JingsongLi commented on code in PR #9667:
URL: https://github.com/apache/paimon/pull/9667#discussion_r4051926683
##########
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:
[P1] Don't let a table property pin the streaming commit user
`configuredCommitUser` also reads `write.stream.commit-user` from
`originTable.options()`, so the identity can be set in `TBLPROPERTIES`, the
channel the connector documents for persistent write options
(`docs/docs/spark/configuration.mdx`). But this option is not a preference, it
is a writer identity, and a table-wide identity is not per query:
`filterCommitted` drops every committable whose identifier is not greater than
the latest one published under the same user, and it drops it silently.
Trigger, traced through the three paths involved (code-level trace, not a
runtime run): table `T` with `TBLPROPERTIES ('write.stream.commit-user' =
'job')`. Streaming query A commits up to identifier 100. A second streaming
query B on the same table starts at batch 0, so every committable with
identifier 1..100 is filtered out by `FileStoreCommitImpl.filterCommitted`, and
since both queries advance one identifier per batch while A keeps committing, B
never catches up and none of its data is ever written. Nothing is logged, no
exception is raised, `PaimonSparkWriter.commit` returns normally, Spark marks
the batch completed and never retries it, and the files B wrote stay
uncommitted until orphan cleaning. From the outside this is indistinguishable
from a genuine replay, although B's batch holds entirely new data: the
identifier comparison is content-blind, so skipping is only correct when the
new run re-reads exactly what the previous run committed.
The four-spark guide this PR adds advertises only two channels, the writer
option and the `spark.paimon.write.stream.commit-user` session conf, and both
are per query; the selected option's description states no scope. The read-side
analogy in the comment on this method (`read.stream.*` taken from the table)
does not carry over to a writer identity, whose whole point is to be unique per
streaming writer, and the existing table-level knob for commit identity,
`commit.user-prefix`, deliberately keeps a runtime-unique suffix so that a
table property cannot alias two writers.
Fix: take the option from the per-query sources only, i.e. drop the
`originTable.options()` lookup; if the table scope is really intended, then the
option description and the guide should require the user to be unique to one
streaming writer, and a committable dropped because its identifier was already
published should at least be logged so the loss is diagnosable.
--
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]
