[KYUUBI #5446][AUTHZ] Support Create/Drop/Show/Reresh index command for Hudi

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

@@ -1635,6 +1635,20 @@
   } ],
   "opType" : "CREATETABLE",
   "queryDescs" : [ ]
+}, {
+  "classname" : "org.apache.spark.sql.hudi.command.CreateIndexCommand",
+  "tableDescs" : [ {
+    "fieldName" : "table",
+    "fieldExtractor" : "CatalogTableTableExtractor",
+    "columnDesc" : null,
+    "actionTypeDesc" : null,
+    "tableTypeDesc" : null,
+    "catalogDesc" : null,
+    "isInput" : false,
+    "setCurrentDatabaseIfMissing" : false
+  } ],
+  "opType" : "CREATEINDEX",
+  "queryDescs" : [ ]
 }, {
   "classname" : "org.apache.spark.sql.hudi.command.DeleteHoodieTableCommand",
   "tableDescs" : [ {
@@ -1674,6 +1688,20 @@
   } ],
   "opType" : "DROPTABLE",
   "queryDescs" : [ ]
+}, {
+  "classname" : "org.apache.spark.sql.hudi.command.DropIndexCommand",
+  "tableDescs" : [ {
+    "fieldName" : "table",
+    "fieldExtractor" : "CatalogTableTableExtractor",
+    "columnDesc" : null,
+    "actionTypeDesc" : null,
+    "tableTypeDesc" : null,
+    "catalogDesc" : null,
+    "isInput" : false,
+    "setCurrentDatabaseIfMissing" : false
+  } ],
+  "opType" : "DROPINDEX",
+  "queryDescs" : [ ]
 }, {
   "classname" : "org.apache.spark.sql.hudi.command.InsertIntoHoodieTableCommand",
   "tableDescs" : [ {
@@ -1716,6 +1744,20 @@
     "fieldName" : "mergeInto",
     "fieldExtractor" : "HudiMergeIntoSourceTableExtractor"
   } ]
+}, {
+  "classname" : "org.apache.spark.sql.hudi.command.RefreshIndexCommand",
+  "tableDescs" : [ {
+    "fieldName" : "table",
+    "fieldExtractor" : "CatalogTableTableExtractor",
+    "columnDesc" : null,
+    "actionTypeDesc" : null,
+    "tableTypeDesc" : null,
+    "catalogDesc" : null,
+    "isInput" : true,
+    "setCurrentDatabaseIfMissing" : false
+  } ],
+  "opType" : "ALTERINDEX_REBUILD",
+  "queryDescs" : [ ]
 }, {
   "classname" : "org.apache.spark.sql.hudi.command.RepairHoodieTableCommand",
   "tableDescs" : [ {
@@ -1747,6 +1789,20 @@
   } ],
   "opType" : "SHOWPARTITIONS",
   "queryDescs" : [ ]
+}, {
+  "classname" : "org.apache.spark.sql.hudi.command.ShowIndexesCommand",
+  "tableDescs" : [ {
+    "fieldName" : "table",
+    "fieldExtractor" : "CatalogTableTableExtractor",
+    "columnDesc" : null,
+    "actionTypeDesc" : null,
+    "tableTypeDesc" : null,
+    "catalogDesc" : null,
+    "isInput" : true,
+    "setCurrentDatabaseIfMissing" : false
+  } ],
+  "opType" : "SHOWINDEXES",
+  "queryDescs" : [ ]
 }, {
   "classname" : "org.apache.spark.sql.hudi.command.Spark31AlterTableCommand",
   "tableDescs" : [ {

extensions/spark/kyuubi-spark-authz/src/main/scala/org/apache/kyuubi/plugin/spark/authz/OperationType.scala

@@ -28,5 +28,6 @@ object OperationType extends Enumeration {
       CREATETABLE_AS_SELECT, CREATEFUNCTION, CREATEVIEW, DESCDATABASE, DESCFUNCTION, DESCTABLE,
       DROPDATABASE, DROPFUNCTION, DROPTABLE, DROPVIEW, EXPLAIN, LOAD, MSCK, QUERY, RELOADFUNCTION,
       SHOWCONF, SHOW_CREATETABLE, SHOWCOLUMNS, SHOWDATABASES, SHOWFUNCTIONS, SHOWPARTITIONS,
-      SHOWTABLES, SHOW_TBLPROPERTIES, SWITCHDATABASE, TRUNCATETABLE = Value
+      SHOWTABLES, SHOW_TBLPROPERTIES, SWITCHDATABASE, TRUNCATETABLE,
+      CREATEINDEX, DROPINDEX, ALTERINDEX_REBUILD, SHOWINDEXES = Value
 }

extensions/spark/kyuubi-spark-authz/src/main/scala/org/apache/kyuubi/plugin/spark/authz/ranger/AccessType.scala

@@ -25,7 +25,7 @@ object AccessType extends Enumeration {
 
   type AccessType = Value
 
-  val NONE, CREATE, ALTER, DROP, SELECT, UPDATE, USE, READ, WRITE, ALL, ADMIN = Value
+  val NONE, CREATE, ALTER, DROP, SELECT, UPDATE, USE, READ, WRITE, ALL, ADMIN, INDEX = Value
 
   def apply(obj: PrivilegeObject, opType: OperationType, isInput: Boolean): AccessType = {
     obj.actionType match {
@@ -48,14 +48,16 @@ object AccessType extends Enumeration {
               ALTERTABLE_REPLACECOLS |
               ALTERTABLE_SERDEPROPERTIES |
               ALTERVIEW_RENAME |
-              MSCK => ALTER
+              MSCK |
+              ALTERINDEX_REBUILD => ALTER
           case ALTERVIEW_AS => if (isInput) SELECT else ALTER
-          case DROPDATABASE | DROPTABLE | DROPFUNCTION | DROPVIEW => DROP
+          case DROPDATABASE | DROPTABLE | DROPFUNCTION | DROPVIEW | DROPINDEX => DROP
           case LOAD => if (isInput) SELECT else UPDATE
           case QUERY |
               SHOW_CREATETABLE |
               SHOW_TBLPROPERTIES |
               SHOWPARTITIONS |
+              SHOWINDEXES |
               ANALYZE_TABLE => SELECT
           case SHOWCOLUMNS | DESCTABLE => SELECT
           case SHOWDATABASES |
@@ -65,6 +67,7 @@ object AccessType extends Enumeration {
               SHOWFUNCTIONS |
               DESCFUNCTION => USE
           case TRUNCATETABLE => UPDATE
+          case CREATEINDEX => INDEX
           case _ => NONE
         }
       case PrivilegeObjectActionType.DELETE => DROP

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

@@ -145,6 +145,30 @@ object HudiCommands {
     TableCommandSpec(cmd, Seq(tableDesc), SHOW_TBLPROPERTIES)
   }
 
+  val CreateIndexCommand = {
+    val cmd = "org.apache.spark.sql.hudi.command.CreateIndexCommand"
+    val tableDesc = TableDesc("table", classOf[CatalogTableTableExtractor])
+    TableCommandSpec(cmd, Seq(tableDesc), CREATEINDEX)
+  }
+
+  val DropIndexCommand = {
+    val cmd = "org.apache.spark.sql.hudi.command.DropIndexCommand"
+    val tableDesc = TableDesc("table", classOf[CatalogTableTableExtractor])
+    TableCommandSpec(cmd, Seq(tableDesc), DROPINDEX)
+  }
+
+  val ShowIndexCommand = {
+    val cmd = "org.apache.spark.sql.hudi.command.ShowIndexesCommand"
+    val tableDesc = TableDesc("table", classOf[CatalogTableTableExtractor], isInput = true)
+    TableCommandSpec(cmd, Seq(tableDesc), SHOWINDEXES)
+  }
+
+  val RefreshIndexCommand = {
+    val cmd = "org.apache.spark.sql.hudi.command.RefreshIndexCommand"
+    val tableDesc = TableDesc("table", classOf[CatalogTableTableExtractor], isInput = true)
+    TableCommandSpec(cmd, Seq(tableDesc), ALTERINDEX_REBUILD)
+  }
+
   val InsertIntoHoodieTableCommand = {
     val cmd = "org.apache.spark.sql.hudi.command.InsertIntoHoodieTableCommand"
     val tableDesc = TableDesc(
@@ -228,13 +252,17 @@ object HudiCommands {
     CreateHoodieTableAsSelectCommand,
     CreateHoodieTableCommand,
     CreateHoodieTableLikeCommand,
+    CreateIndexCommand,
     CompactionHoodieTableCommand,
     CompactionShowHoodieTableCommand,
     DeleteHoodieTableCommand,
     DropHoodieTableCommand,
+    DropIndexCommand,
     InsertIntoHoodieTableCommand,
     MergeIntoHoodieTableCommand,
+    RefreshIndexCommand,
     RepairHoodieTableCommand,
+    ShowIndexCommand,
     TruncateHoodieTableCommand,
     ShowHoodieTablePartitionsCommand,
     Spark31AlterTableCommand,

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

@@ -53,6 +53,7 @@ class HudiCatalogRangerSparkExtensionSuite extends RangerSparkExtensionSuite {
   val table1 = "table1_hoodie"
   val table2 = "table2_hoodie"
   val outputTable1 = "outputTable_hoodie"
+  val index1 = "table_hoodie_index1"
 
   override def withFixture(test: NoArgTest): Outcome = {
     assume(isSupportedVersion)
@@ -522,4 +523,58 @@ class HudiCatalogRangerSparkExtensionSuite extends RangerSparkExtensionSuite {
       }
     }
   }
+
+  test("IndexBasedCommand") {
+    assume(
+      !isSparkV33OrGreater,
+      "Hudi index creation not supported on Spark 3.3 or greater currently")
+    withCleanTmpResources(Seq((s"$namespace1.$table1", "table"), (namespace1, "database"))) {
+      doAs(admin, sql(s"CREATE DATABASE IF NOT EXISTS $namespace1"))
+      doAs(
+        admin,
+        sql(
+          s"""
+             |CREATE TABLE IF NOT EXISTS $namespace1.$table1(id int, name string, city string)
+             |USING HUDI
+             |OPTIONS (
+             | type = 'cow',
+             | primaryKey = 'id',
+             | 'hoodie.datasource.hive_sync.enable' = 'false'
+             |)
+             |PARTITIONED BY(city)
+             |""".stripMargin))
+
+      // CreateIndexCommand
+      val createIndex = s"CREATE INDEX $index1 ON $namespace1.$table1 USING LUCENE (id)"
+      interceptContains[AccessControlException](
+        doAs(
+          someone,
+          sql(createIndex)))(s"does not have [alter] privilege on [$namespace1/$table1]")
+      doAs(admin, sql(createIndex))
+
+      // RefreshIndexCommand
+      val refreshIndex = s"REFRESH INDEX $index1 ON $namespace1.$table1"
+      interceptContains[AccessControlException](
+        doAs(
+          someone,
+          sql(refreshIndex)))(s"does not have [select] privilege on [$namespace1/$table1]")
+      doAs(admin, sql(refreshIndex))
+
+      // ShowIndexesCommand
+      val showIndex = s"SHOW INDEXES FROM TABLE $namespace1.$table1"
+      interceptContains[AccessControlException](
+        doAs(
+          someone,
+          sql(showIndex)))(s"does not have [select] privilege on [$namespace1/$table1]")
+      doAs(admin, sql(showIndex))
+
+      // DropIndexCommand
+      val dropIndex = s"DROP INDEX $index1 ON $namespace1.$table1"
+      interceptContains[AccessControlException](
+        doAs(
+          someone,
+          sql(dropIndex)))(s"does not have [alter] privilege on [$namespace1/$table1]")
+      doAs(admin, sql(dropIndex))
+    }
+  }
 }