zhuxiangyi commented on code in PR #9667:
URL: https://github.com/apache/paimon/pull/9667#discussion_r4052067027
##########
paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/sources/PaimonSink.scala:
##########
@@ -19,23 +19,95 @@
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.
+ *
+ * What the commit user has to identify is one incarnation of a checkpoint,
not the place it is
+ * stored. Paimon skips a batch whose id a previous run committed under the
same user, so reusing
+ * a user across two different queries drops the data of the second one,
while changing it within
+ * one query brings back the duplicate. The query id Spark persists in the
checkpoint metadata is
+ * exactly that identity: it is new when a checkpoint is recreated,
unchanged when a query resumes
+ * from one, and independent of how the location is spelled.
+ *
+ * Resolved lazily: neither the query id nor the checkpoint location is
available on the thread
+ * that constructs the sink.
+ */
+ private lazy val commitUser: String = {
+ configuredCommitUser.getOrElse {
+ queryId
+ .map(derivedCommitUser("query", _))
+ // Only reachable outside a stream execution, e.g. a direct addBatch
call. A location
+ // cannot tell a recreated checkpoint from a resumed one, so it is a
last resort.
+ .orElse(checkpointLocation.map(derivedCommitUser("checkpoint", _)))
+ .getOrElse {
+ logWarning(
+ "This streaming write has neither a query id nor a checkpoint
location to derive a " +
+ "stable commit user from, so a replayed micro-batch cannot be
recognised and may " +
+ s"be committed twice. Set
'${SparkConnectorOptions.STREAM_WRITE_COMMIT_USER.key}' " +
+ "to make the write idempotent.")
+ UUID.randomUUID().toString
+ }
+ }
+ }
+
+ // Like the read side, which takes its 'read.stream.*' options from the
table, so that a
+ // 'spark.paimon.<key>' session conf works the same as an option of the
writer.
+ private def configuredCommitUser: Option[String] = {
+ val fromWriter =
options.get(SparkConnectorOptions.STREAM_WRITE_COMMIT_USER)
+ val fromTable =
+
Options.fromMap(originTable.options()).get(SparkConnectorOptions.STREAM_WRITE_COMMIT_USER)
Review Comment:
Confirmed and reproduced: with the user set as a table property, a second
query's batch was dropped as a replay and its data lost. Fixed in e34909848:
the user is read from the writer option and the session conf only, a table
property is ignored with a warning, and a micro-batch skipped as a replay is
logged. Description and guide now require the user to be unique to one writer.
Also rebased onto master and renamed the history-bound switch after #9960.
--
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]
