[core] Support latest snapshot delta scan in batch

docs/docs/flink/sql-query.mdx

@@ -110,6 +110,21 @@ If you want see `DELETE` records, you can use audit_log table:
 SELECT * FROM t$audit_log /*+ OPTIONS('incremental-between' = '12,20') */;
 ```
 
+### Batch Latest Delta
+
+For a one-shot batch query, you can read newly changed files from only the latest snapshot without
+first looking up its snapshot ID:
+
+```sql
+SELECT * FROM t /*+ OPTIONS('scan.mode' = 'latest-delta') */;
+```
+
+The latest snapshot is selected when the query is planned. This mode does not keep consumer
+progress and does not search backwards for an `APPEND` snapshot. It uses the same semantics as
+`incremental-between-scan-mode = 'delta'`, so a latest `COMPACT` or `OVERWRITE` snapshot produces
+no records. If multiple snapshots must be consumed reliably, use `incremental-between` with
+explicit boundaries instead.
+
 ### Batch Incremental between Auto-created Tags
 
 You can use `incremental-between` to query incremental changes between two tags. But for auto-created tag, the tag may

docs/generated/core_configuration.html

@@ -1474,7 +1474,7 @@
             <td><h5>scan.mode</h5></td>
             <td style="word-wrap: break-word;">default</td>
             <td><p>Enum</p></td>
-            <td>Specify the scanning behavior of the source.<br /><br />Possible values:<ul><li>"default": Determines actual startup mode according to other table properties. If "scan.timestamp-millis" is set the actual startup mode will be "from-timestamp", and if "scan.snapshot-id" or "scan.tag-name" is set the actual startup mode will be "from-snapshot". Otherwise the actual startup mode will be "latest-full".</li><li>"latest-full": For streaming sources, produces the latest snapshot on the table upon first startup, and continue to read the latest changes. For batch sources, just produce the latest snapshot but does not read new changes.</li><li>"full": Deprecated. Same as "latest-full".</li><li>"latest": For streaming sources, continuously reads latest changes without producing a snapshot at the beginning. For batch sources, behaves the same as the "latest-full" startup mode.</li><li>"compacted-full": For streaming sources, produces a snapshot after the latest compaction on the table upon first startup, and continue to read the latest changes. For batch sources, just produce a snapshot after the latest compaction but does not read new changes. Snapshots of full compaction are picked when scheduled full-compaction is enabled.</li><li>"from-timestamp": For streaming sources, continuously reads changes starting from timestamp specified by "scan.timestamp-millis", without producing a snapshot at the beginning. For batch sources, produces a snapshot at timestamp specified by "scan.timestamp-millis" but does not read new changes.</li><li>"from-creation-timestamp": For streaming sources and batch sources, If timestamp specified by "scan.creation-time-millis" is during in the range of earliest snapshot and latest snapshot: mode is from-snapshot which snapshot is equal or later the timestamp. If timestamp is earlier than earliest snapshot or later than latest snapshot, mode is from-file-creation-time.</li><li>"from-file-creation-time": For streaming and batch sources, consumes a snapshot and filters the data files by creation time. For streaming sources, upon first startup, and continue to read the latest changes.</li><li>"from-snapshot": For streaming sources, continuously reads changes starting from snapshot specified by "scan.snapshot-id", without producing a snapshot at the beginning. For batch sources, produces a snapshot specified by "scan.snapshot-id" or "scan.tag-name" but does not read new changes.</li><li>"from-snapshot-full": For streaming sources, produces from snapshot specified by "scan.snapshot-id" on the table upon first startup, and continuously reads changes. For batch sources, produces a snapshot specified by "scan.snapshot-id" but does not read new changes.</li><li>"incremental": Read incremental changes between start and end snapshot or timestamp.</li></ul></td>
+            <td>Specify the scanning behavior of the source.<br /><br />Possible values:<ul><li>"default": Determines actual startup mode according to other table properties. If "scan.timestamp-millis" is set the actual startup mode will be "from-timestamp", and if "scan.snapshot-id" or "scan.tag-name" is set the actual startup mode will be "from-snapshot". Otherwise the actual startup mode will be "latest-full".</li><li>"latest-full": For streaming sources, produces the latest snapshot on the table upon first startup, and continue to read the latest changes. For batch sources, just produce the latest snapshot but does not read new changes.</li><li>"full": Deprecated. Same as "latest-full".</li><li>"latest": For streaming sources, continuously reads latest changes without producing a snapshot at the beginning. For batch sources, behaves the same as the "latest-full" startup mode.</li><li>"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.</li><li>"compacted-full": For streaming sources, produces a snapshot after the latest compaction on the table upon first startup, and continue to read the latest changes. For batch sources, just produce a snapshot after the latest compaction but does not read new changes. Snapshots of full compaction are picked when scheduled full-compaction is enabled.</li><li>"from-timestamp": For streaming sources, continuously reads changes starting from timestamp specified by "scan.timestamp-millis", without producing a snapshot at the beginning. For batch sources, produces a snapshot at timestamp specified by "scan.timestamp-millis" but does not read new changes.</li><li>"from-creation-timestamp": For streaming sources and batch sources, If timestamp specified by "scan.creation-time-millis" is during in the range of earliest snapshot and latest snapshot: mode is from-snapshot which snapshot is equal or later the timestamp. If timestamp is earlier than earliest snapshot or later than latest snapshot, mode is from-file-creation-time.</li><li>"from-file-creation-time": For streaming and batch sources, consumes a snapshot and filters the data files by creation time. For streaming sources, upon first startup, and continue to read the latest changes.</li><li>"from-snapshot": For streaming sources, continuously reads changes starting from snapshot specified by "scan.snapshot-id", without producing a snapshot at the beginning. For batch sources, produces a snapshot specified by "scan.snapshot-id" or "scan.tag-name" but does not read new changes.</li><li>"from-snapshot-full": For streaming sources, produces from snapshot specified by "scan.snapshot-id" on the table upon first startup, and continuously reads changes. For batch sources, produces a snapshot specified by "scan.snapshot-id" but does not read new changes.</li><li>"incremental": Read incremental changes between start and end snapshot or timestamp.</li></ul></td>
         </tr>
         <tr>
             <td><h5>scan.plan-auto-tag-for-read.time-retained</h5></td>

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 "

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(

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();
             }
         }

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)) {

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(

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<String, String> options = new HashMap<>();
+        options.put(CoreOptions.SCAN_MODE.key(), CoreOptions.StartupMode.LATEST_DELTA.toString());
+        assertThatNoException().isThrownBy(() -> validateTableSchemaExec(options));
+
+        Map<String, String> 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<String, String> 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<String, String> options = new HashMap<>();

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<String, String> 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);

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<String, String> 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);

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(