MaxGekk commented on code in PR #59573:
URL: https://github.com/apache/spark/pull/59573#discussion_r4310689727
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala:
##########
@@ -148,6 +148,52 @@ trait GeneratePredicateHelper extends PredicateHelper {
nonNullAttrExprIds)
}
+ /**
+ * Returns a fresh name for the flag that records a conjunct evaluated to
NULL, or None if the
+ * row is dropped at the first NULL conjunct (the legacy behavior) or none
of `conjuncts`, bound
+ * with the nullability they are generated with, can be NULL.
+ *
+ * A row is kept only when the whole condition is TRUE, so both FALSE and
NULL drop it. Only
+ * a FALSE conjunct decides the result of the AND on its own, though: after
a NULL one, the
+ * remaining conjuncts must still be evaluated, as in `And.eval`, so that an
error they raise
+ * on the row is raised (SPARK-60315). The flag drops the row once all of
them are evaluated.
+ */
+ protected def nullConjunctFlag(
+ ctx: CodegenContext,
+ conjuncts: Seq[Expression]): Option[String] = {
+ if (conf.legacyFilterShortCircuitOnNullConjunct ||
!conjuncts.exists(_.nullable)) {
+ None
+ } else {
+ Some(ctx.freshName("nullConjunct"))
+ }
+ }
+
+ /**
+ * Generates the check that drops the row when the conjunct evaluated into
`ev` is FALSE, or
+ * when it is NULL and `nullFlag` is None. With `nullFlag`, a NULL conjunct
sets the flag.
+ */
+ protected def conjunctCheck(ev: ExprCode, nullable: Boolean, nullFlag:
Option[String]): String = {
+ (nullable, nullFlag) match {
+ case (false, _) => s"if (!${ev.value}) continue;"
+ case (true, None) => s"if (${ev.isNull} || !${ev.value}) continue;"
+ case (true, Some(flag)) =>
+ s"if (${ev.isNull}) { $flag = true; } else if (!${ev.value}) continue;"
Review Comment:
Good catch, thanks. I reproduced it on 69318f6 with a table `(a, b, y, x) =
(NULL, 1, 2147483647, NULL), (0, 1, 0, 1)` and ANSI on:
| condition | codegen on, legacy off | codegen on, legacy on | codegen off |
|---|---|---|---|
| `NOT (a > 0 AND b > 0) AND y + 1 > 0 AND x > 0` | ARITHMETIC_OVERFLOW |
`[0]` | `[0]` |
| `x IS NOT NULL AND NOT (a > 0 AND b > 0) AND y + 1 > 0` |
ARITHMETIC_OVERFLOW | `[0]` | `[0]` |
| `NOT (a > 0 AND b > 0) AND y + 1 > 0 AND x IS NOT NULL` |
ARITHMETIC_OVERFLOW | `[0]` | ARITHMETIC_OVERFLOW |
| the guard-first condition in an aggregate `FILTER` | ARITHMETIC_OVERFLOW |
`[1]` | `[1]` |
So the first two and the `FILTER` clause are exactly the regression you
describe, and the third confirms that a guard written after the overflowing
conjunct has to stay there.
One more data point. The lazy placement of the `IsNotNull` checks already
diverges from the interpreted path without any NULL conjunct:
```sql
SELECT a FROM t WHERE x IS NOT NULL AND y + 1 > 0 AND x > 0
```
With codegen on this raises ARITHMETIC_OVERFLOW with the legacy flag on and
off, i.e. on master too, while codegen off returns `[0]`: `isnotnull(x)` is
generated just before `x > 0`, after `y + 1 > 0`. This PR did not create that
class of divergence, it widened it to the shapes that used to stop at the NULL
conjunct. I filed SPARK-60329 for it and mention it in the description.
Fixed in 3264b00d897 the way you suggest. `splitNotNullPreds` keeps how many
`IsNotNull` checks precede each other conjunct in the condition, and once a
conjunct that can be NULL has been generated, `precedingNotNullChecks` returns
the pending checks that precede the next conjunct, which are generated before
it, ahead of its column loads and CSE precomputes, in both
`generatePredicateCode` and the CSE branch. The conjuncts before the first one
that can be NULL are never reached after a NULL conjunct, so the checks before
them keep the lazy placement, and the generated code is unchanged with the
legacy flag or when no conjunct can be NULL.
One difference from your sketch: the checks on expressions, such as
`isnotnull(cast(s AS INT))`, are placed the same way, not only the plain-column
ones. At its own position in the condition the check runs on a subset of the
rows the interpreted path evaluates it on (the lazily placed checks can only
drop a row earlier), so its inner expression cannot raise where codegen off
does not. Leaving it lazy keeps the inconsistency you describe for that shape,
and the migration guide's "in the order they appear" would be false for it.
Tests added for the inferred check, a user-written check first, a
user-written check after the overflowing conjunct, a check on an expression, a
NULL conjunct from `OR`, and the aggregate `FILTER` clause, in the non-CSE and
CSE paths. A note on the CSE test: its raising conjunct is not the common
subexpression, because the interpreted predicate evaluates its common
subexpressions up front, which is a separate pre-existing difference.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/WholeStageCodegenSuite.scala:
##########
@@ -2814,4 +2815,144 @@ class WholeStageCodegenSuite extends SharedSparkSession
}
}
}
+
+ // A Parquet table, so that the filter is not evaluated by the optimizer on
a local relation.
+ // On the first row `NOT (a > 0 AND b > 0)` is NULL, and no `isnotnull(a)`
is inferred from it.
+ // The second row passes every condition of the tests below, except `b < 0`.
The rows can be
+ // written to different files, so the filter is not pushed down, which would
skip the first
+ // row's file by its statistics.
+ private def withNullConjunctTable(y: Int)(f: => Unit): Unit = {
+ withTable("t") {
+ sql("CREATE TABLE t USING parquet AS " +
+ s"SELECT * FROM VALUES (NULL, 1, $y), (0, 1, 0) AS v(a, b, y)")
+ f
+ }
+ }
+
+ // Runs `f` with whole-stage codegen on and off and with the legacy behavior
on and off, and
+ // passes it whether the filter drops a row at its first NULL conjunct.
+ private def withNullConjunctConfs(otherConfs: (String, String)*)(f: Boolean
=> Unit): Unit = {
+ for (codegen <- Seq(true, false); legacy <- Seq(true, false)) {
+ withSQLConf(otherConfs ++ Seq(
+ SQLConf.PARQUET_FILTER_PUSHDOWN_ENABLED.key -> "false",
+ SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> codegen.toString,
+ SQLConf.LEGACY_FILTER_SHORT_CIRCUIT_ON_NULL_CONJUNCT.key ->
legacy.toString): _*) {
+ withClue(s"wholeStage=$codegen, legacy=$legacy: ") {
+ f(codegen && legacy)
+ }
+ }
+ }
+ }
+
+ test("SPARK-60315: a filter evaluates the conjuncts after a NULL one") {
+ withNullConjunctTable(Int.MaxValue) {
+ // A common subexpression after the NULL conjunct takes the CSE path of
FilterExec.
+ val cseQuery = "SELECT a FROM t WHERE NOT (a > 0 AND b > 0) AND y + 1 >
0 AND y + 1 > -1"
+ // The CSE path evaluates the common `y + 1` once, the non-CSE path at
each use. Both
+ // generate the flag that keeps the row's evaluation going after a NULL
conjunct.
+ def cseQueryCode(filterExecCSE: Boolean): String = {
+ withSQLConf(
+ SQLConf.ANSI_ENABLED.key -> "true",
+ SQLConf.SUBEXPRESSION_ELIMINATION_FILTER_EXEC_ENABLED.key ->
filterExecCSE.toString) {
+ val code = codegenString(sql(cseQuery).queryExecution.executedPlan)
+ assert(code.contains("nullConjunct"))
+ code
+ }
+ }
+ assert("addExact".r.findAllIn(cseQueryCode(filterExecCSE = true)).length
<
+ "addExact".r.findAllIn(cseQueryCode(filterExecCSE = false)).length)
+
+ Seq("true", "false").foreach { filterExecCSE =>
+ Seq(
+ "SELECT a FROM t WHERE NOT (a > 0 AND b > 0) AND y + 1 > 0",
+ cseQuery
+ ).foreach { query =>
+ withNullConjunctConfs(
+ SQLConf.ANSI_ENABLED.key -> "true",
+ SQLConf.SUBEXPRESSION_ELIMINATION_FILTER_EXEC_ENABLED.key ->
filterExecCSE) {
+ skipsAtNull =>
+ if (skipsAtNull) {
+ checkAnswer(sql(query), Row(0))
+ } else {
+ checkError(
+ exception = intercept[SparkArithmeticException] {
+ sql(query).collect()
+ },
+ condition = "ARITHMETIC_OVERFLOW",
+ parameters = Map(
+ "message" -> "overflow",
+ "alternative" -> " Use 'try_add' to tolerate overflow and
return NULL instead.",
+ "config" -> toSQLConf(SQLConf.ANSI_ENABLED.key)),
+ context = ExpectedContext(
+ fragment = "y + 1",
+ start = query.indexOf("y + 1"),
+ stop = query.indexOf("y + 1") + 4))
+ }
+ }
+ }
+ }
+ }
+ }
+
+ test("SPARK-60315: a filter evaluates the conjuncts after a NULL one in
non-ANSI mode") {
+ // Year-month interval arithmetic raises on overflow in every mode.
+ withNullConjunctTable(178956970) {
+ val query = "SELECT a FROM t WHERE NOT (a > 0 AND b > 0) AND " +
+ "make_ym_interval(y, 0) + INTERVAL '1' YEAR > INTERVAL '0' YEAR"
+ withNullConjunctConfs(SQLConf.ANSI_ENABLED.key -> "false") { skipsAtNull
=>
+ if (skipsAtNull) {
+ checkAnswer(sql(query), Row(0))
+ } else {
+ checkError(
+ exception = intercept[SparkArithmeticException] {
+ sql(query).collect()
+ },
+ condition = "INTERVAL_ARITHMETIC_OVERFLOW.WITH_SUGGESTION",
+ parameters = Map("functionName" -> toSQLId("try_add")))
+ }
+ }
+ }
+ }
+
+ test("SPARK-60315: a filter drops a row with a NULL conjunct and stops at a
FALSE one") {
+ withNullConjunctTable(Int.MaxValue) {
+ withNullConjunctConfs(SQLConf.ANSI_ENABLED.key -> "true") { _ =>
+ // The first row is dropped after its last conjunct is evaluated.
+ checkAnswer(sql("SELECT a FROM t WHERE NOT (a > 0 AND b > 0) AND y >
-1"), Row(0))
Review Comment:
Right, `cseQuery` only ran on the `Int.MaxValue` table, so the CSE branch's
trailing `if (flag) continue;` was never checked at the result level. In
3264b00d897 the same query runs against `withNullConjunctTable(0)` with
`filterExec` CSE on and is checked to return `Row(0)` in every
`withNullConjunctConfs` combination. The data does not change the plan, so the
existing `addExact` assertion still shows that the CSE path is taken.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala:
##########
@@ -156,6 +202,10 @@ trait GeneratePredicateHelper extends PredicateHelper {
notNullPreds: Seq[Expression],
otherPreds: Seq[Expression],
nonNullAttrExprIds: Seq[ExprId]): String = {
+ // The IsNotNull checks are never NULL, so only `otherPreds` can need the
flag.
+ val nullFlag =
+ nullConjunctFlag(ctx, otherPreds.map(BindReferences.bindReference(_,
outputAttrs)))
Review Comment:
Agreed, the argument was evaluated before the legacy check in
`nullConjunctFlag`, so the bind ran in legacy mode as well. In 3264b00d897
`otherPreds` are bound once against `outputAttrs` and the bound expressions are
reused when generating the bodies, as the CSE branch already does with
`boundOtherPreds`. The `IsNotNull` checks bind against `inputAttrs`, so they
stay as they are.
--
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]
