Allow retrieving a config directly from a Map

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

@@ -107,6 +107,10 @@ public interface PolarisConfigurationStore {
    * Retrieve the current value for a configuration, overriding with a catalog config if it is
    * present.
    *
+   * <p>Prefer using {@link #getConfiguration(RealmContext, Map, PolarisConfiguration)} when the
+   * catalog properties are already available or are needed repeatedly to prevent unnecessary JSON
+   * deserialization.
+   *
    * @param realmContext the current realm context
    * @param catalogEntity the catalog to check for an override
    * @param config the configuration to load
@@ -117,15 +121,31 @@ public interface PolarisConfigurationStore {
       @Nonnull RealmContext realmContext,
       @Nonnull CatalogEntity catalogEntity,
       PolarisConfiguration<T> config) {
+    return getConfiguration(realmContext, catalogEntity.getPropertiesAsMap(), config);
+  }
+
+  /**
+   * Retrieve the current value for a configuration, overriding with a catalog config if it is
+   * present.
+   *
+   * @param realmContext the current realm context
+   * @param catalogProperties the catalog configuration to check for an override
+   * @param config the configuration to load
+   * @return the current value set for the configuration key or null if not set
+   * @param <T> the type of the configuration value
+   */
+  default <T> @Nonnull T getConfiguration(
+      @Nonnull RealmContext realmContext,
+      @Nonnull Map<String, String> catalogProperties,
+      PolarisConfiguration<T> config) {
     if (config.hasCatalogConfig() || config.hasCatalogConfigUnsafe()) {
-      Map<String, String> propertiesMap = catalogEntity.getPropertiesAsMap();
       String propertyValue = null;
       if (config.hasCatalogConfig()) {
-        propertyValue = propertiesMap.get(config.catalogConfig());
+        propertyValue = catalogProperties.get(config.catalogConfig());
       }
       if (propertyValue == null) {
         if (config.hasCatalogConfigUnsafe()) {
-          propertyValue = propertiesMap.get(config.catalogConfigUnsafe());
+          propertyValue = catalogProperties.get(config.catalogConfigUnsafe());
         }
         if (propertyValue != null) {
           LOGGER.warn(

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

@@ -19,6 +19,7 @@
 package org.apache.polaris.core.config;
 
 import jakarta.annotation.Nullable;
+import java.util.Map;
 import org.apache.polaris.core.entity.CatalogEntity;
 
 /** Realm-specific configuration used to retrieve runtime parameters. */
@@ -57,10 +58,24 @@ public interface RealmConfig {
    * Retrieve the current value for a configuration, overriding with a catalog config if it is
    * present.
    *
+   * <p>Prefer using {@link #getConfig(PolarisConfiguration, Map)} when the catalog properties are
+   * already available or are needed repeatedly to prevent unnecessary JSON deserialization.
+   *
    * @param <T> the type of the configuration value
    * @param config the configuration to load
    * @param catalogEntity the catalog to check for an override
    * @return the current value set for the configuration key or null if not set
    */
   <T> T getConfig(PolarisConfiguration<T> config, CatalogEntity catalogEntity);
+
+  /**
+   * Retrieve the current value for a configuration, overriding with a catalog config if it is
+   * present.
+   *
+   * @param <T> the type of the configuration value
+   * @param config the configuration to load
+   * @param catalogProperties the catalog configuration to check for an override
+   * @return the current value set for the configuration key or null if not set
+   */
+  <T> T getConfig(PolarisConfiguration<T> config, Map<String, String> catalogProperties);
 }

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

@@ -19,6 +19,7 @@
 package org.apache.polaris.core.config;
 
 import jakarta.annotation.Nullable;
+import java.util.Map;
 import org.apache.polaris.core.context.RealmContext;
 import org.apache.polaris.core.entity.CatalogEntity;
 
@@ -51,4 +52,9 @@ public <T> T getConfig(PolarisConfiguration<T> config) {
   public <T> T getConfig(PolarisConfiguration<T> config, CatalogEntity catalogEntity) {
     return configurationStore.getConfiguration(realmContext, catalogEntity, config);
   }
+
+  @Override
+  public <T> T getConfig(PolarisConfiguration<T> config, Map<String, String> catalogProperties) {
+    return configurationStore.getConfiguration(realmContext, catalogProperties, config);
+  }
 }

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

@@ -89,7 +89,8 @@ void setUp() throws Exception {
     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()))
+            eq(FeatureConfiguration.ENABLE_SUB_CATALOG_RBAC_FOR_FEDERATED_CATALOGS),
+            (CatalogEntity) Mockito.any()))
         .thenReturn(true);
 
     when(resolutionManifestFactory.createResolutionManifest(any(), any()))
@@ -353,7 +354,8 @@ void testGrantPrivilegeOnNamespaceToRole_PassthroughFacade_FeatureDisabled() thr
     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()))
+            eq(FeatureConfiguration.ENABLE_SUB_CATALOG_RBAC_FOR_FEDERATED_CATALOGS),
+            (CatalogEntity) Mockito.any()))
         .thenReturn(false);
 
     PolarisEntity catalogEntity = createEntity(catalogName, PolarisEntityType.CATALOG);
@@ -516,7 +518,8 @@ void testGrantPrivilegeOnTableLikeToRole_PassthroughFacade_FeatureDisabled() thr
     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()))
+            eq(FeatureConfiguration.ENABLE_SUB_CATALOG_RBAC_FOR_FEDERATED_CATALOGS),
+            (CatalogEntity) Mockito.any()))
         .thenReturn(false);
 
     PolarisEntity catalogEntity = createEntity(catalogName, PolarisEntityType.CATALOG);

runtime/service/src/test/java/org/apache/polaris/service/catalog/iceberg/IcebergCatalogHandlerAuthzTest.java

@@ -1164,6 +1164,17 @@ public <T> T getConfig(PolarisConfiguration<T> config, CatalogEntity catalogEnti
             }
             return realmConfig.getConfig(config, catalogEntity);
           }
+
+          @Override
+          @SuppressWarnings("unchecked")
+          public <T> T getConfig(
+              PolarisConfiguration<T> config, Map<String, String> catalogProperties) {
+            // Override the specific configuration we want to test
+            if (config.equals(FeatureConfiguration.ENABLE_FINE_GRAINED_UPDATE_TABLE_PRIVILEGES)) {
+              return (T) Boolean.valueOf(fineGrainedAuthzEnabled);
+            }
+            return realmConfig.getConfig(config, catalogProperties);
+          }
         };
 
     // Mock the regular CallContext calls