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

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

@@ -7,13 +7,13 @@
 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.
+ * 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 {
@@ -72,8 +72,8 @@ enum Type {
   /**
    * 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/rel/Schema.java → api/src/main/java/com/datastrato/gravitino/Schema.java

@@ -2,9 +2,8 @@
  * Copyright 2023 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.Auditable;
 import com.datastrato.gravitino.annotation.Evolving;
 import java.util.Collections;
 import java.util.Map;
@@ -16,7 +15,7 @@
  * which means it can be schema1.schema2.table.
  *
  * <p>This defines the basic properties of a schema. A catalog implementation with {@link
- * SupportsSchemas} should implement this interface.
+ * com.datastrato.gravitino.SupportsSchemas} should implement this interface.
  */
 @Evolving
 public interface Schema extends Auditable {

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

@@ -18,7 +18,7 @@
 // Referred from Apache Spark's connector/catalog implementation
 // sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/NamespaceChange.java
 
-package com.datastrato.gravitino.rel;
+package com.datastrato.gravitino;
 
 import com.datastrato.gravitino.annotation.Evolving;
 import java.util.Objects;

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.

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,113 @@
+/*
+ * 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 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/test/java/com/datastrato/gravitino/TestSchemaChange.java

@@ -9,9 +9,8 @@
 import static org.junit.jupiter.api.Assertions.assertNotEquals;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
-import com.datastrato.gravitino.rel.SchemaChange;
-import com.datastrato.gravitino.rel.SchemaChange.RemoveProperty;
-import com.datastrato.gravitino.rel.SchemaChange.SetProperty;
+import com.datastrato.gravitino.SchemaChange.RemoveProperty;
+import com.datastrato.gravitino.SchemaChange.SetProperty;
 import org.junit.jupiter.api.Test;
 
 public class TestSchemaChange {

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

@@ -7,8 +7,6 @@
 import com.datastrato.gravitino.connector.BaseCatalog;
 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 java.util.Map;
 
 /**
@@ -33,14 +31,4 @@ protected CatalogOperations newOps(Map<String, String> config) {
   protected Capability newCapability() {
     return new HadoopCatalogCapability();
   }
-
-  @Override
-  public SupportsSchemas asSchemas() {
-    return (HadoopCatalogOperations) ops();
-  }
-
-  @Override
-  public FilesetCatalog asFilesetCatalog() {
-    return (HadoopCatalogOperations) ops();
-  }
 }

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

@@ -11,10 +11,13 @@
 import com.datastrato.gravitino.GravitinoEnv;
 import com.datastrato.gravitino.NameIdentifier;
 import com.datastrato.gravitino.Namespace;
+import com.datastrato.gravitino.Schema;
+import com.datastrato.gravitino.SchemaChange;
 import com.datastrato.gravitino.StringIdentifier;
 import com.datastrato.gravitino.connector.CatalogInfo;
 import com.datastrato.gravitino.connector.CatalogOperations;
 import com.datastrato.gravitino.connector.PropertiesMetadata;
+import com.datastrato.gravitino.connector.SupportsSchemas;
 import com.datastrato.gravitino.exceptions.AlreadyExistsException;
 import com.datastrato.gravitino.exceptions.FilesetAlreadyExistsException;
 import com.datastrato.gravitino.exceptions.NoSuchCatalogException;
@@ -29,9 +32,6 @@
 import com.datastrato.gravitino.meta.AuditInfo;
 import com.datastrato.gravitino.meta.FilesetEntity;
 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.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/TestHadoopCatalogOperations.java

@@ -19,15 +19,15 @@
 import com.datastrato.gravitino.EntityStoreFactory;
 import com.datastrato.gravitino.NameIdentifier;
 import com.datastrato.gravitino.Namespace;
+import com.datastrato.gravitino.Schema;
+import com.datastrato.gravitino.SchemaChange;
 import com.datastrato.gravitino.StringIdentifier;
 import com.datastrato.gravitino.exceptions.NoSuchFilesetException;
 import com.datastrato.gravitino.exceptions.NoSuchSchemaException;
 import com.datastrato.gravitino.exceptions.NonEmptySchemaException;
 import com.datastrato.gravitino.exceptions.SchemaAlreadyExistsException;
 import com.datastrato.gravitino.file.Fileset;
 import com.datastrato.gravitino.file.FilesetChange;
-import com.datastrato.gravitino.rel.Schema;
-import com.datastrato.gravitino.rel.SchemaChange;
 import com.datastrato.gravitino.storage.IdGenerator;
 import com.datastrato.gravitino.storage.RandomIdGenerator;
 import com.google.common.collect.ImmutableMap;

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

@@ -7,6 +7,7 @@
 import com.datastrato.gravitino.Catalog;
 import com.datastrato.gravitino.NameIdentifier;
 import com.datastrato.gravitino.Namespace;
+import com.datastrato.gravitino.Schema;
 import com.datastrato.gravitino.client.GravitinoMetalake;
 import com.datastrato.gravitino.exceptions.FilesetAlreadyExistsException;
 import com.datastrato.gravitino.exceptions.NoSuchFilesetException;
@@ -16,7 +17,6 @@
 import com.datastrato.gravitino.integration.test.container.HiveContainer;
 import com.datastrato.gravitino.integration.test.util.AbstractIT;
 import com.datastrato.gravitino.integration.test.util.GravitinoITUtils;
-import com.datastrato.gravitino.rel.Schema;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Maps;
 import java.io.IOException;
@@ -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,6 @@
 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 java.util.Map;
 import java.util.Optional;
 
@@ -43,26 +41,6 @@ public Capability newCapability() {
     return new HiveCatalogCapability();
   }
 
-  /**
-   * Returns the Hive catalog operations as a {@link SupportsSchemas}.
-   *
-   * @return The Hive catalog operations as {@link HiveCatalogOperations}.
-   */
-  @Override
-  public SupportsSchemas asSchemas() {
-    return (SupportsSchemas) ops();
-  }
-
-  /**
-   * Returns the Hive catalog operations as a {@link TableCatalog}.
-   *
-   * @return The Hive catalog operations as {@link HiveCatalogOperations}.
-   */
-  @Override
-  public TableCatalog asTableCatalog() {
-    return (TableCatalog) ops();
-  }
-
   @Override
   protected Optional<ProxyPlugin> newProxyPlugin(Map<String, String> config) {
     boolean impersonationEnabled =