dongjoon-hyun commented on code in PR #59562:
URL: https://github.com/apache/spark/pull/59562#discussion_r4299557428
##########
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:
Thank you. Moved it right above that return.
##########
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:
Thank you. Added it as a new test right after this one. sq3 keeps the
Project that computes `b > 2` as `flag`. With a plain `where(b > 2)`, the
Project wraps `pf_0` in `If(pf_2, ...)` because np doesn't compute it, so np
gets a new alias instead of reusing `pf_0`.
--
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]
