[SPARK-57525][CONNECT] Declarative Pipelines should not throw NoSuchElementException when a run fails without an attached cause

### What changes were proposed in this pull request?
`PipelinesHandler.startRun` rethrows a failed pipeline run to the Spark Connect client via `runFailureEvent.foreach { event => throw event.error.get }`. But `event.error` is `None` for run termination reasons that carry no cause - `UnexpectedRunFailure` and `FailureStoppingFlow` both have `cause = None` - so `event.error.get` raised a `NoSuchElementException`, crashing the handler and hiding the real failure from the client.

This PR extracts the rethrow into `throwRunFailure`: when the failure has a cause it is rethrown unchanged; when it does not, a `SparkException` with a new `PIPELINE_RUN_FAILED` error condition is thrown, carrying the run's termination message. `PIPELINE_RUN_FAILED` (rather than `INTERNAL_ERROR`) is used so that operational outcomes such as `FailureStoppingFlow` are not mislabeled as Spark bugs.

### Why are the changes needed?
A run that fails without an attached cause (e.g. `UnexpectedRunFailure`, or a flow that fails to stop) currently surfaces to the Connect client as an opaque `NoSuchElementException` ("None.get") instead of the actual run-failure message. That masks the real problem and looks like an internal error. These reasons reach this code via the asynchronous `onCompletion` path, where `PipelineExecution.runPipeline`'s own catch never fires.

### Does this PR introduce _any_ user-facing change?
Yes. When a pipeline run fails without an attached cause, the Spark Connect client now receives a `PIPELINE_RUN_FAILED` error carrying the run's termination message (e.g. "Run failed unexpectedly.") instead of a `NoSuchElementException`.

### How was this patch tested?
New `PipelinesHandlerSuite` unit-tests `throwRunFailure` for both cases: the cause-present case rethrows the original cause, and the no-cause case throws a `PIPELINE_RUN_FAILED` `SparkException` carrying the termination message (verified with `checkError`, using the real `UnexpectedRunFailure` and `FailureStoppingFlow` messages). The cause-less termination reasons cannot be triggered deterministically through the end-to-end run path, so the rethrow is unit-tested directly. `SparkThrowableSuite` validates the new error condition.

### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 4.8)

Closes #56594 from LuciferYang/sdp-run-failure-no-cause.

Authored-by: YangJie <yangjie01@baidu.com>
Signed-off-by: yangjie01 <yangjie01@baidu.com>

Author: LuciferYang

common/utils/src/main/resources/error/error-conditions.json

@@ -6226,6 +6226,12 @@
     ],
     "sqlState" : "0A000"
   },
+  "PIPELINE_RUN_FAILED" : {
+    "message" : [
+      "<message>"
+    ],
+    "sqlState" : "42K0S"
+  },
   "PIPELINE_SQL_GRAPH_ELEMENT_REGISTRATION_ERROR" : {
     "message" : [
       "<message>",

common/utils/src/main/resources/error/error-states.json

@@ -4655,6 +4655,12 @@
         "standard": "N",
         "usedBy": ["Spark"]
     },
+    "42K0S": {
+        "description": "Pipeline run failed.",
+        "origin": "Spark",
+        "standard": "N",
+        "usedBy": ["Spark"]
+    },
     "42KD0": {
         "description": "Ambiguous name reference.",
         "origin": "Databricks",

sql/connect/server/src/main/scala/org/apache/spark/sql/connect/pipelines/PipelinesHandler.scala

@@ -22,6 +22,7 @@ import scala.util.Using
 
 import io.grpc.stub.StreamObserver
 
+import org.apache.spark.SparkException
 import org.apache.spark.connect.proto
 import org.apache.spark.connect.proto.{ExecutePlanResponse, PipelineCommandResult, Relation, ResolvedIdentifier}
 import org.apache.spark.connect.proto.PipelineCommand.DefineFlow.AutoCdcFlowDetails
@@ -563,12 +564,27 @@ private[connect] object PipelinesHandler extends Logging {
 
       // Rethrow any exceptions that caused the pipeline run to fail so that the exception is
       // propagated back to the SC client / CLI.
-      runFailureEvent.foreach { event =>
-        throw event.error.get
-      }
+      runFailureEvent.foreach(throwRunFailure)
     }
   }
 
+  /**
+   * Rethrows the failure behind a terminal run-failure event so it reaches the Spark Connect
+   * client. Most failures carry the underlying cause (e.g. a flow's QueryExecutionFailure), but
+   * some termination reasons (UnexpectedRunFailure, FailureStoppingFlow) have none. When the
+   * cause is absent, throw a PIPELINE_RUN_FAILED error built from the event message rather than
+   * calling Option.get, which would throw a NoSuchElementException and hide the real failure.
+   * Using PIPELINE_RUN_FAILED instead of INTERNAL_ERROR avoids mislabeling operational failures
+   * as bugs.
+   */
+  private[connect] def throwRunFailure(failureEvent: PipelineEvent): Nothing = {
+    throw failureEvent.error.getOrElse(
+      new SparkException(
+        errorClass = "PIPELINE_RUN_FAILED",
+        messageParameters = Map("message" -> failureEvent.message),
+        cause = null))
+  }
+
   /**
    * Creates the table filters for the full refresh and refresh operations based on the StartRun
    * command user provided. Also validates the command parameters to ensure that they are

sql/connect/server/src/test/scala/org/apache/spark/sql/connect/pipelines/PipelinesHandlerSuite.scala

@@ -0,0 +1,70 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.connect.pipelines
+
+import org.apache.spark.{SparkException, SparkFunSuite}
+import org.apache.spark.sql.catalyst.TableIdentifier
+import org.apache.spark.sql.pipelines.common.RunState
+import org.apache.spark.sql.pipelines.graph.{FailureStoppingFlow, UnexpectedRunFailure}
+import org.apache.spark.sql.pipelines.logging.{ConstructPipelineEvent, EventLevel, PipelineEventOrigin, RunProgress}
+
+class PipelinesHandlerSuite extends SparkFunSuite {
+
+  private def runFailedEvent(message: String, error: Option[Throwable]) =
+    ConstructPipelineEvent(
+      origin =
+        PipelineEventOrigin(datasetName = None, flowName = None, sourceCodeLocation = None),
+      // throwRunFailure only reads message and exception; the remaining fields are filled with
+      // valid placeholder values to construct the event.
+      level = EventLevel.INFO,
+      message = message,
+      details = RunProgress(RunState.FAILED),
+      exception = error)
+
+  // Use the real no-cause termination-reason messages so the tests break if their wording drifts.
+  private val unexpectedRunFailureMessage = UnexpectedRunFailure().message
+
+  private val failureStoppingFlowMessage =
+    FailureStoppingFlow(
+      Seq(TableIdentifier("t1", Some("db")), TableIdentifier("t2", Some("db")))).message
+
+  // Regression guard rather than a fix-validation test: the old buggy code (throw error.get) also
+  // rethrew the cause unchanged, so this case passes against both implementations. The no-cause
+  // test below is the one that genuinely exercises this PR's fix.
+  test("throwRunFailure rethrows the underlying cause when present") {
+    val cause = new RuntimeException("boom")
+    val thrown = intercept[RuntimeException] {
+      PipelinesHandler.throwRunFailure(runFailedEvent("Run failed.", Some(cause)))
+    }
+    assert(thrown eq cause)
+  }
+
+  test("throwRunFailure surfaces the message when the failure has no cause") {
+    // No-cause reasons must fall back to a PIPELINE_RUN_FAILED error built from the event message
+    // rather than raising NoSuchElementException; the message is forwarded verbatim.
+    Seq(unexpectedRunFailureMessage, failureStoppingFlowMessage).foreach { message =>
+      val thrown = intercept[SparkException] {
+        PipelinesHandler.throwRunFailure(runFailedEvent(message, None))
+      }
+      checkError(
+        thrown,
+        condition = "PIPELINE_RUN_FAILED",
+        parameters = Map("message" -> message))
+    }
+  }
+}