Make ENABLE_SUB_CATALOG_RBAC_FOR_FEDERATED_CATALOGS configurable per catalog (#2688)

* Update ENABLE_SUB_CATALOG_RBAC_FOR_FEDERATED_CATALOGS to be configurable per catalog

Author: HonahX

polaris-core/src/main/java/org/apache/polaris/core/config/FeatureConfiguration.java

@@ -277,6 +277,7 @@ public static void enforceFeatureEnabledOrThrow(
           .description(
               "When enabled, allows RBAC operations to create synthetic entities for"
                   + " entities in federated catalogs that don't exist in the local metastore.")
+          .catalogConfig("polaris.config.enable-sub-catalog-rbac-for-federated-catalogs")
           .defaultValue(false)
           .buildFeatureConfiguration();
 

runtime/service/src/main/java/org/apache/polaris/service/admin/PolarisAdminService.java

@@ -511,13 +511,18 @@ private void authorizeGrantOnTableLikeOperationOrThrow(
       }
     }
 
+    CatalogEntity catalogEntity =
+        CatalogEntity.of(
+            findCatalogByName(catalogName)
+                .orElseThrow(() -> new NotFoundException("Catalog %s not found", catalogName)));
     PolarisResolvedPathWrapper tableLikeWrapper =
         resolutionManifest.getResolvedPath(
             identifier, PolarisEntityType.TABLE_LIKE, PolarisEntitySubType.ANY_SUBTYPE, true);
     boolean rbacForFederatedCatalogsEnabled =
         getCurrentPolarisContext()
             .getRealmConfig()
-            .getConfig(FeatureConfiguration.ENABLE_SUB_CATALOG_RBAC_FOR_FEDERATED_CATALOGS);
+            .getConfig(
+                FeatureConfiguration.ENABLE_SUB_CATALOG_RBAC_FOR_FEDERATED_CATALOGS, catalogEntity);
     if (!(resolutionManifest.getIsPassthroughFacade() && rbacForFederatedCatalogsEnabled)
         && !subTypes.contains(tableLikeWrapper.getRawLeafEntity().getSubType())) {
       CatalogHandler.throwNotFoundExceptionForTableLikeEntity(identifier, subTypes);
@@ -1710,7 +1715,9 @@ public PrivilegeResult grantPrivilegeOnNamespaceToRole(
       boolean rbacForFederatedCatalogsEnabled =
           getCurrentPolarisContext()
               .getRealmConfig()
-              .getConfig(FeatureConfiguration.ENABLE_SUB_CATALOG_RBAC_FOR_FEDERATED_CATALOGS);
+              .getConfig(
+                  FeatureConfiguration.ENABLE_SUB_CATALOG_RBAC_FOR_FEDERATED_CATALOGS,
+                  catalogEntity);
       if (resolutionManifest.getIsPassthroughFacade() && rbacForFederatedCatalogsEnabled) {
         resolvedPathWrapper =
             createSyntheticNamespaceEntities(catalogEntity, namespace, resolvedPathWrapper);
@@ -2136,7 +2143,9 @@ private PrivilegeResult grantPrivilegeOnTableLikeToRole(
       boolean rbacForFederatedCatalogsEnabled =
           getCurrentPolarisContext()
               .getRealmConfig()
-              .getConfig(FeatureConfiguration.ENABLE_SUB_CATALOG_RBAC_FOR_FEDERATED_CATALOGS);
+              .getConfig(
+                  FeatureConfiguration.ENABLE_SUB_CATALOG_RBAC_FOR_FEDERATED_CATALOGS,
+                  catalogEntity);
       if (resolutionManifest.getIsPassthroughFacade() && rbacForFederatedCatalogsEnabled) {
         resolvedPathWrapper =
             createSyntheticTableLikeEntities(

runtime/service/src/test/java/org/apache/polaris/service/admin/PolarisAdminServiceTest.java

@@ -61,6 +61,7 @@
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.mockito.Mock;
+import org.mockito.Mockito;
 import org.mockito.MockitoAnnotations;
 
 public class PolarisAdminServiceTest {
@@ -90,6 +91,9 @@ void setUp() throws Exception {
     // Default feature configuration - enabled by default
     when(realmConfig.getConfig(FeatureConfiguration.ENABLE_SUB_CATALOG_RBAC_FOR_FEDERATED_CATALOGS))
         .thenReturn(true);
+    when(realmConfig.getConfig(
+            eq(FeatureConfiguration.ENABLE_SUB_CATALOG_RBAC_FOR_FEDERATED_CATALOGS), Mockito.any()))
+        .thenReturn(true);
 
     when(resolutionManifestFactory.createResolutionManifest(any(), any(), any()))
         .thenReturn(resolutionManifest);
@@ -358,6 +362,9 @@ void testGrantPrivilegeOnNamespaceToRole_PassthroughFacade_FeatureDisabled() thr
     // Disable the feature configuration
     when(realmConfig.getConfig(FeatureConfiguration.ENABLE_SUB_CATALOG_RBAC_FOR_FEDERATED_CATALOGS))
         .thenReturn(false);
+    when(realmConfig.getConfig(
+            eq(FeatureConfiguration.ENABLE_SUB_CATALOG_RBAC_FOR_FEDERATED_CATALOGS), Mockito.any()))
+        .thenReturn(false);
 
     PolarisEntity catalogEntity = createEntity(catalogName, PolarisEntityType.CATALOG);
     PolarisResolvedPathWrapper catalogWrapper = mock(PolarisResolvedPathWrapper.class);
@@ -522,6 +529,9 @@ void testGrantPrivilegeOnTableLikeToRole_PassthroughFacade_FeatureDisabled() thr
     // Disable the feature configuration
     when(realmConfig.getConfig(FeatureConfiguration.ENABLE_SUB_CATALOG_RBAC_FOR_FEDERATED_CATALOGS))
         .thenReturn(false);
+    when(realmConfig.getConfig(
+            eq(FeatureConfiguration.ENABLE_SUB_CATALOG_RBAC_FOR_FEDERATED_CATALOGS), Mockito.any()))
+        .thenReturn(false);
 
     PolarisEntity catalogEntity = createEntity(catalogName, PolarisEntityType.CATALOG);
     PolarisResolvedPathWrapper catalogWrapper = mock(PolarisResolvedPathWrapper.class);

runtime/service/src/test/java/org/apache/polaris/service/admin/PolarisAuthzTestBase.java

@@ -128,7 +128,6 @@ public Map<String, String> getConfigOverrides() {
           .put("polaris.features.\"DROP_WITH_PURGE_ENABLED\"", "true")
           .put("polaris.behavior-changes.\"ALLOW_NAMESPACE_CUSTOM_LOCATION\"", "true")
           .put("polaris.features.\"ENABLE_CATALOG_FEDERATION\"", "true")
-          .put("polaris.features.\"ENABLE_SUB_CATALOG_RBAC_FOR_FEDERATED_CATALOGS\"", "true")
           .build();
     }
   }
@@ -303,6 +302,7 @@ public void before(TestInfo testInfo) {
                 realmConfig,
                 storageConfigModelForFederatedCatalog,
                 storageLocationForFederatedCatalog)
+            .addProperty("polaris.config.enable-sub-catalog-rbac-for-federated-catalogs", "true")
             .build();
     ExternalCatalog externalCatalog =
         ExternalCatalog.builder()