wenchen feedback and try to fix build errors

Author: sryza

common/utils/src/main/resources/error/error-conditions.json

@@ -4607,12 +4607,6 @@
     ],
     "sqlState" : "42K03"
   },
-  "PERSISTED_VIEW_READS_FROM_TEMPORARY_VIEW" : {
-    "message" : [
-      "Persisted view <persistedViewName> cannot reference temporary view <temporaryViewName> that will not be available outside the pipeline scope. Either make the persisted view temporary or persist the temporary view."
-    ],
-    "sqlState" : "42K0F"
-  },
   "PIPE_OPERATOR_AGGREGATE_EXPRESSION_CONTAINS_NO_AGGREGATE_FUNCTION" : {
     "message" : [
       "Non-grouping expression <expr> is provided as an argument to the |> AGGREGATE pipe operator but does not contain any aggregate function; please update it to include an aggregate function and then retry the query again."

sql/pipelines/pom.xml

@@ -59,12 +59,6 @@
             <groupId>org.apache.spark</groupId>
             <artifactId>spark-sql_${scala.binary.version}</artifactId>
             <version>${project.version}</version>
-            <exclusions>
-                <exclusion>
-                    <groupId>org.apache.spark</groupId>
-                    <artifactId>spark-connect-shims_${scala.binary.version}</artifactId>
-                </exclusion>
-            </exclusions>
         </dependency>
         <dependency>
             <groupId>org.apache.spark</groupId>

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/GraphErrors.scala

@@ -128,22 +128,4 @@ object GraphErrors {
       cause = Option(cause.orNull)
     )
   }
-
-  /**
-   * Throws an error when a persisted view is trying to read from a temporary view.
-   *
-   * @param persistedViewIdentifier the identifier of the persisted view
-   * @param temporaryViewIdentifier the identifier of the temporary view
-   */
-  def persistedViewReadsFromTemporaryView(
-      persistedViewIdentifier: TableIdentifier,
-      temporaryViewIdentifier: TableIdentifier): AnalysisException = {
-    new AnalysisException(
-      "PERSISTED_VIEW_READS_FROM_TEMPORARY_VIEW",
-      Map(
-        "persistedViewName" -> persistedViewIdentifier.toString,
-        "temporaryViewName" -> temporaryViewIdentifier.toString
-      )
-    )
-  }
 }

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

@@ -252,9 +252,13 @@ trait GraphValidations extends Logging {
           .flatMap(view.get)
           .foreach {
             case tempView: TemporaryView =>
-              throw GraphErrors.persistedViewReadsFromTemporaryView(
-                persistedViewIdentifier = persistedView.identifier,
-                temporaryViewIdentifier = tempView.identifier
+              throw new AnalysisException(
+                errorClass = "INVALID_TEMP_OBJ_REFERENCE",
+                messageParameters = Map(
+                  "persistedViewName" -> persistedView.identifier.toString,
+                  "temporaryViewName" -> tempView.identifier.toString
+                ),
+                cause = null
               )
             case _ =>
           }

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
 }