Support CH

Author: AntiO2

pixels-core/src/main/java/io/pixelsdb/pixels/core/vector/ByteColumnVector.java

@@ -75,7 +75,7 @@ public void add(byte value)
     @Override
     public void add(byte[] value)
     {
-        if(checkBytesNull(value))
+        if (checkBytesNull(value))
         {
             return;
         }

pixels-core/src/main/java/io/pixelsdb/pixels/core/vector/ColumnVector.java

@@ -515,14 +515,14 @@ protected void deserializeBase(ColumnVectorBaseFlat base)
      */
     protected boolean checkBytesNull(byte[] value)
     {
-        if (value == null)
+        if (value == null || value.length == 0)
         {
             if(writeIndex >= getLength())
             {
                 ensureSize(writeIndex * 2, true);
-                isNull[writeIndex++] = true;
-                return true;
             }
+            isNull[writeIndex++] = true;
+            return true;
         }
         return false;
     }

pixels-index/pixels-index-rocksdb/src/main/java/io/pixelsdb/pixels/index/rocksdb/RocksDBFactory.java

@@ -167,8 +167,8 @@ private static ColumnFamilyDescriptor createCFDescriptor(byte[] name, Integer ke
 
         long blockSize = Long.parseLong(config.getProperty("index.rocksdb.block.size"));
         BlockBasedTableConfig tableConfig = new BlockBasedTableConfig()
-                .setFilterPolicy(new BloomFilter(10, false))
-                .setWholeKeyFiltering(false)
+                .setFilterPolicy(new BloomFilter(20, false))
+                .setWholeKeyFiltering(true)
                 .setBlockSize(blockSize)
                 .setCacheIndexAndFilterBlocks(true)
                 .setPinL0FilterAndIndexBlocksInCache(true)
@@ -188,7 +188,7 @@ private static ColumnFamilyDescriptor createCFDescriptor(byte[] name, Integer ke
         int fixedLengthPrefix = Integer.parseInt(config.getProperty("index.rocksdb.prefix.length"));
         if(keyLen != null)
         {
-            fixedLengthPrefix = keyLen + Long.BYTES; // key buffer + index id
+            fixedLengthPrefix = Long.BYTES + keyLen; // index id + key buffer
         }
         CompactionStyle compactionStyle = CompactionStyle.valueOf(config.getProperty("index.rocksdb.compaction.style"));
 
@@ -200,7 +200,8 @@ private static ColumnFamilyDescriptor createCFDescriptor(byte[] name, Integer ke
                 .setWriteBufferSize(writeBufferSize)
                 .setMaxWriteBufferNumber(maxWriteBufferNumber)
                 .setMinWriteBufferNumberToMerge(minWriteBufferNumberToMerge)
-                .setMemtablePrefixBloomSizeRatio(0.1)
+                .setMemtablePrefixBloomSizeRatio(0.2)
+                .setOptimizeFiltersForHits(false)
                 .setTableFormatConfig(tableConfig)
                 .setLevel0FileNumCompactionTrigger(level0FileNumCompactionTrigger)
                 .setMaxBytesForLevelBase(maxBytesForLevelBase)

pixels-index/pixels-index-rocksdb/src/main/java/io/pixelsdb/pixels/index/rocksdb/RocksDBIndex.java

@@ -98,7 +98,9 @@ public long getUniqueRowIdInternal(IndexProto.IndexKey key) throws SinglePointIn
             throw new SinglePointIndexException("getUniqueRowId should only be called on unique index");
         }
         ReadOptions readOptions = RocksDBThreadResources.getReadOptions();
-        readOptions.setPrefixSameAsStart(true);
+        readOptions.setPrefixSameAsStart(true)
+                    .setTotalOrderSeek(false)
+                    .setVerifyChecksums(false);
         ByteBuffer keyBuffer = toKeyBuffer(key);
         long rowId = -1L;
         try (RocksIterator iterator = rocksDB.newIterator(columnFamilyHandle, readOptions))
@@ -131,7 +133,8 @@ public List<Long> getRowIds(IndexProto.IndexKey key) throws SinglePointIndexExce
             return ImmutableList.of(getUniqueRowId(key));
         }
         Set<Long> rowIds = new HashSet<>();
-        ReadOptions readOptions = RocksDBThreadResources.getReadOptions();
+//        ReadOptions readOptions = RocksDBThreadResources.getReadOptions();
+        ReadOptions readOptions = new ReadOptions();
         readOptions.setPrefixSameAsStart(true);
         ByteBuffer keyBuffer = toKeyBuffer(key);
         // use RocksDB iterator for prefix search

pixels-index/pixels-index-rocksdb/src/test/java/io/pixelsdb/pixels/index/rocksdb/TestRocksDB.java

@@ -25,7 +25,6 @@
 
 import java.util.*;
 import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.stream.Collectors;
@@ -350,50 +349,46 @@ public void testBasic()
     @Test
     public void testFullCompaction() throws RocksDBException
     {
-        // Define the list of RocksDB paths
+        // List of RocksDB storage paths
         List<String> dbPaths = new ArrayList<>();
         dbPaths.add("/home/ubuntu/disk6/collected_indexes/realtime-pixels-retina/" + "/rocksdb");
-        for (int i = 2; i <= 8; i++)
-        {
-            dbPaths.add(
-                    "/home/ubuntu/disk6/collected_indexes/realtime-pixels-retina-" + i + "/rocksdb"
-            );
-        }
+//        for (int i = 2; i <= 8; i++)
+//        {
+//            dbPaths.add(
+//                    "/home/ubuntu/disk6/collected_indexes/realtime-pixels-retina-" + i + "/rocksdb"
+//            );
+//        }
 
         long totalStart = System.currentTimeMillis();
         System.out.println("Starting parallel compaction for " + dbPaths.size() + " databases.");
-        // Define parallelism level (e.g., number of disks or cores)
-        int parallelism = 4;
-        ExecutorService executor = Executors.newFixedThreadPool(parallelism);
+
+        // Parallelism level for multiple databases (usually matches the number of disks)
+        int dbParallelism = 4;
+        ExecutorService dbExecutor = Executors.newFixedThreadPool(dbParallelism);
+
         try
         {
+            // Map each database path to an asynchronous compaction task
             List<CompletableFuture<Void>> futures = dbPaths.stream()
                     .map(dbPath -> CompletableFuture.runAsync(() ->
                     {
                         try
                         {
                             executeSingleDbCompaction(dbPath);
-                        } catch (RocksDBException e)
+                        }
+                        catch (RocksDBException e)
                         {
-                            throw new RuntimeException(e);
+                            throw new RuntimeException("Compaction failed for " + dbPath, e);
                         }
-                    }, executor))
+                    }, dbExecutor))
                     .collect(Collectors.toList());
 
-            // Block until all compaction tasks are complete
-            for(CompletableFuture<Void> future: futures)
-            {
-                future.get();
-            }
-        } catch (ExecutionException e)
-        {
-            throw new RuntimeException(e);
-        } catch (InterruptedException e)
-        {
-            throw new RuntimeException(e);
-        } finally
+            // Wait for all DB-level tasks to finish
+            CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
+        }
+        finally
         {
-            executor.shutdown();
+            dbExecutor.shutdown();
         }
 
         long totalEnd = System.currentTimeMillis();
@@ -402,36 +397,59 @@ public void testFullCompaction() throws RocksDBException
 
     private void executeSingleDbCompaction(String dbPath) throws RocksDBException
     {
-        long openStart = System.currentTimeMillis();
-        System.out.println("Thread [" + Thread.currentThread().getName() + "] Start Open: " + dbPath);
+        long startTime = System.currentTimeMillis();
+        System.out.println("Thread [" + Thread.currentThread().getName() + "] Opening DB: " + dbPath);
+
         RocksDB rocksDB;
         Map<String, ColumnFamilyHandle> cfHandles;
 
-        long start = System.currentTimeMillis();
+        // Synchronized block to safely initialize the RocksDB instance and retrieve handles
         synchronized (this)
         {
             rocksDB = RocksDBFactory.createRocksDB(dbPath);
-            long openEnd = System.currentTimeMillis();
-            System.out.println("Open Duration: " + (openEnd - openStart) + "ms\tPath: " + dbPath);
             cfHandles = new HashMap<>(RocksDBFactory.getAllCfHandles());
+            // Clear handles in the factory to prevent memory leaks or reuse conflicts
             RocksDBFactory.clearCFHandles();
         }
 
-        System.out.println("Path: " + dbPath + " | Column Family Count: " + cfHandles.size());
-        // Iterate through each Column Family for manual compaction
-        for (Map.Entry<String, ColumnFamilyHandle> entry : cfHandles.entrySet())
+        System.out.println("Path: " + dbPath + " | CF Count: " + cfHandles.size());
+
+        // Use parallel stream to compact multiple Column Families simultaneously
+        // This allows RocksDB background threads to work on different CFs at once
+        cfHandles.entrySet().parallelStream().forEach(entry ->
         {
             String cfName = entry.getKey();
             ColumnFamilyHandle handle = entry.getValue();
 
-            System.out.println("Compacting CF [" + cfName + "] in " + dbPath);
+            try
+            {
+                System.out.println("Compacting CF [" + cfName + "] in " + dbPath);
 
-            // This is a blocking call per Column Family
-            rocksDB.compactRange(handle);
-            handle.close();
-        }
-        long end = System.currentTimeMillis();
-        System.out.println("SUCCESS: Compaction Duration: " + (end - start) + "ms\tPath: " + dbPath);
+                // Configure CompactRangeOptions to force a rewrite of the bottommost level
+                // This is mandatory to apply the new 20-bit Bloom Filter configuration
+                CompactRangeOptions options = new CompactRangeOptions()
+                        .setBottommostLevelCompaction(CompactRangeOptions.BottommostLevelCompaction.kForce)
+                        .setMaxSubcompactions(4) // Parallel threads within a single CF compaction job
+                        .setAllowWriteStall(true);
+
+                // Synchronous call per CF, but parallelized by the parallelStream
+                rocksDB.compactRange(handle, null, null, options);
+
+                System.out.println("Finished CF [" + cfName + "] in " + dbPath);
+            }
+            catch (RocksDBException e)
+            {
+                System.err.println("Error during compaction for CF " + cfName + ": " + e.getMessage());
+            }
+            finally
+            {
+                // Close handles to release native memory
+                handle.close();
+            }
+        });
+
+        long endTime = System.currentTimeMillis();
+        System.out.println("SUCCESS: Path: " + dbPath + " | Duration: " + (endTime - startTime) + "ms");
         rocksDB.close();
     }
 }