dongjoon-hyun commented on code in PR #59944:
URL: https://github.com/apache/spark/pull/59944#discussion_r4419194752
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/FilterPushdownSuite.scala:
##########
@@ -750,6 +750,120 @@ class FilterPushdownSuite extends PlanTest {
assertTopFilterConjuncts(Optimize.execute(originalQuery),
Seq(nonDeterministic, unplaced))
}
+ // Raises for the rows with a <= 5, so it must only see the rows the
conditions before it kept.
+ private val throwableOnA = attrA > 5 || IsNull(RaiseError(Literal("boom")))
+
+ test("SPARK-60633: a throwable condition is not pushed through a Python UDF
evaluation " +
+ "ahead of the conditions before it") {
+ Seq(PythonEvalType.SQL_BATCHED_UDF,
PythonEvalType.SQL_SCALAR_PANDAS_UDF).foreach { evalType =>
+ withClue(s"eval type $evalType: ") {
+ val pythonUDF = PythonUDF(
+ "pythonUDF", null, IntegerType, Seq(attrA), evalType,
udfDeterministic = true)
+ val f = AttributeReference("f", IntegerType)()
+ def evalPython(child: LogicalPlan): LogicalPlan =
+ if (evalType == PythonEvalType.SQL_BATCHED_UDF) {
+ BatchEvalPython(Seq(pythonUDF), Seq(f), child)
+ } else {
+ ArrowEvalPython(Seq(pythonUDF), Seq(f), child, evalType)
+ }
+
+ // f > 5 cannot be pushed, so the throwable condition after it stays
above too. b > 1
+ // is still pushed, as that only reduces the rows the throwable
condition sees.
+ val originalQuery = Filter(f > 5 && throwableOnA && attrB > 1,
evalPython(testRelation))
+ val correctAnswer =
+ Filter(f > 5 && throwableOnA, evalPython(Filter(attrB > 1,
testRelation)))
+ comparePlans(Optimize.execute(originalQuery), correctAnswer)
+
+ // With nothing left above before it, a throwable condition is still
pushed.
+ val throwableFirst = Filter(throwableOnA && f > 5,
evalPython(testRelation))
+ val throwableFirstAnswer = Filter(f > 5,
evalPython(Filter(throwableOnA, testRelation)))
+ comparePlans(Optimize.execute(throwableFirst), throwableFirstAnswer)
+
+ // The conditions left above keep their written order.
+ val nonDeterministic = Rand(10) > 0.5
+ val mixed = Filter(nonDeterministic && f > 5 && throwableOnA,
evalPython(testRelation))
+ assertTopFilterConjuncts(
+ Optimize.execute(mixed), Seq(nonDeterministic, f > 5, throwableOnA))
+ }
+ }
+ }
+
+ test("SPARK-60633: a throwable condition is not pushed through a sort ahead
of a " +
+ "non-deterministic condition before it") {
+ val originalQuery = testRelation.orderBy($"b".asc)
+ .where(Rand(10) > 0.5 && throwableOnA && $"b" > 1)
+ .analyze
+ val Seq(nonDeterministic, throwable, _) = splitConjunctivePredicates(
+ originalQuery.asInstanceOf[Filter].condition)
+ val correctAnswer = testRelation.where($"b" > 1).orderBy($"b".asc)
+ .where(Rand(10) > 0.5 && throwableOnA)
+ .analyze
+
+ val optimized = Optimize.execute(originalQuery)
+ comparePlans(optimized, correctAnswer)
+ assertTopFilterConjuncts(optimized, Seq(nonDeterministic, throwable))
+ }
+
+ test("SPARK-60633: a throwable condition is not pushed through an aggregate
ahead of the " +
+ "conditions before it") {
+ val originalQuery = testRelation.groupBy($"a")($"a", count($"b") as "c")
+ .where($"c" > 1 && throwableOnA && $"a" > 1)
+ .analyze
+ val correctAnswer = testRelation.where($"a" > 1).groupBy($"a")($"a",
count($"b") as "c")
+ .where($"c" > 1 && throwableOnA)
+ .analyze
+ comparePlans(Optimize.execute(originalQuery), correctAnswer)
+
+ // With nothing left above before it, a throwable condition is still
pushed.
+ val throwableFirst = testRelation.groupBy($"a")($"a", count($"b") as "c")
+ .where(throwableOnA && $"c" > 1)
+ .analyze
+ val throwableFirstAnswer = testRelation.where(throwableOnA)
+ .groupBy($"a")($"a", count($"b") as "c")
+ .where($"c" > 1)
+ .analyze
+ comparePlans(Optimize.execute(throwableFirst), throwableFirstAnswer)
+ }
+
+ test("SPARK-60633: a throwable condition is not pushed through a window
ahead of the " +
+ "conditions before it") {
+ val winExpr = windowExpr(count($"b"),
+ windowSpec($"a" :: Nil, $"b".asc :: Nil, UnspecifiedFrame))
+ val originalQuery = testRelation
+ .window(winExpr.as("window") :: Nil, $"a" :: Nil, $"b".asc :: Nil)
+ .where($"window" > 1 && throwableOnA && $"a" > 1)
+ .analyze
+ val correctAnswer = testRelation.where($"a" > 1)
+ .window(winExpr.as("window") :: Nil, $"a" :: Nil, $"b".asc :: Nil)
+ .where($"window" > 1 && throwableOnA)
+ .analyze
+ comparePlans(Optimize.execute(originalQuery), correctAnswer)
Review Comment:
**2. [Low] No test checks the order of the conditions left above a `Window`**
The description says that the conditions left above the operator in the
generic and `Window` cases now keep their written order. The `Sort` test checks
this for the generic case, but this test uses `comparePlans`, which sorts the
conjuncts, and has no non-deterministic condition. So it would still pass if
the `Window` case went back to `rest ++ nonDeterministic`. The `mixed` check in
the Python UDF test does not cover it either: nothing is pushed there, so the
rule returns the filter unchanged. Something like this would cover it:
```scala
val withRand = testRelation
.window(winExpr.as("window") :: Nil, $"a" :: Nil, $"b".asc :: Nil)
.where(Rand(10) > 0.5 && $"window" > 1 && $"a" > 1)
.analyze
val Seq(nonDeterministic, stayUp, _) = splitConjunctivePredicates(
withRand.asInstanceOf[Filter].condition)
assertTopFilterConjuncts(Optimize.execute(withRand),
Seq(nonDeterministic, stayUp))
```
Without this PR, the top filter would be `window > 1` and then `rand(10) >
0.5`.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:
##########
@@ -2724,15 +2725,10 @@ object PushPredicateThroughNonJoin extends
Rule[LogicalPlan] with PredicateHelpe
if w.partitionSpec.forall(_.isInstanceOf[AttributeReference]) =>
val partitionAttrs = AttributeSet(w.partitionSpec.flatMap(_.references))
- val (candidates, nonDeterministic) =
- splitConjunctivePredicates(condition).partition(_.deterministic)
-
- val (pushDown, rest) = candidates.partition { cond =>
- cond.references.subsetOf(partitionAttrs)
+ val (pushDown, stayUp) = splitPushableConjuncts(condition) { cond =>
+ cond.deterministic && cond.references.subsetOf(partitionAttrs)
}
Review Comment:
**3. [Nit] The comment on the `Window` case still lists the old condition 3**
The comment above this case (L2719-L2723) says that a condition must be
"Placed before any non-deterministic predicates" to be pushed. That was not
true before this PR either, since every deterministic condition on the
partition keys was pushed, and it is still not true: `a > 1` is pushed when it
is written after `rand(10) > 0.5`. The rule that this PR adds, that a throwable
condition written after one that stays up is not pushed, is not mentioned.
Could it say, like the new sentence on the `Aggregate` case:
```scala
// 1. All the expressions are part of window partitioning key. The
expressions can be compound.
// 2. Deterministic.
// 3. Not throwable, if a predicate written before it stays up (see
`splitPushableConjuncts`).
```
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:
##########
@@ -2821,15 +2817,10 @@ object PushPredicateThroughNonJoin extends
Rule[LogicalPlan] with PredicateHelpe
// come from grandchild.
// TODO: non-deterministic predicates could be pushed through some
operators that do not change
// the rows.
- val (candidates, nonDeterministic) =
- splitConjunctivePredicates(filter.condition).partition(_.deterministic)
-
- val (pushDown, rest) = candidates.partition { cond =>
- cond.references.subsetOf(grandchild.outputSet)
+ val (pushDown, stayUp) = splitPushableConjuncts(filter.condition) { cond =>
+ cond.deterministic && cond.references.subsetOf(grandchild.outputSet)
}
Review Comment:
**1. [Medium] A leading throwable condition is still pushed below a
`Generate` that drops rows**
`splitPushableConjuncts` still pushes a throwable condition that has only
pushable conditions before it. The comment on the `Aggregate` case gives the
reason why that is safe there: the filter won't see any new data after push
down. That holds for the operators of `canPushThrough` that keep every row, but
not for a `Generate` with `outer = false`, which drops an input row whose array
is empty or null. Below it, the throwable condition runs on the rows that the
generator would have dropped. For example:
```scala
spark.range(10)
.select($"id", explode(filter(array($"id"), x => x > 5)) as "x")
.where(assert_true($"id" > 5).isNull)
```
The `Generate` only emits the rows with `id > 5`, so the filter as written
never raises. But the condition is the first one and only references `id`, so
it is pushed below the `Generate`, and the query fails with
`USER_RAISED_EXCEPTION` on `id` 0 to 5. The same holds for the other generators
that can emit no row for an input row, such as `posexplode` and `inline`, and
possibly for a `ScriptTransformation` whose script drops rows, with a condition
that references no column (`raise_error('x') IS NULL`).
The description lists `Generate` among the fixed operators, but the
`Generate` query in `DataFrameSuite` uses `explode(array($"id"))`, which emits
exactly one row per input row, so it does not cover this. Could the generic
case keep throwable conditions above the operators that can drop rows? For
example, `canPush` could also require `!cond.throwable` when `u` is a
`Generate` with `outer = false` or a `ScriptTransformation`. A throwable
condition then stops the `span` like any other condition that stays up.
##########
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
+ * conjunct below a throwable one is fine, as it can only reduce the rows
the throwable one
+ * sees. The conjuncts to keep above are returned in the order they were
written.
+ */
+ private def splitPushableConjuncts(condition: Expression)(
+ canPush: Expression => Boolean): (Seq[Expression], Seq[Expression]) = {
+ val (leading, rest) = splitConjunctivePredicates(condition).span(canPush)
+ val (pushedRest, stayUp) = rest.partition(cond => canPush(cond) &&
!cond.throwable)
Review Comment:
**5. [Nit] `canPush` runs twice on the first condition that stays up**
`span` stops at the first condition for which `canPush` is false, and
`partition` then calls `canPush` on that condition again. In the `Aggregate`
case, that includes a `replaceAlias`. Since the head of `rest` is known to stay
up, the split could skip it:
```scala
val (leading, rest) = splitConjunctivePredicates(condition).span(canPush)
val (pushedRest, stayUp) = rest.drop(1).partition(c => canPush(c) &&
!c.throwable)
(leading ++ pushedRest, rest.take(1) ++ stayUp)
```
##########
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:
**6. [Low] Errors under ANSI mode are not `throwable`**
This is not caused by this PR. `Expression.throwable` is only set for
`raise_error` (which `assert_true` is replaced with), `sequence` with a step,
and some shapes of `JsonArray` and `JsonObjectExpr`. An expression that fails
under ANSI mode, the default since 4.0, is not marked, so the same failure
remains for it:
```scala
spark.range(10).groupBy($"id").agg(max($"id") as "m")
.where($"m" > 5 && lit(10) / ($"id" - 5) > 0)
```
`m > 5` keeps no row with `id = 5`, but the division only references the
grouping key, so it is pushed below the `Aggregate` and fails with
`DIVIDE_BY_ZERO`. I understand that `CombineFilters` says the optimizer does
not guarantee the order of the conditions, and that this PR only aims at
`throwable` expressions. Still, since the description says that queries like
these now succeed, it may help to say there that this only covers the
expressions marked `throwable`. Whether the errors of ANSI mode should count as
well could be a separate JIRA.
##########
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
+ * conjunct below a throwable one is fine, as it can only reduce the rows
the throwable one
+ * sees. The conjuncts to keep above are returned in the order they were
written.
+ */
+ private def splitPushableConjuncts(condition: Expression)(
+ canPush: Expression => Boolean): (Seq[Expression], Seq[Expression]) = {
Review Comment:
**4. [Nit] The `Project` case implements the same split inline**
The `Project` case (L2635-L2641) does the same `span` and then `partition`
with `!throwable` over its `(condition, used aliases, replaced)` tuples, only
checking `replaced.throwable` instead of `cond.throwable`. If this helper took
the elements and the throwable check as parameters, both cases could use it,
and a later change to the rule (item 1, for example) would only have to be made
in one place:
```scala
private def splitPushable[T](conds: Seq[T])(
canPush: T => Boolean, throwable: T => Boolean): (Seq[T], Seq[T]) = {
val (leading, rest) = conds.span(canPush)
val (pushedRest, stayUp) = rest.partition(c => canPush(c) &&
!throwable(c))
(leading ++ pushedRest, stayUp)
}
```
##########
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:
**8. [Low] Partition pruning evaluates a throwable condition on every
partition**
This is not caused by this PR either, and it is outside the optimizer. To
prune partitions,
[`DataSourceStrategy.getPushedDownFilters`](https://github.com/apache/spark/blob/5c0e6eef31576f8e47f5bef5852de43a8f3e7ec3/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSourceStrategy.scala#L587-L599)
for a path-based relation and
[`DataSourceUtils.getPartitionFiltersAndDataFilters`](https://github.com/apache/spark/blob/5c0e6eef31576f8e47f5bef5852de43a8f3e7ec3/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSourceUtils.scala#L380-L389)
for a catalog table take every condition that only references partition
columns, whatever its position and whether it is `throwable`. The file index
then evaluates them on the values of every partition. For a table partitioned
by `p`, written with `p` equal to `id`:
```scala
spark.read.parquet(path).where($"id" > 5 && assert_true($"p" > 5).isNull)
```
`id > 5` keeps no row with `p` up to 5, but listing the files evaluates
`assert_true(p > 5)` on those partitions and raises. If this helper moved to
`PredicateHelper`, these two places and item 7 could follow the same rule.
##########
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:
**7. [Low] `PushDownLeftSemiAntiJoin` pushes through the same operators
without looking at `throwable`**
This is not caused by this PR either. `PushDownLeftSemiAntiJoin` has the
same cases as this rule for the condition of a left semi or anti join:
`Aggregate` (this case), `Window`, `Union`, and the operators of
`canPushThrough`, which it calls. Its
[`pushDownJoin`](https://github.com/apache/spark/blob/5c0e6eef31576f8e47f5bef5852de43a8f3e7ec3/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PushDownLeftSemiAntiJoin.scala#L132-L141)
splits the condition with `partition` and never looks at `throwable`. So a
throwable condition of the join can run on rows that it would not see
otherwise. With the `Generate` of item 1:
```scala
val l = spark.range(10)
.select($"id", explode(filter(array($"id"), x => x > 5)) as "x")
l.join(spark.range(10).toDF("rid"), $"id" === $"rid" && assert_true($"id" >
5).isNull,
"left_semi")
```
`PushPredicateThroughJoin` keeps the throwable condition in the join, and
then the join is pushed below the `Generate`, where the condition sees the
matches of `id` 0 to 5. The ordering issue that this PR fixes is narrower
there, since `PushPredicateThroughJoin` already moves the non-throwable
conditions on the left side of a semi join into the left side, and an anti join
is only pushed when its whole condition can be.
--
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]
