[KYUUBI #7136] Support more test cases about iceberg ranger check

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

@@ -728,6 +728,22 @@
     "isInput" : false,
     "comment" : ""
   } ]
+}, {
+  "classname" : "org.apache.spark.sql.catalyst.plans.logical.SetViewProperties",
+  "tableDescs" : [ {
+    "fieldName" : "child",
+    "fieldExtractor" : "ResolvedTableTableExtractor",
+    "columnDesc" : null,
+    "actionTypeDesc" : null,
+    "tableTypeDesc" : null,
+    "catalogDesc" : null,
+    "isInput" : true,
+    "setCurrentDatabaseIfMissing" : false,
+    "comment" : ""
+  } ],
+  "opType" : "ALTERTABLE_PROPERTIES",
+  "queryDescs" : [ ],
+  "uriDescs" : [ ]
 }, {
   "classname" : "org.apache.spark.sql.catalyst.plans.logical.ShowCreateTable",
   "tableDescs" : [ {
@@ -808,6 +824,22 @@
   "opType" : "ALTERTABLE_PROPERTIES",
   "queryDescs" : [ ],
   "uriDescs" : [ ]
+}, {
+  "classname" : "org.apache.spark.sql.catalyst.plans.logical.UnsetViewProperties",
+  "tableDescs" : [ {
+    "fieldName" : "child",
+    "fieldExtractor" : "ResolvedTableTableExtractor",
+    "columnDesc" : null,
+    "actionTypeDesc" : null,
+    "tableTypeDesc" : null,
+    "catalogDesc" : null,
+    "isInput" : true,
+    "setCurrentDatabaseIfMissing" : false,
+    "comment" : ""
+  } ],
+  "opType" : "ALTERTABLE_PROPERTIES",
+  "queryDescs" : [ ],
+  "uriDescs" : [ ]
 }, {
   "classname" : "org.apache.spark.sql.catalyst.plans.logical.UpdateTable",
   "tableDescs" : [ {
@@ -2077,6 +2109,42 @@
   "opType" : "QUERY",
   "queryDescs" : [ ],
   "uriDescs" : [ ]
+}, {
+  "classname" : "org.apache.spark.sql.catalyst.plans.logical.views.CreateIcebergView",
+  "tableDescs" : [ {
+    "fieldName" : "child",
+    "fieldExtractor" : "ResolvedIdentifierTableExtractor",
+    "columnDesc" : null,
+    "actionTypeDesc" : null,
+    "tableTypeDesc" : null,
+    "catalogDesc" : null,
+    "isInput" : false,
+    "setCurrentDatabaseIfMissing" : false,
+    "comment" : ""
+  } ],
+  "opType" : "CREATEVIEW",
+  "queryDescs" : [ {
+    "fieldName" : "query",
+    "fieldExtractor" : "LogicalPlanQueryExtractor",
+    "comment" : ""
+  } ],
+  "uriDescs" : [ ]
+}, {
+  "classname" : "org.apache.spark.sql.catalyst.plans.logical.views.DropIcebergView",
+  "tableDescs" : [ {
+    "fieldName" : "child",
+    "fieldExtractor" : "ResolvedIdentifierTableExtractor",
+    "columnDesc" : null,
+    "actionTypeDesc" : null,
+    "tableTypeDesc" : null,
+    "catalogDesc" : null,
+    "isInput" : false,
+    "setCurrentDatabaseIfMissing" : false,
+    "comment" : ""
+  } ],
+  "opType" : "DROPVIEW",
+  "queryDescs" : [ ],
+  "uriDescs" : [ ]
 }, {
   "classname" : "org.apache.spark.sql.hudi.command.AlterHoodieTableAddColumnsCommand",
   "tableDescs" : [ {

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

@@ -140,8 +140,13 @@ class ResolvedTableTableExtractor extends TableExtractor {
     val catalog = lookupExtractor[CatalogPluginCatalogExtractor].apply(catalogVal)
     val identifier = invokeAs[AnyRef](v1, "identifier")
     val maybeTable = lookupExtractor[IdentifierTableExtractor].apply(spark, identifier)
-    val maybeOwner = TableExtractor.getOwner(v1)
-    maybeTable.map(_.copy(catalog = catalog, owner = maybeOwner))
+    // Iceberg show create table $viewName use ResolvedV2View
+    if (!v1.getClass.getName
+        .equals("org.apache.spark.sql.catalyst.plans.logical.views.ResolvedV2View")) {
+      val maybeOwner = TableExtractor.getOwner(v1)
+      maybeTable.map(_.copy(catalog = catalog, owner = maybeOwner))
+    }
+    maybeTable
   }
 }
 

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

@@ -18,6 +18,7 @@
 package org.apache.kyuubi.plugin.spark.authz.gen
 
 import org.apache.kyuubi.plugin.spark.authz.OperationType
+import org.apache.kyuubi.plugin.spark.authz.OperationType.{CREATEVIEW, DROPVIEW}
 import org.apache.kyuubi.plugin.spark.authz.PrivilegeObjectActionType._
 import org.apache.kyuubi.plugin.spark.authz.serde._
 
@@ -113,6 +114,31 @@ object IcebergCommands extends CommandSpecs[TableCommandSpec] {
     AddPartitionFiled.copy(cmd)
   }
 
+  val CreateIcebergView = {
+    val cmd = "org.apache.spark.sql.catalyst.plans.logical.views.CreateIcebergView"
+    val tableDesc = TableDesc(
+      "child",
+      classOf[ResolvedIdentifierTableExtractor],
+      isInput = false)
+    val queryDesc = QueryDesc("query")
+    TableCommandSpec(
+      cmd,
+      Seq(tableDesc),
+      CREATEVIEW,
+      queryDescs = Seq(queryDesc))
+  }
+
+  val DropIcebergView = {
+    val cmd = "org.apache.spark.sql.catalyst.plans.logical.views.DropIcebergView"
+    val tableDesc = TableDesc(
+      "child",
+      classOf[ResolvedIdentifierTableExtractor])
+    TableCommandSpec(
+      cmd,
+      Seq(tableDesc),
+      DROPVIEW)
+  }
+
   override def specs: Seq[TableCommandSpec] = Seq(
     CallProcedure,
     DeleteFromIcebergTable,
@@ -128,6 +154,8 @@ object IcebergCommands extends CommandSpecs[TableCommandSpec] {
     CreateOrReplaceTag,
     DropBranch,
     DropTag,
+    CreateIcebergView,
+    DropIcebergView,
     MergeIntoIcebergTable.copy(classname =
       "org.apache.spark.sql.catalyst.plans.logical.UnresolvedMergeIntoIcebergTable"))
 }

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

@@ -674,6 +674,20 @@ object TableCommands extends CommandSpecs[TableCommandSpec] {
     TableCommandSpec(cmd, Nil, ADD, uriDescs = Seq(uriDesc))
   }
 
+  val UnsetViewProperties = {
+    val cmd = "org.apache.spark.sql.catalyst.plans.logical.UnsetViewProperties"
+    val tableDesc =
+      TableDesc("child", classOf[ResolvedTableTableExtractor], isInput = true)
+    TableCommandSpec(cmd, Seq(tableDesc), ALTERTABLE_PROPERTIES)
+  }
+
+  val SetViewProperties = {
+    val cmd = "org.apache.spark.sql.catalyst.plans.logical.SetViewProperties"
+    val tableDesc =
+      TableDesc("child", classOf[ResolvedTableTableExtractor], isInput = true)
+    TableCommandSpec(cmd, Seq(tableDesc), ALTERTABLE_PROPERTIES)
+  }
+
   override def specs: Seq[TableCommandSpec] = Seq(
     AddArchivesCommand,
     AddArchivesCommand.copy(classname = "org.apache.spark.sql.execution.command.AddFilesCommand"),
@@ -764,5 +778,7 @@ object TableCommands extends CommandSpecs[TableCommandSpec] {
     ShowTablePropertiesV2,
     TruncateTable,
     TruncateTableV2,
+    SetViewProperties,
+    UnsetViewProperties,
     UpdateTable)
 }