MinMaxAggregateFn

Signed-off-by: Nicholas Gates <nick@nickgates.com>

Author: gatesn

encodings/runend/src/compute/min_max.rs

@@ -0,0 +1,47 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+use vortex_array::ArrayRef;
+use vortex_array::ExecutionCtx;
+use vortex_array::aggregate_fn::AggregateFnRef;
+use vortex_array::aggregate_fn::fns::min_max::MinMax;
+use vortex_array::aggregate_fn::fns::min_max::make_struct_dtype;
+use vortex_array::aggregate_fn::fns::min_max::min_max;
+use vortex_array::aggregate_fn::kernels::DynAggregateKernel;
+use vortex_array::scalar::Scalar;
+use vortex_error::VortexResult;
+
+use crate::RunEnd;
+
+/// RunEnd-specific min/max kernel.
+///
+/// Run-end encoded arrays store each unique run value once, so min/max can be computed directly
+/// on the values array without decoding.
+#[derive(Debug)]
+pub(crate) struct RunEndMinMaxKernel;
+
+impl DynAggregateKernel for RunEndMinMaxKernel {
+    fn aggregate(
+        &self,
+        aggregate_fn: &AggregateFnRef,
+        batch: &ArrayRef,
+        ctx: &mut ExecutionCtx,
+    ) -> VortexResult<Option<Scalar>> {
+        if !aggregate_fn.is::<MinMax>() {
+            return Ok(None);
+        }
+
+        let Some(run_end) = batch.as_opt::<RunEnd>() else {
+            return Ok(None);
+        };
+
+        let struct_dtype = make_struct_dtype(batch.dtype());
+        match min_max(run_end.values(), ctx)? {
+            Some(result) => Ok(Some(Scalar::struct_(
+                struct_dtype,
+                vec![result.min, result.max],
+            ))),
+            None => Ok(Some(Scalar::null(struct_dtype))),
+        }
+    }
+}

encodings/sequence/src/compute/min_max.rs

@@ -0,0 +1,80 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+use vortex_array::ArrayRef;
+use vortex_array::ExecutionCtx;
+use vortex_array::aggregate_fn::AggregateFnRef;
+use vortex_array::aggregate_fn::fns::min_max::MinMax;
+use vortex_array::aggregate_fn::fns::min_max::make_struct_dtype;
+use vortex_array::aggregate_fn::kernels::DynAggregateKernel;
+use vortex_array::dtype::DType;
+use vortex_array::dtype::Nullability;
+use vortex_array::match_each_pvalue;
+use vortex_array::scalar::PValue;
+use vortex_array::scalar::Scalar;
+use vortex_array::scalar::ScalarValue;
+use vortex_error::VortexResult;
+
+use crate::Sequence;
+
+/// Sequence-specific min/max kernel.
+///
+/// A sequence array represents `A[i] = base + i * multiplier`, so min/max can be computed
+/// algebraically from `base` and `last` based on the sign of the multiplier.
+#[derive(Debug)]
+pub(crate) struct SequenceMinMaxKernel;
+
+impl DynAggregateKernel for SequenceMinMaxKernel {
+    fn aggregate(
+        &self,
+        aggregate_fn: &AggregateFnRef,
+        batch: &ArrayRef,
+        _ctx: &mut ExecutionCtx,
+    ) -> VortexResult<Option<Scalar>> {
+        if !aggregate_fn.is::<MinMax>() {
+            return Ok(None);
+        }
+
+        let Some(seq) = batch.as_opt::<Sequence>() else {
+            return Ok(None);
+        };
+
+        let struct_dtype = make_struct_dtype(batch.dtype());
+
+        // Empty sequences shouldn't exist (try_new validates length), but handle gracefully.
+        if seq.is_empty() {
+            return Ok(Some(Scalar::null(struct_dtype)));
+        }
+
+        let base = seq.base();
+        let last = seq.last();
+
+        // Determine min and max based on multiplier direction.
+        // For unsigned types, multiplier is always >= 0.
+        let (min_pvalue, max_pvalue) = match_each_pvalue!(
+            seq.multiplier(),
+            uint: |_v| { (base, last) },
+            int: |v| {
+                if v >= 0 {
+                    (base, last)
+                } else {
+                    (last, base)
+                }
+            },
+            float: |_v| { unreachable!("float multiplier not supported for SequenceArray") }
+        );
+
+        let non_nullable_dtype = DType::Primitive(seq.ptype(), Nullability::NonNullable);
+        let min_scalar = Scalar::try_new(
+            non_nullable_dtype.clone(),
+            Some(ScalarValue::Primitive(min_pvalue)),
+        )?;
+        let max_scalar =
+            Scalar::try_new(non_nullable_dtype, Some(ScalarValue::Primitive(max_pvalue)))?;
+
+        Ok(Some(Scalar::struct_(
+            struct_dtype,
+            vec![min_scalar, max_scalar],
+        )))
+    }
+}

vortex-array/src/arrays/dict/compute/min_max.rs

@@ -0,0 +1,57 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+use vortex_error::VortexResult;
+use vortex_mask::Mask;
+
+use crate::ArrayRef;
+use crate::ExecutionCtx;
+use crate::aggregate_fn::AggregateFnRef;
+use crate::aggregate_fn::fns::min_max::MinMax;
+use crate::aggregate_fn::fns::min_max::make_struct_dtype;
+use crate::aggregate_fn::fns::min_max::min_max;
+use crate::aggregate_fn::kernels::DynAggregateKernel;
+use crate::arrays::Dict;
+use crate::scalar::Scalar;
+
+/// Dict-specific min/max kernel.
+///
+/// When all dictionary values are referenced, min/max can be computed directly on the values
+/// array. Otherwise, unreferenced values are filtered out first.
+#[derive(Debug)]
+pub(crate) struct DictMinMaxKernel;
+
+impl DynAggregateKernel for DictMinMaxKernel {
+    fn aggregate(
+        &self,
+        aggregate_fn: &AggregateFnRef,
+        batch: &ArrayRef,
+        ctx: &mut ExecutionCtx,
+    ) -> VortexResult<Option<Scalar>> {
+        if !aggregate_fn.is::<MinMax>() {
+            return Ok(None);
+        }
+
+        let Some(dict) = batch.as_opt::<Dict>() else {
+            return Ok(None);
+        };
+
+        let struct_dtype = make_struct_dtype(batch.dtype());
+
+        let result = if dict.has_all_values_referenced() {
+            // All values are referenced, compute min/max directly on the values array.
+            min_max(dict.values(), ctx)?
+        } else {
+            // Filter to only referenced values, then compute min/max.
+            let referenced_mask = dict.compute_referenced_values_mask(true)?;
+            let mask = Mask::from(referenced_mask);
+            let filtered_values = dict.values().filter(mask)?;
+            min_max(&filtered_values, ctx)?
+        };
+
+        match result {
+            Some(r) => Ok(Some(Scalar::struct_(struct_dtype, vec![r.min, r.max]))),
+            None => Ok(Some(Scalar::null(struct_dtype))),
+        }
+    }
+}