peter-toth opened a new pull request, #58600:
URL: https://github.com/apache/spark/pull/58600
This is the `branch-4.2` backport of #58501 (master `c6e2d26ac66`). The
cherry-pick was not clean and the change had to be trimmed, so it is a separate
PR.
**What was tailored.**
- **The union defect does not exist on this branch.**
`KeyedPartitioning.concat` arrived in 4.3, and 4.2's `UnionExec` does not merge
keyed partitionings at all, so there is nothing to miscount. Its test is
dropped and the section describing it is marked below.
- `branch-4.2` does not have SPARK-59123's per-key loops in
`KeyedPartitioning.projectKeys` and `reduceKeys`, so those two methods keep
their 4.2 shape and only take the type list off the wrapper factory.
- `keyedPartitioningsOf` and `PartitioningCollection.flatten` do not exist
here, so the test helper is added to the suite with its own local flatten.
**The remaining defect is live here**, verified rather than assumed: with
the erasure ablated on this branch, both `KeyGroupedPartitioningSuite` tests
fail.
### What changes were proposed in this pull request?
`InternalRowComparableWrapper`'s factory builds every partition key row at
`comparableTypes`, the given types with struct field names, every nullability
and a field's metadata erased and nothing else touched. It is `asNullable` for
the nullability half plus a positional `StructType` rename for the naming half.
Three places that report a key list's types alongside it erase them the same
way, so a partitioning's key types are always the types its keys are compared
at: `KeyedPartitioning.keyDataTypes`, in its no-key fallback, and
`KeyedPartitioning.projectKeys` and `reduceKeys`, the latter over types that
come from a connector's `Reducer`.
The erasure is idempotent and keeps the list it was given, so a caller that
hands its own types in and the result back to another factory gets one
instance, and `equals` keeps its reference fast path.
### Why are the changes needed?
`InternalRowComparableWrapper.equals` compares its `dataTypes` before its
values, so two key rows of one value never matched when the columns they came
from were named differently. A storage-partitioned join is about the opposite:
a key value belongs where its value says, not where its column name says. Two
consequences, both measured on master.
**A join between two keyed sides whose struct key fields are named
differently throws.** `identity` carries the column's own struct type into the
key type, and the analyzer accepts an equi-join across `struct<a:int>` and
`struct<b:int>` -- `BinaryComparison.sameType` is
`DataType.equalsStructurally(_, _, ignoreNullability = true)`, so no `Cast` is
inserted. Both sides are keyed and the values match, but the key rows never
matched, so the co-partitioned fast path in `KeyedShuffleSpec.isCompatibleWith`
was out, and a pair of attributes has no reducer, so the reduced-types check
threw `STORAGE_PARTITION_JOIN_INCOMPATIBLE_REDUCED_TYPES` for a join that
reduced nothing:
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 -- threw, now runs
with no shuffle
**A union over a key that two children hold under two namings silently drops
rows. NOT APPLICABLE TO 4.2**, kept here for the record, since
`KeyedPartitioning.concat` does not exist on this branch.
`KeyedPartitioning.concat` puts the children's key rows in one list and asks
whether any repeats. Rows of two namings never matched, so the merged
partitioning reported unique keys when they were not. Nothing regroups it then,
so `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:
t1(k1 struct<a:int>) partitioned by identity(k1), keys
('a',1), ('a',2)
t2(k2 struct<b:int>) partitioned by identity(k2), 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 k1 AS k FROM t1 UNION ALL SELECT k2 AS k FROM t2
) u JOIN s4 s ON u.k = s.k4
returns 2 rows on master and 3 with this change. An inner join loses a row
it should return.
`ShuffleExchangeExec` already knew about this and worked around it,
re-wrapping the partitioner's map keys through the same factory as its per-row
lookup keys so the naming could not decide (SPARK-59054). Doing the erasure
where rows are built removes that workaround's reason. This PR leaves the
re-wrap in place under a different one: the stored keys were built at
`keyDataTypes`, and re-wrapping is what makes that list and the lookups' list
agree. A mismatch there is silent, since `KeyGroupedPartitioner.getPartition`
answers a miss with the key's hash.
The factory is a chokepoint: six sites in production build partition-key
wrappers, none of them wants un-erased types, and the only readers of a
wrapper's `dataTypes` are its own `equals` and `keyDataTypes`. No site can opt
out. The class already erased half of this: `structTypeCache` names every
top-level field `"f"`, and `RowOrdering.createNaturalAscendingOrdering` forces
`nullable = true`, so `hashCode` was naming-blind while `equals` was not.
What goes is the struct field names, every nullability, and a field's
metadata, which travels with its name. Nothing else: a collation, a decimal
precision, a `char` length and a UDT all decide where a value belongs, so they
still tell two rows apart. Metadata goes because nothing that compares or
hashes a row reads it, and because `DataType.equalsStructurally` ignores it
too, so keeping it would make the erasure answer differently from the predicate
it stands for. The erasure is exactly `DataType.equalsStructurally(_, _,
ignoreNullability = true)` expressed as a canonical value rather than a
predicate, and the new suite pins it against that primitive. A value is what is
needed rather than a predicate, because the result is a `NonFateSharingCache`
key, the `dataTypes` field two wrappers compare, and what `keyDataTypes`
reports.
Nothing that compares or hashes a row reads a field name:
`GenerateOrdering.genComparisons` rebuilds each field's `SortOrder`
positionally, and `Murmur3HashFunction` hashes through the field types. So the
erasure cannot move a row or change a sort order, and the
`KeyedPartitioning.toGrouped` / `GroupPartitionsExec.groupAndSortByKeys` sort
contract is unaffected. It can only make more keys compare equal, so
`isGrouped` moves toward "not unique" and a regroup is added, never skipped.
### Does this PR introduce _any_ user-facing change?
Yes, three things.
- The join above ran into an error and now returns its rows without a
shuffle.
- The union above dropped a row and now returns it.
- `STORAGE_PARTITION_JOIN_INCOMPATIBLE_REDUCED_TYPES` prints a struct key's
field names positionally, ``STRUCT<`0`: INT>`` rather than `STRUCT<a: INT>`.
Deliberate: the message fires only on a real structural mismatch now, and the
connector's own names would point a reader at a difference that is not the
cause.
### How was this patch tested?
`InternalRowComparableWrapperSuite`, new:
- "comparableTypes erases the naming and nothing else", over 22 type pairs
including collation, decimal precision, `char` length, two UDTs, field
metadata, struct nullability and nested arrays and maps, each checked against
`DataType.equalsStructurally(ignoreNullability = true)`.
- "erasing is idempotent, and keeps the list it was given".
- "two rows of one value are equal however their columns were named", which
also asserts they hash alike and collapse in a set.
`KeyGroupedPartitioningSuite`, three end-to-end tests, all failing on master:
- "two keyed sides whose struct field names differ join without a shuffle"
-- the first query above. Fails with
`STORAGE_PARTITION_JOIN_INCOMPATIBLE_REDUCED_TYPES`.
- (the union test is not backported, see above)
- "a shuffled side keeps its own struct field names over shared empty keys"
-- a join whose two members carry the two sides' own expressions over one
shared, pruned-to-nothing key list. Fails because the members answer for two
key spaces.
`ShuffleSpecSuite`, "reduceKeys reports the types its keys are compared at",
for a `Reducer` whose result type names a struct field.
Each production hunk was ablated in turn and each has a test that fails
without it.
`KeyGroupedPartitioningSuite`, `KeyGroupedPartitioningRuntimeFilterSuite`,
`EnsureRequirementsSuite`, `GroupPartitionsExecSuite`, `PlannerSuite`,
`ExchangeSuite`, `ExplainSuite`, `DataFrameSetOperationsSuite`,
`DataSourceV2Suite`, `DataSourceV2CatalystRuntimeFilterSuite`,
`ProjectedOrderingAndPartitioningSuite`, `DistributionSuite`,
`ShuffleSpecSuite`, `TransformExpressionSuite` and
`InternalRowComparableWrapperSuite` on master. On this branch:
`KeyGroupedPartitioningSuite`, `GroupPartitionsExecSuite`,
`EnsureRequirementsSuite`, `DistributionSuite`, `ShuffleSpecSuite` and
`InternalRowComparableWrapperSuite`, 242 tests. Scalastyle and scalafmt clean.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 5)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
