fix

Author: JingsongLi

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

@@ -86,6 +86,8 @@ abstract class AbstractFileStoreTable implements FileStoreTable {
 
     private static final long serialVersionUID = 1L;
 
+    private static final String WATERMARK_PREFIX = "watermark-";
+
     protected final FileIO fileIO;
     protected final Path path;
     protected final TableSchema tableSchema;
@@ -188,7 +190,7 @@ public String uuid() {
 
     @Override
     public Optional<Statistics> statistics() {
-        Snapshot snapshot = TimeTravelUtil.resolveSnapshot(this);
+        Snapshot snapshot = TimeTravelUtil.tryTravelOrLatest(this);
         if (snapshot != null) {
             String file = snapshot.statistics();
             if (file == null) {
@@ -479,8 +481,8 @@ private Optional<TableSchema> tryTimeTravel(Options options) {
         Snapshot snapshot;
         try {
             snapshot =
-                    TimeTravelUtil.resolveSnapshotFromOptions(
-                            options, snapshotManager(), tagManager());
+                    TimeTravelUtil.tryTravelToSnapshot(options, snapshotManager(), tagManager())
+                            .orElse(null);
         } catch (Exception e) {
             return Optional.empty();
         }

paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/TimeTravelUtil.java

@@ -37,6 +37,7 @@
 import java.io.FileNotFoundException;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Optional;
 
 import static org.apache.paimon.utils.Preconditions.checkArgument;
 import static org.apache.paimon.utils.SnapshotManager.EARLIEST_SNAPSHOT_DEFAULT_RETRY_NUM;
@@ -55,14 +56,16 @@ public class TimeTravelUtil {
         CoreOptions.SCAN_TIMESTAMP_MILLIS.key()
     };
 
-    @Nullable
-    public static Snapshot resolveSnapshot(FileStoreTable table) {
-        return resolveSnapshotFromOptions(
+    public static Snapshot tryTravelOrLatest(FileStoreTable table) {
+        return tryTravelToSnapshot(table).orElseGet(() -> table.latestSnapshot().orElse(null));
+    }
+
+    public static Optional<Snapshot> tryTravelToSnapshot(FileStoreTable table) {
+        return tryTravelToSnapshot(
                 table.coreOptions().toConfiguration(), table.snapshotManager(), table.tagManager());
     }
 
-    @Nullable
-    public static Snapshot resolveSnapshotFromOptions(
+    public static Optional<Snapshot> tryTravelToSnapshot(
             Options options, SnapshotManager snapshotManager, TagManager tagManager) {
         adaptScanVersion(options, tagManager);
 
@@ -74,7 +77,7 @@ public static Snapshot resolveSnapshotFromOptions(
         }
 
         if (scanHandleKey.isEmpty()) {
-            return snapshotManager.latestSnapshot();
+            return Optional.empty();
         }
 
         checkArgument(
@@ -87,8 +90,8 @@ public static Snapshot resolveSnapshotFromOptions(
                         CoreOptions.SCAN_TIMESTAMP_MILLIS.key()));
 
         String key = scanHandleKey.get(0);
-        Snapshot snapshot = null;
         CoreOptions coreOptions = new CoreOptions(options);
+        Snapshot snapshot;
         if (key.equals(CoreOptions.SCAN_SNAPSHOT_ID.key())) {
             snapshot =
                     new StaticFromSnapshotStartingScanner(
@@ -108,9 +111,10 @@ public static Snapshot resolveSnapshotFromOptions(
             snapshot =
                     new StaticFromTagStartingScanner(snapshotManager, coreOptions.scanTagName())
                             .getSnapshot();
+        } else {
+            throw new UnsupportedOperationException("Unsupported time travel mode: " + key);
         }
-
-        return snapshot;
+        return Optional.of(snapshot);
     }
 
     private static void adaptScanVersion(Options options, TagManager tagManager) {

paimon-core/src/main/java/org/apache/paimon/table/system/ManifestsTable.java

@@ -229,7 +229,7 @@ private InternalRow toRow(
 
     private static List<ManifestFileMeta> allManifests(FileStoreTable dataTable) {
         CoreOptions options = dataTable.coreOptions();
-        Snapshot snapshot = TimeTravelUtil.resolveSnapshot(dataTable);
+        Snapshot snapshot = TimeTravelUtil.tryTravelOrLatest(dataTable);
         if (snapshot == null) {
             LOG.warn("Check if your snapshot is empty.");
             return Collections.emptyList();

paimon-core/src/main/java/org/apache/paimon/table/system/TableIndexesTable.java

@@ -230,7 +230,7 @@ private InternalRow toRow(
 
     private static List<IndexManifestEntry> allIndexEntries(FileStoreTable dataTable) {
         IndexFileHandler indexFileHandler = dataTable.store().newIndexFileHandler();
-        Snapshot snapshot = TimeTravelUtil.resolveSnapshot(dataTable);
+        Snapshot snapshot = TimeTravelUtil.tryTravelOrLatest(dataTable);
         if (snapshot == null) {
             LOG.warn("Check if your snapshot is empty.");
             return Collections.emptyList();

paimon-core/src/test/java/org/apache/paimon/table/source/snapshot/TimeTravelUtilsTest.java

@@ -37,7 +37,7 @@
 public class TimeTravelUtilsTest extends ScannerTestBase {
 
     @Test
-    public void testResolveSnapshotFromOptions() throws Exception {
+    public void testtryTravelToSnapshot() throws Exception {
         SnapshotManager snapshotManager = table.snapshotManager();
         StreamTableWrite write = table.newWrite(commitUser);
         StreamTableCommit commit = table.newCommit(commitUser);
@@ -57,26 +57,26 @@ public void testResolveSnapshotFromOptions() throws Exception {
         optMap.put("scan.snapshot-id", "2");
         CoreOptions options = CoreOptions.fromMap(optMap);
         Snapshot snapshot =
-                TimeTravelUtil.resolveSnapshotFromOptions(
-                        options.toConfiguration(), snapshotManager, null);
+                TimeTravelUtil.tryTravelToSnapshot(options.toConfiguration(), snapshotManager, null)
+                        .orElse(null);
         assertNotNull(snapshot);
         assertTrue(snapshot.id() == 2);
 
         optMap.clear();
         optMap.put("scan.timestamp-millis", ts + "");
         options = CoreOptions.fromMap(optMap);
         snapshot =
-                TimeTravelUtil.resolveSnapshotFromOptions(
-                        options.toConfiguration(), snapshotManager, null);
+                TimeTravelUtil.tryTravelToSnapshot(options.toConfiguration(), snapshotManager, null)
+                        .orElse(null);
         assertTrue(snapshot.id() == 1);
 
         table.createTag("tag3", 3);
         optMap.clear();
         optMap.put("scan.tag-name", "tag3");
         options = CoreOptions.fromMap(optMap);
         snapshot =
-                TimeTravelUtil.resolveSnapshotFromOptions(
-                        options.toConfiguration(), snapshotManager, null);
+                TimeTravelUtil.tryTravelToSnapshot(options.toConfiguration(), snapshotManager, null)
+                        .orElse(null);
         assertTrue(snapshot.id() == 3);
 
         // if contain more scan.xxx config would throw out
@@ -85,7 +85,7 @@ public void testResolveSnapshotFromOptions() throws Exception {
         assertThrows(
                 IllegalArgumentException.class,
                 () ->
-                        TimeTravelUtil.resolveSnapshotFromOptions(
+                        TimeTravelUtil.tryTravelToSnapshot(
                                 options1.toConfiguration(), snapshotManager, null),
                 "scan.snapshot-id scan.tag-name scan.watermark and scan.timestamp-millis can contains only one");