andygrove opened a new pull request, #7322: URL: https://github.com/apache/datafusion-comet/pull/7322
## 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, and nothing in it calls out to the JVM. Otherwise it keeps using the dispatcher, as before. - Under ANSI mode a body that uses `AND`, `OR` or `IN` stays on the dispatcher unless every expression in it cannot 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 #6006. `CASE`, `IF` and `COALESCE` match Spark natively and are not restricted. - `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 30 queries against Spark, with the native path on and off: every supported element type, NULL, empty and all-NULL arrays, NaN and -0.0, captured columns, a parameter that shadows a column, nested filters, and bodies that must stay on the dispatcher. - `array_filter_native_lambda_ansi.sql` covers the short-circuit queries under ANSI mode. - `CometCodegenHOFSuite` asserts that the dispatcher is idle when the lambda runs natively, and active for a body that calls out to the JVM, a nested lambda, a two-parameter lambda, and an ANSI short-circuit. - 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` and `CometCodegenHOFSuite` pass on Spark 4.1. The new SQL file also passes 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. -- 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]
