andygrove commented on issue #5419:
URL:
https://github.com/apache/datafusion-comet/issues/5419#issuecomment-5464233245
Another route into this, found while reviewing #5537.
You do not need `spark.comet.shuffle.enabled=false` to reach the bad shape.
With the ordinary `CometShuffleManager` and `spark.comet.shuffle.mode=native`,
`columnarShuffleFailureReasons` short-circuits on `isCometJVMShuffleMode`
(`CometShuffleExchangeExec.scala:585`), so an exchange whose native path fails
has no columnar fallback either and reverts to a plain Spark exchange with the
native partial still underneath it. Any unsupported native partitioning will do
it, and unlike the decimal AVG case this one crashes instead of returning a
wrong answer.
```scala
spark.range(0, 18, 1, 4).selectExpr("id % 3 AS k", "id % 5 AS
v").createOrReplaceTempView("t")
```
with `spark.comet.shuffle.mode=native` and `spark.sql.shuffle.partitions=4`:
```sql
SELECT array(k) AS ak, percentile(v, 0.5) FROM t GROUP BY array(k)
```
The array hash key is unsupported for native partitioning, so the plan comes
out in exactly the shape this issue describes:
```
ObjectHashAggregate(keys=[_groupingexpression#25],
functions=[percentile(v#2L, 0.5, 1, 0, 0, false)])
+- Exchange hashpartitioning(_groupingexpression#25, 4), ENSURE_REQUIREMENTS
+- CometHashAggregate [v#2L, _groupingexpression#25], [Partial],
[partial_percentile(v#2L, 0.5, 1, 0, 0, false)]
```
and it dies with:
```
java.io.EOFException
at java.io.DataInputStream.readFully(DataInputStream.java:203)
at
org.apache.spark.sql.catalyst.expressions.aggregate.TypedAggregateWithHashMapAsBuffer.deserialize(interfaces.scala:688)
at
org.apache.spark.sql.catalyst.expressions.aggregate.TypedImperativeAggregate.merge(interfaces.scala:589)
at
org.apache.spark.sql.execution.aggregate.AggregationIterator$$anonfun$1.$anonfun$applyOrElse$3(AggregationIterator.scala:201)
```
`collect_list` produces the same plan shape and fails slightly differently:
```
java.lang.NullPointerException
at
org.apache.spark.sql.catalyst.expressions.aggregate.Collect.deserialize(collect.scala:86)
at
org.apache.spark.sql.catalyst.expressions.aggregate.TypedImperativeAggregate.merge(interfaces.scala:589)
```
Both reproduce with AQE on and off, on Spark 4.1.3 / Scala 2.13 / JDK 17.
`sum(v)` on the same query is fine, which matches `supportsMixedPartialFinal`.
The reason I am adding this here rather than opening a new issue is that
#5537 has already built the fix. Its `preserveSparkAggregateBuffers` in
`CometExecRule` walks down from a fallen-back exchange, reverts an incompatible
native Partial back to Spark, and restores any native ancestors in between,
which is what this issue asks for. It is gated behind
`isCometCelebornShuffleManagerEnabled`, so the ordinary shuffle manager does
not get it.
I tried simply deleting that gate on the #5537 branch. Both queries above
then pass, but three existing `CometAggregateSuite` tests fail with "Expected 1
Comet aggregate operators, but found 0": `group-by on variable length types`,
`SUM decimal with DF` and `Decimal Avg with DF`. All three assert a Comet
Partial under a Spark Final with `spark.comet.shuffle.enabled=false`, and
`Decimal Avg with DF` is covering the same decimal AVG shape this issue reports
as a bug. So ungating is not a one-liner: someone needs to decide which of
those assertions are pinning intended behaviour and which are pinning this bug,
and #5509 probably belongs in that conversation too.
--
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]
