[#1231] Fix bugs in drop MySQL schemas

catalogs/catalog-jdbc-mysql/src/main/java/com/datastrato/gravitino/catalog/mysql/operation/MysqlDatabaseOperations.java

@@ -12,6 +12,7 @@
 import java.sql.PreparedStatement;
 import java.sql.ResultSet;
 import java.sql.SQLException;
+import java.sql.Statement;
 import java.util.HashMap;
 import java.util.Map;
 import org.apache.commons.collections4.MapUtils;
@@ -47,11 +48,28 @@ public String generateCreateDatabaseSql(
 
   @Override
   public String generateDropDatabaseSql(String databaseName, boolean cascade) {
+    final String dropDatabaseSql = "DROP DATABASE `" + databaseName + "`";
     if (cascade) {
-      throw new UnsupportedOperationException(
-          "MySQL does not support CASCADE option for DROP DATABASE.");
+      return dropDatabaseSql;
     }
-    return "DROP DATABASE `" + databaseName + "`";
+
+    try (final Connection connection = this.dataSource.getConnection()) {
+      String query = "SHOW TABLES IN " + databaseName;
+      try (Statement statement = connection.createStatement()) {
+        // Execute the query and check if there exists any tables in the database
+        try (ResultSet resultSet = statement.executeQuery(query)) {
+          if (resultSet.next()) {
+            throw new IllegalStateException(
+                String.format(
+                    "Database %s is not empty, the value of cascade should be true.",
+                    databaseName));
+          }
+        }
+      }
+    } catch (SQLException sqlException) {
+      throw this.exceptionMapper.toGravitinoException(sqlException);
+    }
+    return dropDatabaseSql;
   }
 
   @Override

integration-test/src/test/java/com/datastrato/gravitino/integration/test/catalog/jdbc/TestJdbcAbstractIT.java

@@ -61,17 +61,7 @@ protected void testBaseOperation(
 
   protected static void testDropDatabase(String databaseName) {
     List<String> databases;
-    // delete database.
-    UnsupportedOperationException unsupportedOperationException =
-        Assertions.assertThrows(
-            UnsupportedOperationException.class,
-            () -> DATABASE_OPERATIONS.delete(databaseName, true));
-    Assertions.assertTrue(
-        unsupportedOperationException
-            .getMessage()
-            .contains("does not support CASCADE option for DROP DATABASE."));
-
-    Assertions.assertDoesNotThrow(() -> DATABASE_OPERATIONS.delete(databaseName, false));
+    DATABASE_OPERATIONS.delete(databaseName, true);
 
     Assertions.assertThrows(
         NoSuchSchemaException.class, () -> DATABASE_OPERATIONS.load(databaseName));

integration-test/src/test/java/com/datastrato/gravitino/integration/test/catalog/jdbc/mysql/CatalogMysqlIT.java

@@ -123,7 +123,7 @@ private static void clearTableAndSchema() {
     NameIdentifier[] nameIdentifiers =
         catalog.asTableCatalog().listTables(Namespace.of(metalakeName, catalogName, schemaName));
     for (NameIdentifier nameIdentifier : nameIdentifiers) {
-      catalog.asTableCatalog().dropTable(nameIdentifier);
+      catalog.asTableCatalog().purgeTable(nameIdentifier);
     }
     catalog.asSchemas().dropSchema(NameIdentifier.of(metalakeName, catalogName, schemaName), false);
   }
@@ -458,4 +458,40 @@ void testAlterAndDropMysqlTable() {
           catalog.asTableCatalog().dropTable(tableIdentifier);
         });
   }
+
+  @Test
+  void testDropMySQLDatabase() {
+    String schemaName = GravitinoITUtils.genRandomName("mysql_schema").toLowerCase();
+    String tableName = GravitinoITUtils.genRandomName("mysql_table").toLowerCase();
+
+    catalog
+        .asSchemas()
+        .createSchema(
+            NameIdentifier.of(metalakeName, catalogName, schemaName),
+            "Created by gravitino client",
+            ImmutableMap.<String, String>builder().build());
+
+    catalog
+        .asTableCatalog()
+        .createTable(
+            NameIdentifier.of(metalakeName, catalogName, schemaName, tableName),
+            createColumns(),
+            "Created by gravitino client",
+            ImmutableMap.<String, String>builder().build());
+
+    // Try to drop a database, and cascade equals to false, it should not be allowed.
+    catalog.asSchemas().dropSchema(NameIdentifier.of(metalakeName, catalogName, schemaName), false);
+    // Check the database still exists
+    catalog.asSchemas().loadSchema(NameIdentifier.of(metalakeName, catalogName, schemaName));
+
+    // Try to drop a database, and cascade equals to true, it should be allowed.
+    catalog.asSchemas().dropSchema(NameIdentifier.of(metalakeName, catalogName, schemaName), true);
+    // Check database has been dropped
+    Assertions.assertThrows(
+        NoSuchSchemaException.class,
+        () ->
+            catalog
+                .asSchemas()
+                .loadSchema(NameIdentifier.of(metalakeName, catalogName, schemaName)));
+  }
 }