[
https://issues.apache.org/jira/browse/SPARK-60730?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ASF GitHub Bot updated SPARK-60730:
-----------------------------------
Labels: pull-request-available (was: )
> 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
> Labels: pull-request-available
>
> 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]
