[
https://issues.apache.org/jira/browse/SPARK-59121?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18109680#comment-18109680
]
Peter Toth edited comment on SPARK-59121 at 8/30/26 1:56 PM:
-------------------------------------------------------------
I'm working on it.
was (Author: ptoth):
I am working on it.
> 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
> Priority: Major
>
> 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: [email protected]
For additional commands, e-mail: [email protected]
