dongjoon-hyun commented on code in PR #59217:
URL: https://github.com/apache/spark/pull/59217#discussion_r4187715493
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteWithExpression.scala:
##########
@@ -67,6 +67,18 @@ 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. 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 it is substituted instead. Analysis lets
no command but
+ // `CreateVariable` hold a nondeterministic expression; one that does
keeps its `With` rather
+ // than be read twice. `CreateVariable` is left out: its default is
evaluated rather than
+ // stored, so a definition there should be memoized, not inlined.
+ case c: Command if !c.isInstanceOf[CreateVariable] &&
+ c.expressions.exists(_.containsPattern(WITH_EXPRESSION)) =>
+
c.mapExpressions(_.transformUpWithPruning(_.containsPattern(WITH_EXPRESSION)) {
+ case w: With => inlineDefs(w)(_.child.deterministic)
Review Comment:
**This selects definitions by `deterministic` alone, which is weaker than
`isSafeToDuplicate` (L161-196).** `applyForExpression` (L101) uses that check
for the same situation, where there is no plan to pre-evaluate into, and its
scaladoc lists foldable-but-impure cases that `deterministic` accepts:
`aes_encrypt` with a random IV, and a Hive UDF that declares itself
deterministic. For example, in `ALTER TABLE t ADD CONSTRAINT c CHECK
(aes_encrypt(data, k) BETWEEN lo AND hi)` or a DELETE condition of the same
shape, both copies of the definition are kept, and the two comparisons, or
NULLIF's test and its result, read different values. Before this PR the
non-leaf command path hoisted the definition, so it was evaluated once (in the
paths that did not fail). Could this use `isSafeToDuplicate` for multiply
referenced definitions, keeping the `With` otherwise, or at least note this gap
next to the one documented on `canSubstitute` (L232-236)?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteWithExpression.scala:
##########
@@ -67,6 +67,18 @@ 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. 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 it is substituted instead. Analysis lets
no command but
+ // `CreateVariable` hold a nondeterministic expression; one that does
keeps its `With` rather
+ // than be read twice. `CreateVariable` is left out: its default is
evaluated rather than
+ // stored, so a definition there should be memoized, not inlined.
+ case c: Command if !c.isInstanceOf[CreateVariable] &&
Review Comment:
Minor: `!c.isInstanceOf[CreateVariable]` hardcodes whether a command stores
its expressions or evaluates them. `CheckAnalysis` hardcodes `CreateVariable`
the same way (CheckAnalysis.scala:1173). Any other `Command` that evaluates its
own expressions, such as one from an extension or a future scripting command,
now gets every deterministic definition inlined instead of memoized, and
nothing points its author here. Could this be a property of the command, for
example a method on `Command` or a marker trait that both `CheckAnalysis` and
this rule consult, either here or in a follow-up?
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/DeleteFromTableSuiteBase.scala:
##########
@@ -259,6 +259,21 @@ 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.
+ 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" }
+ |""".stripMargin)
+
+ sql(s"DELETE FROM $tableNameAsString WHERE (id + pk) BETWEEN 2 AND 4")
Review Comment:
Minor: the neighboring tests call `checkDeleteMetrics`, and this one checks
only the remaining rows. The failure this PR fixes (`Could not find
_common_expr_4 in [dep]`) was in the runtime group filter, so if that filter
stopped being injected for this condition, this test would still pass without
covering the fixed path. Could it also call `checkDeleteMetrics`, with the
copied-row count that group filtering produces, so that the filter is pinned?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteWithExpression.scala:
##########
@@ -67,6 +67,18 @@ 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. 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 it is substituted instead. Analysis lets
no command but
+ // `CreateVariable` hold a nondeterministic expression; one that does
keeps its `With` rather
+ // than be read twice. `CreateVariable` is left out: its default is
evaluated rather than
+ // stored, so a definition there should be memoized, not inlined.
+ case c: Command if !c.isInstanceOf[CreateVariable] &&
+ c.expressions.exists(_.containsPattern(WITH_EXPRESSION)) =>
+
c.mapExpressions(_.transformUpWithPruning(_.containsPattern(WITH_EXPRESSION)) {
+ case w: With => inlineDefs(w)(_.child.deterministic)
Review Comment:
Nit: this keeps a nondeterministic definition even when it is referenced
only once. `canSubstitute` (L242) inlines any definition with at most one
reference, since that cannot change semantics. A kept `With` cannot be
translated into a V2 predicate and hides the expression from later folding
(L273-278). Before this PR, a leaf command inlined such a definition through
the `childPlanIndex == -1` path. Analysis keeps nondeterministic expressions
out of these commands today, so this is about consistency more than a reachable
bug. Could the selector be `d => !multiplyReferenced.contains(d.id) ||
d.child.deterministic`, using `multiplyReferencedIds(w.child, w.defs)`?
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/RewriteWithExpressionSuite.scala:
##########
@@ -486,6 +486,48 @@ 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 }
+ comparePlans(
+ Optimizer.execute(DeleteFromTable(testRelation, condition)),
+ DeleteFromTable(testRelation, (a + a) < 10 && (a + a) > 0))
+
+ // Nor is one in a conditional branch kept: a `With` translates to no
source predicate.
+ val inBranch = If(a > 0, With(a + a) { case Seq(ref) => ref < 10 && ref >
0 }, Literal(false))
+ comparePlans(
+ Optimizer.execute(DeleteFromTable(testRelation, inBranch)),
+ DeleteFromTable(testRelation, If(a > 0, (a + a) < 10 && (a + a) > 0,
Literal(false))))
+ }
+
+ test("SPARK-59962: a nondeterministic definition in a command keeps its
With") {
+ // 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 delete = DeleteFromTable(testRelation, With(Rand(Literal(0L)) + a) {
case Seq(ref) =>
+ ref < 10 && ref > 0
+ })
+ assert(Optimizer.execute(delete) == delete)
+ }
+
+ 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
+ // must not be substituted at each of its references the way a command's
is. Until SPARK-59954
+ // keeps it in the default, the generic case pre-evaluates it under the
variable names instead;
+ // either way there is at most one copy.
+ 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)
+ val copies = Optimizer.execute(declare).expressions
+ .map(_.collect { case u: ScalaUDF => u }.size).sum
+ assert(copies <= 1, s"the definition was substituted at $copies
references")
Review Comment:
Minor: this counts copies only in `Optimizer.execute(declare).expressions`,
which are the `CreateVariable`'s own expressions. On this branch the generic
case hoists the UDF into `Project(..., ResolvedIdentifier)`, so the default
holds only `_common_expr_0` and `copies` is 0. The test would also pass if the
definition were dropped or folded away, so it catches only the 2-copy
regression that HyukjinKwon's point asked about. Could it count over the whole
plan, e.g. `Optimizer.execute(declare).collectWithSubqueries { case p =>
p.expressions }.flatten`, and assert `== 1`?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteWithExpression.scala:
##########
@@ -67,6 +67,18 @@ 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
Review Comment:
Nit: the object scaladoc (L31-40) still says the rule either adds a
`Project` to pre-evaluate the common expressions or inlines them if they are
cheap. With this case there is a third outcome: in a command, every
deterministic definition is inlined whatever it costs, and a nondeterministic
one stays a `With`. Could the scaladoc mention it?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteWithExpression.scala:
##########
@@ -67,6 +67,18 @@ 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. 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 it is substituted instead. Analysis lets
no command but
+ // `CreateVariable` hold a nondeterministic expression; one that does
keeps its `With` rather
+ // than be read twice. `CreateVariable` is left out: its default is
evaluated rather than
+ // stored, so a definition there should be memoized, not inlined.
+ case c: Command if !c.isInstanceOf[CreateVariable] &&
+ c.expressions.exists(_.containsPattern(WITH_EXPRESSION)) =>
+
c.mapExpressions(_.transformUpWithPruning(_.containsPattern(WITH_EXPRESSION)) {
Review Comment:
**The condition of a row-level command is not only metadata: it is also
evaluated per row.** For a group-based or delta-based DELETE,
`RowLevelOperationRuntimeGroupFiltering`
(RowLevelOperationRuntimeGroupFiltering.scala:55-65) wraps `cond` in a `Filter`
over the table scan to find the groups to rewrite. Inlining every deterministic
definition regardless of cost means that `DELETE FROM t WHERE expensive_udf(id)
BETWEEN 2 AND 4` now evaluates `expensive_udf` twice per scanned row there, and
a scalar subquery in the definition is copied twice with the same exprId.
`canSubstitute` (L242) avoids this with `CollapseProject.isCheap`. Could the
trade-off be noted in the comment above, or a non-cheap definition be kept as a
`With` in that injected filter?
--
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]
