peter-toth commented on code in PR #59294:
URL: https://github.com/apache/spark/pull/59294#discussion_r4219450680
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala:
##########
@@ -553,13 +547,14 @@ class PlanMerger(
existingNPFilter match {
case Some(reusedFilter) =>
// np matches an existing side: no new alias, OR
condition unchanged.
- (childProject.projectList, cp.condition,
(reusedFilter, false))
+ (childProject.projectList, cp.condition,
reusedFilter)
Review Comment:
Done in
[`0af2d628552`](https://github.com/apache/spark/pull/59294/commits/0af2d628552a7d90d8b39929dbfc9f32b79a684d).
The lookup now only matches the aliases `cp.condition` references. The new
plan test puts a computed `b > 2 AS flag` under a `b > 2` filter. Without the
restriction the merged plan gets `count(a) FILTER (WHERE flag)` under the
unchanged `Filter (propagatedFilter_0 OR propagatedFilter_1)`, and the test
fails.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala:
##########
@@ -311,12 +311,8 @@ class PlanMerger(
* @param newPlanFilter A boolean [[Attribute]] in the merged plan that
encodes the filter
* condition from the new plan's side, to be applied as
an aggregate
* `FILTER (WHERE ...)` clause when the propagation
reaches an enclosing
- * [[Aggregate]] node. The boolean component is `true`
if the attribute was
- * freshly aliased and must be appended to enclosing
[[Project]] nodes, or
- * `false` if it was reused from an existing alias
already present in the
- * merged plan. `None` when no differing filter was
propagated.
- * @param cachedPlanFilter Like `newPlanFilter` but for the cached plan's
side. Always a freshly
- * created alias when present, so no `isNew` flag is
needed.
+ * [[Aggregate]] node. `None` when no differing filter
was propagated.
Review Comment:
Done in
[`0af2d628552`](https://github.com/apache/spark/pull/59294/commits/0af2d628552a7d90d8b39929dbfc9f32b79a684d).
It says again that `newPlanFilter` can be a reused alias that an enclosing
`Project` may have to append, and that `cachedPlanFilter` is always fresh.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala:
##########
@@ -1105,11 +1098,9 @@ class PlanMerger(
}
}
- newPlanFilter.foreach {
- case (f, true) => mergedExpressions += f
- case _ =>
+ (newPlanFilter ++ cachedPlanFilter).foreach { f =>
Review Comment:
Agreed, restored `cachedPlanFilter.foreach(mergedExpressions += _)` in
[`0af2d628552`](https://github.com/apache/spark/pull/59294/commits/0af2d628552a7d90d8b39929dbfc9f32b79a684d).
The comment above `mergeNamedExpressions` now spells out the asymmetry.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/MergeSubplansSuite.scala:
##########
@@ -1981,6 +1981,63 @@ class MergeSubplansSuite extends PlanTest {
}
}
+ test("SPARK-60084: Reuse a filter that an earlier merge folded into a
post-join filter") {
Review Comment:
Added "Reuse a folded filter when the new plan has no Project above the
join" in
[`0af2d628552`](https://github.com/apache/spark/pull/59294/commits/0af2d628552a7d90d8b39929dbfc9f32b79a684d),
with `.select($"a")` dropped from `subquery3`. On master it fails with
`propagatedFilter_0` missing from `a, propagatedFilter_2, propagatedFilter_1,
b, c, d, e`, so it goes through `mergeNamedExpressions(np.output,
cp.projectList, ...)`.
The `(np: Project, cp)` arm is fixed by the same line, but I couldn't build
a case where it's the first to miss the filter. A non-`Project` cp (`Filter`,
`Join`) only lacks the reused filter when a `Project` below it dropped it.
Merging that lower `Project` already needs the append, so a test for this arm
fails one level lower on master, in an arm the two tests cover.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/PlanMergingSuite.scala:
##########
@@ -439,4 +439,42 @@ class PlanMergingSuite extends SharedSparkSession
}
}
}
+
+ test("SPARK-60084: Reuse a filter that an earlier merge folded into a
post-join filter") {
+ // Merging the 2nd subquery into the 1st aliases `l.b = 2` below the join,
and the 2nd
+ // subquery's post-join filter folds that alias into a new one, so only
the new alias is
+ // carried up to the aggregate. The 3rd subquery then reuses the `l.b = 2`
alias on its own.
+ Seq(false, true).foreach { enableAQE =>
+ withSQLConf(
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> enableAQE.toString,
+ SQLConf.MERGE_SUBPLANS_SYMMETRIC_FILTER_PROPAGATION_ENABLED.key ->
"true",
+ SQLConf.MERGE_SUBPLANS_FILTER_PROPAGATION_THROUGH_JOIN_ENABLED.key ->
"true",
+ SQLConf.OPTIMIZER_EXCLUDED_RULES.key ->
Review Comment:
It's needed. Without it the 3 subqueries end up in 2 `SubqueryExec`s. Added
the same comment as the other two tests in
[`0af2d628552`](https://github.com/apache/spark/pull/59294/commits/0af2d628552a7d90d8b39929dbfc9f32b79a684d).
--
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]
