[
https://issues.apache.org/jira/browse/FLINK-22099?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
godfrey he updated FLINK-22099:
-------------------------------
Fix Version/s: 1.14.0
> Fix bug for semi/anti window join.
> ----------------------------------
>
> Key: FLINK-22099
> URL: https://issues.apache.org/jira/browse/FLINK-22099
> Project: Flink
> Issue Type: Sub-task
> Components: Table SQL / Planner
> Reporter: Andy
> Assignee: Andy
> Priority: Minor
> Labels: pull-request-available
> Fix For: 1.13.0, 1.14.0
>
>
> Fix bug for Semi/Anti WindowJoin.
> {code:java}
> //代码占位符
> @Test
> def testSemiJoinIN(): Unit = {
> val sql =
> """
> |SELECT * FROM (
> | SELECT
> | a,
> | window_start,
> | window_end,
> | window_time,
> | count(*) as cnt,
> | count(distinct c) AS uv
> | FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(rowtime), INTERVAL '15'
> MINUTE))
> | GROUP BY a, window_start, window_end, window_time
> |) L WHERE L.a IN (
> |SELECT a FROM (
> | SELECT
> | a,
> | window_start,
> | window_end,
> | window_time,
> | count(*) as cnt,
> | count(distinct c) AS uv
> | FROM TABLE(TUMBLE(TABLE MyTable2, DESCRIPTOR(rowtime), INTERVAL '15'
> MINUTE))
> | GROUP BY a, window_start, window_end, window_time
> |) R
> |WHERE L.window_start = R.window_start AND L.window_end = R.window_end)
> """.stripMargin
> util.verifyRelPlan(sql)
> }
> @Test
> def testSemiExist(): Unit = {
> val sql =
> """
> |SELECT * FROM (
> | SELECT
> | a,
> | window_start,
> | window_end,
> | window_time,
> | count(*) as cnt,
> | count(distinct c) AS uv
> | FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(rowtime), INTERVAL '15'
> MINUTE))
> | GROUP BY a, window_start, window_end, window_time
> |) L WHERE EXISTS (
> |SELECT * FROM (
> | SELECT
> | a,
> | window_start,
> | window_end,
> | window_time,
> | count(*) as cnt,
> | count(distinct c) AS uv
> | FROM TABLE(TUMBLE(TABLE MyTable2, DESCRIPTOR(rowtime), INTERVAL '15'
> MINUTE))
> | GROUP BY a, window_start, window_end, window_time
> |) R
> |WHERE L.window_start = R.window_start AND L.window_end = R.window_end
> AND L.a = R.a)
> """.stripMargin
> util.verifyRelPlan(sql)
> }
> @Test
> def testAntiJoinNotExist(): Unit = {
> val sql =
> """
> |SELECT * FROM (
> | SELECT
> | a,
> | window_start,
> | window_end,
> | window_time,
> | count(*) as cnt,
> | count(distinct c) AS uv
> | FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(rowtime), INTERVAL '15'
> MINUTE))
> | GROUP BY a, window_start, window_end, window_time
> |) L WHERE NOT EXISTS (
> |SELECT * FROM (
> | SELECT
> | a,
> | window_start,
> | window_end,
> | window_time,
> | count(*) as cnt,
> | count(distinct c) AS uv
> | FROM TABLE(TUMBLE(TABLE MyTable2, DESCRIPTOR(rowtime), INTERVAL '15'
> MINUTE))
> | GROUP BY a, window_start, window_end, window_time
> |) R
> |WHERE L.window_start = R.window_start AND L.window_end = R.window_end
> AND L.a = R.a)
> """.stripMargin
> util.verifyRelPlan(sql)
> }
> @Test
> def testAntiJoinNotIN(): Unit = {
> val sql =
> """
> |SELECT * FROM (
> | SELECT
> | a,
> | window_start,
> | window_end,
> | window_time,
> | count(*) as cnt,
> | count(distinct c) AS uv
> | FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(rowtime), INTERVAL '15'
> MINUTE))
> | GROUP BY a, window_start, window_end, window_time
> |) L WHERE L.a NOT IN (
> |SELECT a FROM (
> | SELECT
> | a,
> | window_start,
> | window_end,
> | window_time,
> | count(*) as cnt,
> | count(distinct c) AS uv
> | FROM TABLE(TUMBLE(TABLE MyTable2, DESCRIPTOR(rowtime), INTERVAL '15'
> MINUTE))
> | GROUP BY a, window_start, window_end, window_time
> |) R
> |WHERE L.window_start = R.window_start AND L.window_end = R.window_end)
> """.stripMargin
> util.verifyRelPlan(sql)
> }{code}
> Now run the above sql, an `ArrayIndexOutOfBoundsException` would be thrown
> out.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
