From e009f10f394f5be12919b0425704db4219db0759 Mon Sep 17 00:00:00 2001
From: sanshi <1715734693@qq.com>
Date: Thu, 3 Sep 2026 16:41:33 +0800
Subject: [PATCH 1/6] [FLINK] Add configurable initial scan mode for streaming
compaction
---
docs/generated/core_configuration.html | 6 +
.../java/org/apache/paimon/CoreOptions.java | 37 ++++++
.../table/source/AbstractDataTableScan.java | 4 +-
.../ContinuousCompactorStartingScanner.java | 19 +++
...ontinuousCompactorStartingScannerTest.java | 123 ++++++++++++++++++
5 files changed, 188 insertions(+), 1 deletion(-)
diff --git a/docs/generated/core_configuration.html b/docs/generated/core_configuration.html
index d25ad4a0c06f..8e3ce4f9b725 100644
--- a/docs/generated/core_configuration.html
+++ b/docs/generated/core_configuration.html
@@ -380,6 +380,12 @@
| MemorySize |
When incremental size is bigger than this threshold, force a full compaction. |
+
+ compaction.initial-scan-mode |
+ earliest |
+ Enum |
+ 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.
Possible values:- "earliest": Read snapshots from the earliest available snapshot.
- "latest": Read the latest snapshot as the initial full baseline.
|
+
compaction.max-size-amplification-percent |
200 |
diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
index c9c2c32a9343..abd50d71d3e2 100644
--- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
@@ -1637,6 +1637,19 @@ public String toString() {
"Only used to force TableScan to construct suitable 'StartingUpScanner' and 'FollowUpScanner' "
+ "dedicated internal streaming scan.");
+ public static final ConfigOption COMPACTION_INITIAL_SCAN_MODE =
+ key("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 BATCH_SCAN_MODE =
key("batch-scan-mode")
@@ -5129,6 +5142,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."),
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractDataTableScan.java b/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractDataTableScan.java
index 24bf32e50cf0..c5653fcb8d84 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractDataTableScan.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractDataTableScan.java
@@ -434,7 +434,9 @@ 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.COMPACTION_INITIAL_SCAN_MODE));
case FILE_MONITOR:
return new FullStartingScanner(snapshotManager);
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/ContinuousCompactorStartingScanner.java b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/ContinuousCompactorStartingScanner.java
index 5f24366ca1cf..0dd095b9368f 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/ContinuousCompactorStartingScanner.java
+++ b/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);
diff --git a/paimon-core/src/test/java/org/apache/paimon/table/source/snapshot/ContinuousCompactorStartingScannerTest.java b/paimon-core/src/test/java/org/apache/paimon/table/source/snapshot/ContinuousCompactorStartingScannerTest.java
index a55a36f014fa..75fd16ed12a5 100644
--- a/paimon-core/src/test/java/org/apache/paimon/table/source/snapshot/ContinuousCompactorStartingScannerTest.java
+++ b/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.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 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();
+ }
}
From 9902146959360c126f51a36f0d2983d357b58a33 Mon Sep 17 00:00:00 2001
From: sanshi <1715734693@qq.com>
Date: Thu, 3 Sep 2026 21:05:27 +0800
Subject: [PATCH 2/6] trigger ci
From 0968dd8f19a3fa97a924da2cb78eea90739f7440 Mon Sep 17 00:00:00 2001
From: sanshi <1715734693@qq.com>
Date: Thu, 3 Sep 2026 22:00:15 +0800
Subject: [PATCH 3/6] trigger ci
From 49643a6dbab6fe2d90d1fafcb08362daa16625e0 Mon Sep 17 00:00:00 2001
From: sanshi <1715734693@qq.com>
Date: Fri, 4 Sep 2026 10:35:08 +0800
Subject: [PATCH 4/6] rename compaction.initial_scan_mode to
continuous-compaction.initial-scan-mode
---
docs/generated/core_configuration.html | 2 +-
paimon-api/src/main/java/org/apache/paimon/CoreOptions.java | 4 ++--
.../org/apache/paimon/table/source/AbstractDataTableScan.java | 2 +-
.../snapshot/ContinuousCompactorStartingScannerTest.java | 2 +-
4 files changed, 5 insertions(+), 5 deletions(-)
diff --git a/docs/generated/core_configuration.html b/docs/generated/core_configuration.html
index 8e3ce4f9b725..fdaabe2a0a72 100644
--- a/docs/generated/core_configuration.html
+++ b/docs/generated/core_configuration.html
@@ -381,7 +381,7 @@
When incremental size is bigger than this threshold, force a full compaction. |
- compaction.initial-scan-mode |
+ continuous-compaction.initial-scan-mode |
earliest |
Enum |
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.
Possible values:- "earliest": Read snapshots from the earliest available snapshot.
- "latest": Read the latest snapshot as the initial full baseline.
|
diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
index abd50d71d3e2..f67014688bb8 100644
--- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
@@ -1637,8 +1637,8 @@ public String toString() {
"Only used to force TableScan to construct suitable 'StartingUpScanner' and 'FollowUpScanner' "
+ "dedicated internal streaming scan.");
- public static final ConfigOption COMPACTION_INITIAL_SCAN_MODE =
- key("compaction.initial-scan-mode")
+ public static final ConfigOption CONTINUOUS_COMPACTION_INITIAL_SCAN_MODE =
+ key("continuous-compaction.initial-scan-mode")
.enumType(CompactionInitialScanMode.class)
.defaultValue(CompactionInitialScanMode.EARLIEST)
.withDescription(
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractDataTableScan.java b/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractDataTableScan.java
index c5653fcb8d84..7d4677a33c68 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractDataTableScan.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractDataTableScan.java
@@ -436,7 +436,7 @@ protected StartingScanner createStartingScanner(boolean isStreaming) {
isStreaming, "Set 'streaming-compact' in batch mode. This is unexpected.");
return new ContinuousCompactorStartingScanner(
snapshotManager,
- options.toConfiguration().get(CoreOptions.COMPACTION_INITIAL_SCAN_MODE));
+ options.toConfiguration().get(CoreOptions.CONTINUOUS_COMPACTION_INITIAL_SCAN_MODE));
case FILE_MONITOR:
return new FullStartingScanner(snapshotManager);
}
diff --git a/paimon-core/src/test/java/org/apache/paimon/table/source/snapshot/ContinuousCompactorStartingScannerTest.java b/paimon-core/src/test/java/org/apache/paimon/table/source/snapshot/ContinuousCompactorStartingScannerTest.java
index 75fd16ed12a5..ecd798ec972a 100644
--- a/paimon-core/src/test/java/org/apache/paimon/table/source/snapshot/ContinuousCompactorStartingScannerTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/table/source/snapshot/ContinuousCompactorStartingScannerTest.java
@@ -93,7 +93,7 @@ public void testLatestBaselineIsFollowedByDeltaScan() throws Exception {
options.set(CoreOptions.WRITE_ONLY, true);
options.set(CoreOptions.STREAM_SCAN_MODE, CoreOptions.StreamScanMode.COMPACT_BUCKET_TABLE);
options.set(
- CoreOptions.COMPACTION_INITIAL_SCAN_MODE,
+ CoreOptions.CONTINUOUS_COMPACTION_INITIAL_SCAN_MODE,
CoreOptions.CompactionInitialScanMode.LATEST);
createAppendOnlyTable(options);
StreamTableWrite write = table.newWrite(commitUser);
From d661ed46d9be779e8a9d1ef2f52f1d764bc49b0c Mon Sep 17 00:00:00 2001
From: sanshi <1715734693@qq.com>
Date: Fri, 4 Sep 2026 10:39:04 +0800
Subject: [PATCH 5/6] fix code format
---
.../java/org/apache/paimon/CoreOptions.java | 25 ++++++++++---------
.../table/source/AbstractDataTableScan.java | 3 ++-
2 files changed, 15 insertions(+), 13 deletions(-)
diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
index f67014688bb8..42779e7b2092 100644
--- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
@@ -1637,18 +1637,19 @@ public String toString() {
"Only used to force TableScan to construct suitable 'StartingUpScanner' and 'FollowUpScanner' "
+ "dedicated internal streaming scan.");
- public static final ConfigOption 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.");
+ public static final ConfigOption
+ 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 BATCH_SCAN_MODE =
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractDataTableScan.java b/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractDataTableScan.java
index 7d4677a33c68..0cdef986fe16 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractDataTableScan.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractDataTableScan.java
@@ -436,7 +436,8 @@ protected StartingScanner createStartingScanner(boolean isStreaming) {
isStreaming, "Set 'streaming-compact' in batch mode. This is unexpected.");
return new ContinuousCompactorStartingScanner(
snapshotManager,
- options.toConfiguration().get(CoreOptions.CONTINUOUS_COMPACTION_INITIAL_SCAN_MODE));
+ options.toConfiguration()
+ .get(CoreOptions.CONTINUOUS_COMPACTION_INITIAL_SCAN_MODE));
case FILE_MONITOR:
return new FullStartingScanner(snapshotManager);
}
From 35019891bc38fb3937bbf8b646620ded0e96db03 Mon Sep 17 00:00:00 2001
From: sanshi <1715734693@qq.com>
Date: Fri, 4 Sep 2026 14:13:21 +0800
Subject: [PATCH 6/6] trigger ci