[#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("Creating a catalog with secrets is not supported");
+  }
 
   /**
    * 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("Creating a schema with secrets is not supported");
+  }
 
   /**
    * Load metadata properties for a schema.

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<String, String> storageLocations() {
-    throw new UnsupportedOperationException("Not implemented");
+    throw new UnsupportedOperationException("Fileset does not support storageLocations.");
   }
 
   /**

api/src/main/java/org/apache/gravitino/file/FilesetCatalog.java

@@ -170,7 +170,7 @@ default Fileset createMultipleLocationFileset(
       Map<String, SecretBinding> secretBindings,
       Map<String, SecretReference> 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");
   }
 }

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<String, String> 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<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/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<String, String> storageLocations,
+      Map<String, String> 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

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<String, String> 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<String, String> properties,
+      Map<String, SecretBinding> secretBindings,
+      Map<String, SecretReference> secretReferences)
+      throws NoSuchMetalakeException, CatalogAlreadyExistsException {
+    return getMetalake()
+        .createCatalog(
+            catalogName, type, provider, comment, properties, secretBindings, secretReferences);
   }
 
   @Override