[
https://issues.apache.org/jira/browse/SPARK-60260?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Peter Toth resolved SPARK-60260.
--------------------------------
Fix Version/s: 5.0.0
Resolution: Fixed
Issue resolved by pull request 59789
[https://github.com/apache/spark/pull/59789]
> Fix wrong results when spark.sql.sources.v2.bucketing.enabled changes after a
> storage-partitioned join is planned
> -----------------------------------------------------------------------------------------------------------------
>
> Key: SPARK-60260
> URL: https://issues.apache.org/jira/browse/SPARK-60260
> Project: Spark
> Issue Type: Bug
> Components: SQL
> Affects Versions: 4.3.0, 4.2.1, 4.0.5, 4.1.4
> Reporter: Peter Toth
> Assignee: Naveen Kumar Puppala
> Priority: Major
> Labels: pull-request-available
> Fix For: 5.0.0
>
>
> A storage-partitioned join can return wrong results, or fail, when
> {{spark.sql.sources.v2.bucketing.enabled}} is turned off between planning and
> executing a query. It needs dynamic partition pruning on the keyed scan of a
> join with no {{GroupPartitionsExec}} above it, e.g. two sources that already
> report one split per partition key.
> Measured with
> {{spark.sql.optimizer.dynamicPartitionPruning.reuseBroadcastOnly=false}} and
> no broadcast, on {{items}} and {{purchases}} identity-partitioned on {{id}} /
> {{item_id}}, 7 keys, one split each:
> {code:scala}
> val df = sql("SELECT i.id, i.name, p.price FROM items i JOIN purchases p ON
> i.id = p.item_id WHERE p.price > 0.0")
> df.queryExecution.executedPlan // planned with the config on
> spark.conf.set("spark.sql.sources.v2.bucketing.enabled", "false")
> df.collect()
> {code}
> * On master and branch-4.2, with AQE on, it returns 2 rows instead of 7. When
> the runtime filter prunes splits ({{p.price > 5.0}}), it fails with {{Can't
> zip RDDs with unequal numbers of partitions: List(4, 7)}}. With AQE off it is
> correct.
> * On branch-4.1 and branch-4.0 the same happens with AQE on, and with AQE off
> it returns no rows.
> The cause on master: {{PlanAdaptiveDynamicPruningFilters}} makes a new
> {{BatchScanExec}} when the stage is created, and the new instance reads the
> config again. It then reports no {{KeyedPartitioning}}, so
> {{PushDownUtils.replanWithRuntimeFilters}} returns the splits in the
> connector's order and without the empty partitions for pruned keys. The other
> side keeps the sorted and padded layout the join was planned with, so the
> sort-merge join pairs the wrong partitions.
> This is the third bug of this kind in storage-partitioned join: a config read
> again after planning changes what a planned node does (SPARK-59279,
> SPARK-60044). The fixes so far freeze one config in one node. It may be worth
> looking at it more generally. A node that is rebuilt or executed after
> planning reads the session config as it is then, so the plan that runs is not
> always the plan that was planned.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
