flink-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From twal...@apache.org
Subject flink git commit: [FLINK-7698] [table] Tests joins with null literals
Date Thu, 16 Nov 2017 10:06:26 GMT
Repository: flink
Updated Branches:
  refs/heads/release-1.4 13962e1ff -> 397f0d150


[FLINK-7698] [table] Tests joins with null literals


Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/397f0d15
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/397f0d15
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/397f0d15

Branch: refs/heads/release-1.4
Commit: 397f0d150255ea24fd5fd4cc38848441bb0fff58
Parents: 13962e1
Author: twalthr <twalthr@apache.org>
Authored: Thu Nov 16 11:00:30 2017 +0100
Committer: twalthr <twalthr@apache.org>
Committed: Thu Nov 16 11:06:14 2017 +0100

----------------------------------------------------------------------
 .../flink/table/api/stream/sql/JoinTest.scala   | 45 ++++++++++++++++++++
 1 file changed, 45 insertions(+)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/flink/blob/397f0d15/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/sql/JoinTest.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/sql/JoinTest.scala
b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/sql/JoinTest.scala
index 8c1865c..c14b698 100644
--- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/sql/JoinTest.scala
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/sql/JoinTest.scala
@@ -19,7 +19,9 @@ package org.apache.flink.table.api.stream.sql
 
 import org.apache.calcite.rel.logical.LogicalJoin
 import org.apache.flink.api.scala._
+import org.apache.flink.table.api.Types
 import org.apache.flink.table.api.scala._
+import org.apache.flink.table.expressions.Null
 import org.apache.flink.table.plan.logical.TumblingGroupWindow
 import org.apache.flink.table.runtime.join.WindowJoinUtil
 import org.apache.flink.table.utils.TableTestUtil.{term, _}
@@ -245,6 +247,49 @@ class JoinTest extends TableTestBase {
   }
 
   @Test
+  def testJoinWithNullLiteral(): Unit = {
+    val streamUtil: StreamTableTestUtil = streamTestUtil()
+
+    val t1 = streamUtil.addTable[(Int, Long, String)]("Table1", 'a, 'b, 'c, 'proctime.proctime)
+      .select('a, 'b, 'c, 'proctime, Null(Types.LONG) as 'nullField)
+
+    val t2 = streamUtil.addTable[(Int, Long, String)]("Table2", 'a, 'b, 'c, 'proctime.proctime)
+      .select('a, 'b, 'c, 'proctime, 12L as 'nullField)
+
+    streamUtil.tableEnv.registerTable("T1", t1)
+    streamUtil.tableEnv.registerTable("T2", t2)
+
+    val sqlQuery =
+      """
+        |SELECT t2.a, t2.c, t1.c
+        |FROM T1 AS t1
+        |JOIN T2 AS t2 ON t1.a = t2.a AND t1.nullField = t2.nullField AND
+        |  t1.proctime BETWEEN t2.proctime - INTERVAL '5' SECOND AND
+        |  t2.proctime + INTERVAL '5' SECOND
+        |""".stripMargin
+
+    val expected =
+      unaryNode("DataStreamCalc",
+        binaryNode("DataStreamWindowJoin",
+          unaryNode("DataStreamCalc",
+            streamTableNode(0),
+            term("select", "a", "c", "proctime", "null AS nullField")
+          ),
+          unaryNode("DataStreamCalc",
+            streamTableNode(1),
+            term("select", "a", "c", "proctime", "12 AS nullField")
+          ),
+          term("where", "AND(=(a, a0), =(nullField, nullField0), >=(proctime, " +
+            "-(proctime0, 5000)), <=(proctime, DATETIME_PLUS(proctime0, 5000)))"),
+          term("join", "a", "c", "proctime", "nullField", "a0", "c0", "proctime0", "nullField0"),
+          term("joinType", "InnerJoin")
+        ),
+        term("select", "a0 AS a", "c0 AS c", "c AS c0")
+      )
+    streamUtil.verifySql(sqlQuery, expected)
+  }
+
+  @Test
   def testRowTimeInnerJoinAndWindowAggregationOnFirst(): Unit = {
 
     val sqlQuery =


Mime
View raw message