You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kyuubi.apache.org by bo...@apache.org on 2026/08/04 18:30:42 UTC
(kyuubi) branch master updated: [KYUUBI #6718][AUTHZ] Support {OWNER} variable for Hudi update/delete/merge
This is an automated email from the ASF dual-hosted git repository.
bowenliang123 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/kyuubi.git
The following commit(s) were added to refs/heads/master by this push:
new 7a70a3cbb9 [KYUUBI #6718][AUTHZ] Support {OWNER} variable for Hudi update/delete/merge
7a70a3cbb9 is described below
commit 7a70a3cbb964c329766e090e18a297f4bbb20dc9
Author: lifumao <li...@tencent.com>
AuthorDate: Wed Aug 5 02:30:29 2026 +0800
[KYUUBI #6718][AUTHZ] Support {OWNER} variable for Hudi update/delete/merge
### Why are the changes needed?
Fix https://github.com/apache/kyuubi/issues/6718.
Hudi `UPDATE` / `DELETE` / `MERGE INTO` commands are handled by `HudiDataSourceV2RelationTableExtractor` and `HudiMergeIntoTargetTableExtractor` in `table_command_spec.json`. Previously, both extractors delegated the `SubqueryAlias(identifier, _)` case to `StringTableExtractor`, which only parses `database.table` from a string and always leaves `Table.owner` as `None`.
As a result, `AccessResource#setOwnerUser` is never called for these commands, so Ranger cannot resolve the `{OWNER}` variable, and a policy like `{OWNER}` -> `ALL on database/table/column` never matches — even when the user running the statement is exactly the table's creator/owner.
### How was this patch tested?
Added UT.
### Was this patch authored or co-authored using generative AI tooling?
Assisted-by: Claude:claude-opus-4-7
Closes #7606 from maomaodev/kyuubi_6718.
Closes #6718
6555756f5 [lifumao] [KYUUBI #6718][AUTHZ] Fix hudi table owner can not update/delete/merge tables they have created
Authored-by: lifumao <li...@tencent.com>
Signed-off-by: Bowen Liang <li...@qq.com>
---
.../plugin/spark/authz/serde/tableExtractors.scala | 23 +++++-
.../spark/authz/gen/PolicyJsonFileGenerator.scala | 8 +-
.../src/test/resources/sparkSql_hive_jenkins.json | 2 +-
.../HudiCatalogRangerSparkExtensionSuite.scala | 85 ++++++++++++++++++++++
4 files changed, 113 insertions(+), 5 deletions(-)
diff --git a/extensions/spark/kyuubi-spark-authz/src/main/scala/org/apache/kyuubi/plugin/spark/authz/serde/tableExtractors.scala b/extensions/spark/kyuubi-spark-authz/src/main/scala/org/apache/kyuubi/plugin/spark/authz/serde/tableExtractors.scala
index 9729857a73..3ef2172d67 100644
--- a/extensions/spark/kyuubi-spark-authz/src/main/scala/org/apache/kyuubi/plugin/spark/authz/serde/tableExtractors.scala
+++ b/extensions/spark/kyuubi-spark-authz/src/main/scala/org/apache/kyuubi/plugin/spark/authz/serde/tableExtractors.scala
@@ -72,6 +72,19 @@ object TableExtractor {
case _: Exception => None
}
}
+
+ /**
+ * Get owner from a `org.apache.spark.sql.execution.datasources.LogicalRelation`
+ * that wraps a Spark `CatalogTable`.
+ */
+ def getLogicalRelationOwner(v: AnyRef): Option[String] = {
+ try {
+ val maybeCatalogTable = invokeAs[Option[CatalogTable]](v, "catalogTable")
+ maybeCatalogTable.flatMap(ct => Option(ct.owner).filter(_.nonEmpty))
+ } catch {
+ case _: Exception => None
+ }
+ }
}
/**
@@ -321,11 +334,13 @@ class HudiDataSourceV2RelationTableExtractor extends TableExtractor {
override def apply(spark: SparkSession, v1: AnyRef): Option[Table] = {
invokeAs[LogicalPlan](v1, "table") match {
// Match multipartIdentifier with tableAlias
- case SubqueryAlias(_, SubqueryAlias(identifier, _)) =>
+ case SubqueryAlias(_, SubqueryAlias(identifier, relation)) =>
lookupExtractor[StringTableExtractor].apply(spark, identifier.toString())
+ .map(_.copy(owner = TableExtractor.getLogicalRelationOwner(relation)))
// Match multipartIdentifier without tableAlias
- case SubqueryAlias(identifier, _) =>
+ case SubqueryAlias(identifier, relation) =>
lookupExtractor[StringTableExtractor].apply(spark, identifier.toString())
+ .map(_.copy(owner = TableExtractor.getLogicalRelationOwner(relation)))
case _ => None
}
}
@@ -350,9 +365,11 @@ class HudiMergeIntoTargetTableExtractor extends TableExtractor {
// Match multipartIdentifier with tableAlias
case SubqueryAlias(_, SubqueryAlias(identifier, relation)) =>
lookupExtractor[StringTableExtractor].apply(spark, identifier.toString())
+ .map(_.copy(owner = TableExtractor.getLogicalRelationOwner(relation)))
// Match multipartIdentifier without tableAlias
- case SubqueryAlias(identifier, _) =>
+ case SubqueryAlias(identifier, relation) =>
lookupExtractor[StringTableExtractor].apply(spark, identifier.toString())
+ .map(_.copy(owner = TableExtractor.getLogicalRelationOwner(relation)))
case _ => None
}
}
diff --git a/extensions/spark/kyuubi-spark-authz/src/test/gen/scala/org/apache/kyuubi/plugin/spark/authz/gen/PolicyJsonFileGenerator.scala b/extensions/spark/kyuubi-spark-authz/src/test/gen/scala/org/apache/kyuubi/plugin/spark/authz/gen/PolicyJsonFileGenerator.scala
index a1f4b77e9e..3277de2bf2 100644
--- a/extensions/spark/kyuubi-spark-authz/src/test/gen/scala/org/apache/kyuubi/plugin/spark/authz/gen/PolicyJsonFileGenerator.scala
+++ b/extensions/spark/kyuubi-spark-authz/src/test/gen/scala/org/apache/kyuubi/plugin/spark/authz/gen/PolicyJsonFileGenerator.scala
@@ -175,7 +175,13 @@ class PolicyJsonFileGenerator extends KyuubiFunSuite {
name = "all - database, udf",
description = "Policy for all - database, udf",
resources = Map(
- databaseRes(defaultDb, sparkCatalog, icebergNamespace, namespace1, paimonNamespace),
+ databaseRes(
+ defaultDb,
+ sparkCatalog,
+ icebergNamespace,
+ namespace1,
+ paimonNamespace,
+ hudiNamespace),
allTableRes,
allColumnRes),
policyItems = List(
diff --git a/extensions/spark/kyuubi-spark-authz/src/test/resources/sparkSql_hive_jenkins.json b/extensions/spark/kyuubi-spark-authz/src/test/resources/sparkSql_hive_jenkins.json
index 6928feaf95..5d8dc7d28f 100644
--- a/extensions/spark/kyuubi-spark-authz/src/test/resources/sparkSql_hive_jenkins.json
+++ b/extensions/spark/kyuubi-spark-authz/src/test/resources/sparkSql_hive_jenkins.json
@@ -198,7 +198,7 @@
"isRecursive" : false
},
"database" : {
- "values" : [ "default", "spark_catalog", "iceberg_ns", "ns1", "paimon_ns" ],
+ "values" : [ "default", "spark_catalog", "iceberg_ns", "ns1", "paimon_ns", "hudi_ns" ],
"isExcludes" : false,
"isRecursive" : false
},
diff --git a/extensions/spark/kyuubi-spark-authz/src/test/scala/org/apache/kyuubi/plugin/spark/authz/ranger/HudiCatalogRangerSparkExtensionSuite.scala b/extensions/spark/kyuubi-spark-authz/src/test/scala/org/apache/kyuubi/plugin/spark/authz/ranger/HudiCatalogRangerSparkExtensionSuite.scala
index 52ec46e5e3..68d63c1d9b 100644
--- a/extensions/spark/kyuubi-spark-authz/src/test/scala/org/apache/kyuubi/plugin/spark/authz/ranger/HudiCatalogRangerSparkExtensionSuite.scala
+++ b/extensions/spark/kyuubi-spark-authz/src/test/scala/org/apache/kyuubi/plugin/spark/authz/ranger/HudiCatalogRangerSparkExtensionSuite.scala
@@ -17,6 +17,7 @@
package org.apache.kyuubi.plugin.spark.authz.ranger
import org.apache.spark.SparkConf
+import org.apache.spark.sql.catalyst.TableIdentifier
import org.apache.kyuubi.Utils
import org.apache.kyuubi.plugin.spark.authz.AccessControlException
@@ -494,6 +495,90 @@ class HudiCatalogRangerSparkExtensionSuite extends RangerSparkExtensionSuite {
}
}
+ test("Support {OWNER} variable for Hudi update/delete/merge") {
+ withSingleCallEnabled {
+ withCleanTmpResources(Seq(
+ (s"$namespace1.$table1", "table"),
+ (s"$namespace1.$table2", "table"),
+ (namespace1, "database"))) {
+ doAs(admin, sql(s"CREATE DATABASE IF NOT EXISTS $namespace1"))
+
+ // createOnlyUser creates and owns table1
+ doAs(
+ createOnlyUser,
+ 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))
+
+ // createOnlyUser creates and owns table2 as merge source
+ doAs(
+ createOnlyUser,
+ sql(
+ s"""
+ |CREATE TABLE IF NOT EXISTS $namespace1.$table2(id int, name string, city string)
+ |USING HUDI
+ |OPTIONS (
+ | type = 'cow',
+ | primaryKey = 'id',
+ | 'hoodie.datasource.hive_sync.enable' = 'false'
+ |)
+ |PARTITIONED BY(city)
+ |""".stripMargin))
+
+ // The in-memory catalog does not automatically populate `CatalogTable.owner`
+ // (unlike Hive Metastore). Manually set it to simulate the production case
+ // where the creator becomes the owner, which is required by Ranger {OWNER}.
+ Seq(table1, table2).foreach { t =>
+ val ident = TableIdentifier(t, Some(namespace1))
+ val ct = spark.sessionState.catalog.getTableMetadata(ident)
+ spark.sessionState.catalog.alterTable(ct.copy(owner = createOnlyUser))
+ }
+
+ val deleteSql = s"DELETE FROM $namespace1.$table1 WHERE id = 10"
+ // owner should be allowed
+ doAs(createOnlyUser, sql(deleteSql))
+ // non-owner should still be denied
+ interceptEndsWith[AccessControlException] {
+ doAs(someone, sql(deleteSql))
+ }(s"does not have [update] privilege on [$namespace1/$table1]")
+
+ val updateSql = s"UPDATE $namespace1.$table1 SET name = 'test' WHERE id > 10"
+ // owner should be allowed
+ doAs(createOnlyUser, sql(updateSql))
+ // non-owner should still be denied
+ interceptEndsWith[AccessControlException] {
+ doAs(someone, sql(updateSql))
+ }(s"does not have [update] privilege on [$namespace1/$table1]")
+
+ val mergeIntoSql =
+ s"""
+ |MERGE INTO $namespace1.$table1 target
+ |USING $namespace1.$table2 source
+ |ON target.id = source.id
+ |WHEN MATCHED
+ |AND target.name == 'test'
+ | THEN UPDATE SET id = source.id, name = source.name, city = source.city
+ |""".stripMargin
+ // owner (of both target and source) should be allowed
+ doAs(createOnlyUser, sql(mergeIntoSql))
+ // non-owner should still be denied
+ interceptEndsWith[AccessControlException] {
+ doAs(someone, sql(mergeIntoSql))
+ }(s"does not have [select] privilege on " +
+ s"[$namespace1/$table2/city,$namespace1/$table2/id,$namespace1/$table2/name], " +
+ s"[update] privilege on [$namespace1/$table1]")
+ }
+ }
+ }
+
test("CallProcedureHoodieCommand") {
withSingleCallEnabled {
withCleanTmpResources(Seq(