Repository: apache/flink
File: flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/CorrelateITCase.scala (Scala)

Code before:
    result.executeInsert("MySink").await()

    val expected = List("1,a")
    assertThat(sink.getAppendResults.sorted).isEqualTo(expected.sorted)
  }

  @Test
  def testLeftJoinPartialProjectWithEmptyOutput(): Unit = {
    val data = List((1, 2, ""), (1, 3, ""))

    val t1 = env.fromCollection(data).toTable(tEnv, 'a, 'b, 'c)
    tEnv.createTemporaryView("T1", t1)

    val sql = "SELECT b, v FROM T1 left join lateral table(STRING_SPLIT(c, '|')) as T(v) on true"

    val result = tEnv.sqlQuery(sql)
    val sink = TestSinkUtil.configureSink(result, new TestingAppendTableSink)
    tEnv.asInstanceOf[TableEnvironmentInternal].registerTableSinkInternal("MySink", sink)
    result.executeInsert("MySink").await()

    val expected = List("2,null", "3,null")
    assertThat(sink.getAppendResults.sorted).isEqualTo(expected.sorted)
  }

  @Test
  def testL1eftJoinPartialProjectWithEmptyOutput(): Unit = {
    val data = List((1, 2, "x|y"))

    val t1 = env.fromCollection(data).toTable(tEnv, 'a, 'b, 'c)
    tEnv.createTemporaryView("T1", t1)

    val sql =
      "SELECT * FROM T1 as t, LATERAL TABLE(STRING_SPLIT(t.c,'|')) CROSS JOIN (VALUES ('A'), ('B'));"

    val result = tEnv.sqlQuery(sql)
    val sink = TestSinkUtil.configureSink(result, new TestingAppendTableSink)
    tEnv.asInstanceOf[TableEnvironmentInternal].registerTableSinkInternal("MySink", sink)
    result.executeInsert("MySink").await()

    val expected = List("1,2,x|y,x,A", "1,2,x|y,x,B", "1,2,x|y,y,A", "1,2,x|y,y,B")
    assertThat(sink.getAppendResults.sorted).isEqualTo(expected.sorted)
  }

  // TODO support agg
//  @Test
//  def testCountStarOnCorrelate(): Unit = {
//    val data = List(
//      (1, 2, "3018-06-10|2018-06-03"),
//      (1, 2, "2018-06-01"),
//      (1, 2, "2018-06-02"))
//

Review comment (suggestion):
```suggestion
  def testLeftJoinPartialProjectWithEmptyOutput(): Unit = {
```

Code after:
    result.executeInsert("MySink").await()

    val expected = List("1,a")
    assertThat(sink.getAppendResults.sorted).isEqualTo(expected.sorted)
  }

  @Test
  def testLeftJoinPartialProjectWithEmptyOutput(): Unit = {
    val data = List((1, 2, ""), (1, 3, ""))

    val t1 = env.fromCollection(data).toTable(tEnv, 'a, 'b, 'c)
    tEnv.createTemporaryView("T1", t1)

    val sql = "SELECT b, v FROM T1 left join lateral table(STRING_SPLIT(c, '|')) as T(v) on true"

    val result = tEnv.sqlQuery(sql)
    val sink = TestSinkUtil.configureSink(result, new TestingAppendTableSink)
    tEnv.asInstanceOf[TableEnvironmentInternal].registerTableSinkInternal("MySink", sink)
    result.executeInsert("MySink").await()

    val expected = List("2,null", "3,null")
    assertThat(sink.getAppendResults.sorted).isEqualTo(expected.sorted)
  }

  @Test
  def testLateralCrossJoin(): Unit = {
    val data = List((1, 2, "x|y"))

    val t1 = env.fromCollection(data).toTable(tEnv, 'a, 'b, 'c)
    tEnv.createTemporaryView("T1", t1)

    val sql =
      "SELECT * FROM T1 as t, LATERAL TABLE(STRING_SPLIT(t.c,'|')) CROSS JOIN (VALUES ('A'), ('B'));"

    val result = tEnv.sqlQuery(sql)
    val sink = TestSinkUtil.configureSink(result, new TestingAppendTableSink)
    tEnv.asInstanceOf[TableEnvironmentInternal].registerTableSinkInternal("MySink", sink)
    result.executeInsert("MySink").await()

    val expected = List("1,2,x|y,x,A", "1,2,x|y,x,B", "1,2,x|y,y,A", "1,2,x|y,y,B")
    assertThat(sink.getAppendResults.sorted).isEqualTo(expected.sorted)
  }

  // TODO support agg
//  @Test
//  def testCountStarOnCorrelate(): Unit = {
//    val data = List(
//      (1, 2, "3018-06-10|2018-06-03"),
//      (1, 2, "2018-06-01"),
//      (1, 2, "2018-06-02"))
//