diff --git a/vortex-array/src/expr/analysis/fallible.rs b/vortex-array/src/expr/analysis/fallible.rs
index ff43d51603b..4b39f9c09b5 100644
--- a/vortex-array/src/expr/analysis/fallible.rs
+++ b/vortex-array/src/expr/analysis/fallible.rs
@@ -10,8 +10,11 @@ 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,
+            // These add no fallibility of their own. Note this is the *self* label: a lambda's
+            // body is one of its children, so the folded label at a lambda node is the body's
+            // fallibility. A higher-order function therefore picks the body up through the
+            // ordinary fold instead of walking it by hand.
+            Expression::Root | Expression::Variable(_) | Expression::Lambda(_) => false,
         },
         |acc, &child| acc | child,
     )
@@ -82,3 +85,27 @@ mod tests {
         assert_eq!(labels.get(&expr), Some(&false));
     }
 }
+
+#[cfg(test)]
+mod lambda_tests {
+    use super::*;
+    use crate::expr::checked_add;
+    use crate::expr::lambda;
+    use crate::expr::lit;
+    use crate::expr::var;
+
+    /// A lambda contributes no fallibility of its own, but its body is one of its children, so the
+    /// label at the lambda node is the body's. That is what lets a future higher-order function
+    /// pick the body up through the ordinary fold rather than walking it by hand.
+    #[test]
+    fn a_lambdas_label_is_its_bodys_fallibility() {
+        let fallible = Expression::from(lambda(["x"], checked_add(var("x"), lit(1i32))));
+        assert_eq!(label_is_fallible(&fallible).get(&fallible), Some(&true));
+
+        let infallible = Expression::from(lambda(["x"], var("x")));
+        assert_eq!(
+            label_is_fallible(&infallible).get(&infallible),
+            Some(&false)
+        );
+    }
+}
diff --git a/vortex-array/src/expr/analysis/immediate_access.rs b/vortex-array/src/expr/analysis/immediate_access.rs
index 6c2e4975a92..8f97c0e862a 100644
--- a/vortex-array/src/expr/analysis/immediate_access.rs
+++ b/vortex-array/src/expr/analysis/immediate_access.rs
@@ -67,7 +67,13 @@ 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();
+            // Only the scope root reads every field. A variable resolves against a frame, so it
+            // reads none of them, and saying otherwise would defeat column pruning.
+            return if expr.is_root() {
+                scope.names().iter().cloned().collect()
+            } else {
+                vec![]
+            };
         };
 
         if let Some(selection) = scalar_fn.as_opt::<Select>() {
@@ -87,3 +93,39 @@ pub fn make_bound_free_field_annotator(
         vec![]
     }
 }
+
+#[cfg(test)]
+mod variable_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::lambda;
+    use crate::expr::test_harness::struct_dtype;
+    use crate::expr::var;
+
+    /// Only the scope root reads every field. A variable resolves against a frame, so treating it
+    /// like a root would mark a lambda body as reading the whole struct and defeat column pruning.
+    #[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 = lambda(["x"], var("x")).bind(
+            &Scope::new(scope_dtype.clone()),
+            [DType::Primitive(PType::I32, Nullability::NonNullable)],
+        )?;
+
+        let annotator = make_bound_free_field_annotator(fields);
+        assert!(
+            annotator(bound.body()).is_empty(),
+            "a variable should read no root fields"
+        );
+        Ok(())
+    }
+}
diff --git a/vortex-array/src/expr/analysis/referenced_field_paths.rs b/vortex-array/src/expr/analysis/referenced_field_paths.rs
index 48ef285c20a..7ec5064489c 100644
--- a/vortex-array/src/expr/analysis/referenced_field_paths.rs
+++ b/vortex-array/src/expr/analysis/referenced_field_paths.rs
@@ -84,7 +84,7 @@ impl NodeFolderContext for ReferencedFieldPaths {
         {
             let child_fields = node.children()[0]
                 .dtype()
-                .as_struct_fields_opt()
+                .and_then(|dtype| dtype.as_struct_fields_opt())
                 .ok_or_else(|| vortex_err!("Select child is not a struct"))?;
             let included_fields = selection.normalize_to_included_fields(child_fields.names())?;
 
diff --git a/vortex-array/src/expr/analysis/strict.rs b/vortex-array/src/expr/analysis/strict.rs
index 01cc71d91c3..fda14fd398d 100644
--- a/vortex-array/src/expr/analysis/strict.rs
+++ b/vortex-array/src/expr/analysis/strict.rs
@@ -14,8 +14,11 @@ 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,
+            // Vacuously strict: nullary, so no argument can be null.
+            Expression::Root | Expression::Variable { .. } => true,
+            // A lambda is not a value, so strictness is not a meaningful question about it. Take
+            // the contract's conservative default rather than inventing an answer.
+            Expression::Lambda { .. } => false,
         },
         |acc, &child| acc & child,
     )
diff --git a/vortex-array/src/expr/bound_expression.rs b/vortex-array/src/expr/bound_expression.rs
index 51aa5917dec..eca6182243d 100644
--- a/vortex-array/src/expr/bound_expression.rs
+++ b/vortex-array/src/expr/bound_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 vortex_error::vortex_err;
 
 use crate::dtype::DType;
 use crate::expr::Expression;
 use crate::expr::display::DisplayTreeExpr;
+use crate::expr::lambda::Lambda;
+use crate::expr::scope::Frame;
 use crate::expr::scope::Scope;
+use crate::expr::variable::Variable;
 use crate::scalar_fn::ScalarFnRef;
 
 /// An [`Expression`] that has been type-checked against a [`Scope`].
@@ -45,6 +50,86 @@ pub enum BoundExpression {
         /// 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 variable that was resolved.
+        variable: Variable,
+        /// Index of the frame it resolved in, counted from the outermost. Kept so a capture check
+        /// can compare it against the depth at a binder without redoing resolution.
+        depth: usize,
+    },
+    /// A lambda.
+    ///
+    /// The only variant without a dtype: a lambda is not a value. Its function type is recorded
+    /// structurally on [`BoundLambda`] instead — `param_dtypes` is the argument side and
+    /// [`body_dtype`](BoundLambda::body_dtype) the result side.
+    Lambda(BoundLambda),
+}
+
+/// A bound lambda.
+///
+/// A struct as well as a variant, mirroring [`Expression::Lambda`]: the struct lets an API that
+/// wants a lambda — a higher-order function — demand one in its signature, while the variant keeps
+/// a node for traversal to see.
+#[derive(Clone, Debug, PartialEq, Eq, Hash)]
+pub struct BoundLambda {
+    /// The parameters and their dtypes: the argument side of the function type.
+    ///
+    /// Storing the [`Frame`] the body was bound under, rather than two parallel arrays, makes the
+    /// name/dtype pairing and the duplicate-name rejection structural instead of maintained by
+    /// hand.
+    frame: Frame,
+    body: Arc<BoundExpression>,
+}
+
+impl BoundLambda {
+    /// The frame this lambda's body was bound under.
+    pub fn frame(&self) -> &Frame {
+        &self.frame
+    }
+
+    /// The parameters paired with their dtypes, in declaration order.
+    pub fn bindings(&self) -> &[(Variable, DType)] {
+        self.frame.bindings()
+    }
+
+    /// The variables this lambda binds, in declaration order.
+    ///
+    /// An iterator rather than a slice, because the names and dtypes are interleaved in the frame.
+    pub fn params(&self) -> impl ExactSizeIterator<Item = &Variable> {
+        self.frame.bindings().iter().map(|(variable, _)| variable)
+    }
+
+    /// The dtypes of the parameters, in declaration order.
+    pub fn param_dtypes(&self) -> impl ExactSizeIterator<Item = &DType> {
+        self.frame.bindings().iter().map(|(_, dtype)| dtype)
+    }
+
+    /// The bound body.
+    pub fn body(&self) -> &BoundExpression {
+        &self.body
+    }
+
+    /// Return this lambda with a different body, keeping its parameters and their dtypes.
+    pub fn with_body(&self, body: BoundExpression) -> Self {
+        Self {
+            frame: self.frame.clone(),
+            body: Arc::new(body),
+        }
+    }
+
+    /// The dtype the body evaluates to — the result side of the function type.
+    pub fn body_dtype(&self) -> Option<&DType> {
+        self.body.dtype()
+    }
+}
+
+impl From<BoundLambda> for BoundExpression {
+    fn from(lambda: BoundLambda) -> Self {
+        BoundExpression::Lambda(lambda)
+    }
 }
 
 /// A bound-expression wrapper that compares shared tree identity instead of structure.
@@ -76,8 +161,27 @@ impl PartialEq for ExactBoundExpr {
             }
             // 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::Root { .. }, BoundExpression::Scalar { .. })
-            | (BoundExpression::Scalar { .. }, BoundExpression::Root { .. }) => false,
+            (
+                BoundExpression::Variable {
+                    dtype: lhs_dtype,
+                    variable: lhs_var,
+                    depth: lhs_depth,
+                },
+                BoundExpression::Variable {
+                    dtype: rhs_dtype,
+                    variable: rhs_var,
+                    depth: rhs_depth,
+                },
+            ) => lhs_var == rhs_var && lhs_depth == rhs_depth && lhs_dtype == rhs_dtype,
+            (BoundExpression::Lambda(lhs), BoundExpression::Lambda(rhs)) => {
+                lhs.frame == rhs.frame && Arc::ptr_eq(&lhs.body, &rhs.body)
+            }
+            // 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::Variable { .. }, _)
+            | (BoundExpression::Lambda(_), _) => false,
         }
     }
 }
@@ -90,6 +194,18 @@ impl Hash for ExactBoundExpr {
         // identity-keyed cache lookups from deserializing an entire schema just to compute a hash.
         match &self.0 {
             BoundExpression::Root { .. } => state.write_u8(0),
+            BoundExpression::Variable {
+                variable, depth, ..
+            } => {
+                state.write_u8(2);
+                variable.hash(state);
+                depth.hash(state);
+            }
+            BoundExpression::Lambda(lambda) => {
+                state.write_u8(3);
+                lambda.frame.hash(state);
+                Arc::as_ptr(&lambda.body).hash(state);
+            }
             BoundExpression::Scalar {
                 scalar_fn,
                 children,
@@ -124,8 +240,12 @@ impl BoundExpression {
 
         let arg_dtypes = children
             .iter()
-            .map(|child| child.dtype().clone())
-            .collect_vec();
+            .map(|child| {
+                child.dtype().cloned().ok_or_else(|| {
+                    vortex_err!("a scalar function argument must be a value, got {child}")
+                })
+            })
+            .collect::<VortexResult<Vec<_>>>()?;
         let dtype = scalar_fn.return_dtype(&arg_dtypes)?;
 
         Ok(Self::Scalar {
@@ -141,22 +261,48 @@ impl BoundExpression {
         children: impl IntoIterator<Item = BoundExpression>,
     ) -> VortexResult<Self> {
         let children = Vec::from_iter(children);
-        let BoundExpression::Scalar { scalar_fn, .. } = &self 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::Lambda(lambda) => {
+                vortex_ensure!(
+                    children.len() == 1,
+                    "a lambda expects 1 child but got {}",
+                    children.len()
+                );
+                let body = children
+                    .into_iter()
+                    .next()
+                    .vortex_expect("length checked above");
+                Ok(BoundExpression::Lambda(lambda.with_body(body)))
+            }
+            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 {
+    /// The dtype this expression evaluates to, erroring for a lambda, which is not a value.
+    ///
+    /// Prefer this where the node is a value by construction; it turns what would otherwise be an
+    /// `unwrap` at each call site into one clear error.
+    pub fn value_dtype(&self) -> VortexResult<&DType> {
+        self.dtype()
+            .ok_or_else(|| vortex_err!("expected a value, got a lambda: {self}"))
+    }
+
+    /// The dtype this expression evaluates to, or `None` for a lambda, which is not a value.
+    pub fn dtype(&self) -> Option<&DType> {
         match self {
-            Self::Scalar { dtype, .. } | Self::Root { dtype } => dtype,
+            Self::Scalar { dtype, .. } | Self::Root { dtype } | Self::Variable { dtype, .. } => {
+                Some(dtype)
+            }
+            Self::Lambda(_) => None,
         }
     }
 
@@ -164,7 +310,8 @@ impl BoundExpression {
     pub fn children(&self) -> &[BoundExpression] {
         match self {
             Self::Scalar { children, .. } => children.as_slice(),
-            Self::Root { .. } => &[],
+            Self::Lambda(lambda) => std::slice::from_ref(&lambda.body),
+            Self::Root { .. } | Self::Variable { .. } => &[],
         }
     }
 
@@ -172,7 +319,17 @@ impl BoundExpression {
     pub fn as_scalar(&self) -> Option<&ScalarFnRef> {
         match self {
             Self::Scalar { scalar_fn, .. } => Some(scalar_fn),
-            Self::Root { .. } => None,
+            Self::Root { .. } | Self::Variable { .. } | Self::Lambda(_) => None,
+        }
+    }
+
+    /// The variable and its resolution depth, if this node is a variable reference.
+    pub fn as_variable(&self) -> Option<(&Variable, usize)> {
+        match self {
+            Self::Variable {
+                variable, depth, ..
+            } => Some((variable, *depth)),
+            Self::Scalar { .. } | Self::Root { .. } | Self::Lambda(_) => None,
         }
     }
 
@@ -199,6 +356,20 @@ impl BoundExpression {
         while let Some((node, visited)) = pending.pop() {
             match node {
                 BoundExpression::Root { .. } => expressions.push(crate::expr::root()),
+                BoundExpression::Variable { variable, .. } => {
+                    expressions.push(Expression::Variable(variable.clone()))
+                }
+                BoundExpression::Lambda(lambda) if visited => {
+                    let body = expressions.pop().vortex_expect("body was pushed");
+                    expressions.push(Expression::from(Lambda::new(
+                        lambda.params().cloned(),
+                        body,
+                    )));
+                }
+                BoundExpression::Lambda(lambda) => {
+                    pending.push((node, true));
+                    pending.push((&lambda.body, false));
+                }
                 BoundExpression::Scalar {
                     scalar_fn,
                     children,
@@ -229,6 +400,10 @@ impl Display for BoundExpression {
         match self {
             Self::Scalar { scalar_fn, .. } => scalar_fn.fmt_sql(self, f),
             Self::Root { .. } => f.write_str("$"),
+            Self::Variable { variable, .. } => write!(f, "${variable}"),
+            Self::Lambda(lambda) => {
+                write!(f, "({}) -> {}", lambda.params().join(", "), lambda.body)
+            }
         }
     }
 }
@@ -244,20 +419,78 @@ impl Expression {
     }
 
     /// Bind this expression against an explicit [`Scope`].
+    ///
+    /// Errors on a lambda: a lambda has no dtype, so only whoever supplies its parameter types can
+    /// bind it. See [`Lambda::bind`].
     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 (dtype, depth) = scope.resolve(variable).ok_or_else(|| {
+                    vortex_err!(
+                        "unbound variable '{variable}'; the scope binds {} frame(s)",
+                        scope.depth()
+                    )
+                })?;
+                Ok(BoundExpression::Variable {
+                    dtype: dtype.clone(),
+                    variable: variable.clone(),
+                    depth,
+                })
+            }
+            Expression::Lambda(lambda) => vortex_bail!(
+                "a lambda ({}) is not a value and cannot be bound here; use Lambda::bind, \
+                 which supplies its parameter types",
+                lambda.params().iter().join(", ")
+            ),
+            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)
+            }
         }
+    }
+}
 
-        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)
+impl Lambda {
+    /// Bind this lambda, supplying the dtypes of its parameters.
+    ///
+    /// The parameter dtypes come from the caller because a lambda cannot know them — they are
+    /// determined by whatever applies it. A higher-order function derives them from its own
+    /// arguments and calls this.
+    ///
+    /// The body is bound under `scope` extended with one frame holding the parameters, so a
+    /// parameter shadows an outer binding of the same name.
+    pub fn bind(
+        &self,
+        scope: &Scope,
+        param_dtypes: impl IntoIterator<Item = DType>,
+    ) -> VortexResult<BoundLambda> {
+        let param_dtypes = Vec::from_iter(param_dtypes);
+        vortex_ensure!(
+            param_dtypes.len() == self.params().len(),
+            "lambda binds {} parameter(s) but {} dtype(s) were supplied",
+            self.params().len(),
+            param_dtypes.len()
+        );
+
+        let frame = Frame::try_new(
+            self.params()
+                .iter()
+                .cloned()
+                .zip(param_dtypes.iter().cloned()),
+        )?;
+        let body = self.body().bind_scope(&scope.push_frame(frame.clone()))?;
+
+        Ok(BoundLambda {
+            frame,
+            body: Arc::new(body),
+        })
     }
 }
 
@@ -303,7 +536,7 @@ mod tests {
     fn root_binds_to_the_scope() -> VortexResult<()> {
         let bound = root().bind_scope(&scope())?;
         assert!(bound.is_root());
-        assert_eq!(bound.dtype(), &struct_dtype());
+        assert_eq!(bound.dtype(), Some(&struct_dtype()));
         assert_eq!(bound.unbind(), root());
         Ok(())
     }
@@ -313,14 +546,14 @@ mod tests {
         let expr = eq(col("a"), lit(1_i32));
         let bound = expr.bind_scope(&scope())?;
 
-        assert_eq!(bound.dtype(), &DType::Bool(Nullability::NonNullable));
+        assert_eq!(bound.dtype(), Some(&DType::Bool(Nullability::NonNullable)));
 
         let lhs = &bound.children()[0];
         assert_eq!(
             lhs.dtype(),
-            &DType::Primitive(PType::I32, Nullability::NonNullable)
+            Some(&DType::Primitive(PType::I32, Nullability::NonNullable))
         );
-        assert_eq!(lhs.children()[0].dtype(), &struct_dtype());
+        assert_eq!(lhs.children()[0].dtype(), Some(&struct_dtype()));
         Ok(())
     }
 
@@ -329,7 +562,7 @@ mod tests {
         for expr in [root(), col("a"), eq(col("a"), lit(1_i32)), lit(true)] {
             assert_eq!(
                 expr.bind(&struct_dtype())?.dtype(),
-                &expr.return_dtype(&struct_dtype())?,
+                Some(&expr.return_dtype(&struct_dtype())?),
                 "disagreement for {expr}"
             );
         }
@@ -396,3 +629,247 @@ mod tests {
         assert!(expr.bind_scope(&scope()).is_err());
     }
 }
+
+#[cfg(test)]
+mod lambda_tests {
+    use vortex_error::VortexResult;
+
+    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;
+
+    fn i32_() -> DType {
+        DType::Primitive(PType::I32, Nullability::NonNullable)
+    }
+
+    fn scope() -> Scope {
+        Scope::new(struct_dtype())
+    }
+
+    #[test]
+    fn a_lambda_body_sees_its_parameters() -> VortexResult<()> {
+        let l = lambda(["x"], checked_add(var("x"), lit(1i32)));
+        let bound = l.bind(&scope(), [i32_()])?;
+
+        assert_eq!(
+            bound.param_dtypes().cloned().collect::<Vec<_>>(),
+            vec![i32_()]
+        );
+        assert_eq!(bound.body_dtype(), Some(&i32_()));
+        Ok(())
+    }
+
+    /// A lambda has no dtype, so it cannot be bound as a value.
+    #[test]
+    fn binding_a_bare_lambda_fails() {
+        let l = lambda(["x"], var("x"));
+        assert!(Expression::from(l).bind_scope(&scope()).is_err());
+    }
+
+    /// The same reason: a lambda in an argument position is not a value.
+    #[test]
+    fn a_lambda_in_a_value_position_fails() {
+        let expr = eq(Expression::from(lambda(["x"], var("x"))), lit(1i32));
+        assert!(expr.bind_scope(&scope()).is_err());
+    }
+
+    #[test]
+    fn an_unbound_variable_fails() {
+        assert!(var("nope").bind_scope(&scope()).is_err());
+        // ...and also when it is merely the wrong name inside a lambda.
+        let l = lambda(["x"], var("y"));
+        assert!(l.bind(&scope(), [i32_()]).is_err());
+    }
+
+    #[test]
+    fn parameter_count_must_match_supplied_dtypes() {
+        let l = lambda(["x", "y"], var("x"));
+        assert!(l.bind(&scope(), [i32_()]).is_err());
+    }
+
+    #[test]
+    fn duplicate_parameters_are_rejected() {
+        let l = lambda(["x", "x"], var("x"));
+        assert!(l.bind(&scope(), [i32_(), i32_()]).is_err());
+    }
+
+    /// The body binds under a pushed frame, so a parameter shadows an outer binding, and the
+    /// recorded depth distinguishes the two.
+    #[test]
+    fn an_inner_parameter_shadows_an_outer_binding() -> VortexResult<()> {
+        let outer = scope().push_frame(Frame::try_new([(
+            Variable::new("x"),
+            DType::Utf8(Nullability::NonNullable),
+        )])?);
+
+        let bound = lambda(["x"], var("x")).bind(&outer, [i32_()])?;
+        assert_eq!(
+            bound.body_dtype(),
+            Some(&i32_()),
+            "inner parameter should win"
+        );
+
+        let (_, depth) = bound
+            .body()
+            .as_variable()
+            .vortex_expect("body is a variable");
+        assert_eq!(
+            depth, 1,
+            "resolved in the lambda's own frame, not the outer"
+        );
+        Ok(())
+    }
+
+    /// Root still resolves inside a lambda body: it is a capture, which nothing rejects yet.
+    #[test]
+    fn root_is_still_reachable_from_a_body() -> VortexResult<()> {
+        let bound = lambda(["x"], col("a")).bind(&scope(), [i32_()])?;
+        assert_eq!(bound.body_dtype(), Some(&i32_()));
+        Ok(())
+    }
+
+    #[test]
+    fn a_variable_round_trips_through_unbind() -> VortexResult<()> {
+        let bound = lambda(["x"], var("x")).bind(&scope(), [i32_()])?;
+        assert_eq!(bound.body().unbind(), var("x"));
+        Ok(())
+    }
+
+    /// `ExactBoundExpr` is a `HashMap` key, so `eq` and `hash` must agree. If `eq` falls through
+    /// to a catch-all for a variant that `hash` distinguishes, a key stops being equal to itself
+    /// and lookups miss.
+    #[test]
+    fn a_bound_variable_is_a_usable_map_key() -> VortexResult<()> {
+        use std::collections::hash_map::RandomState;
+        use std::hash::BuildHasher;
+
+        let bound = lambda(["x"], var("x")).bind(&scope(), [i32_()])?;
+        let key = ExactBoundExpr(bound.body().clone());
+        let same = key.clone();
+
+        // `ExactBoundExpr` is a map key, so `eq` and `hash` must agree. A catch-all `eq` arm made
+        // a variable unequal to itself while `hash` still distinguished it, so lookups missed.
+        assert_eq!(key, same, "a variable key must equal itself");
+
+        let state = RandomState::new();
+        assert_eq!(state.hash_one(&key), state.hash_one(&same));
+        Ok(())
+    }
+
+    /// Execution has no variable environment, so a bound variable must error rather than silently
+    /// evaluating as the scope.
+    #[test]
+    fn applying_a_bound_variable_errors() -> VortexResult<()> {
+        use crate::IntoArray;
+        use crate::arrays::PrimitiveArray;
+
+        let array = PrimitiveArray::from_iter([1i32, 2, 3]).into_array();
+        let bound = lambda(["x"], var("x")).bind(&scope(), [i32_()])?;
+
+        assert!(array.apply_bound(bound.body()).is_err());
+        Ok(())
+    }
+
+    /// The unbound path must error too, rather than panicking on the missing scalar fn.
+    #[test]
+    fn applying_an_unbound_variable_or_lambda_errors() {
+        use crate::IntoArray;
+        use crate::arrays::PrimitiveArray;
+
+        let array = PrimitiveArray::from_iter([1i32, 2, 3]).into_array();
+        assert!(array.clone().apply(&var("x")).is_err());
+        assert!(
+            array
+                .apply(&Expression::from(lambda(["x"], var("x"))))
+                .is_err()
+        );
+    }
+
+    /// A deep lambda chain must drop iteratively; recursing through bodies overflows the stack.
+    #[test]
+    fn a_deep_lambda_chain_drops_without_overflowing() {
+        let mut expr = var("x");
+        for _ in 0..100_000 {
+            expr = Expression::from(lambda(["x"], expr));
+        }
+        drop(expr);
+    }
+
+    /// `iter_children` and `children_count` count a lambda's body as a child, so `apply_children`
+    /// and `map_children` must visit it too. When they did not, `LabelingVisitor::visit_up` folded
+    /// over a child that had never been labelled and panicked.
+    #[test]
+    fn labelling_visits_a_bound_lambda_body() -> VortexResult<()> {
+        use crate::expr::analysis::label_tree;
+        use crate::expr::traversal::Node;
+
+        let bound = BoundExpression::from(lambda(["x"], var("x")).bind(&scope(), [i32_()])?);
+        assert_eq!(bound.children_count(), 1, "a lambda's body is a child");
+
+        // Counts nodes, so an unvisited child panics rather than miscounting.
+        let labels = label_tree(&bound, |_| 1usize, |acc, child| acc + child);
+        assert_eq!(labels.get(&bound), Some(&2), "the lambda and its body");
+        Ok(())
+    }
+
+    /// Rewriting must reach the body too, and rebuild the lambda around the result.
+    #[test]
+    fn rewriting_reaches_a_bound_lambda_body() -> VortexResult<()> {
+        use crate::expr::traversal::NodeExt;
+        use crate::expr::traversal::Transformed;
+
+        let bound = BoundExpression::from(lambda(["x"], var("x")).bind(&scope(), [i32_()])?);
+
+        let mut visited = 0;
+        let rewritten = bound.clone().transform_up(|node| {
+            visited += 1;
+            Ok(Transformed::no(node))
+        })?;
+        assert_eq!(visited, 2, "the lambda and its body");
+        assert_eq!(rewritten.into_inner(), bound);
+        Ok(())
+    }
+
+    /// A bound lambda stores the frame its body was bound under, so the name/dtype pairing cannot
+    /// drift the way two parallel arrays could.
+    #[test]
+    fn a_bound_lambda_keeps_the_frame_it_bound_under() -> VortexResult<()> {
+        let bound = lambda(["x"], var("x")).bind(&scope(), [i32_()])?;
+
+        assert_eq!(
+            bound.params().collect::<Vec<_>>(),
+            vec![&Variable::new("x")]
+        );
+        assert_eq!(bound.param_dtypes().collect::<Vec<_>>(), vec![&i32_()]);
+        assert_eq!(bound.frame().get(&Variable::new("x")), Some(&i32_()));
+        Ok(())
+    }
+
+    /// Parameters are named, so equality is structural rather than alpha-equivalence: `x -> x`
+    /// and `y -> y` denote the same function but compare unequal. Nothing depends on the stronger
+    /// property today; pinning it here makes the choice explicit rather than accidental. Switching
+    /// to de Bruijn coordinates would make alpha-equivalence structural.
+    #[test]
+    fn lambdas_are_not_alpha_equivalent() {
+        assert_ne!(lambda(["x"], var("x")), lambda(["y"], var("y")));
+        assert_eq!(lambda(["x"], var("x")), lambda(["x"], var("x")));
+    }
+
+    #[test]
+    fn display_shows_the_binder_and_the_reference() {
+        assert_eq!(var("x").to_string(), "$x");
+        assert_eq!(
+            lambda(["x"], checked_add(var("x"), lit(1i32))).to_string(),
+            "(x) -> ($x + 1i32)"
+        );
+        assert_eq!(root().to_string(), "$");
+    }
+}
diff --git a/vortex-array/src/expr/display.rs b/vortex-array/src/expr/display.rs
index a7642325b67..b1f359c2a1e 100644
--- a/vortex-array/src/expr/display.rs
+++ b/vortex-array/src/expr/display.rs
@@ -5,6 +5,8 @@ use std::fmt;
 use std::fmt::Display;
 use std::fmt::Formatter;
 
+use itertools::Itertools;
+
 use crate::expr::BoundExpression;
 use crate::expr::Expression;
 use crate::scalar_fn::ChildName;
@@ -65,7 +67,10 @@ 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::Lambda { .. } => ChildName::from("body"),
+            Expression::Root | Expression::Variable(_) => {
+                unreachable!("a leaf expression has no children")
+            }
         }
     }
 
@@ -73,6 +78,10 @@ impl DisplayTreeNode for Expression {
         match self {
             Expression::Scalar { scalar_fn, .. } => Display::fmt(scalar_fn, f),
             Expression::Root => write!(f, "{ROOT_DISPLAY}"),
+            Expression::Variable(variable) => write!(f, "${variable}"),
+            Expression::Lambda(lambda) => {
+                write!(f, "lambda({})", lambda.params().iter().join(", "))
+            }
         }
     }
 }
@@ -85,7 +94,10 @@ impl DisplayTreeNode for BoundExpression {
     fn tree_child_name(&self, index: usize) -> ChildName {
         match self {
             BoundExpression::Scalar { scalar_fn, .. } => scalar_fn.signature().child_name(index),
-            BoundExpression::Root { .. } => unreachable!("the scope root has no children"),
+            BoundExpression::Lambda(_) => ChildName::from("body"),
+            BoundExpression::Root { .. } | BoundExpression::Variable { .. } => {
+                unreachable!("a leaf bound node has no children")
+            }
         }
     }
 
@@ -93,6 +105,10 @@ impl DisplayTreeNode for BoundExpression {
         match self {
             BoundExpression::Scalar { scalar_fn, .. } => Display::fmt(scalar_fn, f),
             BoundExpression::Root { .. } => write!(f, "{ROOT_DISPLAY}"),
+            BoundExpression::Variable { variable, .. } => write!(f, "${variable}"),
+            BoundExpression::Lambda(lambda) => {
+                write!(f, "lambda({})", lambda.params().join(", "))
+            }
         }
     }
 }
diff --git a/vortex-array/src/expr/expression.rs b/vortex-array/src/expr/expression.rs
index 44e2381b29b..acf8f4b7258 100644
--- a/vortex-array/src/expr/expression.rs
+++ b/vortex-array/src/expr/expression.rs
@@ -11,13 +11,16 @@ 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_session::VortexSession;
 
 use crate::dtype::DType;
 use crate::expr::display::DisplayTreeExpr;
+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::scalar_fn::ScalarFnRef;
 use crate::scalar_fn::ScalarFnVTable;
 use crate::stats::rewrite::StatsRewriteCtx;
@@ -42,6 +45,18 @@ pub enum Expression {
     },
     /// The full scope of the expression evaluation.
     Root,
+    /// A reference to a name bound by an enclosing [`Expression::Lambda`].
+    ///
+    /// Like [`Expression::Root`], its dtype comes from the scope rather than from children, so it
+    /// is resolved during binding.
+    Variable(Variable),
+    /// A body evaluated under a frame binding `params`.
+    ///
+    /// A lambda is **not a value**: its parameter dtypes are determined by whatever applies it, so
+    /// it has no dtype of its own and cannot be bound by
+    /// [`bind_scope`](Expression::bind_scope). Use
+    /// [`Lambda::bind`], which takes the parameter types.
+    Lambda(Lambda),
 }
 
 impl Expression {
@@ -70,11 +85,27 @@ impl Expression {
         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,
+        }
+    }
+
+    /// The lambda this expression holds, if it is one.
+    pub fn as_lambda(&self) -> Option<&Lambda> {
+        match self {
+            Self::Lambda(lambda) => Some(lambda),
+            _ => None,
+        }
+    }
+
     /// 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,
+            Self::Root | Self::Variable(_) | Self::Lambda(_) => None,
         }
     }
 
@@ -99,10 +130,16 @@ impl Expression {
     }
 
     /// Returns the children of this expression.
+    /// Returns the sub-expressions of this node.
+    ///
+    /// A [`Expression::Lambda`] yields its body, so generic traversal reaches it — but only by
+    /// passing through the `Lambda` node, which is a scope boundary. A pass that is not
+    /// scope-aware must handle that variant rather than descending blindly.
     pub fn children(&self) -> &[Expression] {
         match self {
             Self::Scalar { children, .. } => children.as_slice(),
-            Self::Root => NO_CHILDREN,
+            Self::Lambda(lambda) => std::slice::from_ref(lambda.body_arc()),
+            Self::Root | Self::Variable(_) => NO_CHILDREN,
         }
     }
 
@@ -118,13 +155,24 @@ impl Expression {
     ) -> VortexResult<Self> {
         let children = Vec::from_iter(children);
         match &self {
-            Self::Root => {
+            Self::Root | Self::Variable(_) => {
                 vortex_ensure!(
                     children.is_empty(),
-                    "Expression arity mismatch: root expects 0 children but got {}",
+                    "Expression arity mismatch: {self} expects 0 children but got {}",
+                    children.len()
+                );
+                Ok(self.clone())
+            }
+            Self::Lambda(lambda) => {
+                vortex_ensure!(
+                    children.len() == 1,
+                    "Expression arity mismatch: a lambda expects 1 child but got {}",
                     children.len()
                 );
-                Ok(Self::Root)
+                Ok(Self::Lambda(Lambda::new(
+                    lambda.params().iter().cloned(),
+                    children.into_iter().next().vortex_expect("checked above"),
+                )))
             }
             Self::Scalar { scalar_fn, .. } => {
                 vortex_ensure!(
@@ -145,6 +193,14 @@ impl Expression {
     pub fn return_dtype(&self, scope: &DType) -> VortexResult<DType> {
         match self {
             Self::Root => Ok(scope.clone()),
+            // A variable resolves against a frame, which this entry point does not carry. Erroring
+            // keeps callers that only have a root dtype from silently mistyping a lambda body.
+            Self::Variable(variable) => vortex_bail!(
+                "variable '{variable}' can only be typed by binding against a scope with frames"
+            ),
+            Self::Lambda(_) => {
+                vortex_bail!("a lambda has no data type; use Lambda::bind to type its body")
+            }
             Self::Scalar {
                 scalar_fn,
                 children,
@@ -165,6 +221,11 @@ impl Expression {
         match self {
             // The scope is exactly as valid as itself.
             Self::Root => Ok(Self::Root),
+            // A variable is exactly as valid as whatever it is bound to.
+            Self::Variable(_) => Ok(self.clone()),
+            Self::Lambda(_) => {
+                vortex_bail!("a lambda has no validity; it is not a value")
+            }
             Self::Scalar { scalar_fn, .. } => scalar_fn.validity(self),
         }
     }
@@ -204,6 +265,8 @@ 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::Lambda(lambda) => Display::fmt(lambda, f),
             Self::Scalar { scalar_fn, .. } => scalar_fn.fmt_sql(self, f),
         }
     }
@@ -289,6 +352,14 @@ impl Expression {
     }
 }
 
+/// `Root` stands in as the default so that a body can be moved out during the iterative [`Drop`]
+/// below. It is never observed by callers.
+impl Default for Expression {
+    fn default() -> Self {
+        Self::Root
+    }
+}
+
 /// The default display implementation for expressions uses the 'SQL'-style format.
 impl Display for Expression {
     fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
@@ -299,19 +370,34 @@ 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, .. } => {
+                if let Some(children) = Arc::get_mut(children) {
+                    children_to_drop.append(children);
+                }
+            }
+            Self::Lambda(lambda) => {
+                if let Some(body) = lambda.take_unique_body() {
+                    children_to_drop.push(body);
+                }
+            }
+            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, .. } => {
+                    if let Some(expr_children) = Arc::get_mut(children) {
+                        children_to_drop.append(expr_children);
+                    }
+                }
+                Self::Lambda(lambda) => {
+                    if let Some(body) = lambda.take_unique_body() {
+                        children_to_drop.push(body);
+                    }
+                }
+                Self::Root | Self::Variable(_) => {}
             }
         }
     }
diff --git a/vortex-array/src/expr/exprs.rs b/vortex-array/src/expr/exprs.rs
index d603ec3b52c..5c7d752754c 100644
--- a/vortex-array/src/expr/exprs.rs
+++ b/vortex-array/src/expr/exprs.rs
@@ -16,6 +16,8 @@ use crate::dtype::FieldName;
 use crate::dtype::FieldNames;
 use crate::dtype::Nullability;
 use crate::expr::Expression;
+use crate::expr::lambda::Lambda;
+use crate::expr::variable::Variable;
 use crate::scalar::Scalar;
 use crate::scalar::ScalarValue;
 use crate::scalar_fn::EmptyOptions;
@@ -64,6 +66,23 @@ pub fn root() -> Expression {
     Expression::Root
 }
 
+/// Creates an expression referencing the value bound to `name` in the enclosing scope.
+///
+/// A variable has no dtype of its own; it is resolved when the expression is bound. Binding fails
+/// if no enclosing frame introduces the name.
+pub fn var(name: impl AsRef<str>) -> Expression {
+    Expression::Variable(Variable::new(name))
+}
+
+/// Creates a lambda binding `params` over `body`.
+///
+/// A lambda is not a value: its parameter dtypes are supplied by whatever applies it, so binding
+/// one requires [`Lambda::bind`] rather than
+/// [`bind_scope`](Expression::bind_scope).
+pub fn lambda(params: impl IntoIterator<Item = impl Into<Variable>>, body: Expression) -> Lambda {
+    Lambda::new(params, body)
+}
+
 /// 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..c583b372e82
--- /dev/null
+++ b/vortex-array/src/expr/lambda.rs
@@ -0,0 +1,74 @@
+// 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 crate::expr::Expression;
+use crate::expr::variable::Variable;
+
+/// A body evaluated under a frame binding `params`.
+///
+/// A lambda is **not a value**: its parameter dtypes are determined by whatever applies it, so it
+/// has no dtype of its own and cannot be bound by
+/// [`bind_scope`](Expression::bind_scope). Bind it with [`Lambda::bind`], which takes the parameter
+/// types.
+///
+/// It is a struct rather than only an enum variant so that an API expecting a lambda — a
+/// higher-order function, for instance — can say so in its signature and reject anything else at
+/// compile time. A lambda is still reachable as [`Expression::Lambda`], because traversal needs a
+/// node to see: that node is the scope boundary.
+#[derive(Clone, Debug, PartialEq, Eq, Hash)]
+pub struct Lambda {
+    params: Box<[Variable]>,
+    body: Arc<Expression>,
+}
+
+impl Lambda {
+    /// Create a lambda binding `params` over `body`.
+    pub fn new(params: impl IntoIterator<Item = impl Into<Variable>>, body: Expression) -> Self {
+        Self {
+            params: params.into_iter().map(Into::into).collect(),
+            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 frame.
+    pub fn body(&self) -> &Expression {
+        &self.body
+    }
+
+    /// Take the body if this lambda holds the only reference to it.
+    ///
+    /// Used by `Expression`'s iterative [`Drop`] to drain a lambda chain onto a worklist instead of
+    /// recursing through it, which would overflow the stack on a deeply nested chain.
+    pub(crate) fn take_unique_body(&mut self) -> Option<Expression> {
+        Arc::get_mut(&mut self.body).map(std::mem::take)
+    }
+
+    /// The body as a shared handle, so a caller can hand back a one-element slice.
+    pub(crate) fn body_arc(&self) -> &Arc<Expression> {
+        &self.body
+    }
+}
+
+impl Display for Lambda {
+    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
+        write!(f, "({}) -> {}", self.params.iter().join(", "), self.body)
+    }
+}
+
+impl From<Lambda> for Expression {
+    fn from(lambda: Lambda) -> Self {
+        Expression::Lambda(lambda)
+    }
+}
diff --git a/vortex-array/src/expr/mod.rs b/vortex-array/src/expr/mod.rs
index 5cb0cf12168..f3d2bedc101 100644
--- a/vortex-array/src/expr/mod.rs
+++ b/vortex-array/src/expr/mod.rs
@@ -62,18 +62,22 @@ 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 expression::*;
 pub use exprs::*;
+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
@@ -116,6 +120,8 @@ 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::Lambda(lhs), Expression::Lambda(rhs)) => lhs == rhs,
             (
                 Expression::Scalar {
                     scalar_fn: lhs_fn,
@@ -136,6 +142,14 @@ 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::Lambda(lambda) => {
+                state.write_u8(3);
+                lambda.hash(state);
+            }
             Expression::Scalar {
                 scalar_fn,
                 children,
diff --git a/vortex-array/src/expr/optimize.rs b/vortex-array/src/expr/optimize.rs
index 82579ac564c..8813b572572 100644
--- a/vortex-array/src/expr/optimize.rs
+++ b/vortex-array/src/expr/optimize.rs
@@ -44,7 +44,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),
+            Expression::Root | Expression::Variable { .. } | Expression::Lambda { .. } => Ok(None),
         }
     }
 
@@ -52,7 +52,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),
+            Expression::Root | Expression::Variable { .. } | Expression::Lambda { .. } => Ok(None),
         }
     }
 
@@ -64,7 +64,7 @@ impl Expression {
     ) -> VortexResult<Option<ReduceNodeRef>> {
         match self {
             Expression::Scalar { scalar_fn, .. } => scalar_fn.reduce(node, ctx),
-            Expression::Root => Ok(None),
+            Expression::Root | Expression::Variable { .. } | Expression::Lambda { .. } => Ok(None),
         }
     }
 
@@ -167,6 +167,13 @@ impl Expression {
         let mut current = self.clone();
         let mut any_optimizations = false;
 
+        // A lambda's body types against a parameter frame that this pass does not carry, so
+        // descending would try to resolve a variable against the root dtype and fail. Leave it to
+        // whoever binds the lambda and knows the parameter types.
+        if self.as_lambda().is_some() {
+            return Ok(None);
+        }
+
         // First optimize the root
         if let Some(optimized) = current.clone().try_optimize(scope, cache)? {
             current = optimized;
@@ -408,3 +415,47 @@ mod tests {
         Ok(())
     }
 }
+
+#[cfg(test)]
+mod lambda_tests {
+    use vortex_error::VortexResult;
+
+    use crate::expr::Expression;
+    use crate::expr::col;
+    use crate::expr::fill_null;
+    use crate::expr::lambda;
+    use crate::expr::lit;
+    use crate::expr::test_harness::struct_dtype;
+    use crate::expr::var;
+
+    /// A lambda body types against a parameter frame this pass does not carry, so descending into
+    /// one would try to resolve the variable against the root dtype and error.
+    #[test]
+    fn a_lambda_is_an_optimization_boundary() -> VortexResult<()> {
+        let expr = Expression::from(lambda(["x"], fill_null(var("x"), lit(0_i32))));
+        assert_eq!(expr.clone().optimize_recursive(&struct_dtype())?, expr);
+        Ok(())
+    }
+
+    /// The boundary must not stop optimization of everything around it. `cast` of a literal folds,
+    /// which is a rewrite the optimizer actually performs.
+    #[test]
+    fn optimization_still_applies_outside_a_lambda() -> VortexResult<()> {
+        use crate::dtype::DType;
+        use crate::dtype::Nullability;
+        use crate::dtype::PType;
+        use crate::expr::cast;
+        use crate::expr::lt_eq;
+
+        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..ef828362ce7 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,8 @@ use vortex_proto::expr as pb;
 use vortex_session::VortexSession;
 
 use crate::expr::Expression;
+use crate::expr::Lambda;
+use crate::expr::Variable;
 use crate::scalar_fn::ForeignScalarFnVTable;
 use crate::scalar_fn::ScalarFnId;
 use crate::scalar_fn::session::ScalarFnSessionExt;
@@ -22,14 +25,51 @@ pub trait ExprSerializeProtoExt {
 /// already-serialized expressions keep round-tripping.
 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`]. Its body is the message's single child.
+pub(crate) const LAMBDA_ID: &str = "vortex.lambda";
+
 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::Lambda(lambda) => {
+                return Ok(pb::Expr {
+                    id: LAMBDA_ID.to_string(),
+                    children: vec![lambda.body().serialize_proto()?],
+                    metadata: Some(
+                        pb::LambdaOpts {
+                            params: lambda
+                                .params()
+                                .iter()
+                                .map(|v| v.name().to_string())
+                                .collect(),
+                        }
+                        .encode_to_vec(),
+                    ),
+                });
+            }
+            Expression::Scalar { scalar_fn, .. } => scalar_fn,
         };
 
         let children = self
@@ -56,7 +96,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 +107,30 @@ 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)));
+        }
+
+        if expr.id == LAMBDA_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())?;
+            let body = Expression::from_proto(&expr.children[0], session)?;
+            return Ok(Expression::from(Lambda::new(
+                opts.params.into_iter().map(Variable::new),
+                body,
+            )));
+        }
+
         #[expect(clippy::disallowed_methods, reason = "interning a dynamic id")]
         let expr_id = ScalarFnId::new(expr.id.as_str());
         let children = expr
@@ -98,6 +163,7 @@ pub fn deserialize_expr_proto(
 #[cfg(test)]
 mod tests {
     use prost::Message;
+    use vortex_error::VortexResult;
     use vortex_proto::expr as pb;
     use vortex_session::VortexSession;
 
@@ -141,6 +207,49 @@ mod tests {
         assert_eq!(&deser_expr, &expr);
     }
 
+    /// Variables and lambdas round-trip on reserved ids, like `Root`. The body travels as the
+    /// lambda message's single child, so a generic walk of the wire format still sees it.
+    #[test]
+    fn variables_and_lambdas_round_trip() -> VortexResult<()> {
+        use crate::expr::checked_add;
+        use crate::expr::lambda;
+        use crate::expr::var;
+
+        for expr in [
+            var("x"),
+            Expression::from(lambda(["x"], var("x"))),
+            Expression::from(lambda(
+                ["x", "y"],
+                checked_add(var("x"), checked_add(var("y"), lit(1_i32))),
+            )),
+            // Nested, so the inner lambda travels inside the outer one's body.
+            Expression::from(lambda(["x"], Expression::from(lambda(["y"], var("y"))))),
+        ] {
+            let bytes = expr.serialize_proto()?.encode_to_vec();
+            let decoded =
+                Expression::from_proto(&pb::Expr::decode(bytes.as_slice())?, &array_session())?;
+            assert_eq!(decoded, expr, "round trip changed {expr}");
+        }
+        Ok(())
+    }
+
+    /// 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..069c8dc54a9 100644
--- a/vortex-array/src/expr/scope.rs
+++ b/vortex-array/src/expr/scope.rs
@@ -1,28 +1,102 @@
 // 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 crate::dtype::DType;
+use crate::expr::variable::Variable;
+
+/// 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 bindings = Vec::from_iter(bindings);
+        for (idx, (name, _)) in bindings.iter().enumerate() {
+            if bindings[..idx].iter().any(|(seen, _)| seen == name) {
+                vortex_bail!("duplicate binding '{name}' in a single frame");
+            }
+        }
+        Ok(Self(Arc::new(bindings)))
+    }
+
+    /// The bindings in this frame, in declaration order.
+    pub fn bindings(&self) -> &[(Variable, DType)] {
+        &self.0
+    }
+
+    /// The dtype bound to `name` in this frame, if any.
+    pub fn get(&self, name: &Variable) -> Option<&DType> {
+        self.0
+            .iter()
+            .find_map(|(bound, dtype)| (bound == name).then_some(dtype))
+    }
+}
 
 /// 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 is what distinguishes "this binder's own name" from "a name
+/// introduced by an enclosing binder", which is the information a capture check needs.
 #[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,
+        }
+    }
+
+    /// Resolve `name`, searching innermost-first so that inner bindings shadow outer ones.
+    ///
+    /// Returns the bound dtype and the index of the frame it was found in, counted from the
+    /// outermost so the value stays stable as further frames are pushed. Comparing it against the
+    /// [`depth`](Scope::depth) at a binder distinguishes a reference to that binder's own parameter
+    /// from a capture of something further out.
+    pub fn resolve(&self, name: &Variable) -> Option<(&DType, usize)> {
+        self.frames
+            .iter()
+            .enumerate()
+            .rev()
+            .find_map(|(depth, frame)| frame.get(name).map(|dtype| (dtype, depth)))
+    }
 }
 
 impl From<DType> for Scope {
@@ -33,8 +107,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 +127,52 @@ 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_reports_the_frame_it_was_found_in() -> VortexResult<()> {
+        let scope = Scope::new(i32_())
+            .push_frame(Frame::try_new([(Variable::new("a"), i32_())])?)
+            .push_frame(Frame::try_new([(Variable::new("b"), utf8())])?);
+
+        assert_eq!(scope.resolve(&Variable::new("a")), Some((&i32_(), 0)));
+        assert_eq!(scope.resolve(&Variable::new("b")), Some((&utf8(), 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_eq!(scope.resolve(&Variable::new("x")), Some((&utf8(), 1)));
+        Ok(())
+    }
+
+    /// A name bound twice in one frame is ambiguous, so it is rejected rather than silently
+    /// resolved by position.
+    #[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/bound_partition.rs b/vortex-array/src/expr/transform/bound_partition.rs
index 95fe3f4a7ea..e6ea17cbab6 100644
--- a/vortex-array/src/expr/transform/bound_partition.rs
+++ b/vortex-array/src/expr/transform/bound_partition.rs
@@ -293,7 +293,12 @@ fn partition_root_dtype(names: &FieldNames, partitions: &[BoundExpression]) -> D
             names.clone(),
             partitions
                 .iter()
-                .map(|partition| partition.dtype().clone())
+                .map(|partition| {
+                    partition
+                        .dtype()
+                        .cloned()
+                        .vortex_expect("a partition is always a value")
+                })
                 .collect(),
         ),
         Nullability::NonNullable,
@@ -468,7 +473,7 @@ mod tests {
 
         partitioned.replace_partitions(vec![replacement].into_boxed_slice())?;
 
-        assert_eq!(partitioned.root.dtype(), &field_dtype);
+        assert_eq!(partitioned.root.dtype(), Some(&field_dtype));
         Ok(())
     }
 }
diff --git a/vortex-array/src/expr/transform/coerce.rs b/vortex-array/src/expr/transform/coerce.rs
index 48e52b1b4a9..de35620e20d 100644
--- a/vortex-array/src/expr/transform/coerce.rs
+++ b/vortex-array/src/expr/transform/coerce.rs
@@ -8,7 +8,6 @@ use vortex_error::VortexResult;
 use crate::dtype::DType;
 use crate::expr::Expression;
 use crate::expr::cast;
-use crate::expr::traversal::NodeExt;
 use crate::expr::traversal::Transformed;
 use crate::scalar_fn::fns::literal::Literal;
 
@@ -18,53 +17,85 @@ use crate::scalar_fn::fns::literal::Literal;
 /// The rewrite is bottom-up: children are coerced first, then each parent node checks whether
 /// its children match the coerced argument types.
 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.
-        if node.is_root() || node.is::<Literal>() || node.children().is_empty() {
+    // A lambda is a coercion boundary. Its body types against a parameter frame that this pass
+    // does not carry, so descending into one would try to type a variable against the root dtype
+    // and fail. Leave it for whoever binds the lambda and knows the parameter types. Recursing
+    // explicitly rather than using `transform_up` is what makes skipping the body possible.
+    fn coerce_node(node: Expression, scope: &DType) -> VortexResult<Transformed<Expression>> {
+        if node.as_lambda().is_some() {
             return Ok(Transformed::no(node));
         }
 
-        // Compute the current child return types.
-        let child_dtypes: Vec<DType> = node
-            .children()
-            .iter()
-            .map(|c| c.return_dtype(&scope))
-            .collect::<VortexResult<_>>()?;
+        // Rebuild only when a child actually changed. Rebuilding unconditionally would allocate a
+        // fresh children `Arc` for every node and break the pointer identity `ExactExpr` keys on.
+        let mut changed = false;
+        let mut coerced_children = Vec::with_capacity(node.children().len());
+        for child in node.children() {
+            let coerced = coerce_node(child.clone(), scope)?;
+            changed |= coerced.changed;
+            coerced_children.push(coerced.value);
+        }
 
-        // Ask the scalar function what types it wants.
-        let Some(scalar_fn) = node.as_scalar() else {
-            return Ok(Transformed::no(node));
+        let node = if changed {
+            node.with_children(coerced_children)?
+        } else {
+            node
         };
-        let coerced_dtypes = scalar_fn.coerce_args(&child_dtypes)?;
 
-        // If nothing changed, skip.
-        if child_dtypes == coerced_dtypes {
-            return Ok(Transformed::no(node));
+        let coerced = coerce_one(node, scope)?;
+        Ok(Transformed {
+            changed: changed || coerced.changed,
+            ..coerced
+        })
+    }
+
+    fn coerce_one(node: Expression, scope: &DType) -> VortexResult<Transformed<Expression>> {
+        {
+            // Leaf nodes (Root, Literal) have no children to coerce.
+            if node.is_root() || node.is::<Literal>() || node.children().is_empty() {
+                return Ok(Transformed::no(node));
+            }
+
+            // Compute the current child return types.
+            let child_dtypes: Vec<DType> = node
+                .children()
+                .iter()
+                .map(|c| c.return_dtype(scope))
+                .collect::<VortexResult<_>>()?;
+
+            // Ask the scalar function what types it wants.
+            let Some(scalar_fn) = node.as_scalar() else {
+                return Ok(Transformed::no(node));
+            };
+            let coerced_dtypes = scalar_fn.coerce_args(&child_dtypes)?;
+
+            // If nothing changed, skip.
+            if child_dtypes == coerced_dtypes {
+                return Ok(Transformed::no(node));
+            }
+
+            // Build new children, inserting casts where needed.
+            let new_children: Vec<Expression> = node
+                .children()
+                .iter()
+                .zip(coerced_dtypes.iter())
+                .map(|(child, target)| {
+                    let child_dtype = child.return_dtype(scope)?;
+                    if child_dtype.eq_ignore_nullability(target)
+                        && child_dtype.nullability() == target.nullability()
+                    {
+                        Ok(child.clone())
+                    } else {
+                        Ok(cast(child.clone(), target.clone()))
+                    }
+                })
+                .collect::<VortexResult<_>>()?;
+
+            node.with_children(new_children).map(Transformed::yes)
         }
+    }
 
-        // Build new children, inserting casts where needed.
-        let new_children: Vec<Expression> = node
-            .children()
-            .iter()
-            .zip(coerced_dtypes.iter())
-            .map(|(child, target)| {
-                let child_dtype = child.return_dtype(&scope)?;
-                if child_dtype.eq_ignore_nullability(target)
-                    && child_dtype.nullability() == target.nullability()
-                {
-                    Ok(child.clone())
-                } else {
-                    Ok(cast(child.clone(), target.clone()))
-                }
-            })
-            .collect::<VortexResult<_>>()?;
-
-        let new_expr = node.with_children(new_children)?;
-        Ok(Transformed::yes(new_expr))
-    })
-    .map(|t| t.into_inner())
+    coerce_node(expr, scope).map(Transformed::into_inner)
 }
 
 #[cfg(test)]
@@ -219,3 +250,58 @@ mod tests {
         Ok(())
     }
 }
+
+#[cfg(test)]
+mod lambda_tests {
+    use vortex_error::VortexResult;
+
+    use super::*;
+    use crate::expr::Expression;
+    use crate::expr::checked_add;
+    use crate::expr::col;
+    use crate::expr::lambda;
+    use crate::expr::lit;
+    use crate::expr::test_harness::struct_dtype;
+    use crate::expr::var;
+
+    /// A lambda body types against a parameter frame, which this pass does not carry. Descending
+    /// into one would try to type the variable against the root dtype and fail, so a lambda is a
+    /// coercion boundary and is returned untouched.
+    #[test]
+    fn a_lambda_is_a_coercion_boundary() -> VortexResult<()> {
+        let l = Expression::from(lambda(["x"], checked_add(var("x"), lit(1i32))));
+        assert_eq!(coerce_expression(l.clone(), &struct_dtype())?, l);
+        Ok(())
+    }
+
+    /// Coercion that changes nothing must return the original tree, not a rebuilt copy: rebuilding
+    /// allocates a fresh children `Arc` and breaks the pointer identity `ExactExpr` keys on.
+    #[test]
+    fn a_no_op_coercion_preserves_identity() -> VortexResult<()> {
+        use crate::expr::ExactExpr;
+
+        // Already well-typed, so nothing should be coerced.
+        let expr = checked_add(col("a"), lit(1i32));
+        let coerced = coerce_expression(expr.clone(), &struct_dtype())?;
+
+        assert_eq!(coerced, expr);
+        assert_eq!(
+            ExactExpr(coerced),
+            ExactExpr(expr),
+            "an unchanged tree should keep its identity"
+        );
+        Ok(())
+    }
+
+    /// The boundary must not stop coercion of everything around it.
+    #[test]
+    fn coercion_still_applies_outside_a_lambda() -> VortexResult<()> {
+        let expr = checked_add(col("a"), lit(1i64));
+        let coerced = coerce_expression(expr.clone(), &struct_dtype())?;
+        assert_ne!(
+            coerced, expr,
+            "an i32 column against an i64 literal should coerce"
+        );
+        Ok(())
+    }
+}
diff --git a/vortex-array/src/expr/traversal/mod.rs b/vortex-array/src/expr/traversal/mod.rs
index 952a73f2657..603e9fbb8c0 100644
--- a/vortex-array/src/expr/traversal/mod.rs
+++ b/vortex-array/src/expr/traversal/mod.rs
@@ -533,8 +533,15 @@ impl Node for BoundExpression {
         &'a self,
         mut f: F,
     ) -> VortexResult<TraversalOrder> {
-        let BoundExpression::Scalar { children, .. } = self else {
-            return Ok(TraversalOrder::Continue);
+        // Must stay consistent with `iter_children`/`children_count` below: a lambda's body is a
+        // child, and `LabelingVisitor::visit_up` folds over `iter_children` expecting every one to
+        // already have a label.
+        let children: &[Self] = match self {
+            BoundExpression::Scalar { children, .. } => children,
+            BoundExpression::Lambda(lambda) => std::slice::from_ref(lambda.body()),
+            BoundExpression::Root { .. } | BoundExpression::Variable { .. } => {
+                return Ok(TraversalOrder::Continue);
+            }
         };
 
         for child in children.iter() {
@@ -551,8 +558,12 @@ impl Node for BoundExpression {
         self,
         mut f: F,
     ) -> VortexResult<Transformed<Self>> {
-        let BoundExpression::Scalar { children, .. } = &self else {
-            return Ok(Transformed::no(self));
+        let children: &[Self] = match &self {
+            BoundExpression::Scalar { children, .. } => children,
+            BoundExpression::Lambda(lambda) => std::slice::from_ref(lambda.body()),
+            BoundExpression::Root { .. } | BoundExpression::Variable { .. } => {
+                return Ok(Transformed::no(self));
+            }
         };
 
         let mut order = TraversalOrder::Continue;
@@ -584,14 +595,18 @@ impl Node for BoundExpression {
     fn iter_children<T>(&self, f: impl FnOnce(&mut dyn Iterator<Item = &Self>) -> T) -> T {
         match self {
             BoundExpression::Scalar { children, .. } => f(&mut children.iter()),
-            BoundExpression::Root { .. } => f(&mut std::iter::empty()),
+            BoundExpression::Lambda(lambda) => f(&mut std::iter::once(lambda.body())),
+            BoundExpression::Root { .. } | BoundExpression::Variable { .. } => {
+                f(&mut std::iter::empty())
+            }
         }
     }
 
     fn children_count(&self) -> usize {
         match self {
             BoundExpression::Scalar { children, .. } => children.len(),
-            BoundExpression::Root { .. } => 0,
+            BoundExpression::Lambda(_) => 1,
+            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..631ed75212f
--- /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;
+
+/// The name of a value bound in a [`Scope`](crate::expr::Scope).
+///
+/// Deliberately distinct from [`FieldName`](crate::dtype::FieldName): a variable and a struct field
+/// live in different namespaces, and giving them the same type would let one be passed where the
+/// other is meant.
+///
+/// The same type is used in both positions a variable appears — as a parameter of
+/// [`Expression::Lambda`](crate::expr::Expression::Lambda), which introduces the name, and inside
+/// [`Expression::Variable`](crate::expr::Expression::Variable), which references it.
+#[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<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 d0590f2bf58..4c24a2a71b0 100644
--- a/vortex-array/src/expression.rs
+++ b/vortex-array/src/expression.rs
@@ -2,8 +2,8 @@
 // SPDX-FileCopyrightText: Copyright the Vortex contributors
 
 use itertools::Itertools;
-use vortex_error::VortexExpect;
 use vortex_error::VortexResult;
+use vortex_error::vortex_bail;
 
 use crate::ArrayRef;
 use crate::IntoArray;
@@ -17,13 +17,22 @@ 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 BoundExpression::Scalar {
-            scalar_fn,
-            children,
-            ..
-        } = expr
-        else {
-            return Ok(self);
+        let (scalar_fn, children) = match expr {
+            // Root evaluates to the scope, which is this array.
+            BoundExpression::Root { .. } => return Ok(self),
+            // Execution has no variable environment, so a variable cannot be evaluated. It must be
+            // substituted by whatever bound it before the tree reaches the array layer.
+            BoundExpression::Variable { variable, .. } => vortex_bail!(
+                "cannot evaluate variable '{variable}': execution has no variable environment"
+            ),
+            BoundExpression::Lambda(lambda) => {
+                vortex_bail!("cannot evaluate a lambda ({lambda:?}); it must be applied first")
+            }
+            BoundExpression::Scalar {
+                scalar_fn,
+                children,
+                ..
+            } => (scalar_fn, children),
         };
 
         if let Some(scalar) = scalar_fn.as_opt::<Literal>() {
@@ -43,10 +52,17 @@ impl ArrayRef {
 
     /// 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);
-        }
+        let scalar_fn = match expr {
+            // Root evaluates to the scope, which is this array.
+            Expression::Root => return Ok(self),
+            Expression::Variable(variable) => vortex_bail!(
+                "cannot evaluate variable '{variable}': execution has no variable environment"
+            ),
+            Expression::Lambda(lambda) => {
+                vortex_bail!("cannot evaluate a lambda ({lambda}); it must be applied first")
+            }
+            Expression::Scalar { scalar_fn, .. } => scalar_fn,
+        };
 
         // Manually convert literals to ConstantArray.
         if let Some(scalar) = expr.as_opt::<Literal>() {
@@ -59,11 +75,6 @@ impl ArrayRef {
             .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();
 
diff --git a/vortex-layout/src/layouts/chunked/reader.rs b/vortex-layout/src/layouts/chunked/reader.rs
index 949ae24d01a..05fa4e75b95 100644
--- a/vortex-layout/src/layouts/chunked/reader.rs
+++ b/vortex-layout/src/layouts/chunked/reader.rs
@@ -302,7 +302,9 @@ impl LayoutReader for ChunkedReader {
         mask: MaskFuture,
     ) -> VortexResult<BoxFuture<'static, VortexResult<ArrayRef>>> {
         if row_range.is_empty() {
-            return Ok(future::ready(Ok(Canonical::empty(expr.dtype()).into_array())).boxed());
+            return Ok(
+                future::ready(Ok(Canonical::empty(expr.value_dtype()?).into_array())).boxed(),
+            );
         }
 
         let mut chunk_evals = vec![];
diff --git a/vortex-layout/src/layouts/list/reader.rs b/vortex-layout/src/layouts/list/reader.rs
index 53227635de6..5a601dde139 100644
--- a/vortex-layout/src/layouts/list/reader.rs
+++ b/vortex-layout/src/layouts/list/reader.rs
@@ -210,7 +210,7 @@ impl ListReader {
     ) -> VortexResult<ArrayFuture> {
         // Crop to the smallest contiguous row range containing every selected list.
         let Some(selected_rows) = selected_row_range(&mask) else {
-            let empty = Canonical::empty(expr.dtype()).into_array();
+            let empty = Canonical::empty(expr.value_dtype()?).into_array();
             return Ok(async move { Ok(empty) }.boxed());
         };
 
diff --git a/vortex-layout/src/layouts/partitioned.rs b/vortex-layout/src/layouts/partitioned.rs
index 94d1c3f782d..f95c4913cf9 100644
--- a/vortex-layout/src/layouts/partitioned.rs
+++ b/vortex-layout/src/layouts/partitioned.rs
@@ -54,7 +54,7 @@ impl<P: Send + Sync + 'static> BoundPartitionedExprEval<P> for BoundPartitionedE
             .zip_eq(self.partitions.iter())
             .map(|(annotation, expr)| {
                 Ok::<_, VortexError>(
-                    if matches!(expr.dtype(), DType::Bool(Nullability::NonNullable)) {
+                    if matches!(expr.dtype(), Some(DType::Bool(Nullability::NonNullable))) {
                         // If the partition evaluates to a boolean, we can evaluate it as a mask which
                         // can often be more efficient since nulls are turned into `false` early on,
                         // and layouts can perform predicate pruning / indexing.
diff --git a/vortex-layout/src/scan/filter.rs b/vortex-layout/src/scan/filter.rs
index 29b9165741f..f7abff6ceda 100644
--- a/vortex-layout/src/scan/filter.rs
+++ b/vortex-layout/src/scan/filter.rs
@@ -186,12 +186,12 @@ mod tests {
         assert_eq!(
             conjuncts
                 .iter()
-                .map(|expr| expr.dtype().clone())
+                .map(|expr| expr.dtype().cloned())
                 .collect::<Vec<_>>(),
             vec![
-                DType::Bool(Nullability::Nullable),
-                DType::Bool(Nullability::Nullable),
-                DType::Bool(Nullability::NonNullable),
+                Some(DType::Bool(Nullability::Nullable)),
+                Some(DType::Bool(Nullability::Nullable)),
+                Some(DType::Bool(Nullability::NonNullable)),
             ]
         );
         Ok(())
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>,
+}