Add change log and a new feature flag

Author: HonahX

CHANGELOG.md

@@ -35,6 +35,8 @@ request adding CHANGELOG notes for breaking (!) changes and possibly other secti
 
 ### New Features
 
+- Support credential vending for federated catalogs. `ALLOW_FEDERATED_CATALOGS_CREDENTIAL_VENDING` (default: true) was added to toggle this feature.
+
 ### Changes
 
 ### Deprecations

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

@@ -429,4 +429,13 @@ public static void enforceFeatureEnabledOrThrow(
               "When true, enables finer grained update table privileges which are passed to the authorizer for update table operations")
           .defaultValue(true)
           .buildFeatureConfiguration();
+
+  public static final FeatureConfiguration<Boolean> ALLOW_FEDERATED_CATALOGS_CREDENTIAL_VENDING =
+      PolarisConfiguration.<Boolean>builder()
+          .key("ALLOW_FEDERATED_CATALOGS_CREDENTIAL_VENDING")
+          .catalogConfig("polaris.config.allow-federated-catalogs-credential-vending")
+          .description(
+              "If set to true (default), allow credential vending for external catalogs. Note this requires ALLOW_EXTERNAL_CATALOG_CREDENTIAL_VENDING to be true first.")
+          .defaultValue(true)
+          .buildFeatureConfiguration();
 }

runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/IcebergCatalogHandler.java

@@ -18,7 +18,7 @@
  */
 package org.apache.polaris.service.catalog.iceberg;
 
-import static org.apache.polaris.core.config.FeatureConfiguration.ALLOW_EXTERNAL_CATALOG_CREDENTIAL_VENDING;
+import static org.apache.polaris.core.config.FeatureConfiguration.ALLOW_FEDERATED_CATALOGS_CREDENTIAL_VENDING;
 import static org.apache.polaris.core.config.FeatureConfiguration.LIST_PAGINATION_ENABLED;
 import static org.apache.polaris.service.catalog.AccessDelegationMode.VENDED_CREDENTIALS;
 
@@ -426,6 +426,12 @@ public void authorizeCreateTableDirect(
           PolarisAuthorizableOperation.CREATE_TABLE_DIRECT_WITH_WRITE_DELEGATION,
           TableIdentifier.of(namespace, request.name()));
     }
+
+    CatalogEntity catalog = getResolvedCatalogEntity();
+    if (catalog.isStaticFacade()) {
+      throw new BadRequestException("Cannot create table on static-facade external catalogs.");
+    }
+    checkAllowExternalCatalogCredentialVending(delegationModes);
   }
 
   public LoadTableResponse createTableDirect(
@@ -436,10 +442,6 @@ public LoadTableResponse createTableDirect(
 
     authorizeCreateTableDirect(namespace, request, delegationModes);
 
-    CatalogEntity catalog = getResolvedCatalogEntity();
-    if (catalog.isStaticFacade()) {
-      throw new BadRequestException("Cannot create table on static-facade external catalogs.");
-    }
     request.validate();
 
     TableIdentifier tableIdentifier = TableIdentifier.of(namespace, request.name());
@@ -550,6 +552,12 @@ private void authorizeCreateTableStaged(
           PolarisAuthorizableOperation.CREATE_TABLE_STAGED_WITH_WRITE_DELEGATION,
           TableIdentifier.of(namespace, request.name()));
     }
+
+    CatalogEntity catalog = getResolvedCatalogEntity();
+    if (catalog.isStaticFacade()) {
+      throw new BadRequestException("Cannot create table on static-facade external catalogs.");
+    }
+    checkAllowExternalCatalogCredentialVending(delegationModes);
   }
 
   public LoadTableResponse createTableStaged(
@@ -560,10 +568,6 @@ public LoadTableResponse createTableStaged(
 
     authorizeCreateTableStaged(namespace, request, delegationModes);
 
-    CatalogEntity catalog = getResolvedCatalogEntity();
-    if (catalog.isStaticFacade()) {
-      throw new BadRequestException("Cannot create table on static-facade external catalogs.");
-    }
     TableIdentifier ident = TableIdentifier.of(namespace, request.name());
     TableMetadata metadata = stageTableCreateHelper(namespace, request);
 
@@ -728,23 +732,7 @@ private Set<PolarisStorageActions> authorizeLoadTable(
           read, PolarisEntitySubType.ICEBERG_TABLE, tableIdentifier);
     }
 
-    CatalogEntity catalogEntity = getResolvedCatalogEntity();
-
-    LOGGER.info("Catalog type: {}", catalogEntity.getCatalogType());
-    LOGGER.info(
-        "allow external catalog credential vending: {}",
-        realmConfig.getConfig(
-            FeatureConfiguration.ALLOW_EXTERNAL_CATALOG_CREDENTIAL_VENDING, catalogEntity));
-    if (catalogEntity
-            .getCatalogType()
-            .equals(org.apache.polaris.core.admin.model.Catalog.TypeEnum.EXTERNAL)
-        && !realmConfig.getConfig(
-            FeatureConfiguration.ALLOW_EXTERNAL_CATALOG_CREDENTIAL_VENDING, catalogEntity)) {
-      throw new ForbiddenException(
-          "Access Delegation is not enabled for this catalog. Please consult applicable "
-              + "documentation for the catalog config property '%s' to enable this feature",
-          FeatureConfiguration.ALLOW_EXTERNAL_CATALOG_CREDENTIAL_VENDING.catalogConfig());
-    }
+    checkAllowExternalCatalogCredentialVending(delegationModes);
 
     return actionsRequested;
   }
@@ -814,13 +802,14 @@ private LoadTableResponse.Builder buildLoadTableResponseWithDelegationCredential
         CatalogUtils.findResolvedStorageEntity(resolutionManifest, tableIdentifier);
 
     if (resolvedStoragePath == null) {
-        LOGGER.debug("Unable to find storage configuration information for table {}", tableIdentifier);
+      LOGGER.debug(
+          "Unable to find storage configuration information for table {}", tableIdentifier);
       return responseBuilder;
     }
 
     if (baseCatalog instanceof IcebergCatalog
         || realmConfig.getConfig(
-            ALLOW_EXTERNAL_CATALOG_CREDENTIAL_VENDING, getResolvedCatalogEntity())) {
+            ALLOW_FEDERATED_CATALOGS_CREDENTIAL_VENDING, getResolvedCatalogEntity())) {
       AccessConfig accessConfig =
           accessConfigProvider.getAccessConfig(
               callContext,
@@ -1223,6 +1212,31 @@ private EnumSet<PolarisAuthorizableOperation> getUpdateTableAuthorizableOperatio
     }
   }
 
+  private void checkAllowExternalCatalogCredentialVending(
+      EnumSet<AccessDelegationMode> delegationModes) {
+
+    if (delegationModes.isEmpty()) {
+      return;
+    }
+    CatalogEntity catalogEntity = getResolvedCatalogEntity();
+
+    LOGGER.info("Catalog type: {}", catalogEntity.getCatalogType());
+    LOGGER.info(
+        "allow external catalog credential vending: {}",
+        realmConfig.getConfig(
+            FeatureConfiguration.ALLOW_EXTERNAL_CATALOG_CREDENTIAL_VENDING, catalogEntity));
+    if (catalogEntity
+            .getCatalogType()
+            .equals(org.apache.polaris.core.admin.model.Catalog.TypeEnum.EXTERNAL)
+        && !realmConfig.getConfig(
+            FeatureConfiguration.ALLOW_EXTERNAL_CATALOG_CREDENTIAL_VENDING, catalogEntity)) {
+      throw new ForbiddenException(
+          "Access Delegation is not enabled for this catalog. Please consult applicable "
+              + "documentation for the catalog config property '%s' to enable this feature",
+          FeatureConfiguration.ALLOW_EXTERNAL_CATALOG_CREDENTIAL_VENDING.catalogConfig());
+    }
+  }
+
   @Override
   public void close() throws Exception {
     if (baseCatalog instanceof Closeable closeable) {

runtime/spark-tests/src/intTest/java/org/apache/polaris/service/spark/it/CatalogFederationIT.java

@@ -40,6 +40,7 @@ public Map<String, String> getConfigOverrides() {
           .put("polaris.features.\"ALLOW_DROPPING_NON_EMPTY_PASSTHROUGH_FACADE_CATALOG\"", "true")
           .put("polaris.features.\"SKIP_CREDENTIAL_SUBSCOPING_INDIRECTION\"", "false")
           .put("polaris.features.\"ALLOW_EXTERNAL_CATALOG_CREDENTIAL_VENDING\"", "true")
+          .put("polaris.features.\"ALLOW_FEDERATED_CATALOGS_CREDENTIAL_VENDING\"", "true")
           .put("polaris.storage.aws.access-key", CatalogFederationIntegrationTest.MINIO_ACCESS_KEY)
           .put("polaris.storage.aws.secret-key", CatalogFederationIntegrationTest.MINIO_SECRET_KEY)
           .build();