You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ur...@apache.org on 2026/09/21 18:11:23 UTC
(spark) branch branch-4.3 updated: [SPARK-59496][SQL] Avoid exponential NullIf plan growth
This is an automated email from the ASF dual-hosted git repository.
uros-b pushed a commit to branch branch-4.3
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.3 by this push:
new fba9e4a2554c [SPARK-59496][SQL] Avoid exponential NullIf plan growth
fba9e4a2554c is described below
commit fba9e4a2554c41a978d1bdfcac61be15908a2901
Author: Stefan Kandic <st...@databricks.com>
AuthorDate: Mon Sep 21 18:10:52 2026 +0000
[SPARK-59496][SQL] Avoid exponential NullIf plan growth
### What changes were proposed in this pull request?
Replace `TypedNullLiteral` with `NullIfResult`, an analysis-time wrapper that exposes only the
predicate and returned value as children. It expands to `If` after analysis, when the null branch
can take the returned value's final type.
### Why are the changes needed?
`TypedNullLiteral(left)` adds a third recursive occurrence of `left`, causing `3^n` plan growth
when common expressions are explicitly inlined. `NullIfResult` avoids that extra copy while also
preventing stale types or collations when analysis changes the operand. Default `With` mode remains
linear; explicit inlining retains its expected two copies.
### Does this PR introduce _any_ user-facing change?
No. Query results and supported SQL remain unchanged.
### How was this patch tested?
Added and ran tests for nested plan growth, unresolved inputs, ANSI and non-ANSI coercion, both
common-expression modes, final result typing, and optimizer replacement.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Codex (GPT-5)
Closes #58779 from stefankandic/avoid-exponential-nullif-plan-growth.
Authored-by: Stefan Kandic <st...@databricks.com>
Signed-off-by: Uros Bojanic <22...@users.noreply.github.com>
(cherry picked from commit d8a7143d4fd819eb5b6bfab74bfcbec65d235f5e)
Signed-off-by: Uros Bojanic <22...@users.noreply.github.com>
---
.../sql/catalyst/expressions/nullExpressions.scala | 70 ++++++++++++++++------
.../expressions/NullExpressionsSuite.scala | 51 +++++++++++++++-
.../sql/catalyst/optimizer/OptimizerSuite.scala | 27 +++++++--
.../apache/spark/sql/DataFrameFunctionsSuite.scala | 31 ++++++++++
.../scala/org/apache/spark/sql/ExplainSuite.scala | 2 +-
5 files changed, 154 insertions(+), 27 deletions(-)
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/nullExpressions.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/nullExpressions.scala
index db77d6b82b9f..9d95baa91c69 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/nullExpressions.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/nullExpressions.scala
@@ -19,11 +19,14 @@ package org.apache.spark.sql.catalyst.expressions
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult.DataTypeMismatch
import org.apache.spark.sql.catalyst.expressions.Cast._
import org.apache.spark.sql.catalyst.expressions.codegen._
import org.apache.spark.sql.catalyst.expressions.codegen.Block._
+import org.apache.spark.sql.catalyst.trees.BinaryLike
import org.apache.spark.sql.catalyst.trees.TreePattern.{COALESCE, NULL_CHECK, TreePattern}
import org.apache.spark.sql.catalyst.util.TypeUtils
+import org.apache.spark.sql.catalyst.util.TypeUtils.{ordinalNumber, toSQLExpr, toSQLType}
import org.apache.spark.sql.errors.QueryCompilationErrors
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types._
@@ -175,22 +178,6 @@ case class Coalesce(children: Seq[Expression])
copy(children = newChildren)
}
-private case class TypedNullLiteral(child: Expression)
- extends UnaryExpression with RuntimeReplaceable {
- override def nullable: Boolean = true
-
- override def dataType: DataType = child.dataType
-
- override def toString: String = "null"
-
- override def sql: String = "NULL"
-
- override lazy val replacement: Expression = Literal.create(null, child.dataType)
-
- override protected def withNewChildInternal(newChild: Expression): TypedNullLiteral =
- copy(child = newChild)
-}
-
@ExpressionDescription(
usage = "_FUNC_(expr1, expr2) - Returns null if `expr1` equals to `expr2`, or `expr1` otherwise.",
arguments = """
@@ -214,10 +201,10 @@ case class NullIf(left: Expression, right: Expression, replacement: Expression)
this(left, right,
if (!SQLConf.get.getConf(SQLConf.ALWAYS_INLINE_COMMON_EXPR)) {
With(left) { case Seq(ref) =>
- If(EqualTo(ref, right), TypedNullLiteral(ref), ref)
+ NullIfResult(EqualTo(ref, right), ref)
}
} else {
- If(EqualTo(left, right), TypedNullLiteral(left), left)
+ NullIfResult(EqualTo(left, right), left)
}
)
}
@@ -229,6 +216,53 @@ case class NullIf(left: Expression, right: Expression, replacement: Expression)
}
}
+/**
+ * Carries [[NullIf]]'s predicate and result value through analysis before expanding them into an
+ * `If`. The null branch takes its type from the analyzed result value.
+ */
+private case class NullIfResult(predicate: Expression, value: Expression)
+ extends RuntimeReplaceable with BinaryLike[Expression] with ConditionalExpression {
+ override def left: Expression = predicate
+
+ override def right: Expression = value
+
+ override def nullable: Boolean = true
+
+ override def dataType: DataType = value.dataType
+
+ override lazy val replacement: Expression =
+ If(predicate, Literal.create(null, value.dataType), value)
+
+ override def alwaysEvaluatedInputs: Seq[Expression] = predicate :: Nil
+
+ override def withNewAlwaysEvaluatedInputs(
+ alwaysEvaluatedInputs: Seq[Expression]): NullIfResult = {
+ copy(predicate = alwaysEvaluatedInputs.head)
+ }
+
+ override def branchGroups: Seq[Seq[Expression]] = Nil
+
+ override def checkInputDataTypes(): TypeCheckResult = {
+ if (predicate.dataType == BooleanType) {
+ TypeCheckResult.TypeCheckSuccess
+ } else {
+ DataTypeMismatch(
+ errorSubClass = "UNEXPECTED_INPUT_TYPE",
+ messageParameters = Map(
+ "paramIndex" -> ordinalNumber(0),
+ "requiredType" -> toSQLType(BooleanType),
+ "inputSql" -> toSQLExpr(predicate),
+ "inputType" -> toSQLType(predicate.dataType)))
+ }
+ }
+
+ override protected def withNewChildrenInternal(
+ newPredicate: Expression,
+ newValue: Expression): NullIfResult = {
+ copy(predicate = newPredicate, value = newValue)
+ }
+}
+
@ExpressionDescription(
usage = "_FUNC_(expr) - Returns null if `expr` is equal to zero, or `expr` otherwise.",
arguments = """
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/NullExpressionsSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/NullExpressionsSuite.scala
index 3d957b662d5d..2ebbd5c2b4f7 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/NullExpressionsSuite.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/NullExpressionsSuite.scala
@@ -148,9 +148,13 @@ class NullExpressionsSuite extends SparkFunSuite with ExpressionEvalHelper {
test("SPARK-56840: NullIf replacement preserves its data type before type coercion") {
Seq(true, false).foreach { alwaysInlineCommonExpr =>
withSQLConf(SQLConf.ALWAYS_INLINE_COMMON_EXPR.key -> alwaysInlineCommonExpr.toString) {
- val nullIf = new NullIf(Literal(1), Literal(1))
+ val nullIf = new NullIf(Literal(1), Literal(2.1d))
assert(nullIf.dataType == IntegerType)
assert(nullIf.replacement.dataType == IntegerType)
+
+ val plan = SimpleAnalyzer.execute(
+ Project(Alias(nullIf, "out")() :: Nil, LocalRelation()))
+ assert(plan.expressions.head.dataType == IntegerType)
}
}
}
@@ -158,13 +162,54 @@ class NullExpressionsSuite extends SparkFunSuite with ExpressionEvalHelper {
test(
"SPARK-56840: NullIf accepts unresolved nested fields during inlined function construction") {
withSQLConf(SQLConf.ALWAYS_INLINE_COMMON_EXPR.key -> "true") {
- val nullIf = FunctionRegistry.builtin.lookupFunction(
+ val unresolvedNullIf = FunctionRegistry.builtin.lookupFunction(
FunctionIdentifier("nullif"),
Seq(
UnresolvedAttribute(Seq("c", "provider")),
Lower(Literal("ERROR_MULTIPLE_PROVIDERS"))))
+ .asInstanceOf[NullIf]
+
+ assert(unresolvedNullIf.exists(_.isInstanceOf[NullIfResult]))
+
+ val resolvedNullIf = unresolvedNullIf.transformUp {
+ case _: UnresolvedAttribute => Literal("lit")
+ }
+ val result = resolvedNullIf.collectFirst { case result: NullIfResult => result }.get
+
+ assert(result.dataType == StringType)
+ assert(result.replacement.exists {
+ case Literal(null, StringType) => true
+ case _ => false
+ })
+ }
+ }
+
+ test("SPARK-59496: NullIf result validates only its always-evaluated predicate") {
+ val value = UnresolvedAttribute("value")
+ val result = NullIfResult(Literal(true), value)
- assert(nullIf.isInstanceOf[NullIf])
+ assert(result.children == Seq(Literal(true), value))
+ assert(result.alwaysEvaluatedInputs == Seq(Literal(true)))
+ assert(result.branchGroups.isEmpty)
+ assert(result.checkInputDataTypes().isSuccess)
+ assert(NullIfResult(Literal(1), value).checkInputDataTypes().isFailure)
+ }
+
+ test("SPARK-59496: NullIf has linear or doubling construction growth") {
+ Seq(true, false).foreach { alwaysInlineCommonExpr =>
+ withSQLConf(SQLConf.ALWAYS_INLINE_COMMON_EXPR.key -> alwaysInlineCommonExpr.toString) {
+ val depth = 8
+ val nestedNullIf = (1 to depth).foldLeft[Expression](Literal(0)) {
+ case (left, right) => new NullIf(left, Literal(right))
+ }
+
+ val expectedCount = if (alwaysInlineCommonExpr) (1 << depth) - 1 else depth
+ assert(nestedNullIf.collect { case _: NullIf => 1 }.size == expectedCount)
+
+ val analyzed = SimpleAnalyzer.execute(
+ Project(Alias(nestedNullIf, "out")() :: Nil, LocalRelation())).expressions.head
+ assert(analyzed.collect { case _: NullIf => 1 }.size == expectedCount)
+ }
}
}
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/OptimizerSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/OptimizerSuite.scala
index 57b9df6512b5..1f77c6b4909b 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/OptimizerSuite.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/OptimizerSuite.scala
@@ -21,7 +21,22 @@ import org.apache.spark.SparkException
import org.apache.spark.sql.AnalysisException
import org.apache.spark.sql.catalyst.dsl.expressions._
import org.apache.spark.sql.catalyst.dsl.plans._
-import org.apache.spark.sql.catalyst.expressions.{Add, Alias, ArrayCompact, AttributeReference, CreateArray, CreateStruct, IntegerLiteral, Literal, MapFromEntries, Multiply, NamedExpression, NullIf, Remainder, RuntimeReplaceable}
+import org.apache.spark.sql.catalyst.expressions.{
+ Add,
+ Alias,
+ ArrayCompact,
+ AttributeReference,
+ CreateArray,
+ CreateStruct,
+ If,
+ IntegerLiteral,
+ Literal,
+ MapFromEntries,
+ Multiply,
+ NamedExpression,
+ NullIf,
+ Remainder,
+ RuntimeReplaceable}
import org.apache.spark.sql.catalyst.expressions.aggregate.Sum
import org.apache.spark.sql.catalyst.plans.PlanTest
import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, LocalRelation, LogicalPlan, OneRowRelation, Project}
@@ -335,7 +350,7 @@ class OptimizerSuite extends PlanTest {
StructType(StructField("map", MapType(IntegerType, IntegerType, false), false) :: Nil))
}
- test("SPARK-56840: NullIf typed null branch is replaced with a null literal") {
+ test("SPARK-56840: NullIf is replaced with an If containing a typed null") {
val optimizer = new SimpleTestOptimizer() {
override def defaultBatches: Seq[Batch] =
Batch("test", fixedPoint,
@@ -346,12 +361,14 @@ class OptimizerSuite extends PlanTest {
val nullIf = new NullIf(Literal(true), Literal(true))
val plan = Project(Alias(nullIf, "out")() :: Nil, OneRowRelation()).analyze
val optimized = optimizer.execute(plan)
+ val replacement = optimized.expressions.head.asInstanceOf[Alias].child
- assert(optimized.expressions.exists(_.exists {
+ assert(replacement.isInstanceOf[If])
+ assert(replacement.exists {
case Literal(null, BooleanType) => true
case _ => false
- }))
- assert(optimized.expressions.forall(!_.exists(_.isInstanceOf[RuntimeReplaceable])))
+ })
+ assert(!replacement.exists(_.isInstanceOf[RuntimeReplaceable]))
}
}
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameFunctionsSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameFunctionsSuite.scala
index 24f2607354df..184b6fe8538a 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameFunctionsSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameFunctionsSuite.scala
@@ -362,6 +362,37 @@ class DataFrameFunctionsSuite extends SharedSparkSession {
}
}
+ test("nullif keeps the coerced type of its first argument") {
+ Seq(true, false).foreach { alwaysInlineCommonExpr =>
+ Seq(true, false).foreach { ansiEnabled =>
+ withSQLConf(
+ SQLConf.ALWAYS_INLINE_COMMON_EXPR.key -> alwaysInlineCommonExpr.toString,
+ SQLConf.ANSI_ENABLED.key -> ansiEnabled.toString,
+ SQLConf.CONCAT_BINARY_AS_STRING.key -> "true") {
+ // The analyzers apply different comparison coercions to binary concat. Exercise result
+ // typing under each analyzer independently. Pure single-pass does not support the RDD
+ // deserialization check performed by the suite's checkAnswer helper.
+ Seq(false, true).foreach { singlePassResolverEnabled =>
+ withSQLConf(
+ SQLConf.ANALYZER_DUAL_RUN_LEGACY_AND_SINGLE_PASS_RESOLVER.key -> "false",
+ SQLConf.ANALYZER_SINGLE_PASS_RESOLVER_ENABLED_TENTATIVELY.key -> "false",
+ SQLConf.ANALYZER_SINGLE_PASS_RESOLVER_ENABLED.key ->
+ singlePassResolverEnabled.toString) {
+ QueryTest.checkAnswer(
+ sql("SELECT nullif(concat(X'61', X'62'), 'z')"),
+ Row("ab") :: Nil,
+ checkToRDD = false)
+ }
+ }
+
+ val result = sql("SELECT nullif(1, 2.1D)")
+ checkAnswer(result, Row(1))
+ assert(result.schema.head.dataType == IntegerType)
+ }
+ }
+ }
+ }
+
test("equal_null function") {
val df = Seq[(Integer, Integer)]((null, 8)).toDF("a", "b")
checkAnswer(df.selectExpr("equal_null(a, b)"), Seq(Row(false)))
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/ExplainSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/ExplainSuite.scala
index 618f0ab675e1..365559ab63e9 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/ExplainSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/ExplainSuite.scala
@@ -251,7 +251,7 @@ class ExplainSuite extends ExplainSuiteHelper with DisableAdaptiveExecutionSuite
checkKeywordsExistsInExplain(df,
"Project [id#xL AS ifnull(id, 1)#xL, if ((id#xL = 1)) null " +
"else id#xL AS nullif(id, 1)#xL, id#xL AS nvl(id, 1)#xL, 1 AS nvl2(id, 1, 2)#x]")
- checkKeywordsNotExistsInExplain(df, ExtendedMode, "typednullliteral")
+ checkKeywordsNotExistsInExplain(df, ExtendedMode, "nullifresult")
}
test("SPARK-26659: explain of DataWritingCommandExec should not contain duplicate cmd.nodeName") {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org