UpdateTerminationReason -> RunTerminationReason

Author: sryza

sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/GraphExecution.scala

@@ -184,9 +184,9 @@ abstract class GraphExecution(
   /**
    * Returns the reason why this flow execution has terminated.
    * If the function is called before the flow has not terminated yet, the behavior is undefined,
-   * and may return [[UnexpectedUpdateFailure]].
+   * and may return [[UnexpectedRunFailure]].
    */
-  def getUpdateTerminationReason: UpdateTerminationReason
+  def getUpdateTerminationReason: RunTerminationReason
 
   def maxRetryAttemptsForFlow(flowName: TableIdentifier): Int = {
     val flow = graphForExecution.flow(flowName)
@@ -225,7 +225,7 @@ object GraphExecution extends Logging {
   sealed trait FlowExecutionStopReason {
     def cause: Throwable
     def flowDisplayName: String
-    def updateTerminationReason: UpdateTerminationReason
+    def updateTerminationReason: RunTerminationReason
     def failureMessage: String
     // If true, we record this flow execution as STOPPED with a WARNING instead a FAILED with ERROR.
     def warnInsteadOfError: Boolean = false
@@ -240,7 +240,7 @@ object GraphExecution extends Logging {
       flowDisplayName: String,
       maxAllowedRetries: Int
   ) extends FlowExecutionStopReason {
-    override lazy val updateTerminationReason: UpdateTerminationReason = {
+    override lazy val updateTerminationReason: RunTerminationReason = {
       QueryExecutionFailure(flowDisplayName, maxAllowedRetries, Option(cause))
     }
     override lazy val failureMessage: String = {

sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/PipelinesErrors.scala

@@ -155,7 +155,7 @@ object PipelinesErrors extends Logging {
             )
           }
           if (shouldRethrow) {
-            throw UpdateTerminationException(reason.updateTerminationReason)
+            throw RunTerminationException(reason.updateTerminationReason)
           }
       }
     }

sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/UpdateTerminationReason.scala renamed to sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/RunTerminationReason.scala

@@ -20,93 +20,73 @@ package org.apache.spark.sql.pipelines.graph
 import org.apache.spark.sql.catalyst.TableIdentifier
 import org.apache.spark.sql.pipelines.common.RunState
 
-sealed trait UpdateTerminationReason {
+sealed trait RunTerminationReason {
 
-  /** Terminal state for the given update. */
+  /** Terminal state for the given run. */
   def terminalState: RunState
 
   /**
-   * User visible message associated with update termination. This will also be set as the message
-   * in the associated terminal update progress log.
+   * User visible message associated with run termination. This will also be set as the message
+   * in the associated terminal run progress log.
    */
   def message: String
 
   /**
-   * Exception associated with the given update termination. This exception will be
-   * included in the error details in the associated terminal update progress event.
+   * Exception associated with the given run termination. This exception will be
+   * included in the error details in the associated terminal run progress event.
    */
   def cause: Option[Throwable]
-
-  /**
-   * Whether this termination reason should override the partial execution failure thrown during
-   * flow execution.
-   */
-  def overridesPartialExecutionFailure: Boolean = false
 }
 
 /**
- * Helper exception class that indicates that an update has to be terminated and
+ * Helper exception class that indicates that a run has to be terminated and
  * tracks the associated termination reason.
  */
-case class UpdateTerminationException(reason: UpdateTerminationReason) extends Exception
+case class RunTerminationException(reason: RunTerminationReason) extends Exception
 
 // ===============================================================
-// ============ Graceful update termination states ===============
+// ============ Graceful run termination states ==================
 // ===============================================================
 
-/** Indicates that a triggered update has successfully completed execution. */
-case class UpdateCompletion() extends UpdateTerminationReason {
+/** Indicates that a triggered run has successfully completed execution. */
+case class RunCompletion() extends RunTerminationReason {
   override def terminalState: RunState = RunState.COMPLETED
-  override def message: String = s"Update is $terminalState."
+  override def message: String = s"Run is $terminalState."
   override def cause: Option[Throwable] = None
 }
 
-/**
- * Indicates that the update is being terminated since the schema for a given flow
- * changed during execution.
- */
-case class UpdateSchemaChange(flowName: String, override val cause: Option[Throwable])
-    extends UpdateTerminationReason {
-  override def terminalState: RunState = RunState.CANCELED
-  override def message: String =
-    s"Update has been cancelled due to a schema change in $flowName, " +
-    s"and will be restarted."
-
-  override def overridesPartialExecutionFailure: Boolean = true
-}
-
 // ===============================================================
-// ======================= Update failures =======================
+// ======================= Run failures ==========================
 // ===============================================================
 
-/** Indicates that an update entered the failed state.. */
-abstract sealed class UpdateFailure extends UpdateTerminationReason {
+/** Indicates that an run entered the failed state.. */
+abstract sealed class RunFailure extends RunTerminationReason {
 
   /** Whether or not this failure is considered fatal / irrecoverable. */
   def isFatal: Boolean
 
   override def terminalState: RunState = RunState.FAILED
 }
 
-/** Indicates that update has failed due to a query execution failure. */
+/** Indicates that run has failed due to a query execution failure. */
 case class QueryExecutionFailure(
     flowName: String,
     maxRetries: Int,
     override val cause: Option[Throwable])
-    extends UpdateFailure {
+    extends RunFailure {
   override def isFatal: Boolean = false
 
   override def message: String =
     if (maxRetries == 0) {
-      s"Update is $terminalState since flow '$flowName' has failed."
+      s"Run is $terminalState since flow '$flowName' has failed."
     } else {
-      s"Update is $terminalState since flow '$flowName' has failed more " +
+      s"Run is $terminalState since flow '$flowName' has failed more " +
       s"than $maxRetries times."
     }
 }
 
 /** Abstract class used to identify failures related to failures stopping an operation/timeouts. */
-abstract class FailureStoppingOperation extends UpdateFailure {
+abstract class FailureStoppingOperation extends RunFailure {
 
   /** Name of the operation that failed to stop. */
   def operation: String
@@ -120,22 +100,22 @@ case class FailureStoppingFlow(flowIdentifiers: Seq[TableIdentifier])
   override def message: String = {
     if (flowIdentifiers.nonEmpty) {
       val flowNamesToPrint = flowIdentifiers.map(_.toString).sorted.take(5).mkString(", ")
-      s"Update is $terminalState since following flows have failed to stop: " +
+      s"Run is $terminalState since following flows have failed to stop: " +
       s"$flowNamesToPrint."
     } else {
-      s"Update is $terminalState since stopping flow execution has failed."
+      s"Run is $terminalState since stopping flow execution has failed."
     }
   }
   override def cause: Option[Throwable] = None
 }
 
 /**
- * Update could not be associated with a proper root cause.
+ * Run could not be associated with a proper root cause.
  * This is not expected and likely indicates a bug.
  */
-case class UnexpectedUpdateFailure() extends UpdateFailure {
+case class UnexpectedRunFailure() extends RunFailure {
   override def isFatal: Boolean = false
   override def message: String =
-    s"Update $terminalState unexpectedly."
+    s"Run $terminalState unexpectedly."
   override def cause: Option[Throwable] = None
 }

sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/TriggeredGraphExecution.scala

@@ -44,7 +44,7 @@ import org.apache.spark.util.{Clock, SystemClock, ThreadUtils, Utils}
 class TriggeredGraphExecution(
     graphForExecution: DataflowGraph,
     env: PipelineUpdateContext,
-    onCompletion: UpdateTerminationReason => Unit = _ => (),
+    onCompletion: RunTerminationReason => Unit = _ => (),
     clock: Clock = new SystemClock()
 ) extends GraphExecution(graphForExecution, env) {
 
@@ -118,7 +118,7 @@ class TriggeredGraphExecution(
             case ex: Throwable =>
               logError(s"Exception thrown while stopping the update...", ex)
           } finally {
-            onCompletion(UnexpectedUpdateFailure())
+            onCompletion(UnexpectedRunFailure())
           }
       }
     )
@@ -393,7 +393,7 @@ class TriggeredGraphExecution(
       .map(_.get._1)
 
     if (flowsFailedToStop.nonEmpty) {
-      throw UpdateTerminationException(FailureStoppingFlow(flowsFailedToStop))
+      throw RunTerminationException(FailureStoppingFlow(flowsFailedToStop))
     }
   }
 
@@ -403,11 +403,11 @@ class TriggeredGraphExecution(
 
   override def stop(): Unit = { stopInternal(stopTopologicalExecutionThread = true) }
 
-  override def getUpdateTerminationReason: UpdateTerminationReason = {
+  override def getUpdateTerminationReason: RunTerminationReason = {
     val success =
       pipelineState.valuesIterator.forall(TERMINAL_NON_FAILURE_STREAM_STATES.contains)
     if (success) {
-      return UpdateCompletion()
+      return RunCompletion()
     }
 
     val executionFailureOpt = failureTracker.iterator
@@ -424,7 +424,7 @@ class TriggeredGraphExecution(
           reason.updateTerminationReason
       }
 
-    executionFailureOpt.getOrElse(UnexpectedUpdateFailure())
+    executionFailureOpt.getOrElse(UnexpectedRunFailure())
   }
 }
 

sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/TriggeredGraphExecutionSuite.scala

@@ -463,7 +463,7 @@ class TriggeredGraphExecutionSuite extends ExecutionTest {
       expectedEventLevel = EventLevel.ERROR,
       msgChecker = msg =>
         msg.contains(
-          "Update is FAILED since flow 'spark_catalog.test_db.branch_2' has failed more than " +
+          "Run is FAILED since flow 'spark_catalog.test_db.branch_2' has failed more than " +
           "2 times"
         )
     )
@@ -901,7 +901,7 @@ class TriggeredGraphExecutionSuite extends ExecutionTest {
       expectedEventLevel = EventLevel.ERROR,
       msgChecker = msg =>
         msg.contains(
-          "Update is FAILED since flow 'spark_catalog.test_db.a' has failed more than 2 times"
+          "Run is FAILED since flow 'spark_catalog.test_db.a' has failed more than 2 times"
         )
     )
   }