Reduce lazy masks into arrays with all-valid metadata (#10016)

## Summary

Allows lazy masks to attach directly as validity when an encoding can
inspect validity through metadata and the input is `AllValid` or
`NonNullable`. Array-backed validity and encodings that have not opted
in retain the existing lazy-mask fallback.

The reduction rule, in pseudocode, is:

```text
Mask(Array(values, validity = AllValid), m)
    -> Array(values, validity = m)

Mask(Array(values, validity = NonNullable), m)
    -> Array(values, validity = m)
```

This applies when the input encoding opts into
`VALIDITY_IS_METADATA_ONLY` and supports mask reduction. `m` is a
non-nullable Boolean array with the same length as the input, and it may
be lazy. The reduction reuses the values and attaches `m` as validity
without executing it. The output dtype is nullable.

The identity is `AllValid AND m = m`, so no intermediate validity array
is needed.

## Changes

Adds `MaskReduce::VALIDITY_IS_METADATA_ONLY`, defaulting to false, and
enables it for Boolean, primitive, decimal, string, list, struct, map,
and ByteBool encodings. String, decimal, and list reducers preserve
existing buffers and children when rebuilding. List-view masking also
preserves its metadata without revalidating the zero-copy flag.

Two focused tests cover direct attachment to an all-valid primitive
array and the fallback for array-backed validity, including an all-true
bitmap. Existing constant-mask tests remain unchanged, and all tests
stay inline. The Boolean device-buffer fix landed in #10018.

Validation before the rebase: 168 focused comparison, mask, and RowFn
tests passed on the combined stack through #9979, along with `cargo
clippy -p vortex-array --all-targets --all-features -- -D warnings`. The
rebase preserves this PR's patches. Tests, formatting, and benchmarks
were not rerun locally after the rebase.

---------

Signed-off-by: Connor Tsui <connor.tsui20@gmail.com>

Author: connortsui20

encodings/bytebool/src/compute.rs

@@ -73,6 +73,8 @@ impl CastKernel for ByteBool {
 }
 
 impl MaskReduce for ByteBool {
+    const VALIDITY_IS_METADATA_ONLY: bool = true;
+
     fn mask(array: ArrayView<'_, Self>, mask: &ArrayRef) -> VortexResult<Option<ArrayRef>> {
         Ok(Some(
             ByteBool::new(

vortex-array/src/arrays/bool/compute/mask.rs

@@ -12,6 +12,8 @@ use crate::scalar_fn::fns::mask::MaskReduce;
 use crate::validity::Validity;
 
 impl MaskReduce for Bool {
+    const VALIDITY_IS_METADATA_ONLY: bool = true;
+
     fn mask(array: ArrayView<'_, Bool>, mask: &ArrayRef) -> VortexResult<Option<ArrayRef>> {
         Ok(Some(
             BoolArray::try_new_from_handle(

vortex-array/src/arrays/decimal/compute/mask.rs

@@ -8,25 +8,22 @@ use crate::IntoArray;
 use crate::array::ArrayView;
 use crate::arrays::Decimal;
 use crate::arrays::DecimalArray;
-use crate::match_each_decimal_value_type;
 use crate::scalar_fn::fns::mask::MaskReduce;
 use crate::validity::Validity;
 
 impl MaskReduce for Decimal {
+    const VALIDITY_IS_METADATA_ONLY: bool = true;
+
     fn mask(array: ArrayView<'_, Decimal>, mask: &ArrayRef) -> VortexResult<Option<ArrayRef>> {
-        Ok(Some(match_each_decimal_value_type!(
-            array.values_type(),
-            |D| {
-                // SAFETY: masking the validity does not affect the invariants
-                unsafe {
-                    DecimalArray::new_unchecked(
-                        array.buffer::<D>(),
-                        array.decimal_dtype(),
-                        array.validity()?.and(Validity::Array(mask.clone()))?,
-                    )
-                }
-                .into_array()
-            }
-        )))
+        // SAFETY: the values and decimal type are unchanged, and masking only removes valid rows.
+        Ok(Some(unsafe {
+            DecimalArray::new_unchecked_handle(
+                array.buffer_handle().clone(),
+                array.values_type(),
+                array.decimal_dtype(),
+                array.validity()?.and(Validity::Array(mask.clone()))?,
+            )
+            .into_array()
+        }))
     }
 }

vortex-array/src/arrays/fixed_size_list/compute/mask.rs

@@ -14,6 +14,8 @@ use crate::scalar_fn::fns::mask::MaskReduce;
 use crate::validity::Validity;
 
 impl MaskReduce for FixedSizeList {
+    const VALIDITY_IS_METADATA_ONLY: bool = true;
+
     fn mask(
         array: ArrayView<'_, FixedSizeList>,
         mask: &ArrayRef,

vortex-array/src/arrays/list/compute/mask.rs

@@ -13,12 +13,17 @@ use crate::scalar_fn::fns::mask::MaskReduce;
 use crate::validity::Validity;
 
 impl MaskReduce for List {
+    const VALIDITY_IS_METADATA_ONLY: bool = true;
+
     fn mask(array: ArrayView<'_, List>, mask: &ArrayRef) -> VortexResult<Option<ArrayRef>> {
-        ListArray::try_new(
-            array.elements().clone(),
-            array.offsets().clone(),
-            array.validity()?.and(Validity::Array(mask.clone()))?,
-        )
-        .map(|a| Some(a.into_array()))
+        // SAFETY: elements and offsets are unchanged, and masking only removes valid rows.
+        Ok(Some(unsafe {
+            ListArray::new_unchecked(
+                array.elements().clone(),
+                array.offsets().clone(),
+                array.validity()?.and(Validity::Array(mask.clone()))?,
+            )
+            .into_array()
+        }))
     }
 }

vortex-array/src/arrays/listview/compute/mask.rs

@@ -5,27 +5,39 @@ use vortex_error::VortexResult;
 
 use crate::ArrayRef;
 use crate::IntoArray;
+use crate::array::ArrayParts;
 use crate::array::ArrayView;
 use crate::arrays::ListView;
 use crate::arrays::ListViewArray;
 use crate::arrays::listview::ListViewArraySlotsExt;
+use crate::arrays::listview::ListViewData;
 use crate::scalar_fn::fns::mask::MaskReduce;
 use crate::validity::Validity;
 
 impl MaskReduce for ListView {
+    const VALIDITY_IS_METADATA_ONLY: bool = true;
+
     fn mask(array: ArrayView<'_, ListView>, mask: &ArrayRef) -> VortexResult<Option<ArrayRef>> {
-        // SAFETY: masking the validity does not affect the invariants
+        let validity = array.validity()?.and(Validity::Array(mask.clone()))?;
+        let slots = ListViewData::make_slots(
+            array.elements(),
+            array.offsets(),
+            array.sizes(),
+            &validity,
+            array.len(),
+        );
+        let parts = ArrayParts::new(
+            ListView,
+            array.dtype().as_nullable(),
+            array.len(),
+            array.data().clone(),
+        )
+        .with_slots(slots);
+
+        // SAFETY: elements, offsets, sizes, and their metadata are unchanged. Masking only removes
+        // valid rows, so the existing zero-copy-to-list guarantee still holds.
         Ok(Some(
-            unsafe {
-                ListViewArray::new_unchecked(
-                    array.elements().clone(),
-                    array.offsets().clone(),
-                    array.sizes().clone(),
-                    array.validity()?.and(Validity::Array(mask.clone()))?,
-                )
-                .with_zero_copy_to_list(array.is_zero_copy_to_list())
-            }
-            .into_array(),
+            unsafe { ListViewArray::from_parts_unchecked(parts) }.into_array(),
         ))
     }
 }

vortex-array/src/arrays/map/compute/mask.rs

@@ -15,6 +15,8 @@ use crate::scalar_fn::fns::mask::MaskKernel;
 use crate::scalar_fn::fns::mask::MaskReduce;
 
 impl MaskReduce for Map {
+    const VALIDITY_IS_METADATA_ONLY: bool = true;
+
     fn mask(array: ArrayView<'_, Self>, mask: &ArrayRef) -> VortexResult<Option<ArrayRef>> {
         let Some(entries) =
             <ListView as MaskReduce>::mask(array.entries().as_::<ListView>(), mask)?

vortex-array/src/arrays/primitive/compute/mask.rs

@@ -12,6 +12,8 @@ use crate::scalar_fn::fns::mask::MaskReduce;
 use crate::validity::Validity;
 
 impl MaskReduce for Primitive {
+    const VALIDITY_IS_METADATA_ONLY: bool = true;
+
     fn mask(array: ArrayView<'_, Primitive>, mask: &ArrayRef) -> VortexResult<Option<ArrayRef>> {
         // SAFETY: validity and data buffer still have same length
         Ok(Some(unsafe {

vortex-array/src/arrays/struct_/compute/mask.rs

@@ -13,6 +13,8 @@ use crate::scalar_fn::fns::mask::MaskReduce;
 use crate::validity::Validity;
 
 impl MaskReduce for Struct {
+    const VALIDITY_IS_METADATA_ONLY: bool = true;
+
     fn mask(array: ArrayView<'_, Struct>, mask: &ArrayRef) -> VortexResult<Option<ArrayRef>> {
         StructArray::try_new_with_dtype(
             array.iter_unmasked_fields().cloned(),

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

@@ -13,16 +13,19 @@ use crate::scalar_fn::fns::mask::MaskReduce;
 use crate::validity::Validity;
 
 impl MaskReduce for VarBin {
+    const VALIDITY_IS_METADATA_ONLY: bool = true;
+
     fn mask(array: ArrayView<'_, VarBin>, mask: &ArrayRef) -> VortexResult<Option<ArrayRef>> {
-        Ok(Some(
-            VarBinArray::try_new(
+        // SAFETY: offsets and bytes are unchanged, and masking only removes valid rows.
+        Ok(Some(unsafe {
+            VarBinArray::new_unchecked_from_handle(
                 array.offsets().clone(),
-                array.bytes().clone(),
+                array.bytes_handle().clone(),
                 array.dtype().as_nullable(),
                 array.validity()?.and(Validity::Array(mask.clone()))?,
-            )?
-            .into_array(),
-        ))
+            )
+            .into_array()
+        }))
     }
 }