IsConstant Aggregate Function

encodings/datetime-parts/public-api.lock

@@ -22,10 +22,6 @@ impl vortex_array::arrays::slice::SliceReduce for vortex_datetime_parts::DateTim
 
 pub fn vortex_datetime_parts::DateTimeParts::slice(array: &Self::Array, range: core::ops::range::Range<usize>) -> vortex_error::VortexResult<core::option::Option<vortex_array::array::ArrayRef>>
 
-impl vortex_array::compute::is_constant::IsConstantKernel for vortex_datetime_parts::DateTimeParts
-
-pub fn vortex_datetime_parts::DateTimeParts::is_constant(&self, array: &vortex_datetime_parts::DateTimePartsArray, opts: &vortex_array::compute::is_constant::IsConstantOpts) -> vortex_error::VortexResult<core::option::Option<bool>>
-
 impl vortex_array::scalar_fn::fns::binary::compare::CompareKernel for vortex_datetime_parts::DateTimeParts
 
 pub fn vortex_datetime_parts::DateTimeParts::compare(lhs: &vortex_datetime_parts::DateTimePartsArray, rhs: &vortex_array::array::ArrayRef, operator: vortex_array::scalar_fn::fns::operators::CompareOperator, _ctx: &mut vortex_array::executor::ExecutionCtx) -> vortex_error::VortexResult<core::option::Option<vortex_array::array::ArrayRef>>
@@ -220,4 +216,6 @@ pub vortex_datetime_parts::TemporalParts::seconds: vortex_array::array::ArrayRef
 
 pub vortex_datetime_parts::TemporalParts::subseconds: vortex_array::array::ArrayRef
 
+pub fn vortex_datetime_parts::initialize(session: &mut vortex_session::VortexSession)
+
 pub fn vortex_datetime_parts::split_temporal(array: vortex_array::arrays::datetime::TemporalArray) -> vortex_error::VortexResult<vortex_datetime_parts::TemporalParts>

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

@@ -1,45 +1,41 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright the Vortex contributors
 
-use vortex_array::compute::IsConstantKernel;
-use vortex_array::compute::IsConstantKernelAdapter;
-use vortex_array::compute::IsConstantOpts;
-use vortex_array::compute::is_constant_opts;
-use vortex_array::register_kernel;
+use vortex_array::ArrayRef;
+use vortex_array::ExecutionCtx;
+use vortex_array::aggregate_fn::AggregateFnRef;
+use vortex_array::aggregate_fn::fns::is_constant::IsConstant;
+use vortex_array::aggregate_fn::fns::is_constant::is_constant;
+use vortex_array::aggregate_fn::kernels::DynAggregateKernel;
+use vortex_array::scalar::Scalar;
 use vortex_error::VortexResult;
 
 use crate::DateTimeParts;
-use crate::DateTimePartsArray;
 
-impl IsConstantKernel for DateTimeParts {
-    fn is_constant(
-        &self,
-        array: &DateTimePartsArray,
-        opts: &IsConstantOpts,
-    ) -> VortexResult<Option<bool>> {
-        let Some(days) = is_constant_opts(array.days(), opts)? else {
-            return Ok(None);
-        };
-        if !days {
-            return Ok(Some(false));
-        }
+/// DateTimeParts-specific is_constant kernel.
+///
+/// Checks each component (days, seconds, subseconds) individually.
+#[derive(Debug)]
+pub(crate) struct DateTimePartsIsConstantKernel;
 
-        let Some(seconds) = is_constant_opts(array.seconds(), opts)? else {
+impl DynAggregateKernel for DateTimePartsIsConstantKernel {
+    fn aggregate(
+        &self,
+        aggregate_fn: &AggregateFnRef,
+        batch: &ArrayRef,
+        ctx: &mut ExecutionCtx,
+    ) -> VortexResult<Option<Scalar>> {
+        if !aggregate_fn.is::<IsConstant>() {
             return Ok(None);
-        };
-        if !seconds {
-            return Ok(Some(false));
         }
 
-        let Some(subseconds) = is_constant_opts(array.subseconds(), opts)? else {
+        let Some(array) = batch.as_opt::<DateTimeParts>() else {
             return Ok(None);
         };
-        if !subseconds {
-            return Ok(Some(false));
-        }
 
-        Ok(Some(true))
+        let result = is_constant(array.days(), ctx)?
+            && is_constant(array.seconds(), ctx)?
+            && is_constant(array.subseconds(), ctx)?;
+        Ok(Some(IsConstant::make_partial(batch, result)?))
     }
 }
-
-register_kernel!(IsConstantKernelAdapter(DateTimeParts).lift());

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

@@ -4,7 +4,7 @@
 mod cast;
 mod compare;
 mod filter;
-mod is_constant;
+pub(crate) mod is_constant;
 pub(crate) mod kernel;
 mod mask;
 pub(super) mod rules;

encodings/datetime-parts/src/lib.rs

@@ -11,6 +11,23 @@ mod compute;
 mod ops;
 mod timestamp;
 
+use vortex_array::aggregate_fn::AggregateFnVTable;
+use vortex_array::aggregate_fn::fns::is_constant::IsConstant;
+use vortex_array::aggregate_fn::session::AggregateFnSessionExt;
+use vortex_array::session::ArraySessionExt;
+use vortex_session::VortexSession;
+
+/// Initialize datetime-parts encoding in the given session.
+pub fn initialize(session: &mut VortexSession) {
+    session.arrays().register(DateTimeParts::ID, DateTimeParts);
+
+    session.aggregate_fns().register_aggregate_kernel(
+        DateTimeParts::ID,
+        Some(IsConstant.id()),
+        &compute::is_constant::DateTimePartsIsConstantKernel,
+    );
+}
+
 #[cfg(test)]
 mod test {
     use vortex_array::ProstMetadata;

encodings/decimal-byte-parts/public-api.lock

@@ -22,10 +22,6 @@ impl vortex_array::arrays::slice::SliceReduce for vortex_decimal_byte_parts::Dec
 
 pub fn vortex_decimal_byte_parts::DecimalByteParts::slice(array: &vortex_decimal_byte_parts::DecimalBytePartsArray, range: core::ops::range::Range<usize>) -> vortex_error::VortexResult<core::option::Option<vortex_array::array::ArrayRef>>
 
-impl vortex_array::compute::is_constant::IsConstantKernel for vortex_decimal_byte_parts::DecimalByteParts
-
-pub fn vortex_decimal_byte_parts::DecimalByteParts::is_constant(&self, array: &vortex_decimal_byte_parts::DecimalBytePartsArray, opts: &vortex_array::compute::is_constant::IsConstantOpts) -> vortex_error::VortexResult<core::option::Option<bool>>
-
 impl vortex_array::scalar_fn::fns::binary::compare::CompareKernel for vortex_decimal_byte_parts::DecimalByteParts
 
 pub fn vortex_decimal_byte_parts::DecimalByteParts::compare(lhs: &Self::Array, rhs: &vortex_array::array::ArrayRef, operator: vortex_array::scalar_fn::fns::operators::CompareOperator, _ctx: &mut vortex_array::executor::ExecutionCtx) -> vortex_error::VortexResult<core::option::Option<vortex_array::array::ArrayRef>>
@@ -167,3 +163,5 @@ impl prost::message::Message for vortex_decimal_byte_parts::DecimalBytesPartsMet
 pub fn vortex_decimal_byte_parts::DecimalBytesPartsMetadata::clear(&mut self)
 
 pub fn vortex_decimal_byte_parts::DecimalBytesPartsMetadata::encoded_len(&self) -> usize
+
+pub fn vortex_decimal_byte_parts::initialize(session: &mut vortex_session::VortexSession)

encodings/decimal-byte-parts/src/decimal_byte_parts/compute/is_constant.rs

@@ -1,24 +1,39 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright the Vortex contributors
 
-use vortex_array::compute::IsConstantKernel;
-use vortex_array::compute::IsConstantKernelAdapter;
-use vortex_array::compute::IsConstantOpts;
-use vortex_array::compute::is_constant_opts;
-use vortex_array::register_kernel;
+use vortex_array::ArrayRef;
+use vortex_array::ExecutionCtx;
+use vortex_array::aggregate_fn::AggregateFnRef;
+use vortex_array::aggregate_fn::fns::is_constant::IsConstant;
+use vortex_array::aggregate_fn::fns::is_constant::is_constant;
+use vortex_array::aggregate_fn::kernels::DynAggregateKernel;
+use vortex_array::scalar::Scalar;
 use vortex_error::VortexResult;
 
 use crate::DecimalByteParts;
-use crate::DecimalBytePartsArray;
 
-impl IsConstantKernel for DecimalByteParts {
-    fn is_constant(
+/// DecimalByteParts-specific is_constant kernel.
+///
+/// Delegates to checking if the MSP (most significant part) is constant.
+#[derive(Debug)]
+pub(crate) struct DecimalBytePartsIsConstantKernel;
+
+impl DynAggregateKernel for DecimalBytePartsIsConstantKernel {
+    fn aggregate(
         &self,
-        array: &DecimalBytePartsArray,
-        opts: &IsConstantOpts,
-    ) -> VortexResult<Option<bool>> {
-        is_constant_opts(&array.msp, opts)
+        aggregate_fn: &AggregateFnRef,
+        batch: &ArrayRef,
+        ctx: &mut ExecutionCtx,
+    ) -> VortexResult<Option<Scalar>> {
+        if !aggregate_fn.is::<IsConstant>() {
+            return Ok(None);
+        }
+
+        let Some(array) = batch.as_opt::<DecimalByteParts>() else {
+            return Ok(None);
+        };
+
+        let result = is_constant(array.msp(), ctx)?;
+        Ok(Some(IsConstant::make_partial(batch, result)?))
     }
 }
-
-register_kernel!(IsConstantKernelAdapter(DecimalByteParts).lift());

encodings/decimal-byte-parts/src/decimal_byte_parts/compute/mod.rs

@@ -4,7 +4,7 @@
 mod cast;
 mod compare;
 mod filter;
-mod is_constant;
+pub(crate) mod is_constant;
 pub(crate) mod kernel;
 mod mask;
 mod take;

encodings/decimal-byte-parts/src/decimal_byte_parts/mod.rs

@@ -1,7 +1,7 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright the Vortex contributors
 
-mod compute;
+pub(crate) mod compute;
 mod rules;
 mod slice;
 

encodings/decimal-byte-parts/src/lib.rs

@@ -3,6 +3,7 @@
 
 mod decimal_byte_parts;
 
+use decimal_byte_parts::compute::is_constant::DecimalBytePartsIsConstantKernel;
 /// This encoding allow compression of decimals using integer compression schemes.
 /// Decimals can be compressed by narrowing the signed decimal value into the smallest signed value,
 /// then integer compression if that is a value `ptype`, otherwise the decimal can be split into
@@ -12,3 +13,21 @@ mod decimal_byte_parts;
 /// an i128 decimal could be converted into a [i64, u64] with further narrowing applied to either
 /// value.
 pub use decimal_byte_parts::*;
+use vortex_array::aggregate_fn::AggregateFnVTable;
+use vortex_array::aggregate_fn::fns::is_constant::IsConstant;
+use vortex_array::aggregate_fn::session::AggregateFnSessionExt;
+use vortex_array::session::ArraySessionExt;
+use vortex_session::VortexSession;
+
+/// Initialize decimal-byte-parts encoding in the given session.
+pub fn initialize(session: &mut VortexSession) {
+    session
+        .arrays()
+        .register(DecimalByteParts::ID, DecimalByteParts);
+
+    session.aggregate_fns().register_aggregate_kernel(
+        DecimalByteParts::ID,
+        Some(IsConstant.id()),
+        &DecimalBytePartsIsConstantKernel,
+    );
+}

encodings/fastlanes/public-api.lock

@@ -134,10 +134,6 @@ impl vortex_array::arrays::slice::SliceKernel for vortex_fastlanes::BitPacked
 
 pub fn vortex_fastlanes::BitPacked::slice(array: &vortex_fastlanes::BitPackedArray, range: core::ops::range::Range<usize>, _ctx: &mut vortex_array::executor::ExecutionCtx) -> vortex_error::VortexResult<core::option::Option<vortex_array::array::ArrayRef>>
 
-impl vortex_array::compute::is_constant::IsConstantKernel for vortex_fastlanes::BitPacked
-
-pub fn vortex_fastlanes::BitPacked::is_constant(&self, array: &vortex_fastlanes::BitPackedArray, opts: &vortex_array::compute::is_constant::IsConstantOpts) -> vortex_error::VortexResult<core::option::Option<bool>>
-
 impl vortex_array::scalar_fn::fns::cast::kernel::CastReduce for vortex_fastlanes::BitPacked
 
 pub fn vortex_fastlanes::BitPacked::cast(array: &vortex_fastlanes::BitPackedArray, dtype: &vortex_array::dtype::DType) -> vortex_error::VortexResult<core::option::Option<vortex_array::array::ArrayRef>>
@@ -424,10 +420,6 @@ impl vortex_array::arrays::slice::SliceReduce for vortex_fastlanes::FoR
 
 pub fn vortex_fastlanes::FoR::slice(array: &Self::Array, range: core::ops::range::Range<usize>) -> vortex_error::VortexResult<core::option::Option<vortex_array::array::ArrayRef>>
 
-impl vortex_array::compute::is_constant::IsConstantKernel for vortex_fastlanes::FoR
-
-pub fn vortex_fastlanes::FoR::is_constant(&self, array: &vortex_fastlanes::FoRArray, opts: &vortex_array::compute::is_constant::IsConstantOpts) -> vortex_error::VortexResult<core::option::Option<bool>>
-
 impl vortex_array::compute::is_sorted::IsSortedKernel for vortex_fastlanes::FoR
 
 pub fn vortex_fastlanes::FoR::is_sorted(&self, array: &vortex_fastlanes::FoRArray) -> vortex_error::VortexResult<core::option::Option<bool>>
@@ -683,3 +675,5 @@ impl vortex_array::vtable::validity::ValidityChildSliceHelper for vortex_fastlan
 pub fn vortex_fastlanes::RLEArray::unsliced_child_and_slice(&self) -> (&vortex_array::array::ArrayRef, usize, usize)
 
 pub fn vortex_fastlanes::delta_compress(array: &vortex_array::arrays::primitive::array::PrimitiveArray) -> vortex_error::VortexResult<(vortex_array::arrays::primitive::array::PrimitiveArray, vortex_array::arrays::primitive::array::PrimitiveArray)>
+
+pub fn vortex_fastlanes::initialize(session: &mut vortex_session::VortexSession)

encodings/fastlanes/src/bitpacking/compute/is_constant.rs

@@ -5,41 +5,52 @@ use std::ops::Range;
 
 use itertools::Itertools;
 use lending_iterator::LendingIterator;
+use vortex_array::ArrayRef;
+use vortex_array::ExecutionCtx;
 use vortex_array::ToCanonical;
+use vortex_array::aggregate_fn::AggregateFnRef;
+use vortex_array::aggregate_fn::fns::is_constant::IsConstant;
+use vortex_array::aggregate_fn::fns::is_constant::primitive::IS_CONST_LANE_WIDTH;
+use vortex_array::aggregate_fn::fns::is_constant::primitive::compute_is_constant;
+use vortex_array::aggregate_fn::kernels::DynAggregateKernel;
 use vortex_array::arrays::PrimitiveArray;
-use vortex_array::arrays::primitive::IS_CONST_LANE_WIDTH;
-use vortex_array::arrays::primitive::compute_is_constant;
-use vortex_array::compute::IsConstantKernel;
-use vortex_array::compute::IsConstantKernelAdapter;
-use vortex_array::compute::IsConstantOpts;
 use vortex_array::dtype::IntegerPType;
 use vortex_array::match_each_integer_ptype;
 use vortex_array::match_each_unsigned_integer_ptype;
-use vortex_array::register_kernel;
+use vortex_array::scalar::Scalar;
 use vortex_error::VortexResult;
 
 use crate::BitPacked;
 use crate::BitPackedArray;
 use crate::unpack_iter::BitPacked as BitPackedUnpack;
 
-impl IsConstantKernel for BitPacked {
-    fn is_constant(
+/// BitPacked-specific is_constant kernel with SIMD support.
+#[derive(Debug)]
+pub(crate) struct BitPackedIsConstantKernel;
+
+impl DynAggregateKernel for BitPackedIsConstantKernel {
+    fn aggregate(
         &self,
-        array: &BitPackedArray,
-        opts: &IsConstantOpts,
-    ) -> VortexResult<Option<bool>> {
-        if opts.is_negligible_cost() {
+        aggregate_fn: &AggregateFnRef,
+        batch: &ArrayRef,
+        _ctx: &mut ExecutionCtx,
+    ) -> VortexResult<Option<Scalar>> {
+        if !aggregate_fn.is::<IsConstant>() {
             return Ok(None);
         }
-        match_each_integer_ptype!(array.ptype(), |P| {
-            bitpacked_is_constant::<P, { IS_CONST_LANE_WIDTH / size_of::<P>() }>(array)
-        })
-        .map(Some)
+
+        let Some(array) = batch.as_opt::<BitPacked>() else {
+            return Ok(None);
+        };
+
+        let result = match_each_integer_ptype!(array.ptype(), |P| {
+            bitpacked_is_constant::<P, { IS_CONST_LANE_WIDTH / size_of::<P>() }>(array)?
+        });
+
+        Ok(Some(IsConstant::make_partial(batch, result)?))
     }
 }
 
-register_kernel!(IsConstantKernelAdapter(BitPacked).lift());
-
 fn bitpacked_is_constant<T: BitPackedUnpack, const WIDTH: usize>(
     array: &BitPackedArray,
 ) -> VortexResult<bool> {
@@ -170,14 +181,19 @@ fn apply_patches_idx_typed<T: BitPackedUnpack, I: IntegerPType>(
 #[cfg(test)]
 mod tests {
     use vortex_array::IntoArray;
-    use vortex_array::compute::is_constant;
+    use vortex_array::LEGACY_SESSION;
+    use vortex_array::VortexSessionExecute;
+    use vortex_array::aggregate_fn::fns::is_constant::is_constant;
     use vortex_buffer::buffer;
+    use vortex_error::VortexResult;
 
     use crate::BitPackedArray;
 
     #[test]
-    fn is_constant_with_patches() {
-        let array = BitPackedArray::encode(&buffer![4; 1025].into_array(), 2).unwrap();
-        assert!(is_constant(&array.into_array()).unwrap().unwrap());
+    fn is_constant_with_patches() -> VortexResult<()> {
+        let array = BitPackedArray::encode(&buffer![4; 1025].into_array(), 2)?;
+        let mut ctx = LEGACY_SESSION.create_execution_ctx();
+        assert!(is_constant(&array.into_array(), &mut ctx)?);
+        Ok(())
     }
 }

encodings/fastlanes/src/bitpacking/compute/mod.rs

@@ -3,7 +3,7 @@
 
 mod cast;
 mod filter;
-mod is_constant;
+pub(crate) mod is_constant;
 mod slice;
 mod take;
 

encodings/fastlanes/src/bitpacking/mod.rs

@@ -8,7 +8,7 @@ pub use array::bitpack_compress;
 pub use array::bitpack_decompress;
 pub use array::unpack_iter;
 
-mod compute;
+pub(crate) mod compute;
 
 mod vtable;
 pub use vtable::BitPacked;