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/11/07 16:09:00 UTC
[jira] [Updated] (FLINK-38397) Stale Sort Trait causes BatchPhysicalSort.explainTerms to read a non-existent index
[ https://issues.apache.org/jira/browse/FLINK-38397?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Ahmad Humayun updated FLINK-38397:
----------------------------------
Summary: Stale Sort Trait causes BatchPhysicalSort.explainTerms to read a non-existent index (was: Runtime Exception in Planner)
> Stale Sort Trait causes BatchPhysicalSort.explainTerms to read a non-existent index
> -----------------------------------------------------------------------------------
>
> 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: python 3.10.12
> apache-flink 2.0.0
> Reporter: Ahmad Humayun
> Priority: Major
>
>
> 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)