[FLINK-11433] JOIN on a table having a column of type MULTISET gives a NPE

• Type: Bug
• Status: Reopened
• Priority: Not a Priority
• Resolution: Unresolved
• Affects Version/s: 1.7.0, 1.7.1
• Component/s: Table SQL / API
• Labels: stale-assigned
• Assignee: TANG Wen-hui
• Reporter: Elias Saalmann
• Created: 25/Jan/19 11:48
• Updated: 29/Jul/21 07:37

Description

I get an error (Error while applying rule FlinkLogicalJoinConverter) when performing a JOIN on a table having a column of type MULTISET (e.g. a COLLECT as aggregation of a GROUP BY), for instance:

SELECT a, d
FROM TableA JOIN (
  SELECT b, COLLECT(c) AS d
  FROM TableB
  GROUP BY b
) TableC ON a = b

Full stacktrace:

Exception in thread "main" java.lang.RuntimeException: Error while applying rule FlinkLogicalJoinConverter, args rel#71:LogicalJoin.NONE(left=rel#69:Subset#3.NONE,right=rel#70:Subset#4.NONE,condition==($2, $0),joinType=inner)
    at org.apache.calcite.plan.volcano.VolcanoRuleCall.onMatch(VolcanoRuleCall.java:236)
    at org.apache.calcite.plan.volcano.VolcanoPlanner.findBestExp(VolcanoPlanner.java:646)
    at org.apache.calcite.tools.Programs$RuleSetProgram.run(Programs.java:339)
    at org.apache.flink.table.api.TableEnvironment.runVolcanoPlanner(TableEnvironment.scala:373)
    at org.apache.flink.table.api.TableEnvironment.optimizeLogicalPlan(TableEnvironment.scala:292)
    at org.apache.flink.table.api.BatchTableEnvironment.optimize(BatchTableEnvironment.scala:455)
    at org.apache.flink.table.api.BatchTableEnvironment.translate(BatchTableEnvironment.scala:475)
    at org.apache.flink.table.api.java.BatchTableEnvironment.toDataSet(BatchTableEnvironment.scala:165)
    at org.myorg.quickstart.TableJob2.main(TableJob2.java:40)
Caused by: java.lang.RuntimeException: Error occurred while applying rule FlinkLogicalJoinConverter
    at org.apache.calcite.plan.volcano.VolcanoRuleCall.transformTo(VolcanoRuleCall.java:149)
    at org.apache.calcite.plan.RelOptRuleCall.transformTo(RelOptRuleCall.java:234)
    at org.apache.calcite.rel.convert.ConverterRule.onMatch(ConverterRule.java:141)
    at org.apache.calcite.plan.volcano.VolcanoRuleCall.onMatch(VolcanoRuleCall.java:212)
    ... 8 more
Caused by: java.lang.NullPointerException
    at org.apache.flink.table.plan.nodes.FlinkRelNode$class.estimateDataTypeSize(FlinkRelNode.scala:84)
    at org.apache.flink.table.plan.nodes.logical.FlinkLogicalJoinBase.estimateDataTypeSize(FlinkLogicalJoinBase.scala:29)
    at org.apache.flink.table.plan.nodes.FlinkRelNode$class.estimateDataTypeSize(FlinkRelNode.scala:104)
    at org.apache.flink.table.plan.nodes.logical.FlinkLogicalJoinBase.estimateDataTypeSize(FlinkLogicalJoinBase.scala:29)
    at org.apache.flink.table.plan.nodes.FlinkRelNode$$anonfun$estimateRowSize$2.apply(FlinkRelNode.scala:80)
    at org.apache.flink.table.plan.nodes.FlinkRelNode$$anonfun$estimateRowSize$2.apply(FlinkRelNode.scala:79)
    at scala.collection.IndexedSeqOptimized$class.foldl(IndexedSeqOptimized.scala:57)
    at scala.collection.IndexedSeqOptimized$class.foldLeft(IndexedSeqOptimized.scala:66)
    at scala.collection.mutable.ArrayBuffer.foldLeft(ArrayBuffer.scala:48)
    at org.apache.flink.table.plan.nodes.FlinkRelNode$class.estimateRowSize(FlinkRelNode.scala:79)
    at org.apache.flink.table.plan.nodes.logical.FlinkLogicalJoinBase.estimateRowSize(FlinkLogicalJoinBase.scala:29)
    at org.apache.flink.table.plan.nodes.logical.FlinkLogicalJoinBase.computeSelfCost(FlinkLogicalJoinBase.scala:48)
    at org.apache.calcite.rel.metadata.RelMdPercentageOriginalRows.getNonCumulativeCost(RelMdPercentageOriginalRows.java:162)
    at GeneratedMetadataHandler_NonCumulativeCost.getNonCumulativeCost_$(Unknown Source)
    at GeneratedMetadataHandler_NonCumulativeCost.getNonCumulativeCost(Unknown Source)
    at org.apache.calcite.rel.metadata.RelMetadataQuery.getNonCumulativeCost(RelMetadataQuery.java:301)
    at org.apache.calcite.plan.volcano.VolcanoPlanner.getCost(VolcanoPlanner.java:953)
    at org.apache.calcite.plan.volcano.RelSubset.propagateCostImprovements0(RelSubset.java:339)
    at org.apache.calcite.plan.volcano.RelSubset.propagateCostImprovements(RelSubset.java:322)
    at org.apache.calcite.plan.volcano.VolcanoPlanner.addRelToSet(VolcanoPlanner.java:1643)
    at org.apache.calcite.plan.volcano.VolcanoPlanner.registerImpl(VolcanoPlanner.java:1579)
    at org.apache.calcite.plan.volcano.VolcanoPlanner.register(VolcanoPlanner.java:859)
    at org.apache.calcite.plan.volcano.VolcanoPlanner.ensureRegistered(VolcanoPlanner.java:879)
    at org.apache.calcite.plan.volcano.VolcanoPlanner.ensureRegistered(VolcanoPlanner.java:1755)
    at org.apache.calcite.plan.volcano.VolcanoRuleCall.transformTo(VolcanoRuleCall.java:135)
    ... 11 more

Full example reproducing the error: GitHub

Issue Links

• is duplicated by: FLINK-11769 The estimateDataTypesSize method in FlinkRelNode causes NPE for Multiset