[AMORO-2977]Mixed-Format Unkeyed Table will read the full amount of data even if configured with 'scan.startup.mode'='latest'

amoro-mixed-format/amoro-mixed-format-flink/amoro-mixed-format-flink-common/src/main/java/org/apache/amoro/flink/table/FlinkSource.java

@@ -48,13 +48,18 @@
 import org.apache.flink.table.data.RowData;
 import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
 import org.apache.flink.table.types.logical.RowType;
+import org.apache.iceberg.Snapshot;
 import org.apache.iceberg.expressions.Expression;
 import org.apache.iceberg.flink.FlinkSchemaUtil;
 import org.apache.iceberg.flink.source.FlinkInputFormat;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
 
 /** An util class create mixed-format source data stream. */
 public class FlinkSource {
@@ -66,6 +71,7 @@ public static Builder forRowData() {
 
   public static final class Builder {
 
+    private static final Logger LOG = LoggerFactory.getLogger(Builder.class);
     private static final String MIXED_FORMAT_FILE_TRANSFORMATION = "mixed-format-file";
     private ProviderContext context;
     private StreamExecutionEnvironment env;
@@ -148,7 +154,8 @@ public DataStream<RowData> build() {
       loadTableIfNeeded();
 
       if (mixedTable.isUnkeyedTable()) {
-        return buildUnkeyedTableSource();
+        String scanStartupMode = properties.get(MixedFormatValidator.SCAN_STARTUP_MODE.key());
+        return buildUnkeyedTableSource(scanStartupMode);
       }
 
       boolean dimTable =
@@ -214,17 +221,40 @@ private void loadTableIfNeeded() {
       properties.putAll(mixedTable.properties());
     }
 
-    public DataStream<RowData> buildUnkeyedTableSource() {
-      DataStream<RowData> origin =
+    public DataStream<RowData> buildUnkeyedTableSource(String scanStartupMode) {
+      scanStartupMode = scanStartupMode == null ? null : scanStartupMode.toLowerCase();
+      Preconditions.checkArgument(
+          Objects.isNull(scanStartupMode)
+              || Objects.equals(scanStartupMode, MixedFormatValidator.SCAN_STARTUP_MODE_EARLIEST)
+              || Objects.equals(scanStartupMode, MixedFormatValidator.SCAN_STARTUP_MODE_LATEST),
+          String.format(
+              "only support %s, %s when %s is %s",
+              MixedFormatValidator.SCAN_STARTUP_MODE_EARLIEST,
+              MixedFormatValidator.SCAN_STARTUP_MODE_LATEST,
+              MixedFormatValidator.MIXED_FORMAT_READ_MODE,
+              MixedFormatValidator.MIXED_FORMAT_READ_FILE));
+      org.apache.iceberg.flink.source.FlinkSource.Builder builder =
           org.apache.iceberg.flink.source.FlinkSource.forRowData()
               .env(env)
               .project(org.apache.amoro.flink.FlinkSchemaUtil.filterWatermark(projectedSchema))
               .tableLoader(tableLoader)
               .filters(filters)
               .properties(properties)
               .flinkConf(flinkConf)
-              .limit(limit)
-              .build();
+              .limit(limit);
+      if (MixedFormatValidator.SCAN_STARTUP_MODE_LATEST.equalsIgnoreCase(scanStartupMode)) {
+        Optional<Snapshot> startSnapshotOptional =
+            Optional.ofNullable(tableLoader.loadTable().currentSnapshot());
+        if (startSnapshotOptional.isPresent()) {
+          Snapshot snapshot = startSnapshotOptional.get();
+          LOG.info(
+              "Get starting snapshot id {} based on scan startup mode {}",
+              snapshot.snapshotId(),
+              scanStartupMode);
+          builder.startSnapshotId(snapshot.snapshotId());
+        }
+      }
+      DataStream<RowData> origin = builder.build();
       return wrapKrb(origin).assignTimestampsAndWatermarks(watermarkStrategy);
     }
 

amoro-mixed-format/amoro-mixed-format-flink/amoro-mixed-format-flink-common/src/test/java/org/apache/amoro/flink/table/TestUnkeyed.java

@@ -352,34 +352,73 @@ public void testSinkStreamRead() throws Exception {
     sql("CREATE TABLE IF NOT EXISTS mixed_catalog." + db + "." + TABLE + "(id INT, name STRING)");
 
     sql("insert into mixed_catalog." + db + "." + TABLE + " select * from input");
-    sql("insert into mixed_catalog." + db + "." + TABLE + " select * from input");
-
-    MixedTable table = mixedFormatCatalog.loadTable(TableIdentifier.of(catalog, db, TABLE));
 
-    Iterable<Snapshot> snapshots = table.asUnkeyedTable().snapshots();
-    Snapshot s = snapshots.iterator().next();
-
-    TableResult result =
+    // verify in earliest scan-startup-mode file read
+    TableResult resultWithEarliestPosition =
         exec(
             "select * from mixed_catalog."
                 + db
                 + "."
                 + TABLE
                 + "/*+ OPTIONS("
-                + "'mixed-format.read.mode'='file'"
-                + ", 'start-snapshot-id'='"
-                + s.snapshotId()
-                + "'"
+                + "'streaming'='true'"
+                + ", 'mixed-format.read.mode'='file'"
+                + ", 'scan.startup.mode'='earliest'"
+                + ", 'monitor-interval'='1s'"
                 + ")*/");
 
     Set<Row> actual = new HashSet<>();
-    try (CloseableIterator<Row> iterator = result.collect()) {
+    try (CloseableIterator<Row> iterator = resultWithEarliestPosition.collect()) {
       for (int i = 0; i < data.size(); i++) {
         actual.add(iterator.next());
       }
     }
-    result.getJobClient().ifPresent(TestUtil::cancelJob);
+    resultWithEarliestPosition.getJobClient().ifPresent(TestUtil::cancelJob);
     Assert.assertEquals(DataUtil.toRowSet(data), actual);
+
+    // verify in latest scan-startup-mode file read
+    TableResult resultWithLatestPosition =
+        exec(
+            "select * from mixed_catalog."
+                + db
+                + "."
+                + TABLE
+                + "/*+ OPTIONS("
+                + "'streaming'='true'"
+                + ", 'mixed-format.read.mode'='file'"
+                + ", 'scan.startup.mode'='latest'"
+                + ", 'monitor-interval'='1s'"
+                + ")*/");
+
+    List<Object[]> appendData = new LinkedList<>();
+    appendData.add(new Object[] {2000004, "a"});
+    appendData.add(new Object[] {2000015, "b"});
+    appendData.add(new Object[] {2000011, "c"});
+    appendData.add(new Object[] {2000014, "d"});
+    appendData.add(new Object[] {2000021, "d"});
+    appendData.add(new Object[] {2000007, "e"});
+
+    List<ApiExpression> appendRows = DataUtil.toRows(appendData);
+
+    Table appendInput =
+        getTableEnv()
+            .fromValues(
+                DataTypes.ROW(
+                    DataTypes.FIELD("id", DataTypes.INT()),
+                    DataTypes.FIELD("name", DataTypes.STRING())),
+                appendRows);
+    getTableEnv().createTemporaryView("appendInput", appendInput);
+
+    actual.clear();
+    try (CloseableIterator<Row> iterator = resultWithLatestPosition.collect()) {
+      sql("insert into mixed_catalog." + db + "." + TABLE + " select * from appendInput");
+      for (int i = 0; i < appendData.size(); i++) {
+        Assert.assertTrue("Should have more records", iterator.hasNext());
+        actual.add(iterator.next());
+      }
+    }
+    resultWithLatestPosition.getJobClient().ifPresent(TestUtil::cancelJob);
+    Assert.assertEquals(DataUtil.toRowSet(appendData), actual);
   }
 
   @Test