sunchao opened a new issue, #5509:
URL: https://github.com/apache/datafusion-comet/issues/5509
### Describe the bug
Grouped high-precision decimal AVG can disagree with Spark when Spark uses
`ObjectHashAggregateExec`, including its sort-based fallback. Spark can retain
a temporary decimal sum that exceeds the declared buffer precision and later
returns to range; native grouped AVG currently treats that intermediate value
as overflow.
This is a remaining compatibility limitation found while reviewing #5420,
not a regression introduced by its new global-AVG guard. Exact native base/head
comparisons at `5baa6b035270a31faaa61cb49b047f62a6a3f29e` and
`e5aa8d78451059174f20df4ce3b09049198288da` produced the same behavior before
and after that PR.
### Steps to reproduce
Use Spark 4.1.3 with the Comet extension and `CometShuffleManager`
configured, native execution/scan/shuffle enabled, shuffle mode `auto`, and two
shuffle partitions. Disable AQE for this reproducer. Write the input once with
Comet disabled so both engines read the same Parquet rows:
```scala
import spark.implicits._
val path =
java.nio.file.Files.createTempDirectory("comet-object-avg").toString
spark.conf.set("spark.comet.enabled", "false")
Seq((1, 1, "0.6"), (1, 2, "0.6"), (1, 3, "-0.4"), (2, 4, null))
.toDF("g", "ord", "raw")
.selectExpr("g", "ord", "CAST(raw AS DECIMAL(38,38)) AS v")
.coalesce(1).write.mode("overwrite").parquet(path)
spark.read.parquet(path).createOrReplaceTempView("object_avg_case")
```
Run this query with Comet disabled and enabled, under both ANSI settings:
```sql
SET spark.sql.adaptive.enabled=false;
SET spark.sql.shuffle.partitions=2;
SET spark.sql.execution.useObjectHashAggregateExec=true;
SET spark.sql.objectHashAggregate.sortBased.fallbackThreshold=128;
SELECT g, AVG(v), sort_array(collect_list(ord))
FROM object_avg_case GROUP BY g;
```
`collect_list` makes Spark use an object aggregation buffer. Require
`ObjectHashAggregateExec` in the Spark plan and native `CometHashAggregateExec`
in the Comet plan, rather than accepting a silently falling-back native
control. Repeat with
`spark.sql.objectHashAggregate.sortBased.fallbackThreshold=1`; in the Spark
control, assert the object aggregate's `numTasksFallBacked` metric is positive.
### Expected behavior
For group 1, Spark returns `0.26666666666666666666666666666666666667` and
`[1,2,3]`. Group 2 returns a null AVG and `[4]`. The observed Comet result for
group 1 is null with ANSI disabled and `ARITHMETIC_OVERFLOW` with ANSI enabled.
The ordinary object and forced-spill cases reproduced the discrepancy under
both ANSI settings. The forced-spill Spark controls each recorded one fallback
task. As a separate control, disabling `useObjectHashAggregateExec` made this
query remain in Spark's sort aggregation path and match Spark under both ANSI
settings; that fallback case is not evidence of a native sort-aggregate defect.
### Additional context
The usual grouped `HashAggregateExec` uses an `UnsafeRow` decimal buffer
whose updates enforce declared precision. Grouping alone does not imply that
representation: `ObjectAggregationIterator` uses `SpecificInternalRow`,
including for the sort-based accumulator. A fix needs to preserve that
distinction or conservatively decline the affected native cases; it should not
claim that every grouped Spark aggregate truncates at the same point.
Validation used the pinned #5420 build for the Spark probes and separately
compiled the exact native base/head AVG implementations in twelve
mode/sign/batch comparisons. Those comparisons confirmed the limitation
predates #5420. This issue tracks that remaining grouped scope separately from
#5418's empty-partial/global-AVG fixes and #3079's wide-decimal hash
compatibility.
--
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]
