This is an automated email from the ASF dual-hosted git repository.
asf-gitbox-commits pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 7f10fc410716 [SPARK-58627][SQL] Mark raise error Throwable & fix
sequence throwable
7f10fc410716 is described below
commit 7f10fc4107168747f4b0599fc35e2e4e7a317519
Author: Holden Karau <[email protected]>
AuthorDate: Fri Aug 21 12:32:27 2026 -0700
[SPARK-58627][SQL] Mark raise error Throwable & fix sequence throwable
### What changes were proposed in this pull request?
Corrects the `Expression.throwable` metadata on two expressions:
1. Overrides `throwable` to `true` on `RaiseError`.
2. Makes `Sequence`'s existing override fall back to its children rather
than discarding the inherited default.
`Expression.throwable` (added in SPARK-46707) is opt-in metadata that tells
the optimizer an expression may raise a runtime error, so a predicate
containing it must not be relocated to a position where it runs on rows the
original plan would not have evaluated it on. `RaiseError` always throws when
evaluated but never declared the flag, so it inherited
`children.exists(_.throwable)` -- false for the usual case of a literal error
class and parameters.
`Sequence` was the only expression in the tree overriding the flag, and it
did so as `stepOpt.isDefined`, dropping the inherited
`children.exists(_.throwable)` term entirely. A throwing child under a stepless
`sequence(...)` therefore reported non-throwable, which would also have masked
the new flag on `RaiseError`.
### Why are the changes needed?
Without the flag, `CombineFilters` and `PushPredicateThroughJoin` treat a
predicate containing `raise_error` (or `assert_true`, which is rewritten to
`If(cond, null, RaiseError(...))`) as freely movable. Pushing such a predicate
below a selective join, or merging it into a filter that would have removed the
offending rows, can make a query fail at runtime that previously succeeded.
The join below matches no rows, so the predicate should never be evaluated.
But the predicate has no column references, so
`references.subsetOf(left.outputSet)` holds trivially and it is pushed onto the
left side, where it fires on the first row of `t1`:
```sql
CREATE OR REPLACE TEMP VIEW t1 AS SELECT * FROM VALUES (1), (2), (3) AS
t(a);
CREATE OR REPLACE TEMP VIEW t2 AS SELECT * FROM VALUES (4), (5), (6) AS
t(b);
SELECT * FROM t1 JOIN t2 ON t1.a = t2.b WHERE raise_error('boom') IS NULL;
-- [USER_RAISED_EXCEPTION] boom SQLSTATE: P0001
```
With this change the query returns an empty result.
### Does this PR introduce _any_ user-facing change?
Yes, as a bug fix. A predicate containing `raise_error` or `assert_true` is
no longer pushed through a join, pushed into a join condition, or combined with
an adjacent filter, so the error is raised only on the rows the unoptimized
plan would have evaluated it on. Queries that previously failed spuriously now
succeed. The same now holds for a throwing expression nested under a stepless
`sequence(...)`.
### How was this patch tested?
Added UTs:
- `MiscExpressionsSuite`: asserts the flag on `RaiseError`
- `FilterPushdownSuite`: a `raise_error` predicate is not pushed through a
join and not combined with an adjacent filter, each paired with a non-throwing
predicate of the same shape that still is.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 5) human review and cleanup after by Holden
Closes #57975 from holdenk/SPARK-58627-mark-raise-error-throwable.
Lead-authored-by: Holden Karau <[email protected]>
Co-authored-by: Holden Karau <[email protected]>
Signed-off-by: Holden Karau <[email protected]>
---
.../expressions/collectionOperations.scala | 4 +-
.../spark/sql/catalyst/expressions/misc.scala | 1 +
.../expressions/MiscExpressionsSuite.scala | 4 ++
.../catalyst/optimizer/FilterPushdownSuite.scala | 49 ++++++++++++++++++++++
4 files changed, 56 insertions(+), 2 deletions(-)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/collectionOperations.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/collectionOperations.scala
index 0d598aecfe8e..cb6d20eff068 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/collectionOperations.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/collectionOperations.scala
@@ -3573,8 +3573,8 @@ case class Sequence(
override def nullable: Boolean = children.exists(_.nullable)
- // If step is defined, then an error will be thrown if the start and stop do
not satisfy the step.
- override lazy val throwable: Boolean = stepOpt.isDefined
+ // Can throw if step is defined and start and stop don't match or any of the
children can throw.
+ override lazy val throwable: Boolean = stepOpt.isDefined ||
children.exists(_.throwable)
override def dataType: ArrayType = ArrayType(start.dataType, containsNull =
false)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/misc.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/misc.scala
index fa456d8955bf..61e04b9f761c 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/misc.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/misc.scala
@@ -84,6 +84,7 @@ case class RaiseError(errorClass: Expression, errorParms:
Expression, dataType:
override def foldable: Boolean = false
override def nullable: Boolean = true
+ override lazy val throwable: Boolean = true
override def inputTypes: Seq[AbstractDataType] =
Seq(
StringTypeWithCollation(supportsTrimCollation = true),
diff --git
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/MiscExpressionsSuite.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/MiscExpressionsSuite.scala
index 032762470900..a586e33afdd2 100644
---
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/MiscExpressionsSuite.scala
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/MiscExpressionsSuite.scala
@@ -55,6 +55,10 @@ class MiscExpressionsSuite extends SparkFunSuite with
ExpressionEvalHelper {
)
}
+ test("SPARK-58627: RaiseError is throwable") {
+ assert(RaiseError(Literal("error!")).throwable)
+ }
+
test("SPARK-55109: RaiseError.sql uses single-argument form only for known
error classes") {
assert(RaiseError(Literal("error!")).sql === "raise_error('error!')")
diff --git
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/FilterPushdownSuite.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/FilterPushdownSuite.scala
index a43be9a1c0a6..517d31021632 100644
---
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/FilterPushdownSuite.scala
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/FilterPushdownSuite.scala
@@ -1645,6 +1645,55 @@ class FilterPushdownSuite extends PlanTest {
comparePlans(optimizedQueryWithoutStep, correctAnswer)
}
+ test("SPARK-58627: do not push down predicate with raise_error through
joins") {
+ val x = testStringRelation.subquery("x")
+ val y = testRelation1.subquery("y")
+
+ // raise_error over literals references no columns, so it looks evaluable
on either side and
+ // gets pushed into the left relation, where it fires on rows the join
would have dropped.
+ val queryWithRaiseError = x.join(y, joinType = Inner, condition =
Some($"x.a" === $"y.d"))
+ .where(IsNull(RaiseError(Literal("boom"))))
+ .analyze
+ comparePlans(Optimize.execute(queryWithRaiseError), queryWithRaiseError)
+ }
+
+ 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)
+ }
+
test("push down deterministic predicate through BinBy") {
// Relation: ts_start, ts_end, value (DISTRIBUTE), label (pass-through).
val tsStart = AttributeReference("ts_start", TimestampType, nullable =
false)()
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
