hudi-agent commented on code in PR #19046: URL: https://github.com/apache/hudi/pull/19046#discussion_r3732293924
########## rfc/rfc-107/rfc-107.md: ########## @@ -0,0 +1,320 @@ + <!-- + Licensed to the Apache Software Foundation (ASF) under one or more + contributor license agreements. See the NOTICE file distributed with + this work for additional information regarding copyright ownership. + The ASF licenses this file to You under the Apache License, Version 2.0 + (the "License"); you may not use this file except in compliance with + the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +--> +# RFC-107: Support data partition aware RocksDB RecordIndexBackend + +## Proposers + +- @zhenqiu-huang + +## Approvers + - TBD + +## Status + - In Progress + +## Abstract + +[RFC-106](../rfc-106/rfc-106.md) introduces Record Level Index (RLI) support for Flink streaming upsert writes, including a simple in-memory cache for index lookups in the `BucketAssigner` operator. While the in-memory cache works well for small to moderate workloads, it faces scalability challenges for large tables with billions of records: the cache either consumes excessive JVM heap memory or suffers from high eviction rates that degrade lookup performance. +In modern CloudLake systems that rely on object storage platforms such as GCS, OCI Object Storage, and Amazon S3, data is typically transitioned to lower storage tiers over time to optimize storage costs. However, using an in-memory cache to accelerate index lookups may result in increased data processing overhead. + +Before this proposal, RocksDBIndexBackend, GlobalRecordLevelIndexBackend and PartitionedIndexBackend have been supported for different Flink upsert scenarios. +- **RocksDBIndexBackend** implements GlobalIndexBackend using an embedded RocksDB instance (RocksDBDAO) as local persistent storage, global data needs to loaded into RocksDB before job starting to processing DB events +- **GlobalRecordLevelIndexBackend** implements MinibatchIndexBackend, backing global record-level-index (RLI) lookups by Hudi's metadata table (HoodieBackedTableMetadata). Keyed only by record key (global, non-partitioned lookups). +- **RecordLevelIndexBackend** maintains one BucketCache per data partition in a LinkedHashMap, with Lazy bootstrap and optimized memory management. + +This RFC proposes a **Support data partition aware RecordLevelIndexBackend** backed by RocksDB that serves as a local materialized replica of the MDT RLI. The provides: + +- **O(1) local lookups** for record location resolution during streaming writes, eliminating per-record MDT I/O +- **Partition-aware storage** using RocksDB column families, enabling efficient TTL-based eviction of stale partitions +- **Bounded resource consumption** by caching only the partitions actively written to, keeping and memory and storage proportional to the working set rather than total table size +- **Incremental maintenance** through in-line index updates during the write path, with MDT as the authoritative source of truth for bootstrap and cross-engine compatibility + +## Background + +### The Index Lookup Bottleneck + +In Hudi's Flink upsert pipeline, the `BucketAssigner` operator must determine whether each incoming record is an insert or an update by looking up its record key in the index. RFC-106 introduces an in-memory cache to accelerate these lookups, but for large-scale streaming workloads, this approach has fundamental limitations: + +1. **Unbounded Cost**: Each RLI entry requires approximately 50–70 bytes of memory. For a table containing 1 billion records, caching the entire index would consume 50–70 GB of JVM heap. In addition, a record buffer is required to improve RLI lookup efficiency. For CDC workloads with high event throughput (QPS) and large record sizes, maintaining a two-minute buffer can further increase memory consumption significantly. As a result, the compute cost of upsert ingestion workloads can rise substantially. +2. **Cache thrashing**: With bounded memory, the cache must evict entries aggressively. For workloads that access records across many partitions, this leads to frequent cache misses and fallback to MDT queries (10+ ms per record), severely degrading throughput. +3. **Cold start latency**: On job restart or task failover, the in-memory cache starts empty. Warming the cache through individual MDT lookups creates a prolonged period of degraded performance. + +**Relationship to MDT's native file-level cache**: MDT's HFile reader already maintains its own local block cache per index shard (see RFC-106), so it is fair to ask why another local cache is needed on top of it. That cache is effective for *batch* lookups — a `BucketAssigner` task issuing one MDT read per micro-batch of keys amortizes the read path (deserialization, block decompression, seek) across many keys. The streaming upsert path, however, needs a location decision per *individual* incoming record as it flows through the operator; even with a warm block cache, each such lookup still pays the cost of the MDT reader stack (HFile seek, Avro/record deserialization, shard routing) once per record. The Dynamic Partitioned Cache avoids that per-record path entirely by keeping a decoded, ready-to-read `record key → location` mapping in-process, so point lookups are a direct RocksDB `Get()` rather than an MDT read. The two caches are complementary and operate at different layers: MDT's file-level cache speeds up the reads that populate and refresh this cache; this cache is what removes the need to go through MDT for the common-case per-record lookup at all. + +### Why RocksDB + +RocksDB record index is already supported. It is a proven embedded key-value store that addresses these limitations: + +- **Off-heap storage**: RocksDB stores data in SST files on local disk with a configurable block cache in off-heap memory, avoiding GC pressure on the JVM heap. +- **Column families**: RocksDB supports column families, which provide logical separation of data within a single database instance. Each partition's index entries can be stored in a dedicated column family, enabling efficient bulk operations (e.g., dropping an entire partition's cache) without affecting other partitions. +- **Compression**: RocksDB applies block-level compression (Snappy by default), keeping the on-disk footprint manageable. +- **Mature Flink integration**: Flink already uses RocksDB as its primary state backend, so operational expertise and deployment patterns are well-established. + +### Alternatives Considered + +**Flink-managed keyed state (MapState with TTL)**: Since Flink already runs RocksDB as its state backend, an alternative is to store the partitioned index directly in Flink `MapState`, scoped per key group, with state TTL for eviction. This gets incremental checkpointing and automatic recovery "for free" and avoids operating a second, unmanaged RocksDB instance alongside Flink's own state backend. + +This RFC uses a standalone RocksDB instance instead, for three reasons: + +- **Partition-level bulk operations**: Flink's state TTL evicts at the *key* granularity (lazily on access, or via a background scan of all keys). It has no notion of dropping an entire partition's entries in O(1); the column-family-per-partition design in this RFC depends on that operation for both TTL eviction and on-demand loading of older partitions. +- **Cross-operator sharing**: the design shares one RocksDB instance between `RLIBootstrapOperator` and `BucketAssigner` (see [Incremental Cache Maintenance](#incremental-cache-maintenance)). Flink's keyed state is private to the operator/task that owns it and cannot be read directly by a sibling operator without routing every lookup through the network, which reintroduces the per-record I/O this RFC is trying to eliminate. +- **Independent lifecycle from checkpointing**: because the cache is disposable and always rebuildable from MDT, it does not need to participate in Flink's checkpoint/restore protocol at all — bootstrapping directly from MDT on failover is simpler and faster than replaying a large keyed-state snapshot, and avoids inflating checkpoint size with data that already lives durably in MDT. + +The tradeoff is that this RFC's cache must implement its own lifecycle management (open/close/bootstrap/discard) rather than reusing Flink's, and does not benefit from incremental checkpointing of the cache contents. Given the cache is fully derivable from MDT (see [Consistency and Failure Handling](#consistency-and-failure-handling)), this is considered an acceptable tradeoff. + +## High Level Design + +The Dynamic Partitioned Cache introduces a RocksDB-based local index replica as a scalable substitute for the *committed* portion of RFC-106's in-memory cache. RFC-106's uncommitted-records buffer — which tracks records written within the current open checkpoint but not yet committed to MDT — is retained unchanged and remains the first tier consulted on every lookup. The Dynamic Partitioned Cache sits between that buffer and MDT, giving a three-tier lookup order: + +1. **Uncommitted-records buffer (in-memory, RFC-106)** — records inserted/updated within the current, not-yet-committed checkpoint. This tier is authoritative for in-flight state and is always consulted first, which is what prevents an update to a key inserted earlier in the same checkpoint from being misclassified as a new insert. +2. **RocksDB cache** — materialized replica of *committed* index state, i.e. record keys that have already landed in an MDT commit. +3. **MDT RLI** — authoritative remote index, consulted only on a miss in both of the above (e.g. cold partition, or first bootstrap). + +Because the uncommitted buffer is checkpoint-scoped and small (bounded by the records written since the last commit), it keeps the memory-bound properties RFC-106 already relies on; the Dynamic Partitioned Cache only takes over the much larger *committed* working set that previously forced the RFC-106 cache to grow unbounded or thrash. + +The RocksDB cache is partitioned by Hudi data partition path using column families. This partition-aware design enables: + +- Bootstrapping only the partitions that the writer actively touches +- Evicting cold partitions via TTL without scanning individual keys +- Bounding cache size to `O(active_partitions)` rather than `O(total_table_size)` + +A fundamental design is needed to handle cache misses by reading from the MDT RLI and storing the loaded index in RocksDB for future lookups. It will be discussed in following sections. + +### Detailed Design + +### Cache Structure and Partitioning + +The RocksDB instance is organized using **column families**, one per Hudi data partition path. Each column family stores key-value pairs where: + +- **Key**: Record key (byte-serialized) +- **Value**: Record location (file group ID + file slice info) + +Using column families provides two critical advantages over a single flat keyspace: + +1. **Efficient partition eviction**: Dropping a column family is an O(1) metadata operation in RocksDB, compared to O(N) individual deletes. +2. **Partition-level TTL**: Each column family can have its own TTL configuration, enabling automatic eviction of partitions that haven't been written to recently. + +``` +RocksDB Instance +├── CF: "default" (metadata: partition registry, timestamps) +├── CF: "dt=2025-01-15" (RLI entries for partition dt=2025-01-15) +├── CF: "dt=2025-01-16" (RLI entries for partition dt=2025-01-16) +└── CF: "dt=2025-01-17" (RLI entries for partition dt=2025-01-17) +``` + +**Partition completeness invariant**: a partition's column family is only visible to lookups once it has been *fully* loaded — there is no partially-cached partition state at Bootstrap time. Bootstrap and on-demand loading (see [Bootstrap Strategy](#bootstrap-strategy)) build the column family off to the side (or via bulk-load into a not-yet-registered CF) and only register it in the `default` CF's partition registry after the load completes. A cache miss during runtime will trigger an on-demand load of smaller granularity of record index into RocksDB. + +### Bootstrap Strategy + +On job start or task failover, the RocksDB cache must be populated from the MDT RLI. The bootstrap strategy differs based on the index scope: + +#### Global RLI Bootstrap + +For global RLI (cross-partition upsert), the cache must contain all record keys across the entire table: + +1. Close and discard any existing RocksDB state (container-local storage is ephemeral). +2. Scan the full MDT RLI partition assigned to this task (based on `hash(record_key) % num_index_shards` assignment from RFC-106). +3. Bulk-load entries into RocksDB using `SSTFileWriter` for optimal ingestion performance. +4. Open the RocksDB instance for read-write access. + +**Bootstrap latency**: For a table with 1 billion records and ~50–70 bytes per entry, scanning and loading the full index requires approximately 50–70 GB of data transfer. At 500 MB/s network throughput, this takes ~2 minutes. This cost is incurred on every job restart. + +#### Partitioned RLI Bootstrap (Recommended) + +This strategy leverages Hudi's [Partitioned Record Index](https://hudi.apache.org/docs/indexes/) (introduced in 1.1.0), which organizes the MDT record-level index by partition path. Unlike the Global Record Index, the Partitioned Record Index guarantees uniqueness within each `(partition_path, record_key)` pair and supports partition-scoped lookups, making it possible to load only a subset of the index during bootstrap. + +**Time-bounded bootstrap**: On job start or task failover, the cache loads only the most recent **X days** of partitioned record index entries (configurable via `hoodie.index.cache.rocksdb.bootstrap.days`). This bounds bootstrap time and resource consumption to a predictable window: + +1. Close and discard any existing RocksDB state. +2. Determine the bootstrap window: compute the set of partitions whose partition path falls within the last X days (e.g., `dt=2025-01-15` through `dt=2025-01-21` for a 7-day window). +3. For each partition in the bootstrap window, scan only its corresponding MDT Partitioned Record Index entries and bulk-load them into a dedicated RocksDB column family. +4. Record the bootstrap boundary timestamp (the oldest partition loaded) in the `default` column family metadata. + +**Non-temporal partition schemes**: the "last X days" window assumes partition paths are date-based (e.g. `dt=2025-01-15`) and can be ordered chronologically. For tables partitioned by a non-temporal key (e.g. `region=us-west`, `category=electronics`) or by a composite key without a leading date component, there is no meaningful notion of "the most recent X days of partitions." For these tables, time-bounded bootstrap falls back to **access-recency bootstrap**: the cache starts empty (equivalent to `bootstrap.days = 0`), and every partition is loaded on demand on first access via the on-demand loading path described below. TTL-based eviction still applies per column family based on last-access time regardless of partition naming scheme. This RFC recommends the time-bounded bootstrap optimization only for date-partitioned tables initially; supporting user-supplied partition-recency hints for non-temporal schemes is left as future work. + +**On-demand loading for older partitions**: During ingestion, when the `BucketAssigner` encounters an incoming record whose partition path is **older than the bootstrap window** (i.e., not yet cached in RocksDB): + +1. Detect the cache miss: the target partition's column family does not exist in RocksDB. +2. Trigger an **on-demand partition load**: read the Partitioned Record Index entries for that specific partition from MDT and materialize them into a new RocksDB column family. +3. Once loaded, perform the record lookup against the newly populated column family and continue normal processing. +4. The on-demand loaded partition is subject to the same TTL-based eviction as bootstrapped partitions (see [TTL-Based Partition Eviction](#ttl-based-partition-eviction)). + +**Handling bursty cold-partition access (backfills/corrections)**: a workload that touches many cold partitions in a short window (e.g. a backfill correcting records across 30 old partitions) would otherwise trigger 30 sequential synchronous loads on the `BucketAssigner`/`RLIBootstrapOperator` thread, each taking on the order of seconds, causing pipeline backpressure. To bound this in a future iteration: + +- On-demand loads for distinct partitions are dispatched to a small bounded thread pool rather than the task's main processing thread, so independent partitions can load in parallel instead of serially. +- Records whose partition is currently being loaded are buffered (bounded by a configurable in-flight record limit) rather than blocking the operator thread; once the column family finishes loading, buffered records are replayed against it. +- If the in-flight buffer limit is exceeded, the operator applies backpressure to upstream by not requesting more input, rather than growing the buffer unboundedly. + +This is scoped as follow-up work beyond the initial synchronous implementation (see [Implementation Plan](#implementation-plan)); the first phase accepts synchronous on-demand loading and measures actual stall impact before investing in the async path. + +This two-tier approach — time-bounded bootstrap plus on-demand loading — ensures that: + +- **Bootstrap is fast and predictable**: loading X days of index data is bounded and proportional to recent write volume, not total table size. For a daily-partitioned table with 10M records/day at ~60 bytes/entry, a 7-day bootstrap loads ~4.2 GB — completing in seconds rather than minutes. +- **Late-arriving data is handled correctly**: updates to partitions older than X days (e.g., backfills, corrections, late-arriving events) trigger on-demand loading of only the affected partition, avoiding a full re-bootstrap. +- **Cache growth remains bounded**: combined with TTL eviction, the cache holds at most `bootstrap_days + on-demand loaded` partitions, with cold partitions automatically evicted. + +``` +Bootstrap Timeline (X = 7 days) + +◄──── Older partitions ────┤◄──── Bootstrap window (7 days) ────►│ Today + │ │ + dt=2025-01-10 dt=2025-01-13 dt=2025-01-15 ... dt=2025-01-21 dt=2025-01-22 + │ │ │ │ + Not loaded Not loaded Bootstrapped at Bootstrapped + (load on (load on job start at job start + demand if demand if + needed) needed) + +When a record arrives for dt=2025-01-10: + 1. Column family "dt=2025-01-10" not found in RocksDB + 2. On-demand load: read Partitioned Record Index for dt=2025-01-10 from MDT + 3. Create column family, bulk-load entries + 4. Perform lookup and continue +``` + +### Incremental Cache Maintenance + +To achieve the on demand RLI load for an older partition, RLIBootstrapOperator needs to be revised to access the shared RocksDB instance with BucketAssign operator. After bootstrap of RLIBootstrapOperator, the RocksDB cache is maintained incrementally during normal write operations: + +**Operators**: `RLIBootstrapOperator` and `BucketAssigner` must run in the same task slot for shared, in-process RocksDB access to be possible at all — routing lookups across the network would defeat the purpose of the cache. The RLI has a bounded number of file groups, so the effective parallelism of RLIBootstrapOperator is capped by that number. Chaining it with BucketAssigner forces both operators to share the same parallelism, which also caps BucketAssigner at the total number of RLI file groups. This prevents scaling BucketAssigner independently when it becomes a performance hotspot. + +**Concurrency model**: within a chained pair, Flink's runtime already guarantees that only one operator's `processElement` executes at a time per subtask (chained operators share a single task thread), so `RLIBootstrapOperator` and `BucketAssigner` never call into RocksDB concurrently for the same record. The only cross-thread access is the background TTL eviction thread (see [TTL-Based Partition Eviction](#ttl-based-partition-eviction)) dropping a column family while the task thread might be reading from it; this is serialized with a per-column-family read-write lock so an in-flight lookup completes (or cleanly misses) before the column family is dropped. Review Comment: 🤖 The concurrency model argues the per-column-family read-write lock makes TTL eviction safe, but that lock only spans a single lookup — it doesn't cover the window between RLIBootstrapOperator confirming the column family exists and BucketAssigner reading from it. If the background TTL thread drops that CF in between, BucketAssigner sees a "clean miss" and (per the pseudocode) treats an existing committed key as an INSERT, creating a duplicate file group for that key. @danny0405 could you clarify whether eviction is blocked for partitions with in-flight records, or whether BucketAssigner must re-trigger an on-demand load on a missing CF rather than assuming INSERT? <sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub> ########## rfc/rfc-107/rfc-107.md: ########## @@ -0,0 +1,320 @@ + <!-- + Licensed to the Apache Software Foundation (ASF) under one or more + contributor license agreements. See the NOTICE file distributed with + this work for additional information regarding copyright ownership. + The ASF licenses this file to You under the Apache License, Version 2.0 + (the "License"); you may not use this file except in compliance with + the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +--> +# RFC-107: Support data partition aware RocksDB RecordIndexBackend + +## Proposers + +- @zhenqiu-huang + +## Approvers + - TBD + +## Status + - In Progress + +## Abstract + +[RFC-106](../rfc-106/rfc-106.md) introduces Record Level Index (RLI) support for Flink streaming upsert writes, including a simple in-memory cache for index lookups in the `BucketAssigner` operator. While the in-memory cache works well for small to moderate workloads, it faces scalability challenges for large tables with billions of records: the cache either consumes excessive JVM heap memory or suffers from high eviction rates that degrade lookup performance. +In modern CloudLake systems that rely on object storage platforms such as GCS, OCI Object Storage, and Amazon S3, data is typically transitioned to lower storage tiers over time to optimize storage costs. However, using an in-memory cache to accelerate index lookups may result in increased data processing overhead. + +Before this proposal, RocksDBIndexBackend, GlobalRecordLevelIndexBackend and PartitionedIndexBackend have been supported for different Flink upsert scenarios. +- **RocksDBIndexBackend** implements GlobalIndexBackend using an embedded RocksDB instance (RocksDBDAO) as local persistent storage, global data needs to loaded into RocksDB before job starting to processing DB events +- **GlobalRecordLevelIndexBackend** implements MinibatchIndexBackend, backing global record-level-index (RLI) lookups by Hudi's metadata table (HoodieBackedTableMetadata). Keyed only by record key (global, non-partitioned lookups). +- **RecordLevelIndexBackend** maintains one BucketCache per data partition in a LinkedHashMap, with Lazy bootstrap and optimized memory management. + +This RFC proposes a **Support data partition aware RecordLevelIndexBackend** backed by RocksDB that serves as a local materialized replica of the MDT RLI. The provides: + +- **O(1) local lookups** for record location resolution during streaming writes, eliminating per-record MDT I/O +- **Partition-aware storage** using RocksDB column families, enabling efficient TTL-based eviction of stale partitions +- **Bounded resource consumption** by caching only the partitions actively written to, keeping and memory and storage proportional to the working set rather than total table size +- **Incremental maintenance** through in-line index updates during the write path, with MDT as the authoritative source of truth for bootstrap and cross-engine compatibility + +## Background + +### The Index Lookup Bottleneck + +In Hudi's Flink upsert pipeline, the `BucketAssigner` operator must determine whether each incoming record is an insert or an update by looking up its record key in the index. RFC-106 introduces an in-memory cache to accelerate these lookups, but for large-scale streaming workloads, this approach has fundamental limitations: + +1. **Unbounded Cost**: Each RLI entry requires approximately 50–70 bytes of memory. For a table containing 1 billion records, caching the entire index would consume 50–70 GB of JVM heap. In addition, a record buffer is required to improve RLI lookup efficiency. For CDC workloads with high event throughput (QPS) and large record sizes, maintaining a two-minute buffer can further increase memory consumption significantly. As a result, the compute cost of upsert ingestion workloads can rise substantially. +2. **Cache thrashing**: With bounded memory, the cache must evict entries aggressively. For workloads that access records across many partitions, this leads to frequent cache misses and fallback to MDT queries (10+ ms per record), severely degrading throughput. +3. **Cold start latency**: On job restart or task failover, the in-memory cache starts empty. Warming the cache through individual MDT lookups creates a prolonged period of degraded performance. + +**Relationship to MDT's native file-level cache**: MDT's HFile reader already maintains its own local block cache per index shard (see RFC-106), so it is fair to ask why another local cache is needed on top of it. That cache is effective for *batch* lookups — a `BucketAssigner` task issuing one MDT read per micro-batch of keys amortizes the read path (deserialization, block decompression, seek) across many keys. The streaming upsert path, however, needs a location decision per *individual* incoming record as it flows through the operator; even with a warm block cache, each such lookup still pays the cost of the MDT reader stack (HFile seek, Avro/record deserialization, shard routing) once per record. The Dynamic Partitioned Cache avoids that per-record path entirely by keeping a decoded, ready-to-read `record key → location` mapping in-process, so point lookups are a direct RocksDB `Get()` rather than an MDT read. The two caches are complementary and operate at different layers: MDT's file-level cache speeds up the reads that populate and refresh this cache; this cache is what removes the need to go through MDT for the common-case per-record lookup at all. + +### Why RocksDB + +RocksDB record index is already supported. It is a proven embedded key-value store that addresses these limitations: + +- **Off-heap storage**: RocksDB stores data in SST files on local disk with a configurable block cache in off-heap memory, avoiding GC pressure on the JVM heap. +- **Column families**: RocksDB supports column families, which provide logical separation of data within a single database instance. Each partition's index entries can be stored in a dedicated column family, enabling efficient bulk operations (e.g., dropping an entire partition's cache) without affecting other partitions. +- **Compression**: RocksDB applies block-level compression (Snappy by default), keeping the on-disk footprint manageable. +- **Mature Flink integration**: Flink already uses RocksDB as its primary state backend, so operational expertise and deployment patterns are well-established. + +### Alternatives Considered + +**Flink-managed keyed state (MapState with TTL)**: Since Flink already runs RocksDB as its state backend, an alternative is to store the partitioned index directly in Flink `MapState`, scoped per key group, with state TTL for eviction. This gets incremental checkpointing and automatic recovery "for free" and avoids operating a second, unmanaged RocksDB instance alongside Flink's own state backend. + +This RFC uses a standalone RocksDB instance instead, for three reasons: + +- **Partition-level bulk operations**: Flink's state TTL evicts at the *key* granularity (lazily on access, or via a background scan of all keys). It has no notion of dropping an entire partition's entries in O(1); the column-family-per-partition design in this RFC depends on that operation for both TTL eviction and on-demand loading of older partitions. +- **Cross-operator sharing**: the design shares one RocksDB instance between `RLIBootstrapOperator` and `BucketAssigner` (see [Incremental Cache Maintenance](#incremental-cache-maintenance)). Flink's keyed state is private to the operator/task that owns it and cannot be read directly by a sibling operator without routing every lookup through the network, which reintroduces the per-record I/O this RFC is trying to eliminate. +- **Independent lifecycle from checkpointing**: because the cache is disposable and always rebuildable from MDT, it does not need to participate in Flink's checkpoint/restore protocol at all — bootstrapping directly from MDT on failover is simpler and faster than replaying a large keyed-state snapshot, and avoids inflating checkpoint size with data that already lives durably in MDT. + +The tradeoff is that this RFC's cache must implement its own lifecycle management (open/close/bootstrap/discard) rather than reusing Flink's, and does not benefit from incremental checkpointing of the cache contents. Given the cache is fully derivable from MDT (see [Consistency and Failure Handling](#consistency-and-failure-handling)), this is considered an acceptable tradeoff. + +## High Level Design + +The Dynamic Partitioned Cache introduces a RocksDB-based local index replica as a scalable substitute for the *committed* portion of RFC-106's in-memory cache. RFC-106's uncommitted-records buffer — which tracks records written within the current open checkpoint but not yet committed to MDT — is retained unchanged and remains the first tier consulted on every lookup. The Dynamic Partitioned Cache sits between that buffer and MDT, giving a three-tier lookup order: + +1. **Uncommitted-records buffer (in-memory, RFC-106)** — records inserted/updated within the current, not-yet-committed checkpoint. This tier is authoritative for in-flight state and is always consulted first, which is what prevents an update to a key inserted earlier in the same checkpoint from being misclassified as a new insert. +2. **RocksDB cache** — materialized replica of *committed* index state, i.e. record keys that have already landed in an MDT commit. +3. **MDT RLI** — authoritative remote index, consulted only on a miss in both of the above (e.g. cold partition, or first bootstrap). + +Because the uncommitted buffer is checkpoint-scoped and small (bounded by the records written since the last commit), it keeps the memory-bound properties RFC-106 already relies on; the Dynamic Partitioned Cache only takes over the much larger *committed* working set that previously forced the RFC-106 cache to grow unbounded or thrash. + +The RocksDB cache is partitioned by Hudi data partition path using column families. This partition-aware design enables: + +- Bootstrapping only the partitions that the writer actively touches +- Evicting cold partitions via TTL without scanning individual keys +- Bounding cache size to `O(active_partitions)` rather than `O(total_table_size)` + +A fundamental design is needed to handle cache misses by reading from the MDT RLI and storing the loaded index in RocksDB for future lookups. It will be discussed in following sections. + +### Detailed Design + +### Cache Structure and Partitioning + +The RocksDB instance is organized using **column families**, one per Hudi data partition path. Each column family stores key-value pairs where: + +- **Key**: Record key (byte-serialized) +- **Value**: Record location (file group ID + file slice info) + +Using column families provides two critical advantages over a single flat keyspace: + +1. **Efficient partition eviction**: Dropping a column family is an O(1) metadata operation in RocksDB, compared to O(N) individual deletes. +2. **Partition-level TTL**: Each column family can have its own TTL configuration, enabling automatic eviction of partitions that haven't been written to recently. + +``` +RocksDB Instance +├── CF: "default" (metadata: partition registry, timestamps) +├── CF: "dt=2025-01-15" (RLI entries for partition dt=2025-01-15) +├── CF: "dt=2025-01-16" (RLI entries for partition dt=2025-01-16) +└── CF: "dt=2025-01-17" (RLI entries for partition dt=2025-01-17) +``` + +**Partition completeness invariant**: a partition's column family is only visible to lookups once it has been *fully* loaded — there is no partially-cached partition state at Bootstrap time. Bootstrap and on-demand loading (see [Bootstrap Strategy](#bootstrap-strategy)) build the column family off to the side (or via bulk-load into a not-yet-registered CF) and only register it in the `default` CF's partition registry after the load completes. A cache miss during runtime will trigger an on-demand load of smaller granularity of record index into RocksDB. + +### Bootstrap Strategy + +On job start or task failover, the RocksDB cache must be populated from the MDT RLI. The bootstrap strategy differs based on the index scope: + +#### Global RLI Bootstrap + +For global RLI (cross-partition upsert), the cache must contain all record keys across the entire table: + +1. Close and discard any existing RocksDB state (container-local storage is ephemeral). +2. Scan the full MDT RLI partition assigned to this task (based on `hash(record_key) % num_index_shards` assignment from RFC-106). +3. Bulk-load entries into RocksDB using `SSTFileWriter` for optimal ingestion performance. +4. Open the RocksDB instance for read-write access. + +**Bootstrap latency**: For a table with 1 billion records and ~50–70 bytes per entry, scanning and loading the full index requires approximately 50–70 GB of data transfer. At 500 MB/s network throughput, this takes ~2 minutes. This cost is incurred on every job restart. + +#### Partitioned RLI Bootstrap (Recommended) + +This strategy leverages Hudi's [Partitioned Record Index](https://hudi.apache.org/docs/indexes/) (introduced in 1.1.0), which organizes the MDT record-level index by partition path. Unlike the Global Record Index, the Partitioned Record Index guarantees uniqueness within each `(partition_path, record_key)` pair and supports partition-scoped lookups, making it possible to load only a subset of the index during bootstrap. + +**Time-bounded bootstrap**: On job start or task failover, the cache loads only the most recent **X days** of partitioned record index entries (configurable via `hoodie.index.cache.rocksdb.bootstrap.days`). This bounds bootstrap time and resource consumption to a predictable window: + +1. Close and discard any existing RocksDB state. +2. Determine the bootstrap window: compute the set of partitions whose partition path falls within the last X days (e.g., `dt=2025-01-15` through `dt=2025-01-21` for a 7-day window). +3. For each partition in the bootstrap window, scan only its corresponding MDT Partitioned Record Index entries and bulk-load them into a dedicated RocksDB column family. +4. Record the bootstrap boundary timestamp (the oldest partition loaded) in the `default` column family metadata. + +**Non-temporal partition schemes**: the "last X days" window assumes partition paths are date-based (e.g. `dt=2025-01-15`) and can be ordered chronologically. For tables partitioned by a non-temporal key (e.g. `region=us-west`, `category=electronics`) or by a composite key without a leading date component, there is no meaningful notion of "the most recent X days of partitions." For these tables, time-bounded bootstrap falls back to **access-recency bootstrap**: the cache starts empty (equivalent to `bootstrap.days = 0`), and every partition is loaded on demand on first access via the on-demand loading path described below. TTL-based eviction still applies per column family based on last-access time regardless of partition naming scheme. This RFC recommends the time-bounded bootstrap optimization only for date-partitioned tables initially; supporting user-supplied partition-recency hints for non-temporal schemes is left as future work. + +**On-demand loading for older partitions**: During ingestion, when the `BucketAssigner` encounters an incoming record whose partition path is **older than the bootstrap window** (i.e., not yet cached in RocksDB): + +1. Detect the cache miss: the target partition's column family does not exist in RocksDB. +2. Trigger an **on-demand partition load**: read the Partitioned Record Index entries for that specific partition from MDT and materialize them into a new RocksDB column family. +3. Once loaded, perform the record lookup against the newly populated column family and continue normal processing. +4. The on-demand loaded partition is subject to the same TTL-based eviction as bootstrapped partitions (see [TTL-Based Partition Eviction](#ttl-based-partition-eviction)). + +**Handling bursty cold-partition access (backfills/corrections)**: a workload that touches many cold partitions in a short window (e.g. a backfill correcting records across 30 old partitions) would otherwise trigger 30 sequential synchronous loads on the `BucketAssigner`/`RLIBootstrapOperator` thread, each taking on the order of seconds, causing pipeline backpressure. To bound this in a future iteration: + +- On-demand loads for distinct partitions are dispatched to a small bounded thread pool rather than the task's main processing thread, so independent partitions can load in parallel instead of serially. +- Records whose partition is currently being loaded are buffered (bounded by a configurable in-flight record limit) rather than blocking the operator thread; once the column family finishes loading, buffered records are replayed against it. +- If the in-flight buffer limit is exceeded, the operator applies backpressure to upstream by not requesting more input, rather than growing the buffer unboundedly. + +This is scoped as follow-up work beyond the initial synchronous implementation (see [Implementation Plan](#implementation-plan)); the first phase accepts synchronous on-demand loading and measures actual stall impact before investing in the async path. + +This two-tier approach — time-bounded bootstrap plus on-demand loading — ensures that: + +- **Bootstrap is fast and predictable**: loading X days of index data is bounded and proportional to recent write volume, not total table size. For a daily-partitioned table with 10M records/day at ~60 bytes/entry, a 7-day bootstrap loads ~4.2 GB — completing in seconds rather than minutes. +- **Late-arriving data is handled correctly**: updates to partitions older than X days (e.g., backfills, corrections, late-arriving events) trigger on-demand loading of only the affected partition, avoiding a full re-bootstrap. +- **Cache growth remains bounded**: combined with TTL eviction, the cache holds at most `bootstrap_days + on-demand loaded` partitions, with cold partitions automatically evicted. + +``` +Bootstrap Timeline (X = 7 days) + +◄──── Older partitions ────┤◄──── Bootstrap window (7 days) ────►│ Today + │ │ + dt=2025-01-10 dt=2025-01-13 dt=2025-01-15 ... dt=2025-01-21 dt=2025-01-22 + │ │ │ │ + Not loaded Not loaded Bootstrapped at Bootstrapped + (load on (load on job start at job start + demand if demand if + needed) needed) + +When a record arrives for dt=2025-01-10: + 1. Column family "dt=2025-01-10" not found in RocksDB + 2. On-demand load: read Partitioned Record Index for dt=2025-01-10 from MDT + 3. Create column family, bulk-load entries + 4. Perform lookup and continue +``` + +### Incremental Cache Maintenance + +To achieve the on demand RLI load for an older partition, RLIBootstrapOperator needs to be revised to access the shared RocksDB instance with BucketAssign operator. After bootstrap of RLIBootstrapOperator, the RocksDB cache is maintained incrementally during normal write operations: + +**Operators**: `RLIBootstrapOperator` and `BucketAssigner` must run in the same task slot for shared, in-process RocksDB access to be possible at all — routing lookups across the network would defeat the purpose of the cache. The RLI has a bounded number of file groups, so the effective parallelism of RLIBootstrapOperator is capped by that number. Chaining it with BucketAssigner forces both operators to share the same parallelism, which also caps BucketAssigner at the total number of RLI file groups. This prevents scaling BucketAssigner independently when it becomes a performance hotspot. + +**Concurrency model**: within a chained pair, Flink's runtime already guarantees that only one operator's `processElement` executes at a time per subtask (chained operators share a single task thread), so `RLIBootstrapOperator` and `BucketAssigner` never call into RocksDB concurrently for the same record. The only cross-thread access is the background TTL eviction thread (see [TTL-Based Partition Eviction](#ttl-based-partition-eviction)) dropping a column family while the task thread might be reading from it; this is serialized with a per-column-family read-write lock so an in-flight lookup completes (or cleanly misses) before the column family is dropped. + +#### On Record Processing + +``` +for each incoming record r: + 1. In RLIBootstrapOperator, look up through shared RocksDB for r.partitionPath + → If column family does not exist: + a. Load Partitioned Record Index for r.partitionPath from MDT (on-demand) + b. Create column family and bulk-load entries + → If found: forward to next operator BucketAssign + 2. In BucketAssign operator, check RocksDB cache for r.key in column family r.partitionPath + → If found in RocksDB: use cached location (committed record) + → If not found: this is an INSERT, assign new file group + → Update RocksDB cache with r.key → assigned location (there is no side effect, even it is not materialized into remote RLI) Review Comment: 🤖 Here the assigned location is written into RocksDB inline during processing, before the checkpoint commits to MDT — but tier 2 is defined as "committed index state," and the cache is only discarded on task failover / job restart, not on an aborted checkpoint. If a checkpoint is declined/expired without failover, the pending instant rolls back yet these uncommitted key→location entries stay in RocksDB, so a later record for the same key could be misclassified as an UPDATE against a rolled-back location. Could you reconcile this with the "update on checkpoint completion" approach discussed above, and spell out how an aborted checkpoint (no failover) resets the cache? @nsivabalan <sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub> -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
