LuciferYang commented on code in PR #59217:
URL: https://github.com/apache/spark/pull/59217#discussion_r4192296567
##########
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:
`isSafeToDuplicate` only accepts literal trees, so any definition that reads
a column fails it. With it as the selector the CHECK and DELETE tests fail
again: the `With` stays in an expression that is stored or translated into a
source predicate, and neither can hold a memoized value. So the gap is the one
`canSubstitute` already documents for impure expressions that declare
themselves deterministic; the comment on this case now says so and points there.
##########
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:
Measured with a counting UDF over 4 rows in 3 groups on a group-based table:
`WHERE f(id) BETWEEN 2 AND 3` calls it 11 times with the runtime group filter
on and 4 with it off. Written out as `f(id) >= 2 AND f(id) <= 3` it is 14 both
before and after this PR, and before this PR the BETWEEN form failed with the
filter on. Keeping a `With` in that filter would need the condition in two
forms, since the same expression is also translated for pushdown, so I noted
the trade-off in the comment rather than change it here.
##########
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:
Done. It counts over the whole plan with `collectWithSubqueries` and asserts
`== 1`; without the `CreateVariable` guard it counts 2.
##########
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:
Done. Added a row in a `finance` group the condition does not match, and the
test calls `checkDeleteMetrics(numDeletedRows = 2, numCopiedRows = 1)`. With
the runtime group filter disabled it fails with 2 copied rows.
##########
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:
Agreed that this belongs on the command, and `CheckAnalysis` should read the
same property. It changes the `Command` API and both rules, so I'd rather do it
separately: SPARK-59994.
--
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]
