[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.

Release notes

Two behaviours of a Spark Structured Streaming write change with this PR.

• A micro-batch is now committed under commit identifier batchId + 1, the way Flink numbers its
  checkpoints, instead of the batch sentinel Long.MAX_VALUE. The $snapshots system table shows
  it, and a compacted-full scan recognises a scheduled full compaction by it. The full compaction
  schedule itself is unchanged.
• Because a micro-batch is no longer a batch commit, visibility-callback.enabled no longer makes
  the sink wait for visibility after each micro-batch. This matches the documented scope of the
  option, batch mode or bounded stream, and the Flink streaming behaviour; a streaming query that
  relied on the wait needs to be aware of it.

Replay recognition is bounded by the retained snapshot history: once every snapshot of a query'''s
commit user has expired, a replay of a batch older than that cannot be recognised.

[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.

[JingsongLi]

The original two findings are addressed in the inspected code: the persisted query ID is now preferred, and inlineMaintenance restores the one-shot commit lifecycle. Replayed micro-batch deduplication has clear value. One direct-postpone path still bypasses the promised replay behavior; details are inline.

A focused local-filesystem core probe used the actual builder/commit sequences: normal stable-user overwrite replay stayed at snapshot 1; direct postpone replay created snapshot 2, both with its default identity and when only the user was stabilized. I did not rerun the full Spark streaming suite.

[JingsongLi]

[P2] Define replay handling for the direct postpone committer too

With bucket=-2, postpone.default-bucket-num=1, and complete output mode, each overwrite takes this direct branch and never reaches the staged-path limitation warning. The actual PostponeFixedBucketWriteBuilder has its own random commit user, bypassing the stable user configured on writeBuilder. A replay therefore commits another overwrite snapshot instead of being a no-op. I reproduced snapshot 1 becoming snapshot 2; the ordinary stable-user overwrite control stayed at snapshot 1. This does not imply duplicate visible rows in complete mode, but it repeats a commit and its side effects.

Propagating the user alone is insufficient: this builder also resets commit.strict-mode.last-safe-snapshot to the current latest snapshot on each attempt, and filterCommitted searches only after that bound. The second probe with a stable user still committed snapshot 2. Handle replay lookup independently of that conflict bound and pass the stable identity to the actual committer, or explicitly exclude and warn for the direct path alongside the staged path. Please cover complete-mode replay with this configuration.

[zhuxiangyi]

Confirmed and reproduced first: with bucket = -2, postpone.default-bucket-num = 1 and complete mode, replaying batch 0 took the table from snapshot 1 to snapshot 2, exactly as you saw. Both halves of your analysis held up on inspection.

Fixed in 587ad9c along the first line you suggested:

• PostponeFixedBucketWriteBuilder gained withCommitUser, mirroring the one on BatchWriteBuilderImpl, and the direct builder is created with the stable user.
• The direct path looks the replay up itself, with latestSnapshotOfUser unbounded, and commits under the batch id only when nothing at or above it has been committed. The strict mode bound is left alone: it is what conflict detection needs, and I traced its use in filterCommitted to [core] Optimize FileStoreCommitImpl#filterCommitted when commit strict mode is set #7239/[core] Fix performance issue in FileStoreCommitImpl#filterCommitted #7275, a lookup optimisation whose premise is a commit user that never outlives its base snapshot. That premise is exactly what a stable user breaks, so the lookup for this path has to be independent of it rather than the bound relaxed for everyone.

The test for this configuration also commits a later batch after the replay and checks it lands, so that a higher batch id cannot be mistaken for a replay — the failure mode in that direction is silent data loss, which is worse than the duplicate.

Two things worth recording. First, while looking for a way to observe maintenance on this path I found that PrimaryKeyFileStoreTable.newExpireRunnable() returns null for postpone bucket tables, so snapshot expiration is never part of a commit there; a probe that expects the direct path to expire snapshots will not see it, regardless of inlineMaintenance. I restructured the sink so inlineMaintenance(true) is set once above both streaming branches, where the existing expiration test on an append table guards it, and added a core test in SimpleTableTestBase that commits through filterAndCommit with a committer closed after each batch and asserts the previous snapshot is expired before the close. Second, the staged committer is still outside the replay handling and keeps its warning; I did not extend the change there.

The suite is at 11 cases and was run on the 3.2, 3.4, 3.5, 4.0 and 4.1 modules again.

[JingsongLi]

The stable query identity, inline maintenance and unbounded direct-postpone lookup now address the previous replay findings. The new direct path correctly avoids a second snapshot for a replay and still permits a higher batch identifier; its strict conflict guard against another writer remains intact. However, one end-to-end recovery gap remains: successful deduplication must also finish callbacks that failed after snapshot publication. The attached P2 is reproduced using the changed core classes and the built-in partition-registration callback, following the new Spark branch; it is not a full Spark/metastore integration run.

[JingsongLi]

[P2] Retry commit callbacks before acknowledging an already-published batch

For the direct postpone path (bucket=-2, postpone.default-bucket-num=1) on a partitioned table with metastore.partitioned-table=true, a commit can publish its snapshot and then fail in the built-in AddPartitionCommitCallback, for example when the metastore partition-registration RPC fails. On replay, this branch only logs and returns successfully, bypassing FileStoreCommitImpl.filterCommitted, which deliberately calls callback.retry for already-committed batches. I reproduced this with a real local Paimon table and a fail-once PartitionModification: the first call throws after snapshot 1 is published; the replay succeeds with snapshot 1 still current, registrationAttempts=1 and registered=false. The normal filterAndCommit control retries registration (attempts=2, registered=true) without another snapshot. Please preserve the unbounded deduplication lookup while also completing the pending callbacks before reporting replay success; otherwise the checkpoint can advance with its committed partition still absent from the metastore.

[zhuxiangyi]

Confirmed, and reproduced first with a fail-once PartitionModification through the Spark direct path: after the first attempt failed past snapshot publication, the replay left attempts=1 and the partition unregistered, exactly as you saw. The lookup I had moved into the connector returned before filterCommitted, and with it the callback.retry that half of the semantics lives in. I had seen that line and dropped it.

The callbacks live in core, so the direct path commits through filterAndCommit again, in e7a9976. What had forced the connector-side lookup was the strict mode bound on filterCommitted; I traced it to #7239/#7275, an optimisation for a commit user created for one run, which cannot have committed before its base snapshot. A caller-provided user that survives a restart can have, so FileStoreCommit can now look the previous commit up without the bound, and both write builders enable that when withCommitUser was called. Conflict detection keeps the bound, and so does a committer created for an explicitly passed user — the staged committer with its per-run user goes through that overload, and a core test pins both sides for each builder.

While reading filterAndCommitMultiple for this I found a second cost of the same kind as the file-existence check: it passes checkAppendFiles=true, so every micro-batch on the ordinary path scanned the base files of the partitions it touches for conflicts, which an append of files the batch just wrote cannot have. A batch commit never did that scan before this series and the Flink committer does not in steady state either; InnerTableCommit can now turn it off and the sink does. For an append commit the scan has no observable effect, so this is covered by the existing suites rather than by a test of its own.

Suite is at 12 cases, run on 3.2, 3.4, 3.5, 4.0 and 4.1; core regression (FileStoreCommitTest, TableCommitTest, both SimpleTableTest subclasses) is green.

The two red jobs are UTCase and ITCase Others on JDK 11 and 17, failing in S3FileIOTest because Testcontainers cannot pull minio/minio:RELEASE.2022-02-07T08-17-33Z; the same workflow fails the same way on the last four runs of master (e.g. run 34697050592), so it is independent of this PR.

[JingsongLi]

[P2] Recover chain-table overwrite cleanup before acknowledging a replay

For a Complete-mode query writing a chain table's delta branch, the delta OVERWRITE snapshot can be published before a transient I/O failure prevents ChainTableOverwriteCommitCallback.call from clearing the corresponding snapshot-branch partitions. On replay, this filtered commit recognizes the published batch and invokes ChainTableOverwriteCommitCallback.retry, which is a no-op. The batch then returns successfully while chain-table reads still prefer the old snapshot data over the new delta data.

The empty retry method predates this PR, but switching the Spark sink to filtered replay introduces this recovery regression: the previous commit(List) path would commit the replayed overwrite and invoke call again, completing cleanup after the transient failure. This issue dates to the initial commit-path change in this PR.

I reproduced the difference on real local core tables at this revision, comparing the previous batch-commit sequence with the new filtered sequence and injecting a temporary snapshot-branch read failure after delta publication. After successful recovery, the previous path returned [2:value-2] from the main chain table; the filtered path returned accepted=0 and still read [1:value-1]. The probe exercised core commit/read paths; Spark Complete-mode reachability was verified by source tracing, rather than a full Spark fault-injection test.

Please make chain overwrite recovery complete the missing cleanup before acknowledging the replay. Resolve the original committed overwrite snapshot and its manifest changes, including partitions containing only removed files, and add a failure/replay regression test.

[zhuxiangyi]

Confirmed, and reproduced first at the core level: after an overwrite of the delta branch whose cleanup failed once the snapshot was published, the replay was recognised — no second snapshot — but the chain table still read value-1 from the snapshot branch. The empty retry predates this PR, as you say, and it was empty on purpose: its comment relied on the replay being committed again. Recognising the replay removed that.

Fixed in bb81c16, along the lines of IcebergCommitCallback.retry: the callback now gets its commit user, and on retry resolves the overwrite snapshot of the identifier with findSnapshotsForIdentifiers and redoes the cleanup from the delta manifest changes of that snapshot, so partitions the overwrite only removed files from are covered too. Truncating the partitions again is idempotent. One choice worth your look: a snapshot that can no longer be found is logged and skipped rather than failed on, unlike the Iceberg callback — it has either expired, in which case a later snapshot has superseded it, or the replay would fail forever.

The callback had no test of its own before, so 6a96de4 covers its normal path, the failure/replay case, the same for a partition the overwrite only removed files from — a mutant that takes partitions from ADD entries only fails it — and a retry for an identifier no snapshot carries. The failure is a real one: the delta table is told to clean a snapshot branch that does not exist, so the commit publishes its snapshot and then throws, which is the same window as a transient failure of the snapshot branch, without any file system fault injection. All chain table suites and the core regression pass.

While surveying every CommitCallback.retry for this class of problem I found one more thing you should decide on. VisibilityWaitCallback.call only acts when the identifier is BatchWriteBuilder.COMMIT_IDENTIFIER. Before this PR the sink committed micro-batches as batch commits, so a query with visibility-callback.enabled waited for visibility after each one; now the identifier is the batch id and the wait does not run. That matches the option's stated scope, batch mode or bounded stream, and the Flink streaming behaviour, so I have left it, but it is a behaviour change for such a query and I would rather you call it than have it surface later. The other callbacks (AddPartitionCommitCallback, TagPreviewCommitCallback, IcebergCommitCallback) already implement retry.

[JingsongLi]

Reviewed 6a96de4. Requirement fit: SUPPORTED. Implementation: FINDINGS.

Preventing duplicate Spark micro-batches after commit/checkpoint failure has clear end-to-end value. The previous direct-postpone callback and chain-overwrite retry findings are fixed: the replay takes the filtered commit path, and chain cleanup reads the original snapshot's DELTA changes, including DELETE-only partitions. Eight focused real-table recovery/callback tests passed with the changed core classes on JDK 8.

The remaining new issue is the interaction between raw zero-based batch IDs and full-compaction snapshot recognition. A real table/scanner probe selects snapshot 5 and value 103 instead of the scheduled full-compaction snapshot 4 and value 102; the previous sentinel path and a one-based identifier control both select snapshot 4. Details are inline. Current CI is successful, but these focused probes are not a new full Spark integration run.

The disclosed visibility-wait behavior change should remain in release guidance. Replay recognition also remains bounded by retained snapshot history; these checks do not establish recovery after all snapshots for the old commit user have expired.

[JingsongLi]

[P2] Align streaming commit IDs with full-compaction numbering

Publishing the raw zero-based batchId breaks FullCompactedStartingScanner's recognition of Spark's scheduled full compactions. DataWrite.needFullCompaction triggers when (batchId + 1) % deltaCommits == 0, while the scanner accepts commitIdentifier % deltaCommits == 0 (or the old Long.MAX_VALUE sentinel). With deltaCommits=3, every scheduled full compaction now has an identifier that the scanner rejects.

I reproduced this with four real core writes using the exact Spark compaction schedule and filtered commit sequence: batch 2 publishes COMPACT snapshot 4 with identifier 2; batch 3 publishes APPEND snapshot 5. A compacted-full scan falls back to snapshot 5 and reads value 103, instead of snapshot 4/value 102. The previous sentinel sequence and changing only the published identifier to batchId + 1 both select the correct full snapshot. Please use numbering consistent with the existing scanner contract, or explicitly preserve full-compaction recognition, and add a Spark write-to-compacted-full-read regression with a delta count greater than one.

[zhuxiangyi]

Reproduced through the sink (delta-commits = 3, four batches: compacted-full read 103 instead of 102) and fixed in f9995d4 with your first option: the identifier is batchId + 1, defined once and used for both the published commit and the full compaction schedule, so the schedule shares the scanner's rule. The schedule itself is unchanged.

The regression test asserts the identifiers of the full compaction (3) and the batch after it (4) and reads the full compaction through compacted-full; a zero-based identifier fails it and the existing schedule test, nothing else.

Both behaviour changes — the identifier numbering and the visibility wait — are in a release notes section of the description now, with the bound on replay recognition by retained snapshot history.