[SPARK-58627][SQL] Mark raise error Throwable & fix sequence throwable

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/collectionOperations.scala

@@ -3475,8 +3475,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)
 

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),

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!')")
 

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)
+  }
+
   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)()