zhuxiangyi commented on code in PR #9667:
URL: https://github.com/apache/paimon/pull/9667#discussion_r4000068987
##########
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:
Confirmed, and reproduced first at the core level: after an overwrite of the
delta branch whose cleanup failed once the snapshot was published, the replay
was recognised — no second snapshot — but the chain table still read `value-1`
from the snapshot branch. The empty `retry` predates this PR, as you say, and
it was empty on purpose: its comment relied on the replay being committed
again. Recognising the replay removed that.
Fixed in bb81c169d, along the lines of `IcebergCommitCallback.retry`: the
callback now gets its commit user, and on retry resolves the overwrite snapshot
of the identifier with `findSnapshotsForIdentifiers` and redoes the cleanup
from the delta manifest changes of that snapshot, so partitions the overwrite
only removed files from are covered too. Truncating the partitions again is
idempotent. One choice worth your look: a snapshot that can no longer be found
is logged and skipped rather than failed on, unlike the Iceberg callback — it
has either expired, in which case a later snapshot has superseded it, or the
replay would fail forever.
The callback had no test of its own before, so 6a96de471 covers its normal
path, the failure/replay case, the same for a partition the overwrite only
removed files from — a mutant that takes partitions from ADD entries only fails
it — and a retry for an identifier no snapshot carries. The failure is a real
one: the delta table is told to clean a snapshot branch that does not exist, so
the commit publishes its snapshot and then throws, which is the same window as
a transient failure of the snapshot branch, without any file system fault
injection. All chain table suites and the core regression pass.
While surveying every `CommitCallback.retry` for this class of problem I
found one more thing you should decide on. `VisibilityWaitCallback.call` only
acts when the identifier is `BatchWriteBuilder.COMMIT_IDENTIFIER`. Before this
PR the sink committed micro-batches as batch commits, so a query with
`visibility-callback.enabled` waited for visibility after each one; now the
identifier is the batch id and the wait does not run. That matches the option's
stated scope, batch mode or bounded stream, and the Flink streaming behaviour,
so I have left it, but it is a behaviour change for such a query and I would
rather you call it than have it surface later. The other callbacks
(`AddPartitionCommitCallback`, `TagPreviewCommitCallback`,
`IcebergCommitCallback`) already implement retry.
--
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]
