[core] when use from-timestamp read data from snapshot, if time < ear…

…liestSnapshot time , no data return (#3904) (#3910)
apache · Aug 11, 2024 · 8d76bdb · 8d76bdb
1 parent 0c0a083
commit 8d76bdb
Show file tree

Hide file tree

Showing 4 changed files with 17 additions and 6 deletions.
diff --git a/...ain/java/org/apache/paimon/table/source/snapshot/IncrementalTimeStampStartingScanner.java b/...ain/java/org/apache/paimon/table/source/snapshot/IncrementalTimeStampStartingScanner.java
@@ -52,8 +52,16 @@ public Result scan(SnapshotReader reader) {
                 || endTimestamp < earliestSnapshot.timeMillis()) {
             return new NoSnapshot();
         }
+        // in org.apache.paimon.utils.SnapshotManager.earlierOrEqualTimeMills
+        // 1. if earliestSnapshotId or latestSnapshotId is null startingSnapshotId will be null
+        // 2. if earliestSnapShot.timeMillis() > startTimestamp startingSnapshotId will be
+        // earliestSnapShotId
+        // if  earliestSnapShot.timeMillis() > startTimestamp we should include the earliestSnapShot
+        // data
         Long startSnapshotId =
-                (startingSnapshotId == null) ? earliestSnapshot.id() - 1 : startingSnapshotId;
+                (startingSnapshotId == null || earliestSnapshot.timeMillis() > startTimestamp)
+                        ? earliestSnapshot.id() - 1
+                        : startingSnapshotId;
         Snapshot endSnapshot = snapshotManager.earlierOrEqualTimeMills(endTimestamp);
         Long endSnapshotId = (endSnapshot == null) ? latestSnapshot.id() : endSnapshot.id();
         IncrementalStartingScanner incrementalStartingScanner =

diff --git a/paimon-core/src/main/java/org/apache/paimon/utils/SnapshotManager.java b/paimon-core/src/main/java/org/apache/paimon/utils/SnapshotManager.java
@@ -270,8 +270,9 @@ private Snapshot changelogOrSnapshot(long snapshotId) {
             return null;
         }
 
-        if (snapshot(earliest).timeMillis() > timestampMills) {
-            return null;
+        Snapshot earliestSnapShot = snapshot(earliest);
+        if (earliestSnapShot.timeMillis() > timestampMills) {
+            return earliestSnapShot;
         }
         Snapshot finalSnapshot = null;
         while (earliest <= latest) {

diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/TimeTravelITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/TimeTravelITCase.java
@@ -112,8 +112,10 @@ public void testTravelToOldSchema() throws Exception {
     public void testTravelToNonExistedTimestamp() {
         sql("CREATE TABLE t (k INT, v STRING)");
         sql("INSERT INTO t VALUES(1, 'hello'), (2, 'world')");
-        assertThat(sql("SELECT * FROM t FOR SYSTEM_TIME AS OF TIMESTAMP '1900-01-01 00:00:00'"))
-                .isEmpty();
+        assertThat(
+                        sql("SELECT * FROM t FOR SYSTEM_TIME AS OF TIMESTAMP '1900-01-01 00:00:00'")
+                                .toString())
+                .isEqualTo("[+I[1, hello], +I[2, world]]");
     }
 
     @Test

diff --git a/...park-common/src/test/java/org/apache/paimon/spark/SparkTimeTravelWithDataFrameITCase.java b/...park-common/src/test/java/org/apache/paimon/spark/SparkTimeTravelWithDataFrameITCase.java
@@ -132,7 +132,7 @@ public void testTravelToNonExistedTimestamp() {
                         .option("path", tablePath1.toString())
                         .option(CoreOptions.SCAN_TIMESTAMP_MILLIS.key(), 0)
                         .load();
-        assertThat(dataset.collectAsList()).isEmpty();
+        assertThat(dataset.collectAsList().toString()).isEqualTo("[[1,2,1], [5,6,3]]");
     }
 
     @Test