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

docs/docs/spark/structured-streaming.md

@@ -56,6 +56,44 @@ 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.
+
+The commit user is derived from the checkpoint location of the query, so a query keeps it as long
+as it keeps its checkpoint. If the checkpoint location never reaches the sink (for example when it
+comes from `spark.sql.streaming.checkpointLocation`), the query id Spark stores in the checkpoint
+is used instead. 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 change of checkpoint location:
+
+```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.
+
+Starting a query from a new checkpoint location gives it a new commit user, so a micro-batch
+committed by the previous run is not recognised and its data is written again.
+
+A table using postpone bucket with `postpone.batch-write-fixed-bucket` commits 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 checkpoint location of the query, so it is stable across restarts; set it explicitly only if the same query has to keep its identity across a change of checkpoint location.</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/table/sink/BatchWriteBuilderImpl.java

@@ -36,7 +36,8 @@ public class BatchWriteBuilderImpl implements BatchWriteBuilder {
     private static final long serialVersionUID = 1L;
 
     private final InnerTable table;
-    private final String commitUser;
+
+    private String commitUser;
 
     private Map<String, String> staticPartition;
     private @Nullable Long rowIdCheckFromSnapshot = null;
@@ -61,6 +62,19 @@ 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;
+        return this;
+    }
+
     @Override
     public BatchWriteBuilder withOverwrite(@Nullable Map<String, String> staticPartition) {
         this.staticPartition = staticPartition;
@@ -73,7 +87,7 @@ public BatchTableWrite newWrite() {
     }
 
     @Override
-    public BatchTableCommit newCommit() {
+    public InnerTableCommit newCommit() {
         InnerTableCommit commit =
                 table.newCommit(commitUser)
                         .withOverwrite(staticPartition)

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

@@ -54,6 +54,17 @@ 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);
+
     InnerTableCommit appendCommitCheckConflict(boolean appendCommitCheckConflict);
 
     InnerTableCommit rowIdCheckConflict(@Nullable Long rowIdCheckFromSnapshot);

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

@@ -94,6 +94,7 @@ public class TableCommitImpl implements InnerTableCommit {
     @Nullable private List<BinaryRow> overwriteStaticPartitions = null;
     private boolean batchCommitted = false;
     private boolean expireForEmptyCommit = true;
+    private boolean checkFilesExistence = true;
 
     public TableCommitImpl(
             FileStoreCommit commit,
@@ -169,6 +170,12 @@ public TableCommitImpl expireForEmptyCommit(boolean expireForEmptyCommit) {
         return this;
     }
 
+    @Override
+    public TableCommitImpl checkFilesExistence(boolean checkFilesExistence) {
+        this.checkFilesExistence = checkFilesExistence;
+        return this;
+    }
+
     @Override
     public TableCommitImpl appendCommitCheckConflict(boolean appendCommitCheckConflict) {
         commit.appendCommitCheckConflict(appendCommitCheckConflict);
@@ -333,13 +340,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());

paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkConnectorOptions.java

@@ -109,6 +109,19 @@ public class SparkConnectorOptions {
                             "Wait time in milliseconds between retry attempts for Spark V1 UPDATE "
                                     + "on data-evolution tables after row-id range update conflicts.");
 
+    public static final ConfigOption<String> STREAM_WRITE_COMMIT_USER =
+            key("write.stream.commit-user")
+                    .stringType()
+                    .noDefaultValue()
+                    .withDescription(
+                            "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 "
+                                    + "checkpoint location of the query, so it is stable across "
+                                    + "restarts; set it explicitly only if the same query has to "
+                                    + "keep its identity across a change of checkpoint location.");
+
     public static final ConfigOption<Integer> MAX_FILES_PER_TRIGGER =
             key("read.stream.maxFilesPerTrigger")
                     .intType()

paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala

@@ -44,21 +44,24 @@ import org.apache.paimon.types.RowKind
 import org.apache.paimon.utils.{SerializationUtils, UriReaderFactory}
 
 import org.apache.spark.{Partitioner, TaskContext}
+import org.apache.spark.internal.Logging
 import org.apache.spark.rdd.RDD
 import org.apache.spark.sql._
 import org.apache.spark.sql.functions._
 
 import java.io.IOException
-import java.util.{Map => JMap}
+import java.util.{Collections, Map => JMap}
 import java.util.Collections.singletonMap
 
 import scala.collection.JavaConverters._
 
 case class PaimonSparkWriter(
     table: FileStoreTable,
     writeRowTracking: Boolean = false,
-    batchId: Option[Long] = None)
-  extends WriteHelper {
+    batchId: Option[Long] = None,
+    commitUser: Option[String] = None)
+  extends WriteHelper
+  with Logging {
 
   private lazy val tableSchema = table.schema
 
@@ -99,7 +102,13 @@ case class PaimonSparkWriter(
     if (bucketNum.isPresent) Some(bucketNum.get().intValue()) else None
   }
 
-  val writeBuilder: BatchWriteBuilder = table.newBatchWriteBuilder()
+  val writeBuilder: BatchWriteBuilder = {
+    val builder = table.newBatchWriteBuilder()
+    // A streaming write commits under a commit user that survives a restart, so that a replayed
+    // micro-batch can be recognised as already committed.
+    commitUser.foreach(builder.asInstanceOf[BatchWriteBuilderImpl].withCommitUser)
+    builder
+  }
 
   def withOverwrite(): PaimonSparkWriter = withOverwrite(java.util.Collections.emptyMap())
 
@@ -454,6 +463,16 @@ case class PaimonSparkWriter(
     writeBuilder.asInstanceOf[BatchWriteBuilderImpl].rowIdCheckConflict(rowIdCheckFromSnapshot)
   }
 
+  /**
+   * The commit identifier to deduplicate on, present only for a streaming write that has both a
+   * batch id and a commit user that is stable across restarts.
+   */
+  private def idempotentCommitIdentifier: Option[Long] =
+    for {
+      identifier <- batchId
+      _ <- commitUser
+    } yield identifier
+
   def commit(commitMessages: Seq[CommitMessage]): Unit = {
     commit(commitMessages, null)
   }
@@ -463,6 +482,13 @@ case class PaimonSparkWriter(
       if (stagedSparkSession == null) {
         throw new IllegalStateException("Postpone staged write has no SparkSession.")
       }
+      idempotentCommitIdentifier.foreach {
+        identifier =>
+          logWarning(
+            s"Micro-batch $identifier is written to a postpone bucket table through a staged " +
+              "commit, which cannot deduplicate a replayed batch. A failure of this query may " +
+              "duplicate the batch.")
+      }
       val finalOperation = Option(operation).getOrElse(Snapshot.Operation.WRITE)
       val finalMessages = new SparkPostponeStagedCommitter(
         table,
@@ -472,14 +498,29 @@ case class PaimonSparkWriter(
       postCommit(finalMessages)
       return
     }
-    val activeWriteBuilder =
-      Option(directPostponeWriteBuilder).getOrElse(writeBuilder)
-    val tableCommit = activeWriteBuilder.newCommit()
+    val tableCommit: InnerTableCommit =
+      if (directPostponeWriteBuilder != null) {
+        directPostponeWriteBuilder.newCommit()
+      } else {
+        writeBuilder.asInstanceOf[BatchWriteBuilderImpl].newCommit()
+      }
     if (operation != null) {
       tableCommit.withOperation(operation)
     }
     try {
-      tableCommit.commit(commitMessages.toList.asJava)
+      idempotentCommitIdentifier match {
+        case Some(identifier) =>
+          // Structured Streaming replays a micro-batch with its original batch id after a failure.
+          // Committing under a stable commit user lets Paimon skip a replay it already committed,
+          // instead of duplicating the whole batch. The files being committed were written by this
+          // very batch, so there is no need to list them to prove that they still exist.
+          tableCommit
+            .checkFilesExistence(false)
+            .filterAndCommit(
+              Collections.singletonMap(Long.box(identifier), commitMessages.toList.asJava))
+        case None =>
+          tableCommit.commit(commitMessages.toList.asJava)
+      }
     } catch {
       case e: Throwable => throw new RuntimeException(e);
     } finally {

paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/WriteIntoPaimonTable.scala

@@ -40,7 +40,8 @@ case class WriteIntoPaimonTable(
     saveMode: SaveMode,
     _data: DataFrame,
     options: Options,
-    batchId: Option[Long] = None)
+    batchId: Option[Long] = None,
+    commitUser: Option[String] = None)
   extends RunnableCommand
   with ExpressionHelper
   with SchemaEvolutionHelper
@@ -58,7 +59,7 @@ case class WriteIntoPaimonTable(
     updateTableWithOptions(
       Map(DYNAMIC_PARTITION_OVERWRITE.key -> dynamicPartitionOverwriteMode.toString))
 
-    val writer = PaimonSparkWriter(table, batchId = batchId)
+    val writer = PaimonSparkWriter(table, batchId = batchId, commitUser = commitUser)
     if (overwritePartition != null) {
       writer.withOverwrite(overwritePartition.asJava)
     }