diff --git a/vortex-array/src/arrays/scalar_fn/vtable/mod.rs b/vortex-array/src/arrays/scalar_fn/vtable/mod.rs
index 42059bb7493..87d2b572a1c 100644
--- a/vortex-array/src/arrays/scalar_fn/vtable/mod.rs
+++ b/vortex-array/src/arrays/scalar_fn/vtable/mod.rs
@@ -13,7 +13,6 @@ use std::hash::Hasher;
 use std::marker::PhantomData;
 use std::ops::Deref;
 use std::ops::Range;
-use std::sync::Arc;
 
 use itertools::Itertools;
 use vortex_dtype::DType;
@@ -35,11 +34,12 @@ use crate::executor::ExecutionCtx;
 use crate::expr;
 use crate::expr::Arity;
 use crate::expr::ChildName;
+use crate::expr::ExecutionArgs;
+use crate::expr::ExecutionResult;
 use crate::expr::ExprId;
 use crate::expr::Expression;
 use crate::expr::ScalarFn;
 use crate::expr::VTableExt;
-use crate::expr::lit;
 use crate::matchers::Matcher;
 use crate::serde::ArrayChildren;
 use crate::vtable;
@@ -127,29 +127,16 @@ impl VTable for ScalarFnVTable {
     }
 
     fn execute(array: &Self::Array, ctx: &mut ExecutionCtx) -> VortexResult<Canonical> {
-        let inputs: Arc<[_]> = array
-            .children
-            .iter()
-            .map(|child| {
-                if let Some(scalar) = child.as_constant() {
-                    return Ok(lit(scalar));
-                }
-                Expression::try_new(
-                    ScalarFn::new(
-                        ArrayExpr,
-                        FakeEq(child.clone().execute::<Canonical>(ctx)?.into_array()),
-                    ),
-                    [],
-                )
-            })
-            .collect::<VortexResult<_>>()?;
+        let args = ExecutionArgs {
+            inputs: array.children.clone(),
+            row_count: array.len,
+            ctx,
+        };
 
         array
             .scalar_fn
-            .evaluate(
-                &Expression::try_new(array.scalar_fn.clone(), inputs)?,
-                &array.to_array(),
-            )?
+            .execute(args)?
+            .into_array()
             .execute::<Canonical>(ctx)
     }
 
@@ -340,13 +327,12 @@ impl expr::VTable for ArrayExpr {
         Ok(options.0.dtype().clone())
     }
 
-    fn evaluate(
+    fn execute(
         &self,
         options: &Self::Options,
-        _expr: &Expression,
-        _scope: &ArrayRef,
-    ) -> VortexResult<ArrayRef> {
-        Ok(options.0.clone())
+        args: ExecutionArgs,
+    ) -> VortexResult<ExecutionResult> {
+        crate::Executable::execute(options.0.clone(), args.ctx)
     }
 
     fn validity(
diff --git a/vortex-array/src/arrays/scalar_fn/vtable/operations.rs b/vortex-array/src/arrays/scalar_fn/vtable/operations.rs
index c0d785072d6..7f0a6740b60 100644
--- a/vortex-array/src/arrays/scalar_fn/vtable/operations.rs
+++ b/vortex-array/src/arrays/scalar_fn/vtable/operations.rs
@@ -1,42 +1,49 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright the Vortex contributors
 
-use std::sync::Arc;
-
 use vortex_error::VortexExpect;
 use vortex_scalar::Scalar;
 
 use crate::Array;
+use crate::IntoArray;
+use crate::LEGACY_SESSION;
+use crate::VortexSessionExecute;
+use crate::arrays::ConstantArray;
 use crate::arrays::scalar_fn::array::ScalarFnArray;
 use crate::arrays::scalar_fn::vtable::ScalarFnVTable;
-use crate::expr::Expression;
-use crate::expr::lit;
+use crate::expr::ExecutionArgs;
+use crate::expr::ExecutionResult;
 use crate::vtable::OperationsVTable;
 
 impl OperationsVTable<ScalarFnVTable> for ScalarFnVTable {
     fn scalar_at(array: &ScalarFnArray, index: usize) -> Scalar {
-        // TODO(ngates): we should evaluate the scalar function over the scalar inputs.
-        let inputs: Arc<[_]> = array
+        let inputs: Vec<_> = array
             .children
             .iter()
-            .map(|child| lit(child.scalar_at(index)))
+            .map(|child| ConstantArray::new(child.scalar_at(index), 1).into_array())
             .collect::<_>();
 
+        let mut ctx = LEGACY_SESSION.create_execution_ctx();
+        let args = ExecutionArgs {
+            inputs,
+            row_count: 1,
+            ctx: &mut ctx,
+        };
+
         let result = array
             .scalar_fn
-            .evaluate(
-                &Expression::try_new(array.scalar_fn.clone(), inputs)
-                    .vortex_expect("create expr must not fail"),
-                &array.to_array(),
-            )
-            .vortex_expect("execute cannot fail");
+            .execute(args)
+            .vortex_expect("todo vortex result return");
 
-        result.as_constant().unwrap_or_else(|| {
-            tracing::info!(
-                "Scalar function {} returned non-constant array from execution over all scalar inputs",
-                array.scalar_fn,
-            );
-            result.scalar_at(0)
-        })
+        match result {
+            ExecutionResult::Array(arr) => {
+                tracing::info!(
+                    "Scalar function {} returned non-constant array from execution over all scalar inputs",
+                    array.scalar_fn,
+                );
+                arr.as_ref().scalar_at(0)
+            }
+            ExecutionResult::Scalar(scalar) => scalar.scalar().clone(),
+        }
     }
 }
diff --git a/vortex-array/src/arrays/scalar_fn/vtable/validity.rs b/vortex-array/src/arrays/scalar_fn/vtable/validity.rs
index 824a0a1e055..4896575ee4d 100644
--- a/vortex-array/src/arrays/scalar_fn/vtable/validity.rs
+++ b/vortex-array/src/arrays/scalar_fn/vtable/validity.rs
@@ -7,21 +7,57 @@ use vortex_error::VortexExpect;
 use vortex_error::VortexResult;
 use vortex_mask::Mask;
 
+use crate::ArrayRef;
 use crate::IntoArray;
 use crate::LEGACY_SESSION;
 use crate::VortexSessionExecute;
-use crate::arrays::NullArray;
 use crate::arrays::scalar_fn::array::ScalarFnArray;
 use crate::arrays::scalar_fn::vtable::ArrayExpr;
 use crate::arrays::scalar_fn::vtable::FakeEq;
 use crate::arrays::scalar_fn::vtable::ScalarFnVTable;
 use crate::executor::CanonicalOutput;
+use crate::expr::ExecutionArgs;
 use crate::expr::Expression;
+use crate::expr::Literal;
+use crate::expr::Root;
 use crate::expr::ScalarFn;
 use crate::expr::lit;
 use crate::validity::Validity;
 use crate::vtable::ValidityVTable;
 
+/// Execute an expression tree recursively.
+///
+/// This assumes all leaf expressions are either ArrayExpr (wrapping actual arrays) or Literals.
+fn execute_expr(expr: &Expression, row_count: usize) -> VortexResult<ArrayRef> {
+    let mut ctx = LEGACY_SESSION.create_execution_ctx();
+
+    // Handle Root expression - this should not happen in validity expressions
+    if expr.is::<Root>() {
+        vortex_error::vortex_bail!("Root expression cannot be executed in validity context");
+    }
+
+    // Handle Literal expression - create a constant array
+    if expr.is::<Literal>() {
+        let scalar = expr.as_::<Literal>();
+        return Ok(crate::arrays::ConstantArray::new(scalar.clone(), row_count).into_array());
+    }
+
+    // Recursively execute child expressions to get input arrays
+    let inputs: Vec<ArrayRef> = expr
+        .children()
+        .iter()
+        .map(|child| execute_expr(child, row_count))
+        .collect::<VortexResult<_>>()?;
+
+    let args = ExecutionArgs {
+        inputs,
+        row_count,
+        ctx: &mut ctx,
+    };
+
+    Ok(expr.scalar_fn().execute(args)?.into_array())
+}
+
 impl ValidityVTable<ScalarFnVTable> for ScalarFnVTable {
     fn validity(array: &ScalarFnArray) -> VortexResult<Validity> {
         let inputs: Arc<[_]> = array
@@ -38,11 +74,8 @@ impl ValidityVTable<ScalarFnVTable> for ScalarFnVTable {
         let expr = Expression::try_new(array.scalar_fn.clone(), inputs)?;
         let validity_expr = array.scalar_fn().validity(&expr)?;
 
-        // We can evaluate the validity expression against an empty scope because we know all
-        // leaves are ArrayExpr.
-        Ok(Validity::Array(
-            validity_expr.evaluate(&NullArray::new(array.len()).into_array())?,
-        ))
+        // Execute the validity expression. All leaves are ArrayExpr nodes.
+        Ok(Validity::Array(execute_expr(&validity_expr, array.len())?))
     }
 
     fn validity_mask(array: &ScalarFnArray) -> Mask {
diff --git a/vortex-array/src/canonical.rs b/vortex-array/src/canonical.rs
index c599df0a981..31be776b397 100644
--- a/vortex-array/src/canonical.rs
+++ b/vortex-array/src/canonical.rs
@@ -117,6 +117,20 @@ impl Canonical {
         }
     }
 
+    pub fn dtype(&self) -> &DType {
+        match self {
+            Canonical::Null(c) => c.dtype(),
+            Canonical::Bool(c) => c.dtype(),
+            Canonical::Primitive(c) => c.dtype(),
+            Canonical::Decimal(c) => c.dtype(),
+            Canonical::VarBinView(c) => c.dtype(),
+            Canonical::List(c) => c.dtype(),
+            Canonical::FixedSizeList(c) => c.dtype(),
+            Canonical::Struct(c) => c.dtype(),
+            Canonical::Extension(c) => c.dtype(),
+        }
+    }
+
     pub fn is_empty(&self) -> bool {
         match self {
             Canonical::Null(c) => c.is_empty(),
diff --git a/vortex-array/src/compute/list_contains.rs b/vortex-array/src/compute/list_contains.rs
index 005d7fe7036..4189ff5b6e3 100644
--- a/vortex-array/src/compute/list_contains.rs
+++ b/vortex-array/src/compute/list_contains.rs
@@ -102,6 +102,8 @@ pub(crate) fn warm_up_vtable() -> usize {
 ///
 /// assert_eq!(matches.to_bool().bit_buffer(), &bitbuffer![false, true, false]);
 /// ```
+// TODO(joe): ensure that list_contains_scalar from (548303761b4270b583ef34f6ca6e3c2b134a242a)
+// is implemented here.
 pub fn list_contains(array: &dyn Array, value: &dyn Array) -> VortexResult<ArrayRef> {
     LIST_CONTAINS_FN
         .invoke(&InvocationArgs {
diff --git a/vortex-array/src/expr/expression.rs b/vortex-array/src/expr/expression.rs
index 3f6f565c3f3..4525b80553e 100644
--- a/vortex-array/src/expr/expression.rs
+++ b/vortex-array/src/expr/expression.rs
@@ -105,11 +105,41 @@ impl Expression {
     }
 
     /// Evaluates the expression in the given scope, returning an array.
+    ///
+    /// This is a convenience method that recursively evaluates child expressions
+    /// and calls the scalar function's execute method.
     pub fn evaluate(&self, scope: &ArrayRef) -> VortexResult<ArrayRef> {
+        use crate::IntoArray;
+        use crate::LEGACY_SESSION;
+        use crate::VortexSessionExecute;
+        use crate::arrays::ConstantArray;
+        use crate::expr::ExecutionArgs;
+        use crate::expr::Literal;
+
         if self.is::<Root>() {
             return Ok(scope.clone());
         }
-        self.scalar_fn.evaluate(self, scope)
+
+        if self.is::<Literal>() {
+            let scalar = self.as_::<Literal>();
+            return Ok(ConstantArray::new(scalar.clone(), scope.len()).into_array());
+        }
+
+        // Recursively evaluate child expressions to get input arrays
+        let inputs: Vec<ArrayRef> = self
+            .children
+            .iter()
+            .map(|child| child.evaluate(scope))
+            .try_collect()?;
+
+        let mut ctx = LEGACY_SESSION.create_execution_ctx();
+        let args = ExecutionArgs {
+            inputs,
+            row_count: scope.len(),
+            ctx: &mut ctx,
+        };
+
+        Ok(self.scalar_fn.execute(args)?.into_array())
     }
 
     /// Returns a new expression representing the validity mask output of this expression.
diff --git a/vortex-array/src/expr/exprs/between.rs b/vortex-array/src/expr/exprs/between.rs
index 67d345755ed..84d2858cae1 100644
--- a/vortex-array/src/expr/exprs/between.rs
+++ b/vortex-array/src/expr/exprs/between.rs
@@ -11,14 +11,16 @@ use vortex_error::VortexResult;
 use vortex_error::vortex_bail;
 use vortex_error::vortex_err;
 use vortex_proto::expr as pb;
-use vortex_vector::Datum;
 
 use crate::ArrayRef;
+use crate::Canonical;
+use crate::arrays::ConstantVTable;
 use crate::compute::BetweenOptions;
 use crate::compute::between as between_compute;
 use crate::expr::Arity;
 use crate::expr::ChildName;
 use crate::expr::ExecutionArgs;
+use crate::expr::ExecutionResult;
 use crate::expr::ExprId;
 use crate::expr::StatsCatalog;
 use crate::expr::VTable;
@@ -138,50 +140,20 @@ impl VTable for Between {
         ))
     }
 
-    fn evaluate(
+    fn execute(
         &self,
         options: &Self::Options,
-        expr: &Expression,
-        scope: &ArrayRef,
-    ) -> VortexResult<ArrayRef> {
-        let arr = expr.child(0).evaluate(scope)?;
-        let lower = expr.child(1).evaluate(scope)?;
-        let upper = expr.child(2).evaluate(scope)?;
-        between_compute(&arr, &lower, &upper, options)
-    }
-
-    fn execute(&self, options: &Self::Options, args: ExecutionArgs) -> VortexResult<Datum> {
-        let [arr, lower, upper]: [Datum; _] = args
-            .datums
+        args: ExecutionArgs,
+    ) -> VortexResult<ExecutionResult> {
+        let [arr, lower, upper]: [ArrayRef; _] = args
+            .inputs
             .try_into()
             .map_err(|_| vortex_err!("Expected 3 arguments for Between expression",))?;
-        let [arr_dt, lower_dt, upper_dt]: [DType; _] = args
-            .dtypes
-            .try_into()
-            .map_err(|_| vortex_err!("Expected 3 dtypes for Between expression",))?;
-
-        let lower_bound = Binary
-            .bind(options.lower_strict.to_operator().into())
-            .execute(ExecutionArgs {
-                datums: vec![lower, arr.clone()],
-                dtypes: vec![lower_dt, arr_dt.clone()],
-                row_count: args.row_count,
-                return_dtype: args.return_dtype.clone(),
-            })?;
-        let upper_bound = Binary
-            .bind(options.upper_strict.to_operator().into())
-            .execute(ExecutionArgs {
-                datums: vec![arr, upper],
-                dtypes: vec![arr_dt, upper_dt],
-                row_count: args.row_count,
-                return_dtype: args.return_dtype.clone(),
-            })?;
 
-        Binary.bind(Operator::And).execute(ExecutionArgs {
-            datums: vec![lower_bound, upper_bound],
-            dtypes: vec![args.return_dtype.clone(), args.return_dtype.clone()],
-            row_count: args.row_count,
-            return_dtype: args.return_dtype,
+        let result = between_compute(arr.as_ref(), lower.as_ref(), upper.as_ref(), options)?;
+        Ok(match result.try_into::<ConstantVTable>() {
+            Ok(constant) => ExecutionResult::Scalar(constant),
+            Err(arr) => ExecutionResult::Array(arr.execute::<Canonical>(args.ctx)?),
         })
     }
 
diff --git a/vortex-array/src/expr/exprs/binary.rs b/vortex-array/src/expr/exprs/binary.rs
index 95e408321c0..7f553cb69a1 100644
--- a/vortex-array/src/expr/exprs/binary.rs
+++ b/vortex-array/src/expr/exprs/binary.rs
@@ -3,27 +3,19 @@
 
 use std::fmt::Formatter;
 
-use arrow_ord::cmp;
 use prost::Message;
-use vortex_compute::arrow::IntoArrow;
-use vortex_compute::arrow::IntoVector;
-use vortex_compute::logical::LogicalAndKleene;
-use vortex_compute::logical::LogicalOrKleene;
 use vortex_dtype::DType;
 use vortex_error::VortexExpect;
 use vortex_error::VortexResult;
 use vortex_error::vortex_bail;
 use vortex_error::vortex_err;
 use vortex_proto::expr as pb;
-use vortex_vector::Datum;
-use vortex_vector::VectorOps;
 
 use crate::ArrayRef;
 use crate::compute;
 use crate::compute::add;
 use crate::compute::and_kleene;
 use crate::compute::compare;
-use crate::compute::compare_nested_arrow_arrays;
 use crate::compute::div;
 use crate::compute::mul;
 use crate::compute::or_kleene;
@@ -31,6 +23,7 @@ use crate::compute::sub;
 use crate::expr::Arity;
 use crate::expr::ChildName;
 use crate::expr::ExecutionArgs;
+use crate::expr::ExecutionResult;
 use crate::expr::ExprId;
 use crate::expr::StatsCatalog;
 use crate::expr::VTable;
@@ -106,16 +99,13 @@ impl VTable for Binary {
         Ok(DType::Bool((lhs.is_nullable() || rhs.is_nullable()).into()))
     }
 
-    fn evaluate(
-        &self,
-        operator: &Operator,
-        expr: &Expression,
-        scope: &ArrayRef,
-    ) -> VortexResult<ArrayRef> {
-        let lhs = expr.child(0).evaluate(scope)?;
-        let rhs = expr.child(1).evaluate(scope)?;
+    fn execute(&self, op: &Operator, args: ExecutionArgs) -> VortexResult<ExecutionResult> {
+        let [lhs, rhs]: [ArrayRef; _] = args
+            .inputs
+            .try_into()
+            .map_err(|_| vortex_err!("Wrong arg count"))?;
 
-        match operator {
+        match op {
             Operator::Eq => compare(&lhs, &rhs, compute::Operator::Eq),
             Operator::NotEq => compare(&lhs, &rhs, compute::Operator::NotEq),
             Operator::Lt => compare(&lhs, &rhs, compute::Operator::Lt),
@@ -128,89 +118,8 @@ impl VTable for Binary {
             Operator::Sub => sub(&lhs, &rhs),
             Operator::Mul => mul(&lhs, &rhs),
             Operator::Div => div(&lhs, &rhs),
-        }
-    }
-
-    fn execute(&self, op: &Operator, args: ExecutionArgs) -> VortexResult<Datum> {
-        let [lhs, rhs]: [Datum; _] = args
-            .datums
-            .try_into()
-            .map_err(|_| vortex_err!("Wrong arg count"))?;
-
-        // Handle logical operators.
-        match op {
-            Operator::And => {
-                return Ok(LogicalAndKleene::and_kleene(&lhs.into_bool(), &rhs.into_bool()).into());
-            }
-            Operator::Or => {
-                return Ok(LogicalOrKleene::or_kleene(&lhs.into_bool(), &rhs.into_bool()).into());
-            }
-            _ => {}
-        }
-
-        // Arrow's vectorized comparison kernels (`cmp::eq`, etc.) don't support nested types
-        // (Struct, List, FixedSizeList). For those, we use `compare_nested_arrow_arrays` which does
-        // element-wise comparison via `make_comparator`.
-        if let Some(cmp_op) = op.maybe_cmp_operator()
-            && (lhs.is_nested() || rhs.is_nested())
-        {
-            // Treat scalars as 1-element arrow arrays.
-            let lhs_arr = lhs.into_arrow()?;
-            let rhs_arr = rhs.into_arrow()?;
-
-            let bool_array = compare_nested_arrow_arrays(lhs_arr.get().0, rhs_arr.get().0, cmp_op)?;
-            let vector = bool_array.into_vector()?;
-
-            let both_are_scalar = lhs_arr.get().1 && rhs_arr.get().1;
-
-            return Ok(if both_are_scalar {
-                Datum::Scalar(vortex_vector::Scalar::Bool(vector.scalar_at(0)))
-            } else {
-                Datum::Vector(vortex_vector::Vector::Bool(vector))
-            });
-        }
-
-        let lhs = lhs.into_arrow()?;
-        let rhs = rhs.into_arrow()?;
-
-        let vector = match op {
-            // Handle comparison operators.
-            Operator::Eq => cmp::eq(lhs.as_ref(), rhs.as_ref())?.into_vector()?.into(),
-            Operator::NotEq => cmp::neq(lhs.as_ref(), rhs.as_ref())?.into_vector()?.into(),
-            Operator::Gt => cmp::gt(lhs.as_ref(), rhs.as_ref())?.into_vector()?.into(),
-            Operator::Gte => cmp::gt_eq(lhs.as_ref(), rhs.as_ref())?
-                .into_vector()?
-                .into(),
-            Operator::Lt => cmp::lt(lhs.as_ref(), rhs.as_ref())?.into_vector()?.into(),
-            Operator::Lte => cmp::lt_eq(lhs.as_ref(), rhs.as_ref())?
-                .into_vector()?
-                .into(),
-
-            // Handle arithmetic operators.
-            Operator::Add => {
-                arrow_arith::numeric::add(lhs.as_ref(), rhs.as_ref())?.into_vector()?
-            }
-            Operator::Sub => {
-                arrow_arith::numeric::sub(lhs.as_ref(), rhs.as_ref())?.into_vector()?
-            }
-            Operator::Mul => {
-                arrow_arith::numeric::mul(lhs.as_ref(), rhs.as_ref())?.into_vector()?
-            }
-            Operator::Div => {
-                arrow_arith::numeric::div(lhs.as_ref(), rhs.as_ref())?.into_vector()?
-            }
-
-            // Logical operators were handled above.
-            Operator::And | Operator::Or => unreachable!("Already dealt with above"),
-        };
-
-        let both_are_scalar = lhs.get().1 && rhs.get().1;
-
-        Ok(if both_are_scalar {
-            Datum::Scalar(vector.scalar_at(0))
-        } else {
-            Datum::Vector(vector)
-        })
+        }?
+        .execute(args.ctx)
     }
 
     fn stat_falsification(
@@ -612,12 +521,8 @@ pub fn checked_add(lhs: Expression, rhs: Expression) -> Expression {
 #[cfg(test)]
 mod tests {
     use vortex_dtype::DType;
-    use vortex_dtype::FieldNames;
     use vortex_dtype::Nullability;
-    use vortex_dtype::PType;
-    use vortex_dtype::StructFields;
     use vortex_scalar::Scalar;
-    use vortex_vector::ScalarOps;
 
     use super::*;
     use crate::expr::Expression;
@@ -711,82 +616,63 @@ mod tests {
     /// using `make_comparator` instead of Arrow's `cmp` functions which don't support nested types.
     #[test]
     fn test_struct_comparison() {
-        // Create a struct dtype for testing.
-        let struct_dtype = DType::Struct(
-            StructFields::new(
-                FieldNames::from(["a", "b"]),
-                vec![
-                    DType::Primitive(PType::I32, Nullability::NonNullable),
-                    DType::Primitive(PType::I32, Nullability::NonNullable),
-                ],
+        use crate::IntoArray;
+        use crate::arrays::StructArray;
+
+        // Create a struct array with one element for testing.
+        let lhs_struct = StructArray::from_fields(&[
+            (
+                "a",
+                crate::arrays::PrimitiveArray::from_iter([1i32]).into_array(),
             ),
-            Nullability::NonNullable,
-        );
+            (
+                "b",
+                crate::arrays::PrimitiveArray::from_iter([3i32]).into_array(),
+            ),
+        ])
+        .unwrap()
+        .into_array();
+
+        let rhs_struct_equal = StructArray::from_fields(&[
+            (
+                "a",
+                crate::arrays::PrimitiveArray::from_iter([1i32]).into_array(),
+            ),
+            (
+                "b",
+                crate::arrays::PrimitiveArray::from_iter([3i32]).into_array(),
+            ),
+        ])
+        .unwrap()
+        .into_array();
+
+        let rhs_struct_different = StructArray::from_fields(&[
+            (
+                "a",
+                crate::arrays::PrimitiveArray::from_iter([1i32]).into_array(),
+            ),
+            (
+                "b",
+                crate::arrays::PrimitiveArray::from_iter([4i32]).into_array(),
+            ),
+        ])
+        .unwrap()
+        .into_array();
 
-        // Test 1: Equal structs should return true.
-        let lhs_scalar = Scalar::struct_(
-            struct_dtype.clone(),
-            vec![Scalar::from(1i32), Scalar::from(3i32)],
-        );
-        let rhs_scalar = Scalar::struct_(
-            struct_dtype.clone(),
-            vec![Scalar::from(1i32), Scalar::from(3i32)],
+        // Test using compare compute function directly
+        let result_equal = compare(&lhs_struct, &rhs_struct_equal, compute::Operator::Eq).unwrap();
+        assert_eq!(
+            result_equal.scalar_at(0),
+            Scalar::bool(true, Nullability::NonNullable),
+            "Equal structs should be equal"
         );
 
-        let lhs_datum = Datum::Scalar(lhs_scalar.to_vector_scalar());
-        let rhs_datum = Datum::Scalar(rhs_scalar.to_vector_scalar());
-
-        let result = Binary.bind(Operator::Eq).execute(ExecutionArgs {
-            datums: vec![lhs_datum, rhs_datum],
-            dtypes: vec![struct_dtype.clone(), struct_dtype.clone()],
-            row_count: 1,
-            return_dtype: DType::Bool(Nullability::NonNullable),
-        });
-
-        assert!(result.is_ok(), "Expected success, but got: {:?}", result);
-        let datum = result.unwrap();
-        if let Datum::Scalar(vortex_vector::Scalar::Bool(bool_scalar)) = datum {
-            assert!(bool_scalar.is_valid());
-            assert_eq!(
-                bool_scalar.value(),
-                Some(true),
-                "Equal structs should be equal"
-            );
-        } else {
-            panic!("Expected Scalar::Bool, got {:?}", datum);
-        }
-
-        // Test 2: Different structs should return false.
-        let lhs_scalar = Scalar::struct_(
-            struct_dtype.clone(),
-            vec![Scalar::from(1i32), Scalar::from(3i32)],
-        );
-        let rhs_scalar = Scalar::struct_(
-            struct_dtype.clone(),
-            vec![Scalar::from(1i32), Scalar::from(4i32)], // Different value.
+        let result_different =
+            compare(&lhs_struct, &rhs_struct_different, compute::Operator::Eq).unwrap();
+        assert_eq!(
+            result_different.scalar_at(0),
+            Scalar::bool(false, Nullability::NonNullable),
+            "Different structs should not be equal"
         );
-
-        let lhs_datum = Datum::Scalar(lhs_scalar.to_vector_scalar());
-        let rhs_datum = Datum::Scalar(rhs_scalar.to_vector_scalar());
-
-        let result = Binary.bind(Operator::Eq).execute(ExecutionArgs {
-            datums: vec![lhs_datum, rhs_datum],
-            dtypes: vec![struct_dtype.clone(), struct_dtype],
-            row_count: 1,
-            return_dtype: DType::Bool(Nullability::NonNullable),
-        });
-
-        assert!(result.is_ok(), "Expected success, but got: {:?}", result);
-        let datum = result.unwrap();
-        if let Datum::Scalar(vortex_vector::Scalar::Bool(bool_scalar)) = datum {
-            assert!(bool_scalar.is_valid());
-            assert_eq!(
-                bool_scalar.value(),
-                Some(false),
-                "Different structs should not be equal"
-            );
-        } else {
-            panic!("Expected Scalar::Bool, got {:?}", datum);
-        }
     }
 }
diff --git a/vortex-array/src/expr/exprs/cast.rs b/vortex-array/src/expr/exprs/cast.rs
index 3c9bce98028..18993ee60d7 100644
--- a/vortex-array/src/expr/exprs/cast.rs
+++ b/vortex-array/src/expr/exprs/cast.rs
@@ -2,7 +2,6 @@
 // SPDX-FileCopyrightText: Copyright the Vortex contributors
 
 use std::fmt::Formatter;
-use std::ops::Deref;
 
 use prost::Message;
 use vortex_dtype::DType;
@@ -10,13 +9,12 @@ use vortex_error::VortexExpect;
 use vortex_error::VortexResult;
 use vortex_error::vortex_err;
 use vortex_proto::expr as pb;
-use vortex_vector::Datum;
 
-use crate::ArrayRef;
 use crate::compute::cast as compute_cast;
 use crate::expr::Arity;
 use crate::expr::ChildName;
 use crate::expr::ExecutionArgs;
+use crate::expr::ExecutionResult;
 use crate::expr::ExprId;
 use crate::expr::ReduceCtx;
 use crate::expr::ReduceNode;
@@ -77,28 +75,16 @@ impl VTable for Cast {
         Ok(dtype.clone())
     }
 
-    fn evaluate(
+    fn execute(
         &self,
-        dtype: &DType,
-        expr: &Expression,
-        scope: &ArrayRef,
-    ) -> VortexResult<ArrayRef> {
-        let array = expr.children()[0].evaluate(scope)?;
-        compute_cast(&array, dtype).map_err(|e| {
-            e.with_context(format!(
-                "Failed to cast array of dtype {} to {}",
-                array.dtype(),
-                expr.deref()
-            ))
-        })
-    }
-
-    fn execute(&self, target_dtype: &DType, mut args: ExecutionArgs) -> VortexResult<Datum> {
+        target_dtype: &DType,
+        mut args: ExecutionArgs,
+    ) -> VortexResult<ExecutionResult> {
         let input = args
-            .datums
+            .inputs
             .pop()
             .vortex_expect("missing input for Cast expression");
-        vortex_compute::cast::Cast::cast(&input, target_dtype)
+        compute_cast(input.as_ref(), target_dtype)?.execute(args.ctx)
     }
 
     fn reduce(
diff --git a/vortex-array/src/expr/exprs/dynamic.rs b/vortex-array/src/expr/exprs/dynamic.rs
index 39fc91cec9a..cccb6cb3740 100644
--- a/vortex-array/src/expr/exprs/dynamic.rs
+++ b/vortex-array/src/expr/exprs/dynamic.rs
@@ -15,20 +15,16 @@ use vortex_error::VortexResult;
 use vortex_error::vortex_bail;
 use vortex_scalar::Scalar;
 use vortex_scalar::ScalarValue;
-use vortex_vector::Datum;
-use vortex_vector::Scalar as VectorScalar;
-use vortex_vector::bool::BoolScalar;
 
-use crate::Array;
 use crate::ArrayRef;
 use crate::IntoArray;
 use crate::arrays::ConstantArray;
 use crate::compute::Operator;
-use crate::compute::compare;
 use crate::expr::Arity;
 use crate::expr::Binary;
 use crate::expr::ChildName;
 use crate::expr::ExecutionArgs;
+use crate::expr::ExecutionResult;
 use crate::expr::ExprId;
 use crate::expr::Expression;
 use crate::expr::StatsCatalog;
@@ -94,50 +90,25 @@ impl VTable for DynamicComparison {
         ))
     }
 
-    fn evaluate(
-        &self,
-        dynamic: &DynamicComparisonExpr,
-        expr: &Expression,
-        scope: &ArrayRef,
-    ) -> VortexResult<ArrayRef> {
-        if let Some(value) = dynamic.rhs.scalar() {
-            let lhs = expr.child(0).evaluate(scope)?;
-            let rhs = ConstantArray::new(value, scope.len());
-            return compare(lhs.as_ref(), rhs.as_ref(), dynamic.operator);
-        }
-
-        // Otherwise, we return the default value.
-        let lhs = expr.return_dtype(scope.dtype())?;
-        Ok(ConstantArray::new(
-            Scalar::new(
-                DType::Bool(lhs.nullability() | dynamic.rhs.dtype.nullability()),
-                dynamic.default.into(),
-            ),
-            scope.len(),
-        )
-        .into_array())
-    }
-
-    fn execute(&self, data: &Self::Options, args: ExecutionArgs) -> VortexResult<Datum> {
+    fn execute(&self, data: &Self::Options, args: ExecutionArgs) -> VortexResult<ExecutionResult> {
         if let Some(scalar) = data.rhs.scalar() {
-            let [lhs]: [Datum; _] = args
-                .datums
+            let [lhs]: [ArrayRef; _] = args
+                .inputs
                 .try_into()
                 .map_err(|_| vortex_error::vortex_err!("Wrong arg count for DynamicComparison"))?;
-            let rhs_vector_scalar = scalar.to_vector_scalar();
-            let rhs = Datum::Scalar(rhs_vector_scalar);
+            let rhs = ConstantArray::new(scalar, args.row_count).into_array();
 
             return Binary.bind(data.operator.into()).execute(ExecutionArgs {
-                datums: vec![lhs, rhs],
-                dtypes: args.dtypes,
+                inputs: vec![lhs, rhs],
                 row_count: args.row_count,
-                return_dtype: args.return_dtype,
+                ctx: args.ctx,
             });
         }
 
-        Ok(Datum::Scalar(VectorScalar::Bool(BoolScalar::new(Some(
-            data.default,
-        )))))
+        Ok(ExecutionResult::Scalar(ConstantArray::new(
+            false,
+            args.row_count,
+        )))
     }
 
     fn stat_falsification(
diff --git a/vortex-array/src/expr/exprs/get_item.rs b/vortex-array/src/expr/exprs/get_item.rs
index 4437004040b..fb00cbf653c 100644
--- a/vortex-array/src/expr/exprs/get_item.rs
+++ b/vortex-array/src/expr/exprs/get_item.rs
@@ -13,18 +13,15 @@ use vortex_error::VortexExpect;
 use vortex_error::VortexResult;
 use vortex_error::vortex_err;
 use vortex_proto::expr as pb;
-use vortex_vector::Datum;
-use vortex_vector::ScalarOps;
-use vortex_vector::VectorOps;
 
-use crate::ArrayRef;
-use crate::ToCanonical;
+use crate::arrays::StructArray;
 use crate::builtins::ExprBuiltins;
 use crate::compute::mask;
 use crate::expr::Arity;
 use crate::expr::ChildName;
 use crate::expr::EmptyOptions;
 use crate::expr::ExecutionArgs;
+use crate::expr::ExecutionResult;
 use crate::expr::ExprId;
 use crate::expr::Expression;
 use crate::expr::Literal;
@@ -104,41 +101,23 @@ impl VTable for GetItem {
         Ok(field_dtype)
     }
 
-    fn evaluate(
+    fn execute(
         &self,
         field_name: &FieldName,
-        expr: &Expression,
-        scope: &ArrayRef,
-    ) -> VortexResult<ArrayRef> {
-        let input = expr.children()[0].evaluate(scope)?.to_struct();
+        mut args: ExecutionArgs,
+    ) -> VortexResult<ExecutionResult> {
+        let input = args
+            .inputs
+            .pop()
+            .vortex_expect("missing input for GetItem expression")
+            .execute::<StructArray>(args.ctx)?;
         let field = input.field_by_name(field_name).cloned()?;
 
         match input.dtype().nullability() {
             Nullability::NonNullable => Ok(field),
             Nullability::Nullable => mask(&field, &input.validity_mask().not()),
-        }
-    }
-
-    fn execute(&self, field_name: &FieldName, mut args: ExecutionArgs) -> VortexResult<Datum> {
-        let struct_dtype = args.dtypes[0]
-            .as_struct_fields_opt()
-            .ok_or_else(|| vortex_err!("Expected struct dtype for child of GetItem expression"))?;
-        let field_idx = struct_dtype
-            .find(field_name)
-            .ok_or_else(|| vortex_err!("Field {} not found in struct dtype", field_name))?;
-
-        match args.datums.pop().vortex_expect("missing input") {
-            Datum::Scalar(s) => {
-                let mut field = s.as_struct().field(field_idx);
-                field.mask_validity(s.is_valid());
-                Ok(Datum::Scalar(field))
-            }
-            Datum::Vector(v) => {
-                let mut field = v.as_struct().fields()[field_idx].clone();
-                field.mask_validity(v.validity());
-                Ok(Datum::Vector(field))
-            }
-        }
+        }?
+        .execute(args.ctx)
     }
 
     fn reduce(
diff --git a/vortex-array/src/expr/exprs/is_null.rs b/vortex-array/src/expr/exprs/is_null.rs
index da02d823c28..3a62892af9c 100644
--- a/vortex-array/src/expr/exprs/is_null.rs
+++ b/vortex-array/src/expr/exprs/is_null.rs
@@ -2,28 +2,18 @@
 // SPDX-FileCopyrightText: Copyright the Vortex contributors
 
 use std::fmt::Formatter;
-use std::ops::Not;
 
 use vortex_dtype::DType;
 use vortex_dtype::Nullability;
 use vortex_error::VortexExpect;
 use vortex_error::VortexResult;
-use vortex_mask::Mask;
-use vortex_vector::Datum;
-use vortex_vector::ScalarOps;
-use vortex_vector::VectorOps;
-use vortex_vector::bool::BoolScalar;
-use vortex_vector::bool::BoolVector;
-
-use crate::Array;
-use crate::ArrayRef;
-use crate::IntoArray;
-use crate::arrays::ConstantArray;
+
 use crate::builtins::ArrayBuiltins;
 use crate::expr::Arity;
 use crate::expr::ChildName;
 use crate::expr::EmptyOptions;
 use crate::expr::ExecutionArgs;
+use crate::expr::ExecutionResult;
 use crate::expr::ExprId;
 use crate::expr::Expression;
 use crate::expr::StatsCatalog;
@@ -78,29 +68,22 @@ impl VTable for IsNull {
         Ok(DType::Bool(Nullability::NonNullable))
     }
 
-    fn evaluate(
+    fn execute(
         &self,
-        _options: &Self::Options,
-        expr: &Expression,
-        scope: &ArrayRef,
-    ) -> VortexResult<ArrayRef> {
-        let array = expr.child(0).evaluate(scope)?;
-        Ok(match array.validity()? {
+        _data: &Self::Options,
+        mut args: ExecutionArgs,
+    ) -> VortexResult<ExecutionResult> {
+        let child = args.inputs.pop().vortex_expect("Missing input child");
+        if let Some(scalar) = child.as_constant() {
+            return Ok(ExecutionResult::constant(scalar.is_null(), args.row_count));
+        }
+
+        Ok(match child.validity()? {
             Validity::NonNullable | Validity::AllValid => {
-                ConstantArray::new(false, array.len()).into_array()
+                ExecutionResult::constant(false, args.row_count)
             }
-            Validity::AllInvalid => ConstantArray::new(true, array.len()).into_array(),
-            Validity::Array(a) => a.not()?,
-        })
-    }
-
-    fn execute(&self, _data: &Self::Options, mut args: ExecutionArgs) -> VortexResult<Datum> {
-        let child = args.datums.pop().vortex_expect("Missing input child");
-        Ok(match child {
-            Datum::Scalar(s) => Datum::Scalar(BoolScalar::new(Some(s.is_null())).into()),
-            Datum::Vector(v) => Datum::Vector(
-                BoolVector::new(v.validity().to_bit_buffer().not(), Mask::new_true(v.len())).into(),
-            ),
+            Validity::AllInvalid => ExecutionResult::constant(true, args.row_count),
+            Validity::Array(a) => a.not()?.execute(args.ctx)?,
         })
     }
 
diff --git a/vortex-array/src/expr/exprs/like.rs b/vortex-array/src/expr/exprs/like.rs
index 43ea22226b3..4c876db00a5 100644
--- a/vortex-array/src/expr/exprs/like.rs
+++ b/vortex-array/src/expr/exprs/like.rs
@@ -4,15 +4,11 @@
 use std::fmt::Formatter;
 
 use prost::Message;
-use vortex_compute::arrow::IntoArrow;
-use vortex_compute::arrow::IntoVector;
 use vortex_dtype::DType;
 use vortex_error::VortexResult;
 use vortex_error::vortex_bail;
 use vortex_error::vortex_err;
 use vortex_proto::expr as pb;
-use vortex_vector::Datum;
-use vortex_vector::VectorOps;
 
 use crate::ArrayRef;
 use crate::compute::LikeOptions;
@@ -20,6 +16,7 @@ use crate::compute::like as like_compute;
 use crate::expr::Arity;
 use crate::expr::ChildName;
 use crate::expr::ExecutionArgs;
+use crate::expr::ExecutionResult;
 use crate::expr::ExprId;
 use crate::expr::Expression;
 use crate::expr::VTable;
@@ -103,40 +100,17 @@ impl VTable for Like {
         ))
     }
 
-    fn evaluate(
+    fn execute(
         &self,
         options: &Self::Options,
-        expr: &Expression,
-        scope: &ArrayRef,
-    ) -> VortexResult<ArrayRef> {
-        let child = expr.child(0).evaluate(scope)?;
-        let pattern = expr.child(1).evaluate(scope)?;
-        like_compute(&child, &pattern, *options)
-    }
-
-    fn execute(&self, options: &Self::Options, args: ExecutionArgs) -> VortexResult<Datum> {
-        let [child, pattern]: [Datum; _] = args
-            .datums
+        args: ExecutionArgs,
+    ) -> VortexResult<ExecutionResult> {
+        let [child, pattern]: [ArrayRef; _] = args
+            .inputs
             .try_into()
             .map_err(|_| vortex_err!("Wrong argument count"))?;
 
-        let child = child.into_arrow()?;
-        let pattern = pattern.into_arrow()?;
-
-        let array = match (options.negated, options.case_insensitive) {
-            (false, false) => arrow_string::like::like(child.as_ref(), pattern.as_ref()),
-            (false, true) => arrow_string::like::ilike(child.as_ref(), pattern.as_ref()),
-            (true, false) => arrow_string::like::nlike(child.as_ref(), pattern.as_ref()),
-            (true, true) => arrow_string::like::nilike(child.as_ref(), pattern.as_ref()),
-        }?;
-
-        let vector = array.into_vector()?;
-        if vector.len() == 1 && args.row_count != 1 {
-            // Arrow returns a scalar datum result
-            return Ok(Datum::Scalar(vector.scalar_at(0).into()));
-        }
-
-        Ok(Datum::Vector(array.into_vector()?.into()))
+        like_compute(&child, &pattern, *options)?.execute(args.ctx)
     }
 
     fn validity(
diff --git a/vortex-array/src/expr/exprs/list_contains.rs b/vortex-array/src/expr/exprs/list_contains.rs
index 50ff7356b7e..fc01309e4c6 100644
--- a/vortex-array/src/expr/exprs/list_contains.rs
+++ b/vortex-array/src/expr/exprs/list_contains.rs
@@ -3,37 +3,19 @@
 
 use std::fmt::Formatter;
 use std::ops::BitOr;
-use std::ops::Deref;
 
-use arrow_buffer::bit_iterator::BitIndexIterator;
-use vortex_buffer::BitBuffer;
-use vortex_compute::logical::LogicalOr;
 use vortex_dtype::DType;
-use vortex_dtype::IntegerPType;
-use vortex_dtype::Nullability;
-use vortex_dtype::PTypeDowncastExt;
-use vortex_dtype::match_each_integer_ptype;
 use vortex_error::VortexResult;
 use vortex_error::vortex_bail;
 use vortex_error::vortex_err;
-use vortex_mask::Mask;
-use vortex_vector::BoolDatum;
-use vortex_vector::Datum;
-use vortex_vector::Vector;
-use vortex_vector::VectorOps;
-use vortex_vector::bool::BoolScalar;
-use vortex_vector::bool::BoolVector;
-use vortex_vector::listview::ListViewScalar;
-use vortex_vector::listview::ListViewVector;
-use vortex_vector::primitive::PVector;
 
 use crate::ArrayRef;
 use crate::compute::list_contains as compute_list_contains;
 use crate::expr::Arity;
-use crate::expr::Binary;
 use crate::expr::ChildName;
 use crate::expr::EmptyOptions;
 use crate::expr::ExecutionArgs;
+use crate::expr::ExecutionResult;
 use crate::expr::ExprId;
 use crate::expr::Expression;
 use crate::expr::StatsCatalog;
@@ -45,7 +27,6 @@ use crate::expr::exprs::binary::lt;
 use crate::expr::exprs::binary::or;
 use crate::expr::exprs::literal::Literal;
 use crate::expr::exprs::literal::lit;
-use crate::expr::operators;
 
 pub struct ListContains;
 
@@ -109,45 +90,17 @@ impl VTable for ListContains {
         Ok(DType::Bool(nullability))
     }
 
-    fn evaluate(
+    fn execute(
         &self,
         _options: &Self::Options,
-        expr: &Expression,
-        scope: &ArrayRef,
-    ) -> VortexResult<ArrayRef> {
-        let list_array = expr.child(0).evaluate(scope)?;
-        let value_array = expr.child(1).evaluate(scope)?;
-        compute_list_contains(list_array.as_ref(), value_array.as_ref())
-    }
-
-    fn execute(&self, _options: &Self::Options, args: ExecutionArgs) -> VortexResult<Datum> {
-        let [lhs, rhs]: [Datum; _] = args
-            .datums
+        args: ExecutionArgs,
+    ) -> VortexResult<ExecutionResult> {
+        let [list_array, value_array]: [ArrayRef; _] = args
+            .inputs
             .try_into()
             .map_err(|_| vortex_err!("Wrong number of arguments for ListContains expression"))?;
 
-        match (lhs, rhs) {
-            (Datum::Scalar(list_scalar), Datum::Scalar(needle_scalar)) => {
-                let list = list_scalar.into_list();
-                let found = list_contains_scalar_scalar(&list, &needle_scalar)?;
-                Ok(Datum::Scalar(BoolScalar::new(Some(found)).into()))
-            }
-            (Datum::Scalar(list_scalar), Datum::Vector(needle_vector)) => {
-                let matches =
-                    constant_list_scalar_contains(list_scalar.into_list(), needle_vector)?;
-                Ok(Datum::Vector(matches.into()))
-            }
-            (Datum::Vector(list_vector), Datum::Scalar(needle_scalar)) => {
-                let matches =
-                    list_contains_scalar(list_vector.into_list(), needle_scalar.into_list())?;
-                Ok(Datum::Vector(matches.into()))
-            }
-            (Datum::Vector(_), Datum::Vector(_)) => {
-                vortex_bail!(
-                    "ListContains currently only supports constant needle (RHS) or constant list (LHS)"
-                )
-            }
-        }
+        compute_list_contains(list_array.as_ref(), value_array.as_ref())?.execute(args.ctx)
     }
 
     fn stat_falsification(
@@ -212,181 +165,6 @@ pub fn list_contains(list: Expression, value: Expression) -> Expression {
     ListContains.new_expr(EmptyOptions, [list, value])
 }
 
-/// Returns a [`BoolVector`] where each bit represents if a list contains the scalar.
-// FIXME(ngates): test implementation and move to vortex-compute
-fn list_contains_scalar(list: ListViewVector, value: ListViewScalar) -> VortexResult<BoolVector> {
-    // If the list array is constant, we perform a single comparison.
-    // if list.len() > 1 && list.is_constant() {
-    //     let contains = list_contains_scalar(&array.slice(0..1), value, nullability)?;
-    //     return Ok(ConstantArray::new(contains.scalar_at(0), array.len()).into_array());
-    // }
-
-    let elems = list.elements();
-    if elems.is_empty() {
-        // Must return false when a list is empty (but valid), or null when the list itself is null.
-        // return crate::compute::list_contains::list_false_or_null(&list_array, nullability);
-        todo!()
-    }
-
-    let matches = Binary
-        .bind(operators::Operator::Eq)
-        .execute(ExecutionArgs {
-            datums: vec![
-                Datum::Vector(elems.deref().clone()),
-                Datum::Scalar(value.into()),
-            ],
-            // FIXME(ngates): dtypes
-            dtypes: vec![],
-            row_count: elems.len(),
-            return_dtype: DType::Bool(Nullability::Nullable),
-        })?
-        .unwrap_into_vector(elems.len())
-        .into_bool()
-        .into_bits();
-
-    // // Fast path: no elements match.
-    // if let Some(pred) = matches.as_constant() {
-    //     return match pred.as_bool().value() {
-    //         // All comparisons are invalid (result in `null`), and search is not null because
-    //         // we already checked for null above.
-    //         None => {
-    //             assert!(
-    //                 !rhs.scalar().is_null(),
-    //                 "Search value must not be null here"
-    //             );
-    //             // False, unless the list itself is null in which case we return null.
-    //             crate::compute::list_contains::list_false_or_null(&list_array, nullability)
-    //         }
-    //         // No elements match, and all comparisons are valid (result in `false`).
-    //         Some(false) => {
-    //             // False, but match the nullability to the input list array.
-    //             Ok(
-    //                 ConstantArray::new(Scalar::bool(false, nullability), list_array.len())
-    //                     .into_array(),
-    //             )
-    //         }
-    //         // All elements match, and all comparisons are valid (result in `true`).
-    //         Some(true) => {
-    //             // True, unless the list itself is empty or NULL.
-    //             crate::compute::list_contains::list_is_not_empty(&list_array, nullability)
-    //         }
-    //     };
-    // }
-
-    // Get the offsets and sizes as primitive arrays.
-    let offsets = list.offsets();
-    let sizes = list.sizes();
-
-    // Process based on the offset and size types.
-    let list_matches = match_each_integer_ptype!(offsets.ptype(), |O| {
-        match_each_integer_ptype!(sizes.ptype(), |S| {
-            process_matches::<O, S>(
-                matches,
-                list.len(),
-                offsets.downcast::<O>(),
-                sizes.downcast::<S>(),
-            )
-        })
-    });
-
-    Ok(BoolVector::new(list_matches, list.validity().clone()))
-}
-
-// Then there is a constant list scalar (haystack) being compared to an array of needles.
-// FIXME(ngates): test implementation and move to vortex-compute
-fn constant_list_scalar_contains(list: ListViewScalar, values: Vector) -> VortexResult<BoolVector> {
-    let elements = list.value().elements();
-
-    // For each element in the list, we perform a full comparison over the values and OR
-    // the results together.
-    let mut result: BoolVector = BoolVector::new(
-        BitBuffer::new_unset(values.len()),
-        Mask::new(values.len(), true),
-    );
-    for i in 0..elements.len() {
-        let element = Datum::Scalar(elements.scalar_at(i));
-        let compared: BoolDatum = Binary
-            .bind(operators::Operator::Eq)
-            .execute(ExecutionArgs {
-                datums: vec![Datum::Vector(values.clone()), element],
-                dtypes: vec![
-                    // FIXME(ngates): call compute function directly!
-                ],
-                row_count: values.len(),
-                return_dtype: DType::Bool(Nullability::Nullable),
-            })?
-            .into_bool();
-        let compared = Datum::from(compared)
-            .unwrap_into_vector(values.len())
-            .into_bool();
-
-        result = LogicalOr::or(&result, &compared);
-    }
-
-    Ok(result)
-}
-
-/// Used when the needle is a scalar checked for containment in a single list.
-fn list_contains_scalar_scalar(
-    list: &ListViewScalar,
-    needle: &vortex_vector::Scalar,
-) -> VortexResult<bool> {
-    let elements = list.value().elements();
-
-    // Note: If the comparison becomes a bottleneck, look into faster ways to check for list
-    // containment. `execute` allocates the returned vector on the heap. Further, the `eq`
-    // comparison does not short-circuit on the first match found.
-    let found = Binary
-        .bind(operators::Operator::Eq)
-        .execute(ExecutionArgs {
-            datums: vec![
-                Datum::Vector(elements.deref().clone()),
-                Datum::Scalar(needle.clone()),
-            ],
-            dtypes: vec![],
-            row_count: elements.len(),
-            return_dtype: DType::Bool(Nullability::Nullable),
-        })?
-        .unwrap_into_vector(elements.len())
-        .into_bool()
-        .into_bits();
-
-    let mut true_bits = BitIndexIterator::new(found.inner().as_ref(), 0, found.len());
-    Ok(true_bits.next().is_some())
-}
-
-/// Returns a [`BitBuffer`] where each bit represents if a list contains the scalar, derived from a
-/// [`BoolArray`] of matches on the child elements array.
-///
-/// TODO(ngates): replace this for aggregation function.
-fn process_matches<O, S>(
-    matches: BitBuffer,
-    list_array_len: usize,
-    offsets: &PVector<O>,
-    sizes: &PVector<S>,
-) -> BitBuffer
-where
-    O: IntegerPType,
-    S: IntegerPType,
-{
-    let offsets_slice = offsets.elements().as_slice();
-    let sizes_slice = sizes.elements().as_slice();
-
-    (0..list_array_len)
-        .map(|i| {
-            // TODO(ngates): does validity render this invalid?
-            let offset = offsets_slice[i].as_();
-            let size = sizes_slice[i].as_();
-
-            // BitIndexIterator yields indices of true bits only. If `.next()` returns
-            // `Some(_)`, at least one element in this list's range matches.
-            let mut set_bits =
-                BitIndexIterator::new(matches.inner().as_slice(), matches.offset() + offset, size);
-            set_bits.next().is_some()
-        })
-        .collect::<BitBuffer>()
-}
-
 #[cfg(test)]
 mod tests {
     use std::sync::Arc;
diff --git a/vortex-array/src/expr/exprs/literal.rs b/vortex-array/src/expr/exprs/literal.rs
index 4ca0eb0f298..67893fda1ec 100644
--- a/vortex-array/src/expr/exprs/literal.rs
+++ b/vortex-array/src/expr/exprs/literal.rs
@@ -10,15 +10,11 @@ use vortex_error::VortexResult;
 use vortex_error::vortex_err;
 use vortex_proto::expr as pb;
 use vortex_scalar::Scalar;
-use vortex_vector::Datum;
 
-use crate::Array;
-use crate::ArrayRef;
-use crate::IntoArray;
-use crate::arrays::ConstantArray;
 use crate::expr::Arity;
 use crate::expr::ChildName;
 use crate::expr::ExecutionArgs;
+use crate::expr::ExecutionResult;
 use crate::expr::ExprId;
 use crate::expr::Expression;
 use crate::expr::StatsCatalog;
@@ -74,18 +70,8 @@ impl VTable for Literal {
         Ok(options.dtype().clone())
     }
 
-    fn evaluate(
-        &self,
-        scalar: &Scalar,
-        _expr: &Expression,
-        scope: &ArrayRef,
-    ) -> VortexResult<ArrayRef> {
-        Ok(ConstantArray::new(scalar.clone(), scope.len()).into_array())
-    }
-
-    fn execute(&self, scalar: &Scalar, _args: ExecutionArgs) -> VortexResult<Datum> {
-        let vector_scalar = scalar.to_vector_scalar();
-        Ok(Datum::Scalar(vector_scalar))
+    fn execute(&self, scalar: &Scalar, args: ExecutionArgs) -> VortexResult<ExecutionResult> {
+        Ok(ExecutionResult::constant(scalar.clone(), args.row_count))
     }
 
     fn stat_expression(
diff --git a/vortex-array/src/expr/exprs/mask.rs b/vortex-array/src/expr/exprs/mask.rs
index f24e2347ae9..a9216770e04 100644
--- a/vortex-array/src/expr/exprs/mask.rs
+++ b/vortex-array/src/expr/exprs/mask.rs
@@ -2,6 +2,7 @@
 // SPDX-FileCopyrightText: Copyright the Vortex contributors
 
 use std::fmt::Formatter;
+use std::ops::Not;
 
 use vortex_dtype::DType;
 use vortex_dtype::Nullability;
@@ -10,18 +11,15 @@ use vortex_error::VortexResult;
 use vortex_error::vortex_ensure;
 use vortex_error::vortex_err;
 use vortex_scalar::Scalar;
-use vortex_vector::BoolDatum;
-use vortex_vector::Datum;
-use vortex_vector::ScalarOps;
-use vortex_vector::VectorMutOps;
-use vortex_vector::VectorOps;
 
 use crate::ArrayRef;
-use crate::ToCanonical;
+use crate::arrays::BoolArray;
+use crate::compute;
 use crate::expr::Arity;
 use crate::expr::ChildName;
 use crate::expr::EmptyOptions;
 use crate::expr::ExecutionArgs;
+use crate::expr::ExecutionResult;
 use crate::expr::ExprId;
 use crate::expr::Expression;
 use crate::expr::Literal;
@@ -86,54 +84,19 @@ impl VTable for Mask {
         Ok(arg_dtypes[0].as_nullable())
     }
 
-    fn evaluate(
+    fn execute(
         &self,
         _options: &Self::Options,
-        expr: &Expression,
-        scope: &ArrayRef,
-    ) -> VortexResult<ArrayRef> {
-        let child = expr.child(0).evaluate(scope)?;
-
-        // The expr::Mask semantics are: mask=true means retain, mask=false means null.
-        // But compute::mask has: mask=true means null, mask=false means retain.
-        // So we need to invert the mask before passing to compute::mask.
-        let mask = expr.child(1).evaluate(scope)?.to_bool().into_bit_buffer();
-
-        crate::compute::mask(&child, &vortex_mask::Mask::from_buffer(!mask))
-    }
-
-    fn execute(&self, _options: &Self::Options, args: ExecutionArgs) -> VortexResult<Datum> {
-        let [input, mask]: [Datum; _] = args
-            .datums
+        args: ExecutionArgs,
+    ) -> VortexResult<ExecutionResult> {
+        let [input, mask_array]: [ArrayRef; _] = args
+            .inputs
             .try_into()
             .map_err(|_| vortex_err!("Wrong arg count"))?;
-        let mask = mask.into_bool();
-
-        match (input, mask) {
-            (Datum::Scalar(input), BoolDatum::Scalar(mask)) => {
-                let mut result = input;
-                result.mask_validity(mask.value().vortex_expect("mask is non-nullable"));
-                Ok(Datum::Scalar(result))
-            }
-            (Datum::Scalar(input), BoolDatum::Vector(mask)) => {
-                let mut result = input.repeat(args.row_count).freeze();
-                result.mask_validity(&vortex_mask::Mask::from(mask.into_bits()));
-                Ok(Datum::Vector(result))
-            }
-            (Datum::Vector(input_array), BoolDatum::Scalar(mask)) => {
-                let mut result = input_array;
-                result.mask_validity(&vortex_mask::Mask::new(
-                    args.row_count,
-                    mask.value().vortex_expect("mask is non-nullable"),
-                ));
-                Ok(Datum::Vector(result))
-            }
-            (Datum::Vector(input_array), BoolDatum::Vector(mask)) => {
-                let mut result = input_array;
-                result.mask_validity(&vortex_mask::Mask::from(mask.into_bits()));
-                Ok(Datum::Vector(result))
-            }
-        }
+
+        let mask_bool = mask_array.execute::<BoolArray>(args.ctx)?;
+        let inverted = mask_bool.bit_buffer().not();
+        compute::mask(&input, &vortex_mask::Mask::from(inverted))?.execute(args.ctx)
     }
 
     fn simplify(
diff --git a/vortex-array/src/expr/exprs/merge.rs b/vortex-array/src/expr/exprs/merge.rs
index a802344b834..33fb390d703 100644
--- a/vortex-array/src/expr/exprs/merge.rs
+++ b/vortex-array/src/expr/exprs/merge.rs
@@ -15,16 +15,13 @@ use vortex_error::VortexExpect;
 use vortex_error::VortexResult;
 use vortex_error::vortex_bail;
 use vortex_utils::aliases::hash_set::HashSet;
-use vortex_vector::Datum;
 
-use crate::Array;
-use crate::ArrayRef;
 use crate::IntoArray as _;
-use crate::ToCanonical;
 use crate::arrays::StructArray;
 use crate::expr::Arity;
 use crate::expr::ChildName;
 use crate::expr::ExecutionArgs;
+use crate::expr::ExecutionResult;
 use crate::expr::ExprId;
 use crate::expr::Expression;
 use crate::expr::GetItem;
@@ -135,36 +132,32 @@ impl VTable for Merge {
         ))
     }
 
-    fn evaluate(
+    fn execute(
         &self,
         options: &Self::Options,
-        expr: &Expression,
-        scope: &ArrayRef,
-    ) -> VortexResult<ArrayRef> {
+        args: ExecutionArgs,
+    ) -> VortexResult<ExecutionResult> {
         // Collect fields in order of appearance. Later fields overwrite earlier fields.
         let mut field_names = Vec::new();
         let mut arrays = Vec::new();
         let mut duplicate_names = HashSet::<_>::new();
 
-        for child in expr.children().iter() {
-            // TODO(marko): When nullable, we need to merge struct validity into field validity.
-            let array = child.evaluate(scope)?;
+        for input in args.inputs {
+            let array = input.execute::<StructArray>(args.ctx)?;
             if array.dtype().is_nullable() {
                 vortex_bail!("merge expects non-nullable input");
             }
-            if !array.dtype().is_struct() {
-                vortex_bail!("merge expects struct input");
-            }
-            let array = array.to_struct();
 
-            for (field_name, array) in array.names().iter().zip_eq(array.fields().iter().cloned()) {
+            for (field_name, field_array) in
+                array.names().iter().zip_eq(array.fields().iter().cloned())
+            {
                 // Update or insert field.
                 if let Some(idx) = field_names.iter().position(|name| name == field_name) {
                     duplicate_names.insert(field_name.clone());
-                    arrays[idx] = array;
+                    arrays[idx] = field_array;
                 } else {
                     field_names.push(field_name.clone());
-                    arrays.push(array);
+                    arrays.push(field_array);
                 }
             }
         }
@@ -178,15 +171,10 @@ impl VTable for Merge {
 
         // TODO(DK): When children are allowed to be nullable, this needs to change.
         let validity = Validity::NonNullable;
-        let len = scope.len();
-        Ok(
-            StructArray::try_new(FieldNames::from(field_names), arrays, len, validity)?
-                .into_array(),
-        )
-    }
-
-    fn execute(&self, _data: &Self::Options, _args: ExecutionArgs) -> VortexResult<Datum> {
-        todo!()
+        let len = args.row_count;
+        StructArray::try_new(FieldNames::from(field_names), arrays, len, validity)?
+            .into_array()
+            .execute(args.ctx)
     }
 
     fn reduce(
diff --git a/vortex-array/src/expr/exprs/not.rs b/vortex-array/src/expr/exprs/not.rs
index 5185af58c97..b67b5d766bc 100644
--- a/vortex-array/src/expr/exprs/not.rs
+++ b/vortex-array/src/expr/exprs/not.rs
@@ -3,19 +3,17 @@
 
 use std::fmt::Formatter;
 
-use vortex_compute::logical::LogicalNot;
 use vortex_dtype::DType;
 use vortex_error::VortexExpect;
 use vortex_error::VortexResult;
 use vortex_error::vortex_bail;
-use vortex_vector::Datum;
 
-use crate::ArrayRef;
 use crate::compute::invert;
 use crate::expr::Arity;
 use crate::expr::ChildName;
 use crate::expr::EmptyOptions;
 use crate::expr::ExecutionArgs;
+use crate::expr::ExecutionResult;
 use crate::expr::ExprId;
 use crate::expr::Expression;
 use crate::expr::VTable;
@@ -72,19 +70,13 @@ impl VTable for Not {
         Ok(child_dtype.clone())
     }
 
-    fn evaluate(
+    fn execute(
         &self,
-        _options: &Self::Options,
-        expr: &Expression,
-        scope: &ArrayRef,
-    ) -> VortexResult<ArrayRef> {
-        let child_result = expr.child(0).evaluate(scope)?;
-        invert(&child_result)
-    }
-
-    fn execute(&self, _data: &Self::Options, mut args: ExecutionArgs) -> VortexResult<Datum> {
-        let child = args.datums.pop().vortex_expect("Missing input child");
-        Ok(child.into_bool().not().into())
+        _data: &Self::Options,
+        mut args: ExecutionArgs,
+    ) -> VortexResult<ExecutionResult> {
+        let child = args.inputs.pop().vortex_expect("Missing input child");
+        invert(&child)?.execute(args.ctx)
     }
 
     fn is_null_sensitive(&self, _options: &Self::Options) -> bool {
diff --git a/vortex-array/src/expr/exprs/pack.rs b/vortex-array/src/expr/exprs/pack.rs
index bca42a2b184..ccd36e8b685 100644
--- a/vortex-array/src/expr/exprs/pack.rs
+++ b/vortex-array/src/expr/exprs/pack.rs
@@ -4,7 +4,6 @@
 use std::fmt::Display;
 use std::fmt::Formatter;
 use std::hash::Hash;
-use std::sync::Arc;
 
 use itertools::Itertools as _;
 use prost::Message;
@@ -13,22 +12,15 @@ use vortex_dtype::FieldName;
 use vortex_dtype::FieldNames;
 use vortex_dtype::Nullability;
 use vortex_dtype::StructFields;
-use vortex_error::VortexExpect;
 use vortex_error::VortexResult;
-use vortex_mask::Mask;
 use vortex_proto::expr as pb;
-use vortex_vector::Datum;
-use vortex_vector::ScalarOps;
-use vortex_vector::VectorMutOps;
-use vortex_vector::VectorOps;
-use vortex_vector::struct_::StructVector;
 
-use crate::ArrayRef;
 use crate::IntoArray;
 use crate::arrays::StructArray;
 use crate::expr::Arity;
 use crate::expr::ChildName;
 use crate::expr::ExecutionArgs;
+use crate::expr::ExecutionResult;
 use crate::expr::ExprId;
 use crate::expr::Expression;
 use crate::expr::VTable;
@@ -125,30 +117,6 @@ impl VTable for Pack {
         ))
     }
 
-    fn evaluate(
-        &self,
-        options: &Self::Options,
-        expr: &Expression,
-        scope: &ArrayRef,
-    ) -> VortexResult<ArrayRef> {
-        let len = scope.len();
-        let value_arrays = expr
-            .children()
-            .iter()
-            .zip_eq(options.names.iter())
-            .map(|(child_expr, name)| {
-                child_expr
-                    .evaluate(scope)
-                    .map_err(|e| e.with_context(format!("Can't evaluate '{name}'")))
-            })
-            .process_results(|it| it.collect::<Vec<_>>())?;
-        let validity = match options.nullability {
-            Nullability::NonNullable => Validity::NonNullable,
-            Nullability::Nullable => Validity::AllValid,
-        };
-        Ok(StructArray::try_new(options.names.clone(), value_arrays, len, validity)?.into_array())
-    }
-
     fn validity(
         &self,
         _options: &Self::Options,
@@ -157,32 +125,17 @@ impl VTable for Pack {
         Ok(Some(lit(true)))
     }
 
-    fn execute(&self, _options: &Self::Options, args: ExecutionArgs) -> VortexResult<Datum> {
-        // If any datum is a vector, we must convert them all to vectors.
-        if args.datums.iter().any(|d| matches!(d, Datum::Vector(_))) {
-            let fields: Box<[_]> = args
-                .datums
-                .into_iter()
-                .map(|v| v.unwrap_into_vector(args.row_count))
-                .collect();
-            return Ok(Datum::Vector(
-                StructVector::try_new(Arc::new(fields), Mask::new_true(args.row_count))?.into(),
-            ));
-        }
-
-        // Otherwise, we can produce a scalar datum by constructing a length-1 struct vector.
-        let fields: Box<[_]> = args
-            .datums
-            .into_iter()
-            .map(|d| {
-                d.into_scalar()
-                    .vortex_expect("all scalars")
-                    .repeat(1)
-                    .freeze()
-            })
-            .collect();
-        let vector = StructVector::new(Arc::new(fields), Mask::new_true(1));
-        Ok(Datum::Scalar(vector.scalar_at(0).into()))
+    fn execute(
+        &self,
+        options: &Self::Options,
+        args: ExecutionArgs,
+    ) -> VortexResult<ExecutionResult> {
+        let len = args.row_count;
+        let value_arrays = args.inputs;
+        let validity: Validity = options.nullability.into();
+        StructArray::try_new(options.names.clone(), value_arrays, len, validity)?
+            .into_array()
+            .execute(args.ctx)
     }
 
     // This applies a nullability
diff --git a/vortex-array/src/expr/exprs/root.rs b/vortex-array/src/expr/exprs/root.rs
index 2433ada7ea0..9b2136613fb 100644
--- a/vortex-array/src/expr/exprs/root.rs
+++ b/vortex-array/src/expr/exprs/root.rs
@@ -8,13 +8,12 @@ use vortex_dtype::FieldPath;
 use vortex_error::VortexExpect;
 use vortex_error::VortexResult;
 use vortex_error::vortex_bail;
-use vortex_vector::Datum;
 
-use crate::ArrayRef;
 use crate::expr::Arity;
 use crate::expr::ChildName;
 use crate::expr::EmptyOptions;
 use crate::expr::ExecutionArgs;
+use crate::expr::ExecutionResult;
 use crate::expr::ExprId;
 use crate::expr::StatsCatalog;
 use crate::expr::VTable;
@@ -65,16 +64,11 @@ impl VTable for Root {
         vortex_bail!("Root expression does not support return_dtype")
     }
 
-    fn evaluate(
+    fn execute(
         &self,
-        _options: &Self::Options,
-        _expr: &Expression,
-        scope: &ArrayRef,
-    ) -> VortexResult<ArrayRef> {
-        Ok(scope.clone())
-    }
-
-    fn execute(&self, _data: &Self::Options, _args: ExecutionArgs) -> VortexResult<Datum> {
+        _data: &Self::Options,
+        _args: ExecutionArgs,
+    ) -> VortexResult<ExecutionResult> {
         vortex_bail!("Root expression is not executable")
     }
 
diff --git a/vortex-array/src/expr/exprs/select.rs b/vortex-array/src/expr/exprs/select.rs
index 25c5d757577..5cabb52fc7a 100644
--- a/vortex-array/src/expr/exprs/select.rs
+++ b/vortex-array/src/expr/exprs/select.rs
@@ -3,7 +3,6 @@
 
 use std::fmt::Display;
 use std::fmt::Formatter;
-use std::sync::Arc;
 
 use itertools::Itertools;
 use prost::Message;
@@ -16,17 +15,13 @@ use vortex_error::vortex_err;
 use vortex_proto::expr::FieldNames as ProtoFieldNames;
 use vortex_proto::expr::SelectOpts;
 use vortex_proto::expr::select_opts::Opts;
-use vortex_vector::Datum;
-use vortex_vector::StructDatum;
-use vortex_vector::VectorOps;
-use vortex_vector::struct_::StructVector;
 
-use crate::ArrayRef;
 use crate::IntoArray;
-use crate::ToCanonical;
+use crate::arrays::StructArray;
 use crate::expr::Arity;
 use crate::expr::ChildName;
 use crate::expr::ExecutionArgs;
+use crate::expr::ExecutionResult;
 use crate::expr::ExprId;
 use crate::expr::SimplifyCtx;
 use crate::expr::VTable;
@@ -136,71 +131,31 @@ impl VTable for Select {
         Ok(DType::Struct(projected, child_dtype.nullability()))
     }
 
-    fn evaluate(
+    fn execute(
         &self,
         selection: &FieldSelection,
-        expr: &Expression,
-        scope: &ArrayRef,
-    ) -> VortexResult<ArrayRef> {
-        let batch = expr.child(0).evaluate(scope)?.to_struct();
-        Ok(match selection {
-            FieldSelection::Include(f) => batch.project(f.as_ref()),
+        mut args: ExecutionArgs,
+    ) -> VortexResult<ExecutionResult> {
+        let child = args
+            .inputs
+            .pop()
+            .vortex_expect("Missing input child")
+            .execute::<StructArray>(args.ctx)?;
+
+        let result = match selection {
+            FieldSelection::Include(f) => child.project(f.as_ref()),
             FieldSelection::Exclude(names) => {
-                let included_names = batch
+                let included_names = child
                     .names()
                     .iter()
                     .filter(|&f| !names.as_ref().contains(f))
                     .cloned()
                     .collect::<Vec<_>>();
-                batch.project(included_names.as_slice())
+                child.project(included_names.as_slice())
             }
-        }?
-        .into_array())
-    }
-
-    fn execute(&self, selection: &FieldSelection, mut args: ExecutionArgs) -> VortexResult<Datum> {
-        let child_fields = args
-            .dtypes
-            .pop()
-            .vortex_expect("Missing input dtype")
-            .into_struct_fields();
-
-        let field_indices: Vec<usize> = match selection {
-            FieldSelection::Include(f) => f
-                .iter()
-                .map(|name| {
-                    child_fields
-                        .find(name)
-                        .ok_or_else(|| vortex_err!("Field {} not found in struct dtype", name))
-                })
-                .try_collect(),
-            FieldSelection::Exclude(names) => child_fields
-                .names()
-                .iter()
-                .filter(|&f| !names.as_ref().contains(f))
-                .map(|name| {
-                    child_fields
-                        .find(name)
-                        .ok_or_else(|| vortex_err!("Field {} not found in struct dtype", name))
-                })
-                .try_collect(),
         }?;
 
-        let child = args
-            .datums
-            .pop()
-            .vortex_expect("Missing input child")
-            .into_struct();
-
-        Ok(match child {
-            StructDatum::Scalar(s) => StructDatum::Scalar(
-                select_from_struct_vector(s.value(), &field_indices)?.scalar_at(0),
-            ),
-            StructDatum::Vector(v) => {
-                StructDatum::Vector(select_from_struct_vector(&v, &field_indices)?)
-            }
-        }
-        .into())
+        result.into_array().execute(args.ctx)
     }
 
     fn simplify(
@@ -248,17 +203,6 @@ impl VTable for Select {
     }
 }
 
-fn select_from_struct_vector(
-    vec: &StructVector,
-    field_indices: &[usize],
-) -> VortexResult<StructVector> {
-    let new_fields = field_indices
-        .iter()
-        .map(|&idx| vec.fields()[idx].clone())
-        .collect();
-    Ok(unsafe { StructVector::new_unchecked(Arc::new(new_fields), vec.validity().clone()) })
-}
-
 /// Creates an expression that selects (includes) specific fields from an array.
 ///
 /// Projects only the specified fields from the child expression, which must be of DType struct.
diff --git a/vortex-array/src/expr/scalar_fn.rs b/vortex-array/src/expr/scalar_fn.rs
index 575d476ba72..c2c778f4237 100644
--- a/vortex-array/src/expr/scalar_fn.rs
+++ b/vortex-array/src/expr/scalar_fn.rs
@@ -13,11 +13,10 @@ use vortex_dtype::DType;
 use vortex_error::VortexExpect;
 use vortex_error::VortexResult;
 use vortex_utils::debug_with::DebugWith;
-use vortex_vector::Datum;
 
-use crate::ArrayRef;
 use crate::expr::EmptyOptions;
 use crate::expr::ExecutionArgs;
+use crate::expr::ExecutionResult;
 use crate::expr::ExprId;
 use crate::expr::ExprVTable;
 use crate::expr::Expression;
@@ -117,14 +116,6 @@ impl ScalarFn {
             .return_dtype(self.options.deref(), arg_types)
     }
 
-    /// Evaluate the expression, returning an ArrayRef.
-    ///
-    /// NOTE: this function will soon be deprecated as all expressions will evaluate trivially
-    ///  into an ExprArray.
-    pub fn evaluate(&self, expr: &Expression, scope: &ArrayRef) -> VortexResult<ArrayRef> {
-        self.vtable.as_dyn().evaluate(expr, scope)
-    }
-
     /// Transforms the expression into one representing the validity of this expression.
     pub fn validity(&self, expr: &Expression) -> VortexResult<Expression> {
         Ok(self.vtable.as_dyn().validity(expr)?.unwrap_or_else(|| {
@@ -138,7 +129,7 @@ impl ScalarFn {
     }
 
     /// Execute the expression given the input arguments.
-    pub fn execute(&self, ctx: ExecutionArgs) -> VortexResult<Datum> {
+    pub fn execute(&self, ctx: ExecutionArgs) -> VortexResult<ExecutionResult> {
         self.vtable.as_dyn().execute(self.options.deref(), ctx)
     }
 
diff --git a/vortex-array/src/expr/vtable.rs b/vortex-array/src/expr/vtable.rs
index e15d539944c..fd5c2fe7f79 100644
--- a/vortex-array/src/expr/vtable.rs
+++ b/vortex-array/src/expr/vtable.rs
@@ -15,11 +15,15 @@ use vortex_dtype::DType;
 use vortex_error::VortexExpect;
 use vortex_error::VortexResult;
 use vortex_error::vortex_bail;
-use vortex_mask::Mask;
-use vortex_vector::Datum;
-use vortex_vector::VectorOps;
+use vortex_scalar::Scalar;
 
 use crate::ArrayRef;
+use crate::Canonical;
+use crate::Executable;
+use crate::ExecutionCtx;
+use crate::IntoArray;
+use crate::arrays::ConstantArray;
+use crate::arrays::ConstantVTable;
 use crate::expr::ExprId;
 use crate::expr::StatsCatalog;
 use crate::expr::expression::Expression;
@@ -78,29 +82,14 @@ pub trait VTable: 'static + Sized + Send + Sync {
     /// Compute the return [`DType`] of the expression if evaluated over the given input types.
     fn return_dtype(&self, options: &Self::Options, arg_dtypes: &[DType]) -> VortexResult<DType>;
 
-    /// Evaluate the expression in the given scope.
-    ///
-    /// This function will be deprecated in a future release in favor of [`VTable::execute`].
-    fn evaluate(
-        &self,
-        options: &Self::Options,
-        expr: &Expression,
-        scope: &ArrayRef,
-    ) -> VortexResult<ArrayRef> {
-        _ = options;
-        _ = expr;
-        _ = scope;
-        vortex_bail!("Expression {} does not support evaluation", self.id());
-    }
-
     /// Execute the expression on the given vector with the given dtype.
     ///
     /// This function will become required in a future release.
-    fn execute(&self, options: &Self::Options, args: ExecutionArgs) -> VortexResult<Datum> {
-        _ = options;
-        drop(args);
-        vortex_bail!("Expression {} does not support execution", self.id());
-    }
+    fn execute(
+        &self,
+        options: &Self::Options,
+        args: ExecutionArgs,
+    ) -> VortexResult<ExecutionResult>;
 
     /// Implement an abstract reduction rule over a tree of scalar functions.
     ///
@@ -305,23 +294,64 @@ pub trait SimplifyCtx {
 }
 
 /// Arguments for expression execution.
-pub struct ExecutionArgs {
-    /// The input datums for the expression, one per child.
-    pub datums: Vec<Datum>,
-    /// The input dtypes for the expression, one per child.
-    pub dtypes: Vec<DType>,
+pub struct ExecutionArgs<'a> {
+    /// The inputs for the expression, one per child.
+    pub inputs: Vec<ArrayRef>,
     /// The row count of the execution scope.
     pub row_count: usize,
-    /// The expected return dtype of the expression, as computed by [`Expression::return_dtype`].
-    pub return_dtype: DType,
+    /// The execution context.
+    pub ctx: &'a mut ExecutionCtx,
 }
 
-/// Arguments for expression validity execution.
-pub struct ValidityExecutionArgs {
-    /// The input masks for the expression, one per child.
-    pub inputs: Vec<Mask>,
-    /// The row count of the execution scope.
-    pub row_count: usize,
+/// The result of expression execution.
+pub enum ExecutionResult {
+    Array(Canonical),
+    Scalar(ConstantArray),
+}
+
+impl ExecutionResult {
+    pub fn constant<S: Into<Scalar>>(scalar: S, len: usize) -> Self {
+        ExecutionResult::Scalar(ConstantArray::new(scalar.into(), len))
+    }
+
+    /// Returns the length of this execution result.
+    pub fn len(&self) -> usize {
+        match self {
+            ExecutionResult::Array(canonical) => canonical.len(),
+            ExecutionResult::Scalar(constant) => constant.len(),
+        }
+    }
+
+    /// Returns true if this execution result has no elements.
+    pub fn is_empty(&self) -> bool {
+        self.len() == 0
+    }
+
+    /// Returns the data type of this execution result.
+    pub fn dtype(&self) -> &DType {
+        match self {
+            ExecutionResult::Array(canonical) => canonical.dtype(),
+            ExecutionResult::Scalar(constant) => constant.dtype(),
+        }
+    }
+}
+
+impl Executable for ExecutionResult {
+    fn execute(array: ArrayRef, ctx: &mut ExecutionCtx) -> VortexResult<Self> {
+        Ok(match array.as_opt::<ConstantVTable>() {
+            None => ExecutionResult::Array(array.execute::<Canonical>(ctx)?),
+            Some(constant) => ExecutionResult::Scalar(constant.clone()),
+        })
+    }
+}
+
+impl IntoArray for ExecutionResult {
+    fn into_array(self) -> ArrayRef {
+        match self {
+            ExecutionResult::Array(canonical) => canonical.into_array(),
+            ExecutionResult::Scalar(constant) => constant.into_array(),
+        }
+    }
 }
 
 #[derive(Clone, Debug, PartialEq, Eq, Hash)]
@@ -391,8 +421,7 @@ pub trait DynExprVTable: 'static + Send + Sync + private::Sealed {
     ) -> VortexResult<Option<Expression>>;
     fn simplify_untyped(&self, expression: &Expression) -> VortexResult<Option<Expression>>;
     fn validity(&self, expression: &Expression) -> VortexResult<Option<Expression>>;
-    fn execute(&self, options: &dyn Any, args: ExecutionArgs) -> VortexResult<Datum>;
-    fn evaluate(&self, expression: &Expression, scope: &ArrayRef) -> VortexResult<ArrayRef>;
+    fn execute(&self, options: &dyn Any, args: ExecutionArgs) -> VortexResult<ExecutionResult>;
     fn reduce(
         &self,
         options: &dyn Any,
@@ -504,52 +533,46 @@ impl<V: VTable> DynExprVTable for VTableAdapter<V> {
         )
     }
 
-    fn execute(&self, options: &dyn Any, args: ExecutionArgs) -> VortexResult<Datum> {
+    fn execute(&self, options: &dyn Any, args: ExecutionArgs) -> VortexResult<ExecutionResult> {
         let options = downcast::<V>(options);
 
         let expected_row_count = args.row_count;
         #[cfg(debug_assertions)]
-        let expected_dtype = args.return_dtype.clone();
+        let expected_dtype = {
+            let args_dtypes: Vec<DType> = args
+                .inputs
+                .iter()
+                .map(|array| array.dtype().clone())
+                .collect();
+            V::return_dtype(&self.0, options, &args_dtypes)
+        }?;
 
         let result = V::execute(&self.0, options, args)?;
 
-        if let Datum::Vector(v) = &result {
-            assert_eq!(
-                v.len(),
-                expected_row_count,
-                "Expression execution {} returned vector of length {}, but expected {}",
-                self.0.id(),
-                v.len(),
-                expected_row_count,
-            );
-        }
+        assert_eq!(
+            result.len(),
+            expected_row_count,
+            "Expression execution {} returned vector of length {}, but expected {}",
+            self.0.id(),
+            result.len(),
+            expected_row_count,
+        );
 
         // In debug mode, validate that the output dtype matches the expected return dtype.
         #[cfg(debug_assertions)]
         {
-            use vortex_vector::datum_matches_dtype;
-
-            if !datum_matches_dtype(&result, &expected_dtype) {
-                vortex_bail!(
-                    "Expression execution returned datum of invalid dtype. Expected {}, got {:?}",
-                    expected_dtype,
-                    result
-                );
-            }
+            vortex_error::vortex_ensure!(
+                result.dtype() == &expected_dtype,
+                "Expression execution {} returned vector of invalid dtype. Expected {}, got {}",
+                self.0.id(),
+                expected_dtype,
+                result.dtype(),
+            );
         }
 
         Ok(result)
     }
 
-    fn evaluate(&self, expression: &Expression, scope: &ArrayRef) -> VortexResult<ArrayRef> {
-        V::evaluate(
-            &self.0,
-            downcast::<V>(expression.options().as_any()),
-            expression,
-            scope,
-        )
-    }
-
     fn reduce(
         &self,
         options: &dyn Any,
diff --git a/vortex-layout/src/layouts/row_idx/expr.rs b/vortex-layout/src/layouts/row_idx/expr.rs
index 0d029ffd9d7..09a0fbcb2e5 100644
--- a/vortex-layout/src/layouts/row_idx/expr.rs
+++ b/vortex-layout/src/layouts/row_idx/expr.rs
@@ -3,11 +3,11 @@
 
 use std::fmt::Formatter;
 
-use vortex_array::ArrayRef;
 use vortex_array::expr::Arity;
 use vortex_array::expr::ChildName;
 use vortex_array::expr::EmptyOptions;
 use vortex_array::expr::ExecutionArgs;
+use vortex_array::expr::ExecutionResult;
 use vortex_array::expr::ExprId;
 use vortex_array::expr::Expression;
 use vortex_array::expr::VTable;
@@ -17,7 +17,6 @@ use vortex_dtype::Nullability;
 use vortex_dtype::PType;
 use vortex_error::VortexResult;
 use vortex_error::vortex_bail;
-use vortex_vector::Datum;
 
 pub struct RowIdx;
 
@@ -49,20 +48,13 @@ impl VTable for RowIdx {
         Ok(DType::Primitive(PType::U64, Nullability::NonNullable))
     }
 
-    fn evaluate(
+    fn execute(
         &self,
         _options: &Self::Options,
-        _expr: &Expression,
-        _scope: &ArrayRef,
-    ) -> VortexResult<ArrayRef> {
-        vortex_bail!(
-            "RowIdxExpr should not be evaluated directly, use it in the context of a Vortex scan and it will be substituted for a row index array"
-        );
-    }
-
-    fn execute(&self, _options: &Self::Options, _args: ExecutionArgs) -> VortexResult<Datum> {
+        _args: ExecutionArgs,
+    ) -> VortexResult<ExecutionResult> {
         vortex_bail!(
-            "RowIdxExpr should not be eecuted directly, use it in the context of a Vortex scan and it will be substituted for a row index array"
+            "RowIdxExpr should not be executed directly, use it in the context of a Vortex scan and it will be substituted for a row index array"
         );
     }
 }