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

[NestDream]

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 about 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.

[NestDream]

Repro without any code change (master 8b7fbff): stream into a primary-key table created with

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.

[JingsongLi]

+1