[#12297] feat(secret): Add catalog and schema create-time secrets

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.
+   * <p>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<String, String> 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.
+   *
+   * <p>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<String, String> properties,
+      Map<String, SecretBinding> secretBindings,
+      Map<String, SecretReference> secretReferences)
+      throws NoSuchMetalakeException, CatalogAlreadyExistsException {
+    throw new UnsupportedOperationException("Not implemented");
+  }
 
   /**
    * Create a managed catalog with specified catalog name, type, comment, and properties.

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.
    *
+   * <p>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<String, String> properties)
+      throws NoSuchCatalogException, SchemaAlreadyExistsException {
+    return createSchema(
+        schemaName, comment, properties, Collections.emptyMap(), Collections.emptyMap());
+  }
+
+  /**
+   * Creates a schema with optional secret maps.
+   *
+   * <p>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<String, String> properties)
-      throws NoSuchCatalogException, SchemaAlreadyExistsException;
+  default Schema createSchema(
+      String schemaName,
+      String comment,
+      Map<String, String> properties,
+      Map<String, SecretBinding> secretBindings,
+      Map<String, SecretReference> secretReferences)
+      throws NoSuchCatalogException, SchemaAlreadyExistsException {
+    throw new UnsupportedOperationException("Not implemented");
+  }
 
   /**
    * Load metadata properties for a schema.

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,20 +171,35 @@ public String[] listSchemas(String parentSchema)
   }
 
   /**
-   * Create a new schema with specified identifier, comment and metadata.
+   * 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<String, String> properties)
+  public Schema createSchema(
+      String schemaName,
+      String comment,
+      Map<String, String> properties,
+      Map<String, SecretBinding> secretBindings,
+      Map<String, SecretReference> 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 =

clients/client-java/src/main/java/org/apache/gravitino/client/GravitinoClient.java

@@ -67,6 +67,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;
@@ -137,9 +139,13 @@ public Catalog createCatalog(
       Catalog.Type type,
       String provider,
       String comment,
-      Map<String, String> properties)
+      Map<String, String> properties,
+      Map<String, SecretBinding> secretBindings,
+      Map<String, SecretReference> secretReferences)
       throws NoSuchMetalakeException, CatalogAlreadyExistsException {
-    return getMetalake().createCatalog(catalogName, type, provider, comment, properties);
+    return getMetalake()
+        .createCatalog(
+            catalogName, type, provider, comment, properties, secretBindings, secretReferences);
   }
 
   @Override

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;
@@ -236,13 +240,18 @@ public Catalog loadCatalog(String catalogName) throws NoSuchCatalogException {
   }
 
   /**
-   * Create a new catalog with specified identifier, type, comment and properties.
+   * 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.
@@ -253,10 +262,19 @@ public Catalog createCatalog(
       Catalog.Type type,
       String provider,
       String comment,
-      Map<String, String> properties)
+      Map<String, String> properties,
+      Map<String, SecretBinding> secretBindings,
+      Map<String, SecretReference> 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 =

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,14 +84,21 @@ 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.
 
         Args:
             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.