You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@polaris.apache.org by "jbonofre (via GitHub)" <gi...@apache.org> on 2026/06/04 09:48:58 UTC
[PR] Add Directories feature for managing objects (including unstructured data) on object stores [polaris]
jbonofre opened a new pull request, #4613:
URL: https://github.com/apache/polaris/pull/4613
Directories allow users to point Polaris at a location/prefix on an object store (S3, GCS, Azure) containing objects (including unstructured data like images, videos, documents) and automatically track the objects via an Iceberg table with objecr-level metadata (location, size, content type, checksum, last modified, etc.).
When a directory is created, Polaris stores the directory configuration (name, location, scan filters, scan schedule) as a DIRECTORY entity and also creates an associated Iceberg table with the predefined schema.
## Checklist
- [X] ๐ก๏ธ Don't disclose security issues! (contact security@apache.org)
- [X] ๐ Clearly explained why the changes are needed, or linked related issues: Fixes #
- [X] ๐งช Added/updated tests with good coverage, or manually tested (and explained how)
- [X] ๐ก Added comments for complex logic
- [ ] ๐งพ 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 Directories feature for managing objects (including unstructured data) on object stores [polaris]
Posted by "adutra (via GitHub)" <gi...@apache.org>.
adutra commented on code in PR #4613:
URL: https://github.com/apache/polaris/pull/4613#discussion_r3389738729
##########
site/content/in-dev/unreleased/directories.md:
##########
@@ -0,0 +1,227 @@
+---
+#
+# 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.
+#
+title: Directories
+type: docs
+weight: 450
+---
+
+## Overview
+
+Directories make objects (including unstructured data like images, videos, documents, and other objects) discoverable alongside
+structured Iceberg tables within a Polaris catalog.
+A directory points to a base location/prefix on an object store and automatically tracks the objects it contains by maintaining
+an Iceberg table with object-level metadata such as URI, size, content type, checksum, ...
+
+This means query engines and tools that already know how to read Iceberg tables can discover and
+access unstructured data with little or no extra work (accessing the object itself).
+
+## Concepts
+
+A directory has two main parts:
+
+1. **Directory configuration** โ stored by the Polaris server. It describes _where_ the data lives,
+ how to authenticate, which objects to include, and how often to re-scan. The configuration "lives" in a namespace.
+2. **Directory table** โ an Iceberg table serving as the inventory of all objects contained in the directory, one row per object discovered during a scan.
+ The directory table uses the configuration name.
+
+The Polaris server itself does not perform scans. Instead, external services (e.g. directory table scanning service) read the directory configuration through the REST API,
+walk the object store, and write the results into the directory table.
+
+## Configuration
+
+A directory is described by the following fields:
+
+| Field | Type | Required | Description |
+|-------|------|----------|-------------|
+| `name` | `string` | Yes | The name of the directory. It is used in REST endpoint paths and becomes the name of the corresponding Iceberg table. |
+| `base_location` | `string` | Yes | The base location to use as a root for scanning objects in the object store (for example `s3://my-bucket/images/`, `gs://my-bucket/docs/`, or `file:///data/local/`). |
+| `filter` | `object` | No | Include and exclude patterns that control which objects are added to the directory table during a scan. See [Filter](#filter). |
+| `scan-schedule` | `object` | No | Object representing a scan schedule (trigger, cron, ...). |
+
+### Storage access
+
+Directories use Polaris `StorageAccessConfig` to access the object store.
+
+### Filter
+
+The `filter` object controls which objects are included in the directory table during a scan.
+
+If no filter is set, all objects found under the directory `base_location` are included.
+
+`include` and `exclude` are lists of regular expressions matched against the object's full URI. An
+object is included if it matches at least one `include` pattern (or `include` is omitted) and does
+not match any `exclude` pattern.
+
+**Example** โ include only JPEG and PNG images, but exclude thumbnails:
+
+```json
+{
+ "filter": {
+ "include": [".*\\.jpg$", ".*\\.png$"],
+ "exclude": [".*thumbs/.*"]
+ }
+}
+```
+
+### Example: creating a directory
+
+```json
+POST /v1/{prefix}/namespaces/{namespace}/directories
+
+{
+ "name": "product-images",
+ "base_location": "s3://warehouse/product-images/",
+ "filter": {
+ "include": [".*\\.jpg$", ".*\\.png$"]
+ },
+ "scan-schedule": {
+ "cron": "0 * * * *"
+ }
+}
+```
+
+## REST endpoints
+
+Directories are managed through the following REST endpoints. All paths are relative to the catalog
+base URL.
+
+### List directories
+
+**GET** `/v1/{prefix}/namespaces/{namespace}/directories`
+
+Returns the list of directory identifiers in the given namespace.
+
+### Create a directory
+
+**POST** `/v1/{prefix}/namespaces/{namespace}/directories`
+
+Creates a new directory and its corresponding Iceberg table. The request body must contain the
+directory configuration (see [Configuration](#configuration)).
+
+### Get directory details
+
+**GET** `/v1/{prefix}/namespaces/{namespace}/directories/{directory}`
+
+Returns the full configuration of the specified directory, including filter and scan schedule.
+
+### Drop a directory
+
+**DELETE** `/v1/{prefix}/namespaces/{namespace}/directories/{directory}`
+
+Removes the directory and its associated Iceberg table from the namespace.
+
+## Directory table
+
+When a directory is created, Polaris creates an Iceberg table in the same namespace using the directory
+`name` as the table name. The table uses the following schema:
+
+| Field Id | Field Name | Type | Required | Description |
+|----------|------------|------|----------|-------------|
+| 1 | `file_uri` | `string` | Yes | The fully qualified URI of the object (for example `s3://my-bucket/images/photo.jpg`). |
+| 2 | `content_type` | `string` | No | The MIME content type (RFC 2045) of the object (for example `image/jpeg`, `application/pdf`). |
+| 3 | `size` | `long` | No | The size of the object in bytes. |
+| 4 | `checksum_algorithm` | `string` | No | The name of the checksum algorithm (for example `MD5`, `SHA-256`, `CRC32`). |
+| 5 | `checksum` | `string` | No | The object checksum value computed with the algorithm specified in `checksum_algorithm`. |
+| 6 | `last_modified` | `timestamptz` | No | The last modification timestamp of the object as reported by the object store. |
+| 7 | `metadata` | `map<string, string>` | No | Additional labels and tags from the object store (for example S3 user-defined metadata, storage class, or content encoding). |
+
+Clients (query engines, analytics tools, etc.) interact with the directory table as a regular Iceberg
+table. For instance, you can query it with Spark SQL:
+
+```sql
+SELECT file_uri, size, last_modified
+FROM my_catalog.my_namespace.product_images
+WHERE content_type = 'image/jpeg'
+ AND size > 1048576
+ORDER BY last_modified DESC;
+```
+
+### Credential vending
+
+The directory table might contain additional fields for credential vending.
Review Comment:
I would like to understand how we envision credential vending.
Right now, if the spec was live, clients could request vended credentials *for the inventory table*, but would necessarily have to be in possession of their own storage credentials, in order to access the actual data.
I think it would be a real added-value if Polaris could also be the governing entity that grants access to the remote files, via some form of credential vending.
It's too soon to sort out the details, but my gut feeling is that we'll need a form of "fast authz checks", similar to what was proposed in #3995 for remote signing. In which case, it's likely that the inventory table would have to store some encrypted payload that will be used later for generating the storage credentials.
--
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 Directories feature for managing objects (including unstructured data) on object stores [polaris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #4613:
URL: https://github.com/apache/polaris/pull/4613#issuecomment-4949588044
This PR is stale because it has been open 30 days with no activity. Remove stale label or comment or this will be closed in 5 days.
--
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 Directories feature for managing objects (including unstructured data) on object stores [polaris]
Posted by "jbonofre (via GitHub)" <gi...@apache.org>.
jbonofre commented on code in PR #4613:
URL: https://github.com/apache/polaris/pull/4613#discussion_r3364921398
##########
site/content/in-dev/unreleased/directories.md:
##########
@@ -0,0 +1,227 @@
+---
+#
+# 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.
+#
+title: Directories
+type: docs
+weight: 450
+---
+
+## Overview
+
+Directories make objects (including unstructured data like images, videos, documents, and other objects) discoverable alongside
+structured Iceberg tables within a Polaris catalog.
+A directory points to a base location/prefix on an object store and automatically tracks the objects it contains by maintaining
+an Iceberg table with object-level metadata such as URI, size, content type, checksum, ...
+
+This means query engines and tools that already know how to read Iceberg tables can discover and
+access unstructured data with little or no extra work (accessing the object itself).
+
+## Concepts
+
+A directory has two main parts:
+
+1. **Directory configuration** โ stored by the Polaris server. It describes _where_ the data lives,
+ how to authenticate, which objects to include, and how often to re-scan. The configuration "lives" in a namespace.
+2. **Directory table** โ an Iceberg table serving as the inventory of all objects contained in the directory, one row per object discovered during a scan.
+ The directory table uses the configuration name.
+
+The Polaris server itself does not perform scans. Instead, external services (e.g. directory table scanning service) read the directory configuration through the REST API,
+walk the object store, and write the results into the directory table.
+
+## Configuration
+
+A directory is described by the following fields:
+
+| Field | Type | Required | Description |
+|-------|------|----------|-------------|
+| `name` | `string` | Yes | The name of the directory. It is used in REST endpoint paths and becomes the name of the corresponding Iceberg table. |
Review Comment:
You are right: the directory table and the directory configuration share the same name (the configuration name is used as table name) and they are located in the same namespace. The "distinction" is based on object type (configuration vs table).
--
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 Directories feature for managing objects (including unstructured data) on object stores [polaris]
Posted by "dimas-b (via GitHub)" <gi...@apache.org>.
dimas-b commented on code in PR #4613:
URL: https://github.com/apache/polaris/pull/4613#discussion_r4418525764
##########
runtime/service/src/main/java/org/apache/polaris/service/catalog/directory/DirectoryConfigEndpoints.java:
##########
@@ -0,0 +1,63 @@
+/*
+ * 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 com.google.common.collect.ImmutableSet;
+import jakarta.annotation.Priority;
+import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.inject.Inject;
+import java.util.Set;
+import org.apache.iceberg.rest.Endpoint;
+import org.apache.polaris.core.config.FeatureConfiguration;
+import org.apache.polaris.core.config.RealmConfig;
+import org.apache.polaris.core.rest.DirectoryEndpoints;
+import org.apache.polaris.service.catalog.spi.CatalogConfigEndpointContributor;
+
+@ApplicationScoped
+@Priority(350)
+public class DirectoryConfigEndpoints implements CatalogConfigEndpointContributor {
Review Comment:
I'm not sure what we gain by exposing Directory endpoints in the IRC config.
Previous discussion on this topic identified only a very narrow use case for Spark Clients discovering Generic Tables:
https://lists.apache.org/thread/84yc07xdo7p7385vvtm9xomm7oxwh7o8
Do we expect a Directories API client to check IRC config for the presence of Directory endpoints?
##########
polaris-core/src/main/java/org/apache/polaris/core/rest/PolarisResourcePaths.java:
##########
@@ -94,6 +102,27 @@ public String credentialsPath(TableIdentifier ident) {
"credentials");
}
+ public String directories(Namespace ns) {
Review Comment:
This is used only in tests. Could we move it to test utilities?
##########
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,
Review Comment:
Why `CallContext`? We only need `RealmConfig` (line 57) ๐ค
##########
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 =
+ PolarisConfiguration.<Boolean>builder()
+ .key("ENABLE_DIRECTORIES")
+ .description("If true, the directories endpoints are enabled")
+ .defaultValue(true)
+ .buildFeatureConfiguration();
+
+ public static final FeatureConfiguration<Boolean> ENABLE_DIRECTORY_SCAN =
+ PolarisConfiguration.<Boolean>builder()
+ .key("ENABLE_DIRECTORY_SCAN")
Review Comment:
Do we need a flag for this? It could be disabled via the Authorizer, I guess ๐ค
##########
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);
Review Comment:
`PolarisPrincipal` is generally available in `@RequestScoped` beans via CDI... Do we need the `SecurityContext` complication? ๐ค
##########
runtime/service/src/main/java/org/apache/polaris/service/catalog/directory/DirectoryCatalogHandler.java:
##########
@@ -0,0 +1,412 @@
+/*
+ * 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.Optional;
+import java.util.Set;
+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();
+ }
+
+ 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:
This call always throws, does it not? ๐ค
##########
runtime/service/src/main/java/org/apache/polaris/service/catalog/directory/DirectoryCatalogHandler.java:
##########
@@ -0,0 +1,412 @@
+/*
+ * 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.Optional;
+import java.util.Set;
+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();
+ }
+
+ 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(
Review Comment:
@adutra : could review from the recent location check improvements perspective?
##########
runtime/service/build.gradle.kts:
##########
@@ -60,6 +60,9 @@ dependencies {
implementation(platform(libs.iceberg.bom))
implementation("org.apache.iceberg:iceberg-api")
implementation("org.apache.iceberg:iceberg-core")
+ implementation("org.apache.iceberg:iceberg-data")
+ implementation("org.apache.iceberg:iceberg-parquet")
+ implementation(libs.parquet.hadoop)
Review Comment:
Could we put Directories impl. code into a separate module to avoid these new deps in `runtime/service`?
This is our usual practice to avoid automatically pushing new feature deps to downstream servers. Pretty much all downstream servers have to depend on `runtime/service`. Ideally optional features should only be bundled in `runtime/server` to allow downstream project to opt in/out.
##########
runtime/service/src/test/java/org/apache/polaris/service/catalog/directory/PolarisDirectoryCatalogRelationalTest.java:
##########
@@ -0,0 +1,27 @@
+/*
+ * 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 org.apache.polaris.service.Profiles;
+
+@QuarkusTest
+@TestProfile(Profiles.DefaultProfile.class)
+public class PolarisDirectoryCatalogRelationalTest extends AbstractPolarisDirectoryCatalogTest {}
Review Comment:
What makes it "relational"? Did you mean `RelationalJdbcProfile`? ๐ค
##########
polaris-core/src/main/java/org/apache/polaris/core/rest/PolarisResourcePaths.java:
##########
@@ -40,6 +40,14 @@ public class PolarisResourcePaths {
public static final String V1_GENERIC_TABLE =
"polaris/v1/{prefix}/namespaces/{namespace}/generic-tables/{generic-table}";
+ // Directory endpoints
+ public static final String V1_DIRECTORIES =
Review Comment:
Why not keep it in `DirectoryEndpoints` for better feature isolation?
##########
polaris-core/src/main/java/org/apache/polaris/core/catalog/FederatedCatalogFactory.java:
##########
@@ -63,4 +63,23 @@ GenericTableCatalog createGenericCatalog(
ConnectionConfigInfoDpo connectionConfig,
PolarisCredentialManager polarisCredentialManager,
Map<String, String> catalogProperties);
+
+ /**
+ * Creates a directory catalog for the given connection configuration.
+ *
+ * @param connectionConfig the connection configuration
+ * @param polarisCredentialManager the credential manager for generating connection credentials
+ * that Polaris uses to access external systems
+ * @param catalogProperties additional properties from the ExternalCatalog entity that should be
+ * passed through to the underlying catalog
+ * @return the initialized catalog
+ * @throws IllegalStateException if the connection configuration is invalid
+ * @throws UnsupportedOperationException if directory federation is not supported
+ */
+ default DirectoryCatalog createDirectoryCatalog(
+ ConnectionConfigInfoDpo connectionConfig,
+ PolarisCredentialManager polarisCredentialManager,
+ Map<String, String> catalogProperties) {
+ throw new UnsupportedOperationException("Directory federation is not supported.");
Review Comment:
This is awkward... why should a "federated" Catalog Factory care about Directories at all? It does not look like we have any support for Directories in Federated catalogs ATM ๐ค
##########
runtime/service/src/main/java/org/apache/polaris/service/catalog/directory/DirectoryCatalogHandler.java:
##########
@@ -0,0 +1,412 @@
+/*
+ * 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.Optional;
+import java.util.Set;
+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();
+ }
+
+ 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());
+
+ DirectoryScanner.validateFilters(parseJsonArray(filterInclude), parseJsonArray(filterExclude));
+ TableIdentifier tableId = inventoryTableIdentifier(identifier);
Review Comment:
duplicates line 273
##########
runtime/service/src/main/java/org/apache/polaris/service/catalog/directory/DirectoryCatalogHandler.java:
##########
@@ -0,0 +1,412 @@
+/*
+ * 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.Optional;
+import java.util.Set;
+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 =
Review Comment:
How do we version this schema? Should it be part of the GET Directory API (as a property perhaps)?
##########
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);
Review Comment:
`handlerFactory` is request-scoped. so it should be able to have an implicit (CDI-based) `principal`.
--
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 Directories feature for managing objects (including unstructured data) on object stores [polaris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #4613:
URL: https://github.com/apache/polaris/pull/4613#issuecomment-5746802511
This PR is stale because it has been open 30 days with no activity. Remove stale label or comment or this will be closed in 5 days.
--
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 Directories feature for managing objects (including unstructured data) on object stores [polaris]
Posted by "adutra (via GitHub)" <gi...@apache.org>.
adutra commented on code in PR #4613:
URL: https://github.com/apache/polaris/pull/4613#discussion_r3389550749
##########
spec/polaris-catalog-apis/directories-api.yaml:
##########
@@ -0,0 +1,334 @@
+#
+# 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.
+#
+
+---
+paths:
+ /polaris/v1/{prefix}/namespaces/{namespace}/directories:
+ parameters:
+ - $ref: '../iceberg-rest-catalog-open-api.yaml#/components/parameters/prefix'
+ - $ref: '../iceberg-rest-catalog-open-api.yaml#/components/parameters/namespace'
+
+ get:
+ tags:
+ - Directory API
+ summary: List all directory identifiers underneath a given namespace
+ description: Return all directory identifiers under this namespace
+ operationId: listDirectories
+ parameters:
+ - $ref: '../iceberg-rest-catalog-open-api.yaml#/components/parameters/page-token'
+ - $ref: '../iceberg-rest-catalog-open-api.yaml#/components/parameters/page-size'
+ responses:
+ 200:
+ $ref: '#/components/responses/ListDirectoriesResponse'
+ 400:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/responses/BadRequestErrorResponse'
+ 401:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/responses/UnauthorizedResponse'
+ 403:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/responses/ForbiddenResponse'
+ 404:
+ description: Not Found - The namespace specified does not exist
+ content:
+ application/json:
+ schema:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/schemas/IcebergErrorResponse'
+ examples:
+ NamespaceNotFound:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/examples/NoSuchNamespaceError'
+ 503:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/responses/ServiceUnavailableResponse'
+ 5XX:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/responses/ServerErrorResponse'
+
+ post:
+ tags:
+ - Directory API
+ summary: Create a directory under the given namespace
+ description:
+ Create a directory under the given namespace, and return the created directory information as a response.
+ operationId: createDirectory
+ parameters:
+ - $ref: '#/components/parameters/polaris-directory-data-access'
+ requestBody:
+ required: true
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/CreateDirectoryRequest'
+ responses:
+ 200:
+ $ref: '#/components/responses/CreateDirectoryResponse'
+ 400:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/responses/BadRequestErrorResponse'
+ 401:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/responses/UnauthorizedResponse'
+ 403:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/responses/ForbiddenResponse'
+ 404:
+ description: Not Found - The namespace specified does not exist
+ content:
+ application/json:
+ schema:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/schemas/IcebergErrorResponse'
+ examples:
+ NamespaceNotFound:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/examples/NoSuchNamespaceError'
+ 409:
+ description: Conflict - The directory already exists under the given namespace
+ content:
+ application/json:
+ schema:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/schemas/IcebergErrorResponse'
+ examples:
+ TableAlreadyExists:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/examples/TableAlreadyExistsError'
+ 503:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/responses/ServiceUnavailableResponse'
+ 5XX:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/responses/ServerErrorResponse'
+
+ /polaris/v1/{prefix}/namespaces/{namespace}/directories/{directory}:
+ parameters:
+ - $ref: '../iceberg-rest-catalog-open-api.yaml#/components/parameters/prefix'
+ - $ref: '../iceberg-rest-catalog-open-api.yaml#/components/parameters/namespace'
+ - $ref: '#/components/parameters/directory'
+
+ get:
+ tags:
+ - Directory API
+ summary: Load a directory under the given namespace from the catalog
+ operationId: loadDirectory
+ description:
+ Load a directory from the catalog under the given namespace.
+
+ The response contains all directory information passed during create.
+
+ parameters:
+ - $ref: '#/components/parameters/polaris-directory-data-access'
+ responses:
+ 200:
+ $ref: '#/components/responses/LoadDirectoryResponse'
+ 400:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/responses/BadRequestErrorResponse'
+ 401:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/responses/UnauthorizedResponse'
+ 403:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/responses/ForbiddenResponse'
+ 404:
+ description:
+ Not Found - NoSuchTableError, directory to load does not exist
+ content:
+ application/json:
+ schema:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/schemas/IcebergErrorResponse'
+ examples:
+ DirectoryToLoadDoesNotExist:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/examples/NoSuchTableError'
+ 503:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/responses/ServiceUnavailableResponse'
+ 5XX:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/responses/ServerErrorResponse'
+
+ delete:
+ tags:
+ - Directory API
+ summary: Drop a directory under the given namespace from the catalog
+ operationId: dropDirectory
+ description: Remove a directory under the given namespace from the catalog
+ responses:
+ 204:
+ description: Success, no content
+ 400:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/responses/BadRequestErrorResponse'
+ 401:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/responses/UnauthorizedResponse'
+ 403:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/responses/ForbiddenResponse'
+ 404:
+ description:
+ Not Found - NoSuchTableError, directory to drop does not exist
+ content:
+ application/json:
+ schema:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/schemas/IcebergErrorResponse'
+ examples:
+ DirectoryToDeleteDoesNotExist:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/examples/NoSuchTableError'
+ 503:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/responses/ServiceUnavailableResponse'
+ 5XX:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/responses/ServerErrorResponse'
+
+components:
+ parameters:
+ directory:
+ name: directory
+ in: path
+ description: A directory name
+ required: true
+ schema:
+ type: string
+ example: "product-images"
+
+ polaris-directory-data-access:
+ name: Polaris-Directory-Access-Delegation
+ in: header
+ description: >
+ Optional header indicating that the client supports delegated access
+ using a specific mechanism. If the header is present, the server must
+ either provide access using the requested mechanism or fail the request.
+ If `vended-credentials` is specified, the server must return scoped
+ storage credentials in the `storage-access-configs` field of the response.
+ required: false
+ schema:
+ type: string
+ enum:
+ - vended-credentials
+ style: simple
+ explode: false
+ example: "vended-credentials"
+
+ schemas:
+ DirectoryFilter:
+ type: object
+ description: >
+ Controls which objects are included in the directory table during a scan.
+ If no filter is set, all objects found under the directory base_location are included.
+ properties:
+ include:
+ type: array
+ items:
+ type: string
+ description: A list of regular expressions matched against the object URI. An object is included if it matches at least one pattern (or if `include` is omitted).
+ exclude:
+ type: array
+ items:
+ type: string
+ description: A list of regular expressions matched against the object URI. An object matching any pattern is excluded, even if it also matches an include pattern.
+
+ ScanSchedule:
+ type: object
+ description: >
+ Describes when an external scanning service should re-scan the directory.
+ properties:
+ cron:
+ type: string
+ description: Cron expression describing when scans should run (e.g. `0 * * * *` for the top of every hour).
+
+ CreateDirectoryRequest:
+ type: object
+ required:
+ - name
+ - base_location
+ properties:
+ name:
+ type: string
+ description: The name of the directory. It is used in REST endpoint paths and becomes the name of the corresponding Iceberg table.
+ base_location:
+ type: string
+ description: The base location to use as a root for scanning objects in the object store (e.g. s3://bucket/path/).
+ filter:
+ $ref: '#/components/schemas/DirectoryFilter'
+ scan-schedule:
+ $ref: '#/components/schemas/ScanSchedule'
+ properties:
+ type: object
+ additionalProperties:
+ type: string
+
+ Directory:
+ type: object
+ description: >
+ Directory information.
+
+ - `name` (REQUIRED): name of the directory
+
+ - `base_location` (REQUIRED): root location of the directory on the object store
+
+ - `filter` (OPTIONAL): include and exclude patterns for object filtering
+
+ - `scan-schedule` (OPTIONAL): schedule describing when an external service should re-scan the directory
+
+ - `properties` (OPTIONAL): properties for the directory passed on creation
+ required:
+ - name
+ - base_location
+ properties:
+ name:
+ type: string
+ base_location:
+ type: string
+ filter:
+ $ref: '#/components/schemas/DirectoryFilter'
+ scan-schedule:
+ $ref: '#/components/schemas/ScanSchedule'
+ properties:
+ type: object
+ additionalProperties:
+ type: string
+
+ LoadDirectoryResponse:
+ description: Result used when a directory is successfully loaded.
+ type: object
+ required:
+ - directory
+ properties:
+ directory:
+ $ref: '#/components/schemas/Directory'
+ storage-access-configs:
+ type: array
+ description: |
+ Storage configurations required to access the directory, encompassing both credentials and other necessary
+ settings for storage access, such as the client region for S3.
+ items:
+ $ref: './generic-tables-api.yaml#/components/schemas/StorageAccessConfig'
Review Comment:
Would be worthwhile extracting this schema to a common set of schemas for various APIs, rather than referencing the generic-tables API spec directly?
##########
spec/polaris-catalog-apis/directories-api.yaml:
##########
@@ -0,0 +1,334 @@
+#
+# 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.
+#
+
+---
+paths:
+ /polaris/v1/{prefix}/namespaces/{namespace}/directories:
+ parameters:
+ - $ref: '../iceberg-rest-catalog-open-api.yaml#/components/parameters/prefix'
+ - $ref: '../iceberg-rest-catalog-open-api.yaml#/components/parameters/namespace'
+
+ get:
+ tags:
+ - Directory API
+ summary: List all directory identifiers underneath a given namespace
+ description: Return all directory identifiers under this namespace
+ operationId: listDirectories
+ parameters:
+ - $ref: '../iceberg-rest-catalog-open-api.yaml#/components/parameters/page-token'
+ - $ref: '../iceberg-rest-catalog-open-api.yaml#/components/parameters/page-size'
+ responses:
+ 200:
+ $ref: '#/components/responses/ListDirectoriesResponse'
+ 400:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/responses/BadRequestErrorResponse'
+ 401:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/responses/UnauthorizedResponse'
+ 403:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/responses/ForbiddenResponse'
+ 404:
+ description: Not Found - The namespace specified does not exist
+ content:
+ application/json:
+ schema:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/schemas/IcebergErrorResponse'
+ examples:
+ NamespaceNotFound:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/examples/NoSuchNamespaceError'
+ 503:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/responses/ServiceUnavailableResponse'
+ 5XX:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/responses/ServerErrorResponse'
+
+ post:
+ tags:
+ - Directory API
+ summary: Create a directory under the given namespace
+ description:
+ Create a directory under the given namespace, and return the created directory information as a response.
+ operationId: createDirectory
+ parameters:
+ - $ref: '#/components/parameters/polaris-directory-data-access'
+ requestBody:
+ required: true
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/CreateDirectoryRequest'
+ responses:
+ 200:
+ $ref: '#/components/responses/CreateDirectoryResponse'
+ 400:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/responses/BadRequestErrorResponse'
+ 401:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/responses/UnauthorizedResponse'
+ 403:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/responses/ForbiddenResponse'
+ 404:
+ description: Not Found - The namespace specified does not exist
+ content:
+ application/json:
+ schema:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/schemas/IcebergErrorResponse'
+ examples:
+ NamespaceNotFound:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/examples/NoSuchNamespaceError'
+ 409:
+ description: Conflict - The directory already exists under the given namespace
+ content:
+ application/json:
+ schema:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/schemas/IcebergErrorResponse'
+ examples:
+ TableAlreadyExists:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/examples/TableAlreadyExistsError'
+ 503:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/responses/ServiceUnavailableResponse'
+ 5XX:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/responses/ServerErrorResponse'
+
+ /polaris/v1/{prefix}/namespaces/{namespace}/directories/{directory}:
+ parameters:
+ - $ref: '../iceberg-rest-catalog-open-api.yaml#/components/parameters/prefix'
+ - $ref: '../iceberg-rest-catalog-open-api.yaml#/components/parameters/namespace'
+ - $ref: '#/components/parameters/directory'
+
+ get:
+ tags:
+ - Directory API
+ summary: Load a directory under the given namespace from the catalog
+ operationId: loadDirectory
+ description:
+ Load a directory from the catalog under the given namespace.
+
+ The response contains all directory information passed during create.
+
+ parameters:
+ - $ref: '#/components/parameters/polaris-directory-data-access'
+ responses:
+ 200:
+ $ref: '#/components/responses/LoadDirectoryResponse'
+ 400:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/responses/BadRequestErrorResponse'
+ 401:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/responses/UnauthorizedResponse'
+ 403:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/responses/ForbiddenResponse'
+ 404:
+ description:
+ Not Found - NoSuchTableError, directory to load does not exist
+ content:
+ application/json:
+ schema:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/schemas/IcebergErrorResponse'
+ examples:
+ DirectoryToLoadDoesNotExist:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/examples/NoSuchTableError'
+ 503:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/responses/ServiceUnavailableResponse'
+ 5XX:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/responses/ServerErrorResponse'
+
+ delete:
+ tags:
+ - Directory API
+ summary: Drop a directory under the given namespace from the catalog
+ operationId: dropDirectory
+ description: Remove a directory under the given namespace from the catalog
+ responses:
+ 204:
+ description: Success, no content
+ 400:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/responses/BadRequestErrorResponse'
+ 401:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/responses/UnauthorizedResponse'
+ 403:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/responses/ForbiddenResponse'
+ 404:
+ description:
+ Not Found - NoSuchTableError, directory to drop does not exist
+ content:
+ application/json:
+ schema:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/schemas/IcebergErrorResponse'
+ examples:
+ DirectoryToDeleteDoesNotExist:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/examples/NoSuchTableError'
+ 503:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/responses/ServiceUnavailableResponse'
+ 5XX:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/responses/ServerErrorResponse'
+
+components:
+ parameters:
+ directory:
+ name: directory
+ in: path
+ description: A directory name
+ required: true
+ schema:
+ type: string
+ example: "product-images"
+
+ polaris-directory-data-access:
+ name: Polaris-Directory-Access-Delegation
+ in: header
+ description: >
+ Optional header indicating that the client supports delegated access
+ using a specific mechanism. If the header is present, the server must
+ either provide access using the requested mechanism or fail the request.
+ If `vended-credentials` is specified, the server must return scoped
+ storage credentials in the `storage-access-configs` field of the response.
+ required: false
+ schema:
+ type: string
+ enum:
+ - vended-credentials
+ style: simple
+ explode: false
+ example: "vended-credentials"
Review Comment:
```suggestion
schema:
type: array
items:
type: string
enum:
- vended-credentials
example: "vended-credentials"
```
##########
spec/polaris-catalog-apis/directories-api.yaml:
##########
@@ -0,0 +1,334 @@
+#
+# 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.
+#
+
+---
+paths:
+ /polaris/v1/{prefix}/namespaces/{namespace}/directories:
+ parameters:
+ - $ref: '../iceberg-rest-catalog-open-api.yaml#/components/parameters/prefix'
+ - $ref: '../iceberg-rest-catalog-open-api.yaml#/components/parameters/namespace'
+
+ get:
+ tags:
+ - Directory API
+ summary: List all directory identifiers underneath a given namespace
+ description: Return all directory identifiers under this namespace
+ operationId: listDirectories
+ parameters:
+ - $ref: '../iceberg-rest-catalog-open-api.yaml#/components/parameters/page-token'
+ - $ref: '../iceberg-rest-catalog-open-api.yaml#/components/parameters/page-size'
+ responses:
+ 200:
+ $ref: '#/components/responses/ListDirectoriesResponse'
+ 400:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/responses/BadRequestErrorResponse'
+ 401:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/responses/UnauthorizedResponse'
+ 403:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/responses/ForbiddenResponse'
+ 404:
+ description: Not Found - The namespace specified does not exist
+ content:
+ application/json:
+ schema:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/schemas/IcebergErrorResponse'
+ examples:
+ NamespaceNotFound:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/examples/NoSuchNamespaceError'
+ 503:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/responses/ServiceUnavailableResponse'
+ 5XX:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/responses/ServerErrorResponse'
+
+ post:
+ tags:
+ - Directory API
+ summary: Create a directory under the given namespace
+ description:
+ Create a directory under the given namespace, and return the created directory information as a response.
+ operationId: createDirectory
+ parameters:
+ - $ref: '#/components/parameters/polaris-directory-data-access'
+ requestBody:
+ required: true
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/CreateDirectoryRequest'
+ responses:
+ 200:
+ $ref: '#/components/responses/CreateDirectoryResponse'
+ 400:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/responses/BadRequestErrorResponse'
+ 401:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/responses/UnauthorizedResponse'
+ 403:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/responses/ForbiddenResponse'
+ 404:
+ description: Not Found - The namespace specified does not exist
+ content:
+ application/json:
+ schema:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/schemas/IcebergErrorResponse'
+ examples:
+ NamespaceNotFound:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/examples/NoSuchNamespaceError'
+ 409:
+ description: Conflict - The directory already exists under the given namespace
+ content:
+ application/json:
+ schema:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/schemas/IcebergErrorResponse'
+ examples:
+ TableAlreadyExists:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/examples/TableAlreadyExistsError'
+ 503:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/responses/ServiceUnavailableResponse'
+ 5XX:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/responses/ServerErrorResponse'
+
+ /polaris/v1/{prefix}/namespaces/{namespace}/directories/{directory}:
+ parameters:
+ - $ref: '../iceberg-rest-catalog-open-api.yaml#/components/parameters/prefix'
+ - $ref: '../iceberg-rest-catalog-open-api.yaml#/components/parameters/namespace'
+ - $ref: '#/components/parameters/directory'
+
+ get:
+ tags:
+ - Directory API
+ summary: Load a directory under the given namespace from the catalog
+ operationId: loadDirectory
+ description:
+ Load a directory from the catalog under the given namespace.
+
+ The response contains all directory information passed during create.
+
+ parameters:
+ - $ref: '#/components/parameters/polaris-directory-data-access'
+ responses:
+ 200:
+ $ref: '#/components/responses/LoadDirectoryResponse'
+ 400:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/responses/BadRequestErrorResponse'
+ 401:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/responses/UnauthorizedResponse'
+ 403:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/responses/ForbiddenResponse'
+ 404:
+ description:
+ Not Found - NoSuchTableError, directory to load does not exist
+ content:
+ application/json:
+ schema:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/schemas/IcebergErrorResponse'
+ examples:
+ DirectoryToLoadDoesNotExist:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/examples/NoSuchTableError'
+ 503:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/responses/ServiceUnavailableResponse'
+ 5XX:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/responses/ServerErrorResponse'
+
+ delete:
+ tags:
+ - Directory API
+ summary: Drop a directory under the given namespace from the catalog
+ operationId: dropDirectory
+ description: Remove a directory under the given namespace from the catalog
+ responses:
+ 204:
+ description: Success, no content
+ 400:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/responses/BadRequestErrorResponse'
+ 401:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/responses/UnauthorizedResponse'
+ 403:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/responses/ForbiddenResponse'
+ 404:
+ description:
+ Not Found - NoSuchTableError, directory to drop does not exist
+ content:
+ application/json:
+ schema:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/schemas/IcebergErrorResponse'
+ examples:
+ DirectoryToDeleteDoesNotExist:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/examples/NoSuchTableError'
+ 503:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/responses/ServiceUnavailableResponse'
+ 5XX:
+ $ref: '../iceberg-rest-catalog-open-api.yaml#/components/responses/ServerErrorResponse'
+
+components:
+ parameters:
+ directory:
+ name: directory
+ in: path
+ description: A directory name
+ required: true
+ schema:
+ type: string
+ example: "product-images"
+
+ polaris-directory-data-access:
+ name: Polaris-Directory-Access-Delegation
+ in: header
+ description: >
+ Optional header indicating that the client supports delegated access
+ using a specific mechanism. If the header is present, the server must
+ either provide access using the requested mechanism or fail the request.
+ If `vended-credentials` is specified, the server must return scoped
+ storage credentials in the `storage-access-configs` field of the response.
+ required: false
+ schema:
+ type: string
+ enum:
+ - vended-credentials
+ style: simple
+ explode: false
+ example: "vended-credentials"
+
+ schemas:
+ DirectoryFilter:
+ type: object
+ description: >
+ Controls which objects are included in the directory table during a scan.
+ If no filter is set, all objects found under the directory base_location are included.
+ properties:
+ include:
+ type: array
+ items:
+ type: string
+ description: A list of regular expressions matched against the object URI. An object is included if it matches at least one pattern (or if `include` is omitted).
+ exclude:
+ type: array
+ items:
+ type: string
+ description: A list of regular expressions matched against the object URI. An object matching any pattern is excluded, even if it also matches an include pattern.
+
+ ScanSchedule:
+ type: object
+ description: >
+ Describes when an external scanning service should re-scan the directory.
+ properties:
+ cron:
+ type: string
+ description: Cron expression describing when scans should run (e.g. `0 * * * *` for the top of every hour).
+
+ CreateDirectoryRequest:
+ type: object
+ required:
+ - name
+ - base_location
+ properties:
+ name:
+ type: string
+ description: The name of the directory. It is used in REST endpoint paths and becomes the name of the corresponding Iceberg table.
+ base_location:
+ type: string
+ description: The base location to use as a root for scanning objects in the object store (e.g. s3://bucket/path/).
Review Comment:
Just to clarify: is this the external directory base location (where the items to scan live), or the Iceberg inventory table base location? Is it even possible to provide the inventory table location, or is it always assigned by Polaris?
##########
site/content/in-dev/unreleased/configuration/config-sections/flags-polaris_features.md:
##########
@@ -317,6 +317,15 @@ When enabled, allows RBAC operations to create synthetic entities for entities i
---
+##### `polaris.features."ENABLE_VOLUMES"`
Review Comment:
The name is out of date, it seems.
--
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 Directories feature for managing objects (including unstructured data) on object stores [polaris]
Posted by "jbonofre (via GitHub)" <gi...@apache.org>.
jbonofre commented on code in PR #4613:
URL: https://github.com/apache/polaris/pull/4613#discussion_r4156689500
##########
site/content/in-dev/unreleased/directories.md:
##########
@@ -0,0 +1,227 @@
+---
+#
+# 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.
+#
+title: Directories
+type: docs
+weight: 450
+---
+
+## Overview
+
+Directories make objects (including unstructured data like images, videos, documents, and other objects) discoverable alongside
+structured Iceberg tables within a Polaris catalog.
+A directory points to a base location/prefix on an object store and automatically tracks the objects it contains by maintaining
+an Iceberg table with object-level metadata such as URI, size, content type, checksum, ...
+
+This means query engines and tools that already know how to read Iceberg tables can discover and
+access unstructured data with little or no extra work (accessing the object itself).
+
+## Concepts
+
+A directory has two main parts:
+
+1. **Directory configuration** โ stored by the Polaris server. It describes _where_ the data lives,
+ how to authenticate, which objects to include, and how often to re-scan. The configuration "lives" in a namespace.
+2. **Directory table** โ an Iceberg table serving as the inventory of all objects contained in the directory, one row per object discovered during a scan.
+ The directory table uses the configuration name.
+
+The Polaris server itself does not perform scans. Instead, external services (e.g. directory table scanning service) read the directory configuration through the REST API,
+walk the object store, and write the results into the directory table.
+
+## Configuration
+
+A directory is described by the following fields:
+
+| Field | Type | Required | Description |
+|-------|------|----------|-------------|
+| `name` | `string` | Yes | The name of the directory. It is used in REST endpoint paths and becomes the name of the corresponding Iceberg table. |
+| `base_location` | `string` | Yes | The base location to use as a root for scanning objects in the object store (for example `s3://my-bucket/images/`, `gs://my-bucket/docs/`, or `file:///data/local/`). |
+| `filter` | `object` | No | Include and exclude patterns that control which objects are added to the directory table during a scan. See [Filter](#filter). |
+| `scan-schedule` | `object` | No | Object representing a scan schedule (trigger, cron, ...). |
+
+### Storage access
+
+Directories use Polaris `StorageAccessConfig` to access the object store.
+
+### Filter
+
+The `filter` object controls which objects are included in the directory table during a scan.
+
+If no filter is set, all objects found under the directory `base_location` are included.
+
+`include` and `exclude` are lists of regular expressions matched against the object's full URI. An
+object is included if it matches at least one `include` pattern (or `include` is omitted) and does
+not match any `exclude` pattern.
+
+**Example** โ include only JPEG and PNG images, but exclude thumbnails:
+
+```json
+{
+ "filter": {
+ "include": [".*\\.jpg$", ".*\\.png$"],
+ "exclude": [".*thumbs/.*"]
+ }
+}
+```
+
+### Example: creating a directory
+
+```json
+POST /v1/{prefix}/namespaces/{namespace}/directories
+
+{
+ "name": "product-images",
+ "base_location": "s3://warehouse/product-images/",
+ "filter": {
+ "include": [".*\\.jpg$", ".*\\.png$"]
+ },
+ "scan-schedule": {
+ "cron": "0 * * * *"
+ }
+}
+```
+
+## REST endpoints
+
+Directories are managed through the following REST endpoints. All paths are relative to the catalog
+base URL.
+
+### List directories
+
+**GET** `/v1/{prefix}/namespaces/{namespace}/directories`
+
+Returns the list of directory identifiers in the given namespace.
+
+### Create a directory
+
+**POST** `/v1/{prefix}/namespaces/{namespace}/directories`
+
+Creates a new directory and its corresponding Iceberg table. The request body must contain the
+directory configuration (see [Configuration](#configuration)).
+
+### Get directory details
+
+**GET** `/v1/{prefix}/namespaces/{namespace}/directories/{directory}`
+
+Returns the full configuration of the specified directory, including filter and scan schedule.
+
+### Drop a directory
+
+**DELETE** `/v1/{prefix}/namespaces/{namespace}/directories/{directory}`
+
+Removes the directory and its associated Iceberg table from the namespace.
+
+## Directory table
+
+When a directory is created, Polaris creates an Iceberg table in the same namespace using the directory
+`name` as the table name. The table uses the following schema:
+
+| Field Id | Field Name | Type | Required | Description |
+|----------|------------|------|----------|-------------|
+| 1 | `file_uri` | `string` | Yes | The fully qualified URI of the object (for example `s3://my-bucket/images/photo.jpg`). |
+| 2 | `content_type` | `string` | No | The MIME content type (RFC 2045) of the object (for example `image/jpeg`, `application/pdf`). |
+| 3 | `size` | `long` | No | The size of the object in bytes. |
+| 4 | `checksum_algorithm` | `string` | No | The name of the checksum algorithm (for example `MD5`, `SHA-256`, `CRC32`). |
+| 5 | `checksum` | `string` | No | The object checksum value computed with the algorithm specified in `checksum_algorithm`. |
+| 6 | `last_modified` | `timestamptz` | No | The last modification timestamp of the object as reported by the object store. |
+| 7 | `metadata` | `map<string, string>` | No | Additional labels and tags from the object store (for example S3 user-defined metadata, storage class, or content encoding). |
+
+Clients (query engines, analytics tools, etc.) interact with the directory table as a regular Iceberg
+table. For instance, you can query it with Spark SQL:
+
+```sql
+SELECT file_uri, size, last_modified
+FROM my_catalog.my_namespace.product_images
+WHERE content_type = 'image/jpeg'
+ AND size > 1048576
+ORDER BY last_modified DESC;
+```
+
+### Credential vending
+
+The directory table might contain additional fields for credential vending.
Review Comment:
I added a note related to credential vending in the description for now.
--
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 Directories feature for managing objects (including unstructured data) on object stores [polaris]
Posted by "orca-review-bot (via GitHub)" <gi...@apache.org>.
orca-review-bot commented on code in PR #4613:
URL: https://github.com/apache/polaris/pull/4613#discussion_r4392456848
##########
runtime/service/src/main/java/org/apache/polaris/service/catalog/directory/DirectoryCatalogAdapter.java:
##########
@@ -0,0 +1,166 @@
+/*
+ * 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,
+ RealmContext realmContext,
+ SecurityContext securityContext) {
+ DirectoryCatalogHandler handler = newHandler(securityContext, prefix);
+
+ DirectoryFilter filter = createDirectoryRequest.getFilter();
+ String filterInclude =
+ filter != null ? DirectoryCatalogHandler.toJsonArray(filter.getInclude()) : null;
+ String filterExclude =
+ filter != null ? DirectoryCatalogHandler.toJsonArray(filter.getExclude()) : null;
+
+ LoadDirectoryResponse response =
+ handler.createDirectory(
+ TableIdentifier.of(
+ NamespaceUtils.splitNamespace(
+ namespace, NamespaceUtils.DEFAULT_NAMESPACE_SEPARATOR),
+ createDirectoryRequest.getName()),
+ createDirectoryRequest.getBaseLocation(),
+ filterInclude,
+ filterExclude,
+ DirectoryCatalogHandler.scanScheduleToJson(createDirectoryRequest.getScanSchedule()),
+ reservedProperties.removeReservedProperties(createDirectoryRequest.getProperties()));
+
+ return Response.ok(response).build();
+ }
+
+ @Override
+ public Response dropDirectory(
+ String prefix,
+ String namespace,
+ String directory,
+ RealmContext realmContext,
+ SecurityContext securityContext) {
+ DirectoryCatalogHandler handler = newHandler(securityContext, prefix);
+ handler.dropDirectory(
+ TableIdentifier.of(
+ NamespaceUtils.splitNamespace(namespace, NamespaceUtils.DEFAULT_NAMESPACE_SEPARATOR),
+ directory));
+ return Response.noContent().build();
+ }
+
+ @Override
+ public Response scanDirectory(
+ String prefix,
+ String namespace,
+ String directory,
+ RealmContext realmContext,
+ SecurityContext securityContext) {
+ DirectoryCatalogHandler handler = newHandler(securityContext, prefix);
+ ScanDirectoryResponse response =
+ handler.scanDirectory(
+ TableIdentifier.of(
+ NamespaceUtils.splitNamespace(
+ namespace, NamespaceUtils.DEFAULT_NAMESPACE_SEPARATOR),
+ directory));
+ return Response.ok(response).build();
+ }
+
+ @Override
+ public Response listDirectories(
+ String prefix,
+ String namespace,
+ String pageToken,
Review Comment:
`pageToken`/`pageSize` are accepted here but never passed down, and `PolarisDirectoryCatalog` always uses `PageToken.readEverything()`. So a namespace with many directories returns one unbounded page. Can we follow `GenericTableCatalogHandler.listGenericTables` (PageToken.build + rejectIncompleteListing + setNextPageToken)?
<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,408 @@
+/*
+ * 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.Optional;
+import java.util.Set;
+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.AlreadyExistsException;
+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.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.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();
+ }
+
+ 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) {
+ PolarisAuthorizableOperation op = PolarisAuthorizableOperation.LIST_TABLES;
+ authorizeBasicNamespaceOperationOrThrow(op, parent);
+
+ return ListDirectoriesResponse.builder()
+ .setIdentifiers(new LinkedHashSet<>(directoryCatalog.listDirectories(parent)))
+ .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(
Review Comment:
Unlike table create, this doesn't run the location-overlap check (`LocalIcebergCatalog.validateNoLocationOverlap`, gated by `ALLOW_TABLE_LOCATION_OVERLAP`). So a directory can point at a sibling table's location, and a scan writes that table's file names and sizes into an inventory the caller can read. Is that intended, or should directories get the same overlap check?
<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/DirectoryScanner.java:
##########
@@ -0,0 +1,147 @@
+/*
+ * 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 java.io.IOException;
+import java.io.UncheckedIOException;
+import java.net.URLConnection;
+import java.time.Instant;
+import java.time.OffsetDateTime;
+import java.time.ZoneOffset;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.UUID;
+import java.util.regex.Pattern;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.OverwriteFiles;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.data.GenericRecord;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.data.parquet.GenericParquetWriter;
+import org.apache.iceberg.exceptions.CommitStateUnknownException;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.io.DataWriter;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.FileInfo;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.io.SupportsPrefixOperations;
+import org.apache.iceberg.parquet.Parquet;
+import org.apache.polaris.service.catalog.io.ExceptionMappingFileIO;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Lists the objects under a directory base location and replaces the content of the directory
+ * inventory table (see {@link DirectoryCatalogHandler#DIRECTORY_TABLE_SCHEMA}) with them, in a
+ * single commit.
+ */
+final class DirectoryScanner {
+ private static final Logger LOGGER = LoggerFactory.getLogger(DirectoryScanner.class);
+
+ private DirectoryScanner() {}
+
+ /**
+ * @param sourceIO the {@link FileIO} used to list the objects, it must support prefix operations
+ * @return the number of objects recorded in the table
+ */
+ static long scan(
+ Table table,
+ FileIO sourceIO,
+ String baseLocation,
+ List<String> include,
+ List<String> exclude) {
+ // The FileIO factory wraps the FileIO, and the wrapper does not expose the prefix operations
+ FileIO io = sourceIO instanceof ExceptionMappingFileIO w ? w.getInnerIo() : sourceIO;
+ if (!(io instanceof SupportsPrefixOperations prefixIO)) {
+ throw new UnsupportedOperationException(
+ "Scanning is not supported for location " + baseLocation);
+ }
+ List<Pattern> includePatterns = compile(include);
+ List<Pattern> excludePatterns = compile(exclude);
+
+ List<Record> records = new ArrayList<>();
+ for (FileInfo file : prefixIO.listPrefix(baseLocation)) {
Review Comment:
This lists everything under the base location, including the inventory table's own files when they live under it. That happens with `allow.unstructured.table.location=true` and a base location above the namespace location. Old data files stay on storage after an overwrite, so each rescan also indexes the Parquet and metadata files from earlier scans. Can we skip `table.location()` here, or reject that layout at create time?
<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,408 @@
+/*
+ * 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.Optional;
+import java.util.Set;
+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.AlreadyExistsException;
+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.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.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();
+ }
+
+ 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) {
+ PolarisAuthorizableOperation op = PolarisAuthorizableOperation.LIST_TABLES;
+ authorizeBasicNamespaceOperationOrThrow(op, parent);
+
+ return ListDirectoriesResponse.builder()
+ .setIdentifiers(new LinkedHashSet<>(directoryCatalog.listDirectories(parent)))
+ .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);
+
+ // Check that the inventory table name does not collide with an existing entity
+ TableIdentifier tableId = inventoryTableIdentifier(identifier);
+ if (this.icebergCatalog.tableExists(tableId)) {
+ throw new AlreadyExistsException(
+ "Cannot create directory %s: associated table %s already exists", identifier, tableId);
+ }
+
+ 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) {
+ // Roll back the directory entity if table creation fails
+ LOGGER.warn(
+ "Failed to create Iceberg table {} for directory {}, rolling back",
+ tableId,
+ identifier,
+ e);
+ try {
+ this.directoryCatalog.dropDirectory(identifier);
+ } catch (RuntimeException rollbackFailure) {
+ e.addSuppressed(rollbackFailure);
+ }
+ throw e;
+ }
+
+ Directory createdDirectory =
+ Directory.builder()
+ .setName(createdEntity.getName())
+ .setBaseLocation(createdEntity.getBaseLocation())
+ .setFilter(toDirectoryFilter(createdEntity))
+ .setScanSchedule(scanScheduleFromJson(createdEntity.getScanSchedule()))
+ .setProperties(createdEntity.getPropertiesAsMap())
+ .build();
+
+ return LoadDirectoryResponse.builder().setDirectory(createdDirectory).build();
+ }
+
+ public boolean dropDirectory(TableIdentifier identifier) {
+ this.inventoryTable = inventoryTableIdentifier(identifier);
+ PolarisAuthorizableOperation op = PolarisAuthorizableOperation.DROP_TABLE_WITHOUT_PURGE;
+ authorizeCreateTableLikeUnderNamespaceOperationOrThrow(op, identifier);
+
+ // Verify the directory exists before attempting to drop the associated table
+ this.directoryCatalog.loadDirectory(identifier);
+
+ // Drop the associated inventory table first โ if this fails the directory entity is still
+ // intact
+ // and the operation can be retried.
+ TableIdentifier tableId = inventoryTableIdentifier(identifier);
+ try {
+ this.icebergCatalog.dropTable(tableId, false);
+ LOGGER.debug("Dropped associated Iceberg table {} for directory {}", tableId, identifier);
+ } catch (Exception e) {
+ LOGGER.error("Failed to drop associated Iceberg table for directory {}", identifier, e);
Review Comment:
Log-then-rethrow: the exception gets logged here and again by the framework. Can we drop the try/catch?
<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,408 @@
+/*
+ * 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.Optional;
+import java.util.Set;
+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.AlreadyExistsException;
+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.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.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();
+ }
+
+ 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) {
+ PolarisAuthorizableOperation op = PolarisAuthorizableOperation.LIST_TABLES;
+ authorizeBasicNamespaceOperationOrThrow(op, parent);
+
+ return ListDirectoriesResponse.builder()
+ .setIdentifiers(new LinkedHashSet<>(directoryCatalog.listDirectories(parent)))
+ .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);
+
+ // Check that the inventory table name does not collide with an existing entity
+ TableIdentifier tableId = inventoryTableIdentifier(identifier);
+ if (this.icebergCatalog.tableExists(tableId)) {
Review Comment:
This pre-check races with a concurrent `createTable` of `<dir>__inventory`. The rollback below already handles the case where `createTable` fails. Can we drop the pre-check and rely on the `AlreadyExistsException` from `createTable`?
<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/DirectoryScanner.java:
##########
@@ -0,0 +1,147 @@
+/*
+ * 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 java.io.IOException;
+import java.io.UncheckedIOException;
+import java.net.URLConnection;
+import java.time.Instant;
+import java.time.OffsetDateTime;
+import java.time.ZoneOffset;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.UUID;
+import java.util.regex.Pattern;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.OverwriteFiles;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.data.GenericRecord;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.data.parquet.GenericParquetWriter;
+import org.apache.iceberg.exceptions.CommitStateUnknownException;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.io.DataWriter;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.FileInfo;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.io.SupportsPrefixOperations;
+import org.apache.iceberg.parquet.Parquet;
+import org.apache.polaris.service.catalog.io.ExceptionMappingFileIO;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Lists the objects under a directory base location and replaces the content of the directory
+ * inventory table (see {@link DirectoryCatalogHandler#DIRECTORY_TABLE_SCHEMA}) with them, in a
+ * single commit.
+ */
+final class DirectoryScanner {
+ private static final Logger LOGGER = LoggerFactory.getLogger(DirectoryScanner.class);
+
+ private DirectoryScanner() {}
+
+ /**
+ * @param sourceIO the {@link FileIO} used to list the objects, it must support prefix operations
+ * @return the number of objects recorded in the table
+ */
+ static long scan(
+ Table table,
+ FileIO sourceIO,
+ String baseLocation,
+ List<String> include,
+ List<String> exclude) {
+ // The FileIO factory wraps the FileIO, and the wrapper does not expose the prefix operations
+ FileIO io = sourceIO instanceof ExceptionMappingFileIO w ? w.getInnerIo() : sourceIO;
+ if (!(io instanceof SupportsPrefixOperations prefixIO)) {
+ throw new UnsupportedOperationException(
+ "Scanning is not supported for location " + baseLocation);
+ }
+ List<Pattern> includePatterns = compile(include);
+ List<Pattern> excludePatterns = compile(exclude);
+
+ List<Record> records = new ArrayList<>();
+ for (FileInfo file : prefixIO.listPrefix(baseLocation)) {
+ String uri = file.location();
+ if (matches(uri, includePatterns, excludePatterns)) {
+ Record record = GenericRecord.create(table.schema());
+ record.setField("file_uri", uri);
+ record.setField("content_type", URLConnection.guessContentTypeFromName(uri));
+ record.setField("size", file.size());
+ record.setField(
+ "last_modified",
+ OffsetDateTime.ofInstant(Instant.ofEpochMilli(file.createdAtMillis()), ZoneOffset.UTC));
+ records.add(record);
+ }
+ }
+
+ OverwriteFiles overwrite = table.newOverwrite().overwriteByRowFilter(Expressions.alwaysTrue());
+ String dataFileLocation = null;
+ try {
+ if (!records.isEmpty()) {
+ dataFileLocation = table.locationProvider().newDataLocation(UUID.randomUUID() + ".parquet");
+ overwrite.addFile(write(table, dataFileLocation, records));
+ }
+ overwrite.commit();
+ } catch (CommitStateUnknownException e) {
+ // The commit may have landed: deleting the file could leave the table pointing to nothing
+ throw e;
+ } catch (RuntimeException e) {
+ if (dataFileLocation != null) {
+ deleteQuietly(table, dataFileLocation);
+ }
+ throw e;
+ }
+ LOGGER.debug("Scanned {} objects under {}", records.size(), baseLocation);
+ return records.size();
+ }
+
+ private static DataFile write(Table table, String location, List<Record> records) {
+ OutputFile output = table.io().newOutputFile(location);
+ try (DataWriter<Record> writer =
+ Parquet.writeData(output)
+ .schema(table.schema())
+ .withSpec(table.spec())
+ .createWriterFunc(GenericParquetWriter::create)
+ .overwrite()
+ .build()) {
+ records.forEach(writer::write);
+ writer.close();
+ return writer.toDataFile();
+ } catch (IOException e) {
+ throw new UncheckedIOException("Failed to write the directory inventory file", e);
+ }
+ }
+
+ private static void deleteQuietly(Table table, String location) {
+ try {
+ table.io().deleteFile(location);
+ } catch (RuntimeException e) {
+ LOGGER.warn("Failed to delete the orphan inventory file {}", location, e);
+ }
+ }
+
+ private static List<Pattern> compile(List<String> regexes) {
Review Comment:
Filters are only compiled at scan time, so a bad regex is accepted on create and only fails on the first scan. Can we compile them in `createDirectory` and reject bad ones there? Capping pattern count and length would also limit backtracking cost.
<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/catalog/FederatedCatalogFactory.java:
##########
@@ -63,4 +63,20 @@ GenericTableCatalog createGenericCatalog(
ConnectionConfigInfoDpo connectionConfig,
PolarisCredentialManager polarisCredentialManager,
Map<String, String> catalogProperties);
+
+ /**
+ * Creates a directory catalog for the given connection configuration.
+ *
+ * @param connectionConfig the connection configuration
+ * @param polarisCredentialManager the credential manager for generating connection credentials
+ * that Polaris uses to access external systems
+ * @param catalogProperties additional properties from the ExternalCatalog entity that should be
+ * passed through to the underlying catalog
+ * @return the initialized catalog
+ * @throws IllegalStateException if the connection configuration is invalid
+ */
+ DirectoryCatalog createDirectoryCatalog(
Review Comment:
Adding an abstract method to this SPI breaks every out-of-tree `FederatedCatalogFactory` at compile time. Every in-tree implementation just throws `UnsupportedOperationException` anyway. Can this be a `default` method that throws, so we can drop the three overrides and their TODOs?
<sub>๐ค AI-generated review comment โ agents make mistakes, so please verify before acting on it.</sub>
##########
CHANGELOG.md:
##########
@@ -48,6 +48,12 @@ request adding CHANGELOG notes for breaking (!) changes and possibly other secti
does not provide is refused at create and update, and whenever a credential is vended for it,
with "S3 credential vending mechanism `<id>` is not available in this server".
+- Polaris now supports Directories, which make objects stored on an object store (including unstructured
+ data) discoverable alongside Iceberg tables: a directory tracks the objects under a base location in an
+ Iceberg table. A simple scan endpoint (`POST .../directories/{directory}/scan`) populates that table; it
+ is disabled by default and can be enabled with the `ENABLE_DIRECTORY_SCAN` feature flag. The whole
+ feature can be disabled with `ENABLE_DIRECTORIES`.
Review Comment:
Can we mark Directories as beta and say `ENABLE_DIRECTORIES` defaults to `true`? Every catalog's config response now lists the directory endpoints.
<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,408 @@
+/*
+ * 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.Optional;
+import java.util.Set;
+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.AlreadyExistsException;
+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.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.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();
+ }
+
+ 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) {
+ PolarisAuthorizableOperation op = PolarisAuthorizableOperation.LIST_TABLES;
+ authorizeBasicNamespaceOperationOrThrow(op, parent);
+
+ return ListDirectoriesResponse.builder()
+ .setIdentifiers(new LinkedHashSet<>(directoryCatalog.listDirectories(parent)))
+ .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);
+
+ // Check that the inventory table name does not collide with an existing entity
+ TableIdentifier tableId = inventoryTableIdentifier(identifier);
+ if (this.icebergCatalog.tableExists(tableId)) {
+ throw new AlreadyExistsException(
+ "Cannot create directory %s: associated table %s already exists", identifier, tableId);
+ }
+
+ 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) {
+ // Roll back the directory entity if table creation fails
+ LOGGER.warn(
+ "Failed to create Iceberg table {} for directory {}, rolling back",
+ tableId,
+ identifier,
+ e);
+ try {
+ this.directoryCatalog.dropDirectory(identifier);
+ } catch (RuntimeException rollbackFailure) {
+ e.addSuppressed(rollbackFailure);
+ }
+ throw e;
+ }
+
+ Directory createdDirectory =
+ Directory.builder()
+ .setName(createdEntity.getName())
+ .setBaseLocation(createdEntity.getBaseLocation())
+ .setFilter(toDirectoryFilter(createdEntity))
+ .setScanSchedule(scanScheduleFromJson(createdEntity.getScanSchedule()))
+ .setProperties(createdEntity.getPropertiesAsMap())
+ .build();
+
+ return LoadDirectoryResponse.builder().setDirectory(createdDirectory).build();
+ }
+
+ public boolean dropDirectory(TableIdentifier identifier) {
+ this.inventoryTable = inventoryTableIdentifier(identifier);
+ PolarisAuthorizableOperation op = PolarisAuthorizableOperation.DROP_TABLE_WITHOUT_PURGE;
+ authorizeCreateTableLikeUnderNamespaceOperationOrThrow(op, identifier);
Review Comment:
Drop checks DROP_TABLE_WITHOUT_PURGE on the parent namespace, not on the directory. Generic tables check the leaf (`resolveAndAuthorizeBasicTableLikeOperationOrThrow` in GenericTableCatalogHandler). Two effects: a TABLE_DROP grant on the directory itself does nothing, and an external authorizer sees a namespace target, so a per-directory deny won't apply. Can we switch to the leaf helper with `PolarisEntitySubType.DIRECTORY`?
<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 Directories feature for managing objects (including unstructured data) on object stores [polaris]
Posted by "dimas-b (via GitHub)" <gi...@apache.org>.
dimas-b commented on code in PR #4613:
URL: https://github.com/apache/polaris/pull/4613#discussion_r4417898761
##########
spec/polaris-catalog-apis/directories-api.yaml:
##########
@@ -0,0 +1,393 @@
+#
+# 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.
+#
+
+---
+paths:
+ /polaris/v1/{prefix}/namespaces/{namespace}/directories:
+ parameters:
+ - $ref: '../iceberg-rest-catalog-open-api.yaml#/components/parameters/prefix'
Review Comment:
non-blocking: I do not think yaml cross-references are convenience for downstream users. If someone wants to copy this API definition to generate clients, the IRC spec has to be copied too, but the IRC spec does not appear to be a conceptual prerequisite to the Directories spec.
Also (perhaps more critical): changes to IRC spec do not have to automatically trickle down to the directories spec, IMHO. It may be a weak point with respect to these two parameters, but in general I think it makes sense ๐ค
##########
spec/polaris-catalog-apis/directories-api.yaml:
##########
@@ -0,0 +1,393 @@
+#
+# 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.
+#
+
+---
+paths:
+ /polaris/v1/{prefix}/namespaces/{namespace}/directories:
+ parameters:
+ - $ref: '../iceberg-rest-catalog-open-api.yaml#/components/parameters/prefix'
+ - $ref: '../iceberg-rest-catalog-open-api.yaml#/components/parameters/namespace'
+
+ get:
+ tags:
+ - Directory API
+ summary: List all directory identifiers underneath a given namespace
+ description: Return all directory identifiers under this namespace
+ operationId: listDirectories
+ parameters:
+ - $ref: '../iceberg-rest-catalog-open-api.yaml#/components/parameters/page-token'
+ - $ref: '../iceberg-rest-catalog-open-api.yaml#/components/parameters/page-size'
Review Comment:
Aside from yaml cross-references, I do not think pagination parameters are defined with due rigour in IRC. Specifically, the treatment of empty page tokens and defauting to "all". I think it would be nicer to define pagination locally in a Polaris spec.
##########
spec/polaris-catalog-apis/directories-api.yaml:
##########
@@ -0,0 +1,393 @@
+#
+# 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.
+#
+
+---
+paths:
+ /polaris/v1/{prefix}/namespaces/{namespace}/directories:
Review Comment:
Non-blocking: this path prefix interleaves with other Polaris API (e.g. Generic Tables, Policies...). Since endpoints are distinct, it is not a problem. However, it makes separating APIs more difficult.
The `v1` prefix is shared and it is usually perceived as the "API root".
There is no static prefix just for the directories API, which can complicate network routing if someone wants to segregate traffic by API.
I know this fit the existing pattern, so my comment is non-blocking ๐
--
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 Directories feature for managing objects (including unstructured data) on object stores [polaris]
Posted by "dimas-b (via GitHub)" <gi...@apache.org>.
dimas-b commented on code in PR #4613:
URL: https://github.com/apache/polaris/pull/4613#discussion_r3364896532
##########
site/content/in-dev/unreleased/directories.md:
##########
@@ -0,0 +1,227 @@
+---
+#
+# 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.
+#
+title: Directories
+type: docs
+weight: 450
+---
+
+## Overview
+
+Directories make objects (including unstructured data like images, videos, documents, and other objects) discoverable alongside
+structured Iceberg tables within a Polaris catalog.
+A directory points to a base location/prefix on an object store and automatically tracks the objects it contains by maintaining
+an Iceberg table with object-level metadata such as URI, size, content type, checksum, ...
+
+This means query engines and tools that already know how to read Iceberg tables can discover and
+access unstructured data with little or no extra work (accessing the object itself).
+
+## Concepts
+
+A directory has two main parts:
+
+1. **Directory configuration** โ stored by the Polaris server. It describes _where_ the data lives,
+ how to authenticate, which objects to include, and how often to re-scan. The configuration "lives" in a namespace.
+2. **Directory table** โ an Iceberg table serving as the inventory of all objects contained in the directory, one row per object discovered during a scan.
+ The directory table uses the configuration name.
+
+The Polaris server itself does not perform scans. Instead, external services (e.g. directory table scanning service) read the directory configuration through the REST API,
+walk the object store, and write the results into the directory table.
+
+## Configuration
+
+A directory is described by the following fields:
+
+| Field | Type | Required | Description |
+|-------|------|----------|-------------|
+| `name` | `string` | Yes | The name of the directory. It is used in REST endpoint paths and becomes the name of the corresponding Iceberg table. |
Review Comment:
Does this mean that the "directory" and the "table" share the same name?
Do they live in the same namespace in Polaris (basically being distinguishable only by object type)?
(this is not an objection, just trying to understand the situation better ๐
)
--
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 Directories feature for managing objects (including unstructured data) on object stores [polaris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #4613:
URL: https://github.com/apache/polaris/pull/4613#issuecomment-5275126657
This PR is stale because it has been open 30 days with no activity. Remove stale label or comment or this will be closed in 5 days.
--
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 Directories feature for managing objects (including unstructured data) on object stores [polaris]
Posted by "jbonofre (via GitHub)" <gi...@apache.org>.
jbonofre commented on PR #4613:
URL: https://github.com/apache/polaris/pull/4613#issuecomment-6503808287
As discussed during the Community Meeting, I will split this in three PRs:
- PR1: only the spec change
- PR2: entity + SPI
- PR3: simple scanner service
--
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