MB-23898: skip get() call if KeyOnly is true

Previously we would retrieve the value of a key even if the stream
was key only. Instead, check if this is the case, and skip the retrieval
 if the key is not a SystemEvent (which still requires the value).

Also add unit-testing for this.

Change-Id: I6bdc567cd2c983ee5ef67f4064ba1198776cf487
Reviewed-on: https://review.couchbase.org/c/kv_engine/+/171581
Tested-by: Build Bot <[email protected]>
Reviewed-by: Jim Walker <[email protected]>

Author: redm2

engines/ep/src/dcp/backfill_disk.cc

@@ -70,19 +70,44 @@ void CacheCallback::callback(CacheLookup& lookup) {
         return;
     }
 
-    auto gv = get(*vb, lookup, *stream_);
-    if (gv.getStatus() == cb::engine_errc::success) {
-        if (gv.item->getBySeqno() == lookup.getBySeqno()) {
-            if (stream_->backfillReceived(std::move(gv.item),
-                                          BACKFILL_FROM_MEMORY)) {
-                setStatus(cb::engine_errc::key_already_exists);
-                return;
-            }
-            setStatus(cb::engine_errc::no_memory); // Pause the backfill
+    auto item = std::unique_ptr<Item>();
+
+    // We don't need to read the value if the stream is KeyOnly AND this key
+    // is not a SystemEvent (which requires the full payload for all streams)
+    if (stream_->isKeyOnly() &&
+        !(lookup.getKey().getDocKey().isInSystemCollection())) {
+        // Create an empty value to backfill with
+        item = std::make_unique<Item>(
+                lookup.getKey().getDocKey(), /* docKey */
+                0, /* flags */
+                0, /* expiry */
+                nullptr, /* value */
+                0, /* how much memory to allocate */
+                PROTOCOL_BINARY_RAW_BYTES, /* binary protocol */
+                0, /* cas */
+                lookup.getBySeqno()); /* seqNo */
+
+    } else {
+        auto gv = get(*vb, lookup, *stream_);
+        // If we could not retrieve the value or the Item seqNo does not match,
+        // return success so the stream continues onto the next key; no backfill
+        if (gv.getStatus() != cb::engine_errc::success ||
+            gv.item->getBySeqno() != lookup.getBySeqno()) {
+            setStatus(cb::engine_errc::success);
             return;
         }
-    }
-    setStatus(cb::engine_errc::success);
+        item = std::move(gv.item);
+    }
+
+    // Perform the backfill and set status to key_already_exists if successful.
+    // Otherwise pause stream backfill as op failed - not enough free memory
+    bool backfillSuccess =
+            stream_->backfillReceived(std::move(item), BACKFILL_FROM_MEMORY);
+    cb::engine_errc newStatus = (backfillSuccess)
+                                        ? cb::engine_errc::key_already_exists
+                                        : cb::engine_errc::no_memory;
+    setStatus(newStatus);
+    return;
 }
 
 DiskCallback::DiskCallback(std::shared_ptr<ActiveStream> s) : streamPtr(s) {

engines/ep/src/dcp/backfill_disk.h

@@ -33,7 +33,23 @@ enum backfill_state_t {
 class CacheCallback : public StatusCallback<CacheLookup> {
 public:
     CacheCallback(KVBucket& bucket, std::shared_ptr<ActiveStream> s);
-
+    /**
+     * Function called for each key during stream backfill. Informs the caller
+     * if the information required to populate the DCP stream for the item is
+     * in-memory.
+     *
+     * If all required information is in-memory, the item is backfilled and
+     * status is set to cb::engine_errc::key_already_exists.
+     *
+     * If there is additional information required that isn't in-memory, or the
+     * item should not be included in the stream for some reason, the status is
+     * set to cb::engine_errc::success.
+     *
+     * The status is set to cb::engine_errc::no_memory only if there is not
+     * enough memory to backfill, pausing backfilling temporarily.
+     *
+     * @param lookup a reference to a CacheLookup
+     */
     void callback(CacheLookup& lookup) override;
 
 private:

engines/ep/src/vbucket.cc

@@ -2784,11 +2784,10 @@ GetValue VBucket::getInternal(const CookieIface* cookie,
     const bool getDeletedValue = (options & GET_DELETED_VALUE);
     const bool bgFetchRequired = (options & QUEUE_BG_FETCH);
 
-    auto res = fetchValidValue(WantsDeleted::Yes,
-                               trackReference,
-                               cHandle,
-                               getReplicaItem);
+    auto res = fetchValidValue(
+            WantsDeleted::Yes, trackReference, cHandle, getReplicaItem);
 
+    this->isCalledHook();
     auto* v = res.storedValue;
     if (v) {
         // If the fetched value is a Prepared SyncWrite which may already have

engines/ep/src/vbucket.h

@@ -2516,6 +2516,9 @@ class VBucket : public std::enable_shared_from_this<VBucket> {
     // prepare (if the associated Commit/Abort may have been deduped)
     int64_t allowedDuplicatePrepareThreshold = 0;
 
+    // Test hook used to determine if the method .getInternal() is called
+    TestingHook<> isCalledHook;
+
     friend class DurabilityMonitorTest;
     friend class SingleThreadedActiveStreamTest;
     friend class VBucketTestBase;

engines/ep/tests/module_tests/backfill_disk_test.cc

@@ -21,12 +21,20 @@
 #include "tests/mock/mock_kvstore.h"
 #include "tests/mock/mock_stream.h"
 #include "tests/mock/mock_synchronous_ep_engine.h"
+#include "tests/module_tests/vbucket_utils.h"
+
 #include <kv_bucket.h>
 
 using namespace ::testing;
 
 /// Test fixture for DCPBackfillDisk class tests.
-class DCPBackfillDiskTest : public SingleThreadedEPBucketTest {};
+class DCPBackfillDiskTest : public SingleThreadedEPBucketTest {
+protected:
+    void backfillGetDriver(IncludeValue incVal,
+                           IncludeXattrs incXattr,
+                           IncludeDeletedUserXattrs incDeletedXattr,
+                           int expectedGetCalls);
+};
 
 /**
  * Regression test for MB-47790 - if a backfill fails during the scan() phase
@@ -96,3 +104,88 @@ TEST_F(DCPBackfillDiskTest, ScanDiskError) {
     // Replace the MockKVStore with the real one so we can tidy up correctly
     MockKVStore::restoreOriginalRWKVStore(*store);
 }
+
+void DCPBackfillDiskTest::backfillGetDriver(
+        IncludeValue incVal,
+        IncludeXattrs incXattr,
+        IncludeDeletedUserXattrs incDeletedXattr,
+        int expectedGetCalls) {
+    setVBucketStateAndRunPersistTask(vbid, vbucket_state_active);
+    // Create item, checkpoint and flush so item is to be backfilled from disk
+    store_item(vbid, makeStoredDocKey("key"), "value");
+    flushAndRemoveCheckpoints(vbid);
+
+    // Set up and define expectations for the hook that is called in the body of
+    // CacheCallback::get(), i.e., when an item's value is retrieved from cache
+    auto& vb = *engine->getKVBucket()->getVBucket(vbid);
+    testing::StrictMock<testing::MockFunction<void()>> getInternalHook;
+    VBucketTestIntrospector::setIsCalledHook(vb,
+                                             getInternalHook.AsStdFunction());
+    EXPECT_CALL(getInternalHook, Call()).Times(expectedGetCalls);
+
+    // Items now only on disk, create producer
+    auto producer = std::make_shared<MockDcpProducer>(
+            *engine, cookie, "test-producer", 0 /*flags*/, false /*startTask*/);
+
+    auto stream = std::make_shared<MockActiveStream>(
+            engine.get(),
+            producer,
+            0, /* flags */
+            0, /* opaque */
+            vb, /* vbucket */
+            0, /* start seqNo */
+            1, /* end seqNo */
+            0, /* vbucket uuid */
+            0, /* snapshot start seqNo */
+            0, /* snapshot end seqNo */
+            incVal, /* includeValue */
+            incXattr, /* includeXattrs */
+            incDeletedXattr, /* includeDeletedUserXattrs */
+            std::string{}); /* jsonFilter */
+
+    stream->setActive();
+    ASSERT_TRUE(stream->isBackfilling());
+
+    auto& bfm = producer->getBFM();
+    ASSERT_EQ(backfill_success, bfm.backfill()); // initialize backfill
+
+    auto backfillRemaining = stream->getNumBackfillItemsRemaining();
+    ASSERT_TRUE(backfillRemaining);
+
+    ASSERT_EQ(backfill_success, bfm.backfill()); // scan
+    ASSERT_EQ(2, stream->public_readyQSize());
+
+    auto response = stream->public_popFromReadyQ();
+    EXPECT_EQ(DcpResponse::Event::SnapshotMarker, response->getEvent());
+
+    // Ensure an item mutation is in the DCP stream
+    response = stream->public_popFromReadyQ();
+    EXPECT_EQ(DcpResponse::Event::Mutation, response->getEvent());
+
+    // Ensure this item has the correct key, and value (if IncludeValue::Yes)
+    MutationResponse mutResponse = dynamic_cast<MutationResponse&>(*response);
+    SingleThreadedRCPtr item = mutResponse.getItem();
+    EXPECT_EQ(item->getKey(), makeStoredDocKey("key"));
+
+    if (!(stream->isKeyOnly())) {
+        EXPECT_EQ(item->getValue()->to_s(), "value");
+    }
+}
+
+// Tests that CacheCallback::get is never called when a stream is keyOnly.
+TEST_F(DCPBackfillDiskTest, KeyOnlyBackfillSkipsScan) {
+    DCPBackfillDiskTest::backfillGetDriver(IncludeValue::No,
+                                           IncludeXattrs::No,
+                                           IncludeDeletedUserXattrs::No,
+                                           0);
+}
+
+// Complement to KeyOnlyBackfillSkipGet. Other tests already cover all cases,
+// but not using the hook. This test validates the hook, and thus
+// KeyOnlyBackfillSkipGet itself, is performing correctly and can be trusted.
+TEST_F(DCPBackfillDiskTest, ValueBackfillRegressionTest) {
+    DCPBackfillDiskTest::backfillGetDriver(IncludeValue::Yes,
+                                           IncludeXattrs::Yes,
+                                           IncludeDeletedUserXattrs::Yes,
+                                           1);
+}

engines/ep/tests/module_tests/vbucket_utils.h

@@ -38,4 +38,8 @@ class VBucketTestIntrospector {
     static void destroyDM(VBucket& vb) {
         vb.durabilityMonitor.reset();
     }
+
+    static void setIsCalledHook(VBucket& vb, std::function<void()> hook) {
+        vb.isCalledHook = hook;
+    }
 };