MB-54729: Add historyStartSeqno to ScanContext

jimwwalker · jimwwalker · commit a3707d8deb54 · 2023-01-17T11:00:42.000Z
Add a new member, historyStartSeqno to ScanContext for later use in
DCP backfill when "change_stream=true".

This will always be 0 for a KVStore which reports:

    StorageProperties::HistoryRetentionAvailable::No

For a KVStore which reports:

    StorageProperties::HistoryRetentionAvailable::Yes

This will be set to:

    0: If no history is configured (retention_bytes=0)
    &gt;0: When history is available, this value represents the lowest
        seqno at which a continuous change stream could be made
        available.

This commit is not dependent upon Magma's changes to add history, but
adds the ability for MockMagmaKVStore to "inject" a history start.

Change-Id: I66db1d84af54f4a999f260d30b356a9a149ddf31
diff --git a/engines/ep/src/kvstore/kvstore.cc b/engines/ep/src/kvstore/kvstore.cc
@@ -68,6 +68,31 @@ ScanContext::ScanContext(
     Expects(lookup != nullptr);
 }
 
+ScanContext::ScanContext(
+        Vbid vbid,
+        std::unique_ptr<KVFileHandle> handle,
+        DocumentFilter docFilter,
+        ValueFilter valFilter,
+        std::unique_ptr<StatusCallback<GetValue>> cb,
+        std::unique_ptr<StatusCallback<CacheLookup>> cl,
+        const std::vector<Collections::KVStore::DroppedCollection>&
+                droppedCollections,
+        uint64_t maxSeqno,
+        uint64_t historyStartSeqno)
+    : vbid(vbid),
+      handle(std::move(handle)),
+      docFilter(docFilter),
+      valFilter(valFilter),
+      logger(getGlobalBucketLogger().get()),
+      collectionsContext(droppedCollections),
+      maxSeqno(maxSeqno),
+      historyStartSeqno(historyStartSeqno),
+      callback(std::move(cb)),
+      lookup(std::move(cl)) {
+    Expects(callback != nullptr);
+    Expects(lookup != nullptr);
+}
+
 BySeqnoScanContext::BySeqnoScanContext(
         std::unique_ptr<StatusCallback<GetValue>> cb,
         std::unique_ptr<StatusCallback<CacheLookup>> cl,
@@ -100,6 +125,40 @@ BySeqnoScanContext::BySeqnoScanContext(
       timestamp(std::move(timestamp)) {
 }
 
+BySeqnoScanContext::BySeqnoScanContext(
+        std::unique_ptr<StatusCallback<GetValue>> cb,
+        std::unique_ptr<StatusCallback<CacheLookup>> cl,
+        Vbid vb,
+        std::unique_ptr<KVFileHandle> handle,
+        uint64_t start,
+        uint64_t end,
+        uint64_t purgeSeqno,
+        uint64_t historyStartSeqno,
+        DocumentFilter _docFilter,
+        ValueFilter _valFilter,
+        uint64_t _documentCount,
+        const vbucket_state& vbucketState,
+        const std::vector<Collections::KVStore::DroppedCollection>&
+                droppedCollections,
+        std::optional<uint64_t> timestamp)
+
+    : ScanContext(vb,
+                  std::move(handle),
+                  _docFilter,
+                  _valFilter,
+                  std::move(cb),
+                  std::move(cl),
+                  droppedCollections,
+                  end,
+                  historyStartSeqno),
+      startSeqno(start),
+      purgeSeqno(purgeSeqno),
+      documentCount(_documentCount),
+      maxVisibleSeqno(vbucketState.maxVisibleSeqno),
+      persistedCompletedSeqno(vbucketState.persistedCompletedSeqno),
+      timestamp(std::move(timestamp)) {
+}
+
 ByIdScanContext::ByIdScanContext(
         std::unique_ptr<StatusCallback<GetValue>> cb,
         std::unique_ptr<StatusCallback<CacheLookup>> cl,
@@ -123,6 +182,31 @@ ByIdScanContext::ByIdScanContext(
       lastReadKey(nullptr, 0) {
 }
 
+ByIdScanContext::ByIdScanContext(
+        std::unique_ptr<StatusCallback<GetValue>> cb,
+        std::unique_ptr<StatusCallback<CacheLookup>> cl,
+        Vbid vb,
+        std::unique_ptr<KVFileHandle> handle,
+        std::vector<ByIdRange> ranges,
+        DocumentFilter _docFilter,
+        ValueFilter _valFilter,
+        const std::vector<Collections::KVStore::DroppedCollection>&
+                droppedCollections,
+        uint64_t maxSeqno,
+        uint64_t historyStartSeqno)
+    : ScanContext(vb,
+                  std::move(handle),
+                  _docFilter,
+                  _valFilter,
+                  std::move(cb),
+                  std::move(cl),
+                  droppedCollections,
+                  maxSeqno,
+                  historyStartSeqno),
+      ranges(std::move(ranges)),
+      lastReadKey(nullptr, 0) {
+}
+
 bool ByIdRange::operator==(const ByIdRange& other) const {
     return startKey == other.startKey && endKey == other.endKey &&
            rangeScanSuccess == other.rangeScanSuccess;
diff --git a/engines/ep/src/kvstore/kvstore.h b/engines/ep/src/kvstore/kvstore.h
@@ -394,6 +394,17 @@ class ScanContext {
                         droppedCollections,
                 uint64_t maxSeqno);
 
+    ScanContext(Vbid vbid,
+                std::unique_ptr<KVFileHandle> handle,
+                DocumentFilter docFilter,
+                ValueFilter valFilter,
+                std::unique_ptr<StatusCallback<GetValue>> cb,
+                std::unique_ptr<StatusCallback<CacheLookup>> cl,
+                const std::vector<Collections::KVStore::DroppedCollection>&
+                        droppedCollections,
+                uint64_t maxSeqno,
+                uint64_t historyStartSeqno);
+
     virtual ~ScanContext() = default;
 
     virtual const StatusCallback<GetValue>& getValueCallback() const {
@@ -412,14 +423,15 @@ class ScanContext {
         return *lookup;
     }
 
-    const Vbid vbid;
+    const Vbid vbid{0};
     uint64_t lastReadSeqno{0};
     std::unique_ptr<KVFileHandle> handle;
-    const DocumentFilter docFilter;
-    const ValueFilter valFilter;
-    BucketLogger* logger;
+    const DocumentFilter docFilter{DocumentFilter::ALL_ITEMS};
+    const ValueFilter valFilter{ValueFilter::KEYS_ONLY};
+    BucketLogger* logger{nullptr};
     const Collections::VB::ScanContext collectionsContext;
-    uint64_t maxSeqno;
+    uint64_t maxSeqno{0};
+    uint64_t historyStartSeqno{0};
 
     /**
      * Cumulative count of bytes read from disk during this scan. Counts
@@ -456,6 +468,23 @@ class BySeqnoScanContext : public ScanContext {
                     droppedCollections,
             std::optional<uint64_t> timestamp = {});
 
+    BySeqnoScanContext(
+            std::unique_ptr<StatusCallback<GetValue>> cb,
+            std::unique_ptr<StatusCallback<CacheLookup>> cl,
+            Vbid vb,
+            std::unique_ptr<KVFileHandle> handle,
+            uint64_t start,
+            uint64_t end,
+            uint64_t purgeSeqno,
+            uint64_t historyStartSeqno,
+            DocumentFilter _docFilter,
+            ValueFilter _valFilter,
+            uint64_t _documentCount,
+            const vbucket_state& vbucketState,
+            const std::vector<Collections::KVStore::DroppedCollection>&
+                    droppedCollections,
+            std::optional<uint64_t> timestamp = {});
+
     const uint64_t startSeqno;
     const uint64_t purgeSeqno;
     const uint64_t documentCount;
@@ -517,6 +546,18 @@ class ByIdScanContext : public ScanContext {
                     const std::vector<Collections::KVStore::DroppedCollection>&
                             droppedCollections,
                     uint64_t maxSeqno);
+
+    ByIdScanContext(std::unique_ptr<StatusCallback<GetValue>> cb,
+                    std::unique_ptr<StatusCallback<CacheLookup>> cl,
+                    Vbid vb,
+                    std::unique_ptr<KVFileHandle> handle,
+                    std::vector<ByIdRange> ranges,
+                    DocumentFilter _docFilter,
+                    ValueFilter _valFilter,
+                    const std::vector<Collections::KVStore::DroppedCollection>&
+                            droppedCollections,
+                    uint64_t maxSeqno,
+                    uint64_t historyStartSeqno);
     std::vector<ByIdRange> ranges;
     // Key should be set by KVStore when a scan must be paused, this is where
     // a scan can resume from
diff --git a/engines/ep/tests/mock/mock_magma_kvstore.cc b/engines/ep/tests/mock/mock_magma_kvstore.cc
@@ -90,3 +90,44 @@ magma::Status MockMagmaKVStore::runImplicitCompactKVStore(Vbid vbid) {
 magma::Status MockMagmaKVStore::newCheckpoint(Vbid vbid) {
     return magma->NewCheckpoint(vbid.get());
 }
+
+std::unique_ptr<BySeqnoScanContext> MockMagmaKVStore::initBySeqnoScanContext(
+        std::unique_ptr<StatusCallback<GetValue>> cb,
+        std::unique_ptr<StatusCallback<CacheLookup>> cl,
+        Vbid vbid,
+        uint64_t startSeqno,
+        DocumentFilter options,
+        ValueFilter valOptions,
+        SnapshotSource source,
+        std::unique_ptr<KVFileHandle> fileHandle) const {
+    auto scanContext =
+            MagmaKVStore::initBySeqnoScanContext(std::move(cb),
+                                                 std::move(cl),
+                                                 vbid,
+                                                 startSeqno,
+                                                 options,
+                                                 valOptions,
+                                                 source,
+                                                 std::move(fileHandle));
+
+    if (historyStartSeqno) {
+        scanContext->historyStartSeqno = historyStartSeqno.value();
+    }
+    return scanContext;
+}
+
+std::unique_ptr<ByIdScanContext> MockMagmaKVStore::initByIdScanContext(
+        std::unique_ptr<StatusCallback<GetValue>> cb,
+        std::unique_ptr<StatusCallback<CacheLookup>> cl,
+        Vbid vbid,
+        const std::vector<ByIdRange>& ranges,
+        DocumentFilter options,
+        ValueFilter valOptions) const {
+    auto scanContext = MagmaKVStore::initByIdScanContext(
+            std::move(cb), std::move(cl), vbid, ranges, options, valOptions);
+
+    if (historyStartSeqno) {
+        scanContext->historyStartSeqno = historyStartSeqno.value();
+    }
+    return scanContext;
+}
diff --git a/engines/ep/tests/mock/mock_magma_kvstore.h b/engines/ep/tests/mock/mock_magma_kvstore.h
@@ -19,6 +19,34 @@ class MockMagmaKVStore : public MagmaKVStore {
 public:
     explicit MockMagmaKVStore(MagmaKVStoreConfig& config);
 
+    /**
+     * See base-class comments for usage. However this "mock" version will
+     * override the ScanContext::historyStartSeqno if this class defines
+     * historyStartSeqno (see historyStartSeqno member below).
+     */
+    std::unique_ptr<BySeqnoScanContext> initBySeqnoScanContext(
+            std::unique_ptr<StatusCallback<GetValue>> cb,
+            std::unique_ptr<StatusCallback<CacheLookup>> cl,
+            Vbid vbid,
+            uint64_t startSeqno,
+            DocumentFilter options,
+            ValueFilter valOptions,
+            SnapshotSource source,
+            std::unique_ptr<KVFileHandle> fileHandle = nullptr) const override;
+
+    /**
+     * See base-class comments for usage. However this "mock" version will
+     * override the ScanContext::historyStartSeqno if this class defines
+     * historyStartSeqno (see historyStartSeqno member below).
+     */
+    std::unique_ptr<ByIdScanContext> initByIdScanContext(
+            std::unique_ptr<StatusCallback<GetValue>> cb,
+            std::unique_ptr<StatusCallback<CacheLookup>> cl,
+            Vbid vbid,
+            const std::vector<ByIdRange>& ranges,
+            DocumentFilter options,
+            ValueFilter valOptions) const override;
+
     ReadVBStateResult readVBStateFromDisk(Vbid vbid);
 
     ReadVBStateResult readVBStateFromDisk(
@@ -75,6 +103,13 @@ class MockMagmaKVStore : public MagmaKVStore {
     std::function<bool()> snapshotVBucketErrorInjector;
 
     StorageProperties storageProperties;
+
+    /**
+     * The historyStartSeqno in this class when set will override the value
+     * MagmaKVStore sets. This allows arbitrary placement of the history range
+     * for testing of various two-phase backfills.
+     */
+    std::optional<uint64_t> historyStartSeqno;
 };
 
 #endif
