move towards BitBuffer/{Mut}View over owned bit buffers

vortex-array/src/aggregate_fn/fns/all_non_distinct/bool.rs

@@ -10,5 +10,5 @@ where
     L: BoolArrayExt,
     R: BoolArrayExt,
 {
-    Ok(lhs.to_bit_buffer() == rhs.to_bit_buffer())
+    Ok(lhs.bit_buffer_view() == rhs.bit_buffer_view())
 }

vortex-array/src/aggregate_fn/fns/is_constant/bool.rs

@@ -5,6 +5,6 @@ use crate::arrays::BoolArray;
 use crate::arrays::bool::BoolArrayExt;
 
 pub(super) fn check_bool_constant(array: &BoolArray) -> bool {
-    let true_count = array.to_bit_buffer().true_count();
+    let true_count = array.bit_buffer_view().true_count();
     true_count == array.len() || true_count == 0
 }

vortex-array/src/aggregate_fn/fns/is_sorted/bool.rs

@@ -21,7 +21,7 @@ pub(super) fn check_bool_sorted(
     {
         Mask::AllFalse(_) => Ok(!strict),
         Mask::AllTrue(_) => {
-            let values = array.to_bit_buffer();
+            let values = array.bit_buffer_view();
             Ok(if strict {
                 values.iter().is_strict_sorted()
             } else {
@@ -31,15 +31,15 @@ pub(super) fn check_bool_sorted(
         Mask::Values(mask_values) => {
             if strict {
                 let validity_buffer = mask_values.bit_buffer();
-                let values = array.to_bit_buffer();
+                let values = array.bit_buffer_view();
                 Ok(validity_buffer
                     .iter()
                     .zip(values.iter())
                     .map(|(is_valid, value)| is_valid.then_some(value))
                     .is_strict_sorted())
             } else {
                 let set_indices = mask_values.bit_buffer().set_indices();
-                let values = array.to_bit_buffer();
+                let values = array.bit_buffer_view();
                 let values_iter = set_indices.map(|idx|
                     // Safety:
                     // All idxs are in-bounds for the array.

vortex-array/src/aggregate_fn/fns/min_max/bool.rs

@@ -25,7 +25,7 @@ pub(super) fn accumulate_bool(
         .execute_mask(array.as_ref().len(), ctx)?;
     let (true_count, valid_count) = match mask.bit_buffer() {
         AllOr::None => return Ok(()),
-        AllOr::All => (array.to_bit_buffer().true_count(), array.as_ref().len()),
+        AllOr::All => (array.bit_buffer_view().true_count(), array.as_ref().len()),
         AllOr::Some(validity) => (
             array.to_bit_buffer().bitand(validity).true_count(),
             validity.true_count(),

vortex-array/src/aggregate_fn/fns/sum/bool.rs

@@ -25,7 +25,7 @@ pub(super) fn accumulate_bool(
     let mask = b.as_ref().validity()?.execute_mask(b.as_ref().len(), ctx)?;
     let true_count = match mask.bit_buffer() {
         AllOr::None => return Ok(false),
-        AllOr::All => b.to_bit_buffer().true_count() as u64,
+        AllOr::All => b.bit_buffer_view().true_count() as u64,
         AllOr::Some(validity) => b.to_bit_buffer().bitand(validity).true_count() as u64,
     };
 

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

@@ -7,7 +7,9 @@ use std::fmt::Formatter;
 use arrow_array::BooleanArray;
 use smallvec::smallvec;
 use vortex_buffer::BitBuffer;
+use vortex_buffer::BitBufferMeta;
 use vortex_buffer::BitBufferMut;
+use vortex_buffer::BitBufferView;
 use vortex_error::VortexExpect;
 use vortex_error::VortexResult;
 use vortex_error::vortex_ensure;
@@ -69,19 +71,18 @@ pub(super) const SLOT_NAMES: [&str; NUM_SLOTS] = ["validity"];
 #[derive(Clone, Debug)]
 pub struct BoolData {
     pub(super) bits: BufferHandle,
-    pub(super) offset: usize,
+    pub(super) meta: BitBufferMeta,
 }
 
 impl Display for BoolData {
     fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
-        write!(f, "offset: {}", self.offset)
+        write!(f, "offset: {}", self.meta.offset())
     }
 }
 
 pub struct BoolDataParts {
     pub bits: BufferHandle,
-    pub offset: usize,
-    pub len: usize,
+    pub meta: BitBufferMeta,
 }
 
 pub trait BoolArrayExt: TypedArrayRef<Bool> {
@@ -101,7 +102,12 @@ pub trait BoolArrayExt: TypedArrayRef<Bool> {
 
     fn to_bit_buffer(&self) -> BitBuffer {
         let buffer = self.bits.as_host().clone();
-        BitBuffer::new_with_offset(buffer, self.as_ref().len(), self.offset)
+        BitBuffer::new_with_offset(buffer, self.meta.len(), self.meta.offset())
+    }
+
+    /// Borrow the array's packed bits as a [`BitBufferView`] without cloning the backing buffer.
+    fn bit_buffer_view(&self) -> BitBufferView<'_> {
+        BitBufferView::from_meta(self.bits.as_host().as_slice(), self.meta)
     }
 
     fn maybe_execute_mask(&self, ctx: &mut ExecutionCtx) -> VortexResult<Option<Mask>> {
@@ -141,8 +147,7 @@ impl BoolData {
     pub fn into_parts(self, len: usize) -> BoolDataParts {
         BoolDataParts {
             bits: self.bits,
-            offset: self.offset,
-            len,
+            meta: BitBufferMeta::new(self.meta.offset(), len),
         }
     }
 
@@ -242,7 +247,7 @@ impl Array<Bool> {
         let len = self.len();
         let data = self.into_data();
         let buffer = data.bits.unwrap_host();
-        BitBuffer::new_with_offset(buffer, len, data.offset)
+        BitBuffer::new_with_offset(buffer, len, data.meta.offset())
     }
 }
 
@@ -252,11 +257,11 @@ impl BoolData {
         let bits = bits.shrink_offset();
         Self::validate(&bits, &validity)?;
 
-        let (offset, _len, buffer) = bits.into_inner();
+        let (offset, len, buffer) = bits.into_inner();
 
         Ok(Self {
             bits: BufferHandle::new_host(buffer),
-            offset,
+            meta: BitBufferMeta::new(offset, len),
         })
     }
 
@@ -281,18 +286,21 @@ impl BoolData {
             bits.len() * 8,
         );
 
-        Ok(Self { bits, offset })
+        Ok(Self {
+            bits,
+            meta: BitBufferMeta::new(offset, len),
+        })
     }
 
     pub(super) unsafe fn new_unchecked(bits: BitBuffer, validity: Validity) -> Self {
         if cfg!(debug_assertions) {
             Self::try_new(bits, validity).vortex_expect("Failed to create BoolData")
         } else {
-            let (offset, _len, buffer) = bits.into_inner();
+            let (offset, len, buffer) = bits.into_inner();
 
             Self {
                 bits: BufferHandle::new_host(buffer),
-                offset,
+                meta: BitBufferMeta::new(offset, len),
             }
         }
     }

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

@@ -4,6 +4,7 @@
 use itertools::Itertools as _;
 use num_traits::AsPrimitive;
 use vortex_buffer::BitBuffer;
+use vortex_buffer::BitBufferView;
 use vortex_buffer::get_bit;
 use vortex_error::VortexResult;
 use vortex_mask::Mask;
@@ -42,7 +43,10 @@ impl TakeExecute for Bool {
         };
         let indices_nulls_zeroed = indices_nulls_zeroed.execute::<PrimitiveArray>(ctx)?;
         let buffer = match_each_integer_ptype!(indices_nulls_zeroed.ptype(), |I| {
-            take_valid_indices(&array.to_bit_buffer(), indices_nulls_zeroed.as_slice::<I>())
+            take_valid_indices(
+                array.bit_buffer_view(),
+                indices_nulls_zeroed.as_slice::<I>(),
+            )
         });
 
         Ok(Some(
@@ -51,7 +55,7 @@ impl TakeExecute for Bool {
     }
 }
 
-fn take_valid_indices<I: AsPrimitive<usize>>(bools: &BitBuffer, indices: &[I]) -> BitBuffer {
+fn take_valid_indices<I: AsPrimitive<usize>>(bools: BitBufferView<'_>, indices: &[I]) -> BitBuffer {
     // For boolean arrays that roughly fit into a single page (at least, on Linux), it's worth
     // the overhead to convert to a Vec<bool>.
     if bools.len() <= 4096 {
@@ -68,9 +72,9 @@ fn take_byte_bool<I: AsPrimitive<usize>>(bools: Vec<bool>, indices: &[I]) -> Bit
     })
 }
 
-fn take_bool_impl<I: AsPrimitive<usize>>(bools: &BitBuffer, indices: &[I]) -> BitBuffer {
+fn take_bool_impl<I: AsPrimitive<usize>>(bools: BitBufferView<'_>, indices: &[I]) -> BitBuffer {
     // We dereference to underlying buffer to avoid access cost on every index.
-    let buffer = bools.inner().as_ref();
+    let buffer = bools.inner();
     BitBuffer::collect_bool(indices.len(), |idx| {
         // SAFETY: we can take from the indices unchecked since collect_bool just iterates len.
         let idx = unsafe { indices.get_unchecked(idx).as_() };

vortex-array/src/arrays/bool/patch.rs

@@ -31,7 +31,7 @@ impl BoolArray {
             for (idx, value) in indices
                 .as_slice::<I>()
                 .iter()
-                .zip_eq(values.to_bit_buffer().iter())
+                .zip_eq(values.bit_buffer_view().iter())
             {
                 #[allow(clippy::cast_possible_truncation)]
                 own_values.set_to(*idx as usize - offset, value);

vortex-array/src/arrays/bool/vtable/mod.rs

@@ -53,13 +53,13 @@ pub struct BoolMetadata {
 impl ArrayHash for BoolData {
     fn array_hash<H: Hasher>(&self, state: &mut H, precision: Precision) {
         self.bits.array_hash(state, precision);
-        self.offset.hash(state);
+        self.meta.offset().hash(state);
     }
 }
 
 impl ArrayEq for BoolData {
     fn array_eq(&self, other: &Self, precision: Precision) -> bool {
-        self.offset == other.offset && self.bits.array_eq(&other.bits, precision)
+        self.meta.offset() == other.meta.offset() && self.bits.array_eq(&other.bits, precision)
     }
 }
 
@@ -96,10 +96,11 @@ impl VTable for Bool {
         array: ArrayView<'_, Self>,
         _session: &VortexSession,
     ) -> VortexResult<Option<Vec<u8>>> {
-        assert!(array.offset < 8, "Offset must be <8, got {}", array.offset);
+        let offset = array.meta.offset();
+        assert!(offset < 8, "Offset must be <8, got {offset}");
         Ok(Some(
             BoolMetadata {
-                offset: u32::try_from(array.offset).vortex_expect("checked"),
+                offset: u32::try_from(offset).vortex_expect("checked"),
             }
             .encode_to_vec(),
         ))
@@ -116,9 +117,9 @@ impl VTable for Bool {
             vortex_bail!("Expected bool dtype, got {dtype:?}");
         };
         vortex_ensure!(
-            data.bits.len() * 8 >= data.offset + len,
+            data.bits.len() * 8 >= data.meta.offset() + len,
             "BoolArray buffer with offset {} cannot back outer length {} (buffer bits = {})",
-            data.offset,
+            data.meta.offset(),
             len,
             data.bits.len() * 8
         );

vortex-array/src/arrays/bool/vtable/operations.rs

@@ -17,7 +17,7 @@ impl OperationsVTable<Bool> for Bool {
         _ctx: &mut ExecutionCtx,
     ) -> VortexResult<Scalar> {
         Ok(Scalar::bool(
-            array.to_bit_buffer().value(index),
+            array.bit_buffer_view().value(index),
             array.dtype().nullability(),
         ))
     }

vortex-array/src/arrays/dict/compute/fill_null.rs

@@ -41,7 +41,7 @@ impl FillNullKernel for Dict {
 
         // We found the fill value already in the values at this given index.
         let Some(existing_fill_value_index) =
-            found_fill_values.to_bit_buffer().set_indices().next()
+            found_fill_values.bit_buffer_view().set_indices().next()
         else {
             // No fill values found, so we must canonicalize and fill_null.
             return Ok(Some(

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

@@ -55,9 +55,10 @@ impl CompareKernel for Patched {
 
         let validity = child_to_validity(result.slots()[0].as_ref(), result.dtype().nullability());
         let len = result.len();
-        let BoolDataParts { bits, offset, len } = result.into_data().into_parts(len);
+        let BoolDataParts { bits, meta } = result.into_data().into_parts(len);
 
-        let mut bits = BitBufferMut::from_buffer(bits.unwrap_host().into_mut(), offset, len);
+        let mut bits =
+            BitBufferMut::from_buffer(bits.unwrap_host().into_mut(), meta.offset(), meta.len());
 
         let lane_offsets = lhs.lane_offsets().clone().execute::<PrimitiveArray>(ctx)?;
         let indices = lhs.patch_indices().clone().execute::<PrimitiveArray>(ctx)?;

vortex-array/src/arrays/primitive/array/mod.rs

@@ -549,7 +549,7 @@ impl PrimitiveData {
             Validity::Array(is_valid) => {
                 #[expect(deprecated)]
                 let bool_array = is_valid.to_bool();
-                let bool_buffer = bool_array.to_bit_buffer();
+                let bool_buffer = bool_array.bit_buffer_view();
                 let mut bytes = ByteBufferMut::zeroed_aligned(n_rows * byte_width, alignment);
                 for (i, valid_i) in bool_buffer.set_indices().enumerate() {
                     bytes[valid_i * byte_width..(valid_i + 1) * byte_width]

vortex-array/src/canonical.rs

@@ -563,9 +563,14 @@ impl Executable for CanonicalValidity {
             Canonical::Bool(b) => {
                 let validity = child_to_validity(b.slots()[0].as_ref(), b.dtype().nullability());
                 let len = b.len();
-                let BoolDataParts { bits, offset, len } = b.into_data().into_parts(len);
+                let BoolDataParts { bits, meta } = b.into_data().into_parts(len);
                 Ok(CanonicalValidity(Canonical::Bool(
-                    BoolArray::try_new_from_handle(bits, offset, len, validity.execute(ctx)?)?,
+                    BoolArray::try_new_from_handle(
+                        bits,
+                        meta.offset(),
+                        meta.len(),
+                        validity.execute(ctx)?,
+                    )?,
                 )))
             }
             Canonical::Primitive(p) => {
@@ -713,9 +718,14 @@ impl Executable for RecursiveCanonical {
             Canonical::Bool(b) => {
                 let validity = child_to_validity(b.slots()[0].as_ref(), b.dtype().nullability());
                 let len = b.len();
-                let BoolDataParts { bits, offset, len } = b.into_data().into_parts(len);
+                let BoolDataParts { bits, meta } = b.into_data().into_parts(len);
                 Ok(RecursiveCanonical(Canonical::Bool(
-                    BoolArray::try_new_from_handle(bits, offset, len, validity.execute(ctx)?)?,
+                    BoolArray::try_new_from_handle(
+                        bits,
+                        meta.offset(),
+                        meta.len(),
+                        validity.execute(ctx)?,
+                    )?,
                 )))
             }
             Canonical::Primitive(p) => {

vortex-array/src/scalar_fn/fns/list_contains/mod.rs

@@ -356,7 +356,7 @@ where
 {
     let offsets_slice = offsets.as_slice::<O>();
     let sizes_slice = sizes.as_slice::<S>();
-    let bits = matches.to_bit_buffer();
+    let bits = matches.bit_buffer_view();
 
     (0..list_array_len)
         .map(|i| {
@@ -365,7 +365,7 @@ where
 
             // BitIndexIterator yields indices of true bits only. If `.next()` returns
             // `Some(_)`, at least one element in this list's range matches.
-            let mut set_bits = BitIndexIterator::new(bits.inner().as_ref(), offset, size);
+            let mut set_bits = BitIndexIterator::new(bits.inner(), offset, size);
             set_bits.next().is_some()
         })
         .collect::<BitBuffer>()

vortex-buffer/src/bit/buf_mut.rs

@@ -17,7 +17,7 @@ use crate::buffer_mut;
 
 /// Sets all bits in the bit-range `[start_bit, end_bit)` of `slice` to `value`.
 #[inline(always)]
-fn fill_bits(slice: &mut [u8], start_bit: usize, end_bit: usize, value: bool) {
+pub(crate) fn fill_bits(slice: &mut [u8], start_bit: usize, end_bit: usize, value: bool) {
     if start_bit >= end_bit {
         return;
     }