staticPartition) {
this.staticPartition = staticPartition;
@@ -73,11 +89,12 @@ public BatchTableWrite newWrite() {
}
@Override
- public BatchTableCommit newCommit() {
+ public InnerTableCommit newCommit() {
InnerTableCommit commit =
table.newCommit(commitUser)
.withOverwrite(staticPartition)
- .rowIdCheckConflict(rowIdCheckFromSnapshot);
+ .rowIdCheckConflict(rowIdCheckFromSnapshot)
+ .filterCommittedIgnoresStrictModeBound(commitUserProvided);
commit.ignoreEmptyCommit(
Options.fromMap(table.options())
.getOptional(CoreOptions.SNAPSHOT_IGNORE_EMPTY_COMMIT)
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 43f98d0e7933..d592f49fd44e 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/sink/InnerTableCommit.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/sink/InnerTableCommit.java
@@ -54,6 +54,50 @@ public interface InnerTableCommit extends StreamTableCommit, BatchTableCommit {
InnerTableCommit expireForEmptyCommit(boolean expireForEmptyCommit);
+ /**
+ * If this is set to true, {@link StreamTableCommit#filterAndCommit} verifies that every file it
+ * is about to commit still exists. By default it does.
+ *
+ * The check guards a committable that was restored from an engine's state and may reference
+ * files deleted long ago. A caller which filters a committable it has just produced itself
+ * knows those files exist, and can skip a file listing proportional to the size of the
+ * committable.
+ */
+ InnerTableCommit checkFilesExistence(boolean checkFilesExistence);
+
+ /**
+ * Whether {@link StreamTableCommit#filterAndCommit} checks the append files of a committable
+ * against the files of the latest snapshot before committing them. By default it does.
+ *
+ *
The check guards a committable restored from an engine's state, whose files may have been
+ * committed, or removed, by an attempt the engine did not see complete. A caller filtering a
+ * committable it has just produced knows its files are new, and can skip a scan of the base
+ * files of every partition the committable touches. {@link #appendCommitCheckConflict} still
+ * forces the check regardless of this setting.
+ */
+ InnerTableCommit checkAppendFiles(boolean checkAppendFiles);
+
+ /**
+ * If this is set to true, maintenance runs on the committing thread and its failure is thrown
+ * to the caller, instead of running through an executor which stores the failure for the next
+ * commit to report.
+ *
+ *
A committer which commits once and is then closed has to do this: it is about to shut the
+ * executor down, so maintenance dispatched to it may never run, and there is no next commit to
+ * report a failure to. {@link BatchTableCommit#commit(List)} already behaves this way; a caller
+ * which commits through {@link StreamTableCommit#filterAndCommit} with the same one-shot
+ * lifecycle has to ask for it.
+ */
+ InnerTableCommit inlineMaintenance(boolean inlineMaintenance);
+
+ /**
+ * See {@link
+ * org.apache.paimon.operation.FileStoreCommit#filterCommittedIgnoresStrictModeBound}. A write
+ * builder enables this when it was given its commit user, since such a user can have committed
+ * before the base snapshot of the current write.
+ */
+ InnerTableCommit filterCommittedIgnoresStrictModeBound(boolean ignoresStrictModeBound);
+
InnerTableCommit appendCommitCheckConflict(boolean appendCommitCheckConflict);
InnerTableCommit rowIdCheckConflict(@Nullable Long rowIdCheckFromSnapshot);
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..44a514bef1df 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/sink/PostponeFixedBucketWriteBuilder.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/sink/PostponeFixedBucketWriteBuilder.java
@@ -38,7 +38,9 @@ public class PostponeFixedBucketWriteBuilder implements BatchWriteBuilder {
private static final long serialVersionUID = 1L;
private final FileStoreTable table;
- private final String commitUser;
+
+ private String commitUser;
+ private boolean commitUserProvided = false;
@Nullable private Map staticPartition;
@@ -50,6 +52,16 @@ public PostponeFixedBucketWriteBuilder(FileStoreTable table) {
this.commitUser = createCommitUser(new Options(table.options()));
}
+ /**
+ * Use a caller-provided commit user instead of the random one, for the same reason as {@link
+ * BatchWriteBuilderImpl#withCommitUser}.
+ */
+ public PostponeFixedBucketWriteBuilder withCommitUser(String commitUser) {
+ this.commitUser = commitUser;
+ this.commitUserProvided = true;
+ return this;
+ }
+
@Override
public String tableName() {
return table.name();
@@ -87,7 +99,8 @@ public TableCommitImpl newCommit() {
Options.fromMap(table.options())
.getOptional(CoreOptions.SNAPSHOT_IGNORE_EMPTY_COMMIT)
.orElse(true);
- return newCommit(commitUser, ignoreEmpty);
+ return newCommit(commitUser, ignoreEmpty)
+ .filterCommittedIgnoresStrictModeBound(commitUserProvided);
}
public TableCommitImpl newCommit(String commitUser, boolean ignoreEmptyCommit) {
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 014b5e64daa1..826c11d3a137 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/sink/TableCommitImpl.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/sink/TableCommitImpl.java
@@ -94,6 +94,9 @@ public class TableCommitImpl implements InnerTableCommit {
@Nullable private List overwriteStaticPartitions = null;
private boolean batchCommitted = false;
private boolean expireForEmptyCommit = true;
+ private boolean checkFilesExistence = true;
+ private boolean checkAppendFiles = true;
+ private boolean inlineMaintenance = false;
public TableCommitImpl(
FileStoreCommit commit,
@@ -169,6 +172,30 @@ public TableCommitImpl expireForEmptyCommit(boolean expireForEmptyCommit) {
return this;
}
+ @Override
+ public TableCommitImpl checkFilesExistence(boolean checkFilesExistence) {
+ this.checkFilesExistence = checkFilesExistence;
+ return this;
+ }
+
+ @Override
+ public TableCommitImpl checkAppendFiles(boolean checkAppendFiles) {
+ this.checkAppendFiles = checkAppendFiles;
+ return this;
+ }
+
+ @Override
+ public TableCommitImpl inlineMaintenance(boolean inlineMaintenance) {
+ this.inlineMaintenance = inlineMaintenance;
+ return this;
+ }
+
+ @Override
+ public TableCommitImpl filterCommittedIgnoresStrictModeBound(boolean ignoresStrictModeBound) {
+ commit.filterCommittedIgnoresStrictModeBound(ignoresStrictModeBound);
+ return this;
+ }
+
@Override
public TableCommitImpl appendCommitCheckConflict(boolean appendCommitCheckConflict) {
commit.appendCommitCheckConflict(appendCommitCheckConflict);
@@ -265,7 +292,8 @@ public int filterAndCommit(Map> commitIdentifiersAndMe
return filterAndCommitMultiple(
commitIdentifiersAndMessages.entrySet().stream()
.map(e -> createManifestCommittable(e.getKey(), e.getValue()))
- .collect(Collectors.toList()));
+ .collect(Collectors.toList()),
+ checkAppendFiles);
}
private ManifestCommittable createManifestCommittable(
@@ -333,13 +361,15 @@ public int filterAndCommitMultiple(
List retryCommittables = commit.filterCommitted(sortedCommittables);
if (!retryCommittables.isEmpty()) {
- checkFilesExistence(retryCommittables);
+ if (checkFilesExistence) {
+ verifyFilesExist(retryCommittables);
+ }
commitMultiple(retryCommittables, checkAppendFiles);
}
return retryCommittables.size();
}
- private void checkFilesExistence(List committables) {
+ private void verifyFilesExist(List committables) {
List files = new ArrayList<>();
DataFilePathFactories factories = new DataFilePathFactories(commit.pathFactory());
IndexFilePathFactories indexFactories = new IndexFilePathFactories(commit.pathFactory());
@@ -411,7 +441,7 @@ private void maintain(long identifier, ExecutorService executor, boolean doExpir
throw new RuntimeException(maintainError.get());
}
- if (batchCommitted) {
+ if (batchCommitted || inlineMaintenance) {
maintain(identifier, doExpire);
} else {
executor.execute(
diff --git a/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeySimpleTableTest.java b/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeySimpleTableTest.java
index 6a7a722e855b..c97d112732ae 100644
--- a/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeySimpleTableTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeySimpleTableTest.java
@@ -88,6 +88,7 @@
import org.apache.paimon.utils.JsonSerdeUtil;
import org.apache.paimon.utils.Pair;
import org.apache.paimon.utils.RoaringBitmap32;
+import org.apache.paimon.utils.SnapshotManager;
import org.apache.commons.math3.random.RandomDataGenerator;
import org.assertj.core.api.Assertions;
@@ -124,6 +125,7 @@
import static org.apache.paimon.CoreOptions.CHANGELOG_NUM_RETAINED_MAX;
import static org.apache.paimon.CoreOptions.CHANGELOG_NUM_RETAINED_MIN;
import static org.apache.paimon.CoreOptions.CHANGELOG_PRODUCER;
+import static org.apache.paimon.CoreOptions.COMMIT_STRICT_MODE_LAST_SAFE_SNAPSHOT;
import static org.apache.paimon.CoreOptions.ChangelogProducer.LOOKUP;
import static org.apache.paimon.CoreOptions.DELETION_VECTORS_ENABLED;
import static org.apache.paimon.CoreOptions.FILE_FORMAT;
@@ -284,6 +286,50 @@ public void testPostponeBucket() throws Exception {
assertThat(file.valueStatsCols()).isEmpty();
}
+ @Test
+ public void testPostponeFixedBucketReplayLookupWithProvidedUser() throws Exception {
+ FileStoreTable table =
+ createFileStoreTable(options -> options.set(BUCKET, BucketMode.POSTPONE_BUCKET));
+ SnapshotManager sm = table.snapshotManager();
+
+ // A first run commits identifier 0 under a caller-provided user.
+ PostponeFixedBucketWriteBuilder first =
+ table.newPostponeFixedBucketWriteBuilder().withCommitUser("user");
+ try (TableWriteImpl write = first.newWrite();
+ InnerTableCommit commit = first.newCommit()) {
+ write.writeAndReturn(rowData(1, 1, 1L), 0, 1);
+ commit.filterAndCommit(Collections.singletonMap(0L, write.prepareCommit()));
+ }
+ long committed = sm.latestSnapshotId();
+
+ // A restarted run replays identifier 0 in strict mode bounded by that snapshot, the way
+ // a direct postpone write starts from the latest snapshot. The provided user has to be
+ // looked up beyond the bound for the replay to be recognised.
+ Map strict = new HashMap<>();
+ strict.put(COMMIT_STRICT_MODE_LAST_SAFE_SNAPSHOT.key(), String.valueOf(committed));
+ FileStoreTable strictTable = table.copy(strict);
+ PostponeFixedBucketWriteBuilder replay =
+ strictTable.newPostponeFixedBucketWriteBuilder().withCommitUser("user");
+ try (TableWriteImpl write = replay.newWrite();
+ InnerTableCommit commit = replay.newCommit()) {
+ write.writeAndReturn(rowData(1, 1, 1L), 0, 1);
+ commit.filterAndCommit(Collections.singletonMap(0L, write.prepareCommit()));
+ }
+ assertThat(sm.latestSnapshotId())
+ .as("a replay by a provided user must be recognised across the strict mode bound")
+ .isEqualTo(committed);
+
+ // A committer created for an explicitly passed user is one the caller manages itself,
+ // like the staged committer with its per-run user, and keeps the bound.
+ PostponeFixedBucketWriteBuilder explicit = strictTable.newPostponeFixedBucketWriteBuilder();
+ try (TableWriteImpl write = explicit.newWrite("user", null);
+ InnerTableCommit commit = explicit.newCommit("user", true)) {
+ write.writeAndReturn(rowData(2, 2, 2L), 0, 1);
+ commit.filterAndCommit(Collections.singletonMap(0L, write.prepareCommit()));
+ }
+ assertThat(sm.latestSnapshotId()).isEqualTo(committed + 1);
+ }
+
@Test
public void testPostponeFixedBucketWriteBuilder() throws Exception {
FileStoreTable table =
diff --git a/paimon-core/src/test/java/org/apache/paimon/table/SimpleTableTestBase.java b/paimon-core/src/test/java/org/apache/paimon/table/SimpleTableTestBase.java
index f768753d7fe7..99fb4bf45875 100644
--- a/paimon-core/src/test/java/org/apache/paimon/table/SimpleTableTestBase.java
+++ b/paimon-core/src/test/java/org/apache/paimon/table/SimpleTableTestBase.java
@@ -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;
@@ -107,6 +108,7 @@
import static org.apache.paimon.CoreOptions.BUCKET_KEY;
import static org.apache.paimon.CoreOptions.CHANGELOG_NUM_RETAINED_MAX;
import static org.apache.paimon.CoreOptions.CHANGELOG_NUM_RETAINED_MIN;
+import static org.apache.paimon.CoreOptions.COMMIT_STRICT_MODE_LAST_SAFE_SNAPSHOT;
import static org.apache.paimon.CoreOptions.CONSUMER_IGNORE_PROGRESS;
import static org.apache.paimon.CoreOptions.DELETION_VECTORS_ENABLED;
import static org.apache.paimon.CoreOptions.ExpireExecutionMode;
@@ -1594,6 +1596,100 @@ 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
+ public void testFilterAndCommitWithProvidedUserUnderStrictMode() throws Exception {
+ FileStoreTable table = createFileStoreTable(conf -> {});
+ SnapshotManager sm = table.snapshotManager();
+
+ // A first run commits identifier 0 under a caller-provided user.
+ BatchWriteBuilderImpl first =
+ ((BatchWriteBuilderImpl) table.newBatchWriteBuilder()).withCommitUser("user");
+ try (BatchTableWrite write = first.newWrite();
+ InnerTableCommit commit = first.newCommit()) {
+ write.write(rowData(1, 10, 100L));
+ commit.filterAndCommit(Collections.singletonMap(0L, write.prepareCommit()));
+ }
+ long committed = sm.latestSnapshotId();
+
+ // A restarted run replays identifier 0 with strict mode bounded by the snapshot it starts
+ // from, which is the snapshot that identifier produced. The bound only saves lookup work
+ // for a user created for one run; a provided user has to be looked up beyond it, or the
+ // replay is committed again.
+ Map strict = new HashMap<>();
+ strict.put(COMMIT_STRICT_MODE_LAST_SAFE_SNAPSHOT.key(), String.valueOf(committed));
+ BatchWriteBuilderImpl replay =
+ ((BatchWriteBuilderImpl) table.copy(strict).newBatchWriteBuilder())
+ .withCommitUser("user");
+ try (BatchTableWrite write = replay.newWrite();
+ InnerTableCommit commit = replay.newCommit()) {
+ write.write(rowData(1, 10, 100L));
+ commit.filterAndCommit(Collections.singletonMap(0L, write.prepareCommit()));
+ }
+ assertThat(sm.latestSnapshotId())
+ .as("a replay by a provided user must be recognised across the strict mode bound")
+ .isEqualTo(committed);
+
+ // The bound stays in place for a user the builder created itself.
+ BatchWriteBuilderImpl generated =
+ (BatchWriteBuilderImpl) table.copy(strict).newBatchWriteBuilder();
+ try (BatchTableWrite write = generated.newWrite();
+ InnerTableCommit commit = generated.newCommit()) {
+ write.write(rowData(2, 20, 200L));
+ commit.filterAndCommit(Collections.singletonMap(0L, write.prepareCommit()));
+ }
+ assertThat(sm.latestSnapshotId()).isEqualTo(committed + 1);
+ }
+
@Test
@Timeout(120)
public void testExpireWithLimit() throws Exception {
diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkConnectorOptions.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkConnectorOptions.java
index 4dd9329d1c4c..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
@@ -109,6 +109,20 @@ public class SparkConnectorOptions {
"Wait time in milliseconds between retry attempts for Spark V1 UPDATE "
+ "on data-evolution tables after row-id range update conflicts.");
+ public static final ConfigOption STREAM_WRITE_COMMIT_USER =
+ key("write.stream.commit-user")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "The commit user of a Structured Streaming write. Paimon skips a "
+ + "micro-batch that a previous run of the same query already "
+ + "committed under this user, which is what makes a replayed "
+ + "micro-batch idempotent. By default it is derived from the "
+ + "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")
.intType()
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 91efc3d541b1..62a8466b3db9 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala
@@ -44,12 +44,13 @@ import org.apache.paimon.types.RowKind
import org.apache.paimon.utils.{SerializationUtils, UriReaderFactory}
import org.apache.spark.{Partitioner, TaskContext}
+import org.apache.spark.internal.Logging
import org.apache.spark.rdd.RDD
import org.apache.spark.sql._
import org.apache.spark.sql.functions._
import java.io.IOException
-import java.util.{Map => JMap}
+import java.util.{Collections, Map => JMap}
import java.util.Collections.singletonMap
import scala.collection.JavaConverters._
@@ -57,8 +58,10 @@ import scala.collection.JavaConverters._
case class PaimonSparkWriter(
table: FileStoreTable,
writeRowTracking: Boolean = false,
- batchId: Option[Long] = None)
- extends WriteHelper {
+ batchId: Option[Long] = None,
+ commitUser: Option[String] = None)
+ extends WriteHelper
+ with Logging {
private lazy val tableSchema = table.schema
@@ -99,7 +102,13 @@ case class PaimonSparkWriter(
if (bucketNum.isPresent) Some(bucketNum.get().intValue()) else None
}
- val writeBuilder: BatchWriteBuilder = table.newBatchWriteBuilder()
+ val writeBuilder: BatchWriteBuilder = {
+ val builder = table.newBatchWriteBuilder()
+ // A streaming write commits under a commit user that survives a restart, so that a replayed
+ // micro-batch can be recognised as already committed.
+ commitUser.foreach(builder.asInstanceOf[BatchWriteBuilderImpl].withCommitUser)
+ builder
+ }
def withOverwrite(): PaimonSparkWriter = withOverwrite(java.util.Collections.emptyMap())
@@ -144,6 +153,7 @@ case class PaimonSparkWriter(
COMMIT_STRICT_MODE_LAST_SAFE_SNAPSHOT.key(),
postponeBaseSnapshotId.getOrElse(0L).toString)
val builder = table.copy(directWriteOptions).newPostponeFixedBucketWriteBuilder()
+ commitUser.foreach(builder.withCommitUser)
overwritePartitionSpec.foreach(spec => builder.withOverwrite(spec.asJava))
directPostponeWriteBuilder = builder
builder
@@ -454,6 +464,16 @@ case class PaimonSparkWriter(
writeBuilder.asInstanceOf[BatchWriteBuilderImpl].rowIdCheckConflict(rowIdCheckFromSnapshot)
}
+ /**
+ * The commit identifier to deduplicate on, present only for a streaming write that has both a
+ * batch id and a commit user that is stable across restarts.
+ */
+ private def idempotentCommitIdentifier: Option[Long] =
+ for {
+ identifier <- batchId
+ _ <- commitUser
+ } yield identifier
+
def commit(commitMessages: Seq[CommitMessage]): Unit = {
commit(commitMessages, null)
}
@@ -463,6 +483,13 @@ case class PaimonSparkWriter(
if (stagedSparkSession == null) {
throw new IllegalStateException("Postpone staged write has no SparkSession.")
}
+ idempotentCommitIdentifier.foreach {
+ identifier =>
+ logWarning(
+ s"Micro-batch $identifier is written to a postpone bucket table through a staged " +
+ "commit, which cannot deduplicate a replayed batch. A failure of this query may " +
+ "duplicate the batch.")
+ }
val finalOperation = Option(operation).getOrElse(Snapshot.Operation.WRITE)
val finalMessages = new SparkPostponeStagedCommitter(
table,
@@ -472,14 +499,38 @@ case class PaimonSparkWriter(
postCommit(finalMessages)
return
}
- val activeWriteBuilder =
- Option(directPostponeWriteBuilder).getOrElse(writeBuilder)
- val tableCommit = activeWriteBuilder.newCommit()
+ val tableCommit: InnerTableCommit =
+ if (directPostponeWriteBuilder != null) {
+ directPostponeWriteBuilder.newCommit()
+ } else {
+ writeBuilder.asInstanceOf[BatchWriteBuilderImpl].newCommit()
+ }
if (operation != null) {
tableCommit.withOperation(operation)
}
try {
- tableCommit.commit(commitMessages.toList.asJava)
+ idempotentCommitIdentifier match {
+ case Some(identifier) =>
+ // Structured Streaming replays a micro-batch with its original batch id after a failure.
+ // Committing under a stable commit user lets Paimon skip a replay it already committed,
+ // instead of duplicating the whole batch, 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.
+ //
+ // 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)
+ }
} catch {
case e: Throwable => throw new RuntimeException(e);
} finally {
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/WriteIntoPaimonTable.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/WriteIntoPaimonTable.scala
index 937a47c526e9..2a803a761c1d 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/WriteIntoPaimonTable.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/WriteIntoPaimonTable.scala
@@ -40,7 +40,8 @@ case class WriteIntoPaimonTable(
saveMode: SaveMode,
_data: DataFrame,
options: Options,
- batchId: Option[Long] = None)
+ batchId: Option[Long] = None,
+ commitUser: Option[String] = None)
extends RunnableCommand
with ExpressionHelper
with SchemaEvolutionHelper
@@ -58,7 +59,7 @@ case class WriteIntoPaimonTable(
updateTableWithOptions(
Map(DYNAMIC_PARTITION_OVERWRITE.key -> dynamicPartitionOverwriteMode.toString))
- val writer = PaimonSparkWriter(table, batchId = batchId)
+ val writer = PaimonSparkWriter(table, batchId = batchId, commitUser = commitUser)
if (overwritePartition != null) {
writer.withOverwrite(overwritePartition.asJava)
}
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 9d0a1795b589..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
@@ -19,15 +19,21 @@
package org.apache.paimon.spark.sources
import org.apache.paimon.options.Options
-import org.apache.paimon.spark.{InsertInto, Overwrite}
+import org.apache.paimon.spark.{InsertInto, Overwrite, SparkConnectorOptions}
import org.apache.paimon.spark.commands.{SchemaEvolutionHelper, WriteIntoPaimonTable}
import org.apache.paimon.table.FileStoreTable
+import org.apache.spark.internal.Logging
import org.apache.spark.sql.{DataFrame, PaimonUtils, SQLContext}
import org.apache.spark.sql.execution.streaming.Sink
import org.apache.spark.sql.sources.AlwaysTrue
import org.apache.spark.sql.streaming.OutputMode
+import java.nio.charset.StandardCharsets.UTF_8
+import java.util.UUID
+
+import scala.collection.JavaConverters._
+
class PaimonSink(
sqlContext: SQLContext,
override val originTable: FileStoreTable,
@@ -35,7 +41,73 @@ class PaimonSink(
outputMode: OutputMode,
options: Options)
extends Sink
- with SchemaEvolutionHelper {
+ with SchemaEvolutionHelper
+ with Logging {
+
+ /**
+ * Structured Streaming replays a micro-batch with its original batch id when a query is restarted
+ * after failing between this sink returning from [[addBatch]] and Spark recording the batch as
+ * 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.
+ *
+ * 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 {
+ 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 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.")
+ UUID.randomUUID().toString
+ }
+ }
+ }
+
+ // Like the read side, which takes its 'read.stream.*' options from the table, so that a
+ // 'spark.paimon.' session conf works the same as an option of the writer.
+ private def configuredCommitUser: Option[String] = {
+ val fromWriter = options.get(SparkConnectorOptions.STREAM_WRITE_COMMIT_USER)
+ val fromTable =
+ Options.fromMap(originTable.options()).get(SparkConnectorOptions.STREAM_WRITE_COMMIT_USER)
+ Seq(fromWriter, fromTable).find(user => user != null && user.nonEmpty)
+ }
+
+ // Spark hands the sink its options case-insensitively, but keeps whatever case the user wrote.
+ private def checkpointLocation: Option[String] =
+ options.toMap.asScala.collectFirst {
+ case (key, value)
+ if key.equalsIgnoreCase(PaimonSink.CHECKPOINT_LOCATION) && value != null &&
+ value.nonEmpty =>
+ value
+ }
+
+ /**
+ * 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)
+
+ private def derivedCommitUser(kind: String, value: String): String = {
+ val user = s"spark-$kind-${UUID.nameUUIDFromBytes(value.getBytes(UTF_8))}"
+ logInfo(s"Streaming writes to ${originTable.name()} commit as '$user'.")
+ user
+ }
override def addBatch(batchId: Long, data: DataFrame): Unit = {
val saveMode = if (outputMode == OutputMode.Complete()) {
@@ -44,7 +116,18 @@ class PaimonSink(
InsertInto
}
val newData = PaimonUtils.createNewDataFrame(data)
- WriteIntoPaimonTable(originTable, saveMode, newData, options, Some(batchId)).run(
- sqlContext.sparkSession)
+ WriteIntoPaimonTable(originTable, saveMode, newData, options, Some(batchId), Some(commitUser))
+ .run(sqlContext.sparkSession)
}
}
+
+object PaimonSink {
+
+ private val CHECKPOINT_LOCATION = "checkpointLocation"
+
+ /**
+ * `org.apache.spark.sql.execution.streaming.StreamExecution.QUERY_ID_KEY`, inlined because that
+ * class is not in the same package across all supported Spark versions.
+ */
+ private val QUERY_ID_KEY = "sql.streaming.queryId"
+}
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
new file mode 100644
index 000000000000..ca35348efbda
--- /dev/null
+++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/PaimonSinkIdempotencyTest.scala
@@ -0,0 +1,625 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+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}
+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, List => JList, Map => JMap}
+
+import scala.collection.JavaConverters._
+
+/**
+ * Structured Streaming guarantees exactly-once only if the sink is idempotent for a repeated
+ * batchId: when a query fails between the sink returning from `addBatch` and Spark recording the
+ * batch as completed, the restarted query replays that micro-batch with its original batchId.
+ */
+class PaimonSinkIdempotencyTest extends PaimonSparkTestBase with StreamTest {
+
+ override protected def sparkConf: SparkConf = {
+ super.sparkConf.set("spark.sql.catalog.paimon.cache-enabled", "false")
+ }
+
+ import testImplicits._
+
+ private def snapshotCount(tableName: String): Long =
+ loadTable(tableName).snapshotManager().snapshotCount()
+
+ 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()
+ } finally {
+ query.stop()
+ }
+ }
+
+ /**
+ * Leave the checkpoint in the state a driver failure leaves behind when it dies after the sink
+ * returned from `addBatch` but before Spark recorded the batch: the offset log still has the
+ * batch, the commit log does not. The restarted query replays it with the same batchId.
+ */
+ private def dropCommitLogEntry(checkpointPath: String, batchId: Long): Unit = {
+ val commitsDir = new File(checkpointPath, "commits")
+ val names = Set(batchId.toString, s".$batchId.crc")
+ val entries = commitsDir.listFiles().filter(f => names.contains(f.getName))
+ assert(
+ entries.exists(_.getName == batchId.toString),
+ s"no commit log entry for batch $batchId in $commitsDir")
+ entries.foreach(f => assert(f.delete()))
+ }
+
+ test("Paimon Sink: replayed micro-batch must not be committed twice") {
+ 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"), (2, "b"), (3, "c"))
+
+ def start(): StreamingQuery =
+ df.writeStream
+ .option("checkpointLocation", checkpointPath)
+ .format("paimon")
+ .start(location)
+
+ runToCompletion(start())
+
+ val expected = Row(1, "a") :: Row(2, "b") :: Row(3, "c") :: Nil
+ checkAnswer(spark.sql("SELECT * FROM T ORDER BY a"), expected)
+ assert(snapshotCount("T") == 1)
+ assert(
+ latestCommitUser("T").startsWith("spark-query-"),
+ s"expected a commit user derived from the query id, " +
+ s"but got '${latestCommitUser("T")}'"
+ )
+
+ dropCommitLogEntry(checkpointPath, 0)
+ runToCompletion(start())
+
+ // The replayed batch must be recognised as already committed.
+ checkAnswer(spark.sql("SELECT * FROM T ORDER BY a"), expected)
+ assert(
+ snapshotCount("T") == 1,
+ s"replaying batch 0 created a second snapshot (${snapshotCount("T")} in total)")
+ }
+ }
+ }
+
+ test("Paimon Sink: replay is recognised when only the query id is available") {
+ failAfter(streamingTimeout) {
+ withTempDir {
+ checkpointRoot =>
+ spark.sql("CREATE TABLE T (a INT, b STRING)")
+ val location = loadTable("T").location().toString
+ val queryName = "paimon_idempotency"
+ // The location never reaches the sink options this way, so the commit user has to come
+ // from the query id that Spark persists in the checkpoint metadata.
+ val checkpointPath = new File(checkpointRoot, queryName).getCanonicalPath
+
+ withSQLConf("spark.sql.streaming.checkpointLocation" -> checkpointRoot.getCanonicalPath) {
+ val inputData = MemoryStream[(Int, String)]
+ val df = inputData.toDS().toDF("a", "b")
+ inputData.addData((1, "a"), (2, "b"), (3, "c"))
+
+ def start(): StreamingQuery =
+ df.writeStream
+ .queryName(queryName)
+ .format("paimon")
+ .start(location)
+
+ runToCompletion(start())
+
+ val expected = Row(1, "a") :: Row(2, "b") :: Row(3, "c") :: Nil
+ checkAnswer(spark.sql("SELECT * FROM T ORDER BY a"), expected)
+ assert(snapshotCount("T") == 1)
+ assert(
+ latestCommitUser("T").startsWith("spark-query-"),
+ s"expected a commit user derived from the query id, " +
+ s"but got '${latestCommitUser("T")}'")
+
+ dropCommitLogEntry(checkpointPath, 0)
+ runToCompletion(start())
+
+ checkAnswer(spark.sql("SELECT * FROM T ORDER BY a"), expected)
+ assert(
+ snapshotCount("T") == 1,
+ s"replaying batch 0 created a second snapshot (${snapshotCount("T")} in total)")
+ }
+ }
+ }
+ }
+
+ test("Paimon Sink: replay of a batch that is not the first one is recognised") {
+ 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")
+
+ def start(): StreamingQuery =
+ df.writeStream
+ .option("checkpointLocation", checkpointPath)
+ .format("paimon")
+ .start(location)
+
+ val query = start()
+ try {
+ inputData.addData((1, "a"))
+ query.processAllAvailable()
+ inputData.addData((2, "b"))
+ query.processAllAvailable()
+ inputData.addData((3, "c"))
+ query.processAllAvailable()
+ } finally {
+ query.stop()
+ }
+
+ val expected = Row(1, "a") :: Row(2, "b") :: Row(3, "c") :: Nil
+ checkAnswer(spark.sql("SELECT * FROM T ORDER BY a"), expected)
+ assert(snapshotCount("T") == 3)
+
+ dropCommitLogEntry(checkpointPath, 2)
+ runToCompletion(start())
+
+ checkAnswer(spark.sql("SELECT * FROM T ORDER BY a"), expected)
+ assert(
+ snapshotCount("T") == 3,
+ s"replaying batch 2 created another snapshot (${snapshotCount("T")} in total)")
+ }
+ }
+ }
+
+ test("Paimon Sink: replayed micro-batch of a complete mode query is not committed twice") {
+ failAfter(streamingTimeout) {
+ withTempDir {
+ checkpointDir =>
+ spark.sql("CREATE TABLE T (city STRING, population LONG)")
+ 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")
+
+ 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)"
+ )
+ }
+ }
+ }
+
+ 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 {
+ 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"), (2, "b"), (3, "c"))
+
+ def start(): StreamingQuery =
+ df.writeStream
+ .option("checkpointLocation", checkpointPath)
+ .option("write.stream.commit-user", "my-streaming-job")
+ .format("paimon")
+ .start(location)
+
+ runToCompletion(start())
+
+ val expected = Row(1, "a") :: Row(2, "b") :: Row(3, "c") :: Nil
+ checkAnswer(spark.sql("SELECT * FROM T ORDER BY a"), expected)
+ assert(latestCommitUser("T") == "my-streaming-job")
+
+ dropCommitLogEntry(checkpointPath, 0)
+ runToCompletion(start())
+
+ checkAnswer(spark.sql("SELECT * FROM T ORDER BY a"), expected)
+ assert(
+ snapshotCount("T") == 1,
+ s"replaying batch 0 created a second snapshot (${snapshotCount("T")} in total)")
+ }
+ }
+ }
+
+ test("Paimon Sink: write.stream.commit-user can come from a session conf") {
+ failAfter(streamingTimeout) {
+ withTempDir {
+ checkpointDir =>
+ spark.sql("CREATE TABLE T (a INT, b STRING)")
+ val location = loadTable("T").location().toString
+
+ withSQLConf("spark.paimon.write.stream.commit-user" -> "job-from-conf") {
+ val inputData = MemoryStream[(Int, String)]
+ val df = inputData.toDS().toDF("a", "b")
+ inputData.addData((1, "a"))
+
+ runToCompletion(
+ df.writeStream
+ .option("checkpointLocation", checkpointDir.getCanonicalPath)
+ .format("paimon")
+ .start(location))
+ }
+
+ checkAnswer(spark.sql("SELECT * FROM T"), Row(1, "a") :: Nil)
+ assert(latestCommitUser("T") == "job-from-conf")
+ }
+ }
+ }
+
+ 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 =>
+ 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")}")
+ }
+ }
+ }
+}
+
+/** 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 = {}
+ }
+}