diff --git a/encodings/alp/src/alp/array.rs b/encodings/alp/src/alp/array.rs
index 99e166bac18..4550bd1a613 100644
--- a/encodings/alp/src/alp/array.rs
+++ b/encodings/alp/src/alp/array.rs
@@ -3,6 +3,7 @@
 
 use std::fmt::Debug;
 use std::hash::Hash;
+use std::sync::Arc;
 
 use vortex_array::ArrayEq;
 use vortex_array::ArrayHash;
@@ -10,7 +11,7 @@ use vortex_array::ArrayRef;
 use vortex_array::DeserializeMetadata;
 use vortex_array::DynArray;
 use vortex_array::ExecutionCtx;
-use vortex_array::ExecutionStep;
+use vortex_array::ExecutionResult;
 use vortex_array::IntoArray;
 use vortex_array::Precision;
 use vortex_array::ProstMetadata;
@@ -235,10 +236,10 @@ impl VTable for ALPVTable {
         Ok(())
     }
 
-    fn execute(array: &Self::Array, ctx: &mut ExecutionCtx) -> VortexResult<ExecutionStep> {
-        // TODO(joe): take by value
-        Ok(ExecutionStep::Done(
-            execute_decompress(array.clone(), ctx)?.into_array(),
+    fn execute(array: Arc<Self::Array>, ctx: &mut ExecutionCtx) -> VortexResult<ExecutionResult> {
+        let array = Arc::try_unwrap(array).unwrap_or_else(|arc| (*arc).clone());
+        Ok(ExecutionResult::done(
+            execute_decompress(array, ctx)?.into_array(),
         ))
     }
 
diff --git a/encodings/alp/src/alp_rd/array.rs b/encodings/alp/src/alp_rd/array.rs
index d98b596eaa8..cf9c6f22cd6 100644
--- a/encodings/alp/src/alp_rd/array.rs
+++ b/encodings/alp/src/alp_rd/array.rs
@@ -3,6 +3,7 @@
 
 use std::fmt::Debug;
 use std::hash::Hash;
+use std::sync::Arc;
 
 use itertools::Itertools;
 use vortex_array::ArrayEq;
@@ -11,7 +12,7 @@ use vortex_array::ArrayRef;
 use vortex_array::DeserializeMetadata;
 use vortex_array::DynArray;
 use vortex_array::ExecutionCtx;
-use vortex_array::ExecutionStep;
+use vortex_array::ExecutionResult;
 use vortex_array::IntoArray;
 use vortex_array::Precision;
 use vortex_array::ProstMetadata;
@@ -296,7 +297,7 @@ impl VTable for ALPRDVTable {
         Ok(())
     }
 
-    fn execute(array: &Self::Array, ctx: &mut ExecutionCtx) -> VortexResult<ExecutionStep> {
+    fn execute(array: Arc<Self::Array>, ctx: &mut ExecutionCtx) -> VortexResult<ExecutionResult> {
         let left_parts = array.left_parts().clone().execute::<PrimitiveArray>(ctx)?;
         let right_parts = array.right_parts().clone().execute::<PrimitiveArray>(ctx)?;
 
@@ -335,7 +336,7 @@ impl VTable for ALPRDVTable {
             )
         };
 
-        Ok(ExecutionStep::Done(decoded_array.into_array()))
+        Ok(ExecutionResult::done(decoded_array.into_array()))
     }
 
     fn reduce_parent(
diff --git a/encodings/bytebool/src/array.rs b/encodings/bytebool/src/array.rs
index 8c5c7c9fa31..9290ceae5ff 100644
--- a/encodings/bytebool/src/array.rs
+++ b/encodings/bytebool/src/array.rs
@@ -3,13 +3,14 @@
 
 use std::fmt::Debug;
 use std::hash::Hash;
+use std::sync::Arc;
 
 use vortex_array::ArrayEq;
 use vortex_array::ArrayHash;
 use vortex_array::ArrayRef;
 use vortex_array::EmptyMetadata;
 use vortex_array::ExecutionCtx;
-use vortex_array::ExecutionStep;
+use vortex_array::ExecutionResult;
 use vortex_array::IntoArray;
 use vortex_array::Precision;
 use vortex_array::arrays::BoolArray;
@@ -183,10 +184,10 @@ impl VTable for ByteBoolVTable {
         crate::rules::RULES.evaluate(array, parent, child_idx)
     }
 
-    fn execute(array: &Self::Array, _ctx: &mut ExecutionCtx) -> VortexResult<ExecutionStep> {
+    fn execute(array: Arc<Self::Array>, _ctx: &mut ExecutionCtx) -> VortexResult<ExecutionResult> {
         let boolean_buffer = BitBuffer::from(array.as_slice());
         let validity = array.validity().clone();
-        Ok(ExecutionStep::Done(
+        Ok(ExecutionResult::done(
             BoolArray::new(boolean_buffer, validity).into_array(),
         ))
     }
diff --git a/encodings/datetime-parts/src/array.rs b/encodings/datetime-parts/src/array.rs
index e1f51d5ab66..e3b83e02850 100644
--- a/encodings/datetime-parts/src/array.rs
+++ b/encodings/datetime-parts/src/array.rs
@@ -3,6 +3,7 @@
 
 use std::fmt::Debug;
 use std::hash::Hash;
+use std::sync::Arc;
 
 use vortex_array::ArrayEq;
 use vortex_array::ArrayHash;
@@ -10,7 +11,7 @@ use vortex_array::ArrayRef;
 use vortex_array::DeserializeMetadata;
 use vortex_array::DynArray;
 use vortex_array::ExecutionCtx;
-use vortex_array::ExecutionStep;
+use vortex_array::ExecutionResult;
 use vortex_array::IntoArray;
 use vortex_array::Precision;
 use vortex_array::ProstMetadata;
@@ -222,9 +223,9 @@ impl VTable for DateTimePartsVTable {
         Ok(())
     }
 
-    fn execute(array: &Self::Array, ctx: &mut ExecutionCtx) -> VortexResult<ExecutionStep> {
-        Ok(ExecutionStep::Done(
-            decode_to_temporal(array, ctx)?.into_array(),
+    fn execute(array: Arc<Self::Array>, ctx: &mut ExecutionCtx) -> VortexResult<ExecutionResult> {
+        Ok(ExecutionResult::done(
+            decode_to_temporal(&array, ctx)?.into_array(),
         ))
     }
 
diff --git a/encodings/decimal-byte-parts/src/decimal_byte_parts/mod.rs b/encodings/decimal-byte-parts/src/decimal_byte_parts/mod.rs
index 79e24e1b0cd..cc7f0c25743 100644
--- a/encodings/decimal-byte-parts/src/decimal_byte_parts/mod.rs
+++ b/encodings/decimal-byte-parts/src/decimal_byte_parts/mod.rs
@@ -6,6 +6,7 @@ mod rules;
 mod slice;
 
 use std::hash::Hash;
+use std::sync::Arc;
 
 use prost::Message as _;
 use vortex_array::ArrayEq;
@@ -13,7 +14,7 @@ use vortex_array::ArrayHash;
 use vortex_array::ArrayRef;
 use vortex_array::DynArray;
 use vortex_array::ExecutionCtx;
-use vortex_array::ExecutionStep;
+use vortex_array::ExecutionResult;
 use vortex_array::IntoArray;
 use vortex_array::Precision;
 use vortex_array::ProstMetadata;
@@ -190,8 +191,8 @@ impl VTable for DecimalBytePartsVTable {
         PARENT_RULES.evaluate(array, parent, child_idx)
     }
 
-    fn execute(array: &Self::Array, ctx: &mut ExecutionCtx) -> VortexResult<ExecutionStep> {
-        to_canonical_decimal(array, ctx).map(ExecutionStep::Done)
+    fn execute(array: Arc<Self::Array>, ctx: &mut ExecutionCtx) -> VortexResult<ExecutionResult> {
+        to_canonical_decimal(&array, ctx).map(ExecutionResult::done)
     }
 
     fn execute_parent(
diff --git a/encodings/fastlanes/src/bitpacking/vtable/mod.rs b/encodings/fastlanes/src/bitpacking/vtable/mod.rs
index facb2aa5a4f..a5f45839cff 100644
--- a/encodings/fastlanes/src/bitpacking/vtable/mod.rs
+++ b/encodings/fastlanes/src/bitpacking/vtable/mod.rs
@@ -2,13 +2,14 @@
 // SPDX-FileCopyrightText: Copyright the Vortex contributors
 
 use std::hash::Hash;
+use std::sync::Arc;
 
 use vortex_array::ArrayEq;
 use vortex_array::ArrayHash;
 use vortex_array::ArrayRef;
 use vortex_array::DeserializeMetadata;
 use vortex_array::ExecutionCtx;
-use vortex_array::ExecutionStep;
+use vortex_array::ExecutionResult;
 use vortex_array::IntoArray;
 use vortex_array::Precision;
 use vortex_array::ProstMetadata;
@@ -354,8 +355,10 @@ impl VTable for BitPackedVTable {
         })
     }
 
-    fn execute(array: &Self::Array, ctx: &mut ExecutionCtx) -> VortexResult<ExecutionStep> {
-        Ok(ExecutionStep::Done(unpack_array(array, ctx)?.into_array()))
+    fn execute(array: Arc<Self::Array>, ctx: &mut ExecutionCtx) -> VortexResult<ExecutionResult> {
+        Ok(ExecutionResult::done(
+            unpack_array(&array, ctx)?.into_array(),
+        ))
     }
 
     fn execute_parent(
diff --git a/encodings/fastlanes/src/delta/vtable/mod.rs b/encodings/fastlanes/src/delta/vtable/mod.rs
index 5bf67a561c5..917a93cbcb5 100644
--- a/encodings/fastlanes/src/delta/vtable/mod.rs
+++ b/encodings/fastlanes/src/delta/vtable/mod.rs
@@ -2,6 +2,7 @@
 // SPDX-FileCopyrightText: Copyright the Vortex contributors
 
 use std::hash::Hash;
+use std::sync::Arc;
 
 use fastlanes::FastLanes;
 use prost::Message;
@@ -9,7 +10,7 @@ use vortex_array::ArrayEq;
 use vortex_array::ArrayHash;
 use vortex_array::ArrayRef;
 use vortex_array::ExecutionCtx;
-use vortex_array::ExecutionStep;
+use vortex_array::ExecutionResult;
 use vortex_array::IntoArray;
 use vortex_array::Precision;
 use vortex_array::ProstMetadata;
@@ -190,9 +191,9 @@ impl VTable for DeltaVTable {
         DeltaArray::try_new(bases, deltas, metadata.0.offset as usize, len)
     }
 
-    fn execute(array: &Self::Array, ctx: &mut ExecutionCtx) -> VortexResult<ExecutionStep> {
-        Ok(ExecutionStep::Done(
-            delta_decompress(array, ctx)?.into_array(),
+    fn execute(array: Arc<Self::Array>, ctx: &mut ExecutionCtx) -> VortexResult<ExecutionResult> {
+        Ok(ExecutionResult::done(
+            delta_decompress(&array, ctx)?.into_array(),
         ))
     }
 }
diff --git a/encodings/fastlanes/src/for/vtable/mod.rs b/encodings/fastlanes/src/for/vtable/mod.rs
index 1560c721e21..251421cb22a 100644
--- a/encodings/fastlanes/src/for/vtable/mod.rs
+++ b/encodings/fastlanes/src/for/vtable/mod.rs
@@ -3,12 +3,13 @@
 
 use std::fmt::Debug;
 use std::hash::Hash;
+use std::sync::Arc;
 
 use vortex_array::ArrayEq;
 use vortex_array::ArrayHash;
 use vortex_array::ArrayRef;
 use vortex_array::ExecutionCtx;
-use vortex_array::ExecutionStep;
+use vortex_array::ExecutionResult;
 use vortex_array::IntoArray;
 use vortex_array::Precision;
 use vortex_array::buffer::BufferHandle;
@@ -166,8 +167,8 @@ impl VTable for FoRVTable {
         PARENT_RULES.evaluate(array, parent, child_idx)
     }
 
-    fn execute(array: &Self::Array, ctx: &mut ExecutionCtx) -> VortexResult<ExecutionStep> {
-        Ok(ExecutionStep::Done(decompress(array, ctx)?.into_array()))
+    fn execute(array: Arc<Self::Array>, ctx: &mut ExecutionCtx) -> VortexResult<ExecutionResult> {
+        Ok(ExecutionResult::done(decompress(&array, ctx)?.into_array()))
     }
 
     fn execute_parent(
diff --git a/encodings/fastlanes/src/rle/vtable/mod.rs b/encodings/fastlanes/src/rle/vtable/mod.rs
index ab12a22294d..5d20351d495 100644
--- a/encodings/fastlanes/src/rle/vtable/mod.rs
+++ b/encodings/fastlanes/src/rle/vtable/mod.rs
@@ -2,13 +2,14 @@
 // SPDX-FileCopyrightText: Copyright the Vortex contributors
 
 use std::hash::Hash;
+use std::sync::Arc;
 
 use prost::Message;
 use vortex_array::ArrayEq;
 use vortex_array::ArrayHash;
 use vortex_array::ArrayRef;
 use vortex_array::ExecutionCtx;
-use vortex_array::ExecutionStep;
+use vortex_array::ExecutionResult;
 use vortex_array::IntoArray;
 use vortex_array::Precision;
 use vortex_array::ProstMetadata;
@@ -231,9 +232,9 @@ impl VTable for RLEVTable {
         PARENT_KERNELS.execute(array, parent, child_idx, ctx)
     }
 
-    fn execute(array: &Self::Array, ctx: &mut ExecutionCtx) -> VortexResult<ExecutionStep> {
-        Ok(ExecutionStep::Done(
-            rle_decompress(array, ctx)?.into_array(),
+    fn execute(array: Arc<Self::Array>, ctx: &mut ExecutionCtx) -> VortexResult<ExecutionResult> {
+        Ok(ExecutionResult::done(
+            rle_decompress(&array, ctx)?.into_array(),
         ))
     }
 }
diff --git a/encodings/fsst/src/array.rs b/encodings/fsst/src/array.rs
index 2f10b0b1a47..6ee6d75d1bb 100644
--- a/encodings/fsst/src/array.rs
+++ b/encodings/fsst/src/array.rs
@@ -17,7 +17,7 @@ use vortex_array::Canonical;
 use vortex_array::DeserializeMetadata;
 use vortex_array::DynArray;
 use vortex_array::ExecutionCtx;
-use vortex_array::ExecutionStep;
+use vortex_array::ExecutionResult;
 use vortex_array::IntoArray;
 use vortex_array::Precision;
 use vortex_array::ProstMetadata;
@@ -339,8 +339,8 @@ impl VTable for FSSTVTable {
         Ok(())
     }
 
-    fn execute(array: &Self::Array, ctx: &mut ExecutionCtx) -> VortexResult<ExecutionStep> {
-        canonicalize_fsst(array, ctx).map(ExecutionStep::Done)
+    fn execute(array: Arc<Self::Array>, ctx: &mut ExecutionCtx) -> VortexResult<ExecutionResult> {
+        canonicalize_fsst(&array, ctx).map(ExecutionResult::done)
     }
 
     fn execute_parent(
diff --git a/encodings/pco/src/array.rs b/encodings/pco/src/array.rs
index 4b87e49eb70..cc418b30a58 100644
--- a/encodings/pco/src/array.rs
+++ b/encodings/pco/src/array.rs
@@ -4,6 +4,7 @@
 use std::cmp;
 use std::fmt::Debug;
 use std::hash::Hash;
+use std::sync::Arc;
 
 use pco::ChunkConfig;
 use pco::PagingSpec;
@@ -20,7 +21,7 @@ use vortex_array::ArrayHash;
 use vortex_array::ArrayRef;
 use vortex_array::DynArray;
 use vortex_array::ExecutionCtx;
-use vortex_array::ExecutionStep;
+use vortex_array::ExecutionResult;
 use vortex_array::IntoArray;
 use vortex_array::Precision;
 use vortex_array::ProstMetadata;
@@ -263,8 +264,8 @@ impl VTable for PcoVTable {
         Ok(())
     }
 
-    fn execute(array: &Self::Array, _ctx: &mut ExecutionCtx) -> VortexResult<ExecutionStep> {
-        Ok(ExecutionStep::Done(array.decompress()?.into_array()))
+    fn execute(array: Arc<Self::Array>, _ctx: &mut ExecutionCtx) -> VortexResult<ExecutionResult> {
+        Ok(ExecutionResult::done(array.decompress()?.into_array()))
     }
 
     fn reduce_parent(
diff --git a/encodings/runend/src/array.rs b/encodings/runend/src/array.rs
index f943b747cda..8ab44108424 100644
--- a/encodings/runend/src/array.rs
+++ b/encodings/runend/src/array.rs
@@ -3,6 +3,7 @@
 
 use std::fmt::Debug;
 use std::hash::Hash;
+use std::sync::Arc;
 
 use vortex_array::ArrayEq;
 use vortex_array::ArrayHash;
@@ -10,7 +11,7 @@ use vortex_array::ArrayRef;
 use vortex_array::DeserializeMetadata;
 use vortex_array::DynArray;
 use vortex_array::ExecutionCtx;
-use vortex_array::ExecutionStep;
+use vortex_array::ExecutionResult;
 use vortex_array::IntoArray;
 use vortex_array::Precision;
 use vortex_array::ProstMetadata;
@@ -203,8 +204,8 @@ impl VTable for RunEndVTable {
         PARENT_KERNELS.execute(array, parent, child_idx, ctx)
     }
 
-    fn execute(array: &Self::Array, ctx: &mut ExecutionCtx) -> VortexResult<ExecutionStep> {
-        run_end_canonicalize(array, ctx).map(ExecutionStep::Done)
+    fn execute(array: Arc<Self::Array>, ctx: &mut ExecutionCtx) -> VortexResult<ExecutionResult> {
+        run_end_canonicalize(&array, ctx).map(ExecutionResult::done)
     }
 }
 
diff --git a/encodings/sequence/src/array.rs b/encodings/sequence/src/array.rs
index d7149bb3dab..d493a16ae76 100644
--- a/encodings/sequence/src/array.rs
+++ b/encodings/sequence/src/array.rs
@@ -2,12 +2,13 @@
 // SPDX-FileCopyrightText: Copyright the Vortex contributors
 
 use std::hash::Hash;
+use std::sync::Arc;
 
 use num_traits::cast::FromPrimitive;
 use vortex_array::ArrayRef;
 use vortex_array::DeserializeMetadata;
 use vortex_array::ExecutionCtx;
-use vortex_array::ExecutionStep;
+use vortex_array::ExecutionResult;
 use vortex_array::Precision;
 use vortex_array::ProstMetadata;
 use vortex_array::SerializeMetadata;
@@ -381,8 +382,8 @@ impl VTable for SequenceVTable {
         Ok(())
     }
 
-    fn execute(array: &Self::Array, _ctx: &mut ExecutionCtx) -> VortexResult<ExecutionStep> {
-        sequence_decompress(array).map(ExecutionStep::Done)
+    fn execute(array: Arc<Self::Array>, _ctx: &mut ExecutionCtx) -> VortexResult<ExecutionResult> {
+        sequence_decompress(&array).map(ExecutionResult::done)
     }
 
     fn execute_parent(
diff --git a/encodings/sparse/src/lib.rs b/encodings/sparse/src/lib.rs
index b2bf1e81ae9..eca81d3cd5c 100644
--- a/encodings/sparse/src/lib.rs
+++ b/encodings/sparse/src/lib.rs
@@ -3,6 +3,7 @@
 
 use std::fmt::Debug;
 use std::hash::Hash;
+use std::sync::Arc;
 
 use kernel::PARENT_KERNELS;
 use prost::Message as _;
@@ -11,7 +12,7 @@ use vortex_array::ArrayHash;
 use vortex_array::ArrayRef;
 use vortex_array::DynArray;
 use vortex_array::ExecutionCtx;
-use vortex_array::ExecutionStep;
+use vortex_array::ExecutionResult;
 use vortex_array::IntoArray;
 use vortex_array::Precision;
 use vortex_array::ToCanonical;
@@ -256,8 +257,8 @@ impl VTable for SparseVTable {
         PARENT_KERNELS.execute(array, parent, child_idx, ctx)
     }
 
-    fn execute(array: &Self::Array, ctx: &mut ExecutionCtx) -> VortexResult<ExecutionStep> {
-        execute_sparse(array, ctx).map(ExecutionStep::Done)
+    fn execute(array: Arc<Self::Array>, ctx: &mut ExecutionCtx) -> VortexResult<ExecutionResult> {
+        execute_sparse(&array, ctx).map(ExecutionResult::done)
     }
 }
 
diff --git a/encodings/zigzag/src/array.rs b/encodings/zigzag/src/array.rs
index f8139246880..4f6829b9517 100644
--- a/encodings/zigzag/src/array.rs
+++ b/encodings/zigzag/src/array.rs
@@ -2,6 +2,7 @@
 // SPDX-FileCopyrightText: Copyright the Vortex contributors
 
 use std::hash::Hash;
+use std::sync::Arc;
 
 use vortex_array::ArrayEq;
 use vortex_array::ArrayHash;
@@ -9,7 +10,7 @@ use vortex_array::ArrayRef;
 use vortex_array::DynArray;
 use vortex_array::EmptyMetadata;
 use vortex_array::ExecutionCtx;
-use vortex_array::ExecutionStep;
+use vortex_array::ExecutionResult;
 use vortex_array::IntoArray;
 use vortex_array::Precision;
 use vortex_array::buffer::BufferHandle;
@@ -149,8 +150,8 @@ impl VTable for ZigZagVTable {
         Ok(())
     }
 
-    fn execute(array: &Self::Array, ctx: &mut ExecutionCtx) -> VortexResult<ExecutionStep> {
-        Ok(ExecutionStep::Done(
+    fn execute(array: Arc<Self::Array>, ctx: &mut ExecutionCtx) -> VortexResult<ExecutionResult> {
+        Ok(ExecutionResult::done(
             zigzag_decode(array.encoded().clone().execute(ctx)?).into_array(),
         ))
     }
diff --git a/encodings/zstd/src/array.rs b/encodings/zstd/src/array.rs
index 52665408abb..e5509ecb2ea 100644
--- a/encodings/zstd/src/array.rs
+++ b/encodings/zstd/src/array.rs
@@ -13,7 +13,7 @@ use vortex_array::ArrayRef;
 use vortex_array::Canonical;
 use vortex_array::DynArray;
 use vortex_array::ExecutionCtx;
-use vortex_array::ExecutionStep;
+use vortex_array::ExecutionResult;
 use vortex_array::IntoArray;
 use vortex_array::Precision;
 use vortex_array::ProstMetadata;
@@ -272,11 +272,11 @@ impl VTable for ZstdVTable {
         Ok(())
     }
 
-    fn execute(array: &Self::Array, ctx: &mut ExecutionCtx) -> VortexResult<ExecutionStep> {
+    fn execute(array: Arc<Self::Array>, ctx: &mut ExecutionCtx) -> VortexResult<ExecutionResult> {
         array
             .decompress()?
             .execute::<ArrayRef>(ctx)
-            .map(ExecutionStep::Done)
+            .map(ExecutionResult::done)
     }
 
     fn reduce_parent(
diff --git a/encodings/zstd/src/zstd_buffers.rs b/encodings/zstd/src/zstd_buffers.rs
index ff733475a95..1673b156352 100644
--- a/encodings/zstd/src/zstd_buffers.rs
+++ b/encodings/zstd/src/zstd_buffers.rs
@@ -10,7 +10,7 @@ use vortex_array::ArrayEq;
 use vortex_array::ArrayHash;
 use vortex_array::ArrayRef;
 use vortex_array::ExecutionCtx;
-use vortex_array::ExecutionStep;
+use vortex_array::ExecutionResult;
 use vortex_array::Precision;
 use vortex_array::ProstMetadata;
 use vortex_array::buffer::BufferHandle;
@@ -467,12 +467,12 @@ impl VTable for ZstdBuffersVTable {
         Ok(())
     }
 
-    fn execute(array: &Self::Array, ctx: &mut ExecutionCtx) -> VortexResult<ExecutionStep> {
+    fn execute(array: Arc<Self::Array>, ctx: &mut ExecutionCtx) -> VortexResult<ExecutionResult> {
         let session = ctx.session();
         let inner_array = array.decompress_and_build_inner(session)?;
         inner_array
             .execute::<ArrayRef>(ctx)
-            .map(ExecutionStep::Done)
+            .map(ExecutionResult::done)
     }
 }
 
diff --git a/fuzz/src/array/mod.rs b/fuzz/src/array/mod.rs
index acd461c3a32..acb05fa4dc4 100644
--- a/fuzz/src/array/mod.rs
+++ b/fuzz/src/array/mod.rs
@@ -376,7 +376,7 @@ impl<'a> Arbitrary<'a> for FuzzArrayAction {
                         current_array
                             .to_canonical()
                             .vortex_expect("to_canonical should succeed in fuzz test"),
-                        &Mask::from_iter(mask.clone()),
+                        &!Mask::from_iter(mask.clone()),
                     )
                     .vortex_expect("mask_canonical_array should succeed in fuzz test");
                     // Update current_array to the result for chaining
diff --git a/vortex-array/src/array/mod.rs b/vortex-array/src/array/mod.rs
index fa141141e33..f291aec7068 100644
--- a/vortex-array/src/array/mod.rs
+++ b/vortex-array/src/array/mod.rs
@@ -98,7 +98,7 @@ pub trait DynArray:
     fn dtype(&self) -> &DType;
 
     /// Returns the vtable of the array.
-    fn vtable(&self) -> &dyn DynVTable;
+    fn vtable(&self) -> &'static dyn DynVTable;
 
     /// Returns the encoding ID of the array.
     fn encoding_id(&self) -> ArrayId;
@@ -192,7 +192,7 @@ impl DynArray for Arc<dyn DynArray> {
         self.as_ref().dtype()
     }
 
-    fn vtable(&self) -> &dyn DynVTable {
+    fn vtable(&self) -> &'static dyn DynVTable {
         self.as_ref().vtable()
     }
 
@@ -402,6 +402,11 @@ impl<V: VTable> ArrayAdapter<V> {
     pub fn as_inner(&self) -> &V::Array {
         &self.0
     }
+
+    /// Consume the adapter and return the underlying array.
+    pub fn into_inner(self) -> V::Array {
+        self.0
+    }
 }
 
 impl<V: VTable> Debug for ArrayAdapter<V> {
@@ -454,7 +459,7 @@ impl<V: VTable> DynArray for ArrayAdapter<V> {
         V::dtype(&self.0)
     }
 
-    fn vtable(&self) -> &dyn DynVTable {
+    fn vtable(&self) -> &'static dyn DynVTable {
         V::vtable()
     }
 
diff --git a/vortex-array/src/arrays/bool/vtable/mod.rs b/vortex-array/src/arrays/bool/vtable/mod.rs
index 9063bd384bc..1a6ca30eaab 100644
--- a/vortex-array/src/arrays/bool/vtable/mod.rs
+++ b/vortex-array/src/arrays/bool/vtable/mod.rs
@@ -12,8 +12,7 @@ use vortex_session::VortexSession;
 use crate::ArrayRef;
 use crate::DeserializeMetadata;
 use crate::ExecutionCtx;
-use crate::ExecutionStep;
-use crate::IntoArray;
+use crate::ExecutionResult;
 use crate::ProstMetadata;
 use crate::SerializeMetadata;
 use crate::arrays::BoolArray;
@@ -32,6 +31,7 @@ mod operations;
 mod validity;
 
 use std::hash::Hash;
+use std::sync::Arc;
 
 use crate::Precision;
 use crate::arrays::bool::compute::rules::RULES;
@@ -39,6 +39,7 @@ use crate::hash::ArrayEq;
 use crate::hash::ArrayHash;
 use crate::stats::StatsSetRef;
 use crate::vtable::ArrayId;
+use crate::vtable::upcast_array;
 
 vtable!(Bool);
 
@@ -185,8 +186,8 @@ impl VTable for BoolVTable {
         Ok(())
     }
 
-    fn execute(array: &Self::Array, _ctx: &mut ExecutionCtx) -> VortexResult<ExecutionStep> {
-        Ok(ExecutionStep::Done(array.clone().into_array()))
+    fn execute(array: Arc<Self::Array>, _ctx: &mut ExecutionCtx) -> VortexResult<ExecutionResult> {
+        Ok(ExecutionResult::done(upcast_array::<Self>(array)))
     }
 
     fn reduce_parent(
diff --git a/vortex-array/src/arrays/chunked/vtable/mod.rs b/vortex-array/src/arrays/chunked/vtable/mod.rs
index d5b6b6259da..e972d6feb0e 100644
--- a/vortex-array/src/arrays/chunked/vtable/mod.rs
+++ b/vortex-array/src/arrays/chunked/vtable/mod.rs
@@ -2,6 +2,7 @@
 // SPDX-FileCopyrightText: Copyright the Vortex contributors
 
 use std::hash::Hash;
+use std::sync::Arc;
 
 use itertools::Itertools;
 use vortex_error::VortexResult;
@@ -15,7 +16,7 @@ use crate::ArrayRef;
 use crate::Canonical;
 use crate::EmptyMetadata;
 use crate::ExecutionCtx;
-use crate::ExecutionStep;
+use crate::ExecutionResult;
 use crate::IntoArray;
 use crate::Precision;
 use crate::ToCanonical;
@@ -240,8 +241,10 @@ impl VTable for ChunkedVTable {
         Ok(())
     }
 
-    fn execute(array: &Self::Array, ctx: &mut ExecutionCtx) -> VortexResult<ExecutionStep> {
-        Ok(ExecutionStep::Done(_canonicalize(array, ctx)?.into_array()))
+    fn execute(array: Arc<Self::Array>, ctx: &mut ExecutionCtx) -> VortexResult<ExecutionResult> {
+        Ok(ExecutionResult::done(
+            _canonicalize(&array, ctx)?.into_array(),
+        ))
     }
 
     fn reduce(array: &Self::Array) -> VortexResult<Option<ArrayRef>> {
diff --git a/vortex-array/src/arrays/constant/vtable/mod.rs b/vortex-array/src/arrays/constant/vtable/mod.rs
index 12fee7869d2..f011ed3ff20 100644
--- a/vortex-array/src/arrays/constant/vtable/mod.rs
+++ b/vortex-array/src/arrays/constant/vtable/mod.rs
@@ -3,6 +3,7 @@
 
 use std::fmt::Debug;
 use std::hash::Hash;
+use std::sync::Arc;
 
 use vortex_buffer::ByteBufferMut;
 use vortex_error::VortexExpect;
@@ -13,7 +14,7 @@ use vortex_session::VortexSession;
 
 use crate::ArrayRef;
 use crate::ExecutionCtx;
-use crate::ExecutionStep;
+use crate::ExecutionResult;
 use crate::IntoArray;
 use crate::Precision;
 use crate::arrays::ConstantArray;
@@ -178,9 +179,9 @@ impl VTable for ConstantVTable {
         PARENT_RULES.evaluate(array, parent, child_idx)
     }
 
-    fn execute(array: &Self::Array, _ctx: &mut ExecutionCtx) -> VortexResult<ExecutionStep> {
-        Ok(ExecutionStep::Done(
-            constant_canonicalize(array)?.into_array(),
+    fn execute(array: Arc<Self::Array>, _ctx: &mut ExecutionCtx) -> VortexResult<ExecutionResult> {
+        Ok(ExecutionResult::done(
+            constant_canonicalize(&array)?.into_array(),
         ))
     }
 
diff --git a/vortex-array/src/arrays/decimal/vtable/mod.rs b/vortex-array/src/arrays/decimal/vtable/mod.rs
index 7eb78acb832..24289693ef1 100644
--- a/vortex-array/src/arrays/decimal/vtable/mod.rs
+++ b/vortex-array/src/arrays/decimal/vtable/mod.rs
@@ -13,8 +13,7 @@ use vortex_session::VortexSession;
 use crate::ArrayRef;
 use crate::DeserializeMetadata;
 use crate::ExecutionCtx;
-use crate::ExecutionStep;
-use crate::IntoArray;
+use crate::ExecutionResult;
 use crate::ProstMetadata;
 use crate::SerializeMetadata;
 use crate::arrays::DecimalArray;
@@ -35,6 +34,7 @@ mod operations;
 mod validity;
 
 use std::hash::Hash;
+use std::sync::Arc;
 
 use crate::Precision;
 use crate::arrays::decimal::compute::rules::RULES;
@@ -42,6 +42,7 @@ use crate::hash::ArrayEq;
 use crate::hash::ArrayHash;
 use crate::stats::StatsSetRef;
 use crate::vtable::ArrayId;
+use crate::vtable::upcast_array;
 vtable!(Decimal);
 
 // The type of the values can be determined by looking at the type info...right?
@@ -207,8 +208,8 @@ impl VTable for DecimalVTable {
         Ok(())
     }
 
-    fn execute(array: &Self::Array, _ctx: &mut ExecutionCtx) -> VortexResult<ExecutionStep> {
-        Ok(ExecutionStep::Done(array.clone().into_array()))
+    fn execute(array: Arc<Self::Array>, _ctx: &mut ExecutionCtx) -> VortexResult<ExecutionResult> {
+        Ok(ExecutionResult::done(upcast_array::<Self>(array)))
     }
 
     fn reduce_parent(
diff --git a/vortex-array/src/arrays/dict/vtable/mod.rs b/vortex-array/src/arrays/dict/vtable/mod.rs
index ffde0e4223b..6b184ac3c69 100644
--- a/vortex-array/src/arrays/dict/vtable/mod.rs
+++ b/vortex-array/src/arrays/dict/vtable/mod.rs
@@ -2,6 +2,7 @@
 // SPDX-FileCopyrightText: Copyright the Vortex contributors
 
 use std::hash::Hash;
+use std::sync::Arc;
 
 use kernel::PARENT_KERNELS;
 use vortex_error::VortexResult;
@@ -29,7 +30,7 @@ use crate::dtype::DType;
 use crate::dtype::Nullability;
 use crate::dtype::PType;
 use crate::executor::ExecutionCtx;
-use crate::executor::ExecutionStep;
+use crate::executor::ExecutionResult;
 use crate::hash::ArrayEq;
 use crate::hash::ArrayHash;
 use crate::scalar::Scalar;
@@ -191,9 +192,9 @@ impl VTable for DictVTable {
         Ok(())
     }
 
-    fn execute(array: &Self::Array, ctx: &mut ExecutionCtx) -> VortexResult<ExecutionStep> {
-        if let Some(canonical) = execute_fast_path(array, ctx)? {
-            return Ok(ExecutionStep::Done(canonical));
+    fn execute(array: Arc<Self::Array>, ctx: &mut ExecutionCtx) -> VortexResult<ExecutionResult> {
+        if let Some(canonical) = execute_fast_path(&array, ctx)? {
+            return Ok(ExecutionResult::done(canonical));
         }
 
         // TODO(joe): if the values are constant return a constant
@@ -209,7 +210,7 @@ impl VTable for DictVTable {
         // TODO(ngates): if indices min is quite high, we could slice self and offset the indices
         //  such that canonicalize does less work.
 
-        Ok(ExecutionStep::Done(
+        Ok(ExecutionResult::done(
             take_canonical(values, &codes, ctx)?.into_array(),
         ))
     }
diff --git a/vortex-array/src/arrays/extension/vtable/mod.rs b/vortex-array/src/arrays/extension/vtable/mod.rs
index e99db442573..397d1b4a18b 100644
--- a/vortex-array/src/arrays/extension/vtable/mod.rs
+++ b/vortex-array/src/arrays/extension/vtable/mod.rs
@@ -6,6 +6,7 @@ mod operations;
 mod validity;
 
 use std::hash::Hash;
+use std::sync::Arc;
 
 use kernel::PARENT_KERNELS;
 use vortex_error::VortexExpect;
@@ -18,8 +19,7 @@ use vortex_session::VortexSession;
 use crate::ArrayRef;
 use crate::EmptyMetadata;
 use crate::ExecutionCtx;
-use crate::ExecutionStep;
-use crate::IntoArray;
+use crate::ExecutionResult;
 use crate::Precision;
 use crate::arrays::ExtensionArray;
 use crate::arrays::extension::compute::rules::PARENT_RULES;
@@ -33,6 +33,7 @@ use crate::vtable;
 use crate::vtable::ArrayId;
 use crate::vtable::VTable;
 use crate::vtable::ValidityVTableFromChild;
+use crate::vtable::upcast_array;
 
 vtable!(Extension);
 
@@ -150,8 +151,8 @@ impl VTable for ExtensionVTable {
         Ok(())
     }
 
-    fn execute(array: &Self::Array, _ctx: &mut ExecutionCtx) -> VortexResult<ExecutionStep> {
-        Ok(ExecutionStep::Done(array.clone().into_array()))
+    fn execute(array: Arc<Self::Array>, _ctx: &mut ExecutionCtx) -> VortexResult<ExecutionResult> {
+        Ok(ExecutionResult::done(upcast_array::<Self>(array)))
     }
 
     fn reduce_parent(
diff --git a/vortex-array/src/arrays/filter/vtable.rs b/vortex-array/src/arrays/filter/vtable.rs
index 28d31c60d2b..1cb736457d4 100644
--- a/vortex-array/src/arrays/filter/vtable.rs
+++ b/vortex-array/src/arrays/filter/vtable.rs
@@ -4,6 +4,7 @@
 use std::fmt::Debug;
 use std::fmt::Formatter;
 use std::hash::Hasher;
+use std::sync::Arc;
 
 use vortex_error::VortexExpect;
 use vortex_error::VortexResult;
@@ -27,7 +28,7 @@ use crate::arrays::filter::rules::RULES;
 use crate::buffer::BufferHandle;
 use crate::dtype::DType;
 use crate::executor::ExecutionCtx;
-use crate::executor::ExecutionStep;
+use crate::executor::ExecutionResult;
 use crate::scalar::Scalar;
 use crate::serde::ArrayChildren;
 use crate::stats::StatsSetRef;
@@ -155,9 +156,9 @@ impl VTable for FilterVTable {
         Ok(())
     }
 
-    fn execute(array: &Self::Array, ctx: &mut ExecutionCtx) -> VortexResult<ExecutionStep> {
-        if let Some(canonical) = execute_filter_fast_paths(array, ctx)? {
-            return Ok(ExecutionStep::Done(canonical));
+    fn execute(array: Arc<Self::Array>, ctx: &mut ExecutionCtx) -> VortexResult<ExecutionResult> {
+        if let Some(canonical) = execute_filter_fast_paths(&array, ctx)? {
+            return Ok(ExecutionResult::done(canonical));
         }
         let Mask::Values(mask_values) = &array.mask else {
             unreachable!("`execute_filter_fast_paths` handles AllTrue and AllFalse")
@@ -165,7 +166,7 @@ impl VTable for FilterVTable {
 
         // We rely on the optimization pass that runs prior to this execution for filter pushdown,
         // so now we can just execute the filter without worrying.
-        Ok(ExecutionStep::Done(
+        Ok(ExecutionResult::done(
             execute_filter(array.child.clone().execute(ctx)?, mask_values).into_array(),
         ))
     }
diff --git a/vortex-array/src/arrays/fixed_size_list/vtable/mod.rs b/vortex-array/src/arrays/fixed_size_list/vtable/mod.rs
index 6922bd12e7f..023110156f0 100644
--- a/vortex-array/src/arrays/fixed_size_list/vtable/mod.rs
+++ b/vortex-array/src/arrays/fixed_size_list/vtable/mod.rs
@@ -2,6 +2,7 @@
 // SPDX-FileCopyrightText: Copyright the Vortex contributors
 
 use std::hash::Hash;
+use std::sync::Arc;
 
 use vortex_error::VortexExpect;
 use vortex_error::VortexResult;
@@ -13,8 +14,7 @@ use vortex_session::VortexSession;
 use crate::ArrayRef;
 use crate::EmptyMetadata;
 use crate::ExecutionCtx;
-use crate::ExecutionStep;
-use crate::IntoArray;
+use crate::ExecutionResult;
 use crate::Precision;
 use crate::arrays::FixedSizeListArray;
 use crate::arrays::fixed_size_list::compute::rules::PARENT_RULES;
@@ -29,6 +29,7 @@ use crate::vtable;
 use crate::vtable::ArrayId;
 use crate::vtable::VTable;
 use crate::vtable::ValidityVTableFromValidityHelper;
+use crate::vtable::upcast_array;
 use crate::vtable::validity_nchildren;
 use crate::vtable::validity_to_child;
 mod kernel;
@@ -219,7 +220,7 @@ impl VTable for FixedSizeListVTable {
         Ok(())
     }
 
-    fn execute(array: &Self::Array, _ctx: &mut ExecutionCtx) -> VortexResult<ExecutionStep> {
-        Ok(ExecutionStep::Done(array.clone().into_array()))
+    fn execute(array: Arc<Self::Array>, _ctx: &mut ExecutionCtx) -> VortexResult<ExecutionResult> {
+        Ok(ExecutionResult::done(upcast_array::<Self>(array)))
     }
 }
diff --git a/vortex-array/src/arrays/list/vtable/mod.rs b/vortex-array/src/arrays/list/vtable/mod.rs
index 13c5afcd2e0..6ba4819b795 100644
--- a/vortex-array/src/arrays/list/vtable/mod.rs
+++ b/vortex-array/src/arrays/list/vtable/mod.rs
@@ -2,6 +2,7 @@
 // SPDX-FileCopyrightText: Copyright the Vortex contributors
 
 use std::hash::Hash;
+use std::sync::Arc;
 
 use vortex_error::VortexExpect;
 use vortex_error::VortexResult;
@@ -13,7 +14,7 @@ use vortex_session::VortexSession;
 use crate::ArrayRef;
 use crate::DynArray;
 use crate::ExecutionCtx;
-use crate::ExecutionStep;
+use crate::ExecutionResult;
 use crate::IntoArray;
 use crate::Precision;
 use crate::ProstMetadata;
@@ -211,9 +212,9 @@ impl VTable for ListVTable {
         Ok(())
     }
 
-    fn execute(array: &Self::Array, ctx: &mut ExecutionCtx) -> VortexResult<ExecutionStep> {
-        Ok(ExecutionStep::Done(
-            list_view_from_list(array.clone(), ctx)?.into_array(),
+    fn execute(array: Arc<Self::Array>, ctx: &mut ExecutionCtx) -> VortexResult<ExecutionResult> {
+        Ok(ExecutionResult::done(
+            list_view_from_list(ListArray::clone(&array), ctx)?.into_array(),
         ))
     }
 
diff --git a/vortex-array/src/arrays/listview/vtable/mod.rs b/vortex-array/src/arrays/listview/vtable/mod.rs
index 45fd1f96cfb..9288d95edfb 100644
--- a/vortex-array/src/arrays/listview/vtable/mod.rs
+++ b/vortex-array/src/arrays/listview/vtable/mod.rs
@@ -2,6 +2,7 @@
 // SPDX-FileCopyrightText: Copyright the Vortex contributors
 
 use std::hash::Hash;
+use std::sync::Arc;
 
 use vortex_error::VortexExpect;
 use vortex_error::VortexResult;
@@ -13,8 +14,7 @@ use vortex_session::VortexSession;
 use crate::ArrayRef;
 use crate::DeserializeMetadata;
 use crate::ExecutionCtx;
-use crate::ExecutionStep;
-use crate::IntoArray;
+use crate::ExecutionResult;
 use crate::Precision;
 use crate::ProstMetadata;
 use crate::SerializeMetadata;
@@ -33,6 +33,7 @@ use crate::vtable;
 use crate::vtable::ArrayId;
 use crate::vtable::VTable;
 use crate::vtable::ValidityVTableFromValidityHelper;
+use crate::vtable::upcast_array;
 use crate::vtable::validity_nchildren;
 use crate::vtable::validity_to_child;
 mod operations;
@@ -239,8 +240,8 @@ impl VTable for ListViewVTable {
         Ok(())
     }
 
-    fn execute(array: &Self::Array, _ctx: &mut ExecutionCtx) -> VortexResult<ExecutionStep> {
-        Ok(ExecutionStep::Done(array.clone().into_array()))
+    fn execute(array: Arc<Self::Array>, _ctx: &mut ExecutionCtx) -> VortexResult<ExecutionResult> {
+        Ok(ExecutionResult::done(upcast_array::<Self>(array)))
     }
 
     fn reduce_parent(
diff --git a/vortex-array/src/arrays/masked/vtable/mod.rs b/vortex-array/src/arrays/masked/vtable/mod.rs
index dafd40a50ee..36c4e41b38d 100644
--- a/vortex-array/src/arrays/masked/vtable/mod.rs
+++ b/vortex-array/src/arrays/masked/vtable/mod.rs
@@ -5,6 +5,7 @@ mod operations;
 mod validity;
 
 use std::hash::Hash;
+use std::sync::Arc;
 
 use vortex_error::VortexExpect;
 use vortex_error::VortexResult;
@@ -25,7 +26,7 @@ use crate::arrays::masked::mask_validity_canonical;
 use crate::buffer::BufferHandle;
 use crate::dtype::DType;
 use crate::executor::ExecutionCtx;
-use crate::executor::ExecutionStep;
+use crate::executor::ExecutionResult;
 use crate::hash::ArrayEq;
 use crate::hash::ArrayHash;
 use crate::scalar::Scalar;
@@ -161,12 +162,12 @@ impl VTable for MaskedVTable {
         MaskedArray::try_new(child, validity)
     }
 
-    fn execute(array: &Self::Array, ctx: &mut ExecutionCtx) -> VortexResult<ExecutionStep> {
+    fn execute(array: Arc<Self::Array>, ctx: &mut ExecutionCtx) -> VortexResult<ExecutionResult> {
         let validity_mask = array.validity_mask()?;
 
         // Fast path: all masked means result is all nulls.
         if validity_mask.all_false() {
-            return Ok(ExecutionStep::Done(
+            return Ok(ExecutionResult::done(
                 ConstantArray::new(Scalar::null(array.dtype().as_nullable()), array.len())
                     .into_array(),
             ));
@@ -179,7 +180,7 @@ impl VTable for MaskedVTable {
         // `AllTrue` masks (no data copying), so there's no benefit.
 
         let child = array.child().clone().execute::<Canonical>(ctx)?;
-        Ok(ExecutionStep::Done(
+        Ok(ExecutionResult::done(
             mask_validity_canonical(child, &validity_mask, ctx)?.into_array(),
         ))
     }
diff --git a/vortex-array/src/arrays/null/mod.rs b/vortex-array/src/arrays/null/mod.rs
index 5b83164333a..ab291c5489f 100644
--- a/vortex-array/src/arrays/null/mod.rs
+++ b/vortex-array/src/arrays/null/mod.rs
@@ -2,6 +2,7 @@
 // SPDX-FileCopyrightText: Copyright the Vortex contributors
 
 use std::hash::Hash;
+use std::sync::Arc;
 
 use vortex_error::VortexResult;
 use vortex_error::vortex_ensure;
@@ -11,8 +12,7 @@ use vortex_session::VortexSession;
 use crate::ArrayRef;
 use crate::EmptyMetadata;
 use crate::ExecutionCtx;
-use crate::ExecutionStep;
-use crate::IntoArray;
+use crate::ExecutionResult;
 use crate::Precision;
 use crate::arrays::null::compute::rules::PARENT_RULES;
 use crate::buffer::BufferHandle;
@@ -27,6 +27,7 @@ use crate::vtable::ArrayId;
 use crate::vtable::OperationsVTable;
 use crate::vtable::VTable;
 use crate::vtable::ValidityVTable;
+use crate::vtable::upcast_array;
 
 pub(crate) mod compute;
 
@@ -132,8 +133,8 @@ impl VTable for NullVTable {
         PARENT_RULES.evaluate(array, parent, child_idx)
     }
 
-    fn execute(array: &Self::Array, _ctx: &mut ExecutionCtx) -> VortexResult<ExecutionStep> {
-        Ok(ExecutionStep::Done(array.clone().into_array()))
+    fn execute(array: Arc<Self::Array>, _ctx: &mut ExecutionCtx) -> VortexResult<ExecutionResult> {
+        Ok(ExecutionResult::done(upcast_array::<Self>(array)))
     }
 }
 
diff --git a/vortex-array/src/arrays/primitive/vtable/mod.rs b/vortex-array/src/arrays/primitive/vtable/mod.rs
index 3418eb1f822..612c34f5bfb 100644
--- a/vortex-array/src/arrays/primitive/vtable/mod.rs
+++ b/vortex-array/src/arrays/primitive/vtable/mod.rs
@@ -11,8 +11,7 @@ use vortex_error::vortex_panic;
 use crate::ArrayRef;
 use crate::EmptyMetadata;
 use crate::ExecutionCtx;
-use crate::ExecutionStep;
-use crate::IntoArray;
+use crate::ExecutionResult;
 use crate::arrays::PrimitiveArray;
 use crate::buffer::BufferHandle;
 use crate::dtype::DType;
@@ -30,6 +29,7 @@ mod validity;
 
 use std::hash::Hash;
 use std::hash::Hasher;
+use std::sync::Arc;
 
 use vortex_buffer::Alignment;
 use vortex_session::VortexSession;
@@ -40,6 +40,7 @@ use crate::hash::ArrayEq;
 use crate::hash::ArrayHash;
 use crate::stats::StatsSetRef;
 use crate::vtable::ArrayId;
+use crate::vtable::upcast_array;
 
 vtable!(Primitive);
 
@@ -199,8 +200,8 @@ impl VTable for PrimitiveVTable {
         Ok(())
     }
 
-    fn execute(array: &Self::Array, _ctx: &mut ExecutionCtx) -> VortexResult<ExecutionStep> {
-        Ok(ExecutionStep::Done(array.clone().into_array()))
+    fn execute(array: Arc<Self::Array>, _ctx: &mut ExecutionCtx) -> VortexResult<ExecutionResult> {
+        Ok(ExecutionResult::done(upcast_array::<Self>(array)))
     }
 
     fn reduce_parent(
diff --git a/vortex-array/src/arrays/scalar_fn/vtable/mod.rs b/vortex-array/src/arrays/scalar_fn/vtable/mod.rs
index c1dca2b3168..46cd0ba7ef4 100644
--- a/vortex-array/src/arrays/scalar_fn/vtable/mod.rs
+++ b/vortex-array/src/arrays/scalar_fn/vtable/mod.rs
@@ -8,6 +8,7 @@ use std::hash::Hash;
 use std::hash::Hasher;
 use std::marker::PhantomData;
 use std::ops::Deref;
+use std::sync::Arc;
 
 use itertools::Itertools;
 use vortex_error::VortexExpect;
@@ -30,7 +31,7 @@ use crate::arrays::scalar_fn::rules::RULES;
 use crate::buffer::BufferHandle;
 use crate::dtype::DType;
 use crate::executor::ExecutionCtx;
-use crate::executor::ExecutionStep;
+use crate::executor::ExecutionResult;
 use crate::expr::Expression;
 use crate::matcher::Matcher;
 use crate::scalar_fn;
@@ -195,10 +196,13 @@ impl VTable for ScalarFnVTable {
         Ok(())
     }
 
-    fn execute(array: &Self::Array, ctx: &mut ExecutionCtx) -> VortexResult<ExecutionStep> {
+    fn execute(array: Arc<Self::Array>, ctx: &mut ExecutionCtx) -> VortexResult<ExecutionResult> {
         ctx.log(format_args!("scalar_fn({}): executing", array.scalar_fn));
         let args = VecExecutionArgs::new(array.children.clone(), array.len);
-        array.scalar_fn.execute(&args, ctx).map(ExecutionStep::Done)
+        array
+            .scalar_fn
+            .execute(&args, ctx)
+            .map(ExecutionResult::done)
     }
 
     fn reduce(array: &Self::Array) -> VortexResult<Option<ArrayRef>> {
diff --git a/vortex-array/src/arrays/shared/vtable.rs b/vortex-array/src/arrays/shared/vtable.rs
index 12106400338..857b0bc3739 100644
--- a/vortex-array/src/arrays/shared/vtable.rs
+++ b/vortex-array/src/arrays/shared/vtable.rs
@@ -2,6 +2,7 @@
 // SPDX-FileCopyrightText: Copyright the Vortex contributors
 
 use std::hash::Hash;
+use std::sync::Arc;
 
 use vortex_error::VortexExpect;
 use vortex_error::VortexResult;
@@ -12,7 +13,7 @@ use crate::ArrayRef;
 use crate::Canonical;
 use crate::EmptyMetadata;
 use crate::ExecutionCtx;
-use crate::ExecutionStep;
+use crate::ExecutionResult;
 use crate::Precision;
 use crate::arrays::SharedArray;
 use crate::buffer::BufferHandle;
@@ -145,10 +146,10 @@ impl VTable for SharedVTable {
         Ok(())
     }
 
-    fn execute(array: &Self::Array, ctx: &mut ExecutionCtx) -> VortexResult<ExecutionStep> {
+    fn execute(array: Arc<Self::Array>, ctx: &mut ExecutionCtx) -> VortexResult<ExecutionResult> {
         array
             .get_or_compute(|source| source.clone().execute::<Canonical>(ctx))
-            .map(ExecutionStep::Done)
+            .map(ExecutionResult::done)
     }
 }
 impl OperationsVTable<SharedVTable> for SharedVTable {
diff --git a/vortex-array/src/arrays/slice/vtable.rs b/vortex-array/src/arrays/slice/vtable.rs
index a8aaafb1d61..d24fc125c08 100644
--- a/vortex-array/src/arrays/slice/vtable.rs
+++ b/vortex-array/src/arrays/slice/vtable.rs
@@ -6,6 +6,7 @@ use std::fmt::Formatter;
 use std::hash::Hash;
 use std::hash::Hasher;
 use std::ops::Range;
+use std::sync::Arc;
 
 use vortex_error::VortexExpect;
 use vortex_error::VortexResult;
@@ -26,7 +27,7 @@ use crate::arrays::slice::rules::PARENT_RULES;
 use crate::buffer::BufferHandle;
 use crate::dtype::DType;
 use crate::executor::ExecutionCtx;
-use crate::executor::ExecutionStep;
+use crate::executor::ExecutionResult;
 use crate::scalar::Scalar;
 use crate::serde::ArrayChildren;
 use crate::stats::StatsSetRef;
@@ -155,7 +156,7 @@ impl VTable for SliceVTable {
         Ok(())
     }
 
-    fn execute(array: &Self::Array, ctx: &mut ExecutionCtx) -> VortexResult<ExecutionStep> {
+    fn execute(array: Arc<Self::Array>, ctx: &mut ExecutionCtx) -> VortexResult<ExecutionResult> {
         // Execute the child to get canonical form, then slice it
         let Some(canonical) = array.child.as_opt::<AnyCanonical>() else {
             // If the child is not canonical, recurse.
@@ -164,14 +165,14 @@ impl VTable for SliceVTable {
                 .clone()
                 .execute::<ArrayRef>(ctx)?
                 .slice(array.slice_range().clone())
-                .map(ExecutionStep::Done);
+                .map(ExecutionResult::done);
         };
 
         // TODO(ngates): we should inline canonical slice logic here.
         Canonical::from(canonical)
             .as_ref()
             .slice(array.range.clone())
-            .map(ExecutionStep::Done)
+            .map(ExecutionResult::done)
     }
 
     fn reduce_parent(
diff --git a/vortex-array/src/arrays/struct_/vtable/mod.rs b/vortex-array/src/arrays/struct_/vtable/mod.rs
index bbc35b0a67c..44bcc9ed16b 100644
--- a/vortex-array/src/arrays/struct_/vtable/mod.rs
+++ b/vortex-array/src/arrays/struct_/vtable/mod.rs
@@ -15,8 +15,7 @@ use vortex_session::VortexSession;
 use crate::ArrayRef;
 use crate::EmptyMetadata;
 use crate::ExecutionCtx;
-use crate::ExecutionStep;
-use crate::IntoArray;
+use crate::ExecutionResult;
 use crate::arrays::StructArray;
 use crate::arrays::struct_::compute::rules::PARENT_RULES;
 use crate::buffer::BufferHandle;
@@ -38,6 +37,7 @@ use crate::hash::ArrayEq;
 use crate::hash::ArrayHash;
 use crate::stats::StatsSetRef;
 use crate::vtable::ArrayId;
+use crate::vtable::upcast_array;
 
 vtable!(Struct);
 
@@ -207,8 +207,8 @@ impl VTable for StructVTable {
         Ok(())
     }
 
-    fn execute(array: &Self::Array, _ctx: &mut ExecutionCtx) -> VortexResult<ExecutionStep> {
-        Ok(ExecutionStep::Done(array.clone().into_array()))
+    fn execute(array: Arc<Self::Array>, _ctx: &mut ExecutionCtx) -> VortexResult<ExecutionResult> {
+        Ok(ExecutionResult::done(upcast_array::<Self>(array)))
     }
 
     fn reduce_parent(
diff --git a/vortex-array/src/arrays/varbin/vtable/mod.rs b/vortex-array/src/arrays/varbin/vtable/mod.rs
index 3f35cd793c4..02e4a2f1e9c 100644
--- a/vortex-array/src/arrays/varbin/vtable/mod.rs
+++ b/vortex-array/src/arrays/varbin/vtable/mod.rs
@@ -10,7 +10,7 @@ use vortex_error::vortex_panic;
 use crate::ArrayRef;
 use crate::DeserializeMetadata;
 use crate::ExecutionCtx;
-use crate::ExecutionStep;
+use crate::ExecutionResult;
 use crate::IntoArray;
 use crate::ProstMetadata;
 use crate::SerializeMetadata;
@@ -32,6 +32,7 @@ mod kernel;
 mod operations;
 mod validity;
 use std::hash::Hash;
+use std::sync::Arc;
 
 use canonical::varbin_to_canonical;
 use kernel::PARENT_KERNELS;
@@ -219,9 +220,9 @@ impl VTable for VarBinVTable {
         PARENT_KERNELS.execute(array, parent, child_idx, ctx)
     }
 
-    fn execute(array: &Self::Array, ctx: &mut ExecutionCtx) -> VortexResult<ExecutionStep> {
-        Ok(ExecutionStep::Done(
-            varbin_to_canonical(array, ctx)?.into_array(),
+    fn execute(array: Arc<Self::Array>, ctx: &mut ExecutionCtx) -> VortexResult<ExecutionResult> {
+        Ok(ExecutionResult::done(
+            varbin_to_canonical(&array, ctx)?.into_array(),
         ))
     }
 }
diff --git a/vortex-array/src/arrays/varbinview/vtable/mod.rs b/vortex-array/src/arrays/varbinview/vtable/mod.rs
index 9df6a68636c..d85c21c18d5 100644
--- a/vortex-array/src/arrays/varbinview/vtable/mod.rs
+++ b/vortex-array/src/arrays/varbinview/vtable/mod.rs
@@ -17,8 +17,7 @@ use vortex_session::VortexSession;
 use crate::ArrayRef;
 use crate::EmptyMetadata;
 use crate::ExecutionCtx;
-use crate::ExecutionStep;
-use crate::IntoArray;
+use crate::ExecutionResult;
 use crate::Precision;
 use crate::arrays::VarBinViewArray;
 use crate::arrays::varbinview::BinaryView;
@@ -34,6 +33,7 @@ use crate::vtable;
 use crate::vtable::ArrayId;
 use crate::vtable::VTable;
 use crate::vtable::ValidityVTableFromValidityHelper;
+use crate::vtable::upcast_array;
 use crate::vtable::validity_nchildren;
 use crate::vtable::validity_to_child;
 mod kernel;
@@ -243,7 +243,7 @@ impl VTable for VarBinViewVTable {
         PARENT_KERNELS.execute(array, parent, child_idx, ctx)
     }
 
-    fn execute(array: &Self::Array, _ctx: &mut ExecutionCtx) -> VortexResult<ExecutionStep> {
-        Ok(ExecutionStep::Done(array.clone().into_array()))
+    fn execute(array: Arc<Self::Array>, _ctx: &mut ExecutionCtx) -> VortexResult<ExecutionResult> {
+        Ok(ExecutionResult::done(upcast_array::<Self>(array)))
     }
 }
diff --git a/vortex-array/src/executor.rs b/vortex-array/src/executor.rs
index da05450f8de..ccacb3b6e8a 100644
--- a/vortex-array/src/executor.rs
+++ b/vortex-array/src/executor.rs
@@ -141,22 +141,24 @@ impl dyn DynArray + '_ {
                 continue;
             }
 
-            // Execute the array itself
-            match current.vtable().execute(&current, ctx)? {
+            // Execute the array itself.
+            let result = execute_step(current, ctx)?;
+            let (array, step) = result.into_parts();
+            match step {
                 ExecutionStep::ExecuteChild(i, done) => {
-                    let child = current
+                    let child = array
                         .nth_child(i)
                         .vortex_expect("ExecuteChild index in bounds");
                     ctx.log(format_args!(
                         "ExecuteChild({i}): pushing {}, focusing on {}",
-                        current, child
+                        array, child
                     ));
-                    stack.push((current, i, done));
+                    stack.push((array, i, done));
                     current = child.optimize()?;
                 }
-                ExecutionStep::Done(result) => {
-                    ctx.log(format_args!("Done: {} -> {}", current, result));
-                    current = result;
+                ExecutionStep::Done => {
+                    ctx.log(format_args!("Done: {}", array));
+                    current = array;
                 }
             }
         }
@@ -305,12 +307,14 @@ impl Executable for ArrayRef {
             }
         }
 
-        // 4. execute (returns an ExecutionStep)
+        // 4. execute (returns an ExecutionResult)
         ctx.log(format_args!("executing {}", array));
-        match array.vtable().execute(&array, ctx)? {
-            ExecutionStep::Done(result) => {
-                ctx.log(format_args!("-> {}", result.as_ref()));
-                Ok(result)
+        let result = execute_step(array, ctx)?;
+        let (array, step) = result.into_parts();
+        match step {
+            ExecutionStep::Done => {
+                ctx.log(format_args!("-> {}", array.as_ref()));
+                Ok(array)
             }
             ExecutionStep::ExecuteChild(i, _) => {
                 // For single-step execution, handle ExecuteChild by executing the child,
@@ -323,6 +327,14 @@ impl Executable for ArrayRef {
     }
 }
 
+/// Execute a single step on an array, consuming it.
+///
+/// Extracts the vtable before consuming the array to avoid borrow conflicts.
+fn execute_step(array: ArrayRef, ctx: &mut ExecutionCtx) -> VortexResult<ExecutionResult> {
+    let vtable = array.vtable();
+    vtable.execute(array, ctx)
+}
+
 /// Try execute_parent on each child of the array.
 fn try_execute_parent(array: &ArrayRef, ctx: &mut ExecutionCtx) -> VortexResult<Option<ArrayRef>> {
     for child_idx in 0..array.nchildren() {
@@ -343,7 +355,7 @@ fn try_execute_parent(array: &ArrayRef, ctx: &mut ExecutionCtx) -> VortexResult<
 /// A predicate that determines when an array has reached a desired form during execution.
 pub type DonePredicate = fn(&dyn DynArray) -> bool;
 
-/// The result of a single execution step on an array encoding.
+/// Metadata-only step indicator returned alongside an array in [`ExecutionResult`].
 ///
 /// Instead of recursively executing children, encodings return an `ExecutionStep` that tells the
 /// scheduler what to do next. This enables the scheduler to manage execution iteratively using
@@ -355,25 +367,11 @@ pub enum ExecutionStep {
     ///
     /// Between steps, the scheduler runs reduce/reduce_parent rules to fixpoint, enabling
     /// cross-step optimization (e.g., pushing scalar functions through newly-decoded children).
-    ///
-    /// Use [`ExecutionStep::execute_child`] instead of constructing this variant directly.
     ExecuteChild(usize, DonePredicate),
 
-    /// Execution is complete. The result may be in any encoding — not necessarily canonical.
-    /// The scheduler will continue executing the result if it has not yet reached the target form.
-    Done(ArrayRef),
-}
-
-impl ExecutionStep {
-    /// Request execution of child at `child_idx` until it matches the given [`Matcher`].
-    pub fn execute_child<M: Matcher>(child_idx: usize) -> Self {
-        ExecutionStep::ExecuteChild(child_idx, M::matches)
-    }
-
-    /// Signal that execution is complete with the given result.
-    pub fn done(result: ArrayRef) -> Self {
-        ExecutionStep::Done(result)
-    }
+    /// Execution is complete. The array in the accompanying [`ExecutionResult`] is the result.
+    /// The scheduler will continue executing if it has not yet reached the target form.
+    Done,
 }
 
 impl fmt::Debug for ExecutionStep {
@@ -382,9 +380,62 @@ impl fmt::Debug for ExecutionStep {
             ExecutionStep::ExecuteChild(idx, _) => {
                 f.debug_tuple("ExecuteChild").field(idx).finish()
             }
-            ExecutionStep::Done(result) => f.debug_tuple("Done").field(result).finish(),
+            ExecutionStep::Done => write!(f, "Done"),
+        }
+    }
+}
+
+/// The result of a single execution step on an array encoding.
+///
+/// Combines an [`ArrayRef`] with an [`ExecutionStep`] to tell the scheduler both what to do next
+/// and what array to work with.
+pub struct ExecutionResult {
+    array: ArrayRef,
+    step: ExecutionStep,
+}
+
+impl ExecutionResult {
+    /// Signal that execution is complete with the given result array.
+    pub fn done(result: impl IntoArray) -> Self {
+        Self {
+            array: result.into_array(),
+            step: ExecutionStep::Done,
         }
     }
+
+    /// Request execution of child at `child_idx` until it matches the given [`Matcher`].
+    ///
+    /// The provided array is the (possibly modified) parent that still needs its child executed.
+    pub fn execute_child<M: Matcher>(array: impl IntoArray, child_idx: usize) -> Self {
+        Self {
+            array: array.into_array(),
+            step: ExecutionStep::ExecuteChild(child_idx, M::matches),
+        }
+    }
+
+    /// Returns a reference to the array.
+    pub fn array(&self) -> &ArrayRef {
+        &self.array
+    }
+
+    /// Returns a reference to the step.
+    pub fn step(&self) -> &ExecutionStep {
+        &self.step
+    }
+
+    /// Decompose into parts.
+    pub fn into_parts(self) -> (ArrayRef, ExecutionStep) {
+        (self.array, self.step)
+    }
+}
+
+impl fmt::Debug for ExecutionResult {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("ExecutionResult")
+            .field("array", &self.array)
+            .field("step", &self.step)
+            .finish()
+    }
 }
 
 /// Extension trait for creating an execution context from a session.
diff --git a/vortex-array/src/vtable/dyn_.rs b/vortex-array/src/vtable/dyn_.rs
index 64ca5e99c50..09e831de7d7 100644
--- a/vortex-array/src/vtable/dyn_.rs
+++ b/vortex-array/src/vtable/dyn_.rs
@@ -6,6 +6,7 @@ use std::fmt;
 use std::fmt::Debug;
 use std::fmt::Formatter;
 use std::marker::PhantomData;
+use std::sync::Arc;
 
 use arcref::ArcRef;
 use vortex_error::VortexExpect;
@@ -16,6 +17,7 @@ use vortex_session::VortexSession;
 use crate::ArrayAdapter;
 use crate::ArrayRef;
 use crate::DynArray;
+use crate::ExecutionResult;
 use crate::ExecutionStep;
 use crate::IntoArray;
 use crate::buffer::BufferHandle;
@@ -61,7 +63,7 @@ pub trait DynVTable: 'static + private::Sealed + Send + Sync + Debug {
     ) -> VortexResult<Option<ArrayRef>>;
 
     /// See [`VTable::execute`]
-    fn execute(&self, array: &ArrayRef, ctx: &mut ExecutionCtx) -> VortexResult<ExecutionStep>;
+    fn execute(&self, array: ArrayRef, ctx: &mut ExecutionCtx) -> VortexResult<ExecutionResult>;
 
     /// See [`VTable::execute_parent`]
     fn execute_parent(
@@ -146,31 +148,34 @@ impl<V: VTable> DynVTable for ArrayVTableAdapter<V> {
         Ok(Some(reduced))
     }
 
-    fn execute(&self, array: &ArrayRef, ctx: &mut ExecutionCtx) -> VortexResult<ExecutionStep> {
-        let step = V::execute(downcast::<V>(array), ctx)?;
+    fn execute(&self, array: ArrayRef, ctx: &mut ExecutionCtx) -> VortexResult<ExecutionResult> {
+        // Capture metadata before the move for post-validation and stats inheritance.
+        let len = array.len();
+        let dtype = array.dtype().clone();
+        let stats = array.statistics().to_owned();
 
-        if let ExecutionStep::Done(ref result) = step {
+        let owned = downcast_owned::<V>(array);
+        let result = V::execute(owned, ctx)?;
+
+        if matches!(result.step(), ExecutionStep::Done) {
             if cfg!(debug_assertions) {
                 vortex_ensure!(
-                    result.as_ref().len() == array.len(),
+                    result.array().len() == len,
                     "Result length mismatch for {:?}",
                     self
                 );
                 vortex_ensure!(
-                    result.as_ref().dtype() == array.dtype(),
+                    result.array().dtype() == &dtype,
                     "Executed canonical dtype mismatch for {:?}",
                     self
                 );
             }
 
             // TODO(ngates): do we want to do this on every execution? We used to in to_canonical.
-            result
-                .as_ref()
-                .statistics()
-                .inherit_from(array.statistics());
+            result.array().statistics().set_iter(stats.into_iter());
         }
 
-        Ok(step)
+        Ok(result)
     }
 
     fn execute_parent(
@@ -207,6 +212,33 @@ fn downcast<V: VTable>(array: &ArrayRef) -> &V::Array {
         .as_inner()
 }
 
+/// Downcast an `ArrayRef` into an `Arc<V::Array>` without cloning.
+///
+/// This is a zero-cost pointer cast leveraging the `#[repr(transparent)]` layout of
+/// [`ArrayAdapter`].
+fn downcast_owned<V: VTable>(array: ArrayRef) -> Arc<V::Array> {
+    let adapter: Arc<ArrayAdapter<V>> = array
+        .as_any_arc()
+        .downcast::<ArrayAdapter<V>>()
+        .ok()
+        .vortex_expect("Failed to downcast array to expected encoding type");
+    // SAFETY: ArrayAdapter<V> is #[repr(transparent)] over V::Array,
+    // so Arc<ArrayAdapter<V>> and Arc<V::Array> have identical layout.
+    let raw = Arc::into_raw(adapter) as *const V::Array;
+    unsafe { Arc::from_raw(raw) }
+}
+
+/// Upcast an `Arc<V::Array>` into an `ArrayRef` without cloning.
+///
+/// This is a zero-cost pointer cast leveraging the `#[repr(transparent)]` layout of
+/// [`ArrayAdapter`]. It is the reverse of `downcast_owned`.
+pub fn upcast_array<V: VTable>(array: Arc<V::Array>) -> ArrayRef {
+    // SAFETY: ArrayAdapter<V> is #[repr(transparent)] over V::Array,
+    // so Arc<V::Array> and Arc<ArrayAdapter<V>> have identical layout.
+    let raw = Arc::into_raw(array) as *const ArrayAdapter<V>;
+    unsafe { Arc::from_raw(raw) }
+}
+
 impl<V: VTable> Debug for ArrayVTableAdapter<V> {
     fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
         write!(f, "Encoding<{}>", type_name::<V>())
diff --git a/vortex-array/src/vtable/mod.rs b/vortex-array/src/vtable/mod.rs
index a038a83639a..945106fc361 100644
--- a/vortex-array/src/vtable/mod.rs
+++ b/vortex-array/src/vtable/mod.rs
@@ -10,6 +10,7 @@ mod validity;
 use std::fmt::Debug;
 use std::hash::Hasher;
 use std::ops::Deref;
+use std::sync::Arc;
 
 pub use dyn_::*;
 pub use operations::*;
@@ -22,7 +23,7 @@ use vortex_session::VortexSession;
 use crate::ArrayRef;
 use crate::Canonical;
 use crate::DynArray;
-use crate::ExecutionStep;
+use crate::ExecutionResult;
 use crate::IntoArray;
 use crate::Precision;
 use crate::arrays::ConstantArray;
@@ -181,13 +182,12 @@ pub trait VTable: 'static + Sized + Send + Sync + Debug {
     /// of children must be expected.
     fn with_children(array: &mut Self::Array, children: Vec<ArrayRef>) -> VortexResult<()>;
 
-    /// Execute this array by returning an [`ExecutionStep`] that tells the scheduler what to
+    /// Execute this array by returning an [`ExecutionResult`] that tells the scheduler what to
     /// do next.
     ///
     /// Instead of recursively executing children, implementations should return
-    /// [`ExecutionStep::ExecuteChild(i)`] to request that the scheduler execute a child first,
-    /// or [`ExecutionStep::Done(result)`] when the
-    /// encoding can produce a result directly.
+    /// [`ExecutionResult::execute_child`] to request that the scheduler execute a child first,
+    /// or [`ExecutionResult::done`] when the encoding can produce a result directly.
     ///
     /// Array execution is designed such that repeated execution of an array will eventually
     /// converge to a canonical representation. Implementations of this function should therefore
@@ -198,7 +198,7 @@ pub trait VTable: 'static + Sized + Send + Sync + Debug {
     ///
     /// Debug builds will panic if the returned array is of the wrong type, wrong length, or
     /// incorrectly contains null values.
-    fn execute(array: &Self::Array, ctx: &mut ExecutionCtx) -> VortexResult<ExecutionStep>;
+    fn execute(array: Arc<Self::Array>, ctx: &mut ExecutionCtx) -> VortexResult<ExecutionResult>;
 
     /// Attempt to execute the parent of this array.
     ///
diff --git a/vortex-python/src/arrays/py/vtable.rs b/vortex-python/src/arrays/py/vtable.rs
index be3b2ad26a3..447d74367cb 100644
--- a/vortex-python/src/arrays/py/vtable.rs
+++ b/vortex-python/src/arrays/py/vtable.rs
@@ -10,7 +10,7 @@ use pyo3::prelude::*;
 use pyo3::types::PyBytes;
 use vortex::array::ArrayRef;
 use vortex::array::ExecutionCtx;
-use vortex::array::ExecutionStep;
+use vortex::array::ExecutionResult;
 use vortex::array::Precision;
 use vortex::array::RawMetadata;
 use vortex::array::SerializeMetadata;
@@ -156,7 +156,7 @@ impl VTable for PythonVTable {
         Ok(())
     }
 
-    fn execute(_array: &Self::Array, _ctx: &mut ExecutionCtx) -> VortexResult<ExecutionStep> {
+    fn execute(_array: Arc<Self::Array>, _ctx: &mut ExecutionCtx) -> VortexResult<ExecutionResult> {
         todo!()
     }
 }