zhuxiangyi commented on code in PR #9667:
URL: https://github.com/apache/paimon/pull/9667#discussion_r3966085666
##########
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:
Confirmed both, and reproduced both before changing anything.
A fresh query at a reused location: the first query wrote `[1,old]` under
`spark-checkpoint-b492463a…`; after deleting the checkpoint, a new query
writing `(2,"new")` derived the identical user, and the table still held only
`[1,old]` — the new row was skipped as an already committed replay, with the
query reporting success. It is not limited to the first batch: the comparison
is against the previous run’\s last committed identifier, so a new query
silently loses every batch up to that id.
The trailing separator: `spark-checkpoint-ae4544b6…` became
`spark-checkpoint-9ff6c207…` for the same checkpoint, and the replayed batch
produced two rows.
The underlying mistake was binding the identity to where the checkpoint is
stored rather than to which incarnation of it is running, so it could come
apart in both directions. The failure modes are asymmetric — a fresh identity
for the same query duplicates, a stale identity for a new query loses data —
which is the stronger argument for your ordering, so it is now: explicit
override, then the persisted query id, then the location, which is only
reachable outside a stream execution (a direct `addBatch` call has no query id).
Both cases you named are covered: "a new query reusing a checkpoint location
must not skip its batches" and "an equivalent spelling of the checkpoint
location keeps the identity", asserting the commit user differs in the first
and is unchanged in the second.
Since the query id is now the default identity rather than a fallback, I
also ran the suite on the 3.2, 3.4, 3.5, 4.0 and 4.1 modules to confirm the
property is populated on every supported version. That surfaced an unrelated
problem of my own: the suite left a second table behind, which Spark 3.2 cannot
drop because its `dropNamespace` has no cascade overload. Fixed with
`withTable`.
--
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]
