zhuxiangyi commented on code in PR #9667:
URL: https://github.com/apache/paimon/pull/9667#discussion_r3995564323
##########
paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala:
##########
@@ -472,14 +498,32 @@ case class PaimonSparkWriter(
postCommit(finalMessages)
return
}
- val activeWriteBuilder =
- Option(directPostponeWriteBuilder).getOrElse(writeBuilder)
- val tableCommit = activeWriteBuilder.newCommit()
+ val tableCommit: InnerTableCommit =
+ if (directPostponeWriteBuilder != null) {
+ directPostponeWriteBuilder.newCommit()
Review Comment:
Confirmed and reproduced first: with `bucket = -2`,
`postpone.default-bucket-num = 1` and complete mode, replaying batch 0 took the
table from snapshot 1 to snapshot 2, exactly as you saw. Both halves of your
analysis held up on inspection.
Fixed in 587ad9c14 along the first line you suggested:
- `PostponeFixedBucketWriteBuilder` gained `withCommitUser`, mirroring the
one on `BatchWriteBuilderImpl`, and the direct builder is created with the
stable user.
- The direct path looks the replay up itself, with `latestSnapshotOfUser`
unbounded, and commits under the batch id only when nothing at or above it has
been committed. The strict mode bound is left alone: it is what conflict
detection needs, and I traced its use in `filterCommitted` to #7239/#7275, a
lookup optimisation whose premise is a commit user that never outlives its base
snapshot. That premise is exactly what a stable user breaks, so the lookup for
this path has to be independent of it rather than the bound relaxed for
everyone.
The test for this configuration also commits a later batch after the replay
and checks it lands, so that a higher batch id cannot be mistaken for a replay
— the failure mode in that direction is silent data loss, which is worse than
the duplicate.
Two things worth recording. First, while looking for a way to observe
maintenance on this path I found that
`PrimaryKeyFileStoreTable.newExpireRunnable()` returns null for postpone bucket
tables, so snapshot expiration is never part of a commit there; a probe that
expects the direct path to expire snapshots will not see it, regardless of
`inlineMaintenance`. I restructured the sink so `inlineMaintenance(true)` is
set once above both streaming branches, where the existing expiration test on
an append table guards it, and added a core test in `SimpleTableTestBase` that
commits through `filterAndCommit` with a committer closed after each batch and
asserts the previous snapshot is expired before the close. Second, the staged
committer is still outside the replay handling and keeps its warning; I did not
extend the change there.
The suite is at 11 cases and was run on the 3.2, 3.4, 3.5, 4.0 and 4.1
modules again.
--
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]
