diff --git a/docs/source/contributor-guide/expression-audits/math_funcs.md b/docs/source/contributor-guide/expression-audits/math_funcs.md index f997480f57..6c9607efc5 100644 --- a/docs/source/contributor-guide/expression-audits/math_funcs.md +++ b/docs/source/contributor-guide/expression-audits/math_funcs.md @@ -119,6 +119,12 @@ Internal decimal wrapper emitted around every decimal `+ - * /`, `sum`, and `avg - Spark 3.4.3, 3.5.8, 4.0.1, 4.1.1 (audited 2026-05-27): `IntegralDivide(left, right, evalMode)`. Non-decimal operands are cast to `DecimalType(19, 0)`; result is recomputed per `IntegralDivide.resultDecimalType`, wrapped in `CheckOverflow`, then cast to `Long`. ANSI overflow for `Long.MinValue div -1` and decimal-overflow ANSI cases are covered by existing tests. +## DecimalRescaleCheckOverflow (internal) + +Internal fused expression that rescales a Decimal128 value (changing scale) and checks output precision in one pass, replacing the `CheckOverflow(Cast(expr, Decimal128(p, s)))` pattern used by decimal-to-decimal casts. Native impl: `math_funcs/internal/decimal_rescale_check.rs`. + +- Performance (tuned 2026-07-15, PR #4938): the legacy path ran `null_if_overflow_precision` (a second full pass that allocates a new array) on every batch to turn overflow sentinels into nulls, even when nothing overflowed. Now that pass runs only when a sentinel is present (`contains(&i128::MAX)`, short-circuiting), so the common no-overflow case skips the allocation. 8 to 26% faster on no-overflow shapes; overflow and ANSI shapes unchanged. Benchmark: `benches/decimal_rescale.rs`. + ## e - Foldable; rewritten to a literal by ConstantFolding (like `pi`). diff --git a/native/spark-expr/Cargo.toml b/native/spark-expr/Cargo.toml index e3a9026053..63df30c916 100644 --- a/native/spark-expr/Cargo.toml +++ b/native/spark-expr/Cargo.toml @@ -148,6 +148,10 @@ harness = false name = "to_json" harness = false +[[bench]] +name = "decimal_rescale" +harness = false + [[bench]] name = "check_overflow" harness = false diff --git a/native/spark-expr/benches/decimal_rescale.rs b/native/spark-expr/benches/decimal_rescale.rs new file mode 100644 index 0000000000..03eb355f37 --- /dev/null +++ b/native/spark-expr/benches/decimal_rescale.rs @@ -0,0 +1,93 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use arrow::array::{Array, Decimal128Array}; +use arrow::datatypes::{Field, Schema}; +use arrow::record_batch::RecordBatch; +use criterion::{criterion_group, criterion_main, Criterion}; +use datafusion::physical_expr::expressions::Column; +use datafusion::physical_expr::PhysicalExpr; +use datafusion_comet_spark_expr::DecimalRescaleCheckOverflow; +use std::hint::black_box; +use std::sync::Arc; + +const IN_PRECISION: u8 = 38; +const IN_SCALE: i8 = 2; +// A value that overflows the narrow output precision used in the overflow shapes. +const OVERFLOW_VALUE: i128 = 1_000_000_000_000; + +/// Build a Decimal128(38, 2) column of `size` rows. +/// - `null_every`: null every Nth row (0 = none). +/// - `overflow_every`: make every Nth value large enough to overflow the narrow output (0 = none). +fn build(size: usize, null_every: usize, overflow_every: usize) -> RecordBatch { + let arr: Decimal128Array = (0..size) + .map(|i| { + if null_every != 0 && i % null_every == 0 { + None + } else if overflow_every != 0 && i % overflow_every == 0 { + Some(OVERFLOW_VALUE) + } else { + Some((i as i128 % 100_000) * 100) + } + }) + .collect::() + .with_precision_and_scale(IN_PRECISION, IN_SCALE) + .unwrap(); + let schema = Schema::new(vec![Field::new("col", arr.data_type().clone(), true)]); + RecordBatch::try_new(Arc::new(schema), vec![Arc::new(arr)]).unwrap() +} + +fn criterion_benchmark(c: &mut Criterion) { + let size = 8192; + let col = || Arc::new(Column::new("col", 0)) as Arc; + + // Widen scale 2 -> 6 into a roomy precision: the common decimal-cast case, no overflow. + let scale_up = DecimalRescaleCheckOverflow::new(col(), IN_SCALE, 38, 6, false); + // Narrow scale 2 -> 0 (HALF_UP rounding), roomy precision, no overflow. + let scale_down = DecimalRescaleCheckOverflow::new(col(), IN_SCALE, 38, 0, false); + // Scale up into a narrow precision so large values overflow. + let narrow = DecimalRescaleCheckOverflow::new(col(), IN_SCALE, 12, 6, false); + // ANSI widening, no overflow. + let ansi = DecimalRescaleCheckOverflow::new(col(), IN_SCALE, 38, 6, true); + + let no_overflow = build(size, 0, 0); + let no_overflow_nulls = build(size, 17, 0); + let sparse_overflow = build(size, 0, 17); + let dense_overflow = build(size, 0, 2); + + c.bench_function("decimal_rescale: scale up, no overflow", |b| { + b.iter(|| black_box(scale_up.evaluate(black_box(&no_overflow)).unwrap())) + }); + c.bench_function("decimal_rescale: scale up, no overflow, nulls", |b| { + b.iter(|| black_box(scale_up.evaluate(black_box(&no_overflow_nulls)).unwrap())) + }); + c.bench_function("decimal_rescale: scale down, no overflow", |b| { + b.iter(|| black_box(scale_down.evaluate(black_box(&no_overflow)).unwrap())) + }); + c.bench_function("decimal_rescale: sparse overflow", |b| { + b.iter(|| black_box(narrow.evaluate(black_box(&sparse_overflow)).unwrap())) + }); + c.bench_function("decimal_rescale: dense overflow", |b| { + b.iter(|| black_box(narrow.evaluate(black_box(&dense_overflow)).unwrap())) + }); + c.bench_function("decimal_rescale: ansi scale up, no overflow", |b| { + b.iter(|| black_box(ansi.evaluate(black_box(&no_overflow)).unwrap())) + }); +} + +criterion_group!(benches, criterion_benchmark); +criterion_main!(benches); diff --git a/native/spark-expr/src/math_funcs/internal/decimal_rescale_check.rs b/native/spark-expr/src/math_funcs/internal/decimal_rescale_check.rs index 667f76939a..fea2399202 100644 --- a/native/spark-expr/src/math_funcs/internal/decimal_rescale_check.rs +++ b/native/spark-expr/src/math_funcs/internal/decimal_rescale_check.rs @@ -197,7 +197,13 @@ impl PhysicalExpr for DecimalRescaleCheckOverflow { rescale_and_check(value, delta, scale_factor, bound, fail_on_error) })?; - let result = if !fail_on_error { + let result = if !fail_on_error && result.values().contains(&i128::MAX) { + // The rescale pass writes i128::MAX as an overflow sentinel for values that + // do not fit the output precision. Only when a sentinel is present do we need + // the extra null-masking pass (which allocates a new array); `contains` + // short-circuits at the first sentinel, so the common no-overflow case skips + // that allocation entirely. ANSI mode raises on overflow and never produces a + // sentinel, so it also skips this pass. result.null_if_overflow_precision(p_out) } else { result @@ -343,6 +349,42 @@ mod tests { assert!(result.is_err()); } + #[test] + fn test_overflow_with_nulls_legacy() { + // Mixes valid, overflowing, and null inputs so the sentinel fallback path runs with + // nulls present: overflow and null both yield null, valid values are preserved. + let batch = make_batch(vec![Some(150), Some(10_000), None, Some(250)], 10, 2); + let result = eval_expr(&batch, 2, 4, 2, false).unwrap(); + let arr = result.as_primitive::(); + assert_eq!(arr.value(0), 150); + assert!(arr.is_null(1)); // 10000 > 9999 (max for precision 4) -> null + assert!(arr.is_null(2)); // input null stays null + assert_eq!(arr.value(3), 250); + } + + #[test] + fn test_all_values_overflow_legacy() { + // Every value overflows, so the sentinel sits at index 0: `contains` finds it immediately + // and the masking pass nulls the whole array. + let batch = make_batch(vec![Some(10_000), Some(20_000), Some(30_000)], 10, 2); + let result = eval_expr(&batch, 2, 4, 2, false).unwrap(); + let arr = result.as_primitive::(); + assert!(arr.is_null(0)); // all > 9999 (max for precision 4) -> null + assert!(arr.is_null(1)); + assert!(arr.is_null(2)); + } + + #[test] + fn test_precision_boundary_legacy() { + // Pins the exact bound the overflow predicate turns on: 10^p - 1 fits, 10^p overflows. + // Output precision 4, scale 0: 9999 (= 10^4 - 1) fits, 10000 (= 10^4) does not. + let batch = make_batch(vec![Some(9999), Some(10_000)], 10, 0); + let result = eval_expr(&batch, 0, 4, 0, false).unwrap(); + let arr = result.as_primitive::(); + assert_eq!(arr.value(0), 9999); // fits precision 4 + assert!(arr.is_null(1)); // overflows precision 4 -> null + } + #[test] fn test_null_propagation() { let batch = make_batch(vec![Some(100), None, Some(200)], 10, 2);