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:
  def testRetractSink(): Unit = {
    util.tableEnv
      .executeSql(
        "CREATE TABLE retract_snk WITH (\n"
          + "  'sink-changelog-mode-enforced' = 'I,UA,UB,D'"
          + ") LIKE snk (\n"
          + "  OVERWRITING OPTIONS\n"
          + ")")

    util.verifyRelPlanInsert(
      "insert into retract_snk select * from src1 join src2 " +
        "on src1.a1 = src2.b1 " +
        "and src1.a2 = src2.b2")
  }

  @Test
  def testAppendSink(): Unit = {
    util.tableEnv
      .executeSql(
        "CREATE TABLE append_snk WITH (\n"
          + "  'sink-insert-only' = 'true',\n"
          + "  'sink-changelog-mode-enforced' = 'I'\n"
          + ") LIKE snk (\n"
          + "  OVERWRITING OPTIONS\n"
          + ")")

    util.verifyRelPlanInsert(
      "insert into append_snk select * from src1 join src2 " +
        "on src1.a1 = src2.b1 " +
        "and src1.a2 = src2.b2")
  }

  private def addTable(
      tableName: String,
      schema: Schema,
      extraOptions: JMap[String, String] = Collections.emptyMap()): Unit = {
    val currentCatalog = tEnv.getCurrentCatalog
    val currentDatabase = tEnv.getCurrentDatabase
    val tablePath = new ObjectPath(currentDatabase, tableName)
    val catalog = tEnv.getCatalog(currentCatalog).get()
    val schemaResolver = tEnv.getCatalogManager.getSchemaResolver

    val options = new JHashMap[String, String](testValuesTableOptions)
    options.putAll(extraOptions)

    val original = CatalogTable
      .newBuilder()
      .schema(schema)
      .comment(testComment)
      .partitionKeys(Collections.emptyList())
      .options(options)

Review comment (performance):
Can we set TABLE_OPTIMIZER_DELTA_JOIN_STRATEGY to FORCE and provide some different exception messages to indicate the reason for not supporting it

Code after:
        .newBuilder()
        .column("l0", DataTypes.INT.notNull)
        .column("l1", DataTypes.DOUBLE)
        .primaryKey("l0")
        .build()
    )

    // no joins on this query
    util.verifyRelPlanInsert("insert into tmp_snk select a0, a1 from src1")
  }

  @Test
  def testRetractSink(): Unit = {
    util.tableEnv
      .executeSql(
        "CREATE TABLE retract_snk WITH (\n"
          + "  'sink-changelog-mode-enforced' = 'I,UA,UB,D'"
          + ") LIKE snk (\n"
          + "  OVERWRITING OPTIONS\n"
          + ")")

    util.verifyRelPlanInsert(
      "insert into retract_snk select * from src1 join src2 " +
        "on src1.a1 = src2.b1 " +
        "and src1.a2 = src2.b2")
  }

  @Test
  def testAppendSink(): Unit = {
    util.tableEnv
      .executeSql(
        "CREATE TABLE append_snk WITH (\n"
          + "  'sink-insert-only' = 'true',\n"
          + "  'sink-changelog-mode-enforced' = 'I'\n"
          + ") LIKE snk (\n"
          + "  OVERWRITING OPTIONS\n"
          + ")")

    util.verifyRelPlanInsert(
      "insert into append_snk select * from src1 join src2 " +
        "on src1.a1 = src2.b1 " +
        "and src1.a2 = src2.b2")
  }

  private def addTable(
      tableName: String,
      schema: Schema,
      extraOptions: JMap[String, String] = Collections.emptyMap()): Unit = {
    val currentCatalog = tEnv.getCurrentCatalog
    val currentDatabase = tEnv.getCurrentDatabase
    val tablePath = new ObjectPath(currentDatabase, tableName)