[core] Compacted StartingScanner is incorrect

paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/CompactedStartingScanner.java

@@ -35,16 +35,29 @@ public class CompactedStartingScanner implements StartingScanner {
     @Override
     @Nullable
     public Result scan(SnapshotManager snapshotManager, SnapshotSplitReader snapshotSplitReader) {
-        Long startingSnapshotId = snapshotManager.latestCompactedSnapshotId();
+        Long startingSnapshotId = pick(snapshotManager);
         if (startingSnapshotId == null) {
-            LOG.debug("There is currently no compact snapshot. Waiting for snapshot generation.");
-            return null;
+            startingSnapshotId = snapshotManager.latestSnapshotId();
+            if (startingSnapshotId == null) {
+                LOG.debug("There is currently no snapshot. Wait for the snapshot generation.");
+                return null;
+            } else {
+                LOG.debug(
+                        "No compact snapshot found, reading from the latest snapshot {}.",
+                        startingSnapshotId);
+            }
         }
+
         return new Result(
                 startingSnapshotId,
                 snapshotSplitReader
                         .withKind(ScanKind.ALL)
                         .withSnapshot(startingSnapshotId)
                         .splits());
     }
+
+    @Nullable
+    protected Long pick(SnapshotManager snapshotManager) {
+        return snapshotManager.latestCompactedSnapshotId();
+    }
 }

paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/FullCompactedStartingScanner.java

@@ -21,21 +21,15 @@
 import org.apache.paimon.CoreOptions.StartupMode;
 import org.apache.paimon.Snapshot;
 import org.apache.paimon.Snapshot.CommitKind;
-import org.apache.paimon.operation.ScanKind;
 import org.apache.paimon.utils.SnapshotManager;
 
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
 import javax.annotation.Nullable;
 
 /**
  * {@link StartingScanner} for the {@link StartupMode#COMPACTED_FULL} startup mode with
  * 'full-compaction.delta-commits'.
  */
-public class FullCompactedStartingScanner implements StartingScanner {
-
-    private static final Logger LOG = LoggerFactory.getLogger(FullCompactedStartingScanner.class);
+public class FullCompactedStartingScanner extends CompactedStartingScanner {
 
     private final int deltaCommits;
 
@@ -45,18 +39,8 @@ public FullCompactedStartingScanner(int deltaCommits) {
 
     @Override
     @Nullable
-    public Result scan(SnapshotManager snapshotManager, SnapshotSplitReader snapshotSplitReader) {
-        Long startingSnapshotId = snapshotManager.pickSnapshot(this::picked);
-        if (startingSnapshotId == null) {
-            LOG.debug("There is currently no compact snapshot. Waiting for snapshot generation.");
-            return null;
-        }
-        return new Result(
-                startingSnapshotId,
-                snapshotSplitReader
-                        .withKind(ScanKind.ALL)
-                        .withSnapshot(startingSnapshotId)
-                        .splits());
+    protected Long pick(SnapshotManager snapshotManager) {
+        return snapshotManager.pickFromLatest(this::picked);
     }
 
     private boolean picked(Snapshot snapshot) {

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

@@ -100,10 +100,10 @@ public boolean snapshotExists(long snapshotId) {
     }
 
     public @Nullable Long latestCompactedSnapshotId() {
-        return pickSnapshot(s -> s.commitKind() == CommitKind.COMPACT);
+        return pickFromLatest(s -> s.commitKind() == CommitKind.COMPACT);
     }
 
-    public @Nullable Long pickSnapshot(Predicate<Snapshot> predicate) {
+    public @Nullable Long pickFromLatest(Predicate<Snapshot> predicate) {
         Long latestId = latestSnapshotId();
         Long earliestId = earliestSnapshotId();
         if (latestId == null || earliestId == null) {

paimon-core/src/test/java/org/apache/paimon/table/source/snapshot/CompactedStartingScannerTest.java

@@ -86,7 +86,9 @@ public void testNoCompactSnapshot() throws Exception {
         assertThat(snapshotManager.latestSnapshotId()).isEqualTo(1);
 
         CompactedStartingScanner scanner = new CompactedStartingScanner();
-        assertThat(scanner.scan(snapshotManager, snapshotSplitReader)).isNull();
+
+        // No compact snapshot found, reading from the latest snapshot
+        assertThat(scanner.scan(snapshotManager, snapshotSplitReader).snapshotId()).isEqualTo(1);
 
         write.close();
         commit.close();

paimon-core/src/test/java/org/apache/paimon/table/source/snapshot/FullCompactedStartingScannerTest.java

@@ -0,0 +1,94 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.table.source.snapshot;
+
+import org.apache.paimon.table.sink.StreamTableCommit;
+import org.apache.paimon.table.sink.StreamTableWrite;
+import org.apache.paimon.types.RowKind;
+import org.apache.paimon.utils.SnapshotManager;
+
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link FullCompactedStartingScanner}. */
+public class FullCompactedStartingScannerTest extends ScannerTestBase {
+
+    @Test
+    public void testScan() throws Exception {
+        SnapshotManager snapshotManager = table.snapshotManager();
+        StreamTableWrite write = table.newWrite(commitUser);
+        StreamTableCommit commit = table.newCommit(commitUser);
+
+        for (int i = 0; i < 5; i++) {
+            write.write(rowData(1, 10, 102L));
+            write.write(rowData(1, 20, 201L));
+            write.compact(binaryRow(1), 0, true);
+            commit.commit(i, write.prepareCommit(true, i));
+        }
+
+        assertThat(snapshotManager.latestSnapshotId()).isEqualTo(10);
+
+        FullCompactedStartingScanner scanner = new FullCompactedStartingScanner(3);
+        StartingScanner.Result result = scanner.scan(snapshotManager, snapshotSplitReader);
+        assertThat(result.snapshotId()).isEqualTo(8);
+
+        write.close();
+        commit.close();
+    }
+
+    @Test
+    public void testNoSnapshot() {
+        SnapshotManager snapshotManager = table.snapshotManager();
+        FullCompactedStartingScanner scanner = new FullCompactedStartingScanner(3);
+        assertThat(scanner.scan(snapshotManager, snapshotSplitReader)).isNull();
+    }
+
+    @Test
+    public void testNoCompactSnapshot() throws Exception {
+        SnapshotManager snapshotManager = table.snapshotManager();
+        StreamTableWrite write = table.newWrite(commitUser);
+        StreamTableCommit commit = table.newCommit(commitUser);
+
+        write.write(rowData(1, 10, 100L));
+        write.write(rowData(1, 20, 200L));
+        write.write(rowData(1, 40, 400L));
+        commit.commit(0, write.prepareCommit(true, 0));
+
+        write.write(rowData(1, 10, 101L));
+        write.write(rowData(1, 30, 300L));
+        write.write(rowDataWithKind(RowKind.DELETE, 1, 40, 400L));
+        write.compact(binaryRow(1), 0, true);
+        commit.commit(1, write.prepareCommit(true, 1));
+
+        write.write(rowData(1, 10, 102L));
+        write.write(rowData(1, 20, 201L));
+        commit.commit(2, write.prepareCommit(true, 2));
+
+        assertThat(snapshotManager.latestSnapshotId()).isEqualTo(4);
+
+        FullCompactedStartingScanner scanner = new FullCompactedStartingScanner(3);
+
+        // No compact snapshot found, reading from the latest snapshot
+        assertThat(scanner.scan(snapshotManager, snapshotSplitReader).snapshotId()).isEqualTo(4);
+
+        write.close();
+        commit.close();
+    }
+}

paimon-flink/paimon-flink-1.14/src/test/java/org/apache/paimon/flink/BatchFileStoreITCase.java

@@ -51,13 +51,6 @@ public void testOverwriteEmpty() {
         assertThat(batchSql("SELECT * FROM T")).isEmpty();
     }
 
-    @Test
-    public void testCompactedScanModeEmpty() {
-        batchSql("INSERT INTO T VALUES (1, 11, 111), (2, 22, 222)");
-        assertThat(batchSql("SELECT * FROM T /*+ OPTIONS('scan.mode'='compacted-full') */"))
-                .isEmpty();
-    }
-
     @Test
     public void testTimeTravelRead() throws InterruptedException {
         batchSql("INSERT INTO T VALUES (1, 11, 111), (2, 22, 222)");

paimon-flink/paimon-flink-1.15/src/test/java/org/apache/paimon/flink/BatchFileStoreITCase.java

@@ -51,13 +51,6 @@ public void testOverwriteEmpty() {
         assertThat(batchSql("SELECT * FROM T")).isEmpty();
     }
 
-    @Test
-    public void testCompactedScanModeEmpty() {
-        batchSql("INSERT INTO T VALUES (1, 11, 111), (2, 22, 222)");
-        assertThat(batchSql("SELECT * FROM T /*+ OPTIONS('scan.mode'='compacted-full') */"))
-                .isEmpty();
-    }
-
     @Test
     public void testTimeTravelRead() throws InterruptedException {
         batchSql("INSERT INTO T VALUES (1, 11, 111), (2, 22, 222)");

paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/BatchFileStoreITCase.java

@@ -46,13 +46,6 @@ public void testOverwriteEmpty() {
         assertThat(batchSql("SELECT * FROM T")).isEmpty();
     }
 
-    @Test
-    public void testCompactedScanModeEmpty() {
-        batchSql("INSERT INTO T VALUES (1, 11, 111), (2, 22, 222)");
-        assertThat(batchSql("SELECT * FROM T /*+ OPTIONS('scan.mode'='compacted-full') */"))
-                .isEmpty();
-    }
-
     @Test
     public void testTimeTravelRead() throws InterruptedException {
         batchSql("INSERT INTO T VALUES (1, 11, 111), (2, 22, 222)");