dongjoon-hyun commented on code in PR #59217:
URL: https://github.com/apache/spark/pull/59217#discussion_r4208792783
##########
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:
Minor: this test checks the shape of `groupFilterCondition`, but not that
the runtime group filter still works with the `With` kept in it, which is what
f397054 changes. The DELETE test pins that with
`checkDeleteMetrics(numDeletedRows = 2, numCopiedRows = 1)`, but a group-based
UPDATE builds the filter another way: `buildMatchingRowsPlan` remaps the
attributes of `cond` through `attrMap`
(RowLevelOperationRuntimeGroupFiltering.scala:110-115) before
`optimizeSubqueries` hoists the `With`. Here only pk 2 matches, so with the
filter the group-based run should copy no rows, and without it, both `hr` rows.
The neighboring tests call `checkUpdateMetrics`. The second UPDATE checks only
its group filter condition, not the rows it updates. Could this call
`checkUpdateMetrics(numUpdatedRows = 1, numCopiedRows = 0)` after the first
UPDATE, and `checkAnswer` after the second?
##########
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:
Minor: this comment, like the scaladoc at L33, puts `Call` with the commands
whose expressions are stored or translated rather than evaluated. But
`InvokeProcedures` evaluates every argument (`args.map(_.eval())`,
InvokeProcedures.scala:82), and "its default is evaluated rather than stored"
is the reason this comment gives for leaving `CreateVariable` out. Its
expressions aren't evaluated over its children's rows either, since those
children are just the `ResolvedIdentifier` names. What sets `Call` apart is
that its arguments must fold to literals over a fixed `ResolvedProcedure`, so
each copy is folded once at planning time. The consequence is that an impure
foldable in an argument, such as `aes_encrypt` over literals, is folded per
copy. Could the comment and the scaladoc give that as the reason for `Call`,
rather than putting it with the stored expressions?
##########
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:
Nit: putting the `deterministic`/`isCheap` term first makes
`CollapseProject.isCheap` run for every definition on the generic and
conditional-branch paths. That includes definitions read once, which the old
order settled with the set lookup. There `multiplyReferenced` is already a
strict value, so only `inlineCommandWiths` gains from the by-name parameter. It
is a small tree walk, so this is cosmetic. Could the old order stay when
`requireCheap` is set, e.g. `if (requireCheap) !multiplyReferenced.contains(id)
|| (CollapseProject.isCheap(child) && child.deterministic) else
child.deterministic || !multiplyReferenced.contains(id)`?
--
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]
