[andygrove]

Which issue does this PR close?

Closes #7321.

Rationale for this change

filter(arr, x -> ...) ran natively only for x IS NOT NULL. Any other lambda went through the JVM codegen dispatcher, which evaluates it one element at a time like Spark does, plus the transfer of each batch, so such queries ran at about Spark's speed. DataFusion 55 evaluates a lambda once over all the elements of a batch, so Comet can use it when the body is made of natively supported expressions.

Release build, Spark 4.1, 1M rows of ten-element arrays (every fifth element is NULL), best time:

Query	Spark	Comet before	Comet now
filter(strs, x -> nvl(x, '') != '')	616 ms	547 ms	307 ms
array_remove(array_compact(strs), '')	587 ms	252 ms	252 ms
filter(ints, x -> nvl(x, '') != '')	508 ms	456 ms	331 ms
filter(ints, x -> x IS NOT NULL)	286 ms	124 ms	125 ms
filter(ints, x -> x > threshold)	346 ms	335 ms	110 ms

What changes are included in this PR?

• A HigherOrderFunc and a LambdaVariable expression in the protobuf, and a planner arm that builds DataFusion's HigherOrderFunctionExpr with a LambdaExpr. The lambda body is planned against the input schema plus the lambda parameter, so captured columns keep their indexes.
• CometArrayFilter serializes the body natively when all of these hold: the lambda has one parameter, the array has a primitive, decimal, string, binary, date or timestamp element type, the body is deterministic, has no nested lambda and uses the default string collation, it reads no captured column that is not fixed-width, and every subexpression of it converts natively. Otherwise it keeps using the dispatcher, as before.
• The JVM codegen dispatcher now declines a tree that reads a lambda parameter bound outside the tree. Its kernel is fed only columns, so such a parameter would read as NULL. A body that needs the dispatcher therefore fails to convert natively and the whole filter stays on the dispatcher.
• DataFusion copies each column the lambda captures once per element of the array before it evaluates the body, and those copies are not reserved with the memory pool. A body that reads a captured string, binary, array, struct or map column stays on the dispatcher. Numeric, boolean, date, timestamp and decimal captures run natively.
• A body that uses AND, OR or IN stays on the dispatcher unless no expression in it can raise. Spark evaluates those lazily and the native operators evaluate every operand, so x != 0 AND 100 / x > 1 would otherwise raise on the elements Spark skips. This is the per-row case in Track evaluation masks for data-dependent errors beyond unbase64 operator fallback #6006. The check classifies each expression and fails closed: arithmetic and casts raise only in the ANSI eval mode they carry, and element_at raises for the index 0 in any mode, so an expression that is not known to be safe counts as able to raise. CASE, IF and COALESCE match Spark natively and are not restricted.
• NullShortCircuit evaluates the later arguments of functions such as array_contains on a projection of just the columns they read. It now projects and re-indexes a LambdaVariable like a Column, which filter(ints, x -> array_contains(other, x)) needs. case_when.rs treats a LambdaVariable like a Column when it decides whether an expression can fail.
• spark.comet.exec.nativeLambda.enabled (default true) turns the native path off.
• A new "Expression Tuning" page in the user guide covers lambdas: which forms run natively, the equivalent built-in array functions, and why nvl(x, '') != '' on a numeric array is slow. CometHigherOrderFunctionBenchmark produces the numbers above.

Not included: transform, exists and forall still use the dispatcher.

How are these changes tested?

• array_filter_native_lambda.sql runs about 45 queries against Spark with ANSI mode off: every supported element type including tinyint and smallint, NULL, empty and all-NULL arrays and batches, literal arrays, NaN and -0.0, captured columns, a parameter that shadows a column, nested filters, and bodies that must stay on the dispatcher. Each query pins its path with expect_native(filter) or expect_dispatch(filter).
• array_filter_native_lambda_ansi.sql covers the short-circuit queries under ANSI mode, and array_filter_native_lambda_captured.sql covers bodies that read captured columns of each kind and array functions that take the lambda parameter, with ANSI on and off. array_filter_native_lambda_disabled.sql checks that spark.comet.exec.nativeLambda.enabled=false runs every lambda on the dispatcher.
• Rust unit tests cover the planner arm (result, non-array argument, unknown function) and NullShortCircuit with a lambda variable in a masked argument. Without the fix, the second fails with LambdaVariable references column … but batch only has 0 columns.
• A one-off comparison against Spark on 300k rows, with 25 lambda bodies and ANSI on and off, matched in every case where Spark does not raise. The array SQL tests, CometArrayExpressionSuite, the codegen suites and the spark-expr Rust tests pass on Spark 4.1. The array_filter SQL files also pass on Spark 3.4 and 3.5, and the code compiles on 4.0.
• Spark's own SQL suite has not been run yet. The run-* labels on this PR will run it.

[andygrove]

cc @hsiang-c

[comphead]

DataFusion 55.2's HigherOrderFunctionExpr calls remove_list_null_values before it evaluates the body, and list_values slices from the first offset to the last, so values under a NULL array or outside a sliced range never reach the lambda. A NULL predicate drops the element and a NULL array stays NULL, as in Spark's ArrayFilter, and captured columns are re-indexed correctly. Three comments.

[comphead]

DataFusion's lambda copies every captured column once per element before evaluating the body: evaluate_single_list_lambda in datafusion-functions-nested 55.2 spreads the captures with take_arrays(arrays, &list_values_row_number(&original_list)?). For an INT capture such as threshold, the shape the benchmark measures, that is cheap. For a string, binary or array capture the copy grows with the value size. filter(a, x -> array_contains(b, x)) over 100-element INT arrays copies b 100 times per row, roughly 330 MB for one 8192-row batch by my arithmetic, and as far as I can tell those allocations are not reserved with the memory pool. Spark reads the captured value in place, and DuckDB's list_filter references captures through a selection vector. Could the native path decline when the body captures a column that is not fixed-width, or could the benchmark add a captured string or array case to show where it stops paying off? Not measured.

[andygrove]

Good catch, and it applies to more than array_contains. Any captured string, binary, array, struct or map column is copied once per element before the body runs. I haven't checked the memory-pool accounting myself, but declining the native path for these captures sidesteps the question.

nativeLambdaSupported now declines when the body reads a captured column that isn't fixed-width, so those lambdas stay on the dispatcher. Numeric, boolean, date, timestamp and decimal captures still run natively, which is the case the benchmark measures. I didn't add a captured-string benchmark case, because that case now takes the dispatcher path and would only measure the old behaviour. Fixed in a16bdba. The tuning page now says to capture only fixed-width columns.

[comphead]

Some natively supported expressions raise with ANSI off too. Spark's ElementAt throws INVALID_INDEX_OF_ZERO for index 0 outside the failOnError branch, and Comet's one_based_index in list_extract.rs does the same. So with spark.sql.ansi.enabled=false, filter(ints, x -> x > 0 AND element_at(other, x) > 1) takes the native path, where Spark never evaluates element_at(other, 0) because And stops at x > 0. DataFusion's BinaryExpr pre-selects rows for the right side only when the left side is at most 20% true (PRE_SELECTION_THRESHOLD), so for array(0, 1, 2) it evaluates element_at on every element and raises. I have not run it. Could the check classify each node instead, treating ElementAt as raising in any mode and judging arithmetic and casts by their own evalMode, with that query in the non-ANSI fixture?

[andygrove]

You're right. The ANSI check missed expressions that raise in every mode, and element_at with the index 0 is one.

The check no longer looks at the session's ANSI setting. It classifies each expression and fails closed. Arithmetic and casts count as able to raise only when they carry the ANSI eval mode, and anything not on the known-safe list, including element_at, counts as able to raise. A body that combines AND, OR or IN with such an expression stays on the dispatcher. Non-ANSI bodies like x > 0 AND x % 3 = 0 still run natively.

I added your case to the non-ANSI fixture as v > 0 AND element_at(array(threshold, 5), v) > 1. I used a fixed-width array instead of the other column from your example, because a captured array now takes the dispatcher for the reason in the first thread, which would hide this rule. It is pinned with expect_dispatch(filter) and matches Spark. Fixed in a16bdba.

[comphead]

A plain query passes whether the lambda ran natively or in the dispatcher, so this file cannot show which path a query took, and the false arm runs the same queries through the dispatcher a second time. sql-file-tests.md documents query expect_native(...) and query expect_dispatch(...) for a serde that splits between the two paths. With expect_native(filter) on the native queries and expect_dispatch(filter) on the RLIKE, nested-lambda and two-parameter ones, here and in _ansi.sql, the ConfigMatrix and the three new CometCodegenHOFSuite tests could go, since they assert the same split through dispatcher counters. isLambdaElementType also admits tinyint and smallint elements, but the table has no column of either type.

[andygrove]

Agreed, a plain query couldn't show which path ran. Every query in array_filter_native_lambda.sql, the _ansi file and the new captured-column file now carries expect_native(filter) or expect_dispatch(filter). The nativeLambda ConfigMatrix is gone. A small array_filter_native_lambda_disabled.sql checks that the kill switch sends everything to the dispatcher. The three new CometCodegenHOFSuite tests and the assertNoCodegen helper I had added are removed, so those two files match apache/main again.

I also added tinyint and smallint columns, since isLambdaElementType admits both and nothing exercised them. Fixed in a16bdba.

[andygrove]

Thanks for the review. All three are addressed in a16bdba, and the PR description is updated to match. The Spark SQL suite is still running through the run-* labels.