[
https://issues.apache.org/jira/browse/SPARK-60232?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Peter Toth updated SPARK-60232:
-------------------------------
Affects Version/s: 4.0.5
4.1.4
(was: 4.1.3)
(was: 4.0.4)
> 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.0.5, 4.1.4
> Reporter: Peter Toth
> Assignee: Ammar Chalifah
> Priority: Major
> Labels: correctness, pull-request-available
> Fix For: 4.1.5
>
>
> 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
(v10.3.26#10030026)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
