gustavodemorais commented on code in PR #28199:
URL: https://github.com/apache/flink/pull/28199#discussion_r3280422142
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ToChangelogTestPrograms.java:
##########
@@ -215,6 +215,37 @@ public class ToChangelogTestPrograms {
+ "op_mapping =>
MAP['INSERT,UPDATE_AFTER', 'C', 'DELETE', 'D'])")
.build();
+ public static final TableTestProgram UPSERT_PARTITION_BY_KEY_ONLY_DELETES =
+ TableTestProgram.of(
+
"to-changelog-upsert-partition-by-key-only-deletes",
+ "PARTITION BY upsert key + mapping without
UB/DELETE handles key-only deletes")
+ .setupTableSource(
+ SourceTestStep.newBuilder("t")
+ .addSchema(
+ "name STRING PRIMARY KEY NOT
ENFORCED", "score BIGINT")
+ .addMode(ChangelogMode.upsert(true))
+ .producedValues(
+ Row.ofKind(RowKind.INSERT,
"Alice", 10L),
+ Row.ofKind(RowKind.INSERT, "Bob",
20L),
+ Row.ofKind(RowKind.UPDATE_AFTER,
"Alice", 30L),
+ // Key-only delete: source only
knows the key.
+ Row.ofKind(RowKind.DELETE, "Bob",
null))
+ .build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema("name STRING", "op STRING",
"score BIGINT")
+ .consumedValues(
+ "+I[Alice, U, 10]",
+ "+I[Bob, U, 20]",
+ "+I[Alice, U, 30]")
Review Comment:
Yes, should be doable with row semantics, without partition by since the the
ptf is chained with the source and we don't add a ChangelogNormalize
--
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]
