[core] StreamScan should skip empty plan to more timely

Author: JingsongLi

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

@@ -220,6 +220,9 @@ private Plan nextPlan() {
                 SnapshotReader.Plan overwritePlan = handleOverwriteSnapshot(snapshot);
                 if (overwritePlan != null) {
                     nextSnapshotId++;
+                    if (overwritePlan.splits().isEmpty()) {
+                        continue;
+                    }
                     return overwritePlan;
                 }
             }
@@ -229,6 +232,9 @@ private Plan nextPlan() {
                 SnapshotReader.Plan plan = followUpScanner.scan(snapshot, snapshotReader);
                 currentWatermark = plan.watermark();
                 nextSnapshotId++;
+                if (plan.splits().isEmpty()) {
+                    continue;
+                }
                 return plan;
             } else {
                 nextSnapshotId++;

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

@@ -45,6 +45,52 @@
 /** Tests for {@link StreamTableScan}. */
 public class StreamTableScanTest extends ScannerTestBase {
 
+    @Test
+    public void testIgnoreEmptyPlan() throws Exception {
+        TableRead read = table.newRead();
+        StreamTableWrite write = table.newWrite(commitUser);
+        StreamTableCommit commit = table.newCommit(commitUser);
+
+        // only plan partition 1
+        ReadBuilder readBuilder = table.newReadBuilder();
+        PredicateBuilder predicateBuilder = new PredicateBuilder(readBuilder.readType());
+        readBuilder.withFilter(predicateBuilder.lessOrEqual(0, 1));
+        StreamTableScan scan = readBuilder.newStreamScan();
+
+        // first call without any snapshot, should return empty plan
+        assertThat(scan.plan().splits()).isEmpty();
+
+        // write partition 1
+        write.write(rowData(1, 10, 100L));
+        commit.commit(0, write.prepareCommit(true, 0));
+
+        // first call with snapshot, should return complete records from 1 commit
+        TableScan.Plan plan = scan.plan();
+        assertThat(getResult(read, plan.splits()))
+                .hasSameElementsAs(Collections.singletonList("+I 1|10|100"));
+
+        // write partition 2
+        write.write(rowData(2, 10, 100L));
+        commit.commit(1, write.prepareCommit(true, 1));
+
+        // write partition 1
+        write.write(rowData(1, 20, 200L));
+        commit.commit(2, write.prepareCommit(true, 2));
+
+        // second call with snapshot, should ignore 2 commit and return complete records from 3
+        // commit
+        plan = scan.plan();
+        assertThat(getResult(read, plan.splits()))
+                .hasSameElementsAs(Collections.singletonList("+I 1|20|200"));
+
+        // third call without snapshot
+        plan = scan.plan();
+        assertThat(plan.splits()).isEmpty();
+
+        write.close();
+        commit.close();
+    }
+
     @Test
     public void testPlan() throws Exception {
         TableRead read = table.newRead();

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/FullCacheLookupTable.java

@@ -265,20 +265,11 @@ public void refresh() throws Exception {
     }
 
     private void doRefresh() throws Exception {
-        Long latestSnapshotId = context.table.snapshotManager().latestSnapshotId();
-        if (latestSnapshotId == null) {
-            return;
-        }
-
         while (true) {
             try (RecordReaderIterator<InternalRow> batch =
                     new RecordReaderIterator<>(reader.nextBatch(false))) {
                 if (!batch.hasNext()) {
-                    if (reader.nextSnapshotId() > latestSnapshotId) {
-                        return;
-                    } else {
-                        continue;
-                    }
+                    return;
                 }
                 refresh(batch);
             }