[
https://issues.apache.org/jira/browse/SPARK-59252?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Peter Toth updated SPARK-59252:
-------------------------------
Description:
h2. Problem
Three physical nodes recompute {{outputPartitioning}} on every call, and the
planner asks for it many times per node. All three do real work in the body,
and all three read only values that are fixed for the instance.
Counted by instrumenting the bodies and running {{KeyGroupedPartitioningSuite}}
(133 tests), each measured with the previous one already memoized:
|| node || body executions as {{def}} || as {{lazy val}} ||
| {{PartitioningPreservingUnaryExecNode.outputPartitioning}} | 23,512 | 3,663 |
| {{DataSourceV2ScanExecBase.outputPartitioning}} | 33,051 | 1,262 |
| {{GroupPartitionsExec.outputPartitioning}} | 8,821 | 1,326 |
They are one chain, which is why they belong together: 19,105 of the scan's
33,051 reads arrive through {{PartitioningPreservingUnaryExecNode}}, mostly as
{{Project -> Filter -> scan}}. Memoizing only the scan leaves the projection
node rebuilding its answer 23,512 times; memoizing the projection node takes
the scan's reads down to 15,888 on its own.
h2. What each body costs
{{DataSourceV2ScanExecBase}}
({{sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2ScanExecBase.scala}})
sorts every partition key and hands them to {{KeyedPartitioning.apply}}, which
wraps each one and runs a {{distinct}} over them. In a microbenchmark over a
two-column key that measured 35 us at 100 partitions, 124 us at 1,000 and 1,515
us at 10,000.
{{PartitioningPreservingUnaryExecNode}}
({{sql/core/src/main/scala/org/apache/spark/sql/execution/AliasAwareOutputExpression.scala}})
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 keys building a
{{mutable.HashMap}} whenever a position is dropped. More allocation per call
than the scan's sort, at 18x the frequency.
{{GroupPartitionsExec}}
({{sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExec.scala}})
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.
{{FileSourceScanExec}}, the V1 twin of the scan node, is already {{override
lazy val (outputPartitioning, outputOrdering)}} over a conf-derived
{{bucketedScan}} {{lazy val}}. {{BroadcastHashJoinExec}} and
{{AQEShuffleReadExec}} do the same for {{outputPartitioning}}.
h2. The one behaviour change to state
The scan node's body reads {{conf.v2BucketingEnabled}}, and {{conf}} is
{{session.sessionState.conf}}, so memoizing reads it once per node instead of
once per call. Where that is visible is a cached plan: {{CacheManager}} holds
one plan across conf changes and sessions and {{InMemoryTableScanExec}} reads
{{cachedPlan.outputPartitioning}}, so a cached V2 scan keeps the bucketing
setting it was first materialised under. {{FileSourceScanExec}} already behaves
that way.
{{PartitioningPreservingUnaryExecNode}} adds no new freezing of its own: its
{{aliasCandidateLimit}} is already a {{val}} read at construction.
h2. Not in scope
{{DataSourceV2ScanExecBase.outputOrdering}} stays a {{def}}. Making it a {{lazy
val}} fails {{KeyGroupedPartitioningSuite}}'s "SPARK-56241: scan with
KeyedPartitioning reports key-derived outputOrdering", which flips
{{spark.sql.sources.v2.bucketing.partitionKeyOrdering.enabled}} and re-reads it
off a plan it already built.
{{supportsColumnar}} on the same trait ran 5,908 times, each walking
{{inputPartitions}} up to three times, but it memoizes a connector call
({{scan.columnarSupportMode()}}), which is a different question from these
three.
h2. Context
Found while measuring
[SPARK-59249|https://issues.apache.org/jira/browse/SPARK-59249].
was:
h2. Problem
{{DataSourceV2ScanExecBase.outputPartitioning}}
({{sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2ScanExecBase.scala}})
is a {{def}}, and on the storage-partitioned join path it does real work on
every call:
{code:scala}
override def outputPartitioning: physical.Partitioning = {
keyGroupedPartitioning match {
case Some(exprs) if conf.v2BucketingEnabled &&
KeyedPartitioning.supportsExpressions(exprs) &&
inputPartitions.nonEmpty &&
inputPartitions.forall(_.isInstanceOf[HasPartitionKey]) =>
val rowOrdering =
KeyedPartitioning.groupedKeyRowOrdering(exprs.map(_.dataType))
val partitionKeys =
inputPartitions.map(_.asInstanceOf[HasPartitionKey].partitionKey()).sorted(rowOrdering)
KeyedPartitioning(exprs, partitionKeys)
case _ =>
super.outputPartitioning
}
}
{code}
That is a sort of every partition key plus a {{KeyedPartitioning.apply}}, which
wraps every key and runs a {{distinct}} to decide {{isGrouped}}.
h2. How often
Instrumented run of {{KeyGroupedPartitioningSuite}}: *31,922 calls across 1,167
scan instances, a mean of 27 per scan and a maximum of 63.* {{outputOrdering}}
on the same trait calls it as well.
Measured per call for a two-column key: 9.8 us sort + 25.1 us apply at 100
partitions, 45.7 + 78.0 at 1,000, and 712 + 803 at 10,000.
h2. Proposal
Make it a {{lazy val}}. Measured with that change, the call count drops from
*31,922 to 1,167*, exactly one per scan, and all 133
{{KeyGroupedPartitioningSuite}} tests pass.
The value is stable per instance: {{keyGroupedPartitioning}} is a constructor
field, and every implementor of {{inputPartitions}} is already a {{lazy val}}
({{BatchScanExec}}, {{MicroBatchScanExec}}, {{ContinuousScanExec}},
{{RealTimeStreamScanExec}}).
{{BroadcastHashJoinExec}} and {{AQEShuffleReadExec}} already override
{{outputPartitioning}} as a {{lazy val}}, so the shape is established.
h2. The one thing to settle first
{{conf}} is {{session.sessionState.conf}}, so a {{lazy val}} freezes
{{v2BucketingEnabled}} at the first read rather than re-reading it per call.
That cuts both ways and is worth stating rather than assuming. A {{def}} that
reads a mutable conf can also return *different* partitionings across calls
within one planning run, and the planner treats {{outputPartitioning}} as a
property of the node: {{EnsureRequirements}} and {{ValidateRequirements}} ask
for it repeatedly and compare the answers. So memoizing may remove an
inconsistency as well as a cost. Any test that flips the conf between calls on
one plan instance needs checking.
h2. Context
Found while measuring
[SPARK-59249|https://issues.apache.org/jira/browse/SPARK-59249], which removed
one ~100 us term from this method. At 10,000 partitions that term is about 6%
of the call, and the other 94% still runs 27 more times, so this is the larger
remaining win at the site.
Summary: Memoize outputPartitioning on the plan nodes that recompute it
per call (was: Memoize DataSourceV2ScanExecBase.outputPartitioning)
> Memoize outputPartitioning on the plan nodes that recompute it per call
> -----------------------------------------------------------------------
>
> Key: SPARK-59252
> URL: https://issues.apache.org/jira/browse/SPARK-59252
> Project: Spark
> Issue Type: Improvement
> Components: SQL
> Affects Versions: 5.0.0
> Reporter: Peter Toth
> Priority: Major
>
> h2. Problem
> Three physical nodes recompute {{outputPartitioning}} on every call, and the
> planner asks for it many times per node. All three do real work in the body,
> and all three read only values that are fixed for the instance.
> Counted by instrumenting the bodies and running
> {{KeyGroupedPartitioningSuite}} (133 tests), each measured with the previous
> one already memoized:
> || node || body executions as {{def}} || as {{lazy val}} ||
> | {{PartitioningPreservingUnaryExecNode.outputPartitioning}} | 23,512 | 3,663
> |
> | {{DataSourceV2ScanExecBase.outputPartitioning}} | 33,051 | 1,262 |
> | {{GroupPartitionsExec.outputPartitioning}} | 8,821 | 1,326 |
> They are one chain, which is why they belong together: 19,105 of the scan's
> 33,051 reads arrive through {{PartitioningPreservingUnaryExecNode}}, mostly
> as {{Project -> Filter -> scan}}. Memoizing only the scan leaves the
> projection node rebuilding its answer 23,512 times; memoizing the projection
> node takes the scan's reads down to 15,888 on its own.
> h2. What each body costs
> {{DataSourceV2ScanExecBase}}
> ({{sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2ScanExecBase.scala}})
> sorts every partition key and hands them to {{KeyedPartitioning.apply}},
> which wraps each one and runs a {{distinct}} over them. In a microbenchmark
> over a two-column key that measured 35 us at 100 partitions, 124 us at 1,000
> and 1,515 us at 10,000.
> {{PartitioningPreservingUnaryExecNode}}
> ({{sql/core/src/main/scala/org/apache/spark/sql/execution/AliasAwareOutputExpression.scala}})
> 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 keys building a
> {{mutable.HashMap}} whenever a position is dropped. More allocation per call
> than the scan's sort, at 18x the frequency.
> {{GroupPartitionsExec}}
> ({{sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExec.scala}})
> 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.
> {{FileSourceScanExec}}, the V1 twin of the scan node, is already {{override
> lazy val (outputPartitioning, outputOrdering)}} over a conf-derived
> {{bucketedScan}} {{lazy val}}. {{BroadcastHashJoinExec}} and
> {{AQEShuffleReadExec}} do the same for {{outputPartitioning}}.
> h2. The one behaviour change to state
> The scan node's body reads {{conf.v2BucketingEnabled}}, and {{conf}} is
> {{session.sessionState.conf}}, so memoizing reads it once per node instead of
> once per call. Where that is visible is a cached plan: {{CacheManager}} holds
> one plan across conf changes and sessions and {{InMemoryTableScanExec}} reads
> {{cachedPlan.outputPartitioning}}, so a cached V2 scan keeps the bucketing
> setting it was first materialised under. {{FileSourceScanExec}} already
> behaves that way.
> {{PartitioningPreservingUnaryExecNode}} adds no new freezing of its own: its
> {{aliasCandidateLimit}} is already a {{val}} read at construction.
> h2. Not in scope
> {{DataSourceV2ScanExecBase.outputOrdering}} stays a {{def}}. Making it a
> {{lazy val}} fails {{KeyGroupedPartitioningSuite}}'s "SPARK-56241: scan with
> KeyedPartitioning reports key-derived outputOrdering", which flips
> {{spark.sql.sources.v2.bucketing.partitionKeyOrdering.enabled}} and re-reads
> it off a plan it already built.
> {{supportsColumnar}} on the same trait ran 5,908 times, each walking
> {{inputPartitions}} up to three times, but it memoizes a connector call
> ({{scan.columnarSupportMode()}}), which is a different question from these
> three.
> h2. Context
> Found while measuring
> [SPARK-59249|https://issues.apache.org/jira/browse/SPARK-59249].
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
