[FLINK-22099] Fix bug for semi/anti window join.

• Type: Sub-task
• Status: Closed
• Priority: Minor
• Resolution: Fixed
• Fix Version/s: 1.13.0, 1.14.0
• Component/s: Table SQL / Planner
• Labels: pull-request-available
• Assignee: Jing Zhang
• Reporter: Jing Zhang
• Created: 02/Apr/21 05:04
• Updated: 28/May/21 09:10
• Resolved: 19/Apr/21 06:48

Description

Fix bug for Semi/Anti WindowJoin.

//代码占位符
@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)
}

Now run the above sql, an `ArrayIndexOutOfBoundsException` would be thrown out.

Issue Links

• relates to: FLINK-19606 Implement streaming window join operator
• links to: GitHub Pull Request #15477