Peter Toth created SPARK-60552:
----------------------------------
Summary: 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: 5.0.0
Reporter: Peter Toth
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]
