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:
                                            "-D[one, 2, bb]",
                                            "+I[two, 0, dd]",
                                            "-U[two, 0, dd]",
                                            "+U[two, 1, dd]")
                                    .build())
                    .runSql("INSERT INTO sink_t SELECT a, b, c FROM source_t WHERE b < 3")
                    .build();

    static final TableTestProgram KAFKA_SOURCE_WITH_KEY_FILTER =
            TableTestProgram.of(
                            "kafka-with-key-filter",
                            "validates kafka source (with partial deletes) with key filter")
                    .setupTableSource(
                            SourceTestStep.newBuilder("source_t")
                                    .inMode(KAFKA_WITH_PARTIAL_DELETES_CHANGELOG)
                                    .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")
                                    .inMode(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 KAFKA_SOURCE_WITH_NO_FILTER =
            TableTestProgram.of(
                            "kafka-with-no-filter",
                            "validates kafka source (with partial deletes) with no filter")
                    .setupTableSource(
                            SourceTestStep.newBuilder("source_t")
                                    .inMode(KAFKA_WITH_PARTIAL_DELETES_CHANGELOG)

Review comment (suggestion):
ensure that we only enter partial deletes i.e. set all columns `null` that are not the key

Code after:
                                            "-D[one, 2, bb]",
                                            "+I[two, 0, dd]",
                                            "-U[two, 0, dd]",
                                            "+U[two, 1, dd]")
                                    .build())
                    .runSql("INSERT INTO sink_t SELECT a, b, c FROM source_t WHERE b < 3")
                    .build();

    static final TableTestProgram KAFKA_SOURCE_WITH_KEY_FILTER =
            TableTestProgram.of(
                            "kafka-with-key-filter",
                            "validates kafka source (with partial deletes) with key filter")
                    .setupTableSource(
                            SourceTestStep.newBuilder("source_t")
                                    .addMode(KAFKA_WITH_PARTIAL_DELETES_CHANGELOG)
                                    .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", null, null),
                                            Row.ofKind(RowKind.DELETE, "two", null, null))
                                    .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 KAFKA_SOURCE_WITH_NO_FILTER =
            TableTestProgram.of(
                            "kafka-with-no-filter",
                            "validates kafka source (with partial deletes) with no filter")
                    .setupTableSource(
                            SourceTestStep.newBuilder("source_t")
                                    .addMode(KAFKA_WITH_PARTIAL_DELETES_CHANGELOG)