gustavodemorais commented on code in PR #29474:
URL: https://github.com/apache/flink/pull/29474#discussion_r4334457010
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ToChangelogTestPrograms.java:
##########
@@ -481,6 +484,107 @@ public class ToChangelogTestPrograms {
.runTableApi(env ->
env.from("t").partitionBy($("id")).toChangelog(), "sink")
.build();
+ public static final TableTestProgram TABLE_API_CUSTOM_OP_MAPPING =
+ TableTestProgram.of(
+ "to-changelog-table-api-custom-op-mapping",
+ "Table.toChangelog() with a map() op_mapping maps
change operations to user-defined codes")
+ .setupTableSource(
+ SourceTestStep.newBuilder("t")
+ .addSchema(
+ "name STRING PRIMARY KEY NOT
ENFORCED", "score BIGINT")
+ .addMode(ChangelogMode.all())
+ .producedValues(
+ Row.ofKind(RowKind.INSERT,
"Alice", 10L),
+ Row.ofKind(RowKind.INSERT, "Bob",
20L),
+ Row.ofKind(RowKind.UPDATE_BEFORE,
"Alice", 10L),
+ Row.ofKind(RowKind.UPDATE_AFTER,
"Alice", 30L),
+ Row.ofKind(RowKind.DELETE, "Bob",
20L))
+ .build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema("op_code STRING", "name
STRING", "score BIGINT")
+ .consumedValues(
+ "+I[I, Alice, 10]",
+ "+I[I, Bob, 20]",
+ "+I[U, Alice, 30]")
+ .build())
+ .runTableApi(
+ env ->
+ env.from("t")
+ .toChangelog(
+
descriptor("op_code").asArgument("op"),
+ map("INSERT", "I",
"UPDATE_AFTER", "U")
+
.asArgument("op_mapping")),
+ "sink")
+ .build();
+
+ public static final TableTestProgram TABLE_API_DELETION_FLAG =
+ TableTestProgram.of(
+ "to-changelog-table-api-deletion-flag",
+ "Table.toChangelog() with a map() op_mapping
produces deletion flag output")
+ .setupTableSource(
+ SourceTestStep.newBuilder("t")
+ .addSchema(
+ "name STRING PRIMARY KEY NOT
ENFORCED", "score BIGINT")
+ .addMode(ChangelogMode.all())
+ .producedValues(
+ Row.ofKind(RowKind.INSERT,
"Alice", 10L),
+ Row.ofKind(RowKind.INSERT, "Bob",
20L),
+ Row.ofKind(RowKind.UPDATE_BEFORE,
"Alice", 10L),
+ Row.ofKind(RowKind.UPDATE_AFTER,
"Alice", 30L),
+ Row.ofKind(RowKind.DELETE, "Bob",
20L))
+ .build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema("deleted STRING", "name
STRING", "score BIGINT")
+ .consumedValues(
+ "+I[false, Alice, 10]",
+ "+I[false, Bob, 20]",
+ "+I[false, Alice, 30]",
+ "+I[true, Bob, 20]")
+ .build())
+ .runTableApi(
+ env ->
+ env.from("t")
+ .toChangelog(
+
descriptor("deleted").asArgument("op"),
+ map(
+ "INSERT,
UPDATE_AFTER",
+ "false",
+ "DELETE",
+ "true")
+
.asArgument("op_mapping")),
+ "sink")
+ .build();
+
+ public static final TableTestProgram TABLE_API_PRODUCES_PARTIAL_DELETES =
+ TableTestProgram.of(
+ "to-changelog-table-api-produces-partial-deletes",
+ "Table.toChangelog() with
produces_full_deletes=false keeps only the "
+ + "upsert key on DELETE")
+ .setupTableSource(
+ SourceTestStep.newBuilder("t")
+ .addSchema(
+ "name STRING PRIMARY KEY NOT
ENFORCED", "score BIGINT")
+ .addMode(ChangelogMode.upsert())
+ .producedValues(
+ Row.ofKind(RowKind.INSERT,
"Alice", 10L),
+ Row.ofKind(RowKind.DELETE,
"Alice", 10L))
+ .build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema("op STRING", "name STRING",
"score BIGINT")
+ .consumedValues(
+ "+I[INSERT, Alice, 10]",
"+I[DELETE, Alice, null]")
+ .build())
+ .runTableApi(
+ env ->
+ env.from("t")
+ .toChangelog(
+
lit(false).asArgument("produces_full_deletes")),
+ "sink")
+ .build();
+
Review Comment:
Just one small ask for completeness. Could you add negative tests: map()
with nested call, non literal and NULL entry? We expect the test to fail with
the proper error msg.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
