dongjoon-hyun commented on code in PR #59321:
URL: https://github.com/apache/spark/pull/59321#discussion_r4294718882
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala:
##########
@@ -319,14 +325,26 @@ class PlanMerger(
* pair that is NOT the innermost one still recognize the
merge below it and
* apply the DSv2-symmetric exemption. It only ever gates
behaviour when
* `dsv2SymmetricFilterPropagationEnabled` is on.
+ * @param cachedSideFiltered Whether every plan merged into the cached plan
filters this subtree
+ * with a filter that the new plan doesn't share.
It is set whenever
+ * `cachedPlanFilter` is. It is also set when a
merged Filter from an
+ * earlier round filters the subtree, unless the
new plan reuses one of
+ * its sides. That Filter's condition is not in
+ * `cachedPlanFilter`, because the cached
aggregates already carry it.
+ * The fact goes up to the nearest Aggregate,
which consumes the filters
+ * below it, so a Join arm can tell which of its
children the cached
+ * plans filter. The merged Filter case only
matters without
+ * `symmetricFilterPropagationEnabled`, where such
a Filter can only
+ * come from the DSv2-symmetric exemption.
*/
case class TryMergeResult(
mergedPlan: LogicalPlan,
newPlanMapping: AttributeMap[Attribute],
newPlanFilter: Option[Attribute] = None,
cachedPlanFilter: Option[Attribute] = None,
dsv2DeferredScan: Option[DSv2DeferredScan] = None,
- dsv2Merged: Boolean = false)
+ dsv2Merged: Boolean = false,
+ cachedSideFiltered: Boolean = false)
Review Comment:
**This flag defaults to `false`, which is the unsafe direction.** `false`
lets the Join arm at L704-706 allow a merge that builds cross pairs, and
nothing fails when that happens, because the flag only ever declines merges.
This PR already had to forward it by hand in the identical-condition Filter
branch (L539-540) and in the `(np: Filter, cp)` arm (L662-663). A future arm,
or a refactor that rebuilds a `TryMergeResult` with named arguments, would
compile without it and silently bring back the 1/(p+q) join work. Could you
drop the default, so every construction site has to pass it explicitly? The
sites that want `false` today (`identical` at L428, the two Aggregate results,
and the DSv2 leaf) would then each say so once.
##########
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 " +
+ "different children of the join, the merge also needs " +
Review Comment:
**`spark.sql.optimizer.mergeSubplans.filterPropagation.throughJoin.enabled`
shipped in Apache Spark 4.2.0, and this PR changes what it does.** A user who
enabled it without the symmetric config (the 4.2.0 doc only says the two are
"usually" set together) will see subqueries that filter different join children
stop merging after upgrading, with nothing in the release notes to explain the
plan change. `docs/sql-migration-guide.md` already has a "Since Spark 4.3"
entry for `MergeSubplans`. Could you add one for this change too, naming
`spark.sql.optimizer.mergeSubplans.filterPropagation.symmetricFilterPropagation.enabled`
as the config that brings these merges back?
##########
docs/sql-performance-tuning.md:
##########
@@ -354,7 +354,7 @@ SELECT
(SELECT avg(ss_net_paid) FROM store_sales WHERE ss_quantity BETWEEN 21 AND
40)
```
-In TPC-DS benchmark runs, enabling symmetric filter propagation made `q9` and
`q28` about 3.5x faster, and enabling it together with propagation through
joins made `q88` about 7x and `q90` about 2x faster. See
[SPARK-40193](https://issues.apache.org/jira/browse/SPARK-40193) and
[SPARK-56677](https://issues.apache.org/jira/browse/SPARK-56677) for these
measurements. The trade-off depends on the tables: the gain is largest when the
differing filters are on columns the data source cannot prune on, and the risk
is highest on heavily partitioned or file-pruned tables where the widened
filter loses that pruning. Validate it on your own workload before enabling it
in production.
+In TPC-DS benchmark runs, enabling symmetric filter propagation made `q9` and
`q28` about 3.5x faster, and enabling it together with propagation through
joins made `q88` about 7x and `q90` about 2x faster. See
[SPARK-40193](https://issues.apache.org/jira/browse/SPARK-40193) and
[SPARK-56677](https://issues.apache.org/jira/browse/SPARK-56677) for these
measurements. The trade-off depends on the tables: the gain is largest when the
differing filters are on columns the data source cannot prune on, and the risk
is highest on heavily partitioned or file-pruned tables where the widened
filter loses that pruning. With propagation through joins, two subplans that
filter different join children also merge, and so do two that filter both join
children with different conditions. The merged join then builds join pairs that
neither subplan needs, which can multiply its work. Validate it on your own
workload before enabling it in production.
Review Comment:
Minor: "With propagation through joins, two subplans that filter different
join children also merge" reads as if the through-join config alone is enough.
After this PR they merge only when symmetric propagation is on as well. Since
the paragraph starts from the two configs enabled together, could this sentence
say so explicitly, e.g. "With both configs on, two subplans that filter
different join children also merge, ..."?
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/MergeSubplansSuite.scala:
##########
@@ -2597,6 +2653,70 @@ class MergeSubplansSuite extends PlanTest {
}
}
+ gridTest("SPARK-60106: Merge over a merged DSv2 filter on a join child")(
+ for {
+ v2OnLeft <- Seq(true, false)
+ shape <- Seq("dropped", "extended", "reused", "firstMerge")
+ } yield (v2OnLeft, shape)) { case (v2OnLeft, shape) =>
+ // sub1 and sub2 filter the DSv2 scan differently and merge under the DSv2
exemption into a
+ // merged Filter. A subplan that also filters the other join child would
join a child that the
+ // cached plans filter with a child that it filters, so without symmetric
filter propagation
+ // it stays apart. The shapes:
+ // - dropped: sub3 filters only the other child, so its merge drops the
merged Filter.
+ // - extended: sub3 filters the DSv2 scan too, so its merge extends the
merged Filter.
+ // - reused: sub3 repeats sub1's filter on the DSv2 scan, so sub1 covers
sub3's rows there and
+ // the merge builds no extra pairs. It merges without symmetric filter
propagation.
+ // - firstMerge: sub2 itself filters the other child too, so the first
merge declines.
+ // The DSv2 scan is tried as both the left and the right join child.
+ def join(v2: LogicalPlan, other: LogicalPlan): LogicalPlan = if (v2OnLeft)
{
+ v2.join(other, Inner, Some($"a" === $"d"))
+ } else {
+ other.join(v2, Inner, Some($"a" === $"d"))
+ }
+ val sub1 = ScalarSubquery(
+ join(v2ScanReading("a", "b").where($"a" > 1), testRelation2)
+ .groupBy()(sum($"a").as("sum_a")))
+ val other2 = if (shape == "firstMerge") testRelation2.where($"e" > 3) else
testRelation2
+ val sub2 = ScalarSubquery(
+ join(v2ScanReading("a", "b").where($"b" > 2), other2)
+ .groupBy()(max($"a").as("max_a")))
+ val v2In3 = shape match {
+ case "extended" => v2ScanReading("a", "b").where($"a" < 0)
+ case "reused" => v2ScanReading("a", "b").where($"a" > 1)
+ case _ => v2ScanReading("a", "b")
+ }
+ val sub3 = ScalarSubquery(
+ join(v2In3, testRelation2.where($"e" > 3))
+ .groupBy()(min($"a").as("min_a")))
+ val originalQuery = if (shape == "firstMerge") {
+ testRelation.select(sub1, sub2)
+ } else {
+ testRelation.select(sub1, sub2, sub3)
+ }
+
+ def mergedAggregates(symmetric: Boolean): Seq[Seq[String]] = withSQLConf(
Review Comment:
Minor: `mergedAggregates` only compares aggregate names. In the `reused`
case, which merges without symmetric propagation, nothing checks that `min_a`
gets the reused filter attribute together with the `e > 3` alias. If either
went wrong, the names would still be `sum_a, max_a, min_a`, the test would
pass, and the merged query would return wrong results. The only `checkAnswer`
test (`PlanMergingSuite`) covers the plain V1 shape. Could the `reused` case
also compare the merged plan, or at least check each aggregate's FILTER?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala:
##########
@@ -660,18 +688,23 @@ class PlanMerger(
mergedChild.output.toList :+ newCPFilterAlias,
mergedChild)
TryMergeResult(project, npMapping, npFilter, Some(newCPFilter),
- dsv2Merged = dsv2Merged)
+ dsv2Merged = dsv2Merged, cachedSideFiltered = true)
}
}
case (np: Join, cp: Join) if np.joinType == cp.joinType && np.hint ==
cp.hint =>
tryMergePlans(np.left, cp.left, context.copy(filterAboveScan =
false)).flatMap {
case TryMergeResult(mergedLeft, leftNPMapping, leftNPFilter,
leftCPFilter, _,
- leftDsv2Merged) =>
+ leftDsv2Merged, leftCPSideFiltered) =>
tryMergePlans(np.right, cp.right, context.copy(filterAboveScan =
false)).flatMap {
case TryMergeResult(mergedRight, rightNPMapping,
rightNPFilter, rightCPFilter, _,
- rightDsv2Merged)
- if // Gate join-crossing filter propagation behind its own
config flag.
+ rightDsv2Merged, rightCPSideFiltered)
+ if // When the two plans filter different children, the
merge is symmetric
+ // filter propagation and needs its config, see the
class doc.
+ (symmetricFilterPropagationEnabled ||
Review Comment:
Nit: The new check is a fifth clause in a guard that already spans about 20
lines, and it relies on `&&` binding tighter than `||` inside the negation.
Could you pull it out into a named value, e.g. `val crossFiltered =
leftNPFilter.isDefined && rightCPSideFiltered || rightNPFilter.isDefined &&
leftCPSideFiltered`, and test `symmetricFilterPropagationEnabled ||
!crossFiltered`? Moving the guard into the case body with a final `else None`
keeps the same behavior.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala:
##########
@@ -430,27 +448,30 @@ class PlanMerger(
(newPlan, cachedPlan) match {
case (np: Project, cp: Project) =>
tryMergePlans(np.child, cp.child, context).map {
- case TryMergeResult(mergedChild, npMapping, npFilter, cpFilter,
deferred, dsv2Merged) =>
+ case TryMergeResult(mergedChild, npMapping, npFilter, cpFilter,
deferred, dsv2Merged,
Review Comment:
Nit: Adding one field meant editing every positional `case
TryMergeResult(...)` extractor, about 10 of them. With seven positional fields,
including two `Option[Attribute]` and two `Boolean`, swapping `dsv2Merged` and
`cachedSideFiltered` in one of them would still compile. The three Project arms
only change `mergedPlan` and `newPlanMapping`. Could they bind the result as
`r` and return `r.copy(mergedPlan = Project(mergedProjectList, r.mergedPlan),
newPlanMapping = newNPMapping)`?
--
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]
