[#2085] improvement(jdbc-mysql): Remove depends on JDBC system table.

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

@@ -77,6 +77,18 @@ public List<String> listDatabases() {
     }
   }
 
+  /**
+   * Load the schema with the given name.
+   *
+   * @param connection The connection to the database.
+   * @param schemaName The name of the schema.
+   * @return The result set of the schema. Some database may not support fuzzy query and the result
+   *     set may be empty or is not the schema with the given name.
+   * @throws SQLException
+   */
+  protected abstract ResultSet getSchema(Connection connection, String schemaName)
+      throws SQLException;
+
   /**
    * @param databaseName The name of the database.
    * @param comment The comment of the database.

catalogs/catalog-jdbc-common/src/test/java/com/datastrato/gravitino/catalog/jdbc/operation/SqliteDatabaseOperations.java

@@ -15,6 +15,7 @@
 import java.io.File;
 import java.sql.Connection;
 import java.sql.DriverManager;
+import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.util.Arrays;
 import java.util.List;
@@ -91,4 +92,9 @@ public String generateCreateDatabaseSql(
   public String generateDropDatabaseSql(String databaseName, boolean cascade) {
     return null;
   }
+
+  @Override
+  protected ResultSet getSchema(Connection connection, String schemaName) throws SQLException {
+    throw new UnsupportedOperationException("Not supported for SQLite");
+  }
 }

catalogs/catalog-jdbc-doris/src/main/java/com/datastrato/gravitino/catalog/doris/operation/DorisDatabaseOperations.java

@@ -7,6 +7,9 @@
 import com.datastrato.gravitino.catalog.jdbc.JdbcSchema;
 import com.datastrato.gravitino.catalog.jdbc.operation.JdbcDatabaseOperations;
 import com.datastrato.gravitino.exceptions.NoSuchSchemaException;
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.SQLException;
 import java.util.Map;
 
 /** Database operations for Doris. */
@@ -28,4 +31,9 @@ protected String generateCreateDatabaseSql(
   protected String generateDropDatabaseSql(String databaseName, boolean cascade) {
     throw new UnsupportedOperationException();
   }
+
+  @Override
+  protected ResultSet getSchema(Connection connection, String schemaName) throws SQLException {
+    throw new UnsupportedOperationException();
+  }
 }

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

@@ -9,16 +9,17 @@
 import com.datastrato.gravitino.catalog.jdbc.operation.JdbcDatabaseOperations;
 import com.datastrato.gravitino.exceptions.NoSuchSchemaException;
 import com.datastrato.gravitino.meta.AuditInfo;
+import com.google.common.collect.ImmutableMap;
 import java.sql.Connection;
-import java.sql.PreparedStatement;
+import java.sql.DatabaseMetaData;
 import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.sql.Statement;
 import java.util.Collections;
-import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Locale;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Set;
 import org.apache.commons.collections4.MapUtils;
 import org.apache.commons.lang3.StringUtils;
@@ -90,36 +91,36 @@ public String generateDropDatabaseSql(String databaseName, boolean cascade) {
     return dropDatabaseSql;
   }
 
+  @Override
+  protected ResultSet getSchema(Connection connection, String databaseName) throws SQLException {
+    final DatabaseMetaData metaData = connection.getMetaData();
+    // It'd indeed need to call getCatalogs() to get the schema not `getSchemas()` for MySQL.
+    return metaData.getCatalogs();
+  }
+
   @Override
   public JdbcSchema load(String databaseName) throws NoSuchSchemaException {
     try (final Connection connection = this.dataSource.getConnection()) {
-      String query = "SELECT * FROM information_schema.SCHEMATA WHERE SCHEMA_NAME = ?";
-      try (PreparedStatement preparedStatement = connection.prepareStatement(query)) {
-        preparedStatement.setString(1, databaseName);
-
-        // Execute the query
-        try (ResultSet resultSet = preparedStatement.executeQuery()) {
-          if (!resultSet.next()) {
-            throw new NoSuchSchemaException(
-                "Database %s could not be found in information_schema.SCHEMATA", databaseName);
-          }
-          String schemaName = resultSet.getString("SCHEMA_NAME");
-          // Mysql currently only supports these two attributes
-          String characterSetName = resultSet.getString("DEFAULT_CHARACTER_SET_NAME");
-          String collationName = resultSet.getString("DEFAULT_COLLATION_NAME");
-          Map<String, String> properties = new HashMap<>();
-          properties.put("CHARACTER SET", characterSetName);
-          properties.put("COLLATE", collationName);
+      ResultSet resultSet = getSchema(connection, databaseName);
 
-          JdbcSchema.Builder builder =
-              JdbcSchema.builder()
-                  .withName(schemaName)
-                  .withProperties(properties)
-                  .withAuditInfo(AuditInfo.EMPTY);
-
-          return builder.build();
+      boolean found = false;
+      while (resultSet.next()) {
+        if (Objects.equals(resultSet.getString(1), databaseName)) {
+          found = true;
+          break;
         }
       }
+
+      if (!found) {
+        throw new NoSuchSchemaException("Database %s could not be found", databaseName);
+      }
+
+      return JdbcSchema.builder()
+          .withName(databaseName)
+          .withProperties(ImmutableMap.of())
+          .withAuditInfo(AuditInfo.EMPTY)
+          .build();
+
     } catch (final SQLException se) {
       throw this.exceptionMapper.toGravitinoException(se);
     }

catalogs/catalog-jdbc-postgresql/src/main/java/com/datastrato/gravitino/catalog/postgresql/operation/PostgreSqlSchemaOperations.java

@@ -11,6 +11,7 @@
 import com.datastrato.gravitino.exceptions.NoSuchSchemaException;
 import com.datastrato.gravitino.meta.AuditInfo;
 import java.sql.Connection;
+import java.sql.DatabaseMetaData;
 import java.sql.PreparedStatement;
 import java.sql.ResultSet;
 import java.sql.SQLException;
@@ -20,6 +21,7 @@
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Set;
 import javax.sql.DataSource;
 import org.apache.commons.collections4.MapUtils;
@@ -50,25 +52,27 @@ public void initialize(
   @Override
   public JdbcSchema load(String schema) throws NoSuchSchemaException {
     try (Connection connection = getConnection()) {
-      String sql =
-          "SELECT schema_name FROM information_schema.schemata WHERE schema_name = ? AND catalog_name = ?";
-      try (PreparedStatement statement = connection.prepareStatement(sql)) {
-        statement.setString(1, schema);
-        statement.setString(2, database);
-        try (ResultSet resultSet = statement.executeQuery()) {
-          if (!resultSet.next()) {
-            throw new NoSuchSchemaException("No such schema: %s", schema);
-          }
-          String schemaName = resultSet.getString(1);
-          String comment = getSchemaComment(schema, connection);
-          return JdbcSchema.builder()
-              .withName(schemaName)
-              .withComment(comment)
-              .withAuditInfo(AuditInfo.EMPTY)
-              .withProperties(Collections.emptyMap())
-              .build();
+      ResultSet resultSet = getSchema(connection, schema);
+
+      boolean found = false;
+      while (resultSet.next()) {
+        if (Objects.equals(resultSet.getString(1), schema)) {
+          found = true;
+          break;
         }
       }
+
+      if (!found) {
+        throw new NoSuchSchemaException("No such schema: %s", schema);
+      }
+
+      String comment = getSchemaComment(schema, connection);
+      return JdbcSchema.builder()
+          .withName(schema)
+          .withComment(comment)
+          .withAuditInfo(AuditInfo.EMPTY)
+          .withProperties(Collections.emptyMap())
+          .build();
     } catch (SQLException e) {
       throw exceptionMapper.toGravitinoException(e);
     }
@@ -78,16 +82,11 @@ public JdbcSchema load(String schema) throws NoSuchSchemaException {
   public List<String> listDatabases() {
     List<String> result = new ArrayList<>();
     try (Connection connection = getConnection()) {
-      try (PreparedStatement statement =
-          connection.prepareStatement(
-              "SELECT schema_name FROM information_schema.schemata WHERE catalog_name = ?")) {
-        statement.setString(1, database);
-        ResultSet resultSet = statement.executeQuery();
-        while (resultSet.next()) {
-          String databaseName = resultSet.getString(1);
-          if (!isSystemDatabase(databaseName)) {
-            result.add(databaseName);
-          }
+      ResultSet resultSet = getSchema(connection, null);
+      while (resultSet.next()) {
+        String schemaName = resultSet.getString(1);
+        if (!isSystemDatabase(schemaName)) {
+          result.add(resultSet.getString(1));
         }
       }
     } catch (final SQLException se) {
@@ -116,6 +115,12 @@ public String generateCreateDatabaseSql(
     return sqlBuilder.toString();
   }
 
+  @Override
+  protected ResultSet getSchema(Connection connection, String schemaName) throws SQLException {
+    final DatabaseMetaData metaData = connection.getMetaData();
+    return metaData.getSchemas(database, schemaName);
+  }
+
   @Override
   public String generateDropDatabaseSql(String schema, boolean cascade) {
     StringBuilder sqlBuilder = new StringBuilder(String.format("DROP SCHEMA %s", schema));