danny0405 commented on code in PR #19046: URL: https://github.com/apache/hudi/pull/19046#discussion_r3654251235
########## rfc/rfc-107/rfc-107.md: ########## @@ -0,0 +1,358 @@ + <!-- + 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: Dynamic Partitioned Cache for Flink Hudi Upsert + +## 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 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. + +This RFC proposes a **Dynamic Partitioned Cache** backed by RocksDB that serves as a local materialized replica of the MDT RLI. The cache 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 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 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. Review Comment: So it looks like an optimization to existing RocksDB bootstrap based RLI solution? To make the bootstrap and eviction both partition aware? Then can we modify the title of the RFC, it looks more like an enhancement to existing solution instead of a substitute of the native MDT RLI support. -- 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]
