MB-41405: Read persistedDeletes from RW store

Currently when we query the KVStore for the persisted deletes stat we
query the RO store which has its own cache and does not track persisted
deletes (as they are tracked in the RW store). This means that querying
persisted deletes returns the first value correctly (as we read it from
disk) but every subsequent query will return the same value. Correct
this by querying persisted deletes from the RW store.

Change-Id: I2524ca0df1b23a428a34bc645dd606ae48ddaefe
Reviewed-on: http://review.couchbase.org/c/kv_engine/+/136523
Well-Formed: Build Bot <[email protected]>
Tested-by: Build Bot <[email protected]>
Reviewed-by: Paolo Cocchi <[email protected]>
Reviewed-by: Jim Walker <[email protected]>

Author: BenHuddleston

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

@@ -2811,6 +2811,10 @@ ENGINE_ERROR_CODE CouchKVStore::couchErr2EngineErr(couchstore_error_t errCode) {
 }
 
 size_t CouchKVStore::getNumPersistedDeletes(Vbid vbid) {
+    // cachedDeletes isn't tracked correctly for the RO store as it's only set
+    // on write so we can't read the stat from it.
+    Expects(!readOnly);
+
     size_t delCount = cachedDeleteCount[vbid.get()];
     if (delCount != (size_t) -1) {
         return delCount;

engines/ep/src/ep_bucket.cc

@@ -118,7 +118,8 @@ bool BloomFilterCallback::initTempFilter(Vbid vbucketId) {
     size_t estimated_count;
     size_t num_deletes = 0;
     try {
-        num_deletes = store.getROUnderlying(vbucketId)->getNumPersistedDeletes(vbucketId);
+        num_deletes = store.getRWUnderlying(vbucketId)->getNumPersistedDeletes(
+                vbucketId);
     } catch (std::runtime_error& re) {
         EP_LOG_WARN(
                 "BloomFilterCallback::initTempFilter: runtime error while "

engines/ep/src/ep_vb.cc

@@ -809,7 +809,7 @@ size_t EPVBucket::getNumPersistedDeletes() const {
         // If creation is true then no disk file exists
         return 0;
     }
-    return shard->getROUnderlying()->getNumPersistedDeletes(getId());
+    return shard->getRWUnderlying()->getNumPersistedDeletes(getId());
 }
 
 void EPVBucket::dropKey(int64_t bySeqno,

engines/ep/tests/module_tests/evp_store_single_threaded_test.cc

@@ -320,11 +320,12 @@ void SingleThreadedKVBucketTest::createDcpStream(MockDcpProducer& producer,
 }
 
 void SingleThreadedKVBucketTest::runCompaction(uint64_t purgeBeforeTime,
-                                               uint64_t purgeBeforeSeq) {
+                                               uint64_t purgeBeforeSeq,
+                                               bool dropDeletes) {
     CompactionConfig compactConfig;
     compactConfig.purge_before_ts = purgeBeforeTime;
     compactConfig.purge_before_seq = purgeBeforeSeq;
-    compactConfig.drop_deletes = false;
+    compactConfig.drop_deletes = dropDeletes;
     compactConfig.db_file_id = vbid;
     store->scheduleCompaction(vbid, compactConfig, nullptr);
     // run the compaction task
@@ -5151,6 +5152,97 @@ TEST_P(STParamPersistentBucketTest, FlushFailureAtPerstingDelete_VBDeletion) {
     testFlushFailureAtPersistDelete(COUCHSTORE_ERROR_WRITE, true);
 }
 
+TEST_P(STParameterizedBucketTest, DeleteUpdatesPersistedDeletes) {
+    store->setVBucketState(vbid, vbucket_state_active);
+
+    auto vb = store->getVBucket(vbid);
+    ASSERT_TRUE(vb);
+    EXPECT_EQ(0, vb->getNumPersistedDeletes());
+
+    store_item(vbid,
+               makeStoredDocKey("keyA"),
+               "value",
+               0 /*exptime*/,
+               {cb::engine_errc::success} /*expected*/,
+               PROTOCOL_BINARY_RAW_BYTES);
+    delete_item(vbid, makeStoredDocKey("keyA"));
+
+    flushVBucketToDiskIfPersistent(vbid, 1);
+
+    // Before the bug fix the stat would be wrong as we'd read from the RO
+    // store but only update the cached value in the RW store.
+    EXPECT_EQ(1, vb->getNumPersistedDeletes());
+}
+
+void STParamPersistentBucketTest::testCompactionPersistedDeletes(
+        bool dropDeletes) {
+    store->setVBucketState(vbid, vbucket_state_active);
+
+    flushVBucketToDiskIfPersistent(vbid, 0);
+
+    auto vb = store->getVBucket(vbid);
+    ASSERT_TRUE(vb);
+    ASSERT_NE(0, vb->getFilterSize());
+
+    // Stat should be correct and we should populate the cached value
+    EXPECT_EQ(0, vb->getNumPersistedDeletes());
+
+    // Persist first delete
+    store_item(vbid,
+               makeStoredDocKey("keyA"),
+               "value",
+               0 /*exptime*/,
+               {cb::engine_errc::success} /*expected*/,
+               PROTOCOL_BINARY_RAW_BYTES);
+    delete_item(vbid, makeStoredDocKey("keyA"));
+
+    store_item(vbid,
+               makeStoredDocKey("keyB"),
+               "value",
+               0 /*exptime*/,
+               {cb::engine_errc::success} /*expected*/,
+               PROTOCOL_BINARY_RAW_BYTES);
+    delete_item(vbid, makeStoredDocKey("keyB"));
+
+    flushVBucketToDiskIfPersistent(vbid, 2);
+
+    EXPECT_EQ(2, vb->getNumPersistedDeletes());
+
+    runCompaction(0, 0, dropDeletes);
+}
+
+TEST_P(STParamPersistentBucketTest, CompactionUpdatesPersistedDeletes) {
+    testCompactionPersistedDeletes(true /*dropDeletes*/);
+
+    auto vb = store->getVBucket(vbid);
+    ASSERT_TRUE(vb);
+
+    // Before the bug fix the stat would be wrong as we'd read from the RO
+    // store but only update the cached value in the RW store. This won't be 0
+    // even though we have 2 deletes as we keep the last item during a
+    // compaction.
+    EXPECT_EQ(1, vb->getNumPersistedDeletes());
+}
+
+TEST_P(STParamPersistentBucketTest, CompactionUpdatesBloomFilter) {
+    engine->getConfiguration().setBfilterKeyCount(1);
+
+    testCompactionPersistedDeletes(false /*dropDeletes*/);
+
+    auto vb = store->getVBucket(vbid);
+    ASSERT_TRUE(vb);
+
+    // Before the bug fix the stat would be wrong as we'd read from the RO
+    // store but only update the cached value in the RW store.
+    EXPECT_EQ(2, vb->getNumPersistedDeletes());
+
+    auto expected = 29;
+    if (fullEviction()) {
+        expected = 10;
+    }
+    EXPECT_EQ(expected, vb->getFilterSize());
+}
+
 TEST_P(STParamPersistentBucketTest, FlusherMarksCleanBySeqno) {
     setVBucketStateAndRunPersistTask(vbid, vbucket_state_active);
 

engines/ep/tests/module_tests/evp_store_single_threaded_test.h

@@ -102,9 +102,11 @@ class SingleThreadedKVBucketTest : public KVBucketTest {
      * Run the compaction task
      * @param purgeBeforeTime purge tombstones with timestamps less than this
      * @param purgeBeforeSeq purge tombstones with seqnos less than this
+     * @param dropDeletes drop all deletes
      */
     void runCompaction(uint64_t purgeBeforeTime = 0,
-                       uint64_t purgeBeforeSeq = 0);
+                       uint64_t purgeBeforeSeq = 0,
+                       bool dropDeletes = false);
 
     /**
      * Run the task responsible for iterating the documents and erasing them
@@ -364,4 +366,11 @@ class STParamPersistentBucketTest : public STParameterizedBucketTest {
             couchstore_error_t failureCode, bool vbDeletion = false);
     void testFlushFailureAtPersistDelete(couchstore_error_t failureCode,
                                          bool vbDeletion = false);
+
+    /**
+     * Test to check that we update and use persistedDeletes correctly.
+     *
+     * @param dropDeletes compaction config param
+     */
+    void testCompactionPersistedDeletes(bool dropDeletes);
 };