You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by we...@apache.org on 2026/08/20 06:16:40 UTC
(spark) branch branch-4.x updated: [SPARK-51356][SQL] Emit IsNotNull on a complex child before the FilterExec predicate that uses it
This is an automated email from the ASF dual-hosted git repository.
cloud-fan 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 f4ad8a5d0d7d [SPARK-51356][SQL] Emit IsNotNull on a complex child before the FilterExec predicate that uses it
f4ad8a5d0d7d is described below
commit f4ad8a5d0d7da63b4039d3445df2eeb5527a7ede
Author: Eric Yang <ji...@gmail.com>
AuthorDate: Thu Aug 20 14:16:07 2026 +0800
[SPARK-51356][SQL] Emit IsNotNull on a complex child before the FilterExec predicate that uses it
### What changes were proposed in this pull request?
Currently `FilterExec` matches an `otherPred` to its guarding `IsNotNull` by attribute reference, so an `IsNotNull` over a complex child (`IsNotNull(b.c)`) never matches and is deferred until after the predicates that use it. This fix emits it just before an otherPred containing that child, in both `generatePredicateCode` and the CSE branch of `doConsume` (which must stay in lockstep).
### Why are the changes needed?
The guard runs after the predicate it guards:
```scala
case class C(d: Int); case class B(c: C = null); case class A(b: B = null)
val ds = spark.createDataset(Seq(A(null), A(B(null)), A(B(C(0))), A(B(C(1))))).map(identity)
.where($"b".isNotNull).select($"b".as[B])
val isDZero = udf((c: C) => c.d == 0)
ds.filter($"c".isNotNull).filter(not(isDZero($"c"))).show()
```
ScalaUDF passes null to its deserializer instead of short-circuiting, so this throws EXPRESSION_DECODING_FAILED under whole-stage codegen while interpreted mode returns the correct row. Only UDFs taking the nested struct are affected.
### Does this PR introduce any user-facing change?
No API change. Other changes:
- (issue fix) The failed query above now succeeds.
- A guard over a conditionally-evaluated throwing expression, e.g. `cast(s AS int) IS NOT NULL AND (CASE WHEN a = 1 THEN cast(s AS int) ELSE 0 END) > 0` under ANSI is now evaluated up front and can raise under WSCG, matching the interpreted path.
### How was this patch tested?
Unit test cases added.
### Was this patch authored or co-authored using generative AI tooling?
Yes
Closes #58058 from jiwen624/SPARK-51356.
Authored-by: Eric Yang <ji...@gmail.com>
Signed-off-by: Wenchen Fan <we...@databricks.com>
(cherry picked from commit eaf9d3a5da52d9d3180c84a744219baa3074c2dc)
Signed-off-by: Wenchen Fan <we...@databricks.com>
---
.../sql/execution/basicPhysicalOperators.scala | 23 ++++
.../sql/execution/WholeStageCodegenSuite.scala | 118 ++++++++++++++++++++-
2 files changed, 140 insertions(+), 1 deletion(-)
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 a45ea80c2ee5..9f9eae3d86f2 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
@@ -208,10 +208,19 @@ trait GeneratePredicateHelper extends PredicateHelper {
}
}.mkString("\n").trim
+ val nestedNullChecks = 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)
+ }.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"""
|$nullChecks
+ |$nestedNullChecks
|${genPredicate(c, inputExprCode, outputAttrs)}
""".stripMargin.trim
}.mkString("\n")
@@ -401,6 +410,20 @@ case class FilterExec(condition: Expression, child: SparkPlan)
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 _ =>
+ }
statesByFirstUse.get(idx).foreach { states =>
parts.append(ctx.evaluateSubExprEliminationState(states))
parts.append('\n')
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 85e961d71ff9..6af7d066fccd 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
@@ -23,7 +23,7 @@ import java.time.Duration
import org.apache.spark.SparkException
import org.apache.spark.rdd.MapPartitionsWithEvaluatorRDD
import org.apache.spark.sql.{Dataset, Row, SaveMode}
-import org.apache.spark.sql.catalyst.expressions.{And, Cast, CodegenObjectFactoryMode, Expression, IsNotNull}
+import org.apache.spark.sql.catalyst.expressions.{And, Attribute, Cast, CodegenObjectFactoryMode, Expression, IsNotNull}
import org.apache.spark.sql.catalyst.expressions.codegen.{ByteCodeStats, CodeAndComment, CodeGenerator}
import org.apache.spark.sql.execution.adaptive.DisableAdaptiveExecutionSuite
import org.apache.spark.sql.execution.aggregate.{HashAggregateExec, SortAggregateExec}
@@ -35,6 +35,12 @@ import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.test.SharedSparkSession
import org.apache.spark.sql.types.{DayTimeIntervalType, DecimalType, DoubleType, FloatType, IntegerType, LongType, StringType, StructField, StructType}
+// Nested-struct fixtures for the SPARK-51356 test below. They are top-level so the Dataset
+// encoders resolve without an outer scope.
+case class Spark51356Inner(d: Int)
+case class Spark51356Mid(c: Spark51356Inner = null)
+case class Spark51356Outer(b: Spark51356Mid = null)
+
// Disable AQE because the WholeStageCodegenExec is added when running QueryStageExec
class WholeStageCodegenSuite extends SharedSparkSession
with DisableAdaptiveExecutionSuite {
@@ -1328,6 +1334,116 @@ class WholeStageCodegenSuite extends SharedSparkSession
}
}
+ test("SPARK-51356: FilterExec emits IsNotNull on a nested field before its otherPred") {
+ // `IsNotNull(b.c)` is null-intolerant and references only `b`, so it is classified as a
+ // notNullPred -- but its child is a complex expression, so it never matches an otherPred's
+ // bare attribute reference. It used to be deferred to the trailing leftover block, i.e.
+ // emitted *after* the UDF that dereferences `b.c`, and `ScalaUDF` hands a null argument to
+ // its deserializer rather than short-circuiting, so `newInstance(Spark51356Inner)` threw.
+ // The interpreted path evaluates the conjunction in order and was unaffected.
+ val data = Seq(
+ Spark51356Outer(null),
+ Spark51356Outer(Spark51356Mid(null)), // the row that used to trigger the failure
+ Spark51356Outer(Spark51356Mid(Spark51356Inner(0))),
+ Spark51356Outer(Spark51356Mid(Spark51356Inner(1))))
+ val isDZero = udf((c: Spark51356Inner) => c.d == 0)
+ def newDf(): Dataset[Spark51356Mid] = {
+ // `map(identity)` keeps the input from being folded into a LocalRelation, so the filter
+ // really goes through whole-stage codegen.
+ val mids = spark.createDataset(data).map(identity)
+ .where(col("b").isNotNull).select(col("b").as[Spark51356Mid])
+ mids.filter(col("c").isNotNull).filter(not(isDZero(col("c"))))
+ }
+
+ withSQLConf(SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> "true") {
+ val df = newDf()
+ val plan = df.queryExecution.executedPlan
+ assert(plan.exists(_.isInstanceOf[WholeStageCodegenExec]),
+ "Filter should be in whole-stage codegen")
+ // Guard against optimizer drift: the regression only exists when a single FilterExec
+ // carries an `IsNotNull` over a complex child together with a non-IsNotNull conjunct
+ // that consumes the same expression.
+ def conjuncts(e: Expression): Seq[Expression] = e match {
+ case And(l, r) => conjuncts(l) ++ conjuncts(r)
+ case other => Seq(other)
+ }
+ val matchingFilter = plan.collect {
+ case f: FilterExec =>
+ val cs = conjuncts(f.condition)
+ val nestedIsNotNulls = cs.collect {
+ case IsNotNull(child) if !child.isInstanceOf[Attribute] => child
+ }
+ nestedIsNotNulls.exists { child =>
+ cs.exists {
+ case _: IsNotNull => false
+ case other => other.exists(_.semanticEquals(child))
+ }
+ }
+ }.exists(identity)
+ assert(matchingFilter,
+ "expected a FilterExec carrying IsNotNull(<complex>) plus a non-IsNotNull conjunct " +
+ "over the same expression")
+ checkAnswer(df.toDF(), Row(Row(1)))
+ }
+ // Cross-check the codegen path against the interpreted path.
+ withSQLConf(SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> "false") {
+ val df = newDf()
+ assert(!df.queryExecution.executedPlan.exists(_.isInstanceOf[WholeStageCodegenExec]),
+ "the cross-check must be planned without whole-stage codegen")
+ checkAnswer(df.toDF(), Row(Row(1)))
+ }
+ }
+
+ test("SPARK-51356: FilterExec CSE emits IsNotNull on a nested field before its otherPred") {
+ // Same defect as above, on the CSE branch of `FilterExec.doConsume`, which inlines its own
+ // copy of the interleaving. Two otherPreds share the non-cheap `f(c)`, so the branch is
+ // taken (a bare `b.c` is cheap and would fall back to `generatePredicateCode`). The
+ // guarding `IsNotNull(b.c)` has to be emitted ahead of the shared CSE precompute, not after
+ // the predicates that consume it.
+ val data = Seq(
+ Spark51356Outer(null),
+ Spark51356Outer(Spark51356Mid(null)), // the row that used to trigger the failure
+ Spark51356Outer(Spark51356Mid(Spark51356Inner(0))),
+ Spark51356Outer(Spark51356Mid(Spark51356Inner(1))))
+ val dOf = udf((c: Spark51356Inner) => c.d)
+ val mids = spark.createDataset(data).map(identity)
+ .where(col("b").isNotNull).select(col("b").as[Spark51356Mid])
+ val df = mids
+ .filter(col("c").isNotNull)
+ .filter(dOf(col("c")) > 0)
+ .filter(dOf(col("c")) < 100)
+
+ withSQLConf(
+ SQLConf.SUBEXPRESSION_ELIMINATION_ENABLED.key -> "true",
+ SQLConf.SUBEXPRESSION_ELIMINATION_FILTER_EXEC_ENABLED.key -> "true",
+ SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> "true") {
+ val plan = df.queryExecution.executedPlan
+ assert(plan.exists(_.isInstanceOf[WholeStageCodegenExec]),
+ "Filter should be in whole-stage codegen")
+ checkAnswer(df.toDF(), Row(Row(1)))
+ }
+ }
+
+ test("SPARK-51356: FilterExec CSE guards an IsNotNull whose child is itself a subexpression") {
+ // A dynamic-gap `session_window` guards the session struct with an IsNotNull over the same
+ // cast the struct is built from, so the guarded expression is itself the common
+ // subexpression. The check must not reference the CSE state, which is emitted later.
+ val df = spark.sql(
+ """
+ |SELECT a, count(*) AS cnt
+ |FROM VALUES ('A1', '2021-01-01 00:00:00'), ('A1', '2021-01-01 00:04:30'),
+ | ('A2', '2021-01-01 00:01:00') AS tab(a, b)
+ |GROUP BY a, session_window(b, CASE WHEN a = 'A1' THEN '5 minutes' ELSE '1 minute' END)
+ """.stripMargin)
+
+ withSQLConf(
+ SQLConf.SUBEXPRESSION_ELIMINATION_ENABLED.key -> "true",
+ SQLConf.SUBEXPRESSION_ELIMINATION_FILTER_EXEC_ENABLED.key -> "true",
+ SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> "true") {
+ checkAnswer(df, Seq(Row("A1", 2), Row("A2", 1)))
+ }
+ }
+
test("SPARK-56032: FilterExec CSE handles shared otherPred refs with guard") {
// Filter shape: kind = 'numeric' AND cast(s as int) > 0 AND cast(s as int) < 100.
// Exercises invariant (b) on a shape where two cast otherPreds share a ref: the
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org