diff --git a/docs/generated/core_configuration.html b/docs/generated/core_configuration.html index 434ad801394a..3f39d29c592b 100644 --- a/docs/generated/core_configuration.html +++ b/docs/generated/core_configuration.html @@ -1474,7 +1474,7 @@
scan.mode
default

Enum

- Specify the scanning behavior of the source.

Possible values: + Specify the scanning behavior of the source.

Possible values:
scan.plan-auto-tag-for-read.time-retained
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 bc1a2102bc78..260e5fa40d53 100644 --- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java +++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java @@ -4842,6 +4842,12 @@ public enum StartupMode implements DescribedEnum { + "without producing a snapshot at the beginning. " + "For batch sources, behaves the same as the \"latest-full\" startup mode."), + LATEST_DELTA( + "latest-delta", + "For batch sources, reads newly changed files from the latest snapshot. " + + "This mode does not search backwards for an APPEND snapshot, so a latest " + + "COMPACT or OVERWRITE snapshot produces no records. Streaming sources are not supported."), + COMPACTED_FULL( "compacted-full", "For streaming sources, produces a snapshot after the latest compaction on the table " diff --git a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java index 1fd428a85187..4f18d9b4e39e 100644 --- a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java +++ b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java @@ -76,16 +76,20 @@ import static org.apache.paimon.CoreOptions.FIELDS_SEPARATOR; import static org.apache.paimon.CoreOptions.FULL_COMPACTION_DELTA_COMMITS; import static org.apache.paimon.CoreOptions.INCREMENTAL_BETWEEN; +import static org.apache.paimon.CoreOptions.INCREMENTAL_BETWEEN_SCAN_MODE; +import static org.apache.paimon.CoreOptions.INCREMENTAL_BETWEEN_TAG_TO_SNAPSHOT; import static org.apache.paimon.CoreOptions.INCREMENTAL_BETWEEN_TIMESTAMP; import static org.apache.paimon.CoreOptions.INCREMENTAL_TO_AUTO_TAG; import static org.apache.paimon.CoreOptions.MAP_STORAGE_LAYOUT; import static org.apache.paimon.CoreOptions.PRIMARY_KEY; +import static org.apache.paimon.CoreOptions.SCAN_CREATION_TIME_MILLIS; import static org.apache.paimon.CoreOptions.SCAN_FILE_CREATION_TIME_MILLIS; import static org.apache.paimon.CoreOptions.SCAN_MODE; import static org.apache.paimon.CoreOptions.SCAN_SNAPSHOT_ID; import static org.apache.paimon.CoreOptions.SCAN_TAG_NAME; import static org.apache.paimon.CoreOptions.SCAN_TIMESTAMP; import static org.apache.paimon.CoreOptions.SCAN_TIMESTAMP_MILLIS; +import static org.apache.paimon.CoreOptions.SCAN_VERSION; import static org.apache.paimon.CoreOptions.SCAN_WATERMARK; import static org.apache.paimon.CoreOptions.SNAPSHOT_NUM_RETAINED_MAX; import static org.apache.paimon.CoreOptions.SNAPSHOT_NUM_RETAINED_MIN; @@ -522,6 +526,24 @@ private static void validateStartupMode(CoreOptions options) { INCREMENTAL_BETWEEN, INCREMENTAL_TO_AUTO_TAG), Collections.singletonList(SCAN_FILE_CREATION_TIME_MILLIS)); + } else if (options.startupMode() == CoreOptions.StartupMode.LATEST_DELTA) { + for (ConfigOption option : + Arrays.asList( + SCAN_TIMESTAMP_MILLIS, + SCAN_FILE_CREATION_TIME_MILLIS, + SCAN_CREATION_TIME_MILLIS, + SCAN_TIMESTAMP, + SCAN_SNAPSHOT_ID, + SCAN_TAG_NAME, + SCAN_WATERMARK, + SCAN_VERSION, + INCREMENTAL_BETWEEN_TIMESTAMP, + INCREMENTAL_BETWEEN, + INCREMENTAL_TO_AUTO_TAG, + INCREMENTAL_BETWEEN_SCAN_MODE, + INCREMENTAL_BETWEEN_TAG_TO_SNAPSHOT)) { + checkOptionNotExistInMode(options, option, options.startupMode()); + } } else { checkOptionNotExistInMode(options, SCAN_TIMESTAMP_MILLIS, options.startupMode()); checkOptionNotExistInMode( diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractBatchTableScan.java b/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractBatchTableScan.java index 9e4e77cf4286..b52424b937b6 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractBatchTableScan.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractBatchTableScan.java @@ -77,7 +77,8 @@ protected AbstractBatchTableScan( if (options.toConfiguration() .get(CoreOptions.BATCH_SCAN_MODE) .equals(CoreOptions.BatchScanMode.NONE) - && options.startupMode() != CoreOptions.StartupMode.INCREMENTAL) { + && options.startupMode() != CoreOptions.StartupMode.INCREMENTAL + && options.startupMode() != CoreOptions.StartupMode.LATEST_DELTA) { snapshotReader.withLevelFilter(level -> level > 0).enableValueFilter(); } } 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 6695e5cefe84..87257a744fae 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 @@ -286,6 +286,19 @@ protected StartingScanner createStartingScanner(boolean isStreaming) { return isStreaming ? new ContinuousLatestStartingScanner(snapshotManager) : new FullStartingScanner(snapshotManager); + case LATEST_DELTA: + checkArgument( + !isStreaming, + "'latest-delta' scan mode is only supported for batch sources."); + Snapshot latestSnapshot = snapshotManager.latestSnapshot(); + if (latestSnapshot == null) { + return new EmptyResultStartingScanner(snapshotManager); + } + return IncrementalDeltaStartingScanner.betweenSnapshotIds( + latestSnapshot.id() - 1, + latestSnapshot.id(), + snapshotManager, + ScanMode.DELTA); case COMPACTED_FULL: if (options.changelogProducer() == ChangelogProducer.FULL_COMPACTION || options.toConfiguration().contains(FULL_COMPACTION_DELTA_COMMITS)) { diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/PostponeMergeReadBuilder.java b/paimon-core/src/main/java/org/apache/paimon/table/source/PostponeMergeReadBuilder.java index 147d12e05a69..09e04e1cca12 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/PostponeMergeReadBuilder.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/PostponeMergeReadBuilder.java @@ -313,6 +313,7 @@ private static void validateReadMode(FileStoreTable table) { } CoreOptions.StartupMode startupMode = table.coreOptions().startupMode(); if (startupMode == CoreOptions.StartupMode.INCREMENTAL + || startupMode == CoreOptions.StartupMode.LATEST_DELTA || startupMode == CoreOptions.StartupMode.FROM_FILE_CREATION_TIME || startupMode == CoreOptions.StartupMode.FROM_CREATION_TIMESTAMP) { throw new UnsupportedOperationException( diff --git a/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java b/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java index 417a00b93b6a..71f674f3aab0 100644 --- a/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java @@ -133,6 +133,37 @@ public void testFromTimestampConflict() { "must set only one key in [scan.timestamp-millis,scan.timestamp] when you use from-timestamp for scan.mode"); } + @Test + public void testLatestDeltaOnlyAcceptsScanMode() { + Map options = new HashMap<>(); + options.put(CoreOptions.SCAN_MODE.key(), CoreOptions.StartupMode.LATEST_DELTA.toString()); + assertThatNoException().isThrownBy(() -> validateTableSchemaExec(options)); + + Map incompatibleOptions = new HashMap<>(); + incompatibleOptions.put(CoreOptions.SCAN_TIMESTAMP_MILLIS.key(), "1"); + incompatibleOptions.put(CoreOptions.SCAN_FILE_CREATION_TIME_MILLIS.key(), "1"); + incompatibleOptions.put(CoreOptions.SCAN_CREATION_TIME_MILLIS.key(), "1"); + incompatibleOptions.put(CoreOptions.SCAN_TIMESTAMP.key(), "2026-08-10 00:00:00"); + incompatibleOptions.put(CoreOptions.SCAN_SNAPSHOT_ID.key(), "1"); + incompatibleOptions.put(CoreOptions.SCAN_TAG_NAME.key(), "tag1"); + incompatibleOptions.put(CoreOptions.SCAN_WATERMARK.key(), "1"); + incompatibleOptions.put(CoreOptions.SCAN_VERSION.key(), "1"); + incompatibleOptions.put(CoreOptions.INCREMENTAL_BETWEEN_TIMESTAMP.key(), "1,2"); + incompatibleOptions.put(CoreOptions.INCREMENTAL_BETWEEN.key(), "1,2"); + incompatibleOptions.put(CoreOptions.INCREMENTAL_TO_AUTO_TAG.key(), "tag1"); + incompatibleOptions.put(CoreOptions.INCREMENTAL_BETWEEN_SCAN_MODE.key(), "delta"); + incompatibleOptions.put(CoreOptions.INCREMENTAL_BETWEEN_TAG_TO_SNAPSHOT.key(), "true"); + + incompatibleOptions.forEach( + (key, value) -> { + Map invalidOptions = new HashMap<>(options); + invalidOptions.put(key, value); + assertThatThrownBy(() -> validateTableSchemaExec(invalidOptions)) + .hasMessageContaining( + key + " must be null when you use latest-delta for scan.mode"); + }); + } + @Test public void testTargetFileRowNumMustBePositive() { Map options = new HashMap<>(); diff --git a/paimon-core/src/test/java/org/apache/paimon/table/source/StartupModeTest.java b/paimon-core/src/test/java/org/apache/paimon/table/source/StartupModeTest.java index 2d3b60f3ae35..73010cae63d1 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/source/StartupModeTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/source/StartupModeTest.java @@ -19,7 +19,9 @@ package org.apache.paimon.table.source; import org.apache.paimon.CoreOptions; +import org.apache.paimon.Snapshot; import org.apache.paimon.data.GenericRow; +import org.apache.paimon.disk.IOManager; import org.apache.paimon.fs.FileIOFinder; import org.apache.paimon.fs.Path; import org.apache.paimon.options.Options; @@ -88,6 +90,60 @@ public void testStartFromLatest() throws Exception { .isEqualTo(snapshotReader.withSnapshot(4).withMode(ScanMode.ALL).read().splits()); } + @Test + public void testStartFromLatestDelta() throws Exception { + initializeTable(StartupMode.LATEST_DELTA); + initializeTestData(); // initialize 3 commits + + TableScan.Plan plan = table.newScan().plan(); + assertThat(plan.splits()) + .isEqualTo(snapshotReader.withSnapshot(3).withMode(ScanMode.DELTA).read().splits()); + + // Do not search backwards for an APPEND snapshot when the latest snapshot is COMPACT. + write.compact(binaryRow(1), 0, true); + commit.commit(4, write.prepareCommit(true, 4)); + assertThat(table.snapshotManager().latestSnapshot().id()).isEqualTo(4); + assertThat(table.snapshotManager().latestSnapshot().commitKind()) + .isEqualTo(Snapshot.CommitKind.COMPACT); + assertThat(table.newScan().plan().splits()).isEmpty(); + + writeAndCommit(5, rowData(1, 10, 103L)); + assertThat(table.newScan().plan().splits()) + .isEqualTo(snapshotReader.withSnapshot(5).withMode(ScanMode.DELTA).read().splits()); + } + + @Test + public void testStartFromLatestDeltaWithoutSnapshot() throws Exception { + initializeTable(StartupMode.LATEST_DELTA); + + assertThat(table.newScan().plan().splits()).isEmpty(); + assertThatThrownBy(() -> table.newStreamScan().plan()) + .satisfies( + anyCauseMatches( + IllegalArgumentException.class, + "'latest-delta' scan mode is only supported for batch sources.")); + } + + @Test + public void testStartFromLatestDeltaDoesNotSkipLevelZero() throws Exception { + Map properties = new HashMap<>(); + properties.put( + CoreOptions.MERGE_ENGINE.key(), CoreOptions.MergeEngine.FIRST_ROW.toString()); + initializeTable(StartupMode.LATEST_DELTA, properties); + try (IOManager ioManager = IOManager.create(tempDir.resolve("latest-delta").toString()); + StreamTableWrite levelZeroWrite = + table.newWrite(commitUser).withIOManager(ioManager); + StreamTableCommit levelZeroCommit = table.newCommit(commitUser)) { + levelZeroWrite.write(rowData(1, 10, 100L)); + levelZeroCommit.commit(1, levelZeroWrite.prepareCommit(false, 1)); + } + + TableScan.Plan plan = table.newScan().plan(); + assertThat(plan.splits()).isNotEmpty(); + assertThat(plan.splits()) + .isEqualTo(snapshotReader.withSnapshot(1).withMode(ScanMode.DELTA).read().splits()); + } + @Test public void testStartFromLatestFull() throws Exception { initializeTable(StartupMode.LATEST_FULL); diff --git a/paimon-core/src/test/java/org/apache/paimon/table/source/TableScanTest.java b/paimon-core/src/test/java/org/apache/paimon/table/source/TableScanTest.java index 2f8f8959ed82..a60893aa6543 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/source/TableScanTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/source/TableScanTest.java @@ -496,6 +496,21 @@ public void testPostponeMergeReadBuilderAndPushDown() throws Exception { .deleteReadTag(readProtectionTag); } + @Test + public void testPostponeMergeRejectsLatestDelta() { + Map dynamicOptions = new HashMap<>(); + dynamicOptions.put(CoreOptions.BUCKET.key(), "-2"); + dynamicOptions.put(CoreOptions.POSTPONE_MERGE_ON_READ.key(), "true"); + dynamicOptions.put(CoreOptions.SCAN_MODE.key(), "latest-delta"); + FileStoreTable latestDeltaTable = table.copy(dynamicOptions); + + assertThatThrownBy( + () -> PostponeMergeReadBuilder.createSnapshotBound(latestDeltaTable, null)) + .isInstanceOf(UnsupportedOperationException.class) + .hasMessageContaining("requires a full snapshot scan") + .hasMessageContaining("latest-delta"); + } + @Test public void testPostponeMergePlanAndRead() throws Exception { StreamTableWrite realWrite = table.newWrite(commitUser); diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/BatchFileStoreITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/BatchFileStoreITCase.java index 6ce5f53387f3..6c9570dddc85 100644 --- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/BatchFileStoreITCase.java +++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/BatchFileStoreITCase.java @@ -1165,6 +1165,19 @@ public void testEmptyTableIncrementalBetweenTimestamp() { .isEmpty(); } + @Test + public void testLatestDeltaScanMode() { + sql("CREATE TABLE latest_delta (id INT, v STRING)"); + sql("INSERT INTO latest_delta VALUES (1, 'A'), (2, 'B')"); + sql("INSERT INTO latest_delta VALUES (3, 'C'), (4, 'D')"); + + assertThat( + sql( + "SELECT * FROM latest_delta " + + "/*+ OPTIONS('scan.mode'='latest-delta') */")) + .containsExactlyInAnyOrder(Row.of(3, "C"), Row.of(4, "D")); + } + @Test public void testIncrementScanMode() throws Exception { sql( diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FlinkCatalogTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FlinkCatalogTest.java index 2bb9c580e576..cc0d2a8083d8 100644 --- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FlinkCatalogTest.java +++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FlinkCatalogTest.java @@ -913,7 +913,9 @@ private static Stream> optionProvider(boolean isStreaming) { options.put("incremental-between", "2,5"); } - if (isStreaming && mode == CoreOptions.StartupMode.INCREMENTAL) { + if (isStreaming + && (mode == CoreOptions.StartupMode.INCREMENTAL + || mode == CoreOptions.StartupMode.LATEST_DELTA)) { continue; } allOptions.add(options);