fix conflicts

Author: SCHJonathan

sql/pipelines/pom.xml

@@ -33,10 +33,6 @@
     <name>Spark Project Declarative Pipelines Library</name>
     <url>https://spark.apache.org/</url>
     <dependencies>
-        <dependency>
-            <groupId>org.scala-lang</groupId>
-            <artifactId>scala-library</artifactId>
-        </dependency>
         <dependency>
             <groupId>org.apache.spark</groupId>
             <artifactId>spark-core_${scala.binary.version}</artifactId>

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

@@ -68,14 +68,14 @@ case class ResolvedInput(input: Input, aliasIdentifier: AliasIdentifier)
 trait FlowFunction extends Logging {
 
   /**
-   * This function defines the transformations performed by a flow, expressed as a DataFrame.
+   * This function defines the transformations performed by a flow, expressed as a [[DataFrame]].
    *
    * @param allInputs the set of identifiers for all the [[Input]]s defined in the
    *                  [[DataflowGraph]].
    * @param availableInputs the list of all [[Input]]s available to this flow
    * @param configuration the spark configurations that apply to this flow.
    * @param queryContext The context of the query being evaluated.
-   * @return the inputs actually used, and the DataFrame expression for the flow
+   * @return the inputs actually used, and the [[DataFrame]] expression for the flow
    */
   def call(
       allInputs: Set[TableIdentifier],

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

@@ -61,6 +61,7 @@ abstract class GraphExecution(
     triggerFor = streamTrigger
   )
 
+  // Listener to process events and metrics.
   private val streamListener = new StreamListener(env, graphForExecution)
 
   /**
@@ -186,7 +187,7 @@ abstract class GraphExecution(
    * If the function is called before the flow has not terminated yet, the behavior is undefined,
    * and may return [[UnexpectedRunFailure]].
    */
-  def getUpdateTerminationReason: RunTerminationReason
+  def getRunTerminationReason: RunTerminationReason
 
   def maxRetryAttemptsForFlow(flowName: TableIdentifier): Int = {
     val flow = graphForExecution.flow(flowName)
@@ -225,7 +226,7 @@ object GraphExecution extends Logging {
   sealed trait FlowExecutionStopReason {
     def cause: Throwable
     def flowDisplayName: String
-    def updateTerminationReason: RunTerminationReason
+    def runTerminationReason: 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 +241,7 @@ object GraphExecution extends Logging {
       flowDisplayName: String,
       maxAllowedRetries: Int
   ) extends FlowExecutionStopReason {
-    override lazy val updateTerminationReason: RunTerminationReason = {
+    override lazy val runTerminationReason: 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 RunTerminationException(reason.updateTerminationReason)
+            throw RunTerminationException(reason.runTerminationReason)
           }
       }
     }

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

@@ -255,7 +255,7 @@ class TriggeredGraphExecution(
       }
     }
     if (allFlowsDone) {
-      onCompletion(getUpdateTerminationReason)
+      onCompletion(getRunTerminationReason)
     }
   }
 
@@ -403,7 +403,7 @@ class TriggeredGraphExecution(
 
   override def stop(): Unit = { stopInternal(stopTopologicalExecutionThread = true) }
 
-  override def getUpdateTerminationReason: RunTerminationReason = {
+  override def getRunTerminationReason: RunTerminationReason = {
     val success =
       pipelineState.valuesIterator.forall(TERMINAL_NON_FAILURE_STREAM_STATES.contains)
     if (success) {
@@ -421,7 +421,7 @@ class TriggeredGraphExecution(
       }
       .collectFirst {
         case (_, _, GraphExecution.StopFlowExecution(reason)) =>
-          reason.updateTerminationReason
+          reason.runTerminationReason
       }
 
     executionFailureOpt.getOrElse(UnexpectedRunFailure())

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

@@ -67,9 +67,9 @@ trait GraphElement {
 trait Input extends GraphElement {
 
   /**
-   * Returns a DataFrame that is a result of loading data from this [[Input]].
+   * Returns a [[DataFrame]] that is a result of loading data from this [[Input]].
    * @param readOptions Type of input. Used to determine streaming/batch
-   * @return Streaming or batch DataFrame of this Input's data.
+   * @return Streaming or batch [[DataFrame]] of this Input's data.
    */
   def load(readOptions: InputReadOptions): DataFrame
 }
@@ -82,7 +82,7 @@ sealed trait Output {
 
   /**
    * Normalized storage location used for storing materializations for this [[Output]].
-   * If None, it means this [[Output]] has not been normalized yet.
+   * If [[None]], it means this [[Output]] has not been normalized yet.
    */
   def normalizedPath: Option[String]