[snapshot] Optimize watermark check in SnapshotManager which can redu…

…ce FileIO with Filesystem (#3636)
apache · Jun 29, 2024 · 6c4a11d · 6c4a11d
1 parent 088586d
commit 6c4a11d
Show file tree

Hide file tree

Showing 2 changed files with 38 additions and 1 deletion.
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
@@ -285,7 +285,9 @@ private Snapshot changelogOrSnapshot(long snapshotId) {
     public @Nullable Snapshot laterOrEqualWatermark(long watermark) {
         Long earliest = earliestSnapshotId();
         Long latest = latestSnapshotId();
-        if (earliest == null || latest == null) {
+        // If latest == Long.MIN_VALUE don't need next binary search for watermark
+        // which can reduce IO cost with snapshot
+        if (earliest == null || latest == null || snapshot(latest).watermark() == Long.MIN_VALUE) {
             return null;
         }
         Long earliestWatermark = null;

diff --git a/paimon-core/src/test/java/org/apache/paimon/utils/SnapshotManagerTest.java b/paimon-core/src/test/java/org/apache/paimon/utils/SnapshotManagerTest.java
@@ -123,6 +123,21 @@ public void testEarlierOrEqualTimeMills() throws IOException {
                 .isEqualTo(millis + 1000);
     }
 
+    @Test
+    public void testlaterOrEqualWatermark() throws IOException {
+        long millis = Long.MIN_VALUE;
+        FileIO localFileIO = LocalFileIO.create();
+        SnapshotManager snapshotManager =
+                new SnapshotManager(localFileIO, new Path(tempDir.toString()));
+        // create 10 snapshots
+        for (long i = 0; i < 10; i++) {
+            Snapshot snapshot = createSnapshotWithMillis(i, millis, Long.MIN_VALUE);
+            localFileIO.writeFileUtf8(snapshotManager.snapshotPath(i), snapshot.toJson());
+        }
+        // smaller than the second snapshot
+        assertThat(snapshotManager.laterOrEqualWatermark(millis + 999)).isNull();
+    }
+
     private Snapshot createSnapshotWithMillis(long id, long millis) {
         return new Snapshot(
                 id,
@@ -143,6 +158,26 @@ private Snapshot createSnapshotWithMillis(long id, long millis) {
                 null);
     }
 
+    private Snapshot createSnapshotWithMillis(long id, long millis, long watermark) {
+        return new Snapshot(
+                id,
+                0L,
+                null,
+                null,
+                null,
+                null,
+                null,
+                0L,
+                Snapshot.CommitKind.APPEND,
+                millis,
+                null,
+                null,
+                null,
+                null,
+                watermark,
+                null);
+    }
+
     private Changelog createChangelogWithMillis(long id, long millis) {
         return new Changelog(
                 new Snapshot(