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: [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]
