[SPARK-57635][SQL] Make Declarative Pipelines run-termination reason deterministic

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 {

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