From 101d0005c8b0ecff3007cec361e248be027e2fc9 Mon Sep 17 00:00:00 2001
From: Matt Katz <mhkatz97@gmail.com>
Date: Tue, 21 Jul 2026 11:50:22 -0700
Subject: [PATCH 1/7] Implement decimal array mult and div

Signed-off-by: Matt Katz <mhkatz97@gmail.com>
---
 Cargo.lock                                    |   2 +
 vortex-array/Cargo.toml                       |   2 +
 vortex-array/benches/binary_ops.rs            | 206 ++++++++++++++
 .../src/compute/conformance/binary_numeric.rs |  34 ++-
 .../scalar/typed_view/decimal/arithmetic.rs   |   4 +-
 .../src/scalar/typed_view/decimal/mod.rs      |   1 +
 .../scalar_fn/fns/binary/numeric/decimal.rs   | 253 +++++++++++++-----
 .../src/scalar_fn/fns/binary/numeric/mod.rs   |  10 +-
 .../src/scalar_fn/fns/binary/numeric/tests.rs | 158 +++++++++--
 9 files changed, 561 insertions(+), 109 deletions(-)

diff --git a/Cargo.lock b/Cargo.lock
index 81c0bb17416..9b6872d4e1c 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -9456,6 +9456,8 @@ dependencies = [
  "arbitrary",
  "arc-swap",
  "arcref",
+ "arrow-arith 58.4.0",
+ "arrow-array 58.4.0",
  "arrow-buffer 58.4.0",
  "async-lock",
  "bytes",
diff --git a/vortex-array/Cargo.toml b/vortex-array/Cargo.toml
index 70fba4acdcf..f87b07d5b78 100644
--- a/vortex-array/Cargo.toml
+++ b/vortex-array/Cargo.toml
@@ -82,6 +82,8 @@ _test-harness = ["dep:goldenfile", "dep:rstest", "dep:rstest_reuse"]
 serde = ["dep:serde", "vortex-buffer/serde", "vortex-mask/serde"]
 
 [dev-dependencies]
+arrow-arith = { workspace = true }
+arrow-array = { workspace = true }
 divan = { workspace = true }
 futures = { workspace = true, features = ["executor"] }
 insta = { workspace = true }
diff --git a/vortex-array/benches/binary_ops.rs b/vortex-array/benches/binary_ops.rs
index 14c4c9fbe3f..8af51b1c1ec 100644
--- a/vortex-array/benches/binary_ops.rs
+++ b/vortex-array/benches/binary_ops.rs
@@ -9,6 +9,16 @@
 
 use std::sync::LazyLock;
 
+use arrow_arith::numeric::div as arrow_div;
+use arrow_arith::numeric::mul as arrow_mul;
+use arrow_array::ArrowPrimitiveType;
+use arrow_array::PrimitiveArray as ArrowPrimitiveArray;
+use arrow_array::types::Decimal32Type;
+use arrow_array::types::Decimal64Type;
+use arrow_array::types::Decimal128Type;
+use arrow_array::types::Decimal256Type;
+use arrow_array::types::DecimalType as ArrowDecimalType;
+use arrow_buffer::ArrowNativeType;
 use divan::Bencher;
 use divan::counter::ItemsCount;
 use mimalloc::MiMalloc;
@@ -22,7 +32,9 @@ use vortex_array::arrays::ConstantArray;
 use vortex_array::arrays::DecimalArray;
 use vortex_array::arrays::PrimitiveArray;
 use vortex_array::builtins::ArrayBuiltins;
+use vortex_array::dtype::BigCast;
 use vortex_array::dtype::DecimalDType;
+use vortex_array::dtype::NativeDecimalType;
 use vortex_array::scalar_fn::fns::operators::Operator;
 use vortex_session::VortexSession;
 
@@ -190,6 +202,124 @@ fn add_decimal_i128_nullable(bencher: Bencher) {
     bench_decimal(bencher, lhs, rhs, Operator::Add);
 }
 
+macro_rules! decimal_compare_benches {
+    (
+        $vortex_mul:ident,
+        $arrow_mul:ident,
+        $vortex_div:ident,
+        $arrow_div:ident,
+        $vortex_type:ty,
+        $arrow_type:ty,
+        $precision:expr,
+        $scale:expr
+    ) => {
+        #[divan::bench]
+        fn $vortex_mul(bencher: Bencher) {
+            let dtype = DecimalDType::new($precision, $scale);
+            let lhs = comparison_vortex_decimal::<$vortex_type>(dtype, 1).into_array();
+            let rhs = comparison_vortex_decimal::<$vortex_type>(dtype, 17).into_array();
+            bench_decimal(bencher, lhs, rhs, Operator::Mul);
+        }
+
+        #[divan::bench]
+        fn $arrow_mul(bencher: Bencher) {
+            let lhs = comparison_arrow_decimal::<$arrow_type>($precision, $scale, 1);
+            let rhs = comparison_arrow_decimal::<$arrow_type>($precision, $scale, 17);
+            bench_arrow_decimal(bencher, lhs, rhs, ArrowDecimalOp::Mul);
+        }
+
+        #[divan::bench]
+        fn $vortex_div(bencher: Bencher) {
+            let dtype = DecimalDType::new($precision, $scale);
+            let lhs = comparison_vortex_decimal::<$vortex_type>(dtype, 1).into_array();
+            let rhs = comparison_vortex_decimal::<$vortex_type>(dtype, 17).into_array();
+            bench_decimal(bencher, lhs, rhs, Operator::Div);
+        }
+
+        #[divan::bench]
+        fn $arrow_div(bencher: Bencher) {
+            let lhs = comparison_arrow_decimal::<$arrow_type>($precision, $scale, 1);
+            let rhs = comparison_arrow_decimal::<$arrow_type>($precision, $scale, 17);
+            bench_arrow_decimal(bencher, lhs, rhs, ArrowDecimalOp::Div);
+        }
+    };
+}
+
+decimal_compare_benches!(
+    vortex_mul_decimal_i32_nonnull,
+    arrow_mul_decimal_i32_nonnull,
+    vortex_div_decimal_i32_nonnull,
+    arrow_div_decimal_i32_nonnull,
+    i32,
+    Decimal32Type,
+    4,
+    1
+);
+decimal_compare_benches!(
+    vortex_mul_decimal_i64_nonnull,
+    arrow_mul_decimal_i64_nonnull,
+    vortex_div_decimal_i64_nonnull,
+    arrow_div_decimal_i64_nonnull,
+    i64,
+    Decimal64Type,
+    8,
+    2
+);
+decimal_compare_benches!(
+    vortex_mul_decimal_i128_nonnull,
+    arrow_mul_decimal_i128_nonnull,
+    vortex_div_decimal_i128_nonnull,
+    arrow_div_decimal_i128_nonnull,
+    i128,
+    Decimal128Type,
+    18,
+    2
+);
+decimal_compare_benches!(
+    vortex_mul_decimal_i256_nonnull,
+    arrow_mul_decimal_i256_nonnull,
+    vortex_div_decimal_i256_nonnull,
+    arrow_div_decimal_i256_nonnull,
+    vortex_array::dtype::i256,
+    Decimal256Type,
+    38,
+    2
+);
+
+#[divan::bench]
+fn vortex_mul_decimal_i256_nullable(bencher: Bencher) {
+    let dtype = DecimalDType::new(38, 2);
+    let lhs =
+        comparison_vortex_decimal_nullable::<vortex_array::dtype::i256>(dtype, 1, 7).into_array();
+    let rhs =
+        comparison_vortex_decimal_nullable::<vortex_array::dtype::i256>(dtype, 17, 5).into_array();
+    bench_decimal(bencher, lhs, rhs, Operator::Mul);
+}
+
+#[divan::bench]
+fn arrow_mul_decimal_i256_nullable(bencher: Bencher) {
+    let lhs = comparison_arrow_decimal_nullable::<Decimal256Type>(38, 2, 1, 7);
+    let rhs = comparison_arrow_decimal_nullable::<Decimal256Type>(38, 2, 17, 5);
+    bench_arrow_decimal(bencher, lhs, rhs, ArrowDecimalOp::Mul);
+}
+
+#[divan::bench]
+fn vortex_div_decimal_i256_nullable(bencher: Bencher) {
+    let dtype = DecimalDType::new(38, 2);
+    let lhs =
+        comparison_vortex_decimal_nullable::<vortex_array::dtype::i256>(dtype, 1, 7).into_array();
+    let rhs =
+        comparison_vortex_decimal_nullable::<vortex_array::dtype::i256>(dtype, 17, 5).into_array();
+    bench_decimal(bencher, lhs, rhs, Operator::Div);
+}
+
+#[divan::bench]
+fn arrow_div_decimal_i256_nullable(bencher: Bencher) {
+    let lhs = comparison_arrow_decimal_nullable::<Decimal256Type>(38, 2, 1, 7);
+    let rhs = comparison_arrow_decimal_nullable::<Decimal256Type>(38, 2, 17, 5);
+    bench_arrow_decimal(bencher, lhs, rhs, ArrowDecimalOp::Div);
+}
+
 #[divan::bench]
 fn eq_i64_constant(bencher: Bencher) {
     let lhs = primitive_nonnull(0).into_array();
@@ -230,6 +360,28 @@ fn bench_decimal(bencher: Bencher, lhs: ArrayRef, rhs: ArrayRef, operator: Opera
     bench_binary::<DecimalArray>(bencher, lhs, rhs, operator);
 }
 
+#[derive(Clone, Copy)]
+enum ArrowDecimalOp {
+    Mul,
+    Div,
+}
+
+fn bench_arrow_decimal<T>(
+    bencher: Bencher,
+    lhs: ArrowPrimitiveArray<T>,
+    rhs: ArrowPrimitiveArray<T>,
+    op: ArrowDecimalOp,
+) where
+    T: ArrowPrimitiveType,
+{
+    bencher
+        .counter(ItemsCount::new(LEN))
+        .bench_local(|| match op {
+            ArrowDecimalOp::Mul => arrow_mul(&lhs, &rhs).unwrap(),
+            ArrowDecimalOp::Div => arrow_div(&lhs, &rhs).unwrap(),
+        });
+}
+
 fn bench_bool(bencher: Bencher, lhs: ArrayRef, rhs: ArrayRef, operator: Operator) {
     bench_binary::<BoolArray>(bencher, lhs, rhs, operator);
 }
@@ -266,6 +418,60 @@ fn decimal_i128_nullable(base: i128, null_every: usize) -> DecimalArray {
     )
 }
 
+fn comparison_vortex_decimal<T>(dtype: DecimalDType, offset: usize) -> DecimalArray
+where
+    T: NativeDecimalType,
+{
+    DecimalArray::from_iter::<T, _>(
+        (0..LEN).map(|idx| <T as BigCast>::from(((idx + offset) % 89 + 1) as i64).unwrap()),
+        dtype,
+    )
+}
+
+fn comparison_vortex_decimal_nullable<T>(
+    dtype: DecimalDType,
+    offset: usize,
+    null_every: usize,
+) -> DecimalArray
+where
+    T: NativeDecimalType,
+{
+    DecimalArray::from_option_iter::<T, _>(
+        (0..LEN).map(|idx| {
+            (!idx.is_multiple_of(null_every))
+                .then(|| <T as BigCast>::from(((idx + offset) % 89 + 1) as i64).unwrap())
+        }),
+        dtype,
+    )
+}
+
+fn comparison_arrow_decimal<T>(precision: u8, scale: i8, offset: usize) -> ArrowPrimitiveArray<T>
+where
+    T: ArrowPrimitiveType + ArrowDecimalType,
+{
+    ArrowPrimitiveArray::<T>::from_iter_values(
+        (0..LEN).map(|idx| T::Native::usize_as((idx + offset) % 89 + 1)),
+    )
+    .with_precision_and_scale(precision, scale)
+    .unwrap()
+}
+
+fn comparison_arrow_decimal_nullable<T>(
+    precision: u8,
+    scale: i8,
+    offset: usize,
+    null_every: usize,
+) -> ArrowPrimitiveArray<T>
+where
+    T: ArrowPrimitiveType + ArrowDecimalType,
+{
+    ArrowPrimitiveArray::<T>::from_iter((0..LEN).map(|idx| {
+        (!idx.is_multiple_of(null_every)).then(|| T::Native::usize_as((idx + offset) % 89 + 1))
+    }))
+    .with_precision_and_scale(precision, scale)
+    .unwrap()
+}
+
 fn primitive_small_nonnull(offset: i64) -> PrimitiveArray {
     PrimitiveArray::from_iter((0..LEN as i64).map(|i| ((i + offset) % 1024) + 1))
 }
diff --git a/vortex-array/src/compute/conformance/binary_numeric.rs b/vortex-array/src/compute/conformance/binary_numeric.rs
index a0d91715028..1750ffc8e5d 100644
--- a/vortex-array/src/compute/conformance/binary_numeric.rs
+++ b/vortex-array/src/compute/conformance/binary_numeric.rs
@@ -26,6 +26,8 @@ use std::fmt::Debug;
 use itertools::Itertools;
 use num_traits::Bounded;
 use num_traits::CheckedAdd;
+use num_traits::CheckedDiv;
+use num_traits::CheckedMul;
 use num_traits::CheckedSub;
 use num_traits::Float;
 use num_traits::Num;
@@ -303,8 +305,16 @@ fn test_decimal_binary_numeric_with_scalar(
 
     let scalar = Scalar::decimal(value, decimal_dtype, array.dtype().nullability());
 
-    // Decimal Mul/Div are not yet implemented.
-    for operator in [NumericOperator::Add, NumericOperator::Sub] {
+    let mut operators = vec![
+        NumericOperator::Add,
+        NumericOperator::Sub,
+        NumericOperator::Mul,
+    ];
+    if !value.is_zero() {
+        operators.push(NumericOperator::Div);
+    }
+
+    for operator in operators {
         for lhs_is_array in [true, false] {
             test_decimal_binary_numeric_direction(
                 array,
@@ -329,7 +339,7 @@ fn test_decimal_binary_numeric_direction(
     ctx: &mut ExecutionCtx,
 ) {
     let result_decimal_dtype = numeric_op_result_decimal_dtype(decimal_dtype, operator)
-        .vortex_expect("decimal Add/Sub must have a result dtype");
+        .vortex_expect("decimal arithmetic must have a result dtype");
     let result_dtype = DType::Decimal(result_decimal_dtype, array.dtype().nullability());
     let expected_results = expected_decimal_results(
         original_values,
@@ -382,10 +392,22 @@ fn expected_decimal_results(
             let (Some(lhs), Some(rhs)) = (lhs.decimal_value(), rhs.decimal_value()) else {
                 return Some(Scalar::null(result_dtype.clone()));
             };
+            let lhs = lhs.as_i256();
+            let rhs = rhs.as_i256();
             let value = match operator {
-                NumericOperator::Add => lhs.as_i256().checked_add(&rhs.as_i256()),
-                NumericOperator::Sub => lhs.as_i256().checked_sub(&rhs.as_i256()),
-                NumericOperator::Mul | NumericOperator::Div => unreachable!(),
+                NumericOperator::Add => lhs.checked_add(&rhs),
+                NumericOperator::Sub => lhs.checked_sub(&rhs),
+                NumericOperator::Mul => lhs.checked_mul(&rhs),
+                NumericOperator::Div => {
+                    let scale_power = result_decimal_dtype.scale();
+                    let factor =
+                        i256::from_i128(10).checked_pow(scale_power.unsigned_abs() as u32)?;
+                    if scale_power >= 0 {
+                        lhs.checked_mul(&factor)?.checked_div(&rhs)
+                    } else {
+                        lhs.checked_div(&rhs.checked_mul(&factor)?)
+                    }
+                }
             }?;
             let value = DecimalValue::try_from_i256(value, result_decimal_dtype).ok()?;
             Some(Scalar::decimal(
diff --git a/vortex-array/src/scalar/typed_view/decimal/arithmetic.rs b/vortex-array/src/scalar/typed_view/decimal/arithmetic.rs
index ff3007ca31d..2a47c9cf977 100644
--- a/vortex-array/src/scalar/typed_view/decimal/arithmetic.rs
+++ b/vortex-array/src/scalar/typed_view/decimal/arithmetic.rs
@@ -108,7 +108,7 @@ pub(crate) fn checked_decimal_numeric(
     result: DecimalDType,
     op: NumericOperator,
 ) -> Option<DecimalValue> {
-    let work = work_decimal_dtype(input, result, op);
+    let work = decimal_numeric_work_dtype(input, result, op);
     match_each_decimal_value_type!(DecimalType::smallest_decimal_value_type(&work), |W| {
         let value = checked_at_width::<W>(lhs.cast::<W>()?, rhs.cast::<W>()?, result.scale(), op)?;
         DecimalValue::from(value).normalize(result)
@@ -120,7 +120,7 @@ pub(crate) fn checked_decimal_numeric(
 /// The result precision covers Add, Sub and Mul, whose intermediates are the result itself. Div
 /// scales the dividend (or the divisor, for a negative result scale) by `10^result_scale` before
 /// dividing, so it needs room for `p + |result_scale|` digits.
-fn work_decimal_dtype(
+pub(crate) fn decimal_numeric_work_dtype(
     input: DecimalDType,
     result: DecimalDType,
     op: NumericOperator,
diff --git a/vortex-array/src/scalar/typed_view/decimal/mod.rs b/vortex-array/src/scalar/typed_view/decimal/mod.rs
index dd7219cf8dd..149fc8aeadf 100644
--- a/vortex-array/src/scalar/typed_view/decimal/mod.rs
+++ b/vortex-array/src/scalar/typed_view/decimal/mod.rs
@@ -8,6 +8,7 @@ mod dvalue;
 mod scalar;
 
 pub(crate) use arithmetic::decimal_numeric_result_dtype;
+pub(crate) use arithmetic::decimal_numeric_work_dtype;
 pub use dvalue::DecimalValue;
 pub use scalar::DecimalScalar;
 
diff --git a/vortex-array/src/scalar_fn/fns/binary/numeric/decimal.rs b/vortex-array/src/scalar_fn/fns/binary/numeric/decimal.rs
index a380aaa7996..6b0e8d1bc82 100644
--- a/vortex-array/src/scalar_fn/fns/binary/numeric/decimal.rs
+++ b/vortex-array/src/scalar_fn/fns/binary/numeric/decimal.rs
@@ -4,20 +4,27 @@
 //! Native execution of the arithmetic operators over decimal arrays.
 //!
 //! Both operands share a logical [`DecimalDType`] (equal precision and scale). Add and Sub apply
-//! directly to the unscaled stored integers and are exact at that shared scale. The result reserves
-//! one additional precision digit for a carry, capped at Vortex's maximum decimal precision.
+//! directly to the unscaled stored integers and are exact at that shared scale. Mul takes the raw
+//! product, which the doubled result scale leaves correctly scaled, and Div rescales the dividend
+//! (or the divisor, for a negative result scale) before integer division. Result precision and
+//! scale follow Arrow's rules — see [`decimal_numeric_result_dtype`].
 //!
-//! Lanes execute in a working width chosen so that in-precision inputs cannot spuriously
-//! overflow an intermediate value. An operation that overflows the result precision on a valid
-//! lane is an error; invalid lanes never error.
+//! Lanes execute in a working width wide enough that in-precision inputs cannot spuriously
+//! overflow an intermediate, then narrow to the result's own storage width. Multiplication skips
+//! the redundant overflow check when the input precision proves every product fits. Otherwise, an
+//! operation that overflows the result precision on a valid lane is an error; invalid lanes never
+//! error.
+
+use std::ops::Mul;
 
 use num_traits::CheckedAdd;
+use num_traits::CheckedDiv;
+use num_traits::CheckedMul;
 use num_traits::CheckedSub;
 use vortex_buffer::Buffer;
 use vortex_compute::lane_kernels::LaneZip;
 use vortex_error::VortexExpect;
 use vortex_error::VortexResult;
-use vortex_error::vortex_bail;
 use vortex_error::vortex_err;
 use vortex_mask::Mask;
 
@@ -33,27 +40,16 @@ use crate::arrays::decimal::DecimalArrayExt;
 use crate::dtype::BigCast;
 use crate::dtype::DType;
 use crate::dtype::DecimalDType;
+use crate::dtype::MAX_PRECISION;
 use crate::dtype::NativeDecimalType;
 use crate::match_each_decimal_value_type;
 use crate::scalar::DecimalValue;
 use crate::scalar::NumericOperator;
 use crate::scalar::Scalar;
 use crate::scalar::decimal_numeric_result_dtype;
+use crate::scalar::decimal_numeric_work_dtype;
 use crate::validity::Validity;
 
-/// Derive the result decimal dtype for a numeric operation over same-typed operands.
-pub(crate) fn result_decimal_dtype(
-    input: DecimalDType,
-    op: NumericOperator,
-) -> VortexResult<DecimalDType> {
-    match op {
-        NumericOperator::Add | NumericOperator::Sub => decimal_numeric_result_dtype(input, op),
-        NumericOperator::Mul | NumericOperator::Div => {
-            vortex_bail!("numeric operator {op} is not yet supported for decimal arrays")
-        }
-    }
-}
-
 /// Execute a numeric operation between two decimal arrays sharing a decimal dtype.
 pub(super) fn execute_numeric_decimal(
     lhs: &ArrayRef,
@@ -66,7 +62,7 @@ pub(super) fn execute_numeric_decimal(
         .as_decimal_opt()
         .vortex_expect("inputs are both decimals");
 
-    let result_decimal_dtype = result_decimal_dtype(*decimal_dtype, op)?;
+    let result_decimal_dtype = decimal_numeric_result_dtype(*decimal_dtype, op)?;
     let result_dtype = DType::Decimal(
         result_decimal_dtype,
         lhs.dtype().nullability() | rhs.dtype().nullability(),
@@ -89,20 +85,25 @@ pub(super) fn execute_numeric_decimal(
     let validity = lhs.validity().and(rhs.validity())?;
     let valid_rows = validity.execute_mask(len, ctx)?;
 
-    match_each_decimal_value_type!(
-        DecimalType::smallest_decimal_value_type(&result_decimal_dtype),
-        |W| {
-            execute_decimal_at_width::<W>(
-                &lhs,
-                &rhs,
-                op,
-                result_decimal_dtype,
-                &result_dtype,
-                validity,
-                &valid_rows,
-            )
-        }
-    )
+    let work_dtype = decimal_numeric_work_dtype(*decimal_dtype, result_decimal_dtype, op);
+    match_each_decimal_value_type!(DecimalType::smallest_decimal_value_type(&work_dtype), |W| {
+        let plan = DecimalOpPlan::<W>::new(*decimal_dtype, result_decimal_dtype, op)?;
+        match_each_decimal_value_type!(
+            DecimalType::smallest_decimal_value_type(&result_decimal_dtype),
+            |O| {
+                execute_decimal_at_widths::<W, O>(
+                    &lhs,
+                    &rhs,
+                    op,
+                    result_decimal_dtype,
+                    &result_dtype,
+                    validity,
+                    &valid_rows,
+                    &plan,
+                )
+            }
+        )
+    })
 }
 
 fn is_null_constant(array: &ArrayRef) -> bool {
@@ -180,7 +181,7 @@ struct DecimalValueBounds<W> {
 
 impl<W: NativeDecimalType> DecimalValueBounds<W> {
     fn new(dtype: DecimalDType) -> Self {
-        let precision = dtype.precision() as usize;
+        let precision = usize::from(dtype.precision());
         Self {
             lower_bound: W::MIN_BY_PRECISION[precision],
             upper_bound: W::MAX_BY_PRECISION[precision],
@@ -193,42 +194,139 @@ impl<W: NativeDecimalType> DecimalValueBounds<W> {
     }
 }
 
-/// A checked fixed-point decimal operation on unscaled values at working width `W`.
+/// Per-execution constants for a decimal operation at working width `W`.
+struct DecimalOpPlan<W> {
+    bounds: DecimalValueBounds<W>,
+    lhs_scale_factor: W,
+    rhs_scale_factor: W,
+    guaranteed_mul: bool,
+}
+
+impl<W> DecimalOpPlan<W>
+where
+    W: NativeDecimalType + CheckedMul,
+{
+    fn new(input: DecimalDType, result: DecimalDType, op: NumericOperator) -> VortexResult<Self> {
+        let one = cast_work_value::<W, i8>(1);
+        let (lhs_scale_factor, rhs_scale_factor) = if op == NumericOperator::Div {
+            // Arrow scales the quotient by 10^(result_scale - lhs_scale + rhs_scale). Both
+            // Vortex operands share a dtype, so this simplifies to 10^result_scale. A negative
+            // exponent scales the divisor instead of the dividend.
+            let exponent = <u32 as From<u8>>::from(result.scale().unsigned_abs());
+            if result.scale() >= 0 {
+                (decimal_scale_factor::<W>(exponent)?, one)
+            } else {
+                (one, decimal_scale_factor::<W>(exponent)?)
+            }
+        } else {
+            (one, one)
+        };
+
+        Ok(Self {
+            bounds: DecimalValueBounds::new(result),
+            lhs_scale_factor,
+            rhs_scale_factor,
+            guaranteed_mul: op == NumericOperator::Mul
+                && input.precision().saturating_mul(2) <= MAX_PRECISION,
+        })
+    }
+}
+
+fn decimal_scale_factor<W>(exp: u32) -> VortexResult<W>
+where
+    W: NativeDecimalType + CheckedMul,
+{
+    let ten = cast_work_value::<W, i8>(10);
+    let mut factor = cast_work_value::<W, i8>(1);
+    for _ in 0..exp {
+        factor = factor.checked_mul(&ten).ok_or_else(|| {
+            vortex_err!(
+                InvalidArgument:
+                "decimal scale factor 10^{exp} cannot be represented at the working width"
+            )
+        })?;
+    }
+    Ok(factor)
+}
+
+/// A checked decimal operation on unscaled values at working width `W`.
 trait CheckedDecimalOp {
     const ERROR: &'static str;
 
-    fn apply<W>(lhs: W, rhs: W, bounds: &DecimalValueBounds<W>) -> Option<W>
+    fn apply<W>(lhs: W, rhs: W, plan: &DecimalOpPlan<W>) -> Option<W>
     where
-        W: NativeDecimalType + CheckedAdd + CheckedSub;
+        W: NativeDecimalType + CheckedAdd + CheckedSub + CheckedMul + CheckedDiv + Mul<Output = W>;
 }
 
 struct CheckedDecimalAdd;
 
 struct CheckedDecimalSub;
 
+struct CheckedDecimalMul;
+
+struct GuaranteedDecimalMul;
+
+struct CheckedDecimalDiv;
+
 impl CheckedDecimalOp for CheckedDecimalAdd {
     const ERROR: &'static str = "decimal overflow in checked add";
 
-    fn apply<W>(lhs: W, rhs: W, bounds: &DecimalValueBounds<W>) -> Option<W>
+    fn apply<W>(lhs: W, rhs: W, plan: &DecimalOpPlan<W>) -> Option<W>
     where
-        W: NativeDecimalType + CheckedAdd + CheckedSub,
+        W: NativeDecimalType + CheckedAdd + CheckedSub + CheckedMul + CheckedDiv + Mul<Output = W>,
     {
-        bounds.in_precision(lhs.checked_add(&rhs)?)
+        plan.bounds.in_precision(lhs.checked_add(&rhs)?)
     }
 }
 
 impl CheckedDecimalOp for CheckedDecimalSub {
     const ERROR: &'static str = "decimal overflow in checked sub";
 
-    fn apply<W>(lhs: W, rhs: W, bounds: &DecimalValueBounds<W>) -> Option<W>
+    fn apply<W>(lhs: W, rhs: W, plan: &DecimalOpPlan<W>) -> Option<W>
+    where
+        W: NativeDecimalType + CheckedAdd + CheckedSub + CheckedMul + CheckedDiv + Mul<Output = W>,
+    {
+        plan.bounds.in_precision(lhs.checked_sub(&rhs)?)
+    }
+}
+
+impl CheckedDecimalOp for CheckedDecimalMul {
+    const ERROR: &'static str = "decimal overflow in checked mul";
+
+    fn apply<W>(lhs: W, rhs: W, plan: &DecimalOpPlan<W>) -> Option<W>
+    where
+        W: NativeDecimalType + CheckedAdd + CheckedSub + CheckedMul + CheckedDiv + Mul<Output = W>,
+    {
+        plan.bounds.in_precision(lhs.checked_mul(&rhs)?)
+    }
+}
+
+impl CheckedDecimalOp for GuaranteedDecimalMul {
+    const ERROR: &'static str = "decimal overflow in guaranteed mul";
+
+    fn apply<W>(lhs: W, rhs: W, _plan: &DecimalOpPlan<W>) -> Option<W>
+    where
+        W: NativeDecimalType + CheckedAdd + CheckedSub + CheckedMul + CheckedDiv + Mul<Output = W>,
+    {
+        Some(lhs * rhs)
+    }
+}
+
+impl CheckedDecimalOp for CheckedDecimalDiv {
+    const ERROR: &'static str = "decimal overflow or division by zero in checked div";
+
+    fn apply<W>(lhs: W, rhs: W, plan: &DecimalOpPlan<W>) -> Option<W>
     where
-        W: NativeDecimalType + CheckedAdd + CheckedSub,
+        W: NativeDecimalType + CheckedAdd + CheckedSub + CheckedMul + CheckedDiv + Mul<Output = W>,
     {
-        bounds.in_precision(lhs.checked_sub(&rhs)?)
+        let lhs = lhs.checked_mul(&plan.lhs_scale_factor)?;
+        let rhs = rhs.checked_mul(&plan.rhs_scale_factor)?;
+        plan.bounds.in_precision(lhs.checked_div(&rhs)?)
     }
 }
 
-fn execute_decimal_at_width<W>(
+#[expect(clippy::too_many_arguments, reason = "internal width-dispatch shim")]
+fn execute_decimal_at_widths<W, O>(
     lhs: &DecimalOperand,
     rhs: &DecimalOperand,
     op: NumericOperator,
@@ -236,20 +334,23 @@ fn execute_decimal_at_width<W>(
     result_dtype: &DType,
     validity: Validity,
     valid_rows: &Mask,
+    plan: &DecimalOpPlan<W>,
 ) -> VortexResult<ArrayRef>
 where
-    W: NativeDecimalType + CheckedAdd + CheckedSub,
-    DecimalValue: From<W>,
+    W: NativeDecimalType + CheckedAdd + CheckedSub + CheckedMul + CheckedDiv + Mul<Output = W>,
+    O: NativeDecimalType,
+    DecimalValue: From<O>,
 {
     macro_rules! execute_typed {
         ($Op:ty) => {
-            execute_decimal_typed::<W, $Op>(
+            execute_decimal_typed::<W, O, $Op>(
                 lhs,
                 rhs,
                 result_decimal_dtype,
                 result_dtype,
                 validity,
                 valid_rows,
+                plan,
             )
         };
     }
@@ -257,39 +358,42 @@ where
     match op {
         NumericOperator::Add => execute_typed!(CheckedDecimalAdd),
         NumericOperator::Sub => execute_typed!(CheckedDecimalSub),
-        NumericOperator::Mul | NumericOperator::Div => vortex_bail!(
-            "numeric operator {:?} is not yet supported for decimal arrays",
-            op
-        ),
+        // A product of two valid p-digit values needs at most 2p digits. When 2p is within
+        // MAX_PRECISION, both native and declared-precision overflow are impossible at the
+        // selected working width. This includes p=38 producing a precision-76 i256 result.
+        NumericOperator::Mul if plan.guaranteed_mul => execute_typed!(GuaranteedDecimalMul),
+        NumericOperator::Mul => execute_typed!(CheckedDecimalMul),
+        NumericOperator::Div => execute_typed!(CheckedDecimalDiv),
     }
 }
 
-fn execute_decimal_typed<W, Op>(
+fn execute_decimal_typed<W, O, Op>(
     lhs: &DecimalOperand,
     rhs: &DecimalOperand,
     result_decimal_dtype: DecimalDType,
     result_dtype: &DType,
     validity: Validity,
     valid_rows: &Mask,
+    plan: &DecimalOpPlan<W>,
 ) -> VortexResult<ArrayRef>
 where
-    W: NativeDecimalType + CheckedAdd + CheckedSub,
-    DecimalValue: From<W>,
+    W: NativeDecimalType + CheckedAdd + CheckedSub + CheckedMul + CheckedDiv + Mul<Output = W>,
+    O: NativeDecimalType,
+    DecimalValue: From<O>,
     Op: CheckedDecimalOp,
 {
     let len = lhs.len();
-    let bounds = DecimalValueBounds::<W>::new(result_decimal_dtype);
 
     let values = match (lhs, rhs) {
         (DecimalOperand::Array { values: lhs, .. }, DecimalOperand::Array { values: rhs, .. }) => {
-            checked_decimal_arrays::<W, Op>(lhs, rhs, &bounds, valid_rows)
+            checked_decimal_arrays::<W, O, Op>(lhs, rhs, plan, valid_rows)
         }
         (DecimalOperand::Array { values: lhs, .. }, DecimalOperand::Constant { value, .. }) => {
             let rhs = typed_constant::<W>(value);
             match_each_decimal_value_type!(lhs.values_type(), |L| {
                 let lhs = lhs.buffer::<L>();
                 checked_lanes(lhs.as_slice(), valid_rows, |lhs| {
-                    Op::apply(<W as BigCast>::from(lhs)?, rhs, &bounds)
+                    Op::apply(<W as BigCast>::from(lhs)?, rhs, plan).map(cast_result_value::<W, O>)
                 })
             })
         }
@@ -298,7 +402,7 @@ where
             match_each_decimal_value_type!(rhs.values_type(), |R| {
                 let rhs = rhs.buffer::<R>();
                 checked_lanes(rhs.as_slice(), valid_rows, |rhs| {
-                    Op::apply(lhs, <W as BigCast>::from(rhs)?, &bounds)
+                    Op::apply(lhs, <W as BigCast>::from(rhs)?, plan).map(cast_result_value::<W, O>)
                 })
             })
         }
@@ -308,8 +412,9 @@ where
         ) => {
             let lhs = typed_constant::<W>(lhs);
             let rhs = typed_constant::<W>(rhs);
-            let value = Op::apply(lhs, rhs, &bounds)
+            let value = Op::apply(lhs, rhs, plan)
                 .ok_or_else(|| vortex_err!(InvalidArgument: "{}", Op::ERROR))?;
+            let value = cast_result_value::<W, O>(value);
             return Ok(ConstantArray::new(
                 Scalar::decimal(
                     DecimalValue::from(value),
@@ -331,14 +436,15 @@ where
     .into_array())
 }
 
-fn checked_decimal_arrays<W, Op>(
+fn checked_decimal_arrays<W, O, Op>(
     lhs: &DecimalArray,
     rhs: &DecimalArray,
-    bounds: &DecimalValueBounds<W>,
+    plan: &DecimalOpPlan<W>,
     valid_rows: &Mask,
-) -> Result<Buffer<W>, usize>
+) -> Result<Buffer<O>, usize>
 where
-    W: NativeDecimalType + CheckedAdd + CheckedSub,
+    W: NativeDecimalType + CheckedAdd + CheckedSub + CheckedMul + CheckedDiv + Mul<Output = W>,
+    O: NativeDecimalType,
     Op: CheckedDecimalOp,
 {
     debug_assert_eq!(lhs.len(), rhs.len());
@@ -350,17 +456,26 @@ where
                 LaneZip::new(lhs.as_slice(), rhs.as_slice()),
                 valid_rows,
                 |(lhs, rhs)| {
-                    Op::apply(
-                        <W as BigCast>::from(lhs)?,
-                        <W as BigCast>::from(rhs)?,
-                        bounds,
-                    )
+                    Op::apply(<W as BigCast>::from(lhs)?, <W as BigCast>::from(rhs)?, plan)
+                        .map(cast_result_value::<W, O>)
                 },
             )
         })
     })
 }
 
+#[inline(always)]
+fn cast_work_value<W: NativeDecimalType, T: NativeDecimalType>(value: T) -> W {
+    <W as BigCast>::from(value)
+        .vortex_expect("valid decimal input must fit the arithmetic working width")
+}
+
+#[inline(always)]
+fn cast_result_value<W: NativeDecimalType, O: NativeDecimalType>(value: W) -> O {
+    <O as BigCast>::from(value)
+        .vortex_expect("precision-checked decimal result must fit the output width")
+}
+
 fn typed_constant<W: NativeDecimalType>(value: &DecimalValue) -> W {
     value
         .cast::<W>()
diff --git a/vortex-array/src/scalar_fn/fns/binary/numeric/mod.rs b/vortex-array/src/scalar_fn/fns/binary/numeric/mod.rs
index 7211ae9ed6a..6622e08f82b 100644
--- a/vortex-array/src/scalar_fn/fns/binary/numeric/mod.rs
+++ b/vortex-array/src/scalar_fn/fns/binary/numeric/mod.rs
@@ -13,8 +13,6 @@ mod primitive;
 mod tests;
 
 use decimal::execute_numeric_decimal;
-use decimal::result_decimal_dtype;
-pub(crate) use decimal::result_decimal_dtype as numeric_op_result_decimal_dtype;
 pub(crate) use primitive::PrimitiveOperand;
 use primitive::execute_numeric_primitive;
 use vortex_error::VortexResult;
@@ -26,6 +24,7 @@ use crate::ExecutionCtx;
 use crate::IntoArray;
 use crate::dtype::DType;
 use crate::scalar::NumericOperator;
+pub(crate) use crate::scalar::decimal_numeric_result_dtype as numeric_op_result_decimal_dtype;
 
 /// Execute a numeric operation between two arrays.
 pub(crate) fn execute_numeric(
@@ -74,9 +73,10 @@ fn build_empty_result(
     let nullability = lhs.dtype().nullability() | rhs.dtype().nullability();
     let result_dtype = match lhs.dtype() {
         DType::Primitive(..) => lhs.dtype().with_nullability(nullability),
-        DType::Decimal(decimal_dtype, _) => {
-            DType::Decimal(result_decimal_dtype(*decimal_dtype, op)?, nullability)
-        }
+        DType::Decimal(decimal_dtype, _) => DType::Decimal(
+            numeric_op_result_decimal_dtype(*decimal_dtype, op)?,
+            nullability,
+        ),
         _ => unreachable!("dtype is either Primitive or Decimal"),
     };
 
diff --git a/vortex-array/src/scalar_fn/fns/binary/numeric/tests.rs b/vortex-array/src/scalar_fn/fns/binary/numeric/tests.rs
index d0bba9913e9..b998eac7873 100644
--- a/vortex-array/src/scalar_fn/fns/binary/numeric/tests.rs
+++ b/vortex-array/src/scalar_fn/fns/binary/numeric/tests.rs
@@ -5,7 +5,7 @@ use rstest::rstest;
 use vortex_buffer::buffer;
 use vortex_error::VortexResult;
 
-use super::result_decimal_dtype;
+use super::numeric_op_result_decimal_dtype as result_decimal_dtype;
 use crate::ArrayRef;
 use crate::Columnar;
 use crate::IntoArray;
@@ -250,6 +250,12 @@ fn decimal_constant(value: impl Into<DecimalValue>, dtype: DecimalDType, len: us
 #[rstest]
 #[case::add(NumericOperator::Add, [150i64, 225], [1050i64, 1225])]
 #[case::sub(NumericOperator::Sub, [150i64, 225], [750i64, 775])]
+#[case::mul(NumericOperator::Mul, [150i64, 225], [135_000i64, 225_000])]
+#[case::div(
+    NumericOperator::Div,
+    [150i64, 225],
+    [6_000_000i64, 4_444_444]
+)]
 fn test_decimal_array_array(
     #[case] op: NumericOperator,
     #[case] rhs: [i64; 2],
@@ -370,6 +376,39 @@ fn test_decimal_overflow_on_null_lane_ignored() {
     );
 }
 
+#[test]
+fn test_decimal_divide_by_zero_on_null_lane_ignored() -> VortexResult<()> {
+    let mut ctx = array_session().create_execution_ctx();
+    let dtype = DecimalDType::new(10, 2);
+    let lhs = DecimalArray::new(
+        buffer![100i64, 1_000],
+        dtype,
+        Validity::from_iter([false, true]),
+    )
+    .into_array();
+    let rhs = DecimalArray::from_iter::<i64, _>([0, 200], dtype).into_array();
+
+    let result = decimal_binary(lhs, rhs, Operator::Div)?;
+    assert_arrays_eq!(
+        result,
+        DecimalArray::from_option_iter::<i64, _>(
+            [None, Some(5_000_000)],
+            result_decimal_dtype(dtype, NumericOperator::Div)?,
+        ),
+        &mut ctx
+    );
+    Ok(())
+}
+
+#[test]
+fn test_decimal_divide_by_zero_on_valid_lane_errors() {
+    let dtype = DecimalDType::new(10, 2);
+    let lhs = DecimalArray::from_iter::<i64, _>([100], dtype).into_array();
+    let rhs = DecimalArray::from_iter::<i64, _>([0], dtype).into_array();
+
+    assert!(decimal_binary(lhs, rhs, Operator::Div).is_err());
+}
+
 #[test]
 fn test_decimal_add_reserves_carry_digit() {
     let mut ctx = array_session().create_execution_ctx();
@@ -385,6 +424,47 @@ fn test_decimal_add_reserves_carry_digit() {
     );
 }
 
+#[test]
+fn test_decimal_mul_widens_before_multiplying() -> VortexResult<()> {
+    let mut ctx = array_session().create_execution_ctx();
+    let dtype = DecimalDType::new(38, 0);
+    let max = <i128 as NativeDecimalType>::MAX_BY_PRECISION[38];
+    let widened_max = i256::from_i128(max);
+    let result_dtype = result_decimal_dtype(dtype, NumericOperator::Mul)?;
+
+    let result = decimal_binary(
+        DecimalArray::from_iter::<i128, _>([max], dtype).into_array(),
+        DecimalArray::from_iter::<i128, _>([max], dtype).into_array(),
+        Operator::Mul,
+    )?;
+    assert_arrays_eq!(
+        result,
+        DecimalArray::from_iter::<i256, _>([widened_max * widened_max], result_dtype),
+        &mut ctx
+    );
+    Ok(())
+}
+
+#[test]
+fn test_decimal_mul_above_guaranteed_precision_checks_logical_bounds() {
+    let dtype = DecimalDType::new(39, 0);
+    let one = i256::from_i128(1);
+    let ten_to_38 = <i256 as NativeDecimalType>::MAX_BY_PRECISION[38] + one;
+    let value = ten_to_38 * i256::from_i128(2);
+    let product = value * value;
+
+    // The product fits the native i256 width but not the capped precision-76 result.
+    assert!(product > <i256 as NativeDecimalType>::MAX_BY_PRECISION[76]);
+    assert!(
+        decimal_binary(
+            DecimalArray::from_iter::<i256, _>([value], dtype).into_array(),
+            DecimalArray::from_iter::<i256, _>([value], dtype).into_array(),
+            Operator::Mul,
+        )
+        .is_err()
+    );
+}
+
 #[rstest]
 #[case::precision_2(
     DecimalArray::from_iter::<i8, _>([10, 20], DecimalDType::new(2, 0)),
@@ -510,9 +590,16 @@ fn test_decimal_nullable_constant_preserves_nullable_output() -> VortexResult<()
 }
 
 #[rstest]
-#[case::null_lhs(true)]
-#[case::null_rhs(false)]
-fn test_decimal_null_constant_yields_all_null(#[case] null_lhs: bool) -> VortexResult<()> {
+#[case::add_null_lhs(true, NumericOperator::Add)]
+#[case::add_null_rhs(false, NumericOperator::Add)]
+#[case::mul_null_lhs(true, NumericOperator::Mul)]
+#[case::mul_null_rhs(false, NumericOperator::Mul)]
+#[case::div_null_lhs(true, NumericOperator::Div)]
+#[case::div_null_rhs(false, NumericOperator::Div)]
+fn test_decimal_null_constant_yields_all_null(
+    #[case] null_lhs: bool,
+    #[case] op: NumericOperator,
+) -> VortexResult<()> {
     let mut ctx = array_session().create_execution_ctx();
     let dtype = DecimalDType::new(10, 2);
     let values = DecimalArray::from_iter::<i64, _>([100, 200], dtype).into_array();
@@ -527,16 +614,11 @@ fn test_decimal_null_constant_yields_all_null(#[case] null_lhs: bool) -> VortexR
         (values, null_constant)
     };
 
-    let result = lhs
-        .binary(rhs, Operator::Add)?
-        .execute::<Columnar>(&mut ctx)?;
+    let result = lhs.binary(rhs, op.into())?.execute::<Columnar>(&mut ctx)?;
     assert!(matches!(&result, Columnar::Constant(_)));
     assert_arrays_eq!(
         result.into_array(),
-        DecimalArray::from_option_iter::<i64, _>(
-            [None, None],
-            result_decimal_dtype(dtype, NumericOperator::Add)?,
-        ),
+        DecimalArray::from_option_iter::<i256, _>([None, None], result_decimal_dtype(dtype, op)?,),
         &mut ctx
     );
     Ok(())
@@ -563,34 +645,45 @@ fn test_decimal_constant_wider_than_array_storage() -> VortexResult<()> {
     Ok(())
 }
 
-#[test]
-fn test_decimal_empty() -> VortexResult<()> {
+#[rstest]
+#[case::add(NumericOperator::Add)]
+#[case::sub(NumericOperator::Sub)]
+#[case::mul(NumericOperator::Mul)]
+#[case::div(NumericOperator::Div)]
+fn test_decimal_empty(#[case] op: NumericOperator) -> VortexResult<()> {
     let mut ctx = array_session().create_execution_ctx();
     let dtype = DecimalDType::new(10, 2);
     let empty = DecimalArray::from_iter::<i64, _>([], dtype).into_array();
 
-    let result = decimal_binary(empty.clone(), empty, Operator::Add)?;
+    let result = decimal_binary(empty.clone(), empty, op.into())?;
     assert_arrays_eq!(
         result,
-        DecimalArray::from_iter::<i64, _>([], result_decimal_dtype(dtype, NumericOperator::Add)?,),
+        DecimalArray::from_iter::<i256, _>([], result_decimal_dtype(dtype, op)?,),
         &mut ctx
     );
     Ok(())
 }
 
-#[test]
-fn test_decimal_constant_constant_folds() -> VortexResult<()> {
+#[rstest]
+#[case::add(NumericOperator::Add, 200)]
+#[case::sub(NumericOperator::Sub, 100)]
+#[case::mul(NumericOperator::Mul, 7_500)]
+#[case::div(NumericOperator::Div, 3_000_000)]
+fn test_decimal_constant_constant_folds(
+    #[case] op: NumericOperator,
+    #[case] expected: i128,
+) -> VortexResult<()> {
     let mut ctx = array_session().create_execution_ctx();
     let dtype = DecimalDType::new(10, 2);
     let lhs = decimal_constant(150i64, dtype, 3);
     let rhs = decimal_constant(50i64, dtype, 3);
 
-    let result = decimal_binary(lhs, rhs, Operator::Add)?;
+    let result = decimal_binary(lhs, rhs, op.into())?;
     assert_arrays_eq!(
         result,
-        DecimalArray::from_iter::<i64, _>(
-            [200, 200, 200],
-            result_decimal_dtype(dtype, NumericOperator::Add)?,
+        DecimalArray::from_iter::<i256, _>(
+            [i256::from_i128(expected); 3],
+            result_decimal_dtype(dtype, op)?,
         ),
         &mut ctx
     );
@@ -598,11 +691,22 @@ fn test_decimal_constant_constant_folds() -> VortexResult<()> {
 }
 
 #[rstest]
-#[case::mul(Operator::Mul)]
-#[case::div(Operator::Div)]
-fn test_decimal_mul_div_unsupported(#[case] op: Operator) {
-    let dtype = DecimalDType::new(10, 2);
-    let values = DecimalArray::from_iter::<i64, _>([100], dtype).into_array();
+#[case::add(NumericOperator::Add, DecimalDType::new(11, 2))]
+#[case::sub(NumericOperator::Sub, DecimalDType::new(11, 2))]
+#[case::mul(NumericOperator::Mul, DecimalDType::new(21, 4))]
+#[case::div(NumericOperator::Div, DecimalDType::new(16, 6))]
+fn test_decimal_result_dtype(
+    #[case] op: NumericOperator,
+    #[case] expected: DecimalDType,
+) -> VortexResult<()> {
+    assert_eq!(
+        result_decimal_dtype(DecimalDType::new(10, 2), op)?,
+        expected
+    );
+    Ok(())
+}
 
-    assert!(decimal_binary(values.clone(), values, op).is_err());
+#[test]
+fn test_decimal_mul_result_scale_overflow_errors() {
+    assert!(result_decimal_dtype(DecimalDType::new(40, 40), NumericOperator::Mul).is_err());
 }

From 61cccbf202017ad808c39bd5c75c0e19e82c0f94 Mon Sep 17 00:00:00 2001
From: Matt Katz <mhkatz97@gmail.com>
Date: Wed, 5 Aug 2026 16:00:25 -0700
Subject: [PATCH 2/7] always bounds-check decimal array mul

Signed-off-by: Matt Katz <mhkatz97@gmail.com>
---
 .../scalar_fn/fns/binary/numeric/decimal.rs   | 121 +++++++-----------
 .../src/scalar_fn/fns/binary/numeric/tests.rs |  27 +++-
 2 files changed, 74 insertions(+), 74 deletions(-)

diff --git a/vortex-array/src/scalar_fn/fns/binary/numeric/decimal.rs b/vortex-array/src/scalar_fn/fns/binary/numeric/decimal.rs
index 6b0e8d1bc82..d121c4990db 100644
--- a/vortex-array/src/scalar_fn/fns/binary/numeric/decimal.rs
+++ b/vortex-array/src/scalar_fn/fns/binary/numeric/decimal.rs
@@ -10,10 +10,11 @@
 //! scale follow Arrow's rules — see [`decimal_numeric_result_dtype`].
 //!
 //! Lanes execute in a working width wide enough that in-precision inputs cannot spuriously
-//! overflow an intermediate, then narrow to the result's own storage width. Multiplication skips
-//! the redundant overflow check when the input precision proves every product fits. Otherwise, an
-//! operation that overflows the result precision on a valid lane is an error; invalid lanes never
-//! error.
+//! overflow an intermediate, then narrow to the result's own storage width. Every lane is still
+//! checked at that width: [`DecimalArray`] does not validate its stored values against the
+//! declared precision, so an out-of-precision value can reach a kernel and must not be able to
+//! overflow it. An operation that overflows the result precision on a valid lane is an error;
+//! invalid lanes never error.
 
 use std::ops::Mul;
 
@@ -40,7 +41,6 @@ use crate::arrays::decimal::DecimalArrayExt;
 use crate::dtype::BigCast;
 use crate::dtype::DType;
 use crate::dtype::DecimalDType;
-use crate::dtype::MAX_PRECISION;
 use crate::dtype::NativeDecimalType;
 use crate::match_each_decimal_value_type;
 use crate::scalar::DecimalValue;
@@ -87,7 +87,7 @@ pub(super) fn execute_numeric_decimal(
 
     let work_dtype = decimal_numeric_work_dtype(*decimal_dtype, result_decimal_dtype, op);
     match_each_decimal_value_type!(DecimalType::smallest_decimal_value_type(&work_dtype), |W| {
-        let plan = DecimalOpPlan::<W>::new(*decimal_dtype, result_decimal_dtype, op)?;
+        let constants = DecimalOpConstants::<W>::new(result_decimal_dtype, op)?;
         match_each_decimal_value_type!(
             DecimalType::smallest_decimal_value_type(&result_decimal_dtype),
             |O| {
@@ -99,7 +99,7 @@ pub(super) fn execute_numeric_decimal(
                     &result_dtype,
                     validity,
                     &valid_rows,
-                    &plan,
+                    &constants,
                 )
             }
         )
@@ -168,45 +168,26 @@ impl DecimalOperand {
     }
 }
 
-/// Per-execution bounds for checked decimal lane operations at working width `W`.
+/// Per-execution constants for a decimal operation at working width `W`, hoisted out of the
+/// lane loop.
 ///
 /// Native-width checked arithmetic only detects overflow of `W`, whose range may exceed the
 /// declared decimal precision. In particular, `i256` can represent values outside precision 76,
 /// so every native result must also be checked against these logical bounds.
-struct DecimalValueBounds<W> {
+struct DecimalOpConstants<W> {
     /// Inclusive stored-value bounds implied by the result precision.
     lower_bound: W,
     upper_bound: W,
-}
-
-impl<W: NativeDecimalType> DecimalValueBounds<W> {
-    fn new(dtype: DecimalDType) -> Self {
-        let precision = usize::from(dtype.precision());
-        Self {
-            lower_bound: W::MIN_BY_PRECISION[precision],
-            upper_bound: W::MAX_BY_PRECISION[precision],
-        }
-    }
-
-    /// Bounds-check a candidate result against the result precision.
-    fn in_precision(&self, value: W) -> Option<W> {
-        (self.lower_bound <= value && value <= self.upper_bound).then_some(value)
-    }
-}
-
-/// Per-execution constants for a decimal operation at working width `W`.
-struct DecimalOpPlan<W> {
-    bounds: DecimalValueBounds<W>,
+    /// Arrow's division rescaling factors. Both are one for every other operator.
     lhs_scale_factor: W,
     rhs_scale_factor: W,
-    guaranteed_mul: bool,
 }
 
-impl<W> DecimalOpPlan<W>
+impl<W> DecimalOpConstants<W>
 where
     W: NativeDecimalType + CheckedMul,
 {
-    fn new(input: DecimalDType, result: DecimalDType, op: NumericOperator) -> VortexResult<Self> {
+    fn new(result: DecimalDType, op: NumericOperator) -> VortexResult<Self> {
         let one = cast_work_value::<W, i8>(1);
         let (lhs_scale_factor, rhs_scale_factor) = if op == NumericOperator::Div {
             // Arrow scales the quotient by 10^(result_scale - lhs_scale + rhs_scale). Both
@@ -222,14 +203,19 @@ where
             (one, one)
         };
 
+        let precision = usize::from(result.precision());
         Ok(Self {
-            bounds: DecimalValueBounds::new(result),
+            lower_bound: W::MIN_BY_PRECISION[precision],
+            upper_bound: W::MAX_BY_PRECISION[precision],
             lhs_scale_factor,
             rhs_scale_factor,
-            guaranteed_mul: op == NumericOperator::Mul
-                && input.precision().saturating_mul(2) <= MAX_PRECISION,
         })
     }
+
+    /// Bounds-check a candidate result against the result precision.
+    fn in_precision(&self, value: W) -> Option<W> {
+        (self.lower_bound <= value && value <= self.upper_bound).then_some(value)
+    }
 }
 
 fn decimal_scale_factor<W>(exp: u32) -> VortexResult<W>
@@ -253,7 +239,7 @@ where
 trait CheckedDecimalOp {
     const ERROR: &'static str;
 
-    fn apply<W>(lhs: W, rhs: W, plan: &DecimalOpPlan<W>) -> Option<W>
+    fn apply<W>(lhs: W, rhs: W, constants: &DecimalOpConstants<W>) -> Option<W>
     where
         W: NativeDecimalType + CheckedAdd + CheckedSub + CheckedMul + CheckedDiv + Mul<Output = W>;
 }
@@ -264,64 +250,51 @@ struct CheckedDecimalSub;
 
 struct CheckedDecimalMul;
 
-struct GuaranteedDecimalMul;
-
 struct CheckedDecimalDiv;
 
 impl CheckedDecimalOp for CheckedDecimalAdd {
     const ERROR: &'static str = "decimal overflow in checked add";
 
-    fn apply<W>(lhs: W, rhs: W, plan: &DecimalOpPlan<W>) -> Option<W>
+    fn apply<W>(lhs: W, rhs: W, constants: &DecimalOpConstants<W>) -> Option<W>
     where
         W: NativeDecimalType + CheckedAdd + CheckedSub + CheckedMul + CheckedDiv + Mul<Output = W>,
     {
-        plan.bounds.in_precision(lhs.checked_add(&rhs)?)
+        constants.in_precision(lhs.checked_add(&rhs)?)
     }
 }
 
 impl CheckedDecimalOp for CheckedDecimalSub {
     const ERROR: &'static str = "decimal overflow in checked sub";
 
-    fn apply<W>(lhs: W, rhs: W, plan: &DecimalOpPlan<W>) -> Option<W>
+    fn apply<W>(lhs: W, rhs: W, constants: &DecimalOpConstants<W>) -> Option<W>
     where
         W: NativeDecimalType + CheckedAdd + CheckedSub + CheckedMul + CheckedDiv + Mul<Output = W>,
     {
-        plan.bounds.in_precision(lhs.checked_sub(&rhs)?)
+        constants.in_precision(lhs.checked_sub(&rhs)?)
     }
 }
 
 impl CheckedDecimalOp for CheckedDecimalMul {
     const ERROR: &'static str = "decimal overflow in checked mul";
 
-    fn apply<W>(lhs: W, rhs: W, plan: &DecimalOpPlan<W>) -> Option<W>
+    fn apply<W>(lhs: W, rhs: W, constants: &DecimalOpConstants<W>) -> Option<W>
     where
         W: NativeDecimalType + CheckedAdd + CheckedSub + CheckedMul + CheckedDiv + Mul<Output = W>,
     {
-        plan.bounds.in_precision(lhs.checked_mul(&rhs)?)
-    }
-}
-
-impl CheckedDecimalOp for GuaranteedDecimalMul {
-    const ERROR: &'static str = "decimal overflow in guaranteed mul";
-
-    fn apply<W>(lhs: W, rhs: W, _plan: &DecimalOpPlan<W>) -> Option<W>
-    where
-        W: NativeDecimalType + CheckedAdd + CheckedSub + CheckedMul + CheckedDiv + Mul<Output = W>,
-    {
-        Some(lhs * rhs)
+        constants.in_precision(lhs.checked_mul(&rhs)?)
     }
 }
 
 impl CheckedDecimalOp for CheckedDecimalDiv {
     const ERROR: &'static str = "decimal overflow or division by zero in checked div";
 
-    fn apply<W>(lhs: W, rhs: W, plan: &DecimalOpPlan<W>) -> Option<W>
+    fn apply<W>(lhs: W, rhs: W, constants: &DecimalOpConstants<W>) -> Option<W>
     where
         W: NativeDecimalType + CheckedAdd + CheckedSub + CheckedMul + CheckedDiv + Mul<Output = W>,
     {
-        let lhs = lhs.checked_mul(&plan.lhs_scale_factor)?;
-        let rhs = rhs.checked_mul(&plan.rhs_scale_factor)?;
-        plan.bounds.in_precision(lhs.checked_div(&rhs)?)
+        let lhs = lhs.checked_mul(&constants.lhs_scale_factor)?;
+        let rhs = rhs.checked_mul(&constants.rhs_scale_factor)?;
+        constants.in_precision(lhs.checked_div(&rhs)?)
     }
 }
 
@@ -334,7 +307,7 @@ fn execute_decimal_at_widths<W, O>(
     result_dtype: &DType,
     validity: Validity,
     valid_rows: &Mask,
-    plan: &DecimalOpPlan<W>,
+    constants: &DecimalOpConstants<W>,
 ) -> VortexResult<ArrayRef>
 where
     W: NativeDecimalType + CheckedAdd + CheckedSub + CheckedMul + CheckedDiv + Mul<Output = W>,
@@ -350,7 +323,7 @@ where
                 result_dtype,
                 validity,
                 valid_rows,
-                plan,
+                constants,
             )
         };
     }
@@ -358,10 +331,6 @@ where
     match op {
         NumericOperator::Add => execute_typed!(CheckedDecimalAdd),
         NumericOperator::Sub => execute_typed!(CheckedDecimalSub),
-        // A product of two valid p-digit values needs at most 2p digits. When 2p is within
-        // MAX_PRECISION, both native and declared-precision overflow are impossible at the
-        // selected working width. This includes p=38 producing a precision-76 i256 result.
-        NumericOperator::Mul if plan.guaranteed_mul => execute_typed!(GuaranteedDecimalMul),
         NumericOperator::Mul => execute_typed!(CheckedDecimalMul),
         NumericOperator::Div => execute_typed!(CheckedDecimalDiv),
     }
@@ -374,7 +343,7 @@ fn execute_decimal_typed<W, O, Op>(
     result_dtype: &DType,
     validity: Validity,
     valid_rows: &Mask,
-    plan: &DecimalOpPlan<W>,
+    constants: &DecimalOpConstants<W>,
 ) -> VortexResult<ArrayRef>
 where
     W: NativeDecimalType + CheckedAdd + CheckedSub + CheckedMul + CheckedDiv + Mul<Output = W>,
@@ -386,14 +355,15 @@ where
 
     let values = match (lhs, rhs) {
         (DecimalOperand::Array { values: lhs, .. }, DecimalOperand::Array { values: rhs, .. }) => {
-            checked_decimal_arrays::<W, O, Op>(lhs, rhs, plan, valid_rows)
+            checked_decimal_arrays::<W, O, Op>(lhs, rhs, constants, valid_rows)
         }
         (DecimalOperand::Array { values: lhs, .. }, DecimalOperand::Constant { value, .. }) => {
             let rhs = typed_constant::<W>(value);
             match_each_decimal_value_type!(lhs.values_type(), |L| {
                 let lhs = lhs.buffer::<L>();
                 checked_lanes(lhs.as_slice(), valid_rows, |lhs| {
-                    Op::apply(<W as BigCast>::from(lhs)?, rhs, plan).map(cast_result_value::<W, O>)
+                    Op::apply(<W as BigCast>::from(lhs)?, rhs, constants)
+                        .map(cast_result_value::<W, O>)
                 })
             })
         }
@@ -402,7 +372,8 @@ where
             match_each_decimal_value_type!(rhs.values_type(), |R| {
                 let rhs = rhs.buffer::<R>();
                 checked_lanes(rhs.as_slice(), valid_rows, |rhs| {
-                    Op::apply(lhs, <W as BigCast>::from(rhs)?, plan).map(cast_result_value::<W, O>)
+                    Op::apply(lhs, <W as BigCast>::from(rhs)?, constants)
+                        .map(cast_result_value::<W, O>)
                 })
             })
         }
@@ -412,7 +383,7 @@ where
         ) => {
             let lhs = typed_constant::<W>(lhs);
             let rhs = typed_constant::<W>(rhs);
-            let value = Op::apply(lhs, rhs, plan)
+            let value = Op::apply(lhs, rhs, constants)
                 .ok_or_else(|| vortex_err!(InvalidArgument: "{}", Op::ERROR))?;
             let value = cast_result_value::<W, O>(value);
             return Ok(ConstantArray::new(
@@ -439,7 +410,7 @@ where
 fn checked_decimal_arrays<W, O, Op>(
     lhs: &DecimalArray,
     rhs: &DecimalArray,
-    plan: &DecimalOpPlan<W>,
+    constants: &DecimalOpConstants<W>,
     valid_rows: &Mask,
 ) -> Result<Buffer<O>, usize>
 where
@@ -456,8 +427,12 @@ where
                 LaneZip::new(lhs.as_slice(), rhs.as_slice()),
                 valid_rows,
                 |(lhs, rhs)| {
-                    Op::apply(<W as BigCast>::from(lhs)?, <W as BigCast>::from(rhs)?, plan)
-                        .map(cast_result_value::<W, O>)
+                    Op::apply(
+                        <W as BigCast>::from(lhs)?,
+                        <W as BigCast>::from(rhs)?,
+                        constants,
+                    )
+                    .map(cast_result_value::<W, O>)
                 },
             )
         })
diff --git a/vortex-array/src/scalar_fn/fns/binary/numeric/tests.rs b/vortex-array/src/scalar_fn/fns/binary/numeric/tests.rs
index b998eac7873..d36b436581b 100644
--- a/vortex-array/src/scalar_fn/fns/binary/numeric/tests.rs
+++ b/vortex-array/src/scalar_fn/fns/binary/numeric/tests.rs
@@ -335,6 +335,31 @@ fn test_decimal_value_outside_working_width_errors() {
     assert!(decimal_binary(lhs, rhs, Operator::Add).is_err());
 }
 
+#[test]
+fn test_decimal_mul_value_outside_precision_errors() {
+    // `DecimalArray::new` does not validate stored values against the declared precision, so Mul
+    // cannot assume its inputs are in-precision: 500 * 500 is 250_000, well past the 99_999 that
+    // the decimal(5, 0) result can represent.
+    let dtype = DecimalDType::new(2, 0);
+    let value = i256::from_i128(500);
+    let lhs = DecimalArray::new(buffer![value], dtype, Validity::NonNullable).into_array();
+    let rhs = DecimalArray::new(buffer![value], dtype, Validity::NonNullable).into_array();
+
+    assert!(decimal_binary(lhs, rhs, Operator::Mul).is_err());
+}
+
+#[test]
+fn test_decimal_mul_value_outside_working_width_errors() {
+    // 50_000 * 50_000 overflows the i32 working width chosen for a decimal(5, 0) result, which
+    // an unchecked multiply would wrap in release and panic on in debug.
+    let dtype = DecimalDType::new(2, 0);
+    let value = i256::from_i128(50_000);
+    let lhs = DecimalArray::new(buffer![value], dtype, Validity::NonNullable).into_array();
+    let rhs = DecimalArray::new(buffer![value], dtype, Validity::NonNullable).into_array();
+
+    assert!(decimal_binary(lhs, rhs, Operator::Mul).is_err());
+}
+
 #[test]
 fn test_decimal_value_outside_working_width_on_null_lane_ignored() -> VortexResult<()> {
     let mut ctx = array_session().create_execution_ctx();
@@ -446,7 +471,7 @@ fn test_decimal_mul_widens_before_multiplying() -> VortexResult<()> {
 }
 
 #[test]
-fn test_decimal_mul_above_guaranteed_precision_checks_logical_bounds() {
+fn test_decimal_mul_above_result_precision_errors() {
     let dtype = DecimalDType::new(39, 0);
     let one = i256::from_i128(1);
     let ten_to_38 = <i256 as NativeDecimalType>::MAX_BY_PRECISION[38] + one;

From 6efe3d435b6eeec7f9b136ac23e44461a353cf63 Mon Sep 17 00:00:00 2001
From: Matt Katz <mhkatz97@gmail.com>
Date: Wed, 5 Aug 2026 16:14:53 -0700
Subject: [PATCH 3/7] drop arrow comparison benches

Signed-off-by: Matt Katz <mhkatz97@gmail.com>
---
 Cargo.lock                         |   2 -
 vortex-array/Cargo.toml            |   2 -
 vortex-array/benches/binary_ops.rs | 156 ++++-------------------------
 3 files changed, 18 insertions(+), 142 deletions(-)

diff --git a/Cargo.lock b/Cargo.lock
index 9b6872d4e1c..81c0bb17416 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -9456,8 +9456,6 @@ dependencies = [
  "arbitrary",
  "arc-swap",
  "arcref",
- "arrow-arith 58.4.0",
- "arrow-array 58.4.0",
  "arrow-buffer 58.4.0",
  "async-lock",
  "bytes",
diff --git a/vortex-array/Cargo.toml b/vortex-array/Cargo.toml
index f87b07d5b78..70fba4acdcf 100644
--- a/vortex-array/Cargo.toml
+++ b/vortex-array/Cargo.toml
@@ -82,8 +82,6 @@ _test-harness = ["dep:goldenfile", "dep:rstest", "dep:rstest_reuse"]
 serde = ["dep:serde", "vortex-buffer/serde", "vortex-mask/serde"]
 
 [dev-dependencies]
-arrow-arith = { workspace = true }
-arrow-array = { workspace = true }
 divan = { workspace = true }
 futures = { workspace = true, features = ["executor"] }
 insta = { workspace = true }
diff --git a/vortex-array/benches/binary_ops.rs b/vortex-array/benches/binary_ops.rs
index 8af51b1c1ec..094c8ee820a 100644
--- a/vortex-array/benches/binary_ops.rs
+++ b/vortex-array/benches/binary_ops.rs
@@ -9,16 +9,6 @@
 
 use std::sync::LazyLock;
 
-use arrow_arith::numeric::div as arrow_div;
-use arrow_arith::numeric::mul as arrow_mul;
-use arrow_array::ArrowPrimitiveType;
-use arrow_array::PrimitiveArray as ArrowPrimitiveArray;
-use arrow_array::types::Decimal32Type;
-use arrow_array::types::Decimal64Type;
-use arrow_array::types::Decimal128Type;
-use arrow_array::types::Decimal256Type;
-use arrow_array::types::DecimalType as ArrowDecimalType;
-use arrow_buffer::ArrowNativeType;
 use divan::Bencher;
 use divan::counter::ItemsCount;
 use mimalloc::MiMalloc;
@@ -202,92 +192,45 @@ fn add_decimal_i128_nullable(bencher: Bencher) {
     bench_decimal(bencher, lhs, rhs, Operator::Add);
 }
 
-macro_rules! decimal_compare_benches {
-    (
-        $vortex_mul:ident,
-        $arrow_mul:ident,
-        $vortex_div:ident,
-        $arrow_div:ident,
-        $vortex_type:ty,
-        $arrow_type:ty,
-        $precision:expr,
-        $scale:expr
-    ) => {
+macro_rules! decimal_mul_div_benches {
+    ($mul:ident, $div:ident, $native:ty, $precision:expr, $scale:expr) => {
         #[divan::bench]
-        fn $vortex_mul(bencher: Bencher) {
+        fn $mul(bencher: Bencher) {
             let dtype = DecimalDType::new($precision, $scale);
-            let lhs = comparison_vortex_decimal::<$vortex_type>(dtype, 1).into_array();
-            let rhs = comparison_vortex_decimal::<$vortex_type>(dtype, 17).into_array();
+            let lhs = comparison_vortex_decimal::<$native>(dtype, 1).into_array();
+            let rhs = comparison_vortex_decimal::<$native>(dtype, 17).into_array();
             bench_decimal(bencher, lhs, rhs, Operator::Mul);
         }
 
         #[divan::bench]
-        fn $arrow_mul(bencher: Bencher) {
-            let lhs = comparison_arrow_decimal::<$arrow_type>($precision, $scale, 1);
-            let rhs = comparison_arrow_decimal::<$arrow_type>($precision, $scale, 17);
-            bench_arrow_decimal(bencher, lhs, rhs, ArrowDecimalOp::Mul);
-        }
-
-        #[divan::bench]
-        fn $vortex_div(bencher: Bencher) {
+        fn $div(bencher: Bencher) {
             let dtype = DecimalDType::new($precision, $scale);
-            let lhs = comparison_vortex_decimal::<$vortex_type>(dtype, 1).into_array();
-            let rhs = comparison_vortex_decimal::<$vortex_type>(dtype, 17).into_array();
+            let lhs = comparison_vortex_decimal::<$native>(dtype, 1).into_array();
+            let rhs = comparison_vortex_decimal::<$native>(dtype, 17).into_array();
             bench_decimal(bencher, lhs, rhs, Operator::Div);
         }
-
-        #[divan::bench]
-        fn $arrow_div(bencher: Bencher) {
-            let lhs = comparison_arrow_decimal::<$arrow_type>($precision, $scale, 1);
-            let rhs = comparison_arrow_decimal::<$arrow_type>($precision, $scale, 17);
-            bench_arrow_decimal(bencher, lhs, rhs, ArrowDecimalOp::Div);
-        }
     };
 }
 
-decimal_compare_benches!(
-    vortex_mul_decimal_i32_nonnull,
-    arrow_mul_decimal_i32_nonnull,
-    vortex_div_decimal_i32_nonnull,
-    arrow_div_decimal_i32_nonnull,
-    i32,
-    Decimal32Type,
-    4,
-    1
-);
-decimal_compare_benches!(
-    vortex_mul_decimal_i64_nonnull,
-    arrow_mul_decimal_i64_nonnull,
-    vortex_div_decimal_i64_nonnull,
-    arrow_div_decimal_i64_nonnull,
-    i64,
-    Decimal64Type,
-    8,
-    2
-);
-decimal_compare_benches!(
-    vortex_mul_decimal_i128_nonnull,
-    arrow_mul_decimal_i128_nonnull,
-    vortex_div_decimal_i128_nonnull,
-    arrow_div_decimal_i128_nonnull,
+decimal_mul_div_benches!(mul_decimal_i32_nonnull, div_decimal_i32_nonnull, i32, 4, 1);
+decimal_mul_div_benches!(mul_decimal_i64_nonnull, div_decimal_i64_nonnull, i64, 8, 2);
+decimal_mul_div_benches!(
+    mul_decimal_i128_nonnull,
+    div_decimal_i128_nonnull,
     i128,
-    Decimal128Type,
     18,
     2
 );
-decimal_compare_benches!(
-    vortex_mul_decimal_i256_nonnull,
-    arrow_mul_decimal_i256_nonnull,
-    vortex_div_decimal_i256_nonnull,
-    arrow_div_decimal_i256_nonnull,
+decimal_mul_div_benches!(
+    mul_decimal_i256_nonnull,
+    div_decimal_i256_nonnull,
     vortex_array::dtype::i256,
-    Decimal256Type,
     38,
     2
 );
 
 #[divan::bench]
-fn vortex_mul_decimal_i256_nullable(bencher: Bencher) {
+fn mul_decimal_i256_nullable(bencher: Bencher) {
     let dtype = DecimalDType::new(38, 2);
     let lhs =
         comparison_vortex_decimal_nullable::<vortex_array::dtype::i256>(dtype, 1, 7).into_array();
@@ -297,14 +240,7 @@ fn vortex_mul_decimal_i256_nullable(bencher: Bencher) {
 }
 
 #[divan::bench]
-fn arrow_mul_decimal_i256_nullable(bencher: Bencher) {
-    let lhs = comparison_arrow_decimal_nullable::<Decimal256Type>(38, 2, 1, 7);
-    let rhs = comparison_arrow_decimal_nullable::<Decimal256Type>(38, 2, 17, 5);
-    bench_arrow_decimal(bencher, lhs, rhs, ArrowDecimalOp::Mul);
-}
-
-#[divan::bench]
-fn vortex_div_decimal_i256_nullable(bencher: Bencher) {
+fn div_decimal_i256_nullable(bencher: Bencher) {
     let dtype = DecimalDType::new(38, 2);
     let lhs =
         comparison_vortex_decimal_nullable::<vortex_array::dtype::i256>(dtype, 1, 7).into_array();
@@ -313,13 +249,6 @@ fn vortex_div_decimal_i256_nullable(bencher: Bencher) {
     bench_decimal(bencher, lhs, rhs, Operator::Div);
 }
 
-#[divan::bench]
-fn arrow_div_decimal_i256_nullable(bencher: Bencher) {
-    let lhs = comparison_arrow_decimal_nullable::<Decimal256Type>(38, 2, 1, 7);
-    let rhs = comparison_arrow_decimal_nullable::<Decimal256Type>(38, 2, 17, 5);
-    bench_arrow_decimal(bencher, lhs, rhs, ArrowDecimalOp::Div);
-}
-
 #[divan::bench]
 fn eq_i64_constant(bencher: Bencher) {
     let lhs = primitive_nonnull(0).into_array();
@@ -360,28 +289,6 @@ fn bench_decimal(bencher: Bencher, lhs: ArrayRef, rhs: ArrayRef, operator: Opera
     bench_binary::<DecimalArray>(bencher, lhs, rhs, operator);
 }
 
-#[derive(Clone, Copy)]
-enum ArrowDecimalOp {
-    Mul,
-    Div,
-}
-
-fn bench_arrow_decimal<T>(
-    bencher: Bencher,
-    lhs: ArrowPrimitiveArray<T>,
-    rhs: ArrowPrimitiveArray<T>,
-    op: ArrowDecimalOp,
-) where
-    T: ArrowPrimitiveType,
-{
-    bencher
-        .counter(ItemsCount::new(LEN))
-        .bench_local(|| match op {
-            ArrowDecimalOp::Mul => arrow_mul(&lhs, &rhs).unwrap(),
-            ArrowDecimalOp::Div => arrow_div(&lhs, &rhs).unwrap(),
-        });
-}
-
 fn bench_bool(bencher: Bencher, lhs: ArrayRef, rhs: ArrayRef, operator: Operator) {
     bench_binary::<BoolArray>(bencher, lhs, rhs, operator);
 }
@@ -445,33 +352,6 @@ where
     )
 }
 
-fn comparison_arrow_decimal<T>(precision: u8, scale: i8, offset: usize) -> ArrowPrimitiveArray<T>
-where
-    T: ArrowPrimitiveType + ArrowDecimalType,
-{
-    ArrowPrimitiveArray::<T>::from_iter_values(
-        (0..LEN).map(|idx| T::Native::usize_as((idx + offset) % 89 + 1)),
-    )
-    .with_precision_and_scale(precision, scale)
-    .unwrap()
-}
-
-fn comparison_arrow_decimal_nullable<T>(
-    precision: u8,
-    scale: i8,
-    offset: usize,
-    null_every: usize,
-) -> ArrowPrimitiveArray<T>
-where
-    T: ArrowPrimitiveType + ArrowDecimalType,
-{
-    ArrowPrimitiveArray::<T>::from_iter((0..LEN).map(|idx| {
-        (!idx.is_multiple_of(null_every)).then(|| T::Native::usize_as((idx + offset) % 89 + 1))
-    }))
-    .with_precision_and_scale(precision, scale)
-    .unwrap()
-}
-
 fn primitive_small_nonnull(offset: i64) -> PrimitiveArray {
     PrimitiveArray::from_iter((0..LEN as i64).map(|i| ((i + offset) % 1024) + 1))
 }

From 70a969d749639bc4024af5f0291d2c4738c7de94 Mon Sep 17 00:00:00 2001
From: Matt Katz <mhkatz97@gmail.com>
Date: Wed, 5 Aug 2026 16:18:03 -0700
Subject: [PATCH 4/7] simplify decimal mul/div benches

Signed-off-by: Matt Katz <mhkatz97@gmail.com>
---
 vortex-array/benches/binary_ops.rs | 100 +++++++----------------------
 1 file changed, 23 insertions(+), 77 deletions(-)

diff --git a/vortex-array/benches/binary_ops.rs b/vortex-array/benches/binary_ops.rs
index 094c8ee820a..3950d286f46 100644
--- a/vortex-array/benches/binary_ops.rs
+++ b/vortex-array/benches/binary_ops.rs
@@ -22,9 +22,7 @@ use vortex_array::arrays::ConstantArray;
 use vortex_array::arrays::DecimalArray;
 use vortex_array::arrays::PrimitiveArray;
 use vortex_array::builtins::ArrayBuiltins;
-use vortex_array::dtype::BigCast;
 use vortex_array::dtype::DecimalDType;
-use vortex_array::dtype::NativeDecimalType;
 use vortex_array::scalar_fn::fns::operators::Operator;
 use vortex_session::VortexSession;
 
@@ -192,60 +190,35 @@ fn add_decimal_i128_nullable(bencher: Bencher) {
     bench_decimal(bencher, lhs, rhs, Operator::Add);
 }
 
-macro_rules! decimal_mul_div_benches {
-    ($mul:ident, $div:ident, $native:ty, $precision:expr, $scale:expr) => {
-        #[divan::bench]
-        fn $mul(bencher: Bencher) {
-            let dtype = DecimalDType::new($precision, $scale);
-            let lhs = comparison_vortex_decimal::<$native>(dtype, 1).into_array();
-            let rhs = comparison_vortex_decimal::<$native>(dtype, 17).into_array();
-            bench_decimal(bencher, lhs, rhs, Operator::Mul);
-        }
-
-        #[divan::bench]
-        fn $div(bencher: Bencher) {
-            let dtype = DecimalDType::new($precision, $scale);
-            let lhs = comparison_vortex_decimal::<$native>(dtype, 1).into_array();
-            let rhs = comparison_vortex_decimal::<$native>(dtype, 17).into_array();
-            bench_decimal(bencher, lhs, rhs, Operator::Div);
-        }
-    };
-}
-
-decimal_mul_div_benches!(mul_decimal_i32_nonnull, div_decimal_i32_nonnull, i32, 4, 1);
-decimal_mul_div_benches!(mul_decimal_i64_nonnull, div_decimal_i64_nonnull, i64, 8, 2);
-decimal_mul_div_benches!(
-    mul_decimal_i128_nonnull,
-    div_decimal_i128_nonnull,
-    i128,
-    18,
-    2
-);
-decimal_mul_div_benches!(
-    mul_decimal_i256_nonnull,
-    div_decimal_i256_nonnull,
-    vortex_array::dtype::i256,
-    38,
-    2
-);
+#[divan::bench]
+fn mul_decimal_i64_nonnull(bencher: Bencher) {
+    let lhs = decimal_i64_nonnull(0).into_array();
+    let rhs = decimal_i64_nonnull(1_000_000).into_array();
+
+    bench_decimal(bencher, lhs, rhs, Operator::Mul);
+}
 
 #[divan::bench]
-fn mul_decimal_i256_nullable(bencher: Bencher) {
-    let dtype = DecimalDType::new(38, 2);
-    let lhs =
-        comparison_vortex_decimal_nullable::<vortex_array::dtype::i256>(dtype, 1, 7).into_array();
-    let rhs =
-        comparison_vortex_decimal_nullable::<vortex_array::dtype::i256>(dtype, 17, 5).into_array();
+fn mul_decimal_i128_nullable(bencher: Bencher) {
+    let lhs = decimal_i128_nullable(0, 7).into_array();
+    let rhs = decimal_i128_nullable(1_000_000, 5).into_array();
+
     bench_decimal(bencher, lhs, rhs, Operator::Mul);
 }
 
 #[divan::bench]
-fn div_decimal_i256_nullable(bencher: Bencher) {
-    let dtype = DecimalDType::new(38, 2);
-    let lhs =
-        comparison_vortex_decimal_nullable::<vortex_array::dtype::i256>(dtype, 1, 7).into_array();
-    let rhs =
-        comparison_vortex_decimal_nullable::<vortex_array::dtype::i256>(dtype, 17, 5).into_array();
+fn div_decimal_i64_nonnull(bencher: Bencher) {
+    let lhs = decimal_i64_nonnull(0).into_array();
+    let rhs = decimal_i64_nonnull(1_000_000).into_array();
+
+    bench_decimal(bencher, lhs, rhs, Operator::Div);
+}
+
+#[divan::bench]
+fn div_decimal_i128_nullable(bencher: Bencher) {
+    let lhs = decimal_i128_nullable(0, 7).into_array();
+    let rhs = decimal_i128_nullable(1_000_000, 5).into_array();
+
     bench_decimal(bencher, lhs, rhs, Operator::Div);
 }
 
@@ -325,33 +298,6 @@ fn decimal_i128_nullable(base: i128, null_every: usize) -> DecimalArray {
     )
 }
 
-fn comparison_vortex_decimal<T>(dtype: DecimalDType, offset: usize) -> DecimalArray
-where
-    T: NativeDecimalType,
-{
-    DecimalArray::from_iter::<T, _>(
-        (0..LEN).map(|idx| <T as BigCast>::from(((idx + offset) % 89 + 1) as i64).unwrap()),
-        dtype,
-    )
-}
-
-fn comparison_vortex_decimal_nullable<T>(
-    dtype: DecimalDType,
-    offset: usize,
-    null_every: usize,
-) -> DecimalArray
-where
-    T: NativeDecimalType,
-{
-    DecimalArray::from_option_iter::<T, _>(
-        (0..LEN).map(|idx| {
-            (!idx.is_multiple_of(null_every))
-                .then(|| <T as BigCast>::from(((idx + offset) % 89 + 1) as i64).unwrap())
-        }),
-        dtype,
-    )
-}
-
 fn primitive_small_nonnull(offset: i64) -> PrimitiveArray {
     PrimitiveArray::from_iter((0..LEN as i64).map(|i| ((i + offset) % 1024) + 1))
 }

From a6916141c2859b6683193e5d21d91e621c0087fe Mon Sep 17 00:00:00 2001
From: Matt Katz <mhkatz97@gmail.com>
Date: Wed, 5 Aug 2026 16:21:13 -0700
Subject: [PATCH 5/7] keep DecimalValueBounds separate

Signed-off-by: Matt Katz <mhkatz97@gmail.com>
---
 .../scalar_fn/fns/binary/numeric/decimal.rs   | 43 ++++++++++++-------
 1 file changed, 28 insertions(+), 15 deletions(-)

diff --git a/vortex-array/src/scalar_fn/fns/binary/numeric/decimal.rs b/vortex-array/src/scalar_fn/fns/binary/numeric/decimal.rs
index d121c4990db..a34d06e15d9 100644
--- a/vortex-array/src/scalar_fn/fns/binary/numeric/decimal.rs
+++ b/vortex-array/src/scalar_fn/fns/binary/numeric/decimal.rs
@@ -168,16 +168,36 @@ impl DecimalOperand {
     }
 }
 
-/// Per-execution constants for a decimal operation at working width `W`, hoisted out of the
-/// lane loop.
+/// Per-execution bounds for checked decimal lane operations at working width `W`.
 ///
 /// Native-width checked arithmetic only detects overflow of `W`, whose range may exceed the
 /// declared decimal precision. In particular, `i256` can represent values outside precision 76,
 /// so every native result must also be checked against these logical bounds.
-struct DecimalOpConstants<W> {
+struct DecimalValueBounds<W> {
     /// Inclusive stored-value bounds implied by the result precision.
     lower_bound: W,
     upper_bound: W,
+}
+
+impl<W: NativeDecimalType> DecimalValueBounds<W> {
+    fn new(dtype: DecimalDType) -> Self {
+        let precision = usize::from(dtype.precision());
+        Self {
+            lower_bound: W::MIN_BY_PRECISION[precision],
+            upper_bound: W::MAX_BY_PRECISION[precision],
+        }
+    }
+
+    /// Bounds-check a candidate result against the result precision.
+    fn in_precision(&self, value: W) -> Option<W> {
+        (self.lower_bound <= value && value <= self.upper_bound).then_some(value)
+    }
+}
+
+/// Per-execution constants for a decimal operation at working width `W`, hoisted out of the
+/// lane loop.
+struct DecimalOpConstants<W> {
+    bounds: DecimalValueBounds<W>,
     /// Arrow's division rescaling factors. Both are one for every other operator.
     lhs_scale_factor: W,
     rhs_scale_factor: W,
@@ -203,19 +223,12 @@ where
             (one, one)
         };
 
-        let precision = usize::from(result.precision());
         Ok(Self {
-            lower_bound: W::MIN_BY_PRECISION[precision],
-            upper_bound: W::MAX_BY_PRECISION[precision],
+            bounds: DecimalValueBounds::new(result),
             lhs_scale_factor,
             rhs_scale_factor,
         })
     }
-
-    /// Bounds-check a candidate result against the result precision.
-    fn in_precision(&self, value: W) -> Option<W> {
-        (self.lower_bound <= value && value <= self.upper_bound).then_some(value)
-    }
 }
 
 fn decimal_scale_factor<W>(exp: u32) -> VortexResult<W>
@@ -259,7 +272,7 @@ impl CheckedDecimalOp for CheckedDecimalAdd {
     where
         W: NativeDecimalType + CheckedAdd + CheckedSub + CheckedMul + CheckedDiv + Mul<Output = W>,
     {
-        constants.in_precision(lhs.checked_add(&rhs)?)
+        constants.bounds.in_precision(lhs.checked_add(&rhs)?)
     }
 }
 
@@ -270,7 +283,7 @@ impl CheckedDecimalOp for CheckedDecimalSub {
     where
         W: NativeDecimalType + CheckedAdd + CheckedSub + CheckedMul + CheckedDiv + Mul<Output = W>,
     {
-        constants.in_precision(lhs.checked_sub(&rhs)?)
+        constants.bounds.in_precision(lhs.checked_sub(&rhs)?)
     }
 }
 
@@ -281,7 +294,7 @@ impl CheckedDecimalOp for CheckedDecimalMul {
     where
         W: NativeDecimalType + CheckedAdd + CheckedSub + CheckedMul + CheckedDiv + Mul<Output = W>,
     {
-        constants.in_precision(lhs.checked_mul(&rhs)?)
+        constants.bounds.in_precision(lhs.checked_mul(&rhs)?)
     }
 }
 
@@ -294,7 +307,7 @@ impl CheckedDecimalOp for CheckedDecimalDiv {
     {
         let lhs = lhs.checked_mul(&constants.lhs_scale_factor)?;
         let rhs = rhs.checked_mul(&constants.rhs_scale_factor)?;
-        constants.in_precision(lhs.checked_div(&rhs)?)
+        constants.bounds.in_precision(lhs.checked_div(&rhs)?)
     }
 }
 

From 4011a575df6c784d2a92d5a4a9093cb23d3fe4aa Mon Sep 17 00:00:00 2001
From: Matt Katz <mhkatz97@gmail.com>
Date: Wed, 5 Aug 2026 16:52:27 -0700
Subject: [PATCH 6/7] dispatch decimal kernels on one width

Signed-off-by: Matt Katz <mhkatz97@gmail.com>
---
 .../scalar_fn/fns/binary/numeric/decimal.rs   | 112 ++++++++----------
 .../src/scalar_fn/fns/binary/numeric/tests.rs |  19 +++
 2 files changed, 67 insertions(+), 64 deletions(-)

diff --git a/vortex-array/src/scalar_fn/fns/binary/numeric/decimal.rs b/vortex-array/src/scalar_fn/fns/binary/numeric/decimal.rs
index a34d06e15d9..d0e1bea8d7c 100644
--- a/vortex-array/src/scalar_fn/fns/binary/numeric/decimal.rs
+++ b/vortex-array/src/scalar_fn/fns/binary/numeric/decimal.rs
@@ -41,6 +41,7 @@ use crate::arrays::decimal::DecimalArrayExt;
 use crate::dtype::BigCast;
 use crate::dtype::DType;
 use crate::dtype::DecimalDType;
+use crate::dtype::DecimalType;
 use crate::dtype::NativeDecimalType;
 use crate::match_each_decimal_value_type;
 use crate::scalar::DecimalValue;
@@ -88,21 +89,26 @@ pub(super) fn execute_numeric_decimal(
     let work_dtype = decimal_numeric_work_dtype(*decimal_dtype, result_decimal_dtype, op);
     match_each_decimal_value_type!(DecimalType::smallest_decimal_value_type(&work_dtype), |W| {
         let constants = DecimalOpConstants::<W>::new(result_decimal_dtype, op)?;
-        match_each_decimal_value_type!(
-            DecimalType::smallest_decimal_value_type(&result_decimal_dtype),
-            |O| {
-                execute_decimal_at_widths::<W, O>(
+        macro_rules! execute_typed {
+            ($Op:ty) => {
+                execute_decimal_typed::<W, $Op>(
                     &lhs,
                     &rhs,
-                    op,
                     result_decimal_dtype,
                     &result_dtype,
                     validity,
                     &valid_rows,
                     &constants,
                 )
-            }
-        )
+            };
+        }
+
+        match op {
+            NumericOperator::Add => execute_typed!(CheckedDecimalAdd),
+            NumericOperator::Sub => execute_typed!(CheckedDecimalSub),
+            NumericOperator::Mul => execute_typed!(CheckedDecimalMul),
+            NumericOperator::Div => execute_typed!(CheckedDecimalDiv),
+        }
     })
 }
 
@@ -311,11 +317,9 @@ impl CheckedDecimalOp for CheckedDecimalDiv {
     }
 }
 
-#[expect(clippy::too_many_arguments, reason = "internal width-dispatch shim")]
-fn execute_decimal_at_widths<W, O>(
+fn execute_decimal_typed<W, Op>(
     lhs: &DecimalOperand,
     rhs: &DecimalOperand,
-    op: NumericOperator,
     result_decimal_dtype: DecimalDType,
     result_dtype: &DType,
     validity: Validity,
@@ -324,51 +328,14 @@ fn execute_decimal_at_widths<W, O>(
 ) -> VortexResult<ArrayRef>
 where
     W: NativeDecimalType + CheckedAdd + CheckedSub + CheckedMul + CheckedDiv + Mul<Output = W>,
-    O: NativeDecimalType,
-    DecimalValue: From<O>,
-{
-    macro_rules! execute_typed {
-        ($Op:ty) => {
-            execute_decimal_typed::<W, O, $Op>(
-                lhs,
-                rhs,
-                result_decimal_dtype,
-                result_dtype,
-                validity,
-                valid_rows,
-                constants,
-            )
-        };
-    }
-
-    match op {
-        NumericOperator::Add => execute_typed!(CheckedDecimalAdd),
-        NumericOperator::Sub => execute_typed!(CheckedDecimalSub),
-        NumericOperator::Mul => execute_typed!(CheckedDecimalMul),
-        NumericOperator::Div => execute_typed!(CheckedDecimalDiv),
-    }
-}
-
-fn execute_decimal_typed<W, O, Op>(
-    lhs: &DecimalOperand,
-    rhs: &DecimalOperand,
-    result_decimal_dtype: DecimalDType,
-    result_dtype: &DType,
-    validity: Validity,
-    valid_rows: &Mask,
-    constants: &DecimalOpConstants<W>,
-) -> VortexResult<ArrayRef>
-where
-    W: NativeDecimalType + CheckedAdd + CheckedSub + CheckedMul + CheckedDiv + Mul<Output = W>,
-    O: NativeDecimalType,
-    DecimalValue: From<O>,
+    DecimalValue: From<W>,
     Op: CheckedDecimalOp,
 {
     let len = lhs.len();
 
     let values = match (lhs, rhs) {
         (DecimalOperand::Array { values: lhs, .. }, DecimalOperand::Array { values: rhs, .. }) => {
-            checked_decimal_arrays::<W, O, Op>(lhs, rhs, constants, valid_rows)
+            checked_decimal_arrays::<W, Op>(lhs, rhs, constants, valid_rows)
         }
         (DecimalOperand::Array { values: lhs, .. }, DecimalOperand::Constant { value, .. }) => {
             let rhs = typed_constant::<W>(value);
@@ -376,7 +343,6 @@ where
                 let lhs = lhs.buffer::<L>();
                 checked_lanes(lhs.as_slice(), valid_rows, |lhs| {
                     Op::apply(<W as BigCast>::from(lhs)?, rhs, constants)
-                        .map(cast_result_value::<W, O>)
                 })
             })
         }
@@ -386,7 +352,6 @@ where
                 let rhs = rhs.buffer::<R>();
                 checked_lanes(rhs.as_slice(), valid_rows, |rhs| {
                     Op::apply(lhs, <W as BigCast>::from(rhs)?, constants)
-                        .map(cast_result_value::<W, O>)
                 })
             })
         }
@@ -398,13 +363,11 @@ where
             let rhs = typed_constant::<W>(rhs);
             let value = Op::apply(lhs, rhs, constants)
                 .ok_or_else(|| vortex_err!(InvalidArgument: "{}", Op::ERROR))?;
-            let value = cast_result_value::<W, O>(value);
+            let value = DecimalValue::from(value)
+                .normalize(result_decimal_dtype)
+                .vortex_expect("bounds-checked result fits the result precision");
             return Ok(ConstantArray::new(
-                Scalar::decimal(
-                    DecimalValue::from(value),
-                    result_decimal_dtype,
-                    result_dtype.nullability(),
-                ),
+                Scalar::decimal(value, result_decimal_dtype, result_dtype.nullability()),
                 len,
             )
             .into_array());
@@ -412,23 +375,45 @@ where
     }
     .map_err(|_lane| vortex_err!(InvalidArgument: "{}", Op::ERROR))?;
 
-    Ok(DecimalArray::new(
+    Ok(decimal_array_narrowed(
         values,
         result_decimal_dtype,
         validity.union_nullability(result_dtype.nullability()),
-    )
-    .into_array())
+    ))
+}
+
+/// Build the result array, narrowing to the dtype's own storage width when the working width is
+/// wider than it. Only division picks a working width above the result precision, and only for a
+/// negative result scale, so this copies in a corner case rather than on the common path.
+fn decimal_array_narrowed<W: NativeDecimalType>(
+    values: Buffer<W>,
+    decimal_dtype: DecimalDType,
+    validity: Validity,
+) -> ArrayRef {
+    let target = DecimalType::smallest_decimal_value_type(&decimal_dtype);
+    if target == W::DECIMAL_TYPE {
+        return DecimalArray::new(values, decimal_dtype, validity).into_array();
+    }
+
+    match_each_decimal_value_type!(target, |O| {
+        let narrowed: Buffer<O> = values
+            .as_slice()
+            .iter()
+            .copied()
+            .map(cast_result_value::<W, O>)
+            .collect();
+        DecimalArray::new(narrowed, decimal_dtype, validity).into_array()
+    })
 }
 
-fn checked_decimal_arrays<W, O, Op>(
+fn checked_decimal_arrays<W, Op>(
     lhs: &DecimalArray,
     rhs: &DecimalArray,
     constants: &DecimalOpConstants<W>,
     valid_rows: &Mask,
-) -> Result<Buffer<O>, usize>
+) -> Result<Buffer<W>, usize>
 where
     W: NativeDecimalType + CheckedAdd + CheckedSub + CheckedMul + CheckedDiv + Mul<Output = W>,
-    O: NativeDecimalType,
     Op: CheckedDecimalOp,
 {
     debug_assert_eq!(lhs.len(), rhs.len());
@@ -445,7 +430,6 @@ where
                         <W as BigCast>::from(rhs)?,
                         constants,
                     )
-                    .map(cast_result_value::<W, O>)
                 },
             )
         })
diff --git a/vortex-array/src/scalar_fn/fns/binary/numeric/tests.rs b/vortex-array/src/scalar_fn/fns/binary/numeric/tests.rs
index d36b436581b..376a79b9814 100644
--- a/vortex-array/src/scalar_fn/fns/binary/numeric/tests.rs
+++ b/vortex-array/src/scalar_fn/fns/binary/numeric/tests.rs
@@ -335,6 +335,25 @@ fn test_decimal_value_outside_working_width_errors() {
     assert!(decimal_binary(lhs, rhs, Operator::Add).is_err());
 }
 
+#[test]
+fn test_decimal_div_negative_result_scale() -> VortexResult<()> {
+    let mut ctx = array_session().create_execution_ctx();
+    // A negative result scale scales the divisor rather than the dividend: 5e12 / 2e8 is 25_000,
+    // truncated to 2 at the decimal(6, -4) result scale. The i64 working width is also wider than
+    // the i32 result storage, so this narrows on the way out.
+    let dtype = DecimalDType::new(10, -8);
+    let lhs = DecimalArray::from_iter::<i64, _>([50_000], dtype).into_array();
+    let rhs = DecimalArray::from_iter::<i64, _>([2], dtype).into_array();
+
+    let result = decimal_binary(lhs, rhs, Operator::Div)?;
+    assert_arrays_eq!(
+        result,
+        DecimalArray::from_iter::<i32, _>([2], DecimalDType::new(6, -4)),
+        &mut ctx
+    );
+    Ok(())
+}
+
 #[test]
 fn test_decimal_mul_value_outside_precision_errors() {
     // `DecimalArray::new` does not validate stored values against the declared precision, so Mul

From 5087468311e0d8bf4f2b12c3b7402fa9d6bad8e4 Mon Sep 17 00:00:00 2001
From: Matt Katz <mhkatz97@gmail.com>
Date: Wed, 5 Aug 2026 16:56:07 -0700
Subject: [PATCH 7/7] inline decimal cast helpers

Signed-off-by: Matt Katz <mhkatz97@gmail.com>
---
 .../scalar_fn/fns/binary/numeric/decimal.rs   | 24 +++++++------------
 1 file changed, 8 insertions(+), 16 deletions(-)

diff --git a/vortex-array/src/scalar_fn/fns/binary/numeric/decimal.rs b/vortex-array/src/scalar_fn/fns/binary/numeric/decimal.rs
index d0e1bea8d7c..fa9ffb0b5f1 100644
--- a/vortex-array/src/scalar_fn/fns/binary/numeric/decimal.rs
+++ b/vortex-array/src/scalar_fn/fns/binary/numeric/decimal.rs
@@ -214,7 +214,7 @@ where
     W: NativeDecimalType + CheckedMul,
 {
     fn new(result: DecimalDType, op: NumericOperator) -> VortexResult<Self> {
-        let one = cast_work_value::<W, i8>(1);
+        let one = <W as BigCast>::from(1_i8).vortex_expect("one fits every decimal working width");
         let (lhs_scale_factor, rhs_scale_factor) = if op == NumericOperator::Div {
             // Arrow scales the quotient by 10^(result_scale - lhs_scale + rhs_scale). Both
             // Vortex operands share a dtype, so this simplifies to 10^result_scale. A negative
@@ -241,8 +241,9 @@ fn decimal_scale_factor<W>(exp: u32) -> VortexResult<W>
 where
     W: NativeDecimalType + CheckedMul,
 {
-    let ten = cast_work_value::<W, i8>(10);
-    let mut factor = cast_work_value::<W, i8>(1);
+    let ten = <W as BigCast>::from(10_i8).vortex_expect("ten fits every decimal working width");
+    let mut factor =
+        <W as BigCast>::from(1_i8).vortex_expect("one fits every decimal working width");
     for _ in 0..exp {
         factor = factor.checked_mul(&ten).ok_or_else(|| {
             vortex_err!(
@@ -400,7 +401,10 @@ fn decimal_array_narrowed<W: NativeDecimalType>(
             .as_slice()
             .iter()
             .copied()
-            .map(cast_result_value::<W, O>)
+            .map(|value| {
+                <O as BigCast>::from(value)
+                    .vortex_expect("precision-checked decimal result must fit the output width")
+            })
             .collect();
         DecimalArray::new(narrowed, decimal_dtype, validity).into_array()
     })
@@ -436,18 +440,6 @@ where
     })
 }
 
-#[inline(always)]
-fn cast_work_value<W: NativeDecimalType, T: NativeDecimalType>(value: T) -> W {
-    <W as BigCast>::from(value)
-        .vortex_expect("valid decimal input must fit the arithmetic working width")
-}
-
-#[inline(always)]
-fn cast_result_value<W: NativeDecimalType, O: NativeDecimalType>(value: W) -> O {
-    <O as BigCast>::from(value)
-        .vortex_expect("precision-checked decimal result must fit the output width")
-}
-
 fn typed_constant<W: NativeDecimalType>(value: &DecimalValue) -> W {
     value
         .cast::<W>()