Revert "pull breaking changes from column capture draft PR"

This reverts commit 5b42e48.

Author: gstvg

datafusion/expr/src/higher_order_function.rs

@@ -22,7 +22,7 @@ use crate::{ColumnarValue, Documentation, Expr};
 use arrow::array::{ArrayRef, RecordBatch};
 use arrow::datatypes::{DataType, FieldRef, Schema};
 use datafusion_common::config::ConfigOptions;
-use datafusion_common::{Result, ScalarValue, exec_err, not_impl_err};
+use datafusion_common::{Result, ScalarValue, not_impl_err};
 use datafusion_expr_common::dyn_eq::{DynEq, DynHash};
 use datafusion_expr_common::signature::Volatility;
 use datafusion_physical_expr_common::physical_expr::PhysicalExpr;
@@ -208,45 +208,20 @@ pub struct LambdaArgument {
     /// For example, for `array_transform([2], v -> -v)`,
     /// this will be the physical expression of `-v`
     body: Arc<dyn PhysicalExpr>,
-    /// A RecordBatch with the captured columns inside the lambda body, if any
-    ///
-    /// For example, for `array_transform([2], v -> v + a + b)`,
-    /// this will be a `RecordBatch` with columns `a` and `b`
-    captures: Option<RecordBatch>,
 }
 
 impl LambdaArgument {
-    /// Create a new LambdaArgument
-    ///
-    /// Note that capture is not supported yet and must be `None` for now,
-    /// otherwise [LambdaArgument::evaluate] will fail
-    pub fn new(
-        params: Vec<FieldRef>,
-        body: Arc<dyn PhysicalExpr>,
-        captures: Option<RecordBatch>,
-    ) -> Self {
-        Self {
-            params,
-            body,
-            captures,
-        }
+    pub fn new(params: Vec<FieldRef>, body: Arc<dyn PhysicalExpr>) -> Self {
+        Self { params, body }
     }
 
     /// Evaluate this lambda
     /// `args` should evaluate to the value of each parameter
     /// of the correspondent lambda returned in [HigherOrderUDF::lambda_parameters].
-    ///
-    /// `adjust` should adjust the length of captured columns of this
-    /// lambda relative to it's parameters
     pub fn evaluate(
         &self,
         args: &[&dyn Fn() -> Result<ArrayRef>],
-        _adjust: impl FnOnce(&[ArrayRef]) -> Result<Vec<ArrayRef>>,
     ) -> Result<ColumnarValue> {
-        if self.captures.is_some() {
-            return exec_err!("lambda column capture is not supported yet");
-        }
-
         let columns = args
             .iter()
             .take(self.params.len())

datafusion/functions-nested/src/array_transform.rs

@@ -22,7 +22,7 @@ use arrow::{
     datatypes::{DataType, Field, FieldRef},
 };
 use datafusion_common::{
-    Result, ScalarValue, exec_err, not_impl_err, plan_err,
+    Result, ScalarValue, exec_err, plan_err,
     utils::{adjust_offsets_for_slice, list_values, take_function_args},
 };
 use datafusion_expr::{
@@ -203,11 +203,7 @@ impl HigherOrderUDF for ArrayTransform {
 
         // call the transforming lambda
         let transformed_values = lambda
-            .evaluate(&[&values_param], |_| {
-                not_impl_err!(
-                    "array_transform column capture support is not implemented yet"
-                )
-            })?
+            .evaluate(&[&values_param])?
             .into_array(list_values.len())?;
 
         let field = match args.return_field.data_type() {

datafusion/physical-expr/src/higher_order_function.rs

@@ -340,7 +340,6 @@ impl PhysicalExpr for HigherOrderFunctionExpr {
                     Ok(ValueOrLambda::Lambda(LambdaArgument::new(
                         params,
                         Arc::clone(lambda.body()),
-                        None,
                     )))
                 } else {
                     let value = arg.evaluate(batch)?;
@@ -536,10 +535,9 @@ mod tests {
             args: HigherOrderFunctionArgs,
         ) -> Result<ColumnarValue> {
             match &args.args[0] {
-                ValueOrLambda::Lambda(lambda) => lambda.evaluate(
-                    &[&|| Ok(Arc::new(NullArray::new(args.number_rows)))],
-                    |_| unimplemented!("mock_function"),
-                ),
+                ValueOrLambda::Lambda(lambda) => {
+                    lambda.evaluate(&[&|| Ok(Arc::new(NullArray::new(args.number_rows)))])
+                }
                 ValueOrLambda::Value(value) => Ok(value.clone()),
             }
         }