[
https://issues.apache.org/jira/browse/SPARK-59252?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ASF GitHub Bot updated SPARK-59252:
-----------------------------------
Labels: pull-request-available (was: )
> 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
> Labels: pull-request-available
>
> 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]
