NestDream opened a new pull request, #9959: URL: https://github.com/apache/paimon/pull/9959
### Purpose fix #9958 With `changelog-producer=full-compaction` the Flink writer (`GlobalFullCompactionSinkWrite`) keeps a set of checkpoint identifiers whose full compaction it still has to confirm. On every `prepareCommit` while that set is non-empty, `checkSuccessfulFullCompaction` walks the snapshot files from the latest one downwards (one LATEST hint read, one EARLIEST hint read, then one snapshot file per step) until it finds the COMPACT snapshot of one of those identifiers. When it finds it, it clears `writtenBuckets.headMap(id, true)` (inclusive) but `commitIdentifiersToCheck.headSet(id)` (exclusive), so the identifier it just found stays in the set. From then on every checkpoint walks the snapshots again, down to that COMPACT snapshot, until the next full compaction replaces the identifier and the walk starts over. The set never becomes empty once the first full compaction has happened. On a 60-checkpoint MiniCluster run with `full-compaction.delta-commits=10` that is 50 snapshot walks and abo ut 280 snapshot file reads plus 100 hint file reads from this check, against 5 walks, 5 reads and 10 hint reads with the fix. I measured it on a local MiniCluster (Flink 1.20.1, parallelism 1, checkpoint interval 500 ms, datagen source into a primary-key table with `changelog-producer=full-compaction` and `full-compaction.delta-commits=10`, warehouse on a `LocalFileIO` subclass that counts opened snapshot files and checks the call stack). Over 60 checkpoints, unmodified master (36760b4) ran `checkSuccessfulFullCompaction` on all 50 checkpoints after the first full compaction and read 277 to 283 snapshot files plus 100 hint files from it (six runs), a sawtooth of 1, 2, ..., 10 snapshot files per checkpoint, 5.6 per checkpoint on average, 11 at the peak. The writer also logged `Found full compaction snapshot #... with identifier ...` at DEBUG level 50 times for 6 full compactions. With the fix the same run does 5 traversals and reads 5 snapshot files and 10 hint files from that method, one traversal per observed full compaction (the sixth compaction lands on the last checkpoint), and logs the DEBUG line 5 times. Snapshot file reads from all origins in the MiniCluster JVM drop from 453 to 462 down to 181 to 184. Every writer subtask does this, so on a real job the extra snapshot-directory reads per checkpoint scale with the parallelism and with `full-compaction.delta-commits` (up to that many snapshot files per subtask per checkpoint, each a GET on an object store). The fix is the inclusive `headSet(commitIdentifier, true)`, matching the `headMap(commitIdentifier, true)` on the line above and the other `headSet`/`headMap` calls in paimon-flink. ### Tests New `GlobalFullCompactionSinkWriteTest#testStopsCheckingSnapshotsAfterFullCompactionIsObserved`: drives the writer through four checkpoints with `deltaCommits=4` (the fourth triggers the full compaction and commits its COMPACT snapshot), lets checkpoint 5 observe it, and records which snapshot files `prepareCommit` opens for checkpoints 6 and 7 through a `LocalFileIO` subclass. The table writer opens the latest snapshot once to find its last committed identifier, so the assertion is that nothing older than the latest snapshot is opened. Without the fix checkpoint 6 also opens snapshot 5 (the COMPACT snapshot of checkpoint 4) and the test fails: `[snapshot files read by prepareCommit for checkpoint 6] Expecting: [6L, 5L, 6L] to be subset of [6L] but found these extra elements: [5L]`. `FullCompactionFileStoreITCase` still passes. The MiniCluster measurement above was run six times per variant against the unmodified and the fixed module jar; the numbers were stable across runs. -- 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]
