Yang Jie created SPARK-59122:
--------------------------------
Summary: UnionExec fused into whole-stage codegen can fail with
"key not found: numOutputRows"
Key: SPARK-59122
URL: https://issues.apache.org/jira/browse/SPARK-59122
Project: Spark
Issue Type: Bug
Components: SQL
Affects Versions: 5.0.0
Reporter: Yang Jie
A {{UnionExec}} that whole-stage codegen fused can fail at execution with:
{code}
java.util.NoSuchElementException: key not found: numOutputRows
at org.apache.spark.sql.execution.SparkPlan.longMetric(SparkPlan.scala:154)
at
org.apache.spark.sql.execution.CodegenSupport.metricTerm(WholeStageCodegenExec.scala:71)
at
org.apache.spark.sql.execution.UnionExec.doProduce(basicPhysicalOperators.scala:1148)
at
org.apache.spark.sql.execution.WholeStageCodegenExec.doCodeGen(WholeStageCodegenExec.scala:676)
{code}
Reproduced on master with default configuration:
{code}
spark.range(0, 200, 1, 4).selectExpr("id % 10 AS k", "id AS v")
.groupBy("k").agg(sum("v").as("s")).createOrReplaceTempView("v")
spark.catalog.cacheTable("v")
spark.sql("SELECT k, abs(s) AS s FROM v UNION ALL SELECT k, s FROM v").collect()
{code}
{{UnionExec}} derives three things from its children's {{outputPartitioning}}:
whether codegen
fusion applies, whether {{numOutputRows}} is registered, and which RDD
{{doExecute}} builds. That
input is not stable while the plan is prepared -- {{InMemoryTableScanExec}}
reads
{{cachedPlan.outputPartitioning}}, and the inner {{AdaptiveSparkPlanExec}}
answers
{{UnknownPartitioning}} until its final plan exists.
In the query above an expression on one branch only keeps a {{ProjectExec}}
from being collapsed
away, so the two children are shaped differently, {{comparePartitioning}}
rejects the pair, and the
union looks plain and is fused. {{CollapseCodegenStages.insertInputAdapter}}
then rebuilds the node
through {{withNewChildren}} and puts that copy inside the
{{WholeStageCodegenExec}} it just built,
so the copy evaluates the gate for the first time after the cache stages have
finalised. By then
both children report the same {{HashPartitioning}}, the gate answers
"partitioning-aware",
{{metrics}} comes back empty, and the generated code still increments the
metric.
Registering the metric unconditionally would turn the crash into a wrong
answer: a fused union
concatenates its children's partitions, so if it went on claiming their
{{HashPartitioning}} a
parent could satisfy a clustered distribution from an RDD that does not have it.
Affects the union codegen fusion added in SPARK-56482, which is present in
4.2.0, branch-4.3 and
master.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
