You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "Ahmad Humayun (Jira)" <ji...@apache.org> on 2025/09/21 17:31:00 UTC
[jira] [Created] (FLINK-38397) Runtime Exception in Planner
Ahmad Humayun created FLINK-38397:
-------------------------------------
Summary: Runtime Exception in Planner
Key: FLINK-38397
URL: https://issues.apache.org/jira/browse/FLINK-38397
Project: Flink
Issue Type: Bug
Components: Table SQL / Planner
Affects Versions: 2.0.0
Environment:
{code:java}
python 3.10.12
apache-flink 2.0.0
{code}
Reporter: Ahmad Humayun
The following results in the planner throwing a Runtime exception. My best guess is that it seems to be an issue with the planner/optimizer incorrectly estimating the number of columns available at a certain point.
{code:java}
from pyflink.table import EnvironmentSettings, TableEnvironment, Table
from pyflink.common import Configuration
from pyflink.table.expressions import col
cfg = Configuration()
settings = (
EnvironmentSettings.new_instance()
.in_batch_mode()
.with_configuration(cfg)
.build()
)
table_env = TableEnvironment.create(settings)
# =============
data = [
(1, "AAAAAAAABAAAAAAA", "Jimmy Allen", "3rd", -5.00),
(2, "AAAAAAAACAAAAAAA", "Jimmy Bullock", "Cedar Spruce", -5.00),
(3, "AAAAAAAACAAAAAAA", "Floyd Christian", "8th", -5.00),
(4, "AAAAAAAAEAAAAAAA", "James Lachance", "6th", -5.00),
(5, "AAAAAAAAEAAAAAAA", "James Lachance", "9th 12th", -5.00),
(6, "AAAAAAAAEAAAAAAA", "Joaquin Washington", "Adams", -5.00),
(7, "AAAAAAAAHAAAAAAA", "Michael Burton", "3rd", -5.00)
]
schema = [
"s_store_sk",
"s_store_id",
"s_manager",
"s_street_name",
"s_gmt_offset"
]
# =====================
source_table = table_env.from_elements(
data,
schema=schema
)
ordered = source_table.order_by(col('s_manager'))
aggregated = ordered.group_by(col('s_street_name')).select(col('s_gmt_offset').count.alias('s_gmt_offset'))
print(aggregated.explain())
{code}
This code throws the following error:
{code:java}
py4j.protocol.Py4JJavaError: An error occurred while calling o70.explain.: java.lang.RuntimeException: Error while applying rule FlinkExpandConversionRule, args [rel#219:AbstractConverter.BATCH_PHYSICAL.hash[0]true.[2](input=RelSubset#217,convention=BATCH_PHYSICAL,FlinkRelDistributionTraitDef=hash[0]true,sort=[2]), rel#216:BatchPhysicalLocalHashAggregate.BATCH_PHYSICAL.any.[2](input=RelSubset#215,groupBy=s_street_name,select=s_street_name, Partial_COUNT(s_gmt_offset) AS count$0)] at org.apache.calcite.plan.volcano.VolcanoRuleCall.onMatch(VolcanoRuleCall.java:250) at org.apache.calcite.plan.volcano.IterativeRuleDriver.drive(IterativeRuleDriver.java:59) at org.apache.calcite.plan.volcano.VolcanoPlanner.findBestExp(VolcanoPlanner.java:523) at org.apache.calcite.tools.Programs$RuleSetProgram.run(Programs.java:317) at org.apache.flink.table.planner.plan.optimize.program.FlinkVolcanoProgram.optimize(FlinkVolcanoProgram.scala:62) at org.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgram.$anonfun$optimize$1(FlinkChainedProgram.scala:59) at scala.collection.TraversableOnce$folder$1.apply(TraversableOnce.scala:196) at scala.collection.TraversableOnce$folder$1.apply(TraversableOnce.scala:194) at scala.collection.Iterator.foreach(Iterator.scala:943) at scala.collection.Iterator.foreach$(Iterator.scala:943) at scala.collection.AbstractIterator.foreach(Iterator.scala:1431) at scala.collection.IterableLike.foreach(IterableLike.scala:74) at scala.collection.IterableLike.foreach$(IterableLike.scala:73) at scala.collection.AbstractIterable.foreach(Iterable.scala:56) at scala.collection.TraversableOnce.foldLeft(TraversableOnce.scala:199) at scala.collection.TraversableOnce.foldLeft$(TraversableOnce.scala:192) at scala.collection.AbstractTraversable.foldLeft(Traversable.scala:108) at org.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgram.optimize(FlinkChainedProgram.scala:55) at org.apache.flink.table.planner.plan.optimize.BatchCommonSubGraphBasedOptimizer.optimizeTree(BatchCommonSubGraphBasedOptimizer.scala:93) at org.apache.flink.table.planner.plan.optimize.BatchCommonSubGraphBasedOptimizer.optimizeBlock(BatchCommonSubGraphBasedOptimizer.scala:58) at org.apache.flink.table.planner.plan.optimize.BatchCommonSubGraphBasedOptimizer.$anonfun$doOptimize$1(BatchCommonSubGraphBasedOptimizer.scala:45) at org.apache.flink.table.planner.plan.optimize.BatchCommonSubGraphBasedOptimizer.$anonfun$doOptimize$1$adapted(BatchCommonSubGraphBasedOptimizer.scala:45) at scala.collection.immutable.List.foreach(List.scala:431) at org.apache.flink.table.planner.plan.optimize.BatchCommonSubGraphBasedOptimizer.doOptimize(BatchCommonSubGraphBasedOptimizer.scala:45) at org.apache.flink.table.planner.plan.optimize.CommonSubGraphBasedOptimizer.optimize(CommonSubGraphBasedOptimizer.scala:87) at org.apache.flink.table.planner.delegation.PlannerBase.optimize(PlannerBase.scala:390) at org.apache.flink.table.planner.delegation.PlannerBase.getExplainGraphs(PlannerBase.scala:625) at org.apache.flink.table.planner.delegation.BatchPlanner.explain(BatchPlanner.scala:149) at org.apache.flink.table.planner.delegation.BatchPlanner.explain(BatchPlanner.scala:49) at org.apache.flink.table.api.internal.TableEnvironmentImpl.explainInternal(TableEnvironmentImpl.java:753) at org.apache.flink.table.api.internal.TableImpl.explain(TableImpl.java:482) at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method) at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:77) at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) at java.base/java.lang.reflect.Method.invoke(Method.java:569) at org.apache.flink.api.python.shaded.py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244) at org.apache.flink.api.python.shaded.py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:374) at org.apache.flink.api.python.shaded.py4j.Gateway.invoke(Gateway.java:282) at org.apache.flink.api.python.shaded.py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132) at org.apache.flink.api.python.shaded.py4j.commands.CallCommand.execute(CallCommand.java:79) at org.apache.flink.api.python.shaded.py4j.GatewayConnection.run(GatewayConnection.java:238) at java.base/java.lang.Thread.run(Thread.java:840)Caused by: java.lang.RuntimeException: Error occurred while applying rule FlinkExpandConversionRule at org.apache.calcite.plan.volcano.VolcanoRuleCall.transformTo(VolcanoRuleCall.java:157) at org.apache.calcite.plan.RelOptRuleCall.transformTo(RelOptRuleCall.java:273) at org.apache.calcite.plan.RelOptRuleCall.transformTo(RelOptRuleCall.java:288) at org.apache.flink.table.planner.plan.rules.physical.FlinkExpandConversionRule.satisfyTraitsBySelf(FlinkExpandConversionRule.scala:72) at org.apache.flink.table.planner.plan.rules.physical.FlinkExpandConversionRule.onMatch(FlinkExpandConversionRule.scala:52) at org.apache.calcite.plan.volcano.VolcanoRuleCall.onMatch(VolcanoRuleCall.java:223) ... 41 moreCaused by: java.lang.ArrayIndexOutOfBoundsException: Index 2 out of bounds for length 2 at org.apache.flink.calcite.shaded.com.google.common.collect.RegularImmutableList.get(RegularImmutableList.java:77) at org.apache.calcite.util.Util$TransformingList.get(Util.java:2794) at scala.collection.convert.Wrappers$JListWrapper.apply(Wrappers.scala:100) at org.apache.flink.table.planner.plan.utils.RelExplainUtil$.$anonfun$collationToString$1(RelExplainUtil.scala:83) at scala.collection.TraversableLike.$anonfun$map$1(TraversableLike.scala:286) at scala.collection.Iterator.foreach(Iterator.scala:943) at scala.collection.Iterator.foreach$(Iterator.scala:943) at scala.collection.AbstractIterator.foreach(Iterator.scala:1431) at scala.collection.IterableLike.foreach(IterableLike.scala:74) at scala.collection.IterableLike.foreach$(IterableLike.scala:73) at scala.collection.AbstractIterable.foreach(Iterable.scala:56) at scala.collection.TraversableLike.map(TraversableLike.scala:286) at scala.collection.TraversableLike.map$(TraversableLike.scala:279) at scala.collection.AbstractTraversable.map(Traversable.scala:108) at org.apache.flink.table.planner.plan.utils.RelExplainUtil$.collationToString(RelExplainUtil.scala:83) at org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalSort.explainTerms(BatchPhysicalSort.scala:61) at org.apache.calcite.rel.AbstractRelNode.getDigestItems(AbstractRelNode.java:414) at org.apache.calcite.rel.AbstractRelNode.deepHashCode(AbstractRelNode.java:396) at org.apache.calcite.rel.AbstractRelNode$InnerRelDigest.hashCode(AbstractRelNode.java:448) at java.base/java.util.HashMap.hash(HashMap.java:338) at java.base/java.util.HashMap.getNode(HashMap.java:568) at java.base/java.util.HashMap.get(HashMap.java:556) at org.apache.calcite.plan.volcano.VolcanoPlanner.registerImpl(VolcanoPlanner.java:1289) at org.apache.calcite.plan.volcano.VolcanoPlanner.register(VolcanoPlanner.java:598) at org.apache.calcite.plan.volcano.VolcanoPlanner.ensureRegistered(VolcanoPlanner.java:613) at org.apache.calcite.plan.volcano.VolcanoRuleCall.transformTo(VolcanoRuleCall.java:144) ... 46 more// code placeholder
{code}
However, if I just remove the columns that are irrelevant to the query, it works without issues i.e., just replace the data and schema with the following:
{code:java}
data = [
("Jimmy Allen", "3rd", -5.00),
("Jimmy Bullock", "Cedar Spruce", -5.00),
("Floyd Christian", "8th", -5.00),
("James Lachance", "6th", -5.00),
("James Lachance", "9th 12th", -5.00),
("Joaquin Washington", "Adams", -5.00),
("Michael Burton", "3rd", -5.00)
]
schema = [
"s_manager",
"s_street_name",
"s_gmt_offset"
]
{code}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)