[FLASH-798] Rows not match after delete range (#382)

* Check rows count after delete rows * Fix bug FLASH-798: DM Rows not match after delete range.
pingcap · Jan 2, 2020 · 585f0af · 585f0af
1 parent b2517bd
commit 585f0af
Show file tree

Hide file tree

Showing 2 changed files with 46 additions and 21 deletions.
diff --git a/dbms/src/Storages/DeltaMerge/DeltaTree.h b/dbms/src/Storages/DeltaMerge/DeltaTree.h
@@ -1043,8 +1043,8 @@ void DT_CLASS::addDelete(const UInt64 rid)
     std::tie(leaf, delta) = findLeftLeaf<true>(rid);
     std::tie(pos, delta)  = leaf->searchRid(rid, delta);
 
-    bool   merge = false;
-    size_t merge_pos;
+    bool          merge = false;
+    EntryIterator merge_it;
 
     bool exists = pos != leaf->count && leaf->rid(pos, delta) == rid;
     if (exists && leaf->type(pos) == DT_DEL)
@@ -1054,11 +1054,16 @@ void DT_CLASS::addDelete(const UInt64 rid)
         EntryIterator leaf_end(this->end());
         while (leaf_it != leaf_end && leaf_it.getRid() == rid && leaf_it.getType() == DT_DEL)
         {
-            merge     = true;
-            merge_pos = leaf_it.getPos();
-
+            merge = true;
             ++leaf_it;
         }
+
+        if (merge)
+        {
+            merge_it = leaf_it;
+            --merge_it;
+        }
+
         leaf  = leaf_it.getLeaf();
         pos   = leaf_it.getPos();
         delta = leaf_it.getDelta();
@@ -1116,7 +1121,7 @@ void DT_CLASS::addDelete(const UInt64 rid)
     if (merge)
     {
         /// Simply increase delete count at the last one of delete chain.
-        ++(leaf->mutations[merge_pos].count);
+        ++(merge_it.getLeaf()->mutations[merge_it.getPos()].count);
     }
     else
     {

diff --git a/dbms/src/Storages/StorageDeltaMerge.cpp b/dbms/src/Storages/StorageDeltaMerge.cpp
@@ -603,22 +603,25 @@ BlockInputStreams StorageDeltaMerge::read( //
 
 void StorageDeltaMerge::checkStatus(const Context & context) { store->check(context); }
 
-void StorageDeltaMerge::deleteRows(const Context & context, size_t rows)
+
+size_t getRows(DM::DeltaMergeStorePtr & store, const Context & context, const DM::HandleRange & range)
 {
-    size_t total_rows = 0;
+    size_t rows = 0;
 
-    {
-        ColumnDefines to_read{getExtraHandleColumnDefine()};
-        auto stream = store->read(context, context.getSettingsRef(), to_read, {DM::HandleRange::newAll()}, 1, MAX_UINT64, EMPTY_FILTER)[0];
-        stream->readPrefix();
-        Block block;
-        while ((block = stream->read()))
-            total_rows += block.rows();
-        stream->readSuffix();
-    }
+    ColumnDefines to_read{getExtraHandleColumnDefine()};
+    auto stream = store->read(context, context.getSettingsRef(), to_read, {range}, 1, MAX_UINT64, EMPTY_FILTER)[0];
+    stream->readPrefix();
+    Block block;
+    while ((block = stream->read()))
+        rows += block.rows();
+    stream->readSuffix();
+
+    return rows;
+}
 
-    rows = std::min(total_rows, rows);
-    auto start_index = rand() % (total_rows - rows + 1);
+DM::HandleRange getRange(DM::DeltaMergeStorePtr & store, const Context & context, size_t total_rows, size_t delete_rows)
+{
+    auto start_index = rand() % (total_rows - delete_rows + 1);
 
     DM::HandleRange range = DM::HandleRange::newAll();
     {
@@ -634,15 +637,32 @@ void StorageDeltaMerge::deleteRows(const Context & context, size_t rows)
             {
                 if (index == start_index)
                     range.start = data[i];
-                if (index == start_index + rows)
+                if (index == start_index + delete_rows)
                     range.end = data[i];
                 ++index;
             }
         }
         stream->readSuffix();
     }
 
-    store->deleteRange(context, context.getSettingsRef(), range);
+    return range;
+}
+
+void StorageDeltaMerge::deleteRows(const Context & context, size_t delete_rows)
+{
+    size_t total_rows = getRows(store, context, DM::HandleRange::newAll());
+    delete_rows = std::min(total_rows, delete_rows);
+    auto delete_range = getRange(store, context, total_rows, delete_rows);
+    size_t actual_delete_rows = getRows(store, context, delete_range);
+    if (actual_delete_rows != delete_rows)
+        throw Exception("Expected delete rows: " + DB::toString(delete_rows) + ", got: " + DB::toString(actual_delete_rows));
+
+    store->deleteRange(context, context.getSettingsRef(), delete_range);
+
+    size_t after_delete_rows = getRows(store, context, DM::HandleRange::newAll());
+    if (after_delete_rows != total_rows - delete_rows)
+        throw Exception("Rows after delete range not match, expected: " + DB::toString(total_rows - delete_rows)
+            + ", got: " + DB::toString(after_delete_rows));
 }
 
 //==========================================================================================