peter-toth commented on code in PR #59321:
URL: https://github.com/apache/spark/pull/59321#discussion_r4297595045
##########
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:
The symmetric and through-join docs, the tuning guide rows, the class doc
and the new migration guide entry now state the rule. 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 two shapes are examples now
([0fb8bcce413](https://github.com/apache/spark/pull/59321/commits/0fb8bcce413b83ac6c4ab82e9a8e30e513f2d271)).
##########
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:
Added "SPARK-60106: Do not merge subqueries that filter different children
of nested joins". It expects no merge with symmetric off and a merge with it
on. It fails if the Join arm stops forwarding `cachedSideFiltered`
([0fb8bcce413](https://github.com/apache/spark/pull/59321/commits/0fb8bcce413b83ac6c4ab82e9a8e30e513f2d271)).
##########
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:
Added "SPARK-60106: Merge subqueries that share a filter on one join child".
Your example doesn't reach this line, because identical subtrees are reused
before the Filter arms run. So in the test both subqueries filter `a > 1` above
different projections. It fails if this line reports `true`
([0fb8bcce413](https://github.com/apache/spark/pull/59321/commits/0fb8bcce413b83ac6c4ab82e9a8e30e513f2d271)).
##########
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:
Dropped the default. The identical result, the two Aggregate results and the
two DSv2 leaf results now pass `false` explicitly
([0fb8bcce413](https://github.com/apache/spark/pull/59321/commits/0fb8bcce413b83ac6c4ab82e9a8e30e513f2d271)).
##########
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:
Added an entry to the 4.3 to 4.4 section. It names the symmetric config and
says that it also enables the other symmetric merges. Since this is a plan
change under a non-default config, it goes to master and branch-4.x only. I
changed the JIRA to an Improvement and won't backport it to 4.3 or 4.2
([0fb8bcce413](https://github.com/apache/spark/pull/59321/commits/0fb8bcce413b83ac6c4ab82e9a8e30e513f2d271)).
##########
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:
Done. The sentence now starts with "With both configurations on" and states
the rule
([0fb8bcce413](https://github.com/apache/spark/pull/59321/commits/0fb8bcce413b83ac6c4ab82e9a8e30e513f2d271)).
--
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]
