[RFC] Generic byte-reporting alignment for coordinator-driven early commits

[tuantran0910]

Background

Coordinated sinks (Iceberg, Delta Lake, StarRocks, etc. — anything using SinkCommitCoordinator/CoordinatedLogSinker) currently only have one way to control commit cadence: commit_checkpoint_interval, a fixed count of checkpoints. This is purely time/barrier-driven and has no notion of how much data has actually accumulated.

Concretely, this hurts the Iceberg upsert sink today. Its writer (DeltaWriter in iceberg-rust) tracks an in-memory HashMap<OwnedRow, PositionDeleteInput> (inserted_row) to convert same-epoch update/delete sequences into position deletes correctly. This map is populated on every inserted row and is only cleared when the writer commits — it has no other bound. Under high-throughput upsert ingestion, if the checkpoint-count-based interval doesn't trigger a commit soon enough, this map grows unbounded between commits and can drive the sink actor to OOM. A size-based early commit trigger — commit as soon as buffered/tracked data crosses a threshold, independent of checkpoint count — directly bounds this map's growth and avoids the OOM, which is the primary motivation for this feature.

More generally, for any sink writing to file-based formats, a fixed checkpoint interval makes it hard to keep output file sizes in a healthy range under bursty or variable-rate ingestion: an interval short enough to avoid this memory buildup risks too many small output files; one long enough to avoid small files risks the exact memory buildup above.

We tried this once for Iceberg specifically (#25058), implemented as a per-writer local decision: each parallel writer independently checked its own buffered bytes and, once over threshold, unilaterally forced a commit on its next checkpoint barrier. This was reverted (#25614) because it could hang commits indefinitely: under data skew across parallel writers, one writer crosses the threshold and requests a commit for an epoch while its peers — still under threshold — send nothing for that epoch. The coordinator only finalizes a commit once every writer's vnodes are represented for that epoch, so the lone request never completes, and every later epoch queues up behind the stuck one.

The core problem is that "should we commit early" was decided locally, per-writer, with no coordination — which is unsafe for any sink where commits must be vnode-aligned across parallel writers. A correct implementation needs that decision to be made centrally, by the coordinator, based on input from all writers, the same way epoch/commit alignment already works for the two-phase commit path.

This RFC proposes making that a generic framework capability — available to any coordinated sink, not just Iceberg — rather than reintroducing it as Iceberg-only logic (as prototyped in #26551, which was correctly flagged as introducing framework changes inside a connector PR).

Design

New RPC message pair, in connector_service.proto, alongside the existing CommitRequest/CommitResponse in CoordinateRequest/CoordinateResponse:

message CoordinateRequest {
  message ReportBytesRequest {
    uint64 epoch = 1;
    uint64 buffered_bytes = 2;
  }
  oneof msg {
    ...
    ReportBytesRequest report_bytes_request = N;
  }
}

message CoordinateResponse {
  message ReportBytesResponse {
    uint64 epoch = 1;
    bool should_commit = 2;
  }
  oneof msg {
    ...
    ReportBytesResponse report_bytes_response = N;
  }
}

New optional trait method on SinkWriter (src/connector/src/sink/writer.rs), defaulted to 0 so it's a no-op for sinks that don't opt in:

trait SinkWriter {
    /// Number of uncommitted write bytes buffered by this writer.
    fn buffered_bytes(&self) -> u64 { 0 }
    ...
}

For the Iceberg upsert writer specifically, this would report a measure that tracks (or approximates) the size of the in-memory inserted_row map described above, not just raw written bytes — so the trigger actually bounds the structure causing the OOM risk.

Generic writer-side loop change, in CoordinatedLogSinker (src/connector/src/sink/coordinate.rs): on a non-forced checkpoint barrier, if a size threshold is configured, the writer reports its buffered_bytes() to the coordinator via report_bytes() instead of deciding locally, and only commits if the coordinator's response says so. Writers must always report, including zero — skipping the report for an idle writer leaves that epoch permanently unaligned and stalls every later epoch behind it (this was the exact failure mode of #25058/#25614, reproduced and confirmed while prototyping this).

Generic coordinator-side alignment, in src/meta/src/manager/sink_coordination/coordinator_worker.rs: a new pending_byte_reports: BTreeMap<u64, AligningRequests<u64>> track, parallel to the existing pending_epochs used for commit alignment, reusing the same vnode-bitmap alignment logic (AligningRequests) already proven for commits. Once every writer's vnodes are represented for an epoch, the coordinator sums the reported bytes, compares against a configured threshold, and broadcasts one should_commit decision to every writer for that epoch via a new ReportBytesResponse. No writer can unilaterally trigger a commit — the whole cohort commits together or not at all, structurally ruling out the #25058 hang.

The threshold itself (e.g. Iceberg's commit_checkpoint_size_threshold_mb) stays a connector-specific config concern — the framework only needs to accept "is there a threshold, and here's the aggregate byte total" and hand back a boolean. Any sink wanting this capability implements buffered_bytes() and passes its own threshold through to the shared alignment path.

Future Optimizations

• Only byte-size thresholds are covered here; row-count-based or time-since-last-write triggers could reuse the same alignment mechanism (AligningRequests<R> is already generic over the reported value type) if a future need arises.
• Reporting happens on every non-forced checkpoint barrier regardless of whether any threshold is configured for that sink; if this proves to add measurable overhead for large sink fleets, reporting could be skipped entirely when no writer has a threshold configured (today it's already skipped when threshold is None).

Discussions

• Compatibility with existing sinks: this is purely additive — a new optional trait method with a zero-cost default, a new proto oneof variant, and a new coordinator-side alignment track that's inert unless a sink actually calls report_bytes(). No existing sink's behavior changes unless it opts in.

Q&A

• Why not just let each writer decide locally, like feat(iceberg): support iceberg sink commit checkpoint size threshold mb #25058 did? Because commits must be vnode-aligned across all parallel writers, and under data skew one writer can cross a local threshold while its peers never do for that epoch — causing the epoch (and every epoch after it) to hang forever with no timeout. Confirmed as the actual root cause of the fix(iceberg): revert 25058 #25614 revert.
• What happens if a writer never reports (e.g. it's idle)? The design requires writers to always report, even zero. If a writer implementing buffered_bytes() skipped reporting when idle, that epoch would never align — this is called out explicitly as a hard requirement, not an optimization detail.
• Does this replace commit_checkpoint_interval? No — it's an additional, independent trigger. A checkpoint-count-based commit and a size-based commit can both be configured; whichever fires first wins for a given barrier.
• Why does the Iceberg upsert sink need this specifically, beyond file sizing? Because its DeltaWriter holds an unbounded in-memory dedup map (inserted_row) between commits; under high-throughput ingestion with a checkpoint-interval that doesn't trigger often enough, that map is the actual OOM risk, not just Parquet file size.

[wenym1]

I think the current proposal is coupling two separate concerns: controlling writer-side memory usage and coordinating sink commits.

The motivation seems to be that each writer parallelism needs a way to bound its in-memory buffered state (e.g. Iceberg’s inserted_row map). However, a commit is not necessarily the mechanism that flushes this state. The commit phase is mainly about committing metadata / making the previously flushed data visible, while the lifecycle of the writer’s in-memory state can be decoupled from commit coordination.

A simpler approach could be:

• Let each writer decide independently when its local in-memory state needs to be flushed based on its own memory threshold.
• If multiple flushes happen within the same epoch, accumulate the generated files/state changes locally.
• When the coordinator commit happens, the writer submits all accumulated outputs from those flushes together for the commit.

This would keep the existing commit coordination semantics unchanged and avoid introducing a new framework-level coordination path for a problem that is fundamentally local to each writer.

The coordinator should only need to coordinate things that require global consistency (e.g. when a commit becomes visible across all vnodes). Whether a single writer internally needs to spill/flush its buffered state before that point feels like an implementation detail of the writer.

I think it would be worth exploring whether we can solve the Iceberg OOM issue at the writer layer first before adding a generic coordinator-level byte reporting mechanism.