cloud-fan commented on code in PR #59573:
URL: https://github.com/apache/spark/pull/59573#discussion_r4309696444
##########
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:
**Non-blocking (P2):** With the flag, a NULL conjunct no longer ends the
row's evaluation. The IsNotNull checks from `notNullPreds`, however, are still
emitted lazily, in both `generatePredicateCode` and the CSE branch of
`doConsume`: just before the first conjunct that references their column, or
after all conjuncts. With whole-stage codegen off, `And` evaluates the
condition in order, and the `isnotnull(...)` filters that
`InferFiltersFromConstraints` prepends come first.
Example, in ANSI mode, on a row `(a, b, y, x) = (NULL, 1, 2147483647, NULL)`
that reaches the filter:
```sql
SELECT a FROM t WHERE NOT (a > 0 AND b > 0) AND y + 1 > 0 AND x > 0
```
The optimizer prepends `isnotnull(y)` and `isnotnull(x)`. With codegen off,
evaluation stops at `isnotnull(x)` = FALSE and the row is dropped. With this
PR, codegen sets the flag on the NULL conjunct, passes the `isnotnull(y)` check
emitted before `y + 1 > 0`, evaluates `y + 1`, and fails with
`ARITHMETIC_OVERFLOW`, because `isnotnull(x)` is only checked before `x > 0`.
Before the PR, both modes dropped the row without an error. A user-written
guard placed first behaves the same way: in `WHERE x IS NOT NULL AND NOT (a > 0
AND b > 0) AND y + 1 > 0`, no other conjunct references `x`, so `isnotnull(x)`
is emitted after all of them. The aggregate `FILTER` clause goes through
`generatePredicateCode` and has the same problem.
So with the default config, some queries now fail only with whole-stage
codegen on. That is the inverse of SPARK-60315, and it contradicts the
migration guide's statement that this is what Spark "already did with
whole-stage codegen disabled". The new tests miss it because `NOT (a > 0 AND b
> 0)` infers no `IsNotNull`, and no later conjunct references a column that is
NULL on the overflowing row.
This extends an [existing
discussion](https://github.com/apache/spark/pull/59573#issuecomment-6233128726).
**Recommended change:** In both FilterExec predicate codegen paths,
generatePredicateCode (which the aggregate FILTER clause also uses) and the CSE
branch of FilterExec.doConsume, when a NULL-conjunct flag is generated, emit
before each conjunct that follows the first nullable conjunct, and before that
conjunct's column loads or CSE precomputes, every not-yet-emitted IsNotNull
check on a plain column that precedes it in the condition's conjunct order.
Keep each conjunct's position from splitConjunctivePredicates when the
condition is partitioned into IsNotNull checks and other conjuncts. Add
whole-stage codegen filter regression tests for an inferred IsNotNull, a
user-written IsNotNull written first, and a user-written IsNotNull written
after the overflowing conjunct, each on a column that is NULL on the row where
that conjunct overflows.
**Why this works:** With codegen off, the condition is evaluated in its
conjunct order, and the physical condition keeps that order:
InferFiltersFromConstraints puts the inferred isnotnull(...) filters first, and
a user-written IsNotNull stays where it was written. So a FALSE plain-column
isnotnull drops the row before any conjunct that follows it, but not before a
conjunct that comes earlier. Once a nullable conjunct can fall through,
whole-stage codegen must not reach a later conjunct before the plain-column
IsNotNull checks that precede it in that order, and must not run a check before
a conjunct that precedes the check. Emitting the pending preceding checks just
before each conjunct that follows the first nullable one meets both rules: a
FALSE check placed earlier in the condition drops the row first, and a check
placed after the conjunct still runs after it, so the conjunct's error is
raised in both modes. Plain-column checks are never NULL and cannot throw, so
running them e
arlier changes no result. Checks already emitted, complex IsNotNull(expr)
checks, and code with no flag keep their current placement.
**Scope:** When a NULL-conjunct flag is in play, emit each pending
plain-column IsNotNull check just before the first later conjunct that it
precedes in the condition, in both lockstep FilterExec predicate codegen paths,
and add regression tests for the codegen-only failure and for a check written
after the failing conjunct.
**Compatibility:** Outside the NULL-flag case, IsNotNull checks keep their
lazy placement just before the first conjunct that references them; codegen-off
evaluation is unchanged; a NULL-conjunct row with no FALSE IsNotNull ahead of a
raising later conjunct in the condition still raises that conjunct's error, as
the PR intends; and a plain-column IsNotNull written after that conjunct still
runs after it.
**Risks:** Moving the checks earlier loads their columns on rows that a
later FALSE conjunct would have rejected first. The cost is limited to
plain-column null checks and applies only when a flag is generated. The change
must not move any complex IsNotNull(expr), whose inner expression can throw.
The two lockstep paths both need each conjunct's condition position. If only
one path carries it, the IsNotNull checks are reordered in only one branch, and
the non-CSE and CSE paths disagree for the same filter.
**Constraints:** The non-CSE and CSE paths must make the same change,
because their comments require them to stay in lockstep. With
spark.sql.legacy.filterShortCircuitOnNullConjunct set to true, or when no
conjunct can be NULL after binding, the generated predicate code must stay
exactly as the PR emits it today. Every IsNotNull check must still run before
any CSE precompute or conjunct that references its column under tightened
nullability. The new placement must not run a plain-column IsNotNull check
before a conjunct that precedes it in the condition; the existing placement
just before the first conjunct that references its column is unchanged.
**Exclusions:**
- Complex IsNotNull(expr) checks stay where they are, so one that is FALSE
on a non-null input can still be reached after a raising conjunct; this
narrower residual is not repaired here.
- The older divergence where codegen checks a user-written IsNotNull earlier
than its position in the condition, and therefore skips a raising conjunct that
codegen off evaluates, is left unchanged.
**Success:** With whole-stage codegen on and the legacy flag off, `SELECT a
FROM t WHERE NOT (a > 0 AND b > 0) AND y + 1 > 0 AND x > 0` drops the row
(NULL, 1, 2147483647, NULL) without an error, as with codegen off. The same
holds for `WHERE x IS NOT NULL AND NOT (a > 0 AND b > 0) AND y + 1 > 0` and for
the same guard in an aggregate FILTER clause. `SELECT a FROM t WHERE NOT (a > 0
AND b > 0) AND y + 1 > 0 AND x IS NOT NULL` still fails with
ARITHMETIC_OVERFLOW on the row (NULL, 1, 2147483647, NULL) with whole-stage
codegen on and off, because the user-written check follows the overflowing
conjunct. The existing SPARK-60315 error-versus-skip matrix and generated-code
assertions still hold, and the generated code is unchanged when no conjunct can
be NULL or when the legacy flag is set.
**Verification:**
- **Regression:** With whole-stage codegen on and off and the legacy flag
off, a filter whose first conjunct is NULL on a row, whose later conjunct
overflows on that row, and whose plain-column IsNotNull (inferred from a later
comparison, or written first by the user) is FALSE on that row drops the row
without an error in both the non-CSE and the filterExec CSE paths.
- **Regression:** With whole-stage codegen on and off and the legacy flag
off, a filter whose first conjunct is NULL on a row, whose later conjunct
overflows on that row, and whose user-written plain-column IsNotNull follows
the overflowing conjunct and is FALSE on that row still fails with the overflow
error in both the non-CSE and the filterExec CSE paths.
- **Regression:** An aggregate FILTER clause with a user-written IsNotNull
guard first, a NULL second conjunct, and an overflowing third conjunct returns
the same count with whole-stage codegen on as with it off, without an error.
- **Compatibility:** With the legacy flag set, or when no conjunct can be
NULL after binding, the generated predicate code matches the current output,
and the existing error-versus-skip results across whole-stage codegen on/off
and legacy on/off are unchanged.
##########
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:
**Nit (P3):** This drop assertion, like the FALSE-stop one after it, has no
common subexpression, so it only exercises `generatePredicateCode`. The only
query that takes the FilterExec CSE branch is `cseQuery` in the first test, and
it runs only on the `Int.MaxValue` table. With the legacy config off, it raises
before the CSE branch's own trailing `if (flag) continue;`
(`withNullConjunctFlag(predCode + "\n" + leftoverNotNull, nullFlag)` in
`doConsume`). With it on, no flag is generated. The `nullConjunct` substring
check only shows that the flag exists. If that branch stopped dropping a row
whose condition is NULL, the suite would still pass, even though the comments
require the branch to be kept in lockstep with `generatePredicateCode` by hand.
To close the gap, run `cseQuery` against `withNullConjunctTable(0)` with
filterExec CSE on, and check that it returns `Row(0)` in every
`withNullConjunctConfs` combination.
**Verification:**
- **Behavior:** With filterExec subexpression elimination on, a filter with
a NULL first conjunct followed by conjuncts sharing a non-cheap common
subexpression returns, on non-overflowing data, only the rows whose condition
is TRUE in every combination of whole-stage codegen and the legacy flag, and
the default configuration's generated code takes the CSE path.
--
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]
