kazantsev-maksim commented on code in PR #4744:
URL: https://github.com/apache/datafusion-comet/pull/4744#discussion_r4048785857
##########
native/core/src/execution/planner.rs:
##########
@@ -3670,6 +3689,134 @@ impl PhysicalPlanner {
}
}
+ fn create_high_order_function_expr(
+ &self,
+ expr: &HigherOrderFunc,
+ input_schema: SchemaRef,
+ ) -> Result<Arc<dyn PhysicalExpr>, ExecutionError> {
+ let udf = create_comet_hof_func(&expr.func_name,
&self.session_ctx.state())?;
+
+ // 1. Plan value args.
+ let value_args: Vec<Arc<dyn PhysicalExpr>> = expr
+ .value_args
+ .iter()
+ .map(|e| self.create_expr(e, Arc::clone(&input_schema)))
+ .collect::<Result<_, _>>()?;
+
+ // 2. Resolve lambda param field types via the UDF (mirrors runtime).
+ let param_fields = Self::resolve_lambda_param_fields(
+ &udf,
+ &expr.func_name,
+ &value_args,
+ expr.lambdas.len(),
+ input_schema.as_ref(),
+ )?;
+
+ // 3. Plan lambdas with resolved param fields.
+ let lambdas: Vec<Arc<dyn PhysicalExpr>> = expr
+ .lambdas
+ .iter()
+ .zip(¶m_fields)
+ .map(|(l, fields)| self.create_lambda_expr(l, &input_schema,
fields))
+ .collect::<Result<_, _>>()?;
+
+ // 4. NOTE: assumes value args precede lambdas (holds for
array_filter).
+ let mut args = value_args;
+ args.extend(lambdas);
+
+ Ok(Arc::new(HigherOrderFunctionExpr::try_new_with_schema(
+ udf,
+ args,
+ &input_schema,
+ Arc::new(ConfigOptions::default()),
+ )?))
Review Comment:
Thanks for the detailed analysis and the reproduction case, @sunchao!
I completely agree that evaluating the predicate on zero elements breaks
Spark's short-circuit contract in ANSI mode. Looking closely at
`evaluate_single_list_lambda`, this seems to be an upstream DataFusion gap: it
already has an early return for `all_null`, but misses an early return when
`flattened.is_empty()`.
To avoid blocking this PR while we address it upstream, what do you think
about the following approach?
Instead of a heavy physical expression wrapper around
`HigherOrderFunctionExpr` (which would require manual array downcasting, offset
inspection, and null-mask tracking), we could add a minimal guard inside our
lambda wrapper (`PhysicalPlanner` / `lambda.rs`):
```rust
if batch.num_rows() == 0 {
return
Ok(ColumnarValue::Array(arrow::array::new_empty_array(self.data_type())));
}
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
