rewrite

Author: eric-maynard

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

@@ -20,20 +20,17 @@
 
 import java.util.ArrayList;
 import java.util.List;
-import java.util.Map;
-import java.util.Objects;
 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
- * TODO: deprecate unsafe catalog configs and remove related code
+ * 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
  */
@@ -63,21 +60,20 @@ private static void registerConfiguration(PolarisConfiguration<?> configuration)
         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()) {
+        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);
@@ -168,6 +164,10 @@ public Builder<T> catalogConfig(String catalogConfig) {
     @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;
     }
@@ -177,7 +177,8 @@ public FeatureConfiguration<T> buildFeatureConfiguration() {
         throw new IllegalArgumentException("key, description, and defaultValue are required");
       }
       FeatureConfiguration<T> config =
-          new FeatureConfiguration<>(key, description, defaultValue, catalogConfig, catalogConfigUnsafe);
+          new FeatureConfiguration<>(
+              key, description, defaultValue, catalogConfig, catalogConfigUnsafe);
       PolarisConfiguration.registerConfiguration(config);
       return config;
     }
@@ -191,7 +192,8 @@ public BehaviorChangeConfiguration<T> buildBehaviorChangeConfiguration() {
             "catalog configs are not valid for behavior change configs");
       }
       BehaviorChangeConfiguration<T> config =
-          new BehaviorChangeConfiguration<>(key, description, defaultValue, catalogConfig, catalogConfigUnsafe);
+          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,8 +23,6 @@
 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;
@@ -129,5 +127,4 @@ public interface PolarisConfigurationStore {
       return getConfiguration(ctx, config);
     }
   }
-
 }

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,27 @@ 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.allConfigurations.forEach(
+          c -> {
+            if (c.hasCatalogConfig()) {
+              allowlist.add(c.catalogConfig());
+            }
+            if (c.hasCatalogConfigUnsafe()) {
+              allowlist.add(c.catalogConfigUnsafe());
+            }
+          });
+      return allowlist;
+    }
+  }
 }

service/common/src/main/java/org/apache/polaris/service/config/ReservedProperties.java

@@ -22,6 +22,7 @@
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.stream.Collectors;
 import org.apache.iceberg.MetadataUpdate;
 import org.slf4j.Logger;
@@ -43,6 +44,11 @@ public interface ReservedProperties {
         public List<String> prefixes() {
           return List.of();
         }
+
+        @Override
+        public Set<String> allowlist() {
+          return Set.of();
+        }
       };
 
   /**
@@ -51,6 +57,12 @@ public List<String> prefixes() {
    */
   List<String> prefixes();
 
+  /**
+   * A list of properties that are *not* considered reserved, even if they start with a reserved
+   * prefix
+   */
+  Set<String> allowlist();
+
   /** If true, attempts to modify a reserved property should throw an exception. */
   default boolean shouldThrow() {
     return true;
@@ -94,15 +106,17 @@ default Map<String, String> removeReservedProperties(Map<String, String> propert
     List<String> prefixes = prefixes();
     for (var entry : properties.entrySet()) {
       boolean isReserved = false;
-      for (String prefix : prefixes) {
-        if (entry.getKey().startsWith(prefix)) {
-          isReserved = true;
-          String message =
-              String.format("Property '%s' matches reserved prefix '%s'", entry.getKey(), prefix);
-          if (shouldThrow()) {
-            throw new IllegalArgumentException(message);
-          } else {
-            LOGGER.debug(message);
+      if (!allowlist().contains(entry.getKey())) {
+        for (String prefix : prefixes) {
+          if (entry.getKey().startsWith(prefix)) {
+            isReserved = true;
+            String message =
+                String.format("Property '%s' matches reserved prefix '%s'", entry.getKey(), prefix);
+            if (shouldThrow()) {
+              throw new IllegalArgumentException(message);
+            } else {
+              LOGGER.debug(message);
+            }
           }
         }
       }