MB-54729: Add KVStore::scanAllVersions

Add a new scan method that scans the seqno index.

For now this function is the same as ::scan, but will be updated
to scan the retained history.

Change-Id: I7a2c424b212962634120f2ba3f831f47e72c9820
Reviewed-on: https://review.couchbase.org/c/kv_engine/+/184676
Well-Formed: Restriction Checker
Tested-by: Jim Walker <[email protected]>
Reviewed-by: Paolo Cocchi <[email protected]>

Author: jimwwalker

engines/ep/src/kvstore/couch-kvstore/couch-kvstore.cc

@@ -2376,6 +2376,11 @@ scan_error_t CouchKVStore::scan(BySeqnoScanContext& ctx) const {
     return scan_success;
 }
 
+scan_error_t CouchKVStore::scanAllVersions(BySeqnoScanContext& sctx) const {
+    // StorageProperties reported this feature is not available
+    throw std::runtime_error("CouchKVStore::scanAllVersions was invoked");
+}
+
 scan_error_t CouchKVStore::scan(ByIdScanContext& ctx) const {
     TRACE_EVENT_START2("CouchKVStore",
                        "scan by id",

engines/ep/src/kvstore/couch-kvstore/couch-kvstore.h

@@ -333,6 +333,7 @@ class CouchKVStore : public KVStore
             ValueFilter valOptions) const override;
 
     scan_error_t scan(BySeqnoScanContext& sctx) const override;
+    scan_error_t scanAllVersions(BySeqnoScanContext& sctx) const override;
     scan_error_t scan(ByIdScanContext& sctx) const override;
 
     std::unique_ptr<KVFileHandle> makeFileHandle(Vbid vbid) const override;

engines/ep/src/kvstore/kvstore_iface.h

@@ -521,6 +521,20 @@ class KVStoreIface {
      */
     virtual scan_error_t scan(BySeqnoScanContext& sctx) const = 0;
 
+    /**
+     * Run a BySeqno scan that retrieves all versions of a document from the
+     * configured "window of history".
+     *
+     * This function is only valid to call when the KVStore reports
+     * StorageProperties::HistoryRetentionAvailable::Yes and that the scan
+     * is with the range of retained history as reported by
+     * ScanContext::historyStartSeqno
+     *
+     * @param sctx non-const reference to the context, internal callbacks may
+     *        write to the object as progress is made through the scan
+     */
+    virtual scan_error_t scanAllVersions(BySeqnoScanContext& sctx) const = 0;
+
     /**
      * Run a ById scan
      * @param sctx non-const reference to the context, internal callbacks may

engines/ep/src/kvstore/magma-kvstore/magma-kvstore.cc

@@ -1805,6 +1805,12 @@ std::unique_ptr<ByIdScanContext> MagmaKVStore::initByIdScanContext(
     return sctx;
 }
 
+scan_error_t MagmaKVStore::scanAllVersions(BySeqnoScanContext& ctx) const {
+    // @todo use magma's mode
+    // return scan(ctx, magma::Magma::SeqIterator::Mode::History);
+    return scan(ctx);
+}
+
 scan_error_t MagmaKVStore::scan(BySeqnoScanContext& ctx) const {
     if (ctx.lastReadSeqno == ctx.maxSeqno) {
         logger->TRACE("MagmaKVStore::scan {} lastReadSeqno:{} == maxSeqno:{}",

engines/ep/src/kvstore/magma-kvstore/magma-kvstore.h

@@ -233,6 +233,7 @@ class MagmaKVStore : public KVStore {
             ValueFilter valOptions) const override;
 
     scan_error_t scan(BySeqnoScanContext& sctx) const override;
+    scan_error_t scanAllVersions(BySeqnoScanContext& sctx) const override;
     scan_error_t scan(ByIdScanContext& ctx) const override;
 
     std::unique_ptr<KVFileHandle> makeFileHandle(Vbid vbid) const override;

engines/ep/src/kvstore/nexus-kvstore/nexus-kvstore.cc

@@ -2789,6 +2789,10 @@ scan_error_t NexusKVStore::scan(BySeqnoScanContext& ctx) const {
     return primaryScanResult;
 }
 
+scan_error_t NexusKVStore::scanAllVersions(BySeqnoScanContext& ctx) const {
+    throw std::runtime_error("NexusKVStore::scanAllVersions unimplemented");
+}
+
 scan_error_t NexusKVStore::scan(ByIdScanContext& ctx) const {
     auto& nexusCtx = dynamic_cast<NexusKVStoreByIdScanContext&>(ctx);
     auto& primaryCtx = *nexusCtx.primaryCtx;

engines/ep/src/kvstore/nexus-kvstore/nexus-kvstore.h

@@ -103,6 +103,7 @@ class NexusKVStore : public KVStoreIface {
             DocumentFilter options,
             ValueFilter valOptions) const override;
     scan_error_t scan(BySeqnoScanContext& sctx) const override;
+    scan_error_t scanAllVersions(BySeqnoScanContext& sctx) const override;
     scan_error_t scan(ByIdScanContext& sctx) const override;
     std::unique_ptr<KVFileHandle> makeFileHandle(Vbid vbid) const override;
     std::pair<GetCollectionStatsStatus, Collections::VB::PersistedStats>

engines/ep/src/kvstore/rocksdb-kvstore/rocksdb-kvstore.h

@@ -276,6 +276,11 @@ class RocksDBKVStore : public KVStore {
     }
 
     scan_error_t scan(BySeqnoScanContext& sctx) const override;
+
+    scan_error_t scanAllVersions(BySeqnoScanContext& sctx) const override {
+        throw std::runtime_error("RocksDB no support for scanAllVersions");
+    }
+
     scan_error_t scan(ByIdScanContext& sctx) const override {
         throw std::runtime_error("RocksDB no support for byID scan");
     }

engines/ep/tests/mock/mock_kvstore.h

@@ -158,6 +158,10 @@ class MockKVStore : public KVStore {
                 scan,
                 (BySeqnoScanContext & sctx),
                 (const, override));
+    MOCK_METHOD(scan_error_t,
+                scanAllVersions,
+                (BySeqnoScanContext & sctx),
+                (const, override));
     MOCK_METHOD(scan_error_t,
                 scan,
                 (ByIdScanContext & sctx),

engines/ep/tests/module_tests/magma-kvstore_test.cc

@@ -57,6 +57,21 @@ class MagmaKVStoreTest : public KVStoreTest {
         rollbackTest = true;
     }
 
+    queued_item doWrite(uint64_t seqno,
+                        bool expected,
+                        const std::string& key = "key") {
+        auto ctx =
+                kvstore->begin(vbid, std::make_unique<PersistenceCallback>());
+        auto qi = makeCommittedItem(makeStoredDocKey(key),
+                                    "value" + std::to_string(seqno));
+        qi->setBySeqno(seqno);
+        flush.proposedVBState.lastSnapStart = seqno;
+        flush.proposedVBState.lastSnapEnd = seqno;
+        kvstore->set(*ctx, qi);
+        EXPECT_EQ(expected, kvstore->commit(std::move(ctx), flush));
+        return qi;
+    };
+
 private:
     bool rollbackTest{false};
 };
@@ -697,4 +712,34 @@ TEST_F(MagmaKVStoreTest, makeFileHandleSyncFailed) {
     setupSyncStatus(magma::Status(magma::Status::Internal, "Internal"));
     fileHandle = kvstore->makeFileHandle(vbid);
     EXPECT_FALSE(fileHandle);
+}
+
+// @todo: This is a basic test that will be expanded to cover scanning history
+// at the moment this test is equivalent to "scan"
+TEST_F(MagmaKVStoreTest, scanAllVersions) {
+    initialize_kv_store(kvstore.get(), vbid);
+    std::vector<queued_item> expectedItems;
+    expectedItems.push_back(doWrite(1, true, "k1"));
+    expectedItems.push_back(doWrite(2, true, "k2"));
+    auto validate = [&expectedItems](GetValue gv) {
+        ASSERT_TRUE(gv.item);
+        ASSERT_GE(expectedItems.size(), size_t(gv.item->getBySeqno()));
+        EXPECT_EQ(*expectedItems[gv.item->getBySeqno() - 1], *gv.item);
+    };
+    auto bySeq = kvstore->initBySeqnoScanContext(
+            std::make_unique<CustomCallback<GetValue>>(validate),
+            std::make_unique<CustomCallback<CacheLookup>>(),
+            vbid,
+            1,
+            DocumentFilter::ALL_ITEMS,
+            ValueFilter::VALUES_COMPRESSED,
+            SnapshotSource::Head);
+    ASSERT_TRUE(bySeq);
+    // @todo: This must be the expected seqno where history begins
+    EXPECT_EQ(0, bySeq->historyStartSeqno);
+    EXPECT_EQ(scan_success, kvstore->scanAllVersions(*bySeq));
+
+    auto& cb =
+            static_cast<CustomCallback<GetValue>&>(bySeq->getValueCallback());
+    EXPECT_EQ(2, cb.getProcessedCount());
 }