You are viewing a plain text version of this content. The canonical link for it is here.

Posted to issues@paimon.apache.org by "NestDream (via GitHub)" <gi...@apache.org> on 2026/09/18 09:01:08 UTC

[PR] [flink] Fix GlobalFullCompactionSinkWrite re-reading snapshots on every checkpoint after a full compaction [paimon]

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: issues-unsubscribe@paimon.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [flink] Fix GlobalFullCompactionSinkWrite re-reading snapshots on every checkpoint after a full compaction [paimon]

Posted by "NestDream (via GitHub)" <gi...@apache.org>.

NestDream commented on PR #9959:
URL: https://github.com/apache/paimon/pull/9959#issuecomment-5727712070

   Repro without any code change (master 8b7fbffcc): stream into a primary-key table created with
   
   ```sql
   CREATE TABLE T (id INT, v STRING, PRIMARY KEY (id) NOT ENFORCED) WITH (
     'bucket' = '1',
     'changelog-producer' = 'full-compaction',
     'full-compaction.delta-commits' = '10'
   );
   ```
   
   with a 500 ms checkpoint interval and `org.apache.paimon.flink.sink.GlobalFullCompactionSinkWrite` at DEBUG. After the first full compaction the line `Found full compaction snapshot #... with identifier ...` shows up on every checkpoint (50 times in 60 checkpoints on master); each one is a walk over the snapshot files down to that COMPACT snapshot. With this PR it shows up once per full compaction (5 times in the same run). The snapshot and hint file read counts in the description come from a `LocalFileIO` subclass that counts the files `prepareCommit` opens, run three times per variant on the unmodified and the fixed module jar.
   


-- 
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: issues-unsubscribe@paimon.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [flink] Fix GlobalFullCompactionSinkWrite re-reading snapshots on every checkpoint after a full compaction [paimon]

Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.

JingsongLi commented on PR #9959:
URL: https://github.com/apache/paimon/pull/9959#issuecomment-5742492649

   +1


-- 
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: issues-unsubscribe@paimon.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [flink] Fix GlobalFullCompactionSinkWrite re-reading snapshots on every checkpoint after a full compaction [paimon]

Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.

JingsongLi merged PR #9959:
URL: https://github.com/apache/paimon/pull/9959


-- 
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: issues-unsubscribe@paimon.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org