dongjoon-hyun commented on code in PR #52185:
URL: https://github.com/apache/spark/pull/52185#discussion_r4459312711
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/GroupBasedMergeIntoTableSuite.scala:
##########
@@ -218,6 +219,253 @@ class GroupBasedMergeIntoTableSuite extends
MergeIntoTableSuiteBase {
}
}
+ test("merge pushes conditional NOT MATCHED BY SOURCE predicates to target") {
Review Comment:
**2. [Low] No test keeps an untouched row in a group that is read**
The two tests that check replaced partitions (this one and the `ON 1 = 0`
one) update or delete every row in the groups that are read. In the other
tests, the in-memory table cannot evaluate the pushed `OR`, so every group is
read.
So no test has a group that is read because of the new condition and also
has a row that matches nothing and must be carried over. A small change to this
test would cover that case. Add `(5, 100, 'hr')` to the target and use this
clause:
```sql
WHEN NOT MATCHED BY SOURCE AND t.dep IN ('hr', 'finance') AND t.salary > 150
THEN
DELETE
```
`BooleanSimplification` turns `D OR (D AND x)` into `D`, so the in-memory
table still prunes `software` and the `cond.sql` assertion stays the same. Then
`(5, 100, 'hr')` must still be in the replaced `hr` partition.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteMergeIntoTable.scala:
##########
@@ -174,16 +174,29 @@ object RewriteMergeIntoTable extends
RewriteRowLevelCommand with PredicateHelper
readRelation, joinPlan, matchedActions, notMatchedActions,
notMatchedBySourceActions, metadataAttrs, checkCardinality)
- // predicates of the ON condition can be used to filter the target table
(planning & runtime)
+ // predicates that can be used to filter the target table (planning):
+ // - without NOT MATCHED BY SOURCE clauses, target rows that don't satisfy
the ON condition
+ // match no action and are simply carried over, so the ON condition can
filter the target
+ // - if every NOT MATCHED BY SOURCE clause is conditional, a target row
can be skipped only
+ // if it satisfies neither the ON condition nor any of those conditions,
so the filter is
+ // their disjunction; only the target-only part of the ON condition can
take part, as a
+ // disjunction that references the source is not a valid filter for the
target table
+ // - an unconditional NOT MATCHED BY SOURCE clause matches any target row
and disables this
+ val pushableCond = if (notMatchedBySourceActions.isEmpty) {
+ cond
+ } else if (notMatchedBySourceActions.exists(_.condition.isEmpty)) {
+ TrueLiteral
+ } else {
+ val (targetCond, _) = splitTargetPredicates(readRelation, cond)
+ (targetCond +: notMatchedBySourceActions.flatMap(_.condition)).reduce(Or)
Review Comment:
**1. [Medium] DELETE-only MERGE on an unpadded CHAR column deletes too
little**
With this change, a MERGE whose actions are all DELETE can now prune groups
with a CHAR condition that the rest of the plan evaluates differently.
`ApplyCharTypePadding` runs after this rule. It puts read-side padding over
the read relation and remaps the join and `MergeRows` to the padded column. It
does not remap `ReplaceData.condition`. That is because
`ReplaceData.references` is `query.outputSet`
([v2Commands.scala#L400](https://github.com/apache/spark/blob/f2b3c156118b115caf6a382d2d0b017719abbaf8/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/v2Commands.scala#L400)),
and `transformUpWithNewOutput` only remaps attributes in `references`
([QueryPlan.scala#L394-L398](https://github.com/apache/spark/blob/f2b3c156118b115caf6a382d2d0b017719abbaf8/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/QueryPlan.scala#L394-L398)).
Its string comparison pass still pads the literals of that condition
([ApplyCharTypePaddingHelper.scala#L101-L121](https://github.com/apache/spark/blob/f2b3c156118b115caf6a382d2d0b017719abbaf8/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ApplyCha
rTypePaddingHelper.scala#L101-L121)).
So the pushed predicate compares the stored value with a padded literal,
while `MergeRows` compares the padded value with the same literal. A MERGE with
assignments is different: `ResolveRowLevelCommandAssignments` strips the CHAR
metadata first, so both sides compare the stored value, and the pushdown adds
no new difference there.
Here is an example. `dep CHAR(8)` is the partition column, and the rows are
stored without padding, for example because they were written with
`spark.sql.legacy.charVarcharAsString=true` or by another engine:
```sql
-- t: (1, 'hr'), (3, 'finance'); s: (1)
MERGE INTO t USING s
ON t.pk = s.pk AND t.dep IN ('hr', 'finance')
WHEN MATCHED THEN DELETE
WHEN NOT MATCHED BY SOURCE AND t.dep IN ('hr', 'finance') THEN DELETE
```
- **master:** The NOT MATCHED BY SOURCE clause makes the pushed condition
`true`, so every group is read and both rows are deleted.
- **This PR:** The pushed condition is `dep IN ('hr ', 'finance ')`. No
stored partition value matches it, so no group is read and nothing is deleted.
There is no error.
I found this by reading the code and did not run this PR.
On master, the same mismatch already affects DELETE-only MERGE without NOT
MATCHED BY SOURCE clauses, where the ON condition is pushed in the same way. So
the root cause is in `ApplyCharTypePadding` and should be fixed there
separately. Could this PR avoid widening it, for example by keeping
`TrueLiteral` when the new disjunction references a CHAR column? Otherwise, the
PR description's "No change in query results" would need a caveat.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteMergeIntoTable.scala:
##########
@@ -174,16 +174,29 @@ object RewriteMergeIntoTable extends
RewriteRowLevelCommand with PredicateHelper
readRelation, joinPlan, matchedActions, notMatchedActions,
notMatchedBySourceActions, metadataAttrs, checkCardinality)
- // predicates of the ON condition can be used to filter the target table
(planning & runtime)
+ // predicates that can be used to filter the target table (planning):
+ // - without NOT MATCHED BY SOURCE clauses, target rows that don't satisfy
the ON condition
+ // match no action and are simply carried over, so the ON condition can
filter the target
+ // - if every NOT MATCHED BY SOURCE clause is conditional, a target row
can be skipped only
+ // if it satisfies neither the ON condition nor any of those conditions,
so the filter is
+ // their disjunction; only the target-only part of the ON condition can
take part, as a
+ // disjunction that references the source is not a valid filter for the
target table
+ // - an unconditional NOT MATCHED BY SOURCE clause matches any target row
and disables this
+ val pushableCond = if (notMatchedBySourceActions.isEmpty) {
+ cond
+ } else if (notMatchedBySourceActions.exists(_.condition.isEmpty)) {
+ TrueLiteral
+ } else {
+ val (targetCond, _) = splitTargetPredicates(readRelation, cond)
+ (targetCond +: notMatchedBySourceActions.flatMap(_.condition)).reduce(Or)
+ }
+
+ // predicates of the ON condition can be used to filter the target table
at runtime
// only if there is no NOT MATCHED BY SOURCE clause
- val (pushableCond, groupFilterCond) = if
(notMatchedBySourceActions.isEmpty) {
- if (groupFilterEnabled) {
- (cond, Some(toGroupFilterCondition(relation, source, cond)))
- } else {
- (cond, None)
- }
+ val groupFilterCond = if (notMatchedBySourceActions.isEmpty &&
groupFilterEnabled) {
Review Comment:
**5. [Low] `ON t.pk = s.pk` still reads every group**
As [cloud-fan
noted](https://github.com/apache/spark/pull/52185#discussion_r4360904644), the
disjunction folds to `true` when ON has no target-only conjunct, which is the
most common shape. The runtime group filter is still off whenever there is a
NOT MATCHED BY SOURCE clause, so such a MERGE reads every group, as before.
When every NOT MATCHED BY SOURCE clause is conditional, the argument for the
static disjunction also holds at runtime. A target row that matches no source
row and fails every NOT MATCHED BY SOURCE condition is only carried over. So
the group filter could be:
```scala
val nmbsConds = notMatchedBySourceActions.flatMap(_.condition)
Some((toGroupFilterCondition(relation, source, cond) +:
nmbsConds).reduce(Or))
```
`RowLevelOperationRuntimeGroupFiltering` already rewrites the MERGE group
filter with `RewritePredicateSubquery`, which handles `EXISTS` under `OR`. This
could go in a separate PR. I mention it because it would cover the shape that
the static disjunction cannot.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/GroupBasedMergeIntoTableSuite.scala:
##########
@@ -218,6 +219,253 @@ class GroupBasedMergeIntoTableSuite extends
MergeIntoTableSuiteBase {
}
}
+ test("merge pushes conditional NOT MATCHED BY SOURCE predicates to target") {
+ withTempView("source") {
+ 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": "finance" }
+ |""".stripMargin)
+
+ val sourceDF = Seq(1, 3, 6).toDF("pk")
+ sourceDF.createOrReplaceTempView("source")
+
+ // every NOT MATCHED BY SOURCE clause is conditional, so the target-only
part of the ON
+ // condition and that clause's condition can be pushed to the target
table as a
+ // disjunction; a target row satisfying neither is only carried over, so
its group
+ // doesn't have to be read at all
+ val (cond, groupFilterCond) = executeAndKeepConditions {
+ sql(
+ s"""MERGE INTO $tableNameAsString t
+ |USING source s
+ |ON t.pk = s.pk AND t.dep IN ('hr', 'finance')
+ |WHEN MATCHED THEN
+ | UPDATE SET t.salary = t.salary + 1
+ |WHEN NOT MATCHED THEN
+ | INSERT (pk, salary, dep) VALUES (s.pk, 0, 'hr')
+ |WHEN NOT MATCHED BY SOURCE AND t.dep IN ('hr', 'finance') THEN
+ | DELETE
+ |""".stripMargin)
+ }
+
+ assert(cond.sql == "(t.dep IN ('hr', 'finance'))", s"unexpected pushable
condition: $cond")
Review Comment:
**3. [Nit] The `cond.sql` assertions depend on optimizer simplifications**
These assertions read `ReplaceData.condition` from the optimized plan. They
rely on how `BooleanSimplification` folds the disjunction (`D OR D` to `D`, `(1
= 0) OR D` to `D`, `true OR D` to `true`) and on how the result renders as SQL.
[cloud-fan's
review](https://github.com/apache/spark/pull/52185#pullrequestreview-5637730797)
sees the exact match as a strength, because any change to the emitted
disjuncts fails it, and I agree with that. The cost is that an unrelated
optimizer change also fails these tests, for example one that rewrites `dep =
'hr' OR dep = 'finance'` as an `IN` list.
This is optional. A short comment above each assertion that names the
simplification it relies on would help whoever has to update it.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteMergeIntoTable.scala:
##########
@@ -391,13 +404,23 @@ object RewriteMergeIntoTable extends
RewriteRowLevelCommand with PredicateHelper
targetTable: LogicalPlan,
cond: Expression): (LogicalPlan, Expression) = {
+ val (targetCond, joinCond) = splitTargetPredicates(targetTable, cond)
+ (Filter(targetCond, targetTable), joinCond)
+ }
+
+ // splits the given condition into a part that can be evaluated against the
target table alone
+ // and a part that requires the source as well, each returned as a
conjunction
+ private def splitTargetPredicates(
Review Comment:
**4. [Low] `extractPredicatesWithinOutputSet` would also cover an `OR` in
the ON condition**
`splitTargetPredicates` keeps only whole top-level conjuncts. Take an ON
condition like `(t.pk = s.pk AND t.dep = 'hr') OR (t.pk = s.pk AND t.dep =
'it')`. Its only conjunct references the source, so `targetCond` is
`TrueLiteral` and the disjunction folds to `true`.
`PredicateHelper.extractPredicatesWithinOutputSet`, which this object
already mixes in, returns the strongest target-only predicate implied by ON.
Here that is `t.dep = 'hr' OR t.dep = 'it'`. Because ON implies it, it is a
safe disjunct. A part that mixes target and source references under `NOT` is
treated as one leaf and dropped, so this also holds before the optimizer runs.
Could the group-based branch use it?
```scala
val targetCond = extractPredicatesWithinOutputSet(cond,
readRelation.outputSet)
.getOrElse(TrueLiteral)
```
The delta-based path still needs `splitTargetPredicates`, because it also
uses the remaining join condition.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteMergeIntoTable.scala:
##########
@@ -391,13 +404,23 @@ object RewriteMergeIntoTable extends
RewriteRowLevelCommand with PredicateHelper
targetTable: LogicalPlan,
cond: Expression): (LogicalPlan, Expression) = {
+ val (targetCond, joinCond) = splitTargetPredicates(targetTable, cond)
Review Comment:
**6. [Low] The delta-based path keeps the old rule**
I understand that the delta-based path is left for a separate PR. In case it
helps there, the same disjunction seems to fit without other changes.
`buildWriteDeltaPlan` could put `Filter(targetCond OR c1 OR ..., readRelation)`
under the join and keep the full ON condition as the join condition
([L296-L302](https://github.com/apache/spark/blob/f2b3c156118b115caf6a382d2d0b017719abbaf8/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteMergeIntoTable.scala#L296-L302)).
A target row that the filter drops can neither match nor fire a NOT MATCHED BY
SOURCE action, so it would produce no delta row.
Until then, the same MERGE reads only the matching groups on a group-based
connector but the whole table on a delta-based one. A shared helper that builds
the disjunction for both paths would keep them from drifting apart.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteMergeIntoTable.scala:
##########
@@ -174,16 +174,29 @@ object RewriteMergeIntoTable extends
RewriteRowLevelCommand with PredicateHelper
readRelation, joinPlan, matchedActions, notMatchedActions,
notMatchedBySourceActions, metadataAttrs, checkCardinality)
- // predicates of the ON condition can be used to filter the target table
(planning & runtime)
+ // predicates that can be used to filter the target table (planning):
+ // - without NOT MATCHED BY SOURCE clauses, target rows that don't satisfy
the ON condition
+ // match no action and are simply carried over, so the ON condition can
filter the target
+ // - if every NOT MATCHED BY SOURCE clause is conditional, a target row
can be skipped only
+ // if it satisfies neither the ON condition nor any of those conditions,
so the filter is
+ // their disjunction; only the target-only part of the ON condition can
take part, as a
+ // disjunction that references the source is not a valid filter for the
target table
+ // - an unconditional NOT MATCHED BY SOURCE clause matches any target row
and disables this
+ val pushableCond = if (notMatchedBySourceActions.isEmpty) {
Review Comment:
**7. [Nit] No conf restores the previous plan**
Before this PR, a group-based MERGE with any NOT MATCHED BY SOURCE clause
pushed nothing to the connector. Now, when those clauses are all conditional,
it pushes a disjunction, and that changes which groups are read and replaced.
If a connector evaluates such a predicate wrongly, or in the CHAR case of
item 1, users cannot get the old plan back without a code change. Would an
internal conf, enabled by default, that keeps `TrueLiteral` be worth adding?
--
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]
