Introduce RealmConfig (#2015)

Getting a config value currently requires quite a ceremony:
```
ctx.getPolarisCallContext()
   .getConfigurationStore()
   .getConfiguration(ctx.getRealmContext(), "ALLOW_WILDCARD_LOCATION", false))
```
since a `PolarisConfigurationStore` cant be used without a `RealmContext` it makes sense to add a dedicated interface. this allows removal of verbose code and also moves towards injecting that interface via CDI at a request/realm scope in the future.

Author: XN137

persistence/eclipselink/src/main/java/org/apache/polaris/extension/persistence/impl/eclipselink/EclipseLinkPolarisMetaStoreManagerFactory.java

@@ -25,7 +25,6 @@
 import jakarta.inject.Inject;
 import java.nio.file.Path;
 import org.apache.polaris.core.PolarisDiagnostics;
-import org.apache.polaris.core.config.PolarisConfigurationStore;
 import org.apache.polaris.core.context.RealmContext;
 import org.apache.polaris.core.persistence.LocalPolarisMetaStoreManagerFactory;
 import org.apache.polaris.core.persistence.PolarisMetaStoreManager;
@@ -47,13 +46,12 @@ public class EclipseLinkPolarisMetaStoreManagerFactory
   @Inject PolarisStorageIntegrationProvider storageIntegrationProvider;
 
   protected EclipseLinkPolarisMetaStoreManagerFactory() {
-    this(null, null);
+    this(null);
   }
 
   @Inject
-  protected EclipseLinkPolarisMetaStoreManagerFactory(
-      PolarisDiagnostics diagnostics, PolarisConfigurationStore configurationStore) {
-    super(diagnostics, configurationStore);
+  protected EclipseLinkPolarisMetaStoreManagerFactory(PolarisDiagnostics diagnostics) {
+    super(diagnostics);
   }
 
   @Override

persistence/relational-jdbc/src/main/java/org/apache/polaris/persistence/relational/jdbc/JdbcMetaStoreManagerFactory.java

@@ -31,7 +31,7 @@
 import org.apache.polaris.core.PolarisCallContext;
 import org.apache.polaris.core.PolarisDefaultDiagServiceImpl;
 import org.apache.polaris.core.PolarisDiagnostics;
-import org.apache.polaris.core.config.PolarisConfigurationStore;
+import org.apache.polaris.core.config.RealmConfig;
 import org.apache.polaris.core.context.CallContext;
 import org.apache.polaris.core.context.RealmContext;
 import org.apache.polaris.core.entity.PolarisEntity;
@@ -78,7 +78,6 @@ public class JdbcMetaStoreManagerFactory implements MetaStoreManagerFactory {
   @Inject PolarisStorageIntegrationProvider storageIntegrationProvider;
   @Inject Instance<DataSource> dataSource;
   @Inject RelationalJdbcConfiguration relationalJdbcConfiguration;
-  @Inject PolarisConfigurationStore configurationStore;
 
   protected JdbcMetaStoreManagerFactory() {}
 
@@ -221,22 +220,23 @@ public synchronized Supplier<BasePersistence> getOrCreateSessionSupplier(
 
   @Override
   public synchronized StorageCredentialCache getOrCreateStorageCredentialCache(
-      RealmContext realmContext) {
+      RealmContext realmContext, RealmConfig realmConfig) {
     if (!storageCredentialCacheMap.containsKey(realmContext.getRealmIdentifier())) {
       storageCredentialCacheMap.put(
-          realmContext.getRealmIdentifier(), new StorageCredentialCache(configurationStore));
+          realmContext.getRealmIdentifier(), new StorageCredentialCache());
     }
 
     return storageCredentialCacheMap.get(realmContext.getRealmIdentifier());
   }
 
   @Override
-  public synchronized EntityCache getOrCreateEntityCache(RealmContext realmContext) {
+  public synchronized EntityCache getOrCreateEntityCache(
+      RealmContext realmContext, RealmConfig realmConfig) {
     if (!entityCacheMap.containsKey(realmContext.getRealmIdentifier())) {
       PolarisMetaStoreManager metaStoreManager = getOrCreateMetaStoreManager(realmContext);
       entityCacheMap.put(
           realmContext.getRealmIdentifier(),
-          new InMemoryEntityCache(realmContext, configurationStore, metaStoreManager));
+          new InMemoryEntityCache(realmConfig, metaStoreManager));
     }
 
     return entityCacheMap.get(realmContext.getRealmIdentifier());

polaris-core/src/main/java/org/apache/polaris/core/PolarisCallContext.java

@@ -22,6 +22,8 @@
 import java.time.Clock;
 import java.time.ZoneId;
 import org.apache.polaris.core.config.PolarisConfigurationStore;
+import org.apache.polaris.core.config.RealmConfig;
+import org.apache.polaris.core.config.RealmConfigImpl;
 import org.apache.polaris.core.context.CallContext;
 import org.apache.polaris.core.context.RealmContext;
 import org.apache.polaris.core.persistence.BasePersistence;
@@ -42,8 +44,9 @@ public class PolarisCallContext implements CallContext {
 
   private final Clock clock;
 
-  // will make it final once we remove deprecated constructor
-  private RealmContext realmContext = null;
+  private final RealmContext realmContext;
+
+  private final RealmConfig realmConfig;
 
   public PolarisCallContext(
       @Nonnull RealmContext realmContext,
@@ -56,17 +59,19 @@ public PolarisCallContext(
     this.diagServices = diagServices;
     this.configurationStore = configurationStore;
     this.clock = clock;
+    this.realmConfig = new RealmConfigImpl(this.configurationStore, this.realmContext);
   }
 
   public PolarisCallContext(
       @Nonnull RealmContext realmContext,
       @Nonnull BasePersistence metaStore,
       @Nonnull PolarisDiagnostics diagServices) {
-    this.realmContext = realmContext;
-    this.metaStore = metaStore;
-    this.diagServices = diagServices;
-    this.configurationStore = new PolarisConfigurationStore() {};
-    this.clock = Clock.system(ZoneId.systemDefault());
+    this(
+        realmContext,
+        metaStore,
+        diagServices,
+        new PolarisConfigurationStore() {},
+        Clock.system(ZoneId.systemDefault()));
   }
 
   public BasePersistence getMetaStore() {
@@ -77,10 +82,6 @@ public PolarisDiagnostics getDiagServices() {
     return diagServices;
   }
 
-  public PolarisConfigurationStore getConfigurationStore() {
-    return configurationStore;
-  }
-
   public Clock getClock() {
     return clock;
   }
@@ -90,6 +91,11 @@ public RealmContext getRealmContext() {
     return realmContext;
   }
 
+  @Override
+  public RealmConfig getRealmConfig() {
+    return realmConfig;
+  }
+
   @Override
   public PolarisCallContext getPolarisCallContext() {
     return this;

polaris-core/src/main/java/org/apache/polaris/core/auth/PolarisAuthorizerImpl.java

@@ -114,7 +114,6 @@
 import java.util.stream.Collectors;
 import org.apache.iceberg.exceptions.ForbiddenException;
 import org.apache.polaris.core.config.FeatureConfiguration;
-import org.apache.polaris.core.config.PolarisConfigurationStore;
 import org.apache.polaris.core.context.CallContext;
 import org.apache.polaris.core.entity.PolarisBaseEntity;
 import org.apache.polaris.core.entity.PolarisEntityConstants;
@@ -531,12 +530,8 @@ public class PolarisAuthorizerImpl implements PolarisAuthorizer {
         List.of(TABLE_DETACH_POLICY, CATALOG_MANAGE_METADATA, CATALOG_MANAGE_CONTENT));
   }
 
-  private final PolarisConfigurationStore featureConfig;
-
   @Inject
-  public PolarisAuthorizerImpl(PolarisConfigurationStore featureConfig) {
-    this.featureConfig = featureConfig;
-  }
+  public PolarisAuthorizerImpl() {}
 
   /**
    * Checks whether the {@code grantedPrivilege} is sufficient to confer {@code desiredPrivilege},
@@ -583,9 +578,10 @@ public void authorizeOrThrow(
       @Nullable List<PolarisResolvedPathWrapper> targets,
       @Nullable List<PolarisResolvedPathWrapper> secondaries) {
     boolean enforceCredentialRotationRequiredState =
-        featureConfig.getConfiguration(
-            callContext.getRealmContext(),
-            FeatureConfiguration.ENFORCE_PRINCIPAL_CREDENTIAL_ROTATION_REQUIRED_CHECKING);
+        callContext
+            .getRealmConfig()
+            .getConfig(
+                FeatureConfiguration.ENFORCE_PRINCIPAL_CREDENTIAL_ROTATION_REQUIRED_CHECKING);
     if (enforceCredentialRotationRequiredState
         && authenticatedPrincipal
             .getPrincipalEntity()

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

@@ -49,11 +49,7 @@ protected FeatureConfiguration(
    */
   public static void enforceFeatureEnabledOrThrow(
       CallContext callContext, FeatureConfiguration<Boolean> featureConfig) {
-    boolean enabled =
-        callContext
-            .getPolarisCallContext()
-            .getConfigurationStore()
-            .getConfiguration(callContext.getRealmContext(), featureConfig);
+    boolean enabled = callContext.getRealmConfig().getConfig(featureConfig);
     if (!enabled) {
       throw new UnsupportedOperationException("Feature not enabled: " + featureConfig.key());
     }

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

@@ -0,0 +1,66 @@
+/*
+ * 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.config;
+
+import jakarta.annotation.Nullable;
+import org.apache.polaris.core.entity.CatalogEntity;
+
+/** Realm-specific configuration used to retrieve runtime parameters. */
+public interface RealmConfig {
+
+  /**
+   * Retrieve the current value for a configuration key. May be null if not set.
+   *
+   * @param <T> the type of the configuration value
+   * @param configName the name of the configuration key to check
+   * @return the current value set for the configuration key, or null if not set
+   */
+  <T> @Nullable T getConfig(String configName);
+
+  /**
+   * Retrieve the current value for a configuration key. If not set, return the non-null default
+   * value.
+   *
+   * @param <T> the type of the configuration value
+   * @param configName the name of the configuration key to check
+   * @param defaultValue the default value if the configuration key has no value
+   * @return the current value or the supplied default value
+   */
+  <T> T getConfig(String configName, T defaultValue);
+
+  /**
+   * Retrieve the current value for a configuration.
+   *
+   * @param <T> the type of the configuration value
+   * @param config the configuration to load
+   * @return the current value set for the configuration key or null if not set
+   */
+  <T> T getConfig(PolarisConfiguration<T> config);
+
+  /**
+   * 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 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);
+}

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

@@ -0,0 +1,54 @@
+/*
+ * 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.config;
+
+import jakarta.annotation.Nullable;
+import org.apache.polaris.core.context.RealmContext;
+import org.apache.polaris.core.entity.CatalogEntity;
+
+public class RealmConfigImpl implements RealmConfig {
+
+  private final PolarisConfigurationStore configurationStore;
+  private final RealmContext realmContext;
+
+  public RealmConfigImpl(PolarisConfigurationStore configurationStore, RealmContext realmContext) {
+    this.configurationStore = configurationStore;
+    this.realmContext = realmContext;
+  }
+
+  @Override
+  public <T> @Nullable T getConfig(String configName) {
+    return configurationStore.getConfiguration(realmContext, configName);
+  }
+
+  @Override
+  public <T> T getConfig(String configName, T defaultValue) {
+    return configurationStore.getConfiguration(realmContext, configName, defaultValue);
+  }
+
+  @Override
+  public <T> T getConfig(PolarisConfiguration<T> config) {
+    return configurationStore.getConfiguration(realmContext, config);
+  }
+
+  @Override
+  public <T> T getConfig(PolarisConfiguration<T> config, CatalogEntity catalogEntity) {
+    return configurationStore.getConfiguration(realmContext, catalogEntity, config);
+  }
+}

polaris-core/src/main/java/org/apache/polaris/core/context/CallContext.java

@@ -19,6 +19,7 @@
 package org.apache.polaris.core.context;
 
 import org.apache.polaris.core.PolarisCallContext;
+import org.apache.polaris.core.config.RealmConfig;
 
 /**
  * Stores elements associated with an individual REST request such as RealmContext, caller
@@ -53,4 +54,6 @@ static void unsetCurrentContext() {
    * @return the inner context used for delegating services
    */
   PolarisCallContext getPolarisCallContext();
+
+  RealmConfig getRealmConfig();
 }

polaris-core/src/main/java/org/apache/polaris/core/entity/CatalogEntity.java

@@ -313,11 +313,8 @@ private void validateMaxAllowedLocations(
         CallContext callContext, Collection<String> allowedLocations) {
       int maxAllowedLocations =
           callContext
-              .getPolarisCallContext()
-              .getConfigurationStore()
-              .getConfiguration(
-                  callContext.getRealmContext(),
-                  BehaviorChangeConfiguration.STORAGE_CONFIGURATION_MAX_LOCATIONS);
+              .getRealmConfig()
+              .getConfig(BehaviorChangeConfiguration.STORAGE_CONFIGURATION_MAX_LOCATIONS);
       if (maxAllowedLocations != -1 && allowedLocations.size() > maxAllowedLocations) {
         throw new IllegalArgumentException(
             String.format(

polaris-core/src/main/java/org/apache/polaris/core/persistence/AtomicOperationMetaStoreManager.java

@@ -1508,9 +1508,8 @@ private void revokeGrantRecord(
                   PolarisObjectMapperUtil.parseTaskState(entity);
               long taskAgeTimeout =
                   callCtx
-                      .getConfigurationStore()
-                      .getConfiguration(
-                          callCtx.getRealmContext(),
+                      .getRealmConfig()
+                      .getConfig(
                           PolarisTaskConstants.TASK_TIMEOUT_MILLIS_CONFIG,
                           PolarisTaskConstants.TASK_TIMEOUT_MILLIS);
               return taskState == null