switch Buffer -> BufferHandle in PrimitiveArray

encodings/alp/src/alp/array.rs

@@ -718,7 +718,7 @@ mod tests {
             );
 
             if let Some(expected_val) = expected_value {
-                let buf = result_primitive.buffer::<f64>();
+                let buf = result_primitive.to_buffer::<f64>();
                 let result_val = buf.as_slice()[idx];
                 assert_eq!(result_val, expected_val, "Value mismatch at idx={idx}",);
             }

encodings/bytebool/src/array.rs

@@ -97,7 +97,7 @@ impl VTable for ByteBoolVTable {
         if buffers.len() != 1 {
             vortex_bail!("Expected 1 buffer, got {}", buffers.len());
         }
-        let buffer = buffers[0].clone().try_to_bytes()?;
+        let buffer = buffers[0].clone().try_to_host()?;
 
         Ok(ByteBoolArray::new(buffer, validity))
     }

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

@@ -238,7 +238,7 @@ impl CanonicalVTable<DecimalBytePartsVTable> for DecimalBytePartsVTable {
             // The decimal dtype matches the array's dtype, and validity is preserved.
             Canonical::Decimal(unsafe {
                 DecimalArray::new_unchecked(
-                    prim.buffer::<P>(),
+                    prim.to_buffer::<P>(),
                     *array.decimal_dtype(),
                     prim.validity().clone(),
                 )

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

@@ -176,7 +176,7 @@ impl VTable for BitPackedVTable {
         if buffers.len() != 1 {
             vortex_bail!("Expected 1 buffer, got {}", buffers.len());
         }
-        let packed = buffers[0].clone().try_to_bytes()?;
+        let packed = buffers[0].clone().try_to_host()?;
 
         let load_validity = |child_idx: usize| {
             if children.len() == child_idx {

encodings/fsst/src/array.rs

@@ -116,8 +116,8 @@ impl VTable for FSSTVTable {
         if buffers.len() != 2 {
             vortex_bail!(InvalidArgument: "Expected 2 buffers, got {}", buffers.len());
         }
-        let symbols = Buffer::<Symbol>::from_byte_buffer(buffers[0].clone().try_to_bytes()?);
-        let symbol_lengths = Buffer::<u8>::from_byte_buffer(buffers[1].clone().try_to_bytes()?);
+        let symbols = Buffer::<Symbol>::from_byte_buffer(buffers[0].clone().try_to_host()?);
+        let symbol_lengths = Buffer::<u8>::from_byte_buffer(buffers[1].clone().try_to_host()?);
 
         if children.len() != 2 {
             vortex_bail!(InvalidArgument: "Expected 2 children, got {}", children.len());

encodings/fsst/src/kernel.rs

@@ -86,7 +86,7 @@ impl ExecuteParentKernel<FSSTVTable> for FSSTFilterKernel {
             .cast(DType::Primitive(PType::U32, Nullability::NonNullable))?
             .execute::<Canonical>(ctx)?
             .into_primitive()
-            .buffer::<u32>();
+            .to_buffer::<u32>();
 
         let decompressor = array.decompressor();
 
@@ -97,7 +97,7 @@ impl ExecuteParentKernel<FSSTVTable> for FSSTFilterKernel {
                 &codes_offsets,
                 mask_values,
                 &validity,
-                &uncompressed_lens.buffer::<S>(),
+                &uncompressed_lens.to_buffer::<S>(),
             )
         });
 

encodings/pco/src/array.rs

@@ -141,11 +141,11 @@ impl VTable for PcoVTable {
         vortex_ensure!(buffers.len() >= metadata.0.chunks.len());
         let chunk_metas = buffers[..metadata.0.chunks.len()]
             .iter()
-            .map(|b| b.clone().try_to_bytes())
+            .map(|b| b.clone().try_to_host())
             .collect::<VortexResult<Vec<_>>>()?;
         let pages = buffers[metadata.0.chunks.len()..]
             .iter()
-            .map(|b| b.clone().try_to_bytes())
+            .map(|b| b.clone().try_to_host())
             .collect::<VortexResult<Vec<_>>>()?;
 
         let expected_n_pages = metadata
@@ -293,7 +293,7 @@ impl PcoArray {
                 number_type,
                 NumberType<T> => {
                     let chunk_end = cmp::min(n_values, chunk_start + values_per_chunk);
-                    let values = values.buffer::<T>();
+                    let values = values.to_buffer::<T>();
                     let chunk = &values.as_slice()[chunk_start..chunk_end];
                     fc
                         .chunk_compressor(chunk, &chunk_config)

encodings/sparse/src/canonical.rs

@@ -449,7 +449,7 @@ fn canonicalize_varbin(
     let len = array.len();
 
     match_each_integer_ptype!(indices.ptype(), |I| {
-        let indices = indices.buffer::<I>();
+        let indices = indices.to_buffer::<I>();
         canonicalize_varbin_inner::<I>(fill_value, indices, values, dtype, validity, len)
     })
 }

encodings/sparse/src/lib.rs

@@ -137,7 +137,7 @@ impl VTable for SparseVTable {
         }
         let fill_value = Scalar::new(
             dtype.clone(),
-            ScalarValue::from_protobytes(&buffers[0].clone().try_to_bytes()?)?,
+            ScalarValue::from_protobytes(&buffers[0].clone().try_to_host()?)?,
         );
 
         SparseArray::try_new(patch_indices, patch_values, len, fill_value)

encodings/zstd/src/array.rs

@@ -142,16 +142,16 @@ impl VTable for ZstdVTable {
                 None,
                 buffers
                     .iter()
-                    .map(|b| b.clone().try_to_bytes())
+                    .map(|b| b.clone().try_to_host())
                     .collect::<VortexResult<Vec<_>>>()?,
             )
         } else {
             // with dictionary
             (
-                Some(buffers[0].clone().try_to_bytes()?),
+                Some(buffers[0].clone().try_to_host()?),
                 buffers[1..]
                     .iter()
-                    .map(|b| b.clone().try_to_bytes())
+                    .map(|b| b.clone().try_to_host())
                     .collect::<VortexResult<Vec<_>>>()?,
             )
         };
@@ -366,7 +366,7 @@ impl ZstdArray {
             n_values
         };
 
-        let value_bytes = values.byte_buffer();
+        let value_bytes = values.buffer_handle().try_to_host()?;
         // Align frames to buffer alignment. This is necessary for overaligned buffers.
         let alignment = *value_bytes.alignment();
         let step_width = (values_per_frame * byte_width).div_ceil(alignment) * alignment;
@@ -379,7 +379,7 @@ impl ZstdArray {
             frames,
             frame_metas,
         } = Self::compress_values(
-            value_bytes,
+            &value_bytes,
             &frame_byte_starts,
             level,
             values_per_frame,

fuzz/src/array/fill_null.rs

@@ -90,12 +90,9 @@ fn fill_primitive_array(
             .vortex_expect("fill value conversion should succeed in fuzz test");
 
         match array.validity() {
-            Validity::NonNullable | Validity::AllValid => PrimitiveArray::from_byte_buffer(
-                array.byte_buffer().clone(),
-                array.ptype(),
-                result_nullability.into(),
-            )
-            .into_array(),
+            Validity::NonNullable | Validity::AllValid => {
+                PrimitiveArray::new(array.to_buffer::<T>(), result_nullability.into()).into_array()
+            }
             Validity::AllInvalid => {
                 ConstantArray::new(fill_value.clone(), array.len()).into_array()
             }

fuzz/src/array/mask.rs

@@ -36,8 +36,8 @@ pub fn mask_canonical_array(canonical: Canonical, mask: &Mask) -> VortexResult<A
         }
         Canonical::Primitive(array) => {
             let new_validity = array.validity().mask(mask);
-            PrimitiveArray::from_byte_buffer(
-                array.byte_buffer().clone(),
+            PrimitiveArray::from_buffer_handle(
+                array.buffer_handle().clone(),
                 array.ptype(),
                 new_validity,
             )

fuzz/src/array/slice.rs

@@ -41,10 +41,11 @@ pub fn slice_canonical_array(
         DType::Primitive(p, _) => {
             let primitive_array = array.to_primitive();
             match_each_native_ptype!(p, |P| {
-                Ok(
-                    PrimitiveArray::new(primitive_array.buffer::<P>().slice(start..stop), validity)
-                        .into_array(),
+                Ok(PrimitiveArray::new(
+                    primitive_array.to_buffer::<P>().slice(start..stop),
+                    validity,
                 )
+                .into_array())
             })
         }
         DType::Utf8(_) | DType::Binary(_) => {

vortex-array/src/array/visitor.rs

@@ -11,6 +11,7 @@ use crate::Array;
 use crate::ArrayRef;
 use crate::IntoArray;
 use crate::arrays::ConstantArray;
+use crate::buffer::BufferHandle;
 use crate::patches::Patches;
 use crate::validity::Validity;
 
@@ -113,6 +114,10 @@ pub trait ArrayVisitorExt: Array {
 impl<A: Array + ?Sized> ArrayVisitorExt for A {}
 
 pub trait ArrayBufferVisitor {
+    fn visit_buffer_handle(&mut self, handle: &BufferHandle) -> VortexResult<()> {
+        self.visit_buffer(&handle.clone().try_to_host()?);
+        Ok(())
+    }
     fn visit_buffer(&mut self, buffer: &ByteBuffer);
 }
 

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

@@ -103,7 +103,7 @@ impl VTable for BoolVTable {
             vortex_bail!("Expected 0 or 1 child, got {}", children.len());
         };
 
-        let buffer = buffers[0].clone().try_to_bytes()?;
+        let buffer = buffers[0].clone().try_to_host()?;
         let bits = BitBuffer::new_with_offset(buffer, len, metadata.offset as usize);
 
         BoolArray::try_new(bits, validity)

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

@@ -117,7 +117,7 @@ impl ChunkedArray {
 
     #[inline]
     pub fn chunk_offsets(&self) -> Buffer<u64> {
-        self.chunk_offsets.buffer()
+        self.chunk_offsets.to_buffer()
     }
 
     pub(crate) fn find_chunk_idx(&self, index: usize) -> (usize, usize) {

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

@@ -98,7 +98,7 @@ impl VTable for ChunkedVTable {
             )?
             .to_primitive();
 
-        let chunk_offsets_buf = chunk_offsets_array.buffer::<u64>();
+        let chunk_offsets_buf = chunk_offsets_array.to_buffer::<u64>();
 
         // The remaining children contain the actual data of the chunks
         let chunks = chunk_offsets_buf
@@ -139,7 +139,7 @@ impl VTable for ChunkedVTable {
 
         let nchunks = children.len() - 1;
         let chunk_offsets_array = children[0].to_primitive();
-        let chunk_offsets_buf = chunk_offsets_array.buffer::<u64>();
+        let chunk_offsets_buf = chunk_offsets_array.to_buffer::<u64>();
 
         vortex_ensure!(
             chunk_offsets_buf.len() == nchunks + 1,

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

@@ -81,7 +81,7 @@ impl VTable for ConstantVTable {
         if buffers.len() != 1 {
             vortex_bail!("Expected 1 buffer, got {}", buffers.len());
         }
-        let buffer = buffers[0].clone().try_to_bytes()?;
+        let buffer = buffers[0].clone().try_to_host()?;
         let sv = ScalarValue::from_protobytes(&buffer)?;
         let scalar = Scalar::new(dtype.clone(), sv);
         Ok(ConstantArray::new(scalar, len))

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

@@ -95,7 +95,7 @@ impl VTable for DecimalVTable {
         if buffers.len() != 1 {
             vortex_bail!("Expected 1 buffer, got {}", buffers.len());
         }
-        let buffer = buffers[0].clone().try_to_bytes()?;
+        let buffer = buffers[0].clone().try_to_host()?;
 
         let validity = if children.is_empty() {
             Validity::from(dtype.nullability())

vortex-array/src/arrays/extension/vtable/rules.rs

@@ -104,7 +104,7 @@ mod tests {
         assert_eq!(ext_result.ext_dtype().as_ref(), ext_dtype.as_ref());
 
         // Check the storage values
-        let storage_result: &[i64] = &ext_result.storage().to_primitive().buffer::<i64>();
+        let storage_result: &[i64] = &ext_result.storage().to_primitive().to_buffer::<i64>();
         assert_eq!(storage_result, &[1, 3, 5]);
     }
 

vortex-array/src/arrays/masked/execute.rs

@@ -66,7 +66,14 @@ fn mask_validity_primitive(array: PrimitiveArray, mask: &Mask) -> PrimitiveArray
     let len = array.len();
     let ptype = array.ptype();
     let new_validity = combine_validity(array.validity(), mask, len);
-    PrimitiveArray::from_byte_buffer(array.into_byte_buffer(), ptype, new_validity)
+    // SAFETY: validity has same length as values
+    unsafe {
+        PrimitiveArray::new_unchecked_from_handle(
+            array.buffer_handle().clone(),
+            ptype,
+            new_validity,
+        )
+    }
 }
 
 fn mask_validity_decimal(array: DecimalArray, mask: &Mask) -> DecimalArray {

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

@@ -5,6 +5,7 @@ use vortex_buffer::Buffer;
 use vortex_dtype::DType;
 use vortex_dtype::NativePType;
 use vortex_dtype::PType;
+use vortex_error::VortexExpect;
 use vortex_error::VortexResult;
 use vortex_error::vortex_panic;
 
@@ -18,6 +19,10 @@ impl PrimitiveArray {
     /// Return a slice of the array's buffer.
     ///
     /// NOTE: these values may be nonsense if the validity buffer indicates that the value is null.
+    ///
+    /// # Panic
+    ///
+    /// This operation will panic if the array is not backed by host memory.
     pub fn as_slice<T: NativePType>(&self) -> &[T] {
         if T::PTYPE != self.ptype() {
             vortex_panic!(
@@ -26,11 +31,15 @@ impl PrimitiveArray {
                 self.ptype()
             )
         }
-        let raw_slice = self.byte_buffer().as_ptr();
+
+        let byte_buffer = self
+            .buffer
+            .as_host_opt()
+            .vortex_expect("as_slice must be called on host buffer");
+        let raw_slice = byte_buffer.as_ptr();
+
         // SAFETY: alignment of Buffer is checked on construction
-        unsafe {
-            std::slice::from_raw_parts(raw_slice.cast(), self.byte_buffer().len() / size_of::<T>())
-        }
+        unsafe { std::slice::from_raw_parts(raw_slice.cast(), byte_buffer.len() / size_of::<T>()) }
     }
 
     pub fn reinterpret_cast(&self, ptype: PType) -> Self {
@@ -44,7 +53,11 @@ impl PrimitiveArray {
             "can't reinterpret cast between integers of two different widths"
         );
 
-        PrimitiveArray::from_byte_buffer(self.byte_buffer().clone(), ptype, self.validity().clone())
+        PrimitiveArray::from_buffer_handle(
+            self.buffer_handle().clone(),
+            ptype,
+            self.validity().clone(),
+        )
     }
 
     /// Narrow the array to the smallest possible integer type that can represent all values.