[api] Add REST policy management

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();
+ }
+}

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;
+ }
+}

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();
+ }
+}

paimon-api/src/main/java/org/apache/paimon/management/PermissionResource.java

@@ -122,6 +122,13 @@ public String getView() {
  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) {

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());
+ }
+ }
+}