[Issue #1220] add a toggle switch to control whether Retina is enabled (#1227)

gengdy1545 · web-flow · commit 4fb1fbc0537e · 2025-12-01T00:15:42.000+08:00
diff --git a/pixels-cli/src/main/java/io/pixelsdb/pixels/cli/executor/CompactExecutor.java b/pixels-cli/src/main/java/io/pixelsdb/pixels/cli/executor/CompactExecutor.java
@@ -214,18 +214,21 @@ public void execute(Namespace ns, String command) throws Exception
         while (!compactExecutor.awaitTermination(100, TimeUnit.SECONDS));
         metadataService.addFiles(compactFiles);
 
-        Iterator<File> fileIterator = compactFiles.iterator();
-        Iterator<Path> pathIterator = compactPaths.iterator();
-        while (fileIterator.hasNext() && pathIterator.hasNext())
+        if (retinaService.isEnabled())
         {
-            File file = fileIterator.next();
-            Path path = pathIterator.next();
-            try
+            Iterator<File> fileIterator = compactFiles.iterator();
+            Iterator<Path> pathIterator = compactPaths.iterator();
+            while (fileIterator.hasNext() && pathIterator.hasNext())
             {
-                retinaService.addVisibility(File.getFilePath(path, file));
-            } catch (RetinaException e)
-            {
-                System.out.println("add visibility for compact file '" + file + "' failed");
+                File file = fileIterator.next();
+                Path path = pathIterator.next();
+                try
+                {
+                    retinaService.addVisibility(File.getFilePath(path, file));
+                } catch (RetinaException e)
+                {
+                    System.out.println("add visibility for compact file '" + file + "' failed");
+                }
             }
         }
 
diff --git a/pixels-cli/src/main/java/io/pixelsdb/pixels/cli/executor/LoadExecutor.java b/pixels-cli/src/main/java/io/pixelsdb/pixels/cli/executor/LoadExecutor.java
@@ -100,11 +100,17 @@ public void execute(Namespace ns, String command) throws Exception
                     NodeProto.NodeInfo nodeInfo = loadedInfo.loadedRetinaNode;
                     if(nodeInfo == null)
                     {
-                        defaultRetinaService.addVisibility(File.getFilePath(path, file));
+                        if (defaultRetinaService.isEnabled())
+                        {
+                            defaultRetinaService.addVisibility(File.getFilePath(path, file));
+                        }
                     } else
                     {
                         RetinaService retinaService = RetinaService.CreateInstance(nodeInfo.getAddress(), retinaServerPort);
-                        retinaService.addVisibility(File.getFilePath(path, file));
+                        if (retinaService.isEnabled())
+                        {
+                            retinaService.addVisibility(File.getFilePath(path, file));
+                        }
                     }
 
                 } catch (RetinaException e)
diff --git a/pixels-common/src/main/java/io/pixelsdb/pixels/common/retina/RetinaService.java b/pixels-common/src/main/java/io/pixelsdb/pixels/common/retina/RetinaService.java
@@ -49,9 +49,11 @@ public class RetinaService
 
     static
     {
-        String retinaHost = ConfigFactory.Instance().getProperty("retina.server.host");
-        int retinaPort = Integer.parseInt(ConfigFactory.Instance().getProperty("retina.server.port"));
-        defaultInstance = new RetinaService(retinaHost, retinaPort);
+        ConfigFactory config = ConfigFactory.Instance();
+        String retinaHost = config.getProperty("retina.server.host");
+        int retinaPort = Integer.parseInt(config.getProperty("retina.server.port"));
+        boolean enabled = Boolean.parseBoolean(config.getProperty("retina.enable"));
+        defaultInstance = new RetinaService(retinaHost, retinaPort, enabled);
         ShutdownHookManager.Instance().registerShutdownHook(RetinaService.class, false, () -> {
             try
             {
@@ -91,36 +93,54 @@ public static RetinaService Instance()
     public static synchronized RetinaService CreateInstance(String host, int port)
     {
         HostAddress address = HostAddress.fromParts(host, port);
+        // For other instances, we also follow the global configuration.
+        String retinaEnable = ConfigFactory.Instance().getProperty("retina.enable");
+        boolean enabled = Boolean.parseBoolean(retinaEnable);
         return otherInstances.computeIfAbsent(
-                address, addr -> new RetinaService(addr.getHostText(), addr.getPort())
+                address, addr -> new RetinaService(addr.getHostText(), addr.getPort(), enabled)
         );
     }
 
     private final ManagedChannel channel;
     private final RetinaWorkerServiceGrpc.RetinaWorkerServiceBlockingStub stub;
     private final RetinaWorkerServiceGrpc.RetinaWorkerServiceStub asyncStub;
     private boolean isShutdown;
+    private final boolean enabled;
 
-    private RetinaService(String host, int port)
+    private RetinaService(String host, int port, boolean enabled)
     {
-        assert (host != null);
-        assert (port > 0 && port <= 65535);
-        this.channel = ManagedChannelBuilder.forAddress(host, port).usePlaintext()
-//                .keepAliveTime(1, TimeUnit.SECONDS)
-//                .keepAliveTimeout(3, TimeUnit.SECONDS)
-//                .keepAliveWithoutCalls(true)
-                .build();
-        this.stub = RetinaWorkerServiceGrpc.newBlockingStub(this.channel);
-        this.asyncStub = RetinaWorkerServiceGrpc.newStub(this.channel);
+        this.enabled = enabled;
+        if (enabled)
+        {
+            assert (host != null);
+            assert (port > 0 && port <= 65535);
+            this.channel = ManagedChannelBuilder.forAddress(host, port).
+                    usePlaintext().build();
+            this.stub = RetinaWorkerServiceGrpc.newBlockingStub(this.channel);
+            this.asyncStub = RetinaWorkerServiceGrpc.newStub(this.channel);
+        } else
+        {
+            this.channel = null;
+            this.stub = null;
+            this.asyncStub = null;
+        }
         this.isShutdown = false;
     }
 
+    public boolean isEnabled()
+    {
+        return enabled;
+    }
+
     private synchronized void shutdown() throws InterruptedException
     {
         if (!this.isShutdown)
         {
-            // Wait for at most 5 seconds, this should be enough to shut down an RPC client.
-            this.channel.shutdown().awaitTermination(5, TimeUnit.SECONDS);
+            if (this.channel != null)
+            {
+                // Wait for at most 5 seconds, this should be enough to shut down an RPC client.
+                this.channel.shutdown().awaitTermination(5, TimeUnit.SECONDS);
+            }
             this.isShutdown = true;
         }
     }
diff --git a/pixels-common/src/main/resources/pixels.properties b/pixels-common/src/main/resources/pixels.properties
@@ -267,6 +267,8 @@ memspl = [1,8,16,32,64]
 
 ### pixels-retina - write buffer flush configuration ###
 
+# set to true to enable pixels-retina
+retina.enable=false
 # number of rows recorded in memTable, must be a multiple of 64
 retina.buffer.memTable.size=10240
 # the scheme of the storage for retina buffer object storage, e.g., s3, minio
diff --git a/pixels-common/src/test/java/io/pixelsdb/pixels/common/retina/TestRetinaService.java b/pixels-common/src/test/java/io/pixelsdb/pixels/common/retina/TestRetinaService.java
@@ -65,6 +65,7 @@ public class TestRetinaService
     @BeforeAll
     public static void setUp() throws MetadataException, InterruptedException, JsonProcessingException
     {
+        Assertions.assertTrue(RetinaService.Instance().isEnabled(), "Retina service should be enabled for tests");
         MetadataService metadataService = MetadataService.Instance();
         Table table = metadataService.getTable(schemaName, tableName);
         List<Column> columns = metadataService.getColumns(schemaName, tableName, false);
diff --git a/pixels-core/src/main/java/io/pixelsdb/pixels/core/reader/PixelsRecordReaderBufferImpl.java b/pixels-core/src/main/java/io/pixelsdb/pixels/core/reader/PixelsRecordReaderBufferImpl.java
@@ -77,6 +77,7 @@ public class PixelsRecordReaderBufferImpl implements PixelsRecordReader
 
     private final long tableId;
     private final String retinaBufferStorageFolder;
+    private final boolean retinaEnabled;
 
     private final TypeDescription typeDescription;
     private final int colNum;
@@ -112,6 +113,7 @@ public PixelsRecordReaderBufferImpl(PixelsReaderOption option,
     {
         ConfigFactory configFactory = ConfigFactory.Instance();
         this.retinaBufferStorageFolder = configFactory.getProperty("retina.buffer.object.storage.folder");
+        this.retinaEnabled = Boolean.parseBoolean(configFactory.getProperty("retina.enable"));
 
         this.option = option;
         this.activeMemtableData = activeMemtableData;
@@ -158,7 +160,7 @@ private void checkBeforeRead() throws IOException
         }
 
         // check retina
-        if (visibilityBitmap.size() != fileIds.size() + 1)
+        if (retinaEnabled && visibilityBitmap != null && visibilityBitmap.size() != fileIds.size() + 1)
         {
             checkValid = false;
             throw new IOException("visibilityBitmap.getSize is " + visibilityBitmap.size() +
@@ -266,7 +268,7 @@ public VectorizedRowBatch readBatch() throws IOException
         for (int i = 0; i < curBatchSize; i++)
         {
             if ((hiddenTimestampVector == null || hiddenTimestampVector.vector[i] <= this.option.getTransTimestamp())
-                    && (!checkBit(visibilityBitmap.get(fileIdIndex), i)))
+                    && (!retinaEnabled || visibilityBitmap == null || !checkBit(visibilityBitmap.get(fileIdIndex), i)))
             {
                 selectedRows.set(i);
                 addedRows++;
diff --git a/pixels-core/src/main/java/io/pixelsdb/pixels/core/reader/PixelsRecordReaderImpl.java b/pixels-core/src/main/java/io/pixelsdb/pixels/core/reader/PixelsRecordReaderImpl.java
@@ -501,7 +501,7 @@ else if (predicate.matchesNone())
         targetRGNum = targetRGIdx;
 
         // query visibility bitmap of target row groups
-        if (this.option.hasValidTransTimestamp())
+        if (this.option.hasValidTransTimestamp() && retinaService.isEnabled())
         {
             try
             {
@@ -1132,7 +1132,7 @@ public VectorizedRowBatch readBatch(int batchSize, boolean reuse)
                 for (int i = 0; i < curBatchSize; i++)
                 {
                     if ((hiddenTimestampVector == null || hiddenTimestampVector.vector[i] <= this.transTimestamp)
-                        && (!checkBit(rgVisibilityBitmaps[curRGIdx], curRowInRG + i)))
+                        && (rgVisibilityBitmaps == null || !checkBit(rgVisibilityBitmaps[curRGIdx], curRowInRG + i)))
                     {
                         selectedRows.set(i);
                         addedRows++;
@@ -1475,4 +1475,4 @@ public ChunkId(int rowGroupId, int columnId, long offset, int length)
             this.length = length;
         }
     }
-}
+}
diff --git a/pixels-core/src/test/java/io/pixelsdb/pixels/core/reader/TestPixelsRecordReaderBufferImpl.java b/pixels-core/src/test/java/io/pixelsdb/pixels/core/reader/TestPixelsRecordReaderBufferImpl.java
@@ -49,6 +49,7 @@ public void testReadBatch() throws RetinaException, IOException
 
 
         RetinaService retinaService = RetinaService.Instance();
+        assert retinaService.isEnabled();
         long timeStamp = 100000;
         RetinaProto.GetWriterBufferResponse superVersion = retinaService.getWriterBuffer(schemaName, tableName, timeStamp);
         TypeDescription typeDescription = null;
diff --git a/pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/metadata/MetadataServiceImpl.java b/pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/metadata/MetadataServiceImpl.java
@@ -363,14 +363,17 @@ public void createTable(MetadataProto.CreateTableRequest request,
                                             for(NodeProto.NodeInfo retinaNode : retinaList)
                                             {
                                                 RetinaService retinaService = RetinaService.CreateInstance(retinaNode.getAddress(), retinaPort);
-                                                if (!retinaService.addWriterBuffer(request.getSchemaName(), request.getTableName()))
+                                                if (retinaService.isEnabled())
                                                 {
-                                                    headerBuilder.setErrorCode(METADATA_ADD_RETINA_BUFFER_FAILED)
-                                                            .setErrorMsg("failed to add retina's writer buffer for table '" +
-                                                                    request.getSchemaName() + "." + request.getTableName() + "'");
-                                                } else
-                                                {
-                                                    headerBuilder.setErrorCode(SUCCESS).setErrorMsg("");
+                                                    if (!retinaService.addWriterBuffer(request.getSchemaName(), request.getTableName()))
+                                                    {
+                                                        headerBuilder.setErrorCode(METADATA_ADD_RETINA_BUFFER_FAILED)
+                                                                .setErrorMsg("failed to add retina's writer buffer for table '" +
+                                                                        request.getSchemaName() + "." + request.getTableName() + "'");
+                                                    } else
+                                                    {
+                                                        headerBuilder.setErrorCode(SUCCESS).setErrorMsg("");
+                                                    }
                                                 }
                                             }
                                         } else

Original file line number	Diff line number	Diff line change
`@@ -100,11 +100,17 @@ public void execute(Namespace ns, String command) throws Exception`
`100`	`100`	`NodeProto.NodeInfo nodeInfo = loadedInfo.loadedRetinaNode;`
`101`	`101`	`if(nodeInfo == null)`
`102`	`102`	`{`
`103`		`- defaultRetinaService.addVisibility(File.getFilePath(path, file));`
	`103`	`+ if (defaultRetinaService.isEnabled())`
	`104`	`+ {`
	`105`	`+ defaultRetinaService.addVisibility(File.getFilePath(path, file));`
	`106`	`+ }`
`104`	`107`	`} else`
`105`	`108`	`{`
`106`	`109`	`RetinaService retinaService = RetinaService.CreateInstance(nodeInfo.getAddress(), retinaServerPort);`
`107`		`- retinaService.addVisibility(File.getFilePath(path, file));`
	`110`	`+ if (retinaService.isEnabled())`
	`111`	`+ {`
	`112`	`+ retinaService.addVisibility(File.getFilePath(path, file));`
	`113`	`+ }`
`108`	`114`	`}`
`109`	`115`
`110`	`116`	`} catch (RetinaException e)`
Original file line number	Diff line number	Diff line change
`@@ -65,6 +65,7 @@ public class TestRetinaService`
`65`	`65`	`@BeforeAll`
`66`	`66`	`public static void setUp() throws MetadataException, InterruptedException, JsonProcessingException`
`67`	`67`	`{`
	`68`	`+ Assertions.assertTrue(RetinaService.Instance().isEnabled(), "Retina service should be enabled for tests");`
`68`	`69`	`MetadataService metadataService = MetadataService.Instance();`
`69`	`70`	`Table table = metadataService.getTable(schemaName, tableName);`
`70`	`71`	`List<Column> columns = metadataService.getColumns(schemaName, tableName, false);`
Original file line number	Diff line number	Diff line change
`@@ -77,6 +77,7 @@ public class PixelsRecordReaderBufferImpl implements PixelsRecordReader`
`77`	`77`
`78`	`78`	`private final long tableId;`
`79`	`79`	`private final String retinaBufferStorageFolder;`
	`80`	`+ private final boolean retinaEnabled;`
`80`	`81`
`81`	`82`	`private final TypeDescription typeDescription;`
`82`	`83`	`private final int colNum;`
`@@ -112,6 +113,7 @@ public PixelsRecordReaderBufferImpl(PixelsReaderOption option,`
`112`	`113`	`{`
`113`	`114`	`ConfigFactory configFactory = ConfigFactory.Instance();`
`114`	`115`	`this.retinaBufferStorageFolder = configFactory.getProperty("retina.buffer.object.storage.folder");`
	`116`	`+ this.retinaEnabled = Boolean.parseBoolean(configFactory.getProperty("retina.enable"));`
`115`	`117`
`116`	`118`	`this.option = option;`
`117`	`119`	`this.activeMemtableData = activeMemtableData;`
`@@ -158,7 +160,7 @@ private void checkBeforeRead() throws IOException`
`158`	`160`	`}`
`159`	`161`
`160`	`162`	`// check retina`
`161`		`- if (visibilityBitmap.size() != fileIds.size() + 1)`
	`163`	`+ if (retinaEnabled && visibilityBitmap != null && visibilityBitmap.size() != fileIds.size() + 1)`
`162`	`164`	`{`
`163`	`165`	`checkValid = false;`
`164`	`166`	`throw new IOException("visibilityBitmap.getSize is " + visibilityBitmap.size() +`
`@@ -266,7 +268,7 @@ public VectorizedRowBatch readBatch() throws IOException`
`266`	`268`	`for (int i = 0; i < curBatchSize; i++)`
`267`	`269`	`{`
`268`	`270`	`if ((hiddenTimestampVector == null \|\| hiddenTimestampVector.vector[i] <= this.option.getTransTimestamp())`
`269`		`- && (!checkBit(visibilityBitmap.get(fileIdIndex), i)))`
	`271`	`+ && (!retinaEnabled \|\| visibilityBitmap == null \|\| !checkBit(visibilityBitmap.get(fileIdIndex), i)))`
`270`	`272`	`{`
`271`	`273`	`selectedRows.set(i);`
`272`	`274`	`addedRows++;`
Original file line number	Diff line number	Diff line change
`@@ -501,7 +501,7 @@ else if (predicate.matchesNone())`
`501`	`501`	`targetRGNum = targetRGIdx;`
`502`	`502`
`503`	`503`	`// query visibility bitmap of target row groups`
`504`		`- if (this.option.hasValidTransTimestamp())`
	`504`	`+ if (this.option.hasValidTransTimestamp() && retinaService.isEnabled())`
`505`	`505`	`{`
`506`	`506`	`try`
`507`	`507`	`{`
`@@ -1132,7 +1132,7 @@ public VectorizedRowBatch readBatch(int batchSize, boolean reuse)`
`1132`	`1132`	`for (int i = 0; i < curBatchSize; i++)`
`1133`	`1133`	`{`
`1134`	`1134`	`if ((hiddenTimestampVector == null \|\| hiddenTimestampVector.vector[i] <= this.transTimestamp)`
`1135`		`- && (!checkBit(rgVisibilityBitmaps[curRGIdx], curRowInRG + i)))`
	`1135`	`+ && (rgVisibilityBitmaps == null \|\| !checkBit(rgVisibilityBitmaps[curRGIdx], curRowInRG + i)))`
`1136`	`1136`	`{`
`1137`	`1137`	`selectedRows.set(i);`
`1138`	`1138`	`addedRows++;`
`@@ -1475,4 +1475,4 @@ public ChunkId(int rowGroupId, int columnId, long offset, int length)`
`1475`	`1475`	`this.length = length;`
`1476`	`1476`	`}`
`1477`	`1477`	`}`
`1478`		`-}`
	`1478`	`+}`