vortex_array/arrays/decimal/compute/

cast.rs

// SPDX-License-Identifier: Apache-2.0
// SPDX-FileCopyrightText: Copyright the Vortex contributors

use num_traits::AsPrimitive;
use num_traits::CheckedMul;
use num_traits::ToPrimitive as NumToPrimitive;
use vortex_buffer::Buffer;
use vortex_buffer::BufferMut;
use vortex_compute::lane_kernels::IndexedSourceExt;
use vortex_error::VortexError;
use vortex_error::VortexExpect;
use vortex_error::VortexResult;
use vortex_error::vortex_bail;
use vortex_error::vortex_err;
use vortex_error::vortex_panic;
use vortex_mask::Mask;

use crate::ArrayRef;
use crate::ExecutionCtx;
use crate::IntoArray;
use crate::array::ArrayView;
use crate::arrays::ConstantArray;
use crate::arrays::Decimal;
use crate::arrays::DecimalArray;
use crate::arrays::PrimitiveArray;
use crate::arrays::decimal::DecimalArrayExt;
use crate::dtype::BigCast;
use crate::dtype::DType;
use crate::dtype::DecimalDType;
use crate::dtype::DecimalType;
use crate::dtype::IntegerPType;
use crate::dtype::NativeDecimalType;
use crate::dtype::Nullability;
use crate::dtype::PType;
use crate::dtype::ToI256;
use crate::dtype::i256;
use crate::match_each_decimal_value_type;
use crate::match_each_integer_ptype;
use crate::scalar::DecimalToIntegerCast;
use crate::scalar::DecimalValue;
use crate::scalar::Scalar;
use crate::scalar_fn::fns::cast::CastKernel;
use crate::scalar_fn::fns::cast::CastReduce;
use crate::validity::Validity;

impl CastReduce for Decimal {
    fn cast(array: ArrayView<'_, Decimal>, dtype: &DType) -> VortexResult<Option<ArrayRef>> {
        // Only nullability changes within the same decimal dtype are reducible without execution.
        // Precision/scale changes need the kernel.
        let DType::Decimal(to_decimal_dtype, to_nullability) = dtype else {
            return Ok(None);
        };
        let DType::Decimal(from_decimal_dtype, _) = array.dtype() else {
            vortex_panic!(
                "DecimalArray must have decimal dtype, got {:?}",
                array.dtype()
            );
        };

        if from_decimal_dtype != to_decimal_dtype {
            return Ok(None);
        }

        let Some(new_validity) = array
            .validity()?
            .trivially_cast_nullability(*to_nullability, array.len())?
        else {
            return Ok(None);
        };

        // SAFETY: validity has the same length, only its nullability tag changes.
        unsafe {
            Ok(Some(
                DecimalArray::new_unchecked_handle(
                    array.buffer_handle().clone(),
                    array.values_type(),
                    *to_decimal_dtype,
                    new_validity,
                )
                .into_array(),
            ))
        }
    }
}

impl CastKernel for Decimal {
    fn cast(
        array: ArrayView<'_, Decimal>,
        dtype: &DType,
        ctx: &mut ExecutionCtx,
    ) -> VortexResult<Option<ArrayRef>> {
        let DType::Decimal(from_decimal_dtype, _) = array.dtype() else {
            vortex_panic!(
                "DecimalArray must have decimal dtype, got {:?}",
                array.dtype()
            );
        };
        if let DType::Primitive(ptype, nullability) = dtype
            && ptype.is_int()
        {
            return match_each_integer_ptype!(*ptype, |T| {
                cast_to_integer::<T>(array, *nullability, ctx).map(Some)
            });
        }
        if let DType::Primitive(PType::F64, nullability) = dtype {
            let scale = from_decimal_dtype.scale();
            return cast_to_f64(array, scale, *nullability, ctx).map(Some);
        }
        let DType::Decimal(to_decimal_dtype, to_nullability) = dtype else {
            return Ok(None);
        };

        // If the dtype is exactly the same, return self
        if array.dtype() == dtype {
            return Ok(Some(array.array().clone()));
        }

        let validity = array.validity()?;

        // Cast the validity to the new nullability
        let new_validity = validity
            .clone()
            .cast_nullability(*to_nullability, array.len(), ctx)?;

        // Reuse the values buffer untouched when no rescale is required, the target precision
        // only widens (so every value still fits), and the current physical type is already wide
        // enough to hold the target precision. This keeps the common precision-widening cast
        // (and pure nullability changes) zero-copy instead of allocating and re-scanning.
        if from_decimal_dtype.scale() == to_decimal_dtype.scale()
            && to_decimal_dtype.precision() >= from_decimal_dtype.precision()
            && array
                .values_type()
                .is_compatible_decimal_value_type(*to_decimal_dtype)
        {
            // SAFETY: the source values are bit-identical and remain in range for the wider
            // precision, and new_validity has the same length, only its nullability tag changes.
            unsafe {
                return Ok(Some(
                    DecimalArray::new_unchecked_handle(
                        array.buffer_handle().clone(),
                        array.values_type(),
                        *to_decimal_dtype,
                        new_validity,
                    )
                    .into_array(),
                ));
            }
        }

        let valid_values = validity.execute_mask(array.len(), ctx)?;
        if !array.is_empty() && matches!(valid_values, Mask::AllFalse(_)) {
            return Ok(Some(
                ConstantArray::new(Scalar::null(dtype.clone()), array.len()).into_array(),
            ));
        }
        let target_values_type = DecimalType::smallest_decimal_value_type(to_decimal_dtype);

        match_each_decimal_value_type!(array.values_type(), |F| {
            match_each_decimal_value_type!(target_values_type, |T| {
                cast_decimal_values::<F, T>(
                    array,
                    *from_decimal_dtype,
                    *to_decimal_dtype,
                    new_validity,
                    &valid_values,
                )
                .map(Some)
            })
        })
    }
}

fn cast_to_integer<T: IntegerPType + BigCast>(
    array: ArrayView<'_, Decimal>,
    nullability: Nullability,
    ctx: &mut ExecutionCtx,
) -> VortexResult<ArrayRef> {
    let source_validity = array.validity()?;
    let mask = source_validity.execute_mask(array.len(), ctx)?;
    let validity = source_validity.cast_nullability(nullability, array.len(), ctx)?;
    if !array.is_empty() && matches!(mask, Mask::AllFalse(_)) {
        return Ok(ConstantArray::new(
            Scalar::null(DType::Primitive(T::PTYPE, nullability)),
            array.len(),
        )
        .into_array());
    }
    let cast = DecimalToIntegerCast::<T>::new(array.decimal_dtype().scale());
    let buffer = match_each_decimal_value_type!(array.values_type(), |F| {
        let values = array.buffer::<F>();
        cast_decimal_buffer(values.as_slice(), &mask, |value: F| {
            cast.cast(value.to_i256()?)
        })
        .map_err(|index| cast.error(DecimalValue::from(values[index]).as_i256()))?
    });
    Ok(PrimitiveArray::new(buffer, validity).into_array())
}

fn cast_to_f64(
    array: ArrayView<'_, Decimal>,
    scale: i8,
    nullability: Nullability,
    ctx: &mut ExecutionCtx,
) -> VortexResult<ArrayRef> {
    let source_validity = array.validity()?;
    let n = array.len();
    let mask = source_validity.execute_mask(n, ctx)?;
    let validity = source_validity.cast_nullability(nullability, n, ctx)?;
    if n > 0 && matches!(mask, Mask::AllFalse(_)) {
        return Ok(
            ConstantArray::new(Scalar::null(DType::Primitive(PType::F64, nullability)), n)
                .into_array(),
        );
    }

    let scale: i32 = scale.as_();
    let inv_factor: f64 = 10f64.powi(-scale);

    let buffer = match &mask {
        Mask::AllFalse(_) => BufferMut::<f64>::zeroed(n),
        Mask::AllTrue(_) => match_each_decimal_value_type!(array.values_type(), |F| {
            let values = array.buffer::<F>();
            let values = values.as_slice();
            let mut out = BufferMut::<f64>::with_capacity(n);
            values.map_into(&mut out.spare_capacity_mut()[..n], |v: F| {
                to_f64_lossy::<F>(v) * inv_factor
            });
            // SAFETY: map_into wrote every lane before returning.
            unsafe { out.set_len(n) };
            out
        }),
        Mask::Values(mask_values) => match_each_decimal_value_type!(array.values_type(), |F| {
            let values = array.buffer::<F>();
            let values = values.as_slice();
            let mut out = BufferMut::<f64>::with_capacity(n);
            let write_result = values.try_map_masked_into(
                mask_values.bit_buffer(),
                &mut out.spare_capacity_mut()[..n],
                |v: F| Some(to_f64_lossy::<F>(v) * inv_factor),
            );
            debug_assert!(write_result.is_ok());
            // SAFETY: try_map_masked_into wrote every lane before returning Ok.
            unsafe { out.set_len(n) };
            out
        }),
    };

    Ok(PrimitiveArray::new(buffer, validity).into_array())
}

#[inline]
fn to_f64_lossy<F: NativeDecimalType>(value: F) -> f64 {
    NumToPrimitive::to_f64(&value).unwrap_or(f64::NAN)
}

fn cast_decimal_values<F, T>(
    array: ArrayView<'_, Decimal>,
    from_decimal_dtype: DecimalDType,
    to_decimal_dtype: DecimalDType,
    validity: Validity,
    valid_values: &Mask,
) -> VortexResult<ArrayRef>
where
    F: NativeDecimalType,
    T: NativeDecimalType + CheckedMul,
    DecimalValue: From<F>,
{
    let values = array.buffer::<F>();
    let values = values.as_slice();
    let cast_result = match DecimalCastPlan::<T>::new(from_decimal_dtype, to_decimal_dtype) {
        DecimalCastPlan::SameScale { min, max } => {
            cast_decimal_buffer(values, valid_values, |value| {
                let value = <T as BigCast>::from(value)?;
                (value >= min && value <= max).then_some(value)
            })
        }
        cast_plan => cast_decimal_buffer(values, valid_values, |value| cast_plan.cast(value)),
    };
    let buffer = cast_result.map_err(|idx| {
        decimal_cast_error::<F, T>(values[idx], from_decimal_dtype, to_decimal_dtype)
    })?;

    Ok(DecimalArray::new(buffer, to_decimal_dtype, validity).into_array())
}

fn cast_decimal_buffer<F, T>(
    values: &[F],
    valid_values: &Mask,
    cast: impl Fn(F) -> Option<T>,
) -> Result<Buffer<T>, usize>
where
    F: NativeDecimalType,
    T: Copy + Default,
{
    let mut buffer = BufferMut::<T>::with_capacity(values.len());
    match valid_values {
        Mask::AllTrue(_) => {
            values.try_map_into(&mut buffer.spare_capacity_mut()[..values.len()], &cast)?;
        }
        Mask::AllFalse(_) => return Ok(BufferMut::<T>::zeroed(values.len()).freeze()),
        Mask::Values(mask) => {
            values.try_map_masked_into(
                mask.bit_buffer(),
                &mut buffer.spare_capacity_mut()[..values.len()],
                &cast,
            )?;
        }
    }
    // SAFETY: the selected map kernel initialized every lane before returning Ok.
    unsafe { buffer.set_len(values.len()) };
    Ok(buffer.freeze())
}

#[cold]
fn decimal_cast_error<F, T>(
    value: F,
    from_decimal_dtype: DecimalDType,
    to_decimal_dtype: DecimalDType,
) -> VortexError
where
    F: NativeDecimalType,
    T: NativeDecimalType,
    DecimalValue: From<F>,
{
    match DecimalValue::from(value)
        .cast_decimal(from_decimal_dtype, to_decimal_dtype)
        .and_then(|value| {
            value.cast::<T>().ok_or_else(|| {
                vortex_err!(
                    "decimal value cannot be represented as {} after casting to {}",
                    T::DECIMAL_TYPE,
                    to_decimal_dtype
                )
            })
        }) {
        Ok(_) => {
            // The fast path only returns `None` for values the slow path also rejects, so this
            // arm should be unreachable. If it is hit, the fast and slow paths have drifted and
            // we are erroring on a value that is actually representable.
            debug_assert!(
                false,
                "decimal fast-path cast rejected value {value} that the slow path accepts \
                 (from {from_decimal_dtype} to {to_decimal_dtype})"
            );
            vortex_err!(
                "decimal value cannot be represented as {} after casting from {} to {}",
                T::DECIMAL_TYPE,
                from_decimal_dtype,
                to_decimal_dtype
            )
        }
        Err(error) => error,
    }
}

#[derive(Debug, Clone, Copy)]
enum DecimalCastPlan<T> {
    SameScale { min: T, max: T },
    ScaleUp { factor: T, min: T, max: T },
    ScaleUpOverflow,
    ScaleDown { factor: i256, min: i256, max: i256 },
    ScaleDownOverflow,
}

impl<T> DecimalCastPlan<T>
where
    T: NativeDecimalType + CheckedMul,
{
    fn new(from_decimal_dtype: DecimalDType, to_decimal_dtype: DecimalDType) -> Self {
        let scale_delta = to_decimal_dtype.scale() as i16 - from_decimal_dtype.scale() as i16;
        if scale_delta == 0 {
            let (min, max) = decimal_precision_range::<T>(to_decimal_dtype);
            return Self::SameScale { min, max };
        }

        if scale_delta > 0 {
            let Some(factor) = decimal_scale_factor::<T>(scale_delta as u32) else {
                return Self::ScaleUpOverflow;
            };
            let (min, max) = decimal_precision_range::<T>(to_decimal_dtype);
            return Self::ScaleUp { factor, min, max };
        }

        let Some(factor) = decimal_scale_factor::<i256>((-scale_delta) as u32) else {
            return Self::ScaleDownOverflow;
        };
        let (min, max) = decimal_precision_range::<i256>(to_decimal_dtype);
        Self::ScaleDown { factor, min, max }
    }

    #[inline]
    fn cast<F>(&self, value: F) -> Option<T>
    where
        F: NativeDecimalType,
    {
        match *self {
            DecimalCastPlan::SameScale { min, max } => {
                let value = <T as BigCast>::from(value)?;
                (value >= min && value <= max).then_some(value)
            }
            DecimalCastPlan::ScaleUp { factor, min, max } => {
                let value = <T as BigCast>::from(value)?;
                let value = value.checked_mul(&factor)?;
                (value >= min && value <= max).then_some(value)
            }
            DecimalCastPlan::ScaleUpOverflow | DecimalCastPlan::ScaleDownOverflow => {
                (value == F::default()).then_some(T::default())
            }
            DecimalCastPlan::ScaleDown { factor, min, max } => {
                let value = <i256 as BigCast>::from(value)?;
                if value == i256::ZERO {
                    return Some(T::default());
                }
                if value % factor != i256::ZERO {
                    return None;
                }

                let value = value / factor;
                if value < min || value > max {
                    return None;
                }
                <T as BigCast>::from(value)
            }
        }
    }
}

fn decimal_precision_range<T: NativeDecimalType>(decimal_dtype: DecimalDType) -> (T, T) {
    let precision = usize::from(decimal_dtype.precision());
    (
        T::MIN_BY_PRECISION[precision],
        T::MAX_BY_PRECISION[precision],
    )
}

fn decimal_scale_factor<T>(exp: u32) -> Option<T>
where
    T: NativeDecimalType + CheckedMul,
{
    let ten = <T as BigCast>::from(10_i8)?;
    let mut factor = <T as BigCast>::from(1_i8)?;
    for _ in 0..exp {
        factor = factor.checked_mul(&ten)?;
    }
    Some(factor)
}

/// Upcast a DecimalArray to a wider physical representation (e.g., i32 -> i64) while keeping
/// the same precision and scale.
///
/// This is useful when you need to widen the underlying storage type to accommodate operations
/// that might overflow the current representation, or to match the physical type expected by
/// downstream consumers.
///
/// # Errors
///
/// Returns an error if `to_values_type` is narrower than the array's current values type.
/// Only upcasting (widening) is supported.
pub fn upcast_decimal_values(
    array: ArrayView<'_, Decimal>,
    to_values_type: DecimalType,
) -> VortexResult<DecimalArray> {
    let from_values_type = array.values_type();

    // If already the target type, just clone
    if from_values_type == to_values_type {
        return Ok(array.array().as_::<Decimal>().into_owned());
    }

    // Only allow upcasting (widening)
    if to_values_type < from_values_type {
        vortex_bail!(
            "Cannot downcast decimal values from {:?} to {:?}. Only upcasting is supported.",
            from_values_type,
            to_values_type
        );
    }

    let decimal_dtype = array.decimal_dtype();
    let validity = array.validity()?;

    // Use match_each_decimal_value_type to dispatch based on source and target types
    match_each_decimal_value_type!(from_values_type, |F| {
        let from_buffer = array.buffer::<F>();
        match_each_decimal_value_type!(to_values_type, |T| {
            let to_buffer = upcast_decimal_buffer::<F, T>(from_buffer);
            Ok(DecimalArray::new(to_buffer, decimal_dtype, validity))
        })
    })
}

/// Upcast a buffer of decimal values from type F to type T.
/// Since T is wider than F, this conversion never fails.
fn upcast_decimal_buffer<F: NativeDecimalType, T: NativeDecimalType>(from: Buffer<F>) -> Buffer<T> {
    from.iter()
        .map(|&v| T::from(v).vortex_expect("upcast should never fail"))
        .collect()
}

#[cfg(test)]
mod tests {
    use rstest::rstest;
    use vortex_buffer::buffer;
    use vortex_error::VortexResult;

    use super::upcast_decimal_values;
    use crate::Canonical;
    use crate::IntoArray;
    use crate::VortexSessionExecute;
    use crate::array_session;
    use crate::arrays::DecimalArray;
    use crate::arrays::PrimitiveArray;
    use crate::assert_arrays_eq;
    use crate::builtins::ArrayBuiltins;
    use crate::compute::conformance::cast::test_cast_conformance;
    use crate::dtype::BigCast;
    use crate::dtype::DType;
    use crate::dtype::DecimalDType;
    use crate::dtype::DecimalType;
    use crate::dtype::Nullability;
    use crate::dtype::PType;
    use crate::dtype::i256;
    use crate::match_each_decimal_value_type;
    use crate::match_each_integer_ptype;
    use crate::scalar::Scalar;
    use crate::validity::Validity;

    #[rstest]
    #[case::workflow(0, vec![0, 1], vec![0, 1])]
    #[case::fractional(1, vec![-199, -1, 0, 1, 199], vec![-19, 0, 0, 0, 19])]
    #[case::negative_scale(-2, vec![-12, 0, 12], vec![-1200, 0, 1200])]
    #[case::large_exact(
        0,
        vec![1_786_639_777_684_000_001],
        vec![1_786_639_777_684_000_001]
    )]
    #[case::large_scale(76, vec![-1, 1], vec![0, 0])]
    #[case::zero_extreme_scale(-128, vec![0, 0], vec![0, 0])]
    fn cast_decimal_to_integer_policy(
        #[case] scale: i8,
        #[case] values: Vec<i128>,
        #[case] expected: Vec<i64>,
    ) -> VortexResult<()> {
        let mut ctx = array_session().create_execution_ctx();
        let decimal_dtype = DecimalDType::new(76, scale);
        let array = DecimalArray::new(
            values.iter().copied().map(i256::from_i128).collect(),
            decimal_dtype,
            Validity::NonNullable,
        );
        let target = DType::Primitive(PType::I64, Nullability::NonNullable);
        let casted = array
            .into_array()
            .cast(target.clone())?
            .execute::<PrimitiveArray>(&mut ctx)?;
        assert_arrays_eq!(
            casted,
            PrimitiveArray::from_iter(expected.iter().copied()),
            &mut ctx
        );
        for (value, expected) in values.into_iter().zip(expected) {
            let scalar = Scalar::decimal(value.into(), decimal_dtype, Nullability::NonNullable);
            assert_eq!(scalar.cast(&target)?, Scalar::from(expected));
        }
        Ok(())
    }

    #[rstest]
    fn cast_decimal_to_integer_storage_and_nulls(
        #[values(
            DecimalType::I8,
            DecimalType::I16,
            DecimalType::I32,
            DecimalType::I64,
            DecimalType::I128,
            DecimalType::I256
        )]
        storage: DecimalType,
        #[values(
            PType::I8, PType::I16, PType::I32, PType::I64, PType::U8, PType::U16, PType::U32,
            PType::U64
        )]
        target: PType,
    ) -> VortexResult<()> {
        let mut ctx = array_session().create_execution_ctx();
        let decimal_dtype = DecimalDType::new(3, 1);
        let array = match_each_decimal_value_type!(storage, |F| {
            DecimalArray::from_option_iter(
                [Some(19i8), None, Some(123)].map(|value| value.and_then(<F as BigCast>::from)),
                decimal_dtype,
            )
        });
        let dtype = DType::Primitive(target, Nullability::Nullable);
        let casted = array
            .into_array()
            .cast(dtype.clone())?
            .execute::<PrimitiveArray>(&mut ctx)?;
        match_each_integer_ptype!(target, |T| {
            assert_arrays_eq!(
                casted,
                PrimitiveArray::from_option_iter([Some(1 as T), None, Some(12 as T)]),
                &mut ctx
            );
            for (value, expected) in [(19i8, 1 as T), (123, 12 as T)] {
                let scalar = Scalar::decimal(value.into(), decimal_dtype, Nullability::Nullable);
                assert_eq!(
                    scalar.cast(&dtype)?,
                    Scalar::primitive(expected, Nullability::Nullable)
                );
            }
        });
        Ok(())
    }

    #[rstest]
    fn cast_decimal_to_integer_bounds(
        #[values(
            PType::I8, PType::I16, PType::I32, PType::I64, PType::U8, PType::U16, PType::U32,
            PType::U64
        )]
        target: PType,
    ) -> VortexResult<()> {
        let mut ctx = array_session().create_execution_ctx();
        let decimal_dtype = DecimalDType::new(76, 1);
        let dtype = DType::Primitive(target, Nullability::NonNullable);
        match_each_integer_ptype!(target, |T| {
            let ten = i256::from_i128(10);
            let minimum = i256::from_i128(T::MIN as i128) * ten;
            let maximum = i256::from_i128(T::MAX as i128) * ten;
            let array = DecimalArray::new(
                buffer![minimum, maximum],
                decimal_dtype,
                Validity::NonNullable,
            );
            let casted = array
                .into_array()
                .cast(dtype.clone())?
                .execute::<PrimitiveArray>(&mut ctx)?;
            assert_arrays_eq!(
                casted,
                PrimitiveArray::from_iter([T::MIN, T::MAX]),
                &mut ctx
            );
            for (value, expected) in [(minimum, T::MIN), (maximum, T::MAX)] {
                let scalar = Scalar::decimal(value.into(), decimal_dtype, Nullability::NonNullable);
                assert_eq!(
                    scalar.cast(&dtype)?,
                    Scalar::primitive(expected, Nullability::NonNullable)
                );
            }
            // The scalar policy rejects values outside the range even when truncation would fit.
            for value in [
                minimum - i256::ONE,
                maximum + i256::ONE,
                minimum - ten,
                maximum + ten,
            ] {
                let array = DecimalArray::new(
                    buffer![i256::ZERO, value],
                    decimal_dtype,
                    Validity::NonNullable,
                );
                let error = array
                    .into_array()
                    .cast(dtype.clone())?
                    .execute::<PrimitiveArray>(&mut ctx)
                    .unwrap_err();
                assert!(error.to_string().contains("out of range"));
                let scalar = Scalar::decimal(value.into(), decimal_dtype, Nullability::NonNullable);
                assert!(scalar.cast(&dtype).is_err());
            }
        });
        Ok(())
    }

    #[test]
    fn cast_decimal_to_integer_masks_and_empty() -> VortexResult<()> {
        let mut ctx = array_session().create_execution_ctx();
        let decimal_dtype = DecimalDType::new(21, 0);
        let target = DType::Primitive(PType::I8, Nullability::Nullable);
        for validity in [Validity::from_iter([false, true]), Validity::AllInvalid] {
            let expected = PrimitiveArray::new(buffer![0i8, 12], validity.clone());
            let array = DecimalArray::new(buffer![999i128, 12], decimal_dtype, validity);
            let casted = array
                .into_array()
                .cast(target.clone())?
                .execute::<PrimitiveArray>(&mut ctx)?;
            assert_arrays_eq!(casted, expected, &mut ctx);
        }
        let array = DecimalArray::new(
            buffer![999i128, 12],
            decimal_dtype,
            Validity::from_iter([false, true]),
        );
        assert!(
            array
                .into_array()
                .cast(target.as_nonnullable())?
                .execute::<PrimitiveArray>(&mut ctx)
                .is_err()
        );
        let empty = DecimalArray::from_option_iter([] as [Option<i128>; 0], decimal_dtype);
        let casted = empty
            .into_array()
            .cast(target.as_nonnullable())?
            .execute::<PrimitiveArray>(&mut ctx)?;
        assert_arrays_eq!(casted, PrimitiveArray::from_iter([] as [i8; 0]), &mut ctx);
        Ok(())
    }

    #[test]
    fn cast_decimal_to_integer_wide_storage() -> VortexResult<()> {
        let mut ctx = array_session().create_execution_ctx();
        let factor = i256::from_i128(10).checked_pow(40).unwrap();
        let value = i256::from_i128(123) * factor + i256::ONE;
        let decimal_dtype = DecimalDType::new(76, 40);
        let target = DType::Primitive(PType::I64, Nullability::NonNullable);
        let array = DecimalArray::new(buffer![value, -value], decimal_dtype, Validity::NonNullable);
        let casted = array
            .into_array()
            .cast(target.clone())?
            .execute::<PrimitiveArray>(&mut ctx)?;
        assert_arrays_eq!(casted, PrimitiveArray::from_iter([123i64, -123]), &mut ctx);
        for (value, expected) in [(value, 123i64), (-value, -123)] {
            let scalar = Scalar::decimal(value.into(), decimal_dtype, Nullability::NonNullable);
            assert_eq!(scalar.cast(&target)?, Scalar::from(expected));
        }
        let huge = DecimalArray::new(
            buffer![i256::ONE],
            DecimalDType::new(1, -128),
            Validity::NonNullable,
        );
        assert!(
            huge.into_array()
                .cast(target)?
                .execute::<PrimitiveArray>(&mut ctx)
                .is_err()
        );
        Ok(())
    }

    #[test]
    fn cast_decimal_to_nullable() {
        let mut ctx = array_session().create_execution_ctx();
        let decimal_dtype = DecimalDType::new(10, 2);
        let array = DecimalArray::new(
            buffer![100i32, 200, 300],
            decimal_dtype,
            Validity::NonNullable,
        );

        // Cast to nullable
        let nullable_dtype = DType::Decimal(decimal_dtype, Nullability::Nullable);
        let casted = array
            .into_array()
            .cast(nullable_dtype.clone())
            .unwrap()
            .execute::<DecimalArray>(&mut ctx)
            .unwrap();

        assert_eq!(casted.dtype(), &nullable_dtype);
        assert!(matches!(casted.validity(), Ok(Validity::AllValid)));
        assert_eq!(casted.len(), 3);
    }

    #[test]
    fn cast_nullable_to_non_nullable() {
        let mut ctx = array_session().create_execution_ctx();
        let decimal_dtype = DecimalDType::new(10, 2);

        // Create nullable array with no nulls
        let array = DecimalArray::new(buffer![100i32, 200, 300], decimal_dtype, Validity::AllValid);

        // Cast to non-nullable
        let non_nullable_dtype = DType::Decimal(decimal_dtype, Nullability::NonNullable);
        let casted = array
            .into_array()
            .cast(non_nullable_dtype.clone())
            .unwrap()
            .execute::<DecimalArray>(&mut ctx)
            .unwrap();

        assert_eq!(casted.dtype(), &non_nullable_dtype);
        assert!(matches!(casted.validity(), Ok(Validity::NonNullable)));
    }

    #[test]
    #[should_panic(expected = "Cannot cast array with invalid values to non-nullable type")]
    fn cast_nullable_with_nulls_to_non_nullable_fails() {
        let mut ctx = array_session().create_execution_ctx();
        let decimal_dtype = DecimalDType::new(10, 2);

        // Create nullable array with nulls
        let array = DecimalArray::from_option_iter([Some(100i32), None, Some(300)], decimal_dtype);

        // Attempt to cast to non-nullable should fail
        let non_nullable_dtype = DType::Decimal(decimal_dtype, Nullability::NonNullable);
        array
            .into_array()
            .cast(non_nullable_dtype)
            .and_then(|a| a.execute::<Canonical>(&mut ctx).map(|c| c.into_array()))
            .unwrap();
    }

    #[test]
    fn cast_different_scale_rescales() {
        let mut ctx = array_session().create_execution_ctx();
        let array = DecimalArray::new(
            buffer![100i32],
            DecimalDType::new(10, 2),
            Validity::NonNullable,
        );

        // Cast 1.00 to scale 3, where it is stored as 1000.
        let different_dtype = DType::Decimal(DecimalDType::new(15, 3), Nullability::NonNullable);
        let casted = array
            .into_array()
            .cast(different_dtype)
            .unwrap()
            .execute::<DecimalArray>(&mut ctx)
            .unwrap();

        assert_eq!(casted.precision(), 15);
        assert_eq!(casted.scale(), 3);
        assert_eq!(casted.values_type(), DecimalType::I64);
        assert_eq!(casted.buffer::<i64>().as_ref(), &[1000]);
    }

    #[test]
    fn cast_downcast_precision_succeeds_when_values_fit() {
        let mut ctx = array_session().create_execution_ctx();
        let array = DecimalArray::new(
            buffer![100i64],
            DecimalDType::new(18, 2),
            Validity::NonNullable,
        );

        // Downcasting precision is allowed when every value fits.
        let smaller_dtype = DType::Decimal(DecimalDType::new(10, 2), Nullability::NonNullable);
        let casted = array
            .into_array()
            .cast(smaller_dtype)
            .unwrap()
            .execute::<DecimalArray>(&mut ctx)
            .unwrap();

        assert_eq!(casted.precision(), 10);
        assert_eq!(casted.scale(), 2);
        assert_eq!(casted.buffer::<i64>().as_ref(), &[100]);
    }

    #[test]
    fn cast_downcast_precision_checks_values() {
        let mut ctx = array_session().create_execution_ctx();
        let array = DecimalArray::new(
            buffer![1000i64],
            DecimalDType::new(18, 0),
            Validity::NonNullable,
        );

        let smaller_dtype = DType::Decimal(DecimalDType::new(3, 0), Nullability::NonNullable);
        let result = array
            .into_array()
            .cast(smaller_dtype)
            .and_then(|a| a.execute::<Canonical>(&mut ctx).map(|c| c.into_array()));

        assert!(result.is_err());
        assert!(
            result
                .unwrap_err()
                .to_string()
                .contains("does not fit in precision")
        );
    }

    #[test]
    fn cast_lower_scale_requires_exact_rescale() {
        let mut ctx = array_session().create_execution_ctx();
        let array = DecimalArray::new(
            buffer![123456i64],
            DecimalDType::new(10, 4),
            Validity::NonNullable,
        );

        let lower_scale_dtype = DType::Decimal(DecimalDType::new(10, 2), Nullability::NonNullable);
        let result = array
            .into_array()
            .cast(lower_scale_dtype)
            .and_then(|a| a.execute::<Canonical>(&mut ctx).map(|c| c.into_array()));

        assert!(result.is_err());
        assert!(
            result
                .unwrap_err()
                .to_string()
                .contains("would lose precision")
        );
    }

    #[test]
    fn cast_lower_scale_ignores_null_lane_failures() {
        let mut ctx = array_session().create_execution_ctx();
        let array = DecimalArray::new(
            buffer![100i64, 123456],
            DecimalDType::new(10, 4),
            Validity::from_iter([true, false]),
        );

        let lower_scale_dtype = DType::Decimal(DecimalDType::new(3, 2), Nullability::Nullable);
        let casted = array
            .into_array()
            .cast(lower_scale_dtype)
            .unwrap()
            .execute::<DecimalArray>(&mut ctx)
            .unwrap();

        let mask = casted
            .as_ref()
            .validity()
            .unwrap()
            .execute_mask(
                casted.as_ref().len(),
                &mut array_session().create_execution_ctx(),
            )
            .unwrap();
        assert!(mask.value(0));
        assert!(!mask.value(1));
        assert_eq!(casted.buffer::<i16>().as_ref()[0], 1);
    }

    #[test]
    fn cast_upcast_precision_succeeds() {
        let mut ctx = array_session().create_execution_ctx();
        let array = DecimalArray::new(
            buffer![100i32, 200, 300],
            DecimalDType::new(10, 2),
            Validity::NonNullable,
        );

        // Cast to higher precision with same scale - should succeed
        let wider_dtype = DType::Decimal(DecimalDType::new(38, 2), Nullability::NonNullable);
        let casted = array
            .into_array()
            .cast(wider_dtype)
            .unwrap()
            .execute::<DecimalArray>(&mut ctx)
            .unwrap();

        assert_eq!(casted.precision(), 38);
        assert_eq!(casted.scale(), 2);
        assert_eq!(casted.len(), 3);
        // Should be stored in i128 now (precision 38 requires i128)
        assert_eq!(casted.values_type(), DecimalType::I128);
    }

    #[test]
    fn cast_widening_same_physical_type_is_zero_copy() {
        let mut ctx = array_session().create_execution_ctx();
        // Decimal(10,2) and Decimal(18,2) are both physically i64 with the same scale, so widening
        // the precision must reuse the values buffer rather than allocate and re-scan it.
        let array = DecimalArray::new(
            buffer![100i64, 200, 300],
            DecimalDType::new(10, 2),
            Validity::NonNullable,
        );
        let src_ptr = array.buffer::<i64>().as_ptr();

        let wider_dtype = DType::Decimal(DecimalDType::new(18, 2), Nullability::NonNullable);
        let casted = array
            .into_array()
            .cast(wider_dtype)
            .unwrap()
            .execute::<DecimalArray>(&mut ctx)
            .unwrap();

        assert_eq!(casted.precision(), 18);
        assert_eq!(casted.scale(), 2);
        assert_eq!(casted.values_type(), DecimalType::I64);
        assert_eq!(casted.buffer::<i64>().as_ref(), &[100, 200, 300]);
        // The values buffer must be shared with the source (zero-copy), not reallocated.
        assert_eq!(
            casted.buffer::<i64>().as_ptr(),
            src_ptr,
            "precision-widening cast must reuse the source values buffer"
        );
    }

    #[test]
    fn cast_to_non_decimal_returns_err() {
        let mut ctx = array_session().create_execution_ctx();
        let array = DecimalArray::new(
            buffer![100i32],
            DecimalDType::new(10, 2),
            Validity::NonNullable,
        );

        // Try to cast to non-decimal type - should fail since no kernel can handle it
        let result = array
            .into_array()
            .cast(DType::Utf8(Nullability::NonNullable))
            .and_then(|a| a.execute::<Canonical>(&mut ctx).map(|c| c.into_array()));

        assert!(result.is_err());
        assert!(
            result
                .unwrap_err()
                .to_string()
                .contains("No CastKernel to cast canonical array")
        );
    }

    #[rstest]
    #[case(DecimalArray::new(buffer![100i32, 200, 300], DecimalDType::new(10, 2), Validity::NonNullable))]
    #[case(DecimalArray::new(buffer![10000i64, 20000, 30000], DecimalDType::new(18, 4), Validity::NonNullable))]
    #[case(DecimalArray::from_option_iter([Some(100i32), None, Some(300)], DecimalDType::new(10, 2)))]
    #[case(DecimalArray::new(buffer![42i32], DecimalDType::new(5, 1), Validity::NonNullable))]
    fn test_cast_decimal_conformance(#[case] array: DecimalArray) {
        test_cast_conformance(
            &array.into_array(),
            &mut array_session().create_execution_ctx(),
        );
    }

    #[test]
    fn upcast_decimal_values_i32_to_i64() {
        let decimal_dtype = DecimalDType::new(10, 2);
        let array = DecimalArray::new(
            buffer![100i32, 200, 300],
            decimal_dtype,
            Validity::NonNullable,
        );

        assert_eq!(array.values_type(), DecimalType::I32);

        let array = array.as_view();
        let casted = upcast_decimal_values(array, DecimalType::I64).unwrap();

        assert_eq!(casted.values_type(), DecimalType::I64);
        assert_eq!(casted.decimal_dtype(), decimal_dtype);
        assert_eq!(casted.len(), 3);

        // Verify values are preserved
        let buffer = casted.buffer::<i64>();
        assert_eq!(buffer.as_ref(), &[100i64, 200, 300]);
    }

    #[test]
    fn upcast_decimal_values_i64_to_i128() {
        let decimal_dtype = DecimalDType::new(18, 4);
        let array = DecimalArray::new(
            buffer![10000i64, 20000, 30000],
            decimal_dtype,
            Validity::NonNullable,
        );

        let array = array.as_view();
        let casted = upcast_decimal_values(array, DecimalType::I128).unwrap();

        assert_eq!(casted.values_type(), DecimalType::I128);
        assert_eq!(casted.decimal_dtype(), decimal_dtype);

        let buffer = casted.buffer::<i128>();
        assert_eq!(buffer.as_ref(), &[10000i128, 20000, 30000]);
    }

    #[test]
    fn upcast_decimal_values_same_type_returns_clone() {
        let decimal_dtype = DecimalDType::new(10, 2);
        let array = DecimalArray::new(
            buffer![100i32, 200, 300],
            decimal_dtype,
            Validity::NonNullable,
        );

        let array = array.as_view();
        let casted = upcast_decimal_values(array, DecimalType::I32).unwrap();

        assert_eq!(casted.values_type(), DecimalType::I32);
        assert_eq!(casted.decimal_dtype(), decimal_dtype);
    }

    #[test]
    fn upcast_decimal_values_with_nulls() {
        let decimal_dtype = DecimalDType::new(10, 2);
        let array = DecimalArray::from_option_iter([Some(100i32), None, Some(300)], decimal_dtype);

        let array = array.as_view();
        let casted = upcast_decimal_values(array, DecimalType::I64).unwrap();

        assert_eq!(casted.values_type(), DecimalType::I64);
        assert_eq!(casted.len(), 3);

        // Check validity is preserved
        let mask = casted
            .as_ref()
            .validity()
            .unwrap()
            .execute_mask(
                casted.as_ref().len(),
                &mut array_session().create_execution_ctx(),
            )
            .unwrap();
        assert!(mask.value(0));
        assert!(!mask.value(1));
        assert!(mask.value(2));

        // Check non-null values
        let buffer = casted.buffer::<i64>();
        assert_eq!(buffer[0], 100);
        assert_eq!(buffer[2], 300);
    }

    #[test]
    fn upcast_decimal_values_downcast_fails() {
        let decimal_dtype = DecimalDType::new(18, 4);
        let array = DecimalArray::new(
            buffer![10000i64, 20000, 30000],
            decimal_dtype,
            Validity::NonNullable,
        );

        // Attempt to downcast from i64 to i32 should fail
        let array = array.as_view();
        let result = upcast_decimal_values(array, DecimalType::I32);
        assert!(result.is_err());
        assert!(
            result
                .unwrap_err()
                .to_string()
                .contains("Cannot downcast decimal values")
        );
    }

    #[test]
    fn cast_decimal_f64() {
        let dtype = DecimalDType::new(6, 2);
        let array = DecimalArray::new(buffer![125i32, 250, -375], dtype, Validity::NonNullable);
        let target = DType::Primitive(PType::F64, Nullability::NonNullable);
        let casted = array.into_array().cast(target.clone()).unwrap();
        assert_eq!(casted.dtype(), &target);

        let mut ctx = array_session().create_execution_ctx();
        let primitive = casted
            .execute::<Canonical>(&mut ctx)
            .unwrap()
            .into_primitive();
        assert_eq!(primitive.to_buffer::<f64>().as_ref(), &[1.25, 2.5, -3.75]);
    }

    #[test]
    fn cast_decimal_f64_null() {
        let values = [Some(100i32), None, Some(-200)];
        let array = DecimalArray::from_option_iter(values, DecimalDType::new(6, 2));
        let target = DType::Primitive(PType::F64, Nullability::Nullable);
        let casted = array.into_array().cast(target.clone()).unwrap();
        assert_eq!(casted.dtype(), &target);

        let mut ctx = array_session().create_execution_ctx();
        let primitive = casted
            .execute::<Canonical>(&mut ctx)
            .unwrap()
            .into_primitive();

        assert!(primitive.is_valid(0, &mut ctx).unwrap());
        assert!(!primitive.is_valid(1, &mut ctx).unwrap());
        assert!(primitive.is_valid(2, &mut ctx).unwrap());

        assert_eq!(
            primitive.execute_scalar(0, &mut ctx).unwrap(),
            Scalar::from(1f64)
        );
        assert_eq!(
            primitive.execute_scalar(2, &mut ctx).unwrap(),
            Scalar::from(-2f64)
        );
    }

    #[test]
    fn cast_decimal_f64_all_null() {
        let dtype = DecimalDType::new(6, 2);
        let buf = buffer![i32::MAX, i32::MIN, 12345];
        let array = DecimalArray::new(buf, dtype, Validity::AllInvalid);
        let target = DType::Primitive(PType::F64, Nullability::Nullable);
        let casted = array.into_array().cast(target.clone()).unwrap();
        assert_eq!(casted.dtype(), &target);

        let primitive = casted
            .execute::<Canonical>(&mut array_session().create_execution_ctx())
            .unwrap()
            .into_primitive();
        let mut ctx = array_session().create_execution_ctx();
        for i in 0..2 {
            assert!(!primitive.is_valid(i, &mut ctx).unwrap());
        }
        assert_eq!(primitive.to_buffer::<f64>().as_ref(), &[0.0, 0.0, 0.0]);
    }
}