This is an automated email from the ASF dual-hosted git repository.
peter-toth pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 1bb8a1857cf3 [SPARK-58562][SQL][DOC] Document subplan merging in the
SQL performance tuning guide
1bb8a1857cf3 is described below
commit 1bb8a1857cf3f0a8b3d50c31bdd8ca022e6f3bbb
Author: Peter Toth <[email protected]>
AuthorDate: Wed Aug 5 14:48:41 2026 +0200
[SPARK-58562][SQL][DOC] Document subplan merging in the SQL performance
tuning guide
### What changes were proposed in this pull request?
This PR adds a "Merging Subplans" section to
`docs/sql-performance-tuning.md`, between "Join Strategy Hints" and "Adaptive
Query Execution". Docs only, no code change.
The section covers:
- what `MergeSubplans` does and which subplans qualify (non-correlated
deterministic scalar subqueries and non-grouping aggregates), with a small
example that merges by default,
- how a merged subplan looks in `EXPLAIN` (a subquery whose single output
column is `mergedValue`, shared via `ReusedSubquery`),
- when two subplans are merged, and that a query still carrying a
non-inlined `WITH` clause is skipped,
- why the one-sided filter case is always beneficial while the symmetric
one is not, with the TPC-DS numbers measured in SPARK-40193 and SPARK-56677 and
guidance on when to consider enabling it,
- the 4 configs under
`spark.sql.optimizer.mergeSubplans.filterPropagation.*` in a config table, and
how to turn the rule off via `spark.sql.optimizer.excludedRules`.
### Why are the changes needed?
The optimization and its configs are not described anywhere in the docs
today. The 4 configs are public, so they show up in the generated SQL config
table in `configuration.md`, but nothing tells users what the optimization
does, which query shapes it applies to, or what the trade-offs of the 3
non-default configs are. The only other mention in `docs/` is the
migration-guide note about the rule's package move.
This matters most for symmetric filter propagation. It is disabled by
default because merging two filtered scans widens the scan filter to `OR(f1,
f2)`, which can lose partition or file pruning. But on queries that compute
several differently filtered aggregates over the same table it is a big win: q9
and q28 got about 3.5x faster in TPC-DS runs, and together with propagation
through joins q88 got about 7x and q90 about 2x faster. Users have no way to
discover that today.
The section also makes the plan shape discoverable: a merged subplan shows
up as a subquery with a `mergedValue` column, which is otherwise hard to trace
back to its source.
### Does this PR introduce _any_ user-facing change?
No, documentation only.
### How was this patch tested?
No tests, docs only. I verified every statement against the implementation
on master: the config keys, defaults and versions against `SQLConf`, the merge
conditions and the join nullability rules against `PlanMerger`, the skipped-CTE
and candidate conditions against `MergeSubplans`, the `SCAN_MERGING` opt-in
against `TableCapability` and its (test-only) implementors, and the rule name
used in `excludedRules` against `Rule.ruleName` and the non-excludable rule
list. The `mergedValue` o [...]
The added table follows the same `spark-config` markup as the other tables
on the page.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 5)
Closes #57765 from peter-toth/SPARK-58562-document-plan-merging.
Authored-by: Peter Toth <[email protected]>
Signed-off-by: Peter Toth <[email protected]>
(cherry picked from commit 190ea9b3f0acc4f9ae2ba3db51607c87f9c31b85)
Signed-off-by: Peter Toth <[email protected]>
---
docs/sql-performance-tuning.md | 72 ++++++++++++++++++++++++++++++++++++++++++
1 file changed, 72 insertions(+)
diff --git a/docs/sql-performance-tuning.md b/docs/sql-performance-tuning.md
index 3ba3a6c749ca..e0cba9362736 100644
--- a/docs/sql-performance-tuning.md
+++ b/docs/sql-performance-tuning.md
@@ -258,6 +258,78 @@ SELECT /*+ BROADCAST(r) */ * FROM src s JOIN records r ON
s.key = r.key
For more details please refer to the documentation of [Join
Hints](sql-ref-syntax-qry-select-hints.html#join-hints).
+## Merging Subplans
+
+Spark merges subplans that return a single row and read the same input, so
that input is scanned once instead of once per subplan. The candidates are
non-correlated deterministic scalar subqueries and non-grouping aggregates
(aggregates without `GROUP BY`). The merged subplan is evaluated once and
outputs a single struct, and each original site reads its own field out of that
struct. This optimization is enabled by default.
+
+For example, in the following query both subqueries scan `store_sales`:
+
+```sql
+SELECT
+ (SELECT min(ss_net_paid) FROM store_sales),
+ (SELECT max(ss_net_paid) FROM store_sales)
+```
+
+They are merged into one aggregate that computes `min` and `max` together, so
`store_sales` is read once. In `EXPLAIN` output a merged subplan shows up as a
subquery whose single output column is named `mergedValue`, and the sites that
share it as `ReusedSubquery`.
+
+Two subplans are merged when their plans match node by node: `Project` lists
are unioned, `Aggregate`s must have the same grouping and use the same
aggregation implementation (so a `min` is not merged with a `collect_list`),
`Filter`s must have the same condition, `Join`s must have the same type,
condition and hints, and the leaves must read the same input. Subplans that
differ only in their `WHERE` conditions can be merged as well, by turning each
side's condition into a boolean column [...]
+
+When only one of the two subplans has a filter, merging is always beneficial,
because the unfiltered side reads all the data anyway. This case is on by
default, unless the filter has to cross a `Join` to reach the aggregate, which
needs the through-join configuration below. When both sides have a filter (the
symmetric case), the merged scan filter becomes `OR(f1, f2)`, which is less
selective than either original filter and can therefore read more data - for
example when the filters prun [...]
+
+Still, it is worth considering on queries that compute several differently
filtered aggregates over the same table, which is a common analytical shape:
+
+```sql
+SELECT
+ (SELECT avg(ss_net_paid) FROM store_sales WHERE ss_quantity BETWEEN 1 AND
20),
+ (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 pr [...]
+
+<table class="spark-config">
+ <thead><tr><th>Property Name</th><th>Default</th><th>Meaning</th><th>Since
Version</th></tr></thead>
+ <tr>
+
<td><code>spark.sql.optimizer.mergeSubplans.filterPropagation.enabled</code></td>
+ <td>true</td>
+ <td>
+ When true, subplans that differ only in their filter conditions can be
merged by propagating the filters up to the enclosing non-grouping aggregates.
This is the umbrella configuration of the three below: none of them has any
effect when this is false. Subplans with the same filter condition are merged
regardless of this configuration.
+ </td>
+ <td>4.2.0</td>
+ </tr>
+ <tr>
+
<td><code>spark.sql.optimizer.mergeSubplans.filterPropagation.symmetricFilterPropagation.enabled</code></td>
+ <td>false</td>
+ <td>
+ When true, two non-grouping aggregate subplans that both have a filter
condition can also be merged. Disabled by default because the merged filter is
widened to <code>OR(f1, f2)</code>, which may read more data than the two
original filters, especially on heavily partitioned or file-pruned tables.
+ </td>
+ <td>4.2.0</td>
+ </tr>
+ <tr>
+
<td><code>spark.sql.optimizer.mergeSubplans.filterPropagation.throughJoin.enabled</code></td>
+ <td>false</td>
+ <td>
+ When true, filter conditions can also propagate through
<code>Join</code> nodes, which lets subplans that differ only in their filter
conditions and share a common join be merged. When false, no filter is
propagated across a join, not even when only one of the two subplans has a
filter. A filter only propagates from the preserved side of the join: the left
side of <code>LEFT OUTER</code>/<code>LEFT SEMI</code>/<code>LEFT ANTI</code>,
the right side of <code>RIGHT OUTER</code>, or e [...]
+ </td>
+ <td>4.2.0</td>
+ </tr>
+ <tr>
+
<td><code>spark.sql.optimizer.mergeSubplans.filterPropagation.dsv2SymmetricFilterPropagation.enabled</code></td>
+ <td>false</td>
+ <td>
+ When true, two DataSource V2 scans that pushed the same strictly
enforced filters but carry different best-effort (post-scan) filters can be
merged even when
<code>spark.sql.optimizer.mergeSubplans.filterPropagation.symmetricFilterPropagation.enabled</code>
is false. In this case widening cannot change the set of rows the scan is
required to return, as the strict filters are re-pushed unchanged and the
enclosing <code>Filter</code> re-checks the rest above the scan. This applies
on [...]
+ </td>
+ <td>4.3.0</td>
+ </tr>
+</table>
+
+Subplan merging can be turned off entirely by adding the rule to
`spark.sql.optimizer.excludedRules`:
+
+```
+spark.sql.optimizer.excludedRules=org.apache.spark.sql.execution.planmerging.MergeSubplans
+```
+
+Use exactly that name: the rule was called `MergeScalarSubqueries` before
Spark 4.2 and sat in a different package before Spark 4.3, and unknown names in
`spark.sql.optimizer.excludedRules` are silently ignored, so an older name
carried over from a previous version does not turn the rule off. See the [SQL
migration guide](sql-migration-guide.html) for the old names.
+
## Adaptive Query Execution
Adaptive Query Execution (AQE) is an optimization technique in Spark SQL that
makes use of the runtime statistics to choose the most efficient query
execution plan, which is enabled by default since Apache Spark 3.2.0. Spark SQL
can turn on and off AQE by `spark.sql.adaptive.enabled` as an umbrella
configuration.
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
