diff --git a/vortex-layout/src/plan/mod.rs b/vortex-layout/src/plan/mod.rs
index 06f7acd0031..802e210b07e 100644
--- a/vortex-layout/src/plan/mod.rs
+++ b/vortex-layout/src/plan/mod.rs
@@ -11,6 +11,7 @@ mod children;
 mod display;
 mod lower;
 mod optimize;
+pub mod optimizer;
 mod plans;
 mod typed;
 mod vtable;
@@ -38,12 +39,13 @@ pub use plans::PackPlan;
 pub use plans::RowIdx;
 pub use plans::RowIdxData;
 pub use plans::RowIdxPlan;
-pub use plans::RowIdxPlanMetadata;
 pub use plans::SegmentScan;
 pub use plans::SegmentScanData;
 pub use plans::SegmentScanPlan;
 pub use plans::Take;
 pub use plans::TakePlan;
+pub use plans::plan_row_idx_expression;
+pub use plans::row_idx_dtype;
 pub use typed::DynPlan;
 pub use typed::Plan;
 pub use typed::PlanParts;
diff --git a/vortex-layout/src/plan/optimize.rs b/vortex-layout/src/plan/optimize.rs
index d78798cdb65..f471b7744f9 100644
--- a/vortex-layout/src/plan/optimize.rs
+++ b/vortex-layout/src/plan/optimize.rs
@@ -1,15 +1,35 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright the Vortex contributors
 
-//! Generic bottom-up optimization over physical plans.
+//! Plan optimization.
+//!
+//! The optimizer applies static rewrites top-down, optimizes children, then retries rewrites
+//! exposed by the optimized children.
 
 use vortex_error::VortexResult;
 
-use crate::plan::Eval;
 use crate::plan::PlanRef;
+use crate::plan::optimizer::reduce_parent;
+use crate::plan::optimizer::reduce_plan;
+
+fn reduce(plan: &PlanRef) -> VortexResult<Option<PlanRef>> {
+    if let Some(rewritten) = reduce_plan(plan)? {
+        return Ok(Some(rewritten));
+    }
+    for child_idx in 0..plan.child_count() {
+        if let Some(rewritten) = reduce_parent(plan, child_idx)? {
+            return Ok(Some(rewritten));
+        }
+    }
+    Ok(None)
+}
 
 /// Optimizes `plan`, preserving its dtype and row domain.
 pub fn optimize(plan: PlanRef) -> VortexResult<PlanRef> {
+    if let Some(rewritten) = reduce(&plan)? {
+        return optimize(rewritten);
+    }
+
     let mut children = Vec::with_capacity(plan.child_count());
     let mut changed = false;
     for child in plan.children().iter() {
@@ -19,17 +39,13 @@ pub fn optimize(plan: PlanRef) -> VortexResult<PlanRef> {
         children.push(optimized);
     }
 
-    let plan = if changed {
-        plan.with_children(children)?
-    } else {
-        plan
-    };
-
-    let Some(eval) = plan.as_opt::<Eval>() else {
+    if !changed {
         return Ok(plan);
-    };
-    if eval.expression().is_root() {
-        return eval.child_plan();
+    }
+
+    let plan = plan.with_children(children)?;
+    if let Some(rewritten) = reduce(&plan)? {
+        return optimize(rewritten);
     }
     Ok(plan)
 }
diff --git a/vortex-layout/src/plan/optimizer/mod.rs b/vortex-layout/src/plan/optimizer/mod.rs
new file mode 100644
index 00000000000..a63006be62e
--- /dev/null
+++ b/vortex-layout/src/plan/optimizer/mod.rs
@@ -0,0 +1,57 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+//! Static rewrite rules for physical plans.
+
+mod rules;
+
+pub use rules::DynPlanParentReduceRule;
+pub use rules::DynPlanReduceRule;
+pub use rules::PlanParentReduceRule;
+pub use rules::PlanParentReduceRuleAdapter;
+pub use rules::PlanParentRuleSet;
+pub use rules::PlanReduceRule;
+pub use rules::PlanReduceRuleAdapter;
+pub use rules::PlanRuleSet;
+use vortex_error::VortexResult;
+
+use super::Concat;
+use super::Eval;
+use super::Pack;
+use super::PlanRef;
+use super::Take;
+use super::plans::EvalIdentityRule;
+use super::plans::ExpressionConcatRule;
+use super::plans::ExpressionPackRule;
+use super::plans::ExpressionTakeRule;
+
+static EVAL_IDENTITY_RULE: PlanReduceRuleAdapter<Eval, EvalIdentityRule> =
+    PlanReduceRuleAdapter::new(EvalIdentityRule);
+
+static PLAN_RULES: PlanRuleSet = PlanRuleSet::new(&[&EVAL_IDENTITY_RULE]);
+
+static EXPRESSION_CONCAT_RULE: PlanParentReduceRuleAdapter<Concat, ExpressionConcatRule> =
+    PlanParentReduceRuleAdapter::new(ExpressionConcatRule);
+static EXPRESSION_TAKE_RULE: PlanParentReduceRuleAdapter<Take, ExpressionTakeRule> =
+    PlanParentReduceRuleAdapter::new(ExpressionTakeRule);
+static EXPRESSION_PACK_RULE: PlanParentReduceRuleAdapter<Pack, ExpressionPackRule> =
+    PlanParentReduceRuleAdapter::new(ExpressionPackRule);
+
+static PARENT_RULES: PlanParentRuleSet = PlanParentRuleSet::new(&[
+    &EXPRESSION_CONCAT_RULE,
+    &EXPRESSION_TAKE_RULE,
+    &EXPRESSION_PACK_RULE,
+]);
+
+/// Attempts a static rewrite for `plan`.
+pub(crate) fn reduce_plan(plan: &PlanRef) -> VortexResult<Option<PlanRef>> {
+    PLAN_RULES.evaluate(plan)
+}
+
+/// Attempts a static rewrite for `parent` and its child at `child_idx`.
+pub(crate) fn reduce_parent(parent: &PlanRef, child_idx: usize) -> VortexResult<Option<PlanRef>> {
+    let Some(child) = parent.child(child_idx)? else {
+        return Ok(None);
+    };
+    PARENT_RULES.evaluate(&child, parent, child_idx)
+}
diff --git a/vortex-layout/src/plan/optimizer/rules.rs b/vortex-layout/src/plan/optimizer/rules.rs
new file mode 100644
index 00000000000..3276aab22f6
--- /dev/null
+++ b/vortex-layout/src/plan/optimizer/rules.rs
@@ -0,0 +1,257 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+//! Typed and type-erased interfaces for plan rewrites.
+
+use std::any::type_name;
+use std::fmt::Debug;
+use std::marker::PhantomData;
+
+use vortex_error::VortexResult;
+
+use crate::plan::Plan;
+use crate::plan::PlanRef;
+use crate::plan::PlanVTable;
+
+/// A rewrite over one concrete plan operator.
+///
+/// Rules return one rewrite without recursively optimizing the replacement. The plan optimizer
+/// owns traversal and drives further rewrites.
+pub trait PlanReduceRule<P: PlanVTable>: Debug + Send + Sync + 'static {
+    /// Attempts to replace `plan`.
+    fn reduce(&self, plan: &Plan<P>) -> VortexResult<Option<PlanRef>>;
+}
+
+/// Type-erased interface used by [`PlanRuleSet`].
+pub trait DynPlanReduceRule: Debug + Send + Sync + 'static {
+    /// Returns whether this rule supports the concrete plan operator.
+    fn matches(&self, plan: &PlanRef) -> bool;
+
+    /// Attempts to replace `plan`.
+    fn reduce(&self, plan: &PlanRef) -> VortexResult<Option<PlanRef>>;
+}
+
+/// Bridges a typed [`PlanReduceRule`] to a type-erased static registry.
+pub struct PlanReduceRuleAdapter<P, R> {
+    rule: R,
+    _plan: PhantomData<fn() -> P>,
+}
+
+impl<P, R> PlanReduceRuleAdapter<P, R> {
+    /// Creates an adapter for a typed plan rule.
+    pub const fn new(rule: R) -> Self {
+        Self {
+            rule,
+            _plan: PhantomData,
+        }
+    }
+}
+
+impl<P, R> Debug for PlanReduceRuleAdapter<P, R>
+where
+    P: PlanVTable,
+    R: PlanReduceRule<P>,
+{
+    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        formatter
+            .debug_struct("PlanReduceRuleAdapter")
+            .field("plan", &type_name::<P>())
+            .field("rule", &self.rule)
+            .finish()
+    }
+}
+
+impl<P, R> DynPlanReduceRule for PlanReduceRuleAdapter<P, R>
+where
+    P: PlanVTable,
+    R: PlanReduceRule<P>,
+{
+    fn matches(&self, plan: &PlanRef) -> bool {
+        plan.is::<P>()
+    }
+
+    fn reduce(&self, plan: &PlanRef) -> VortexResult<Option<PlanRef>> {
+        let Some(plan) = plan.as_opt::<P>() else {
+            return Ok(None);
+        };
+        self.rule.reduce(plan)
+    }
+}
+
+/// An ordered static collection of single-plan rewrite rules.
+pub struct PlanRuleSet {
+    rules: &'static [&'static dyn DynPlanReduceRule],
+}
+
+impl PlanRuleSet {
+    /// Creates a rule set whose first successful rewrite wins.
+    pub const fn new(rules: &'static [&'static dyn DynPlanReduceRule]) -> Self {
+        Self { rules }
+    }
+
+    /// Evaluates rules registered for the concrete plan operator.
+    pub fn evaluate(&self, plan: &PlanRef) -> VortexResult<Option<PlanRef>> {
+        for rule in self.rules {
+            if !rule.matches(plan) {
+                continue;
+            }
+            let Some(reduced) = rule.reduce(plan)? else {
+                continue;
+            };
+
+            #[cfg(debug_assertions)]
+            {
+                vortex_error::vortex_ensure!(
+                    reduced.row_count() == plan.row_count(),
+                    "Plan rewrite from {rule:?} changed row count from {} to {}",
+                    plan.row_count(),
+                    reduced.row_count()
+                );
+                vortex_error::vortex_ensure!(
+                    reduced.dtype() == plan.dtype(),
+                    "Plan rewrite from {rule:?} changed dtype from {} to {}",
+                    plan.dtype(),
+                    reduced.dtype()
+                );
+            }
+
+            return Ok(Some(reduced));
+        }
+        Ok(None)
+    }
+}
+
+/// A metadata-only rewrite where a child plan rewrites its parent plan.
+///
+/// Rules return one rewrite without recursively optimizing the replacement. The plan optimizer
+/// owns traversal and drives further rewrites.
+pub trait PlanParentReduceRule<C: PlanVTable>: Debug + Send + Sync + 'static {
+    /// The concrete parent operator matched by this rule.
+    type Parent: PlanVTable;
+
+    /// Attempts to replace `parent` based on its child at `child_idx`.
+    fn reduce_parent(
+        &self,
+        child: &Plan<C>,
+        parent: &Plan<Self::Parent>,
+        child_idx: usize,
+    ) -> VortexResult<Option<PlanRef>>;
+}
+
+/// Type-erased interface used by [`PlanParentRuleSet`].
+pub trait DynPlanParentReduceRule: Debug + Send + Sync + 'static {
+    /// Returns whether this rule supports the concrete child and parent operators.
+    fn matches(&self, child: &PlanRef, parent: &PlanRef) -> bool;
+
+    /// Attempts to replace `parent` based on `child` at `child_idx`.
+    fn reduce_parent(
+        &self,
+        child: &PlanRef,
+        parent: &PlanRef,
+        child_idx: usize,
+    ) -> VortexResult<Option<PlanRef>>;
+}
+
+/// Bridges a typed [`PlanParentReduceRule`] to a type-erased static registry.
+pub struct PlanParentReduceRuleAdapter<C, R> {
+    rule: R,
+    _child: PhantomData<fn() -> C>,
+}
+
+impl<C, R> PlanParentReduceRuleAdapter<C, R> {
+    /// Creates an adapter for a typed parent-child rule.
+    pub const fn new(rule: R) -> Self {
+        Self {
+            rule,
+            _child: PhantomData,
+        }
+    }
+}
+
+impl<C, R> Debug for PlanParentReduceRuleAdapter<C, R>
+where
+    C: PlanVTable,
+    R: PlanParentReduceRule<C>,
+{
+    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        formatter
+            .debug_struct("PlanParentReduceRuleAdapter")
+            .field("parent", &type_name::<R::Parent>())
+            .field("child", &type_name::<C>())
+            .field("rule", &self.rule)
+            .finish()
+    }
+}
+
+impl<C, R> DynPlanParentReduceRule for PlanParentReduceRuleAdapter<C, R>
+where
+    C: PlanVTable,
+    R: PlanParentReduceRule<C>,
+{
+    fn matches(&self, child: &PlanRef, parent: &PlanRef) -> bool {
+        child.is::<C>() && parent.is::<R::Parent>()
+    }
+
+    fn reduce_parent(
+        &self,
+        child: &PlanRef,
+        parent: &PlanRef,
+        child_idx: usize,
+    ) -> VortexResult<Option<PlanRef>> {
+        let Some(child) = child.as_opt::<C>() else {
+            return Ok(None);
+        };
+        let Some(parent) = parent.as_opt::<R::Parent>() else {
+            return Ok(None);
+        };
+        self.rule.reduce_parent(child, parent, child_idx)
+    }
+}
+
+/// An ordered static collection of parent-child plan rewrite rules.
+pub struct PlanParentRuleSet {
+    rules: &'static [&'static dyn DynPlanParentReduceRule],
+}
+
+impl PlanParentRuleSet {
+    /// Creates a rule set whose first successful rewrite wins.
+    pub const fn new(rules: &'static [&'static dyn DynPlanParentReduceRule]) -> Self {
+        Self { rules }
+    }
+
+    /// Evaluates rules registered for the concrete `(parent, child)` pair.
+    pub fn evaluate(
+        &self,
+        child: &PlanRef,
+        parent: &PlanRef,
+        child_idx: usize,
+    ) -> VortexResult<Option<PlanRef>> {
+        for rule in self.rules {
+            if !rule.matches(child, parent) {
+                continue;
+            }
+            let Some(reduced) = rule.reduce_parent(child, parent, child_idx)? else {
+                continue;
+            };
+
+            #[cfg(debug_assertions)]
+            {
+                vortex_error::vortex_ensure!(
+                    reduced.row_count() == parent.row_count(),
+                    "Plan rewrite from {rule:?} changed row count from {} to {}",
+                    parent.row_count(),
+                    reduced.row_count()
+                );
+                vortex_error::vortex_ensure!(
+                    reduced.dtype() == parent.dtype(),
+                    "Plan rewrite from {rule:?} changed dtype from {} to {}",
+                    parent.dtype(),
+                    reduced.dtype()
+                );
+            }
+
+            return Ok(Some(reduced));
+        }
+        Ok(None)
+    }
+}
diff --git a/vortex-layout/src/plan/plans/concat.rs b/vortex-layout/src/plan/plans/concat.rs
index 2e2893fde5f..2e814156309 100644
--- a/vortex-layout/src/plan/plans/concat.rs
+++ b/vortex-layout/src/plan/plans/concat.rs
@@ -10,12 +10,15 @@ use vortex_error::VortexResult;
 use vortex_error::vortex_bail;
 use vortex_session::registry::CachedId;
 
+use crate::plan::Eval;
+use crate::plan::EvalPlan;
 use crate::plan::Plan;
 use crate::plan::PlanChildren;
 use crate::plan::PlanId;
 use crate::plan::PlanParts;
 use crate::plan::PlanRef;
 use crate::plan::PlanVTable;
+use crate::plan::optimizer::PlanParentReduceRule;
 
 /// Concatenates its children row-wise.
 #[derive(Clone, Debug)]
@@ -140,3 +143,28 @@ impl PlanVTable for Concat {
         Cow::Owned(format!("chunks[{index}]"))
     }
 }
+
+/// Pushes an expression into every chunk of a [`Concat`].
+#[derive(Debug)]
+pub(crate) struct ExpressionConcatRule;
+
+impl PlanParentReduceRule<Concat> for ExpressionConcatRule {
+    type Parent = Eval;
+
+    fn reduce_parent(
+        &self,
+        child: &Plan<Concat>,
+        parent: &Plan<Eval>,
+        _child_idx: usize,
+    ) -> VortexResult<Option<PlanRef>> {
+        let expression = parent.expression();
+        let chunks = child
+            .children()
+            .iter()
+            .map(|chunk| Ok(EvalPlan::try_new(expression.clone(), chunk?)?.into_plan()))
+            .collect::<VortexResult<Vec<_>>>()?;
+        Ok(Some(
+            ConcatPlan::try_new(expression.dtype().clone(), chunks)?.into_plan(),
+        ))
+    }
+}
diff --git a/vortex-layout/src/plan/plans/eval.rs b/vortex-layout/src/plan/plans/eval.rs
index 70a645da687..f1edb967e41 100644
--- a/vortex-layout/src/plan/plans/eval.rs
+++ b/vortex-layout/src/plan/plans/eval.rs
@@ -17,6 +17,7 @@ use crate::plan::PlanParts;
 use crate::plan::PlanRef;
 use crate::plan::PlanVTable;
 use crate::plan::check_child_count;
+use crate::plan::optimizer::PlanReduceRule;
 
 /// Applies an expression to the output of its child.
 #[derive(Clone, Debug)]
@@ -123,3 +124,17 @@ fn validate_expression_child(expression: &BoundExpression, child: &PlanRef) -> V
     }
     Ok(())
 }
+
+/// Removes an [`Eval`] whose expression is the identity expression.
+#[derive(Debug)]
+pub(crate) struct EvalIdentityRule;
+
+impl PlanReduceRule<Eval> for EvalIdentityRule {
+    fn reduce(&self, plan: &Plan<Eval>) -> VortexResult<Option<PlanRef>> {
+        if plan.expression().is_root() {
+            Ok(Some(plan.child_plan()?))
+        } else {
+            Ok(None)
+        }
+    }
+}
diff --git a/vortex-layout/src/plan/plans/mod.rs b/vortex-layout/src/plan/plans/mod.rs
index 2e4a6dbad5e..7b6a1893d0b 100644
--- a/vortex-layout/src/plan/plans/mod.rs
+++ b/vortex-layout/src/plan/plans/mod.rs
@@ -2,7 +2,7 @@
 // SPDX-FileCopyrightText: Copyright the Vortex contributors
 
 mod concat;
-mod eval;
+pub(crate) mod eval;
 mod list_pack;
 mod pack;
 mod row_idx;
@@ -12,21 +12,26 @@ mod take;
 pub use concat::Concat;
 pub use concat::ConcatData;
 pub use concat::ConcatPlan;
+pub(crate) use concat::ExpressionConcatRule;
 pub use eval::Eval;
 pub use eval::EvalData;
+pub(crate) use eval::EvalIdentityRule;
 pub use eval::EvalPlan;
 pub use list_pack::ListPack;
 pub use list_pack::ListPackData;
 pub use list_pack::ListPackPlan;
+pub(crate) use pack::ExpressionPackRule;
 pub use pack::Pack;
 pub use pack::PackData;
 pub use pack::PackPlan;
 pub use row_idx::RowIdx;
 pub use row_idx::RowIdxData;
 pub use row_idx::RowIdxPlan;
-pub use row_idx::RowIdxPlanMetadata;
+pub use row_idx::plan_row_idx_expression;
+pub use row_idx::row_idx_dtype;
 pub use segment_scan::SegmentScan;
 pub use segment_scan::SegmentScanData;
 pub use segment_scan::SegmentScanPlan;
+pub(crate) use take::ExpressionTakeRule;
 pub use take::Take;
 pub use take::TakePlan;
diff --git a/vortex-layout/src/plan/plans/pack.rs b/vortex-layout/src/plan/plans/pack.rs
index 9cf47ebe4b7..6497cc63d45 100644
--- a/vortex-layout/src/plan/plans/pack.rs
+++ b/vortex-layout/src/plan/plans/pack.rs
@@ -5,20 +5,39 @@ use std::borrow::Cow;
 
 use vortex_array::EmptyMetadata;
 use vortex_array::dtype::DType;
+use vortex_array::dtype::FieldName;
+use vortex_array::dtype::FieldNames;
 use vortex_array::dtype::Nullability;
 use vortex_array::dtype::StructFields;
+use vortex_array::expr::BoundExpression;
+use vortex_array::expr::ExactBoundExpr;
+use vortex_array::expr::descendent_bound_annotations;
+use vortex_array::expr::make_bound_free_field_annotator;
+use vortex_array::expr::transform::partition_bound;
+use vortex_array::expr::traversal::NodeExt;
+use vortex_array::expr::traversal::Transformed;
+use vortex_array::expr::traversal::TraversalOrder;
+use vortex_array::scalar_fn::ScalarFnVTableExt;
+use vortex_array::scalar_fn::fns::get_item::GetItem;
+use vortex_array::scalar_fn::fns::pack::Pack as PackFn;
+use vortex_array::scalar_fn::fns::pack::PackOptions;
+use vortex_array::scalar_fn::fns::select::Select;
 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::registry::CachedId;
 
+use crate::plan::Eval;
+use crate::plan::EvalPlan;
 use crate::plan::Plan;
 use crate::plan::PlanChildren;
 use crate::plan::PlanId;
 use crate::plan::PlanParts;
 use crate::plan::PlanRef;
 use crate::plan::PlanVTable;
+use crate::plan::optimizer::PlanParentReduceRule;
 
 /// Assembles a struct from one child per field, plus an optional trailing validity child.
 #[derive(Clone, Debug)]
@@ -224,3 +243,339 @@ fn validate_validity_child(expected_row_count: u64, child: &PlanRef) -> VortexRe
     }
     Ok(())
 }
+
+impl PackPlan {
+    /// Rebuilds this plan with only `fields`, which must be a subset of the current fields.
+    ///
+    /// Pruning is only sound for a non-nullable struct: dropping a field of a nullable struct
+    /// would drop the validity child that the remaining fields depend on.
+    pub(crate) fn with_pruned_fields(
+        &self,
+        fields: Vec<(FieldName, PlanRef)>,
+    ) -> VortexResult<Self> {
+        vortex_ensure!(
+            !self.dtype().is_nullable(),
+            "Cannot prune fields from a nullable Pack"
+        );
+        let struct_fields = StructFields::from_iter(
+            fields
+                .iter()
+                .map(|(name, plan)| (name.clone(), plan.dtype().clone())),
+        );
+        let field_plans = fields.into_iter().map(|(_, plan)| plan).collect::<Vec<_>>();
+        PackPlan::try_new(
+            struct_fields,
+            Nullability::NonNullable,
+            self.row_count(),
+            field_plans,
+            None,
+        )
+    }
+}
+
+/// Pushes an expression into the referenced fields of a [`Pack`], pruning the rest.
+#[derive(Debug)]
+pub(crate) struct ExpressionPackRule;
+
+impl PlanParentReduceRule<Pack> for ExpressionPackRule {
+    type Parent = Eval;
+
+    fn reduce_parent(
+        &self,
+        child: &Plan<Pack>,
+        parent: &Plan<Eval>,
+        _child_idx: usize,
+    ) -> VortexResult<Option<PlanRef>> {
+        if child.dtype().is_nullable() {
+            return Ok(None);
+        }
+
+        let expression = parent.expression();
+        let fields = child.fields();
+        let referenced_fields =
+            descendent_bound_annotations(expression, make_bound_free_field_annotator(fields))
+                .get(&ExactBoundExpr(expression.clone()))
+                .vortex_expect("Bound expression missing free-field annotations")
+                .clone();
+        let expanded_root = expanded_struct_root(child.dtype(), fields)?;
+        let expanded = expand_struct_root(expression.clone(), &expanded_root, fields)?;
+        let partitioned =
+            partition_bound(expanded.clone(), make_bound_free_field_annotator(fields))?;
+
+        if partitioned.partition_names.is_empty() {
+            let selected_indices = fields
+                .names()
+                .iter()
+                .enumerate()
+                .filter_map(|(index, name)| referenced_fields.contains(name).then_some(index))
+                .collect::<Vec<_>>();
+            if selected_indices.len() == fields.nfields() {
+                return Ok(None);
+            }
+
+            let pruned_fields = selected_indices
+                .into_iter()
+                .map(|field_index| {
+                    Ok((
+                        field_name(fields, field_index)?,
+                        field_plan(child, field_index)?,
+                    ))
+                })
+                .collect::<VortexResult<Vec<_>>>()?;
+            let rewritten = child.with_pruned_fields(pruned_fields)?.into_plan();
+            return Ok(Some(
+                EvalPlan::try_new(expression.clone(), rewritten)?.into_plan(),
+            ));
+        }
+
+        if partitioned.partition_names.len() == 1 {
+            let name = partitioned
+                .partition_names
+                .get(0)
+                .ok_or_else(|| vortex_err!("Struct expression partition has no field"))?;
+            let index = fields.find(name).ok_or_else(|| {
+                vortex_err!("Struct expression references unknown field '{name}'")
+            })?;
+            let field = field_plan(child, index)?;
+            let lowered = step_into_struct_field(expanded, name, field.dtype().clone())?;
+            return Ok(Some(EvalPlan::try_new(lowered, field)?.into_plan()));
+        }
+
+        let residual = partitioned.root;
+        let mut collapsed = Vec::with_capacity(partitioned.partitions.len());
+        let mut field_expressions = vec![None; fields.nfields()];
+        for index in 0..partitioned.partitions.len() {
+            let name = &partitioned.partition_names[index];
+            let partition = &partitioned.partitions[index];
+            let field_index = fields.find(name).ok_or_else(|| {
+                vortex_err!("Struct expression references unknown field '{name}'")
+            })?;
+            let field = field_plan(child, field_index)?;
+            let lowered = if let Some(pack) = partition
+                .as_scalar()
+                .and_then(|scalar_fn| scalar_fn.as_opt::<PackFn>())
+                && partition.children().len() == 1
+            {
+                let value_name = pack
+                    .names
+                    .get(0)
+                    .ok_or_else(|| vortex_err!("Struct expression partition pack is empty"))?;
+                collapsed.push((name.clone(), value_name.clone()));
+                partition.children()[0].clone()
+            } else {
+                partition.clone()
+            };
+            let lowered = step_into_struct_field(lowered, name, field.dtype().clone())?;
+            field_expressions[field_index] = Some(lowered);
+        }
+
+        let mut fields_changed = partitioned.partition_names.len() != fields.nfields();
+        let mut pruned_fields = Vec::with_capacity(partitioned.partition_names.len());
+        for (field_index, expression) in field_expressions.into_iter().enumerate() {
+            let Some(expression) = expression else {
+                continue;
+            };
+            let field = field_plan(child, field_index)?;
+            let field = if is_identity_expression(&expression, field.dtype())? {
+                field
+            } else {
+                fields_changed = true;
+                EvalPlan::try_new(expression, field)?.into_plan()
+            };
+            pruned_fields.push((field_name(fields, field_index)?, field));
+        }
+        let rewritten = if fields_changed {
+            child.with_pruned_fields(pruned_fields)?.into_plan()
+        } else {
+            child.to_plan()
+        };
+        let residual = rewrite_partition_root(residual, rewritten.dtype().clone(), &collapsed)?;
+
+        if !fields_changed && residual == *expression {
+            return Ok(None);
+        }
+
+        Ok(Some(EvalPlan::try_new(residual, rewritten)?.into_plan()))
+    }
+}
+
+/// Rebinds a partitioned residual expression after collapsing single-value partitions.
+///
+/// # Arguments
+///
+/// * `expression` - The residual recombination expression returned by `partition_bound`.
+/// * `root_dtype` - The dtype produced by the rewritten plan and used to rebind every root.
+/// * `collapsed` - `(partition_name, value_name)` pairs whose one-field `Pack` was removed;
+///   each `$.partition_name.value_name` access is rewritten to `$.partition_name`.
+pub(super) fn rewrite_partition_root(
+    expression: BoundExpression,
+    root_dtype: DType,
+    collapsed: &[(FieldName, FieldName)],
+) -> VortexResult<BoundExpression> {
+    Ok(expression
+        .transform_down(|node| {
+            if let Some(value_name) = node.as_opt::<GetItem>() {
+                let partition_access = &node.children()[0];
+                if let Some(partition_name) = partition_access.as_opt::<GetItem>()
+                    && partition_access.children()[0].is_root()
+                    && collapsed.iter().any(|(partition, value)| {
+                        partition == partition_name && value == value_name
+                    })
+                {
+                    return Ok(Transformed {
+                        value: BoundExpression::try_new(
+                            GetItem.bind(partition_name.clone()),
+                            [BoundExpression::new_root(root_dtype.clone())],
+                        )?,
+                        changed: true,
+                        order: TraversalOrder::Skip,
+                    });
+                }
+            }
+
+            if node.is_root() {
+                Ok(Transformed {
+                    value: BoundExpression::new_root(root_dtype.clone()),
+                    changed: true,
+                    order: TraversalOrder::Skip,
+                })
+            } else {
+                Ok(Transformed::no(node))
+            }
+        })?
+        .into_inner())
+}
+
+fn field_name(fields: &StructFields, index: usize) -> VortexResult<FieldName> {
+    Ok(fields
+        .field_name(index)
+        .ok_or_else(|| vortex_err!("Struct field {index} has no name"))?
+        .clone())
+}
+
+fn field_plan(plan: &Plan<Pack>, index: usize) -> VortexResult<PlanRef> {
+    plan.child(index)?
+        .ok_or_else(|| vortex_err!("Struct field {index} has no plan"))
+}
+
+fn expanded_struct_root(
+    root_dtype: &DType,
+    fields: &StructFields,
+) -> VortexResult<BoundExpression> {
+    let root = BoundExpression::new_root(root_dtype.clone());
+    let children = fields
+        .names()
+        .iter()
+        .map(|name| BoundExpression::try_new(GetItem.bind(name.clone()), [root.clone()]))
+        .collect::<VortexResult<Vec<_>>>()?;
+    bound_pack(fields.names().clone(), children)
+}
+
+fn is_identity_expression(expression: &BoundExpression, input_dtype: &DType) -> VortexResult<bool> {
+    if expression.is_root() {
+        return Ok(expression.dtype() == input_dtype);
+    }
+    if input_dtype.is_nullable() {
+        return Ok(false);
+    }
+    let Some(fields) = input_dtype.as_struct_fields_opt() else {
+        return Ok(false);
+    };
+    Ok(expression == &expanded_struct_root(input_dtype, fields)?)
+}
+
+fn expand_struct_root(
+    expression: BoundExpression,
+    expanded_root: &BoundExpression,
+    fields: &StructFields,
+) -> VortexResult<BoundExpression> {
+    Ok(expression
+        .transform_down(|node| {
+            if node.is_root() {
+                return Ok(Transformed {
+                    value: expanded_root.clone(),
+                    changed: true,
+                    order: TraversalOrder::Skip,
+                });
+            }
+
+            let Some(scalar_fn) = node.as_scalar() else {
+                return Ok(Transformed::no(node));
+            };
+            if !node
+                .children()
+                .first()
+                .is_some_and(BoundExpression::is_root)
+            {
+                return Ok(Transformed::no(node));
+            }
+
+            if let Some(field_name) = scalar_fn.as_opt::<GetItem>() {
+                let index = fields.find(field_name).ok_or_else(|| {
+                    vortex_err!("Field {field_name} not found while expanding struct root")
+                })?;
+                return Ok(Transformed {
+                    value: expanded_root.children()[index].clone(),
+                    changed: true,
+                    order: TraversalOrder::Skip,
+                });
+            }
+
+            if let Some(selection) = scalar_fn.as_opt::<Select>() {
+                let names = selection.normalize_to_included_fields(fields.names())?;
+                let children = names
+                    .iter()
+                    .map(|name| {
+                        let index = fields
+                            .find(name)
+                            .vortex_expect("normalized selection fields must exist in the root");
+                        expanded_root.children()[index].clone()
+                    })
+                    .collect();
+                return Ok(Transformed {
+                    value: bound_pack(names, children)?,
+                    changed: true,
+                    order: TraversalOrder::Skip,
+                });
+            }
+
+            Ok(Transformed::no(node))
+        })?
+        .into_inner())
+}
+
+fn step_into_struct_field(
+    expression: BoundExpression,
+    field_name: &FieldName,
+    field_dtype: DType,
+) -> VortexResult<BoundExpression> {
+    Ok(expression
+        .transform_down(|node| {
+            let is_field_access = node
+                .as_scalar()
+                .and_then(|scalar_fn| scalar_fn.as_opt::<GetItem>())
+                .is_some_and(|name| name == field_name)
+                && node.children()[0].is_root();
+
+            if is_field_access {
+                Ok(Transformed {
+                    value: BoundExpression::new_root(field_dtype.clone()),
+                    changed: true,
+                    order: TraversalOrder::Skip,
+                })
+            } else {
+                Ok(Transformed::no(node))
+            }
+        })?
+        .into_inner())
+}
+
+fn bound_pack(names: FieldNames, children: Vec<BoundExpression>) -> VortexResult<BoundExpression> {
+    BoundExpression::try_new(
+        PackFn.bind(PackOptions {
+            names,
+            nullability: Nullability::NonNullable,
+        }),
+        children,
+    )
+}
diff --git a/vortex-layout/src/plan/plans/row_idx.rs b/vortex-layout/src/plan/plans/row_idx.rs
index 99d8f2f963d..893980825d3 100644
--- a/vortex-layout/src/plan/plans/row_idx.rs
+++ b/vortex-layout/src/plan/plans/row_idx.rs
@@ -1,12 +1,29 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright the Vortex contributors
 
-use std::borrow::Cow;
+use std::fmt::Display;
+use std::fmt::Formatter;
 
-use vortex_array::ProstMetadata;
+use vortex_array::EmptyMetadata;
+use vortex_array::dtype::DType;
+use vortex_array::dtype::FieldName;
+use vortex_array::dtype::Nullability;
+use vortex_array::dtype::PType;
+use vortex_array::dtype::StructFields;
+use vortex_array::expr::BoundExpression;
+use vortex_array::expr::transform::partition_bound;
+use vortex_array::expr::traversal::NodeExt;
+use vortex_array::expr::traversal::Transformed;
+use vortex_array::expr::traversal::TraversalOrder;
+use vortex_array::scalar_fn::fns::pack::Pack as PackFn;
 use vortex_error::VortexResult;
+use vortex_error::vortex_ensure;
+use vortex_error::vortex_err;
 use vortex_session::registry::CachedId;
 
+use crate::layouts::row_idx::RowIdx as RowIdxFn;
+use crate::plan::EvalPlan;
+use crate::plan::PackPlan;
 use crate::plan::Plan;
 use crate::plan::PlanChildren;
 use crate::plan::PlanId;
@@ -14,93 +31,226 @@ use crate::plan::PlanParts;
 use crate::plan::PlanRef;
 use crate::plan::PlanVTable;
 use crate::plan::check_child_count;
+use crate::plan::plans::pack::rewrite_partition_root;
 
-/// Adds row-index support to its child, offsetting row numbers into the file's row domain.
+const ROW_IDX_PARTITION_NAME: &str = "row_idx";
+const CHILD_PARTITION_NAME: &str = "child";
+
+/// Generates global row indices for the current execution row domain.
 #[derive(Clone, Debug)]
 pub struct RowIdx;
 
-/// The row offset applied to the child domain.
+/// Operator-specific data for a [`RowIdx`] plan.
 #[derive(Clone, Debug)]
-pub struct RowIdxData {
-    row_offset: u64,
-}
+pub struct RowIdxData;
 
-/// A plan that adds row-index support to its child.
+/// A childless source of global row indices supplied by its execution row domain.
 pub type RowIdxPlan = Plan<RowIdx>;
 
 impl RowIdxPlan {
-    /// Creates a row-index plan with `row_offset` applied to its child domain.
-    pub fn new(row_offset: u64, child: PlanRef) -> Self {
+    /// Creates a row-index source covering `row_count` rows.
+    pub fn new(row_count: u64) -> Self {
         PlanParts {
             vtable: RowIdx,
-            dtype: child.dtype().clone(),
-            row_count: child.row_count(),
-            children: vec![child].into(),
-            data: RowIdxData { row_offset },
+            dtype: row_idx_dtype(),
+            row_count,
+            children: PlanChildren::default(),
+            data: RowIdxData,
         }
         .into_typed()
     }
+}
 
-    /// Returns the row offset applied to the child domain.
-    pub fn row_offset(&self) -> u64 {
-        self.data().row_offset
-    }
-
-    /// Returns the child plan.
-    pub fn child_plan(&self) -> VortexResult<PlanRef> {
-        self.child_required(0)
-    }
+/// Returns the dtype of a generated row index.
+pub fn row_idx_dtype() -> DType {
+    DType::Primitive(PType::U64, Nullability::NonNullable)
 }
 
 impl PlanVTable for RowIdx {
     type PlanData = RowIdxData;
-    type Metadata = ProstMetadata<RowIdxPlanMetadata>;
+    type Metadata = EmptyMetadata;
 
     fn id(&self) -> PlanId {
         static ID: CachedId = CachedId::new("vortex.plan.row_idx");
         *ID
     }
 
-    fn metadata(plan: &Plan<Self>) -> Option<Self::Metadata> {
-        Some(ProstMetadata(RowIdxPlanMetadata {
-            row_offset: plan.data().row_offset,
-        }))
+    fn metadata(_plan: &Plan<Self>) -> Option<Self::Metadata> {
+        Some(EmptyMetadata)
     }
 
     fn with_children(
-        plan: &Plan<Self>,
+        _plan: &Plan<Self>,
         children: &PlanChildren,
         _data: &mut Self::PlanData,
     ) -> VortexResult<()> {
-        check_child_count("RowIdx", children, 1)?;
-        let child = children
-            .get(0)?
-            .ok_or_else(|| vortex_error::vortex_err!("RowIdx child is absent"))?;
-        if child.dtype() != plan.dtype() || child.row_count() != plan.row_count() {
-            vortex_error::vortex_bail!(
-                "RowIdx child shape changed from ({}, {}) to ({}, {})",
-                plan.dtype(),
-                plan.row_count(),
-                child.dtype(),
-                child.row_count()
-            );
-        }
-        Ok(())
+        check_child_count("RowIdx", children, 0)
     }
+}
 
-    fn child_name(_plan: &Plan<Self>, index: usize) -> Cow<'_, str> {
-        if index == 0 {
-            Cow::Borrowed("child")
+/// Plans an expression over a data source and its global row-index domain.
+///
+/// Expressions that only use data bypass the row-index source, expressions that only use
+/// `#row_idx` bypass `child`, and mixed expressions combine independently planned branches with a
+/// [`PackPlan`]. The file row offset is supplied by the execution row domain.
+pub fn plan_row_idx_expression(
+    expression: BoundExpression,
+    child: PlanRef,
+) -> VortexResult<PlanRef> {
+    let partitioned = partition_bound(expression.clone(), |node| {
+        if node
+            .as_scalar()
+            .is_some_and(|scalar_fn| scalar_fn.is::<RowIdxFn>())
+        {
+            vec![RowIdxExpressionPartition::RowIdx]
+        } else if node.is_root() {
+            vec![RowIdxExpressionPartition::Child]
         } else {
-            Cow::Owned(format!("child[{index}]"))
+            vec![]
+        }
+    })?;
+
+    if partitioned.partition_annotations.is_empty() {
+        let row_domain = PackPlan::try_new(
+            StructFields::empty(),
+            Nullability::NonNullable,
+            child.row_count(),
+            Vec::new(),
+            None,
+        )?
+        .into_plan();
+        return Ok(EvalPlan::try_new(expression, row_domain)?.into_plan());
+    }
+
+    if partitioned.partition_annotations.len() == 1 {
+        return match partitioned.partition_annotations[0] {
+            RowIdxExpressionPartition::RowIdx => {
+                let expression = replace_row_idx(expression)?;
+                let values = RowIdxPlan::new(child.row_count()).into_plan();
+                Ok(EvalPlan::try_new(expression, values)?.into_plan())
+            }
+            RowIdxExpressionPartition::Child => {
+                Ok(EvalPlan::try_new(expression, child)?.into_plan())
+            }
+        };
+    }
+
+    vortex_ensure!(
+        partitioned.partition_annotations.len() == 2,
+        "Row-index expression produced more than two partitions"
+    );
+    let row_idx_index = partitioned
+        .partition_annotations
+        .iter()
+        .position(|partition| *partition == RowIdxExpressionPartition::RowIdx)
+        .ok_or_else(|| vortex_err!("Row-index expression has no row-index partition"))?;
+    let child_index = partitioned
+        .partition_annotations
+        .iter()
+        .position(|partition| *partition == RowIdxExpressionPartition::Child)
+        .ok_or_else(|| vortex_err!("Row-index expression has no data partition"))?;
+
+    let row_idx_partition = &partitioned.partitions[row_idx_index];
+    let child_partition = &partitioned.partitions[child_index];
+    let (Some(row_idx_pack), Some(child_pack)) = (
+        row_idx_partition
+            .as_scalar()
+            .and_then(|scalar_fn| scalar_fn.as_opt::<PackFn>()),
+        child_partition
+            .as_scalar()
+            .and_then(|scalar_fn| scalar_fn.as_opt::<PackFn>()),
+    ) else {
+        return Err(vortex_err!(
+            "Row-index expression partitions must be struct packs"
+        ));
+    };
+    let row_idx_partition_name = partitioned.partition_names[row_idx_index].clone();
+    let child_partition_name = partitioned.partition_names[child_index].clone();
+    let mut collapsed = Vec::with_capacity(2);
+
+    let row_idx_expression = if row_idx_partition.children().len() == 1 {
+        let Some(value_name) = row_idx_pack.names.get(0) else {
+            return Err(vortex_err!("Row-index expression partition is empty"));
+        };
+        collapsed.push((row_idx_partition_name, value_name.clone()));
+        row_idx_partition.children()[0].clone()
+    } else {
+        row_idx_partition.clone()
+    };
+    let child_expression = if child_partition.children().len() == 1 {
+        let Some(value_name) = child_pack.names.get(0) else {
+            return Err(vortex_err!("Data expression partition is empty"));
+        };
+        collapsed.push((child_partition_name, value_name.clone()));
+        child_partition.children()[0].clone()
+    } else {
+        child_partition.clone()
+    };
+
+    let row_count = child.row_count();
+    let row_idx_expression = replace_row_idx(row_idx_expression)?;
+    let row_idx_plan =
+        EvalPlan::try_new(row_idx_expression, RowIdxPlan::new(row_count).into_plan())?.into_plan();
+    let child_plan = EvalPlan::try_new(child_expression, child)?.into_plan();
+    let fields = StructFields::from_iter([
+        (ROW_IDX_PARTITION_NAME, row_idx_plan.dtype().clone()),
+        (CHILD_PARTITION_NAME, child_plan.dtype().clone()),
+    ]);
+    let partitions = PackPlan::try_new(
+        fields,
+        Nullability::NonNullable,
+        row_count,
+        vec![row_idx_plan, child_plan],
+        None,
+    )?;
+    let residual =
+        rewrite_partition_root(partitioned.root, partitions.dtype().clone(), &collapsed)?;
+
+    Ok(EvalPlan::try_new(residual, partitions.into_plan())?.into_plan())
+}
+
+fn replace_row_idx(expression: BoundExpression) -> VortexResult<BoundExpression> {
+    Ok(expression
+        .transform_down(|node| {
+            if node
+                .as_scalar()
+                .is_some_and(|scalar_fn| scalar_fn.is::<RowIdxFn>())
+            {
+                Ok(Transformed {
+                    value: BoundExpression::new_root(row_idx_dtype()),
+                    changed: true,
+                    order: TraversalOrder::Skip,
+                })
+            } else {
+                Ok(Transformed::no(node))
+            }
+        })?
+        .into_inner())
+}
+
+#[derive(Clone, Copy, PartialEq, Eq, Hash)]
+enum RowIdxExpressionPartition {
+    RowIdx,
+    Child,
+}
+
+impl RowIdxExpressionPartition {
+    fn name(self) -> &'static str {
+        match self {
+            Self::RowIdx => ROW_IDX_PARTITION_NAME,
+            Self::Child => CHILD_PARTITION_NAME,
         }
     }
 }
 
-/// Serialized metadata for a [`RowIdx`] plan.
-#[derive(Clone, PartialEq, Eq, ::prost::Message)]
-pub struct RowIdxPlanMetadata {
-    /// The row offset applied to the child domain.
-    #[prost(uint64, tag = "1")]
-    pub row_offset: u64,
+impl Display for RowIdxExpressionPartition {
+    fn fmt(&self, formatter: &mut Formatter<'_>) -> std::fmt::Result {
+        formatter.write_str(self.name())
+    }
+}
+
+impl From<RowIdxExpressionPartition> for FieldName {
+    fn from(partition: RowIdxExpressionPartition) -> Self {
+        FieldName::from(partition.name())
+    }
 }
diff --git a/vortex-layout/src/plan/plans/take.rs b/vortex-layout/src/plan/plans/take.rs
index 1e68281029b..af9eca55da1 100644
--- a/vortex-layout/src/plan/plans/take.rs
+++ b/vortex-layout/src/plan/plans/take.rs
@@ -5,9 +5,13 @@ use std::borrow::Cow;
 
 use vortex_array::EmptyMetadata;
 use vortex_array::dtype::DType;
+use vortex_array::expr::ExactBoundExpr;
+use vortex_array::expr::label_bound_tree;
 use vortex_error::VortexResult;
 use vortex_session::registry::CachedId;
 
+use crate::plan::Eval;
+use crate::plan::EvalPlan;
 use crate::plan::Plan;
 use crate::plan::PlanChildren;
 use crate::plan::PlanId;
@@ -15,6 +19,7 @@ use crate::plan::PlanParts;
 use crate::plan::PlanRef;
 use crate::plan::PlanVTable;
 use crate::plan::check_child_count;
+use crate::plan::optimizer::PlanParentReduceRule;
 
 const CODES: usize = 0;
 const VALUES: usize = 1;
@@ -113,3 +118,47 @@ impl PlanVTable for Take {
         }
     }
 }
+
+/// Pushes a strict, infallible boolean expression onto the dictionary values of a [`Take`].
+#[derive(Debug)]
+pub(crate) struct ExpressionTakeRule;
+
+impl PlanParentReduceRule<Take> for ExpressionTakeRule {
+    type Parent = Eval;
+
+    fn reduce_parent(
+        &self,
+        child: &Plan<Take>,
+        parent: &Plan<Eval>,
+        _child_idx: usize,
+    ) -> VortexResult<Option<PlanRef>> {
+        let expression = parent.expression();
+        if !expression.dtype().is_boolean() {
+            return Ok(None);
+        }
+        // Evaluating over values rather than codes is only sound when the expression reads the
+        // root, is strict, and cannot fail: otherwise per-row behaviour is not preserved.
+        let labels = label_bound_tree(
+            expression,
+            |node| match node.as_scalar() {
+                Some(scalar_fn) => (
+                    false,
+                    scalar_fn.signature().is_strict(),
+                    scalar_fn.signature().is_fallible(),
+                ),
+                None => (true, true, false),
+            },
+            |acc, &child| (acc.0 | child.0, acc.1 & child.1, acc.2 | child.2),
+        );
+        let (references_root, is_strict, is_fallible) = labels
+            .get(&ExactBoundExpr(expression.clone()))
+            .copied()
+            .unwrap_or((false, false, true));
+        if !references_root || !is_strict || is_fallible {
+            return Ok(None);
+        }
+
+        let values = EvalPlan::try_new(expression.clone(), child.values()?)?.into_plan();
+        Ok(Some(TakePlan::new(child.codes()?, values).into_plan()))
+    }
+}
diff --git a/vortex-layout/src/plan/tests.rs b/vortex-layout/src/plan/tests.rs
index 019ebbaf1ba..2b87f7ae3c5 100644
--- a/vortex-layout/src/plan/tests.rs
+++ b/vortex-layout/src/plan/tests.rs
@@ -8,7 +8,14 @@ use vortex_array::dtype::DType;
 use vortex_array::dtype::Nullability;
 use vortex_array::dtype::PType;
 use vortex_array::dtype::StructFields;
+use vortex_array::expr::Expression;
+use vortex_array::expr::and;
+use vortex_array::expr::checked_add;
 use vortex_array::expr::get_item;
+use vortex_array::expr::gt;
+use vortex_array::expr::is_null;
+use vortex_array::expr::lit;
+use vortex_array::expr::pack;
 use vortex_array::expr::root;
 use vortex_error::VortexResult;
 use vortex_error::vortex_err;
@@ -50,6 +57,20 @@ fn make_plan(layout: LayoutRef) -> VortexResult<PlanRef> {
     lower(&layout)
 }
 
+fn make_eval(expression: Expression, child: PlanRef) -> VortexResult<EvalPlan> {
+    let expression = expression
+        .optimize_recursive(child.dtype())?
+        .bind(child.dtype())?;
+    EvalPlan::try_new(expression, child)
+}
+
+fn make_row_idx_plan(expression: Expression, child: PlanRef) -> VortexResult<PlanRef> {
+    let expression = expression
+        .optimize_recursive(child.dtype())?
+        .bind(child.dtype())?;
+    plan_row_idx_expression(expression, child)
+}
+
 fn child_of(plan: &PlanRef, index: usize) -> VortexResult<PlanRef> {
     plan.child(index)?
         .ok_or_else(|| vortex_err!("missing child {index}"))
@@ -479,17 +500,588 @@ fn list_plan_display_handles_optional_validity() -> VortexResult<()> {
 }
 
 #[test]
-fn row_idx_plan_preserves_row_index_expressions() -> VortexResult<()> {
+fn row_idx_only_expression_uses_row_idx_source() -> VortexResult<()> {
     let layout = flat(3, primitive(PType::I32, Nullability::NonNullable), 0);
-    let plan = RowIdxPlan::new(10, make_plan(layout)?).into_plan();
-    let bound_expression = row_idx().bind(plan.dtype())?;
-    let plan = optimize(EvalPlan::try_new(bound_expression.clone(), plan)?.into_plan())?;
-    let expression = plan
+    let plan = make_row_idx_plan(row_idx(), make_plan(layout)?)?;
+
+    insta::assert_snapshot!(plan.display_tree(), @"
+    root: vortex.plan.eval(u64, rows=3) expr=$
+      child: vortex.plan.row_idx(u64, rows=3)
+    ");
+
+    let optimized = optimize(plan)?;
+    insta::assert_snapshot!(optimized.display_tree(), @"root: vortex.plan.row_idx(u64, rows=3)");
+    assert!(optimized.is::<RowIdx>());
+    Ok(())
+}
+
+#[test]
+fn expression_partitions_across_row_idx_and_struct() -> VortexResult<()> {
+    let value_dtype = primitive(PType::I32, Nullability::NonNullable);
+    let dictionary = DictLayout::new(
+        flat(2, value_dtype.clone(), 0),
+        flat(3, primitive(PType::U8, Nullability::NonNullable), 1),
+    )
+    .into_layout();
+    let layout = StructLayout::new(
+        3,
+        DType::Struct(
+            StructFields::from_iter([("a", value_dtype.clone()), ("b", value_dtype.clone())]),
+            Nullability::NonNullable,
+        ),
+        vec![dictionary, flat(3, value_dtype, 2)],
+    )
+    .into_layout();
+    let expression = and(
+        gt(row_idx(), lit(11_u64)),
+        and(
+            gt(get_item("a", root()), lit(5_i32)),
+            gt(get_item("b", root()), lit(7_i32)),
+        ),
+    );
+    let plan = make_row_idx_plan(expression, make_plan(layout)?)?;
+
+    let optimized = optimize(plan)?;
+    insta::assert_snapshot!(optimized.display_tree(), @"
+    root: vortex.plan.eval(bool, rows=3) expr=(($.row_idx and $.child.child_0) and $.child.child_1)
+      child: vortex.plan.pack({row_idx=bool, child={child_0=bool, child_1=bool}}, rows=3)
+        row_idx: vortex.plan.eval(bool, rows=3) expr=($ > 11u64)
+          child: vortex.plan.row_idx(u64, rows=3)
+        child: vortex.plan.eval({child_0=bool, child_1=bool}, rows=3) expr=pack(child_0: $.a, child_1: $.b)
+          child: vortex.plan.pack({a=bool, b=bool}, rows=3)
+            a: vortex.plan.take(bool, rows=3)
+              codes: vortex.plan.segment_scan(u8, rows=3)
+              values: vortex.plan.eval(bool, rows=2) expr=($ > 5i32)
+                child: vortex.plan.segment_scan(i32, rows=2)
+            b: vortex.plan.eval(bool, rows=3) expr=($ > 7i32)
+              child: vortex.plan.segment_scan(i32, rows=3)
+    ");
+    Ok(())
+}
+
+#[test]
+fn empty_projection_prunes_row_idx_child_fields() -> VortexResult<()> {
+    let value_dtype = primitive(PType::I32, Nullability::NonNullable);
+    let layout = StructLayout::new(
+        3,
+        DType::Struct(
+            StructFields::from_iter([("a", value_dtype.clone()), ("b", value_dtype.clone())]),
+            Nullability::NonNullable,
+        ),
+        vec![flat(3, value_dtype.clone(), 0), flat(3, value_dtype, 1)],
+    )
+    .into_layout();
+    let projection = pack(
+        std::iter::empty::<(&str, Expression)>(),
+        Nullability::NonNullable,
+    );
+    let plan = make_row_idx_plan(projection, make_plan(layout)?)?;
+
+    let optimized = optimize(plan)?;
+    let projection = optimized
         .as_opt::<Eval>()
-        .ok_or_else(|| vortex_err!("optimized plan is not an expression plan"))?;
+        .ok_or_else(|| vortex_err!("optimized plan has no projection expression"))?;
+    let child = projection.child_plan()?;
+    let empty_struct = child
+        .as_opt::<Pack>()
+        .ok_or_else(|| vortex_err!("empty projection did not prune the RowIdx child"))?;
+
+    assert_eq!(empty_struct.nfields(), 0);
+    assert_eq!(empty_struct.children().len(), 0);
+    Ok(())
+}
+
+#[test]
+fn row_idx_and_data_expression_pushes_data_into_chunks() -> VortexResult<()> {
+    let dtype = primitive(PType::I32, Nullability::NonNullable);
+    let layout = ChunkedLayout::new(
+        3,
+        dtype.clone(),
+        OwnedLayoutChildren::layout_children(vec![flat(1, dtype.clone(), 0), flat(2, dtype, 1)]),
+    )
+    .into_layout();
+    let child = make_plan(layout)?;
+    let expression = and(gt(row_idx(), lit(10_u64)), gt(root(), lit(5_i32)));
+    let plan = optimize(make_row_idx_plan(expression, child)?)?;
+
+    insta::assert_snapshot!(plan.display_tree(), @r"
+    root: vortex.plan.eval(bool, rows=3) expr=($.row_idx and $.child)
+      child: vortex.plan.pack({row_idx=bool, child=bool}, rows=3)
+        row_idx: vortex.plan.eval(bool, rows=3) expr=($ > 10u64)
+          child: vortex.plan.row_idx(u64, rows=3)
+        child: vortex.plan.concat(bool, rows=3)
+          chunks[0]: vortex.plan.eval(bool, rows=1) expr=($ > 5i32)
+            child: vortex.plan.segment_scan(i32, rows=1)
+          chunks[1]: vortex.plan.eval(bool, rows=2) expr=($ > 5i32)
+            child: vortex.plan.segment_scan(i32, rows=2)
+    ");
+    Ok(())
+}
+
+#[test]
+fn expression_pushes_through_struct_field_and_dictionary_values() -> VortexResult<()> {
+    let value_dtype = primitive(PType::I32, Nullability::NonNullable);
+    let codes_dtype = primitive(PType::U8, Nullability::NonNullable);
+    let dictionary =
+        DictLayout::new(flat(2, value_dtype.clone(), 0), flat(3, codes_dtype, 1)).into_layout();
+    let layout = StructLayout::new(
+        3,
+        DType::Struct(
+            StructFields::from_iter([("a", value_dtype.clone()), ("b", value_dtype.clone())]),
+            Nullability::NonNullable,
+        ),
+        vec![dictionary, flat(3, value_dtype, 2)],
+    )
+    .into_layout();
+    let plan = make_eval(gt(get_item("a", root()), lit(5_i32)), make_plan(layout)?)?.into_plan();
+
+    insta::assert_snapshot!(plan.display_tree(), @"
+    root: vortex.plan.eval(bool, rows=3) expr=($.a > 5i32)
+      child: vortex.plan.pack({a=i32, b=i32}, rows=3)
+        a: vortex.plan.take(i32, rows=3)
+          codes: vortex.plan.segment_scan(u8, rows=3)
+          values: vortex.plan.segment_scan(i32, rows=2)
+        b: vortex.plan.segment_scan(i32, rows=3)
+    ");
+
+    let optimized = optimize(plan)?;
+    insta::assert_snapshot!(optimized.display_tree(), @"
+    root: vortex.plan.take(bool, rows=3)
+      codes: vortex.plan.segment_scan(u8, rows=3)
+      values: vortex.plan.eval(bool, rows=2) expr=($ > 5i32)
+        child: vortex.plan.segment_scan(i32, rows=2)
+    ");
+    Ok(())
+}
+
+#[test]
+fn expression_pushes_through_struct_field_with_heterogeneous_chunks() -> VortexResult<()> {
+    let value_dtype = primitive(PType::I32, Nullability::NonNullable);
+    let dictionary = DictLayout::new(
+        flat(2, value_dtype.clone(), 0),
+        flat(3, primitive(PType::U8, Nullability::NonNullable), 1),
+    )
+    .into_layout();
+    let chunks = ChunkedLayout::new(
+        5,
+        value_dtype.clone(),
+        OwnedLayoutChildren::layout_children(vec![dictionary, flat(2, value_dtype.clone(), 2)]),
+    )
+    .into_layout();
+    let layout = StructLayout::new(
+        5,
+        DType::Struct(
+            StructFields::from_iter([("a", value_dtype.clone()), ("b", value_dtype.clone())]),
+            Nullability::NonNullable,
+        ),
+        vec![chunks, flat(5, value_dtype, 3)],
+    )
+    .into_layout();
+    let plan = make_eval(gt(get_item("a", root()), lit(5_i32)), make_plan(layout)?)?.into_plan();
+
+    insta::assert_snapshot!(plan.display_tree(), @"
+    root: vortex.plan.eval(bool, rows=5) expr=($.a > 5i32)
+      child: vortex.plan.pack({a=i32, b=i32}, rows=5)
+        a: vortex.plan.concat(i32, rows=5)
+          chunks[0]: vortex.plan.take(i32, rows=3)
+            codes: vortex.plan.segment_scan(u8, rows=3)
+            values: vortex.plan.segment_scan(i32, rows=2)
+          chunks[1]: vortex.plan.segment_scan(i32, rows=2)
+        b: vortex.plan.segment_scan(i32, rows=5)
+    ");
+
+    let optimized = optimize(plan)?;
+    insta::assert_snapshot!(optimized.display_tree(), @"
+    root: vortex.plan.concat(bool, rows=5)
+      chunks[0]: vortex.plan.take(bool, rows=3)
+        codes: vortex.plan.segment_scan(u8, rows=3)
+        values: vortex.plan.eval(bool, rows=2) expr=($ > 5i32)
+          child: vortex.plan.segment_scan(i32, rows=2)
+      chunks[1]: vortex.plan.eval(bool, rows=2) expr=($ > 5i32)
+        child: vortex.plan.segment_scan(i32, rows=2)
+    ");
+    Ok(())
+}
+
+#[test]
+fn expression_pushes_through_nested_struct_fields_in_one_pass() -> VortexResult<()> {
+    let value_dtype = primitive(PType::I32, Nullability::NonNullable);
+    let nested_dtype = DType::Struct(
+        StructFields::from_iter([("x", value_dtype.clone()), ("y", value_dtype.clone())]),
+        Nullability::NonNullable,
+    );
+    let nested = StructLayout::new(
+        3,
+        nested_dtype.clone(),
+        vec![
+            flat(3, value_dtype.clone(), 0),
+            flat(3, value_dtype.clone(), 1),
+        ],
+    )
+    .into_layout();
+    let layout = StructLayout::new(
+        3,
+        DType::Struct(
+            StructFields::from_iter([("nested", nested_dtype), ("z", value_dtype.clone())]),
+            Nullability::NonNullable,
+        ),
+        vec![nested, flat(3, value_dtype, 2)],
+    )
+    .into_layout();
+    let expression = gt(get_item("x", get_item("nested", root())), lit(5_i32));
+    let plan = make_eval(expression, make_plan(layout)?)?.into_plan();
+
+    let optimized = optimize(plan)?;
+    insta::assert_snapshot!(optimized.display_tree(), @r"
+    root: vortex.plan.eval(bool, rows=3) expr=($ > 5i32)
+      child: vortex.plan.segment_scan(i32, rows=3)
+    ");
+    Ok(())
+}
+
+#[test]
+fn expression_pushes_through_single_field_nested_structs() -> VortexResult<()> {
+    let value_dtype = primitive(PType::I32, Nullability::NonNullable);
+    let inner_dtype = DType::Struct(
+        StructFields::from_iter([("b", value_dtype.clone())]),
+        Nullability::NonNullable,
+    );
+    let inner =
+        StructLayout::new(3, inner_dtype.clone(), vec![flat(3, value_dtype, 0)]).into_layout();
+    let layout = StructLayout::new(
+        3,
+        DType::Struct(
+            StructFields::from_iter([("a", inner_dtype)]),
+            Nullability::NonNullable,
+        ),
+        vec![inner],
+    )
+    .into_layout();
+    let expression = gt(get_item("b", get_item("a", root())), lit(5_i32));
+    let plan = make_eval(expression, make_plan(layout)?)?.into_plan();
+
+    insta::assert_snapshot!(plan.display_tree(), @r"
+    root: vortex.plan.eval(bool, rows=3) expr=($.a.b > 5i32)
+      child: vortex.plan.pack({a={b=i32}}, rows=3)
+        a: vortex.plan.pack({b=i32}, rows=3)
+          b: vortex.plan.segment_scan(i32, rows=3)
+    ");
+
+    let optimized = optimize(plan)?;
+    insta::assert_snapshot!(optimized.display_tree(), @r"
+    root: vortex.plan.eval(bool, rows=3) expr=($ > 5i32)
+      child: vortex.plan.segment_scan(i32, rows=3)
+    ");
+    Ok(())
+}
+
+#[test]
+fn expression_pushes_through_three_single_field_structs() -> VortexResult<()> {
+    let value_dtype = primitive(PType::I32, Nullability::NonNullable);
+    let c_dtype = DType::Struct(
+        StructFields::from_iter([("c", value_dtype.clone())]),
+        Nullability::NonNullable,
+    );
+    let b_dtype = DType::Struct(
+        StructFields::from_iter([("b", c_dtype.clone())]),
+        Nullability::NonNullable,
+    );
+    let c = StructLayout::new(3, c_dtype, vec![flat(3, value_dtype, 0)]).into_layout();
+    let b = StructLayout::new(3, b_dtype.clone(), vec![c]).into_layout();
+    let layout = StructLayout::new(
+        3,
+        DType::Struct(
+            StructFields::from_iter([("a", b_dtype)]),
+            Nullability::NonNullable,
+        ),
+        vec![b],
+    )
+    .into_layout();
+    let expression = gt(
+        get_item("c", get_item("b", get_item("a", root()))),
+        lit(5_i32),
+    );
+    let plan = make_eval(expression, make_plan(layout)?)?.into_plan();
+
+    let optimized = optimize(plan)?;
+    insta::assert_snapshot!(optimized.display_tree(), @r"
+    root: vortex.plan.eval(bool, rows=3) expr=($ > 5i32)
+      child: vortex.plan.segment_scan(i32, rows=3)
+    ");
+    Ok(())
+}
+
+#[test]
+fn compound_expression_pushes_through_nested_struct_and_dictionary() -> VortexResult<()> {
+    let value_dtype = primitive(PType::I32, Nullability::NonNullable);
+    let codes_dtype = primitive(PType::U8, Nullability::NonNullable);
+    let dictionary =
+        DictLayout::new(flat(2, value_dtype.clone(), 0), flat(3, codes_dtype, 1)).into_layout();
+    let inner_dtype = DType::Struct(
+        StructFields::from_iter([
+            ("b", value_dtype.clone()),
+            ("c", value_dtype.clone()),
+            ("unused", value_dtype.clone()),
+        ]),
+        Nullability::NonNullable,
+    );
+    let inner = StructLayout::new(
+        3,
+        inner_dtype.clone(),
+        vec![
+            dictionary,
+            flat(3, value_dtype.clone(), 2),
+            flat(3, value_dtype.clone(), 3),
+        ],
+    )
+    .into_layout();
+    let layout = StructLayout::new(
+        3,
+        DType::Struct(
+            StructFields::from_iter([("a", inner_dtype), ("unused", value_dtype.clone())]),
+            Nullability::NonNullable,
+        ),
+        vec![inner, flat(3, value_dtype, 4)],
+    )
+    .into_layout();
+    let expression = and(
+        gt(get_item("b", get_item("a", root())), lit(5_i32)),
+        gt(get_item("c", get_item("a", root())), lit(7_i32)),
+    );
+    let plan = make_eval(expression, make_plan(layout)?)?.into_plan();
+
+    let optimized = optimize(plan)?;
+    insta::assert_snapshot!(optimized.display_tree(), @r"
+    root: vortex.plan.eval(bool, rows=3) expr=($.b and $.c)
+      child: vortex.plan.pack({b=bool, c=bool}, rows=3)
+        b: vortex.plan.take(bool, rows=3)
+          codes: vortex.plan.segment_scan(u8, rows=3)
+          values: vortex.plan.eval(bool, rows=2) expr=($ > 5i32)
+            child: vortex.plan.segment_scan(i32, rows=2)
+        c: vortex.plan.eval(bool, rows=3) expr=($ > 7i32)
+          child: vortex.plan.segment_scan(i32, rows=3)
+    ");
+    Ok(())
+}
+
+#[test]
+fn expression_pushes_through_dictionary_of_struct_values() -> VortexResult<()> {
+    let value_dtype = primitive(PType::I32, Nullability::NonNullable);
+    let values_dtype = DType::Struct(
+        StructFields::from_iter([("b", value_dtype.clone()), ("unused", value_dtype.clone())]),
+        Nullability::NonNullable,
+    );
+    let values = StructLayout::new(
+        2,
+        values_dtype.clone(),
+        vec![
+            flat(2, value_dtype.clone(), 0),
+            flat(2, value_dtype.clone(), 1),
+        ],
+    )
+    .into_layout();
+    let dictionary = DictLayout::new(
+        values,
+        flat(3, primitive(PType::U8, Nullability::NonNullable), 2),
+    )
+    .into_layout();
+    let layout = StructLayout::new(
+        3,
+        DType::Struct(
+            StructFields::from_iter([("a", values_dtype), ("unused", value_dtype.clone())]),
+            Nullability::NonNullable,
+        ),
+        vec![dictionary, flat(3, value_dtype, 3)],
+    )
+    .into_layout();
+    let expression = gt(get_item("b", get_item("a", root())), lit(5_i32));
+    let plan = make_eval(expression, make_plan(layout)?)?.into_plan();
+
+    let optimized = optimize(plan)?;
+    insta::assert_snapshot!(optimized.display_tree(), @r"
+    root: vortex.plan.take(bool, rows=3)
+      codes: vortex.plan.segment_scan(u8, rows=3)
+      values: vortex.plan.eval(bool, rows=2) expr=($ > 5i32)
+        child: vortex.plan.segment_scan(i32, rows=2)
+    ");
+    Ok(())
+}
+
+#[test]
+fn multi_field_struct_expression_pushes_into_each_field() -> VortexResult<()> {
+    let value_dtype = primitive(PType::I32, Nullability::NonNullable);
+    let dictionary = DictLayout::new(
+        flat(2, value_dtype.clone(), 0),
+        flat(3, primitive(PType::U8, Nullability::NonNullable), 1),
+    )
+    .into_layout();
+    let layout = StructLayout::new(
+        3,
+        DType::Struct(
+            StructFields::from_iter([
+                ("a", value_dtype.clone()),
+                ("b", value_dtype.clone()),
+                ("c", value_dtype.clone()),
+            ]),
+            Nullability::NonNullable,
+        ),
+        vec![
+            dictionary,
+            flat(3, value_dtype.clone(), 2),
+            flat(3, value_dtype, 3),
+        ],
+    )
+    .into_layout();
+    let expression = and(
+        gt(get_item("a", root()), lit(5_i32)),
+        gt(get_item("b", root()), lit(7_i32)),
+    );
+    let plan = make_eval(expression, make_plan(layout)?)?.into_plan();
+
+    insta::assert_snapshot!(plan.display_tree(), @"
+    root: vortex.plan.eval(bool, rows=3) expr=(($.a > 5i32) and ($.b > 7i32))
+      child: vortex.plan.pack({a=i32, b=i32, c=i32}, rows=3)
+        a: vortex.plan.take(i32, rows=3)
+          codes: vortex.plan.segment_scan(u8, rows=3)
+          values: vortex.plan.segment_scan(i32, rows=2)
+        b: vortex.plan.segment_scan(i32, rows=3)
+        c: vortex.plan.segment_scan(i32, rows=3)
+    ");
+
+    let optimized = optimize(plan)?;
+    insta::assert_snapshot!(optimized.display_tree(), @"
+    root: vortex.plan.eval(bool, rows=3) expr=($.a and $.b)
+      child: vortex.plan.pack({a=bool, b=bool}, rows=3)
+        a: vortex.plan.take(bool, rows=3)
+          codes: vortex.plan.segment_scan(u8, rows=3)
+          values: vortex.plan.eval(bool, rows=2) expr=($ > 5i32)
+            child: vortex.plan.segment_scan(i32, rows=2)
+        b: vortex.plan.eval(bool, rows=3) expr=($ > 7i32)
+          child: vortex.plan.segment_scan(i32, rows=3)
+    ");
+    let reoptimized = optimize(optimized.clone())?;
+    assert!(PlanRef::ptr_eq(&optimized, &reoptimized));
+    Ok(())
+}
+
+#[test]
+fn repeated_cross_field_expressions_reach_a_fixed_point() -> VortexResult<()> {
+    let value_dtype = primitive(PType::I32, Nullability::NonNullable);
+    let layout = StructLayout::new(
+        3,
+        DType::Struct(
+            StructFields::from_iter([("a", value_dtype.clone()), ("b", value_dtype.clone())]),
+            Nullability::NonNullable,
+        ),
+        vec![flat(3, value_dtype.clone(), 0), flat(3, value_dtype, 1)],
+    )
+    .into_layout();
+    let a = get_item("a", root());
+    let b = get_item("b", root());
+    let expression = and(
+        gt(checked_add(a.clone(), b.clone()), lit(10_i32)),
+        gt(checked_add(a, b), lit(20_i32)),
+    );
+    let plan = make_eval(expression, make_plan(layout)?)?.into_plan();
 
-    assert_eq!(expression.expression(), &bound_expression);
-    assert!(expression.child_plan()?.is::<RowIdx>());
-    assert_eq!(expression.row_count(), 3);
+    let optimized = optimize(plan)?;
+    let reoptimized = optimize(optimized.clone())?;
+    assert!(PlanRef::ptr_eq(&optimized, &reoptimized));
+    Ok(())
+}
+
+#[test]
+fn multi_field_struct_expression_keeps_cross_field_refinement() -> VortexResult<()> {
+    let value_dtype = primitive(PType::I32, Nullability::NonNullable);
+    let dictionary = DictLayout::new(
+        flat(2, value_dtype.clone(), 0),
+        flat(3, primitive(PType::U8, Nullability::NonNullable), 1),
+    )
+    .into_layout();
+    let layout = StructLayout::new(
+        3,
+        DType::Struct(
+            StructFields::from_iter([
+                ("a", value_dtype.clone()),
+                ("b", value_dtype.clone()),
+                ("c", value_dtype.clone()),
+            ]),
+            Nullability::NonNullable,
+        ),
+        vec![
+            dictionary,
+            flat(3, value_dtype.clone(), 2),
+            flat(3, value_dtype, 3),
+        ],
+    )
+    .into_layout();
+    let expression = gt(
+        checked_add(get_item("a", root()), get_item("b", root())),
+        lit(10_i32),
+    );
+    let plan = make_eval(expression, make_plan(layout)?)?.into_plan();
+
+    let optimized = optimize(plan)?;
+    insta::assert_snapshot!(optimized.display_tree(), @"
+    root: vortex.plan.eval(bool, rows=3) expr=(($.a + $.b) > 10i32)
+      child: vortex.plan.pack({a=i32, b=i32}, rows=3)
+        a: vortex.plan.take(i32, rows=3)
+          codes: vortex.plan.segment_scan(u8, rows=3)
+          values: vortex.plan.segment_scan(i32, rows=2)
+        b: vortex.plan.segment_scan(i32, rows=3)
+    ");
+    assert_eq!(
+        optimized.display_tree().to_string(),
+        optimize(optimized.clone())?.display_tree().to_string()
+    );
+    Ok(())
+}
+
+#[test]
+fn dictionary_pushdown_rejects_unsafe_expressions() -> VortexResult<()> {
+    let value_dtype = primitive(PType::I32, Nullability::NonNullable);
+    let dictionary = DictLayout::new(
+        flat(2, value_dtype, 0),
+        flat(3, primitive(PType::U8, Nullability::NonNullable), 1),
+    )
+    .into_layout();
+
+    for expression in [
+        lit(false),
+        is_null(root()),
+        gt(checked_add(root(), lit(1_i32)), lit(5_i32)),
+    ] {
+        let plan = make_eval(expression.clone(), make_plan(Arc::clone(&dictionary))?)?.into_plan();
+        let optimized = optimize(plan)?;
+        let eval = optimized.as_opt::<Eval>().ok_or_else(|| {
+            vortex_err!("Expression unexpectedly pushed into dictionary: {expression}")
+        })?;
+        assert!(eval.child_plan()?.is::<Take>());
+    }
+    Ok(())
+}
+
+#[test]
+fn nullable_struct_keeps_expression_above_parent_validity() -> VortexResult<()> {
+    let field_dtype = primitive(PType::I32, Nullability::NonNullable);
+    let layout = StructLayout::new(
+        3,
+        DType::Struct(
+            StructFields::from_iter([("a", field_dtype.clone())]),
+            Nullability::Nullable,
+        ),
+        vec![
+            flat(3, DType::Bool(Nullability::NonNullable), 0),
+            flat(3, field_dtype, 1),
+        ],
+    )
+    .into_layout();
+    let plan = make_eval(gt(get_item("a", root()), lit(5_i32)), make_plan(layout)?)?.into_plan();
+
+    let optimized = optimize(plan)?;
+    let eval = optimized
+        .as_opt::<Eval>()
+        .ok_or_else(|| vortex_err!("Nullable struct expression unexpectedly pushed down"))?;
+    assert!(eval.child_plan()?.is::<Pack>());
     Ok(())
 }