This is an automated email from the ASF dual-hosted git repository.
MartijnVisser 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 dd2bc4b41ce [FLINK-40540][table-planner] Put the rule's reason first
in the planning error
dd2bc4b41ce is described below
commit dd2bc4b41ce78a5b361d5336b8399b1d5ac0e8a9
Author: Martijn Visser <[email protected]>
AuthorDate: Wed Sep 2 17:17:58 2026 +0200
[FLINK-40540][table-planner] Put the rule's reason first in the planning
error
A rule that rejects a query already explains why, but the message buried
that sentence behind the full logical plan. Put the reason on the header
line and mark the plan with "Plan:" so that tools rewriting the message
can find it. The wrapper type and cause chain stay as they are. A
ValidationException thrown while applying a rule now gets the same
treatment instead of leaking as Calcite's RuntimeException.
Generated-by: Claude Code (Claude Fable 5.1)
---
.../optimize/program/FlinkVolcanoProgram.scala | 63 +++++++++++++++-------
.../plan/batch/sql/agg/GroupWindowTest.scala | 8 ++-
.../optimize/program/FlinkVolcanoProgramTest.scala | 41 +++++++++++++-
.../plan/stream/sql/agg/OverAggregateTest.scala | 4 ++
4 files changed, 94 insertions(+), 22 deletions(-)
diff --git
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/optimize/program/FlinkVolcanoProgram.scala
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/optimize/program/FlinkVolcanoProgram.scala
index ac626c94738..32a4c9809fc 100644
---
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/optimize/program/FlinkVolcanoProgram.scala
+++
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/optimize/program/FlinkVolcanoProgram.scala
@@ -17,7 +17,8 @@
*/
package org.apache.flink.table.planner.plan.optimize.program
-import org.apache.flink.table.api.TableException
+import org.apache.flink.annotation.VisibleForTesting
+import org.apache.flink.table.api.{TableException, ValidationException}
import org.apache.flink.table.planner.plan.metadata.FlinkRelMdNonCumulativeCost
import org.apache.flink.table.planner.plan.utils.FlinkRelOptUtil
import org.apache.flink.util.Preconditions
@@ -29,6 +30,8 @@ import org.apache.calcite.plan.volcano.VolcanoPlanner
import org.apache.calcite.rel.RelNode
import org.apache.calcite.tools.{Programs, RuleSet}
+import scala.annotation.tailrec
+
/**
* A FlinkRuleSetProgram that runs with
[[org.apache.calcite.plan.volcano.VolcanoPlanner]].
*
@@ -63,32 +66,54 @@ class FlinkVolcanoProgram[OC <: FlinkOptimizeContext]
extends FlinkRuleSetProgra
} catch {
case e: CannotPlanException =>
throw new TableException(
- s"Cannot generate a valid execution plan for the given query: \n\n" +
- s"${FlinkRelOptUtil.toString(root)}\n" +
- s"This exception indicates that the query uses an unsupported SQL
feature.\n" +
- s"Please check the documentation for the set of currently
supported SQL features.",
+ rejectionMessage(
+ "This exception indicates that the query uses an unsupported SQL
feature. " +
+ "Please check the documentation for the set of currently
supported SQL features.",
+ root
+ ),
e)
- case t: TableException =>
- throw new TableException(
- s"Cannot generate a valid execution plan for the given query: \n\n" +
- s"${FlinkRelOptUtil.toString(root)}\n" +
- s"${t.getMessage}\n" +
- s"Please check the documentation for the set of currently
supported SQL features.",
- t)
case a: AssertionError =>
throw new AssertionError(s"Sql optimization: Assertion error:
${a.getMessage}", a)
- case r: RuntimeException if r.getCause.isInstanceOf[TableException] =>
- throw new TableException(
- s"Sql optimization: Cannot generate a valid execution plan for the
given query: \n\n" +
- s"${FlinkRelOptUtil.toString(root)}\n" +
- s"${r.getCause.getMessage}\n" +
- s"Please check the documentation for the set of currently
supported SQL features.",
- r.getCause)
+ case r: RuntimeException =>
+ unwrapRuleException(r) match {
+ case t: TableException =>
+ throw new TableException(rejectionMessage(t.getMessage, root), t)
+ case v: ValidationException =>
+ throw new ValidationException(rejectionMessage(v.getMessage,
root), v)
+ case other => throw other
+ }
} finally {
FlinkRelMdNonCumulativeCost.THREAD_PLANNER.remove()
}
}
+ /**
+ * Puts the reason first so that it is the first line users see, and marks
the plan with a prefix
+ * so that tools rewriting the message can find it.
+ */
+ private def rejectionMessage(reason: String, root: RelNode): String = {
+ s"Cannot generate a valid execution plan for the given query: $reason\n\n"
+
+ s"Plan:\n${FlinkRelOptUtil.toString(root)}"
+ }
+
+ /**
+ * Returns the [[TableException]] or [[ValidationException]] thrown by a
rule, looking through the
+ * plain [[RuntimeException]]s Calcite wraps it in while applying the rule.
Any other exception is
+ * returned unchanged.
+ */
+ @VisibleForTesting
+ private[program] def unwrapRuleException(r: RuntimeException):
RuntimeException = {
+ @tailrec
+ def findRuleException(t: Throwable): Option[RuntimeException] = t match {
+ case e: TableException => Some(e)
+ case e: ValidationException => Some(e)
+ case e if e != null && e.getClass == classOf[RuntimeException] =>
+ findRuleException(e.getCause)
+ case _ => None
+ }
+ findRuleException(r).getOrElse(r)
+ }
+
/** Sets required output traits. */
def setRequiredOutputTraits(relTraits: Array[RelTrait]): Unit = {
Preconditions.checkNotNull(relTraits)
diff --git
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/agg/GroupWindowTest.scala
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/agg/GroupWindowTest.scala
index dc38aa53fc3..ca715963f67 100644
---
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/agg/GroupWindowTest.scala
+++
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/agg/GroupWindowTest.scala
@@ -290,8 +290,12 @@ class GroupWindowTest(aggStrategy: AggregatePhaseStrategy)
extends TableTestBase
def testNonPartitionedSessionWindow(): Unit = {
val sqlQuery = "SELECT COUNT(*) AS cnt FROM MyTable2 GROUP BY SESSION(ts,
INTERVAL '30' MINUTE)"
assertThatThrownBy(() => util.verifyExecPlan(sqlQuery))
- .hasMessageContaining("Cannot generate a valid execution plan for the
given query")
- .isInstanceOf[TableException]
+ .isInstanceOf(classOf[TableException])
+ // the rule's reason comes first, the plan is marked so that tools can
strip it
+ .hasMessageStartingWith(
+ "Cannot generate a valid execution plan for the given query: Window
SessionGroupWindow(")
+ .hasMessageContaining("is not supported right
now.\n\nPlan:\nFlinkLogicalWindowAggregate(")
+ .hasCauseInstanceOf(classOf[TableException])
}
@TestTemplate
diff --git
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/optimize/program/FlinkVolcanoProgramTest.scala
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/optimize/program/FlinkVolcanoProgramTest.scala
index 5b987230ca1..c286665384e 100644
---
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/optimize/program/FlinkVolcanoProgramTest.scala
+++
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/optimize/program/FlinkVolcanoProgramTest.scala
@@ -17,11 +17,14 @@
*/
package org.apache.flink.table.planner.plan.optimize.program
+import org.apache.flink.table.api.{TableException, ValidationException}
+
import org.apache.calcite.plan.Convention
+import org.apache.calcite.plan.RelOptPlanner.CannotPlanException
import org.apache.calcite.rel.RelNode
import org.apache.calcite.rel.rules._
import org.apache.calcite.tools.RuleSets
-import org.assertj.core.api.Assertions.assertThatThrownBy
+import org.assertj.core.api.Assertions.{assertThat, assertThatThrownBy}
import org.junit.jupiter.api.Test
/** Tests for [[FlinkVolcanoProgramTest]]. */
@@ -48,4 +51,40 @@ class FlinkVolcanoProgramTest {
.isInstanceOf(classOf[NullPointerException])
}
+ @Test
+ def testUnwrapRuleExceptionReturnsFlinkExceptionsAsIs(): Unit = {
+ val program = FlinkVolcanoProgramBuilder.newBuilder.build()
+ val tableException = new TableException("rejected by rule")
+ val validationException = new ValidationException("invalid for rule")
+
+
assertThat(program.unwrapRuleException(tableException)).isSameAs(tableException)
+
assertThat(program.unwrapRuleException(validationException)).isSameAs(validationException)
+ }
+
+ @Test
+ def testUnwrapRuleExceptionLooksThroughCalciteWrappers(): Unit = {
+ val program = FlinkVolcanoProgramBuilder.newBuilder.build()
+ val cause = new ValidationException("invalid for rule")
+ val wrappedOnce = new RuntimeException("Error while applying rule", cause)
+ val wrappedTwice = new RuntimeException("Error occurred while applying
rule", wrappedOnce)
+
+ assertThat(program.unwrapRuleException(wrappedOnce)).isSameAs(cause)
+ assertThat(program.unwrapRuleException(wrappedTwice)).isSameAs(cause)
+ }
+
+ @Test
+ def testUnwrapRuleExceptionKeepsOtherExceptionsWrapped(): Unit = {
+ val program = FlinkVolcanoProgramBuilder.newBuilder.build()
+ val wrappedBug = new RuntimeException("Error while applying rule", new
NullPointerException())
+ val wrappedCannotPlan =
+ new RuntimeException("Error while applying rule", new
CannotPlanException("no plan"))
+ val subclassWrapper = new IllegalStateException("wrapped", new
TableException("rejected"))
+ val withoutCause = new RuntimeException("Error while applying rule")
+
+ assertThat(program.unwrapRuleException(wrappedBug)).isSameAs(wrappedBug)
+
assertThat(program.unwrapRuleException(wrappedCannotPlan)).isSameAs(wrappedCannotPlan)
+
assertThat(program.unwrapRuleException(subclassWrapper)).isSameAs(subclassWrapper)
+
assertThat(program.unwrapRuleException(withoutCause)).isSameAs(withoutCause)
+ }
+
}
diff --git
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/agg/OverAggregateTest.scala
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/agg/OverAggregateTest.scala
index c8c7c087802..bb57aba9c4e 100644
---
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/agg/OverAggregateTest.scala
+++
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/agg/OverAggregateTest.scala
@@ -66,6 +66,10 @@ class OverAggregateTest extends TableTestBase {
""".stripMargin
assertThatThrownBy(() => util.verifyExecPlan(sqlQuery))
+ // thrown while applying a rule: the exception keeps its type instead of
Calcite's wrapper
+ .isInstanceOf(classOf[ValidationException])
+ .hasMessageStartingWith("Cannot generate a valid execution plan for the
given query: " +
+ "Frame exclusion 'EXCLUDE GROUP' is not supported in over
windows.\n\nPlan:\n")
.hasRootCauseInstanceOf(classOf[ValidationException])
.hasRootCauseMessage("Frame exclusion 'EXCLUDE GROUP' is not supported
in over windows.")
}
