Repository navigation
Description
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:
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 Fix listagg-collation.sql test in Spark 4.0.0 #1947 / fix: fall back to Spark for shuffle/sort/aggregate on non-default collated strings [Spark 4] #4035, rows equal under the collation reached
different partitions and a downstream collation-awareDISTINCTin Spark then
producedaabbinstead ofab. - CalendarInterval -- allowed as a shuffle data column, but the native
hasher has noIntervalbranch and fails with "Unsupported data type in
hasher" (Hashing a CalendarInterval value fails with "Unsupported data type in hasher: Interval(MonthDayNano)" #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.
Activity