[spark] Commit a streaming micro-batch through the stream write API

docs/docs/spark/structured-streaming.md

@@ -66,6 +66,62 @@ Use a durable checkpoint location accessible to the cluster for deployed jobs. G
 its own checkpoint directory. Stop this example with `writer.stop()`.
 Streaming writes also support [Schema Evolution on Write](./schema-evolution).
 
+### 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. It is the identity of one streaming writer and must be unique
+to it: it is not read from table properties, which every writer of the table would share, and two
+queries given the same identity would each drop the other's batches as replays. `commit.user-prefix`
+is read from the table as usual and prefixes the derived user, so a job keeps the name its table was
+configured with without two queries sharing an identity.
+
+```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 streaming write to a postpone bucket table writes the postpone bucket, as a Flink streaming job
+does, whatever `postpone.batch-write-fixed-bucket` says: that option is for a batch job which ends
+with its commit. The rows become readable once a
+[compaction](../primary-key-table/data-distribution#postpone-bucket) has sorted them into real
+buckets.
+
+:::
+
+The committer of a query is created with its first micro-batch and lives until the query
+terminates, again as a Flink committer lives across checkpoints. Only the first micro-batch of a
+run looks up what a previous run committed; the maintenance a commit schedules, such as tag
+creation and snapshot or partition expiration, runs while the next micro-batch is written when
+`snapshot.expire.execution-mode` is `async`.
+
 ## Streaming Query
 
 :::info

docs/generated/spark_connector_configuration.html

@@ -146,6 +146,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. It is the identity of one streaming writer and must be unique to it: set it as an option of the writer or as a session conf, never as a table property, which every writer of the table would share.</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

@@ -442,7 +442,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,20 +19,41 @@
 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.io.CompactIncrement;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.DataIncrement;
+import org.apache.paimon.manifest.FileEntry;
 import org.apache.paimon.manifest.ManifestCommittable;
 import org.apache.paimon.manifest.ManifestEntry;
+import org.apache.paimon.operation.FileStoreCommit;
+import org.apache.paimon.options.Options;
 import org.apache.paimon.table.FileStoreTable;
 import org.apache.paimon.table.sink.BatchTableCommit;
+import org.apache.paimon.table.sink.BatchWriteBuilder;
 import org.apache.paimon.table.sink.CommitCallback;
+import org.apache.paimon.table.sink.CommitMessageImpl;
+import org.apache.paimon.table.source.ScanMode;
 import org.apache.paimon.utils.ChainTableUtils;
 import org.apache.paimon.utils.InternalRowPartitionComputer;
+import org.apache.paimon.utils.SnapshotManager;
+import org.apache.paimon.utils.Triple;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.stream.Collectors;
 
+import static org.apache.paimon.CoreOptions.createCommitUser;
+
 /**
  * A {@link CommitCallback} implementation to maintain chain table snapshot branch for overwrite
  * commits.
@@ -50,52 +71,179 @@
  */
 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);
+    }
 
+    /**
+     * 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;
+            }
+            List<BinaryRow> overwritePartitions =
+                    overwritePartitions(
+                            table.store()
+                                    .newScan()
+                                    .withKind(ScanMode.DELTA)
+                                    .withSnapshot(snapshot.id())
+                                    .plan()
+                                    .files());
+            clearSnapshotFilesAsOf(overwritePartitions, snapshot.timeMillis());
+        }
+    }
 
-        if (context.snapshot.commitKind() != CommitKind.OVERWRITE) {
+    /**
+     * Clear the files of the given partitions that the snapshot branch held when the overwrite was
+     * published, and nothing that landed there since. That is what {@link #call} cleared at the
+     * time; a replay that repeats it after later data arrived must not take that data with it,
+     * whether the original cleanup had completed or not.
+     */
+    private void clearSnapshotFilesAsOf(List<BinaryRow> partitions, long overwriteCommitMillis) {
+        if (partitions.isEmpty()) {
+            return;
+        }
+        FileStoreTable snapshotTable = snapshotTable();
+        SnapshotManager snapshotManager = snapshotTable.snapshotManager();
+        Snapshot asOf = snapshotManager.earlierOrEqualTimeMills(overwriteCommitMillis);
+        Snapshot latest = snapshotManager.latestSnapshot();
+        if (asOf == null || latest == null) {
+            return;
+        }
+        Set<FileEntry.Identifier> current =
+                filesOf(snapshotTable, latest, partitions).stream()
+                        .map(ManifestEntry::identifier)
+                        .collect(Collectors.toSet());
+        // Files cleared since, by the original callback or an earlier retry, are gone already.
+        Map<Triple<BinaryRow, Integer, Integer>, List<DataFileMeta>> superseded = new HashMap<>();
+        for (ManifestEntry entry : filesOf(snapshotTable, asOf, partitions)) {
+            if (current.contains(entry.identifier())) {
+                superseded
+                        .computeIfAbsent(
+                                Triple.of(
+                                        entry.partition().copy(),
+                                        entry.bucket(),
+                                        entry.totalBuckets()),
+                                k -> new ArrayList<>())
+                        .add(entry.file());
+            }
+        }
+        if (superseded.isEmpty()) {
             return;
         }
+        ManifestCommittable committable =
+                new ManifestCommittable(BatchWriteBuilder.COMMIT_IDENTIFIER);
+        superseded.forEach(
+                (key, files) ->
+                        committable.addFileCommittable(
+                                new CommitMessageImpl(
+                                        key.f0,
+                                        key.f1,
+                                        key.f2,
+                                        new DataIncrement(
+                                                Collections.emptyList(),
+                                                files,
+                                                Collections.emptyList()),
+                                        CompactIncrement.emptyIncrement())));
+        try (FileStoreCommit commit =
+                snapshotTable
+                        .store()
+                        .newCommit(
+                                createCommitUser(new Options(snapshotTable.options())),
+                                snapshotTable)) {
+            // The files being removed must still be there; a concurrent change to them is a
+            // conflict to report, not to skip over.
+            commit.commit(committable, true);
+        } catch (Exception e) {
+            throw new RuntimeException(
+                    String.format(
+                            "Failed to clear the files of partitions %s in the snapshot table.",
+                            partitions),
+                    e);
+        }
+    }
+
+    private static List<ManifestEntry> filesOf(
+            FileStoreTable table, Snapshot snapshot, List<BinaryRow> partitions) {
+        return table.store()
+                .newScan()
+                .withSnapshot(snapshot)
+                .withPartitionFilter(partitions)
+                .plan()
+                .files();
+    }
 
+    private FileStoreTable snapshotTable() {
         FileStoreTable candidateTable = ChainTableUtils.resolveChainPrimaryTable(table);
+        return candidateTable.switchToBranch(coreOptions.scanFallbackSnapshotBranch());
+    }
 
-        FileStoreTable snapshotTable =
-                candidateTable.switchToBranch(coreOptions.scanFallbackSnapshotBranch());
+    private static List<BinaryRow> overwritePartitions(List<ManifestEntry> deltaFiles) {
+        return deltaFiles.stream()
+                .map(ManifestEntry::partition)
+                .distinct()
+                .collect(Collectors.toList());
+    }
 
+    private void truncateSnapshotPartitions(List<ManifestEntry> deltaFiles) {
+        List<BinaryRow> overwritePartitions = overwritePartitions(deltaFiles);
+        if (overwritePartitions.isEmpty()) {
+            return;
+        }
         InternalRowPartitionComputer partitionComputer =
                 new InternalRowPartitionComputer(
                         coreOptions.partitionDefaultName(),
                         table.schema().logicalPartitionType(),
                         table.schema().partitionKeys().toArray(new String[0]),
                         coreOptions.legacyPartitionName());
-
-        List<BinaryRow> overwritePartitions =
-                context.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());
-
+        FileStoreTable snapshotTable = snapshotTable();
         try (BatchTableCommit commit = snapshotTable.newBatchWriteBuilder().newCommit()) {
             commit.truncatePartitions(candidatePartitions);
         } catch (Exception e) {
@@ -107,12 +255,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