You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/09/15 13:51:24 UTC
(datafusion-comet) branch gh-readonly-queue/main/pr-5160-a8e8157ead46e7971a72049e9a4534c9e6f2b5aa created (now 4479e722ef)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-5160-a8e8157ead46e7971a72049e9a4534c9e6f2b5aa
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
at 4479e722ef refactor: use Arrow decimal precision validation (#5160)
This branch includes the following new commits:
new 4479e722ef refactor: use Arrow decimal precision validation (#5160)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org
(datafusion-comet) 01/01: refactor: use Arrow decimal precision validation (#5160)
Posted by gi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-5160-a8e8157ead46e7971a72049e9a4534c9e6f2b5aa
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
commit 4479e722efebd33620f29a246f7863bd04f5df1a
Author: Han-Yin Chang <ni...@gmail.com>
AuthorDate: Tue Sep 15 13:50:59 2026 +0000
refactor: use Arrow decimal precision validation (#5160)
* refactor: use Arrow decimal precision validation
Closes #5089
* fix: update decimal precision check after helper removal
---------
Co-authored-by: Andy Grove <ag...@apache.org>
---
native/spark-expr/src/agg_funcs/avg_decimal.rs | 17 +++++++++-----
native/spark-expr/src/agg_funcs/sum_decimal.rs | 31 ++++++++++++++++++-------
native/spark-expr/src/iceberg_funcs/truncate.rs | 8 ++++---
native/spark-expr/src/utils.rs | 15 +-----------
4 files changed, 39 insertions(+), 32 deletions(-)
diff --git a/native/spark-expr/src/agg_funcs/avg_decimal.rs b/native/spark-expr/src/agg_funcs/avg_decimal.rs
index 0e02dd03d4..c58dd213de 100644
--- a/native/spark-expr/src/agg_funcs/avg_decimal.rs
+++ b/native/spark-expr/src/agg_funcs/avg_decimal.rs
@@ -18,7 +18,7 @@
use arrow::array::{
builder::PrimitiveBuilder,
cast::AsArray,
- types::{Decimal128Type, Int64Type},
+ types::{Decimal128Type, DecimalType, Int64Type},
Array, ArrayRef, Decimal128Array, Int64Array, PrimitiveArray,
};
use arrow::datatypes::{DataType, Field, FieldRef};
@@ -30,7 +30,7 @@ use datafusion::logical_expr::{
use datafusion::physical_expr::expressions::format_state_name;
use std::sync::Arc;
-use crate::utils::{build_bool_state, is_valid_decimal_precision, unlikely};
+use crate::utils::{build_bool_state, unlikely};
use crate::{decimal_sum_overflow_error, EvalMode, SparkErrorWithContext};
use arrow::array::ArrowNativeTypeOp;
use arrow::datatypes::{
@@ -272,7 +272,7 @@ impl AvgDecimalAccumulator {
None => (v, false),
};
- if is_overflow || !is_valid_decimal_precision(new_sum, self.sum_precision) {
+ if is_overflow || !Decimal128Type::is_valid_decimal_precision(new_sum, self.sum_precision) {
// Overflow: set to null. Error will be thrown during evaluate in ANSI mode.
// This matches Spark's DecimalAddNoOverflowCheck behavior.
self.is_not_null = false;
@@ -347,7 +347,8 @@ impl Accumulator for AvgDecimalAccumulator {
let v = self.sum.get_or_insert(0);
let (result, overflowed) = v.overflowing_add(x);
- if overflowed || !is_valid_decimal_precision(result, self.sum_precision) {
+ if overflowed || !Decimal128Type::is_valid_decimal_precision(result, self.sum_precision)
+ {
// Overflow during merge: set to null, error will be thrown during evaluate in ANSI mode
self.is_not_null = false;
self.sum = None;
@@ -485,7 +486,9 @@ impl AvgDecimalGroupsAccumulator {
self.counts[group_index] += 1;
self.sums[group_index] = new_sum;
- if unlikely(is_overflow || !is_valid_decimal_precision(new_sum, self.sum_precision)) {
+ if unlikely(
+ is_overflow || !Decimal128Type::is_valid_decimal_precision(new_sum, self.sum_precision),
+ ) {
// Overflow: set to null. Error will be thrown during evaluate in ANSI mode.
// This matches Spark's DecimalAddNoOverflowCheck behavior.
self.is_not_null.set_bit(group_index, false);
@@ -572,7 +575,9 @@ impl GroupsAccumulator for AvgDecimalGroupsAccumulator {
let sum = self.sums[group_index];
let (new_sum, is_overflow) = sum.overflowing_add(new_value);
- if is_overflow || !is_valid_decimal_precision(new_sum, self.sum_precision) {
+ if is_overflow
+ || !Decimal128Type::is_valid_decimal_precision(new_sum, self.sum_precision)
+ {
if self.eval_mode == EvalMode::Ansi {
let error = decimal_sum_overflow_error("avg");
return Err(self.wrap_error_with_context(error));
diff --git a/native/spark-expr/src/agg_funcs/sum_decimal.rs b/native/spark-expr/src/agg_funcs/sum_decimal.rs
index cae0a70b42..cc2bc10566 100644
--- a/native/spark-expr/src/agg_funcs/sum_decimal.rs
+++ b/native/spark-expr/src/agg_funcs/sum_decimal.rs
@@ -15,10 +15,11 @@
// specific language governing permissions and limitations
// under the License.
-use crate::utils::is_valid_decimal_precision;
use crate::{decimal_sum_overflow_error, EvalMode, SparkErrorWithContext};
use arrow::array::{
- cast::AsArray, types::Decimal128Type, Array, ArrayRef, BooleanArray, Decimal128Array,
+ cast::AsArray,
+ types::{Decimal128Type, DecimalType},
+ Array, ArrayRef, BooleanArray, Decimal128Array,
};
use arrow::datatypes::{DataType, Field, FieldRef};
use datafusion::common::{not_impl_err, DataFusionError, Result as DFResult, ScalarValue};
@@ -208,7 +209,7 @@ impl SumDecimalAccumulator {
let running_sum = self.sum.unwrap_or(0);
let (new_sum, is_overflow) = running_sum.overflowing_add(v);
- if is_overflow || !is_valid_decimal_precision(new_sum, self.precision) {
+ if is_overflow || !Decimal128Type::is_valid_decimal_precision(new_sum, self.precision) {
if self.eval_mode == EvalMode::Ansi {
let error = decimal_sum_overflow_error("sum");
return Err(self.wrap_error_with_context(error));
@@ -276,7 +277,9 @@ impl Accumulator for SumDecimalAccumulator {
)
} else {
match self.sum {
- Some(sum_value) if is_valid_decimal_precision(sum_value, self.precision) => {
+ Some(sum_value)
+ if Decimal128Type::is_valid_decimal_precision(sum_value, self.precision) =>
+ {
ScalarValue::try_new_decimal128(sum_value, self.precision, self.scale)
}
_ => ScalarValue::new_primitive::<Decimal128Type>(
@@ -351,7 +354,7 @@ impl Accumulator for SumDecimalAccumulator {
let right = that_sum.unwrap();
let (new_sum, is_overflow) = left.overflowing_add(right);
- if is_overflow || !is_valid_decimal_precision(new_sum, self.precision) {
+ if is_overflow || !Decimal128Type::is_valid_decimal_precision(new_sum, self.precision) {
if self.eval_mode == EvalMode::Ansi {
let error = decimal_sum_overflow_error("sum");
return Err(self.wrap_error_with_context(error));
@@ -424,7 +427,7 @@ impl SumDecimalGroupsAccumulator {
let running_sum = self.sum[group_index].unwrap_or(0);
let (new_sum, is_overflow) = running_sum.overflowing_add(value);
- if is_overflow || !is_valid_decimal_precision(new_sum, self.precision) {
+ if is_overflow || !Decimal128Type::is_valid_decimal_precision(new_sum, self.precision) {
if self.eval_mode == EvalMode::Ansi {
let error = decimal_sum_overflow_error("sum");
return Err(self.wrap_error_with_context(error));
@@ -484,7 +487,12 @@ impl GroupsAccumulator for SumDecimalGroupsAccumulator {
None
} else {
match sum {
- Some(v) if is_valid_decimal_precision(v, self.precision) => {
+ Some(v)
+ if Decimal128Type::is_valid_decimal_precision(
+ v,
+ self.precision,
+ ) =>
+ {
Some(v)
}
_ => None,
@@ -508,7 +516,12 @@ impl GroupsAccumulator for SumDecimalGroupsAccumulator {
None
} else {
match sum {
- Some(v) if is_valid_decimal_precision(v, self.precision) => {
+ Some(v)
+ if Decimal128Type::is_valid_decimal_precision(
+ v,
+ self.precision,
+ ) =>
+ {
Some(v)
}
_ => None,
@@ -587,7 +600,7 @@ impl GroupsAccumulator for SumDecimalGroupsAccumulator {
let right = that_sum_val.unwrap();
let (new_sum, is_overflow) = left.overflowing_add(right);
- if is_overflow || !is_valid_decimal_precision(new_sum, self.precision) {
+ if is_overflow || !Decimal128Type::is_valid_decimal_precision(new_sum, self.precision) {
if self.eval_mode == EvalMode::Ansi {
let error = decimal_sum_overflow_error("sum");
return Err(self.wrap_error_with_context(error));
diff --git a/native/spark-expr/src/iceberg_funcs/truncate.rs b/native/spark-expr/src/iceberg_funcs/truncate.rs
index 70b638f52b..b18ca614e6 100644
--- a/native/spark-expr/src/iceberg_funcs/truncate.rs
+++ b/native/spark-expr/src/iceberg_funcs/truncate.rs
@@ -20,10 +20,11 @@
//! points of a string, and the first `W` bytes of a binary value.
use super::{apply_unary, positive_int_param, unpacked_type, unsupported_type};
-use crate::utils::is_valid_decimal_precision;
use arrow::array::{Array, ArrayRef, AsArray, Decimal128Array, OffsetSizeTrait};
use arrow::compute::kernels::substring::{substring, substring_by_char};
-use arrow::datatypes::{DataType, Decimal128Type, Int16Type, Int32Type, Int64Type, Int8Type};
+use arrow::datatypes::{
+ DataType, Decimal128Type, DecimalType, Int16Type, Int32Type, Int64Type, Int8Type,
+};
use datafusion::common::{utils::take_function_args, Result};
use datafusion::logical_expr::{
ColumnarValue, ScalarFunctionArgs, ScalarUDFImpl, Signature, Volatility,
@@ -108,7 +109,8 @@ fn truncate_array(fn_name: &str, array: &ArrayRef, width: i32) -> Result<ArrayRe
let truncated: Decimal128Array =
array.as_primitive::<Decimal128Type>().unary_opt(|v| {
let truncated = truncate_i128(v, width as i128);
- is_valid_decimal_precision(truncated, *precision).then_some(truncated)
+ Decimal128Type::is_valid_decimal_precision(truncated, *precision)
+ .then_some(truncated)
});
Arc::new(truncated.with_precision_and_scale(*precision, *scale)?)
}
diff --git a/native/spark-expr/src/utils.rs b/native/spark-expr/src/utils.rs
index 7ab665c195..8007275ae3 100644
--- a/native/spark-expr/src/utils.rs
+++ b/native/spark-expr/src/utils.rs
@@ -15,7 +15,7 @@
// specific language governing permissions and limitations
// under the License.
-use arrow::datatypes::{DataType, TimeUnit, DECIMAL128_MAX_PRECISION};
+use arrow::datatypes::{DataType, TimeUnit};
use arrow::{
array::{
cast::as_primitive_array,
@@ -29,7 +29,6 @@ use std::sync::Arc;
use arrow::array::timezone::Tz;
use arrow::array::types::TimestampMillisecondType;
-use arrow::datatypes::{MAX_DECIMAL128_FOR_EACH_PRECISION, MIN_DECIMAL128_FOR_EACH_PRECISION};
use arrow::error::ArrowError;
use arrow::{
array::{as_dictionary_array, Array, ArrayRef, PrimitiveArray},
@@ -315,18 +314,6 @@ fn pre_timestamp_cast(array: ArrayRef, timezone: String) -> Result<ArrayRef, Arr
}
}
-/// Adapted from arrow-rs `validate_decimal_precision` but returns bool
-/// instead of Err to avoid the cost of formatting the error strings and is
-/// optimized to remove a memcpy that exists in the original function
-/// we can remove this code once we upgrade to a version of arrow-rs that
-/// includes https://github.com/apache/arrow-rs/pull/6419
-#[inline]
-pub fn is_valid_decimal_precision(value: i128, precision: u8) -> bool {
- precision <= DECIMAL128_MAX_PRECISION
- && value >= MIN_DECIMAL128_FOR_EACH_PRECISION[precision as usize]
- && value <= MAX_DECIMAL128_FOR_EACH_PRECISION[precision as usize]
-}
-
/// Build a boolean buffer from the state and reset the state, based on the emit_to
/// strategy.
pub fn build_bool_state(state: &mut BooleanBufferBuilder, emit_to: &EmitTo) -> BooleanBuffer {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org