viirya opened a new issue, #5566:
URL: https://github.com/apache/datafusion-comet/issues/5566
## Is your feature request related to a problem or challenge?
`CometShuffleExchangeExec.supportedHashPartitioningDataType` rejects nested
types
as hash partitioning keys, so a query that repartitions on a struct, array
or map
column falls back to Spark for the entire shuffle:
```scala
def supportedHashPartitioningDataType(dt: DataType): Boolean = dt match {
case st: StringType if isStringCollationType(st) => false
case _: BooleanType | ... | _: DateType => true
case _: DecimalType => true
case dt if isTimeType(dt) => true
case _ => false // struct / array / map land here
}
```
The comment above it explains the restriction as:
> Native code does not support hashing complex types, see hash_funcs/utils.rs
That is no longer accurate. `native/spark-expr/src/hash_funcs/utils.rs`
hashes
nested types recursively -- struct fields, `List`/`LargeList`/`FixedSizeList`
elements, and map keys and values -- and shuffle partitioning calls the same
`create_murmur3_hashes` entry point, with the same seed, that the `hash`
expression uses. Nested hashing is exercised today through `hash()` /
`xxhash64()`,
whose gate (`HashUtils.unsupportedReasonFor` in `serde/hash.scala`) recurses
through struct/array/map, with coverage in `CometHashExpressionSuite`.
So a shuffle on a nested key falls back even though the machinery to run it
natively is present and in use elsewhere.
## Describe the solution you'd like
Allow struct/array/map as hash partitioning keys for native shuffle, checked
recursively so that a leaf type which cannot be hashed natively disqualifies
the
whole key and the shuffle still falls back to Spark. Two such leaves exist
today:
- **collated strings** -- Comet hashes raw bytes. This is a correctness
matter,
not just a difference: in #1947 / #4035, rows equal under the collation
reached
different partitions and a downstream collation-aware `DISTINCT` in Spark
then
produced `aabb` instead of `ab`.
- **CalendarInterval** -- allowed as a shuffle *data* column, but the native
hasher has no `Interval` branch and fails with "Unsupported data type in
hasher" (#5059).
Map keys need one extra restriction. Map entry order is not semantically
meaningful, so two equal maps must hash alike. Spark 4.0+ normalizes a map
shuffle key by wrapping it in `mapsort(...)` (#1941); earlier versions
insert no
such normalization, so Comet would hash physical entry order and could route
equal maps to different partitions. Map keys should therefore be admitted
only on
Spark 4.0+, and only when the `mapsort` is itself convertible --
`CometMapSort`
supports scalar map keys only, and otherwise the existing
partitioning-expression
check already forces a fallback.
Worth putting behind a config so the previous behavior remains available.
## Describe alternatives you've considered
Leaving the gate as it is and only correcting the stale comment. That keeps a
supported, tested native capability unreachable from shuffle for no technical
reason.
## Additional context
Notably, Spark rewrites a deeply nested partitioning key into a
`transform(...)`
containing a nested `mapsort(...)`, e.g. for
`struct<a: array<struct<m: map<string, array<int>>, s: string>>, i: int>`:
```
hashpartitioning(if (isnull(k)) null else named_struct(
a, transform(k.a, lambdafunction(
if (isnull(x)) null else named_struct(m, mapsort(x.m), s, x.s), x,
false)),
i, k.i), 10)
```
so the normalization applies at any depth, not only to a top-level map key.
--
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]
