peter-toth commented on code in PR #59321:
URL: https://github.com/apache/spark/pull/59321#discussion_r4297597668
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/MergeSubplansSuite.scala:
##########
@@ -2597,6 +2653,70 @@ class MergeSubplansSuite extends PlanTest {
}
}
+ gridTest("SPARK-60106: Merge over a merged DSv2 filter on a join child")(
+ for {
+ v2OnLeft <- Seq(true, false)
+ shape <- Seq("dropped", "extended", "reused", "firstMerge")
+ } yield (v2OnLeft, shape)) { case (v2OnLeft, shape) =>
+ // sub1 and sub2 filter the DSv2 scan differently and merge under the DSv2
exemption into a
+ // merged Filter. A subplan that also filters the other join child would
join a child that the
+ // cached plans filter with a child that it filters, so without symmetric
filter propagation
+ // it stays apart. The shapes:
+ // - dropped: sub3 filters only the other child, so its merge drops the
merged Filter.
+ // - extended: sub3 filters the DSv2 scan too, so its merge extends the
merged Filter.
+ // - reused: sub3 repeats sub1's filter on the DSv2 scan, so sub1 covers
sub3's rows there and
+ // the merge builds no extra pairs. It merges without symmetric filter
propagation.
+ // - firstMerge: sub2 itself filters the other child too, so the first
merge declines.
+ // The DSv2 scan is tried as both the left and the right join child.
+ def join(v2: LogicalPlan, other: LogicalPlan): LogicalPlan = if (v2OnLeft)
{
+ v2.join(other, Inner, Some($"a" === $"d"))
+ } else {
+ other.join(v2, Inner, Some($"a" === $"d"))
+ }
+ val sub1 = ScalarSubquery(
+ join(v2ScanReading("a", "b").where($"a" > 1), testRelation2)
+ .groupBy()(sum($"a").as("sum_a")))
+ val other2 = if (shape == "firstMerge") testRelation2.where($"e" > 3) else
testRelation2
+ val sub2 = ScalarSubquery(
+ join(v2ScanReading("a", "b").where($"b" > 2), other2)
+ .groupBy()(max($"a").as("max_a")))
+ val v2In3 = shape match {
+ case "extended" => v2ScanReading("a", "b").where($"a" < 0)
+ case "reused" => v2ScanReading("a", "b").where($"a" > 1)
+ case _ => v2ScanReading("a", "b")
+ }
+ val sub3 = ScalarSubquery(
+ join(v2In3, testRelation2.where($"e" > 3))
+ .groupBy()(min($"a").as("min_a")))
+ val originalQuery = if (shape == "firstMerge") {
+ testRelation.select(sub1, sub2)
+ } else {
+ testRelation.select(sub1, sub2, sub3)
+ }
+
+ def mergedAggregates(symmetric: Boolean): Seq[Seq[String]] = withSQLConf(
Review Comment:
The `reused` case now also checks each aggregate's `FILTER`, with the
propagated aliases inlined. For example, `min_a` has to get `((a > 1) AND (e >
3))`. It fails if the reuse picks the wrong alias
([0fb8bcce413](https://github.com/apache/spark/pull/59321/commits/0fb8bcce413b83ac6c4ab82e9a8e30e513f2d271)).
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala:
##########
@@ -430,27 +448,30 @@ class PlanMerger(
(newPlan, cachedPlan) match {
case (np: Project, cp: Project) =>
tryMergePlans(np.child, cp.child, context).map {
- case TryMergeResult(mergedChild, npMapping, npFilter, cpFilter,
deferred, dsv2Merged) =>
+ case TryMergeResult(mergedChild, npMapping, npFilter, cpFilter,
deferred, dsv2Merged,
Review Comment:
Done, the three Project arms use `copy` now
([0fb8bcce413](https://github.com/apache/spark/pull/59321/commits/0fb8bcce413b83ac6c4ab82e9a8e30e513f2d271)).
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala:
##########
@@ -660,18 +688,23 @@ class PlanMerger(
mergedChild.output.toList :+ newCPFilterAlias,
mergedChild)
TryMergeResult(project, npMapping, npFilter, Some(newCPFilter),
- dsv2Merged = dsv2Merged)
+ dsv2Merged = dsv2Merged, cachedSideFiltered = true)
}
}
case (np: Join, cp: Join) if np.joinType == cp.joinType && np.hint ==
cp.hint =>
tryMergePlans(np.left, cp.left, context.copy(filterAboveScan =
false)).flatMap {
case TryMergeResult(mergedLeft, leftNPMapping, leftNPFilter,
leftCPFilter, _,
- leftDsv2Merged) =>
+ leftDsv2Merged, leftCPSideFiltered) =>
tryMergePlans(np.right, cp.right, context.copy(filterAboveScan =
false)).flatMap {
case TryMergeResult(mergedRight, rightNPMapping,
rightNPFilter, rightCPFilter, _,
- rightDsv2Merged)
- if // Gate join-crossing filter propagation behind its own
config flag.
+ rightDsv2Merged, rightCPSideFiltered)
+ if // When the two plans filter different children, the
merge is symmetric
+ // filter propagation and needs its config, see the
class doc.
+ (symmetricFilterPropagationEnabled ||
Review Comment:
I added parentheses around the two `&&` terms. I'd keep the check in the
guard, because it only protects this case, like the other clauses there with
their own comments
([0fb8bcce413](https://github.com/apache/spark/pull/59321/commits/0fb8bcce413b83ac6c4ab82e9a8e30e513f2d271)).
--
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]
