Peter Toth created SPARK-59121:
----------------------------------
Summary: 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: 5.0.0
Reporter: Peter Toth
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]
