Core: Add data compaction policy content parser and validator (#1238)

Author: flyrain

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

@@ -0,0 +1,22 @@
+/*
+ * 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;
+
+/** A marker interface for policy content */
+public interface PolicyContent {}

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

@@ -0,0 +1,36 @@
+/*
+ * 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.validator;
+
+import org.apache.polaris.core.exceptions.PolarisException;
+
+/** Exception thrown when a policy is invalid or violates defined rules. */
+public class InvalidPolicyException extends PolarisException {
+  public InvalidPolicyException(String message) {
+    super(message);
+  }
+
+  public InvalidPolicyException(String message, Throwable cause) {
+    super(message, cause);
+  }
+
+  public InvalidPolicyException(Throwable cause) {
+    super("Invalid policy", cause);
+  }
+}

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

@@ -0,0 +1,47 @@
+/*
+ * 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.validator;
+
+import org.apache.polaris.core.entity.PolarisEntitySubType;
+import org.apache.polaris.core.entity.PolarisEntityType;
+
+/** Validates and parses a given policy content string against its defined schema. */
+public interface PolicyValidator {
+
+  /**
+   * Validates the provided policy content.
+   *
+   * @param content the policy content to parse and validate
+   * @throws InvalidPolicyException if the content is not valid
+   */
+  void validate(String content) throws InvalidPolicyException;
+
+  /**
+   * Determines whether the policy is attachable to a target entity.
+   *
+   * <p>This method examines the provided {@link PolarisEntityType} and {@link PolarisEntitySubType}
+   * to decide if a policy is applicable for attachment to the target entity.
+   *
+   * @param entityType the type of the target entity
+   * @param entitySubType the subtype of the target entity
+   * @return {@code true} if the policy can be attached to the target entity; {@code false}
+   *     otherwise
+   */
+  boolean canAttach(PolarisEntityType entityType, PolarisEntitySubType entitySubType);
+}

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

@@ -0,0 +1,35 @@
+/*
+ * 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.validator;
+
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+public class PolicyValidatorUtil {
+  public static final ObjectMapper MAPPER = configureMapper();
+
+  private static ObjectMapper configureMapper() {
+    ObjectMapper mapper = new ObjectMapper();
+    // Fails if a required field (in the constructor) is missing
+    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);
+    return mapper;
+  }
+}

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

@@ -0,0 +1,95 @@
+/*
+ * 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.validator;
+
+import com.google.common.base.Preconditions;
+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.validator.datacompaction.DataCompactionPolicyValidator;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Validates a given {@link PolicyEntity} against its defined policy type.
+ *
+ * <p>This class maps the policy type code from the {@code PolicyEntity} to a predefined policy
+ * type, then delegates parsing/validation to a specific validator implementation.
+ */
+public class PolicyValidators {
+  private static final Logger LOGGER = LoggerFactory.getLogger(PolicyValidators.class);
+
+  /**
+   * Validates the given policy.
+   *
+   * @param policy the policy entity to validate
+   * @throws InvalidPolicyException if the policy type is unknown or unsupported, or if the policy
+   *     content is invalid
+   */
+  public static void validate(PolicyEntity policy) {
+    Preconditions.checkNotNull(policy, "Policy must not be null");
+
+    var type = PredefinedPolicyTypes.fromCode(policy.getPolicyTypeCode());
+    Preconditions.checkArgument(type != null, "Unknown policy type: " + policy.getPolicyTypeCode());
+
+    switch (type) {
+      case DATA_COMPACTION:
+        DataCompactionPolicyValidator.INSTANCE.validate(policy.getContent());
+        break;
+
+      // To support additional policy types in the future, add cases here.
+      case METADATA_COMPACTION:
+      case SNAPSHOT_RETENTION:
+      case ORPHAN_FILE_REMOVAL:
+      default:
+        throw new InvalidPolicyException("Unsupported policy type: " + type.getName());
+    }
+
+    LOGGER.info("Policy validated successfully: {}", type.getName());
+  }
+
+  /**
+   * Determines whether the given policy can be attached to the specified target entity.
+   *
+   * @param policy the policy entity to check
+   * @param targetEntity the target Polaris entity to attach the policy to
+   * @return {@code true} if the policy is attachable to the target entity; {@code false} otherwise
+   */
+  public static boolean canAttach(PolicyEntity policy, PolarisEntity targetEntity) {
+    Preconditions.checkNotNull(policy, "Policy must not be null");
+    Preconditions.checkNotNull(targetEntity, "Target entity must not be null");
+
+    var policyType = PredefinedPolicyTypes.fromCode(policy.getPolicyTypeCode());
+    Preconditions.checkArgument(
+        policyType != null, "Unknown policy type: " + policy.getPolicyTypeCode());
+
+    switch (policyType) {
+      case DATA_COMPACTION:
+        return DataCompactionPolicyValidator.INSTANCE.canAttach(
+            targetEntity.getType(), targetEntity.getSubType());
+      // To support additional policy types in the future, add cases here.
+      case METADATA_COMPACTION:
+      case SNAPSHOT_RETENTION:
+      case ORPHAN_FILE_REMOVAL:
+      default:
+        LOGGER.warn("Attachment not supported for policy type: {}", policyType.getName());
+        return false;
+    }
+  }
+}

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

@@ -0,0 +1,38 @@
+/*
+ * 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.validator;
+
+import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.databind.DeserializationContext;
+import com.fasterxml.jackson.databind.JsonDeserializer;
+import java.io.IOException;
+
+public class StrictBooleanDeserializer extends JsonDeserializer<Boolean> {
+  @Override
+  public Boolean deserialize(JsonParser p, DeserializationContext ctxt) throws IOException {
+    String text = p.getText();
+    if ("true".equals(text)) {
+      return Boolean.TRUE;
+    } else if ("false".equals(text)) {
+      return Boolean.FALSE;
+    } else {
+      throw new InvalidPolicyException("Invalid boolean value: " + text);
+    }
+  }
+}

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

@@ -0,0 +1,97 @@
+/*
+ * 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.validator.datacompaction;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import com.google.common.base.Strings;
+import java.util.Map;
+import java.util.Set;
+import org.apache.polaris.core.policy.PolicyContent;
+import org.apache.polaris.core.policy.validator.InvalidPolicyException;
+import org.apache.polaris.core.policy.validator.PolicyValidatorUtil;
+import org.apache.polaris.core.policy.validator.StrictBooleanDeserializer;
+
+public class DataCompactionPolicyContent implements PolicyContent {
+  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);
+
+  @JsonDeserialize(using = StrictBooleanDeserializer.class)
+  private Boolean enable;
+
+  private String version;
+  private Map<String, String> config;
+
+  @JsonCreator
+  public DataCompactionPolicyContent(
+      @JsonProperty(value = "enable", required = true) boolean enable) {
+    this.enable = enable;
+  }
+
+  public String getVersion() {
+    return version;
+  }
+
+  public void setVersion(String version) {
+    this.version = version;
+  }
+
+  public Boolean enabled() {
+    return enable;
+  }
+
+  public void setEnabled(Boolean enable) {
+    this.enable = enable;
+  }
+
+  public Map<String, String> getConfig() {
+    return config;
+  }
+
+  public void setConfig(Map<String, String> config) {
+    this.config = config;
+  }
+
+  public static DataCompactionPolicyContent fromString(String content) {
+    if (Strings.isNullOrEmpty(content)) {
+      throw new InvalidPolicyException("Policy is empty");
+    }
+
+    try {
+      DataCompactionPolicyContent policy =
+          PolicyValidatorUtil.MAPPER.readValue(content, DataCompactionPolicyContent.class);
+      if (policy == null) {
+        throw new InvalidPolicyException("Invalid policy");
+      }
+
+      if (Strings.isNullOrEmpty(policy.getVersion())) {
+        policy.setVersion(DEFAULT_POLICY_SCHEMA_VERSION);
+      }
+
+      if (!POLICY_SCHEMA_VERSIONS.contains(policy.getVersion())) {
+        throw new InvalidPolicyException("Invalid policy version: " + policy.getVersion());
+      }
+
+      return policy;
+    } catch (Exception e) {
+      throw new InvalidPolicyException(e);
+    }
+  }
+}