write.use-v2-write |
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/sink/InnerTableCommit.java b/paimon-core/src/main/java/org/apache/paimon/table/sink/InnerTableCommit.java
index d513dd27baec..48ddb2cd4c42 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/sink/InnerTableCommit.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/sink/InnerTableCommit.java
@@ -65,6 +65,19 @@ public interface InnerTableCommit extends StreamTableCommit, BatchTableCommit {
*/
InnerTableCommit checkFilesExistence(boolean checkFilesExistence);
+ /**
+ * 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.
+ *
+ * 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);
+
InnerTableCommit appendCommitCheckConflict(boolean appendCommitCheckConflict);
InnerTableCommit rowIdCheckConflict(@Nullable Long rowIdCheckFromSnapshot);
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/sink/TableCommitImpl.java b/paimon-core/src/main/java/org/apache/paimon/table/sink/TableCommitImpl.java
index 48cb61eecc78..55f4336d92b1 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/sink/TableCommitImpl.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/sink/TableCommitImpl.java
@@ -95,6 +95,7 @@ public class TableCommitImpl implements InnerTableCommit {
private boolean batchCommitted = false;
private boolean expireForEmptyCommit = true;
private boolean checkFilesExistence = true;
+ private boolean inlineMaintenance = false;
public TableCommitImpl(
FileStoreCommit commit,
@@ -176,6 +177,12 @@ public TableCommitImpl checkFilesExistence(boolean checkFilesExistence) {
return this;
}
+ @Override
+ public TableCommitImpl inlineMaintenance(boolean inlineMaintenance) {
+ this.inlineMaintenance = inlineMaintenance;
+ return this;
+ }
+
@Override
public TableCommitImpl appendCommitCheckConflict(boolean appendCommitCheckConflict) {
commit.appendCommitCheckConflict(appendCommitCheckConflict);
@@ -420,7 +427,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(
diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkConnectorOptions.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkConnectorOptions.java
index c3f8f804e799..956e4ec4f90d 100644
--- a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkConnectorOptions.java
+++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkConnectorOptions.java
@@ -118,9 +118,10 @@ public class SparkConnectorOptions {
+ "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.");
+ + "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.");
public static final ConfigOption MAX_FILES_PER_TRIGGER =
key("read.stream.maxFilesPerTrigger")
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala
index f91916349217..61f6d86ec1dc 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala
@@ -516,6 +516,9 @@ case class PaimonSparkWriter(
// very batch, so there is no need to list them to prove that they still exist.
tableCommit
.checkFilesExistence(false)
+ // This committer is closed right after the batch, so maintenance cannot be left to an
+ // executor that is about to be shut down, nor a failure to a commit that never comes.
+ .inlineMaintenance(true)
.filterAndCommit(
Collections.singletonMap(Long.box(identifier), commitMessages.toList.asJava))
case None =>
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/sources/PaimonSink.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/sources/PaimonSink.scala
index 6bfec18c4ac1..8256af6c3591 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/sources/PaimonSink.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/sources/PaimonSink.scala
@@ -50,17 +50,26 @@ class PaimonSink(
* completed. Committing every batch under a commit user that is stable across restarts lets
* Paimon skip such a replay instead of committing its data twice.
*
- * Resolved lazily: neither the checkpoint location nor the query id is available on the thread
+ * What the commit user has to identify is one incarnation of a checkpoint, not the place it is
+ * stored. Paimon skips a batch whose id a previous run committed under the same user, so reusing
+ * a user across two different queries drops the data of the second one, while changing it within
+ * one query brings back the duplicate. The query id Spark persists in the checkpoint metadata is
+ * exactly that identity: it is new when a checkpoint is recreated, unchanged when a query resumes
+ * from one, and independent of how the location is spelled.
+ *
+ * Resolved lazily: neither the query id nor the checkpoint location is available on the thread
* that constructs the sink.
*/
private lazy val commitUser: String = {
configuredCommitUser.getOrElse {
- checkpointLocation
- .map(derivedCommitUser("checkpoint", _))
- .orElse(queryId.map(derivedCommitUser("query", _)))
+ queryId
+ .map(derivedCommitUser("query", _))
+ // Only reachable outside a stream execution, e.g. a direct addBatch call. A location
+ // cannot tell a recreated checkpoint from a resumed one, so it is a last resort.
+ .orElse(checkpointLocation.map(derivedCommitUser("checkpoint", _)))
.getOrElse {
logWarning(
- "This streaming write has neither a checkpoint location nor a query id to derive a " +
+ "This streaming write has neither a query id nor a checkpoint location to derive a " +
"stable commit user from, so a replayed micro-batch cannot be recognised and may " +
s"be committed twice. Set '${SparkConnectorOptions.STREAM_WRITE_COMMIT_USER.key}' " +
"to make the write idempotent.")
@@ -88,10 +97,8 @@ class PaimonSink(
}
/**
- * The id Spark persists in the checkpoint metadata, hence stable across restarts of the same
- * query. It covers the case of a checkpoint location that never reaches the sink options, for
- * example one taken from `spark.sql.streaming.checkpointLocation`. It is a thread local of the
- * stream execution thread, so it can only be read from within [[addBatch]].
+ * The id Spark persists in the checkpoint metadata. It is a thread local of the stream execution
+ * thread, so it can only be read from within [[addBatch]].
*/
private def queryId: Option[String] =
Option(sqlContext.sparkContext.getLocalProperty(PaimonSink.QUERY_ID_KEY)).filter(_.nonEmpty)
diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/PaimonSinkIdempotencyTest.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/PaimonSinkIdempotencyTest.scala
index ea0543141a4d..4c10651ad4ea 100644
--- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/PaimonSinkIdempotencyTest.scala
+++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/PaimonSinkIdempotencyTest.scala
@@ -48,6 +48,13 @@ class PaimonSinkIdempotencyTest extends PaimonSparkTestBase with StreamTest {
private def latestCommitUser(tableName: String): String =
loadTable(tableName).snapshotManager().latestSnapshot().commitUser()
+ private def deleteRecursively(file: File): Unit = {
+ if (file.isDirectory) {
+ file.listFiles().foreach(deleteRecursively)
+ }
+ file.delete()
+ }
+
private def runToCompletion(query: StreamingQuery): Unit = {
try {
query.processAllAvailable()
@@ -95,8 +102,8 @@ class PaimonSinkIdempotencyTest extends PaimonSparkTestBase with StreamTest {
checkAnswer(spark.sql("SELECT * FROM T ORDER BY a"), expected)
assert(snapshotCount("T") == 1)
assert(
- latestCommitUser("T").startsWith("spark-checkpoint-"),
- s"expected a commit user derived from the checkpoint location, " +
+ latestCommitUser("T").startsWith("spark-query-"),
+ s"expected a commit user derived from the query id, " +
s"but got '${latestCommitUser("T")}'"
)
@@ -305,19 +312,142 @@ class PaimonSinkIdempotencyTest extends PaimonSparkTestBase with StreamTest {
}
test("Paimon Sink: addBatch with a repeated batchId must be a no-op") {
- spark.sql("CREATE TABLE T2 (a INT, b STRING)")
- val sink = new PaimonSink(
- spark.sqlContext,
- loadTable("T2"),
- Nil,
- OutputMode.Append(),
- Options.fromMap(Collections.singletonMap("write.stream.commit-user", "direct-api")))
-
- val batch: DataFrame = Seq((1, "a"), (2, "b")).toDF("a", "b")
- sink.addBatch(0L, batch)
- sink.addBatch(0L, batch)
-
- checkAnswer(spark.sql("SELECT * FROM T2 ORDER BY a"), Row(1, "a") :: Row(2, "b") :: Nil)
- assert(snapshotCount("T2") == 1)
+ withTempDir {
+ checkpointDir =>
+ withTable("T2") {
+ spark.sql("CREATE TABLE T2 (a INT, b STRING)")
+ // Called outside a stream execution there is no query id, so this also covers the
+ // checkpoint location fallback.
+ val sink = new PaimonSink(
+ spark.sqlContext,
+ loadTable("T2"),
+ Nil,
+ OutputMode.Append(),
+ Options.fromMap(
+ Collections.singletonMap("checkpointLocation", checkpointDir.getCanonicalPath))
+ )
+
+ val batch: DataFrame = Seq((1, "a"), (2, "b")).toDF("a", "b")
+ sink.addBatch(0L, batch)
+ sink.addBatch(0L, batch)
+
+ checkAnswer(spark.sql("SELECT * FROM T2 ORDER BY a"), Row(1, "a") :: Row(2, "b") :: Nil)
+ assert(snapshotCount("T2") == 1)
+ assert(
+ latestCommitUser("T2").startsWith("spark-checkpoint-"),
+ s"expected a commit user derived from the checkpoint location, " +
+ s"but got '${latestCommitUser("T2")}'"
+ )
+ }
+ }
+ }
+
+ test("Paimon Sink: a new query reusing a checkpoint location must not skip its batches") {
+ failAfter(streamingTimeout) {
+ withTempDir {
+ dir =>
+ spark.sql("CREATE TABLE T (a INT, b STRING)")
+ val location = loadTable("T").location().toString
+ val checkpointDir = new File(dir, "cp")
+
+ def runOneBatch(row: (Int, String)): Unit = {
+ val inputData = MemoryStream[(Int, String)]
+ val df = inputData.toDS().toDF("a", "b")
+ inputData.addData(row)
+ runToCompletion(
+ df.writeStream
+ .option("checkpointLocation", checkpointDir.getCanonicalPath)
+ .format("paimon")
+ .start(location))
+ }
+
+ runOneBatch((1, "old"))
+ val firstCommitUser = latestCommitUser("T")
+
+ // The checkpoint is dropped and an unrelated query starts at the same location. Its
+ // batch ids start at 0 again, so reusing the identity of the previous query would
+ // make Paimon skip its data as an already committed replay.
+ deleteRecursively(checkpointDir)
+ runOneBatch((2, "new"))
+
+ checkAnswer(
+ spark.sql("SELECT * FROM T ORDER BY a"),
+ Row(1, "old") :: Row(2, "new") :: Nil)
+ assert(
+ latestCommitUser("T") != firstCommitUser,
+ "a query that does not continue the previous checkpoint must not reuse its " +
+ "commit user")
+ }
+ }
+ }
+
+ test("Paimon Sink: an equivalent spelling of the checkpoint location keeps the identity") {
+ failAfter(streamingTimeout) {
+ withTempDir {
+ checkpointDir =>
+ spark.sql("CREATE TABLE T (a INT, b STRING)")
+ val location = loadTable("T").location().toString
+ val checkpointPath = checkpointDir.getCanonicalPath
+
+ val inputData = MemoryStream[(Int, String)]
+ val df = inputData.toDS().toDF("a", "b")
+ inputData.addData((1, "a"))
+
+ runToCompletion(
+ df.writeStream
+ .option("checkpointLocation", checkpointPath)
+ .format("paimon")
+ .start(location))
+ val firstCommitUser = latestCommitUser("T")
+
+ dropCommitLogEntry(checkpointPath, 0)
+ // The same checkpoint, written with a trailing separator.
+ runToCompletion(
+ df.writeStream
+ .option("checkpointLocation", checkpointPath + "/")
+ .format("paimon")
+ .start(location))
+
+ checkAnswer(spark.sql("SELECT * FROM T"), Row(1, "a") :: Nil)
+ assert(
+ latestCommitUser("T") == firstCommitUser,
+ "the same query resuming the same checkpoint must keep its commit user")
+ }
+ }
+ }
+
+ test("Paimon Sink: expiration of a micro-batch completes before the committer closes") {
+ failAfter(streamingTimeout) {
+ withTempDir {
+ checkpointDir =>
+ // Async expiration plus a committer that is closed after every micro-batch:
+ // maintenance has to run before that close, or expiration never happens.
+ spark.sql(
+ "CREATE TABLE T (a INT, b STRING) TBLPROPERTIES (" +
+ "'snapshot.expire.execution-mode' = 'async', " +
+ "'snapshot.num-retained.min' = '1', " +
+ "'snapshot.num-retained.max' = '1')")
+ val location = loadTable("T").location().toString
+
+ val inputData = MemoryStream[(Int, String)]
+ val df = inputData.toDS().toDF("a", "b")
+ val query = df.writeStream
+ .option("checkpointLocation", checkpointDir.getCanonicalPath)
+ .format("paimon")
+ .start(location)
+ try {
+ for (i <- 1 to 4) {
+ inputData.addData((i, s"v$i"))
+ query.processAllAvailable()
+ }
+ } finally {
+ query.stop()
+ }
+
+ assert(
+ snapshotCount("T") == 1,
+ s"expiration should retain a single snapshot, found ${snapshotCount("T")}")
+ }
+ }
}
}
From 587ad9c14ceb7909e6a335fd08ea930c6d2511a2 Mon Sep 17 00:00:00 2001
From: Xiangyi Zhu <82511136+zhuxiangyi@users.noreply.github.com>
Date: Sat, 12 Sep 2026 15:45:23 +0800
Subject: [PATCH 3/4] [core][spark] Recognise a replay on the direct postpone
committer too
Review found that an overwrite of a postpone bucket table with a default
bucket number, such as a micro-batch in complete mode, commits through the
direct fixed-bucket committer, which the replay handling did not reach: the
builder created its own random commit user, and even with a stable one the
committer runs in strict mode, whose lower bound on the lookup of the
previous commit is the snapshot the write started from, while the batch
being replayed was committed before it. A replay therefore committed a
second overwrite snapshot.
Give PostponeFixedBucketWriteBuilder a withCommitUser like the batch
builder has, and on the direct path look the replay up without the bound
before committing under the batch id. The strict mode bound stays what it
is for conflict detection; it was a lookup optimisation premised on a
commit user that never outlives its base snapshot, which a stable user does.
The test for this configuration also commits a later batch after the
replay, to make sure a higher batch id is not mistaken for a replay, and
the core test for inlineMaintenance commits through filterAndCommit with a
committer that is closed after each batch, the way the sink does.
---
docs/docs/spark/structured-streaming.md | 6 +-
.../sink/PostponeFixedBucketWriteBuilder.java | 12 +++-
.../paimon/table/SimpleTableTestBase.java | 51 ++++++++++++++
.../spark/commands/PaimonSparkWriter.scala | 40 ++++++++---
.../spark/PaimonSinkIdempotencyTest.scala | 66 +++++++++++++++++++
5 files changed, 163 insertions(+), 12 deletions(-)
diff --git a/docs/docs/spark/structured-streaming.md b/docs/docs/spark/structured-streaming.md
index e30ad081385f..b614a9797150 100644
--- a/docs/docs/spark/structured-streaming.md
+++ b/docs/docs/spark/structured-streaming.md
@@ -91,8 +91,10 @@ 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 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.
+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.
:::
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/sink/PostponeFixedBucketWriteBuilder.java b/paimon-core/src/main/java/org/apache/paimon/table/sink/PostponeFixedBucketWriteBuilder.java
index a3b28e47cc8d..750cb7defa3c 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/sink/PostponeFixedBucketWriteBuilder.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/sink/PostponeFixedBucketWriteBuilder.java
@@ -38,7 +38,8 @@ public class PostponeFixedBucketWriteBuilder implements BatchWriteBuilder {
private static final long serialVersionUID = 1L;
private final FileStoreTable table;
- private final String commitUser;
+
+ private String commitUser;
@Nullable private Map staticPartition;
@@ -50,6 +51,15 @@ 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;
+ return this;
+ }
+
@Override
public String tableName() {
return table.name();
diff --git a/paimon-core/src/test/java/org/apache/paimon/table/SimpleTableTestBase.java b/paimon-core/src/test/java/org/apache/paimon/table/SimpleTableTestBase.java
index f768753d7fe7..6b2a1fbdcd79 100644
--- a/paimon-core/src/test/java/org/apache/paimon/table/SimpleTableTestBase.java
+++ b/paimon-core/src/test/java/org/apache/paimon/table/SimpleTableTestBase.java
@@ -50,6 +50,7 @@
import org.apache.paimon.table.sink.BatchTableCommit;
import org.apache.paimon.table.sink.BatchTableWrite;
import org.apache.paimon.table.sink.BatchWriteBuilder;
+import org.apache.paimon.table.sink.BatchWriteBuilderImpl;
import org.apache.paimon.table.sink.CommitMessage;
import org.apache.paimon.table.sink.CommitMessageImpl;
import org.apache.paimon.table.sink.InnerTableCommit;
@@ -1594,6 +1595,56 @@ public void testBatchWriteAsyncExpireFallbackToSync() throws Exception {
}
}
+ @Test
+ public void testFilterAndCommitWithInlineMaintenance() throws Exception {
+ // async expire but retain only the last snapshot, like
+ // testBatchWriteAsyncExpireFallbackToSync
+ Map opts = new HashMap<>();
+ opts.put(SNAPSHOT_EXPIRE_EXECUTION_MODE.key(), ExpireExecutionMode.ASYNC.toString());
+ opts.put(SNAPSHOT_NUM_RETAINED_MIN.key(), "1");
+ opts.put(SNAPSHOT_NUM_RETAINED_MAX.key(), "1");
+ opts.put(SNAPSHOT_EXPIRE_LIMIT.key(), "100");
+
+ FileStoreTable table = createFileStoreTable(conf -> {});
+ table = table.copy(opts);
+ SnapshotManager sm = table.snapshotManager();
+
+ // A committer that commits once through filterAndCommit and is then closed, the way an
+ // engine which replays a batch with a stable identifier does. Without inline maintenance
+ // the expiration dispatched to the executor is cut short by the close.
+ BatchWriteBuilderImpl builder =
+ ((BatchWriteBuilderImpl) table.newBatchWriteBuilder()).withCommitUser("user");
+ long previous = 0;
+ for (long identifier = 0; identifier < 3; identifier++) {
+ try (BatchTableWrite write = builder.newWrite();
+ InnerTableCommit commit = builder.newCommit()) {
+ write.write(rowData((int) identifier, (int) identifier * 10, identifier * 100L));
+ commit.inlineMaintenance(true)
+ .filterAndCommit(
+ Collections.singletonMap(identifier, write.prepareCommit()));
+ }
+
+ long latest = sm.latestSnapshotId();
+ assertThat(latest).isGreaterThan(previous);
+ if (previous > 0) {
+ assertThat(sm.snapshotExists(previous))
+ .as("the previous snapshot should be expired before the committer closes")
+ .isFalse();
+ assertThat(sm.earliestSnapshotId()).isEqualTo(latest);
+ }
+ previous = latest;
+ }
+
+ // A replayed identifier is recognised and does not create a snapshot.
+ try (BatchTableWrite write = builder.newWrite();
+ InnerTableCommit commit = builder.newCommit()) {
+ write.write(rowData(2, 20, 200L));
+ commit.inlineMaintenance(true)
+ .filterAndCommit(Collections.singletonMap(2L, write.prepareCommit()));
+ }
+ assertThat(sm.latestSnapshotId()).isEqualTo(previous);
+ }
+
@Test
@Timeout(120)
public void testExpireWithLimit() throws Exception {
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala
index 61f6d86ec1dc..cf2eec2b2a00 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala
@@ -153,6 +153,7 @@ case class PaimonSparkWriter(
COMMIT_STRICT_MODE_LAST_SAFE_SNAPSHOT.key(),
postponeBaseSnapshotId.getOrElse(0L).toString)
val builder = table.copy(directWriteOptions).newPostponeFixedBucketWriteBuilder()
+ commitUser.foreach(builder.withCommitUser)
overwritePartitionSpec.foreach(spec => builder.withOverwrite(spec.asJava))
directPostponeWriteBuilder = builder
builder
@@ -473,6 +474,14 @@ case class PaimonSparkWriter(
_ <- commitUser
} yield identifier
+ /** Whether the stable commit user has already committed this identifier, or a later one. */
+ private def alreadyCommitted(identifier: Long): Boolean =
+ commitUser.exists {
+ user =>
+ val latest = table.snapshotManager().latestSnapshotOfUser(user)
+ latest.isPresent && latest.get.commitIdentifier() >= identifier
+ }
+
def commit(commitMessages: Seq[CommitMessage]): Unit = {
commit(commitMessages, null)
}
@@ -512,15 +521,28 @@ case class PaimonSparkWriter(
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)
- // This committer is closed right after the batch, so maintenance cannot be left to an
- // executor that is about to be shut down, nor a failure to a commit that never comes.
- .inlineMaintenance(true)
- .filterAndCommit(
- Collections.singletonMap(Long.box(identifier), commitMessages.toList.asJava))
+ // instead of duplicating the whole batch.
+ //
+ // This committer is closed right after the batch, so maintenance cannot be left to an
+ // executor that is about to be shut down, nor a failure to a commit that never comes.
+ tableCommit.inlineMaintenance(true)
+ if (directPostponeWriteBuilder != null) {
+ // The direct postpone committer runs in strict mode, and filterAndCommit bounds its
+ // lookup of the previous commit by the snapshot this write started from. The batch
+ // being replayed was committed before that snapshot, so look it up without the bound.
+ if (alreadyCommitted(identifier)) {
+ logInfo(s"Micro-batch $identifier is already committed, skipping the replay.")
+ } else {
+ tableCommit.commit(identifier, commitMessages.toList.asJava)
+ }
+ } else {
+ // 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)
}
diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/PaimonSinkIdempotencyTest.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/PaimonSinkIdempotencyTest.scala
index 4c10651ad4ea..e35649e7172f 100644
--- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/PaimonSinkIdempotencyTest.scala
+++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/PaimonSinkIdempotencyTest.scala
@@ -250,6 +250,72 @@ class PaimonSinkIdempotencyTest extends PaimonSparkTestBase with StreamTest {
}
}
+ test("Paimon Sink: complete mode replay on a postpone bucket table is not committed twice") {
+ failAfter(streamingTimeout) {
+ withTempDir {
+ checkpointDir =>
+ // A postpone bucket table with a default bucket number takes the direct fixed-bucket
+ // write path for an overwrite, which has its own committer.
+ spark.sql(
+ "CREATE TABLE T (city STRING, population LONG) TBLPROPERTIES (" +
+ "'primary-key' = 'city', 'bucket' = '-2', 'postpone.default-bucket-num' = '1')")
+ val location = loadTable("T").location().toString
+ val checkpointPath = checkpointDir.getCanonicalPath
+
+ val inputData = MemoryStream[(Int, String)]
+ val df = inputData
+ .toDS()
+ .toDF("uid", "city")
+ .groupBy("city")
+ .count()
+ .toDF("city", "population")
+ inputData.addData((1, "HZ"), (2, "BJ"), (3, "BJ"))
+
+ def start(): StreamingQuery =
+ df.writeStream
+ .outputMode("complete")
+ .option("checkpointLocation", checkpointPath)
+ .format("paimon")
+ .start(location)
+
+ runToCompletion(start())
+
+ val expected = Row("BJ", 2L) :: Row("HZ", 1L) :: Nil
+ checkAnswer(spark.sql("SELECT * FROM T ORDER BY city"), expected)
+ val snapshotsAfterFirstBatch = snapshotCount("T")
+ // The direct committer has to commit under the stable identity, or the replay below
+ // could not be recognised.
+ assert(
+ latestCommitUser("T").startsWith("spark-query-"),
+ s"expected the direct postpone committer to use the commit user derived from the " +
+ s"query id, but got '${latestCommitUser("T")}'"
+ )
+
+ dropCommitLogEntry(checkpointPath, 0)
+ runToCompletion(start())
+
+ checkAnswer(spark.sql("SELECT * FROM T ORDER BY city"), expected)
+ assert(
+ snapshotCount("T") == snapshotsAfterFirstBatch,
+ s"replaying batch 0 created another snapshot (${snapshotCount("T")} in total, " +
+ s"$snapshotsAfterFirstBatch before the replay)"
+ )
+
+ // A later batch is not a replay and has to be committed: the replay lookup must not
+ // mistake a higher batch id for one that was already committed.
+ inputData.addData((4, "SH"))
+ runToCompletion(start())
+
+ checkAnswer(
+ spark.sql("SELECT * FROM T ORDER BY city"),
+ Row("BJ", 2L) :: Row("HZ", 1L) :: Row("SH", 1L) :: Nil)
+ assert(
+ snapshotCount("T") == snapshotsAfterFirstBatch + 1,
+ s"the batch after the replay was not committed (${snapshotCount("T")} snapshots)")
+ }
+ }
+ }
+
test("Paimon Sink: write.stream.commit-user overrides the derived commit user") {
failAfter(streamingTimeout) {
withTempDir {
From e7a9976bb8c775706968648c330b5b0760c2b589 Mon Sep 17 00:00:00 2001
From: Xiangyi Zhu <82511136+zhuxiangyi@users.noreply.github.com>
Date: Sun, 13 Sep 2026 00:43:23 +0800
Subject: [PATCH 4/4] [core][spark] Retry commit callbacks for a replayed
micro-batch on the direct postpone path
Review found that the replay lookup the previous commit added on the direct
postpone path returned before reaching filterCommitted, which retries the
commit callbacks of a batch it recognises as already committed. A partition
registration that failed after the snapshot was published was therefore
never retried, and the checkpoint advanced with the partition absent from
the metastore.
The callbacks live in core, so the direct path commits through
filterAndCommit again. Its lookup was bounded by the strict mode safe
snapshot, which is what had forced the connector-side lookup: the bound is
an optimisation for a commit user created for one run, which cannot have
committed before its base snapshot, and a caller-provided user that
survives a restart can have. FileStoreCommit can now look the previous
commit up without that bound, and both write builders enable it when they
were given their commit user. Conflict detection keeps the bound, and so
does a committer created for an explicitly passed user, like the staged
postpone committer with its per-run user.
filterAndCommit also checked the append files of every committable against
the base files of the partitions it touches, a scan that guards a
committable restored from an engine's state and does nothing for one made
of files the batch has just written. InnerTableCommit can now turn that off
too, and the sink does; a batch commit never did the scan before this
series, and neither does the Flink committer in steady state.
Reproduced with a fail-once partition registration before the change, and
covered by that test, by a core test of the unbounded lookup for each
builder, including the overload that keeps the bound, and by the existing
suites.
---
.../paimon/operation/FileStoreCommit.java | 10 ++
.../paimon/operation/FileStoreCommitImpl.java | 9 +-
.../table/sink/BatchWriteBuilderImpl.java | 5 +-
.../paimon/table/sink/InnerTableCommit.java | 20 ++++
.../sink/PostponeFixedBucketWriteBuilder.java | 5 +-
.../paimon/table/sink/TableCommitImpl.java | 16 ++-
.../table/PrimaryKeySimpleTableTest.java | 46 ++++++++
.../paimon/table/SimpleTableTestBase.java | 45 ++++++++
.../spark/commands/PaimonSparkWriter.scala | 43 +++----
.../spark/PaimonSinkIdempotencyTest.scala | 108 +++++++++++++++++-
10 files changed, 273 insertions(+), 34 deletions(-)
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommit.java b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommit.java
index b039ffb9e9fc..3bc029fe05ad 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommit.java
+++ b/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(
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
index 2d9c94ec72fc..0ec10272f606 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
+++ b/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 filterCommitted(List commi
Optional optionalStrictSnapshot = options.commitStrictModeLastSafeSnapshot();
Optional latestSnapshot;
- if (optionalStrictSnapshot.isPresent()) {
+ if (optionalStrictSnapshot.isPresent() && !filterCommittedIgnoresStrictModeBound) {
latestSnapshot =
snapshotManager.latestSnapshotOfUser(
commitUser,
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/sink/BatchWriteBuilderImpl.java b/paimon-core/src/main/java/org/apache/paimon/table/sink/BatchWriteBuilderImpl.java
index 9f72e702a313..ce9002fa15f3 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/sink/BatchWriteBuilderImpl.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/sink/BatchWriteBuilderImpl.java
@@ -38,6 +38,7 @@ public class BatchWriteBuilderImpl implements BatchWriteBuilder {
private final InnerTable table;
private String commitUser;
+ private boolean commitUserProvided = false;
private Map staticPartition;
private @Nullable Long rowIdCheckFromSnapshot = null;
@@ -72,6 +73,7 @@ public Optional newWriteSelector() {
*/
public BatchWriteBuilderImpl withCommitUser(String commitUser) {
this.commitUser = commitUser;
+ this.commitUserProvided = true;
return this;
}
@@ -91,7 +93,8 @@ 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)
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/sink/InnerTableCommit.java b/paimon-core/src/main/java/org/apache/paimon/table/sink/InnerTableCommit.java
index 48ddb2cd4c42..d592f49fd44e 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/sink/InnerTableCommit.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/sink/InnerTableCommit.java
@@ -65,6 +65,18 @@ public interface InnerTableCommit extends StreamTableCommit, BatchTableCommit {
*/
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.
+ *
+ * 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
@@ -78,6 +90,14 @@ public interface InnerTableCommit extends StreamTableCommit, BatchTableCommit {
*/
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);
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/sink/PostponeFixedBucketWriteBuilder.java b/paimon-core/src/main/java/org/apache/paimon/table/sink/PostponeFixedBucketWriteBuilder.java
index 750cb7defa3c..44a514bef1df 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/sink/PostponeFixedBucketWriteBuilder.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/sink/PostponeFixedBucketWriteBuilder.java
@@ -40,6 +40,7 @@ public class PostponeFixedBucketWriteBuilder implements BatchWriteBuilder {
private final FileStoreTable table;
private String commitUser;
+ private boolean commitUserProvided = false;
@Nullable private Map staticPartition;
@@ -57,6 +58,7 @@ public PostponeFixedBucketWriteBuilder(FileStoreTable table) {
*/
public PostponeFixedBucketWriteBuilder withCommitUser(String commitUser) {
this.commitUser = commitUser;
+ this.commitUserProvided = true;
return this;
}
@@ -97,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) {
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/sink/TableCommitImpl.java b/paimon-core/src/main/java/org/apache/paimon/table/sink/TableCommitImpl.java
index 55f4336d92b1..826c11d3a137 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/sink/TableCommitImpl.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/sink/TableCommitImpl.java
@@ -95,6 +95,7 @@ public class TableCommitImpl implements InnerTableCommit {
private boolean batchCommitted = false;
private boolean expireForEmptyCommit = true;
private boolean checkFilesExistence = true;
+ private boolean checkAppendFiles = true;
private boolean inlineMaintenance = false;
public TableCommitImpl(
@@ -177,12 +178,24 @@ public TableCommitImpl checkFilesExistence(boolean 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);
@@ -279,7 +292,8 @@ public int filterAndCommit(Map> commitIdentifiersAndMe
return filterAndCommitMultiple(
commitIdentifiersAndMessages.entrySet().stream()
.map(e -> createManifestCommittable(e.getKey(), e.getValue()))
- .collect(Collectors.toList()));
+ .collect(Collectors.toList()),
+ checkAppendFiles);
}
private ManifestCommittable createManifestCommittable(
diff --git a/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeySimpleTableTest.java b/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeySimpleTableTest.java
index 6a7a722e855b..c97d112732ae 100644
--- a/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeySimpleTableTest.java
+++ b/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 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 =
diff --git a/paimon-core/src/test/java/org/apache/paimon/table/SimpleTableTestBase.java b/paimon-core/src/test/java/org/apache/paimon/table/SimpleTableTestBase.java
index 6b2a1fbdcd79..99fb4bf45875 100644
--- a/paimon-core/src/test/java/org/apache/paimon/table/SimpleTableTestBase.java
+++ b/paimon-core/src/test/java/org/apache/paimon/table/SimpleTableTestBase.java
@@ -108,6 +108,7 @@
import static org.apache.paimon.CoreOptions.BUCKET_KEY;
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.COMMIT_STRICT_MODE_LAST_SAFE_SNAPSHOT;
import static org.apache.paimon.CoreOptions.CONSUMER_IGNORE_PROGRESS;
import static org.apache.paimon.CoreOptions.DELETION_VECTORS_ENABLED;
import static org.apache.paimon.CoreOptions.ExpireExecutionMode;
@@ -1645,6 +1646,50 @@ public void testFilterAndCommitWithInlineMaintenance() throws Exception {
assertThat(sm.latestSnapshotId()).isEqualTo(previous);
}
+ @Test
+ public void testFilterAndCommitWithProvidedUserUnderStrictMode() throws Exception {
+ FileStoreTable table = createFileStoreTable(conf -> {});
+ SnapshotManager sm = table.snapshotManager();
+
+ // A first run commits identifier 0 under a caller-provided user.
+ BatchWriteBuilderImpl first =
+ ((BatchWriteBuilderImpl) table.newBatchWriteBuilder()).withCommitUser("user");
+ try (BatchTableWrite write = first.newWrite();
+ InnerTableCommit commit = first.newCommit()) {
+ write.write(rowData(1, 10, 100L));
+ commit.filterAndCommit(Collections.singletonMap(0L, write.prepareCommit()));
+ }
+ long committed = sm.latestSnapshotId();
+
+ // A restarted run replays identifier 0 with strict mode bounded by the snapshot it starts
+ // from, which is the snapshot that identifier produced. The bound only saves lookup work
+ // for a user created for one run; a provided user has to be looked up beyond it, or the
+ // replay is committed again.
+ Map strict = new HashMap<>();
+ strict.put(COMMIT_STRICT_MODE_LAST_SAFE_SNAPSHOT.key(), String.valueOf(committed));
+ BatchWriteBuilderImpl replay =
+ ((BatchWriteBuilderImpl) table.copy(strict).newBatchWriteBuilder())
+ .withCommitUser("user");
+ try (BatchTableWrite write = replay.newWrite();
+ InnerTableCommit commit = replay.newCommit()) {
+ write.write(rowData(1, 10, 100L));
+ 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);
+
+ // The bound stays in place for a user the builder created itself.
+ BatchWriteBuilderImpl generated =
+ (BatchWriteBuilderImpl) table.copy(strict).newBatchWriteBuilder();
+ try (BatchTableWrite write = generated.newWrite();
+ InnerTableCommit commit = generated.newCommit()) {
+ write.write(rowData(2, 20, 200L));
+ commit.filterAndCommit(Collections.singletonMap(0L, write.prepareCommit()));
+ }
+ assertThat(sm.latestSnapshotId()).isEqualTo(committed + 1);
+ }
+
@Test
@Timeout(120)
public void testExpireWithLimit() throws Exception {
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala
index cf2eec2b2a00..62a8466b3db9 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala
@@ -474,14 +474,6 @@ case class PaimonSparkWriter(
_ <- commitUser
} yield identifier
- /** Whether the stable commit user has already committed this identifier, or a later one. */
- private def alreadyCommitted(identifier: Long): Boolean =
- commitUser.exists {
- user =>
- val latest = table.snapshotManager().latestSnapshotOfUser(user)
- latest.isPresent && latest.get.commitIdentifier() >= identifier
- }
-
def commit(commitMessages: Seq[CommitMessage]): Unit = {
commit(commitMessages, null)
}
@@ -521,28 +513,21 @@ case class PaimonSparkWriter(
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.
+ // instead of duplicating the whole batch, while still retrying the commit callbacks
+ // that may have failed after the snapshot was published. Either builder was given the
+ // stable user, so its lookup of the previous commit is not bounded by strict mode.
//
- // This committer is closed right after the batch, so maintenance cannot be left to an
- // executor that is about to be shut down, nor a failure to a commit that never comes.
- tableCommit.inlineMaintenance(true)
- if (directPostponeWriteBuilder != null) {
- // The direct postpone committer runs in strict mode, and filterAndCommit bounds its
- // lookup of the previous commit by the snapshot this write started from. The batch
- // being replayed was committed before that snapshot, so look it up without the bound.
- if (alreadyCommitted(identifier)) {
- logInfo(s"Micro-batch $identifier is already committed, skipping the replay.")
- } else {
- tableCommit.commit(identifier, commitMessages.toList.asJava)
- }
- } else {
- // 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))
- }
+ // The files being committed were written by this very batch, so there is no need to
+ // list them to prove that they still exist, nor to scan the base files they could
+ // conflict with. And this committer is closed right after the batch, so maintenance
+ // cannot be left to an executor that is about to be shut down, nor a failure to a
+ // commit that never comes.
+ tableCommit
+ .checkFilesExistence(false)
+ .checkAppendFiles(false)
+ .inlineMaintenance(true)
+ .filterAndCommit(
+ Collections.singletonMap(Long.box(identifier), commitMessages.toList.asJava))
case None =>
tableCommit.commit(commitMessages.toList.asJava)
}
diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/PaimonSinkIdempotencyTest.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/PaimonSinkIdempotencyTest.scala
index e35649e7172f..ca35348efbda 100644
--- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/PaimonSinkIdempotencyTest.scala
+++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/PaimonSinkIdempotencyTest.scala
@@ -18,8 +18,10 @@
package org.apache.paimon.spark
+import org.apache.paimon.catalog.{Catalog, CatalogLoader, DelegateCatalog, Identifier}
import org.apache.paimon.options.Options
import org.apache.paimon.spark.sources.PaimonSink
+import org.apache.paimon.table.{CatalogEnvironment, FileStoreTableFactory}
import org.apache.spark.SparkConf
import org.apache.spark.sql.{DataFrame, Row}
@@ -27,7 +29,9 @@ import org.apache.spark.sql.paimon.shims.memstream.MemoryStream
import org.apache.spark.sql.streaming.{OutputMode, StreamingQuery, StreamTest}
import java.io.File
-import java.util.Collections
+import java.util.{Collections, List => JList, Map => JMap}
+
+import scala.collection.JavaConverters._
/**
* Structured Streaming guarantees exactly-once only if the sink is idempotent for a repeated
@@ -377,6 +381,67 @@ class PaimonSinkIdempotencyTest extends PaimonSparkTestBase with StreamTest {
}
}
+ test("Paimon Sink: a replay on the direct postpone path retries the partition registration") {
+ withTempDir {
+ checkpointDir =>
+ // A commit publishes its snapshot and only then registers the partition in the metastore.
+ // If that registration fails, the replay of the batch has to retry it: the snapshot
+ // already exists, so the replay must not commit again, but it must not report success
+ // before the partition is registered either.
+ spark.sql(
+ "CREATE TABLE T (city STRING, population LONG, dt STRING) PARTITIONED BY (dt) " +
+ "TBLPROPERTIES ('primary-key' = 'city,dt', 'bucket' = '-2', " +
+ "'postpone.default-bucket-num' = '1', 'metastore.partitioned-table' = 'true')")
+ val base = loadTable("T")
+ FailOnceRegistration.reset(paimonCatalog)
+ val environment = new CatalogEnvironment(
+ base.catalogEnvironment().identifier(),
+ base.catalogEnvironment().uuid(),
+ FailOnceRegistration.loader,
+ null,
+ null,
+ null,
+ false,
+ true)
+ val table = FileStoreTableFactory.create(
+ base.fileIO(),
+ base.location(),
+ base.schema(),
+ new Options(base.options()),
+ environment)
+
+ def newSink(): PaimonSink = new PaimonSink(
+ spark.sqlContext,
+ table,
+ Nil,
+ OutputMode.Complete(),
+ Options.fromMap(
+ Collections.singletonMap("checkpointLocation", checkpointDir.getCanonicalPath)))
+
+ val batch: DataFrame = Seq(("HZ", 1L, "2026-09-12")).toDF("city", "population", "dt")
+
+ // The first attempt fails after the snapshot is published.
+ val failure = intercept[Exception](newSink().addBatch(0L, batch))
+ assert(
+ failure.getMessage.contains("metastore unavailable") ||
+ Option(failure.getCause).exists(_.getMessage.contains("metastore unavailable")))
+ assert(snapshotCount("T") == 1)
+ assert(FailOnceRegistration.attempts == 1)
+ assert(FailOnceRegistration.registered.isEmpty)
+
+ // The restarted query replays the batch.
+ newSink().addBatch(0L, batch)
+
+ assert(snapshotCount("T") == 1, "the replay must not commit a second snapshot")
+ assert(
+ FailOnceRegistration.attempts == 2,
+ "the replay must retry the partition registration that failed after the snapshot")
+ assert(
+ FailOnceRegistration.registered.contains("2026-09-12"),
+ "the partition committed by the replayed batch must be registered")
+ }
+ }
+
test("Paimon Sink: addBatch with a repeated batchId must be a no-op") {
withTempDir {
checkpointDir =>
@@ -517,3 +582,44 @@ class PaimonSinkIdempotencyTest extends PaimonSparkTestBase with StreamTest {
}
}
}
+
+/** A catalog whose first partition registration fails, the way a metastore RPC can. */
+private[spark] object FailOnceRegistration {
+
+ @volatile private var wrapped: Catalog = _
+ @volatile var attempts: Int = 0
+ val registered: java.util.Set[String] =
+ Collections.synchronizedSet(new java.util.HashSet[String]())
+
+ def reset(catalog: Catalog): Unit = {
+ wrapped = catalog
+ attempts = 0
+ registered.clear()
+ }
+
+ // Refers to this object only, so that the loader stays serializable with the table.
+ val loader: CatalogLoader = () => new FailOnceCatalog(wrapped)
+
+ private class FailOnceCatalog(catalog: Catalog) extends DelegateCatalog(catalog) {
+
+ override def catalogLoader(): CatalogLoader = loader
+
+ override def createPartitions(
+ identifier: Identifier,
+ partitions: JList[JMap[String, String]]): Unit = {
+ attempts += 1
+ if (attempts == 1) {
+ throw new RuntimeException("metastore unavailable")
+ }
+ partitions.asScala.foreach(p => registered.add(p.get("dt")))
+ }
+
+ override def alterPartitions(
+ identifier: Identifier,
+ partitions: JList[org.apache.paimon.partition.PartitionStatistics]): Unit = {}
+
+ override def dropPartitions(
+ identifier: Identifier,
+ partitions: JList[JMap[String, String]]): Unit = {}
+ }
+}