[
https://issues.apache.org/jira/browse/SPARK-59252?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Peter Toth updated SPARK-59252:
-------------------------------
Affects Version/s: 4.2.0
4.3.0
4.4.0
Description:
h2. Problem
{{DataSourceV2ScanExecBase.outputPartitioning}} is a {{def}} that reads
{{conf.v2BucketingEnabled}} off the live session conf, so it can answer
differently at execution time than it did at planning time. When it does, the
plan is already committed to the answer it gave the planner.
{code:scala}
override def outputPartitioning: physical.Partitioning = {
keyGroupedPartitioning match {
case Some(exprs) if conf.v2BucketingEnabled &&
KeyedPartitioning.supportsExpressions(exprs) &&
inputPartitions.nonEmpty &&
inputPartitions.forall(_.isInstanceOf[HasPartitionKey]) =>
...
KeyedPartitioning(exprs, partitionKeys)
case _ =>
super.outputPartitioning
}
}
{code}
h2. Reproduction, measured
Two tables bucketed the same way, joined on the bucket column. Force the plan,
then turn the config off, then run it:
{code:scala}
val df = sql("SELECT l.id FROM testcat.ns.l4 l JOIN testcat.ns.r4 r ON l.id =
r.id")
df.queryExecution.executedPlan // 0 shuffles, 2 GroupPartitionsExec
withSQLConf(SQLConf.V2_BUCKETING_ENABLED.key -> "false") {
df.collect()
}
{code}
{noformat}
java.lang.ClassCastException: class
org.apache.spark.sql.catalyst.plans.physical.UnknownPartitioning
cannot be cast to class org.apache.spark.sql.catalyst.expressions.Expression
{noformat}
The planner dropped both shuffles on the strength of the key-grouped layout and
inserted a {{GroupPartitionsExec}} on each side. Those nodes ask the child for
its partitioning again at execution, {{GroupPartitionsExec.grouping}} casts it
to {{Partitioning with Expression}}, and by then the scan reports
{{UnknownPartitioning}}.
*Measured on master, {{branch-4.3}} and {{branch-4.2}}*: the same exception on
all three, with the same plan shape (0 shuffles, 2 {{GroupPartitionsExec}}).
{{branch-4.1}} and older do not have {{GroupPartitionsExec}} at all, so the
crash site does not exist there.
h2. Fix
Make it a {{lazy val}}, so a node answers for its whole life what it answered
the planner.
The value is fixed for the instance otherwise: {{keyGroupedPartitioning}} is a
constructor field, every implementation of {{inputPartitions}} is already a
{{lazy val}} ({{BatchScanExec}}, {{MicroBatchScanExec}},
{{ContinuousScanExec}}, {{RealTimeStreamScanExec}}), and
{{BatchScanExec.filteredPartitions}} derives a new sequence rather than
replacing {{inputPartitions}}. {{FileSourceScanExec}}, the V1 twin, is already
{{override lazy val (outputPartitioning, outputOrdering)}} over a conf-derived
{{bucketedScan}} {{lazy val}}, so this aligns V2 with V1.
h2. It also removes repeated work
Not the reason for the ticket, but worth recording. As a {{def}} the
key-grouped arm sorts every partition key and hands them to
{{KeyedPartitioning.apply}}, which wraps each one and runs a {{distinct}}.
Instrumenting the body and running {{KeyGroupedPartitioningSuite}} counted
*33,051 executions as a {{def}} against 1,262 as a {{lazy val}}*, so 26 per
scan instance instead of one. Separately, in a microbenchmark over a two-column
key, one call measured 35 us at 100 partitions, 124 us at 1,000 and 1,515 us at
10,000.
h2. Follow-up, not in scope
Two other nodes recompute {{outputPartitioning}} the same way and are pure
performance, so they go separately, master only:
{{PartitioningPreservingUnaryExecNode}} (23,512 body executions, 3,663
memoized, and the source of 19,105 of the scan's reads) and
{{GroupPartitionsExec}} (8,821). {{supportsColumnar}} on this trait repeats
too, but it memoizes a connector call.
was:
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].
Issue Type: Bug (was: Improvement)
Summary: SPJ scan reports a different partitioning at execution
than it did at planning (was: Memoize outputPartitioning on the plan nodes
that recompute it per call)
> SPJ scan reports a different partitioning at execution than it did at planning
> ------------------------------------------------------------------------------
>
> Key: SPARK-59252
> URL: https://issues.apache.org/jira/browse/SPARK-59252
> Project: Spark
> Issue Type: Bug
> Components: SQL
> Affects Versions: 4.2.0, 4.3.0, 5.0.0, 4.4.0
> Reporter: Peter Toth
> Priority: Major
>
> h2. Problem
> {{DataSourceV2ScanExecBase.outputPartitioning}} is a {{def}} that reads
> {{conf.v2BucketingEnabled}} off the live session conf, so it can answer
> differently at execution time than it did at planning time. When it does, the
> plan is already committed to the answer it gave the planner.
> {code:scala}
> override def outputPartitioning: physical.Partitioning = {
> keyGroupedPartitioning match {
> case Some(exprs) if conf.v2BucketingEnabled &&
> KeyedPartitioning.supportsExpressions(exprs) &&
> inputPartitions.nonEmpty &&
> inputPartitions.forall(_.isInstanceOf[HasPartitionKey]) =>
> ...
> KeyedPartitioning(exprs, partitionKeys)
> case _ =>
> super.outputPartitioning
> }
> }
> {code}
> h2. Reproduction, measured
> Two tables bucketed the same way, joined on the bucket column. Force the
> plan, then turn the config off, then run it:
> {code:scala}
> val df = sql("SELECT l.id FROM testcat.ns.l4 l JOIN testcat.ns.r4 r ON l.id =
> r.id")
> df.queryExecution.executedPlan // 0 shuffles, 2 GroupPartitionsExec
> withSQLConf(SQLConf.V2_BUCKETING_ENABLED.key -> "false") {
> df.collect()
> }
> {code}
> {noformat}
> java.lang.ClassCastException: class
> org.apache.spark.sql.catalyst.plans.physical.UnknownPartitioning
> cannot be cast to class org.apache.spark.sql.catalyst.expressions.Expression
> {noformat}
> The planner dropped both shuffles on the strength of the key-grouped layout
> and inserted a {{GroupPartitionsExec}} on each side. Those nodes ask the
> child for its partitioning again at execution,
> {{GroupPartitionsExec.grouping}} casts it to {{Partitioning with
> Expression}}, and by then the scan reports {{UnknownPartitioning}}.
> *Measured on master, {{branch-4.3}} and {{branch-4.2}}*: the same exception
> on all three, with the same plan shape (0 shuffles, 2
> {{GroupPartitionsExec}}). {{branch-4.1}} and older do not have
> {{GroupPartitionsExec}} at all, so the crash site does not exist there.
> h2. Fix
> Make it a {{lazy val}}, so a node answers for its whole life what it answered
> the planner.
> The value is fixed for the instance otherwise: {{keyGroupedPartitioning}} is
> a constructor field, every implementation of {{inputPartitions}} is already a
> {{lazy val}} ({{BatchScanExec}}, {{MicroBatchScanExec}},
> {{ContinuousScanExec}}, {{RealTimeStreamScanExec}}), and
> {{BatchScanExec.filteredPartitions}} derives a new sequence rather than
> replacing {{inputPartitions}}. {{FileSourceScanExec}}, the V1 twin, is
> already {{override lazy val (outputPartitioning, outputOrdering)}} over a
> conf-derived {{bucketedScan}} {{lazy val}}, so this aligns V2 with V1.
> h2. It also removes repeated work
> Not the reason for the ticket, but worth recording. As a {{def}} the
> key-grouped arm sorts every partition key and hands them to
> {{KeyedPartitioning.apply}}, which wraps each one and runs a {{distinct}}.
> Instrumenting the body and running {{KeyGroupedPartitioningSuite}} counted
> *33,051 executions as a {{def}} against 1,262 as a {{lazy val}}*, so 26 per
> scan instance instead of one. Separately, in a microbenchmark over a
> two-column key, one call measured 35 us at 100 partitions, 124 us at 1,000
> and 1,515 us at 10,000.
> h2. Follow-up, not in scope
> Two other nodes recompute {{outputPartitioning}} the same way and are pure
> performance, so they go separately, master only:
> {{PartitioningPreservingUnaryExecNode}} (23,512 body executions, 3,663
> memoized, and the source of 19,105 of the scan's reads) and
> {{GroupPartitionsExec}} (8,821). {{supportsColumnar}} on this trait repeats
> too, but it memoizes a connector call.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
