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