sunchao commented on code in PR #5567:
URL: https://github.com/apache/datafusion-comet/pull/5567#discussion_r3910167338
##########
spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometShuffleExchangeExec.scala:
##########
@@ -424,6 +433,22 @@ object CometShuffleExchangeExec
true
case dt if isTimeType(dt) =>
true
+ case StructType(fields) if nestedHashPartitioningEnabled =>
+ // `fields.nonEmpty` mirrors the guard on the data-column gate below.
An empty struct is
+ // not reachable end-to-end anyway: Parquet cannot store an empty
group, and an in-memory
+ // relation with one does not survive scan conversion.
+ fields.nonEmpty && fields.forall(f =>
supportedHashPartitioningDataType(f.dataType))
+ case ArrayType(elementType, _) if nestedHashPartitioningEnabled =>
+ supportedHashPartitioningDataType(elementType)
+ case MapType(keyType, valueType, _) if nestedHashPartitioningEnabled =>
+ // 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(...)`, which is
+ // gated separately by CometMapSort (scalar map keys only) and, when
unsupported, fails
+ // the expression check below. Earlier Spark versions insert no such
normalization, so
+ // Comet would hash physical entry order and could route equal maps
differently.
+ isSpark40Plus &&
+ supportedHashPartitioningDataType(keyType) &&
+ supportedHashPartitioningDataType(valueType)
Review Comment:
[P2] Rebase sliced map offsets before enabling map shuffle keys
Could we fix `spark_map_sort` for sliced maps before admitting this case?
With the nested-hash config enabled on Spark 4.0, a native OFFSET below a map
repartition can pass a MapArray whose first entry offset is nonzero.
DataFusion's limit uses `batch.slice(skip, ...)`, and Arrow preserves the map's
original entry offsets. The unified native shuffle evaluates `mapsort(m)` on
that slice without a JVM round trip.
`map_sort.rs` builds its `take` indices only for the visible maps, but
passes the original offsets to `MapArray::try_new`. For two two-entry maps,
skipping the first leaves offsets `[2,4]`. Sorting creates two entries, so
Arrow rejects the result with `Max offset of 4 exceeds length of entries 2`.
The query fails instead of repartitioning. The map-sort code predates this PR,
but the base gate rejected map hash keys, so the newly admitted path exposes it.
Please rebase the output offsets, or retain fallback for this case, and
cover OFFSET followed by map repartition while asserting both operators run
natively. This finding is source-derived. I have not executed the query.
--
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]
