JingsongLi commented on code in PR #9667:
URL: https://github.com/apache/paimon/pull/9667#discussion_r3996246106
##########
paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala:
##########
@@ -472,14 +507,45 @@ 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.
+ //
+ // 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.inlineMaintenance(true)
+ if (directPostponeWriteBuilder != null) {
+ // The direct postpone committer runs in strict mode, and
filterAndCommit bounds its
+ // lookup of the previous commit by the snapshot this write
started from. The batch
+ // being replayed was committed before that snapshot, so look it
up without the bound.
+ if (alreadyCommitted(identifier)) {
+ logInfo(s"Micro-batch $identifier is already committed, skipping
the replay.")
Review Comment:
[P2] Retry commit callbacks before acknowledging an already-published batch
For the direct postpone path (bucket=-2, postpone.default-bucket-num=1) on a
partitioned table with metastore.partitioned-table=true, a commit can publish
its snapshot and then fail in the built-in AddPartitionCommitCallback, for
example when the metastore partition-registration RPC fails. On replay, this
branch only logs and returns successfully, bypassing
FileStoreCommitImpl.filterCommitted, which deliberately calls callback.retry
for already-committed batches. I reproduced this with a real local Paimon table
and a fail-once PartitionModification: the first call throws after snapshot 1
is published; the replay succeeds with snapshot 1 still current,
registrationAttempts=1 and registered=false. The normal filterAndCommit control
retries registration (attempts=2, registered=true) without another snapshot.
Please preserve the unbounded deduplication lookup while also completing the
pending callbacks before reporting replay success; otherwise the checkpoint can
advance with its comm
itted partition still absent from the metastore.
--
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]
