From f06d2cec610358be37a6d36da8b4a486466cb54f Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Wed, 15 Jul 2026 10:35:28 -0600 Subject: [PATCH 1/2] 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. --- native/spark-expr/Cargo.toml | 4 + .../spark-expr/benches/cast_int_to_decimal.rs | 98 ++++++++++++++ .../src/conversion_funcs/numeric.rs | 125 ++++++++++++------ 3 files changed, 188 insertions(+), 39 deletions(-) create mode 100644 native/spark-expr/benches/cast_int_to_decimal.rs diff --git a/native/spark-expr/Cargo.toml b/native/spark-expr/Cargo.toml index 6f90181ee0..0615c64e72 100644 --- a/native/spark-expr/Cargo.toml +++ b/native/spark-expr/Cargo.toml @@ -139,3 +139,7 @@ harness = false [[bench]] name = "to_json" harness = false + +[[bench]] +name = "cast_int_to_decimal" +harness = false 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 ce2fe515b6..0c8517ecc4 100644 --- a/native/spark-expr/src/conversion_funcs/numeric.rs +++ b/native/spark-expr/src/conversion_funcs/numeric.rs @@ -19,9 +19,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, GenericStringArray, Int16Array, Int32Array, Int64Array, Int8Array, - OffsetSizeTrait, PrimitiveArray, StringBuilder, TimestampMicrosecondBuilder, + Array, ArrayRef, AsArray, BooleanBuilder, Decimal128Array, Float32Array, Float64Array, + GenericStringArray, Int16Array, Int32Array, Int64Array, Int8Array, OffsetSizeTrait, + PrimitiveArray, StringBuilder, TimestampMicrosecondBuilder, }; use arrow::datatypes::{ i256, is_validate_decimal_precision, ArrowPrimitiveType, DataType, Decimal128Type, Float32Type, @@ -663,48 +663,41 @@ where T: ArrowPrimitiveType, T::Native: Into, { - 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 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| { + let v: i128 = v.into(); + v.checked_mul(multiplier) + .filter(|scaled| is_validate_decimal_precision(*scaled, precision)) + }); + + // 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(); + let overflows = v + .checked_mul(multiplier) + .map(|scaled| !is_validate_decimal_precision(scaled, precision)) + .unwrap_or(true); + if overflows { + 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( @@ -1160,6 +1153,60 @@ 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 sentinel + masking 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::(); + 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_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::(); + 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() { + let array: ArrayRef = Arc::new(Int32Array::from(vec![Some(9), Some(1000)])); + let result = cast_int_to_decimal128( + &array, + EvalMode::Ansi, + &DataType::Int32, + &DataType::Decimal128(3, 2), + 3, + 2, + ); + assert!(result.is_err()); + } #[test] fn test_cast_int_to_timestamp() { let timezones: [Option>; 6] = [ From d62d695179be92c325b5dd5963c66dd015423aa5 Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Wed, 15 Jul 2026 10:35:28 -0600 Subject: [PATCH 2/2] docs: record integer-to-decimal cast performance audit entry --- .../contributor-guide/expression-audits/conversion_funcs.md | 1 + 1 file changed, 1 insertion(+) diff --git a/docs/source/contributor-guide/expression-audits/conversion_funcs.md b/docs/source/contributor-guide/expression-audits/conversion_funcs.md index 45b6a8f03b..8f53dbf9b8 100644 --- a/docs/source/contributor-guide/expression-audits/conversion_funcs.md +++ b/docs/source/contributor-guide/expression-audits/conversion_funcs.md @@ -35,5 +35,6 @@ - `spark.sql.legacy.castComplexTypesToString.enabled=true` is not honoured by Comet (https://github.com/apache/datafusion-comet/issues/4492). - `CAST( AS DECIMAL)` rounding may differ from Spark (`Incompatible`, gated by `spark.comet.expression.Cast.allowIncompatible`, tracked at https://github.com/apache/datafusion-comet/issues/1371). - 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-15, PR #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