cloud-fan commented on code in PR #57901:
URL: https://github.com/apache/spark/pull/57901#discussion_r3758099639
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PushdownPredicatesAndPruneColumnsForCTEDef.scala:
##########
@@ -156,6 +174,79 @@ object PushdownPredicatesAndPruneColumnsForCTEDef extends
Rule[LogicalPlan] {
}
}
+ /**
+ * Removes the previous push-down filter (identified by its condition,
`predicate`) from
+ * `plan`, wherever predicate push-down rules sharing the fixedPoint batches
with this rule
+ * (e.g. `PushDownPredicates`) may have moved it. The descent mirrors the
cases of
+ * `PushPredicateThroughNonJoin` and `PushPredicateThroughJoin`, translating
`predicate` back
+ * the same way they translate the pushed condition (for `Project` and
`Aggregate` it uses
+ * the very same `AliasHelper` utilities, so the two cannot drift apart):
through projection
+ * and grouping-key aliases, positionally into each branch (`Union`), and
unchanged
+ * through operators that pass the referenced attributes verbatim (`Join`,
`Window`, and
+ * output-preserving unary nodes like `Filter`, `Sort`, `Repartition`).
Descent stops at
+ * operators that remap attributes in other ways (e.g. `Generate`,
`Expand`): if the filter
+ * cannot be located, the input plan is returned unchanged, and the caller
re-pushes on top,
+ * which is redundant but always semantics-preserving.
+ */
+ private def removePushedDownFilter(plan: LogicalPlan, predicate:
Expression): LogicalPlan = {
+ def remove(current: LogicalPlan, target: Expression): (LogicalPlan,
Boolean) = current match {
+ case Filter(cond, inner) if cond.canonicalized == target.canonicalized =>
+ (inner, true)
+ case p: Project =>
+ // Mirror PushPredicateThroughNonJoin: translate the target through
the projection's
+ // aliases with the same helper it uses to move the filter below the
projection.
+ val translated = replaceAlias(target, getAliasMap(p))
+ val (newChild, removed) = remove(p.child, translated)
+ if (removed) (p.copy(child = newChild), true) else (p, false)
+ case j: Join =>
+ val (newLeft, removedFromLeft) = remove(j.left, target)
+ if (removedFromLeft) {
+ (j.copy(left = newLeft), true)
+ } else {
+ val (newRight, removedFromRight) = remove(j.right, target)
+ if (removedFromRight) (j.copy(right = newRight), true) else (j,
false)
+ }
+ case u: Union =>
+ // PushDownPredicates copies the filter into every branch, mapping the
union output
+ // attributes to each branch's output positionally; remove it from
every branch where
+ // it is found.
+ var removedAny = false
+ val newChildren = u.children.map { branch =>
+ val branchTarget = target.transform {
+ case a: Attribute if u.output.exists(_.exprId == a.exprId) =>
+ branch.output(u.output.indexWhere(_.exprId == a.exprId))
Review Comment:
**Non-blocking:**
Please build an ExprId-to-output-index map once before iterating the
branches. This currently scans `u.output` with both `exists` and `indexWhere`
for every referenced attribute in every branch, which becomes unnecessarily
quadratic for wide unions.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PushdownPredicatesAndPruneColumnsForCTEDef.scala:
##########
@@ -156,6 +174,79 @@ object PushdownPredicatesAndPruneColumnsForCTEDef extends
Rule[LogicalPlan] {
}
}
+ /**
+ * Removes the previous push-down filter (identified by its condition,
`predicate`) from
+ * `plan`, wherever predicate push-down rules sharing the fixedPoint batches
with this rule
+ * (e.g. `PushDownPredicates`) may have moved it. The descent mirrors the
cases of
+ * `PushPredicateThroughNonJoin` and `PushPredicateThroughJoin`, translating
`predicate` back
+ * the same way they translate the pushed condition (for `Project` and
`Aggregate` it uses
+ * the very same `AliasHelper` utilities, so the two cannot drift apart):
through projection
+ * and grouping-key aliases, positionally into each branch (`Union`), and
unchanged
+ * through operators that pass the referenced attributes verbatim (`Join`,
`Window`, and
+ * output-preserving unary nodes like `Filter`, `Sort`, `Repartition`).
Descent stops at
+ * operators that remap attributes in other ways (e.g. `Generate`,
`Expand`): if the filter
+ * cannot be located, the input plan is returned unchanged, and the caller
re-pushes on top,
+ * which is redundant but always semantics-preserving.
+ */
+ private def removePushedDownFilter(plan: LogicalPlan, predicate:
Expression): LogicalPlan = {
+ def remove(current: LogicalPlan, target: Expression): (LogicalPlan,
Boolean) = current match {
+ case Filter(cond, inner) if cond.canonicalized == target.canonicalized =>
+ (inner, true)
+ case p: Project =>
+ // Mirror PushPredicateThroughNonJoin: translate the target through
the projection's
+ // aliases with the same helper it uses to move the filter below the
projection.
+ val translated = replaceAlias(target, getAliasMap(p))
+ val (newChild, removed) = remove(p.child, translated)
+ if (removed) (p.copy(child = newChild), true) else (p, false)
Review Comment:
**Non-blocking:**
Could we rebuild these ancestors with `withNewChildren` (or explicitly copy
tags)? Direct case-class copies drop `TreeNode` tags; for example, a tagged
`Project` loses `hiddenOutputTag`. The same applies to the direct `Join`,
`Aggregate`, and `Window` copies, so a tagged-node regression case would make
the invariant explicit.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:
##########
@@ -2194,6 +2194,10 @@ object PushDownPredicates extends Rule[LogicalPlan] {
* 2) the predicate is deterministic and the operator will not change any of
rows.
* 3) We don't add double evaluation OR double evaluation would be cheap OR
we're configured to.
*
+ * Note: if a new push-through case is added here, or the translation applied
to pushed
+ * conditions changes (e.g. how aliases are substituted), also update
+ * [[PushdownPredicatesAndPruneColumnsForCTEDef.removePushedDownFilter]],
which mirrors this
Review Comment:
**Nit:**
Please link `[[PushdownPredicatesAndPruneColumnsForCTEDef]]` and render
`removePushedDownFilter` as code. The helper is private, so it is not a stable
generated-Scaladoc member target.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PushdownPredicatesAndPruneColumnsForCTEDef.scala:
##########
@@ -113,32 +114,49 @@ object PushdownPredicatesAndPruneColumnsForCTEDef extends
Rule[LogicalPlan] {
* In order to guarantee idempotency, we keep the predicates (if any) being
pushed down by the
* last iteration of this rule in a temporary field of `CTERelationDef`, so
that on the current
* iteration, we only push down predicates for a CTE def if there exists any
new predicate that
- * has not been pushed before. Also, since part of a new predicate might
overlap with some
- * existing predicate and it can be hard to extract only the non-overlapping
part, we also keep
- * the original CTE definition plan without any predicate push-down in that
temporary field so
- * that when we do a new predicate push-down, we can construct a new plan
with all latest
- * predicates over the original plan without having to figure out the exact
predicate difference.
+ * has not been pushed before. When such a new predicate push-down happens,
the CTE definition
+ * is rebuilt from its CURRENT child: the push-down filter this rule placed
in the previous
+ * iteration is removed (wherever it sits) and the result is wrapped with
the latest combined
+ * predicate. This preserves any change other rules made to the CTE
definition's child in
+ * between (e.g. filters injected by `InferFiltersFromConstraints`, which
runs in the `Once`
+ * batch sandwiched between the two fixedPoint batches containing this
rule). If the previous
+ * push-down can no longer be found (another rule rewrote or merged it with
other filters),
+ * the current child is used as-is: re-pushing the combined predicate is
redundant but always
+ * semantics-preserving, since the disjunction of the reference predicates
is valid for every
+ * row of the CTE definition.
*/
private def pushdownPredicatesAndAttributes(
plan: LogicalPlan,
cteMap: CTEMap): LogicalPlan = plan.transformWithSubqueries {
case cteDef @ CTERelationDef(child, id, originalPlanWithPredicates, _, _,
_) =>
val (_, _, newPreds, newAttrSet) = cteMap(id)
- val originalPlan = originalPlanWithPredicates.map(_._1).getOrElse(child)
val preds = originalPlanWithPredicates.map(_._2).getOrElse(Seq.empty)
if (!isTruePredicate(newPreds) &&
newPreds.exists(newPred =>
!preds.exists(_.semanticEquals(newPred)))) {
+ val basePlan = originalPlanWithPredicates match {
+ case Some((_, prevPreds)) if prevPreds.nonEmpty =>
+ // Remove the push-down filter this rule placed in the previous
iteration. It is
+ // usually the top-level node of the child, but rules sharing the
fixedPoint
+ // batches with this rule (e.g. `PushDownPredicates`) may have
moved it deeper,
+ // possibly across attribute-renaming projections - hence the
comparison is done
+ // on canonicalized conditions. If the previous push-down can no
longer be found
+ // (another rule rewrote or merged it with other filters), the
current child is
+ // used as-is: re-pushing the combined predicate is redundant but
always
+ // semantics-preserving, since the disjunction of the reference
predicates is
+ // valid for every row of the CTE definition.
+ removePushedDownFilter(child, prevPreds.reduce(Or))
+ case _ => child
+ }
val newCombinedPred = newPreds.reduce(Or)
- val newChild = if (needsPruning(originalPlan, newAttrSet)) {
- Project(newAttrSet.toSeq, originalPlan)
+ val newChild = if (needsPruning(basePlan, newAttrSet)) {
+ Project(newAttrSet.toSeq, basePlan)
} else {
- originalPlan
+ basePlan
}
cteDef.copy(child = Filter(newCombinedPred, newChild),
- originalPlanWithPredicates = Some((originalPlan, newPreds)))
+ originalPlanWithPredicates = Some((basePlan, newPreds)))
Review Comment:
**Nit:**
Please update the `CTERelationDef.originalPlanWithPredicates` parameter
documentation with this semantic change. It now stores the current base plan
after removing the previous pushed filter, not necessarily the original query
plan before predicate pushdown.
--
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]
