[
https://issues.apache.org/jira/browse/FLINK-8428?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16377007#comment-16377007
]
ASF GitHub Bot commented on FLINK-8428:
---------------------------------------
Github user walterddr commented on a diff in the pull request:
https://github.com/apache/flink/pull/5327#discussion_r170610792
--- Diff:
flink-libraries/flink-table/src/test/scala/org/apache/flink/table/runtime/stream/sql/JoinITCase.scala
---
@@ -977,6 +922,254 @@ class JoinITCase extends StreamingWithStateTestBase {
expected.add("D,R-8,null")
StreamITCase.compareWithList(expected)
}
+
+ /** test non-window inner join **/
+ @Test
+ def testNonWindowInnerJoin(): Unit = {
+ val env = StreamExecutionEnvironment.getExecutionEnvironment
+ val tEnv = TableEnvironment.getTableEnvironment(env)
+ env.setStateBackend(getStateBackend)
+ StreamITCase.clear
+
+ val data1 = new mutable.MutableList[(Int, Long, String)]
+ data1.+=((1, 1L, "Hi1"))
+ data1.+=((1, 2L, "Hi2"))
+ data1.+=((1, 2L, "Hi2"))
+ data1.+=((1, 5L, "Hi3"))
+ data1.+=((2, 7L, "Hi5"))
+ data1.+=((1, 9L, "Hi6"))
+ data1.+=((1, 8L, "Hi8"))
+ data1.+=((3, 8L, "Hi9"))
+
+ val data2 = new mutable.MutableList[(Int, Long, String)]
+ data2.+=((1, 1L, "HiHi"))
+ data2.+=((2, 2L, "HeHe"))
+ data2.+=((3, 2L, "HeHe"))
+
+ val t1 = env.fromCollection(data1).toTable(tEnv, 'a, 'b, 'c)
+ .select(('a === 3) ? (Null(Types.INT), 'a) as 'a, 'b, 'c)
+ val t2 = env.fromCollection(data2).toTable(tEnv, 'a, 'b, 'c)
+ .select(('a === 3) ? (Null(Types.INT), 'a) as 'a, 'b, 'c)
+
+ tEnv.registerTable("T1", t1)
+ tEnv.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.b > t2.b
+ |""".stripMargin
+
+ val result = tEnv.sqlQuery(sqlQuery).toAppendStream[Row]
+ result.addSink(new StreamITCase.StringSink[Row])
+ env.execute()
+
+ val expected = mutable.MutableList(
+ "1,HiHi,Hi2",
+ "1,HiHi,Hi2",
+ "1,HiHi,Hi3",
+ "1,HiHi,Hi6",
+ "1,HiHi,Hi8",
+ "2,HeHe,Hi5",
+ "null,HeHe,Hi9")
+
+ assertEquals(expected.sorted, StreamITCase.testResults.sorted)
+ }
+
+ @Test
+ def testJoin(): Unit = {
--- End diff --
Can this be more specific? like, inner equality join
> Implement stream-stream non-window left outer join
> --------------------------------------------------
>
> Key: FLINK-8428
> URL: https://issues.apache.org/jira/browse/FLINK-8428
> Project: Flink
> Issue Type: Sub-task
> Components: Table API & SQL
> Reporter: Hequn Cheng
> Assignee: Hequn Cheng
> Priority: Major
>
> Implement stream-stream non-window left outer join for sql/table-api. A
> simple design doc can be found
> [here|https://docs.google.com/document/d/1u7bJHeEBP_hFhi8Jm4oT3FqQDOm2pJDqCtq1U1WMHDo/edit?usp=sharing]
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
