@@ -28,6 +28,7 @@ prost = { workspace = true }
 rand = { workspace = true, optional = true }
 vortex-array = { workspace = true }
 vortex-buffer = { workspace = true }
+vortex-compute = { workspace = true }
 vortex-error = { workspace = true }
 vortex-mask = { workspace = true }
 vortex-session = { workspace = true }
 
@@ -11,6 +11,7 @@ use vortex_array::ExecutionCtx;
 use vortex_array::IntoArray;
 use vortex_array::arrays::ConstantArray;
 use vortex_array::arrays::PrimitiveArray;
+use vortex_array::arrays::primitive::PrimitiveArrayExt;
 use vortex_array::dtype::NativePType;
 use vortex_array::expr::stats::Stat;
 use vortex_array::match_each_integer_ptype;
@@ -19,9 +20,11 @@ use vortex_array::validity::Validity;
 use vortex_buffer::BitBuffer;
 use vortex_buffer::Buffer;
 use vortex_buffer::BufferMut;
+use vortex_compute::lane_kernels::IndexedSourceExt;
 use vortex_error::VortexResult;
 use vortex_error::vortex_err;
 use vortex_mask::AllOr;
+use vortex_mask::Mask;
 
 use crate::FL_CHUNK_SIZE;
 use crate::FoR;
@@ -53,21 +56,29 @@ impl FoRData {
     }
 }
 
+/// Subtracts `min` from every valid value. Null slots are written as zero so they cost no bits in
+/// a downstream bit-packing, and the validity bit selects the zero without a branch per value.
 fn encode_primitive<T: NativePType + WrappingSub + PrimInt>(
     parray: PrimitiveArray,
     min: T,
     ctx: &mut ExecutionCtx,
 ) -> VortexResult<PrimitiveArray> {
-    // Set null values to the min value, ensuring that decompress into a value in the primitive
-    // range (and stop them wrapping around).
-    let encoded = parray.map_each_with_validity::<T, _, _>(ctx, |(v, bool)| {
-        if bool {
-            v.wrapping_sub(&min)
-        } else {
-            T::zero()
-        }
-    })?;
-    Ok(encoded)
+    let validity = parray.validity()?;
+    let len = parray.len();
+    let values = parray.as_slice::<T>();
+    let subtract = |v: T| v.wrapping_sub(&min);
+
+    let mut encoded = BufferMut::<T>::with_capacity_in(len, ctx.allocator().clone());
+    let out = &mut encoded.spare_capacity_mut()[..len];
+    match validity.execute_mask(len, ctx)? {
+        Mask::AllTrue(_) => values.map_into(out, subtract),
+        Mask::AllFalse(_) => out.fill(MaybeUninit::new(T::zero())),
+        Mask::Values(mask) => values.map_masked_into(mask.bit_buffer(), out, subtract),
+    }
+    // SAFETY: each branch writes every lane of `out`, which spans exactly `len` items.
+    unsafe { encoded.set_len(len) };
+
+    Ok(PrimitiveArray::new(encoded.freeze(), validity))
 }
 
 fn encode_chunked_typed<T: NativePType + WrappingSub + PrimInt>(
@@ -331,6 +342,30 @@ mod test {
         assert_eq!(encoded, Scalar::from(0i32));
     }
 
+    #[test]
+    fn test_compress_nullable_zeroes_null_slots() -> VortexResult<()> {
+        let mut ctx = SESSION.create_execution_ctx();
+        let array = PrimitiveArray::from_option_iter(
+            (0..200i32).map(|i| (i % 3 != 0).then_some(1_000 + i)),
+        );
+        let compressed = FoRData::encode(array.clone(), &mut ctx)?;
+        let reference = compressed
+            .constant_reference()
+            .ok_or_else(|| vortex_err!("expected a constant reference"))?;
+        assert_eq!(i32::try_from(&reference)?, 1_001);
+
+        let encoded = compressed
+            .encoded()
+            .clone()
+            .execute::<PrimitiveArray>(&mut ctx)?;
+        let expected: Vec<i32> = (0..200i32)
+            .map(|i| if i % 3 != 0 { i - 1 } else { 0 })
+            .collect();
+        assert_eq!(encoded.as_slice::<i32>(), expected.as_slice());
+        assert_arrays_eq!(compressed, array, &mut ctx);
+        Ok(())
+    }
+
     #[test]
     fn test_overflow() -> VortexResult<()> {
         let mut ctx = SESSION.create_execution_ctx();
 
@@ -26,6 +26,8 @@ use vortex_array::scalar::Scalar;
 use vortex_buffer::Buffer;
 use vortex_buffer::BufferAllocatorRef;
 use vortex_buffer::BufferMut;
+use vortex_compute::lane_kernels::IndexedSinkExt;
+use vortex_compute::lane_kernels::IndexedSourceExt;
 use vortex_error::VortexExpect;
 use vortex_error::VortexResult;
 use vortex_error::vortex_err;
@@ -56,7 +58,7 @@ fn decompress_one_ref(
         return fused_decompress(array, bp, ctx);
     }
 
-    add_reference(array, reference)
+    add_reference(array, reference, ctx)
 }
 
 /// Unpack a BitPacked child and add the constant reference in one pass.
@@ -88,15 +90,20 @@ fn fused_decompress_typed<
 }
 
 /// Decode `encoded`, then add `reference` to every value.
-fn add_reference(array: &FoRArray, reference: &Scalar) -> VortexResult<PrimitiveArray> {
+fn add_reference(
+    array: &FoRArray,
+    reference: &Scalar,
+    ctx: &mut ExecutionCtx,
+) -> VortexResult<PrimitiveArray> {
     match_each_integer_ptype!(array.ptype(), |T| {
-        add_reference_typed::<T>(array, reference)
+        add_reference_typed::<T>(array, reference, ctx)
     })
 }
 
 fn add_reference_typed<T: NativePType + WrappingAdd + PrimInt>(
     array: &FoRArray,
     reference: &Scalar,
+    ctx: &mut ExecutionCtx,
 ) -> VortexResult<PrimitiveArray> {
     let encoded = array.encoded().as_::<Primitive>().into_owned();
     let min = reference
@@ -108,18 +115,38 @@ fn add_reference_typed<T: NativePType + WrappingAdd + PrimInt>(
     }
     let validity = encoded.validity()?;
     Ok(PrimitiveArray::new(
-        decompress_primitive(encoded.into_buffer::<T>(), min),
+        decompress_primitive(encoded.try_into_buffer_mut::<T>(), min, ctx),
         validity,
     ))
 }
 
+/// Adds `min` to every value. A uniquely owned buffer is mapped in place, a shared one is mapped
+/// into a new allocation.
 fn decompress_primitive<T: NativePType + WrappingAdd + PrimInt>(
-    values: Buffer<T>,
+    values: Result<BufferMut<T>, Buffer<T>>,
     min: T,
+    ctx: &mut ExecutionCtx,
 ) -> Buffer<T> {
-    values
-        .map_each_in_place(move |v| v.wrapping_add(&min))
-        .freeze()
+    let add = |v: T| v.wrapping_add(&min);
+
+    match values {
+        Ok(mut values) => {
+            values.as_mut_slice().map_into_in_place(add);
+            values.freeze()
+        }
+        Err(values) => {
+            let len = values.len();
+            let mut decoded = BufferMut::<T>::with_capacity_in(len, ctx.allocator().clone());
+            values
+                .as_slice()
+                .map_into(&mut decoded.spare_capacity_mut()[..len], add);
+
+            // SAFETY: `map_into` writes every lane of the `len` items.
+            unsafe { decoded.set_len(len) };
+
+            decoded.freeze()
+        }
+    }
 }
 
 /// Decompress an array whose chunks have different references.
 
@@ -16,6 +16,7 @@ version = { workspace = true }
 [dependencies]
 vortex-array = { workspace = true }
 vortex-buffer = { workspace = true }
+vortex-compute = { workspace = true }
 vortex-error = { workspace = true }
 vortex-mask = { workspace = true }
 vortex-session = { workspace = true }
 
@@ -9,6 +9,8 @@ use vortex_array::dtype::NativePType;
 use vortex_array::dtype::PType;
 use vortex_array::validity::Validity;
 use vortex_buffer::BufferMut;
+use vortex_compute::lane_kernels::IndexedSinkExt;
+use vortex_compute::lane_kernels::ReinterpretSink;
 use vortex_error::VortexExpect;
 use vortex_error::VortexResult;
 use vortex_error::vortex_bail;
@@ -34,16 +36,19 @@ pub fn zigzag_encode(parray: ArrayView<'_, Primitive>) -> VortexResult<ZigZagArr
 }
 
 fn zigzag_encode_primitive<T: ExternalZigZag + NativePType>(
-    values: BufferMut<T>,
+    mut values: BufferMut<T>,
     validity: Validity,
 ) -> PrimitiveArray
 where
     <T as ExternalZigZag>::UInt: NativePType,
 {
-    PrimitiveArray::new(
-        values.map_each_in_place(|v| T::encode(v)).freeze(),
-        validity,
-    )
+    ReinterpretSink::<T, T::UInt>::new(values.as_mut_slice()).map_into_in_place(T::encode);
+
+    // SAFETY: `ReinterpretSink::new` checks that both types have the same size and alignment,
+    // and `map_into_in_place` writes a valid value into every lane.
+    let encoded = unsafe { values.transmute::<T::UInt>() };
+
+    PrimitiveArray::new(encoded.freeze(), validity)
 }
 
 pub fn zigzag_decode(parray: PrimitiveArray) -> PrimitiveArray {
@@ -63,16 +68,19 @@ pub fn zigzag_decode(parray: PrimitiveArray) -> PrimitiveArray {
 }
 
 fn zigzag_decode_primitive<T: ExternalZigZag + NativePType>(
-    values: BufferMut<T::UInt>,
+    mut values: BufferMut<T::UInt>,
     validity: Validity,
 ) -> PrimitiveArray
 where
     <T as ExternalZigZag>::UInt: NativePType,
 {
-    PrimitiveArray::new(
-        values.map_each_in_place(|v| T::decode(v)).freeze(),
-        validity,
-    )
+    ReinterpretSink::<T::UInt, T>::new(values.as_mut_slice()).map_into_in_place(T::decode);
+
+    // SAFETY: `ReinterpretSink::new` checks that both types have the same size and alignment,
+    // and `map_into_in_place` writes a valid value into every lane.
+    let decoded = unsafe { values.transmute::<T>() };
+
+    PrimitiveArray::new(decoded.freeze(), validity)
 }
 
 #[cfg(test)]
 
@@ -3,7 +3,6 @@
 
 use std::fmt::Display;
 use std::fmt::Formatter;
-use std::iter::repeat;
 
 use smallvec::smallvec;
 use vortex_buffer::Alignment;
@@ -486,31 +485,6 @@ impl Array<Primitive> {
             validity,
         }
     }
-
-    pub fn map_each_with_validity<T, R, F>(self, ctx: &mut ExecutionCtx, f: F) -> VortexResult<Self>
-    where
-        T: NativePType,
-        R: NativePType,
-        F: FnMut((T, bool)) -> R,
-    {
-        let validity = PrimitiveArrayExt::validity(&self);
-        let data = self.into_data();
-        let buf_iter = data.to_buffer::<T>().into_iter();
-
-        let buffer = match &validity {
-            Validity::NonNullable | Validity::AllValid => {
-                Buffer::<R>::from_trusted_len_iter(buf_iter.zip(repeat(true)).map(f))
-            }
-            Validity::AllInvalid => {
-                Buffer::<R>::from_trusted_len_iter(buf_iter.zip(repeat(false)).map(f))
-            }
-            Validity::Array(val) => {
-                let val = val.clone().execute::<BoolArray>(ctx)?.into_bit_buffer();
-                Buffer::<R>::from_trusted_len_iter(buf_iter.zip(val.iter()).map(f))
-            }
-        };
-        Ok(PrimitiveArray::new(buffer, validity))
-    }
 }
 
 impl PrimitiveData {
 
@@ -12,7 +12,6 @@ use num_traits::PrimInt;
 use vortex_buffer::Buffer;
 use vortex_buffer::BufferMut;
 use vortex_error::VortexExpect;
-use vortex_error::vortex_err;
 
 #[global_allocator]
 static GLOBAL: MiMalloc = MiMalloc;
@@ -47,66 +46,6 @@ fn from_iter<B: FromIterator<i32>>(n: i32) {
     B::from_iter((0..n).map(|i| i % i32::MAX));
 }
 
-trait MapEach<T, R> {
-    type Output;
-
-    fn map_each<F>(self, f: F) -> Self::Output
-    where
-        F: FnMut(T) -> R;
-}
-
-impl<T: ArrowNativeType, R: ArrowNativeType> MapEach<T, R> for Arrow<ScalarBuffer<T>> {
-    type Output = Arrow<ScalarBuffer<R>>;
-
-    fn map_each<F>(self, f: F) -> Self::Output
-    where
-        F: FnMut(T) -> R,
-    {
-        Arrow(ScalarBuffer::from(
-            self.0
-                .into_inner()
-                .into_vec::<T>()
-                .map_err(|_| vortex_err!("Failed to convert Arrow buffer into a mut vec"))
-                .vortex_expect("Failed to convert Arrow buffer into a mut vec")
-                .into_iter()
-                .map(f)
-                .collect::<Vec<R>>(),
-        ))
-    }
-}
-
-impl<T: Copy, R> MapEach<T, R> for Buffer<T> {
-    type Output = BufferMut<R>;
-
-    fn map_each<F>(self, f: F) -> Self::Output
-    where
-        F: FnMut(T) -> R,
-    {
-        Buffer::<T>::map_each_in_place(self, f)
-    }
-}
-
-impl<T: Copy, R> MapEach<T, R> for BufferMut<T> {
-    type Output = BufferMut<R>;
-
-    fn map_each<F>(self, f: F) -> Self::Output
-    where
-        F: FnMut(T) -> R,
-    {
-        BufferMut::<T>::map_each_in_place(self, f)
-    }
-}
-
-#[divan::bench(
-    types = [Arrow<ScalarBuffer<i32>>, Buffer<i32>, BufferMut<i32>],
-    args = INPUT_SIZE,
-)]
-fn map_each<B: MapEach<i32, u32> + FromIterator<i32>>(bencher: Bencher, n: i32) {
-    bencher
-        .with_inputs(|| B::from_iter((0..n).map(|i| i % i32::MAX)))
-        .bench_values(|buffer| B::map_each(buffer, |i| (i as u32) + 1));
-}
-
 /// Element-at-a-time pushes cost tens of nanoseconds each, so cap the top size to keep the
 /// benchmark runtime bounded.
 const PUSH_INPUT_SIZE: &[i32] = &[128, 1024, 2048, 16_384, 32_768];