This is an automated email from the ASF dual-hosted git repository.
andygrove pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
The following commit(s) were added to refs/heads/main by this push:
new 4cfd0739c3 perf: vectorize integer-to-decimal cast (#4939)
4cfd0739c3 is described below
commit 4cfd0739c353df378d2d0a1d4fcb4d6f5dd987e7
Author: Andy Grove <[email protected]>
AuthorDate: Wed Aug 5 06:44:11 2026 -0600
perf: vectorize integer-to-decimal cast (#4939)
* perf: vectorize integer-to-decimal cast
cast_int_to_decimal128_internal used a per-element Decimal128Builder loop
with a
null check and branch per row. Replace it with a single vectorized unary_opt
pass: a value that overflows the multiply or does not fit the output
precision
maps to null, and the input null buffer is carried over.
ANSI mode must raise on out-of-range values rather than nulling them.
unary_opt
only nulls non-null inputs that overflow, so a null count beyond the input's
signals an overflow; that O(1) check gates a rare element-wise rescan that
reports the first offending value with Spark's exact error.
This is a single pass for every eval mode and shape, so all shapes are
faster
(28-50%) with no regression, including the overflow case. Add unit tests
for the
legacy null-on-overflow, ANSI no-overflow, and ANSI overflow-error paths,
plus a
criterion benchmark.
Part of #4936.
* docs: record integer-to-decimal cast performance audit entry
* refactor: share fit-at-precision predicate; test EvalMode::Try; fix
outdated comment
Extracts the 'multiply-then-fits-target-precision' predicate into a single
'fits' closure so the vectorized pass and the ANSI rescan cannot drift out of
sync. Adds test_cast_int_to_decimal128_overflow_try_nulls asserting
EvalMode::Try shares the null-on-overflow branch. Replaces the outdated
'sentinel + masking path' comment on the Legacy test with an accurate
description of the null-on-overflow path.
* test: pin ANSI overflow error variant and first-offending-value scan order
test_cast_int_to_decimal128_overflow_ansi_errors asserted only is_err(), so
a
refactor that scanned in reverse or returned a different variant would still
pass. Match on SparkError::NumericValueOutOfRange with its precision, scale
and
value, and add a second overflowing value after the first so the assertion
also
pins the documented 'reports the first offending value' behavior.
---
.../expression-audits/conversion_funcs.md | 1 +
native/spark-expr/Cargo.toml | 3 +
native/spark-expr/benches/cast_int_to_decimal.rs | 98 +++++++++++++
native/spark-expr/src/conversion_funcs/numeric.rs | 159 ++++++++++++++++-----
4 files changed, 222 insertions(+), 39 deletions(-)
diff --git
a/docs/source/contributor-guide/expression-audits/conversion_funcs.md
b/docs/source/contributor-guide/expression-audits/conversion_funcs.md
index 64e01822dc..ea9a9bf737 100644
--- a/docs/source/contributor-guide/expression-audits/conversion_funcs.md
+++ b/docs/source/contributor-guide/expression-audits/conversion_funcs.md
@@ -37,5 +37,6 @@
- Spark registers the type-name conversion functions (`bigint`, `binary`,
`boolean`, `date`, `decimal`, `double`, `float`, `int`, `smallint`, `string`,
`timestamp`, `tinyint`) as cast aliases. Each lowers to the same `Cast` node,
so Comet handles it via the `cast` implementation with the same compatibility
profile.
- Performance (tuned 2026-07-14, PR
[#4920](https://github.com/apache/datafusion-comet/pull/4920)): narrowing
integer casts (`spark_cast_int_to_int`) map the values buffer in a single pass
with Arrow `unary`/`try_unary` and carry the null buffer over untouched,
replacing an element-by-element `Option`/`Result` iterator-collect. Up to 100x
faster on narrowing casts. Benchmark: `benches/cast_numeric.rs`.
- Performance (tuned 2026-07-15, PR
[#4940](https://github.com/apache/datafusion-comet/pull/4940)):
float/double-to-decimal casts (`cast_floating_point_to_decimal128`) now convert
in a single vectorized `unary_opt` pass that maps out-of-range values (NaN,
infinity, precision overflow) to null, replacing the per-element
`Decimal128Builder` loop. ANSI raises via an O(1) null-count check plus a rare
element-wise rescan. 15-36% faster with no regression on any shape. Benchmark:
`benches/cast [...]
+- Performance (tuned 2026-07-15, PR
[#4939](https://github.com/apache/datafusion-comet/pull/4939)):
integer-to-decimal casts (`cast_int_to_decimal128_internal`) now convert in a
single vectorized `unary_opt` pass that maps overflowing values to null,
replacing the per-element `Decimal128Builder` loop. ANSI raises via an O(1)
null-count check plus a rare element-wise rescan. 28-50% faster with no
regression on any shape. Benchmark: `benches/cast_int_to_decimal.rs`.
[Spark Expression Support]: ../../user-guide/latest/expressions.md
diff --git a/native/spark-expr/Cargo.toml b/native/spark-expr/Cargo.toml
index b85153a59d..cd7cd89849 100644
--- a/native/spark-expr/Cargo.toml
+++ b/native/spark-expr/Cargo.toml
@@ -215,3 +215,6 @@ harness = false
name = "unscaled_value"
harness = false
+[[bench]]
+name = "cast_int_to_decimal"
+harness = false
\ No newline at end of file
diff --git a/native/spark-expr/benches/cast_int_to_decimal.rs
b/native/spark-expr/benches/cast_int_to_decimal.rs
new file mode 100644
index 0000000000..8949c5c327
--- /dev/null
+++ b/native/spark-expr/benches/cast_int_to_decimal.rs
@@ -0,0 +1,98 @@
+// 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::{Int32Array, Int64Array, RecordBatch};
+use arrow::datatypes::{DataType, Field, Schema};
+use criterion::{criterion_group, criterion_main, Criterion};
+use datafusion::physical_expr::{expressions::Column, PhysicalExpr};
+use datafusion_comet_spark_expr::{Cast, EvalMode, SparkCastOptions};
+use std::hint::black_box;
+use std::sync::Arc;
+
+fn i32_batch(size: usize, null_every: usize) -> RecordBatch {
+ let a: Int32Array = (0..size)
+ .map(|i| {
+ if null_every != 0 && i % null_every == 0 {
+ None
+ } else {
+ Some((i as i32) % 100_000)
+ }
+ })
+ .collect();
+ let schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Int32,
true)]));
+ RecordBatch::try_new(schema, vec![Arc::new(a)]).unwrap()
+}
+
+fn i64_batch(size: usize, big: bool) -> RecordBatch {
+ let a: Int64Array = (0..size)
+ .map(|i| {
+ if big {
+ // Large enough that value * 10^4 overflows Decimal128(15, 4).
+ Some(1_000_000_000_000_000_i64 + i as i64)
+ } else {
+ Some((i as i64) % 100_000)
+ }
+ })
+ .collect();
+ let schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Int64,
true)]));
+ RecordBatch::try_new(schema, vec![Arc::new(a)]).unwrap()
+}
+
+fn cast(col: &str, to: DataType, mode: EvalMode) -> Cast {
+ Cast::new(
+ Arc::new(Column::new(col, 0)),
+ to,
+ SparkCastOptions::new_without_timezone(mode, false),
+ None,
+ None,
+ )
+}
+
+fn criterion_benchmark(c: &mut Criterion) {
+ let size = 8192;
+ let dec_15_4 = DataType::Decimal128(15, 4);
+ let dec_38_4 = DataType::Decimal128(38, 4);
+
+ let i32_no_nulls = i32_batch(size, 0);
+ let i32_nulls = i32_batch(size, 10);
+ let i64_small = i64_batch(size, false);
+ let i64_big = i64_batch(size, true);
+
+ let c_i32 = cast("a", dec_15_4.clone(), EvalMode::Legacy);
+ let c_i32_ansi = cast("a", dec_15_4.clone(), EvalMode::Ansi);
+ let c_i64 = cast("a", dec_38_4, EvalMode::Legacy);
+ let c_i64_overflow = cast("a", dec_15_4, EvalMode::Legacy);
+
+ c.bench_function("cast_int_to_decimal: i32 -> dec(15,4)", |b| {
+ b.iter(|| black_box(c_i32.evaluate(black_box(&i32_no_nulls)).unwrap()))
+ });
+ c.bench_function("cast_int_to_decimal: i32 -> dec(15,4), nulls", |b| {
+ b.iter(|| black_box(c_i32.evaluate(black_box(&i32_nulls)).unwrap()))
+ });
+ c.bench_function("cast_int_to_decimal: i64 -> dec(38,4)", |b| {
+ b.iter(|| black_box(c_i64.evaluate(black_box(&i64_small)).unwrap()))
+ });
+ c.bench_function("cast_int_to_decimal: i32 -> dec(15,4) ansi", |b| {
+ b.iter(||
black_box(c_i32_ansi.evaluate(black_box(&i32_no_nulls)).unwrap()))
+ });
+ c.bench_function("cast_int_to_decimal: i64 -> dec(15,4) overflow", |b| {
+ b.iter(||
black_box(c_i64_overflow.evaluate(black_box(&i64_big)).unwrap()))
+ });
+}
+
+criterion_group!(benches, criterion_benchmark);
+criterion_main!(benches);
diff --git a/native/spark-expr/src/conversion_funcs/numeric.rs
b/native/spark-expr/src/conversion_funcs/numeric.rs
index 90fc5ffc5b..029a38e3e7 100644
--- a/native/spark-expr/src/conversion_funcs/numeric.rs
+++ b/native/spark-expr/src/conversion_funcs/numeric.rs
@@ -20,9 +20,9 @@ use crate::conversion_funcs::utils::cast_overflow;
use crate::conversion_funcs::utils::MICROS_PER_SECOND;
use crate::{EvalMode, SparkError, SparkResult};
use arrow::array::{
- Array, ArrayRef, AsArray, BooleanBuilder, Decimal128Array,
Decimal128Builder, Float32Array,
- Float64Array, GenericStringBuilder, Int16Array, Int32Array, Int64Array,
Int8Array,
- OffsetSizeTrait, PrimitiveArray, StringBuilder,
TimestampMicrosecondBuilder,
+ Array, ArrayRef, AsArray, BooleanBuilder, Decimal128Array, Float32Array,
Float64Array,
+ GenericStringBuilder, Int16Array, Int32Array, Int64Array, Int8Array,
OffsetSizeTrait,
+ PrimitiveArray, StringBuilder, TimestampMicrosecondBuilder,
};
use arrow::datatypes::{
i256, is_validate_decimal_precision, ArrowPrimitiveType, DataType,
Decimal128Type, Float32Type,
@@ -734,48 +734,40 @@ where
T: ArrowPrimitiveType,
T::Native: Into<i128>,
{
- let mut builder = Decimal128Builder::with_capacity(array.len());
let multiplier = 10_i128.pow(scale as u32);
- for i in 0..array.len() {
- if array.is_null(i) {
- builder.append_null();
- } else {
- let v = array.value(i).into();
- let scaled = v.checked_mul(multiplier);
- match scaled {
- Some(scaled) => {
- if !is_validate_decimal_precision(scaled, precision) {
- match eval_mode {
- EvalMode::Ansi => {
- return Err(SparkError::NumericValueOutOfRange {
- value: v.to_string(),
- precision,
- scale,
- });
- }
- EvalMode::Try | EvalMode::Legacy =>
builder.append_null(),
- }
- } else {
- builder.append_value(scaled);
- }
+ // Single spelling of the "does this value fit at the target precision
after scaling" check,
+ // shared between the vectorized pass and the ANSI rescan below.
+ let fits = |v: i128| -> Option<i128> {
+ v.checked_mul(multiplier)
+ .filter(|scaled| is_validate_decimal_precision(*scaled, precision))
+ };
+
+ // Single vectorized pass: a value that overflows the multiply or does not
fit the output
+ // precision maps to null. `unary_opt` only applies the closure to
non-null slots and carries
+ // the input null buffer over, replacing the per-element builder loop
without a second pass.
+ let result: Decimal128Array = array.unary_opt::<_, Decimal128Type>(|v|
fits(v.into()));
+
+ // ANSI must raise on out-of-range values instead of nulling them.
`unary_opt` only nulls
+ // non-null inputs that overflow, so a null count beyond the input's
signals an overflow to
+ // report. This check is O(1); the element-wise rescan runs only on the
rare error path and
+ // reports the first offending value with Spark's exact error.
+ if eval_mode == EvalMode::Ansi && result.null_count() > array.null_count()
{
+ for i in 0..array.len() {
+ if !array.is_null(i) {
+ let v: i128 = array.value(i).into();
+ if fits(v).is_none() {
+ return Err(SparkError::NumericValueOutOfRange {
+ value: v.to_string(),
+ precision,
+ scale,
+ });
}
- _ => match eval_mode {
- EvalMode::Ansi => {
- return Err(SparkError::NumericValueOutOfRange {
- value: v.to_string(),
- precision,
- scale,
- })
- }
- EvalMode::Legacy | EvalMode::Try => builder.append_null(),
- },
}
}
}
- Ok(Arc::new(
- builder.with_precision_and_scale(precision, scale)?.finish(),
- ))
+
+ Ok(Arc::new(result.with_precision_and_scale(precision, scale)?))
}
pub(crate) fn cast_int_to_decimal128(
@@ -1322,6 +1314,95 @@ mod tests {
assert_eq!(decimal_array.value(1), -10000); // -100 * 10^2
assert!(decimal_array.is_null(2));
}
+
+ #[test]
+ fn test_cast_int_to_decimal128_overflow_legacy_nulls() {
+ // 1000 * 10^2 = 100000 does not fit precision 3 -> null (legacy).
Valid values and the
+ // input null are preserved, exercising the vectorized
null-on-overflow path.
+ let array: ArrayRef = Arc::new(Int32Array::from(vec![Some(9),
Some(1000), None, Some(-9)]));
+ let result = cast_int_to_decimal128(
+ &array,
+ EvalMode::Legacy,
+ &DataType::Int32,
+ &DataType::Decimal128(3, 2),
+ 3,
+ 2,
+ )
+ .unwrap();
+ let d = result.as_primitive::<Decimal128Type>();
+ assert_eq!(d.value(0), 900); // 9.00
+ assert!(d.is_null(1)); // overflow -> null
+ assert!(d.is_null(2)); // input null preserved
+ assert_eq!(d.value(3), -900);
+ assert_eq!(d.data_type(), &DataType::Decimal128(3, 2));
+ }
+
+ #[test]
+ fn test_cast_int_to_decimal128_overflow_try_nulls() {
+ // Try shares the Legacy null-on-overflow branch but is a distinct
enum arm; assert it
+ // explicitly so a future refactor cannot regress it silently.
+ let array: ArrayRef = Arc::new(Int32Array::from(vec![Some(9),
Some(1000), None, Some(-9)]));
+ let result = cast_int_to_decimal128(
+ &array,
+ EvalMode::Try,
+ &DataType::Int32,
+ &DataType::Decimal128(3, 2),
+ 3,
+ 2,
+ )
+ .unwrap();
+ let d = result.as_primitive::<Decimal128Type>();
+ assert_eq!(d.value(0), 900);
+ assert!(d.is_null(1));
+ assert!(d.is_null(2));
+ assert_eq!(d.value(3), -900);
+ }
+
+ #[test]
+ fn test_cast_int_to_decimal128_no_overflow_ansi() {
+ let array: ArrayRef = Arc::new(Int32Array::from(vec![Some(9), None,
Some(-9)]));
+ let result = cast_int_to_decimal128(
+ &array,
+ EvalMode::Ansi,
+ &DataType::Int32,
+ &DataType::Decimal128(3, 2),
+ 3,
+ 2,
+ )
+ .unwrap();
+ let d = result.as_primitive::<Decimal128Type>();
+ assert_eq!(d.value(0), 900);
+ assert!(d.is_null(1));
+ assert_eq!(d.value(2), -900);
+ }
+
+ #[test]
+ fn test_cast_int_to_decimal128_overflow_ansi_errors() {
+ // Two overflowing values: the rescan is documented to report the
first one, so asserting
+ // on 1000 rather than 2000 pins the scan order as well as the error
variant and payload.
+ let array: ArrayRef = Arc::new(Int32Array::from(vec![Some(9),
Some(1000), Some(2000)]));
+ let result = cast_int_to_decimal128(
+ &array,
+ EvalMode::Ansi,
+ &DataType::Int32,
+ &DataType::Decimal128(3, 2),
+ 3,
+ 2,
+ );
+ let err = result.unwrap_err();
+ assert!(
+ matches!(
+ err,
+ SparkError::NumericValueOutOfRange {
+ ref value,
+ precision: 3,
+ scale: 2
+ } if value == "1000"
+ ),
+ "unexpected error: {err:?}"
+ );
+ }
+
#[test]
fn test_cast_int_to_timestamp() {
let timezones: [Option<Arc<str>>; 6] = [
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
