uros-b commented on code in PR #57975:
URL: https://github.com/apache/spark/pull/57975#discussion_r3791448862
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/FilterPushdownSuite.scala:
##########
@@ -1645,6 +1645,43 @@ class FilterPushdownSuite extends PlanTest {
comparePlans(optimizedQueryWithoutStep, correctAnswer)
}
+ test("SPARK-58627: do not push down predicate with a throwing child of
sequence through joins") {
+ val x = testStringRelation.subquery("x")
+ val y = testRelation1.subquery("y")
+
+ // Sequence overrides `throwable` for its step check, so it also has to
fall back to its
+ // children. Without that fallback a RaiseError under a stepless sequence
reports
+ // non-throwable and the predicate gets pushed below the join.
+ val raiseErrorInt = RaiseError(
+ Literal("USER_RAISED_EXCEPTION"),
+ CreateMap(Seq(Literal("errorMessage"), $"x.e")),
+ IntegerType)
+ val queryWithRaiseError = x.join(y, joinType = Inner, condition =
Some($"x.a" === $"y.d"))
+ .where(IsNotNull(Sequence($"x.a", raiseErrorInt, None)))
+ .analyze
+ comparePlans(Optimize.execute(queryWithRaiseError), queryWithRaiseError)
+ }
+
+ test("SPARK-58627: do not combine predicate with raise_error with other
filters") {
+ val x = testStringRelation.subquery("x")
+
+ // Do not combine. Two stacked Filters pin raise_error above the inner
predicate, while a
+ // single merged And does not: execution does not guarantee the conjuncts
are evaluated in
+ // order, and later rules are free to re-split and relocate them
independently. Either way
+ // raise_error can end up evaluated on rows the inner filter would have
removed.
+ val queryWithRaiseError = x.where($"x.a" > 1)
+ .where(IsNull(RaiseError($"x.e")))
+ .analyze
+ comparePlans(Optimize.execute(queryWithRaiseError), queryWithRaiseError)
+
+ // The same shape without raise_error is combined into a single filter.
+ val queryWithoutRaiseError = x.where($"x.a" > 1)
+ .where(IsNotNull($"x.e"))
+ .analyze
+ val correctAnswer = x.where(IsNotNull($"x.e") && $"x.a" > 1).analyze
+ comparePlans(Optimize.execute(queryWithoutRaiseError), correctAnswer)
+ }
+
Review Comment:
The PR motivating example (WHERE raise_error('boom') IS NULL over a join)
has no corresponding unit test. The two new tests cover CombineFilters and a
Sequence-wrapped path through PushPredicateThroughJoin, but not the direct
RaiseError in a join-predicate scenario. A bare RaiseError(Literal(...)) has no
column references, so pre-fix references.subsetOf(left.outputSet) is vacuously
true and the predicate is pushed; after the fix throwable=true blocks it. This
case should be explicitly tested.
--
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]
