Repository: apache/flink
File: flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/physical/stream/DuplicateChangesInferRuleTest.java (Java)

Code before:

        String sql = "insert into pk_upsert_snk select a,b,c from retract_src";
        verifyRelPlanInsert(sql);
    }

    @TestTemplate
    void testSinkWithMaterialize() {
        assumeTrue(testSinkWithPk);

        util.tableEnv()
                .executeSql(
                        "CREATE TABLE another_pk_upsert_snk (\n"
                                + "  primary key (b) not enforced\n"
                                + ") LIKE pk_upsert_snk (\n"
                                + "  EXCLUDING CONSTRAINTS\n"
                                + ")");

        String sql = "insert into another_pk_upsert_snk select a,b,c from retract_src";
        verifyRelPlanInsert(sql);
    }

    @TestTemplate
    void testRetractSink() {
        assumeTrue(testSinkWithPk);

        String sql = "insert into pk_retract_snk select a,b,c from retract_src";
        verifyRelPlanInsert(sql);
    }

    @TestTemplate
    void testChangelogNormalize() {
        assumeTrue(testSinkWithPk);

        String sql = "insert into pk_retract_snk select a,b,c from upsert_src";
        verifyRelPlanInsert(sql);
    }

    @TestTemplate
    void testJoin() {
        String sql =
                String.format(
                        "insert into %s select T1.a, T2.b, T1.c "
                                + "from append_src1 T1 join append_src2 T2 "
                                + "on T1.a = T2.a",
                        getSinkTableName());
        verifyRelPlanInsert(sql);
    }

    @TestTemplate
    void testWindowJoin() {
        assumeTrue(testSinkWithPk);

Review comment (refactor):
Maybe there is essentially no difference between testRetractSink and testChangelogNormalize, both of which are caused by pk_detract_stnk resulting in duplicateChanges being set to disallow. Therefore, they can be merged into one test and the ddl of pk_detract_stnk can be moved here

Code after:

        String sql = "insert into pk_upsert_snk select a, b, c from retract_src";
        verifyRelPlanInsert(sql);
    }

    @TestTemplate
    void testSinkWithMaterialize() {
        assumeTrue(testSinkWithPk);

        util.tableEnv()
                .executeSql(
                        "CREATE TABLE another_pk_upsert_snk (\n"
                                + "  primary key (b) not enforced\n"
                                + ") LIKE pk_upsert_snk (\n"
                                + "  EXCLUDING CONSTRAINTS\n"
                                + ")");

        String sql = "insert into another_pk_upsert_snk select a, b, c from retract_src";
        verifyRelPlanInsert(sql);
    }

    @TestTemplate
    void testRetractSink() {
        assumeTrue(testSinkWithPk);

        String sql = "insert into pk_retract_snk select a, b, c from retract_src";
        verifyRelPlanInsert(sql);
    }

    @TestTemplate
    void testChangelogNormalize() {
        assumeTrue(testSinkWithPk);

        String sql = "insert into pk_retract_snk select a, b, c from upsert_src";
        verifyRelPlanInsert(sql);
    }

    @TestTemplate
    void testJoin() {
        String sql =
                String.format(
                        "insert into %s select T1.a, T2.b, T1.c "
                                + "from append_src1 T1 join append_src2 T2 "
                                + "on T1.a = T2.a",
                        getSinkTableName());
        verifyRelPlanInsert(sql);
    }

    @TestTemplate
    void testWindowJoin() {
        assumeTrue(testSinkWithPk);