clean up numeric compute

encodings/datetime-parts/src/canonical.rs

@@ -12,7 +12,6 @@ use vortex_array::dtype::PType;
 use vortex_array::extension::datetime::TimeUnit;
 use vortex_array::extension::datetime::Timestamp;
 use vortex_array::match_each_integer_ptype;
-use vortex_array::validity::Validity;
 use vortex_buffer::BufferMut;
 use vortex_error::VortexExpect as _;
 use vortex_error::VortexResult;
@@ -95,8 +94,7 @@ pub fn decode_to_temporal(
     }
 
     Ok(TemporalArray::new_timestamp(
-        PrimitiveArray::new(values.freeze(), Validity::copy_from_array(array.as_ref())?)
-            .into_array(),
+        PrimitiveArray::new(values.freeze(), array.validity()?).into_array(),
         options.unit,
         options.tz.clone(),
     ))

encodings/fastlanes/src/rle/array/rle_decompress.rs

@@ -11,7 +11,6 @@ use vortex_array::arrays::PrimitiveArray;
 use vortex_array::dtype::NativePType;
 use vortex_array::match_each_native_ptype;
 use vortex_array::match_each_unsigned_integer_ptype;
-use vortex_array::validity::Validity;
 use vortex_buffer::BufferMut;
 use vortex_error::VortexResult;
 use vortex_error::vortex_panic;
@@ -102,6 +101,6 @@ where
         buffer
             .freeze()
             .slice(offset_within_chunk..(offset_within_chunk + array.len())),
-        Validity::copy_from_array(array.as_ref())?,
+        array.validity()?,
     ))
 }

encodings/fsst/public-api.lock

@@ -106,7 +106,7 @@ pub fn vortex_fsst::FSSTVTable::slice(array: &Self::Array, range: core::ops::ran
 
 impl vortex_array::expr::exprs::binary::compare::CompareKernel for vortex_fsst::FSSTVTable
 
-pub fn vortex_fsst::FSSTVTable::compare(lhs: &vortex_fsst::FSSTArray, rhs: &dyn vortex_array::array::Array, operator: vortex_array::expr::exprs::operators::CompareOperator, _ctx: &mut vortex_array::executor::ExecutionCtx) -> vortex_error::VortexResult<core::option::Option<vortex_array::array::ArrayRef>>
+pub fn vortex_fsst::FSSTVTable::compare(lhs: &vortex_fsst::FSSTArray, rhs: &dyn vortex_array::array::Array, operator: vortex_array::expr::exprs::operators::CompareOperator, ctx: &mut vortex_array::executor::ExecutionCtx) -> vortex_error::VortexResult<core::option::Option<vortex_array::array::ArrayRef>>
 
 impl vortex_array::expr::exprs::cast::kernel::CastReduce for vortex_fsst::FSSTVTable
 

encodings/fsst/src/compute/compare.rs

@@ -5,18 +5,14 @@ use vortex_array::Array;
 use vortex_array::ArrayRef;
 use vortex_array::ExecutionCtx;
 use vortex_array::IntoArray;
-use vortex_array::ToCanonical;
 use vortex_array::arrays::BoolArray;
 use vortex_array::arrays::ConstantArray;
 use vortex_array::builtins::ArrayBuiltins;
-use vortex_array::compute::compare_lengths_to_empty;
 use vortex_array::dtype::DType;
 use vortex_array::expr::CompareKernel;
 use vortex_array::expr::CompareOperator;
 use vortex_array::expr::Operator;
-use vortex_array::match_each_integer_ptype;
 use vortex_array::scalar::Scalar;
-use vortex_array::validity::Validity;
 use vortex_buffer::BitBuffer;
 use vortex_buffer::ByteBuffer;
 use vortex_error::VortexExpect;
@@ -31,10 +27,10 @@ impl CompareKernel for FSSTVTable {
         lhs: &FSSTArray,
         rhs: &dyn Array,
         operator: CompareOperator,
-        _ctx: &mut ExecutionCtx,
+        ctx: &mut ExecutionCtx,
     ) -> VortexResult<Option<ArrayRef>> {
         match rhs.as_constant() {
-            Some(constant) => compare_fsst_constant(lhs, &constant, operator),
+            Some(constant) => compare_fsst_constant(lhs, &constant, operator, ctx),
             // Otherwise, fall back to the default comparison behavior.
             _ => Ok(None),
         }
@@ -46,6 +42,7 @@ fn compare_fsst_constant(
     left: &FSSTArray,
     right: &Scalar,
     operator: CompareOperator,
+    ctx: &mut ExecutionCtx,
 ) -> VortexResult<Option<ArrayRef>> {
     let is_rhs_empty = match right.dtype() {
         DType::Binary(_) => right
@@ -64,21 +61,24 @@ fn compare_fsst_constant(
             CompareOperator::Gte => BitBuffer::new_set(left.len()),
             // No value is lt ""
             CompareOperator::Lt => BitBuffer::new_unset(left.len()),
-            _ => {
-                let uncompressed_lengths = left.uncompressed_lengths().to_primitive();
-                match_each_integer_ptype!(uncompressed_lengths.ptype(), |P| {
-                    compare_lengths_to_empty(
-                        uncompressed_lengths.as_slice::<P>().iter().copied(),
-                        operator,
+            _ => left
+                .uncompressed_lengths()
+                .to_array()
+                .binary(
+                    ConstantArray::new(
+                        Scalar::zero_value(left.uncompressed_lengths().dtype()),
+                        left.uncompressed_lengths().len(),
                     )
-                })
-            }
+                    .into_array(),
+                    operator.into(),
+                )?
+                .execute(ctx)?,
         };
 
         return Ok(Some(
             BoolArray::new(
                 buffer,
-                Validity::copy_from_array(left.as_ref())?
+                left.validity()?
                     .union_nullability(right.dtype().nullability()),
             )
             .into_array(),

encodings/fsst/src/compute/filter.rs

@@ -22,8 +22,9 @@ impl FilterKernel for FSSTVTable {
         // Directly invoke VarBin's FilterKernel to get a concrete VarBinArray back.
         let filtered_codes = <VarBinVTable as FilterKernel>::filter(array.codes(), mask, ctx)?
             .vortex_expect("VarBin filter kernel always returns Some")
-            .as_::<VarBinVTable>()
-            .clone();
+            .try_into::<VarBinVTable>()
+            .ok()
+            .vortex_expect("must be VarBinVTable");
 
         Ok(Some(
             FSSTArray::try_new(

vortex-array/public-api.lock

@@ -6092,10 +6092,6 @@ impl vortex_array::compute::Options for vortex_array::expr::BetweenOptions
 
 pub fn vortex_array::expr::BetweenOptions::as_any(&self) -> &dyn core::any::Any
 
-impl vortex_array::compute::Options for vortex_array::scalar::NumericOperator
-
-pub fn vortex_array::scalar::NumericOperator::as_any(&self) -> &dyn core::any::Any
-
 pub trait vortex_array::compute::SumKernel: vortex_array::vtable::VTable
 
 pub fn vortex_array::compute::SumKernel::sum(&self, array: &Self::Array, accumulator: &vortex_array::scalar::Scalar) -> vortex_error::VortexResult<vortex_array::scalar::Scalar>
@@ -6134,8 +6130,6 @@ pub fn vortex_array::compute::arrow_filter_fn(array: &dyn vortex_array::Array, m
 
 pub fn vortex_array::compute::cast(array: &dyn vortex_array::Array, dtype: &vortex_array::dtype::DType) -> vortex_error::VortexResult<vortex_array::ArrayRef>
 
-pub fn vortex_array::compute::compare_lengths_to_empty<P, I>(lengths: I, op: vortex_array::expr::CompareOperator) -> vortex_buffer::bit::buf::BitBuffer where P: vortex_array::dtype::IntegerPType, I: core::iter::traits::iterator::Iterator<Item = P>
-
 pub fn vortex_array::compute::div(lhs: &dyn vortex_array::Array, rhs: &dyn vortex_array::Array) -> vortex_error::VortexResult<vortex_array::ArrayRef>
 
 pub fn vortex_array::compute::div_scalar(lhs: &dyn vortex_array::Array, rhs: vortex_array::scalar::Scalar) -> vortex_error::VortexResult<vortex_array::ArrayRef>
@@ -12120,8 +12114,16 @@ pub fn vortex_array::expr::cast(child: vortex_array::expr::Expression, target: v
 
 pub fn vortex_array::expr::checked_add(lhs: vortex_array::expr::Expression, rhs: vortex_array::expr::Expression) -> vortex_array::expr::Expression
 
+pub fn vortex_array::expr::checked_div(lhs: vortex_array::expr::Expression, rhs: vortex_array::expr::Expression) -> vortex_array::expr::Expression
+
+pub fn vortex_array::expr::checked_mul(lhs: vortex_array::expr::Expression, rhs: vortex_array::expr::Expression) -> vortex_array::expr::Expression
+
+pub fn vortex_array::expr::checked_sub(lhs: vortex_array::expr::Expression, rhs: vortex_array::expr::Expression) -> vortex_array::expr::Expression
+
 pub fn vortex_array::expr::col(field: impl core::convert::Into<vortex_array::dtype::FieldName>) -> vortex_array::expr::Expression
 
+pub fn vortex_array::expr::compare_nested_arrow_arrays(lhs: &dyn arrow_array::array::Array, rhs: &dyn arrow_array::array::Array, operator: vortex_array::expr::CompareOperator) -> vortex_error::VortexResult<arrow_array::array::boolean_array::BooleanArray>
+
 pub fn vortex_array::expr::descendent_annotations<A: vortex_array::expr::AnnotationFn>(expr: &vortex_array::expr::Expression, annotate: A) -> vortex_array::expr::Annotations<'_, <A as vortex_array::expr::AnnotationFn>::Annotation>
 
 pub fn vortex_array::expr::dynamic(operator: vortex_array::expr::CompareOperator, rhs_value: impl core::ops::function::Fn() -> core::option::Option<vortex_array::scalar::ScalarValue> + core::marker::Send + core::marker::Sync + 'static, rhs_dtype: vortex_array::dtype::DType, default: bool, lhs: vortex_array::expr::Expression) -> vortex_array::expr::Expression
@@ -13218,10 +13220,6 @@ impl core::marker::Copy for vortex_array::scalar::NumericOperator
 
 impl core::marker::StructuralPartialEq for vortex_array::scalar::NumericOperator
 
-impl vortex_array::compute::Options for vortex_array::scalar::NumericOperator
-
-pub fn vortex_array::scalar::NumericOperator::as_any(&self) -> &dyn core::any::Any
-
 pub enum vortex_array::scalar::PValue
 
 pub vortex_array::scalar::PValue::F16(half::binary16::f16)
@@ -15428,8 +15426,6 @@ pub fn vortex_array::validity::Validity::as_array(&self) -> core::option::Option
 
 pub fn vortex_array::validity::Validity::cast_nullability(self, nullability: vortex_array::dtype::Nullability, len: usize) -> vortex_error::VortexResult<vortex_array::validity::Validity>
 
-pub fn vortex_array::validity::Validity::copy_from_array(array: &dyn vortex_array::Array) -> vortex_error::VortexResult<Self>
-
 pub fn vortex_array::validity::Validity::filter(&self, mask: &vortex_mask::Mask) -> vortex_error::VortexResult<Self>
 
 pub fn vortex_array::validity::Validity::into_array(self) -> core::option::Option<vortex_array::ArrayRef>
@@ -17900,6 +17896,10 @@ impl vortex_array::Executable for vortex_array::arrays::VarBinViewArray
 
 pub fn vortex_array::arrays::VarBinViewArray::execute(array: vortex_array::ArrayRef, ctx: &mut vortex_array::ExecutionCtx) -> vortex_error::VortexResult<Self>
 
+impl vortex_array::Executable for vortex_buffer::bit::buf::BitBuffer
+
+pub fn vortex_buffer::bit::buf::BitBuffer::execute(array: vortex_array::ArrayRef, ctx: &mut vortex_array::ExecutionCtx) -> vortex_error::VortexResult<Self>
+
 impl vortex_array::Executable for vortex_mask::Mask
 
 pub fn vortex_mask::Mask::execute(array: vortex_array::ArrayRef, ctx: &mut vortex_array::ExecutionCtx) -> vortex_error::VortexResult<Self>

vortex-array/src/arrays/chunked/vtable/canonical.rs

@@ -36,17 +36,13 @@ pub(super) fn _canonicalize(
 
     Ok(match array.dtype() {
         DType::Struct(struct_dtype, _) => {
-            let struct_array = pack_struct_chunks(
-                array.chunks(),
-                Validity::copy_from_array(array.as_ref())?,
-                struct_dtype,
-                ctx,
-            )?;
+            let struct_array =
+                pack_struct_chunks(array.chunks(), array.validity()?, struct_dtype, ctx)?;
             Canonical::Struct(struct_array)
         }
         DType::List(elem_dtype, _) => Canonical::List(swizzle_list_chunks(
             array.chunks(),
-            Validity::copy_from_array(array.as_ref())?,
+            array.validity()?,
             elem_dtype,
             ctx,
         )?),

vortex-array/src/arrays/varbin/compute/compare.rs

@@ -4,7 +4,6 @@
 use arrow_array::BinaryArray;
 use arrow_array::StringArray;
 use arrow_ord::cmp;
-use itertools::Itertools;
 use vortex_buffer::BitBuffer;
 use vortex_error::VortexExpect as _;
 use vortex_error::VortexResult;
@@ -23,7 +22,6 @@ use crate::arrays::VarBinVTable;
 use crate::arrow::Datum;
 use crate::arrow::from_arrow_array_with_len;
 use crate::builtins::ArrayBuiltins;
-use crate::compute::compare_lengths_to_empty;
 use crate::dtype::DType;
 use crate::dtype::IntegerPType;
 use crate::expr::CompareKernel;
@@ -60,13 +58,16 @@ impl CompareKernel for VarBinVTable {
                 let buffer = match operator {
                     CompareOperator::Gte => BitBuffer::new_set(len), // Every possible value is >= ""
                     CompareOperator::Lt => BitBuffer::new_unset(len), // No value is < ""
-                    CompareOperator::Eq
-                    | CompareOperator::NotEq
-                    | CompareOperator::Gt
-                    | CompareOperator::Lte => {
+                    CompareOperator::Eq | CompareOperator::Lte => {
                         let lhs_offsets = lhs.offsets().to_primitive();
                         match_each_integer_ptype!(lhs_offsets.ptype(), |P| {
-                            compare_offsets_to_empty::<P>(lhs_offsets, operator)
+                            compare_offsets_to_empty::<P>(lhs_offsets, true)
+                        })
+                    }
+                    CompareOperator::NotEq | CompareOperator::Gt => {
+                        let lhs_offsets = lhs.offsets().to_primitive();
+                        match_each_integer_ptype!(lhs_offsets.ptype(), |P| {
+                            compare_offsets_to_empty::<P>(lhs_offsets, false)
                         })
                     }
                 };
@@ -129,16 +130,14 @@ impl CompareKernel for VarBinVTable {
     }
 }
 
-fn compare_offsets_to_empty<P: IntegerPType>(
-    offsets: PrimitiveArray,
-    operator: CompareOperator,
-) -> BitBuffer {
-    let lengths_iter = offsets
-        .as_slice::<P>()
-        .iter()
-        .tuple_windows()
-        .map(|(&s, &e)| e - s);
-    compare_lengths_to_empty(lengths_iter, operator)
+fn compare_offsets_to_empty<P: IntegerPType>(offsets: PrimitiveArray, eq: bool) -> BitBuffer {
+    let fn_ = if eq { P::eq } else { P::ne };
+    let offsets = offsets.as_slice::<P>();
+    BitBuffer::collect_bool(offsets.len() - 1, |idx| {
+        let left = unsafe { offsets.get_unchecked(idx) };
+        let right = unsafe { offsets.get_unchecked(idx + 1) };
+        fn_(left, right)
+    })
 }
 
 #[cfg(test)]

vortex-array/src/canonical.rs

@@ -5,6 +5,7 @@
 
 use std::sync::Arc;
 
+use vortex_buffer::BitBuffer;
 use vortex_buffer::Buffer;
 use vortex_error::VortexExpect;
 use vortex_error::VortexResult;
@@ -782,6 +783,20 @@ impl Executable for BoolArray {
     }
 }
 
+/// Execute the array to a [`BitBuffer`], aka a non-nullable  [`BoolArray`].
+///
+/// This will panic if the array's dtype is not non-nullable bool.
+impl Executable for BitBuffer {
+    fn execute(array: ArrayRef, ctx: &mut ExecutionCtx) -> VortexResult<Self> {
+        let bool = BoolArray::execute(array, ctx)?;
+        assert!(
+            !bool.dtype().is_nullable(),
+            "bit buffer execute only works with non-nullable bool arrays"
+        );
+        Ok(bool.into_bit_buffer())
+    }
+}
+
 /// Execute the array to canonical form and unwrap as a [`NullArray`].
 ///
 /// This will panic if the array's dtype is not null.