From a84601c068d1398c97af4cedf017828cc5ad89c3 Mon Sep 17 00:00:00 2001 From: Holden Karau Date: Fri, 7 Aug 2026 02:46:07 +0000 Subject: [PATCH 1/2] [SPARK-58627][SQL] Mark RaiseError as 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` and its propagation through `AssertTrue`'s replacement. - `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; plus a `raise_error` nested under a stepless `sequence(...)`, covering the `Sequence` override. - `ColumnExpressionSuite`: end-to-end, the query above returns an empty result, with a control that still raises once the join does produce rows. All five new tests fail without the source changes, the end-to-end one with exactly the `[USER_RAISED_EXCEPTION] boom` shown above. Also ran the full `catalyst` suite (375 suites, 10460 tests) -- all green -- and the full `sql` suite. In `sql`, every failure is a pre-existing environment defect in my local setup, where the RocksDB native library cannot be loaded (`UnsatisfiedLinkError: librocksdbjni*.so: libstdc++.so.6`); this takes out the state-store and stateful-streaming suites and leaks a SparkContext that aborts unrelated suites sharing the forked JVM. Re-running those aborted suites in isolation, they pass. Relevant to this change, all plan-shape-sensitive suites are green: `SQLQueryTestSuite` (golden files), the eight `TPCDS*`/`TPCH*` `PlanStability*` suites, `TPCDSQuerySuite`, `ExplainSuite`, `SubquerySuite` and `DataFrameJoinSuite`. Relying on CI for the state-store coverage. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Code (Opus 5) --- .../expressions/collectionOperations.scala | 3 +- .../spark/sql/catalyst/expressions/misc.scala | 3 + .../expressions/MiscExpressionsSuite.scala | 11 ++++ .../optimizer/FilterPushdownSuite.scala | 58 +++++++++++++++++++ .../spark/sql/ColumnExpressionSuite.scala | 25 ++++++++ 5 files changed, 99 insertions(+), 1 deletion(-) 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 4a842f2c27f9a..4cfb6e4345721 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 @@ -3476,7 +3476,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 + // Either way this must still fall back to the children, which may throw on their own. + 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 fa456d8955bf1..5e22ff37140a6 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,9 @@ case class RaiseError(errorClass: Expression, errorParms: Expression, dataType: override def foldable: Boolean = false override def nullable: Boolean = true + // Always throws when evaluated. This has to be set explicitly: the inherited default infers + // the flag from the children, which here are usually non-throwing literals. + 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 0327624709001..00731af360fc2 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,17 @@ class MiscExpressionsSuite extends SparkFunSuite with ExpressionEvalHelper { ) } + test("SPARK-58627: RaiseError is throwable") { + assert(RaiseError(Literal("error!")).throwable) + + // An expression wrapping a RaiseError picks the flag up from the inherited default. Note that + // assert_true is rewritten to its If(cond, null, RaiseError(...)) replacement by + // ReplaceExpressions before any rule reads the flag, so it is the replacement, not the + // AssertTrue node, that the optimizer actually consults. + assert(AssertTrue(Literal(true)).throwable) + assert(AssertTrue(Literal(true)).replacement.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 a43be9a1c0a66..ddb0aaee90728 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,64 @@ 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 push down predicate with raise_error through joins") { + val x = testStringRelation.subquery("x") + val y = testRelation1.subquery("y") + + // Do not push down: below the join the predicate would run on rows the join discards, so + // raise_error could fire for a query that succeeds without pushdown. + val queryWithRaiseError = x.join(y, joinType = Inner, condition = Some($"x.a" === $"y.d")) + .where(IsNull(RaiseError($"x.e"))) + .analyze + comparePlans(Optimize.execute(queryWithRaiseError), queryWithRaiseError) + + // A predicate over the same column that cannot throw is still pushed down. + val queryWithoutRaiseError = x.join(y, joinType = Inner, condition = Some($"x.a" === $"y.d")) + .where(IsNotNull($"x.e")) + .analyze + val correctAnswer = x.where(IsNotNull($"x.e")) + .join(y, joinType = Inner, condition = Some($"x.a" === $"y.d")) + .analyze + comparePlans(Optimize.execute(queryWithoutRaiseError), correctAnswer) + } + + 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)() diff --git a/sql/core/src/test/scala/org/apache/spark/sql/ColumnExpressionSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/ColumnExpressionSuite.scala index 379b7320a299e..18c02b2bc35e4 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/ColumnExpressionSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/ColumnExpressionSuite.scala @@ -2620,6 +2620,31 @@ class ColumnExpressionSuite extends SharedSparkSession { parameters = Map("errorMessage" -> "hello")) } + test("SPARK-58627: raise_error in a filter is not pushed through a join") { + withTempView("t1", "t2") { + Seq(1, 2, 3).toDF("a").createOrReplaceTempView("t1") + Seq(4, 5, 6).toDF("b").createOrReplaceTempView("t2") + + // The join yields no rows, so the predicate is never evaluated and the query must not + // raise. Before SPARK-58627, RaiseError did not report throwable, so the predicate was + // pushed below the join and raise_error fired on the rows of t1. + checkAnswer( + spark.sql("SELECT * FROM t1 JOIN t2 ON t1.a = t2.b WHERE raise_error('boom') IS NULL"), + Seq.empty[Row]) + + // Control: the predicate is retained rather than dropped, so it still raises once the join + // actually produces rows to evaluate it on. + checkError( + exception = intercept[SparkRuntimeException] { + spark.sql( + "SELECT * FROM t1 JOIN t2 ON t1.a = t2.b - 3 WHERE raise_error('boom') IS NULL") + .collect() + }, + condition = "USER_RAISED_EXCEPTION", + parameters = Map("errorMessage" -> "boom")) + } + } + test("SPARK-34677: negate/add/subtract year-month and day-time intervals") { import testImplicits._ val df = Seq((Period.ofMonths(10), Duration.ofDays(10), Period.ofMonths(1), Duration.ofDays(1))) From 27723d4cbb9dac92942f4976dad6cb9aa1caa9f7 Mon Sep 17 00:00:00 2001 From: Holden Karau Date: Wed, 12 Aug 2026 12:05:34 -0700 Subject: [PATCH 2/2] Clean up some verbose comments and duplicated test coverage from Claude. Co-authored-by: Holden Karau --- .../expressions/collectionOperations.scala | 3 +-- .../spark/sql/catalyst/expressions/misc.scala | 2 -- .../expressions/MiscExpressionsSuite.scala | 7 ------ .../optimizer/FilterPushdownSuite.scala | 21 ---------------- .../spark/sql/ColumnExpressionSuite.scala | 25 ------------------- 5 files changed, 1 insertion(+), 57 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 4cfb6e4345721..9cff812397b02 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 @@ -3475,8 +3475,7 @@ 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. - // Either way this must still fall back to the children, which may throw on their own. + // 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 5e22ff37140a6..61e04b9f761ce 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,8 +84,6 @@ case class RaiseError(errorClass: Expression, errorParms: Expression, dataType: override def foldable: Boolean = false override def nullable: Boolean = true - // Always throws when evaluated. This has to be set explicitly: the inherited default infers - // the flag from the children, which here are usually non-throwing literals. override lazy val throwable: Boolean = true override def inputTypes: Seq[AbstractDataType] = Seq( 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 00731af360fc2..a586e33afdd26 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 @@ -57,13 +57,6 @@ class MiscExpressionsSuite extends SparkFunSuite with ExpressionEvalHelper { test("SPARK-58627: RaiseError is throwable") { assert(RaiseError(Literal("error!")).throwable) - - // An expression wrapping a RaiseError picks the flag up from the inherited default. Note that - // assert_true is rewritten to its If(cond, null, RaiseError(...)) replacement by - // ReplaceExpressions before any rule reads the flag, so it is the replacement, not the - // AssertTrue node, that the optimizer actually consults. - assert(AssertTrue(Literal(true)).throwable) - assert(AssertTrue(Literal(true)).replacement.throwable) } test("SPARK-55109: RaiseError.sql uses single-argument form only for known error classes") { 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 ddb0aaee90728..9609e407f6dc9 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 @@ -1662,27 +1662,6 @@ class FilterPushdownSuite extends PlanTest { comparePlans(Optimize.execute(queryWithRaiseError), queryWithRaiseError) } - test("SPARK-58627: do not push down predicate with raise_error through joins") { - val x = testStringRelation.subquery("x") - val y = testRelation1.subquery("y") - - // Do not push down: below the join the predicate would run on rows the join discards, so - // raise_error could fire for a query that succeeds without pushdown. - val queryWithRaiseError = x.join(y, joinType = Inner, condition = Some($"x.a" === $"y.d")) - .where(IsNull(RaiseError($"x.e"))) - .analyze - comparePlans(Optimize.execute(queryWithRaiseError), queryWithRaiseError) - - // A predicate over the same column that cannot throw is still pushed down. - val queryWithoutRaiseError = x.join(y, joinType = Inner, condition = Some($"x.a" === $"y.d")) - .where(IsNotNull($"x.e")) - .analyze - val correctAnswer = x.where(IsNotNull($"x.e")) - .join(y, joinType = Inner, condition = Some($"x.a" === $"y.d")) - .analyze - comparePlans(Optimize.execute(queryWithoutRaiseError), correctAnswer) - } - test("SPARK-58627: do not combine predicate with raise_error with other filters") { val x = testStringRelation.subquery("x") diff --git a/sql/core/src/test/scala/org/apache/spark/sql/ColumnExpressionSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/ColumnExpressionSuite.scala index 18c02b2bc35e4..379b7320a299e 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/ColumnExpressionSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/ColumnExpressionSuite.scala @@ -2620,31 +2620,6 @@ class ColumnExpressionSuite extends SharedSparkSession { parameters = Map("errorMessage" -> "hello")) } - test("SPARK-58627: raise_error in a filter is not pushed through a join") { - withTempView("t1", "t2") { - Seq(1, 2, 3).toDF("a").createOrReplaceTempView("t1") - Seq(4, 5, 6).toDF("b").createOrReplaceTempView("t2") - - // The join yields no rows, so the predicate is never evaluated and the query must not - // raise. Before SPARK-58627, RaiseError did not report throwable, so the predicate was - // pushed below the join and raise_error fired on the rows of t1. - checkAnswer( - spark.sql("SELECT * FROM t1 JOIN t2 ON t1.a = t2.b WHERE raise_error('boom') IS NULL"), - Seq.empty[Row]) - - // Control: the predicate is retained rather than dropped, so it still raises once the join - // actually produces rows to evaluate it on. - checkError( - exception = intercept[SparkRuntimeException] { - spark.sql( - "SELECT * FROM t1 JOIN t2 ON t1.a = t2.b - 3 WHERE raise_error('boom') IS NULL") - .collect() - }, - condition = "USER_RAISED_EXCEPTION", - parameters = Map("errorMessage" -> "boom")) - } - } - test("SPARK-34677: negate/add/subtract year-month and day-time intervals") { import testImplicits._ val df = Seq((Period.ofMonths(10), Duration.ofDays(10), Period.ofMonths(1), Duration.ofDays(1)))