@@ -262,25 +262,33 @@ impl LambdaArgument {
     /// `args` should evaluate to the value of each parameter
     /// of the correspondent lambda returned in [HigherOrderUDF::lambda_parameters].
     ///
-    /// `adjust` should adjust the captured columns of this
-    /// lambda, if any, relative to it's parameters
+    /// `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
+    /// function may evaluate the lambda body over a batch with a different number
+    /// of rows. It is the function responsibility to provide the appropriate `spread_captures`
+    /// closure to expand (or otherwise reshape) the captures to match.
+    /// Function working on lists, for example `array_transform(arr, v -> v + 1)`
+    /// flattens all list elements into a single batch, duplicating captured
+    /// values for rows with multiple elements and dropping them for empty lists.
+    /// If the lambda has no captures, `spread_captures` is never called.
     pub fn evaluate(
         &self,
         args: &[&dyn Fn() -> Result<ArrayRef>],
-        adjust: impl FnOnce(&[ArrayRef]) -> Result<Vec<ArrayRef>>,
+        spread_captures: impl FnOnce(&[ArrayRef]) -> Result<Vec<ArrayRef>>,
     ) -> Result<ColumnarValue> {
-        let adjusted_captures = self
+        let spread_captures = self
             .captures
             .as_ref()
             .map(|captures| {
-                let adjusted_columns = adjust(captures.columns())?;
+                let spread_columns = spread_captures(captures.columns())?;
 
-                RecordBatch::try_new(captures.schema(), adjusted_columns)
+                RecordBatch::try_new(captures.schema(), spread_columns)
             })
             .transpose()?;
 
         let merged = merge_captures_with_variables(
-            adjusted_captures.as_ref(),
+            spread_captures.as_ref(),
             Arc::clone(&self.schema),
             &self.params,
             args,
 
@@ -33,8 +33,9 @@ use datafusion_common::{
     internal_datafusion_err, internal_err,
 };
 use datafusion_expr::ColumnarValue;
-use indexmap::{IndexMap, IndexSet};
+use indexmap::IndexMap;
 use std::borrow::Cow;
+use std::collections::BTreeSet;
 use std::hash::Hash;
 use std::sync::Arc;
 
@@ -128,7 +129,8 @@ impl CaseBody {
     /// Derives a [ProjectedCaseBody] from this [CaseBody].
     fn project(&self) -> Result<ProjectedCaseBody> {
         // Determine the set of columns that are used in all the expressions of the case body.
-        let mut used_column_indices = IndexSet::<usize>::new();
+        // Use an ordered set so lambda variables continue to be positioned after columns
+        let mut used_column_indices = BTreeSet::<usize>::new();
         let mut collect_column_indices = |expr: &Arc<dyn PhysicalExpr>| {
             expr.apply(|expr| {
                 if let Some(column) = expr.downcast_ref::<Column>() {
 
@@ -124,6 +124,7 @@ impl PhysicalExpr for Column {
     }
 
     fn return_field(&self, input_schema: &Schema) -> Result<FieldRef> {
+        self.bounds_check(input_schema)?;
         Ok(input_schema.field(self.index).clone().into())
     }
 
 
@@ -107,7 +107,8 @@ impl PhysicalExpr for LambdaVariable {
 
         if self.field.as_ref() != batch.schema_ref().field(self.index) {
             return exec_err!(
-                "Physical LambdaVariable field doesn't match batch field during evaluation {} != {}",
+                "Field of physical LambdaVariable with index {} doesn't match batch field during evaluation {} != {}",
+                self.index,
                 self.field,
                 batch.schema_ref().field(self.index)
             );
 
@@ -568,7 +568,7 @@ mod tests {
             match &args.args[0] {
                 ValueOrLambda::Lambda(lambda) => lambda.evaluate(
                     &[&|| Ok(Arc::new(NullArray::new(args.number_rows)))],
-                    |_| unreachable!(),
+                    |arrays| Ok(arrays.to_vec()),
                 ),
                 ValueOrLambda::Value(value) => Ok(value.clone()),
             }
 
@@ -221,6 +221,40 @@ a [1, 50] 10 [[[11, 11]]]
 b [4, 50] 40 [[[48, 94]]]
 c [7, 50] 60 [[[74, 117]]]
 
+# case with inner nested higher order function where the condition is a lambda function
+query T?I?
+select
+    t.text,
+    t.list,
+    t.number,
+    case
+        when array_any_match(t.list, v -> v = 7) then array_transform(
+                [[t.list]],
+                list -> array_transform(
+                    list,
+                    list -> array_transform(
+                        list,
+                        v -> number + v + list[1]
+                    )
+                )
+            )
+        else array_transform(
+                [[t.list]],
+                list -> array_transform(
+                    list,
+                    list -> array_transform(
+                        list,
+                        v -> number + list[1]
+                    )
+                )
+            )
+    end
+from t
+order by t.number;
+----
+a [1, 50] 10 [[[11, 11]]]
+b [4, 50] 40 [[[44, 44]]]
+c [7, 50] 60 [[[74, 117]]]
 
 # array_transform coercion rules
 query TT
Original file line number	Diff line number	Diff line change
`@@ -124,6 +124,7 @@ impl PhysicalExpr for Column {`
`124`	`124`	`}`
`125`	`125`
`126`	`126`	`fn return_field(&self, input_schema: &Schema) -> Result<FieldRef> {`
	`127`	`+ self.bounds_check(input_schema)?;`
`127`	`128`	`Ok(input_schema.field(self.index).clone().into())`
`128`	`129`	`}`
`129`	`130`
Original file line number	Diff line number	Diff line change
`@@ -568,7 +568,7 @@ mod tests {`
`568`	`568`	`match &args.args[0] {`
`569`	`569`	`ValueOrLambda::Lambda(lambda) => lambda.evaluate(`
`570`	`570`	`&[&\|\| Ok(Arc::new(NullArray::new(args.number_rows)))],`
`571`		`- \|_\| unreachable!(),`
	`571`	`+ \|arrays\| Ok(arrays.to_vec()),`
`572`	`572`	`),`
`573`	`573`	`ValueOrLambda::Value(value) => Ok(value.clone()),`
`574`	`574`	`}`