spark-test commented on code in PR #59985:
URL: https://github.com/apache/spark/pull/59985#discussion_r4465352168
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala:
##########
@@ -256,22 +263,25 @@ trait GeneratePredicateHelper extends PredicateHelper {
k: Int,
conjunct: Expression,
afterNullable: Boolean,
- generated: Array[Boolean]): Seq[Expression] = {
- if (afterNullable) {
- val pending =
notNullPreds.take(precedingNotNulls(k)).zipWithIndex.filterNot {
- case (_, idx) => generated(idx)
- }
- if (pending.nonEmpty && (mayRaise(conjunct) || pending.exists(p =>
mayRaise(p._1)))) {
- pending.map { case (p, idx) =>
- generated(idx) = true
- p
- }
- } else {
- Nil
+ generated: Array[Boolean]): Seq[Int] = {
+ val pending =
notNullPreds.take(precedingNotNulls(k)).zipWithIndex.filterNot {
+ case (_, idx) => generated(idx)
+ }
+ val conjunctMayRaise = mayRaise(conjunct)
+ val positioned = if (afterNullable &&
+ (conjunctMayRaise || pending.exists(p => mayRaise(p._1)))) {
+ pending
+ } else if (conjunctMayRaise) {
+ pending.filter { case (p, _) =>
+ !mayRaise(p) &&
p.getTagValue(InferFiltersFromConstraints.INFERRED).isEmpty
Review Comment:
**Non-blocking (P2):** The fix stops at this filter: a check moves only when
its own expression is inside the `canEvaluateUnconditionally` whitelist and it
has no `INFERRED` tag. Every other preceding check keeps the lazy placement, so
with whole-stage codegen on, outside the SPARK-60315 path after a conjunct that
can be NULL, the query still fails on any row the check rejects and a later
conjunct raises on, while codegen off drops the row:
- Inferred checks, from both the Filter and the Join paths of
`InferFiltersFromConstraints`: `SELECT a FROM t WHERE y + 1 > 0 AND x > 0`
raises `ARITHMETIC_OVERFLOW` with codegen on and drops the row with codegen
off, as the new test asserts. This form needs no `IS NOT NULL` in the query, so
it is likely the most common one.
- Written checks on an expression that cannot raise but is outside the
whitelist, such as `upper(s) IS NOT NULL AND y + 1 > 0` on a row where `s` is
NULL. The migration-guide entry describes such a check, "written in the query
that cannot raise an error", as fixed.
- Written checks on an expression that can raise, such as `cast(s AS INT) IS
NOT NULL AND y + 1 > 0` on the same row, where the check is FALSE without
raising.
See **Shared repair plan 1** in the review body.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala:
##########
@@ -256,22 +263,25 @@ trait GeneratePredicateHelper extends PredicateHelper {
k: Int,
conjunct: Expression,
afterNullable: Boolean,
- generated: Array[Boolean]): Seq[Expression] = {
- if (afterNullable) {
- val pending =
notNullPreds.take(precedingNotNulls(k)).zipWithIndex.filterNot {
- case (_, idx) => generated(idx)
- }
- if (pending.nonEmpty && (mayRaise(conjunct) || pending.exists(p =>
mayRaise(p._1)))) {
- pending.map { case (p, idx) =>
- generated(idx) = true
- p
- }
- } else {
- Nil
+ generated: Array[Boolean]): Seq[Int] = {
+ val pending =
notNullPreds.take(precedingNotNulls(k)).zipWithIndex.filterNot {
+ case (_, idx) => generated(idx)
+ }
+ val conjunctMayRaise = mayRaise(conjunct)
+ val positioned = if (afterNullable &&
+ (conjunctMayRaise || pending.exists(p => mayRaise(p._1)))) {
+ pending
+ } else if (conjunctMayRaise) {
Review Comment:
**Blocking (P1):** This branch moves the pending written checks that cannot
raise ahead of any conjunct for which `mayRaise` holds, and `mayRaise` is the
conservative `canEvaluateUnconditionally` whitelist: it is also true for
deterministic conjuncts that can never raise, such as `upper(name) = 'X'`, `s
LIKE 'a%'`, `length(s) > 3`, integral arithmetic with
`spark.sql.ansi.enabled=false`, or the `might_contain` conjunct that the
runtime bloom filter appends by default to the filter of a large fact scan.
Ahead of such a conjunct, moving the check changes neither results nor errors,
but `genNotNull` emits the column's input variable at the new position, and for
a columnar scan that decodes the value along with the null bit.
For example, with whole-stage codegen on, `WHERE d IS NOT NULL AND
upper(name) = 'X' AND d > 5` over Parquet now decodes `d` on every row before
`upper(name) = 'X'`, while the base branch decoded it only on the rows where
`upper(name) = 'X'`. That is the placement you measured for the inferred check
(314 ms against 520 ms with `d DECIMAL(38, 10)`), now applied to written
checks, and the legacy flag does not turn it off. The extra decode happens in
`FilterExec` without subexpression elimination, and with it only for a checked
column that no other conjunct references; in an aggregate `FILTER` clause the
inputs are already loaded, so only the null test moves. Moving checks ahead of
conjuncts that can actually raise, or that are nondeterministic, is the
accepted cost of the fix and is not what this comment is about.
See **Shared repair plan 1** in the review body.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala:
##########
@@ -256,22 +263,25 @@ trait GeneratePredicateHelper extends PredicateHelper {
k: Int,
conjunct: Expression,
afterNullable: Boolean,
- generated: Array[Boolean]): Seq[Expression] = {
- if (afterNullable) {
- val pending =
notNullPreds.take(precedingNotNulls(k)).zipWithIndex.filterNot {
- case (_, idx) => generated(idx)
- }
- if (pending.nonEmpty && (mayRaise(conjunct) || pending.exists(p =>
mayRaise(p._1)))) {
- pending.map { case (p, idx) =>
- generated(idx) = true
- p
- }
- } else {
- Nil
+ generated: Array[Boolean]): Seq[Int] = {
+ val pending =
notNullPreds.take(precedingNotNulls(k)).zipWithIndex.filterNot {
+ case (_, idx) => generated(idx)
+ }
+ val conjunctMayRaise = mayRaise(conjunct)
+ val positioned = if (afterNullable &&
+ (conjunctMayRaise || pending.exists(p => mayRaise(p._1)))) {
+ pending
+ } else if (conjunctMayRaise) {
+ pending.filter { case (p, _) =>
+ !mayRaise(p) &&
p.getTagValue(InferFiltersFromConstraints.INFERRED).isEmpty
Review Comment:
**Non-blocking (P2):** `!mayRaise(p)` only looks at the check's own
expression, where a column is a plain attribute. Under whole-stage codegen,
though, that column can be deferred code of the child operator in the same
stage: `FilterExec.usedInputs` is empty, `ProjectExec` evaluates only its
nondeterministic outputs before `consume`, and the grouped `HashAggregateExec`
defers its deterministic result expressions. `genNotNull` then emits that
computation wherever the check is moved.
With ANSI on, `SELECT k, sum(a) / sum(b) AS r FROM t GROUP BY k HAVING r IS
NOT NULL AND max(y) + 1 > 0` now evaluates `sum(a) / sum(b)` before `max(y) + 1
> 0` and raises `DIVIDE_BY_ZERO` for a group with `sum(b) = 0` and `max(y) =
-5`, which the base branch dropped without error with codegen on. A filter kept
above a projection whose alias can raise behaves the same way, and an expensive
alias, such as a UDF kept above the projection so that it is not evaluated
twice, now runs on rows the conjunct rejects. With codegen off this query
already fails, because the aggregate computes `r` eagerly. But the PR keeps the
lazy placement for `cast(s AS INT) IS NOT NULL` precisely so that a check's
evaluation does not raise on a row the conjunct would reject, and a deferred
input gets around that rule.
**Recommended change:** Let the placement rule see how each input column is
produced. Give the codegen interface a way for the operator that produces
FilterExec's input variables in the same stage to report the expression behind
a lazily computed output: ProjectExec for its aliases, the aggregate operators
for their result expressions, and pass-through operators by forwarding to their
child. When deciding whether a check on such a column can move, treat it as a
check on that expression, so a computation that can raise moves only through
the columns it reads, or not at all. Add whole-stage codegen tests for a HAVING
clause and for a filter kept above a projection.
**Why this works:** Once the producing expression stands in for the column,
the same classification that keeps a raising check in its base-branch position
also applies to a raising projection or aggregate result. No computation that
can raise then runs before a conjunct that may reject the row, while plain
column reads keep moving as before.
**Scope:** Expose the producers of deferred outputs to the filter placement
rule, and use them when classifying checks.
**Compatibility:** Written checks on columns read directly from a scan still
drop rows before a later conjunct that can raise.
**Risks:** Operators that do not report their deferred outputs are treated
as possibly raising, so filters above them keep the base-branch placement and
the failure the PR fixes remains there.
**Constraints:** Inputs read straight from a scan, a columnar batch or a
stage boundary stay eligible to move.
**Success:** With ANSI on and whole-stage codegen on, SELECT k, sum(a) /
sum(b) AS r FROM t GROUP BY k HAVING r IS NOT NULL AND max(y) + 1 > 0 returns
without DIVIDE_BY_ZERO for a group whose sum(b) is 0 and whose max(y) is -5, as
on the base branch. A filter kept above a projection whose alias can raise does
not evaluate the alias before a later conjunct that rejects the row.
**Verification:**
- **Regression:** With ANSI on and whole-stage codegen on, the HAVING query
above, and a filter kept above a nondeterministic projection with an ANSI cast
alias, do not raise the computation's error on rows the later conjunct rejects.
- **Behavior:** Written checks on scan columns still move ahead of a raising
conjunct, so the PR's overflow examples still drop the NULL row with
whole-stage codegen on.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala:
##########
@@ -474,7 +484,8 @@ case class FilterExec(condition: Expression, child:
SparkPlan)
// got synthetic coverage) emit after all otherPreds. With the
NULL-conjunct
// flag, once an otherPred that can be NULL is emitted, the
notNullPreds that
// precede a later otherPred in the condition are emitted before it
first, when it
- // or one of them can raise (`precedingNotNullChecks`), and a complex
+ // or one of them can raise, and the plain-column ones are before an
otherPred that
Review Comment:
**Nit (P3):** This still describes the rule of the first commit, that "the
plain-column ones are before an otherPred that can raise in any case".
`precedingNotNullChecks` now moves every pending check that cannot raise and
has no `INFERRED` tag, so a written `st.x IS NOT NULL` moves and an inferred
`isnotnull(x)` does not, and it does so with or without the NULL-conjunct flag,
while this sentence sits inside the flag clause. Since the NOTE in
`generatePredicateCode` asks for both code paths to be kept in lockstep, a
maintainer who follows this invariant would bring the old rule back. The new
`INFERRED` scaladoc in `Optimizer.scala` is off in the same way: it says
codegen generates a written check "at its position", but that happens only
before a conjunct that may raise and only when the check cannot raise itself,
and inferred checks move too once the SPARK-60315 path applies.
**Verification:**
- **Inspection:** The CSE-branch invariant and every summary of the
placement rule name the same conditions and the same set of checks as the
helper's code, and none of them describes a plain-column-only rule.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/WholeStageCodegenSuite.scala:
##########
@@ -3156,6 +3156,71 @@ class WholeStageCodegenSuite extends SharedSparkSession
}
}
+ test("SPARK-60329: a filter checks a column IS NOT NULL before a conjunct
that can raise") {
+ withNullConjunctTable(Int.MaxValue) {
+ // `x` is NULL on the first row and `y + 1` overflows on it. With
codegen off, the FALSE
+ // `x IS NOT NULL` drops the row before the conjunct after it is
evaluated. With codegen
+ // on, the check was generated just before the first conjunct that
references `x`, or
+ // after all of them, so the overflow was raised. A check written in the
query that cannot
+ // raise is now generated before a conjunct that can raise, as with the
NULL-conjunct flag,
+ // whatever the legacy config, since it is independent of a NULL
conjunct. A check that the
+ // optimizer inferred keeps the lazy placement, so that its column is
not loaded before the
+ // conjunct it was inferred from.
+ //
+ // The common `b * 2` takes the CSE path of FilterExec.
+ val cseQuery = "SELECT a FROM t WHERE x IS NOT NULL AND y + 1 > 0 AND " +
+ "b * 2 > 0 AND b * 2 > -1"
+ def multiplyExactCount(filterExecCSE: Boolean): Int = {
+ withSQLConf(
+ SQLConf.ANSI_ENABLED.key -> "true",
+ SQLConf.SUBEXPRESSION_ELIMINATION_FILTER_EXEC_ENABLED.key ->
filterExecCSE.toString) {
+ val code = codegenString(sql(cseQuery).queryExecution.executedPlan)
+ "multiplyExact".r.findAllIn(code).length
+ }
+ }
+ assert(multiplyExactCount(filterExecCSE = true) <
multiplyExactCount(filterExecCSE = false))
+
+ Seq("true", "false").foreach { filterExecCSE =>
+ withNullConjunctConfs(
+ SQLConf.ANSI_ENABLED.key -> "true",
+ SQLConf.SUBEXPRESSION_ELIMINATION_FILTER_EXEC_ENABLED.key ->
filterExecCSE) { _ =>
+ // A check written in the query, with and without a later conjunct
that references `x`.
+ checkAnswer(sql("SELECT a FROM t WHERE x IS NOT NULL AND y + 1 > 0
AND x > 0"), Row(0))
+ checkAnswer(sql("SELECT a FROM t WHERE x IS NOT NULL AND y + 1 >
0"), Row(0))
+ // The same with a common subexpression.
+ checkAnswer(sql(cseQuery), Row(0))
+ // A check on an expression that cannot raise is generated at its
position too, while
+ // one that can raise, `cast(s AS INT) IS NOT NULL`, keeps the lazy
placement and is
+ // evaluated after the overflowing conjunct, on the second row only,
where it holds.
+ checkAnswer(sql("SELECT a FROM t WHERE st.x IS NOT NULL AND y + 1 >
0"), Row(0))
+ checkAnswer(
+ sql("SELECT a FROM t WHERE x IS NOT NULL AND cast(s AS INT) IS NOT
NULL AND " +
Review Comment:
**Nit (P3):** The written `x IS NOT NULL` before the cast check is moved
first and drops row 1, the only row on which the cast or `y + 1` could fail,
before either placement of the cast check runs; on row 2 the cast of `'1'`
succeeds. So this assertion holds whether or not the cast check keeps its lazy
placement: removing `!mayRaise(p) &&` from the new branch still returns
`Row(0)` here, and no other case in the suite covers a written check that can
raise ahead of a raising conjunct without the NULL-conjunct flag. The comment
above the case says more than the assertion verifies. On this table the two
placements differ only when no earlier check drops row 1: for `cast(s AS INT)
IS NOT NULL AND y + 1 > 0` with codegen on, the lazy placement raises
`ARITHMETIC_OVERFLOW`, and a moved cast check would raise `CAST_INVALID_INPUT`.
**Verification:**
- **Regression:** With ANSI on and both legacy-flag settings, a filter with
a written check on cast(s AS INT) before an overflowing conjunct, and no
earlier check that drops the row, raises the overflow error with whole-stage
codegen on and the cast error with it off.
##########
docs/sql-migration-guide.md:
##########
@@ -28,6 +28,7 @@ license: |
## Upgrading from Spark SQL 4.3 to 4.4
+- Since Spark 4.4, with whole-stage codegen enabled, a filter condition,
including the `FILTER` clause of an aggregate function, evaluates an `IS NOT
NULL` check written in the query that cannot raise an error, such as a check on
a column or on a struct field, before a later conjunct that can raise an error,
in the order of the condition, as Spark already did with whole-stage codegen
disabled. Previously, the check was evaluated just before the first conjunct
that references the column, or after all conjuncts, so a query such as `SELECT
a FROM t WHERE x IS NOT NULL AND y + 1 > 0` could fail with
`ARITHMETIC_OVERFLOW` in ANSI mode on a row where `x` is `NULL` and `y` is
`2147483647`, which is dropped with whole-stage codegen disabled. An `IS NOT
NULL` check that the optimizer infers from another conjunct, and a check on an
expression that can raise an error itself, such as `cast(s AS INT) IS NOT
NULL`, keep the previous placement, so a later conjunct can still raise an
error on a row
that such a check rejects; and a check written after a conjunct that
references its column is still evaluated before that conjunct.
Review Comment:
**Nit (P3):** The new placement also applies before every nondeterministic
or UDF conjunct, because `mayRaise` holds for them. With whole-stage codegen
on, `WHERE x IS NOT NULL AND rand(42) < 0.5` now evaluates `rand(42)` only on
rows where `x` is not NULL, so the same seed and data return a different sample
whenever some `x` is NULL, and a side-effecting UDF conjunct is no longer
called on those rows. That matches codegen off, but this entry only talks about
conjuncts that can raise an error, while the SPARK-60315 entry further down
documents the same kind of effect, on `rand(seed)` values and UDF calls, for
its own change.
**Verification:**
- **Inspection:** The migration-guide entry says that, under whole-stage
codegen, a nondeterministic conjunct such as rand(seed) and a side-effecting
user-defined function now run on a different set of rows.
--
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]
