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/25 18:16:49 UTC
(datafusion-comet) 01/01: perf: track decimal overflow without rescanning results (#5044)
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-5044-8ca8ca6b82e4e72ad69d49fabf1e1d7e7ffd8795
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
commit e1f2ba3f21026bf246b9b87de1807036c8249cb7
Author: Peter Lee <pe...@gmail.com>
AuthorDate: Fri Sep 25 16:24:22 2026 +0000
perf: track decimal overflow without rescanning results (#5044)
* perf: skip wide decimal null masking when nothing overflows
* docs: explain wide decimal overflow guard
* perf: track wide decimal overflow during evaluation
* perf: avoid decimal rescale overflow scan
* refine comment
* docs: explain overflow guard invariant and Cell cost
Restores the safety-invariant comment from review at the surviving guard
site and documents why the Cell write costs 3-5% on overflow shapes.
Co-Authored-By: Claude Fable 5 <no...@anthropic.com>
---------
Co-authored-by: Claude Fable 5 <no...@anthropic.com>
---
native/spark-expr/benches/wide_decimal.rs | 76 ++++++++++++++++++----
.../src/math_funcs/wide_decimal_binary_expr.rs | 65 ++++++++++++++++--
2 files changed, 124 insertions(+), 17 deletions(-)
diff --git a/native/spark-expr/benches/wide_decimal.rs b/native/spark-expr/benches/wide_decimal.rs
index ec932ae68f..b38ddbfaa0 100644
--- a/native/spark-expr/benches/wide_decimal.rs
+++ b/native/spark-expr/benches/wide_decimal.rs
@@ -19,7 +19,7 @@
//! for Decimal128 arithmetic that requires wider intermediate precision.
use arrow::array::builder::Decimal128Builder;
-use arrow::array::RecordBatch;
+use arrow::array::{Array, Decimal128Array, RecordBatch};
use arrow::datatypes::{DataType, Field, Schema};
use criterion::{criterion_group, criterion_main, BenchmarkId, Criterion};
use datafusion::logical_expr::Operator;
@@ -28,6 +28,7 @@ use datafusion::physical_expr::PhysicalExpr;
use datafusion_comet_spark_expr::{
Cast, EvalMode, SparkCastOptions, WideDecimalBinaryExpr, WideDecimalOp,
};
+use std::hint::black_box;
use std::sync::Arc;
const BATCH_SIZE: usize = 8192;
@@ -49,6 +50,30 @@ fn make_decimal_batch(p1: u8, s1: i8, p2: u8, s2: i8) -> RecordBatch {
RecordBatch::try_new(Arc::new(schema), vec![Arc::new(left), Arc::new(right)]).unwrap()
}
+fn make_overflow_batch(null_every: usize, overflow_every: usize) -> RecordBatch {
+ let left: Decimal128Array = (0..BATCH_SIZE)
+ .map(|i| {
+ if null_every != 0 && i % null_every == 0 {
+ None
+ } else if overflow_every != 0 && (i + 1) % overflow_every == 0 {
+ Some(10_000_000_000)
+ } else {
+ Some((i as i128 % 100_000) * 100)
+ }
+ })
+ .collect::<Decimal128Array>()
+ .with_precision_and_scale(38, 2)
+ .unwrap();
+ let right = Decimal128Array::from_value(0, BATCH_SIZE)
+ .with_precision_and_scale(38, 2)
+ .unwrap();
+ let schema = Schema::new(vec![
+ Field::new("left", left.data_type().clone(), true),
+ Field::new("right", right.data_type().clone(), false),
+ ]);
+ RecordBatch::try_new(Arc::new(schema), vec![Arc::new(left), Arc::new(right)]).unwrap()
+}
+
/// Old approach: Cast(Decimal128->Decimal256) both sides, BinaryExpr, Cast(Decimal256->Decimal128).
fn build_old_expr(
p1: u8,
@@ -80,16 +105,16 @@ fn build_old_expr(
}
/// New approach: single fused WideDecimalBinaryExpr.
-fn build_new_expr(op: WideDecimalOp, p_out: u8, s_out: i8) -> Arc<dyn PhysicalExpr> {
+fn build_new_expr(
+ op: WideDecimalOp,
+ p_out: u8,
+ s_out: i8,
+ eval_mode: EvalMode,
+) -> Arc<dyn PhysicalExpr> {
let left_col: Arc<dyn PhysicalExpr> = Arc::new(Column::new("left", 0));
let right_col: Arc<dyn PhysicalExpr> = Arc::new(Column::new("right", 1));
Arc::new(WideDecimalBinaryExpr::new(
- left_col,
- right_col,
- op,
- p_out,
- s_out,
- EvalMode::Legacy,
+ left_col, right_col, op, p_out, s_out, eval_mode,
))
}
@@ -116,7 +141,7 @@ fn criterion_benchmark(c: &mut Criterion) {
{
let batch = make_decimal_batch(38, 10, 38, 10);
let old = build_old_expr(38, 10, 38, 10, Operator::Plus, DataType::Decimal128(38, 10));
- let new = build_new_expr(WideDecimalOp::Add, 38, 10);
+ let new = build_new_expr(WideDecimalOp::Add, 38, 10, EvalMode::Legacy);
bench_case(&mut group, "add_same_scale", &batch, &old, &new);
}
@@ -124,7 +149,7 @@ fn criterion_benchmark(c: &mut Criterion) {
{
let batch = make_decimal_batch(38, 6, 38, 4);
let old = build_old_expr(38, 6, 38, 4, Operator::Plus, DataType::Decimal128(38, 6));
- let new = build_new_expr(WideDecimalOp::Add, 38, 6);
+ let new = build_new_expr(WideDecimalOp::Add, 38, 6, EvalMode::Legacy);
bench_case(&mut group, "add_diff_scale", &batch, &old, &new);
}
@@ -140,7 +165,7 @@ fn criterion_benchmark(c: &mut Criterion) {
Operator::Multiply,
DataType::Decimal128(38, 6),
);
- let new = build_new_expr(WideDecimalOp::Multiply, 38, 6);
+ let new = build_new_expr(WideDecimalOp::Multiply, 38, 6, EvalMode::Legacy);
bench_case(&mut group, "multiply", &batch, &old, &new);
}
@@ -155,11 +180,38 @@ fn criterion_benchmark(c: &mut Criterion) {
Operator::Minus,
DataType::Decimal128(38, 18),
);
- let new = build_new_expr(WideDecimalOp::Subtract, 38, 18);
+ let new = build_new_expr(WideDecimalOp::Subtract, 38, 18, EvalMode::Legacy);
bench_case(&mut group, "subtract", &batch, &old, &new);
}
group.finish();
+
+ let no_overflow = make_overflow_batch(0, 0);
+ let no_overflow_nulls = make_overflow_batch(17, 0);
+ let sparse_overflow = make_overflow_batch(0, 17);
+ let dense_overflow = make_overflow_batch(0, 2);
+ let overflow_at_end = make_overflow_batch(0, BATCH_SIZE);
+ let legacy = build_new_expr(WideDecimalOp::Add, 10, 2, EvalMode::Legacy);
+ let ansi = build_new_expr(WideDecimalOp::Add, 10, 2, EvalMode::Ansi);
+
+ c.bench_function("wide_decimal: no overflow", |b| {
+ b.iter(|| black_box(legacy.evaluate(black_box(&no_overflow)).unwrap()))
+ });
+ c.bench_function("wide_decimal: no overflow, nulls", |b| {
+ b.iter(|| black_box(legacy.evaluate(black_box(&no_overflow_nulls)).unwrap()))
+ });
+ c.bench_function("wide_decimal: sparse overflow", |b| {
+ b.iter(|| black_box(legacy.evaluate(black_box(&sparse_overflow)).unwrap()))
+ });
+ c.bench_function("wide_decimal: dense overflow", |b| {
+ b.iter(|| black_box(legacy.evaluate(black_box(&dense_overflow)).unwrap()))
+ });
+ c.bench_function("wide_decimal: overflow at end of batch", |b| {
+ b.iter(|| black_box(legacy.evaluate(black_box(&overflow_at_end)).unwrap()))
+ });
+ c.bench_function("wide_decimal: ansi no overflow", |b| {
+ b.iter(|| black_box(ansi.evaluate(black_box(&no_overflow)).unwrap()))
+ });
}
criterion_group!(benches, criterion_benchmark);
diff --git a/native/spark-expr/src/math_funcs/wide_decimal_binary_expr.rs b/native/spark-expr/src/math_funcs/wide_decimal_binary_expr.rs
index ca4869357e..37c848c024 100644
--- a/native/spark-expr/src/math_funcs/wide_decimal_binary_expr.rs
+++ b/native/spark-expr/src/math_funcs/wide_decimal_binary_expr.rs
@@ -29,6 +29,7 @@ use arrow::record_batch::RecordBatch;
use datafusion::common::Result;
use datafusion::logical_expr::ColumnarValue;
use datafusion::physical_expr::PhysicalExpr;
+use std::cell::Cell;
use std::fmt::{Display, Formatter};
use std::hash::Hash;
use std::sync::Arc;
@@ -214,6 +215,13 @@ impl PhysicalExpr for WideDecimalBinaryExpr {
let bound = max_for_precision(p_out);
let neg_bound = i256::ZERO.wrapping_sub(bound);
+ // Records overflow during evaluation so no-overflow batches can skip the
+ // null-masking pass below. The `Cell` write is an observable side effect
+ // inside the kernel closure and costs 3-5% on overflow-bearing batches even
+ // when only a single row overflows, so the cost comes from inhibited
+ // optimization of the closure rather than the number of stores. #5309
+ // tracks removing the sentinel and `Cell` by writing null bits directly.
+ let overflowed = Cell::new(false);
let result: Decimal128Array = match op {
WideDecimalOp::Add | WideDecimalOp::Subtract => {
@@ -249,7 +257,7 @@ impl PhysicalExpr for WideDecimalBinaryExpr {
} else {
raw
};
- check_overflow_and_convert(result, bound, neg_bound, eval_mode)
+ check_overflow_and_convert(result, bound, neg_bound, eval_mode, &overflowed)
})?
}
WideDecimalOp::Multiply => {
@@ -276,12 +284,16 @@ impl PhysicalExpr for WideDecimalBinaryExpr {
} else {
raw
};
- check_overflow_and_convert(result, bound, neg_bound, eval_mode)
+ check_overflow_and_convert(result, bound, neg_bound, eval_mode, &overflowed)
})?
}
};
- let result = if eval_mode != EvalMode::Ansi {
+ let result = if overflowed.get() {
+ // Every non-sentinel value is already within ±(10^p_out - 1), so the extra
+ // null-masking pass can only null sentinels. Checking the overflow flag lets
+ // the common no-overflow case skip this pass and its allocation entirely.
+ // ANSI mode errors before setting the flag, so it also skips this pass.
result.null_if_overflow_precision(p_out)
} else {
result
@@ -329,20 +341,24 @@ impl PhysicalExpr for WideDecimalBinaryExpr {
}
/// Check if the i256 result fits in the output precision. In Ansi mode, return an error
-/// on overflow. In Legacy/Try mode, return i128::MAX as a sentinel value that will be
-/// nullified by `null_if_overflow_precision`.
+/// on overflow. In Legacy/Try mode, record the overflow and return i128::MAX as a sentinel
+/// value that will be nullified by `null_if_overflow_precision`.
#[inline]
fn check_overflow_and_convert(
result: i256,
bound: i256,
neg_bound: i256,
eval_mode: EvalMode,
+ overflowed: &Cell<bool>,
) -> Result<i128, ArrowError> {
if result > bound || result < neg_bound {
if eval_mode == EvalMode::Ansi {
return Err(ArrowError::ComputeError("Arithmetic overflow".to_string()));
}
+ // TODO: Write the null bit directly and remove the sentinel/masking pass.
+ // https://github.com/apache/datafusion-comet/issues/5309
// Sentinel value — will be nullified by null_if_overflow_precision
+ overflowed.set(true);
Ok(i128::MAX)
} else {
Ok(result.to_i128().unwrap())
@@ -506,6 +522,23 @@ mod tests {
assert!(arr.is_null(0));
}
+ #[test]
+ fn test_overflow_with_nulls_legacy_mode() {
+ let batch = make_batch(
+ vec![Some(4), Some(5), None],
+ 38,
+ 0,
+ vec![Some(5), Some(5), Some(1)],
+ 38,
+ 0,
+ );
+ let result = eval_expr(&batch, WideDecimalOp::Add, 1, 0, EvalMode::Legacy).unwrap();
+ let arr = result.as_primitive::<Decimal128Type>();
+ assert_eq!(arr.value(0), 9);
+ assert!(arr.is_null(1));
+ assert!(arr.is_null(2));
+ }
+
#[test]
fn test_overflow_ansi_mode_returns_error() {
let batch = make_batch(vec![Some(5)], 38, 0, vec![Some(5)], 38, 0);
@@ -620,6 +653,28 @@ mod tests {
}
}
+ #[test]
+ fn test_scalar_scalar_overflow_returns_null_scalar() {
+ use datafusion::common::ScalarValue;
+ use datafusion::physical_expr::expressions::Literal;
+
+ let value = ScalarValue::Decimal128(Some(5), 38, 0);
+ let expr = WideDecimalBinaryExpr::new(
+ Arc::new(Literal::new(value.clone())),
+ Arc::new(Literal::new(value)),
+ WideDecimalOp::Multiply,
+ 1,
+ 0,
+ EvalMode::Legacy,
+ );
+ let batch = RecordBatch::new_empty(Arc::new(Schema::empty()));
+
+ assert!(matches!(
+ expr.evaluate(&batch).unwrap(),
+ ColumnarValue::Scalar(ScalarValue::Decimal128(None, 1, 0))
+ ));
+ }
+
/// Companion test: when at least one input is an Array, the result must remain an Array.
/// Guards against over-eager scalar-unwrapping in the fix.
#[test]
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org