refactor: extract LambdaExpr breaking change to apache#22853

Per maintainer request on PR apache#22689, the `LambdaExpr::try_new` /
`expressions::lambda(...)` signature change (adding `outer_columns_count`)
is being reviewed separately in apache#22853 because it's a
breaking change to the physical-expr public API and warrants its own
attention, distinct from this additive UDF.

This commit reverts the `lambda.rs` / `higher_order_function.rs` /
`planner.rs` files to their `upstream/main` state, removes the
sqllogictest file (every query in it uses `(k, v) -> body` lambdas that
require the upstream fix), and marks the unit tests that exercise
multi-parameter lambdas with
`#[ignore = "blocked on apache#22853: multi-param lambda projection fix"]`.

`transform_values_uses_keys_via_case` and
`transform_values_all_null_rows_returns_null_array` still pass because
the former references both `k` and `v` (so projection is a no-op) and
the latter short-circuits before evaluating the lambda. This PR will be
rebased onto main once apache#22853 merges, at which point the ignore
markers will be removed and the sqllogictest file restored.

Author: Adam-Alani

datafusion/functions-nested/src/transform_values.rs

@@ -337,6 +337,7 @@ mod tests {
     }
 
     #[test]
+    #[ignore = "blocked on apache/datafusion#22853: multi-param lambda projection fix"]
     fn transform_values_doubles_values() {
         let map = create_str_int_map(
             vec!["a", "b", "c"],
@@ -389,6 +390,7 @@ mod tests {
     }
 
     #[test]
+    #[ignore = "blocked on apache/datafusion#22853: multi-param lambda projection fix"]
     fn transform_values_preserves_null_rows() {
         let map = create_str_int_map(
             vec!["a", "b", "c", "d"],
@@ -422,6 +424,7 @@ mod tests {
     }
 
     #[test]
+    #[ignore = "blocked on apache/datafusion#22853: multi-param lambda projection fix"]
     fn transform_values_empty_map() {
         let map = create_str_int_map(
             vec![],
@@ -440,6 +443,7 @@ mod tests {
     }
 
     #[test]
+    #[ignore = "blocked on apache/datafusion#22853: multi-param lambda projection fix"]
     fn transform_values_mixed_empty_null_populated_rows() {
         // Row 0: empty {}, row 1: NULL, row 2: {a: 1, b: 2, c: 3}, row 3: {d: 4}
         let map = create_str_int_map(
@@ -474,6 +478,7 @@ mod tests {
     }
 
     #[test]
+    #[ignore = "blocked on apache/datafusion#22853: multi-param lambda projection fix"]
     fn transform_values_on_sliced_map_should_not_evaluate_on_unreachable_values() {
         // Build 4 map rows then slice off the first row. The value `0` lives in
         // the unreachable prefix; if the lambda `100/v` were evaluated on it we

datafusion/physical-expr/src/expressions/lambda.rs

@@ -43,11 +43,6 @@ pub struct LambdaExpr {
     body: Arc<dyn PhysicalExpr>,
     projected_body: Arc<dyn PhysicalExpr>,
     projection: Vec<usize>,
-    /// Number of columns in the outer input schema. Column/LambdaVariable
-    /// indices below this value reference outer captures; indices at or above
-    /// reference lambda parameters (whose position in the merged evaluation
-    /// batch is fixed by the higher-order function, not by the projection).
-    outer_columns_count: usize,
 }
 
 // Manually derive PartialEq and Hash to work around https://github.com/rust-lang/rust/issues/78808 [https://github.com/apache/datafusion/issues/13196]
@@ -65,19 +60,8 @@ impl Hash for LambdaExpr {
 }
 
 impl LambdaExpr {
-    /// Create a new lambda expression with the given parameters and body.
-    ///
-    /// `outer_columns_count` is the number of columns in the outer input
-    /// schema this lambda was planned against. Column/LambdaVariable indices
-    /// strictly below `outer_columns_count` reference outer captures and get
-    /// compressed to the front of the evaluation batch; indices at or above
-    /// reference lambda parameters and keep their fixed position relative to
-    /// the captures (so unused lambda parameters do not shift used ones).
-    pub fn try_new(
-        params: Vec<String>,
-        body: Arc<dyn PhysicalExpr>,
-        outer_columns_count: usize,
-    ) -> Result<Self> {
+    /// 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!(
                 "lambda params must be unique, got ({})",
@@ -87,14 +71,10 @@ impl LambdaExpr {
 
         check_async_udf(&body)?;
 
-        Ok(Self::new(params, body, outer_columns_count))
+        Ok(Self::new(params, body))
     }
 
-    fn new(
-        params: Vec<String>,
-        body: Arc<dyn PhysicalExpr>,
-        outer_columns_count: usize,
-    ) -> Self {
+    fn new(params: Vec<String>, body: Arc<dyn PhysicalExpr>) -> Self {
         let mut used_column_indices = HashSet::new();
 
         body.apply(|node| {
@@ -112,26 +92,10 @@ impl LambdaExpr {
 
         projection.sort();
 
-        // Captures (outer column refs) get compressed to the front of the
-        // merged batch. Lambda variables (indices >= outer_columns_count)
-        // keep their fixed offset from the start of the lambda parameter
-        // block, because the higher-order function always pushes all
-        // declared parameters into the merged batch in order.
-        let used_captures_count = projection
-            .iter()
-            .filter(|i| **i < outer_columns_count)
-            .count();
         let column_index_map = projection
             .iter()
             .enumerate()
-            .map(|(captures_pos, original)| {
-                let projected = if *original < outer_columns_count {
-                    captures_pos
-                } else {
-                    used_captures_count + (*original - outer_columns_count)
-                };
-                (*original, projected)
-            })
+            .map(|(projected, original)| (*original, projected))
             .collect::<HashMap<_, _>>();
 
         let projected_body = Arc::clone(&body)
@@ -165,7 +129,6 @@ impl LambdaExpr {
             body,
             projected_body,
             projection,
-            outer_columns_count,
         }
     }
 
@@ -224,31 +187,22 @@ impl PhysicalExpr for LambdaExpr {
 
         check_async_udf(body)?;
 
-        Ok(Arc::new(Self::new(
-            self.params.clone(),
-            Arc::clone(body),
-            self.outer_columns_count,
-        )))
+        Ok(Arc::new(Self::new(self.params.clone(), Arc::clone(body))))
     }
 
     fn fmt_sql(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         write!(f, "({}) -> {}", self.params.join(", "), self.body)
     }
 }
 
-/// Create a lambda expression.
-///
-/// `outer_columns_count` is the number of columns in the outer input schema
-/// this lambda was planned against. See [`LambdaExpr::try_new`] for details.
+/// Create a lambda expression
 pub fn lambda(
     params: impl IntoIterator<Item = impl Into<String>>,
     body: Arc<dyn PhysicalExpr>,
-    outer_columns_count: usize,
 ) -> Result<Arc<dyn PhysicalExpr>> {
     Ok(Arc::new(LambdaExpr::try_new(
         params.into_iter().map(Into::into).collect(),
         body,
-        outer_columns_count,
     )?))
 }
 
@@ -286,13 +240,13 @@ mod tests {
 
     #[test]
     fn test_lambda_evaluate() {
-        let lambda = lambda(["a"], Arc::new(NoOp::new()), 0).unwrap();
+        let lambda = lambda(["a"], Arc::new(NoOp::new())).unwrap();
         let batch = RecordBatch::new_empty(Arc::new(Schema::empty()));
         assert!(lambda.evaluate(&batch).is_err());
     }
 
     #[test]
     fn test_lambda_duplicate_name() {
-        assert!(lambda(["a", "a"], Arc::new(NoOp::new()), 0).is_err());
+        assert!(lambda(["a", "a"], Arc::new(NoOp::new())).is_err());
     }
 }

datafusion/physical-expr/src/higher_order_function.rs

@@ -628,7 +628,7 @@ mod tests {
 
         let hof = HigherOrderFunctionExpr::try_new_with_schema(
             fun,
-            vec![lambda(["a"], Arc::new(Literal::new(expected.clone())), 0).unwrap()],
+            vec![lambda(["a"], Arc::new(Literal::new(expected.clone()))).unwrap()],
             &Schema::empty(),
             Arc::new(ConfigOptions::new()),
         )
@@ -664,12 +664,10 @@ mod tests {
         let hof = HigherOrderFunctionExpr::try_new_with_schema(
             fun,
             vec![
-                not(lambda(
-                    ["a"],
-                    Arc::new(Literal::new(ScalarValue::Int32(Some(42)))),
-                    0,
+                not(
+                    lambda(["a"], Arc::new(Literal::new(ScalarValue::Int32(Some(42)))))
+                        .unwrap(),
                 )
-                .unwrap())
                 .unwrap(),
             ],
             &Schema::empty(),
@@ -709,7 +707,7 @@ mod tests {
         .unwrap();
 
         let result = Arc::new(hof)
-            .with_new_children(vec![lambda(["a"], Arc::new(NoOp::new()), 0).unwrap()])
+            .with_new_children(vec![lambda(["a"], Arc::new(NoOp::new())).unwrap()])
             .unwrap_err();
 
         assert_contains!(

datafusion/physical-expr/src/planner.rs

@@ -513,20 +513,10 @@ pub fn create_physical_expr(
                 config_options,
             )?))
         }
-        Expr::Lambda(Lambda { params, body }) => {
-            // The lambda was planned against a schema that appends the
-            // lambda's parameters to the outer input schema. Subtract
-            // `params.len()` to recover the outer column count, which the
-            // physical lambda needs to distinguish outer captures from
-            // lambda parameters during projection.
-            let outer_columns_count =
-                input_dfschema.fields().len().saturating_sub(params.len());
-            expressions::lambda(
-                params,
-                create_physical_expr(body, input_dfschema, execution_props)?,
-                outer_columns_count,
-            )
-        }
+        Expr::Lambda(Lambda { params, body }) => expressions::lambda(
+            params,
+            create_physical_expr(body, input_dfschema, execution_props)?,
+        ),
         Expr::LambdaVariable(LambdaVariable {
             name,
             field,