dongjoon-hyun commented on code in PR #59321:
URL: https://github.com/apache/spark/pull/59321#discussion_r4298937919
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala:
##########
@@ -660,18 +687,24 @@ 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 each plan filters a different join child with a
condition that the
+ // other doesn't share, the merge is symmetric filter
propagation and needs
+ // its config, see the class doc.
+ (symmetricFilterPropagationEnabled ||
+ !((leftNPFilter.isDefined && rightCPSideFiltered) ||
Review Comment:
**This decides "a filter the other doesn't share" by comparing whole
conditions per child, without the join condition, so it also declines merges
that build no unneeded pairs.** Here are two realistic cases. Both assume
through-join and `dsv2SymmetricFilterPropagation` on, symmetric off, and
SCAN_MERGING DSv2 scans whose data filters stay post-scan (e.g. Parquet V2 with
`parquet` removed from `spark.sql.sources.useV1SourceList`):
- **Join-key filters.** Take `(SELECT sum(l.v) FROM l JOIN r ON l.k = r.k
WHERE l.k > 10)` and `(SELECT max(l.v) FROM l JOIN r ON l.k = r.k WHERE l.x >
5)`. `InferFiltersFromConstraints` copies `k > 10` onto `r`, so the cached
plans look like they filter `r`, and this line declines. But a merged pair
whose `l` row has `l.x <= 5` has `l.k > 10`, so it can only join `r` rows with
`r.k > 10`, which sub1 needs. No unneeded pair is built.
- **Implied filters.** With a nullable `k`, an unfiltered subquery carries
`isnotnull(k)` on both children, and `(... WHERE l.x > 1 AND r.y > 3)` carries
`isnotnull(k) AND ...` on both. Both child pairs take the first-time branch
(L639), so the merge is declined. Yet `OR(isnotnull(k) AND x > 1,
isnotnull(k))` is just `isnotnull(k)`, and the merged join is the first
subquery's own join.
Both shapes merge at the merge base (SPARK-59897) and didn't merge in 4.3.
So this isn't a regression against a release, but it gives up part of what
SPARK-59897 added. The tests don't show it because `MergeSubplansSuite` runs
only `MergeSubplans`. For example, the gridTest's `a > 1` with `a = d` has no
inferred `d > 1`.
Could the docs and the class doc say the merged join *can* build pairs that
neither subplan needs, instead of *builds*? Deciding by conjunct containment
would be the more general fix, but that may be out of scope here.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala:
##########
@@ -660,18 +687,24 @@ 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 each plan filters a different join child with a
condition that the
+ // other doesn't share, the merge is symmetric filter
propagation and needs
+ // its config, see the class doc.
+ (symmetricFilterPropagationEnabled ||
Review Comment:
**With this check, the result depends on the order of the subqueries, and in
one order it runs one more subquery than 4.3.** `merge()` takes the first cache
entry that merges and never revisits earlier entries (L183). Take through-join
on, symmetric off and non-nullable join keys, as in the PR description and the
e2e test. Then add `(SELECT count(*) FROM l JOIN r ON l.k = r.k)` as the third
item of the PR description's query:
1. sub2 is declined against entry 0 and becomes entry 1.
2. sub3 merges into entry 0. The untagged `(np, cp: Filter)` arm turns
sub1's filter into an alias, so entry 0 now joins all of `l x r`, 4,000,000
rows.
3. Entry 1 is never retried, so `max(r.v)` still runs its own join.
4.3 and the merge base run a single 4,000,000-row join. With `count(*)`
listed first or second, this PR does too.
Plan-level repro: append `testRelation.join(testRelation2, Inner, Some($"a"
=== $"d")).groupBy()(min($"a"))` as a third subquery in the first SPARK-60106
test. You get one CTE with `sum_a` and `min_a`, plus a separate `max_a`.
The greedy order dependence is older, but this regression comes from the new
check. Could a test or the migration guide at least pin this behavior?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala:
##########
@@ -574,8 +596,12 @@ class PlanMerger(
projectList = newProjectList, child = prunedChild)
val newFilter = Filter(newCondition, newProject)
newFilter.copyTagsFrom(cp)
+ // The cached plans' side of the OR stays hidden in
cp's aggregates. A
+ // reused np filter is one of the OR's sides, so one
cached plan shares
+ // it, and the cached plans don't filter this
subtree against np.
TryMergeResult(newFilter, npMapping,
Some(newNPFilterOut), None,
Review Comment:
**This isn't introduced by this PR, but it is a wrong-result bug in the
branches this PR touches.** Both merged-Filter branches return
`cachedPlanFilter = None`, here (extend/reuse) and at L679 (drop). So they
throw away a `cpFilter` that the child merge produced. That happens when a
Filter that all cached plans share sits below the merged Filter and the new
plan lacks it: the untagged `(np, cp: Filter)` arm turns that Filter into an
alias. Correctness then depends on `mergeNamedExpressions` rewriting each OR
side as `If(cpFilter, side, null)`, but it skips aliases whose child is an
`Attribute` (L1148) and cached expressions that a new-plan expression matched
(`matchedCachedIndices`).
For example, with symmetric propagation on and `rel(a int, fa boolean, fb
boolean)`:
- sub1 = `Filter(fa, Filter(a > 0, rel))` with `sum(a)`
- sub2 = `Filter(fb, Filter(a > 0, rel))` with `max(a)`
- sub3 = `rel` with `count(a)`
Merging sub2 builds a merged `Filter(pf0 OR pf1)` over `Project[..., fb AS
pf0, fa AS pf1](Filter(a > 0, rel))`. Merging sub3 turns `Filter(a > 0)` into a
`cpFilter`, L679 drops it, and `pf0`/`pf1` stay unwrapped. With the rows `(-5,
true, true)` and `(3, true, false)`, sub1 returns -2 instead of 3, and sub2
returns -5 instead of NULL. The same happens through this line when a new
plan's Project computes one of the OR sides. For example, sub1 and sub2 filter
`c = 'x'` and then `a > 1` / `b > 2`, and sub3 is `T.select(a, b, (b >
2).as("flag")).where(a < 0)`. Then `pf0` (`b > 2`) is matched, not wrapped, and
`max_a FILTER pf0` counts rows with `c != 'x'`.
I could build these shapes only in the `MergeSubplansSuite` harness. With
the default rules (`CombineFilters`, `PushDownPredicates`,
`InferFiltersFromConstraints`) I couldn't find a SQL query that reaches them.
The same logic is in 4.2.0 and later (SPARK-40193, SPARK-56570), so it
shouldn't block this PR and could be tracked in a separate JIRA. Note that L604
now reads the child's `cpSideFiltered` while still dropping its `cpFilter`.
##########
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 both configurations on, two subplans also merge
when one filters a join child with a condition that the other doesn't share,
and the other does so on the other join child. For example, one filters only
the left child and the other only the right child, or both filter both children
with different conditions. The merged join then builds join pairs that neither
subplan needs, wh
ich can multiply its work. Validate it on your own workload before enabling it
in production.
Review Comment:
Nit: Two of the new sentences claim more than the code does.
- Here, "With both configurations on, two subplans also merge when ..."
holds only for inner and cross joins. For outer, semi and anti joins, this
shape always puts a filter on a null-supplying or dropped side, which
`filterSafeForJoin` rejects under any configs. For example, `L.where(a > 1)
LEFT JOIN R` and `L LEFT JOIN R.where(d > 1)` never merge. Saying "two subplans
of an inner or cross join" would fix it.
- In the through-join row (L381), "Such a merge also needs
`...symmetricFilterPropagation.enabled`" also covers differing filters on the
same join child. But `dsv2SymmetricFilterPropagation` is enough there
(`PlanMerger` L547-549), and the gridTest relies on it for sub1 and sub2. The
next sentence only implies the exception. Could the row say so explicitly, e.g.
"unless `...dsv2SymmetricFilterPropagation.enabled` lets DataSource V2 scans
merge, which doesn't apply when ..."?
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/MergeSubplansSuite.scala:
##########
@@ -2121,6 +2121,125 @@ class MergeSubplansSuite extends PlanTest {
}
}
+ test("SPARK-60106: Do not merge subqueries that filter different join
children") {
+ // subquery1 filters the right child and subquery2 the left child. The
merged join would read
+ // both children unfiltered and build pairs neither subquery needs, so
without symmetric filter
+ // propagation the subqueries stay separate.
+ val subquery1 = ScalarSubquery(
+ testRelation.join(testRelation2.where($"d" > 1), Inner, Some($"a" ===
$"d"))
+ .groupBy()(sum($"a").as("sum_a")))
+ val subquery2 = ScalarSubquery(
+ testRelation.where($"a" > 1).join(testRelation2, Inner, 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)
+ }
+ }
+
+ test("SPARK-60106: Merge subqueries that filter different join children when
symmetric") {
+ val subquery1 = ScalarSubquery(
+ testRelation.join(testRelation2.where($"d" > 1), Inner, Some($"a" ===
$"d"))
+ .groupBy()(sum($"a").as("sum_a")))
+ val subquery2 = ScalarSubquery(
+ testRelation.where($"a" > 1).join(testRelation2, Inner, Some($"a" ===
$"d"))
+ .groupBy()(max($"a").as("max_a")))
+ val originalQuery = testRelation.select(subquery1, subquery2)
+
+ val f0Alias = Alias($"a" > 1, "propagatedFilter_0")()
+ val f0 = f0Alias.toAttribute
+ val f1Alias = Alias($"d" > 1, "propagatedFilter_1")()
+ val f1 = f1Alias.toAttribute
+ val mergedSubquery = testRelation
+ .select(testRelation.output ++ Seq(f0Alias): _*)
+ .join(
+ testRelation2.select(testRelation2.output ++ Seq(f1Alias): _*),
+ Inner, Some($"a" === $"d"))
+ .groupBy()(
+ sum($"a", Some(f1)).as("sum_a"),
+ max($"a", Some(f0)).as("max_a"))
+ .select(CreateNamedStruct(Seq(
+ Literal("sum_a"), $"sum_a",
+ Literal("max_a"), $"max_a"
+ )).as("mergedValue"))
+ val analyzedMergedSubquery = mergedSubquery.analyze
+ val correctAnswer = WithCTE(
+ testRelation.select(
+ extractorExpression(0, analyzedMergedSubquery.output, 0),
+ extractorExpression(0, analyzedMergedSubquery.output, 1)),
+ Seq(definitionNode(analyzedMergedSubquery, 0)))
+
+ withSQLConf(
+ SQLConf.MERGE_SUBPLANS_SYMMETRIC_FILTER_PROPAGATION_ENABLED.key ->
"true",
+ SQLConf.MERGE_SUBPLANS_FILTER_PROPAGATION_THROUGH_JOIN_ENABLED.key ->
"true") {
+ comparePlans(Optimize.execute(originalQuery.analyze),
correctAnswer.analyze)
+ }
+ }
+
+ test("SPARK-60106: Do not merge subqueries that filter different children of
nested joins") {
+ // subquery1 filters a child of the inner join and subquery2 a child of
the outer join. The
+ // outer join has to know that the cached plans filter its left child.
+ val testRelation3 = LocalRelation($"g".int, $"h".int)
+ val subquery1 = ScalarSubquery(
+ testRelation.join(testRelation2.where($"d" > 1), Inner, Some($"a" ===
$"d"))
+ .join(testRelation3, Inner, Some($"a" === $"g"))
+ .groupBy()(sum($"a").as("sum_a")))
+ val subquery2 = ScalarSubquery(
+ testRelation.join(testRelation2, Inner, Some($"a" === $"d"))
+ .join(testRelation3.where($"h" > 1), Inner, Some($"a" === $"g"))
+ .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)
+ }
+ withSQLConf(
+ SQLConf.MERGE_SUBPLANS_SYMMETRIC_FILTER_PROPAGATION_ENABLED.key ->
"true",
+ SQLConf.MERGE_SUBPLANS_FILTER_PROPAGATION_THROUGH_JOIN_ENABLED.key ->
"true") {
+
assert(Optimize.execute(originalQuery.analyze).exists(_.isInstanceOf[CTERelationDef]))
Review Comment:
Nit: This half only checks that some `CTERelationDef` exists, so it doesn't
pin which aggregate gets which FILTER. It's also the only test of filter
propagation through nested joins. Suppose the outer Join paired `d > 1` and `h
> 1` with the wrong aggregates, e.g. `sum_a FILTER (h > 1)`. The answer would
be wrong, and this assertion would still pass.
Could it compare against an expected plan, like the flat test at L2141?
Within this test, the declined run above consumes `PlanMerger.newId`, so the
aliases would be `propagatedFilter_2`/`_3`. A separate test avoids that.
##########
docs/sql-migration-guide.md:
##########
@@ -37,6 +37,7 @@ license: |
- Since Spark 4.4, when an already-analyzed Data Source V2 query is refreshed,
if rebinding an expanded struct used as a map key causes distinct current keys
to collide under the analyzed schema, the query fails with `DUPLICATED_MAP_KEY`
by default instead of returning duplicate keys. With
`spark.sql.mapKeyDedupPolicy=LAST_WIN`, the last value is kept.
- Since Spark 4.4, the Oracle JDBC datasource maps bare `NUMBER` columns
(declared without explicit precision or scale) to `DecimalType(38, 18)` instead
of `DecimalType(38, 10)`. This uses `DecimalType.DEFAULT_SCALE` and preserves
up to 18 fractional digits instead of 10, reducing silent precision loss.
However, the integer range shrinks from 28 to 20 digits; bare `NUMBER` values
with more than 20 integer digits that previously read correctly will raise
`NUMERIC_VALUE_OUT_OF_RANGE`. The same change applies to Oracle `FLOAT`
columns, which JDBC reports as `NUMERIC` with `scale = -127`. Columns with
explicit `NUMBER(p,s)` are not affected. To restore the previous
`DecimalType(38, 10)` mapping, set
`spark.sql.legacy.oracle.numberMapping.enabled` to `true`.
- Since Spark 4.4, the optimizer rule `SimplifyCaseConversionExpressions` no
longer simplifies mixed case conversions `LOWER(UPPER(str))` to `LOWER(str)` or
`UPPER(LOWER(str))` to `UPPER(str)`. This preserves standard Unicode case
mapping semantics for non-ASCII characters such as Latin small letter dotless i
(`ı` U+0131) and micro sign (`µ` U+00B5).
+- Since Spark 4.4,
`spark.sql.optimizer.mergeSubplans.filterPropagation.throughJoin.enabled` alone
no longer merges subplans that filter different children of a join. More
precisely, when one subplan filters a join child with a condition that the
other doesn't share, and the other subplan does so on the other join child, the
merge also needs
`spark.sql.optimizer.mergeSubplans.filterPropagation.symmetricFilterPropagation.enabled`.
Such a merged join builds join pairs that neither subplan needs, which can
multiply its work. Query results don't change. To merge these subplans again,
set
`spark.sql.optimizer.mergeSubplans.filterPropagation.symmetricFilterPropagation.enabled`
to `true`. This also enables the other symmetric merges described in [Merging
Subplans](sql-performance-tuning.html#merging-subplans).
Review Comment:
Minor: The Join check deliberately doesn't take the DSv2 exemption that the
`(Filter, Filter)` arm has, but only the through-join docs say so ("even when
... lets their scans merge"). This entry opens with "`...throughJoin.enabled`
alone". The `dsv2SymmetricFilterPropagation` doc and tuning row still say DSv2
scans merge "even when ... is false". So a 4.3 user who runs through-join with
`dsv2SymmetricFilterPropagation` and without symmetric may not see this entry
as theirs.
For example, with non-nullable keys, `F.where(a > 1) JOIN D`, `F.where(b >
2) JOIN D` and `F JOIN D.where(e > 3)` merged into one subquery in 4.3 (the
gridTest's `dropped` shape). After this PR the third stays separate.
Could this entry add "even when
`spark.sql.optimizer.mergeSubplans.filterPropagation.dsv2SymmetricFilterPropagation.enabled`
lets their scans merge"? The class doc's "like the other such cases"
(`PlanMerger` L122-123) and the guard comment (L701-703) could also say that
the DSv2 exemption doesn't apply here. Then nobody adds it later to match the
`(Filter, Filter)` arm.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -8050,7 +8053,12 @@ 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 one subplan
filters a join " +
Review Comment:
Nit: The rule is stated for two subplans, but with three or more, the new
subplan is checked against the merged ones. They count as filtering a join
child only if every one of them filters it with a condition the new one doesn't
share, and a reused OR side doesn't count (`PlanMerger` L604). So a pair that
the docs say needs symmetric propagation can merge without it, depending on the
order.
In the gridTest's `reused` case (symmetric off), sub2 and sub3 are such a
pair, yet all three merge. That is correct, because sub1 covers sub3's pairs.
In the order sub2, sub3, sub1, sub3 stays separate.
Could the docs add something like "With more subplans, each one is checked
against the subplans merged before it, so the result can depend on their order"?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala:
##########
@@ -660,18 +687,24 @@ 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 {
Review Comment:
Nit: Like the existing through-join check, the new check runs only after
this right-side merge has finished. When the left result already makes the
decline certain, the right merge still runs to the end and is then thrown away.
That includes DSv2 scan rebuilds: `rebuildScan` plus
`V2ScanPartitioningAndOrdering`, up to two per Filter pair. Each declined
subplan also becomes its own cache entry, so with n left-filtered and m
right-filtered subplans this repeats n x m times. Before this PR those merges
succeeded, so the work was used. The gridTest's `(v2OnLeft = false,
firstMerge)` case with symmetric off is one example.
Passing "np filter on the left" and "cached-side filter on the left" down
through `MergeContext` would let the right side's Filter arms return `None`
before rebuilding. It's planning time only and behind non-default configs, so a
separate change is fine.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/MergeSubplansSuite.scala:
##########
@@ -2121,6 +2121,125 @@ class MergeSubplansSuite extends PlanTest {
}
}
+ test("SPARK-60106: Do not merge subqueries that filter different join
children") {
+ // subquery1 filters the right child and subquery2 the left child. The
merged join would read
+ // both children unfiltered and build pairs neither subquery needs, so
without symmetric filter
+ // propagation the subqueries stay separate.
+ val subquery1 = ScalarSubquery(
+ testRelation.join(testRelation2.where($"d" > 1), Inner, Some($"a" ===
$"d"))
+ .groupBy()(sum($"a").as("sum_a")))
+ val subquery2 = ScalarSubquery(
+ testRelation.where($"a" > 1).join(testRelation2, Inner, 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)
+ }
+ }
+
+ test("SPARK-60106: Merge subqueries that filter different join children when
symmetric") {
Review Comment:
Nit: This test and the one above build the same `subquery1`, `subquery2` and
`originalQuery` (L2128-2134 and L2142-2148). If one of them is edited later,
the no-merge and merge tests can stop covering the same plan pair without any
failure. The nested-join test in this PR already checks both configs in one
test.
If you fold them, run the merge check with the expected
`propagatedFilter_0`/`_1` first, because the declined run also consumes
`PlanMerger.newId` (L656, L684). A `gridTest` over the config would also work.
This one is optional.
##########
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:
Minor, related to the earlier L754 thread: the nested-join test uses a plain
cached filter, so at the inner Join `cachedSideFiltered` always equals
`cachedPlanFilter.isDefined`. As a result, changing this line to
`cachedSideFiltered = cpFilter.isDefined`, or to `rightCPSideFiltered`, still
passes every test.
This OR exists for a hidden flag from a dropped merged Filter (L679-680)
that reaches an outer Join, and no test covers that. The gridTest uses single
joins, and the nested test has no DSv2 scan. For example, take through-join and
`dsv2SymmetricFilterPropagation` on and symmetric off:
- `(V2.where(a > 1) JOIN B) JOIN C`
- `(V2.where(b > 2) JOIN B) JOIN C`
- `(V2 JOIN B) JOIN C.where(h > 1)`
The first two merge into a merged Filter on `V2`. The third should be
declined, but under either change it merges and builds `NOT(a > 1 OR b > 2) x
NOT(h > 1)` pairs. Could you add this shape next to the gridTest?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala:
##########
@@ -513,8 +535,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:
Minor: No test reaches this line, L662, or the `|| cpSideFiltered` term at
L604 with a child that reports `true` while symmetric propagation is off. So
changing any of them to `false` passes every test, and `false` is the fail-open
direction that dropping the default was meant to prevent. The test added for
the earlier L539/L540 thread only pins that this line doesn't report `true`.
Here is a small shape that pins the other direction:
```scala
val subquery1 = ScalarSubquery(
testRelation.where($"c" === "x").where($"a" > 1)
.join(testRelation2, Inner, Some($"a" === $"d"))
.groupBy()(sum($"a").as("sum_a")))
val subquery2 = ScalarSubquery(
testRelation.where($"a" > 1)
.join(testRelation2.where($"e" > 3), Inner, Some($"a" === $"d"))
.groupBy()(max($"a").as("max_a")))
```
The left pair has the same `a > 1` on both sides and recurses into `(T,
Filter(c = 'x', T))`, which reports `true`. This line forwards it, so the Join
arm declines. If this line returned `false`, the plans would merge and build
`(a > 1 AND c != 'x') x (e <= 3)` pairs, and no test would fail. To pin it,
assert no merge with only through-join on, and a merge with symmetric on as
well.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/MergeSubplansSuite.scala:
##########
@@ -2597,6 +2716,96 @@ 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:
Nit: Three small things in this gridTest:
- `mergedAggregates` and `mergedFilters` each repeat the same three configs
and their own `Optimize.execute`, so the `reused` case runs the optimizer four
times for two configs.
- Both `shape match` expressions end in `case _`. If a shape label in the
generator (L2722) is renamed or mistyped, that cell silently runs as a second
`dropped` cell and passes.
- The `firstMerge` cell builds an unused `v2In3`/`sub3`. It also checks for
no merge with `mergedAggregates(false).isEmpty` instead of the suite's usual
`comparePlans(Optimize.execute(q), q)`.
How about one `def optimized(symmetric: Boolean): LogicalPlan`, with the two
helpers as plain extractors over its result? Explicit `case "dropped" |
"firstMerge" =>` and `case "dropped" | "extended" =>` would fix the fallbacks.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -7987,11 +7987,14 @@ object SQLConf {
"FILTER (WHERE ...) clauses on each aggregate expression. " +
"Merging two filtered scans broadens the combined filter to OR(f1, f2),
which may " +
"reduce IO pruning (e.g. partition or file skipping) compared to the
individual " +
- "filters. When both subplans filter both join children, the merged join
can also " +
- "produce pairs that neither subplan needs, increasing join work or
making a formerly " +
- "broadcastable input too large to broadcast. Disabled by default; enable
once the " +
- "behaviour has been validated in your " +
- "workload, particularly on heavily partitioned or file-pruned tables. " +
+ "filters. Through a join, a merge also counts as symmetric when one
subplan filters a " +
Review Comment:
Minor, a design question rather than a blocker: this ties the
different-children case to the symmetric config, which mixes two different
costs. The symmetric config was about OR-widened scans losing pruning
(SPARK-40193). In the case this PR gates, nothing is OR-widened: each merged
child reads what the unfiltered side already read, and the cost is the extra
join pairs. That has two consequences:
- The tuning guide recommends setting through-join together with symmetric,
and the q88/q90 numbers use both. With both on, the new check is skipped
(`PlanMerger` L704), so the PR description's `l.id < 100` / `r.id < 100`
example still builds 4,000,000 rows instead of 2 x 20,000.
- A `dsv2SymmetricFilterPropagation` user who wants these merges has to turn
on V1 OR-widening as well.
q88 and q90 differ only on `time_dim`. So a separate config for this case,
internal and false by default, would keep the documented gains. If one config
is intended, feel free to ignore this.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala:
##########
@@ -319,14 +325,28 @@ 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
Review Comment:
Nit: Two doc details.
- "The merged Filter case only matters without
`symmetricFilterPropagationEnabled`" is true of the whole field. Its only
consumer is the Join check (L704), which short-circuits on that config. How
about the one-line form that `dsv2Merged` uses: "It only ever gates behaviour
when `symmetricFilterPropagationEnabled` is off, where a merged Filter can only
come from the DSv2-symmetric exemption."
- The `tryMergePlans` Scaladoc (L417-418) still says "filter differences on
either or both children can merge". The first SPARK-60106 test's shape is now
declined without symmetric propagation, but only the class doc was updated.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala:
##########
@@ -633,24 +659,25 @@ class PlanMerger(
mergedChild.output.toList :+ newNPFilterAlias,
mergedChild)
TryMergeResult(project, npMapping, Some(newNPFilter), cpFilter,
- dsv2Merged = dsv2Merged)
+ dsv2Merged = dsv2Merged, cachedSideFiltered = cpSideFiltered)
}
case (np, cp: Filter) if context.filterPropagationSupported =>
tryMergePlans(np, cp.child, context.copy(filterAboveScan =
false)).collect {
// If the np side already propagated a filter from deeper
recursion, the merge is
// effectively symmetric (both sides have a filter condition).
Abort unless
// symmetricFilterPropagationEnabled.
- case TryMergeResult(mergedChild, npMapping, npFilter, cpFilter, _,
dsv2Merged)
- if npFilter.isEmpty || symmetricFilterPropagationEnabled =>
+ case TryMergeResult(mergedChild, npMapping, npFilter, cpFilter, _,
dsv2Merged,
+ cpSideFiltered) if npFilter.isEmpty ||
symmetricFilterPropagationEnabled =>
Review Comment:
Nit: This arm binds `cpSideFiltered` but never reads it, since both branches
return `true` (L680, L690). `true` is right here, because cp's own Filter
filters this subtree. But the sibling arm at L652 binds the same name and
forwards it, so this one reads as if it forwards too. How about `_`, as in the
Aggregate arms?
```scala
case TryMergeResult(mergedChild, npMapping, npFilter, cpFilter, _,
dsv2Merged, _)
if npFilter.isEmpty || symmetricFilterPropagationEnabled =>
```
--
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]
