Peter Toth created SPARK-60460:
----------------------------------
Summary: Fix storage-partitioned join failures over a nested join
that shuffles one side
Key: SPARK-60460
URL: https://issues.apache.org/jira/browse/SPARK-60460
Project: Spark
Issue Type: Bug
Components: SQL
Affects Versions: 5.0.0
Reporter: Peter Toth
On 4.0 and 4.1, a storage-partitioned join planned over another one fails when
the inner join shuffles one of its sides
(`spark.sql.sources.v2.bucketing.shuffle.enabled=true`). The outer join pushes
its merged partition values into the inner join's partitioned scan, so that
scan now has more partitions. The exchange that shuffled the inner join's other
side keeps the partition count it was planned with.
4.2 and later regroup above the inner join with `GroupPartitionsExec`
(SPARK-55535), so they are not affected.
{code:sql}
-- pa and pc PARTITIONED BY (id), pd not partitioned
-- pa: 1, 2, 3 pd: 1, 2, 3 pc: 1, 2, 3, 4, 5
SELECT /*+ MERGE(pa, pd, pc) */ pa.id, pd.id, pc.id
FROM pa JOIN pd ON pa.id = pd.id
JOIN pc ON pa.id = pc.id
{code}
With `spark.sql.sources.v2.bucketing.shuffle.enabled=true`, measured on
branch-4.0 (2be20e25823) and branch-4.1 (9762e41a34f):
* AQE off: {{IllegalArgumentException: Can't zip RDDs with unequal numbers of
partitions: List(5, 3)}}
* AQE on: {{requirement failed: PartitioningCollection requires all of its
partitionings have the same numPartitions}}
branch-4.2 and master return the 3 rows with one shuffle.
Stopping the push at the exchange does not fix it, since the exchange still has
the old partition count. A fix has to either re-plan that exchange onto the
merged values, or refuse the SPJ for such a child.
Found while reviewing apache/spark#59599 (SPARK-60232). That PR handles the
broadcast build-side half of the same push.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
