Kousuke Saruta created SPARK-60731:
--------------------------------------
Summary: Do not compute an expensive alias ahead of the filter
conditions written before it when splitting a projection
Key: SPARK-60731
URL: https://issues.apache.org/jira/browse/SPARK-60731
Project: Spark
Issue Type: Bug
Components: SQL
Affects Versions: 4.4.0
Reporter: Kousuke Saruta
The projection split from SPARK-55014
(`spark.sql.optimizer.splitProjectionForExpensiveFilters`, default `true`)
orders the conditions of a `Filter` over a `Project` with expensive aliases by
how many not-yet-computed aliases they need, and computes the aliases of the
least demanding group in the lowest projection. So it can compute an expensive
alias, such as a UDF, ahead of the conditions written before the condition that
reads it, on rows those conditions would have filtered out. If the UDF fails on
such a row, the query fails.
Reproduction (spark-shell on master; `f` and `h` are no-op UDFs, `g` requires
`id > 5`):
{code:scala}
val f = udf((i: Long) => i)
val h = udf((i: Long) => i)
val g = udf((i: Long) => { require(i > 5); i })
spark.range(0, 10, 1, 1)
.select($"id", f($"id") as "f", h($"id") as "h", g($"id") as "g")
.where($"f" + $"h" > 10 && $"g" > 0)
.collect()
{code}
||whole-stage codegen||split on||split off||
|on (default)|FAILED_EXECUTE_UDF|OK (4 rows)|
|off|FAILED_EXECUTE_UDF|FAILED_EXECUTE_UDF|
The `g > 0` group needs one alias and `f + h > 10` needs two, so the split
computes `g` first, on every row. Without the split, whole-stage codegen
computes `g` only for the rows `f + h > 10` kept. With whole-stage codegen off,
the interpreted projection computes every alias on every row, so the query
fails with or without the split.
SPARK-60602 already keeps a condition over a throwable alias out of the split
for this reason, and SPARK-XXXXX keeps the conditions its check rejects out of
it, and holds the conditions written after them back in that pass. In later
passes the split orders those conditions by cost again, and throwable or
non-deterministic conditions do not hold the later ones back at all. Neither
can tell whether a UDF alias can fail, since `Expression.throwable` is opt-in
metadata that UDFs do not report. The same goes for a cheap part that can fail
inside an expensive alias, e.g. `udf(x) + 10 / (y - 3) AS f`, since the split
only checks the aliases for `throwable`.
Proposal: build the split in the order the conditions are written. A group is
split off into a lower projection only when every condition written before it
is already placed below it (or pushed below the projection), so neither a
condition nor the aliases it needs is evaluated ahead of the conditions written
before it. This keeps the main benefit of the split, an expensive alias only
computed for the rows the conditions before it kept, for filters written from
the cheapest to the most demanding condition. It gives up the cost-based
reordering for filters written the other way round.
Alternative: treat this as expected and document it. A `Project` computes its
expressions on all its input rows, and the interpreted path already fails here
without the split, so only the deferral that whole-stage codegen happens to do
keeps the query running. The docs would then say that a projected expression
may be evaluated on rows that a later filter drops. This would be at odds with
SPARK-60602 keeping throwable aliases out of the split, and leaves the
difference between the split on and off with whole-stage codegen on.
Since the split is unreleased, either choice changes nothing for released
versions.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
