dawidwys commented on code in PR #26591: URL: https://github.com/apache/flink/pull/26591#discussion_r2111206408
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ChangelogNormalizeTestPrograms.java:
##########
@@ -199,4 +200,87 @@ public class ChangelogNormalizeTestPrograms {
.build())
.runSql("INSERT INTO sink_t SELECT a, b, c FROM source_t
WHERE b < 10")
.build();
+
+ static final TableTestProgram UPSERT_SOURCE_WITH_FILTER_ON_WATERMARK =
+ TableTestProgram.of(
+ "changelog-normalize-upsert-filter-watermark",
+ "validates changelog normalize upsert with filter
using current_watermark")
+ .setupConfig(
+
ExecutionConfigOptions.TABLE_EXEC_SOURCE_CDC_EVENTS_DUPLICATE, true)
+ .setupTableSource(
+ SourceTestStep.newBuilder("source_t")
+ .addOption("changelog-mode", "I,UA,D")
+ .addSchema(
+ "a VARCHAR",
+ "b INT NOT NULL",
+ "c VARCHAR",
+ "d TIMESTAMP_LTZ(3)",
+ "WATERMARK FOR d AS d",
+ "PRIMARY KEY(a) NOT ENFORCED")
+ .producedBeforeRestore(
+ Row.ofKind(
+ RowKind.UPDATE_AFTER,
+ "one",
+ 1,
+ "a",
+ Instant.ofEpochMilli(1L)),
+ Row.ofKind(
+ RowKind.UPDATE_AFTER,
+ "one",
+ 2,
+ "b",
+ Instant.ofEpochMilli(1L)),
+ Row.ofKind(
+ RowKind.UPDATE_AFTER,
+ "one",
+ 12,
+ "b",
+ Instant.ofEpochMilli(1L)),
+ Row.ofKind(
+ RowKind.UPDATE_AFTER,
+ "one",
+ 13,
+ "b",
+ Instant.ofEpochMilli(1L)),
+ Row.ofKind(
+ RowKind.UPDATE_AFTER,
+ "three",
+ 3,
+ "cc",
+ Instant.ofEpochMilli(1L)))
+ .producedAfterRestore(
+ Row.ofKind(
+ RowKind.UPDATE_AFTER,
+ "one",
+ 15,
+ "aa",
+ Instant.ofEpochMilli(1L)),
+ Row.ofKind(
+ RowKind.DELETE,
+ "one",
+ 15,
+ "c",
+ Instant.ofEpochMilli(1L)),
+ Row.ofKind(
+ RowKind.DELETE,
+ "three",
+ 3,
+ "cc",
+ Instant.ofEpochMilli(1L)))
+ .build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink_t")
+ .addSchema(SINK_SCHEMA)
+ .consumedBeforeRestore(
+ "+I[one, 1, a]",
+ "-U[one, 1, a]",
+ "+U[one, 2, b]",
+ "-D[one, 2, b]",
+ "+I[three, 3, cc]")
+ .consumedAfterRestore("-D[three, 3, cc]")
+ .build())
+ .runSql(
+ "INSERT INTO sink_t SELECT a, b, c FROM source_t
WHERE b < 10 AND "
+ + "CURRENT_WATERMARK(d) IS NULL")
Review Comment:
Correct
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ChangelogNormalizeRestoreTest.java:
##########
@@ -40,6 +40,7 @@ public List<TableTestProgram> programs() {
ChangelogNormalizeTestPrograms.CHANGELOG_SOURCE,
ChangelogNormalizeTestPrograms.CHANGELOG_SOURCE_MINI_BATCH,
ChangelogNormalizeTestPrograms.UPSERT_SOURCE,
- ChangelogNormalizeTestPrograms.UPSERT_SOURCE_WITH_FILTER);
+ ChangelogNormalizeTestPrograms.UPSERT_SOURCE_WITH_FILTER,
+
ChangelogNormalizeTestPrograms.UPSERT_SOURCE_WITH_FILTER_ON_WATERMARK);
Review Comment:
I forgot to commit them, sorry
--
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: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
