Improvements to PolarisConfiguration

polaris-core/src/main/java/io/polaris/core/PolarisConfiguration.java

@@ -15,30 +15,136 @@
  */
 package io.polaris.core;
 
-public class PolarisConfiguration {
+import java.util.Optional;
 
-  public static final String ENFORCE_PRINCIPAL_CREDENTIAL_ROTATION_REQUIRED_CHECKING =
-      "ENFORCE_PRINCIPAL_CREDENTIAL_ROTATION_REQUIRED_CHECKING";
-  public static final String ALLOW_TABLE_LOCATION_OVERLAP = "ALLOW_TABLE_LOCATION_OVERLAP";
-  public static final String ALLOW_NAMESPACE_LOCATION_OVERLAP = "ALLOW_NAMESPACE_LOCATION_OVERLAP";
-  public static final String ALLOW_EXTERNAL_METADATA_FILE_LOCATION =
-      "ALLOW_EXTERNAL_METADATA_FILE_LOCATION";
+public class PolarisConfiguration<T> {
 
-  public static final String ALLOW_OVERLAPPING_CATALOG_URLS = "ALLOW_OVERLAPPING_CATALOG_URLS";
+  public final String key;
+  public final String description;
+  public final T defaultValue;
+  private final Optional<String> catalogConfigImpl;
+  private final Class<T> typ;
 
-  public static final String CATALOG_ALLOW_UNSTRUCTURED_TABLE_LOCATION =
-      "allow.unstructured.table.location";
-  public static final String CATALOG_ALLOW_EXTERNAL_TABLE_LOCATION =
-      "allow.external.table.location";
+  @SuppressWarnings("unchecked")
+  public PolarisConfiguration(
+      String key, String description, T defaultValue, Optional<String> catalogConfig) {
+    this.key = key;
+    this.description = description;
+    this.defaultValue = defaultValue;
+    this.catalogConfigImpl = catalogConfig;
+    this.typ = (Class<T>) defaultValue.getClass();
+  }
 
-  /*
-   * Default values for the configuration properties
-   */
+  public boolean hasCatalogConfig() {
+    return catalogConfigImpl.isPresent();
+  }
 
-  public static final boolean DEFAULT_ALLOW_OVERLAPPING_CATALOG_URLS = false;
-  public static final boolean DEFAULT_ALLOW_TABLE_LOCATION_OVERLAP = false;
-  public static final boolean DEFAULT_ALLOW_EXTERNAL_METADATA_FILE_LOCATION = false;
-  public static final boolean DEFAULT_ALLOW_NAMESPACE_LOCATION_OVERLAP = false;
+  public String catalogConfig() {
+    return catalogConfigImpl.orElseThrow(
+        () ->
+            new IllegalStateException(
+                "Attempted to read a catalog config key from a configuration that doesn't have one."));
+  }
 
-  private PolarisConfiguration() {}
+  T cast(Object value) {
+    return this.typ.cast(value);
+  }
+
+  public static class Builder<T> {
+    private String key;
+    private String description;
+    private T defaultValue;
+    private Optional<String> catalogConfig = Optional.empty();
+
+    public Builder<T> key(String key) {
+      this.key = key;
+      return this;
+    }
+
+    public Builder<T> description(String description) {
+      this.description = description;
+      return this;
+    }
+
+    public Builder<T> defaultValue(T defaultValue) {
+      this.defaultValue = defaultValue;
+      return this;
+    }
+
+    public Builder<T> catalogConfig(String catalogConfig) {
+      this.catalogConfig = Optional.of(catalogConfig);
+      return this;
+    }
+
+    public PolarisConfiguration<T> build() {
+      if (key == null || description == null || defaultValue == null) {
+        throw new IllegalArgumentException("key, description, and defaultValue are required");
+      }
+      return new PolarisConfiguration<>(key, description, defaultValue, catalogConfig);
+    }
+  }
+
+  public static <T> Builder<T> builder() {
+    return new Builder<>();
+  }
+
+  public static final PolarisConfiguration<Boolean>
+      ENFORCE_PRINCIPAL_CREDENTIAL_ROTATION_REQUIRED_CHECKING =
+          PolarisConfiguration.<Boolean>builder()
+              .key("ENFORCE_PRINCIPAL_CREDENTIAL_ROTATION_REQUIRED_CHECKING")
+              .description(
+                  "If set to true, require that principals must rotate their credentials before being used "
+                      + "for anything else.")
+              .defaultValue(false)
+              .build();
+
+  public static final PolarisConfiguration<Boolean> ALLOW_TABLE_LOCATION_OVERLAP =
+      PolarisConfiguration.<Boolean>builder()
+          .key("ALLOW_TABLE_LOCATION_OVERLAP")
+          .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.")
+          .defaultValue(false)
+          .build();
+
+  public static final PolarisConfiguration<Boolean> ALLOW_NAMESPACE_LOCATION_OVERLAP =
+      PolarisConfiguration.<Boolean>builder()
+          .key("ALLOW_NAMESPACE_LOCATION_OVERLAP")
+          .description(
+              "If set to true, allow one table's location to reside within another table's location. "
+                  + "This is only enforced within a parent catalog or namespace.")
+          .defaultValue(false)
+          .build();
+
+  public static final PolarisConfiguration<Boolean> ALLOW_EXTERNAL_METADATA_FILE_LOCATION =
+      PolarisConfiguration.<Boolean>builder()
+          .key("ALLOW_EXTERNAL_METADATA_FILE_LOCATION")
+          .description(
+              "If set to true, allows metadata files to be located outside the default metadata directory.")
+          .defaultValue(false)
+          .build();
+
+  public static final PolarisConfiguration<Boolean> ALLOW_OVERLAPPING_CATALOG_URLS =
+      PolarisConfiguration.<Boolean>builder()
+          .key("ALLOW_OVERLAPPING_CATALOG_URLS")
+          .description("If set to true, allows catalog URLs to overlap.")
+          .defaultValue(false)
+          .build();
+
+  public static final PolarisConfiguration<Boolean> ALLOW_UNSTRUCTURED_TABLE_LOCATION =
+      PolarisConfiguration.<Boolean>builder()
+          .key("ALLOW_UNSTRUCTURED_TABLE_LOCATION")
+          .catalogConfig("allow.unstructured.table.location")
+          .description("If set to true, allows unstructured table locations.")
+          .defaultValue(false)
+          .build();
+
+  public static final PolarisConfiguration<Boolean> ALLOW_EXTERNAL_TABLE_LOCATION =
+      PolarisConfiguration.<Boolean>builder()
+          .key("ALLOW_EXTERNAL_TABLE_LOCATION")
+          .catalogConfig("allow.external.table.location")
+          .description(
+              "If set to true, allows tables to have external locations outside the default structure.")
+          .defaultValue(false)
+          .build();
 }

polaris-core/src/main/java/io/polaris/core/PolarisConfigurationStore.java

@@ -16,14 +16,18 @@
 package io.polaris.core;
 
 import com.google.common.base.Preconditions;
+import io.polaris.core.entity.CatalogEntity;
 import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;
+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.
@@ -53,4 +57,56 @@ public interface PolarisConfigurationStore {
     T configValue = getConfiguration(ctx, configName);
     return configValue != null ? configValue : defaultValue;
   }
+
+  /**
+   * In some cases, we may extract a value that doesn't match the expected type for a config. This
+   * method can be used to attempt to force-cast it using `String.valueOf`
+   */
+  private <T> @NotNull T tryCast(PolarisConfiguration<T> config, Object value) {
+    if (value == null) {
+      return config.defaultValue;
+    }
+
+    if (config.defaultValue instanceof Boolean) {
+      return config.cast(Boolean.valueOf(String.valueOf(value)));
+    } else {
+      return config.cast(value);
+    }
+  }
+
+  /**
+   * Retrieve the current value for a configuration.
+   *
+   * @param ctx the current call context
+   * @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> @NotNull T getConfiguration(PolarisCallContext ctx, PolarisConfiguration<T> config) {
+    T result = getConfiguration(ctx, config.key, config.defaultValue);
+    return tryCast(config, result);
+  }
+
+  /**
+   * Retrieve the current value for a configuration, overriding with a catalog config if it is
+   * present.
+   *
+   * @param ctx the current call context
+   * @param catalogEntity the catalog 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> @NotNull T getConfiguration(
+      PolarisCallContext ctx,
+      @NotNull CatalogEntity catalogEntity,
+      PolarisConfiguration<T> config) {
+    if (config.hasCatalogConfig()
+        && catalogEntity.getPropertiesAsMap().containsKey(config.catalogConfig())) {
+      LOGGER.debug("Loaded config from catalog: {}", config.catalogConfig());
+      return tryCast(config, catalogEntity.getPropertiesAsMap().get(config.catalogConfig()));
+    } else {
+      return getConfiguration(ctx, config);
+    }
+  }
 }

polaris-core/src/main/java/io/polaris/core/auth/PolarisAuthorizer.java

@@ -500,8 +500,7 @@ public void authorizeOrThrow(
     boolean enforceCredentialRotationRequiredState =
         featureConfig.getConfiguration(
             CallContext.getCurrentContext().getPolarisCallContext(),
-            PolarisConfiguration.ENFORCE_PRINCIPAL_CREDENTIAL_ROTATION_REQUIRED_CHECKING,
-            false);
+            PolarisConfiguration.ENFORCE_PRINCIPAL_CREDENTIAL_ROTATION_REQUIRED_CHECKING);
     if (enforceCredentialRotationRequiredState
         && authenticatedPrincipal
             .getPrincipalEntity()

polaris-core/src/main/java/io/polaris/core/storage/PolarisStorageConfigurationInfo.java

@@ -25,6 +25,7 @@
 import io.polaris.core.PolarisConfiguration;
 import io.polaris.core.PolarisDiagnostics;
 import io.polaris.core.admin.model.Catalog;
+import io.polaris.core.context.CallContext;
 import io.polaris.core.entity.CatalogEntity;
 import io.polaris.core.entity.PolarisEntity;
 import io.polaris.core.entity.PolarisEntityConstants;
@@ -147,19 +148,16 @@ public static Optional<PolarisStorageConfigurationInfo> forEntityPath(
                       .orElse(null);
               CatalogEntity catalog = CatalogEntity.of(entityPath.get(0));
               boolean allowEscape =
-                  Optional.ofNullable(
-                          catalog
-                              .getPropertiesAsMap()
-                              .get(PolarisConfiguration.CATALOG_ALLOW_UNSTRUCTURED_TABLE_LOCATION))
-                      .map(
-                          val -> {
-                            LOGGER.debug(
-                                "Found catalog level property to allow unstructured table location: {}",
-                                val);
-                            return Boolean.parseBoolean(val);
-                          })
-                      .orElseGet(() -> Catalog.TypeEnum.EXTERNAL.equals(catalog.getCatalogType()));
-              if (!allowEscape && baseLocation != null) {
+                  CallContext.getCurrentContext()
+                      .getPolarisCallContext()
+                      .getConfigurationStore()
+                      .getConfiguration(
+                          CallContext.getCurrentContext().getPolarisCallContext(),
+                          catalog,
+                          PolarisConfiguration.ALLOW_UNSTRUCTURED_TABLE_LOCATION);
+              if (!allowEscape
+                  && catalog.getCatalogType() != Catalog.TypeEnum.EXTERNAL
+                  && baseLocation != null) {
                 LOGGER.debug(
                     "Not allowing unstructured table location for entity: {}",
                     entityPath.getLast().getName());

polaris-service/src/main/java/io/polaris/service/admin/PolarisAdminService.java

@@ -496,14 +496,10 @@ private String terminateWithSlash(String path) {
    */
   private boolean catalogOverlapsWithExistingCatalog(CatalogEntity catalogEntity) {
     boolean allowOverlappingCatalogUrls =
-        Boolean.parseBoolean(
-            String.valueOf(
-                getCurrentPolarisContext()
-                    .getConfigurationStore()
-                    .getConfiguration(
-                        getCurrentPolarisContext(),
-                        PolarisConfiguration.ALLOW_OVERLAPPING_CATALOG_URLS,
-                        PolarisConfiguration.DEFAULT_ALLOW_OVERLAPPING_CATALOG_URLS)));
+        getCurrentPolarisContext()
+            .getConfigurationStore()
+            .getConfiguration(
+                getCurrentPolarisContext(), PolarisConfiguration.ALLOW_OVERLAPPING_CATALOG_URLS);
 
     if (allowOverlappingCatalogUrls) {
       return false;

polaris-service/src/main/java/io/polaris/service/catalog/BasePolarisCatalog.java

@@ -431,8 +431,7 @@ private void createNamespaceInternal(
         .getConfigurationStore()
         .getConfiguration(
             callContext.getPolarisCallContext(),
-            PolarisConfiguration.ALLOW_NAMESPACE_LOCATION_OVERLAP,
-            PolarisConfiguration.DEFAULT_ALLOW_NAMESPACE_LOCATION_OVERLAP)) {
+            PolarisConfiguration.ALLOW_NAMESPACE_LOCATION_OVERLAP)) {
       LOG.debug("Validating no overlap for {} with sibling tables or namespaces", namespace);
       validateNoLocationOverlap(
           entity.getBaseLocation(), resolvedParent.getRawFullPath(), entity.getName());
@@ -577,8 +576,7 @@ public boolean setProperties(Namespace namespace, Map<String, String> properties
         .getConfigurationStore()
         .getConfiguration(
             callContext.getPolarisCallContext(),
-            PolarisConfiguration.ALLOW_NAMESPACE_LOCATION_OVERLAP,
-            PolarisConfiguration.DEFAULT_ALLOW_NAMESPACE_LOCATION_OVERLAP)) {
+            PolarisConfiguration.ALLOW_NAMESPACE_LOCATION_OVERLAP)) {
       LOG.debug("Validating no overlap with sibling tables or namespaces");
       validateNoLocationOverlap(
           NamespaceEntity.of(updatedEntity).getBaseLocation(),
@@ -915,8 +913,7 @@ private void validateNoLocationOverlap(
         .getConfigurationStore()
         .getConfiguration(
             callContext.getPolarisCallContext(),
-            PolarisConfiguration.ALLOW_TABLE_LOCATION_OVERLAP,
-            PolarisConfiguration.DEFAULT_ALLOW_TABLE_LOCATION_OVERLAP)) {
+            PolarisConfiguration.ALLOW_TABLE_LOCATION_OVERLAP)) {
       LOG.debug("Skipping location overlap validation for identifier '{}'", identifier);
     } else { // if (entity.getSubType().equals(PolarisEntitySubType.TABLE)) {
       // TODO - is this necessary for views? overlapping views do not expose subdirectories via the
@@ -1256,17 +1253,16 @@ protected String tableName() {
   private void validateMetadataFileInTableDir(
       TableIdentifier identifier, TableMetadata metadata, CatalogEntity catalog) {
     PolarisCallContext polarisCallContext = callContext.getPolarisCallContext();
-    String allowEscape =
-        catalog
-            .getPropertiesAsMap()
-            .get(PolarisConfiguration.CATALOG_ALLOW_EXTERNAL_TABLE_LOCATION);
-    if (!Boolean.parseBoolean(allowEscape)
+    boolean allowEscape =
+        polarisCallContext
+            .getConfigurationStore()
+            .getConfiguration(
+                polarisCallContext, PolarisConfiguration.ALLOW_EXTERNAL_TABLE_LOCATION);
+    if (!allowEscape
         && !polarisCallContext
             .getConfigurationStore()
             .getConfiguration(
-                polarisCallContext,
-                PolarisConfiguration.ALLOW_EXTERNAL_METADATA_FILE_LOCATION,
-                PolarisConfiguration.DEFAULT_ALLOW_EXTERNAL_METADATA_FILE_LOCATION)) {
+                polarisCallContext, PolarisConfiguration.ALLOW_EXTERNAL_METADATA_FILE_LOCATION)) {
       LOG.debug(
           "Validating base location {} for table {} in metadata file {}",
           metadata.location(),

polaris-service/src/test/java/io/polaris/service/admin/PolarisAuthzTestBase.java

@@ -127,7 +127,7 @@ public abstract class PolarisAuthzTestBase {
       new PolarisAuthorizer(
           new DefaultConfigurationStore(
               Map.of(
-                  PolarisConfiguration.ENFORCE_PRINCIPAL_CREDENTIAL_ROTATION_REQUIRED_CHECKING,
+                  PolarisConfiguration.ENFORCE_PRINCIPAL_CREDENTIAL_ROTATION_REQUIRED_CHECKING.key,
                   true)));
 
   protected BasePolarisCatalog baseCatalog;