MaxGekk commented on code in PR #59573:
URL: https://github.com/apache/spark/pull/59573#discussion_r4326311343
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala:
##########
@@ -215,42 +323,44 @@ trait GeneratePredicateHelper extends PredicateHelper {
if !generatedIsNotNullChecks(idx) && !n.isInstanceOf[Attribute] &&
Review Comment:
Done in 1fce0247cf8. In both emitters, `afterNullable` is captured before
`nullableSeen` is updated, so the first conjunct that can be NULL keeps its
nested check, and the nested branch is skipped once `afterNullable` is set. The
checks written before the conjunct are already generated by
`precedingNotNullChecks` by then, and the columns of the expression stay
covered by the synthetic `IsNotNull(r)`.
The new test `SPARK-60315: a filter checks IS NOT NULL on an expression
after the conjunct it follows` runs your query in the non-CSE and CSE paths (`b
* 2 > 0 AND b * 2 > -1` added for the latter, with the `multiplyExact` count
asserted): the row is dropped in every combination when `coalesce(b, cast(s AS
INT)) < 0` is FALSE, and the query raises `CAST_INVALID_INPUT` when the
conjunct is `< 2` and TRUE, except with the legacy config, which skips the row
at the NULL conjunct. The SPARK-51356 tests still pass.
##########
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. 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`, unless 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 statis
tics. 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:
Right, thanks. In 1fce0247cf8 the entry lists it as the one exception: an
`IS NOT NULL` check on a column that a later conjunct references is still
evaluated before that conjunct, even when it is written after it, and can drop
the row before an error the conjunct raises. The `precedingNotNullChecks`
scaladoc says the same, with the reason (the conjunct is bound with the column
non-nullable) and that such a check only drops a row that `And.eval` drops too,
unless the conjunct raises first. The first paragraph of the PR description now
says "keeps evaluating the conjuncts after a NULL one", as you suggested, and
the "Not fixed here" part mentions this placement next to SPARK-60329.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/WholeStageCodegenSuite.scala:
##########
@@ -2814,4 +2815,219 @@ 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 the interpreted predicate evaluates its common
subexpressions up front.
Review Comment:
Thanks, you are right about `GeneratePredicate`. Took the suggestion in
1fce0247cf8.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala:
##########
@@ -137,47 +137,146 @@ 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,
+ 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)) {
+ 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;"
+ }
+ }
+
+ /** Wraps the checks of all conjuncts in `code` with the declaration and
test of `nullFlag`. */
+ protected def withNullConjunctFlag(code: String, nullFlag: Option[String]):
String = {
+ nullFlag match {
+ case Some(flag) =>
+ s"""
+ |boolean $flag = false;
+ |$code
+ |if ($flag) continue;
+ """.stripMargin
+ case None => code
+ }
+ }
+
+ /**
+ * Returns the checks of `notNullPreds` that precede the k-th other conjunct
in the condition
+ * and are not generated yet, and marks them generated, when `afterNullable`
is set, or Nil.
+ *
+ * With the NULL-conjunct flag, a NULL conjunct no longer drops the row, so
the conjuncts after
+ * it are evaluated in the order of the condition, as `And.eval` does: a
FALSE IsNotNull check
+ * drops the row before an error a later conjunct raises on it
(SPARK-60315). `afterNullable`
+ * is set once a conjunct that can be NULL has been generated. The conjuncts
before it are
+ * never reached after a NULL conjunct, so the checks before them keep the
lazy placement.
+ */
+ protected def precedingNotNullChecks(
+ notNullPreds: Seq[Expression],
+ precedingNotNulls: Seq[Int],
+ k: Int,
+ afterNullable: Boolean,
+ generated: Array[Boolean]): Seq[Expression] = {
+ if (afterNullable) {
+ notNullPreds.take(precedingNotNulls(k)).zipWithIndex.collect {
+ case (p, idx) if !generated(idx) =>
+ generated(idx) = true
+ p
+ }
+ } else {
+ Nil
+ }
+ }
+
protected def generatePredicateCode(
ctx: CodegenContext,
inputAttrs: Seq[Attribute],
inputExprCode: Seq[ExprCode],
outputAttrs: Seq[Attribute],
notNullPreds: Seq[Expression],
otherPreds: Seq[Expression],
+ boundOtherPreds: Seq[Expression],
Review Comment:
Removed in 1fce0247cf8, from the signature and both callers, and the comment
now says that the caller binds `boundOtherPreds` against the output with the
checked columns tightened to non-nullable.
--
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]
