SS-148 Iceberg sink, don't attempt commit unless it's definitely not a duplicate

[ublubu]

replaces #38333

We rely on MaterializeInc/iceberg-rust#7 making the Transaction commit internals public.

Then we reimplement Transaction commit, but without the built-in rebase+retry.

Previously, iceberg-rust loaded the table from the Catalog on every commit attempt, applying the transaction's actions on top of the latest state of the table.

With this PR, Mz loads the table from the Catalog on every commit attempt, but it does not apply the transaction's action on top of the latest state of the table until after it inspects the table state (for funny business like a previously successful attempt or another writer taking over).

[def-]

QA LLM Review

1. MEDIUM -- Pre-commit guard only rejects an exactly-equal frontier, so a writer that committed a strict prefix of our batch still produces duplicate rows

src/storage/src/sink/iceberg.rs:883

The new guard proves "not a duplicate" only when last_frontier is exactly equal to frontier (idempotent retry) or >= it (fenced). It never checks last_frontier against batch_lower, so when another incarnation committed a snapshot that lands strictly inside our batch range (batch_lower < last_frontier < frontier) all three conditions fall through and we commit [batch_lower, frontier) on top of it, silently duplicating every row in [batch_lower, last_frontier) with no error surfaced.

Details

The reachable window is a sink version handoff, which is a normal ALTER SINK (see workflow_alter_commit_interval) or a replica restart where the old incarnation is still running. Version V reads the table, is not fenced (nothing newer is on it yet), and commits [F, G). Version V+1 had already read resume_upper = F at mint_batch_descriptions (src/storage/src/sink/iceberg.rs:1228) and minted its catch-up batch as [F, H) with H > G. At commit time V+1 sees last_version = V, so last_version > sink_version is false; last_frontier = G != H, so the idempotency arm is skipped; and PartialOrder::less_equal(H, G) is false, so the fencing arm at line 913 is skipped too. V+1 commits [F, H), duplicating [F, G). The same shape applies to two same-version incarnations racing after a replica restart, where mz-sink-version provides no fence at all.

last_frontier == batch_lower is an invariant of the minter in every single-writer case, so it can be asserted directly: no snapshots means a fresh table and batch_lower == as_of; the catch-up batch's lower is resume_upper, which is by construction the last committed frontier; and every subsequent batch's lower is the previous batch's upper. Third-party snapshots that copy our summary (as workflow_commit_conflict injects) and replace compactions both leave last_frontier at the sink's last committed frontier, so neither is disturbed.

Suggested fix, after the two existing fatal arms:

if last_frontier != *batch_lower {
    // Another writer committed inside our batch range. Committing now would
    // duplicate [batch_lower, last_frontier). Restart and re-resume instead.
    return (
        table,
        RetryResult::FatalErr(anyhow!(
            "Iceberg table '{}' advanced to frontier {:?} inside batch [{}, {}); \
             another writer is active.",
            conn_table,
            last_frontier,
            batch_lower.pretty(),
            batch_upper.pretty(),
        )),
    );
}

2. MEDIUM -- workflow_idempotent_retry can drop the response to an empty commit, in which case it passes with the pre-fix duplicate-write behavior

test/iceberg/mzcompose.py:405

The drop is armed as soon as commits_ok >= 1 and before the first INSERT, so the commit whose response gets dropped is frequently one that carries no data files. When that happens the test proves nothing about duplicate writes: replaying an empty RowDeltaAction adds no rows, so idempotent-retry-verify.td's exact count of 13 still holds, messages_committed >= 13 still holds, and the sink is still running. The regression this workflow exists to catch (reverting to Transaction::commit, whose rebase-and-retry re-adds the dropped batch's data files) would slip through.

Details

Empty commits are the steady state here: the sink commits on the 2s interval whether or not data arrived, the three setup rows all land in the snapshot batch [as_of, as_of+1), and the catch-up batch [as_of+1, observed_frontier) that immediately follows it is therefore empty. So the armed drop lands on a data-carrying commit only if the first c.sql INSERT round-trip completes before the next batch boundary, which is a timing race rather than something the test establishes.

The proxy already has the request body in hand at test/iceberg/polaris_proxy.py:96, so the fire condition can require a data-carrying commit, for example by only setting should_drop when the body's add-snapshot update has a non-zero added-data-files/added-records in its summary. Asserting after the fact that the dropped commit carried data would work equally well.

[patrickwwbutler]

I think just fix this main issue and it looks ok? Would also take a look at the test setup issue in the LLM review but that's less important to me.

Merge issues probably just from my iceberg stuff last week, should be pretty easy although you'll need to rebase your iceberg-rust changes and update the revision

[patrickwwbutler]

Suggested change

	if PartialOrder::less_equal(frontier, &last_frontier) {
	if PartialOrder::less_equal(frontier, &last_frontier) || PartialOrder::less_than(batch_lower, &last_frontier) {

I think this should address the issue in the LLM review and catch when another writer has written a partially overlapping batch, as the last frontier should be <= all items in this batch.

Given the generally longer commit intervals for iceberg sinks, this actually seems fairly likely to happen (i.e. the other writer committed 5 seconds ago, so this batch does have a later frontier, but there's still overlap. Thankfully it looks like an easy fix?

[patrickwwbutler]

Also just don't forget this lol