Peter Toth created SPARK-59261:
----------------------------------
Summary: Memoize outputPartitioning on the projection and SPJ
grouping nodes
Key: SPARK-59261
URL: https://issues.apache.org/jira/browse/SPARK-59261
Project: Spark
Issue Type: Improvement
Components: SQL
Affects Versions: 5.0.0
Reporter: Peter Toth
h2. Problem
Two physical nodes recompute {{outputPartitioning}} on every call, and the
planner asks for it many times per node. Both do real work in the body and both
read only values fixed for the instance.
Counted by instrumenting the bodies and running
{{KeyGroupedPartitioningSuite}}, each measured with the previous change applied:
|| node || body executions as {{def}} || as {{lazy val}} ||
| {{PartitioningPreservingUnaryExecNode.outputPartitioning}}
({{sql/core/src/main/scala/org/apache/spark/sql/execution/AliasAwareOutputExpression.scala}})
| 23,512 | 3,663 |
| {{GroupPartitionsExec.outputPartitioning}}
({{sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExec.scala}})
| 8,821 | 1,326 |
h2. What the bodies cost
{{PartitioningPreservingUnaryExecNode}} flattens and partitions the child's
partitioning, allocates an {{AttributeSet}} and an {{ExpressionSet}} per key
position, cross-products the per-position alternatives through {{LazyList}}s,
and calls {{kps.head.project(positions)}}, which walks all N partition keys
building a {{mutable.HashMap}} whenever a position is dropped. It is the single
biggest caller of the scan's {{outputPartitioning}}: 19,105 of that method's
33,051 reads arrive through it, mostly as {{Project -> Filter -> scan}}.
{{GroupPartitionsExec}} rebuilds every {{KeyedPartitioning}} in the child's
partitioning through {{p.transform}}.
h2. Why memoizing is sound
The planner already treats {{outputPartitioning}} as a property of the node
rather than a question to re-ask. {{ValidateRequirements}} reads one child's
partitioning twice and compares specs built from the two reads.
{{EnsureRequirements}} reads it at five points and threads the results between
them, and at {{EnsureRequirements.scala:335}} it is the pruning predicate of a
{{multiTransformDownWithPruning}}, re-evaluated per generated alternative,
which is only sound if the answer is fixed.
Neither node freezes a config it does not already freeze.
{{PartitioningPreservingUnaryExecNode}} reads {{child.outputPartitioning}},
{{outputExpressions}} and {{aliasCandidateLimit}}, and that last one is already
a {{protected val}} read at construction.
{{GroupPartitionsExec.outputPartitioning}} reads {{child.outputPartitioning}}
and its own {{grouping}}, itself a {{lazy val}}, and no config at all.
{{BroadcastHashJoinExec}}, {{AQEShuffleReadExec}} and {{FileSourceScanExec}}
already override {{outputPartitioning}} as a {{lazy val}}.
{{PartitioningPreservingUnaryExecNode.outputPartitioning}} is {{final override
def}} today, so it becomes {{final override lazy val}}.
h2. Not in scope
{{GroupPartitionsExec.outputOrdering}} stays a {{def}}: it reads
{{conf.v2BucketingPreserveKeyOrderingOnCoalesceEnabled}}, so memoizing it would
freeze a config, which is a separate question from this one.
h2. Context
Split out of [SPARK-59252|https://issues.apache.org/jira/browse/SPARK-59252].
That one turned out to fix a crash and needs backporting, so the two pure
performance changes were separated from it to keep the bugfix minimal.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
