Repository: apache/flink
File: flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ChangelogNormalizeSemanticTestPrograms.java (Java)

Code before:
                                    .producedValues(
                                            Row.ofKind(RowKind.INSERT, "one", 1, "a"),
                                            Row.ofKind(RowKind.UPDATE_AFTER, "one", 2, "bb"),
                                            Row.ofKind(RowKind.UPDATE_AFTER, "three", 3, "ccc"),
                                            Row.ofKind(RowKind.UPDATE_AFTER, "three", 3, "ccc"),
                                            Row.ofKind(RowKind.INSERT, "one", 4, "aaaa"),
                                            Row.ofKind(RowKind.UPDATE_AFTER, "one", 4, "aaaa"),
                                            Row.ofKind(RowKind.INSERT, "two", 1, "d"),
                                            Row.ofKind(RowKind.UPDATE_AFTER, "two", 1, "d"),
                                            Row.ofKind(RowKind.DELETE, "three", 3, "cc"),
                                            Row.ofKind(RowKind.DELETE, "two", 1, "d"))
                                    .build())
                    .setupTableSink(
                            SinkTestStep.newBuilder("sink_t")
                                    .addSchema(SINK_SCHEMA)
                                    .consumedValues(
                                            "+I[one, 1, a]",
                                            "+U[one, 2, bb]",
                                            "+U[one, 4, aaaa]",
                                            "+I[two, 1, d]",
                                            "-D[two, 1, d]")
                                    .build())
                    .runSql("INSERT INTO sink_t SELECT a, b, c FROM source_t WHERE a <> 'three'")
                    .build();

    static final TableTestProgram UPSERT_SOURCE =
            TableTestProgram.of("upsert", "validates changelog normalize")
                    .setupConfig(
                            ExecutionConfigOptions.TABLE_EXEC_SOURCE_CDC_EVENTS_DUPLICATE, true)
                    .setupTableSource(
                            SourceTestStep.newBuilder("source_t")
                                    .addOption("changelog-mode", "UA,D")
                                    .addSchema(SOURCE_SCHEMA)
                                    .producedValues(
                                            Row.ofKind(RowKind.INSERT, "one", 1, "a"),
                                            Row.ofKind(RowKind.UPDATE_AFTER, "one", 2, "bb"),
                                            Row.ofKind(RowKind.UPDATE_AFTER, "three", 3, "ccc"),
                                            Row.ofKind(RowKind.UPDATE_AFTER, "three", 3, "ccc"),
                                            Row.ofKind(RowKind.INSERT, "one", 4, "aaaa"),
                                            Row.ofKind(RowKind.UPDATE_AFTER, "one", 4, "aaaa"),
                                            Row.ofKind(RowKind.DELETE, "three", 3, "cc"))
                                    .build())
                    .setupTableSink(
                            SinkTestStep.newBuilder("sink_t")
                                    .addSchema(SINK_SCHEMA)
                                    .consumedValues(
                                            "+I[one, 1, a]",
                                            "+U[one, 2, bb]",
                                            "+I[three, 3, ccc]",
                                            "+U[one, 4, aaaa]",
                                            "-D[three, 3, ccc]")

Review comment (suggestion):
```suggestion
    static final TableTestProgram UPSERT_SOURCE_WITH_RETRACT_SINK =
```

Code after:
                                    .build())
                    .runSql("INSERT INTO sink_t SELECT a, b, c FROM source_t WHERE b < 3")
                    .build();

    static final TableTestProgram UPSERT_SOURCE_WITH_KEY_FILTER =
            TableTestProgram.of("upsert-with-key-filter", "validates upsert with key filter")
                    .setupTableSource(
                            SourceTestStep.newBuilder("source_t")
                                    .addMode(ChangelogMode.upsert())
                                    .addSchema(SOURCE_SCHEMA)
                                    .producedValues(
                                            Row.ofKind(RowKind.INSERT, "one", 1, "a"),
                                            Row.ofKind(RowKind.UPDATE_AFTER, "one", 2, "bb"),
                                            Row.ofKind(RowKind.UPDATE_AFTER, "three", 3, "ccc"),
                                            Row.ofKind(RowKind.UPDATE_AFTER, "three", 3, "ccc"),
                                            Row.ofKind(RowKind.INSERT, "one", 4, "aaaa"),
                                            Row.ofKind(RowKind.UPDATE_AFTER, "one", 4, "aaaa"),
                                            Row.ofKind(RowKind.INSERT, "two", 1, "d"),
                                            Row.ofKind(RowKind.UPDATE_AFTER, "two", 1, "d"),
                                            Row.ofKind(RowKind.DELETE, "three", 3, "ccc"),
                                            Row.ofKind(RowKind.DELETE, "two", 1, "d"))
                                    .build())
                    .setupTableSink(
                            SinkTestStep.newBuilder("sink_t")
                                    .addMode(ChangelogMode.all())
                                    .addSchema(SINK_SCHEMA)
                                    .consumedValues(
                                            "+I[one, 1, a]",
                                            "-U[one, 1, a]",
                                            "+U[one, 2, bb]",
                                            "-U[one, 2, bb]",
                                            "+U[one, 4, aaaa]",
                                            "+I[two, 1, d]",
                                            "-D[two, 1, d]")
                                    .build())
                    .runSql("INSERT INTO sink_t SELECT a, b, c FROM source_t WHERE a <> 'three'")
                    .build();

    static final TableTestProgram UPSERT_SOURCE_WITH_NO_FILTER =
            TableTestProgram.of("upsert-with-no-filter", "validates upsert with no filter")
                    .setupTableSource(
                            SourceTestStep.newBuilder("source_t")
                                    .addMode(ChangelogMode.upsert())
                                    .addSchema(SOURCE_SCHEMA)
                                    .producedValues(
                                            Row.ofKind(RowKind.INSERT, "one", 1, "a"),
                                            Row.ofKind(RowKind.UPDATE_AFTER, "one", 2, "bb"),
                                            Row.ofKind(RowKind.INSERT, "three", 3, "ccc"),
                                            Row.ofKind(RowKind.UPDATE_AFTER, "three", 3, "ccc"),
                                            Row.ofKind(RowKind.INSERT, "one", 4, "aaaa"),
                                            Row.ofKind(RowKind.UPDATE_AFTER, "one", 4, "aaaa"),