LuciferYang commented on code in PR #59217:
URL: https://github.com/apache/spark/pull/59217#discussion_r4206239227
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteWithExpression.scala:
##########
@@ -67,6 +69,23 @@ 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
+ // A command's own expressions are not evaluated over its children's
rows: most are stored as
+ // metadata or turned into source predicates, neither of which can hold
a memoized value. 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, so a
+ // deterministic one is substituted whatever it costs. Where the
expression is evaluated
Review Comment:
The comment now names this path. As you note in the next comment, building
the stored predicate from the analyzed expression would keep `predicateSql` as
the source and avoid the branch case; I've filed that as SPARK-60032.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/CheckConstraintSuite.scala:
##########
@@ -147,6 +147,21 @@ class CheckConstraintSuite extends QueryTest with
CommandSuiteBase with DDLComma
}
}
+ test("SPARK-59962: Predicate of a check reading a computed value twice") {
+ // NULLIF reads `id + 1` twice. The predicate stored in the catalog is
built from the check as
+ // the optimizer leaves it, so it must not refer to a column pre-evaluated
under the command.
Review Comment:
Agreed, and the ALTER-time freeze reproduces: after `ALTER TABLE ... ADD
CONSTRAINT vt CHECK (ts <= current_timestamp())` the stored predicate is `ts <=
<literal>`, and an INSERT of `current_timestamp()` a second later fails with
CHECK_CONSTRAINT_VIOLATION. The test now asserts the enforcement, and building
the predicate from the analyzed expression is SPARK-60032.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/CheckConstraintSuite.scala:
##########
@@ -147,6 +147,21 @@ class CheckConstraintSuite extends QueryTest with
CommandSuiteBase with DDLComma
}
}
+ test("SPARK-59962: Predicate of a check reading a computed value twice") {
+ // NULLIF reads `id + 1` twice. The predicate stored in the catalog is
built from the check as
+ // the optimizer leaves it, so it must not refer to a column pre-evaluated
under the command.
+ 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)")
Review Comment:
I haven't reproduced the upgrade path, since it needs a catalog that
persists the V2 predicate across versions. Filed the `predicateSql` fallback as
SPARK-60033.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteWithExpression.scala:
##########
@@ -67,6 +69,23 @@ 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
+ // A command's own expressions are not evaluated over its children's
rows: most are stored as
+ // metadata or turned into source predicates, neither of which can hold
a memoized value. 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, so a
+ // deterministic one is substituted whatever it costs. Where the
expression is evaluated
+ // after all, as by the runtime group filter of a row-level command,
that evaluates it at
+ // every reference, and it duplicates an impure one that declares itself
deterministic (see
+ // `canSubstitute`). Analysis lets no command but `CreateVariable` hold
a nondeterministic
+ // expression; one read more than once keeps its `With`.
`CreateVariable` is left out: its
+ // default is evaluated rather than stored, so a definition there should
be memoized.
+ case c: Command if !c.isInstanceOf[CreateVariable] &&
Review Comment:
Covered since b2a5c098330, which extends the case to `Call`. Your CALL
returns 2 now; on f26b155 it failed with Unexpected plan.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteWithExpression.scala:
##########
@@ -67,6 +69,23 @@ 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
+ // A command's own expressions are not evaluated over its children's
rows: most are stored as
+ // metadata or turned into source predicates, neither of which can hold
a memoized value. 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, so a
+ // deterministic one is substituted whatever it costs. Where the
expression is evaluated
+ // after all, as by the runtime group filter of a row-level command,
that evaluates it at
+ // every reference, and it duplicates an impure one that declares itself
deterministic (see
Review Comment:
Done: the command case goes through `canSubstitute(..., requireCheap =
false)`, and the doc on `canSubstitute` says that without the cost check the
gap widens to impure definitions that read a column.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteWithExpression.scala:
##########
@@ -67,6 +69,23 @@ 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
+ // A command's own expressions are not evaluated over its children's
rows: most are stored as
+ // metadata or turned into source predicates, neither of which can hold
a memoized value. 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, so a
+ // deterministic one is substituted whatever it costs. Where the
expression is evaluated
+ // after all, as by the runtime group filter of a row-level command,
that evaluates it at
+ // every reference, and it duplicates an impure one that declares itself
deterministic (see
+ // `canSubstitute`). Analysis lets no command but `CreateVariable` hold
a nondeterministic
+ // expression; one read more than once keeps its `With`.
`CreateVariable` is left out: its
+ // default is evaluated rather than stored, so a definition there should
be memoized.
+ case c: Command if !c.isInstanceOf[CreateVariable] &&
+ c.expressions.exists(_.containsPattern(WITH_EXPRESSION)) =>
+
c.mapExpressions(_.transformUpWithPruning(_.containsPattern(WITH_EXPRESSION)) {
Review Comment:
The case now calls a shared helper through `mapExpressions`, since the
row-level arm applies the same helper to `condition` alone.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteWithExpression.scala:
##########
@@ -67,6 +69,23 @@ 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
+ // A command's own expressions are not evaluated over its children's
rows: most are stored as
+ // metadata or turned into source predicates, neither of which can hold
a memoized value. 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, so a
+ // deterministic one is substituted whatever it costs. Where the
expression is evaluated
+ // after all, as by the runtime group filter of a row-level command,
that evaluates it at
+ // every reference, and it duplicates an impure one that declares itself
deterministic (see
+ // `canSubstitute`). Analysis lets no command but `CreateVariable` hold
a nondeterministic
+ // expression; one read more than once keeps its `With`.
`CreateVariable` is left out: its
+ // default is evaluated rather than stored, so a definition there should
be memoized.
+ case c: Command if !c.isInstanceOf[CreateVariable] &&
+ c.expressions.exists(_.containsPattern(WITH_EXPRESSION)) =>
+
c.mapExpressions(_.transformUpWithPruning(_.containsPattern(WITH_EXPRESSION)) {
+ case w: With =>
+ val multiplyReferenced = multiplyReferencedIds(w.child, w.defs)
Review Comment:
Done, as a by-name parameter of `canSubstitute` backed by a `lazy val`.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala:
##########
@@ -2683,6 +2683,23 @@ 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)")
+ sql(s"INSERT INTO $t VALUES (2L, 'a', 2), (2L, 'b', 3), (3L, 'c', 3)")
+ 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]"))
+ assert(spark.table(t).count() === 3)
Review Comment:
Dropped the INSERT and the count. I kept it as its own test rather than a
table with its neighbor, so the SPARK-59962 name stays on the case this PR
fixes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
