sunchao commented on code in PR #5420:
URL: https://github.com/apache/datafusion-comet/pull/5420#discussion_r4316318017
##########
spark/src/main/scala/org/apache/spark/sql/comet/operators.scala:
##########
@@ -1979,6 +1979,30 @@ trait CometBaseAggregate {
case _ => false
})
+ protected def hasMaxPrecisionDecimalAvg(op: BaseAggregateExec): Boolean =
+ op.aggregateExpressions.exists(_.aggregateFunction match {
+ case avg: Average =>
+ avg.sumDataType match {
+ case decimal: DecimalType => decimal.precision ==
DecimalType.MAX_PRECISION
+ case _ => false
+ }
+ case _ => false
+ })
+
+ protected def aggregateSupportLevel(op: BaseAggregateExec): SupportLevel = {
+ val unsupportedAverage = op.groupingExpressions.isEmpty &&
hasMaxPrecisionDecimalAvg(op)
+
+ if (unsupportedAverage) {
+ // Spark's global buffer can retain a wider sum until division; native
overflow is sticky.
+ // Both stages must fall back because decimal AVG buffers cannot cross
engines.
+ Unsupported(
+ Some(
+ "Ungrouped AVG on DECIMAL with maximum-precision intermediate state
is not supported"))
+ } else {
+ Compatible()
+ }
+ }
Review Comment:
Addressed in `8c1658a9a4f4459b470ebec7848c73f3eed484be`.
Rebased onto main and took the separate-follow-up option: #7116 tracks a
wider scalar AVG running sum, the codegen-off boundary, partial-state checks,
and expanding windows. The Hash Aggregation and Window Functions sections now
link it. This keeps the existing correctness fallback and main's object/sort
policies while making the native-coverage optimization explicit.
##########
native/spark-expr/src/agg_funcs/avg_decimal.rs:
##########
@@ -671,31 +609,281 @@ impl GroupsAccumulator for AvgDecimalGroupsAccumulator {
}
/// Returns the `sum`/`count` as a i128 Decimal128 with
-/// target_scale and target_precision and return None if overflows.
+/// target_scale and target_precision, returning the rounded wide value on
overflow.
///
/// * sum: The total sum value stored as Decimal128 with sum_scale
/// * count: total count, stored as a i128 (*NOT* a Decimal128 value)
/// * target_min: The minimum output value possible to represent with the
target precision
/// * target_max: The maximum output value possible to represent with the
target precision
/// * scaler: scale factor for avg
#[inline(always)]
-fn avg(sum: i128, count: i128, target_min: i128, target_max: i128, scaler:
i128) -> Option<i128> {
- if let Some(value) = sum.checked_mul(scaler) {
- // `sum / count` with ROUND_HALF_UP
+fn avg(
+ sum: i128,
+ count: i128,
+ target_min: i128,
+ target_max: i128,
+ scaler: i128,
+) -> std::result::Result<i128, i256> {
+ // The scaled numerator can exceed i128 while both the sum and final
average fit.
+ // Use the inexpensive path when it fits, and widen only that intermediate
otherwise.
+ let rounded = if let Some(value) = sum.checked_mul(scaler) {
let (div, rem) = value.div_rem(&count);
let half = div_ceil(count, 2);
- let half_neg = half.neg_wrapping();
- let new_value = match value >= 0 {
+ let rounded = match value >= 0 {
true if rem >= half => div.add_wrapping(1),
- false if rem <= half_neg => div.sub_wrapping(1),
+ false if rem <= half.neg_wrapping() => div.sub_wrapping(1),
_ => div,
};
- if new_value >= target_min && new_value <= target_max {
- Some(new_value)
+ i256::from_i128(rounded)
+ } else {
+ // A Decimal128 numerator times the Decimal128 scale factor fits
within i256.
+ let value = i256::from_i128(sum) * i256::from_i128(scaler);
+ let divisor = i256::from_i128(count);
+ let (div, rem) = (value / divisor, value % divisor);
+ let half = i256::from_i128(div_ceil(count, 2));
+ if rem >= half {
+ div + i256::ONE
+ } else if rem <= -half {
+ div - i256::ONE
} else {
- None
+ div
}
+ };
+ if rounded >= i256::from_i128(target_min) && rounded <=
i256::from_i128(target_max) {
+ Ok(rounded.to_i128().expect("bounded Decimal128 average"))
} else {
- None
+ Err(rounded)
+ }
+}
+
+fn avg_result_overflow_error(value: i256, precision: u8, scale: i8) ->
crate::SparkError {
+ let unscaled = value.to_string();
+ let digits = unscaled.trim_start_matches('-').len();
+ crate::SparkError::NumericValueOutOfRange {
+ value: format_decimal_str(&unscaled, digits, scale),
+ precision,
+ scale,
+ }
+}
Review Comment:
Addressed in `8c1658a9a4f4459b470ebec7848c73f3eed484be`.
Changed avg_result_overflow_error to Decimal256Type::format_decimal(value,
DECIMAL256_MAX_PRECISION, scale), matching SUM and removing the manual digit
counting. The five focused native AVG tests pass, including widened division
and final-result overflow.
##########
native/spark-expr/src/agg_funcs/avg_decimal.rs:
##########
@@ -671,31 +609,281 @@ impl GroupsAccumulator for AvgDecimalGroupsAccumulator {
}
/// Returns the `sum`/`count` as a i128 Decimal128 with
-/// target_scale and target_precision and return None if overflows.
+/// target_scale and target_precision, returning the rounded wide value on
overflow.
///
/// * sum: The total sum value stored as Decimal128 with sum_scale
/// * count: total count, stored as a i128 (*NOT* a Decimal128 value)
/// * target_min: The minimum output value possible to represent with the
target precision
/// * target_max: The maximum output value possible to represent with the
target precision
/// * scaler: scale factor for avg
#[inline(always)]
-fn avg(sum: i128, count: i128, target_min: i128, target_max: i128, scaler:
i128) -> Option<i128> {
- if let Some(value) = sum.checked_mul(scaler) {
- // `sum / count` with ROUND_HALF_UP
+fn avg(
+ sum: i128,
+ count: i128,
+ target_min: i128,
+ target_max: i128,
+ scaler: i128,
+) -> std::result::Result<i128, i256> {
+ // The scaled numerator can exceed i128 while both the sum and final
average fit.
+ // Use the inexpensive path when it fits, and widen only that intermediate
otherwise.
+ let rounded = if let Some(value) = sum.checked_mul(scaler) {
let (div, rem) = value.div_rem(&count);
let half = div_ceil(count, 2);
- let half_neg = half.neg_wrapping();
- let new_value = match value >= 0 {
+ let rounded = match value >= 0 {
true if rem >= half => div.add_wrapping(1),
- false if rem <= half_neg => div.sub_wrapping(1),
+ false if rem <= half.neg_wrapping() => div.sub_wrapping(1),
_ => div,
};
- if new_value >= target_min && new_value <= target_max {
- Some(new_value)
+ i256::from_i128(rounded)
+ } else {
+ // A Decimal128 numerator times the Decimal128 scale factor fits
within i256.
+ let value = i256::from_i128(sum) * i256::from_i128(scaler);
+ let divisor = i256::from_i128(count);
+ let (div, rem) = (value / divisor, value % divisor);
+ let half = i256::from_i128(div_ceil(count, 2));
+ if rem >= half {
+ div + i256::ONE
+ } else if rem <= -half {
+ div - i256::ONE
} else {
- None
+ div
}
+ };
+ if rounded >= i256::from_i128(target_min) && rounded <=
i256::from_i128(target_max) {
+ Ok(rounded.to_i128().expect("bounded Decimal128 average"))
} else {
- None
+ Err(rounded)
+ }
+}
+
+fn avg_result_overflow_error(value: i256, precision: u8, scale: i8) ->
crate::SparkError {
Review Comment:
Addressed in `8c1658a9a4f4459b470ebec7848c73f3eed484be`.
Confirmed with Spark 4.0.4: Decimal division retains a quotient truncated to
39 fractional digits, and the precision error prints that value. I took the
documentation option and added a note beside grouped AVG in the
known-divergences list: Comet currently prints the average rounded to its
result scale. The SQL ANSI regression checks NUMERIC_VALUE_OUT_OF_RANGE for
both engines; this does not claim identical message values.
--
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]
