You are viewing a plain text version of this content. The canonical link for it is here.

Posted to dev@hudi.apache.org by "vinothchandar (via GitHub)" <gi...@apache.org> on 2026/03/09 18:25:22 UTC

[D] RocksDB as The Replica of MDT/RLI [hudi]

GitHub user vinothchandar created a discussion: RocksDB as The Replica of MDT/RLI

### RocksDB as The Replica of MDT

### The update to RocksDB

The rocksDB instantces are initialized and bootstrapped from scratch by reading the full MDT RLI index for each job restart or task failover.

The incremental index upserts inferred from the data inputs are applied directly on these RocksDB instances, these index upserts are pass along with the data payloads altogether to the `IndexWrite` op for actual MDT update. The MDT update happens in the same lifecycle of data records write and the incremental upserts are a replica image of the upserts into the RocksDB.

The RLI would be utilized for two cases:

- serves as the source of truth of the index mapping and been utilized in the RocksDB bootstrap
- cross engine compatibility

The new write flow with RockDB replica:
<img width="4696" height="1386" alt="image" src="https://github.com/user-attachments/assets/d99a624e-84b8-4c46-9a60-2799b7cb79d3" />


### The Clean/Eviction of Index Payloads in RocksDB

For global RLI, the rocksDB instance would be closed and removed each time a task fails over or got a job restart.

For partitioned RLI, for local RocksDB instance per `BucketAssign` task, the paylods under the same data partition is stored as a separate column family, when the data partition is based on datetime, the column family can be dropped very efficiently with a configurable partition lookup TTL.

### The Additional Storage Cost

A local transcoding from HFile to RocksDB sst shows that the storage size diff mainly comes from the compression codec, Hudi uses gzip as default while RocksDB uses snappy, and the RocksDB storage size turns out to be a nearly 2x size storage against the native HFile(the RocksDB WAL has been disabled explicitly since it is not necessary in our use cases).

<img width="616" height="244" alt="image" src="https://github.com/user-attachments/assets/ba01023f-f744-454c-839e-e5bebf0ba28c" />
<img width="727" height="192" alt="image" src="https://github.com/user-attachments/assets/708e1d98-b778-43f9-a157-7d3fa131ee5f" />



GitHub link: https://github.com/apache/hudi/discussions/18296

----
This is an automatically sent email for dev@hudi.apache.org.
To unsubscribe, please send an email to: dev-unsubscribe@hudi.apache.org

Re: [D] RocksDB as The Replica of MDT/RLI [hudi]

Posted by "danny0405 (via GitHub)" <gi...@apache.org>.

GitHub user danny0405 added a comment to the discussion: RocksDB as The Replica of MDT/RLI

> I don't like how we are coupling index choices and concurrency models.

yeah, the simple bucket index is required to impl NBCC now and we may need more flexible and general design for concurrent modifications in streaming concurrent write scenarios.

> can you please explain in detail,how the failover and OCC handling are related.

I think we can categorize the concurrent write cases into two: write with conflicts and write without conflicts.

1. If the write detects conflicts, the whole job/task will trigger failover and the RocksDB replica will rebootstrap from scrach, which can ensure the consistency of the index backend akka to MDT RLI index, but this needs to enable the early confclit detection: https://github.com/apache/hudi/pull/6133, the pre-commit conflict resolution does not work well for Flink streaming because it happens after a successful checkpoint, Hudi deems the write as failed with there is conflict while Flink deems the write as successful(from the latest successful checkpoint), there is gap, the early confclit resolutuon is required here.

2. If the write does not detect confclits, there are still cases that another concurrent write modify the table with new record locations, the solution is we might need a early detection of the index backend freshness before each write: maintain a mappings between job-id to instant time so we can load the index changes maded from concurrent writers incrementally.(put the job-id in commit metadata or maintain it on the coordinator). This introduces a lot of complexities though, I'm expecting a more general solution for NBCC that is index type agnostic and not struggle in this index concurrent index trap.

Here is the table for support of cuncurrent modifications with Flink RLI:

| use case/concurrency mode | OCC | NBCC |
|---|---|---|
| write & write | Y(with early conflic detection and index refreshing) | N |
| write & compaction | Y | N |
| write & clustering | Y(with early index refreshing) | N |

GitHub link: https://github.com/apache/hudi/discussions/18296#discussioncomment-16171800

----
This is an automatically sent email for dev@hudi.apache.org.
To unsubscribe, please send an email to: dev-unsubscribe@hudi.apache.org

Re: [D] RocksDB as The Replica of MDT/RLI [hudi]

Posted by "cshuo (via GitHub)" <gi...@apache.org>.

GitHub user cshuo added a comment to the discussion: RocksDB as The Replica of MDT/RLI

> Is there no way to make hudi commit first, and then flink proceeds to checkpoint.

@vinothchandar My previous explanation was addressing this question—why simply moving the Hudi commit before checkpoint completion would not be safe.

> I heard the opposite, where checkpoint can complete, but hudi commit can fail. Can that happen as well?

Yes, with the current implementation based on flink standard two-phase commit, the checkpoint completes first and then Hudi commits afterward. It's possible that hudi commits can still fail after checkpoint has completed successfully. That failure does not invalidate the completed checkpoint. 

GitHub link: https://github.com/apache/hudi/discussions/18296#discussioncomment-18359781

----
This is an automatically sent email for dev@hudi.apache.org.
To unsubscribe, please send an email to: dev-unsubscribe@hudi.apache.org

Re: [D] RocksDB as The Replica of MDT/RLI [hudi]

Posted by "codope (via GitHub)" <gi...@apache.org>.

GitHub user codope added a comment to the discussion: RocksDB as The Replica of MDT/RLI

This is a very interesting proposal! I had a few questions to understand the motivation and mechanics of the proposal:

1. What is the motivation behind the proposal? Is it just to cut down cloud storage i/o and log scan overhead (and hence latency per write request)? I can see how local, materialized index can help esp for streaming writes, but what were other concerns? Is it just for Flink, or Spark is also being considered? I think if we can update the discussion with just a short para on motivation/background, it would be very helpful.
2. On the concurrent upsert consistency discussion above, re-bootstrapping from MDT on failover is correct for crash recovery. However, what happens in case of the following scenario with NBCC enabled?
- Writer A starts, bootstraps RocksDB from MDT at instant T1
- Writer B starts, bootstraps RocksDB from MDT at instant T1
- Writer A commits records {k1, k2} --> MDT updated at T2
- Writer B is still running with stale RocksDB (knows nothing about k1, k2)
- Writer B receives an upsert for k1 --> RocksDB says "not found" --> Will it then tag as INSERT instead of UPDATE which will lead to duplicate record in a different file group?
The question is whether NBCC's existing mechanisms are sufficient, or if the stale RocksDB creates scenarios that require additional handling. Another related question is how does the RocksDB replica handle file group replacement from concurrent compaction/clustering?
3. On the storage overhead, the 2x number is for the initial static size (compression difference) right? During active writes, esp under heavy load, it could be more due to unmerged L0 files and multiple levels of SST files. So, total disk usage could peak at `compression * compaction` overhead. It would be nice to benchmark the prototype with active writes. Also, on memory overhead, wouldn't the RocksDB block cache compete with JVM heap used by Hudi's own processing?
4. Bootstrap latency could be a concern for large tables. IIRC, RLI is about 50-70 bytes per record. Even for a table with 1B records, and assuming about 500 Mbps cloud throughput, that's about 2 minutes. Is there a way to persist the RocksDB state across restarts?
5. How does this coexist with or replace the existing `RecordIndexCache`? Should updates be buffered in memory and only flushed to RocksDB on commit success? (This is essentially what `RecordIndexCache` does with checkpoint-based eviction right?)
6. Looking at the new write flow diagram, the third shuffle by RLI/Si shard ids is new. Is this shuffle strictly necessary? Could the `IndexWrite Op` be collocated with the `StreamWrite Op` to avoid one shuffle?

GitHub link: https://github.com/apache/hudi/discussions/18296#discussioncomment-16152659

----
This is an automatically sent email for dev@hudi.apache.org.
To unsubscribe, please send an email to: dev-unsubscribe@hudi.apache.org

Re: [D] RocksDB as The Replica of MDT/RLI [hudi]

Posted by "vinothchandar (via GitHub)" <gi...@apache.org>.

GitHub user vinothchandar edited a discussion: RocksDB as The Replica of MDT/RLI

### RocksDB as The Replica of MDT

### The update to RocksDB

The rocksDB instantces are initialized and bootstrapped from scratch by reading the full MDT RLI index for each job restart or task failover.

The incremental index upserts inferred from the data inputs are applied directly on these RocksDB instances, these index upserts are passed along with the data payloads altogether to the `IndexWrite` op for actual MDT update. The MDT update happens in the same lifecycle of data records write and the incremental upserts are a replica image of the upserts into the RocksDB.

The RLI would be utilized for two cases:

- serves as the source of truth of the index mapping and been utilized in the RocksDB bootstrap
- cross engine compatibility

The new write flow with RockDB replica:
<img width="4696" height="1386" alt="image" src="https://github.com/user-attachments/assets/d99a624e-84b8-4c46-9a60-2799b7cb79d3" />


### The Clean/Eviction of Index Payloads in RocksDB

For global RLI, the rocksDB instance would be closed and removed each time a task fails over or got a job restart.

For partitioned RLI, for local RocksDB instance per `BucketAssign` task, the paylods under the same data partition is stored as a separate column family, when the data partition is based on datetime, the column family can be dropped very efficiently with a configurable partition lookup TTL.

### The Additional Storage Cost

A local transcoding from HFile to RocksDB sst shows that the storage size diff mainly comes from the compression codec, Hudi uses gzip as default while RocksDB uses snappy, and the RocksDB storage size turns out to be a nearly 2x size storage against the native HFile(the RocksDB WAL has been disabled explicitly since it is not necessary in our use cases).

<img width="616" height="244" alt="image" src="https://github.com/user-attachments/assets/ba01023f-f744-454c-839e-e5bebf0ba28c" />
<img width="727" height="192" alt="image" src="https://github.com/user-attachments/assets/708e1d98-b778-43f9-a157-7d3fa131ee5f" />



GitHub link: https://github.com/apache/hudi/discussions/18296

----
This is an automatically sent email for dev@hudi.apache.org.
To unsubscribe, please send an email to: dev-unsubscribe@hudi.apache.org

Re: [D] RocksDB as The Replica of MDT/RLI [hudi]

Posted by "HuangZhenQiu (via GitHub)" <gi...@apache.org>.

GitHub user HuangZhenQiu added a comment to the discussion: RocksDB as The Replica of MDT/RLI

@vinothchandar 
It is a known and resolved issue. Basically, even for append-only use cases, Flink ingestion could first finished checkpoint and later fail due the Hudi commit failure. In this case, job will fail and restart.

 How do we recover from this issue? 

We include Kafka offset into each of Hudi commit metadata. When job restart from failure, it will use the kafka offset in the last completed hudi commit to start rather than the kafka offset in flink checkpoint. We simply only use flink checkpoint mechanism for two phase commit protocol implementation not for any state management and recovery.

Will the solution usable for Flink upsert?

Yes, it will still work for Fink upsert, not matter we bootstrap record index or not, we don't relies on the state in Flink checkpoint to recovery. We may still use the Kafka offset in last completed Hudi commit to recover to guarantee exactly once data write to the target Hudi table






GitHub link: https://github.com/apache/hudi/discussions/18296#discussioncomment-18390480

----
This is an automatically sent email for dev@hudi.apache.org.
To unsubscribe, please send an email to: dev-unsubscribe@hudi.apache.org

Re: [D] RocksDB as The Replica of MDT/RLI [hudi]

Posted by "vinothchandar (via GitHub)" <gi...@apache.org>.

GitHub user vinothchandar added a comment to the discussion: RocksDB as The Replica of MDT/RLI

> The concurrent upsert consistency still works under OCC since the task failover would anyway retrigger the full bootstrap of the RocksDB replica, as of now, simple bucket index is a prerequisite for NBCC so it should not be a strong concern or blocker for the RLI.

I don't like how we are coupling index choices and concurrency models. Before we go to NBCC, can you please explain in detail,how the failover and OCC handling are related.. Even if there is no multi-writer, just some async clustering -- the location of the record can change (even without failovers) and the writers need to know this right. 

@danny0405 I think we need to handle this. Restricting clustering also is not a good solution to this

GitHub link: https://github.com/apache/hudi/discussions/18296#discussioncomment-16167851

----
This is an automatically sent email for dev@hudi.apache.org.
To unsubscribe, please send an email to: dev-unsubscribe@hudi.apache.org

Re: [D] RocksDB as The Replica of MDT/RLI [hudi]

Posted by "vinothchandar (via GitHub)" <gi...@apache.org>.

GitHub user vinothchandar added a comment to the discussion: RocksDB as The Replica of MDT/RLI

Starting a thread here in the discussion around issues flagged in RFC-107 (https://github.com/apache/hudi/pull/19046)

For background, we landed the first version of RFC-107 that adds a rocksDB backed partitioned index backend. The idea is simple, we cache fully all "hot" partitions affected by a write such that the rocksDB state scales `O(num_storage_partitions_actively_written_to)` . RocksDB is the source of truth i.e assume if there is a location entry in local rocksDB state, it is correct.

**Few scenarios complicate this:**

1. Concurrent async clustering, that can change the file group location of a record.
2. Concurrent writer (OCC), which can write records into same file group as a writer with Rocksdb record index backend. But, this does not change the location i.e key to file group mapping.
3. Concurrent writer (NBCC) which can write records into same file group as writer with Rocksdb record index backend. Even this should not change the record location.

**Questions that arise:**

1. From https://github.com/apache/hudi/pull/19046#issuecomment-5351517103 , `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.` => What does this mean for scenarios 2,3? Flink won't replay uncommitted input streams, right since checkpoint is successful. Hudi cannot commit either because writes are conflicting?
2. For scenario 1, we need some safe, atomic recovery protocol? What is that? we cannot disallow clustering altogether.
3. We can throw an exception for OCC, with rocksDB RecordIndexBackend is enabled? i.e I am saying we only support NBCC (this does not need the OCC like error-retry loop and is way more streaming friendly). (note this is 180* from my original thinking that NBCC is the harder of the two. I realize now that the record location does not change.). Poke holes at this?

GitHub link: https://github.com/apache/hudi/discussions/18296#discussioncomment-18240183

----
This is an automatically sent email for dev@hudi.apache.org.
To unsubscribe, please send an email to: dev-unsubscribe@hudi.apache.org

Re: [D] RocksDB as The Replica of MDT/RLI [hudi]

Posted by "vinothchandar (via GitHub)" <gi...@apache.org>.

GitHub user vinothchandar added a comment to the discussion: RocksDB as The Replica of MDT/RLI

> we meed some early conflict resolution happens within the checkpoint lifecycle to intervene and fail the checkpoint 

This is still best effort right.. Is there no way to make hudi commit first, and then flink proceeds to checkpoint. like a pre-checkpoint hook? in flink? is this a flink limitation?

GitHub link: https://github.com/apache/hudi/discussions/18296#discussioncomment-18259162

----
This is an automatically sent email for dev@hudi.apache.org.
To unsubscribe, please send an email to: dev-unsubscribe@hudi.apache.org

Re: [D] RocksDB as The Replica of MDT/RLI [hudi]

Posted by "danny0405 (via GitHub)" <gi...@apache.org>.

GitHub user danny0405 added a comment to the discussion: RocksDB as The Replica of MDT/RLI

I'm assuming you already read the [Flink RLI RFC](https://github.com/apache/hudi/pull/17610).

1. we did some local benchmark with MDT(plus hotspot cache in memory) as the index backend, and it turns out to be an average 50ms access latency per record query which does not scale well for steaming. Thus a local fast lookup replica is necessary, we have to solutions on table:

1. use RocksDB replica as the mirror image of the MDT, and always replicate the index payloads from MDT to local RocksDB to ensure the performance;
2. still uses the MDT, but introduces more caches: data block cache, local index metadata, bloomfilter cache, local files cache on SSD(secondary cache) and a fallback to remote MDT queries.

The 2nd one would take a lot of efforts and we deem it as long-term solution, as of now, we prefer 1.

2. NBCC is only working with simple bucket index, so not very related.

3. MDT does not clean the legacy files on each compaction immediately so the estimation still makes sense, based on the compression frequency configured(both can adjust the cleaning/compaction strategy), the RocksDB block cache utilizes off-heap memory I think. But yes, there could be resouce contention just like the Flink-State index which proves good perf in production already. And we will definitely do a lot more benchmarks for long-running job, that's a collaboration with Uber team.

4. the bootstrap op parallelism is scalable, the bottleneck might be the full scan load time for a single file group of MDT. As long as the time is less than the checkpoint timeout, it works well and the bootstrap only happens once for a job restart. Why not persistent RocksDB instances across jobs recovery: a). don't think we have good way to main the consistency between the RocksDB and MDT based on the complexities of DT and MDT consistency. b). another reason is the RocksDB local storage is within local container which will be release once the task is killed(we can not use remote storage which kills the perf).

5. The RocksDB is updated per-record in the `BucketAssign` op, and the `RocksDB` itself has memtable to manage the buffer and flush of the index records. Each time the task failsover, the bootstrap retriggers to ensure the integrity.

6. we want the write scalable as independent and we don't want to want to every file group from a single write task to limit the small files.

GitHub link: https://github.com/apache/hudi/discussions/18296#discussioncomment-16154189

----
This is an automatically sent email for dev@hudi.apache.org.
To unsubscribe, please send an email to: dev-unsubscribe@hudi.apache.org

Re: [D] RocksDB as The Replica of MDT/RLI [hudi]

Posted by "vinothchandar (via GitHub)" <gi...@apache.org>.

GitHub user vinothchandar added a comment to the discussion: RocksDB as The Replica of MDT/RLI

Early conflict resolution is best-effort. It cannot guarantee catching conflicts before we reach pre-commit conflict resolution. 

GitHub link: https://github.com/apache/hudi/discussions/18296#discussioncomment-16201536

----
This is an automatically sent email for dev@hudi.apache.org.
To unsubscribe, please send an email to: dev-unsubscribe@hudi.apache.org

Re: [D] RocksDB as The Replica of MDT/RLI [hudi]

Posted by "vinothchandar (via GitHub)" <gi...@apache.org>.

GitHub user vinothchandar edited a comment on the discussion: RocksDB as The Replica of MDT/RLI

> The incremental index upserts inferred from the data inputs are applied directly on these RocksDB instances

**Question 1**: If we treat rocksDB as the primary source of truth during writes, how does concurrent updates from another writer get visible correctly during `BucketAssignOp` stage? We need a solution that can work with NBCC and multiple writers. Can you expand on this in a comment below? 

> the paylods under the same data partition is stored as a separate column family,  

+1. This will also mean that for partitioned RLI i.e records always have a immutable partitioning field(s), the index size can be simply `O(size_of_data_partitions_written_to)` and not `O(total_size_of_table)` (as is the case for global indexing or mutable partition fields) 

> the RocksDB storage size turns out to be a nearly 2x size storage against the native HFile

**Question 2** : the 2x is due to just compression?  I think there will be an additional 2x additional storage for un-compacted updates, since rocksdb will also do its own async compaction periodically. 

GitHub link: https://github.com/apache/hudi/discussions/18296#discussioncomment-16056769

----
This is an automatically sent email for dev@hudi.apache.org.
To unsubscribe, please send an email to: dev-unsubscribe@hudi.apache.org

Re: [D] RocksDB as The Replica of MDT/RLI [hudi]

Posted by "vinothchandar (via GitHub)" <gi...@apache.org>.

GitHub user vinothchandar added a comment to the discussion: RocksDB as The Replica of MDT/RLI

> “Record location does not change” only covers existing keys. For a new key, multiple writers may all miss it in their local RLI replicas and independently assign it to different file groups.

you are right. that was my miss. So, how would we make this safe in face of multi-writers? (for general inserts, doing uniqueness check is doable, but expensive. we have not taken that up. separate topic). 

GitHub link: https://github.com/apache/hudi/discussions/18296#discussioncomment-18259149

----
This is an automatically sent email for dev@hudi.apache.org.
To unsubscribe, please send an email to: dev-unsubscribe@hudi.apache.org

Re: [D] RocksDB as The Replica of MDT/RLI [hudi]

Posted by "vinothchandar (via GitHub)" <gi...@apache.org>.

GitHub user vinothchandar added a comment to the discussion: RocksDB as The Replica of MDT/RLI

@danny0405 @HuangZhenQiu Can we do a pointed discussion on these scenarios here. and mitigations/how we guard users from bad scenarios. 

GitHub link: https://github.com/apache/hudi/discussions/18296#discussioncomment-18240195

----
This is an automatically sent email for dev@hudi.apache.org.
To unsubscribe, please send an email to: dev-unsubscribe@hudi.apache.org

Re: [D] RocksDB as The Replica of MDT/RLI [hudi]

Posted by "vinothchandar (via GitHub)" <gi...@apache.org>.

GitHub user vinothchandar added a comment to the discussion: RocksDB as The Replica of MDT/RLI

Thanks for confirming @cshuo . 

> Even if the sink successfully commits to Hudi, a failure in any other operator can still abort the overall checkpoint.

this is actually workable. I heard the opposite, where checkpoint can complete, but hudi commit can fail. Can that happen as well?

GitHub link: https://github.com/apache/hudi/discussions/18296#discussioncomment-18356728

----
This is an automatically sent email for dev@hudi.apache.org.
To unsubscribe, please send an email to: dev-unsubscribe@hudi.apache.org

Re: [D] RocksDB as The Replica of MDT/RLI [hudi]

Posted by "vinothchandar (via GitHub)" <gi...@apache.org>.

GitHub user vinothchandar edited a discussion: RocksDB as The Replica of MDT/RLI

### RocksDB as The Replica of MDT

The initial Flink RLI RFC: https://github.com/apache/hudi/pull/17610

### The update to RocksDB

The rocksDB instantces are initialized and bootstrapped from scratch by reading the full MDT RLI index for each job restart or task failover.

The incremental index upserts inferred from the data inputs are applied directly on these RocksDB instances, these index upserts are passed along with the data payloads altogether to the `IndexWrite` op for actual MDT update. The MDT update happens in the same lifecycle of data records write and the incremental upserts are a replica image of the upserts into the RocksDB.

The RLI would be utilized for two cases:

- serves as the source of truth of the index mapping and been utilized in the RocksDB bootstrap
- cross engine compatibility

The new write flow with RockDB replica:
<img width="4696" height="1386" alt="image" src="https://github.com/user-attachments/assets/d99a624e-84b8-4c46-9a60-2799b7cb79d3" />


### The Clean/Eviction of Index Payloads in RocksDB

For global RLI, the rocksDB instance would be closed and removed each time a task fails over or got a job restart.

For partitioned RLI, for local RocksDB instance per `BucketAssign` task, the paylods under the same data partition is stored as a separate column family, when the data partition is based on datetime, the column family can be dropped very efficiently with a configurable partition lookup TTL.

### The Additional Storage Cost

A local transcoding from HFile to RocksDB sst shows that the storage size diff mainly comes from the compression codec, Hudi uses gzip as default while RocksDB uses snappy, and the RocksDB storage size turns out to be a nearly 2x size storage against the native HFile(the RocksDB WAL has been disabled explicitly since it is not necessary in our use cases).

<img width="616" height="244" alt="image" src="https://github.com/user-attachments/assets/ba01023f-f744-454c-839e-e5bebf0ba28c" />
<img width="727" height="192" alt="image" src="https://github.com/user-attachments/assets/708e1d98-b778-43f9-a157-7d3fa131ee5f" />



GitHub link: https://github.com/apache/hudi/discussions/18296

----
This is an automatically sent email for dev@hudi.apache.org.
To unsubscribe, please send an email to: dev-unsubscribe@hudi.apache.org

Re: [D] RocksDB as The Replica of MDT/RLI [hudi]

Posted by "vinothchandar (via GitHub)" <gi...@apache.org>.

GitHub user vinothchandar edited a comment on the discussion: RocksDB as The Replica of MDT/RLI

Early conflict resolution is best-effort. It cannot guarantee catching conflicts before we reach pre-commit conflict resolution. 

Can we revise the comment based on this assumption? @danny0405 

GitHub link: https://github.com/apache/hudi/discussions/18296#discussioncomment-16201536

----
This is an automatically sent email for dev@hudi.apache.org.
To unsubscribe, please send an email to: dev-unsubscribe@hudi.apache.org

Re: [D] RocksDB as The Replica of MDT/RLI [hudi]

Posted by "codope (via GitHub)" <gi...@apache.org>.

GitHub user codope added a comment to the discussion: RocksDB as The Replica of MDT/RLI

Thanks @danny0405 , I had not read the Flink RLI RFC. Will go through it first. Would be helpful to link that RFC in the description as well.

GitHub link: https://github.com/apache/hudi/discussions/18296#discussioncomment-16154807

----
This is an automatically sent email for dev@hudi.apache.org.
To unsubscribe, please send an email to: dev-unsubscribe@hudi.apache.org

Re: [D] RocksDB as The Replica of MDT/RLI [hudi]

Posted by "cshuo (via GitHub)" <gi...@apache.org>.

GitHub user cshuo added a comment to the discussion: RocksDB as The Replica of MDT/RLI

> This is still best effort right.. Is there no way to make hudi commit first, and then flink proceeds to checkpoint. like a pre-checkpoint hook? in flink? is this a flink limitation?

@vinothchandar Yeah, that's a limitation or transactional sink contract for flink.

A Flink checkpoint is completed only after all operators asynchronously finish and acknowledge their snapshots. Even if the sink successfully commits to Hudi, a failure in any other operator can still abort the overall checkpoint. Moreover, an aborted checkpoint does not necessarily fail or restart the job—Flink may simply continue and let a later checkpoint cover a longer processing interval.

This is why Flink’s transactional sink model uses checkpoint-coordinated two-phase commit: prepare during the checkpoint and make the external commit visible only after the global checkpoint completes. Flink writers for Iceberg, Delta etc,  follow the same pattern.

GitHub link: https://github.com/apache/hudi/discussions/18296#discussioncomment-18261200

----
This is an automatically sent email for dev@hudi.apache.org.
To unsubscribe, please send an email to: dev-unsubscribe@hudi.apache.org

Re: [D] RocksDB as The Replica of MDT/RLI [hudi]

Posted by "danny0405 (via GitHub)" <gi...@apache.org>.

GitHub user danny0405 edited a comment on the discussion: RocksDB as The Replica of MDT/RLI

I'm assuming you already read the [Flink RLI RFC](https://github.com/apache/hudi/pull/17610).

The 2nd one would take a lot of efforts and we deem it as long-term solution, as of now, we prefer 1.

2. NBCC is only working with simple bucket index, so not very related.

6. we want the write scalable as independent and we don't want to want to every file group from a single write task to limit the small files.

GitHub link: https://github.com/apache/hudi/discussions/18296#discussioncomment-16154189

----
This is an automatically sent email for dev@hudi.apache.org.
To unsubscribe, please send an email to: dev-unsubscribe@hudi.apache.org

Re: [D] RocksDB as The Replica of MDT/RLI [hudi]

Posted by "danny0405 (via GitHub)" <gi...@apache.org>.

GitHub user danny0405 added a comment to the discussion: RocksDB as The Replica of MDT/RLI

> Question 1: If we treat rocksDB as the primary source of truth during writes, how does concurrent updates from another writer get visible correctly during BucketAssignOp stage

The concurrent upsert consistency still works under OCC since the task failover would anyway retrigger the full bootstrap of the RocksDB replica, as of now, simple bucket index is a prerequisite for NBCC so it should not be a strong concern or blocker for the RLI.

> Question 2 : the 2x is due to just compression? I think there will be an additional 2x additional storage for un-compacted updates

While the RocksDB suggests light compression(LZ4/Snappy) for L0 ~ L2 to get best write throughput, and heavier compression (Zstd/Zlib) for L3+, for updates that un-compacted, the native MDT also got the similar case for its payloads in log files, it may relies on the compression frequency gap between MDT(10 delta commits trigger a compaction) and RocksDB(4 sst triggers a compaction).

GitHub link: https://github.com/apache/hudi/discussions/18296#discussioncomment-16061388

----
This is an automatically sent email for dev@hudi.apache.org.
To unsubscribe, please send an email to: dev-unsubscribe@hudi.apache.org

Re: [D] RocksDB as The Replica of MDT/RLI [hudi]

Posted by "danny0405 (via GitHub)" <gi...@apache.org>.

GitHub user danny0405 edited a comment on the discussion: RocksDB as The Replica of MDT/RLI

> I don't like how we are coupling index choices and concurrency models.

yeah, the simple bucket index is required to impl NBCC now and we may need more flexible and general design for concurrent modifications in streaming concurrent write scenarios.

> can you please explain in detail,how the failover and OCC handling are related.

I think we can categorize the concurrent write cases into two: write with conflicts and write without conflicts.

1. If the write detects conflicts, the whole job/task will trigger failover and the RocksDB replica will rebootstrap from scrach, which can ensure the consistency of the index backend akka to MDT RLI index, but this needs to enable the early confclit detection: https://github.com/apache/hudi/pull/6133, the pre-commit conflict resolution does not work well for Flink streaming because it happens after a successful checkpoint, Hudi deems the write as failed if there is conflict while Flink deems the write as successful(from the latest successful checkpoint), to fix gap, the early confclit resolutuon is required here.

2. If the write does not detect confclits, there are still cases that another concurrent write modify the table with new record locations, the solution is we might need a early detection of the index backend freshness before each write: maintain a mappings between job-id to instant time so we can load the index changes maded from concurrent writers incrementally.(put the job-id in commit metadata or maintain it on the coordinator). This introduces a lot of complexities though, I'm expecting a more general solution for NBCC that is index type agnostic and not struggle in this index concurrent modification trap.

Here is the table for support of cuncurrent modifications with Flink RLI:

| use case/concurrency mode | OCC | NBCC |
|---|---|---|
| write & write | Y(with early conflic detection and index refreshing) | N |
| write & compaction | Y | N |
| write & clustering | Y(with early index refreshing) | N |

GitHub link: https://github.com/apache/hudi/discussions/18296#discussioncomment-16171800

----
This is an automatically sent email for dev@hudi.apache.org.
To unsubscribe, please send an email to: dev-unsubscribe@hudi.apache.org

Re: [D] RocksDB as The Replica of MDT/RLI [hudi]

Posted by "danny0405 (via GitHub)" <gi...@apache.org>.

GitHub user danny0405 added a comment to the discussion: RocksDB as The Replica of MDT/RLI

updated

GitHub link: https://github.com/apache/hudi/discussions/18296#discussioncomment-16216239

----
This is an automatically sent email for dev@hudi.apache.org.
To unsubscribe, please send an email to: dev-unsubscribe@hudi.apache.org

Re: [D] RocksDB as The Replica of MDT/RLI [hudi]

Posted by "danny0405 (via GitHub)" <gi...@apache.org>.

GitHub user danny0405 edited a comment on the discussion: RocksDB as The Replica of MDT/RLI

I'm assuming you already read the [Flink RLI RFC](https://github.com/apache/hudi/pull/17610).

The 2nd one would take a lot of efforts and we deem it as long-term solution, as of now, we prefer 1.

2. NBCC is only working with simple bucket index, so not very related.

4. the bootstrap op parallelism is scalable, the bottleneck might be the full scan load time for a single file group of MDT. As long as the time is less than the checkpoint timeout, it works well and the bootstrap only happens once for a job restart. Why not persistent RocksDB instances across jobs recovery: a). the RocksDB local storage is within local container which will be release once the task is killed(we can not use remote storage which kills the perf). b). there is no good ways to main the consistency between the RocksDB and MDT based on the complexities of DT and MDT consistency.

6. we want the write scalable as independent and we don't want to want to every file group from a single write task to limit the small files.

GitHub link: https://github.com/apache/hudi/discussions/18296#discussioncomment-16154189

----
This is an automatically sent email for dev@hudi.apache.org.
To unsubscribe, please send an email to: dev-unsubscribe@hudi.apache.org

Re: [D] RocksDB as The Replica of MDT/RLI [hudi]

Posted by "danny0405 (via GitHub)" <gi...@apache.org>.

GitHub user danny0405 edited a comment on the discussion: RocksDB as The Replica of MDT/RLI

> I don't like how we are coupling index choices and concurrency models.

yeah, the simple bucket index is required to impl NBCC now and we may need more flexible and general design for concurrent modifications in streaming concurrent write scenarios.

> can you please explain in detail,how the failover and OCC handling are related.

I think we can categorize the concurrent write cases into two: write with conflicts and write without conflicts.

* If the write detects conflicts, the whole job/task will trigger failover and the RocksDB replica will rebootstrap from scrach, which can ensure the consistency of the index backend akka to MDT RLI index, but this needs to introduce specific early conflict detection just in the checkpoint lifecycle:
  * persist the uncommitted write metadata under the Hudi table path;
  * in the last step of the #snapshot of write function, send a request to the coordinator to detect the conflicts;
  * need a customized conflict resolution strategy to combine all the existing uncommited write metadata with the latest timeline to validate where there are conflicts;
        

The pre-commit conflict resolution does not work well for Flink streaming because it happens after a successful checkpoint, Hudi deems the write as failed if there is conflict while Flink deems the write as successful(from the latest successful checkpoint), to fix gap, the early confclit resolutuon is required here.

* If the write does not detect confclits, there are still cases that another concurrent write modify the table with new record locations, the solution is we might need a early detection of the index backend freshness before each write: maintain a mappings between job-id to instant time so we can load the index changes maded from concurrent writers incrementally.(put the job-id in commit metadata or maintain it on the coordinator). This introduces a lot of complexities though, I'm expecting a more general solution for NBCC that is index type agnostic and not struggle in this index concurrent modification trap.

Here is the table for support of cuncurrent modifications with Flink RLI:

| use case/concurrency mode | OCC | NBCC |
|---|---|---|
| write & write | Y(with early conflic detection and index refreshing) | N |
| write & compaction | Y | N |
| write & clustering | Y(with early index refreshing) | N |

GitHub link: https://github.com/apache/hudi/discussions/18296#discussioncomment-16171800

----
This is an automatically sent email for dev@hudi.apache.org.
To unsubscribe, please send an email to: dev-unsubscribe@hudi.apache.org

Re: [D] RocksDB as The Replica of MDT/RLI [hudi]

Posted by "danny0405 (via GitHub)" <gi...@apache.org>.

GitHub user danny0405 added a comment to the discussion: RocksDB as The Replica of MDT/RLI

> Is there no way to make hudi commit first, and then flink proceeds to checkpoint.

Not sure why we can do this, we can not expose invalid dataset to the downstream right? That breaks the SI isolation.

GitHub link: https://github.com/apache/hudi/discussions/18296#discussioncomment-18259988

----
This is an automatically sent email for dev@hudi.apache.org.
To unsubscribe, please send an email to: dev-unsubscribe@hudi.apache.org

Re: [D] RocksDB as The Replica of MDT/RLI [hudi]

Posted by "danny0405 (via GitHub)" <gi...@apache.org>.

GitHub user danny0405 added a comment to the discussion: RocksDB as The Replica of MDT/RLI

> For scenario 1, we need some safe, atomic recovery protocol? What is that? we cannot disallow clustering altogether.

Yes, we meed some early conflict resolution happens within the checkpoint lifecycle to intervene and fail the checkpoint when there is conflict instead of let it complete to get rid of the inconsistency of validity(checkpoint and Hudi instant).

GitHub link: https://github.com/apache/hudi/discussions/18296#discussioncomment-18243428

----
This is an automatically sent email for dev@hudi.apache.org.
To unsubscribe, please send an email to: dev-unsubscribe@hudi.apache.org

Re: [D] RocksDB as The Replica of MDT/RLI [hudi]

Posted by "vinothchandar (via GitHub)" <gi...@apache.org>.

GitHub user vinothchandar added a comment to the discussion: RocksDB as The Replica of MDT/RLI

> The incremental index upserts inferred from the data inputs are applied directly on these RocksDB instances

**Question 1**: If we treat rocksDB as the primary source of truth during writes, how does concurrent updates from another writer get visible correctly during `BucketAssignOp` stage? We need a solution that can work with NBCC and multiple writers. Can you expand on this in a comment below? 

> the paylods under the same data partition is stored as a separate column family,  

+1. This will also mean that for partitioned RLI i.e records always have a immutable partitioning field(s), the index size can be simply `O(size_of_data_partitions_written_to)` and not `O(total_size_of_table)` (as is the case for global indexing or mutable partition fields) 

> the RocksDB storage size turns out to be a nearly 2x size storage against the native HFile
**Question 2** : the 2x is due to just compression?  I think there will be an additional 2x additional storage for un-compacted updates, since rocksdb will also do its own async compaction periodically. 

GitHub link: https://github.com/apache/hudi/discussions/18296#discussioncomment-16056769

----
This is an automatically sent email for dev@hudi.apache.org.
To unsubscribe, please send an email to: dev-unsubscribe@hudi.apache.org

Re: [D] RocksDB as The Replica of MDT/RLI [hudi]

Posted by "danny0405 (via GitHub)" <gi...@apache.org>.

GitHub user danny0405 edited a comment on the discussion: RocksDB as The Replica of MDT/RLI

I'm assuming you already read the [Flink RLI RFC](https://github.com/apache/hudi/pull/17610).

The 2nd one would take a lot of efforts and we deem it as long-term solution, as of now, we prefer 1.

2. NBCC is only working with simple bucket index, so not very related.

4. the bootstrap op parallelism is scalable, the bottleneck might be the full scan load time for a single file group of MDT. As long as the time is less than the checkpoint timeout, it works well and the bootstrap only happens once for a job restart. Why not persistent RocksDB instances across jobs recovery: a). the RocksDB local storage is within local container which will be release once the task is killed(we can not use remote storage which kills the perf). b). there is no good ways to main the consistency between the RocksDB and MDT based on the complexities of DT and MDT consistency.

6. we want the write scalable as independent and we don't want to write to every file group from a single write task to limit the small files.

GitHub link: https://github.com/apache/hudi/discussions/18296#discussioncomment-16154189

----
This is an automatically sent email for dev@hudi.apache.org.
To unsubscribe, please send an email to: dev-unsubscribe@hudi.apache.org

Re: [D] RocksDB as The Replica of MDT/RLI [hudi]

Posted by "danny0405 (via GitHub)" <gi...@apache.org>.

GitHub user danny0405 edited a comment on the discussion: RocksDB as The Replica of MDT/RLI

> I don't like how we are coupling index choices and concurrency models.

yeah, the simple bucket index is required to impl NBCC now and we may need more flexible and general design for concurrent modifications in streaming concurrent write scenarios.

> can you please explain in detail,how the failover and OCC handling are related.

I think we can categorize the concurrent write cases into two: write with conflicts and write without conflicts.

1. If the write detects conflicts, the whole job/task will trigger failover and the RocksDB replica will rebootstrap from scrach, which can ensure the consistency of the index backend akka to MDT RLI index, but this needs to enable the early confclit detection: https://github.com/apache/hudi/pull/6133, the pre-commit conflict resolution does not work well for Flink streaming because it happens after a successful checkpoint, Hudi deems the write as failed if there is conflict while Flink deems the write as successful(from the latest successful checkpoint), to fix gap, the early confclit resolutuon is required here.

2. If the write does not detect confclits, there are still cases that another concurrent write modify the table with new record locations, the solution is we might need a early detection of the index backend freshness before each write: maintain a mappings between job-id to instant time so we can load the index changes maded from concurrent writers incrementally.(put the job-id in commit metadata or maintain it on the coordinator). This introduces a lot of complexities though, I'm expecting a more general solution for NBCC that is index type agnostic and not struggle in this index concurrent index trap.

Here is the table for support of cuncurrent modifications with Flink RLI:

| use case/concurrency mode | OCC | NBCC |
|---|---|---|
| write & write | Y(with early conflic detection and index refreshing) | N |
| write & compaction | Y | N |
| write & clustering | Y(with early index refreshing) | N |

GitHub link: https://github.com/apache/hudi/discussions/18296#discussioncomment-16171800

----
This is an automatically sent email for dev@hudi.apache.org.
To unsubscribe, please send an email to: dev-unsubscribe@hudi.apache.org

Re: [D] RocksDB as The Replica of MDT/RLI [hudi]

Posted by "vinothchandar (via GitHub)" <gi...@apache.org>.

GitHub user vinothchandar added a comment to the discussion: RocksDB as The Replica of MDT/RLI

@HuangZhenQiu Can you please chime in here.

GitHub link: https://github.com/apache/hudi/discussions/18296#discussioncomment-18379264

----
This is an automatically sent email for dev@hudi.apache.org.
To unsubscribe, please send an email to: dev-unsubscribe@hudi.apache.org

Re: [D] RocksDB as The Replica of MDT/RLI [hudi]

Posted by "cshuo (via GitHub)" <gi...@apache.org>.

GitHub user cshuo added a comment to the discussion: RocksDB as The Replica of MDT/RLI

> What does this mean for scenarios 2,3? Flink won't replay uncommitted input streams, right since checkpoint is successful. Hudi cannot commit either because writes are conflicting?

Yes, that's a common issue for flink writer, regardless of the index type.

>  (note this is 180* from my original thinking that NBCC is the harder of the two. I realize now that the record location does not change.). Poke holes at this?

“Record location does not change” only covers existing keys. For a new key, multiple writers may all miss it in their local RLI replicas and independently assign it to different file groups. NBCC cannot reconcile these writes because they do not conflict on the same file group, potentially creating duplicates. This is why NBCC currently requires the simple bucket index: its deterministic key -> bucket -> file group mapping makes every writer choose the same location.

Supporting Flink RLI with NBCC therefore requires a logically shared, atomic allocator for new keys—`record key -> file group`, so all writers agree on the first assignment.



GitHub link: https://github.com/apache/hudi/discussions/18296#discussioncomment-18243039

----
This is an automatically sent email for dev@hudi.apache.org.
To unsubscribe, please send an email to: dev-unsubscribe@hudi.apache.org