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 d296f56d8e31..c6f3e668ca72 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 5e58ae7ab00fd5766660b42fc3ebdadf23e09d33 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/9] [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 08dcd002b8c8..1f955852287b 100644
--- a/docs/docs/spark/structured-streaming.md
+++ b/docs/docs/spark/structured-streaming.md
@@ -101,8 +101,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 c6f3e668ca72..ba1026ac9f2f 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_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 f641688f9a5499b6dba98bcb2a41ee6692e9dd55 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/9] [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..8a142fa0438c 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_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. Strict mode checks keep the bound either way.
+ */
+ FileStoreCommit filterCommittedIgnoresLastSafeSnapshot(boolean ignoresLastSafeSnapshot);
+
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 e867a1cc85a8..db83e26da8c6 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 filterCommittedIgnoresLastSafeSnapshot = false;
private CommitMetrics commitMetrics;
private boolean appendCommitCheckConflict = false;
private long lastCommittedSnapshotId = -1L;
@@ -250,6 +251,12 @@ public FileStoreCommit ignoreEmptyCommit(boolean ignoreEmptyCommit) {
return this;
}
+ @Override
+ public FileStoreCommit filterCommittedIgnoresLastSafeSnapshot(boolean ignoresLastSafeSnapshot) {
+ this.filterCommittedIgnoresLastSafeSnapshot = ignoresLastSafeSnapshot;
+ return this;
+ }
+
@Override
public FileStoreCommit withPartitionExpire(PartitionExpire partitionExpire) {
this.conflictDetection.withPartitionExpire(partitionExpire);
@@ -297,7 +304,7 @@ public List filterCommitted(List commi
Optional lastSafeSnapshot = options.commitLastSafeSnapshot();
Optional latestSnapshot;
- if (lastSafeSnapshot.isPresent()) {
+ if (lastSafeSnapshot.isPresent() && !filterCommittedIgnoresLastSafeSnapshot) {
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..53a27f02776c 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)
+ .filterCommittedIgnoresLastSafeSnapshot(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..ee96cff32062 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#filterCommittedIgnoresLastSafeSnapshot}. 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 filterCommittedIgnoresLastSafeSnapshot(boolean ignoresLastSafeSnapshot);
+
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..af2889e09831 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)
+ .filterCommittedIgnoresLastSafeSnapshot(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..48a4198d2155 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 filterCommittedIgnoresLastSafeSnapshot(boolean ignoresLastSafeSnapshot) {
+ commit.filterCommittedIgnoresLastSafeSnapshot(ignoresLastSafeSnapshot);
+ 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 140f549d50e8..31cc8ae3fae3 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_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 with the history 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_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 history 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..1a02cac4b3fe 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_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 testFilterAndCommitWithProvidedUserAcrossHistoryBound() 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 the history 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_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 history 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 ba1026ac9f2f..1d79f8fee7b3 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 = {}
+ }
+}
From ead1fe83fd75db8923ef084022b498b33d2edcac Mon Sep 17 00:00:00 2001
From: Xiangyi Zhu <82511136+zhuxiangyi@users.noreply.github.com>
Date: Sun, 13 Sep 2026 23:36:49 +0800
Subject: [PATCH 5/9] [core] Redo the chain table overwrite cleanup when a
replayed commit is retried
An overwrite of the delta branch of a chain table publishes its snapshot
and only then clears the same partitions of the snapshot branch, so that
reads fall through to the delta. ChainTableOverwriteCommitCallback left
its retry empty on purpose: a replayed batch used to be committed again,
and the cleanup ran with it. Now that a replay of an already published
batch is recognised and only retried, the cleanup that a callback failing
after the snapshot was written did not complete was never completed at
all, and the batch was acknowledged while chain reads still preferred the
old snapshot data.
Give the callback its commit user, the way IcebergCommitCallback has it,
and on retry resolve the overwrite snapshot of the identifier and redo the
cleanup from the manifest changes of that snapshot, which also cover
partitions the overwrite wrote no new file to. Truncating the partitions
again is idempotent. A snapshot that can no longer be found is logged
rather than failed on, since it has either expired or been superseded.
The test commits the overwrite through a table on which the callback does
nothing, which leaves exactly the state a failed callback leaves, then
replays it and expects the chain table to read the delta.
---
.../org/apache/paimon/AbstractFileStore.java | 2 +-
.../ChainTableOverwriteCommitCallback.java | 71 ++++++++++++++-----
.../table/ChainTableFileStoreTableTest.java | 54 ++++++++++++++
3 files changed, 110 insertions(+), 17 deletions(-)
diff --git a/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java b/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java
index c8719aae32de..97bf697f56c7 100644
--- a/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java
+++ b/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java
@@ -442,7 +442,7 @@ private List createCommitCallbacks(String commitUser, FileStoreT
}
if (options.isChainTable()) {
- callbacks.add(new ChainTableOverwriteCommitCallback(table));
+ callbacks.add(new ChainTableOverwriteCommitCallback(table, commitUser));
}
if (options.visibilityCallbackEnabled() && shouldWaitForVisibility(table)) {
diff --git a/paimon-core/src/main/java/org/apache/paimon/metastore/ChainTableOverwriteCommitCallback.java b/paimon-core/src/main/java/org/apache/paimon/metastore/ChainTableOverwriteCommitCallback.java
index 2d1a0cd7050f..1d2040fcd6ce 100644
--- a/paimon-core/src/main/java/org/apache/paimon/metastore/ChainTableOverwriteCommitCallback.java
+++ b/paimon-core/src/main/java/org/apache/paimon/metastore/ChainTableOverwriteCommitCallback.java
@@ -19,6 +19,7 @@
package org.apache.paimon.metastore;
import org.apache.paimon.CoreOptions;
+import org.apache.paimon.Snapshot;
import org.apache.paimon.Snapshot.CommitKind;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.manifest.ManifestCommittable;
@@ -26,9 +27,14 @@
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.sink.BatchTableCommit;
import org.apache.paimon.table.sink.CommitCallback;
+import org.apache.paimon.table.source.ScanMode;
import org.apache.paimon.utils.ChainTableUtils;
import org.apache.paimon.utils.InternalRowPartitionComputer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
@@ -50,52 +56,91 @@
*/
public class ChainTableOverwriteCommitCallback implements CommitCallback {
+ private static final Logger LOG =
+ LoggerFactory.getLogger(ChainTableOverwriteCommitCallback.class);
+
private transient FileStoreTable table;
private transient CoreOptions coreOptions;
+ private final String commitUser;
- public ChainTableOverwriteCommitCallback(FileStoreTable table) {
+ public ChainTableOverwriteCommitCallback(FileStoreTable table, String commitUser) {
this.table = table;
this.coreOptions = table.coreOptions();
+ this.commitUser = commitUser;
}
@Override
public void call(Context context) {
-
if (!ChainTableUtils.isScanFallbackDeltaBranch(coreOptions)) {
return;
}
-
if (context.snapshot.commitKind() != CommitKind.OVERWRITE) {
return;
}
+ truncateSnapshotPartitions(context.deltaFiles);
+ }
- FileStoreTable candidateTable = ChainTableUtils.resolveChainPrimaryTable(table);
+ /**
+ * The commit of this committable was published by an earlier attempt whose callback may not
+ * have completed, for example because the snapshot branch was unreachable right after the delta
+ * snapshot was written. Resolve that snapshot and redo the cleanup, which is idempotent. The
+ * partitions are taken from the manifest changes of the snapshot rather than from the
+ * committable, since an overwrite also clears partitions it wrote no new file to.
+ */
+ @Override
+ public void retry(ManifestCommittable committable) {
+ if (!ChainTableUtils.isScanFallbackDeltaBranch(coreOptions)) {
+ return;
+ }
+ List snapshots =
+ table.snapshotManager()
+ .findSnapshotsForIdentifiers(
+ commitUser, Collections.singletonList(committable.identifier()));
+ if (snapshots.isEmpty()) {
+ LOG.warn(
+ "No snapshot of commit user {} with identifier {} in table {}, "
+ + "cannot redo the snapshot branch cleanup of its overwrite.",
+ commitUser,
+ committable.identifier(),
+ table.name());
+ return;
+ }
+ for (Snapshot snapshot : snapshots) {
+ if (snapshot.commitKind() != CommitKind.OVERWRITE) {
+ continue;
+ }
+ truncateSnapshotPartitions(
+ table.store()
+ .newScan()
+ .withKind(ScanMode.DELTA)
+ .withSnapshot(snapshot.id())
+ .plan()
+ .files());
+ }
+ }
+ private void truncateSnapshotPartitions(List deltaFiles) {
+ FileStoreTable candidateTable = ChainTableUtils.resolveChainPrimaryTable(table);
FileStoreTable snapshotTable =
candidateTable.switchToBranch(coreOptions.scanFallbackSnapshotBranch());
-
InternalRowPartitionComputer partitionComputer =
new InternalRowPartitionComputer(
coreOptions.partitionDefaultName(),
table.schema().logicalPartitionType(),
table.schema().partitionKeys().toArray(new String[0]),
coreOptions.legacyPartitionName());
-
List overwritePartitions =
- context.deltaFiles.stream()
+ deltaFiles.stream()
.map(ManifestEntry::partition)
.distinct()
.collect(Collectors.toList());
-
if (overwritePartitions.isEmpty()) {
return;
}
-
List> candidatePartitions =
overwritePartitions.stream()
.map(partitionComputer::generatePartValues)
.collect(Collectors.toList());
-
try (BatchTableCommit commit = snapshotTable.newBatchWriteBuilder().newCommit()) {
commit.truncatePartitions(candidatePartitions);
} catch (Exception e) {
@@ -107,12 +152,6 @@ public void call(Context context) {
}
}
- @Override
- public void retry(ManifestCommittable committable) {
- // No-op. Truncating the same partitions again is safe, but we prefer to only rely on the
- // successful commit callback.
- }
-
@Override
public void close() throws Exception {
// no resources to close
diff --git a/paimon-core/src/test/java/org/apache/paimon/table/ChainTableFileStoreTableTest.java b/paimon-core/src/test/java/org/apache/paimon/table/ChainTableFileStoreTableTest.java
index b65b2e6ec0c8..f4b8b8a55d62 100644
--- a/paimon-core/src/test/java/org/apache/paimon/table/ChainTableFileStoreTableTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/table/ChainTableFileStoreTableTest.java
@@ -41,7 +41,10 @@
import org.apache.paimon.schema.SchemaChange;
import org.apache.paimon.schema.SchemaManager;
import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.table.sink.BatchTableWrite;
+import org.apache.paimon.table.sink.BatchWriteBuilderImpl;
import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.table.sink.InnerTableCommit;
import org.apache.paimon.table.sink.InnerTableWrite;
import org.apache.paimon.table.sink.StreamTableCommit;
import org.apache.paimon.table.source.ChainSplit;
@@ -70,6 +73,7 @@
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
@@ -657,6 +661,56 @@ private List readSplits(FileStoreTable table, List splits) th
return result;
}
+ @Test
+ public void testChainOverwriteReplayCompletesSnapshotCleanup() throws Exception {
+ createChainTable(options -> {});
+ FileStoreTable chainTable = loadTable();
+ FileStoreTable snapshotTable = chainTable.switchToBranch(SNAPSHOT_BRANCH);
+ FileStoreTable deltaTable = chainTable.switchToBranch(DELTA_BRANCH);
+ Map partition =
+ ImmutableMap.of("region", "CN", "dt", "20250810", "hour", "20");
+
+ writeWithCommit(snapshotTable, row(1L, 1L, "value-1", "CN", "20250810", "20"));
+ assertThat(getResult(loadTable(), partition))
+ .containsExactly(row(1L, 1L, "value-1", "CN", "20250810", "20"));
+
+ // An overwrite of the delta branch publishes its snapshot, and only then clears the same
+ // partition of the snapshot branch so that reads fall through to the delta. Commit it
+ // through a copy on which that callback does nothing: the state it leaves behind is the
+ // one a callback that failed after the snapshot was published leaves behind.
+ Map noCleanup = new HashMap<>();
+ noCleanup.put(CoreOptions.SCAN_FALLBACK_DELTA_BRANCH.key(), "none");
+ overwriteDelta(deltaTable.copy(noCleanup), partition, "value-2");
+ assertThat(getResult(loadTable(), partition))
+ .as("the snapshot branch still hides the delta until its partition is cleared")
+ .containsExactly(row(1L, 1L, "value-1", "CN", "20250810", "20"));
+
+ // A restarted job replays the batch under the same user and identifier. The commit is
+ // recognised as already published, so the callback is retried rather than called, and
+ // the retry has to complete the cleanup the first attempt did not.
+ overwriteDelta(deltaTable, partition, "value-2");
+
+ assertThat(deltaTable.snapshotManager().snapshotCount())
+ .as("the replay must not publish a second snapshot")
+ .isEqualTo(1);
+ assertThat(getResult(loadTable(), partition))
+ .as("the replay must complete the cleanup of the snapshot branch")
+ .containsExactly(row(1L, 2L, "value-2", "CN", "20250810", "20"));
+ }
+
+ private void overwriteDelta(FileStoreTable deltaTable, Map partition, String v)
+ throws Exception {
+ BatchWriteBuilderImpl builder =
+ ((BatchWriteBuilderImpl) deltaTable.newBatchWriteBuilder()).withCommitUser("user");
+ builder.withOverwrite(partition);
+ try (BatchTableWrite write =
+ builder.newWrite().withIOManager(new IOManagerImpl(tempDir.toString()));
+ InnerTableCommit commit = builder.newCommit()) {
+ write.write(row(1L, 2L, v, "CN", "20250810", "20"));
+ commit.filterAndCommit(Collections.singletonMap(0L, write.prepareCommit()));
+ }
+ }
+
private FileStoreTable loadTable() {
return loadTable(CatalogEnvironment.empty());
}
From ba632e29169136deedd573bcb03948a2c00b6f33 Mon Sep 17 00:00:00 2001
From: Xiangyi Zhu <82511136+zhuxiangyi@users.noreply.github.com>
Date: Sun, 13 Sep 2026 23:42:00 +0800
Subject: [PATCH 6/9] [core] Cover the chain table overwrite callback on its
normal and retried paths
The callback had no test of its own. Cover the normal path, the retry of an
overwrite whose cleanup failed after its snapshot was published, the same
for an overwrite that only removed files from a partition, which the commit
messages of the batch do not mention, and a retry for an identifier no
snapshot carries.
The failure is a real one: the delta table is told to clean a snapshot
branch that does not exist, so the commit publishes its snapshot and then
throws, which is the window a transient failure of the snapshot branch
opens.
---
.../table/ChainTableFileStoreTableTest.java | 134 ++++++++++++++++--
1 file changed, 119 insertions(+), 15 deletions(-)
diff --git a/paimon-core/src/test/java/org/apache/paimon/table/ChainTableFileStoreTableTest.java b/paimon-core/src/test/java/org/apache/paimon/table/ChainTableFileStoreTableTest.java
index f4b8b8a55d62..7d8e7236b6e1 100644
--- a/paimon-core/src/test/java/org/apache/paimon/table/ChainTableFileStoreTableTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/table/ChainTableFileStoreTableTest.java
@@ -662,25 +662,48 @@ private List readSplits(FileStoreTable table, List splits) th
}
@Test
- public void testChainOverwriteReplayCompletesSnapshotCleanup() throws Exception {
+ public void testChainOverwriteClearsSnapshotPartition() throws Exception {
createChainTable(options -> {});
FileStoreTable chainTable = loadTable();
FileStoreTable snapshotTable = chainTable.switchToBranch(SNAPSHOT_BRANCH);
FileStoreTable deltaTable = chainTable.switchToBranch(DELTA_BRANCH);
- Map partition =
- ImmutableMap.of("region", "CN", "dt", "20250810", "hour", "20");
+ Map partition = partition("20");
writeWithCommit(snapshotTable, row(1L, 1L, "value-1", "CN", "20250810", "20"));
assertThat(getResult(loadTable(), partition))
.containsExactly(row(1L, 1L, "value-1", "CN", "20250810", "20"));
- // An overwrite of the delta branch publishes its snapshot, and only then clears the same
- // partition of the snapshot branch so that reads fall through to the delta. Commit it
- // through a copy on which that callback does nothing: the state it leaves behind is the
- // one a callback that failed after the snapshot was published leaves behind.
- Map noCleanup = new HashMap<>();
- noCleanup.put(CoreOptions.SCAN_FALLBACK_DELTA_BRANCH.key(), "none");
- overwriteDelta(deltaTable.copy(noCleanup), partition, "value-2");
+ // An overwrite of the delta branch clears the same partition of the snapshot branch, so
+ // that reads fall through to the delta.
+ commitOverwrite(deltaTable, partition, 0, row(1L, 2L, "value-2", "CN", "20250810", "20"));
+ assertThat(getResult(loadTable(), partition))
+ .containsExactly(row(1L, 2L, "value-2", "CN", "20250810", "20"));
+ }
+
+ @Test
+ public void testChainOverwriteReplayCompletesSnapshotCleanup() throws Exception {
+ createChainTable(options -> {});
+ FileStoreTable chainTable = loadTable();
+ FileStoreTable snapshotTable = chainTable.switchToBranch(SNAPSHOT_BRANCH);
+ FileStoreTable deltaTable = chainTable.switchToBranch(DELTA_BRANCH);
+ Map partition = partition("20");
+
+ writeWithCommit(snapshotTable, row(1L, 1L, "value-1", "CN", "20250810", "20"));
+
+ // The overwrite publishes its snapshot, and only then clears the snapshot branch. Make
+ // that cleanup fail, the way a transient failure of the snapshot branch would: the delta
+ // snapshot is published, the commit throws, and the partition is not cleared.
+ assertThatThrownBy(
+ () ->
+ commitOverwrite(
+ withFailingCleanup(deltaTable),
+ partition,
+ 0,
+ row(1L, 2L, "value-2", "CN", "20250810", "20")))
+ .hasMessageContaining("does not exist");
+ assertThat(deltaTable.snapshotManager().snapshotCount())
+ .as("the delta snapshot was published before the cleanup failed")
+ .isEqualTo(1);
assertThat(getResult(loadTable(), partition))
.as("the snapshot branch still hides the delta until its partition is cleared")
.containsExactly(row(1L, 1L, "value-1", "CN", "20250810", "20"));
@@ -688,7 +711,7 @@ public void testChainOverwriteReplayCompletesSnapshotCleanup() throws Exception
// A restarted job replays the batch under the same user and identifier. The commit is
// recognised as already published, so the callback is retried rather than called, and
// the retry has to complete the cleanup the first attempt did not.
- overwriteDelta(deltaTable, partition, "value-2");
+ commitOverwrite(deltaTable, partition, 0, row(1L, 2L, "value-2", "CN", "20250810", "20"));
assertThat(deltaTable.snapshotManager().snapshotCount())
.as("the replay must not publish a second snapshot")
@@ -698,16 +721,97 @@ public void testChainOverwriteReplayCompletesSnapshotCleanup() throws Exception
.containsExactly(row(1L, 2L, "value-2", "CN", "20250810", "20"));
}
- private void overwriteDelta(FileStoreTable deltaTable, Map partition, String v)
+ @Test
+ public void testChainOverwriteReplayClearsPartitionWithOnlyRemovedFiles() throws Exception {
+ createChainTable(options -> {});
+ FileStoreTable chainTable = loadTable();
+ FileStoreTable snapshotTable = chainTable.switchToBranch(SNAPSHOT_BRANCH);
+ FileStoreTable deltaTable = chainTable.switchToBranch(DELTA_BRANCH);
+
+ writeWithCommit(
+ snapshotTable,
+ row(1L, 1L, "value-1", "CN", "20250810", "20"),
+ row(2L, 1L, "value-1", "CN", "20250810", "21"));
+ writeWithCommit(deltaTable, row(1L, 1L, "delta-old", "CN", "20250810", "20"));
+
+ // A static overwrite of the whole delta branch that writes only to hour 21 removes the
+ // file of hour 20 without adding one there. The cleanup has to cover hour 20 as well,
+ // which the commit messages of the batch do not mention; only the manifest changes of the
+ // snapshot do. (A dynamic partition overwrite, the default, would leave hour 20 alone.)
+ Map staticOverwrite = new HashMap<>();
+ staticOverwrite.put(CoreOptions.DYNAMIC_PARTITION_OVERWRITE.key(), "false");
+ FileStoreTable staticDelta = deltaTable.copy(staticOverwrite);
+ assertThatThrownBy(
+ () ->
+ commitOverwrite(
+ withFailingCleanup(staticDelta),
+ Collections.emptyMap(),
+ 0,
+ row(2L, 2L, "value-2", "CN", "20250810", "21")))
+ .hasMessageContaining("does not exist");
+ commitOverwrite(
+ staticDelta,
+ Collections.emptyMap(),
+ 0,
+ row(2L, 2L, "value-2", "CN", "20250810", "21"));
+
+ assertThat(deltaTable.snapshotManager().snapshotCount()).isEqualTo(2);
+ assertThat(getResult(loadTable(), partition("20")))
+ .as("a partition the overwrite only removed files from must be cleared too")
+ .isEmpty();
+ assertThat(getResult(loadTable(), partition("21")))
+ .containsExactly(row(2L, 2L, "value-2", "CN", "20250810", "21"));
+ }
+
+ @Test
+ public void testChainOverwriteReplayWithoutSnapshotIsSkipped() throws Exception {
+ createChainTable(options -> {});
+ FileStoreTable deltaTable = loadTable().switchToBranch(DELTA_BRANCH);
+ Map partition = partition("20");
+
+ commitOverwrite(deltaTable, partition, 0, row(1L, 1L, "value-1", "CN", "20250810", "20"));
+ commitOverwrite(deltaTable, partition, 2, row(1L, 2L, "value-2", "CN", "20250810", "20"));
+ long snapshots = deltaTable.snapshotManager().snapshotCount();
+
+ // Identifier 1 is below the latest committed one, so it is treated as a replay and
+ // retried, but no snapshot carries it. That is not an error: nothing is left to clean.
+ commitOverwrite(deltaTable, partition, 1, row(1L, 3L, "value-3", "CN", "20250810", "20"));
+
+ assertThat(deltaTable.snapshotManager().snapshotCount()).isEqualTo(snapshots);
+ assertThat(getResult(loadTable(), partition))
+ .containsExactly(row(1L, 2L, "value-2", "CN", "20250810", "20"));
+ }
+
+ private static Map partition(String hour) {
+ return ImmutableMap.of("region", "CN", "dt", "20250810", "hour", hour);
+ }
+
+ /**
+ * The delta table with the snapshot branch cleanup of an overwrite failing after the delta
+ * snapshot is published, since the branch it is told to clean does not exist.
+ */
+ private static FileStoreTable withFailingCleanup(FileStoreTable deltaTable) {
+ Map options = new HashMap<>();
+ options.put(CoreOptions.SCAN_FALLBACK_SNAPSHOT_BRANCH.key(), "unreachable");
+ return deltaTable.copy(options);
+ }
+
+ private void commitOverwrite(
+ FileStoreTable deltaTable,
+ Map overwrite,
+ long identifier,
+ GenericRow... rows)
throws Exception {
BatchWriteBuilderImpl builder =
((BatchWriteBuilderImpl) deltaTable.newBatchWriteBuilder()).withCommitUser("user");
- builder.withOverwrite(partition);
+ builder.withOverwrite(overwrite);
try (BatchTableWrite write =
builder.newWrite().withIOManager(new IOManagerImpl(tempDir.toString()));
InnerTableCommit commit = builder.newCommit()) {
- write.write(row(1L, 2L, v, "CN", "20250810", "20"));
- commit.filterAndCommit(Collections.singletonMap(0L, write.prepareCommit()));
+ for (GenericRow r : rows) {
+ write.write(r);
+ }
+ commit.filterAndCommit(Collections.singletonMap(identifier, write.prepareCommit()));
}
}
From 848c78b25db7086e280a08573ce8c027e4bba8f3 Mon Sep 17 00:00:00 2001
From: Xiangyi Zhu <82511136+zhuxiangyi@users.noreply.github.com>
Date: Mon, 14 Sep 2026 12:19:06 +0800
Subject: [PATCH 7/9] [spark] Commit a micro-batch under an identifier counted
from 1
Paimon numbers the commits of a stream from 1, the way Flink numbers its
checkpoints, and FullCompactedStartingScanner recognises a full compaction
by an identifier that is a multiple of full-compaction.delta-commits. The
sink scheduled its full compactions by (batchId + 1) % deltaCommits, which
encodes exactly that, but published the zero-based batch id itself, so a
compacted-full scan rejected every scheduled full compaction and fell back
to a later snapshot.
Define the identifier once, as the batch id plus one, and use it both to
publish the commit and to schedule the full compaction, which now shares
the scanner's rule instead of carrying its own. The schedule is unchanged.
The test writes four micro-batches with a delta count of three, checks the
identifiers of the full compaction and of the batch after it, and expects
a compacted-full scan to read the state of the full compaction; a
zero-based identifier fails it and the existing schedule test alike.
---
docs/docs/spark/structured-streaming.md | 5 +-
.../spark/commands/PaimonSparkWriter.scala | 13 ++++-
.../apache/paimon/spark/write/DataWrite.scala | 12 ++--
.../paimon/spark/write/PaimonDataWrite.scala | 2 +-
.../spark/write/PaimonV2DataWriter.scala | 2 +-
.../apache/paimon/spark/PaimonSinkTest.scala | 55 +++++++++++++++++++
6 files changed, 80 insertions(+), 9 deletions(-)
diff --git a/docs/docs/spark/structured-streaming.md b/docs/docs/spark/structured-streaming.md
index 1f955852287b..c16a9967b0f6 100644
--- a/docs/docs/spark/structured-streaming.md
+++ b/docs/docs/spark/structured-streaming.md
@@ -71,7 +71,10 @@ Streaming writes also support [Schema Evolution on Write](./schema-evolution).
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.
+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
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 1d79f8fee7b3..c3742250b965 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
@@ -63,6 +63,15 @@ case class PaimonSparkWriter(
extends WriteHelper
with Logging {
+ /**
+ * Paimon numbers the commits of a stream from 1, the way Flink numbers its checkpoints, and
+ * [[org.apache.paimon.table.source.snapshot.FullCompactedStartingScanner]] recognises a full
+ * compaction by an identifier that is a multiple of 'full-compaction.delta-commits'. Spark
+ * numbers its micro-batches from 0, so the third one is batch 2; both the full compaction
+ * schedule and the identifier it is published under have to count it as 3.
+ */
+ private val commitIdentifier: Option[Long] = batchId.map(_ + 1)
+
private lazy val tableSchema = table.schema
private lazy val bucketMode = table.bucketMode
@@ -183,7 +192,7 @@ case class PaimonSparkWriter(
rowKindColIdx,
writeRowTracking,
fullCompactionDeltaCommits,
- batchId,
+ commitIdentifier,
uriReaderFactory,
postponePartitionBucketComputer
)
@@ -470,7 +479,7 @@ case class PaimonSparkWriter(
*/
private def idempotentCommitIdentifier: Option[Long] =
for {
- identifier <- batchId
+ identifier <- commitIdentifier
_ <- commitUser
} yield identifier
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/DataWrite.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/DataWrite.scala
index 603ec7ecff1c..907905b0891e 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/DataWrite.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/DataWrite.scala
@@ -114,14 +114,18 @@ abstract class abstractInnerTableDataWrite[T] extends InnerTableDataWrite[T] wit
val fullCompactionDeltaCommits: Option[Int]
- /** For batch write, batchId is None, for streaming write, batchId is the current batch id (>= 0). */
- val batchId: Option[Long]
+ /**
+ * None for a batch write. For a streaming write, the identifier the micro-batch is committed
+ * under, which counts from 1.
+ */
+ val commitIdentifier: Option[Long]
private lazy val needFullCompaction: Boolean = {
fullCompactionDeltaCommits match {
case Some(deltaCommits) if deltaCommits > 0 =>
- batchId match {
- case Some(id) => (id + 1) % deltaCommits == 0
+ commitIdentifier match {
+ // The same rule by which a compacted-full scan recognises the full compaction.
+ case Some(identifier) => identifier % deltaCommits == 0
// When fullCompactionDeltaCommits is set, always trigger full compaction for batch write.
case None => true
}
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonDataWrite.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonDataWrite.scala
index af20144f5233..eda41f008400 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonDataWrite.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonDataWrite.scala
@@ -36,7 +36,7 @@ case class PaimonDataWrite(
rowKindColIdx: Int = -1,
writeRowTracking: Boolean = false,
fullCompactionDeltaCommits: Option[Int],
- batchId: Option[Long],
+ commitIdentifier: Option[Long],
uriReaderFactory: UriReaderFactory,
postponePartitionBucketComputer: Option[BinaryRow => Integer])
extends abstractInnerTableDataWrite[Row]
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonV2DataWriter.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonV2DataWriter.scala
index 05b6bfd307c4..5db5843173ab 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonV2DataWriter.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonV2DataWriter.scala
@@ -38,7 +38,7 @@ case class PaimonV2DataWriter(
dataSchema: StructType,
coreOptions: CoreOptions,
uriReaderFactory: UriReaderFactory,
- batchId: Option[Long] = None,
+ commitIdentifier: Option[Long] = None,
paimonWriteType: Option[RowType] = None,
metadataSchema: Option[StructType] = None,
plainWriteSchema: Option[StructType] = None)
diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/PaimonSinkTest.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/PaimonSinkTest.scala
index f22dab4b9cd0..bbbfa7cc7081 100644
--- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/PaimonSinkTest.scala
+++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/PaimonSinkTest.scala
@@ -364,6 +364,61 @@ class PaimonSinkTest extends PaimonSparkTestBase with StreamTest {
}
}
+ test("Paimon Sink: full compaction of a micro-batch is recognised by a compacted-full scan") {
+ failAfter(streamingTimeout) {
+ withTempDir {
+ checkpointDir =>
+ spark.sql(s"""
+ |CREATE TABLE T (a INT, b INT)
+ |TBLPROPERTIES (
+ | 'primary-key'='a',
+ | 'bucket'='1',
+ | 'full-compaction.delta-commits'='3'
+ |)
+ |""".stripMargin)
+ val table = loadTable("T")
+ val location = table.location().toString
+
+ val inputData = MemoryStream[(Int, Int)]
+ val stream = inputData
+ .toDS()
+ .toDF("a", "b")
+ .writeStream
+ .option("checkpointLocation", checkpointDir.getCanonicalPath)
+ .format("paimon")
+ .start(location)
+
+ try {
+ // The third micro-batch is scheduled for a full compaction, the fourth is not.
+ for (b <- 100 to 103) {
+ inputData.addData((1, b))
+ stream.processAllAvailable()
+ }
+ } finally {
+ stream.stop()
+ }
+ // Micro-batch n is committed under identifier n + 1, the way Flink numbers checkpoints;
+ // that is what a compacted-full scan recognises a scheduled full compaction by.
+ val snapshots = table.snapshotManager()
+ assert(
+ snapshots.latestSnapshot().commitKind == APPEND,
+ "the last micro-batch must not have compacted")
+ assert(snapshots.latestSnapshot().commitIdentifier() == 4)
+ val fullCompaction = snapshots.snapshot(snapshots.latestSnapshotId() - 1)
+ assert(fullCompaction.commitKind == COMPACT)
+ assert(fullCompaction.commitIdentifier() == 3)
+
+ // A compacted-full scan reads the latest full compaction, that is the state after the
+ // third micro-batch, and must recognise it by the commit identifier the sink published.
+ val compactedFull = spark.read
+ .format("paimon")
+ .option("scan.mode", "compacted-full")
+ .load(location)
+ checkAnswer(compactedFull, Row(1, 102) :: Nil)
+ }
+ }
+ }
+
test("Paimon Sink: batch then stream should not overwrite batch data") {
failAfter(streamingTimeout) {
withTempDir {
From 92407b7af99311f7e44ac4bfd70ddda3b943edf2 Mon Sep 17 00:00:00 2001
From: Xiangyi Zhu <82511136+zhuxiangyi@users.noreply.github.com>
Date: Thu, 17 Sep 2026 19:00:21 +0800
Subject: [PATCH 8/9] [core] Clear only what a replayed chain overwrite
superseded
The retry of the chain table overwrite cleanup truncated the partitions of
the snapshot branch outright. A replay of a batch does not mean its
callback never ran: the engine may have failed after the sink returned,
with the cleanup long done. Data written to those partitions of the
snapshot branch since then, which takes precedence over the delta, was
removed by the replay, and reads reverted to the delta.
Clear on retry only what the callback cleared, or would have cleared, when
the overwrite was published: the files the snapshot branch held in those
partitions at the commit time of the overwrite snapshot, intersected with
the files it holds now. Files that landed since are kept, and files cleared
since, by the callback or an earlier retry, are not touched again, which
makes the retry idempotent. The files are removed by a commit of their own
that checks they are still present, so a concurrent change surfaces as a
conflict instead of being skipped over. The cleanup at commit time is
unchanged.
Two tests pin the two halves: data written after a completed cleanup
survives the replay, and after a failed cleanup the replay removes the
superseded file and keeps the later one. Replacing the as-of snapshot by
the latest, or the intersection by a truncation, fails them.
---
.../ChainTableOverwriteCommitCallback.java | 139 +++++++++++++++---
.../table/ChainTableFileStoreTableTest.java | 67 +++++++++
2 files changed, 188 insertions(+), 18 deletions(-)
diff --git a/paimon-core/src/main/java/org/apache/paimon/metastore/ChainTableOverwriteCommitCallback.java b/paimon-core/src/main/java/org/apache/paimon/metastore/ChainTableOverwriteCommitCallback.java
index 1d2040fcd6ce..ae123cd9afb2 100644
--- a/paimon-core/src/main/java/org/apache/paimon/metastore/ChainTableOverwriteCommitCallback.java
+++ b/paimon-core/src/main/java/org/apache/paimon/metastore/ChainTableOverwriteCommitCallback.java
@@ -22,23 +22,38 @@
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.
@@ -109,38 +124,126 @@ public void retry(ManifestCommittable committable) {
if (snapshot.commitKind() != CommitKind.OVERWRITE) {
continue;
}
- truncateSnapshotPartitions(
- table.store()
- .newScan()
- .withKind(ScanMode.DELTA)
- .withSnapshot(snapshot.id())
- .plan()
- .files());
+ List overwritePartitions =
+ overwritePartitions(
+ table.store()
+ .newScan()
+ .withKind(ScanMode.DELTA)
+ .withSnapshot(snapshot.id())
+ .plan()
+ .files());
+ clearSnapshotFilesAsOf(overwritePartitions, snapshot.timeMillis());
}
}
- private void truncateSnapshotPartitions(List deltaFiles) {
+ /**
+ * 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 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 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, List> 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 filesOf(
+ FileStoreTable table, Snapshot snapshot, List partitions) {
+ return table.store()
+ .newScan()
+ .withSnapshot(snapshot)
+ .withPartitionFilter(partitions)
+ .plan()
+ .files();
+ }
+
+ private FileStoreTable snapshotTable() {
FileStoreTable candidateTable = ChainTableUtils.resolveChainPrimaryTable(table);
- FileStoreTable snapshotTable =
- candidateTable.switchToBranch(coreOptions.scanFallbackSnapshotBranch());
+ return candidateTable.switchToBranch(coreOptions.scanFallbackSnapshotBranch());
+ }
+
+ private static List overwritePartitions(List deltaFiles) {
+ return deltaFiles.stream()
+ .map(ManifestEntry::partition)
+ .distinct()
+ .collect(Collectors.toList());
+ }
+
+ private void truncateSnapshotPartitions(List deltaFiles) {
+ List 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 overwritePartitions =
- deltaFiles.stream()
- .map(ManifestEntry::partition)
- .distinct()
- .collect(Collectors.toList());
- if (overwritePartitions.isEmpty()) {
- return;
- }
List> candidatePartitions =
overwritePartitions.stream()
.map(partitionComputer::generatePartValues)
.collect(Collectors.toList());
+ FileStoreTable snapshotTable = snapshotTable();
try (BatchTableCommit commit = snapshotTable.newBatchWriteBuilder().newCommit()) {
commit.truncatePartitions(candidatePartitions);
} catch (Exception e) {
diff --git a/paimon-core/src/test/java/org/apache/paimon/table/ChainTableFileStoreTableTest.java b/paimon-core/src/test/java/org/apache/paimon/table/ChainTableFileStoreTableTest.java
index 7d8e7236b6e1..1545df92152d 100644
--- a/paimon-core/src/test/java/org/apache/paimon/table/ChainTableFileStoreTableTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/table/ChainTableFileStoreTableTest.java
@@ -763,6 +763,73 @@ public void testChainOverwriteReplayClearsPartitionWithOnlyRemovedFiles() throws
.containsExactly(row(2L, 2L, "value-2", "CN", "20250810", "21"));
}
+ @Test
+ public void testChainOverwriteReplayKeepsSnapshotDataWrittenAfterTheOverwrite()
+ throws Exception {
+ createChainTable(options -> {});
+ FileStoreTable chainTable = loadTable();
+ FileStoreTable snapshotTable = chainTable.switchToBranch(SNAPSHOT_BRANCH);
+ FileStoreTable deltaTable = chainTable.switchToBranch(DELTA_BRANCH);
+ Map partition = partition("20");
+
+ writeWithCommit(snapshotTable, row(1L, 1L, "value-1", "CN", "20250810", "20"));
+ // The overwrite and its cleanup both succeed.
+ commitOverwrite(deltaTable, partition, 0, row(1L, 2L, "value-2", "CN", "20250810", "20"));
+ assertThat(getResult(loadTable(), partition))
+ .containsExactly(row(1L, 2L, "value-2", "CN", "20250810", "20"));
+
+ // New data lands in the same partition of the snapshot branch afterwards and, being the
+ // snapshot, takes precedence over the delta.
+ writeWithCommit(snapshotTable, row(2L, 1L, "value-3", "CN", "20250810", "20"));
+ assertThat(getResult(loadTable(), partition))
+ .containsExactly(row(2L, 1L, "value-3", "CN", "20250810", "20"));
+
+ // The batch is replayed although its callback had completed, for example because the
+ // engine failed after the sink returned. The retry must not clear what landed later.
+ commitOverwrite(deltaTable, partition, 0, row(1L, 2L, "value-2", "CN", "20250810", "20"));
+
+ assertThat(getResult(snapshotTable, partition))
+ .as(
+ "data written to the snapshot branch after the overwrite must survive its replay")
+ .containsExactly(row(2L, 1L, "value-3", "CN", "20250810", "20"));
+ assertThat(getResult(loadTable(), partition))
+ .containsExactly(row(2L, 1L, "value-3", "CN", "20250810", "20"));
+ }
+
+ @Test
+ public void testChainOverwriteReplayClearsOnlyWhatTheOverwriteSuperseded() throws Exception {
+ createChainTable(options -> {});
+ FileStoreTable chainTable = loadTable();
+ FileStoreTable snapshotTable = chainTable.switchToBranch(SNAPSHOT_BRANCH);
+ FileStoreTable deltaTable = chainTable.switchToBranch(DELTA_BRANCH);
+ Map partition = partition("20");
+
+ writeWithCommit(snapshotTable, row(1L, 1L, "value-1", "CN", "20250810", "20"));
+ // The overwrite is published, its cleanup fails.
+ assertThatThrownBy(
+ () ->
+ commitOverwrite(
+ withFailingCleanup(deltaTable),
+ partition,
+ 0,
+ row(1L, 2L, "value-2", "CN", "20250810", "20")))
+ .hasMessageContaining("does not exist");
+
+ // New data lands in the same partition of the snapshot branch before the replay.
+ writeWithCommit(snapshotTable, row(2L, 1L, "value-3", "CN", "20250810", "20"));
+
+ // Had the cleanup run when the overwrite was published, it would have cleared value-1
+ // and left value-3, which came later, in place. The retry has to do the same.
+ commitOverwrite(deltaTable, partition, 0, row(1L, 2L, "value-2", "CN", "20250810", "20"));
+
+ assertThat(getResult(snapshotTable, partition))
+ .as(
+ "the retry must clear what the overwrite superseded and nothing that came later")
+ .containsExactly(row(2L, 1L, "value-3", "CN", "20250810", "20"));
+ assertThat(getResult(loadTable(), partition))
+ .containsExactly(row(2L, 1L, "value-3", "CN", "20250810", "20"));
+ }
+
@Test
public void testChainOverwriteReplayWithoutSnapshotIsSkipped() throws Exception {
createChainTable(options -> {});
From e349098489d819348d74eb94613db63b637c4633 Mon Sep 17 00:00:00 2001
From: Xiangyi Zhu <82511136+zhuxiangyi@users.noreply.github.com>
Date: Sat, 19 Sep 2026 11:11:51 +0800
Subject: [PATCH 9/9] [spark] Take the streaming commit user only from sources
scoped to the query
The commit user was also read from the options of the table, so that a
spark.paimon session conf would reach the sink. Those options include the
table properties, and a commit user set there was shared by every streaming
writer of the table. A commit user is the identity of one writer: with a
shared one, the batches of a second query carry identifiers the first has
already published, and filterCommitted drops them as replays, silently, so
the second query writes nothing and reports success.
Read the option from the writer and from the session conf directly, and
ignore it in the table properties, with a warning. The option description
and the guide now say the user has to be unique to one streaming writer. A
micro-batch skipped as a replay is logged, which is the only trace of a
misconfigured identity.
The test starts a second query on a table that carries the property and
expects its data and its own derived identity; it fails when the table
options are read. Another checks that an option of the writer takes
precedence over the session conf.
---
docs/docs/spark/structured-streaming.md | 4 +-
.../spark_connector_configuration.html | 2 +-
.../paimon/spark/SparkConnectorOptions.java | 6 +-
.../spark/commands/PaimonSparkWriter.scala | 8 +-
.../paimon/spark/sources/PaimonSink.scala | 23 ++++--
.../paimon/spark/util/OptionUtils.scala | 2 +-
.../spark/PaimonSinkIdempotencyTest.scala | 76 +++++++++++++++++++
7 files changed, 111 insertions(+), 10 deletions(-)
diff --git a/docs/docs/spark/structured-streaming.md b/docs/docs/spark/structured-streaming.md
index c16a9967b0f6..f908371cb110 100644
--- a/docs/docs/spark/structured-streaming.md
+++ b/docs/docs/spark/structured-streaming.md
@@ -83,7 +83,9 @@ id that Spark persists in the checkpoint, which is new when a checkpoint is recr
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:
+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.
```scala
val stream = df
diff --git a/docs/generated/spark_connector_configuration.html b/docs/generated/spark_connector_configuration.html
index ca92a5331238..d36ab2ee3f45 100644
--- a/docs/generated/spark_connector_configuration.html
+++ b/docs/generated/spark_connector_configuration.html
@@ -144,7 +144,7 @@
write.stream.commit-user |
(none) |
String |
- 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. |
+ 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. |