[#1231] Fix bugs in drop MySQL schemas (#1251)

### What changes were proposed in this pull request? - Allow dropping a database when `cascase` is `true`. - If the database is not empty and `cascade` is `false`, disable the drop operation. ### Why are the changes needed? Align the logic with Gravitino EntityStore. Fix: #1231 ### Does this PR introduce _any_ user-facing change? N/A ### How was this patch tested? Add UT `testDropMySQLDatabase` Co-authored-by: Qi Yu <yuqi@datastrato.com>
apache · Dec 25, 2023 · ccea15c · ccea15c
1 parent 58cbf05
commit ccea15c
Show file tree

Hide file tree

Showing 3 changed files with 59 additions and 15 deletions.
diff --git a/...c/main/java/com/datastrato/gravitino/catalog/mysql/operation/MysqlDatabaseOperations.java b/...c/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

diff --git a/.../test/java/com/datastrato/gravitino/integration/test/catalog/jdbc/TestJdbcAbstractIT.java b/.../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));

diff --git a/...est/java/com/datastrato/gravitino/integration/test/catalog/jdbc/mysql/CatalogMysqlIT.java b/...est/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)));
+  }
 }