[
https://issues.apache.org/jira/browse/SPARK-59261?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18111676#comment-18111676
]
Peter Toth commented on SPARK-59261:
------------------------------------
I'm working on this.
> 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
> Priority: Major
>
> 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]
