Introduce an option to add object storage prefix to table locations

CHANGELOG.md

@@ -41,6 +41,9 @@ request adding CHANGELOG notes for breaking (!) changes and possibly other secti
 providing authentication parameters to Polaris. When the authentication type is set to `IMPLICIT`, 
 the authentication parameters are picked from the environment or configuration files. 
 
+- The `DEFAULT_LOCATION_OBJECT_STORAGE_PREFIX_ENABLED` feature was added to support placing tables
+at locations that better optimize for object storage.
+
 ### Changes
 
 ### Deprecations

LICENSE

@@ -222,6 +222,7 @@ This product includes code from Apache Iceberg.
 * service/common/src/main/java/org/apache/polaris/service/catalog/iceberg/CatalogHandlerUtils.java
 * plugins/spark/v3.5/spark/src/main/java/org/apache/polaris/spark/PolarisRESTCatalog.java
 * plugins/spark/v3.5/spark/src/main/java/org/apache/polaris/spark/SparkCatalog.java
+* service/common/src/main/java/org/apache/polaris/service/catalog/common/LocationUtils.java
 
 Copyright: Copyright 2017-2025 The Apache Software Foundation
 Home page: https://iceberg.apache.org

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

@@ -339,4 +339,17 @@ public static void enforceFeatureEnabledOrThrow(
                   + "to enforce this when new locations are added. Only supported by the JDBC metastore.")
           .defaultValue(false)
           .buildFeatureConfiguration();
+
+  public static final FeatureConfiguration<Boolean> DEFAULT_LOCATION_OBJECT_STORAGE_PREFIX_ENABLED =
+      PolarisConfiguration.<Boolean>builder()
+          .key("DEFAULT_LOCATION_OBJECT_STORAGE_PREFIX_ENABLED")
+          .catalogConfig("polaris.config.default-table-location-object-storage-prefix.enabled")
+          .description(
+              "When enabled, Iceberg tables and views created without a location specified will have a prefix "
+                  + "applied to the location within the catalog's base location, rather than a location directly "
+                  + "inside the parent namespace. Note that this requires ALLOW_EXTERNAL_TABLE_LOCATION to be "
+                  + "enabled, but with OPTIMIZED_SIBLING_CHECK enabled "
+                  + "it is still possible to enforce the uniqueness of table locations within a catalog.")
+          .defaultValue(false)
+          .buildFeatureConfiguration();
 }

polaris-core/src/main/java/org/apache/polaris/core/persistence/transactional/TransactionalMetaStoreManagerImpl.java

@@ -2328,7 +2328,12 @@ public Map<String, String> getInternalPropertyMap(
   public <T extends PolarisEntity & LocationBasedEntity>
       Optional<Optional<String>> hasOverlappingSiblings(
           @Nonnull PolarisCallContext callContext, T entity) {
-    return Optional.empty();
+    TransactionalPersistence ms = ((TransactionalPersistence) callContext.getMetaStore());
+    return ms.runInTransaction(
+        callContext,
+        () -> {
+          return callContext.getMetaStore().hasOverlappingSiblings(callContext, entity);
+        });
   }
 
   /** {@inheritDoc} */

polaris-core/src/main/java/org/apache/polaris/core/persistence/transactional/TreeMapMetaStore.java

@@ -81,6 +81,10 @@ public T read(String key) {
      */
     public List<T> readRange(String prefix) {
       TreeMapMetaStore.this.ensureReadTr();
+      if (prefix.isEmpty()) {
+        return new ArrayList<>(this.slice.values());
+      }
+
       // end of the key
       String endKey =
           prefix.substring(0, prefix.length() - 1)

polaris-core/src/main/java/org/apache/polaris/core/persistence/transactional/TreeMapTransactionalPersistenceImpl.java

@@ -35,8 +35,10 @@
 import org.apache.polaris.core.entity.PolarisChangeTrackingVersions;
 import org.apache.polaris.core.entity.PolarisEntitiesActiveKey;
 import org.apache.polaris.core.entity.PolarisEntity;
+import org.apache.polaris.core.entity.PolarisEntityConstants;
 import org.apache.polaris.core.entity.PolarisEntityCore;
 import org.apache.polaris.core.entity.PolarisEntityId;
+import org.apache.polaris.core.entity.PolarisEntitySubType;
 import org.apache.polaris.core.entity.PolarisEntityType;
 import org.apache.polaris.core.entity.PolarisGrantRecord;
 import org.apache.polaris.core.entity.PolarisPrincipalSecrets;
@@ -50,6 +52,7 @@
 import org.apache.polaris.core.storage.PolarisStorageConfigurationInfo;
 import org.apache.polaris.core.storage.PolarisStorageIntegration;
 import org.apache.polaris.core.storage.PolarisStorageIntegrationProvider;
+import org.apache.polaris.core.storage.StorageLocation;
 
 public class TreeMapTransactionalPersistenceImpl extends AbstractTransactionalPersistence {
 
@@ -667,11 +670,44 @@ record -> this.store.getSlicePolicyMappingRecordsByPolicy().delete(record));
         .readRange(this.store.buildPrefixKeyComposite(policyTypeCode, policyCatalogId, policyId));
   }
 
+  private Optional<String> getEntityLocationWithoutScheme(PolarisBaseEntity entity) {
+    if (entity.getType() == PolarisEntityType.TABLE_LIKE) {
+      if (entity.getSubType() == PolarisEntitySubType.ICEBERG_TABLE
+          || entity.getSubType() == PolarisEntitySubType.ICEBERG_VIEW) {
+        return Optional.of(
+            StorageLocation.of(
+                    entity.getPropertiesAsMap().get(PolarisEntityConstants.ENTITY_BASE_LOCATION))
+                .withoutScheme());
+      }
+    }
+    if (entity.getType() == PolarisEntityType.NAMESPACE) {
+      return Optional.of(
+          StorageLocation.of(
+                  entity.getPropertiesAsMap().get(PolarisEntityConstants.ENTITY_BASE_LOCATION))
+              .withoutScheme());
+    }
+    return Optional.empty();
+  }
+
   /** {@inheritDoc} */
   @Override
   public <T extends PolarisEntity & LocationBasedEntity>
       Optional<Optional<String>> hasOverlappingSiblings(
           @Nonnull PolarisCallContext callContext, T entity) {
-    return Optional.empty();
+    // TODO we could optimize this full scan
+    StorageLocation entityLocationWithoutScheme =
+        StorageLocation.of(StorageLocation.of(entity.getBaseLocation()).withoutScheme());
+    List<PolarisBaseEntity> allEntities = this.store.getSliceEntities().readRange("");
+    for (PolarisBaseEntity siblingEntity : allEntities) {
+      Optional<StorageLocation> maybeSiblingLocationWithoutScheme =
+          getEntityLocationWithoutScheme(siblingEntity).map(StorageLocation::of);
+      if (maybeSiblingLocationWithoutScheme.isPresent()) {
+        if (maybeSiblingLocationWithoutScheme.get().isChildOf(entityLocationWithoutScheme)
+            || entityLocationWithoutScheme.isChildOf(maybeSiblingLocationWithoutScheme.get())) {
+          return Optional.of(Optional.of(maybeSiblingLocationWithoutScheme.toString()));
+        }
+      }
+    }
+    return Optional.of(Optional.empty());
   }
 }

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

@@ -36,10 +36,10 @@
 import java.util.Optional;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
+import org.apache.polaris.core.PolarisCallContext;
 import org.apache.polaris.core.PolarisDiagnostics;
 import org.apache.polaris.core.admin.model.Catalog;
 import org.apache.polaris.core.config.FeatureConfiguration;
-import org.apache.polaris.core.context.CallContext;
 import org.apache.polaris.core.entity.CatalogEntity;
 import org.apache.polaris.core.entity.PolarisEntity;
 import org.apache.polaris.core.entity.PolarisEntityConstants;
@@ -136,12 +136,12 @@ public static PolarisStorageConfigurationInfo deserialize(
   }
 
   public static Optional<PolarisStorageConfigurationInfo> forEntityPath(
-      PolarisDiagnostics diagnostics, List<PolarisEntity> entityPath) {
+      PolarisCallContext callContext, List<PolarisEntity> entityPath) {
     return findStorageInfoFromHierarchy(entityPath)
         .map(
             storageInfo ->
                 deserialize(
-                    diagnostics,
+                    callContext.getDiagServices(),
                     storageInfo
                         .getInternalPropertiesAsMap()
                         .get(PolarisEntityConstants.getStorageConfigInfoPropertyName())))
@@ -162,11 +162,10 @@ public static Optional<PolarisStorageConfigurationInfo> forEntityPath(
                       .orElse(null);
               CatalogEntity catalog = CatalogEntity.of(entityPath.get(0));
               boolean allowEscape =
-                  CallContext.getCurrentContext()
-                      .getPolarisCallContext()
+                  callContext
                       .getConfigurationStore()
                       .getConfiguration(
-                          CallContext.getCurrentContext().getRealmContext(),
+                          callContext.getRealmContext(),
                           catalog,
                           FeatureConfiguration.ALLOW_UNSTRUCTURED_TABLE_LOCATION);
               if (!allowEscape