zhuxiangyi commented on code in PR #9667:
URL: https://github.com/apache/paimon/pull/9667#discussion_r3966088161
##########
paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala:
##########
@@ -472,14 +498,29 @@ 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. The files being committed
were written by this
+ // very batch, so there is no need to list them to prove that they
still exist.
+ tableCommit
+ .checkFilesExistence(false)
+ .filterAndCommit(
+ Collections.singletonMap(Long.box(identifier),
commitMessages.toList.asJava))
Review Comment:
Confirmed. `maintain` branches on `batchCommitted`, which only
`checkCommitted()` sets, so the filtered path left maintenance to the executor
while the writer kept the one-shot lifecycle of a batch committer — the two
halves no longer matched.
The asynchronous half reproduced clearly. With
`snapshot.expire.execution-mode=async` and min/max retained set to 1, four
micro-batches left `snapshots=4 earliest=1 latest=4`: `shutdownNow()` drained
the queued task, so expiration did not run at all rather than being cut short
partway. For a long-running streaming write that means snapshots grow without
bound.
The synchronous half I verified by reading rather than by test: the wrapper
stores the failure in `maintainError`, which is only rethrown at the start of
the next `maintain`, and this committer is discarded before there is one.
Rather than test the swallowing, I made it structurally impossible.
`InnerTableCommit` now has `inlineMaintenance(boolean)`: maintenance runs on
the committing thread and its failure is thrown to the caller, which is what
`commit(List)` already gets from `batchCommitted`. The Spark sink asks for it
on the filtered path. The default is unchanged, so Flink keeps the executor.
`testBatchWriteAsyncExpireFallbackToSync` was the right reference — the
invariant it fixes is exactly the one I had broken, and the new test asserts
the same property through the streaming sink.
Worth noting for the record: expiration now runs inline on each micro-batch,
which restores the behaviour before this PR rather than adding cost. Making it
asynchronous again would mean keeping one committer alive across batches, the
way `CommitterOperator` does; that is a larger change than this fix and I have
left it out.
--
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]
