This is an automated email from the ASF dual-hosted git repository.
gustavodemorais pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 8cc54fb6ba5 [FLINK-40018][table] Throw a clear error when a PTF table
argument input cannot satisfy the required changelog mode
8cc54fb6ba5 is described below
commit 8cc54fb6ba5decfb1139f153117d0927d112e956
Author: Gustavo de Morais <[email protected]>
AuthorDate: Thu Jul 2 10:36:20 2026 +0200
[FLINK-40018][table] Throw a clear error when a PTF table argument input
cannot satisfy the required changelog mode
This closes #28584.
---
.../FlinkChangelogModeInferenceProgram.scala | 25 +++++++++++++---------
.../plan/stream/sql/ProcessTableFunctionTest.java | 13 +++++++++++
2 files changed, 28 insertions(+), 10 deletions(-)
diff --git
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/optimize/program/FlinkChangelogModeInferenceProgram.scala
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/optimize/program/FlinkChangelogModeInferenceProgram.scala
index 0987d022150..53efd6e8f50 100644
---
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/optimize/program/FlinkChangelogModeInferenceProgram.scala
+++
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/optimize/program/FlinkChangelogModeInferenceProgram.scala
@@ -857,7 +857,7 @@ class FlinkChangelogModeInferenceProgram extends
FlinkOptimizeProgram[StreamOpti
// input traits, partition keys, and upsert keys
val inputArgs = StreamPhysicalProcessTableFunction
.getProvidedInputArgs(process.getCall)
- val children = process.getInputs
+ val childOpts = process.getInputs
.map(_.asInstanceOf[StreamPhysicalRel])
.zipWithIndex
.map {
@@ -888,15 +888,20 @@ class FlinkChangelogModeInferenceProgram extends
FlinkOptimizeProgram[StreamOpti
}
}
.toList
- .flatten
- // Query PTF for upsert vs. retract
- val providedUpdateTrait = queryPtfChangelogMode(
- process,
- children,
- toChangelogMode(process, Some(requiredUpdateTrait), None),
- UpdateKindTrait.fromChangelogMode,
- UpdateKindTrait.NONE)
- createNewNode(rel, Some(children), providedUpdateTrait)
+
+ if (childOpts.exists(_.isEmpty)) {
+ None
+ } else {
+ val children = childOpts.flatten
+ // Query PTF for upsert vs. retract
+ val providedUpdateTrait = queryPtfChangelogMode(
+ process,
+ children,
+ toChangelogMode(process, Some(requiredUpdateTrait), None),
+ UpdateKindTrait.fromChangelogMode,
+ UpdateKindTrait.NONE)
+ createNewNode(rel, Some(children), providedUpdateTrait)
+ }
case multiJoin: StreamPhysicalMultiJoin =>
val onlyAfterByParent = requiredUpdateTrait.updateKind ==
UpdateKind.ONLY_UPDATE_AFTER
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/ProcessTableFunctionTest.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/ProcessTableFunctionTest.java
index 9f06973ce7c..4606e2ec493 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/ProcessTableFunctionTest.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/ProcessTableFunctionTest.java
@@ -43,6 +43,7 @@ import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctio
import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.ScalarArgsFunction;
import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.SetSemanticTableFunction;
import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.SetSemanticTablePassThroughFunction;
+import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.SetSemanticTableRetractArgFunction;
import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.TypedRowSemanticTableFunction;
import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.TypedSetSemanticTableFunction;
import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.UpdatingRetractRowSemanticFunction;
@@ -95,6 +96,11 @@ class ProcessTableFunctionTest extends TableTestBase {
.executeSql("CREATE VIEW t_type_diff AS SELECT 'Bob' AS name,
TRUE AS isValid");
util.tableEnv()
.executeSql("CREATE VIEW t_updating AS SELECT name, COUNT(*)
FROM t GROUP BY name");
+
+ util.addTemporarySystemFunction("upsertPtf",
UpdatingUpsertFunction.class);
+ util.tableEnv()
+ .executeSql(
+ "CREATE VIEW t_upsert AS SELECT * FROM upsertPtf(r =>
TABLE t_updating PARTITION BY name)");
util.tableEnv()
.executeSql(
"CREATE TABLE t_watermarked (name STRING, score INT,
ts TIMESTAMP_LTZ(3), WATERMARK FOR ts AS ts) "
@@ -539,6 +545,13 @@ class ProcessTableFunctionTest extends TableTestBase {
"SELECT name, SUM(`count`) OVER (PARTITION BY name
ORDER BY name) "
+ "FROM f(r => TABLE t_updating PARTITION BY
name)",
"Can't generate a valid execution plan for the given
query:\n"),
+ ErrorSpec.ofSelect(
+ // t_upsert produces an upsert changelog.
+ // the table argument for f requires a retract
changelog
+ "retract requirement on an upsert table arg",
+ SetSemanticTableRetractArgFunction.class,
+ "SELECT * FROM f(r => TABLE t_upsert PARTITION BY
name)",
+ "Can't generate a valid execution plan for the given
query:\n"),
ErrorSpec.ofInsertInto(
"upsert conflict buried below a calc",
UpdatingUpsertFunction.class,
