dongjoon-hyun commented on code in PR #59573:
URL: https://github.com/apache/spark/pull/59573#discussion_r4327369276
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala:
##########
@@ -186,13 +290,24 @@ trait GeneratePredicateHelper extends PredicateHelper {
// This has the property of not doing redundant IsNotNull checks and
taking better advantage of
// short-circuiting, not loading attributes until they are needed.
// This is very perf sensitive.
+ // With the NULL-conjunct flag, once a predicate that can be NULL is
generated, the IsNotNull
+ // checks that precede a later predicate in the condition are generated
before it, see
+ // `precedingNotNullChecks`, and a check on an expression that is written
after the predicate
+ // is not generated before it, see the nested checks below.
// NOTE: `FilterExec.doConsume`'s CSE branch inlines the same interleaving
so it can
// emit CSE state precomputes between the IsNotNull checks and each
otherPred body.
// Any change to this algorithm must be mirrored there (and vice versa).
// TODO: revisit this. We can consider reordering predicates as well.
val generatedIsNotNullChecks = new Array[Boolean](notNullPreds.length)
val extraIsNotNullAttrs = mutable.Set[Attribute]()
- val generated = otherPreds.map { c =>
+ var nullableSeen = false
+ val generated = otherPreds.zip(boundOtherPreds).zipWithIndex.map { case
((c, bound), k) =>
+ val afterNullable = nullFlag.isDefined && nullableSeen
Review Comment:
**`afterNullable` is decided at codegen time, so the moved checks and the
skipped nested checks also apply to rows where no conjunct is NULL.** Once some
conjunct can be NULL after binding, `precedingNotNullChecks` and the
nested-branch skip (L334; L530 and L560 in the CSE branch) are in the generated
code for every row. So with the default config, rows without any NULL conjunct
also behave differently from 4.3, in both directions, always toward the
codegen-off result. With ANSI on and Parquet filter pushdown off:
- On `withNullConjunctTable(Int.MaxValue)`, `SELECT a FROM t WHERE cast(s AS
INT) IS NOT NULL AND (a > 0 OR b > 0) AND y < 0` returns no rows on master and
with the legacy config, because `y < 0` is FALSE before the leftover cast check
runs. Now it raises `CAST_INVALID_INPUT`: on the first row the `OR` is TRUE
(`NULL OR TRUE`), and `isnotnull(cast(s AS INT))` is generated before `y < 0`.
Codegen off raises too.
- On a row `(a, b, s) = (0, 1, 'x')`, `WHERE NOT (a > 0 AND b > 0) AND
coalesce(b, cast(s AS INT)) < 0 AND cast(s AS INT) IS NOT NULL` raised
`CAST_INVALID_INPUT` on master and still does with the legacy config, but now
returns no rows, as codegen off does.
The migration guide ("on a row after one of them evaluates to `NULL`" ...
"on that row"), the legacy config doc (`SQLConf.scala:1751`) and the
user-facing section of the PR description ("Rows whose earlier conjunct is
FALSE are not affected") describe only rows where a conjunct is NULL. Also, the
SPARK-60377 example under "Not fixed here" in the PR description, which becomes
the commit message, now raises only with the legacy config; `WHERE coalesce(b,
cast(s AS INT)) < 0 AND cast(s AS INT) IS NOT NULL` on `(b, s) = (1, 'x')`
still shows it with the default config.
Could you add a sentence to the migration guide and the legacy config doc
saying that, once a condition has a conjunct that can be NULL, the `IS NOT
NULL` checks after that conjunct are evaluated in the order of the condition on
every row, not only on rows where a conjunct is NULL, update that example, and
pin the first query in a test?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala:
##########
@@ -137,47 +137,151 @@ trait GeneratePredicateHelper extends PredicateHelper {
condition: Expression,
inputAttrs: Seq[Attribute],
inputExprCode: Seq[ExprCode]): String = {
- val (notNullPreds, otherPreds) =
splitConjunctivePredicates(condition).partition {
- case IsNotNull(a) => isNullIntolerant(a) &&
a.references.subsetOf(AttributeSet(inputAttrs))
- case _ => false
- }
+ val (notNullPreds, otherPreds, precedingNotNulls) =
+ splitNotNullPreds(condition, AttributeSet(inputAttrs))
val nonNullAttrExprIds =
notNullPreds.flatMap(_.references).distinct.map(_.exprId)
val outputAttrs = outputWithNullability(inputAttrs, nonNullAttrExprIds)
generatePredicateCode(
- ctx, inputAttrs, inputExprCode, outputAttrs, notNullPreds, otherPreds,
+ ctx, inputAttrs, inputExprCode, notNullPreds, otherPreds,
+ otherPreds.map(BindReferences.bindReference(_, outputAttrs)),
precedingNotNulls,
nonNullAttrExprIds)
}
+ /**
+ * Splits `condition` into the `IsNotNull` checks on null-intolerant
expressions over
+ * `inputAttrs`, the other conjuncts, and for each of the other conjuncts
the number of the
+ * checks that precede it in `condition`. The conjuncts keep their order
within each group, so
+ * the checks before the k-th other conjunct are the first
`precedingNotNulls(k)` checks.
+ */
+ protected def splitNotNullPreds(
+ condition: Expression,
+ inputAttrs: AttributeSet): (Seq[Expression], Seq[Expression], Seq[Int])
= {
+ val notNullPreds = mutable.ArrayBuffer.empty[Expression]
+ val otherPreds = mutable.ArrayBuffer.empty[Expression]
+ val precedingNotNulls = mutable.ArrayBuffer.empty[Int]
+ splitConjunctivePredicates(condition).foreach {
+ case p @ IsNotNull(a) if isNullIntolerant(a) &&
a.references.subsetOf(inputAttrs) =>
+ notNullPreds += p
+ case p =>
+ otherPreds += p
+ precedingNotNulls += notNullPreds.length
+ }
+ (notNullPreds.toSeq, otherPreds.toSeq, precedingNotNulls.toSeq)
+ }
+
+ /**
+ * 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)) {
Review Comment:
**The flag, and with it the hoisted checks, is generated whenever a conjunct
can be NULL, even when nothing after it can raise an error.** This condition
looks only at nullability, so any `OR`, `IN`, `CASE` or `coalesce` over
nullable columns turns it on. After that conjunct, `precedingNotNullChecks`
generates all pending checks before the next conjunct, including every
`isnotnull(...)` that `InferFiltersFromConstraints` prepends, and on the
non-CSE path each check loads its column value through
`evaluateRequiredVariables` (L269). That code runs on every row that reaches
the next conjunct, not only on rows where a conjunct is NULL.
For example, on a Parquet table `t(a INT, b INT, c INT, d DECIMAL(38,10))`,
`WHERE (a > 5 OR b > 5) AND c = 1 AND d = 2` becomes `isnotnull(c) AND
isnotnull(d) AND (a > 5 OR b > 5) AND c = 1 AND d = 2`. Master decodes `d` only
for rows with `c = 1`, while this PR decodes it, allocating a `BigInteger` and
a `BigDecimal`, on every row where the `OR` is TRUE or NULL. No conjunct there
can raise, so neither the flag nor the hoisting changes any result. This is the
cost that d6036f470d7 (the SPARK-56032 follow-up for TPC-DS q28) removed from
the CSE prologue, and it goes beyond the risk noted in the first review
("limited to plain-column null checks"), because the values are decoded too and
common `OR`/`IN` filters trigger it.
Could you gate both with `ExprUtils.canEvaluateUnconditionally`, applied to
the unbound expressions? Return `None` here unless a conjunct after the first
nullable one, or one of the `IsNotNull` checks, cannot be evaluated
unconditionally, and in `precedingNotNullChecks` hoist before the k-th conjunct
only if it or one of the pending checks cannot. `precedingNotNulls` never
decreases, so the next conjunct that can raise still gets every check that
precedes it, and the legacy path stays as it is.
##########
docs/sql-migration-guide.md:
##########
@@ -51,6 +51,7 @@ license: |
- Since Spark 4.4, the value columns of `UNPIVOT` and `Dataset.unpivot` merge
the nested nullability (array element, map value, and struct field nullability)
of all unpivoted values. Previously, a value column took the nested nullability
of the first listed value, so a `NULL` array element, map value, or struct
field of a later value could be returned as a non-`NULL` value such as `0`, and
`IS NULL` on it returned `false`. A view created by an earlier Spark version
over such an `UNPIVOT` may have recorded the non-nullable nested type in its
schema and can fail to read; recreate it with `CREATE OR REPLACE VIEW`. To
restore the previous behavior, set
`spark.sql.legacy.unpivot.useFirstValueNestedNullability` to `true`.
- Since Spark 4.4, when `spark.sql.parquet.fieldId.read.enabled` is true, a
Parquet read fails with `PARQUET_FIELD_REQUESTED_MORE_THAN_ONCE` if two or more
requested fields of a struct (or of the top-level schema) resolve to the same
Parquet field. This happens when one requested field is matched by field ID and
another by name, or when two requested fields have the same field ID. In Spark
4.3 and earlier, such a read returned wrong values without an error: with the
vectorized reader one of the fields read as non-null default values (for
example `0`) or uninitialized data, and with the parquet-mr reader
(`spark.sql.parquet.enableVectorizedReader` set to `false`) one of the fields
read as `null`. To read the Parquet field more than once, request it once and
refer to the resulting column more than once in the query.
- Since Spark 4.4, when `spark.sql.groupByOrdinal` is true (the default), the
pipe `AGGREGATE` operator rejects an aliased integer literal in its `GROUP BY`
clause, such as `TABLE t |> AGGREGATE sum(x) AS s GROUP BY 2 AS grp`, with
`UNSUPPORTED_FEATURE.PIPE_OPERATOR_AGGREGATE_UNSUPPORTED_CASE`, because a bare
integer literal in the same position is an ordinal. Previously the aliased
literal was grouped as a constant. Bare ordinals and aliases on other grouping
expressions are not affected. To restore the previous behavior, set
`spark.sql.legacy.allowAliasedPipeAggregateOrdinals` to `true`.
+- Since Spark 4.4, with whole-stage codegen enabled, a filter condition,
including the `FILTER` clause of an aggregate function, evaluates the remaining
top-level conjuncts of the condition on a row after one of them evaluates to
`NULL`, as the SQL standard requires, and does so in the order they appear in
the condition, as Spark already did with whole-stage codegen disabled. Only a
`FALSE` conjunct stops the evaluation. The one exception is an `IS NOT NULL`
check on a column that a later conjunct references: it is still evaluated
before that conjunct, even when it is written after it, and can drop the row
before an error the conjunct raises. A row whose condition is `NULL` is still
dropped, but an error raised by a later conjunct on that row, such as
`ARITHMETIC_OVERFLOW` in ANSI mode, now fails the query. For example, `SELECT a
FROM t WHERE NOT (a > 0 AND b > 0) AND y + 1 > 0` can fail with
`ARITHMETIC_OVERFLOW` on a row where `a` is `NULL`, `b` is `1`, and `y` is
`2147483647`, un
less the row is dropped earlier by a `FALSE` conjunct or `IS NOT NULL` check
that precedes `y + 1 > 0` in the condition, or skipped before the filter, for
example by data source filter pushdown such as Parquet row group statistics.
Previously the query skipped such a row with whole-stage codegen enabled. The
later conjuncts also run on more rows than before, so a nondeterministic
conjunct such as `rand(seed)` can return different values on the remaining
rows, and a side-effecting user-defined function is called on rows that are
dropped. To restore the previous behavior, set
`spark.sql.legacy.filterShortCircuitOnNullConjunct` to `true`.
Review Comment:
Minor: "as Spark already did with whole-stage codegen disabled. Only a
`FALSE` conjunct stops the evaluation." does not hold when the conjuncts after
a FALSE one share a subexpression. With whole-stage codegen off, `FilterExec`
uses `Predicate.create`, and `GeneratePredicate` evaluates the common
subexpressions before the condition (`GeneratePredicate.scala:63-64`); the
generated projection for an aggregate `FILTER` does the same. So on
`withNullConjunctTable(Int.MaxValue)` with ANSI on and Parquet filter pushdown
off, `SELECT a FROM t WHERE NOT (a > 0 AND b > 0) AND b < 0 AND y + 1 > 0 AND y
+ 1 > -1` returns no rows with codegen on, whatever the legacy and filterExec
CSE configs, and raises `ARITHMETIC_OVERFLOW` with codegen off. This is the
pre-existing difference you mentioned in the review thread, and it also happens
without the NULL conjunct, so it is not a regression. Could you list it under
"Not fixed here" in the PR description, or file a JIRA, and qualify this
sentence?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala:
##########
@@ -451,10 +597,11 @@ case class FilterExec(condition: Expression, child:
SparkPlan)
case (p, i) if !generatedIsNotNullChecks(i) => genNotNull(p)
}.mkString("\n")
- (inputVarsEvalCode, predCode + "\n" + leftoverNotNull)
+ (inputVarsEvalCode, withNullConjunctFlag(predCode + "\n" +
leftoverNotNull, nullFlag))
} else {
("", generatePredicateCode(
- ctx, child.output, input, output, notNullPreds, otherPreds,
notNullAttributes))
+ ctx, child.output, input, notNullPreds, otherPreds, boundOtherPreds,
Review Comment:
Nit: this call now passes `boundOtherPreds` to the non-CSE path, where it
decides the NULL-conjunct flag and supplies the bodies of the conjuncts, but
the comment on `boundOtherPreds` (L381-383) still says it is "shared between
the CSE gate in `doConsume` and the CSE codegen itself". Could you update it,
e.g. "shared by the CSE gate in `doConsume` and both predicate code paths (the
CSE branch and `generatePredicateCode`)"?
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/WholeStageCodegenSuite.scala:
##########
@@ -2814,4 +2815,274 @@ 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, NULL, 'x'), (0, 1, 0, 1, '1') AS
v(a, b, y, x, s)")
+ f
+ }
+ }
+
+ // A common subexpression after the NULL conjunct takes the CSE path of
FilterExec.
+ private val nullConjunctCSEQuery =
+ "SELECT a FROM t WHERE NOT (a > 0 AND b > 0) AND y + 1 > 0 AND y + 1 > -1"
+
+ private def checkArithmeticOverflow(query: String): Unit = {
+ 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))
+ }
+
+ // 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) {
+ // 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(nullConjunctCSEQuery).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",
+ nullConjunctCSEQuery
+ ).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 {
+ checkArithmeticOverflow(query)
+ }
+ }
+ }
+ }
+ }
+ // A NULL conjunct from OR. The second row does not pass it.
+ withNullConjunctTable(Int.MaxValue) {
+ val query = "SELECT a FROM t WHERE (a > 0 OR b < 0) AND y + 1 > 0"
+ withNullConjunctConfs(SQLConf.ANSI_ENABLED.key -> "true") { skipsAtNull
=>
+ if (skipsAtNull) {
+ checkAnswer(sql(query), Seq.empty)
+ } else {
+ checkArithmeticOverflow(query)
+ }
+ }
+ }
+ // Without an error, the CSE path drops the first row after its last
conjunct is evaluated.
+ withNullConjunctTable(0) {
+ withNullConjunctConfs(
+ SQLConf.ANSI_ENABLED.key -> "true",
+ SQLConf.SUBEXPRESSION_ELIMINATION_FILTER_EXEC_ENABLED.key -> "true")
{ _ =>
+ checkAnswer(sql(nullConjunctCSEQuery), Row(0))
+ }
+ }
+ }
+
+ 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 condition = "NOT (a > 0 AND b > 0) AND " +
+ "make_ym_interval(y, 0) + INTERVAL '1' YEAR > INTERVAL '0' YEAR"
+ val query = s"SELECT a FROM t WHERE $condition"
+ 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")))
+ }
+ // A check on an expression, not on a column, that precedes the NULL
conjunct:
+ // `cast(s AS INT)` is NULL on the first row, so the check is FALSE
and drops the row
+ // before the overflowing conjunct, as with codegen off.
+ checkAnswer(
+ sql(s"SELECT a FROM t WHERE cast(s AS INT) IS NOT NULL AND
$condition"), Row(0))
+ }
+ }
+ }
+
+ 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))
+ // FALSE decides the AND, so the overflowing conjunct after it is not
evaluated.
+ checkAnswer(
+ sql("SELECT a FROM t WHERE NOT (a > 0 AND b > 0) AND b < 0 AND y + 1
> 0"),
+ Seq.empty)
+ }
+ // `b > 0` cannot be NULL after the inferred `isnotnull(b)`, so no flag
is generated.
+ val df = sql("SELECT a FROM t WHERE b > 0 AND y > -1")
+ checkAnswer(df, Seq(Row(null), Row(0)))
+
assert(!codegenString(df.queryExecution.executedPlan).contains("nullConjunct"))
+ }
+ }
+
+ test("SPARK-60315: a filter checks IS NOT NULL before the conjuncts it
precedes") {
+ withNullConjunctTable(Int.MaxValue) {
+ // `x` is NULL on the first row. With codegen off, a FALSE
`isnotnull(x)` drops the row
+ // before a conjunct after it in the condition is evaluated. With
codegen on, the check is
+ // generated just before the first conjunct that references `x`, or
after all of them, so
+ // after a NULL conjunct it has to be generated before the conjuncts it
precedes.
+ //
+ // The common `b * 2` takes the CSE path of FilterExec. The overflowing
`y + 1` is not
+ // common because `GeneratePredicate` (codegen off) evaluates common
subexpressions first.
+ val cseQuery = "SELECT a FROM t WHERE x IS NOT NULL AND NOT (a > 0 AND b
> 0) 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) {
+ skipsAtNull =>
+ // `isnotnull(x)` is inferred from `x > 0` and precedes the
overflowing conjunct.
+ checkAnswer(
+ sql("SELECT a FROM t WHERE NOT (a > 0 AND b > 0) AND y + 1 > 0
AND x > 0"),
+ Row(0))
+ // A user-written check that no other conjunct references.
+ checkAnswer(
+ sql("SELECT a FROM t WHERE x IS NOT NULL AND NOT (a > 0 AND b >
0) AND y + 1 > 0"),
+ Row(0))
+ // The same with a common subexpression.
+ checkAnswer(sql(cseQuery), Row(0))
+ // A check written after the overflowing conjunct stays after it.
+ val checkAfter =
+ "SELECT a FROM t WHERE NOT (a > 0 AND b > 0) AND y + 1 > 0 AND x
IS NOT NULL"
Review Comment:
Minor: this query, like the inferred-check one at L2993, has no common
subexpression, so it takes `generatePredicateCode` even when `filterExecCSE` is
true. Only `cseQuery`, whose check is written first, runs the preceding checks
through the CSE branch of `FilterExec.doConsume`. If a later change to that
branch emitted the pending column checks right after the first nullable
conjunct (the shape that cloud-fan's first review warned would bring back the
original inconsistency), `... AND y + 1 > 0 AND x IS NOT NULL AND b * 2 > 0 AND
b * 2 > -1` would return `[0]` with codegen on and raise with it off, and the
suite would still pass; the new expression-check test only catches hoisting
that includes the expression check. Could you also check `checkAfter + " AND b
* 2 > 0 AND b * 2 > -1"` with the same `skipsAtNull` expectation? With
`precedingNotNulls = [2, 2, 3, 3]`, it would pin the per-conjunct placement as
well.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala:
##########
@@ -373,29 +503,43 @@ case class FilterExec(condition: Expression, child:
SparkPlan)
// Emit an IsNotNull check, binding against child.output so the inner
expression
// is evaluated with its original nullability (the
tightened-nullability `output`
// is only appropriate inside otherPreds, where the required checks
have fired).
- // The `bound.nullable` gate below is defensive -- `IsNotNull` always
produces a
- // non-nullable Boolean, so this branch is structurally unreachable
today; it
- // mirrors the non-CSE helper's shape to keep future refactors safe.
+ // Passing `bound.nullable` to `conjunctCheck` is defensive --
`IsNotNull` always
+ // produces a non-nullable Boolean, so its NULL handling is
unreachable here; the
+ // non-CSE helper generates its IsNotNull checks the same way.
def genNotNull(pred: Expression): String = {
val bound = BindReferences.bindReference(pred, child.output)
val evaluated = evaluateRequiredVariables(child.output, input,
pred.references)
val ev = ExpressionCanonicalizer.execute(bound).genCode(ctx)
- val nullCheck = if (bound.nullable) s"${ev.isNull} || " else ""
s"""
|$evaluated
|${ev.code}
- |if (${nullCheck}!${ev.value}) continue;
+ |${conjunctCheck(ev, bound.nullable, nullFlag)}
Review Comment:
Nit: with a flag, passing `nullFlag` here turns the "defensive" NULL
handling of a check into `if (isNull) { flag = true; } else if (!value)
continue;`. So if a check could ever be NULL, it would no longer drop the row
before the later conjuncts and CSE precomputes that are bound with its column
non-nullable, which is what invariant (c) above is about. Master's `if (isNull
|| !value) continue;` was really defensive, and `genNotNull` in
`generatePredicateCode` (L277, L283) has the same issue. Could you pass `None`
for the `IsNotNull` checks in both helpers? The generated code does not change,
since `IsNotNull` is never nullable.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/WholeStageCodegenSuite.scala:
##########
@@ -2814,4 +2815,274 @@ 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, NULL, 'x'), (0, 1, 0, 1, '1') AS
v(a, b, y, x, s)")
+ f
+ }
+ }
+
+ // A common subexpression after the NULL conjunct takes the CSE path of
FilterExec.
+ private val nullConjunctCSEQuery =
+ "SELECT a FROM t WHERE NOT (a > 0 AND b > 0) AND y + 1 > 0 AND y + 1 > -1"
+
+ private def checkArithmeticOverflow(query: String): Unit = {
+ 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))
+ }
+
+ // 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) {
+ // 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(nullConjunctCSEQuery).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",
+ nullConjunctCSEQuery
+ ).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 {
+ checkArithmeticOverflow(query)
+ }
+ }
+ }
+ }
+ }
+ // A NULL conjunct from OR. The second row does not pass it.
+ withNullConjunctTable(Int.MaxValue) {
+ val query = "SELECT a FROM t WHERE (a > 0 OR b < 0) AND y + 1 > 0"
+ withNullConjunctConfs(SQLConf.ANSI_ENABLED.key -> "true") { skipsAtNull
=>
+ if (skipsAtNull) {
+ checkAnswer(sql(query), Seq.empty)
+ } else {
+ checkArithmeticOverflow(query)
+ }
+ }
+ }
+ // Without an error, the CSE path drops the first row after its last
conjunct is evaluated.
+ withNullConjunctTable(0) {
+ withNullConjunctConfs(
+ SQLConf.ANSI_ENABLED.key -> "true",
+ SQLConf.SUBEXPRESSION_ELIMINATION_FILTER_EXEC_ENABLED.key -> "true")
{ _ =>
+ checkAnswer(sql(nullConjunctCSEQuery), Row(0))
+ }
+ }
+ }
+
+ 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 condition = "NOT (a > 0 AND b > 0) AND " +
+ "make_ym_interval(y, 0) + INTERVAL '1' YEAR > INTERVAL '0' YEAR"
+ val query = s"SELECT a FROM t WHERE $condition"
+ 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")))
+ }
+ // A check on an expression, not on a column, that precedes the NULL
conjunct:
+ // `cast(s AS INT)` is NULL on the first row, so the check is FALSE
and drops the row
+ // before the overflowing conjunct, as with codegen off.
+ checkAnswer(
+ sql(s"SELECT a FROM t WHERE cast(s AS INT) IS NOT NULL AND
$condition"), Row(0))
+ }
+ }
+ }
+
+ 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))
+ // FALSE decides the AND, so the overflowing conjunct after it is not
evaluated.
+ checkAnswer(
+ sql("SELECT a FROM t WHERE NOT (a > 0 AND b > 0) AND b < 0 AND y + 1
> 0"),
+ Seq.empty)
+ }
+ // `b > 0` cannot be NULL after the inferred `isnotnull(b)`, so no flag
is generated.
+ val df = sql("SELECT a FROM t WHERE b > 0 AND y > -1")
+ checkAnswer(df, Seq(Row(null), Row(0)))
+
assert(!codegenString(df.queryExecution.executedPlan).contains("nullConjunct"))
Review Comment:
Nit: when the plan has no whole-stage codegen subtree, `codegenString`
returns only "Found 0 WholeStageCodegen subtrees.", so this assertion would
also pass if the filter were not in whole-stage codegen. The other `FilterExec`
codegen tests here guard with
`plan.exists(_.isInstanceOf[WholeStageCodegenExec])`, and `genCode` (L889)
asserts it. Could you use
`assert(genCode(df).forall(!_.body.contains("nullConjunct")))` instead?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala:
##########
@@ -201,56 +316,64 @@ trait GeneratePredicateHelper extends PredicateHelper {
if (idx != -1 && !generatedIsNotNullChecks(idx)) {
generatedIsNotNullChecks(idx) = true
// Use the child's output. The nullability is what the child
produced.
- genPredicate(notNullPreds(idx), inputExprCode, inputAttrs)
+ genNotNull(notNullPreds(idx))
} else if (nonNullAttrExprIds.contains(r.exprId) &&
!extraIsNotNullAttrs.contains(r)) {
Review Comment:
Nit: when `precedingNotNullChecks` has already generated the direct
`IsNotNull(c)` for a column, this branch adds a synthetic `IsNotNull(c)` for
the same column right before the conjunct that references it, e.g. for `c` in
`(a = 1 OR b = 2) AND c = 5`. It is harmless at runtime, but it adds code to
the per-row loop and goes against "not doing redundant IsNotNull checks"
(L290). Master already does this for the second conjunct that references a
column, e.g. `c <= hi` of a `BETWEEN`. Could you guard this branch with `idx ==
-1` (and `ni == -1` at L552 in the CSE branch)? A synthetic check is only
needed for a column that only a check on an expression covers. If the generated
code without a flag has to stay identical to master, `idx == -1 ||
!afterNullable` works too.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala:
##########
@@ -373,29 +503,43 @@ case class FilterExec(condition: Expression, child:
SparkPlan)
// Emit an IsNotNull check, binding against child.output so the inner
expression
// is evaluated with its original nullability (the
tightened-nullability `output`
// is only appropriate inside otherPreds, where the required checks
have fired).
- // The `bound.nullable` gate below is defensive -- `IsNotNull` always
produces a
- // non-nullable Boolean, so this branch is structurally unreachable
today; it
- // mirrors the non-CSE helper's shape to keep future refactors safe.
+ // Passing `bound.nullable` to `conjunctCheck` is defensive --
`IsNotNull` always
+ // produces a non-nullable Boolean, so its NULL handling is
unreachable here; the
+ // non-CSE helper generates its IsNotNull checks the same way.
def genNotNull(pred: Expression): String = {
val bound = BindReferences.bindReference(pred, child.output)
val evaluated = evaluateRequiredVariables(child.output, input,
pred.references)
val ev = ExpressionCanonicalizer.execute(bound).genCode(ctx)
- val nullCheck = if (bound.nullable) s"${ev.isNull} || " else ""
s"""
|$evaluated
|${ev.code}
- |if (${nullCheck}!${ev.value}) continue;
+ |${conjunctCheck(ev, bound.nullable, nullFlag)}
""".stripMargin
}
val generatedIsNotNullChecks = new Array[Boolean](notNullPreds.length)
val extraIsNotNullAttrs = mutable.Set[Attribute]()
+ var nullableSeen = false
val predCode: String = {
val parts = new StringBuilder
ctx.withSubExprEliminationExprs(subExprs.states) {
otherPreds.zip(boundOtherPreds).zipWithIndex.foreach {
case ((orig, bound), idx) =>
+ val afterNullable = nullFlag.isDefined && nullableSeen
+ precedingNotNullChecks(
+ notNullPreds, precedingNotNulls, idx, afterNullable,
+ generatedIsNotNullChecks).foreach { p =>
+ // The inner expression of a check is evaluated with no CSE
state, as below.
Review Comment:
Nit: this block (L534-540) is a copy of the one below (L566-572), and its
only explanation is "as below"; why a check is generated without the CSE state
is not written down in the main code. Could you move it into a local helper
next to `genNotNull`, e.g. `genNotNullWithoutCSE`, with a short comment on the
reason? The calls stay in the same order, so the generated code does not change.
##########
docs/sql-migration-guide.md:
##########
@@ -51,6 +51,7 @@ license: |
- Since Spark 4.4, the value columns of `UNPIVOT` and `Dataset.unpivot` merge
the nested nullability (array element, map value, and struct field nullability)
of all unpivoted values. Previously, a value column took the nested nullability
of the first listed value, so a `NULL` array element, map value, or struct
field of a later value could be returned as a non-`NULL` value such as `0`, and
`IS NULL` on it returned `false`. A view created by an earlier Spark version
over such an `UNPIVOT` may have recorded the non-nullable nested type in its
schema and can fail to read; recreate it with `CREATE OR REPLACE VIEW`. To
restore the previous behavior, set
`spark.sql.legacy.unpivot.useFirstValueNestedNullability` to `true`.
- Since Spark 4.4, when `spark.sql.parquet.fieldId.read.enabled` is true, a
Parquet read fails with `PARQUET_FIELD_REQUESTED_MORE_THAN_ONCE` if two or more
requested fields of a struct (or of the top-level schema) resolve to the same
Parquet field. This happens when one requested field is matched by field ID and
another by name, or when two requested fields have the same field ID. In Spark
4.3 and earlier, such a read returned wrong values without an error: with the
vectorized reader one of the fields read as non-null default values (for
example `0`) or uninitialized data, and with the parquet-mr reader
(`spark.sql.parquet.enableVectorizedReader` set to `false`) one of the fields
read as `null`. To read the Parquet field more than once, request it once and
refer to the resulting column more than once in the query.
- Since Spark 4.4, when `spark.sql.groupByOrdinal` is true (the default), the
pipe `AGGREGATE` operator rejects an aliased integer literal in its `GROUP BY`
clause, such as `TABLE t |> AGGREGATE sum(x) AS s GROUP BY 2 AS grp`, with
`UNSUPPORTED_FEATURE.PIPE_OPERATOR_AGGREGATE_UNSUPPORTED_CASE`, because a bare
integer literal in the same position is an ordinal. Previously the aliased
literal was grouped as a constant. Bare ordinals and aliases on other grouping
expressions are not affected. To restore the previous behavior, set
`spark.sql.legacy.allowAliasedPipeAggregateOrdinals` to `true`.
+- Since Spark 4.4, with whole-stage codegen enabled, a filter condition,
including the `FILTER` clause of an aggregate function, evaluates the remaining
top-level conjuncts of the condition on a row after one of them evaluates to
`NULL`, as the SQL standard requires, and does so in the order they appear in
the condition, as Spark already did with whole-stage codegen disabled. Only a
`FALSE` conjunct stops the evaluation. The one exception is an `IS NOT NULL`
check on a column that a later conjunct references: it is still evaluated
before that conjunct, even when it is written after it, and can drop the row
before an error the conjunct raises. A row whose condition is `NULL` is still
dropped, but an error raised by a later conjunct on that row, such as
`ARITHMETIC_OVERFLOW` in ANSI mode, now fails the query. For example, `SELECT a
FROM t WHERE NOT (a > 0 AND b > 0) AND y + 1 > 0` can fail with
`ARITHMETIC_OVERFLOW` on a row where `a` is `NULL`, `b` is `1`, and `y` is
`2147483647`, un
less the row is dropped earlier by a `FALSE` conjunct or `IS NOT NULL` check
that precedes `y + 1 > 0` in the condition, or skipped before the filter, for
example by data source filter pushdown such as Parquet row group statistics.
Previously the query skipped such a row with whole-stage codegen enabled. The
later conjuncts also run on more rows than before, so a nondeterministic
conjunct such as `rand(seed)` can return different values on the remaining
rows, and a side-effecting user-defined function is called on rows that are
dropped. To restore the previous behavior, set
`spark.sql.legacy.filterShortCircuitOnNullConjunct` to `true`.
Review Comment:
Minor: "The one exception" is narrower than what the code does. On
`withNullConjunctTable(Int.MaxValue)` with ANSI on and Parquet filter pushdown
off:
1. A check on an expression also makes its columns non-nullable, so a
synthetic `isnotnull(x)` is generated before the first later conjunct that
references `x` (`basicPhysicalOperators.scala:320-322`). `SELECT count(*)
FILTER (WHERE NOT (a > 0 AND b > 0) AND y + coalesce(x, 1) > 0 AND -x IS NOT
NULL) FROM t` returns 1 with whole-stage codegen on and raises
`ARITHMETIC_OVERFLOW` with it off; an aggregate `FILTER` gets no inferred
checks.
2. The moved check also runs before the conjuncts between that conjunct and
its own position, so it can drop the row before an error raised by a conjunct
that does not reference the column. `WHERE NOT (a > 0 AND b > 0) AND (x > 0 OR
b > 0) AND y + 1 > 0 AND x IS NOT NULL` drops the first row with codegen on,
while codegen off evaluates the `OR` (TRUE) and fails at `y + 1 > 0`. The
`precedingNotNullChecks` scaladoc ("unless the conjunct raises an error on the
row first") has the same gap.
3. The first conjunct that can be NULL keeps the nested hoisting, since
`afterNullable` is false there. `WHERE NOT (a > 0 AND coalesce(b, cast(s AS
INT)) > 0) AND b < 0 AND cast(s AS INT) IS NOT NULL` raises
`CAST_INVALID_INPUT` on the first row, whose first conjunct is NULL, with
codegen on, and returns no rows with codegen off.
All three behave as on master, so this is only about the text. Could you
describe the exception more broadly (a check on a column, or on an expression
over it, runs before the first later conjunct that references the column and
every conjunct up to its own position) and mention the first nullable conjunct,
or say "One exception"? The "Without a NULL conjunct" of the SPARK-60377 bullet
in the PR description could cover case 3 too.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/WholeStageCodegenSuite.scala:
##########
@@ -2814,4 +2815,274 @@ 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, NULL, 'x'), (0, 1, 0, 1, '1') AS
v(a, b, y, x, s)")
+ f
+ }
+ }
+
+ // A common subexpression after the NULL conjunct takes the CSE path of
FilterExec.
+ private val nullConjunctCSEQuery =
+ "SELECT a FROM t WHERE NOT (a > 0 AND b > 0) AND y + 1 > 0 AND y + 1 > -1"
+
+ private def checkArithmeticOverflow(query: String): Unit = {
+ 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))
+ }
+
+ // 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) {
+ // 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(nullConjunctCSEQuery).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",
+ nullConjunctCSEQuery
+ ).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 {
+ checkArithmeticOverflow(query)
+ }
+ }
+ }
+ }
+ }
+ // A NULL conjunct from OR. The second row does not pass it.
+ withNullConjunctTable(Int.MaxValue) {
+ val query = "SELECT a FROM t WHERE (a > 0 OR b < 0) AND y + 1 > 0"
+ withNullConjunctConfs(SQLConf.ANSI_ENABLED.key -> "true") { skipsAtNull
=>
+ if (skipsAtNull) {
+ checkAnswer(sql(query), Seq.empty)
+ } else {
+ checkArithmeticOverflow(query)
+ }
+ }
+ }
+ // Without an error, the CSE path drops the first row after its last
conjunct is evaluated.
+ withNullConjunctTable(0) {
+ withNullConjunctConfs(
+ SQLConf.ANSI_ENABLED.key -> "true",
+ SQLConf.SUBEXPRESSION_ELIMINATION_FILTER_EXEC_ENABLED.key -> "true")
{ _ =>
+ checkAnswer(sql(nullConjunctCSEQuery), Row(0))
+ }
+ }
+ }
+
+ 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 condition = "NOT (a > 0 AND b > 0) AND " +
+ "make_ym_interval(y, 0) + INTERVAL '1' YEAR > INTERVAL '0' YEAR"
+ val query = s"SELECT a FROM t WHERE $condition"
+ 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")))
+ }
+ // A check on an expression, not on a column, that precedes the NULL
conjunct:
+ // `cast(s AS INT)` is NULL on the first row, so the check is FALSE
and drops the row
+ // before the overflowing conjunct, as with codegen off.
+ checkAnswer(
+ sql(s"SELECT a FROM t WHERE cast(s AS INT) IS NOT NULL AND
$condition"), Row(0))
+ }
+ }
+ }
+
+ 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))
+ // FALSE decides the AND, so the overflowing conjunct after it is not
evaluated.
+ checkAnswer(
+ sql("SELECT a FROM t WHERE NOT (a > 0 AND b > 0) AND b < 0 AND y + 1
> 0"),
+ Seq.empty)
+ }
+ // `b > 0` cannot be NULL after the inferred `isnotnull(b)`, so no flag
is generated.
+ val df = sql("SELECT a FROM t WHERE b > 0 AND y > -1")
+ checkAnswer(df, Seq(Row(null), Row(0)))
+
assert(!codegenString(df.queryExecution.executedPlan).contains("nullConjunct"))
+ }
+ }
+
+ test("SPARK-60315: a filter checks IS NOT NULL before the conjuncts it
precedes") {
+ withNullConjunctTable(Int.MaxValue) {
+ // `x` is NULL on the first row. With codegen off, a FALSE
`isnotnull(x)` drops the row
+ // before a conjunct after it in the condition is evaluated. With
codegen on, the check is
+ // generated just before the first conjunct that references `x`, or
after all of them, so
+ // after a NULL conjunct it has to be generated before the conjuncts it
precedes.
+ //
+ // The common `b * 2` takes the CSE path of FilterExec. The overflowing
`y + 1` is not
+ // common because `GeneratePredicate` (codegen off) evaluates common
subexpressions first.
+ val cseQuery = "SELECT a FROM t WHERE x IS NOT NULL AND NOT (a > 0 AND b
> 0) 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) {
+ skipsAtNull =>
+ // `isnotnull(x)` is inferred from `x > 0` and precedes the
overflowing conjunct.
+ checkAnswer(
+ sql("SELECT a FROM t WHERE NOT (a > 0 AND b > 0) AND y + 1 > 0
AND x > 0"),
+ Row(0))
+ // A user-written check that no other conjunct references.
+ checkAnswer(
+ sql("SELECT a FROM t WHERE x IS NOT NULL AND NOT (a > 0 AND b >
0) AND y + 1 > 0"),
+ Row(0))
+ // The same with a common subexpression.
+ checkAnswer(sql(cseQuery), Row(0))
+ // A check written after the overflowing conjunct stays after it.
+ val checkAfter =
+ "SELECT a FROM t WHERE NOT (a > 0 AND b > 0) AND y + 1 > 0 AND x
IS NOT NULL"
+ if (skipsAtNull) {
+ checkAnswer(sql(checkAfter), Row(0))
+ } else {
+ checkArithmeticOverflow(checkAfter)
+ }
+ }
+ }
+ // The aggregate FILTER clause goes through the same code generation.
+ withNullConjunctConfs(SQLConf.ANSI_ENABLED.key -> "true") { _ =>
+ checkAnswer(
+ sql("SELECT count(1) FILTER (WHERE x IS NOT NULL AND NOT (a > 0 AND
b > 0) AND " +
+ "y + 1 > 0) FROM t"),
+ Row(1))
+ }
+ }
+ }
+
+ test("SPARK-60315: a filter checks IS NOT NULL on an expression after the
conjunct it follows") {
+ withNullConjunctTable(Int.MaxValue) {
+ // `cast(s AS INT)` raises on the first row. With codegen on, a check on
an expression is
+ // generated just before the first conjunct that contains the expression
(SPARK-51356),
+ // even when it is written after that conjunct. After a NULL conjunct,
it has to stay after
+ // the conjunct, as with codegen off: `coalesce(b, cast(s AS INT)) < 0`
is FALSE on the row
+ // and drops it before the check raises, while `coalesce(b, cast(s AS
INT)) < 2` is TRUE and
+ // the check raises after it. The second row passes the latter condition.
+ //
+ // The common `b * 2` takes the CSE path of FilterExec.
+ def query(conjunct: String, cse: Boolean): String = {
+ val common = if (cse) "b * 2 > 0 AND b * 2 > -1 AND " else ""
+ s"SELECT a FROM t WHERE NOT (a > 0 AND b > 0) AND $common$conjunct AND
" +
+ "cast(s AS INT) IS NOT NULL"
+ }
+ def multiplyExactCount(filterExecCSE: Boolean): Int = {
+ withSQLConf(
+ SQLConf.ANSI_ENABLED.key -> "true",
+ SQLConf.SUBEXPRESSION_ELIMINATION_FILTER_EXEC_ENABLED.key ->
filterExecCSE.toString) {
+ val plan = sql(query("coalesce(b, cast(s AS INT)) < 0", cse =
true)).queryExecution
+ "multiplyExact".r.findAllIn(codegenString(plan.executedPlan)).length
+ }
+ }
+ assert(multiplyExactCount(filterExecCSE = true) <
multiplyExactCount(filterExecCSE = false))
+
+ Seq(true, false).foreach { cse =>
+ withNullConjunctConfs(
+ SQLConf.ANSI_ENABLED.key -> "true",
+ SQLConf.SUBEXPRESSION_ELIMINATION_FILTER_EXEC_ENABLED.key ->
cse.toString) {
+ skipsAtNull =>
+ checkAnswer(sql(query("coalesce(b, cast(s AS INT)) < 0", cse)),
Seq.empty)
+ val raising = query("coalesce(b, cast(s AS INT)) < 2", cse)
+ if (skipsAtNull) {
+ checkAnswer(sql(raising), Row(0))
+ } else {
+ checkError(
+ exception = intercept[SparkNumberFormatException] {
+ sql(raising).collect()
+ },
+ condition = "CAST_INVALID_INPUT",
+ parameters = Map(
+ "expression" -> "'x'",
+ "sourceType" -> "\"STRING\"",
+ "targetType" -> "\"INT\"",
+ "ansiConfig" -> toSQLConf(SQLConf.ANSI_ENABLED.key)),
+ context = ExpectedContext(
+ fragment = "cast(s AS INT)",
+ start = raising.lastIndexOf("cast(s AS INT)"),
+ stop = raising.lastIndexOf("cast(s AS INT)") + 13))
+ }
+ }
+ }
+ }
+ }
+
+ test("SPARK-60315: an aggregate FILTER evaluates the conjuncts after a NULL
one") {
+ withNullConjunctTable(Int.MaxValue) {
+ val query = "SELECT count(a) FILTER (WHERE NOT (a > 0 AND b > 0) AND y +
1 > 0) FROM t"
Review Comment:
Minor (pre-existing, not from this PR): the aggregate `FILTER` clause also
differs between codegen on and off in the opposite direction.
`AggregateCodegenSupport.scala:237-238` eliminates the common subexpressions of
the update expressions without their `FILTER` and evaluates them before the `do
{ ... } while (false)` of each `FILTER` (L270-272 and L297-305 there; the
with-keys paths of `HashAggregateExec` do the same), while codegen off
evaluates `If(filter, updateExpr, buffer)`
(`AggregationIterator.scala:166-169`). So with ANSI on, `SELECT avg(y + 1)
FILTER (WHERE b < 0) FROM t` on this table raises `ARITHMETIC_OVERFLOW` with
whole-stage codegen on, because `Average` uses `y + 1` twice, and returns NULL
with it off, on master too. `spark.sql.subexpressionElimination.enabled=false`
does not turn this off. The tests here use `count(a)` and `count(1)`, which
have no common subexpression. Could you add it to "Not fixed here" or file a
JIRA?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala:
##########
@@ -186,13 +290,24 @@ trait GeneratePredicateHelper extends PredicateHelper {
// This has the property of not doing redundant IsNotNull checks and
taking better advantage of
// short-circuiting, not loading attributes until they are needed.
// This is very perf sensitive.
+ // With the NULL-conjunct flag, once a predicate that can be NULL is
generated, the IsNotNull
+ // checks that precede a later predicate in the condition are generated
before it, see
+ // `precedingNotNullChecks`, and a check on an expression that is written
after the predicate
+ // is not generated before it, see the nested checks below.
// NOTE: `FilterExec.doConsume`'s CSE branch inlines the same interleaving
so it can
Review Comment:
Nit (follow-up): as cloud-fan noted in the first review, this ordering is
mirrored by hand in the CSE branch of `FilterExec.doConsume`, and this PR adds
two rules to both copies (the preceding checks at L305-308 and L529-541, and
the nested skip at L334 and L560). SPARK-60329 and SPARK-60377 will each need
the same pair of edits, and a mismatch shows only for filters with a non-cheap
common subexpression, which is why three of the new tests add `y + 1 > -1` or
`b * 2 > 0 AND b * 2 > -1` and count `addExact` or `multiplyExact`. Earlier
CSE-only bugs came from the same split, e.g. the one fixed by a4399f0c6e7,
which added this NOTE. Could a follow-up compute the order once, as a list of
checks and conjuncts that both emitters consume, ideally before SPARK-60329 and
SPARK-60377?
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/WholeStageCodegenSuite.scala:
##########
@@ -2814,4 +2815,274 @@ 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
Review Comment:
Nit: the second row also fails `(a > 0 OR b < 0)` (L2903) and `coalesce(b,
cast(s AS INT)) < 0` (L3051), not only `b < 0`. Could you list them, or say
"unless a test says otherwise"?
--
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]
