apache / apache/datafusion-comet
Match Spark ObjectHashAggregate decimal AVG buffer semantics
- Dominant language
- Scala
- Stars
- 1.3k
- Forks
- 373
- Avg merge
- 2d 6h
- Merged PRs (30d)
- 190
Description
### 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.
Contributor guide
Assessment
This issue has not been assessed yet.