diff --git a/vortex-array/src/arrays/higher_order_fn/array.rs b/vortex-array/src/arrays/higher_order_fn/array.rs
new file mode 100644
index 00000000000..ef47ac9b0d5
--- /dev/null
+++ b/vortex-array/src/arrays/higher_order_fn/array.rs
@@ -0,0 +1,132 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+use std::fmt::Display;
+use std::fmt::Formatter;
+
+use vortex_error::VortexExpect;
+use vortex_error::VortexResult;
+use vortex_error::vortex_ensure;
+
+use crate::ArrayRef;
+use crate::ArraySlots;
+use crate::array::Array;
+use crate::array::ArrayParts;
+use crate::array::TypedArrayRef;
+use crate::arrays::HigherOrderFn;
+use crate::higher_order_fn::HigherOrderFunctionRef;
+use crate::higher_order_fn::LambdaClosure;
+
+/// Per-array data for [`HigherOrderFnArray`].
+#[derive(Clone, Debug)]
+pub struct HigherOrderFnData {
+    pub(super) higher_order_fn: HigherOrderFunctionRef,
+    pub(super) arg_count: usize,
+    pub(super) lambdas: Box<[LambdaClosure]>,
+}
+
+impl Display for HigherOrderFnData {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        write!(f, "higher_order_fn: {}", self.higher_order_fn)
+    }
+}
+
+pub trait HigherOrderFnArrayExt: TypedArrayRef<HigherOrderFn> {
+    /// The function represented by this array.
+    fn higher_order_fn(&self) -> &HigherOrderFunctionRef {
+        &self.higher_order_fn
+    }
+
+    /// The number of ordinary function arguments, before capture slots.
+    fn arg_count(&self) -> usize {
+        self.arg_count
+    }
+
+    /// The lambdas closed over the capture slots.
+    fn lambdas(&self) -> &[LambdaClosure] {
+        &self.lambdas
+    }
+
+    /// The function's ordinary argument arrays.
+    fn args(&self) -> Vec<ArrayRef> {
+        self.as_ref().slots()[..self.arg_count()]
+            .iter()
+            .map(|slot| {
+                slot.as_ref()
+                    .vortex_expect("HigherOrderFnArray argument slot")
+                    .clone()
+            })
+            .collect()
+    }
+
+    /// Arrays captured by its lambdas, in closure-slot order.
+    fn captures(&self) -> Vec<ArrayRef> {
+        self.as_ref().slots()[self.arg_count()..]
+            .iter()
+            .map(|slot| {
+                slot.as_ref()
+                    .vortex_expect("HigherOrderFnArray capture slot")
+                    .clone()
+            })
+            .collect()
+    }
+}
+impl<T: TypedArrayRef<HigherOrderFn>> HigherOrderFnArrayExt for T {}
+
+impl Array<HigherOrderFn> {
+    /// Build a lazy higher-order-function array from ordinary arguments and lexical closures.
+    pub(crate) fn try_new_with_len(
+        higher_order_fn: HigherOrderFunctionRef,
+        args: Vec<ArrayRef>,
+        lambdas: Vec<LambdaClosure>,
+        capture_slots: Vec<ArrayRef>,
+        len: usize,
+    ) -> VortexResult<Self> {
+        vortex_ensure!(
+            args.iter().all(|arg| arg.len() == len),
+            "HigherOrderFnArray arguments must have the array length"
+        );
+        vortex_ensure!(
+            higher_order_fn.arity().matches(args.len()),
+            "{} takes {} ordinary arguments, got {}",
+            higher_order_fn,
+            higher_order_fn.arity(),
+            args.len()
+        );
+        vortex_ensure!(
+            lambdas.len() == higher_order_fn.lambda_arity(),
+            "{} takes {} lambda arguments, got {}",
+            higher_order_fn,
+            higher_order_fn.lambda_arity(),
+            lambdas.len()
+        );
+
+        let arg_dtypes = args
+            .iter()
+            .map(|arg| arg.dtype().clone())
+            .collect::<Vec<_>>();
+        let arg_count = args.len();
+        let lambdas = lambdas.into_boxed_slice();
+        let mut slots = args;
+        slots.extend(capture_slots);
+        let bound_lambdas = lambdas
+            .iter()
+            .map(|lambda| lambda.lambda().clone())
+            .collect::<Vec<_>>();
+        let dtype = higher_order_fn.return_dtype(&arg_dtypes, &bound_lambdas)?;
+        let data = HigherOrderFnData {
+            higher_order_fn: higher_order_fn.clone(),
+            arg_count,
+            lambdas,
+        };
+        let vtable = HigherOrderFn {
+            id: higher_order_fn.id(),
+        };
+        Ok(unsafe {
+            Array::from_parts_unchecked(
+                ArrayParts::new(vtable, dtype, len, data)
+                    .with_slots(slots.into_iter().map(Some).collect::<ArraySlots>()),
+            )
+        })
+    }
+}
diff --git a/vortex-array/src/arrays/higher_order_fn/mod.rs b/vortex-array/src/arrays/higher_order_fn/mod.rs
new file mode 100644
index 00000000000..7884b18548d
--- /dev/null
+++ b/vortex-array/src/arrays/higher_order_fn/mod.rs
@@ -0,0 +1,8 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+mod array;
+mod vtable;
+
+pub use array::HigherOrderFnArrayExt;
+pub use vtable::*;
diff --git a/vortex-array/src/arrays/higher_order_fn/vtable.rs b/vortex-array/src/arrays/higher_order_fn/vtable.rs
new file mode 100644
index 00000000000..12c9e076a96
--- /dev/null
+++ b/vortex-array/src/arrays/higher_order_fn/vtable.rs
@@ -0,0 +1,221 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+use std::fmt::Display;
+use std::fmt::Formatter;
+use std::hash::Hash;
+use std::hash::Hasher;
+
+use vortex_error::VortexResult;
+use vortex_error::vortex_bail;
+use vortex_error::vortex_ensure;
+use vortex_error::vortex_panic;
+use vortex_session::VortexSession;
+
+use crate::ArrayEq;
+use crate::ArrayHash;
+use crate::ArrayRef;
+use crate::Canonical;
+use crate::EqMode;
+use crate::ExecutionCtx;
+use crate::ExecutionResult;
+use crate::IntoArray;
+use crate::array::Array;
+use crate::array::ArrayId;
+use crate::array::ArrayParts;
+use crate::array::ArrayView;
+use crate::array::OperationsVTable;
+use crate::array::VTable;
+use crate::array::ValidityVTable;
+use crate::array::with_empty_buffers;
+use crate::arrays::higher_order_fn::array::HigherOrderFnArrayExt;
+use crate::arrays::higher_order_fn::array::HigherOrderFnData;
+use crate::buffer::BufferHandle;
+use crate::dtype::DType;
+use crate::higher_order_fn::HigherOrderFunctionId;
+use crate::serde::ArrayChildren;
+use crate::validity::Validity;
+
+/// A lazy array produced by a higher-order function.
+pub type HigherOrderFnArray = Array<HigherOrderFn>;
+
+/// The Vortex array vtable for a registered higher-order function.
+#[derive(Clone, Debug)]
+pub struct HigherOrderFn {
+    pub(super) id: HigherOrderFunctionId,
+}
+
+impl Display for HigherOrderFn {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        self.id.fmt(f)
+    }
+}
+
+impl ArrayHash for HigherOrderFnData {
+    fn array_hash<H: Hasher>(&self, state: &mut H, _accuracy: EqMode) {
+        self.higher_order_fn.hash(state);
+        self.arg_count.hash(state);
+        self.lambdas.hash(state);
+    }
+}
+
+impl ArrayEq for HigherOrderFnData {
+    fn array_eq(&self, other: &Self, _accuracy: EqMode) -> bool {
+        self.higher_order_fn == other.higher_order_fn
+            && self.arg_count == other.arg_count
+            && self.lambdas == other.lambdas
+    }
+}
+
+impl VTable for HigherOrderFn {
+    type TypedArrayData = HigherOrderFnData;
+    type OperationsVTable = Self;
+    type ValidityVTable = Self;
+
+    fn id(&self) -> ArrayId {
+        self.id
+    }
+
+    fn validate(
+        &self,
+        data: &HigherOrderFnData,
+        dtype: &DType,
+        len: usize,
+        slots: &[Option<ArrayRef>],
+    ) -> VortexResult<()> {
+        vortex_ensure!(
+            data.higher_order_fn.id() == self.id,
+            "HigherOrderFnArray data function does not match vtable"
+        );
+        vortex_ensure!(
+            slots.len() >= data.arg_count,
+            "HigherOrderFnArray has fewer slots than ordinary arguments"
+        );
+        vortex_ensure!(
+            data.higher_order_fn.arity().matches(data.arg_count),
+            "HigherOrderFnArray argument count does not match function arity"
+        );
+        vortex_ensure!(
+            slots.iter().all(Option::is_some),
+            "HigherOrderFnArray slots must not be empty"
+        );
+        vortex_ensure!(
+            slots.iter().flatten().all(|slot| slot.len() == len),
+            "HigherOrderFnArray slots must have the array length"
+        );
+        vortex_ensure!(
+            data.lambdas.len() == data.higher_order_fn.lambda_arity(),
+            "HigherOrderFnArray lambda count does not match function arity"
+        );
+
+        let capture_count = slots.len() - data.arg_count;
+        for lambda in &data.lambdas {
+            lambda.validate(capture_count)?;
+        }
+
+        let arg_dtypes = slots[..data.arg_count]
+            .iter()
+            .flatten()
+            .map(|slot| slot.dtype().clone())
+            .collect::<Vec<_>>();
+        let lambdas = data
+            .lambdas
+            .iter()
+            .map(|lambda| lambda.lambda().clone())
+            .collect::<Vec<_>>();
+        vortex_ensure!(
+            data.higher_order_fn.return_dtype(&arg_dtypes, &lambdas)? == *dtype,
+            "HigherOrderFnArray dtype does not match higher-order function return dtype"
+        );
+        Ok(())
+    }
+
+    fn nbuffers(_array: ArrayView<'_, Self>) -> usize {
+        0
+    }
+
+    fn buffer(_array: ArrayView<'_, Self>, idx: usize) -> BufferHandle {
+        vortex_panic!("HigherOrderFnArray buffer index {idx} out of bounds")
+    }
+
+    fn buffer_name(_array: ArrayView<'_, Self>, _idx: usize) -> Option<String> {
+        None
+    }
+
+    fn with_buffers(
+        &self,
+        array: ArrayView<'_, Self>,
+        buffers: &[BufferHandle],
+    ) -> VortexResult<ArrayParts<Self>> {
+        with_empty_buffers(self, array, buffers)
+    }
+
+    fn serialize(
+        _array: ArrayView<'_, Self>,
+        _session: &VortexSession,
+    ) -> VortexResult<Option<Vec<u8>>> {
+        // Runtime closures carry array references and are intentionally not serializable.
+        Ok(None)
+    }
+
+    fn deserialize(
+        &self,
+        _dtype: &DType,
+        _len: usize,
+        _metadata: &[u8],
+        _buffers: &[BufferHandle],
+        _children: &dyn ArrayChildren,
+        _session: &VortexSession,
+    ) -> VortexResult<ArrayParts<Self>> {
+        vortex_bail!("Deserialization of HigherOrderFnArray metadata is not supported")
+    }
+
+    fn slot_name(array: ArrayView<'_, Self>, idx: usize) -> String {
+        if idx < array.arg_count() {
+            array.higher_order_fn().child_name(idx).to_string()
+        } else {
+            format!("capture[{}]", idx - array.arg_count())
+        }
+    }
+
+    fn execute(array: Array<Self>, ctx: &mut ExecutionCtx) -> VortexResult<ExecutionResult> {
+        let args = array.args();
+        let captures = array.captures();
+        let lambdas = array
+            .lambdas()
+            .iter()
+            .map(|lambda| lambda.call(&captures))
+            .collect::<Vec<_>>();
+        array
+            .higher_order_fn()
+            .execute(&args, &lambdas, ctx)
+            .map(ExecutionResult::done)
+    }
+}
+
+impl OperationsVTable<HigherOrderFn> for HigherOrderFn {
+    fn scalar_at(
+        array: ArrayView<'_, HigherOrderFn>,
+        index: usize,
+        ctx: &mut ExecutionCtx,
+    ) -> VortexResult<crate::scalar::Scalar> {
+        array
+            .array()
+            .clone()
+            .execute::<Canonical>(ctx)?
+            .into_array()
+            .execute_scalar(index, ctx)
+    }
+}
+
+impl ValidityVTable<HigherOrderFn> for HigherOrderFn {
+    fn validity(array: ArrayView<'_, HigherOrderFn>) -> VortexResult<Validity> {
+        let args = array.args();
+        let lambdas = array
+            .lambdas()
+            .iter()
+            .map(|lambda| lambda.lambda().clone())
+            .collect::<Vec<_>>();
+        array.higher_order_fn().validity(&args, &lambdas)
+    }
+}
diff --git a/vortex-array/src/arrays/mod.rs b/vortex-array/src/arrays/mod.rs
index f96eebead65..7ec3f64e84f 100644
--- a/vortex-array/src/arrays/mod.rs
+++ b/vortex-array/src/arrays/mod.rs
@@ -66,6 +66,10 @@ pub mod fixed_size_list;
 pub use fixed_size_list::FixedSizeList;
 pub use fixed_size_list::FixedSizeListArray;
 
+pub mod higher_order_fn;
+pub use higher_order_fn::HigherOrderFn;
+pub use higher_order_fn::HigherOrderFnArray;
+
 pub mod interleave;
 pub use interleave::Interleave;
 pub use interleave::InterleaveArray;
diff --git a/vortex-array/src/expr/analysis/fallible.rs b/vortex-array/src/expr/analysis/fallible.rs
index ff43d51603b..cde0b209111 100644
--- a/vortex-array/src/expr/analysis/fallible.rs
+++ b/vortex-array/src/expr/analysis/fallible.rs
@@ -10,8 +10,8 @@ pub fn label_is_fallible(expr: &Expression) -> BooleanLabels<'_> {
         expr,
         |expr| match expr {
             Expression::Scalar { scalar_fn, .. } => scalar_fn.signature().is_fallible(),
-            // The scope itself cannot fail.
-            Expression::Root => false,
+            Expression::HigherOrder { .. } => true,
+            Expression::Root | Expression::Variable(_) => false,
         },
         |acc, &child| acc | child,
     )
diff --git a/vortex-array/src/expr/analysis/immediate_access.rs b/vortex-array/src/expr/analysis/immediate_access.rs
index 6c2e4975a92..bd4491af00a 100644
--- a/vortex-array/src/expr/analysis/immediate_access.rs
+++ b/vortex-array/src/expr/analysis/immediate_access.rs
@@ -67,7 +67,11 @@ pub fn make_bound_free_field_annotator(
 ) -> impl AnnotationFn<BoundExpression, Annotation = FieldName> {
     move |expr: &BoundExpression| {
         let Some(scalar_fn) = expr.as_scalar() else {
-            return scope.names().iter().cloned().collect();
+            return if expr.is_root() {
+                scope.names().iter().cloned().collect()
+            } else {
+                vec![]
+            };
         };
 
         if let Some(selection) = scalar_fn.as_opt::<Select>() {
@@ -87,3 +91,37 @@ pub fn make_bound_free_field_annotator(
         vec![]
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use vortex_error::VortexResult;
+
+    use super::*;
+    use crate::dtype::DType;
+    use crate::dtype::Nullability;
+    use crate::dtype::PType;
+    use crate::expr::Scope;
+    use crate::expr::Variable;
+    use crate::expr::test_harness::struct_dtype;
+    use crate::expr::var;
+
+    #[test]
+    fn a_bound_variable_accesses_no_root_fields() -> VortexResult<()> {
+        let scope_dtype = struct_dtype();
+        let fields = scope_dtype
+            .as_struct_fields_opt()
+            .vortex_expect("test scope is a struct");
+
+        let bound = var("x").bind_scope(&Scope::new(scope_dtype.clone()).with_bindings([(
+            Variable::new("x"),
+            DType::Primitive(PType::I32, Nullability::NonNullable),
+        )])?)?;
+
+        let annotator = make_bound_free_field_annotator(fields);
+        assert!(
+            annotator(&bound).is_empty(),
+            "a variable should read no root fields"
+        );
+        Ok(())
+    }
+}
diff --git a/vortex-array/src/expr/analysis/strict.rs b/vortex-array/src/expr/analysis/strict.rs
index 01cc71d91c3..4823cc1e7c4 100644
--- a/vortex-array/src/expr/analysis/strict.rs
+++ b/vortex-array/src/expr/analysis/strict.rs
@@ -14,8 +14,9 @@ pub fn label_strict(expr: &Expression) -> BooleanLabels<'_> {
         expr,
         |expr| match expr {
             Expression::Scalar { scalar_fn, .. } => scalar_fn.signature().is_strict(),
-            // Vacuously strict.
-            Expression::Root => true,
+            Expression::HigherOrder { .. } => false,
+            // Vacuously strict: nullary, so no argument can be null.
+            Expression::Root | Expression::Variable { .. } => true,
         },
         |acc, &child| acc & child,
     )
diff --git a/vortex-array/src/expr/bound_expression.rs b/vortex-array/src/expr/bound_expression.rs
index 45040ab2cd1..3c6d82a0391 100644
--- a/vortex-array/src/expr/bound_expression.rs
+++ b/vortex-array/src/expr/bound_expression.rs
@@ -11,15 +11,20 @@ use std::sync::Arc;
 use itertools::Itertools;
 use vortex_error::VortexExpect;
 use vortex_error::VortexResult;
+use vortex_error::vortex_bail;
 use vortex_error::vortex_ensure;
+use vortex_error::vortex_err;
 use vortex_session::VortexSession;
 
 use crate::dtype::DType;
 use crate::expr::Expression;
 use crate::expr::display::DisplayTreeExpr;
 use crate::expr::scope::Scope;
+use crate::expr::scope::VariableRef;
 use crate::expr::traversal::TraversalOrder;
 use crate::expr::traversal::pre_order_visit_down;
+use crate::expr::variable::Variable;
+use crate::higher_order_fn::HigherOrderFunctionRef;
 use crate::scalar_fn::ScalarFnRef;
 use crate::scalar_fn::ScalarFnVTable;
 use crate::stats::rewrite::StatsRewriteCtx;
@@ -32,17 +37,11 @@ use crate::stats::rewrite::StatsRewriteCtx;
 /// Binding is purely logical: it deals only in [`DType`]s and never sees an array, a length, or an
 /// encoding.
 #[derive(Clone, Debug, PartialEq, Eq, Hash)]
-pub struct BoundExpression {
-    kind: BoundKind,
-    dtype: DType,
-}
-
-/// The per-variant contents of a [`BoundExpression`], mirroring the logical variants of
-/// [`Expression`].
-#[derive(Clone, Debug, PartialEq, Eq, Hash)]
-pub enum BoundKind {
+pub enum BoundExpression {
     /// A scalar function applied to bound children.
     Scalar {
+        /// The dtype this node evaluates to.
+        dtype: DType,
         /// The scalar function for this node.
         scalar_fn: ScalarFnRef,
         /// The bound children, in argument order.
@@ -51,8 +50,190 @@ pub enum BoundKind {
         /// consumers from destructuring a `BoundExpression` by value.
         children: Arc<Vec<BoundExpression>>,
     },
+    /// A type-checked higher-order function call.
+    HigherOrder {
+        /// The dtype this node evaluates to.
+        dtype: DType,
+        /// The higher-order function implementation.
+        higher_order_fn: HigherOrderFunctionRef,
+        /// Bound ordinary expression children.
+        children: Arc<Vec<BoundExpression>>,
+        /// Bound lambda arguments.
+        lambdas: Arc<[TypedLambda]>,
+    },
     /// The scope itself. Its dtype is the scope's root dtype.
-    Root,
+    Root {
+        /// The dtype this node evaluates to.
+        dtype: DType,
+    },
+    /// A resolved reference to a bound variable.
+    Variable {
+        /// The dtype this node evaluates to.
+        dtype: DType,
+        /// The source-level name, retained for display and diagnostics.
+        variable: Variable,
+        /// The lexical location resolved while binding.
+        variable_ref: VariableRef,
+    },
+}
+
+/// A type-checked lambda expression.
+///
+/// A higher-order function's type-checked lambda argument.
+///
+/// This is deliberately separate from [`BoundExpression`]: a lambda is not a value and has no
+/// dtype. The higher-order function establishes the parameter bindings, binds the body, and stores
+/// the resulting typed lambda in its own state.
+#[derive(Clone, Debug, PartialEq, Eq, Hash)]
+pub struct TypedLambda {
+    /// The parameters and their dtypes: the argument side of the function type.
+    params: Box<[Variable]>,
+    param_dtypes: Box<[DType]>,
+    /// The lexical locations assigned to parameters during binding.
+    param_refs: Box<[VariableRef]>,
+    /// The frame containing this lambda's parameters.
+    parameter_frame: usize,
+    body: Arc<BoundExpression>,
+}
+
+impl TypedLambda {
+    /// Bind `lambda` against a scope containing its parameter bindings.
+    ///
+    /// The higher-order function determines each parameter type and installs it in `scope`
+    /// before binding. Keeping the bindings in one scope makes that scope the single source of
+    /// truth for both the lambda body and its function signature.
+    pub fn bind(lambda: &crate::expr::Lambda, scope: &Scope) -> VortexResult<Self> {
+        vortex_ensure!(
+            scope.depth() > 0,
+            "lambda parameters must be bound in a lexical frame"
+        );
+        let parameter_frame = scope.depth() - 1;
+        let parameter_bindings = lambda
+            .params()
+            .iter()
+            .map(|param| {
+                scope
+                    .resolve(param)
+                    .map(|(dtype, variable_ref)| (dtype.clone(), variable_ref))
+                    .ok_or_else(|| {
+                        vortex_err!("lambda parameter '{param}' is not bound in its scope")
+                    })
+            })
+            .collect::<VortexResult<Vec<_>>>()?;
+        vortex_ensure!(
+            parameter_bindings
+                .iter()
+                .all(|(_, variable_ref)| variable_ref.frame() == parameter_frame),
+            "lambda parameters must be bound in the innermost lexical frame"
+        );
+        let body = lambda.body().bind_scope(scope)?;
+
+        Ok(Self {
+            params: lambda.params().to_vec().into_boxed_slice(),
+            param_dtypes: parameter_bindings
+                .iter()
+                .map(|(dtype, _)| dtype.clone())
+                .collect(),
+            param_refs: parameter_bindings
+                .into_iter()
+                .map(|(_, variable_ref)| variable_ref)
+                .collect(),
+            parameter_frame,
+            body: Arc::new(body),
+        })
+    }
+
+    /// The variables this lambda binds, in declaration order.
+    pub fn params(&self) -> &[Variable] {
+        &self.params
+    }
+
+    /// The dtypes of the parameters, in declaration order.
+    pub fn param_dtypes(&self) -> &[DType] {
+        &self.param_dtypes
+    }
+
+    /// The lexical locations of this lambda's parameters.
+    pub(crate) fn param_refs(&self) -> &[VariableRef] {
+        &self.param_refs
+    }
+
+    /// The bound body.
+    pub fn body(&self) -> &BoundExpression {
+        &self.body
+    }
+
+    /// The dtype the body evaluates to — the result side of the function type.
+    pub fn body_dtype(&self) -> &DType {
+        self.body.dtype()
+    }
+
+    /// The outer lexical bindings read directly by this lambda body.
+    ///
+    /// Nested lambdas are deliberately not traversed: they become closures when their enclosing
+    /// higher-order expression is applied, at which point their own free bindings are available.
+    pub(crate) fn free_variables(&self) -> Vec<VariableRef> {
+        fn collect(
+            expr: &BoundExpression,
+            parameter_frame: usize,
+            variables: &mut Vec<VariableRef>,
+        ) {
+            match expr {
+                BoundExpression::Variable { variable_ref, .. }
+                    if variable_ref.frame() < parameter_frame
+                        && !variables.contains(variable_ref) =>
+                {
+                    variables.push(*variable_ref);
+                }
+                BoundExpression::Scalar { children, .. }
+                | BoundExpression::HigherOrder { children, .. } => {
+                    for child in children.iter() {
+                        collect(child, parameter_frame, variables);
+                    }
+                }
+                BoundExpression::Root { .. } | BoundExpression::Variable { .. } => {}
+            }
+        }
+
+        let mut variables = Vec::new();
+        collect(&self.body, self.parameter_frame, &mut variables);
+        variables.sort_by_key(|variable_ref| (variable_ref.frame(), variable_ref.slot()));
+        variables
+    }
+
+    /// Validate arrays supplied to this lambda's parameter frame.
+    pub(crate) fn validate_arguments(
+        &self,
+        root: &crate::ArrayRef,
+        args: &[crate::ArrayRef],
+    ) -> VortexResult<()> {
+        vortex_ensure!(
+            args.len() == self.params.len(),
+            "lambda takes {} parameters but was applied with {} arguments",
+            self.params.len(),
+            args.len()
+        );
+        for ((param, dtype), arg) in self.params.iter().zip(&self.param_dtypes).zip(args) {
+            vortex_ensure!(
+                arg.dtype() == dtype,
+                "lambda parameter '{param}' expects dtype {dtype}, got {}",
+                arg.dtype()
+            );
+            vortex_ensure!(
+                arg.len() == root.len(),
+                "lambda parameter '{param}' has length {}, expected {}",
+                arg.len(),
+                root.len()
+            );
+        }
+        Ok(())
+    }
+}
+
+impl Display for TypedLambda {
+    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
+        write!(f, "({}) -> {}", self.params.iter().join(", "), self.body)
+    }
 }
 
 /// A bound-expression wrapper that compares shared tree identity instead of structure.
@@ -61,23 +242,66 @@ pub struct ExactBoundExpr(pub BoundExpression);
 
 impl PartialEq for ExactBoundExpr {
     fn eq(&self, other: &Self) -> bool {
-        match (&self.0.kind, &other.0.kind) {
-            (BoundKind::Root, BoundKind::Root) => self.0.dtype == other.0.dtype,
+        match (&self.0, &other.0) {
             (
-                BoundKind::Scalar {
+                BoundExpression::Root { dtype: lhs_dtype },
+                BoundExpression::Root { dtype: rhs_dtype },
+            ) => lhs_dtype == rhs_dtype,
+            (
+                BoundExpression::Scalar {
+                    dtype: lhs_dtype,
                     scalar_fn: lhs_fn,
                     children: lhs_children,
                 },
-                BoundKind::Scalar {
+                BoundExpression::Scalar {
+                    dtype: rhs_dtype,
                     scalar_fn: rhs_fn,
                     children: rhs_children,
                 },
             ) => {
                 lhs_fn == rhs_fn
                     && Arc::ptr_eq(lhs_children, rhs_children)
-                    && self.0.dtype == other.0.dtype
+                    && lhs_dtype == rhs_dtype
+            }
+            (
+                BoundExpression::HigherOrder {
+                    dtype: lhs_dtype,
+                    higher_order_fn: lhs_fn,
+                    children: lhs_children,
+                    lambdas: lhs_lambdas,
+                },
+                BoundExpression::HigherOrder {
+                    dtype: rhs_dtype,
+                    higher_order_fn: rhs_fn,
+                    children: rhs_children,
+                    lambdas: rhs_lambdas,
+                },
+            ) => {
+                lhs_fn == rhs_fn
+                    && Arc::ptr_eq(lhs_children, rhs_children)
+                    && Arc::ptr_eq(lhs_lambdas, rhs_lambdas)
+                    && lhs_dtype == rhs_dtype
             }
-            _ => false,
+            // No catch-all: a new variant must state its own identity rather than silently
+            // comparing unequal, which would put `eq` out of step with `hash`.
+            (
+                BoundExpression::Variable {
+                    dtype: lhs_dtype,
+                    variable: lhs_var,
+                    variable_ref: lhs_ref,
+                },
+                BoundExpression::Variable {
+                    dtype: rhs_dtype,
+                    variable: rhs_var,
+                    variable_ref: rhs_ref,
+                },
+            ) => lhs_var == rhs_var && lhs_ref == rhs_ref && lhs_dtype == rhs_dtype,
+            // No catch-all: a new variant must state its own identity, or `eq` drifts out of step
+            // with `hash` and keys stop equalling themselves.
+            (BoundExpression::Root { .. }, _)
+            | (BoundExpression::Scalar { .. }, _)
+            | (BoundExpression::HigherOrder { .. }, _)
+            | (BoundExpression::Variable { .. }, _) => false,
         }
     }
 }
@@ -88,16 +312,37 @@ impl Hash for ExactBoundExpr {
     fn hash<H: Hasher>(&self, state: &mut H) {
         // DType differences are resolved by equality. Omitting the potentially lazy dtype keeps
         // identity-keyed cache lookups from deserializing an entire schema just to compute a hash.
-        match &self.0.kind {
-            BoundKind::Root => state.write_u8(0),
-            BoundKind::Scalar {
+        match &self.0 {
+            BoundExpression::Root { .. } => state.write_u8(0),
+            BoundExpression::Variable {
+                variable,
+                variable_ref,
+                ..
+            } => {
+                state.write_u8(2);
+                variable.hash(state);
+                variable_ref.hash(state);
+            }
+            BoundExpression::Scalar {
                 scalar_fn,
                 children,
+                ..
             } => {
                 state.write_u8(1);
                 scalar_fn.hash(state);
                 Arc::as_ptr(children).hash(state);
             }
+            BoundExpression::HigherOrder {
+                higher_order_fn,
+                children,
+                lambdas,
+                ..
+            } => {
+                state.write_u8(3);
+                higher_order_fn.hash(state);
+                Arc::as_ptr(children).hash(state);
+                Arc::as_ptr(lambdas).hash(state);
+            }
         }
     }
 }
@@ -105,10 +350,7 @@ impl Hash for ExactBoundExpr {
 impl BoundExpression {
     /// Create a bound root expression with the given dtype.
     pub fn new_root(dtype: DType) -> Self {
-        Self {
-            kind: BoundKind::Root,
-            dtype,
-        }
+        Self::Root { dtype }
     }
 
     /// Create a bound scalar node from a scalar function and already-bound children.
@@ -130,12 +372,43 @@ impl BoundExpression {
             .collect_vec();
         let dtype = scalar_fn.return_dtype(&arg_dtypes)?;
 
-        Ok(Self {
-            kind: BoundKind::Scalar {
-                scalar_fn,
-                children: children.into(),
-            },
+        Ok(Self::Scalar {
+            dtype,
+            scalar_fn,
+            children: children.into(),
+        })
+    }
+
+    /// Create a bound higher-order node from its bound children and lambdas.
+    pub fn try_new_higher_order(
+        higher_order_fn: HigherOrderFunctionRef,
+        children: impl IntoIterator<Item = BoundExpression>,
+        lambdas: impl Into<Box<[TypedLambda]>>,
+    ) -> VortexResult<Self> {
+        let children = Vec::from_iter(children);
+        let lambdas: Arc<[TypedLambda]> = Arc::from(lambdas.into());
+        vortex_ensure!(
+            higher_order_fn.arity().matches(children.len()),
+            "Higher-order expression arity mismatch: expected {} children but got {}",
+            higher_order_fn.arity(),
+            children.len()
+        );
+        vortex_ensure!(
+            higher_order_fn.lambda_arity() == lambdas.len(),
+            "Higher-order expression lambda arity mismatch: expected {} lambdas but got {}",
+            higher_order_fn.lambda_arity(),
+            lambdas.len()
+        );
+        let arg_dtypes = children
+            .iter()
+            .map(|child| child.dtype().clone())
+            .collect_vec();
+        let dtype = higher_order_fn.return_dtype(&arg_dtypes, &lambdas)?;
+        Ok(Self::HigherOrder {
             dtype,
+            higher_order_fn,
+            children: children.into(),
+            lambdas,
         })
     }
 
@@ -145,33 +418,41 @@ impl BoundExpression {
         children: impl IntoIterator<Item = BoundExpression>,
     ) -> VortexResult<Self> {
         let children = Vec::from_iter(children);
-        let BoundKind::Scalar { scalar_fn, .. } = &self.kind else {
-            vortex_ensure!(
-                children.is_empty(),
-                "Root expression cannot have {} children",
-                children.len()
-            );
-            return Ok(self);
-        };
-
-        Self::try_new(scalar_fn.clone(), children)
+        match &self {
+            BoundExpression::Scalar { scalar_fn, .. } => Self::try_new(scalar_fn.clone(), children),
+            BoundExpression::HigherOrder {
+                higher_order_fn,
+                lambdas,
+                ..
+            } => Self::try_new_higher_order(higher_order_fn.clone(), children, lambdas.to_vec()),
+            BoundExpression::Root { .. } | BoundExpression::Variable { .. } => {
+                vortex_ensure!(
+                    children.is_empty(),
+                    "{self} cannot have {} children",
+                    children.len()
+                );
+                Ok(self)
+            }
+        }
     }
 
     /// The dtype this expression evaluates to.
     pub fn dtype(&self) -> &DType {
-        &self.dtype
-    }
-
-    /// The per-variant contents of this node.
-    pub fn kind(&self) -> &BoundKind {
-        &self.kind
+        match self {
+            Self::Scalar { dtype, .. }
+            | Self::HigherOrder { dtype, .. }
+            | Self::Root { dtype }
+            | Self::Variable { dtype, .. } => dtype,
+        }
     }
 
-    /// The bound children of this node, in argument order. Empty for [`BoundKind::Root`].
+    /// The bound children of this node, in argument order. Empty for [`BoundExpression::Root`].
     pub fn children(&self) -> &[BoundExpression] {
-        match &self.kind {
-            BoundKind::Scalar { children, .. } => children.as_slice(),
-            BoundKind::Root => &[],
+        match self {
+            Self::Scalar { children, .. } | Self::HigherOrder { children, .. } => {
+                children.as_slice()
+            }
+            Self::Root { .. } | Self::Variable { .. } => &[],
         }
     }
 
@@ -182,9 +463,27 @@ impl BoundExpression {
 
     /// The scalar function for this node, or `None` if it is the scope root.
     pub fn as_scalar(&self) -> Option<&ScalarFnRef> {
-        match &self.kind {
-            BoundKind::Scalar { scalar_fn, .. } => Some(scalar_fn),
-            BoundKind::Root => None,
+        match self {
+            Self::Scalar { scalar_fn, .. } => Some(scalar_fn),
+            Self::HigherOrder { .. } | Self::Root { .. } | Self::Variable { .. } => None,
+        }
+    }
+
+    /// The higher-order function for this node, if any.
+    pub fn as_higher_order(&self) -> Option<&HigherOrderFunctionRef> {
+        match self {
+            Self::HigherOrder {
+                higher_order_fn, ..
+            } => Some(higher_order_fn),
+            Self::Scalar { .. } | Self::Root { .. } | Self::Variable { .. } => None,
+        }
+    }
+
+    /// The bound lambda arguments of this node.
+    pub fn lambdas(&self) -> &[TypedLambda] {
+        match self {
+            Self::HigherOrder { lambdas, .. } => lambdas,
+            Self::Scalar { .. } | Self::Root { .. } | Self::Variable { .. } => &[],
         }
     }
 
@@ -221,9 +520,17 @@ impl BoundExpression {
             .vortex_expect("Bound expression options type mismatch")
     }
 
+    /// The variable this node resolves to, if this node is a variable reference.
+    pub fn as_variable(&self) -> Option<&Variable> {
+        match self {
+            Self::Variable { variable, .. } => Some(variable),
+            Self::Scalar { .. } | Self::HigherOrder { .. } | Self::Root { .. } => None,
+        }
+    }
+
     /// Whether this node is the scope root.
     pub fn is_root(&self) -> bool {
-        matches!(self.kind, BoundKind::Root)
+        matches!(self, Self::Root { .. })
     }
 
     /// Return whether every scope root in this expression has `dtype`.
@@ -260,9 +567,31 @@ impl BoundExpression {
 
 impl Display for BoundExpression {
     fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
-        match self.kind() {
-            BoundKind::Scalar { scalar_fn, .. } => scalar_fn.fmt_sql(self, f),
-            BoundKind::Root => f.write_str("$"),
+        match self {
+            Self::Scalar { scalar_fn, .. } => scalar_fn.fmt_sql(self, f),
+            Self::HigherOrder {
+                higher_order_fn,
+                children,
+                lambdas,
+                ..
+            } => {
+                write!(f, "{higher_order_fn}(")?;
+                for (index, child) in children.iter().enumerate() {
+                    if index > 0 {
+                        write!(f, ", ")?;
+                    }
+                    Display::fmt(child, f)?;
+                }
+                for (index, lambda) in lambdas.iter().enumerate() {
+                    if !children.is_empty() || index > 0 {
+                        write!(f, ", ")?;
+                    }
+                    Display::fmt(lambda, f)?;
+                }
+                write!(f, ")")
+            }
+            Self::Root { .. } => f.write_str("$"),
+            Self::Variable { variable, .. } => write!(f, "${variable}"),
         }
     }
 }
@@ -279,26 +608,52 @@ impl Expression {
 
     /// Bind this expression against an explicit [`Scope`].
     pub fn bind_scope(&self, scope: &Scope) -> VortexResult<BoundExpression> {
-        if self.is_root() {
-            return Ok(BoundExpression::new_root(scope.root().clone()));
+        match self {
+            Expression::Root => Ok(BoundExpression::new_root(scope.root().clone())),
+            Expression::Variable(variable) => {
+                let Some((dtype, variable_ref)) = scope.resolve(variable) else {
+                    vortex_bail!("unbound variable '{variable}'");
+                };
+                Ok(BoundExpression::Variable {
+                    dtype: dtype.clone(),
+                    variable: variable.clone(),
+                    variable_ref,
+                })
+            }
+            Expression::Scalar {
+                scalar_fn,
+                children,
+            } => {
+                let children: Vec<_> = children
+                    .iter()
+                    .map(|child| child.bind_scope(scope))
+                    .try_collect()?;
+                BoundExpression::try_new(scalar_fn.clone(), children)
+            }
+            Expression::HigherOrder {
+                higher_order_fn,
+                children,
+                lambdas,
+            } => {
+                let children: Vec<_> = children
+                    .iter()
+                    .map(|child| child.bind_scope(scope))
+                    .try_collect()?;
+                let dtypes = children
+                    .iter()
+                    .map(|child| child.dtype().clone())
+                    .collect_vec();
+                let lambdas = higher_order_fn.bind_lambdas(scope, &dtypes, lambdas)?;
+                BoundExpression::try_new_higher_order(higher_order_fn.clone(), children, lambdas)
+            }
         }
-
-        let children: Vec<_> = self
-            .children()
-            .iter()
-            .map(|child| child.bind_scope(scope))
-            .try_collect()?;
-        let scalar_fn = self
-            .as_scalar()
-            .vortex_expect("root was handled above, so this is a scalar node");
-        BoundExpression::try_new(scalar_fn.clone(), children)
     }
 }
 
 /// Iterative drop to avoid stack overflows on deep trees.
 impl Drop for BoundExpression {
     fn drop(&mut self) {
-        let BoundKind::Scalar { children, .. } = &mut self.kind else {
+        let (Self::Scalar { children, .. } | Self::HigherOrder { children, .. }) = self else {
             return;
         };
         let Some(children) = Arc::get_mut(children) else {
@@ -307,7 +662,8 @@ impl Drop for BoundExpression {
 
         let mut to_drop = std::mem::take(children);
         while let Some(mut child) = to_drop.pop() {
-            if let BoundKind::Scalar { children, .. } = &mut child.kind
+            if let BoundExpression::Scalar { children, .. }
+            | BoundExpression::HigherOrder { children, .. } = &mut child
                 && let Some(grandchildren) = Arc::get_mut(children)
             {
                 to_drop.append(grandchildren);
@@ -323,11 +679,15 @@ mod tests {
     use super::*;
     use crate::dtype::Nullability;
     use crate::dtype::PType;
+    use crate::expr::checked_add;
     use crate::expr::col;
     use crate::expr::eq;
+    use crate::expr::lambda;
     use crate::expr::lit;
     use crate::expr::root;
     use crate::expr::test_harness::struct_dtype;
+    use crate::expr::var;
+    use crate::scalar_fn::fns::is_not_null::IsNotNull;
     use crate::scalar_fn::fns::literal::Literal;
 
     fn scope() -> Scope {
@@ -407,25 +767,84 @@ mod tests {
     }
 
     #[test]
-    fn clone_shares_children() -> VortexResult<()> {
-        let bound = eq(col("a"), lit(1_i32)).bind_scope(&scope())?;
-        let cloned = bound.clone();
+    fn variable_binds_to_its_scope() -> VortexResult<()> {
+        let scope = scope().with_bindings([(
+            Variable::new("value"),
+            DType::Primitive(PType::I64, Nullability::Nullable),
+        )])?;
+
+        let bound = var("value").bind_scope(&scope)?;
+        let variable = bound
+            .as_variable()
+            .vortex_expect("variable must remain bound");
+
+        assert_eq!(variable, &Variable::new("value"));
+        assert_eq!(
+            bound.dtype(),
+            &DType::Primitive(PType::I64, Nullability::Nullable)
+        );
+        Ok(())
+    }
 
-        let (BoundKind::Scalar { children: a, .. }, BoundKind::Scalar { children: b, .. }) =
-            (bound.kind(), cloned.kind())
-        else {
-            unreachable!("eq is a scalar node")
-        };
-        assert!(Arc::ptr_eq(a, b));
+    #[test]
+    fn variable_validity_is_deferred_to_its_bound_array() -> VortexResult<()> {
+        let scope = scope().with_bindings([(
+            Variable::new("value"),
+            DType::Primitive(PType::I32, Nullability::Nullable),
+        )])?;
+        let expression = checked_add(var("value"), lit(1_i32));
+
+        let validity = expression.validity()?;
+        assert!(validity.contains::<IsNotNull>()?);
+        assert_eq!(
+            validity.bind_scope(&scope)?.dtype(),
+            &DType::Bool(Nullability::NonNullable)
+        );
         Ok(())
     }
 
     #[test]
-    fn repeated_subtree_is_bound_per_occurrence() -> VortexResult<()> {
-        let shared = col("a");
-        let bound = eq(shared.clone(), shared).bind_scope(&scope())?;
-        let children = bound.children();
-        assert_eq!(children[0].dtype(), children[1].dtype());
+    fn duplicate_lambda_parameters_are_rejected() {
+        assert!(lambda(["x", "x"], var("x")).is_err());
+    }
+
+    #[test]
+    fn lambda_signature_comes_from_its_scope() -> VortexResult<()> {
+        let lambda = lambda(["value"], var("value"))?;
+        let value_dtype = DType::Primitive(PType::I64, Nullability::Nullable);
+        let lambda_scope =
+            scope().with_bindings([(Variable::new("value"), value_dtype.clone())])?;
+
+        let bound = TypedLambda::bind(&lambda, &lambda_scope)?;
+        assert_eq!(bound.body_dtype(), &value_dtype);
+        assert_eq!(bound.param_dtypes(), &[value_dtype]);
+        assert!(TypedLambda::bind(&lambda, &scope()).is_err());
+        Ok(())
+    }
+
+    #[test]
+    fn lambda_captures_only_direct_free_variables() -> VortexResult<()> {
+        let dtype = DType::Primitive(PType::I32, Nullability::NonNullable);
+        let lambda_scope = scope()
+            .with_bindings([(Variable::new("outer"), dtype.clone())])?
+            .with_bindings([(Variable::new("middle"), dtype.clone())])?
+            .with_bindings([(Variable::new("value"), dtype)])?;
+        let lambda = lambda(
+            ["value"],
+            checked_add(checked_add(var("value"), var("middle")), var("outer")),
+        )?;
+
+        let typed = TypedLambda::bind(&lambda, &lambda_scope)?;
+        let free_variables = typed.free_variables();
+
+        assert_eq!(free_variables.len(), 2);
+        assert_eq!(
+            free_variables
+                .iter()
+                .map(|variable_ref| (variable_ref.frame(), variable_ref.slot()))
+                .collect_vec(),
+            vec![(0, 0), (1, 0)]
+        );
         Ok(())
     }
 
@@ -443,7 +862,6 @@ mod tests {
 
     #[test]
     fn binding_reports_a_type_error() {
-        let expr = eq(col("a"), lit("nope"));
-        assert!(expr.bind_scope(&scope()).is_err());
+        assert!(eq(col("a"), lit("nope")).bind_scope(&scope()).is_err());
     }
 }
diff --git a/vortex-array/src/expr/display.rs b/vortex-array/src/expr/display.rs
index 345f393e779..2b2cf5c3779 100644
--- a/vortex-array/src/expr/display.rs
+++ b/vortex-array/src/expr/display.rs
@@ -9,7 +9,6 @@ use vortex_utils::tree::TreeDisplayAdapter;
 use vortex_utils::tree::write_branch_tree;
 
 use crate::expr::BoundExpression;
-use crate::expr::BoundKind;
 use crate::expr::Expression;
 use crate::scalar_fn::ChildName;
 
@@ -69,14 +68,23 @@ impl DisplayTreeNode for Expression {
     fn tree_child_name(&self, index: usize) -> ChildName {
         match self {
             Expression::Scalar { scalar_fn, .. } => scalar_fn.signature().child_name(index),
-            Expression::Root => unreachable!("the scope root has no children"),
+            Expression::HigherOrder {
+                higher_order_fn, ..
+            } => higher_order_fn.child_name(index),
+            Expression::Root | Expression::Variable(_) => {
+                unreachable!("a leaf expression has no children")
+            }
         }
     }
 
     fn fmt_tree_node(&self, f: &mut Formatter<'_>) -> fmt::Result {
         match self {
             Expression::Scalar { scalar_fn, .. } => Display::fmt(scalar_fn, f),
+            Expression::HigherOrder {
+                higher_order_fn, ..
+            } => Display::fmt(higher_order_fn, f),
             Expression::Root => write!(f, "{ROOT_DISPLAY}"),
+            Expression::Variable(variable) => write!(f, "${variable}"),
         }
     }
 }
@@ -87,16 +95,25 @@ impl DisplayTreeNode for BoundExpression {
     }
 
     fn tree_child_name(&self, index: usize) -> ChildName {
-        match self.kind() {
-            BoundKind::Scalar { scalar_fn, .. } => scalar_fn.signature().child_name(index),
-            BoundKind::Root => unreachable!("the scope root has no children"),
+        match self {
+            BoundExpression::Scalar { scalar_fn, .. } => scalar_fn.signature().child_name(index),
+            BoundExpression::HigherOrder {
+                higher_order_fn, ..
+            } => higher_order_fn.child_name(index),
+            BoundExpression::Root { .. } | BoundExpression::Variable { .. } => {
+                unreachable!("a leaf bound node has no children")
+            }
         }
     }
 
     fn fmt_tree_node(&self, f: &mut Formatter<'_>) -> fmt::Result {
-        match self.kind() {
-            BoundKind::Scalar { scalar_fn, .. } => Display::fmt(scalar_fn, f),
-            BoundKind::Root => write!(f, "{ROOT_DISPLAY}"),
+        match self {
+            BoundExpression::Scalar { scalar_fn, .. } => Display::fmt(scalar_fn, f),
+            BoundExpression::HigherOrder {
+                higher_order_fn, ..
+            } => Display::fmt(higher_order_fn, f),
+            BoundExpression::Root { .. } => write!(f, "{ROOT_DISPLAY}"),
+            BoundExpression::Variable { variable, .. } => write!(f, "${variable}"),
         }
     }
 }
diff --git a/vortex-array/src/expr/evaluator.rs b/vortex-array/src/expr/evaluator.rs
new file mode 100644
index 00000000000..4b78d38f33f
--- /dev/null
+++ b/vortex-array/src/expr/evaluator.rs
@@ -0,0 +1,130 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+use itertools::Itertools;
+use vortex_error::VortexResult;
+use vortex_error::vortex_err;
+use vortex_utils::aliases::hash_map::HashMap;
+
+use crate::ArrayRef;
+use crate::IntoArray;
+use crate::arrays::ConstantArray;
+use crate::arrays::HigherOrderFnArray;
+use crate::arrays::ScalarFnArray;
+use crate::expr::BoundExpression;
+use crate::expr::TypedLambda;
+use crate::expr::VariableRef;
+use crate::higher_order_fn::LambdaClosure;
+use crate::optimizer::ArrayOptimizer;
+use crate::scalar_fn::fns::literal::Literal;
+
+/// Applies bound expressions against their root array and lexical bindings.
+///
+/// Application constructs lazy arrays. A higher-order expression therefore produces a
+/// [`HigherOrderFnArray`], retaining lambda closures for the array vtable to execute later.
+pub struct ExprApplyCtx {
+    root: ArrayRef,
+    bindings: HashMap<VariableRef, ArrayRef>,
+}
+
+impl ExprApplyCtx {
+    /// Create an application context for a top-level expression.
+    pub fn new(root: ArrayRef) -> Self {
+        Self::with_bindings(root, HashMap::new())
+    }
+
+    /// Create an application context for a lambda invocation.
+    pub(crate) fn with_bindings(root: ArrayRef, bindings: HashMap<VariableRef, ArrayRef>) -> Self {
+        Self { root, bindings }
+    }
+
+    fn binding(&self, variable_ref: VariableRef) -> Option<&ArrayRef> {
+        self.bindings.get(&variable_ref)
+    }
+
+    fn close_lambda(
+        &self,
+        lambda: TypedLambda,
+        capture_slots: &mut Vec<ArrayRef>,
+    ) -> VortexResult<LambdaClosure> {
+        let captures = lambda
+            .free_variables()
+            .into_iter()
+            .map(|variable_ref| {
+                self.binding(variable_ref)
+                    .cloned()
+                    .map(|array| (variable_ref, array))
+                    .ok_or_else(|| {
+                        vortex_err!(
+                            "cannot capture missing lexical binding at frame {}, slot {}",
+                            variable_ref.frame(),
+                            variable_ref.slot()
+                        )
+                    })
+            })
+            .collect::<VortexResult<Vec<_>>>()?;
+        Ok(LambdaClosure::new(lambda, captures, capture_slots))
+    }
+}
+
+impl BoundExpression {
+    /// Apply this expression in `ctx`'s current lexical and row domain.
+    pub(crate) fn apply(&self, ctx: &mut ExprApplyCtx) -> VortexResult<ArrayRef> {
+        let (scalar_fn, children) = match self {
+            BoundExpression::Root { .. } => return Ok(ctx.root.clone()),
+            BoundExpression::Variable {
+                variable,
+                variable_ref,
+                ..
+            } => {
+                return ctx
+                    .binding(*variable_ref)
+                    .cloned()
+                    .ok_or_else(|| vortex_err!("cannot apply unbound variable '{variable}'"));
+            }
+            BoundExpression::HigherOrder {
+                higher_order_fn,
+                children,
+                lambdas,
+                ..
+            } => {
+                let children = children
+                    .iter()
+                    .map(|child| child.apply(ctx))
+                    .try_collect()?;
+                let mut capture_slots = Vec::new();
+                let lambdas = lambdas
+                    .iter()
+                    .cloned()
+                    .map(|lambda| ctx.close_lambda(lambda, &mut capture_slots))
+                    .collect::<VortexResult<Vec<_>>>()?;
+                return Ok(HigherOrderFnArray::try_new_with_len(
+                    higher_order_fn.clone(),
+                    children,
+                    lambdas,
+                    capture_slots,
+                    ctx.root.len(),
+                )?
+                .into_array());
+            }
+            BoundExpression::Scalar {
+                scalar_fn,
+                children,
+                ..
+            } => (scalar_fn, children),
+        };
+
+        if let Some(scalar) = scalar_fn.as_opt::<Literal>() {
+            return Ok(ConstantArray::new(scalar.clone(), ctx.root.len()).into_array());
+        }
+
+        let children = children
+            .iter()
+            .map(|child| child.apply(ctx))
+            .try_collect()?;
+        let array = ScalarFnArray::try_new_with_len(scalar_fn.clone(), children, ctx.root.len())?
+            .into_array();
+
+        array.optimize()
+    }
+}
diff --git a/vortex-array/src/expr/expression.rs b/vortex-array/src/expr/expression.rs
index db1a76853cd..707ae72f9e8 100644
--- a/vortex-array/src/expr/expression.rs
+++ b/vortex-array/src/expr/expression.rs
@@ -11,12 +11,17 @@ use std::sync::Arc;
 use itertools::Itertools;
 use vortex_error::VortexExpect;
 use vortex_error::VortexResult;
+use vortex_error::vortex_bail;
 use vortex_error::vortex_ensure;
 
 use crate::dtype::DType;
 use crate::expr::display::DisplayTreeExpr;
+use crate::expr::is_not_null;
+use crate::expr::lambda::Lambda;
 use crate::expr::traversal::TraversalOrder;
 use crate::expr::traversal::pre_order_visit_down;
+use crate::expr::variable::Variable;
+use crate::higher_order_fn::HigherOrderFunctionRef;
 use crate::scalar_fn::ScalarFnRef;
 use crate::scalar_fn::ScalarFnVTable;
 
@@ -38,8 +43,16 @@ pub enum Expression {
         /// Any children of this expression.
         children: Arc<Vec<Expression>>,
     },
+    /// A higher-order function applied to ordinary children and owned lambda syntax.
+    HigherOrder {
+        higher_order_fn: HigherOrderFunctionRef,
+        children: Arc<Vec<Expression>>,
+        lambdas: Arc<Vec<Lambda>>,
+    },
     /// The full scope of the expression evaluation.
     Root,
+    /// A name to be bound to a value.
+    Variable(Variable),
 }
 
 impl Expression {
@@ -63,16 +76,67 @@ impl Expression {
         })
     }
 
+    /// Create a higher-order function from ordinary children and owned lambdas.
+    pub fn try_new_higher_order(
+        higher_order_fn: HigherOrderFunctionRef,
+        children: impl IntoIterator<Item = Expression>,
+        lambdas: impl IntoIterator<Item = Lambda>,
+    ) -> VortexResult<Self> {
+        let children = Vec::from_iter(children);
+        let lambdas = Vec::from_iter(lambdas);
+        vortex_ensure!(
+            higher_order_fn.arity().matches(children.len()),
+            "Higher-order expression arity mismatch: expected {} children but got {}",
+            higher_order_fn.arity(),
+            children.len()
+        );
+        vortex_ensure!(
+            higher_order_fn.lambda_arity() == lambdas.len(),
+            "Higher-order expression lambda arity mismatch: expected {} lambdas but got {}",
+            higher_order_fn.lambda_arity(),
+            lambdas.len()
+        );
+        Ok(Self::HigherOrder {
+            higher_order_fn,
+            children: children.into(),
+            lambdas: lambdas.into(),
+        })
+    }
+
     /// Whether this expression is the scope root.
     pub fn is_root(&self) -> bool {
         matches!(self, Self::Root)
     }
 
+    /// The variable this expression references, if it is a variable.
+    pub fn as_variable(&self) -> Option<&Variable> {
+        match self {
+            Self::Variable(variable) => Some(variable),
+            _ => None,
+        }
+    }
+
+    pub fn as_higher_order(&self) -> Option<&HigherOrderFunctionRef> {
+        match self {
+            Self::HigherOrder {
+                higher_order_fn, ..
+            } => Some(higher_order_fn),
+            _ => None,
+        }
+    }
+
+    pub fn lambdas(&self) -> &[Lambda] {
+        match self {
+            Self::HigherOrder { lambdas, .. } => lambdas,
+            _ => &[],
+        }
+    }
+
     /// Returns the scalar fn for this expression, or `None` if it is not a scalar node.
     pub fn as_scalar(&self) -> Option<&ScalarFnRef> {
         match self {
             Self::Scalar { scalar_fn, .. } => Some(scalar_fn),
-            Self::Root => None,
+            _ => None,
         }
     }
 
@@ -96,11 +160,13 @@ impl Expression {
             .vortex_expect("Expression options type mismatch")
     }
 
-    /// Returns the children of this expression.
+    /// Returns the sub-expressions of this node.
     pub fn children(&self) -> &[Expression] {
         match self {
-            Self::Scalar { children, .. } => children.as_slice(),
-            Self::Root => NO_CHILDREN,
+            Self::Scalar { children, .. } | Self::HigherOrder { children, .. } => {
+                children.as_slice()
+            }
+            _ => NO_CHILDREN,
         }
     }
 
@@ -116,14 +182,6 @@ impl Expression {
     ) -> VortexResult<Self> {
         let children = Vec::from_iter(children);
         match &self {
-            Self::Root => {
-                vortex_ensure!(
-                    children.is_empty(),
-                    "Expression arity mismatch: root expects 0 children but got {}",
-                    children.len()
-                );
-                Ok(Self::Root)
-            }
             Self::Scalar { scalar_fn, .. } => {
                 vortex_ensure!(
                     scalar_fn.signature().arity().matches(children.len()),
@@ -136,6 +194,19 @@ impl Expression {
                     children: children.into(),
                 })
             }
+            Self::HigherOrder {
+                higher_order_fn,
+                lambdas,
+                ..
+            } => Self::try_new_higher_order(higher_order_fn.clone(), children, lambdas.to_vec()),
+            _ => {
+                vortex_ensure!(
+                    children.is_empty(),
+                    "Expression arity mismatch: a leaf expects 0 children but got {}",
+                    children.len()
+                );
+                Ok(self)
+            }
         }
     }
 
@@ -143,6 +214,10 @@ impl Expression {
     pub fn return_dtype(&self, scope: &DType) -> VortexResult<DType> {
         match self {
             Self::Root => Ok(scope.clone()),
+            Self::Variable(variable) => vortex_bail!(
+                "variable '{variable}' can only be typed by binding against a scope with bindings"
+            ),
+            Self::HigherOrder { .. } => Ok(self.bind(scope)?.dtype().clone()),
             Self::Scalar {
                 scalar_fn,
                 children,
@@ -163,6 +238,10 @@ impl Expression {
         match self {
             // The scope is exactly as valid as itself.
             Self::Root => Ok(Self::Root),
+            // This is evaluated later against the array bound to the variable by a higher-order
+            // function, yielding that array's validity as a non-nullable boolean mask.
+            Self::Variable(_) => Ok(is_not_null(self.clone())),
+            Self::HigherOrder { .. } => Ok(is_not_null(self.clone())),
             Self::Scalar { scalar_fn, .. } => scalar_fn.validity(self),
         }
     }
@@ -174,6 +253,27 @@ impl Expression {
     pub fn fmt_sql(&self, f: &mut Formatter<'_>) -> fmt::Result {
         match self {
             Self::Root => write!(f, "$"),
+            Self::Variable(variable) => write!(f, "${variable}"),
+            Self::HigherOrder {
+                higher_order_fn,
+                children,
+                lambdas,
+            } => {
+                write!(f, "{higher_order_fn}(")?;
+                for (index, child) in children.iter().enumerate() {
+                    if index > 0 {
+                        write!(f, ", ")?;
+                    }
+                    Display::fmt(child, f)?;
+                }
+                for (index, lambda) in lambdas.iter().enumerate() {
+                    if !children.is_empty() || index > 0 {
+                        write!(f, ", ")?;
+                    }
+                    Display::fmt(lambda, f)?;
+                }
+                write!(f, ")")
+            }
             Self::Scalar { scalar_fn, .. } => scalar_fn.fmt_sql(self, f),
         }
     }
@@ -269,19 +369,24 @@ impl Display for Expression {
 /// Iterative drop for expression to avoid stack overflows.
 impl Drop for Expression {
     fn drop(&mut self) {
-        let Self::Scalar { children, .. } = self else {
-            return;
-        };
-        let Some(children) = Arc::get_mut(children) else {
-            return;
-        };
+        let mut children_to_drop = Vec::new();
+        match self {
+            Self::Scalar { children, .. } | Self::HigherOrder { children, .. } => {
+                if let Some(children) = Arc::get_mut(children) {
+                    children_to_drop.append(children);
+                }
+            }
+            Self::Root | Self::Variable(_) => return,
+        }
 
-        let mut children_to_drop = std::mem::take(children);
         while let Some(mut child) = children_to_drop.pop() {
-            if let Self::Scalar { children, .. } = &mut child
-                && let Some(expr_children) = Arc::get_mut(children)
-            {
-                children_to_drop.append(expr_children);
+            match &mut child {
+                Self::Scalar { children, .. } | Self::HigherOrder { children, .. } => {
+                    if let Some(expr_children) = Arc::get_mut(children) {
+                        children_to_drop.append(expr_children);
+                    }
+                }
+                Self::Root | Self::Variable(_) => {}
             }
         }
     }
diff --git a/vortex-array/src/expr/exprs.rs b/vortex-array/src/expr/exprs.rs
index fb8bfe227aa..e1b8a6968b5 100644
--- a/vortex-array/src/expr/exprs.rs
+++ b/vortex-array/src/expr/exprs.rs
@@ -17,6 +17,10 @@ use crate::dtype::FieldNames;
 use crate::dtype::Nullability;
 use crate::expr::BoundExpression;
 use crate::expr::Expression;
+use crate::expr::lambda::Lambda;
+use crate::expr::variable::Variable;
+use crate::higher_order_fn::HigherOrderFunctionRef;
+use crate::higher_order_fn::fns::list_transform::ListTransform;
 use crate::scalar::Scalar;
 use crate::scalar::ScalarValue;
 use crate::scalar_fn::EmptyOptions;
@@ -59,7 +63,7 @@ use crate::scalar_fn::fns::zip::Zip;
 
 /// Creates an expression that references the root scope.
 ///
-/// Returns the entire input array as passed to the expression evaluator.
+/// Returns the entire input array as passed to the expression application context.
 /// This is commonly used as the starting point for field access and other operations.
 pub fn root() -> Expression {
     Expression::Root
@@ -70,6 +74,33 @@ pub fn bound_root(dtype: DType) -> BoundExpression {
     BoundExpression::new_root(dtype)
 }
 
+/// Creates an expression referencing the value bound to `name` in the surrounding lexical scope.
+pub fn var(name: impl AsRef<str>) -> Expression {
+    Variable::new(name).into()
+}
+
+/// Creates standalone lambda syntax binding `params` over `body`.
+///
+/// Returns an error when a parameter name is repeated in the same lambda.
+pub fn lambda(
+    params: impl IntoIterator<Item = impl Into<Variable>>,
+    body: Expression,
+) -> VortexResult<Lambda> {
+    Lambda::try_new(params, body)
+}
+
+/// Transform every element in a list with a single-parameter `lambda`.
+///
+/// The lambda is syntax rather than an ordinary expression child: its parameter type comes from
+/// the input list's element type when the expression is bound.
+pub fn list_transform(input: Expression, lambda: Lambda) -> VortexResult<Expression> {
+    Expression::try_new_higher_order(
+        HigherOrderFunctionRef::new(ListTransform),
+        [input],
+        [lambda],
+    )
+}
+
 /// Return whether the expression is a root expression.
 pub fn is_root(expr: &Expression) -> bool {
     expr.is_root()
diff --git a/vortex-array/src/expr/lambda.rs b/vortex-array/src/expr/lambda.rs
new file mode 100644
index 00000000000..4506f60969b
--- /dev/null
+++ b/vortex-array/src/expr/lambda.rs
@@ -0,0 +1,75 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+use std::fmt;
+use std::fmt::Display;
+use std::fmt::Formatter;
+use std::sync::Arc;
+
+use itertools::Itertools;
+use vortex_error::VortexResult;
+use vortex_error::vortex_bail;
+use vortex_utils::aliases::hash_set::HashSet;
+
+use crate::expr::Expression;
+use crate::expr::variable::Variable;
+
+/// An expression `body` evaluated with named bindings `params`.
+#[derive(Clone, Debug, PartialEq, Eq, Hash)]
+pub struct Lambda {
+    params: Arc<Vec<Variable>>,
+    body: Arc<Expression>,
+}
+
+impl Lambda {
+    /// Create a lambda binding `params` over `body`.
+    ///
+    /// Returns an error when a parameter name is repeated.
+    pub fn try_new(
+        params: impl IntoIterator<Item = impl Into<Variable>>,
+        body: Expression,
+    ) -> VortexResult<Self> {
+        let mut vars = Vec::new();
+        let mut seen = HashSet::new();
+
+        for param in params {
+            let var: Variable = param.into();
+            if !seen.insert(var.clone()) {
+                vortex_bail!("duplicate parameter");
+            }
+
+            vars.push(var)
+        }
+
+        Ok(Self {
+            params: Arc::new(vars),
+            body: Arc::new(body),
+        })
+    }
+
+    /// The variables this lambda binds, in declaration order.
+    pub fn params(&self) -> &[Variable] {
+        &self.params
+    }
+
+    /// The expression evaluated under the parameter bindings.
+    pub fn body(&self) -> &Expression {
+        &self.body
+    }
+}
+
+impl Display for Lambda {
+    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
+        write!(f, "({}) -> {}", self.params.iter().join(", "), self.body)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn duplicate_parameters_are_rejected() {
+        assert!(Lambda::try_new(["x", "x"], Expression::Root).is_err());
+    }
+}
diff --git a/vortex-array/src/expr/mod.rs b/vortex-array/src/expr/mod.rs
index fb2fbf9b478..eff3cb02687 100644
--- a/vortex-array/src/expr/mod.rs
+++ b/vortex-array/src/expr/mod.rs
@@ -58,25 +58,30 @@ pub mod analysis;
 pub mod arbitrary;
 pub mod bound_expression;
 pub mod display;
+pub mod evaluator;
 pub(crate) mod expression;
 mod exprs;
 pub(crate) mod field;
 pub mod forms;
+pub mod lambda;
 mod optimize;
 pub mod proto;
 pub mod scope;
 pub mod stats;
 pub mod transform;
 pub mod traversal;
+pub mod variable;
 
 pub use analysis::*;
 pub use bound_expression::*;
+pub use evaluator::*;
 pub use expression::*;
 pub use exprs::and;
 pub use exprs::and_collect;
 pub use exprs::between;
 pub use exprs::binary;
 pub use exprs::bound;
+pub use exprs::bound_root;
 pub use exprs::byte_length;
 pub use exprs::case_when;
 pub use exprs::case_when_no_else;
@@ -95,11 +100,13 @@ pub use exprs::ilike;
 pub use exprs::is_not_null;
 pub use exprs::is_null;
 pub use exprs::is_root;
+pub use exprs::lambda;
 pub use exprs::like;
 pub use exprs::list_contains;
 pub use exprs::list_length;
 pub use exprs::list_sum;
 pub use exprs::list_sum_opts;
+pub use exprs::list_transform;
 pub use exprs::lit;
 pub use exprs::lt;
 pub use exprs::lt_eq;
@@ -118,9 +125,12 @@ pub use exprs::root;
 pub use exprs::select;
 pub use exprs::select_exclude;
 pub use exprs::union_child_validities;
+pub use exprs::var;
 pub use exprs::variant_get;
 pub use exprs::zip_expr;
+pub use lambda::*;
 pub use scope::*;
+pub use variable::*;
 
 pub trait VortexExprExt {
     /// Accumulate all field references from this expression and its children in a set
@@ -163,6 +173,7 @@ impl PartialEq for ExactExpr {
     fn eq(&self, other: &Self) -> bool {
         match (&self.0, &other.0) {
             (Expression::Root, Expression::Root) => true,
+            (Expression::Variable(lhs), Expression::Variable(rhs)) => lhs == rhs,
             (
                 Expression::Scalar {
                     scalar_fn: lhs_fn,
@@ -173,6 +184,22 @@ impl PartialEq for ExactExpr {
                     children: rhs_children,
                 },
             ) => lhs_fn == rhs_fn && Arc::ptr_eq(lhs_children, rhs_children),
+            (
+                Expression::HigherOrder {
+                    higher_order_fn: lhs_fn,
+                    children: lhs_children,
+                    lambdas: lhs_lambdas,
+                },
+                Expression::HigherOrder {
+                    higher_order_fn: rhs_fn,
+                    children: rhs_children,
+                    lambdas: rhs_lambdas,
+                },
+            ) => {
+                lhs_fn == rhs_fn
+                    && Arc::ptr_eq(lhs_children, rhs_children)
+                    && Arc::ptr_eq(lhs_lambdas, rhs_lambdas)
+            }
             _ => false,
         }
     }
@@ -183,6 +210,10 @@ impl Hash for ExactExpr {
     fn hash<H: Hasher>(&self, state: &mut H) {
         match &self.0 {
             Expression::Root => state.write_u8(0),
+            Expression::Variable(variable) => {
+                state.write_u8(2);
+                variable.hash(state);
+            }
             Expression::Scalar {
                 scalar_fn,
                 children,
@@ -191,6 +222,16 @@ impl Hash for ExactExpr {
                 scalar_fn.hash(state);
                 Arc::as_ptr(children).hash(state);
             }
+            Expression::HigherOrder {
+                higher_order_fn,
+                children,
+                lambdas,
+            } => {
+                state.write_u8(4);
+                higher_order_fn.hash(state);
+                Arc::as_ptr(children).hash(state);
+                Arc::as_ptr(lambdas).hash(state);
+            }
         }
     }
 }
@@ -227,6 +268,7 @@ mod tests {
     use vortex_array::expr::eq;
     use vortex_array::expr::lit;
     use vortex_array::expr::root;
+    use vortex_error::VortexResult;
 
     use super::*;
     use crate::dtype::DType;
@@ -235,8 +277,6 @@ mod tests {
     use crate::dtype::PType;
     use crate::dtype::StructFields;
     use crate::expr::and;
-    use crate::expr::bound;
-    use crate::expr::case_when;
     use crate::expr::col;
     use crate::expr::get_item;
     use crate::expr::gt;
@@ -286,7 +326,7 @@ mod tests {
     }
 
     #[test]
-    fn bound_constructors_preserve_order_and_types() -> vortex_error::VortexResult<()> {
+    fn bound_constructors_preserve_order_and_types() -> VortexResult<()> {
         let value_dtype = DType::Primitive(PType::I32, Nullability::NonNullable);
         let scope = DType::Struct(
             StructFields::from_iter([("value", value_dtype.clone())]),
@@ -326,9 +366,14 @@ mod tests {
     }
 
     #[test]
-    fn expr_display() {
+    fn expr_display() -> VortexResult<()> {
         assert_eq!(col("a").to_string(), "$.a");
         assert_eq!(root().to_string(), "$");
+        assert_eq!(var("value").to_string(), "$value");
+        assert_eq!(
+            lambda(["left", "right"], var("left"))?.to_string(),
+            "(left, right) -> $left"
+        );
 
         let col1: Expression = col("col1");
         let col2: Expression = col("col2");
@@ -422,6 +467,7 @@ mod tests {
             .to_string(),
             "{dog: 32u32, cat: \"rufus\"}"
         );
+        Ok(())
     }
 
     #[test]
diff --git a/vortex-array/src/expr/optimize.rs b/vortex-array/src/expr/optimize.rs
index 3e625324a18..78063a06913 100644
--- a/vortex-array/src/expr/optimize.rs
+++ b/vortex-array/src/expr/optimize.rs
@@ -32,7 +32,7 @@ impl Expression {
     fn simplify_untyped_node(&self) -> VortexResult<Option<Expression>> {
         match self {
             Expression::Scalar { scalar_fn, .. } => scalar_fn.simplify_untyped(self),
-            Expression::Root => Ok(None),
+            _ => Ok(None),
         }
     }
 
@@ -40,7 +40,7 @@ impl Expression {
     fn simplify_node(&self, ctx: &dyn SimplifyCtx) -> VortexResult<Option<Expression>> {
         match self {
             Expression::Scalar { scalar_fn, .. } => scalar_fn.simplify(self, ctx),
-            Expression::Root => Ok(None),
+            _ => Ok(None),
         }
     }
 
@@ -51,7 +51,7 @@ impl Expression {
     ) -> VortexResult<Option<ExpressionReduceNode<'a>>> {
         match self {
             Expression::Scalar { scalar_fn, .. } => scalar_fn.reduce_expression(node),
-            Expression::Root => Ok(None),
+            _ => Ok(None),
         }
     }
 
@@ -214,12 +214,14 @@ mod tests {
     use crate::dtype::PType;
     use crate::dtype::StructFields;
     use crate::expr::cast;
+    use crate::expr::col;
     use crate::expr::eq;
     use crate::expr::get_item;
     use crate::expr::lit;
     use crate::expr::lt_eq;
     use crate::expr::or;
     use crate::expr::root;
+    use crate::expr::test_harness::struct_dtype;
     use crate::scalar::Scalar;
     use crate::scalar_fn::fns::literal::Literal;
 
@@ -272,4 +274,18 @@ mod tests {
         assert_eq!(rhs, &Scalar::primitive(3.0f64, Nullability::NonNullable));
         Ok(())
     }
+
+    #[test]
+    fn optimization_folds_a_literal_cast() -> VortexResult<()> {
+        let expr = lt_eq(
+            col("a"),
+            cast(
+                lit(3_i32),
+                DType::Primitive(PType::I64, Nullability::NonNullable),
+            ),
+        );
+        let optimized = expr.optimize_recursive(&struct_dtype())?;
+        assert_ne!(optimized, expr, "casting a literal should fold");
+        Ok(())
+    }
 }
diff --git a/vortex-array/src/expr/proto.rs b/vortex-array/src/expr/proto.rs
index 4f544fecafb..f98a7bfd2f1 100644
--- a/vortex-array/src/expr/proto.rs
+++ b/vortex-array/src/expr/proto.rs
@@ -2,6 +2,7 @@
 // SPDX-FileCopyrightText: Copyright the Vortex contributors
 
 use itertools::Itertools;
+use prost::Message;
 use vortex_error::VortexResult;
 use vortex_error::vortex_ensure;
 use vortex_error::vortex_err;
@@ -9,6 +10,10 @@ use vortex_proto::expr as pb;
 use vortex_session::VortexSession;
 
 use crate::expr::Expression;
+use crate::expr::Lambda;
+use crate::expr::Variable;
+use crate::higher_order_fn::HigherOrderFunctionId;
+use crate::higher_order_fn::session::HigherOrderFunctionSessionExt;
 use crate::scalar_fn::ForeignScalarFnVTable;
 use crate::scalar_fn::ScalarFnId;
 use crate::scalar_fn::session::ScalarFnSessionExt;
@@ -18,18 +23,104 @@ pub trait ExprSerializeProtoExt {
     fn serialize_proto(&self) -> VortexResult<pb::Expr>;
 }
 
-/// The wire id for [`Expression::Root`], retained from when `Root` was a scalar function so that
-/// already-serialized expressions keep round-tripping.
+/// The wire id for [`Expression::Root`].
 pub(crate) const ROOT_ID: &str = "vortex.root";
 
+/// The wire id for [`Expression::Variable`].
+pub(crate) const VARIABLE_ID: &str = "vortex.var";
+
+/// The wire id for [`Expression::Lambda`].
+pub(crate) const LAMBDA_ID: &str = "vortex.lambda";
+
+pub trait LambdaSerializeProtoExt {
+    fn serialize_proto(&self) -> VortexResult<pb::Expr>;
+}
+
+impl LambdaSerializeProtoExt for Lambda {
+    fn serialize_proto(&self) -> VortexResult<pb::Expr> {
+        Ok(pb::Expr {
+            id: LAMBDA_ID.to_string(),
+            children: vec![self.body().serialize_proto()?],
+            metadata: Some(
+                pb::LambdaOpts {
+                    params: self
+                        .params()
+                        .iter()
+                        .map(|variable| variable.name().to_string())
+                        .collect(),
+                }
+                .encode_to_vec(),
+            ),
+        })
+    }
+}
+
+impl Lambda {
+    pub fn from_proto(expr: &pb::Expr, session: &VortexSession) -> VortexResult<Self> {
+        vortex_ensure!(
+            expr.id == LAMBDA_ID,
+            "expected lambda expression id '{LAMBDA_ID}', got '{}'",
+            expr.id
+        );
+        vortex_ensure!(
+            expr.children.len() == 1,
+            "a lambda must have exactly one child, its body, got {}",
+            expr.children.len()
+        );
+        let opts = pb::LambdaOpts::decode(expr.metadata())?;
+        Self::try_new(
+            opts.params.into_iter().map(Variable::new),
+            Expression::from_proto(&expr.children[0], session)?,
+        )
+    }
+}
+
 impl ExprSerializeProtoExt for Expression {
     fn serialize_proto(&self) -> VortexResult<pb::Expr> {
-        let Some(scalar_fn) = self.as_scalar() else {
-            return Ok(pb::Expr {
-                id: ROOT_ID.to_string(),
-                children: vec![],
-                metadata: Some(vec![]),
-            });
+        let scalar_fn = match self {
+            Expression::Root => {
+                return Ok(pb::Expr {
+                    id: ROOT_ID.to_string(),
+                    children: vec![],
+                    metadata: Some(vec![]),
+                });
+            }
+            Expression::Variable(variable) => {
+                return Ok(pb::Expr {
+                    id: VARIABLE_ID.to_string(),
+                    children: vec![],
+                    metadata: Some(
+                        pb::VariableOpts {
+                            name: variable.name().to_string(),
+                        }
+                        .encode_to_vec(),
+                    ),
+                });
+            }
+            Expression::HigherOrder {
+                higher_order_fn,
+                children,
+                lambdas,
+            } => {
+                let children = children
+                    .iter()
+                    .map(ExprSerializeProtoExt::serialize_proto)
+                    .chain(lambdas.iter().map(LambdaSerializeProtoExt::serialize_proto))
+                    .collect::<VortexResult<Vec<_>>>()?;
+                let metadata = higher_order_fn.serialize()?.ok_or_else(|| {
+                    vortex_err!(
+                        "Expression '{}' is not serializable: {}",
+                        higher_order_fn.id(),
+                        self
+                    )
+                })?;
+                return Ok(pb::Expr {
+                    id: higher_order_fn.id().to_string(),
+                    children,
+                    metadata: Some(metadata),
+                });
+            }
+            Expression::Scalar { scalar_fn, .. } => scalar_fn,
         };
 
         let children = self
@@ -56,7 +147,8 @@ impl ExprSerializeProtoExt for Expression {
 
 impl Expression {
     pub fn from_proto(expr: &pb::Expr, session: &VortexSession) -> VortexResult<Expression> {
-        // Root is not a registered scalar fn, so it must be resolved before the registry lookup.
+        // These are language primitives rather than registered scalar fns, so they must be
+        // resolved before the registry lookup below.
         if expr.id == ROOT_ID {
             vortex_ensure!(
                 expr.children.is_empty(),
@@ -66,6 +158,49 @@ impl Expression {
             return Ok(Expression::Root);
         }
 
+        if expr.id == VARIABLE_ID {
+            vortex_ensure!(
+                expr.children.is_empty(),
+                "a variable must have no children, got {}",
+                expr.children.len()
+            );
+            let opts = pb::VariableOpts::decode(expr.metadata())?;
+            return Ok(Expression::Variable(Variable::new(opts.name)));
+        }
+
+        #[expect(clippy::disallowed_methods, reason = "interning a dynamic id")]
+        let higher_order_id = HigherOrderFunctionId::new(expr.id.as_str());
+        if let Some(plugin) = session
+            .higher_order_functions()
+            .registry()
+            .get(&higher_order_id)
+        {
+            let higher_order_fn = plugin.deserialize(expr.metadata(), session)?;
+            vortex_ensure!(
+                expr.children.len() >= higher_order_fn.lambda_arity(),
+                "higher-order expression '{}' has too few children for {} lambdas",
+                higher_order_fn.id(),
+                higher_order_fn.lambda_arity()
+            );
+            let ordinary_children = expr.children.len() - higher_order_fn.lambda_arity();
+            vortex_ensure!(
+                higher_order_fn.arity().matches(ordinary_children),
+                "higher-order expression '{}' expected {} ordinary children, got {}",
+                higher_order_fn.id(),
+                higher_order_fn.arity(),
+                ordinary_children
+            );
+            let children = expr.children[..ordinary_children]
+                .iter()
+                .map(|child| Expression::from_proto(child, session))
+                .collect::<VortexResult<Vec<_>>>()?;
+            let lambdas = expr.children[ordinary_children..]
+                .iter()
+                .map(|child| Lambda::from_proto(child, session))
+                .collect::<VortexResult<Vec<_>>>()?;
+            return Expression::try_new_higher_order(higher_order_fn, children, lambdas);
+        }
+
         #[expect(clippy::disallowed_methods, reason = "interning a dynamic id")]
         let expr_id = ScalarFnId::new(expr.id.as_str());
         let children = expr
@@ -141,6 +276,23 @@ mod tests {
         assert_eq!(&deser_expr, &expr);
     }
 
+    /// A lambda carries its body as a child, so a malformed message must be rejected rather than
+    /// silently producing a lambda with the wrong arity.
+    #[test]
+    fn a_lambda_without_a_body_is_rejected() {
+        let malformed = pb::Expr {
+            id: "vortex.lambda".to_string(),
+            children: vec![],
+            metadata: Some(
+                pb::LambdaOpts {
+                    params: vec!["x".to_string()],
+                }
+                .encode_to_vec(),
+            ),
+        };
+        assert!(Expression::from_proto(&malformed, &array_session()).is_err());
+    }
+
     #[test]
     fn unknown_expression_id_allow_unknown() {
         let session = VortexSession::empty().with::<ScalarFnSession>();
diff --git a/vortex-array/src/expr/scope.rs b/vortex-array/src/expr/scope.rs
index 04de3202cf5..f42d8db94f2 100644
--- a/vortex-array/src/expr/scope.rs
+++ b/vortex-array/src/expr/scope.rs
@@ -1,28 +1,171 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright the Vortex contributors
 
+use std::sync::Arc;
+
+use vortex_error::VortexResult;
+use vortex_error::vortex_bail;
+use vortex_utils::aliases::hash_set::HashSet;
+
 use crate::dtype::DType;
+use crate::expr::variable::Variable;
+
+/// A stable location for a variable resolved in a lexical [`Scope`].
+///
+/// Frames are indexed from the outermost binding and slots are indexed in declaration order.
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
+pub struct VariableRef {
+    frame: usize,
+    slot: usize,
+}
+
+impl VariableRef {
+    fn new(frame: usize, slot: usize) -> Self {
+        Self { frame, slot }
+    }
+
+    /// The lexical frame containing this binding, counted from the outermost frame.
+    pub fn frame(&self) -> usize {
+        self.frame
+    }
+
+    /// The binding's position in its frame, in declaration order.
+    pub fn slot(&self) -> usize {
+        self.slot
+    }
+}
+
+/// A set of named bindings introduced by a single binder.
+///
+/// A lambda pushes exactly one frame, holding its parameters. Names within a frame must be unique;
+/// a name in an inner frame shadows the same name in an outer one.
+#[derive(Clone, Debug, PartialEq, Eq, Hash)]
+pub struct Frame(Arc<Vec<(Variable, DType)>>);
+
+impl Frame {
+    /// Create a frame from its bindings, in declaration order.
+    ///
+    /// Errors if a name is bound more than once, since a reference to it would be ambiguous.
+    pub fn try_new(bindings: impl IntoIterator<Item = (Variable, DType)>) -> VortexResult<Self> {
+        let mut seen = HashSet::new();
+        let mut frame = Vec::new();
+        for (variable, dtype) in bindings {
+            if !seen.insert(variable.clone()) {
+                vortex_bail!("duplicate binding in a single frame");
+            }
+
+            frame.push((variable, dtype));
+        }
+
+        Ok(Self(Arc::new(frame)))
+    }
+
+    /// The bindings in this frame, in declaration order.
+    pub fn bindings(&self) -> &[(Variable, DType)] {
+        &self.0
+    }
+
+    /// The dtype and declaration slot bound to `name` in this frame, if any.
+    fn get(&self, name: &Variable) -> Option<(&DType, usize)> {
+        self.0
+            .iter()
+            .enumerate()
+            .find_map(|(slot, (bound, dtype))| (bound == name).then_some((dtype, slot)))
+    }
+}
 
 /// The context an [`Expression`](crate::expr::Expression) is bound against.
 ///
-/// Today a scope is just the dtype that [`root`](crate::expr::root) resolves to. It is an opaque
-/// struct rather than a bare [`DType`] so that lexical bindings can be added later without changing
-/// [`Expression::bind_scope`](crate::expr::Expression::bind_scope)'s signature.
+/// A scope is the dtype that [`root`](crate::expr::root) resolves to, plus a stack of [`Frame`]s
+/// holding named bindings. The stack resolves names lexically, including inner-frame shadowing.
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub struct Scope {
     root: DType,
+    /// Innermost frame last.
+    frames: Vec<Frame>,
 }
 
 impl Scope {
-    /// Create a scope in which `root` resolves to the given dtype.
+    /// Create a scope in which `root` resolves to the given dtype, with no bindings.
     pub fn new(root: DType) -> Self {
-        Self { root }
+        Self {
+            root,
+            frames: Vec::new(),
+        }
     }
 
     /// The dtype that `root` resolves to.
     pub fn root(&self) -> &DType {
         &self.root
     }
+
+    /// The number of enclosing binders.
+    pub fn depth(&self) -> usize {
+        self.frames.len()
+    }
+
+    /// Return this scope extended with one more frame, which becomes the innermost.
+    pub fn push_frame(&self, frame: Frame) -> Self {
+        let mut frames = self.frames.clone();
+        frames.push(frame);
+        Self {
+            root: self.root.clone(),
+            frames,
+        }
+    }
+
+    /// Add a named binding to this scope's current frame.
+    ///
+    /// Prefer [`Self::push_frame`] for a lexical binder. This mutating helper is retained for
+    /// callers constructing one frame incrementally.
+    pub fn bind(&mut self, variable: Variable, dtype: DType) -> VortexResult<()> {
+        if self.frames.is_empty() {
+            self.frames.push(Frame(Arc::new(Vec::new())));
+        }
+        let Some(frame) = self.frames.last_mut() else {
+            unreachable!("a frame was inserted above")
+        };
+        let bindings = Arc::make_mut(&mut frame.0);
+        if bindings.iter().any(|(bound, _)| bound == &variable) {
+            vortex_bail!("variable '{variable}' is already bound in the current frame");
+        }
+        bindings.push((variable, dtype));
+        Ok(())
+    }
+
+    /// Return this scope extended with one frame containing `bindings`.
+    pub fn with_bindings(
+        self,
+        bindings: impl IntoIterator<Item = (Variable, DType)>,
+    ) -> VortexResult<Self> {
+        Ok(self.push_frame(Frame::try_new(bindings)?))
+    }
+
+    /// Return this scope with the same lexical frames and a different `root` dtype.
+    ///
+    /// A higher-order function uses this when its lambda body has a new implicit input while
+    /// retaining access to surrounding lambda frames.
+    pub fn with_root(&self, root: DType) -> Self {
+        Self {
+            root,
+            frames: self.frames.clone(),
+        }
+    }
+
+    /// Resolve `name`, searching innermost-first so that inner bindings shadow outer ones.
+    ///
+    /// Returns the bound dtype and stable lexical location of `name`.
+    pub fn resolve(&self, name: &Variable) -> Option<(&DType, VariableRef)> {
+        self.frames
+            .iter()
+            .enumerate()
+            .rev()
+            .find_map(|(frame, bindings)| {
+                bindings
+                    .get(name)
+                    .map(|(dtype, slot)| (dtype, VariableRef::new(frame, slot)))
+            })
+    }
 }
 
 impl From<DType> for Scope {
@@ -33,8 +176,19 @@ impl From<DType> for Scope {
 
 #[cfg(test)]
 mod tests {
+    use vortex_error::VortexResult;
+
     use super::*;
     use crate::dtype::Nullability;
+    use crate::dtype::PType;
+
+    fn i32_() -> DType {
+        DType::Primitive(PType::I32, Nullability::NonNullable)
+    }
+
+    fn utf8() -> DType {
+        DType::Utf8(Nullability::NonNullable)
+    }
 
     #[test]
     fn root_round_trips() {
@@ -42,4 +196,63 @@ mod tests {
         assert_eq!(Scope::new(dtype.clone()).root(), &dtype);
         assert_eq!(Scope::from(dtype.clone()).root(), &dtype);
     }
+
+    #[test]
+    fn an_empty_scope_resolves_nothing() {
+        let scope = Scope::new(i32_());
+        assert_eq!(scope.depth(), 0);
+        assert!(scope.resolve(&Variable::new("x")).is_none());
+    }
+
+    #[test]
+    fn resolve_tracks_the_frame_and_slot() -> VortexResult<()> {
+        let scope = Scope::new(i32_())
+            .push_frame(Frame::try_new([(Variable::new("a"), i32_())])?)
+            .push_frame(Frame::try_new([
+                (Variable::new("b"), utf8()),
+                (Variable::new("c"), i32_()),
+            ])?);
+
+        assert_eq!(
+            scope.resolve(&Variable::new("a")),
+            Some((&i32_(), VariableRef::new(0, 0)))
+        );
+        assert_eq!(
+            scope.resolve(&Variable::new("b")),
+            Some((&utf8(), VariableRef::new(1, 0)))
+        );
+        assert_eq!(
+            scope.resolve(&Variable::new("c")),
+            Some((&i32_(), VariableRef::new(1, 1)))
+        );
+        assert_eq!(scope.depth(), 2);
+        Ok(())
+    }
+
+    #[test]
+    fn inner_frames_shadow_outer_ones() -> VortexResult<()> {
+        let scope = Scope::new(i32_())
+            .push_frame(Frame::try_new([(Variable::new("x"), i32_())])?)
+            .push_frame(Frame::try_new([(Variable::new("x"), utf8())])?);
+
+        assert!(scope.resolve(&Variable::new("x")) == Some((&utf8(), VariableRef::new(1, 0))));
+        Ok(())
+    }
+
+    #[test]
+    fn duplicate_names_in_one_frame_are_rejected() {
+        assert!(
+            Frame::try_new([(Variable::new("x"), i32_()), (Variable::new("x"), utf8())]).is_err()
+        );
+    }
+
+    #[test]
+    fn pushing_a_frame_does_not_mutate_the_original() -> VortexResult<()> {
+        let outer = Scope::new(i32_());
+        let inner = outer.push_frame(Frame::try_new([(Variable::new("x"), i32_())])?);
+
+        assert!(outer.resolve(&Variable::new("x")).is_none());
+        assert!(inner.resolve(&Variable::new("x")).is_some());
+        Ok(())
+    }
 }
diff --git a/vortex-array/src/expr/transform/coerce.rs b/vortex-array/src/expr/transform/coerce.rs
index 48e52b1b4a9..94c81afaaa5 100644
--- a/vortex-array/src/expr/transform/coerce.rs
+++ b/vortex-array/src/expr/transform/coerce.rs
@@ -20,8 +20,9 @@ use crate::scalar_fn::fns::literal::Literal;
 pub fn coerce_expression(expr: Expression, scope: &DType) -> VortexResult<Expression> {
     // We capture scope by reference for the closure.
     let scope = scope.clone();
+
     expr.transform_up(|node| {
-        // Leaf nodes (Root, Literal) have no children to coerce.
+        // Leaf nodes have no children to coerce.
         if node.is_root() || node.is::<Literal>() || node.children().is_empty() {
             return Ok(Transformed::no(node));
         }
diff --git a/vortex-array/src/expr/traversal/mod.rs b/vortex-array/src/expr/traversal/mod.rs
index b97ae2536f7..dd82697a779 100644
--- a/vortex-array/src/expr/traversal/mod.rs
+++ b/vortex-array/src/expr/traversal/mod.rs
@@ -24,7 +24,6 @@ use vortex_error::VortexResult;
 
 use crate::expr::BoundExpression;
 use crate::expr::Expression;
-use crate::expr::bound_expression::BoundKind;
 use crate::expr::traversal::fold::NodeFolderContextWrapper;
 
 /// Signal to control a traversal's flow
@@ -534,8 +533,12 @@ impl Node for BoundExpression {
         &'a self,
         mut f: F,
     ) -> VortexResult<TraversalOrder> {
-        let BoundKind::Scalar { children, .. } = self.kind() else {
-            return Ok(TraversalOrder::Continue);
+        let children: &[Self] = match self {
+            BoundExpression::Scalar { children, .. }
+            | BoundExpression::HigherOrder { children, .. } => children,
+            BoundExpression::Root { .. } | BoundExpression::Variable { .. } => {
+                return Ok(TraversalOrder::Continue);
+            }
         };
 
         for child in children.iter() {
@@ -552,8 +555,12 @@ impl Node for BoundExpression {
         self,
         mut f: F,
     ) -> VortexResult<Transformed<Self>> {
-        let BoundKind::Scalar { children, .. } = self.kind() else {
-            return Ok(Transformed::no(self));
+        let children: &[Self] = match &self {
+            BoundExpression::Scalar { children, .. }
+            | BoundExpression::HigherOrder { children, .. } => children,
+            BoundExpression::Root { .. } | BoundExpression::Variable { .. } => {
+                return Ok(Transformed::no(self));
+            }
         };
 
         let mut order = TraversalOrder::Continue;
@@ -583,16 +590,20 @@ impl Node for BoundExpression {
     }
 
     fn iter_children<T>(&self, f: impl FnOnce(&mut dyn Iterator<Item = &Self>) -> T) -> T {
-        match self.kind() {
-            BoundKind::Scalar { children, .. } => f(&mut children.iter()),
-            BoundKind::Root => f(&mut std::iter::empty()),
+        match self {
+            BoundExpression::Scalar { children, .. }
+            | BoundExpression::HigherOrder { children, .. } => f(&mut children.iter()),
+            BoundExpression::Root { .. } | BoundExpression::Variable { .. } => {
+                f(&mut std::iter::empty())
+            }
         }
     }
 
     fn children_count(&self) -> usize {
-        match self.kind() {
-            BoundKind::Scalar { children, .. } => children.len(),
-            BoundKind::Root => 0,
+        match self {
+            BoundExpression::Scalar { children, .. }
+            | BoundExpression::HigherOrder { children, .. } => children.len(),
+            BoundExpression::Root { .. } | BoundExpression::Variable { .. } => 0,
         }
     }
 }
diff --git a/vortex-array/src/expr/variable.rs b/vortex-array/src/expr/variable.rs
new file mode 100644
index 00000000000..68e6e63ca00
--- /dev/null
+++ b/vortex-array/src/expr/variable.rs
@@ -0,0 +1,96 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+use std::fmt;
+use std::fmt::Display;
+use std::fmt::Formatter;
+use std::hash::Hash;
+use std::hash::Hasher;
+use std::sync::Arc;
+
+use crate::expr::Expression;
+
+/// The name of a value bound in a [`Scope`](crate::expr::Scope).
+#[derive(Clone, Debug)]
+pub struct Variable(Arc<str>);
+
+impl Variable {
+    /// Create a variable with the given name.
+    pub fn new(name: impl AsRef<str>) -> Self {
+        Self(Arc::from(name.as_ref()))
+    }
+
+    /// The variable's name.
+    pub fn name(&self) -> &str {
+        &self.0
+    }
+}
+
+/// Compares by name, with a pointer-equality fast path for the common case of a cloned handle.
+impl PartialEq for Variable {
+    fn eq(&self, other: &Self) -> bool {
+        Arc::ptr_eq(&self.0, &other.0) || self.0 == other.0
+    }
+}
+
+impl Eq for Variable {}
+
+impl Hash for Variable {
+    fn hash<H: Hasher>(&self, state: &mut H) {
+        // Hash the name, not the pointer, so equal names hash equally.
+        self.0.hash(state);
+    }
+}
+
+impl Display for Variable {
+    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
+        write!(f, "{}", self.0)
+    }
+}
+
+impl From<Variable> for Expression {
+    fn from(variable: Variable) -> Self {
+        Expression::Variable(variable)
+    }
+}
+
+impl<T: AsRef<str>> From<T> for Variable {
+    fn from(name: T) -> Self {
+        Self::new(name)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn equality_is_by_name_not_identity() {
+        let x1 = Variable::new("x");
+        let x2 = Variable::new("x");
+        let y = Variable::new("y");
+
+        // Independently constructed, so no shared allocation to short-circuit on.
+        assert_eq!(x1, x2);
+        assert_ne!(x1, y);
+    }
+
+    #[test]
+    fn equal_variables_hash_equally() {
+        use std::collections::hash_map::RandomState;
+        use std::hash::BuildHasher;
+
+        let state = RandomState::new();
+        assert_eq!(
+            state.hash_one(Variable::new("x")),
+            state.hash_one(Variable::new("x"))
+        );
+    }
+
+    #[test]
+    fn cloning_shares_the_name() {
+        let original = Variable::new("x");
+        let cloned = original.clone();
+        assert!(Arc::ptr_eq(&original.0, &cloned.0));
+    }
+}
diff --git a/vortex-array/src/expression.rs b/vortex-array/src/expression.rs
index a1a46196766..6c6f8995fff 100644
--- a/vortex-array/src/expression.rs
+++ b/vortex-array/src/expression.rs
@@ -1,73 +1,22 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright the Vortex contributors
 
-use itertools::Itertools;
-use vortex_error::VortexExpect;
 use vortex_error::VortexResult;
 
 use crate::ArrayRef;
-use crate::IntoArray;
-use crate::arrays::ConstantArray;
-use crate::arrays::ScalarFnArray;
 use crate::expr::BoundExpression;
-use crate::expr::BoundKind;
+use crate::expr::ExprApplyCtx;
 use crate::expr::Expression;
-use crate::optimizer::ArrayOptimizer;
-use crate::scalar_fn::fns::literal::Literal;
 
 impl ArrayRef {
     /// Apply a bound expression to this array, producing a new array in constant time.
     pub fn apply_bound(self, expr: &BoundExpression) -> VortexResult<ArrayRef> {
-        let BoundKind::Scalar {
-            scalar_fn,
-            children,
-        } = expr.kind()
-        else {
-            return Ok(self);
-        };
-
-        if let Some(scalar) = scalar_fn.as_opt::<Literal>() {
-            return Ok(ConstantArray::new(scalar.clone(), self.len()).into_array());
-        }
-
-        let children: Vec<_> = children
-            .iter()
-            .map(|child| self.clone().apply_bound(child))
-            .try_collect()?;
-
-        let array =
-            ScalarFnArray::try_new_with_len(scalar_fn.clone(), children, self.len())?.into_array();
-
-        array.optimize()
+        expr.apply(&mut ExprApplyCtx::new(self))
     }
 
     /// Apply the expression to this array, producing a new array in constant time.
     pub fn apply(self, expr: &Expression) -> VortexResult<ArrayRef> {
-        // If the expression is a root, return self.
-        if expr.is_root() {
-            return Ok(self);
-        }
-
-        // Manually convert literals to ConstantArray.
-        if let Some(scalar) = expr.as_opt::<Literal>() {
-            return Ok(ConstantArray::new(scalar.clone(), self.len()).into_array());
-        }
-
-        // Otherwise, collect the child arrays.
-        let children: Vec<_> = expr
-            .children()
-            .iter()
-            .map(|e| self.clone().apply(e))
-            .try_collect()?;
-
-        // And wrap the scalar function up in an array.
-        let scalar_fn = expr
-            .as_scalar()
-            .vortex_expect("root and literal were handled above, so this is a scalar node");
-        let array =
-            ScalarFnArray::try_new_with_len(scalar_fn.clone(), children, self.len())?.into_array();
-
-        // Optimize the resulting array's root.
-        array.optimize()
+        let bound = expr.bind(self.dtype())?;
+        self.apply_bound(&bound)
     }
 }
diff --git a/vortex-array/src/higher_order_fn/erased.rs b/vortex-array/src/higher_order_fn/erased.rs
new file mode 100644
index 00000000000..c8f8cc353b3
--- /dev/null
+++ b/vortex-array/src/higher_order_fn/erased.rs
@@ -0,0 +1,209 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+use std::any::Any;
+use std::fmt;
+use std::fmt::Debug;
+use std::fmt::Display;
+use std::fmt::Formatter;
+use std::hash::Hash;
+use std::sync::Arc;
+
+use vortex_error::VortexResult;
+use vortex_session::VortexSession;
+
+use crate::ArrayRef;
+use crate::dtype::DType;
+use crate::expr::Lambda;
+use crate::expr::Scope;
+use crate::expr::TypedLambda;
+use crate::higher_order_fn::HigherOrderFunctionId;
+use crate::higher_order_fn::HigherOrderFunctionVTable;
+use crate::higher_order_fn::LambdaCall;
+use crate::scalar_fn::Arity;
+use crate::scalar_fn::ChildName;
+use crate::validity::Validity;
+
+trait DynHigherOrderFunction: 'static + Send + Sync {
+    fn as_any(&self) -> &dyn Any;
+    fn id(&self) -> HigherOrderFunctionId;
+    fn arity(&self) -> Arity;
+    fn lambda_arity(&self) -> usize;
+    fn child_name(&self, child_idx: usize) -> ChildName;
+    fn serialize(&self) -> VortexResult<Option<Vec<u8>>>;
+    fn deserialize(&self, metadata: &[u8], session: &VortexSession) -> VortexResult<()>;
+    fn bind_lambdas(
+        &self,
+        scope: &Scope,
+        arg_dtypes: &[DType],
+        lambdas: &[Lambda],
+    ) -> VortexResult<Box<[TypedLambda]>>;
+    fn return_dtype(&self, args: &[DType], lambdas: &[TypedLambda]) -> VortexResult<DType>;
+    fn validity(&self, args: &[ArrayRef], lambdas: &[TypedLambda]) -> VortexResult<Validity>;
+    fn execute(
+        &self,
+        args: &[ArrayRef],
+        lambdas: &[LambdaCall<'_>],
+        ctx: &mut crate::ExecutionCtx,
+    ) -> VortexResult<ArrayRef>;
+    fn is_fallible(&self, lambdas: &[TypedLambda]) -> bool;
+}
+
+impl<V: HigherOrderFunctionVTable> DynHigherOrderFunction for V {
+    fn as_any(&self) -> &dyn Any {
+        self
+    }
+
+    fn id(&self) -> HigherOrderFunctionId {
+        V::id(self)
+    }
+
+    fn arity(&self) -> Arity {
+        V::arity(self)
+    }
+
+    fn lambda_arity(&self) -> usize {
+        V::lambda_arity(self)
+    }
+
+    fn child_name(&self, child_idx: usize) -> ChildName {
+        V::child_name(self, child_idx)
+    }
+
+    fn serialize(&self) -> VortexResult<Option<Vec<u8>>> {
+        V::serialize(self)
+    }
+
+    fn deserialize(&self, metadata: &[u8], session: &VortexSession) -> VortexResult<()> {
+        V::deserialize(self, metadata, session)
+    }
+
+    fn bind_lambdas(
+        &self,
+        scope: &Scope,
+        arg_dtypes: &[DType],
+        lambdas: &[Lambda],
+    ) -> VortexResult<Box<[TypedLambda]>> {
+        V::bind_lambdas(self, scope, arg_dtypes, lambdas)
+    }
+
+    fn return_dtype(&self, args: &[DType], lambdas: &[TypedLambda]) -> VortexResult<DType> {
+        V::return_dtype(self, args, lambdas)
+    }
+
+    fn validity(&self, args: &[ArrayRef], lambdas: &[TypedLambda]) -> VortexResult<Validity> {
+        V::validity(self, args, lambdas)
+    }
+
+    fn execute(
+        &self,
+        args: &[ArrayRef],
+        lambdas: &[LambdaCall<'_>],
+        ctx: &mut crate::ExecutionCtx,
+    ) -> VortexResult<ArrayRef> {
+        V::execute(self, args, lambdas, ctx)
+    }
+
+    fn is_fallible(&self, lambdas: &[TypedLambda]) -> bool {
+        V::is_fallible(self, lambdas)
+    }
+}
+
+/// A type-erased higher-order function vtable.
+#[derive(Clone)]
+pub struct HigherOrderFunctionRef(Arc<dyn DynHigherOrderFunction>);
+
+impl HigherOrderFunctionRef {
+    /// Create a reference to `vtable`.
+    pub fn new<V: HigherOrderFunctionVTable>(vtable: V) -> Self {
+        Self(Arc::new(vtable))
+    }
+
+    /// The function's global identifier.
+    pub fn id(&self) -> HigherOrderFunctionId {
+        self.0.id()
+    }
+
+    /// Whether this is an instance of `V`.
+    pub fn is<V: HigherOrderFunctionVTable>(&self) -> bool {
+        self.0.as_any().is::<V>()
+    }
+
+    pub fn arity(&self) -> Arity {
+        self.0.arity()
+    }
+
+    pub fn lambda_arity(&self) -> usize {
+        self.0.lambda_arity()
+    }
+
+    pub fn child_name(&self, child_idx: usize) -> ChildName {
+        self.0.child_name(child_idx)
+    }
+
+    pub fn serialize(&self) -> VortexResult<Option<Vec<u8>>> {
+        self.0.serialize()
+    }
+
+    pub(crate) fn deserialize(&self, metadata: &[u8], session: &VortexSession) -> VortexResult<()> {
+        self.0.deserialize(metadata, session)
+    }
+
+    pub(crate) fn bind_lambdas(
+        &self,
+        scope: &Scope,
+        arg_dtypes: &[DType],
+        lambdas: &[Lambda],
+    ) -> VortexResult<Box<[TypedLambda]>> {
+        self.0.bind_lambdas(scope, arg_dtypes, lambdas)
+    }
+
+    pub fn return_dtype(&self, args: &[DType], lambdas: &[TypedLambda]) -> VortexResult<DType> {
+        self.0.return_dtype(args, lambdas)
+    }
+
+    pub fn validity(&self, args: &[ArrayRef], lambdas: &[TypedLambda]) -> VortexResult<Validity> {
+        self.0.validity(args, lambdas)
+    }
+
+    pub fn execute(
+        &self,
+        args: &[ArrayRef],
+        lambdas: &[LambdaCall<'_>],
+        ctx: &mut crate::ExecutionCtx,
+    ) -> VortexResult<ArrayRef> {
+        self.0.execute(args, lambdas, ctx)
+    }
+
+    pub fn is_fallible(&self, lambdas: &[TypedLambda]) -> bool {
+        self.0.is_fallible(lambdas)
+    }
+}
+
+impl Debug for HigherOrderFunctionRef {
+    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
+        f.debug_tuple("HigherOrderFunctionRef")
+            .field(&self.id())
+            .finish()
+    }
+}
+
+impl Display for HigherOrderFunctionRef {
+    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
+        Display::fmt(&self.id(), f)
+    }
+}
+
+impl PartialEq for HigherOrderFunctionRef {
+    fn eq(&self, other: &Self) -> bool {
+        self.id() == other.id()
+    }
+}
+
+impl Eq for HigherOrderFunctionRef {}
+
+impl Hash for HigherOrderFunctionRef {
+    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
+        self.id().hash(state);
+    }
+}
diff --git a/vortex-array/src/higher_order_fn/fns/list_transform.rs b/vortex-array/src/higher_order_fn/fns/list_transform.rs
new file mode 100644
index 00000000000..9dc90575a7f
--- /dev/null
+++ b/vortex-array/src/higher_order_fn/fns/list_transform.rs
@@ -0,0 +1,575 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+use vortex_buffer::BufferMut;
+use vortex_error::VortexResult;
+use vortex_error::vortex_bail;
+use vortex_error::vortex_ensure;
+use vortex_session::VortexSession;
+use vortex_session::registry::CachedId;
+
+use crate::ArrayRef;
+use crate::Canonical;
+use crate::IntoArray;
+use crate::arrays::FixedSizeList;
+use crate::arrays::FixedSizeListArray;
+use crate::arrays::List;
+use crate::arrays::ListArray;
+use crate::arrays::ListView;
+use crate::arrays::ListViewArray;
+use crate::arrays::PrimitiveArray;
+use crate::arrays::fixed_size_list::FixedSizeListArrayExt;
+use crate::arrays::fixed_size_list::FixedSizeListArraySlotsExt;
+use crate::arrays::list::ListArrayExt;
+use crate::arrays::list::ListArraySlotsExt;
+use crate::arrays::listview::ListViewArrayExt;
+use crate::arrays::listview::ListViewArraySlotsExt;
+use crate::builtins::ArrayBuiltins;
+use crate::dtype::DType;
+use crate::dtype::Nullability;
+use crate::dtype::PType;
+use crate::expr::Frame;
+use crate::expr::Lambda;
+use crate::expr::Scope;
+use crate::expr::TypedLambda;
+use crate::higher_order_fn::HigherOrderFunctionId;
+use crate::higher_order_fn::HigherOrderFunctionVTable;
+use crate::higher_order_fn::LambdaCall;
+use crate::matcher::Matcher;
+use crate::scalar_fn::Arity;
+use crate::scalar_fn::ChildName;
+use crate::validity::Validity;
+
+/// Applies a single-argument lambda to every element of a list.
+///
+/// The input's list structure and list-level validity are retained while the lambda transforms
+/// the elements array. Nested invocation rebases captured outer parameters from list rows to
+/// element rows before evaluating the lambda body.
+#[derive(Clone, Debug)]
+pub struct ListTransform;
+
+impl HigherOrderFunctionVTable for ListTransform {
+    fn id(&self) -> HigherOrderFunctionId {
+        static ID: CachedId = CachedId::new("vortex.list.transform");
+        *ID
+    }
+
+    fn serialize(&self) -> VortexResult<Option<Vec<u8>>> {
+        Ok(Some(vec![]))
+    }
+
+    fn deserialize(&self, metadata: &[u8], _session: &VortexSession) -> VortexResult<()> {
+        if metadata.is_empty() {
+            Ok(())
+        } else {
+            vortex_bail!("list_transform() does not accept metadata")
+        }
+    }
+
+    fn arity(&self) -> Arity {
+        Arity::Exact(1)
+    }
+
+    fn lambda_arity(&self) -> usize {
+        1
+    }
+
+    fn child_name(&self, child_idx: usize) -> ChildName {
+        match child_idx {
+            0 => ChildName::from("input"),
+            _ => unreachable!("invalid child index {child_idx} for list_transform()"),
+        }
+    }
+
+    fn bind_lambdas(
+        &self,
+        scope: &Scope,
+        args: &[DType],
+        lambdas: &[Lambda],
+    ) -> VortexResult<Box<[TypedLambda]>> {
+        let [lambda] = lambdas else {
+            vortex_bail!("list_transform() requires exactly one lambda");
+        };
+        vortex_ensure!(
+            lambda.params().len() == 1,
+            "list_transform() lambda must take exactly one parameter, got {}",
+            lambda.params().len()
+        );
+        let element = match &args[0] {
+            DType::List(element, _) | DType::FixedSizeList(element, ..) => element,
+            dtype => vortex_bail!("list_transform() requires List or FixedSizeList, got {dtype}"),
+        };
+        Ok(Box::new([bind_lambda(lambda, scope, element)?]))
+    }
+
+    fn return_dtype(&self, args: &[DType], lambdas: &[TypedLambda]) -> VortexResult<DType> {
+        let [lambda] = lambdas else {
+            vortex_bail!("list_transform() requires exactly one lambda");
+        };
+        match &args[0] {
+            DType::List(_, nullability) => Ok(DType::List(
+                lambda.body_dtype().clone().into(),
+                *nullability,
+            )),
+            DType::FixedSizeList(_, size, nullability) => Ok(DType::FixedSizeList(
+                lambda.body_dtype().clone().into(),
+                *size,
+                *nullability,
+            )),
+            dtype => vortex_bail!("list_transform() requires List or FixedSizeList, got {dtype}"),
+        }
+    }
+
+    fn validity(&self, args: &[ArrayRef], _lambdas: &[TypedLambda]) -> VortexResult<Validity> {
+        let [input] = args else {
+            vortex_bail!("list_transform() requires exactly one input");
+        };
+        input.validity()
+    }
+
+    fn execute(
+        &self,
+        args: &[ArrayRef],
+        lambdas: &[LambdaCall<'_>],
+        ctx: &mut crate::ExecutionCtx,
+    ) -> VortexResult<ArrayRef> {
+        let [input] = args else {
+            vortex_bail!("list_transform() requires exactly one input");
+        };
+        let [lambda] = lambdas else {
+            vortex_bail!("list_transform() requires exactly one lambda");
+        };
+        let lambda_definitions = [lambda.lambda().clone()];
+        let output_dtype =
+            self.return_dtype(std::slice::from_ref(input.dtype()), &lambda_definitions)?;
+        if input.is_empty() {
+            return Ok(Canonical::empty(&output_dtype).into_array());
+        }
+
+        let input = input.clone().execute_until::<AnyList>(ctx)?;
+        transform_list(&input, lambda, lambda_is_fallible(lambda.lambda()), ctx)
+    }
+}
+
+fn bind_lambda(
+    lambda: &Lambda,
+    parent_scope: &Scope,
+    element_dtype: &DType,
+) -> VortexResult<TypedLambda> {
+    let frame = Frame::try_new(
+        lambda
+            .params()
+            .iter()
+            .cloned()
+            .map(|parameter| (parameter, element_dtype.clone())),
+    )?;
+    let lambda_scope = parent_scope
+        .with_root(element_dtype.clone())
+        .push_frame(frame);
+    TypedLambda::bind(lambda, &lambda_scope)
+}
+
+fn transform_list(
+    input: &ArrayRef,
+    lambda: &LambdaCall<'_>,
+    lambda_is_fallible: bool,
+    ctx: &mut crate::ExecutionCtx,
+) -> VortexResult<ArrayRef> {
+    if let Some(fixed_size_list) = input.as_opt::<FixedSizeList>() {
+        let parent_indices = lambda.has_captures().then(|| {
+            fixed_size_list_parent_indices(
+                fixed_size_list.list_size() as usize,
+                fixed_size_list.len(),
+            )
+        });
+        let elements = fixed_size_list.elements();
+        return Ok(FixedSizeListArray::new(
+            lambda.apply(
+                elements.clone(),
+                std::slice::from_ref(elements),
+                parent_indices.as_ref(),
+            )?,
+            fixed_size_list.list_size(),
+            fixed_size_list.fixed_size_list_validity(),
+            fixed_size_list.len(),
+        )
+        .into_array());
+    }
+
+    if let Some(list) = input.as_opt::<List>() {
+        let validity = list.list_validity();
+        if lambda.has_captures()
+            || (lambda_is_fallible
+                && !matches!(validity, Validity::NonNullable | Validity::AllValid))
+        {
+            let (elements, offsets, parent_indices) = gather_referenced_elements(
+                list.elements(),
+                list.offsets(),
+                None,
+                &validity,
+                list.len(),
+                ctx,
+            )?;
+            return Ok(ListArray::try_new(
+                lambda.apply(
+                    elements.clone(),
+                    std::slice::from_ref(&elements),
+                    lambda.has_captures().then_some(&parent_indices),
+                )?,
+                offsets,
+                validity,
+            )?
+            .into_array());
+        }
+
+        return Ok(ListArray::try_new(
+            lambda.apply(
+                list.elements().clone(),
+                std::slice::from_ref(list.elements()),
+                None,
+            )?,
+            list.offsets().clone(),
+            validity,
+        )?
+        .into_array());
+    }
+
+    if let Some(list_view) = input.as_opt::<ListView>() {
+        let validity = list_view.listview_validity();
+        if lambda_is_fallible || lambda.has_captures() {
+            let (elements, offsets, parent_indices) = gather_referenced_elements(
+                list_view.elements(),
+                list_view.offsets(),
+                Some(list_view.sizes()),
+                &validity,
+                list_view.len(),
+                ctx,
+            )?;
+            return Ok(ListArray::try_new(
+                lambda.apply(
+                    elements.clone(),
+                    std::slice::from_ref(&elements),
+                    lambda.has_captures().then_some(&parent_indices),
+                )?,
+                offsets,
+                validity,
+            )?
+            .into_array());
+        }
+
+        return Ok(ListViewArray::new(
+            lambda.apply(
+                list_view.elements().clone(),
+                std::slice::from_ref(list_view.elements()),
+                None,
+            )?,
+            list_view.offsets().clone(),
+            list_view.sizes().clone(),
+            validity,
+        )
+        .into_array());
+    }
+
+    let dtype = input.dtype();
+    vortex_bail!("list_transform() requires List, ListView, or FixedSizeList, got {dtype}")
+}
+
+fn lambda_is_fallible(lambda: &TypedLambda) -> bool {
+    match lambda.body() {
+        crate::expr::BoundExpression::Scalar {
+            scalar_fn,
+            children,
+            ..
+        } => scalar_fn.signature().is_fallible() || children.iter().any(bound_is_fallible),
+        crate::expr::BoundExpression::HigherOrder { .. } => true,
+        crate::expr::BoundExpression::Root { .. }
+        | crate::expr::BoundExpression::Variable { .. } => false,
+    }
+}
+
+fn bound_is_fallible(expr: &crate::expr::BoundExpression) -> bool {
+    match expr {
+        crate::expr::BoundExpression::Scalar {
+            scalar_fn,
+            children,
+            ..
+        } => scalar_fn.signature().is_fallible() || children.iter().any(bound_is_fallible),
+        crate::expr::BoundExpression::HigherOrder { .. } => true,
+        crate::expr::BoundExpression::Root { .. }
+        | crate::expr::BoundExpression::Variable { .. } => false,
+    }
+}
+
+/// Gather elements referenced by valid lists into a dense `List` layout.
+///
+/// A fallible lambda must only see elements in valid list rows. `ListView` can also leave gaps in
+/// its element array, so its ranges are gathered even when every list row is valid.
+fn gather_referenced_elements(
+    elements: &ArrayRef,
+    offsets: &ArrayRef,
+    sizes: Option<&ArrayRef>,
+    validity: &Validity,
+    len: usize,
+    ctx: &mut crate::ExecutionCtx,
+) -> VortexResult<(ArrayRef, ArrayRef, ArrayRef)> {
+    let u64_dtype = DType::Primitive(PType::U64, Nullability::NonNullable);
+    let offsets = offsets
+        .cast(u64_dtype.clone())?
+        .execute::<PrimitiveArray>(ctx)?;
+    let offsets = offsets.as_slice::<u64>();
+    let sizes = sizes
+        .map(|sizes| sizes.cast(u64_dtype)?.execute::<PrimitiveArray>(ctx))
+        .transpose()?;
+    let sizes = sizes.as_ref().map(|sizes| sizes.as_slice::<u64>());
+
+    let mask = validity.execute_mask(len, ctx)?;
+    let mut indices = BufferMut::<u64>::empty();
+    let mut parent_indices = BufferMut::<u64>::empty();
+    let mut new_offsets = BufferMut::<u64>::with_capacity(len + 1);
+    new_offsets.push(0);
+    for row in 0..len {
+        if mask.value(row) {
+            let start = offsets[row];
+            let end = match sizes {
+                Some(sizes) => start + sizes[row],
+                None => offsets[row + 1],
+            };
+            for index in start..end {
+                indices.push(index);
+                parent_indices.push(row as u64);
+            }
+        }
+        new_offsets.push(indices.len() as u64);
+    }
+
+    Ok((
+        elements.take(indices.freeze().into_array())?,
+        new_offsets.freeze().into_array(),
+        parent_indices.freeze().into_array(),
+    ))
+}
+
+fn fixed_size_list_parent_indices(list_size: usize, len: usize) -> ArrayRef {
+    let mut indices = BufferMut::<u64>::with_capacity(list_size * len);
+    for row in 0..len {
+        indices.extend(std::iter::repeat_n(row as u64, list_size));
+    }
+    indices.freeze().into_array()
+}
+
+/// Matches a concrete variable- or fixed-size list encoding.
+struct AnyList;
+
+impl Matcher for AnyList {
+    type Match<'a> = ();
+
+    fn try_match(array: &ArrayRef) -> Option<Self::Match<'_>> {
+        (array.as_opt::<List>().is_some()
+            || array.as_opt::<ListView>().is_some()
+            || array.as_opt::<FixedSizeList>().is_some())
+        .then_some(())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use vortex_buffer::buffer;
+    use vortex_error::VortexResult;
+
+    use super::*;
+    use crate::IntoArray;
+    use crate::VortexSessionExecute;
+    use crate::array_session;
+    use crate::arrays::BoolArray;
+    use crate::arrays::HigherOrderFn;
+    use crate::arrays::ListArray;
+    use crate::arrays::PrimitiveArray;
+    use crate::assert_arrays_eq;
+    use crate::expr::Expression;
+    use crate::expr::cast;
+    use crate::expr::checked_add;
+    use crate::expr::lambda as lambda_expr;
+    use crate::expr::list_length;
+    use crate::expr::list_transform;
+    use crate::expr::lit;
+    use crate::expr::proto::ExprSerializeProtoExt;
+    use crate::expr::root;
+    use crate::expr::var;
+    use crate::validity::Validity;
+
+    #[test]
+    fn transforms_elements_with_the_lambda_parameter() -> VortexResult<()> {
+        let input = ListArray::try_new(
+            PrimitiveArray::from_iter([1_i32, 2, 3]).into_array(),
+            buffer![0_u32, 2, 3].into_array(),
+            Validity::NonNullable,
+        )?
+        .into_array();
+        let expression = list_transform(
+            root(),
+            lambda_expr(["element"], checked_add(var("element"), lit(1_i32)))?,
+        )?;
+
+        let result = input.apply(&expression)?;
+        assert!(result.is::<HigherOrderFn>());
+        let expected = ListArray::try_new(
+            PrimitiveArray::from_iter([2_i32, 3, 4]).into_array(),
+            buffer![0_u32, 2, 3].into_array(),
+            Validity::NonNullable,
+        )?
+        .into_array();
+        let mut ctx = array_session().create_execution_ctx();
+        assert_arrays_eq!(result, expected, &mut ctx);
+        Ok(())
+    }
+
+    #[test]
+    fn rejects_a_lambda_with_the_wrong_arity() -> VortexResult<()> {
+        let expression = list_transform(root(), lambda_expr(["x", "i"], var("x"))?)?;
+        let dtype = DType::List(
+            std::sync::Arc::new(DType::Primitive(PType::I32, Nullability::NonNullable)),
+            Nullability::NonNullable,
+        );
+        assert!(expression.return_dtype(&dtype).is_err());
+        Ok(())
+    }
+
+    #[test]
+    fn nested_transforms_capture_outer_parameters() -> VortexResult<()> {
+        let inner_lists = ListArray::try_new(
+            PrimitiveArray::from_iter([1_i32, 2, 3, 4]).into_array(),
+            buffer![0_u32, 2, 3, 4].into_array(),
+            Validity::NonNullable,
+        )?
+        .into_array();
+        let input = ListArray::try_new(
+            inner_lists,
+            buffer![0_u32, 2, 3].into_array(),
+            Validity::NonNullable,
+        )?
+        .into_array();
+
+        let inner = list_transform(
+            var("outer"),
+            lambda_expr(
+                ["inner"],
+                checked_add(
+                    var("inner"),
+                    cast(
+                        list_length(var("outer")),
+                        DType::Primitive(PType::I32, Nullability::NonNullable),
+                    ),
+                ),
+            )?,
+        )?;
+        let expression = list_transform(root(), lambda_expr(["outer"], inner)?)?;
+
+        let result = input.apply(&expression)?;
+        let expected_inner_lists = ListArray::try_new(
+            PrimitiveArray::from_iter([3_i32, 4, 4, 5]).into_array(),
+            buffer![0_u32, 2, 3, 4].into_array(),
+            Validity::NonNullable,
+        )?
+        .into_array();
+        let expected = ListArray::try_new(
+            expected_inner_lists,
+            buffer![0_u32, 2, 3].into_array(),
+            Validity::NonNullable,
+        )?
+        .into_array();
+        let mut ctx = array_session().create_execution_ctx();
+        assert_arrays_eq!(result, expected, &mut ctx);
+        Ok(())
+    }
+
+    #[test]
+    fn inner_parameters_shadow_outer_parameters() -> VortexResult<()> {
+        let inner_lists = ListArray::try_new(
+            PrimitiveArray::from_iter([1_i32, 2, 3, 4]).into_array(),
+            buffer![0_u32, 2, 3, 4].into_array(),
+            Validity::NonNullable,
+        )?
+        .into_array();
+        let input = ListArray::try_new(
+            inner_lists,
+            buffer![0_u32, 2, 3].into_array(),
+            Validity::NonNullable,
+        )?
+        .into_array();
+
+        let inner = list_transform(
+            var("value"),
+            lambda_expr(["value"], checked_add(var("value"), lit(1_i32)))?,
+        )?;
+        let expression = list_transform(root(), lambda_expr(["value"], inner)?)?;
+
+        let result = input.apply(&expression)?;
+        let expected_inner_lists = ListArray::try_new(
+            PrimitiveArray::from_iter([2_i32, 3, 4, 5]).into_array(),
+            buffer![0_u32, 2, 3, 4].into_array(),
+            Validity::NonNullable,
+        )?
+        .into_array();
+        let expected = ListArray::try_new(
+            expected_inner_lists,
+            buffer![0_u32, 2, 3].into_array(),
+            Validity::NonNullable,
+        )?
+        .into_array();
+        let mut ctx = array_session().create_execution_ctx();
+        assert_arrays_eq!(result, expected, &mut ctx);
+        Ok(())
+    }
+
+    #[test]
+    fn fallible_lambdas_skip_elements_of_null_lists() -> VortexResult<()> {
+        let input = ListArray::try_new(
+            PrimitiveArray::from_iter([1_i32, i32::MAX]).into_array(),
+            buffer![0_u32, 1, 2].into_array(),
+            Validity::Array(BoolArray::from_iter([true, false]).into_array()),
+        )?
+        .into_array();
+        let expression = list_transform(
+            root(),
+            lambda_expr(["element"], checked_add(var("element"), lit(1_i32)))?,
+        )?;
+
+        let result = input.apply(&expression)?;
+        let expected = ListArray::try_new(
+            PrimitiveArray::from_iter([2_i32]).into_array(),
+            buffer![0_u32, 1, 1].into_array(),
+            Validity::Array(BoolArray::from_iter([true, false]).into_array()),
+        )?
+        .into_array();
+        let mut ctx = array_session().create_execution_ctx();
+        assert_arrays_eq!(result, expected, &mut ctx);
+        Ok(())
+    }
+
+    #[test]
+    fn higher_order_expression_round_trips() -> VortexResult<()> {
+        let expression = list_transform(
+            root(),
+            lambda_expr(["element"], checked_add(var("element"), lit(1_i32)))?,
+        )?;
+        let session = array_session();
+
+        let round_tripped = Expression::from_proto(&expression.serialize_proto()?, &session)?;
+        assert_eq!(round_tripped, expression);
+        Ok(())
+    }
+
+    #[test]
+    fn nested_higher_order_expression_round_trips() -> VortexResult<()> {
+        let inner = list_transform(
+            var("outer"),
+            lambda_expr(["inner"], checked_add(var("inner"), lit(1_i32)))?,
+        )?;
+        let expression = list_transform(root(), lambda_expr(["outer"], inner)?)?;
+        let session = array_session();
+
+        let round_tripped = Expression::from_proto(&expression.serialize_proto()?, &session)?;
+        assert_eq!(round_tripped, expression);
+        Ok(())
+    }
+}
diff --git a/vortex-array/src/higher_order_fn/fns/mod.rs b/vortex-array/src/higher_order_fn/fns/mod.rs
new file mode 100644
index 00000000000..3fe3265ccc0
--- /dev/null
+++ b/vortex-array/src/higher_order_fn/fns/mod.rs
@@ -0,0 +1,4 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+pub mod list_transform;
diff --git a/vortex-array/src/higher_order_fn/lambda.rs b/vortex-array/src/higher_order_fn/lambda.rs
new file mode 100644
index 00000000000..ed4fcaa4bcc
--- /dev/null
+++ b/vortex-array/src/higher_order_fn/lambda.rs
@@ -0,0 +1,126 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+use vortex_error::VortexResult;
+use vortex_error::vortex_ensure;
+use vortex_utils::aliases::hash_map::HashMap;
+
+use crate::ArrayRef;
+use crate::expr::ExprApplyCtx;
+use crate::expr::TypedLambda;
+use crate::expr::VariableRef;
+
+#[derive(Clone, Debug, PartialEq, Eq, Hash)]
+struct Capture {
+    variable_ref: VariableRef,
+    slot: usize,
+}
+
+/// A type-checked lambda plus the locations of its captured array slots.
+///
+/// Captures are retained as slots of the enclosing
+/// [`HigherOrderFnArray`](crate::arrays::HigherOrderFnArray), so normal array traversal,
+/// execution, equality, and explain machinery sees every dependency of the closure.
+#[derive(Clone, Debug, PartialEq, Eq, Hash)]
+pub struct LambdaClosure {
+    lambda: TypedLambda,
+    captures: Box<[Capture]>,
+}
+
+impl LambdaClosure {
+    /// Construct a closure, moving each free lexical binding into an enclosing array slot.
+    pub(crate) fn new(
+        lambda: TypedLambda,
+        captures: Vec<(VariableRef, ArrayRef)>,
+        capture_slots: &mut Vec<ArrayRef>,
+    ) -> Self {
+        let captures = captures
+            .into_iter()
+            .map(|(variable_ref, array)| {
+                let slot = capture_slots.len();
+                capture_slots.push(array);
+                Capture { variable_ref, slot }
+            })
+            .collect();
+        Self { lambda, captures }
+    }
+
+    /// The typed lambda expression.
+    pub fn lambda(&self) -> &TypedLambda {
+        &self.lambda
+    }
+
+    pub(crate) fn validate(&self, capture_count: usize) -> VortexResult<()> {
+        vortex_ensure!(
+            self.captures
+                .iter()
+                .all(|capture| capture.slot < capture_count),
+            "HigherOrderFnArray lambda capture references a missing slot"
+        );
+        Ok(())
+    }
+
+    pub(crate) fn call<'a>(&'a self, capture_slots: &[ArrayRef]) -> LambdaCall<'a> {
+        let captures = self
+            .captures
+            .iter()
+            .map(|capture| (capture.variable_ref, capture_slots[capture.slot].clone()))
+            .collect();
+        LambdaCall {
+            lambda: &self.lambda,
+            captures,
+        }
+    }
+}
+
+/// A short-lived invocation view of a [`LambdaClosure`].
+///
+/// The enclosing higher-order-function array resolves its capture slots before creating this
+/// value. Calling [`Self::apply`] builds the lambda body array; it never executes that array.
+#[derive(Debug)]
+pub struct LambdaCall<'a> {
+    lambda: &'a TypedLambda,
+    captures: Vec<(VariableRef, ArrayRef)>,
+}
+
+impl LambdaCall<'_> {
+    /// The typed lambda expression being invoked.
+    pub fn lambda(&self) -> &TypedLambda {
+        self.lambda
+    }
+
+    /// Whether this lambda reads any enclosing lexical binding.
+    pub fn has_captures(&self) -> bool {
+        !self.captures.is_empty()
+    }
+
+    /// Apply the lambda to arrays in its invocation domain.
+    ///
+    /// When `parent_indices` is provided, it maps the invocation rows to the rows of the captured
+    /// arrays. A list function, for example, can use it to map flattened element rows back to
+    /// their parent list rows before applying the lambda body.
+    pub fn apply(
+        &self,
+        root: ArrayRef,
+        args: &[ArrayRef],
+        parent_indices: Option<&ArrayRef>,
+    ) -> VortexResult<ArrayRef> {
+        self.lambda.validate_arguments(&root, args)?;
+
+        let mut bindings = HashMap::new();
+        for (variable_ref, array) in &self.captures {
+            let array = match parent_indices {
+                Some(indices) => array.take(indices.clone())?,
+                None => array.clone(),
+            };
+            bindings.insert(*variable_ref, array);
+        }
+        for (variable_ref, array) in self.lambda.param_refs().iter().zip(args) {
+            bindings.insert(*variable_ref, array.clone());
+        }
+
+        self.lambda
+            .body()
+            .apply(&mut ExprApplyCtx::with_bindings(root, bindings))
+    }
+}
diff --git a/vortex-array/src/higher_order_fn/mod.rs b/vortex-array/src/higher_order_fn/mod.rs
new file mode 100644
index 00000000000..e137530b065
--- /dev/null
+++ b/vortex-array/src/higher_order_fn/mod.rs
@@ -0,0 +1,29 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+//! Higher-order function vtable machinery.
+//!
+//! Higher-order functions combine ordinary expression children with lambdas. They are distinct
+//! from scalar functions because they establish lambda parameter types and evaluation bindings.
+
+use vortex_session::registry::Id;
+
+mod erased;
+pub use erased::HigherOrderFunctionRef;
+
+mod lambda;
+pub use lambda::LambdaCall;
+pub use lambda::LambdaClosure;
+
+mod plugin;
+pub use plugin::*;
+
+pub mod session;
+
+mod vtable;
+pub use vtable::*;
+
+pub mod fns;
+
+/// A globally unique identifier for a higher-order function.
+pub type HigherOrderFunctionId = Id;
diff --git a/vortex-array/src/higher_order_fn/plugin.rs b/vortex-array/src/higher_order_fn/plugin.rs
new file mode 100644
index 00000000000..917ff0dd6f7
--- /dev/null
+++ b/vortex-array/src/higher_order_fn/plugin.rs
@@ -0,0 +1,47 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+use std::sync::Arc;
+
+use vortex_error::VortexResult;
+use vortex_session::VortexSession;
+
+use crate::higher_order_fn::HigherOrderFunctionId;
+use crate::higher_order_fn::HigherOrderFunctionRef;
+use crate::higher_order_fn::HigherOrderFunctionVTable;
+
+/// A session-registered higher-order-function vtable.
+pub trait HigherOrderFunctionPlugin: 'static + Send + Sync {
+    fn id(&self) -> HigherOrderFunctionId;
+    fn deserialize(
+        &self,
+        metadata: &[u8],
+        session: &VortexSession,
+    ) -> VortexResult<HigherOrderFunctionRef>;
+}
+
+impl std::fmt::Debug for dyn HigherOrderFunctionPlugin {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_tuple("HigherOrderFunctionPlugin")
+            .field(&self.id())
+            .finish()
+    }
+}
+
+impl<V: HigherOrderFunctionVTable> HigherOrderFunctionPlugin for V {
+    fn id(&self) -> HigherOrderFunctionId {
+        V::id(self)
+    }
+
+    fn deserialize(
+        &self,
+        metadata: &[u8],
+        session: &VortexSession,
+    ) -> VortexResult<HigherOrderFunctionRef> {
+        let function = HigherOrderFunctionRef::new(self.clone());
+        function.deserialize(metadata, session)?;
+        Ok(function)
+    }
+}
+
+pub type HigherOrderFunctionPluginRef = Arc<dyn HigherOrderFunctionPlugin>;
diff --git a/vortex-array/src/higher_order_fn/session.rs b/vortex-array/src/higher_order_fn/session.rs
new file mode 100644
index 00000000000..6beffe5c96d
--- /dev/null
+++ b/vortex-array/src/higher_order_fn/session.rs
@@ -0,0 +1,64 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+use std::any::Any;
+use std::sync::Arc;
+
+use vortex_session::ArcSwapMap;
+use vortex_session::SessionExt;
+use vortex_session::SessionGuard;
+use vortex_session::SessionVar;
+use vortex_session::registry::Id;
+
+use crate::higher_order_fn::HigherOrderFunctionPluginRef;
+use crate::higher_order_fn::HigherOrderFunctionVTable;
+use crate::higher_order_fn::fns::list_transform::ListTransform;
+
+/// Registry of higher-order function vtables.
+pub type HigherOrderFunctionRegistry = ArcSwapMap<Id, HigherOrderFunctionPluginRef>;
+
+/// Session state for higher-order function vtables.
+#[derive(Clone, Debug)]
+pub struct HigherOrderFunctionSession {
+    registry: HigherOrderFunctionRegistry,
+}
+
+impl HigherOrderFunctionSession {
+    pub fn registry(&self) -> &HigherOrderFunctionRegistry {
+        &self.registry
+    }
+
+    /// Register a vtable, replacing any existing vtable with the same ID.
+    pub fn register<V: HigherOrderFunctionVTable>(&self, vtable: V) {
+        self.registry.insert(vtable.id(), Arc::new(vtable));
+    }
+}
+
+impl Default for HigherOrderFunctionSession {
+    fn default() -> Self {
+        let this = Self {
+            registry: HigherOrderFunctionRegistry::default(),
+        };
+        this.register(ListTransform);
+        this
+    }
+}
+
+impl SessionVar for HigherOrderFunctionSession {
+    fn as_any(&self) -> &dyn Any {
+        self
+    }
+
+    fn as_any_mut(&mut self) -> &mut dyn Any {
+        self
+    }
+}
+
+/// Extension trait for accessing higher-order-function session state.
+pub trait HigherOrderFunctionSessionExt: SessionExt {
+    fn higher_order_functions(&self) -> SessionGuard<'_, HigherOrderFunctionSession> {
+        self.get::<HigherOrderFunctionSession>()
+    }
+}
+
+impl<S: SessionExt> HigherOrderFunctionSessionExt for S {}
diff --git a/vortex-array/src/higher_order_fn/vtable.rs b/vortex-array/src/higher_order_fn/vtable.rs
new file mode 100644
index 00000000000..5fbd432a4b3
--- /dev/null
+++ b/vortex-array/src/higher_order_fn/vtable.rs
@@ -0,0 +1,94 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+use std::fmt::Debug;
+
+use vortex_error::VortexResult;
+use vortex_error::vortex_bail;
+use vortex_session::VortexSession;
+
+use crate::ArrayRef;
+use crate::dtype::DType;
+use crate::expr::Lambda;
+use crate::expr::Scope;
+use crate::expr::TypedLambda;
+use crate::higher_order_fn::HigherOrderFunctionId;
+use crate::higher_order_fn::LambdaCall;
+use crate::scalar_fn::Arity;
+use crate::scalar_fn::ChildName;
+use crate::validity::Validity;
+
+/// The implementation contract for a higher-order function.
+///
+/// Unlike a [`ScalarFnVTable`](crate::scalar_fn::ScalarFnVTable), a higher-order function owns
+/// one or more lambdas in addition to its ordinary expression children. It determines the lambda
+/// parameter types while binding, then receives the resulting [`TypedLambda`]s at execution.
+pub trait HigherOrderFunctionVTable: 'static + Sized + Clone + Debug + Send + Sync {
+    /// The globally unique identifier for this higher-order function.
+    fn id(&self) -> HigherOrderFunctionId;
+
+    /// The arity of the ordinary expression children.
+    fn arity(&self) -> Arity;
+
+    /// The number of lambda arguments.
+    fn lambda_arity(&self) -> usize;
+
+    /// The name of an ordinary expression child.
+    fn child_name(&self, child_idx: usize) -> ChildName;
+
+    /// Serialize any vtable-level metadata.
+    ///
+    /// The lambda syntax is serialized by [`Expression`](crate::expr::Expression), not here.
+    fn serialize(&self) -> VortexResult<Option<Vec<u8>>> {
+        Ok(None)
+    }
+
+    /// Validate metadata while deserializing an instance of this vtable.
+    fn deserialize(&self, _metadata: &[u8], _session: &VortexSession) -> VortexResult<()> {
+        vortex_bail!("higher-order function {} is not deserializable", self.id());
+    }
+
+    /// Bind `lambdas` after the ordinary children have been bound.
+    ///
+    /// The implementation establishes each lambda's parameter frame and root scope from
+    /// `scope` and `arg_dtypes`.
+    fn bind_lambdas(
+        &self,
+        scope: &Scope,
+        arg_dtypes: &[DType],
+        lambdas: &[Lambda],
+    ) -> VortexResult<Box<[TypedLambda]>>;
+
+    /// Return the dtype after both ordinary children and lambdas have been bound.
+    fn return_dtype(&self, arg_dtypes: &[DType], lambdas: &[TypedLambda]) -> VortexResult<DType>;
+
+    /// Return the result validity without executing the function.
+    ///
+    /// Most functions only know that a nullable result is potentially valid everywhere, but a
+    /// function such as `list_transform` can preserve validity from one of its inputs exactly.
+    fn validity(&self, args: &[ArrayRef], lambdas: &[TypedLambda]) -> VortexResult<Validity> {
+        let arg_dtypes = args
+            .iter()
+            .map(|arg| arg.dtype().clone())
+            .collect::<Vec<_>>();
+        Ok(Validity::from(
+            self.return_dtype(&arg_dtypes, lambdas)?.nullability(),
+        ))
+    }
+
+    /// Execute the function over its already-applied ordinary arguments and closed lambdas.
+    ///
+    /// Lambdas build their bodies lazily when invoked. The returned array is then executed by the
+    /// normal array scheduler, so expression application remains distinct from execution.
+    fn execute(
+        &self,
+        args: &[ArrayRef],
+        lambdas: &[LambdaCall<'_>],
+        ctx: &mut crate::ExecutionCtx,
+    ) -> VortexResult<ArrayRef>;
+
+    /// Whether evaluation of the function can raise a semantic error.
+    fn is_fallible(&self, _lambdas: &[TypedLambda]) -> bool {
+        true
+    }
+}
diff --git a/vortex-array/src/lib.rs b/vortex-array/src/lib.rs
index 9439dc7dd1b..df11b6146a1 100644
--- a/vortex-array/src/lib.rs
+++ b/vortex-array/src/lib.rs
@@ -96,6 +96,7 @@ use vortex_session::registry::Interner;
 
 use crate::aggregate_fn::session::AggregateFnSession;
 use crate::dtype::session::DTypeSession;
+use crate::higher_order_fn::session::HigherOrderFunctionSession;
 use crate::memory::MemorySession;
 use crate::optimizer::kernels::KernelSession;
 use crate::scalar_fn::session::ScalarFnSession;
@@ -122,6 +123,7 @@ pub mod expr;
 mod expression;
 pub mod extension;
 mod hash;
+pub mod higher_order_fn;
 pub mod iter;
 pub mod kernel;
 pub mod mask;
@@ -175,6 +177,7 @@ pub fn array_session() -> VortexSession {
         .with::<KernelSession>()
         .with::<DTypeSession>()
         .with::<ScalarFnSession>()
+        .with::<HigherOrderFunctionSession>()
         .with::<StatsSession>()
         .with::<AggregateFnSession>()
         .with::<MemorySession>()
diff --git a/vortex-proto/proto/expr.proto b/vortex-proto/proto/expr.proto
index 00ffaac433c..482bb204475 100644
--- a/vortex-proto/proto/expr.proto
+++ b/vortex-proto/proto/expr.proto
@@ -125,3 +125,13 @@ message SelectOpts {
 message CaseWhenOpts {
   uint32 num_children = 1;
 }
+
+// Options for `vortex.var`, a reference to a name bound in an enclosing scope.
+message VariableOpts {
+  string name = 1;
+}
+
+// Options for `vortex.lambda`. The body is the expression's single child.
+message LambdaOpts {
+  repeated string params = 1;
+}
diff --git a/vortex-proto/src/generated/vortex.expr.rs b/vortex-proto/src/generated/vortex.expr.rs
index a44328623e3..605f30c07ad 100644
--- a/vortex-proto/src/generated/vortex.expr.rs
+++ b/vortex-proto/src/generated/vortex.expr.rs
@@ -207,3 +207,15 @@ pub struct CaseWhenOpts {
     #[prost(uint32, tag = "1")]
     pub num_children: u32,
 }
+/// Options for `vortex.var`, a reference to a name bound in an enclosing scope.
+#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct VariableOpts {
+    #[prost(string, tag = "1")]
+    pub name: ::prost::alloc::string::String,
+}
+/// Options for `vortex.lambda`. The body is the expression's single child.
+#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct LambdaOpts {
+    #[prost(string, repeated, tag = "1")]
+    pub params: ::prost::alloc::vec::Vec<::prost::alloc::string::String>,
+}