Support optimize path-based table for Delta Lake in Authz

Author: zml1206

extensions/spark/kyuubi-spark-authz/src/main/resources/META-INF/services/org.apache.kyuubi.plugin.spark.authz.serde.URIExtractor

@@ -23,7 +23,10 @@ org.apache.kyuubi.plugin.spark.authz.serde.IdentifierURIExtractor
 org.apache.kyuubi.plugin.spark.authz.serde.PartitionLocsSeqURIExtractor
 org.apache.kyuubi.plugin.spark.authz.serde.PropertiesLocationUriExtractor
 org.apache.kyuubi.plugin.spark.authz.serde.PropertiesPathUriExtractor
+org.apache.kyuubi.plugin.spark.authz.serde.ResolvedTableURIExtractor
 org.apache.kyuubi.plugin.spark.authz.serde.StringSeqURIExtractor
 org.apache.kyuubi.plugin.spark.authz.serde.StringURIExtractor
 org.apache.kyuubi.plugin.spark.authz.serde.SubqueryAliasURIExtractor
+org.apache.kyuubi.plugin.spark.authz.serde.TableIdentifierOptionURIExtractor
+org.apache.kyuubi.plugin.spark.authz.serde.TableIdentifierURIExtractor
 org.apache.kyuubi.plugin.spark.authz.serde.TableSpecURIExtractor

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

@@ -2226,7 +2226,19 @@
   } ],
   "opType" : "ALTERTABLE_COMPACT",
   "queryDescs" : [ ],
-  "uriDescs" : [ ]
+  "uriDescs" : [ {
+    "fieldName" : "child",
+    "fieldExtractor" : "ResolvedTableURIExtractor",
+    "isInput" : false
+  }, {
+    "fieldName" : "tableId",
+    "fieldExtractor" : "TableIdentifierOptionURIExtractor",
+    "isInput" : false
+  }, {
+    "fieldName" : "path",
+    "fieldExtractor" : "StringURIExtractor",
+    "isInput" : false
+  } ]
 }, {
   "classname" : "org.apache.spark.sql.delta.commands.UpdateCommand",
   "tableDescs" : [ {

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

@@ -17,8 +17,7 @@
 
 package org.apache.kyuubi.plugin.spark.authz.serde
 
-import java.util.{Map => JMap}
-import java.util.LinkedHashMap
+import java.util.{LinkedHashMap, Map => JMap}
 
 import scala.collection.JavaConverters._
 
@@ -81,14 +80,18 @@ object TableExtractor {
 class TableIdentifierTableExtractor extends TableExtractor {
   override def apply(spark: SparkSession, v1: AnyRef): Option[Table] = {
     val identifier = v1.asInstanceOf[TableIdentifier]
-    val owner =
-      try {
-        val catalogTable = spark.sessionState.catalog.getTableMetadata(identifier)
-        Option(catalogTable.owner).filter(_.nonEmpty)
-      } catch {
-        case _: Exception => None
-      }
-    Some(Table(None, identifier.database, identifier.table, owner))
+    if (isPathIdentifier(identifier.table, spark)) {
+      None
+    } else {
+      val owner =
+        try {
+          val catalogTable = spark.sessionState.catalog.getTableMetadata(identifier)
+          Option(catalogTable.owner).filter(_.nonEmpty)
+        } catch {
+          case _: Exception => None
+        }
+      Some(Table(None, identifier.database, identifier.table, owner))
+    }
   }
 }
 

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

@@ -18,6 +18,7 @@
 package org.apache.kyuubi.plugin.spark.authz.serde
 
 import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.TableIdentifier
 import org.apache.spark.sql.catalyst.catalog.{CatalogStorageFormat, CatalogTable}
 import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, SubqueryAlias}
 import org.apache.spark.sql.connector.catalog.Identifier
@@ -133,3 +134,26 @@ class DataSourceV2RelationURIExtractor extends URIExtractor {
     }
   }
 }
+
+class ResolvedTableURIExtractor extends URIExtractor {
+  override def apply(spark: SparkSession, v1: AnyRef): Seq[Uri] = {
+    val identifier = invokeAs[AnyRef](v1, "identifier")
+    lookupExtractor[IdentifierURIExtractor].apply(spark, identifier)
+  }
+}
+
+class TableIdentifierURIExtractor extends URIExtractor {
+  override def apply(spark: SparkSession, v1: AnyRef): Seq[Uri] = v1 match {
+    case tableIdentifier: TableIdentifier if isPathIdentifier(tableIdentifier.table, spark) =>
+      Seq(tableIdentifier.table).map(Uri)
+    case _ => Nil
+  }
+}
+
+class TableIdentifierOptionURIExtractor extends URIExtractor {
+  override def apply(spark: SparkSession, v1: AnyRef): Seq[Uri] = v1 match {
+    case Some(tableIdentifier: TableIdentifier) =>
+      lookupExtractor[TableIdentifierURIExtractor].apply(spark, tableIdentifier)
+    case _ => Nil
+  }
+}

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

@@ -49,7 +49,11 @@ object DeltaCommands extends CommandSpecs[TableCommandSpec] {
     val cmd = "org.apache.spark.sql.delta.commands.OptimizeTableCommand"
     val childDesc = TableDesc("child", classOf[ResolvedTableTableExtractor])
     val tableDesc = TableDesc("tableId", classOf[TableIdentifierOptionTableExtractor])
-    TableCommandSpec(cmd, Seq(childDesc, tableDesc), ALTERTABLE_COMPACT)
+    val uriDescs = Seq(
+      UriDesc("child", classOf[ResolvedTableURIExtractor]),
+      UriDesc("tableId", classOf[TableIdentifierOptionURIExtractor]),
+      UriDesc("path", classOf[StringURIExtractor]))
+    TableCommandSpec(cmd, Seq(childDesc, tableDesc), ALTERTABLE_COMPACT, uriDescs = uriDescs)
   }
 
   val VacuumTableCommand = {

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

@@ -432,6 +432,25 @@ class DeltaCatalogRangerSparkExtensionSuite extends RangerSparkExtensionSuite {
       }
     }
   }
+
+  test("optimize path-based table") {
+    assume(isSparkV32OrGreater)
+
+    withTempDir(path => {
+      doAs(admin, sql(createPathBasedTableSql(path)))
+      val optimizeTableSql1 = s"OPTIMIZE delta.`$path`"
+      interceptContains[AccessControlException](
+        doAs(someone, sql(optimizeTableSql1)))(
+        s"does not have [write] privilege on [[$path, $path/]]")
+      doAs(admin, sql(optimizeTableSql1))
+
+      val optimizeTableSql2 = s"OPTIMIZE '$path'"
+      interceptContains[AccessControlException](
+        doAs(someone, sql(optimizeTableSql2)))(
+        s"does not have [write] privilege on [[$path, $path/]]")
+      doAs(admin, sql(optimizeTableSql2))
+    })
+  }
 }
 
 object DeltaCatalogRangerSparkExtensionSuite {