MB-54850: Skip magma key-lookup optimization if Disk snap is Historical

As an optimization for improving the performance at flush, we tell
magma whether a replica is receiving the initial disk snapshot, ie the
vbucket is empty. In that case magma can skip the key lookups required
for maintaining the item-count, as any mutation received is an Insert
by logic.

That isn't the case anymore for when the initial disk snapshot is
Historical, as the snapshot might contain duplicates.

Change-Id: I453339fd9dcbe9cbd4dcd8f4c54a18cd8543d5ca
Reviewed-on: https://review.couchbase.org/c/kv_engine/+/184967
Well-Formed: Restriction Checker
Tested-by: Build Bot <[email protected]>
Reviewed-by: Jim Walker <[email protected]>

Author: paolococchi

engines/ep/src/checkpoint.h

@@ -434,6 +434,10 @@ class Checkpoint {
         checkpointType = type;
     }
 
+    void setHistorical(CheckpointHistorical hist) {
+        historical = hist;
+    }
+
     void setHighCompletedSeqno(std::optional<uint64_t> seqno) {
         highCompletedSeqno = seqno;
     }
@@ -629,6 +633,10 @@ class Checkpoint {
      */
     void applyQueuedItemsMemUsageDecrement(size_t size);
 
+    CheckpointHistorical getHistorical() const {
+        return historical;
+    }
+
     // Memory overhead of the toWrite container (a list), ie 3 ptrs (forward,
     // backwards and element pointers) per element in the list.
     static constexpr uint8_t per_item_queue_overhead = 3 * sizeof(uintptr_t);
@@ -764,7 +772,7 @@ class Checkpoint {
 
     // Whether the snapshot stored in this checkpoint is part of a historical
     // sequence of mutation.
-    const CheckpointHistorical historical;
+    CheckpointHistorical historical;
 
     friend std::ostream& operator <<(std::ostream& os, const Checkpoint& m);
 };

engines/ep/src/checkpoint_manager.cc

@@ -106,6 +106,11 @@ CheckpointType CheckpointManager::getOpenCheckpointType() const {
     return getOpenCheckpoint(lh).getCheckpointType();
 }
 
+CheckpointHistorical CheckpointManager::getOpenCheckpointHistorical() const {
+    std::lock_guard<std::mutex> lh(queueLock);
+    return getOpenCheckpoint(lh).getHistorical();
+}
+
 Checkpoint& CheckpointManager::getOpenCheckpoint(
         const std::lock_guard<std::mutex>&) const {
     // During its lifetime, the checkpointList can only be in one of the
@@ -966,11 +971,12 @@ CheckpointManager::ItemsForCursor CheckpointManager::getItemsForCursor(
 
     // Fetch whole checkpoints; as long as we don't exceed the approx item
     // limit.
-    ItemsForCursor result(
-            (*cursor.getCheckpoint())->getCheckpointType(),
-            (*cursor.getCheckpoint())->getMaxDeletedRevSeqno(),
-            (*cursor.getCheckpoint())->getHighCompletedSeqno(),
-            (*cursor.getCheckpoint())->getVisibleSnapshotEndSeqno());
+    const auto& checkpoint = **cursor.getCheckpoint();
+    ItemsForCursor result(checkpoint.getCheckpointType(),
+                          checkpoint.getMaxDeletedRevSeqno(),
+                          checkpoint.getHighCompletedSeqno(),
+                          checkpoint.getVisibleSnapshotEndSeqno(),
+                          checkpoint.getHistorical());
 
     // Only enforce a hard limit for Disk Checkpoints (i.e backfill). This will
     // prevent huge memory growth due to flushing vBuckets on replicas during a
@@ -1309,6 +1315,7 @@ void CheckpointManager::createSnapshot(
         openCkpt.setSnapshotStartSeqno(snapStartSeqno);
         openCkpt.setSnapshotEndSeqno(snapEndSeqno, visibleSnapEnd);
         openCkpt.setCheckpointType(checkpointType);
+        openCkpt.setHistorical(historical);
         openCkpt.setHighCompletedSeqno(highCompletedSeqno);
         return;
     }

engines/ep/src/checkpoint_manager.h

@@ -83,8 +83,10 @@ class CheckpointManager {
         ItemsForCursor(CheckpointType checkpointType,
                        std::optional<uint64_t> maxDeletedRevSeqno,
                        std::optional<uint64_t> highCompletedSeqno,
-                       uint64_t visibleSeqno)
+                       uint64_t visibleSeqno,
+                       CheckpointHistorical historical)
             : checkpointType(checkpointType),
+              historical(historical),
               maxDeletedRevSeqno(maxDeletedRevSeqno),
               highCompletedSeqno(highCompletedSeqno),
               visibleSeqno(visibleSeqno) {
@@ -99,6 +101,8 @@ class CheckpointManager {
          */
         CheckpointType checkpointType = CheckpointType::Memory;
 
+        CheckpointHistorical historical = CheckpointHistorical::No;
+
         std::optional<uint64_t> maxDeletedRevSeqno = {};
 
         /**
@@ -180,6 +184,8 @@ class CheckpointManager {
 
     CheckpointType getOpenCheckpointType() const;
 
+    CheckpointHistorical getOpenCheckpointHistorical() const;
+
     /**
      * Removes closed unreferenced checkpoints from the checkpoint-list and
      * frees up their used memory.

engines/ep/src/ep_bucket.cc

@@ -483,7 +483,8 @@ EPBucket::FlushResult EPBucket::flushVBucket_UNLOCKED(LockedVBucketPtr vb) {
     // KVStore implementations since a lookup isn't needed. This scenario
     // commonly occurs in rebalance where a vbucket is taken over by a new node.
     if (toFlush.ranges.size() == 1 &&
-        toFlush.checkpointType == CheckpointType::InitialDisk) {
+        toFlush.checkpointType == CheckpointType::InitialDisk &&
+        toFlush.historical == CheckpointHistorical::No) {
         writeOp = WriteOperation::Insert;
     }
 

engines/ep/src/vbucket.cc

@@ -442,6 +442,7 @@ ItemsToFlush VBucket::getItemsToPersist(size_t approxLimit) {
     result.ranges = std::move(rangeInfo.ranges);
     result.maxDeletedRevSeqno = rangeInfo.maxDeletedRevSeqno;
     result.checkpointType = rangeInfo.checkpointType;
+    result.historical = rangeInfo.historical;
     result.flushHandle = std::move(rangeInfo.flushHandle);
     result.moreAvailable = rangeInfo.moreAvailable;
 

engines/ep/src/vbucket_types.h

@@ -74,6 +74,8 @@ struct ItemsToFlush {
     std::optional<uint64_t> maxDeletedRevSeqno = {};
     CheckpointType checkpointType = CheckpointType::Memory;
 
+    CheckpointHistorical historical = CheckpointHistorical::No;
+
     // See CM::ItemsForCursor for details.
     UniqueFlushHandle flushHandle;
 };

engines/ep/tests/module_tests/dcp_stream_test.cc

@@ -14,6 +14,7 @@
 #include "checkpoint_manager.h"
 #include "checkpoint_utils.h"
 #include "collections/collections_test_helpers.h"
+#include "collections/events_generated.h"
 #include "collections/vbucket_manifest_handles.h"
 #include "dcp/backfill_disk.h"
 #include "dcp/response.h"
@@ -5242,45 +5243,177 @@ INSTANTIATE_TEST_SUITE_P(Persistent,
                          STParameterizedBucketTest::magmaConfigValues(),
                          STParameterizedBucketTest::PrintToStringParamName);
 
-TEST_P(CDCPassiveStreamTest, HistorySnapshotReceived) {
-    // Replica receives Snap{1, 3, Disk|History}, with 1->3 mutations of the
-    // same key.
-    // The test verifies that replica is resilient to duplicates in the disk
-    // snapshot and that duplicates are successfully queued into the same
-    // checkpoint.
+void CDCPassiveStreamTest::SetUp() {
+    // Note: Checkpoint removal isn't under test at all here.
+    // Eager checkpoint removal, default prod setting in Neo and post-Neo.
+    // That helps in cleaning up the CheckpointManager during the test and
+    // we won't need to fix the testsuite when merging into the master
+    // branch.
+    if (!config_string.empty()) {
+        config_string += ";";
+    }
+    config_string += "checkpoint_removal_mode=eager";
+
+    STPassiveStreamPersistentTest::SetUp();
+}
+
+void CDCPassiveStreamTest::createHistoricalCollection(uint64_t snapStart,
+                                                      uint64_t snapEnd) {
+    const uint32_t opaque = 1;
+    ASSERT_EQ(cb::engine_errc::success,
+              consumer->snapshotMarker(
+                      opaque,
+                      vbid,
+                      snapStart,
+                      snapEnd,
+                      MARKER_FLAG_DISK | MARKER_FLAG_CHK | MARKER_FLAG_HISTORY,
+                      {0},
+                      {}));
+
+    flatbuffers::FlatBufferBuilder fb;
+    Collections::ManifestUid manifestUid;
+    const auto collection = CollectionEntry::historical;
+    auto fbPayload = Collections::VB::CreateCollection(
+            fb,
+            manifestUid,
+            uint32_t(ScopeEntry::defaultS.getId()),
+            uint32_t(collection.getId()),
+            false,
+            0,
+            fb.CreateString(collection.name.data(), collection.name.size()),
+            true /*history*/);
+    fb.Finish(fbPayload);
+    ASSERT_EQ(cb::engine_errc::success,
+              consumer->systemEvent(
+                      1 /*opaque*/,
+                      vbid,
+                      mcbp::systemevent::id::CreateCollection,
+                      1,
+                      mcbp::systemevent::version::version2,
+                      {reinterpret_cast<const uint8_t*>(collection.name.data()),
+                       collection.name.size()},
+                      {fb.GetBufferPointer(), fb.GetSize()}));
 
     const auto& vb = *store->getVBucket(vbid);
-    ASSERT_EQ(0, vb.getHighSeqno());
-    const auto& manager = *vb.checkpointManager;
+    ASSERT_EQ(0, vb.getNumTotalItems());
+    ASSERT_EQ(1, vb.getHighSeqno());
+    auto& manager = *vb.checkpointManager;
+    ASSERT_EQ(1, manager.getNumCheckpoints());
+    ASSERT_EQ(1, manager.getNumOpenChkItems());
+    ASSERT_EQ(CheckpointType::InitialDisk, manager.getOpenCheckpointType());
+    ASSERT_EQ(CheckpointHistorical::Yes, manager.getOpenCheckpointHistorical());
+    flush_vbucket_to_disk(vbid, 1);
+}
+
+/*
+ * HistorySnapshotReceived tests verify (on different scenarios) that:
+ * - replica stream is resilient to duplicates on disk snapshot
+ * - duplicates are queued in checkpoint (ie, not deduplicated)
+ * - duplicates are persisted on disk (again, not deduplicated)
+ * - stats (eg item-count) are updated correctly
+ */
+
+TEST_P(CDCPassiveStreamTest, HistorySnapshotReceived_Disk) {
+    // Replica receives Snap{start, end, Disk|History}, with start->end
+    // mutations for the same key.
+
+    createHistoricalCollection(1, 1);
+
+    // Clear CM
+    const auto& vb = *store->getVBucket(vbid);
+    const auto initialHighSeqno = vb.getHighSeqno();
+    ASSERT_EQ(1, initialHighSeqno);
+    auto& manager = *vb.checkpointManager;
+    manager.createNewCheckpoint(true);
     ASSERT_EQ(1, manager.getNumCheckpoints());
     ASSERT_EQ(0, manager.getNumOpenChkItems());
 
-    const uint32_t opaque = 0;
+    const uint32_t opaque = 1;
     SnapshotMarker snapshotMarker(
             opaque,
             vbid,
-            1 /*start*/,
-            3 /*end*/,
+            initialHighSeqno + 1 /*start*/,
+            initialHighSeqno + 3 /*end*/,
             MARKER_FLAG_CHK | MARKER_FLAG_DISK | MARKER_FLAG_HISTORY,
             std::optional<uint64_t>(0), /*HCS*/
             {}, /*maxVisibleSeqno*/
             {}, /*timestamp*/
             {} /*streamId*/);
     stream->processMarker(&snapshotMarker);
+    ASSERT_EQ(1, manager.getNumCheckpoints());
+    ASSERT_EQ(0, manager.getNumOpenChkItems());
+    ASSERT_EQ(CheckpointType::Disk, manager.getOpenCheckpointType());
+    ASSERT_EQ(CheckpointHistorical::Yes, manager.getOpenCheckpointHistorical());
+
+    const auto collection = CollectionEntry::historical;
+    const std::string key("key");
+    const std::string value("value");
+    const size_t numItems = 3;
+    for (size_t seqno = initialHighSeqno + 1;
+         seqno <= initialHighSeqno + numItems;
+         ++seqno) {
+        EXPECT_EQ(
+                cb::engine_errc::success,
+                stream->messageReceived(makeMutationConsumerMessage(
+                        opaque, seqno, vbid, value, key, collection.getId())));
+    }
+
+    EXPECT_EQ(initialHighSeqno + numItems, vb.getHighSeqno());
     EXPECT_EQ(1, manager.getNumCheckpoints());
-    EXPECT_EQ(0, manager.getNumOpenChkItems());
+    EXPECT_EQ(numItems, manager.getNumOpenChkItems());
+
+    // All duplicates persisted
+    flush_vbucket_to_disk(vbid, numItems);
+
+    // Item count doesn't account for historical revisions
+    EXPECT_EQ(1, vb.getNumTotalItems());
+}
+
+TEST_P(CDCPassiveStreamTest, HistorySnapshotReceived_InitialDisk) {
+    // Replica receives Snap{start, end, InitialDisk|History}, with start->end
+    // mutations for the same key.
 
+    createHistoricalCollection(1, 10);
+
+    const auto& vb = *store->getVBucket(vbid);
+    const auto initialHighSeqno = vb.getHighSeqno();
+    ASSERT_EQ(1, initialHighSeqno);
+    auto& manager = *vb.checkpointManager;
+    ASSERT_EQ(1, manager.getNumCheckpoints());
+    ASSERT_EQ(1, manager.getNumOpenChkItems());
+
+    // Historical items received within the same snapshot
+    const auto collection = CollectionEntry::historical;
     const std::string key("key");
     const std::string value("value");
-    for (size_t seqno = 1; seqno <= 3; ++seqno) {
+    const size_t numItems = 3;
+    for (size_t seqno = initialHighSeqno + 1;
+         seqno <= initialHighSeqno + numItems;
+         ++seqno) {
         EXPECT_EQ(cb::engine_errc::success,
-                  stream->messageReceived(makeMutationConsumerMessage(
-                          seqno, vbid, value, opaque, key)));
+                  stream->messageReceived(
+                          makeMutationConsumerMessage(1 /*opaque*/,
+                                                      seqno,
+                                                      vbid,
+                                                      value,
+                                                      key,
+                                                      collection.getId())));
     }
 
-    EXPECT_EQ(3, vb.getHighSeqno());
+    // Important: In this scenario historical mutations are queued into the
+    // Initial disk checkpoint
+    ASSERT_EQ(CheckpointType::InitialDisk, manager.getOpenCheckpointType());
+    ASSERT_EQ(CheckpointHistorical::Yes, manager.getOpenCheckpointHistorical());
+
+    EXPECT_EQ(initialHighSeqno + numItems, vb.getHighSeqno());
     EXPECT_EQ(1, manager.getNumCheckpoints());
-    EXPECT_EQ(3, manager.getNumOpenChkItems());
+    EXPECT_EQ(initialHighSeqno + numItems, manager.getNumOpenChkItems());
+
+    // All duplicates persisted
+    flush_vbucket_to_disk(vbid, numItems);
+
+    // Item count doesn't account for historical revisions
+    EXPECT_EQ(1, vb.getNumTotalItems());
 }
 
 INSTANTIATE_TEST_SUITE_P(Persistent,

engines/ep/tests/module_tests/dcp_stream_test.h

@@ -229,4 +229,15 @@ class CDCActiveStreamTest : public STActiveStreamPersistentTest {
  * connections.
  */
 class CDCPassiveStreamTest : public STPassiveStreamPersistentTest {
+protected:
+    void SetUp() override;
+
+    /**
+     * Create an historical collection at replica.
+     * SysEvent received in a snapshot with range defined by the user.
+     *
+     * @param snapStart
+     * @parma snapEnd
+     */
+    void createHistoricalCollection(uint64_t snapStart, uint64_t snapEnd);
 };

engines/ep/tests/module_tests/dcp_utils.cc

@@ -127,3 +127,32 @@ std::unique_ptr<MutationConsumerMessage> makeMutationConsumerMessage(
             nullptr,
             cb::mcbp::DcpStreamId{});
 }
+
+std::unique_ptr<MutationConsumerMessage> makeMutationConsumerMessage(
+        uint64_t opaque,
+        uint64_t seqno,
+        Vbid vbid,
+        const std::string& value,
+        const std::string& key,
+        CollectionID cid) {
+    queued_item qi(new Item(makeStoredDocKey(key, cid),
+                            0,
+                            0,
+                            value.c_str(),
+                            value.size(),
+                            PROTOCOL_BINARY_RAW_BYTES,
+                            0,
+                            seqno,
+                            vbid,
+                            1));
+    return std::make_unique<MutationConsumerMessage>(
+            std::move(qi),
+            opaque,
+            IncludeValue::Yes,
+            IncludeXattrs::Yes,
+            IncludeDeleteTime::No,
+            IncludeDeletedUserXattrs::Yes,
+            DocKeyEncodesCollectionId::No,
+            nullptr,
+            cb::mcbp::DcpStreamId{});
+}