perf: vectorize integer-to-decimal cast

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_float_to_decimal.rs`.
+- 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

native/spark-expr/Cargo.toml

@@ -215,3 +215,6 @@ harness = false
 name = "unscaled_value"
 harness = false
 
+[[bench]]
+name = "cast_int_to_decimal"
+harness = false

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);

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] = [