fix

Author: JingsongLi

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

@@ -148,7 +148,7 @@ public void testSnapshotsTableWithRecordCount() throws Exception {
                 sql(
                         "SELECT snapshot_id, total_record_count, delta_record_count, changelog_record_count FROM T$snapshots");
         assertThat(result)
-                .containsExactlyInAnyOrder(Row.of(1L, 1L, 1L, 0L), Row.of(2L, 2L, 1L, 0L));
+                .containsExactlyInAnyOrder(Row.of(1L, 1L, 1L, null), Row.of(2L, 2L, 1L, null));
     }
 
     @Test

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

@@ -27,7 +27,6 @@
 import org.apache.paimon.shade.guava30.com.google.common.collect.ImmutableList;
 
 import org.apache.flink.table.api.StatementSet;
-import org.apache.flink.table.api.ValidationException;
 import org.apache.flink.table.planner.factories.TestValuesTableFactory;
 import org.apache.flink.types.Row;
 import org.apache.flink.types.RowKind;
@@ -227,7 +226,7 @@ public void testContinuousLatest() throws Exception {
 
         BlockingIterator<Row, Row> iterator =
                 BlockingIterator.of(
-                        streamSqlIter("SELECT * FROM T1 /*+ OPTIONS('log.scan'='latest') */"));
+                        streamSqlIter("SELECT * FROM T1 /*+ OPTIONS('scan.mode'='latest') */"));
 
         batchSql("INSERT INTO T1 VALUES ('7', '8', '9'), ('10', '11', '12')");
         assertThat(iterator.collect(2))
@@ -255,7 +254,7 @@ public void testContinuousLatestStartingFromEmpty() throws Exception {
     @Test
     public void testContinuousFromTimestamp() throws Exception {
         String sql =
-                "SELECT * FROM T1 /*+ OPTIONS('log.scan'='from-timestamp', 'log.scan.timestamp-millis'='%s') */";
+                "SELECT * FROM T1 /*+ OPTIONS('scan.mode'='from-timestamp', 'scan.timestamp-millis'='%s') */";
 
         // empty table
         BlockingIterator<Row, Row> iterator = BlockingIterator.of(streamSqlIter(sql, 0));
@@ -323,31 +322,31 @@ public void testLackStartupTimestamp() {
         assertThatThrownBy(
                         () ->
                                 streamSqlIter(
-                                        "SELECT * FROM T1 /*+ OPTIONS('log.scan'='from-timestamp') */"))
+                                        "SELECT * FROM T1 /*+ OPTIONS('scan.mode'='from-timestamp') */"))
                 .hasCauseInstanceOf(IllegalArgumentException.class)
                 .hasRootCauseMessage(
                         "must set only one key in [scan.timestamp-millis,scan.timestamp] when you use from-timestamp for scan.mode");
     }
 
     @Test
     public void testConfigureStartupTimestamp() throws Exception {
-        // Configure 'log.scan.timestamp-millis' without 'log.scan'.
+        // Configure 'scan.timestamp-millis' without 'scan.mode'.
         BlockingIterator<Row, Row> iterator =
                 BlockingIterator.of(
                         streamSqlIter(
-                                "SELECT * FROM T1 /*+ OPTIONS('log.scan.timestamp-millis'='%s') */",
+                                "SELECT * FROM T1 /*+ OPTIONS('scan.timestamp-millis'='%s') */",
                                 0));
         batchSql("INSERT INTO T1 VALUES ('1', '2', '3'), ('4', '5', '6')");
         batchSql("INSERT INTO T1 VALUES ('7', '8', '9'), ('10', '11', '12')");
         assertThat(iterator.collect(2))
                 .containsExactlyInAnyOrder(Row.of("1", "2", "3"), Row.of("4", "5", "6"));
         iterator.close();
 
-        // Configure 'log.scan.timestamp-millis' with 'log.scan=latest'.
+        // Configure 'scan.timestamp-millis' with 'scan.mode=latest'.
         assertThatThrownBy(
                         () ->
                                 streamSqlIter(
-                                        "SELECT * FROM T1 /*+ OPTIONS('log.scan'='latest', 'log.scan.timestamp-millis'='%s') */",
+                                        "SELECT * FROM T1 /*+ OPTIONS('scan.mode'='latest', 'scan.timestamp-millis'='%s') */",
                                         0))
                 .hasCauseInstanceOf(IllegalArgumentException.class)
                 .hasRootCauseMessage(
@@ -466,28 +465,6 @@ public void testIgnoreOverwrite() throws Exception {
         iterator.close();
     }
 
-    @Test
-    public void testUnsupportedUpsert() {
-        assertThatThrownBy(
-                        () ->
-                                streamSqlIter(
-                                        "SELECT * FROM T1 /*+ OPTIONS('log.changelog-mode'='upsert') */"))
-                .hasCauseInstanceOf(ValidationException.class)
-                .hasRootCauseMessage(
-                        "File store continuous reading does not support upsert changelog mode.");
-    }
-
-    @Test
-    public void testUnsupportedEventual() {
-        assertThatThrownBy(
-                        () ->
-                                streamSqlIter(
-                                        "SELECT * FROM T1 /*+ OPTIONS('log.consistency'='eventual') */"))
-                .hasCauseInstanceOf(ValidationException.class)
-                .hasRootCauseMessage(
-                        "File store continuous reading does not support eventual consistency mode.");
-    }
-
     @Test
     public void testFlinkMemoryPool() {
         // Check if the configuration is effective