Optimize the location overlap check with an index

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

@@ -334,7 +334,7 @@ public void testIcebergCreateNamespaceInExternalCatalog() throws IOException {
           .isNotNull()
           .isNotEmpty()
           .containsEntry(
-              PolarisEntityConstants.ENTITY_BASE_LOCATION, "s3://my-bucket/path/to/data/db1");
+              PolarisEntityConstants.ENTITY_BASE_LOCATION, "s3://my-bucket/path/to/data/db1/");
     }
   }
 

persistence/eclipselink/src/main/java/org/apache/polaris/extension/persistence/impl/eclipselink/PolarisEclipseLinkMetaStoreSessionImpl.java

@@ -32,6 +32,7 @@
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
+import java.util.Optional;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.function.Function;
 import java.util.function.Predicate;
@@ -41,9 +42,11 @@
 import org.apache.polaris.core.PolarisCallContext;
 import org.apache.polaris.core.context.RealmContext;
 import org.apache.polaris.core.entity.EntityNameLookupRecord;
+import org.apache.polaris.core.entity.LocationBasedEntity;
 import org.apache.polaris.core.entity.PolarisBaseEntity;
 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.PolarisEntityCore;
 import org.apache.polaris.core.entity.PolarisEntityId;
 import org.apache.polaris.core.entity.PolarisEntityType;
@@ -778,4 +781,12 @@ public void rollback() {
       session.getTransaction().rollback();
     }
   }
+
+  /** {@inheritDoc} */
+  @Override
+  public <T extends PolarisEntity & LocationBasedEntity>
+      Optional<Optional<String>> hasOverlappingSiblings(
+          @Nonnull PolarisCallContext callContext, T entity) {
+    return Optional.empty();
+  }
 }

persistence/relational-jdbc/src/main/java/org/apache/polaris/persistence/relational/jdbc/DatabaseType.java

@@ -44,6 +44,6 @@ public static DatabaseType fromDisplayName(String displayName) {
   }
 
   public String getInitScriptResource() {
-    return String.format("%s/schema-v1.sql", this.getDisplayName());
+    return String.format("%s/schema-v2.sql", this.getDisplayName());
   }
 }

persistence/relational-jdbc/src/main/java/org/apache/polaris/persistence/relational/jdbc/JdbcBasePersistenceImpl.java

@@ -31,13 +31,16 @@
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.function.Function;
 import java.util.function.Predicate;
 import java.util.stream.Collectors;
 import org.apache.polaris.core.PolarisCallContext;
 import org.apache.polaris.core.entity.EntityNameLookupRecord;
+import org.apache.polaris.core.entity.LocationBasedEntity;
 import org.apache.polaris.core.entity.PolarisBaseEntity;
 import org.apache.polaris.core.entity.PolarisChangeTrackingVersions;
+import org.apache.polaris.core.entity.PolarisEntity;
 import org.apache.polaris.core.entity.PolarisEntityCore;
 import org.apache.polaris.core.entity.PolarisEntityId;
 import org.apache.polaris.core.entity.PolarisEntityType;
@@ -59,10 +62,12 @@
 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;
 import org.apache.polaris.persistence.relational.jdbc.models.ModelEntity;
 import org.apache.polaris.persistence.relational.jdbc.models.ModelGrantRecord;
 import org.apache.polaris.persistence.relational.jdbc.models.ModelPolicyMappingRecord;
 import org.apache.polaris.persistence.relational.jdbc.models.ModelPrincipalAuthenticationData;
+import org.apache.polaris.persistence.relational.jdbc.models.SchemaVersion;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -74,6 +79,10 @@ public class JdbcBasePersistenceImpl implements BasePersistence, IntegrationPers
   private final PrincipalSecretsGenerator secretsGenerator;
   private final PolarisStorageIntegrationProvider storageIntegrationProvider;
   private final String realmId;
+  private final int version;
+
+  // The max number of components a location can have before the optimized sibling check is not used
+  private static final int MAX_LOCATION_COMPONENTS = 40;
 
   public JdbcBasePersistenceImpl(
       DatasourceOperations databaseOperations,
@@ -84,6 +93,7 @@ public JdbcBasePersistenceImpl(
     this.secretsGenerator = secretsGenerator;
     this.storageIntegrationProvider = storageIntegrationProvider;
     this.realmId = realmId;
+    this.version = loadVersion();
   }
 
   @Override
@@ -622,6 +632,64 @@ public boolean hasChildren(
     }
   }
 
+  private int loadVersion() {
+    PreparedQuery query = QueryGenerator.generateVersionQuery();
+    try {
+      List<SchemaVersion> schemaVersion =
+          datasourceOperations.executeSelect(query, new SchemaVersion());
+      if (schemaVersion == null || schemaVersion.size() != 1) {
+        throw new RuntimeException("Failed to retrieve schema version");
+      }
+      return schemaVersion.getFirst().getValue();
+    } catch (SQLException e) {
+      LOGGER.error("Failed to load schema version due to {}", e.getMessage(), e);
+      throw new IllegalStateException("Failed to retrieve schema version", e);
+    }
+  }
+
+  /** {@inheritDoc} */
+  @Override
+  public <T extends PolarisEntity & LocationBasedEntity>
+      Optional<Optional<String>> hasOverlappingSiblings(
+          @Nonnull PolarisCallContext callContext, T entity) {
+    if (this.version < 2) {
+      return Optional.empty();
+    }
+    if (entity.getBaseLocation().chars().filter(ch -> ch == '/').count()
+        > MAX_LOCATION_COMPONENTS) {
+      return Optional.empty();
+    }
+
+    PreparedQuery query =
+        QueryGenerator.generateOverlapQuery(
+            realmId, entity.getCatalogId(), entity.getBaseLocation());
+    try {
+      var results = datasourceOperations.executeSelect(query, new ModelEntity());
+      if (!results.isEmpty()) {
+        StorageLocation entityLocation = StorageLocation.of(entity.getBaseLocation());
+        for (PolarisBaseEntity result : results) {
+          StorageLocation potentialSiblingLocation =
+              StorageLocation.of(((LocationBasedEntity) result).getBaseLocation());
+          if (entityLocation.isChildOf(potentialSiblingLocation)
+              || potentialSiblingLocation.isChildOf(entityLocation)) {
+            return Optional.of(Optional.of(potentialSiblingLocation.toString()));
+          }
+        }
+      }
+      return Optional.of(Optional.empty());
+    } catch (SQLException e) {
+      LOGGER.error(
+          "Failed to retrieve location overlap for location {} due to {}",
+          entity.getBaseLocation(),
+          e.getMessage(),
+          e);
+      throw new RuntimeException(
+          String.format(
+              "Failed to retrieve location overlap for location: %s", entity.getBaseLocation()),
+          e);
+    }
+  }
+
   @Nullable
   @Override
   public PolarisPrincipalSecrets loadPrincipalSecrets(

persistence/relational-jdbc/src/main/java/org/apache/polaris/persistence/relational/jdbc/QueryGenerator.java

@@ -30,6 +30,7 @@
 import java.util.stream.Collectors;
 import org.apache.polaris.core.entity.PolarisEntityCore;
 import org.apache.polaris.core.entity.PolarisEntityId;
+import org.apache.polaris.core.storage.StorageLocation;
 import org.apache.polaris.persistence.relational.jdbc.models.ModelEntity;
 import org.apache.polaris.persistence.relational.jdbc.models.ModelGrantRecord;
 
@@ -208,6 +209,59 @@ static QueryFragment generateWhereClause(
     return new QueryFragment(clause, parameters);
   }
 
+  @VisibleForTesting
+  static PreparedQuery generateVersionQuery() {
+    return new PreparedQuery("SELECT version_value FROM POLARIS_SCHEMA.VERSION", List.of());
+  }
+
+  /**
+   * Generate a SELECT query to find any entities that have a given realm & parent and that may with
+   * a given location. The check is performed without consideration for the scheme, so a path on one
+   * storage type may give a false positive for overlapping with another storage type. This should
+   * be combined with a check using `StorageLocation`.
+   *
+   * @param realmId A realm to search within
+   * @param catalogId A catalog entity to search within
+   * @param baseLocation The base location to look for overlap with, with or without a scheme
+   * @return The list of possibly overlapping entities that meet the criteria
+   */
+  @VisibleForTesting
+  public static PreparedQuery generateOverlapQuery(
+      String realmId, long catalogId, String baseLocation) {
+    StorageLocation baseStorageLocation = StorageLocation.of(baseLocation);
+    String locationWithoutScheme = baseStorageLocation.withoutScheme();
+
+    List<String> conditions = new ArrayList<>();
+    List<Object> parameters = new ArrayList<>();
+
+    String[] components = locationWithoutScheme.split("/");
+    StringBuilder pathBuilder = new StringBuilder();
+
+    for (String component : components) {
+      pathBuilder.append(component).append("/");
+      conditions.add("location_without_scheme = ?");
+      parameters.add(pathBuilder.toString());
+    }
+
+    // Add LIKE condition to match children
+    conditions.add("location_without_scheme LIKE ?");
+    parameters.add(locationWithoutScheme + "%");
+
+    String locationClause = String.join(" OR ", conditions);
+    String clause = " WHERE realm_id = ? AND catalog_id = ? AND (" + locationClause + ")";
+
+    // realmId and parentId go first
+    List<Object> finalParams = new ArrayList<>();
+    finalParams.add(realmId);
+    finalParams.add(catalogId);
+    finalParams.addAll(parameters);
+
+    QueryFragment where = new QueryFragment(clause, finalParams);
+    PreparedQuery query =
+        generateSelectQuery(ModelEntity.ALL_COLUMNS, ModelEntity.TABLE_NAME, where.sql());
+    return new PreparedQuery(query.sql(), where.parameters());
+  }
+
   private static String getFullyQualifiedTableName(String tableName) {
     // TODO: make schema name configurable.
     return "POLARIS_SCHEMA." + tableName;

persistence/relational-jdbc/src/main/java/org/apache/polaris/persistence/relational/jdbc/models/ModelEntity.java

@@ -24,8 +24,10 @@
 import java.util.List;
 import java.util.Map;
 import org.apache.polaris.core.entity.PolarisBaseEntity;
+import org.apache.polaris.core.entity.PolarisEntityConstants;
 import org.apache.polaris.core.entity.PolarisEntitySubType;
 import org.apache.polaris.core.entity.PolarisEntityType;
+import org.apache.polaris.core.storage.StorageLocation;
 import org.apache.polaris.persistence.relational.jdbc.DatabaseType;
 
 public class ModelEntity implements Converter<PolarisBaseEntity> {
@@ -47,7 +49,8 @@ public class ModelEntity implements Converter<PolarisBaseEntity> {
           "last_update_timestamp",
           "properties",
           "internal_properties",
-          "grant_records_version");
+          "grant_records_version",
+          "location_without_scheme");
 
   // the id of the catalog associated to that entity. use 0 if this entity is top-level
   // like a catalog
@@ -95,6 +98,9 @@ public class ModelEntity implements Converter<PolarisBaseEntity> {
   // current version for that entity, will be monotonically incremented
   private int grantRecordsVersion;
 
+  // location for the entity but without a scheme, when applicable
+  private String locationWithoutScheme;
+
   public long getId() {
     return id;
   }
@@ -155,6 +161,10 @@ public int getGrantRecordsVersion() {
     return grantRecordsVersion;
   }
 
+  public String getLocationWithoutScheme() {
+    return locationWithoutScheme;
+  }
+
   public static Builder builder() {
     return new Builder();
   }
@@ -180,6 +190,7 @@ public PolarisBaseEntity fromResultSet(ResultSet r) throws SQLException {
             // JSONB: use getString(), not getObject().
             .internalProperties(r.getString("internal_properties"))
             .grantRecordsVersion(r.getObject("grant_records_version", Integer.class))
+            .locationWithoutScheme(r.getString("location_without_scheme"))
             .build();
 
     return toEntity(modelEntity);
@@ -208,6 +219,7 @@ public Map<String, Object> toMap(DatabaseType databaseType) {
       map.put("internal_properties", this.getInternalProperties());
     }
     map.put("grant_records_version", this.getGrantRecordsVersion());
+    map.put("location_without_scheme", this.getLocationWithoutScheme());
     return map;
   }
 
@@ -293,29 +305,52 @@ public Builder grantRecordsVersion(int grantRecordsVersion) {
       return this;
     }
 
+    public Builder locationWithoutScheme(String location) {
+      entity.locationWithoutScheme = location;
+      return this;
+    }
+
     public ModelEntity build() {
       return entity;
     }
   }
 
   public static ModelEntity fromEntity(PolarisBaseEntity entity) {
-    return ModelEntity.builder()
-        .catalogId(entity.getCatalogId())
-        .id(entity.getId())
-        .parentId(entity.getParentId())
-        .typeCode(entity.getTypeCode())
-        .name(entity.getName())
-        .entityVersion(entity.getEntityVersion())
-        .subTypeCode(entity.getSubTypeCode())
-        .createTimestamp(entity.getCreateTimestamp())
-        .dropTimestamp(entity.getDropTimestamp())
-        .purgeTimestamp(entity.getPurgeTimestamp())
-        .toPurgeTimestamp(entity.getToPurgeTimestamp())
-        .lastUpdateTimestamp(entity.getLastUpdateTimestamp())
-        .properties(entity.getProperties())
-        .internalProperties(entity.getInternalProperties())
-        .grantRecordsVersion(entity.getGrantRecordsVersion())
-        .build();
+    var builder =
+        ModelEntity.builder()
+            .catalogId(entity.getCatalogId())
+            .id(entity.getId())
+            .parentId(entity.getParentId())
+            .typeCode(entity.getTypeCode())
+            .name(entity.getName())
+            .entityVersion(entity.getEntityVersion())
+            .subTypeCode(entity.getSubTypeCode())
+            .createTimestamp(entity.getCreateTimestamp())
+            .dropTimestamp(entity.getDropTimestamp())
+            .purgeTimestamp(entity.getPurgeTimestamp())
+            .toPurgeTimestamp(entity.getToPurgeTimestamp())
+            .lastUpdateTimestamp(entity.getLastUpdateTimestamp())
+            .properties(entity.getProperties())
+            .internalProperties(entity.getInternalProperties())
+            .grantRecordsVersion(entity.getGrantRecordsVersion());
+
+    if (entity.getType() == PolarisEntityType.TABLE_LIKE) {
+      if (entity.getSubType() == PolarisEntitySubType.ICEBERG_TABLE
+          || entity.getSubType() == PolarisEntitySubType.ICEBERG_VIEW) {
+        builder.locationWithoutScheme(
+            StorageLocation.of(
+                    entity.getPropertiesAsMap().get(PolarisEntityConstants.ENTITY_BASE_LOCATION))
+                .withoutScheme());
+      }
+    }
+    if (entity.getType() == PolarisEntityType.NAMESPACE) {
+      builder.locationWithoutScheme(
+          StorageLocation.of(
+                  entity.getPropertiesAsMap().get(PolarisEntityConstants.ENTITY_BASE_LOCATION))
+              .withoutScheme());
+    }
+
+    return builder.build();
   }
 
   public static PolarisBaseEntity toEntity(ModelEntity model) {