beryllw commented on code in PR #3420:
URL: https://github.com/apache/fluss/pull/3420#discussion_r3541102128
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/committer/TieringCommitOperator.java:
##########
@@ -232,14 +233,35 @@ private CommitResult commitWriteResults(
Map<TableBucket, Long> logEndOffsets = new HashMap<>();
Map<TableBucket, Long> logMaxTieredTimestamps = new HashMap<>();
+ Long watermark = null;
for (TableBucketWriteResult<WriteResult> writeResult :
nonEmptyResults) {
TableBucket tableBucket = writeResult.tableBucket();
logEndOffsets.put(tableBucket, writeResult.logEndOffset());
logMaxTieredTimestamps.put(tableBucket,
writeResult.maxTimestamp());
+ Long writeResultWatermark =
writeResult.writeResult().getWatermark();
+ if (writeResultWatermark == null) {
+ continue;
+ }
+ watermark =
Review Comment:
nit: Consider adding a brief comment explaining why it's safe for the
aggregated watermark to be lower than previously committed — Paimon's
FileStoreCommitImpl#tryCommitOnce guarantees snapshot watermark monotonicity
via Math.max(current, latestSnapshot.watermark()), so it will never regress.
This would save future readers from needing to trace into Paimon source to
verify correctness.
https://github.com/apache/paimon/blob/5b797ef63fba5e98bf4fd285fb5a34877c91ee81/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java#L1020-L1026
--
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]
