(flink) branch master updated: [FLINK-40112][table-planner] Report changelog mode mismatch instead of a generic planning error

[twalthr]

This is an automated email from the ASF dual-hosted git repository.

twalthr 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 8b7ebcacddc [FLINK-40112][table-planner] Report changelog mode 
mismatch instead of a generic planning error
8b7ebcacddc is described below

commit 8b7ebcacddcb43168879f1d6135b9714270cae6d
Author: Gustavo de Morais <gustavopg...@gmail.com>
AuthorDate: Fri Jul 10 09:11:33 2026 +0200

    [FLINK-40112][table-planner] Report changelog mode mismatch instead of a 
generic planning error
    
    This closes #28703.
---
 .../plan/optimize/program/FlinkChangelogModeInferenceProgram.scala  | 4 +++-
 .../table/planner/plan/stream/sql/ProcessTableFunctionTest.java     | 6 ++++--
 2 files changed, 7 insertions(+), 3 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 53efd6e8f50..77fc7208d24 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
@@ -141,7 +141,9 @@ class FlinkChangelogModeInferenceProgram extends 
FlinkOptimizeProgram[StreamOpti
         "The conflict is at:\n" + conflict
     } else {
       val plan = FlinkRelOptUtil.toString(rootWithModifyKindSet, 
withChangelogTraits = true)
-      "Can't generate a valid execution plan for the given query:\n" + plan
+      "Can't generate a valid execution plan for the given query because of a 
changelog mismatch: " +
+        "an operator cannot produce the changelog its consumer requires. 
Review the changelog " +
+        "modes of the operators in the plan below:\n" + plan
     }
   }
 
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 4606e2ec493..d34b3f3bc8d 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
@@ -544,14 +544,16 @@ class ProcessTableFunctionTest extends TableTestBase {
                         UpdatingUpsertFunction.class,
                         "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"),
+                        "Can't generate a valid execution plan for the given 
query because of a "
+                                + "changelog mismatch"),
                 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"),
+                        "Can't generate a valid execution plan for the given 
query because of a "
+                                + "changelog mismatch"),
                 ErrorSpec.ofInsertInto(
                         "upsert conflict buried below a calc",
                         UpdatingUpsertFunction.class,