[core] Fix race condition for earliest snapshot

paimon-core/src/main/java/org/apache/paimon/Changelog.java

@@ -28,6 +28,7 @@
 
 import javax.annotation.Nullable;
 
+import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.util.Map;
 
@@ -105,9 +106,19 @@ public static Changelog fromJson(String json) {
     }
 
     public static Changelog fromPath(FileIO fileIO, Path path) {
+        try {
+            return tryFromPath(fileIO, path);
+        } catch (FileNotFoundException e) {
+            throw new RuntimeException("Fails to read changelog from path " + path, e);
+        }
+    }
+
+    public static Changelog tryFromPath(FileIO fileIO, Path path) throws FileNotFoundException {
         try {
             String json = fileIO.readFileUtf8(path);
             return Changelog.fromJson(json);
+        } catch (FileNotFoundException e) {
+            throw e;
         } catch (IOException e) {
             throw new RuntimeException("Fails to read changelog from path " + path, e);
         }

paimon-core/src/main/java/org/apache/paimon/utils/SnapshotManager.java

@@ -168,6 +168,11 @@ public Changelog changelog(long snapshotId) {
         return Changelog.fromPath(fileIO, changelogPath);
     }
 
+    private Changelog tryGetChangelog(long snapshotId) throws FileNotFoundException {
+        Path changelogPath = longLivedChangelogPath(snapshotId);
+        return Changelog.tryFromPath(fileIO, changelogPath);
+    }
+
     public Changelog longLivedChangelog(long snapshotId) {
         return Changelog.fromPath(fileIO, longLivedChangelogPath(snapshotId));
     }
@@ -208,8 +213,41 @@ public boolean longLivedChangelogExists(long snapshotId) {
     }
 
     public @Nullable Snapshot earliestSnapshot() {
-        Long snapshotId = earliestSnapshotId();
-        return snapshotId == null ? null : snapshot(snapshotId);
+        return earliestSnapshot(false);
+    }
+
+    private @Nullable Snapshot earliestSnapshot(boolean includeChangelog) {
+        Long snapshotId = null;
+        if (includeChangelog) {
+            snapshotId = earliestLongLivedChangelogId();
+        }
+        if (snapshotId == null) {
+            snapshotId = earliestSnapshotId();
+        }
+        if (snapshotId == null) {
+            return null;
+        }
+
+        FunctionWithException<Long, Snapshot, FileNotFoundException> snapshotFunction =
+                includeChangelog ? this::tryGetChangelogOrSnapshot : this::tryGetSnapshot;
+
+        // The loss of the earliest snapshot is an event of small probability, so the retry number
+        // here need not be too large.
+        int retry = 0;
+        do {
+            try {
+                return snapshotFunction.apply(snapshotId);
+            } catch (FileNotFoundException e) {
+                if (retry++ >= 3) {
+                    throw new RuntimeException(e);
+                }
+                LOG.warn(
+                        "The earliest snapshot or changelog was once identified but disappeared. "
+                                + "It might have been expired by other jobs operating on this table. "
+                                + "Searching for the second earliest snapshot or changelog instead. ");
+                snapshotId++;
+            }
+        } while (true);
     }
 
     public @Nullable Long earliestSnapshotId() {
@@ -268,28 +306,34 @@ private Snapshot changelogOrSnapshot(long snapshotId) {
         }
     }
 
+    private Snapshot tryGetChangelogOrSnapshot(long snapshotId) throws FileNotFoundException {
+        if (longLivedChangelogExists(snapshotId)) {
+            return tryGetChangelog(snapshotId);
+        } else {
+            return tryGetSnapshot(snapshotId);
+        }
+    }
+
     /**
      * Returns the latest snapshot earlier than the timestamp mills. A non-existent snapshot may be
      * returned if all snapshots are equal to or later than the timestamp mills.
      */
     public @Nullable Long earlierThanTimeMills(long timestampMills, boolean startFromChangelog) {
-        Long earliestSnapshot = earliestSnapshotId();
-        Long earliest;
-        if (startFromChangelog) {
-            Long earliestChangelog = earliestLongLivedChangelogId();
-            earliest = earliestChangelog == null ? earliestSnapshot : earliestChangelog;
-        } else {
-            earliest = earliestSnapshot;
-        }
         Long latest = latestSnapshotId();
-        if (earliest == null || latest == null) {
+        if (latest == null) {
+            return null;
+        }
+
+        Snapshot earliestSnapshot = earliestSnapshot(startFromChangelog);
+        if (earliestSnapshot == null) {
             return null;
         }
 
-        if (changelogOrSnapshot(earliest).timeMillis() >= timestampMills) {
-            return earliest - 1;
+        if (earliestSnapshot.timeMillis() >= timestampMills) {
+            return earliestSnapshot.id() - 1;
         }
 
+        long earliest = earliestSnapshot.id();
         while (earliest < latest) {
             long mid = (earliest + latest + 1) / 2;
             if (changelogOrSnapshot(mid).timeMillis() < timestampMills) {
@@ -306,16 +350,17 @@ private Snapshot changelogOrSnapshot(long snapshotId) {
      * mills. If there is no such a snapshot, returns null.
      */
     public @Nullable Snapshot earlierOrEqualTimeMills(long timestampMills) {
-        Long earliest = earliestSnapshotId();
         Long latest = latestSnapshotId();
-        if (earliest == null || latest == null) {
+        if (latest == null) {
             return null;
         }
 
-        Snapshot earliestSnapShot = snapshot(earliest);
-        if (earliestSnapShot.timeMillis() > timestampMills) {
+        Snapshot earliestSnapShot = earliestSnapshot();
+        if (earliestSnapShot == null || earliestSnapShot.timeMillis() > timestampMills) {
             return earliestSnapShot;
         }
+        long earliest = earliestSnapShot.id();
+
         Snapshot finalSnapshot = null;
         while (earliest <= latest) {
             long mid = earliest + (latest - earliest) / 2; // Avoid overflow
@@ -368,16 +413,22 @@ private Snapshot changelogOrSnapshot(long snapshotId) {
     }
 
     public @Nullable Snapshot earlierOrEqualWatermark(long watermark) {
-        Long earliest = earliestSnapshotId();
         Long latest = latestSnapshotId();
         // 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) {
+        if (latest == null || snapshot(latest).watermark() == Long.MIN_VALUE) {
             return null;
         }
+
+        Snapshot earliestSnapShot = earliestSnapshot();
+        if (earliestSnapShot == null) {
+            return null;
+        }
+        long earliest = earliestSnapShot.id();
+
         Long earliestWatermark = null;
         // find the first snapshot with watermark
-        if ((earliestWatermark = snapshot(earliest).watermark()) == null) {
+        if ((earliestWatermark = earliestSnapShot.watermark()) == null) {
             while (earliest < latest) {
                 earliest++;
                 earliestWatermark = snapshot(earliest).watermark();
@@ -427,16 +478,22 @@ private Snapshot changelogOrSnapshot(long snapshotId) {
     }
 
     public @Nullable Snapshot laterOrEqualWatermark(long watermark) {
-        Long earliest = earliestSnapshotId();
         Long latest = latestSnapshotId();
         // 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) {
+        if (latest == null || snapshot(latest).watermark() == Long.MIN_VALUE) {
             return null;
         }
+
+        Snapshot earliestSnapShot = earliestSnapshot();
+        if (earliestSnapShot == null) {
+            return null;
+        }
+        long earliest = earliestSnapShot.id();
+
         Long earliestWatermark = null;
         // find the first snapshot with watermark
-        if ((earliestWatermark = snapshot(earliest).watermark()) == null) {
+        if ((earliestWatermark = earliestSnapShot.watermark()) == null) {
             while (earliest < latest) {
                 earliest++;
                 earliestWatermark = snapshot(earliest).watermark();