Skip to content

Native shuffle rejects nested types as hash partitioning keys although the native hasher supports them #5566

Description

@viirya

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:

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

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Metadata

Metadata

Assignees

No one assigned

    Labels

    area:shuffleShuffle (JVM and native)enhancementNew feature or request

    Type

    No type

    Projects

    No projects

      Milestone

      No milestone

      Relationships

      None yet

      Development

      No branches or pull requests

      Issue actions