[
https://issues.apache.org/jira/browse/SPARK-60460?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18132917#comment-18132917
]
Peter Toth commented on SPARK-60460:
------------------------------------
Two more shapes from the same push, measured on branch-4.0 (076b9d674fb),
branch-4.1 (627c1fb1758) and the head of apache/spark#59599 (c2bdb4df323), AQE
on and off. Both need {{spark.sql.sources.v2.bucketing.shuffle.enabled=true}},
{{spark.sql.sources.v2.bucketing.allowJoinKeysSubsetOfPartitionKeys.enabled=true}}
and {{spark.sql.requireAllClusterKeysForCoPartition=false}}.
*Reordered values return wrong rows, with no error.*
{code:sql}
-- sa (store INT, dept INT) PARTITIONED BY (store, dept): (1, 20), (2, 10)
-- sb (store INT, dept INT), not partitioned: (0, 10), (0, 20)
-- sc (store INT, dept INT) PARTITIONED BY (dept): (0, 10), (0, 20)
SELECT /*+ MERGE(sa, sb, sc) */ sa.store, sa.dept, sb.dept, sc.dept
FROM sa JOIN sb ON sa.dept = sb.dept
JOIN sc ON sa.dept = sc.dept
{code}
It returns no rows instead of 2. The inner join shuffles {{sb}} into {{[20],
[10]}}, the order in which {{sa}}'s keys appear once projected onto {{dept}}.
The outer join then pushes the sorted {{[10], [20]}} into {{sa}}'s scan, while
the exchange keeps the old order.
*Narrowed positions fail at planning.*
{code:sql}
-- da (store INT, dept INT) PARTITIONED BY (store, dept): (1, 20), (3, 20), (2,
10)
-- db (store INT, dept INT), not partitioned: the same rows
-- de (k INT, dept INT) PARTITIONED BY (k, dept): (1, 20), (2, 10)
SELECT /*+ MERGE(da, db, de) */ da.store, da.dept, db.store, de.k
FROM da JOIN db ON da.store = db.store AND da.dept = db.dept
JOIN de ON da.dept = de.dept
{code}
It fails with {{requirement failed: PartitioningCollection requires all of its
partitionings have the same numPartitions}}. The push narrows {{da}} to
{{dept}}, so it has 2 partitions. {{db}}'s exchange keeps 3, one per {{(store,
dept)}}.
master (972b507214f) returns the right rows for both. branch-4.2 (aef8fb7426a)
does for the second one. It returns no rows for the first one with AQE on, but
for another reason: there the inner join alone already loses the rows, see
SPARK-58988.
Refusing the SPJ for such a child, as the description says, has to happen in
two places:
* {{checkKeyGroupCompatible}}, when the push changes what the exchange below
has routed: other values, another order, or reducers.
* {{matchedIndexes}} in the one-side shuffle path, when the child's join key
positions narrow its layout. Otherwise the second query fails there once
{{checkKeyGroupCompatible}} refuses.
A prototype on top of #59599 that refuses in both places, for any child with a
key-grouped exchange below, returns the right rows for a query of the ticket's
shape and for both queries above. It is too coarse, though, since it also
shuffles such a child when the push changes nothing.
> 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]
