[KYUUBI #6979] Support check paimon system producers

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

@@ -2589,6 +2589,22 @@
     "isInput" : false,
     "comment" : "Delta"
   } ]
+}, {
+  "classname" : "org.apache.paimon.spark.catalyst.plans.logical.PaimonCallCommand",
+  "tableDescs" : [ {
+    "fieldName" : "args",
+    "fieldExtractor" : "ExpressionSeqTableExtractor",
+    "columnDesc" : null,
+    "actionTypeDesc" : null,
+    "tableTypeDesc" : null,
+    "catalogDesc" : null,
+    "isInput" : false,
+    "setCurrentDatabaseIfMissing" : false,
+    "comment" : "Paimon"
+  } ],
+  "opType" : "ALTERTABLE_PROPERTIES",
+  "queryDescs" : [ ],
+  "uriDescs" : [ ]
 }, {
   "classname" : "org.apache.paimon.spark.commands.DeleteFromPaimonTableCommand",
   "tableDescs" : [ {

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

@@ -17,6 +17,7 @@
 
 package org.apache.kyuubi.plugin.spark.authz.gen
 
+import org.apache.kyuubi.plugin.spark.authz.OperationType
 import org.apache.kyuubi.plugin.spark.authz.PrivilegeObjectActionType._
 import org.apache.kyuubi.plugin.spark.authz.serde._
 
@@ -56,8 +57,14 @@ object PaimonCommands extends CommandSpecs[TableCommandSpec] {
     TableCommandSpec(cmd, Seq(targetTableDesc, sourceTableDesc))
   }
 
+  val PaimonCallCommand = {
+    val cmd = "org.apache.paimon.spark.catalyst.plans.logical.PaimonCallCommand"
+    val td = TableDesc("args", classOf[ExpressionSeqTableExtractor], comment = "Paimon")
+    TableCommandSpec(cmd, Seq(td), opType = OperationType.ALTERTABLE_PROPERTIES)
+  }
   override def specs: Seq[TableCommandSpec] = Seq(
     UpdatePaimonTable,
     DeleteFromPaimonTable,
-    MergeIntoPaimonTable)
+    MergeIntoPaimonTable,
+    PaimonCallCommand)
 }

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

@@ -568,6 +568,63 @@ class PaimonCatalogRangerSparkExtensionSuite extends RangerSparkExtensionSuite {
     }
   }
 
+  test("Producers") {
+
+    withCleanTmpResources(Seq(
+      (s"$catalogV2.$namespace1.$table1", "table"))) {
+      try {
+        doAs(admin, sql(createTableSql(namespace1, table1)))
+        doAs(admin, sql(s"INSERT INTO $catalogV2.$namespace1.$table1 VALUES (1, 'a'), (2, 'b')"))
+
+        var currentCatalogName =
+          doAs(admin, spark.sessionState.catalogManager.currentCatalog.name())
+
+        doAs(admin, sql(s"use $catalogV2"))
+        currentCatalogName = doAs(admin, spark.sessionState.catalogManager.currentCatalog.name())
+        assert(currentCatalogName.equals(catalogV2))
+
+        // Create Tag
+        val createTagSql = s"Call sys.create_tag(table =>" +
+          s"'$catalogV2.$namespace1.$table1', tag => 'test_tag', snapshot => 1)"
+        interceptEndsWith[AccessControlException] {
+          doAs(table1OnlyUserForNs, sql(createTagSql))
+        }(s"does not have [alter] privilege on [$namespace1/$table1]")
+        interceptEndsWith[AccessControlException] {
+          doAs(someone, sql(createTagSql))
+        }(s"does not have [alter] privilege on [$namespace1/$table1]")
+        doAs(admin, sql(createTagSql))
+
+        // Delete Tag
+        val deleteTagSql = s"Call sys.delete_tag(table =>" +
+          s"'$catalogV2.$namespace1.$table1', tag => 'test_tag')"
+        interceptEndsWith[AccessControlException] {
+          doAs(table1OnlyUserForNs, sql(deleteTagSql))
+        }(s"does not have [alter] privilege on [$namespace1/$table1]")
+        interceptEndsWith[AccessControlException] {
+          doAs(someone, sql(deleteTagSql))
+        }(s"does not have [alter] privilege on [$namespace1/$table1]")
+        doAs(admin, sql(deleteTagSql))
+
+        // Rollback
+        doAs(admin, sql(s"INSERT INTO $catalogV2.$namespace1.$table1 VALUES (3, 'a'), (4, 'b')"))
+        doAs(admin, sql(s"INSERT INTO $catalogV2.$namespace1.$table1 VALUES (5, 'a'), (6, 'b')"))
+        val rollbackTagSql = s"Call sys.rollback(table =>" +
+          s"'$catalogV2.$namespace1.$table1', version => '2')"
+        interceptEndsWith[AccessControlException] {
+          doAs(table1OnlyUserForNs, sql(rollbackTagSql))
+        }(s"does not have [alter] privilege on [$namespace1/$table1]")
+        interceptEndsWith[AccessControlException] {
+          doAs(someone, sql(rollbackTagSql))
+        }(s"does not have [alter] privilege on [$namespace1/$table1]")
+        doAs(admin, sql(rollbackTagSql))
+
+      } finally {
+        doAs(admin, sql(s"use spark_catalog"))
+      }
+    }
+
+  }
+
   def createTableSql(namespace: String, table: String): String =
     s"""
        |CREATE TABLE IF NOT EXISTS $catalogV2.$namespace.$table

pom.xml

@@ -183,7 +183,7 @@
         <netty.version>4.1.108.Final</netty.version>
         <openai.java.version>0.12.0</openai.java.version>
         <retrofit.version>2.9.0</retrofit.version>
-        <paimon.version>0.8.2</paimon.version>
+        <paimon.version>0.9.0</paimon.version>
         <paimon.artifact>paimon-spark-${spark.binary.version}</paimon.artifact>
         <phoenix.version>6.0.0</phoenix.version>
         <postgresql.version>42.7.2</postgresql.version>