feat: support delete in pre-aggregator #2689

XLC-2 · 2022-10-19T09:04:23Z

What kind of change does this PR introduce? (Bug fix, feature, docs update, ...)
Bug fix
What is the current behavior? (You can also link to an open issue here)
After deleting a record from a table, the pre aggregation table corresponding to this table will not be updated.
What is the new behavior (if this is a feature change)?
After deleting a record from a table, the pre aggregation table corresponding to this table will be updated.

…f new and old interfaces

…deleting record

… generate combined_key, 2. RemoveDeletedKey add combined_key to distinguish deletion granularity

…eded; 2. remove the extra process of parsing records; 3. fix cpplint error

…ata when do FreeTsEntry

…last record is deleted, the keyEntry will also be deleted

…nMLDB into fix/data-consistency-pre-aggr2

dl239 · 2022-10-19T09:12:46Z

hybridse/src/planv2/ast_node_converter.cc

-                    op = node::FnOperator::kFnOpRLike;
-                    break;
-                }
+                // case zetasql::ASTBinaryExpression::Op::RLIKE: {


revert this file

dl239 · 2022-10-19T09:18:39Z

src/storage/aggregator.h


    bool Delete(const std::string& key);

    bool FlushAll();

    bool Init(std::shared_ptr<LogReplicator> base_replicator);

+    void SetReletedTable(std::shared_ptr<Table> base_table);


pass base table in Init function

dl239 · 2022-10-19T09:21:04Z

src/storage/aggregator.cc

@@ -447,6 +495,8 @@ bool Aggregator::FlushAggrBuffer(const std::string& key, const std::string& filt
    row_builder_.SetTimestamp(row_ptr, 1, buffer.ts_begin_);
    row_builder_.SetTimestamp(row_ptr, 2, buffer.ts_end_);
    row_builder_.SetInt32(row_ptr, 3, buffer.aggr_cnt_);
+    std::cout << "[DEBUG]: aggr_cnt: " << buffer.aggr_cnt_


DLOG(INFO) << "aggr_cnt: " << buffer.aggr_cnt_ << " aggr_val: " << aggr_val;

codecov · 2022-10-19T09:28:43Z

Codecov Report

❗ No coverage uploaded for pull request base (dev@770f04b). Click here to learn what that means.
Patch has no changes to coverable lines.

Additional details and impacted files

@@          Coverage Diff           @@
##             dev    #2689   +/-   ##
======================================
  Coverage       ?   36.70%           
  Complexity     ?      392           
======================================
  Files          ?      156           
  Lines          ?     9257           
  Branches       ?     1060           
======================================
  Hits           ?     3398           
  Misses         ?     5622           
  Partials       ?      237

Help us with your feedback. Take ten seconds to tell us how you rate us. Have a feature suggestion? Share it here.

☔ View full report at Codecov.
📢 Do you have feedback about the report comment? Let us know in this issue.

dl239 · 2022-10-19T09:43:58Z

src/storage/aggregator.cc

+    if (reverse) {
+        ok = UpdateAggrValAfterDeleteRange(base_row_view_, row_ptr, &aggr_buffer);
+    } else {
+        ok = UpdateAggrVal(base_row_view_, row_ptr, &aggr_buffer, reverse);


why pass the reverse argument that is always false

Because UpdateAggrVal will be called in other places, and the paremeter 'reverse' in other places may be true

github-actions · 2022-10-19T09:44:59Z

Linux Test Report

577 tests 567 ✔️ 26m 25s ⏱️
  57 suites     7 💤
  51 files     3 ❌

For more details on these failures, see this check.

Results for commit a4cf474.

♻️ This comment has been updated with latest results.

github-actions · 2022-10-19T09:46:11Z

SDK Test Report

  83 files   83 suites 1m 40s ⏱️
174 tests 168 ✔️ 6 💤 0 ❌
220 runs 214 ✔️ 6 💤 0 ❌

Results for commit a4cf474.

♻️ This comment has been updated with latest results.

dl239 · 2022-10-19T09:52:28Z

src/storage/aggregator.cc

+    ::openmldb::api::Dimension* dimension = entry.add_dimensions();
+    dimension->set_key(key);
+    dimension->set_idx(aggr_index_pos_);
+    ok = aggr_replicator_->AppendEntry(entry);


no need to call AppendEntry of agg_replicator

github-actions · 2022-10-19T10:15:16Z

HybridSE Linux Test Report

19 719 tests 19 697 ✔️ 8m 6s ⏱️
    240 suites         2 💤
      67 files       20 ❌

For more details on these failures, see this check.

Results for commit f9780e4.

github-actions · 2022-10-19T10:15:27Z

HybridSE Mac Test Report

19 719 tests 19 696 ✔️ 8m 51s ⏱️
    240 suites         2 💤
      67 files       21 ❌

For more details on these failures, see this check.

Results for commit f9780e4.

dl239 · 2022-10-20T03:40:08Z

src/storage/aggregator_test.cc

+        AddDefaultAggregatorSchema(&aggr_table_meta);
+        std::map<std::string, uint32_t> mapping;
+        mapping.insert(std::make_pair("idx", 0));
+        std::shared_ptr<Table> aggr_table =


std::shared_ptr<Table> aggr_table = std::make_shared<MemTable>(aggr_table_meta);

…le can be updated after deleting a record

dl239 · 2022-10-24T03:50:22Z

src/storage/aggregator.cc

    if (!ok) {
        PDLOG(ERROR, "Update aggr value failed");
        return false;
    }
    return true;
 }

+bool Aggregator::DeleteAndUpdate(const std::string& key, const std::string& row,
+                                 const uint64_t& offset, bool recover) {


where this function is used except test case

dl239 · 2022-10-24T03:53:30Z

src/tablet/tablet_impl.cc

@@ -3122,7 +3122,7 @@ int TabletImpl::LoadTableInternal(uint32_t tid, uint32_t pid, std::shared_ptr<::
                auto aggrs = GetAggregators(tid, pid);
                if (aggrs != nullptr) {
                    for (auto& aggr : *aggrs) {
-                        if (!aggr->Init(replicator)) {
+                        if (!aggr->Init(replicator, nullptr)) {


why pass nullptr

dl239 · 2022-10-24T04:04:46Z

src/storage/aggregator.cc

            break;
        }
        case DataType::kInt: {
            int32_t val;
            row_view.GetValue(row_ptr, aggr_col_idx_, aggr_col_type_, &val);
-            aggr_buffer->aggr_val_.vlong += val;
+            if (reverse) {
+                aggr_buffer->aggr_val_.vlong -= val;    


cpplint error

dl239 · 2022-10-24T04:06:12Z

src/storage/aggregator.cc

+            auto tmp_val = it->GetValue();
+            std::string origin_data = tmp_val.ToString();
+            const int8_t* row_ptr = reinterpret_cast<const int8_t*>(it->GetValue().data());
+            codec::RowView base_row_view(base_table_->GetTableMeta()->column_desc(),


use row_view passed from the function arguments

dl239 · 2022-10-24T04:07:51Z

src/storage/aggregator.cc

+            const int8_t* row_ptr = reinterpret_cast<const int8_t*>(it->GetValue().data());
+            codec::RowView base_row_view(base_table_->GetTableMeta()->column_desc(),
+                                         reinterpret_cast<int8_t*>(const_cast<char*>(origin_data.c_str())),
+                                         origin_data.size());


call UpdateAggrVal(..., false) recursively？

dl239 · 2022-10-27T09:25:32Z

src/tablet/tablet_impl.cc

@@ -5850,7 +5850,7 @@ bool TabletImpl::CreateAggregatorInternal(const ::openmldb::api::CreateAggregato
    }

    auto base_replicator = GetReplicator(base_meta->tid(), base_meta->pid());
-    if (!aggregator->Init(base_replicator)) {
+    if (!aggregator->Init(base_replicator, nullptr)) {


pass base table?

dl239 · 2022-10-27T09:31:35Z

src/storage/aggregator_test.cc

@@ -173,6 +429,45 @@ void CheckSumAggrResult(std::shared_ptr<Table> aggr_table, DataType data_type, s
    return;
 }

+template <typename T>
+void CheckSumAggrResultAfterDelete(std::shared_ptr<Table> aggr_table, DataType data_type, 


resolve cpplint error

dl239 · 2022-10-27T09:42:21Z

src/storage/aggregator.h

@@ -127,13 +127,14 @@ class Aggregator {

    ~Aggregator();

-    bool Update(const std::string& key, const std::string& row, const uint64_t& offset, bool recover = false);
+    bool Update(const std::string& key, const std::string& row, const uint64_t& offset, bool recover = false,


call this function in Delete method of tablet_impl.cc

zhanghaohit · 2022-10-31T03:21:31Z

src/storage/aggregator.cc

-    bool ok = UpdateAggrVal(base_row_view_, base_row_ptr, &tmp_buffer);
+    bool ok = false;
+    if (reverse) {
+        ok = aggr_table_->Delete(key, aggr_index_pos_, tmp_buffer.ts_begin_);


should remove the cur_ts?

zhanghaohit · 2022-10-31T03:41:42Z

src/storage/aggregator.cc

-            if (aggr_buffer->AggrValEmpty() || val < aggr_buffer->aggr_val_.vint) {
-                aggr_buffer->aggr_val_.vint = val;
+        aggr_buffer->non_null_cnt_ = 0;
+        auto it = base_table_->NewTraverseIterator(0);


we should use:

NewWindowIterator(index_pos_)

seek(key) and then seek(ts_begin_)

reset the aggr_buffer

iterate and update all the rows within the [ts_end_, ts_begin_]

The first two steps can also use NewTraverseIterator(index_pos_); and Seek(key, ts).

zhanghaohit · 2022-10-31T03:43:00Z

src/storage/aggregator.cc

+        aggr_buffer->non_null_cnt_ = 0;
+        auto it = base_table_->NewTraverseIterator(0);
+        it->SeekToFirst();
+        while (it->Valid() && it->GetKey() >= aggr_buffer->ts_begin_) {


can use the same common function for min and max

…Aggr

zhanghaohit · 2022-11-16T02:12:32Z

src/storage/aggregator.cc

+    int64_t ts_end = aggr_buffer->ts_end_;
+    it->Seek(key, ts_end);
+    aggr_buffer->clear();
+    if (this->GetWindowType() == WindowType::kRowsRange) {


for both kRowsRange and kkRowsNum, the ts_begin_ and ts_end_ should keep the same.

zhanghaohit · 2022-11-16T02:13:58Z

src/storage/aggregator.cc

+            if (this->GetWindowType() == WindowType::kRowsRange) {
+                if (aggr_buffer->ts_begin_ == -1) {
+                    // the first record.
+                    aggr_buffer->ts_end_ = it->GetKey();
+                }
+                aggr_buffer->ts_begin_ = it->GetKey();
+            }


why check the kRowsRange again here? you intend to check kRowsNum?

as we keep the ts_end_ and ts_start_ as previous buffer, can remove the code

zhanghaohit · 2022-11-16T02:20:31Z

src/storage/aggregator.cc

+        PDLOG(ERROR, "base table is nullptr, cannot update MinAggr table");
+        return false;
+    }
+    auto it = base_table_->NewTraverseIterator(this->GetIndexPos());


if you prefer to use TraverseIterator, you have to check the key and GetPK are the same. So here better to use WindowIterator to simplify the code and performance concern.

zhanghaohit · 2022-11-16T02:25:54Z

src/storage/aggregator.cc

-        }
-        default: {
-            PDLOG(ERROR, "Unsupported data type");
-            return false;
        }
    }
    aggr_buffer->non_null_cnt_++;


this should be put for the else branch only

zhanghaohit · 2022-11-16T02:26:18Z

src/storage/aggregator.cc

-        }
-        default: {
-            PDLOG(ERROR, "Unsupported data type");
-            return false;
        }
    }
    aggr_buffer->non_null_cnt_++;


non_null_cnt_++ for the else branch only

dl239 · 2023-06-25T04:17:52Z

has done in #3327

XLC-2 and others added 16 commits August 12, 2022 07:35

feat: Delete record interface of specified timestamp

935fef1

fix: protect deletion process with lock and add judgment on the use o…

39f9221

…f new and old interfaces

fix: use idx instead of idx_name when rollback

2991374

fix: delete redundant code

4c52a8f

fix: fix bugs found in code review

b4a3b5c

fix: judge nullptr after deleting a record from skiplist

4aa4228

fix: skip deleting a record when it is in use

ddfd423

fix: snapshot remove deleted key when ts_col in index and add test cases

aa5c0d4

fix: fix compile bugs about snapshot_test and improve the process of …

97b6fa5

…deleting record

fix: 1. CollectDeletedKey and RemoveDeletedKey have the same logic to…

5e6dcfb

… generate combined_key, 2. RemoveDeletedKey add combined_key to distinguish deletion granularity

fix: fix cpplint error

245e986

fix: 1. parse the ts column only instead of the entire record when ne…

1ae48b3

…eded; 2. remove the extra process of parsing records; 3. fix cpplint error

fix: 1. use version decoder decode ts column; 2. update statistical d…

62ee01f

…ata when do FreeTsEntry

fix: when there is no record in the corresponding keyEntry after the …

82628f2

…last record is deleted, the keyEntry will also be deleted

Merge branch 'fix/data-consistency-pre-aggr2' of github.com:XLC-2/Ope…

fa3a8a6

…nMLDB into fix/data-consistency-pre-aggr2

feat: update aggr table when deleting a record

f9780e4

github-actions bot added execute-engine hybridse sql engine storage-engine openmldb storage engine. nameserver & tablet labels Oct 19, 2022

dl239 changed the title ~~Fix/data consistency pre aggr2~~ feat: support delete in pre-aggregator Oct 19, 2022

dl239 reviewed Oct 19, 2022

View reviewed changes

dl239 reviewed Oct 20, 2022

View reviewed changes

fix: The maximum pre aggregation table or minimum pre aggregation tab…

5b7228d

…le can be updated after deleting a record

github-actions bot removed the execute-engine hybridse sql engine label Oct 24, 2022

dl239 reviewed Oct 24, 2022

View reviewed changes

fix: modify the process of updating the extreme value aggregation table

e51b609

dl239 reviewed Oct 27, 2022

View reviewed changes

dl239 requested a review from zhanghaohit October 27, 2022 09:32

dl239 marked this pull request as ready for review October 27, 2022 09:32

dl239 reviewed Oct 27, 2022

View reviewed changes

fix: updating aggr table when tablet deletes a record

efd164d

zhanghaohit requested changes Nov 7, 2022

View reviewed changes

fix: use the same function to rebuild aggr buffer for MinAggr and Max…

a4cf474

…Aggr

zhanghaohit requested changes Nov 16, 2022

View reviewed changes

tobegit3hub assigned dl239 Feb 28, 2023

dl239 closed this Jun 25, 2023

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat: support delete in pre-aggregator #2689

feat: support delete in pre-aggregator #2689

XLC-2 commented Oct 19, 2022

dl239 Oct 19, 2022

dl239 Oct 19, 2022

dl239 Oct 19, 2022

codecov bot commented Oct 19, 2022 •

edited

Loading

dl239 Oct 19, 2022

XLC-2 Oct 20, 2022

github-actions bot commented Oct 19, 2022 •

edited

Loading

github-actions bot commented Oct 19, 2022 •

edited

Loading

dl239 Oct 19, 2022

github-actions bot commented Oct 19, 2022

github-actions bot commented Oct 19, 2022

dl239 Oct 20, 2022 •

edited

Loading

dl239 Oct 24, 2022 •

edited

Loading

dl239 Oct 24, 2022

dl239 Oct 24, 2022

dl239 Oct 24, 2022

dl239 Oct 24, 2022

dl239 Oct 27, 2022

dl239 Oct 27, 2022

dl239 Oct 27, 2022

zhanghaohit Oct 31, 2022

zhanghaohit Oct 31, 2022

zhanghaohit Oct 31, 2022

zhanghaohit Nov 16, 2022

zhanghaohit Nov 16, 2022

zhanghaohit Nov 16, 2022

zhanghaohit Nov 16, 2022

zhanghaohit Nov 16, 2022

dl239 commented Jun 25, 2023

feat: support delete in pre-aggregator #2689

feat: support delete in pre-aggregator #2689

Conversation

XLC-2 commented Oct 19, 2022

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

codecov bot commented Oct 19, 2022 • edited Loading

Codecov Report

Choose a reason for hiding this comment

Choose a reason for hiding this comment

github-actions bot commented Oct 19, 2022 • edited Loading

Linux Test Report

github-actions bot commented Oct 19, 2022 • edited Loading

SDK Test Report

Choose a reason for hiding this comment

github-actions bot commented Oct 19, 2022

HybridSE Linux Test Report

github-actions bot commented Oct 19, 2022

HybridSE Mac Test Report

dl239 Oct 20, 2022 • edited Loading

Choose a reason for hiding this comment

dl239 Oct 24, 2022 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dl239 commented Jun 25, 2023

codecov bot commented Oct 19, 2022 •

edited

Loading

github-actions bot commented Oct 19, 2022 •

edited

Loading

github-actions bot commented Oct 19, 2022 •

edited

Loading

dl239 Oct 20, 2022 •

edited

Loading

dl239 Oct 24, 2022 •

edited

Loading