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
index 8cd159997865..5ad2f1c254ec 100644
--- a/paimon-core/src/main/java/org/apache/paimon/utils/SnapshotManager.java
+++ b/paimon-core/src/main/java/org/apache/paimon/utils/SnapshotManager.java
@@ -360,16 +360,20 @@ public long repairEarliestSnapshot(long snapshotId) {
      * mills. If there is no such a snapshot, returns null.
      */
     public @Nullable Snapshot earlierOrEqualTimeMills(long timestampMills) {
-        Long latest = latestSnapshotId();
-        if (latest == null) {
+        Snapshot latestSnapshot = latestSnapshot();
+        if (latestSnapshot == null) {
             return null;
         }
+        if (latestSnapshot.timeMillis() <= timestampMills) {
+            return latestSnapshot;
+        }
 
-        Snapshot earliestSnapShot = earliestSnapshot(latest);
-        if (earliestSnapShot == null || earliestSnapShot.timeMillis() > timestampMills) {
+        Snapshot earliestSnapshot = earliestSnapshot(latestSnapshot.id());
+        if (earliestSnapshot == null || earliestSnapshot.timeMillis() > timestampMills) {
             return null;
         }
-        long earliest = earliestSnapShot.id();
+        long earliest = earliestSnapshot.id();
+        long latest = latestSnapshot.id() - 1;
 
         Snapshot finalSnapshot = null;
         while (earliest <= latest) {
@@ -382,8 +386,8 @@ public long repairEarliestSnapshot(long snapshotId) {
                 earliest = mid + 1; // Search in the right half
                 finalSnapshot = snapshot;
             } else {
-                finalSnapshot = snapshot; // Found the exact match
-                break;
+                finalSnapshot = snapshot;
+                earliest = mid + 1;
             }
         }
         return finalSnapshot;
@@ -394,16 +398,21 @@ public long repairEarliestSnapshot(long snapshotId) {
      * If there is no such a snapshot, returns null.
      */
     public @Nullable Snapshot laterOrEqualTimeMills(long timestampMills) {
-        Long earliest = earliestSnapshotId();
-        Long latest = latestSnapshotId();
-        if (earliest == null || latest == null) {
+        Snapshot latestSnapshot = latestSnapshot();
+        if (latestSnapshot == null || latestSnapshot.timeMillis() < timestampMills) {
             return null;
         }
 
-        Snapshot latestSnapShot = snapshot(latest);
-        if (latestSnapShot.timeMillis() < timestampMills) {
+        Snapshot earliestSnapshot = earliestSnapshot(latestSnapshot.id());
+        if (earliestSnapshot == null) {
             return null;
         }
+        if (earliestSnapshot.timeMillis() >= timestampMills) {
+            return earliestSnapshot;
+        }
+
+        long earliest = earliestSnapshot.id() + 1;
+        long latest = latestSnapshot.id();
         Snapshot finalSnapshot = null;
         while (earliest <= latest) {
             long mid = earliest + (latest - earliest) / 2; // Avoid overflow
@@ -415,8 +424,8 @@ public long repairEarliestSnapshot(long snapshotId) {
             } else if (commitTime < timestampMills) {
                 earliest = mid + 1; // Search in the right half
             } else {
-                finalSnapshot = snapshot; // Found the exact match
-                break;
+                finalSnapshot = snapshot;
+                latest = mid - 1;
             }
         }
         return finalSnapshot;
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
index bf1a005878a9..a66a00ab8541 100644
--- a/paimon-core/src/test/java/org/apache/paimon/utils/SnapshotManagerTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/utils/SnapshotManagerTest.java
@@ -423,6 +423,64 @@ public void testLaterOrEqualTimeMills() throws IOException {
         assertThat(snapshotManager.laterOrEqualTimeMills(millis + 10001)).isNull();
     }
 
+    @Test
+    public void testEarlierOrEqualTimeMillsWithDuplicateCommitTimes() throws IOException {
+        long millis = 1684726826L;
+        FileIO localFileIO = LocalFileIO.create();
+        SnapshotManager snapshotManager =
+                newSnapshotManager(localFileIO, new Path(tempDir.toString()));
+        for (long i = 0; i < 3; i++) {
+            Snapshot snapshot = createSnapshotWithMillis(i, millis);
+            localFileIO.tryToWriteAtomic(snapshotManager.snapshotPath(i), snapshot.toJson());
+        }
+
+        assertThat(snapshotManager.earlierOrEqualTimeMills(millis).id()).isEqualTo(2);
+    }
+
+    @Test
+    public void testLaterOrEqualTimeMillsWithDuplicateCommitTimes() throws IOException {
+        long millis = 1684726826L;
+        FileIO localFileIO = LocalFileIO.create();
+        SnapshotManager snapshotManager =
+                newSnapshotManager(localFileIO, new Path(tempDir.toString()));
+        for (long i = 0; i < 3; i++) {
+            Snapshot snapshot = createSnapshotWithMillis(i, millis);
+            localFileIO.tryToWriteAtomic(snapshotManager.snapshotPath(i), snapshot.toJson());
+        }
+
+        assertThat(snapshotManager.laterOrEqualTimeMills(millis).id()).isEqualTo(0);
+    }
+
+    @Test
+    public void testEarlierOrEqualTimeMillsWithConcurrentRollback() throws IOException {
+        long millis = 1684726826L;
+        FileIO localFileIO = LocalFileIO.create();
+        SnapshotManager snapshotManager =
+                new LatestSnapshotRollbackRaceManager(localFileIO, new Path(tempDir.toString()));
+        for (long i = 0; i < 3; i++) {
+            Snapshot snapshot = createSnapshotWithMillis(i, millis);
+            localFileIO.tryToWriteAtomic(snapshotManager.snapshotPath(i), snapshot.toJson());
+        }
+        snapshotManager.commitLatestHint(2);
+
+        assertThat(snapshotManager.earlierOrEqualTimeMills(millis).id()).isEqualTo(1);
+    }
+
+    @Test
+    public void testLaterOrEqualTimeMillsWithConcurrentExpiration() throws IOException {
+        long millis = 1684726826L;
+        FileIO localFileIO = LocalFileIO.create();
+        SnapshotManager snapshotManager =
+                new TestSnapshotManager(localFileIO, new Path(tempDir.toString()), true);
+        for (long i = 0; i < 3; i++) {
+            Snapshot snapshot = createSnapshotWithMillis(i, millis);
+            localFileIO.tryToWriteAtomic(snapshotManager.snapshotPath(i), snapshot.toJson());
+        }
+        snapshotManager.commitEarliestHint(0);
+
+        assertThat(snapshotManager.laterOrEqualTimeMills(millis).id()).isEqualTo(1);
+    }
+
     @ParameterizedTest
     @ValueSource(booleans = {true, false})
     public void testLaterOrEqualWatermark(boolean isRaceCondition) throws IOException {
@@ -955,4 +1013,28 @@ private LatestSnapshotRaceManager(FileIO fileIO, Path tablePath, long millis) {
             return snapshotId;
         }
     }
+
+    /** Simulates a rollback after finding the latest snapshot ID. */
+    private static class LatestSnapshotRollbackRaceManager extends SnapshotManager {
+        private boolean rollbackLatestSnapshot = true;
+
+        private LatestSnapshotRollbackRaceManager(FileIO fileIO, Path tablePath) {
+            super(fileIO, tablePath, DEFAULT_MAIN_BRANCH, null, null);
+        }
+
+        @Override
+        public @Nullable Long latestSnapshotIdFromFileSystem() {
+            Long snapshotId = super.latestSnapshotIdFromFileSystem();
+            if (snapshotId != null && rollbackLatestSnapshot) {
+                try {
+                    commitLatestHint(snapshotId - 1);
+                    fileIO().delete(snapshotPath(snapshotId), true);
+                } catch (IOException e) {
+                    throw new RuntimeException(e);
+                }
+                rollbackLatestSnapshot = false;
+            }
+            return snapshotId;
+        }
+    }
 }