[KYUUBI #5529][AUTHZ] Support create table command for Delta Lake

extensions/spark/kyuubi-spark-authz/pom.xml

@@ -329,6 +329,12 @@
             <artifactId>paimon-spark-${paimon.spark.binary.version}</artifactId>
             <scope>test</scope>
         </dependency>
+
+        <dependency>
+            <groupId>io.delta</groupId>
+            <artifactId>${delta.artifact}_${scala.binary.version}</artifactId>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
     <build>

extensions/spark/kyuubi-spark-authz/src/main/resources/table_command_spec.json

@@ -671,6 +671,20 @@
   } ],
   "opType" : "QUERY",
   "queryDescs" : [ ]
+}, {
+  "classname" : "org.apache.spark.sql.delta.commands.CreateDeltaTableCommand",
+  "tableDescs" : [ {
+    "fieldName" : "table",
+    "fieldExtractor" : "CatalogTableTableExtractor",
+    "columnDesc" : null,
+    "actionTypeDesc" : null,
+    "tableTypeDesc" : null,
+    "catalogDesc" : null,
+    "isInput" : false,
+    "setCurrentDatabaseIfMissing" : false
+  } ],
+  "opType" : "CREATETABLE",
+  "queryDescs" : [ ]
 }, {
   "classname" : "org.apache.spark.sql.execution.command.AlterTableAddColumnsCommand",
   "tableDescs" : [ {

extensions/spark/kyuubi-spark-authz/src/test/scala/org/apache/kyuubi/plugin/spark/authz/RangerTestResources.scala

@@ -41,6 +41,7 @@ object RangerTestNamespace {
   val sparkCatalog = "spark_catalog"
   val icebergNamespace = "iceberg_ns"
   val hudiNamespace = "hudi_ns"
+  val deltaNamespace = "delta_ns"
   val namespace1 = "ns1"
   val namespace2 = "ns2"
 }

extensions/spark/kyuubi-spark-authz/src/test/scala/org/apache/kyuubi/plugin/spark/authz/gen/DeltaCommands.scala

@@ -0,0 +1,33 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.plugin.spark.authz.gen
+
+import org.apache.kyuubi.plugin.spark.authz.OperationType._
+import org.apache.kyuubi.plugin.spark.authz.serde._
+
+object DeltaCommands {
+
+  val CreateDeltaTableCommand = {
+    val cmd = "org.apache.spark.sql.delta.commands.CreateDeltaTableCommand"
+    val tableDesc = TableDesc("table", classOf[CatalogTableTableExtractor])
+    TableCommandSpec(cmd, Seq(tableDesc), CREATETABLE)
+  }
+
+  val data: Array[TableCommandSpec] = Array(
+    CreateDeltaTableCommand)
+}

extensions/spark/kyuubi-spark-authz/src/test/scala/org/apache/kyuubi/plugin/spark/authz/gen/JsonSpecFileGenerator.scala

@@ -43,7 +43,9 @@ class JsonSpecFileGenerator extends AnyFunSuite {
   // scalastyle:on
   test("check spec json files") {
     writeCommandSpecJson("database", DatabaseCommands.data)
-    writeCommandSpecJson("table", TableCommands.data ++ IcebergCommands.data ++ HudiCommands.data)
+    writeCommandSpecJson(
+      "table",
+      TableCommands.data ++ IcebergCommands.data ++ HudiCommands.data ++ DeltaCommands.data)
     writeCommandSpecJson("function", FunctionCommands.data)
     writeCommandSpecJson("scan", Scans.data)
   }

extensions/spark/kyuubi-spark-authz/src/test/scala/org/apache/kyuubi/plugin/spark/authz/ranger/DeltaCatalogRangerSparkExtensionSuite.scala

@@ -0,0 +1,123 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kyuubi.plugin.spark.authz.ranger
+
+import org.scalatest.Outcome
+
+import org.apache.kyuubi.Utils
+import org.apache.kyuubi.plugin.spark.authz.AccessControlException
+import org.apache.kyuubi.plugin.spark.authz.RangerTestNamespace._
+import org.apache.kyuubi.plugin.spark.authz.RangerTestUsers._
+import org.apache.kyuubi.tags.DeltaTest
+import org.apache.kyuubi.util.AssertionUtils._
+
+/**
+ * Tests for RangerSparkExtensionSuite on Delta Lake
+ */
+@DeltaTest
+class DeltaCatalogRangerSparkExtensionSuite extends RangerSparkExtensionSuite {
+  override protected val catalogImpl: String = "hive"
+
+  val namespace1 = deltaNamespace
+  val table1 = "table1_delta"
+
+  override def withFixture(test: NoArgTest): Outcome = {
+    test()
+  }
+
+  override def beforeAll(): Unit = {
+    spark.conf.set(
+      s"spark.sql.catalog.$sparkCatalog",
+      "org.apache.spark.sql.delta.catalog.DeltaCatalog")
+    spark.conf.set(
+      s"spark.sql.catalog.$sparkCatalog.warehouse",
+      Utils.createTempDir("delta-hadoop").toString)
+    super.beforeAll()
+    doAs(admin, sql(s"CREATE DATABASE IF NOT EXISTS $namespace1"))
+  }
+
+  override def afterAll(): Unit = {
+    doAs(admin, sql(s"DROP DATABASE IF EXISTS $namespace1"))
+    super.afterAll()
+    spark.sessionState.catalog.reset()
+    spark.sessionState.conf.clear()
+  }
+
+  test("CreateDeltaTableCommand") {
+    withCleanTmpResources(Seq((s"$namespace1.$table1", "table"))) {
+      val createTableSql =
+        s"""
+           |CREATE TABLE IF NOT EXISTS $namespace1.$table1 (
+           |  id INT,
+           |  firstName STRING,
+           |  middleName STRING,
+           |  lastName STRING,
+           |  gender STRING,
+           |  birthDate TIMESTAMP,
+           |  ssn STRING,
+           |  salary INT
+           |) USING DELTA
+           |""".stripMargin
+      interceptContains[AccessControlException] {
+        doAs(someone, sql(createTableSql))
+      }(s"does not have [create] privilege on [$namespace1/$table1]")
+      doAs(admin, createTableSql)
+    }
+
+    withCleanTmpResources(Seq((s"$namespace1.$table1", "table"))) {
+      val createOrReplaceTableSql =
+        s"""
+           |CREATE OR REPLACE TABLE $namespace1.$table1 (
+           |  id INT,
+           |  firstName STRING,
+           |  middleName STRING,
+           |  lastName STRING,
+           |  gender STRING,
+           |  birthDate TIMESTAMP,
+           |  ssn STRING,
+           |  salary INT
+           |) USING DELTA
+           |""".stripMargin
+      interceptContains[AccessControlException] {
+        doAs(someone, sql(createOrReplaceTableSql))
+      }(s"does not have [create] privilege on [$namespace1/$table1]")
+      doAs(admin, createOrReplaceTableSql)
+    }
+
+    withCleanTmpResources(Seq((s"$namespace1.$table1", "table"))) {
+      val createPartitionTableSql =
+        s"""
+           |CREATE TABLE IF NOT EXISTS $namespace1.$table1 (
+           |  id INT,
+           |  firstName STRING,
+           |  middleName STRING,
+           |  lastName STRING,
+           |  gender STRING,
+           |  birthDate TIMESTAMP,
+           |  ssn STRING,
+           |  salary INT
+           |)
+           |USING DELTA
+           |PARTITIONED BY (gender)
+           |""".stripMargin
+      interceptContains[AccessControlException] {
+        doAs(someone, sql(createPartitionTableSql))
+      }(s"does not have [create] privilege on [$namespace1/$table1]")
+      doAs(admin, createPartitionTableSql)
+    }
+  }
+}