Add SUPPORTED_CATALOG_STORAGE_TYPES to PolarisConfiguration (#165)

apache · Sep 24, 2024 · 413eca7 · 413eca7
1 parent 3396e22
commit 413eca7
Show file tree

Hide file tree

Showing 5 changed files with 65 additions and 25 deletions.
diff --git a/polaris-core/src/main/java/org/apache/polaris/core/PolarisConfiguration.java b/polaris-core/src/main/java/org/apache/polaris/core/PolarisConfiguration.java
@@ -18,7 +18,9 @@
  */
 package org.apache.polaris.core;
 
+import java.util.List;
 import java.util.Optional;
+import org.apache.polaris.core.admin.model.StorageConfigInfo;
 
 public class PolarisConfiguration<T> {
 
@@ -115,7 +117,7 @@ public static <T> Builder<T> builder() {
       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. "
+              "If set to true, allow one namespace's location to reside within another namespace's location. "
                   + "This is only enforced within a parent catalog or namespace.")
           .defaultValue(false)
           .build();
@@ -152,6 +154,19 @@ public static <T> Builder<T> builder() {
           .defaultValue(false)
           .build();
 
+  public static final PolarisConfiguration<List<String>> SUPPORTED_CATALOG_STORAGE_TYPES =
+      PolarisConfiguration.<List<String>>builder()
+          .key("SUPPORTED_CATALOG_STORAGE_TYPES")
+          .catalogConfig("supported.storage.types")
+          .description("The list of supported storage types for a catalog")
+          .defaultValue(
+              List.of(
+                  StorageConfigInfo.StorageTypeEnum.S3.name(),
+                  StorageConfigInfo.StorageTypeEnum.AZURE.name(),
+                  StorageConfigInfo.StorageTypeEnum.GCS.name(),
+                  StorageConfigInfo.StorageTypeEnum.FILE.name()))
+          .build();
+
   public static final PolarisConfiguration<Boolean> CLEANUP_ON_NAMESPACE_DROP =
       PolarisConfiguration.<Boolean>builder()
           .key("CLEANUP_ON_NAMESPACE_DROP")

diff --git a/polaris-core/src/main/java/org/apache/polaris/core/PolarisConfigurationStore.java b/polaris-core/src/main/java/org/apache/polaris/core/PolarisConfigurationStore.java
@@ -19,6 +19,7 @@
 package org.apache.polaris.core;
 
 import com.google.common.base.Preconditions;
+import java.util.List;
 import org.apache.polaris.core.entity.CatalogEntity;
 import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;
@@ -68,6 +69,8 @@ public interface PolarisConfigurationStore {
 
     if (config.defaultValue instanceof Boolean) {
       return config.cast(Boolean.valueOf(String.valueOf(value)));
+    } else if (config.defaultValue instanceof List<?>) {
+      return config.cast(List.copyOf((List<?>) value));
     } else {
       return config.cast(value);
     }

diff --git a/polaris-service/src/main/java/org/apache/polaris/service/admin/PolarisServiceImpl.java b/polaris-service/src/main/java/org/apache/polaris/service/admin/PolarisServiceImpl.java
@@ -25,6 +25,7 @@
 import org.apache.iceberg.catalog.TableIdentifier;
 import org.apache.iceberg.exceptions.NotAuthorizedException;
 import org.apache.polaris.core.PolarisCallContext;
+import org.apache.polaris.core.PolarisConfiguration;
 import org.apache.polaris.core.admin.model.AddGrantRequest;
 import org.apache.polaris.core.admin.model.Catalog;
 import org.apache.polaris.core.admin.model.CatalogGrant;
@@ -118,13 +119,7 @@ private void validateStorageConfig(StorageConfigInfo storageConfigInfo) {
         polarisCallContext
             .getConfigurationStore()
             .getConfiguration(
-                polarisCallContext,
-                "SUPPORTED_CATALOG_STORAGE_TYPES",
-                List.of(
-                    StorageConfigInfo.StorageTypeEnum.S3.name(),
-                    StorageConfigInfo.StorageTypeEnum.AZURE.name(),
-                    StorageConfigInfo.StorageTypeEnum.GCS.name(),
-                    StorageConfigInfo.StorageTypeEnum.FILE.name()));
+                polarisCallContext, PolarisConfiguration.SUPPORTED_CATALOG_STORAGE_TYPES);
     if (!allowedStorageTypes.contains(storageConfigInfo.getStorageType().name())) {
       LOGGER
           .atWarn()

diff --git a/polaris-service/src/main/java/org/apache/polaris/service/catalog/BasePolarisCatalog.java b/polaris-service/src/main/java/org/apache/polaris/service/catalog/BasePolarisCatalog.java
@@ -75,6 +75,7 @@
 import org.apache.iceberg.view.ViewUtil;
 import org.apache.polaris.core.PolarisCallContext;
 import org.apache.polaris.core.PolarisConfiguration;
+import org.apache.polaris.core.admin.model.StorageConfigInfo;
 import org.apache.polaris.core.auth.AuthenticatedPolarisPrincipal;
 import org.apache.polaris.core.catalog.PolarisCatalogHelpers;
 import org.apache.polaris.core.context.CallContext;
@@ -986,14 +987,23 @@ private void validateLocationsForTableLike(
           // }
         },
         () -> {
-          List<String> invalidLocations =
-              locations.stream()
-                  .filter(location -> location.startsWith("file:") || location.startsWith("http"))
-                  .collect(Collectors.toList());
-          if (!invalidLocations.isEmpty()) {
-            throw new ForbiddenException(
-                "Invalid locations '%s' for identifier '%s': File locations are not allowed",
-                invalidLocations, identifier);
+          List<String> allowedStorageTypes =
+              callContext
+                  .getPolarisCallContext()
+                  .getConfigurationStore()
+                  .getConfiguration(
+                      callContext.getPolarisCallContext(),
+                      PolarisConfiguration.SUPPORTED_CATALOG_STORAGE_TYPES);
+          if (!allowedStorageTypes.contains(StorageConfigInfo.StorageTypeEnum.FILE.name())) {
+            List<String> invalidLocations =
+                locations.stream()
+                    .filter(location -> location.startsWith("file:") || location.startsWith("http"))
+                    .collect(Collectors.toList());
+            if (!invalidLocations.isEmpty()) {
+              throw new ForbiddenException(
+                  "Invalid locations '%s' for identifier '%s': File locations are not allowed",
+                  invalidLocations, identifier);
+            }
           }
         });
   }

diff --git a/polaris-service/src/test/java/org/apache/polaris/service/catalog/BasePolarisCatalogTest.java b/polaris-service/src/test/java/org/apache/polaris/service/catalog/BasePolarisCatalogTest.java
@@ -688,9 +688,15 @@ public void testUpdateNotificationCreateTableWithLocalFilePrefix() {
         metadataLocation,
         TableMetadataParser.toJson(createSampleTableMetadata(metadataLocation)).getBytes(UTF_8));
 
-    Assertions.assertThatThrownBy(() -> catalog.sendNotification(table, request))
-        .isInstanceOf(ForbiddenException.class)
-        .hasMessageContaining("Invalid location");
+    PolarisCallContext polarisCallContext = callContext.getPolarisCallContext();
+    if (!polarisCallContext
+        .getConfigurationStore()
+        .getConfiguration(polarisCallContext, PolarisConfiguration.SUPPORTED_CATALOG_STORAGE_TYPES)
+        .contains("FILE")) {
+      Assertions.assertThatThrownBy(() -> catalog.sendNotification(table, request))
+          .isInstanceOf(ForbiddenException.class)
+          .hasMessageContaining("Invalid location");
+    }
   }
 
   @Test
@@ -748,9 +754,15 @@ public void testUpdateNotificationCreateTableWithHttpPrefix() {
         metadataLocation,
         TableMetadataParser.toJson(createSampleTableMetadata(metadataLocation)).getBytes(UTF_8));
 
-    Assertions.assertThatThrownBy(() -> catalog.sendNotification(table, request))
-        .isInstanceOf(ForbiddenException.class)
-        .hasMessageContaining("Invalid location");
+    PolarisCallContext polarisCallContext = callContext.getPolarisCallContext();
+    if (!polarisCallContext
+        .getConfigurationStore()
+        .getConfiguration(polarisCallContext, PolarisConfiguration.SUPPORTED_CATALOG_STORAGE_TYPES)
+        .contains("FILE")) {
+      Assertions.assertThatThrownBy(() -> catalog.sendNotification(table, request))
+          .isInstanceOf(ForbiddenException.class)
+          .hasMessageContaining("Invalid location");
+    }
 
     // It also fails if we try to use https
     final String httpsMetadataLocation = "https://maliciousdomain.com/metadata.json";
@@ -764,9 +776,14 @@ public void testUpdateNotificationCreateTableWithHttpPrefix() {
         httpsMetadataLocation,
         TableMetadataParser.toJson(createSampleTableMetadata(metadataLocation)).getBytes(UTF_8));
 
-    Assertions.assertThatThrownBy(() -> catalog.sendNotification(table, newRequest))
-        .isInstanceOf(ForbiddenException.class)
-        .hasMessageContaining("Invalid location");
+    if (!polarisCallContext
+        .getConfigurationStore()
+        .getConfiguration(polarisCallContext, PolarisConfiguration.SUPPORTED_CATALOG_STORAGE_TYPES)
+        .contains("FILE")) {
+      Assertions.assertThatThrownBy(() -> catalog.sendNotification(table, newRequest))
+          .isInstanceOf(ForbiddenException.class)
+          .hasMessageContaining("Invalid location");
+    }
   }
 
   @Test