feat: produce gcSnapshotBitmap during Memory GC for zero-traversal checkpointing

Author: gengdy1545

cpp/pixels-retina/include/RGVisibility.h

@@ -33,7 +33,7 @@ class RGVisibility : public pixels::RetinaBase<RGVisibility<CAPACITY>> {
     void deleteRGRecord(uint32_t rowId, uint64_t timestamp);
     uint64_t* getRGVisibilityBitmap(uint64_t timestamp);
 
-    void collectRGGarbage(uint64_t timestamp);
+    std::vector<uint64_t> collectRGGarbage(uint64_t timestamp);
 
     uint64_t getBitmapSize() const;
 

cpp/pixels-retina/include/RGVisibilityJni.h

@@ -97,7 +97,7 @@ class TileVisibility : public pixels::RetinaBase<TileVisibility<CAPACITY>> {
     ~TileVisibility() override;
     void deleteTileRecord(uint16_t rowId, uint64_t ts);
     void getTileVisibilityBitmap(uint64_t ts, uint64_t* outBitmap) const;
-    void collectTileGarbage(uint64_t ts);
+    void collectTileGarbage(uint64_t ts, uint64_t* gcSnapshotBitmap);
 
   private:
     TileVisibility(const TileVisibility &) = delete;

cpp/pixels-retina/include/TileVisibility.h

@@ -62,11 +62,14 @@ RGVisibility<CAPACITY>::~RGVisibility() {
 }
 
 template<size_t CAPACITY>
-void RGVisibility<CAPACITY>::collectRGGarbage(uint64_t timestamp) {
-// TileVisibility::collectTileGarbage uses COW + Epoch, so it's safe to call concurrently
-    for (uint64_t i = 0; i < tileCount; i++) {
-        tileVisibilities[i].collectTileGarbage(timestamp);
+std::vector<uint64_t> RGVisibility<CAPACITY>::collectRGGarbage(uint64_t timestamp) {
+    size_t totalWords = tileCount * BITMAP_SIZE_PER_TILE_VISIBILITY;
+    std::vector<uint64_t> rgSnapshot(totalWords, 0);
+    for (uint32_t t = 0; t < tileCount; t++) {
+        tileVisibilities[t].collectTileGarbage(timestamp,
+            rgSnapshot.data() + t * BITMAP_SIZE_PER_TILE_VISIBILITY);
     }
+    return rgSnapshot;
 }
 
 template<size_t CAPACITY>

cpp/pixels-retina/lib/RGVisibility.cpp

@@ -129,15 +129,20 @@ JNIEXPORT jlongArray JNICALL Java_io_pixelsdb_pixels_retina_RGVisibility_getVisi
 /*
  * Class:     io_pixelsdb_pixels_retina_RGVisibility
  * Method:    garbageCollect
- * Signature: (JJ)V
+ * Signature: (JJ)[J
  */
-JNIEXPORT void JNICALL Java_io_pixelsdb_pixels_retina_RGVisibility_garbageCollect
+JNIEXPORT jlongArray JNICALL Java_io_pixelsdb_pixels_retina_RGVisibility_garbageCollect
   (JNIEnv* env, jobject, jlong timestamp, jlong handle) {
     try {
         auto* rgVisibility = reinterpret_cast<RGVisibilityInstance*>(handle);
-        rgVisibility->collectRGGarbage(timestamp);
+        std::vector<uint64_t> snapshot = rgVisibility->collectRGGarbage(timestamp);
+        jlongArray result = env->NewLongArray(snapshot.size());
+        env->SetLongArrayRegion(result, 0, snapshot.size(),
+            reinterpret_cast<const jlong*>(snapshot.data()));
+        return result;
     } catch (const std::exception& e) {
         env->ThrowNew(env->FindClass("java/lang/RuntimeException"), e.what());
+        return nullptr;
     }
 }
 

cpp/pixels-retina/lib/RGVisibilityJni.cpp

@@ -238,7 +238,7 @@ void TileVisibility<CAPACITY>::getTileVisibilityBitmap(uint64_t ts, uint64_t* ou
 }
 
 template<size_t CAPACITY>
-void TileVisibility<CAPACITY>::collectTileGarbage(uint64_t ts) {
+void TileVisibility<CAPACITY>::collectTileGarbage(uint64_t ts, uint64_t* gcSnapshotBitmap) {
     // Drain the pending retirement slot left by deleteTileRecord's empty-chain path.
     VersionedData<CAPACITY>* pending = pendingRetire.exchange(nullptr, std::memory_order_acquire);
     if (pending) {
@@ -248,7 +248,12 @@ void TileVisibility<CAPACITY>::collectTileGarbage(uint64_t ts) {
 
     // Load old version
     VersionedData<CAPACITY>* oldVer = currentVersion.load(std::memory_order_acquire);
-    if (ts <= oldVer->baseTimestamp) return;
+
+    // Early return A: safeGcTs <= baseTimestamp, nothing to compact
+    if (ts <= oldVer->baseTimestamp) {
+        std::memcpy(gcSnapshotBitmap, oldVer->baseBitmap, NUM_WORDS * sizeof(uint64_t));
+        return;
+    }
 
     // Find the last block that should be compacted
     DeleteIndexBlock *blk = oldVer->head;
@@ -275,7 +280,22 @@ void TileVisibility<CAPACITY>::collectTileGarbage(uint64_t ts) {
         blk = blk->next.load(std::memory_order_acquire);
     }
 
-    if (!lastFullBlk) return;
+    // Early return B: no compactable block
+    if (!lastFullBlk) {
+        std::memcpy(gcSnapshotBitmap, oldVer->baseBitmap, NUM_WORDS * sizeof(uint64_t));
+        if (oldVer->head) {
+            auto* tailSnap = tail.load(std::memory_order_acquire);
+            size_t tailUsedSnap = tailUsed.load(std::memory_order_acquire);
+            size_t cnt = (oldVer->head == tailSnap) ? tailUsedSnap : DeleteIndexBlock::BLOCK_CAPACITY;
+            for (size_t i = 0; i < cnt; i++) {
+                uint64_t item = oldVer->head->items[i];
+                if (item == 0) break;
+                if (extractTimestamp(item) <= ts) SET_BITMAP_BIT(gcSnapshotBitmap, extractRowId(item));
+                else break;
+            }
+        }
+        return;
+    }
 
     // Create new version with Copy-on-Write
     // Manually compute the new base bitmap from oldVer
@@ -304,8 +324,22 @@ void TileVisibility<CAPACITY>::collectTileGarbage(uint64_t ts) {
         blk = blk->next.load(std::memory_order_acquire);
     }
 
-    // Get new head and break the chain to avoid double-free
+    // Compact path: build gcSnapshotBitmap by scanning the boundary block
     DeleteIndexBlock* newHead = lastFullBlk->next.load(std::memory_order_acquire);
+    std::memcpy(gcSnapshotBitmap, newBaseBitmap, NUM_WORDS * sizeof(uint64_t));
+    if (newHead) {
+        auto* tailSnap = tail.load(std::memory_order_acquire);
+        size_t tailUsedSnap = tailUsed.load(std::memory_order_acquire);
+        size_t cnt = (newHead == tailSnap) ? tailUsedSnap : DeleteIndexBlock::BLOCK_CAPACITY;
+        for (size_t i = 0; i < cnt; i++) {
+            uint64_t item = newHead->items[i];
+            if (item == 0) break;
+            if (extractTimestamp(item) <= ts) SET_BITMAP_BIT(gcSnapshotBitmap, extractRowId(item));
+            else break;
+        }
+    }
+
+    // Break the chain to avoid double-free
     lastFullBlk->next.store(nullptr, std::memory_order_release);
 
     // Create new version with new head - this is the atomic COW update