You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ya...@apache.org on 2026/07/27 15:24:53 UTC
(spark) branch branch-4.1 updated: [SPARK-57635][SQL] Make Declarative Pipelines run-termination reason deterministic
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 fe16861dc5e2 [SPARK-57635][SQL] Make Declarative Pipelines run-termination reason deterministic
fe16861dc5e2 is described below
commit fe16861dc5e24e2b6c2d760b2bce80e4bd1db174
Author: YangJie <ya...@baidu.com>
AuthorDate: Mon Jul 27 23:22:32 2026 +0800
[SPARK-57635][SQL] Make Declarative Pipelines run-termination reason deterministic
### What changes were proposed in this pull request?
`TriggeredGraphExecution.getRunTerminationReason` decided which failed flow's reason to report by calling `collectFirst` over `failureTracker`, a `ConcurrentHashMap` whose iteration order is unspecified. When more than one flow exhausts its retries (a non-retryable `StopFlowExecution`), the flow whose reason gets surfaced therefore varied from run to run.
This extracts a small pure helper, `chooseRunTerminationReason`, that considers only the stopped flows and picks the earliest one by `(lastFailTimestamp, flowName)`, so the reported reason is stable across otherwise-identical runs. `getRunTerminationReason` now calls it and falls back to `UnexpectedRunFailure()`. The previous code also computed `graphForExecution.flow(...)` and `lastException` only to discard them; those are dropped.
### Why are the changes needed?
Two runs that fail the same way could report different termination reasons (different flow name and cause), which is confusing in logs and events and makes the outcome non-reproducible.
### Does this PR introduce _any_ user-facing change?
No. The reported reason was already one of the failing flows; it is now chosen deterministically.
### How was this patch tested?
Added unit tests for `chooseRunTerminationReason` in `TriggeredGraphExecutionSuite`: the earliest failure wins regardless of iteration order, ties are broken by flow name, and retryable failures are ignored. They fail against the previous order-dependent selection and pass with this change.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 4.8)
Closes #56698 from LuciferYang/sdp-deterministic-termination-reason.
Authored-by: YangJie <ya...@baidu.com>
Signed-off-by: yangjie01 <ya...@baidu.com>
(cherry picked from commit effd582afa3980a6b674d212a705c0d5212f2948)
Signed-off-by: yangjie01 <ya...@baidu.com>
---
.../pipelines/graph/TriggeredGraphExecution.scala | 35 +++++++++-------
.../graph/TriggeredGraphExecutionSuite.scala | 49 ++++++++++++++++++++++
2 files changed, 69 insertions(+), 15 deletions(-)
diff --git a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/TriggeredGraphExecution.scala b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/TriggeredGraphExecution.scala
index 02871e64aa76..22b64798f598 100644
--- a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/TriggeredGraphExecution.scala
+++ b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/TriggeredGraphExecution.scala
@@ -418,21 +418,9 @@ class TriggeredGraphExecution(
return RunCompletion()
}
- val executionFailureOpt = failureTracker.iterator
- .map {
- case (flowIdentifier, failureInfo) =>
- (
- graphForExecution.flow(flowIdentifier),
- failureInfo.lastException,
- failureInfo.lastExceptionAction
- )
- }
- .collectFirst {
- case (_, _, GraphExecution.StopFlowExecution(reason)) =>
- reason.runTerminationReason
- }
-
- executionFailureOpt.getOrElse(UnexpectedRunFailure())
+ TriggeredGraphExecution
+ .chooseRunTerminationReason(failureTracker.iterator)
+ .getOrElse(UnexpectedRunFailure())
}
}
@@ -450,6 +438,23 @@ case class TriggeredFailureInfo(
object TriggeredGraphExecution {
+ /**
+ * Picks the run-termination reason from the flows whose execution was stopped because they
+ * exhausted their retries. Several flows can stop a run and `failures` comes from an unordered
+ * map, so the earliest failure is chosen - ties broken by flow name - to keep the reported
+ * reason stable across otherwise-identical runs.
+ */
+ private[graph] def chooseRunTerminationReason(
+ failures: Iterator[(TableIdentifier, TriggeredFailureInfo)]): Option[RunTerminationReason] = {
+ failures
+ .collect {
+ case (id, TriggeredFailureInfo(ts, _, _, GraphExecution.StopFlowExecution(r))) =>
+ (ts, id.unquotedString, r.runTerminationReason)
+ }
+ .minByOption { case (ts, flowName, _) => (ts, flowName) }
+ .map { case (_, _, reason) => reason }
+ }
+
// All possible states of a data stream for a flow
sealed trait StreamState
object StreamState {
diff --git a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/TriggeredGraphExecutionSuite.scala b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/TriggeredGraphExecutionSuite.scala
index 57baf4c2d5b1..db8c368ca89d 100644
--- a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/TriggeredGraphExecutionSuite.scala
+++ b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/TriggeredGraphExecutionSuite.scala
@@ -1060,4 +1060,53 @@ class TriggeredGraphExecutionSuite extends ExecutionTest with SharedSparkSession
assert(warnCount == 2 && errorCount == 1)
}
+
+ /** A non-retryable (retries exhausted) failure, i.e. one that stops the run. */
+ private def stopFailure(ts: Long, flowName: String, cause: Throwable): TriggeredFailureInfo = {
+ // currentNumTries > maxAllowedRetries yields a StopFlowExecution.
+ val action = GraphExecution.determineFlowExecutionActionFromError(
+ ex = cause,
+ flowDisplayName = flowName,
+ currentNumTries = 2,
+ maxAllowedRetries = 1)
+ TriggeredFailureInfo(ts, numFailures = 2, lastException = cause, lastExceptionAction = action)
+ }
+
+ test("chooseRunTerminationReason surfaces the earliest non-retryable failure deterministically") {
+ val earliestCause = new RuntimeException("earliest")
+ // A retryable failure with an earlier timestamp must be ignored: it does not stop the run.
+ val retryable = {
+ val cause = new RuntimeException("retryable")
+ val action = GraphExecution.determineFlowExecutionActionFromError(
+ ex = cause, flowDisplayName = "flow_retry", currentNumTries = 1, maxAllowedRetries = 3)
+ TableIdentifier("flow_retry") -> TriggeredFailureInfo(
+ 50, numFailures = 1, lastException = cause, lastExceptionAction = action)
+ }
+ val entries = Seq(
+ TableIdentifier("flow_c") -> stopFailure(300, "flow_c", new RuntimeException("c")),
+ TableIdentifier("flow_b") -> stopFailure(100, "flow_b", earliestCause),
+ TableIdentifier("flow_a") -> stopFailure(200, "flow_a", new RuntimeException("a")),
+ retryable)
+
+ // The earliest non-retryable failure (flow_b @ 100) wins regardless of iteration order.
+ Seq(entries, entries.reverse).foreach { ordered =>
+ assert(
+ TriggeredGraphExecution.chooseRunTerminationReason(ordered.iterator)
+ .contains(QueryExecutionFailure("flow_b", 1, Some(earliestCause))))
+ }
+ }
+
+ test("chooseRunTerminationReason breaks ties between equal timestamps by flow name") {
+ val aCause = new RuntimeException("a")
+ val entries = Seq(
+ TableIdentifier("flow_z") -> stopFailure(100, "flow_z", new RuntimeException("z")),
+ TableIdentifier("flow_a") -> stopFailure(100, "flow_a", aCause))
+ assert(
+ TriggeredGraphExecution.chooseRunTerminationReason(entries.iterator)
+ .contains(QueryExecutionFailure("flow_a", 1, Some(aCause))))
+ }
+
+ test("chooseRunTerminationReason has no reason when no flow stopped the run") {
+ assert(TriggeredGraphExecution.chooseRunTerminationReason(Iterator.empty).isEmpty)
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org