diff --git a/api/src/main/java/org/apache/gravitino/SupportsCatalogs.java b/api/src/main/java/org/apache/gravitino/SupportsCatalogs.java index 088279b69f4..203be19fe49 100644 --- a/api/src/main/java/org/apache/gravitino/SupportsCatalogs.java +++ b/api/src/main/java/org/apache/gravitino/SupportsCatalogs.java @@ -18,6 +18,7 @@ */ package org.apache.gravitino; +import java.util.Collections; import java.util.Map; import org.apache.gravitino.annotation.Evolving; import org.apache.gravitino.exceptions.CatalogAlreadyExistsException; @@ -26,6 +27,8 @@ import org.apache.gravitino.exceptions.NoSuchCatalogException; import org.apache.gravitino.exceptions.NoSuchMetalakeException; import org.apache.gravitino.exceptions.NonEmptyEntityException; +import org.apache.gravitino.secret.SecretBinding; +import org.apache.gravitino.secret.SecretReference; /** * Client interface for supporting catalogs. It includes methods for listing, loading, creating, @@ -84,22 +87,66 @@ default boolean catalogExists(String catalogName) { * the created catalog is the managed catalog, like model, fileset catalog. For the details of the * provider definition, see {@link CatalogProvider}. * - * @param catalogName the name of the catalog. - * @param type the type of the catalog. - * @param provider the provider of the catalog, or null if the catalog is a managed catalog. - * @param comment the comment of the catalog. - * @param properties the properties of the catalog. - * @return The created catalog. - * @throws NoSuchMetalakeException If the metalake does not exist. - * @throws CatalogAlreadyExistsException If the catalog already exists. + *

Delegates to {@link #createCatalog(String, Catalog.Type, String, String, Map, Map, Map)} + * with empty secret maps. + * + * @param catalogName the name of the catalog + * @param type the type of the catalog + * @param provider the provider of the catalog, or null if the catalog is a managed catalog + * @param comment the comment of the catalog + * @param properties the properties of the catalog + * @return the created catalog + * @throws NoSuchMetalakeException if the metalake does not exist + * @throws CatalogAlreadyExistsException if the catalog already exists */ - Catalog createCatalog( + default Catalog createCatalog( String catalogName, Catalog.Type type, String provider, String comment, Map properties) - throws NoSuchMetalakeException, CatalogAlreadyExistsException; + throws NoSuchMetalakeException, CatalogAlreadyExistsException { + return createCatalog( + catalogName, + type, + provider, + comment, + properties, + Collections.emptyMap(), + Collections.emptyMap()); + } + + /** + * Create a catalog with optional secret maps. + * + *

The default implementation rejects create-time secrets. Implementations that support secrets + * must override this method. + * + * @param catalogName the name of the catalog + * @param type the type of the catalog + * @param provider the provider of the catalog, or null if managed + * @param comment the comment of the catalog + * @param properties the properties of the catalog + * @param secretBindings optional property key → binding ({@code provider} + {@code plaintext}) + * for write-through + * @param secretReferences optional property key → secret locator ({@code provider} plus + * provider-specific attributes) + * @return the created catalog + * @throws NoSuchMetalakeException if the metalake does not exist + * @throws CatalogAlreadyExistsException if the catalog already exists + * @throws UnsupportedOperationException if create-time secrets are not supported + */ + default Catalog createCatalog( + String catalogName, + Catalog.Type type, + String provider, + String comment, + Map properties, + Map secretBindings, + Map secretReferences) + throws NoSuchMetalakeException, CatalogAlreadyExistsException { + throw new UnsupportedOperationException("Creating a catalog with secrets is not supported"); + } /** * Create a managed catalog with specified catalog name, type, comment, and properties. diff --git a/api/src/main/java/org/apache/gravitino/SupportsSchemas.java b/api/src/main/java/org/apache/gravitino/SupportsSchemas.java index 42284ed24db..cc059d9f095 100644 --- a/api/src/main/java/org/apache/gravitino/SupportsSchemas.java +++ b/api/src/main/java/org/apache/gravitino/SupportsSchemas.java @@ -20,12 +20,15 @@ package org.apache.gravitino; +import java.util.Collections; import java.util.Map; import org.apache.gravitino.annotation.Evolving; import org.apache.gravitino.exceptions.NoSuchCatalogException; import org.apache.gravitino.exceptions.NoSuchSchemaException; import org.apache.gravitino.exceptions.NonEmptySchemaException; import org.apache.gravitino.exceptions.SchemaAlreadyExistsException; +import org.apache.gravitino.secret.SecretBinding; +import org.apache.gravitino.secret.SecretReference; /** * The client interface to support schema operations. The server side should use the other one with @@ -93,15 +96,48 @@ default boolean schemaExists(String schemaName) { * need the schema with default values applied, use the {@link #loadSchema(String)} method after * creation. * + *

Delegates to {@link #createSchema(String, String, Map, Map, Map)} with empty secret maps. + * + * @param schemaName The name of the schema. + * @param comment The comment of the schema. + * @param properties The properties of the schema. + * @return The schema as defined by the caller, without all default values. + * @throws NoSuchCatalogException If the catalog does not exist. + * @throws SchemaAlreadyExistsException If the schema already exists. + */ + default Schema createSchema(String schemaName, String comment, Map properties) + throws NoSuchCatalogException, SchemaAlreadyExistsException { + return createSchema( + schemaName, comment, properties, Collections.emptyMap(), Collections.emptyMap()); + } + + /** + * Creates a schema with optional secret maps. + * + *

The default implementation rejects create-time secrets. Implementations that support secrets + * must override this method. + * * @param schemaName The name of the schema. * @param comment The comment of the schema. * @param properties The properties of the schema. + * @param secretBindings optional property key → binding ({@code provider} + {@code plaintext}) + * for write-through + * @param secretReferences optional property key → secret locator ({@code provider} plus + * provider-specific attributes) * @return The schema as defined by the caller, without all default values. * @throws NoSuchCatalogException If the catalog does not exist. * @throws SchemaAlreadyExistsException If the schema already exists. + * @throws UnsupportedOperationException if create-time secrets are not supported */ - Schema createSchema(String schemaName, String comment, Map properties) - throws NoSuchCatalogException, SchemaAlreadyExistsException; + default Schema createSchema( + String schemaName, + String comment, + Map properties, + Map secretBindings, + Map secretReferences) + throws NoSuchCatalogException, SchemaAlreadyExistsException { + throw new UnsupportedOperationException("Creating a schema with secrets is not supported"); + } /** * Load metadata properties for a schema. diff --git a/api/src/main/java/org/apache/gravitino/file/Fileset.java b/api/src/main/java/org/apache/gravitino/file/Fileset.java index 7089213984c..2f93630c9e8 100644 --- a/api/src/main/java/org/apache/gravitino/file/Fileset.java +++ b/api/src/main/java/org/apache/gravitino/file/Fileset.java @@ -231,7 +231,7 @@ default String storageLocation() { * location and the value is the storage location path. */ default Map storageLocations() { - throw new UnsupportedOperationException("Not implemented"); + throw new UnsupportedOperationException("Fileset does not support storageLocations."); } /** diff --git a/api/src/main/java/org/apache/gravitino/file/FilesetCatalog.java b/api/src/main/java/org/apache/gravitino/file/FilesetCatalog.java index 68bddf96edb..54922a69f8f 100644 --- a/api/src/main/java/org/apache/gravitino/file/FilesetCatalog.java +++ b/api/src/main/java/org/apache/gravitino/file/FilesetCatalog.java @@ -170,7 +170,7 @@ default Fileset createMultipleLocationFileset( Map secretBindings, Map secretReferences) throws NoSuchSchemaException, FilesetAlreadyExistsException { - throw new UnsupportedOperationException("Not implemented"); + throw new UnsupportedOperationException("Creating a fileset with secrets is not supported"); } /** @@ -258,6 +258,6 @@ default String getFileLocation(NameIdentifier ident, String subPath) */ default String getFileLocation(NameIdentifier ident, String subPath, String locationName) throws NoSuchFilesetException, NoSuchLocationNameException { - throw new UnsupportedOperationException("Not implemented"); + throw new UnsupportedOperationException("getFileLocation is not supported"); } } diff --git a/clients/client-java/src/main/java/org/apache/gravitino/client/BaseSchemaCatalog.java b/clients/client-java/src/main/java/org/apache/gravitino/client/BaseSchemaCatalog.java index 0553abfc4bf..7ea349b22b5 100644 --- a/clients/client-java/src/main/java/org/apache/gravitino/client/BaseSchemaCatalog.java +++ b/clients/client-java/src/main/java/org/apache/gravitino/client/BaseSchemaCatalog.java @@ -42,6 +42,8 @@ import org.apache.gravitino.dto.responses.DropResponse; import org.apache.gravitino.dto.responses.EntityListResponse; import org.apache.gravitino.dto.responses.SchemaResponse; +import org.apache.gravitino.dto.secret.SecretBindingDTO; +import org.apache.gravitino.dto.secret.SecretReferenceDTO; import org.apache.gravitino.exceptions.NoSuchCatalogException; import org.apache.gravitino.exceptions.NoSuchPolicyException; import org.apache.gravitino.exceptions.NoSuchSchemaException; @@ -56,6 +58,8 @@ import org.apache.gravitino.policy.Policy; import org.apache.gravitino.policy.SupportsPolicies; import org.apache.gravitino.rest.RESTUtils; +import org.apache.gravitino.secret.SecretBinding; +import org.apache.gravitino.secret.SecretReference; import org.apache.gravitino.tag.SupportsTags; import org.apache.gravitino.tag.Tag; @@ -167,7 +171,7 @@ public String[] listSchemas(String parentSchema) } /** - * Create a new schema with specified identifier, comment and metadata. + * Create a new schema with specified identifier, comment and properties. * * @param schemaName The name identifier of the schema. * @param comment The comment of the schema. @@ -179,8 +183,40 @@ public String[] listSchemas(String parentSchema) @Override public Schema createSchema(String schemaName, String comment, Map properties) throws NoSuchCatalogException, SchemaAlreadyExistsException { + return createSchema( + schemaName, comment, properties, Collections.emptyMap(), Collections.emptyMap()); + } + + /** + * Create a new schema with specified identifier, comment, properties, and optional secret maps. + * + * @param schemaName The name identifier of the schema. + * @param comment The comment of the schema. + * @param properties The properties of the schema. + * @param secretBindings Optional property key → binding ({@code provider} + {@code plaintext}) + * for write-through. + * @param secretReferences Optional property key → secret locator ({@code provider} plus + * provider-specific attributes). + * @return The created {@link Schema}. + * @throws NoSuchCatalogException if the catalog with specified namespace does not exist. + * @throws SchemaAlreadyExistsException if the schema with specified identifier already exists. + */ + @Override + public Schema createSchema( + String schemaName, + String comment, + Map properties, + Map secretBindings, + Map secretReferences) + throws NoSuchCatalogException, SchemaAlreadyExistsException { - SchemaCreateRequest req = new SchemaCreateRequest(schemaName, comment, properties); + SchemaCreateRequest req = + new SchemaCreateRequest( + schemaName, + comment, + properties, + SecretBindingDTO.fromSecretBindings(secretBindings), + SecretReferenceDTO.fromSecretReferences(secretReferences)); req.validate(); SchemaResponse resp = diff --git a/clients/client-java/src/main/java/org/apache/gravitino/client/FilesetCatalog.java b/clients/client-java/src/main/java/org/apache/gravitino/client/FilesetCatalog.java index 40dc2d72869..ad61a82b401 100644 --- a/clients/client-java/src/main/java/org/apache/gravitino/client/FilesetCatalog.java +++ b/clients/client-java/src/main/java/org/apache/gravitino/client/FilesetCatalog.java @@ -137,8 +137,38 @@ public Fileset loadFileset(NameIdentifier ident) throws NoSuchFilesetException { * @param type The type of the fileset. * @param storageLocations The location names and storage locations of the fileset. * @param properties The properties of the fileset. - * @param secretBindings Optional property key → binding ({ provider} + { plaintext}) for - * write-through. + * @return The created fileset metadata + * @throws NoSuchSchemaException If the schema does not exist. + * @throws FilesetAlreadyExistsException If the fileset already exists. + */ + @Override + public Fileset createMultipleLocationFileset( + NameIdentifier ident, + String comment, + Fileset.Type type, + Map storageLocations, + Map properties) + throws NoSuchSchemaException, FilesetAlreadyExistsException { + return createMultipleLocationFileset( + ident, + comment, + type, + storageLocations, + properties, + Collections.emptyMap(), + Collections.emptyMap()); + } + + /** + * Create a fileset metadata with multiple storage locations in the catalog. + * + * @param ident A fileset identifier. + * @param comment The comment of the fileset. + * @param type The type of the fileset. + * @param storageLocations The location names and storage locations of the fileset. + * @param properties The properties of the fileset. + * @param secretBindings Optional property key → binding ({@code provider} + {@code plaintext}) + * for write-through. * @param secretReferences Optional property key → secret locator ({@code provider} plus * provider-specific attributes). * @return The created fileset metadata diff --git a/clients/client-java/src/main/java/org/apache/gravitino/client/GravitinoClient.java b/clients/client-java/src/main/java/org/apache/gravitino/client/GravitinoClient.java index 16b222ba8d6..120435b3b0d 100644 --- a/clients/client-java/src/main/java/org/apache/gravitino/client/GravitinoClient.java +++ b/clients/client-java/src/main/java/org/apache/gravitino/client/GravitinoClient.java @@ -21,6 +21,7 @@ import com.google.common.base.Preconditions; import com.google.common.collect.Sets; +import java.util.Collections; import java.util.List; import java.util.Map; import java.util.Optional; @@ -67,6 +68,8 @@ import org.apache.gravitino.policy.PolicyChange; import org.apache.gravitino.policy.PolicyContent; import org.apache.gravitino.policy.PolicyOperations; +import org.apache.gravitino.secret.SecretBinding; +import org.apache.gravitino.secret.SecretReference; import org.apache.gravitino.tag.Tag; import org.apache.gravitino.tag.TagChange; import org.apache.gravitino.tag.TagOperations; @@ -139,7 +142,29 @@ public Catalog createCatalog( String comment, Map properties) throws NoSuchMetalakeException, CatalogAlreadyExistsException { - return getMetalake().createCatalog(catalogName, type, provider, comment, properties); + return createCatalog( + catalogName, + type, + provider, + comment, + properties, + Collections.emptyMap(), + Collections.emptyMap()); + } + + @Override + public Catalog createCatalog( + String catalogName, + Catalog.Type type, + String provider, + String comment, + Map properties, + Map secretBindings, + Map secretReferences) + throws NoSuchMetalakeException, CatalogAlreadyExistsException { + return getMetalake() + .createCatalog( + catalogName, type, provider, comment, properties, secretBindings, secretReferences); } @Override diff --git a/clients/client-java/src/main/java/org/apache/gravitino/client/GravitinoMetalake.java b/clients/client-java/src/main/java/org/apache/gravitino/client/GravitinoMetalake.java index f88a33a227d..671f87eaa8f 100644 --- a/clients/client-java/src/main/java/org/apache/gravitino/client/GravitinoMetalake.java +++ b/clients/client-java/src/main/java/org/apache/gravitino/client/GravitinoMetalake.java @@ -95,6 +95,8 @@ import org.apache.gravitino.dto.responses.TagResponse; import org.apache.gravitino.dto.responses.UserListResponse; import org.apache.gravitino.dto.responses.UserResponse; +import org.apache.gravitino.dto.secret.SecretBindingDTO; +import org.apache.gravitino.dto.secret.SecretReferenceDTO; import org.apache.gravitino.exceptions.CatalogAlreadyExistsException; import org.apache.gravitino.exceptions.CatalogInUseException; import org.apache.gravitino.exceptions.GroupAlreadyExistsException; @@ -128,6 +130,8 @@ import org.apache.gravitino.policy.PolicyContent; import org.apache.gravitino.policy.PolicyOperations; import org.apache.gravitino.rest.RESTUtils; +import org.apache.gravitino.secret.SecretBinding; +import org.apache.gravitino.secret.SecretReference; import org.apache.gravitino.tag.Tag; import org.apache.gravitino.tag.TagChange; import org.apache.gravitino.tag.TagOperations; @@ -255,8 +259,52 @@ public Catalog createCatalog( String comment, Map properties) throws NoSuchMetalakeException, CatalogAlreadyExistsException { + return createCatalog( + catalogName, + type, + provider, + comment, + properties, + Collections.emptyMap(), + Collections.emptyMap()); + } + + /** + * Create a new catalog with specified identifier, type, comment, properties, and optional secret + * maps. + * + * @param catalogName The identifier of the catalog. + * @param type The type of the catalog. + * @param provider The provider of the catalog. + * @param comment The comment of the catalog. + * @param properties The properties of the catalog. + * @param secretBindings Optional property key → binding ({@code provider} + {@code plaintext}) + * for write-through. + * @param secretReferences Optional property key → secret locator ({@code provider} plus + * provider-specific attributes). + * @return The created {@link Catalog}. + * @throws NoSuchMetalakeException if the metalake with specified namespace does not exist. + * @throws CatalogAlreadyExistsException if the catalog with specified identifier already exists. + */ + @Override + public Catalog createCatalog( + String catalogName, + Catalog.Type type, + String provider, + String comment, + Map properties, + Map secretBindings, + Map secretReferences) + throws NoSuchMetalakeException, CatalogAlreadyExistsException { CatalogCreateRequest req = - new CatalogCreateRequest(catalogName, type, provider, comment, properties); + new CatalogCreateRequest( + catalogName, + type, + provider, + comment, + properties, + SecretBindingDTO.fromSecretBindings(secretBindings), + SecretReferenceDTO.fromSecretReferences(secretReferences)); req.validate(); CatalogResponse resp = diff --git a/clients/client-python/gravitino/api/supports_schemas.py b/clients/client-python/gravitino/api/supports_schemas.py index 7a61400b6f9..b81d74c1539 100644 --- a/clients/client-python/gravitino/api/supports_schemas.py +++ b/clients/client-python/gravitino/api/supports_schemas.py @@ -16,12 +16,17 @@ # under the License. from abc import ABC, abstractmethod -from typing import Dict, List, Optional +from types import MappingProxyType +from typing import Dict, List, Mapping, Optional from gravitino.api.schema import Schema from gravitino.api.schema_change import SchemaChange +from gravitino.api.secret import SecretBinding, SecretReference from gravitino.exceptions.base import NoSuchSchemaException +_EMPTY_SECRET_BINDINGS: Mapping[str, SecretBinding] = MappingProxyType({}) +_EMPTY_SECRET_REFERENCES: Mapping[str, SecretReference] = MappingProxyType({}) + class SupportsSchemas(ABC): """ @@ -79,7 +84,12 @@ def schema_exists(self, schema_name: str) -> bool: @abstractmethod def create_schema( - self, schema_name: str, comment: str, properties: Dict[str, str] + self, + schema_name: str, + comment: str, + properties: Dict[str, str], + secret_bindings: Mapping[str, SecretBinding] = _EMPTY_SECRET_BINDINGS, + secret_references: Mapping[str, SecretReference] = _EMPTY_SECRET_REFERENCES, ) -> Schema: """Create a schema in the catalog. @@ -87,6 +97,8 @@ def create_schema( schema_name: The name of the schema. comment: The comment of the schema. properties: The properties of the schema. + secret_bindings: Optional property key → binding (provider + plaintext) for write-through. + secret_references: Optional property key → locator attributes. Raises: NoSuchCatalogException: If the catalog does not exist. diff --git a/clients/client-python/gravitino/client/base_schema_catalog.py b/clients/client-python/gravitino/client/base_schema_catalog.py index a2cbae094b5..1d21c45e8c3 100644 --- a/clients/client-python/gravitino/client/base_schema_catalog.py +++ b/clients/client-python/gravitino/client/base_schema_catalog.py @@ -16,7 +16,8 @@ # under the License. import logging -from typing import Dict, List, Optional +from types import MappingProxyType +from typing import Dict, List, Mapping, Optional from gravitino.api.catalog import Catalog from gravitino.api.function.function import Function @@ -28,6 +29,7 @@ from gravitino.api.metadata_objects import MetadataObjects from gravitino.api.schema import Schema from gravitino.api.schema_change import SchemaChange +from gravitino.api.secret import SecretBinding, SecretReference from gravitino.api.supports_schemas import SupportsSchemas from gravitino.api.tag.supports_tags import SupportsTags from gravitino.api.tag.tag import Tag @@ -54,6 +56,9 @@ logger = logging.getLogger(__name__) +_EMPTY_SECRET_BINDINGS: Mapping[str, SecretBinding] = MappingProxyType({}) +_EMPTY_SECRET_REFERENCES: Mapping[str, SecretReference] = MappingProxyType({}) + class BaseSchemaCatalog( CatalogDTO, @@ -159,6 +164,8 @@ def create_schema( schema_name: str = None, comment: str = None, properties: Dict[str, str] = None, + secret_bindings: Mapping[str, SecretBinding] = _EMPTY_SECRET_BINDINGS, + secret_references: Mapping[str, SecretReference] = _EMPTY_SECRET_REFERENCES, ) -> Schema: """Create a new schema with specified identifier, comment and metadata. @@ -166,6 +173,8 @@ def create_schema( schema_name: The name of the schema. comment: The comment of the schema. properties: The properties of the schema. + secret_bindings: Optional property key → binding (provider + plaintext) for write-through. + secret_references: Optional property key → locator attributes. Raises: NoSuchCatalogException if the catalog with specified namespace does not exist. @@ -174,7 +183,13 @@ def create_schema( Returns: The created Schema. """ - req = SchemaCreateRequest(encode_string(schema_name), comment, properties) + req = SchemaCreateRequest( + encode_string(schema_name), + comment, + properties, + secret_bindings, + secret_references, + ) req.validate() resp = self.rest_client.post( diff --git a/clients/client-python/gravitino/client/gravitino_client.py b/clients/client-python/gravitino/client/gravitino_client.py index 3e936d2eb0f..303877f3d4b 100644 --- a/clients/client-python/gravitino/client/gravitino_client.py +++ b/clients/client-python/gravitino/client/gravitino_client.py @@ -17,7 +17,8 @@ from __future__ import annotations -from typing import Dict, List, Optional +from types import MappingProxyType +from typing import Dict, List, Mapping, Optional from gravitino.api.authorization.group import Group from gravitino.api.authorization.owner import Owner @@ -32,6 +33,7 @@ from gravitino.api.job.job_template_change import JobTemplateChange from gravitino.api.job.supports_jobs import SupportsJobs from gravitino.api.metadata_object import MetadataObject +from gravitino.api.secret import SecretBinding, SecretReference from gravitino.api.tag.tag_operations import TagOperations from gravitino.auth.auth_data_provider import AuthDataProvider from gravitino.client.gravitino_client_base import GravitinoClientBase @@ -39,6 +41,9 @@ from ..api.tag.tag import Tag +_EMPTY_SECRET_BINDINGS: Mapping[str, SecretBinding] = MappingProxyType({}) +_EMPTY_SECRET_REFERENCES: Mapping[str, SecretReference] = MappingProxyType({}) + class GravitinoClient(GravitinoClientBase, SupportsJobs, TagOperations): """Gravitino Client for a user to interact with the Gravitino API, allowing the client to list, @@ -103,9 +108,17 @@ def create_catalog( provider: str, comment: str, properties: Dict[str, str], + secret_bindings: Mapping[str, SecretBinding] = _EMPTY_SECRET_BINDINGS, + secret_references: Mapping[str, SecretReference] = _EMPTY_SECRET_REFERENCES, ) -> Catalog: return self.get_metalake().create_catalog( - name, catalog_type, provider, comment, properties + name, + catalog_type, + provider, + comment, + properties, + secret_bindings, + secret_references, ) def alter_catalog(self, name: str, *changes: CatalogChange): diff --git a/clients/client-python/gravitino/client/gravitino_metalake.py b/clients/client-python/gravitino/client/gravitino_metalake.py index beb4baf96aa..a48b720f4a6 100644 --- a/clients/client-python/gravitino/client/gravitino_metalake.py +++ b/clients/client-python/gravitino/client/gravitino_metalake.py @@ -16,7 +16,8 @@ # under the License. # pylint: disable=too-many-lines import logging -from typing import Dict, List, Optional +from types import MappingProxyType +from typing import Dict, List, Mapping, Optional from gravitino.api.authorization.group import Group from gravitino.api.authorization.owner import Owner @@ -31,6 +32,7 @@ from gravitino.api.job.job_template_change import JobTemplateChange from gravitino.api.job.supports_jobs import SupportsJobs from gravitino.api.metadata_object import MetadataObject +from gravitino.api.secret import SecretBinding, SecretReference from gravitino.api.tag.tag import Tag from gravitino.api.tag.tag_operations import TagOperations from gravitino.client.dto_converters import DTOConverters @@ -104,6 +106,9 @@ logger = logging.getLogger(__name__) +_EMPTY_SECRET_BINDINGS: Mapping[str, SecretBinding] = MappingProxyType({}) +_EMPTY_SECRET_REFERENCES: Mapping[str, SecretReference] = MappingProxyType({}) + class GravitinoMetalake( MetalakeDTO, @@ -217,6 +222,8 @@ def create_catalog( provider: str, comment: str, properties: Dict[str, str], + secret_bindings: Mapping[str, SecretBinding] = _EMPTY_SECRET_BINDINGS, + secret_references: Mapping[str, SecretReference] = _EMPTY_SECRET_REFERENCES, ) -> Catalog: """Create a new catalog with specified name, catalog type, comment and properties. @@ -228,6 +235,8 @@ def create_catalog( None provider. For the details, please refer to the Catalog.Type. comment: The comment of the catalog. properties: The properties of the catalog. + secret_bindings: Optional property key → binding (provider + plaintext) for write-through. + secret_references: Optional property key → locator attributes. Raises: NoSuchMetalakeException if the metalake does not exist. @@ -243,6 +252,8 @@ def create_catalog( provider=provider, comment=comment, properties=properties, + secret_bindings=secret_bindings, + secret_references=secret_references, ) catalog_create_request.validate() @@ -251,6 +262,7 @@ def create_catalog( url, json=catalog_create_request, error_handler=CATALOG_ERROR_HANDLER ) catalog_resp = CatalogResponse.from_json(response.body, infer_missing=True) + catalog_resp.validate() return DTOConverters.to_catalog( self.name(), catalog_resp.catalog(), self.rest_client diff --git a/clients/client-python/gravitino/dto/requests/catalog_create_request.py b/clients/client-python/gravitino/dto/requests/catalog_create_request.py index 885011118dc..f0cfdb62a1e 100644 --- a/clients/client-python/gravitino/dto/requests/catalog_create_request.py +++ b/clients/client-python/gravitino/dto/requests/catalog_create_request.py @@ -16,13 +16,18 @@ # under the License. from dataclasses import dataclass, field -from typing import Optional, Dict +from types import MappingProxyType +from typing import Dict, Mapping, Optional from dataclasses_json import config from gravitino.api.catalog import Catalog +from gravitino.api.secret import SecretBinding, SecretReference from gravitino.rest.rest_message import RESTRequest +_EMPTY_SECRET_BINDINGS: Mapping[str, SecretBinding] = MappingProxyType({}) +_EMPTY_SECRET_REFERENCES: Mapping[str, SecretReference] = MappingProxyType({}) + @dataclass class CatalogCreateRequest(RESTRequest): @@ -41,6 +46,12 @@ class CatalogCreateRequest(RESTRequest): _properties: Optional[Dict[str, str]] = field( metadata=config(field_name="properties") ) + _secret_bindings: Dict[str, SecretBinding] = field( + default_factory=dict, metadata=config(field_name="secretBindings") + ) + _secret_references: Dict[str, SecretReference] = field( + default_factory=dict, metadata=config(field_name="secretReferences") + ) def __init__( self, @@ -49,12 +60,16 @@ def __init__( provider: str = None, comment: str = None, properties: Dict[str, str] = None, + secret_bindings: Mapping[str, SecretBinding] = _EMPTY_SECRET_BINDINGS, + secret_references: Mapping[str, SecretReference] = _EMPTY_SECRET_REFERENCES, ): self._name = name self._type = catalog_type self._provider = provider self._comment = comment self._properties = properties + self._secret_bindings = dict(secret_bindings) + self._secret_references = dict(secret_references) def validate(self): """Validates the fields of the request. diff --git a/clients/client-python/gravitino/dto/requests/schema_create_request.py b/clients/client-python/gravitino/dto/requests/schema_create_request.py index 3baae7e857a..6cbcad81f32 100644 --- a/clients/client-python/gravitino/dto/requests/schema_create_request.py +++ b/clients/client-python/gravitino/dto/requests/schema_create_request.py @@ -16,12 +16,17 @@ # under the License. from dataclasses import dataclass, field -from typing import Optional, Dict +from types import MappingProxyType +from typing import Dict, Mapping, Optional from dataclasses_json import config +from gravitino.api.secret import SecretBinding, SecretReference from gravitino.rest.rest_message import RESTRequest +_EMPTY_SECRET_BINDINGS: Mapping[str, SecretBinding] = MappingProxyType({}) +_EMPTY_SECRET_REFERENCES: Mapping[str, SecretReference] = MappingProxyType({}) + @dataclass class SchemaCreateRequest(RESTRequest): @@ -32,14 +37,32 @@ class SchemaCreateRequest(RESTRequest): _properties: Optional[Dict[str, str]] = field( metadata=config(field_name="properties") ) + _secret_bindings: Dict[str, SecretBinding] = field( + default_factory=dict, metadata=config(field_name="secretBindings") + ) + _secret_references: Dict[str, SecretReference] = field( + default_factory=dict, metadata=config(field_name="secretReferences") + ) def __init__( - self, name: str, comment: Optional[str], properties: Optional[Dict[str, str]] + self, + name: str, + comment: Optional[str], + properties: Optional[Dict[str, str]], + secret_bindings: Mapping[str, SecretBinding] = _EMPTY_SECRET_BINDINGS, + secret_references: Mapping[str, SecretReference] = _EMPTY_SECRET_REFERENCES, ): self._name = name self._comment = comment self._properties = properties + self._secret_bindings = dict(secret_bindings) + self._secret_references = dict(secret_references) def validate(self): + """Validates the request. + + Raises: + IllegalArgumentException if the request is invalid. + """ if not self._name: raise ValueError('"name" field is required and cannot be empty') diff --git a/common/src/main/java/org/apache/gravitino/dto/requests/CatalogCreateRequest.java b/common/src/main/java/org/apache/gravitino/dto/requests/CatalogCreateRequest.java index 3223fce6c8c..acf8442c4d0 100644 --- a/common/src/main/java/org/apache/gravitino/dto/requests/CatalogCreateRequest.java +++ b/common/src/main/java/org/apache/gravitino/dto/requests/CatalogCreateRequest.java @@ -19,8 +19,10 @@ package org.apache.gravitino.dto.requests; import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonInclude; import com.fasterxml.jackson.annotation.JsonProperty; import com.google.common.base.Preconditions; +import java.util.Collections; import java.util.Map; import javax.annotation.Nullable; import lombok.EqualsAndHashCode; @@ -29,6 +31,8 @@ import org.apache.commons.lang3.StringUtils; import org.apache.gravitino.Catalog; import org.apache.gravitino.CatalogProvider; +import org.apache.gravitino.dto.secret.SecretBindingDTO; +import org.apache.gravitino.dto.secret.SecretReferenceDTO; import org.apache.gravitino.rest.RESTRequest; /** Represents a request to create a catalog. */ @@ -54,6 +58,32 @@ public class CatalogCreateRequest implements RESTRequest { @JsonProperty("properties") private final Map properties; + @JsonInclude(JsonInclude.Include.NON_EMPTY) + @JsonProperty("secretBindings") + private final Map secretBindings; + + @JsonInclude(JsonInclude.Include.NON_EMPTY) + @JsonProperty("secretReferences") + private final Map secretReferences; + + /** + * Constructor for CatalogCreateRequest without secret maps. + * + * @param name The name of the catalog. + * @param type The type of the catalog. + * @param provider The provider of the catalog. + * @param comment The comment for the catalog. + * @param properties The properties for the catalog. + */ + public CatalogCreateRequest( + String name, + Catalog.Type type, + String provider, + String comment, + Map properties) { + this(name, type, provider, comment, properties, Collections.emptyMap(), Collections.emptyMap()); + } + /** * Constructor for CatalogCreateRequest. * @@ -62,6 +92,10 @@ public class CatalogCreateRequest implements RESTRequest { * @param provider The provider of the catalog. * @param comment The comment for the catalog. * @param properties The properties for the catalog. + * @param secretBindings Optional property key → binding DTO ({@code provider} + {@code + * plaintext}) for write-through secrets. + * @param secretReferences Optional property key → secret locator DTO ({@code provider} plus + * provider-specific attributes). */ @JsonCreator public CatalogCreateRequest( @@ -69,11 +103,17 @@ public CatalogCreateRequest( @JsonProperty("type") Catalog.Type type, @JsonProperty("provider") String provider, @JsonProperty("comment") String comment, - @JsonProperty("properties") Map properties) { + @JsonProperty("properties") Map properties, + @JsonProperty("secretBindings") Map secretBindings, + @JsonProperty("secretReferences") Map secretReferences) { this.name = name; this.type = type; this.comment = comment; this.properties = properties; + // Match FilesetCreateRequest field defaults when JSON omits secret maps (@JsonCreator passes + // null for absent properties). + this.secretBindings = secretBindings == null ? Collections.emptyMap() : secretBindings; + this.secretReferences = secretReferences == null ? Collections.emptyMap() : secretReferences; if (StringUtils.isNotBlank(provider)) { this.provider = provider; diff --git a/common/src/main/java/org/apache/gravitino/dto/requests/SchemaCreateRequest.java b/common/src/main/java/org/apache/gravitino/dto/requests/SchemaCreateRequest.java index ec103c0629d..37456860120 100644 --- a/common/src/main/java/org/apache/gravitino/dto/requests/SchemaCreateRequest.java +++ b/common/src/main/java/org/apache/gravitino/dto/requests/SchemaCreateRequest.java @@ -18,14 +18,19 @@ */ package org.apache.gravitino.dto.requests; +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonInclude; import com.fasterxml.jackson.annotation.JsonProperty; import com.google.common.base.Preconditions; +import java.util.Collections; import java.util.Map; import javax.annotation.Nullable; import lombok.EqualsAndHashCode; import lombok.Getter; import lombok.ToString; import org.apache.commons.lang3.StringUtils; +import org.apache.gravitino.dto.secret.SecretBindingDTO; +import org.apache.gravitino.dto.secret.SecretReferenceDTO; import org.apache.gravitino.rest.RESTRequest; /** Represents a request to create a schema. */ @@ -45,22 +50,53 @@ public class SchemaCreateRequest implements RESTRequest { @JsonProperty("properties") private final Map properties; + @JsonInclude(JsonInclude.Include.NON_EMPTY) + @JsonProperty("secretBindings") + private final Map secretBindings; + + @JsonInclude(JsonInclude.Include.NON_EMPTY) + @JsonProperty("secretReferences") + private final Map secretReferences; + /** Default constructor for Jackson deserialization. */ public SchemaCreateRequest() { - this(null, null, null); + this(null, null, null, Collections.emptyMap(), Collections.emptyMap()); } /** - * Creates a new SchemaCreateRequest. + * Creates a new SchemaCreateRequest without secret maps. * * @param name The name of the schema. * @param comment The comment of the schema. * @param properties The properties of the schema. */ public SchemaCreateRequest(String name, String comment, Map properties) { + this(name, comment, properties, Collections.emptyMap(), Collections.emptyMap()); + } + + /** + * Creates a new SchemaCreateRequest. + * + * @param name The name of the schema. + * @param comment The comment of the schema. + * @param properties The properties of the schema. + * @param secretBindings Optional property key → binding DTO ({@code provider} + {@code + * plaintext}) for write-through secrets. + * @param secretReferences Optional property key → secret locator DTO ({@code provider} plus + * provider-specific attributes). + */ + @JsonCreator + public SchemaCreateRequest( + @JsonProperty("name") String name, + @JsonProperty("comment") String comment, + @JsonProperty("properties") Map properties, + @JsonProperty("secretBindings") Map secretBindings, + @JsonProperty("secretReferences") Map secretReferences) { this.name = name; this.comment = comment; this.properties = properties; + this.secretBindings = secretBindings == null ? Collections.emptyMap() : secretBindings; + this.secretReferences = secretReferences == null ? Collections.emptyMap() : secretReferences; } /** diff --git a/core/src/main/java/org/apache/gravitino/GravitinoEnv.java b/core/src/main/java/org/apache/gravitino/GravitinoEnv.java index def00e8cbc1..14f9061e06c 100644 --- a/core/src/main/java/org/apache/gravitino/GravitinoEnv.java +++ b/core/src/main/java/org/apache/gravitino/GravitinoEnv.java @@ -745,8 +745,20 @@ private void initGravitinoServerComponents() { this.credentialOperationDispatcher = new CredentialOperationDispatcher(catalogManager, entityStore, idGenerator, secretManager); + // Fileset dispatcher is created before schema dispatcher so schema can take it directly. + FilesetOperationDispatcher filesetOperationDispatcher = + new FilesetOperationDispatcher(catalogManager, entityStore, idGenerator, secretManager); + FilesetNormalizeDispatcher filesetNormalizeDispatcher = + new FilesetNormalizeDispatcher(filesetOperationDispatcher, catalogManager); + this.internalFilesetDispatcher = filesetNormalizeDispatcher; + FilesetEventDispatcher filesetEventDispatcher = + new FilesetEventDispatcher(eventBus, filesetNormalizeDispatcher); + this.filesetDispatcher = new FilesetHookDispatcher(filesetEventDispatcher); + catalogManager.setFilesetDispatcher(filesetNormalizeDispatcher); + SchemaOperationDispatcher schemaOperationDispatcher = - new SchemaOperationDispatcher(catalogManager, entityStore, idGenerator, secretManager); + new SchemaOperationDispatcher( + catalogManager, entityStore, idGenerator, secretManager, filesetNormalizeDispatcher); this.internalSchemaDispatcher = schemaOperationDispatcher; SchemaNormalizeDispatcher schemaNormalizeDispatcher = new SchemaNormalizeDispatcher(schemaOperationDispatcher, catalogManager); @@ -782,15 +794,6 @@ private void initGravitinoServerComponents() { new PartitionNormalizeDispatcher(partitionOperationDispatcher, catalogManager); this.partitionDispatcher = new PartitionEventDispatcher(eventBus, partitionNormalizeDispatcher); - FilesetOperationDispatcher filesetOperationDispatcher = - new FilesetOperationDispatcher(catalogManager, entityStore, idGenerator, secretManager); - FilesetNormalizeDispatcher filesetNormalizeDispatcher = - new FilesetNormalizeDispatcher(filesetOperationDispatcher, catalogManager); - this.internalFilesetDispatcher = filesetNormalizeDispatcher; - FilesetEventDispatcher filesetEventDispatcher = - new FilesetEventDispatcher(eventBus, filesetNormalizeDispatcher); - this.filesetDispatcher = new FilesetHookDispatcher(filesetEventDispatcher); - TopicOperationDispatcher topicOperationDispatcher = new TopicOperationDispatcher(catalogManager, entityStore, idGenerator, secretManager); TopicNormalizeDispatcher topicNormalizeDispatcher = diff --git a/core/src/main/java/org/apache/gravitino/catalog/CatalogManager.java b/core/src/main/java/org/apache/gravitino/catalog/CatalogManager.java index 72c0139cf43..07135a4a2e5 100644 --- a/core/src/main/java/org/apache/gravitino/catalog/CatalogManager.java +++ b/core/src/main/java/org/apache/gravitino/catalog/CatalogManager.java @@ -25,6 +25,7 @@ import static org.apache.gravitino.catalog.PropertiesMetadataHelpers.validatePropertyForCreate; import static org.apache.gravitino.connector.BaseCatalogPropertiesMetadata.PROPERTY_METALAKE_IN_USE; import static org.apache.gravitino.metalake.MetalakeManager.checkMetalake; +import static org.apache.gravitino.utils.NameIdentifierUtil.ofFileset; import com.github.benmanes.caffeine.cache.Cache; import com.github.benmanes.caffeine.cache.Caffeine; @@ -100,13 +101,18 @@ import org.apache.gravitino.messaging.TopicCatalog; import org.apache.gravitino.meta.AuditInfo; import org.apache.gravitino.meta.CatalogEntity; +import org.apache.gravitino.meta.FilesetEntity; import org.apache.gravitino.meta.SchemaEntity; import org.apache.gravitino.model.ModelCatalog; import org.apache.gravitino.rel.SupportsPartitions; import org.apache.gravitino.rel.Table; import org.apache.gravitino.rel.TableCatalog; import org.apache.gravitino.rel.ViewCatalog; +import org.apache.gravitino.secret.SecretBinding; import org.apache.gravitino.secret.SecretManager; +import org.apache.gravitino.secret.SecretMaterial; +import org.apache.gravitino.secret.SecretPropertyUtils; +import org.apache.gravitino.secret.SecretReference; import org.apache.gravitino.storage.IdGenerator; import org.apache.gravitino.storage.relational.SupportsEntityChangeLog; import org.apache.gravitino.utils.ClassLoaderKey; @@ -364,6 +370,8 @@ private ModelCatalog asModels() { @SuppressWarnings("UnusedVariable") private final SecretManager secretManager; + private FilesetDispatcher filesetDispatcher; + private final List> removalListeners = Lists.newArrayList(); private final ConcurrentHashMap localMutationCounts = new ConcurrentHashMap<>(); @@ -379,7 +387,7 @@ private ModelCatalog asModels() { * @param config The configuration for the manager. * @param store The entity store to use. * @param idGenerator The id generator to use. - * @param secretManager The secret manager used by catalog operations. + * @param secretManager The secret manager to use for create-time secret bindings/references. */ public CatalogManager( Config config, EntityStore store, IdGenerator idGenerator, SecretManager secretManager) { @@ -429,6 +437,16 @@ public CatalogManager( } } + /** + * Sets the {@link FilesetDispatcher} used when force-dropping a catalog. Must be called after the + * fileset dispatcher is constructed (CatalogManager is created first). + * + * @param filesetDispatcher The fileset dispatcher. + */ + public void setFilesetDispatcher(FilesetDispatcher filesetDispatcher) { + this.filesetDispatcher = Preconditions.checkNotNull(filesetDispatcher); + } + /** * Closes the CatalogManager and releases any resources associated with it. This method * invalidates all cached catalog instances and clears the cache. @@ -595,10 +613,31 @@ public Catalog createCatalog( String comment, Map properties) throws NoSuchMetalakeException, CatalogAlreadyExistsException { + return createCatalog( + ident, type, provider, comment, properties, Collections.emptyMap(), Collections.emptyMap()); + } + + @Override + public Catalog createCatalog( + NameIdentifier ident, + Catalog.Type type, + String provider, + String comment, + Map properties, + Map secretBindings, + Map secretReferences) + throws NoSuchMetalakeException, CatalogAlreadyExistsException { NameIdentifier metalakeIdent = NameIdentifier.of(ident.namespace().levels()); - Map mergedConfig = buildCatalogConf(provider, properties); + Map mergedConfig = + SecretPropertyUtils.copyEntityProperties( + buildCatalogConf(provider, properties), secretBindings, secretReferences); long uid = idGenerator.nextId(); + + List secretMaterials = + secretManager.assembleSecretMaterials( + properties, mergedConfig, "catalog", uid, secretBindings, secretReferences); + StringIdentifier stringId = StringIdentifier.fromId(uid); Instant now = Instant.now(); String creator = PrincipalUtils.getCurrentPrincipal().getName(); @@ -627,6 +666,8 @@ public Catalog createCatalog( checkMetalake(metalakeIdent, store); boolean needClean = true; try { + secretManager.writeSecrets(secretMaterials); + store.put(e, false /* overwrite */); CatalogWrapper wrapper = catalogCache.get(ident, id -> createCatalogWrapper(e, mergedConfig)); @@ -635,7 +676,10 @@ public Catalog createCatalog( return wrapper.catalog; } catch (EntityAlreadyExistsException e1) { + // Catalog already exists: do not delete it, but roll back secrets written for this + // attempt (needClean is false so finally will not roll back). needClean = false; + secretManager.rollbackSecrets(secretMaterials); LOG.warn("Catalog {} already exists", ident, e1); throw new CatalogAlreadyExistsException("Catalog %s already exists", ident); @@ -652,6 +696,9 @@ public Catalog createCatalog( } finally { if (needClean) { + // Create failed after writeSecrets (or writeSecrets itself failed — rollback is + // best-effort / idempotent with writeSecrets' own partial cleanup). + secretManager.rollbackSecrets(secretMaterials); // since we put the catalog entity into the store but failed to create the catalog // instance, // we need to clean up the entity stored. @@ -955,29 +1002,69 @@ public boolean dropCatalog(NameIdentifier ident, boolean force) "Catalog %s has schemas, please drop them first or use force option", ident); } - if (isManagedStorageCatalog(catalogWrapper)) { + // Drop filesets via FilesetDispatcher first so each fileset cleans its own + // write-through secrets. Do not snapshot/delete fileset secrets here. + for (SchemaEntity schema : schemaEntities) { + NameIdentifier schemaIdent = schema.nameIdentifier(); + Namespace filesetNs = + Namespace.of( + schemaIdent.namespace().level(0), + schemaIdent.namespace().level(1), + schemaIdent.name()); + List filesets = + store.list(filesetNs, FilesetEntity.class, EntityType.FILESET); + for (FilesetEntity fileset : filesets) { + filesetDispatcher.dropFileset( + ofFileset( + filesetNs.level(0), + filesetNs.level(1), + filesetNs.level(2), + fileset.name())); + } + } + + // Snapshot schema properties (write-through secret URNs) before entities are removed. + // store.delete(cascade) only soft-deletes meta rows and does not call secret providers. + List> schemaSecretPropertySnapshots = new ArrayList<>(); + for (SchemaEntity schema : schemaEntities) { + schemaSecretPropertySnapshots.add(copyProperties(schema.properties())); + } + + boolean managedStorage = isManagedStorageCatalog(catalogWrapper); + if (managedStorage) { // For managed catalog, we need to call drop schema API to drop the underlying // entities as well as the related resource first. Directly deleting the metadata from - // the store is not enough. - schemaEntities.forEach( - schema -> { - try { - catalogWrapper.doWithSchemaOps( - ops -> ops.dropSchema(schema.nameIdentifier(), true)); - } catch (Exception e) { - LOG.warn("Failed to drop schema {}", schema.nameIdentifier()); - throw new RuntimeException( - "Failed to drop schema " + schema.nameIdentifier(), e); - } - }); + // the store is not enough. Delete schema secrets only after a successful drop. + for (int i = 0; i < schemaEntities.size(); i++) { + SchemaEntity schema = schemaEntities.get(i); + try { + catalogWrapper.doWithSchemaOps( + ops -> ops.dropSchema(schema.nameIdentifier(), true)); + secretManager.deleteSecretsFromProperties(schemaSecretPropertySnapshots.get(i)); + } catch (Exception e) { + LOG.warn("Failed to drop schema {}", schema.nameIdentifier()); + throw new RuntimeException("Failed to drop schema " + schema.nameIdentifier(), e); + } + } } // Finally, delete the catalog entity as well as all its sub-entities from the store. // Invalidate after store.delete() to prevent a background thread from repopulating // the cache with stale data between invalidate and delete. + Map catalogProperties = + catalogWrapper.catalog().entity().getProperties(); boolean deleted = store.delete(ident, EntityType.CATALOG, true); if (deleted) { markLocalMutation(ident); + // Unmanaged / mixed: schemas were removed only via store cascade — clean their + // secrets now. Managed path already cleaned per successful dropSchema above. + // Fileset secrets were already cleaned via FilesetDispatcher above. + if (!managedStorage) { + for (Map schemaProperties : schemaSecretPropertySnapshots) { + secretManager.deleteSecretsFromProperties(schemaProperties); + } + } + secretManager.deleteSecretsFromProperties(catalogProperties); } catalogCache.invalidate(ident); return deleted; @@ -992,6 +1079,12 @@ public boolean dropCatalog(NameIdentifier ident, boolean force) }); } + private static Map copyProperties(Map properties) { + return properties == null || properties.isEmpty() + ? Collections.emptyMap() + : new HashMap<>(properties); + } + /** * Check if the given list of schema entities contains any currently existing user-created * schemas. @@ -1304,7 +1397,6 @@ private BaseCatalog createBaseCatalog(IsolatedClassLoader classLoader, Catalo // Load Catalog class instance BaseCatalog catalog = createCatalogInstance(classLoader, entity.getProvider()); // Resolve secret URNs to plaintext for connector init only; entity storage keeps URNs. - // Fileset FS merge assumes catalog conf is already plaintext at this boundary. catalog .withCatalogConf(secretManager.toPlaintextProperties(entity.getProperties())) .withCatalogEntity(entity); diff --git a/core/src/main/java/org/apache/gravitino/catalog/CatalogNormalizeDispatcher.java b/core/src/main/java/org/apache/gravitino/catalog/CatalogNormalizeDispatcher.java index 56c75a63180..84e2ecbfcb1 100644 --- a/core/src/main/java/org/apache/gravitino/catalog/CatalogNormalizeDispatcher.java +++ b/core/src/main/java/org/apache/gravitino/catalog/CatalogNormalizeDispatcher.java @@ -22,6 +22,7 @@ import com.google.common.collect.ImmutableSet; import java.util.Arrays; +import java.util.Collections; import java.util.Map; import java.util.Set; import org.apache.gravitino.Catalog; @@ -34,6 +35,8 @@ import org.apache.gravitino.exceptions.NoSuchCatalogException; import org.apache.gravitino.exceptions.NoSuchMetalakeException; import org.apache.gravitino.exceptions.NonEmptyEntityException; +import org.apache.gravitino.secret.SecretBinding; +import org.apache.gravitino.secret.SecretReference; public class CatalogNormalizeDispatcher implements CatalogDispatcher { private static final Set RESERVED_WORDS = @@ -84,8 +87,23 @@ public Catalog createCatalog( String comment, Map properties) throws NoSuchMetalakeException, CatalogAlreadyExistsException { + return createCatalog( + ident, type, provider, comment, properties, Collections.emptyMap(), Collections.emptyMap()); + } + + @Override + public Catalog createCatalog( + NameIdentifier ident, + Catalog.Type type, + String provider, + String comment, + Map properties, + Map secretBindings, + Map secretReferences) + throws NoSuchMetalakeException, CatalogAlreadyExistsException { validateCatalogName(ident.name()); - return dispatcher.createCatalog(ident, type, provider, comment, properties); + return dispatcher.createCatalog( + ident, type, provider, comment, properties, secretBindings, secretReferences); } @Override diff --git a/core/src/main/java/org/apache/gravitino/catalog/FilesetOperationDispatcher.java b/core/src/main/java/org/apache/gravitino/catalog/FilesetOperationDispatcher.java index b8fe860afe5..2587efe69e7 100644 --- a/core/src/main/java/org/apache/gravitino/catalog/FilesetOperationDispatcher.java +++ b/core/src/main/java/org/apache/gravitino/catalog/FilesetOperationDispatcher.java @@ -18,10 +18,12 @@ */ package org.apache.gravitino.catalog; +import static org.apache.gravitino.Entity.EntityType.FILESET; import static org.apache.gravitino.catalog.PropertiesMetadataHelpers.validatePropertyForCreate; import static org.apache.gravitino.utils.NameIdentifierUtil.getCatalogIdentifier; import java.util.Arrays; +import java.util.HashMap; import java.util.List; import java.util.Map; import org.apache.gravitino.EntityStore; @@ -39,6 +41,7 @@ import org.apache.gravitino.file.FilesetChange; import org.apache.gravitino.lock.LockType; import org.apache.gravitino.lock.TreeLockUtils; +import org.apache.gravitino.meta.FilesetEntity; import org.apache.gravitino.secret.SecretBinding; import org.apache.gravitino.secret.SecretManager; import org.apache.gravitino.secret.SecretMaterial; @@ -154,7 +157,8 @@ public Fileset createMultipleLocationFileset( throws NoSuchSchemaException, FilesetAlreadyExistsException { NameIdentifier catalogIdent = getCatalogIdentifier(ident); long uid = idGenerator.nextId(); - Map entityProperties = SecretPropertyUtils.copyEntityProperties(properties); + Map entityProperties = + SecretPropertyUtils.copyEntityProperties(properties, secretBindings, secretReferences); List secretMaterials = secretManager.assembleSecretMaterials( properties, entityProperties, "fileset", uid, secretBindings, secretReferences); @@ -167,7 +171,6 @@ public Fileset createMultipleLocationFileset( return null; }), IllegalArgumentException.class); - secretManager.writeSecrets(secretMaterials); StringIdentifier stringId = StringIdentifier.fromId(uid); // Same split as CatalogManager: create/storage properties keep secret URNs. Connectors that // need plaintext for runtime (e.g. Fileset FS) resolve at the conf boundary — see @@ -175,34 +178,41 @@ public Fileset createMultipleLocationFileset( Map updatedProperties = StringIdentifier.newPropertiesWithId(stringId, entityProperties); - try { - Fileset createdFileset = - TreeLockUtils.doWithTreeLock( - // Lock at fileset level (not schema level) to allow concurrent fileset creation. - // Trade-off: listFilesets() may temporarily miss in-progress creations until - // complete. - ident, - LockType.WRITE, - () -> - doWithCatalog( - catalogIdent, - c -> - c.doWithFilesetOps( - f -> - f.createMultipleLocationFileset( - ident, comment, type, storageLocations, updatedProperties)), - NoSuchSchemaException.class, - FilesetAlreadyExistsException.class)); - return EntityCombinedFileset.of(createdFileset) - .withHiddenProperties( - getHiddenPropertyNames( - catalogIdent, - HasPropertyMetadata::filesetPropertiesMetadata, - createdFileset.properties())); - } catch (RuntimeException e) { - secretManager.rollbackSecrets(secretMaterials); - throw e; - } + return TreeLockUtils.doWithTreeLock( + // Lock at fileset level (not schema level) to allow concurrent fileset creation. + // Trade-off: listFilesets() may temporarily miss in-progress creations until + // complete. + ident, + LockType.WRITE, + () -> { + // Same pattern as CatalogManager.createCatalog: writeSecrets inside the locked try; + // only roll back when the underlying create did not succeed (needClean stays true). + boolean needClean = true; + try { + secretManager.writeSecrets(secretMaterials); + Fileset createdFileset = + doWithCatalog( + catalogIdent, + c -> + c.doWithFilesetOps( + f -> + f.createMultipleLocationFileset( + ident, comment, type, storageLocations, updatedProperties)), + NoSuchSchemaException.class, + FilesetAlreadyExistsException.class); + needClean = false; + return EntityCombinedFileset.of(createdFileset) + .withHiddenProperties( + getHiddenPropertyNames( + catalogIdent, + HasPropertyMetadata::filesetPropertiesMetadata, + createdFileset.properties())); + } finally { + if (needClean) { + secretManager.rollbackSecrets(secretMaterials); + } + } + }); } /** @@ -266,17 +276,26 @@ public boolean dropFileset(NameIdentifier ident) { LockType.WRITE, () -> { NameIdentifier catalogIdent = getCatalogIdentifier(ident); - // Capture properties (including write-through secret URNs) before drop. - Map filesetProperties; - try { - Fileset fileset = - doWithCatalog( - catalogIdent, - c -> c.doWithFilesetOps(f -> f.loadFileset(ident)), - NoSuchFilesetException.class); - filesetProperties = fileset.properties(); - } catch (NoSuchFilesetException e) { - return false; + // Secret URNs live on FilesetEntity in the store (catalog loadFileset may omit them). + Map filesetProperties = new HashMap<>(); + FilesetEntity filesetEntity = getEntity(ident, FILESET, FilesetEntity.class); + if (filesetEntity != null + && filesetEntity.properties() != null + && !filesetEntity.properties().isEmpty()) { + filesetProperties = new HashMap<>(filesetEntity.properties()); + } else { + try { + Fileset fileset = + doWithCatalog( + catalogIdent, + c -> c.doWithFilesetOps(f -> f.loadFileset(ident)), + NoSuchFilesetException.class); + if (fileset.properties() != null) { + filesetProperties = new HashMap<>(fileset.properties()); + } + } catch (NoSuchFilesetException e) { + return false; + } } boolean dropped = diff --git a/core/src/main/java/org/apache/gravitino/catalog/SchemaDispatcher.java b/core/src/main/java/org/apache/gravitino/catalog/SchemaDispatcher.java index 1b4ef6f178c..f9efb3737a2 100644 --- a/core/src/main/java/org/apache/gravitino/catalog/SchemaDispatcher.java +++ b/core/src/main/java/org/apache/gravitino/catalog/SchemaDispatcher.java @@ -19,7 +19,15 @@ package org.apache.gravitino.catalog; +import java.util.Collections; +import java.util.Map; +import org.apache.gravitino.NameIdentifier; +import org.apache.gravitino.Schema; import org.apache.gravitino.connector.SupportsSchemas; +import org.apache.gravitino.exceptions.NoSuchCatalogException; +import org.apache.gravitino.exceptions.SchemaAlreadyExistsException; +import org.apache.gravitino.secret.SecretBinding; +import org.apache.gravitino.secret.SecretReference; /** * {@code SchemaDispatcher} interface acts as a specialization of the {@link SupportsSchemas} @@ -27,4 +35,52 @@ * to dispatching or handling schema-related events or actions that are not covered by the standard * {@code SupportsSchemas} operations. */ -public interface SchemaDispatcher extends SupportsSchemas {} +public interface SchemaDispatcher extends SupportsSchemas { + + /** + * Create a schema in the catalog. + * + *

Delegates to {@link #createSchema(NameIdentifier, String, Map, Map, Map)} with empty secret + * maps. + * + * @param ident The name identifier of the schema. + * @param comment The comment of the schema. + * @param properties The properties of the schema. + * @return The created schema. + * @throws NoSuchCatalogException If the catalog does not exist. + * @throws SchemaAlreadyExistsException If the schema already exists. + */ + @Override + default Schema createSchema(NameIdentifier ident, String comment, Map properties) + throws NoSuchCatalogException, SchemaAlreadyExistsException { + return createSchema(ident, comment, properties, Collections.emptyMap(), Collections.emptyMap()); + } + + /** + * Create a schema in the catalog with optional secret maps. + * + *

The default implementation rejects create-time secrets. Implementations that support secrets + * must override this method. + * + * @param ident The name identifier of the schema. + * @param comment The comment of the schema. + * @param properties The properties of the schema. + * @param secretBindings optional property key → binding ({@code provider} + {@code plaintext}) + * for write-through + * @param secretReferences optional property key → secret locator ({@code provider} plus + * provider-specific attributes). + * @return The created schema. + * @throws NoSuchCatalogException If the catalog does not exist. + * @throws SchemaAlreadyExistsException If the schema already exists. + * @throws UnsupportedOperationException if create-time secrets are not supported + */ + default Schema createSchema( + NameIdentifier ident, + String comment, + Map properties, + Map secretBindings, + Map secretReferences) + throws NoSuchCatalogException, SchemaAlreadyExistsException { + throw new UnsupportedOperationException("Creating a schema with secrets is not supported"); + } +} diff --git a/core/src/main/java/org/apache/gravitino/catalog/SchemaNormalizeDispatcher.java b/core/src/main/java/org/apache/gravitino/catalog/SchemaNormalizeDispatcher.java index 564d7778209..31d77ef307c 100644 --- a/core/src/main/java/org/apache/gravitino/catalog/SchemaNormalizeDispatcher.java +++ b/core/src/main/java/org/apache/gravitino/catalog/SchemaNormalizeDispatcher.java @@ -22,6 +22,7 @@ import static org.apache.gravitino.catalog.CapabilityHelpers.applyCaseSensitive; import static org.apache.gravitino.catalog.CapabilityHelpers.getCapability; +import java.util.Collections; import java.util.Map; import org.apache.gravitino.NameIdentifier; import org.apache.gravitino.Namespace; @@ -32,6 +33,8 @@ import org.apache.gravitino.exceptions.NoSuchSchemaException; import org.apache.gravitino.exceptions.NonEmptySchemaException; import org.apache.gravitino.exceptions.SchemaAlreadyExistsException; +import org.apache.gravitino.secret.SecretBinding; +import org.apache.gravitino.secret.SecretReference; /** * Note on list operations: names returned by list methods (e.g. {@link #listSchemas(Namespace)}) @@ -73,7 +76,19 @@ public boolean schemaExists(NameIdentifier ident) { @Override public Schema createSchema(NameIdentifier ident, String comment, Map properties) throws NoSuchCatalogException, SchemaAlreadyExistsException { - return dispatcher.createSchema(normalizeNameIdentifier(ident), comment, properties); + return createSchema(ident, comment, properties, Collections.emptyMap(), Collections.emptyMap()); + } + + @Override + public Schema createSchema( + NameIdentifier ident, + String comment, + Map properties, + Map secretBindings, + Map secretReferences) + throws NoSuchCatalogException, SchemaAlreadyExistsException { + return dispatcher.createSchema( + normalizeNameIdentifier(ident), comment, properties, secretBindings, secretReferences); } @Override diff --git a/core/src/main/java/org/apache/gravitino/catalog/SchemaOperationDispatcher.java b/core/src/main/java/org/apache/gravitino/catalog/SchemaOperationDispatcher.java index 31c752b21bf..f88e02e99a7 100644 --- a/core/src/main/java/org/apache/gravitino/catalog/SchemaOperationDispatcher.java +++ b/core/src/main/java/org/apache/gravitino/catalog/SchemaOperationDispatcher.java @@ -18,11 +18,18 @@ */ package org.apache.gravitino.catalog; +import static org.apache.gravitino.Entity.EntityType.FILESET; import static org.apache.gravitino.Entity.EntityType.SCHEMA; import static org.apache.gravitino.catalog.PropertiesMetadataHelpers.validatePropertyForCreate; import static org.apache.gravitino.utils.NameIdentifierUtil.getCatalogIdentifier; +import static org.apache.gravitino.utils.NameIdentifierUtil.ofFileset; +import com.google.common.base.Preconditions; +import java.io.IOException; import java.time.Instant; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; import java.util.Map; import org.apache.gravitino.EntityAlreadyExistsException; import org.apache.gravitino.EntityStore; @@ -41,8 +48,13 @@ import org.apache.gravitino.lock.LockType; import org.apache.gravitino.lock.TreeLockUtils; import org.apache.gravitino.meta.AuditInfo; +import org.apache.gravitino.meta.FilesetEntity; import org.apache.gravitino.meta.SchemaEntity; +import org.apache.gravitino.secret.SecretBinding; import org.apache.gravitino.secret.SecretManager; +import org.apache.gravitino.secret.SecretMaterial; +import org.apache.gravitino.secret.SecretPropertyUtils; +import org.apache.gravitino.secret.SecretReference; import org.apache.gravitino.storage.IdGenerator; import org.apache.gravitino.utils.PrincipalUtils; import org.apache.gravitino.utils.SchemaEntityCleaner; @@ -53,6 +65,8 @@ public class SchemaOperationDispatcher extends OperationDispatcher implements Sc private static final Logger LOG = LoggerFactory.getLogger(SchemaOperationDispatcher.class); + private final FilesetDispatcher filesetDispatcher; + /** * Creates a new SchemaOperationDispatcher instance. * @@ -60,13 +74,16 @@ public class SchemaOperationDispatcher extends OperationDispatcher implements Sc * @param store The EntityStore instance to be used for schema operations. * @param idGenerator The IdGenerator instance to be used for schema operations. * @param secretManager The SecretManager instance to be used for secret operations. + * @param filesetDispatcher The fileset dispatcher used to drop filesets on cascade schema drop. */ public SchemaOperationDispatcher( CatalogManager catalogManager, EntityStore store, IdGenerator idGenerator, - SecretManager secretManager) { + SecretManager secretManager, + FilesetDispatcher filesetDispatcher) { super(catalogManager, store, idGenerator, secretManager); + this.filesetDispatcher = Preconditions.checkNotNull(filesetDispatcher); } /** @@ -103,81 +120,115 @@ public NameIdentifier[] listSchemas(Namespace namespace) throws NoSuchCatalogExc @Override public Schema createSchema(NameIdentifier ident, String comment, Map properties) throws NoSuchCatalogException, SchemaAlreadyExistsException { + return createSchema(ident, comment, properties, Collections.emptyMap(), Collections.emptyMap()); + } + + @Override + public Schema createSchema( + NameIdentifier ident, + String comment, + Map properties, + Map secretBindings, + Map secretReferences) + throws NoSuchCatalogException, SchemaAlreadyExistsException { NameIdentifier catalogIdent = getCatalogIdentifier(ident); + long uid = idGenerator.nextId(); + Map entityProperties = + SecretPropertyUtils.copyEntityProperties(properties, secretBindings, secretReferences); + List secretMaterials = + secretManager.assembleSecretMaterials( + properties, entityProperties, "schema", uid, secretBindings, secretReferences); doWithCatalog( catalogIdent, c -> c.doWithPropertiesMeta( p -> { - validatePropertyForCreate(p.schemaPropertiesMetadata(), properties); + validatePropertyForCreate(p.schemaPropertiesMetadata(), entityProperties); return null; }), IllegalArgumentException.class); - long uid = idGenerator.nextId(); // Add StringIdentifier to the properties, the specific catalog will handle this // StringIdentifier to make sure only when the operation is successful, the related // SchemaEntity will be visible. + // + // Same split as CatalogManager: create/storage properties keep secret URNs. Connectors that + // need plaintext for runtime (e.g. Fileset FS) resolve at the conf boundary — see + // FilesetCatalogOperations.mergeUpLevelConfigurations / CatalogManager.createBaseCatalog. StringIdentifier stringId = StringIdentifier.fromId(uid); Map updatedProperties = - StringIdentifier.newPropertiesWithId(stringId, properties); + StringIdentifier.newPropertiesWithId(stringId, entityProperties); return TreeLockUtils.doWithTreeLock( catalogIdent, LockType.WRITE, () -> { - // we do not retrieve the schema again (to obtain some values generated by underlying - // catalog) - // since some catalogs' API is async and the schema may not be created immediately - Schema schema = - doWithCatalog( - catalogIdent, - c -> c.doWithSchemaOps(s -> s.createSchema(ident, comment, updatedProperties)), - NoSuchCatalogException.class, - SchemaAlreadyExistsException.class); - - // If the Schema is maintained by the Gravitino's store, we don't have to store again. - boolean isManagedSchema = isManagedEntity(catalogIdent, Capability.Scope.SCHEMA); - if (isManagedSchema) { - return EntityCombinedSchema.of(schema) - .withHiddenProperties( - getHiddenPropertyNames( - catalogIdent, - HasPropertyMetadata::schemaPropertiesMetadata, - schema.properties())); - } + // Same pattern as CatalogManager.createCatalog: writeSecrets inside the locked try; + // only roll back when the underlying create did not succeed (needClean stays true). + boolean needClean = true; + try { + secretManager.writeSecrets(secretMaterials); + // we do not retrieve the schema again (to obtain some values generated by underlying + // catalog) + // since some catalogs' API is async and the schema may not be created immediately + Schema schema = + doWithCatalog( + catalogIdent, + c -> c.doWithSchemaOps(s -> s.createSchema(ident, comment, updatedProperties)), + NoSuchCatalogException.class, + SchemaAlreadyExistsException.class); + needClean = false; + + // If the Schema is maintained by the Gravitino's store, we don't have to store again. + boolean isManagedSchema = isManagedEntity(catalogIdent, Capability.Scope.SCHEMA); + if (isManagedSchema) { + return EntityCombinedSchema.of(schema) + .withHiddenProperties( + getHiddenPropertyNames( + catalogIdent, + HasPropertyMetadata::schemaPropertiesMetadata, + schema.properties())); + } - SchemaEntity schemaEntity = - SchemaEntity.builder() - .withId(uid) - .withName(ident.name()) - .withNamespace(ident.namespace()) - .withAuditInfo( - AuditInfo.builder() - .withCreator(PrincipalUtils.getCurrentPrincipal().getName()) - .withCreateTime(Instant.now()) - .build()) - .build(); + // Persist properties (including secret URNs) in the entity store so cleanup still works + // when the underlying catalog does not retain schema properties. + SchemaEntity schemaEntity = + SchemaEntity.builder() + .withId(uid) + .withName(ident.name()) + .withNamespace(ident.namespace()) + .withProperties(updatedProperties) + .withAuditInfo( + AuditInfo.builder() + .withCreator(PrincipalUtils.getCurrentPrincipal().getName()) + .withCreateTime(Instant.now()) + .build()) + .build(); + + try { + store.put(schemaEntity, true /* overwrite */); + } catch (Exception e) { + LOG.error(FormattedErrorMessages.STORE_OP_FAILURE, "put", ident, e); + return EntityCombinedSchema.of(schema) + .withHiddenProperties( + getHiddenPropertyNames( + catalogIdent, + HasPropertyMetadata::schemaPropertiesMetadata, + schema.properties())); + } - try { - store.put(schemaEntity, true /* overwrite */); - } catch (Exception e) { - LOG.error(FormattedErrorMessages.STORE_OP_FAILURE, "put", ident, e); - return EntityCombinedSchema.of(schema) + // Merge both the metadata from catalog operation and the metadata from entity store. + return EntityCombinedSchema.of(schema, schemaEntity) .withHiddenProperties( getHiddenPropertyNames( catalogIdent, HasPropertyMetadata::schemaPropertiesMetadata, schema.properties())); + } finally { + if (needClean) { + secretManager.rollbackSecrets(secretMaterials); + } } - - // Merge both the metadata from catalog operation and the metadata from entity store. - return EntityCombinedSchema.of(schema, schemaEntity) - .withHiddenProperties( - getHiddenPropertyNames( - catalogIdent, - HasPropertyMetadata::schemaPropertiesMetadata, - schema.properties())); }); } @@ -294,6 +345,8 @@ public Schema alterSchema(NameIdentifier ident, SchemaChange... changes) .withId(schemaEntity.id()) .withName(schemaEntity.name()) .withNamespace(ident.namespace()) + .withProperties( + propertiesForSchemaEntityAlter(schemaEntity, changes)) .withAuditInfo( AuditInfo.builder() .withCreator(schemaEntity.auditInfo().creator()) @@ -327,10 +380,38 @@ public Schema alterSchema(NameIdentifier ident, SchemaChange... changes) @Override public boolean dropSchema(NameIdentifier ident, boolean cascade) throws NonEmptySchemaException { NameIdentifier catalogIdent = getCatalogIdentifier(ident); + + // Cascade: drop filesets via FilesetDispatcher first so each fileset cleans its own + // write-through secrets. Do this before the catalog lock to avoid nested TreeLocks. + if (cascade) { + Namespace filesetNs = + Namespace.of(ident.namespace().level(0), ident.namespace().level(1), ident.name()); + List filesets; + try { + filesets = store.list(filesetNs, FilesetEntity.class, FILESET); + } catch (IOException e) { + throw new RuntimeException("Failed to list filesets under schema " + ident, e); + } + for (FilesetEntity fileset : filesets) { + filesetDispatcher.dropFileset( + ofFileset(filesetNs.level(0), filesetNs.level(1), filesetNs.level(2), fileset.name())); + } + } + return TreeLockUtils.doWithTreeLock( catalogIdent, LockType.WRITE, () -> { + // Schema secret URNs live on SchemaEntity in the store (catalog loadSchema may omit + // them). + Map schemaProperties = new HashMap<>(); + SchemaEntity schemaEntity = getEntity(ident, SCHEMA, SchemaEntity.class); + if (schemaEntity != null + && schemaEntity.properties() != null + && !schemaEntity.properties().isEmpty()) { + schemaProperties = new HashMap<>(schemaEntity.properties()); + } + boolean droppedFromCatalog = doWithCatalog( catalogIdent, @@ -341,6 +422,9 @@ public boolean dropSchema(NameIdentifier ident, boolean cascade) throws NonEmpty // For managed schema, we don't need to drop the schema from the store again. boolean isManagedSchema = isManagedEntity(catalogIdent, Capability.Scope.SCHEMA); if (isManagedSchema) { + if (droppedFromCatalog) { + secretManager.deleteSecretsFromProperties(schemaProperties); + } return droppedFromCatalog; } @@ -369,10 +453,33 @@ public boolean dropSchema(NameIdentifier ident, boolean cascade) throws NonEmpty catalogIdent, c -> c.doWithSchemaOps(s -> s.schemaExists(schemaIdent)), RuntimeException.class)); + if (droppedFromCatalog) { + secretManager.deleteSecretsFromProperties(schemaProperties); + } return droppedFromCatalog; }); } + /** + * Builds properties to persist on {@link SchemaEntity} after alter, matching catalog alter: start + * from existing entity properties and apply set/remove changes so write-through secret URNs are + * preserved when the underlying catalog omits them. + */ + private static Map propertiesForSchemaEntityAlter( + SchemaEntity existing, SchemaChange[] changes) { + Map newProps = + existing.properties() == null ? new HashMap<>() : new HashMap<>(existing.properties()); + for (SchemaChange change : changes) { + if (change instanceof SchemaChange.SetProperty) { + SchemaChange.SetProperty setProperty = (SchemaChange.SetProperty) change; + newProps.put(setProperty.getProperty(), setProperty.getValue()); + } else if (change instanceof SchemaChange.RemoveProperty) { + newProps.remove(((SchemaChange.RemoveProperty) change).getProperty()); + } + } + return newProps; + } + private void importSchema(NameIdentifier identifier) { EntityCombinedSchema schema = internalLoadSchema(identifier); if (schema.imported()) { @@ -406,6 +513,8 @@ private void importSchema(NameIdentifier identifier) { .withId(uid) .withName(identifier.name()) .withNamespace(identifier.namespace()) + .withProperties( + schema.properties() == null ? Collections.emptyMap() : schema.properties()) .withAuditInfo( AuditInfo.builder() .withCreator(schema.auditInfo().creator()) diff --git a/core/src/main/java/org/apache/gravitino/catalog/SupportsCatalogs.java b/core/src/main/java/org/apache/gravitino/catalog/SupportsCatalogs.java index c83c70cb822..044825588b9 100644 --- a/core/src/main/java/org/apache/gravitino/catalog/SupportsCatalogs.java +++ b/core/src/main/java/org/apache/gravitino/catalog/SupportsCatalogs.java @@ -18,6 +18,7 @@ */ package org.apache.gravitino.catalog; +import java.util.Collections; import java.util.Map; import org.apache.gravitino.Catalog; import org.apache.gravitino.CatalogChange; @@ -31,6 +32,8 @@ import org.apache.gravitino.exceptions.NoSuchCatalogException; import org.apache.gravitino.exceptions.NoSuchMetalakeException; import org.apache.gravitino.exceptions.NonEmptyEntityException; +import org.apache.gravitino.secret.SecretBinding; +import org.apache.gravitino.secret.SecretReference; /** * Interface for supporting catalogs. It includes methods for listing, loading, creating, altering @@ -88,6 +91,9 @@ default boolean catalogExists(NameIdentifier ident) { * catalog should be created. The short name should be the same as the {@link CatalogProvider} * interface provided. * + *

Delegates to {@link #createCatalog(NameIdentifier, Catalog.Type, String, String, Map, Map, + * Map)} with empty secret maps. + * * @param ident the identifier of the catalog. * @param type the type of the catalog. * @param comment the comment of the catalog. @@ -97,13 +103,48 @@ default boolean catalogExists(NameIdentifier ident) { * @throws NoSuchMetalakeException If the metalake does not exist. * @throws CatalogAlreadyExistsException If the catalog already exists. */ - Catalog createCatalog( + default Catalog createCatalog( NameIdentifier ident, Catalog.Type type, String provider, String comment, Map properties) - throws NoSuchMetalakeException, CatalogAlreadyExistsException; + throws NoSuchMetalakeException, CatalogAlreadyExistsException { + return createCatalog( + ident, type, provider, comment, properties, Collections.emptyMap(), Collections.emptyMap()); + } + + /** + * Create a catalog with optional secret maps. + * + *

The default implementation rejects create-time secrets. Implementations that support secrets + * must override this method. + * + * @param ident the identifier of the catalog. + * @param type the type of the catalog. + * @param comment the comment of the catalog. + * @param provider the provider of the catalog. + * @param properties the properties of the catalog. + * @param secretBindings optional property key → binding ({@code provider} + {@code plaintext}) + * for write-through + * @param secretReferences optional property key → secret locator ({@code provider} plus + * provider-specific attributes). + * @return The created catalog. + * @throws NoSuchMetalakeException If the metalake does not exist. + * @throws CatalogAlreadyExistsException If the catalog already exists. + * @throws UnsupportedOperationException if create-time secrets are not supported + */ + default Catalog createCatalog( + NameIdentifier ident, + Catalog.Type type, + String provider, + String comment, + Map properties, + Map secretBindings, + Map secretReferences) + throws NoSuchMetalakeException, CatalogAlreadyExistsException { + throw new UnsupportedOperationException("Creating a catalog with secrets is not supported"); + } /** * Alter a catalog with specified identifier. diff --git a/core/src/main/java/org/apache/gravitino/hook/CatalogHookDispatcher.java b/core/src/main/java/org/apache/gravitino/hook/CatalogHookDispatcher.java index 81d324856ea..0b9dc9e2ce2 100644 --- a/core/src/main/java/org/apache/gravitino/hook/CatalogHookDispatcher.java +++ b/core/src/main/java/org/apache/gravitino/hook/CatalogHookDispatcher.java @@ -18,6 +18,7 @@ */ package org.apache.gravitino.hook; +import java.util.Collections; import java.util.List; import java.util.Map; import org.apache.gravitino.Catalog; @@ -38,6 +39,8 @@ import org.apache.gravitino.exceptions.NoSuchCatalogException; import org.apache.gravitino.exceptions.NoSuchMetalakeException; import org.apache.gravitino.exceptions.NonEmptyEntityException; +import org.apache.gravitino.secret.SecretBinding; +import org.apache.gravitino.secret.SecretReference; import org.apache.gravitino.utils.NameIdentifierUtil; import org.apache.gravitino.utils.PrincipalUtils; import org.slf4j.Logger; @@ -79,7 +82,23 @@ public Catalog createCatalog( String comment, Map properties) throws NoSuchMetalakeException, CatalogAlreadyExistsException { - Catalog catalog = dispatcher.createCatalog(ident, type, provider, comment, properties); + return createCatalog( + ident, type, provider, comment, properties, Collections.emptyMap(), Collections.emptyMap()); + } + + @Override + public Catalog createCatalog( + NameIdentifier ident, + Catalog.Type type, + String provider, + String comment, + Map properties, + Map secretBindings, + Map secretReferences) + throws NoSuchMetalakeException, CatalogAlreadyExistsException { + Catalog catalog = + dispatcher.createCatalog( + ident, type, provider, comment, properties, secretBindings, secretReferences); try { // Set the creator as the owner of the catalog. diff --git a/core/src/main/java/org/apache/gravitino/hook/SchemaHookDispatcher.java b/core/src/main/java/org/apache/gravitino/hook/SchemaHookDispatcher.java index dae07ffbc6f..7bd644df16e 100644 --- a/core/src/main/java/org/apache/gravitino/hook/SchemaHookDispatcher.java +++ b/core/src/main/java/org/apache/gravitino/hook/SchemaHookDispatcher.java @@ -41,6 +41,8 @@ import org.apache.gravitino.exceptions.SchemaAlreadyExistsException; import org.apache.gravitino.lock.LockType; import org.apache.gravitino.lock.TreeLockUtils; +import org.apache.gravitino.secret.SecretBinding; +import org.apache.gravitino.secret.SecretReference; import org.apache.gravitino.utils.HierarchicalSchemaUtil; import org.apache.gravitino.utils.NameIdentifierUtil; import org.apache.gravitino.utils.PrincipalUtils; @@ -65,6 +67,17 @@ public NameIdentifier[] listSchemas(Namespace namespace) throws NoSuchCatalogExc @Override public Schema createSchema(NameIdentifier ident, String comment, Map properties) throws NoSuchCatalogException, SchemaAlreadyExistsException { + return createSchema(ident, comment, properties, Collections.emptyMap(), Collections.emptyMap()); + } + + @Override + public Schema createSchema( + NameIdentifier ident, + String comment, + Map properties, + Map secretBindings, + Map secretReferences) + throws NoSuchCatalogException, SchemaAlreadyExistsException { // The inner NormalizeDispatcher case-folds the schema name based on catalog capabilities, so // the entity is stored under the normalized identifier. Normalize here too so ownership is // attached to the identifiers the manager sees and ancestor probing matches stored names. @@ -91,7 +104,8 @@ public Schema createSchema(NameIdentifier ident, String comment, Map newAncestors = findMissingAncestors(normalizedIdent); - Schema schema = dispatcher.createSchema(ident, comment, properties); + Schema schema = + dispatcher.createSchema(ident, comment, properties, secretBindings, secretReferences); // Set the creator as the owner of the new schema and of any ancestors it created. This // mirrors IcebergNamespaceHookDispatcher.createNamespace so ownership-based diff --git a/core/src/main/java/org/apache/gravitino/listener/CatalogEventDispatcher.java b/core/src/main/java/org/apache/gravitino/listener/CatalogEventDispatcher.java index ce95aa2b6ab..4b8675d0daf 100644 --- a/core/src/main/java/org/apache/gravitino/listener/CatalogEventDispatcher.java +++ b/core/src/main/java/org/apache/gravitino/listener/CatalogEventDispatcher.java @@ -19,6 +19,7 @@ package org.apache.gravitino.listener; +import java.util.Collections; import java.util.Map; import org.apache.gravitino.Catalog; import org.apache.gravitino.CatalogChange; @@ -53,6 +54,8 @@ import org.apache.gravitino.listener.api.event.LoadCatalogFailureEvent; import org.apache.gravitino.listener.api.event.LoadCatalogPreEvent; import org.apache.gravitino.listener.api.info.CatalogInfo; +import org.apache.gravitino.secret.SecretBinding; +import org.apache.gravitino.secret.SecretReference; import org.apache.gravitino.utils.PrincipalUtils; /** @@ -137,12 +140,28 @@ public Catalog createCatalog( String comment, Map properties) throws NoSuchMetalakeException, CatalogAlreadyExistsException { + return createCatalog( + ident, type, provider, comment, properties, Collections.emptyMap(), Collections.emptyMap()); + } + + @Override + public Catalog createCatalog( + NameIdentifier ident, + Catalog.Type type, + String provider, + String comment, + Map properties, + Map secretBindings, + Map secretReferences) + throws NoSuchMetalakeException, CatalogAlreadyExistsException { CatalogInfo catalogInfo = new CatalogInfo(ident.name(), type, provider, comment, properties, null); eventBus.dispatchEvent( new CreateCatalogPreEvent(PrincipalUtils.getCurrentUserName(), ident, catalogInfo)); try { - Catalog catalog = dispatcher.createCatalog(ident, type, provider, comment, properties); + Catalog catalog = + dispatcher.createCatalog( + ident, type, provider, comment, properties, secretBindings, secretReferences); eventBus.dispatchEvent( new CreateCatalogEvent( PrincipalUtils.getCurrentUserName(), ident, new CatalogInfo(catalog))); diff --git a/core/src/main/java/org/apache/gravitino/listener/SchemaEventDispatcher.java b/core/src/main/java/org/apache/gravitino/listener/SchemaEventDispatcher.java index 179609d5846..e5ad603ebc5 100644 --- a/core/src/main/java/org/apache/gravitino/listener/SchemaEventDispatcher.java +++ b/core/src/main/java/org/apache/gravitino/listener/SchemaEventDispatcher.java @@ -19,6 +19,7 @@ package org.apache.gravitino.listener; +import java.util.Collections; import java.util.Map; import org.apache.gravitino.NameIdentifier; import org.apache.gravitino.Namespace; @@ -46,6 +47,8 @@ import org.apache.gravitino.listener.api.event.LoadSchemaFailureEvent; import org.apache.gravitino.listener.api.event.LoadSchemaPreEvent; import org.apache.gravitino.listener.api.info.SchemaInfo; +import org.apache.gravitino.secret.SecretBinding; +import org.apache.gravitino.secret.SecretReference; import org.apache.gravitino.utils.PrincipalUtils; /** @@ -96,11 +99,23 @@ public boolean schemaExists(NameIdentifier ident) { @Override public Schema createSchema(NameIdentifier ident, String comment, Map properties) throws NoSuchCatalogException, SchemaAlreadyExistsException { + return createSchema(ident, comment, properties, Collections.emptyMap(), Collections.emptyMap()); + } + + @Override + public Schema createSchema( + NameIdentifier ident, + String comment, + Map properties, + Map secretBindings, + Map secretReferences) + throws NoSuchCatalogException, SchemaAlreadyExistsException { SchemaInfo createSchemaRequest = new SchemaInfo(ident.name(), comment, properties, null); eventBus.dispatchEvent( new CreateSchemaPreEvent(PrincipalUtils.getCurrentUserName(), ident, createSchemaRequest)); try { - Schema schema = dispatcher.createSchema(ident, comment, properties); + Schema schema = + dispatcher.createSchema(ident, comment, properties, secretBindings, secretReferences); eventBus.dispatchEvent( new CreateSchemaEvent( PrincipalUtils.getCurrentUserName(), ident, new SchemaInfo(schema))); diff --git a/core/src/main/java/org/apache/gravitino/secret/SecretManager.java b/core/src/main/java/org/apache/gravitino/secret/SecretManager.java index 3a989209df2..46abd86f4d2 100644 --- a/core/src/main/java/org/apache/gravitino/secret/SecretManager.java +++ b/core/src/main/java/org/apache/gravitino/secret/SecretManager.java @@ -127,7 +127,8 @@ public void checkSecretKeys( * is the mutable map that will be stored (may already contain merged catalog conf). * * @param properties properties used for key uniqueness checks (may be null) - * @param targetProperties mutable properties that receive URN values + * @param targetProperties mutable properties that receive URN values (may be null only when both + * secret maps are null or empty) * @param entityType {@code catalog}, {@code schema}, or {@code fileset} * @param entityId stable numeric entity id * @param secretBindings property key → write-through binding (may be null) @@ -137,12 +138,17 @@ public void checkSecretKeys( */ public List assembleSecretMaterials( @Nullable Map properties, - Map targetProperties, + @Nullable Map targetProperties, String entityType, long entityId, @Nullable Map secretBindings, @Nullable Map secretReferences) { checkSecretKeys(properties, secretBindings, secretReferences); + if (!SecretPropertyUtils.hasSecretMaps(secretBindings, secretReferences)) { + return List.of(); + } + Preconditions.checkArgument( + targetProperties != null, "targetProperties must not be null when secrets are present"); Map bindings = secretBindings == null ? Map.of() : secretBindings; Map references = secretReferences == null ? Map.of() : secretReferences; diff --git a/core/src/main/java/org/apache/gravitino/secret/SecretPropertyUtils.java b/core/src/main/java/org/apache/gravitino/secret/SecretPropertyUtils.java index b7988cd6dcc..77fa159e70a 100644 --- a/core/src/main/java/org/apache/gravitino/secret/SecretPropertyUtils.java +++ b/core/src/main/java/org/apache/gravitino/secret/SecretPropertyUtils.java @@ -49,15 +49,40 @@ public static boolean isSecretProperty(@Nullable String key, @Nullable String va } /** - * Returns a mutable copy of a property map for create-time assembly. + * Returns whether either secret map has at least one entry. * - *

{@code null} becomes an empty {@link HashMap}; otherwise returns a new {@link HashMap} copy. - * Used for request properties and for merged catalog conf (which may be unmodifiable). + * @param secretBindings write-through bindings (may be null) + * @param secretReferences secret locators (may be null) + * @return true when at least one secret map is non-empty + */ + public static boolean hasSecretMaps( + @Nullable Map secretBindings, @Nullable Map secretReferences) { + return (secretBindings != null && !secretBindings.isEmpty()) + || (secretReferences != null && !secretReferences.isEmpty()); + } + + /** + * Returns a mutable property map for create-time assembly, or {@code null} when the caller + * supplied no properties and no secrets. + * + *

When {@code properties} is {@code null} and both secret maps are null or empty, returns + * {@code null} so {@code validatePropertyForCreate} can skip required-key checks (historical + * behavior). When secrets are present but {@code properties} is null, returns an empty {@link + * HashMap} for URN assembly. Otherwise returns a new {@link HashMap} copy of {@code properties}. * * @param properties property map to copy (may be null) - * @return a mutable property map, never null + * @param secretBindings write-through bindings (may be null) + * @param secretReferences secret locators (may be null) + * @return a mutable property map, or null when there are no properties and no secrets */ - public static Map copyEntityProperties(@Nullable Map properties) { + @Nullable + public static Map copyEntityProperties( + @Nullable Map properties, + @Nullable Map secretBindings, + @Nullable Map secretReferences) { + if (properties == null && !hasSecretMaps(secretBindings, secretReferences)) { + return null; + } return properties == null ? new HashMap<>() : new HashMap<>(properties); } diff --git a/core/src/test/java/org/apache/gravitino/catalog/TestCatalogManager.java b/core/src/test/java/org/apache/gravitino/catalog/TestCatalogManager.java index c55c4044692..23b7a9abe7e 100644 --- a/core/src/test/java/org/apache/gravitino/catalog/TestCatalogManager.java +++ b/core/src/test/java/org/apache/gravitino/catalog/TestCatalogManager.java @@ -37,8 +37,10 @@ import java.time.Instant; import java.util.List; import java.util.Map; +import java.util.Properties; import java.util.Set; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicReference; import org.apache.commons.lang3.reflect.FieldUtils; import org.apache.gravitino.Catalog; @@ -65,7 +67,14 @@ import org.apache.gravitino.meta.CatalogEntity; import org.apache.gravitino.meta.SchemaEntity; import org.apache.gravitino.meta.SchemaVersion; +import org.apache.gravitino.secret.SecretBinding; +import org.apache.gravitino.secret.SecretConstants; import org.apache.gravitino.secret.SecretManager; +import org.apache.gravitino.secret.SecretPropertyUtils; +import org.apache.gravitino.secret.SecretProviderRegistry; +import org.apache.gravitino.secret.SecretUrn; +import org.apache.gravitino.secret.memory.InMemorySecretsProvider; +import org.apache.gravitino.storage.IdGenerator; import org.apache.gravitino.storage.RandomIdGenerator; import org.apache.gravitino.storage.memory.TestMemoryEntityStore; import org.apache.gravitino.storage.memory.TestMemoryEntityStore.InMemoryEntityStore; @@ -1340,4 +1349,141 @@ private void testProperties(Map expectedProps, Map props = + ImmutableMap.of( + "provider", + "test", + PROPERTY_KEY1, + "value1", + PROPERTY_KEY2, + "value2", + PROPERTY_KEY5_PREFIX + "1", + "value3"); + Map bindings = + Map.of(PROPERTY_KEY4, new SecretBinding("memory", "s3cr3t")); + + Catalog catalog = + manager.createCatalog( + ident, Catalog.Type.RELATIONAL, provider, "comment", props, bindings, Map.of()); + Assertions.assertFalse(catalog.properties().containsKey(PROPERTY_KEY4)); + + String urn = + entityStore + .get(ident, EntityType.CATALOG, CatalogEntity.class) + .getProperties() + .get(PROPERTY_KEY4); + Assertions.assertTrue(SecretPropertyUtils.isSecretProperty(PROPERTY_KEY4, urn)); + Assertions.assertEquals("s3cr3t", secrets.readSecret(SecretUrn.parse(urn))); + + Assertions.assertTrue(manager.dropCatalog(ident, true)); + Assertions.assertThrows( + IllegalArgumentException.class, () -> secrets.readSecret(SecretUrn.parse(urn))); + manager.close(); + } + } + + @Test + void testCreateSecretRollback() throws Exception { + try (SecretManager secrets = memorySecretManager()) { + AtomicLong nextId = new AtomicLong(4242L); + IdGenerator ids = nextId::getAndIncrement; + CatalogManager manager = Mockito.spy(new CatalogManager(config, entityStore, ids, secrets)); + NameIdentifier ident = NameIdentifier.of("metalake", "secret_catalog_fail"); + Map props = + ImmutableMap.of( + "provider", + "test", + PROPERTY_KEY1, + "value1", + PROPERTY_KEY2, + "value2", + PROPERTY_KEY5_PREFIX + "1", + "value3"); + Mockito.doThrow(new RuntimeException("init failed")) + .when(manager) + .createCatalogWrapper(any(CatalogEntity.class), any()); + + SecretUrn urn = + SecretUrn.buildWriteThrough( + "memory", + Map.of( + SecretConstants.ATTR_ENTITY_TYPE, "catalog", + SecretConstants.ATTR_ENTITY_ID, "4242", + SecretConstants.ATTR_PROPERTY_KEY, PROPERTY_KEY4)); + + Assertions.assertThrows( + RuntimeException.class, + () -> + manager.createCatalog( + ident, + Catalog.Type.RELATIONAL, + provider, + "comment", + props, + Map.of(PROPERTY_KEY4, new SecretBinding("memory", "x")), + Map.of())); + Assertions.assertFalse(entityStore.exists(ident, EntityType.CATALOG)); + Assertions.assertThrows(IllegalArgumentException.class, () -> secrets.readSecret(urn)); + manager.close(); + } + } + + @Test + void testCreateSecretSkippedNoMetalake() throws Exception { + try (SecretManager secrets = memorySecretManager()) { + AtomicLong nextId = new AtomicLong(4343L); + IdGenerator ids = nextId::getAndIncrement; + CatalogManager manager = new CatalogManager(config, entityStore, ids, secrets); + NameIdentifier ident = NameIdentifier.of("missing_metalake", "secret_catalog"); + SecretUrn urn = + SecretUrn.buildWriteThrough( + "memory", + Map.of( + SecretConstants.ATTR_ENTITY_TYPE, "catalog", + SecretConstants.ATTR_ENTITY_ID, "4343", + SecretConstants.ATTR_PROPERTY_KEY, PROPERTY_KEY4)); + + Assertions.assertThrows( + NoSuchMetalakeException.class, + () -> + manager.createCatalog( + ident, + Catalog.Type.RELATIONAL, + provider, + "comment", + ImmutableMap.of( + "provider", + "test", + PROPERTY_KEY1, + "value1", + PROPERTY_KEY2, + "value2", + PROPERTY_KEY5_PREFIX + "1", + "value3"), + Map.of(PROPERTY_KEY4, new SecretBinding("memory", "x")), + Map.of())); + Assertions.assertThrows(IllegalArgumentException.class, () -> secrets.readSecret(urn)); + manager.close(); + } + } + + private static SecretManager memorySecretManager() { + Config c = new Config(false) {}; + Properties p = new Properties(); + p.setProperty(SecretProviderRegistry.GRAVITINO_SECRET_PROVIDERS, "memory"); + p.setProperty( + SecretProviderRegistry.GRAVITINO_SECRET_PROVIDER_PREFIX + + "memory." + + SecretProviderRegistry.CLASS_NAME, + InMemorySecretsProvider.class.getName()); + c.loadFromProperties(p); + return new SecretManager(c); + } } diff --git a/core/src/test/java/org/apache/gravitino/catalog/TestFilesetOperationDispatcher.java b/core/src/test/java/org/apache/gravitino/catalog/TestFilesetOperationDispatcher.java index b0a17ccd488..980fffe9d76 100644 --- a/core/src/test/java/org/apache/gravitino/catalog/TestFilesetOperationDispatcher.java +++ b/core/src/test/java/org/apache/gravitino/catalog/TestFilesetOperationDispatcher.java @@ -56,10 +56,11 @@ public class TestFilesetOperationDispatcher extends TestOperationDispatcher { @BeforeAll public static void initialize() throws IOException { - schemaOperationDispatcher = - new SchemaOperationDispatcher(catalogManager, entityStore, idGenerator, secretManager); filesetOperationDispatcher = new FilesetOperationDispatcher(catalogManager, entityStore, idGenerator, secretManager); + schemaOperationDispatcher = + new SchemaOperationDispatcher( + catalogManager, entityStore, idGenerator, secretManager, filesetOperationDispatcher); } public static FilesetOperationDispatcher getFilesetOperationDispatcher() { @@ -332,7 +333,7 @@ public void testCreateWithSecrets() throws Exception { IdGenerator ids = nextId::getAndIncrement; FilesetOperationDispatcher filesets = new FilesetOperationDispatcher(catalogManager, entityStore, ids, secrets); - new SchemaOperationDispatcher(catalogManager, entityStore, ids, secrets) + new SchemaOperationDispatcher(catalogManager, entityStore, ids, secrets, filesets) .createSchema( NameIdentifier.of(metalake, catalog, "schema_secret_fileset"), "comment", diff --git a/core/src/test/java/org/apache/gravitino/catalog/TestModelOperationDispatcher.java b/core/src/test/java/org/apache/gravitino/catalog/TestModelOperationDispatcher.java index 06a43a0e8ee..793bb6b24f9 100644 --- a/core/src/test/java/org/apache/gravitino/catalog/TestModelOperationDispatcher.java +++ b/core/src/test/java/org/apache/gravitino/catalog/TestModelOperationDispatcher.java @@ -67,7 +67,12 @@ public static void initialize() throws IOException, IllegalAccessException { modelOperationDispatcher = new ModelOperationDispatcher(catalogManager, entityStore, idGenerator, secretManager); schemaOperationDispatcher = - new SchemaOperationDispatcher(catalogManager, entityStore, idGenerator, secretManager); + new SchemaOperationDispatcher( + catalogManager, + entityStore, + idGenerator, + secretManager, + Mockito.mock(FilesetDispatcher.class)); } @Test diff --git a/core/src/test/java/org/apache/gravitino/catalog/TestPartitionOperationDispatcher.java b/core/src/test/java/org/apache/gravitino/catalog/TestPartitionOperationDispatcher.java index 25a9ea6ebba..fb402360774 100644 --- a/core/src/test/java/org/apache/gravitino/catalog/TestPartitionOperationDispatcher.java +++ b/core/src/test/java/org/apache/gravitino/catalog/TestPartitionOperationDispatcher.java @@ -79,7 +79,8 @@ public static void initialize() throws IllegalAccessException { protected static void prepareTable() throws IllegalAccessException { schemaOperationDispatcher = - new SchemaOperationDispatcher(catalogManager, entityStore, idGenerator, secretManager); + new SchemaOperationDispatcher( + catalogManager, entityStore, idGenerator, secretManager, mock(FilesetDispatcher.class)); tableOperationDispatcher = new TableOperationDispatcher(catalogManager, entityStore, idGenerator, secretManager); partitionOperationDispatcher = diff --git a/core/src/test/java/org/apache/gravitino/catalog/TestSchemaOperationDispatcher.java b/core/src/test/java/org/apache/gravitino/catalog/TestSchemaOperationDispatcher.java index bb2b0d0576c..9a2678f30c0 100644 --- a/core/src/test/java/org/apache/gravitino/catalog/TestSchemaOperationDispatcher.java +++ b/core/src/test/java/org/apache/gravitino/catalog/TestSchemaOperationDispatcher.java @@ -36,6 +36,7 @@ import java.util.HashMap; import java.util.Map; import java.util.Optional; +import java.util.Properties; import org.apache.commons.lang3.reflect.FieldUtils; import org.apache.gravitino.Config; import org.apache.gravitino.Configs; @@ -48,9 +49,17 @@ import org.apache.gravitino.SchemaChange; import org.apache.gravitino.auth.AuthConstants; import org.apache.gravitino.exceptions.NoSuchEntityException; +import org.apache.gravitino.exceptions.SchemaAlreadyExistsException; import org.apache.gravitino.lock.LockManager; import org.apache.gravitino.meta.AuditInfo; import org.apache.gravitino.meta.SchemaEntity; +import org.apache.gravitino.secret.SecretBinding; +import org.apache.gravitino.secret.SecretConstants; +import org.apache.gravitino.secret.SecretManager; +import org.apache.gravitino.secret.SecretPropertyUtils; +import org.apache.gravitino.secret.SecretProviderRegistry; +import org.apache.gravitino.secret.SecretUrn; +import org.apache.gravitino.secret.memory.InMemorySecretsProvider; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; @@ -62,7 +71,8 @@ public class TestSchemaOperationDispatcher extends TestOperationDispatcher { @BeforeAll public static void initialize() throws IOException, IllegalAccessException { dispatcher = - new SchemaOperationDispatcher(catalogManager, entityStore, idGenerator, secretManager); + new SchemaOperationDispatcher( + catalogManager, entityStore, idGenerator, secretManager, mock(FilesetDispatcher.class)); Config config = mock(Config.class); doReturn(100000L).when(config).get(Configs.TREE_LOCK_MAX_NODE_IN_MEMORY); @@ -413,6 +423,56 @@ public void testDropHierarchicalSchemaKeepsAncestorsThatStillExist() throws IOEx Assertions.assertTrue(entityStore.exists(ancestorA, SCHEMA)); } + @Test + public void testCreateWithSecrets() throws Exception { + try (SecretManager secrets = memorySecretManager()) { + SchemaOperationDispatcher d = + new SchemaOperationDispatcher( + catalogManager, entityStore, idGenerator, secrets, mock(FilesetDispatcher.class)); + NameIdentifier ident = NameIdentifier.of(metalake, catalog, "schema_secret_1"); + Map props = ImmutableMap.of("k1", "v1"); + Map bindings = Map.of("k2", new SecretBinding("memory", "s3cr3t")); + + Schema schema = d.createSchema(ident, "comment", props, bindings, Map.of()); + Assertions.assertFalse(schema.properties().containsKey("k2")); + + Schema stored = + catalogManager + .loadCatalogAndWrap(NameIdentifier.of(metalake, catalog)) + .doWithSchemaOps(ops -> ops.loadSchema(ident)); + Assertions.assertTrue( + SecretPropertyUtils.isSecretProperty("k2", stored.properties().get("k2"))); + + SchemaEntity entity = entityStore.get(ident, SCHEMA, SchemaEntity.class); + SecretUrn urn = + SecretUrn.buildWriteThrough( + "memory", + Map.of( + SecretConstants.ATTR_ENTITY_TYPE, "schema", + SecretConstants.ATTR_ENTITY_ID, String.valueOf(entity.id()), + SecretConstants.ATTR_PROPERTY_KEY, "k2")); + Assertions.assertEquals("s3cr3t", secrets.readSecret(urn)); + Assertions.assertThrows( + SchemaAlreadyExistsException.class, + () -> d.createSchema(ident, "comment", props, bindings, Map.of())); + Assertions.assertTrue(d.dropSchema(ident, false)); + Assertions.assertThrows(IllegalArgumentException.class, () -> secrets.readSecret(urn)); + } + } + + private static SecretManager memorySecretManager() { + Config c = new Config(false) {}; + Properties p = new Properties(); + p.setProperty(SecretProviderRegistry.GRAVITINO_SECRET_PROVIDERS, "memory"); + p.setProperty( + SecretProviderRegistry.GRAVITINO_SECRET_PROVIDER_PREFIX + + "memory." + + SecretProviderRegistry.CLASS_NAME, + InMemorySecretsProvider.class.getName()); + c.loadFromProperties(p); + return new SecretManager(c); + } + private void putSchemaEntity(NameIdentifier ident) throws IOException { SchemaEntity entity = SchemaEntity.builder() diff --git a/core/src/test/java/org/apache/gravitino/catalog/TestTableOperationDispatcher.java b/core/src/test/java/org/apache/gravitino/catalog/TestTableOperationDispatcher.java index a834398c71f..d2d49a3f6c2 100644 --- a/core/src/test/java/org/apache/gravitino/catalog/TestTableOperationDispatcher.java +++ b/core/src/test/java/org/apache/gravitino/catalog/TestTableOperationDispatcher.java @@ -80,7 +80,8 @@ public class TestTableOperationDispatcher extends TestOperationDispatcher { @BeforeAll public static void initialize() throws IOException, IllegalAccessException { schemaOperationDispatcher = - new SchemaOperationDispatcher(catalogManager, entityStore, idGenerator, secretManager); + new SchemaOperationDispatcher( + catalogManager, entityStore, idGenerator, secretManager, mock(FilesetDispatcher.class)); tableOperationDispatcher = new TableOperationDispatcher( catalogManager, diff --git a/core/src/test/java/org/apache/gravitino/catalog/TestTopicOperationDispatcher.java b/core/src/test/java/org/apache/gravitino/catalog/TestTopicOperationDispatcher.java index 084f12132b8..b4778924a8f 100644 --- a/core/src/test/java/org/apache/gravitino/catalog/TestTopicOperationDispatcher.java +++ b/core/src/test/java/org/apache/gravitino/catalog/TestTopicOperationDispatcher.java @@ -66,7 +66,8 @@ public class TestTopicOperationDispatcher extends TestOperationDispatcher { @BeforeAll public static void initialize() throws IOException, IllegalAccessException { schemaOperationDispatcher = - new SchemaOperationDispatcher(catalogManager, entityStore, idGenerator, secretManager); + new SchemaOperationDispatcher( + catalogManager, entityStore, idGenerator, secretManager, mock(FilesetDispatcher.class)); topicOperationDispatcher = new TopicOperationDispatcher(catalogManager, entityStore, idGenerator, secretManager); diff --git a/core/src/test/java/org/apache/gravitino/catalog/TestViewOperationDispatcher.java b/core/src/test/java/org/apache/gravitino/catalog/TestViewOperationDispatcher.java index 26372f662e6..69fc4cbbac5 100644 --- a/core/src/test/java/org/apache/gravitino/catalog/TestViewOperationDispatcher.java +++ b/core/src/test/java/org/apache/gravitino/catalog/TestViewOperationDispatcher.java @@ -72,7 +72,8 @@ public class TestViewOperationDispatcher extends TestOperationDispatcher { @BeforeAll public static void initialize() throws IOException, IllegalAccessException { schemaOperationDispatcher = - new SchemaOperationDispatcher(catalogManager, entityStore, idGenerator, secretManager); + new SchemaOperationDispatcher( + catalogManager, entityStore, idGenerator, secretManager, mock(FilesetDispatcher.class)); viewOperationDispatcher = new ViewOperationDispatcher( catalogManager, diff --git a/core/src/test/java/org/apache/gravitino/hook/TestCatalogHookDispatcher.java b/core/src/test/java/org/apache/gravitino/hook/TestCatalogHookDispatcher.java index 1803646fc26..1d17ff01f3e 100644 --- a/core/src/test/java/org/apache/gravitino/hook/TestCatalogHookDispatcher.java +++ b/core/src/test/java/org/apache/gravitino/hook/TestCatalogHookDispatcher.java @@ -23,7 +23,7 @@ import static org.junit.jupiter.api.Assertions.assertTrue; import java.util.Arrays; -import java.util.Collections; +import java.util.Map; import org.apache.commons.lang3.reflect.FieldUtils; import org.apache.gravitino.Catalog; import org.apache.gravitino.GravitinoEnv; @@ -36,70 +36,48 @@ public class TestCatalogHookDispatcher { @Test - public void testCreateCatalogThrowsPostHookExceptionWhenRollbackSucceeds() throws Exception { - GravitinoEnv gravitinoEnv = GravitinoEnv.getInstance(); - Object originalOwnerDispatcher = FieldUtils.readField(gravitinoEnv, "ownerDispatcher", true); - Object originalFutureGrantManager = - FieldUtils.readField(gravitinoEnv, "futureGrantManager", true); - - CatalogDispatcher dispatcher = Mockito.mock(CatalogDispatcher.class); - Catalog catalog = Mockito.mock(Catalog.class); - NameIdentifier ident = NameIdentifier.of("metalake", "catalog"); - RuntimeException postHookException = new RuntimeException("post-hook failed"); - - OwnerDispatcher ownerDispatcher = Mockito.mock(OwnerDispatcher.class); - Mockito.doThrow(postHookException) - .when(ownerDispatcher) - .setOwner(Mockito.anyString(), Mockito.any(), Mockito.anyString(), Mockito.any()); - Mockito.when( - dispatcher.createCatalog( - Mockito.eq(ident), - Mockito.eq(Catalog.Type.RELATIONAL), - Mockito.eq("provider"), - Mockito.eq("comment"), - Mockito.anyMap())) - .thenReturn(catalog); - - FieldUtils.writeField(gravitinoEnv, "ownerDispatcher", ownerDispatcher, true); - FieldUtils.writeField(gravitinoEnv, "futureGrantManager", null, true); + public void testCreatePostHookRollback() throws Exception { + RuntimeException postHook = new RuntimeException("post-hook failed"); + runCreateWithPostHookFailure( + postHook, + null, + (dispatcher, ident, thrown) -> { + assertSame(postHook, thrown); + Mockito.verify(dispatcher).dropCatalog(ident, true); + }); + } - try { - CatalogHookDispatcher hookDispatcher = new CatalogHookDispatcher(dispatcher); - RuntimeException thrown = - assertThrowsExactly( - RuntimeException.class, - () -> - hookDispatcher.createCatalog( - ident, - Catalog.Type.RELATIONAL, - "provider", - "comment", - Collections.emptyMap())); - assertSame(postHookException, thrown); + @Test + public void testCreateRollbackSuppressed() throws Exception { + RuntimeException postHook = new RuntimeException("post-hook failed"); + RuntimeException rollback = new RuntimeException("rollback failed"); + runCreateWithPostHookFailure( + postHook, + rollback, + (dispatcher, ident, thrown) -> { + assertSame(postHook, thrown); + assertTrue(Arrays.stream(thrown.getSuppressed()).anyMatch(t -> t == rollback)); + Mockito.verify(dispatcher).dropCatalog(ident, true); + }); + } - Mockito.verify(dispatcher).dropCatalog(ident, true); - } finally { - FieldUtils.writeField(gravitinoEnv, "ownerDispatcher", originalOwnerDispatcher, true); - FieldUtils.writeField(gravitinoEnv, "futureGrantManager", originalFutureGrantManager, true); - } + @FunctionalInterface + private interface PostHookAssert { + void check(CatalogDispatcher dispatcher, NameIdentifier ident, RuntimeException thrown); } - @Test - public void testCreateCatalogRollbackExceptionDoesNotMaskPostHookException() throws Exception { - GravitinoEnv gravitinoEnv = GravitinoEnv.getInstance(); - Object originalOwnerDispatcher = FieldUtils.readField(gravitinoEnv, "ownerDispatcher", true); - Object originalFutureGrantManager = - FieldUtils.readField(gravitinoEnv, "futureGrantManager", true); + private static void runCreateWithPostHookFailure( + RuntimeException postHook, RuntimeException dropFailure, PostHookAssert asserts) + throws Exception { + GravitinoEnv env = GravitinoEnv.getInstance(); + Object savedOwner = FieldUtils.readField(env, "ownerDispatcher", true); + Object savedFutureGrant = FieldUtils.readField(env, "futureGrantManager", true); CatalogDispatcher dispatcher = Mockito.mock(CatalogDispatcher.class); - Catalog catalog = Mockito.mock(Catalog.class); NameIdentifier ident = NameIdentifier.of("metalake", "catalog"); - RuntimeException postHookException = new RuntimeException("post-hook failed"); - RuntimeException rollbackException = new RuntimeException("rollback failed"); - - OwnerDispatcher ownerDispatcher = Mockito.mock(OwnerDispatcher.class); - Mockito.doThrow(postHookException) - .when(ownerDispatcher) + OwnerDispatcher owner = Mockito.mock(OwnerDispatcher.class); + Mockito.doThrow(postHook) + .when(owner) .setOwner(Mockito.anyString(), Mockito.any(), Mockito.anyString(), Mockito.any()); Mockito.when( dispatcher.createCatalog( @@ -107,32 +85,28 @@ public void testCreateCatalogRollbackExceptionDoesNotMaskPostHookException() thr Mockito.eq(Catalog.Type.RELATIONAL), Mockito.eq("provider"), Mockito.eq("comment"), + Mockito.anyMap(), + Mockito.anyMap(), Mockito.anyMap())) - .thenReturn(catalog); - Mockito.doThrow(rollbackException).when(dispatcher).dropCatalog(ident, true); - - FieldUtils.writeField(gravitinoEnv, "ownerDispatcher", ownerDispatcher, true); - FieldUtils.writeField(gravitinoEnv, "futureGrantManager", null, true); + .thenReturn(Mockito.mock(Catalog.class)); + if (dropFailure != null) { + Mockito.doThrow(dropFailure).when(dispatcher).dropCatalog(ident, true); + } + FieldUtils.writeField(env, "ownerDispatcher", owner, true); + FieldUtils.writeField(env, "futureGrantManager", null, true); try { - CatalogHookDispatcher hookDispatcher = new CatalogHookDispatcher(dispatcher); + CatalogHookDispatcher hook = new CatalogHookDispatcher(dispatcher); RuntimeException thrown = assertThrowsExactly( RuntimeException.class, () -> - hookDispatcher.createCatalog( - ident, - Catalog.Type.RELATIONAL, - "provider", - "comment", - Collections.emptyMap())); - assertSame(postHookException, thrown); - assertTrue(Arrays.stream(thrown.getSuppressed()).anyMatch(t -> t == rollbackException)); - - Mockito.verify(dispatcher).dropCatalog(ident, true); + hook.createCatalog( + ident, Catalog.Type.RELATIONAL, "provider", "comment", Map.of())); + asserts.check(dispatcher, ident, thrown); } finally { - FieldUtils.writeField(gravitinoEnv, "ownerDispatcher", originalOwnerDispatcher, true); - FieldUtils.writeField(gravitinoEnv, "futureGrantManager", originalFutureGrantManager, true); + FieldUtils.writeField(env, "ownerDispatcher", savedOwner, true); + FieldUtils.writeField(env, "futureGrantManager", savedFutureGrant, true); } } } diff --git a/core/src/test/java/org/apache/gravitino/hook/TestSchemaHookDispatcher.java b/core/src/test/java/org/apache/gravitino/hook/TestSchemaHookDispatcher.java index 3e05c3c83ce..d5ebd8893df 100644 --- a/core/src/test/java/org/apache/gravitino/hook/TestSchemaHookDispatcher.java +++ b/core/src/test/java/org/apache/gravitino/hook/TestSchemaHookDispatcher.java @@ -22,11 +22,13 @@ import static org.apache.gravitino.Configs.TREE_LOCK_MAX_NODE_IN_MEMORY; import static org.apache.gravitino.Configs.TREE_LOCK_MIN_NODE_IN_MEMORY; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyBoolean; import static org.mockito.ArgumentMatchers.anyList; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.doReturn; import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -116,7 +118,7 @@ private static Config newLockConfig() { public void testCreateSchemaThrowsWhenSetOwnerFails() { NameIdentifier ident = NameIdentifier.of("test_metalake", "test_catalog", "test_schema"); Schema mockSchema = mock(Schema.class); - when(mockDispatcher.createSchema(any(), any(), any())).thenReturn(mockSchema); + when(mockDispatcher.createSchema(any(), any(), any(), any(), any())).thenReturn(mockSchema); doThrow(new RuntimeException("Set owner failed")) .when(mockOwnerDispatcher) @@ -127,7 +129,9 @@ public void testCreateSchemaThrowsWhenSetOwnerFails() { RuntimeException.class, () -> hookDispatcher.createSchema(ident, "comment", Collections.emptyMap())); Assertions.assertEquals("Set owner failed", thrown.getMessage()); - verify(mockDispatcher).createSchema(any(), any(), any()); + verify(mockDispatcher).createSchema(any(), any(), any(), any(), any()); + // Align with FilesetHookDispatcher / #12366: do not drop the schema when setOwner fails. + verify(mockDispatcher, never()).dropSchema(any(), anyBoolean()); } @Test @@ -138,7 +142,7 @@ public void testCreateSchemaSetsOwnerWithNormalizedIdentifier() throws Exception NameIdentifier ident = NameIdentifier.of("test_metalake", "test_catalog", "MY_SCHEMA"); Schema mockSchema = mock(Schema.class); - when(mockDispatcher.createSchema(any(), any(), any())).thenReturn(mockSchema); + when(mockDispatcher.createSchema(any(), any(), any(), any(), any())).thenReturn(mockSchema); hookDispatcher.createSchema(ident, "comment", Collections.emptyMap()); @@ -167,7 +171,7 @@ public void testCreateHierarchicalSchemaOwnsNewAncestors() throws Exception { NameIdentifier ident = NameIdentifier.of("test_metalake", "test_catalog", "A:B:C"); Schema mockSchema = mock(Schema.class); - when(mockDispatcher.createSchema(any(), any(), any())).thenReturn(mockSchema); + when(mockDispatcher.createSchema(any(), any(), any(), any(), any())).thenReturn(mockSchema); // No ancestor exists yet, so creating "A:B:C" auto-creates "A" and "A:B". when(mockDispatcher.schemaExists(any())).thenReturn(false); @@ -187,7 +191,7 @@ public void testCreateHierarchicalSchemaKeepsExistingAncestorOwner() throws Exce NameIdentifier ident = NameIdentifier.of("test_metalake", "test_catalog", "A:B:C"); Schema mockSchema = mock(Schema.class); - when(mockDispatcher.createSchema(any(), any(), any())).thenReturn(mockSchema); + when(mockDispatcher.createSchema(any(), any(), any(), any(), any())).thenReturn(mockSchema); // "A" already exists (and has its own owner); only "A:B" and the leaf are newly created. NameIdentifier existingA = NameIdentifier.of("test_metalake", "test_catalog", "A"); when(mockDispatcher.schemaExists(any())).thenReturn(false); diff --git a/core/src/test/java/org/apache/gravitino/listener/api/event/TestCatalogEvent.java b/core/src/test/java/org/apache/gravitino/listener/api/event/TestCatalogEvent.java index eaf01f9fb32..51f26585e5f 100644 --- a/core/src/test/java/org/apache/gravitino/listener/api/event/TestCatalogEvent.java +++ b/core/src/test/java/org/apache/gravitino/listener/api/event/TestCatalogEvent.java @@ -20,6 +20,7 @@ package org.apache.gravitino.listener.api.event; import static org.mockito.ArgumentMatchers.anyBoolean; +import static org.mockito.ArgumentMatchers.nullable; import static org.mockito.Mockito.any; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; @@ -382,7 +383,9 @@ private CatalogDispatcher mockCatalogDispatcher() { any(Catalog.Type.class), any(String.class), any(String.class), - any(Map.class))) + nullable(Map.class), + nullable(Map.class), + nullable(Map.class))) .thenReturn(catalog); when(dispatcher.loadCatalog(any(NameIdentifier.class))).thenReturn(catalog); when(dispatcher.dropCatalog(any(NameIdentifier.class), anyBoolean())).thenReturn(true); diff --git a/core/src/test/java/org/apache/gravitino/listener/api/event/TestSchemaEvent.java b/core/src/test/java/org/apache/gravitino/listener/api/event/TestSchemaEvent.java index 9e2e8fa293d..646fc220356 100644 --- a/core/src/test/java/org/apache/gravitino/listener/api/event/TestSchemaEvent.java +++ b/core/src/test/java/org/apache/gravitino/listener/api/event/TestSchemaEvent.java @@ -19,6 +19,7 @@ package org.apache.gravitino.listener.api.event; +import static org.mockito.ArgumentMatchers.nullable; import static org.mockito.Mockito.any; import static org.mockito.Mockito.eq; import static org.mockito.Mockito.mock; @@ -274,7 +275,12 @@ private Schema mockSchema() { private SchemaDispatcher mockSchemaDispatcher() { SchemaDispatcher dispatcher = mock(SchemaDispatcher.class); - when(dispatcher.createSchema(any(NameIdentifier.class), any(String.class), any(Map.class))) + when(dispatcher.createSchema( + any(NameIdentifier.class), + any(String.class), + nullable(Map.class), + nullable(Map.class), + nullable(Map.class))) .thenReturn(schema); when(dispatcher.loadSchema(any(NameIdentifier.class))).thenReturn(schema); when(dispatcher.dropSchema(any(NameIdentifier.class), eq(true))).thenReturn(false); diff --git a/core/src/test/java/org/apache/gravitino/secret/TestSecretPropertyUtils.java b/core/src/test/java/org/apache/gravitino/secret/TestSecretPropertyUtils.java index 89fd5d065b8..f556d9639b0 100644 --- a/core/src/test/java/org/apache/gravitino/secret/TestSecretPropertyUtils.java +++ b/core/src/test/java/org/apache/gravitino/secret/TestSecretPropertyUtils.java @@ -35,7 +35,8 @@ void testAssembleAndWrite() { Map properties = Map.of("jdbc-user", "root"); Map bindings = Map.of("jdbc-password", new SecretBinding("memory", "s3cr3t")); - Map entityProps = SecretPropertyUtils.copyEntityProperties(properties); + Map entityProps = + SecretPropertyUtils.copyEntityProperties(properties, bindings, Map.of()); List writes = sm.assembleSecretMaterials(properties, entityProps, "catalog", 42L, bindings, Map.of()); sm.writeSecrets(writes); @@ -50,9 +51,17 @@ void testAssembleAndWrite() { @Test void testCopyEntityProperties() { - Assertions.assertTrue(SecretPropertyUtils.copyEntityProperties(null).isEmpty()); + Assertions.assertNull(SecretPropertyUtils.copyEntityProperties(null, null, null)); + Assertions.assertNull(SecretPropertyUtils.copyEntityProperties(null, Map.of(), Map.of())); + + Map bindings = + Map.of("jdbc-password", new SecretBinding("memory", "s3cr3t")); + Map forSecrets = SecretPropertyUtils.copyEntityProperties(null, bindings, null); + Assertions.assertNotNull(forSecrets); + Assertions.assertTrue(forSecrets.isEmpty()); + Map original = Map.of("a", "b"); - Map copy = SecretPropertyUtils.copyEntityProperties(original); + Map copy = SecretPropertyUtils.copyEntityProperties(original, null, null); Assertions.assertEquals(original, copy); copy.put("c", "d"); Assertions.assertFalse(original.containsKey("c")); @@ -70,6 +79,15 @@ void testEmptySecretsNoOp() { } } + @Test + void testAssembleWithNullTargetWhenNoSecrets() { + try (SecretManager sm = memorySecretManager()) { + List writes = + sm.assembleSecretMaterials(null, null, "schema", 1L, null, null); + Assertions.assertTrue(writes.isEmpty()); + } + } + private static SecretManager memorySecretManager() { Config config = new Config(false) {}; Properties properties = new Properties(); diff --git a/docs/open-api/catalogs.yaml b/docs/open-api/catalogs.yaml index f568b04c527..f549166b6c7 100644 --- a/docs/open-api/catalogs.yaml +++ b/docs/open-api/catalogs.yaml @@ -390,6 +390,22 @@ components: default: { } additionalProperties: type: string + secretBindings: + type: object + description: > + Optional map of property key to write-through binding. Persisted value + becomes a URN. + nullable: true + additionalProperties: + $ref: "./secrets.yaml#/components/schemas/SecretBinding" + secretReferences: + type: object + description: > + Optional map of property key to secret locator. Server builds and + persists the URN. Must not overlap with secretBindings. + nullable: true + additionalProperties: + $ref: "./secrets.yaml#/components/schemas/SecretReference" CatalogSetRequest: type: object diff --git a/docs/open-api/schemas.yaml b/docs/open-api/schemas.yaml index 06f90f50564..81c3ea3f4e5 100644 --- a/docs/open-api/schemas.yaml +++ b/docs/open-api/schemas.yaml @@ -194,6 +194,22 @@ components: default: { } additionalProperties: type: string + secretBindings: + type: object + description: > + Optional map of property key to write-through binding. Persisted value + becomes a URN. + nullable: true + additionalProperties: + $ref: "./secrets.yaml#/components/schemas/SecretBinding" + secretReferences: + type: object + description: > + Optional map of property key to secret locator. Server builds and + persists the URN. Must not overlap with secretBindings. + nullable: true + additionalProperties: + $ref: "./secrets.yaml#/components/schemas/SecretReference" Schema: type: object diff --git a/server/src/main/java/org/apache/gravitino/server/web/rest/CatalogOperations.java b/server/src/main/java/org/apache/gravitino/server/web/rest/CatalogOperations.java index 7c66b4a8525..987d9e0b32d 100644 --- a/server/src/main/java/org/apache/gravitino/server/web/rest/CatalogOperations.java +++ b/server/src/main/java/org/apache/gravitino/server/web/rest/CatalogOperations.java @@ -20,6 +20,7 @@ import com.codahale.metrics.annotation.ResponseMetered; import com.codahale.metrics.annotation.Timed; +import java.util.Arrays; import javax.inject.Inject; import javax.servlet.http.HttpServletRequest; import javax.ws.rs.Consumes; @@ -52,6 +53,8 @@ import org.apache.gravitino.dto.responses.CatalogResponse; import org.apache.gravitino.dto.responses.DropResponse; import org.apache.gravitino.dto.responses.EntityListResponse; +import org.apache.gravitino.dto.secret.SecretBindingDTO; +import org.apache.gravitino.dto.secret.SecretReferenceDTO; import org.apache.gravitino.dto.util.DTOConverters; import org.apache.gravitino.metrics.MetricNames; import org.apache.gravitino.server.authorization.MetadataAuthzHelper; @@ -100,15 +103,17 @@ public Response listCatalogs( Namespace catalogNS = NamespaceUtil.ofCatalog(metalake); // Lock the root and the metalake with WRITE lock to ensure the consistency of the list. if (verbose) { - Catalog[] catalogs = catalogDispatcher.listCatalogsInfo(catalogNS); - catalogs = + // Authorize on identifiers first, then resolve catalog details (including secrets) + // only for authorized catalogs via loadCatalog. + NameIdentifier[] idents = catalogDispatcher.listCatalogs(catalogNS); + idents = MetadataAuthzHelper.filterByExpression( metalake, AuthorizationExpressionConstants.LOAD_CATALOG_AUTHORIZATION_EXPRESSION, Entity.EntityType.CATALOG, - catalogs, - (catalogEntity) -> - NameIdentifierUtil.ofCatalog(metalake, catalogEntity.name())); + idents); + Catalog[] catalogs = + Arrays.stream(idents).map(catalogDispatcher::loadCatalog).toArray(Catalog[]::new); Response response = Utils.ok(new CatalogListResponse(DTOConverters.toDTOs(catalogs))); LOG.info("List {} catalogs info under metalake: {}", catalogs.length, metalake); return response; @@ -154,7 +159,9 @@ public Response createCatalog( request.getType(), request.getProvider(), request.getComment(), - request.getProperties()); + request.getProperties(), + SecretBindingDTO.toSecretBindings(request.getSecretBindings()), + SecretReferenceDTO.toSecretReferences(request.getSecretReferences())); Response response = Utils.ok(new CatalogResponse(DTOConverters.toDTO(catalog))); LOG.info("Catalog created: {}.{}", metalake, catalog.name()); return response; diff --git a/server/src/main/java/org/apache/gravitino/server/web/rest/SchemaOperations.java b/server/src/main/java/org/apache/gravitino/server/web/rest/SchemaOperations.java index 1b8167331a3..49fc72941a2 100644 --- a/server/src/main/java/org/apache/gravitino/server/web/rest/SchemaOperations.java +++ b/server/src/main/java/org/apache/gravitino/server/web/rest/SchemaOperations.java @@ -49,6 +49,8 @@ import org.apache.gravitino.dto.responses.DropResponse; import org.apache.gravitino.dto.responses.EntityListResponse; import org.apache.gravitino.dto.responses.SchemaResponse; +import org.apache.gravitino.dto.secret.SecretBindingDTO; +import org.apache.gravitino.dto.secret.SecretReferenceDTO; import org.apache.gravitino.dto.util.DTOConverters; import org.apache.gravitino.metrics.MetricNames; import org.apache.gravitino.server.authorization.MetadataAuthzHelper; @@ -150,7 +152,12 @@ public Response createSchema( NameIdentifier ident = NameIdentifierUtil.ofSchema(metalake, catalog, request.getName()); Schema schema = - dispatcher.createSchema(ident, request.getComment(), request.getProperties()); + dispatcher.createSchema( + ident, + request.getComment(), + request.getProperties(), + SecretBindingDTO.toSecretBindings(request.getSecretBindings()), + SecretReferenceDTO.toSecretReferences(request.getSecretReferences())); Response response = Utils.ok(new SchemaResponse(DTOConverters.toDTO(schema))); LOG.info("Schema created: {}.{}.{}", metalake, catalog, schema.name()); return response; diff --git a/server/src/test/java/org/apache/gravitino/server/web/rest/TestCatalogOperations.java b/server/src/test/java/org/apache/gravitino/server/web/rest/TestCatalogOperations.java index 4298c5d2c77..f92d76c58bc 100644 --- a/server/src/test/java/org/apache/gravitino/server/web/rest/TestCatalogOperations.java +++ b/server/src/test/java/org/apache/gravitino/server/web/rest/TestCatalogOperations.java @@ -169,8 +169,12 @@ public void testListCatalogs() { public void testListCatalogsInfo() { TestCatalog catalog1 = buildCatalog("metalake1", "catalog1"); TestCatalog catalog2 = buildCatalog("metalake1", "catalog2"); + NameIdentifier ident1 = NameIdentifier.of("metalake1", "catalog1"); + NameIdentifier ident2 = NameIdentifier.of("metalake1", "catalog2"); - when(manager.listCatalogsInfo(any())).thenReturn(new Catalog[] {catalog1, catalog2}); + when(manager.listCatalogs(any())).thenReturn(new NameIdentifier[] {ident1, ident2}); + when(manager.loadCatalog(ident1)).thenReturn(catalog1); + when(manager.loadCatalog(ident2)).thenReturn(catalog2); Response resp = target("/metalakes/metalake1/catalogs") @@ -202,7 +206,7 @@ public void testListCatalogsInfo() { Assertions.assertEquals( ImmutableMap.of("key", "value", PROPERTY_IN_USE, "true"), catalogDTO2.properties()); - doThrow(new NoSuchMetalakeException("mock error")).when(manager).listCatalogsInfo(any()); + doThrow(new NoSuchMetalakeException("mock error")).when(manager).listCatalogs(any()); Response resp1 = target("/metalakes/metalake1/catalogs") .queryParam("details", "true") @@ -229,7 +233,8 @@ public void testCreateCatalog() { ImmutableMap.of("key", "value")); TestCatalog catalog = buildCatalog("metalake1", "catalog1"); - when(manager.createCatalog(any(), any(), any(), any(), any())).thenReturn(catalog); + when(manager.createCatalog(any(), any(), any(), any(), any(), any(), any())) + .thenReturn(catalog); Response resp = target("/metalakes/metalake1/catalogs") @@ -253,7 +258,7 @@ public void testCreateCatalog() { // Test throw NoSuchMetalakeException doThrow(new NoSuchMetalakeException("mock error")) .when(manager) - .createCatalog(any(), any(), any(), any(), any()); + .createCatalog(any(), any(), any(), any(), any(), any(), any()); Response resp1 = target("/metalakes/metalake1/catalogs") .request(MediaType.APPLICATION_JSON_TYPE) @@ -270,7 +275,7 @@ public void testCreateCatalog() { // Test throw CatalogAlreadyExistsException doThrow(new CatalogAlreadyExistsException("mock error")) .when(manager) - .createCatalog(any(), any(), any(), any(), any()); + .createCatalog(any(), any(), any(), any(), any(), any(), any()); Response resp2 = target("/metalakes/metalake1/catalogs") .request(MediaType.APPLICATION_JSON_TYPE) @@ -287,7 +292,7 @@ public void testCreateCatalog() { // Test throw internal RuntimeException doThrow(new RuntimeException("mock error")) .when(manager) - .createCatalog(any(), any(), any(), any(), any()); + .createCatalog(any(), any(), any(), any(), any(), any(), any()); Response resp3 = target("/metalakes/metalake1/catalogs") .request(MediaType.APPLICATION_JSON_TYPE) diff --git a/server/src/test/java/org/apache/gravitino/server/web/rest/TestSchemaOperations.java b/server/src/test/java/org/apache/gravitino/server/web/rest/TestSchemaOperations.java index 8981e5e29d9..08f2fb2be9d 100644 --- a/server/src/test/java/org/apache/gravitino/server/web/rest/TestSchemaOperations.java +++ b/server/src/test/java/org/apache/gravitino/server/web/rest/TestSchemaOperations.java @@ -223,7 +223,7 @@ public void testCreateSchema() { new SchemaCreateRequest("schema1", "comment", ImmutableMap.of("key", "value")); Schema mockSchema = mockSchema("schema1", "comment", ImmutableMap.of("key", "value")); - when(dispatcher.createSchema(any(), any(), any())).thenReturn(mockSchema); + when(dispatcher.createSchema(any(), any(), any(), any(), any())).thenReturn(mockSchema); Response resp = target("/metalakes/" + metalake + "/catalogs/" + catalog + "/schemas") @@ -245,7 +245,7 @@ public void testCreateSchema() { // Test throw NoSuchCatalogException doThrow(new NoSuchCatalogException("mock error")) .when(dispatcher) - .createSchema(any(), any(), any()); + .createSchema(any(), any(), any(), any(), any()); Response resp1 = target("/metalakes/" + metalake + "/catalogs/" + catalog + "/schemas") .request(MediaType.APPLICATION_JSON_TYPE) @@ -262,7 +262,7 @@ public void testCreateSchema() { // Test throw SchemaAlreadyExistsException doThrow(new SchemaAlreadyExistsException("mock error")) .when(dispatcher) - .createSchema(any(), any(), any()); + .createSchema(any(), any(), any(), any(), any()); Response resp2 = target("/metalakes/" + metalake + "/catalogs/" + catalog + "/schemas") @@ -279,7 +279,9 @@ public void testCreateSchema() { SchemaAlreadyExistsException.class.getSimpleName(), errorResp2.getType()); // Test throw RuntimeException - doThrow(new RuntimeException("mock error")).when(dispatcher).createSchema(any(), any(), any()); + doThrow(new RuntimeException("mock error")) + .when(dispatcher) + .createSchema(any(), any(), any(), any(), any()); Response resp3 = target("/metalakes/" + metalake + "/catalogs/" + catalog + "/schemas")