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 d64164d53e43..947313feb3ae 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
@@ -665,7 +665,7 @@ public static List<Snapshot> findOverlappedSnapshots(
         List<Snapshot> overlappedSnapshots = new ArrayList<>();
         int right = findPreviousSnapshot(sortedSnapshots, endExclusive);
         if (right >= 0) {
-            int left = Math.max(findPreviousOrEqualSnapshot(sortedSnapshots, beginInclusive), 0);
+            int left = Math.max(findNextOrEqualTag(sortedSnapshots, beginInclusive), 0);
             for (int i = left; i <= right; i++) {
                 overlappedSnapshots.add(sortedSnapshots.get(i));
             }
@@ -682,10 +682,10 @@ public static int findPreviousSnapshot(List<Snapshot> sortedSnapshots, long targ
         return -1;
     }
 
-    private static int findPreviousOrEqualSnapshot(
-            List<Snapshot> sortedSnapshots, long targetSnapshotId) {
-        for (int i = sortedSnapshots.size() - 1; i >= 0; i--) {
-            if (sortedSnapshots.get(i).id() <= targetSnapshotId) {
+    private static int findNextOrEqualTag(
+            List<Snapshot> taggedSnapshots, long targetSnapshotId) {
+        for (int i = 0; i < taggedSnapshots.size(); i++) {
+            if (taggedSnapshots.get(i).id() >= targetSnapshotId) {
                 return i;
             }
         }
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 bdc86fff3e15..b879dadf963f 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
@@ -138,7 +138,7 @@ public void testlaterOrEqualWatermark() throws IOException {
         assertThat(snapshotManager.laterOrEqualWatermark(millis + 999)).isNull();
     }
 
-    public Snapshot createSnapshotWithMillis(long id, long millis) {
+    private Snapshot createSnapshotWithMillis(long id, long millis) {
         return new Snapshot(
                 id,
                 0L,
@@ -366,4 +366,23 @@ public void testLongLivedChangelog() throws Exception {
         Assertions.assertThat(snapshotManager.latestSnapshotId()).isEqualTo(10);
         Assertions.assertThat(snapshotManager.changelog(1)).isNotNull();
     }
+
+    @Test
+    public void testOverlappedSnapshots() {
+        SnapshotManagerTest snapshotManagerTest = new SnapshotManagerTest();
+        List<Snapshot> taggedSnapshot = new ArrayList<>();
+        long millis = System.currentTimeMillis();
+        long[] snapshotId = new long[] {8, 9, 11, 12, 15};
+        for (long id : snapshotId) {
+            taggedSnapshot.add(snapshotManagerTest.createSnapshotWithMillis(id, millis));
+        }
+
+        int beginInclusive = 10, endExclusive = 15;
+        // overlapped snapshot is [11,12]
+        List<Snapshot> overlappedSnapshots =
+                SnapshotManager.findOverlappedSnapshots(taggedSnapshot, beginInclusive, endExclusive);
+        List<Long> overlappedIds =
+                overlappedSnapshots.stream().map(Snapshot::id).collect(Collectors.toList());
+        assertThat(overlappedIds).containsExactly(11L, 12L);
+    }
 }
diff --git a/paimon-core/src/test/java/org/apache/paimon/utils/TagManagerTest.java b/paimon-core/src/test/java/org/apache/paimon/utils/TagManagerTest.java
index 86b928ca26fb..c114c830621e 100644
--- a/paimon-core/src/test/java/org/apache/paimon/utils/TagManagerTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/utils/TagManagerTest.java
@@ -133,25 +133,6 @@ public void testCreateTagWithTimeRetained() throws Exception {
         assertThat(tags.get(0).getValue()).contains("tag");
     }
 
-    @Test
-    public void testOverlappedSnapshots() {
-        SnapshotManagerTest snapshotManagerTest = new SnapshotManagerTest();
-        List<Snapshot> taggedSnapshot = new ArrayList<>();
-        long millis = System.currentTimeMillis();
-        long[] snapshotId = new long[] {7, 9, 11, 12};
-        for (long id : snapshotId) {
-            taggedSnapshot.add(snapshotManagerTest.createSnapshotWithMillis(id, millis));
-        }
-
-        int beginInclusive = 10, endExclusive = 15;
-        // overlapped snapshot is [11,12]
-        List<Snapshot> overlappedSnapshots =
-                TagManager.findOverlappedSnapshots(taggedSnapshot, beginInclusive, endExclusive);
-        List<Long> overlappedIds =
-                overlappedSnapshots.stream().map(Snapshot::id).collect(Collectors.toList());
-        assertThat(overlappedIds).containsExactly(11L, 12L);
-    }
-
     private TestFileStore createStore(TestKeyValueGenerator.GeneratorMode mode, int buckets)
             throws Exception {
         ThreadLocalRandom random = ThreadLocalRandom.current();