Migrate statistics consumers and file summaries to aggregates

encodings/datetime-parts/src/compute/compare.rs

@@ -19,6 +19,8 @@ use vortex_error::VortexResult;
 
 use crate::array::DateTimeParts;
 use crate::array::DateTimePartsArraySlotsExt;
+use crate::compute::MAX_SKIP_NANS;
+use crate::compute::MIN_SKIP_NANS;
 use crate::timestamp;
 
 impl CompareKernel for DateTimeParts {
@@ -74,7 +76,12 @@ fn compare_eq(
     ctx: &mut ExecutionCtx,
 ) -> VortexResult<Option<ArrayRef>> {
     let mut comparison = compare_dtp(lhs.days(), ts_parts.days, CompareOperator::Eq, nullability)?;
-    if comparison.statistics().compute_max::<bool>(ctx) == Some(false) {
+    if matches!(
+        comparison
+            .aggregations()
+            .compute_as::<bool>(&MAX_SKIP_NANS, ctx),
+        Ok(false),
+    ) {
         // All values are different.
         return Ok(Some(comparison));
     }
@@ -87,7 +94,12 @@ fn compare_eq(
     )?
     .binary(comparison, Operator::And)?;
 
-    if comparison.statistics().compute_max::<bool>(ctx) == Some(false) {
+    if matches!(
+        comparison
+            .aggregations()
+            .compute_as::<bool>(&MAX_SKIP_NANS, ctx),
+        Ok(false),
+    ) {
         // All values are different.
         return Ok(Some(comparison));
     }
@@ -115,7 +127,12 @@ fn compare_ne(
         CompareOperator::NotEq,
         nullability,
     )?;
-    if comparison.statistics().compute_min::<bool>(ctx) == Some(true) {
+    if matches!(
+        comparison
+            .aggregations()
+            .compute_as::<bool>(&MIN_SKIP_NANS, ctx),
+        Ok(true),
+    ) {
         // All values are different.
         return Ok(Some(comparison));
     }
@@ -128,7 +145,12 @@ fn compare_ne(
     )?
     .binary(comparison, Operator::Or)?;
 
-    if comparison.statistics().compute_min::<bool>(ctx) == Some(true) {
+    if matches!(
+        comparison
+            .aggregations()
+            .compute_as::<bool>(&MIN_SKIP_NANS, ctx),
+        Ok(true),
+    ) {
         // All values are different.
         return Ok(Some(comparison));
     }
@@ -151,7 +173,12 @@ fn compare_lt(
     ctx: &mut ExecutionCtx,
 ) -> VortexResult<Option<ArrayRef>> {
     let days_lt = compare_dtp(lhs.days(), ts_parts.days, CompareOperator::Lt, nullability)?;
-    if days_lt.statistics().compute_min::<bool>(ctx) == Some(true) {
+    if matches!(
+        days_lt
+            .aggregations()
+            .compute_as::<bool>(&MIN_SKIP_NANS, ctx),
+        Ok(true),
+    ) {
         // All values on the lhs are smaller.
         return Ok(Some(days_lt));
     }
@@ -166,7 +193,12 @@ fn compare_gt(
     ctx: &mut ExecutionCtx,
 ) -> VortexResult<Option<ArrayRef>> {
     let days_gt = compare_dtp(lhs.days(), ts_parts.days, CompareOperator::Gt, nullability)?;
-    if days_gt.statistics().compute_min::<bool>(ctx) == Some(true) {
+    if matches!(
+        days_gt
+            .aggregations()
+            .compute_as::<bool>(&MIN_SKIP_NANS, ctx),
+        Ok(true),
+    ) {
         // All values on the lhs are larger.
         return Ok(Some(days_gt));
     }

encodings/datetime-parts/src/compute/mod.rs

@@ -1,6 +1,14 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright the Vortex contributors
 
+use std::sync::LazyLock;
+
+use vortex_array::aggregate_fn::AggregateFnRef;
+use vortex_array::aggregate_fn::AggregateFnVTableExt;
+use vortex_array::aggregate_fn::NumericalAggregateOpts;
+use vortex_array::aggregate_fn::fns::max::Max;
+use vortex_array::aggregate_fn::fns::min::Min;
+
 mod cast;
 mod compare;
 mod filter;
@@ -11,6 +19,11 @@ pub(super) mod rules;
 mod slice;
 mod take;
 
+static MIN_SKIP_NANS: LazyLock<AggregateFnRef> =
+    LazyLock::new(|| Min.bind(NumericalAggregateOpts::skip_nans()));
+static MAX_SKIP_NANS: LazyLock<AggregateFnRef> =
+    LazyLock::new(|| Max.bind(NumericalAggregateOpts::skip_nans()));
+
 #[cfg(test)]
 mod tests {
     use rstest::rstest;

encodings/datetime-parts/src/compute/take.rs

@@ -9,14 +9,13 @@ use vortex_array::arrays::PrimitiveArray;
 use vortex_array::arrays::dict::TakeExecute;
 use vortex_array::builtins::ArrayBuiltins;
 use vortex_array::dtype::Nullability;
-use vortex_array::expr::stats::Stat;
-use vortex_array::expr::stats::StatsProvider;
 use vortex_array::scalar::Scalar;
 use vortex_error::VortexResult;
 use vortex_error::vortex_panic;
 
 use crate::DateTimeParts;
 use crate::array::DateTimePartsArraySlotsExt;
+use crate::compute::MIN_SKIP_NANS;
 fn take_datetime_parts(
     array: ArrayView<DateTimeParts>,
     indices: &ArrayRef,
@@ -63,18 +62,20 @@ fn take_datetime_parts(
 
     let seconds_fill = array
         .seconds()
-        .statistics()
-        .get(Stat::Min)
+        .aggregations()
+        .get_result(&MIN_SKIP_NANS)
         .into_inner()
+        .filter(|s| !s.is_null())
         .unwrap_or_else(|| Scalar::primitive(0i64, Nullability::NonNullable))
         .cast(array.seconds().dtype())?;
     let taken_seconds = taken_seconds.fill_null(seconds_fill)?;
 
     let subseconds_fill = array
         .subseconds()
-        .statistics()
-        .get(Stat::Min)
+        .aggregations()
+        .get_result(&MIN_SKIP_NANS)
         .into_inner()
+        .filter(|s| !s.is_null())
         .unwrap_or_else(|| Scalar::primitive(0i64, Nullability::NonNullable))
         .cast(array.subseconds().dtype())?;
     let taken_subseconds = taken_subseconds.fill_null(subseconds_fill)?;

encodings/fastlanes/src/bitpacking/array/bitpack_compress.rs

@@ -1,12 +1,18 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright the Vortex contributors
 
+use std::sync::LazyLock;
+
 use fastlanes::BitPacking;
 use itertools::Itertools;
 use num_traits::PrimInt;
 use vortex_array::ArrayView;
 use vortex_array::ExecutionCtx;
 use vortex_array::IntoArray;
+use vortex_array::aggregate_fn::AggregateFnRef;
+use vortex_array::aggregate_fn::AggregateFnVTableExt;
+use vortex_array::aggregate_fn::NumericalAggregateOpts;
+use vortex_array::aggregate_fn::fns::min::Min;
 use vortex_array::arrays::Primitive;
 use vortex_array::arrays::PrimitiveArray;
 use vortex_array::arrays::primitive::PrimitiveArrayExt;
@@ -31,6 +37,9 @@ use crate::BitPacked;
 use crate::BitPackedArray;
 use crate::bitpack_decompress;
 
+static MIN_SKIP_NANS: LazyLock<AggregateFnRef> =
+    LazyLock::new(|| Min.bind(NumericalAggregateOpts::skip_nans()));
+
 pub fn bitpack_to_best_bit_width(
     array: &PrimitiveArray,
     ctx: &mut ExecutionCtx,
@@ -55,7 +64,11 @@ pub fn bitpack_encode(
     // Check array contains no negative values.
     if array.ptype().is_signed_int() {
         let has_negative_values = match_each_integer_ptype!(array.ptype(), |P| {
-            array.statistics().compute_min::<P>(ctx).unwrap_or_default() < 0
+            array
+                .aggregations()
+                .compute_as::<P>(&MIN_SKIP_NANS, ctx)
+                .unwrap_or_default()
+                < 0
         });
         if has_negative_values {
             vortex_bail!(InvalidArgument: "cannot bitpack_encode array containing negative integers")
@@ -88,7 +101,7 @@ pub fn bitpack_encode(
         array.len(),
         0,
     )?;
-    bitpacked.statistics().inherit_from(array.statistics());
+    bitpacked.aggregations().inherit_from(array.aggregations());
     Ok(bitpacked)
 }
 
@@ -107,7 +120,6 @@ pub unsafe fn bitpack_encode_unchecked(
     // SAFETY: non-negativity of input checked by caller.
     let packed = unsafe { bitpack_unchecked(&array, bit_width) };
 
-    let arr_ref = array.clone().into_array();
     let bitpacked = BitPacked::try_new(
         BufferHandle::new_host(packed),
         array.ptype(),
@@ -118,7 +130,7 @@ pub unsafe fn bitpack_encode_unchecked(
         0,
     )
     .vortex_expect("bitpacked array construction should succeed");
-    bitpacked.statistics().inherit_from(arr_ref.statistics());
+    bitpacked.aggregations().inherit_from(array.aggregations());
     Ok(bitpacked)
 }
 
@@ -427,16 +439,20 @@ pub mod test_harness {
 }
 
 #[cfg(test)]
-mod test {
+mod tests {
     use std::sync::LazyLock;
 
     use rand::SeedableRng;
     use rand::rngs::StdRng;
     use vortex_array::VortexSessionExecute;
+    use vortex_array::aggregate_fn::EmptyOptions;
+    use vortex_array::aggregate_fn::fns::count::Count;
+    use vortex_array::aggregate_fn::fns::uncompressed_size_in_bytes::UncompressedSizeInBytes;
     use vortex_array::arrays::ChunkedArray;
     use vortex_array::assert_arrays_eq;
     use vortex_array::builders::ArrayBuilder;
     use vortex_array::builders::PrimitiveBuilder;
+    use vortex_array::expr::stats::Precision;
     use vortex_buffer::Buffer;
     use vortex_error::VortexError;
     use vortex_error::vortex_err;
@@ -453,6 +469,24 @@ mod test {
         session
     });
 
+    #[test]
+    fn bitpack_preserves_logical_results_and_discards_physical_size() -> VortexResult<()> {
+        let source = PrimitiveArray::from_iter([1u32, 2, 3]);
+        let mut ctx = SESSION.create_execution_ctx();
+        let count = Count.bind(NumericalAggregateOpts::skip_nans());
+        let size = UncompressedSizeInBytes.bind(EmptyOptions);
+        source.aggregations().compute_result(&count, &mut ctx)?;
+        source.aggregations().compute_result(&size, &mut ctx)?;
+
+        let packed = bitpack_encode(&source, 2, None, &mut ctx)?;
+        assert_eq!(
+            packed.aggregations().get_result_as::<u64>(&count)?,
+            Precision::Exact(3)
+        );
+        assert_eq!(packed.aggregations().get_result(&size), Precision::Absent);
+        Ok(())
+    }
+
     #[test]
     fn test_best_bit_width() {
         // 10 1-bit values, 20 2-bit, etc.

encodings/fastlanes/src/for/array/for_compress.rs

@@ -2,39 +2,46 @@
 // SPDX-FileCopyrightText: Copyright the Vortex contributors
 
 use std::mem::MaybeUninit;
+use std::sync::LazyLock;
 
 use itertools::Itertools;
 use num_traits::AsPrimitive;
 use num_traits::PrimInt;
 use num_traits::WrappingSub;
 use vortex_array::ExecutionCtx;
 use vortex_array::IntoArray;
+use vortex_array::aggregate_fn::AggregateFnRef;
+use vortex_array::aggregate_fn::AggregateFnVTableExt;
+use vortex_array::aggregate_fn::NumericalAggregateOpts;
+use vortex_array::aggregate_fn::fns::min::Min;
 use vortex_array::arrays::ConstantArray;
 use vortex_array::arrays::PrimitiveArray;
 use vortex_array::dtype::NativePType;
-use vortex_array::expr::stats::Stat;
 use vortex_array::match_each_integer_ptype;
 use vortex_array::scalar::Scalar;
 use vortex_array::validity::Validity;
 use vortex_buffer::BitBuffer;
 use vortex_buffer::Buffer;
 use vortex_buffer::BufferMut;
 use vortex_error::VortexResult;
-use vortex_error::vortex_err;
+use vortex_error::vortex_ensure;
 use vortex_mask::AllOr;
 
 use crate::FL_CHUNK_SIZE;
 use crate::FoR;
 use crate::FoRArray;
 use crate::FoRData;
 
+static MIN_SKIP_NANS: LazyLock<AggregateFnRef> =
+    LazyLock::new(|| Min.bind(NumericalAggregateOpts::skip_nans()));
+
 impl FoRData {
     pub fn encode(array: PrimitiveArray, ctx: &mut ExecutionCtx) -> VortexResult<FoRArray> {
         let array_ref = array.clone().into_array();
         let min = array_ref
-            .statistics()
-            .compute_stat(Stat::Min, ctx)?
-            .ok_or_else(|| vortex_err!("Min stat not found"))?;
+            .aggregations()
+            .compute_result(&MIN_SKIP_NANS, ctx)?;
+        vortex_ensure!(!min.is_null(), "Min stat not found");
 
         let encoded = match_each_integer_ptype!(array.ptype(), |T| {
             encode_primitive::<T>(array, T::try_from(&min)?, ctx)?.into_array()
@@ -254,7 +261,7 @@ fn select<T: PrimInt>(mask: T, a: T, b: T) -> T {
 }
 
 #[cfg(test)]
-mod test {
+mod tests {
     use std::sync::LazyLock;
 
     use itertools::Itertools;
@@ -263,7 +270,6 @@ mod test {
     use vortex_array::arrays::primitive::PrimitiveArrayExt;
     use vortex_array::assert_arrays_eq;
     use vortex_array::dtype::PType;
-    use vortex_array::expr::stats::StatsProvider;
     use vortex_array::scalar::Scalar;
     use vortex_array::validity::Validity;
     use vortex_buffer::Buffer;
@@ -313,7 +319,14 @@ mod test {
     fn test_zeros() {
         let mut ctx = SESSION.create_execution_ctx();
         let array = PrimitiveArray::new(buffer![0i32; 100], Validity::NonNullable);
-        assert_eq!(array.statistics().len(), 0);
+        assert!(
+            array
+                .aggregations()
+                .snapshot_results()
+                .iter()
+                .next()
+                .is_none()
+        );
 
         let dtype = array.dtype().clone();
         let compressed = FoRData::encode(array, &mut ctx).unwrap();
@@ -369,4 +382,19 @@ mod test {
         assert_arrays_eq!(decompressed, array, &mut ctx);
         Ok(())
     }
+
+    #[test]
+    fn encode_without_valid_values_keeps_missing_min_error() -> VortexResult<()> {
+        let mut ctx = SESSION.create_execution_ctx();
+        let arrays = [
+            PrimitiveArray::new(Buffer::<i32>::empty(), Validity::NonNullable),
+            PrimitiveArray::from_option_iter([None::<i32>, None]),
+        ];
+        for array in arrays {
+            let error =
+                FoRData::encode(array, &mut ctx).expect_err("no valid values provide a minimum");
+            assert!(error.to_string().contains("Min stat not found"));
+        }
+        Ok(())
+    }
 }

encodings/pco/src/tests.rs

@@ -197,6 +197,7 @@ fn test_serde() -> VortexResult<()> {
             &SerializeOptions {
                 offset: 0,
                 include_padding: true,
+                ..Default::default()
             },
         )?
         .into_iter()