[#3483] refactor(API): separate API for client using: SupportsSchemas

api/src/main/java/com/datastrato/gravitino/Catalog.java

@@ -7,73 +7,21 @@
 import com.datastrato.gravitino.annotation.Evolving;
 import com.datastrato.gravitino.file.FilesetCatalog;
 import com.datastrato.gravitino.messaging.TopicCatalog;
-import com.datastrato.gravitino.rel.SupportsSchemas;
 import com.datastrato.gravitino.rel.TableCatalog;
-import java.util.Map;
 
 /**
- * The interface of a catalog. The catalog is the second level entity in the gravitino system,
- * containing a set of tables.
+ * The client interface of a catalog. The catalog is the second level entity in the gravitino
+ * system, containing a set of tables. The server side should use the other one with the same name
+ * in the core module.
  */
 @Evolving
-public interface Catalog extends Auditable {
-
-  /** The type of the catalog. */
-  enum Type {
-    /** Catalog Type for Relational Data Structure, like db.table, catalog.db.table. */
-    RELATIONAL,
-
-    /** Catalog Type for Fileset System (including HDFS, S3, etc.), like path/to/file */
-    FILESET,
-
-    /** Catalog Type for Message Queue, like Kafka://topic */
-    MESSAGING,
-
-    /** Catalog Type for test only. */
-    UNSUPPORTED
-  }
-
-  /**
-   * A reserved property to specify the package location of the catalog. The "package" is a string
-   * of path to the folder where all the catalog related dependencies is located. The dependencies
-   * under the "package" will be loaded by Gravitino to create the catalog.
-   *
-   * <p>The property "package" is not needed if the catalog is a built-in one, Gravitino will search
-   * the proper location using "provider" to load the dependencies. Only when the folder is in
-   * different location, the "package" property is needed.
-   */
-  String PROPERTY_PACKAGE = "package";
-
-  /** @return The name of the catalog. */
-  String name();
-
-  /** @return The type of the catalog. */
-  Type type();
-
-  /** @return The provider of the catalog. */
-  String provider();
-
-  /**
-   * The comment of the catalog. Note. this method will return null if the comment is not set for
-   * this catalog.
-   *
-   * @return The comment of the catalog.
-   */
-  String comment();
-
-  /**
-   * The properties of the catalog. Note, this method will return null if the properties are not
-   * set.
-   *
-   * @return The properties of the catalog.
-   */
-  Map<String, String> properties();
+public interface Catalog extends CatalogBasicInfo {
 
   /**
    * Return the {@link SupportsSchemas} if the catalog supports schema operations.
    *
-   * @throws UnsupportedOperationException if the catalog does not support schema operations.
    * @return The {@link SupportsSchemas} if the catalog supports schema operations.
+   * @throws UnsupportedOperationException if the catalog does not support schema operations.
    */
   default SupportsSchemas asSchemas() throws UnsupportedOperationException {
     throw new UnsupportedOperationException("Catalog does not support schema operations");

api/src/main/java/com/datastrato/gravitino/CatalogBasicInfo.java

@@ -0,0 +1,64 @@
+/*
+ * Copyright 2023 Datastrato Pvt Ltd.
+ * This software is licensed under the Apache License version 2.
+ */
+package com.datastrato.gravitino;
+
+import com.datastrato.gravitino.annotation.Evolving;
+import java.util.Map;
+
+/** The base interface of a catalog, defining the basic attributes and catalog types. */
+@Evolving
+public interface CatalogBasicInfo extends Auditable {
+
+  /** The type of the catalog. */
+  enum Type {
+    /** Catalog Type for Relational Data Structure, like db.table, catalog.db.table. */
+    RELATIONAL,
+
+    /** Catalog Type for Fileset System (including HDFS, S3, etc.), like path/to/file */
+    FILESET,
+
+    /** Catalog Type for Message Queue, like Kafka://topic */
+    MESSAGING,
+
+    /** Catalog Type for test only. */
+    UNSUPPORTED
+  }
+
+  /**
+   * A reserved property to specify the package location of the catalog. The "package" is a string
+   * of path to the folder where all the catalog related dependencies is located. The dependencies
+   * under the "package" will be loaded by Gravitino to create the catalog.
+   *
+   * <p>The property "package" is not needed if the catalog is a built-in one, Gravitino will search
+   * the proper location using "provider" to load the dependencies. Only when the folder is in
+   * different location, the "package" property is needed.
+   */
+  String PROPERTY_PACKAGE = "package";
+
+  /** @return The name of the catalog. */
+  String name();
+
+  /** @return The type of the catalog. */
+  Type type();
+
+  /** @return The provider of the catalog. */
+  String provider();
+
+  /**
+   * The comment of the catalog. Note. this method will return null if the comment is not set for
+   * this catalog.
+   *
+   * @return The comment of the catalog.
+   */
+  String comment();
+
+  /**
+   * The properties of the catalog. Note, this method will return null if the properties are not
+   * set.
+   *
+   * @return The properties of the catalog.
+   */
+  Map<String, String> properties();
+}

api/src/main/java/com/datastrato/gravitino/rel/SupportsCatalogs.java → api/src/main/java/com/datastrato/gravitino/SupportsCatalogs.java

@@ -2,12 +2,8 @@
  * Copyright 2024 Datastrato Pvt Ltd.
  * This software is licensed under the Apache License version 2.
  */
-package com.datastrato.gravitino.rel;
+package com.datastrato.gravitino;
 
-import com.datastrato.gravitino.Catalog;
-import com.datastrato.gravitino.CatalogChange;
-import com.datastrato.gravitino.CatalogProvider;
-import com.datastrato.gravitino.NameIdentifier;
 import com.datastrato.gravitino.annotation.Evolving;
 import com.datastrato.gravitino.exceptions.CatalogAlreadyExistsException;
 import com.datastrato.gravitino.exceptions.NoSuchCatalogException;
@@ -65,8 +61,8 @@ default boolean catalogExists(String catalogName) {
    * Create a catalog with specified identifier.
    *
    * <p>The parameter "provider" is a short name of the catalog, used to tell Gravitino which
-   * catalog should be created. The short name should be the same as the {@link CatalogProvider}
-   * interface provided.
+   * catalog should be created. The short name should be the same as the {@link
+   * com.datastrato.gravitino.CatalogProvider} interface provided.
    *
    * @param catalogName the name of the catalog.
    * @param type the type of the catalog.
@@ -79,7 +75,7 @@ default boolean catalogExists(String catalogName) {
    */
   Catalog createCatalog(
       String catalogName,
-      Catalog.Type type,
+      CatalogBasicInfo.Type type,
       String provider,
       String comment,
       Map<String, String> properties)

api/src/main/java/com/datastrato/gravitino/rel/SupportsMetalakes.java → api/src/main/java/com/datastrato/gravitino/SupportsMetalakes.java

@@ -2,10 +2,8 @@
  * Copyright 2024 Datastrato Pvt Ltd.
  * This software is licensed under the Apache License version 2.
  */
-package com.datastrato.gravitino.rel;
+package com.datastrato.gravitino;
 
-import com.datastrato.gravitino.Metalake;
-import com.datastrato.gravitino.MetalakeChange;
 import com.datastrato.gravitino.annotation.Evolving;
 import com.datastrato.gravitino.exceptions.MetalakeAlreadyExistsException;
 import com.datastrato.gravitino.exceptions.NoSuchMetalakeException;

api/src/main/java/com/datastrato/gravitino/SupportsSchemas.java

@@ -0,0 +1,115 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+// Referred from Apache Spark's connector/catalog implementation
+// sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/SupportNamespaces.java
+
+package com.datastrato.gravitino;
+
+import com.datastrato.gravitino.annotation.Evolving;
+import com.datastrato.gravitino.exceptions.NoSuchCatalogException;
+import com.datastrato.gravitino.exceptions.NoSuchSchemaException;
+import com.datastrato.gravitino.exceptions.NonEmptySchemaException;
+import com.datastrato.gravitino.exceptions.SchemaAlreadyExistsException;
+import com.datastrato.gravitino.rel.Schema;
+import com.datastrato.gravitino.rel.SchemaChange;
+import java.util.Map;
+
+/**
+ * The client interface to support schema operations. The server side should use the other one with
+ * the same name in the core module.
+ */
+@Evolving
+public interface SupportsSchemas {
+
+  /**
+   * List schemas under the entity.
+   *
+   * <p>If an entity such as a table, view exists, its parent schemas must also exist and must be
+   * returned by this discovery method. For example, if table a.b.t exists, this method invoked as
+   * listSchemas(a) must return [a.b] in the result array
+   *
+   * @return An array of schema identifier under the namespace.
+   * @throws NoSuchCatalogException If the catalog does not exist.
+   */
+  NameIdentifier[] listSchemas() throws NoSuchCatalogException;
+
+  /**
+   * Check if a schema exists.
+   *
+   * <p>If an entity such as a table, view exists, its parent namespaces must also exist. For
+   * example, if table a.b.t exists, this method invoked as schemaExists(a.b) must return true.
+   *
+   * @param schemaName The name of the schema.
+   * @return True if the schema exists, false otherwise.
+   */
+  default boolean schemaExists(String schemaName) {
+    try {
+      loadSchema(schemaName);
+      return true;
+    } catch (NoSuchSchemaException e) {
+      return false;
+    }
+  }
+
+  /**
+   * Create a schema in the catalog.
+   *
+   * @param schemaName The name 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.
+   */
+  Schema createSchema(String schemaName, String comment, Map<String, String> properties)
+      throws NoSuchCatalogException, SchemaAlreadyExistsException;
+
+  /**
+   * Load metadata properties for a schema.
+   *
+   * @param schemaName The name of the schema.
+   * @return A schema.
+   * @throws NoSuchSchemaException If the schema does not exist (optional).
+   */
+  Schema loadSchema(String schemaName) throws NoSuchSchemaException;
+
+  /**
+   * Apply the metadata change to a schema in the catalog.
+   *
+   * @param schemaName The name of the schema.
+   * @param changes The metadata changes to apply.
+   * @return The altered schema.
+   * @throws NoSuchSchemaException If the schema does not exist.
+   */
+  Schema alterSchema(String schemaName, SchemaChange... changes) throws NoSuchSchemaException;
+
+  /**
+   * Drop a schema from the catalog. If cascade option is true, recursively drop all objects within
+   * the schema.
+   *
+   * <p>If the catalog implementation does not support this operation, it may throw {@link
+   * UnsupportedOperationException}.
+   *
+   * @param schemaName The name of the schema.
+   * @param cascade If true, recursively drop all objects within the schema.
+   * @return True if the schema exists and is dropped successfully, false if the schema doesn't
+   *     exist.
+   * @throws NonEmptySchemaException If the schema is not empty and cascade is false.
+   */
+  boolean dropSchema(String schemaName, boolean cascade) throws NonEmptySchemaException;
+}

api/src/main/java/com/datastrato/gravitino/rel/Schema.java

@@ -5,6 +5,7 @@
 package com.datastrato.gravitino.rel;
 
 import com.datastrato.gravitino.Auditable;
+import com.datastrato.gravitino.SupportsSchemas;
 import com.datastrato.gravitino.annotation.Evolving;
 import java.util.Collections;
 import java.util.Map;

catalogs/catalog-hadoop/src/main/java/com/datastrato/gravitino/catalog/hadoop/HadoopCatalog.java

@@ -8,7 +8,7 @@
 import com.datastrato.gravitino.connector.CatalogOperations;
 import com.datastrato.gravitino.connector.capability.Capability;
 import com.datastrato.gravitino.file.FilesetCatalog;
-import com.datastrato.gravitino.rel.SupportsSchemas;
+import com.datastrato.gravitino.schema.SupportsSchemas;
 import java.util.Map;
 
 /**

catalogs/catalog-hadoop/src/main/java/com/datastrato/gravitino/catalog/hadoop/HadoopCatalogOperations.java

@@ -31,7 +31,7 @@
 import com.datastrato.gravitino.meta.SchemaEntity;
 import com.datastrato.gravitino.rel.Schema;
 import com.datastrato.gravitino.rel.SchemaChange;
-import com.datastrato.gravitino.rel.SupportsSchemas;
+import com.datastrato.gravitino.schema.SupportsSchemas;
 import com.datastrato.gravitino.utils.PrincipalUtils;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;

catalogs/catalog-hadoop/src/test/java/com/datastrato/gravitino/catalog/hadoop/integration/test/HadoopCatalogIT.java

@@ -100,15 +100,14 @@ private static void createCatalog() {
   }
 
   private static void createSchema() {
-    NameIdentifier ident = NameIdentifier.of(metalakeName, catalogName, schemaName);
     Map<String, String> properties = Maps.newHashMap();
     properties.put("key1", "val1");
     properties.put("key2", "val2");
     properties.put("location", defaultBaseLocation());
     String comment = "comment";
 
-    catalog.asSchemas().createSchema(ident, comment, properties);
-    Schema loadSchema = catalog.asSchemas().loadSchema(ident);
+    catalog.asSchemas().createSchema(schemaName, comment, properties);
+    Schema loadSchema = catalog.asSchemas().loadSchema(schemaName);
     Assertions.assertEquals(schemaName, loadSchema.name());
     Assertions.assertEquals(comment, loadSchema.comment());
     Assertions.assertEquals("val1", loadSchema.properties().get("key1"));
@@ -117,9 +116,8 @@ private static void createSchema() {
   }
 
   private static void dropSchema() {
-    NameIdentifier ident = NameIdentifier.of(metalakeName, catalogName, schemaName);
-    catalog.asSchemas().dropSchema(ident, true);
-    Assertions.assertFalse(catalog.asSchemas().schemaExists(ident));
+    catalog.asSchemas().dropSchema(schemaName, true);
+    Assertions.assertFalse(catalog.asSchemas().schemaExists(schemaName));
   }
 
   @Test
@@ -548,22 +546,13 @@ public void testDropCatalogWithEmptySchema() {
     // Create a schema without specifying location.
     String schemaName =
         GravitinoITUtils.genRandomName("test_drop_catalog_with_empty_schema_schema");
-    filesetCatalog
-        .asSchemas()
-        .createSchema(
-            NameIdentifier.of(metalakeName, catalogName, schemaName), "comment", ImmutableMap.of());
+    filesetCatalog.asSchemas().createSchema(schemaName, "comment", ImmutableMap.of());
 
     // Drop the empty schema.
-    boolean dropped =
-        filesetCatalog
-            .asSchemas()
-            .dropSchema(NameIdentifier.of(metalakeName, catalogName, schemaName), true);
+    boolean dropped = filesetCatalog.asSchemas().dropSchema(schemaName, true);
     Assertions.assertTrue(dropped, "schema should be dropped");
     Assertions.assertFalse(
-        filesetCatalog
-            .asSchemas()
-            .schemaExists(NameIdentifier.of(metalakeName, catalogName, schemaName)),
-        "schema should not be exists");
+        filesetCatalog.asSchemas().schemaExists(schemaName), "schema should not be exists");
 
     // Drop the catalog.
     dropped = metalake.dropCatalog(catalogName);

catalogs/catalog-hive/src/main/java/com/datastrato/gravitino/catalog/hive/HiveCatalog.java

@@ -8,8 +8,8 @@
 import com.datastrato.gravitino.connector.CatalogOperations;
 import com.datastrato.gravitino.connector.ProxyPlugin;
 import com.datastrato.gravitino.connector.capability.Capability;
-import com.datastrato.gravitino.rel.SupportsSchemas;
 import com.datastrato.gravitino.rel.TableCatalog;
+import com.datastrato.gravitino.schema.SupportsSchemas;
 import java.util.Map;
 import java.util.Optional;