You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ma...@apache.org on 2026/10/22 07:39:09 UTC
(spark) branch branch-4.x updated: [SPARK-60315][SQL] Fix inconsistent filter behavior with whole-stage codegen on and off when a conjunct is NULL
This is an automated email from the ASF dual-hosted git repository.
MaxGekk pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new ba2dec7f05df [SPARK-60315][SQL] Fix inconsistent filter behavior with whole-stage codegen on and off when a conjunct is NULL
ba2dec7f05df is described below
commit ba2dec7f05df27ef6b1a8d6427bc7be6ffdc8542
Author: Max Gekk <ma...@gmail.com>
AuthorDate: Thu Oct 22 09:38:44 2026 +0200
[SPARK-60315][SQL] Fix inconsistent filter behavior with whole-stage codegen on and off when a conjunct is NULL
### What changes were proposed in this pull request?
This PR fixes inconsistent behavior of a filter between whole-stage codegen on and off: when an earlier conjunct of the condition is NULL on a row and a later one raises an error on that row, the query failed with codegen off but returned rows with codegen on. After this PR, codegen keeps evaluating the conjuncts after a NULL one, so it raises the error in that case too. An `IS NOT NULL` check that precedes the raising conjunct in the condition still drops the row first in both modes.
Whole-stage codegen of a filter condition (`FilterExec`) and of the `FILTER` clause of an aggregate function (`AggregateCodegenSupport`) generates each top-level conjunct separately and drops the row as soon as one of them is FALSE **or NULL** (`if (isNull || !value) continue;`). The interpreted path evaluates the whole condition with `And.eval`, which stops only at FALSE. So when an earlier conjunct is NULL and a later one raises on that row, the query fails with codegen off and retu [...]
This PR changes the generated code so that only a FALSE conjunct stops the evaluation:
- `GeneratePredicateHelper` gets six helpers: `splitNotNullPreds` (splits the condition into the `IsNotNull` checks and the other conjuncts, keeping the number of checks that precede each conjunct), `mayRaise` (whether an expression can raise an error, is nondeterministic or may have side effects, via `ExprUtils.canEvaluateUnconditionally`), `nullConjunctFlag` (a fresh flag name, or `None`), `conjunctCheck` (the per-conjunct check), `withNullConjunctFlag` (declares the flag before the [...]
- A NULL conjunct sets the flag, and the remaining conjuncts are still evaluated: `if (isNull) { flag = true; } else if (!value) continue;`. The row is dropped after all of them: `if (flag) continue;`.
- The flag is generated only when some conjunct can be NULL after binding, and a conjunct after the first such one, or one of the `IsNotNull` checks, can raise an error, is nondeterministic or may have side effects (`ExprUtils.canEvaluateUnconditionally`). Otherwise the evaluation after a NULL conjunct is not observable and the generated code is unchanged, e.g. for `WHERE (a > 5 OR b > 5) AND c = 1 AND d = 2`. The `IsNotNull` checks are never NULL and keep their early `continue`, so t [...]
- With the flag, once a conjunct that can be NULL has been generated, the `IsNotNull` checks that precede a later conjunct in the condition are generated before it when the conjunct or one of the checks can raise, as `And.eval` evaluates them, instead of lazily before the first conjunct that references their column. So a FALSE check drops the row before an error the later conjunct raises on it, as with codegen off (`splitNotNullPreds` keeps the number of checks that precede each conju [...]
- With the flag, after a conjunct that can be NULL, an `IS NOT NULL` check on an expression (SPARK-51356) that is written after the conjunct containing the expression is no longer generated before that conjunct. It stays at its position, as `And.eval` evaluates it, so it does not raise an error on a row that the conjunct drops as FALSE.
- Both code paths are changed: `generatePredicateCode`, which the aggregate `FILTER` clause also uses, and the CSE branch of `FilterExec.doConsume`, which the code comments require to stay in step with it.
- New legacy config `spark.sql.legacy.filterShortCircuitOnNullConjunct` (internal, default `false`, version 4.4.0). Set to `true`, it restores the previous codegen behavior. The interpreted path is unchanged in both modes.
Join conditions and `MergeRowsExec` generate the whole condition through `And.doGenCode`, which already evaluates the right side after a NULL left side, so they are not changed.
Not fixed here, on master as well as after this PR:
- With whole-stage codegen on, an `IS NOT NULL` check on a column is generated just before the first conjunct that references the column, so `WHERE x IS NOT NULL AND y + 1 > 0 AND x > 0` raises `ARITHMETIC_OVERFLOW` with codegen on and drops the row with codegen off. That is tracked in SPARK-60329. The check is also generated before that conjunct when it is written after it, and so before the conjuncts between that one and its own position, and a check on an expression over the column [...]
- Before the first conjunct that can be NULL, including that conjunct itself, or with the legacy config, an `IS NOT NULL` check on an expression is still generated before the first conjunct that contains the expression, even when it is written after it, so `WHERE coalesce(b, cast(s AS INT)) < 0 AND cast(s AS INT) IS NOT NULL` raises `CAST_INVALID_INPUT` with codegen on and drops the row with codegen off on `(b, s) = (1, 'x')`. That is tracked in SPARK-60377.
- With whole-stage codegen off, `GeneratePredicate` evaluates a subexpression shared by several conjuncts before the condition, so `WHERE NOT (a > 0 AND b > 0) AND b < 0 AND y + 1 > 0 AND y + 1 > -1` raises `ARITHMETIC_OVERFLOW` with codegen off and returns no rows with codegen on, where the FALSE `b < 0` stops the evaluation. That is tracked in SPARK-60392.
- Whole-stage codegen evaluates the common subexpressions of the update expressions of an aggregate function before its `FILTER` clause, so `SELECT avg(y + 1) FILTER (WHERE b < 0) FROM t` raises `ARITHMETIC_OVERFLOW` with codegen on and returns NULL with codegen off. That is tracked in SPARK-60393.
- The interleaving of the checks and the conjuncts is mirrored by hand in `generatePredicateCode` and in the CSE branch of `FilterExec.doConsume`. SPARK-60394 is to compute it once for both.
#### SQL standard
- 7.11 `<where clause>`, GR 2 (and 10.9 `<aggregate function>`, GR 4a, for `<filter clause>`): the search condition is effectively evaluated for each row, and the result keeps the rows for which it is True.
- 6.35 `<boolean value expression>`, Table 11: Unknown AND True is Unknown, Unknown AND False is False. After an Unknown conjunct the result of the search condition still depends on the remaining ones.
- 9075-1 "Rule evaluation order" (SQL-92 3.3.4.4): only the *inessential* parts, those not needed to determine the result of the expression or search condition, may be skipped, and whether their exceptions are raised is implementation-dependent. After a NULL conjunct the rest is essential. The order of evaluation is implementation-dependent too, but in every order the raising conjunct is either evaluated first or is still essential, so the exception is raised. Only a FALSE conjunct ma [...]
#### Other DBMSes
Table `t(a INT, b INT, y)` with rows `(NULL, 1, <raises>)` and `(0, 1, <ok>)`. `<E>` raises on the first row:
- `y + 1 > 0` with `y` at the maximum of its type: `INT`, or `BIGINT` on MySQL and MariaDB, which widen `INT + INT`;
- `10 / y > 0` with `y = 0` on Oracle, whose `NUMBER` does not overflow at 32 bits;
- `abs(y) > 0` with `y` at the minimum `BIGINT` on SQLite, where `+` switches to floating point.
The images and credentials are those of `connector/docker-integration-tests`.
| Query | PostgreSQL 18.2 | MySQL 9.6.0 | MariaDB 12.2.2 | SQL Server 2022 | Oracle Free 23.26 | DB2 11.5.9 | H2 2.3.232 | Derby 10.16 | SQLite 3.46 |
|---|---|---|---|---|---|---|---|---|---|
| `WHERE NOT (a > 0 AND b > 0) AND <E>` | skip | skip | skip | **raise** | skip | **raise** | **raise** | skip | skip |
| `WHERE <E> AND NOT (a > 0 AND b > 0)` | raise | raise | raise | raise | **skip** | raise | raise | **skip** | raise |
| `WHERE a > 0 AND <E>` | skip | skip | skip | skip | skip | **raise** | **raise** | skip | skip |
| `WHERE b < 0 AND <E>` (FALSE first) | - | - | - | - | - | raise | - | - | - |
| `count(a) FILTER (WHERE NOT (a > 0 AND b > 0) AND <E>)` | **raise** | n/a | n/a | n/a | **raise** | n/a | **raise** | n/a | skip |
"-" means no error and an empty result; "n/a" means the `FILTER` clause is not supported.
- In WHERE, most engines drop the row at the NULL conjunct. Oracle and Derby do it even when the raising conjunct is written first, so they reorder.
- DB2 and H2 raise. SQL Server raises for the JIRA's shape but not for a plain `a > 0` first.
- For the aggregate `FILTER` clause, PostgreSQL, Oracle and H2 evaluate the whole `AND` and raise, as this PR does.
We considered the opposite fix: make the interpreted path stop at the first NULL conjunct too. It matches most engines' WHERE behavior, but it departs from the standard's text and from what Spark already does with codegen off. Setting `spark.sql.legacy.filterShortCircuitOnNullConjunct` to `true` only restores the previous whole-stage codegen behavior; with whole-stage codegen off, the query still raises.
### Why are the changes needed?
The same query must not raise with whole-stage codegen off and return rows with it on (SPARK-60315). Reproducer (ANSI mode):
```sql
CREATE TABLE t USING parquet AS
SELECT * FROM VALUES (NULL, 1, 2147483647), (0, 1, 0) AS v(a, b, y);
-- The rows can be written to different files, and `NOT (a > 0 AND b > 0)` is pushed down as
-- `a <= 0 OR b <= 0`, which skips the first row's file by its statistics.
SET spark.sql.parquet.filterPushdown=false;
SET spark.sql.codegen.wholeStage=true;
SELECT a FROM t WHERE NOT (a > 0 AND b > 0) AND y + 1 > 0;
-- before: 0
-- after: [ARITHMETIC_OVERFLOW] overflow. ... SQLSTATE: 22003
SET spark.sql.codegen.wholeStage=false;
SELECT a FROM t WHERE NOT (a > 0 AND b > 0) AND y + 1 > 0;
-- before and after: [ARITHMETIC_OVERFLOW] overflow. ... SQLSTATE: 22003
```
The same happens in non-ANSI mode with year-month interval arithmetic, and with the aggregate `FILTER` clause.
### Does this PR introduce _any_ user-facing change?
Yes. With whole-stage codegen on, an error raised by a later conjunct on a row where an earlier conjunct is NULL now fails the query, as it already did with codegen off. A row can still be skipped before the filter, for example by Parquet row-group statistics, and then no error is raised.
The later conjuncts also run on more rows than before:
- a nondeterministic conjunct such as `rand(seed)` can return different values on the remaining rows;
- a side-effecting UDF is called on rows that are then dropped;
- an expensive later conjunct costs more on rows where an earlier, nullable conjunct is NULL.
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, so a check written before a conjunct can now raise an error on a row that the conjunct dropped before, as it does with codegen off. Setting `spark.sql.legacy.filterShortCircuitOnNullConjunct=true` restores the previous behavior. The migration guide has an entry for this.
### How was this patch tested?
Six new tests in `WholeStageCodegenSuite`. Each runs with whole-stage codegen on and off, and with the legacy config on and off; it expects the row to be skipped only with codegen on and the legacy config set, and the error in every other combination:
- the ANSI reproducer, and a variant with a common `y + 1` that takes the CSE path of `FilterExec` (checked from the generated code, with `subexpressionElimination.filterExec.enabled` on and off), and a NULL conjunct from `OR`; without an error, the CSE path drops the row whose condition is NULL;
- the non-ANSI year-month interval reproducer (`INTERVAL_ARITHMETIC_OVERFLOW`), and a user-written `cast(s AS INT) IS NOT NULL` check, which is FALSE on the row, before the NULL conjunct: the row is dropped in every combination;
- a NULL conjunct followed by a FALSE one, which must not raise; it also checks that no flag is generated when no conjunct can be NULL, or when nothing after the first conjunct that can be NULL can raise, and that it is generated once a conjunct can raise;
- an `IS NOT NULL` check on a column that is FALSE on the row and precedes the overflowing conjunct in the condition, inferred or user-written, in the non-CSE and CSE paths of `FilterExec` and in the aggregate `FILTER` clause: the row is dropped in every combination. A check written after the overflowing conjunct stays after it, in both paths, so the query still raises;
- an `IS NOT NULL` check on `cast(s AS INT)`, which raises on the row, written after a conjunct that contains the expression, in the non-CSE and CSE paths: the row is dropped in every combination when the conjunct is FALSE, and the query raises `CAST_INVALID_INPUT` when it is TRUE. The same check written before a conjunct that can be NULL raises on a row where that conjunct is TRUE and the next one FALSE, since the checks after the first conjunct that can be NULL are placed in the ord [...]
- the aggregate `FILTER` clause.
The tests use a Parquet table with filter pushdown off, so the filter is neither evaluated on a local relation by the optimizer nor skipped by row-group statistics.
Also ran `WholeStageCodegenSuite`, `SQLQuerySuite`, `DataFrameSuite`, `DataFrameAggregateSuite`, `SubexpressionEliminationSuite`, `FilterPushdownSuite` and `SQLQueryTestSuite` locally. The only failure is `udtf/udtf.sql - Python UDTFs`, whose golden output differs from the local Python's `ValueError` message (`'input' is not in list` vs `list.index(x): x not in list`), which is unrelated to this change.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 5.5)
Closes #59573 from MaxGekk/SPARK-60315-filter-null-conjunct.
Authored-by: Max Gekk <ma...@gmail.com>
Signed-off-by: Max Gekk <ma...@gmail.com>
(cherry picked from commit 930c571628d7ca0bd94e455b1532e3fc2bbdca3a)
Signed-off-by: Max Gekk <ma...@gmail.com>
---
docs/sql-migration-guide.md | 1 +
.../org/apache/spark/sql/internal/SQLConf.scala | 20 ++
.../sql/execution/basicPhysicalOperators.scala | 311 +++++++++++++++++----
.../sql/execution/WholeStageCodegenSuite.scala | 311 ++++++++++++++++++++-
4 files changed, 582 insertions(+), 61 deletions(-)
diff --git a/docs/sql-migration-guide.md b/docs/sql-migration-guide.md
index 1e6c8891679a..72947866f4aa 100644
--- a/docs/sql-migration-guide.md
+++ b/docs/sql-migration-guide.md
@@ -52,6 +52,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` [...]
- 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 [...]
- 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 affecte [...]
+- 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, which also evaluates a subexpression shared by several conjuncts before the condition. Only a `FALSE` conjunct st [...]
- Since Spark 4.4, when a format is specified, `to_timestamp`, `try_to_timestamp`, and `to_timestamp_ltz` cast a numeric first argument to a string and parse it with the format, for example `to_timestamp(20201231, 'yyyyMMdd')` returns `2020-12-31 00:00:00`. This matches the existing behavior when `spark.sql.timestampType` is `TIMESTAMP_NTZ`. Previously, with the default `TIMESTAMP_LTZ` timestamp type, such calls failed with `INTERNAL_ERROR`. Note that the string form of the number is par [...]
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
index f957a9363345..f0fb0fc026da 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
@@ -1763,6 +1763,23 @@ object SQLConf {
.booleanConf
.createWithDefault(true)
+ val LEGACY_FILTER_SHORT_CIRCUIT_ON_NULL_CONJUNCT =
+ buildConf("spark.sql.legacy.filterShortCircuitOnNullConjunct")
+ .internal()
+ .doc("When true, whole stage codegen of a filter condition, including the FILTER clause " +
+ "of an aggregate function, drops a row at the first top-level conjunct that evaluates " +
+ "to NULL without evaluating the remaining conjuncts, so an error that one of them " +
+ "would raise on that row is not raised. When false, only a FALSE conjunct stops the " +
+ "evaluation, as the SQL standard requires and as the interpreted evaluation does, and " +
+ "the row is dropped after the remaining conjuncts are evaluated. Once a condition has " +
+ "a conjunct that can be NULL, the IS NOT NULL checks after that conjunct are then " +
+ "evaluated in the order of the condition on every row, not only on rows where a " +
+ "conjunct is NULL.")
+ .version("4.4.0")
+ .withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE)
+ .booleanConf
+ .createWithDefault(false)
+
val CASE_SENSITIVE = buildConf(SqlApiConfHelper.CASE_SENSITIVE_KEY)
.internal()
.doc("Whether the query analyzer should be case sensitive or not. " +
@@ -9618,6 +9635,9 @@ class SQLConf extends Serializable with Logging with SqlApiConf {
def subexpressionEliminationFilterExecEnabled: Boolean =
getConf(SUBEXPRESSION_ELIMINATION_FILTER_EXEC_ENABLED)
+ def legacyFilterShortCircuitOnNullConjunct: Boolean =
+ getConf(LEGACY_FILTER_SHORT_CIRCUIT_ON_NULL_CONJUNCT)
+
def autoBroadcastJoinThreshold: Long = getConf(AUTO_BROADCASTJOIN_THRESHOLD)
def limitInitialNumPartitions: Int = getConf(LIMIT_INITIAL_NUM_PARTITIONS)
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala
index d8314b8807a0..3c94cab249b5 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala
@@ -137,47 +137,182 @@ 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)
+ }
+
+ /**
+ * Whether evaluating `e` on a row can be observed other than through its value: it can raise
+ * an error, is nondeterministic or may have side effects, such as a UDF. Such an expression
+ * has to be evaluated on the rows, and in the order, that `And.eval` evaluates it on.
+ */
+ protected def mayRaise(e: Expression): Boolean = !ExprUtils.canEvaluateUnconditionally(e)
+
+ /**
+ * 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), if none of `otherPreds`,
+ * bound with the nullability they are generated with in `boundOtherPreds`, can be NULL, or if
+ * neither a conjunct after the first one that can be NULL nor one of the `notNullPreds` checks
+ * `mayRaise`: the evaluation after a NULL conjunct is then not observable, and the generated
+ * code is the same as without the flag.
+ *
+ * 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,
+ otherPreds: Seq[Expression],
+ boundOtherPreds: Seq[Expression],
+ notNullPreds: Seq[Expression]): Option[String] = {
+ val firstNullable = boundOtherPreds.indexWhere(_.nullable)
+ if (conf.legacyFilterShortCircuitOnNullConjunct || firstNullable == -1 ||
+ !(otherPreds.drop(firstNullable + 1).exists(mayRaise) || notNullPreds.exists(mayRaise))) {
+ 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 `conjunct` in the
+ * condition and are not generated yet, and marks them generated, when `afterNullable` is set
+ * and the conjunct or one of these checks `mayRaise`, 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. So
+ * do the checks before a conjunct when neither it nor one of them can raise: their order is
+ * not observable, and the checks are still generated before the next conjunct that can raise.
+ *
+ * The exception to the order is a check on a column, or on an expression over the column, that
+ * a later conjunct references: the check on the column is generated before the first such
+ * conjunct even when it is written after it, since the conjunct is bound with the column
+ * non-nullable, and so before the conjuncts between that one and its own position. The check
+ * is never NULL, so it only drops a row that `And.eval` drops too, unless one of these
+ * conjuncts raises an error on the row first. Also, the first conjunct that can be NULL keeps
+ * the checks on the expressions it contains before it (SPARK-51356).
+ */
+ protected def precedingNotNullChecks(
+ notNullPreds: Seq[Expression],
+ precedingNotNulls: Seq[Int],
+ k: Int,
+ conjunct: Expression,
+ afterNullable: Boolean,
+ generated: Array[Boolean]): Seq[Expression] = {
+ if (afterNullable) {
+ val pending = notNullPreds.take(precedingNotNulls(k)).zipWithIndex.filterNot {
+ case (_, idx) => generated(idx)
+ }
+ if (pending.nonEmpty && (mayRaise(conjunct) || pending.exists(p => mayRaise(p._1)))) {
+ pending.map { case (p, idx) =>
+ generated(idx) = true
+ p
+ }
+ } else {
+ Nil
+ }
+ } 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],
+ precedingNotNulls: Seq[Int],
nonNullAttrExprIds: Seq[ExprId]): String = {
+ // The IsNotNull checks are never NULL, so only `otherPreds` can need the flag. The caller
+ // binds them against the output with the checked columns tightened to non-nullable, in
+ // `boundOtherPreds`.
+ val nullFlag = nullConjunctFlag(ctx, otherPreds, boundOtherPreds, notNullPreds)
+
/**
- * Generates code for `c`, using `in` for input attributes and `attrs` for nullability.
+ * Generates code for `c`, given as `bound` after binding against the attributes whose
+ * nullability it is generated with. A NULL value sets `flag`, or drops the row if None.
*/
- def genPredicate(c: Expression, in: Seq[ExprCode], attrs: Seq[Attribute]): String = {
- val bound = BindReferences.bindReference(c, attrs)
- val evaluated = evaluateRequiredVariables(inputAttrs, in, c.references)
+ def genPredicate(c: Expression, bound: Expression, flag: Option[String]): String = {
+ val evaluated = evaluateRequiredVariables(inputAttrs, inputExprCode, c.references)
// Generate the code for the predicate.
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, flag)}
""".stripMargin
}
+ /**
+ * Generates an IsNotNull check with the nullability the child produced. The check is never
+ * NULL, and must drop the row before the conjuncts bound with its column non-nullable, so
+ * it does not get the flag.
+ */
+ def genNotNull(c: Expression): String =
+ genPredicate(c, BindReferences.bindReference(c, inputAttrs), None)
+
// To generate the predicates we will follow this algorithm.
// For each predicate that is not IsNotNull, we will generate them one by one loading attributes
// as necessary. For each of both attributes, if there is an IsNotNull predicate we will
@@ -186,13 +321,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
+ val precedingChecks = precedingNotNullChecks(
+ notNullPreds, precedingNotNulls, k, c, afterNullable,
+ generatedIsNotNullChecks).map(genNotNull).mkString("\n").trim
+ nullableSeen ||= bound.nullable
+
val nullChecks = c.references.map { r =>
val idx = notNullPreds.indexWhere {
case IsNotNull(n) => n.semanticEquals(r)
@@ -201,44 +347,58 @@ 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)
- } else if (nonNullAttrExprIds.contains(r.exprId) && !extraIsNotNullAttrs.contains(r)) {
+ genNotNull(notNullPreds(idx))
+ } else if ((idx == -1 || !afterNullable) && nonNullAttrExprIds.contains(r.exprId) &&
+ !extraIsNotNullAttrs.contains(r)) {
+ // A synthetic check for a column covered only by a check on an expression, or, before
+ // the flag applies, for a column whose check was generated before an earlier predicate.
+ // After a predicate that can be NULL, the check generated by `precedingNotNullChecks`
+ // already covers the column.
extraIsNotNullAttrs += r
- genPredicate(IsNotNull(r), inputExprCode, inputAttrs)
+ genNotNull(IsNotNull(r))
} else {
""
}
}.mkString("\n").trim
- val nestedNullChecks = notNullPreds.zipWithIndex.collect {
+ // A check on an expression that the predicate contains is generated before the predicate
+ // (SPARK-51356). After a predicate that can be NULL, the checks written before this one
+ // are generated above when one of them or the predicate can raise, and a check written
+ // after it is left to the remaining checks, as `And.eval` evaluates it after the
+ // predicate, so that it does not raise an error on a row that the predicate drops. The
+ // columns of the expression are covered by the checks above.
+ val nestedNullChecks = if (afterNullable) "" else notNullPreds.zipWithIndex.collect {
case (p @ IsNotNull(n), idx)
if !generatedIsNotNullChecks(idx) && !n.isInstanceOf[Attribute] &&
c.exists(_.semanticEquals(n)) =>
generatedIsNotNullChecks(idx) = true
- genPredicate(p, inputExprCode, inputAttrs)
+ genNotNull(p)
}.mkString("\n").trim
// Here we use *this* operator's output with this output's nullability since we already
// enforced them with the IsNotNull checks above.
s"""
+ |$precedingChecks
|$nullChecks
|$nestedNullChecks
- |${genPredicate(c, inputExprCode, outputAttrs)}
+ |${genPredicate(c, bound, nullFlag)}
""".stripMargin.trim
}.mkString("\n")
val nullChecks = notNullPreds.zipWithIndex.map { case (c, idx) =>
if (!generatedIsNotNullChecks(idx)) {
- genPredicate(c, inputExprCode, inputAttrs)
+ genNotNull(c)
} else {
""
}
}.mkString("\n")
- s"""
- |$generated
- |$nullChecks
- """.stripMargin
+ withNullConjunctFlag(
+ s"""
+ |$generated
+ |$nullChecks
+ """.stripMargin,
+ nullFlag)
}
}
@@ -246,18 +406,18 @@ trait GeneratePredicateHelper extends PredicateHelper {
case class FilterExec(condition: Expression, child: SparkPlan)
extends UnaryExecNode with CodegenSupport with GeneratePredicateHelper with SafeForKWayMerge {
- // Split out all the IsNotNulls from condition.
- private val (notNullPreds, otherPreds) = splitConjunctivePredicates(condition).partition {
- case IsNotNull(a) => isNullIntolerant(a) && a.references.subsetOf(child.outputSet)
- case _ => false
- }
+ // Split out all the IsNotNulls from condition, keeping how many of them precede each of the
+ // other conjuncts.
+ private val (notNullPreds, otherPreds, precedingNotNulls) =
+ splitNotNullPreds(condition, child.outputSet)
// The columns that will filtered out by `IsNotNull` could be considered as not nullable.
private val notNullAttributes = notNullPreds.flatMap(_.references).distinct.map(_.exprId)
- // `otherPreds` bound against this operator's `output`, shared between the CSE gate in
- // `doConsume` and the CSE codegen itself. Codegen-only derived state, so `@transient`: it is
- // computed on the driver during code generation and never accessed on executors.
+ // `otherPreds` bound against this operator's `output`, shared by the CSE gate in `doConsume`
+ // and both predicate code paths, the CSE branch and `generatePredicateCode`. Codegen-only
+ // derived state, so `@transient`: it is computed on the driver during code generation and
+ // never accessed on executors.
@transient private lazy val boundOtherPreds: Seq[Expression] =
otherPreds.map(BindReferences.bindReference(_, output))
@@ -311,7 +471,13 @@ case class FilterExec(condition: Expression, child: SparkPlan)
// the tightened-output binding below is safe without evaluating the
// throw-capable inner expression. Then emit the otherPred's CSE precompute
// and body. Remaining notNullPreds (including complex-child forms whose refs
- // got synthetic coverage) emit after all otherPreds.
+ // got synthetic coverage) emit after all otherPreds. With the NULL-conjunct
+ // flag, once an otherPred that can be NULL is emitted, the notNullPreds that
+ // precede a later otherPred in the condition are emitted before it first, when it
+ // or one of them can raise (`precedingNotNullChecks`), and a complex
+ // `IsNotNull(expr)` written after the otherPred is left to the remaining
+ // notNullPreds, so it does not raise on a row the otherPred rejects, as in the
+ // non-CSE branch (SPARK-60315).
// (c) CSE state materialization: binding otherPreds (and the CSE analysis)
// against `output` (with `notNullAttributes` tightened to non-nullable)
// requires that the matching IsNotNull has already short-circuited -- else
@@ -360,6 +526,8 @@ case class FilterExec(condition: Expression, child: SparkPlan)
val subExprs =
ctx.subexpressionEliminationForWholeStageCodegen(otherPredsEquivalentExpressions)
+ val nullFlag = nullConjunctFlag(ctx, otherPreds, boundOtherPreds, notNullPreds)
+
// Group CSE states by the index of the first otherPred that references them.
// `evaluateSubExprEliminationState` recursively emits each state's children
// before the state itself and marks emitted code as `EmptyBlock`, so emitting
@@ -373,29 +541,52 @@ 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. The check does not get
+ // the NULL-conjunct flag: it must drop the row before the otherPreds and CSE
+ // precomputes bound with its column non-nullable, see (c).
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, None)}
""".stripMargin
}
+ // Emit a check on a complex expression with no CSE state: the expression is bound
+ // against child.output, with its original nullability, while the CSE states are
+ // computed over `boundOtherPreds`, bound against the tightened `output`, and a state
+ // precomputed for the check could materialize a NULL into a shared variable, see (c).
+ def genNotNullWithoutCSE(pred: Expression): String = {
+ var checkCode: String = null
+ ctx.withSubExprEliminationExprs(Map.empty) {
+ checkCode = genNotNull(pred)
+ Seq.empty
+ }
+ checkCode
+ }
+
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, orig, afterNullable,
+ generatedIsNotNullChecks).foreach { p =>
+ parts.append(genNotNullWithoutCSE(p))
+ parts.append('\n')
+ }
+ nullableSeen ||= bound.nullable
orig.references.foreach { r =>
val ni = notNullPreds.indexWhere {
case IsNotNull(c) => c.semanticEquals(r)
@@ -405,35 +596,34 @@ case class FilterExec(condition: Expression, child: SparkPlan)
generatedIsNotNullChecks(ni) = true
parts.append(genNotNull(notNullPreds(ni)))
parts.append('\n')
- } else if (notNullAttributes.contains(r.exprId) &&
+ } else if ((ni == -1 || !afterNullable) && notNullAttributes.contains(r.exprId) &&
!extraIsNotNullAttrs.contains(r)) {
+ // As in `generatePredicateCode`: after a nullable otherPred, a check
+ // generated by `precedingNotNullChecks` already covers the column.
extraIsNotNullAttrs += r
parts.append(genNotNull(IsNotNull(r)))
parts.append('\n')
}
}
- notNullPreds.zipWithIndex.foreach {
- case (p @ IsNotNull(n), ni)
- if !generatedIsNotNullChecks(ni) && !n.isInstanceOf[Attribute] &&
- orig.exists(_.semanticEquals(n)) =>
- generatedIsNotNullChecks(ni) = true
- var checkCode: String = null
- ctx.withSubExprEliminationExprs(Map.empty) {
- checkCode = genNotNull(p)
- Seq.empty
- }
- parts.append(checkCode)
- parts.append('\n')
- case _ =>
+ // A complex `IsNotNull(expr)` is skipped after a nullable otherPred, see (b).
+ if (!afterNullable) {
+ notNullPreds.zipWithIndex.foreach {
+ case (p @ IsNotNull(n), ni)
+ if !generatedIsNotNullChecks(ni) && !n.isInstanceOf[Attribute] &&
+ orig.exists(_.semanticEquals(n)) =>
+ generatedIsNotNullChecks(ni) = true
+ parts.append(genNotNullWithoutCSE(p))
+ parts.append('\n')
+ case _ =>
+ }
}
statesByFirstUse.get(idx).foreach { states =>
parts.append(ctx.evaluateSubExprEliminationState(states))
parts.append('\n')
}
val ev = ExpressionCanonicalizer.execute(bound).genCode(ctx)
- val nullCheck = if (bound.nullable) s"${ev.isNull} || " else ""
parts.append(ev.code.toString)
- parts.append(s"\nif (${nullCheck}!${ev.value}) continue;\n")
+ parts.append(s"\n${conjunctCheck(ev, bound.nullable, nullFlag)}\n")
}
Seq.empty
}
@@ -451,10 +641,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,
+ precedingNotNulls, notNullAttributes))
}
// Reset the isNull to false for the not-null columns, then the followed operators could
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/WholeStageCodegenSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/WholeStageCodegenSuite.scala
index 1a17dd700c5f..15396c3c661c 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/WholeStageCodegenSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/WholeStageCodegenSuite.scala
@@ -26,7 +26,7 @@ import scala.util.{Failure, Success, Try}
import org.apache.logging.log4j.Level
import org.scalatest.time.SpanSugar._
-import org.apache.spark.SparkException
+import org.apache.spark.{SparkArithmeticException, SparkException, SparkNumberFormatException}
import org.apache.spark.metrics.source.CodegenMetrics
import org.apache.spark.rdd.MapPartitionsWithEvaluatorRDD
import org.apache.spark.sql.{DataFrame, Dataset, Row, SaveMode}
@@ -35,6 +35,7 @@ import org.apache.spark.sql.catalyst.expressions.codegen.{ByteCodeStats, CodeAnd
import org.apache.spark.sql.catalyst.plans.logical.MergeRows
import org.apache.spark.sql.catalyst.plans.logical.MergeRows.{Keep, Update}
import org.apache.spark.sql.classic
+import org.apache.spark.sql.errors.DataTypeErrors.{toSQLConf, toSQLId}
import org.apache.spark.sql.execution.adaptive.DisableAdaptiveExecutionSuite
import org.apache.spark.sql.execution.aggregate.{HashAggregateExec, SortAggregateExec}
import org.apache.spark.sql.execution.columnar.InMemoryTableScanExec
@@ -2847,6 +2848,314 @@ 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`, `(a > 0 OR b < 0)`
+ // and `coalesce(b, cast(s AS INT)) < 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))
+ }
+
+ private def checkCastInvalidInput(query: String): Unit = {
+ val start = query.lastIndexOf("cast(s AS INT)")
+ checkError(
+ exception = intercept[SparkNumberFormatException] {
+ sql(query).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 = start, stop = start + 13))
+ }
+
+ // 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)
+ }
+ def hasNullConjunctFlag(query: String): Boolean =
+ genCode(sql(query)).exists(_.body.contains("nullConjunct"))
+ // `b > 0` cannot be NULL after the inferred `isnotnull(b)`, so no flag is generated.
+ val noNullConjunct = "SELECT a FROM t WHERE b > 0 AND y > -1"
+ checkAnswer(sql(noNullConjunct), Seq(Row(null), Row(0)))
+ assert(!hasNullConjunctFlag(noNullConjunct))
+ // `a > 0 OR b > 0` can be NULL, but neither the conjuncts after it nor the inferred checks
+ // can raise an error, so the evaluation after a NULL conjunct is not observable and no
+ // flag is generated either: the generated code is the one of the legacy behavior.
+ val noRaise = "SELECT a FROM t WHERE (a > 0 OR b > 0) AND x = 1 AND y = 0"
+ checkAnswer(sql(noRaise), Row(0))
+ assert(!hasNullConjunctFlag(noRaise))
+ // `y + 1 > 0` can raise, so the flag is generated.
+ val raise = "SELECT a FROM t WHERE (a > 0 OR b > 0) AND x = 1 AND y + 1 > 0"
+ checkAnswer(sql(raise), Row(0))
+ assert(hasNullConjunctFlag(raise))
+ }
+ }
+
+ 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 same through the CSE branch of FilterExec, which generates the pending checks
+ // before `y + 1 > 0`, the first conjunct after the NULL one that can raise, and not
+ // before the conjuncts after it.
+ val checkAfterCSE = checkAfter + " AND b * 2 > 0 AND b * 2 > -1"
+ if (skipsAtNull) {
+ checkAnswer(sql(checkAfterCSE), Row(0))
+ } else {
+ checkArithmeticOverflow(checkAfterCSE)
+ }
+ }
+ }
+ // 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 {
+ checkCastInvalidInput(raising)
+ }
+ }
+ }
+ // The placement of the checks is decided at code generation, so once a conjunct can be
+ // NULL, the checks after it are evaluated in the order of the condition on every row. Here
+ // the check written first is generated before `y < 0` and raises on the first row, as with
+ // codegen off, although `a > 0 OR b > 0` is TRUE on the row. The legacy behavior evaluates
+ // `y < 0` first, which is FALSE on both rows, and never reaches the check.
+ val checkFirst =
+ "SELECT a FROM t WHERE cast(s AS INT) IS NOT NULL AND (a > 0 OR b > 0) AND y < 0"
+ withNullConjunctConfs(SQLConf.ANSI_ENABLED.key -> "true") { skipsAtNull =>
+ if (skipsAtNull) {
+ checkAnswer(sql(checkFirst), Seq.empty)
+ } else {
+ checkCastInvalidInput(checkFirst)
+ }
+ }
+ }
+ }
+
+ 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"
+ withNullConjunctConfs(SQLConf.ANSI_ENABLED.key -> "true") { skipsAtNull =>
+ if (skipsAtNull) {
+ checkAnswer(sql(query), Row(1))
+ } else {
+ checkArithmeticOverflow(query)
+ }
+ }
+ }
+ }
+
test("SPARK-60403: full outer joins compile with their functions spilled to nested classes") {
// A full outer join outputs its rows through a function of its own, and generates the code of
// the operators above it, for the body of that function, before adding the function: here a
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org