[jira] [Assigned] (SPARK-59121) Fix wrong results when a storage-partitioned join reduces the partition keys of both sides

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

     [ https://issues.apache.org/jira/browse/SPARK-59121?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Peter Toth reassigned SPARK-59121:
----------------------------------

    Assignee: Peter Toth

> Fix wrong results when a storage-partitioned join reduces the partition keys of both sides
> ------------------------------------------------------------------------------------------
>
>                 Key: SPARK-59121
>                 URL: https://issues.apache.org/jira/browse/SPARK-59121
>             Project: Spark
>          Issue Type: Bug
>          Components: SQL
>    Affects Versions: 4.0.0, 4.2.0, 5.0.0
>            Reporter: Peter Toth
>            Assignee: Peter Toth
>            Priority: Major
>              Labels: pull-request-available
>
> With {{spark.sql.sources.v2.bucketing.allowCompatibleTransforms.enabled}} a storage-partitioned join can reduce *both* sides' partition keys onto a common key space. SPARK-47094 does that for {{bucket(12)}} joined to {{bucket(8)}}, where {{BucketFunction.reducer}} hands both sides {{BucketReducer(gcd = 4)}}, so the shared space is a third one, {{bucket(4, x)}}, and neither side's transform describes it.
> The reduce itself is self-contained. {{GroupPartitionsExec}} is handed the reducers and types the keys from {{Reducer.resultType()}}, never from the reported expressions. The bug is in what the reduced partitioning advertises upwards: it keeps reporting the partition expressions it was built from, and every consumer above it believes them. Only certain questions break. Clustering stays sound, because it only asks which attributes the keys are a function of.
> Five consequences, all measured, all needing a consumer above the reduced join:
> * A second join reduces already reduced keys, deriving a reducer from the stale transform. {{bucket(12)}} / {{bucket(8)}} / {{bucket(6)}} chained returns 4 of 12 rows, no shuffle.
> * A union merges a reduced partitioning with an unreduced one. {{(bucket12 JOIN bucket8) UNION ALL bucket12}} grouped by the key returns every id twice, because the merge claims {{bucket(12, id)}} for keys that are {{id % 4}}.
> * Two independently reduced sides compare as compatible on their stale transforms. {{bucket12 JOIN bucket8}} reduced onto {{id % 4}} joined with {{bucket12 JOIN bucket18}} reduced onto {{id % 6}} returns 0,1,2,3 instead of 0,1,2,3,6,7,8,9.
> * Another child is shuffled onto the reduced keys. {{ShuffleExchangeExec}} places its rows by evaluating the stale expressions, so a key that is not in the map goes to {{hashCode % numPartitions}} and rows are lost. With {{bucket(12)}} and {{bucket(8)}} reducing onto {{bucket(4)}} plus an unpartitioned third table, 8 of 12 rows come back.
> * {{EnsureRequirements}}' {{OrderedDistribution}} arm orders the key rows by the partition attributes, which for reduced keys is the wrong key space. {{identity(ts)}} joined to {{years(ts)}} with {{ORDER BY}} on the partition key throws {{ClassCastException}} at planning.
> The type half of this is being fixed separately in SPARK-59120: reading a key row at the types it was written with removes two {{ClassCastException}}s and refuses a reduced partitioning as a shuffle target. That refusal is an explicit proxy. A reduction that keeps the type passes it, which is the fourth case above, so the type fix does not close it.
> The remedy is for the partitioning to carry, per partition expression position, the reduction target its keys were mapped onto, and to refuse the four questions above when the targets do not pair up. SPARK-59045 covers the case where one side reduces onto the other side's transform, since an exact expression exists there and can be reported. This ticket is the case where both sides reduce and no expression exists.
> Affects 4.0.0 onwards, where the feature and the reporting of reduced keys under the original transforms arrived together (SPARK-47094). On 4.0 and 4.1 the symptoms are wrong results; the {{ClassCastException}}s start in 4.2.0 with the {{KeyedPartitioning}} refactor.



--
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