zhuxiangyi commented on code in PR #9667:
URL: https://github.com/apache/paimon/pull/9667#discussion_r3998730175
##########
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:
Confirmed, and reproduced first with a fail-once `PartitionModification`
through the Spark direct path: after the first attempt failed past snapshot
publication, the replay left `attempts=1` and the partition unregistered,
exactly as you saw. The lookup I had moved into the connector returned before
`filterCommitted`, and with it the `callback.retry` that half of the semantics
lives in. I had seen that line and dropped it.
The callbacks live in core, so the direct path commits through
`filterAndCommit` again, in e7a9976bb. What had forced the connector-side
lookup was the strict mode bound on `filterCommitted`; I traced it to
#7239/#7275, an optimisation for a commit user created for one run, which
cannot have committed before its base snapshot. A caller-provided user that
survives a restart can have, so `FileStoreCommit` can now look the previous
commit up without the bound, and both write builders enable that when
`withCommitUser` was called. Conflict detection keeps the bound, and so does a
committer created for an explicitly passed user — the staged committer with its
per-run user goes through that overload, and a core test pins both sides for
each builder.
While reading `filterAndCommitMultiple` for this I found a second cost of
the same kind as the file-existence check: it passes `checkAppendFiles=true`,
so every micro-batch on the ordinary path scanned the base files of the
partitions it touches for conflicts, which an append of files the batch just
wrote cannot have. A batch commit never did that scan before this series and
the Flink committer does not in steady state either; `InnerTableCommit` can now
turn it off and the sink does. For an append commit the scan has no observable
effect, so this is covered by the existing suites rather than by a test of its
own.
Suite is at 12 cases, run on 3.2, 3.4, 3.5, 4.0 and 4.1; core regression
(`FileStoreCommitTest`, `TableCommitTest`, both `SimpleTableTest` subclasses)
is green.
The two red jobs are `UTCase and ITCase Others` on JDK 11 and 17, failing in
`S3FileIOTest` because Testcontainers cannot pull
`minio/minio:RELEASE.2022-02-07T08-17-33Z`; the same workflow fails the same
way on the last four runs of master (e.g. run 34697050592), so it is
independent of this PR.
--
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]
