[core][spark] Deduplicate a replayed Structured Streaming micro-batch

[zhuxiangyi]

Purpose

Closes #9666.

Structured Streaming delivers exactly-once only if the sink is idempotent for a repeated batch id.
When a query fails after the sink returns from addBatch but before Spark records the batch as
completed, the restarted query replays that micro-batch with its original batch id.

PaimonSink received the batch id but used it only to pace full compaction, and committed through
table.newBatchWriteBuilder(), whose commit user is a fresh random UUID per builder and whose
commit identifier is always BatchWriteBuilder.COMMIT_IDENTIFIER = Long.MAX_VALUE. Neither of the
dimensions Paimon deduplicates on could identify a replay, so the batch was committed a second
time: every row duplicated in an append-only table, and silently wrong values in an aggregation
merge-engine table (writing (1, 10) and replaying that batch yields v = 20).

The machinery already exists in core and is what the Flink sink uses; the Spark sink simply took
the batch write path.

Tests

PaimonSinkIdempotencyTest (new, 7 cases), each asserting the correct behaviour so that it fails
without the fix:

• a replayed micro-batch of an append-only table, driven end to end by deleting the commit log
  entry of the batch, which is exactly the checkpoint state a driver failure leaves behind;
• the same when only the query id is available, i.e. the checkpoint location never reaches the sink
  options because it comes from spark.sql.streaming.checkpointLocation;
• a replay of a batch that is not the first one of the query;
• a replay in complete output mode;
• write.stream.commit-user as an option of the writer and as a spark.paimon. session conf;
• addBatch called twice with the same batch id through the API directly.

Two of them assert the prefix of the commit user recorded in the snapshot, so that the case which
is meant to exercise the query id derivation cannot pass through the checkpoint derivation.

The full set of Spark streaming suites was run on the spark3 and spark4 profiles (Spark 3.2, 3.4,
3.5, 4.1): 34 suites, 349 tests.

API and Format

BatchWriteBuilderImpl gains withCommitUser, and its newCommit() return type is narrowed from
BatchTableCommit to InnerTableCommit. The narrowing keeps the caller in the connector free of a
downcast that could only fail at runtime; it is source compatible, and BatchWriteBuilderImpl has
no subclasses.

InnerTableCommit gains checkFilesExistence(boolean). filterAndCommit verified that every file
it is about to commit still exists, which guards a committable restored from an engine's state that
may reference files deleted long ago. A caller filtering a committable it has just produced knows
those files exist, so the Spark sink turns the check off; otherwise every micro-batch would pay a
file listing proportional to the number of files it wrote. The default is unchanged, so Flink keeps
the check.

Documentation

docs/docs/spark/structured-streaming.md gains an "Exactly-once" section covering how the commit
user is derived, the new write.stream.commit-user option, and the limits: starting from a new
checkpoint location gives a query a new commit user, the data files of a skipped replay are left to
orphan file cleaning, and a postpone bucket table committing through the staged committer cannot
deduplicate and logs a warning per micro-batch. The generated option reference is regenerated.

[JingsongLi]

Two issues found in the checkpoint identity and commit maintenance lifecycle. The existing seven Spark 3 tests pass; additional checkpoint edge-case tests and a focused maintenance probe reproduce the issues below.

[JingsongLi]

[P1] Prefer the persisted query ID over the checkpoint path

If a checkpoint is deleted and a new query starts at the same path, Spark assigns a new query ID and restarts batch IDs at 0. This code nevertheless reuses the previous commit user, so filterCommitted silently skips fresh batches whose IDs are at or below the previous query's last committed ID. I reproduced this with different input in the new query: the query completed successfully, but the table contained only the old row instead of both rows.

The reverse also fails: restarting the same checkpoint with only a trailing / added to its path changes the commit user despite an unchanged query ID. Replaying batch 0 then produced two rows instead of one.

Please keep the explicit override, but prefer the persisted Spark query ID for the default identity and use the checkpoint path only as a fallback. Add coverage for both a fresh query reusing a checkpoint path and an existing query using an equivalent path spelling.

[zhuxiangyi]

Confirmed both, and reproduced both before changing anything.

A fresh query at a reused location: the first query wrote [1,old] under spark-checkpoint-b492463a…; after deleting the checkpoint, a new query writing (2,"new") derived the identical user, and the table still held only [1,old] — the new row was skipped as an already committed replay, with the query reporting success. It is not limited to the first batch: the comparison is against the previous run’\s last committed identifier, so a new query silently loses every batch up to that id.

The trailing separator: spark-checkpoint-ae4544b6… became spark-checkpoint-9ff6c207… for the same checkpoint, and the replayed batch produced two rows.

The underlying mistake was binding the identity to where the checkpoint is stored rather than to which incarnation of it is running, so it could come apart in both directions. The failure modes are asymmetric — a fresh identity for the same query duplicates, a stale identity for a new query loses data — which is the stronger argument for your ordering, so it is now: explicit override, then the persisted query id, then the location, which is only reachable outside a stream execution (a direct addBatch call has no query id).

Both cases you named are covered: "a new query reusing a checkpoint location must not skip its batches" and "an equivalent spelling of the checkpoint location keeps the identity", asserting the commit user differs in the first and is unchanged in the second.

Since the query id is now the default identity rather than a fallback, I also ran the suite on the 3.2, 3.4, 3.5, 4.0 and 4.1 modules to confirm the property is populated on every supported version. That surfaced an unrelated problem of my own: the suite left a second table behind, which Spark 3.2 cannot drop because its dropNamespace has no cascade overload. Fixed with withTable.

[JingsongLi]

[P2] Preserve the batch maintenance lifecycle when filtering commits

Unlike commit(List), filterAndCommit does not set TableCommitImpl.batchCommitted. Maintenance therefore runs through the streaming executor wrapper, but this writer still closes and discards the committer immediately after each batch. With snapshot.expire.execution-mode=async, close() calls shutdownNow() and can interrupt snapshot expiration before it finishes. Even with the default synchronous mode, the wrapper catches maintenance exceptions and stores them for the next commit; because this instance is discarded, those failures are never propagated to the caller.

A focused probe against the built classes reproduced both the asynchronous interruption and the loss of synchronous error propagation. The existing testBatchWriteAsyncExpireFallbackToSync also establishes that a batch committer must finish maintenance before closing.

Please preserve the one-shot batch maintenance semantics in the filtered commit path, or explicitly wait for maintenance and propagate its failure before closing the committer.

[zhuxiangyi]

Confirmed. maintain branches on batchCommitted, which only checkCommitted() sets, so the filtered path left maintenance to the executor while the writer kept the one-shot lifecycle of a batch committer — the two halves no longer matched.

The asynchronous half reproduced clearly. With snapshot.expire.execution-mode=async and min/max retained set to 1, four micro-batches left snapshots=4 earliest=1 latest=4: shutdownNow() drained the queued task, so expiration did not run at all rather than being cut short partway. For a long-running streaming write that means snapshots grow without bound.

The synchronous half I verified by reading rather than by test: the wrapper stores the failure in maintainError, which is only rethrown at the start of the next maintain, and this committer is discarded before there is one. Rather than test the swallowing, I made it structurally impossible.

InnerTableCommit now has inlineMaintenance(boolean): maintenance runs on the committing thread and its failure is thrown to the caller, which is what commit(List) already gets from batchCommitted. The Spark sink asks for it on the filtered path. The default is unchanged, so Flink keeps the executor. testBatchWriteAsyncExpireFallbackToSync was the right reference — the invariant it fixes is exactly the one I had broken, and the new test asserts the same property through the streaming sink.

Worth noting for the record: expiration now runs inline on each micro-batch, which restores the behaviour before this PR rather than adding cost. Making it asynchronous again would mean keeping one committer alive across batches, the way CommitterOperator does; that is a larger change than this fix and I have left it out.