Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 7 additions & 5 deletions docs/source/contributor-guide/spark_configs_support.md
Original file line number Diff line number Diff line change
Expand Up @@ -62,11 +62,13 @@ least one result is `null`, `exists` returns `null` when the config is `true` an
is `false`. A matching element still returns `true`, while a null input array still returns
`null`, under either value.

**Comet status.** `CometArrayExists` routes Spark's `ArrayExists` expression through the JVM
codegen dispatcher. The dispatcher evaluates the Spark expression carrying the captured
`followThreeValuedLogic` value inside the Comet pipeline, so both config values preserve Spark's
semantics. If the dispatcher is disabled, the affected Comet operator falls back to Spark rather
than executing with different semantics.
**Comet status.** With the setting `true` (the default), `CometArrayExists` runs a lambda whose body Comet
supports natively under DataFusion's `array_any_match`, which has the same three-valued result: NULL when no
element matches but the predicate was NULL for some. With the setting `false`, and for every other lambda,
`CometArrayExists` routes Spark's `ArrayExists` expression through the JVM codegen dispatcher. The dispatcher
evaluates the Spark expression carrying the captured `followThreeValuedLogic` value inside the Comet pipeline, so
both config values preserve Spark's semantics. If the dispatcher is disabled, the affected Comet operator falls
back to Spark rather than executing with different semantics.

**Test coverage.** `spark/src/test/resources/sql-tests/expressions/array/exists.sql` uses a
`ConfigMatrix` to run both values and covers null arrays, empty arrays, matching predicates,
Expand Down
4 changes: 2 additions & 2 deletions docs/source/user-guide/latest/expressions.md
Original file line number Diff line number Diff line change
Expand Up @@ -387,8 +387,8 @@ to Spark ([#2837](https://github.com/apache/datafusion-comet/issues/2837)). Enab
| --- | --- | --- | --- |
| `aggregate` | ✅ | Codegen dispatch | |
| `array_sort` | ✅ | Native | The default comparator runs natively; custom comparators and arrays with non-default string collations use the JVM codegen dispatcher |
| `exists` | ✅ | Codegen dispatch | |
| `filter` | ✅ | Native | General lambda routed through the JVM codegen dispatcher; the `array_compact` form runs natively |
| `exists` | ✅ | Native | A one-parameter lambda whose body is natively supported and cannot raise runs natively; other lambdas, and `exists` with `spark.sql.legacy.followThreeValuedLogicInArrayExists=false`, use the JVM codegen dispatcher. See [Expression Tuning](tuning/expressions.md#lambdas-and-higher-order-functions) |
| `filter` | ✅ | Native | A one-parameter lambda whose body is natively supported runs natively, as does the `array_compact` form; other lambdas use the JVM codegen dispatcher. See [Expression Tuning](tuning/expressions.md#lambdas-and-higher-order-functions) |
| `forall` | ✅ | Codegen dispatch | |
| `map_filter` | ✅ | Codegen dispatch | |
| `map_zip_with` | ✅ | Codegen dispatch | |
Expand Down
1 change: 1 addition & 0 deletions docs/source/user-guide/latest/index.rst
Original file line number Diff line number Diff line change
Expand Up @@ -96,6 +96,7 @@ the navigation menu to read more.
Shuffle <tuning/shuffle>
Celeborn Shuffle <tuning/celeborn>
Operators <tuning/operators>
Expressions <tuning/expressions>
Row/Columnar Transitions <tuning/transitions>

.. toctree::
Expand Down
2 changes: 2 additions & 0 deletions docs/source/user-guide/latest/tuning.md
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,8 @@ guide is split into the following pages:
`spark.comet.scan.icebergNative.dataFileConcurrencyLimit`.
- [Operator Tuning](tuning/operators.md): joins, adaptive partial aggregation, and sorting on
floating-point values.
- [Expression Tuning](tuning/expressions.md): lambdas and higher-order functions such as `filter`, and the
built-in array functions that replace them.
- [Reducing Row/Columnar Conversion Overhead](tuning/transitions.md): stages in which many
operators fall back to Spark, and the experimental direct columnar-to-row converter.

Expand Down
104 changes: 104 additions & 0 deletions docs/source/user-guide/latest/tuning/expressions.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,104 @@
<!---
Licensed to the Apache Software Foundation (ASF) under one
or more contributor license agreements. See the NOTICE file
distributed with this work for additional information
regarding copyright ownership. The ASF licenses this file
to you under the Apache License, Version 2.0 (the
"License"); you may not use this file except in compliance
with the License. You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing,
software distributed under the License is distributed on an
"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
KIND, either express or implied. See the License for the
specific language governing permissions and limitations
under the License.
-->

# Expression Tuning

## Lambdas and Higher-Order Functions

A higher-order function takes a lambda: `filter(arr, x -> ...)`, `transform`, `exists`, `forall`, `aggregate`,
`zip_with` and `map_filter`. How fast Comet runs one depends on whether it can run the lambda natively.

| Lambda | How Comet runs it |
| -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `filter(arr, x -> x IS NOT NULL)` (what `array_compact` expands to), and `filter(arr, x -> nvl(x, '') != '')` on an array of numbers, dates or timestamps | Natively, as `array_compact`. |
| `filter` and `exists` with a one-parameter lambda whose body uses only natively supported expressions, over an array of primitive, decimal, string, binary, date or timestamp elements | Natively. The lambda runs once over all the elements of a batch, vectorized, instead of once per element. |
| Any other lambda, and every other higher-order function | Through the [JVM codegen dispatcher](../scala_java_udfs.md). Spark's interpreter evaluates the lambda for each element inside a Comet kernel, so expect Spark's speed, not more. |

`exists` runs natively under DataFusion's `array_any_match`, which returns what Spark's three-valued logic returns: NULL
when no element matches but the lambda was NULL for some element. With
`spark.sql.legacy.followThreeValuedLogicInArrayExists=false`, Spark returns false there, and `exists` stays on the
dispatcher.

The dispatcher has no advantage over Spark for a lambda, because Spark also evaluates a lambda one element at a time.
It adds the cost of handing each batch to the JVM and back. The rest of the query stays native either way.

To get the most out of Comet when a query uses lambdas:

- **Prefer a built-in array function to a lambda.** Most of them run natively with no per-element evaluation, for
example `array_compact`, `array_remove`, `array_distinct` and `array_position`. See the
[Supported Expressions](../expressions.md) tables for the caveats of each.
`array_compact(arr)` returns what `filter(arr, x -> x IS NOT NULL)` does for any array, and
`array_remove(array_compact(arr), '')` returns what `filter(arr, x -> nvl(x, '') != '')` does for an array of
strings with the default collation.
- **Do not test a numeric array for nulls with a string comparison.** With ANSI mode off, Spark casts every element
of `filter(int_arr, x -> nvl(x, '') != '')` to a string to compare it with `''`; with ANSI mode on the query does
not analyze. Comet recognizes exactly this form, with either operand first, and runs it as `array_compact`, which
never casts. A variation, such as another default value or an extra condition, runs as a general lambda and does
cast every element. Write `x IS NOT NULL`, or call `array_compact`.
- **Keep the lambda body to natively supported expressions.** A body that calls a regular expression function, a
UDF or another lambda cannot be split from the dispatcher, so the whole lambda runs there. Use
[`spark.comet.explain.codegen.enabled`](../understanding-comet-plans.md#sparkcometexplaincodegenenabled) to see which
expressions of a query went through the dispatcher.
- **Keep `AND`, `OR` and `IN` away from operations that can raise.** Spark evaluates them lazily, so
`x != 0 AND 100 / x > 1` never divides by zero, but the native lambda evaluates every operand for every element. A
body that combines them with an operation that can raise stays on the dispatcher. That covers a division, an
addition or a cast under ANSI mode, which raise only in that mode, and `element_at`, which raises for the index 0 in
any mode. `CASE WHEN`, `IF` and `COALESCE` are lazy natively too, so
`filter(arr, x -> CASE WHEN x = 0 THEN false ELSE 100 / x > 1 END)` runs natively.
- **Keep operations that can raise out of an `exists` lambda.** Spark stops at the first element that matches, so
under ANSI mode `exists(array(1, 0), x -> 100 / x > 1)` returns true without dividing by zero, but the native lambda
evaluates every element. An `exists` lambda that contains any operation that can raise stays on the dispatcher,
whether or not it uses `AND`, `OR`, `IN` or `CASE WHEN`.
- **Keep date and timestamp conversions out of the lambda.** A cast between a date or timestamp and a string, date or
timestamp fails natively for years beyond about 262,143, which Spark supports up to year 294,247
([#7471](https://github.com/apache/datafusion-comet/issues/7471)), so a lambda that contains one stays on the
dispatcher. So does a lambda with a `try_cast` that can make a `NULL` map key.
- **Capture only numeric, decimal, boolean, date or timestamp columns.** DataFusion copies each column the lambda reads
once per element of the array before it evaluates the lambda, so a captured string, binary, array, struct or map
column is copied as many times as the array has elements. A lambda that reads one stays on the dispatcher.
- **Use a lambda of one parameter.** `filter(arr, (x, i) -> i > 0)`, which also receives the element's index, runs
through the dispatcher.

`spark.comet.exec.nativeLambda.enabled=false` makes Comet run every lambda through the dispatcher.

### Example

`CometHigherOrderFunctionBenchmark` runs `filter` over 1 million rows of ten-element arrays, in which every fifth element is
`NULL`. These are the best times of the Spark baseline and of Comet on an Apple M-series laptop, with
`spark.comet.exec.nativeLambda.enabled` off and at its default:

| Query | Spark | Comet, nativeLambda off | Comet, default |
| ----------------------------------------------------------- | ------ | ----------------------- | -------------- |
| `filter(strs, x -> nvl(x, '') != '')` (string array) | 616 ms | 547 ms | 307 ms |
| `array_remove(array_compact(strs), '')` (same result) | 587 ms | 252 ms | 252 ms |
| `filter(ints, x -> x IS NOT NULL)` | 286 ms | 124 ms | 125 ms |
| `filter(ints, x -> x > threshold)`, `threshold` is a column | 346 ms | 335 ms | 110 ms |

The same benchmark runs `exists` over 1 million rows of ten-element arrays, in which every fifth element is `NULL`
(Spark 3.5, same laptop):

| Query | Spark | Comet, nativeLambda off | Comet, default |
| ----------------------------------------------------------- | ------ | ----------------------- | -------------- |
| `exists(ints, x -> x > 90)` | 266 ms | 234 ms | 108 ms |
| `exists(dbls, x -> x IS NULL OR x < 0 OR x > 90 OR x = 0)` | 266 ms | 225 ms | 94 ms |
| `exists(ints, x -> x > threshold)`, `threshold` is a column | 251 ms | 196 ms | 105 ms |
| `size(dbls) < 1 OR exists(dbls, x -> x > 90)` | 355 ms | 403 ms | 110 ms |
| `exists(strs, x -> x IS NULL OR x = '')` | 357 ms | 183 ms | 168 ms |

Rewriting a lambda to built-in functions is still the fastest form, but a native lambda gets most of the way there.
Loading
Loading