(hudi) branch master updated: feat(flink): add RocksDB partitioned RLI Cache Flink Options (#19641)

[da...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

danny0405 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git


The following commit(s) were added to refs/heads/master by this push:
     new e3ee9600fc5b feat(flink): add RocksDB partitioned RLI Cache Flink Options (#19641)
e3ee9600fc5b is described below

commit e3ee9600fc5be950f3332fd9ffe2413697485fe0
Author: Peter Huang <hu...@gmail.com>
AuthorDate: Tue Aug 25 20:01:51 2026 -0700

    feat(flink): add RocksDB partitioned RLI Cache Flink Options (#19641)
    
    * feat(flink): add RocksDB partitioned RLI Cache Flink Options
    
    Exposes the 7 configuration properties defined by RFC-107 (Support data
    partition aware RocksDB RecordIndexBackend) in
    FlinkOptions
    
    * replace opt in config to index.rli.backend.type
---
 .../apache/hudi/configuration/FlinkOptions.java    | 58 ++++++++++++++++++++++
 rfc/rfc-107/rfc-107.md                             | 16 +++---
 2 files changed, 66 insertions(+), 8 deletions(-)

diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/FlinkOptions.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/FlinkOptions.java
index 9b02e2a202df..88662af61cb3 100644
--- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/FlinkOptions.java
+++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/FlinkOptions.java
@@ -349,6 +349,64 @@ public class FlinkOptions extends HoodieConfig {
       .noDefaultValue()
       .withDescription("Parallelism of tasks that do the index writing, default is the parallelism of the execution environment");
 
+  // ------------------------------------------------------------------------
+  //  RLI RocksDB Partitioned Cache Options
+  // ------------------------------------------------------------------------
+
+  @AdvancedConfig
+  public static final ConfigOption<String> INDEX_RLI_BACKEND_TYPE = ConfigOptions
+      .key("index.rli.backend.type")
+      .stringType()
+      .defaultValue("mdt")
+      .withDescription("The backend used to serve record level index lookups. Supported values are "
+          + "mdt (default), which reads the record level index directly from the metadata table, and "
+          + "rocksdb, which uses a local RocksDB-based partitioned cache in front of the metadata table "
+          + "to accelerate lookups.");
+
+  @AdvancedConfig
+  public static final ConfigOption<String> INDEX_RLI_CACHE_ROCKSDB_BASE_PATH = ConfigOptions
+      .key("index.rli.cache.rocksdb.base.path")
+      .stringType()
+      .defaultValue("/tmp/hudi-index-cache")
+      .withDescription("Local directory for RocksDB partitioned record level index cache data.");
+
+  @AdvancedConfig
+  public static final ConfigOption<Integer> INDEX_RLI_CACHE_ROCKSDB_BOOTSTRAP_DAYS = ConfigOptions
+      .key("index.rli.cache.rocksdb.bootstrap.days")
+      .intType()
+      .defaultValue(7)
+      .withDescription("Number of days of Partitioned Record Index to load during bootstrap. Only partitions "
+          + "within this window are pre-loaded; older partitions are loaded on demand when updates are observed.");
+
+  @AdvancedConfig
+  public static final ConfigOption<Long> INDEX_RLI_CACHE_ROCKSDB_PARTITION_TTL_HOURS = ConfigOptions
+      .key("index.rli.cache.rocksdb.partition.ttl.hours")
+      .longType()
+      .defaultValue(168L) // default 7 days
+      .withDescription("TTL for partition column families in the RocksDB partitioned record level index cache, default 168 hours (7 days).");
+
+  @AdvancedConfig
+  public static final ConfigOption<Long> INDEX_RLI_CACHE_ROCKSDB_BLOCK_CACHE_MB = ConfigOptions
+      .key("index.rli.cache.rocksdb.block.cache.mb")
+      .longType()
+      .defaultValue(256L)
+      .withDescription("RocksDB block cache size (off-heap) in MB for the partitioned record level index cache, default 256.");
+
+  @AdvancedConfig
+  public static final ConfigOption<String> INDEX_RLI_CACHE_ROCKSDB_COMPACTION_STYLE = ConfigOptions
+      .key("index.rli.cache.rocksdb.compaction.style")
+      .stringType()
+      .defaultValue("LEVEL")
+      .withDescription("RocksDB compaction style for the partitioned record level index cache, default LEVEL.");
+
+  @AdvancedConfig
+  public static final ConfigOption<Boolean> INDEX_RLI_CACHE_ROCKSDB_INVALIDATE_ON_REPLACE_COMMIT = ConfigOptions
+      .key("index.rli.cache.rocksdb.invalidate.on.replacecommit")
+      .booleanType()
+      .defaultValue(true)
+      .withDescription("Drop a cached partition's column family when a completed REPLACE_COMMIT (clustering, "
+          + "insert-overwrite, delete-partition) is observed for it, ahead of its TTL, default true.");
+
   // ------------------------------------------------------------------------
   //  Read Options
   // ------------------------------------------------------------------------
diff --git a/rfc/rfc-107/rfc-107.md b/rfc/rfc-107/rfc-107.md
index 667388279e01..56f5f0baa1a2 100644
--- a/rfc/rfc-107/rfc-107.md
+++ b/rfc/rfc-107/rfc-107.md
@@ -223,13 +223,13 @@ Configuration:
 
 | Property                                                | Default | Description |
 |---------------------------------------------------------|---|---|
-| `hoodie.record.index.cache.rocksdb.enabled`             | `false` | Enable RocksDB-based partitioned cache |
-| `hoodie.record.index.cache.rocksdb.base.path`           | `/tmp/hudi-index-cache` | Local directory for RocksDB data |
-| `hoodie.record.index.cache.rocksdb.bootstrap.days`      | `7` | Number of days of Partitioned Record Index to load during bootstrap. Only partitions within this window are pre-loaded; older partitions are loaded on demand when updates are observed. |
-| `hoodie.record.index.cache.rocksdb.partition.ttl.hours` | `168` (7 days) | TTL for partition column families |
-| `hoodie.record.index.cache.rocksdb.block.cache.mb`      | `256` | RocksDB block cache size (off-heap) |
-| `hoodie.record.index.cache.rocksdb.compaction.style`    | `LEVEL` | RocksDB compaction style |
-| `hoodie.record.index.cache.rocksdb.invalidate.on.replacecommit` | `true` | Drop a cached partition's column family when a completed `REPLACE_COMMIT` (clustering, insert-overwrite, delete-partition) is observed for it, ahead of its TTL (see [Interaction with Table Services and Concurrent Writers](#interaction-with-table-services-and-concurrent-writers)) |
+| `index.rli.backend.type`                      | `mdt` | Backend for record level index lookups: `mdt` (reads directly from the metadata table) or `rocksdb` (RocksDB-based partitioned cache in front of the metadata table) |
+| `index.rli.cache.rocksdb.base.path`           | `/tmp/hudi-index-cache` | Local directory for RocksDB data |
+| `index.rli.cache.rocksdb.bootstrap.days`      | `7` | Number of days of Partitioned Record Index to load during bootstrap. Only partitions within this window are pre-loaded; older partitions are loaded on demand when updates are observed. |
+| `index.rli.cache.rocksdb.partition.ttl.hours` | `168` (7 days) | TTL for partition column families |
+| `index.rli.cache.rocksdb.block.cache.mb`      | `256` | RocksDB block cache size (off-heap) |
+| `index.rli.cache.rocksdb.compaction.style`    | `LEVEL` | RocksDB compaction style |
+| `index.rli.cache.rocksdb.invalidate.on.replacecommit` | `true` | Drop a cached partition's column family when a completed `REPLACE_COMMIT` (clustering, insert-overwrite, delete-partition) is observed for it, ahead of its TTL (see [Interaction with Table Services and Concurrent Writers](#interaction-with-table-services-and-concurrent-writers)) |
 
 ### Storage Overhead
 
@@ -320,7 +320,7 @@ The implementation builds on top of RFC-106's infrastructure:
 
 ## Rollout/Adoption Plan
 
-- **No impact on existing users**: The RocksDB cache is disabled by default (`hoodie.index.cache.rocksdb.enabled = false`). Existing Flink users continue using their current index types unchanged.
+- **No impact on existing users**: The RocksDB cache is opt-in via `index.rli.backend.type = rocksdb` (default is `mdt`). Existing Flink users continue using their current index types unchanged.
 - **Opt-in activation**: Users enable the cache by setting the configuration flag. The feature requires RLI to be enabled (RFC-106) as a prerequisite.
 - **Recommended for**: Large tables (100M+ records) with streaming upsert workloads where the in-memory cache from RFC-106 is insufficient.
 - **Not recommended for**: Small tables where the in-memory cache provides adequate performance, or batch workloads where bootstrap overhead dominates.