re-define drop catalog

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

@@ -98,6 +98,9 @@ enum CloudName {
    */
   String PROPERTY_PACKAGE = "package";
 
+  /** The property indicating the catalog is in use. */
+  String PROPERTY_IN_USE = "in-use";
+
   /**
    * The property to specify the cloud that the catalog is running on. The value should be one of
    * the {@link CloudName}.

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

@@ -21,8 +21,10 @@
 import java.util.Map;
 import org.apache.gravitino.annotation.Evolving;
 import org.apache.gravitino.exceptions.CatalogAlreadyExistsException;
+import org.apache.gravitino.exceptions.EntityInUseException;
 import org.apache.gravitino.exceptions.NoSuchCatalogException;
 import org.apache.gravitino.exceptions.NoSuchMetalakeException;
+import org.apache.gravitino.exceptions.NonEmptyEntityException;
 
 /**
  * Client interface for supporting catalogs. It includes methods for listing, loading, creating,
@@ -48,9 +50,9 @@ public interface SupportsCatalogs {
   Catalog[] listCatalogsInfo() throws NoSuchMetalakeException;
 
   /**
-   * Load a catalog by its identifier.
+   * Load a catalog by its name.
    *
-   * @param catalogName the identifier of the catalog.
+   * @param catalogName the name of the catalog.
    * @return The catalog.
    * @throws NoSuchCatalogException If the catalog does not exist.
    */
@@ -59,7 +61,7 @@ public interface SupportsCatalogs {
   /**
    * Check if a catalog exists.
    *
-   * @param catalogName The identifier of the catalog.
+   * @param catalogName The name of the catalog.
    * @return True if the catalog exists, false otherwise.
    */
   default boolean catalogExists(String catalogName) {
@@ -72,7 +74,7 @@ default boolean catalogExists(String catalogName) {
   }
 
   /**
-   * Create a catalog with specified identifier.
+   * Create a catalog with specified catalog name, type, provider, comment, and properties.
    *
    * <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}
@@ -96,9 +98,9 @@ Catalog createCatalog(
       throws NoSuchMetalakeException, CatalogAlreadyExistsException;
 
   /**
-   * Alter a catalog with specified identifier.
+   * Alter a catalog with specified catalog name and changes.
    *
-   * @param catalogName the identifier of the catalog.
+   * @param catalogName the name of the catalog.
    * @param changes the changes to apply to the catalog.
    * @return The altered catalog.
    * @throws NoSuchCatalogException If the catalog does not exist.
@@ -108,12 +110,72 @@ Catalog alterCatalog(String catalogName, CatalogChange... changes)
       throws NoSuchCatalogException, IllegalArgumentException;
 
   /**
-   * Drop a catalog with specified identifier.
+   * Drop a catalog with specified name. Please make sure:
+   *
+   * <ul>
+   *   <li>There is no schema in the catalog. Otherwise, a {@link NonEmptyEntityException} will be
+   *       thrown.
+   *   <li>The method {@link #deactivateCatalog(String)} has been called before dropping the
+   *       catalog. Otherwise, a {@link EntityInUseException} will be thrown.
+   * </ul>
+   *
+   * It is equivalent to calling {@code dropCatalog(ident, false)}.
    *
    * @param catalogName the name of the catalog.
-   * @return True if the catalog was dropped, false otherwise.
+   * @return True if the catalog was dropped, false if the catalog does not exist.
+   * @throws NonEmptyEntityException If the catalog is not empty.
+   * @throws EntityInUseException If the catalog is in use.
+   */
+  default boolean dropCatalog(String catalogName)
+      throws NonEmptyEntityException, EntityInUseException {
+    return dropCatalog(catalogName, false);
+  }
+
+  /**
+   * Drop a catalog with specified name. If the force flag is true, it will:
+   *
+   * <ul>
+   *   <li>Cascade drop all sub-entities (schemas, tables, etc.) of the catalog in Gravitino store.
+   *   <li>Drop the catalog even if it is in use.
+   *   <li>External resources (e.g. database, table, etc.) associated with sub-entities will not be
+   *       dropped unless it is managed (such as managed fileset).
+   * </ul>
+   *
+   * If the force flag is false, it is equivalent to calling {@link #dropCatalog(String)}.
+   *
+   * @param catalogName The identifier of the catalog.
+   * @param force Whether to force the drop.
+   * @return True if the catalog was dropped, false if the catalog does not exist.
+   * @throws NonEmptyEntityException If the catalog is not empty and force is false.
+   * @throws EntityInUseException If the catalog is in use and force is false.
+   */
+  boolean dropCatalog(String catalogName, boolean force)
+      throws NonEmptyEntityException, EntityInUseException;
+
+  /**
+   * Activate a catalog. If the catalog is already activated, this method does nothing.
+   *
+   * @param catalogName The identifier of the catalog.
+   * @throws NoSuchCatalogException If the catalog does not exist.
+   */
+  void activateCatalog(String catalogName) throws NoSuchCatalogException;
+
+  /**
+   * Deactivate a catalog. If the catalog is already deactivated, this method does nothing. Once a
+   * catalog is deactivated:
+   *
+   * <ul>
+   *   <li>It can only be listed, loaded, dropped, or activated.
+   *   <li>Any other operations on the catalog will throw an {@link
+   *       org.apache.gravitino.exceptions.NonInUseEntityException}.
+   *   <li>Any operation on the sub-entities (schemas, tables, etc.) will throw an {@link
+   *       org.apache.gravitino.exceptions.NonInUseEntityException}.
+   * </ul>
+   *
+   * @param catalogName The identifier of the catalog.
+   * @throws NoSuchCatalogException If the catalog does not exist.
    */
-  boolean dropCatalog(String catalogName);
+  void deactivateCatalog(String catalogName) throws NoSuchCatalogException;
 
   /**
    * Test whether the catalog with specified parameters can be connected to before creating it.

api/src/main/java/org/apache/gravitino/exceptions/EntityInUseException.java

@@ -0,0 +1,36 @@
+/*
+ * 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.
+ */
+package org.apache.gravitino.exceptions;
+
+import com.google.errorprone.annotations.FormatMethod;
+import com.google.errorprone.annotations.FormatString;
+
+/** Exception thrown when an entity is in use and cannot be deleted. */
+public class EntityInUseException extends GravitinoRuntimeException {
+  /**
+   * Constructs a new exception with the specified detail message.
+   *
+   * @param message the detail message.
+   * @param args the arguments to the message.
+   */
+  @FormatMethod
+  public EntityInUseException(@FormatString String message, Object... args) {
+    super(message, args);
+  }
+}

api/src/main/java/org/apache/gravitino/exceptions/NonInUseEntityException.java

@@ -0,0 +1,36 @@
+/*
+ * 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.
+ */
+package org.apache.gravitino.exceptions;
+
+import com.google.errorprone.annotations.FormatMethod;
+import com.google.errorprone.annotations.FormatString;
+
+/** An exception thrown when operating on a non-in-use entity. */
+public class NonInUseEntityException extends GravitinoRuntimeException {
+  /**
+   * Constructs a new exception with the specified detail message.
+   *
+   * @param message the detail message.
+   * @param args the arguments to the message.
+   */
+  @FormatMethod
+  public NonInUseEntityException(@FormatString String message, Object... args) {
+    super(message, args);
+  }
+}

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

@@ -189,10 +189,7 @@ public static void stop() throws IOException {
                 catalog.asSchemas().dropSchema(schema, true);
               }));
       Arrays.stream(metalake.listCatalogs())
-          .forEach(
-              (catalogName -> {
-                metalake.dropCatalog(catalogName);
-              }));
+          .forEach((catalogName -> metalake.dropCatalog(catalogName, true)));
       client.dropMetalake(metalakeName);
     }
     if (sparkSession != null) {

catalogs/catalog-hadoop/src/main/java/org/apache/gravitino/catalog/hadoop/SecureHadoopCatalogOperations.java

@@ -156,6 +156,10 @@ public boolean dropSchema(NameIdentifier ident, boolean cascade) throws NonEmpty
       UserContext.clearUserContext(ident);
 
       return r;
+    } catch (NoSuchEntityException e) {
+      LOG.warn("Schema {} does not exist", ident);
+      return false;
+
     } catch (IOException ioe) {
       throw new RuntimeException("Failed to delete schema " + ident, ioe);
     }

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

@@ -89,7 +89,7 @@ public static void setup() throws IOException {
   public static void stop() throws IOException {
     Catalog catalog = metalake.loadCatalog(catalogName);
     catalog.asSchemas().dropSchema(schemaName, true);
-    metalake.dropCatalog(catalogName);
+    metalake.dropCatalog(catalogName, true);
     client.dropMetalake(metalakeName);
     if (hdfs != null) {
       hdfs.close();
@@ -162,7 +162,7 @@ void testAlterCatalogLocation() {
 
     Assertions.assertEquals(newLocation, modifiedCatalog.properties().get("location"));
 
-    metalake.dropCatalog(catalogName);
+    metalake.dropCatalog(catalogName, true);
   }
 
   @Test
@@ -608,7 +608,7 @@ public void testDropCatalogWithEmptySchema() {
         filesetCatalog.asSchemas().schemaExists(schemaName), "schema should not be exists");
 
     // Drop the catalog.
-    dropped = metalake.dropCatalog(catalogName);
+    dropped = metalake.dropCatalog(catalogName, true);
     Assertions.assertTrue(dropped, "catalog should be dropped");
     Assertions.assertFalse(metalake.catalogExists(catalogName), "catalog should not be exists");
   }

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

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

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

@@ -22,6 +22,7 @@
 import static org.apache.gravitino.Catalog.AUTHORIZATION_PROVIDER;
 import static org.apache.gravitino.Catalog.CLOUD_NAME;
 import static org.apache.gravitino.Catalog.CLOUD_REGION_CODE;
+import static org.apache.gravitino.Catalog.PROPERTY_IN_USE;
 import static org.apache.gravitino.catalog.hive.HiveCatalogPropertiesMeta.CHECK_INTERVAL_SEC;
 import static org.apache.gravitino.catalog.hive.HiveCatalogPropertiesMeta.CLIENT_POOL_CACHE_EVICTION_INTERVAL_MS;
 import static org.apache.gravitino.catalog.hive.HiveCatalogPropertiesMeta.CLIENT_POOL_SIZE;
@@ -73,10 +74,11 @@ void testPropertyMeta() {
     Map<String, PropertyEntry<?>> propertyEntryMap =
         HIVE_PROPERTIES_METADATA.catalogPropertiesMetadata().propertyEntries();
 
-    Assertions.assertEquals(20, propertyEntryMap.size());
+    Assertions.assertEquals(21, propertyEntryMap.size());
     Assertions.assertTrue(propertyEntryMap.containsKey(METASTORE_URIS));
     Assertions.assertTrue(propertyEntryMap.containsKey(Catalog.PROPERTY_PACKAGE));
     Assertions.assertTrue(propertyEntryMap.containsKey(BaseCatalog.CATALOG_OPERATION_IMPL));
+    Assertions.assertTrue(propertyEntryMap.containsKey(PROPERTY_IN_USE));
     Assertions.assertTrue(propertyEntryMap.containsKey(AUTHORIZATION_PROVIDER));
     Assertions.assertTrue(propertyEntryMap.containsKey(CLIENT_POOL_SIZE));
     Assertions.assertTrue(propertyEntryMap.containsKey(IMPERSONATION_ENABLE));

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

@@ -230,10 +230,7 @@ public void stop() throws IOException {
                 catalog.asSchemas().dropSchema(schema, true);
               }));
       Arrays.stream(metalake.listCatalogs())
-          .forEach(
-              (catalogName -> {
-                metalake.dropCatalog(catalogName);
-              }));
+          .forEach((catalogName -> metalake.dropCatalog(catalogName, true)));
       client.dropMetalake(metalakeName);
     }
     if (hiveClientPool != null) {
@@ -1674,7 +1671,7 @@ void testAlterCatalogProperties() {
         });
 
     newCatalog.asSchemas().dropSchema("schema", true);
-    metalake.dropCatalog(nameOfCatalog);
+    metalake.dropCatalog(nameOfCatalog, true);
   }
 
   private void createCatalogWithCustomOperation(String catalogName, String customImpl) {

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

@@ -271,7 +271,7 @@ public void testUserAuthentication() {
     Assertions.assertFalse(catalog.asSchemas().schemaExists(SCHEMA_NAME));
 
     // Drop catalog
-    Assertions.assertTrue(gravitinoMetalake.dropCatalog(CATALOG_NAME));
+    Assertions.assertTrue(gravitinoMetalake.dropCatalog(CATALOG_NAME, true));
   }
 
   @AfterAll

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

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

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

@@ -95,7 +95,7 @@ public void testAuditCatalog() throws Exception {
     Assertions.assertEquals(expectUser, catalog.auditInfo().creator());
     Assertions.assertEquals(expectUser, catalog.auditInfo().lastModifier());
 
-    metalake.dropCatalog(catalogName);
+    metalake.dropCatalog(catalogName, true);
   }
 
   @Test
@@ -109,7 +109,7 @@ public void testAuditSchema() throws Exception {
     Assertions.assertNull(schema.auditInfo().lastModifier());
 
     catalog.asSchemas().dropSchema(schemaName, true);
-    metalake.dropCatalog(catalogName);
+    metalake.dropCatalog(catalogName, true);
   }
 
   @Test
@@ -144,7 +144,7 @@ public void testAuditTable() throws Exception {
 
     catalog.asTableCatalog().dropTable(NameIdentifier.of(schemaName, tableName));
     catalog.asSchemas().dropSchema(schemaName, true);
-    metalake.dropCatalog(catalogName);
+    metalake.dropCatalog(catalogName, true);
   }
 
   private static Catalog createCatalog(String catalogName) throws SQLException {

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

@@ -142,6 +142,7 @@ public void startup() throws IOException, SQLException {
   @AfterAll
   public void stop() {
     clearTableAndSchema();
+    metalake.deactivateCatalog(catalogName);
     metalake.dropCatalog(catalogName);
     client.dropMetalake(metalakeName);
     mysqlService.close();
@@ -1987,6 +1988,6 @@ void testAlterCatalogProperties() throws SQLException {
     Assertions.assertDoesNotThrow(() -> loadCatalog.asSchemas().createSchema("test", "", null));
 
     loadCatalog.asSchemas().dropSchema("test", true);
-    metalake.dropCatalog(testCatalogName);
+    metalake.dropCatalog(testCatalogName, true);
   }
 }

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

@@ -128,6 +128,7 @@ public void stop() {
     for (String schemaName : schemaNames) {
       catalog.asSchemas().dropSchema(schemaName, true);
     }
+    metalake.deactivateCatalog(catalogName);
     metalake.dropCatalog(catalogName);
     client.dropMetalake(metalakeName);
     postgreSqlService.close();

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

@@ -123,6 +123,7 @@ public static void shutdown() {
     Arrays.stream(metalake.listCatalogs())
         .forEach(
             (catalogName -> {
+              metalake.deactivateCatalog(catalogName);
               metalake.dropCatalog(catalogName);
             }));
     client.dropMetalake(METALAKE_NAME);
@@ -171,7 +172,7 @@ public void testCatalog() throws ExecutionException, InterruptedException {
     Assertions.assertFalse(alteredCatalog.properties().containsKey("key1"));
 
     // test drop catalog
-    boolean dropped = metalake.dropCatalog(catalogName);
+    boolean dropped = metalake.dropCatalog(catalogName, true);
     Assertions.assertTrue(dropped);
     Exception exception =
         Assertions.assertThrows(