[flink-table-store] branch release-0.2 updated: [FLINK-29687] Do not throw exception when Lookup table is empty

[lz...@apache.org]

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

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


The following commit(s) were added to refs/heads/release-0.2 by this push:
     new 56c952e0 [FLINK-29687] Do not throw exception when Lookup table is empty
56c952e0 is described below

commit 56c952e059686a5096163356f563548bb1dc3f2e
Author: Jane Chan <55...@users.noreply.github.com>
AuthorDate: Fri Oct 21 11:22:53 2022 +0800

    [FLINK-29687] Do not throw exception when Lookup table is empty
    
    This closes #323
---
 .../table/store/connector/LookupJoinITCase.java    | 27 ++++++++++++++++++++++
 .../store/table/source/TableStreamingReader.java   |  6 +++--
 2 files changed, 31 insertions(+), 2 deletions(-)

diff --git a/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/LookupJoinITCase.java b/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/LookupJoinITCase.java
index b5cf0013..6b06aef0 100644
--- a/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/LookupJoinITCase.java
+++ b/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/LookupJoinITCase.java
@@ -59,6 +59,33 @@ public class LookupJoinITCase extends AbstractTestBase {
                         + " ('continuous.discovery-interval'='1 ms')");
     }
 
+    @Test
+    public void testLookupEmptyTable() throws Exception {
+        String query =
+                "SELECT T.i, D.j, D.k1, D.k2 FROM T LEFT JOIN DIM for system_time as of T.proctime AS D ON T.i = D.i";
+        BlockingIterator<Row, Row> iterator = BlockingIterator.of(env.executeSql(query).collect());
+
+        executeSql("INSERT INTO T VALUES (1), (2), (3)");
+
+        List<Row> result = iterator.collect(3);
+        assertThat(result)
+                .containsExactlyInAnyOrder(
+                        Row.of(1, null, null, null),
+                        Row.of(2, null, null, null),
+                        Row.of(3, null, null, null));
+
+        executeSql("INSERT INTO DIM VALUES (1, 11, 111, 1111), (2, 22, 222, 2222)");
+        Thread.sleep(2000); // wait refresh
+        executeSql("INSERT INTO T VALUES (1), (2), (4)");
+        result = iterator.collect(3);
+        assertThat(result)
+                .containsExactlyInAnyOrder(
+                        Row.of(1, 11, 111, 1111),
+                        Row.of(2, 22, 222, 2222),
+                        Row.of(4, null, null, null));
+        iterator.close();
+    }
+
     @Test
     public void testLookup() throws Exception {
         executeSql("INSERT INTO DIM VALUES (1, 11, 111, 1111), (2, 22, 222, 2222)");
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/source/TableStreamingReader.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/source/TableStreamingReader.java
index 58a420e2..0b79a512 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/source/TableStreamingReader.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/source/TableStreamingReader.java
@@ -35,7 +35,6 @@ import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Iterator;
 import java.util.List;
-import java.util.Objects;
 import java.util.function.IntUnaryOperator;
 import java.util.stream.IntStream;
 
@@ -90,7 +89,10 @@ public class TableStreamingReader {
                 scan.withFilter(predicate);
             }
             TableScan.Plan plan = scan.plan();
-            long snapshotId = Objects.requireNonNull(plan.snapshotId);
+            if (plan.snapshotId == null) {
+                return null;
+            }
+            long snapshotId = plan.snapshotId;
             enumerator =
                     new SnapshotEnumerator(
                             table.location(), scan.withIncremental(true), snapshotId);