viirya commented on code in PR #59848:
URL: https://github.com/apache/spark/pull/59848#discussion_r4404109261
##########
sql/core/src/test/scala/org/apache/spark/sql/DataFrameSuite.scala:
##########
@@ -2905,6 +2905,29 @@ class DataFrameSuite extends SharedSparkSession
}
}
+ test("SPARK-60602: a throwable filter condition is not evaluated ahead of
its order") {
+ val f = udf((i: Long) => i)
+ val g = udf((i: Long) => i)
+ // `assert_true(g > 5)` only holds for the ids the first conjunct (`f + g
> 10`) keeps, so the
+ // `AND` short-circuit must keep it away from the other rows however the
filter is pushed.
+ def splitQuery(): DataFrame = spark.range(0, 10, 1, 1)
+ .select($"id", f($"id") as "f", g($"id") as "g")
+ .where($"f" + $"g" > 10 && assert_true($"g" > 5).isNull)
Review Comment:
Good catch. Both queries now also run with an extra `$"id" >= 0` conjunct
across the whole matrix. With the non-splittable conditions emitted first, the
split variant fails with `USER_RAISED_EXCEPTION`. I also added `SPARK-60602:
the conditions left above the projection keep their written order`, which
checks the order at the plan level instead of relying on the next pass
re-pushing the condition.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:
##########
@@ -2618,22 +2618,25 @@ object PushPredicateThroughNonJoin extends
Rule[LogicalPlan] with PredicateHelpe
val (replaced, usedAliases) = replaceAliasWhileTracking(cond,
aliasMap)
(cond, usedAliases, replaced)
}
+ // A component is cheap if it uses nothing expensive from the
projection: either it
+ // uses nothing at all, or only inexpensive items. Then we can push it
because it might
+ // eliminate more data quickly and it may also be something which
could be evaluated at
+ // the storage layer. We may wish to improve this heuristic in the
future.
+ def isCheap(used: AttributeMap[Alias]): Boolean =
!used.exists(_._2.child.expensive)
+ val firstExpensive =
originalAndRewrittenConditionsWithUsedAlias.indexWhere {
+ case (_, used, _) => !isCheap(used)
+ }
// Split the filter's components into cheap and expensive while
keeping track of
- // what each references from the projection.
- val (cheapWithUsed, expensiveWithUsed) =
originalAndRewrittenConditionsWithUsedAlias
- .partition { case (cond, used, replaced) =>
- // Didn't use anything? We're good
- if (used.isEmpty) {
- true
- } else if (!used.exists(_._2.child.expensive)) {
- // If it's cheap we can push it because it might eliminate more
data quickly and
- // it may also be something which could be evaluated at the
storage layer.
- // We may wish to improve this heuristic in the future.
- true
- } else {
- false
- }
+ // what each references from the projection. A throwable component
(e.g. `raise_error`)
+ // must not be evaluated on rows the components before it would have
filtered out, so
+ // it is only pushed when no expensive component, which stays above,
comes before it.
+ val (cheapWithIndex, expensiveWithIndex) =
originalAndRewrittenConditionsWithUsedAlias
+ .zipWithIndex
+ .partition { case ((cond, used, _), idx) =>
+ isCheap(used) && (!cond.throwable || firstExpensive < 0 || idx <
firstExpensive)
Review Comment:
Done, through the `span` version, which checks `replaced.throwable`. I added
the variant to the end-to-end matrix. It only runs with codegen on, or with
`avoidDoubleFilterEval` off. An interpreted projection computes `t` for every
input row whatever the filter does, so with codegen off and the filter above
the projection it raises regardless of this change.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:
##########
@@ -2644,15 +2647,20 @@ object PushPredicateThroughNonJoin extends
Rule[LogicalPlan] with PredicateHelpe
// Case 3: split the projection around the expensive conditions.
Non-deterministic
// conditions stay in the top filter, as they do everywhere else in
this rule -- the
// split orders conditions by cost, and cost is no way to decide how
many rows a
- // non-deterministic expression sees.
+ // non-deterministic expression sees. Throwable conditions stay there
too: ordering by
+ // cost could evaluate them ahead of the conditions written before
them.
val (splitChild, splitAliases, stayUp) =
if (expensiveWithUsed.nonEmpty &&
SQLConf.get.splitProjectionForExpensiveFilters) {
- val (deterministic, nonDeterministic) =
- expensiveWithUsed.partition { case (cond, _, _) =>
cond.deterministic }
+ val (splittable, notSplittable) = expensiveWithUsed.partition {
+ case (cond, _, _) => cond.deterministic && !cond.throwable
Review Comment:
Done. I added a plan test and your query to the end-to-end matrix, with the
same codegen caveat as [1]. I agree to leave the case where `unitOf` pulls in a
throwable alias.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:
##########
@@ -2618,22 +2618,25 @@ object PushPredicateThroughNonJoin extends
Rule[LogicalPlan] with PredicateHelpe
val (replaced, usedAliases) = replaceAliasWhileTracking(cond,
aliasMap)
(cond, usedAliases, replaced)
}
+ // A component is cheap if it uses nothing expensive from the
projection: either it
+ // uses nothing at all, or only inexpensive items. Then we can push it
because it might
+ // eliminate more data quickly and it may also be something which
could be evaluated at
+ // the storage layer. We may wish to improve this heuristic in the
future.
+ def isCheap(used: AttributeMap[Alias]): Boolean =
!used.exists(_._2.child.expensive)
+ val firstExpensive =
originalAndRewrittenConditionsWithUsedAlias.indexWhere {
Review Comment:
Done, thanks.
--
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]
