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:
        "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 testExplainPlanAdviceWithLookupOptions(): Unit = {
    val lookupOptions = new JHashMap[String, String]()
    lookupOptions.put("lookup.cache", "partial")
    lookupOptions.put("lookup.partial-cache.max-rows", "1000")
    addTable(
      "src1_v2",
      Schema
        .newBuilder()
        .column("a0", DataTypes.INT.notNull)
        .column("a1", DataTypes.DOUBLE.notNull)
        .column("a2", DataTypes.STRING)
        .column("a3", DataTypes.INT)
        .index("a1", "a2")
        .index("a3")
        .build(),
      lookupOptions
    )

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

  @Test
  def testWithWatermarkAssigner(): Unit = {
    addTable(
      "wm_source",
      Schema

Review comment (refactor):
Nit if you pull the latest master, you can simplify this part by 
```
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)
```

Code after:
  @Test
  def testNonDeterministicFilterFieldsBeforeJoin2(): Unit = {
    util.verifyRelPlanInsert("""
                               |insert into snk
                               |  select * from src1
                               |  join (
                               |    select * from src2 where b0 > rand(10)
                               |  )
                               |  on a1 = b1
                               |  and a2 = b2
                               |""".stripMargin)
  }

  @Test
  def testNonDeterministicFilterFieldsBeforeJoinWithFilterPushDown(): Unit = {
    replaceTable("src1", "src1", Maps.newHashMap("filterable-fields", "a0"))

    // actually, 'values' source will not push down filter 'a0 > rand(10)' into source
    util.verifyRelPlanInsert("""
                               |insert into snk(l0, l1, r0, r2, r1)
                               |  select a0, a1, b0, b2, b1 from (
                               |    select a0, a2, a1 from src1 where a0 > rand(10)
                               |  ) join src2
                               |  on a1 = b1
                               |  and a2 = b2
                               |""".stripMargin)
  }

  @Test
  def testPartitionPushDown(): Unit = {
    addTable(
      "src1WithPartition",
      Schema
        .newBuilder()
        .column("a0", DataTypes.INT.notNull)
        .column("a1", DataTypes.DOUBLE.notNull)
        .column("a2", DataTypes.STRING)
        .column("pt", DataTypes.INT)
        .index("a1", "a2")
        .build(),
      Maps.newHashMap("partition-list", "pt:1;pt:2"),
      Lists.newArrayList("pt")
    )

    util.verifyRelPlanInsert("""
                               |insert into snk(l0, r0, r2)
                               |  select a0, b0, b2 from (
                               |    select a0, a2, a1 from src1WithPartition where pt = 1
                               |  ) join src2
                               |  on a1 = b1
                               |  and a2 = b2