viirya commented on code in PR #59562:
URL: https://github.com/apache/spark/pull/59562#discussion_r4299366668
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala:
##########
@@ -532,6 +532,8 @@ class PlanMerger(
// condition. A tagged filter is always built with a Project
child (see the
// first-time branch below), so a non-Project child should
not happen; decline the
// merge rather than fail, keeping the merge best-effort.
+ // A cpFilter from the child merge still flows up to cp's
aggregates: it is a
+ // filter below this one that np lacks, and the OR's aliases
don't include it.
Review Comment:
nit: this note explains the `cpFilter` passed in `TryMergeResult(newFilter,
npMapping, Some(newNPFilterOut), cpFilter, ...)` about 40 lines below, but it
sits at the end of the block that describes the tagged branch as a whole.
Moving it next to that return would make it easier to connect the two.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/MergeSubplansSuite.scala:
##########
@@ -1727,6 +1727,109 @@ 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") {
Review Comment:
Optional: this covers the path where np gets a new alias. The other path,
where np reuses an existing side (`existingNPFilter` is `Some`) while the child
also brings a cpFilter, isn't covered. For example, sq3 could filter on `b > 2`
without the `c = 'x'` filter. Then np should get plain `pf_0` and cp should get
`pf_2 AND pf_i`. If you think it's worth guarding, this would be a cheap
addition.
--
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]
