Add a scan for the range of valid decimal values

vortex-array/benches/aggregate_max.rs

@@ -11,11 +11,17 @@ use vortex_array::VortexSessionExecute;
 use vortex_array::aggregate_fn::NumericalAggregateOpts;
 use vortex_array::aggregate_fn::fns::min_max::min_max;
 use vortex_array::aggregate_fn::session::AggregateFnSession;
+use vortex_array::arrays::DecimalArray;
 use vortex_array::arrays::PrimitiveArray;
 use vortex_array::arrays::VarBinViewArray;
 use vortex_array::dtype::DType;
+use vortex_array::dtype::DecimalDType;
+use vortex_array::dtype::NativeDecimalType;
 use vortex_array::dtype::Nullability;
+use vortex_array::dtype::i256;
 use vortex_array::optimizer::kernels::KernelSession;
+use vortex_array::validity::Validity;
+use vortex_buffer::Buffer;
 use vortex_session::VortexSession;
 
 #[global_allocator]
@@ -63,6 +69,62 @@ fn max_i64(bencher: Bencher) {
         .bench_refs(|(a, ctx)| a.statistics().compute_max::<i64>(ctx));
 }
 
+#[derive(Clone, Copy, Debug)]
+enum DecimalValidity {
+    NonNullable,
+    ScatteredNulls,
+}
+
+const DECIMAL_VALIDITIES: &[DecimalValidity] = &[
+    DecimalValidity::NonNullable,
+    DecimalValidity::ScatteredNulls,
+];
+
+fn bench_decimal_min_max<T: NativeDecimalType>(
+    bencher: Bencher,
+    values: Buffer<T>,
+    decimal_dtype: DecimalDType,
+    validity: DecimalValidity,
+) {
+    let validity = match validity {
+        DecimalValidity::NonNullable => Validity::NonNullable,
+        DecimalValidity::ScatteredNulls => {
+            let mut rng = StdRng::seed_from_u64(8);
+            Validity::from_iter((0..N).map(|_| rng.random_bool(0.5)))
+        }
+    };
+
+    bencher
+        .with_inputs(|| {
+            (
+                DecimalArray::new(values.clone(), decimal_dtype, validity.clone()).into_array(),
+                SESSION.create_execution_ctx(),
+            )
+        })
+        .bench_refs(|(array, ctx)| min_max(array, ctx, NumericalAggregateOpts::default()));
+}
+
+#[divan::bench(args = DECIMAL_VALIDITIES)]
+fn min_max_decimal_i128(bencher: Bencher, validity: DecimalValidity) {
+    let mut rng = StdRng::seed_from_u64(9);
+    let bound = 10i128.pow(38) - 1;
+    let values = (0..N).map(|_| rng.random_range(-bound..=bound)).collect();
+
+    bench_decimal_min_max(bencher, values, DecimalDType::new(38, 2), validity);
+}
+
+#[divan::bench(args = DECIMAL_VALIDITIES)]
+fn min_max_decimal_i256(bencher: Bencher, validity: DecimalValidity) {
+    let mut rng = StdRng::seed_from_u64(10);
+    let factor = i256::from_i128(10i128.pow(38));
+    let bound = 10i128.pow(37) - 1;
+    let values = (0..N)
+        .map(|_| i256::from_i128(rng.random_range(-bound..=bound)) * factor)
+        .collect();
+
+    bench_decimal_min_max(bencher, values, DecimalDType::new(76, 2), validity);
+}
+
 #[divan::bench]
 fn max_f64(bencher: Bencher) {
     let mut rng = StdRng::seed_from_u64(3);

vortex-array/src/aggregate_fn/fns/min_max/decimal.rs

@@ -11,7 +11,6 @@ use crate::ExecutionCtx;
 use crate::aggregate_fn::AggregateArgs;
 use crate::aggregate_fn::NumericalAggregateOpts;
 use crate::arrays::DecimalArray;
-use crate::dtype::DecimalDType;
 use crate::dtype::NativeDecimalType;
 use crate::dtype::Nullability::NonNullable;
 use crate::match_each_decimal_value_type;
@@ -38,45 +37,67 @@ fn compute_min_max_with_validity<D>(
 where
     D: Into<DecimalValue> + NativeDecimalType,
 {
+    let validity = array
+        .as_ref()
+        .validity()?
+        .execute_mask(array.as_ref().len(), ctx)?;
+    if matches!(&validity, Mask::AllFalse(_)) {
+        return Ok(None);
+    }
+
+    let decimal_dtype = array.decimal_dtype();
+
     Ok(
-        match array
-            .as_ref()
-            .validity()?
-            .execute_mask(array.as_ref().len(), ctx)?
-        {
-            Mask::AllTrue(_) => compute_min_max(array.buffer::<D>().iter(), array.decimal_dtype()),
-            Mask::AllFalse(_) => None,
-            Mask::Values(v) => compute_min_max(
-                array
-                    .buffer::<D>()
-                    .iter()
-                    .zip(v.bit_buffer().iter())
-                    .filter_map(|(v, m)| m.then_some(v)),
-                array.decimal_dtype(),
-            ),
-        },
+        valid_range(array.buffer::<D>().as_slice(), &validity).map(|(min, max)| {
+            let min_scalar = Scalar::decimal(min.into(), decimal_dtype, NonNullable);
+            let max_scalar = if min == max {
+                min_scalar.clone()
+            } else {
+                Scalar::decimal(max.into(), decimal_dtype, NonNullable)
+            };
+
+            MinMaxResult {
+                min: min_scalar,
+                max: max_scalar,
+            }
+        }),
     )
 }
 
-fn compute_min_max<'a, T>(
-    iter: impl Iterator<Item = &'a T>,
-    decimal_dtype: DecimalDType,
-) -> Option<MinMaxResult>
+/// Returns the smallest and the largest valid value, or `None` when no value is valid.
+///
+/// Null slots can hold any value and do not contribute to the range. The scan allocates nothing
+/// and does not check decimal precision.
+///
+/// # Panics
+///
+/// Panics if `validity` and `values` have different lengths.
+pub fn valid_range<T: NativeDecimalType>(values: &[T], validity: &Mask) -> Option<(T, T)> {
+    assert_eq!(
+        validity.len(),
+        values.len(),
+        "valid_range requires one validity bit per value"
+    );
+
+    match validity {
+        Mask::AllTrue(_) => compute_min_max(values.iter()),
+        Mask::AllFalse(_) => None,
+        Mask::Values(validity) => compute_min_max(
+            values
+                .iter()
+                .zip(validity.bit_buffer().iter())
+                .filter_map(|(value, valid)| valid.then_some(value)),
+        ),
+    }
+}
+
+fn compute_min_max<'a, T>(iter: impl Iterator<Item = &'a T>) -> Option<(T, T)>
 where
-    T: Into<DecimalValue> + NativeDecimalType + Ord + Copy + 'a,
+    T: NativeDecimalType,
 {
     match iter.minmax_by(|a, b| a.cmp(b)) {
         itertools::MinMaxResult::NoElements => None,
-        itertools::MinMaxResult::OneElement(&x) => {
-            let scalar = Scalar::decimal(x.into(), decimal_dtype, NonNullable);
-            Some(MinMaxResult {
-                min: scalar.clone(),
-                max: scalar,
-            })
-        }
-        itertools::MinMaxResult::MinMax(&min, &max) => Some(MinMaxResult {
-            min: Scalar::decimal(min.into(), decimal_dtype, NonNullable),
-            max: Scalar::decimal(max.into(), decimal_dtype, NonNullable),
-        }),
+        itertools::MinMaxResult::OneElement(&value) => Some((value, value)),
+        itertools::MinMaxResult::MinMax(&min, &max) => Some((min, max)),
     }
 }

vortex-array/src/aggregate_fn/fns/min_max/mod.rs

@@ -3,6 +3,7 @@
 
 mod bool;
 mod decimal;
+pub use decimal::valid_range;
 mod extension;
 mod fixed_size_list;
 mod primitive;
@@ -535,6 +536,7 @@ mod tests {
     use std::sync::Arc;
     use std::sync::LazyLock;
 
+    use rstest::rstest;
     use vortex_buffer::BitBuffer;
     use vortex_buffer::Buffer;
     use vortex_buffer::buffer;
@@ -566,8 +568,10 @@ mod tests {
     use crate::arrays::NullArray;
     use crate::arrays::PrimitiveArray;
     use crate::arrays::VarBinArray;
+    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::extension::ExtDType;
@@ -576,6 +580,7 @@ mod tests {
     use crate::expr::stats::Precision;
     use crate::expr::stats::Stat;
     use crate::expr::stats::StatsProvider;
+    use crate::match_each_decimal_value_type;
     use crate::scalar::DecimalValue;
     use crate::scalar::Scalar;
     use crate::scalar::ScalarValue;
@@ -995,6 +1000,66 @@ mod tests {
         Ok(())
     }
 
+    #[rstest]
+    #[case::non_nullable(vec![Some(-3), Some(7), Some(5)], false, Some((-3, 7)))]
+    #[case::all_valid(vec![Some(-3), Some(7), Some(5)], true, Some((-3, 7)))]
+    #[case::null_extremes(vec![None, Some(-3), Some(7), None], true, Some((-3, 7)))]
+    #[case::single_valid(vec![None, Some(5), None], true, Some((5, 5)))]
+    #[case::single_row(vec![Some(5)], false, Some((5, 5)))]
+    #[case::equal_values(vec![Some(5), Some(5)], true, Some((5, 5)))]
+    #[case::all_null(vec![None, None], true, None)]
+    #[case::empty_non_nullable(vec![], false, None)]
+    #[case::empty_nullable(vec![], true, None)]
+    fn test_decimal_valid_range(
+        #[case] values: Vec<Option<i8>>,
+        #[case] nullable: bool,
+        #[case] expected: Option<(i8, i8)>,
+        #[values(
+            DecimalType::I8,
+            DecimalType::I16,
+            DecimalType::I32,
+            DecimalType::I64,
+            DecimalType::I128,
+            DecimalType::I256
+        )]
+        storage: DecimalType,
+    ) -> VortexResult<()> {
+        let mut ctx = SESSION.create_execution_ctx();
+        let decimal_dtype = DecimalDType::new(9, 2);
+        let validity = if nullable {
+            Validity::from_iter(values.iter().map(Option::is_some))
+        } else {
+            Validity::NonNullable
+        };
+        let expected = expected.map(|(min, max)| MinMaxResult {
+            min: Scalar::decimal(min.into(), decimal_dtype, Nullability::NonNullable),
+            max: Scalar::decimal(max.into(), decimal_dtype, Nullability::NonNullable),
+        });
+
+        match_each_decimal_value_type!(storage, |T| {
+            let buffer: Buffer<T> = values
+                .iter()
+                .enumerate()
+                .map(|(row, value)| match value {
+                    Some(value) => <T as BigCast>::from(*value)
+                        .vortex_expect("i8 fixture values fit every decimal storage type"),
+                    None if row % 2 == 0 => T::MIN,
+                    None => T::MAX,
+                })
+                .collect();
+            let array = DecimalArray::new(buffer, decimal_dtype, validity).into_array();
+            let actual = min_max(&array, &mut ctx, NumericalAggregateOpts::default())?;
+            assert_eq!(actual, expected);
+            if let Some(actual) = actual {
+                let dtype = DType::Decimal(decimal_dtype, Nullability::NonNullable);
+                assert_eq!(actual.min.dtype(), &dtype);
+                assert_eq!(actual.max.dtype(), &dtype);
+            }
+
+            Ok(())
+        })
+    }
+
     #[test]
     fn list_and_fixed_size_list_return_dtype() {
         let element_dtype = DType::Primitive(PType::I32, Nullability::Nullable);

vortex-array/src/arrays/decimal/utils.rs

@@ -13,6 +13,7 @@ use vortex_error::vortex_ensure_eq;
 use vortex_mask::AllOr;
 use vortex_mask::Mask;
 
+pub use crate::aggregate_fn::fns::min_max::valid_range;
 use crate::arrays::DecimalArray;
 use crate::arrays::decimal::DecimalArrayExt;
 use crate::buffer::BufferHandle;
@@ -318,6 +319,7 @@ mod tests {
     use super::check_precision;
     use super::converted_buffer;
     use super::copy_to_canonical_width;
+    use super::valid_range;
     use crate::arrays::DecimalArray;
     use crate::dtype::BigCast;
     use crate::dtype::DecimalDType;
@@ -362,6 +364,82 @@ mod tests {
             .collect()
     }
 
+    #[rstest]
+    #[case::all_valid(vec![5, -3, 7], vec![true, true, true], Some((-3, 7)))]
+    #[case::partly_valid(vec![5, -3, 7], vec![true, true, false], Some((-3, 5)))]
+    #[case::all_null(vec![5, -3, 7], vec![false, false, false], None)]
+    #[case::empty(vec![], vec![], None)]
+    #[case::one_valid(vec![5, -3, 7], vec![false, true, false], Some((-3, -3)))]
+    #[case::outside_i8_precision(vec![-100, 100], vec![true, true], Some((-100, 100)))]
+    fn test_valid_range(
+        #[case] values: Vec<i8>,
+        #[case] validity: Vec<bool>,
+        #[case] expected: Option<(i8, i8)>,
+        #[values(
+            DecimalType::I8,
+            DecimalType::I16,
+            DecimalType::I32,
+            DecimalType::I64,
+            DecimalType::I128,
+            DecimalType::I256
+        )]
+        storage: DecimalType,
+    ) -> VortexResult<()> {
+        let validity = Mask::from_iter(validity);
+        match_each_decimal_value_type!(storage, |T| {
+            let values: Vec<T> = values
+                .iter()
+                .map(|&value| {
+                    <T as BigCast>::from(value).vortex_expect("i8 values fit every decimal type")
+                })
+                .collect();
+            let expected = expected.map(|(min, max)| {
+                (
+                    <T as BigCast>::from(min).vortex_expect("i8 values fit every decimal type"),
+                    <T as BigCast>::from(max).vortex_expect("i8 values fit every decimal type"),
+                )
+            });
+            assert_eq!(valid_range(&values, &validity), expected);
+        });
+
+        Ok(())
+    }
+
+    #[rstest]
+    fn test_valid_range_ignores_null_extremes(
+        #[values(
+            DecimalType::I8,
+            DecimalType::I16,
+            DecimalType::I32,
+            DecimalType::I64,
+            DecimalType::I128,
+            DecimalType::I256
+        )]
+        storage: DecimalType,
+    ) -> VortexResult<()> {
+        let validity = Mask::from_iter([false, true, true, false]);
+        match_each_decimal_value_type!(storage, |T| {
+            let min = <T as BigCast>::from(-3i8).vortex_expect("-3 fits every decimal type");
+            let max = <T as BigCast>::from(7i8).vortex_expect("7 fits every decimal type");
+            assert_eq!(
+                valid_range(&[T::MIN, min, max, T::MAX], &validity),
+                Some((min, max))
+            );
+            assert_eq!(
+                valid_range(&[T::MIN, T::MAX], &Mask::AllTrue(2)),
+                Some((T::MIN, T::MAX))
+            );
+        });
+
+        Ok(())
+    }
+
+    #[test]
+    #[should_panic(expected = "valid_range requires one validity bit per value")]
+    fn test_valid_range_rejects_mismatched_mask() {
+        valid_range(&[1i8, 2], &Mask::AllTrue(3));
+    }
+
     /// Each pair selects a different widening or narrowing conversion.
     #[rstest]
     fn converted_buffer_every_storage_pair(