[kv] support shared block cache for kv tablets

Author: platinumhamburg

fluss-common/src/main/java/com/alibaba/fluss/config/ConfigOptions.java

@@ -1535,6 +1535,19 @@ public class ConfigOptions {
                             "The amount of the cache for data blocks in RocksDB. "
                                     + "The default block-cache size is `8MB`.");
 
+    public static final ConfigOption<Boolean> KV_SHARED_BLOCK_CACHE_ENABLED =
+            key("kv.rocksdb.shared-block-cache.enabled")
+                    .booleanType()
+                    .defaultValue(false)
+                    .withDescription(
+                            "Whether to enable the shared block cache across all column families.");
+
+    public static final ConfigOption<MemorySize> KV_SHARED_BLOCK_CACHE_SIZE =
+            key("kv.rocksdb.shared-block-cache.size")
+                    .memoryType()
+                    .defaultValue(MemorySize.parse("4gb"))
+                    .withDescription("The size of the shared block cache if enabled.");
+
     public static final ConfigOption<Boolean> KV_USE_BLOOM_FILTER =
             key("kv.rocksdb.use-bloom-filter")
                     .booleanType()

fluss-server/src/main/java/com/alibaba/fluss/server/kv/KvManager.java

@@ -33,6 +33,7 @@
 import com.alibaba.fluss.metadata.TableInfo;
 import com.alibaba.fluss.metadata.TablePath;
 import com.alibaba.fluss.server.TabletManagerBase;
+import com.alibaba.fluss.server.kv.rocksdb.RocksDBSharedResource;
 import com.alibaba.fluss.server.kv.rowmerger.RowMerger;
 import com.alibaba.fluss.server.log.LogManager;
 import com.alibaba.fluss.server.log.LogTablet;
@@ -86,6 +87,9 @@ public final class KvManager extends TabletManagerBase {
 
     private final FileSystem remoteFileSystem;
 
+    /** Global shared RocksDB resource. */
+    private final RocksDBSharedResource sharedResource;
+
     private KvManager(
             File dataDir,
             Configuration conf,
@@ -100,6 +104,9 @@ private KvManager(
         this.zkClient = zkClient;
         this.remoteKvDir = FlussPaths.remoteKvDir(conf);
         this.remoteFileSystem = remoteKvDir.getFileSystem();
+
+        // Initialize global shared resource
+        this.sharedResource = RocksDBSharedResource.getInstance(conf);
     }
 
     public static KvManager create(
@@ -129,6 +136,16 @@ public void shutdown() {
                 LOG.warn("Exception while closing kv tablet {}.", kvTablet.getTableBucket(), e);
             }
         }
+
+        // Clean up shared resources
+        if (sharedResource != null) {
+            try {
+                sharedResource.close();
+            } catch (Exception e) {
+                LOG.warn("Exception while closing shared RocksDB resource.", e);
+            }
+        }
+
         arrowBufferAllocator.close();
         memorySegmentPool.close();
         LOG.info("Shut down KvManager complete.");

fluss-server/src/main/java/com/alibaba/fluss/server/kv/rocksdb/RocksDBResourceContainer.java

@@ -26,6 +26,7 @@
 
 import org.rocksdb.BlockBasedTableConfig;
 import org.rocksdb.BloomFilter;
+import org.rocksdb.Cache;
 import org.rocksdb.ColumnFamilyOptions;
 import org.rocksdb.CompactionStyle;
 import org.rocksdb.CompressionType;
@@ -74,6 +75,9 @@ public class RocksDBResourceContainer implements AutoCloseable {
     /** The handles to be closed when the container is closed. */
     private final ArrayList<AutoCloseable> handlesToClose;
 
+    /** Global shared RocksDB resource. */
+    private final RocksDBSharedResource sharedResource;
+
     @VisibleForTesting
     RocksDBResourceContainer() {
         this(new Configuration(), null, false);
@@ -96,6 +100,10 @@ public RocksDBResourceContainer(
         this.enableStatistics = enableStatistics;
 
         this.handlesToClose = new ArrayList<>();
+
+        // Get global shared resource and increase reference count
+        this.sharedResource = RocksDBSharedResource.getInstance(configuration);
+        this.sharedResource.acquire();
     }
 
     /** Gets the RocksDB {@link DBOptions} to be used for RocksDB instances. */
@@ -153,10 +161,21 @@ public ReadOptions getReadOptions() {
         return opt;
     }
 
+    @VisibleForTesting
+    @Nullable
+    Cache getSharedBlockCache() {
+        return sharedResource.getSharedBlockCache();
+    }
+
     @Override
     public void close() throws Exception {
         handlesToClose.forEach(IOUtils::closeQuietly);
         handlesToClose.clear();
+
+        // Release reference to shared resource
+        if (sharedResource != null) {
+            sharedResource.release();
+        }
     }
 
     /** Create a {@link DBOptions} for RocksDB, including some common settings. */
@@ -253,14 +272,22 @@ private ColumnFamilyOptions setColumnFamilyOptionsFromConfigurableOptions(
             }
         }
 
+        Cache sharedCache = getSharedBlockCache();
+        if (sharedCache != null) {
+            blockBasedTableConfig.setBlockCache(sharedCache);
+        } else {
+            // Original per-CF block cache settings
+            blockBasedTableConfig.setBlockCacheSize(
+                    internalGetOption(ConfigOptions.KV_BLOCK_CACHE_SIZE).getBytes());
+        }
+
         blockBasedTableConfig.setBlockSize(
                 internalGetOption(ConfigOptions.KV_BLOCK_SIZE).getBytes());
 
         blockBasedTableConfig.setMetadataBlockSize(
                 internalGetOption(ConfigOptions.KV_METADATA_BLOCK_SIZE).getBytes());
 
-        blockBasedTableConfig.setBlockCacheSize(
-                internalGetOption(ConfigOptions.KV_BLOCK_CACHE_SIZE).getBytes());
+        blockBasedTableConfig.setPinL0FilterAndIndexBlocksInCache(true);
 
         if (internalGetOption(ConfigOptions.KV_USE_BLOOM_FILTER)) {
             final double bitsPerKey = internalGetOption(ConfigOptions.KV_BLOOM_FILTER_BITS_PER_KEY);