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 4cda48260e [flink] Support grant, revoke and list_permissions
procedures (#10437)
4cda48260e is described below
commit 4cda48260eee573bd6f5a3f47453cd4aae706655
Author: Wenchao Wu <[email protected]>
AuthorDate: Mon Oct 12 21:28:40 2026 +0800
[flink] Support grant, revoke and list_permissions procedures (#10437)
---
docs/docs/concepts/rest/management-api.md | 35 ++
docs/docs/flink/procedures.md | 1 +
docs/docs/flink/procedures/permissions.md | 176 +++++++++
.../flink/procedure/BasePermissionProcedure.java | 159 ++++++++
.../flink/procedure/GrantPermissionProcedure.java | 98 +++++
.../flink/procedure/ListPermissionsProcedure.java | 128 ++++++
.../flink/procedure/RevokePermissionProcedure.java | 81 ++++
.../services/org.apache.paimon.factories.Factory | 3 +
.../flink/procedure/PermissionProcedureITCase.java | 428 +++++++++++++++++++++
...ermissionProcedureUnsupportedCatalogITCase.java | 41 ++
10 files changed, 1150 insertions(+)
diff --git a/docs/docs/concepts/rest/management-api.md
b/docs/docs/concepts/rest/management-api.md
index d604b800b6..bd05c10941 100644
--- a/docs/docs/concepts/rest/management-api.md
+++ b/docs/docs/concepts/rest/management-api.md
@@ -662,6 +662,41 @@ CALL paimon.sys.drop_policy(
);
```
+## Flink SQL procedures
+
+Flink 1.19 and later can call the same three permission procedures on a REST
catalog. Arguments,
+identity rules, and output columns match the Spark procedures above. A
filesystem catalog fails
+with `Catalog does not support permission management.`
+
+```sql
+CALL sys.grant_permission(
+ resource_type => 'TABLE',
+ `database` => 'sales',
+ `table` => 'orders',
+ access => 'SELECT',
+ principal => 'user:alice'
+);
+
+CALL sys.list_permissions(
+ resource_type => 'TABLE',
+ `database` => 'sales',
+ `table` => 'orders'
+);
+
+CALL sys.revoke_permission(
+ resource_type => 'TABLE',
+ `database` => 'sales',
+ `table` => 'orders',
+ access => 'SELECT',
+ principal => 'user:alice'
+);
+```
+
+Column ranges are a JSON array of exact column names, for example
+`column_names => '["order_id", "region"]'`. `list_permissions` returns those
names as
+`ARRAY<STRING>`. Argument details and positional calls are in
+[Flink permission procedures](../../flink/procedures/permissions).
+
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/docs/docs/flink/procedures.md b/docs/docs/flink/procedures.md
index 883324e889..1e7747ef55 100644
--- a/docs/docs/flink/procedures.md
+++ b/docs/docs/flink/procedures.md
@@ -74,3 +74,4 @@ defaults are specific to the procedure.
| [Consumers and Query Service](./procedures/consumers) |
[`reset_consumer`](./procedures/consumers#reset_consumer),
[`clear_consumers`](./procedures/consumers#clear_consumers),
[`query_service`](./procedures/consumers#query_service) |
| [Cleanup and Repair](./procedures/repair) |
[`remove_orphan_files`](./procedures/repair#remove_orphan_files),
[`remove_orphan_blobs`](./procedures/repair#remove_orphan_blobs),
[`remove_unexisting_files`](./procedures/repair#remove_unexisting_files),
[`remove_unexisting_manifests`](./procedures/repair#remove_unexisting_manifests),
[`repair`](./procedures/repair#repair),
[`repair_earliest_snapshot`](./procedures/repair#repair_earliest_snapshot) |
| [Views and Functions](./procedures/catalog) |
[`alter_view_dialect`](./procedures/catalog#alter_view_dialect),
[`create_function`](./procedures/catalog#create_function),
[`alter_function`](./procedures/catalog#alter_function),
[`drop_function`](./procedures/catalog#drop_function) |
+| [Permissions](./procedures/permissions) |
[`grant_permission`](./procedures/permissions#grant_permission),
[`revoke_permission`](./procedures/permissions#revoke_permission),
[`list_permissions`](./procedures/permissions#list_permissions) |
diff --git a/docs/docs/flink/procedures/permissions.md
b/docs/docs/flink/procedures/permissions.md
new file mode 100644
index 0000000000..117e7da64e
--- /dev/null
+++ b/docs/docs/flink/procedures/permissions.md
@@ -0,0 +1,176 @@
+---
+title: "Permissions"
+sidebar_position: 9
+---
+
+<!--
+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.
+-->
+
+# Permissions
+
+Grant, revoke, and list direct permission assignments on a [REST
catalog](../../concepts/rest/management-api).
+These procedures require Flink 1.19 or later. Named and positional calls both
use that API.
+Other catalogs fail with `Catalog does not support permission management.`
+
+`resource_type` is one of `CATALOG`, `CATALOG_ALL`, `DATABASE`,
`DATABASE_ALL`, `TABLE`, `COLUMN`,
+`VIEW`, or `FUNCTION`. Names are case-insensitive. `CATALOG_ALL` and
`DATABASE_ALL` grant an access
+on every applicable object in that scope; they do not grant catalog or
database administration
+access such as `CREATEDATABASE`.
+
+Repeating `grant_permission` for the same resource, access, and principal
replaces `expire_time`
+and the column range. Repeating `revoke_permission` succeeds when the
assignment is already absent.
+`list_permissions` returns only direct assignments on the requested resource.
It does not
+synthesize assignments inherited from a parent scope. `next_page_token` is
opaque and is repeated
+on every row of a page; pass it back unchanged with the same filters.
+
+See [Procedures](../procedures) for Flink version requirements, argument
conventions, and catalog
+selection. The portable contract is described in [REST Management
API](../../concepts/rest/management-api).
+
+## grant_permission
+
+Grant one permission assignment.
+
+**Arguments**
+
+- `resource_type` (`STRING`, required): resource or explicit descendant scope.
+- `access` (`STRING`, required): permission to grant, such as `SELECT` or
`CREATEDATABASE`.
+- `principal` (`STRING`, required): principal that receives the permission.
+- `database` (`STRING`, optional): database locator. Required for database,
table, column, view, and function resources.
+- `table` (`STRING`, optional): table locator. Required for `TABLE` and
`COLUMN`.
+- `function` (`STRING`, optional): function locator. Required for `FUNCTION`.
+- `view` (`STRING`, optional): view locator. Required for `VIEW`.
+- `expire_time` (`STRING`, optional): exclusive expiration instant, for
example `2027-01-01T00:00:00Z`.
+- `column_names` (`STRING`, optional): JSON array allowlist for a `COLUMN`
assignment.
+- `excluded_column_names` (`STRING`, optional): JSON array denylist for a
`COLUMN` assignment.
+
+A `COLUMN` assignment sets exactly one of `column_names` or
`excluded_column_names`. The value is a
+JSON array of strings, for example `["order_id", "region"]`. Each element is
the exact column name,
+including leading or trailing spaces and embedded commas. The table must
already have
+`query-auth.enabled` set to `true`.
+
+**Output**
+
+| Column | Type | Description |
+| --- | --- | --- |
+| `result` | `BOOLEAN` | `true` when the server accepts the assignment. |
+
+**Example**
+
+```sql
+-- Flink 1.19 and later, named arguments
+CALL sys.grant_permission(
+ resource_type => 'TABLE',
+ `database` => 'sales',
+ `table` => 'orders',
+ access => 'SELECT',
+ principal => 'analyst',
+ expire_time => '2027-01-01T00:00:00Z'
+);
+
+CALL sys.grant_permission(
+ resource_type => 'COLUMN',
+ `database` => 'sales',
+ `table` => 'orders',
+ access => 'SELECT',
+ principal => 'analyst',
+ column_names => '["order_id", "region"]'
+);
+
+-- Flink 1.19 and later, positional arguments in parameter order
+CALL sys.grant_permission(
+ 'TABLE', 'SELECT', 'analyst', 'sales', 'orders', '', '',
'2027-01-01T00:00:00Z');
+```
+
+## list_permissions
+
+List direct assignments on one resource.
+
+**Arguments**
+
+- `resource_type` (`STRING`, required): resource or explicit descendant scope
to list.
+- `database` (`STRING`, optional): database locator.
+- `table` (`STRING`, optional): table locator.
+- `function` (`STRING`, optional): function locator.
+- `view` (`STRING`, optional): view locator.
+- `principal` (`STRING`, optional): return only this principal.
+- `access` (`STRING`, optional): return only this access.
+- `max_results` (`INT`, optional): maximum number of assignments in the page.
+- `page_token` (`STRING`, optional): opaque token from the previous page.
+
+**Output**
+
+| Column | Type | Description |
+| --- | --- | --- |
+| `resource_type` | `STRING` | Resource type of the assignment. |
+| `database` | `STRING` | Database locator, or `NULL`. |
+| `table` | `STRING` | Table locator, or `NULL`. |
+| `function` | `STRING` | Function locator, or `NULL`. |
+| `view` | `STRING` | View locator, or `NULL`. |
+| `access` | `STRING` | Granted access. |
+| `principal` | `STRING` | Principal that holds the access. |
+| `column_names` | `ARRAY<STRING>` | Column allowlist, or `NULL`. |
+| `excluded_column_names` | `ARRAY<STRING>` | Column denylist, or `NULL`. |
+| `expire_time` | `STRING` | Exclusive expiration instant, or `NULL`. |
+| `next_page_token` | `STRING` | Opaque token for the next page, or `NULL` on
the last page. |
+
+**Example**
+
+```sql
+CALL sys.list_permissions(
+ resource_type => 'TABLE',
+ `database` => 'sales',
+ `table` => 'orders',
+ principal => 'analyst',
+ max_results => 50,
+ page_token => 'opaque-token-from-previous-row'
+);
+```
+
+## revoke_permission
+
+Remove one permission assignment. `expire_time` and the column range are not
part of the identity.
+Revoking a `COLUMN` assignment removes the whole range for that resource,
access, and principal.
+
+**Arguments**
+
+- `resource_type` (`STRING`, required): resource or explicit descendant scope.
+- `access` (`STRING`, required): permission to revoke.
+- `principal` (`STRING`, required): principal that holds the permission.
+- `database` (`STRING`, optional): database locator.
+- `table` (`STRING`, optional): table locator.
+- `function` (`STRING`, optional): function locator.
+- `view` (`STRING`, optional): view locator.
+
+**Output**
+
+| Column | Type | Description |
+| --- | --- | --- |
+| `result` | `BOOLEAN` | `true` when the server accepts the revocation. |
+
+**Example**
+
+```sql
+CALL sys.revoke_permission(
+ resource_type => 'TABLE',
+ `database` => 'sales',
+ `table` => 'orders',
+ access => 'SELECT',
+ principal => 'analyst'
+);
+```
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/BasePermissionProcedure.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/BasePermissionProcedure.java
new file mode 100644
index 0000000000..8990549c5e
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/BasePermissionProcedure.java
@@ -0,0 +1,159 @@
+/*
+ * 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.flink.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.ResourceType;
+import org.apache.paimon.rest.RESTCatalog;
+import org.apache.paimon.utils.JsonSerdeUtil;
+import org.apache.paimon.utils.StringUtils;
+
+import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.DeserializationFeature;
+import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.JsonNode;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Locale;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/** Shared REST catalog lookup and argument validation for permission
procedures. */
+abstract class BasePermissionProcedure extends ProcedureBase {
+
+ private static final String COLUMN_NAMES_FORMAT_MESSAGE =
+ "Column names must be a JSON array of strings, for example
[\"col1\", \"col2\"].";
+
+ protected PermissionManagement permissionManagement() {
+ return restCatalog().permissionManagement();
+ }
+
+ private RESTCatalog restCatalog() {
+ Catalog root = DelegateCatalog.rootCatalog(catalog);
+ checkArgument(
+ root instanceof RESTCatalog, "Catalog does not support
permission management.");
+ 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));
+ }
+
+ @Nullable
+ protected static PermissionColumns columns(
+ @Nullable String columnNames, @Nullable String
excludedColumnNames) {
+ List<String> included = parseColumnNames(columnNames);
+ List<String> excluded = parseColumnNames(excludedColumnNames);
+ return included == null && excluded == null
+ ? null
+ : new PermissionColumns(included, excluded);
+ }
+
+ protected static <E extends Enum<E>> E enumValue(
+ String value, Class<E> enumClass, String argument) {
+ checkArgument(!StringUtils.isNullOrWhitespaceOnly(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 String[] toArray(@Nullable List<String> values) {
+ return values == null ? null : values.toArray(new String[0]);
+ }
+
+ /**
+ * Parses a JSON array of column names. Each string is kept exactly,
including leading or
+ * trailing spaces. The value must be one complete JSON array; trailing
tokens are rejected so a
+ * second array cannot be dropped before the grant.
+ */
+ @Nullable
+ private static List<String> parseColumnNames(@Nullable String value) {
+ String normalized = emptyToNull(value);
+ if (normalized == null) {
+ return null;
+ }
+ JsonNode node;
+ try {
+ node =
+ JsonSerdeUtil.OBJECT_MAPPER_INSTANCE
+ .readerFor(JsonNode.class)
+
.with(DeserializationFeature.FAIL_ON_TRAILING_TOKENS)
+ .readValue(normalized);
+ } catch (IOException e) {
+ throw new IllegalArgumentException(COLUMN_NAMES_FORMAT_MESSAGE, e);
+ }
+ checkArgument(node.isArray(), COLUMN_NAMES_FORMAT_MESSAGE);
+ List<String> names = new ArrayList<>(node.size());
+ for (JsonNode element : node) {
+ checkArgument(element.isTextual(), "Column name must be a JSON
string.");
+ String name = element.asText();
+ checkArgument(!name.isEmpty(), "Column name cannot be empty.");
+ names.add(name);
+ }
+ return names;
+ }
+
+ @Nullable
+ protected static String emptyToNull(@Nullable String value) {
+ return StringUtils.isNullOrWhitespaceOnly(value) ? null : value;
+ }
+}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/GrantPermissionProcedure.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/GrantPermissionProcedure.java
new file mode 100644
index 0000000000..2cad636df6
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/GrantPermissionProcedure.java
@@ -0,0 +1,98 @@
+/*
+ * 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.flink.procedure;
+
+import org.apache.paimon.management.ResourceType;
+
+import org.apache.flink.table.annotation.ArgumentHint;
+import org.apache.flink.table.annotation.DataTypeHint;
+import org.apache.flink.table.annotation.ProcedureHint;
+import org.apache.flink.table.procedure.ProcedureContext;
+import org.apache.flink.types.Row;
+
+/**
+ * Grants a permission through a REST catalog. Usage:
+ *
+ * <pre><code>
+ * CALL sys.grant_permission(
+ * resource_type => 'TABLE',
+ * access => 'SELECT',
+ * principal => 'analyst',
+ * `database` => 'sales',
+ * `table` => 'orders')
+ * </code></pre>
+ */
+public class GrantPermissionProcedure extends BasePermissionProcedure {
+
+ public static final String IDENTIFIER = "grant_permission";
+
+ @ProcedureHint(
+ argument = {
+ @ArgumentHint(name = "resource_type", type =
@DataTypeHint("STRING")),
+ @ArgumentHint(name = "access", type = @DataTypeHint("STRING")),
+ @ArgumentHint(name = "principal", type =
@DataTypeHint("STRING")),
+ @ArgumentHint(name = "database", type =
@DataTypeHint("STRING"), isOptional = true),
+ @ArgumentHint(name = "table", type = @DataTypeHint("STRING"),
isOptional = true),
+ @ArgumentHint(name = "function", type =
@DataTypeHint("STRING"), isOptional = true),
+ @ArgumentHint(name = "view", type = @DataTypeHint("STRING"),
isOptional = true),
+ @ArgumentHint(
+ name = "expire_time",
+ type = @DataTypeHint("STRING"),
+ isOptional = true),
+ @ArgumentHint(
+ name = "column_names",
+ type = @DataTypeHint("STRING"),
+ isOptional = true),
+ @ArgumentHint(
+ name = "excluded_column_names",
+ type = @DataTypeHint("STRING"),
+ isOptional = true)
+ })
+ public @DataTypeHint("ROW<result BOOLEAN>") Row[] call(
+ ProcedureContext procedureContext,
+ String resourceType,
+ String access,
+ String principal,
+ String database,
+ String table,
+ String function,
+ String view,
+ String expireTime,
+ String columnNames,
+ String excludedColumnNames) {
+ permissionManagement()
+ .grantPermission(
+ assignment(
+ enumValue(resourceType, ResourceType.class,
"resource_type"),
+ access,
+ principal,
+ database,
+ table,
+ function,
+ view,
+ columns(columnNames, excludedColumnNames),
+ expireTime));
+ return new Row[] {Row.of(true)};
+ }
+
+ @Override
+ public String identifier() {
+ return IDENTIFIER;
+ }
+}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/ListPermissionsProcedure.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/ListPermissionsProcedure.java
new file mode 100644
index 0000000000..8a7cdb87d0
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/ListPermissionsProcedure.java
@@ -0,0 +1,128 @@
+/*
+ * 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.flink.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.flink.table.annotation.ArgumentHint;
+import org.apache.flink.table.annotation.DataTypeHint;
+import org.apache.flink.table.annotation.ProcedureHint;
+import org.apache.flink.table.procedure.ProcedureContext;
+import org.apache.flink.types.Row;
+
+import java.util.List;
+
+/**
+ * Lists direct permissions on an exact resource or explicit descendant scope.
Usage:
+ *
+ * <pre><code>
+ * CALL sys.list_permissions(
+ * resource_type => 'TABLE',
+ * `database` => 'sales',
+ * `table` => 'orders')
+ * </code></pre>
+ */
+public class ListPermissionsProcedure extends BasePermissionProcedure {
+
+ public static final String IDENTIFIER = "list_permissions";
+
+ private static final String OUTPUT_TYPE =
+ "ROW<resource_type STRING, `database` STRING, `table` STRING,
`function` STRING, "
+ + "`view` STRING, access STRING, principal STRING, "
+ + "column_names ARRAY<STRING>, excluded_column_names
ARRAY<STRING>, "
+ + "expire_time STRING, next_page_token STRING>";
+
+ @ProcedureHint(
+ argument = {
+ @ArgumentHint(name = "resource_type", type =
@DataTypeHint("STRING")),
+ @ArgumentHint(name = "database", type =
@DataTypeHint("STRING"), isOptional = true),
+ @ArgumentHint(name = "table", type = @DataTypeHint("STRING"),
isOptional = true),
+ @ArgumentHint(name = "function", type =
@DataTypeHint("STRING"), isOptional = true),
+ @ArgumentHint(name = "view", type = @DataTypeHint("STRING"),
isOptional = true),
+ @ArgumentHint(
+ name = "principal",
+ type = @DataTypeHint("STRING"),
+ isOptional = true),
+ @ArgumentHint(name = "access", type = @DataTypeHint("STRING"),
isOptional = true),
+ @ArgumentHint(name = "max_results", type =
@DataTypeHint("INT"), isOptional = true),
+ @ArgumentHint(
+ name = "page_token",
+ type = @DataTypeHint("STRING"),
+ isOptional = true)
+ })
+ public @DataTypeHint(OUTPUT_TYPE) Row[] call(
+ ProcedureContext procedureContext,
+ String resourceType,
+ String database,
+ String table,
+ String function,
+ String view,
+ String principal,
+ String access,
+ Integer maxResults,
+ String pageToken) {
+ PagedList<PermissionAssignment> page =
+ permissionManagement()
+ .listPermissions(
+ new ListPermissionsRequest(
+ enumValue(
+ resourceType,
ResourceType.class, "resource_type"),
+ emptyToNull(database),
+ emptyToNull(table),
+ emptyToNull(function),
+ emptyToNull(view),
+ emptyToNull(principal),
+ emptyToNull(access),
+ emptyToNull(pageToken),
+ maxResults));
+ List<PermissionAssignment> assignments = page.getElements();
+ if (assignments == null || assignments.isEmpty()) {
+ return new Row[0];
+ }
+
+ Row[] rows = new Row[assignments.size()];
+ for (int i = 0; i < assignments.size(); i++) {
+ PermissionAssignment assignment = assignments.get(i);
+ PermissionColumns columns = assignment.getColumns();
+ rows[i] =
+ Row.of(
+ assignment.getResource().getType().name(),
+ assignment.getResource().getDatabase(),
+ assignment.getResource().getTable(),
+ assignment.getResource().getFunction(),
+ assignment.getResource().getView(),
+ assignment.getAccess(),
+ assignment.getPrincipal(),
+ toArray(columns == null ? null :
columns.getColumnNames()),
+ toArray(columns == null ? null :
columns.getExcludedColumnNames()),
+ assignment.getExpireTime(),
+ page.getNextPageToken());
+ }
+ return rows;
+ }
+
+ @Override
+ public String identifier() {
+ return IDENTIFIER;
+ }
+}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/RevokePermissionProcedure.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/RevokePermissionProcedure.java
new file mode 100644
index 0000000000..57a219acf2
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/RevokePermissionProcedure.java
@@ -0,0 +1,81 @@
+/*
+ * 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.flink.procedure;
+
+import org.apache.paimon.management.ResourceType;
+
+import org.apache.flink.table.annotation.ArgumentHint;
+import org.apache.flink.table.annotation.DataTypeHint;
+import org.apache.flink.table.annotation.ProcedureHint;
+import org.apache.flink.table.procedure.ProcedureContext;
+import org.apache.flink.types.Row;
+
+/**
+ * Revokes a permission by its resource identity. Usage:
+ *
+ * <pre><code>
+ * CALL sys.revoke_permission(
+ * resource_type => 'TABLE',
+ * access => 'SELECT',
+ * principal => 'analyst',
+ * `database` => 'sales',
+ * `table` => 'orders')
+ * </code></pre>
+ */
+public class RevokePermissionProcedure extends BasePermissionProcedure {
+
+ public static final String IDENTIFIER = "revoke_permission";
+
+ @ProcedureHint(
+ argument = {
+ @ArgumentHint(name = "resource_type", type =
@DataTypeHint("STRING")),
+ @ArgumentHint(name = "access", type = @DataTypeHint("STRING")),
+ @ArgumentHint(name = "principal", type =
@DataTypeHint("STRING")),
+ @ArgumentHint(name = "database", type =
@DataTypeHint("STRING"), isOptional = true),
+ @ArgumentHint(name = "table", type = @DataTypeHint("STRING"),
isOptional = true),
+ @ArgumentHint(name = "function", type =
@DataTypeHint("STRING"), isOptional = true),
+ @ArgumentHint(name = "view", type = @DataTypeHint("STRING"),
isOptional = true)
+ })
+ public @DataTypeHint("ROW<result BOOLEAN>") Row[] call(
+ ProcedureContext procedureContext,
+ String resourceType,
+ String access,
+ String principal,
+ String database,
+ String table,
+ String function,
+ String view) {
+ permissionManagement()
+ .revokePermission(
+ resource(
+ enumValue(resourceType, ResourceType.class,
"resource_type"),
+ database,
+ table,
+ function,
+ view),
+ access,
+ principal);
+ return new Row[] {Row.of(true)};
+ }
+
+ @Override
+ public String identifier() {
+ return IDENTIFIER;
+ }
+}
diff --git
a/paimon-flink/paimon-flink-common/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory
b/paimon-flink/paimon-flink-common/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory
index b9223d0334..d9e2e8e1bf 100644
---
a/paimon-flink/paimon-flink-common/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory
+++
b/paimon-flink/paimon-flink-common/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory
@@ -112,3 +112,6 @@ org.apache.paimon.flink.procedure.VectorSearchProcedure
org.apache.paimon.flink.procedure.FullTextSearchProcedure
org.apache.paimon.flink.procedure.DropGlobalIndexProcedure
org.apache.paimon.flink.procedure.ImportFilesProcedure
+org.apache.paimon.flink.procedure.GrantPermissionProcedure
+org.apache.paimon.flink.procedure.RevokePermissionProcedure
+org.apache.paimon.flink.procedure.ListPermissionsProcedure
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/PermissionProcedureITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/PermissionProcedureITCase.java
new file mode 100644
index 0000000000..f9208a2621
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/PermissionProcedureITCase.java
@@ -0,0 +1,428 @@
+/*
+ * 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.flink.procedure;
+
+import org.apache.paimon.flink.RESTCatalogITCaseBase;
+
+import org.apache.flink.types.Row;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** IT cases for REST permission procedures. */
+public class PermissionProcedureITCase extends RESTCatalogITCaseBase {
+
+ @BeforeEach
+ @Override
+ public void before() throws IOException {
+ super.before();
+ restCatalogServer.registerManagementPrincipal("analyst");
+ restCatalogServer.registerManagementPrincipal("first");
+ restCatalogServer.registerManagementPrincipal("second");
+ }
+
+ @Test
+ public void testGrantListAndRevoke() {
+ assertThat(grantTable("select",
"2027-01-01T00:00:00Z")).containsExactly(Row.of(true));
+ assertThat(grantTable("SELECT",
"2028-01-01T00:00:00Z")).containsExactly(Row.of(true));
+
+ List<Row> listed =
+ sql(
+ "CALL sys.list_permissions("
+ + "resource_type => 'TABLE', "
+ + "`database` => '%s', "
+ + "`table` => '%s', "
+ + "principal => 'analyst')",
+ DATABASE_NAME, TABLE_NAME);
+ assertThat(listed).hasSize(1);
+ Row assignment = listed.get(0);
+ assertThat(assignment.getField(0)).isEqualTo("TABLE");
+ assertThat(assignment.getField(1)).isEqualTo(DATABASE_NAME);
+ assertThat(assignment.getField(2)).isEqualTo(TABLE_NAME);
+ assertThat(assignment.getField(3)).isNull();
+ assertThat(assignment.getField(4)).isNull();
+ assertThat(assignment.getField(5)).isEqualTo("SELECT");
+ assertThat(assignment.getField(6)).isEqualTo("analyst");
+ assertThat(assignment.getField(7)).isNull();
+ assertThat(assignment.getField(8)).isNull();
+ assertThat(assignment.getField(9)).isEqualTo("2028-01-01T00:00:00Z");
+ assertThat(assignment.getField(10)).isNull();
+
+ String revoke =
+ String.format(
+ "CALL sys.revoke_permission("
+ + "resource_type => 'TABLE', "
+ + "access => 'SELECT', "
+ + "principal => 'analyst', "
+ + "`database` => '%s', "
+ + "`table` => '%s')",
+ DATABASE_NAME, TABLE_NAME);
+ assertThat(sql(revoke)).containsExactly(Row.of(true));
+ assertThat(sql(revoke)).containsExactly(Row.of(true));
+ assertThat(
+ sql(
+ "CALL sys.list_permissions("
+ + "resource_type => 'TABLE', "
+ + "`database` => '%s', "
+ + "`table` => '%s', "
+ + "principal => 'analyst')",
+ DATABASE_NAME, TABLE_NAME))
+ .isEmpty();
+ }
+
+ @Test
+ public void testListPermissionsPagination() {
+ grantCatalog("first");
+ grantCatalog("second");
+
+ Row first =
+ sql("CALL sys.list_permissions(resource_type => 'CATALOG',
max_results => 1)")
+ .get(0);
+ assertThat(first.getField(6)).isEqualTo("first");
+ String pageToken = (String) first.getField(10);
+ assertThat(pageToken).isNotEmpty().isNotEqualTo("1");
+
+ sql(
+ "CALL sys.revoke_permission("
+ + "resource_type => 'CATALOG', "
+ + "access => 'CREATEDATABASE', "
+ + "principal => 'first')");
+
+ Row second =
+ sql(
+ "CALL sys.list_permissions("
+ + "resource_type => 'CATALOG', "
+ + "max_results => 1, "
+ + "page_token => '%s')",
+ pageToken)
+ .get(0);
+ assertThat(second.getField(6)).isEqualTo("second");
+ assertThat(second.getField(10)).isNull();
+ }
+
+ @Test
+ public void testDescendantScopes() {
+ assertThat(
+ sql(
+ "CALL sys.grant_permission("
+ + "resource_type => 'CATALOG_ALL', "
+ + "access => 'SELECT', "
+ + "principal => 'analyst')"))
+ .containsExactly(Row.of(true));
+ Row catalogAll =
+ sql("CALL sys.list_permissions("
+ + "resource_type => 'CATALOG_ALL', "
+ + "principal => 'analyst')")
+ .get(0);
+ assertThat(catalogAll.getField(0)).isEqualTo("CATALOG_ALL");
+ assertThat(catalogAll.getField(1)).isNull();
+ assertThat(catalogAll.getField(5)).isEqualTo("SELECT");
+
+ assertThat(
+ sql(
+ "CALL sys.grant_permission("
+ + "resource_type => 'DATABASE_ALL', "
+ + "`database` => '%s', "
+ + "access => 'UPDATE', "
+ + "principal => 'analyst')",
+ DATABASE_NAME))
+ .containsExactly(Row.of(true));
+ Row databaseAll =
+ sql(
+ "CALL sys.list_permissions("
+ + "resource_type => 'DATABASE_ALL', "
+ + "`database` => '%s', "
+ + "principal => 'analyst')",
+ DATABASE_NAME)
+ .get(0);
+ assertThat(databaseAll.getField(0)).isEqualTo("DATABASE_ALL");
+ assertThat(databaseAll.getField(1)).isEqualTo(DATABASE_NAME);
+ assertThat(databaseAll.getField(2)).isNull();
+ assertThat(databaseAll.getField(5)).isEqualTo("UPDATE");
+ }
+
+ @Test
+ public void testColumnPermissions() {
+ sql("ALTER TABLE %s.%s SET ('query-auth.enabled' = 'true')",
DATABASE_NAME, TABLE_NAME);
+ assertThat(
+ sql(
+ "CALL sys.grant_permission("
+ + "resource_type => 'COLUMN', "
+ + "access => 'SELECT', "
+ + "principal => 'analyst', "
+ + "`database` => '%s', "
+ + "`table` => '%s', "
+ + "column_names => '[\"a\"]')",
+ DATABASE_NAME, TABLE_NAME))
+ .containsExactly(Row.of(true));
+
+ Row included =
+ sql(
+ "CALL sys.list_permissions("
+ + "resource_type => 'COLUMN', "
+ + "`database` => '%s', "
+ + "`table` => '%s', "
+ + "principal => 'analyst')",
+ DATABASE_NAME, TABLE_NAME)
+ .get(0);
+ assertThat(stringArray(included.getField(7))).containsExactly("a");
+ assertThat(included.getField(8)).isNull();
+
+ assertThat(
+ sql(
+ "CALL sys.grant_permission("
+ + "resource_type => 'COLUMN', "
+ + "access => 'SELECT', "
+ + "principal => 'analyst', "
+ + "`database` => '%s', "
+ + "`table` => '%s', "
+ + "excluded_column_names =>
'[\"b\"]')",
+ DATABASE_NAME, TABLE_NAME))
+ .containsExactly(Row.of(true));
+ Row excluded =
+ sql(
+ "CALL sys.list_permissions("
+ + "resource_type => 'COLUMN', "
+ + "`database` => '%s', "
+ + "`table` => '%s', "
+ + "principal => 'analyst')",
+ DATABASE_NAME, TABLE_NAME)
+ .get(0);
+ assertThat(excluded.getField(7)).isNull();
+ assertThat(stringArray(excluded.getField(8))).containsExactly("b");
+ }
+
+ @Test
+ public void testPositionalGrant() {
+ assertThat(
+ sql(
+ "CALL sys.grant_permission("
+ + "'TABLE', 'SELECT', 'analyst', '%s',
'%s', '', '', "
+ + "'2027-01-01T00:00:00Z')",
+ DATABASE_NAME, TABLE_NAME))
+ .containsExactly(Row.of(true));
+ Row assignment = listedTable().get(0);
+ assertThat(assignment.getField(1)).isEqualTo(DATABASE_NAME);
+ assertThat(assignment.getField(2)).isEqualTo(TABLE_NAME);
+ assertThat(assignment.getField(5)).isEqualTo("SELECT");
+ assertThat(assignment.getField(6)).isEqualTo("analyst");
+ assertThat(assignment.getField(9)).isEqualTo("2027-01-01T00:00:00Z");
+
+ sql("ALTER TABLE %s.%s SET ('query-auth.enabled' = 'true')",
DATABASE_NAME, TABLE_NAME);
+ assertThat(
+ sql(
+ "CALL sys.grant_permission("
+ + "'COLUMN', 'SELECT', 'analyst',
'%s', '%s', '', '', '', "
+ + "'[\"a\"]', '')",
+ DATABASE_NAME, TABLE_NAME))
+ .containsExactly(Row.of(true));
+ Row columns =
+ sql(
+ "CALL sys.list_permissions("
+ + "resource_type => 'COLUMN', "
+ + "`database` => '%s', "
+ + "`table` => '%s', "
+ + "principal => 'analyst')",
+ DATABASE_NAME, TABLE_NAME)
+ .get(0);
+ assertThat(stringArray(columns.getField(7))).containsExactly("a");
+ assertThat(columns.getField(8)).isNull();
+ }
+
+ @Test
+ public void testColumnNameIdentity() {
+ sql(
+ "CREATE TABLE %s.space_columns (`secret` STRING, ` secret `
STRING) "
+ + "WITH ('query-auth.enabled' = 'true')",
+ DATABASE_NAME);
+ assertThat(
+ sql(
+ "CALL sys.grant_permission("
+ + "resource_type => 'COLUMN', "
+ + "access => 'SELECT', "
+ + "principal => 'analyst', "
+ + "`database` => '%s', "
+ + "`table` => 'space_columns', "
+ + "excluded_column_names => '[\"
secret \"]')",
+ DATABASE_NAME))
+ .containsExactly(Row.of(true));
+ Row excluded =
+ sql(
+ "CALL sys.list_permissions("
+ + "resource_type => 'COLUMN', "
+ + "`database` => '%s', "
+ + "`table` => 'space_columns', "
+ + "principal => 'analyst')",
+ DATABASE_NAME)
+ .get(0);
+ assertThat(excluded.getField(7)).isNull();
+ assertThat(stringArray(excluded.getField(8))).containsExactly(" secret
");
+
+ assertThat(
+ sql(
+ "CALL sys.grant_permission("
+ + "resource_type => 'COLUMN', "
+ + "access => 'SELECT', "
+ + "principal => 'first', "
+ + "`database` => '%s', "
+ + "`table` => 'space_columns', "
+ + "column_names => '[\" secret \",
\"secret\"]')",
+ DATABASE_NAME))
+ .containsExactly(Row.of(true));
+ Row included =
+ sql(
+ "CALL sys.list_permissions("
+ + "resource_type => 'COLUMN', "
+ + "`database` => '%s', "
+ + "`table` => 'space_columns', "
+ + "principal => 'first')",
+ DATABASE_NAME)
+ .get(0);
+ assertThat(stringArray(included.getField(7))).containsExactly(" secret
", "secret");
+ assertThat(included.getField(8)).isNull();
+
+ assertThatThrownBy(
+ () ->
+ sql(
+ "CALL sys.grant_permission("
+ + "resource_type => 'COLUMN', "
+ + "access => 'SELECT', "
+ + "principal => 'second', "
+ + "`database` => '%s', "
+ + "`table` => 'space_columns',
"
+ + "excluded_column_names => '
secret ')",
+ DATABASE_NAME))
+ .hasMessageContaining("JSON array of strings");
+ assertThatThrownBy(
+ () ->
+ sql(
+ "CALL sys.grant_permission("
+ + "resource_type => 'COLUMN', "
+ + "access => 'SELECT', "
+ + "principal => 'second', "
+ + "`database` => '%s', "
+ + "`table` => 'space_columns',
"
+ + "column_names =>
'[\"a\"],[\"b\"]')",
+ DATABASE_NAME))
+ .hasMessageContaining("JSON array of strings");
+ assertThat(
+ sql(
+ "CALL sys.list_permissions("
+ + "resource_type => 'COLUMN', "
+ + "`database` => '%s', "
+ + "`table` => 'space_columns', "
+ + "principal => 'second')",
+ DATABASE_NAME))
+ .isEmpty();
+ }
+
+ @Test
+ public void testRejectsInvalidArguments() {
+ assertThatThrownBy(
+ () ->
+ sql(
+ "CALL sys.grant_permission("
+ + "resource_type => 'NOPE', "
+ + "access => 'SELECT', "
+ + "principal => 'analyst')"))
+ .hasMessageContaining("Invalid resource_type 'NOPE'");
+ assertThatThrownBy(
+ () ->
+ sql(
+ "CALL sys.grant_permission("
+ + "resource_type => 'CATALOG',
"
+ + "access => 'CREATEDATABASE',
"
+ + "principal => '')"))
+ .hasMessageContaining("principal cannot be empty.");
+ assertThatThrownBy(
+ () ->
+ sql(
+ "CALL sys.grant_permission("
+ + "resource_type => 'COLUMN', "
+ + "access => 'SELECT', "
+ + "principal => 'analyst', "
+ + "`database` => '%s', "
+ + "`table` => '%s', "
+ + "column_names => '[\"a\"]', "
+ + "excluded_column_names =>
'[\"b\"]')",
+ DATABASE_NAME, TABLE_NAME))
+ .hasMessageContaining(
+ "columns must contain exactly one of columnNames or
excludedColumnNames.");
+ assertThatThrownBy(
+ () ->
+ sql(
+ "CALL sys.grant_permission("
+ + "resource_type => 'TABLE', "
+ + "access => 'SELECT', "
+ + "principal => 'analyst', "
+ + "`database` => '%s', "
+ + "`table` => '%s', "
+ + "column_names => '[\"a\"]')",
+ DATABASE_NAME, TABLE_NAME))
+ .hasMessageContaining("columns is only valid for COLUMN
resource.");
+ }
+
+ private List<Row> listedTable() {
+ return sql(
+ "CALL sys.list_permissions("
+ + "resource_type => 'TABLE', "
+ + "`database` => '%s', "
+ + "`table` => '%s', "
+ + "principal => 'analyst')",
+ DATABASE_NAME, TABLE_NAME);
+ }
+
+ private List<Row> grantTable(String access, String expireTime) {
+ return sql(
+ "CALL sys.grant_permission("
+ + "resource_type => 'table', "
+ + "access => '%s', "
+ + "principal => 'analyst', "
+ + "`database` => '%s', "
+ + "`table` => '%s', "
+ + "expire_time => '%s')",
+ access, DATABASE_NAME, TABLE_NAME, expireTime);
+ }
+
+ private void grantCatalog(String principal) {
+ sql(
+ "CALL sys.grant_permission("
+ + "resource_type => 'CATALOG', "
+ + "access => 'CREATEDATABASE', "
+ + "principal => '%s')",
+ principal);
+ }
+
+ private static String[] stringArray(Object value) {
+ if (value instanceof String[]) {
+ return (String[]) value;
+ }
+ Object[] objects = (Object[]) value;
+ String[] strings = new String[objects.length];
+ for (int i = 0; i < objects.length; i++) {
+ strings[i] = String.valueOf(objects[i]);
+ }
+ return strings;
+ }
+}
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/PermissionProcedureUnsupportedCatalogITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/PermissionProcedureUnsupportedCatalogITCase.java
new file mode 100644
index 0000000000..be620f855c
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/PermissionProcedureUnsupportedCatalogITCase.java
@@ -0,0 +1,41 @@
+/*
+ * 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.flink.procedure;
+
+import org.apache.paimon.flink.CatalogITCaseBase;
+
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Permission procedures reject catalogs that are not REST catalogs. */
+public class PermissionProcedureUnsupportedCatalogITCase extends
CatalogITCaseBase {
+
+ @Test
+ public void testFilesystemCatalogRejectsPermissionProcedure() {
+ assertThatThrownBy(
+ () ->
+ sql(
+ "CALL sys.grant_permission("
+ + "resource_type => 'CATALOG',
"
+ + "access => 'CREATEDATABASE',
"
+ + "principal => 'admin')"))
+ .hasMessageContaining("Catalog does not support permission
management.");
+ }
+}
