[core] Handle missing snapshots in branch time travel

docs/generated/core_configuration.html

@@ -1566,6 +1566,12 @@
             <td>String</td>
             <td>When a batch job queries from a table, if a partition does not exist in the current branch, the reader will try to get this partition from this fallback branch.</td>
         </tr>
+        <tr>
+            <td><h5>scan.fallback-branch.allow-earliest-snapshot</h5></td>
+            <td style="word-wrap: break-word;">false</td>
+            <td>Boolean</td>
+            <td>Whether to allow reading the earliest available snapshot of a fallback or primary branch when no snapshot exists at or before the current branch's requested snapshot time. This applies to snapshot-ID time travel, including numeric scan.version. By default, such reads fail unless the other branch has no snapshots. Enabling this option may read data committed after the requested time.</td>
+        </tr>
         <tr>
             <td><h5>scan.fallback-branch.read-fail-fast</h5></td>
             <td style="word-wrap: break-word;">false</td>

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

@@ -2396,6 +2396,17 @@ public String toString() {
                                     + "falls through to the current branch, which can mask data issues. "
                                     + "Set this to true to surface fallback branch errors to the caller instead.");
 
+    public static final ConfigOption<Boolean> SCAN_FALLBACK_BRANCH_ALLOW_EARLIEST_SNAPSHOT =
+            key("scan.fallback-branch.allow-earliest-snapshot")
+                    .booleanType()
+                    .defaultValue(false)
+                    .withDescription(
+                            "Whether to allow reading the earliest available snapshot of a fallback or primary branch "
+                                    + "when no snapshot exists at or before the current branch's requested snapshot time. "
+                                    + "This applies to snapshot-ID time travel, including numeric scan.version. "
+                                    + "By default, such reads fail unless the other branch has no snapshots. "
+                                    + "Enabling this option may read data committed after the requested time.");
+
     public static final ConfigOption<String> SCAN_PRIMARY_BRANCH =
             key("scan.primary-branch")
                     .stringType()

paimon-core/src/main/java/org/apache/paimon/table/FallbackReadFileStoreTable.java

@@ -216,13 +216,34 @@ protected Map<String, String> rewriteOtherOptions(Map<String, String> options) {
             long id = Long.parseLong(scanSnapshotId);
             long millis = wrapped.snapshotManager().snapshot(id).timeMillis();
             Snapshot otherSnapshot = other.snapshotManager().earlierOrEqualTimeMills(millis);
-            long otherId;
             if (otherSnapshot == null) {
-                otherId = Snapshot.FIRST_SNAPSHOT_ID;
-            } else {
-                otherId = otherSnapshot.id();
+                otherSnapshot = other.snapshotManager().earliestSnapshot();
+                if (otherSnapshot != null) {
+                    Preconditions.checkArgument(
+                            new Options(wrapped.options(), options)
+                                    .get(CoreOptions.SCAN_FALLBACK_BRANCH_ALLOW_EARLIEST_SNAPSHOT),
+                            "No snapshot at or before %s is available in branch '%s'. "
+                                    + "The earliest available snapshot is %s (time %s). "
+                                    + "Set '%s' to true to allow reading it.",
+                            millis,
+                            other.coreOptions().branch(),
+                            otherSnapshot.id(),
+                            otherSnapshot.timeMillis(),
+                            CoreOptions.SCAN_FALLBACK_BRANCH_ALLOW_EARLIEST_SNAPSHOT.key());
+                    LOG.warn(
+                            "No snapshot at or before {} is available in branch '{}'; reading earliest "
+                                    + "snapshot {} (time {}) because '{}' is enabled.",
+                            millis,
+                            other.coreOptions().branch(),
+                            otherSnapshot.id(),
+                            otherSnapshot.timeMillis(),
+                            CoreOptions.SCAN_FALLBACK_BRANCH_ALLOW_EARLIEST_SNAPSHOT.key());
+                }
             }
-            result.put(scanSnapshotIdOptionKey, String.valueOf(otherId));
+            // An empty branch is skipped by newFallbackScan instead of loading this ID.
+            result.put(
+                    scanSnapshotIdOptionKey,
+                    otherSnapshot == null ? scanSnapshotId : String.valueOf(otherSnapshot.id()));
         }
 
         // bucket number of main branch and other branch are very likely different,
@@ -246,7 +267,53 @@ public DataTableScan newFallbackScan(Function<FileStoreTable, DataTableScan> sca
         validateSchema();
         FileStoreTable first = wrappedFirst ? wrapped : other;
         FileStoreTable second = wrappedFirst ? other : wrapped;
-        return new FallbackReadScan(first, second, wrapped.schema(), scanCreator);
+        boolean otherIsEmpty =
+                other.coreOptions().scanSnapshotId() != null
+                        && other.snapshotManager().latestSnapshotId() == null;
+        if (otherIsEmpty) {
+            LOG.warn(
+                    "Branch '{}' has no snapshots; skipping it when reading branch '{}'.",
+                    other.coreOptions().branch(),
+                    wrapped.coreOptions().branch());
+        }
+        return new FallbackReadScan(
+                first,
+                second,
+                wrapped.schema(),
+                table -> {
+                    if (table == other && otherIsEmpty) {
+                        return new EmptyBranchScan();
+                    }
+                    return scanCreator.apply(table);
+                });
+    }
+
+    private static class EmptyBranchScan implements DataTableScan {
+
+        @Override
+        public DataTableScan withShard(int indexOfThisSubtask, int numberOfParallelSubtasks) {
+            return this;
+        }
+
+        @Override
+        public InnerTableScan withFilter(Predicate predicate) {
+            return this;
+        }
+
+        @Override
+        public InnerTableScan withInternalPartitionFilter(PartitionPredicate partitionPredicate) {
+            return this;
+        }
+
+        @Override
+        public Plan plan() {
+            return new DataFilePlan<>(Collections.emptyList());
+        }
+
+        @Override
+        public List<PartitionEntry> listPartitionEntries() {
+            return Collections.emptyList();
+        }
     }
 
     protected void validateSchema() {

paimon-core/src/test/java/org/apache/paimon/table/FallbackReadFileStoreTableTest.java

@@ -70,7 +70,9 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.UUID;
 import java.util.function.Consumer;
 import java.util.stream.Collectors;
@@ -368,6 +370,141 @@ public void testScanVersionForFallbackBranch() throws Exception {
                 .containsExactlyInAnyOrder(Pair.of(1, 10), Pair.of(2, 20));
     }
 
+    @ParameterizedTest
+    @CsvSource({"false,false,false", "false,true,true", "true,false,true", "true,true,false"})
+    public void testTimeTravelWithEmptyOtherBranch(
+            boolean wrappedFirst, boolean scanVersion, boolean allowEarliest) throws Exception {
+        FileStoreTable mainTable = createTable();
+        writeDataIntoTable(mainTable, 0, rowData(1, 10));
+        mainTable.createBranch("empty");
+        FileStoreTable branchTable = createTableFromBranch(mainTable, "empty");
+        assertThat(branchTable.snapshotManager().latestSnapshotId()).isNull();
+        Map<String, String> options = new HashMap<>();
+        options.put(
+                scanVersion ? CoreOptions.SCAN_VERSION.key() : CoreOptions.SCAN_SNAPSHOT_ID.key(),
+                "1");
+        options.put(
+                CoreOptions.SCAN_FALLBACK_BRANCH_ALLOW_EARLIEST_SNAPSHOT.key(),
+                String.valueOf(allowEarliest));
+
+        FallbackReadFileStoreTable latest =
+                new FallbackReadFileStoreTable(mainTable, branchTable, wrappedFirst);
+        assertThat(readAndCollect(latest, scan -> {})).containsExactly(Pair.of(1, 10));
+        FallbackReadFileStoreTable table = (FallbackReadFileStoreTable) latest.copy(options);
+
+        assertThat(readAndCollect(table, scan -> {})).containsExactly(Pair.of(1, 10));
+        assertThat(table.newScan().listPartitionEntries())
+                .map(entry -> entry.partition().getInt(0))
+                .containsExactly(1);
+    }
+
+    @ParameterizedTest
+    @CsvSource({"false,false", "false,true", "true,false", "true,true"})
+    public void testTimeTravelBeforeOtherSnapshots(boolean wrappedFirst, boolean expired)
+            throws Exception {
+        FileStoreTable mainTable = createTable();
+        writeDataIntoTable(mainTable, 0, rowData(1, 10));
+        long snapshotTime = mainTable.snapshotManager().snapshot(1).timeMillis();
+        mainTable.createBranch("bc");
+        FileStoreTable branchTable = createTableFromBranch(mainTable, "bc");
+        while (System.currentTimeMillis() <= snapshotTime) {
+            Thread.yield();
+        }
+        writeDataIntoTable(branchTable, 0, rowData(1, 100), rowData(2, 20));
+        if (expired) {
+            writeDataIntoTable(branchTable, 1, rowData(3, 30));
+            branchTable.snapshotManager().deleteSnapshot(1);
+            branchTable.snapshotManager().commitEarliestHint(2);
+        }
+        // The latest snapshot must differ from earliest; its extra row must not be read.
+        writeDataIntoTable(branchTable, 2, rowData(4, 40));
+        if (expired) {
+            branchTable
+                    .schemaManager()
+                    .commitChanges(SchemaChange.addColumn("extra", DataTypes.STRING()));
+            branchTable = branchTable.copyWithLatestSchema();
+        }
+        FallbackReadFileStoreTable table =
+                new FallbackReadFileStoreTable(mainTable, branchTable, wrappedFirst);
+        Map<String, String> options = new HashMap<>();
+        options.put(CoreOptions.SCAN_SNAPSHOT_ID.key(), "1");
+        assertThatThrownBy(() -> table.copy(options))
+                .isInstanceOf(IllegalArgumentException.class)
+                .hasMessageContaining("No snapshot at or before");
+
+        options.put(CoreOptions.SCAN_FALLBACK_BRANCH_ALLOW_EARLIEST_SNAPSHOT.key(), "true");
+        FallbackReadFileStoreTable historical = (FallbackReadFileStoreTable) table.copy(options);
+        long earliestId = expired ? 2L : 1L;
+        assertThat(historical.other().coreOptions().scanSnapshotId()).isEqualTo(earliestId);
+        assertThat(historical.other().rowType()).isEqualTo(mainTable.rowType());
+        List<Pair<Integer, Integer>> expected =
+                new ArrayList<>(Arrays.asList(Pair.of(1, wrappedFirst ? 10 : 100), Pair.of(2, 20)));
+        if (expired) {
+            expected.add(Pair.of(3, 30));
+        }
+        assertThat(readAndCollect(historical, scan -> {}))
+                .containsExactlyInAnyOrderElementsOf(expected);
+
+        options.remove(CoreOptions.SCAN_FALLBACK_BRANCH_ALLOW_EARLIEST_SNAPSHOT.key());
+        assertThat(
+                        ((FallbackReadFileStoreTable) historical.copy(options))
+                                .other()
+                                .coreOptions()
+                                .scanSnapshotId())
+                .isEqualTo(earliestId);
+        for (String override : Arrays.asList("false", null)) {
+            options.put(CoreOptions.SCAN_FALLBACK_BRANCH_ALLOW_EARLIEST_SNAPSHOT.key(), override);
+            assertThatThrownBy(() -> historical.copy(options))
+                    .isInstanceOf(IllegalArgumentException.class)
+                    .hasMessageContaining("No snapshot at or before");
+        }
+    }
+
+    @ParameterizedTest
+    @ValueSource(booleans = {false, true})
+    public void testExplicitMissingSnapshotIsNotReplaced(boolean expired) throws Exception {
+        FileStoreTable table = createTable();
+        if (expired) {
+            writeDataIntoTable(table, 0, rowData(1, 10));
+            writeDataIntoTable(table, 1, rowData(2, 20));
+            table.snapshotManager().deleteSnapshot(1);
+            table.snapshotManager().commitEarliestHint(2);
+        }
+        Map<String, String> options = new HashMap<>();
+        options.put(CoreOptions.SCAN_SNAPSHOT_ID.key(), "1");
+        options.put(CoreOptions.SCAN_FALLBACK_BRANCH_ALLOW_EARLIEST_SNAPSHOT.key(), "true");
+
+        assertThatThrownBy(() -> table.copy(options).newScan().plan())
+                .isInstanceOf(IllegalArgumentException.class)
+                .hasMessageContaining(
+                        expired
+                                ? "out of available snapshotId range"
+                                : "There is currently no snapshot.");
+    }
+
+    @Test
+    public void testTagTimeTravelWithoutActiveOtherSnapshots() throws Exception {
+        FileStoreTable mainTable = createTable();
+        writeDataIntoTable(mainTable, 0, rowData(1, 10));
+        mainTable.createTag("base");
+        mainTable.createBranch("bc");
+        FileStoreTable branchTable = createTableFromBranch(mainTable, "bc");
+        writeDataIntoTable(branchTable, 0, rowData(2, 20));
+        branchTable.createTag("base");
+        branchTable.snapshotManager().deleteSnapshot(1);
+        branchTable.snapshotManager().deleteLatestHint();
+        assertThat(branchTable.snapshotManager().latestSnapshotId()).isNull();
+        FallbackReadFileStoreTable table =
+                (FallbackReadFileStoreTable)
+                        new FallbackReadFileStoreTable(mainTable, branchTable, true)
+                                .copy(
+                                        Collections.singletonMap(
+                                                CoreOptions.SCAN_TAG_NAME.key(), "base"));
+
+        assertThat(readAndCollect(table, scan -> {}))
+                .containsExactlyInAnyOrder(Pair.of(1, 10), Pair.of(2, 20));
+    }
+
     private DataTableScan queryAuthScan(DataTableScan delegate, TableQueryAuthResult authResult) {
         DataTableScan scan =
                 Mockito.mock(DataTableScan.class, AdditionalAnswers.delegatesTo(delegate));