feat(iceberg): support coordinator-driven size-based commit trigger for Iceberg sink - #26551
Open
tuantran0910 wants to merge 5 commits into
Open
Conversation
Add (default 512 MiB) that triggers an early commit when total buffered write bytes across all writers exceeds the threshold. The coordinator sums per-writer byte reports and broadcasts the same commit decision to prevent multi-writer alignment deadlocks. When uncommitted bytes are zero, the coordinator RPC is skipped entirely.
…alignment Iceberg sinks with commit_checkpoint_size_threshold_mb set stalled silently in production: commits stopped entirely within minutes of the post-backfill parallelism reschedule, kv_log_store_buffer_unconsumed_epoch_count climbed at exactly one epoch per second on every actor, and no error was ever surfaced. Recreating the sinks only re-armed the stall. Separately, the config declared a serde default of 512 MiB for the threshold that the runtime never applied. The root cause is a writer/coordinator protocol mismatch: the coordinated writer skipped the report_bytes RPC when an actor had zero buffered bytes at a checkpoint, but the coordinator answers an epoch only once byte reports cover the full vnode bitmap, and only for the oldest pending epoch. One idle actor therefore left the epoch unaligned forever, head-of-line-blocking every later epoch while the reporting actors awaited a response that never came, with no timeout on either side. Low per-actor traffic after backfill (a few events/sec hash-split across 8 actors) made an idle-actor checkpoint near-certain. The dead default arose because both runtime sites read the raw property map, where absence means disabled, while the serde default only populated the parsed IcebergConfig field that nothing consumed. The writer now always sends report_bytes, including zero: the coordinator aligns the epoch and answers should_commit=false when the reported total is below the threshold, so all-idle epochs align and never commit. A regression test (test_size_based_commit_with_zero_byte_report) covers an idle writer aligning an epoch and committing empty metadata once its peer crosses the threshold. The unused 512 MiB default is removed: the field is now Option<u64> and size-based commits are strictly opt-in, so the declared contract matches runtime behavior instead of silently flipping on for every iceberg sink if a future refactor wires the parsed config through. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Fixes formatting flagged by CI's rustfmt check on the coordinator-driven size-based commit trigger changes (chained method calls, struct pattern wrapping, argument list wrapping).
…er sizing guidance Clippy's doc_markdown lint flags bare "MiB" as missing backticks. Also corrects the sizing guidance for commit_checkpoint_size_threshold_mb: the threshold is compared against the sum of buffered bytes across all parallel writers, not any single writer, so a given writer's output file size is closer to threshold / sink_parallelism rather than the full threshold value.
…omment The generated yaml's comments field must match the struct field's doc comment line-for-line (with_options_test.rs::extract_comments joins each `///` line verbatim, blank lines included) — my hand-edit had paraphrased the text instead, failing test_with_options_yaml_up_to_date.
Contributor
|
Thanks for the contribution! I appreciate you taking the time to add this Iceberg sink feature. However, I noticed that this PR also introduces changes to our sink execution framework. The execution framework is a shared abstraction layer used by all connectors, so changes in this area have a much broader impact than a single connector implementation. For framework-level changes, we usually need to discuss and review the design first rather than directly modifying it as part of a connector PR. Could you please open a GitHub issue to propose this framework change separately? It would be helpful if the issue could include:
After the framework change proposal is reviewed and we decide that this is the right direction, we can first implement the framework changes, and then update the Iceberg sink to leverage the new framework capability. Thanks again for the effort on this feature — separating the framework evolution from the connector implementation will make the change easier to review and maintain long term. |
Contributor
Author
|
Thank you @wenym1, I will open a Github issue for the framework change proposal soon :D |
Contributor
Author
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
I hereby agree to the terms of the RisingWave Labs, Inc. Contributor License Agreement.
Close #25058
What's changed and what's your intention?
Iceberg sinks currently only commit after a fixed number of checkpoints (
commit_checkpoint_interval), regardless of how much data has accumulated. Under bursty or variable-rate workloads, this leads to either too many small files or too much memory held between commits.This was already attempted once in #25058 (
commit_checkpoint_size_threshold_mb, merged), but reverted in #25614 because the implementation could hang commits indefinitely. That version made the commit decision locally per writer: each writer independently checked its own buffered bytes and decided whether to force a commit on the checkpoint barrier. Under hash skew across parallel writers, one writer can cross the threshold and request a commit for an epoch while its peers, still under threshold, send nothing for that same epoch. The coordinator only finalizes a commit once every writer's vnodes are represented for that epoch — so the lone request never completes, and since epochs commit in order, every later epoch queues up behind the stuck one.This PR re-implements the feature with the decision moved out of the writer and into the coordinator, structurally avoiding that failure mode:
ReportBytesgRPC message) — it no longer decides anything on its own.should_commitdecision, broadcast back to every writer for that epoch. No writer can unilaterally trigger a commit the others aren't ready for — the whole cohort commits together or not at all.Behavior:
Option<u64>, unset). Setcommit_checkpoint_size_threshold_mbto enable;0disables it explicitly.iceberg_engineoption passthrough, soCREATE TABLE ... WITH (commit_checkpoint_size_threshold_mb = ...) ENGINE = icebergandALTER ... SET commit_checkpoint_size_threshold_mb = ...just work — no extra frontend wiring needed.Known limitation: the threshold measures in-memory buffered bytes, not on-disk Parquet size — actual file size is typically 3–10x smaller after compression, so size it accordingly.
Related to #25058 (reverted in #25614).
Checklist
Documentation
Release note
Added a new Iceberg sink/table option,
commit_checkpoint_size_threshold_mb, that triggers an early commit once total buffered write bytes across all writers exceeds the given threshold (in MiB). Opt-in and off by default. This supersedes the earlier attempt in #25058 (reverted in #25614 due to a commit-hang bug); the commit decision is now made by the coordinator across all writers instead of by each writer independently, avoiding that failure mode.