cloud-fan commented on code in PR #59573: URL: https://github.com/apache/spark/pull/59573#discussion_r4323423318
########## 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: **Nit (P3):** This says that after a NULL conjunct the remaining conjuncts are evaluated "in the order they appear in the condition", and that only a `FALSE` conjunct or an `IS NOT NULL` check that precedes the raising conjunct can drop the row first. The `precedingNotNullChecks` scaladoc in `basicPhysicalOperators.scala` makes the same claim. However, both emitters still generate `IsNotNull(x)` just before the first conjunct that references `x`, even when the check is written after that conjunct, because that conjunct is bound with `x` non-nullable. For example, with ANSI on and the row `(a, b, y, x) = (NULL, 1, 2147483647, NULL)`: ```sql SELECT a FROM t WHERE NOT (a > 0 AND b > 0) AND (y + 1 > 0 OR x > 0) AND x IS NOT NULL ``` Nothing is inferred from the `OR`, so the user-written check is the only one. With whole-stage codegen on, `isnotnull(x)` runs before the `OR` and drops the row. With it off, `And` evaluates the `OR` after the NULL conjunct and fails with `ARITHMETIC_OVERFLOW`. Master also drops this row, at the NULL conjunct, so the runtime result is not new. But the guide now promises a parity that the code does not provide for this shape. Could the entry and the scaladoc list it as an exception: a check on a column that the raising conjunct references runs before that conjunct, even when it is written after it? This extends an [existing discussion](https://github.com/apache/spark/pull/59573#discussion_r4309696444). **Verification:** - **Inspection:** The migration-guide entry and the scaladoc each state that, after a NULL conjunct, an IS NOT NULL check on a column that a later conjunct references may run before that conjunct. Neither promises evaluation strictly in the order of the condition. ########## 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: **Nit (P3):** With `boundOtherPreds` passed in, nothing in this method reads `outputAttrs` (the parameter above) anymore. `genPredicate` takes the bound expression, and `genNotNull` binds against `inputAttrs`. Both callers, the 4-argument overload and `FilterExec.doConsume`, still compute and pass it, so the signature suggests a binding the method no longer does. Could you drop `outputAttrs` here and at both call sites, and change the comment below to say that callers bind `boundOtherPreds` against the tightened output? **Verification:** - **Compile:** The SQL core module compiles with the parameter removed from the shared helper and from both of its callers. - **Compatibility:** Filter and aggregate FILTER results and errors are unchanged across whole-stage codegen on and off, the legacy config on and off, and the filterExec CSE path on and off. ########## 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: **Nit (P3):** `InterpretedPredicate` evaluates common subexpressions lazily, through `ExpressionProxy` and the `SubExprEvaluationRuntime` cache. The predicate that evaluates them up front is `GeneratePredicate`, which runs `subexprFunctionsCode` before the condition. That is what `FilterExec` gets from `Predicate.create` with whole-stage codegen off, under the default factory mode. ```suggestion // common because `GeneratePredicate` (codegen off) evaluates common subexpressions first. ``` **Verification:** - **Inspection:** The comment that explains why the overflowing addition is not a common subexpression attributes the up-front evaluation to the generated predicate used when whole-stage codegen is off, not to the interpreted predicate. ########## 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: Thanks, I checked 3264b00d897. After a nullable conjunct, the checks written before each later conjunct now run before it in both emitters. The inferred, guard-first, and aggregate FILTER cases drop the row, and a guard written after the overflowing conjunct still raises. Placing expression checks the same way makes sense: at their own position they run only on rows codegen off also evaluates them on. One remaining wording point: the migration guide and the precedingNotNullChecks scaladoc still promise condition-order evaluation, but a check written after a conjunct that references its column is still generated before that conjunct. See the review comment on docs/sql-migration-guide.md. <!-- SPARK_DEV_REVIEW_REPLY {"feedback_id":"inline:4309696444","thread_id":"inline:4309696444","verdict_sha256":"c6775db45c1326a69ec4232e7dc0b7dcfc9209f4dde7e4ed49ed9b2788d14034"} --> -- 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]
