vortex_array/aggregate_fn/fns/min_max/
mod.rs

1// SPDX-License-Identifier: Apache-2.0
2// SPDX-FileCopyrightText: Copyright the Vortex contributors
3
4mod bool;
5mod decimal;
6mod extension;
7mod primitive;
8mod varbin;
9
10use std::sync::LazyLock;
11
12use vortex_error::VortexExpect;
13use vortex_error::VortexResult;
14use vortex_error::vortex_bail;
15use vortex_error::vortex_panic;
16use vortex_session::registry::CachedId;
17
18use self::bool::accumulate_bool;
19use self::decimal::accumulate_decimal;
20use self::extension::accumulate_extension;
21use self::primitive::accumulate_primitive;
22use self::varbin::accumulate_varbinview;
23use crate::ArrayRef;
24use crate::Canonical;
25use crate::Columnar;
26use crate::ExecutionCtx;
27use crate::aggregate_fn::Accumulator;
28use crate::aggregate_fn::AggregateArgs;
29use crate::aggregate_fn::AggregateFnId;
30use crate::aggregate_fn::AggregateFnVTable;
31use crate::aggregate_fn::DynAccumulator;
32use crate::aggregate_fn::NumericalAggregateOpts;
33use crate::dtype::DType;
34use crate::dtype::FieldNames;
35use crate::dtype::Nullability;
36use crate::dtype::PType;
37use crate::dtype::StructFields;
38use crate::dtype::half::f16;
39use crate::expr::stats::Precision;
40use crate::expr::stats::Stat;
41use crate::expr::stats::StatsProvider;
42use crate::expr::stats::StatsProviderExt;
43use crate::partial_ord::partial_max;
44use crate::partial_ord::partial_min;
45use crate::scalar::Scalar;
46
47static NAMES: LazyLock<FieldNames> = LazyLock::new(|| FieldNames::from(["min", "max"]));
48
49/// The minimum and maximum non-null values of an array, or `None` if there are no non-null values.
50///
51/// NaN handling for float inputs is controlled by [`NumericalAggregateOpts`]: with `skip_nans` (the
52/// default) NaN values are ignored and the cached `Stat::Min`/`Stat::Max` statistics are consulted
53/// and updated. With `skip_nans=false`, any NaN value in a float array poisons both extrema to
54/// NaN; an exact `Stat::NaNCount` statistic shortcircuits the NaN scan in either direction.
55///
56/// The result scalars have the non-nullable version of the array dtype.
57/// This will update the stats set of the array as a side effect.
58pub fn min_max(
59    array: &ArrayRef,
60    ctx: &mut ExecutionCtx,
61    options: NumericalAggregateOpts,
62) -> VortexResult<Option<MinMaxResult>> {
63    if !options.skip_nans && array.dtype().is_float() {
64        match array.statistics().get_as::<u64>(Stat::NaNCount) {
65            // NaN-free: identical to the NaN-skipping path below, including its stat caching.
66            Precision::Exact(0) => {}
67            // At least one NaN value poisons both extrema.
68            Precision::Exact(_) => return Ok(Some(nan_minmax_result(array.dtype()))),
69            _ => {
70                if array.is_empty() || array.valid_count(ctx)? == 0 {
71                    return Ok(None);
72                }
73                // Compute with NaN-including options; the NaN-skipping `Stat::Min`/`Stat::Max`
74                // caches are neither read nor written.
75                let mut acc = Accumulator::try_new(MinMax, options, array.dtype().clone())?;
76                acc.accumulate(array, ctx)?;
77                return MinMaxResult::from_scalar(acc.finish()?);
78            }
79        }
80    }
81
82    // NaN-skipping path. Also reached for NaN-free not-skipping float arrays and all non-float
83    // arrays, where `skip_nans` has no effect.
84
85    // Short-circuit using cached array statistics.
86    let cached_min = array.statistics().get(Stat::Min).as_exact();
87    let cached_max = array.statistics().get(Stat::Max).as_exact();
88    if let Some((min, max)) = cached_min.zip(cached_max) {
89        let non_nullable_dtype = array.dtype().as_nonnullable();
90        return Ok(Some(MinMaxResult {
91            min: min.cast(&non_nullable_dtype)?,
92            max: max.cast(&non_nullable_dtype)?,
93        }));
94    }
95
96    // Short-circuit for empty arrays or all-null arrays.
97    if array.is_empty() || array.valid_count(ctx)? == 0 {
98        return Ok(None);
99    }
100
101    // Short-circuit for dtypes this helper cannot currently compute.
102    if !minmax_compute_supported_dtype(array.dtype()) {
103        return Ok(None);
104    }
105
106    // Compute using Accumulator<MinMax>.
107    let mut acc = Accumulator::try_new(
108        MinMax,
109        NumericalAggregateOpts::default(),
110        array.dtype().clone(),
111    )?;
112    acc.accumulate(array, ctx)?;
113    let result_scalar = acc.finish()?;
114    let result = MinMaxResult::from_scalar(result_scalar)?;
115
116    // Cache the computed min/max as statistics.
117    if let Some(r) = &result {
118        if let Some(min_value) = r.min.value() {
119            array
120                .statistics()
121                .set(Stat::Min, Precision::Exact(min_value.clone()));
122        }
123        if let Some(max_value) = r.max.value() {
124            array
125                .statistics()
126                .set(Stat::Max, Precision::Exact(max_value.clone()));
127        }
128    }
129
130    Ok(result)
131}
132
133/// A `{min: NaN, max: NaN}` result for a poisoned NaN-including min/max over `dtype`.
134fn nan_minmax_result(dtype: &DType) -> MinMaxResult {
135    let nan = nan_scalar(dtype);
136    MinMaxResult {
137        min: nan.clone(),
138        max: nan,
139    }
140}
141
142/// A non-nullable NaN scalar of the float `dtype`.
143pub(crate) fn nan_scalar(dtype: &DType) -> Scalar {
144    match dtype.as_ptype() {
145        PType::F16 => Scalar::primitive(f16::NAN, Nullability::NonNullable),
146        PType::F32 => Scalar::primitive(f32::NAN, Nullability::NonNullable),
147        PType::F64 => Scalar::primitive(f64::NAN, Nullability::NonNullable),
148        _ => vortex_panic!("NaN scalar requested for non-float dtype {dtype}"),
149    }
150}
151
152/// Whether a scalar holds a primitive float NaN value.
153pub(crate) fn scalar_is_nan(scalar: &Scalar) -> bool {
154    if !scalar.dtype().is_float() {
155        return false;
156    }
157
158    scalar.as_primitive_opt().is_some_and(|p| p.is_nan())
159}
160
161/// The minimum and maximum non-null values of an array.
162#[derive(Debug, Clone, PartialEq, Eq)]
163pub struct MinMaxResult {
164    pub min: Scalar,
165    pub max: Scalar,
166}
167
168impl MinMaxResult {
169    /// Extract a `MinMaxResult` from a struct scalar with `{min, max}` fields.
170    pub fn from_scalar(scalar: Scalar) -> VortexResult<Option<Self>> {
171        if scalar.is_null() {
172            Ok(None)
173        } else {
174            let min = scalar
175                .as_struct()
176                .field_by_idx(0)
177                .vortex_expect("missing min field");
178            let max = scalar
179                .as_struct()
180                .field_by_idx(1)
181                .vortex_expect("missing max field");
182            Ok(Some(MinMaxResult { min, max }))
183        }
184    }
185}
186
187/// Compute the min and max of an array.
188///
189/// Returns a nullable struct scalar `{min: T, max: T}` where `T` is the non-nullable input dtype.
190/// The struct is null when the array is empty or all-null.
191///
192/// NaN handling for float inputs is controlled by [`NumericalAggregateOpts`]: with `skip_nans` (the
193/// default) NaN values are ignored, otherwise any NaN value poisons both extrema to NaN.
194#[derive(Clone, Debug)]
195pub struct MinMax;
196
197/// Partial accumulator state for min/max.
198pub struct MinMaxPartial {
199    min: Option<Scalar>,
200    max: Option<Scalar>,
201}
202
203impl MinMaxPartial {
204    /// Merge a local `MinMaxResult` into this partial state.
205    fn merge(
206        &mut self,
207        args: AggregateArgs<'_, NumericalAggregateOpts>,
208        local: Option<MinMaxResult>,
209    ) {
210        let Some(MinMaxResult { min, max }) = local else {
211            return;
212        };
213
214        // NaN scalars are incomparable under `partial_min`/`partial_max`, so they are handled
215        // explicitly: a NaN extremum poisons the partial state when NaNs participate, and is
216        // dropped when they are skipped.
217        if scalar_is_nan(&min) || scalar_is_nan(&max) || self.is_poisoned() {
218            if !args.options.skip_nans {
219                self.poison(args);
220            }
221            return;
222        }
223
224        self.min = Some(match self.min.take() {
225            Some(current) => partial_min(min, current).vortex_expect("incomparable min scalars"),
226            None => min,
227        });
228
229        self.max = Some(match self.max.take() {
230            Some(current) => partial_max(max, current).vortex_expect("incomparable max scalars"),
231            None => max,
232        });
233    }
234
235    /// Poison the partial state to `{min: NaN, max: NaN}`.
236    fn poison(&mut self, args: AggregateArgs<'_, NumericalAggregateOpts>) {
237        let nan = nan_scalar(args.dtype);
238        self.min = Some(nan.clone());
239        self.max = Some(nan);
240    }
241
242    /// Whether the partial state is poisoned to NaN.
243    fn is_poisoned(&self) -> bool {
244        self.min.as_ref().is_some_and(scalar_is_nan)
245    }
246}
247
248/// Creates the struct dtype `{min: T, max: T}` (nullable) used for min/max aggregate results.
249pub fn make_minmax_dtype(element_dtype: &DType) -> DType {
250    DType::Struct(
251        StructFields::new(
252            NAMES.clone(),
253            vec![
254                element_dtype.as_nonnullable(),
255                element_dtype.as_nonnullable(),
256            ],
257        ),
258        Nullability::Nullable,
259    )
260}
261
262fn minmax_supported_dtype(input_dtype: &DType) -> bool {
263    match input_dtype {
264        DType::Bool(_)
265        | DType::Primitive(..)
266        | DType::Decimal(..)
267        | DType::Utf8(..)
268        | DType::Binary(..)
269        | DType::Extension(..) => true,
270        DType::List(element_dtype, _) => minmax_supported_dtype(element_dtype),
271        DType::FixedSizeList(element_dtype, ..) => minmax_supported_dtype(element_dtype),
272        _ => false,
273    }
274}
275
276/// Returns whether [`min_max`] can currently compute extrema for this logical dtype.
277///
278/// This is intentionally narrower than [`minmax_supported_dtype`]. List and fixed-size-list
279/// extrema have a defined output dtype for aggregate expression lowering, but the accumulator does
280/// not yet implement lexicographic list comparison.
281fn minmax_compute_supported_dtype(input_dtype: &DType) -> bool {
282    matches!(
283        input_dtype,
284        DType::Bool(_)
285            | DType::Primitive(..)
286            | DType::Decimal(..)
287            | DType::Utf8(..)
288            | DType::Binary(..)
289            | DType::Extension(..)
290    )
291}
292
293impl AggregateFnVTable for MinMax {
294    type Options = NumericalAggregateOpts;
295    type Partial = MinMaxPartial;
296
297    fn id(&self) -> AggregateFnId {
298        static ID: CachedId = CachedId::new("vortex.min_max");
299        *ID
300    }
301
302    fn serialize(&self, _options: &Self::Options) -> VortexResult<Option<Vec<u8>>> {
303        Ok(None)
304    }
305
306    fn return_dtype(&self, _options: &Self::Options, input_dtype: &DType) -> Option<DType> {
307        minmax_supported_dtype(input_dtype).then(|| make_minmax_dtype(input_dtype))
308    }
309
310    fn partial_dtype(&self, options: &Self::Options, input_dtype: &DType) -> Option<DType> {
311        self.return_dtype(options, input_dtype)
312    }
313
314    fn empty_partial(
315        &self,
316        _args: AggregateArgs<'_, Self::Options>,
317    ) -> VortexResult<Self::Partial> {
318        Ok(MinMaxPartial {
319            min: None,
320            max: None,
321        })
322    }
323
324    fn partial_from_scalar(
325        &self,
326        args: AggregateArgs<'_, Self::Options>,
327        scalar: Scalar,
328    ) -> VortexResult<Self::Partial> {
329        let mut partial = MinMaxPartial {
330            min: None,
331            max: None,
332        };
333        // `merge` normalizes the parsed extrema: nulls stay empty and NaNs poison or drop.
334        partial.merge(args, MinMaxResult::from_scalar(scalar)?);
335        Ok(partial)
336    }
337
338    fn merge_partials(
339        &self,
340        args: AggregateArgs<'_, Self::Options>,
341        mut first: Self::Partial,
342        second: Self::Partial,
343    ) -> VortexResult<Self::Partial> {
344        if let (Some(min), Some(max)) = (second.min, second.max) {
345            first.merge(args, Some(MinMaxResult { min, max }));
346        }
347        Ok(first)
348    }
349
350    fn to_scalar(
351        &self,
352        args: AggregateArgs<'_, Self::Options>,
353        partial: &Self::Partial,
354    ) -> VortexResult<Scalar> {
355        let dtype = args.partial_dtype.clone();
356        Ok(match (&partial.min, &partial.max) {
357            (Some(min), Some(max)) => Scalar::struct_(dtype, vec![min.clone(), max.clone()]),
358            _ => Scalar::null(dtype),
359        })
360    }
361
362    #[inline]
363    fn is_saturated(
364        &self,
365        _args: AggregateArgs<'_, Self::Options>,
366        partial: &Self::Partial,
367    ) -> bool {
368        // A poisoned NaN-including min/max is fully determined.
369        partial.is_poisoned()
370    }
371
372    fn try_accumulate(
373        &self,
374        args: AggregateArgs<'_, Self::Options>,
375        partial: &mut Self::Partial,
376        batch: &ArrayRef,
377        _ctx: &mut ExecutionCtx,
378    ) -> VortexResult<bool> {
379        // NaN-aware shortcircuits only apply to NaN-including float min/max; everything else
380        // takes the default dispatch path.
381        if args.options.skip_nans || !args.dtype.is_float() {
382            return Ok(false);
383        }
384        match batch.statistics().get_as::<u64>(Stat::NaNCount) {
385            Precision::Exact(0) => {
386                // NaN-free batch: the cached NaN-skipping extrema (if any) are valid.
387                let cached_min = batch.statistics().get(Stat::Min).as_exact();
388                let cached_max = batch.statistics().get(Stat::Max).as_exact();
389                if let Some((min, max)) = cached_min.zip(cached_max) {
390                    // Cached float stats carry the (possibly nullable) array dtype; `to_scalar`
391                    // builds a struct with non-nullable fields, so normalise here.
392                    let non_nullable_dtype = args.dtype.as_nonnullable();
393                    partial.merge(
394                        args,
395                        Some(MinMaxResult {
396                            min: min.cast(&non_nullable_dtype)?,
397                            max: max.cast(&non_nullable_dtype)?,
398                        }),
399                    );
400                    return Ok(true);
401                }
402                Ok(false)
403            }
404            Precision::Exact(_) => {
405                // At least one NaN value poisons both extrema without scanning the batch.
406                partial.poison(args);
407                Ok(true)
408            }
409            _ => Ok(false),
410        }
411    }
412
413    fn accumulate(
414        &self,
415        args: AggregateArgs<'_, Self::Options>,
416        partial: &mut Self::Partial,
417        batch: &Columnar,
418        ctx: &mut ExecutionCtx,
419    ) -> VortexResult<()> {
420        match batch {
421            Columnar::Constant(c) => {
422                let scalar = c.scalar();
423                if scalar.is_null() {
424                    return Ok(());
425                }
426                // NaN float constants are skipped or poison the extrema, per the options.
427                if scalar_is_nan(scalar) {
428                    if !args.options.skip_nans {
429                        partial.poison(args);
430                    }
431                    return Ok(());
432                }
433                let non_nullable_dtype = scalar.dtype().as_nonnullable();
434                let cast = scalar.cast(&non_nullable_dtype)?;
435                partial.merge(
436                    args,
437                    Some(MinMaxResult {
438                        min: cast.clone(),
439                        max: cast,
440                    }),
441                );
442                Ok(())
443            }
444            Columnar::Canonical(c) => match c {
445                Canonical::Primitive(p) => accumulate_primitive(args, partial, p, ctx),
446                Canonical::Bool(b) => accumulate_bool(args, partial, b, ctx),
447                Canonical::VarBinView(v) => accumulate_varbinview(args, partial, v, ctx),
448                Canonical::Decimal(d) => accumulate_decimal(args, partial, d, ctx),
449                Canonical::Extension(e) => accumulate_extension(args, partial, e, ctx),
450                Canonical::Null(_) => Ok(()),
451                Canonical::Union(_) => {
452                    vortex_bail!("TODO(connor)[Union]: implement min_max for Union arrays")
453                }
454                Canonical::Struct(_)
455                | Canonical::List(_)
456                | Canonical::Map(_)
457                | Canonical::FixedSizeList(_)
458                | Canonical::Variant(_) => {
459                    vortex_bail!("Unsupported canonical type for min_max: {}", batch.dtype())
460                }
461            },
462        }
463    }
464
465    fn finalize(
466        &self,
467        _args: AggregateArgs<'_, Self::Options>,
468        partials: ArrayRef,
469    ) -> VortexResult<ArrayRef> {
470        Ok(partials)
471    }
472
473    fn finalize_scalar(
474        &self,
475        args: AggregateArgs<'_, Self::Options>,
476        partial: &Self::Partial,
477    ) -> VortexResult<Scalar> {
478        self.to_scalar(args, partial)
479    }
480}
481
482#[cfg(test)]
483mod tests {
484    use std::sync::Arc;
485    use std::sync::LazyLock;
486
487    use vortex_buffer::BitBuffer;
488    use vortex_buffer::buffer;
489    use vortex_error::VortexExpect;
490    use vortex_error::VortexResult;
491    use vortex_session::VortexSession;
492
493    use crate::IntoArray as _;
494    use crate::VortexSessionExecute;
495    use crate::aggregate_fn::Accumulator;
496    use crate::aggregate_fn::AggregateDTypes;
497    use crate::aggregate_fn::AggregateFnVTable;
498    use crate::aggregate_fn::DynAccumulator;
499    use crate::aggregate_fn::NumericalAggregateOpts;
500    use crate::aggregate_fn::fns::min_max::MinMax;
501    use crate::aggregate_fn::fns::min_max::MinMaxPartial;
502    use crate::aggregate_fn::fns::min_max::MinMaxResult;
503    use crate::aggregate_fn::fns::min_max::make_minmax_dtype;
504    use crate::aggregate_fn::fns::min_max::min_max;
505    use crate::arrays::BoolArray;
506    use crate::arrays::ChunkedArray;
507    use crate::arrays::ConstantArray;
508    use crate::arrays::DecimalArray;
509    use crate::arrays::FixedSizeListArray;
510    use crate::arrays::ListArray;
511    use crate::arrays::NullArray;
512    use crate::arrays::PrimitiveArray;
513    use crate::arrays::VarBinArray;
514    use crate::dtype::DType;
515    use crate::dtype::DecimalDType;
516    use crate::dtype::Nullability;
517    use crate::dtype::PType;
518    use crate::expr::stats::Precision;
519    use crate::expr::stats::Stat;
520    use crate::scalar::DecimalValue;
521    use crate::scalar::Scalar;
522    use crate::scalar::ScalarValue;
523    use crate::validity::Validity;
524
525    static SESSION: LazyLock<VortexSession> = LazyLock::new(vortex_array::array_session);
526
527    #[test]
528    fn test_prim_min_max() -> VortexResult<()> {
529        let p = PrimitiveArray::new(buffer![1, 2, 3], Validity::NonNullable).into_array();
530        let mut ctx = SESSION.create_execution_ctx();
531        assert_eq!(
532            min_max(&p, &mut ctx, NumericalAggregateOpts::default())?,
533            Some(MinMaxResult {
534                min: 1.into(),
535                max: 3.into()
536            })
537        );
538        Ok(())
539    }
540
541    #[test]
542    fn test_prim_min_max_multiple_null_runs() -> VortexResult<()> {
543        // Several disjoint valid runs separated by nulls exercise the per-run fold; the extrema
544        // (min 1, max 9) fall in different runs.
545        let p = PrimitiveArray::from_option_iter([
546            Some(5i32),
547            Some(3),
548            None,
549            None,
550            Some(9),
551            None,
552            Some(1),
553            Some(7),
554        ])
555        .into_array();
556        let mut ctx = SESSION.create_execution_ctx();
557        assert_eq!(
558            min_max(&p, &mut ctx, NumericalAggregateOpts::default())?,
559            Some(MinMaxResult {
560                min: 1.into(),
561                max: 9.into()
562            })
563        );
564        Ok(())
565    }
566
567    #[test]
568    fn test_bool_min_max() -> VortexResult<()> {
569        let mut ctx = SESSION.create_execution_ctx();
570
571        let all_true = BoolArray::new(
572            BitBuffer::from([true, true, true].as_slice()),
573            Validity::NonNullable,
574        )
575        .into_array();
576        assert_eq!(
577            min_max(&all_true, &mut ctx, NumericalAggregateOpts::default())?,
578            Some(MinMaxResult {
579                min: true.into(),
580                max: true.into()
581            })
582        );
583
584        let all_false = BoolArray::new(
585            BitBuffer::from([false, false, false].as_slice()),
586            Validity::NonNullable,
587        )
588        .into_array();
589        assert_eq!(
590            min_max(&all_false, &mut ctx, NumericalAggregateOpts::default())?,
591            Some(MinMaxResult {
592                min: false.into(),
593                max: false.into()
594            })
595        );
596
597        let mixed = BoolArray::new(
598            BitBuffer::from([false, true, false].as_slice()),
599            Validity::NonNullable,
600        )
601        .into_array();
602        assert_eq!(
603            min_max(&mixed, &mut ctx, NumericalAggregateOpts::default())?,
604            Some(MinMaxResult {
605                min: false.into(),
606                max: true.into()
607            })
608        );
609        Ok(())
610    }
611
612    #[test]
613    fn test_null_array() -> VortexResult<()> {
614        let p = NullArray::new(1).into_array();
615        let mut ctx = SESSION.create_execution_ctx();
616        assert_eq!(
617            min_max(&p, &mut ctx, NumericalAggregateOpts::default())?,
618            None
619        );
620        Ok(())
621    }
622
623    #[test]
624    fn test_prim_nan() -> VortexResult<()> {
625        let array = PrimitiveArray::new(
626            buffer![f32::NAN, -f32::NAN, -1.0, 1.0],
627            Validity::NonNullable,
628        );
629        let mut ctx = SESSION.create_execution_ctx();
630        let result = min_max(
631            &array.into_array(),
632            &mut ctx,
633            NumericalAggregateOpts::default(),
634        )?
635        .vortex_expect("should have result");
636        assert_eq!(f32::try_from(&result.min)?, -1.0);
637        assert_eq!(f32::try_from(&result.max)?, 1.0);
638        Ok(())
639    }
640
641    #[test]
642    fn test_prim_inf() -> VortexResult<()> {
643        let array = PrimitiveArray::new(
644            buffer![f32::INFINITY, f32::NEG_INFINITY, -1.0, 1.0],
645            Validity::NonNullable,
646        );
647        let mut ctx = SESSION.create_execution_ctx();
648        let result = min_max(
649            &array.into_array(),
650            &mut ctx,
651            NumericalAggregateOpts::default(),
652        )?
653        .vortex_expect("should have result");
654        assert_eq!(f32::try_from(&result.min)?, f32::NEG_INFINITY);
655        assert_eq!(f32::try_from(&result.max)?, f32::INFINITY);
656        Ok(())
657    }
658
659    #[test]
660    fn test_multi_batch() -> VortexResult<()> {
661        let mut ctx = SESSION.create_execution_ctx();
662        let dtype = DType::Primitive(PType::I32, Nullability::NonNullable);
663        let mut acc = Accumulator::try_new(MinMax, NumericalAggregateOpts::default(), dtype)?;
664
665        let batch1 = PrimitiveArray::new(buffer![10i32, 20, 5], Validity::NonNullable).into_array();
666        acc.accumulate(&batch1, &mut ctx)?;
667
668        let batch2 = PrimitiveArray::new(buffer![3i32, 25], Validity::NonNullable).into_array();
669        acc.accumulate(&batch2, &mut ctx)?;
670
671        let result = MinMaxResult::from_scalar(acc.finish()?)?.vortex_expect("should have result");
672        assert_eq!(result.min, Scalar::from(3i32));
673        assert_eq!(result.max, Scalar::from(25i32));
674        Ok(())
675    }
676
677    #[test]
678    fn test_finish_resets_state() -> VortexResult<()> {
679        let mut ctx = SESSION.create_execution_ctx();
680        let dtype = DType::Primitive(PType::I32, Nullability::NonNullable);
681        let mut acc = Accumulator::try_new(MinMax, NumericalAggregateOpts::default(), dtype)?;
682
683        let batch1 = PrimitiveArray::new(buffer![10i32, 20], Validity::NonNullable).into_array();
684        acc.accumulate(&batch1, &mut ctx)?;
685        let result1 = MinMaxResult::from_scalar(acc.finish()?)?.vortex_expect("should have result");
686        assert_eq!(result1.min, Scalar::from(10i32));
687        assert_eq!(result1.max, Scalar::from(20i32));
688
689        let batch2 = PrimitiveArray::new(buffer![3i32, 6, 9], Validity::NonNullable).into_array();
690        acc.accumulate(&batch2, &mut ctx)?;
691        let result2 = MinMaxResult::from_scalar(acc.finish()?)?.vortex_expect("should have result");
692        assert_eq!(result2.min, Scalar::from(3i32));
693        assert_eq!(result2.max, Scalar::from(9i32));
694        Ok(())
695    }
696
697    #[test]
698    fn test_state_merge() -> VortexResult<()> {
699        let dtype = DType::Primitive(PType::I32, Nullability::NonNullable);
700        let options = NumericalAggregateOpts::default();
701        let owned = AggregateDTypes::try_new(&MinMax, &options, dtype)?;
702        let args = owned.args(&options);
703        let partial_of = |min: i32, max: i32| MinMaxPartial {
704            min: Some(Scalar::from(min)),
705            max: Some(Scalar::from(max)),
706        };
707
708        let state = MinMax.merge_partials(args, partial_of(5, 15), partial_of(2, 10))?;
709
710        let result = MinMaxResult::from_scalar(MinMax.to_scalar(args, &state)?)?
711            .vortex_expect("should have result");
712        assert_eq!(result.min, Scalar::from(2i32));
713        assert_eq!(result.max, Scalar::from(15i32));
714        Ok(())
715    }
716
717    #[test]
718    fn test_constant_nan() -> VortexResult<()> {
719        let scalar = Scalar::primitive(f16::NAN, Nullability::NonNullable);
720        let array = ConstantArray::new(scalar, 2).into_array();
721        let mut ctx = SESSION.create_execution_ctx();
722        assert_eq!(
723            min_max(&array, &mut ctx, NumericalAggregateOpts::default())?,
724            None
725        );
726        Ok(())
727    }
728
729    const KEEP_NANS: NumericalAggregateOpts = NumericalAggregateOpts::include_nans();
730
731    fn assert_poisoned(result: Option<MinMaxResult>) -> VortexResult<()> {
732        let result = result.vortex_expect("should have result");
733        assert!(f64::try_from(&result.min.cast(&result.min.dtype().as_nullable())?)?.is_nan());
734        assert!(f64::try_from(&result.max.cast(&result.max.dtype().as_nullable())?)?.is_nan());
735        Ok(())
736    }
737
738    #[test]
739    fn test_prim_nan_not_skipping() -> VortexResult<()> {
740        let array = PrimitiveArray::new(
741            buffer![f32::NAN, -f32::NAN, -1.0, 1.0],
742            Validity::NonNullable,
743        )
744        .into_array();
745        let mut ctx = SESSION.create_execution_ctx();
746        assert_poisoned(min_max(&array, &mut ctx, KEEP_NANS)?)
747    }
748
749    #[test]
750    fn test_prim_no_nan_not_skipping() -> VortexResult<()> {
751        let array =
752            PrimitiveArray::new(buffer![3.0f32, -1.0, 1.0], Validity::NonNullable).into_array();
753        let mut ctx = SESSION.create_execution_ctx();
754        let result = min_max(&array, &mut ctx, KEEP_NANS)?.vortex_expect("should have result");
755        assert_eq!(f32::try_from(&result.min)?, -1.0);
756        assert_eq!(f32::try_from(&result.max)?, 3.0);
757        Ok(())
758    }
759
760    #[test]
761    fn test_constant_nan_not_skipping() -> VortexResult<()> {
762        let scalar = Scalar::primitive(f64::NAN, Nullability::NonNullable);
763        let array = ConstantArray::new(scalar, 2).into_array();
764        let mut ctx = SESSION.create_execution_ctx();
765        assert_poisoned(min_max(&array, &mut ctx, KEEP_NANS)?)
766    }
767
768    #[test]
769    fn test_not_skipping_shortcircuits_on_exact_nan_count_stat() -> VortexResult<()> {
770        // The array has no NaNs; a planted exact NaNCount stat proves the poisoning came from
771        // the stat rather than a scan.
772        let array =
773            PrimitiveArray::new(buffer![1.0f64, 2.0, 3.0], Validity::NonNullable).into_array();
774        array
775            .statistics()
776            .set(Stat::NaNCount, Precision::Exact(ScalarValue::from(2u64)));
777        let mut ctx = SESSION.create_execution_ctx();
778        assert_poisoned(min_max(&array, &mut ctx, KEEP_NANS)?)
779    }
780
781    #[test]
782    fn test_not_skipping_uses_cached_stats_when_nan_free() -> VortexResult<()> {
783        // With an exact NaNCount of zero, the planted exact Min/Max stats are usable as-is.
784        let array =
785            PrimitiveArray::new(buffer![1.0f64, 2.0, 3.0], Validity::NonNullable).into_array();
786        array
787            .statistics()
788            .set(Stat::NaNCount, Precision::Exact(ScalarValue::from(0u64)));
789        array
790            .statistics()
791            .set(Stat::Min, Precision::Exact(ScalarValue::from(-10.0f64)));
792        array
793            .statistics()
794            .set(Stat::Max, Precision::Exact(ScalarValue::from(10.0f64)));
795        let mut ctx = SESSION.create_execution_ctx();
796        let result = min_max(&array, &mut ctx, KEEP_NANS)?.vortex_expect("should have result");
797        assert_eq!(f64::try_from(&result.min)?, -10.0);
798        assert_eq!(f64::try_from(&result.max)?, 10.0);
799        Ok(())
800    }
801
802    #[test]
803    fn test_accumulator_nan_including_nullable_cached_stats() -> VortexResult<()> {
804        // A nullable float array's cached Min/Max stats are reconstructed as nullable scalars.
805        // The NaN-including accumulator shortcircuit must normalise them to the non-nullable
806        // struct field dtype before building the result scalar.
807        let mut ctx = SESSION.create_execution_ctx();
808        let array =
809            PrimitiveArray::from_option_iter([Some(1.0f64), Some(2.0), Some(3.0)]).into_array();
810        array
811            .statistics()
812            .set(Stat::NaNCount, Precision::Exact(ScalarValue::from(0u64)));
813        array
814            .statistics()
815            .set(Stat::Min, Precision::Exact(ScalarValue::from(1.0f64)));
816        array
817            .statistics()
818            .set(Stat::Max, Precision::Exact(ScalarValue::from(3.0f64)));
819
820        let mut acc = Accumulator::try_new(MinMax, KEEP_NANS, array.dtype().clone())?;
821        acc.accumulate(&array, &mut ctx)?;
822        let result = MinMaxResult::from_scalar(acc.finish()?)?.vortex_expect("should have result");
823        assert_eq!(f64::try_from(&result.min)?, 1.0);
824        assert_eq!(f64::try_from(&result.max)?, 3.0);
825        Ok(())
826    }
827
828    #[test]
829    fn test_multi_batch_nan_poisoning() -> VortexResult<()> {
830        let mut ctx = SESSION.create_execution_ctx();
831        let dtype = DType::Primitive(PType::F64, Nullability::NonNullable);
832        let mut acc = Accumulator::try_new(MinMax, KEEP_NANS, dtype)?;
833
834        let batch1 = PrimitiveArray::new(buffer![1.0f64, 2.0], Validity::NonNullable).into_array();
835        acc.accumulate(&batch1, &mut ctx)?;
836        assert!(!acc.is_saturated());
837
838        let batch2 = PrimitiveArray::new(buffer![f64::NAN], Validity::NonNullable).into_array();
839        acc.accumulate(&batch2, &mut ctx)?;
840        assert!(acc.is_saturated());
841
842        assert_poisoned(MinMaxResult::from_scalar(acc.finish()?)?)
843    }
844
845    #[test]
846    fn test_chunked() -> VortexResult<()> {
847        let chunk1 = PrimitiveArray::from_option_iter([Some(5i32), None, Some(1)]);
848        let chunk2 = PrimitiveArray::from_option_iter([Some(10i32), Some(3), None]);
849        let dtype = chunk1.dtype().clone();
850        let chunked = ChunkedArray::try_new(vec![chunk1.into_array(), chunk2.into_array()], dtype)?;
851        let mut ctx = SESSION.create_execution_ctx();
852        let result = min_max(
853            &chunked.into_array(),
854            &mut ctx,
855            NumericalAggregateOpts::default(),
856        )?
857        .vortex_expect("should have result");
858        assert_eq!(result.min, Scalar::from(1i32));
859        assert_eq!(result.max, Scalar::from(10i32));
860        Ok(())
861    }
862
863    #[test]
864    fn test_all_null() -> VortexResult<()> {
865        let p = PrimitiveArray::from_option_iter::<i32, _>([None, None, None]);
866        let mut ctx = SESSION.create_execution_ctx();
867        assert_eq!(
868            min_max(&p.into_array(), &mut ctx, NumericalAggregateOpts::default())?,
869            None
870        );
871        Ok(())
872    }
873
874    #[test]
875    fn test_varbin() -> VortexResult<()> {
876        let array = VarBinArray::from_iter(
877            vec![
878                Some("hello world"),
879                None,
880                Some("hello world this is a long string"),
881                None,
882            ],
883            DType::Utf8(Nullability::Nullable),
884        );
885        let mut ctx = SESSION.create_execution_ctx();
886        let result = min_max(
887            &array.into_array(),
888            &mut ctx,
889            NumericalAggregateOpts::default(),
890        )?
891        .vortex_expect("should have result");
892        assert_eq!(
893            result.min,
894            Scalar::utf8("hello world", Nullability::NonNullable)
895        );
896        assert_eq!(
897            result.max,
898            Scalar::utf8(
899                "hello world this is a long string",
900                Nullability::NonNullable
901            )
902        );
903        Ok(())
904    }
905
906    #[test]
907    fn test_decimal() -> VortexResult<()> {
908        let decimal = DecimalArray::new(
909            buffer![100i32, 2000i32, 200i32],
910            DecimalDType::new(4, 2),
911            Validity::from_iter([true, false, true]),
912        );
913        let mut ctx = SESSION.create_execution_ctx();
914        let result = min_max(
915            &decimal.into_array(),
916            &mut ctx,
917            NumericalAggregateOpts::default(),
918        )?
919        .vortex_expect("should have result");
920
921        let non_nullable_dtype = DType::Decimal(DecimalDType::new(4, 2), Nullability::NonNullable);
922        let expected_min = Scalar::try_new(
923            non_nullable_dtype.clone(),
924            Some(ScalarValue::from(DecimalValue::from(100i32))),
925        )?;
926        let expected_max = Scalar::try_new(
927            non_nullable_dtype,
928            Some(ScalarValue::from(DecimalValue::from(200i32))),
929        )?;
930        assert_eq!(result.min, expected_min);
931        assert_eq!(result.max, expected_max);
932        Ok(())
933    }
934
935    #[test]
936    fn list_and_fixed_size_list_return_dtype() {
937        let element_dtype = DType::Primitive(PType::I32, Nullability::Nullable);
938        let list_dtype = DType::List(Arc::new(element_dtype.clone()), Nullability::Nullable);
939        let fixed_size_list_dtype =
940            DType::FixedSizeList(Arc::new(element_dtype), 1, Nullability::Nullable);
941
942        assert_eq!(
943            MinMax.return_dtype(&NumericalAggregateOpts::default(), &list_dtype),
944            Some(make_minmax_dtype(&list_dtype))
945        );
946        assert_eq!(
947            MinMax.return_dtype(&NumericalAggregateOpts::default(), &fixed_size_list_dtype),
948            Some(make_minmax_dtype(&fixed_size_list_dtype))
949        );
950    }
951
952    #[test]
953    fn list_and_fixed_size_list_min_max_returns_none() -> VortexResult<()> {
954        let mut ctx = SESSION.create_execution_ctx();
955
956        let list_array = ListArray::try_new(
957            buffer![1i32, 2, 3].into_array(),
958            buffer![0u32, 2, 3].into_array(),
959            Validity::NonNullable,
960        )?
961        .into_array();
962        assert_eq!(
963            min_max(&list_array, &mut ctx, NumericalAggregateOpts::default())?,
964            None
965        );
966
967        let fixed_size_list_array = FixedSizeListArray::try_new(
968            buffer![1i32, 2, 3, 4].into_array(),
969            2,
970            Validity::NonNullable,
971            2,
972        )?
973        .into_array();
974        assert_eq!(
975            min_max(
976                &fixed_size_list_array,
977                &mut ctx,
978                NumericalAggregateOpts::default()
979            )?,
980            None
981        );
982
983        Ok(())
984    }
985
986    use crate::dtype::half::f16;
987
988    #[test]
989    fn test_bool_with_nulls() -> VortexResult<()> {
990        let mut ctx = SESSION.create_execution_ctx();
991
992        let result = min_max(
993            &BoolArray::from_iter(vec![Some(true), Some(true), None, None]).into_array(),
994            &mut ctx,
995            NumericalAggregateOpts::default(),
996        )?;
997        assert_eq!(
998            result,
999            Some(MinMaxResult {
1000                min: Scalar::bool(true, Nullability::NonNullable),
1001                max: Scalar::bool(true, Nullability::NonNullable),
1002            })
1003        );
1004
1005        let result = min_max(
1006            &BoolArray::from_iter(vec![None, Some(true), Some(true)]).into_array(),
1007            &mut ctx,
1008            NumericalAggregateOpts::default(),
1009        )?;
1010        assert_eq!(
1011            result,
1012            Some(MinMaxResult {
1013                min: Scalar::bool(true, Nullability::NonNullable),
1014                max: Scalar::bool(true, Nullability::NonNullable),
1015            })
1016        );
1017
1018        let result = min_max(
1019            &BoolArray::from_iter(vec![None, Some(true), Some(true), None]).into_array(),
1020            &mut ctx,
1021            NumericalAggregateOpts::default(),
1022        )?;
1023        assert_eq!(
1024            result,
1025            Some(MinMaxResult {
1026                min: Scalar::bool(true, Nullability::NonNullable),
1027                max: Scalar::bool(true, Nullability::NonNullable),
1028            })
1029        );
1030
1031        let result = min_max(
1032            &BoolArray::from_iter(vec![Some(false), Some(false), None, None]).into_array(),
1033            &mut ctx,
1034            NumericalAggregateOpts::default(),
1035        )?;
1036        assert_eq!(
1037            result,
1038            Some(MinMaxResult {
1039                min: Scalar::bool(false, Nullability::NonNullable),
1040                max: Scalar::bool(false, Nullability::NonNullable),
1041            })
1042        );
1043        Ok(())
1044    }
1045
1046    /// Regression test for <https://github.com/vortex-data/vortex/issues/7074>.
1047    ///
1048    /// A chunked all-true bool array with an empty first chunk returned min=false because
1049    /// `accumulate_bool` on the empty chunk incorrectly merged min=false,max=false into the
1050    /// partial state.
1051    #[test]
1052    fn test_bool_chunked_with_empty_chunk() -> VortexResult<()> {
1053        let mut ctx = SESSION.create_execution_ctx();
1054
1055        let empty = BoolArray::new(BitBuffer::from([].as_slice()), Validity::NonNullable);
1056        let chunk1 = BoolArray::new(
1057            BitBuffer::from([true, true].as_slice()),
1058            Validity::NonNullable,
1059        );
1060        let chunk2 = BoolArray::new(
1061            BitBuffer::from([true, true, true].as_slice()),
1062            Validity::NonNullable,
1063        );
1064        let chunked = ChunkedArray::try_new(
1065            vec![empty.into_array(), chunk1.into_array(), chunk2.into_array()],
1066            DType::Bool(Nullability::NonNullable),
1067        )?;
1068
1069        let result = min_max(
1070            &chunked.into_array(),
1071            &mut ctx,
1072            NumericalAggregateOpts::default(),
1073        )?;
1074        assert_eq!(
1075            result,
1076            Some(MinMaxResult {
1077                min: Scalar::bool(true, Nullability::NonNullable),
1078                max: Scalar::bool(true, Nullability::NonNullable),
1079            })
1080        );
1081        Ok(())
1082    }
1083
1084    /// Regression test for <https://github.com/vortex-data/vortex/issues/8145>.
1085    ///
1086    /// A chunked array whose first chunk is an *empty* constant array — as produced by
1087    /// `fill_null` on an empty all-null chunk — returned `max = u32::MAX` because
1088    /// `ChunkedArrayAggregate` accumulated the empty chunk, folding its fill scalar into the
1089    /// running min/max. Empty chunks are now skipped during chunked aggregation.
1090    #[test]
1091    fn test_chunked_with_empty_constant_chunk() -> VortexResult<()> {
1092        let mut ctx = SESSION.create_execution_ctx();
1093
1094        let empty = ConstantArray::new(Scalar::primitive(u32::MAX, Nullability::NonNullable), 0)
1095            .into_array();
1096        let chunk1 = PrimitiveArray::new(buffer![7631471u32], Validity::NonNullable).into_array();
1097        let chunk2 = PrimitiveArray::new(buffer![0u32], Validity::NonNullable).into_array();
1098        let chunked = ChunkedArray::try_new(
1099            vec![empty, chunk1, chunk2],
1100            DType::Primitive(PType::U32, Nullability::NonNullable),
1101        )?;
1102
1103        assert_eq!(
1104            min_max(
1105                &chunked.into_array(),
1106                &mut ctx,
1107                NumericalAggregateOpts::default()
1108            )?,
1109            Some(MinMaxResult {
1110                min: Scalar::primitive(0u32, Nullability::NonNullable),
1111                max: Scalar::primitive(7631471u32, Nullability::NonNullable),
1112            })
1113        );
1114        Ok(())
1115    }
1116
1117    #[test]
1118    fn test_varbin_all_nulls() -> VortexResult<()> {
1119        let array = VarBinArray::from_iter(
1120            vec![Option::<&str>::None, None, None],
1121            DType::Utf8(Nullability::Nullable),
1122        );
1123        let mut ctx = SESSION.create_execution_ctx();
1124        assert_eq!(
1125            min_max(
1126                &array.into_array(),
1127                &mut ctx,
1128                NumericalAggregateOpts::default()
1129            )?,
1130            None
1131        );
1132        Ok(())
1133    }
1134}
vortex_array/aggregate_fn/fns/min_max/mod.rs

vortex_array/aggregate_fn/fns/min_max/
mod.rs
