fix: native ansi-errors raised as spark error

native/core/src/execution/expressions/arithmetic.rs

@@ -45,6 +45,10 @@ impl CheckedBinaryExpr {
             query_context,
         }
     }
+
+    pub(crate) fn child(&self) -> &Arc<dyn PhysicalExpr> {
+        &self.child
+    }
 }
 
 impl Display for CheckedBinaryExpr {

native/core/src/execution/planner.rs

@@ -101,6 +101,7 @@ use datafusion::physical_expr::expressions::{Literal, StatsType};
 use datafusion::physical_expr::window::WindowExpr;
 use datafusion::physical_expr::LexOrdering;
 
+use crate::execution::expressions::arithmetic::CheckedBinaryExpr;
 use crate::parquet::parquet_exec::init_datasource_exec;
 use arrow::array::{
     new_empty_array, Array, ArrayRef, BinaryBuilder, BooleanArray, Date32Array, Decimal128Array,
@@ -517,18 +518,41 @@ impl PhysicalPlanner {
                     self.create_expr(expr.child.as_ref().unwrap(), Arc::clone(&input_schema))?;
                 let data_type = to_arrow_datatype(expr.datatype.as_ref().unwrap());
                 let fail_on_error = expr.fail_on_error;
+                let query_context = spark_expr.expr_id.and_then(|expr_id| {
+                    let registry = &self.query_context_registry;
+                    registry.get(expr_id)
+                });
 
                 // WideDecimalBinaryExpr already handles overflow — skip redundant check
-                // but only if its output type matches CheckOverflow's declared type
-                if child.downcast_ref::<WideDecimalBinaryExpr>().is_some() {
+                // but only if its output type matches CheckOverflow's declared type. A binary
+                // expression with query context is already wrapped in CheckedBinaryExpr, so
+                // inspect through that single layer too.
+                let is_wide_decimal = child.downcast_ref::<WideDecimalBinaryExpr>().is_some()
+                    || child
+                        .downcast_ref::<CheckedBinaryExpr>()
+                        .is_some_and(|checked| {
+                            checked
+                                .child()
+                                .downcast_ref::<WideDecimalBinaryExpr>()
+                                .is_some()
+                        });
+                if is_wide_decimal {
                     let child_type = child.data_type(&input_schema)?;
                     if child_type == data_type {
-                        return Ok(child);
+                        return if query_context.is_some()
+                            && child.downcast_ref::<CheckedBinaryExpr>().is_none()
+                        {
+                            Ok(Arc::new(CheckedBinaryExpr::new(child, query_context)))
+                        } else {
+                            Ok(child)
+                        };
                     }
                 }
 
                 // Fuse Cast(Decimal128→Decimal128) + CheckOverflow into single rescale+check
-                // Only fuse when the Cast target type matches the CheckOverflow output type
+                // Only fuse when the Cast target type matches the CheckOverflow output type.
+                // Spark 3.4+ does not currently emit this shape, but keep its errors typed and
+                // contextualized in case a future serializer makes the fusion reachable.
                 if let Some(cast) = child.downcast_ref::<Cast>() {
                     if let (
                         DataType::Decimal128(p_out, s_out),
@@ -537,23 +561,33 @@ impl PhysicalPlanner {
                     {
                         let cast_target = cast.data_type(&input_schema)?;
                         if cast_target == data_type {
-                            return Ok(Arc::new(DecimalRescaleCheckOverflow::new(
-                                Arc::clone(&cast.child),
-                                s_in,
-                                *p_out,
-                                *s_out,
-                                fail_on_error,
-                            )));
+                            let fused: Arc<dyn PhysicalExpr> =
+                                Arc::new(DecimalRescaleCheckOverflow::new(
+                                    Arc::clone(&cast.child),
+                                    s_in,
+                                    *p_out,
+                                    *s_out,
+                                    fail_on_error,
+                                ));
+                            return if query_context.is_some() {
+                                Ok(Arc::new(CheckedBinaryExpr::new(fused, query_context)))
+                            } else {
+                                Ok(fused)
+                            };
                         }
                     }
                 }
 
-                // Look up query context from registry if expr_id is present
-                let query_context = spark_expr.expr_id.and_then(|expr_id| {
-                    let registry = &self.query_context_registry;
-                    registry.get(expr_id)
-                });
-
+                // Generated child protos may not carry an expression id of their own, so retain
+                // the outer CheckOverflow context as a fallback for bare child SparkErrors.
+                let child = if query_context.is_some()
+                    && child.downcast_ref::<CheckedBinaryExpr>().is_none()
+                {
+                    Arc::new(CheckedBinaryExpr::new(child, query_context.clone()))
+                        as Arc<dyn PhysicalExpr>
+                } else {
+                    child
+                };
                 Ok(Arc::new(CheckOverflow::new(
                     child,
                     data_type,
@@ -933,9 +967,14 @@ impl PhysicalPlanner {
                     DataFusionOperator::Multiply => WideDecimalOp::Multiply,
                     _ => unreachable!(),
                 };
-                Ok(Arc::new(WideDecimalBinaryExpr::new(
+                let expr: Arc<dyn PhysicalExpr> = Arc::new(WideDecimalBinaryExpr::new(
                     left, right, wide_op, p_out, s_out, eval_mode,
-                )))
+                ));
+                if query_context.is_some() {
+                    Ok(Arc::new(CheckedBinaryExpr::new(expr, query_context)))
+                } else {
+                    Ok(expr)
+                }
             }
             (
                 DataFusionOperator::Divide,
@@ -960,13 +999,18 @@ impl PhysicalPlanner {
                     Some(options.check_divide_overflow),
                     eval_mode,
                 )?;
-                Ok(Arc::new(ScalarFunctionExpr::new(
+                let expr: Arc<dyn PhysicalExpr> = Arc::new(ScalarFunctionExpr::new(
                     func_name,
                     fun_expr,
                     vec![left, right],
                     Arc::new(Field::new(func_name, data_type, true)),
                     Arc::new(ConfigOptions::default()),
-                )))
+                ));
+                if query_context.is_some() {
+                    Ok(Arc::new(CheckedBinaryExpr::new(expr, query_context)))
+                } else {
+                    Ok(expr)
+                }
             }
             // Date +/- Int8/Int16/Int32: DataFusion 52's arrow-arith kernels only
             // support Date32 +/- Interval types, not raw integers. Use the Spark
@@ -1033,9 +1077,11 @@ impl PhysicalPlanner {
                         Arc::new(ConfigOptions::default()),
                     ));
 
-                    // Wrap with CheckedBinaryExpr to add query_context to errors
-                    use crate::execution::expressions::arithmetic::CheckedBinaryExpr;
-                    Ok(Arc::new(CheckedBinaryExpr::new(scalar_expr, query_context)))
+                    if query_context.is_some() {
+                        Ok(Arc::new(CheckedBinaryExpr::new(scalar_expr, query_context)))
+                    } else {
+                        Ok(scalar_expr)
+                    }
                 } else {
                     Ok(Arc::new(BinaryExpr::new(left, op, right)))
                 }

native/spark-expr/src/conversion_funcs/cast.rs

@@ -45,15 +45,17 @@ use arrow::array::{
     new_null_array, BinaryBuilder, DictionaryArray, GenericByteArray, ListArray, MapArray,
     StringArray, StructArray,
 };
-use arrow::datatypes::{ArrowDictionaryKeyType, ArrowNativeType, DataType, Schema};
+use arrow::datatypes::{
+    format_decimal_str, ArrowDictionaryKeyType, ArrowNativeType, DataType, Schema,
+};
 use arrow::datatypes::{Field, Fields, GenericBinaryType};
 use arrow::error::ArrowError;
 use arrow::{
     array::{
-        cast::AsArray, types::Int32Type, Array, ArrayRef, Int16Array, Int32Array, Int64Array,
-        Int8Array, OffsetSizeTrait, PrimitiveArray,
+        cast::AsArray, types::Decimal128Type, types::Int32Type, Array, ArrayRef, Int16Array,
+        Int32Array, Int64Array, Int8Array, OffsetSizeTrait, PrimitiveArray,
     },
-    compute::{cast_with_options, take, CastOptions},
+    compute::{cast_with_options, rescale_decimal, take, CastOptions},
     record_batch::RecordBatch,
     util::display::FormatOptions,
 };
@@ -339,6 +341,38 @@ pub(crate) fn cast_array(
         (Utf8 | LargeUtf8, Decimal256(precision, scale)) => {
             cast_string_to_decimal(&array, to_type, precision, scale, eval_mode)
         }
+        (Decimal128(input_precision, input_scale), Decimal128(output_precision, output_scale))
+            if eval_mode == EvalMode::Ansi =>
+        {
+            cast_with_options(&array, to_type, &native_cast_options).map_err(|error| {
+                array
+                    .as_primitive::<Decimal128Type>()
+                    .iter()
+                    .flatten()
+                    .find(|value| {
+                        rescale_decimal::<Decimal128Type, Decimal128Type>(
+                            *value,
+                            *input_precision,
+                            *input_scale,
+                            *output_precision,
+                            *output_scale,
+                        )
+                        .is_none()
+                    })
+                    .map_or_else(
+                        || error.into(),
+                        |value| SparkError::NumericValueOutOfRange {
+                            value: format_decimal_str(
+                                &value.to_string(),
+                                *input_precision as usize,
+                                *input_scale,
+                            ),
+                            precision: *output_precision,
+                            scale: *output_scale,
+                        },
+                    )
+            })
+        }
         (Int64, Int32)
         | (Int64, Int16)
         | (Int64, Int8)
@@ -752,8 +786,12 @@ impl PhysicalExpr for Cast {
     }
 
     fn evaluate(&self, batch: &RecordBatch) -> DataFusionResult<ColumnarValue> {
-        let arg = self.child.evaluate(batch)?;
-        let result = spark_cast(arg, &self.data_type, &self.cast_options);
+        // `CometIntegralDivide` builds its inner `CheckOverflow` without an expression id, so a
+        // bare child `SparkError` deliberately inherits the outer Cast's query context here.
+        let result = self
+            .child
+            .evaluate(batch)
+            .and_then(|arg| spark_cast(arg, &self.data_type, &self.cast_options));
 
         // If there's an error and we have query_context, wrap it
         match result {
@@ -908,11 +946,62 @@ fn cast_binary_to_string<O: OffsetSizeTrait>(
 #[cfg(test)]
 mod tests {
     use super::*;
-    use arrow::array::{BinaryArray, ListArray, NullArray, StringArray};
+    use arrow::array::{BinaryArray, Decimal128Array, ListArray, NullArray, StringArray};
     use arrow::buffer::OffsetBuffer;
     use arrow::datatypes::TimestampMicrosecondType;
     use arrow::datatypes::{Field, Fields};
 
+    #[test]
+    fn test_cast_decimal_to_decimal_ansi_overflow_returns_spark_error() {
+        let cases = [
+            (
+                vec![None, Some(1), Some(-123_456_789)],
+                DataType::Decimal128(10, 4),
+                DataType::Decimal128(6, 2),
+                "-12345.6789",
+                6,
+                2,
+            ),
+            (
+                vec![None, Some(1), Some(-999)],
+                DataType::Decimal128(3, 0),
+                DataType::Decimal128(3, 2),
+                "-999",
+                3,
+                2,
+            ),
+        ];
+
+        for (values, input_type, output_type, expected_value, expected_precision, expected_scale) in
+            cases
+        {
+            let input: ArrayRef =
+                Arc::new(Decimal128Array::from(values).with_data_type(input_type));
+            let error = cast_array(
+                input,
+                &output_type,
+                &SparkCastOptions::new_without_timezone(EvalMode::Ansi, false),
+            )
+            .unwrap_err();
+
+            match error {
+                DataFusionError::External(error) => match error.downcast_ref::<SparkError>() {
+                    Some(SparkError::NumericValueOutOfRange {
+                        value,
+                        precision,
+                        scale,
+                    }) => {
+                        assert_eq!(value, expected_value);
+                        assert_eq!(*precision, expected_precision);
+                        assert_eq!(*scale, expected_scale);
+                    }
+                    other => panic!("expected NumericValueOutOfRange, got {other:?}"),
+                },
+                other => panic!("expected external SparkError, got {other:?}"),
+            }
+        }
+    }
+
     #[test]
     fn test_cast_binary_to_string_replaces_invalid_utf8_jvm_compatibly() {
         // Invalid bytes are replaced with U+FFFD instead of reinterpreted as an invalid `str`,

native/spark-expr/src/conversion_funcs/numeric.rs

@@ -24,8 +24,8 @@ use arrow::array::{
     OffsetSizeTrait, PrimitiveArray, StringBuilder, TimestampMicrosecondBuilder,
 };
 use arrow::datatypes::{
-    i256, is_validate_decimal_precision, ArrowPrimitiveType, DataType, Decimal128Type, Float32Type,
-    Float64Type, Int16Type, Int32Type, Int64Type, Int8Type,
+    format_decimal_str, i256, is_validate_decimal_precision, ArrowPrimitiveType, DataType,
+    Decimal128Type, Float32Type, Float64Type, Int16Type, Int32Type, Int64Type, Int8Type,
 };
 use num::{cast::AsPrimitive, ToPrimitive, Zero};
 use std::sync::Arc;
@@ -559,30 +559,6 @@ macro_rules! cast_decimal_to_int32_up {
     }};
 }
 
-// copied from arrow::dataTypes::Decimal128Type since Decimal128Type::format_decimal can't be called directly
-pub(crate) fn format_decimal_str(value_str: &str, precision: usize, scale: i8) -> String {
-    let (sign, rest) = match value_str.strip_prefix('-') {
-        Some(stripped) => ("-", stripped),
-        None => ("", value_str),
-    };
-    let bound = precision.min(rest.len()) + sign.len();
-    let value_str = &value_str[0..bound];
-
-    if scale == 0 {
-        value_str.to_string()
-    } else if scale < 0 {
-        let padding = value_str.len() + scale.unsigned_abs() as usize;
-        format!("{value_str:0<padding$}")
-    } else if rest.len() > scale as usize {
-        // Decimal separator is in the middle of the string
-        let (whole, decimal) = value_str.split_at(value_str.len() - scale as usize);
-        format!("{whole}.{decimal}")
-    } else {
-        // String has to be padded
-        format!("{}0.{:0>width$}", sign, rest, width = scale as usize)
-    }
-}
-
 /// Casts a Decimal128 array to string using Java's BigDecimal.toString() semantics,
 /// which is Spark's LEGACY eval mode behavior. Plain notation when scale >= 0 and
 /// adjusted_exponent >= -6, otherwise scientific notation (e.g. "0E-18" for zero

native/spark-expr/src/error.rs

@@ -19,3 +19,15 @@
 pub use datafusion_comet_common::{
     decimal_overflow_error, SparkError, SparkErrorWithContext, SparkResult,
 };
+
+use arrow::error::ArrowError;
+use datafusion::common::DataFusionError;
+
+/// Arrow's `try_*` kernels require closure errors to be `ArrowError`, so a `SparkError`
+/// travels through `ExternalError`. Unwrap it again so JNI sees the direct `SparkError`.
+pub(crate) fn unwrap_arrow_external_error(error: ArrowError) -> DataFusionError {
+    match error {
+        ArrowError::ExternalError(error) => DataFusionError::External(error),
+        error => error.into(),
+    }
+}