This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun 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 32425ba1c418 [SPARK-59954][SQL] Fix DECLARE VARIABLE failing on a
NULLIF or BETWEEN default
32425ba1c418 is described below
commit 32425ba1c41847e90cb5f680a20162f27129c731
Author: YangJie <[email protected]>
AuthorDate: Wed Oct 7 10:47:24 2026 -0700
[SPARK-59954][SQL] Fix DECLARE VARIABLE failing on a NULLIF or BETWEEN
default
### What changes were proposed in this pull request?
`RewriteWithExpression` pre-evaluates a common expression's definition in a
`Project` over the first child whose output covers the definition's references.
A default value references no column, so for a `CreateVariable` the rule put
that `Project` over the first `ResolvedIdentifier`, which names a variable and
is not an input. `V2CommandStrategy` plans a `CreateVariable` only when every
child is a `ResolvedIdentifier`, so planning failed.
This adds a `CreateVariable` case to the rule. It inlines the definitions
that gain nothing from memoizing, as it does for a `With` in a conditional
branch, and leaves the rest in the default. `CreateVariableExec` evaluates the
default on the driver, and a `With` can be evaluated since SPARK-58818, so a
kept definition is evaluated once per `With`. A later rule that copies its
input, such as `LikeSimplification` for a single pattern, copies the `With`
too, and each copy evaluates the [...]
### Why are the changes needed?
```sql
DECLARE OR REPLACE VARIABLE v1 DOUBLE DEFAULT nullif(rand(), 0.5);
```
fails with
```
[INTERNAL_ERROR] The Spark SQL phase planning failed with an internal
error. ... SQLSTATE: XX000
Caused by: java.lang.AssertionError: assertion failed: No plan for
CreateVariable default(if ((_common_expr_0#0 = 0.5)) null else
_common_expr_0#0) ...
+- Project [rand(...) AS _common_expr_0#0]
+- ResolvedIdentifier ..., session.v1
```
`BETWEEN` fails the same way. Both wrap their first operand in a `With`
that references it twice, so a NULLIF or BETWEEN default whose first operand is
nondeterministic, like `rand()`, or not cheap to evaluate twice, like a Scala
UDF, hits this. A cheap first operand, such as in `nullif(1, 3)`, is inlined
and works.
### Does this PR introduce _any_ user-facing change?
Yes. A DECLARE VARIABLE like the one above now creates the variable instead
of failing with INTERNAL_ERROR.
### How was this patch tested?
- `RewriteWithExpressionSuite`: a `CreateVariable` whose default holds a
`With` worth memoizing comes out of the rule unchanged, and one whose
definition gains nothing is inlined.
- `ColumnExpressionSuite`: a NULLIF default and a BETWEEN default over a
counting UDF each create the variable, call the UDF once, and read back the
expected value. `nullif(rand(5), 0.5)`, a seeded form of the repro above,
creates its variable.
Both fail without the change to `RewriteWithExpression`.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 5
Closes #59209 from LuciferYang/SPARK-59954.
Lead-authored-by: YangJie <[email protected]>
Co-authored-by: reviewer <reviewer@local>
Signed-off-by: Dongjoon Hyun <[email protected]>
(cherry picked from commit 1dd81715a37d545426abb67b24c024cf27a138aa)
Signed-off-by: Dongjoon Hyun <[email protected]>
---
.../catalyst/optimizer/RewriteWithExpression.scala | 22 +++++++++++-----
.../optimizer/RewriteWithExpressionSuite.scala | 22 +++++++++++++---
.../apache/spark/sql/ColumnExpressionSuite.scala | 30 ++++++++++++++++++++++
3 files changed, 65 insertions(+), 9 deletions(-)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteWithExpression.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteWithExpression.scala
index 29187b3e27d9..722753c9958e 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteWithExpression.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteWithExpression.scala
@@ -22,7 +22,7 @@ import scala.collection.mutable
import org.apache.spark.SparkException
import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.catalyst.planning.PhysicalAggregation
-import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, LogicalPlan,
PlanHelper, Project}
+import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, CreateVariable,
LogicalPlan, PlanHelper, Project}
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.catalyst.trees.TreePattern.{COMMON_EXPR_REF,
CURRENT_LIKE, WITH_EXPRESSION}
import org.apache.spark.sql.internal.SQLConf
@@ -67,6 +67,13 @@ object RewriteWithExpression extends Rule[LogicalPlan] {
// changed anything: `mapExpressions` and `withNewChildren` preserve
reference equality when
// they rewrite nothing, which is what makes this detectable.
if ((rewrittenAgg eq agg) && (rewrittenProj eq proj)) p else
rewrittenProj
+ // The children of a `CreateVariable` are the `ResolvedIdentifier`s
naming the variables, not
+ // rows to evaluate the default over, and `V2CommandStrategy` plans the
command only while
+ // they stay so. A `With` in the default is left in place to memoize the
definitions worth
+ // it, as one in a conditional branch is. A later rule that copies its
input, such as
+ // `LikeSimplification`, copies the `With` too, and each copy evaluates
the definition at most
+ // once each time it is evaluated.
+ case c: CreateVariable => c.mapExpressions(inlineWithsThatGainNothing)
case p if p.expressions.exists(_.containsPattern(WITH_EXPRESSION)) =>
applyInternal(p)
}
@@ -253,6 +260,12 @@ object RewriteWithExpression extends Rule[LogicalPlan] {
counts.filter(_._2 > 1).keys.toSet
}
+ /** `e` with `inlineDefsThatGainNothing` applied to every `With`, bottom-up
for nested ones. */
+ private def inlineWithsThatGainNothing(e: Expression): Expression =
+ e.transformUpWithPruning(_.containsPattern(WITH_EXPRESSION)) {
+ case w: With => inlineDefsThatGainNothing(w)
+ }
+
/**
* `w` with every definition that gains nothing from being memoized inlined
into its references:
* one cheap enough to evaluate twice, and one referenced at most once
anyway. This is the test
@@ -373,11 +386,8 @@ object RewriteWithExpression extends Rule[LogicalPlan] {
val newExpr = c.withNewAlwaysEvaluatedInputs(newAlwaysEvaluatedInputs)
// A `With` in a conditional branch cannot go into a project, which is
always evaluated
// while the branch may not be. It stays where it is and memoizes its
definition per entry
- // instead, but only the definitions that gain something from it. Use
transformUp to handle
- // nested With.
- newExpr.transformUpWithPruning(_.containsPattern(WITH_EXPRESSION)) {
- case w: With => inlineDefsThatGainNothing(w)
- }
+ // instead, but only the definitions that gain something from it.
+ inlineWithsThatGainNothing(newExpr)
case other => other.mapChildren(
rewriteWithExprAndInputPlans(
diff --git
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/RewriteWithExpressionSuite.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/RewriteWithExpressionSuite.scala
index 11bd8623033c..81495236b586 100644
---
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/RewriteWithExpressionSuite.scala
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/RewriteWithExpressionSuite.scala
@@ -22,17 +22,17 @@ import java.time.Instant
import org.apache.spark.SparkException
import org.apache.spark.api.python.PythonEvalType
import org.apache.spark.sql.catalyst.InternalRow
-import org.apache.spark.sql.catalyst.analysis.{EmptyFunctionRegistry,
EmptyTableFunctionRegistry, FakeV2SessionCatalog, TempResolvedColumn}
+import org.apache.spark.sql.catalyst.analysis.{EmptyFunctionRegistry,
EmptyTableFunctionRegistry, FakeSystemCatalog, FakeV2SessionCatalog,
ResolvedIdentifier, TempResolvedColumn}
import org.apache.spark.sql.catalyst.catalog.{InMemoryCatalog, SessionCatalog}
import org.apache.spark.sql.catalyst.dsl.expressions._
import org.apache.spark.sql.catalyst.dsl.plans._
import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.catalyst.plans.PlanTest
-import org.apache.spark.sql.catalyst.plans.logical.{LocalRelation,
LogicalPlan, Project}
+import org.apache.spark.sql.catalyst.plans.logical.{CreateVariable,
DefaultValueExpression, LocalRelation, LogicalPlan, Project}
import org.apache.spark.sql.catalyst.rules.RuleExecutor
import org.apache.spark.sql.catalyst.util.DateTimeUtils
+import org.apache.spark.sql.connector.catalog.{DefaultCatalogManager,
Identifier}
import org.apache.spark.sql.connector.catalog.CatalogV2Implicits._
-import org.apache.spark.sql.connector.catalog.DefaultCatalogManager
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types.{DateType, IntegerType, StringType,
TimestampNTZType, TimestampType, TimeType}
@@ -486,6 +486,22 @@ class RewriteWithExpressionSuite extends PlanTest {
)
}
+ test("SPARK-59954: a With in a DECLARE VARIABLE default leaves the variable
names alone") {
+ val ident = ResolvedIdentifier(FakeSystemCatalog,
Identifier.of(Array("session"), "v"))
+ def declare(default: Expression): CreateVariable =
+ CreateVariable(Seq(ident), DefaultValueExpression(default, "v"), replace
= true)
+
+ // The children are the variable names, not rows, so there is no `Project`
to pre-evaluate a
+ // definition in. One worth memoizing stays in the default for
`CreateVariableExec` to evaluate.
+ val kept = declare(With(Rand(Literal(0L))) { case Seq(ref) => ref * ref })
+ comparePlans(Optimizer.execute(kept), kept)
+
+ // One that gains nothing from memoizing is inlined, as anywhere else.
+ comparePlans(
+ Optimizer.execute(declare(With(Literal(3)) { case Seq(ref) => ref * ref
})),
+ declare(Literal(3) * Literal(3)))
+ }
+
test("SPARK-58902: a With left in a conditional branch of an aggregate still
converges") {
val Seq(a, b) = testRelation.output
// Not cheap and referenced twice, so it stays a memoizing `With` rather
than being inlined.
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/ColumnExpressionSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/ColumnExpressionSuite.scala
index c79fd0878ce2..26ada7270069 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/ColumnExpressionSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/ColumnExpressionSuite.scala
@@ -645,6 +645,36 @@ class ColumnExpressionSuite extends SharedSparkSession {
}
}
+ test("SPARK-59954: a With in a DECLARE VARIABLE default is evaluated once") {
+ // `CreateVariableExec` evaluates the default on the driver, so the
accumulator needs no job to
+ // read. `plus_one_counted` is deterministic but not cheap, so its
definition is worth
+ // memoizing, and the NULLIF and BETWEEN defaults below each read it
twice. The conf block pins
+ // the `With` form of `NullIf` and `Between`, so a suite-level override
cannot switch this test
+ // to the inlined form.
+ val calls = sparkContext.longAccumulator("calls")
+ withUserDefinedFunction("plus_one_counted" -> true) {
+ spark.udf.register("plus_one_counted", (i: Int) => { calls.add(1); i + 1
})
+ withSQLConf(SQLConf.ALWAYS_INLINE_COMMON_EXPR.key -> "false") {
+ withSessionVariable("v1", "v2", "v3") {
+ sql("DECLARE OR REPLACE VARIABLE v1 INT DEFAULT
nullif(plus_one_counted(1), 3)")
+ assert(calls.value == 1, s"the NULLIF default called the UDF
${calls.value} times")
+ checkAnswer(sql("SELECT v1"), Row(2))
+
+ calls.reset()
+ sql("DECLARE OR REPLACE VARIABLE v2 BOOLEAN DEFAULT
plus_one_counted(1) BETWEEN 1 AND 3")
+ assert(calls.value == 1, s"the BETWEEN default called the UDF
${calls.value} times")
+ checkAnswer(sql("SELECT v2"), Row(true))
+
+ // A seeded form of the SPARK-59954 repro. This checks that the
command is planned and the
+ // `Rand` inside the definition is initialized, not the value: every
copy of `rand(5)`
+ // would draw the same first number.
+ sql("DECLARE OR REPLACE VARIABLE v3 DOUBLE DEFAULT nullif(rand(5),
0.5)")
+ checkAnswer(sql("SELECT v3 IS NULL OR (v3 >= 0 AND v3 < 1)"),
Row(true))
+ }
+ }
+ }
+ }
+
test("in") {
val df = Seq((1, "x"), (2, "y"), (3, "z")).toDF("a", "b")
checkAnswer(df.filter($"a".isin(1, 2)),
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
