[jira] [Created] (SPARK-60232) Fix wrong results when a storage-partitioned join is planned over another storage-partitioned join

["Peter Toth (Jira)" <ji...@apache.org>]

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: issues-unsubscribe@spark.apache.org
For additional commands, e-mail: issues-help@spark.apache.org