[SPARK-39800][SQL][WIP] DataSourceV2: View Support

Author: jzhuge

sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/ViewCatalog.java

@@ -17,6 +17,8 @@
 
 package org.apache.spark.sql.connector.catalog;
 
+import java.util.Arrays;
+import java.util.List;
 import java.util.Map;
 
 import org.apache.spark.annotation.Experimental;
@@ -31,6 +33,35 @@
 @Experimental
 public interface ViewCatalog extends CatalogPlugin {
 
+  /**
+   * A reserved property to specify the description of the view.
+   */
+  String PROP_COMMENT = "comment";
+
+  /**
+   * A reserved property to specify the owner of the view.
+   */
+  String PROP_OWNER = "owner";
+
+  /**
+   * A reserved property to specify the software version used to create the view.
+   */
+  String PROP_CREATE_ENGINE_VERSION = "create_engine_version";
+
+  /**
+   * A reserved property to specify the software version used to change the view.
+   */
+  String PROP_ENGINE_VERSION = "engine_version";
+
+  /**
+   * All reserved properties of the view.
+   */
+  List<String> RESERVED_PROPERTIES = Arrays.asList(
+        PROP_COMMENT,
+        PROP_OWNER,
+        PROP_CREATE_ENGINE_VERSION,
+        PROP_ENGINE_VERSION);
+
   /**
    * List the views in a namespace from the catalog.
    * <p>

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala

@@ -35,6 +35,7 @@ import org.apache.spark.sql.catalyst.expressions.SubExprUtils._
 import org.apache.spark.sql.catalyst.expressions.aggregate._
 import org.apache.spark.sql.catalyst.expressions.objects._
 import org.apache.spark.sql.catalyst.optimizer.OptimizeUpdateFields
+import org.apache.spark.sql.catalyst.parser.{ParseException, ParserInterface}
 import org.apache.spark.sql.catalyst.plans._
 import org.apache.spark.sql.catalyst.plans.logical._
 import org.apache.spark.sql.catalyst.rules._
@@ -44,7 +45,7 @@ import org.apache.spark.sql.catalyst.trees.CurrentOrigin.withOrigin
 import org.apache.spark.sql.catalyst.trees.TreePattern._
 import org.apache.spark.sql.catalyst.util.{toPrettySQL, CharVarcharUtils}
 import org.apache.spark.sql.catalyst.util.ResolveDefaultColumns._
-import org.apache.spark.sql.connector.catalog.{View => _, _}
+import org.apache.spark.sql.connector.catalog.{View => V2View, _}
 import org.apache.spark.sql.connector.catalog.CatalogV2Implicits._
 import org.apache.spark.sql.connector.catalog.TableChange.{After, ColumnPosition}
 import org.apache.spark.sql.connector.catalog.functions.{AggregateFunction => V2AggregateFunction, BoundFunction, ScalarFunction, UnboundFunction}
@@ -238,6 +239,11 @@ class Analyzer(override val catalogManager: CatalogManager)
       errorOnExceed = true,
       maxIterationsSetting = SQLConf.ANALYZER_MAX_ITERATIONS.key)
 
+  /**
+   * Override to provide additional rules for the "Substitution" batch.
+   */
+  val extendedSubstitutionRules: Seq[Rule[LogicalPlan]] = Nil
+
   /**
    * Override to provide additional rules for the "Resolution" batch.
    */
@@ -262,11 +268,12 @@ class Analyzer(override val catalogManager: CatalogManager)
       // However, when manipulating deeply nested schema, `UpdateFields` expression tree could be
       // very complex and make analysis impossible. Thus we need to optimize `UpdateFields` early
       // at the beginning of analysis.
-      OptimizeUpdateFields,
-      CTESubstitution,
-      WindowsSubstitution,
-      EliminateUnions,
-      SubstituteUnresolvedOrdinals),
+      OptimizeUpdateFields +:
+      CTESubstitution +:
+      WindowsSubstitution +:
+      EliminateUnions +:
+      SubstituteUnresolvedOrdinals +:
+      extendedSubstitutionRules : _*),
     Batch("Disable Hints", Once,
       new ResolveHints.DisableHints),
     Batch("Hints", fixedPoint,
@@ -447,6 +454,74 @@ class Analyzer(override val catalogManager: CatalogManager)
     }
   }
 
+  /**
+   * Substitute persisted views in parsed plans with parsed view sql text.
+   */
+  case class ViewSubstitution(sqlParser: ParserInterface) extends Rule[LogicalPlan] {
+
+    def apply(plan: LogicalPlan): LogicalPlan = plan.resolveOperatorsUp {
+      case u @ UnresolvedRelation(nameParts, _, _) if v1SessionCatalog.isTempView(nameParts) =>
+        u
+      case u @ UnresolvedRelation(
+          parts @ NonSessionCatalogAndIdentifier(catalog, ident), _, _) if !isSQLOnFile(parts) =>
+        CatalogV2Util.loadView(catalog, ident)
+            .map(createViewRelation(parts.quoted, _))
+            .getOrElse(u)
+    }
+
+    private def isSQLOnFile(parts: Seq[String]): Boolean = parts match {
+      case Seq(_, path) if path.contains("/") => true
+      case _ => false
+    }
+
+    private def createViewRelation(name: String, view: V2View): LogicalPlan = {
+      if (!catalogManager.isCatalogRegistered(view.currentCatalog)) {
+        throw new AnalysisException(
+          s"Invalid current catalog '${view.currentCatalog}' in view '$name'")
+      }
+
+      val child = parseViewText(name, view.sql)
+      val desc = V2ViewDescription(name, view)
+      val qualifiedChild = desc.viewCatalogAndNamespace match {
+        case Seq() =>
+          // Views from Spark 2.2 or prior do not store catalog or namespace,
+          // however its sql text should already be fully qualified.
+          child
+        case catalogAndNamespace =>
+          // Substitute CTEs within the view before qualifying table identifiers
+          qualifyTableIdentifiers(CTESubstitution.apply(child), catalogAndNamespace)
+      }
+
+      // The relation is a view, so we wrap the relation by:
+      // 1. Add a [[View]] operator over the relation to keep track of the view desc;
+      // 2. Wrap the logical plan in a [[SubqueryAlias]] which tracks the name of the view.
+      SubqueryAlias(name, View(desc, false, qualifiedChild))
+    }
+
+    private def parseViewText(name: String, viewText: String): LogicalPlan = {
+      try {
+        sqlParser.parsePlan(viewText)
+      } catch {
+        case _: ParseException =>
+          throw QueryCompilationErrors.invalidViewText(viewText, name)
+      }
+    }
+
+    /**
+     * Qualify table identifiers with default catalog and namespace if necessary.
+     */
+    private def qualifyTableIdentifiers(
+        child: LogicalPlan,
+        catalogAndNamespace: Seq[String]): LogicalPlan =
+      child transform {
+        case u @ UnresolvedRelation(Seq(table), _, _) =>
+          u.copy(multipartIdentifier = catalogAndNamespace :+ table)
+        case u @ UnresolvedRelation(parts, _, _)
+            if !catalogManager.isCatalogRegistered(parts.head) =>
+          u.copy(multipartIdentifier = catalogAndNamespace.head +: parts)
+      }
+  }
+
   /**
    * Substitute child plan with WindowSpecDefinitions.
    */
@@ -1003,7 +1078,7 @@ class Analyzer(override val catalogManager: CatalogManager)
       // The view's child should be a logical plan parsed from the `desc.viewText`, the variable
       // `viewText` should be defined, or else we throw an error on the generation of the View
       // operator.
-      case view @ View(desc, isTempView, child) if !child.resolved =>
+      case view @ View(CatalogTableViewDescription(desc), isTempView, child) if !child.resolved =>
         // Resolve all the UnresolvedRelations and Views in the child.
         val newChild = AnalysisContext.withAnalysisContext(desc) {
           val nestedViewDepth = AnalysisContext.get.nestedViewDepth
@@ -1055,8 +1130,9 @@ class Analyzer(override val catalogManager: CatalogManager)
         write.table match {
           case u: UnresolvedRelation if !u.isStreaming =>
             lookupRelation(u).map(unwrapRelationPlan).map {
-              case v: View => throw QueryCompilationErrors.writeIntoViewNotAllowedError(
-                v.desc.identifier, write)
+              case View(CatalogTableViewDescription(desc), _, _) =>
+                throw QueryCompilationErrors.writeIntoViewNotAllowedError(
+                desc.identifier, write)
               case r: DataSourceV2Relation => write.withNewTable(r)
               case u: UnresolvedCatalogRelation =>
                 throw QueryCompilationErrors.writeIntoV1TableNotAllowedError(
@@ -1139,20 +1215,32 @@ class Analyzer(override val catalogManager: CatalogManager)
       }.orElse {
         expandIdentifier(identifier) match {
           case CatalogAndIdentifier(catalog, ident) =>
-            CatalogV2Util.loadTable(catalog, ident).map {
-              case v1Table: V1Table if CatalogV2Util.isSessionCatalog(catalog) &&
-                v1Table.v1Table.tableType == CatalogTableType.VIEW =>
-                val v1Ident = v1Table.catalogTable.identifier
-                val v2Ident = Identifier.of(v1Ident.database.toArray, v1Ident.identifier)
-                ResolvedView(v2Ident, isTemp = false)
-              case table =>
-                ResolvedTable.create(catalog.asTableCatalog, ident, table)
-            }
+            lookupView(catalog, ident)
+                .orElse(lookupTable(catalog, ident))
           case _ => None
         }
       }
     }
 
+    private def lookupView(catalog: CatalogPlugin, ident: Identifier): Option[LogicalPlan] =
+      CatalogV2Util.loadView(catalog, ident).map {
+        case _ if CatalogV2Util.isSessionCatalog(catalog) =>
+          ResolvedView(ident, isTemp = false)
+        case view =>
+          ResolvedV2View(catalog.asViewCatalog, ident, view)
+      }
+
+    private def lookupTable(catalog: CatalogPlugin, ident: Identifier): Option[LogicalPlan] =
+      CatalogV2Util.loadTable(catalog, ident).map {
+        case v1Table: V1Table if CatalogV2Util.isSessionCatalog(catalog) &&
+            v1Table.v1Table.tableType == CatalogTableType.VIEW =>
+          val v1Ident = v1Table.catalogTable.identifier
+          val v2Ident = Identifier.of(v1Ident.database.toArray, v1Ident.identifier)
+          ResolvedView(v2Ident, isTemp = false)
+        case table =>
+          ResolvedTable.create(catalog.asTableCatalog, ident, table)
+      }
+
     private def createRelation(
         catalog: CatalogPlugin,
         ident: Identifier,

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckAnalysis.scala

@@ -131,12 +131,6 @@ trait CheckAnalysis extends PredicateHelper with LookupCatalog {
       case u: UnresolvedTable =>
         u.failAnalysis(s"Table not found: ${u.multipartIdentifier.quoted}")
 
-      case u @ UnresolvedView(NonSessionCatalogAndIdentifier(catalog, ident), cmd, _, _) =>
-        u.failAnalysis(
-          s"Cannot specify catalog `${catalog.name}` for view ${ident.quoted} " +
-            "because view support in v2 catalog has not been implemented yet. " +
-            s"$cmd expects a view.")
-
       case u: UnresolvedView =>
         u.failAnalysis(s"View not found: ${u.multipartIdentifier.quoted}")
 
@@ -521,6 +515,44 @@ trait CheckAnalysis extends PredicateHelper with LookupCatalog {
           case alter: AlterTableCommand =>
             checkAlterTableCommand(alter)
 
+          case c: CreateView =>
+            if (c.originalText.isEmpty) {
+              throw new AnalysisException(
+                "'originalText' must be provided to create permanent view")
+            }
+
+            if (c.allowExisting && c.replace) {
+              throw new AnalysisException(
+                "CREATE VIEW with both IF NOT EXISTS and REPLACE is not allowed.")
+            }
+
+          // If the view output doesn't have the same number of columns neither with the child
+          // output, nor with the query column names, throw an AnalysisException.
+          // If the view's child output can't up cast to the view output,
+          // throw an AnalysisException, too.
+          case v @ View(desc, _, child) if child.resolved && v.output != child.output =>
+            val queryColumnNames = desc.viewQueryColumnNames
+            val queryOutput = if (queryColumnNames.nonEmpty) {
+              if (v.output.length != queryColumnNames.length) {
+                // If the view output doesn't have the same number of columns with the query column
+                // names, throw an AnalysisException.
+                throw new AnalysisException(
+                  s"The view output ${v.output.mkString("[", ",", "]")} doesn't have the same" +
+                    "number of columns with the query column names " +
+                    s"${queryColumnNames.mkString("[", ",", "]")}")
+              }
+              val resolver = SQLConf.get.resolver
+              queryColumnNames.map { colName =>
+                child.output.find { attr =>
+                  resolver(attr.name, colName)
+                }.getOrElse(throw new AnalysisException(
+                  s"Attribute with name '$colName' is not found in " +
+                    s"'${child.output.map(_.name).mkString("(", ",", ")")}'"))
+              }
+            } else {
+              child.output
+            }
+
           case _ => // Falls back to the following checks
         }
 

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveCatalogs.scala

@@ -19,14 +19,17 @@ package org.apache.spark.sql.catalyst.analysis
 
 import org.apache.spark.sql.catalyst.plans.logical._
 import org.apache.spark.sql.catalyst.rules.Rule
-import org.apache.spark.sql.connector.catalog.{CatalogManager, LookupCatalog}
+import org.apache.spark.sql.connector.catalog.{CatalogManager, CatalogPlugin, LookupCatalog, ViewChange}
+import org.apache.spark.sql.errors.QueryCompilationErrors
 
 /**
  * Resolves the catalog of the name parts for table/view/function/namespace.
  */
 class ResolveCatalogs(val catalogManager: CatalogManager)
   extends Rule[LogicalPlan] with LookupCatalog {
 
+  import org.apache.spark.sql.connector.catalog.CatalogV2Implicits._
+
   override def apply(plan: LogicalPlan): LogicalPlan = plan resolveOperators {
     case UnresolvedIdentifier(CatalogAndIdentifier(catalog, identifier)) =>
       ResolvedIdentifier(catalog, identifier)
@@ -44,5 +47,52 @@ class ResolveCatalogs(val catalogManager: CatalogManager)
       ResolvedNamespace(currentCatalog, Seq.empty[String])
     case UnresolvedNamespace(CatalogAndNamespace(catalog, ns)) =>
       ResolvedNamespace(catalog, ns)
+
+    case DescribeRelation(ResolvedV2View(_, ident, view), _, isExtended, _) =>
+      DescribeV2View(V2ViewDescription(ident.quoted, view), isExtended)
+
+    case ShowCreateTable(ResolvedV2View(_, ident, view), _, _) =>
+      ShowCreateV2View(V2ViewDescription(ident.quoted, view))
+
+    case ShowTableProperties(ResolvedV2View(_, ident, view), propertyKeys, _) =>
+      ShowV2ViewProperties(V2ViewDescription(ident.quoted, view), propertyKeys)
+
+    case SetViewProperties(ResolvedV2View(catalog, ident, _), props) =>
+      val changes = props.map {
+        case (property, value) => ViewChange.setProperty(property, value)
+      }.toSeq
+      AlterV2View(catalog, ident, changes)
+
+    case UnsetViewProperties(ResolvedV2View(catalog, ident, _), propertyKeys, ifExists) =>
+      if (!ifExists) {
+        val view = catalog.loadView(ident)
+        propertyKeys.filterNot(view.properties.containsKey).foreach { property =>
+          QueryCompilationErrors.cannotUnsetNonExistentViewProperty(ident, property)
+        }
+      }
+      val changes = propertyKeys.map(ViewChange.removeProperty)
+      AlterV2View(catalog, ident, changes)
+
+    case RenameTable(ResolvedV2View(oldCatalog, oldIdent, _),
+        NonSessionCatalogAndIdentifier(newCatalog, newIdent), true) =>
+      if (oldCatalog.name != newCatalog.name) {
+        QueryCompilationErrors.cannotMoveViewBetweenCatalogs(
+          oldCatalog.name, newCatalog.name)
+      }
+      RenameV2View(oldCatalog, oldIdent, newIdent)
+
+    case RefreshTable(ResolvedV2View(catalog, ident, _)) =>
+      RefreshView(catalog, ident)
+
+    case DropView(ResolvedV2View(catalog, ident, _), ifExists) =>
+      DropV2View(catalog, ident, ifExists)
+  }
+
+  object NonSessionCatalogAndTable {
+    def unapply(nameParts: Seq[String]): Option[(CatalogPlugin, Seq[String])] = nameParts match {
+      case NonSessionCatalogAndIdentifier(catalog, ident) =>
+        Some(catalog -> ident.asMultipartIdentifier)
+      case _ => None
+    }
   }
 }