dongjoon-hyun commented on code in PR #59294:
URL: https://github.com/apache/spark/pull/59294#discussion_r4218812536
##########
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:
This is pre-existing, but since this PR now relies on the reuse branch: the
`existingNPFilter` lookup above matches any `Alias` in
`childProject.projectList` whose child equals `newNPCondition`, while this
branch keeps `cp.condition` (the OR) unchanged.
The Phase 2 comment below says this `Project` can also hold computed-column
aliases that are not filters. If np's condition canonically equals one of them,
that alias is reused as np's filter, but the OR filter never references it, so
rows satisfying only np's condition are dropped and np's aggregate silently
returns a wrong result.
Could we restrict the lookup to the filter aliases, i.e. the same predicate
Phase 2 uses?
```scala
case a: Alias if cp.condition.references.contains(a.toAttribute) &&
a.child.canonicalized == newNPCondition.canonicalized =>
```
##########
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:
This removes the only description of the reuse case from the result-type
contract. Could we keep a sentence that `newPlanFilter` (unlike
`cachedPlanFilter`) can be an attribute reused from a previous merge round,
which an enclosing `Project` may not output yet and therefore must re-append?
Otherwise the hazard behind SPARK-60084 is documented only in the
`mergeNamedExpressions` comment, and a future arm that builds a `Project` from
`mergedChild.output` has no warning.
##########
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:
The two existing tests in this file that exclude `ObjectSerializerPruning`
(L374, L410) explain why in a comment. Could we add the same comment here, or
drop the exclusion if it is not needed to reach the bug path?
##########
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:
This test covers the reused filter going missing at a `(np: Project, cp:
Project)` merge. Could we also cover the `(np, cp: Project)` / `(np: Project,
cp)` arms, whose base list is `np.output` / `cp.output` rather than a
projectList? For example, dropping `.select($"a")` from `subquery3` makes it
`Aggregate(Join)` vs. `Aggregate(Project(Project(Join)))`, so the reused
`propagatedFilter_0` must be appended through `mergeNamedExpressions(np.output,
cp.projectList, ...)`. This PR fixes that path too, but nothing guards it.
##########
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:
nit. `cachedPlanFilter` is always a freshly created alias, so the presence
check never matters for it. Treating both sides the same suggests the cached
side can be reused too, which blurs the asymmetry a maintainer needs to
understand when changing the reuse logic. Either keep
`cachedPlanFilter.foreach(mergedExpressions += _)` or mention that the check is
only defensive for the cached side.
--
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]
