diff --git a/docs/generated/core_configuration.html b/docs/generated/core_configuration.html index d25ad4a0c06f..fdaabe2a0a72 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. + +
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: +
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..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,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 + 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 = 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."), 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..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 @@ -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); } 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..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 @@ -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 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(); + } }