[
https://issues.apache.org/jira/browse/SPARK-60552?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Peter Toth updated SPARK-60552:
-------------------------------
Affects Version/s: 4.1.3
4.0.4
(was: 5.0.0)
> Fix wrong results when an ORDER BY over a storage-partitioned join pushes
> partition values below a shuffle
> ----------------------------------------------------------------------------------------------------------
>
> Key: SPARK-60552
> URL: https://issues.apache.org/jira/browse/SPARK-60552
> Project: Spark
> Issue Type: Bug
> Components: SQL
> Affects Versions: 4.1.3, 4.0.4
> Reporter: Peter Toth
> Priority: Major
>
> On 4.0 and 4.1, an {{ORDER BY}} over a storage-partitioned join that shuffles
> one side can drop that side's rows. With
> {{spark.sql.sources.v2.bucketing.sorting.enabled=true}},
> {{EnsureRequirements.ensureOrdering}} sorts the partition values and pushes
> them into every scan below the join. That includes the scan below the
> {{ShuffleExchangeExec}} of a one-side shuffle. When that scan is partitioned
> on other columns, it keeps only the partitions whose keys are among the
> pushed values.
> 4.2 and later group partitions with {{GroupPartitionsExec}} (SPARK-55535) and
> push nothing into the scans, so they are not affected.
> {code:sql}
> -- sa7 (dept INT) PARTITIONED BY (dept): 1, 2, 3
> -- sb7 (dept INT, region INT) PARTITIONED BY (region): (1, 10), (2, 20), (3,
> 30)
> SELECT /*+ MERGE(sa7, sb7) */ sa7.dept, sb7.region
> FROM sa7 LEFT JOIN sb7 ON sa7.dept = sb7.dept
> ORDER BY sa7.dept
> {code}
> With {{spark.sql.sources.v2.bucketing.shuffle.enabled=true}} and
> {{spark.sql.sources.v2.bucketing.sorting.enabled=true}}, measured on
> branch-4.0 (076b9d674fb) and branch-4.1 (627c1fb1758), AQE on and off:
> * The query returns {{[1,null]}}, {{[2,null]}}, {{[3,null]}} instead of
> {{[1,10]}}, {{[2,20]}}, {{[3,30]}}. {{sb7}} is shuffled into
> {{keygroupedpartitioning(dept, 3, [1], [2], [3])}}, and its scan gets the
> common partition values {{[1], [2], [3]}}, so it drops every {{region}}
> partition.
> * The same query with an inner {{JOIN}} returns no rows.
> * Without the {{ORDER BY}}, or with {{sorting.enabled=false}}, both return
> the right rows, and {{sb7}}'s scan gets no common partition values.
> branch-4.2 (aef8fb7426a) and master (972b507214f) return the right rows in
> every case.
> A nested storage-partitioned join over an inner one-side shuffle loses rows
> the same way, through the outer join's push (SPARK-60460). On top of
> apache/spark#59599 (SPARK-60232), which stops the push at a
> {{BroadcastExchangeLike}}, stopping it at a {{ShuffleExchangeLike}} too fixes
> both.
> Found while reviewing apache/spark#59599.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
