[PR] Add the Directory entity, REST endpoints and DirectoryScanService SPI [polaris]

["jbonofre (via GitHub)" <gi...@apache.org>]

jbonofre opened a new pull request, #5928:
URL: https://github.com/apache/polaris/pull/5928

   This is the second of three PRs split from #4613:
   
   1. #5927: OpenAPI specification and documentation
   2. **This PR**: Directory entity, REST endpoints and `DirectoryScanService` SPI
   3. PR3 (to be linked): simple directory scan service
   
   It is stacked on #5927: until that one is merged, this PR also shows its commit. Only the last commit (`Add the Directory entity, REST endpoints and DirectoryScanService SPI`) belongs to this PR.
   
   ### Changes
   
   - **Entity and persistence**: new `DIRECTORY` table-like sub-type and `DirectoryEntity` (base location, include/exclude filters, scan schedule), with its NoSQL mapping (`DirectoryObj`). The relational persistence needs no schema change.
   - **REST endpoints**: list, create, load and drop directories (`DirectoryCatalogAdapter`, `DirectoryCatalogHandler`, `PolarisDirectoryCatalog`), with the corresponding events (new `DIRECTORY` event category). Creating a directory also creates its Iceberg inventory table with a predefined schema. The table is named `<directory>__inventory`, because a directory and a table cannot share a name in a namespace (both are table-like entities). Dropping the directory drops the table.
   - **Location validation**: `base_location` is validated like a table location: it must be within the catalog's allowed locations, and within the namespace location unless `polaris.config.allow.unstructured.table.location` is `true` on the catalog. It must not overlap the location of a sibling table, view, directory or namespace. Filters must be valid regular expressions.
   - **`DirectoryScanService` SPI**: `POST .../directories/{directory}/scan` authorizes the request (`UPDATE_TABLE` on the directory), vends read/list credentials scoped to `base_location`, and then calls the `DirectoryScanService` bean. An extension can provide its own bean. In this PR, the default bean (`DefaultDirectoryScanService`, a `@DefaultBean`) rejects the scan. The next PR replaces it with the simple scan service.
   - **Feature flags**: `ENABLE_DIRECTORIES` (default `true`) enables the directory endpoints. `ENABLE_DIRECTORY_SCAN` (default `false`) enables the scan endpoint: when disabled, the endpoint is rejected and not advertised in the catalog config. Because `ENABLE_DIRECTORIES` defaults to `true`, every catalog config now lists the four directory endpoints, hence the regtest reference updates.
   
   ### Tests
   
   - `AbstractPolarisDirectoryCatalogTest` (relational and NoSQL in-memory), `PolarisDirectoryCatalogHandlerAuthzTest`, `DirectoryConfigEndpointsTest`, `PolarisResourcePathsTest`, `TestTableLikeMapping`.
   - `DirectoryServiceIntegrationTest`: end-to-end REST test on a file-based catalog (create/load/list/drop, location validation, read-only principal cannot scan or drop).
   
   ### Notes for reviewers
   
   - Compared to #4613, the filter validation moved from `DirectoryScanner` to `DirectoryCatalogHandler`, so creating a directory does not depend on the scanner.
   - Credential vending for the remote files themselves is not addressed here: reading the inventory does not give access to the objects, as documented in `directories.md`.
   
   ## Checklist
   - [X] 🛡️ Don't disclose security issues! (contact security@apache.org)
   - [X] 🔗 Clearly explained why the changes are needed, or linked related issues: Related to #4613
   - [X] 🧪 Added/updated tests with good coverage, or manually tested (and explained how)
   - [X] 💡 Added comments for complex logic
   - [X] 🧾 Updated `CHANGELOG.md` (if needed)
   - [X] 📚 Updated documentation in `site/content/in-dev/unreleased` (if needed)
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Add the Directory entity, REST endpoints and DirectoryScanService SPI [polaris]

["orca-review-bot (via GitHub)" <gi...@apache.org>]

orca-review-bot commented on code in PR #5928:
URL: https://github.com/apache/polaris/pull/5928#discussion_r4427579473


##########
runtime/service/src/main/java/org/apache/polaris/service/catalog/directory/DirectoryCatalogHandler.java:
##########
@@ -0,0 +1,428 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.polaris.service.catalog.directory;
+
+import static org.apache.iceberg.types.Types.NestedField.optional;
+import static org.apache.iceberg.types.Types.NestedField.required;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.smallrye.common.annotation.Identifier;
+import jakarta.enterprise.inject.Instance;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+import java.util.regex.Pattern;
+import java.util.regex.PatternSyntaxException;
+import java.util.stream.Stream;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.catalog.Catalog;
+import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.exceptions.NoSuchNamespaceException;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.types.Types;
+import org.apache.polaris.core.auth.PolarisAuthorizableOperation;
+import org.apache.polaris.core.catalog.DirectoryCatalog;
+import org.apache.polaris.core.catalog.FederatedCatalogFactory;
+import org.apache.polaris.core.catalog.LocalCatalogFactory;
+import org.apache.polaris.core.catalog.PolarisCatalogHelpers;
+import org.apache.polaris.core.config.FeatureConfiguration;
+import org.apache.polaris.core.connection.ConnectionConfigInfoDpo;
+import org.apache.polaris.core.connection.ConnectionType;
+import org.apache.polaris.core.credentials.PolarisCredentialManager;
+import org.apache.polaris.core.entity.CatalogEntity;
+import org.apache.polaris.core.entity.PolarisEntitySubType;
+import org.apache.polaris.core.entity.PolarisEntityType;
+import org.apache.polaris.core.entity.table.DirectoryEntity;
+import org.apache.polaris.core.persistence.PolarisResolvedPathWrapper;
+import org.apache.polaris.core.persistence.pagination.Page;
+import org.apache.polaris.core.persistence.pagination.PageToken;
+import org.apache.polaris.core.persistence.resolver.PolarisResolutionManifest;
+import org.apache.polaris.core.persistence.resolver.ResolvedPathKey;
+import org.apache.polaris.core.persistence.resolver.ResolverPath;
+import org.apache.polaris.core.storage.PolarisStorageActions;
+import org.apache.polaris.core.storage.StorageAccessConfig;
+import org.apache.polaris.immutables.PolarisImmutable;
+import org.apache.polaris.service.catalog.common.CatalogHandler;
+import org.apache.polaris.service.catalog.common.CatalogUtils;
+import org.apache.polaris.service.catalog.iceberg.LocalIcebergCatalog;
+import org.apache.polaris.service.catalog.io.FileIOFactory;
+import org.apache.polaris.service.catalog.io.StorageAccessConfigProvider;
+import org.apache.polaris.service.catalog.spi.DirectoryScanService;
+import org.apache.polaris.service.catalog.validation.IcebergPropertiesValidation;
+import org.apache.polaris.service.types.Directory;
+import org.apache.polaris.service.types.DirectoryFilter;
+import org.apache.polaris.service.types.ListDirectoriesResponse;
+import org.apache.polaris.service.types.LoadDirectoryResponse;
+import org.apache.polaris.service.types.ScanDirectoryResponse;
+import org.apache.polaris.service.types.ScanSchedule;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@PolarisImmutable
+@SuppressWarnings("immutables:incompat")
+public abstract class DirectoryCatalogHandler extends CatalogHandler {
+  private static final Logger LOGGER = LoggerFactory.getLogger(DirectoryCatalogHandler.class);
+
+  /** Predefined Iceberg table schema for directory file metadata. */
+  public static final Schema DIRECTORY_TABLE_SCHEMA =
+      new Schema(
+          required(1, "file_uri", Types.StringType.get(), "The object URI"),
+          optional(
+              2, "content_type", Types.StringType.get(), "The MIME content type of the object"),
+          optional(3, "size", Types.LongType.get(), "The size of the object in bytes"),
+          optional(
+              4,
+              "checksum_algorithm",
+              Types.StringType.get(),
+              "The checksum algorithm (e.g. MD5, SHA-256)"),
+          optional(5, "checksum", Types.StringType.get(), "The file checksum"),
+          optional(
+              6,
+              "last_modified",
+              Types.TimestampType.withZone(),
+              "The last modification timestamp"),
+          optional(
+              7,
+              "metadata",
+              Types.MapType.ofRequired(8, 9, Types.StringType.get(), Types.StringType.get()),
+              "Additional key-value metadata from the object store"));
+
+  /**
+   * Suffix of the inventory table name. A directory and its inventory table cannot share a name:
+   * both are table-like entities, and names are unique among table-like entities of a namespace.
+   */
+  static final String INVENTORY_TABLE_SUFFIX = "__inventory";
+
+  static TableIdentifier inventoryTableIdentifier(TableIdentifier directory) {
+    return TableIdentifier.of(directory.namespace(), directory.name() + INVENTORY_TABLE_SUFFIX);
+  }
+
+  private static final ObjectMapper MAPPER = new ObjectMapper();
+  private static final TypeReference<List<String>> LIST_OF_STRING = new TypeReference<>() {};
+
+  static String toJsonArray(List<String> list) {
+    if (list == null) {
+      return null;
+    }
+    try {
+      return MAPPER.writeValueAsString(list);
+    } catch (JsonProcessingException e) {
+      throw new IllegalArgumentException("Failed to serialize filter list", e);
+    }
+  }
+
+  private static List<String> parseJsonArray(String json) {
+    if (json == null) {
+      return null;
+    }
+    try {
+      return MAPPER.readValue(json, LIST_OF_STRING);
+    } catch (JsonProcessingException e) {
+      throw new IllegalArgumentException("Failed to parse filter list", e);
+    }
+  }
+
+  private static DirectoryFilter toDirectoryFilter(DirectoryEntity entity) {
+    List<String> include = parseJsonArray(entity.getFilterInclude());
+    List<String> exclude = parseJsonArray(entity.getFilterExclude());
+    if (include == null && exclude == null) {
+      return null;
+    }
+    return DirectoryFilter.builder().setInclude(include).setExclude(exclude).build();
+  }
+
+  /** Rejects filters that are not valid regular expressions. */
+  private static void validateFilters(List<String> include, List<String> exclude) {
+    try {
+      Stream.of(include, exclude)
+          .filter(Objects::nonNull)
+          .flatMap(List::stream)
+          .forEach(Pattern::compile);
+    } catch (PatternSyntaxException e) {
+      throw new IllegalArgumentException("Invalid directory filter: " + e.getMessage(), e);
+    }
+  }
+
+  static String scanScheduleToJson(ScanSchedule scanSchedule) {
+    if (scanSchedule == null) {
+      return null;
+    }
+    try {
+      return MAPPER.writeValueAsString(scanSchedule);
+    } catch (JsonProcessingException e) {
+      throw new IllegalArgumentException("Failed to serialize scan schedule", e);
+    }
+  }
+
+  private static ScanSchedule scanScheduleFromJson(String json) {
+    if (json == null) {
+      return null;
+    }
+    try {
+      return MAPPER.readValue(json, ScanSchedule.class);
+    } catch (JsonProcessingException e) {
+      throw new IllegalArgumentException("Failed to parse scan schedule", e);
+    }
+  }
+
+  protected abstract PolarisCredentialManager credentialManager();
+
+  protected abstract Instance<FederatedCatalogFactory> federatedCatalogFactories();
+
+  protected abstract LocalCatalogFactory localCatalogFactory();
+
+  protected abstract StorageAccessConfigProvider storageAccessConfigProvider();
+
+  protected abstract FileIOFactory fileIOFactory();
+
+  protected abstract DirectoryScanService directoryScanService();
+
+  private DirectoryCatalog directoryCatalog;
+  private Catalog icebergCatalog;
+  private TableIdentifier inventoryTable;
+
+  /**
+   * The underlying Iceberg catalog manipulates the inventory table, which is not the identifier
+   * targeted by the request: it must be added as a passthrough resolution path as well.
+   */
+  @Override
+  protected PolarisResolutionManifest newResolutionManifest() {
+    PolarisResolutionManifest manifest = super.newResolutionManifest();
+    if (inventoryTable != null) {
+      manifest.addPassthroughPath(
+          new ResolverPath(
+              PolarisCatalogHelpers.tableIdentifierToList(inventoryTable),
+              PolarisEntityType.TABLE_LIKE,
+              true /* optional */));
+    }
+    return manifest;
+  }
+
+  @Override
+  protected void initializeCatalog() {
+    CatalogEntity resolvedCatalogEntity = resolutionManifest.getResolvedCatalogEntity();
+    ConnectionConfigInfoDpo connectionConfigInfoDpo =
+        resolvedCatalogEntity.getConnectionConfigInfoDpo();
+    if (connectionConfigInfoDpo != null) {
+      LOGGER
+          .atInfo()
+          .addKeyValue("remoteUrl", connectionConfigInfoDpo.getUri())
+          .log("Initializing federated catalog");
+      FeatureConfiguration.enforceFeatureEnabledOrThrow(
+          realmConfig(), FeatureConfiguration.ENABLE_CATALOG_FEDERATION);
+
+      DirectoryCatalog federatedCatalog;
+      ConnectionType connectionType =
+          ConnectionType.fromCode(connectionConfigInfoDpo.getConnectionTypeCode());
+
+      Instance<FederatedCatalogFactory> federatedCatalogFactory =
+          federatedCatalogFactories()
+              .select(Identifier.Literal.of(connectionType.getFactoryIdentifier()));
+      if (federatedCatalogFactory.isResolvable()) {
+        Map<String, String> catalogProperties = resolvedCatalogEntity.getPropertiesAsMap();
+        federatedCatalog =
+            federatedCatalogFactory
+                .get()
+                .createDirectoryCatalog(
+                    connectionConfigInfoDpo, credentialManager(), catalogProperties);
+      } else {
+        throw new UnsupportedOperationException(
+            "External catalog factory for type '" + connectionType + "' is unavailable.");
+      }
+      this.directoryCatalog = federatedCatalog;
+    } else {
+      LOGGER.atInfo().log("Initializing non-federated catalog");
+      this.directoryCatalog =
+          new PolarisDirectoryCatalog(metaStoreManager(), callContext(), this.resolutionManifest);
+      this.directoryCatalog.initialize(catalogName(), Map.of());
+    }
+
+    // Initialize an IcebergCatalog for creating/dropping the associated Iceberg tables
+    this.icebergCatalog = localCatalogFactory().createCatalog(this.resolutionManifest);
+  }
+
+  public ListDirectoriesResponse listDirectories(
+      Namespace parent, String pageToken, Integer pageSize) {
+    PolarisAuthorizableOperation op = PolarisAuthorizableOperation.LIST_TABLES;
+    authorizeBasicNamespaceOperationOrThrow(op, parent);
+
+    PageToken pageRequest =
+        PageToken.build(pageToken, pageSize, maxPageSize(), this::shouldDecodeToken);
+    Page<TableIdentifier> page = directoryCatalog.listDirectories(parent, pageRequest);
+    rejectIncompleteListing(pageToken, pageSize, page.encodedResponseToken());
+    return ListDirectoriesResponse.builder()
+        .setIdentifiers(new LinkedHashSet<>(page.items()))
+        .setNextPageToken(page.encodedResponseToken())
+        .build();
+  }
+
+  public LoadDirectoryResponse createDirectory(
+      TableIdentifier identifier,
+      String baseLocation,
+      String filterInclude,
+      String filterExclude,
+      String scanSchedule,
+      Map<String, String> properties) {
+    this.inventoryTable = inventoryTableIdentifier(identifier);
+    PolarisAuthorizableOperation op = PolarisAuthorizableOperation.CREATE_TABLE_DIRECT;
+    authorizeCreateTableLikeUnderNamespaceOperationOrThrow(op, identifier);
+
+    PolarisResolvedPathWrapper resolvedParent =
+        resolutionManifest.getResolvedPath(ResolvedPathKey.ofNamespace(identifier.namespace()));
+    if (resolvedParent == null) {
+      throw new NoSuchNamespaceException("Namespace does not exist: %s", identifier.namespace());
+    }
+    // The base location is read with vended credentials: it must be within the allowed locations
+    CatalogUtils.validateLocationsForTableLike(
+        realmConfig(), identifier, Set.of(baseLocation), resolvedParent);
+    // The scan exposes the names and sizes of the objects under the base location: it must not
+    // overlap the location of a sibling table, view, directory or namespace
+    ((LocalIcebergCatalog) icebergCatalog)

Review Comment:
   This comment promises no overlap with a sibling directory, but existing directories never get checked. The sibling loop in `LocalIcebergCatalog` resolves them and then passes them through `PolarisEntityUtils.asLocationBasedEntity`, which has no `DIRECTORY` case and returns empty. So directory B can sit under directory A, and its scan would list A's objects. The same gap lets a new Iceberg table land inside an existing directory. Can we add `case DIRECTORY -> DirectoryEntity.of(entity)` there and add a test for both cases?
   
   <sub>🤖 AI-generated review comment — agents make mistakes, so please verify before acting on it.</sub>



##########
runtime/service/src/main/java/org/apache/polaris/service/catalog/directory/DirectoryCatalogAdapter.java:
##########
@@ -0,0 +1,168 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.polaris.service.catalog.directory;
+
+import jakarta.enterprise.context.RequestScoped;
+import jakarta.inject.Inject;
+import jakarta.ws.rs.core.Response;
+import jakarta.ws.rs.core.SecurityContext;
+import java.util.List;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.polaris.core.auth.PolarisPrincipal;
+import org.apache.polaris.core.config.FeatureConfiguration;
+import org.apache.polaris.core.config.RealmConfig;
+import org.apache.polaris.core.context.CallContext;
+import org.apache.polaris.core.context.RealmContext;
+import org.apache.polaris.core.rest.NamespaceUtils;
+import org.apache.polaris.service.catalog.CatalogPrefixParser;
+import org.apache.polaris.service.catalog.api.PolarisCatalogDirectoryApiService;
+import org.apache.polaris.service.catalog.common.CatalogAdapter;
+import org.apache.polaris.service.config.ReservedProperties;
+import org.apache.polaris.service.types.CreateDirectoryRequest;
+import org.apache.polaris.service.types.DirectoryFilter;
+import org.apache.polaris.service.types.ListDirectoriesResponse;
+import org.apache.polaris.service.types.LoadDirectoryResponse;
+import org.apache.polaris.service.types.ScanDirectoryResponse;
+
+@RequestScoped
+public class DirectoryCatalogAdapter implements PolarisCatalogDirectoryApiService, CatalogAdapter {
+
+  private final RealmConfig realmConfig;
+  private final ReservedProperties reservedProperties;
+  private final CatalogPrefixParser prefixParser;
+  private final DirectoryCatalogHandlerFactory handlerFactory;
+
+  @Inject
+  public DirectoryCatalogAdapter(
+      CallContext callContext,
+      CatalogPrefixParser prefixParser,
+      ReservedProperties reservedProperties,
+      DirectoryCatalogHandlerFactory handlerFactory) {
+    this.realmConfig = callContext.getRealmConfig();
+    this.prefixParser = prefixParser;
+    this.reservedProperties = reservedProperties;
+    this.handlerFactory = handlerFactory;
+  }
+
+  private DirectoryCatalogHandler newHandler(SecurityContext securityContext, String prefix) {
+    FeatureConfiguration.enforceFeatureEnabledOrThrow(
+        realmConfig, FeatureConfiguration.ENABLE_DIRECTORIES);
+    PolarisPrincipal principal = validatePrincipal(securityContext);
+    String catalogName = prefixParser.prefixToCatalogName(prefix);
+    return handlerFactory.createHandler(catalogName, principal);
+  }
+
+  @Override
+  public Response createDirectory(
+      String prefix,
+      String namespace,
+      CreateDirectoryRequest createDirectoryRequest,
+      List<String> polarisDirectoryAccessDelegation,

Review Comment:
   `polarisDirectoryAccessDelegation` is accepted but never read, so clients can send the header and nothing happens. Can we drop it from the spec until it's wired through, or pass it to the handler?
   
   <sub>🤖 AI-generated review comment — agents make mistakes, so please verify before acting on it.</sub>



##########
runtime/service/src/main/java/org/apache/polaris/service/catalog/spi/DirectoryScanService.java:
##########
@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.polaris.service.catalog.spi;
+
+import java.util.List;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.io.FileIO;
+
+/**
+ * Scans the objects of a directory and populates its inventory table.
+ *
+ * <p>Polaris provides a default implementation. To customize the scan (for example to compute
+ * checksums, read object-store metadata or update the table incrementally), provide an {@code
+ * ApplicationScoped} CDI bean implementing this interface: it replaces the default one.
+ */
+public interface DirectoryScanService {
+
+  /**
+   * @param identifier the directory identifier, also the identifier of its inventory table

Review Comment:
   The inventory table is `<name>__inventory`, so it isn't the same identifier as the directory. Can we fix this so SPI implementers don't write to the wrong table?
   
   <sub>🤖 AI-generated review comment — agents make mistakes, so please verify before acting on it.</sub>



##########
polaris-core/src/main/java/org/apache/polaris/core/config/FeatureConfiguration.java:
##########
@@ -527,6 +527,22 @@ public static void enforceFeatureEnabledOrThrow(
           .defaultValue(true)
           .buildFeatureConfiguration();
 
+  public static final FeatureConfiguration<Boolean> ENABLE_DIRECTORIES =

Review Comment:
   Defaulting a beta feature to on means every catalog's config response changes on upgrade (hence all the regtest ref churn). Was default-true agreed for 1.9.0, or should this be false like `ENABLE_DIRECTORY_SCAN`?
   
   <sub>🤖 AI-generated review comment — agents make mistakes, so please verify before acting on it.</sub>



##########
runtime/service/src/main/java/org/apache/polaris/service/catalog/directory/DirectoryCatalogHandler.java:
##########
@@ -0,0 +1,428 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.polaris.service.catalog.directory;
+
+import static org.apache.iceberg.types.Types.NestedField.optional;
+import static org.apache.iceberg.types.Types.NestedField.required;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.smallrye.common.annotation.Identifier;
+import jakarta.enterprise.inject.Instance;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+import java.util.regex.Pattern;
+import java.util.regex.PatternSyntaxException;
+import java.util.stream.Stream;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.catalog.Catalog;
+import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.exceptions.NoSuchNamespaceException;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.types.Types;
+import org.apache.polaris.core.auth.PolarisAuthorizableOperation;
+import org.apache.polaris.core.catalog.DirectoryCatalog;
+import org.apache.polaris.core.catalog.FederatedCatalogFactory;
+import org.apache.polaris.core.catalog.LocalCatalogFactory;
+import org.apache.polaris.core.catalog.PolarisCatalogHelpers;
+import org.apache.polaris.core.config.FeatureConfiguration;
+import org.apache.polaris.core.connection.ConnectionConfigInfoDpo;
+import org.apache.polaris.core.connection.ConnectionType;
+import org.apache.polaris.core.credentials.PolarisCredentialManager;
+import org.apache.polaris.core.entity.CatalogEntity;
+import org.apache.polaris.core.entity.PolarisEntitySubType;
+import org.apache.polaris.core.entity.PolarisEntityType;
+import org.apache.polaris.core.entity.table.DirectoryEntity;
+import org.apache.polaris.core.persistence.PolarisResolvedPathWrapper;
+import org.apache.polaris.core.persistence.pagination.Page;
+import org.apache.polaris.core.persistence.pagination.PageToken;
+import org.apache.polaris.core.persistence.resolver.PolarisResolutionManifest;
+import org.apache.polaris.core.persistence.resolver.ResolvedPathKey;
+import org.apache.polaris.core.persistence.resolver.ResolverPath;
+import org.apache.polaris.core.storage.PolarisStorageActions;
+import org.apache.polaris.core.storage.StorageAccessConfig;
+import org.apache.polaris.immutables.PolarisImmutable;
+import org.apache.polaris.service.catalog.common.CatalogHandler;
+import org.apache.polaris.service.catalog.common.CatalogUtils;
+import org.apache.polaris.service.catalog.iceberg.LocalIcebergCatalog;
+import org.apache.polaris.service.catalog.io.FileIOFactory;
+import org.apache.polaris.service.catalog.io.StorageAccessConfigProvider;
+import org.apache.polaris.service.catalog.spi.DirectoryScanService;
+import org.apache.polaris.service.catalog.validation.IcebergPropertiesValidation;
+import org.apache.polaris.service.types.Directory;
+import org.apache.polaris.service.types.DirectoryFilter;
+import org.apache.polaris.service.types.ListDirectoriesResponse;
+import org.apache.polaris.service.types.LoadDirectoryResponse;
+import org.apache.polaris.service.types.ScanDirectoryResponse;
+import org.apache.polaris.service.types.ScanSchedule;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@PolarisImmutable
+@SuppressWarnings("immutables:incompat")
+public abstract class DirectoryCatalogHandler extends CatalogHandler {
+  private static final Logger LOGGER = LoggerFactory.getLogger(DirectoryCatalogHandler.class);
+
+  /** Predefined Iceberg table schema for directory file metadata. */
+  public static final Schema DIRECTORY_TABLE_SCHEMA =
+      new Schema(
+          required(1, "file_uri", Types.StringType.get(), "The object URI"),
+          optional(
+              2, "content_type", Types.StringType.get(), "The MIME content type of the object"),
+          optional(3, "size", Types.LongType.get(), "The size of the object in bytes"),
+          optional(
+              4,
+              "checksum_algorithm",
+              Types.StringType.get(),
+              "The checksum algorithm (e.g. MD5, SHA-256)"),
+          optional(5, "checksum", Types.StringType.get(), "The file checksum"),
+          optional(
+              6,
+              "last_modified",
+              Types.TimestampType.withZone(),
+              "The last modification timestamp"),
+          optional(
+              7,
+              "metadata",
+              Types.MapType.ofRequired(8, 9, Types.StringType.get(), Types.StringType.get()),
+              "Additional key-value metadata from the object store"));
+
+  /**
+   * Suffix of the inventory table name. A directory and its inventory table cannot share a name:
+   * both are table-like entities, and names are unique among table-like entities of a namespace.
+   */
+  static final String INVENTORY_TABLE_SUFFIX = "__inventory";
+
+  static TableIdentifier inventoryTableIdentifier(TableIdentifier directory) {
+    return TableIdentifier.of(directory.namespace(), directory.name() + INVENTORY_TABLE_SUFFIX);
+  }
+
+  private static final ObjectMapper MAPPER = new ObjectMapper();
+  private static final TypeReference<List<String>> LIST_OF_STRING = new TypeReference<>() {};
+
+  static String toJsonArray(List<String> list) {
+    if (list == null) {
+      return null;
+    }
+    try {
+      return MAPPER.writeValueAsString(list);
+    } catch (JsonProcessingException e) {
+      throw new IllegalArgumentException("Failed to serialize filter list", e);
+    }
+  }
+
+  private static List<String> parseJsonArray(String json) {
+    if (json == null) {
+      return null;
+    }
+    try {
+      return MAPPER.readValue(json, LIST_OF_STRING);
+    } catch (JsonProcessingException e) {
+      throw new IllegalArgumentException("Failed to parse filter list", e);
+    }
+  }
+
+  private static DirectoryFilter toDirectoryFilter(DirectoryEntity entity) {
+    List<String> include = parseJsonArray(entity.getFilterInclude());
+    List<String> exclude = parseJsonArray(entity.getFilterExclude());
+    if (include == null && exclude == null) {
+      return null;
+    }
+    return DirectoryFilter.builder().setInclude(include).setExclude(exclude).build();
+  }
+
+  /** Rejects filters that are not valid regular expressions. */
+  private static void validateFilters(List<String> include, List<String> exclude) {
+    try {
+      Stream.of(include, exclude)
+          .filter(Objects::nonNull)
+          .flatMap(List::stream)
+          .forEach(Pattern::compile);
+    } catch (PatternSyntaxException e) {
+      throw new IllegalArgumentException("Invalid directory filter: " + e.getMessage(), e);
+    }
+  }
+
+  static String scanScheduleToJson(ScanSchedule scanSchedule) {
+    if (scanSchedule == null) {
+      return null;
+    }
+    try {
+      return MAPPER.writeValueAsString(scanSchedule);
+    } catch (JsonProcessingException e) {
+      throw new IllegalArgumentException("Failed to serialize scan schedule", e);
+    }
+  }
+
+  private static ScanSchedule scanScheduleFromJson(String json) {
+    if (json == null) {
+      return null;
+    }
+    try {
+      return MAPPER.readValue(json, ScanSchedule.class);
+    } catch (JsonProcessingException e) {
+      throw new IllegalArgumentException("Failed to parse scan schedule", e);
+    }
+  }
+
+  protected abstract PolarisCredentialManager credentialManager();
+
+  protected abstract Instance<FederatedCatalogFactory> federatedCatalogFactories();
+
+  protected abstract LocalCatalogFactory localCatalogFactory();
+
+  protected abstract StorageAccessConfigProvider storageAccessConfigProvider();
+
+  protected abstract FileIOFactory fileIOFactory();
+
+  protected abstract DirectoryScanService directoryScanService();
+
+  private DirectoryCatalog directoryCatalog;
+  private Catalog icebergCatalog;
+  private TableIdentifier inventoryTable;
+
+  /**
+   * The underlying Iceberg catalog manipulates the inventory table, which is not the identifier
+   * targeted by the request: it must be added as a passthrough resolution path as well.
+   */
+  @Override
+  protected PolarisResolutionManifest newResolutionManifest() {
+    PolarisResolutionManifest manifest = super.newResolutionManifest();
+    if (inventoryTable != null) {
+      manifest.addPassthroughPath(
+          new ResolverPath(
+              PolarisCatalogHelpers.tableIdentifierToList(inventoryTable),
+              PolarisEntityType.TABLE_LIKE,
+              true /* optional */));
+    }
+    return manifest;
+  }
+
+  @Override
+  protected void initializeCatalog() {
+    CatalogEntity resolvedCatalogEntity = resolutionManifest.getResolvedCatalogEntity();
+    ConnectionConfigInfoDpo connectionConfigInfoDpo =
+        resolvedCatalogEntity.getConnectionConfigInfoDpo();
+    if (connectionConfigInfoDpo != null) {
+      LOGGER
+          .atInfo()
+          .addKeyValue("remoteUrl", connectionConfigInfoDpo.getUri())
+          .log("Initializing federated catalog");
+      FeatureConfiguration.enforceFeatureEnabledOrThrow(
+          realmConfig(), FeatureConfiguration.ENABLE_CATALOG_FEDERATION);
+
+      DirectoryCatalog federatedCatalog;
+      ConnectionType connectionType =
+          ConnectionType.fromCode(connectionConfigInfoDpo.getConnectionTypeCode());
+
+      Instance<FederatedCatalogFactory> federatedCatalogFactory =
+          federatedCatalogFactories()
+              .select(Identifier.Literal.of(connectionType.getFactoryIdentifier()));
+      if (federatedCatalogFactory.isResolvable()) {
+        Map<String, String> catalogProperties = resolvedCatalogEntity.getPropertiesAsMap();
+        federatedCatalog =
+            federatedCatalogFactory
+                .get()
+                .createDirectoryCatalog(
+                    connectionConfigInfoDpo, credentialManager(), catalogProperties);
+      } else {
+        throw new UnsupportedOperationException(
+            "External catalog factory for type '" + connectionType + "' is unavailable.");
+      }
+      this.directoryCatalog = federatedCatalog;
+    } else {
+      LOGGER.atInfo().log("Initializing non-federated catalog");
+      this.directoryCatalog =
+          new PolarisDirectoryCatalog(metaStoreManager(), callContext(), this.resolutionManifest);
+      this.directoryCatalog.initialize(catalogName(), Map.of());
+    }
+
+    // Initialize an IcebergCatalog for creating/dropping the associated Iceberg tables
+    this.icebergCatalog = localCatalogFactory().createCatalog(this.resolutionManifest);
+  }
+
+  public ListDirectoriesResponse listDirectories(
+      Namespace parent, String pageToken, Integer pageSize) {
+    PolarisAuthorizableOperation op = PolarisAuthorizableOperation.LIST_TABLES;
+    authorizeBasicNamespaceOperationOrThrow(op, parent);
+
+    PageToken pageRequest =
+        PageToken.build(pageToken, pageSize, maxPageSize(), this::shouldDecodeToken);
+    Page<TableIdentifier> page = directoryCatalog.listDirectories(parent, pageRequest);
+    rejectIncompleteListing(pageToken, pageSize, page.encodedResponseToken());
+    return ListDirectoriesResponse.builder()
+        .setIdentifiers(new LinkedHashSet<>(page.items()))
+        .setNextPageToken(page.encodedResponseToken())
+        .build();
+  }
+
+  public LoadDirectoryResponse createDirectory(
+      TableIdentifier identifier,
+      String baseLocation,
+      String filterInclude,
+      String filterExclude,
+      String scanSchedule,
+      Map<String, String> properties) {
+    this.inventoryTable = inventoryTableIdentifier(identifier);
+    PolarisAuthorizableOperation op = PolarisAuthorizableOperation.CREATE_TABLE_DIRECT;
+    authorizeCreateTableLikeUnderNamespaceOperationOrThrow(op, identifier);
+
+    PolarisResolvedPathWrapper resolvedParent =
+        resolutionManifest.getResolvedPath(ResolvedPathKey.ofNamespace(identifier.namespace()));
+    if (resolvedParent == null) {
+      throw new NoSuchNamespaceException("Namespace does not exist: %s", identifier.namespace());
+    }
+    // The base location is read with vended credentials: it must be within the allowed locations
+    CatalogUtils.validateLocationsForTableLike(
+        realmConfig(), identifier, Set.of(baseLocation), resolvedParent);
+    // The scan exposes the names and sizes of the objects under the base location: it must not
+    // overlap the location of a sibling table, view, directory or namespace
+    ((LocalIcebergCatalog) icebergCatalog)
+        .validateNoDirectoryLocationOverlap(
+            identifier, baseLocation, resolvedParent.getRawFullPath());
+
+    validateFilters(parseJsonArray(filterInclude), parseJsonArray(filterExclude));
+    TableIdentifier tableId = inventoryTableIdentifier(identifier);
+
+    DirectoryEntity createdEntity =
+        this.directoryCatalog.createDirectory(
+            identifier, baseLocation, filterInclude, filterExclude, scanSchedule, properties);
+
+    // Create the associated Iceberg table with the predefined schema
+    try {
+      this.icebergCatalog.createTable(tableId, DIRECTORY_TABLE_SCHEMA);
+      LOGGER.debug("Created associated Iceberg table {} for directory {}", tableId, identifier);
+    } catch (RuntimeException e) {

Review Comment:
   This catches every RuntimeException, `CommitStateUnknownException` included. If the inventory commit actually succeeded, we drop the directory but leave `<name>__inventory` behind. Every later create of that name then fails on the leftover table. Should we skip the rollback (or check `tableExists`) when the commit state is unknown?
   
   <sub>🤖 AI-generated review comment — agents make mistakes, so please verify before acting on it.</sub>



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Add the Directory entity, REST endpoints and DirectoryScanService SPI [polaris]

["ArnabKarmakar1108 (via GitHub) via issues" <is...@polaris.apache.org>]

ArnabKarmakar1108 commented on code in PR #5928:
URL: https://github.com/apache/polaris/pull/5928#discussion_r4431392292


##########
runtime/service/src/test/java/org/apache/polaris/service/catalog/directory/AbstractPolarisDirectoryCatalogTest.java:
##########
@@ -0,0 +1,503 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.polaris.service.catalog.directory;
+
+import static org.apache.iceberg.types.Types.NestedField.required;
+import static org.mockito.ArgumentMatchers.isA;
+import static org.mockito.Mockito.when;
+
+import com.google.common.collect.ImmutableMap;
+import io.quarkus.test.junit.QuarkusMock;
+import jakarta.inject.Inject;
+import java.io.IOException;
+import java.lang.reflect.Method;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import org.apache.iceberg.CatalogProperties;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.types.Types;
+import org.apache.polaris.core.PolarisCallContext;
+import org.apache.polaris.core.PolarisDiagnostics;
+import org.apache.polaris.core.admin.model.AwsStorageConfigInfo;
+import org.apache.polaris.core.admin.model.CreateCatalogRequest;
+import org.apache.polaris.core.admin.model.StorageConfigInfo;
+import org.apache.polaris.core.auth.PolarisAuthorizer;
+import org.apache.polaris.core.auth.PolarisAuthorizerImpl;
+import org.apache.polaris.core.auth.PolarisPrincipal;
+import org.apache.polaris.core.auth.PolarisPrincipalAttributes;
+import org.apache.polaris.core.collection.ImmutableAttributeMap;
+import org.apache.polaris.core.config.FeatureConfiguration;
+import org.apache.polaris.core.config.RealmConfig;
+import org.apache.polaris.core.context.CallContext;
+import org.apache.polaris.core.context.RealmContext;
+import org.apache.polaris.core.entity.CatalogEntity;
+import org.apache.polaris.core.entity.PolarisEntity;
+import org.apache.polaris.core.entity.PrincipalEntity;
+import org.apache.polaris.core.entity.table.DirectoryEntity;
+import org.apache.polaris.core.identity.provider.ServiceIdentityProvider;
+import org.apache.polaris.core.persistence.PolarisMetaStoreManager;
+import org.apache.polaris.core.persistence.pagination.PageToken;
+import org.apache.polaris.core.persistence.resolver.ResolutionManifestFactory;
+import org.apache.polaris.core.persistence.resolver.ResolverFactory;
+import org.apache.polaris.core.secrets.UserSecretsManager;
+import org.apache.polaris.core.storage.PolarisStorageIntegrationProvider;
+import org.apache.polaris.core.storage.aws.AwsCredentialsStorageIntegration;
+import org.apache.polaris.core.storage.aws.AwsStorageConfigurationInfo;
+import org.apache.polaris.core.storage.cache.StorageCredentialCache;
+import org.apache.polaris.service.admin.PolarisAdminService;
+import org.apache.polaris.service.admin.PolarisAdminServiceTestSupport;
+import org.apache.polaris.service.catalog.PolarisPassthroughResolutionView;
+import org.apache.polaris.service.catalog.iceberg.LocalIcebergCatalog;
+import org.apache.polaris.service.catalog.io.FileIOFactory;
+import org.apache.polaris.service.catalog.io.StorageAccessConfigProvider;
+import org.apache.polaris.service.config.ReservedProperties;
+import org.apache.polaris.service.context.catalog.PolarisPrincipalHolder;
+import org.apache.polaris.service.events.PolarisEventMetadataFactory;
+import org.apache.polaris.service.events.listeners.InMemoryEventCollector;
+import org.apache.polaris.service.storage.PolarisStorageIntegrationProviderImpl;
+import org.apache.polaris.service.task.TaskExecutor;
+import org.assertj.core.api.Assertions;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInfo;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.NullSource;
+import org.junit.jupiter.params.provider.ValueSource;
+import org.mockito.Mockito;
+import software.amazon.awssdk.services.sts.StsClient;
+import software.amazon.awssdk.services.sts.model.AssumeRoleRequest;
+import software.amazon.awssdk.services.sts.model.AssumeRoleResponse;
+import software.amazon.awssdk.services.sts.model.Credentials;
+
+public abstract class AbstractPolarisDirectoryCatalogTest {
+
+  protected static final Namespace NS = Namespace.of("newdb");
+  protected static final TableIdentifier TABLE = TableIdentifier.of(NS, "table");
+  public static final String CATALOG_NAME = "polaris-catalog";
+  public static final String TEST_ACCESS_KEY = "test_access_key";
+  public static final String SECRET_ACCESS_KEY = "secret_access_key";
+  public static final String SESSION_TOKEN = "session_token";
+
+  @Inject ServiceIdentityProvider serviceIdentityProvider;
+  @Inject StorageCredentialCache storageCredentialCache;
+  @Inject PolarisStorageIntegrationProvider storageIntegrationProvider;
+  @Inject PolarisDiagnostics diagServices;
+  @Inject ResolverFactory resolverFactory;
+  @Inject ResolutionManifestFactory resolutionManifestFactory;
+  @Inject PolarisEventMetadataFactory eventMetadataFactory;
+  @Inject PolarisMetaStoreManager metaStoreManager;
+  @Inject UserSecretsManager userSecretsManager;
+  @Inject CallContext callContext;
+  @Inject RealmConfig realmConfig;
+  @Inject StorageAccessConfigProvider storageAccessConfigProvider;
+  @Inject FileIOFactory fileIOFactory;
+  @Inject PolarisPrincipalHolder polarisPrincipalHolder;
+
+  private PolarisDirectoryCatalog directoryCatalog;
+  private LocalIcebergCatalog icebergCatalog;
+  private AwsStorageConfigInfo storageConfigModel;
+  private String realmName;
+  private PolarisCallContext polarisContext;
+  private PolarisAdminService adminService;
+  private PolarisPrincipal authenticatedRoot;
+  private PolarisEntity catalogEntity;
+
+  protected static final Schema SCHEMA =
+      new Schema(
+          required(3, "id", Types.IntegerType.get(), "unique ID"),
+          required(4, "data", Types.StringType.get()));
+
+  @BeforeAll
+  public static void setUpMocks() {
+    PolarisStorageIntegrationProviderImpl mock =
+        Mockito.mock(PolarisStorageIntegrationProviderImpl.class);
+    QuarkusMock.installMockForType(mock, PolarisStorageIntegrationProviderImpl.class);
+  }
+
+  protected void bootstrapRealm(String realmName) {}
+
+  @BeforeEach
+  @SuppressWarnings("unchecked")
+  public void before(TestInfo testInfo) {
+    storageCredentialCache.invalidateAll();
+
+    realmName =
+        "realm_%s_%s"
+            .formatted(
+                testInfo.getTestMethod().map(Method::getName).orElse("test"), System.nanoTime());
+    bootstrapRealm(realmName);
+
+    RealmContext realmContext = () -> realmName;
+    QuarkusMock.installMockForType(realmContext, RealmContext.class);
+    polarisContext = callContext.getPolarisCallContext();
+
+    PrincipalEntity rootPrincipal =
+        metaStoreManager.findRootPrincipal(polarisContext).orElseThrow();
+    authenticatedRoot =
+        PolarisPrincipal.of(
+            rootPrincipal.getName(),
+            ImmutableAttributeMap.builder()
+                .put(PolarisPrincipalAttributes.PRINCIPAL_ENTITY_ATTRIBUTE_KEY, rootPrincipal)
+                .put(PolarisPrincipalAttributes.PRINCIPAL_ROLE_ALL_ATTRIBUTE_KEY, true)
+                .build(),
+            Set.of());
+    polarisPrincipalHolder.set(authenticatedRoot);
+
+    PolarisAuthorizer authorizer = new PolarisAuthorizerImpl();
+    ReservedProperties reservedProperties = ReservedProperties.NONE;
+
+    adminService =
+        PolarisAdminServiceTestSupport.newAdminService(
+            polarisContext,
+            resolutionManifestFactory,
+            metaStoreManager,
+            userSecretsManager,
+            serviceIdentityProvider,
+            authenticatedRoot,
+            authorizer,
+            reservedProperties);
+
+    String storageLocation = "s3://my-bucket/path/to/data";
+    storageConfigModel =
+        AwsStorageConfigInfo.builder()
+            .setRoleArn("arn:aws:iam::012345678901:role/jdoe")
+            .setExternalId("externalId")
+            .setUserArn("aws::a:user:arn")
+            .setStorageType(StorageConfigInfo.StorageTypeEnum.S3)
+            .setAllowedLocations(List.of(storageLocation, "s3://externally-owned-bucket"))
+            .build();
+    catalogEntity =
+        adminService.createCatalog(
+            new CreateCatalogRequest(
+                new CatalogEntity.Builder()
+                    .setName(CATALOG_NAME)
+                    .setDefaultBaseLocation(storageLocation)
+                    .addProperty(
+                        FeatureConfiguration.ALLOW_EXTERNAL_METADATA_FILE_LOCATION.catalogConfig(),
+                        "true")
+                    .addProperty(
+                        FeatureConfiguration.ALLOW_UNSTRUCTURED_TABLE_LOCATION.catalogConfig(),
+                        "true")
+                    .addProperty(
+                        FeatureConfiguration.DROP_WITH_PURGE_ENABLED.catalogConfig(), "true")
+                    .setStorageConfigurationInfo(realmConfig, storageConfigModel)
+                    .build()
+                    .asCatalog(serviceIdentityProvider)));
+
+    PolarisPassthroughResolutionView passthroughView =
+        new PolarisPassthroughResolutionView(
+            resolutionManifestFactory, authenticatedRoot, CATALOG_NAME);
+    TaskExecutor taskExecutor = Mockito.mock();
+
+    StsClient stsClient = Mockito.mock(StsClient.class);
+    when(stsClient.assumeRole(isA(AssumeRoleRequest.class)))
+        .thenReturn(
+            AssumeRoleResponse.builder()
+                .credentials(
+                    Credentials.builder()
+                        .accessKeyId(TEST_ACCESS_KEY)
+                        .secretAccessKey(SECRET_ACCESS_KEY)
+                        .sessionToken(SESSION_TOKEN)
+                        .build())
+                .build());
+    AwsStorageConfigurationInfo mockAwsConfig =
+        AwsStorageConfigurationInfo.builder()
+            .roleARN("arn:aws:iam::012345678901:role/mock")
+            .build();
+    AwsCredentialsStorageIntegration storageIntegration =
+        new AwsCredentialsStorageIntegration(
+            (destination) -> stsClient,
+            config -> java.util.Optional.empty(),
+            storageCredentialCache,
+            mockAwsConfig,
+            callContext.getRealmConfig());
+    when(storageIntegrationProvider.getStorageIntegration(Mockito.anyList()))
+        .thenReturn(storageIntegration);
+
+    this.directoryCatalog =
+        new PolarisDirectoryCatalog(metaStoreManager, polarisContext, passthroughView);
+    this.directoryCatalog.initialize(CATALOG_NAME, Map.of());
+    this.icebergCatalog =
+        new LocalIcebergCatalog(
+            diagServices,
+            resolverFactory,
+            metaStoreManager,
+            polarisContext,
+            passthroughView,
+            authenticatedRoot,
+            taskExecutor,
+            storageAccessConfigProvider,
+            fileIOFactory,
+            new InMemoryEventCollector(),
+            eventMetadataFactory);
+    this.icebergCatalog.initialize(
+        CATALOG_NAME,
+        ImmutableMap.of(
+            CatalogProperties.FILE_IO_IMPL, "org.apache.iceberg.inmemory.InMemoryFileIO"));
+  }
+
+  @AfterEach
+  public void after() throws IOException {
+    metaStoreManager.purge(polarisContext);
+  }
+
+  @Test
+  public void testCreateDirectoryDoesNotThrow() {
+    Namespace namespace = Namespace.of("ns");
+    icebergCatalog.createNamespace(namespace);
+    Assertions.assertThatCode(
+            () ->
+                directoryCatalog.createDirectory(
+                    TableIdentifier.of("ns", "d1"), "s3://bucket/path", null, null, null, Map.of()))
+        .doesNotThrowAnyException();
+  }
+
+  @Test
+  public void testCreateAndLoadDirectory() {
+    Namespace namespace = Namespace.of("ns");
+    icebergCatalog.createNamespace(namespace);
+
+    directoryCatalog.createDirectory(
+        TableIdentifier.of("ns", "d1"), "s3://bucket/path", null, null, null, Map.of());
+
+    DirectoryEntity entity = directoryCatalog.loadDirectory(TableIdentifier.of("ns", "d1"));
+    Assertions.assertThat(entity.getName()).isEqualTo("d1");
+    Assertions.assertThat(entity.getBaseLocation()).isEqualTo("s3://bucket/path");
+  }
+
+  @Test
+  public void testDirectoryAlreadyExists() {
+    Namespace namespace = Namespace.of("ns");
+    icebergCatalog.createNamespace(namespace);
+    directoryCatalog.createDirectory(
+        TableIdentifier.of("ns", "d1"), "s3://bucket/path", null, null, null, Map.of());
+
+    Assertions.assertThatCode(
+            () ->
+                directoryCatalog.createDirectory(
+                    TableIdentifier.of("ns", "d1"),
+                    "s3://bucket/other",
+                    null,
+                    null,
+                    null,
+                    Map.of()))
+        .hasMessageContaining("already exists");
+
+    Assertions.assertThatCode(
+            () -> icebergCatalog.createTable(TableIdentifier.of("ns", "d1"), SCHEMA))
+        .hasMessageContaining("already exists");
+  }
+
+  @Test
+  public void testInventoryTableCoexistsWithDirectory() {

Review Comment:
   We are only covering inventory created after the directory, not create-directory when __inventory already exists at the handler layer. I think adding a case for that would improve the coverage.



##########
runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/LocalIcebergCatalog.java:
##########


Review Comment:
   I think the message should also mention directories. Same for line 1709.



##########
runtime/service/src/test/java/org/apache/polaris/service/catalog/directory/PolarisDirectoryCatalogHandlerAuthzTest.java:
##########
@@ -0,0 +1,174 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.polaris.service.catalog.directory;
+
+import io.quarkus.test.junit.QuarkusTest;
+import io.quarkus.test.junit.TestProfile;
+import jakarta.enterprise.inject.Any;
+import jakarta.enterprise.inject.Instance;
+import jakarta.inject.Inject;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Stream;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.polaris.core.auth.PolarisPrincipal;
+import org.apache.polaris.core.auth.PolarisPrincipalAttributes;
+import org.apache.polaris.core.catalog.FederatedCatalogFactory;
+import org.apache.polaris.core.catalog.LocalCatalogFactory;
+import org.apache.polaris.core.collection.ImmutableAttributeMap;
+import org.apache.polaris.core.credentials.PolarisCredentialManager;
+import org.apache.polaris.core.entity.PolarisPrivilege;
+import org.apache.polaris.service.Profiles;
+import org.apache.polaris.service.admin.PolarisAuthzTestBase;
+import org.apache.polaris.service.catalog.spi.DirectoryScanService;
+import org.junit.jupiter.api.DynamicNode;
+import org.junit.jupiter.api.TestFactory;
+
+@QuarkusTest
+@TestProfile(Profiles.PolarisAuthzBaseProfile.class)
+public class PolarisDirectoryCatalogHandlerAuthzTest extends PolarisAuthzTestBase {

Review Comment:
   We are covering list/create/load/drop but no scanDirectory matrix.



##########
integration-tests/src/main/java/org/apache/polaris/service/it/test/PolarisDirectoryServiceIntegrationTest.java:
##########
@@ -0,0 +1,230 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.polaris.service.it.test;
+
+import static org.apache.polaris.service.it.env.PolarisClient.polarisClient;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import jakarta.ws.rs.core.Response;
+import java.io.IOException;
+import java.net.URI;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.rest.RESTCatalog;
+import org.apache.iceberg.types.Types;
+import org.apache.polaris.core.admin.model.Catalog;
+import org.apache.polaris.core.admin.model.CatalogGrant;
+import org.apache.polaris.core.admin.model.CatalogPrivilege;
+import org.apache.polaris.core.admin.model.CatalogProperties;
+import org.apache.polaris.core.admin.model.FileStorageConfigInfo;
+import org.apache.polaris.core.admin.model.GrantResource;
+import org.apache.polaris.core.admin.model.PolarisCatalog;
+import org.apache.polaris.core.admin.model.PrincipalWithCredentials;
+import org.apache.polaris.core.admin.model.StorageConfigInfo;
+import org.apache.polaris.service.it.env.ClientCredentials;
+import org.apache.polaris.service.it.env.DirectoryApi;
+import org.apache.polaris.service.it.env.IcebergHelper;
+import org.apache.polaris.service.it.env.IntegrationTestsHelper;
+import org.apache.polaris.service.it.env.ManagementApi;
+import org.apache.polaris.service.it.env.PolarisApiEndpoints;
+import org.apache.polaris.service.it.env.PolarisClient;
+import org.apache.polaris.service.it.ext.PolarisIntegrationTestExtension;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.io.TempDir;
+
+/**
+ * End-to-end test of the directories REST API on a file based catalog: the whole create, load and
+ * drop flow goes through the REST layer, including authorization and location validation.
+ */
+@ExtendWith(PolarisIntegrationTestExtension.class)
+public class PolarisDirectoryServiceIntegrationTest {

Review Comment:
   I think we are missing cases where directory exists then overlapping table, two overlapping directories.



##########
runtime/service/src/main/java/org/apache/polaris/service/catalog/directory/DirectoryCatalogHandler.java:
##########
@@ -0,0 +1,428 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.polaris.service.catalog.directory;
+
+import static org.apache.iceberg.types.Types.NestedField.optional;
+import static org.apache.iceberg.types.Types.NestedField.required;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.smallrye.common.annotation.Identifier;
+import jakarta.enterprise.inject.Instance;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+import java.util.regex.Pattern;
+import java.util.regex.PatternSyntaxException;
+import java.util.stream.Stream;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.catalog.Catalog;
+import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.exceptions.NoSuchNamespaceException;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.types.Types;
+import org.apache.polaris.core.auth.PolarisAuthorizableOperation;
+import org.apache.polaris.core.catalog.DirectoryCatalog;
+import org.apache.polaris.core.catalog.FederatedCatalogFactory;
+import org.apache.polaris.core.catalog.LocalCatalogFactory;
+import org.apache.polaris.core.catalog.PolarisCatalogHelpers;
+import org.apache.polaris.core.config.FeatureConfiguration;
+import org.apache.polaris.core.connection.ConnectionConfigInfoDpo;
+import org.apache.polaris.core.connection.ConnectionType;
+import org.apache.polaris.core.credentials.PolarisCredentialManager;
+import org.apache.polaris.core.entity.CatalogEntity;
+import org.apache.polaris.core.entity.PolarisEntitySubType;
+import org.apache.polaris.core.entity.PolarisEntityType;
+import org.apache.polaris.core.entity.table.DirectoryEntity;
+import org.apache.polaris.core.persistence.PolarisResolvedPathWrapper;
+import org.apache.polaris.core.persistence.pagination.Page;
+import org.apache.polaris.core.persistence.pagination.PageToken;
+import org.apache.polaris.core.persistence.resolver.PolarisResolutionManifest;
+import org.apache.polaris.core.persistence.resolver.ResolvedPathKey;
+import org.apache.polaris.core.persistence.resolver.ResolverPath;
+import org.apache.polaris.core.storage.PolarisStorageActions;
+import org.apache.polaris.core.storage.StorageAccessConfig;
+import org.apache.polaris.immutables.PolarisImmutable;
+import org.apache.polaris.service.catalog.common.CatalogHandler;
+import org.apache.polaris.service.catalog.common.CatalogUtils;
+import org.apache.polaris.service.catalog.iceberg.LocalIcebergCatalog;
+import org.apache.polaris.service.catalog.io.FileIOFactory;
+import org.apache.polaris.service.catalog.io.StorageAccessConfigProvider;
+import org.apache.polaris.service.catalog.spi.DirectoryScanService;
+import org.apache.polaris.service.catalog.validation.IcebergPropertiesValidation;
+import org.apache.polaris.service.types.Directory;
+import org.apache.polaris.service.types.DirectoryFilter;
+import org.apache.polaris.service.types.ListDirectoriesResponse;
+import org.apache.polaris.service.types.LoadDirectoryResponse;
+import org.apache.polaris.service.types.ScanDirectoryResponse;
+import org.apache.polaris.service.types.ScanSchedule;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@PolarisImmutable
+@SuppressWarnings("immutables:incompat")
+public abstract class DirectoryCatalogHandler extends CatalogHandler {
+  private static final Logger LOGGER = LoggerFactory.getLogger(DirectoryCatalogHandler.class);
+
+  /** Predefined Iceberg table schema for directory file metadata. */
+  public static final Schema DIRECTORY_TABLE_SCHEMA =
+      new Schema(
+          required(1, "file_uri", Types.StringType.get(), "The object URI"),
+          optional(
+              2, "content_type", Types.StringType.get(), "The MIME content type of the object"),
+          optional(3, "size", Types.LongType.get(), "The size of the object in bytes"),
+          optional(
+              4,
+              "checksum_algorithm",
+              Types.StringType.get(),
+              "The checksum algorithm (e.g. MD5, SHA-256)"),
+          optional(5, "checksum", Types.StringType.get(), "The file checksum"),
+          optional(
+              6,
+              "last_modified",
+              Types.TimestampType.withZone(),
+              "The last modification timestamp"),
+          optional(
+              7,
+              "metadata",
+              Types.MapType.ofRequired(8, 9, Types.StringType.get(), Types.StringType.get()),
+              "Additional key-value metadata from the object store"));
+
+  /**
+   * Suffix of the inventory table name. A directory and its inventory table cannot share a name:
+   * both are table-like entities, and names are unique among table-like entities of a namespace.
+   */
+  static final String INVENTORY_TABLE_SUFFIX = "__inventory";
+
+  static TableIdentifier inventoryTableIdentifier(TableIdentifier directory) {
+    return TableIdentifier.of(directory.namespace(), directory.name() + INVENTORY_TABLE_SUFFIX);
+  }
+
+  private static final ObjectMapper MAPPER = new ObjectMapper();
+  private static final TypeReference<List<String>> LIST_OF_STRING = new TypeReference<>() {};
+
+  static String toJsonArray(List<String> list) {
+    if (list == null) {
+      return null;
+    }
+    try {
+      return MAPPER.writeValueAsString(list);
+    } catch (JsonProcessingException e) {
+      throw new IllegalArgumentException("Failed to serialize filter list", e);
+    }
+  }
+
+  private static List<String> parseJsonArray(String json) {
+    if (json == null) {
+      return null;
+    }
+    try {
+      return MAPPER.readValue(json, LIST_OF_STRING);
+    } catch (JsonProcessingException e) {
+      throw new IllegalArgumentException("Failed to parse filter list", e);
+    }
+  }
+
+  private static DirectoryFilter toDirectoryFilter(DirectoryEntity entity) {
+    List<String> include = parseJsonArray(entity.getFilterInclude());
+    List<String> exclude = parseJsonArray(entity.getFilterExclude());
+    if (include == null && exclude == null) {
+      return null;
+    }
+    return DirectoryFilter.builder().setInclude(include).setExclude(exclude).build();
+  }
+
+  /** Rejects filters that are not valid regular expressions. */
+  private static void validateFilters(List<String> include, List<String> exclude) {
+    try {
+      Stream.of(include, exclude)
+          .filter(Objects::nonNull)
+          .flatMap(List::stream)
+          .forEach(Pattern::compile);
+    } catch (PatternSyntaxException e) {
+      throw new IllegalArgumentException("Invalid directory filter: " + e.getMessage(), e);
+    }
+  }
+
+  static String scanScheduleToJson(ScanSchedule scanSchedule) {
+    if (scanSchedule == null) {
+      return null;
+    }
+    try {
+      return MAPPER.writeValueAsString(scanSchedule);
+    } catch (JsonProcessingException e) {
+      throw new IllegalArgumentException("Failed to serialize scan schedule", e);
+    }
+  }
+
+  private static ScanSchedule scanScheduleFromJson(String json) {
+    if (json == null) {
+      return null;
+    }
+    try {
+      return MAPPER.readValue(json, ScanSchedule.class);
+    } catch (JsonProcessingException e) {
+      throw new IllegalArgumentException("Failed to parse scan schedule", e);
+    }
+  }
+
+  protected abstract PolarisCredentialManager credentialManager();
+
+  protected abstract Instance<FederatedCatalogFactory> federatedCatalogFactories();
+
+  protected abstract LocalCatalogFactory localCatalogFactory();
+
+  protected abstract StorageAccessConfigProvider storageAccessConfigProvider();
+
+  protected abstract FileIOFactory fileIOFactory();
+
+  protected abstract DirectoryScanService directoryScanService();
+
+  private DirectoryCatalog directoryCatalog;
+  private Catalog icebergCatalog;
+  private TableIdentifier inventoryTable;
+
+  /**
+   * The underlying Iceberg catalog manipulates the inventory table, which is not the identifier
+   * targeted by the request: it must be added as a passthrough resolution path as well.
+   */
+  @Override
+  protected PolarisResolutionManifest newResolutionManifest() {
+    PolarisResolutionManifest manifest = super.newResolutionManifest();
+    if (inventoryTable != null) {
+      manifest.addPassthroughPath(
+          new ResolverPath(
+              PolarisCatalogHelpers.tableIdentifierToList(inventoryTable),
+              PolarisEntityType.TABLE_LIKE,
+              true /* optional */));
+    }
+    return manifest;
+  }
+
+  @Override
+  protected void initializeCatalog() {
+    CatalogEntity resolvedCatalogEntity = resolutionManifest.getResolvedCatalogEntity();
+    ConnectionConfigInfoDpo connectionConfigInfoDpo =
+        resolvedCatalogEntity.getConnectionConfigInfoDpo();
+    if (connectionConfigInfoDpo != null) {
+      LOGGER
+          .atInfo()
+          .addKeyValue("remoteUrl", connectionConfigInfoDpo.getUri())
+          .log("Initializing federated catalog");
+      FeatureConfiguration.enforceFeatureEnabledOrThrow(
+          realmConfig(), FeatureConfiguration.ENABLE_CATALOG_FEDERATION);
+
+      DirectoryCatalog federatedCatalog;
+      ConnectionType connectionType =
+          ConnectionType.fromCode(connectionConfigInfoDpo.getConnectionTypeCode());
+
+      Instance<FederatedCatalogFactory> federatedCatalogFactory =
+          federatedCatalogFactories()
+              .select(Identifier.Literal.of(connectionType.getFactoryIdentifier()));
+      if (federatedCatalogFactory.isResolvable()) {
+        Map<String, String> catalogProperties = resolvedCatalogEntity.getPropertiesAsMap();
+        federatedCatalog =
+            federatedCatalogFactory
+                .get()
+                .createDirectoryCatalog(
+                    connectionConfigInfoDpo, credentialManager(), catalogProperties);
+      } else {
+        throw new UnsupportedOperationException(
+            "External catalog factory for type '" + connectionType + "' is unavailable.");
+      }
+      this.directoryCatalog = federatedCatalog;
+    } else {
+      LOGGER.atInfo().log("Initializing non-federated catalog");
+      this.directoryCatalog =
+          new PolarisDirectoryCatalog(metaStoreManager(), callContext(), this.resolutionManifest);
+      this.directoryCatalog.initialize(catalogName(), Map.of());
+    }
+
+    // Initialize an IcebergCatalog for creating/dropping the associated Iceberg tables
+    this.icebergCatalog = localCatalogFactory().createCatalog(this.resolutionManifest);
+  }
+
+  public ListDirectoriesResponse listDirectories(
+      Namespace parent, String pageToken, Integer pageSize) {
+    PolarisAuthorizableOperation op = PolarisAuthorizableOperation.LIST_TABLES;
+    authorizeBasicNamespaceOperationOrThrow(op, parent);
+
+    PageToken pageRequest =
+        PageToken.build(pageToken, pageSize, maxPageSize(), this::shouldDecodeToken);
+    Page<TableIdentifier> page = directoryCatalog.listDirectories(parent, pageRequest);
+    rejectIncompleteListing(pageToken, pageSize, page.encodedResponseToken());
+    return ListDirectoriesResponse.builder()
+        .setIdentifiers(new LinkedHashSet<>(page.items()))
+        .setNextPageToken(page.encodedResponseToken())
+        .build();
+  }
+
+  public LoadDirectoryResponse createDirectory(
+      TableIdentifier identifier,
+      String baseLocation,
+      String filterInclude,
+      String filterExclude,
+      String scanSchedule,
+      Map<String, String> properties) {
+    this.inventoryTable = inventoryTableIdentifier(identifier);
+    PolarisAuthorizableOperation op = PolarisAuthorizableOperation.CREATE_TABLE_DIRECT;
+    authorizeCreateTableLikeUnderNamespaceOperationOrThrow(op, identifier);
+
+    PolarisResolvedPathWrapper resolvedParent =
+        resolutionManifest.getResolvedPath(ResolvedPathKey.ofNamespace(identifier.namespace()));
+    if (resolvedParent == null) {
+      throw new NoSuchNamespaceException("Namespace does not exist: %s", identifier.namespace());
+    }
+    // The base location is read with vended credentials: it must be within the allowed locations
+    CatalogUtils.validateLocationsForTableLike(
+        realmConfig(), identifier, Set.of(baseLocation), resolvedParent);
+    // The scan exposes the names and sizes of the objects under the base location: it must not
+    // overlap the location of a sibling table, view, directory or namespace
+    ((LocalIcebergCatalog) icebergCatalog)

Review Comment:
   1. This comment is helpful but cant solve not sufficient on its own. ModelEntity.fromEntity only sets location_without_scheme for Iceberg table/view and namespace — not DIRECTORY:
   
   ModelEntity.java
   ```
   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) {
   ```
   Directories store base_location in internal props (DirectoryEntity.BASE_LOCATION_KEY), not location like Iceberg entities. So generateOverlapQuery may never see them even if asLocationBasedEntity is fixed.
   
   2. MutationAttempt.updateLocationsIndex only reads `ENTITY_BASE_LOCATION` ("location") from obj properties. Directories use baseLocation on DirectoryObj / base_location in internal props. Same overlap gap on NoSQL backends.



##########
runtime/service/src/main/java/org/apache/polaris/service/catalog/directory/DirectoryCatalogHandler.java:
##########
@@ -0,0 +1,428 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.polaris.service.catalog.directory;
+
+import static org.apache.iceberg.types.Types.NestedField.optional;
+import static org.apache.iceberg.types.Types.NestedField.required;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.smallrye.common.annotation.Identifier;
+import jakarta.enterprise.inject.Instance;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+import java.util.regex.Pattern;
+import java.util.regex.PatternSyntaxException;
+import java.util.stream.Stream;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.catalog.Catalog;
+import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.exceptions.NoSuchNamespaceException;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.types.Types;
+import org.apache.polaris.core.auth.PolarisAuthorizableOperation;
+import org.apache.polaris.core.catalog.DirectoryCatalog;
+import org.apache.polaris.core.catalog.FederatedCatalogFactory;
+import org.apache.polaris.core.catalog.LocalCatalogFactory;
+import org.apache.polaris.core.catalog.PolarisCatalogHelpers;
+import org.apache.polaris.core.config.FeatureConfiguration;
+import org.apache.polaris.core.connection.ConnectionConfigInfoDpo;
+import org.apache.polaris.core.connection.ConnectionType;
+import org.apache.polaris.core.credentials.PolarisCredentialManager;
+import org.apache.polaris.core.entity.CatalogEntity;
+import org.apache.polaris.core.entity.PolarisEntitySubType;
+import org.apache.polaris.core.entity.PolarisEntityType;
+import org.apache.polaris.core.entity.table.DirectoryEntity;
+import org.apache.polaris.core.persistence.PolarisResolvedPathWrapper;
+import org.apache.polaris.core.persistence.pagination.Page;
+import org.apache.polaris.core.persistence.pagination.PageToken;
+import org.apache.polaris.core.persistence.resolver.PolarisResolutionManifest;
+import org.apache.polaris.core.persistence.resolver.ResolvedPathKey;
+import org.apache.polaris.core.persistence.resolver.ResolverPath;
+import org.apache.polaris.core.storage.PolarisStorageActions;
+import org.apache.polaris.core.storage.StorageAccessConfig;
+import org.apache.polaris.immutables.PolarisImmutable;
+import org.apache.polaris.service.catalog.common.CatalogHandler;
+import org.apache.polaris.service.catalog.common.CatalogUtils;
+import org.apache.polaris.service.catalog.iceberg.LocalIcebergCatalog;
+import org.apache.polaris.service.catalog.io.FileIOFactory;
+import org.apache.polaris.service.catalog.io.StorageAccessConfigProvider;
+import org.apache.polaris.service.catalog.spi.DirectoryScanService;
+import org.apache.polaris.service.catalog.validation.IcebergPropertiesValidation;
+import org.apache.polaris.service.types.Directory;
+import org.apache.polaris.service.types.DirectoryFilter;
+import org.apache.polaris.service.types.ListDirectoriesResponse;
+import org.apache.polaris.service.types.LoadDirectoryResponse;
+import org.apache.polaris.service.types.ScanDirectoryResponse;
+import org.apache.polaris.service.types.ScanSchedule;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@PolarisImmutable
+@SuppressWarnings("immutables:incompat")
+public abstract class DirectoryCatalogHandler extends CatalogHandler {
+  private static final Logger LOGGER = LoggerFactory.getLogger(DirectoryCatalogHandler.class);
+
+  /** Predefined Iceberg table schema for directory file metadata. */
+  public static final Schema DIRECTORY_TABLE_SCHEMA =
+      new Schema(
+          required(1, "file_uri", Types.StringType.get(), "The object URI"),
+          optional(
+              2, "content_type", Types.StringType.get(), "The MIME content type of the object"),
+          optional(3, "size", Types.LongType.get(), "The size of the object in bytes"),
+          optional(
+              4,
+              "checksum_algorithm",
+              Types.StringType.get(),
+              "The checksum algorithm (e.g. MD5, SHA-256)"),
+          optional(5, "checksum", Types.StringType.get(), "The file checksum"),
+          optional(
+              6,
+              "last_modified",
+              Types.TimestampType.withZone(),
+              "The last modification timestamp"),
+          optional(
+              7,
+              "metadata",
+              Types.MapType.ofRequired(8, 9, Types.StringType.get(), Types.StringType.get()),
+              "Additional key-value metadata from the object store"));
+
+  /**
+   * Suffix of the inventory table name. A directory and its inventory table cannot share a name:
+   * both are table-like entities, and names are unique among table-like entities of a namespace.
+   */
+  static final String INVENTORY_TABLE_SUFFIX = "__inventory";
+
+  static TableIdentifier inventoryTableIdentifier(TableIdentifier directory) {
+    return TableIdentifier.of(directory.namespace(), directory.name() + INVENTORY_TABLE_SUFFIX);
+  }
+
+  private static final ObjectMapper MAPPER = new ObjectMapper();
+  private static final TypeReference<List<String>> LIST_OF_STRING = new TypeReference<>() {};
+
+  static String toJsonArray(List<String> list) {
+    if (list == null) {
+      return null;
+    }
+    try {
+      return MAPPER.writeValueAsString(list);
+    } catch (JsonProcessingException e) {
+      throw new IllegalArgumentException("Failed to serialize filter list", e);
+    }
+  }
+
+  private static List<String> parseJsonArray(String json) {
+    if (json == null) {
+      return null;
+    }
+    try {
+      return MAPPER.readValue(json, LIST_OF_STRING);
+    } catch (JsonProcessingException e) {
+      throw new IllegalArgumentException("Failed to parse filter list", e);
+    }
+  }
+
+  private static DirectoryFilter toDirectoryFilter(DirectoryEntity entity) {
+    List<String> include = parseJsonArray(entity.getFilterInclude());
+    List<String> exclude = parseJsonArray(entity.getFilterExclude());
+    if (include == null && exclude == null) {
+      return null;
+    }
+    return DirectoryFilter.builder().setInclude(include).setExclude(exclude).build();
+  }
+
+  /** Rejects filters that are not valid regular expressions. */
+  private static void validateFilters(List<String> include, List<String> exclude) {
+    try {
+      Stream.of(include, exclude)
+          .filter(Objects::nonNull)
+          .flatMap(List::stream)
+          .forEach(Pattern::compile);
+    } catch (PatternSyntaxException e) {
+      throw new IllegalArgumentException("Invalid directory filter: " + e.getMessage(), e);
+    }
+  }
+
+  static String scanScheduleToJson(ScanSchedule scanSchedule) {
+    if (scanSchedule == null) {
+      return null;
+    }
+    try {
+      return MAPPER.writeValueAsString(scanSchedule);
+    } catch (JsonProcessingException e) {
+      throw new IllegalArgumentException("Failed to serialize scan schedule", e);
+    }
+  }
+
+  private static ScanSchedule scanScheduleFromJson(String json) {
+    if (json == null) {
+      return null;
+    }
+    try {
+      return MAPPER.readValue(json, ScanSchedule.class);
+    } catch (JsonProcessingException e) {
+      throw new IllegalArgumentException("Failed to parse scan schedule", e);
+    }
+  }
+
+  protected abstract PolarisCredentialManager credentialManager();
+
+  protected abstract Instance<FederatedCatalogFactory> federatedCatalogFactories();
+
+  protected abstract LocalCatalogFactory localCatalogFactory();
+
+  protected abstract StorageAccessConfigProvider storageAccessConfigProvider();
+
+  protected abstract FileIOFactory fileIOFactory();
+
+  protected abstract DirectoryScanService directoryScanService();
+
+  private DirectoryCatalog directoryCatalog;
+  private Catalog icebergCatalog;
+  private TableIdentifier inventoryTable;
+
+  /**
+   * The underlying Iceberg catalog manipulates the inventory table, which is not the identifier
+   * targeted by the request: it must be added as a passthrough resolution path as well.
+   */
+  @Override
+  protected PolarisResolutionManifest newResolutionManifest() {
+    PolarisResolutionManifest manifest = super.newResolutionManifest();
+    if (inventoryTable != null) {
+      manifest.addPassthroughPath(
+          new ResolverPath(
+              PolarisCatalogHelpers.tableIdentifierToList(inventoryTable),
+              PolarisEntityType.TABLE_LIKE,
+              true /* optional */));
+    }
+    return manifest;
+  }
+
+  @Override
+  protected void initializeCatalog() {
+    CatalogEntity resolvedCatalogEntity = resolutionManifest.getResolvedCatalogEntity();
+    ConnectionConfigInfoDpo connectionConfigInfoDpo =
+        resolvedCatalogEntity.getConnectionConfigInfoDpo();
+    if (connectionConfigInfoDpo != null) {
+      LOGGER
+          .atInfo()
+          .addKeyValue("remoteUrl", connectionConfigInfoDpo.getUri())
+          .log("Initializing federated catalog");
+      FeatureConfiguration.enforceFeatureEnabledOrThrow(
+          realmConfig(), FeatureConfiguration.ENABLE_CATALOG_FEDERATION);
+
+      DirectoryCatalog federatedCatalog;
+      ConnectionType connectionType =
+          ConnectionType.fromCode(connectionConfigInfoDpo.getConnectionTypeCode());
+
+      Instance<FederatedCatalogFactory> federatedCatalogFactory =
+          federatedCatalogFactories()
+              .select(Identifier.Literal.of(connectionType.getFactoryIdentifier()));
+      if (federatedCatalogFactory.isResolvable()) {
+        Map<String, String> catalogProperties = resolvedCatalogEntity.getPropertiesAsMap();
+        federatedCatalog =
+            federatedCatalogFactory
+                .get()
+                .createDirectoryCatalog(

Review Comment:
   After ENABLE_CATALOG_FEDERATION, createDirectoryCatalog defaults to UnsupportedOperationException. Directories enabled on a federated catalog fail at runtime with a generic unsupported error. Rejecting earlier with a clear message would be clearer or we can document “internal catalog only”.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org