Revert "MB-41321: 2/4 Clean-up if collection exists in old and new drop containers"

This reverts commit a3cc422.

Reason for revert: Build break - merged out of order?

Change-Id: Ib709a997cee0e64ff3f66eacf4e778b26b8e0cd4
Reviewed-on: http://review.couchbase.org/c/kv_engine/+/137630
Tested-by: Dave Rigby <[email protected]>
Reviewed-by: Dave Rigby <[email protected]>

Author: daverigby

engines/ep/src/collections/flush.cc

@@ -448,44 +448,29 @@ flatbuffers::DetachedBuffer Flush::encodeOpenCollections(
 
 flatbuffers::DetachedBuffer Flush::encodeDroppedCollections(
         std::vector<Collections::KVStore::DroppedCollection>& existingDropped) {
-    flatbuffers::FlatBufferBuilder builder;
-    std::vector<flatbuffers::Offset<Collections::KVStore::Dropped>> output;
-
-    // Add Collection's to this set only if they are in both old and new
-    // sets of dropped collections - this ensures we generate it once into
-    // the new output.
-    std::unordered_set<CollectionID> skip;
-
     // Iterate through the existing dropped collections and look each up in the
     // commit metadata. If the collection is in both lists, we will just update
     // the existing data (adjusting the endSeqno) and then erase the collection
     // from the commit meta's dropped collections.
     for (auto& collection : existingDropped) {
         if (auto itr = droppedCollections.find(collection.collectionId);
             itr != droppedCollections.end()) {
-            // Collection is in both old and new 'sets' of dropped collections
-            // we only want it in the output once - update the end-seqno here
-            // and add to skip set
             collection.endSeqno = itr->second.endSeqno;
-            skip.emplace(collection.collectionId);
+
+            // Now kick the collection out of collectionsMeta, its contribution
+            // to the final output is complete
+            droppedCollections.erase(itr);
         }
-        auto newEntry = Collections::KVStore::CreateDropped(
-                builder,
-                collection.startSeqno,
-                collection.endSeqno,
-                uint32_t(collection.collectionId));
-        output.push_back(newEntry);
     }
 
-    // Now add the newly dropped collections
-    // Iterate through the set of collections dropped in the commit batch and
+    flatbuffers::FlatBufferBuilder builder;
+    std::vector<flatbuffers::Offset<Collections::KVStore::Dropped>> output;
+
+    // Now merge, first the newly dropped collections
+    // Iterate  through the list collections dropped in the commit batch and
     // and create flatbuffer versions of each one
     for (const auto& [cid, dropped] : droppedCollections) {
         (void)cid;
-        if (skip.count(cid) > 0) {
-            // This collection is already in output
-            continue;
-        }
         auto newEntry = Collections::KVStore::CreateDropped(
                 builder,
                 dropped.startSeqno,
@@ -494,6 +479,16 @@ flatbuffers::DetachedBuffer Flush::encodeDroppedCollections(
         output.push_back(newEntry);
     }
 
+    // and now copy across the existing dropped collections
+    for (const auto& entry : existingDropped) {
+        auto newEntry = Collections::KVStore::CreateDropped(
+                builder,
+                entry.startSeqno,
+                entry.endSeqno,
+                uint32_t(entry.collectionId));
+        output.push_back(newEntry);
+    }
+
     auto vector = builder.CreateVector(output);
     auto final =
             Collections::KVStore::CreateDroppedCollections(builder, vector);

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

@@ -2858,31 +2858,6 @@ TEST_P(CollectionsParameterizedTest, OneScopeStatsByIdParsing) {
     EXPECT_EQ(cm.getUid(), result.getManifestId());
 }
 
-// Test a specific issue spotted, in the case when a collection exists
-// as dropped in storage and is newly dropped - we clean-up
-TEST_P(CollectionsPersistentParameterizedTest, FlushDropCreateDropCleansUp) {
-    CollectionsManifest cm;
-    cm.add(CollectionEntry::fruit); // seq:1
-    setCollections(cookie, std::string{cm});
-    flushVBucketToDiskIfPersistent(vbid, 1);
-    cm.remove(CollectionEntry::fruit); // seq:2
-    setCollections(cookie, std::string{cm});
-    flushVBucketToDiskIfPersistent(vbid, 1);
-    cm.add(CollectionEntry::fruit); // seq:3
-    setCollections(cookie, std::string{cm});
-    flushVBucketToDiskIfPersistent(vbid, 1);
-    cm.remove(CollectionEntry::fruit); // seq:4
-    setCollections(cookie, std::string{cm});
-    flushVBucketToDiskIfPersistent(vbid, 1);
-
-    VBucketPtr vb = store->getVBucket(vbid);
-
-    // Final flush should of removed making this invalid and we get an exception
-    EXPECT_THROW(vb->getManifest().lock().getStatsForFlush(
-                         CollectionEntry::fruit, 4),
-                 std::logic_error);
-}
-
 INSTANTIATE_TEST_SUITE_P(CollectionsExpiryLimitTests,
                          CollectionsExpiryLimitTest,
                          ::testing::Bool(),