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

Code before:
                                            Row.ofKind(RowKind.INSERT, "2", "retract2"),
                                            Row.ofKind(RowKind.INSERT, "3", "retract3"),
                                            Row.ofKind(RowKind.DELETE, "3", "retract3"),
                                            Row.ofKind(RowKind.INSERT, "1", "retract1_new"))
                                    .build())
                    .setupTableSource(
                            SourceTestStep.newBuilder("UpsertTable")
                                    .addSchema(
                                            "key_id STRING PRIMARY KEY NOT ENFORCED, status STRING")
                                    .addOption("changelog-mode", "I,UA,D")
                                    .producedValues(
                                            Row.ofKind(RowKind.INSERT, "1", "active"),
                                            Row.ofKind(RowKind.INSERT, "2", "pending"),
                                            Row.ofKind(RowKind.UPDATE_BEFORE, "2", "pending"),
                                            Row.ofKind(RowKind.UPDATE_AFTER, "2", "active"),
                                            Row.ofKind(RowKind.INSERT, "3", "inactive"),
                                            Row.ofKind(RowKind.DELETE, "3", "inactive"))
                                    .build())
                    .setupTableSink(
                            SinkTestStep.newBuilder("sink")
                                    .addSchema(
                                            "id STRING",
                                            "val STRING",
                                            "data STRING",
                                            "status STRING")
                                    .addOption("changelog-mode", "I,UA,UB,D")
                                    .consumedValues(
                                            "+I[1, append1, retract1, active]",
                                            "+I[2, append2, retract2, active]",
                                            "+I[1, append1, retract1_new, active]",
                                            "+I[3, append3, null, null]")
                                    .testMaterializedData()
                                    .build())
                    .runSql(
                            "INSERT INTO sink "
                                    + "SELECT a.id, a.val, r.data, u.status "
                                    + "FROM AppendTable a "
                                    + "LEFT JOIN RetractTable r ON a.id = r.ref_id "
                                    + "LEFT JOIN UpsertTable u ON a.id = u.key_id")
                    .build();
}


Review comment (suggestion):
sinks use a different flag, otherwise the mode is not considered: `sink-changelog-mode-enforced`

Code after:
                                            Row.ofKind(RowKind.INSERT, "2", "retract2"),
                                            Row.ofKind(RowKind.INSERT, "3", "retract3"),
                                            Row.ofKind(RowKind.DELETE, "3", "retract3"),
                                            Row.ofKind(RowKind.INSERT, "1", "retract1_new"))
                                    .build())
                    .setupTableSource(
                            SourceTestStep.newBuilder("UpsertTable")
                                    .addSchema(
                                            "key_id STRING PRIMARY KEY NOT ENFORCED, status STRING")
                                    .addOption("changelog-mode", "I,UA,D")
                                    .producedValues(
                                            Row.ofKind(RowKind.INSERT, "1", "active"),
                                            Row.ofKind(RowKind.INSERT, "2", "pending"),
                                            Row.ofKind(RowKind.UPDATE_AFTER, "2", "active"),
                                            Row.ofKind(RowKind.INSERT, "3", "inactive"),
                                            Row.ofKind(RowKind.DELETE, "3", "inactive"))
                                    .build())
                    .setupTableSink(
                            SinkTestStep.newBuilder("sink")
                                    .addSchema(
                                            "id STRING",
                                            "val STRING",
                                            "data STRING",
                                            "status STRING")
                                    .addOption("sink-changelog-mode-enforced", "I,UA,UB,D")
                                    .consumedValues(
                                            "+I[1, append1, retract1, active]",
                                            "+I[2, append2, retract2, active]",
                                            "+I[1, append1, retract1_new, active]",
                                            "+I[3, append3, null, null]")
                                    .testMaterializedData()
                                    .build())
                    .runSql(
                            "INSERT INTO sink "
                                    + "SELECT a.id, a.val, r.data, u.status "
                                    + "FROM AppendTable a "
                                    + "LEFT JOIN RetractTable r ON a.id = r.ref_id "
                                    + "LEFT JOIN UpsertTable u ON a.id = u.key_id")
                    .build();
}