peter-toth commented on code in PR #59161:
URL: https://github.com/apache/spark/pull/59161#discussion_r4187246168
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala:
##########
@@ -690,19 +684,43 @@ class PlanMerger(
// rows are NULL-padded so f=NULL, causing FILTER
(WHERE f) to incorrectly
// exclude rows that should contribute to the
aggregate. Right-side
// attributes are also absent from semi/anti join
output.
- (leftNPFilter.isEmpty && leftCPFilter.isEmpty ||
+ (leftNPFilter.isEmpty && leftCPFilter.isEmpty ||
filterSafeForJoin(fromLeft = true, cp.joinType)) &&
(rightNPFilter.isEmpty && rightCPFilter.isEmpty ||
filterSafeForJoin(fromLeft = false, cp.joinType)) =>
val npMapping = leftNPMapping ++ rightNPMapping
val mappedNPCondition = np.condition.map(mapAttributes(_,
npMapping))
// Comparing the canonicalized form is required to ignore
different forms of the
// same expression and `AttributeReference.qualifier`s in
`cp.condition`.
- if (mappedNPCondition.map(_.canonicalized) ==
cp.condition.map(_.canonicalized)) {
- val npFilter = leftNPFilter.orElse(rightNPFilter)
- val cpFilter = leftCPFilter.orElse(rightCPFilter)
- Some(TryMergeResult(cp.withNewChildren(Seq(mergedLeft,
mergedRight)), npMapping,
- npFilter, cpFilter, dsv2Merged = leftDsv2Merged ||
rightDsv2Merged))
+ if (mappedNPCondition.map(_.canonicalized) ==
+ cp.condition.map(_.canonicalized)) {
+ val mergedJoin = cp.withNewChildren(Seq(mergedLeft,
mergedRight))
+ val (npFilter, npAlias) = (leftNPFilter, rightNPFilter)
match {
+ case (Some((left, _)), Some((right, _))) =>
+ val alias = Alias(
+ And(left, right),
s"propagatedFilter_${PlanMerger.newId}")()
Review Comment:
**Finding 1.** Folding `left` and `right` into this `And` means neither is
carried to the `Project`s above the join. A later round can still reuse one of
them on its own. Then the `Aggregate`'s `FILTER` references an attribute its
child doesn't output.
The reuse comes from the tagged `(Filter, Filter)` branch
(`existingNPFilter`, line 546). It returns the attribute with `isNew = false`,
so `mergeNamedExpressions` doesn't append it (line 1125). That relies on the
round that created the attribute having carried it all the way up to the
`Aggregate`. Once it went into this `And`, it wasn't.
A shape that hits it, with the confs of `Repeatedly merge filters from both
Join children` (symmetric and through-join propagation on):
```sql
SELECT
(SELECT sum(l.a) FROM testData2 l JOIN testData2 r ON l.a = r.a WHERE l.b
= 1),
(SELECT max(l.a) FROM testData2 l JOIN testData2 r ON l.a = r.a WHERE l.b
= 2 AND r.b = 2),
(SELECT count(*) FROM testData2 l JOIN testData2 r ON l.a = r.a WHERE l.b
= 2)
```
1. The second subquery merges into the first. The left child becomes a
tagged `Filter` holding `f_l = (b = 2)`. Only `And(f_l, f_r)` is carried into
the `Project` that `ColumnPruning` put between the join and the `Aggregate`,
and `f_l` itself is not.
2. The third subquery reuses `f_l` with `isNew = false`. Its `count(*)
FILTER (WHERE f_l)` then sits above that `Project`, which doesn't output `f_l`.
The expected answer is `Row(12, 3, 6)`. On master the second subquery
doesn't merge, so the third one merges with the first and the query works.
As far as I can trace, a post-join `Filter` can already fold a child's
attribute this way on master (`newNPCondition = And(f, ...)`). So I'd fix it in
`mergeNamedExpressions` rather than here, by appending the np filter whenever
the merged list doesn't output it yet:
```scala
newPlanFilter.foreach { case (f, _) =>
if (!mergedExpressions.exists(_.exprId == f.exprId)) mergedExpressions
+= f
}
```
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/MergeSubplansSuite.scala:
##########
@@ -1859,10 +1859,7 @@ class MergeSubplansSuite extends PlanTest {
}
}
- test("SPARK-56677: Do not merge subqueries when both join children have
independent filters") {
- // np has filters on BOTH left and right join children simultaneously. The
guard in the
- // Join case prevents this merge because combining two independent filter
attributes would
- // require ANDing them into a new alias, which is not yet supported.
+ test("SPARK-56677: Merge subqueries when both join children have independent
filters") {
Review Comment:
**Finding 3.** This renamed test and the four new ones (two below, two in
`PlanMergingSuite`) test the SPARK-59897 change. They'd normally carry
`SPARK-59897:` rather than `SPARK-56677:`.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala:
##########
@@ -115,9 +115,7 @@ object PlanMerger {
* result of expressions like `coalesce(col, default)` in the aggregate: an
originally unmatched row
* would have contributed `default` via `coalesce(NULL, default)`, but in the
merged plan it is
* matched, its real column value fails the filter, and `FILTER (WHERE false)`
discards it entirely.
- * Propagation is also skipped when both the left and right children
simultaneously produce filter
- * attributes, as combining them would require an additional AND alias above
the join (not yet
- * supported).
+ * Filters from both join children are combined with AND above the join.
Review Comment:
**Finding 4.** The `tryMergePlans` scaladoc (lines 399-400) still says the
join arm forwards propagation "so a filter difference on one child can still be
merged". Worth saying "either or both children" there too.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala:
##########
@@ -676,11 +674,7 @@ class PlanMerger(
tryMergePlans(np.right, cp.right, context.copy(filterAboveScan =
false)).flatMap {
case TryMergeResult(mergedRight, rightNPMapping,
rightNPFilter, rightCPFilter, _,
rightDsv2Merged)
- // If both children independently propagate filter
attributes we would need to
- // AND them into a new alias above the join, which is not
yet supported.
- if !(leftNPFilter.isDefined && rightNPFilter.isDefined) &&
- !(leftCPFilter.isDefined && rightCPFilter.isDefined) &&
- // Gate join-crossing filter propagation behind its own
config flag.
+ if // Gate join-crossing filter propagation behind its own
config flag.
Review Comment:
**Finding 2.** Removing this guard leaves `filterSafeForJoin` as the only
check that keeps a filter pair off an outer join. For a `LeftOuter` join it's
the right-side case that rejects the pair, and no test covers it. The existing
negative tests use `RightOuter`, `FullOuter` and the right side of `LeftSemi`.
A test next to the `RightOuter` one would pin it:
```scala
test("SPARK-59897: Do not merge subqueries when both children of a
LeftOuter join have filters") {
val subquery1 = ScalarSubquery(
testRelation.join(testRelation2, LeftOuter, Some($"a" === $"d"))
.groupBy()(sum($"a").as("sum_a")))
val subquery2 = ScalarSubquery(
testRelation.where($"a" > 1).join(testRelation2.where($"d" > 1),
LeftOuter,
Some($"a" === $"d"))
.groupBy()(max($"a").as("max_a")))
val originalQuery = testRelation.select(subquery1, subquery2)
withSQLConf(SQLConf.MERGE_SUBPLANS_FILTER_PROPAGATION_THROUGH_JOIN_ENABLED.key
-> "true") {
comparePlans(Optimize.execute(originalQuery.analyze),
originalQuery.analyze)
}
}
```
--
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]
