You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "dawidwys (via GitHub)" <gi...@apache.org> on 2025/05/22 15:39:19 UTC
[PR] [FLINK-37829] ChangelogNormalize fails with CURRENT_WATERMARK in the filter condition [flink]
dawidwys opened a new pull request, #26591:
URL: https://github.com/apache/flink/pull/26591
## What is the purpose of the change
Fix code generation exceptions when an expression requires access to time service.
## Verifying this change
Added a test in `ChangelogNormalizeRestoreTest`
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): (yes / **no**)
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (yes / **no**)
- The serializers: (yes / **no** / don't know)
- The runtime per-record code paths (performance sensitive): (yes / **no** / don't know)
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (yes / **no** / don't know)
- The S3 file system connector: (yes / **no** / don't know)
## Documentation
- Does this pull request introduce a new feature? (yes / **no**)
- If yes, how is the feature documented? (**not applicable** / docs / JavaDocs / not documented)
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37829] ChangelogNormalize fails with CURRENT_WATERMARK in the filter condition [flink]
Posted by "gustavodemorais (via GitHub)" <gi...@apache.org>.
gustavodemorais commented on code in PR #26591:
URL: https://github.com/apache/flink/pull/26591#discussion_r2106329686
##########
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:
It's always null for all the events here because we don't progress the watermark, correct?
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37829] ChangelogNormalize fails with CURRENT_WATERMARK in the filter condition [flink]
Posted by "dawidwys (via GitHub)" <gi...@apache.org>.
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37829] ChangelogNormalize fails with CURRENT_WATERMARK in the filter condition [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #26591:
URL: https://github.com/apache/flink/pull/26591#issuecomment-2901715558
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1c7a4d5a87922c201d151f684f2c5cdafd88a3f5",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "1c7a4d5a87922c201d151f684f2c5cdafd88a3f5",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 1c7a4d5a87922c201d151f684f2c5cdafd88a3f5 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37829] ChangelogNormalize fails with CURRENT_WATERMARK in the filter condition [flink]
Posted by "dawidwys (via GitHub)" <gi...@apache.org>.
dawidwys commented on PR #26591:
URL: https://github.com/apache/flink/pull/26591#issuecomment-2915359339
@gustavodemorais Testing using `CURRENT_WATERMARK` is generally tricky. Watermarks are generated every 200ms, so it's hard to keep a deterministic behaviour. I modified the test a bit though, so that we also test against a concrete value.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37829] ChangelogNormalize fails with CURRENT_WATERMARK in the filter condition [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #26591:
URL: https://github.com/apache/flink/pull/26591#discussion_r2106880066
##########
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:
mostly lgtm
however seems plan is missing since ci is complaining about that
https://dev.azure.com/apache-flink/apache-flink/_build/results?buildId=67780&view=logs&j=0c940707-2659-5648-cbe6-a1ad63045f0a&t=019d96ac-af5c-5c6d-106d-b67f63cc6e96&l=11806
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37829] ChangelogNormalize fails with CURRENT_WATERMARK in the filter condition [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #26591:
URL: https://github.com/apache/flink/pull/26591#discussion_r2104277157
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/FunctionCodeGenerator.scala:
##########
@@ -204,14 +204,15 @@ object FunctionCodeGenerator {
clazz: Class[F],
bodyCode: String,
input1Term: String = CodeGenUtils.DEFAULT_INPUT1_TERM,
- input2Term: String = CodeGenUtils.DEFAULT_INPUT2_TERM): (String, String) = {
+ input2Term: String = CodeGenUtils.DEFAULT_INPUT2_TERM,
+ contextTerm: String = CodeGenUtils.DEFAULT_CONTEXT_TERM): (String, String) = {
Review Comment:
nit: we should add contextTerm to the javadoc
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37829] ChangelogNormalize fails with CURRENT_WATERMARK in the filter condition [flink]
Posted by "gustavodemorais (via GitHub)" <gi...@apache.org>.
gustavodemorais commented on code in PR #26591:
URL: https://github.com/apache/flink/pull/26591#discussion_r2106329686
##########
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:
It's always null because we don't progress the watermark, correct?
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/generated/FilterCondition.java:
##########
@@ -27,5 +31,66 @@ public interface FilterCondition extends RichFunction {
/**
* @return true if the filter condition stays true for the input row
*/
- boolean apply(RowData in);
+ boolean apply(Context ctx, RowData in);
+
+ /**
+ * Context for generating expressions such as e.g. {@code CURRENT_WATERMARK} or {@code
+ * STREAMRECORD_TIMESTAMP}.
+ *
+ * @see ProcessFunction.Context
+ */
+ interface Context {
+ /**
+ * Timestamp of the element currently being processed or timestamp of a firing timer.
+ *
+ * <p>This might be {@code null}, depending on the stream's watermark strategy.
+ */
+ Long timestamp();
+
+ /** A {@link TimerService} for querying time and registering timers. */
+ TimerService timerService();
+
+ static Context of(KeyedProcessFunction<?, ?, ?>.Context context) {
+ return new Context() {
+ @Override
+ public Long timestamp() {
+ return context.timestamp();
+ }
+
+ @Override
+ public TimerService timerService() {
+ return context.timerService();
+ }
+ };
+ }
+
+ static Context of(ProcessFunction<?, ?>.Context context) {
+ return new Context() {
+ @Override
+ public Long timestamp() {
+ return context.timestamp();
+ }
+
+ @Override
+ public TimerService timerService() {
+ return context.timerService();
+ }
+ };
+ }
+
+ Context INVALID_CONTEXT =
Review Comment:
Nice
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37829] ChangelogNormalize fails with CURRENT_WATERMARK in the filter condition [flink]
Posted by "dawidwys (via GitHub)" <gi...@apache.org>.
dawidwys merged PR #26591:
URL: https://github.com/apache/flink/pull/26591
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org