Peter Toth created SPARK-60232:
----------------------------------
Summary: Fix wrong results when a storage-partitioned join is
planned over another storage-partitioned join
Key: SPARK-60232
URL: https://issues.apache.org/jira/browse/SPARK-60232
Project: Spark
Issue Type: Bug
Components: SQL
Affects Versions: 4.1.3, 4.0.4
Reporter: Peter Toth
On 4.0 and 4.1, a storage-partitioned join planned over another one can return
wrong results or fail. The outer join pushes its SPJ parameters into the scans
below, and it overwrites what the inner join pushed there instead of combining
with it. 4.2 and later regroup above the inner join with
{{GroupPartitionsExec}} (SPARK-55535, SPARK-55092), so they are not affected.
Three shapes, measured on branch-4.0 and branch-4.1, with AQE on and off:
1. Join key positions, with
{{spark.sql.sources.v2.bucketing.allowJoinKeysSubsetOfPartitionKeys.enabled=true}}
and {{spark.sql.requireAllClusterKeysForCoPartition=false}}:
{code:sql}
-- all tables: (store_id INT, dept_id INT, data STRING)
-- t4 PARTITIONED BY (store_id, dept_id): (1, 2, 'a'), (2, 3, 'b'), (3, 1, 'c')
-- t5, t6 PARTITIONED BY (dept_id): (0, 1, 'x1'), (0, 2, 'x2'), (0, 3, 'x3')
SELECT /*+ MERGE(t4, t5, t6) */ t4.store_id, t4.dept_id, t4.data, t5.data,
t6.data
FROM t4 JOIN t5 ON t4.dept_id = t5.dept_id
JOIN t6 ON t4.dept_id = t6.dept_id
{code}
It returns no rows instead of 3. The inner join pushes the positions {{[1]}}
into the {{t4}} scan. The outer join computes {{[0]}} against the already
projected {{[dept_id]}}, and {{populateCommonPartitionInfo}} writes it over
{{[1]}}. So {{t4}} is grouped by {{store_id}}. With partition keys of different
types, e.g. {{identity(date)}} and {{bucket(32, id)}} joined on {{id}}, the
same shape throws {{ClassCastException}}.
2. Reducers, with
{{spark.sql.sources.v2.bucketing.allowCompatibleTransforms.enabled=true}}:
{code:sql}
-- b4 PARTITIONED BY (bucket(4, id)): 4
-- b12 PARTITIONED BY (bucket(12, id)): 4
-- b8 PARTITIONED BY (bucket(8, id)): 1, 4
SELECT /*+ MERGE(b4, b12, b8) */ b4.id, b12.id, b8.id
FROM b4 JOIN b12 ON b4.id = b12.id
JOIN b8 ON b4.id = b8.id
{code}
It returns no rows instead of 1. The inner join reduces {{b12}} onto 4 buckets.
The outer join then pushes {{reducers = None}} into every scan on its left
side, so {{b12}} is no longer reduced.
3. Join key positions over common partition values, with
{{spark.sql.sources.v2.bucketing.shuffle.enabled=true}} and
{{allowJoinKeysSubsetOfPartitionKeys.enabled=true}}. {{items}} and {{items2}}
are partitioned by {{(id, name)}}, and {{purchases}} is not partitioned:
{code:sql}
SELECT /*+ MERGE(i, j, p) */ i.id, i.name, i.price, p.price
FROM items i JOIN items2 j ON i.id = j.id AND i.name = j.name
JOIN purchases p ON i.id = p.item_id
{code}
Planning fails:
* with {{PartitioningCollection requires all of its partitionings have the same
numPartitions}} when an id has two names;
* with {{ArrayIndexOutOfBoundsException}} otherwise.
The outer join pushes positions into scans that already carry the inner join's
common partition values. Those values keep both keys, while the scan's
expressions are cut down to {{id}}.
The same queries return the right rows on branch-4.2 and master.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
