viirya commented on code in PR #59944:
URL: https://github.com/apache/spark/pull/59944#discussion_r4421487031
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:
##########
@@ -2842,6 +2833,22 @@ object PushPredicateThroughNonJoin extends
Rule[LogicalPlan] with PredicateHelpe
}
}
+ /**
+ * Splits `condition` into the conjuncts to push below an operator and the
conjuncts to keep
+ * above it, where `canPush` tells whether the operator lets a conjunct
through. Every conjunct
+ * written before the first one that cannot be pushed is pushed. After it,
only the conjuncts
+ * that are not throwable (e.g. `raise_error`) are pushed, so a throwable
conjunct is not
+ * evaluated on rows the conjuncts before it would have filtered out. Moving
a non-throwable
Review Comment:
Agreed. This PR only covers expressions marked `throwable`, and the
description now says so. Whether ANSI errors should count as well is a wider
question about evaluation order, beyond this rule, so a separate JIRA fits it
better.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:
##########
@@ -2690,15 +2690,16 @@ object PushPredicateThroughNonJoin extends
Rule[LogicalPlan] with PredicateHelpe
// We can push down deterministic predicate through Aggregate, including
throwable predicate.
// If we can push down a filter through Aggregate, it means the filter
only references the
// grouping keys or constants. The Aggregate operator can't reduce
distinct values of grouping
- // keys so the filter won't see any new data after push down.
+ // keys so the filter won't see any new data after push down. A throwable
predicate written
+ // after one that stays up is not pushed, though (see
`splitPushableConjuncts`).
case filter @ Filter(condition, aggregate: Aggregate)
if aggregate.aggregateExpressions.forall(_.deterministic)
&& aggregate.groupingExpressions.nonEmpty =>
val aliasMap = getAliasMap(aggregate)
// For each filter, expand the alias and check if the filter can be
evaluated using
// attributes produced by the aggregate operator's child operator.
- val (pushDown, stayUp) = splitConjunctivePredicates(condition).partition
{ cond =>
+ val (pushDown, stayUp) = splitPushableConjuncts(condition) { cond =>
val replaced = replaceAlias(cond, aliasMap)
cond.deterministic && cond.references.nonEmpty &&
replaced.references.subsetOf(aggregate.child.outputSet)
Review Comment:
Agreed, it is the same problem for the condition of a left semi or anti
join, and the `Generate` example shows it can be hit. I'd keep it for a
separate JIRA so this PR stays on `PushPredicateThroughNonJoin`. If it gets
fixed, `splitPushable` and `canDropRows` could move to `PredicateHelper` so
both rules share them.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:
##########
@@ -2842,6 +2833,22 @@ object PushPredicateThroughNonJoin extends
Rule[LogicalPlan] with PredicateHelpe
}
}
+ /**
+ * Splits `condition` into the conjuncts to push below an operator and the
conjuncts to keep
+ * above it, where `canPush` tells whether the operator lets a conjunct
through. Every conjunct
Review Comment:
Confirmed with a table partitioned by `p`: the query fails while listing the
files. Since it is outside the optimizer, I'd leave it for a separate JIRA too,
together with item 7.
--
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]
