You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by do...@apache.org on 2026/10/17 00:23:22 UTC
(spark) branch branch-4.x updated: [SPARK-60304][SQL] Fix `PlanMerger` losing a shared lower filter when merging into a merged Filter
This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun 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 bd71d0d80bc2 [SPARK-60304][SQL] Fix `PlanMerger` losing a shared lower filter when merging into a merged Filter
bd71d0d80bc2 is described below
commit bd71d0d80bc2aa389844146dedf34404b333a1a0
Author: Dongjoon Hyun <do...@apache.org>
AuthorDate: Fri Oct 16 17:23:05 2026 -0700
[SPARK-60304][SQL] Fix `PlanMerger` losing a shared lower filter when merging into a merged Filter
### What changes were proposed in this pull request?
`PlanMerger.tryMergePlans` merges a new plan (np) into a cached plan (cp). When symmetric filter propagation merges two `Filter`s with different conditions, it tags the merged `Filter` with `MERGED_FILTER_TAG`. Its condition is `OR(pf_0, pf_1, ...)` of propagated filter aliases, and each cached aggregate carries its own `FILTER (WHERE pf_i)`.
This PR changes the two branches that merge np into a tagged cp `Filter`. They now pass the child merge's `cachedPlanFilter` up instead of replacing it with `None`:
- `(np: Filter, cp: Filter)` with a tagged cp, which extends the OR or reuses one of its aliases.
- `(np, cp: Filter)` with a tagged cp, which drops cp's `Filter`.
The enclosing `Aggregate` then applies `FILTER (WHERE cpFilter AND pf_i)` to the cached aggregates. The other branches already pass the child's `cachedPlanFilter` up or fold it into cp's new filter alias.
Note for reviewers: this touches the same two return statements as https://github.com/apache/spark/pull/59321 (SPARK-60106). Whichever PR merges second needs a small rebase that keeps both changes.
### Why are the changes needed?
A child merge sets `cachedPlanFilter` when cp has a `Filter` below the tagged one and np does not. Every plan merged into cp so far had that `Filter`, so it stayed a plain `Filter`, and none of the `pf_i` includes it. Dropping `cachedPlanFilter` therefore lets the cached aggregates read rows that this `Filter` removed.
`mergeNamedExpressions` wraps the cached aliases in `If(cpFilter, ..., null)`, which hides the problem in some plans. However, it skips aliases of a plain attribute and cached expressions that np also computes.
For example, with symmetric filter propagation on:
- sq1: `sum(a)` over `t.where(a > 0).where(fa)`
- sq2: `max(a)` over `t.where(a > 0).where(fb)`
- sq3: `count(a)` over `t`
Merging sq1 and sq2 aliases `fb` as `pf_0` and `fa` as `pf_1` above the shared `a > 0` Filter. Merging sq3 turns `a > 0` into `pf_2` and drops the tagged Filter. The aggregates stay `sum(a) FILTER (WHERE pf_1)` and `max(a) FILTER (WHERE pf_0)`, so they also read the rows where `a <= 0`.
The `None`s come from https://github.com/apache/spark/pull/55298 (SPARK-40193, the tagged `(Filter, Filter)` branch) and https://github.com/apache/spark/pull/55659 (SPARK-56570, the tagged `(np, cp: Filter)` branch).
- https://github.com/apache/spark/pull/55298
- https://github.com/apache/spark/pull/55659
### Does this PR introduce _any_ user-facing change?
Yes, it fixes wrong results.
Since Apache Spark [v4.2.0](https://github.com/apache/spark/releases/tag/v4.2.0) (2026-07-11), merged non-grouping aggregates can ignore such a shared lower filter. This happens when `spark.sql.optimizer.mergeSubplans.filterPropagation.symmetricFilterPropagation.enabled` is `true` (default `false`). In the unreleased 4.3.0 and later, `spark.sql.optimizer.mergeSubplans.filterPropagation.dsv2SymmetricFilterPropagation.enabled` (default `false`) reaches the same code for merged DSv2 scans.
With symmetric filter propagation on, plans reach the fixed branches without excluding any optimizer rule too. For example, with `spark.sql.optimizer.mergeSubplans.filterPropagation.throughJoin.enabled`, the shared filter can sit below a join and the differing filters above it. The answers there happen to stay right, because `mergeNamedExpressions` wraps the cached `pf_i` in `If(cpFilter, ..., null)`. A wrong answer needs a `pf_i` that is a bare attribute, or a cached expression that [...]
The new end-to-end test runs the query below. It excludes `PushDownPredicates`, which would combine the stacked `Filter`s, so that `WHERE fa` stays a bare-attribute `Filter` above the shared `a > 0` one.
```scala
import org.apache.spark.sql.Row
import org.apache.spark.sql.types._
val schema = new StructType()
.add("a", IntegerType, nullable = false)
.add("fa", BooleanType, nullable = false)
.add("fb", BooleanType, nullable = false)
spark.createDataFrame(
spark.sparkContext.parallelize(Seq(Row(-5, true, true), Row(3, true, false))), schema)
.createOrReplaceTempView("t")
spark.conf.set(
"spark.sql.optimizer.mergeSubplans.filterPropagation.symmetricFilterPropagation.enabled", "true")
spark.conf.set(
"spark.sql.optimizer.excludedRules", "org.apache.spark.sql.catalyst.optimizer.PushDownPredicates")
spark.sql("""
SELECT
(SELECT sum(a) FROM (SELECT * FROM t WHERE a > 0) WHERE fa),
(SELECT max(a) FROM (SELECT * FROM t WHERE a > 0) WHERE fb),
(SELECT count(a) FROM t)
""")
```
It returns `[-2,-5,2]` before this PR and `[3,null,2]` after it.
### How was this patch tested?
Added five tests. All five fail without this fix.
- `MergeSubplansSuite`:
- The tagged `(np, cp: Filter)` branch under symmetric filter propagation (the example above).
- The tagged `(Filter, Filter)` branch under symmetric filter propagation. A `c = 'x'` filter is shared below, and the new plan also computes one of the cached filter conditions.
- The same branch when the new plan reuses that cached filter alias. The alias stays right for the new plan, and the cached aggregates still get the lower filter.
- The tagged `(np, cp: Filter)` branch under the DSv2 symmetric config only, with the general symmetric config off.
- `PlanMergingSuite`: the end-to-end query above, with AQE on and off.
Ran the following command. It runs `MergeSubplansSuite`, `PlanMergingSuite`, `DSv2PlanMergingSuite`, `FileSourceV1PlanMergingSuite` and `FileSourceV2PlanMergingSuite`, and all 159 tests pass.
```
build/sbt 'sql/testOnly *MergeSubplansSuite *PlanMergingSuite'
```
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 5.5
Closes #59562 from dongjoon-hyun/SPARK-60304.
Authored-by: Dongjoon Hyun <do...@apache.org>
Signed-off-by: Dongjoon Hyun <do...@apache.org>
(cherry picked from commit b9159d596ad0fb917d55e81d3a509fda365b8ecd)
Signed-off-by: Dongjoon Hyun <do...@apache.org>
---
.../sql/execution/planmerging/PlanMerger.scala | 17 +-
.../execution/planmerging/MergeSubplansSuite.scala | 202 ++++++++++++++++++++-
.../execution/planmerging/PlanMergingSuite.scala | 44 +++++
3 files changed, 256 insertions(+), 7 deletions(-)
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala
index 4b48825af982..bc57176e98dc 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala
@@ -482,8 +482,10 @@ class PlanMerger(
// Apply each propagated boolean attribute as a FILTER (WHERE ...) clause on the
// corresponding side's aggregate expressions.
- // A None filter means the side's aggregate expressions already carry their individual
- // FILTER attributes from a previous merge round and should be left unchanged.
+ // A None filter leaves the side's aggregate expressions unchanged, e.g. ones that
+ // already carry their individual FILTER attributes from a previous merge round. A
+ // Some filter is ANDed with any FILTER they carry, e.g. a cpFilter from a filter
+ // below a merged Filter, which those attributes don't include.
// Filter propagation is consumed here and not passed further up.
val filteredNPAggregateExpressions = npFilterOpt.fold(np.aggregateExpressions)(
applyFilterToAggregateExpressions(np.aggregateExpressions, _))
@@ -574,7 +576,9 @@ class PlanMerger(
projectList = newProjectList, child = prunedChild)
val newFilter = Filter(newCondition, newProject)
newFilter.copyTagsFrom(cp)
- TryMergeResult(newFilter, npMapping, Some(newNPFilterOut), None,
+ // The child's cpFilter is a filter below this one that np lacks. The
+ // OR's aliases don't include it, so pass it up to cp's aggregates.
+ TryMergeResult(newFilter, npMapping, Some(newNPFilterOut), cpFilter,
dsv2Merged = dsv2Merged)
}
case _ =>
@@ -648,9 +652,10 @@ class PlanMerger(
// restrict each aggregation to its originating side. Synthesising a new cpFilter
// alias for cp.condition would just produce `FILTER AND(OR(pf_0, pf_1, ...), pf_i)`
// upstream, which simplifies to `FILTER pf_i` -- wasted work and plan bloat.
- // Drop cp's Filter and let the recursion's result flow up with cpFilter = None so
- // cp's aggregates are left untouched.
- TryMergeResult(mergedChild, npMapping, npFilter, None, dsv2Merged = dsv2Merged)
+ // Drop cp's Filter and let the recursion's result flow up as is. A cpFilter in it
+ // is a filter below cp's that np lacks, which the pf_i don't include, so cp's
+ // aggregates still need it.
+ TryMergeResult(mergedChild, npMapping, npFilter, cpFilter, dsv2Merged = dsv2Merged)
} else {
val newCPCondition = cpFilter.fold(cp.condition)(And(_, cp.condition))
val newCPFilterAlias =
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/MergeSubplansSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/MergeSubplansSuite.scala
index 99340374bab9..564fd617d6d1 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/MergeSubplansSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/MergeSubplansSuite.scala
@@ -36,7 +36,7 @@ import org.apache.spark.sql.connector.read.partitioning.{KeyGroupedPartitioning
import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Relation, DataSourceV2ScanRelation, V2ScanPartitioningAndOrdering, V2ScanRelationPushDown}
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.internal.connector.SupportsPushDownCatalystFilters
-import org.apache.spark.sql.types.{DataType, IntegerType, StringType, StructField, StructType}
+import org.apache.spark.sql.types.{BooleanType, DataType, IntegerType, StringType, StructField, StructType}
import org.apache.spark.sql.util.CaseInsensitiveStringMap
class MergeSubplansSuite extends PlanTest {
@@ -1727,6 +1727,156 @@ class MergeSubplansSuite extends PlanTest {
}
}
+ test("SPARK-60304: `(np, cp: Filter)` dropping a tagged cp Filter keeps the cpFilter " +
+ "from below it") {
+ // Round 1+2: sq1 and sq2 share the inner `a > 0` Filter, so it stays a plain Filter, and their
+ // outer Filters merge via `(Filter, Filter)` first-time into a tagged Filter (condition
+ // `OR(pf_0=fb, pf_1=fa)`). cp's aggregates are `[sum(a) FILTER pf_1, max(a) FILTER pf_0]`,
+ // and only the plain Filter below restricts them to `a > 0`.
+ // Round 3: sq3 has no Filter, so `(np, cp: Filter)` with cp tagged fires. Its recursion turns
+ // the plain Filter into `pf_2 = (a > 0)` as cpFilter. `pf_0` and `pf_1` alias attributes, so
+ // the Project does not wrap them with `pf_2`, and cp's aggregates have to get `pf_2` as a
+ // FILTER. Otherwise `sum(a) FILTER pf_1` also sums the rows where `a <= 0`.
+ val relation = LocalRelation($"a".int, $"fa".boolean, $"fb".boolean)
+ val subquery1 = ScalarSubquery(
+ relation.where($"a" > 0).where($"fa").groupBy()(sum($"a").as("sum_a")))
+ val subquery2 = ScalarSubquery(
+ relation.where($"a" > 0).where($"fb").groupBy()(max($"a").as("max_a")))
+ val subquery3 = ScalarSubquery(relation.groupBy()(count($"a").as("count_a")))
+ val originalQuery = relation.select(subquery1, subquery2, subquery3)
+
+ val f0Alias = Alias($"fb", "propagatedFilter_0")()
+ val f0 = f0Alias.toAttribute
+ val f1Alias = Alias($"fa", "propagatedFilter_1")()
+ val f1 = f1Alias.toAttribute
+ val f2Alias = Alias($"a" > 0, "propagatedFilter_2")()
+ val f2 = f2Alias.toAttribute
+ val mergedSubquery = relation
+ .select(relation.output :+ f2Alias: _*)
+ .select(relation.output ++ Seq(f0Alias, f1Alias, f2): _*)
+ .groupBy()(
+ sum($"a", Some(And(f2, f1))).as("sum_a"),
+ max($"a", Some(And(f2, f0))).as("max_a"),
+ count($"a").as("count_a"))
+ .select(CreateNamedStruct(Seq(
+ Literal("sum_a"), $"sum_a",
+ Literal("max_a"), $"max_a",
+ Literal("count_a"), $"count_a"
+ )).as("mergedValue"))
+ val analyzedMergedSubquery = mergedSubquery.analyze
+ val correctAnswer = WithCTE(
+ relation.select(
+ extractorExpression(0, analyzedMergedSubquery.output, 0),
+ extractorExpression(0, analyzedMergedSubquery.output, 1),
+ extractorExpression(0, analyzedMergedSubquery.output, 2)),
+ Seq(definitionNode(analyzedMergedSubquery, 0)))
+
+ withSQLConf(SQLConf.MERGE_SUBPLANS_SYMMETRIC_FILTER_PROPAGATION_ENABLED.key -> "true") {
+ comparePlans(Optimize.execute(originalQuery.analyze), correctAnswer.analyze)
+ }
+ }
+
+ test("SPARK-60304: tagged `(Filter, Filter)` keeps the cpFilter from below it") {
+ // Round 1+2: sq1 and sq2 share the inner `c = 'x'` Filter, so it stays a plain Filter, and
+ // their outer Filters merge via `(Filter, Filter)` first-time into a tagged Filter (condition
+ // `OR(pf_0=(b>2), pf_1=(a>1))`). Only the plain Filter below restricts cp's aggregates to
+ // `c = 'x'`.
+ // Round 3: sq3 has no `c = 'x'` Filter, so the recursion below the tagged `(Filter, Filter)`
+ // turns the plain Filter into `pf_2 = (c = 'x')` as cpFilter. The Project wraps `pf_1` with
+ // `pf_2`, but not `pf_0`, because sq3 computes `b > 2` too. So cp's aggregates have to get
+ // `pf_2` as a FILTER. Otherwise `max(a) FILTER pf_0` also reads the rows where `c != 'x'`.
+ val subquery1 = ScalarSubquery(
+ testRelation.where($"c" === "x").where($"a" > 1).groupBy()(sum($"a").as("sum_a")))
+ val subquery2 = ScalarSubquery(
+ testRelation.where($"c" === "x").where($"b" > 2).groupBy()(max($"a").as("max_a")))
+ val subquery3 = ScalarSubquery(
+ testRelation
+ .select($"a", $"b", ($"b" > 2).as("flag"))
+ .where($"a" < 0)
+ .groupBy()(min($"a").as("min_a")))
+ val originalQuery = testRelation.select(subquery1, subquery2, subquery3)
+
+ val f0Alias = Alias($"b" > 2, "propagatedFilter_0")()
+ val f0 = f0Alias.toAttribute
+ val f2Alias = Alias($"c" === "x", "propagatedFilter_2")()
+ val f2 = f2Alias.toAttribute
+ val f1Alias = Alias(If(f2, $"a" > 1, Literal(null, BooleanType)), "propagatedFilter_1")()
+ val f1 = f1Alias.toAttribute
+ val f3Alias = Alias($"a" < 0, "propagatedFilter_3")()
+ val f3 = f3Alias.toAttribute
+ val mergedSubquery = testRelation
+ .select(testRelation.output :+ f2Alias: _*)
+ .select(testRelation.output ++ Seq(f0Alias, f1Alias, f2, f3Alias): _*)
+ .where(Or(Or(f0, f1), f3))
+ .groupBy()(
+ sum($"a", Some(And(f2, f1))).as("sum_a"),
+ max($"a", Some(And(f2, f0))).as("max_a"),
+ min($"a", Some(f3)).as("min_a"))
+ .select(CreateNamedStruct(Seq(
+ Literal("sum_a"), $"sum_a",
+ Literal("max_a"), $"max_a",
+ Literal("min_a"), $"min_a"
+ )).as("mergedValue"))
+ val analyzedMergedSubquery = mergedSubquery.analyze
+ val correctAnswer = WithCTE(
+ testRelation.select(
+ extractorExpression(0, analyzedMergedSubquery.output, 0),
+ extractorExpression(0, analyzedMergedSubquery.output, 1),
+ extractorExpression(0, analyzedMergedSubquery.output, 2)),
+ Seq(definitionNode(analyzedMergedSubquery, 0)))
+
+ withSQLConf(SQLConf.MERGE_SUBPLANS_SYMMETRIC_FILTER_PROPAGATION_ENABLED.key -> "true") {
+ comparePlans(Optimize.execute(originalQuery.analyze), correctAnswer.analyze)
+ }
+ }
+
+ test("SPARK-60304: tagged `(Filter, Filter)` reuse keeps the cpFilter from below it") {
+ // Like the test above, but sq3 filters on `b > 2`, which it also computes, so the Project
+ // leaves `pf_0` unwrapped and sq3 reuses it. A plain `pf_0` is right for sq3, which has no
+ // `c = 'x'` Filter, but cp's aggregates still have to get `pf_2` as a FILTER.
+ val subquery1 = ScalarSubquery(
+ testRelation.where($"c" === "x").where($"a" > 1).groupBy()(sum($"a").as("sum_a")))
+ val subquery2 = ScalarSubquery(
+ testRelation.where($"c" === "x").where($"b" > 2).groupBy()(max($"a").as("max_a")))
+ val subquery3 = ScalarSubquery(
+ testRelation
+ .select($"a", $"b", ($"b" > 2).as("flag"))
+ .where($"b" > 2)
+ .groupBy()(min($"a").as("min_a")))
+ val originalQuery = testRelation.select(subquery1, subquery2, subquery3)
+
+ val f0Alias = Alias($"b" > 2, "propagatedFilter_0")()
+ val f0 = f0Alias.toAttribute
+ val f2Alias = Alias($"c" === "x", "propagatedFilter_2")()
+ val f2 = f2Alias.toAttribute
+ val f1Alias = Alias(If(f2, $"a" > 1, Literal(null, BooleanType)), "propagatedFilter_1")()
+ val f1 = f1Alias.toAttribute
+ val mergedSubquery = testRelation
+ .select(testRelation.output :+ f2Alias: _*)
+ .select(testRelation.output ++ Seq(f0Alias, f1Alias, f2): _*)
+ .where(Or(f0, f1))
+ .groupBy()(
+ sum($"a", Some(And(f2, f1))).as("sum_a"),
+ max($"a", Some(And(f2, f0))).as("max_a"),
+ min($"a", Some(f0)).as("min_a"))
+ .select(CreateNamedStruct(Seq(
+ Literal("sum_a"), $"sum_a",
+ Literal("max_a"), $"max_a",
+ Literal("min_a"), $"min_a"
+ )).as("mergedValue"))
+ val analyzedMergedSubquery = mergedSubquery.analyze
+ val correctAnswer = WithCTE(
+ testRelation.select(
+ extractorExpression(0, analyzedMergedSubquery.output, 0),
+ extractorExpression(0, analyzedMergedSubquery.output, 1),
+ extractorExpression(0, analyzedMergedSubquery.output, 2)),
+ Seq(definitionNode(analyzedMergedSubquery, 0)))
+
+ withSQLConf(SQLConf.MERGE_SUBPLANS_SYMMETRIC_FILTER_PROPAGATION_ENABLED.key -> "true") {
+ comparePlans(Optimize.execute(originalQuery.analyze), correctAnswer.analyze)
+ }
+ }
+
test("SPARK-40193: Merge non-grouping subqueries where one aggregate already carries a " +
"FILTER clause") {
val subquery1 = ScalarSubquery(testRelation.groupBy()(max($"a").as("max_a")))
@@ -2880,6 +3030,56 @@ class MergeSubplansSuite extends PlanTest {
}
}
+ test("SPARK-60304: DSv2 symmetric merge keeps the cpFilter from below a dropped tagged Filter") {
+ // The DSv2 counterpart of "`(np, cp: Filter)` dropping a tagged cp Filter keeps the cpFilter
+ // from below it", under the DSv2-specific config only. sub1 and sub2 read different columns,
+ // so their scans go through a DSv2 merge below the shared `a > 0` Filter, which lets the DSv2
+ // exemption merge their outer Filters into a tagged Filter (`OR(pf_0=fb, pf_1=fa)`). sub3 has
+ // no Filter and its scan merges too, so dropping the tagged Filter has to pass
+ // `pf_2 = (a > 0)` up to cp's aggregates.
+ val t = new TestV2Table(StructType(Seq(StructField("a", IntegerType),
+ StructField("fa", BooleanType), StructField("fb", BooleanType))))
+ val sub1 = ScalarSubquery(v2ScanReadingOn(t, Seq("a", "fa"))
+ .where($"a" > 0).where($"fa").groupBy()(sum($"a").as("sum_a")))
+ val sub2 = ScalarSubquery(v2ScanReadingOn(t, Seq("a", "fb"))
+ .where($"a" > 0).where($"fb").groupBy()(max($"a").as("max_a")))
+ val sub3 = ScalarSubquery(v2ScanReadingOn(t, Seq("a")).groupBy()(count($"a").as("count_a")))
+ val originalQuery = testRelation.select(sub1, sub2, sub3)
+
+ val mergedScan = v2ScanReadingOn(t, Seq("a", "fa", "fb"))
+ val f0Alias = Alias($"fb", "propagatedFilter_0")()
+ val f0 = f0Alias.toAttribute
+ val f1Alias = Alias($"fa", "propagatedFilter_1")()
+ val f1 = f1Alias.toAttribute
+ val f2Alias = Alias($"a" > 0, "propagatedFilter_2")()
+ val f2 = f2Alias.toAttribute
+ val mergedSubquery = mergedScan
+ .select(mergedScan.output :+ f2Alias: _*)
+ .select(mergedScan.output ++ Seq(f0Alias, f1Alias, f2): _*)
+ .groupBy()(
+ sum($"a", Some(And(f2, f1))).as("sum_a"),
+ max($"a", Some(And(f2, f0))).as("max_a"),
+ count($"a").as("count_a"))
+ .select(CreateNamedStruct(Seq(
+ Literal("sum_a"), $"sum_a",
+ Literal("max_a"), $"max_a",
+ Literal("count_a"), $"count_a")).as("mergedValue"))
+ val analyzedMergedSubquery = mergedSubquery.analyze
+ val correctAnswer = WithCTE(
+ testRelation.select(
+ extractorExpression(0, analyzedMergedSubquery.output, 0),
+ extractorExpression(0, analyzedMergedSubquery.output, 1),
+ extractorExpression(0, analyzedMergedSubquery.output, 2)),
+ Seq(definitionNode(analyzedMergedSubquery, 0)))
+
+ withSQLConf(
+ SQLConf.MERGE_SUBPLANS_FILTER_PROPAGATION_ENABLED.key -> "true",
+ SQLConf.MERGE_SUBPLANS_SYMMETRIC_FILTER_PROPAGATION_ENABLED.key -> "false",
+ SQLConf.MERGE_SUBPLANS_DSV2_SYMMETRIC_FILTER_PROPAGATION_ENABLED.key -> "true") {
+ comparePlans(Optimize.execute(originalQuery.analyze), correctAnswer.analyze)
+ }
+ }
+
test("SPARK-40259: a non-deterministic filter conjunct is not pushed as a pruning predicate") {
withSQLConf(
SQLConf.MERGE_SUBPLANS_FILTER_PROPAGATION_ENABLED.key -> "true",
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/PlanMergingSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/PlanMergingSuite.scala
index 08925befd102..7f846a969f02 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/PlanMergingSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/PlanMergingSuite.scala
@@ -22,6 +22,7 @@ import org.apache.spark.sql.execution._
import org.apache.spark.sql.execution.adaptive.{AdaptiveSparkPlanHelper}
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.sql.types.{BooleanType, IntegerType, StructType}
class PlanMergingSuite extends SharedSparkSession
with AdaptiveSparkPlanHelper {
@@ -397,6 +398,49 @@ class PlanMergingSuite extends SharedSparkSession
}
}
+ test("SPARK-60304: Merge a subquery without filters into subqueries that share a lower filter") {
+ // Excluding PushDownPredicates keeps the stacked Filters apart, so the first two subqueries
+ // merge their outer `fa` and `fb` Filters above the shared `a > 0` Filter. The third one has
+ // no Filter, and the merge has to keep `a > 0` for the first two aggregates, which would
+ // otherwise also read the row where `a <= 0` and return -2 and -5.
+ val schema = new StructType()
+ .add("a", IntegerType, nullable = false)
+ .add("fa", BooleanType, nullable = false)
+ .add("fb", BooleanType, nullable = false)
+ val rows = Seq(Row(-5, true, true), Row(3, true, false))
+ withTempView("t") {
+ spark.createDataFrame(spark.sparkContext.parallelize(rows), schema)
+ .createOrReplaceTempView("t")
+ Seq(false, true).foreach { enableAQE =>
+ withSQLConf(
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> enableAQE.toString,
+ SQLConf.MERGE_SUBPLANS_SYMMETRIC_FILTER_PROPAGATION_ENABLED.key -> "true",
+ SQLConf.OPTIMIZER_EXCLUDED_RULES.key ->
+ "org.apache.spark.sql.catalyst.optimizer.PushDownPredicates") {
+ val df = sql(
+ """
+ |SELECT
+ | (SELECT sum(a) FROM (SELECT * FROM t WHERE a > 0) WHERE fa),
+ | (SELECT max(a) FROM (SELECT * FROM t WHERE a > 0) WHERE fb),
+ | (SELECT count(a) FROM t)
+ """.stripMargin)
+
+ checkAnswer(df, Row(3, null, 2) :: Nil)
+
+ val plan = df.queryExecution.executedPlan
+ val subqueryIds = collectWithSubqueries(plan) { case s: SubqueryExec => s.id }
+ val reusedSubqueryIds = collectWithSubqueries(plan) {
+ case rs: ReusedSubqueryExec => rs.child.id
+ }
+
+ assert(subqueryIds.size == 1, "Missing or unexpected SubqueryExec in the plan")
+ assert(reusedSubqueryIds.size == 2,
+ "Missing or unexpected ReusedSubqueryExec in the plan")
+ }
+ }
+ }
+ }
+
test("SPARK-56677: Merge scalar subqueries with filter propagation through Join") {
// subquery1 has no filter; subquery2 filters on b > 1 (a column from the right side of the join
// that is not part of the join condition). Predicate pushdown can only push this filter to
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org