Re: [PR] docs: rfc-107 Support data partition aware RocksDB RecordIndexBackend [hudi]
vinothchandar commented on PR #19046: URL: https://github.com/apache/hudi/pull/19046#issuecomment-5500076838 I am going to add a comment to https://github.com/apache/hudi/discussions/18296 , to keep the discussion going someplace where its not closed. @danny0405 @HuangZhenQiu please engage there.. -- 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]
Re: [PR] docs: rfc-107 Support data partition aware RocksDB RecordIndexBackend [hudi]
danny0405 commented on PR #19046: URL: https://github.com/apache/hudi/pull/19046#issuecomment-5351517103 > @danny0405 What are your findings around the OCC/correctness questions we discussed.. The current conflict resolution happens right before the Hudi commit, current Flink try to commit to Hudi timeline after receiving a checkpoint ack msg(which indicates the checkpoint is already successful to Flink signt of view), let's say it is `ckp_n`, when the exception throws because of the confclit resolution, the commit procedure aborted and trigger the job to restart, but the job by default restore from the last successful checkpoint `ckp_n` which is actually invalid to Hudi. I have fire a similar discuss here before for the rocksDB statebackend solution: https://github.com/apache/hudi/discussions/18296, and there are some explanaitons and also proposals in the discussion. -- 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]
Re: [PR] docs: rfc-107 Support data partition aware RocksDB RecordIndexBackend [hudi]
vinothchandar commented on PR #19046: URL: https://github.com/apache/hudi/pull/19046#issuecomment-5350728733 @danny0405 What are your findings around the OCC/correctness questions we discussed.. -- 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]
Re: [PR] docs: rfc-107 Support data partition aware RocksDB RecordIndexBackend [hudi]
bhasudha commented on code in PR #19046: URL: https://github.com/apache/hudi/pull/19046#discussion_r3787477606 ## rfc/rfc-107/rfc-107.md: ## @@ -0,0 +1,358 @@ + +# 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
Re: [PR] docs: rfc-107 Support data partition aware RocksDB RecordIndexBackend [hudi]
voonhous merged PR #19046: URL: https://github.com/apache/hudi/pull/19046 -- 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]
Re: [PR] docs: rfc-107 Support data partition aware RocksDB RecordIndexBackend [hudi]
hudi-bot commented on PR #19046: URL: https://github.com/apache/hudi/pull/19046#issuecomment-5284721049 ## CI report: * 476537b17e09247d7de302946b22c8b16b66b460 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16280) Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] docs: rfc-107 Support data partition aware RocksDB RecordIndexBackend [hudi]
hudi-bot commented on PR #19046: URL: https://github.com/apache/hudi/pull/19046#issuecomment-5283809584 ## CI report: * 782f4fd9d26907d295a3e7a5387ecb3b8e10fd63 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16229) * 476537b17e09247d7de302946b22c8b16b66b460 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16280) Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] docs: rfc-107 Support data partition aware RocksDB RecordIndexBackend [hudi]
hudi-bot commented on PR #19046: URL: https://github.com/apache/hudi/pull/19046#issuecomment-5283640529 ## CI report: * 782f4fd9d26907d295a3e7a5387ecb3b8e10fd63 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16229) * 476537b17e09247d7de302946b22c8b16b66b460 UNKNOWN Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] docs: rfc-107 Support data partition aware RocksDB RecordIndexBackend [hudi]
HuangZhenQiu commented on PR #19046: URL: https://github.com/apache/hudi/pull/19046#issuecomment-5283676940 @vinothchandar Per our discussion in yesterday's Hudi Dev meeting, I added my thinking of cache validation solution for OCC and suggested operation model to ensure no data loss. I don't think the solution fit for the NPCC mode, we main explicit disallow of using the solution with NPCC at the same time. -- 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]
Re: [PR] docs: rfc-107 Support data partition aware RocksDB RecordIndexBackend [hudi]
vinothchandar commented on PR #19046: URL: https://github.com/apache/hudi/pull/19046#issuecomment-5275264674 @HuangZhenQiu thanks for that clarification. I am mainly interested in how we handle parallel writers/clustering (throw errors, safely abort writes) in those cases.. -- 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]
Re: [PR] docs: rfc-107 Support data partition aware RocksDB RecordIndexBackend [hudi]
vinothchandar commented on code in PR #19046: URL: https://github.com/apache/hudi/pull/19046#discussion_r3771958998 ## rfc/rfc-107/rfc-107.md: ## @@ -0,0 +1,306 @@ + +# 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. Review Comment: So, is this now another IndexBackend - how do we prevent thrashing in this one . or can we reuse/fix the existing local cache? -- 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]
Re: [PR] docs: rfc-107 Support data partition aware RocksDB RecordIndexBackend [hudi]
vinothchandar commented on code in PR #19046: URL: https://github.com/apache/hudi/pull/19046#discussion_r3771950776 ## rfc/rfc-107/rfc-107.md: ## @@ -0,0 +1,306 @@ + +# 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 Review Comment: reopening to define the behavior - even if you all don't perform clustering in parallel to MoR writer, we should throw errors safely? prevent any writer from working off uncommitted data. -- 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]
Re: [PR] docs: rfc-107 Support data partition aware RocksDB RecordIndexBackend [hudi]
HuangZhenQiu commented on PR #19046: URL: https://github.com/apache/hudi/pull/19046#issuecomment-5273578064 @vinothchandar Thanks for the feedback. I think we had achieved alignment with @cshuo @danny0405. The assumption of Flink ingestion is that there is no multiple writer. The async compaction job only async compact the data /record index partition. It ensures no change to the mapping of row key to file group info. This proposal added the dynamic cache in rocksDB with bounded disk usage and bootstrap time. The correctness will continue hold if the previous assumption. I will join today's dev sync to align more details with other PMCs. Thanks. -- 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]
Re: [PR] docs: rfc-107 Support data partition aware RocksDB RecordIndexBackend [hudi]
vinothchandar commented on PR #19046: URL: https://github.com/apache/hudi/pull/19046#issuecomment-5273049903 @HuangZhenQiu are you able to join one of the weekly dev syncs for us to discuss this? I see a lot of AI comments here as well. Can you summarize for me in 1 line- if we are reaching alignment here on the design. The correctness/consistency edge cases were my main concern. if the RFC can make it clear, how invariants keep them up, it ll be great! -- 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]
Re: [PR] docs: rfc-107 Support data partition aware RocksDB RecordIndexBackend [hudi]
hudi-agent commented on code in PR #19046: URL: https://github.com/apache/hudi/pull/19046#discussion_r3765081519 ## rfc/rfc-107/rfc-107.md: ## Review Comment: 🤖 **Line 136:** The config table above was renamed to the `hoodie.record.index.cache.rocksdb.*` prefix, but this reference (and the one at line 300, `hoodie.index.cache.rocksdb.enabled`) still uses the old `hoodie.index.cache.rocksdb.*` prefix. Could you update these two so the doc names each config consistently? ⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality. -- 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]
Re: [PR] docs: rfc-107 Support data partition aware RocksDB RecordIndexBackend [hudi]
hudi-bot commented on PR #19046: URL: https://github.com/apache/hudi/pull/19046#issuecomment-5262423820 ## CI report: * 782f4fd9d26907d295a3e7a5387ecb3b8e10fd63 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16229) Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] docs: rfc-107 Support data partition aware RocksDB RecordIndexBackend [hudi]
hudi-bot commented on PR #19046: URL: https://github.com/apache/hudi/pull/19046#issuecomment-5261883618 ## CI report: * 1ede6e3d0ee6504130779acd72bd6020a711869f Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16224) * 782f4fd9d26907d295a3e7a5387ecb3b8e10fd63 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16229) Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] docs: rfc-107 Support data partition aware RocksDB RecordIndexBackend [hudi]
hudi-bot commented on PR #19046: URL: https://github.com/apache/hudi/pull/19046#issuecomment-5261839809 ## CI report: * 1ede6e3d0ee6504130779acd72bd6020a711869f Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16224) * 782f4fd9d26907d295a3e7a5387ecb3b8e10fd63 UNKNOWN Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] docs: rfc-107 Support data partition aware RocksDB RecordIndexBackend [hudi]
HuangZhenQiu commented on code in PR #19046: URL: https://github.com/apache/hudi/pull/19046#discussion_r3763325327 ## rfc/rfc-107/rfc-107.md: ## @@ -0,0 +1,317 @@ + +# 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
Re: [PR] docs: rfc-107 Support data partition aware RocksDB RecordIndexBackend [hudi]
HuangZhenQiu commented on code in PR #19046: URL: https://github.com/apache/hudi/pull/19046#discussion_r3763324208 ## rfc/rfc-107/rfc-107.md: ## @@ -0,0 +1,317 @@ + +# 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
Re: [PR] docs: rfc-107 Support data partition aware RocksDB RecordIndexBackend [hudi]
danny0405 commented on code in PR #19046: URL: https://github.com/apache/hudi/pull/19046#discussion_r3763156783 ## rfc/rfc-107/rfc-107.md: ## @@ -0,0 +1,317 @@ + +# 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
Re: [PR] docs: rfc-107 Support data partition aware RocksDB RecordIndexBackend [hudi]
danny0405 commented on code in PR #19046: URL: https://github.com/apache/hudi/pull/19046#discussion_r3763109581 ## rfc/rfc-107/rfc-107.md: ## @@ -0,0 +1,317 @@ + +# 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
Re: [PR] docs: rfc-107 Support data partition aware RocksDB RecordIndexBackend [hudi]
hudi-bot commented on PR #19046: URL: https://github.com/apache/hudi/pull/19046#issuecomment-5257036202 ## CI report: * 1ede6e3d0ee6504130779acd72bd6020a711869f Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16224) Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] docs: rfc-107 Support data partition aware RocksDB RecordIndexBackend [hudi]
hudi-bot commented on PR #19046: URL: https://github.com/apache/hudi/pull/19046#issuecomment-5256189481 ## CI report: * f544ea0d43f5f6494e0d1d05f02d658d013af9fa Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16205) * 1ede6e3d0ee6504130779acd72bd6020a711869f Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16224) Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] docs: rfc-107 Support data partition aware RocksDB RecordIndexBackend [hudi]
hudi-bot commented on PR #19046: URL: https://github.com/apache/hudi/pull/19046#issuecomment-5256002271 ## CI report: * f544ea0d43f5f6494e0d1d05f02d658d013af9fa Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16205) * 1ede6e3d0ee6504130779acd72bd6020a711869f UNKNOWN Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] docs: rfc-107 Support data partition aware RocksDB RecordIndexBackend [hudi]
HuangZhenQiu commented on PR #19046: URL: https://github.com/apache/hudi/pull/19046#issuecomment-5256095890 @danny0405 Thanks for your comments. I revised corresponding description according to the solution we aligned. -- 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]
Re: [PR] docs: rfc-107 Support data partition aware RocksDB RecordIndexBackend [hudi]
HuangZhenQiu commented on code in PR #19046: URL: https://github.com/apache/hudi/pull/19046#discussion_r3759908448 ## rfc/rfc-107/rfc-107.md: ## @@ -0,0 +1,317 @@ + +# 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
Re: [PR] docs: rfc-107 Support data partition aware RocksDB RecordIndexBackend [hudi]
HuangZhenQiu commented on code in PR #19046: URL: https://github.com/apache/hudi/pull/19046#discussion_r3759777323 ## rfc/rfc-107/rfc-107.md: ## @@ -0,0 +1,317 @@ + +# 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
Re: [PR] docs: rfc-107 Support data partition aware RocksDB RecordIndexBackend [hudi]
HuangZhenQiu commented on code in PR #19046: URL: https://github.com/apache/hudi/pull/19046#discussion_r3759750309 ## rfc/rfc-107/rfc-107.md: ## @@ -0,0 +1,358 @@ + +# 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)
Re: [PR] docs: rfc-107 Support data partition aware RocksDB RecordIndexBackend [hudi]
danny0405 commented on code in PR #19046: URL: https://github.com/apache/hudi/pull/19046#discussion_r3754457407 ## rfc/rfc-107/rfc-107.md: ## @@ -0,0 +1,317 @@ + +# 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
Re: [PR] docs: rfc-107 Support data partition aware RocksDB RecordIndexBackend [hudi]
danny0405 commented on code in PR #19046: URL: https://github.com/apache/hudi/pull/19046#discussion_r3754449732 ## rfc/rfc-107/rfc-107.md: ## @@ -0,0 +1,358 @@ + +# 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
Re: [PR] docs: rfc-107 Support data partition aware RocksDB RecordIndexBackend [hudi]
danny0405 commented on code in PR #19046: URL: https://github.com/apache/hudi/pull/19046#discussion_r3754430182 ## rfc/rfc-107/rfc-107.md: ## @@ -0,0 +1,317 @@ + +# 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
Re: [PR] docs: rfc-107 Support data partition aware RocksDB RecordIndexBackend [hudi]
hudi-bot commented on PR #19046: URL: https://github.com/apache/hudi/pull/19046#issuecomment-5243642889 ## CI report: * f544ea0d43f5f6494e0d1d05f02d658d013af9fa Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16205) Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] docs: rfc-107 Support data partition aware RocksDB RecordIndexBackend [hudi]
hudi-agent commented on code in PR #19046: URL: https://github.com/apache/hudi/pull/19046#discussion_r3751318473 ## rfc/rfc-107/rfc-107.md: ## @@ -0,0 +1,317 @@ + +# 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
Re: [PR] docs: rfc-107 Support data partition aware RocksDB RecordIndexBackend [hudi]
hudi-bot commented on PR #19046: URL: https://github.com/apache/hudi/pull/19046#issuecomment-5242690517 ## CI report: * 0e3b2abca842395b8ff1f1e63a8dcc5d7802d2fa Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16158) * f544ea0d43f5f6494e0d1d05f02d658d013af9fa Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16205) Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] docs: rfc-107 Support data partition aware RocksDB RecordIndexBackend [hudi]
hudi-bot commented on PR #19046: URL: https://github.com/apache/hudi/pull/19046#issuecomment-5242627479 ## CI report: * 0e3b2abca842395b8ff1f1e63a8dcc5d7802d2fa Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16158) * f544ea0d43f5f6494e0d1d05f02d658d013af9fa UNKNOWN Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] docs: rfc-107 Support data partition aware RocksDB RecordIndexBackend [hudi]
HuangZhenQiu commented on code in PR #19046: URL: https://github.com/apache/hudi/pull/19046#discussion_r3751045899 ## rfc/rfc-107/rfc-107.md: ## @@ -0,0 +1,317 @@ + +# 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
Re: [PR] docs: rfc-107 Support data partition aware RocksDB RecordIndexBackend [hudi]
HuangZhenQiu commented on code in PR #19046: URL: https://github.com/apache/hudi/pull/19046#discussion_r3751035811 ## rfc/rfc-107/rfc-107.md: ## @@ -0,0 +1,317 @@ + +# 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
Re: [PR] docs: rfc-107 Support data partition aware RocksDB RecordIndexBackend [hudi]
cshuo commented on code in PR #19046: URL: https://github.com/apache/hudi/pull/19046#discussion_r3748173486 ## rfc/rfc-107/rfc-107.md: ## @@ -0,0 +1,317 @@ + +# 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.
Re: [PR] docs: rfc-107 Support data partition aware RocksDB RecordIndexBackend [hudi]
hudi-agent commented on code in PR #19046: URL: https://github.com/apache/hudi/pull/19046#discussion_r3736660808 ## rfc/rfc-107/rfc-107.md: ## @@ -0,0 +1,317 @@ + +# 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
Re: [PR] docs: rfc-107 Support data partition aware RocksDB RecordIndexBackend [hudi]
hudi-bot commented on PR #19046: URL: https://github.com/apache/hudi/pull/19046#issuecomment-5214715635 ## CI report: * 0e3b2abca842395b8ff1f1e63a8dcc5d7802d2fa Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16158) Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] docs: rfc-107 Support data partition aware RocksDB RecordIndexBackend [hudi]
hudi-bot commented on PR #19046: URL: https://github.com/apache/hudi/pull/19046#issuecomment-5213572495 ## CI report: * 8d2db2db33fc0e778a8767b7ada587c37a61d6c1 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16148) * 0e3b2abca842395b8ff1f1e63a8dcc5d7802d2fa Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16158) Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] docs: rfc-107 Support data partition aware RocksDB RecordIndexBackend [hudi]
hudi-bot commented on PR #19046: URL: https://github.com/apache/hudi/pull/19046#issuecomment-5213533689 ## CI report: * 8d2db2db33fc0e778a8767b7ada587c37a61d6c1 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16148) * 0e3b2abca842395b8ff1f1e63a8dcc5d7802d2fa UNKNOWN Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] docs: rfc-107 Support data partition aware RocksDB RecordIndexBackend [hudi]
hudi-bot commented on PR #19046: URL: https://github.com/apache/hudi/pull/19046#issuecomment-5210838787 ## CI report: * 8d2db2db33fc0e778a8767b7ada587c37a61d6c1 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16148) Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] docs: rfc-107 Support data partition aware RocksDB RecordIndexBackend [hudi]
hudi-bot commented on PR #19046: URL: https://github.com/apache/hudi/pull/19046#issuecomment-5210212375 ## CI report: * 63428dc599ca1b08521f974d76f456cc60756b7c Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16147) * 8d2db2db33fc0e778a8767b7ada587c37a61d6c1 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16148) Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] docs: rfc-107 Support data partition aware RocksDB RecordIndexBackend [hudi]
hudi-bot commented on PR #19046: URL: https://github.com/apache/hudi/pull/19046#issuecomment-5210036240 ## CI report: * 7fc23e6387f45b4b02fbeb4d858b3b86351f5441 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16096) * 63428dc599ca1b08521f974d76f456cc60756b7c Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16147) * 8d2db2db33fc0e778a8767b7ada587c37a61d6c1 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16148) Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] docs: rfc-107 Support data partition aware RocksDB RecordIndexBackend [hudi]
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 @@ + +# 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
Re: [PR] docs: rfc-107 Support data partition aware RocksDB RecordIndexBackend [hudi]
hudi-bot commented on PR #19046: URL: https://github.com/apache/hudi/pull/19046#issuecomment-5209665698 ## CI report: * 7fc23e6387f45b4b02fbeb4d858b3b86351f5441 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16096) * 63428dc599ca1b08521f974d76f456cc60756b7c Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16147) * 8d2db2db33fc0e778a8767b7ada587c37a61d6c1 UNKNOWN Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] docs: rfc-107 Support data partition aware RocksDB RecordIndexBackend [hudi]
hudi-bot commented on PR #19046: URL: https://github.com/apache/hudi/pull/19046#issuecomment-5209637003 ## CI report: * 7fc23e6387f45b4b02fbeb4d858b3b86351f5441 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16096) * 63428dc599ca1b08521f974d76f456cc60756b7c Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16147) Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] docs: rfc-107 Support data partition aware RocksDB RecordIndexBackend [hudi]
hudi-bot commented on PR #19046: URL: https://github.com/apache/hudi/pull/19046#issuecomment-5209609280 ## CI report: * 7fc23e6387f45b4b02fbeb4d858b3b86351f5441 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16096) * 63428dc599ca1b08521f974d76f456cc60756b7c UNKNOWN Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] docs: rfc-107 Support data partition aware RocksDB RecordIndexBackend [hudi]
HuangZhenQiu commented on code in PR #19046: URL: https://github.com/apache/hudi/pull/19046#discussion_r3732216357 ## rfc/rfc-107/rfc-107.md: ## @@ -0,0 +1,358 @@ + +# 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)
Re: [PR] docs: rfc-107 Support data partition aware RocksDB RecordIndexBackend [hudi]
HuangZhenQiu commented on code in PR #19046: URL: https://github.com/apache/hudi/pull/19046#discussion_r3732209034 ## rfc/rfc-107/rfc-107.md: ## @@ -0,0 +1,358 @@ + +# 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)
Re: [PR] docs: rfc-107 Support data partition aware RocksDB RecordIndexBackend [hudi]
HuangZhenQiu commented on code in PR #19046: URL: https://github.com/apache/hudi/pull/19046#discussion_r3732100644 ## rfc/rfc-107/rfc-107.md: ## @@ -0,0 +1,358 @@ + +# 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)
Re: [PR] docs: rfc-107 Support data partition aware RocksDB RecordIndexBackend [hudi]
HuangZhenQiu commented on code in PR #19046: URL: https://github.com/apache/hudi/pull/19046#discussion_r3732096088 ## rfc/rfc-107/rfc-107.md: ## @@ -0,0 +1,358 @@ + +# 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)
Re: [PR] docs: rfc-107 Support data partition aware RocksDB RecordIndexBackend [hudi]
HuangZhenQiu commented on code in PR #19046: URL: https://github.com/apache/hudi/pull/19046#discussion_r3732061352 ## rfc/rfc-107/rfc-107.md: ## @@ -0,0 +1,306 @@ + +# 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. Review Comment: Yes, With in Flink Hudi integration, we already supported three types IndexBackend. Every of these solution has a local cache either in RocksDB or memory. From our testing, each of them has the Cache thrashing for different workload pattern. -- 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]
Re: [PR] docs: rfc-107 Support data partition aware RocksDB RecordIndexBackend [hudi]
HuangZhenQiu commented on code in PR #19046: URL: https://github.com/apache/hudi/pull/19046#discussion_r3732054100 ## rfc/rfc-107/rfc-107.md: ## @@ -0,0 +1,358 @@ + +# 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. Review Comment: Agree. Users may always use memory as unbound resource. From our testing, the largest table has 100TB record index. Each of input message size is 30KB to 50KB, no matter use the GlobalRecordLevelIndexBackend with mini batch or RocksDBIndexBackend, it requires a large job with 20GB memory. The proposal here is to try to find an optimization to minimize the memory, compute and disk usage. -- 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]
Re: [PR] docs: rfc-107 Support data partition aware RocksDB RecordIndexBackend [hudi]
cshuo commented on code in PR #19046: URL: https://github.com/apache/hudi/pull/19046#discussion_r3657182641 ## rfc/rfc-107/rfc-107.md: ## @@ -0,0 +1,358 @@ + +# 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
Re: [PR] docs: rfc-107 Support data partition aware RocksDB RecordIndexBackend [hudi]
hudi-agent commented on code in PR #19046: URL: https://github.com/apache/hudi/pull/19046#discussion_r372432 ## rfc/rfc-107/rfc-107.md: ## @@ -0,0 +1,320 @@ + +# 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
Re: [PR] docs: rfc-107 Support data partition aware RocksDB RecordIndexBackend [hudi]
hudi-bot commented on PR #19046: URL: https://github.com/apache/hudi/pull/19046#issuecomment-5196151368 ## CI report: * 7fc23e6387f45b4b02fbeb4d858b3b86351f5441 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16096) Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] docs: rfc-107 Support data partition aware RocksDB RecordIndexBackend [hudi]
hudi-bot commented on PR #19046: URL: https://github.com/apache/hudi/pull/19046#issuecomment-5195189940 ## CI report: * 3e527ec939a4dd7353681fc6b0529f78203694e1 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16075) * 7fc23e6387f45b4b02fbeb4d858b3b86351f5441 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16096) Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] docs: rfc-107 Support data partition aware RocksDB RecordIndexBackend [hudi]
hudi-bot commented on PR #19046: URL: https://github.com/apache/hudi/pull/19046#issuecomment-5195126589 ## CI report: * 3e527ec939a4dd7353681fc6b0529f78203694e1 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16075) * 7fc23e6387f45b4b02fbeb4d858b3b86351f5441 UNKNOWN Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] docs: rfc-107 Support data partition aware RocksDB RecordIndexBackend [hudi]
hudi-bot commented on PR #19046: URL: https://github.com/apache/hudi/pull/19046#issuecomment-5188644965 ## CI report: * 3e527ec939a4dd7353681fc6b0529f78203694e1 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16075) Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] docs: rfc-107 Support data partition aware RocksDB RecordIndexBackend [hudi]
hudi-bot commented on PR #19046: URL: https://github.com/apache/hudi/pull/19046#issuecomment-5187993189 ## CI report: * a3bcf4303bfc380c95a1ba32675a51de665cc8ec Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15712) * 3e527ec939a4dd7353681fc6b0529f78203694e1 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16075) Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
Re: [PR] docs: rfc-107 Support data partition aware RocksDB RecordIndexBackend [hudi]
hudi-agent commented on code in PR #19046: URL: https://github.com/apache/hudi/pull/19046#discussion_r3718201590 ## rfc/rfc-107/rfc-107.md: ## @@ -0,0 +1,341 @@ + +# 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
Re: [PR] docs: rfc-107 Support data partition aware RocksDB RecordIndexBackend [hudi]
hudi-bot commented on PR #19046: URL: https://github.com/apache/hudi/pull/19046#issuecomment-5187958056 ## CI report: * a3bcf4303bfc380c95a1ba32675a51de665cc8ec Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15712) * 3e527ec939a4dd7353681fc6b0529f78203694e1 UNKNOWN Bot commands @hudi-bot supports the following commands: - `@hudi-bot run azure` re-run the last Azure build -- 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]
