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

Code before:

    val stmt = tEnv.createStatementSet()
    stmt.addInsertSql(
      "insert into snk select * from src1 join src2 " +
        "on src1.a1 = src2.b1 " +
        "and src1.a2 = src2.b2")

    stmt.addInsertSql(
      "insert into snk2 select * from src1 join src2 " +
        "on src1.a1 = src2.b1 " +
        "and src1.a2 = src2.b2")

    util.verifyExecPlan(stmt)
  }

  @Test
  def testExplainPlanAdvice(): Unit = {
    util.verifyExplainInsert(
      "insert into snk select * from src1 join src2 " +
        "on src1.a1 = src2.b1 " +
        "and src1.a2 = src2.b2",
      ExplainDetail.PLAN_ADVICE)
  }

  @Test
  def testWithWatermarkAssigner(): Unit = {
    addTable(
      "wm_source",
      Schema
        .newBuilder()
        .column("a0", DataTypes.INT)
        .column("a1", DataTypes.STRING)
        .column("a2", DataTypes.TIMESTAMP(3))
        .watermark("a2", "a2 - INTERVAL '1' SECOND")
        .index("a1")
        .build()
    )
    addTable(
      "tmp_sink",
      Schema
        .newBuilder()
        .column("l0", DataTypes.INT)
        .column("l1", DataTypes.STRING)
        .column("l2", DataTypes.TIMESTAMP(3))
        .column("r0", DataTypes.INT.notNull)
        .column("r2", DataTypes.STRING)
        .column("r1", DataTypes.DOUBLE)
        .build()
    )

    // could not optimize into delta join because there is a calc between join and source

Review comment (suggestion):
Is there an end to end test we can add to drive this code outside of explain? If this issue is explain only ,then we should mention the situations that the user can hit this in the Jira or PR text - so people can find it easier. 

Code after:

    val stmt = tEnv.createStatementSet()
    stmt.addInsertSql(
      "insert into snk select * from src1 join src2 " +
        "on src1.a1 = src2.b1 " +
        "and src1.a2 = src2.b2")

    stmt.addInsertSql(
      "insert into snk2 select * from src1 join src2 " +
        "on src1.a1 = src2.b1 " +
        "and src1.a2 = src2.b2")

    util.verifyExecPlan(stmt)
  }

  @Test
  def testExplainPlanAdvice(): Unit = {
    util.verifyExplainInsert(
      "insert into snk select * from src1 join src2 " +
        "on src1.a1 = src2.b1 " +
        "and src1.a2 = src2.b2",
      ExplainDetail.PLAN_ADVICE)
  }

  @Test
  def testLookupTableWithCache(): Unit = {
    val lookupOptions = new JHashMap[String, String]()
    lookupOptions.put("lookup.cache", "partial")
    lookupOptions.put("lookup.partial-cache.max-rows", "1000")

    replaceTable("src1", "src1", lookupOptions)

    util.verifyExplainInsert(
      "insert into snk select * from src1 join src2 " +
        "on src1.a1 = src2.b1 " +
        "and src1.a2 = src2.b2",
      ExplainDetail.PLAN_ADVICE)
  }

  @Test
  def testWithWatermarkAssigner(): Unit = {
    addTable(
      "wm_source",
      Schema
        .newBuilder()
        .column("a0", DataTypes.INT)
        .column("a1", DataTypes.STRING)
        .column("a2", DataTypes.TIMESTAMP(3))
        .watermark("a2", "a2 - INTERVAL '1' SECOND")
        .index("a1")
        .build()