Migrate catalog configs to the new reserved prefix

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

@@ -33,8 +33,12 @@
 public class BehaviorChangeConfiguration<T> extends PolarisConfiguration<T> {
 
   protected BehaviorChangeConfiguration(
-      String key, String description, T defaultValue, Optional<String> catalogConfig) {
-    super(key, description, defaultValue, catalogConfig);
+      String key,
+      String description,
+      T defaultValue,
+      Optional<String> catalogConfig,
+      Optional<String> catalogConfigUnsafe) {
+    super(key, description, defaultValue, catalogConfig, catalogConfigUnsafe);
   }
 
   public static final BehaviorChangeConfiguration<Boolean> VALIDATE_VIEW_LOCATION_OVERLAP =

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

@@ -33,8 +33,12 @@
  */
 public class FeatureConfiguration<T> extends PolarisConfiguration<T> {
   protected FeatureConfiguration(
-      String key, String description, T defaultValue, Optional<String> catalogConfig) {
-    super(key, description, defaultValue, catalogConfig);
+      String key,
+      String description,
+      T defaultValue,
+      Optional<String> catalogConfig,
+      Optional<String> catalogConfigUnsafe) {
+    super(key, description, defaultValue, catalogConfig, catalogConfigUnsafe);
   }
 
   /**
@@ -81,7 +85,8 @@ public static void enforceFeatureEnabledOrThrow(
   public static final FeatureConfiguration<Boolean> ALLOW_TABLE_LOCATION_OVERLAP =
       PolarisConfiguration.<Boolean>builder()
           .key("ALLOW_TABLE_LOCATION_OVERLAP")
-          .catalogConfig("allow.overlapping.table.location")
+          .catalogConfig("polaris.config.allow.overlapping.table.location")
+          .catalogConfigUnsafe("allow.overlapping.table.location")
           .description(
               "If set to true, allow one table's location to reside within another table's location. "
                   + "This is only enforced within a given namespace.")
@@ -115,15 +120,17 @@ public static void enforceFeatureEnabledOrThrow(
   public static final FeatureConfiguration<Boolean> ALLOW_UNSTRUCTURED_TABLE_LOCATION =
       PolarisConfiguration.<Boolean>builder()
           .key("ALLOW_UNSTRUCTURED_TABLE_LOCATION")
-          .catalogConfig("allow.unstructured.table.location")
+          .catalogConfig("polaris.config.allow.unstructured.table.location")
+          .catalogConfigUnsafe("allow.unstructured.table.location")
           .description("If set to true, allows unstructured table locations.")
           .defaultValue(false)
           .buildFeatureConfiguration();
 
   public static final FeatureConfiguration<Boolean> ALLOW_EXTERNAL_TABLE_LOCATION =
       PolarisConfiguration.<Boolean>builder()
           .key("ALLOW_EXTERNAL_TABLE_LOCATION")
-          .catalogConfig("allow.external.table.location")
+          .catalogConfig("polaris.config.allow.external.table.location")
+          .catalogConfigUnsafe("allow.external.table.location")
           .description(
               "If set to true, allows tables to have external locations outside the default structure.")
           .defaultValue(false)
@@ -132,15 +139,17 @@ public static void enforceFeatureEnabledOrThrow(
   public static final FeatureConfiguration<Boolean> ALLOW_EXTERNAL_CATALOG_CREDENTIAL_VENDING =
       PolarisConfiguration.<Boolean>builder()
           .key("ALLOW_EXTERNAL_CATALOG_CREDENTIAL_VENDING")
-          .catalogConfig("enable.credential.vending")
+          .catalogConfig("polaris.config.enable.credential.vending")
+          .catalogConfigUnsafe("enable.credential.vending")
           .description("If set to true, allow credential vending for external catalogs.")
           .defaultValue(true)
           .buildFeatureConfiguration();
 
   public static final FeatureConfiguration<List<String>> SUPPORTED_CATALOG_STORAGE_TYPES =
       PolarisConfiguration.<List<String>>builder()
           .key("SUPPORTED_CATALOG_STORAGE_TYPES")
-          .catalogConfig("supported.storage.types")
+          .catalogConfig("polaris.config.supported.storage.types")
+          .catalogConfigUnsafe("supported.storage.types")
           .description("The list of supported storage types for a catalog")
           .defaultValue(
               List.of(
@@ -153,23 +162,26 @@ public static void enforceFeatureEnabledOrThrow(
   public static final FeatureConfiguration<Boolean> CLEANUP_ON_NAMESPACE_DROP =
       PolarisConfiguration.<Boolean>builder()
           .key("CLEANUP_ON_NAMESPACE_DROP")
-          .catalogConfig("cleanup.on.namespace.drop")
+          .catalogConfig("polaris.config.cleanup.on.namespace.drop")
+          .catalogConfigUnsafe("cleanup.on.namespace.drop")
           .description("If set to true, clean up data when a namespace is dropped")
           .defaultValue(false)
           .buildFeatureConfiguration();
 
   public static final FeatureConfiguration<Boolean> CLEANUP_ON_CATALOG_DROP =
       PolarisConfiguration.<Boolean>builder()
           .key("CLEANUP_ON_CATALOG_DROP")
-          .catalogConfig("cleanup.on.catalog.drop")
+          .catalogConfig("polaris.config.cleanup.on.catalog.drop")
+          .catalogConfigUnsafe("cleanup.on.catalog.drop")
           .description("If set to true, clean up data when a catalog is dropped")
           .defaultValue(false)
           .buildFeatureConfiguration();
 
   public static final FeatureConfiguration<Boolean> DROP_WITH_PURGE_ENABLED =
       PolarisConfiguration.<Boolean>builder()
           .key("DROP_WITH_PURGE_ENABLED")
-          .catalogConfig("drop-with-purge.enabled")
+          .catalogConfig("polaris.config.drop-with-purge.enabled")
+          .catalogConfigUnsafe("drop-with-purge.enabled")
           .description(
               "If set to true, allows tables to be dropped with the purge parameter set to true.")
           .defaultValue(true)

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

@@ -21,32 +21,81 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Optional;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
 import org.apache.polaris.core.context.CallContext;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 /**
- * An ABC for Polaris configurations that alter the service's behavior
+ * An ABC for Polaris configurations that alter the service's behavior TODO: deprecate unsafe
+ * catalog configs and remove related code
  *
  * @param <T> The type of the configuration
  */
 public abstract class PolarisConfiguration<T> {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(PolarisConfiguration.class);
 
+  private static final List<PolarisConfiguration<?>> allConfigurations = new ArrayList<>();
+
   public final String key;
   public final String description;
   public final T defaultValue;
   private final Optional<String> catalogConfigImpl;
+  private final Optional<String> catalogConfigUnsafeImpl;
   private final Class<T> typ;
 
+  /** catalog configs are expected to start with this prefix */
+  private static final String SAFE_CATALOG_CONFIG_PREFIX = "polaris.config.";
+
+  /**
+   * Helper method for building `allConfigurations` and checking for duplicate use of keys across
+   * configs.
+   */
+  private static void registerConfiguration(PolarisConfiguration<?> configuration) {
+    for (PolarisConfiguration<?> existingConfiguration : allConfigurations) {
+      if (existingConfiguration.key.equals(configuration.key)) {
+        throw new IllegalArgumentException(
+            String.format("Config '%s' is already in use", configuration.key));
+      } else {
+        var configs =
+            Stream.of(
+                    configuration.catalogConfigImpl,
+                    configuration.catalogConfigUnsafeImpl,
+                    existingConfiguration.catalogConfigImpl,
+                    existingConfiguration.catalogConfigUnsafeImpl)
+                .flatMap(Optional::stream)
+                .collect(Collectors.groupingBy(Function.identity(), Collectors.counting()));
+        for (var entry : configs.entrySet()) {
+          if (entry.getValue() > 1) {
+            throw new IllegalArgumentException(
+                String.format("Catalog config %s is already in use", entry.getKey()));
+          }
+        }
+      }
+    }
+    allConfigurations.add(configuration);
+  }
+
+  /** Returns a list of all PolarisConfigurations that have been registered */
+  public static List<PolarisConfiguration<?>> getAllConfigurations() {
+    return List.copyOf(allConfigurations);
+  }
+
   @SuppressWarnings("unchecked")
   protected PolarisConfiguration(
-      String key, String description, T defaultValue, Optional<String> catalogConfig) {
+      String key,
+      String description,
+      T defaultValue,
+      Optional<String> catalogConfig,
+      Optional<String> catalogConfigUnsafe) {
     this.key = key;
     this.description = description;
     this.defaultValue = defaultValue;
     this.catalogConfigImpl = catalogConfig;
+    this.catalogConfigUnsafeImpl = catalogConfigUnsafe;
     this.typ = (Class<T>) defaultValue.getClass();
   }
 
@@ -61,6 +110,17 @@ public String catalogConfig() {
                 "Attempted to read a catalog config key from a configuration that doesn't have one."));
   }
 
+  public boolean hasCatalogConfigUnsafe() {
+    return catalogConfigUnsafeImpl.isPresent();
+  }
+
+  public String catalogConfigUnsafe() {
+    return catalogConfigUnsafeImpl.orElseThrow(
+        () ->
+            new IllegalStateException(
+                "Attempted to read an unsafe catalog config key from a configuration that doesn't have one."));
+  }
+
   T cast(Object value) {
     return this.typ.cast(value);
   }
@@ -70,6 +130,7 @@ public static class Builder<T> {
     private String description;
     private T defaultValue;
     private Optional<String> catalogConfig = Optional.empty();
+    private Optional<String> catalogConfigUnsafe = Optional.empty();
 
     public Builder<T> key(String key) {
       this.key = key;
@@ -93,26 +154,53 @@ public Builder<T> defaultValue(T defaultValue) {
     }
 
     public Builder<T> catalogConfig(String catalogConfig) {
+      if (!catalogConfig.startsWith(SAFE_CATALOG_CONFIG_PREFIX)) {
+        throw new IllegalArgumentException(
+            "Catalog configs are expected to start with " + SAFE_CATALOG_CONFIG_PREFIX);
+      }
       this.catalogConfig = Optional.of(catalogConfig);
       return this;
     }
 
+    /**
+     * Used to support backwards compatability before there were reserved properties. Usage of this
+     * method should be removed over time.
+     */
+    @Deprecated
+    public Builder<T> catalogConfigUnsafe(String catalogConfig) {
+      LOGGER.info("catalogConfigUnsafe is deprecated! Use catalogConfig() instead.");
+      if (catalogConfig.startsWith(SAFE_CATALOG_CONFIG_PREFIX)) {
+        throw new IllegalArgumentException(
+            "Unsafe catalog configs are not expected to start with " + SAFE_CATALOG_CONFIG_PREFIX);
+      }
+      this.catalogConfigUnsafe = Optional.of(catalogConfig);
+      return this;
+    }
+
     public FeatureConfiguration<T> buildFeatureConfiguration() {
       if (key == null || description == null || defaultValue == null) {
         throw new IllegalArgumentException("key, description, and defaultValue are required");
       }
-      return new FeatureConfiguration<>(key, description, defaultValue, catalogConfig);
+      FeatureConfiguration<T> config =
+          new FeatureConfiguration<>(
+              key, description, defaultValue, catalogConfig, catalogConfigUnsafe);
+      PolarisConfiguration.registerConfiguration(config);
+      return config;
     }
 
     public BehaviorChangeConfiguration<T> buildBehaviorChangeConfiguration() {
       if (key == null || description == null || defaultValue == null) {
         throw new IllegalArgumentException("key, description, and defaultValue are required");
       }
-      if (catalogConfig.isPresent()) {
+      if (catalogConfig.isPresent() || catalogConfigUnsafe.isPresent()) {
         throw new IllegalArgumentException(
-            "catalogConfig is not valid for behavior change configs");
+            "catalog configs are not valid for behavior change configs");
       }
-      return new BehaviorChangeConfiguration<>(key, description, defaultValue, catalogConfig);
+      BehaviorChangeConfiguration<T> config =
+          new BehaviorChangeConfiguration<>(
+              key, description, defaultValue, catalogConfig, catalogConfigUnsafe);
+      PolarisConfiguration.registerConfiguration(config);
+      return config;
     }
   }
 

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

@@ -23,14 +23,19 @@
 import jakarta.annotation.Nullable;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Map;
 import org.apache.polaris.core.PolarisCallContext;
 import org.apache.polaris.core.entity.CatalogEntity;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * Dynamic configuration store used to retrieve runtime parameters, which may vary by realm or by
  * request.
  */
 public interface PolarisConfigurationStore {
+  Logger LOGGER = LoggerFactory.getLogger(PolarisConfigurationStore.class);
+
   /**
    * Retrieve the current value for a configuration key. May be null if not set.
    *
@@ -111,11 +116,22 @@ public interface PolarisConfigurationStore {
       PolarisCallContext ctx,
       @Nonnull CatalogEntity catalogEntity,
       PolarisConfiguration<T> config) {
-    if (config.hasCatalogConfig()
-        && catalogEntity.getPropertiesAsMap().containsKey(config.catalogConfig())) {
-      return tryCast(config, catalogEntity.getPropertiesAsMap().get(config.catalogConfig()));
-    } else {
-      return getConfiguration(ctx, config);
+    if (config.hasCatalogConfig() || config.hasCatalogConfigUnsafe()) {
+      Map<String, String> propertiesMap = catalogEntity.getPropertiesAsMap();
+      String propertyValue = propertiesMap.get(config.catalogConfig());
+      if (propertyValue == null) {
+        propertyValue = propertiesMap.get(config.catalogConfigUnsafe());
+        if (propertyValue != null) {
+          LOGGER.warn(
+              String.format(
+                  "Deprecated config %s is in use and will be removed in a future version",
+                  config.catalogConfigUnsafe()));
+        }
+      }
+      if (propertyValue != null) {
+        return tryCast(config, propertyValue);
+      }
     }
+    return getConfiguration(ctx, config);
   }
 }

polaris-core/src/test/java/org/apache/polaris/service/storage/PolarisConfigurationStoreTest.java

@@ -75,7 +75,7 @@ public void testConfigsCanBeCastedFromString() {
   public void testInvalidCastThrowsException() {
     // Bool not included because Boolean.valueOf turns non-boolean strings to false
     List<PolarisConfiguration<?>> configs =
-        List.of(buildConfig("int", 12), buildConfig("long", 34L), buildConfig("double", 5.6D));
+        List.of(buildConfig("int2", 12), buildConfig("long2", 34L), buildConfig("double2", 5.6D));
 
     PolarisConfigurationStore store =
         new PolarisConfigurationStore() {
@@ -135,7 +135,7 @@ public void testBehaviorAndFeatureConfigs() {
 
     BehaviorChangeConfiguration<Boolean> behaviorChangeConfig =
         PolarisConfiguration.<Boolean>builder()
-            .key("example")
+            .key("example2")
             .description("example")
             .defaultValue(true)
             .buildBehaviorChangeConfiguration();

quarkus/service/src/main/java/org/apache/polaris/service/quarkus/config/QuarkusReservedProperties.java

@@ -19,7 +19,10 @@
 package org.apache.polaris.service.quarkus.config;
 
 import io.smallrye.config.ConfigMapping;
+import java.util.HashSet;
 import java.util.List;
+import java.util.Set;
+import org.apache.polaris.core.config.PolarisConfiguration;
 import org.apache.polaris.service.config.ReservedProperties;
 
 @ConfigMapping(prefix = "polaris.reserved-properties")
@@ -28,4 +31,28 @@ public interface QuarkusReservedProperties extends ReservedProperties {
   default List<String> prefixes() {
     return List.of("polaris.");
   }
+
+  @Override
+  default Set<String> allowlist() {
+    return AllowlistHolder.INSTANCE;
+  }
+
+  class AllowlistHolder {
+    static final Set<String> INSTANCE = computeAllowlist();
+
+    private static Set<String> computeAllowlist() {
+      Set<String> allowlist = new HashSet<>();
+      PolarisConfiguration.getAllConfigurations()
+          .forEach(
+              c -> {
+                if (c.hasCatalogConfig()) {
+                  allowlist.add(c.catalogConfig());
+                }
+                if (c.hasCatalogConfigUnsafe()) {
+                  allowlist.add(c.catalogConfigUnsafe());
+                }
+              });
+      return allowlist;
+    }
+  }
 }