[jira] [Assigned] (SPARK-59187) SPJ across differently named struct partition keys: a join throws and a union drops rows

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

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

Peter Toth reassigned SPARK-59187:
----------------------------------

    Assignee: Peter Toth

> 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
>
> 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: issues-unsubscribe@spark.apache.org
For additional commands, e-mail: issues-help@spark.apache.org