From 28cb9fef814f6b00a006f6c5b014b0dcf28fb4c0 Mon Sep 17 00:00:00 2001 From: Xiangyi Zhu <82511136+zhuxiangyi@users.noreply.github.com> Date: Wed, 21 Oct 2026 21:02:08 +0800 Subject: [PATCH] [spark] Commit a streaming micro-batch through the stream write API A Structured Streaming write went through the batch write API: every micro-batch built its own write builder and committed through a one-shot committer under a random commit user. Such a commit cannot be recognised again, so a query that failed between the sink returning from addBatch and Spark recording that batch in its commit log wrote the whole batch a second time on restart (#9666). For an append-only table nothing merges the duplicate rows away. The sink now commits the way a Flink job does: - one commit user per query, stable across its runs. It is derived from the query id Spark persists in the checkpoint, which is new when a checkpoint is recreated and unchanged when a query resumes from one: batch ids are only unique within one checkpoint, so the identity must not outlive it. 'commit.user-prefix' prefixes it, and a run resuming a checkpoint under another prefix is refused. - one StreamTableCommit per run, created with the first micro-batch and closed when that run of the query terminates. - micro-batch n committed under identifier n + 1, the way Flink numbers its checkpoints, which is also what a compacted-full scan recognises a scheduled full compaction by. - the first micro-batch of a run looks the commit user up. A batch it already committed is not committed again; it goes through the filtered commit without its files, so that the commit callbacks retry what may have failed after the first commit, such as registering partitions. - a replay is recognised by the snapshots of the commit user, which expiration eventually removes. Before committing, the sink records the latest snapshot under the checkpoint of the query; if expiration has gone past it by the replay, the query fails instead of possibly committing the batch twice. - a query in complete mode overwrites the table with the complete result of each micro-batch, so its replay is committed again, as before, and leaves the same rows. - a write to a postpone bucket table writes the postpone bucket, as a Flink streaming write does. --- docs/docs/spark/structured-streaming.md | 52 + .../spark/commands/PaimonSparkWriter.scala | 99 +- .../spark/commands/WriteIntoPaimonTable.scala | 6 +- .../paimon/spark/sources/PaimonSink.scala | 245 ++- .../paimon/spark/write/CommitMarker.scala | 120 ++ .../apache/paimon/spark/write/DataWrite.scala | 14 +- .../paimon/spark/write/PaimonDataWrite.scala | 26 +- .../spark/write/PaimonV2DataWriter.scala | 2 +- .../spark/write/StreamingWriteContext.scala | 209 +++ .../spark/PaimonSinkIdempotencyTest.scala | 1547 +++++++++++++++++ .../apache/paimon/spark/PaimonSinkTest.scala | 55 + 11 files changed, 2344 insertions(+), 31 deletions(-) create mode 100644 paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/CommitMarker.scala create mode 100644 paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/StreamingWriteContext.scala create mode 100644 paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/PaimonSinkIdempotencyTest.scala diff --git a/docs/docs/spark/structured-streaming.md b/docs/docs/spark/structured-streaming.md index cc28c1d652f5..15de812edbb3 100644 --- a/docs/docs/spark/structured-streaming.md +++ b/docs/docs/spark/structured-streaming.md @@ -66,6 +66,58 @@ Use a durable checkpoint location accessible to the cluster for deployed jobs. G its own checkpoint directory. Stop this example with `writer.stop()`. Streaming writes also support [Schema Evolution on Write](./schema-evolution). +### Exactly-once + +Structured Streaming replays a micro-batch with its original batch id when a query is restarted +after failing between the sink writing the batch and Spark recording that batch as completed. +Paimon commits every micro-batch under a commit user that is stable across restarts, and skips a +batch that the same user already committed, so a replay does not write the data twice. A query in +`complete` mode is the exception: each of its micro-batches overwrites the table with the complete +result, so its replay is committed again and leaves the same rows, as before. 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, since the batch ids it +numbers are only unique within one: reusing it across two checkpoints would make Paimon skip the +data of the second one, while changing it within one query would bring the duplicate back. It is +therefore derived from the query id that Spark persists in the checkpoint, which is new when a +checkpoint is recreated, unchanged when a query resumes from one, and independent of how the +location is spelled, and it cannot be set explicitly. `commit.user-prefix` is read from the table as +usual and prefixes the derived user, so a job keeps the name its table was configured with. + +:::note + +A skipped replay leaves the data files it wrote behind, uncommitted. They are removed by +[orphan file cleaning](../maintenance/manage-snapshots#remove-orphan-files), like any other +uncommitted file. + +A query that starts from a new checkpoint gets a new commit user, so a micro-batch the previous +run committed is not recognised and its data is written again. + +A replay is recognised by the snapshots its commit user left behind, which snapshot expiration +eventually removes. Before committing a micro-batch, the sink records under the checkpoint location +of the query which snapshot was the latest before its first attempt. Retries preserve this original +marker rather than advancing it. If filtering cannot recognise the replay and expiration has removed +the snapshot immediately after that boundary, the query fails instead of possibly writing it twice. +An incomplete or corrupt marker also fails the query, since it may belong to an already committed +batch. The error names the marker file; restore it or delete it only after verifying that the batch +was not committed. Retain snapshots for longer than a query may be down. + +A streaming write to a postpone bucket table writes the postpone bucket, as a Flink streaming job +does, whatever `postpone.batch-write-fixed-bucket` says: that option is for a batch job which ends +with its commit. The rows become readable once a +[compaction](../primary-key-table/data-distribution#postpone-bucket) has sorted them into real +buckets. + +::: + +The committer of a query is created with its first micro-batch and lives until the query +terminates, again as a Flink committer lives across checkpoints. Only the first micro-batch of a +run looks up what a previous run committed; the maintenance a commit schedules, such as tag +creation and snapshot or partition expiration, runs while the next micro-batch is written when +`snapshot.expire.execution-mode` is `async`. + ## Streaming Query :::info 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 132b8bfe470f..b1292a41b8cf 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 @@ -37,7 +37,7 @@ import org.apache.paimon.spark.schema.SparkSystemColumns.{BUCKET_COL, ROW_KIND_C import org.apache.paimon.spark.sort.TableSorter import org.apache.paimon.spark.util.OptionUtils.paimonExtensionEnabled import org.apache.paimon.spark.util.SparkRowUtils -import org.apache.paimon.spark.write.{PaimonDataWrite, WriteHelper, WriteTaskResult} +import org.apache.paimon.spark.write.{PaimonDataWrite, StreamingWrite, WriteHelper, WriteTaskResult} import org.apache.paimon.table.{FileStoreTable, SpecialFields} import org.apache.paimon.table.BucketMode._ import org.apache.paimon.table.sink._ @@ -45,6 +45,7 @@ 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._ @@ -55,11 +56,20 @@ import java.util.Collections.singletonMap import scala.collection.JavaConverters._ +/** + * @param streaming + * the micro-batch this writer writes, when it writes for a streaming query. Such a write goes + * through the stream write API, the way a Flink streaming job does: it is committed under an + * identifier by the committer of the query, which recognises a replayed micro-batch. + */ case class PaimonSparkWriter( table: FileStoreTable, writeRowTracking: Boolean = false, - batchId: Option[Long] = None) - extends WriteHelper { + streaming: Option[StreamingWrite] = None) + extends WriteHelper + with Logging { + + private val commitIdentifier: Option[Long] = streaming.map(_.commitIdentifier) private lazy val tableSchema = table.schema @@ -89,8 +99,14 @@ case class PaimonSparkWriter( @transient private lazy val metricRegistry = SparkMetricRegistry() + /** + * A streaming write to a postpone bucket table always goes to bucket -2, to be bucketed by a + * later compaction, the same as a Flink streaming write: the fixed-bucket paths are for a batch + * job that ends with its commit. + */ val postponeBatchWriteFixedBucket: Boolean = - table.bucketMode() == POSTPONE_MODE && coreOptions.postponeBatchWriteFixedBucket() + streaming.isEmpty && table.bucketMode() == POSTPONE_MODE && + coreOptions.postponeBatchWriteFixedBucket() private val postponeBaseSnapshotId = if (postponeBatchWriteFixedBucket) @@ -102,25 +118,34 @@ case class PaimonSparkWriter( if (bucketNum.isPresent) Some(bucketNum.get().intValue()) else None } - val writeBuilder: BatchWriteBuilder = table.newBatchWriteBuilder() + val writeBuilder: WriteBuilder = streaming match { + case Some(streamingWrite) => + // Every micro-batch is committed under the commit user of the query, which is stable across + // its runs, so that a replayed micro-batch can be recognised as already committed. + table.newStreamWriteBuilder().withCommitUser(streamingWrite.context.commitUser) + case None => table.newBatchWriteBuilder() + } def withOverwrite(): PaimonSparkWriter = withOverwrite(java.util.Collections.emptyMap()) def withOverwrite(partition: JMap[String, String]): PaimonSparkWriter = { overwritePartitionSpec = Some(partition.asScala.toMap) - if (!postponeBatchWriteFixedBucket) { - writeBuilder.withOverwrite(partition) + writeBuilder match { + case batchWriteBuilder: BatchWriteBuilder if !postponeBatchWriteFixedBucket => + batchWriteBuilder.withOverwrite(partition) + // A stream write builder is told at commit time; see [[commit]]. + case _ => } this } def writeOnly(): PaimonSparkWriter = { - PaimonSparkWriter(table.copy(singletonMap(WRITE_ONLY.key(), "true"))) + PaimonSparkWriter(table.copy(singletonMap(WRITE_ONLY.key(), "true")), streaming = streaming) } def withRowTracking(): PaimonSparkWriter = { if (coreOptions.rowTrackingEnabled()) { - PaimonSparkWriter(table, writeRowTracking = true) + PaimonSparkWriter(table, writeRowTracking = true, streaming = streaming) } else { this } @@ -140,7 +165,7 @@ case class PaimonSparkWriter( } else { None } - val activeWriteBuilder: BatchWriteBuilder = directPostponeBucketNum match { + val activeWriteBuilder: WriteBuilder = directPostponeBucketNum match { case Some(_) => val directWriteOptions = new java.util.HashMap[String, String]() directWriteOptions.put( @@ -176,9 +201,11 @@ case class PaimonSparkWriter( rowKindColIdx, writeRowTracking, fullCompactionDeltaCommits, - batchId, + commitIdentifier, uriReaderFactory, - postponePartitionBucketComputer + postponePartitionBucketComputer, + // The stream write builder does not know that the commit overwrites, see [[commit]]. + ignorePreviousFiles = streaming.isDefined && overwritePartitionSpec.isDefined ) def sparkParallelism = { @@ -458,6 +485,18 @@ case class PaimonSparkWriter( } def commit(commitMessages: Seq[CommitMessage], operation: Snapshot.Operation): Unit = { + streaming match { + case Some(streamingWrite) => + commitMicroBatch(streamingWrite, commitMessages, operation) + postCommit(commitMessages) + case None => + commitBatch(commitMessages, operation) + } + } + + private def commitBatch( + commitMessages: Seq[CommitMessage], + operation: Snapshot.Operation): Unit = { if (postponeBatchWriteFixedBucket && directPostponeWriteBuilder == null) { if (stagedSparkSession == null) { throw new IllegalStateException("Postpone staged write has no SparkSession.") @@ -471,8 +510,8 @@ case class PaimonSparkWriter( postCommit(finalMessages) return } - val activeWriteBuilder = - Option(directPostponeWriteBuilder).getOrElse(writeBuilder) + val activeWriteBuilder: BatchWriteBuilder = + Option(directPostponeWriteBuilder).getOrElse(writeBuilder.asInstanceOf[BatchWriteBuilder]) val tableCommit = activeWriteBuilder.newCommit() tableCommit.withMetricRegistry(metricRegistry) if (operation != null) { @@ -488,6 +527,38 @@ case class PaimonSparkWriter( postCommit(commitMessages) } + /** + * Commits a micro-batch through the committer of the query, which the first micro-batch of a run + * creates and every later one reuses. + */ + private def commitMicroBatch( + streamingWrite: StreamingWrite, + commitMessages: Seq[CommitMessage], + operation: Snapshot.Operation): Unit = { + streamingWrite.context.openCommitter { + val committer = writeBuilder + .asInstanceOf[StreamWriteBuilder] + .newCommit() + .asInstanceOf[InnerTableCommit] + committer.withMetricRegistry(metricRegistry) + // Both are the same for every micro-batch of a query: its output mode does not change, and + // a batch that wrote nothing is no reason to publish a snapshot, as for a batch write. + overwritePartitionSpec.foreach(spec => committer.withOverwrite(spec.asJava)) + committer.ignoreEmptyCommit( + Option(coreOptions.toConfiguration.get(CoreOptions.SNAPSHOT_IGNORE_EMPTY_COMMIT)) + .forall(_.booleanValue())) + if (operation != null) { + committer.withOperation(operation) + } + committer + } + streamingWrite.context.commit( + streamingWrite, + commitMessages, + table, + overwrite = overwritePartitionSpec.isDefined) + } + private def baseSnapshotHasNoRealBuckets: Boolean = { postponeBaseSnapshotId.forall { snapshotId => 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..70279bb0ce25 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 @@ -24,7 +24,7 @@ import org.apache.paimon.options.Options import org.apache.paimon.spark._ import org.apache.paimon.spark.catalyst.analysis.ReplacePaimonFunctions import org.apache.paimon.spark.catalyst.analysis.expressions.ExpressionHelper -import org.apache.paimon.spark.write.PaimonWriteOptions +import org.apache.paimon.spark.write.{PaimonWriteOptions, StreamingWrite} import org.apache.paimon.table.FileStoreTable import org.apache.spark.internal.Logging @@ -40,7 +40,7 @@ case class WriteIntoPaimonTable( saveMode: SaveMode, _data: DataFrame, options: Options, - batchId: Option[Long] = None) + streaming: Option[StreamingWrite] = None) extends RunnableCommand with ExpressionHelper with SchemaEvolutionHelper @@ -58,7 +58,7 @@ case class WriteIntoPaimonTable( updateTableWithOptions( Map(DYNAMIC_PARTITION_OVERWRITE.key -> dynamicPartitionOverwriteMode.toString)) - val writer = PaimonSparkWriter(table, batchId = batchId) + val writer = PaimonSparkWriter(table, streaming = streaming) 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..09db44d3e668 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 @@ -18,15 +18,24 @@ package org.apache.paimon.spark.sources +import org.apache.paimon.CoreOptions import org.apache.paimon.options.Options import org.apache.paimon.spark.{InsertInto, Overwrite} import org.apache.paimon.spark.commands.{SchemaEvolutionHelper, WriteIntoPaimonTable} +import org.apache.paimon.spark.write.{CommitMarker, StreamingWrite, StreamingWriteContext} 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 org.apache.spark.sql.streaming.{OutputMode, StreamingQueryListener} + +import java.nio.charset.StandardCharsets.UTF_8 +import java.util.UUID + +import scala.collection.JavaConverters._ +import scala.util.Try class PaimonSink( sqlContext: SQLContext, @@ -35,7 +44,167 @@ 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, since the + * micro-batches it numbers are only unique within one. Paimon skips a batch whose id a previous + * run committed under the same user, so reusing a user across two checkpoints drops the data of + * the second one, while changing it within one 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. It is therefore not configurable; 'commit.user-prefix' can name it. + * + * Resolved lazily: neither the query id nor the checkpoint location is available on the thread + * that constructs the sink. + */ + private lazy val commitUser: String = { + val user = withPrefix(identity) + logInfo(s"Streaming writes to ${originTable.name()} commit as '$user'.") + user + } + + /** The part of the commit user that identifies the checkpoint, without the prefix. */ + private lazy val identity: String = + queryId + .map(derivedIdentity("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(derivedIdentity("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 " + + "be committed twice.") + UUID.randomUUID().toString + } + + // 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 derivedIdentity(kind: String, value: String): String = + s"spark-$kind-${UUID.nameUUIDFromBytes(value.getBytes(UTF_8))}" + + /** + * 'commit.user-prefix' names the writers of a table, the way + * [[org.apache.paimon.CoreOptions#createCommitUser]] prefixes the random user of a batch write. + * It prefixes a derived user here too, so that a streaming job keeps the name its table was + * configured with. It cannot be the identity by itself, since every writer of the table shares + * it, so what follows it still has to be unique to the query. Being part of the commit user, it + * cannot change while a checkpoint is resumed; see [[checkPrefixUnchanged]]. + */ + private def withPrefix(user: String): String = + Seq( + options.get(CoreOptions.COMMIT_USER_PREFIX), + originTable.options().get(CoreOptions.COMMIT_USER_PREFIX.key)) + .find(prefix => prefix != null && prefix.nonEmpty) + .fold(user)(prefix => s"${prefix}_$user") + + /** + * The writer and the committer of this run of the query, from its first micro-batch until it + * terminates. Spark creates a sink per run, so a restarted query starts a new context and looks + * up once what the previous run committed. + */ + private var context: StreamingWriteContext = _ + + private def openContext(): StreamingWriteContext = synchronized { + if (context == null) { + val marker = + markerLocation.map( + new CommitMarker(_, sqlContext.sparkSession.sessionState.newHadoopConf())) + marker.foreach(checkPrefixUnchanged) + context = new StreamingWriteContext(commitUser, marker) + } + context + } + + /** Closes the context of this run, if it has created one. */ + private def closeContext(): Unit = synchronized { + if (context != null) { + context.close() + } + } + + /** + * A replayed micro-batch is only recognised under the commit user that committed it, so a run + * resuming a checkpoint under a different 'commit.user-prefix' would commit it twice. The marker + * left by the previous run names its commit user: one with the same identity but another prefix + * is that of this checkpoint, and the run is refused. A marker of another identity belongs to a + * query that used the location before, which says nothing about this one. + */ + private def checkPrefixUnchanged(marker: CommitMarker): Unit = + marker.commitUser.filter(_ != commitUser).foreach { + previous => + if (previous == identity || previous.endsWith(s"_$identity")) { + val previousPrefix = previous.stripSuffix(identity).stripSuffix("_") + throw new IllegalStateException( + s"The streaming query writing to ${originTable.name()} committed as '$previous' " + + s"before it was restarted, but now commits as '$commitUser': " + + s"'${CoreOptions.COMMIT_USER_PREFIX.key}' was changed. A micro-batch replayed by " + + "the restart would not be recognised and would be committed twice. " + + (if (previousPrefix.isEmpty) { + s"Unset '${CoreOptions.COMMIT_USER_PREFIX.key}'" + } else { + s"Set '${CoreOptions.COMMIT_USER_PREFIX.key}' back to '$previousPrefix'" + }) + + " to resume this checkpoint, or start the query from a new checkpoint.") + } + } + + /** + * Where the commit marker is kept: the checkpoint of the query, which the writer names only when + * it passes 'checkpointLocation' as an option. A checkpoint configured through + * 'spark.sql.streaming.checkpointLocation' and a query name never reaches the options, so it is + * taken from the running query instead. A query whose checkpoint cannot be found fails rather + * than write without a marker, which would let a replay after snapshot expiration be committed + * twice. Outside a stream execution there is no replay to guard. + */ + private def markerLocation: Option[String] = + checkpointLocation.orElse { + queryId.map { + id => + runningQueryCheckpoint(id).getOrElse { + throw new IllegalStateException( + s"Cannot find the checkpoint location of streaming query $id writing to " + + s"${originTable.name()}, which is needed to recognise a replayed micro-batch " + + s"after snapshot expiration. Set '${PaimonSink.CHECKPOINT_LOCATION}' as an " + + "option of the writer.") + } + } + } + + /** + * The checkpoint root Spark resolved for the running query. `StreamingQueryWrapper` and + * `StreamExecution` are internal and not in the same package across the supported Spark versions, + * so their public accessors are called by name. + */ + private def runningQueryCheckpoint(id: String): Option[String] = + Try { + val query = sqlContext.sparkSession.streams.get(UUID.fromString(id)) + val execution = query.getClass.getMethod("streamingQuery").invoke(query) + execution.getClass.getMethod("resolvedCheckpointRoot").invoke(execution).asInstanceOf[String] + }.toOption.filter(location => location != null && location.nonEmpty) + + private var closeRegistered = false override def addBatch(batchId: Long, data: DataFrame): Unit = { val saveMode = if (outputMode == OutputMode.Complete()) { @@ -44,7 +213,75 @@ class PaimonSink( InsertInto } val newData = PaimonUtils.createNewDataFrame(data) - WriteIntoPaimonTable(originTable, saveMode, newData, options, Some(batchId)).run( - sqlContext.sparkSession) + val closeAfterBatch = !registerCloseOnTermination() + try { + WriteIntoPaimonTable( + originTable, + saveMode, + newData, + options, + Some(StreamingWrite(openContext(), batchId))) + .run(sqlContext.sparkSession) + } finally { + // Outside a stream execution nothing ever terminates the query, so the committer of a batch + // cannot be kept for the next one. + if (closeAfterBatch) { + closeContext() + } + } } + + /** + * Closes the committer of this run when the query terminates, whether it completes, fails or is + * stopped. Returns whether the query is one whose termination Spark reports; it is not when + * [[addBatch]] is called outside a stream execution. + * + * A restarted query keeps its id and gets a new run id, and Spark reports terminations + * asynchronously, so the termination of the previous run may only be delivered once this run is + * writing. The listener therefore waits for the termination of this run only. + */ + private def registerCloseOnTermination(): Boolean = synchronized { + if (closeRegistered) { + return true + } + val streams = sqlContext.sparkSession.streams + // Spark's own query id is a uuid; anything else cannot be matched against a termination. + val run = for { + id <- queryId.flatMap(id => Try(UUID.fromString(id)).toOption) + query <- Option(streams.get(id)) + } yield (id, query.runId) + run match { + case Some((id, runId)) => + streams.addListener(new StreamingQueryListener { + override def onQueryStarted(event: StreamingQueryListener.QueryStartedEvent): Unit = {} + + override def onQueryProgress(event: StreamingQueryListener.QueryProgressEvent): Unit = {} + + override def onQueryTerminated( + event: StreamingQueryListener.QueryTerminatedEvent): Unit = { + if (event.id == id && event.runId == runId) { + try { + closeContext() + } finally { + streams.removeListener(this) + } + } + } + }) + closeRegistered = true + true + case None => false + } + } +} + +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-common/src/main/scala/org/apache/paimon/spark/write/CommitMarker.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/CommitMarker.scala new file mode 100644 index 000000000000..ea27d836d632 --- /dev/null +++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/CommitMarker.scala @@ -0,0 +1,120 @@ +/* + * 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.write + +import org.apache.hadoop.conf.Configuration +import org.apache.hadoop.fs.Path + +import java.io.IOException +import java.nio.charset.StandardCharsets.UTF_8 +import java.util.UUID + +import scala.io.Source +import scala.util.Try + +/** + * What a streaming query was about to commit, kept under its checkpoint location. + * + * A replayed micro-batch is recognised by the snapshots of its commit user, which snapshot + * expiration eventually removes. Before committing a micro-batch the sink records its batch id and + * the latest snapshot of the table at that moment: whatever that commit produced is newer than that + * snapshot, so as long as the table still retains the snapshot after it, a lookup of the commit + * user sees everything the micro-batch may have committed. Once expiration has gone past it, the + * lookup can no longer tell, and the replay must not be committed blindly. + * + * A marker is published before committing and is never replaced by a retry of the same batch. Only + * a later batch (or a new query) replaces it, after writing its complete contents to a temp file. A + * missing marker therefore means this batch never reached its commit. + */ +class CommitMarker(checkpointLocation: String, hadoopConf: Configuration) { + + val path: Path = new Path(new Path(checkpointLocation, "paimon"), "commit-marker") + + def write(commitUser: String, batchId: Long, latestSnapshotId: Long): Unit = { + if (latestSnapshotIdBefore(commitUser, batchId).isDefined) { + return + } + val fs = path.getFileSystem(hadoopConf) + val temporary = new Path(path.getParent, s".commit-marker-${UUID.randomUUID()}") + try { + val out = fs.create(temporary, false) + try { + out.write(s"$commitUser\n$batchId\n$latestSnapshotId\n".getBytes(UTF_8)) + } finally { + out.close() + } + // Spark has only one writer per checkpoint. The previous marker belongs to a different + // batch/query, so losing it here is safe: this batch has not reached its commit yet. Never + // delete a marker of this same batch, even if its first attempt has already committed. + if (fs.exists(path) && !fs.delete(path, false)) { + throw new IOException(s"Cannot replace commit marker $path") + } + if (!fs.rename(temporary, path)) { + throw new IOException(s"Cannot publish commit marker $path") + } + } finally { + fs.delete(temporary, false) + } + } + + /** + * The boundary of the first attempt of `batchId` of `commitUser`. A marker of another batch or + * query says nothing about this one. A corrupt marker, however, might have been truncated by an + * older sink retrying an already committed batch, so it must not be treated as a missing marker. + */ + def latestSnapshotIdBefore(commitUser: String, batchId: Long): Option[Long] = + read().map(lines => parse(lines).getOrElse(throw corrupt())).collect { + case (user, batch, snapshot) if user == commitUser && batch == batchId => snapshot + } + + /** + * The commit user of the micro-batch the marker was published for. A corrupt marker names none; + * the commit of a batch refuses it. + */ + def commitUser: Option[String] = read().flatMap(parse).map(_._1) + + private def read(): Option[List[String]] = { + val fs = path.getFileSystem(hadoopConf) + if (!fs.exists(path)) { + return None + } + val in = fs.open(path) + try { + Some(Source.fromInputStream(in, UTF_8.name()).getLines().toList) + } finally { + in.close() + } + } + + private def parse(lines: List[String]): Option[(String, Long, Long)] = + lines match { + case user :: batch :: snapshot :: Nil => + for { + id <- Try(batch.toLong).toOption.filter(_ >= 0) + before <- Try(snapshot.toLong).toOption.filter(_ >= 0) + if user.nonEmpty + } yield (user, id, before) + case _ => None + } + + private def corrupt(): IllegalStateException = + new IllegalStateException( + s"Cannot read commit marker $path. It may belong to an already committed micro-batch; " + + "restore a valid marker or verify the table before deleting it and restarting.") +} 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..3676201d5f1e 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 @@ -110,18 +110,22 @@ trait InnerTableV2DataWrite extends InnerTableDataWrite[InternalRow] with V2Data abstract class abstractInnerTableDataWrite[T] extends InnerTableDataWrite[T] with Logging { - val write: BatchTableWrite + val write: TableWrite 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..410953766731 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 @@ -30,15 +30,24 @@ import org.apache.spark.sql.Row import scala.collection.JavaConverters._ +/** + * @param writeBuilder + * a batch write builder, or the stream write builder of a streaming query. + * @param ignorePreviousFiles + * whether the write may ignore the files it would otherwise have to read first, because the + * commit overwrites them. A batch write builder decides this itself for an overwrite; a stream + * write builder does not know about overwrites. + */ case class PaimonDataWrite( - writeBuilder: BatchWriteBuilder, + writeBuilder: WriteBuilder, writeType: RowType, rowKindColIdx: Int = -1, writeRowTracking: Boolean = false, fullCompactionDeltaCommits: Option[Int], - batchId: Option[Long], + commitIdentifier: Option[Long], uriReaderFactory: UriReaderFactory, - postponePartitionBucketComputer: Option[BinaryRow => Integer]) + postponePartitionBucketComputer: Option[BinaryRow => Integer], + ignorePreviousFiles: Boolean = false) extends abstractInnerTableDataWrite[Row] with InnerTableV1DataWrite { @@ -50,6 +59,9 @@ case class PaimonDataWrite( if (writeRowTracking) { _write.withWriteType(writeType) } + if (ignorePreviousFiles) { + _write.withIgnorePreviousFiles(true) + } _write } @@ -72,7 +84,13 @@ case class PaimonDataWrite( } override def commitImpl(): Seq[CommitMessage] = { - write.prepareCommit().asScala.toSeq + val messages = commitIdentifier match { + // A Spark task does not outlive its micro-batch, so unlike a Flink writer it cannot leave a + // compaction running to be collected at the next commit. + case Some(identifier) => write.prepareCommit(true, identifier) + case None => write.prepareCommit() + } + messages.asScala.toSeq } override def close(): Unit = { 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-common/src/main/scala/org/apache/paimon/spark/write/StreamingWriteContext.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/StreamingWriteContext.scala new file mode 100644 index 000000000000..8933533f7a9a --- /dev/null +++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/StreamingWriteContext.scala @@ -0,0 +1,209 @@ +/* + * 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.write + +import org.apache.paimon.io.{CompactIncrement, DataIncrement} +import org.apache.paimon.table.FileStoreTable +import org.apache.paimon.table.sink.{CommitMessage, CommitMessageImpl, InnerTableCommit} + +import org.apache.spark.internal.Logging + +import java.util.Collections + +import scala.collection.JavaConverters._ + +/** + * The committer of one run of a streaming query. It is created with the first micro-batch of the + * run and kept until the query terminates, the way a Flink committer lives across checkpoints: the + * maintenance a commit starts (tag creation, partition and snapshot expiration) can run on while + * the next micro-batch is written, and only a micro-batch that may be a replay has to look up what + * a previous run committed. + * + * @param commitUser + * the commit user of the query, stable across its runs, under which every micro-batch is + * committed so that a replayed one can be recognised. + * @param marker + * where the query records what it is about to commit, when it has a checkpoint location to keep + * it in. + */ +class StreamingWriteContext(val commitUser: String, @transient marker: Option[CommitMarker]) + extends AutoCloseable + with Serializable + with Logging { + + /** Driver-side state; the context itself travels with the write closures. */ + @transient private var committer: InnerTableCommit = _ + + /** The identifier of the last micro-batch this run committed, none before the first one. */ + @transient private var lastCommitted: Option[Long] = None + + /** The committer of the run, created by `create` on first use and kept afterwards. */ + def openCommitter(create: => InnerTableCommit): InnerTableCommit = + synchronized { + if (committer == null) { + committer = create + } + committer + } + + /** + * Commits a micro-batch under its identifier. + * + * Structured Streaming replays a micro-batch with its original batch id when a query is restarted + * after failing between the sink returning and Spark recording the batch as completed, so the + * first micro-batch of a run may already have been committed by a previous run. A batch that + * follows one this run committed cannot be a replay of it, and is committed without looking the + * history of the commit user up. + * + * An overwrite, the commit of a query in complete mode, publishes the complete result of its + * micro-batch: committing a replay of it again leaves the table as its first commit did, and runs + * the callbacks of an overwrite again, as the cleanup of the snapshot branch of a chain table + * needs. Every micro-batch of such a query is committed, as it always was. + */ + def commit( + write: StreamingWrite, + messages: Seq[CommitMessage], + table: FileStoreTable, + overwrite: Boolean): Unit = + synchronized { + require(committer != null, "The committer of the streaming query has not been created.") + val identifier = write.commitIdentifier + if (overwrite) { + committer.commit(identifier, messages.asJava) + } else if (lastCommitted.exists(identifier > _)) { + writeMarker(write, table) + committer.commit(identifier, messages.asJava) + } else { + commitFirstOfRun(write, messages, table) + } + lastCommitted = Some(lastCommitted.fold(identifier)(Math.max(_, identifier))) + } + + private def writeMarker(write: StreamingWrite, table: FileStoreTable): Unit = + marker.foreach( + _.write( + commitUser, + write.batchId, + Option(table.snapshotManager().latestSnapshotId()).map(_.longValue()).getOrElse(0L))) + + /** + * The first micro-batch of a run, which may be a replay. Whether its commit user already + * committed it is looked up here first, and only then is the lookup validated against snapshot + * expiration, so that the validation covers what the lookup saw. + * + * A micro-batch that was not committed is committed directly. One that was is not committed + * again, but is handed to the commit callbacks, through the committer's filtered commit, to retry + * what may have failed after its first commit, such as registering its partitions. It is handed + * over without its files: should expiration remove the first commit between this lookup and the + * one of the filtered commit, what gets published is an empty commit, not the batch a second + * time. + */ + private def commitFirstOfRun( + write: StreamingWrite, + messages: Seq[CommitMessage], + table: FileStoreTable): Unit = { + val identifier = write.commitIdentifier + val previousAttempt = marker.flatMap(_.latestSnapshotIdBefore(commitUser, write.batchId)) + writeMarker(write, table) + val latestOfUser = table.snapshotManager().latestSnapshotOfUser(commitUser) + if (latestOfUser.isPresent && latestOfUser.get().commitIdentifier() >= identifier) { + committer.filterAndCommit( + Collections.singletonMap(Long.box(identifier), withoutFiles(messages).asJava)) + // Expected of a replay; the only trace of a misconfigured identity otherwise. + logInfo( + s"Micro-batch $identifier was already committed to ${table.name()} under commit user " + + s"'$commitUser' and is skipped as a replay.") + } else { + checkReplayIsRecognisable(previousAttempt, identifier, table) + committer.commit(identifier, messages.asJava) + } + } + + /** The partitions and buckets a micro-batch wrote to, without anything it wrote there. */ + private def withoutFiles(messages: Seq[CommitMessage]): Seq[CommitMessage] = + messages.map( + message => + new CommitMessageImpl( + message.partition(), + message.bucket(), + message.totalBuckets(), + DataIncrement.emptyIncrement(), + CompactIncrement.emptyIncrement())) + + /** + * Validates a lookup that found no commit of this micro-batch. A micro-batch whose previous + * attempt left no marker never reached its commit. One that did was committed, if at all, after + * the snapshot its marker recorded; while the snapshot after that one is still retained, every + * commit since was retained when the commit user was looked up, as expiration only removes the + * oldest snapshots, so the lookup is exact. Otherwise the lookup may have missed an expired + * commit, and committing again could write the micro-batch twice. + */ + private def checkReplayIsRecognisable( + previousAttempt: Option[Long], + identifier: Long, + table: FileStoreTable): Unit = { + val snapshots = table.snapshotManager() + previousAttempt.foreach { + before => + // Check the file itself: expiration deletes snapshots before updating the earliest hint. + if ( + Option(snapshots.latestSnapshotId()).exists(_.longValue() > before) && + !snapshots.snapshotExists(before + 1) + ) { + throw new IllegalStateException( + s"Cannot tell whether micro-batch $identifier of this query was committed to " + + s"${table.name()}. A previous attempt of it was about to commit when the latest " + + s"snapshot was $before, but snapshot ${before + 1} is no longer retained, so " + + s"the lookup of commit user '$commitUser' may have missed its commit. " + + "Committing it again could write it twice. Retain snapshots for longer than a " + + "query may be down. If the table shows that the micro-batch was not committed, " + + s"delete ${marker.get.path} and restart the query to commit it; otherwise restart " + + "the query from a new checkpoint.") + } + } + } + + override def close(): Unit = + synchronized { + if (committer != null) { + val toClose = committer + committer = null + toClose.close() + } + } +} + +/** + * One micro-batch of a streaming query, written and committed through the context of the query's + * current run. + * + * @param batchId + * the id Spark gives the micro-batch, counted from 0. + */ +case class StreamingWrite(context: StreamingWriteContext, batchId: Long) { + + /** + * 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. + */ + val commitIdentifier: Long = batchId + 1 +} 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..881cd8c64564 --- /dev/null +++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/PaimonSinkIdempotencyTest.scala @@ -0,0 +1,1547 @@ +/* + * 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.data.{BinaryString, GenericRow} +import org.apache.paimon.manifest.ManifestCommittable +import org.apache.paimon.options.Options +import org.apache.paimon.spark.sources.PaimonSink +import org.apache.paimon.spark.write.{CommitMarker, StreamingWrite, StreamingWriteContext} +import org.apache.paimon.table.{CatalogEnvironment, FileStoreTableFactory} +import org.apache.paimon.table.sink.{CommitCallback, InnerTableCommit} + +import org.apache.hadoop.conf.Configuration +import org.apache.hadoop.fs.{Path, RawLocalFileSystem} +import org.apache.hadoop.fs.permission.FsPermission +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, StreamingQueryListener, StreamTest} + +import java.io.{File, IOException, OutputStream} +import java.util.{Collections, List => JList, Map => JMap, UUID} +import java.util.concurrent.{CountDownLatch, TimeUnit} +import java.util.concurrent.atomic.{AtomicInteger, AtomicReference} + +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 causes(e: Throwable): Seq[Throwable] = + Iterator.iterate(e)(_.getCause).takeWhile(_ != null).take(20).toSeq + + 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: a replayed micro-batch of a complete mode query leaves the table as it was") { + 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()) + + // Complete mode overwrites the table with the complete result of the micro-batch, so its + // replay is committed again, as it always was, and leaves the same rows. + checkAnswer(spark.sql("SELECT * FROM T ORDER BY city"), expected) + assert(snapshotCount("T") == snapshotsAfterFirstBatch + 1) + } + } + } + + test("Paimon Sink: complete mode replay on a postpone bucket table leaves the table as it was") { + failAfter(streamingTimeout) { + withTempDir { + checkpointDir => + // A streaming write to a postpone bucket table writes bucket -2, the way a Flink + // streaming write does: the fixed bucket paths are for a batch job that ends with its + // commit, and 'postpone.batch-write-fixed-bucket' is not for a stream. + 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()) + + // The written rows wait in bucket -2 for a compaction, so the table still reads empty. + checkAnswer(spark.sql("SELECT * FROM T"), Nil) + checkAnswer(spark.sql("SELECT count(*) FROM `T$buckets` WHERE bucket = -2"), Row(1L)) + val snapshotsAfterFirstBatch = snapshotCount("T") + assert( + latestCommitUser("T").startsWith("spark-query-"), + s"expected the streaming committer to use the commit user derived from the query id, " + + s"but got '${latestCommitUser("T")}'" + ) + + dropCommitLogEntry(checkpointPath, 0) + runToCompletion(start()) + + // The replay overwrites the table with the same complete result again. + assert(snapshotCount("T") == snapshotsAfterFirstBatch + 1) + + spark.sql("CALL sys.compact(table => 'T')") + checkAnswer( + spark.sql("SELECT * FROM T ORDER BY city"), + Row("BJ", 2L) :: Row("HZ", 1L) :: Nil) + + // A later batch is committed too. Complete mode overwrites what the compaction has just + // materialized. + val snapshotsBeforeSecondBatch = snapshotCount("T") + inputData.addData((4, "SH")) + runToCompletion(start()) + + assert( + snapshotCount("T") == snapshotsBeforeSecondBatch + 1, + s"the batch after the replay was not committed (${snapshotCount("T")} snapshots)") + spark.sql("CALL sys.compact(table => 'T')") + checkAnswer( + spark.sql("SELECT * FROM T ORDER BY city"), + Row("BJ", 2L) :: Row("HZ", 1L) :: Row("SH", 1L) :: Nil) + } + } + } + + test("Paimon Sink: a replay retries the partition registration of the batch it skips") { + 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.Append(), + 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: a replay whose commit the filtered commit misses publishes no data") { + withTempDir { + dir => + // A replayed micro-batch found committed is handed to the filtered commit only to retry + // the commit callbacks, and it looks the commit user up once more. Should expiration + // remove the first commit between the two lookups, the filtered commit takes the batch + // for a new one. A committer of another user stands in for that second lookup here: what + // it publishes must not be the batch a second time. + spark.sql("CREATE TABLE T (a INT, b STRING)") + val hadoopConf = spark.sessionState.newHadoopConf() + val marker = new CommitMarker(new File(dir, "cp").getCanonicalPath, hadoopConf) + + def attempt(committerUser: String, value: String): Unit = { + val table = loadTable("T") + val context = new StreamingWriteContext("user", Some(marker)) + val write = table.newStreamWriteBuilder().withCommitUser("user").newWrite() + try { + write.write(GenericRow.of(Integer.valueOf(1), BinaryString.fromString(value))) + val messages = write.prepareCommit(true, 6).asScala.toSeq + context.openCommitter( + table + .newStreamWriteBuilder() + .withCommitUser(committerUser) + .newCommit() + .asInstanceOf[InnerTableCommit]) + context.commit(StreamingWrite(context, 5), messages, table, overwrite = false) + } finally { + write.close() + context.close() + } + } + + attempt("user", "first") + checkAnswer(spark.sql("SELECT * FROM T"), Row(1, "first") :: Nil) + + attempt("missing-the-commit", "replay") + checkAnswer(spark.sql("SELECT * FROM T"), Row(1, "first") :: Nil) + } + } + + test("Paimon Sink: commit.user-prefix names the derived commit user") { + failAfter(streamingTimeout) { + withTempDir { + checkpointDir => + // 'commit.user-prefix' names the writers of a table, as it does for a batch write. It + // only prefixes the derived user, so the identity of the query is unchanged and a + // replayed micro-batch is still recognised. + spark.sql( + "CREATE TABLE T (a INT, b STRING) TBLPROPERTIES ('commit.user-prefix' = 'my-job')") + 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")) + + def start(): StreamingQuery = + df.writeStream + .option("checkpointLocation", checkpointPath) + .format("paimon") + .start(location) + + runToCompletion(start()) + + assert( + latestCommitUser("T").startsWith("my-job_spark-query-"), + s"expected the prefix in front of the derived user, but the commit user is " + + s"'${latestCommitUser("T")}'" + ) + + dropCommitLogEntry(checkpointPath, 0) + runToCompletion(start()) + + checkAnswer(spark.sql("SELECT * FROM T"), Row(1, "a") :: Nil) + assert( + snapshotCount("T") == 1, + s"replaying batch 0 created a second snapshot (${snapshotCount("T")} in total)") + } + } + } + + test("Paimon Sink: commit.user-prefix must not make two queries share a commit user") { + failAfter(streamingTimeout) { + withTempDir { + dir => + // The prefix belongs to every writer of the table, so what follows it still has to be + // unique to the query: two queries sharing the prefix must not drop each other's + // batches as replays. + spark.sql( + "CREATE TABLE T (a INT, b STRING) TBLPROPERTIES ('commit.user-prefix' = 'my-job')") + val location = loadTable("T").location().toString + + def runOneBatch(checkpoint: String, row: (Int, String)): Unit = { + val input = MemoryStream[(Int, String)] + input.addData(row) + runToCompletion( + input + .toDS() + .toDF("a", "b") + .writeStream + .option("checkpointLocation", new File(dir, checkpoint).getCanonicalPath) + .format("paimon") + .start(location)) + } + + runOneBatch("a", (1, "a")) + val userA = latestCommitUser("T") + runOneBatch("b", (2, "b")) + val userB = latestCommitUser("T") + + assert(userA != userB, s"both queries committed as '$userA'") + checkAnswer(spark.sql("SELECT * FROM T ORDER BY a"), Row(1, "a") :: Row(2, "b") :: Nil) + } + } + } + + /** + * Commits batch 0 under 'commit.user-prefix' = 'old-job', leaves it to be replayed, and resumes + * the checkpoint under 'new-job' with `start`, which applies the prefix it is given. + */ + private def checkPrefixChangeIsRefused( + checkpointPath: String, + start: String => StreamingQuery): Unit = { + runToCompletion(start("old-job")) + assert(latestCommitUser("T").startsWith("old-job_spark-query-")) + dropCommitLogEntry(checkpointPath, 0) + + // Under another commit user the replay of batch 0 would not be recognised. + val failure = intercept[Exception](runToCompletion(start("new-job"))) + assert( + causes(failure).exists( + e => + e.isInstanceOf[IllegalStateException] && + e.getMessage.contains("Set 'commit.user-prefix' back to 'old-job'")), + s"expected the prefix change to be refused, but got $failure" + ) + checkAnswer(spark.sql("SELECT * FROM T"), Row(1, "a") :: Nil) + assert(snapshotCount("T") == 1) + + // Under the previous prefix the replay is recognised. + runToCompletion(start("old-job")) + checkAnswer(spark.sql("SELECT * FROM T"), Row(1, "a") :: Nil) + assert( + snapshotCount("T") == 1, + s"replaying batch 0 created a second snapshot (${snapshotCount("T")} in total)") + } + + test("Paimon Sink: a checkpoint is not resumed under another table commit.user-prefix") { + 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")) + + checkPrefixChangeIsRefused( + checkpointPath, + prefix => { + spark.sql(s"ALTER TABLE T SET TBLPROPERTIES ('commit.user-prefix' = '$prefix')") + df.writeStream + .option("checkpointLocation", checkpointPath) + .format("paimon") + .start(location) + } + ) + } + } + } + + test("Paimon Sink: a checkpoint is not resumed under another write commit.user-prefix") { + 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")) + + checkPrefixChangeIsRefused( + checkpointPath, + prefix => + df.writeStream + .option("checkpointLocation", checkpointPath) + .option("commit.user-prefix", prefix) + .format("paimon") + .start(location) + ) + } + } + } + + 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: a new checkpoint must not skip its batches under a named commit user") { + failAfter(streamingTimeout) { + withTempDir { + dir => + // A new checkpoint numbers its batches from 0 again, so its identifiers collide with + // those the previous query published. Only the query id, which is new with the + // checkpoint, keeps the two apart: a name configured for the writers of the table, + // which both queries share, must not. + spark.sql("CREATE TABLE T (a INT, b STRING) TBLPROPERTIES ('commit.user-prefix' = 'job')") + val location = loadTable("T").location().toString + val checkpointDir = new File(dir, "cp") + + def runBatches(rows: (Int, String)*): Unit = { + val inputData = MemoryStream[(Int, String)] + val query = inputData + .toDS() + .toDF("a", "b") + .writeStream + .option("checkpointLocation", checkpointDir.getCanonicalPath) + // No longer an option; a leftover setting must not pin the identity either. + .option("write.stream.commit-user", "my-job") + .format("paimon") + .start(location) + try { + rows.foreach { + row => + inputData.addData(row) + query.processAllAvailable() + } + } finally { + query.stop() + } + } + + runBatches((1, "old-1"), (2, "old-2")) + val firstCommitUser = latestCommitUser("T") + deleteRecursively(checkpointDir) + runBatches((3, "new-1"), (4, "new-2"), (5, "new-3")) + + checkAnswer( + spark.sql("SELECT * FROM T ORDER BY a"), + Row(1, "old-1") :: Row(2, "old-2") :: Row(3, "new-1") :: Row(4, "new-2") :: + Row(5, "new-3") :: Nil) + assert(latestCommitUser("T") != firstCommitUser) + assert(latestCommitUser("T").startsWith("job_spark-query-")) + } + } + } + + test("Paimon Sink: a replay whose snapshot may have expired is refused, not committed twice") { + failAfter(streamingTimeout) { + withTempDir { + checkpointDir => + // A replay is recognised by the snapshots of its commit user. Once expiration has + // removed them, the sink cannot tell whether the batch was committed, and must not + // commit it again on the chance that it was not. + spark.sql( + "CREATE TABLE T (a INT, b STRING) TBLPROPERTIES (" + + "'snapshot.num-retained.min' = '1', 'snapshot.num-retained.max' = '1')") + 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, "stream")) + + def start(): StreamingQuery = + df.writeStream + .option("checkpointLocation", checkpointPath) + .format("paimon") + .start(location) + + runToCompletion(start()) + val streamSnapshot = loadTable("T").snapshotManager().latestSnapshotId() + + // Another writer commits, and its expiration removes the snapshot of the batch. + spark.sql("INSERT INTO T VALUES (2, 'batch')") + assert( + !loadTable("T").snapshotManager().snapshotExists(streamSnapshot), + "the snapshot of the streaming batch should have expired") + + // The driver failed before recording batch 0, which is replayed on restart. + dropCommitLogEntry(checkpointPath, 0) + val failure = intercept[Exception](runToCompletion(start())) + assert( + causes(failure).exists( + e => + e.getMessage != null && + e.getMessage.contains("Cannot tell whether micro-batch 1 of this query")), + s"expected the replay to be refused, but got $failure" + ) + + checkAnswer( + spark.sql("SELECT * FROM T ORDER BY a"), + Row(1, "stream") :: Row(2, "batch") :: Nil) + } + } + } + + test("Paimon Sink: a replay is refused after expiration with a session-configured checkpoint") { + failAfter(streamingTimeout) { + withTempDir { + checkpointRoot => + // Reviewer P1 on #10105: a checkpoint configured through the session and a query name + // never reaches the sink options, so the marker has to be kept in the checkpoint the + // query actually uses. + spark.sql( + "CREATE TABLE T (a INT, b STRING) TBLPROPERTIES (" + + "'snapshot.num-retained.min' = '1', 'snapshot.num-retained.max' = '1')") + val location = loadTable("T").location().toString + val queryName = "paimon_session_checkpoint" + 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, "stream")) + + def start(): StreamingQuery = + df.writeStream + .queryName(queryName) + .format("paimon") + .start(location) + + runToCompletion(start()) + assert( + new File(checkpointPath, "paimon/commit-marker").exists(), + "the marker belongs in the checkpoint the query uses") + + spark.sql("INSERT INTO T VALUES (2, 'batch')") + dropCommitLogEntry(checkpointPath, 0) + val failure = intercept[Exception](runToCompletion(start())) + assert( + causes(failure).exists( + e => + e.getMessage != null && + e.getMessage.contains("Cannot tell whether micro-batch 1 of this query")), + s"expected the replay to be refused, but got $failure" + ) + + checkAnswer( + spark.sql("SELECT * FROM T ORDER BY a"), + Row(1, "stream") :: Row(2, "batch") :: Nil) + } + } + } + } + + test("Paimon Sink: a streaming query whose checkpoint cannot be found fails closed") { + withTable("T2") { + // Inside a stream execution, but with no checkpoint in the options and no running query to + // take it from: writing without a marker could commit a replay twice, so the sink refuses. + spark.sql("CREATE TABLE T2 (a INT, b STRING)") + val sink = + new PaimonSink(spark.sqlContext, loadTable("T2"), Nil, OutputMode.Append(), new Options()) + val batch: DataFrame = Seq((1, "a")).toDF("a", "b") + + spark.sparkContext.setLocalProperty( + "sql.streaming.queryId", + java.util.UUID.randomUUID().toString) + try { + val failure = intercept[IllegalStateException](sink.addBatch(0L, batch)) + assert(failure.getMessage.contains("Cannot find the checkpoint location")) + } finally { + spark.sparkContext.setLocalProperty("sql.streaming.queryId", null) + } + assert(snapshotCount("T2") == 0) + } + } + + test("Paimon Sink: the replay check refuses only what it cannot tell") { + withTempDir { + dir => + // The first micro-batch of a run may be a replay. With no snapshot of its commit user + // left, the marker its previous attempt wrote before committing decides: no marker, or + // one from another batch, means that attempt never committed; a marker older than every + // retained snapshot means it may have, and the table cannot say. + spark.sql( + "CREATE TABLE T (a INT, b STRING) TBLPROPERTIES (" + + "'snapshot.num-retained.min' = '1', 'snapshot.num-retained.max' = '1')") + spark.sql("INSERT INTO T VALUES (0, 'seed')") + val hadoopConf = spark.sessionState.newHadoopConf() + + def attempt(name: String, prepare: CommitMarker => Unit, tableName: String = "T"): Unit = { + val table = loadTable(tableName) + val user = s"user-$name" + val marker = new CommitMarker(new File(dir, name).getCanonicalPath, hadoopConf) + prepare(marker) + val context = new StreamingWriteContext(user, Some(marker)) + val builder = table.newStreamWriteBuilder().withCommitUser(user) + val write = builder.newWrite() + try { + write.write(GenericRow.of(Integer.valueOf(1), BinaryString.fromString(name))) + val messages = write.prepareCommit(true, 6).asScala.toSeq + context.openCommitter(builder.newCommit().asInstanceOf[InnerTableCommit]) + context.commit(StreamingWrite(context, 5), messages, table, overwrite = false) + } finally { + write.close() + context.close() + } + } + + def latestId(): Long = loadTable("T").snapshotManager().latestSnapshotId() + + // No marker: the previous attempt never got to its commit. + attempt("none", _ => ()) + // A marker of another batch says nothing about this one. + attempt("other", _.write("user-other", 4, 0)) + // A marker of this batch that no expiration has gone past: a commit would be retained. + attempt("kept", m => m.write("user-kept", 5, latestId())) + assert( + spark.sql("SELECT b FROM T WHERE a = 1").collect().map(_.getString(0)).toSet == + Set("none", "other", "kept"), + "each of these attempts should have been committed" + ) + + // A marker of this batch on a table without any snapshot: nothing can have expired. + withTable("T3") { + spark.sql("CREATE TABLE T3 (a INT, b STRING)") + attempt("empty", m => m.write("user-empty", 5, 0), "T3") + checkAnswer(spark.sql("SELECT * FROM T3"), Row(1, "empty") :: Nil) + } + + // A marker of this batch older than every retained snapshot: refused. + val refused = intercept[IllegalStateException]( + attempt("expired", m => m.write("user-expired", 5, latestId() - 2))) + assert(refused.getMessage.contains("Cannot tell whether micro-batch 6 of this query")) + assert(spark.sql("SELECT * FROM T WHERE b = 'expired'").collect().isEmpty) + + // What the error says to do once the table shows the batch was not committed: delete + // the marker it names, and the next attempt commits the batch. + val named = "(/\\S+/paimon/commit-marker)".r.findFirstIn(refused.getMessage) + assert(named.isDefined, s"the error should name the marker: ${refused.getMessage}") + assert(new File(named.get).delete(), s"cannot delete ${named.get}") + attempt("expired", _ => ()) + checkAnswer(spark.sql("SELECT a FROM T WHERE b = 'expired'"), Row(1) :: Nil) + } + } + + private def commitReplayAttempt(marker: CommitMarker, batchId: Long = 0): Unit = { + val table = loadTable("T") + val user = "recovery-user" + val context = new StreamingWriteContext(user, Some(marker)) + val builder = table.newStreamWriteBuilder().withCommitUser(user) + val write = builder.newWrite() + try { + write.write(GenericRow.of(Integer.valueOf(1), BinaryString.fromString("stream"))) + val messages = write.prepareCommit(true, batchId + 1).asScala.toSeq + context.openCommitter(builder.newCommit().asInstanceOf[InnerTableCommit]) + context.commit(StreamingWrite(context, batchId), messages, table, overwrite = false) + } finally { + write.close() + context.close() + } + } + + test("Paimon Sink: repeated recovery preserves the first attempt's marker") { + withTempDir { + dir => + spark.sql( + "CREATE TABLE T (a INT, b STRING) TBLPROPERTIES (" + + "'snapshot.num-retained.min' = '1', 'snapshot.num-retained.max' = '1')") + val conf = spark.sessionState.newHadoopConf() + val marker = new CommitMarker(dir.getCanonicalPath, conf) + commitReplayAttempt(marker) + val original = loadTable("T").snapshotManager().latestSnapshotId() + val interrupted = new CommitMarker(dir.getCanonicalPath, conf) { + override def write(user: String, batch: Long, latest: Long): Unit = { + super.write(user, batch, latest) + throw new IllegalStateException("driver failed before filtering") + } + } + (1 to 2).foreach { + _ => + assert( + intercept[IllegalStateException](commitReplayAttempt(interrupted)).getMessage + .contains("driver failed")) + } + assert(marker.latestSnapshotIdBefore("recovery-user", 0).contains(0L)) + spark.sql("INSERT INTO T VALUES (2, 'other')") + assert(!loadTable("T").snapshotManager().snapshotExists(original)) + assert( + intercept[IllegalStateException](commitReplayAttempt(marker)).getMessage + .contains("Cannot tell whether")) + checkAnswer(spark.sql("SELECT * FROM T WHERE b = 'stream'"), Row(1, "stream") :: Nil) + } + } + + Seq(false, true).foreach { + staleHint => + test(s"Paimon Sink: expiration before replay filtering is refused (stale hint: $staleHint)") { + withTempDir { + dir => + spark.sql( + "CREATE TABLE T (a INT, b STRING) TBLPROPERTIES (" + + "'snapshot.num-retained.min' = '1', 'snapshot.num-retained.max' = '1')") + val conf = spark.sessionState.newHadoopConf() + val marker = new CommitMarker(dir.getCanonicalPath, conf) + commitReplayAttempt(marker) + val original = loadTable("T").snapshotManager().latestSnapshotId() + val concurrentExpiration = new CommitMarker(dir.getCanonicalPath, conf) { + override def write(user: String, batch: Long, latest: Long): Unit = { + super.write(user, batch, latest) + // Interleave another writer after the old precheck but before actual filtering. + spark.sql("INSERT INTO T VALUES (2, 'other')") + val snapshots = loadTable("T").snapshotManager() + assert(!snapshots.snapshotExists(original)) + if (staleHint) { + // Expiration removes snapshot files before advancing the earliest hint. + snapshots.commitEarliestHint(original) + } + } + } + assert( + intercept[IllegalStateException](commitReplayAttempt(concurrentExpiration)).getMessage + .contains("Cannot tell whether")) + checkAnswer(spark.sql("SELECT * FROM T WHERE b = 'stream'"), Row(1, "stream") :: Nil) + } + } + } + + test("Paimon Sink: a corrupt marker must not be treated as an uncommitted batch") { + withTempDir { + dir => + spark.sql( + "CREATE TABLE T (a INT, b STRING) TBLPROPERTIES (" + + "'snapshot.num-retained.min' = '1', 'snapshot.num-retained.max' = '1')") + val conf = spark.sessionState.newHadoopConf() + val marker = new CommitMarker(dir.getCanonicalPath, conf) + commitReplayAttempt(marker) + spark.sql("INSERT INTO T VALUES (2, 'other')") + // An older sink could truncate the committed batch's marker during recovery. + val out = marker.path.getFileSystem(conf).create(marker.path, true) + out.write("recovery-user\n0".getBytes("UTF-8")) + out.close() + val failure = intercept[IllegalStateException](commitReplayAttempt(marker)) + assert(failure.getMessage.contains(marker.path.toString)) + checkAnswer(spark.sql("SELECT * FROM T WHERE b = 'stream'"), Row(1, "stream") :: Nil) + } + } + + test("Paimon Sink: an interrupted marker update keeps the previous batch's marker") { + withTempDir { + dir => + val conf = new Configuration(spark.sessionState.newHadoopConf()) + conf.setClass( + "fs.file.impl", + classOf[FailingMarkerFileSystem], + classOf[org.apache.hadoop.fs.FileSystem]) + conf.setBoolean("fs.file.impl.disable.cache", true) + val marker = new CommitMarker(dir.getCanonicalPath, conf) + marker.write("recovery-user", 0, 0) + FailingMarkerFileSystem.failWrites = true + try { + // A same-batch retry must not write, even when the filesystem rejects new writes. + marker.write("recovery-user", 0, 1) + intercept[IOException](marker.write("recovery-user", 1, 2)) + } finally { + FailingMarkerFileSystem.failWrites = false + } + assert(marker.latestSnapshotIdBefore("recovery-user", 0).contains(0L)) + marker.write("recovery-user", 1, 2) + assert(marker.latestSnapshotIdBefore("recovery-user", 1).contains(2L)) + // Recreating the checkpoint gives the same batch number a different identity. + marker.write("another-query", 0, 3) + assert(marker.latestSnapshotIdBefore("another-query", 0).contains(3L)) + } + } + + // Every kind of table the sink writes to: the write and commit paths differ by bucket mode and + // by what a commit also produces (compactions, deletion vectors, changelog, row ids). + Seq( + ("dynamic bucket", "'primary-key' = 'k,pt', 'bucket' = '-1'", false), + ("cross partition", "'primary-key' = 'k', 'bucket' = '-1'", false), + ( + "postpone bucket with real buckets", + "'primary-key' = 'k,pt', 'bucket' = '-2', 'postpone.default-bucket-num' = '1'", + true), + ( + "deletion vector", + "'primary-key' = 'k,pt', 'bucket' = '1', 'deletion-vectors.enabled' = 'true'", + false), + ( + "lookup changelog", + "'primary-key' = 'k,pt', 'bucket' = '1', 'changelog-producer' = 'lookup'", + false), + ("row tracking", "'row-tracking.enabled' = 'true'", false) + ).foreach { + case (kind, properties, postpone) => + test(s"Paimon Sink: a replay into a $kind table is not committed twice") { + failAfter(streamingTimeout) { + withTempDir { + checkpointDir => + spark.sql( + "CREATE TABLE T (k INT, v STRING, pt STRING) PARTITIONED BY (pt) " + + s"TBLPROPERTIES ($properties)") + val seed = if (postpone) { + // A batch write puts this into a real bucket, so the streaming batches below + // are appended to a table that already has buckets. + spark.sql("INSERT INTO T VALUES (0, 'seed', 'p1')") + Row(0, "seed", "p1") :: Nil + } else { + Nil + } + val location = loadTable("T").location().toString + val checkpointPath = checkpointDir.getCanonicalPath + + val inputData = MemoryStream[(Int, String, String)] + val df = inputData.toDS().toDF("k", "v", "pt") + def run(): Unit = + runToCompletion( + df.writeStream + .option("checkpointLocation", checkpointPath) + .format("paimon") + .start(location)) + + inputData.addData((1, "a", "p1"), (2, "b", "p2")) + run() + inputData.addData((3, "c", "p1")) + run() + if (postpone) { + checkAnswer( + spark.sql("SELECT count(*) > 0 FROM `T$buckets` WHERE bucket = -2"), + Row(true)) + } + + // A primary key table would hide a duplicate behind its merge, so what shows a + // replay committed again is a new snapshot. + val snapshots = snapshotCount("T") + dropCommitLogEntry(checkpointPath, 1) + run() + assert( + snapshotCount("T") == snapshots, + s"replaying batch 1 created ${snapshotCount("T") - snapshots} more snapshots") + + if (postpone) { + spark.sql("CALL sys.compact(table => 'T')") + } + checkAnswer( + spark.sql("SELECT k, v, pt FROM T ORDER BY k"), + seed ++ (Row(1, "a", "p1") :: Row(2, "b", "p2") :: Row(3, "c", "p1") :: Nil)) + } + } + } + } + + test("Paimon Sink: a batch that failed before its commit is committed once on restart") { + failAfter(streamingTimeout) { + withTempDir { + checkpointDir => + // The query fails while writing batch 1, before anything is committed. The run's + // committer has to be closed although the query did not stop cleanly, and the restart + // replays batch 1, which, never committed, must be committed once. + CountingCommitCallback.reset() + spark.sql( + "CREATE TABLE T (a INT, b STRING) TBLPROPERTIES (" + + s"'commit.callbacks' = '${classOf[CountingCommitCallback].getName}')") + val location = loadTable("T").location().toString + val checkpointPath = checkpointDir.getCanonicalPath + + val inputData = MemoryStream[(Int, String)] + val df = inputData + .toDS() + .map { + row => + if (row._1 == 2 && FailBeforeCommit.armed) { + throw new RuntimeException("injected failure before the commit") + } + row + } + .toDF("a", "b") + def start(): StreamingQuery = + df.writeStream + .option("checkpointLocation", checkpointPath) + .format("paimon") + .start(location) + + FailBeforeCommit.armed = true + try { + val query = start() + try { + inputData.addData((1, "a")) + query.processAllAvailable() + inputData.addData((2, "b")) + intercept[Exception](query.processAllAvailable()) + } finally { + query.stop() + } + } finally { + FailBeforeCommit.armed = false + } + assert(snapshotCount("T") == 1, "only batch 0 was committed before the failure") + // The query terminates asynchronously; bounded by the timeout of this test. + while (CountingCommitCallback.closed < CountingCommitCallback.created) { + Thread.sleep(50) + } + + runToCompletion(start()) + + checkAnswer(spark.sql("SELECT * FROM T ORDER BY a"), Row(1, "a") :: Row(2, "b") :: Nil) + assert(snapshotCount("T") == 2) + } + } + } + + test("Paimon Sink: a query upgraded from per-batch commits keeps committing") { + failAfter(streamingTimeout) { + withTempDir { + checkpointDir => + // Before this sink, every micro-batch was a batch write: a random commit user, the + // batch commit identifier, and nothing under the checkpoint. A query resuming such a + // checkpoint has a commit user with no history and no marker, on a table whose older + // snapshots may have expired; its next batch is new and must be committed. + spark.sql( + "CREATE TABLE T (a INT, b STRING) TBLPROPERTIES (" + + "'snapshot.num-retained.min' = '1', 'snapshot.num-retained.max' = '1')") + val location = loadTable("T").location().toString + val checkpointPath = checkpointDir.getCanonicalPath + + val inputData = MemoryStream[(Int, String)] + val df = inputData.toDS().toDF("a", "b") + + val before = df.writeStream + .option("checkpointLocation", checkpointPath) + .foreachBatch { + (batch: DataFrame, _: Long) => + batch.write.format("paimon").mode("append").save(location) + } + .start() + try { + inputData.addData((1, "a")) + before.processAllAvailable() + inputData.addData((2, "b")) + before.processAllAvailable() + } finally { + before.stop() + } + assert(loadTable("T").snapshotManager().earliestSnapshotId() > 1) + + inputData.addData((3, "c")) + runToCompletion( + df.writeStream + .option("checkpointLocation", checkpointPath) + .format("paimon") + .start(location)) + + checkAnswer( + spark.sql("SELECT * FROM T ORDER BY a"), + Row(1, "a") :: Row(2, "b") :: Row(3, "c") :: Nil) + assert(latestCommitUser("T").startsWith("spark-query-")) + } + } + } + + 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: maintenance of a micro-batch runs while the query is alive") { + failAfter(streamingTimeout) { + withTempDir { + checkpointDir => + // The committer of a query lives until the query terminates, so the maintenance a + // commit starts runs on while the next micro-batch is written. A committer closed after + // every micro-batch would cancel an async expiration before it ever ran. + 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() + } + // Bounded by the timeout of this test. + while (snapshotCount("T") > 1) { + Thread.sleep(50) + } + } finally { + query.stop() + } + } + } + } + + test("Paimon Sink: one committer commits every micro-batch of a query and closes with it") { + failAfter(streamingTimeout) { + withTempDir { + checkpointDir => + // The committer of a query is created with its first micro-batch and closed when the + // query terminates, the way a Flink committer lives across checkpoints, so that the + // maintenance a commit starts is not cancelled and only the first micro-batch of a run + // has to look up what a previous run committed. + CountingCommitCallback.reset() + spark.sql( + "CREATE TABLE T (a INT, b STRING) TBLPROPERTIES (" + + s"'commit.callbacks' = '${classOf[CountingCommitCallback].getName}')") + val location = loadTable("T").location().toString + + val inputData = MemoryStream[(Int, String)] + val query = inputData + .toDS() + .toDF("a", "b") + .writeStream + .option("checkpointLocation", checkpointDir.getCanonicalPath) + .format("paimon") + .start(location) + try { + for (i <- 1 to 3) { + inputData.addData((i, s"v$i")) + query.processAllAvailable() + } + assert( + CountingCommitCallback.commits == 3, + s"expected three committed micro-batches, found ${CountingCommitCallback.commits}") + assert( + CountingCommitCallback.created == 1, + s"expected one committer for the whole query, found " + + s"${CountingCommitCallback.created}" + ) + assert( + CountingCommitCallback.closed == 0, + "the committer must not be closed while the query is running") + } finally { + query.stop() + } + + // The query terminates asynchronously; bounded by the timeout of this test. + while (CountingCommitCallback.closed == 0) { + Thread.sleep(50) + } + assert(CountingCommitCallback.created == 1) + } + } + } + test("Paimon Sink: a late termination of the previous run leaves the restarted run open") { + failAfter(streamingTimeout) { + withTempDir { + checkpointDir => + // A restarted query keeps its id, and Spark delivers terminations asynchronously: the + // termination of the previous run may arrive while the restarted run is writing. It + // must close the committer of the previous run only. + CountingCommitCallback.reset() + spark.sql( + "CREATE TABLE T (a INT, b STRING) TBLPROPERTIES (" + + s"'commit.callbacks' = '${classOf[CountingCommitCallback].getName}')") + val location = loadTable("T").location().toString + + val inputData = MemoryStream[(Int, String)] + def start(): StreamingQuery = + inputData + .toDS() + .toDF("a", "b") + .writeStream + .option("checkpointLocation", checkpointDir.getCanonicalPath) + .format("paimon") + .start(location) + + // Holds back a progress event of the first run, and with it the termination of that run + // that Spark queues behind it. + val firstRun = new AtomicReference[UUID]() + val release = new CountDownLatch(1) + val blocker = new StreamingQueryListener { + override def onQueryStarted(event: StreamingQueryListener.QueryStartedEvent): Unit = {} + + override def onQueryProgress(event: StreamingQueryListener.QueryProgressEvent): Unit = { + if (event.progress.runId == firstRun.get()) { + release.await(streamingTimeout.toMillis, TimeUnit.MILLISECONDS) + } + } + + override def onQueryTerminated( + event: StreamingQueryListener.QueryTerminatedEvent): Unit = {} + } + // Registered after the sink listeners, so it sees a termination once they have. + val delivered = Collections.synchronizedSet(new java.util.HashSet[UUID]()) + val observer = new StreamingQueryListener { + override def onQueryStarted(event: StreamingQueryListener.QueryStartedEvent): Unit = {} + + override def onQueryProgress( + event: StreamingQueryListener.QueryProgressEvent): Unit = {} + + override def onQueryTerminated( + event: StreamingQueryListener.QueryTerminatedEvent): Unit = { + delivered.add(event.runId) + } + } + def awaitDelivered(runId: UUID): Unit = { + // Bounded by the timeout of this test. + while (!delivered.contains(runId)) { + Thread.sleep(50) + } + } + + spark.streams.addListener(blocker) + try { + val first = start() + firstRun.set(first.runId) + inputData.addData((1, "a")) + first.processAllAvailable() + first.stop() + + val restarted = start() + try { + assert(restarted.id == first.id && restarted.runId != first.runId) + inputData.addData((2, "b")) + restarted.processAllAvailable() + spark.streams.addListener(observer) + + release.countDown() + awaitDelivered(first.runId) + assert(restarted.isActive) + assert( + CountingCommitCallback.created == 2 && CountingCommitCallback.closed == 1, + "the termination of the first run must close its own committer only, but " + + s"${CountingCommitCallback.closed} of ${CountingCommitCallback.created} " + + "committers are closed" + ) + + inputData.addData((3, "c")) + restarted.processAllAvailable() + } finally { + restarted.stop() + } + awaitDelivered(restarted.runId) + assert( + CountingCommitCallback.created == 2 && CountingCommitCallback.closed == 2, + s"${CountingCommitCallback.closed} of ${CountingCommitCallback.created} " + + "committers are closed after both runs terminated" + ) + checkAnswer( + spark.sql("SELECT * FROM T ORDER BY a"), + Row(1, "a") :: Row(2, "b") :: Row(3, "c") :: Nil) + } finally { + release.countDown() + spark.streams.removeListener(blocker) + spark.streams.removeListener(observer) + } + } + } + } +} + +/** Counts the committers of a table, and what they commit. */ +private[spark] class CountingCommitCallback extends CommitCallback { + + CountingCommitCallback.createdCount.incrementAndGet() + + override def call(context: CommitCallback.Context): Unit = { + CountingCommitCallback.commitCount.incrementAndGet() + } + + override def retry(committable: ManifestCommittable): Unit = {} + + override def close(): Unit = { + CountingCommitCallback.closedCount.incrementAndGet() + } +} + +private[spark] object CountingCommitCallback { + + private[spark] val createdCount = new AtomicInteger(0) + private[spark] val commitCount = new AtomicInteger(0) + private[spark] val closedCount = new AtomicInteger(0) + + def reset(): Unit = { + createdCount.set(0) + commitCount.set(0) + closedCount.set(0) + } + + def created: Int = createdCount.get() + + def commits: Int = commitCount.get() + + def closed: Int = closedCount.get() +} + +/** Makes a streaming write fail while writing, before anything is committed. */ +private[spark] object FailBeforeCommit { + @volatile var armed: Boolean = false +} + +/** 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 = {} + } +} + +/** Fails after opening and partially writing a marker, as a storage failure can. */ +private[spark] class FailingMarkerFileSystem extends RawLocalFileSystem { + override protected def createOutputStreamWithMode( + path: Path, + append: Boolean, + permission: FsPermission): OutputStream = { + val out = super.createOutputStreamWithMode(path, append, permission) + if (FailingMarkerFileSystem.failWrites) { + out.write("partial".getBytes("UTF-8")) + out.close() + throw new IOException("marker write interrupted") + } + out + } +} + +private[spark] object FailingMarkerFileSystem { + @volatile var failWrites: Boolean = false +} 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 206c2c0bcff5..6e5edd201e5a 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 @@ -369,6 +369,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 {