This is an automated email from the ASF dual-hosted git repository.
yangjie01 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 2ad52c22e835 [SPARK-56431][SQL] Fix NPE in FilterExec CSE when notNull
columns are null before short-circuit
2ad52c22e835 is described below
commit 2ad52c22e8357c38b55cbe61f24f7e9c387c538d
Author: yangjie01 <[email protected]>
AuthorDate: Sun Apr 12 21:49:13 2026 +0800
[SPARK-56431][SQL] Fix NPE in FilterExec CSE when notNull columns are null
before short-circuit
### What changes were proposed in this pull request?
In `FilterExec.doConsume`, bind `otherPreds` against `child.output`
(instead of `output`) for both CSE analysis and predicate code generation.
### Why are the changes needed?
SPARK-56032 added CSE support to `FilterExec`. The CSE precomputation code
runs **before** the `IsNotNull` predicates short-circuit, at the start of the
`do {} while(false)` block. However, the `boundOtherPreds` passed to CSE
analysis were bound against `output`, where columns in `notNullAttributes` have
`nullable=false`. This caused the CSE-generated code to omit null checks for
those columns. At runtime, when a row has a null value in such a column, the
precomputation
calls an arithmetic operator (e.g. `Decimal.subtract`) on the null value
before the `IsNotNull` check can filter the row out, resulting in an NPE.
The original SPARK-56032 design comment states:
> CSE evaluation is placed before predicate short-circuit checks -- This is
safe
> because Spark SQL expressions handle null inputs gracefully (returning
null rather
> than throwing).
This assumption is only correct when the expressions preserve their input
columns' original nullability. Binding against `output` (with tightened
nullability) breaks it.
Reproducer: a `FilterExec` over a source with a nullable `DecimalType`
column, where the filter has both `IsNotNull(col)` and a repeated arithmetic
expression on `col` (creating a CSE candidate), and at least one input row has
`col = null`.
### Does this PR introduce _any_ user-facing change?
No. This optimization has not been released yet.
### How was this patch tested?
Added a new test in `WholeStageCodegenSuite`.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code
Closes #55288 from LuciferYang/SPARK-56431.
Authored-by: yangjie01 <[email protected]>
Signed-off-by: yangjie01 <[email protected]>
---
.../sql/execution/basicPhysicalOperators.scala | 7 +++--
.../sql/execution/WholeStageCodegenSuite.scala | 36 +++++++++++++++++++++-
2 files changed, 40 insertions(+), 3 deletions(-)
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 aaafeccdfc6f..9b457bfdd662 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
@@ -256,8 +256,11 @@ case class FilterExec(condition: Expression, child:
SparkPlan)
// Apply CSE to otherPreds only (notNullPreds are simple IsNotNull checks
with no CSE value).
val (inputVarsCode, subExprsCode, predicateCode) =
if (conf.subexpressionEliminationEnabled && otherPreds.nonEmpty) {
+ // Bind against child.output (not output) so that columns in
+ // notNullAttributes keep their original nullable=true for the
+ // CSE precomputation, which runs BEFORE IsNotNull short-circuits.
val boundOtherPreds = otherPreds.map(
- BindReferences.bindReference(_, output))
+ BindReferences.bindReference(_, child.output))
// Pre-evaluate input variables before CSE analysis: CSE clears
// ctx.currentVars[i].code as a side effect; without this
pre-evaluation, Janino fails
// with "Unknown variable or type" when notNullPreds reference the
same input columns.
@@ -270,7 +273,7 @@ case class FilterExec(condition: Expression, child:
SparkPlan)
var code = ""
ctx.withSubExprEliminationExprs(subExprs.states) {
code = generatePredicateCode(
- ctx, child.output, input, output, notNullPreds, otherPreds,
notNullAttributes)
+ ctx, child.output, input, child.output, notNullPreds,
otherPreds, notNullAttributes)
Seq.empty
}
code
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 652f38183bc7..ad77e125c0cb 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
@@ -17,6 +17,7 @@
package org.apache.spark.sql.execution
+import java.math.{BigDecimal => JBigDecimal}
import java.time.Duration
import org.apache.spark.SparkException
@@ -32,7 +33,7 @@ import
org.apache.spark.sql.execution.joins.{BroadcastHashJoinExec, BroadcastNes
import org.apache.spark.sql.functions._
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.test.SharedSparkSession
-import org.apache.spark.sql.types.{DayTimeIntervalType, IntegerType,
StringType, StructField, StructType}
+import org.apache.spark.sql.types.{DayTimeIntervalType, DecimalType,
IntegerType, StringType, StructField, StructType}
// Disable AQE because the WholeStageCodegenExec is added when running
QueryStageExec
class WholeStageCodegenSuite extends QueryTest with SharedSparkSession
@@ -995,4 +996,37 @@ class WholeStageCodegenSuite extends QueryTest with
SharedSparkSession
s"addExact appears $cseAddCount times with CSE vs $noCseAddCount
times without")
}
}
+
+ test("SPARK-56431: CSE in FilterExec preserves null guards for notNull
columns") {
+ // The CSE precomputation in FilterExec runs BEFORE IsNotNull
short-circuits.
+ // If the CSE-bound expressions use the tightened output nullability
+ // (notNullAttributes marked non-null), the generated code omits null
+ // checks and crashes on types whose arithmetic NPEs on null (Decimal).
+ val schema = StructType(Seq(
+ StructField("a", DecimalType(10, 2), nullable = true),
+ StructField("b", DecimalType(10, 2), nullable = true)))
+ val data = spark.sparkContext.parallelize(Seq(
+ Row(null, new JBigDecimal("100.00")),
+ Row(new JBigDecimal("50.00"), new JBigDecimal("200.00")),
+ Row(new JBigDecimal("80.00"), new JBigDecimal("100.00"))))
+
+ withSQLConf(
+ SQLConf.SUBEXPRESSION_ELIMINATION_ENABLED.key -> "true",
+ SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> "true",
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+ val df = spark.createDataFrame(data, schema)
+ // abs(b - a) appears twice, creating a CSE candidate.
+ // a IS NOT NULL puts a in FilterExec.notNullAttributes.
+ // Without the fix the first row (a=null) causes NPE in the
+ // CSE precomputation because Decimal.subtract skips null check.
+ val filtered = df.where(
+ "a IS NOT NULL AND abs(b - a) > 0.10 AND abs(b - a) < 1000.00")
+ val plan = filtered.queryExecution.executedPlan
+ assert(plan.exists(_.isInstanceOf[WholeStageCodegenExec]),
+ "Filter should be in whole-stage codegen")
+ checkAnswer(filtered, Seq(
+ Row(new JBigDecimal("50.00"), new JBigDecimal("200.00")),
+ Row(new JBigDecimal("80.00"), new JBigDecimal("100.00"))))
+ }
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
