This is an automated email from the ASF dual-hosted git repository.
voonhous 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 37ac38e560ac docs: rfc-107 Support data partition aware RocksDB
RecordIndexBackend (#19046)
37ac38e560ac is described below
commit 37ac38e560ac8f6e7717af17740e70dfefd8c1e0
Author: Peter Huang <[email protected]>
AuthorDate: Fri Aug 14 01:28:17 2026 -0700
docs: rfc-107 Support data partition aware RocksDB RecordIndexBackend
(#19046)
* docs: rfc-107 Support data partition aware RocksDB RecordIndexBackend
* resolve comments
* resolve shuo's comment
* resolve shuo's comment
* change config prefix according's danny's suggestion
* add cache invalidation solution and operation suggestions
---
rfc/README.md | 4 +-
rfc/rfc-107/rfc-107.md | 339 +++++++++++++++++++++++++++++++++++++++++++++++++
2 files changed, 341 insertions(+), 2 deletions(-)
diff --git a/rfc/README.md b/rfc/README.md
index 4e8d08751507..45f4bdc5a500 100644
--- a/rfc/README.md
+++ b/rfc/README.md
@@ -142,7 +142,7 @@ The list of all RFCs can be found here.
| 104 | [Unify schema evolution on
schema-on-read](./rfc-104/rfc-104.md)
| :eyes: `UNDER REVIEW` |
| 105 | [Trino Hudi Connector — Shim/Bundle
Refactor](./rfc-105/rfc-105.md)
| :eyes: `UNDER REVIEW` |
| 106 | [Record Level and Secondary Index Support for Flink
Writers](./rfc-106/rfc-106.md)
| :white_check_mark: `COMPLETED` |
-| 107 | Dynamic Partitioned Cache for Flink upsert
|
:hammer_and_wrench: `IN PROGRESS` |
+| 107 | [Support data partition aware RocksDB
RecordIndexBackend](./rfc-107/rfc-107/md)
| :hammer_and_wrench: `IN PROGRESS` |
| 108 | [Multi-dataset incremental reads in Hudi
Streamer](./rfc-108/rfc-108.md)
| :eyes: `UNDER REVIEW` |
| 109 | Hudi Native Vector Index
| :eyes:
`UNDER REVIEW` |
-| 110 | Native Full-Text Search Index
| :eyes:
`UNDER REVIEW` |
+| 110 | Native Full-Text Search Index
| :eyes:
`UNDER REVIEW` | | :eyes: `UNDER REVIEW` |
diff --git a/rfc/rfc-107/rfc-107.md b/rfc/rfc-107/rfc-107.md
new file mode 100644
index 000000000000..667388279e01
--- /dev/null
+++ b/rfc/rfc-107/rfc-107.md
@@ -0,0 +1,339 @@
+ <!--
+ 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 [...]
+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 *individua [...]
+
+### 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
that adaptively trades off bootstrap time against RLI lookup latency based on
the write patterns of the target dataset. By accounting for workload-specific
write characteristics, the cache can determine how much local index state to
materialize during bootstrap, balancing initialization cost with the lookup
performance achieved once the replica is serving traffic
+1. **RocksDB cache** — materialized replica of *committed* index state, i.e.
record keys that have already landed in an MDT commit.
+2. **MDT RLI** — authoritative remote index, consulted only on a miss in both
of the above (e.g. cold partition, or first bootstrap).
+
+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 o [...]
+
+### 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 (eq [...]
+
+**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` 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.
Thus, they need be decoupled but with the same in-process RocksDB access (The
RLI has a bounded number of file groups, so the effective parallelism of
RLIBootstrapOperator is capped by th [...]
+
+**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 mig [...]
+
+#### 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)
+```
+
+In this process above, `RLIBootstrapOperator` is responsible for checking
whether a record is from an older partition that has not been bootstrapped. If
so, it triggers the on-demand load from MDT and forwards the
`HoodieFlinkInternalRow` to the `BucketAssigner` operator.
+
+#### On Index Write
+
+In the `IndexWrite` operator (from RFC-106), index records are written to MDT.
The RocksDB cache in the `BucketAssigner` is updated in-line as records flow
through the pipeline, ensuring the cache stays ahead of MDT commits.
+
+### TTL-Based Partition Eviction
+
+For partitioned RLI, the cache implements automatic eviction of cold
partitions:
+
+- Each column family tracks the **last access timestamp** (last time a record
was written to or looked up in that partition).
+- The TTL should be set based on the workload's partition access pattern. For
daily-partitioned event data, a TTL of 3–7 days is typical.
+- Given we need to consider the ongoing checkpoint, the cleanup process of
droping column families whose last access exceeds the configured TTL should
happen right after checkpoint completion.
+
+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)) |
+
+### Storage Overhead
+
+RocksDB occupies approximately **2x the storage** compared to native HFile
format in MDT, due to:
+
+1. **Compression codec difference**: RocksDB uses Snappy compression by
default, while Hudi MDT uses gzip, which achieves higher compression ratios.
+2. **Uncompacted SST files**: During active writes, RocksDB maintains multiple
levels of SST files before compaction merges them.
+3. **WAL disabled**: Write-ahead log is disabled since the cache can be
rebuilt from MDT on failure, reducing write amplification.
+
+For a partition with 10 million records at ~60 bytes per entry, the RocksDB
footprint is approximately 1.2 GB on disk.
+
+### Consistency and Failure Handling
+
+The RocksDB cache is a **derived, disposable replica** — MDT remains the
single source of truth. This simplifies consistency handling:
+
+#### Task Failover
+
+1. The RocksDB cache on the failed task's container is discarded (ephemeral
local storage).
+2. The all of recovered tasks bootstrap a fresh RocksDB instance from MDT.
(Local recovery of RocksDBDAO is not scoped in this RFC)
+3. If the failure happens in commit phase file write and index update has been
done, The coordinator recommits any pending Hudi instants (as described in
RFC-106).
+
+#### Job Restart
+
+1. All RocksDB state is discarded.
+2. Full bootstrap from MDT is triggered.
+3. If the restart happens in commit phase file write and index update has been
done, The coordinator handles recommitting of pending instants per RFC-106's
recovery protocol.
+
+#### Interaction with Table Services and Concurrent Writers
+
+The primary deployment target for this RFC is a **single Flink ingestion
writer** per table, with table services (compaction, clustering) running as
separate, periodic jobs — the same assumption RFC-106 makes for its in-memory
cache. This RFC does not target concurrent *ingestion* writers to the data
table. However, table services still write to the table (and, for clustering,
to the MDT's `record_index` partition) concurrently with the Flink writer that
owns the RocksDB cache, so the ca [...]
+
+**Why compaction is safe**: MDT/data-table compaction only merges existing log
files into a new base file *within the same file group* — it does not change
which `fileId` a record key maps to. Since the RocksDB cache's value is
`(fileId + file slice info)`, not a base-file path, compaction never
invalidates a cached entry; the cache's lookup contract is already coarse
enough to tolerate it.
+
+**Why clustering is not safe**: Clustering runs as a `REPLACE_COMMIT` that
rewrites records into brand-new file groups and replaces the old ones
(`HoodieReplaceCommitMetadata.getPartitionToReplaceFileIds()`). Every clustered
record key gets a fresh `record_index` entry pointing at its new `fileId` —
`BaseRecordIndexer` re-derives and upserts a RLI row per key for every write
stat in the replace-commit, the same as for a normal commit — so the
**authoritative MDT RLI is always correct aft [...]
+
+
+**Proposed invalidation mechanism — timeline-driven column family drop**:
rather than trying to patch individual cached keys, the cache reuses the same
O(1) primitive already used for [TTL-based
eviction](#ttl-based-partition-eviction) — dropping a column family — to
invalidate an entire partition whenever a clustering (or
`INSERT_OVERWRITE`/`DELETE_PARTITION`-style) replace-commit has touched it:
+
+1. The `default` column family's partition registry already records, per
cached partition, the instant time as of which it was last materialized
(bootstrap boundary or last on-demand/incremental refresh — see [Cache
Structure and Partitioning](#cache-structure-and-partitioning)).
+2. At the same cadence as TTL eviction (right after checkpoint completion, to
avoid interfering with an in-flight checkpoint), `RLIBootstrapOperator` scans
the active timeline for newly completed `REPLACE_COMMIT` instants since the
last scan.
+3. For each such instant, `getPartitionToReplaceFileIds()` gives the set of
replaced partitions/file groups directly, with no need to read record-level
data. If a partition in this set is currently cached and no new lookup occurs
to RLI of it during the current checkpoint interval, its column family is
dropped. Otherwise, if any record has used an invalidated cache at the current
instant, the job fails fast.
+4. The next lookup against that partition falls through the normal [on-demand
loading](#partitioned-rli-bootstrap-recommended) path, which bulk-loads from
MDT and therefore observes the post-clustering `fileId`s.
+
+This adds only a periodic, bounded timeline scan (proportional to the number
of new instants since the last check, not table size) and no new cross-process
invalidation channel — it composes with mechanisms the cache already has.
+
+**Concurrency-control mode matters for how aggressively this must run**:
+
+- **Under Optimistic Concurrency Control (OCC)**,
`SimpleConcurrentFileWritesConflictResolutionStrategy` detects `(partition,
fileId)` overlap between the Flink writer's pending commit and other
completed/pending commits (including pending and completed replace-commits) at
pre-commit time, under the write lock; on overlap it raises
`HoodieWriteConflictException` and the writer's commit attempt aborts. This
protects the *data table's* file-group integrity, but it does not protect this
RFC [...]
+- **Under Non-Blocking Concurrency Control (NBCC)**, conflict resolution is
skipped for non-bulk-insert operations, and — notably — always skipped for the
metadata table itself once metadata-table streaming writes are enabled, since
MDT's own NB-CC support is still evolving. `HoodieMetadataPayload`'s RLI record
has no NBCC-aware ordering field; concurrent RLI upserts from independent
writers merge by generic commit-completion order, not by NBCC's data-table
log-scan semantics. This is an [...]
+
+In short, this solution works well for large datasets with a **single Flink
ingestion writer**, **OCC** model and **non-clustering** asynchronous table
operations. If clustering is required, we recommend using **OCC** with the
enhanced timeline-based cache validation mechanism. This allows the Flink
ingestion job to fail fast on conflicts and restart with a new RLI. However, if
a conflict occurs after a Flink checkpoint has completed, it cannot be resolved
automatically. In that case, a [...]
+
+### Integration with RFC-106 Pipeline
+
+RFC-106 introduces the following operator chain for Flink streaming upserts:
`BucketAssigner → StreamWriteFunction → ... → IndexWrite → Coordinator`. This
RFC modifies that pipeline as follows:
+
+1. **New operator**: `RLIBootstrapOperator` is inserted immediately before
`BucketAssigner`, chained to it (same parallelism, no shuffle in between — see
[Operator co-location](#incremental-cache-maintenance)). It owns the RocksDB
instance's lifecycle (open at job start, bootstrap per [Bootstrap
Strategy](#bootstrap-strategy), close on job shutdown) and ensures the column
family for an incoming record's partition exists before the record reaches
`BucketAssigner`, triggering an on-demand [...]
+2. **BucketAssigner change**: instead of consulting only RFC-106's in-memory
cache, `BucketAssigner` now consults lookup order from [High Level
Design](#high-level-design) (RocksDB → MDT), using the same RocksDB instance
opened by `RLIBootstrapOperator`.
+3. **IndexWrite change**: unchanged in terms of its output contract to MDT
(RFC-106's index write path itself is not modified), but it now also emits an
in-line update into the shared RocksDB instance so newly-committed keys become
visible to the cache immediately, ahead of the next MDT read, per [On Index
Write](#on-index-write).
+4. **Coordinator**: unchanged from RFC-106. Task failover and job restart
recovery (recommitting pending instants) continue to follow RFC-106's existing
protocol; this RFC only adds the RocksDB rebuild step described in [Task
Failover](#task-failover) and [Job Restart](#job-restart) ahead of resuming
normal processing.
+
+This RFC does not change RFC-106's checkpointing, coordinator, or commit
protocol — the only structural change to the operator graph is the addition of
`RLIBootstrapOperator` ahead of `BucketAssigner`, and the addition of a
cache-write side effect in `IndexWrite`.
+
+## Implementation Plan
+
+The implementation builds on top of RFC-106's infrastructure:
+
+- **Phase 1: RocksDB cache foundation**
+ - Implement RocksDB lifecycle management (open, close, bootstrap, discard)
+ - Implement column-family-per-partition data model
+ - Integrate with `BucketAssigner` as a cache layer between in-memory cache
and MDT
+
+- **Phase 2: Bootstrap and incremental maintenance**
+ - Implement partitioned RLI bootstrap (lazy, on-demand per partition)
+ - Implement global RLI bootstrap (full scan with SST bulk load)
+ - Add incremental cache updates on record processing and checkpoint
completion
+
+- **Phase 3: TTL-based eviction and resource management**
+ - Add configurable RocksDB tuning parameters (block cache size, compaction
style)
+ - Add monitoring metrics (cache hit rate, bootstrap latency, partition count)
+ - Implement partition TTL eviction on checkpoint completion
+
+- **Phase 4: Concurrent table service support**
+ - Track a per-partition "as-of" instant time in the `default` column
family's registry, updated on bootstrap, on-demand load, and incremental refresh
+ - Implement a periodic (post-checkpoint) timeline scan for newly completed
`REPLACE_COMMIT` instants and drop affected column families per [Interaction
with Table Services and Concurrent
Writers](#interaction-with-table-services-and-concurrent-writers)
+ - Bypass TTL trust (force invalidation) for partitions with in-flight or
recently-completed table-service instants, to bound the NBCC ambiguity window
described above
+ - Compaction requires no invalidation handling, since it does not change a
record key's `fileId` assignment
+
+## 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.
+- **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.
+
+## Test Plan
+
+1. **Functional correctness**: Verify that upsert, insert, delete, and
cross-partition update operations produce correct results with the RocksDB
cache enabled.
+2. **Cache consistency**: Validate that the cache remains consistent with MDT
after checkpoint completion, task failover, and job restart scenarios.
+3. **Partition eviction**: Test that TTL-based eviction correctly drops stale
partitions and that subsequent access to evicted partitions triggers
re-bootstrap from MDT.
+4. **Concurrent table services**: Verify that a completed clustering
(`REPLACE_COMMIT`) affecting a cached partition results in that partition's
column family being dropped and correctly reloaded with post-clustering
`fileId`s on next access, before its TTL would otherwise have expired it.
Verify compaction alone (no clustering) never triggers invalidation. Verify
that under NBCC, cached partitions with concurrent table-service activity are
invalidated ahead of their TTL rather than serv [...]
+5. **Performance benchmarks**:
+ - Measure cache hit rate under various workload patterns (hot partition,
uniform random, zipfian).
+ - Compare throughput with in-memory-only cache vs. RocksDB cache for tables
of varying sizes (10M, 100M, 1B records).
+ - Measure bootstrap latency for partitioned vs. global RLI configurations.
+6. **Resource consumption**: Profile RocksDB memory (block cache, memtable)
and disk usage under sustained write loads to validate bounded resource
consumption.
+
