[Flink] Add configurable initial scan mode for streaming compaction

docs/generated/core_configuration.html

@@ -380,6 +380,12 @@
             <td>MemorySize</td>
             <td>When incremental size is bigger than this threshold, force a full compaction.</td>
         </tr>
+        <tr>
+            <td><h5>continuous-compaction.initial-scan-mode</h5></td>
+            <td style="word-wrap: break-word;">earliest</td>
+            <td><p>Enum</p></td>
+            <td>Initial snapshot mode for dedicated streaming compaction. When set to 'earliest' (the default), compaction starts from the earliest available snapshot if no COMPACT snapshot exists; when a COMPACT snapshot exists, compaction always resumes from the snapshot after it. When set to 'latest', the latest snapshot is read in ALL mode as the initial baseline and subsequent scans start from the next snapshot. The 'latest' mode skips historical snapshot changes and should only be used when historical changelog replay is not required.<br /><br />Possible values:<ul><li>"earliest": Read snapshots from the earliest available snapshot.</li><li>"latest": Read the latest snapshot as the initial full baseline.</li></ul></td>
+        </tr>
         <tr>
             <td><h5>compaction.max-size-amplification-percent</h5></td>
             <td style="word-wrap: break-word;">200</td>

paimon-api/src/main/java/org/apache/paimon/CoreOptions.java

@@ -1637,6 +1637,20 @@ public String toString() {
                             "Only used to force TableScan to construct suitable 'StartingUpScanner' and 'FollowUpScanner' "
                                     + "dedicated internal streaming scan.");
 
+    public static final ConfigOption<CompactionInitialScanMode>
+            CONTINUOUS_COMPACTION_INITIAL_SCAN_MODE =
+                    key("continuous-compaction.initial-scan-mode")
+                            .enumType(CompactionInitialScanMode.class)
+                            .defaultValue(CompactionInitialScanMode.EARLIEST)
+                            .withDescription(
+                                    "Initial snapshot mode for dedicated streaming compaction. "
+                                            + "When set to 'earliest' (the default), compaction starts from the earliest available snapshot "
+                                            + "if no COMPACT snapshot exists; when a COMPACT snapshot exists, compaction always resumes from the snapshot after it. "
+                                            + "When set to 'latest', the latest snapshot is read in ALL mode "
+                                            + "as the initial baseline and subsequent scans start from the next snapshot. "
+                                            + "The 'latest' mode skips historical snapshot changes and should only be used when historical "
+                                            + "changelog replay is not required.");
+
     @ExcludeFromDocumentation("Internal use only")
     public static final ConfigOption<BatchScanMode> BATCH_SCAN_MODE =
             key("batch-scan-mode")
@@ -5129,6 +5143,30 @@ public InlineElement getDescription() {
         }
     }
 
+    /** Initial snapshot mode for dedicated streaming compaction. */
+    public enum CompactionInitialScanMode implements DescribedEnum {
+        EARLIEST("earliest", "Read snapshots from the earliest available snapshot."),
+        LATEST("latest", "Read the latest snapshot as the initial full baseline.");
+
+        private final String value;
+        private final String description;
+
+        CompactionInitialScanMode(String value, String description) {
+            this.value = value;
+            this.description = description;
+        }
+
+        @Override
+        public String toString() {
+            return value;
+        }
+
+        @Override
+        public InlineElement getDescription() {
+            return text(description);
+        }
+    }
+
     /** Inner stream scan mode for some internal requirements. */
     public enum StreamScanMode implements DescribedEnum {
         NONE("none", "No requirement."),

paimon-core/src/main/java/org/apache/paimon/table/source/AbstractDataTableScan.java

@@ -434,7 +434,10 @@ protected StartingScanner createStartingScanner(boolean isStreaming) {
             case COMPACT_BUCKET_TABLE:
                 checkArgument(
                         isStreaming, "Set 'streaming-compact' in batch mode. This is unexpected.");
-                return new ContinuousCompactorStartingScanner(snapshotManager);
+                return new ContinuousCompactorStartingScanner(
+                        snapshotManager,
+                        options.toConfiguration()
+                                .get(CoreOptions.CONTINUOUS_COMPACTION_INITIAL_SCAN_MODE));
             case FILE_MONITOR:
                 return new FullStartingScanner(snapshotManager);
         }

paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/ContinuousCompactorStartingScanner.java

@@ -18,7 +18,9 @@
 
 package org.apache.paimon.table.source.snapshot;
 
+import org.apache.paimon.CoreOptions;
 import org.apache.paimon.Snapshot;
+import org.apache.paimon.table.source.ScanMode;
 import org.apache.paimon.utils.SnapshotManager;
 
 import org.slf4j.Logger;
@@ -27,11 +29,21 @@
 /** {@link StartingScanner} used internally for stand-alone streaming compact job sources. */
 public class ContinuousCompactorStartingScanner extends AbstractStartingScanner {
 
+    private final boolean latestInitialSnapshot;
+
     private static final Logger LOG =
             LoggerFactory.getLogger(ContinuousCompactorStartingScanner.class);
 
     public ContinuousCompactorStartingScanner(SnapshotManager snapshotManager) {
+        this(snapshotManager, CoreOptions.CompactionInitialScanMode.EARLIEST);
+    }
+
+    public ContinuousCompactorStartingScanner(
+            SnapshotManager snapshotManager,
+            CoreOptions.CompactionInitialScanMode initialScanMode) {
         super(snapshotManager);
+        this.latestInitialSnapshot =
+                initialScanMode == CoreOptions.CompactionInitialScanMode.LATEST;
         this.startingSnapshotId = snapshotManager.earliestSnapshotId();
     }
 
@@ -52,6 +64,13 @@ public Result scan(SnapshotReader snapshotReader) {
             }
         }
 
+        if (latestInitialSnapshot) {
+            LOG.debug(
+                    "No compact snapshot found, reading the latest snapshot {} as the initial compaction baseline.",
+                    latestSnapshotId);
+            return StartingScanner.fromPlan(
+                    snapshotReader.withMode(ScanMode.ALL).withSnapshot(latestSnapshotId).read());
+        }
         LOG.debug(
                 "No compact snapshot found, reading from the earliest snapshot {}.",
                 earliestSnapshotId);

paimon-core/src/test/java/org/apache/paimon/table/source/snapshot/ContinuousCompactorStartingScannerTest.java

@@ -18,13 +18,24 @@
 
 package org.apache.paimon.table.source.snapshot;
 
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.options.Options;
 import org.apache.paimon.table.sink.StreamTableCommit;
 import org.apache.paimon.table.sink.StreamTableWrite;
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.table.source.Split;
+import org.apache.paimon.table.source.StreamTableScan;
+import org.apache.paimon.table.source.TableScan;
 import org.apache.paimon.types.RowKind;
 import org.apache.paimon.utils.SnapshotManager;
 
 import org.junit.jupiter.api.Test;
 
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.Set;
+
 import static org.assertj.core.api.Assertions.assertThat;
 
 /** Tests for {@link ContinuousCompactorStartingScanner}. */
@@ -75,4 +86,116 @@ public void testNoSnapshot() {
                 new ContinuousCompactorStartingScanner(snapshotManager);
         assertThat(scanner.scan(snapshotReader)).isInstanceOf(StartingScanner.NoSnapshot.class);
     }
+
+    @Test
+    public void testLatestBaselineIsFollowedByDeltaScan() throws Exception {
+        Options options = new Options();
+        options.set(CoreOptions.WRITE_ONLY, true);
+        options.set(CoreOptions.STREAM_SCAN_MODE, CoreOptions.StreamScanMode.COMPACT_BUCKET_TABLE);
+        options.set(
+                CoreOptions.CONTINUOUS_COMPACTION_INITIAL_SCAN_MODE,
+                CoreOptions.CompactionInitialScanMode.LATEST);
+        createAppendOnlyTable(options);
+        StreamTableWrite write = table.newWrite(commitUser);
+        StreamTableCommit commit = table.newCommit(commitUser);
+
+        write.write(rowData(1, 10, 100L));
+        commit.commit(0, write.prepareCommit(true, 0));
+        write.write(rowData(1, 11, 101L));
+        commit.commit(1, write.prepareCommit(true, 1));
+
+        StreamTableScan scan = table.newStreamScan();
+        TableScan.Plan baseline = scan.plan();
+        assertThat(baseline.splits()).allMatch(split -> ((DataSplit) split).snapshotId() == 2L);
+        assertThat(getResult(table.newRead(), baseline.splits()))
+                .hasSameElementsAs(Arrays.asList("+I 1|10|100", "+I 1|11|101"));
+        assertThat(scan.checkpoint()).isEqualTo(3L);
+
+        write.write(rowData(1, 12, 102L));
+        commit.commit(2, write.prepareCommit(true, 2));
+
+        TableScan.Plan delta = scan.plan();
+        assertThat(delta.splits()).allMatch(split -> ((DataSplit) split).snapshotId() == 3L);
+        assertThat(delta.splits()).isNotEmpty();
+        assertThat(getResult(table.newRead(), delta.splits())).containsExactly("+I 1|12|102");
+        assertThat(scan.checkpoint()).isEqualTo(4L);
+
+        write.close();
+        commit.close();
+    }
+
+    @Test
+    public void testNoCompactSnapshotLatestBaselineContainsAllPartitionsAndBuckets()
+            throws Exception {
+        Options options = new Options();
+        options.set(CoreOptions.WRITE_ONLY, true);
+        options.set(CoreOptions.BUCKET, 2);
+        options.set(CoreOptions.BUCKET_KEY, "a");
+        createAppendOnlyTable(options);
+        SnapshotManager snapshotManager = table.snapshotManager();
+        StreamTableWrite write = table.newWrite(commitUser);
+        StreamTableCommit commit = table.newCommit(commitUser);
+
+        write.write(rowData(1, 10, 100L));
+        write.write(rowData(1, 11, 101L));
+        write.write(rowData(2, 10, 200L));
+        write.write(rowData(2, 11, 201L));
+        commit.commit(0, write.prepareCommit(true, 0));
+
+        StartingScanner.NextSnapshot earliestResult =
+                (StartingScanner.NextSnapshot)
+                        new ContinuousCompactorStartingScanner(snapshotManager)
+                                .scan(snapshotReader);
+        assertThat(earliestResult.nextSnapshotId()).isEqualTo(1L);
+
+        ContinuousCompactorStartingScanner scanner =
+                new ContinuousCompactorStartingScanner(
+                        snapshotManager, CoreOptions.CompactionInitialScanMode.LATEST);
+        StartingScanner.ScannedResult result =
+                (StartingScanner.ScannedResult) scanner.scan(snapshotReader);
+
+        Set<String> partitionBuckets = new HashSet<>();
+        for (Split split : result.splits()) {
+            DataSplit dataSplit = (DataSplit) split;
+            partitionBuckets.add(dataSplit.partition().getInt(0) + ":" + dataSplit.bucket());
+        }
+        assertThat(partitionBuckets).hasSize(4);
+        assertThat(result.splits()).allMatch(split -> !((DataSplit) split).dataFiles().isEmpty());
+
+        write.close();
+        commit.close();
+    }
+
+    @Test
+    public void testNoCompactSnapshotReadsLatestAsInitialBaseline() throws Exception {
+        Options options = new Options();
+        options.set(CoreOptions.WRITE_ONLY, true);
+        createAppendOnlyTable(options);
+        SnapshotManager snapshotManager = table.snapshotManager();
+        StreamTableWrite write = table.newWrite(commitUser);
+        StreamTableCommit commit = table.newCommit(commitUser);
+
+        for (int i = 0; i < 5; i++) {
+            write.write(rowData(1, i, (long) i));
+            commit.commit(i, write.prepareCommit(true, i));
+        }
+
+        ContinuousCompactorStartingScanner scanner =
+                new ContinuousCompactorStartingScanner(
+                        snapshotManager, CoreOptions.CompactionInitialScanMode.LATEST);
+        StartingScanner.ScannedResult result =
+                (StartingScanner.ScannedResult) scanner.scan(snapshotReader);
+
+        assertThat(snapshotManager.earliestSnapshotId()).isEqualTo(1L);
+        assertThat(snapshotManager.latestSnapshotId()).isEqualTo(5L);
+        assertThat(snapshotManager.snapshot(5L).commitKind()).isEqualTo(Snapshot.CommitKind.APPEND);
+        assertThat(result.currentSnapshotId()).isEqualTo(5);
+        assertThat(result.plan().snapshotId()).isEqualTo(5);
+        assertThat(result.splits()).isNotEmpty();
+        assertThat(result.splits()).allMatch(split -> ((DataSplit) split).snapshotId() == 5);
+        assertThat(result.splits()).allMatch(split -> !((DataSplit) split).dataFiles().isEmpty());
+
+        write.close();
+        commit.close();
+    }
 }