dongjoon-hyun opened a new pull request, #59562: URL: https://github.com/apache/spark/pull/59562
[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). ### 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. The new end-to-end test runs the query below. It excludes `PushDownPredicates` so that the stacked `Filter`s stay separate until `MergeSubplans`. ```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 four tests. All four 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 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 158 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 -- 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]
