This is an automated email from the ASF dual-hosted git repository.
LuciferYang 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 284bfe14a29c [SPARK-59962][SQL] Inline common expressions in a command
instead of pre-evaluating them under it
284bfe14a29c is described below
commit 284bfe14a29cf9f0e9b3849c4c13709726fc6ecb
Author: YangJie <[email protected]>
AuthorDate: Thu Oct 8 17:11:39 2026 +0800
[SPARK-59962][SQL] Inline common expressions in a command instead of
pre-evaluating them under it
### What changes were proposed in this pull request?
`RewriteWithExpression` pre-evaluates a common expression's definition that
is referenced more than once and is not cheap in a `Project` over the first
child whose output covers the definition's references. That assumes the
operator evaluates its expressions over its children's rows, and a command does
not, nor does a `Call`, whose arguments `InvokeProcedures` expects to find
foldable over its `ResolvedProcedure`. A CHECK constraint added by ALTER TABLE
is stored as a V2 predicate bui [...]
This adds a case for a `Command` or a `Call` to the rule that substitutes
every deterministic definition, and any definition read once, at its
references. A row-level write (`ReplaceData`, `WriteDelta`) gets it for its
`condition` only: its `groupFilterCondition` is read only by the runtime group
filter, which puts it in a filter over the table and optimizes it there, so a
`With` in it is treated as one in a conditional branch, inlined only where
memoizing gains nothing. Analysis reje [...]
`inlineDefsThatGainNothing` is split so the new case reuses its
substitution, and `canSubstitute` takes a flag that skips its cost check for it.
### Why are the changes needed?
With default settings, NULLIF or BETWEEN over a first operand that is not
cheap breaks these commands and CALL:
```sql
-- InMemoryTableCatalog as testcat
CREATE TABLE testcat.g (a INT, b INT GENERATED ALWAYS AS (nullif(a + 1,
3)));
-- [INTERNAL_ERROR] ... No plan for CreateTable ...
-- +- Project [a, b, (a + 1) AS _common_expr_0]
-- +- ResolvedIdentifier ..., g, [a, b]
-- CREATE OR REPLACE TABLE fails the same way
DELETE FROM testcat.t WHERE (a + b) BETWEEN 0 AND 4;
-- [INTERNAL_ERROR] Unexpected table relation: Project [a, b, (a + b) AS
_common_expr_1] ...
-- group-based row-level table partitioned by dep
DELETE FROM rowcat.p WHERE (id + pk) BETWEEN 2 AND 4;
-- [INTERNAL_ERROR_ATTRIBUTE_NOT_FOUND] Could not find _common_expr_4 in
[dep]
CALL cat.ns.sum(nullif((SELECT 5), 3), 1);
-- [INTERNAL_ERROR] Unexpected plan for optimized CALL statement
ALTER TABLE testcat.t ADD CONSTRAINT c1 CHECK (nullif(a + b, 3) > 0);
-- succeeds, but the V2 predicate stored in the catalog is
-- CASE WHEN _common_expr_1 = 3 THEN NULL ELSE _common_expr_1 > 0 END
-- and every later write fails with UNRESOLVED_COLUMN on _common_expr_1
```
### Does this PR introduce _any_ user-facing change?
Yes. The CREATE TABLE and CREATE OR REPLACE TABLE above succeed and INSERT
computes the generated column. The DELETE on `testcat.t` reports that the table
cannot delete by `[(a + b) >= 0, (a + b) <= 4]` instead of an internal error,
so a source that accepts those predicates can run it. The group-based DELETE
deletes the matching rows. The CALL returns 6. Writes to the table after the
ALTER TABLE enforce the check. A definition referenced twice in the expression
of any command but DECL [...]
### How was this patch tested?
- `RewriteWithExpressionSuite`: a DELETE condition holding a `With`,
directly and in a conditional branch, comes out inlined with no `Project` under
the command; a nondeterministic definition keeps its `With` when read twice and
is inlined when read once; and a `DECLARE VARIABLE` default keeps a `With` over
a definition that is deterministic but not cheap, which the new case would
inline.
- `DataSourceV2SQLSuiteV1Filter`: `DELETE ... WHERE (id + p) BETWEEN 0 AND
4` on a table that cannot delete by it fails with `Cannot delete from table ...
where [...]` listing both comparisons over the table's columns.
- `DeleteFromTableSuiteBase`: the same shape of condition deletes the
matching rows on the group-based and delta-based row-level tables, a
group-based delete copies only the rows of the groups the runtime group filter
keeps, and its condition has no `With` while its group filter condition keeps
one.
- `UpdateTableSuiteBase`: an UPDATE with that shape of condition updates
the matching row, its condition has no `With` and its group filter condition
keeps one, and on the group-based tables it copies no rows, as the runtime
group filter keeps only the group of that row. A condition that folds to true
leaves a true group filter condition and updates every row.
- `GeneratedColumnWriteSuite`: a table with `c INT GENERATED ALWAYS AS
(nullif(a + b, 3))` is created by CREATE TABLE and then replaced by CREATE OR
REPLACE TABLE, and INSERT computes `c` after each.
- `ProcedureSuite`: `CALL cat.ns.sum(nullif((SELECT 5), 3), 1)` returns 6.
- `CheckConstraintSuite`: after `ALTER TABLE ... ADD CONSTRAINT c1 CHECK
(nullif(id + 1, 3) > 0)`, an INSERT the check accepts succeeds and one it
rejects fails with CHECK_CONSTRAINT_VIOLATION; it passes with
`SPARK_ANSI_SQL_MODE=false` as well.
All of them fail without the change except the two delta-based DELETE runs,
which were not affected, and the `DECLARE VARIABLE` case, which passes on
master and fails if a `CreateVariable` reaches the new case. The UPDATE runs
fail only because master inlines their group filter condition; the UPDATE
itself works there.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 5
Closes #59217 from LuciferYang/SPARK-59962.
Lead-authored-by: YangJie <[email protected]>
Co-authored-by: reviewer <reviewer@local>
Signed-off-by: yangjie01 <[email protected]>
(cherry picked from commit 6bfccc1e5b166ceb026c1f93157418461fc98efb)
Signed-off-by: yangjie01 <[email protected]>
---
.../catalyst/optimizer/RewriteWithExpression.scala | 96 +++++++++++++++++++---
.../optimizer/RewriteWithExpressionSuite.scala | 46 ++++++++++-
.../spark/sql/connector/DataSourceV2SQLSuite.scala | 15 ++++
.../sql/connector/DeleteFromTableSuiteBase.scala | 30 +++++++
.../sql/connector/GeneratedColumnWriteSuite.scala | 20 +++++
.../spark/sql/connector/ProcedureSuite.scala | 7 ++
.../spark/sql/connector/UpdateTableSuiteBase.scala | 35 ++++++++
.../command/v2/CheckConstraintSuite.scala | 23 ++++++
8 files changed, 260 insertions(+), 12 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 722753c9958e..58bc78bc6e73 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, CreateVariable,
LogicalPlan, PlanHelper, Project}
+import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, Call, Command,
CreateVariable, LogicalPlan, PlanHelper, Project, ReplaceData, WriteDelta}
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
@@ -30,7 +30,11 @@ import org.apache.spark.util.Utils
/**
* Rewrites the `With` expressions by adding a `Project` to pre-evaluate the
common expressions, or
- * just inline them if they are cheap.
+ * just inline them if they are cheap. In a command other than
`CreateVariable`, whose expressions
+ * are mostly stored or translated rather than evaluated, and in a `Call`,
whose arguments must fold
+ * to literals, every deterministic definition is inlined whatever it costs,
and a nondeterministic
+ * one read more than once stays a `With`. The `groupFilterCondition` of a
row-level write keeps
+ * the `With`s worth memoizing for the runtime group filter.
*
* Since this rule can introduce new `Project` operators, it is advised to run
[[CollapseProject]]
* after this rule.
@@ -74,6 +78,36 @@ object RewriteWithExpression extends Rule[LogicalPlan] {
// `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)
+ // A row-level write reads its `groupFilterCondition` only in the
runtime group filter, which
+ // puts it in a filter over the table and optimizes it there. A `With`
in it is treated as one
+ // in a conditional branch: the definitions that gain nothing from
memoizing are inlined, so a
+ // condition that folds to true still injects no filter, and the rest
are left for that
+ // filter. Only `condition` is substituted as below.
+ case rd: ReplaceData =>
+ rowLevelConditions(rd.condition, rd.groupFilterCondition).map { case
(cond, groupFilter) =>
+ rd.copy(condition = cond, groupFilterCondition = groupFilter)
+ }.getOrElse(rd)
+ case wd: WriteDelta =>
+ rowLevelConditions(wd.condition, wd.groupFilterCondition).map { case
(cond, groupFilter) =>
+ wd.copy(condition = cond, groupFilterCondition = groupFilter)
+ }.getOrElse(wd)
+ // A command's own expressions are mostly stored as metadata or turned
into source
+ // predicates, which cannot hold a memoized value, and a definition
hoisted into a child would
+ // leave them reading a column only that `Project` produces, or leave
the command over a child
+ // its planner does not expect. A `Call` evaluates its arguments, but
`InvokeProcedures`
+ // needs them to fold to literals over its `ResolvedProcedure`, which
neither a `With` nor a
+ // `Project` under it allows. So a deterministic definition is
substituted whatever it costs
+ // (see `canSubstitute`), and the copies compound with nesting, as with
+ // `spark.sql.alwaysInlineCommonExpr`. Each copy is evaluated wherever
the expression is: a
+ // CHECK predicate stored by ALTER TABLE at every later write, and a
`Call` argument when it
+ // is folded, so an impure foldable such as `aes_encrypt` over literals
is folded per copy.
+ // Analysis lets no command but `CreateVariable` hold a nondeterministic
expression; a `Call`
+ // can hold one, as `InvokeProcedures` optimizes it before analysis
checks it, and then
+ // rejects it as not foldable. A nondeterministic definition read more
than once keeps its
+ // `With`. `CreateVariable` is matched above instead: its default is
evaluated and need not
+ // fold, so a definition there worth memoizing keeps its `With`.
+ case c @ (_: Command | _: Call) =>
+ c.mapExpressions(inlineCommandWiths)
case p if p.expressions.exists(_.containsPattern(WITH_EXPRESSION)) =>
applyInternal(p)
}
@@ -217,8 +251,9 @@ object RewriteWithExpression extends Rule[LogicalPlan] {
}
/**
- * Whether substituting this definition into its references is as good as
evaluating it once: it
- * is referenced at most once anyway, or it is cheap to evaluate twice and
deterministic.
+ * Whether this definition may be substituted into its references: it is
referenced at most once
+ * anyway, or it is deterministic and, under `requireCheap`, cheap to
evaluate twice. With
+ * `requireCheap`, substituting is as good as evaluating it once.
*
* `CollapseProject.isCheap` answers what one evaluation costs, not whether
a second is allowed --
* it admits a `PythonUDF`, which may be nondeterministic and is still in
the tree here, since
@@ -229,12 +264,23 @@ object RewriteWithExpression extends Rule[LogicalPlan] {
* as an `aes_encrypt` with no IV still gets substituted and folded per
copy. That predates this
* rule keeping a `With` -- the condition it replaced was no stricter -- and
belongs either on
* those expressions or in one purity predicate shared with
`isSafeToDuplicate`.
+ *
+ * Without `requireCheap`, as for a command's expressions
(`inlineCommandWiths`), the cost is not
+ * asked, so the gap widens to an impure definition that reads a column,
such as an
+ * `aes_encrypt` over one.
*/
private def canSubstitute(
child: Expression,
id: CommonExpressionId,
- multiplyReferenced: Set[CommonExpressionId]): Boolean = {
- !multiplyReferenced.contains(id) || (CollapseProject.isCheap(child) &&
child.deterministic)
+ multiplyReferenced: => Set[CommonExpressionId],
+ requireCheap: Boolean = true): Boolean = {
+ if (requireCheap) {
+ !multiplyReferenced.contains(id) || (CollapseProject.isCheap(child) &&
child.deterministic)
+ } else {
+ // Determinism first, so a caller passing the count lazily computes it
only once it meets a
+ // nondeterministic definition.
+ child.deterministic || !multiplyReferenced.contains(id)
+ }
}
/**
@@ -266,6 +312,31 @@ object RewriteWithExpression extends Rule[LogicalPlan] {
case w: With => inlineDefsThatGainNothing(w)
}
+ /**
+ * The `condition` and `groupFilterCondition` of a row-level write rewritten
as `apply` explains,
+ * or `None` if neither changes.
+ */
+ private def rowLevelConditions(
+ condition: Expression,
+ groupFilter: Option[Expression]): Option[(Expression,
Option[Expression])] = {
+ val newCondition = inlineCommandWiths(condition)
+ val newGroupFilter = groupFilter.map(inlineWithsThatGainNothing)
+ val unchanged = (newCondition eq condition) &&
+ newGroupFilter.zip(groupFilter).forall { case (n, o) => n eq o }
+ if (unchanged) None else Some((newCondition, newGroupFilter))
+ }
+
+ /**
+ * `e` with every `With` substituted as a command's expressions need,
bottom-up for nested ones:
+ * each deterministic definition whatever it costs, and any read at most
once.
+ */
+ private def inlineCommandWiths(e: Expression): Expression =
+ e.transformUpWithPruning(_.containsPattern(WITH_EXPRESSION)) {
+ case w: With =>
+ lazy val multiplyReferenced = multiplyReferencedIds(w.child, w.defs)
+ inlineDefs(w)(d => canSubstitute(d.child, d.id, multiplyReferenced,
requireCheap = false))
+ }
+
/**
* `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
@@ -280,13 +351,16 @@ object RewriteWithExpression extends Rule[LogicalPlan] {
*/
private def inlineDefsThatGainNothing(w: With): Expression = {
val multiplyReferenced = multiplyReferencedIds(w.child, w.defs)
- val (toInline, toKeep) = w.defs.partition { d =>
- canSubstitute(d.child, d.id, multiplyReferenced)
- }
- if (toInline.isEmpty) {
+ inlineDefs(w)(d => canSubstitute(d.child, d.id, multiplyReferenced))
+ }
+
+ /** `w` with the definitions `toInline` selects substituted at their
references. */
+ private def inlineDefs(w: With)(toInline: CommonExpressionDef => Boolean):
Expression = {
+ val (inlined, toKeep) = w.defs.partition(toInline)
+ if (inlined.isEmpty) {
w
} else {
- val refToExpr = toInline.map(d => d.id -> d.child).toMap
+ val refToExpr = inlined.map(d => d.id -> d.child).toMap
val newChild =
w.child.transformWithPruning(_.containsPattern(COMMON_EXPR_REF)) {
// A ref of a definition kept here, or of an enclosing `With`, is left
for its owner.
case ref: CommonExpressionRef if refToExpr.contains(ref.id) =>
refToExpr(ref.id)
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 81495236b586..8aaf15f0cd3a 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
@@ -28,7 +28,7 @@ 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.{CreateVariable,
DefaultValueExpression, LocalRelation, LogicalPlan, Project}
+import org.apache.spark.sql.catalyst.plans.logical.{CreateVariable,
DefaultValueExpression, DeleteFromTable, 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}
@@ -486,6 +486,50 @@ class RewriteWithExpressionSuite extends PlanTest {
)
}
+ test("SPARK-59962: a With in a command's condition is inlined instead of
hoisted") {
+ val a = testRelation.output.head
+ // The condition is translated into source predicates, not evaluated over
the child's rows, so a
+ // `Project` under the command would sit where the planner expects the
table.
+ val condition = With(a + a) { case Seq(ref) => ref < 10 && ref > 0 }
+ val inlined = (a + a) < 10 && (a + a) > 0
+ comparePlans(
+ Optimizer.execute(DeleteFromTable(testRelation, condition)),
+ DeleteFromTable(testRelation, inlined))
+
+ // Nor is one in a conditional branch kept: a `With` translates to no
source predicate.
+ comparePlans(
+ Optimizer.execute(DeleteFromTable(testRelation, If(a > 0, condition,
Literal(false)))),
+ DeleteFromTable(testRelation, If(a > 0, inlined, Literal(false))))
+ }
+
+ test("SPARK-59962: a nondeterministic definition in a command keeps its With
if read twice") {
+ // Analysis keeps nondeterministic expressions out of every command but
`CreateVariable`, so
+ // this only pins that the rule would not read such a definition twice.
+ val a = testRelation.output.head
+ val definition = Rand(Literal(0L)) + a
+ val delete = DeleteFromTable(testRelation, With(definition) { case
Seq(ref) =>
+ ref < 10 && ref > 0
+ })
+ assert(Optimizer.execute(delete) == delete)
+
+ // Read once, it is inlined as anywhere else.
+ assert(Optimizer.execute(DeleteFromTable(testRelation,
+ With(definition) { case Seq(ref) => ref < 10 })) ==
+ DeleteFromTable(testRelation, definition < 10))
+ }
+
+ test("SPARK-59962: a definition in a DECLARE VARIABLE default is not inlined
like a command's") {
+ // The default is evaluated rather than stored, so a definition there that
is worth memoizing
+ // stays in its `With` rather than being substituted at each of its
references as a command's.
+ val ident = ResolvedIdentifier(FakeSystemCatalog,
Identifier.of(Array("session"), "v"))
+ // Deterministic but not cheap: the command case would inline it.
+ val plusOne = ScalaUDF((i: Int) => i + 1, IntegerType, Seq(Literal(1)),
udfName = Some("f"))
+ val declare = CreateVariable(
+ Seq(ident), DefaultValueExpression(With(plusOne) { case Seq(ref) => ref
* ref }, "v"),
+ replace = true)
+ comparePlans(Optimizer.execute(declare), declare)
+ }
+
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 =
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala
index bd2eb571da70..e46c68d19d0b 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala
@@ -2683,6 +2683,21 @@ class DataSourceV2SQLSuiteV1Filter
}
}
+ test("SPARK-59962: DeleteFrom: a condition reading a value twice reaches the
source inlined") {
+ // BETWEEN reads `id + p` twice. Pre-evaluating it would put a `Project`
where the planner
+ // expects the table, so the condition has to reach the source inlined.
+ val t = "testcat.ns1.ns2.tbl"
+ withTable(t) {
+ sql(s"CREATE TABLE $t (id bigint, data string, p int) USING foo
PARTITIONED BY (id, p)")
+ checkError(
+ exception = analysisException(s"DELETE FROM $t WHERE (id + p) BETWEEN
0 AND 4"),
+ condition = "_LEGACY_ERROR_TEMP_1110",
+ parameters = Map(
+ "table" -> "testcat.ns1.ns2.tbl",
+ "filters" -> "[(id + CAST(p AS long)) >= 0, (id + CAST(p AS long))
<= 4]"))
+ }
+ }
+
test("UPDATE TABLE") {
val t = "testcat.ns1.ns2.tbl"
withTable(t) {
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/connector/DeleteFromTableSuiteBase.scala
b/sql/core/src/test/scala/org/apache/spark/sql/connector/DeleteFromTableSuiteBase.scala
index f8cc82eb8a89..22ac338fbd6e 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/connector/DeleteFromTableSuiteBase.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/connector/DeleteFromTableSuiteBase.scala
@@ -22,6 +22,7 @@ import org.apache.spark.sql.{AnalysisException, Row}
import org.apache.spark.sql.QueryTest.withQueryExecutionsCaptured
import org.apache.spark.sql.catalyst.expressions.CheckInvariant
import org.apache.spark.sql.catalyst.plans.logical.Filter
+import org.apache.spark.sql.catalyst.trees.TreePattern.WITH_EXPRESSION
import org.apache.spark.sql.connector.catalog.{Aborted, Committed,
InMemoryRowLevelOperationTable, InMemoryTable, InMemoryTruncatableOnlyTable,
InMemoryTruncatableOnlyTableCatalog, TableWritePrivilege}
import org.apache.spark.sql.connector.write.DeleteSummary
import org.apache.spark.sql.execution.datasources.v2.{DeleteFromTableExec,
ReplaceDataExec, TruncateTableExec, WriteDeltaExec}
@@ -259,6 +260,35 @@ abstract class DeleteFromTableSuiteBase extends
RowLevelOperationSuiteBase {
checkDeleteMetrics(numDeletedRows = 1, numCopiedRows = 1)
}
+ test("SPARK-59962: delete with a condition reading a computed value twice") {
+ // BETWEEN reads `id + pk` twice. A group-based delete pushes the
condition down to the scan
+ // and reuses it for runtime group filtering, where a column pre-evaluated
under the write is
+ // not found. The untouched `finance` group is copied only if that filter
is missing.
+ createAndInitTable("pk INT NOT NULL, id INT, dep STRING",
+ """{ "pk": 1, "id": 1, "dep": "hr" }
+ |{ "pk": 2, "id": 2, "dep": "software" }
+ |{ "pk": 3, "id": 3, "dep": "hr" }
+ |{ "pk": 4, "id": 10, "dep": "finance" }
+ |""".stripMargin)
+
+ val (cond, groupFilterCond) = executeAndKeepConditions {
+ sql(s"DELETE FROM $tableNameAsString WHERE (id + pk) BETWEEN 2 AND 4")
+ }
+ // The condition reaches the source inlined. A group-based write also has
a group filter
+ // condition, only evaluated in the runtime group filter, which keeps the
`With` for that
+ // filter to memoize.
+ assert(!cond.containsPattern(WITH_EXPRESSION), s"condition: $cond")
+ assert(groupFilterCond.isDefined == !deltaDelete, s"group filter
condition: $groupFilterCond")
+ assert(groupFilterCond.forall(_.containsPattern(WITH_EXPRESSION)),
+ s"group filter condition: $groupFilterCond")
+
+ checkAnswer(
+ sql(s"SELECT * FROM $tableNameAsString"),
+ Row(3, 3, "hr") :: Row(4, 10, "finance") :: Nil)
+
+ checkDeleteMetrics(numDeletedRows = 2, numCopiedRows = 1)
+ }
+
test("delete with aliases") {
createAndInitTable("pk INT NOT NULL, id INT, dep STRING",
"""{ "pk": 1, "id": 1, "dep": "hr" }
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/connector/GeneratedColumnWriteSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/connector/GeneratedColumnWriteSuite.scala
index c963d91005f1..943d883799e1 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/connector/GeneratedColumnWriteSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/connector/GeneratedColumnWriteSuite.scala
@@ -456,6 +456,26 @@ class GeneratedColumnWriteSuite extends QueryTest with
DatasourceV2SQLBase {
}
}
+ test("SPARK-59962: CREATE and REPLACE TABLE with a generation expression
reading a value twice") {
+ // NULLIF reads `a + b` twice. The table is created from the expression
rather than evaluating
+ // it, so nothing may be pre-evaluated under the CREATE or REPLACE TABLE.
The second statement
+ // replaces the table the first one created.
+ val tblName = "my_tab"
+ withTable(s"testcat.$tblName") {
+ Seq("CREATE TABLE", "CREATE OR REPLACE TABLE").foreach { create =>
+ sql(s"""$create testcat.$tblName(
+ | a INT,
+ | b INT,
+ | c INT GENERATED ALWAYS AS (nullif(a + b, 3))
+ |) USING foo""".stripMargin)
+ sql(s"INSERT INTO testcat.$tblName(a, b) VALUES (1, 1), (1, 2)")
+ checkAnswer(
+ spark.table(s"testcat.$tblName"),
+ Seq(Row(1, 1, 2), Row(1, 2, null)))
+ }
+ }
+ }
+
test("allowNullableIngest config controls missing non-generated columns") {
val tblName = "my_tab"
withTable(s"testcat.$tblName") {
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/connector/ProcedureSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/connector/ProcedureSuite.scala
index cf45f4f416dc..76bed5649ffd 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/connector/ProcedureSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/connector/ProcedureSuite.scala
@@ -98,6 +98,13 @@ class ProcedureSuite extends SharedSparkSession with
BeforeAndAfter {
checkAnswer(sql("CALL cat.ns.sum((1 + 1) * 2, in2 => (2 + 1) / 3)"),
Row(5) :: Nil)
}
+ test("SPARK-59962: an argument reading a value twice") {
+ // NULLIF reads the scalar subquery twice. Pre-evaluating it in a
`Project` over the
+ // `ResolvedProcedure` would leave a plan `InvokeProcedures` does not
expect.
+ catalog.createProcedure(Identifier.of(Array("ns"), "sum"), UnboundSum)
+ checkAnswer(sql("CALL cat.ns.sum(nullif((SELECT 5), 3), 1)"), Row(6) ::
Nil)
+ }
+
test("type coercion") {
catalog.createProcedure(Identifier.of(Array("ns"), "sum"), UnboundLongSum)
checkAnswer(sql("CALL cat.ns.sum(1, 2)"), Row(3) :: Nil)
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/connector/UpdateTableSuiteBase.scala
b/sql/core/src/test/scala/org/apache/spark/sql/connector/UpdateTableSuiteBase.scala
index 3269081227e5..c0506ca03dcc 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/connector/UpdateTableSuiteBase.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/connector/UpdateTableSuiteBase.scala
@@ -21,7 +21,9 @@ import org.apache.spark.SparkRuntimeException
import org.apache.spark.internal.config
import org.apache.spark.sql.{sources, AnalysisException, Row}
import org.apache.spark.sql.QueryTest.withQueryExecutionsCaptured
+import org.apache.spark.sql.catalyst.expressions.Literal.TrueLiteral
import org.apache.spark.sql.catalyst.plans.logical.{ReplaceData, WriteDelta}
+import org.apache.spark.sql.catalyst.trees.TreePattern.WITH_EXPRESSION
import org.apache.spark.sql.connector.catalog.{Aborted, Column,
ColumnDefaultValue, Committed, InMemoryTable, TableChange, TableInfo}
import org.apache.spark.sql.connector.expressions.{GeneralScalarExpression,
LiteralValue}
import org.apache.spark.sql.connector.write.UpdateSummary
@@ -179,6 +181,39 @@ abstract class UpdateTableSuiteBase extends
RowLevelOperationSuiteBase {
checkUpdateMetrics(numUpdatedRows = 1, numCopiedRows = 1)
}
+ test("SPARK-59962: update with a condition reading a computed value twice") {
+ // BETWEEN reads `salary + pk` twice. The condition keeps no `With`, and
the group filter
+ // condition, only evaluated in the runtime group filter, keeps it for
that filter to
+ // memoize; with the filter, a group-based update copies none of the `hr`
rows. One that
+ // gains nothing from memoizing is still inlined there, so a condition
that folds to true
+ // leaves a true group filter condition.
+ createAndInitTable("pk INT NOT NULL, salary INT, dep STRING",
+ """{ "pk": 1, "salary": 100, "dep": "hr" }
+ |{ "pk": 2, "salary": 200, "dep": "software" }
+ |{ "pk": 3, "salary": 300, "dep": "hr" }
+ |""".stripMargin)
+
+ val (cond, groupFilterCond) = executeAndKeepConditions {
+ sql(s"UPDATE $tableNameAsString SET dep = 'x' WHERE (salary + pk)
BETWEEN 150 AND 250")
+ }
+ assert(!cond.containsPattern(WITH_EXPRESSION), s"condition: $cond")
+ assert(groupFilterCond.exists(_.containsPattern(WITH_EXPRESSION)),
+ s"group filter condition: $groupFilterCond")
+ checkAnswer(
+ sql(s"SELECT * FROM $tableNameAsString"),
+ Row(1, 100, "hr") :: Row(2, 200, "x") :: Row(3, 300, "hr") :: Nil)
+ checkUpdateMetrics(numUpdatedRows = 1, numCopiedRows = 0)
+
+ val (_, constantGroupFilterCond) = executeAndKeepConditions {
+ sql(s"UPDATE $tableNameAsString SET dep = 'y' WHERE 1 BETWEEN 0 AND 2")
+ }
+ assert(constantGroupFilterCond.contains(TrueLiteral),
+ s"group filter condition: $constantGroupFilterCond")
+ checkAnswer(
+ sql(s"SELECT * FROM $tableNameAsString"),
+ Row(1, 100, "y") :: Row(2, 200, "y") :: Row(3, 300, "y") :: Nil)
+ }
+
test("update with aliases") {
createAndInitTable("pk INT NOT NULL, salary INT, dep STRING",
"""{ "pk": 1, "salary": 100, "dep": "hr" }
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/CheckConstraintSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/CheckConstraintSuite.scala
index 8e295ecc3d5d..631c7bfb37c9 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/CheckConstraintSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/CheckConstraintSuite.scala
@@ -147,6 +147,29 @@ class CheckConstraintSuite extends QueryTest with
CommandSuiteBase with DDLComma
}
}
+ test("SPARK-59962: Check added by ALTER TABLE reading a computed value
twice") {
+ // NULLIF reads `id + 1` twice. A predicate stored with a column
pre-evaluated under the ALTER
+ // TABLE would make every later write fail to resolve it.
+ withNamespaceAndTable("ns", "tbl", nonPartitionCatalog) { t =>
+ sql(s"CREATE TABLE $t (id bigint, data string) $defaultUsing")
+ sql(s"ALTER TABLE $t ADD CONSTRAINT c1 CHECK (nullif(id + 1, 3) > 0)")
+ // `nullif(3, 3)` is NULL, which a check accepts.
+ sql(s"INSERT INTO $t VALUES (2, 'a')")
+ val error = intercept[SparkRuntimeException] {
+ sql(s"INSERT INTO $t VALUES (-5, 'b')")
+ }
+ checkError(
+ exception = error,
+ condition = "CHECK_CONSTRAINT_VIOLATION",
+ sqlState = "23001",
+ parameters = Map(
+ "constraintName" -> "c1",
+ "expression" -> "nullif(id + 1, 3) > 0",
+ "values" -> " - id : -5"))
+ checkAnswer(sql(s"SELECT * FROM $t"), Row(2, "a"))
+ }
+ }
+
def getConstraintCharacteristics(): Seq[(String, String)] = {
Seq(
("", s"ENFORCED NORELY"),
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
