MB-32361: Trigger compaction when a drop collection is persisted

When the flusher has persisted the end of a collection it is
safe to trigger the compactor to reclaim that resource.

Change-Id: I5989e965c2c378ed793edcd0e931f7b045068699
Reviewed-on: http://review.couchbase.org/102848
Tested-by: Build Bot <[email protected]>
Reviewed-by: Ben Huddleston <[email protected]>

Author: jimwwalker

engines/ep/src/collections/flush.cc

@@ -18,6 +18,7 @@
 #include "collections/flush.h"
 #include "collections/collection_persisted_stats.h"
 #include "collections/vbucket_manifest.h"
+#include "ep_bucket.h"
 #include "item.h"
 
 void Collections::VB::Flush::processManifestChange(const queued_item& item) {
@@ -83,4 +84,13 @@ void Collections::VB::Flush::setPersistedHighSeqno(const DocKey& key,
         mutated.insert(key.getCollectionID());
         manifest.lock(key).setPersistedHighSeqno(value);
     }
+}
+
+void Collections::VB::Flush::checkAndTriggerPurge(Vbid vbid,
+                                                  EPBucket& bucket) const {
+    if (!deletedCollections.empty()) {
+        CompactionConfig config;
+        config.db_file_id = vbid;
+        bucket.scheduleCompaction(vbid, config, nullptr);
+    }
 }

engines/ep/src/collections/flush.h

@@ -23,6 +23,8 @@
 #include <unordered_set>
 #include <vector>
 
+class EPBucket;
+
 namespace Collections {
 namespace VB {
 
@@ -88,6 +90,14 @@ class Flush {
      */
     void setPersistedHighSeqno(const DocKey& key, uint64_t value);
 
+    /**
+     * Check to see if this flush should trigger a collection purge which if
+     * true schedules a task which will iterate the vbucket's documents removing
+     * those of any dropped collections. The actual task currently scheduled is
+     * compaction.
+     */
+    void checkAndTriggerPurge(Vbid vbid, EPBucket& bucket) const;
+
 private:
     /**
      * Keep track of only the collections that have had a insert/delete in

engines/ep/src/ep_bucket.cc

@@ -469,6 +469,8 @@ std::pair<bool, size_t> EPBucket::flushVBucket(Vbid vbid) {
             stats.cumulativeFlushTime.fetch_add(trans_time);
             stats.flusher_todo.store(0);
             stats.totalPersistVBState++;
+
+            sef.getCollectionFlush().checkAndTriggerPurge(vb->getId(), *this);
         }
 
         rwUnderlying->pendingTasks();

engines/ep/tests/module_tests/collections/evp_store_collections_eraser_test.cc

@@ -37,10 +37,6 @@ class CollectionsEraserTest
         SingleThreadedKVBucketTest::TearDown();
     }
 
-    void runEraser() {
-        runCompaction();
-    }
-
     bool isFullEviction() const {
         return GetParam().find("item_eviction_policy=full_eviction") !=
                std::string::npos;
@@ -79,7 +75,7 @@ TEST_P(CollectionsEraserTest, basic) {
     // Deleted, but still exists in the manifest
     EXPECT_TRUE(vb->lockCollections().exists(CollectionEntry::dairy));
 
-    runEraser();
+    runCollectionsEraser();
 
     EXPECT_EQ(0, vb->getNumItems());
 
@@ -119,7 +115,7 @@ TEST_P(CollectionsEraserTest, basic_2_collections) {
 
     flush_vbucket_to_disk(vbid, 2 /* 2 x system */);
 
-    runEraser();
+    runCollectionsEraser();
 
     EXPECT_EQ(0, vb->getNumItems());
 
@@ -159,7 +155,7 @@ TEST_P(CollectionsEraserTest, basic_3_collections) {
 
     flush_vbucket_to_disk(vbid, 1 /* 1 x system */);
 
-    runEraser();
+    runCollectionsEraser();
 
     EXPECT_EQ(2, vb->getNumItems());
 
@@ -200,7 +196,7 @@ TEST_P(CollectionsEraserTest, basic_4_collections) {
 
     flush_vbucket_to_disk(vbid, 3 /* 3x system (2 deletes, 1 create) */);
 
-    runEraser();
+    runCollectionsEraser();
 
     EXPECT_EQ(0, vb->getNumItems());
 
@@ -233,7 +229,7 @@ TEST_P(CollectionsEraserTest, default_Destroy) {
 
     flush_vbucket_to_disk(vbid, 1 /* 1 x system */);
 
-    runEraser();
+    runCollectionsEraser();
 
     EXPECT_EQ(0, vb->getNumItems());
 
@@ -287,7 +283,7 @@ TEST_P(CollectionsEraserTest, erase_and_reset) {
 
     flush_vbucket_to_disk(vbid, 3 /* 3x system (2 deletes, 1 create) */);
 
-    runEraser();
+    runCollectionsEraser();
 
     EXPECT_EQ(0, vb->getNumItems());
 

engines/ep/tests/module_tests/collections/evp_store_collections_test.cc

@@ -864,14 +864,14 @@ TEST_F(CollectionsWarmupTest, warmupIgnoreLogicallyDeleted) {
                   vb->lockCollections().getItemCount(CollectionEntry::meat));
     } // VBucketPtr scope ends
 
+    // Ensure collection purge has executed
+    runCollectionsEraser();
+
     resetEngineAndWarmup();
 
     EXPECT_EQ(0, store->getVBucket(vbid)->ht.getNumInMemoryItems());
-    // Eraser hasn't ran, but the collection deletion will have resulted in
-    // the stat document being removed
-    EXPECT_EQ(0,
-              store->getVBucket(vbid)->lockCollections().getItemCount(
-                      CollectionEntry::meat));
+    EXPECT_FALSE(store->getVBucket(vbid)->lockCollections().exists(
+            CollectionEntry::meat));
 }
 
 //
@@ -912,17 +912,17 @@ TEST_F(CollectionsWarmupTest, warmupIgnoreLogicallyDeletedDefault) {
                           CollectionEntry::defaultC));
     } // VBucketPtr scope ends
 
+    // Ensure collection purge has executed
+    runCollectionsEraser();
+
     resetEngineAndWarmup();
 
     EXPECT_EQ(0, store->getVBucket(vbid)->ht.getNumInMemoryItems());
-    // Eraser hasn't ran, but the collection deletion will have resulted in
-    // the stat document being removed
-    EXPECT_EQ(0,
-              store->getVBucket(vbid)->lockCollections().getItemCount(
-                      CollectionEntry::defaultC));
-    EXPECT_EQ(0,
-              store->getVBucket(vbid)->lockCollections().getPersistedHighSeqno(
-                      CollectionEntry::defaultC));
+    // meat collection still exists
+    EXPECT_TRUE(store->getVBucket(vbid)->lockCollections().exists(
+            CollectionEntry::meat));
+    EXPECT_TRUE(store->getVBucket(vbid)->lockCollections().isCollectionOpen(
+            CollectionEntry::meat));
 }
 
 TEST_F(CollectionsWarmupTest, warmupManifestUidLoadsOnCreate) {
@@ -942,6 +942,10 @@ TEST_F(CollectionsWarmupTest, warmupManifestUidLoadsOnCreate) {
     // validate the manifest uid comes back
     EXPECT_EQ(0xface2,
               store->getVBucket(vbid)->lockCollections().getManifestUid());
+    EXPECT_TRUE(store->getVBucket(vbid)->lockCollections().exists(
+            CollectionEntry::meat));
+    EXPECT_TRUE(store->getVBucket(vbid)->lockCollections().isCollectionOpen(
+            CollectionEntry::meat));
 }
 
 TEST_F(CollectionsWarmupTest, warmupManifestUidLoadsOnDelete) {

engines/ep/tests/module_tests/evp_store_rollback_test.cc

@@ -38,12 +38,12 @@
 #include <mcbp/protocol/framebuilder.h>
 #include <memcached/server_cookie_iface.h>
 
-class RollbackTest : public EPBucketTest,
+class RollbackTest : public SingleThreadedEPBucketTest,
                      public ::testing::WithParamInterface<
                              std::tuple<std::string, std::string>> {
     void SetUp() override {
         config_string += "item_eviction_policy=" + std::get<0>(GetParam());
-        EPBucketTest::SetUp();
+        SingleThreadedEPBucketTest::SetUp();
         if (std::get<1>(GetParam()) == "pending") {
             vbStateAtRollback = vbucket_state_pending;
         } else {

engines/ep/tests/module_tests/evp_store_single_threaded_test.cc

@@ -284,6 +284,15 @@ void SingleThreadedKVBucketTest::runCompaction(uint64_t purgeBeforeTime,
                 "Compact DB file 0");
 }
 
+void SingleThreadedKVBucketTest::runCollectionsEraser() {
+    ASSERT_TRUE(engine->getConfiguration().getBucketType() == "persistent")
+            << "No support for ephemeral collections erase";
+    // run the compaction task. assuming it was scheduled by the test, will
+    // fail the runNextTask expect if not scheduled.
+    runNextTask(*task_executor->getLpTaskQ()[WRITER_TASK_IDX],
+                "Compact DB file 0");
+}
+
 /*
  * MB-31175
  * The following test checks to see that when we call handleSlowStream in an

engines/ep/tests/module_tests/evp_store_single_threaded_test.h

@@ -98,6 +98,13 @@ class SingleThreadedKVBucketTest : public KVBucketTest {
     void runCompaction(uint64_t purgeBeforeTime = 0,
                        uint64_t purgeBeforeSeq = 0);
 
+    /**
+     * Run the task responsible for iterating the documents and erasing them
+     * For persistent buckets integrated into compaction.
+     * For ephemeral buckets integrated into stale item removal task
+     */
+    void runCollectionsEraser();
+
 protected:
     void SetUp() override;