0lai0 opened a new issue, #5971: URL: https://github.com/apache/datafusion-comet/issues/5971
## What is the problem? `CometShuffleExchangeExec` decides whether *columnar* (JVM) shuffle can handle a `RangePartitioning` by asking whether Comet can serialize the sort orders to protobuf: https://github.com/apache/datafusion-comet/blob/main/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometShuffleExchangeExec.scala#L654-L659 ```scala case RangePartitioning(orderings, _) => for (o <- orderings) { if (QueryPlanSerde.exprToProto(o, inputs).isEmpty) { reasons += s"unsupported range partitioning sort order: $o" } } ``` But on the columnar path the range partitioning is performed entirely on the JVM by Spark's own `RangePartitioner`, over an `UnsafeProjection` of the sort keys and a `LazilyGeneratedOrdering`: https://github.com/apache/datafusion-comet/blob/main/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometShuffleExchangeExec.scala#L1002-L1022 ```scala case RangePartitioning(sortingExpressions, numPartitions) => val rddForSampling = rdd.mapPartitionsInternal { iter => val projection = UnsafeProjection.create(sortingExpressions.map(_.child), outputAttributes) ... } implicit val ordering = new LazilyGeneratedOrdering(orderingAttributes) new RangePartitioner(numPartitions, rddForSampling, ascending = true, ...) ``` The serialized `SortOrder` produced by the check is never sent to native code on this path. So the gate rejects columnar shuffle for orderings that the JVM would have partitioned correctly, and the query falls back to Spark's shuffle for no compatibility reason. ## Impact Any ordering expression that Comet cannot serialize disables Comet's columnar shuffle for that exchange, even though nothing about the partitioning would have run natively. Examples: - floating-point keys nested in arrays or structs while `spark.comet.exec.strictFloatingPoint=true` - any `ORDER BY` / `repartitionByRange` expression Comet has no serde for This is an unnecessary-fallback bug rather than a correctness bug. ## History Until #5506 this also fired for **scalar** float and double keys under `spark.comet.exec.strictFloatingPoint=true`, because `CometSortOrder.getSupportLevel` returned `Incompatible` for them and `exprToProto` therefore returned `None`. #5506 narrowed that verdict, so the scalar floating-point case no longer reaches this gate. The structural problem is unchanged. ## Suggested fix Drop the `exprToProto` probe from the columnar branch, keeping the collation check that follows it (collation genuinely affects JVM ordering semantics here). **Please verify before fixing** — I have not confirmed this myself: that no part of the columnar shuffle write path consumes a serialized partitioning for `RangePartitioning`. If some writer does, the gate is load-bearing and the fix is instead to narrow it to whatever that writer actually needs. ## Suggested test Pick an ordering that Comet cannot serialize but the JVM can sort — a struct containing a float with `spark.comet.exec.strictFloatingPoint=true` is a ready-made one — and assert with `spark.comet.exec.shuffle.mode=jvm` that the exchange stays on Comet's columnar shuffle instead of falling back to Spark's. -- 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]
