You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "andygrove (via GitHub)" <gi...@apache.org> on 2026/10/15 15:31:47 UTC
[PR] fix: fall back to Spark for decimal AVG cases that return wrong results [datafusion-comet]
andygrove opened a new pull request, #7017:
URL: https://github.com/apache/datafusion-comet/pull/7017
## Which issue does this PR close?
Closes #7007 and #5509. Also guards the ungrouped and window cases left under #5418.
## Rationale for this change
On main, native decimal AVG returns wrong results with default settings, and no guard covers them. Each row below reproduces at 66c63bd307 on Spark 4.1.3 with a native partial and final aggregate:
| Query | Spark | Comet on main |
| --------------------------------------------------------------------------------- | ---------------------------- | --------------------------------------------------- |
| `avg` of two `DECIMAL(38,18)` rows of 9000000000000000, grouped, global or windowed | 9000000000000000 | `NULL` |
| `avg` of one `DECIMAL(38,18)` row of 1e17 under ANSI | `NUMERIC_VALUE_OUT_OF_RANGE` | `NULL` |
| global `avg` of `DECIMAL(38,38)` values 0.6, 0.6, -0.6 | 0.2 | 0 |
| global `avg` of two partitions holding 0.6 each | 0.6 | `NULL`, or `ARITHMETIC_OVERFLOW` under ANSI |
| grouped `avg` beside `collect_list`, over 0.6, 0.6, -0.6 | 0.2 | `NULL`, or `ARITHMETIC_OVERFLOW` under ANSI |
| running `avg` over 0.6, 0.6, -0.6 | 0.6, 0.6, 0.2 | 0.6, 0.6, 0.6, or `ARITHMETIC_OVERFLOW` under ANSI |
| `avg` over the whole window partition, same rows | 0.2 | 0 |
There are two causes. `avg()` in `avg_decimal.rs` multiplies the `i128` sum by 10^4 before it divides by the count, so it overflows once the sum passes about 1.7e34 unscaled (#7007). For `DECIMAL(38,18)`, the type Spark gives Scala and Java `BigDecimal` fields, that is a sum above about 1.7e16. Separately, once the sum is at precision 38, the scalar accumulator skips the row that takes it out of range and carries on. Spark's codegen, object aggregation and window buffers keep the sum unbounded, and its `UnsafeRow` buffer nulls it, so neither matches (#5418, #5509).
#5420 widens the `avg()` numerator to `i256` and declines the same precision 38 cases, but it conflicts with main and may not land before 1.2.0. This PR makes Comet return Spark's answer until then, as #6041 did for decimal SUM. When #5420 lands it can drop the first guard below and keep or replace the other three.
Wide-decimal hashing, the other known decimal gap, needs nothing new: #6005 keeps wide-decimal keys off native shuffle, and `hash`, `xxhash64` and `approx_count_distinct` already decline precision above 18.
## What changes are included in this PR?
- `CometAverage` reports a decimal AVG as `Incompatible` when the sum's precision plus the rescale to the result scale exceeds 38 digits. That is input precision 25 and up, except `DECIMAL(38,38)`, which needs no rescale. It falls back by default in aggregates and windows, and `spark.comet.expression.Average.allowIncompatible=true` still runs it natively for data whose sums stay small.
- `CometHashAggregateExec` declines an ungrouped decimal AVG whose sum is at precision 38, which is input precision 28 and up. A grouped one stays native, because both engines null the group's sum, or raise under ANSI, once it overflows.
- `CometObjectHashAggregateExec` declines a decimal AVG whose sum is at precision 38, grouped or not. This is the AVG counterpart of the SUM check #6041 added.
- `CometWindowExec` declines a decimal AVG whose sum is at precision 38 for every frame. Sliding frames already fell back.
- Docs: a Hash Aggregation section and a window fallback entry in the operator compatibility guide, and the `avg`, `mean` and `try_avg` notes in the expressions table. The generated aggregate compatibility page lists the new incompatible case.
The cost is that AVG over `DECIMAL(38,18)` and other `DECIMAL(25+, s)` inputs runs in Spark by default, along with any sibling aggregates in the same operator.
## How are these changes tested?
- New tests in `CometAggregateSuite` and `CometWindowExecSuite` cover every row of the table, with ANSI on and off and with `try_avg`. Each asserts Spark's answer, the fallback reason, and that no Comet aggregate or window ran. Controls stay native: `DECIMAL(24,2)`, grouped `DECIMAL(38,38)`, `DECIMAL(10,2)` beside `collect_list`, and `DECIMAL(25,2)` in an aggregate and a window with `allowIncompatible`.
- Removing a guard makes its tests fail with a wrong answer, an error Spark does not raise, or a missing fallback reason. That was checked for all four guards.
- Five existing tests asserted native AVG over `DECIMAL(38, s)`. They now set `allowIncompatible` to keep comparing the native grouped AVG with Spark, and their ungrouped queries at precision 38 now expect the aggregate to run in Spark.
- Locally on Spark 4.1: CometAggregateSuite, CometWindowExecSuite and CometExecRuleSuite (325 tests). Also the aggregate and window SQL file tests, CometFloatSemanticsSuite, CometShuffleSuite, DisableAQECometShuffleSuite, CometShuffleInputConversionSuite, CometFuzzAggregateSuite, and both TPC-DS plan stability suites, which needed no plan changes (766 tests). The new and changed tests also pass on Spark 3.4, 3.5 with `-Pstrict-warnings`, and 4.2, and the semantic scalafix check passes on 3.4.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: fall back to Spark for decimal AVG cases that return wrong results [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove merged PR #7017:
URL: https://github.com/apache/datafusion-comet/pull/7017
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: fall back to Spark for decimal AVG cases that return wrong results [datafusion-comet]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #7017:
URL: https://github.com/apache/datafusion-comet/pull/7017#discussion_r4288426323
##########
docs/source/user-guide/latest/compatibility/operators.md:
##########
@@ -74,6 +74,24 @@ key whose type cannot hold a null element or field, such as `array(coalesce(x, 0
`spark.comet.expression.SortOrder.allowIncompatible=true` to run the other null orders natively
anyway.
+## Hash Aggregation
+
+Decimal `avg` with input precision 25 or more falls back to Spark, except over `DECIMAL(38, 38)`.
Review Comment:
The rescale overflow comes from the `avg` expression, so it applies to sort aggregates and windows as well as hash aggregates. The window section below says so, but this paragraph sits under "Hash Aggregation", and the `avg`, `mean` and `try_avg` rows in `expressions.md` link here. Could the paragraph say explicitly that it applies to every operator that runs `avg`, or move to a spot that isn't tied to one operator? If the i256 fix from #5420 comes in here instead, this paragraph and the three `expressions.md` notes can go away.
##########
spark/src/main/scala/org/apache/spark/sql/comet/CometWindowExec.scala:
##########
@@ -193,6 +193,19 @@ object CometWindowExec extends CometOperatorSerde[WindowExec] {
}
}
+ /**
+ * Whether a decimal AVG keeps its running sum at DecimalType.MAX_PRECISION, which Spark does
+ * for input precision 28 and up. Spark's window buffer keeps that sum unbounded, while the
+ * native AVG skips the row that takes it out of the precision, so DECIMAL(38,38) inputs 0.6,
+ * 0.6, -0.6 average to 0 over the whole partition rather than Spark's 0.2, and to 0.6 rather
+ * than 0.2 in a running frame, where ANSI mode raises instead.
+ * https://github.com/apache/datafusion-comet/issues/5418
+ */
+ private def hasMaxPrecisionDecimalSum(avg: Average): Boolean = avg.sumDataType match {
Review Comment:
This is the same check as `CometBaseAggregate.hasMaxPrecisionDecimalAvg` in `operators.scala`. The name also reads as if it were about SUM, even though it takes an `Average`. Could both call sites share one helper, for example `AggSerde.avgHasMaxPrecisionDecimalSum(avg: Average)`? Then the aggregate and window guards can't drift apart when #5420 changes them.
##########
spark/src/main/scala/org/apache/comet/serde/aggregates.scala:
##########
@@ -235,14 +235,42 @@ object CometAverage extends CometAggregateExpressionSerde[Average] {
// deferred (overflow nulls count differently) and stays unsafe for mixed execution.
!fn.child.dataType.isInstanceOf[DecimalType]
+ private val rescaledSumOverflowReason =
+ "a large sum of a DECIMAL with precision 25 or more overflows when the native average " +
+ "rescales it, which returns NULL " +
+ "(https://github.com/apache/datafusion-comet/issues/7007)"
+
override def getUnsupportedReasons(): Seq[String] = Seq(
"YearMonthIntervalType and DayTimeIntervalType inputs are not supported")
+ override def getIncompatibleReasons(): Seq[String] = Seq(
+ "For a `DECIMAL` input with precision 25 or more, other than `DECIMAL(38, 38)`, the native " +
+ "average rescales the sum within 38 digits before dividing it by the count, so a large " +
+ "sum returns `NULL` where Spark returns the average. Under ANSI, an average that Spark " +
+ "rejects with `NUMERIC_VALUE_OUT_OF_RANGE` also returns `NULL` " +
+ "([#7007](https://github.com/apache/datafusion-comet/issues/7007)).")
+
+ /**
+ * Whether the native average can overflow when it rescales the sum to the result scale. It
+ * multiplies the sum, which holds up to `sumDataType.precision` digits, by 10 to the power of
+ * the scale difference in an `i128`, which holds 38 digits, before dividing by the count, while
+ * Spark divides with `BigDecimal`. That covers input precision 25 and up, except scale 38,
+ * which needs no rescale.
+ */
+ private def rescaledSumCanOverflow(expr: Average): Boolean =
+ (expr.sumDataType, expr.dataType) match {
+ case (sum: DecimalType, result: DecimalType) =>
+ sum.precision + result.scale - sum.scale > DecimalType.MAX_PRECISION
+ case _ => false
+ }
+
override def getSupportLevel(expr: Average): SupportLevel =
- if (AggSerde.avgDataTypeSupported(expr.dataType)) {
- Compatible()
- } else {
+ if (!AggSerde.avgDataTypeSupported(expr.dataType)) {
Unsupported(Some(s"Unsupported data type: ${expr.dataType}"))
+ } else if (rescaledSumCanOverflow(expr)) {
+ Incompatible(Some(rescaledSumOverflowReason))
Review Comment:
Precision 25 and up is common for decimals that come out of arithmetic, not only for `DECIMAL(38,18)` columns. For example, `DECIMAL(15,2) * DECIMAL(10,2)` is `DECIMAL(26,4)`. So with this gate, `avg(price * qty)` falls back by default, and so does every other aggregate in the same operator. For grouped queries that's a sizable regression for a bug that only shows up once the sum passes about 1.7e34 unscaled.
#5420 already has a small, self-contained fix for `avg()`. It keeps the i128 fast path, widens to i256 only when the scaled numerator overflows, and raises `NUMERIC_VALUE_OUT_OF_RANGE` under ANSI when the average doesn't fit the result type. Could we bring just that hunk in here instead of the `Incompatible` gate? That would remove this support level, `getIncompatibleReasons`, the three `expressions.md` notes and the `allowIncompatible` opt-ins in the five existing tests. Grouped AVG over `DECIMAL(25..37, s)` would also stay native and correct. The precision 38 guards would still be needed until the rest of #5420 lands.
If you'd rather keep the gate for now, could the PR description and the docs mention the arithmetic-result case, so users can tell why `avg(a * b)` stopped running natively?
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org