re-define drop metalake

api/src/main/java/org/apache/gravitino/Metalake.java

@@ -29,6 +29,9 @@
 @Evolving
 public interface Metalake extends Auditable {
 
+  /** The property indicating the metalake is in use. */
+  String PROPERTY_IN_USE = "in-use";
+
   /**
    * The name of the metalake.
    *

api/src/main/java/org/apache/gravitino/SupportsMetalakes.java

@@ -20,8 +20,10 @@
 
 import java.util.Map;
 import org.apache.gravitino.annotation.Evolving;
+import org.apache.gravitino.exceptions.EntityInUseException;
 import org.apache.gravitino.exceptions.MetalakeAlreadyExistsException;
 import org.apache.gravitino.exceptions.NoSuchMetalakeException;
+import org.apache.gravitino.exceptions.NonEmptyEntityException;
 
 /**
  * Client interface for supporting metalakes. It includes methods for listing, loading, creating,
@@ -38,7 +40,7 @@ public interface SupportsMetalakes {
   Metalake[] listMetalakes();
 
   /**
-   * Load a metalake by its identifier.
+   * Load a metalake by its name.
    *
    * @param name the name of the metalake.
    * @return The metalake.
@@ -62,7 +64,7 @@ default boolean metalakeExists(String name) {
   }
 
   /**
-   * Create a metalake with specified identifier.
+   * Create a metalake with specified name, comment and properties.
    *
    * @param name The name of the metalake.
    * @param comment The comment of the metalake.
@@ -74,7 +76,7 @@ Metalake createMetalake(String name, String comment, Map<String, String> propert
       throws MetalakeAlreadyExistsException;
 
   /**
-   * Alter a metalake with specified identifier.
+   * Alter a metalake with specified metalake name and changes.
    *
    * @param name The name of the metalake.
    * @param changes The changes to apply.
@@ -86,10 +88,70 @@ Metalake alterMetalake(String name, MetalakeChange... changes)
       throws NoSuchMetalakeException, IllegalArgumentException;
 
   /**
-   * Drop a metalake with specified identifier.
+   * Drop a metalake with specified name. Please make sure:
    *
-   * @param name The identifier of the metalake.
+   * <ul>
+   *   <li>There is no catalog in the metalake. Otherwise, a {@link NonEmptyEntityException} will be
+   *       thrown.
+   *   <li>The method {@link #deactivateMetalake(String)} has been called before dropping the
+   *       metalake. Otherwise, a {@link EntityInUseException} will be thrown.
+   * </ul>
+   *
+   * It is equivalent to calling {@code dropMetalake(ident, false)}.
+   *
+   * @param name The name of the metalake.
    * @return True if the metalake was dropped, false if the metalake does not exist.
+   * @throws NonEmptyEntityException If the metalake is not empty.
+   * @throws EntityInUseException If the metalake is in use.
+   */
+  default boolean dropMetalake(String name) throws NonEmptyEntityException, EntityInUseException {
+    return dropMetalake(name, false);
+  }
+
+  /**
+   * Drop a metalake with specified name. If the force flag is true, it will:
+   *
+   * <ul>
+   *   <li>Cascade drop all sub-entities (tags, catalogs, schemas, tables, etc.) of the metalake in
+   *       Gravitino store.
+   *   <li>Drop the metalake even if it is in use.
+   *   <li>External resources (e.g. database, table, etc.) associated with sub-entities will not be
+   *       deleted unless it is managed (such as managed fileset).
+   * </ul>
+   *
+   * If the force flag is false, it is equivalent to calling {@link #dropMetalake(String)}.
+   *
+   * @param name The name of the metalake.
+   * @param force Whether to force the drop.
+   * @return True if the metalake was dropped, false if the metalake does not exist.
+   * @throws NonEmptyEntityException If the metalake is not empty and force is false.
+   * @throws EntityInUseException If the metalake is in use and force is false.
+   */
+  boolean dropMetalake(String name, boolean force)
+      throws NonEmptyEntityException, EntityInUseException;
+
+  /**
+   * Activate a metalake. If the metalake is already active, this method does nothing.
+   *
+   * @param name The name of the metalake.
+   * @throws NoSuchMetalakeException If the metalake does not exist.
+   */
+  void activateMetalake(String name) throws NoSuchMetalakeException;
+
+  /**
+   * Deactivate a metalake. If the metalake is already inactive, this method does nothing. Once a
+   * metalake is deactivated:
+   *
+   * <ul>
+   *   <li>It can only be listed, loaded, dropped, or activated.
+   *   <li>Any other operations on the metalake will throw an {@link
+   *       org.apache.gravitino.exceptions.NonInUseEntityException}.
+   *   <li>Any operation on the sub-entities (catalogs, schemas, tables, etc.) will throw an {@link
+   *       org.apache.gravitino.exceptions.NonInUseEntityException}.
+   * </ul>
+   *
+   * @param name The name of the metalake.
+   * @throws NoSuchMetalakeException If the metalake does not exist.
    */
-  boolean dropMetalake(String name);
+  void deactivateMetalake(String name) throws NoSuchMetalakeException;
 }

authorizations/authorization-ranger/src/test/java/org/apache/gravitino/authorization/ranger/integration/test/RangerHiveE2EIT.java

@@ -191,6 +191,7 @@ public void stop() throws IOException {
               }));
       Arrays.stream(metalake.listCatalogs())
           .forEach((catalogName -> metalake.dropCatalog(catalogName, true)));
+      client.deactivateMetalake(metalakeName);
       client.dropMetalake(metalakeName);
     }
     if (sparkSession != null) {
@@ -269,10 +270,9 @@ private void createMetalake() {
     GravitinoMetalake[] gravitinoMetalakes = client.listMetalakes();
     Assertions.assertEquals(0, gravitinoMetalakes.length);
 
-    GravitinoMetalake createdMetalake =
-        client.createMetalake(metalakeName, "comment", Collections.emptyMap());
+    client.createMetalake(metalakeName, "comment", Collections.emptyMap());
     GravitinoMetalake loadMetalake = client.loadMetalake(metalakeName);
-    Assertions.assertEquals(createdMetalake, loadMetalake);
+    Assertions.assertEquals(metalakeName, loadMetalake.name());
 
     metalake = loadMetalake;
   }

catalogs/catalog-hadoop/src/test/java/org/apache/gravitino/catalog/hadoop/integration/test/HadoopCatalogIT.java

@@ -90,7 +90,7 @@ public void stop() throws IOException {
     Catalog catalog = metalake.loadCatalog(catalogName);
     catalog.asSchemas().dropSchema(schemaName, true);
     metalake.dropCatalog(catalogName, true);
-    client.dropMetalake(metalakeName);
+    client.dropMetalake(metalakeName, true);
     if (hdfs != null) {
       hdfs.close();
     }
@@ -106,10 +106,9 @@ private void createMetalake() {
     GravitinoMetalake[] gravitinoMetalakes = client.listMetalakes();
     Assertions.assertEquals(0, gravitinoMetalakes.length);
 
-    GravitinoMetalake createdMetalake =
-        client.createMetalake(metalakeName, "comment", Collections.emptyMap());
+    client.createMetalake(metalakeName, "comment", Collections.emptyMap());
     GravitinoMetalake loadMetalake = client.loadMetalake(metalakeName);
-    Assertions.assertEquals(createdMetalake, loadMetalake);
+    Assertions.assertEquals(metalakeName, loadMetalake.name());
 
     metalake = loadMetalake;
   }

catalogs/catalog-hadoop/src/test/java/org/apache/gravitino/catalog/hadoop/integration/test/HadoopUserAuthenticationIT.java

@@ -654,6 +654,6 @@ void testUserImpersonation() {
     catalog.asFilesetCatalog().dropFileset(NameIdentifier.of(SCHEMA_NAME, filesetName));
     catalog.asSchemas().dropSchema(SCHEMA_NAME, true);
     gravitinoMetalake.dropCatalog(catalogName, true);
-    adminClient.dropMetalake(metalakeName);
+    adminClient.dropMetalake(metalakeName, true);
   }
 }

catalogs/catalog-hadoop/src/test/java/org/apache/gravitino/catalog/hadoop/integration/test/HadoopUserImpersonationIT.java

@@ -258,10 +258,9 @@ private void createMetalake() {
     GravitinoMetalake[] gravitinoMetalakes = client.listMetalakes();
     Assertions.assertEquals(0, gravitinoMetalakes.length);
 
-    GravitinoMetalake createdMetalake =
-        client.createMetalake(metalakeName, "comment", Collections.emptyMap());
+    client.createMetalake(metalakeName, "comment", Collections.emptyMap());
     GravitinoMetalake loadMetalake = client.loadMetalake(metalakeName);
-    Assertions.assertEquals(createdMetalake, loadMetalake);
+    Assertions.assertEquals(metalakeName, loadMetalake.name());
 
     metalake = loadMetalake;
   }

catalogs/catalog-hive/src/test/java/org/apache/gravitino/catalog/hive/integration/test/CatalogHiveIT.java

@@ -229,7 +229,7 @@ public void stop() throws IOException {
               }));
       Arrays.stream(metalake.listCatalogs())
           .forEach((catalogName -> metalake.dropCatalog(catalogName, true)));
-      client.dropMetalake(metalakeName);
+      client.dropMetalake(metalakeName, true);
     }
     if (hiveClientPool != null) {
       hiveClientPool.close();
@@ -264,10 +264,9 @@ private void createMetalake() {
     GravitinoMetalake[] gravitinoMetalakes = client.listMetalakes();
     Assertions.assertEquals(0, gravitinoMetalakes.length);
 
-    GravitinoMetalake createdMetalake =
-        client.createMetalake(metalakeName, "comment", Collections.emptyMap());
+    client.createMetalake(metalakeName, "comment", Collections.emptyMap());
     GravitinoMetalake loadMetalake = client.loadMetalake(metalakeName);
-    Assertions.assertEquals(createdMetalake, loadMetalake);
+    Assertions.assertEquals(metalakeName, loadMetalake.name());
 
     metalake = loadMetalake;
   }
@@ -1429,8 +1428,8 @@ void testDropAndRename() {
     client.createMetalake(metalakeName1, "comment", Collections.emptyMap());
     client.createMetalake(metalakeName2, "comment", Collections.emptyMap());
 
-    client.dropMetalake(metalakeName1);
-    client.dropMetalake(metalakeName2);
+    client.dropMetalake(metalakeName1, true);
+    client.dropMetalake(metalakeName2, true);
 
     client.createMetalake(metalakeName1, "comment", Collections.emptyMap());
 

catalogs/catalog-jdbc-doris/src/test/java/org/apache/gravitino/catalog/doris/integration/test/CatalogDorisIT.java

@@ -126,7 +126,7 @@ public void startup() throws IOException {
   public void stop() {
     clearTableAndSchema();
     metalake.dropCatalog(catalogName, true);
-    client.dropMetalake(metalakeName);
+    client.dropMetalake(metalakeName, true);
   }
 
   @AfterEach

catalogs/catalog-jdbc-mysql/src/test/java/org/apache/gravitino/catalog/mysql/integration/test/AuditCatalogMysqlIT.java

@@ -77,7 +77,7 @@ public void startIntegrationTest() throws Exception {
 
   @AfterAll
   public void stopIntegrationTest() throws IOException, InterruptedException {
-    client.dropMetalake(metalakeName);
+    client.dropMetalake(metalakeName, true);
     mysqlService.close();
     super.stopIntegrationTest();
   }

catalogs/catalog-jdbc-mysql/src/test/java/org/apache/gravitino/catalog/mysql/integration/test/CatalogMysqlIT.java

@@ -144,6 +144,7 @@ public void stop() {
     clearTableAndSchema();
     metalake.deactivateCatalog(catalogName);
     metalake.dropCatalog(catalogName);
+    client.deactivateMetalake(metalakeName);
     client.dropMetalake(metalakeName);
     mysqlService.close();
   }

catalogs/catalog-jdbc-postgresql/src/test/java/org/apache/gravitino/catalog/postgresql/integration/test/CatalogPostgreSqlIT.java

@@ -130,6 +130,7 @@ public void stop() {
     }
     metalake.deactivateCatalog(catalogName);
     metalake.dropCatalog(catalogName);
+    client.deactivateMetalake(metalakeName);
     client.dropMetalake(metalakeName);
     postgreSqlService.close();
   }

catalogs/catalog-kafka/src/test/java/org/apache/gravitino/catalog/kafka/integration/test/CatalogKafkaIT.java

@@ -126,6 +126,7 @@ public void shutdown() {
               metalake.deactivateCatalog(catalogName);
               metalake.dropCatalog(catalogName);
             }));
+    client.deactivateMetalake(METALAKE_NAME);
     client.dropMetalake(METALAKE_NAME);
     if (adminClient != null) {
       adminClient.close();

catalogs/catalog-lakehouse-iceberg/src/test/java/org/apache/gravitino/catalog/lakehouse/iceberg/integration/test/CatalogIcebergBaseIT.java

@@ -140,6 +140,7 @@ public void stop() throws Exception {
       clearTableAndSchema();
       metalake.deactivateCatalog(catalogName);
       metalake.dropCatalog(catalogName);
+      client.deactivateMetalake(metalakeName);
       client.dropMetalake(metalakeName);
     } finally {
       if (spark != null) {

catalogs/catalog-lakehouse-paimon/src/test/java/org/apache/gravitino/catalog/lakehouse/paimon/integration/test/CatalogPaimonBaseIT.java

@@ -138,6 +138,7 @@ public void stop() {
     clearTableAndSchema();
     metalake.deactivateCatalog(catalogName);
     metalake.dropCatalog(catalogName);
+    client.deactivateMetalake(metalakeName);
     client.dropMetalake(metalakeName);
     if (spark != null) {
       spark.close();

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

@@ -22,6 +22,7 @@
 import com.google.common.base.Preconditions;
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.stream.Collectors;
@@ -31,10 +32,13 @@
 import org.apache.gravitino.dto.requests.MetalakeUpdateRequest;
 import org.apache.gravitino.dto.requests.MetalakeUpdatesRequest;
 import org.apache.gravitino.dto.responses.DropResponse;
+import org.apache.gravitino.dto.responses.ErrorResponse;
 import org.apache.gravitino.dto.responses.MetalakeListResponse;
 import org.apache.gravitino.dto.responses.MetalakeResponse;
+import org.apache.gravitino.exceptions.EntityInUseException;
 import org.apache.gravitino.exceptions.MetalakeAlreadyExistsException;
 import org.apache.gravitino.exceptions.NoSuchMetalakeException;
+import org.apache.gravitino.exceptions.NonEmptyEntityException;
 
 /**
  * Apache Gravitino Client for the administrator to interact with the Gravitino API, allowing the
@@ -145,24 +149,72 @@ public GravitinoMetalake alterMetalake(String name, MetalakeChange... changes)
   }
 
   /**
-   * Drops a specific Metalake using the Gravitino API.
+   * Drop a metalake with specified name. Please make sure:
    *
-   * @param name The name of the Metalake to be dropped.
-   * @return True if the Metalake was successfully dropped, false if the Metalake does not exist.
+   * <ul>
+   *   <li>There is no catalog in the metalake. Otherwise, a {@link NonEmptyEntityException} will be
+   *       thrown.
+   *   <li>The method {@link #deactivateMetalake(String)} has been called before dropping the
+   *       metalake. Otherwise, a {@link EntityInUseException} will be thrown.
+   * </ul>
+   *
+   * It is equivalent to calling {@code dropMetalake(ident, false)}.
+   *
+   * @param name The name of the metalake.
+   * @return True if the metalake was dropped, false if the metalake does not exist.
+   * @throws NonEmptyEntityException If the metalake is not empty.
+   * @throws EntityInUseException If the metalake is in use.
    */
   @Override
-  public boolean dropMetalake(String name) {
+  public boolean dropMetalake(String name, boolean force)
+      throws NonEmptyEntityException, EntityInUseException {
     checkMetalakeName(name);
+    Map<String, String> params = new HashMap<>();
+    params.put("force", String.valueOf(force));
+
     DropResponse resp =
         restClient.delete(
             API_METALAKES_IDENTIFIER_PATH + name,
+            params,
             DropResponse.class,
             Collections.emptyMap(),
             ErrorHandlers.metalakeErrorHandler());
     resp.validate();
     return resp.dropped();
   }
 
+  @Override
+  public void activateMetalake(String name) throws NoSuchMetalakeException {
+    ErrorResponse resp =
+        restClient.get(
+            API_METALAKES_IDENTIFIER_PATH + name + "/activate",
+            ErrorResponse.class,
+            Collections.emptyMap(),
+            ErrorHandlers.metalakeErrorHandler());
+
+    if (resp.getCode() == 0) {
+      return;
+    }
+
+    ErrorHandlers.metalakeErrorHandler().accept(resp);
+  }
+
+  @Override
+  public void deactivateMetalake(String name) throws NoSuchMetalakeException {
+    ErrorResponse resp =
+        restClient.get(
+            API_METALAKES_IDENTIFIER_PATH + name + "/deactivate",
+            ErrorResponse.class,
+            Collections.emptyMap(),
+            ErrorHandlers.metalakeErrorHandler());
+
+    if (resp.getCode() == 0) {
+      return;
+    }
+
+    ErrorHandlers.metalakeErrorHandler().accept(resp);
+  }
+
   /**
    * Creates a new builder for constructing a GravitinoClient.
    *

clients/client-java/src/test/java/org/apache/gravitino/client/integration/test/AuditIT.java

@@ -60,6 +60,7 @@ public void testAuditMetalake() {
     metaLake = client.alterMetalake(metalakeAuditName, changes);
     Assertions.assertEquals(expectUser, metaLake.auditInfo().creator());
     Assertions.assertEquals(expectUser, metaLake.auditInfo().lastModifier());
+    Assertions.assertDoesNotThrow(() -> client.deactivateMetalake(newName));
     Assertions.assertTrue(client.dropMetalake(newName), "metaLake should be dropped");
     Assertions.assertFalse(client.dropMetalake(newName), "metalake should be non-existent");
   }