JingsongLi commented on code in PR #9667:
URL: https://github.com/apache/paimon/pull/9667#discussion_r3956402943
##########
paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/sources/PaimonSink.scala:
##########
@@ -19,23 +19,88 @@
package org.apache.paimon.spark.sources
import org.apache.paimon.options.Options
-import org.apache.paimon.spark.{InsertInto, Overwrite}
+import org.apache.paimon.spark.{InsertInto, Overwrite, SparkConnectorOptions}
import org.apache.paimon.spark.commands.{SchemaEvolutionHelper,
WriteIntoPaimonTable}
import org.apache.paimon.table.FileStoreTable
+import org.apache.spark.internal.Logging
import org.apache.spark.sql.{DataFrame, PaimonUtils, SQLContext}
import org.apache.spark.sql.execution.streaming.Sink
import org.apache.spark.sql.sources.AlwaysTrue
import org.apache.spark.sql.streaming.OutputMode
+import java.nio.charset.StandardCharsets.UTF_8
+import java.util.UUID
+
+import scala.collection.JavaConverters._
+
class PaimonSink(
sqlContext: SQLContext,
override val originTable: FileStoreTable,
partitionColumns: Seq[String],
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.
+ *
+ * Resolved lazily: neither the checkpoint location nor the query id is
available on the thread
+ * that constructs the sink.
+ */
+ private lazy val commitUser: String = {
+ configuredCommitUser.getOrElse {
+ checkpointLocation
+ .map(derivedCommitUser("checkpoint", _))
+ .orElse(queryId.map(derivedCommitUser("query", _)))
Review Comment:
[P1] Prefer the persisted query ID over the checkpoint path
If a checkpoint is deleted and a new query starts at the same path, Spark
assigns a new query ID and restarts batch IDs at 0. This code nevertheless
reuses the previous commit user, so `filterCommitted` silently skips fresh
batches whose IDs are at or below the previous query's last committed ID. I
reproduced this with different input in the new query: the query completed
successfully, but the table contained only the old row instead of both rows.
The reverse also fails: restarting the same checkpoint with only a trailing
`/` added to its path changes the commit user despite an unchanged query ID.
Replaying batch 0 then produced two rows instead of one.
Please keep the explicit override, but prefer the persisted Spark query ID
for the default identity and use the checkpoint path only as a fallback. Add
coverage for both a fresh query reusing a checkpoint path and an existing query
using an equivalent path spelling.
##########
paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala:
##########
@@ -472,14 +498,29 @@ case class PaimonSparkWriter(
postCommit(finalMessages)
return
}
- val activeWriteBuilder =
- Option(directPostponeWriteBuilder).getOrElse(writeBuilder)
- val tableCommit = activeWriteBuilder.newCommit()
+ val tableCommit: InnerTableCommit =
+ if (directPostponeWriteBuilder != null) {
+ directPostponeWriteBuilder.newCommit()
+ } else {
+ writeBuilder.asInstanceOf[BatchWriteBuilderImpl].newCommit()
+ }
if (operation != null) {
tableCommit.withOperation(operation)
}
try {
- tableCommit.commit(commitMessages.toList.asJava)
+ idempotentCommitIdentifier match {
+ case Some(identifier) =>
+ // Structured Streaming replays a micro-batch with its original
batch id after a failure.
+ // Committing under a stable commit user lets Paimon skip a replay
it already committed,
+ // instead of duplicating the whole batch. The files being committed
were written by this
+ // very batch, so there is no need to list them to prove that they
still exist.
+ tableCommit
+ .checkFilesExistence(false)
+ .filterAndCommit(
+ Collections.singletonMap(Long.box(identifier),
commitMessages.toList.asJava))
Review Comment:
[P2] Preserve the batch maintenance lifecycle when filtering commits
Unlike `commit(List)`, `filterAndCommit` does not set
`TableCommitImpl.batchCommitted`. Maintenance therefore runs through the
streaming executor wrapper, but this writer still closes and discards the
committer immediately after each batch. With
`snapshot.expire.execution-mode=async`, `close()` calls `shutdownNow()` and can
interrupt snapshot expiration before it finishes. Even with the default
synchronous mode, the wrapper catches maintenance exceptions and stores them
for the next commit; because this instance is discarded, those failures are
never propagated to the caller.
A focused probe against the built classes reproduced both the asynchronous
interruption and the loss of synchronous error propagation. The existing
`testBatchWriteAsyncExpireFallbackToSync` also establishes that a batch
committer must finish maintenance before closing.
Please preserve the one-shot batch maintenance semantics in the filtered
commit path, or explicitly wait for maintenance and propagate its failure
before closing the committer.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
