[flink-table-store] branch master updated: [FLINK-28020] TableScan.Plan should accept Null snapshot id

[lzljs3620320]

This is an automated email from the ASF dual-hosted git repository.

lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-table-store.git


The following commit(s) were added to refs/heads/master by this push:
     new f5c8ee0a [FLINK-28020] TableScan.Plan should accept Null snapshot id
f5c8ee0a is described below

commit f5c8ee0ac6c3b7a488fa22388b746797e91ad6bd
Author: JingsongLi <lzljs3620...@aliyun.com>
AuthorDate: Mon Jun 13 18:09:54 2022 +0800

    [FLINK-28020] TableScan.Plan should accept Null snapshot id
---
 .../apache/flink/table/store/connector/AppendOnlyTableTest.java    | 5 +++++
 .../store/connector/source/FileStoreSourceSplitGeneratorTest.java  | 2 +-
 .../java/org/apache/flink/table/store/table/source/TableScan.java  | 7 +++++--
 3 files changed, 11 insertions(+), 3 deletions(-)

diff --git 
a/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/AppendOnlyTableTest.java
 
b/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/AppendOnlyTableTest.java
index 18a88b73..682c623c 100644
--- 
a/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/AppendOnlyTableTest.java
+++ 
b/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/AppendOnlyTableTest.java
@@ -48,6 +48,11 @@ public class AppendOnlyTableTest extends 
FileStoreTableITCase {
                                 + "want to keep the primary key definition.");
     }
 
+    @Test
+    public void testReadEmpty() {
+        assertThat(batchSql("SELECT * FROM append_table")).isEmpty();
+    }
+
     @Test
     public void testReadWrite() {
         batchSql("INSERT INTO append_table VALUES (1, 'AAA'), (2, 'BBB')");
diff --git 
a/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/source/FileStoreSourceSplitGeneratorTest.java
 
b/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/source/FileStoreSourceSplitGeneratorTest.java
index 5f56c136..254e6992 100644
--- 
a/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/source/FileStoreSourceSplitGeneratorTest.java
+++ 
b/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/source/FileStoreSourceSplitGeneratorTest.java
@@ -79,7 +79,7 @@ public class FileStoreSourceSplitGeneratorTest {
                 };
         TableScan.Plan tableScanPlan =
                 new TableScan.Plan(
-                        1,
+                        1L,
                         new DefaultSplitGenerator(
                                         new FileStorePathFactory(new 
Path(tempDir.toString())))
                                 .generate(plan.groupByPartFiles()));
diff --git 
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/source/TableScan.java
 
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/source/TableScan.java
index f49733fa..ada81909 100644
--- 
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/source/TableScan.java
+++ 
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/source/TableScan.java
@@ -27,6 +27,8 @@ import 
org.apache.flink.table.store.file.predicate.PredicateBuilder;
 import org.apache.flink.table.store.file.schema.Schema;
 import org.apache.flink.table.store.file.utils.FileStorePathFactory;
 
+import javax.annotation.Nullable;
+
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Optional;
@@ -123,11 +125,12 @@ public abstract class TableScan {
 
     /** Scanning plan containing snapshot ID and input splits. */
     public static class Plan {
-        public final long snapshotId;
+
+        @Nullable public final Long snapshotId;
         public final List<Split> splits;
 
         @VisibleForTesting
-        public Plan(long snapshotId, List<Split> splits) {
+        public Plan(@Nullable Long snapshotId, List<Split> splits) {
             this.snapshotId = snapshotId;
             this.splits = splits;
         }