This is an automated email from the ASF dual-hosted git repository.
LuciferYang pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new 91619952f072 [SPARK-57525][4.1][CONNECT] Declarative Pipelines should
not throw NoSuchElementException when a run fails without an attached cause
91619952f072 is described below
commit 91619952f0726744c8760a0e4ed0cecc653a00f8
Author: YangJie <[email protected]>
AuthorDate: Tue Jul 7 12:17:40 2026 +0800
[SPARK-57525][4.1][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. `SparkThrow [...]
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 4.8)
Closes #56900 from LuciferYang/SPARK-57525-4.1.
Authored-by: YangJie <[email protected]>
Signed-off-by: yangjie01 <[email protected]>
---
.../src/main/resources/error/error-conditions.json | 6 ++
.../src/main/resources/error/error-states.json | 6 ++
.../sql/connect/pipelines/PipelinesHandler.scala | 22 ++++++-
.../connect/pipelines/PipelinesHandlerSuite.scala | 70 ++++++++++++++++++++++
4 files changed, 101 insertions(+), 3 deletions(-)
diff --git a/common/utils/src/main/resources/error/error-conditions.json
b/common/utils/src/main/resources/error/error-conditions.json
index aa0f0a89f97c..43e293ef8013 100644
--- a/common/utils/src/main/resources/error/error-conditions.json
+++ b/common/utils/src/main/resources/error/error-conditions.json
@@ -5017,6 +5017,12 @@
},
"sqlState" : "42710"
},
+ "PIPELINE_RUN_FAILED" : {
+ "message" : [
+ "<message>"
+ ],
+ "sqlState" : "42K0S"
+ },
"PIPELINE_SQL_GRAPH_ELEMENT_REGISTRATION_ERROR" : {
"message" : [
"<message>",
diff --git a/common/utils/src/main/resources/error/error-states.json
b/common/utils/src/main/resources/error/error-states.json
index 4fddbeed4090..b70a791e8cbe 100644
--- a/common/utils/src/main/resources/error/error-states.json
+++ b/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",
diff --git
a/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/pipelines/PipelinesHandler.scala
b/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/pipelines/PipelinesHandler.scala
index 62f060014117..279a524c8551 100644
---
a/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/pipelines/PipelinesHandler.scala
+++
b/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/pipelines/PipelinesHandler.scala
@@ -23,6 +23,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.internal.Logging
@@ -440,12 +441,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
diff --git
a/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/pipelines/PipelinesHandlerSuite.scala
b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/pipelines/PipelinesHandlerSuite.scala
new file mode 100644
index 000000000000..72d077a40e04
--- /dev/null
+++
b/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))
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
