[
https://issues.apache.org/jira/browse/SPARK-59187?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Peter Toth resolved SPARK-59187.
--------------------------------
Fix Version/s: 4.4.0
Resolution: Fixed
Issue resolved by pull request 58501
[https://github.com/apache/spark/pull/58501]
> SPJ across differently named struct partition keys: a join throws and a union
> drops rows
> ----------------------------------------------------------------------------------------
>
> Key: SPARK-59187
> URL: https://issues.apache.org/jira/browse/SPARK-59187
> Project: Spark
> Issue Type: Bug
> Components: SQL
> Affects Versions: 4.2.0, 4.3.0, 5.0.0, 4.4.0
> Reporter: Peter Toth
> Assignee: Peter Toth
> Priority: Major
> Labels: pull-request-available
> Fix For: 4.4.0
>
>
> A storage-partitioned join between two keyed sides whose struct partition
> keys have differently named fields fails at planning, and a union over such
> keys silently drops rows. Both are the same cause: a partition key row
> compares its data types exactly, so two rows of one value do not match when
> the columns they came from were named differently.
> h3. The failing join
> {code:sql}
> -- s1(id struct<a:int>, v string) partitioned by identity(id), keys
> named_struct('a',1), ('a',2)
> -- s2(k struct<b:int>, w string) partitioned by identity(k), keys
> named_struct('b',1), ('b',2)
> SELECT s1.v, s2.w FROM s1 JOIN s2 ON s1.id = s2.k
> {code}
> with {{spark.sql.sources.v2.bucketing.enabled}} and
> {{spark.sql.sources.v2.bucketing.pushPartValues.enabled}} on. The join is
> legal: {{BinaryComparison.sameType}} is {{DataType.equalsStructurally(_, _,
> ignoreNullability = true)}}, so no {{Cast}} is inserted, both sides are
> keyed, and the key values match. It throws:
> {noformat}
> [STORAGE_PARTITION_JOIN_INCOMPATIBLE_REDUCED_TYPES] Storage-partition join
> partition transforms
> produced incompatible reduced types, left reducers: [] returned: ["STRUCT<a:
> INT>"],
> right reducers: [] returned: ["STRUCT<b: INT>"]
> {noformat}
> for a join that reduced nothing. The path:
> # {{InternalRowComparableWrapper.equals}} compares its {{dataTypes}} before
> its values, so the two sides' key rows never match.
> # {{KeyedShuffleSpec.isCompatibleWith}} therefore answers false and the
> co-partitioned fast path is out.
> # {{EnsureRequirements}} falls into the push-common-partition-values branch,
> where a pair of attributes has no reducer, so each side's reduced types are
> its own and the comparison throws.
> h3. The union that drops rows
> {{KeyedPartitioning.concat}} puts the children's key rows in one list and
> asks whether any repeats. Rows of two namings never match, so a union of two
> children that hold one key under two namings reports unique keys when they
> are not. Nothing regroups it then, {{KeyedShuffleSpec.canCreatePartitioning}}
> accepts it, and the other side is shuffled straight onto those keys.
> {{KeyGroupedPartitioner}}'s map holds one partition per key, so the union
> partition holding the earlier copy of the repeated key receives no rows at
> all.
> {code:sql}
> -- items(id struct<a:int>, name string) partitioned by identity(id), keys
> ('a',1), ('a',2)
> -- purchases(item_id struct<b:int>, ...) unpartitioned
> -- t3(c struct<b:int>) partitioned by identity(c), key
> ('b',1)
> -- s4(k4 struct<b:int>, w string) unpartitioned, rows (('b',1),'x'),
> (('b',2),'y')
> SELECT u.k, s.w FROM (
> SELECT p.item_id AS k FROM purchases p LEFT JOIN items i ON p.item_id = i.id
> UNION ALL SELECT c AS k FROM t3
> ) u JOIN s4 s ON u.k = s.k4
> {code}
> with {{spark.sql.sources.v2.bucketing.shuffle.enabled}} and
> {{spark.sql.union.output.partitioning.enabled}} on, AQE off. Measured: 2 rows
> returned where 3 are correct. An inner join loses a row it should return,
> with no error.
> h3. Fix
> Erase the naming where key rows are built. {{InternalRowComparableWrapper}}'s
> factory builds every row at the given types with struct field names and every
> nullability erased, and nothing else touched: a collation, a decimal
> precision, a {{char}} length and a UDT all still tell two rows apart. Nothing
> that compares or hashes a row reads a field name, so this cannot move a row
> or change a sort order, and it can only make more keys compare equal, so a
> regroup is added rather than skipped.
> h3. Affected versions
> The reduced-types comparison came with SPARK-56046, released in 4.2.0, so the
> failing join affects 4.2.0 and later. The union path
> ({{KeyedPartitioning.concat}}) exists from 4.3.0.
> h3. Note
> This ticket was originally filed as an Improvement, to carry the key data
> types on {{KeyedPartitioning}} instead of sampling them from the first key
> row. That refactor is now SPARK-59285, and this ticket keeps the user-visible
> defects the work uncovered.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
