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

docs/docs/spark/structured-streaming.md

@@ -56,6 +56,48 @@ 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.
+
+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/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);

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

@@ -38,7 +38,9 @@ public class PostponeFixedBucketWriteBuilder implements BatchWriteBuilder {
     private static final long serialVersionUID = 1L;
 
     private final FileStoreTable table;
-    private final String commitUser;
+
+    private String commitUser;
+    private boolean commitUserProvided = false;
 
     @Nullable private Map<String, String> staticPartition;
 
@@ -50,6 +52,16 @@ public PostponeFixedBucketWriteBuilder(FileStoreTable table) {
         this.commitUser = createCommitUser(new Options(table.options()));
     }
 
+    /**
+     * Use a caller-provided commit user instead of the random one, for the same reason as {@link
+     * BatchWriteBuilderImpl#withCommitUser}.
+     */
+    public PostponeFixedBucketWriteBuilder withCommitUser(String commitUser) {
+        this.commitUser = commitUser;
+        this.commitUserProvided = true;
+        return this;
+    }
+
     @Override
     public String tableName() {
         return table.name();
@@ -87,7 +99,8 @@ public TableCommitImpl newCommit() {
                 Options.fromMap(table.options())
                         .getOptional(CoreOptions.SNAPSHOT_IGNORE_EMPTY_COMMIT)
                         .orElse(true);
-        return newCommit(commitUser, ignoreEmpty);
+        return newCommit(commitUser, ignoreEmpty)
+                .filterCommittedIgnoresStrictModeBound(commitUserProvided);
     }
 
     public TableCommitImpl newCommit(String commitUser, boolean ignoreEmptyCommit) {

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

@@ -94,6 +94,9 @@ public class TableCommitImpl implements InnerTableCommit {
     @Nullable private List<BinaryRow> overwriteStaticPartitions = null;
     private boolean batchCommitted = false;
     private boolean expireForEmptyCommit = true;
+    private boolean checkFilesExistence = true;
+    private boolean checkAppendFiles = true;
+    private boolean inlineMaintenance = false;
 
     public TableCommitImpl(
             FileStoreCommit commit,
@@ -169,6 +172,30 @@ public TableCommitImpl expireForEmptyCommit(boolean expireForEmptyCommit) {
         return this;
     }
 
+    @Override
+    public TableCommitImpl checkFilesExistence(boolean checkFilesExistence) {
+        this.checkFilesExistence = checkFilesExistence;
+        return this;
+    }
+
+    @Override
+    public TableCommitImpl checkAppendFiles(boolean checkAppendFiles) {
+        this.checkAppendFiles = checkAppendFiles;
+        return this;
+    }
+
+    @Override
+    public TableCommitImpl inlineMaintenance(boolean inlineMaintenance) {
+        this.inlineMaintenance = inlineMaintenance;
+        return this;
+    }
+
+    @Override
+    public TableCommitImpl filterCommittedIgnoresStrictModeBound(boolean ignoresStrictModeBound) {
+        commit.filterCommittedIgnoresStrictModeBound(ignoresStrictModeBound);
+        return this;
+    }
+
     @Override
     public TableCommitImpl appendCommitCheckConflict(boolean appendCommitCheckConflict) {
         commit.appendCommitCheckConflict(appendCommitCheckConflict);
@@ -265,7 +292,8 @@ public int filterAndCommit(Map<Long, List<CommitMessage>> commitIdentifiersAndMe
         return filterAndCommitMultiple(
                 commitIdentifiersAndMessages.entrySet().stream()
                         .map(e -> createManifestCommittable(e.getKey(), e.getValue()))
-                        .collect(Collectors.toList()));
+                        .collect(Collectors.toList()),
+                checkAppendFiles);
     }
 
     private ManifestCommittable createManifestCommittable(
@@ -333,13 +361,15 @@ public int filterAndCommitMultiple(
         List<ManifestCommittable> retryCommittables = commit.filterCommitted(sortedCommittables);
 
         if (!retryCommittables.isEmpty()) {
-            checkFilesExistence(retryCommittables);
+            if (checkFilesExistence) {
+                verifyFilesExist(retryCommittables);
+            }
             commitMultiple(retryCommittables, checkAppendFiles);
         }
         return retryCommittables.size();
     }
 
-    private void checkFilesExistence(List<ManifestCommittable> committables) {
+    private void verifyFilesExist(List<ManifestCommittable> committables) {
         List<Path> files = new ArrayList<>();
         DataFilePathFactories factories = new DataFilePathFactories(commit.pathFactory());
         IndexFilePathFactories indexFactories = new IndexFilePathFactories(commit.pathFactory());
@@ -411,7 +441,7 @@ private void maintain(long identifier, ExecutorService executor, boolean doExpir
             throw new RuntimeException(maintainError.get());
         }
 
-        if (batchCommitted) {
+        if (batchCommitted || inlineMaintenance) {
             maintain(identifier, doExpire);
         } else {
             executor.execute(

paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeySimpleTableTest.java

@@ -88,6 +88,7 @@
 import org.apache.paimon.utils.JsonSerdeUtil;
 import org.apache.paimon.utils.Pair;
 import org.apache.paimon.utils.RoaringBitmap32;
+import org.apache.paimon.utils.SnapshotManager;
 
 import org.apache.commons.math3.random.RandomDataGenerator;
 import org.assertj.core.api.Assertions;
@@ -124,6 +125,7 @@
 import static org.apache.paimon.CoreOptions.CHANGELOG_NUM_RETAINED_MAX;
 import static org.apache.paimon.CoreOptions.CHANGELOG_NUM_RETAINED_MIN;
 import static org.apache.paimon.CoreOptions.CHANGELOG_PRODUCER;
+import static org.apache.paimon.CoreOptions.COMMIT_STRICT_MODE_LAST_SAFE_SNAPSHOT;
 import static org.apache.paimon.CoreOptions.ChangelogProducer.LOOKUP;
 import static org.apache.paimon.CoreOptions.DELETION_VECTORS_ENABLED;
 import static org.apache.paimon.CoreOptions.FILE_FORMAT;
@@ -284,6 +286,50 @@ public void testPostponeBucket() throws Exception {
         assertThat(file.valueStatsCols()).isEmpty();
     }
 
+    @Test
+    public void testPostponeFixedBucketReplayLookupWithProvidedUser() throws Exception {
+        FileStoreTable table =
+                createFileStoreTable(options -> options.set(BUCKET, BucketMode.POSTPONE_BUCKET));
+        SnapshotManager sm = table.snapshotManager();
+
+        // A first run commits identifier 0 under a caller-provided user.
+        PostponeFixedBucketWriteBuilder first =
+                table.newPostponeFixedBucketWriteBuilder().withCommitUser("user");
+        try (TableWriteImpl<?> write = first.newWrite();
+                InnerTableCommit commit = first.newCommit()) {
+            write.writeAndReturn(rowData(1, 1, 1L), 0, 1);
+            commit.filterAndCommit(Collections.singletonMap(0L, write.prepareCommit()));
+        }
+        long committed = sm.latestSnapshotId();
+
+        // A restarted run replays identifier 0 in strict mode bounded by that snapshot, the way
+        // a direct postpone write starts from the latest snapshot. The provided user has to be
+        // looked up beyond the bound for the replay to be recognised.
+        Map<String, String> strict = new HashMap<>();
+        strict.put(COMMIT_STRICT_MODE_LAST_SAFE_SNAPSHOT.key(), String.valueOf(committed));
+        FileStoreTable strictTable = table.copy(strict);
+        PostponeFixedBucketWriteBuilder replay =
+                strictTable.newPostponeFixedBucketWriteBuilder().withCommitUser("user");
+        try (TableWriteImpl<?> write = replay.newWrite();
+                InnerTableCommit commit = replay.newCommit()) {
+            write.writeAndReturn(rowData(1, 1, 1L), 0, 1);
+            commit.filterAndCommit(Collections.singletonMap(0L, write.prepareCommit()));
+        }
+        assertThat(sm.latestSnapshotId())
+                .as("a replay by a provided user must be recognised across the strict mode bound")
+                .isEqualTo(committed);
+
+        // A committer created for an explicitly passed user is one the caller manages itself,
+        // like the staged committer with its per-run user, and keeps the bound.
+        PostponeFixedBucketWriteBuilder explicit = strictTable.newPostponeFixedBucketWriteBuilder();
+        try (TableWriteImpl<?> write = explicit.newWrite("user", null);
+                InnerTableCommit commit = explicit.newCommit("user", true)) {
+            write.writeAndReturn(rowData(2, 2, 2L), 0, 1);
+            commit.filterAndCommit(Collections.singletonMap(0L, write.prepareCommit()));
+        }
+        assertThat(sm.latestSnapshotId()).isEqualTo(committed + 1);
+    }
+
     @Test
     public void testPostponeFixedBucketWriteBuilder() throws Exception {
         FileStoreTable table =