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

docs/docs/spark/structured-streaming.md

@@ -56,6 +56,51 @@ val stream = df
 
 Streaming write also supports [Write merge schema](./sql-write#write-merge-schema).
 
+### Exactly-once
+
+Structured Streaming replays a micro-batch with its original batch id when a query is restarted
+after failing between the sink writing the batch and Spark recording that batch as completed.
+Paimon commits every micro-batch under a commit user that is stable across restarts, and skips a
+batch that the same user already committed, so a replay does not write the data twice. Micro-batch
+`n` is committed under commit identifier `n + 1`, the way Flink numbers its checkpoints, which is
+what the `$snapshots` system table shows and what a `compacted-full` scan recognises a scheduled
+full compaction by.
+
+What the commit user identifies is one incarnation of a checkpoint, not the place it is stored:
+reusing it across two different queries would make Paimon skip the data of the second one, while
+changing it within one query would bring the duplicate back. It is therefore derived from the query
+id that Spark persists in the checkpoint, which is new when a checkpoint is recreated, unchanged
+when a query resumes from one, and independent of how the location is spelled. Set
+`write.stream.commit-user` to pin it explicitly, either as an option of the writer or as a
+`spark.paimon.write.stream.commit-user` session conf, which is only needed if a query has to keep
+its identity across a new checkpoint:
+
+```scala
+val stream = df
+  .writeStream
+  .outputMode("append")
+  .option("checkpointLocation", "/path/to/checkpoint")
+  .option("write.stream.commit-user", "my-streaming-job")
+  .format("paimon")
+  .start("/path/to/paimon/sink/table")
+```
+
+:::note
+
+A skipped replay leaves the data files it wrote behind, uncommitted. They are removed by
+[orphan file cleaning](../maintenance/manage-snapshots#remove-orphan-files), like any other
+uncommitted file.
+
+A query that starts from a new checkpoint gets a new commit user, so a micro-batch the previous
+run committed is not recognised and its data is written again.
+
+A postpone bucket table with `postpone.default-bucket-num` commits an overwrite, such as a
+micro-batch in `complete` mode, through its direct fixed-bucket committer, where a replay is
+recognised as well. Its other writes go through a staged committer that cannot skip a replay; a
+warning is logged for every such micro-batch.
+
+:::
+
 ## Streaming Query
 
 :::info

docs/generated/spark_connector_configuration.html

@@ -140,6 +140,12 @@
             <td>Boolean</td>
             <td>Only effective when 'write.merge-schema' is true. If true, widen an existing column type when the incoming data has a wider compatible type (e.g. INT -&gt; BIGINT, DECIMAL precision increase). Lossy changes are still rejected unless 'write.merge-schema.explicit-cast' is also true.</td>
         </tr>
+        <tr>
+            <td><h5>write.stream.commit-user</h5></td>
+            <td style="word-wrap: break-word;">(none)</td>
+            <td>String</td>
+            <td>The commit user of a Structured Streaming write. Paimon skips a micro-batch that a previous run of the same query already committed under this user, which is what makes a replayed micro-batch idempotent. By default it is derived from the query id that Spark persists in the checkpoint, so it is kept while a query resumes from its checkpoint and is new when the checkpoint is; set it explicitly only if a query has to keep its identity across a new checkpoint.</td>
+        </tr>
         <tr>
             <td><h5>write.use-v2-write</h5></td>
             <td style="word-wrap: break-word;">false</td>

paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java

@@ -440,7 +440,7 @@ private List<CommitCallback> createCommitCallbacks(String commitUser, FileStoreT
         }
 
         if (options.isChainTable()) {
-            callbacks.add(new ChainTableOverwriteCommitCallback(table));
+            callbacks.add(new ChainTableOverwriteCommitCallback(table, commitUser));
         }
 
         if (options.visibilityCallbackEnabled() && shouldWaitForVisibility(table)) {

paimon-core/src/main/java/org/apache/paimon/metastore/ChainTableOverwriteCommitCallback.java

@@ -19,16 +19,22 @@
 package org.apache.paimon.metastore;
 
 import org.apache.paimon.CoreOptions;
+import org.apache.paimon.Snapshot;
 import org.apache.paimon.Snapshot.CommitKind;
 import org.apache.paimon.data.BinaryRow;
 import org.apache.paimon.manifest.ManifestCommittable;
 import org.apache.paimon.manifest.ManifestEntry;
 import org.apache.paimon.table.FileStoreTable;
 import org.apache.paimon.table.sink.BatchTableCommit;
 import org.apache.paimon.table.sink.CommitCallback;
+import org.apache.paimon.table.source.ScanMode;
 import org.apache.paimon.utils.ChainTableUtils;
 import org.apache.paimon.utils.InternalRowPartitionComputer;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.stream.Collectors;
@@ -50,52 +56,91 @@
  */
 public class ChainTableOverwriteCommitCallback implements CommitCallback {
 
+    private static final Logger LOG =
+            LoggerFactory.getLogger(ChainTableOverwriteCommitCallback.class);
+
     private transient FileStoreTable table;
     private transient CoreOptions coreOptions;
+    private final String commitUser;
 
-    public ChainTableOverwriteCommitCallback(FileStoreTable table) {
+    public ChainTableOverwriteCommitCallback(FileStoreTable table, String commitUser) {
         this.table = table;
         this.coreOptions = table.coreOptions();
+        this.commitUser = commitUser;
     }
 
     @Override
     public void call(Context context) {
-
         if (!ChainTableUtils.isScanFallbackDeltaBranch(coreOptions)) {
             return;
         }
-
         if (context.snapshot.commitKind() != CommitKind.OVERWRITE) {
             return;
         }
+        truncateSnapshotPartitions(context.deltaFiles);
+    }
 
-        FileStoreTable candidateTable = ChainTableUtils.resolveChainPrimaryTable(table);
+    /**
+     * The commit of this committable was published by an earlier attempt whose callback may not
+     * have completed, for example because the snapshot branch was unreachable right after the delta
+     * snapshot was written. Resolve that snapshot and redo the cleanup, which is idempotent. The
+     * partitions are taken from the manifest changes of the snapshot rather than from the
+     * committable, since an overwrite also clears partitions it wrote no new file to.
+     */
+    @Override
+    public void retry(ManifestCommittable committable) {
+        if (!ChainTableUtils.isScanFallbackDeltaBranch(coreOptions)) {
+            return;
+        }
+        List<Snapshot> snapshots =
+                table.snapshotManager()
+                        .findSnapshotsForIdentifiers(
+                                commitUser, Collections.singletonList(committable.identifier()));
+        if (snapshots.isEmpty()) {
+            LOG.warn(
+                    "No snapshot of commit user {} with identifier {} in table {}, "
+                            + "cannot redo the snapshot branch cleanup of its overwrite.",
+                    commitUser,
+                    committable.identifier(),
+                    table.name());
+            return;
+        }
+        for (Snapshot snapshot : snapshots) {
+            if (snapshot.commitKind() != CommitKind.OVERWRITE) {
+                continue;
+            }
+            truncateSnapshotPartitions(
+                    table.store()
+                            .newScan()
+                            .withKind(ScanMode.DELTA)
+                            .withSnapshot(snapshot.id())
+                            .plan()
+                            .files());
+        }
+    }
 
+    private void truncateSnapshotPartitions(List<ManifestEntry> deltaFiles) {
+        FileStoreTable candidateTable = ChainTableUtils.resolveChainPrimaryTable(table);
         FileStoreTable snapshotTable =
                 candidateTable.switchToBranch(coreOptions.scanFallbackSnapshotBranch());
-
         InternalRowPartitionComputer partitionComputer =
                 new InternalRowPartitionComputer(
                         coreOptions.partitionDefaultName(),
                         table.schema().logicalPartitionType(),
                         table.schema().partitionKeys().toArray(new String[0]),
                         coreOptions.legacyPartitionName());
-
         List<BinaryRow> overwritePartitions =
-                context.deltaFiles.stream()
+                deltaFiles.stream()
                         .map(ManifestEntry::partition)
                         .distinct()
                         .collect(Collectors.toList());
-
         if (overwritePartitions.isEmpty()) {
             return;
         }
-
         List<Map<String, String>> candidatePartitions =
                 overwritePartitions.stream()
                         .map(partitionComputer::generatePartValues)
                         .collect(Collectors.toList());
-
         try (BatchTableCommit commit = snapshotTable.newBatchWriteBuilder().newCommit()) {
             commit.truncatePartitions(candidatePartitions);
         } catch (Exception e) {
@@ -107,12 +152,6 @@ public void call(Context context) {
         }
     }
 
-    @Override
-    public void retry(ManifestCommittable committable) {
-        // No-op. Truncating the same partitions again is safe, but we prefer to only rely on the
-        // successful commit callback.
-    }
-
     @Override
     public void close() throws Exception {
         // no resources to close

paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommit.java

@@ -18,6 +18,7 @@
 
 package org.apache.paimon.operation;
 
+import org.apache.paimon.CoreOptions;
 import org.apache.paimon.Snapshot;
 import org.apache.paimon.data.BinaryRow;
 import org.apache.paimon.disk.IOManager;
@@ -44,6 +45,15 @@ public interface FileStoreCommit extends AutoCloseable {
 
     FileStoreCommit appendCommitCheckConflict(boolean appendCommitCheckConflict);
 
+    /**
+     * Whether {@link #filterCommitted} looks the previous commit of this user up without the lower
+     * bound of {@link CoreOptions#COMMIT_STRICT_MODE_LAST_SAFE_SNAPSHOT}. The bound only saves the
+     * lookup work for a commit user that is created for one run and so cannot have committed before
+     * its base snapshot; a caller-provided user that survives a restart can have, and needs the
+     * unbounded lookup to recognise a replay. Conflict detection keeps the bound either way.
+     */
+    FileStoreCommit filterCommittedIgnoresStrictModeBound(boolean ignoresStrictModeBound);
+
     FileStoreCommit rowIdCheckConflict(@Nullable Long rowIdCheckFromSnapshot);
 
     FileStoreCommit rowIdCheckConflictForMaterializeDvCompaction(

paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java

@@ -167,6 +167,7 @@ public class FileStoreCommitImpl implements FileStoreCommit {
     private final CommitCleaner commitCleaner;
 
     private boolean ignoreEmptyCommit;
+    private boolean filterCommittedIgnoresStrictModeBound = false;
     private CommitMetrics commitMetrics;
     private boolean appendCommitCheckConflict = false;
     private long lastCommittedSnapshotId = -1L;
@@ -249,6 +250,12 @@ public FileStoreCommit ignoreEmptyCommit(boolean ignoreEmptyCommit) {
         return this;
     }
 
+    @Override
+    public FileStoreCommit filterCommittedIgnoresStrictModeBound(boolean ignoresStrictModeBound) {
+        this.filterCommittedIgnoresStrictModeBound = ignoresStrictModeBound;
+        return this;
+    }
+
     @Override
     public FileStoreCommit withPartitionExpire(PartitionExpire partitionExpire) {
         this.conflictDetection.withPartitionExpire(partitionExpire);
@@ -296,7 +303,7 @@ public List<ManifestCommittable> filterCommitted(List<ManifestCommittable> commi
 
         Optional<Long> optionalStrictSnapshot = options.commitStrictModeLastSafeSnapshot();
         Optional<Snapshot> latestSnapshot;
-        if (optionalStrictSnapshot.isPresent()) {
+        if (optionalStrictSnapshot.isPresent() && !filterCommittedIgnoresStrictModeBound) {
             latestSnapshot =
                     snapshotManager.latestSnapshotOfUser(
                             commitUser,

paimon-core/src/main/java/org/apache/paimon/table/sink/BatchWriteBuilderImpl.java

@@ -36,7 +36,9 @@ public class BatchWriteBuilderImpl implements BatchWriteBuilder {
     private static final long serialVersionUID = 1L;
 
     private final InnerTable table;
-    private final String commitUser;
+
+    private String commitUser;
+    private boolean commitUserProvided = false;
 
     private Map<String, String> staticPartition;
     private @Nullable Long rowIdCheckFromSnapshot = null;
@@ -61,6 +63,20 @@ public Optional<WriteSelector> newWriteSelector() {
         return table.newWriteSelector();
     }
 
+    /**
+     * Use a caller-provided commit user instead of the random one.
+     *
+     * <p>A batch job has no reason to do this, but an engine which replays a failed batch with a
+     * stable identifier (for example a Spark Structured Streaming micro-batch) needs a commit user
+     * that survives the replay, so that {@link StreamTableCommit#filterAndCommit} can recognise
+     * what has already been committed.
+     */
+    public BatchWriteBuilderImpl withCommitUser(String commitUser) {
+        this.commitUser = commitUser;
+        this.commitUserProvided = true;
+        return this;
+    }
+
     @Override
     public BatchWriteBuilder withOverwrite(@Nullable Map<String, String> staticPartition) {
         this.staticPartition = staticPartition;
@@ -73,11 +89,12 @@ public BatchTableWrite newWrite() {
     }
 
     @Override
-    public BatchTableCommit newCommit() {
+    public InnerTableCommit newCommit() {
         InnerTableCommit commit =
                 table.newCommit(commitUser)
                         .withOverwrite(staticPartition)
-                        .rowIdCheckConflict(rowIdCheckFromSnapshot);
+                        .rowIdCheckConflict(rowIdCheckFromSnapshot)
+                        .filterCommittedIgnoresStrictModeBound(commitUserProvided);
         commit.ignoreEmptyCommit(
                 Options.fromMap(table.options())
                         .getOptional(CoreOptions.SNAPSHOT_IGNORE_EMPTY_COMMIT)

paimon-core/src/main/java/org/apache/paimon/table/sink/InnerTableCommit.java

@@ -54,6 +54,50 @@ public interface InnerTableCommit extends StreamTableCommit, BatchTableCommit {
 
     InnerTableCommit expireForEmptyCommit(boolean expireForEmptyCommit);
 
+    /**
+     * If this is set to true, {@link StreamTableCommit#filterAndCommit} verifies that every file it
+     * is about to commit still exists. By default it does.
+     *
+     * <p>The check guards a committable that was restored from an engine's state and may reference
+     * files deleted long ago. A caller which filters a committable it has just produced itself
+     * knows those files exist, and can skip a file listing proportional to the size of the
+     * committable.
+     */
+    InnerTableCommit checkFilesExistence(boolean checkFilesExistence);
+
+    /**
+     * Whether {@link StreamTableCommit#filterAndCommit} checks the append files of a committable
+     * against the files of the latest snapshot before committing them. By default it does.
+     *
+     * <p>The check guards a committable restored from an engine's state, whose files may have been
+     * committed, or removed, by an attempt the engine did not see complete. A caller filtering a
+     * committable it has just produced knows its files are new, and can skip a scan of the base
+     * files of every partition the committable touches. {@link #appendCommitCheckConflict} still
+     * forces the check regardless of this setting.
+     */
+    InnerTableCommit checkAppendFiles(boolean checkAppendFiles);
+
+    /**
+     * If this is set to true, maintenance runs on the committing thread and its failure is thrown
+     * to the caller, instead of running through an executor which stores the failure for the next
+     * commit to report.
+     *
+     * <p>A committer which commits once and is then closed has to do this: it is about to shut the
+     * executor down, so maintenance dispatched to it may never run, and there is no next commit to
+     * report a failure to. {@link BatchTableCommit#commit(List)} already behaves this way; a caller
+     * which commits through {@link StreamTableCommit#filterAndCommit} with the same one-shot
+     * lifecycle has to ask for it.
+     */
+    InnerTableCommit inlineMaintenance(boolean inlineMaintenance);
+
+    /**
+     * See {@link
+     * org.apache.paimon.operation.FileStoreCommit#filterCommittedIgnoresStrictModeBound}. A write
+     * builder enables this when it was given its commit user, since such a user can have committed
+     * before the base snapshot of the current write.
+     */
+    InnerTableCommit filterCommittedIgnoresStrictModeBound(boolean ignoresStrictModeBound);
+
     InnerTableCommit appendCommitCheckConflict(boolean appendCommitCheckConflict);
 
     InnerTableCommit rowIdCheckConflict(@Nullable Long rowIdCheckFromSnapshot);