Method evaluate

datafusion/physical-expr/src/higher_order_function.rs:296–414

(&self, batch: &RecordBatch)

impl PhysicalExpr for HigherOrderFunctionExpr {
    fn evaluate(&self, batch: &RecordBatch) -> Result<ColumnarValue> {
        let mut arg_fields = Vec::with_capacity(self.args.len());
        let mut fields = Vec::with_capacity(self.args.len());
        for (arg, slot) in self.args.iter().zip(&self.slots) {
            match slot {
                ArgSlot::Lambda(lambda) => {
                    let field = lambda.body().return_field(batch.schema_ref())?;
                    arg_fields.push(ValueOrLambda::Lambda(Arc::clone(&field)));
                    fields.push(ValueOrLambda::Lambda(Some(field)));
                }
                ArgSlot::Value => {
                    let field = arg.return_field(batch.schema_ref())?;
                    arg_fields.push(ValueOrLambda::Value(Arc::clone(&field)));
                    fields.push(ValueOrLambda::Value(field));
                }
            }
        }

        let mut lambda_parameters = self.resolve_lambda_parameters(&fields)?.into_iter();

        let args = self
            .args
            .iter()
            .zip(&self.slots)
            .map(|(arg, slot)| match slot {
                ArgSlot::Lambda(lambda) => {
                    let lambda_params = lambda_parameters.next().ok_or_else(|| {
                        internal_datafusion_err!(
                            "params len should have been checked above"
                        )
                    })?;

                    if lambda.params().len() > lambda_params.len() {
                        return exec_err!(
                            "lambda defined {} params but higher-order function support only {}",
                            lambda.params().len(),
                            lambda_params.len()
                        );
                    }

                    let params = std::iter::zip(lambda.params(), lambda_params)
                        .map(|(name, param)| param.renamed(name.as_str()))
                        .collect();

                    // lambda.projection may include indexes of nested lambda variables not present on this batch
                    let projection = lambda
                        .projection()
                        .iter()
                        .copied()
                        .filter(|i| *i < batch.num_columns())
                        .collect::<Vec<_>>();

                    Ok(ValueOrLambda::Lambda(LambdaArgument::new(
                        params,
                        Arc::clone(lambda.projected_body()),
                        if projection.is_empty() {
                            None
                        } else {