This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 8de456de10 [core][spark] Add REST management procedures (#9410)
8de456de10 is described below
commit 8de456de1017b3784406f51ab72f5033e09625d8
Author: Jingsong Lee <[email protected]>
AuthorDate: Wed Aug 26 22:31:50 2026 +0800
[core][spark] Add REST management procedures (#9410)
---
docs/docs/concepts/rest/management-api.md | 326 +++++++++
.../PaimonSqlExtensions.g4 | 8 +-
.../org/apache/paimon/spark/SparkProcedures.java | 12 +
.../spark/procedure/BasePermissionProcedure.java | 130 ++++
.../spark/procedure/BasePolicyProcedure.java | 62 ++
.../spark/procedure/CreatePolicyProcedure.java | 100 +++
.../spark/procedure/DropPolicyProcedure.java | 94 +++
.../spark/procedure/GrantPermissionProcedure.java | 139 ++++
.../spark/procedure/ListPermissionsProcedure.java | 160 +++++
.../spark/procedure/ListPoliciesProcedure.java | 140 ++++
.../spark/procedure/RevokePermissionProcedure.java | 96 +++
.../AbstractPaimonSparkSqlExtensionsParser.scala | 1 +
.../spark/PaimonSparkTestWithRestCatalogBase.scala | 16 +-
.../spark/procedure/PermissionProcedureTest.scala | 787 +++++++++++++++++++++
.../sql/CatalogQualifiedCreateTableLikeTest.scala | 5 +
15 files changed, 2073 insertions(+), 3 deletions(-)
diff --git a/docs/docs/concepts/rest/management-api.md
b/docs/docs/concepts/rest/management-api.md
index 751eb6f3af..e90e9a71d4 100644
--- a/docs/docs/concepts/rest/management-api.md
+++ b/docs/docs/concepts/rest/management-api.md
@@ -231,3 +231,329 @@ schema-incompatible predicate or transform must also fail
closed rather than omi
This experimental contract deliberately does not define governed tags,
catalog/database policy
inheritance, or tag-driven matching. Those features need explicit match
conditions and conflict
rules before being added.
+
+## Spark SQL procedures
+
+The following examples assume a Spark catalog named `paimon`. Replace it with
the catalog name in
+`spark.sql.catalog.<catalog-name>`.
+
+### Grant permissions
+
+`grant_permission` returns one row with `result = true` when the server
accepts the assignment.
+
+Grant permission to create databases in the catalog:
+
+```sql
+CALL paimon.sys.grant_permission(
+ resource_type => 'CATALOG',
+ access => 'CREATEDATABASE',
+ principal => 'role:catalog_user'
+);
+```
+
+Grant read access to every applicable object currently or subsequently created
in the catalog. This
+does not grant catalog-level operations such as `CREATEDATABASE`:
+
+```sql
+CALL paimon.sys.grant_permission(
+ resource_type => 'CATALOG_ALL',
+ access => 'SELECT',
+ principal => 'role:catalog_reader'
+);
+```
+
+Grant permission to create views in a database with an optional expiration
time:
+
+```sql
+CALL paimon.sys.grant_permission(
+ resource_type => 'DATABASE',
+ database => 'sales',
+ access => 'CREATEVIEW',
+ principal => 'role:data_engineer',
+ expire_time => '2027-01-01T00:00:00Z'
+);
+```
+
+Grant write access to every applicable table currently or subsequently created
in one database.
+`DATABASE_ALL` requires `database` but does not accept a table, function, or
view locator:
+
+```sql
+CALL paimon.sys.grant_permission(
+ resource_type => 'DATABASE_ALL',
+ database => 'sales',
+ access => 'UPDATE',
+ principal => 'role:sales_writer'
+);
+```
+
+Grant table, function, and view access with the matching locator:
+
+```sql
+CALL paimon.sys.grant_permission(
+ resource_type => 'TABLE',
+ database => 'sales',
+ table => 'orders',
+ access => 'SELECT',
+ principal => 'user:alice'
+);
+
+CALL paimon.sys.grant_permission(
+ resource_type => 'FUNCTION',
+ database => 'sales',
+ function => 'calculate_tax',
+ access => 'SELECT',
+ principal => 'role:analyst'
+);
+
+CALL paimon.sys.grant_permission(
+ resource_type => 'VIEW',
+ database => 'sales',
+ view => 'daily_orders',
+ access => 'SELECT',
+ principal => 'service:reporting_job'
+);
+```
+
+Grant access to selected columns. This requires table query authorization;
named arguments are
+recommended because the two column range modes are mutually exclusive:
+
+```sql
+ALTER TABLE paimon.sales.orders
+SET TBLPROPERTIES ('query-auth.enabled' = 'true');
+
+CALL paimon.sys.grant_permission(
+ resource_type => 'COLUMN',
+ database => 'sales',
+ table => 'orders',
+ access => 'SELECT',
+ principal => 'role:analyst',
+ column_names => array('order_id', 'region')
+);
+```
+
+Use `excluded_column_names` for a denylist. Repeating the grant replaces the
preceding allowlist in
+one operation:
+
+```sql
+CALL paimon.sys.grant_permission(
+ resource_type => 'COLUMN',
+ database => 'sales',
+ table => 'orders',
+ access => 'SELECT',
+ principal => 'role:analyst',
+ excluded_column_names => array('email', 'phone_number')
+);
+```
+
+### List permissions
+
+`list_permissions` always addresses one exact resource or explicit descendant
scope. Omit optional
+filters to list every direct assignment on it; effective assignments inherited
from a scope are not
+synthesized:
+
+```sql
+CALL paimon.sys.list_permissions(
+ resource_type => 'TABLE',
+ database => 'sales',
+ table => 'orders'
+);
+```
+
+Filter by principal or access:
+
+```sql
+CALL paimon.sys.list_permissions(
+ resource_type => 'TABLE',
+ database => 'sales',
+ table => 'orders',
+ principal => 'role:analyst',
+ access => 'SELECT'
+);
+```
+
+List the column range attached to a principal. The result exposes
`column_names` and
+`excluded_column_names` as `ARRAY<STRING>` columns, with exactly one populated
for a `COLUMN`
+assignment:
+
+```sql
+CALL paimon.sys.list_permissions(
+ resource_type => 'COLUMN',
+ database => 'sales',
+ table => 'orders',
+ principal => 'role:analyst',
+ access => 'SELECT'
+);
+```
+
+The `next_page_token` output is opaque; pass it back unchanged with the same
filters:
+
+```sql
+CALL paimon.sys.list_permissions(
+ resource_type => 'TABLE',
+ database => 'sales',
+ table => 'orders',
+ max_results => 50,
+ page_token => 'opaque-token-from-previous-row'
+);
+```
+
+### Revoke permissions
+
+Supply the same three identity fields used by the grant. `expire_time` is not
part of identity.
+
+```sql
+CALL paimon.sys.revoke_permission(
+ resource_type => 'TABLE',
+ database => 'sales',
+ table => 'orders',
+ access => 'SELECT',
+ principal => 'role:sales_reader'
+);
+```
+
+Repeating the same call succeeds even when the assignment is already absent.
+
+Column revocation uses the containing table identity and removes the complete
range:
+
+```sql
+CALL paimon.sys.revoke_permission(
+ resource_type => 'COLUMN',
+ database => 'sales',
+ table => 'orders',
+ access => 'SELECT',
+ principal => 'role:analyst'
+);
+```
+
+### Create row-filter policies
+
+Before attaching any policy, enable table query authorization:
+
+```sql
+ALTER TABLE paimon.sales.orders
+SET TBLPROPERTIES ('query-auth.enabled' = 'true');
+```
+
+`create_policy` accepts the canonical `principal` and a serialized Paimon
`Predicate`. The JSON
+below is the same representation accepted in one
`AuthTableQueryResponse.filter` entry. Named
+arguments are recommended because row-filter and column-mask definitions use
different fields:
+
+```sql
+CALL paimon.sys.create_policy(
+ database => 'sales',
+ table => 'orders',
+ policy_type => 'ROW_FILTER',
+ principal => 'group:analysts',
+ predicate_json =>
'{"kind":"LEAF","transform":{"name":"FIELD_REF","fieldRef":{"index":1,"name":"region","type":"STRING"}},"function":"EQUAL","literals":["APAC"]}'
+);
+```
+
+The call fails if that principal already has a row filter on the table. Drop
the existing policy
+before creating a different definition for the same identity. Create another
policy for a second
+principal with a separate call.
+
+### Create column-masking policies
+
+For column masking, `on_column` identifies the protected column and
`transform_json` is the same
+serialized Paimon `Transform` representation used as an
+`AuthTableQueryResponse.columnMasking` value. This example replaces every
visible phone number
+with a fixed string:
+
+```sql
+CALL paimon.sys.create_policy(
+ database => 'sales',
+ table => 'customers',
+ policy_type => 'COLUMN_MASKING',
+ principal => 'role:support',
+ on_column => 'phone_number',
+ transform_json => '{"name":"CONCAT","inputs":["****"]}'
+);
+```
+
+A transform may reference table fields by name. The server remaps their
indices to the current
+schema, rejects missing fields, and verifies that the result type matches
`on_column`:
+
+```sql
+CALL paimon.sys.create_policy(
+ database => 'sales',
+ table => 'customers',
+ policy_type => 'COLUMN_MASKING',
+ principal => 'group:support',
+ on_column => 'email',
+ transform_json =>
'{"name":"CONCAT","inputs":[{"index":1,"name":"region","type":"STRING"},"-masked"]}'
+);
+```
+
+`predicate_json` is required only for `ROW_FILTER`. `on_column` and
`transform_json` are required
+only for `COLUMN_MASKING`. JSON containing a single quote must escape it as
`''` inside the SQL
+string literal.
+
+### List policies
+
+List every policy directly attached to one table:
+
+```sql
+CALL paimon.sys.list_policies(
+ database => 'sales',
+ table => 'orders'
+);
+```
+
+Filter by policy type or principal. A `column` filter is valid only with
+`policy_type => 'COLUMN_MASKING'`:
+
+```sql
+CALL paimon.sys.list_policies(
+ database => 'sales',
+ table => 'orders',
+ policy_type => 'ROW_FILTER',
+ principal => 'group:analysts'
+);
+```
+
+The output columns are `database`, `table`, `policy_type`, `principal`,
`predicate_json`,
+`on_column`, `transform_json`, and `next_page_token`. A row filter has only
`predicate_json`; a
+column mask has only `on_column` and `transform_json`. Pass an opaque
continuation token back
+unchanged with the same filters:
+
+```sql
+CALL paimon.sys.list_policies(
+ database => 'sales',
+ table => 'orders',
+ max_results => 50,
+ page_token => 'opaque-token-from-previous-row'
+);
+```
+
+Management listing follows the existing Paimon pagination contract: an empty
page terminates
+pagination and therefore has no continuation token. Each Spark procedure
returns exactly the page
+selected by `page_token`; pass a non-null `next_page_token` back unchanged to
retrieve the next page.
+
+### Drop policies
+
+Drop an existing policy:
+
+```sql
+CALL paimon.sys.drop_policy(
+ database => 'sales',
+ table => 'orders',
+ policy_type => 'ROW_FILTER',
+ principal => 'group:analysts'
+);
+```
+
+By default an absent policy is an error. Set `if_exists => true` for an
idempotent operation:
+
+```sql
+CALL paimon.sys.drop_policy(
+ database => 'sales',
+ table => 'orders',
+ policy_type => 'ROW_FILTER',
+ principal => 'group:analysts',
+ if_exists => true
+);
+```
+
+Creating, dropping, or inspecting permissions and policies requires the server
to
+authorize the caller for `GRANT` on the relevant resource. Authentication,
principal
+membership, policy persistence, schema validation, and audit logging remain
REST server concerns.
diff --git
a/paimon-spark/paimon-spark-common/src/main/antlr4/org.apache.spark.sql.catalyst.parser.extensions/PaimonSqlExtensions.g4
b/paimon-spark/paimon-spark-common/src/main/antlr4/org.apache.spark.sql.catalyst.parser.extensions/PaimonSqlExtensions.g4
index 8c2e45b34e..bbd2728d9d 100644
---
a/paimon-spark/paimon-spark-common/src/main/antlr4/org.apache.spark.sql.catalyst.parser.extensions/PaimonSqlExtensions.g4
+++
b/paimon-spark/paimon-spark-common/src/main/antlr4/org.apache.spark.sql.catalyst.parser.extensions/PaimonSqlExtensions.g4
@@ -167,6 +167,7 @@ overwriteClause
expression
: constant
| stringMap
+ | stringArray
;
constant
@@ -180,6 +181,10 @@ stringMap
: MAP '(' constant (',' constant)* ')'
;
+stringArray
+ : ARRAY '(' (constant (',' constant)*)? ')'
+ ;
+
booleanValue
: TRUE | FALSE
;
@@ -214,7 +219,7 @@ nonReserved
: ALTER | AS | CALL | CREATE | DAYS | DELETE | EXISTS | HOURS | IF | LIKE
| NOT | OF | OR | TABLE | REPLACE | RETAIN | VERSION | TAG
| TRUE | FALSE
- | MAP
+ | ARRAY | MAP
| COPY | INTO | FROM | FILE_FORMAT | PATTERN | FORCE | ON_ERROR |
ABORT_STATEMENT | CONTINUE | SKIP_FILE | OVERWRITE
| CSV
| JSON
@@ -249,6 +254,7 @@ TRUE: 'TRUE';
FALSE: 'FALSE';
MAP: 'MAP';
+ARRAY: 'ARRAY';
COPY: 'COPY';
INTO: 'INTO';
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkProcedures.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkProcedures.java
index 60b5747e3d..866a54e802 100644
---
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkProcedures.java
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkProcedures.java
@@ -29,16 +29,21 @@ import org.apache.paimon.spark.procedure.CopyFilesProcedure;
import org.apache.paimon.spark.procedure.CreateBranchProcedure;
import org.apache.paimon.spark.procedure.CreateFunctionProcedure;
import org.apache.paimon.spark.procedure.CreateGlobalIndexProcedure;
+import org.apache.paimon.spark.procedure.CreatePolicyProcedure;
import org.apache.paimon.spark.procedure.CreateTagFromTimestampProcedure;
import org.apache.paimon.spark.procedure.CreateTagProcedure;
import org.apache.paimon.spark.procedure.DeleteBranchProcedure;
import org.apache.paimon.spark.procedure.DeleteTagProcedure;
import org.apache.paimon.spark.procedure.DropFunctionProcedure;
import org.apache.paimon.spark.procedure.DropGlobalIndexProcedure;
+import org.apache.paimon.spark.procedure.DropPolicyProcedure;
import org.apache.paimon.spark.procedure.ExpirePartitionsProcedure;
import org.apache.paimon.spark.procedure.ExpireSnapshotsProcedure;
import org.apache.paimon.spark.procedure.ExpireTagsProcedure;
import org.apache.paimon.spark.procedure.FastForwardProcedure;
+import org.apache.paimon.spark.procedure.GrantPermissionProcedure;
+import org.apache.paimon.spark.procedure.ListPermissionsProcedure;
+import org.apache.paimon.spark.procedure.ListPoliciesProcedure;
import org.apache.paimon.spark.procedure.MarkPartitionDoneProcedure;
import org.apache.paimon.spark.procedure.MaterializeDeletionVectorsProcedure;
import org.apache.paimon.spark.procedure.MergeBranchProcedure;
@@ -57,6 +62,7 @@ import org.apache.paimon.spark.procedure.RepairProcedure;
import org.apache.paimon.spark.procedure.ReplaceTagProcedure;
import org.apache.paimon.spark.procedure.RescaleProcedure;
import org.apache.paimon.spark.procedure.ResetConsumerProcedure;
+import org.apache.paimon.spark.procedure.RevokePermissionProcedure;
import org.apache.paimon.spark.procedure.RewriteFileIndexProcedure;
import org.apache.paimon.spark.procedure.RollbackProcedure;
import org.apache.paimon.spark.procedure.RollbackToTimestampProcedure;
@@ -134,6 +140,12 @@ public class SparkProcedures {
procedureBuilders.put("rewrite_file_index",
RewriteFileIndexProcedure::builder);
procedureBuilders.put("copy", CopyFilesProcedure::builder);
procedureBuilders.put("reassign_row_id",
ReassignRowIdProcedure::builder);
+ procedureBuilders.put("grant_permission",
GrantPermissionProcedure::builder);
+ procedureBuilders.put("revoke_permission",
RevokePermissionProcedure::builder);
+ procedureBuilders.put("list_permissions",
ListPermissionsProcedure::builder);
+ procedureBuilders.put("create_policy", CreatePolicyProcedure::builder);
+ procedureBuilders.put("drop_policy", DropPolicyProcedure::builder);
+ procedureBuilders.put("list_policies", ListPoliciesProcedure::builder);
return procedureBuilders.build();
}
}
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/BasePermissionProcedure.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/BasePermissionProcedure.java
new file mode 100644
index 0000000000..8a4ba0b069
--- /dev/null
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/BasePermissionProcedure.java
@@ -0,0 +1,130 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.spark.procedure;
+
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.DelegateCatalog;
+import org.apache.paimon.management.PermissionAssignment;
+import org.apache.paimon.management.PermissionColumns;
+import org.apache.paimon.management.PermissionManagement;
+import org.apache.paimon.management.PermissionResource;
+import org.apache.paimon.management.PolicyManagement;
+import org.apache.paimon.management.ResourceType;
+import org.apache.paimon.rest.RESTCatalog;
+import org.apache.paimon.spark.catalog.WithPaimonCatalog;
+
+import org.apache.spark.sql.connector.catalog.TableCatalog;
+
+import javax.annotation.Nullable;
+
+import java.util.Arrays;
+import java.util.Locale;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/** Shared REST catalog lookup and argument validation for management
procedures. */
+abstract class BasePermissionProcedure extends BaseProcedure {
+
+ protected BasePermissionProcedure(TableCatalog tableCatalog) {
+ super(tableCatalog);
+ }
+
+ protected PermissionManagement permissionManagement() {
+ return restCatalog().permissionManagement();
+ }
+
+ protected PolicyManagement policyManagement() {
+ return restCatalog().policyManagement();
+ }
+
+ private RESTCatalog restCatalog() {
+ checkArgument(
+ tableCatalog() instanceof WithPaimonCatalog,
+ "Catalog '%s' is not a Paimon catalog.",
+ tableCatalog().name());
+ Catalog root =
+ DelegateCatalog.rootCatalog(((WithPaimonCatalog)
tableCatalog()).paimonCatalog());
+ checkArgument(
+ root instanceof RESTCatalog,
+ "Catalog '%s' does not support permission or policy
management.",
+ tableCatalog().name());
+ return (RESTCatalog) root;
+ }
+
+ protected static PermissionAssignment assignment(
+ ResourceType resourceType,
+ String access,
+ String principal,
+ @Nullable String database,
+ @Nullable String table,
+ @Nullable String function,
+ @Nullable String view,
+ @Nullable PermissionColumns columns,
+ @Nullable String expireTime) {
+ return new PermissionAssignment(
+ resource(resourceType, database, table, function, view),
+ access,
+ principal,
+ columns,
+ emptyToNull(expireTime));
+ }
+
+ protected static PermissionResource resource(
+ ResourceType resourceType,
+ @Nullable String database,
+ @Nullable String table,
+ @Nullable String function,
+ @Nullable String view) {
+ return new PermissionResource(
+ resourceType,
+ emptyToNull(database),
+ emptyToNull(table),
+ emptyToNull(function),
+ emptyToNull(view));
+ }
+
+ protected static <E extends Enum<E>> E enumValue(
+ String value, Class<E> enumClass, String argument) {
+ checkArgument(!isBlank(value), "%s cannot be empty.", argument);
+ try {
+ return Enum.valueOf(enumClass, value.toUpperCase(Locale.ROOT));
+ } catch (IllegalArgumentException e) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Invalid %s '%s'. Expected one of %s.",
+ argument, value,
Arrays.toString(enumClass.getEnumConstants())),
+ e);
+ }
+ }
+
+ @Nullable
+ protected static <E extends Enum<E>> E optionalEnum(
+ @Nullable String value, Class<E> enumClass, String argument) {
+ return isBlank(value) ? null : enumValue(value, enumClass, argument);
+ }
+
+ @Nullable
+ protected static String emptyToNull(@Nullable String value) {
+ return isBlank(value) ? null : value;
+ }
+
+ protected static boolean isBlank(@Nullable String value) {
+ return value == null || value.trim().isEmpty();
+ }
+}
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/BasePolicyProcedure.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/BasePolicyProcedure.java
new file mode 100644
index 0000000000..e96d31df77
--- /dev/null
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/BasePolicyProcedure.java
@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.spark.procedure;
+
+import org.apache.paimon.management.ColumnMask;
+import org.apache.paimon.management.DataPolicy;
+import org.apache.paimon.management.PermissionResource;
+import org.apache.paimon.management.PolicyType;
+import org.apache.paimon.management.ResourceType;
+import org.apache.paimon.management.RowFilter;
+
+import org.apache.spark.sql.connector.catalog.TableCatalog;
+
+import javax.annotation.Nullable;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/** Shared typed argument conversion for table policy procedures. */
+abstract class BasePolicyProcedure extends BasePermissionProcedure {
+
+ protected BasePolicyProcedure(TableCatalog tableCatalog) {
+ super(tableCatalog);
+ }
+
+ protected static DataPolicy policy(
+ String database,
+ String table,
+ PolicyType policyType,
+ String principal,
+ @Nullable String predicate,
+ @Nullable String onColumn,
+ @Nullable String transform) {
+ PermissionResource resource = tableResource(database, table);
+ if (policyType == PolicyType.ROW_FILTER) {
+ checkArgument(isBlank(onColumn), "ROW_FILTER policy cannot specify
on_column.");
+ checkArgument(isBlank(transform), "ROW_FILTER policy cannot
specify transform.");
+ return DataPolicy.rowFilter(resource, new RowFilter(predicate),
principal);
+ }
+ checkArgument(isBlank(predicate), "COLUMN_MASKING policy cannot
specify predicate.");
+ return DataPolicy.columnMask(resource, new ColumnMask(onColumn,
transform), principal);
+ }
+
+ protected static PermissionResource tableResource(String database, String
table) {
+ return resource(ResourceType.TABLE, database, table, null, null);
+ }
+}
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CreatePolicyProcedure.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CreatePolicyProcedure.java
new file mode 100644
index 0000000000..52b56a55fa
--- /dev/null
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CreatePolicyProcedure.java
@@ -0,0 +1,100 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.spark.procedure;
+
+import org.apache.paimon.management.DataPolicy;
+import
org.apache.paimon.management.PolicyManagement.PolicyAlreadyExistException;
+import org.apache.paimon.management.PolicyType;
+
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.connector.catalog.TableCatalog;
+import org.apache.spark.sql.types.DataTypes;
+import org.apache.spark.sql.types.Metadata;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+
+import static org.apache.spark.sql.types.DataTypes.StringType;
+
+/** Creates a table row-filter or column-mask policy. */
+public class CreatePolicyProcedure extends BasePolicyProcedure {
+
+ private static final ProcedureParameter[] PARAMETERS =
+ new ProcedureParameter[] {
+ ProcedureParameter.required("database", StringType),
+ ProcedureParameter.required("table", StringType),
+ ProcedureParameter.required("policy_type", StringType),
+ ProcedureParameter.required("principal", StringType),
+ ProcedureParameter.optional("predicate_json", StringType),
+ ProcedureParameter.optional("on_column", StringType),
+ ProcedureParameter.optional("transform_json", StringType)
+ };
+
+ private static final StructType OUTPUT_TYPE =
+ new StructType(
+ new StructField[] {
+ new StructField("result", DataTypes.BooleanType,
false, Metadata.empty())
+ });
+
+ private CreatePolicyProcedure(TableCatalog tableCatalog) {
+ super(tableCatalog);
+ }
+
+ @Override
+ public ProcedureParameter[] parameters() {
+ return PARAMETERS;
+ }
+
+ @Override
+ public StructType outputType() {
+ return OUTPUT_TYPE;
+ }
+
+ @Override
+ public InternalRow[] call(InternalRow args) {
+ DataPolicy policy =
+ policy(
+ args.getString(0),
+ args.getString(1),
+ enumValue(args.getString(2), PolicyType.class,
PARAMETERS[2].name()),
+ args.getString(3),
+ args.isNullAt(4) ? null : args.getString(4),
+ args.isNullAt(5) ? null : args.getString(5),
+ args.isNullAt(6) ? null : args.getString(6));
+ try {
+ policyManagement().createPolicy(policy);
+ } catch (PolicyAlreadyExistException e) {
+ throw new RuntimeException(e);
+ }
+ return new InternalRow[] {newInternalRow(true)};
+ }
+
+ public static ProcedureBuilder builder() {
+ return new Builder<CreatePolicyProcedure>() {
+ @Override
+ protected CreatePolicyProcedure doBuild() {
+ return new CreatePolicyProcedure(tableCatalog());
+ }
+ };
+ }
+
+ @Override
+ public String description() {
+ return "CreatePolicyProcedure";
+ }
+}
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/DropPolicyProcedure.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/DropPolicyProcedure.java
new file mode 100644
index 0000000000..71780d1a84
--- /dev/null
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/DropPolicyProcedure.java
@@ -0,0 +1,94 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.spark.procedure;
+
+import org.apache.paimon.management.PermissionResource;
+import org.apache.paimon.management.PolicyType;
+
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.connector.catalog.TableCatalog;
+import org.apache.spark.sql.types.DataTypes;
+import org.apache.spark.sql.types.Metadata;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+
+import static org.apache.spark.sql.types.DataTypes.BooleanType;
+import static org.apache.spark.sql.types.DataTypes.StringType;
+
+/** Drops one principal's row-filter or column-masking policy. */
+public class DropPolicyProcedure extends BasePolicyProcedure {
+
+ private static final ProcedureParameter[] PARAMETERS =
+ new ProcedureParameter[] {
+ ProcedureParameter.required("database", StringType),
+ ProcedureParameter.required("table", StringType),
+ ProcedureParameter.required("policy_type", StringType),
+ ProcedureParameter.required("principal", StringType),
+ ProcedureParameter.optional("column", StringType),
+ ProcedureParameter.optional("if_exists", BooleanType)
+ };
+
+ private static final StructType OUTPUT_TYPE =
+ new StructType(
+ new StructField[] {
+ new StructField("result", DataTypes.BooleanType,
false, Metadata.empty())
+ });
+
+ private DropPolicyProcedure(TableCatalog tableCatalog) {
+ super(tableCatalog);
+ }
+
+ @Override
+ public ProcedureParameter[] parameters() {
+ return PARAMETERS;
+ }
+
+ @Override
+ public StructType outputType() {
+ return OUTPUT_TYPE;
+ }
+
+ @Override
+ public InternalRow[] call(InternalRow args) {
+ PermissionResource resource = tableResource(args.getString(0),
args.getString(1));
+ PolicyType type = enumValue(args.getString(2), PolicyType.class,
PARAMETERS[2].name());
+ policyManagement()
+ .dropPolicy(
+ resource,
+ type,
+ args.getString(3),
+ args.isNullAt(4) ? null : args.getString(4),
+ !args.isNullAt(5) && args.getBoolean(5));
+ return new InternalRow[] {newInternalRow(true)};
+ }
+
+ public static ProcedureBuilder builder() {
+ return new Builder<DropPolicyProcedure>() {
+ @Override
+ protected DropPolicyProcedure doBuild() {
+ return new DropPolicyProcedure(tableCatalog());
+ }
+ };
+ }
+
+ @Override
+ public String description() {
+ return "DropPolicyProcedure";
+ }
+}
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/GrantPermissionProcedure.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/GrantPermissionProcedure.java
new file mode 100644
index 0000000000..14cbef3b7c
--- /dev/null
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/GrantPermissionProcedure.java
@@ -0,0 +1,139 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.spark.procedure;
+
+import org.apache.paimon.management.PermissionAssignment;
+import org.apache.paimon.management.PermissionColumns;
+import org.apache.paimon.management.ResourceType;
+
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.catalyst.util.ArrayData;
+import org.apache.spark.sql.connector.catalog.TableCatalog;
+import org.apache.spark.sql.types.Metadata;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+import org.apache.spark.unsafe.types.UTF8String;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.apache.spark.sql.types.DataTypes.StringType;
+import static org.apache.spark.sql.types.DataTypes.createArrayType;
+
+/** Grants a permission through a REST catalog. */
+public class GrantPermissionProcedure extends BasePermissionProcedure {
+
+ private static final ProcedureParameter[] PARAMETERS =
+ new ProcedureParameter[] {
+ ProcedureParameter.required("resource_type", StringType),
+ ProcedureParameter.required("access", StringType),
+ ProcedureParameter.required("principal", StringType),
+ ProcedureParameter.optional("database", StringType),
+ ProcedureParameter.optional("table", StringType),
+ ProcedureParameter.optional("function", StringType),
+ ProcedureParameter.optional("view", StringType),
+ ProcedureParameter.optional("expire_time", StringType),
+ ProcedureParameter.optional("column_names",
createArrayType(StringType)),
+ ProcedureParameter.optional("excluded_column_names",
createArrayType(StringType))
+ };
+
+ private static final StructType OUTPUT_TYPE =
+ new StructType(
+ new StructField[] {
+ new StructField(
+ "result",
+
org.apache.spark.sql.types.DataTypes.BooleanType,
+ false,
+ Metadata.empty())
+ });
+
+ private GrantPermissionProcedure(TableCatalog tableCatalog) {
+ super(tableCatalog);
+ }
+
+ @Override
+ public ProcedureParameter[] parameters() {
+ return PARAMETERS;
+ }
+
+ @Override
+ public StructType outputType() {
+ return OUTPUT_TYPE;
+ }
+
+ @Override
+ public InternalRow[] call(InternalRow args) {
+ ResourceType resourceType =
+ enumValue(args.getString(0), ResourceType.class,
PARAMETERS[0].name());
+ PermissionAssignment assignment =
+ assignment(
+ resourceType,
+ args.getString(1),
+ args.getString(2),
+ args.isNullAt(3) ? null : args.getString(3),
+ args.isNullAt(4) ? null : args.getString(4),
+ args.isNullAt(5) ? null : args.getString(5),
+ args.isNullAt(6) ? null : args.getString(6),
+ columns(args, 8, 9),
+ args.isNullAt(7) ? null : args.getString(7));
+
+ permissionManagement().grantPermission(assignment);
+ return new InternalRow[] {newInternalRow(true)};
+ }
+
+ @Nullable
+ private static PermissionColumns columns(
+ InternalRow args, int columnNamesPos, int excludedColumnNamesPos) {
+ List<String> columnNames = stringArray(args, columnNamesPos);
+ List<String> excludedColumnNames = stringArray(args,
excludedColumnNamesPos);
+ return columnNames == null && excludedColumnNames == null
+ ? null
+ : new PermissionColumns(columnNames, excludedColumnNames);
+ }
+
+ @Nullable
+ private static List<String> stringArray(InternalRow args, int position) {
+ if (args.isNullAt(position)) {
+ return null;
+ }
+ ArrayData array = args.getArray(position);
+ List<String> values = new ArrayList<>(array.numElements());
+ for (int i = 0; i < array.numElements(); i++) {
+ UTF8String value = array.isNullAt(i) ? null :
array.getUTF8String(i);
+ values.add(value == null ? null : value.toString());
+ }
+ return values;
+ }
+
+ public static ProcedureBuilder builder() {
+ return new Builder<GrantPermissionProcedure>() {
+ @Override
+ protected GrantPermissionProcedure doBuild() {
+ return new GrantPermissionProcedure(tableCatalog());
+ }
+ };
+ }
+
+ @Override
+ public String description() {
+ return "GrantPermissionProcedure";
+ }
+}
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/ListPermissionsProcedure.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/ListPermissionsProcedure.java
new file mode 100644
index 0000000000..70d256e5c9
--- /dev/null
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/ListPermissionsProcedure.java
@@ -0,0 +1,160 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.spark.procedure;
+
+import org.apache.paimon.PagedList;
+import org.apache.paimon.management.ListPermissionsRequest;
+import org.apache.paimon.management.PermissionAssignment;
+import org.apache.paimon.management.PermissionColumns;
+import org.apache.paimon.management.ResourceType;
+
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.catalyst.util.GenericArrayData;
+import org.apache.spark.sql.connector.catalog.TableCatalog;
+import org.apache.spark.sql.types.Metadata;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+import org.apache.spark.unsafe.types.UTF8String;
+
+import java.util.List;
+
+import static org.apache.spark.sql.types.DataTypes.IntegerType;
+import static org.apache.spark.sql.types.DataTypes.StringType;
+import static org.apache.spark.sql.types.DataTypes.createArrayType;
+
+/** Lists direct permissions on an exact target. */
+public class ListPermissionsProcedure extends BasePermissionProcedure {
+
+ private static final ProcedureParameter[] PARAMETERS =
+ new ProcedureParameter[] {
+ ProcedureParameter.required("resource_type", StringType),
+ ProcedureParameter.optional("database", StringType),
+ ProcedureParameter.optional("table", StringType),
+ ProcedureParameter.optional("function", StringType),
+ ProcedureParameter.optional("view", StringType),
+ ProcedureParameter.optional("principal", StringType),
+ ProcedureParameter.optional("access", StringType),
+ ProcedureParameter.optional("max_results", IntegerType),
+ ProcedureParameter.optional("page_token", StringType)
+ };
+
+ private static final StructType OUTPUT_TYPE =
+ new StructType(
+ new StructField[] {
+ field("resource_type", StringType, false),
+ field("database", StringType, true),
+ field("table", StringType, true),
+ field("function", StringType, true),
+ field("view", StringType, true),
+ field("access", StringType, false),
+ field("principal", StringType, false),
+ field("column_names", createArrayType(StringType),
true),
+ field("excluded_column_names",
createArrayType(StringType), true),
+ field("expire_time", StringType, true),
+ field("next_page_token", StringType, true)
+ });
+
+ private ListPermissionsProcedure(TableCatalog tableCatalog) {
+ super(tableCatalog);
+ }
+
+ @Override
+ public ProcedureParameter[] parameters() {
+ return PARAMETERS;
+ }
+
+ @Override
+ public StructType outputType() {
+ return OUTPUT_TYPE;
+ }
+
+ @Override
+ public InternalRow[] call(InternalRow args) {
+ ResourceType resourceType =
+ enumValue(args.getString(0), ResourceType.class,
PARAMETERS[0].name());
+ Integer maxResults = args.isNullAt(7) ? null : args.getInt(7);
+ ListPermissionsRequest request =
+ new ListPermissionsRequest(
+ resourceType,
+ args.isNullAt(1) ? null :
emptyToNull(args.getString(1)),
+ args.isNullAt(2) ? null :
emptyToNull(args.getString(2)),
+ args.isNullAt(3) ? null :
emptyToNull(args.getString(3)),
+ args.isNullAt(4) ? null :
emptyToNull(args.getString(4)),
+ args.isNullAt(5) ? null :
emptyToNull(args.getString(5)),
+ args.isNullAt(6) ? null :
emptyToNull(args.getString(6)),
+ args.isNullAt(8) ? null :
emptyToNull(args.getString(8)),
+ maxResults);
+ PagedList<PermissionAssignment> page =
permissionManagement().listPermissions(request);
+ List<PermissionAssignment> assignments = page.getElements();
+ if (assignments == null || assignments.isEmpty()) {
+ return new InternalRow[0];
+ }
+
+ InternalRow[] rows = new InternalRow[assignments.size()];
+ for (int i = 0; i < assignments.size(); i++) {
+ PermissionAssignment assignment = assignments.get(i);
+ PermissionColumns columns = assignment.getColumns();
+ rows[i] =
+ newInternalRow(
+ string(assignment.getResource().getType().name()),
+ string(assignment.getResource().getDatabase()),
+ string(assignment.getResource().getTable()),
+ string(assignment.getResource().getFunction()),
+ string(assignment.getResource().getView()),
+ string(assignment.getAccess()),
+ string(assignment.getPrincipal()),
+ stringArray(columns == null ? null :
columns.getColumnNames()),
+ stringArray(columns == null ? null :
columns.getExcludedColumnNames()),
+ string(assignment.getExpireTime()),
+ string(page.getNextPageToken()));
+ }
+ return rows;
+ }
+
+ private static StructField field(
+ String name, org.apache.spark.sql.types.DataType type, boolean
nullable) {
+ return new StructField(name, type, nullable, Metadata.empty());
+ }
+
+ private static UTF8String string(String value) {
+ return value == null ? null : UTF8String.fromString(value);
+ }
+
+ private static GenericArrayData stringArray(List<String> values) {
+ if (values == null) {
+ return null;
+ }
+ return new GenericArrayData(
+
values.stream().map(ListPermissionsProcedure::string).toArray());
+ }
+
+ public static ProcedureBuilder builder() {
+ return new Builder<ListPermissionsProcedure>() {
+ @Override
+ protected ListPermissionsProcedure doBuild() {
+ return new ListPermissionsProcedure(tableCatalog());
+ }
+ };
+ }
+
+ @Override
+ public String description() {
+ return "ListPermissionsProcedure";
+ }
+}
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/ListPoliciesProcedure.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/ListPoliciesProcedure.java
new file mode 100644
index 0000000000..fa4d8a244f
--- /dev/null
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/ListPoliciesProcedure.java
@@ -0,0 +1,140 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.spark.procedure;
+
+import org.apache.paimon.PagedList;
+import org.apache.paimon.management.ColumnMask;
+import org.apache.paimon.management.DataPolicy;
+import org.apache.paimon.management.ListPoliciesRequest;
+import org.apache.paimon.management.PolicyType;
+import org.apache.paimon.management.RowFilter;
+
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.connector.catalog.TableCatalog;
+import org.apache.spark.sql.types.Metadata;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+import org.apache.spark.unsafe.types.UTF8String;
+
+import java.util.List;
+
+import static org.apache.spark.sql.types.DataTypes.IntegerType;
+import static org.apache.spark.sql.types.DataTypes.StringType;
+
+/** Lists policies attached to an exact table. */
+public class ListPoliciesProcedure extends BasePolicyProcedure {
+
+ private static final ProcedureParameter[] PARAMETERS =
+ new ProcedureParameter[] {
+ ProcedureParameter.required("database", StringType),
+ ProcedureParameter.required("table", StringType),
+ ProcedureParameter.optional("policy_type", StringType),
+ ProcedureParameter.optional("principal", StringType),
+ ProcedureParameter.optional("column", StringType),
+ ProcedureParameter.optional("max_results", IntegerType),
+ ProcedureParameter.optional("page_token", StringType)
+ };
+
+ private static final StructType OUTPUT_TYPE =
+ new StructType(
+ new StructField[] {
+ field("database", StringType, false),
+ field("table", StringType, false),
+ field("policy_type", StringType, false),
+ field("principal", StringType, false),
+ field("predicate_json", StringType, true),
+ field("on_column", StringType, true),
+ field("transform_json", StringType, true),
+ field("next_page_token", StringType, true)
+ });
+
+ private ListPoliciesProcedure(TableCatalog tableCatalog) {
+ super(tableCatalog);
+ }
+
+ @Override
+ public ProcedureParameter[] parameters() {
+ return PARAMETERS;
+ }
+
+ @Override
+ public StructType outputType() {
+ return OUTPUT_TYPE;
+ }
+
+ @Override
+ public InternalRow[] call(InternalRow args) {
+ ListPoliciesRequest request =
+ new ListPoliciesRequest(
+ tableResource(args.getString(0), args.getString(1)),
+ optionalEnum(
+ args.isNullAt(2) ? null : args.getString(2),
+ PolicyType.class,
+ PARAMETERS[2].name()),
+ args.isNullAt(3) ? null : args.getString(3),
+ args.isNullAt(4) ? null : args.getString(4),
+ args.isNullAt(6) ? null : args.getString(6),
+ args.isNullAt(5) ? null : args.getInt(5));
+ PagedList<DataPolicy> page = policyManagement().listPolicies(request);
+ List<DataPolicy> policies = page.getElements();
+ if (policies == null || policies.isEmpty()) {
+ return new InternalRow[0];
+ }
+ InternalRow[] rows = new InternalRow[policies.size()];
+ for (int i = 0; i < policies.size(); i++) {
+ DataPolicy policy = policies.get(i);
+ RowFilter rowFilter = policy.getRowFilter();
+ ColumnMask columnMask = policy.getColumnMask();
+ rows[i] =
+ newInternalRow(
+ string(policy.getResource().getDatabase()),
+ string(policy.getResource().getTable()),
+ string(policy.type().name()),
+ string(policy.getPrincipal()),
+ string(rowFilter == null ? null :
rowFilter.getPredicate()),
+ string(columnMask == null ? null :
columnMask.getOnColumn()),
+ string(columnMask == null ? null :
columnMask.getTransform()),
+ string(page.getNextPageToken()));
+ }
+ return rows;
+ }
+
+ private static StructField field(
+ String name, org.apache.spark.sql.types.DataType type, boolean
nullable) {
+ return new StructField(name, type, nullable, Metadata.empty());
+ }
+
+ private static UTF8String string(String value) {
+ return value == null ? null : UTF8String.fromString(value);
+ }
+
+ public static ProcedureBuilder builder() {
+ return new Builder<ListPoliciesProcedure>() {
+ @Override
+ protected ListPoliciesProcedure doBuild() {
+ return new ListPoliciesProcedure(tableCatalog());
+ }
+ };
+ }
+
+ @Override
+ public String description() {
+ return "ListPoliciesProcedure";
+ }
+}
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/RevokePermissionProcedure.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/RevokePermissionProcedure.java
new file mode 100644
index 0000000000..516c8d1ae1
--- /dev/null
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/RevokePermissionProcedure.java
@@ -0,0 +1,96 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.spark.procedure;
+
+import org.apache.paimon.management.PermissionResource;
+import org.apache.paimon.management.ResourceType;
+
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.connector.catalog.TableCatalog;
+import org.apache.spark.sql.types.DataTypes;
+import org.apache.spark.sql.types.Metadata;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+
+import static org.apache.spark.sql.types.DataTypes.StringType;
+
+/** Revokes a permission by its resource identity. */
+public class RevokePermissionProcedure extends BasePermissionProcedure {
+
+ private static final ProcedureParameter[] PARAMETERS =
+ new ProcedureParameter[] {
+ ProcedureParameter.required("resource_type", StringType),
+ ProcedureParameter.required("access", StringType),
+ ProcedureParameter.required("principal", StringType),
+ ProcedureParameter.optional("database", StringType),
+ ProcedureParameter.optional("table", StringType),
+ ProcedureParameter.optional("function", StringType),
+ ProcedureParameter.optional("view", StringType)
+ };
+
+ private static final StructType OUTPUT_TYPE =
+ new StructType(
+ new StructField[] {
+ new StructField("result", DataTypes.BooleanType,
false, Metadata.empty())
+ });
+
+ private RevokePermissionProcedure(TableCatalog tableCatalog) {
+ super(tableCatalog);
+ }
+
+ @Override
+ public ProcedureParameter[] parameters() {
+ return PARAMETERS;
+ }
+
+ @Override
+ public StructType outputType() {
+ return OUTPUT_TYPE;
+ }
+
+ @Override
+ public InternalRow[] call(InternalRow args) {
+ ResourceType resourceType =
+ enumValue(args.getString(0), ResourceType.class,
PARAMETERS[0].name());
+ PermissionResource resource =
+ resource(
+ resourceType,
+ args.isNullAt(3) ? null : args.getString(3),
+ args.isNullAt(4) ? null : args.getString(4),
+ args.isNullAt(5) ? null : args.getString(5),
+ args.isNullAt(6) ? null : args.getString(6));
+
+ permissionManagement().revokePermission(resource, args.getString(1),
args.getString(2));
+ return new InternalRow[] {newInternalRow(true)};
+ }
+
+ public static ProcedureBuilder builder() {
+ return new Builder<RevokePermissionProcedure>() {
+ @Override
+ protected RevokePermissionProcedure doBuild() {
+ return new RevokePermissionProcedure(tableCatalog());
+ }
+ };
+ }
+
+ @Override
+ public String description() {
+ return "RevokePermissionProcedure";
+ }
+}
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/AbstractPaimonSparkSqlExtensionsParser.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/AbstractPaimonSparkSqlExtensionsParser.scala
index 7770143c80..6cede203d4 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/AbstractPaimonSparkSqlExtensionsParser.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/AbstractPaimonSparkSqlExtensionsParser.scala
@@ -71,6 +71,7 @@ abstract class AbstractPaimonSparkSqlExtensionsParser(val
delegate: ParserInterf
PaimonSqlExtensionsParser.TAG,
PaimonSqlExtensionsParser.TRUE,
PaimonSqlExtensionsParser.FALSE,
+ PaimonSqlExtensionsParser.ARRAY,
PaimonSqlExtensionsParser.MAP,
PaimonSqlExtensionsParser.COPY,
PaimonSqlExtensionsParser.INTO,
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/PaimonSparkTestWithRestCatalogBase.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/PaimonSparkTestWithRestCatalogBase.scala
index b8b8d53e21..12b5c8febe 100644
---
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/PaimonSparkTestWithRestCatalogBase.scala
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/PaimonSparkTestWithRestCatalogBase.scala
@@ -30,7 +30,7 @@ import java.util.UUID
class PaimonSparkTestWithRestCatalogBase extends PaimonSparkTestBase {
- private var restCatalogServer: RESTCatalogServer = _
+ protected var restCatalogServer: RESTCatalogServer = _
private var serverUrl: String = _
protected var warehouse: String = _
private val initToken = "init_token"
@@ -43,13 +43,25 @@ class PaimonSparkTestWithRestCatalogBase extends
PaimonSparkTestBase {
"paimon",
CatalogOptions.WAREHOUSE.key,
warehouse),
- ImmutableMap.of())
+ ImmutableMap.of()
+ )
val authProvider = new BearTokenAuthProvider(initToken)
restCatalogServer =
new RESTCatalogServer(tempDBDir.getCanonicalPath, authProvider, config,
warehouse)
restCatalogServer.start()
serverUrl = restCatalogServer.getUrl
super.beforeAll()
+ Seq("analyst", "first", "second", "reader", "function_reader").foreach(
+ restCatalogServer.registerManagementPrincipal)
+ restCatalogServer.registerManagementPrincipal("analysts")
+ restCatalogServer.registerManagementPrincipal("admin")
+ spark.sql("CREATE DATABASE IF NOT EXISTS paimon.sales")
+ spark.sql("""CREATE TABLE IF NOT EXISTS paimon.sales.orders (
+ | id INT,
+ | region STRING,
+ | email STRING)
+ |TBLPROPERTIES ('query-auth.enabled' = 'true')
+ |""".stripMargin)
}
override protected def afterAll(): Unit = {
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/PermissionProcedureTest.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/PermissionProcedureTest.scala
new file mode 100644
index 0000000000..3e34fe56f2
--- /dev/null
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/PermissionProcedureTest.scala
@@ -0,0 +1,787 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.spark.procedure
+
+import org.apache.paimon.catalog.Identifier
+import org.apache.paimon.data.BinaryString
+import org.apache.paimon.management.{PermissionResource, ResourceType}
+import org.apache.paimon.predicate.{ConcatTransform, Equal, FieldRef,
FieldTransform, LeafPredicate}
+import org.apache.paimon.spark.{PaimonSparkTestBase,
PaimonSparkTestWithRestCatalogBase}
+import org.apache.paimon.types.DataTypes
+import org.apache.paimon.utils.JsonSerdeUtil
+
+import org.apache.spark.sql.Row
+import org.assertj.core.api.Assertions.assertThat
+
+import java.util.{Arrays, Collections}
+
+/** End-to-end tests for permission and policy management procedures. */
+class PermissionProcedureTest extends PaimonSparkTestWithRestCatalogBase {
+
+ test("grant, list and idempotently revoke a permission") {
+ checkAnswer(
+ spark.sql("""CALL sys.grant_permission(
+ | resource_type => 'table',
+ | access => 'select',
+ | principal => 'analyst',
+ | database => 'sales',
+ | table => 'orders',
+ | expire_time => '2027-01-01T00:00:00Z')
+ |""".stripMargin),
+ Row(true)
+ )
+
+ checkAnswer(
+ spark.sql("""CALL sys.grant_permission(
+ | resource_type => 'TABLE',
+ | access => 'SELECT',
+ | principal => 'analyst',
+ | database => 'sales',
+ | table => 'orders',
+ | expire_time => '2028-01-01T00:00:00Z')
+ |""".stripMargin),
+ Row(true)
+ )
+
+ val listed = spark.sql("""CALL sys.list_permissions(
+ | resource_type => 'TABLE',
+ | database => 'sales',
+ | table => 'orders',
+ | principal => 'analyst')
+ |""".stripMargin)
+ assertThat(listed.columns).containsExactly(
+ "resource_type",
+ "database",
+ "table",
+ "function",
+ "view",
+ "access",
+ "principal",
+ "column_names",
+ "excluded_column_names",
+ "expire_time",
+ "next_page_token")
+ val assignments = listed.collect()
+ assertThat(assignments).hasSize(1)
+ val assignment = assignments.head
+ assertThat(assignment.getString(0)).isEqualTo("TABLE")
+ assertThat(assignment.getString(1)).isEqualTo("sales")
+ assertThat(assignment.getString(2)).isEqualTo("orders")
+ assertThat(assignment.getString(5)).isEqualTo("SELECT")
+ assertThat(assignment.getString(6)).isEqualTo("analyst")
+ assertThat(assignment.isNullAt(7)).isTrue
+ assertThat(assignment.isNullAt(8)).isTrue
+ assertThat(assignment.getString(9)).isEqualTo("2028-01-01T00:00:00Z")
+ assertThat(assignment.isNullAt(10)).isTrue
+
+ val revoke = """CALL sys.revoke_permission(
+ | resource_type => 'TABLE',
+ | access => 'SELECT',
+ | principal => 'analyst',
+ | database => 'sales',
+ | table => 'orders')
+ |""".stripMargin
+ checkAnswer(spark.sql(revoke), Row(true))
+ checkAnswer(spark.sql(revoke), Row(true))
+ checkAnswer(
+ spark.sql("""CALL sys.list_permissions(
+ | resource_type => 'TABLE',
+ | database => 'sales',
+ | table => 'orders',
+ | principal => 'analyst')
+ |""".stripMargin),
+ Nil
+ )
+ }
+
+ test("list permissions supports opaque pagination tokens") {
+ grantCatalogPermission("first")
+ grantCatalogPermission("second")
+
+ val first = spark
+ .sql("CALL sys.list_permissions(resource_type => 'CATALOG', max_results
=> 1)")
+ .head()
+ assertThat(first.getString(6)).isEqualTo("first")
+ val pageToken = first.getString(10)
+ assertThat(pageToken).isNotEmpty.isNotEqualTo("1")
+
+ spark.sql(
+ "CALL sys.revoke_permission(resource_type => 'CATALOG', " +
+ "access => 'CREATEDATABASE', principal => 'first')")
+
+ val second = spark
+ .sql(
+ "CALL sys.list_permissions(resource_type => 'CATALOG', max_results =>
1, " +
+ s"page_token => ${sqlLiteral(pageToken)})"
+ )
+ .head()
+ assertThat(second.getString(6)).isEqualTo("second")
+ assertThat(second.isNullAt(10)).isTrue
+ }
+
+ test("grant and list explicit descendant scopes") {
+ checkAnswer(
+ spark.sql("""CALL sys.grant_permission(
+ | resource_type => 'CATALOG_ALL',
+ | access => 'SELECT',
+ | principal => 'analyst')
+ |""".stripMargin),
+ Row(true)
+ )
+ val catalogAll = spark
+ .sql(
+ "CALL sys.list_permissions(resource_type => 'CATALOG_ALL', principal
=> 'analyst')"
+ )
+ .head()
+ assertThat(catalogAll.getString(0)).isEqualTo("CATALOG_ALL")
+ assertThat(catalogAll.isNullAt(1)).isTrue
+ assertThat(catalogAll.getString(5)).isEqualTo("SELECT")
+
+ checkAnswer(
+ spark.sql("""CALL sys.grant_permission(
+ | resource_type => 'DATABASE_ALL',
+ | database => 'sales',
+ | access => 'UPDATE',
+ | principal => 'analyst')
+ |""".stripMargin),
+ Row(true)
+ )
+ val databaseAll = spark
+ .sql("""CALL sys.list_permissions(
+ | resource_type => 'DATABASE_ALL',
+ | database => 'sales',
+ | principal => 'analyst')
+ |""".stripMargin)
+ .head()
+ assertThat(databaseAll.getString(0)).isEqualTo("DATABASE_ALL")
+ assertThat(databaseAll.getString(1)).isEqualTo("sales")
+ assertThat(databaseAll.isNullAt(2)).isTrue
+ assertThat(databaseAll.getString(5)).isEqualTo("UPDATE")
+ }
+
+ test("grant, replace, list and enforce column permissions") {
+ restCatalogServer.setQueryPrincipals(Collections.singleton("analyst"))
+
+ checkAnswer(
+ spark.sql("""CALL sys.grant_permission(
+ | resource_type => 'COLUMN',
+ | access => 'SELECT',
+ | principal => 'analyst',
+ | database => 'sales',
+ | table => 'orders',
+ | column_names => array('id', 'region'))
+ |""".stripMargin),
+ Row(true)
+ )
+
+ val included = spark.sql("""CALL sys.list_permissions(
+ | resource_type => 'COLUMN',
+ | database => 'sales',
+ | table => 'orders',
+ | principal => 'analyst')
+ |""".stripMargin)
+ assertThat(included.columns).containsExactly(
+ "resource_type",
+ "database",
+ "table",
+ "function",
+ "view",
+ "access",
+ "principal",
+ "column_names",
+ "excluded_column_names",
+ "expire_time",
+ "next_page_token")
+ assertThat(included.head().getSeq[String](7)).isEqualTo(Seq("id",
"region"))
+ assertThat(included.head().isNullAt(8)).isTrue
+
+ checkAnswer(spark.sql("SELECT id, region FROM paimon.sales.orders"), Nil)
+ val deniedEmail = intercept[Exception] {
+ spark.sql("SELECT email FROM paimon.sales.orders").collect()
+ }
+ assertThat(deniedEmail.getMessage).contains("permission")
+
+ checkAnswer(
+ spark.sql("""CALL sys.grant_permission(
+ | resource_type => 'COLUMN',
+ | access => 'SELECT',
+ | principal => 'analyst',
+ | database => 'sales',
+ | table => 'orders',
+ | excluded_column_names => array('region'))
+ |""".stripMargin),
+ Row(true)
+ )
+ checkAnswer(spark.sql("SELECT id, email FROM paimon.sales.orders"), Nil)
+ val deniedRegion = intercept[Exception] {
+ spark.sql("SELECT region FROM paimon.sales.orders").collect()
+ }
+ assertThat(deniedRegion.getMessage).contains("permission")
+
+ restCatalogServer.registerManagementPrincipal("limited")
+ restCatalogServer.setQueryPrincipals(new
java.util.HashSet(Arrays.asList("analyst", "limited")))
+ checkAnswer(
+ spark.sql("""CALL sys.grant_permission(
+ | resource_type => 'COLUMN', access => 'SELECT',
+ | principal => 'limited', database => 'sales', table =>
'orders',
+ | column_names => array('id', 'region'))
+ |""".stripMargin),
+ Row(true)
+ )
+ checkAnswer(spark.sql("SELECT id FROM paimon.sales.orders"), Nil)
+ val deniedByIntersection = intercept[Exception] {
+ spark.sql("SELECT email FROM paimon.sales.orders").collect()
+ }
+ assertThat(deniedByIntersection.getMessage).contains("permission")
+
+ checkAnswer(
+ spark.sql("""CALL sys.revoke_permission(
+ | resource_type => 'COLUMN', access => 'SELECT',
+ | principal => 'analyst', database => 'sales', table =>
'orders')
+ |""".stripMargin),
+ Row(true)
+ )
+ }
+
+ test("column permission validates query authorization and referenced
columns") {
+ spark.sql("CREATE TABLE paimon.sales.disabled_columns (id INT)")
+ val disabled = intercept[Exception] {
+ spark
+ .sql("""CALL sys.grant_permission(
+ | resource_type => 'COLUMN', access => 'SELECT',
+ | principal => 'analyst', database => 'sales', table =>
'disabled_columns',
+ | column_names => array('id'))
+ |""".stripMargin)
+ .collect()
+ }
+ assertThat(disabled.getMessage).contains("query-auth.enabled=true")
+
+ val missing = intercept[Exception] {
+ spark
+ .sql("""CALL sys.grant_permission(
+ | resource_type => 'COLUMN', access => 'SELECT',
+ | principal => 'analyst', database => 'sales', table =>
'orders',
+ | excluded_column_names => array('missing'))
+ |""".stripMargin)
+ .collect()
+ }
+ assertThat(missing.getMessage).contains("Permission column does not exist")
+ }
+
+ test("column permissions follow table and schema lifecycle") {
+ restCatalogServer.setQueryPrincipals(Collections.singleton("analyst"))
+ spark.sql("""CREATE TABLE paimon.sales.column_lifecycle (
+ | id INT,
+ | secret STRING)
+ |TBLPROPERTIES ('query-auth.enabled' = 'true')
+ |""".stripMargin)
+ spark.sql("INSERT INTO paimon.sales.column_lifecycle VALUES (1, 's1')")
+ checkAnswer(
+ spark.sql("""CALL sys.grant_permission(
+ | resource_type => 'COLUMN', access => 'SELECT', principal
=> 'analyst',
+ | database => 'sales', table => 'column_lifecycle',
+ | excluded_column_names => array('secret'))
+ |""".stripMargin),
+ Row(true)
+ )
+
+ spark.sql(
+ "ALTER TABLE paimon.sales.column_lifecycle RENAME TO
paimon.sales.renamed_column_lifecycle")
+ var assignment = spark
+ .sql(
+ "CALL sys.list_permissions(resource_type => 'COLUMN', database =>
'sales', " +
+ "table => 'renamed_column_lifecycle', principal => 'analyst')")
+ .head()
+ assertThat(assignment.getString(2)).isEqualTo("renamed_column_lifecycle")
+ assertThat(assignment.getSeq[String](8)).isEqualTo(Seq("secret"))
+
+ val disableAuth = intercept[Exception] {
+ spark
+ .sql("ALTER TABLE paimon.sales.renamed_column_lifecycle " +
+ "SET TBLPROPERTIES ('query-auth.enabled' = 'false')")
+ .collect()
+ }
+ assertThat(disableAuth.getMessage).contains("Cannot disable
query-auth.enabled")
+
+ spark.sql(
+ "ALTER TABLE paimon.sales.renamed_column_lifecycle " +
+ "RENAME COLUMN secret TO private_secret")
+ assignment = spark
+ .sql(
+ "CALL sys.list_permissions(resource_type => 'COLUMN', database =>
'sales', " +
+ "table => 'renamed_column_lifecycle', principal => 'analyst')")
+ .head()
+ assertThat(assignment.getSeq[String](8)).isEqualTo(Seq("private_secret"))
+ val denied = intercept[Exception] {
+ spark.sql("SELECT private_secret FROM
paimon.sales.renamed_column_lifecycle").collect()
+ }
+ assertThat(denied.getMessage).contains("permission")
+
+ spark.sql("""CALL sys.grant_permission(
+ | resource_type => 'COLUMN', access => 'SELECT', principal =>
'analyst',
+ | database => 'sales', table => 'renamed_column_lifecycle',
+ | column_names => array('private_secret'))
+ |""".stripMargin)
+ val dropOnlyAllowedColumn = intercept[Exception] {
+ spark
+ .sql("ALTER TABLE paimon.sales.renamed_column_lifecycle DROP COLUMN
private_secret")
+ .collect()
+ }
+ assertThat(dropOnlyAllowedColumn.getMessage).contains("Cannot drop every
allowed column")
+
+ spark.sql("""CALL sys.grant_permission(
+ | resource_type => 'COLUMN', access => 'SELECT', principal =>
'analyst',
+ | database => 'sales', table => 'renamed_column_lifecycle',
+ | excluded_column_names => array('private_secret'))
+ |""".stripMargin)
+ spark.sql("ALTER TABLE paimon.sales.renamed_column_lifecycle DROP COLUMN
private_secret")
+ checkAnswer(
+ spark.sql(
+ "CALL sys.list_permissions(resource_type => 'COLUMN', database =>
'sales', " +
+ "table => 'renamed_column_lifecycle', principal => 'analyst')"),
+ Nil
+ )
+
+ grantTablePermission("renamed_column_lifecycle", "analyst")
+ spark.sql("DROP TABLE paimon.sales.renamed_column_lifecycle")
+ spark.sql("CREATE TABLE paimon.sales.renamed_column_lifecycle (id INT)")
+ checkAnswer(
+ spark.sql(
+ "CALL sys.list_permissions(resource_type => 'TABLE', database =>
'sales', " +
+ "table => 'renamed_column_lifecycle', principal => 'analyst')"),
+ Nil
+ )
+ }
+
+ test("create, reject duplicates, apply and idempotently drop table data
policies") {
+ restCatalogServer.setQueryPrincipals(Collections.singleton("analyst"))
+ spark.sql(
+ "INSERT OVERWRITE paimon.sales.orders VALUES " +
+ "(1, 'APAC', '[email protected]'), (2, 'EMEA', '[email protected]')")
+
+ val apacFilter = stringEqualsPredicate(1, "region", "APAC")
+ val emeaFilter = stringEqualsPredicate(1, "region", "EMEA")
+ val emailMask = concatFieldTransform(1, "region", "-masked")
+
+ checkAnswer(
+ spark.sql(s"""CALL sys.create_policy(
+ | database => 'sales',
+ | table => 'orders',
+ | policy_type => 'ROW_FILTER',
+ | principal => 'analyst',
+ | predicate_json => ${sqlLiteral(apacFilter)})
+ |""".stripMargin),
+ Row(true)
+ )
+
+ val listed = spark.sql("""CALL sys.list_policies(
+ | database => 'sales',
+ | table => 'orders',
+ | policy_type => 'ROW_FILTER',
+ | principal => 'analyst')
+ |""".stripMargin)
+ assertThat(listed.columns).containsExactly(
+ "database",
+ "table",
+ "policy_type",
+ "principal",
+ "predicate_json",
+ "on_column",
+ "transform_json",
+ "next_page_token")
+ val direct = listed.head()
+ assertThat(direct.getString(2)).isEqualTo("ROW_FILTER")
+ assertThat(direct.getString(3)).isEqualTo("analyst")
+ assertThat(direct.getString(4)).contains("\"name\":\"region\"")
+ assertThat(direct.isNullAt(5)).isTrue
+ assertThat(direct.isNullAt(6)).isTrue
+ checkAnswer(
+ spark.sql("SELECT id, region, email FROM paimon.sales.orders ORDER BY
id"),
+ Row(1, "APAC", "[email protected]")
+ )
+
+ val duplicate = intercept[Exception] {
+ spark
+ .sql(s"""CALL sys.create_policy(
+ | database => 'sales',
+ | table => 'orders',
+ | policy_type => 'ROW_FILTER',
+ | principal => 'analyst',
+ | predicate_json => ${sqlLiteral(emeaFilter)})
+ |""".stripMargin)
+ .collect()
+ }
+ assertThat(duplicate.getMessage).contains("already exists")
+ val unchanged = spark
+ .sql("""CALL sys.list_policies(
+ | database => 'sales',
+ | table => 'orders',
+ | policy_type => 'ROW_FILTER',
+ | principal => 'analyst')
+ |""".stripMargin)
+ .head()
+ assertThat(unchanged.getString(4)).contains("APAC").doesNotContain("EMEA")
+
+ checkAnswer(
+ spark.sql(s"""CALL sys.create_policy(
+ | database => 'sales',
+ | table => 'orders',
+ | policy_type => 'COLUMN_MASKING',
+ | principal => 'analyst',
+ | on_column => 'email',
+ | transform_json => ${sqlLiteral(emailMask)})
+ |""".stripMargin),
+ Row(true)
+ )
+
+ val mask = spark
+ .sql("""CALL sys.list_policies(
+ | database => 'sales',
+ | table => 'orders',
+ | policy_type => 'COLUMN_MASKING')
+ |""".stripMargin)
+ .head()
+ assertThat(mask.getString(2)).isEqualTo("COLUMN_MASKING")
+ assertThat(mask.getString(3)).isEqualTo("analyst")
+ assertThat(mask.isNullAt(4)).isTrue
+ assertThat(mask.getString(5)).isEqualTo("email")
+ assertThat(mask.getString(6)).contains("-masked")
+
+ checkAnswer(
+ spark.sql("SELECT id, region, email FROM paimon.sales.orders ORDER BY
id"),
+ Row(1, "APAC", "APAC-masked")
+ )
+ checkAnswer(
+ spark.sql("SELECT email FROM paimon.sales.orders"),
+ Row("APAC-masked")
+ )
+ checkAnswer(
+ spark.sql("""CALL sys.drop_policy(
+ | database => 'sales',
+ | table => 'orders',
+ | policy_type => 'ROW_FILTER',
+ | principal => 'analyst',
+ | if_exists => true)
+ |""".stripMargin),
+ Row(true)
+ )
+ checkAnswer(
+ spark.sql("""CALL sys.drop_policy(
+ | database => 'sales',
+ | table => 'orders',
+ | policy_type => 'ROW_FILTER',
+ | principal => 'analyst',
+ | if_exists => true)
+ |""".stripMargin),
+ Row(true)
+ )
+ checkAnswer(
+ spark.sql("""CALL sys.list_policies(
+ | database => 'sales',
+ | table => 'orders',
+ | policy_type => 'ROW_FILTER',
+ | principal => 'analyst')
+ |""".stripMargin),
+ Nil
+ )
+ spark.sql(
+ "CALL sys.drop_policy(database => 'sales', table => 'orders', " +
+ "policy_type => 'COLUMN_MASKING', " +
+ "principal => 'analyst', column => 'email', if_exists => true)")
+ }
+
+ test("policy creation validates query authorization, JSON and columns") {
+ val validFilter = intEqualsPredicate(0, "id", 1)
+ spark.sql("CREATE TABLE paimon.sales.disabled_orders (id INT)")
+
+ val disabled = intercept[Exception] {
+ spark
+ .sql(s"""CALL sys.create_policy(
+ | database => 'sales', table => 'disabled_orders',
+ | policy_type => 'ROW_FILTER', predicate_json =>
${sqlLiteral(validFilter)},
+ | principal => 'analyst')
+ |""".stripMargin)
+ .collect()
+ }
+ assertThat(disabled.getMessage).contains("query-auth.enabled=true")
+
+ val malformed = intercept[Exception] {
+ spark
+ .sql("""CALL sys.create_policy(
+ | database => 'sales', table => 'orders',
+ | policy_type => 'ROW_FILTER', predicate_json => '{bad',
+ | principal => 'analyst')
+ |""".stripMargin)
+ .collect()
+ }
+ assertThat(malformed.getMessage).contains("Unexpected character")
+
+ val unknownColumnFilter = stringEqualsPredicate(0, "unknown", "APAC")
+ val missingColumn = intercept[Exception] {
+ spark
+ .sql(s"""CALL sys.create_policy(
+ | database => 'sales', table => 'orders',
+ | policy_type => 'ROW_FILTER', predicate_json =>
${sqlLiteral(unknownColumnFilter)},
+ | principal => 'analyst')
+ |""".stripMargin)
+ .collect()
+ }
+ assertThat(missingColumn.getMessage).contains("column unknown")
+
+ val missingPrincipal = intercept[Exception] {
+ spark
+ .sql(s"""CALL sys.create_policy(
+ | database => 'sales', table => 'orders',
+ | policy_type => 'ROW_FILTER', predicate_json =>
${sqlLiteral(validFilter)},
+ | principal => 'missing')
+ |""".stripMargin)
+ .collect()
+ }
+ assertThat(missingPrincipal.getMessage).contains("principal does not
exist")
+
+ val jsonNull = intercept[Exception] {
+ spark
+ .sql("""CALL sys.create_policy(
+ | database => 'sales', table => 'orders',
+ | policy_type => 'ROW_FILTER', predicate_json => 'null',
+ | principal => 'analyst')
+ |""".stripMargin)
+ .collect()
+ }
+ assertThat(jsonNull.getMessage).contains("JSON null")
+
+ val mixedDefinition = intercept[Exception] {
+ spark
+ .sql(s"""CALL sys.create_policy(
+ | database => 'sales', table => 'orders',
+ | policy_type => 'ROW_FILTER', predicate_json =>
${sqlLiteral(validFilter)},
+ | transform_json =>
${sqlLiteral(constantStringTransform("****"))},
+ | principal => 'analyst')
+ |""".stripMargin)
+ .collect()
+ }
+ assertThat(mixedDefinition.getMessage).contains("cannot specify transform")
+ }
+
+ test("list policies supports opaque pagination tokens") {
+ val filter = intEqualsPredicate(0, "id", 1)
+ Seq("first", "second").foreach {
+ principal =>
+ checkAnswer(
+ spark.sql(s"""CALL sys.create_policy(
+ | database => 'sales', table => 'orders',
+ | policy_type => 'ROW_FILTER', principal =>
'$principal',
+ | predicate_json => ${sqlLiteral(filter)})
+ |""".stripMargin),
+ Row(true)
+ )
+ }
+ try {
+ val first = spark
+ .sql("CALL sys.list_policies(database => 'sales', table => 'orders',
max_results => 1)")
+ .head()
+ assertThat(first.getString(3)).isEqualTo("first")
+ val pageToken = first.getString(7)
+ assertThat(pageToken).isNotEmpty.isNotEqualTo("1")
+
+ spark.sql(
+ "CALL sys.drop_policy(database => 'sales', table => 'orders', " +
+ "policy_type => 'ROW_FILTER', principal => 'first', if_exists =>
true)")
+
+ val second = spark
+ .sql(
+ "CALL sys.list_policies(database => 'sales', table => 'orders', " +
+ s"max_results => 1, page_token => ${sqlLiteral(pageToken)})")
+ .head()
+ assertThat(second.getString(3)).isEqualTo("second")
+ assertThat(second.isNullAt(7)).isTrue
+ } finally {
+ Seq("first", "second").foreach {
+ principal =>
+ spark.sql(
+ "CALL sys.drop_policy(database => 'sales', table => 'orders', " +
+ s"policy_type => 'ROW_FILTER', principal => '$principal',
if_exists => true)")
+ }
+ }
+ }
+
+ test("table lifecycle preserves policy enforcement and rejects unsafe schema
changes") {
+ restCatalogServer.setQueryPrincipals(Collections.singleton("analyst"))
+ spark.sql("""CREATE TABLE paimon.sales.lifecycle_orders (
+ | id INT,
+ | region STRING)
+ |TBLPROPERTIES ('query-auth.enabled' = 'true')
+ |""".stripMargin)
+ spark.sql("INSERT INTO paimon.sales.lifecycle_orders VALUES (1, 'APAC'),
(2, 'EMEA')")
+ val lifecycleFilter = stringEqualsPredicate(1, "region", "APAC")
+ checkAnswer(
+ spark.sql(s"""CALL sys.create_policy(
+ | database => 'sales', table => 'lifecycle_orders',
+ | policy_type => 'ROW_FILTER', predicate_json =>
${sqlLiteral(lifecycleFilter)},
+ | principal => 'analyst')
+ |""".stripMargin),
+ Row(true)
+ )
+
+ spark.sql("ALTER TABLE paimon.sales.lifecycle_orders RENAME TO
paimon.sales.renamed_orders")
+ assertThat(
+ spark
+ .sql("CALL sys.list_policies(database => 'sales', table =>
'renamed_orders', " +
+ "policy_type => 'ROW_FILTER', principal => 'analyst')")
+ .head()
+ .getString(1)).isEqualTo("renamed_orders")
+ assertThat(
+ paimonCatalog
+ .authTableQuery(Identifier.create("sales", "renamed_orders"), null)
+ .extractPredicate()).isNotNull
+
+ val disableAuth = intercept[Exception] {
+ spark
+ .sql("ALTER TABLE paimon.sales.renamed_orders " +
+ "SET TBLPROPERTIES ('query-auth.enabled' = 'false')")
+ .collect()
+ }
+ assertThat(disableAuth.getMessage).contains("Cannot disable
query-auth.enabled")
+
+ val renameColumn = intercept[Exception] {
+ spark
+ .sql("ALTER TABLE paimon.sales.renamed_orders RENAME COLUMN region TO
area")
+ .collect()
+ }
+ assertThat(renameColumn.getMessage).contains("column region")
+
+ spark.sql("DROP TABLE paimon.sales.renamed_orders")
+ spark.sql("""CREATE TABLE paimon.sales.renamed_orders (
+ | id INT,
+ | region STRING)
+ |TBLPROPERTIES ('query-auth.enabled' = 'true')
+ |""".stripMargin)
+ spark.sql("INSERT INTO paimon.sales.renamed_orders VALUES (1, 'APAC'), (2,
'EMEA')")
+ checkAnswer(
+ spark.sql("SELECT id, region FROM paimon.sales.renamed_orders ORDER BY
id"),
+ Seq(Row(1, "APAC"), Row(2, "EMEA"))
+ )
+ checkAnswer(
+ spark.sql(
+ "CALL sys.list_policies(database => 'sales', table =>
'renamed_orders', " +
+ "policy_type => 'ROW_FILTER', principal => 'analyst')"),
+ Nil
+ )
+ assertThat(
+ paimonCatalog
+ .authTableQuery(Identifier.create("sales", "renamed_orders"), null)
+ .extractPredicate()).isNull
+ }
+
+ test("management endpoints enforce target authorization") {
+ val resource =
+ new PermissionResource(ResourceType.TABLE, "sales", "orders", null, null)
+ restCatalogServer.denyManagementPermission(resource)
+ try {
+ val permissionError = intercept[Exception] {
+ spark
+ .sql("""CALL sys.grant_permission(
+ | resource_type => 'TABLE', access => 'SELECT',
+ | principal => 'analyst',
+ | database => 'sales', table => 'orders')
+ |""".stripMargin)
+ .collect()
+ }
+ assertThat(permissionError.getMessage).contains("cannot manage
permissions")
+
+ val policyError = intercept[Exception] {
+ spark
+ .sql("CALL sys.list_policies(database => 'sales', table =>
'orders')")
+ .collect()
+ }
+ assertThat(policyError.getMessage).contains("cannot manage permissions")
+ } finally {
+ restCatalogServer.allowManagementPermission(resource)
+ }
+ }
+
+ private def grantCatalogPermission(principal: String): Unit = {
+ checkAnswer(
+ spark.sql(s"""CALL sys.grant_permission(
+ | resource_type => 'CATALOG',
+ | access => 'CREATEDATABASE',
+ | principal => '$principal')
+ |""".stripMargin),
+ Row(true)
+ )
+ }
+
+ private def grantTablePermission(table: String, principal: String): Unit = {
+ checkAnswer(
+ spark.sql(s"""CALL sys.grant_permission(
+ | resource_type => 'TABLE', access => 'SELECT', principal
=> '$principal',
+ | database => 'sales', table => '$table')
+ |""".stripMargin),
+ Row(true)
+ )
+ }
+
+ private def stringEqualsPredicate(index: Int, column: String, constant:
String): String = {
+ JsonSerdeUtil.toFlatJson(
+ LeafPredicate.of(
+ new FieldTransform(new FieldRef(index, column, DataTypes.STRING())),
+ Equal.INSTANCE,
+ Collections.singletonList(BinaryString.fromString(constant))))
+ }
+
+ private def intEqualsPredicate(index: Int, column: String, constant: Int):
String = {
+ JsonSerdeUtil.toFlatJson(
+ LeafPredicate.of(
+ new FieldTransform(new FieldRef(index, column, DataTypes.INT())),
+ Equal.INSTANCE,
+ Collections.singletonList(Integer.valueOf(constant))))
+ }
+
+ private def constantStringTransform(constant: String): String = {
+ JsonSerdeUtil.toFlatJson(
+ new
ConcatTransform(Collections.singletonList(BinaryString.fromString(constant))))
+ }
+
+ private def concatFieldTransform(index: Int, column: String, suffix:
String): String = {
+ JsonSerdeUtil.toFlatJson(
+ new ConcatTransform(Arrays
+ .asList(new FieldRef(index, column, DataTypes.STRING()),
BinaryString.fromString(suffix))))
+ }
+
+ private def sqlLiteral(value: String): String = {
+ "'" + value.replace("'", "''") + "'"
+ }
+}
+
+/** Management procedures must fail clearly for catalogs without the REST
capability. */
+class PermissionProcedureUnsupportedCatalogTest extends PaimonSparkTestBase {
+
+ test("filesystem catalog does not expose permission management") {
+ val error = intercept[IllegalArgumentException] {
+ spark
+ .sql("""CALL sys.grant_permission(
+ | resource_type => 'CATALOG',
+ | access => 'CREATEDATABASE',
+ | principal => 'admin')
+ |""".stripMargin)
+ .collect()
+ }
+ assertThat(error.getMessage).contains("does not support permission or
policy management")
+ }
+}
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/CatalogQualifiedCreateTableLikeTest.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/CatalogQualifiedCreateTableLikeTest.scala
index c4eb2cd644..350462fc1c 100644
---
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/CatalogQualifiedCreateTableLikeTest.scala
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/CatalogQualifiedCreateTableLikeTest.scala
@@ -139,6 +139,11 @@ class CatalogQualifiedCreateTableLikeTest extends
PaimonSparkTestBase {
Assertions.assertEquals("skip_file", skipFileCommand.targetIdent.name())
Assertions.assertEquals(Seq("test"),
skipFileCommand.targetIdent.namespace().toSeq)
+ val arrayCommand =
+ parseCreateTableLikeCommand("CREATE TABLE paimon.test.array LIKE
paimon.test.source_tbl")
+ Assertions.assertEquals("array", arrayCommand.targetIdent.name())
+ Assertions.assertEquals(Seq("test"),
arrayCommand.targetIdent.namespace().toSeq)
+
val nestedIdentifierCommand =
parseCreateTableLikeCommand(
"CREATE TABLE paimon.test.extra.target_tbl LIKE
paimon.test.extra.source_tbl")
