[#3269] fix(jdbc-mysql): Solve DatabaseMetaData#getSchema return null…

… cause list tables error on some jdbc 8.x driver (#3294)

<!--
1. Title: [#<issue>] <type>(<scope>): <subject>
   Examples:
     - "[#123] feat(operator): support xxx"
     - "[#233] fix: check null before access result in xxx"
     - "[MINOR] refactor: fix typo in variable name"
     - "[MINOR] docs: fix typo in README"
     - "[#255] test: fix flaky test NameOfTheTest"
   Reference: https://www.conventionalcommits.org/en/v1.0.0/
2. If the PR is unfinished, please mark this PR as draft.
-->

### What changes were proposed in this pull request?

replace `DatabaseMetaData#getSchema` by `DatabaseMetaData#getCatalog`
and provide a more general way to get tables through jdbc driver

### Why are the changes needed?

jdbc driver `DatabaseMetaData#getSchema` method return null causes list
tables failure on some jdbc mysql 8.x driver
(mysql-connector-java-8.0.11).

Fix: #3269

### Does this PR introduce _any_ user-facing change?

not

### How was this patch tested?

(Please test your changes, and provide instructions on how to test it:
1. If you add a feature or fix a bug, add a test to cover your changes.
2. If you fix a flaky test, repeat it for many times to prove it works.)

---------

Co-authored-by: wangqi <[email protected]>

catalogs/catalog-jdbc-common/src/main/java/com/datastrato/gravitino/catalog/jdbc/operation/JdbcTableOperations.java

@@ -261,8 +261,9 @@ public void purge(String databaseName, String tableName) throws NoSuchTableExcep
 
   protected ResultSet getTables(Connection connection) throws SQLException {
     final DatabaseMetaData metaData = connection.getMetaData();
-    String databaseName = connection.getSchema();
-    return metaData.getTables(databaseName, databaseName, null, JdbcConnectorUtils.getTableTypes());
+    String catalogName = connection.getCatalog();
+    String schemaName = connection.getSchema();
+    return metaData.getTables(catalogName, schemaName, null, JdbcConnectorUtils.getTableTypes());
   }
 
   protected ResultSet getTable(Connection connection, String databaseName, String tableName)

catalogs/catalog-jdbc-doris/src/test/java/com/datastrato/gravitino/catalog/doris/integration/test/CatalogDorisDriverIT.java

@@ -0,0 +1,16 @@
+/*
+ * Copyright 2024 Datastrato Pvt Ltd.
+ * This software is licensed under the Apache License version 2.
+ */
+package com.datastrato.gravitino.catalog.doris.integration.test;
+
+import org.junit.jupiter.api.Tag;
+
+@Tag("gravitino-docker-it")
+public class CatalogDorisDriverIT extends CatalogDorisIT {
+  public CatalogDorisDriverIT() {
+    super();
+    mysqlDriverDownloadUrl =
+        "https://repo1.maven.org/maven2/mysql/mysql-connector-java/8.0.11/mysql-connector-java-8.0.11.jar";
+  }
+}

catalogs/catalog-jdbc-doris/src/test/java/com/datastrato/gravitino/catalog/doris/integration/test/CatalogDorisIT.java

@@ -88,13 +88,15 @@ public class CatalogDorisIT extends AbstractIT {
 
   protected Catalog catalog;
 
+  protected String mysqlDriverDownloadUrl = DOWNLOAD_JDBC_DRIVER_URL;
+
   @BeforeAll
   public void startup() throws IOException {
 
     if (!ITUtils.EMBEDDED_TEST_MODE.equals(AbstractIT.testMode)) {
       String gravitinoHome = System.getenv("GRAVITINO_HOME");
       Path tmpPath = Paths.get(gravitinoHome, "/catalogs/jdbc-doris/libs");
-      JdbcDriverDownloader.downloadJdbcDriver(DOWNLOAD_JDBC_DRIVER_URL, tmpPath.toString());
+      JdbcDriverDownloader.downloadJdbcDriver(mysqlDriverDownloadUrl, tmpPath.toString());
     }
 
     containerSuite.startDorisContainer();

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

@@ -0,0 +1,17 @@
+/*
+ *  Copyright 2024 Datastrato Pvt Ltd.
+ *  This software is licensed under the Apache License version 2.
+ */
+
+package com.datastrato.gravitino.catalog.mysql.integration.test;
+
+import org.junit.jupiter.api.Tag;
+
+@Tag("gravitino-docker-it")
+public class CatalogMysqlDriverIT extends CatalogMysqlIT {
+  public CatalogMysqlDriverIT() {
+    super();
+    mysqlDriverDownloadUrl =
+        "https://repo1.maven.org/maven2/mysql/mysql-connector-java/8.0.11/mysql-connector-java-8.0.11.jar";
+  }
+}

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

@@ -104,6 +104,8 @@ public class CatalogMysqlIT extends AbstractIT {
 
   protected String mysqlImageName = defaultMysqlImageName;
 
+  protected String mysqlDriverDownloadUrl = DOWNLOAD_JDBC_DRIVER_URL;
+
   boolean SupportColumnDefaultValueExpression() {
     return true;
   }
@@ -114,7 +116,7 @@ public void startup() throws IOException, SQLException {
     if (!ITUtils.EMBEDDED_TEST_MODE.equals(testMode)) {
       String gravitinoHome = System.getenv("GRAVITINO_HOME");
       Path tmpPath = Paths.get(gravitinoHome, "/catalogs/jdbc-mysql/libs");
-      JdbcDriverDownloader.downloadJdbcDriver(DOWNLOAD_JDBC_DRIVER_URL, tmpPath.toString());
+      JdbcDriverDownloader.downloadJdbcDriver(mysqlDriverDownloadUrl, tmpPath.toString());
     }
 
     TEST_DB_NAME = TestDatabaseName.MYSQL_CATALOG_MYSQL_IT;
@@ -1054,6 +1056,7 @@ public void testSchemaComment() {
     Assertions.assertTrue(StringUtils.isEmpty(schema.comment()));
     schema = catalog.asSchemas().loadSchema(ident);
     Assertions.assertTrue(StringUtils.isEmpty(schema.comment()));
+    catalog.asSchemas().dropSchema(ident, true);
   }
 
   @Test
@@ -1287,6 +1290,7 @@ void testNameSpec() {
     Assertions.assertTrue(catalog.asTableCatalog().dropTable(tableIdent));
     Assertions.assertFalse(catalog.asTableCatalog().tableExists(tableIdent));
     Assertions.assertFalse(catalog.asTableCatalog().purgeTable(tableIdent));
+    catalog.asSchemas().dropSchema(schemaIdent, true);
   }
 
   @Test
@@ -1351,6 +1355,11 @@ void testMySQLSchemaNameCaseSensitive() {
               .stream()
               .anyMatch(n -> n.equals(tableName)));
     }
+
+    for (String schema : schemas) {
+      NameIdentifier schemaIdentifier = NameIdentifier.of(metalakeName, catalogName, schema);
+      schemaSupport.dropSchema(schemaIdentifier, true);
+    }
   }
 
   @Test