peter-toth opened a new pull request, #59321:
URL: https://github.com/apache/spark/pull/59321
### What changes were proposed in this pull request?
`PlanMerger` now treats a merge through a join as symmetric filter
propagation when the two plans filter different children of the join. Such a
merge then needs
`spark.sql.optimizer.mergeSubplans.filterPropagation.symmetricFilterPropagation.enabled`,
like the other merges where both plans have a differing filter.
A cached join child also counts as filtered by the cached plans when it
holds a merged `Filter` from an earlier round. The merge doesn't report that
filter, because the cached aggregates already carry it.
The docs of the symmetric and the through-join configs mention the new case.
### Why are the changes needed?
When one plan filters one join child and the other plan the other child, the
merged join reads both children unfiltered. It also builds the pairs that
neither plan needs. The answer is right, but the join work can grow a lot:
```scala
spark.range(20000).selectExpr("id", "coalesce(id % 100, 0L) AS k", "id AS
v").createOrReplaceTempView("l")
spark.range(20000).selectExpr("id", "coalesce(id % 100, 0L) AS k", "id AS
v").createOrReplaceTempView("r")
sql("""
SELECT
(SELECT sum(l.v) FROM l JOIN r ON l.k = r.k WHERE l.id < 100),
(SELECT max(r.v) FROM l JOIN r ON l.k = r.k WHERE r.id < 100)
""")
```
With
`spark.sql.optimizer.mergeSubplans.filterPropagation.throughJoin.enabled=true`,
the two subqueries merge into one join that produces 4,000,000 rows. Without
the merge, the two joins produce 20,000 rows each.
If each plan's filter keeps a share p and q of its side, the merged join
does about 1/(p+q) times the work of the two original joins. So the merge only
pays off for filters that keep most of the data, and `PlanMerger` can't see
that. Both plans have a differing filter in this case. The `Filter` arms
already treat that as symmetric propagation, e.g. `(np: Filter, cp)` needs the
config when cp brings a filter from below. The join arm was the one place that
combined the two plans' filters without that check.
The through-join docs recommend turning on the symmetric config too, because
the `isnotnull` filters inferred for a nullable join key make most of these
merges symmetric anyway. With both configs on, this check doesn't apply and the
merge happens as before.
#59161 changes the same guard. Its tests keep passing with this change.
### Does this PR introduce _any_ user-facing change?
Yes. With the through-join config on and the symmetric config off,
subqueries that filter different children of the same join are no longer
merged. With both configs on they merge as before. Both configs are off by
default.
### How was this patch tested?
New tests in `MergeSubplansSuite` (no merge without the symmetric config,
merge with it, and a merged `Filter` from DSv2 scans on either join child) and
`PlanMergingSuite` (end to end). The tests that expect no merge fail without
the fix.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 5.5)
--
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]
