Disable custom namespace locations

integration-tests/src/main/java/org/apache/polaris/service/it/test/PolarisApplicationIntegrationTest.java

@@ -20,9 +20,11 @@
 
 import static org.apache.polaris.service.it.env.PolarisClient.polarisClient;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.awaitility.Awaitility.await;
 
+import com.google.common.collect.ImmutableMap;
 import jakarta.ws.rs.ProcessingException;
 import jakarta.ws.rs.client.Entity;
 import jakarta.ws.rs.client.Invocation;
@@ -67,6 +69,7 @@
 import org.apache.polaris.core.admin.model.PolarisCatalog;
 import org.apache.polaris.core.admin.model.PrincipalRole;
 import org.apache.polaris.core.admin.model.StorageConfigInfo;
+import org.apache.polaris.core.config.BehaviorChangeConfiguration;
 import org.apache.polaris.core.entity.CatalogEntity;
 import org.apache.polaris.core.entity.PolarisEntityConstants;
 import org.apache.polaris.service.it.env.ClientPrincipal;
@@ -76,6 +79,7 @@
 import org.apache.polaris.service.it.env.RestApi;
 import org.apache.polaris.service.it.ext.PolarisIntegrationTestExtension;
 import org.assertj.core.api.InstanceOfAssertFactories;
+import org.assertj.core.api.ThrowableAssert;
 import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeAll;
@@ -84,6 +88,8 @@
 import org.junit.jupiter.api.TestInfo;
 import org.junit.jupiter.api.extension.ExtendWith;
 import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
 
 /**
  * @implSpec This test expects the server to be configured with the following features configured:
@@ -186,11 +192,30 @@ private static void createCatalog(
       String principalRoleName,
       StorageConfigInfo storageConfig,
       String defaultBaseLocation) {
-    CatalogProperties props =
+    createCatalog(
+        catalogName,
+        catalogType,
+        principalRoleName,
+        storageConfig,
+        defaultBaseLocation,
+        ImmutableMap.of());
+  }
+
+  private static void createCatalog(
+      String catalogName,
+      Catalog.TypeEnum catalogType,
+      String principalRoleName,
+      StorageConfigInfo storageConfig,
+      String defaultBaseLocation,
+      Map<String, String> additionalProperties) {
+    CatalogProperties.Builder propsBuilder =
         CatalogProperties.builder(defaultBaseLocation)
             .addProperty(
-                CatalogEntity.REPLACE_NEW_LOCATION_PREFIX_WITH_CATALOG_DEFAULT_KEY, "file:/")
-            .build();
+                CatalogEntity.REPLACE_NEW_LOCATION_PREFIX_WITH_CATALOG_DEFAULT_KEY, "file:/");
+    for (var entry : additionalProperties.entrySet()) {
+      propsBuilder.addProperty(entry.getKey(), entry.getValue());
+    }
+    CatalogProperties props = propsBuilder.build();
     Catalog catalog =
         catalogType.equals(Catalog.TypeEnum.INTERNAL)
             ? PolarisCatalog.builder()
@@ -641,4 +666,53 @@ public void testRequestBodyTooLarge() throws Exception {
               });
     }
   }
+
+  @ParameterizedTest
+  @ValueSource(booleans = {true, false})
+  public void testNamespaceOutsideCatalog(boolean allowNamespaceLocationEscape) throws IOException {
+    String catalogName = client.newEntityName("testNamespaceOutsideCatalog_specificLocation");
+    String catalogLocation = baseLocation.resolve(catalogName + "/catalog").toString();
+    String badLocation = baseLocation.resolve(catalogName + "/ns").toString();
+    createCatalog(
+        catalogName,
+        Catalog.TypeEnum.INTERNAL,
+        principalRoleName,
+        FileStorageConfigInfo.builder(StorageConfigInfo.StorageTypeEnum.FILE)
+            .setAllowedLocations(List.of(catalogLocation))
+            .build(),
+        catalogLocation,
+        ImmutableMap.of(
+            BehaviorChangeConfiguration.ALLOW_NAMESPACE_CUSTOM_LOCATION.catalogConfig(),
+            String.valueOf(allowNamespaceLocationEscape)));
+    try (RESTSessionCatalog sessionCatalog = newSessionCatalog(catalogName)) {
+      SessionCatalog.SessionContext sessionContext = SessionCatalog.SessionContext.createEmpty();
+      sessionCatalog.createNamespace(sessionContext, Namespace.of("good_namespace"));
+      ThrowableAssert.ThrowingCallable createBadNamespace =
+          () ->
+              sessionCatalog.createNamespace(
+                  sessionContext,
+                  Namespace.of("bad_namespace"),
+                  ImmutableMap.of("location", badLocation));
+      if (!allowNamespaceLocationEscape) {
+        assertThatThrownBy(createBadNamespace)
+            .isInstanceOf(BadRequestException.class)
+            .hasMessageContaining("custom location");
+      } else {
+        assertThatCode(createBadNamespace).doesNotThrowAnyException();
+      }
+      ThrowableAssert.ThrowingCallable createBadChildGoodParent =
+          () ->
+              sessionCatalog.createNamespace(
+                  sessionContext,
+                  Namespace.of("good_namespace", "bad_child"),
+                  ImmutableMap.of("location", badLocation));
+      if (!allowNamespaceLocationEscape) {
+        assertThatThrownBy(createBadChildGoodParent)
+            .isInstanceOf(BadRequestException.class)
+            .hasMessageContaining("custom location");
+      } else {
+        assertThatCode(createBadChildGoodParent).doesNotThrowAnyException();
+      }
+    }
+  }
 }

integration-tests/src/main/java/org/apache/polaris/service/it/test/PolarisRestCatalogIntegrationBase.java

@@ -173,7 +173,8 @@ public abstract class PolarisRestCatalogIntegrationBase extends CatalogTests<RES
       Map.of(
           "polaris.config.allow.unstructured.table.location", "true",
           "polaris.config.allow.external.table.location", "true",
-          "polaris.config.list-pagination-enabled", "true");
+          "polaris.config.list-pagination-enabled", "true",
+          "polaris.config.namespace-custom-location.enabled", "true");
 
   /**
    * Get the storage configuration information for the catalog.

plugins/spark/v3.5/integration/src/intTest/java/org/apache/polaris/spark/quarkus/it/SparkIntegrationBase.java

@@ -100,6 +100,7 @@ public void before(
     CatalogProperties props = new CatalogProperties("s3://my-bucket/path/to/data");
     props.putAll(s3Container.getS3ConfigProperties());
     props.put("polaris.config.drop-with-purge.enabled", "true");
+    props.put("polaris.config.namespace-custom-location.enabled", "true");
     Catalog catalog =
         PolarisCatalog.builder()
             .setType(Catalog.TypeEnum.INTERNAL)

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

@@ -22,11 +22,11 @@
 
 /**
  * Internal configuration flags for non-feature behavior changes in Polaris. These flags control
- * subtle behavior adjustments and bug fixes, not user-facing catalog settings. They are intended
- * for internal use only, are inherently unstable, and may be removed at any time. When introducing
- * a new flag, consider the trade-off between maintenance burden and the risk of an unguarded
- * behavior change. Flags here are generally short-lived and should either be removed or promoted to
- * stable feature flags before the next release.
+ * subtle behavior adjustments and bug fixes, not user-facing settings. They are intended for
+ * internal use only, are inherently unstable, and may be removed at any time. When introducing a
+ * new flag, consider the trade-off between maintenance burden and the risk of an unguarded behavior
+ * change. Flags here are generally short-lived and should either be removed or promoted to stable
+ * feature flags before the next release.
  *
  * @param <T> The type of the configuration
  */
@@ -74,4 +74,14 @@ protected BehaviorChangeConfiguration(
                       + " the committed metadata again.")
               .defaultValue(true)
               .buildBehaviorChangeConfiguration();
+
+  public static final BehaviorChangeConfiguration<Boolean> ALLOW_NAMESPACE_CUSTOM_LOCATION =
+      PolarisConfiguration.<Boolean>builder()
+          .key("ALLOW_NAMESPACE_CUSTOM_LOCATION")
+          .catalogConfig("polaris.config.namespace-custom-location.enabled")
+          .description(
+              "If set to true, allow namespaces with completely arbitrary locations. This should not affect"
+                  + " credential vending.")
+          .defaultValue(false)
+          .buildBehaviorChangeConfiguration();
 }

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

@@ -209,10 +209,6 @@ public FeatureConfiguration<T> buildFeatureConfiguration() {
 
     public BehaviorChangeConfiguration<T> buildBehaviorChangeConfiguration() {
       validateOrThrow();
-      if (catalogConfig.isPresent() || catalogConfigUnsafe.isPresent()) {
-        throw new IllegalArgumentException(
-            "catalog configs are not valid for behavior change configs");
-      }
       BehaviorChangeConfiguration<T> config =
           new BehaviorChangeConfiguration<>(
               key, description, defaultValue, catalogConfig, catalogConfigUnsafe);

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

@@ -65,7 +65,7 @@ protected StorageLocation(@Nonnull String location) {
   }
 
   /** If a path doesn't end in `/`, this will add one */
-  protected static String ensureTrailingSlash(String location) {
+  public static String ensureTrailingSlash(String location) {
     if (location == null || location.endsWith("/")) {
       return location;
     } else {

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

@@ -67,4 +67,19 @@ public String getScheme() {
   public String withoutScheme() {
     return locationWithoutScheme;
   }
+
+  @Override
+  public int hashCode() {
+    return withoutScheme().hashCode();
+  }
+
+  /** Checks if two S3Location instances represent the same physical location. */
+  @Override
+  public boolean equals(Object obj) {
+    if (obj instanceof S3Location) {
+      return withoutScheme().equals(((StorageLocation) obj).withoutScheme());
+    } else {
+      return false;
+    }
+  }
 }

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

@@ -126,4 +126,19 @@ public static boolean isAzureLocation(String location) {
     Matcher matcher = URI_PATTERN.matcher(location);
     return matcher.matches();
   }
+
+  @Override
+  public int hashCode() {
+    return withoutScheme().hashCode();
+  }
+
+  /** Checks if two AzureLocation instances represent the same physical location. */
+  @Override
+  public boolean equals(Object obj) {
+    if (obj instanceof AzureLocation) {
+      return withoutScheme().equals(((StorageLocation) obj).withoutScheme());
+    } else {
+      return false;
+    }
+  }
 }

polaris-core/src/test/java/org/apache/polaris/core/storage/aws/S3LocationTest.java

@@ -47,6 +47,7 @@ public void testPrefixValidationIgnoresScheme(String parentScheme, String childS
     Assertions.assertThat(loc1.isChildOf(loc2)).isTrue();
 
     StorageLocation loc3 = StorageLocation.of(childScheme + "://bucket/schema1");
-    Assertions.assertThat(loc2.equals(loc3)).isFalse();
+    Assertions.assertThat(loc2.toString().equals(loc3.toString())).isFalse();
+    Assertions.assertThat(loc2.equals(loc3)).isTrue();
   }
 }

regtests/t_cli/src/test_cli.py

@@ -110,6 +110,8 @@ def test_quickstart_flow():
             ROLE_ARN,
             '--default-base-location',
             f's3://fake-location-{SALT}',
+            '--property',
+            'polaris.config.namespace-custom-location.enabled=true',
             f'test_cli_catalog_{SALT}'), checker=lambda s: s == '')
         check_output(root_cli('catalogs', 'list'),
                      checker=lambda s: f'test_cli_catalog_{SALT}' in s)
@@ -170,7 +172,7 @@ def test_quickstart_flow():
             '--property',
             'foo=bar',
             '--location',
-            's3://custom-namespace-location'
+            f's3://fake-location-{SALT}/custom-namespace-location/'
         ), checker=lambda s: s == '')
         check_output(cli(user_token)('namespaces', 'list', '--catalog', f'test_cli_catalog_{SALT}'),
                      checker=lambda s: f'test_cli_namespace_{SALT}' in s)
@@ -180,7 +182,7 @@ def test_quickstart_flow():
             '--catalog',
             f'test_cli_catalog_{SALT}',
             f'test_cli_namespace_{SALT}'
-        ), checker=lambda s: 's3://custom-namespace-location' in s and '"foo": "bar"' in s)
+        ), checker=lambda s: f's3://fake-location-{SALT}/custom-namespace-location/' in s and '"foo": "bar"' in s)
         check_output(cli(user_token)(
             'namespaces',
             'delete',

runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/IcebergCatalog.java

@@ -508,6 +508,10 @@ private void createNamespaceInternal(
     } else {
       LOGGER.debug("Skipping location overlap validation for namespace '{}'", namespace);
     }
+    if (!realmConfig.getConfig(
+        BehaviorChangeConfiguration.ALLOW_NAMESPACE_CUSTOM_LOCATION, catalogEntity)) {
+      validateNamespaceLocation(entity, resolvedParent);
+    }
     PolarisEntity returnedEntity =
         PolarisEntity.of(
             getMetaStoreManager()
@@ -671,6 +675,12 @@ public boolean setProperties(Namespace namespace, Map<String, String> properties
     } else {
       LOGGER.debug("Skipping location overlap validation for namespace '{}'", namespace);
     }
+    if (!realmConfig.getConfig(
+        BehaviorChangeConfiguration.ALLOW_NAMESPACE_CUSTOM_LOCATION, catalogEntity)) {
+      if (properties.containsKey(PolarisEntityConstants.ENTITY_BASE_LOCATION)) {
+        validateNamespaceLocation(NamespaceEntity.of(entity), resolvedEntities);
+      }
+    }
 
     List<PolarisEntity> parentPath = resolvedEntities.getRawFullPath();
     PolarisEntity returnedEntity =
@@ -1085,6 +1095,76 @@ private void validateNoLocationOverlap(
     }
   }
 
+  /** Checks whether the location of a namespace is valid given its parent */
+  private void validateNamespaceLocation(
+      NamespaceEntity namespace, PolarisResolvedPathWrapper resolvedParent) {
+    StorageLocation namespaceLocation =
+        StorageLocation.of(
+            StorageLocation.ensureTrailingSlash(
+                resolveNamespaceLocation(namespace.asNamespace(), namespace.getPropertiesAsMap())));
+    PolarisEntity parent = resolvedParent.getResolvedLeafEntity().getEntity();
+    Preconditions.checkArgument(
+        parent.getType().equals(PolarisEntityType.CATALOG)
+            || parent.getType().equals(PolarisEntityType.NAMESPACE),
+        "Invalid parent type");
+    if (parent.getType().equals(PolarisEntityType.CATALOG)) {
+      CatalogEntity parentEntity = CatalogEntity.of(parent);
+      LOGGER.debug(
+          "Validating namespace {} given parent catalog {}",
+          namespace.getName(),
+          parentEntity.getName());
+      var storageConfigInfo = parentEntity.getStorageConfigurationInfo();
+      if (storageConfigInfo == null) {
+        throw new IllegalArgumentException(
+            "Cannot create namespace without a parent storage configuration");
+      }
+      List<StorageLocation> defaultLocations =
+          parentEntity.getStorageConfigurationInfo().getAllowedLocations().stream()
+              .filter(java.util.Objects::nonNull)
+              .map(
+                  l ->
+                      StorageLocation.ensureTrailingSlash(
+                          StorageLocation.ensureTrailingSlash(l) + namespace.getName()))
+              .map(StorageLocation::of)
+              .toList();
+      if (!defaultLocations.contains(namespaceLocation)) {
+        throw new IllegalArgumentException(
+            "Namespace "
+                + namespace.getName()
+                + " has a custom location, "
+                + "which is not enabled. Expected a location in: ["
+                + String.join(
+                    ", ", defaultLocations.stream().map(StorageLocation::toString).toList())
+                + "]. Got location: "
+                + namespaceLocation
+                + "]");
+      }
+    } else if (parent.getType().equals(PolarisEntityType.NAMESPACE)) {
+      NamespaceEntity parentEntity = NamespaceEntity.of(parent);
+      LOGGER.debug(
+          "Validating namespace {} given parent namespace {}",
+          namespace.getName(),
+          parentEntity.getName());
+      String parentLocation =
+          resolveNamespaceLocation(parentEntity.asNamespace(), parentEntity.getPropertiesAsMap());
+      StorageLocation defaultLocation =
+          StorageLocation.of(
+              StorageLocation.ensureTrailingSlash(
+                  StorageLocation.ensureTrailingSlash(parentLocation) + namespace.getName()));
+      if (!defaultLocation.equals(namespaceLocation)) {
+        throw new IllegalArgumentException(
+            "Namespace "
+                + namespace.getName()
+                + " has a custom location, "
+                + "which is not enabled. Expected location: ["
+                + defaultLocation
+                + "]. Got location: ["
+                + namespaceLocation
+                + "]");
+      }
+    }
+  }
+
   /**
    * Validate no location overlap exists between the entity path and its sibling entities. This
    * resolves all siblings at the same level as the target entity (namespaces if the target entity

runtime/service/src/test/java/org/apache/polaris/service/admin/PolarisAuthzTestBase.java

@@ -119,6 +119,7 @@ public Map<String, String> getConfigOverrides() {
               "polaris.features.\"ENFORCE_PRINCIPAL_CREDENTIAL_ROTATION_REQUIRED_CHECKING\"",
               "true")
           .put("polaris.features.\"DROP_WITH_PURGE_ENABLED\"", "true")
+          .put("polaris.behavior-changes.\"ALLOW_NAMESPACE_CUSTOM_LOCATION\"", "true")
           .build();
     }
   }

runtime/service/src/test/java/org/apache/polaris/service/catalog/iceberg/AbstractIcebergCatalogTest.java

@@ -199,6 +199,7 @@ public Map<String, String> getConfigOverrides() {
           .putAll(super.getConfigOverrides())
           .put("polaris.features.\"ALLOW_TABLE_LOCATION_OVERLAP\"", "true")
           .put("polaris.features.\"LIST_PAGINATION_ENABLED\"", "true")
+          .put("polaris.features.\"ALLOW_NAMESPACE_CUSTOM_LOCATION\"", "true")
           .build();
     }
   }

runtime/service/src/test/java/org/apache/polaris/service/it/RestCatalogFileIntegrationTest.java

@@ -42,6 +42,8 @@ public Map<String, String> getConfigOverrides() {
           "polaris.features.\"SUPPORTED_CATALOG_STORAGE_TYPES\"",
           "[\"FILE\",\"S3\"]",
           "polaris.readiness.ignore-severe-issues",
+          "true",
+          "polaris.features.\"ALLOW_NAMESPACE_CUSTOM_LOCATION\"",
           "true");
     }
   }