LuciferYang commented on code in PR #59217:
URL: https://github.com/apache/spark/pull/59217#discussion_r4209934715
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/UpdateTableSuiteBase.scala:
##########
@@ -179,6 +181,34 @@ 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. 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)
Review Comment:
Done: `checkUpdateMetrics(numUpdatedRows = 1, numCopiedRows = 0)` after the
first UPDATE and `checkAnswer` after the second. With the runtime group filter
disabled for the first UPDATE, the group-based run copies both `hr` rows and
fails.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteWithExpression.scala:
##########
@@ -67,6 +71,34 @@ 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 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 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, and
+ // a `Call` must see foldable arguments over its `ResolvedProcedure`
(`InvokeProcedures`). 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, or
`InvokeProcedures`, does not
+ // expect, so a deterministic one is substituted whatever it costs (see
`canSubstitute`).
+ // The copies compound with nesting, as with
`spark.sql.alwaysInlineCommonExpr`, and where
+ // the expression is evaluated after all, as a CHECK predicate stored by
ALTER TABLE is at
+ // every later write, each copy is evaluated. 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
left out: its default
+ // is evaluated rather than stored, so a definition there should be
memoized.
+ case c @ (_: Command | _: Call) if !c.isInstanceOf[CreateVariable] =>
Review Comment:
Done: the comment and the scaladoc now give `Call` its own reason. Its
arguments must fold to literals over its `ResolvedProcedure`, so each copy is
folded, and an impure foldable such as `aes_encrypt` over literals is folded
per copy.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteWithExpression.scala:
##########
@@ -222,12 +255,18 @@ 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 = {
+ (child.deterministic && (!requireCheap || CollapseProject.isCheap(child)))
||
Review Comment:
Done, the old order is back when `requireCheap` is set.
--
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]
