Support define owner for datasource api

Author: yaooqinn

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

@@ -0,0 +1,10 @@
+package org.apache.spark.sql.connector.catalog;
+
+/**
+ * An enumeration support for Role-Based Access Control(RBAC) extensions.
+ */
+public enum PrincipalType {
+  USER,
+  GROUP,
+  ROLE
+}

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

@@ -20,6 +20,7 @@
 import org.apache.spark.annotation.Experimental;
 import org.apache.spark.sql.catalyst.analysis.NamespaceAlreadyExistsException;
 import org.apache.spark.sql.catalyst.analysis.NoSuchNamespaceException;
+import org.apache.spark.util.Utils;
 
 import java.util.Arrays;
 import java.util.List;
@@ -64,9 +65,7 @@ public interface SupportsNamespaces extends CatalogPlugin {
   String PROP_OWNER_TYPE = "ownerType";
 
   /**
-   * The list of namespace ownership properties, cannot be used in `CREATE` syntax.
-   *
-   * Only support in:
+   * The list of namespace ownership properties, which can be used in `ALTER` syntax:
    *
    * {{
    *   ALTER (DATABASE|SCHEMA|NAMESPACE) SET OWNER ...
@@ -86,6 +85,18 @@ public interface SupportsNamespaces extends CatalogPlugin {
   List<String> REVERSED_PROPERTIES =
     Arrays.asList(PROP_COMMENT, PROP_LOCATION, PROP_OWNER_NAME, PROP_OWNER_TYPE);
 
+  /**
+   * Specify the default owner name for `CREATE` namespace.
+   *
+   */
+  default String defaultOwner() { return Utils.getCurrentUserName(); }
+
+  /**
+   *
+   * Specify the default owner type for `CREATE` namespace.
+   */
+  default String defaultOwnerType() { return PrincipalType.USER.name(); }
+
   /**
    * Return a default namespace for the catalog.
    * <p>

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

@@ -39,7 +39,7 @@ import org.apache.spark.sql.catalyst.plans.logical._
 import org.apache.spark.sql.catalyst.util.DateTimeUtils.{getZoneId, stringToDate, stringToTimestamp}
 import org.apache.spark.sql.catalyst.util.IntervalUtils
 import org.apache.spark.sql.catalyst.util.IntervalUtils.IntervalUnit
-import org.apache.spark.sql.connector.catalog.SupportsNamespaces
+import org.apache.spark.sql.connector.catalog.{PrincipalType, SupportsNamespaces}
 import org.apache.spark.sql.connector.expressions.{ApplyTransform, BucketTransform, DaysTransform, Expression => V2Expression, FieldReference, HoursTransform, IdentityTransform, LiteralValue, MonthsTransform, Transform, YearsTransform}
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.types._
@@ -2589,10 +2589,17 @@ class AstBuilder(conf: SQLConf) extends SqlBaseBaseVisitor[AnyRef] with Logging
    */
   override def visitSetNamespaceOwner(ctx: SetNamespaceOwnerContext): LogicalPlan = {
     withOrigin(ctx) {
-      AlterNamespaceSetOwner(
-        visitMultipartIdentifier(ctx.multipartIdentifier),
-        ctx.identifier.getText,
-        ctx.ownerType.getText)
+      val ownerType = ctx.ownerType.getText
+      try {
+        AlterNamespaceSetOwner(
+          visitMultipartIdentifier(ctx.multipartIdentifier),
+          ctx.identifier.getText,
+          PrincipalType.valueOf(ownerType).name)
+      } catch {
+        case _: IllegalArgumentException =>
+          throw new ParseException(s"$ownerType is not supported, need to be one of" +
+            s" ${PrincipalType.values().mkString("[", ",", "]")}", ctx)
+      }
     }
   }
 

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

@@ -22,7 +22,7 @@ import scala.collection.JavaConverters.mapAsJavaMapConverter
 import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.analysis.NamespaceAlreadyExistsException
 import org.apache.spark.sql.catalyst.expressions.Attribute
-import org.apache.spark.sql.connector.catalog.SupportsNamespaces
+import org.apache.spark.sql.connector.catalog.{PrincipalType, SupportsNamespaces}
 
 /**
  * Physical plan node for creating a namespace.
@@ -35,11 +35,14 @@ case class CreateNamespaceExec(
     extends V2CommandExec {
   override protected def run(): Seq[InternalRow] = {
     import org.apache.spark.sql.connector.catalog.CatalogV2Implicits._
+    import org.apache.spark.sql.connector.catalog.SupportsNamespaces._
 
     val ns = namespace.toArray
     if (!catalog.namespaceExists(ns)) {
       try {
-        catalog.createNamespace(ns, properties.asJava)
+        val ownership = Map(PROP_OWNER_NAME -> catalog.defaultOwner(),
+          PROP_OWNER_TYPE -> PrincipalType.valueOf(catalog.defaultOwnerType).name())
+        catalog.createNamespace(ns, (properties ++ ownership).asJava)
       } catch {
         case _: NamespaceAlreadyExistsException if ifNotExists =>
           logWarning(s"Namespace ${namespace.quoted} was created concurrently. Ignoring.")

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

@@ -45,13 +45,13 @@ case class DescribeNamespaceExec(
     rows += toCatalystRow("Namespace Name", ns.last)
     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))
+    rows += toCatalystRow("Owner Name", metadata.getOrDefault(PROP_OWNER_NAME, ""))
+    rows += toCatalystRow("Owner Type", metadata.getOrDefault(PROP_OWNER_TYPE, ""))
 
     if (isExtended) {
       val properties = metadata.asScala -- REVERSED_PROPERTIES.asScala
       if (properties.nonEmpty) {
-        rows += toCatalystRow("Properties", properties.mkString("(", ",", ")"))
+        rows += toCatalystRow("Properties", properties.toSeq.mkString("(", ",", ")"))
       }
     }
     rows

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

@@ -226,6 +226,15 @@ class V2SessionCatalog(catalog: SessionCatalog, conf: SQLConf)
     namespace match {
       case Array(db) =>
         val metadata = catalog.getDatabaseMetadata(db).toMetadata
+        // validate that this catalog's reserved properties are not removed
+        changes.foreach {
+          case remove: RemoveProperty
+            if SupportsNamespaces.REVERSED_PROPERTIES.contains(remove.property) =>
+            throw new UnsupportedOperationException(
+              s"Cannot remove reserved property: ${remove.property}")
+          case _ =>
+        }
+
         catalog.alterDatabase(
           toCatalogDatabase(db, CatalogV2Util.applyNamespaceChanges(metadata, changes)))
 

sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala

@@ -29,6 +29,7 @@ import org.apache.spark.sql.internal.SQLConf.V2_SESSION_CATALOG_IMPLEMENTATION
 import org.apache.spark.sql.sources.SimpleScanSource
 import org.apache.spark.sql.types.{BooleanType, LongType, StringType, StructType}
 import org.apache.spark.sql.util.CaseInsensitiveStringMap
+import org.apache.spark.util.Utils
 
 class DataSourceV2SQLSuite
   extends InsertIntoTests(supportsDynamicOverwrite = true, includeSQLOnlyTests = true)
@@ -933,7 +934,9 @@ class DataSourceV2SQLSuite
       assert(description === Seq(
         Row("Namespace Name", "ns2"),
         Row("Description", "test namespace"),
-        Row("Location", "/tmp/ns_test")
+        Row("Location", "/tmp/ns_test"),
+        Row("Owner Name", Utils.getCurrentUserName()),
+        Row("Owner Type", PrincipalType.USER.name())
       ))
     }
   }
@@ -948,6 +951,8 @@ class DataSourceV2SQLSuite
         Row("Namespace Name", "ns2"),
         Row("Description", "test namespace"),
         Row("Location", "/tmp/ns_test"),
+        Row("Owner Name", Utils.getCurrentUserName()),
+        Row("Owner Type", PrincipalType.USER.name()),
         Row("Properties", "((a,b),(b,a),(c,c))")
       ))
     }
@@ -962,7 +967,25 @@ class DataSourceV2SQLSuite
       assert(descriptionDf.collect() === Seq(
         Row("Namespace Name", "ns2"),
         Row("Description", "test namespace"),
-        Row("Location", "/tmp/ns_test_2")
+        Row("Location", "/tmp/ns_test_2"),
+        Row("Owner Name", Utils.getCurrentUserName()),
+        Row("Owner Type", PrincipalType.USER.name())
+      ))
+    }
+  }
+
+  test("AlterNamespaceSetOwner using v2 catalog") {
+    withNamespace("testcat.ns1.ns2") {
+      sql("CREATE NAMESPACE IF NOT EXISTS testcat.ns1.ns2 COMMENT " +
+        "'test namespace' LOCATION '/tmp/ns_test_3'")
+      sql("ALTER NAMESPACE testcat.ns1.ns2 SET OWNER ROLE adminRole")
+      val descriptionDf = sql("DESCRIBE NAMESPACE EXTENDED testcat.ns1.ns2")
+      assert(descriptionDf.collect() === Seq(
+        Row("Namespace Name", "ns2"),
+        Row("Description", "test namespace"),
+        Row("Location", "/tmp/ns_test_3"),
+        Row("Owner Name", "adminRole"),
+        Row("Owner Type", PrincipalType.ROLE.name())
       ))
     }
   }

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

@@ -1010,31 +1010,18 @@ class V2SessionCatalogNamespaceSuite extends V2SessionCatalogBaseSuite {
     assert(exc.getMessage.contains(testNs.quoted))
   }
 
-  test("alterNamespace: fail to remove location") {
+  test("alterNamespace: fail to remove reversed properties") {
     val catalog = newCatalog()
 
     catalog.createNamespace(testNs, emptyProps)
 
-    val exc = intercept[UnsupportedOperationException] {
-      catalog.alterNamespace(testNs, NamespaceChange.removeProperty("location"))
-    }
-
-    assert(exc.getMessage.contains("Cannot remove reserved property: location"))
-
-    catalog.dropNamespace(testNs)
-  }
+    SupportsNamespaces.REVERSED_PROPERTIES.asScala.foreach { p =>
+      val exc = intercept[UnsupportedOperationException] {
+        catalog.alterNamespace(testNs, NamespaceChange.removeProperty(p))
+      }
+      assert(exc.getMessage.contains(s"Cannot remove reserved property: $p"))
 
-  test("alterNamespace: fail to remove comment") {
-    val catalog = newCatalog()
-
-    catalog.createNamespace(testNs, Map("comment" -> "test db").asJava)
-
-    val exc = intercept[UnsupportedOperationException] {
-      catalog.alterNamespace(testNs, NamespaceChange.removeProperty("comment"))
     }
-
-    assert(exc.getMessage.contains("Cannot remove reserved property: comment"))
-
     catalog.dropNamespace(testNs)
   }
 }