Run filter lambdas natively with DataFusion's array_filter #7321

What is the problem the feature request solves?

filter(arr, x -> ...) runs natively only when the lambda is x IS NOT NULL (what array_compact expands to). Every other lambda goes through the JVM codegen dispatcher: each batch is exported over FFI and Spark's interpreter evaluates the lambda one element at a time. That is the same per-element work Spark does, plus the transfer, so such a query runs at about Spark's speed, and a user reported it slower than Spark.

The user's workaround was to rewrite filter(arr, x -> nvl(x, '') != '') to array_remove(array_compact(arr), ''), which is native and much faster. Release build, Spark 4.1, 1M rows of ten-element string arrays:

Describe the potential solution

DataFusion 55 ships higher-order functions (array_filter, array_transform, array_any_match, array_first) that evaluate a lambda once over all the elements of a batch. Comet can serialize the lambda body as ordinary native expressions and plan it into DataFusion's array_filter, falling back to the dispatcher for bodies it cannot evaluate natively.

array_transform, exists and forall could follow. Spark's three-valued logic in exists needs checking first.

Related: the native AND, OR and IN do not short-circuit data-dependent errors under ANSI mode, so the first version has to keep such bodies on the dispatcher (see the comment on #6006).

Part of the effort tracked in #7300.