diff --git a/encodings/alp/benches/alp_compress.rs b/encodings/alp/benches/alp_compress.rs
index 02cc30e44a6..6e769ffc22e 100644
--- a/encodings/alp/benches/alp_compress.rs
+++ b/encodings/alp/benches/alp_compress.rs
@@ -146,8 +146,8 @@ fn compress_rd<T: ALPRDFloat + NativePType>(bencher: Bencher, args: (usize, f64)
     let encoder = RDEncoder::new(primitive.as_slice::<T>());
 
     bencher
-        .with_inputs(|| (&primitive, &encoder, SESSION.create_execution_ctx()))
-        .bench_refs(|(primitive, encoder, ctx)| encoder.encode(primitive.as_view(), ctx))
+        .with_inputs(|| (&primitive, &encoder))
+        .bench_refs(|(primitive, encoder)| encoder.encode(primitive.as_view()))
 }
 
 #[divan::bench(types = [f32, f64], args = RD_BENCH_ARGS)]
@@ -155,7 +155,7 @@ fn decompress_rd<T: ALPRDFloat + NativePType>(bencher: Bencher, args: (usize, f6
     let (n, fraction_patch) = args;
     let primitive = make_rd_array::<T>(n, fraction_patch);
     let encoder = RDEncoder::new(primitive.as_slice::<T>());
-    let encoded = encoder.encode(primitive.as_view(), &mut SESSION.create_execution_ctx());
+    let encoded = encoder.encode(primitive.as_view());
 
     bencher
         .with_inputs(|| (&encoded, SESSION.create_execution_ctx()))
diff --git a/encodings/alp/src/alp/array.rs b/encodings/alp/src/alp/array.rs
index 8e581fe51fc..8c0bc346f0f 100644
--- a/encodings/alp/src/alp/array.rs
+++ b/encodings/alp/src/alp/array.rs
@@ -311,10 +311,15 @@ impl ALPData {
 
         let expected_type = DType::Primitive(T::PTYPE, encoded.dtype().nullability());
         vortex_ensure!(
-            patches.dtype() == &expected_type,
-            "Expected patches type {expected_type}, actual {}",
+            patches.dtype().eq_ignore_nullability(&expected_type),
+            "Expected patches type {expected_type} (ignoring nullability), actual {}",
             patches.dtype(),
         );
+        vortex_ensure!(
+            patches.values().validity()?.definitely_no_nulls(),
+            "ALP patch values must contain no nulls, got {}",
+            patches.values(),
+        );
 
         Ok(())
     }
diff --git a/encodings/alp/src/alp/compute/mask.rs b/encodings/alp/src/alp/compute/mask.rs
index 796b0fa0813..06859fed814 100644
--- a/encodings/alp/src/alp/compute/mask.rs
+++ b/encodings/alp/src/alp/compute/mask.rs
@@ -36,16 +36,13 @@ impl MaskKernel for ALP {
     ) -> VortexResult<Option<ArrayRef>> {
         let vortex_mask = Validity::Array(mask.not()?).execute_mask(array.len(), ctx)?;
         let masked_encoded = array.encoded().clone().mask(mask.clone())?;
-        let masked_dtype = array
-            .dtype()
-            .with_nullability(masked_encoded.dtype().nullability());
+        // Patch values are concrete floats with no nulls; `Patches::mask` filters them and
+        // preserves that, so no cast is needed.
         let masked_patches = array
             .patches()
             .map(|p| p.mask(&vortex_mask, ctx))
             .transpose()?
-            .flatten()
-            .map(|patches| patches.cast_values(&masked_dtype))
-            .transpose()?;
+            .flatten();
         Ok(Some(
             ALP::new(masked_encoded, array.exponents(), masked_patches).into_array(),
         ))
@@ -99,7 +96,7 @@ mod test {
     }
 
     #[test]
-    fn test_mask_alp_with_patches_casts_surviving_patch_values_to_nullable() {
+    fn test_mask_alp_with_patches_keeps_patch_values_without_nulls() {
         let mut ctx = array_session().create_execution_ctx();
         let values = PrimitiveArray::from_iter([1.234f32, f32::NAN, 2.345, f32::INFINITY, 3.456]);
         let alp = alp_encode(values.as_view(), None, &mut ctx).unwrap();
@@ -113,7 +110,15 @@ mod test {
         let masked_alp = masked.as_opt::<crate::ALP>().unwrap();
         let masked_patches = masked_alp.patches().unwrap();
 
+        // Masking introduces nulls into the array, but the surviving patch values stay concrete
+        // (no cast), so they remain free of nulls.
         assert_eq!(masked.dtype().nullability(), Nullability::Nullable);
-        assert_eq!(masked_patches.dtype().nullability(), Nullability::Nullable);
+        assert!(
+            masked_patches
+                .values()
+                .validity()
+                .unwrap()
+                .definitely_no_nulls()
+        );
     }
 }
diff --git a/encodings/alp/src/alp/compute/take.rs b/encodings/alp/src/alp/compute/take.rs
index d7673455de4..ee54e1007fb 100644
--- a/encodings/alp/src/alp/compute/take.rs
+++ b/encodings/alp/src/alp/compute/take.rs
@@ -19,19 +19,14 @@ impl TakeExecute for ALP {
         ctx: &mut ExecutionCtx,
     ) -> VortexResult<Option<ArrayRef>> {
         let taken_encoded = array.encoded().take(indices.clone())?;
+        // Patch values are concrete floats with no nulls; `Patches::take` preserves that, so no
+        // cast is needed — `ALP::validate_patches` checks the dtype (ignoring nullability) and
+        // `definitely_no_nulls`.
         let taken_patches = array
             .patches()
             .map(|p| p.take(indices, ctx))
             .transpose()?
-            .flatten()
-            .map(|patches| {
-                patches.cast_values(
-                    &array
-                        .dtype()
-                        .with_nullability(taken_encoded.dtype().nullability()),
-                )
-            })
-            .transpose()?;
+            .flatten();
         Ok(Some(
             ALP::new(taken_encoded, array.exponents(), taken_patches).into_array(),
         ))
diff --git a/encodings/alp/src/alp_rd/array.rs b/encodings/alp/src/alp_rd/array.rs
index 160723e83b2..a4e16f5ba09 100644
--- a/encodings/alp/src/alp_rd/array.rs
+++ b/encodings/alp/src/alp_rd/array.rs
@@ -17,14 +17,11 @@ use vortex_array::ArrayParts;
 use vortex_array::ArrayRef;
 use vortex_array::ArraySlots;
 use vortex_array::ArrayView;
-use vortex_array::Canonical;
 use vortex_array::EqMode;
 use vortex_array::ExecutionCtx;
 use vortex_array::ExecutionResult;
 use vortex_array::IntoArray;
-use vortex_array::LEGACY_SESSION;
 use vortex_array::TypedArrayRef;
-use vortex_array::VortexSessionExecute;
 use vortex_array::arrays::Primitive;
 use vortex_array::arrays::PrimitiveArray;
 use vortex_array::buffer::BufferHandle;
@@ -209,13 +206,7 @@ impl VTable for ALPRD {
                 )
             })
             .transpose()?;
-        // NOTE: `VTable::deserialize` has a fixed trait signature without `ExecutionCtx`, so we
-        // cannot plumb a ctx in here. We construct a legacy ctx locally at this trait boundary.
-        let left_parts_patches = ALPRDData::canonicalize_patches(
-            &left_parts,
-            left_parts_patches,
-            &mut LEGACY_SESSION.create_execution_ctx(),
-        )?;
+        let left_parts_patches = ALPRDData::validate_patches(&left_parts, left_parts_patches)?;
         let slots = ALPRDData::make_slots(&left_parts, &right_parts, left_parts_patches.as_ref());
         let data = ALPRDData::new(
             left_parts_dictionary,
@@ -368,11 +359,9 @@ impl ALPRD {
         right_parts: ArrayRef,
         right_bit_width: u8,
         left_parts_patches: Option<Patches>,
-        ctx: &mut ExecutionCtx,
     ) -> VortexResult<ALPRDArray> {
         let len = left_parts.len();
-        let left_parts_patches =
-            ALPRDData::canonicalize_patches(&left_parts, left_parts_patches, ctx)?;
+        let left_parts_patches = ALPRDData::validate_patches(&left_parts, left_parts_patches)?;
         let slots = ALPRDData::make_slots(&left_parts, &right_parts, left_parts_patches.as_ref());
         let data = ALPRDData::new(left_parts_dictionary, right_bit_width, left_parts_patches);
         Array::try_from_parts(ArrayParts::new(ALPRD, dtype, len, data).with_slots(slots))
@@ -400,26 +389,31 @@ impl ALPRD {
 }
 
 impl ALPRDData {
-    fn canonicalize_patches(
+    /// Validate that `left_parts_patches` are well-formed for an `ALPRDArray` without performing
+    /// any execution.
+    ///
+    /// Patch values must have the same dtype as `left_parts` (ignoring nullability) and contain no
+    /// nulls. Both are checked statically — `definitely_no_nulls` requires no execution — so the
+    /// values never need to be cast. Callers are responsible for constructing patches with the
+    /// correct value dtype (the encoder, `deserialize`, and the compute kernels all already do so).
+    fn validate_patches(
         left_parts: &ArrayRef,
         left_parts_patches: Option<Patches>,
-        ctx: &mut ExecutionCtx,
     ) -> VortexResult<Option<Patches>> {
-        left_parts_patches
-            .map(|patches| {
-                if !patches.values().all_valid(ctx)? {
-                    vortex_bail!("patches must be all valid: {}", patches.values());
-                }
-                // TODO(ngates): assert the DType, don't cast it.
-                // TODO(joe): assert the DType, don't cast it in the next PR.
-                let mut patches = patches.cast_values(&left_parts.dtype().as_nonnullable())?;
-                // Force execution of the lazy cast so patch values are materialized
-                // before serialization.
-                let canonical = patches.values().clone().execute::<Canonical>(ctx)?;
-                *patches.values_mut() = canonical.into_array();
-                Ok(patches)
-            })
-            .transpose()
+        if let Some(patches) = &left_parts_patches {
+            vortex_ensure!(
+                patches.dtype().eq_ignore_nullability(left_parts.dtype()),
+                "ALPRD patch values dtype {} must match left-parts dtype {}",
+                patches.dtype(),
+                left_parts.dtype(),
+            );
+            vortex_ensure!(
+                patches.values().validity()?.definitely_no_nulls(),
+                "ALPRD patch values must contain no nulls, got {}",
+                patches.values(),
+            );
+        }
+        Ok(left_parts_patches)
     }
 
     /// Build a new `ALPRDArray` from components.
@@ -555,11 +549,9 @@ fn validate_parts(
             left_parts.dtype(),
         );
         vortex_ensure!(
-            patches
-                .values()
-                .all_valid(&mut LEGACY_SESSION.create_execution_ctx())?,
-            "patches must be all valid: {}",
-            patches.values()
+            patches.values().validity()?.definitely_no_nulls(),
+            "patches must contain no nulls: {}",
+            patches.values(),
         );
     }
 
@@ -664,7 +656,7 @@ mod test {
         // Pick a seed that we know will trigger lots of patches.
         let encoder: alp_rd::RDEncoder = alp_rd::RDEncoder::new(&[seed.powi(-2)]);
 
-        let rd_array = encoder.encode(real_array.as_view(), &mut ctx);
+        let rd_array = encoder.encode(real_array.as_view());
 
         let decoded = rd_array
             .as_array()
diff --git a/encodings/alp/src/alp_rd/compute/cast.rs b/encodings/alp/src/alp_rd/compute/cast.rs
index 8a651708d3f..d290827355c 100644
--- a/encodings/alp/src/alp_rd/compute/cast.rs
+++ b/encodings/alp/src/alp_rd/compute/cast.rs
@@ -67,7 +67,7 @@ mod tests {
         let values = vec![1.0f32, 1.1, 1.2, 1.3, 1.4];
         let arr = PrimitiveArray::from_iter(values.clone());
         let encoder = RDEncoder::new(&values);
-        let alprd = encoder.encode(arr.as_view(), &mut ctx);
+        let alprd = encoder.encode(arr.as_view());
 
         let casted = alprd
             .into_array()
@@ -92,7 +92,7 @@ mod tests {
             PrimitiveArray::from_option_iter([Some(10.0f64), None, Some(10.1), Some(10.2), None]);
         let values = vec![10.0f64, 10.1, 10.2];
         let encoder = RDEncoder::new(&values);
-        let alprd = encoder.encode(arr.as_view(), &mut ctx);
+        let alprd = encoder.encode(arr.as_view());
 
         // Cast to NonNullable should fail since we have nulls. The failure surfaces during
         // execution since the reduce path defers when the validity stat is not cached.
@@ -122,31 +122,31 @@ mod tests {
         let values = vec![1.23f32, 4.56, 7.89, 10.11, 12.13];
         let arr = PrimitiveArray::from_iter(values.clone());
         let encoder = RDEncoder::new(&values);
-        encoder.encode(arr.as_view(), &mut array_session().create_execution_ctx())
+        encoder.encode(arr.as_view())
     })]
     #[case::f64({
         let values = vec![100.1f64, 200.2, 300.3, 400.4, 500.5];
         let arr = PrimitiveArray::from_iter(values.clone());
         let encoder = RDEncoder::new(&values);
-        encoder.encode(arr.as_view(), &mut array_session().create_execution_ctx())
+        encoder.encode(arr.as_view())
     })]
     #[case::single({
         let values = vec![42.42f64];
         let arr = PrimitiveArray::from_iter(values.clone());
         let encoder = RDEncoder::new(&values);
-        encoder.encode(arr.as_view(), &mut array_session().create_execution_ctx())
+        encoder.encode(arr.as_view())
     })]
     #[case::negative({
         let values = vec![0.0f32, -1.5, 2.5, -3.5, 4.5];
         let arr = PrimitiveArray::from_iter(values.clone());
         let encoder = RDEncoder::new(&values);
-        encoder.encode(arr.as_view(), &mut array_session().create_execution_ctx())
+        encoder.encode(arr.as_view())
     })]
     #[case::nullable({
         let arr = PrimitiveArray::from_option_iter([Some(1.1f32), None, Some(2.2), Some(3.3), None]);
         let values = vec![1.1f32, 2.2, 3.3];
         let encoder = RDEncoder::new(&values);
-        encoder.encode(arr.as_view(), &mut array_session().create_execution_ctx())
+        encoder.encode(arr.as_view())
     })]
     fn test_cast_alprd_conformance(#[case] alprd: crate::alp_rd::ALPRDArray) {
         test_cast_conformance(
diff --git a/encodings/alp/src/alp_rd/compute/filter.rs b/encodings/alp/src/alp_rd/compute/filter.rs
index 5d98b4561ce..69c26c53e3b 100644
--- a/encodings/alp/src/alp_rd/compute/filter.rs
+++ b/encodings/alp/src/alp_rd/compute/filter.rs
@@ -32,7 +32,6 @@ impl FilterKernel for ALPRD {
                 array.right_parts().filter(mask.clone())?,
                 array.right_bit_width(),
                 left_parts_exceptions,
-                ctx,
             )?
             .into_array(),
         ))
@@ -71,7 +70,7 @@ mod test {
     fn test_filter<T: ALPRDFloat>(#[case] a: T, #[case] b: T, #[case] outlier: T) {
         let mut ctx = SESSION.create_execution_ctx();
         let array = PrimitiveArray::new(buffer![a, b, outlier], Validity::NonNullable);
-        let encoded = RDEncoder::new(&[a, b]).encode(array.as_view(), &mut ctx);
+        let encoded = RDEncoder::new(&[a, b]).encode(array.as_view());
 
         // Make sure that we're testing the exception pathway.
         assert!(encoded.left_parts_patches().is_some());
@@ -90,10 +89,7 @@ mod test {
         let mut ctx = SESSION.create_execution_ctx();
         test_filter_conformance(
             &RDEncoder::new(&[a, b])
-                .encode(
-                    PrimitiveArray::from_iter([a, b, outlier, b, outlier]).as_view(),
-                    &mut ctx,
-                )
+                .encode(PrimitiveArray::from_iter([a, b, outlier, b, outlier]).as_view())
                 .into_array(),
             &mut ctx,
         );
@@ -109,7 +105,6 @@ mod test {
                 .encode(
                     PrimitiveArray::from_option_iter([Some(a), None, Some(outlier), Some(a), None])
                         .as_view(),
-                    &mut ctx,
                 )
                 .into_array(),
             &mut ctx,
diff --git a/encodings/alp/src/alp_rd/compute/mask.rs b/encodings/alp/src/alp_rd/compute/mask.rs
index 335f18a33cb..b32db4ef642 100644
--- a/encodings/alp/src/alp_rd/compute/mask.rs
+++ b/encodings/alp/src/alp_rd/compute/mask.rs
@@ -4,8 +4,6 @@
 use vortex_array::ArrayRef;
 use vortex_array::ArrayView;
 use vortex_array::IntoArray;
-use vortex_array::LEGACY_SESSION;
-use vortex_array::VortexSessionExecute;
 use vortex_array::arrays::scalar_fn::ScalarFnFactoryExt;
 use vortex_array::scalar_fn::EmptyOptions;
 use vortex_array::scalar_fn::fns::mask::Mask as MaskExpr;
@@ -22,8 +20,6 @@ impl MaskReduce for ALPRD {
             EmptyOptions,
             [array.left_parts().clone(), mask.clone()],
         )?;
-        // NOTE: `MaskReduce::mask` has a fixed trait signature without `ExecutionCtx`, so we
-        // construct a legacy ctx locally at this trait boundary.
         Ok(Some(
             ALPRD::try_new(
                 array.dtype().as_nullable(),
@@ -32,7 +28,6 @@ impl MaskReduce for ALPRD {
                 array.right_parts().clone(),
                 array.right_bit_width(),
                 array.left_parts_patches(),
-                &mut LEGACY_SESSION.create_execution_ctx(),
             )?
             .into_array(),
         ))
@@ -58,10 +53,7 @@ mod tests {
         let mut ctx = array_session().create_execution_ctx();
         test_mask_conformance(
             &RDEncoder::new(&[a, b])
-                .encode(
-                    PrimitiveArray::from_iter([a, b, outlier, b, outlier]).as_view(),
-                    &mut ctx,
-                )
+                .encode(PrimitiveArray::from_iter([a, b, outlier, b, outlier]).as_view())
                 .into_array(),
             &mut ctx,
         );
@@ -77,7 +69,6 @@ mod tests {
                 .encode(
                     PrimitiveArray::from_option_iter([Some(a), None, Some(outlier), Some(a), None])
                         .as_view(),
-                    &mut ctx,
                 )
                 .into_array(),
             &mut ctx,
diff --git a/encodings/alp/src/alp_rd/compute/mod.rs b/encodings/alp/src/alp_rd/compute/mod.rs
index f74b23fc2c2..696e622b873 100644
--- a/encodings/alp/src/alp_rd/compute/mod.rs
+++ b/encodings/alp/src/alp_rd/compute/mod.rs
@@ -34,47 +34,47 @@ mod tests {
         let values = vec![1.0f32, 1.1, 1.2, 1.3, 1.4];
         let arr = PrimitiveArray::from_iter(values.clone());
         let encoder = RDEncoder::new(&values);
-        encoder.encode(arr.as_view(), &mut SESSION.create_execution_ctx())
+        encoder.encode(arr.as_view())
     })]
     #[case::f64_array({
         let values = vec![100.0f64, 100.01, 100.02, 100.03, 100.04];
         let arr = PrimitiveArray::from_iter(values.clone());
         let encoder = RDEncoder::new(&values);
-        encoder.encode(arr.as_view(), &mut SESSION.create_execution_ctx())
+        encoder.encode(arr.as_view())
     })]
     // Nullable arrays
     #[case::nullable_f32({
         let values = vec![1.0f32, 1.2, 1.3];
         let arr = PrimitiveArray::from_option_iter([Some(1.0f32), None, Some(1.2), Some(1.3), None]);
         let encoder = RDEncoder::new(&values);
-        encoder.encode(arr.as_view(), &mut SESSION.create_execution_ctx())
+        encoder.encode(arr.as_view())
     })]
     #[case::nullable_f64({
         let values = vec![10.0f64, 10.2, 10.3];
         let arr = PrimitiveArray::from_option_iter([Some(10.0f64), None, Some(10.2), Some(10.3), None]);
         let encoder = RDEncoder::new(&values);
-        encoder.encode(arr.as_view(), &mut SESSION.create_execution_ctx())
+        encoder.encode(arr.as_view())
     })]
     // Edge cases
     #[case::single_element({
         let values = vec![42.42f64];
         let arr = PrimitiveArray::from_iter(values.clone());
         let encoder = RDEncoder::new(&values);
-        encoder.encode(arr.as_view(), &mut SESSION.create_execution_ctx())
+        encoder.encode(arr.as_view())
     })]
     // Arrays with small deltas (good for RD encoding)
     #[case::small_deltas({
         let values = vec![1000.0f32, 1000.001, 1000.002, 1000.003, 1000.004];
         let arr = PrimitiveArray::from_iter(values.clone());
         let encoder = RDEncoder::new(&values);
-        encoder.encode(arr.as_view(), &mut SESSION.create_execution_ctx())
+        encoder.encode(arr.as_view())
     })]
     // Large arrays
     #[case::large_f32({
         let values: Vec<f32> = (0..1000).map(|i| 100.0 + i as f32 * 0.01).collect();
         let arr = PrimitiveArray::from_iter(values.clone());
         let encoder = RDEncoder::new(&values);
-        encoder.encode(arr.as_view(), &mut SESSION.create_execution_ctx())
+        encoder.encode(arr.as_view())
     })]
     fn test_alp_rd_consistency(#[case] array: ALPRDArray) {
         let ctx = &mut SESSION.create_execution_ctx();
@@ -86,25 +86,25 @@ mod tests {
         let values = vec![1.0f32, 1.1, 1.2, 1.3, 1.4];
         let arr = PrimitiveArray::from_iter(values.clone());
         let encoder = RDEncoder::new(&values);
-        encoder.encode(arr.as_view(), &mut SESSION.create_execution_ctx())
+        encoder.encode(arr.as_view())
     })]
     #[case::f64_basic({
         let values = vec![100.0f64, 100.01, 100.02, 100.03, 100.04];
         let arr = PrimitiveArray::from_iter(values.clone());
         let encoder = RDEncoder::new(&values);
-        encoder.encode(arr.as_view(), &mut SESSION.create_execution_ctx())
+        encoder.encode(arr.as_view())
     })]
     #[case::f32_large({
         let values: Vec<f32> = (0..100).map(|i| 50.0 + i as f32 * 0.1).collect();
         let arr = PrimitiveArray::from_iter(values.clone());
         let encoder = RDEncoder::new(&values);
-        encoder.encode(arr.as_view(), &mut SESSION.create_execution_ctx())
+        encoder.encode(arr.as_view())
     })]
     #[case::f64_large({
         let values: Vec<f64> = (0..100).map(|i| 1000.0 + i as f64 * 0.01).collect();
         let arr = PrimitiveArray::from_iter(values.clone());
         let encoder = RDEncoder::new(&values);
-        encoder.encode(arr.as_view(), &mut SESSION.create_execution_ctx())
+        encoder.encode(arr.as_view())
     })]
     fn test_alp_rd_binary_numeric(#[case] array: ALPRDArray) {
         test_binary_numeric_array(&array.into_array(), &mut SESSION.create_execution_ctx());
diff --git a/encodings/alp/src/alp_rd/compute/take.rs b/encodings/alp/src/alp_rd/compute/take.rs
index 6db6eae2978..5718169042d 100644
--- a/encodings/alp/src/alp_rd/compute/take.rs
+++ b/encodings/alp/src/alp_rd/compute/take.rs
@@ -20,19 +20,14 @@ impl TakeExecute for ALPRD {
         ctx: &mut ExecutionCtx,
     ) -> VortexResult<Option<ArrayRef>> {
         let taken_left_parts = array.left_parts().take(indices.clone())?;
+        // `Patches::take` ignores null take-indices, so the resulting patch values contain no
+        // nulls (they may report a nullable dtype, but `ALPRD::try_new` validates that via
+        // `definitely_no_nulls` rather than requiring a cast).
         let left_parts_exceptions = array
             .left_parts_patches()
             .map(|patches| patches.take(indices, ctx))
             .transpose()?
-            .flatten()
-            .map(|p| {
-                let values_dtype = p
-                    .values()
-                    .dtype()
-                    .with_nullability(taken_left_parts.dtype().nullability());
-                p.cast_values(&values_dtype)
-            })
-            .transpose()?;
+            .flatten();
         let right_parts = array
             .right_parts()
             .take(indices.clone())?
@@ -48,7 +43,6 @@ impl TakeExecute for ALPRD {
                 right_parts,
                 array.right_bit_width(),
                 left_parts_exceptions,
-                ctx,
             )?
             .into_array(),
         ))
@@ -87,7 +81,7 @@ mod test {
 
         let mut ctx = SESSION.create_execution_ctx();
         let array = PrimitiveArray::from_iter([a, b, outlier]);
-        let encoded = RDEncoder::new(&[a, b]).encode(array.as_view(), &mut ctx);
+        let encoded = RDEncoder::new(&[a, b]).encode(array.as_view());
 
         assert!(encoded.left_parts_patches().is_some());
         assert!(
@@ -113,7 +107,7 @@ mod test {
     fn take_with_nulls<T: ALPRDFloat>(#[case] a: T, #[case] b: T, #[case] outlier: T) {
         let mut ctx = SESSION.create_execution_ctx();
         let array = PrimitiveArray::from_iter([a, b, outlier]);
-        let encoded = RDEncoder::new(&[a, b]).encode(array.as_view(), &mut ctx);
+        let encoded = RDEncoder::new(&[a, b]).encode(array.as_view());
 
         assert!(encoded.left_parts_patches().is_some());
         assert!(
@@ -144,10 +138,7 @@ mod test {
         let mut ctx = SESSION.create_execution_ctx();
         test_take_conformance(
             &RDEncoder::new(&[a, b])
-                .encode(
-                    PrimitiveArray::from_iter([a, b, outlier, b, outlier]).as_view(),
-                    &mut ctx,
-                )
+                .encode(PrimitiveArray::from_iter([a, b, outlier, b, outlier]).as_view())
                 .into_array(),
             &mut ctx,
         );
@@ -163,7 +154,6 @@ mod test {
                 .encode(
                     PrimitiveArray::from_option_iter([Some(a), None, Some(outlier), Some(a), None])
                         .as_view(),
-                    &mut ctx,
                 )
                 .into_array(),
             &mut ctx,
diff --git a/encodings/alp/src/alp_rd/mod.rs b/encodings/alp/src/alp_rd/mod.rs
index a9ac7ca82fe..07afe6b8adf 100644
--- a/encodings/alp/src/alp_rd/mod.rs
+++ b/encodings/alp/src/alp_rd/mod.rs
@@ -187,15 +187,11 @@ impl RDEncoder {
     ///
     /// Each value will be split into a left and right component, which are compressed individually.
     // TODO(joe): make fallible
-    pub fn encode(&self, array: ArrayView<'_, Primitive>, ctx: &mut ExecutionCtx) -> ALPRDArray {
-        match_each_alp_float_ptype!(array.ptype(), |P| { self.encode_generic::<P>(array, ctx) })
+    pub fn encode(&self, array: ArrayView<'_, Primitive>) -> ALPRDArray {
+        match_each_alp_float_ptype!(array.ptype(), |P| { self.encode_generic::<P>(array) })
     }
 
-    fn encode_generic<T>(
-        &self,
-        array: ArrayView<'_, Primitive>,
-        ctx: &mut ExecutionCtx,
-    ) -> ALPRDArray
+    fn encode_generic<T>(&self, array: ArrayView<'_, Primitive>) -> ALPRDArray
     where
         T: ALPRDFloat + NativePType,
         T::UINT: NativePType,
@@ -294,7 +290,6 @@ impl RDEncoder {
             packed_right,
             self.right_bit_width,
             exceptions,
-            ctx,
         )
         .vortex_expect("ALPRDArray construction in encode")
     }
diff --git a/encodings/alp/src/alp_rd/ops.rs b/encodings/alp/src/alp_rd/ops.rs
index be8d8f88948..0fc0dc40fa1 100644
--- a/encodings/alp/src/alp_rd/ops.rs
+++ b/encodings/alp/src/alp_rd/ops.rs
@@ -90,7 +90,7 @@ mod test {
     fn test_slice<T: ALPRDFloat>(#[case] a: T, #[case] b: T, #[case] outlier: T) {
         let mut ctx = SESSION.create_execution_ctx();
         let array = PrimitiveArray::from_iter([a, b, outlier]);
-        let encoded = RDEncoder::new(&[a, b]).encode(array.as_view(), &mut ctx);
+        let encoded = RDEncoder::new(&[a, b]).encode(array.as_view());
 
         assert!(encoded.left_parts_patches().is_some());
         assert_arrays_eq!(encoded, array, &mut ctx);
@@ -106,7 +106,7 @@ mod test {
     ) {
         let mut ctx = SESSION.create_execution_ctx();
         let array = PrimitiveArray::from_iter([a, b, outlier]);
-        let encoded = RDEncoder::new(&[a, b]).encode(array.as_view(), &mut ctx);
+        let encoded = RDEncoder::new(&[a, b]).encode(array.as_view());
         assert!(encoded.left_parts_patches().is_some());
         assert_arrays_eq!(encoded, array, &mut ctx);
     }
@@ -118,7 +118,7 @@ mod test {
         let b = 0.2f64;
         let outlier = 3e100f64;
         let array = PrimitiveArray::from_option_iter([Some(a), Some(b), Some(outlier)]);
-        let encoded = RDEncoder::new(&[a, b]).encode(array.as_view(), &mut ctx);
+        let encoded = RDEncoder::new(&[a, b]).encode(array.as_view());
         assert!(encoded.left_parts_patches().is_some());
         assert_arrays_eq!(
             encoded,
diff --git a/vortex-array/src/patches.rs b/vortex-array/src/patches.rs
index 5c73aa95d13..77459d4a64a 100644
--- a/vortex-array/src/patches.rs
+++ b/vortex-array/src/patches.rs
@@ -910,7 +910,12 @@ impl Patches {
 
         let new_indices = new_indices.into_array();
         let new_array_len = take_indices.len();
-        let values_validity = take_indices_validity.take(&new_indices)?;
+        // When nulls are excluded every taken value is valid, so the values are non-nullable.
+        let values_validity = if include_nulls {
+            take_indices_validity.take(&new_indices)?
+        } else {
+            Validity::NonNullable
+        };
 
         Ok(Some(Self {
             array_len: new_array_len,
@@ -1058,8 +1063,13 @@ where
     }
 
     let new_sparse_indices = new_sparse_indices.into_array();
-    let values_validity =
-        Validity::from_mask(take_validity, take_nullability).take(&new_sparse_indices)?;
+    // When nulls are excluded every emitted value index is valid, so the value-index array is
+    // non-nullable; only when nulls are included must the take indices' nullability be preserved.
+    let values_validity = if include_nulls {
+        Validity::from_mask(take_validity, take_nullability).take(&new_sparse_indices)?
+    } else {
+        Validity::NonNullable
+    };
     Ok(Some((
         new_sparse_indices,
         PrimitiveArray::new(value_indices, values_validity).into_array(),
diff --git a/vortex-btrblocks/src/schemes/float/alprd.rs b/vortex-btrblocks/src/schemes/float/alprd.rs
index f97ae077882..969afabad49 100644
--- a/vortex-btrblocks/src/schemes/float/alprd.rs
+++ b/vortex-btrblocks/src/schemes/float/alprd.rs
@@ -66,7 +66,7 @@ impl Scheme for ALPRDScheme {
             ptype => vortex_panic!("cannot ALPRD compress ptype {ptype}"),
         };
 
-        let alp_rd = encoder.encode(primitive_array, exec_ctx);
+        let alp_rd = encoder.encode(primitive_array);
         let dtype = alp_rd.dtype().clone();
         let right_bit_width = alp_rd.right_bit_width();
         let mut parts = ALPRDArrayOwnedExt::into_data_parts(alp_rd);
@@ -82,7 +82,6 @@ impl Scheme for ALPRDScheme {
             parts.right_parts,
             right_bit_width,
             parts.left_parts_patches,
-            exec_ctx,
         )?
         .into_array())
     }
diff --git a/vortex-test/compat-gen/src/fixtures/arrays/synthetic/encodings/alprd.rs b/vortex-test/compat-gen/src/fixtures/arrays/synthetic/encodings/alprd.rs
index a073dfc6693..ef41fbf8a58 100644
--- a/vortex-test/compat-gen/src/fixtures/arrays/synthetic/encodings/alprd.rs
+++ b/vortex-test/compat-gen/src/fixtures/arrays/synthetic/encodings/alprd.rs
@@ -46,7 +46,7 @@ impl FlatLayoutFixture for AlprdFixture {
         vec![ALPRD.id()]
     }
 
-    fn build(&self, ctx: &mut ExecutionCtx) -> VortexResult<ArrayRef> {
+    fn build(&self, _ctx: &mut ExecutionCtx) -> VortexResult<ArrayRef> {
         let sensor: PrimitiveArray = (0..N)
             .map(|i| {
                 let noise = ((i * 7 + 13) % 100) as f64 / 1000.0;
@@ -151,31 +151,21 @@ impl FlatLayoutFixture for AlprdFixture {
                 "nullable_specials",
             ]),
             vec![
-                sensor_enc.encode(sensor.as_view(), ctx).into_array(),
-                drift_enc.encode(drift.as_view(), ctx).into_array(),
-                constant_enc
-                    .encode(constant_series.as_view(), ctx)
-                    .into_array(),
-                decreasing_enc
-                    .encode(decreasing.as_view(), ctx)
-                    .into_array(),
-                oscillating_enc
-                    .encode(oscillating.as_view(), ctx)
-                    .into_array(),
+                sensor_enc.encode(sensor.as_view()).into_array(),
+                drift_enc.encode(drift.as_view()).into_array(),
+                constant_enc.encode(constant_series.as_view()).into_array(),
+                decreasing_enc.encode(decreasing.as_view()).into_array(),
+                oscillating_enc.encode(oscillating.as_view()).into_array(),
                 periodic_resets_enc
-                    .encode(periodic_resets.as_view(), ctx)
-                    .into_array(),
-                nullable_enc
-                    .encode(sensor_nullable.as_view(), ctx)
-                    .into_array(),
-                special_enc
-                    .encode(special_values.as_view(), ctx)
+                    .encode(periodic_resets.as_view())
                     .into_array(),
+                nullable_enc.encode(sensor_nullable.as_view()).into_array(),
+                special_enc.encode(special_values.as_view()).into_array(),
                 boundary_enc
-                    .encode(boundary_specials.as_view(), ctx)
+                    .encode(boundary_specials.as_view())
                     .into_array(),
                 nullable_special_enc
-                    .encode(nullable_specials.as_view(), ctx)
+                    .encode(nullable_specials.as_view())
                     .into_array(),
             ],
             N,
diff --git a/vortex/benches/single_encoding_throughput.rs b/vortex/benches/single_encoding_throughput.rs
index a20777af583..b75f832a7a1 100644
--- a/vortex/benches/single_encoding_throughput.rs
+++ b/vortex/benches/single_encoding_throughput.rs
@@ -299,10 +299,10 @@ fn bench_alp_rd_compress_f64(bencher: Bencher) {
     let (_, _, float_array) = setup_primitive_arrays();
 
     with_byte_counter(bencher, NUM_VALUES * 8)
-        .with_inputs(|| (&float_array, SESSION.create_execution_ctx()))
-        .bench_refs(|(a, ctx)| {
+        .with_inputs(|| &float_array)
+        .bench_refs(|a| {
             let encoder = RDEncoder::new(a.as_slice::<f64>());
-            encoder.encode(a.as_view(), ctx)
+            encoder.encode(a.as_view())
         });
 }
 
@@ -310,7 +310,7 @@ fn bench_alp_rd_compress_f64(bencher: Bencher) {
 fn bench_alp_rd_decompress_f64(bencher: Bencher) {
     let (_, _, float_array) = setup_primitive_arrays();
     let encoder = RDEncoder::new(float_array.as_slice::<f64>());
-    let compressed = encoder.encode(float_array.as_view(), &mut SESSION.create_execution_ctx());
+    let compressed = encoder.encode(float_array.as_view());
 
     with_byte_counter(bencher, NUM_VALUES * 8)
         .with_inputs(|| (&compressed, SESSION.create_execution_ctx()))