Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/5140#discussion_r159013964
--- Diff:
flink-libraries/flink-table/src/test/scala/org/apache/flink/table/runtime/stream/sql/JoinITCase.scala
---
@@ -461,6 +462,302 @@ class JoinITCase extends StreamingWithStateTestBase {
StreamITCase.compareWithList(expected)
}
+ // Tests for left outer join
+ @Test
+ def testProcTimeLeftOuterJoin(): Unit = {
+ val env = StreamExecutionEnvironment.getExecutionEnvironment
+ val tEnv = TableEnvironment.getTableEnvironment(env)
+ env.setStateBackend(getStateBackend)
+ StreamITCase.clear
+ env.setParallelism(1)
+
+ val sqlQuery =
+ """
+ |SELECT t2.a, t2.c, t1.c
+ |FROM T1 as t1 LEFT OUTER JOIN T2 as t2 ON
+ | t1.a = t2.a AND
+ | t1.proctime BETWEEN t2.proctime - INTERVAL '5' SECOND AND
+ | t2.proctime + INTERVAL '3' SECOND
+ |""".stripMargin
+
+ val data1 = new mutable.MutableList[(Int, Long, String)]
+ data1.+=((1, 1L, "Hi1"))
+ data1.+=((1, 2L, "Hi2"))
+ data1.+=((1, 5L, "Hi3"))
+ data1.+=((2, 7L, "Hi5"))
+
+ val data2 = new mutable.MutableList[(Int, Long, String)]
+ data2.+=((1, 1L, "HiHi"))
+ data2.+=((2, 2L, "HeHe"))
+
+ val t1 = env.fromCollection(data1).toTable(tEnv, 'a, 'b, 'c,
'proctime.proctime)
+ .select('a, 'b, 'c, 'proctime)
+ val t2 = env.fromCollection(data2).toTable(tEnv, 'a, 'b, 'c,
'proctime.proctime)
+ .select('a, 'b, 'c, 'proctime)
+
+ tEnv.registerTable("T1", t1)
+ tEnv.registerTable("T2", t2)
+
+ val result = tEnv.sqlQuery(sqlQuery).toAppendStream[Row]
+ result.addSink(new StreamITCase.StringSink[Row])
+ env.execute()
+ }
+
+ @Test
+ def testRowTimeLeftOuterJoin(): Unit = {
+ val env = StreamExecutionEnvironment.getExecutionEnvironment
+ val tEnv = TableEnvironment.getTableEnvironment(env)
+ env.setStateBackend(getStateBackend)
+ env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
+ StreamITCase.clear
+
+ val sqlQuery =
+ """
+ |SELECT t2.key, t2.id, t1.id
+ |FROM T1 AS t1 LEFT OUTER JOIN T2 AS t2 ON
+ | t1.key = t2.key AND
+ | t1.rt BETWEEN t2.rt - INTERVAL '5' SECOND AND
+ | t2.rt + INTERVAL '6' SECOND
+ |""".stripMargin
+
+ val data1 = new mutable.MutableList[(String, String, Long)]
+ // for boundary test
+ data1.+=(("A", "L-1", 1000L))
+ data1.+=(("A", "L-2", 2000L))
+ data1.+=(("B", "L-4", 4000L))
+ data1.+=(("A", "L-6", 6000L))
+ data1.+=(("C", "L-7", 7000L))
+ data1.+=(("A", "L-10", 10000L))
+ data1.+=(("A", "L-12", 12000L))
+ data1.+=(("A", "L-20", 20000L))
+
+ val data2 = new mutable.MutableList[(String, String, Long)]
--- End diff --
Add a row to the right data set such that one left row joins with two right
rows.
---
