Part 1 : Adds RLS and CLS control Policies

polaris-core/src/main/java/org/apache/polaris/core/policy/PolicyUtil.java

@@ -0,0 +1,108 @@
+/*
+ * 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.polaris.core.policy;
+
+import static org.apache.polaris.core.policy.PredefinedPolicyTypes.ACCESS_CONTROL;
+
+import com.google.common.collect.Lists;
+import java.util.List;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.expressions.UnboundPredicate;
+import org.apache.polaris.core.auth.AuthenticatedPolarisPrincipal;
+import org.apache.polaris.core.policy.content.AccessControlPolicyContent;
+
+public class PolicyUtil {
+  private PolicyUtil() {}
+
+  public static String replaceContextVariable(
+      String content, PolicyType policyType, AuthenticatedPolarisPrincipal authenticatedPrincipal) {
+    if (policyType == ACCESS_CONTROL) {
+      try {
+        AccessControlPolicyContent policyContent = AccessControlPolicyContent.fromString(content);
+        List<Expression> evaluatedRowFilterExpressions = Lists.newArrayList();
+        for (Expression rowFilterExpression : policyContent.getRowFilters()) {
+          // check if the expression refers to context variable current_principal_role
+          if (rowFilterExpression instanceof UnboundPredicate<?>) {
+            UnboundPredicate<?> boundPredicate = (UnboundPredicate<?>) rowFilterExpression;
+            // check if this references to current_principal
+            if (boundPredicate.ref().name().equals("$current_principal_role")) {
+              // compare the literal and replace
+              String val = (String) boundPredicate.literal().value();
+              if (authenticatedPrincipal.getActivatedPrincipalRoleNames().contains(val)) {
+                // TODO: see if we can utilize the expression evaluation of iceberg SDK
+                Expression result =
+                    boundPredicate.op().equals(Expression.Operation.EQ)
+                        ? Expressions.alwaysTrue()
+                        : Expressions.alwaysFalse();
+                evaluatedRowFilterExpressions.add(result);
+              } else {
+                Expression result =
+                    boundPredicate.op().equals(Expression.Operation.NOT_EQ)
+                        ? Expressions.alwaysTrue()
+                        : Expressions.alwaysFalse();
+                evaluatedRowFilterExpressions.add(result);
+              }
+            } else if (boundPredicate.ref().name().equals("$current_principal")) {
+              String val = (String) boundPredicate.literal().value();
+              if (authenticatedPrincipal.getName().equals(val)) {
+                Expression result =
+                    boundPredicate.op().equals(Expression.Operation.EQ)
+                        ? Expressions.alwaysTrue()
+                        : Expressions.alwaysFalse();
+                evaluatedRowFilterExpressions.add(result);
+              } else {
+                Expression result =
+                    boundPredicate.op().equals(Expression.Operation.NOT_EQ)
+                        ? Expressions.alwaysTrue()
+                        : Expressions.alwaysFalse();
+                evaluatedRowFilterExpressions.add(result);
+              }
+            } else {
+              evaluatedRowFilterExpressions.add(rowFilterExpression);
+            }
+          }
+        }
+
+        policyContent.setRowFilters(evaluatedRowFilterExpressions);
+        return AccessControlPolicyContent.toString(policyContent);
+      } catch (Exception e) {
+        return content;
+      }
+    }
+    return content;
+  }
+
+  public static boolean filterApplicablePolicy(
+      PolicyEntity policyEntity, AuthenticatedPolarisPrincipal authenticatedPrincipal) {
+    if (policyEntity.getPolicyType().equals(ACCESS_CONTROL)) {
+      AccessControlPolicyContent content =
+          AccessControlPolicyContent.fromString(policyEntity.getContent());
+      String applicablePrincipal = content.getPrincipalRole();
+      return applicablePrincipal == null
+          || authenticatedPrincipal.getActivatedPrincipalRoleNames().isEmpty()
+          || authenticatedPrincipal
+              .getActivatedPrincipalRoleNames()
+              .contains(content.getPrincipalRole());
+    }
+
+    return true;
+  }
+}

polaris-core/src/main/java/org/apache/polaris/core/policy/PredefinedPolicyTypes.java

@@ -28,7 +28,8 @@ public enum PredefinedPolicyTypes implements PolicyType {
   DATA_COMPACTION(0, "system.data-compaction", true),
   METADATA_COMPACTION(1, "system.metadata-compaction", true),
   ORPHAN_FILE_REMOVAL(2, "system.orphan-file-removal", true),
-  SNAPSHOT_EXPIRY(3, "system.snapshot-expiry", true);
+  SNAPSHOT_EXPIRY(3, "system.snapshot-expiry", true),
+  ACCESS_CONTROL(4, "system.access-control", false);
 
   private final int code;
   private final String name;

polaris-core/src/main/java/org/apache/polaris/core/policy/content/AccessControlPolicyContent.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.polaris.core.policy.content;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import com.fasterxml.jackson.databind.annotation.JsonSerialize;
+import com.google.common.base.Strings;
+import java.util.List;
+import java.util.Set;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.polaris.core.policy.validator.InvalidPolicyException;
+
+public class AccessControlPolicyContent implements PolicyContent {
+
+  // Optional, if there means policies is applicable to the role
+  private String principalRole;
+
+  // TODO: model them as iceberg transforms
+  private List<String> columnProjections;
+
+  // Iceberg expressions without context functions for now.
+  // Use a custom deserializer for the list of Iceberg Expressions
+  @JsonDeserialize(using = IcebergExpressionListDeserializer.class)
+  @JsonSerialize(using = IcebergExpressionListSerializer.class)
+  private List<Expression> rowFilters;
+
+  private static final String DEFAULT_POLICY_SCHEMA_VERSION = "2025-02-03";
+  private static final Set<String> POLICY_SCHEMA_VERSIONS = Set.of(DEFAULT_POLICY_SCHEMA_VERSION);
+
+  public static AccessControlPolicyContent fromString(String content) {
+    if (Strings.isNullOrEmpty(content)) {
+      throw new InvalidPolicyException("Policy is empty");
+    }
+
+    AccessControlPolicyContent policy;
+    try {
+      policy = PolicyContentUtil.MAPPER.readValue(content, AccessControlPolicyContent.class);
+    } catch (Exception e) {
+      throw new InvalidPolicyException(e);
+    }
+
+    boolean isProjectionsEmpty =
+        policy.getColumnProjections() == null || policy.getColumnProjections().isEmpty();
+    boolean isRowFilterEmpty = policy.getRowFilters() == null || policy.getRowFilters().isEmpty();
+    if (isProjectionsEmpty && isRowFilterEmpty) {
+      throw new InvalidPolicyException("Policy must contain 'columnProjections' or 'rowFilters'.");
+    }
+
+    return policy;
+  }
+
+  public static String toString(AccessControlPolicyContent content) {
+    if (content == null) {
+      return null;
+    }
+    try {
+      return PolicyContentUtil.MAPPER.writeValueAsString(content);
+    } catch (JsonProcessingException e) {
+      throw new InvalidPolicyException("Failed to convert policy content to JSON string", e);
+    }
+  }
+
+  // Constructors, getters, and setters
+  public AccessControlPolicyContent() {}
+
+  public String getPrincipalRole() {
+    return principalRole;
+  }
+
+  public void setPrincipalRole(String principalRole) {
+    this.principalRole = principalRole;
+  }
+
+  public List<String> getColumnProjections() {
+    return columnProjections;
+  }
+
+  public void setAllowedColumns(List<String> columnProjections) {
+    this.columnProjections = columnProjections;
+  }
+
+  public List<Expression> getRowFilters() {
+    return rowFilters;
+  }
+
+  public void setRowFilters(List<Expression> rowFilters) {
+    this.rowFilters = rowFilters;
+  }
+
+  @Override
+  public String toString() {
+    return "AccessControlPolicyContent{"
+        + "principalRole='"
+        + principalRole
+        + '\''
+        + ", columnProjections="
+        + columnProjections
+        + ", rowFilters="
+        + rowFilters
+        + '}';
+  }
+}

polaris-core/src/main/java/org/apache/polaris/core/policy/content/IcebergExpressionListDeserializer.java

@@ -0,0 +1,49 @@
+/*
+ * 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.polaris.core.policy.content;
+
+import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.databind.DeserializationContext;
+import com.fasterxml.jackson.databind.JsonDeserializer;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.ExpressionParser;
+
+public class IcebergExpressionListDeserializer extends JsonDeserializer<List<Expression>> {
+  @Override
+  public List<Expression> deserialize(JsonParser p, DeserializationContext ctxt)
+      throws IOException {
+    ObjectMapper mapper = (ObjectMapper) p.getCodec();
+    JsonNode node = mapper.readTree(p);
+
+    List<Expression> expressions = new ArrayList<>();
+    if (node.isArray()) {
+      for (JsonNode element : node) {
+        // Convert each JSON element back to a string and pass it to ExpressionParser.fromJson
+        expressions.add(ExpressionParser.fromJson(mapper.writeValueAsString(element)));
+      }
+    }
+    return expressions;
+  }
+}

polaris-core/src/main/java/org/apache/polaris/core/policy/content/IcebergExpressionListSerializer.java

@@ -0,0 +1,50 @@
+/*
+ * 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.polaris.core.policy.content;
+
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.databind.JsonSerializer;
+import com.fasterxml.jackson.databind.SerializerProvider;
+import java.io.IOException;
+import java.util.List;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.ExpressionParser;
+
+/**
+ * Custom Jackson JsonSerializer for a List of Iceberg Expression objects. This serializer converts
+ * each Iceberg Expression into its string representation.
+ */
+public class IcebergExpressionListSerializer extends JsonSerializer<List<Expression>> {
+
+  @Override
+  public void serialize(
+      List<Expression> expressions,
+      JsonGenerator jsonGenerator,
+      SerializerProvider serializerProvider)
+      throws IOException {
+    jsonGenerator.writeStartArray();
+    if (expressions != null) {
+      for (Expression expression : expressions) {
+        jsonGenerator.writeString(ExpressionParser.toJson(expression));
+      }
+    }
+    jsonGenerator.writeEndArray();
+  }
+}

polaris-core/src/main/java/org/apache/polaris/core/policy/content/PolicyContentUtil.java

@@ -20,6 +20,7 @@
 
 import com.fasterxml.jackson.databind.DeserializationFeature;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.iceberg.rest.RESTSerializers;
 
 public class PolicyContentUtil {
   public static final ObjectMapper MAPPER = configureMapper();
@@ -30,6 +31,8 @@ private static ObjectMapper configureMapper() {
     mapper.configure(DeserializationFeature.FAIL_ON_MISSING_CREATOR_PROPERTIES, true);
     // Fails if a required field is present but explicitly null, e.g., {"enable": null}
     mapper.configure(DeserializationFeature.FAIL_ON_NULL_CREATOR_PROPERTIES, true);
+    // This will make sure all the Iceberg parsers are loaded.
+    RESTSerializers.registerAll(mapper);
     return mapper;
   }
 }

polaris-core/src/main/java/org/apache/polaris/core/policy/validator/PolicyValidators.java

@@ -22,6 +22,7 @@
 import org.apache.polaris.core.entity.PolarisEntity;
 import org.apache.polaris.core.policy.PolicyEntity;
 import org.apache.polaris.core.policy.PredefinedPolicyTypes;
+import org.apache.polaris.core.policy.content.AccessControlPolicyContent;
 import org.apache.polaris.core.policy.content.maintenance.DataCompactionPolicyContent;
 import org.apache.polaris.core.policy.content.maintenance.MetadataCompactionPolicyContent;
 import org.apache.polaris.core.policy.content.maintenance.OrphanFileRemovalPolicyContent;
@@ -66,6 +67,9 @@ public static void validate(PolicyEntity policy) {
       case ORPHAN_FILE_REMOVAL:
         OrphanFileRemovalPolicyContent.fromString(policy.getContent());
         break;
+      case ACCESS_CONTROL:
+        AccessControlPolicyContent.fromString(policy.getContent());
+        break;
       default:
         throw new IllegalArgumentException("Unsupported policy type: " + type.getName());
     }
@@ -98,6 +102,10 @@ public static boolean canAttach(PolicyEntity policy, PolarisEntity targetEntity)
       case ORPHAN_FILE_REMOVAL:
         return BaseMaintenancePolicyValidator.INSTANCE.canAttach(entityType, entitySubType);
 
+      case ACCESS_CONTROL:
+        // TODO: Add validator for attaching this only to table
+        return true;
+
       default:
         LOGGER.warn("Attachment not supported for policy type: {}", policyType.getName());
         return false;