JingsongLi commented on code in PR #9667:
URL: https://github.com/apache/paimon/pull/9667#discussion_r3999236946
##########
paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala:
##########
@@ -472,14 +499,38 @@ 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, while still retrying the
commit callbacks
+ // that may have failed after the snapshot was published. Either
builder was given the
+ // stable user, so its lookup of the previous commit is not bounded
by strict mode.
+ //
+ // The files being committed were written by this very batch, so
there is no need to
+ // list them to prove that they still exist, nor to scan the base
files they could
+ // conflict with. And this committer is closed right after the
batch, so maintenance
+ // cannot be left to an executor that is about to be shut down, nor
a failure to a
+ // commit that never comes.
+ tableCommit
+ .checkFilesExistence(false)
+ .checkAppendFiles(false)
+ .inlineMaintenance(true)
+ .filterAndCommit(
+ Collections.singletonMap(Long.box(identifier),
commitMessages.toList.asJava))
Review Comment:
[P2] Recover chain-table overwrite cleanup before acknowledging a replay
For a Complete-mode query writing a chain table's delta branch, the delta
OVERWRITE snapshot can be published before a transient I/O failure prevents
`ChainTableOverwriteCommitCallback.call` from clearing the corresponding
snapshot-branch partitions. On replay, this filtered commit recognizes the
published batch and invokes `ChainTableOverwriteCommitCallback.retry`, which is
a no-op. The batch then returns successfully while chain-table reads still
prefer the old snapshot data over the new delta data.
The empty `retry` method predates this PR, but switching the Spark sink to
filtered replay introduces this recovery regression: the previous
`commit(List)` path would commit the replayed overwrite and invoke `call`
again, completing cleanup after the transient failure. This issue dates to the
initial commit-path change in this PR.
I reproduced the difference on real local core tables at this revision,
comparing the previous batch-commit sequence with the new filtered sequence and
injecting a temporary snapshot-branch read failure after delta publication.
After successful recovery, the previous path returned `[2:value-2]` from the
main chain table; the filtered path returned `accepted=0` and still read
`[1:value-1]`. The probe exercised core commit/read paths; Spark Complete-mode
reachability was verified by source tracing, rather than a full Spark
fault-injection test.
Please make chain overwrite recovery complete the missing cleanup before
acknowledging the replay. Resolve the original committed overwrite snapshot and
its manifest changes, including partitions containing only removed files, and
add a failure/replay regression test.
--
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]
