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 <[email protected]>
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.
