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

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

@@ -66,7 +66,8 @@ public interface SupportsNamespaces extends CatalogPlugin {
   /**
    * The list of reserved namespace properties.
    */
-  List<String> RESERVED_PROPERTIES = Arrays.asList(PROP_COMMENT, PROP_LOCATION);
+  List<String> RESERVED_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,18 @@ class ResolveCatalogs(val catalogManager: CatalogManager)
           s"because view support in catalog has not been implemented yet")
 
     case AlterNamespaceSetPropertiesStatement(NonSessionCatalog(catalog, nameParts), properties) =>
-      AlterNamespaceSetProperties(catalog.asNamespaceCatalog, nameParts, properties)
+      val availableProps = properties -- RESERVED_PROPERTIES.asScala
+      AlterNamespaceSetProperties(catalog.asNamespaceCatalog, nameParts, availableProps)
 
     case AlterNamespaceSetLocationStatement(NonSessionCatalog(catalog, nameParts), location) =>
       AlterNamespaceSetProperties(catalog.asNamespaceCatalog, nameParts,
-        Map(SupportsNamespaces.PROP_LOCATION -> location))
+        Map(PROP_LOCATION -> location))
+
+    case AlterNamespaceSetOwnerStatement(NonSessionCatalog(catalog, nameParts), name, typ) =>
+      AlterNamespaceSetProperties(
+        catalog.asNamespaceCatalog,
+        nameParts,
+        Map(PROP_OWNER_NAME -> name, PROP_OWNER_TYPE -> typ))
 
     case RenameTableStatement(NonSessionCatalog(catalog, oldName), newNameParts, isView) =>
       if (isView) {

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 [[AlterNamespaceSetOwnerStatement]] logical plan.
+   *
+   * For example:
+   * {{{
+   *   ALTER (DATABASE|SCHEMA|NAMESPACE) namespace SET OWNER (USER|ROLE|GROUP) identityName;
+   * }}}
+   */
+  override def visitSetNamespaceOwner(ctx: SetNamespaceOwnerContext): LogicalPlan = {
+    withOrigin(ctx) {
+      AlterNamespaceSetOwnerStatement(
+        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 AlterNamespaceSetOwnerStatement(
+    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"),
+      AlterNamespaceSetOwnerStatement(Seq("a", "b", "c"), "user1", "USER"))
+
+    comparePlans(
+      parsePlan("ALTER DATABASE a.b.c SET OWNER ROLE role1"),
+      AlterNamespaceSetOwnerStatement(Seq("a", "b", "c"), "role1", "ROLE"))
+    comparePlans(
+      parsePlan("ALTER DATABASE a.b.c SET OWNER GROUP group1"),
+      AlterNamespaceSetOwnerStatement(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

@@ -181,6 +181,13 @@ class ResolveSessionCatalog(
       }
       AlterDatabaseSetLocationCommand(nameParts.head, location)
 
+    case AlterNamespaceSetOwnerStatement(SessionCatalog(_, nameParts), ownerName, ownerType) =>
+      if (nameParts.length != 1) {
+        throw new AnalysisException(
+          s"The database name is not valid: ${nameParts.quoted}")
+      }
+      AlterDatabaseSetOwnerCommand(nameParts.head, ownerName, ownerType)
+
     case RenameTableStatement(SessionCatalog(_, oldName), newNameParts, isView) =>
       AlterTableRenameCommand(oldName.asTableIdentifier, newNameParts.asTableIdentifier, isView)
 

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

@@ -129,7 +129,8 @@ case class AlterDatabasePropertiesCommand(
   override def run(sparkSession: SparkSession): Seq[Row] = {
     val catalog = sparkSession.sessionState.catalog
     val db: CatalogDatabase = catalog.getDatabaseMetadata(databaseName)
-    catalog.alterDatabase(db.copy(properties = db.properties ++ props))
+    val availableProps = props -- RESERVED_PROPERTIES.asScala
+    catalog.alterDatabase(db.copy(properties = db.properties ++ availableProps))
 
     Seq.empty[Row]
   }
@@ -156,6 +157,26 @@ case class AlterDatabaseSetLocationCommand(databaseName: String, location: Strin
   }
 }
 
+/**
+ * A command for users to set ownership for a database
+ * If the database does not exist, an error message will be issued to indicate the database
+ * does not exist.
+ * The syntax of using this command in SQL is:
+ * {{{
+ *    ALTER (DATABASE|SCHEMA) database_name SET OWNER [USER|ROLE|GROUP] identityName
+ * }}}
+ */
+case class AlterDatabaseSetOwnerCommand(databaseName: String, ownerName: String, ownerType: String)
+  extends RunnableCommand {
+  override def run(sparkSession: SparkSession): Seq[Row] = {
+    val catalog = sparkSession.sessionState.catalog
+    val database = catalog.getDatabaseMetadata(databaseName)
+    val ownerships = Map(PROP_OWNER_NAME -> ownerName, PROP_OWNER_TYPE -> ownerType)
+    catalog.alterDatabase(database.copy(properties = database.properties ++ ownerships))
+    Seq.empty[Row]
+  }
+}
+
 /**
  * A command for users to show the name of the database, its comment (if one has been set), and its
  * root location on the filesystem. When extended is true, it also shows the database's properties
@@ -183,7 +204,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 -- RESERVED_PROPERTIES.asScala
       val propertiesStr =
         if (properties.isEmpty) {
           ""

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,23 @@ 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())
-      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")
+      sql(s"ALTER DATABASE $db1 SET DBPROPERTIES ('a'='a', 'ownerName'='$owner'," +
+        s" 'ownerType'='XXX')")
+      checkOwner(db1, currentUser, "USER")
+      sql(s"ALTER DATABASE $db1 SET OWNER ROLE $owner")
+      checkOwner(db1, owner, "ROLE")
+
+      sql(s"CREATE DATABASE $db2 WITH DBPROPERTIES('ownerName'='$owner', 'ownerType'='XXX')")
+      checkOwner(db2, currentUser, "USER")
+      sql(s"ALTER DATABASE $db2 SET DBPROPERTIES ('a'='a', 'ownerName'='$owner'," +
+        s" 'ownerType'='XXX')")
+      checkOwner(db2, currentUser, "USER")
+      sql(s"ALTER DATABASE $db2 SET OWNER GROUP $owner")
+      checkOwner(db2, owner, "GROUP")
     } finally {
       catalog.reset()
     }