Improve decimal arithmetic execution

Signed-off-by: Matt Katz <mhkatz97@gmail.com>

Author: mhk197

vortex-array/src/compute/conformance/binary_numeric.rs

@@ -313,18 +313,6 @@ fn test_decimal_binary_numeric_with_scalar(
                 (rhs_const.clone(), array.clone())
             };
 
-            let result = lhs
-                .binary(rhs, operator.into())
-                .vortex_expect("apply shouldn't fail")
-                .execute::<RecursiveCanonical>(ctx)
-                .map(|c| c.0.into_array());
-
-            // Skip this operator if the entire operation fails (e.g. an overflowing lane).
-            let Ok(result) = result else {
-                continue;
-            };
-
-            let actual_values = to_vec_of_scalar(&result, ctx);
             let expected_results: Vec<Option<Scalar>> = original_values
                 .iter()
                 .map(|x| {
@@ -337,20 +325,46 @@ fn test_decimal_binary_numeric_with_scalar(
                 })
                 .collect();
 
-            // For elements that didn't overflow, check they match.
+            let result = lhs
+                .binary(rhs, operator.into())
+                .vortex_expect("apply shouldn't fail")
+                .execute::<RecursiveCanonical>(ctx)
+                .map(|c| c.0.into_array());
+
+            let expects_overflow = expected_results.iter().any(Option::is_none);
+            if expects_overflow {
+                assert!(
+                    result.is_err(),
+                    "Decimal binary numeric operation should overflow for encoding {}: \
+                     {operator:?} {scalar} (lhs_is_array: {lhs_is_array})",
+                    array.encoding_id(),
+                );
+                continue;
+            }
+
+            let result = result.unwrap_or_else(|err| {
+                vortex_panic!(
+                    "Decimal binary numeric operation unexpectedly failed for encoding {}: \
+                     {operator:?} {scalar} (lhs_is_array: {lhs_is_array}): {err}",
+                    array.encoding_id(),
+                )
+            });
+
+            let actual_values = to_vec_of_scalar(&result, ctx);
             for (idx, (actual, expected)) in actual_values.iter().zip(&expected_results).enumerate()
             {
-                if let Some(expected_value) = expected {
-                    assert_eq!(
-                        actual,
-                        expected_value,
-                        "Decimal binary numeric operation failed for encoding {} at index {}: \
-                         ({array:?})[{idx}] {operator:?} {scalar} (lhs_is_array: {lhs_is_array}) \
-                         expected {expected_value:?}, got {actual:?}",
-                        array.encoding_id(),
-                        idx,
-                    );
-                }
+                let expected_value = expected
+                    .as_ref()
+                    .vortex_expect("non-overflowing decimal lane must have an expected value");
+                assert_eq!(
+                    actual,
+                    expected_value,
+                    "Decimal binary numeric operation failed for encoding {} at index {}: \
+                     ({array:?})[{idx}] {operator:?} {scalar} (lhs_is_array: {lhs_is_array}) \
+                     expected {expected_value:?}, got {actual:?}",
+                    array.encoding_id(),
+                    idx,
+                );
             }
         }
     }

vortex-array/src/expr/transform/coerce.rs

@@ -70,6 +70,7 @@ mod tests {
     use vortex_error::VortexResult;
 
     use crate::dtype::DType;
+    use crate::dtype::DecimalDType;
     use crate::dtype::Nullability::NonNullable;
     use crate::dtype::PType;
     use crate::dtype::StructFields;
@@ -153,6 +154,52 @@ mod tests {
         Ok(())
     }
 
+    #[test]
+    fn decimal_arithmetic_coerces_precision_and_scale() -> VortexResult<()> {
+        let result_dtype = DType::Decimal(DecimalDType::new(4, 2), NonNullable);
+        let scope = DType::Struct(
+            StructFields::new(
+                ["a", "b"].into(),
+                vec![
+                    DType::Decimal(DecimalDType::new(3, 1), NonNullable),
+                    result_dtype.clone(),
+                ],
+            ),
+            NonNullable,
+        );
+        let expr = Binary.new_expr(Operator::Add, [col("a"), col("b")]);
+
+        let coerced = coerce_expression(expr, &scope)?;
+
+        assert!(coerced.child(0).is::<Cast>());
+        assert!(!coerced.child(1).is::<Cast>());
+        assert_eq!(coerced.return_dtype(&scope)?, result_dtype);
+        Ok(())
+    }
+
+    #[test]
+    fn decimal_arithmetic_coerces_integer_operand() -> VortexResult<()> {
+        let result_dtype = DType::Decimal(DecimalDType::new(5, 2), NonNullable);
+        let scope = DType::Struct(
+            StructFields::new(
+                ["a", "b"].into(),
+                vec![
+                    DType::Decimal(DecimalDType::new(4, 2), NonNullable),
+                    DType::Primitive(PType::I8, NonNullable),
+                ],
+            ),
+            NonNullable,
+        );
+        let expr = Binary.new_expr(Operator::Add, [col("a"), col("b")]);
+
+        let coerced = coerce_expression(expr, &scope)?;
+
+        assert!(coerced.child(0).is::<Cast>());
+        assert!(coerced.child(1).is::<Cast>());
+        assert_eq!(coerced.return_dtype(&scope)?, result_dtype);
+        Ok(())
+    }
+
     #[test]
     fn boolean_operators_no_coercion() -> VortexResult<()> {
         let scope = DType::Struct(