Match Spark ObjectHashAggregate decimal AVG buffer semantics

[sunchao]

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:

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:

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.