1

Author: aakash-db

sql/pipelines/pom.xml

@@ -38,6 +38,13 @@
             <artifactId>spark-core_${scala.binary.version}</artifactId>
             <version>${project.version}</version>
         </dependency>
+        <dependency>
+            <groupId>org.apache.spark</groupId>
+            <artifactId>spark-core_${scala.binary.version}</artifactId>
+            <version>${project.version}</version>
+            <type>test-jar</type>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>org.apache.spark</groupId>
             <artifactId>spark-sql_${scala.binary.version}</artifactId>
@@ -49,13 +56,6 @@
                 </exclusion>
             </exclusions>
         </dependency>
-        <dependency>
-            <groupId>org.apache.spark</groupId>
-            <artifactId>spark-core_${scala.binary.version}</artifactId>
-            <version>${project.version}</version>
-            <type>test-jar</type>
-            <scope>test</scope>
-        </dependency>
         <dependency>
             <groupId>org.apache.spark</groupId>
             <artifactId>spark-catalyst_${scala.binary.version}</artifactId>

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

@@ -183,8 +183,9 @@ private class FlowResolver(rawGraph: DataflowGraph) extends Logging {
           }
           convertResolvedToTypedFlow(flowToResolve, maybeNewFuncResult)
 
-        // If flow failed due to unresolved dataset, throw a retryable exception, otherwise just
-        // return the failed flow.
+        // If the flow failed due to an UnresolvedDatasetException, it means that one of the
+        // flow's inputs wasn't available. After other flows are resolved, these inputs
+        // may become available, so throw a retryable exception in this case.
         case f =>
           f.dataFrame.failed.toOption.collectFirst {
             case e: UnresolvedDatasetException => e

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

@@ -113,7 +113,6 @@ sealed trait TableInput extends Input {
  * @param normalizedPath Normalized storage location for the table based on the user-specified table
  *                       path (if not defined, we will normalize a managed storage path for it).
  * @param properties Table Properties to set in table metadata.
- * @param sqlText For SQL-defined pipelines, the original string of the SELECT query.
  * @param comment User-specified comment that can be placed on the table.
  * @param isStreamingTableOpt if the table is a streaming table, will be None until we have resolved
  *                            flows into table
@@ -124,7 +123,6 @@ case class Table(
     partitionCols: Option[Seq[String]],
     normalizedPath: Option[String],
     properties: Map[String, String] = Map.empty,
-    sqlText: Option[String],
     comment: Option[String],
     baseOrigin: QueryOrigin,
     isStreamingTableOpt: Option[Boolean],
@@ -239,9 +237,6 @@ trait View extends GraphElement {
   /** Properties of this view */
   val properties: Map[String, String]
 
-  /** (SQL-specific) The raw query that defines the [[View]]. */
-  val sqlText: Option[String]
-
   /** User-specified comment that can be placed on the [[View]]. */
   val comment: Option[String]
 }
@@ -251,13 +246,11 @@ trait View extends GraphElement {
  *
  * @param identifier The identifier of this view within the graph.
  * @param properties Properties of the view
- * @param sqlText Raw SQL query that defines the view.
  * @param comment when defining a view
  */
 case class TemporaryView(
     identifier: TableIdentifier,
     properties: Map[String, String],
-    sqlText: Option[String],
     comment: Option[String],
     origin: QueryOrigin
 ) extends View {}
@@ -267,13 +260,11 @@ case class TemporaryView(
  *
  * @param identifier The identifier of this view within the graph.
  * @param properties Properties of the view
- * @param sqlText Raw SQL query that defines the view.
  * @param comment when defining a view
  */
 case class PersistedView(
     identifier: TableIdentifier,
     properties: Map[String, String],
-    sqlText: Option[String],
     comment: Option[String],
     origin: QueryOrigin
 ) extends View {}

sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/utils/PipelineTest.scala

@@ -92,23 +92,6 @@ abstract class PipelineTest
     }
   }
 
-  /**
-   * Runs the given function with the given spark conf, and resets the conf after the function
-   * completes.
-   */
-  def withSparkConfs[T](confs: Map[String, String])(f: => T): T = {
-    val originalConfs = confs.keys.map(k => k -> spark.conf.getOption(k)).toMap
-    confs.foreach { case (k, v) => spark.conf.set(k, v) }
-    try f
-    finally originalConfs.foreach {
-      case (k, v) =>
-        v match {
-          case Some(v) => spark.conf.set(k, v)
-          case None => spark.conf.unset(k)
-        }
-    }
-  }
-
   /**
    * This exists temporarily for compatibility with tests that become invalid when multiple
    * executors are available.

sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/utils/TestGraphRegistrationContext.scala

@@ -55,7 +55,6 @@ class TestGraphRegistrationContext(
       query: Option[FlowFunction] = None,
       sqlConf: Map[String, String] = Map.empty,
       comment: Option[String] = None,
-      sqlText: Option[String] = None,
       specifiedSchema: Option[StructType] = None,
       partitionCols: Option[Seq[String]] = None,
       properties: Map[String, String] = Map.empty,
@@ -70,7 +69,6 @@ class TestGraphRegistrationContext(
       Table(
         identifier = GraphIdentifierManager.parseTableIdentifier(name, spark),
         comment = comment,
-        sqlText = sqlText,
         specifiedSchema = specifiedSchema,
         partitionCols = partitionCols,
         properties = properties,
@@ -105,7 +103,6 @@ class TestGraphRegistrationContext(
       query: FlowFunction,
       sqlConf: Map[String, String] = Map.empty,
       comment: Option[String] = None,
-      sqlText: Option[String] = None,
       origin: QueryOrigin = QueryOrigin.empty,
       viewType: ViewType = LocalTempView,
       catalog: Option[String] = None,
@@ -121,15 +118,13 @@ class TestGraphRegistrationContext(
           TemporaryView(
             identifier = viewIdentifier,
             comment = comment,
-            sqlText = sqlText,
             origin = origin,
             properties = Map.empty
           )
         case _ =>
           PersistedView(
             identifier = viewIdentifier,
             comment = comment,
-            sqlText = sqlText,
             origin = origin,
             properties = Map.empty
           )