feat(retina)!: introduce recovery checkpoint subsystem

Author: gengdy1545

pixels-common/src/main/java/io/pixelsdb/pixels/common/transaction/TransService.java

@@ -20,7 +20,6 @@
 package io.pixelsdb.pixels.common.transaction;
 
 import com.google.common.collect.ImmutableList;
-import com.google.protobuf.Empty;
 import io.grpc.ManagedChannel;
 import io.grpc.ManagedChannelBuilder;
 import io.pixelsdb.pixels.common.error.ErrorCode;
@@ -498,12 +497,22 @@ public boolean bindExternalTraceId(long transId, String externalTraceId) throws
         return true;
     }
 
-    public long getSafeGcTimestamp() throws TransException
+    /**
+     * Get the safe upper bound (inclusive) for folding DELETE timestamps into
+     * the visibility base bitmap.
+     *
+     * @param includeRunningQueries whether the returned timestamp must remain safe for live running queries
+     */
+    public long getSafeVisibilityFoldingTimestamp(boolean includeRunningQueries) throws TransException
     {
-        TransProto.GetSafeGcTimestampResponse response = this.stub.getSafeGcTimestamp(Empty.getDefaultInstance());
+        TransProto.GetSafeVisibilityFoldingTimestampRequest request =
+                TransProto.GetSafeVisibilityFoldingTimestampRequest.newBuilder()
+                        .setIncludeRunningQueries(includeRunningQueries).build();
+        TransProto.GetSafeVisibilityFoldingTimestampResponse response =
+                this.stub.getSafeVisibilityFoldingTimestamp(request);
         if (response.getErrorCode() != ErrorCode.SUCCESS)
         {
-            throw new TransException("failed to get safe garbage collection timestamp"
+            throw new TransException("failed to get safe visibility folding timestamp, error code="
                     + response.getErrorCode());
         }
         return response.getTimestamp();

pixels-common/src/main/java/io/pixelsdb/pixels/common/utils/EtcdUtil.java

@@ -24,7 +24,11 @@
 import io.etcd.jetcd.KeyValue;
 import io.etcd.jetcd.Watch;
 import io.etcd.jetcd.kv.PutResponse;
+import io.etcd.jetcd.kv.TxnResponse;
 import io.etcd.jetcd.lease.LeaseGrantResponse;
+import io.etcd.jetcd.op.Cmp;
+import io.etcd.jetcd.op.CmpTarget;
+import io.etcd.jetcd.op.Op;
 import io.etcd.jetcd.options.DeleteOption;
 import io.etcd.jetcd.options.GetOption;
 import io.etcd.jetcd.options.PutOption;
@@ -260,6 +264,33 @@ public long putKeyValueWithLeaseId(String key, String value, long leaseId)
         return 0L;
     }
 
+    /**
+     * Atomic compare-and-swap put.
+     * 
+     * @param key
+     * @param expectedValue
+     * @param newValue
+     * @return true if the txn committed; false if CAS failed
+     */
+    public boolean compareAndPut(String key, String expectedValue, String newValue)
+            throws ExecutionException, InterruptedException
+    {
+        ByteSequence keyBs = ByteSequence.from(key, StandardCharsets.UTF_8);
+        Cmp cmp = (expectedValue == null)
+                ? new Cmp(keyBs, Cmp.Op.EQUAL, CmpTarget.version(0L))
+                : new Cmp(keyBs, Cmp.Op.EQUAL, CmpTarget.value(
+                        ByteSequence.from(expectedValue, StandardCharsets.UTF_8)));
+        Op putOp = Op.put(keyBs,
+                ByteSequence.from(newValue, StandardCharsets.UTF_8),
+                PutOption.DEFAULT);
+        TxnResponse resp = this.client.getKVClient().txn()
+                .If(cmp)
+                .Then(putOp)
+                .commit()
+                .get();
+        return resp.isSucceeded();
+    }
+
     /**
      * delete key-value by key.
      *

pixels-common/src/main/java/io/pixelsdb/pixels/common/utils/RetinaUtils.java

@@ -29,8 +29,8 @@
 
 public class RetinaUtils
 {
-    public static final String CHECKPOINT_PREFIX_GC = "vis_gc_";
     public static final String CHECKPOINT_PREFIX_OFFLOAD = "vis_offload_";
+    public static final String CHECKPOINT_PREFIX_RECOVERY = "recovery_";
     public static final String CHECKPOINT_SUFFIX = ".bin";
 
     private static volatile RetinaUtils instance;
@@ -132,12 +132,12 @@ public static String getCheckpointPrefix(String typePrefix, String hostname)
     }
 
     /**
-     * Builds the checkpoint file path from a directory, prefix, hostname and timestamp.
+     * Builds the checkpoint file path from a directory, type prefix, hostname and identifier timestamp.
      *
      * @param checkpointDir directory where checkpoint files reside (may or may not end with '/')
-     * @param prefix        {@link #CHECKPOINT_PREFIX_GC} or {@link #CHECKPOINT_PREFIX_OFFLOAD}
+     * @param prefix        {@link #CHECKPOINT_PREFIX_OFFLOAD} or {@link #CHECKPOINT_PREFIX_RECOVERY}
      * @param hostname      the retina host name
-     * @param timestamp     the GC or offload timestamp
+     * @param timestamp     the checkpoint identifier timestamp (offload ts for offload, applied ts for recovery)
      */
     public static String buildCheckpointPath(String checkpointDir, String prefix, String hostname, long timestamp)
     {

pixels-common/src/main/resources/pixels.properties

@@ -291,8 +291,6 @@ retina.buffer.flush.count=20
 retina.buffer.flush.interval=30
 # interval in seconds for retina visibility garbage
 retina.gc.interval=300
-# number of threads for retina checkpoint
-retina.checkpoint.threads=4
 # retina buffer reader prefetch threads num
 retina.reader.prefetch.threads=8
 # retina service init threads num
@@ -305,8 +303,12 @@ retina.upsert-mode.enabled=false
 pixels.transaction.offload.threshold=1800
 # lease duration for retina offload cache in seconds, default 600s
 retina.offload.cache.lease.duration=600
-# snapshot storage directory
-retina.checkpoint.dir=file:///tmp/pixels-checkpoints
+# number of threads for offload checkpoint writers
+retina.offload.checkpoint.threads=4
+# storage URI for long-running query offload visibility snapshots; cleared on Retina startup
+retina.offload.checkpoint.dir=file:///tmp/pixels-offload-checkpoints
+# storage URI for recovery checkpoint body objects (one body per node per round)
+retina.recovery.checkpoint.dir=file:///tmp/pixels-recovery-checkpoints
 # set to true to enable storage GC (rewrites high-deletion-ratio files to reclaim space)
 retina.storage.gc.enabled=false
 # invalidRatio must be strictly greater than this value for a file to be a GC candidate

pixels-core/src/test/java/io/pixelsdb/pixels/core/reader/TestVisibilityCheckpointCache.java

@@ -46,7 +46,7 @@ public class TestVisibilityCheckpointCache
     @Before
     public void setUp() throws IOException
     {
-        testCheckpointDir = ConfigFactory.Instance().getProperty("retina.checkpoint.dir");
+        testCheckpointDir = ConfigFactory.Instance().getProperty("retina.offload.checkpoint.dir");
         storage = StorageFactory.Instance().getStorage(testCheckpointDir);
 
         if (!storage.exists(testCheckpointDir))
@@ -86,7 +86,7 @@ private void createDummyCheckpoint(String path, int numFiles, int rgsPerFile, lo
     public void testCacheLoading() throws Exception
     {
         long timestamp = 1000L;
-        String checkpointPath = resolve(testCheckpointDir, "vis_gc_tencent_100.bin");
+        String checkpointPath = resolve(testCheckpointDir, "vis_offload_tencent_100.bin");
         long[] dummyBitmap = new long[]{0x1L, 0x2L};
         createDummyCheckpoint(checkpointPath, 1, 1, dummyBitmap);
 

pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/retina/RetinaServerImpl.java

@@ -114,7 +114,7 @@ public RetinaServerImpl()
     private void initializeRetinaResources() throws Exception
     {
         logger.info("Pre-loading checkpoints...");
-        this.retinaResourceManager.recoverCheckpoints();
+        this.retinaResourceManager.recoverOffloadCheckpoints();
 
         List<Schema> schemas = this.metadataService.getSchemas();
         for (Schema schema : schemas)
@@ -825,7 +825,7 @@ public void queryVisibility(RetinaProto.QueryVisibilityRequest request,
                     .newBuilder()
                     .setHeader(headerBuilder.build());
 
-            String checkpointPath = this.retinaResourceManager.getCheckpointPath(timestamp);
+            String checkpointPath = this.retinaResourceManager.getOffloadCheckpointPath(timestamp);
             if (checkpointPath != null)
             {
                 responseBuilder.setCheckpointPath(checkpointPath);

pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/transaction/TransServiceImpl.java

@@ -629,14 +629,18 @@ public void dumpTrans(TransProto.DumpTransRequest request,
     }
 
     @Override
-    public void getSafeGcTimestamp(com.google.protobuf.Empty request,
-                                   StreamObserver<TransProto.GetSafeGcTimestampResponse> responseObserver)
+    public void getSafeVisibilityFoldingTimestamp(TransProto.GetSafeVisibilityFoldingTimestampRequest request,
+                                                  StreamObserver<TransProto.GetSafeVisibilityFoldingTimestampResponse> responseObserver)
     {
-        long safeTs = Math.max(0, lowWatermark.get() - 1);
-        TransProto.GetSafeGcTimestampResponse response = TransProto.GetSafeGcTimestampResponse.newBuilder()
-                .setErrorCode(ErrorCode.SUCCESS)
-                .setTimestamp(safeTs)
-                .build();
+        long writerSafeTs = Math.max(0, highWatermark.get() - 1);
+        long safeTs = request.getIncludeRunningQueries()
+                ? Math.min(lowWatermark.get(), writerSafeTs)
+                : writerSafeTs;
+        TransProto.GetSafeVisibilityFoldingTimestampResponse response =
+                TransProto.GetSafeVisibilityFoldingTimestampResponse.newBuilder()
+                        .setErrorCode(ErrorCode.SUCCESS)
+                        .setTimestamp(safeTs)
+                        .build();
         responseObserver.onNext(response);
         responseObserver.onCompleted();
     }

pixels-daemon/src/test/java/io/pixelsdb/pixels/daemon/retina/TestRetinaServer.java

@@ -98,7 +98,7 @@ public void testRetinaServerImplInitializationFailureIsFailClosed() throws Excep
             assertTrue(e.getMessage().contains("Failed to initialize RetinaServerImpl"));
         }
 
-        verify(resourceManager).recoverCheckpoints();
+        verify(resourceManager).recoverOffloadCheckpoints();
         verify(resourceManager, never()).startBackgroundGc();
     }
 
@@ -137,7 +137,7 @@ public void testRetinaServerImplStartsBackgroundGcAfterSuccessfulInitialization(
         doAnswer(invocation -> {
             lifecycleEvents.add("recover");
             return null;
-        }).when(resourceManager).recoverCheckpoints();
+        }).when(resourceManager).recoverOffloadCheckpoints();
         doAnswer(invocation -> {
             lifecycleEvents.add("visibility:" + invocation.getArgument(0));
             return null;
@@ -188,7 +188,7 @@ public void testRetinaServerImplBackgroundGcStartFailureIsFailClosed() throws Ex
         }
 
         InOrder inOrder = inOrder(resourceManager);
-        inOrder.verify(resourceManager).recoverCheckpoints();
+        inOrder.verify(resourceManager).recoverOffloadCheckpoints();
         inOrder.verify(resourceManager).startBackgroundGc();
     }
 

pixels-retina/pom.xml

@@ -88,7 +88,6 @@
         <dependency>
             <groupId>io.etcd</groupId>
             <artifactId>jetcd-core</artifactId>
-            <scope>test</scope>
         </dependency>
 
         <dependency>