Extend MaskValuesRef to filter execution

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

@@ -2,7 +2,6 @@
 // SPDX-FileCopyrightText: Copyright the Vortex contributors
 
 use std::mem::MaybeUninit;
-use std::sync::Arc;
 
 use fastlanes::BitPacking;
 use vortex_array::ArrayRef;
@@ -20,7 +19,7 @@ use vortex_buffer::Buffer;
 use vortex_buffer::BufferMut;
 use vortex_error::VortexResult;
 use vortex_mask::Mask;
-use vortex_mask::MaskValues;
+use vortex_mask::MaskValuesRef;
 
 use super::chunked_indices;
 use super::take::UNPACK_CHUNK_THRESHOLD;
@@ -82,7 +81,7 @@ impl FilterKernel for BitPacked {
 
         let patches = array
             .patches()
-            .map(|patches| patches.filter(&Mask::Values(Arc::clone(values)), ctx))
+            .map(|patches| patches.filter(&Mask::Values(MaskValuesRef::clone(values)), ctx))
             .transpose()?
             .flatten();
 
@@ -109,12 +108,12 @@ impl FilterKernel for BitPacked {
 /// Returns a tuple of (values buffer, validity mask).
 fn filter_primitive_without_patches<U: UnsignedPType + BitPacking>(
     array: ArrayView<'_, BitPacked>,
-    selection: &Arc<MaskValues>,
+    selection: &MaskValuesRef,
 ) -> VortexResult<(Buffer<U>, Validity)> {
     let values = filter_with_indices(array.data(), selection.indices());
     let validity = array
         .validity()?
-        .filter(&Mask::Values(Arc::clone(selection)))?;
+        .filter(&Mask::Values(MaskValuesRef::clone(selection)))?;
 
     Ok((values.freeze(), validity))
 }
@@ -176,7 +175,7 @@ fn filter_with_indices<T: NativePType + BitPacking>(
 }
 
 #[cfg(test)]
-mod test {
+mod tests {
     use std::sync::LazyLock;
 
     use vortex_array::IntoArray as _;

vortex-array/src/arrays/filter/execute/bool.rs

@@ -1,20 +1,18 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright the Vortex contributors
 
-use std::sync::Arc;
-
 use vortex_error::VortexExpect;
-use vortex_mask::MaskValues;
+use vortex_mask::MaskValuesRef;
 
 use crate::arrays::BoolArray;
 use crate::arrays::bool::BoolArrayExt;
 use crate::arrays::filter::execute::bitbuffer;
 use crate::arrays::filter::execute::filter_validity;
 
-pub fn filter_bool(array: &BoolArray, mask: &Arc<MaskValues>) -> BoolArray {
+pub fn filter_bool(array: &BoolArray, mask: &MaskValuesRef) -> BoolArray {
     let validity = array
         .validity()
-        .vortex_expect("bool validity should be derivable");
+        .vortex_expect("validity is derivable for a valid BoolArray");
     let filtered_validity = filter_validity(validity, mask);
 
     let bit_buffer = array.to_bit_buffer();
@@ -24,7 +22,7 @@ pub fn filter_bool(array: &BoolArray, mask: &Arc<MaskValues>) -> BoolArray {
 }
 
 #[cfg(test)]
-mod test {
+mod tests {
     use itertools::Itertools;
     use rstest::rstest;
     use vortex_mask::Mask;

vortex-array/src/arrays/filter/execute/fixed_size_list.rs

@@ -1,12 +1,11 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright the Vortex contributors
 
-use std::sync::Arc;
-
 use vortex_error::VortexExpect;
 use vortex_mask::Mask;
 use vortex_mask::MaskIter;
 use vortex_mask::MaskValues;
+use vortex_mask::MaskValuesRef;
 
 use crate::arrays::FixedSizeListArray;
 use crate::arrays::filter::execute::filter_validity;
@@ -24,12 +23,12 @@ const MASK_EXPANSION_DENSITY_THRESHOLD: f64 = 0.05;
 /// mask down to the child elements array.
 pub fn filter_fixed_size_list(
     array: &FixedSizeListArray,
-    selection_mask: &Arc<MaskValues>,
+    selection_mask: &MaskValuesRef,
 ) -> FixedSizeListArray {
     let filtered_validity = filter_validity(
         array
             .validity()
-            .vortex_expect("fixed-size-list validity should be derivable"),
+            .vortex_expect("validity is derivable for a valid FixedSizeListArray"),
         selection_mask,
     );
 
@@ -38,13 +37,11 @@ pub fn filter_fixed_size_list(
     let list_size = array.list_size();
 
     let new_elements = {
-        // We want to create a new mask specialized to the underlying `elements` of the array.
         if list_size != 0 {
-            // TODO(connor): If we can push down a "indices" or "slices" selection instead that
-            // would be much more performant.
-            let elements_mask = compute_mask_for_fsl_elements(selection_mask, list_size as usize);
+            // TODO(connor): Push down an indices or slices selection to avoid expanding the mask.
+            let elements_mask =
+                compute_mask_for_fsl_elements(selection_mask.as_ref(), list_size as usize);
 
-            // Allow the child array to filter itself.
             let new_elements = elements
                 .filter(elements_mask)
                 .vortex_expect("FixedSizeListArray elements are guaranteed to support filter");
@@ -59,22 +56,14 @@ pub fn filter_fixed_size_list(
                 "degenerate FixedSizeListArray is invalid"
             );
 
-            // NB: The safety comment for the `list_size == 0` case is here for clarity.
-
-            // SAFETY: We have verified that when `list_size == 0`
-            // - `elements` has length 0 (since it came from a valid `FixedSizeListArray`)
-            // - `filtered_validity` has the correct length because we filter with the same
-            //   `selection_mask` as the array itself
             elements.clone()
         }
     };
 
-    // SAFETY: We have verified that
-    // - The case when `list_size == 0` is safe (see above)
-    // - The `new_elements` array is guaranteed to have a length that is a multiple of
-    //   `list_size`
-    // - `filtered_validity` has the correct length because we filter with the same
-    //   `selection_mask` as the array itself
+    // SAFETY:
+    // - A valid zero-width array has no elements, which the zero-width branch preserves.
+    // - Otherwise, the expanded selection retains `list_size` elements for each selected list.
+    // - Filtering validity with `selection_mask` gives it the required `new_len`.
     unsafe {
         FixedSizeListArray::new_unchecked(new_elements, list_size, filtered_validity, new_len)
     }
@@ -120,7 +109,7 @@ fn compute_mask_for_fsl_elements(selection_mask: &MaskValues, list_size: usize)
 }
 
 #[cfg(test)]
-mod test {
+mod tests {
     use vortex_buffer::buffer;
     use vortex_mask::Mask;
 

vortex-array/src/arrays/filter/execute/listview.rs

@@ -1,11 +1,9 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright the Vortex contributors
 
-use std::sync::Arc;
-
 use vortex_error::VortexExpect;
 use vortex_mask::Mask;
-use vortex_mask::MaskValues;
+use vortex_mask::MaskValuesRef;
 
 use crate::arrays::ListViewArray;
 use crate::arrays::filter::execute::filter_validity;
@@ -23,15 +21,15 @@ use crate::arrays::listview::ListViewArraySlotsExt;
 ///
 /// The trade-off is that we may keep unreferenced elements in memory, but this is acceptable since
 /// we're optimizing for read performance and the data isn't being copied.
-pub fn filter_listview(array: &ListViewArray, selection_mask: &Arc<MaskValues>) -> ListViewArray {
+pub fn filter_listview(array: &ListViewArray, selection_mask: &MaskValuesRef) -> ListViewArray {
     let elements = array.elements();
     let offsets = array.offsets();
     let sizes = array.sizes();
 
     let new_validity = filter_validity(
         array
             .validity()
-            .vortex_expect("listview validity should be derivable"),
+            .vortex_expect("validity is derivable for a valid ListViewArray"),
         selection_mask,
     );
     debug_assert!(
@@ -40,24 +38,23 @@ pub fn filter_listview(array: &ListViewArray, selection_mask: &Arc<MaskValues>)
             .is_none_or(|len| len == selection_mask.true_count())
     );
 
-    // Simply filter the offsets and sizes arrays.
-    let mask_for_filter = Mask::Values(Arc::clone(selection_mask));
+    let mask_for_filter = Mask::Values(MaskValuesRef::clone(selection_mask));
     let new_offsets = offsets
         .filter(mask_for_filter.clone())
         .vortex_expect("ListViewArray offsets are guaranteed to support filter");
     let new_sizes = sizes
         .filter(mask_for_filter)
         .vortex_expect("ListViewArray sizes are guaranteed to support filter");
 
-    // SAFETY: Filter operation maintains all `ListViewArray` invariants:
+    // SAFETY:
     // - Offsets and sizes are derived from existing valid child arrays.
-    // - Offsets and sizes have the same length (both filtered by `selection_mask`).
-    // - Validity matches the filtered array's nullability.
+    // - Filtering offsets, sizes, and validity with `selection_mask` preserves their row alignment.
+    // - The retained elements buffer keeps every referenced element in bounds.
     unsafe { ListViewArray::new_unchecked(elements.clone(), new_offsets, new_sizes, new_validity) }
 }
 
 #[cfg(test)]
-mod test {
+mod tests {
     use std::sync::LazyLock;
 
     use vortex_buffer::buffer;

vortex-array/src/arrays/filter/execute/mod.rs

@@ -5,12 +5,10 @@
 //!
 //! The main entrypoint is [`execute_filter`] which filters any [`Canonical`] array.
 
-use std::sync::Arc;
-
 use vortex_error::VortexExpect;
 use vortex_error::VortexResult;
 use vortex_mask::Mask;
-use vortex_mask::MaskValues;
+use vortex_mask::MaskValuesRef;
 
 use crate::ArrayRef;
 use crate::Canonical;
@@ -25,7 +23,6 @@ use crate::arrays::MapArray;
 use crate::arrays::NullArray;
 use crate::arrays::VariantArray;
 use crate::arrays::extension::ExtensionArrayExt;
-use crate::arrays::filter::FilterArraySlotsExt;
 use crate::arrays::filter::FilterReduce;
 use crate::arrays::fixed_width;
 use crate::arrays::variant::VariantArraySlotsExt;
@@ -44,49 +41,36 @@ mod take;
 mod union;
 mod varbinview;
 
-/// A helper function that lazily filters a [`Validity`] with selection mask values.
-pub(crate) fn filter_validity(validity: Validity, mask: &Arc<MaskValues>) -> Validity {
+/// Lazily filters a [`Validity`] with a partially selective mask.
+pub(crate) fn filter_validity(validity: Validity, mask: &MaskValuesRef) -> Validity {
     validity
-        .filter(&Mask::Values(Arc::clone(mask)))
-        .vortex_expect("Somehow unable to wrap filter around a validity array")
+        .filter(&Mask::Values(MaskValuesRef::clone(mask)))
+        .vortex_expect("filtering validity with a partially selective mask is valid")
 }
 
-/// Check for some fast-path execution conditions before calling [`execute_filter`].
-pub(super) fn execute_filter_fast_paths(
+/// Returns an all-null result when the child contains no valid values.
+pub(super) fn execute_all_null_filter_fast_path(
     array: ArrayView<'_, Filter>,
+    selected_count: usize,
     ctx: &mut ExecutionCtx,
 ) -> VortexResult<Option<ArrayRef>> {
-    let true_count = array.mask.true_count();
-
-    // If the mask selects nothing, the output is empty.
-    if true_count == 0 {
-        return Ok(Some(Canonical::empty(array.dtype()).into_array()));
-    }
-
-    // If the mask selects everything, then we can just fully decompress the whole thing.
-    if true_count == array.mask.len() {
-        return Ok(Some(array.child().clone()));
-    }
-
-    // Also check if the array itself is completely null, in which case we only care about the total
-    // number of nulls, not the values.
-    let child_arr = array.array();
-    if child_arr
+    let child = array.array();
+    if child
         .validity()?
-        .execute_mask(child_arr.len(), ctx)?
+        .execute_mask(child.len(), ctx)?
         .true_count()
         == 0
     {
         return Ok(Some(
-            ConstantArray::new(Scalar::null(array.dtype().clone()), true_count).into_array(),
+            ConstantArray::new(Scalar::null(array.dtype().clone()), selected_count).into_array(),
         ));
     }
 
     Ok(None)
 }
 
 /// Filter a canonical array by a mask, returning a new canonical array.
-pub(super) fn execute_filter(canonical: Canonical, mask: &Arc<MaskValues>) -> Canonical {
+pub(super) fn execute_filter(canonical: Canonical, mask: &MaskValuesRef) -> Canonical {
     match canonical {
         Canonical::Null(_) => Canonical::Null(NullArray::new(mask.true_count())),
         Canonical::Bool(a) => Canonical::Bool(bool::filter_bool(&a, mask)),
@@ -103,12 +87,12 @@ pub(super) fn execute_filter(canonical: Canonical, mask: &Arc<MaskValues>) -> Ca
         Canonical::Extension(a) => {
             let filtered_storage = a
                 .storage_array()
-                .filter(Mask::Values(Arc::clone(mask)))
+                .filter(Mask::Values(MaskValuesRef::clone(mask)))
                 .vortex_expect("ExtensionArray storage type somehow could not be filtered");
             Canonical::Extension(ExtensionArray::new(a.ext_dtype().clone(), filtered_storage))
         }
         Canonical::Variant(a) => {
-            let filter_mask = Mask::Values(Arc::clone(mask));
+            let filter_mask = Mask::Values(MaskValuesRef::clone(mask));
             let filtered_core_storage = a
                 .core_storage()
                 .filter(filter_mask.clone())
@@ -126,8 +110,8 @@ pub(super) fn execute_filter(canonical: Canonical, mask: &Arc<MaskValues>) -> Ca
     }
 }
 
-fn filter_map(array: &MapArray, mask: &Arc<MaskValues>) -> MapArray {
-    let filter_mask = Mask::Values(Arc::clone(mask));
+fn filter_map(array: &MapArray, mask: &MaskValuesRef) -> MapArray {
+    let filter_mask = Mask::Values(MaskValuesRef::clone(mask));
     let filtered = <Map as FilterReduce>::filter(array.as_view(), &filter_mask)
         .vortex_expect("MapArray somehow could not be filtered")
         .vortex_expect("Map filter reduce always produces an array");

vortex-array/src/arrays/filter/execute/struct_.rs

@@ -1,26 +1,24 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright the Vortex contributors
 
-use std::sync::Arc;
-
 use vortex_error::VortexExpect;
 use vortex_mask::Mask;
-use vortex_mask::MaskValues;
+use vortex_mask::MaskValuesRef;
 
 use crate::ArrayRef;
 use crate::arrays::StructArray;
 use crate::arrays::filter::execute::filter_validity;
 use crate::arrays::struct_::StructArrayExt;
 
-pub fn filter_struct(array: &StructArray, mask: &Arc<MaskValues>) -> StructArray {
+pub fn filter_struct(array: &StructArray, mask: &MaskValuesRef) -> StructArray {
     let filtered_validity = filter_validity(
         array
             .validity()
-            .vortex_expect("struct validity should be derivable"),
+            .vortex_expect("validity is derivable for a valid StructArray"),
         mask,
     );
 
-    let mask_for_filter = Mask::Values(Arc::clone(mask));
+    let mask_for_filter = Mask::Values(MaskValuesRef::clone(mask));
     let fields: Vec<ArrayRef> = array
         .iter_unmasked_fields()
         .map(|field| {
@@ -45,7 +43,7 @@ pub fn filter_struct(array: &StructArray, mask: &Arc<MaskValues>) -> StructArray
 }
 
 #[cfg(test)]
-mod test {
+mod tests {
     use vortex_mask::Mask;
 
     use crate::IntoArray;

vortex-array/src/arrays/filter/execute/union.rs

@@ -1,19 +1,17 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright the Vortex contributors
 
-use std::sync::Arc;
-
 use vortex_error::VortexExpect;
 use vortex_mask::Mask;
-use vortex_mask::MaskValues;
+use vortex_mask::MaskValuesRef;
 
 use crate::ArrayRef;
 use crate::arrays::UnionArray;
 use crate::arrays::union::UnionArrayExt;
 use crate::arrays::union::UnionArraySlotsExt;
 
-pub fn filter_union(array: &UnionArray, mask: &Arc<MaskValues>) -> UnionArray {
-    let filter_mask = Mask::Values(Arc::clone(mask));
+pub fn filter_union(array: &UnionArray, mask: &MaskValuesRef) -> UnionArray {
+    let filter_mask = Mask::Values(MaskValuesRef::clone(mask));
 
     let type_ids = array
         .type_ids()