Adjustable limit on the number of locations per storage config

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

@@ -43,4 +43,13 @@ protected BehaviorChangeConfiguration(
           .description("If true, validate that view locations don't overlap when views are created")
           .defaultValue(true)
           .buildBehaviorChangeConfiguration();
+
+  public static final BehaviorChangeConfiguration<Integer> STORAGE_CONFIGURATION_MAX_LOCATIONS =
+      PolarisConfiguration.<Integer>builder()
+          .key("STORAGE_CONFIGURATION_MAX_LOCATIONS")
+          .description(
+              "How many locations can be associated with a storage configuration, or -1 for"
+                  + " unlimited locations")
+          .defaultValue(-1)
+          .buildBehaviorChangeConfiguration();
 }

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

@@ -23,6 +23,7 @@
 import jakarta.annotation.Nonnull;
 import jakarta.annotation.Nullable;
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Map;
@@ -39,6 +40,7 @@
 import org.apache.polaris.core.admin.model.GcpStorageConfigInfo;
 import org.apache.polaris.core.admin.model.PolarisCatalog;
 import org.apache.polaris.core.admin.model.StorageConfigInfo;
+import org.apache.polaris.core.config.BehaviorChangeConfiguration;
 import org.apache.polaris.core.storage.FileStorageConfigurationInfo;
 import org.apache.polaris.core.storage.PolarisStorageConfigurationInfo;
 import org.apache.polaris.core.storage.aws.AwsStorageConfigurationInfo;
@@ -237,6 +239,7 @@ public Builder setStorageConfigurationInfo(
           throw new BadRequestException("Must specify default base location");
         }
         allowedLocations.add(defaultBaseLocation);
+        validateMaxAllowedLocations(allowedLocations);
         switch (storageConfigModel.getStorageType()) {
           case S3:
             AwsStorageConfigInfo awsConfigModel = (AwsStorageConfigInfo) storageConfigModel;
@@ -279,6 +282,19 @@ public Builder setStorageConfigurationInfo(
       return this;
     }
 
+    /** Validate the number of allowed locations not exceeding the max value. */
+    private void validateMaxAllowedLocations(Collection<String> allowedLocations) {
+      int maxAllowedLocations =
+          BehaviorChangeConfiguration.loadConfig(
+              BehaviorChangeConfiguration.STORAGE_CONFIGURATION_MAX_LOCATIONS);
+      if (maxAllowedLocations != -1 && allowedLocations.size() > maxAllowedLocations) {
+        throw new IllegalArgumentException(
+            String.format(
+                "Number of configured locations (%s) exceeds the limit of %s",
+                allowedLocations.size(), maxAllowedLocations));
+      }
+    }
+
     @Override
     public CatalogEntity build() {
       return new CatalogEntity(buildBase());

polaris-core/src/main/java/org/apache/polaris/core/storage/PolarisStorageConfigurationInfo.java

@@ -230,14 +230,6 @@ protected void validatePrefixForStorageType(String loc) {
     }
   }
 
-  /** Validate the number of allowed locations not exceeding the max value. */
-  public void validateMaxAllowedLocations(int maxAllowedLocations) {
-    if (allowedLocations.size() > maxAllowedLocations) {
-      throw new IllegalArgumentException(
-          "Number of allowed locations exceeds " + maxAllowedLocations);
-    }
-  }
-
   /** Polaris' storage type, each has a fixed prefix for its location */
   public enum StorageType {
     S3("s3://"),

polaris-core/src/main/java/org/apache/polaris/core/storage/StorageConfigurationOverride.java

@@ -49,9 +49,4 @@ public String getFileIoImplClassName() {
   protected void validatePrefixForStorageType(String loc) {
     parentStorageConfiguration.validatePrefixForStorageType(loc);
   }
-
-  @Override
-  public void validateMaxAllowedLocations(int maxAllowedLocations) {
-    parentStorageConfiguration.validateMaxAllowedLocations(maxAllowedLocations);
-  }
 }

polaris-core/src/main/java/org/apache/polaris/core/storage/aws/AwsStorageConfigurationInfo.java

@@ -32,10 +32,6 @@
 /** Aws Polaris Storage Configuration information */
 public class AwsStorageConfigurationInfo extends PolarisStorageConfigurationInfo {
 
-  // 5 is the approximate max allowed locations for the size of AccessPolicy when LIST is required
-  // for allowed read and write locations for subscoping creds.
-  @JsonIgnore private static final int MAX_ALLOWED_LOCATIONS = 5;
-
   // Technically, it should be ^arn:(aws|aws-cn|aws-us-gov):iam::(\d{12}):role/.+$,
   @JsonIgnore
   public static final String ROLE_ARN_PATTERN = "^arn:(aws|aws-us-gov):iam::(\\d{12}):role/.+$";
@@ -75,7 +71,6 @@ public AwsStorageConfigurationInfo(
     this.roleARN = roleARN;
     this.externalId = externalId;
     this.region = region;
-    validateMaxAllowedLocations(MAX_ALLOWED_LOCATIONS);
   }
 
   @Override

polaris-core/src/main/java/org/apache/polaris/core/storage/azure/AzureStorageConfigurationInfo.java

@@ -19,7 +19,6 @@
 package org.apache.polaris.core.storage.azure;
 
 import com.fasterxml.jackson.annotation.JsonCreator;
-import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.google.common.base.MoreObjects;
 import jakarta.annotation.Nonnull;
@@ -30,9 +29,6 @@
 
 /** Azure storage configuration information. */
 public class AzureStorageConfigurationInfo extends PolarisStorageConfigurationInfo {
-  // technically there is no limitation since expectation for Azure locations are for the same
-  // storage account and same container
-  @JsonIgnore private static final int MAX_ALLOWED_LOCATIONS = 20;
 
   // Azure tenant id
   private final @Nonnull String tenantId;
@@ -52,7 +48,6 @@ public AzureStorageConfigurationInfo(
       @JsonProperty(value = "tenantId", required = true) @Nonnull String tenantId) {
     super(StorageType.AZURE, allowedLocations);
     this.tenantId = tenantId;
-    validateMaxAllowedLocations(MAX_ALLOWED_LOCATIONS);
   }
 
   @Override

polaris-core/src/main/java/org/apache/polaris/core/storage/gcp/GcpStorageConfigurationInfo.java

@@ -19,7 +19,6 @@
 package org.apache.polaris.core.storage.gcp;
 
 import com.fasterxml.jackson.annotation.JsonCreator;
-import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.google.common.base.MoreObjects;
 import jakarta.annotation.Nonnull;
@@ -30,11 +29,6 @@
 /** Gcp storage storage configuration information. */
 public class GcpStorageConfigurationInfo extends PolarisStorageConfigurationInfo {
 
-  // 8 is an experimental result from generating GCP accessBoundaryRules when subscoping creds,
-  // when the rule is too large, GCS only returns error: 400 bad request "Invalid arguments
-  // provided in the request"
-  @JsonIgnore private static final int MAX_ALLOWED_LOCATIONS = 8;
-
   /** The gcp service account */
   @JsonProperty(value = "gcpServiceAccount", required = false)
   private @Nullable String gcpServiceAccount = null;
@@ -44,7 +38,6 @@ public GcpStorageConfigurationInfo(
       @JsonProperty(value = "allowedLocations", required = true) @Nonnull
           List<String> allowedLocations) {
     super(StorageType.GCS, allowedLocations);
-    validateMaxAllowedLocations(MAX_ALLOWED_LOCATIONS);
   }
 
   @Override

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

@@ -28,6 +28,7 @@
 import org.apache.polaris.core.config.PolarisConfigurationStore;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
 
 /** Unit test for the default behaviors of the PolarisConfigurationStore interface. */
 public class PolarisConfigurationStoreTest {
@@ -64,8 +65,9 @@ public void testConfigsCanBeCastedFromString() {
 
     // Ensure that we can fetch all the configs and that the value is what we expect, which
     // is the config's default value based on how we've implemented PolarisConfigurationStore above.
+    PolarisCallContext polarisCallContext = Mockito.mock(PolarisCallContext.class);
     for (PolarisConfiguration<?> c : configs) {
-      Assertions.assertEquals(c.defaultValue, store.getConfiguration(null, c));
+      Assertions.assertEquals(c.defaultValue, store.getConfiguration(polarisCallContext, c));
     }
   }
 
@@ -84,8 +86,10 @@ public <T> T getConfiguration(PolarisCallContext ctx, String configName) {
           }
         };
 
+    PolarisCallContext polarisCallContext = Mockito.mock(PolarisCallContext.class);
     for (PolarisConfiguration<?> c : configs) {
-      Assertions.assertThrows(NumberFormatException.class, () -> store.getConfiguration(null, c));
+      Assertions.assertThrows(
+          NumberFormatException.class, () -> store.getConfiguration(polarisCallContext, c));
     }
   }
 

quarkus/service/src/test/java/org/apache/polaris/service/quarkus/entity/CatalogEntityTest.java

@@ -19,21 +19,38 @@
 package org.apache.polaris.service.quarkus.entity;
 
 import java.util.List;
+import org.apache.polaris.core.PolarisCallContext;
+import org.apache.polaris.core.PolarisDefaultDiagServiceImpl;
 import org.apache.polaris.core.admin.model.AwsStorageConfigInfo;
 import org.apache.polaris.core.admin.model.AzureStorageConfigInfo;
 import org.apache.polaris.core.admin.model.Catalog;
 import org.apache.polaris.core.admin.model.CatalogProperties;
 import org.apache.polaris.core.admin.model.GcpStorageConfigInfo;
 import org.apache.polaris.core.admin.model.PolarisCatalog;
 import org.apache.polaris.core.admin.model.StorageConfigInfo;
+import org.apache.polaris.core.context.CallContext;
 import org.apache.polaris.core.entity.CatalogEntity;
+import org.apache.polaris.core.persistence.MetaStoreManagerFactory;
+import org.apache.polaris.service.persistence.InMemoryPolarisMetaStoreManagerFactory;
 import org.assertj.core.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.ValueSource;
 
 public class CatalogEntityTest {
 
+  @BeforeAll
+  public static void setup() {
+    MetaStoreManagerFactory metaStoreManagerFactory = new InMemoryPolarisMetaStoreManagerFactory();
+    PolarisCallContext polarisCallContext =
+        new PolarisCallContext(
+            metaStoreManagerFactory.getOrCreateSessionSupplier(() -> "realm").get(),
+            new PolarisDefaultDiagServiceImpl());
+    CallContext callContext = CallContext.of(() -> "realm", polarisCallContext);
+    CallContext.setCurrentContext(callContext);
+  }
+
   @Test
   public void testInvalidAllowedLocationPrefix() {
     String storageLocation = "unsupportPrefix://mybucket/path";
@@ -123,9 +140,8 @@ public void testExceedMaxAllowedLocations() {
             .setProperties(prop)
             .setStorageConfigInfo(awsStorageConfigModel)
             .build();
-    Assertions.assertThatThrownBy(() -> CatalogEntity.fromCatalog(awsCatalog))
-        .isInstanceOf(IllegalArgumentException.class)
-        .hasMessageContaining("Number of allowed locations exceeds 5");
+    Assertions.assertThatCode(() -> CatalogEntity.fromCatalog(awsCatalog))
+        .doesNotThrowAnyException();
   }
 
   @Test