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 28a2c01a65 [api] Add REST policy management (#9400)
28a2c01a65 is described below
commit 28a2c01a65bb3246a08d840c11f3c27a264f7cd9
Author: Jingsong Lee <[email protected]>
AuthorDate: Wed Aug 26 15:47:36 2026 +0800
[api] Add REST policy management (#9400)
---
.../org/apache/paimon/management/ColumnMask.java | 77 +++++++
.../org/apache/paimon/management/DataPolicy.java | 120 ++++++++++
.../paimon/management/ListPoliciesRequest.java | 102 +++++++++
.../paimon/management/PermissionResource.java | 7 +
.../apache/paimon/management/PolicyManagement.java | 70 ++++++
.../org/apache/paimon/management/PolicyType.java | 37 +++
.../org/apache/paimon/management/RowFilter.java | 64 ++++++
.../main/java/org/apache/paimon/rest/RESTApi.java | 58 +++++
.../apache/paimon/rest/RESTPolicyManagement.java | 72 ++++++
.../java/org/apache/paimon/rest/ResourcePaths.java | 15 ++
.../paimon/rest/requests/DropPolicyRequest.java | 99 ++++++++
.../apache/paimon/rest/requests/PolicyRequest.java | 96 ++++++++
.../paimon/rest/responses/ErrorResponse.java | 2 +
.../rest/responses/ListPoliciesResponse.java | 72 ++++++
.../management/PolicyManagementJsonTest.java | 176 +++++++++++++++
.../paimon/rest/RESTPolicyManagementTest.java | 251 +++++++++++++++++++++
.../requests/RequestJacksonCompatibilityTest.java | 2 +
.../java/org/apache/paimon/rest/RESTCatalog.java | 6 +
.../org/apache/paimon/rest/ResourcePathsTest.java | 24 ++
19 files changed, 1350 insertions(+)
diff --git
a/paimon-api/src/main/java/org/apache/paimon/management/ColumnMask.java
b/paimon-api/src/main/java/org/apache/paimon/management/ColumnMask.java
new file mode 100644
index 0000000000..c2dbadbbdb
--- /dev/null
+++ b/paimon-api/src/main/java/org/apache/paimon/management/ColumnMask.java
@@ -0,0 +1,77 @@
+/*
+ * 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.management;
+
+import org.apache.paimon.annotation.Experimental;
+
+import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonGetter;
+import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
+import java.beans.ConstructorProperties;
+import java.nio.charset.StandardCharsets;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/** Protected column and serialized Paimon transform for a column-mask policy.
*/
+@Experimental
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class ColumnMask {
+
+ public static final int MAX_TRANSFORM_BYTES = 60 * 1024;
+
+ private static final String FIELD_ON_COLUMN = "onColumn";
+ private static final String FIELD_TRANSFORM = "transform";
+
+ @JsonProperty(FIELD_ON_COLUMN)
+ private final String onColumn;
+
+ @JsonProperty(FIELD_TRANSFORM)
+ private final String transform;
+
+ @JsonCreator
+ @ConstructorProperties({FIELD_ON_COLUMN, FIELD_TRANSFORM})
+ public ColumnMask(
+ @JsonProperty(FIELD_ON_COLUMN) String onColumn,
+ @JsonProperty(FIELD_TRANSFORM) String transform) {
+ checkArgument(!isBlank(onColumn), "onColumn cannot be empty.");
+ checkArgument(!isBlank(transform), "transform cannot be empty.");
+ checkArgument(
+ transform.getBytes(StandardCharsets.UTF_8).length <=
MAX_TRANSFORM_BYTES,
+ "transform must not exceed %s UTF-8 bytes.",
+ MAX_TRANSFORM_BYTES);
+ this.onColumn = onColumn;
+ this.transform = transform;
+ }
+
+ @JsonGetter(FIELD_ON_COLUMN)
+ public String getOnColumn() {
+ return onColumn;
+ }
+
+ @JsonGetter(FIELD_TRANSFORM)
+ public String getTransform() {
+ return transform;
+ }
+
+ private static boolean isBlank(String value) {
+ return value == null || value.trim().isEmpty();
+ }
+}
diff --git
a/paimon-api/src/main/java/org/apache/paimon/management/DataPolicy.java
b/paimon-api/src/main/java/org/apache/paimon/management/DataPolicy.java
new file mode 100644
index 0000000000..08fe3fb035
--- /dev/null
+++ b/paimon-api/src/main/java/org/apache/paimon/management/DataPolicy.java
@@ -0,0 +1,120 @@
+/*
+ * 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.management;
+
+import org.apache.paimon.annotation.Experimental;
+
+import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonGetter;
+import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonInclude;
+import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
+import javax.annotation.Nullable;
+
+import java.beans.ConstructorProperties;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+import static org.apache.paimon.utils.Preconditions.checkNotNull;
+
+/**
+ * Principal-scoped row-filter or column-mask policy attached to one table.
+ *
+ * <p>A principal has at most one row filter per table and at most one mask
per table column. When
+ * policies are enforced, applicable row filters are combined with logical AND
and multiple
+ * effective masks for one column fail closed. Invalid predicates or
transforms also fail closed.
+ */
+@Experimental
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class DataPolicy {
+
+ private static final String FIELD_RESOURCE = "resource";
+ private static final String FIELD_ROW_FILTER = "rowFilter";
+ private static final String FIELD_COLUMN_MASK = "columnMask";
+ private static final String FIELD_PRINCIPAL = "principal";
+
+ @JsonProperty(FIELD_RESOURCE)
+ private final PermissionResource resource;
+
+ @Nullable
+ @JsonProperty(FIELD_ROW_FILTER)
+ @JsonInclude(JsonInclude.Include.NON_NULL)
+ private final RowFilter rowFilter;
+
+ @Nullable
+ @JsonProperty(FIELD_COLUMN_MASK)
+ @JsonInclude(JsonInclude.Include.NON_NULL)
+ private final ColumnMask columnMask;
+
+ @JsonProperty(FIELD_PRINCIPAL)
+ private final String principal;
+
+ @JsonCreator
+ @ConstructorProperties({FIELD_RESOURCE, FIELD_ROW_FILTER,
FIELD_COLUMN_MASK, FIELD_PRINCIPAL})
+ public DataPolicy(
+ @JsonProperty(FIELD_RESOURCE) PermissionResource resource,
+ @Nullable @JsonProperty(FIELD_ROW_FILTER) RowFilter rowFilter,
+ @Nullable @JsonProperty(FIELD_COLUMN_MASK) ColumnMask columnMask,
+ @JsonProperty(FIELD_PRINCIPAL) String principal) {
+ this.resource = checkNotNull(resource, "resource cannot be null");
+ resource.validatePolicyAttachment();
+ checkArgument(
+ (rowFilter == null) != (columnMask == null),
+ "A policy must contain exactly one of rowFilter and
columnMask.");
+ this.rowFilter = rowFilter;
+ this.columnMask = columnMask;
+ this.principal = PermissionAssignment.validatePrincipal(principal);
+ }
+
+ public static DataPolicy rowFilter(
+ PermissionResource resource, RowFilter rowFilter, String
principal) {
+ return new DataPolicy(resource, rowFilter, null, principal);
+ }
+
+ public static DataPolicy columnMask(
+ PermissionResource resource, ColumnMask columnMask, String
principal) {
+ return new DataPolicy(resource, null, columnMask, principal);
+ }
+
+ @JsonGetter(FIELD_RESOURCE)
+ public PermissionResource getResource() {
+ return resource;
+ }
+
+ @Nullable
+ @JsonGetter(FIELD_ROW_FILTER)
+ public RowFilter getRowFilter() {
+ return rowFilter;
+ }
+
+ @Nullable
+ @JsonGetter(FIELD_COLUMN_MASK)
+ public ColumnMask getColumnMask() {
+ return columnMask;
+ }
+
+ public PolicyType type() {
+ return rowFilter == null ? PolicyType.COLUMN_MASKING :
PolicyType.ROW_FILTER;
+ }
+
+ @JsonGetter(FIELD_PRINCIPAL)
+ public String getPrincipal() {
+ return principal;
+ }
+}
diff --git
a/paimon-api/src/main/java/org/apache/paimon/management/ListPoliciesRequest.java
b/paimon-api/src/main/java/org/apache/paimon/management/ListPoliciesRequest.java
new file mode 100644
index 0000000000..e79186be3d
--- /dev/null
+++
b/paimon-api/src/main/java/org/apache/paimon/management/ListPoliciesRequest.java
@@ -0,0 +1,102 @@
+/*
+ * 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.management;
+
+import org.apache.paimon.annotation.Experimental;
+
+import javax.annotation.Nullable;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+import static org.apache.paimon.utils.Preconditions.checkNotNull;
+
+/** Filters for listing policies attached to an exact table resource. */
+@Experimental
+public class ListPoliciesRequest {
+
+ private final PermissionResource resource;
+ @Nullable private final PolicyType type;
+ @Nullable private final String principal;
+ @Nullable private final String column;
+ @Nullable private final String pageToken;
+ @Nullable private final Integer maxResults;
+
+ public ListPoliciesRequest(
+ PermissionResource resource,
+ @Nullable PolicyType type,
+ @Nullable String principal,
+ @Nullable String column,
+ @Nullable String pageToken,
+ @Nullable Integer maxResults) {
+ this.resource = checkNotNull(resource, "resource cannot be null");
+ resource.validatePolicyAttachment();
+ if (!isBlank(principal)) {
+ PermissionAssignment.validatePrincipal(principal);
+ }
+ checkArgument(maxResults == null || maxResults > 0, "maxResults must
be greater than 0.");
+ checkArgument(
+ maxResults == null || maxResults <=
ListPermissionsRequest.MAX_PAGE_SIZE,
+ "maxResults must be at most %s.",
+ ListPermissionsRequest.MAX_PAGE_SIZE);
+ this.type = type;
+ this.principal = isBlank(principal) ? null : principal;
+ checkArgument(
+ isBlank(column) || type == PolicyType.COLUMN_MASKING,
+ "column filter requires type COLUMN_MASKING.");
+ this.column = isBlank(column) ? null : column;
+ this.pageToken = pageToken;
+ this.maxResults = maxResults;
+ }
+
+ public PermissionResource getResource() {
+ return resource;
+ }
+
+ @Nullable
+ public PolicyType getType() {
+ return type;
+ }
+
+ @Nullable
+ public String getPrincipal() {
+ return principal;
+ }
+
+ @Nullable
+ public String getColumn() {
+ return column;
+ }
+
+ @Nullable
+ public String getPageToken() {
+ return pageToken;
+ }
+
+ @Nullable
+ public Integer getMaxResults() {
+ return maxResults;
+ }
+
+ public ListPoliciesRequest withPageToken(@Nullable String newPageToken) {
+ return new ListPoliciesRequest(resource, type, principal, column,
newPageToken, maxResults);
+ }
+
+ private static boolean isBlank(@Nullable String value) {
+ return value == null || value.trim().isEmpty();
+ }
+}
diff --git
a/paimon-api/src/main/java/org/apache/paimon/management/PermissionResource.java
b/paimon-api/src/main/java/org/apache/paimon/management/PermissionResource.java
index f0100af100..b3d15373b8 100644
---
a/paimon-api/src/main/java/org/apache/paimon/management/PermissionResource.java
+++
b/paimon-api/src/main/java/org/apache/paimon/management/PermissionResource.java
@@ -122,6 +122,13 @@ public class PermissionResource {
return view;
}
+ /** Validates that this resource can carry a data policy in the current
contract. */
+ public void validatePolicyAttachment() {
+ checkArgument(
+ type == ResourceType.TABLE,
+ "Policies can currently be attached only to TABLE resources.");
+ }
+
@Override
public boolean equals(Object o) {
if (this == o) {
diff --git
a/paimon-api/src/main/java/org/apache/paimon/management/PolicyManagement.java
b/paimon-api/src/main/java/org/apache/paimon/management/PolicyManagement.java
new file mode 100644
index 0000000000..6b33195b06
--- /dev/null
+++
b/paimon-api/src/main/java/org/apache/paimon/management/PolicyManagement.java
@@ -0,0 +1,70 @@
+/*
+ * 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.management;
+
+import org.apache.paimon.PagedList;
+import org.apache.paimon.annotation.Experimental;
+
+import javax.annotation.Nullable;
+
+/** Control-plane contract for row-filter and column-masking policies. */
+@Experimental
+public interface PolicyManagement {
+
+ PagedList<DataPolicy> listPolicies(ListPoliciesRequest request);
+
+ void createPolicy(DataPolicy policy) throws PolicyAlreadyExistException;
+
+ void dropPolicy(
+ PermissionResource resource,
+ PolicyType type,
+ String principal,
+ @Nullable String column,
+ boolean ignoreIfNotExists);
+
+ /** Exception for trying to create a policy that already exists. */
+ class PolicyAlreadyExistException extends Exception {
+
+ private final DataPolicy policy;
+
+ public PolicyAlreadyExistException(DataPolicy policy) {
+ this(policy, null);
+ }
+
+ public PolicyAlreadyExistException(DataPolicy policy, Throwable cause)
{
+ super(message(policy), cause);
+ this.policy = policy;
+ }
+
+ public DataPolicy policy() {
+ return policy;
+ }
+
+ private static String message(DataPolicy policy) {
+ String target = policy.type().name();
+ if (policy.getColumnMask() != null) {
+ target += "(" + policy.getColumnMask().getOnColumn() + ")";
+ }
+ PermissionResource resource = policy.getResource();
+ return String.format(
+ "%s policy for principal '%s' already exists on table
'%s.%s'.",
+ target, policy.getPrincipal(), resource.getDatabase(),
resource.getTable());
+ }
+ }
+}
diff --git
a/paimon-api/src/main/java/org/apache/paimon/management/PolicyType.java
b/paimon-api/src/main/java/org/apache/paimon/management/PolicyType.java
new file mode 100644
index 0000000000..1312a2632f
--- /dev/null
+++ b/paimon-api/src/main/java/org/apache/paimon/management/PolicyType.java
@@ -0,0 +1,37 @@
+/*
+ * 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.management;
+
+import org.apache.paimon.annotation.Experimental;
+
+import javax.annotation.Nullable;
+
+import java.util.Locale;
+
+/** Fine-grained data policy types. */
+@Experimental
+public enum PolicyType {
+ ROW_FILTER,
+ COLUMN_MASKING;
+
+ @Nullable
+ public static PolicyType fromString(@Nullable String value) {
+ return value == null ? null : valueOf(value.toUpperCase(Locale.ROOT));
+ }
+}
diff --git
a/paimon-api/src/main/java/org/apache/paimon/management/RowFilter.java
b/paimon-api/src/main/java/org/apache/paimon/management/RowFilter.java
new file mode 100644
index 0000000000..4af32042b5
--- /dev/null
+++ b/paimon-api/src/main/java/org/apache/paimon/management/RowFilter.java
@@ -0,0 +1,64 @@
+/*
+ * 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.management;
+
+import org.apache.paimon.annotation.Experimental;
+
+import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonGetter;
+import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
+import java.beans.ConstructorProperties;
+import java.nio.charset.StandardCharsets;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/** Serialized Paimon predicate for a row-filter policy. */
+@Experimental
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class RowFilter {
+
+ public static final int MAX_PREDICATE_BYTES = 60 * 1024;
+
+ private static final String FIELD_PREDICATE = "predicate";
+
+ @JsonProperty(FIELD_PREDICATE)
+ private final String predicate;
+
+ @JsonCreator
+ @ConstructorProperties({FIELD_PREDICATE})
+ public RowFilter(@JsonProperty(FIELD_PREDICATE) String predicate) {
+ checkArgument(!isBlank(predicate), "predicate cannot be empty.");
+ checkArgument(
+ predicate.getBytes(StandardCharsets.UTF_8).length <=
MAX_PREDICATE_BYTES,
+ "predicate must not exceed %s UTF-8 bytes.",
+ MAX_PREDICATE_BYTES);
+ this.predicate = predicate;
+ }
+
+ @JsonGetter(FIELD_PREDICATE)
+ public String getPredicate() {
+ return predicate;
+ }
+
+ private static boolean isBlank(String value) {
+ return value == null || value.trim().isEmpty();
+ }
+}
diff --git a/paimon-api/src/main/java/org/apache/paimon/rest/RESTApi.java
b/paimon-api/src/main/java/org/apache/paimon/rest/RESTApi.java
index e2f5cc3702..b4b110cfec 100644
--- a/paimon-api/src/main/java/org/apache/paimon/rest/RESTApi.java
+++ b/paimon-api/src/main/java/org/apache/paimon/rest/RESTApi.java
@@ -26,9 +26,12 @@ import org.apache.paimon.annotation.VisibleForTesting;
import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.consumer.ConsumerInfo;
import org.apache.paimon.function.FunctionChange;
+import org.apache.paimon.management.DataPolicy;
import org.apache.paimon.management.ListPermissionsRequest;
+import org.apache.paimon.management.ListPoliciesRequest;
import org.apache.paimon.management.PermissionAssignment;
import org.apache.paimon.management.PermissionResource;
+import org.apache.paimon.management.PolicyType;
import org.apache.paimon.options.Options;
import org.apache.paimon.partition.Partition;
import org.apache.paimon.partition.PartitionStatistics;
@@ -51,11 +54,13 @@ import org.apache.paimon.rest.requests.CreateTableRequest;
import org.apache.paimon.rest.requests.CreateTagRequest;
import org.apache.paimon.rest.requests.CreateViewRequest;
import org.apache.paimon.rest.requests.DropPartitionsRequest;
+import org.apache.paimon.rest.requests.DropPolicyRequest;
import org.apache.paimon.rest.requests.ForwardBranchRequest;
import org.apache.paimon.rest.requests.GrantPermissionRequest;
import org.apache.paimon.rest.requests.ListPartitionsByFilterRequest;
import org.apache.paimon.rest.requests.ListPartitionsByNamesRequest;
import org.apache.paimon.rest.requests.MarkDonePartitionsRequest;
+import org.apache.paimon.rest.requests.PolicyRequest;
import org.apache.paimon.rest.requests.RegisterTableRequest;
import org.apache.paimon.rest.requests.RenameTableRequest;
import org.apache.paimon.rest.requests.ReplaceTableRequest;
@@ -86,6 +91,7 @@ import
org.apache.paimon.rest.responses.ListFunctionsGloballyResponse;
import org.apache.paimon.rest.responses.ListFunctionsResponse;
import org.apache.paimon.rest.responses.ListPartitionsResponse;
import org.apache.paimon.rest.responses.ListPermissionsResponse;
+import org.apache.paimon.rest.responses.ListPoliciesResponse;
import org.apache.paimon.rest.responses.ListSnapshotsResponse;
import org.apache.paimon.rest.responses.ListTableDetailsResponse;
import org.apache.paimon.rest.responses.ListTablesGloballyResponse;
@@ -125,6 +131,7 @@ import static
org.apache.paimon.rest.RESTFunctionValidator.isValidFunctionName;
import static org.apache.paimon.rest.RESTUtil.extractPrefixMap;
import static
org.apache.paimon.rest.auth.AuthProviderFactory.createAuthProvider;
import static org.apache.paimon.utils.Preconditions.checkArgument;
+import static org.apache.paimon.utils.Preconditions.checkNotNull;
/**
* REST API for REST Catalog.
@@ -880,6 +887,57 @@ public class RESTApi {
restAuthFunction);
}
+ /** Lists policies attached to an exact table resource. */
+ @Experimental
+ public ListPoliciesResponse listPolicies(ListPoliciesRequest request) {
+ Map<String, String> queryParams = Maps.newHashMap();
+ if (request.getType() != null) {
+ putQueryParameter(queryParams, "type", request.getType().name());
+ }
+ putQueryParameter(queryParams, "principal", request.getPrincipal());
+ putQueryParameter(queryParams, "column", request.getColumn());
+ if (request.getMaxResults() != null) {
+ queryParams.put(MAX_RESULTS, request.getMaxResults().toString());
+ }
+ putQueryParameter(queryParams, PAGE_TOKEN, request.getPageToken());
+ return client.get(
+ resourcePaths.policies(request.getResource()),
+ queryParams,
+ ListPoliciesResponse.class,
+ restAuthFunction);
+ }
+
+ /** Creates a principal policy on its attachment resource. */
+ @Experimental
+ public void createPolicy(DataPolicy policy) {
+ client.post(
+ resourcePaths.policies(policy.getResource()),
+ new PolicyRequest(policy),
+ restAuthFunction);
+ }
+
+ /** Drops a principal policy from its exact attachment resource. */
+ @Experimental
+ public void dropPolicy(
+ PermissionResource resource,
+ PolicyType type,
+ String principal,
+ @Nullable String column,
+ boolean ignoreIfNotExists) {
+ checkNotNull(resource, "resource cannot be
null").validatePolicyAttachment();
+ try {
+ client.post(
+ resourcePaths.dropPolicy(resource),
+ new DropPolicyRequest(type, principal, column),
+ restAuthFunction);
+ } catch (NoSuchResourceException e) {
+ if (!ignoreIfNotExists
+ ||
!ErrorResponse.RESOURCE_TYPE_POLICY.equals(e.resourceType())) {
+ throw e;
+ }
+ }
+ }
+
/**
* Drop table.
*
diff --git
a/paimon-api/src/main/java/org/apache/paimon/rest/RESTPolicyManagement.java
b/paimon-api/src/main/java/org/apache/paimon/rest/RESTPolicyManagement.java
new file mode 100644
index 0000000000..45a42f6cf8
--- /dev/null
+++ b/paimon-api/src/main/java/org/apache/paimon/rest/RESTPolicyManagement.java
@@ -0,0 +1,72 @@
+/*
+ * 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.rest;
+
+import org.apache.paimon.PagedList;
+import org.apache.paimon.annotation.Experimental;
+import org.apache.paimon.management.DataPolicy;
+import org.apache.paimon.management.ListPoliciesRequest;
+import org.apache.paimon.management.PermissionResource;
+import org.apache.paimon.management.PolicyManagement;
+import
org.apache.paimon.management.PolicyManagement.PolicyAlreadyExistException;
+import org.apache.paimon.management.PolicyType;
+import org.apache.paimon.rest.exceptions.AlreadyExistsException;
+import org.apache.paimon.rest.responses.ErrorResponse;
+import org.apache.paimon.rest.responses.ListPoliciesResponse;
+
+import javax.annotation.Nullable;
+
+/** REST implementation of data policy management for a configured catalog
prefix. */
+@Experimental
+public class RESTPolicyManagement implements PolicyManagement {
+
+ private final RESTApi api;
+
+ public RESTPolicyManagement(RESTApi api) {
+ this.api = api;
+ }
+
+ @Override
+ public PagedList<DataPolicy> listPolicies(ListPoliciesRequest request) {
+ ListPoliciesResponse response = api.listPolicies(request);
+ return new PagedList<>(response.getPolicies(),
response.getNextPageToken());
+ }
+
+ @Override
+ public void createPolicy(DataPolicy policy) throws
PolicyAlreadyExistException {
+ try {
+ api.createPolicy(policy);
+ } catch (AlreadyExistsException e) {
+ if (ErrorResponse.RESOURCE_TYPE_POLICY.equals(e.resourceType())) {
+ throw new PolicyAlreadyExistException(policy, e);
+ }
+ throw e;
+ }
+ }
+
+ @Override
+ public void dropPolicy(
+ PermissionResource resource,
+ PolicyType type,
+ String principal,
+ @Nullable String column,
+ boolean ignoreIfNotExists) {
+ api.dropPolicy(resource, type, principal, column, ignoreIfNotExists);
+ }
+}
diff --git a/paimon-api/src/main/java/org/apache/paimon/rest/ResourcePaths.java
b/paimon-api/src/main/java/org/apache/paimon/rest/ResourcePaths.java
index 6ae7ebf8c3..4cef311061 100644
--- a/paimon-api/src/main/java/org/apache/paimon/rest/ResourcePaths.java
+++ b/paimon-api/src/main/java/org/apache/paimon/rest/ResourcePaths.java
@@ -19,6 +19,7 @@
package org.apache.paimon.rest;
import org.apache.paimon.annotation.Experimental;
+import org.apache.paimon.management.PermissionResource;
import org.apache.paimon.options.Options;
import org.apache.paimon.shade.guava30.com.google.common.base.Joiner;
@@ -44,6 +45,7 @@ public class ResourcePaths {
protected static final String FUNCTIONS = "functions";
protected static final String FUNCTION_DETAILS = "function-details";
protected static final String PERMISSIONS = "permissions";
+ protected static final String POLICIES = "policies";
protected static final String ID = "id";
private static final Joiner SLASH = Joiner.on("/").skipNulls();
@@ -77,6 +79,19 @@ public class ResourcePaths {
return SLASH.join(permissions(), "revoke");
}
+ /** Policy collection nested below its attachment resource. */
+ @Experimental
+ public String policies(PermissionResource resource) {
+ resource.validatePolicyAttachment();
+ return SLASH.join(table(resource.getDatabase(), resource.getTable()),
POLICIES);
+ }
+
+ /** Action endpoint for dropping one policy from its attachment resource.
*/
+ @Experimental
+ public String dropPolicy(PermissionResource resource) {
+ return SLASH.join(policies(resource), "drop");
+ }
+
public String databases() {
return SLASH.join(V1, prefix, DATABASES);
}
diff --git
a/paimon-api/src/main/java/org/apache/paimon/rest/requests/DropPolicyRequest.java
b/paimon-api/src/main/java/org/apache/paimon/rest/requests/DropPolicyRequest.java
new file mode 100644
index 0000000000..64b7b83c0e
--- /dev/null
+++
b/paimon-api/src/main/java/org/apache/paimon/rest/requests/DropPolicyRequest.java
@@ -0,0 +1,99 @@
+/*
+ * 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.rest.requests;
+
+import org.apache.paimon.annotation.Experimental;
+import org.apache.paimon.management.PermissionAssignment;
+import org.apache.paimon.management.PolicyType;
+import org.apache.paimon.rest.RESTRequest;
+
+import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonGetter;
+import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonInclude;
+import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
+import javax.annotation.Nullable;
+
+import java.beans.ConstructorProperties;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+import static org.apache.paimon.utils.Preconditions.checkNotNull;
+
+/** Request for dropping one principal's row filter or column mask. */
+@Experimental
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class DropPolicyRequest implements RESTRequest {
+
+ private static final String FIELD_TYPE = "type";
+ private static final String FIELD_PRINCIPAL = "principal";
+ private static final String FIELD_COLUMN = "column";
+
+ private final PolicyType type;
+ private final String principal;
+ @Nullable private final String column;
+
+ @JsonCreator
+ @ConstructorProperties({FIELD_TYPE, FIELD_PRINCIPAL, FIELD_COLUMN})
+ public DropPolicyRequest(
+ @JsonProperty(FIELD_TYPE) PolicyType type,
+ @JsonProperty(FIELD_PRINCIPAL) String principal,
+ @Nullable @JsonProperty(FIELD_COLUMN) String column) {
+ this.type = checkNotNull(type, "policy type cannot be null");
+ this.principal = validatePrincipal(principal);
+ if (type == PolicyType.ROW_FILTER) {
+ checkArgument(isBlank(column), "ROW_FILTER identity cannot contain
a column.");
+ this.column = null;
+ } else {
+ checkArgument(!isBlank(column), "column is required for
COLUMN_MASKING identity.");
+ this.column = column;
+ }
+ }
+
+ @JsonGetter(FIELD_TYPE)
+ public PolicyType getType() {
+ return type;
+ }
+
+ @JsonGetter(FIELD_PRINCIPAL)
+ public String getPrincipal() {
+ return principal;
+ }
+
+ @Nullable
+ @JsonGetter(FIELD_COLUMN)
+ @JsonInclude(JsonInclude.Include.NON_NULL)
+ public String getColumn() {
+ return column;
+ }
+
+ private static String validatePrincipal(String principal) {
+ checkArgument(
+ principal != null && !principal.trim().isEmpty(), "principal
cannot be empty.");
+ checkArgument(
+ principal.length() <=
PermissionAssignment.MAX_PRINCIPAL_LENGTH,
+ "principal must contain at most %s characters.",
+ PermissionAssignment.MAX_PRINCIPAL_LENGTH);
+ return principal;
+ }
+
+ private static boolean isBlank(@Nullable String value) {
+ return value == null || value.trim().isEmpty();
+ }
+}
diff --git
a/paimon-api/src/main/java/org/apache/paimon/rest/requests/PolicyRequest.java
b/paimon-api/src/main/java/org/apache/paimon/rest/requests/PolicyRequest.java
new file mode 100644
index 0000000000..f1e9a0ed53
--- /dev/null
+++
b/paimon-api/src/main/java/org/apache/paimon/rest/requests/PolicyRequest.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.rest.requests;
+
+import org.apache.paimon.annotation.Experimental;
+import org.apache.paimon.management.ColumnMask;
+import org.apache.paimon.management.DataPolicy;
+import org.apache.paimon.management.PermissionResource;
+import org.apache.paimon.management.RowFilter;
+import org.apache.paimon.rest.RESTRequest;
+
+import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonGetter;
+import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonIgnore;
+import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonInclude;
+import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
+import javax.annotation.Nullable;
+
+import java.beans.ConstructorProperties;
+
+/** Create payload for a principal policy whose table is identified by the
request path. */
+@Experimental
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class PolicyRequest implements RESTRequest {
+
+ private static final String FIELD_ROW_FILTER = "rowFilter";
+ private static final String FIELD_COLUMN_MASK = "columnMask";
+ private static final String FIELD_PRINCIPAL = "principal";
+
+ @Nullable private final RowFilter rowFilter;
+ @Nullable private final ColumnMask columnMask;
+ private final String principal;
+
+ public PolicyRequest(DataPolicy policy) {
+ this(policy.getRowFilter(), policy.getColumnMask(),
policy.getPrincipal());
+ }
+
+ @JsonCreator
+ @ConstructorProperties({FIELD_ROW_FILTER, FIELD_COLUMN_MASK,
FIELD_PRINCIPAL})
+ public PolicyRequest(
+ @Nullable @JsonProperty(FIELD_ROW_FILTER) RowFilter rowFilter,
+ @Nullable @JsonProperty(FIELD_COLUMN_MASK) ColumnMask columnMask,
+ @JsonProperty(FIELD_PRINCIPAL) String principal) {
+ this.rowFilter = rowFilter;
+ this.columnMask = columnMask;
+ this.principal = principal;
+ }
+
+ public DataPolicy policy(PermissionResource resource) {
+ return new DataPolicy(resource, rowFilter, columnMask, principal);
+ }
+
+ /** Creating a principal policy cannot be replayed after an ambiguous
server response. */
+ @JsonIgnore
+ @Override
+ public boolean isRetrySafe() {
+ return false;
+ }
+
+ @Nullable
+ @JsonGetter(FIELD_ROW_FILTER)
+ @JsonInclude(JsonInclude.Include.NON_NULL)
+ public RowFilter getRowFilter() {
+ return rowFilter;
+ }
+
+ @Nullable
+ @JsonGetter(FIELD_COLUMN_MASK)
+ @JsonInclude(JsonInclude.Include.NON_NULL)
+ public ColumnMask getColumnMask() {
+ return columnMask;
+ }
+
+ @JsonGetter(FIELD_PRINCIPAL)
+ public String getPrincipal() {
+ return principal;
+ }
+}
diff --git
a/paimon-api/src/main/java/org/apache/paimon/rest/responses/ErrorResponse.java
b/paimon-api/src/main/java/org/apache/paimon/rest/responses/ErrorResponse.java
index 4eb52308dd..4fe52b1d0b 100644
---
a/paimon-api/src/main/java/org/apache/paimon/rest/responses/ErrorResponse.java
+++
b/paimon-api/src/main/java/org/apache/paimon/rest/responses/ErrorResponse.java
@@ -51,6 +51,8 @@ public class ErrorResponse implements RESTResponse {
public static final String RESOURCE_TYPE_FUNCTION = "FUNCTION";
+ public static final String RESOURCE_TYPE_POLICY = "POLICY";
+
public static final String RESOURCE_TYPE_DEFINITION = "DEFINITION";
private static final String FIELD_MESSAGE = "message";
diff --git
a/paimon-api/src/main/java/org/apache/paimon/rest/responses/ListPoliciesResponse.java
b/paimon-api/src/main/java/org/apache/paimon/rest/responses/ListPoliciesResponse.java
new file mode 100644
index 0000000000..95595f066d
--- /dev/null
+++
b/paimon-api/src/main/java/org/apache/paimon/rest/responses/ListPoliciesResponse.java
@@ -0,0 +1,72 @@
+/*
+ * 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.rest.responses;
+
+import org.apache.paimon.annotation.Experimental;
+import org.apache.paimon.management.DataPolicy;
+
+import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonGetter;
+import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonInclude;
+import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
+import javax.annotation.Nullable;
+
+import java.beans.ConstructorProperties;
+import java.util.List;
+
+/** Response for listing data policies. */
+@Experimental
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class ListPoliciesResponse implements PagedResponse<DataPolicy> {
+
+ private static final String FIELD_POLICIES = "policies";
+ private static final String FIELD_NEXT_PAGE_TOKEN = "nextPageToken";
+
+ private final List<DataPolicy> policies;
+ @Nullable private final String nextPageToken;
+
+ @JsonCreator
+ @ConstructorProperties({FIELD_POLICIES, FIELD_NEXT_PAGE_TOKEN})
+ public ListPoliciesResponse(
+ @JsonProperty(FIELD_POLICIES) List<DataPolicy> policies,
+ @Nullable @JsonProperty(FIELD_NEXT_PAGE_TOKEN) String
nextPageToken) {
+ this.policies = policies;
+ this.nextPageToken = nextPageToken;
+ }
+
+ @JsonGetter(FIELD_POLICIES)
+ public List<DataPolicy> getPolicies() {
+ return policies;
+ }
+
+ @Override
+ @Nullable
+ @JsonGetter(FIELD_NEXT_PAGE_TOKEN)
+ @JsonInclude(JsonInclude.Include.NON_NULL)
+ public String getNextPageToken() {
+ return nextPageToken;
+ }
+
+ @Override
+ public List<DataPolicy> data() {
+ return policies;
+ }
+}
diff --git
a/paimon-api/src/test/java/org/apache/paimon/management/PolicyManagementJsonTest.java
b/paimon-api/src/test/java/org/apache/paimon/management/PolicyManagementJsonTest.java
new file mode 100644
index 0000000000..53c1905144
--- /dev/null
+++
b/paimon-api/src/test/java/org/apache/paimon/management/PolicyManagementJsonTest.java
@@ -0,0 +1,176 @@
+/*
+ * 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.management;
+
+import org.apache.paimon.rest.RESTApi;
+import org.apache.paimon.rest.requests.DropPolicyRequest;
+import org.apache.paimon.rest.requests.PolicyRequest;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** JSON and validation tests for data-policy management contracts. */
+public class PolicyManagementJsonTest {
+
+ private static final String PREDICATE_JSON =
+ "{\"kind\":\"LEAF\",\"transform\":{\"name\":\"FIELD_REF\","
+ +
"\"fieldRef\":{\"index\":0,\"name\":\"region\",\"type\":\"STRING\"}},"
+ + "\"function\":\"EQUAL\",\"literals\":[\"APAC\"]}";
+ private static final String TRANSFORM_JSON =
+
"{\"name\":\"CONCAT\",\"inputs\":[{\"index\":0,\"name\":\"region\","
+ + "\"type\":\"STRING\"},\"****\"]}";
+
+ @Test
+ void testPolicyDefinitionsRoundTripWithShadedAndExternalJackson() throws
Exception {
+ DataPolicy policy =
+ DataPolicy.columnMask(
+ tableResource(), new ColumnMask("email",
TRANSFORM_JSON), "analyst");
+ String json = RESTApi.toJson(policy);
+
+ DataPolicy shaded = RESTApi.fromJson(json, DataPolicy.class);
+ DataPolicy external =
+ new
com.fasterxml.jackson.databind.ObjectMapper().readValue(json, DataPolicy.class);
+ for (DataPolicy roundTrip : Arrays.asList(shaded, external)) {
+ assertThat(roundTrip.type()).isEqualTo(PolicyType.COLUMN_MASKING);
+ assertThat(roundTrip.getResource()).isEqualTo(tableResource());
+
assertThat(roundTrip.getColumnMask().getOnColumn()).isEqualTo("email");
+
assertThat(roundTrip.getColumnMask().getTransform()).isEqualTo(TRANSFORM_JSON);
+ assertThat(roundTrip.getPrincipal()).isEqualTo("analyst");
+ }
+
+ PolicyRequest request = new PolicyRequest(policy);
+ assertThat(request.isRetrySafe()).isFalse();
+ Map<?, ?> wire = RESTApi.fromJson(RESTApi.toJson(request), Map.class);
+ assertThat(wire).hasSize(2);
+ assertThat(wire.get("columnMask")).isNotNull();
+ assertThat(wire.get("principal")).isEqualTo("analyst");
+
assertThat(request.policy(tableResource()).getResource()).isEqualTo(tableResource());
+ }
+
+ @Test
+ void testRowFilterRoundTrip() throws Exception {
+ DataPolicy policy =
+ DataPolicy.rowFilter(tableResource(), new
RowFilter(PREDICATE_JSON), "analyst");
+
+ DataPolicy roundTrip = RESTApi.fromJson(RESTApi.toJson(policy),
DataPolicy.class);
+ assertThat(roundTrip.type()).isEqualTo(PolicyType.ROW_FILTER);
+
assertThat(roundTrip.getRowFilter().getPredicate()).isEqualTo(PREDICATE_JSON);
+ assertThat(roundTrip.getColumnMask()).isNull();
+ }
+
+ @Test
+ void testPolicyValidationAndPayloadBounds() {
+ assertThatThrownBy(() -> new RowFilter(" "))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("predicate");
+ assertThatThrownBy(() -> new ColumnMask("email", " "))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("transform");
+ assertThatThrownBy(
+ () ->
+ new DataPolicy(
+ tableResource(),
+ new RowFilter(PREDICATE_JSON),
+ new ColumnMask("email",
TRANSFORM_JSON),
+ "analyst"))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("exactly one");
+ assertThatThrownBy(
+ () ->
+ DataPolicy.columnMask(
+ catalogResource(),
+ new ColumnMask("email",
TRANSFORM_JSON),
+ "analyst"))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("TABLE");
+ assertThatThrownBy(() -> new RowFilter(repeat('p',
RowFilter.MAX_PREDICATE_BYTES + 1)))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("UTF-8 bytes");
+ assertThatThrownBy(
+ () ->
+ new ColumnMask(
+ "email", repeat('t',
ColumnMask.MAX_TRANSFORM_BYTES + 1)))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("UTF-8 bytes");
+ }
+
+ @Test
+ void testDropPolicyRequestRoundTripAndIdentityValidation() throws
Exception {
+ DropPolicyRequest request =
+ new DropPolicyRequest(PolicyType.COLUMN_MASKING, "analyst",
"email");
+ DropPolicyRequest roundTrip =
+ RESTApi.fromJson(RESTApi.toJson(request),
DropPolicyRequest.class);
+
+ assertThat(roundTrip.getType()).isEqualTo(PolicyType.COLUMN_MASKING);
+ assertThat(roundTrip.getPrincipal()).isEqualTo("analyst");
+ assertThat(roundTrip.getColumn()).isEqualTo("email");
+ assertThatThrownBy(() -> new
DropPolicyRequest(PolicyType.COLUMN_MASKING, "analyst", null))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("column is required");
+ assertThatThrownBy(() -> new DropPolicyRequest(PolicyType.ROW_FILTER,
"analyst", "email"))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("cannot contain a column");
+ }
+
+ @Test
+ void testListPoliciesValidationAndOpaquePageToken() {
+ ListPoliciesRequest request =
+ new ListPoliciesRequest(
+ tableResource(), PolicyType.COLUMN_MASKING, "analyst",
"email", null, 25);
+
+ assertThat(request.withPageToken(" \t").getPageToken()).isEqualTo("
\t");
+ assertThatThrownBy(
+ () ->
+ new ListPoliciesRequest(
+ tableResource(), null, null, "email",
null, 25))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("COLUMN_MASKING");
+ assertThatThrownBy(
+ () ->
+ new ListPoliciesRequest(
+ tableResource(), null, null, null,
null, 1001))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("at most 1000");
+ assertThatThrownBy(
+ () ->
+ new ListPoliciesRequest(
+ catalogResource(), null, null, null,
null, 25))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("TABLE");
+ }
+
+ private static PermissionResource catalogResource() {
+ return new PermissionResource(ResourceType.CATALOG, null, null, null,
null);
+ }
+
+ private static PermissionResource tableResource() {
+ return new PermissionResource(ResourceType.TABLE, "sales", "orders",
null, null);
+ }
+
+ private static String repeat(char value, int length) {
+ char[] values = new char[length];
+ Arrays.fill(values, value);
+ return new String(values);
+ }
+}
diff --git
a/paimon-api/src/test/java/org/apache/paimon/rest/RESTPolicyManagementTest.java
b/paimon-api/src/test/java/org/apache/paimon/rest/RESTPolicyManagementTest.java
new file mode 100644
index 0000000000..a09e0e902a
--- /dev/null
+++
b/paimon-api/src/test/java/org/apache/paimon/rest/RESTPolicyManagementTest.java
@@ -0,0 +1,251 @@
+/*
+ * 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.rest;
+
+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.PermissionResource;
+import org.apache.paimon.management.PolicyManagement;
+import
org.apache.paimon.management.PolicyManagement.PolicyAlreadyExistException;
+import org.apache.paimon.management.PolicyType;
+import org.apache.paimon.management.ResourceType;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.rest.exceptions.AlreadyExistsException;
+import org.apache.paimon.rest.exceptions.NoSuchResourceException;
+import org.apache.paimon.utils.JsonSerdeUtil;
+
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpServer;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.apache.paimon.rest.RESTCatalogInternalOptions.PREFIX;
+import static org.apache.paimon.rest.RESTCatalogOptions.TOKEN;
+import static org.apache.paimon.rest.RESTCatalogOptions.TOKEN_PROVIDER;
+import static org.apache.paimon.rest.RESTCatalogOptions.URI;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Behavioral tests for REST data-policy management. */
+public class RESTPolicyManagementTest {
+
+ private static final String COLLECTION_PATH =
+ "/v1/catalog+id/databases/sales/tables/orders/policies";
+ private static final String DROP_PATH = COLLECTION_PATH + "/drop";
+
+ private HttpServer server;
+ private PolicyManagement management;
+ private final AtomicReference<String> listQuery = new AtomicReference<>();
+ private final AtomicReference<String> createBody = new AtomicReference<>();
+ private final AtomicReference<String> createError = new
AtomicReference<>();
+ private final AtomicReference<String> dropBody = new AtomicReference<>();
+ private final AtomicReference<String> dropError = new AtomicReference<>();
+ private final AtomicInteger dropCalls = new AtomicInteger();
+
+ @BeforeEach
+ void setUp() throws Exception {
+ server = HttpServer.create(new InetSocketAddress(0), 0);
+ server.createContext(
+ "/v1/",
+ exchange -> {
+ String path = exchange.getRequestURI().getRawPath();
+ String method = exchange.getRequestMethod();
+ if (COLLECTION_PATH.equals(path) && "GET".equals(method)) {
+ listQuery.set(exchange.getRequestURI().getRawQuery());
+ respond(exchange, 200, listResponse());
+ } else if (COLLECTION_PATH.equals(path) &&
"POST".equals(method)) {
+ createBody.set(readBody(exchange));
+ if (createError.get() == null) {
+ respond(exchange, 200, null);
+ } else {
+ respond(exchange, 409, createError.get());
+ }
+ } else if (DROP_PATH.equals(path) &&
"POST".equals(method)) {
+ dropBody.set(readBody(exchange));
+ dropCalls.incrementAndGet();
+ if (dropError.get() == null) {
+ respond(exchange, 200, null);
+ } else {
+ respond(exchange, 404, dropError.get());
+ }
+ } else {
+ respond(exchange, 404,
"{\"message\":\"missing\",\"code\":404}");
+ }
+ });
+ server.start();
+
+ Options options = new Options();
+ options.set(URI, "http://127.0.0.1:"; + server.getAddress().getPort());
+ options.set(TOKEN_PROVIDER, "bear");
+ options.set(TOKEN, "secret");
+ options.set(PREFIX, "catalog id");
+ management = new RESTPolicyManagement(new RESTApi(options, false));
+ }
+
+ @AfterEach
+ void tearDown() {
+ if (server != null) {
+ server.stop(0);
+ }
+ }
+
+ @Test
+ void testListUsesResourceNestedPathAndIdentityFilters() {
+ PagedList<DataPolicy> policies =
+ management.listPolicies(
+ new ListPoliciesRequest(
+ tableResource(),
+ PolicyType.COLUMN_MASKING,
+ "analyst",
+ "email",
+ "start",
+ 25));
+
+ assertThat(policies.getElements()).hasSize(1);
+ assertThat(policies.getNextPageToken()).isEqualTo("next");
+ assertThat(listQuery.get())
+ .contains("type=COLUMN_MASKING")
+ .contains("principal=analyst")
+ .contains("column=email")
+ .contains("maxResults=25")
+ .contains("pageToken=start");
+ }
+
+ @Test
+ void testCreateAndDropUsePostEndpoints() throws Exception {
+ DataPolicy policy = policy();
+ management.createPolicy(policy);
+ management.dropPolicy(
+ policy.getResource(),
+ policy.type(),
+ policy.getPrincipal(),
+ policy.getColumnMask().getOnColumn(),
+ false);
+
+ assertThat(createBody.get()).contains("\"principal\":\"analyst\"");
+ assertThat(createBody.get()).doesNotContain("\"resource\"");
+ assertThat(dropBody.get())
+ .contains("\"type\":\"COLUMN_MASKING\"")
+ .contains("\"column\":\"email\"");
+ assertThat(dropCalls).hasValue(1);
+ }
+
+ @Test
+ void testCreateMapsOnlyPolicyConflict() {
+ DataPolicy policy = policy();
+ createError.set(
+ "{\"resourceType\":\"POLICY\",\"resourceName\":"
+ + "\"COLUMN_MASKING:analyst:email\","
+ + "\"message\":\"already exists\",\"code\":409}");
+
+ assertThatThrownBy(() -> management.createPolicy(policy))
+ .isInstanceOf(PolicyAlreadyExistException.class)
+ .hasMessageContaining("COLUMN_MASKING(email)")
+ .hasCauseInstanceOf(AlreadyExistsException.class);
+
+ createError.set(
+ "{\"resourceType\":\"TABLE\",\"resourceName\":\"orders\","
+ + "\"message\":\"table conflict\",\"code\":409}");
+ assertThatThrownBy(() -> management.createPolicy(policy))
+ .isInstanceOf(AlreadyExistsException.class)
+ .hasMessageContaining("table conflict");
+ }
+
+ @Test
+ void testDropIfExistsOnlyIgnoresMissingPolicy() {
+ DataPolicy policy = policy();
+ dropError.set(
+ "{\"resourceType\":\"POLICY\",\"resourceName\":"
+ + "\"COLUMN_MASKING:analyst:email\","
+ + "\"message\":\"missing\",\"code\":404}");
+
+ management.dropPolicy(
+ policy.getResource(),
+ policy.type(),
+ policy.getPrincipal(),
+ policy.getColumnMask().getOnColumn(),
+ true);
+
+ dropError.set(
+ "{\"resourceType\":\"TABLE\",\"resourceName\":\"orders\","
+ + "\"message\":\"missing table\",\"code\":404}");
+ assertThatThrownBy(
+ () ->
+ management.dropPolicy(
+ policy.getResource(),
+ policy.type(),
+ policy.getPrincipal(),
+ policy.getColumnMask().getOnColumn(),
+ true))
+ .isInstanceOf(NoSuchResourceException.class)
+ .hasMessageContaining("missing table");
+ }
+
+ private static DataPolicy policy() {
+ return DataPolicy.columnMask(
+ tableResource(),
+ new ColumnMask(
+ "email",
+ "{\"name\":\"FIELD_REF\",\"fieldRef\":{\"index\":0,"
+ + "\"name\":\"region\",\"type\":\"STRING\"}}"),
+ "analyst");
+ }
+
+ private static PermissionResource tableResource() {
+ return new PermissionResource(ResourceType.TABLE, "sales", "orders",
null, null);
+ }
+
+ private static String listResponse() {
+ return "{\"policies\":[" + policyJson() +
"],\"nextPageToken\":\"next\"}";
+ }
+
+ private static String policyJson() {
+ return JsonSerdeUtil.toFlatJson(policy());
+ }
+
+ private static String readBody(HttpExchange exchange) throws IOException {
+ byte[] data = new byte[8192];
+ int read = exchange.getRequestBody().read(data);
+ return read < 0 ? "" : new String(data, 0, read,
StandardCharsets.UTF_8);
+ }
+
+ private static void respond(HttpExchange exchange, int code, String body)
throws IOException {
+ if (body == null) {
+ exchange.sendResponseHeaders(code, 0);
+ exchange.getResponseBody().close();
+ } else {
+ byte[] bytes = body.getBytes(StandardCharsets.UTF_8);
+ exchange.sendResponseHeaders(code, bytes.length);
+ try (OutputStream output = exchange.getResponseBody()) {
+ output.write(bytes);
+ }
+ exchange.close();
+ }
+ }
+}
diff --git
a/paimon-api/src/test/java/org/apache/paimon/rest/requests/RequestJacksonCompatibilityTest.java
b/paimon-api/src/test/java/org/apache/paimon/rest/requests/RequestJacksonCompatibilityTest.java
index 30ebab5463..9e645db3b7 100644
---
a/paimon-api/src/test/java/org/apache/paimon/rest/requests/RequestJacksonCompatibilityTest.java
+++
b/paimon-api/src/test/java/org/apache/paimon/rest/requests/RequestJacksonCompatibilityTest.java
@@ -169,7 +169,9 @@ public class RequestJacksonCompatibilityTest {
CreatePartitionsRequest.class,
CreateTableRequest.class,
CreateViewRequest.class,
+ DropPolicyRequest.class,
GrantPermissionRequest.class,
+ PolicyRequest.class,
RegisterTableRequest.class,
RenameTableRequest.class,
ReplaceTableRequest.class,
diff --git a/paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalog.java
b/paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalog.java
index 3c668aeacb..9a21f3fa41 100644
--- a/paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalog.java
+++ b/paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalog.java
@@ -41,6 +41,7 @@ import org.apache.paimon.fs.cache.LocalCacheManager;
import org.apache.paimon.function.Function;
import org.apache.paimon.function.FunctionChange;
import org.apache.paimon.management.PermissionManagement;
+import org.apache.paimon.management.PolicyManagement;
import org.apache.paimon.options.Options;
import org.apache.paimon.partition.Partition;
import org.apache.paimon.partition.PartitionStatistics;
@@ -144,6 +145,11 @@ public class RESTCatalog implements Catalog {
return new RESTPermissionManagement(api);
}
+ @Experimental
+ public PolicyManagement policyManagement() {
+ return new RESTPolicyManagement(api);
+ }
+
@Override
public List<String> listDatabases() {
return api.listDatabases();
diff --git
a/paimon-core/src/test/java/org/apache/paimon/rest/ResourcePathsTest.java
b/paimon-core/src/test/java/org/apache/paimon/rest/ResourcePathsTest.java
index 432ea8f776..b6f0f39f38 100644
--- a/paimon-core/src/test/java/org/apache/paimon/rest/ResourcePathsTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/rest/ResourcePathsTest.java
@@ -18,9 +18,13 @@
package org.apache.paimon.rest;
+import org.apache.paimon.management.PermissionResource;
+import org.apache.paimon.management.ResourceType;
+
import org.junit.Test;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
/** Test for {@link ResourcePaths}. */
public class ResourcePathsTest {
@@ -47,4 +51,24 @@ public class ResourcePathsTest {
assertEquals("/v1/catalog%2Fid/permissions/grant",
resourcePaths.grantPermission());
assertEquals("/v1/catalog%2Fid/permissions/revoke",
resourcePaths.revokePermission());
}
+
+ @Test
+ public void testPoliciesAreNestedUnderAttachmentResource() {
+ ResourcePaths paths = new ResourcePaths("catalog/id");
+ PermissionResource catalog =
+ new PermissionResource(ResourceType.CATALOG, null, null, null,
null);
+ PermissionResource database =
+ new PermissionResource(ResourceType.DATABASE, "sales db",
null, null, null);
+ PermissionResource table =
+ new PermissionResource(ResourceType.TABLE, "sales db",
"orders/all", null, null);
+
+ assertThrows(IllegalArgumentException.class, () ->
paths.policies(catalog));
+ assertThrows(IllegalArgumentException.class, () ->
paths.policies(database));
+ assertEquals(
+
"/v1/catalog%2Fid/databases/sales+db/tables/orders%2Fall/policies",
+ paths.policies(table));
+ assertEquals(
+
"/v1/catalog%2Fid/databases/sales+db/tables/orders%2Fall/policies/drop",
+ paths.dropPolicy(table));
+ }
}
