[SPARK-30018][SQL] Support ALTER DATABASE SET OWNER syntax

docs/sql-keywords.md

@@ -203,6 +203,7 @@ Below is a list of all the keywords in Spark SQL.
   <tr><td>OVERLAPS</td><td>reserved</td><td>non-reserved</td><td>reserved</td></tr>
   <tr><td>OVERLAY</td><td>non-reserved</td><td>non-reserved</td><td>non-reserved</td></tr>
   <tr><td>OVERWRITE</td><td>non-reserved</td><td>non-reserved</td><td>non-reserved</td></tr>
+  <tr><td>OWNER</td><td>non-reserved</td><td>non-reserved</td><td>non-reserved</td></tr>
   <tr><td>PARTITION</td><td>non-reserved</td><td>non-reserved</td><td>reserved</td></tr>
   <tr><td>PARTITIONED</td><td>non-reserved</td><td>non-reserved</td><td>non-reserved</td></tr>
   <tr><td>PARTITIONS</td><td>non-reserved</td><td>non-reserved</td><td>non-reserved</td></tr>

sql/catalyst/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBase.g4

@@ -97,6 +97,8 @@ statement
         SET (DBPROPERTIES | PROPERTIES) tablePropertyList              #setNamespaceProperties
     | ALTER (database | NAMESPACE) multipartIdentifier
         SET locationSpec                                               #setNamespaceLocation
+    | ALTER (database | NAMESPACE) multipartIdentifier
+        SET OWNER ownerType=(USER | ROLE | GROUP) identifier           #setNamespaceOwner
     | DROP (database | NAMESPACE) (IF EXISTS)? multipartIdentifier
         (RESTRICT | CASCADE)?                                          #dropNamespace
     | SHOW (DATABASES | NAMESPACES) ((FROM | IN) multipartIdentifier)?
@@ -1355,6 +1357,7 @@ nonReserved
     | OVERLAPS
     | OVERLAY
     | OVERWRITE
+    | OWNER
     | PARTITION
     | PARTITIONED
     | PARTITIONS
@@ -1623,6 +1626,7 @@ OVER: 'OVER';
 OVERLAPS: 'OVERLAPS';
 OVERLAY: 'OVERLAY';
 OVERWRITE: 'OVERWRITE';
+OWNER: 'OWNER';
 PARTITION: 'PARTITION';
 PARTITIONED: 'PARTITIONED';
 PARTITIONS: 'PARTITIONS';

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

@@ -64,9 +64,27 @@ public interface SupportsNamespaces extends CatalogPlugin {
   String PROP_OWNER_TYPE = "ownerType";
 
   /**
-   * The list of reserved namespace properties.
+   * The list of namespace ownership properties, cannot be used in `CREATE` syntax.
+   *
+   * Only support in:
+   *
+   * {{
+   *   ALTER (DATABASE|SCHEMA|NAMESPACE) SET OWNER ...
+   * }}
+   */
+  List<String> OWNERSHIPS = Arrays.asList(PROP_OWNER_NAME, PROP_OWNER_TYPE);
+
+  /**
+   * The list of immutable namespace properties, which can not be removed or changed directly by
+   * the syntax:
+   * {{
+   *   ALTER (DATABASE|SCHEMA|NAMESPACE) SET DBPROPERTIES(...)
+   * }}
+   *
+   * They need specific syntax to modify
    */
-  List<String> RESERVED_PROPERTIES = Arrays.asList(PROP_COMMENT, PROP_LOCATION);
+  List<String> REVERSED_PROPERTIES =
+    Arrays.asList(PROP_COMMENT, PROP_LOCATION, PROP_OWNER_NAME, PROP_OWNER_TYPE);
 
   /**
    * Return a default namespace for the catalog.

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

@@ -17,10 +17,13 @@
 
 package org.apache.spark.sql.catalyst.analysis
 
+import scala.collection.JavaConverters._
+
 import org.apache.spark.sql.AnalysisException
 import org.apache.spark.sql.catalyst.plans.logical._
 import org.apache.spark.sql.catalyst.rules.Rule
-import org.apache.spark.sql.connector.catalog.{CatalogManager, CatalogPlugin, LookupCatalog, SupportsNamespaces, TableCatalog, TableChange}
+import org.apache.spark.sql.connector.catalog.{CatalogManager, CatalogPlugin, LookupCatalog, TableCatalog, TableChange}
+import org.apache.spark.sql.connector.catalog.SupportsNamespaces._
 
 /**
  * Resolves catalogs from the multi-part identifiers in SQL statements, and convert the statements
@@ -94,11 +97,21 @@ class ResolveCatalogs(val catalogManager: CatalogManager)
           s"because view support in catalog has not been implemented yet")
 
     case AlterNamespaceSetPropertiesStatement(NonSessionCatalog(catalog, nameParts), properties) =>
+      if (properties.keySet.intersect(REVERSED_PROPERTIES.asScala.toSet).nonEmpty) {
+        throw new AnalysisException(s"Cannot directly modify the reversed properties" +
+          s" ${REVERSED_PROPERTIES.asScala.mkString("[", ",", "]")}.")
+      }
       AlterNamespaceSetProperties(catalog.asNamespaceCatalog, nameParts, properties)
 
     case AlterNamespaceSetLocationStatement(NonSessionCatalog(catalog, nameParts), location) =>
       AlterNamespaceSetProperties(catalog.asNamespaceCatalog, nameParts,
-        Map(SupportsNamespaces.PROP_LOCATION -> location))
+        Map(PROP_LOCATION -> location))
+
+    case AlterNamespaceSetOwner(CatalogAndIdentifierParts(catalog, parts), name, typ) =>
+      AlterNamespaceSetProperties(
+        catalog.asNamespaceCatalog,
+        parts,
+        Map(PROP_OWNER_NAME -> name, PROP_OWNER_TYPE -> typ))
 
     case RenameTableStatement(NonSessionCatalog(catalog, oldName), newNameParts, isView) =>
       if (isView) {
@@ -175,12 +188,11 @@ class ResolveCatalogs(val catalogManager: CatalogManager)
         s"Can not specify catalog `${catalog.name}` for view ${viewName.quoted} " +
           s"because view support in catalog has not been implemented yet")
 
-    case c @ CreateNamespaceStatement(NonSessionCatalog(catalog, nameParts), _, _) =>
-      CreateNamespace(
-        catalog.asNamespaceCatalog,
-        nameParts,
-        c.ifNotExists,
-        c.properties)
+    case c @ CreateNamespaceStatement(NonSessionCatalog(catalog, nameParts), _, properties) =>
+      if (properties.keySet.intersect(OWNERSHIPS.asScala.toSet).nonEmpty) {
+        throw new AnalysisException("Cannot specify the ownership in CREATE NAMESPACE.")
+      }
+      CreateNamespace(catalog.asNamespaceCatalog, nameParts, c.ifNotExists, properties)
 
     case DropNamespaceStatement(NonSessionCatalog(catalog, nameParts), ifExists, cascade) =>
       DropNamespace(catalog, nameParts, ifExists, cascade)

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala

@@ -2579,6 +2579,23 @@ class AstBuilder(conf: SQLConf) extends SqlBaseBaseVisitor[AnyRef] with Logging
     }
   }
 
+  /**
+   * Create an [[AlterNamespaceSetOwner]] logical plan.
+   *
+   * For example:
+   * {{{
+   *   ALTER (DATABASE|SCHEMA|NAMESPACE) namespace SET OWNER (USER|ROLE|GROUP) identityName;
+   * }}}
+   */
+  override def visitSetNamespaceOwner(ctx: SetNamespaceOwnerContext): LogicalPlan = {
+    withOrigin(ctx) {
+      AlterNamespaceSetOwner(
+        visitMultipartIdentifier(ctx.multipartIdentifier),
+        ctx.identifier.getText,
+        ctx.ownerType.getText)
+    }
+  }
+
   /**
    * Create a [[ShowNamespacesStatement]] command.
    */

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/statements.scala

@@ -382,6 +382,14 @@ case class AlterNamespaceSetLocationStatement(
     namespace: Seq[String],
     location: String) extends ParsedStatement
 
+/**
+ * ALTER (DATABASE|SCHEMA|NAMESPACE) ... SET OWNER command, as parsed from SQL.
+ */
+case class AlterNamespaceSetOwner(
+    namespace: Seq[String],
+    ownerName: String,
+    ownerType: String) extends ParsedStatement
+
 /**
  * A SHOW NAMESPACES statement, as parsed from SQL.
  */

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/DDLParserSuite.scala

@@ -1215,6 +1215,19 @@ class DDLParserSuite extends AnalysisTest {
       AlterNamespaceSetLocationStatement(Seq("a", "b", "c"), "/home/user/db"))
   }
 
+  test("set namespace owner") {
+    comparePlans(
+      parsePlan("ALTER DATABASE a.b.c SET OWNER USER user1"),
+      AlterNamespaceSetOwner(Seq("a", "b", "c"), "user1", "USER"))
+
+    comparePlans(
+      parsePlan("ALTER DATABASE a.b.c SET OWNER ROLE role1"),
+      AlterNamespaceSetOwner(Seq("a", "b", "c"), "role1", "ROLE"))
+    comparePlans(
+      parsePlan("ALTER DATABASE a.b.c SET OWNER GROUP group1"),
+      AlterNamespaceSetOwner(Seq("a", "b", "c"), "group1", "GROUP"))
+  }
+
   test("show databases: basic") {
     comparePlans(
       parsePlan("SHOW DATABASES"),

sql/core/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveSessionCatalog.scala

@@ -25,6 +25,7 @@ import org.apache.spark.sql.catalyst.catalog.{BucketSpec, CatalogTable, CatalogT
 import org.apache.spark.sql.catalyst.plans.logical._
 import org.apache.spark.sql.catalyst.rules.Rule
 import org.apache.spark.sql.connector.catalog.{CatalogManager, CatalogPlugin, LookupCatalog, SupportsNamespaces, Table, TableCatalog, TableChange, V1Table}
+import org.apache.spark.sql.connector.catalog.SupportsNamespaces._
 import org.apache.spark.sql.connector.expressions.Transform
 import org.apache.spark.sql.execution.command._
 import org.apache.spark.sql.execution.datasources.{CreateTable, DataSource, RefreshTable}
@@ -172,6 +173,10 @@ class ResolveSessionCatalog(
         throw new AnalysisException(
           s"The database name is not valid: ${nameParts.quoted}")
       }
+      if (properties.keySet.intersect(REVERSED_PROPERTIES.asScala.toSet).nonEmpty) {
+        throw new AnalysisException(s"Cannot directly modify the reversed properties" +
+          s" ${REVERSED_PROPERTIES.asScala.mkString("[", ",", "]")}.")
+      }
       AlterDatabasePropertiesCommand(nameParts.head, properties)
 
     case AlterNamespaceSetLocationStatement(SessionCatalog(_, nameParts), location) =>
@@ -302,10 +307,12 @@ class ResolveSessionCatalog(
         throw new AnalysisException(
           s"The database name is not valid: ${nameParts.quoted}")
       }
-
-      val comment = c.properties.get(SupportsNamespaces.PROP_COMMENT)
-      val location = c.properties.get(SupportsNamespaces.PROP_LOCATION)
-      val newProperties = c.properties -- SupportsNamespaces.RESERVED_PROPERTIES.asScala
+      if (c.properties.keySet.intersect(OWNERSHIPS.asScala.toSet).nonEmpty) {
+        throw new AnalysisException("Cannot specify the ownership in CREATE DATABASE.")
+      }
+      val comment = c.properties.get(PROP_COMMENT)
+      val location = c.properties.get(PROP_LOCATION)
+      val newProperties = c.properties -- REVERSED_PROPERTIES.asScala
       CreateDatabaseCommand(nameParts.head, c.ifNotExists, location, comment, newProperties)
 
     case d @ DropNamespaceStatement(SessionCatalog(_, nameParts), _, _) =>

sql/core/src/main/scala/org/apache/spark/sql/execution/command/ddl.scala

@@ -183,7 +183,7 @@ case class DescribeDatabaseCommand(
         Row("Owner Type", allDbProperties.getOrElse(PROP_OWNER_TYPE, "")) :: Nil
 
     if (extended) {
-      val properties = allDbProperties -- Seq(PROP_OWNER_NAME, PROP_OWNER_TYPE)
+      val properties = allDbProperties -- REVERSED_PROPERTIES.asScala
       val propertiesStr =
         if (properties.isEmpty) {
           ""

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala

@@ -22,8 +22,9 @@ import scala.collection.JavaConverters._
 import org.apache.spark.sql.{AnalysisException, Strategy}
 import org.apache.spark.sql.catalyst.expressions.{And, PredicateHelper, SubqueryExpression}
 import org.apache.spark.sql.catalyst.planning.PhysicalOperation
-import org.apache.spark.sql.catalyst.plans.logical.{AlterNamespaceSetProperties, AlterTable, AppendData, CreateNamespace, CreateTableAsSelect, CreateV2Table, DeleteFromTable, DescribeNamespace, DescribeTable, DropNamespace, DropTable, LogicalPlan, OverwriteByExpression, OverwritePartitionsDynamic, RefreshTable, RenameTable, Repartition, ReplaceTable, ReplaceTableAsSelect, SetCatalogAndNamespace, ShowCurrentNamespace, ShowNamespaces, ShowTableProperties, ShowTables}
+import org.apache.spark.sql.catalyst.plans.logical.{AlterNamespaceSetOwner, AlterNamespaceSetProperties, AlterTable, AppendData, CreateNamespace, CreateTableAsSelect, CreateV2Table, DeleteFromTable, DescribeNamespace, DescribeTable, DropNamespace, DropTable, LogicalPlan, OverwriteByExpression, OverwritePartitionsDynamic, RefreshTable, RenameTable, Repartition, ReplaceTable, ReplaceTableAsSelect, SetCatalogAndNamespace, ShowCurrentNamespace, ShowNamespaces, ShowTableProperties, ShowTables}
 import org.apache.spark.sql.connector.catalog.{StagingTableCatalog, TableCapability}
+import org.apache.spark.sql.connector.catalog.SupportsNamespaces.{PROP_OWNER_NAME, PROP_OWNER_TYPE}
 import org.apache.spark.sql.connector.read.streaming.{ContinuousStream, MicroBatchStream}
 import org.apache.spark.sql.execution.{FilterExec, ProjectExec, SparkPlan}
 import org.apache.spark.sql.execution.datasources.DataSourceStrategy

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DescribeNamespaceExec.scala

@@ -35,19 +35,21 @@ case class DescribeNamespaceExec(
     namespace: Seq[String],
     isExtended: Boolean) extends V2CommandExec {
   private val encoder = RowEncoder(StructType.fromAttributes(output)).resolveAndBind()
+  import SupportsNamespaces._
 
   override protected def run(): Seq[InternalRow] = {
     val rows = new ArrayBuffer[InternalRow]()
     val ns = namespace.toArray
     val metadata = catalog.loadNamespaceMetadata(ns)
 
     rows += toCatalystRow("Namespace Name", ns.last)
-    rows += toCatalystRow("Description", metadata.get(SupportsNamespaces.PROP_COMMENT))
-    rows += toCatalystRow("Location", metadata.get(SupportsNamespaces.PROP_LOCATION))
+    rows += toCatalystRow("Description", metadata.get(PROP_COMMENT))
+    rows += toCatalystRow("Location", metadata.get(PROP_LOCATION))
+    rows += toCatalystRow("Owner Name", metadata.get(PROP_OWNER_NAME))
+    rows += toCatalystRow("Owner Type", metadata.get(PROP_OWNER_TYPE))
+
     if (isExtended) {
-      val properties =
-        metadata.asScala.toSeq.filter(p =>
-          !SupportsNamespaces.RESERVED_PROPERTIES.contains(p._1))
+      val properties = metadata.asScala -- REVERSED_PROPERTIES.asScala
       if (properties.nonEmpty) {
         rows += toCatalystRow("Properties", properties.mkString("(", ",", ")"))
       }

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2SessionCatalog.scala

@@ -225,15 +225,6 @@ class V2SessionCatalog(catalog: SessionCatalog, conf: SQLConf)
   override def alterNamespace(namespace: Array[String], changes: NamespaceChange*): Unit = {
     namespace match {
       case Array(db) =>
-        // validate that this catalog's reserved properties are not removed
-        changes.foreach {
-          case remove: RemoveProperty
-            if SupportsNamespaces.RESERVED_PROPERTIES.contains(remove.property) =>
-            throw new UnsupportedOperationException(
-              s"Cannot remove reserved property: ${remove.property}")
-          case _ =>
-        }
-
         val metadata = catalog.getDatabaseMetadata(db).toMetadata
         catalog.alterDatabase(
           toCatalogDatabase(db, CatalogV2Util.applyNamespaceChanges(metadata, changes)))
@@ -263,6 +254,7 @@ class V2SessionCatalog(catalog: SessionCatalog, conf: SQLConf)
 }
 
 private[sql] object V2SessionCatalog {
+  import SupportsNamespaces._
 
   /**
    * Convert v2 Transforms to v1 partition columns and an optional bucket spec.
@@ -292,12 +284,12 @@ private[sql] object V2SessionCatalog {
       defaultLocation: Option[URI] = None): CatalogDatabase = {
     CatalogDatabase(
       name = db,
-      description = metadata.getOrDefault(SupportsNamespaces.PROP_COMMENT, ""),
-      locationUri = Option(metadata.get(SupportsNamespaces.PROP_LOCATION))
+      description = metadata.getOrDefault(PROP_COMMENT, ""),
+      locationUri = Option(metadata.get(PROP_LOCATION))
           .map(CatalogUtils.stringToURI)
           .orElse(defaultLocation)
           .getOrElse(throw new IllegalArgumentException("Missing database location")),
-      properties = metadata.asScala.toMap -- SupportsNamespaces.RESERVED_PROPERTIES.asScala)
+      properties = metadata.asScala.toMap -- Seq(PROP_COMMENT, PROP_LOCATION))
   }
 
   private implicit class CatalogDatabaseHelper(catalogDatabase: CatalogDatabase) {

sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/V2SessionCatalogSuite.scala

@@ -757,7 +757,7 @@ class V2SessionCatalogNamespaceSuite extends V2SessionCatalogBaseSuite {
       actual: scala.collection.Map[String, String]): Unit = {
     // remove location and comment that are automatically added by HMS unless they are expected
     val toRemove =
-      SupportsNamespaces.RESERVED_PROPERTIES.asScala.filter(expected.contains)
+      SupportsNamespaces.REVERSED_PROPERTIES.asScala.filter(expected.contains)
     assert(expected -- toRemove === actual)
   }
 

sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveDDLSuite.scala

@@ -374,11 +374,14 @@ class HiveCatalogedDDLSuite extends DDLSuite with TestHiveSingleton with BeforeA
     }
   }
 
-  private def checkOwner(db: String, expected: String): Unit = {
-    val owner = sql(s"DESCRIBE DATABASE EXTENDED $db")
-      .where("database_description_item='Owner Name'")
+  private def checkOwner(db: String, expectedOwnerName: String, expectedOwnerType: String): Unit = {
+    val df = sql(s"DESCRIBE DATABASE EXTENDED $db")
+    val owner = df.where("database_description_item='Owner Name'")
       .collect().head.getString(1)
-    assert(owner === expected)
+    val typ = df.where("database_description_item='Owner Type'")
+      .collect().head.getString(1)
+    assert(owner === expectedOwnerName)
+    assert(typ === expectedOwnerType)
   }
 
   test("Database Ownership") {
@@ -387,20 +390,29 @@ class HiveCatalogedDDLSuite extends DDLSuite with TestHiveSingleton with BeforeA
       val db1 = "spark_29425_1"
       val db2 = "spark_29425_2"
       val owner = "spark_29425"
+      val currentUser = Utils.getCurrentUserName()
 
       sql(s"CREATE DATABASE $db1")
-      checkOwner(db1, Utils.getCurrentUserName())
+      checkOwner(db1, currentUser, "USER")
       sql(s"ALTER DATABASE $db1 SET DBPROPERTIES ('a'='a')")
-      checkOwner(db1, Utils.getCurrentUserName())
-
-      // TODO: Specify ownership should be forbidden after we implement `SET OWNER` syntax
-      sql(s"CREATE DATABASE $db2 WITH DBPROPERTIES('ownerName'='$owner')")
-      checkOwner(db2, owner)
-      sql(s"ALTER DATABASE $db2 SET DBPROPERTIES ('a'='a')")
-      checkOwner(db2, owner)
-      // TODO: Changing ownership should be forbidden after we implement `SET OWNER` syntax
-      sql(s"ALTER DATABASE $db2 SET DBPROPERTIES ('ownerName'='a')")
-      checkOwner(db2, "a")
+      checkOwner(db1, currentUser, "USER")
+      val e = intercept[AnalysisException](sql(s"ALTER DATABASE $db1 SET DBPROPERTIES ('a'='a',"
+        + s"'ownerName'='$owner','ownerType'='XXX')"))
+      assert(e.getMessage.contains("ownerName"))
+      sql(s"ALTER DATABASE $db1 SET OWNER ROLE $owner")
+      checkOwner(db1, owner, "ROLE")
+
+      val e2 = intercept[AnalysisException](
+        sql(s"CREATE DATABASE $db2 WITH DBPROPERTIES('ownerName'='$owner', 'ownerType'='XXX')"))
+      assert(e2.getMessage.contains("ownership"))
+      sql(s"CREATE DATABASE $db2 WITH DBPROPERTIES('comment'='$owner')")
+      checkOwner(db2, currentUser, "USER")
+      sql(s"ALTER DATABASE $db2 SET OWNER GROUP $owner")
+      checkOwner(db2, owner, "GROUP")
+      sql(s"ALTER DATABASE $db2 SET OWNER GROUP `$owner`")
+      checkOwner(db2, owner, "GROUP")
+      sql(s"ALTER DATABASE $db2 SET OWNER GROUP OWNER")
+      checkOwner(db2, "OWNER", "GROUP")
     } finally {
       catalog.reset()
     }