Peter Toth created SPARK-59252:
----------------------------------
Summary: Memoize DataSourceV2ScanExecBase.outputPartitioning
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
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.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
