HyukjinKwon commented on code in PR #59321:
URL: https://github.com/apache/spark/pull/59321#discussion_r4286238721
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala:
##########
@@ -717,7 +750,8 @@ class PlanMerger(
mergedJoin
}
Some(TryMergeResult(mergedPlan, npMapping, npFilter,
cpFilter,
- dsv2Merged = leftDsv2Merged || rightDsv2Merged))
+ dsv2Merged = leftDsv2Merged || rightDsv2Merged,
+ cachedSideFiltered = leftCPSideFiltered ||
rightCPSideFiltered))
Review Comment:
**Nit (P3):** This OR is what lets an outer Join see that the cached plans
filter a child of an inner join. Take `(A JOIN B.where(d > 1)) JOIN C` vs `(A
JOIN B) JOIN C.where(x > 0)`: the inner Join returns `cachedSideFiltered =
true` here, so the outer gate declines. If this went back to the default
`false`, that merge would happen without symmetric propagation and build the
cross pairs again. Every SPARK-60106 test, and the existing
SPARK-56677/SPARK-59897 join tests, use a single Join, so nothing would catch
it.
Multi-join subqueries (the q88/q90 shapes the tuning guide cites) are where
this matters. Could you add a small nested-join plan test next to the
SPARK-60106 pair that asserts no merge with symmetric off?
See **Shared repair plan 1** in the review body.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -8050,7 +8051,11 @@ object SQLConf {
"common join to be merged into a single scan. A filter attribute is
only propagated " +
"through a join when it originates from the non-nullable (preserved)
side: the left side " +
"of LeftOuter/LeftSemi/LeftAnti, the right side of RightOuter, or
either side of " +
- "Inner/Cross. FullOuter joins are never eligible. " +
+ "Inner/Cross. FullOuter joins are never eligible. When the two
subplans filter " +
Review Comment:
**Nit (P3):** This sentence, the symmetric doc above, and the matching
tuning-guide rows say when a through-join merge needs symmetric propagation by
naming two shapes. The gate in `PlanMerger` (the new Join-arm conjunct) applies
one rule: it declines when one subplan has a filter on a join child that the
other doesn't share, and the other subplan has such a filter on the other child.
Some shapes fit that rule but neither phrase. The gridTest's
`firstMerge`/`extended` case, `(a > 1 on V2) JOIN R` vs `(b > 2 on V2) JOIN (e
> 3 on R)`, is now declined even with `dsv2SymmetricFilterPropagation` on, and
so are two subplans that filter both DSv2 children differently. Read strictly,
neither 'filter different join children' nor 'filter both children' covers
them, so the 'even when dsv2... lets their scans merge' clause reads as if they
still merge. Read loosely, as 'filter different sets of children', the rule
would wrongly cover `(a > 1 on L) JOIN R` vs `(a > 1 on L) JOIN (e > 3 on R)`.
Those two share the left filter and still merge without symmetric.
Could the three doc sites state the rule directly? For example: through a
join, a merge also counts as symmetric when each subplan filters a join child
with a condition the other does not share, on different children, even when
dsv2SymmetricFilterPropagation lets the scans merge. The two current phrases
would then serve as examples.
**Verification:**
- **Inspection:** The reworded docs give the same verdict as the Join gate
for the pure different-children shape, the mixed shape, the both-children DSv2
shape, and the shared-filter shape.
- **Compile:** The edited SQLConf doc-string concatenation still compiles
and stays within the line-length limit.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala:
##########
@@ -513,8 +536,8 @@ class PlanMerger(
tryBuildFilterDSv2ScanChild(mergedChild, deferred,
Some(cp.condition))
.map { prunedChild =>
val mergedPlan = Filter(cp.condition, prunedChild)
- TryMergeResult(
- mergedPlan, npMapping, npFilter, cpFilter, dsv2Merged =
dsv2Merged)
+ TryMergeResult(mergedPlan, npMapping, npFilter, cpFilter,
+ dsv2Merged = dsv2Merged, cachedSideFiltered =
cpSideFiltered)
Review Comment:
**Nit (P3):** Forwarding only the child's flag here is what keeps a filter
both plans share from counting as cached-side filtering. Take `A.where(a > 1)
JOIN B` vs `A.where(a > 1) JOIN B.where(d > 1)` with throughJoin on and
symmetric off: they should still merge, because the first plan's pairs cover
the second's. No test has this shape. The DSv2 `reused` grid case goes through
the tagged merged-Filter branch instead, so if this line reported `true`, as
its sibling first-time branch does, the merge would be declined and no
assertion would fail.
Could you add a plan test for this shared-filter shape that asserts the
merge?
See **Shared repair plan 1** in the review body.
--
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]
