From 7ca2a657e4b93ce6c650ef570cd8b07f524007c5 Mon Sep 17 00:00:00 2001
From: LiaCastaneda <lia.castaneda@datadoghq.com>
Date: Fri, 7 Aug 2026 12:04:13 +0200
Subject: [PATCH 1/8] fix(lambda): only push referenced params into the merged
 batch

---
 datafusion/expr/src/higher_order_function.rs  |  86 ++++++--
 .../physical-expr/src/expressions/lambda.rs   | 190 ++++++++++++++++--
 .../src/higher_order_function.rs              |   1 +
 3 files changed, 245 insertions(+), 32 deletions(-)

diff --git a/datafusion/expr/src/higher_order_function.rs b/datafusion/expr/src/higher_order_function.rs
index 413714f498164..a004bf2fdca2c 100644
--- a/datafusion/expr/src/higher_order_function.rs
+++ b/datafusion/expr/src/higher_order_function.rs
@@ -24,7 +24,7 @@ use crate::expr::{
 use crate::type_coercion::functions::value_fields_with_higher_order_udf;
 use crate::udf_eq::UdfEq;
 use crate::{ColumnarValue, Documentation, Expr, ExprSchemable};
-use arrow::array::{ArrayRef, RecordBatch};
+use arrow::array::{ArrayRef, RecordBatch, RecordBatchOptions};
 use arrow::datatypes::{DataType, FieldRef, Schema};
 use arrow_schema::SchemaRef;
 use datafusion_common::config::ConfigOptions;
@@ -239,6 +239,19 @@ pub struct LambdaArgument {
     /// For example, for `array_transform([2], v -> -v)`,
     /// this will be `vec![Field::new("v", DataType::Int32, true)]`
     params: Vec<FieldRef>,
+    /// Indices into [`Self::params`] of the parameters that are actually
+    /// referenced by [`Self::body`] (taking nested-lambda shadowing into
+    /// account), in the original declaration order of `params`.
+    ///
+    /// [`Self::evaluate`] only evaluates and pushes the closures whose
+    /// corresponding parameter index appears here, so unused declared
+    /// parameters leave no slot in the merged batch and the body's compressed
+    /// column indices line up directly with what the evaluator built.
+    ///
+    /// Callers who already have a [`LambdaExpr`] should read the used names
+    /// from `LambdaExpr::used_params()` and pass them to [`Self::new`];
+    /// [`Self::new`] handles the name → index translation.
+    used_param_indices: Vec<usize>,
     /// The body of the lambda
     ///
     /// For example, for `array_transform([2], v -> -v)`,
@@ -257,26 +270,49 @@ pub struct LambdaArgument {
 }
 
 impl LambdaArgument {
+    /// Build a [`LambdaArgument`] for a lambda whose body references the
+    /// subset of `params` named in `used_params`.
+    ///
+    /// [`Self::evaluate`] only materialises the closures whose parameter name
+    /// appears in `used_params`, preserving the original declaration order of
+    /// `params`. Unused declared parameters therefore leave no slot in the
+    /// merged batch, so the body's compressed column indices line up directly
+    /// with the columns the evaluator built.
+    ///
+    /// Callers with a `LambdaExpr` in hand should pass `lambda.used_params()`;
+    /// that method already computes the exact set required here (with
+    /// nested-lambda shadow tracking).
     pub fn new(
         params: Vec<FieldRef>,
         body: Arc<dyn PhysicalExpr>,
         captures: Option<RecordBatch>,
+        used_params: &HashSet<String>,
     ) -> Self {
-        let fields = match &captures {
+        let used_param_indices: Vec<usize> = params
+            .iter()
+            .enumerate()
+            .filter(|(_, f)| used_params.contains(f.name()))
+            .map(|(i, _)| i)
+            .collect();
+
+        let effective_params = used_param_indices.iter().map(|i| Arc::clone(&params[*i]));
+
+        let fields: Vec<FieldRef> = match &captures {
             Some(batch) => batch
                 .schema_ref()
                 .fields()
                 .iter()
                 .cloned()
-                .chain(params.clone())
+                .chain(effective_params)
                 .collect(),
-            None => params.clone(),
+            None => effective_params.collect(),
         };
 
         let schema = Arc::new(Schema::new(fields));
 
         Self {
             params,
+            used_param_indices,
             body,
             schema,
             captures,
@@ -344,6 +380,7 @@ impl LambdaArgument {
             spread_captures.as_ref(),
             Arc::clone(&self.schema),
             &self.params,
+            &self.used_param_indices,
             args,
         )?;
 
@@ -355,6 +392,7 @@ fn merge_captures_with_variables(
     captures: Option<&RecordBatch>,
     schema: SchemaRef,
     params: &[FieldRef],
+    used_param_indices: &[usize],
     variables: &[&dyn Fn() -> Result<ArrayRef>],
 ) -> Result<RecordBatch> {
     if variables.len() < params.len() {
@@ -365,23 +403,43 @@ fn merge_captures_with_variables(
         );
     }
 
+    let push_param_arrays = |columns: &mut Vec<ArrayRef>| -> Result<()> {
+        for &i in used_param_indices {
+            columns.push(variables[i]()?);
+        }
+        Ok(())
+    };
+
     let columns = match captures {
         Some(captures) => {
             let mut columns = captures.columns().to_vec();
-
-            for arg in &variables[..params.len()] {
-                columns.push(arg()?);
-            }
-
+            push_param_arrays(&mut columns)?;
+            columns
+        }
+        None => {
+            let mut columns = Vec::with_capacity(used_param_indices.len());
+            push_param_arrays(&mut columns)?;
             columns
         }
-        None => variables
-            .iter()
-            .take(params.len())
-            .map(|arg| arg())
-            .collect::<Result<_>>()?,
     };
 
+    if columns.is_empty() {
+        // Constant lambda body with no captures and no used parameters. We
+        // still need a row count for the merged batch, so evaluate one
+        // variable just to derive it. This is essentially free in the common
+        // case (the variables already exist as closures over arrays the
+        // caller computed up front).
+        let row_count = match variables.first() {
+            Some(first) => first()?.len(),
+            None => 0,
+        };
+        return Ok(RecordBatch::try_new_with_options(
+            schema,
+            vec![],
+            &RecordBatchOptions::new().with_row_count(Some(row_count)),
+        )?);
+    }
+
     Ok(RecordBatch::try_new(schema, columns)?)
 }
 
diff --git a/datafusion/physical-expr/src/expressions/lambda.rs b/datafusion/physical-expr/src/expressions/lambda.rs
index cab2eea64fcf4..de22ec2f6bea4 100644
--- a/datafusion/physical-expr/src/expressions/lambda.rs
+++ b/datafusion/physical-expr/src/expressions/lambda.rs
@@ -31,7 +31,7 @@ use arrow::{
 };
 use datafusion_common::{
     HashMap, plan_err,
-    tree_node::{Transformed, TreeNode, TreeNodeRecursion},
+    tree_node::{Transformed, TreeNode, TreeNodeRecursion, TreeNodeVisitor},
 };
 use datafusion_common::{HashSet, Result, internal_err};
 use datafusion_expr::ColumnarValue;
@@ -43,6 +43,15 @@ pub struct LambdaExpr {
     body: Arc<dyn PhysicalExpr>,
     projected_body: Arc<dyn PhysicalExpr>,
     projection: Vec<usize>,
+    /// Subset of `params` (by name) that the body actually references,
+    /// computed with nested-lambda shadow tracking. Empty when no parameter
+    /// is referenced by this lambda's own body.
+    ///
+    /// The higher-order function uses this to only evaluate and push the
+    /// parameters the body actually needs into the merged evaluation batch,
+    /// which keeps the body's compressed column indices aligned with the
+    /// batch layout produced at runtime.
+    used_params: HashSet<String>,
 }
 
 // Manually derive PartialEq and Hash to work around https://github.com/rust-lang/rust/issues/78808 [https://github.com/apache/datafusion/issues/13196]
@@ -60,7 +69,7 @@ impl Hash for LambdaExpr {
 }
 
 impl LambdaExpr {
-    /// Create a new lambda expression with the given parameters and body
+    /// Create a new lambda expression with the given parameters and body.
     pub fn try_new(params: Vec<String>, body: Arc<dyn PhysicalExpr>) -> Result<Self> {
         if !all_unique(&params) {
             return plan_err!(
@@ -75,27 +84,30 @@ impl LambdaExpr {
     }
 
     fn new(params: Vec<String>, body: Arc<dyn PhysicalExpr>) -> Self {
-        let mut used_column_indices = HashSet::new();
+        let own_params: HashSet<String> = params.iter().cloned().collect();
 
-        body.apply(|node| {
-            if let Some(col) = node.downcast_ref::<Column>() {
-                used_column_indices.insert(col.index());
-            } else if let Some(var) = node.downcast_ref::<LambdaVariable>() {
-                used_column_indices.insert(var.index());
-            }
-
-            Ok(TreeNodeRecursion::Continue)
-        })
-        .expect("closure should be infallible");
+        let mut visitor = CollectUsedVisitor {
+            own_params: &own_params,
+            used_indices: HashSet::new(),
+            used_param_names: HashSet::new(),
+            shadow_stack: Vec::new(),
+        };
+        body.visit(&mut visitor).expect("visitor is infallible");
+        let CollectUsedVisitor {
+            used_indices,
+            used_param_names,
+            ..
+        } = visitor;
 
-        let mut projection = used_column_indices.into_iter().collect::<Vec<_>>();
+        let mut projection = used_indices.into_iter().collect::<Vec<_>>();
 
         projection.sort();
 
         let column_index_map = projection
             .iter()
+            .copied()
             .enumerate()
-            .map(|(projected, original)| (*original, projected))
+            .map(|(new_idx, original)| (original, new_idx))
             .collect::<HashMap<_, _>>();
 
         let projected_body = Arc::clone(&body)
@@ -129,6 +141,7 @@ impl LambdaExpr {
             body,
             projected_body,
             projection,
+            used_params: used_param_names,
         }
     }
 
@@ -170,6 +183,67 @@ impl LambdaExpr {
     pub(crate) fn projected_body(&self) -> &Arc<dyn PhysicalExpr> {
         &self.projected_body
     }
+
+    /// Subset of [`params`](Self::params) (by name) that the body actually
+    /// references, taking nested-lambda shadowing into account. Used by the
+    /// higher-order function evaluator to skip evaluating/pushing parameters
+    /// the lambda body does not need, so that unused declared parameters do
+    /// not shift the merged batch's column positions out of sync with the
+    /// body's compressed indices.
+    pub fn used_params(&self) -> &HashSet<String> {
+        &self.used_params
+    }
+}
+
+/// Walks the body of a [`LambdaExpr`] and collects, on a single pass:
+///
+/// * `used_indices` — every `Column` / `LambdaVariable` index referenced
+///   anywhere in the tree (including inside nested lambdas). This drives
+///   the `projection` used to slice the outer batch.
+/// * `used_param_names` — the subset of *this* lambda's `own_params` that
+///   the body actually references, with nested-lambda parameters shadowing
+///   the outer ones. For example, in
+///   `(k, v) -> func(col, (k, v2) -> k + v2 + v)` the inner `k` shadows the
+///   outer `k`, so only `v` flows up as used.
+///
+/// The shadow stack uses `TreeNodeVisitor`'s `f_down` / `f_up` callbacks
+/// directly: push a frame when entering a nested [`LambdaExpr`], pop it
+/// when leaving.
+struct CollectUsedVisitor<'a> {
+    own_params: &'a HashSet<String>,
+    used_indices: HashSet<usize>,
+    used_param_names: HashSet<String>,
+    shadow_stack: Vec<HashSet<String>>,
+}
+
+impl TreeNodeVisitor<'_> for CollectUsedVisitor<'_> {
+    type Node = Arc<dyn PhysicalExpr>;
+
+    fn f_down(&mut self, node: &Self::Node) -> Result<TreeNodeRecursion> {
+        if let Some(col) = node.downcast_ref::<Column>() {
+            self.used_indices.insert(col.index());
+        } else if let Some(var) = node.downcast_ref::<LambdaVariable>() {
+            self.used_indices.insert(var.index());
+
+            let name = var.name();
+            let shadowed = self.shadow_stack.iter().any(|frame| frame.contains(name));
+            if !shadowed && self.own_params.contains(name) {
+                self.used_param_names.insert(name.to_string());
+            }
+        } else if let Some(nested) = node.downcast_ref::<LambdaExpr>() {
+            self.shadow_stack
+                .push(nested.params.iter().cloned().collect());
+        }
+
+        Ok(TreeNodeRecursion::Continue)
+    }
+
+    fn f_up(&mut self, node: &Self::Node) -> Result<TreeNodeRecursion> {
+        if node.downcast_ref::<LambdaExpr>().is_some() {
+            self.shadow_stack.pop();
+        }
+        Ok(TreeNodeRecursion::Continue)
+    }
 }
 
 impl std::fmt::Display for LambdaExpr {
@@ -234,7 +308,7 @@ impl PhysicalExpr for LambdaExpr {
     }
 }
 
-/// Create a lambda expression
+/// Create a lambda expression.
 pub fn lambda(
     params: impl IntoIterator<Item = impl Into<String>>,
     body: Arc<dyn PhysicalExpr>,
@@ -273,10 +347,15 @@ fn check_async_udf(body: &Arc<dyn PhysicalExpr>) -> Result<()> {
 
 #[cfg(test)]
 mod tests {
-    use crate::expressions::{NoOp, lambda::lambda};
-    use arrow::{array::RecordBatch, datatypes::Schema};
+    use crate::expressions::{Column, LambdaVariable, NoOp, lambda::lambda};
+    use arrow::{
+        array::RecordBatch,
+        datatypes::{DataType, Field, Schema},
+    };
     use std::sync::Arc;
 
+    use super::LambdaExpr;
+
     #[test]
     fn test_lambda_evaluate() {
         let lambda = lambda(["a"], Arc::new(NoOp::new())).unwrap();
@@ -288,4 +367,79 @@ mod tests {
     fn test_lambda_duplicate_name() {
         assert!(lambda(["a", "a"], Arc::new(NoOp::new())).is_err());
     }
+
+    /// A two-parameter lambda whose body only references the second
+    /// parameter (`v`) must report only `v` as used. The higher-order
+    /// function uses this set to push only `v` into the merged batch, so
+    /// the body's compressed `LambdaVariable` index for `v` lines up with
+    /// the batch layout.
+    #[test]
+    fn test_used_params_collects_only_referenced_param() {
+        let v_field = Arc::new(Field::new("v", DataType::Int32, true));
+        let body = Arc::new(LambdaVariable::new(1, Arc::clone(&v_field)));
+
+        let lambda =
+            LambdaExpr::try_new(vec!["k".to_string(), "v".to_string()], body).unwrap();
+
+        assert_eq!(lambda.projection(), &[1]);
+        let used = lambda.used_params();
+        assert!(used.contains("v"));
+        assert!(!used.contains("k"));
+        assert_eq!(used.len(), 1);
+    }
+
+    /// Inside a nested lambda that re-declares one of the outer parameter
+    /// names, only the non-shadowed outer references should be reported as
+    /// used by the outer lambda. In
+    /// `(k, v) -> func(col, (k, v2) -> k + v2 + v)` the inner `k` shadows
+    /// the outer `k`, so the outer lambda must only see `v` as used.
+    #[test]
+    fn test_used_params_handles_shadowing_inside_nested_lambda() {
+        let outer_k_field = Arc::new(Field::new("k", DataType::Int32, true));
+        let outer_v_field = Arc::new(Field::new("v", DataType::Int32, true));
+        let inner_v2_field = Arc::new(Field::new("v2", DataType::Int32, true));
+
+        // Inner lambda body references "k" (inner's), "v2" (inner's), and
+        // "v" (outer's). Build it directly with the dense compressed
+        // indices the inner LambdaExpr::new would produce: sorted referenced
+        // indices, so the names alone matter here — what matters for
+        // shadow tracking is the names, not the indices.
+        let inner_body: Arc<dyn crate::PhysicalExpr> =
+            Arc::new(crate::expressions::BinaryExpr::new(
+                Arc::new(crate::expressions::BinaryExpr::new(
+                    Arc::new(LambdaVariable::new(1, Arc::clone(&outer_k_field))),
+                    datafusion_expr::Operator::Plus,
+                    Arc::new(LambdaVariable::new(2, Arc::clone(&inner_v2_field))),
+                )),
+                datafusion_expr::Operator::Plus,
+                Arc::new(LambdaVariable::new(0, Arc::clone(&outer_v_field))),
+            ));
+        let inner_lambda = Arc::new(
+            LambdaExpr::try_new(vec!["k".to_string(), "v2".to_string()], inner_body)
+                .unwrap(),
+        );
+
+        // Outer body wraps the inner lambda in a binary op next to a
+        // regular column reference so the walk has something non-trivial
+        // to descend through. The outer body references the inner lambda
+        // via `inner_lambda`.
+        let outer_body: Arc<dyn crate::PhysicalExpr> =
+            Arc::new(crate::expressions::BinaryExpr::new(
+                Arc::new(Column::new("col", 0)),
+                datafusion_expr::Operator::Plus,
+                inner_lambda,
+            ));
+
+        let outer_lambda =
+            LambdaExpr::try_new(vec!["k".to_string(), "v".to_string()], outer_body)
+                .unwrap();
+
+        let used = outer_lambda.used_params();
+        assert!(used.contains("v"), "outer's `v` should be reported as used");
+        assert!(
+            !used.contains("k"),
+            "outer's `k` is shadowed inside the nested lambda and should not be reported as used"
+        );
+        assert_eq!(used.len(), 1);
+    }
 }
diff --git a/datafusion/physical-expr/src/higher_order_function.rs b/datafusion/physical-expr/src/higher_order_function.rs
index 7390eb33a0922..79ff9c4e1cfea 100644
--- a/datafusion/physical-expr/src/higher_order_function.rs
+++ b/datafusion/physical-expr/src/higher_order_function.rs
@@ -353,6 +353,7 @@ impl PhysicalExpr for HigherOrderFunctionExpr {
                         } else {
                             Some(batch.project(&projection)?)
                         },
+                        lambda.used_params(),
                     )))
                 }
                 ArgSlot::Value => {

From a2b3dee659676166d4dfa592b02f03247481ae53 Mon Sep 17 00:00:00 2001
From: LiaCastaneda <lia.castaneda@datadoghq.com>
Date: Fri, 7 Aug 2026 12:41:34 +0200
Subject: [PATCH 2/8] Add more tests

---
 .../physical-expr/src/expressions/lambda.rs   | 66 +++++++++++++++++++
 1 file changed, 66 insertions(+)

diff --git a/datafusion/physical-expr/src/expressions/lambda.rs b/datafusion/physical-expr/src/expressions/lambda.rs
index de22ec2f6bea4..08346c0bb0f7a 100644
--- a/datafusion/physical-expr/src/expressions/lambda.rs
+++ b/datafusion/physical-expr/src/expressions/lambda.rs
@@ -388,6 +388,72 @@ mod tests {
         assert_eq!(used.len(), 1);
     }
 
+    /// A lambda whose body references neither declared parameter (e.g. a
+    /// constant expression) must report an empty used-params set. This
+    /// exercises the merged-batch path that has no parameter columns to
+    /// push at all.
+    #[test]
+    fn test_used_params_all_unused() {
+        let body = Arc::new(NoOp::new());
+
+        let lambda =
+            LambdaExpr::try_new(vec!["k".to_string(), "v".to_string()], body).unwrap();
+
+        assert!(lambda.projection().is_empty());
+        assert!(lambda.used_params().is_empty());
+    }
+
+    /// A three-parameter lambda whose body skips the middle parameter must
+    /// report only the first and last as used, preserving declaration
+    /// order regardless of how [`Self::used_params`] (a set) reports them.
+    #[test]
+    fn test_used_params_three_params_middle_unused() {
+        let a_field = Arc::new(Field::new("a", DataType::Int32, true));
+        let c_field = Arc::new(Field::new("c", DataType::Int32, true));
+        let body = Arc::new(crate::expressions::BinaryExpr::new(
+            Arc::new(LambdaVariable::new(0, Arc::clone(&a_field))),
+            datafusion_expr::Operator::Plus,
+            Arc::new(LambdaVariable::new(2, Arc::clone(&c_field))),
+        ));
+
+        let lambda = LambdaExpr::try_new(
+            vec!["a".to_string(), "b".to_string(), "c".to_string()],
+            body,
+        )
+        .unwrap();
+
+        let used = lambda.used_params();
+        assert!(used.contains("a"));
+        assert!(!used.contains("b"));
+        assert!(used.contains("c"));
+        assert_eq!(used.len(), 2);
+    }
+
+    /// A two-parameter lambda whose body references both parameters, but in
+    /// the reverse of their declared order (`v` before `k`), must still
+    /// report both as used. Declaration order (not reference order) is what
+    /// downstream code (`LambdaArgument::new`) relies on when it maps these
+    /// names back to positions in the merged batch.
+    #[test]
+    fn test_used_params_both_used_in_reverse_reference_order() {
+        let k_field = Arc::new(Field::new("k", DataType::Int32, true));
+        let v_field = Arc::new(Field::new("v", DataType::Int32, true));
+        let body = Arc::new(crate::expressions::BinaryExpr::new(
+            Arc::new(LambdaVariable::new(1, Arc::clone(&v_field))),
+            datafusion_expr::Operator::Plus,
+            Arc::new(LambdaVariable::new(0, Arc::clone(&k_field))),
+        ));
+
+        let lambda =
+            LambdaExpr::try_new(vec!["k".to_string(), "v".to_string()], body).unwrap();
+
+        assert_eq!(lambda.projection(), &[0, 1]);
+        let used = lambda.used_params();
+        assert!(used.contains("k"));
+        assert!(used.contains("v"));
+        assert_eq!(used.len(), 2);
+    }
+
     /// Inside a nested lambda that re-declares one of the outer parameter
     /// names, only the non-shadowed outer references should be reported as
     /// used by the outer lambda. In

From f230015ed67169d41c322484ef1bac6817782073 Mon Sep 17 00:00:00 2001
From: LiaCastaneda <lia.castaneda@datadoghq.com>
Date: Fri, 7 Aug 2026 14:32:06 +0200
Subject: [PATCH 3/8] fix broken rustdoc intra-link to LambdaExpr

---
 datafusion/expr/src/higher_order_function.rs | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/datafusion/expr/src/higher_order_function.rs b/datafusion/expr/src/higher_order_function.rs
index a004bf2fdca2c..3a6fdb2ef5ba5 100644
--- a/datafusion/expr/src/higher_order_function.rs
+++ b/datafusion/expr/src/higher_order_function.rs
@@ -248,7 +248,7 @@ pub struct LambdaArgument {
     /// parameters leave no slot in the merged batch and the body's compressed
     /// column indices line up directly with what the evaluator built.
     ///
-    /// Callers who already have a [`LambdaExpr`] should read the used names
+    /// Callers who already have a `LambdaExpr` should read the used names
     /// from `LambdaExpr::used_params()` and pass them to [`Self::new`];
     /// [`Self::new`] handles the name → index translation.
     used_param_indices: Vec<usize>,

From db957ce7d8e27f13f0d40c21ac9922b86122841c Mon Sep 17 00:00:00 2001
From: LiaCastaneda <lia.castaneda@datadoghq.com>
Date: Tue, 11 Aug 2026 11:13:54 +0200
Subject: [PATCH 4/8] Address review comments

---
 datafusion/expr/src/higher_order_function.rs  | 130 ++++++++++++++++--
 .../physical-expr/src/expressions/lambda.rs   |  50 +++----
 2 files changed, 136 insertions(+), 44 deletions(-)

diff --git a/datafusion/expr/src/higher_order_function.rs b/datafusion/expr/src/higher_order_function.rs
index 3a6fdb2ef5ba5..b286e5508176f 100644
--- a/datafusion/expr/src/higher_order_function.rs
+++ b/datafusion/expr/src/higher_order_function.rs
@@ -270,18 +270,6 @@ pub struct LambdaArgument {
 }
 
 impl LambdaArgument {
-    /// Build a [`LambdaArgument`] for a lambda whose body references the
-    /// subset of `params` named in `used_params`.
-    ///
-    /// [`Self::evaluate`] only materialises the closures whose parameter name
-    /// appears in `used_params`, preserving the original declaration order of
-    /// `params`. Unused declared parameters therefore leave no slot in the
-    /// merged batch, so the body's compressed column indices line up directly
-    /// with the columns the evaluator built.
-    ///
-    /// Callers with a `LambdaExpr` in hand should pass `lambda.used_params()`;
-    /// that method already computes the exact set required here (with
-    /// nested-lambda shadow tracking).
     pub fn new(
         params: Vec<FieldRef>,
         body: Arc<dyn PhysicalExpr>,
@@ -323,6 +311,11 @@ impl LambdaArgument {
     /// `args` should evaluate to the value of each parameter
     /// of the correspondent lambda returned in [HigherOrderUDFImpl::lambda_parameters].
     ///
+    /// Only the closures in `args` for parameters the lambda body actually
+    /// references are called; closures for declared-but-unused parameters
+    /// are skipped entirely. Callers should not rely on every closure in
+    /// `args` being invoked.
+    ///
     /// `spread_captures` is responsible for transforming the captured column arrays
     /// so they align with the evaluation batch. Captures are snapshotted from the
     /// outer batch at construction time, giving one value per outer row, but the
@@ -1739,4 +1732,117 @@ mod tests {
             Some(Arc::new(Field::new(name, dt, nullable))),
         ))
     }
+
+    /// A physical expression that reads the column at a fixed index of the
+    /// batch it is evaluated against, for exercising [`LambdaArgument`]
+    /// directly without depending on `datafusion-physical-expr`.
+    #[derive(Debug, Eq, PartialEq, Hash)]
+    struct ColumnAt(usize);
+
+    impl std::fmt::Display for ColumnAt {
+        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+            write!(f, "column_at({})", self.0)
+        }
+    }
+
+    impl PhysicalExpr for ColumnAt {
+        fn evaluate(&self, batch: &RecordBatch) -> Result<ColumnarValue> {
+            Ok(ColumnarValue::Array(Arc::clone(batch.column(self.0))))
+        }
+
+        fn children(&self) -> Vec<&Arc<dyn PhysicalExpr>> {
+            vec![]
+        }
+
+        fn with_new_children(
+            self: Arc<Self>,
+            _children: Vec<Arc<dyn PhysicalExpr>>,
+        ) -> Result<Arc<dyn PhysicalExpr>> {
+            Ok(self)
+        }
+
+        fn fmt_sql(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+            write!(f, "{self}")
+        }
+    }
+
+    /// `(k, v) -> v` with only `v` used must push `v`'s array, not `k`'s.
+    #[test]
+    fn test_lambda_argument_evaluate_pushes_only_used_param() {
+        use arrow::array::Int32Array;
+
+        let k_field = Arc::new(Field::new("k", DataType::Int32, true));
+        let v_field = Arc::new(Field::new("v", DataType::Int32, true));
+
+        let body = Arc::new(ColumnAt(0)) as Arc<dyn PhysicalExpr>;
+        let used_params: HashSet<String> = ["v".to_string()].into_iter().collect();
+
+        let lambda_arg =
+            LambdaArgument::new(vec![k_field, v_field], body, None, &used_params);
+
+        let k_values: ArrayRef = Arc::new(Int32Array::from(vec![100, 200, 300]));
+        let v_values: ArrayRef = Arc::new(Int32Array::from(vec![1, 2, 3]));
+        let k_closure = || -> Result<ArrayRef> { Ok(Arc::clone(&k_values)) };
+        let v_closure = || -> Result<ArrayRef> { Ok(Arc::clone(&v_values)) };
+        let args: Vec<&dyn Fn() -> Result<ArrayRef>> = vec![&k_closure, &v_closure];
+
+        let result = lambda_arg
+            .evaluate(&args, |arrays| Ok(arrays.to_vec()))
+            .unwrap();
+        let ColumnarValue::Array(result) = result else {
+            unreachable!()
+        };
+
+        assert_eq!(
+            result.as_any().downcast_ref::<Int32Array>().unwrap(),
+            &Int32Array::from(vec![1, 2, 3]),
+            "body should read v's values, not k's"
+        );
+    }
+
+    /// Same as above, but with a capture occupying the leading slot.
+    #[test]
+    fn test_lambda_argument_evaluate_pushes_only_used_param_with_captures() {
+        use arrow::array::Int32Array;
+
+        let cap_field = Arc::new(Field::new("cap", DataType::Int32, true));
+        let k_field = Arc::new(Field::new("k", DataType::Int32, true));
+        let v_field = Arc::new(Field::new("v", DataType::Int32, true));
+
+        let body = Arc::new(ColumnAt(1)) as Arc<dyn PhysicalExpr>;
+        let used_params: HashSet<String> = ["v".to_string()].into_iter().collect();
+
+        let cap_values: ArrayRef = Arc::new(Int32Array::from(vec![9, 9, 9]));
+        let captures = RecordBatch::try_new(
+            Arc::new(Schema::new(vec![cap_field])),
+            vec![cap_values],
+        )
+        .unwrap();
+
+        let lambda_arg = LambdaArgument::new(
+            vec![k_field, v_field],
+            body,
+            Some(captures),
+            &used_params,
+        );
+
+        let k_values: ArrayRef = Arc::new(Int32Array::from(vec![100, 200, 300]));
+        let v_values: ArrayRef = Arc::new(Int32Array::from(vec![1, 2, 3]));
+        let k_closure = || -> Result<ArrayRef> { Ok(Arc::clone(&k_values)) };
+        let v_closure = || -> Result<ArrayRef> { Ok(Arc::clone(&v_values)) };
+        let args: Vec<&dyn Fn() -> Result<ArrayRef>> = vec![&k_closure, &v_closure];
+
+        let result = lambda_arg
+            .evaluate(&args, |arrays| Ok(arrays.to_vec()))
+            .unwrap();
+        let ColumnarValue::Array(result) = result else {
+            unreachable!()
+        };
+
+        assert_eq!(
+            result.as_any().downcast_ref::<Int32Array>().unwrap(),
+            &Int32Array::from(vec![1, 2, 3]),
+            "body should read v's values, not k's or the capture's"
+        );
+    }
 }
diff --git a/datafusion/physical-expr/src/expressions/lambda.rs b/datafusion/physical-expr/src/expressions/lambda.rs
index 08346c0bb0f7a..b86b823e68570 100644
--- a/datafusion/physical-expr/src/expressions/lambda.rs
+++ b/datafusion/physical-expr/src/expressions/lambda.rs
@@ -43,14 +43,6 @@ pub struct LambdaExpr {
     body: Arc<dyn PhysicalExpr>,
     projected_body: Arc<dyn PhysicalExpr>,
     projection: Vec<usize>,
-    /// Subset of `params` (by name) that the body actually references,
-    /// computed with nested-lambda shadow tracking. Empty when no parameter
-    /// is referenced by this lambda's own body.
-    ///
-    /// The higher-order function uses this to only evaluate and push the
-    /// parameters the body actually needs into the merged evaluation batch,
-    /// which keeps the body's compressed column indices aligned with the
-    /// batch layout produced at runtime.
     used_params: HashSet<String>,
 }
 
@@ -185,11 +177,7 @@ impl LambdaExpr {
     }
 
     /// Subset of [`params`](Self::params) (by name) that the body actually
-    /// references, taking nested-lambda shadowing into account. Used by the
-    /// higher-order function evaluator to skip evaluating/pushing parameters
-    /// the lambda body does not need, so that unused declared parameters do
-    /// not shift the merged batch's column positions out of sync with the
-    /// body's compressed indices.
+    /// references. See `CollectUsedVisitor` in this module.
     pub fn used_params(&self) -> &HashSet<String> {
         &self.used_params
     }
@@ -201,14 +189,21 @@ impl LambdaExpr {
 ///   anywhere in the tree (including inside nested lambdas). This drives
 ///   the `projection` used to slice the outer batch.
 /// * `used_param_names` — the subset of *this* lambda's `own_params` that
-///   the body actually references, with nested-lambda parameters shadowing
-///   the outer ones. For example, in
-///   `(k, v) -> func(col, (k, v2) -> k + v2 + v)` the inner `k` shadows the
-///   outer `k`, so only `v` flows up as used.
+///   the body actually references.
 ///
-/// The shadow stack uses `TreeNodeVisitor`'s `f_down` / `f_up` callbacks
-/// directly: push a frame when entering a nested [`LambdaExpr`], pop it
-/// when leaving.
+///   A nested lambda can declare its own parameter with the same name as
+///   one of `own_params` — a distinct variable that happens to reuse the
+///   name (variable shadowing). E.g. in
+///   `(k, v) -> func(col, (k, v2) -> k + v2 + v)`, the inner `k` is not
+///   `own_params`' `k`; only `v` should flow up as used, not `k`.
+///
+///   `shadow_stack` holds one frame per nested `LambdaExpr` currently being
+///   visited, each frame being that lambda's own parameter names. A
+///   `LambdaVariable` only counts toward `used_param_names` if its name
+///   isn't in any active frame (i.e. not shadowed).
+///
+/// The stack is maintained via `TreeNodeVisitor`'s `f_down` / `f_up`:
+/// push a frame when entering a nested [`LambdaExpr`], pop it when leaving.
 struct CollectUsedVisitor<'a> {
     own_params: &'a HashSet<String>,
     used_indices: HashSet<usize>,
@@ -388,10 +383,7 @@ mod tests {
         assert_eq!(used.len(), 1);
     }
 
-    /// A lambda whose body references neither declared parameter (e.g. a
-    /// constant expression) must report an empty used-params set. This
-    /// exercises the merged-batch path that has no parameter columns to
-    /// push at all.
+    /// A body that references neither declared parameter reports no used params.
     #[test]
     fn test_used_params_all_unused() {
         let body = Arc::new(NoOp::new());
@@ -403,9 +395,7 @@ mod tests {
         assert!(lambda.used_params().is_empty());
     }
 
-    /// A three-parameter lambda whose body skips the middle parameter must
-    /// report only the first and last as used, preserving declaration
-    /// order regardless of how [`Self::used_params`] (a set) reports them.
+    /// A three-parameter lambda that skips the middle parameter reports only the ends as used.
     #[test]
     fn test_used_params_three_params_middle_unused() {
         let a_field = Arc::new(Field::new("a", DataType::Int32, true));
@@ -429,11 +419,7 @@ mod tests {
         assert_eq!(used.len(), 2);
     }
 
-    /// A two-parameter lambda whose body references both parameters, but in
-    /// the reverse of their declared order (`v` before `k`), must still
-    /// report both as used. Declaration order (not reference order) is what
-    /// downstream code (`LambdaArgument::new`) relies on when it maps these
-    /// names back to positions in the merged batch.
+    /// Referencing params out of declaration order still reports both as used.
     #[test]
     fn test_used_params_both_used_in_reverse_reference_order() {
         let k_field = Arc::new(Field::new("k", DataType::Int32, true));

From 256e19e5cd9bb232a0a099fd6b9f5acfd102499a Mon Sep 17 00:00:00 2001
From: LiaCastaneda <lia.castaneda@datadoghq.com>
Date: Tue, 11 Aug 2026 11:48:26 +0200
Subject: [PATCH 5/8] track used params by index instead of name

---
 datafusion/expr/src/higher_order_function.rs  | 30 +++---------
 .../physical-expr/src/expressions/lambda.rs   | 49 +++++++++----------
 .../src/higher_order_function.rs              |  2 +-
 3 files changed, 32 insertions(+), 49 deletions(-)

diff --git a/datafusion/expr/src/higher_order_function.rs b/datafusion/expr/src/higher_order_function.rs
index b286e5508176f..dcf85ca1b2ed4 100644
--- a/datafusion/expr/src/higher_order_function.rs
+++ b/datafusion/expr/src/higher_order_function.rs
@@ -248,9 +248,9 @@ pub struct LambdaArgument {
     /// parameters leave no slot in the merged batch and the body's compressed
     /// column indices line up directly with what the evaluator built.
     ///
-    /// Callers who already have a `LambdaExpr` should read the used names
-    /// from `LambdaExpr::used_params()` and pass them to [`Self::new`];
-    /// [`Self::new`] handles the name → index translation.
+    /// Callers who already have a `LambdaExpr` should pass
+    /// `LambdaExpr::used_param_indices()` directly to [`Self::new`] — both
+    /// are indices into the same positionally-aligned `params` list.
     used_param_indices: Vec<usize>,
     /// The body of the lambda
     ///
@@ -274,15 +274,9 @@ impl LambdaArgument {
         params: Vec<FieldRef>,
         body: Arc<dyn PhysicalExpr>,
         captures: Option<RecordBatch>,
-        used_params: &HashSet<String>,
+        used_param_indices: &[usize],
     ) -> Self {
-        let used_param_indices: Vec<usize> = params
-            .iter()
-            .enumerate()
-            .filter(|(_, f)| used_params.contains(f.name()))
-            .map(|(i, _)| i)
-            .collect();
-
+        let used_param_indices = used_param_indices.to_vec();
         let effective_params = used_param_indices.iter().map(|i| Arc::clone(&params[*i]));
 
         let fields: Vec<FieldRef> = match &captures {
@@ -1775,10 +1769,7 @@ mod tests {
         let v_field = Arc::new(Field::new("v", DataType::Int32, true));
 
         let body = Arc::new(ColumnAt(0)) as Arc<dyn PhysicalExpr>;
-        let used_params: HashSet<String> = ["v".to_string()].into_iter().collect();
-
-        let lambda_arg =
-            LambdaArgument::new(vec![k_field, v_field], body, None, &used_params);
+        let lambda_arg = LambdaArgument::new(vec![k_field, v_field], body, None, &[1]);
 
         let k_values: ArrayRef = Arc::new(Int32Array::from(vec![100, 200, 300]));
         let v_values: ArrayRef = Arc::new(Int32Array::from(vec![1, 2, 3]));
@@ -1810,7 +1801,6 @@ mod tests {
         let v_field = Arc::new(Field::new("v", DataType::Int32, true));
 
         let body = Arc::new(ColumnAt(1)) as Arc<dyn PhysicalExpr>;
-        let used_params: HashSet<String> = ["v".to_string()].into_iter().collect();
 
         let cap_values: ArrayRef = Arc::new(Int32Array::from(vec![9, 9, 9]));
         let captures = RecordBatch::try_new(
@@ -1819,12 +1809,8 @@ mod tests {
         )
         .unwrap();
 
-        let lambda_arg = LambdaArgument::new(
-            vec![k_field, v_field],
-            body,
-            Some(captures),
-            &used_params,
-        );
+        let lambda_arg =
+            LambdaArgument::new(vec![k_field, v_field], body, Some(captures), &[1]);
 
         let k_values: ArrayRef = Arc::new(Int32Array::from(vec![100, 200, 300]));
         let v_values: ArrayRef = Arc::new(Int32Array::from(vec![1, 2, 3]));
diff --git a/datafusion/physical-expr/src/expressions/lambda.rs b/datafusion/physical-expr/src/expressions/lambda.rs
index b86b823e68570..9e5a4bffb1ab6 100644
--- a/datafusion/physical-expr/src/expressions/lambda.rs
+++ b/datafusion/physical-expr/src/expressions/lambda.rs
@@ -43,7 +43,7 @@ pub struct LambdaExpr {
     body: Arc<dyn PhysicalExpr>,
     projected_body: Arc<dyn PhysicalExpr>,
     projection: Vec<usize>,
-    used_params: HashSet<String>,
+    used_param_indices: Vec<usize>,
 }
 
 // Manually derive PartialEq and Hash to work around https://github.com/rust-lang/rust/issues/78808 [https://github.com/apache/datafusion/issues/13196]
@@ -128,12 +128,19 @@ impl LambdaExpr {
             .expect("closure should be infallible")
             .data;
 
+        let used_param_indices = params
+            .iter()
+            .enumerate()
+            .filter(|(_, name)| used_param_names.contains(*name))
+            .map(|(i, _)| i)
+            .collect();
+
         Self {
             params,
             body,
             projected_body,
             projection,
-            used_params: used_param_names,
+            used_param_indices,
         }
     }
 
@@ -176,10 +183,11 @@ impl LambdaExpr {
         &self.projected_body
     }
 
-    /// Subset of [`params`](Self::params) (by name) that the body actually
-    /// references. See `CollectUsedVisitor` in this module.
-    pub fn used_params(&self) -> &HashSet<String> {
-        &self.used_params
+    /// Indices into [`params`](Self::params) of the parameters the body
+    /// actually references, in declaration order. See `CollectUsedVisitor`
+    /// in this module.
+    pub fn used_param_indices(&self) -> &[usize] {
+        &self.used_param_indices
     }
 }
 
@@ -377,10 +385,7 @@ mod tests {
             LambdaExpr::try_new(vec!["k".to_string(), "v".to_string()], body).unwrap();
 
         assert_eq!(lambda.projection(), &[1]);
-        let used = lambda.used_params();
-        assert!(used.contains("v"));
-        assert!(!used.contains("k"));
-        assert_eq!(used.len(), 1);
+        assert_eq!(lambda.used_param_indices(), &[1]);
     }
 
     /// A body that references neither declared parameter reports no used params.
@@ -392,7 +397,7 @@ mod tests {
             LambdaExpr::try_new(vec!["k".to_string(), "v".to_string()], body).unwrap();
 
         assert!(lambda.projection().is_empty());
-        assert!(lambda.used_params().is_empty());
+        assert!(lambda.used_param_indices().is_empty());
     }
 
     /// A three-parameter lambda that skips the middle parameter reports only the ends as used.
@@ -412,11 +417,7 @@ mod tests {
         )
         .unwrap();
 
-        let used = lambda.used_params();
-        assert!(used.contains("a"));
-        assert!(!used.contains("b"));
-        assert!(used.contains("c"));
-        assert_eq!(used.len(), 2);
+        assert_eq!(lambda.used_param_indices(), &[0, 2]);
     }
 
     /// Referencing params out of declaration order still reports both as used.
@@ -434,10 +435,7 @@ mod tests {
             LambdaExpr::try_new(vec!["k".to_string(), "v".to_string()], body).unwrap();
 
         assert_eq!(lambda.projection(), &[0, 1]);
-        let used = lambda.used_params();
-        assert!(used.contains("k"));
-        assert!(used.contains("v"));
-        assert_eq!(used.len(), 2);
+        assert_eq!(lambda.used_param_indices(), &[0, 1]);
     }
 
     /// Inside a nested lambda that re-declares one of the outer parameter
@@ -486,12 +484,11 @@ mod tests {
             LambdaExpr::try_new(vec!["k".to_string(), "v".to_string()], outer_body)
                 .unwrap();
 
-        let used = outer_lambda.used_params();
-        assert!(used.contains("v"), "outer's `v` should be reported as used");
-        assert!(
-            !used.contains("k"),
-            "outer's `k` is shadowed inside the nested lambda and should not be reported as used"
+        assert_eq!(
+            outer_lambda.used_param_indices(),
+            &[1],
+            "only outer's `v` (index 1) should be reported as used; `k` (index 0) is \
+             shadowed inside the nested lambda"
         );
-        assert_eq!(used.len(), 1);
     }
 }
diff --git a/datafusion/physical-expr/src/higher_order_function.rs b/datafusion/physical-expr/src/higher_order_function.rs
index 79ff9c4e1cfea..5337227c94ece 100644
--- a/datafusion/physical-expr/src/higher_order_function.rs
+++ b/datafusion/physical-expr/src/higher_order_function.rs
@@ -353,7 +353,7 @@ impl PhysicalExpr for HigherOrderFunctionExpr {
                         } else {
                             Some(batch.project(&projection)?)
                         },
-                        lambda.used_params(),
+                        lambda.used_param_indices(),
                     )))
                 }
                 ArgSlot::Value => {

From 81abca06edaca07d6d8e2852710d8c4c657cc958 Mon Sep 17 00:00:00 2001
From: LiaCastaneda <lia.castaneda@datadoghq.com>
Date: Tue, 11 Aug 2026 11:58:06 +0200
Subject: [PATCH 6/8] fail instead of defaulting row count to 0

---
 datafusion/expr/src/higher_order_function.rs | 19 ++++++++++---------
 1 file changed, 10 insertions(+), 9 deletions(-)

diff --git a/datafusion/expr/src/higher_order_function.rs b/datafusion/expr/src/higher_order_function.rs
index dcf85ca1b2ed4..8dfbb36889b1c 100644
--- a/datafusion/expr/src/higher_order_function.rs
+++ b/datafusion/expr/src/higher_order_function.rs
@@ -411,15 +411,16 @@ fn merge_captures_with_variables(
     };
 
     if columns.is_empty() {
-        // Constant lambda body with no captures and no used parameters. We
-        // still need a row count for the merged batch, so evaluate one
-        // variable just to derive it. This is essentially free in the common
-        // case (the variables already exist as closures over arrays the
-        // caller computed up front).
-        let row_count = match variables.first() {
-            Some(first) => first()?.len(),
-            None => 0,
-        };
+        // No columns to derive a row count from, so borrow one variable's
+        // array length instead (all variables have the same length).
+        let row_count = variables
+            .first()
+            .ok_or_else(|| {
+                internal_datafusion_err!(
+                    "merge_captures_with_variables: no variables to derive a row count from"
+                )
+            })?()?
+        .len();
         return Ok(RecordBatch::try_new_with_options(
             schema,
             vec![],

From 204c88955a39ff002a30f1467742a2d5998868af Mon Sep 17 00:00:00 2001
From: LiaCastaneda <lia.castaneda@datadoghq.com>
Date: Tue, 11 Aug 2026 12:19:04 +0200
Subject: [PATCH 7/8] fmt

---
 datafusion/expr/src/higher_order_function.rs | 12 +++++-------
 1 file changed, 5 insertions(+), 7 deletions(-)

diff --git a/datafusion/expr/src/higher_order_function.rs b/datafusion/expr/src/higher_order_function.rs
index 8dfbb36889b1c..4474020138b2e 100644
--- a/datafusion/expr/src/higher_order_function.rs
+++ b/datafusion/expr/src/higher_order_function.rs
@@ -413,13 +413,11 @@ fn merge_captures_with_variables(
     if columns.is_empty() {
         // No columns to derive a row count from, so borrow one variable's
         // array length instead (all variables have the same length).
-        let row_count = variables
-            .first()
-            .ok_or_else(|| {
-                internal_datafusion_err!(
-                    "merge_captures_with_variables: no variables to derive a row count from"
-                )
-            })?()?
+        let row_count = variables.first().ok_or_else(|| {
+            internal_datafusion_err!(
+                "merge_captures_with_variables: no variables to derive a row count from"
+            )
+        })?()?
         .len();
         return Ok(RecordBatch::try_new_with_options(
             schema,

From 67b03000f79c6fefe7bb1832d12d51c7e2b41b35 Mon Sep 17 00:00:00 2001
From: LiaCastaneda <lia.castaneda@datadoghq.com>
Date: Tue, 11 Aug 2026 18:27:54 +0200
Subject: [PATCH 8/8] review comments: Add e2e test + debug_assert if indices
 are out of bounds

---
 datafusion/expr/src/higher_order_function.rs  | 21 +++++
 .../physical-expr/src/expressions/lambda.rs   |  4 +
 .../src/higher_order_function.rs              | 76 +++++++++++++++++--
 3 files changed, 96 insertions(+), 5 deletions(-)

diff --git a/datafusion/expr/src/higher_order_function.rs b/datafusion/expr/src/higher_order_function.rs
index 4474020138b2e..c300be8f6fcfe 100644
--- a/datafusion/expr/src/higher_order_function.rs
+++ b/datafusion/expr/src/higher_order_function.rs
@@ -251,6 +251,13 @@ pub struct LambdaArgument {
     /// Callers who already have a `LambdaExpr` should pass
     /// `LambdaExpr::used_param_indices()` directly to [`Self::new`] — both
     /// are indices into the same positionally-aligned `params` list.
+    ///
+    /// Every index here must be `< params.len()`; see the precondition on
+    /// [`Self::new`].
+    ///
+    /// Relies on captures sorting before this lambda's own params in the
+    /// planner's (un-projected) index space, which is what makes
+    /// `captures ++ used_params` below line up with the projected body.
     used_param_indices: Vec<usize>,
     /// The body of the lambda
     ///
@@ -270,12 +277,26 @@ pub struct LambdaArgument {
 }
 
 impl LambdaArgument {
+    /// # Preconditions
+    ///
+    /// Every index in `used_param_indices` must be `< params.len()`;
+    /// violating this panics on out-of-bounds indexing below. Callers should
+    /// pass `LambdaExpr::used_param_indices()`, which always indexes into the
+    /// same `params` list, rather than constructing indices by hand.
     pub fn new(
         params: Vec<FieldRef>,
         body: Arc<dyn PhysicalExpr>,
         captures: Option<RecordBatch>,
         used_param_indices: &[usize],
     ) -> Self {
+        debug_assert!(
+            used_param_indices.iter().all(|i| *i < params.len()),
+            "used_param_indices contains an index out of bounds for params \
+             (len {}): {:?}",
+            params.len(),
+            used_param_indices
+        );
+
         let used_param_indices = used_param_indices.to_vec();
         let effective_params = used_param_indices.iter().map(|i| Arc::clone(&params[*i]));
 
diff --git a/datafusion/physical-expr/src/expressions/lambda.rs b/datafusion/physical-expr/src/expressions/lambda.rs
index 9e5a4bffb1ab6..95bb5db0b328e 100644
--- a/datafusion/physical-expr/src/expressions/lambda.rs
+++ b/datafusion/physical-expr/src/expressions/lambda.rs
@@ -186,6 +186,10 @@ impl LambdaExpr {
     /// Indices into [`params`](Self::params) of the parameters the body
     /// actually references, in declaration order. See `CollectUsedVisitor`
     /// in this module.
+    ///
+    /// Relies on the planner appending each lambda's own params after
+    /// captures, matching the `captures ++ used_params` layout
+    /// `LambdaArgument::new` builds.
     pub fn used_param_indices(&self) -> &[usize] {
         &self.used_param_indices
     }
diff --git a/datafusion/physical-expr/src/higher_order_function.rs b/datafusion/physical-expr/src/higher_order_function.rs
index 5337227c94ece..e28b38bd7c8c1 100644
--- a/datafusion/physical-expr/src/higher_order_function.rs
+++ b/datafusion/physical-expr/src/higher_order_function.rs
@@ -510,15 +510,18 @@ mod tests {
 
     use super::*;
     use crate::HigherOrderFunctionExpr;
+    use crate::create_physical_expr;
     use crate::expressions::Column;
     use crate::expressions::NoOp;
     use crate::expressions::lambda;
     use crate::expressions::not;
-    use arrow::array::NullArray;
     use arrow::array::RecordBatchOptions;
+    use arrow::array::{ArrayRef, Int32Array};
     use arrow::datatypes::{DataType, Field, Schema};
     use datafusion_common::Result;
     use datafusion_common::assert_contains;
+    use datafusion_expr::execution_props::ExecutionProps;
+    use datafusion_expr::physical_planning_context::PhysicalPlanningContext;
     use datafusion_expr::{
         HigherOrderFunctionArgs, HigherOrderSignature, HigherOrderUDF, HigherOrderUDFImpl,
     };
@@ -546,9 +549,11 @@ mod tests {
             _step: usize,
             _fields: &[ValueOrLambda<FieldRef, Option<FieldRef>>],
         ) -> Result<LambdaParametersProgress> {
-            Ok(LambdaParametersProgress::Complete(vec![vec![Arc::new(
-                Field::new("", DataType::Null, true),
-            )]]))
+            // Offer two params; single-param lambdas just ignore the second.
+            Ok(LambdaParametersProgress::Complete(vec![vec![
+                Arc::new(Field::new("", DataType::Int32, true)),
+                Arc::new(Field::new("", DataType::Int32, true)),
+            ]]))
         }
 
         fn return_field_from_args(
@@ -568,7 +573,18 @@ mod tests {
         ) -> Result<ColumnarValue> {
             match &args.args[0] {
                 ValueOrLambda::Lambda(lambda) => lambda.evaluate(
-                    &[&|| Ok(Arc::new(NullArray::new(args.number_rows)))],
+                    &[
+                        // Sentinel for the first param, distinct from the second's value.
+                        &|| {
+                            Ok(Arc::new(Int32Array::from(vec![-1000; args.number_rows]))
+                                as ArrayRef)
+                        },
+                        &|| {
+                            Ok(Arc::new(Int32Array::from_iter_values(
+                                (0..args.number_rows as i32).map(|i| 10 * (i + 1)),
+                            )) as ArrayRef)
+                        },
+                    ],
                     |arrays| Ok(arrays.to_vec()),
                 ),
                 ValueOrLambda::Value(value) => Ok(value.clone()),
@@ -716,4 +732,54 @@ mod tests {
             "mock_function received a lambda via with_new_children at position 0 that wasn't a lambda before"
         );
     }
+
+    /// Exercises the real planner end to end (not hand-picked indices) to
+    /// check the "captures before own-params" layout invariant.
+    #[test]
+    fn test_higher_order_function_two_lambda_params_capture_and_unused_param() {
+        use datafusion_common::DFSchema;
+        use datafusion_expr::expr::{HigherOrderFunction, LambdaVariable};
+        use datafusion_expr::{Expr, col, lambda as logical_lambda};
+
+        let fun = Arc::new(HigherOrderUDF::new_from_impl(MockHigherOrderUDF {
+            signature: HigherOrderSignature::variadic_any(Volatility::Stable),
+        }));
+
+        // Body uses capture "a" and param "v"; param "k" is left unused.
+        let v = Expr::LambdaVariable(LambdaVariable::new(
+            "v".to_string(),
+            Some(Arc::new(Field::new("v", DataType::Int32, true))),
+        ));
+        let body = col("a") + v;
+        let lambda_expr = logical_lambda(["k", "v"], body);
+
+        let schema = DFSchema::from_unqualified_fields(
+            vec![Field::new("a", DataType::Int32, false)].into(),
+            std::collections::HashMap::new(),
+        )
+        .unwrap();
+
+        let physical_expr = create_physical_expr(
+            &Expr::HigherOrderFunction(HigherOrderFunction::new(fun, vec![lambda_expr])),
+            &schema,
+            &ExecutionProps::new(),
+            &PhysicalPlanningContext::default(),
+        )
+        .unwrap();
+
+        let batch = RecordBatch::try_new(
+            Arc::clone(schema.inner()),
+            vec![Arc::new(Int32Array::from(vec![1, 2, 3])) as ArrayRef],
+        )
+        .unwrap();
+
+        let result = physical_expr.evaluate(&batch).unwrap();
+        let ColumnarValue::Array(result) = result else {
+            unreachable!()
+        };
+
+        // a + v; k's sentinel (-1000) must not leak into the result.
+        let expected = Int32Array::from(vec![11, 22, 33]);
+        assert_eq!(result.as_ref(), &expected);
+    }
 }