[
https://issues.apache.org/jira/browse/SPARK-60730?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Kousuke Saruta updated SPARK-60730:
-----------------------------------
Description:
Since 4.2.0, `PushPredicateThroughNonJoin` can evaluate a filter condition
ahead of the conditions written before it, so a query that ran in 4.1 fails
with the default configs.
SPARK-47672 (`spark.sql.optimizer.avoidDoubleFilterEval`, default `true`)
splits a `Filter` over a `Project` with expensive aliases (UDFs, expensive
regexes) into cheap and expensive conditions. It pushes the cheap ones below
the projection and keeps the expensive ones above. A cheap condition written
after an expensive one is therefore evaluated first, on every row, including
the rows the expensive condition would have filtered out. If the cheap
condition can fail on those rows, the query fails. Before SPARK-47672, the
whole condition was pushed in its written order.
SPARK-60602 fixed this for `raise_error` / `assert_true`, using
`Expression.throwable`. But `throwable` is opt-in metadata that ANSI runtime
errors and user-defined functions do not report, so the same problem remains
for them. With ANSI mode on (the default in 4.x), arithmetic and casts are the
common case.
Reproduction (spark-shell, `f` is a no-op UDF so `f` is an expensive alias):
{code:scala}
val f = udf((i: Long) => i)
val df = spark.range(0, 10, 1, 1).select($"id", f($"id") as "f")
// 1. Division by zero on id = 3, which f > 5 filters out.
df.where($"f" > 5 && (lit(10L) / ($"id" - 3)) > 0).collect()
// 2. Invalid cast on the rows f > 5 filters out.
spark.range(0, 10, 1, 1)
.select($"id", f($"id") as "f", when($"id" > 5, lit("1")).otherwise(lit("x"))
as "s")
.where($"f" > 5 && $"s".cast("int") > 0).collect()
// 3. A UDF written directly in the filter (not through the projection) that
requires id > 5.
val h = udf((i: Long) => { require(i > 5); i })
df.where($"f" > 5 && h($"id") > 0).collect()
{code}
Results (ANSI on, avoidDoubleFilterEval at its default `true`):
||Case||4.1.4||4.2.0||master||
|1. divide|OK (4 rows)|DIVIDE_BY_ZERO|DIVIDE_BY_ZERO|
|2. cast|OK (4 rows)|CAST_INVALID_INPUT|CAST_INVALID_INPUT|
|3. UDF|OK (4 rows)|FAILED_EXECUTE_UDF|FAILED_EXECUTE_UDF|
All cases succeed with `spark.sql.optimizer.avoidDoubleFilterEval=false`, which
is the workaround, at the cost of evaluating the expensive aliases twice.
On master the projection split from SPARK-55014
(`spark.sql.optimizer.splitProjectionForExpensiveFilters`, unreleased) moves
conditions the same way: `f + g > 10 && 10 / (g - 3) > 0` with two expensive
aliases fails with DIVIDE_BY_ZERO when the split is on, and succeeds when it is
off.
The same reordering also changes which rows a non-deterministic condition sees.
`f >= 50 && rand(10) > 0.5` over `range(0, 100)` returns 31 rows in 4.1.4 and
29 rows (a different set) in 4.2.0 and on master.
Proposed fix:
Push the cheap conditions written before the first expensive condition as
today. After it, push only the conditions that
`ExprUtils.canEvaluateUnconditionally` accepts (deterministic attribute /
literal comparisons, `IN`, `IS NULL`, `AND` / `OR` / `NOT`, and the like),
which is the existing check for moving an expression out of a short-circuited
position (SPARK-58066). Apply the same check to the projection split, with the
expensive aliases read as columns and the first expensive condition exempt, so
the split does not place a rejected condition ahead of the conditions written
before it, and holds the conditions written after it back while it is not
placed. The check is an allowlist, so the filters the optimizer adds that can
not fail (dynamic partition pruning, runtime bloom filters, the `size(...) > 0`
from `InferFiltersFromGenerate` over a plain array) are accepted explicitly.
This subsumes the `throwable` check from SPARK-60602, and keeps pushing the
simple conditions that matter most for pruning and data source pushdown.
Alternatives considered:
- Push nothing after the first expensive condition: also correct, but loses the
pushdown of simple conditions such as `part = 1` written after a UDF condition.
- Mark the expressions that can fail under ANSI as `throwable`: does not cover
UDFs, and would stop the pushdown of most arithmetic and casts through joins as
well, since the join pushdown rules check `throwable` too.
Behavior change:
Conditions that the check rejects (e.g. `a + 1 > 5`, `upper(s) = 'X'`, casts)
are no longer pushed below the projection when written after an expensive
condition. A query that relies on such a condition, written after the expensive
one, to filter rows before the expensive one sees them succeeds in 4.2 but
fails after the fix, as it does in 4.1. For example, with `p` a UDF that
requires `id > 5`, `p > 0 && id > 5` fails in 4.1.4, succeeds in 4.2.0, and
still succeeds after the fix since `id > 5` is a simple comparison; `p > 0 &&
id + 1 > 6` fails after the fix, also with ANSI mode off, since the check is an
allowlist. A legacy config,
`spark.sql.legacy.pushFilterAheadOfExpensiveConjunct` (default `false`), pushes
such conditions again, except throwable ones such as `raise_error`
(SPARK-60602), and the migration guide will describe the change.
Backport:
This is a regression from SPARK-47672, which stated that results should not
change. The fix also changes behavior for the queries above, so whether it goes
to branch-4.3 / branch-4.2 is to be decided, either with the fix on by default
and the legacy config to restore 4.2.0, or with the fix off by default. The
`Filter` over `Project` code is the same in 4.2.0, 4.2.1 and 4.3.0.
`ExprUtils.canEvaluateUnconditionally` is not in branch-4.2 and would be added
there. The fix builds on SPARK-60602, whose backports to branch-4.3 and
branch-4.2 are in review, so it would go in after them. The projection split
parts would be left out, as in the SPARK-60602 backports.
This only covers the `Filter` over `Project` case. SPARK-60633 tracks a related
problem with throwable conditions in the generic `UnaryNode` case of the same
rule, which predates SPARK-47672 and is out of scope here.
> Do not push filter conditions that can fail ahead of the expensive conditions
> before them
> -----------------------------------------------------------------------------------------
>
> Key: SPARK-60730
> URL: https://issues.apache.org/jira/browse/SPARK-60730
> Project: Spark
> Issue Type: Bug
> Components: SQL
> Affects Versions: 4.2.0, 4.3.0, 4.2.1
> Reporter: Kousuke Saruta
> Priority: Major
>
> Since 4.2.0, `PushPredicateThroughNonJoin` can evaluate a filter condition
> ahead of the conditions written before it, so a query that ran in 4.1 fails
> with the default configs.
> SPARK-47672 (`spark.sql.optimizer.avoidDoubleFilterEval`, default `true`)
> splits a `Filter` over a `Project` with expensive aliases (UDFs, expensive
> regexes) into cheap and expensive conditions. It pushes the cheap ones below
> the projection and keeps the expensive ones above. A cheap condition written
> after an expensive one is therefore evaluated first, on every row, including
> the rows the expensive condition would have filtered out. If the cheap
> condition can fail on those rows, the query fails. Before SPARK-47672, the
> whole condition was pushed in its written order.
> SPARK-60602 fixed this for `raise_error` / `assert_true`, using
> `Expression.throwable`. But `throwable` is opt-in metadata that ANSI runtime
> errors and user-defined functions do not report, so the same problem remains
> for them. With ANSI mode on (the default in 4.x), arithmetic and casts are
> the common case.
> Reproduction (spark-shell, `f` is a no-op UDF so `f` is an expensive alias):
> {code:scala}
> val f = udf((i: Long) => i)
> val df = spark.range(0, 10, 1, 1).select($"id", f($"id") as "f")
> // 1. Division by zero on id = 3, which f > 5 filters out.
> df.where($"f" > 5 && (lit(10L) / ($"id" - 3)) > 0).collect()
> // 2. Invalid cast on the rows f > 5 filters out.
> spark.range(0, 10, 1, 1)
> .select($"id", f($"id") as "f", when($"id" > 5,
> lit("1")).otherwise(lit("x")) as "s")
> .where($"f" > 5 && $"s".cast("int") > 0).collect()
> // 3. A UDF written directly in the filter (not through the projection) that
> requires id > 5.
> val h = udf((i: Long) => { require(i > 5); i })
> df.where($"f" > 5 && h($"id") > 0).collect()
> {code}
> Results (ANSI on, avoidDoubleFilterEval at its default `true`):
> ||Case||4.1.4||4.2.0||master||
> |1. divide|OK (4 rows)|DIVIDE_BY_ZERO|DIVIDE_BY_ZERO|
> |2. cast|OK (4 rows)|CAST_INVALID_INPUT|CAST_INVALID_INPUT|
> |3. UDF|OK (4 rows)|FAILED_EXECUTE_UDF|FAILED_EXECUTE_UDF|
> All cases succeed with `spark.sql.optimizer.avoidDoubleFilterEval=false`,
> which is the workaround, at the cost of evaluating the expensive aliases
> twice.
> On master the projection split from SPARK-55014
> (`spark.sql.optimizer.splitProjectionForExpensiveFilters`, unreleased) moves
> conditions the same way: `f + g > 10 && 10 / (g - 3) > 0` with two expensive
> aliases fails with DIVIDE_BY_ZERO when the split is on, and succeeds when it
> is off.
> The same reordering also changes which rows a non-deterministic condition
> sees. `f >= 50 && rand(10) > 0.5` over `range(0, 100)` returns 31 rows in
> 4.1.4 and 29 rows (a different set) in 4.2.0 and on master.
> Proposed fix:
> Push the cheap conditions written before the first expensive condition as
> today. After it, push only the conditions that
> `ExprUtils.canEvaluateUnconditionally` accepts (deterministic attribute /
> literal comparisons, `IN`, `IS NULL`, `AND` / `OR` / `NOT`, and the like),
> which is the existing check for moving an expression out of a short-circuited
> position (SPARK-58066). Apply the same check to the projection split, with
> the expensive aliases read as columns and the first expensive condition
> exempt, so the split does not place a rejected condition ahead of the
> conditions written before it, and holds the conditions written after it back
> while it is not placed. The check is an allowlist, so the filters the
> optimizer adds that can not fail (dynamic partition pruning, runtime bloom
> filters, the `size(...) > 0` from `InferFiltersFromGenerate` over a plain
> array) are accepted explicitly. This subsumes the `throwable` check from
> SPARK-60602, and keeps pushing the simple conditions that matter most for
> pruning and data source pushdown.
> Alternatives considered:
> - Push nothing after the first expensive condition: also correct, but loses
> the pushdown of simple conditions such as `part = 1` written after a UDF
> condition.
> - Mark the expressions that can fail under ANSI as `throwable`: does not
> cover UDFs, and would stop the pushdown of most arithmetic and casts through
> joins as well, since the join pushdown rules check `throwable` too.
> Behavior change:
> Conditions that the check rejects (e.g. `a + 1 > 5`, `upper(s) = 'X'`, casts)
> are no longer pushed below the projection when written after an expensive
> condition. A query that relies on such a condition, written after the
> expensive one, to filter rows before the expensive one sees them succeeds in
> 4.2 but fails after the fix, as it does in 4.1. For example, with `p` a UDF
> that requires `id > 5`, `p > 0 && id > 5` fails in 4.1.4, succeeds in 4.2.0,
> and still succeeds after the fix since `id > 5` is a simple comparison; `p >
> 0 && id + 1 > 6` fails after the fix, also with ANSI mode off, since the
> check is an allowlist. A legacy config,
> `spark.sql.legacy.pushFilterAheadOfExpensiveConjunct` (default `false`),
> pushes such conditions again, except throwable ones such as `raise_error`
> (SPARK-60602), and the migration guide will describe the change.
> Backport:
> This is a regression from SPARK-47672, which stated that results should not
> change. The fix also changes behavior for the queries above, so whether it
> goes to branch-4.3 / branch-4.2 is to be decided, either with the fix on by
> default and the legacy config to restore 4.2.0, or with the fix off by
> default. The `Filter` over `Project` code is the same in 4.2.0, 4.2.1 and
> 4.3.0. `ExprUtils.canEvaluateUnconditionally` is not in branch-4.2 and would
> be added there. The fix builds on SPARK-60602, whose backports to branch-4.3
> and branch-4.2 are in review, so it would go in after them. The projection
> split parts would be left out, as in the SPARK-60602 backports.
> This only covers the `Filter` over `Project` case. SPARK-60633 tracks a
> related problem with throwable conditions in the generic `UnaryNode` case of
> the same rule, which predates SPARK-47672 and is out of scope here.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
