[
https://issues.apache.org/jira/browse/SPARK-60460?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18136631#comment-18136631
]
Peter Toth commented on SPARK-60460:
------------------------------------
A third route, through a reducer, measured on branch-4.0 (5192f285fdf),
branch-4.1 (d4521911899) and the head of apache/spark#59599 (785f3a4e14f), AQE
on and off. It needs {{spark.sql.sources.v2.bucketing.shuffle.enabled=true}}
and {{spark.sql.sources.v2.bucketing.allowCompatibleTransforms.enabled=true}}.
{code:sql}
-- l12 (id BIGINT) PARTITIONED BY (bucket(12, id))
-- l4 (id BIGINT) PARTITIONED BY (bucket(4, id))
-- ld (id BIGINT), not partitioned
-- each holds the ids 0 to 23
SELECT /*+ MERGE(l12, ld, l4) */ l12.id, ld.id, l4.id
FROM l12 JOIN ld ON l12.id = ld.id
JOIN l4 ON l12.id = l4.id
{code}
* AQE off: {{IllegalArgumentException: Can't zip RDDs with unequal numbers of
partitions: List(4, 12)}}
* AQE on: {{requirement failed: PartitioningCollection requires all of its
partitionings have the same numPartitions}}
The inner join shuffles {{ld}} into the 12 buckets of {{l12}}. The outer join
then pushes a {{bucket(12)}} to {{bucket(4)}} reducer into {{l12}}'s scan, so
the scan has 4 partitions, while the exchange keeps 12. The inner join alone
returns the 24 rows with one shuffle. This is the reducer case that the fix in
the earlier comment has to refuse in {{checkKeyGroupCompatible}}.
branch-4.2 (3a8d3c3c072) and master (319cc71bf95) return the 24 rows with one
shuffle.
The ids are BIGINT because with INT ids the one-side shuffle of the inner join
already fails with a {{ClassCastException}}, see SPARK-60027.
> 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: 4.1.3, 4.0.4
> Reporter: Peter Toth
> Priority: Major
>
> 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]
