MB-47604: Have CompactionContext update rollbackPurgeSeqno

For NexusKVStore we need to know the highest purged seqno before we
can enable implicit compaction or concurrent flush and compaction. To
get this we could add and track a new seqno but this is only required
for NexusKVStore. It makes more sense to instead call some function on
CompactionContext for each purged item, and defer to subclasses for the
required action. This also allows us to tidy up the purging behaviours
which may/may not update seqnos based on the type of item purged and use
common code for all KVStores.

Change-Id: I8e289529ac4a4bfb2677b9cf77145d21ece9a4a5
Reviewed-on: http://review.couchbase.org/c/kv_engine/+/163092
Reviewed-by: Richard de Mellow <[email protected]>
Tested-by: Build Bot <[email protected]>

Author: BenHuddleston

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

@@ -892,6 +892,7 @@ static int time_purge_hook(Db& d,
             }
         }
         // No need to make 'disk-size' changes here as the collection has gone.
+        ctx.purgedItem(PurgedItemType::LogicalDelete, info->db_seq);
         return COUCHSTORE_COMPACT_DROP_ITEM;
     } else if (docKey.isInSystemCollection()) {
         ctx.eraserContext->processSystemEvent(
@@ -922,13 +923,10 @@ static int time_purge_hook(Db& d,
             }
 
             if (purgeItem) {
-                // Maybe update purge seqno
-                if (ctx.rollbackPurgeSeqno < info->db_seq) {
-                    ctx.rollbackPurgeSeqno = info->db_seq;
-                }
                 // Update stats and return
                 ctx.stats.tombstonesPurged++;
                 maybeAccountForPurgedCollectionData();
+                ctx.purgedItem(PurgedItemType::Tombstone, info->db_seq);
                 return COUCHSTORE_COMPACT_DROP_ITEM;
             }
         }
@@ -947,6 +945,7 @@ static int time_purge_hook(Db& d,
                 // as the stat doc will have already been deleted (for
                 // couchstore)
                 maybeAccountForPurgedCollectionData();
+                ctx.purgedItem(PurgedItemType::Prepare, info->db_seq);
                 return COUCHSTORE_COMPACT_DROP_ITEM;
             }
 

engines/ep/src/kvstore/kvstore.h

@@ -129,6 +129,18 @@ struct CompactionConfig {
     bool retain_erroneous_tombstones = false;
 };
 
+/**
+ * Type of item purged
+ */
+enum class PurgedItemType {
+    // Tombstone - deleted document
+    Tombstone = 0,
+    // Logical deletion - item belonging to dropped collection
+    LogicalDelete,
+    // Prepare - complete prepare (seqno lower than PCS)
+    Prepare,
+};
+
 struct CompactionContext {
     CompactionContext(Vbid vbid,
                       const CompactionConfig& config,
@@ -139,6 +151,34 @@ struct CompactionContext {
           rollbackPurgeSeqno(purgeSeq),
           timeToExpireFrom(timeToExpireFrom) {
     }
+
+    /**
+     * Process a purged item
+     *
+     * @param type The type of the item purged
+     * @param seqno The seqno of the item purged
+     */
+    void purgedItem(PurgedItemType type, uint64_t seqno) {
+        switch (type) {
+        case PurgedItemType::Tombstone:
+            // Only tombstones need to move the rollback purge seqno
+            updateRollbackPurgeSeqno(seqno);
+            break;
+        case PurgedItemType::LogicalDelete:
+        case PurgedItemType::Prepare:
+            break;
+        }
+    }
+
+    /**
+     * Update the rollback purge seqno
+     *
+     * @param seqno The seqno of the item purged
+     */
+    void updateRollbackPurgeSeqno(uint64_t seqno) {
+        rollbackPurgeSeqno = std::max(rollbackPurgeSeqno, seqno);
+    }
+
     Vbid vbid;
 
     /// The configuration for this compaction.

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

@@ -336,6 +336,7 @@ bool MagmaKVStore::compactionCore(MagmaKVStore::MagmaCompactionCB& cbCtx,
                         vbid,
                         userSanitizedItemStr);
             }
+            cbCtx.ctx->purgedItem(PurgedItemType::LogicalDelete, seqno);
             return true;
         }
     }
@@ -376,14 +377,12 @@ bool MagmaKVStore::compactionCore(MagmaKVStore::MagmaCompactionCB& cbCtx,
 
             if (drop) {
                 cbCtx.ctx->stats.tombstonesPurged++;
-                if (cbCtx.ctx->rollbackPurgeSeqno < seqno) {
-                    cbCtx.ctx->rollbackPurgeSeqno = seqno;
-                }
                 auto& dbStats = cbCtx.magmaDbStats;
                 if (dbStats.purgeSeqno < seqno) {
                     dbStats.purgeSeqno = seqno;
                 }
                 maybeUpdatePurgeSeqno();
+                cbCtx.ctx->purgedItem(PurgedItemType::Tombstone, seqno);
                 return true;
             }
         }
@@ -403,6 +402,7 @@ bool MagmaKVStore::compactionCore(MagmaKVStore::MagmaCompactionCB& cbCtx,
                             vbid,
                             userSanitizedItemStr);
                 }
+                cbCtx.ctx->purgedItem(PurgedItemType::Prepare, seqno);
                 return true;
             }
         }