MaxGekk commented on code in PR #59985:
URL: https://github.com/apache/spark/pull/59985#discussion_r4466927685
##########
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:
Agreed, that was the conservative whitelist standing in for "can raise". In
376f3b2e355 `mayRaise` knows the total expressions on top of
`ExprUtils.canEvaluateUnconditionally`: `upper`, `lower`, `length`, the trims,
`substring`, `concat`, the hashes, the Bloom filter predicate of a runtime
filter, and the arithmetic of numbers in the legacy evaluation mode, as a rule
local to the filter codegen so that the join hoisting is not widened. So
`upper(name) = 'X'` no longer moves the checks before it and their columns are
loaded where they were, while a written `upper(s) IS NOT NULL` now moves before
a conjunct that can raise. Before a conjunct that can raise, the decode of the
checked column is the price of the fix; testing only the null bit there is
SPARK-60872. I am measuring the written-check shapes with the benchmark runs on
GitHub Actions, https://github.com/MaxGekk/spark/actions/runs/41822061208
(master) and https://github.com/MaxGekk/spark/actions/runs/41822065625 (this
branch),
and will post the numbers here.
##########
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:
Inferred checks keep the lazy placement by design. An inferred check is
implied by the conjunct it was inferred from, and generating it at its position
decodes its column on every row before any conjunct that can raise:
`upper(name) = 'X' AND d > 5` over 20M Parquet rows with `d DECIMAL(38, 10)`
went from 314 ms to 520 ms with that placement, and back to 316 ms with the
lazy one, and a first conjunct with a function is the common filter shape. A
written check on a total expression such as `upper(s) IS NOT NULL` now moves
(see the thread above), while a written check that can raise itself keeps the
lazy placement so that it does not raise on a row the conjunct rejects; the
migration guide names both as keeping the previous placement, together with the
inferred checks.
##########
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:
Good catch, thanks. In 376f3b2e355 `producedBy` resolves a checked column
through the operators of the stage, the aliases of a `ProjectExec` and the
result expressions of an aggregate, and the check is judged by that expression.
So `HAVING sum(a) / sum(b) IS NOT NULL AND max(y) + 1 > 0` keeps the lazy
placement and returns the second group without `DIVIDE_BY_ZERO` with codegen
on, and a filter kept above a nondeterministic projection with a `cast(s AS
INT)` alias still raises the overflow first; both are tests now. No codegen
interface change was needed: the plan walk stops at an operator that does not
defer its outputs, so a column read from a scan, a columnar batch or an
exchange stays movable.
##########
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:
Added in 376f3b2e355: the entry says that a nondeterministic conjunct such
as `rand(seed)` after such a check now runs only on the rows the check keeps,
and that a side-effecting UDF is no longer called on the rows it rejects, and
the PR description says the same. The entry also names the new config,
`spark.sql.legacy.filterLazyNotNullCheckPlacement`, internal and false by
default, which restores the previous placement, separate from the SPARK-60315
one as you argued.
##########
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:
Fixed in 376f3b2e355: the CSE branch invariant and the `INFERRED` scaladoc
now state the rule the code implements, with or without the flag, the checks
written in the query that cannot raise, inferred ones excluded, before a
conjunct that can raise.
##########
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:
Right. In 376f3b2e355 the case is `cast(s AS INT) IS NOT NULL AND y + 1 >
0`, with no earlier check that drops the row: codegen on raises
`ARITHMETIC_OVERFLOW`, the lazy placement, and codegen off raises
`CAST_INVALID_INPUT`. A moved cast check would fail it. The `x IS NOT NULL AND
cast(s AS INT) IS NOT NULL AND y + 1 > 0` case is gone.
--
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]
