fix

clients/client-java/src/main/java/com/datastrato/gravitino/client/BaseSchemaCatalog.java

@@ -27,16 +27,13 @@
 import java.util.List;
 import java.util.Map;
 import java.util.stream.Collectors;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 /**
  * BaseSchemaCatalog is the base abstract class for all the catalog with schema. It provides the
  * common methods for managing schemas in a catalog. With {@link BaseSchemaCatalog}, users can list,
  * create, load, alter and drop a schema with specified identifier.
  */
 abstract class BaseSchemaCatalog extends CatalogDTO implements SupportsSchemas {
-  private static final Logger LOG = LoggerFactory.getLogger(BaseSchemaCatalog.class);
 
   /** The REST client to send the requests. */
   protected final RESTClient restClient;
@@ -170,32 +167,22 @@ public Schema alterSchema(NameIdentifier ident, SchemaChange... changes)
    *
    * @param ident The name identifier of the schema.
    * @param cascade Whether to drop all the tables under the schema.
-   * @return true if the schema is dropped successfully, false otherwise.
+   * @return true if the schema is dropped successfully, false if the schema does not exist.
    * @throws NonEmptySchemaException if the schema is not empty and cascade is false.
    */
   @Override
   public boolean dropSchema(NameIdentifier ident, boolean cascade) throws NonEmptySchemaException {
     NameIdentifier.checkSchema(ident);
 
-    try {
-      DropResponse resp =
-          restClient.delete(
-              formatSchemaRequestPath(ident.namespace())
-                  + "/"
-                  + RESTUtils.encodeString(ident.name()),
-              Collections.singletonMap("cascade", String.valueOf(cascade)),
-              DropResponse.class,
-              Collections.emptyMap(),
-              ErrorHandlers.schemaErrorHandler());
-      resp.validate();
-      return resp.dropped();
-
-    } catch (NonEmptySchemaException e) {
-      throw e;
-    } catch (Exception e) {
-      LOG.warn("Failed to drop schema {}", ident, e);
-      return false;
-    }
+    DropResponse resp =
+        restClient.delete(
+            formatSchemaRequestPath(ident.namespace()) + "/" + RESTUtils.encodeString(ident.name()),
+            Collections.singletonMap("cascade", String.valueOf(cascade)),
+            DropResponse.class,
+            Collections.emptyMap(),
+            ErrorHandlers.schemaErrorHandler());
+    resp.validate();
+    return resp.dropped();
   }
 
   static String formatSchemaRequestPath(Namespace ns) {

clients/client-java/src/main/java/com/datastrato/gravitino/client/GravitinoAdminClient.java

@@ -44,8 +44,6 @@
 import java.util.Map;
 import java.util.Objects;
 import java.util.stream.Collectors;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 /**
  * Gravitino Client for the administrator to interact with the Gravitino API, allowing the client to
@@ -54,8 +52,6 @@
  * <p>Normal users should use {@link GravitinoClient} to connect with the Gravitino server.
  */
 public class GravitinoAdminClient extends GravitinoClientBase implements SupportsMetalakes {
-
-  private static final Logger LOG = LoggerFactory.getLogger(GravitinoAdminClient.class);
   private static final String API_METALAKES_USERS_PATH = "api/metalakes/%s/users/%s";
   private static final String API_METALAKES_GROUPS_PATH = "api/metalakes/%s/groups/%s";
   private static final String API_METALAKES_ROLES_PATH = "api/metalakes/%s/roles/%s";
@@ -168,26 +164,20 @@ public GravitinoMetalake alterMetalake(NameIdentifier ident, MetalakeChange... c
    * Drops a specific Metalake using the Gravitino API.
    *
    * @param ident The identifier of the Metalake to be dropped.
-   * @return True if the Metalake was successfully dropped, false otherwise.
+   * @return True if the Metalake was successfully dropped, false if the Metalake does not exist.
    */
   @Override
   public boolean dropMetalake(NameIdentifier ident) {
     NameIdentifier.checkMetalake(ident);
 
-    try {
-      DropResponse resp =
-          restClient.delete(
-              API_METALAKES_IDENTIFIER_PATH + ident.name(),
-              DropResponse.class,
-              Collections.emptyMap(),
-              ErrorHandlers.metalakeErrorHandler());
-      resp.validate();
-      return resp.dropped();
-
-    } catch (Exception e) {
-      LOG.warn("Failed to drop metadata {}", ident, e);
-      return false;
-    }
+    DropResponse resp =
+        restClient.delete(
+            API_METALAKES_IDENTIFIER_PATH + ident.name(),
+            DropResponse.class,
+            Collections.emptyMap(),
+            ErrorHandlers.metalakeErrorHandler());
+    resp.validate();
+    return resp.dropped();
   }
 
   /**

clients/client-java/src/main/java/com/datastrato/gravitino/client/GravitinoMetalake.java

@@ -29,8 +29,6 @@
 import java.util.Map;
 import java.util.stream.Collectors;
 import org.apache.commons.lang3.StringUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 /**
  * Gravitino Metalake is the top-level metadata repository for users. It contains a list of catalogs
@@ -39,8 +37,6 @@
  */
 public class GravitinoMetalake extends MetalakeDTO implements SupportsCatalogs {
 
-  private static final Logger LOG = LoggerFactory.getLogger(GravitinoMetalake.class);
-
   private static final String API_METALAKES_CATALOGS_PATH = "api/metalakes/%s/catalogs/%s";
 
   private final RESTClient restClient;
@@ -200,26 +196,20 @@ public Catalog alterCatalog(NameIdentifier ident, CatalogChange... changes)
    * Drop the catalog with specified identifier.
    *
    * @param ident the identifier of the catalog.
-   * @return true if the catalog is dropped successfully, false otherwise.
+   * @return true if the catalog is dropped successfully, false if the catalog does not exist.
    */
   @Override
   public boolean dropCatalog(NameIdentifier ident) {
     NameIdentifier.checkCatalog(ident);
 
-    try {
-      DropResponse resp =
-          restClient.delete(
-              String.format(API_METALAKES_CATALOGS_PATH, ident.namespace().level(0), ident.name()),
-              DropResponse.class,
-              Collections.emptyMap(),
-              ErrorHandlers.catalogErrorHandler());
-      resp.validate();
-      return resp.dropped();
-
-    } catch (Exception e) {
-      LOG.warn("Failed to drop catalog {}", ident, e);
-      return false;
-    }
+    DropResponse resp =
+        restClient.delete(
+            String.format(API_METALAKES_CATALOGS_PATH, ident.namespace().level(0), ident.name()),
+            DropResponse.class,
+            Collections.emptyMap(),
+            ErrorHandlers.catalogErrorHandler());
+    resp.validate();
+    return resp.dropped();
   }
 
   static class Builder extends MetalakeDTO.Builder<Builder> {

clients/client-java/src/main/java/com/datastrato/gravitino/client/RelationalCatalog.java

@@ -197,28 +197,20 @@ public Table alterTable(NameIdentifier ident, TableChange... changes)
    * Drop the table with specified identifier.
    *
    * @param ident The identifier of the table.
-   * @return true if the table is dropped successfully, false otherwise.
+   * @return true if the table is dropped successfully, false if the table does not exist.
    */
   @Override
   public boolean dropTable(NameIdentifier ident) {
     NameIdentifier.checkTable(ident);
 
-    try {
-      DropResponse resp =
-          restClient.delete(
-              formatTableRequestPath(ident.namespace())
-                  + "/"
-                  + RESTUtils.encodeString(ident.name()),
-              DropResponse.class,
-              Collections.emptyMap(),
-              ErrorHandlers.tableErrorHandler());
-      resp.validate();
-      return resp.dropped();
-
-    } catch (Exception e) {
-      LOG.warn("Failed to drop table {}", ident, e);
-      return false;
-    }
+    DropResponse resp =
+        restClient.delete(
+            formatTableRequestPath(ident.namespace()) + "/" + RESTUtils.encodeString(ident.name()),
+            DropResponse.class,
+            Collections.emptyMap(),
+            ErrorHandlers.tableErrorHandler());
+    resp.validate();
+    return resp.dropped();
   }
 
   /**

clients/client-java/src/main/java/com/datastrato/gravitino/client/RelationalTable.java

@@ -222,7 +222,7 @@ public Partition addPartition(Partition partition) throws PartitionAlreadyExists
    * Drops the partition with the given name.
    *
    * @param partitionName The name of the partition.
-   * @return true if the partition is dropped, false otherwise.
+   * @return true if the partition is dropped, false if the partition does not exist.
    */
   @Override
   public boolean dropPartition(String partitionName) {

clients/client-java/src/test/java/com/datastrato/gravitino/client/TestGravitinoClient.java

@@ -218,11 +218,13 @@ public void testAlterMetalake() throws JsonProcessingException {
   public void testDropMetalake() throws JsonProcessingException {
     DropResponse resp = new DropResponse(true);
     buildMockResource(Method.DELETE, "/api/metalakes/mock", null, resp, HttpStatus.SC_OK);
-    Assertions.assertTrue(client.dropMetalake(NameIdentifier.of("mock")));
+    Assertions.assertTrue(
+        client.dropMetalake(NameIdentifier.of("mock")), "metalake should be dropped");
 
     DropResponse resp1 = new DropResponse(false);
     buildMockResource(Method.DELETE, "/api/metalakes/mock", null, resp1, HttpStatus.SC_OK);
-    Assertions.assertFalse(client.dropMetalake(NameIdentifier.of("mock")));
+    Assertions.assertFalse(
+        client.dropMetalake(NameIdentifier.of("mock")), "metalake should be non-existent");
 
     // Test return internal error
     ErrorResponse errorResp = ErrorResponse.internalError("mock error");

clients/client-java/src/test/java/com/datastrato/gravitino/client/TestGravitinoMetalake.java

@@ -374,13 +374,13 @@ public void testDropCatalog() throws JsonProcessingException {
     DropResponse resp = new DropResponse(true);
     buildMockResource(Method.DELETE, path, null, resp, HttpStatus.SC_OK);
     boolean dropped = gravitinoClient.dropCatalog(NameIdentifier.of(metalakeName, catalogName));
-    Assertions.assertTrue(dropped);
+    Assertions.assertTrue(dropped, "catalog should be dropped");
 
     // Test return false
     DropResponse resp1 = new DropResponse(false);
     buildMockResource(Method.DELETE, path, null, resp1, HttpStatus.SC_OK);
     boolean dropped1 = gravitinoClient.dropCatalog(NameIdentifier.of(metalakeName, catalogName));
-    Assertions.assertFalse(dropped1);
+    Assertions.assertFalse(dropped1, "catalog should be non-existent");
   }
 
   static GravitinoMetalake createMetalake(GravitinoAdminClient client, String metalakeName)

integration-test/src/test/java/com/datastrato/gravitino/integration/test/client/CatalogIT.java

@@ -185,6 +185,9 @@ public void testListCatalogsInfo() {
     }
     Assertions.assertTrue(ArrayUtils.contains(catalogs, relCatalog));
     Assertions.assertTrue(ArrayUtils.contains(catalogs, fileCatalog));
+
+    metalake.dropCatalog(relCatalogIdent);
+    metalake.dropCatalog(fileCatalogIdent);
   }
 
   private void assertCatalogEquals(Catalog catalog1, Catalog catalog2) {

integration-test/src/test/java/com/datastrato/gravitino/integration/test/client/filesystem/hadoop/GravitinoVirtualFileSystemIT.java

@@ -78,6 +78,11 @@ public static void startUp() {
 
   @AfterAll
   public static void tearDown() throws IOException {
+    Catalog catalog = metalake.loadCatalog(NameIdentifier.ofCatalog(metalakeName, catalogName));
+    catalog
+        .asSchemas()
+        .dropSchema(NameIdentifier.ofSchema(metalakeName, catalogName, schemaName), true);
+    metalake.dropCatalog(NameIdentifier.of(metalakeName, catalogName));
     client.dropMetalake(NameIdentifier.of(metalakeName));
 
     if (client != null) {

integration-test/src/test/java/com/datastrato/gravitino/integration/test/spark/SparkCommonIT.java

@@ -141,9 +141,14 @@ void init() {
 
   @AfterAll
   void cleanUp() {
-    sql("USE " + getCatalogName());
-    getDatabases()
-        .forEach(database -> sql(String.format("DROP DATABASE IF EXISTS %s CASCADE", database)));
+    getDatabases().stream()
+        .filter(database -> !database.equals("default"))
+        .forEach(
+            database -> {
+              sql("USE " + database);
+              listTableNames().forEach(table -> dropTableIfExists(table));
+              dropDatabaseIfExists(database);
+            });
   }
 
   @Test

integration-test/src/test/java/com/datastrato/gravitino/integration/test/trino/TrinoConnectorIT.java

@@ -1083,11 +1083,16 @@ void testIcebergCatalogCreatedByGravitino() {
     }
 
     // Do not support the cascade drop
-    success =
-        catalog
-            .asSchemas()
-            .dropSchema(NameIdentifier.of(metalakeName, catalogName, schemaName), true);
-    Assertions.assertFalse(success);
+    Throwable excep =
+        Assertions.assertThrows(
+            IllegalArgumentException.class,
+            () ->
+                catalog
+                    .asSchemas()
+                    .dropSchema(NameIdentifier.of(metalakeName, catalogName, schemaName), true));
+    Assertions.assertTrue(
+        excep.getMessage().contains("Iceberg does not support cascading delete operations."));
+
     final String sql3 = String.format("show schemas in %s like '%s'", catalogName, schemaName);
     success = checkTrinoHasLoaded(sql3, 30);
     if (!success) {