[Draft][Do not Review] Make GenericTable a valid policy attachment target

integration-tests/src/main/java/org/apache/polaris/service/it/test/PolarisPolicyServiceIntegrationTest.java

@@ -64,6 +64,7 @@
 import org.apache.polaris.core.policy.exceptions.PolicyInUseException;
 import org.apache.polaris.service.it.env.CatalogConfig;
 import org.apache.polaris.service.it.env.ClientCredentials;
+import org.apache.polaris.service.it.env.GenericTableApi;
 import org.apache.polaris.service.it.env.IcebergHelper;
 import org.apache.polaris.service.it.env.IntegrationTestsHelper;
 import org.apache.polaris.service.it.env.ManagementApi;
@@ -108,7 +109,9 @@ public class PolarisPolicyServiceIntegrationTest {
   private static final PolicyIdentifier NS1_P1 = new PolicyIdentifier(NS1, "P1");
   private static final PolicyIdentifier NS1_P2 = new PolicyIdentifier(NS1, "P2");
   private static final PolicyIdentifier NS1_P3 = new PolicyIdentifier(NS1, "P3");
+  private static final PolicyIdentifier NS1_P4 = new PolicyIdentifier(NS1, "P4");
   private static final TableIdentifier NS2_T1 = TableIdentifier.of(NS2, "T1");
+  private static final TableIdentifier NS2_T2 = TableIdentifier.of(NS2, "T2");
 
   private static final String NS1_NAME = RESTUtil.encodeNamespace(NS1);
   private static final String INVALID_NAMESPACE = "INVALID_NAMESPACE";
@@ -124,6 +127,7 @@ public class PolarisPolicyServiceIntegrationTest {
   private static PolarisClient client;
   private static ManagementApi managementApi;
   private static PolicyApi policyApi;
+  private GenericTableApi genericTableApi;
 
   private RESTCatalog restCatalog;
   private String currentCatalogName;
@@ -221,6 +225,7 @@ public void before(TestInfo testInfo) {
         principalRoleName, currentCatalogName, catalogRole);
 
     policyApi = client.policyApi(principalToken);
+    genericTableApi = client.genericTableApi(principalToken);
   }
 
   @AfterEach
@@ -668,11 +673,19 @@ public void testPolicyMapping() {
             PredefinedPolicyTypes.ORPHAN_FILE_REMOVAL,
             EXAMPLE_TABLE_MAINTENANCE_POLICY_CONTENT,
             "test policy");
+    Policy p4 =
+        policyApi.createPolicy(
+            currentCatalogName,
+            NS1_P4,
+            PredefinedPolicyTypes.TABLE_CONVERSION,
+            "{\"enable\": false}",
+            "test policy");
 
     restCatalog
         .buildTable(
             NS2_T1, new Schema(Types.NestedField.optional(1, "string", Types.StringType.get())))
         .create();
+    genericTableApi.createGenericTable(currentCatalogName, NS2_T2, "delta", Map.of());
 
     PolicyAttachmentTarget catalogTarget =
         PolicyAttachmentTarget.builder().setType(PolicyAttachmentTarget.TypeEnum.CATALOG).build();
@@ -686,10 +699,16 @@ NS2_T1, new Schema(Types.NestedField.optional(1, "string", Types.StringType.get(
             .setType(PolicyAttachmentTarget.TypeEnum.TABLE_LIKE)
             .setPath(PolarisCatalogHelpers.tableIdentifierToList(NS2_T1))
             .build();
+    PolicyAttachmentTarget genericTableTarget =
+        PolicyAttachmentTarget.builder()
+            .setType(PolicyAttachmentTarget.TypeEnum.TABLE_LIKE)
+            .setPath(PolarisCatalogHelpers.tableIdentifierToList(NS2_T2))
+            .build();
 
     policyApi.attachPolicy(currentCatalogName, NS1_P1, catalogTarget, Map.of());
     policyApi.attachPolicy(currentCatalogName, NS1_P2, namespaceTarget, Map.of());
     policyApi.attachPolicy(currentCatalogName, NS1_P3, tableTarget, Map.of());
+    policyApi.attachPolicy(currentCatalogName, NS1_P4, genericTableTarget, Map.of());
 
     List<ApplicablePolicy> applicablePoliciesOnCatalog =
         policyApi.getApplicablePolicies(currentCatalogName, null, null, null);
@@ -715,15 +734,22 @@ NS2_T1, new Schema(Types.NestedField.optional(1, "string", Types.StringType.get(
                 currentCatalogName, NS2, NS2_T1.name(), PredefinedPolicyTypes.METADATA_COMPACTION))
         .containsExactlyInAnyOrder(policyToApplicablePolicy(p2, true, NS1));
 
+    Assertions.assertThat(
+            policyApi.getApplicablePolicies(currentCatalogName, NS2, NS2_T2.name(), null))
+        .containsExactlyInAnyOrder(policyToApplicablePolicy(p4, false, NS1));
+
     policyApi.detachPolicy(currentCatalogName, NS1_P1, catalogTarget);
     policyApi.detachPolicy(currentCatalogName, NS1_P2, namespaceTarget);
     policyApi.detachPolicy(currentCatalogName, NS1_P3, tableTarget);
+    policyApi.detachPolicy(currentCatalogName, NS1_P4, genericTableTarget);
 
     policyApi.dropPolicy(currentCatalogName, NS1_P1);
     policyApi.dropPolicy(currentCatalogName, NS1_P2);
     policyApi.dropPolicy(currentCatalogName, NS1_P3);
+    policyApi.dropPolicy(currentCatalogName, NS1_P4);
 
     restCatalog.dropTable(NS2_T1);
+    genericTableApi.dropGenericTable(currentCatalogName, NS2_T2);
   }
 
   @Test

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),
+  TABLE_CONVERSION(4, "system.table-conversion", true);
 
   private final int code;
   private final String name;

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

@@ -0,0 +1,106 @@
+/*
+ * 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.conversion;
+
+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.content.PolicyContentUtil;
+import org.apache.polaris.core.policy.content.StrictBooleanDeserializer;
+import org.apache.polaris.core.policy.validator.InvalidPolicyException;
+
+public class TableConversionPolicyContent {
+  private static final String DEFAULT_POLICY_SCHEMA_VERSION = "2025-09-09";
+  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 TableConversionPolicyContent(
+      @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;
+  }
+
+  static void validateVersion(
+      String content,
+      TableConversionPolicyContent policy,
+      String defaultVersion,
+      Set<String> allVersions) {
+    if (policy == null) {
+      throw new InvalidPolicyException("Invalid policy: " + content);
+    }
+
+    if (Strings.isNullOrEmpty(policy.getVersion())) {
+      policy.setVersion(defaultVersion);
+    }
+
+    if (!allVersions.contains(policy.getVersion())) {
+      throw new InvalidPolicyException("Invalid policy version: " + policy.getVersion());
+    }
+  }
+
+  public static TableConversionPolicyContent fromString(String content) {
+    if (Strings.isNullOrEmpty(content)) {
+      throw new InvalidPolicyException("Policy is empty");
+    }
+
+    TableConversionPolicyContent policy;
+    try {
+      policy = PolicyContentUtil.MAPPER.readValue(content, TableConversionPolicyContent.class);
+    } catch (Exception e) {
+      throw new InvalidPolicyException(e);
+    }
+
+    validateVersion(content, policy, DEFAULT_POLICY_SCHEMA_VERSION, POLICY_SCHEMA_VERSIONS);
+
+    return policy;
+  }
+}

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

@@ -22,11 +22,13 @@
 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.conversion.TableConversionPolicyContent;
 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;
 import org.apache.polaris.core.policy.content.maintenance.SnapshotExpiryPolicyContent;
 import org.apache.polaris.core.policy.exceptions.PolicyAttachException;
+import org.apache.polaris.core.policy.validator.conversion.TableConversionPolicyValidator;
 import org.apache.polaris.core.policy.validator.maintenance.BaseMaintenancePolicyValidator;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -66,6 +68,9 @@ public static void validate(PolicyEntity policy) {
       case ORPHAN_FILE_REMOVAL:
         OrphanFileRemovalPolicyContent.fromString(policy.getContent());
         break;
+      case TABLE_CONVERSION:
+        TableConversionPolicyContent.fromString(policy.getContent());
+        break;
       default:
         throw new IllegalArgumentException("Unsupported policy type: " + type.getName());
     }
@@ -97,7 +102,8 @@ public static boolean canAttach(PolicyEntity policy, PolarisEntity targetEntity)
       case SNAPSHOT_EXPIRY:
       case ORPHAN_FILE_REMOVAL:
         return BaseMaintenancePolicyValidator.INSTANCE.canAttach(entityType, entitySubType);
-
+      case TABLE_CONVERSION:
+        return TableConversionPolicyValidator.INSTANCE.canAttach(entityType, entitySubType);
       default:
         LOGGER.warn("Attachment not supported for policy type: {}", policyType.getName());
         return false;

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

@@ -0,0 +1,42 @@
+/*
+ * 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.conversion;
+
+import org.apache.polaris.core.entity.PolarisEntitySubType;
+import org.apache.polaris.core.entity.PolarisEntityType;
+import org.apache.polaris.core.policy.PolicyMappingUtil;
+import org.apache.polaris.core.policy.validator.InvalidPolicyException;
+import org.apache.polaris.core.policy.validator.PolicyValidator;
+
+public class TableConversionPolicyValidator implements PolicyValidator {
+  public static final TableConversionPolicyValidator INSTANCE =
+      new TableConversionPolicyValidator();
+
+  @Override
+  public void validate(String content) throws InvalidPolicyException {}
+
+  @Override
+  public boolean canAttach(PolarisEntityType entityType, PolarisEntitySubType entitySubType) {
+    // TODO: refactor this, the util method should be more general
+    return PolicyMappingUtil.isValidTargetEntityType(entityType, entitySubType)
+        || (entityType == PolarisEntityType.TABLE_LIKE
+            && entitySubType == PolarisEntitySubType.GENERIC_TABLE);
+  }
+}

polaris-core/src/test/java/org/apache/polaris/core/policy/PolicyTypeTest.java

@@ -34,7 +34,8 @@ static Stream<Arguments> predefinedPolicyTypes() {
         Arguments.of(0, "system.data-compaction", true),
         Arguments.of(1, "system.metadata-compaction", true),
         Arguments.of(2, "system.orphan-file-removal", true),
-        Arguments.of(3, "system.snapshot-expiry", true));
+        Arguments.of(3, "system.snapshot-expiry", true),
+        Arguments.of(4, "system.table-conversion", true));
   }
 
   @ParameterizedTest

runtime/service/src/main/java/org/apache/polaris/service/catalog/common/CatalogHandler.java

@@ -242,6 +242,13 @@ protected void authorizeCreateTableLikeUnderNamespaceOperationOrThrow(
 
   protected void authorizeBasicTableLikeOperationOrThrow(
       PolarisAuthorizableOperation op, PolarisEntitySubType subType, TableIdentifier identifier) {
+    authorizeBasicTableLikeOperationOrThrow(op, List.of(subType), identifier);
+  }
+
+  protected void authorizeBasicTableLikeOperationOrThrow(
+      PolarisAuthorizableOperation op,
+      List<PolarisEntitySubType> subTypes,
+      TableIdentifier identifier) {
     resolutionManifest =
         resolutionManifestFactory.createResolutionManifest(
             callContext, securityContext, catalogName);
@@ -254,10 +261,17 @@ protected void authorizeBasicTableLikeOperationOrThrow(
             true /* optional */),
         identifier);
     resolutionManifest.resolveAll();
-    PolarisResolvedPathWrapper target =
-        resolutionManifest.getResolvedPath(identifier, PolarisEntityType.TABLE_LIKE, subType, true);
+    PolarisResolvedPathWrapper target = null;
+    for (PolarisEntitySubType subType : subTypes) {
+      target =
+          resolutionManifest.getResolvedPath(
+              identifier, PolarisEntityType.TABLE_LIKE, subType, true);
+      if (target != null) {
+        break;
+      }
+    }
     if (target == null) {
-      throwNotFoundExceptionForTableLikeEntity(identifier, List.of(subType));
+      throwNotFoundExceptionForTableLikeEntity(identifier, subTypes);
     }
     authorizer.authorizeOrThrow(
         polarisPrincipal,