schenksj commented on PR #4827:
URL:
https://github.com/apache/datafusion-comet/pull/4827#issuecomment-6631504867
@andygrove, replying to the Aug 27 review:
- **Short-circuit.** With the root-only rule, a conditional that contains
the declined node goes
to the dispatcher whole, so `CASE`/`IF`/`AND`/`OR`/`COALESCE` run Spark's
own code. Gluten's
`ColumnarPartialProjectExec` takes the same line: it extracts the whole
`If`/`CaseWhen`/
`Coalesce`/`NaNvl` when it contains a UDF. I added ANSI `10 div x` tests
for CASE in a projection
and AND in a filter. To be clear about what they show: with the old
node-level hook put back,
these shapes and five others (OR, IF, COALESCE, NVL2, and IF/OR in a
filter) all still matched
Spark on current main, because the Rust conditionals there evaluate a
branch only on the rows that
reach it. So I couldn't reproduce a failure on current main. The tests pin
the result, and root-only
makes it hold without depending on that.
Evaluation order does matter in three places, all about which rows Spark's
whole-stage code
evaluates an expression for. None is about the values computed for a row.
- **Nondeterministic filter.** Whole-stage `FilterExec` evaluates each
`IsNotNull` only just
before the first conjunct that reads its column. So in `y IS NOT NULL
AND abs(rand(42)) < 0.5`,
`rand` runs for the rows with a NULL `y` too. On the test's 200 rows
Spark keeps 69, and 62
without whole-stage codegen, which evaluates the condition the way the
kernel does. The detour
now declines a nondeterministic filter condition. It walks the tree for
this, because
`Invoke.deterministic` ignores its target. There are tests for that
shape and for a Spark 4
variant hidden under `make_valid_utf8`. The same blind spot affects
projections: Spark defers
a projection item whose nondeterminism is hidden under an `Invoke`, as
it would any
deterministic item, so its `rand` advances only for the rows Spark
evaluates. The detour now
also declines a projection item that reports itself deterministic but
contains a
nondeterministic node, and there's a Spark 4.0+ test for that.
- **Subexpression elimination.** The kernel eliminated common
subexpressions, which whole-stage
`FilterExec` does not do in its predicate. Outside whole-stage codegen
it does, and Spark runs
a filter there when whole-stage codegen is off, when the condition has a
higher-order function
or other non-leaf `CodegenFallback`, or when it exceeds
`spark.sql.codegen.maxFields`. The
detour now mirrors `CollapseCodegenStages` per filter and compiles with
or without elimination
to match, with tests for both forms. Projection items follow Spark's
form too: under
`spark.sql.codegen.factoryMode=NO_CODEGEN` Spark interprets the
projection and evaluates a
common subexpression only on first use, so the kernel doesn't eliminate
them there. This
follows what Spark plans; a run-time fallback of Spark's whole-stage
code (a huge method, or a
compile failure) can't be seen at planning time and is documented as a
difference. Before this
change, the kernel's elimination moved the division in
`x <> 0 AND abs(10 div x) + abs(10 div x) > 1` in front of the `AND`, so
it raised under ANSI
for `x = 0` where whole-stage Spark returns rows; that case now matches.
Spark 4.2 also added
subexpression elimination to whole-stage `FilterExec` (SPARK-56032, from
4.2.0), interleaved
with the conjuncts, which the kernel can't reproduce; from 4.2 the
detour declines a filter
condition where that applies, with a reason and tests. `ProjectExec`
also eliminates
them across items, which a per-item kernel can't. So for two items `x <>
0 AND abs(10 div x) > 1`
and `x <> 0 AND abs(10 div x) > 2`, Spark raises for `x = 0` while the
detour returns rows, as
Comet's Rust projection already does. That's documented and pinned by a
test.
- **Errors on rows Spark skips, or doesn't skip.** This applies to any
expression that can
raise, not only ANSI arithmetic: with ANSI off, a `raise_error` in a
deferred item raises with
the detour where whole-stage Spark returns no rows, and there's a test
for that. The kernel
evaluates every item for every row of the batch. Whole-stage
`ProjectExec` defers deterministic items until something above
in the stage reads them, such as a filter, a broadcast hash join or a
`LIMIT`. So
`SELECT d FROM (SELECT abs(10 div x) AS d, rand(1) AS r FROM t) WHERE r
> 2` returns nothing in
Spark and can raise with the detour, and so can
`SELECT t.d FROM (SELECT x, abs(10 div x) AS d FROM t) t JOIN u ON t.x =
u.x`. Filters differ in both directions,
because whole-stage `FilterExec` places each `IsNotNull` itself while
the kernel goes left to
right, as Spark without whole-stage codegen does:
- `y IS NOT NULL AND abs(10 div x) > 0` with a row `(0, NULL)` raises in
whole-stage Spark,
while the detour and Comet's Rust filter return the other rows.
- `abs(10 div coalesce(x, 0)) > 0 AND x IS NOT NULL` with a NULL `x`
returns rows in
whole-stage Spark, while the detour raises.
I didn't try to reproduce Spark's evaluation order. The compatibility
guide now has a "JVM
expression detour" section describing these cases, and tests pin both
filter directions.
- **"Outermost" wording.** It's gone. The detour is now explicitly at the
root, and the whole-item
tests pin which tree gets dispatched.
- **Benchmark numbers.** I reran the benchmark on the final code and added
the shape you asked
about: the detoured projection (and, separately, the detoured filter)
under a `GROUP BY`
aggregate with a shuffle. Setup: AMD Ryzen 9 8945HX, Spark 4.1, release
build, `local[1]`, 16M
rows, `sqrt` sent through the detour by the testing config in both Comet
cases. Two runs, best
ms, run 1 / run 2:
| Shape | Spark | Detour off |
Detour on |
| ------------------------------------------ | --------- | ---------- |
--------- |
| one detoured item + 6 passthrough columns | 845 / 841 | 353 / 339 |
349 / 356 |
| 7 Rust-implemented items + 1 detoured item | 831 / 844 | 344 / 341 |
361 / 357 |
| filter | 716 / 700 | 379 / 364 |
338 / 324 |
| projection under `GROUP BY` | 516 / 512 | 310 / 329 |
282 / 276 |
| filter under `GROUP BY` | 562 / 570 | 292 / 295 |
295 / 290 |
Each case moved by at most 6% between runs. In the executed plans after
AQE, with the detour
off, the aggregates and the shuffle above the projection or filter fall
back to Spark as well,
above a `CometColumnarToRow` over the scan. With it on, every operator,
inside the shuffle
stage too, is a Comet operator. Your data-dependence point holds. The
detour is 9% to 16% faster for the
projection under `GROUP BY` and about 11% faster for the filter. It's
about the same for the
passthrough shape and for the filter under `GROUP BY`, and about 5% slower
for the mixed
projection. So keeping the plan in Comet doesn't by itself make these
queries faster; the gain depends on
the shape (the filter under `GROUP BY` keeps its aggregate and shuffle in
Comet and gains
nothing, while the plain filter gains 11%). The full table, the plans and
the machine
load are in the PR description.
- **Fallback rollup.** There are two tests. One checks that a detoured item
leaves no fallback
reason when the operator stays in Comet. The other uses `abs` detoured
plus `upper` disabled in
one projection: the operator falls back and reports only the `upper`
reason. The detour now
snapshots the tree's explain state (journal reasons, reason tags, coverage
tags) before the
conversion. It restores that snapshot before the detour, and merges the
conversion's state back
only if the detour is refused. A subtree shared with another item, as
`CollapseProject` can leave
it, therefore keeps that item's tags. There are tests for the
shared-subtree case and for a
refused detour.
- **Batch evaluation.** The config doc says the detour evaluates the item
for every row of the
batch, so an expression that can raise an error, under ANSI or
`raise_error` / `assert_true`, can
raise where Spark does not, or not raise where Spark does. It points at
the compatibility guide's
"JVM expression detour" section, which covers the deferred-item,
filter-ordering and cross-item
subexpression cases. There are tests that pin these, including a
`raise_error` deferred-item
case with ANSI off.
--
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]
