[#5214] Improvement(catalog-lakehouse-paimon): add jdbc-driver to Paimon JDBC catalog backend configuration

catalogs/catalog-lakehouse-paimon/src/main/java/org/apache/gravitino/catalog/lakehouse/paimon/PaimonCatalogPropertiesMetadata.java

@@ -53,6 +53,7 @@ public class PaimonCatalogPropertiesMetadata extends BaseCatalogPropertiesMetada
   public static final String PAIMON_JDBC_USER = "jdbc.user";
   public static final String GRAVITINO_JDBC_PASSWORD = "jdbc-password";
   public static final String PAIMON_JDBC_PASSWORD = "jdbc.password";
+  public static final String GRAVITINO_JDBC_DRIVER = "jdbc-driver";
 
   // S3 properties needed by Paimon
   public static final String S3_ENDPOINT = "s3.endpoint";
@@ -70,7 +71,9 @@ public class PaimonCatalogPropertiesMetadata extends BaseCatalogPropertiesMetada
           GRAVITINO_JDBC_USER,
           PAIMON_JDBC_USER,
           GRAVITINO_JDBC_PASSWORD,
-          PAIMON_JDBC_PASSWORD);
+          PAIMON_JDBC_PASSWORD,
+          GRAVITINO_JDBC_DRIVER,
+          GRAVITINO_JDBC_DRIVER);
   private static final Map<String, PropertyEntry<?>> PROPERTIES_METADATA;
   public static final Map<String, String> KERBEROS_CONFIGURATION =
       ImmutableMap.of(
@@ -131,7 +134,13 @@ public class PaimonCatalogPropertiesMetadata extends BaseCatalogPropertiesMetada
                 "Gravitino Paimon catalog jdbc password",
                 false /* immutable */,
                 null /* defaultValue */,
-                true /* hidden */));
+                true /* hidden */),
+            stringOptionalPropertyEntry(
+                GRAVITINO_JDBC_DRIVER,
+                "The driver of the Jdbc connection",
+                false /* immutable */,
+                null /* defaultValue */,
+                false /* hidden */));
     HashMap<String, PropertyEntry<?>> result = Maps.newHashMap();
     result.putAll(Maps.uniqueIndex(propertyEntries, PropertyEntry::getName));
     result.putAll(KerberosConfig.KERBEROS_PROPERTY_ENTRIES);

catalogs/catalog-lakehouse-paimon/src/main/java/org/apache/gravitino/catalog/lakehouse/paimon/PaimonConfig.java

@@ -66,6 +66,14 @@ public class PaimonConfig extends Config {
           .checkValue(StringUtils::isNotBlank, ConfigConstants.NOT_BLANK_ERROR_MSG)
           .create();
 
+  public static final ConfigEntry<String> CATALOG_JDBC_DRIVER =
+      new ConfigBuilder(PaimonCatalogPropertiesMetadata.GRAVITINO_JDBC_DRIVER)
+          .doc("The driver of the Jdbc connection")
+          .version(ConfigConstants.VERSION_0_7_0)
+          .stringConf()
+          .checkValue(StringUtils::isNotBlank, ConfigConstants.NOT_BLANK_ERROR_MSG)
+          .create();
+
   public PaimonConfig() {
     super(false);
   }
@@ -74,4 +82,8 @@ public PaimonConfig(Map<String, String> properties) {
     super(false);
     loadFromMap(properties, k -> true);
   }
+
+  public String getJdbcDriver() {
+    return get(CATALOG_JDBC_DRIVER);
+  }
 }

catalogs/catalog-lakehouse-paimon/src/main/java/org/apache/gravitino/catalog/lakehouse/paimon/utils/CatalogUtils.java

@@ -124,6 +124,16 @@ private static void checkPaimonConfig(PaimonConfig paimonConfig) {
       Preconditions.checkArgument(
           StringUtils.isNotBlank(uri), "Paimon Catalog uri can not be null or empty.");
     }
+
+    if (PaimonCatalogBackend.JDBC.name().equalsIgnoreCase(metastore)) {
+      String driverClassName = paimonConfig.getJdbcDriver();
+      try {
+        // Load the jdbc driver
+        Class.forName(driverClassName);
+      } catch (ClassNotFoundException e) {
+        throw new IllegalArgumentException("Couldn't load jdbc driver " + driverClassName);
+      }
+    }
   }
 
   public static Map<String, String> toInnerProperty(

catalogs/catalog-lakehouse-paimon/src/test/java/org/apache/gravitino/catalog/lakehouse/paimon/integration/test/CatalogPaimonBaseIT.java

@@ -89,7 +89,7 @@ public abstract class CatalogPaimonBaseIT extends BaseIT {
 
   protected static final ContainerSuite containerSuite = ContainerSuite.getInstance();
   protected static final TestDatabaseName TEST_DB_NAME =
-      TestDatabaseName.PG_TEST_ICEBERG_CATALOG_MULTIPLE_JDBC_LOAD;
+      TestDatabaseName.PG_TEST_PAIMON_CATALOG_MULTIPLE_JDBC_LOAD;
   protected static MySQLContainer mySQLContainer;
   protected String WAREHOUSE;
   protected String TYPE;

catalogs/catalog-lakehouse-paimon/src/test/java/org/apache/gravitino/catalog/lakehouse/paimon/integration/test/CatalogPaimonJdbcIT.java

@@ -53,6 +53,8 @@ protected Map<String, String> initPaimonCatalogProperties() {
     catalogProperties.put(PaimonCatalogPropertiesMetadata.URI, URI);
     catalogProperties.put(PaimonCatalogPropertiesMetadata.GRAVITINO_JDBC_USER, jdbcUser);
     catalogProperties.put(PaimonCatalogPropertiesMetadata.GRAVITINO_JDBC_PASSWORD, jdbcPassword);
+    catalogProperties.put(
+        PaimonCatalogPropertiesMetadata.GRAVITINO_JDBC_DRIVER, "com.mysql.cj.jdbc.Driver");
 
     return catalogProperties;
   }

catalogs/catalog-lakehouse-paimon/src/test/java/org/apache/gravitino/catalog/lakehouse/paimon/integration/test/CatalogPaimonMultipleJDBCLoadIT.java

@@ -0,0 +1,147 @@
+/*
+ * 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.catalog.lakehouse.paimon.integration.test;
+
+import com.google.common.collect.Maps;
+import java.io.IOException;
+import java.sql.SQLException;
+import java.util.Collections;
+import java.util.Map;
+import org.apache.gravitino.Catalog;
+import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.catalog.lakehouse.paimon.PaimonCatalogPropertiesMetadata;
+import org.apache.gravitino.client.GravitinoMetalake;
+import org.apache.gravitino.integration.test.container.MySQLContainer;
+import org.apache.gravitino.integration.test.container.PostgreSQLContainer;
+import org.apache.gravitino.integration.test.util.BaseIT;
+import org.apache.gravitino.integration.test.util.TestDatabaseName;
+import org.apache.gravitino.rel.Column;
+import org.apache.gravitino.rel.types.Types;
+import org.apache.gravitino.utils.RandomNameUtils;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInstance;
+
+@Tag("gravitino-docker-test")
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
+public class CatalogPaimonMultipleJDBCLoadIT extends BaseIT {
+  private static final TestDatabaseName TEST_DB_NAME =
+      TestDatabaseName.PG_TEST_PAIMON_CATALOG_MULTIPLE_JDBC_LOAD;
+
+  private static MySQLContainer mySQLContainer;
+  private static PostgreSQLContainer postgreSQLContainer;
+
+  @BeforeAll
+  public void startup() throws IOException {
+    containerSuite.startMySQLContainer(TEST_DB_NAME);
+    mySQLContainer = containerSuite.getMySQLContainer();
+    containerSuite.startPostgreSQLContainer(TEST_DB_NAME);
+    postgreSQLContainer = containerSuite.getPostgreSQLContainer();
+  }
+
+  @Test
+  public void testCreateMultipleJdbcInPaimon() throws SQLException {
+    String metalakeName = RandomNameUtils.genRandomName("it_metalake");
+    String postgreSqlCatalogName = RandomNameUtils.genRandomName("it_paimon_postgresql");
+    GravitinoMetalake metalake =
+        client.createMetalake(metalakeName, "comment", Collections.emptyMap());
+
+    Map<String, String> paimonPgConf = Maps.newHashMap();
+    String jdbcUrl = postgreSQLContainer.getJdbcUrl(TEST_DB_NAME);
+    paimonPgConf.put(PaimonCatalogPropertiesMetadata.GRAVITINO_CATALOG_BACKEND, "jdbc");
+    paimonPgConf.put(PaimonCatalogPropertiesMetadata.URI, jdbcUrl);
+    paimonPgConf.put(PaimonCatalogPropertiesMetadata.WAREHOUSE, "file:///tmp/paimon-pg-warehouse");
+    paimonPgConf.put(
+        PaimonCatalogPropertiesMetadata.GRAVITINO_JDBC_USER, postgreSQLContainer.getUsername());
+    paimonPgConf.put(
+        PaimonCatalogPropertiesMetadata.GRAVITINO_JDBC_PASSWORD, postgreSQLContainer.getPassword());
+    paimonPgConf.put(
+        PaimonCatalogPropertiesMetadata.GRAVITINO_JDBC_DRIVER,
+        postgreSQLContainer.getDriverClassName(TEST_DB_NAME));
+
+    Catalog postgreSqlCatalog =
+        metalake.createCatalog(
+            postgreSqlCatalogName,
+            Catalog.Type.RELATIONAL,
+            "lakehouse-paimon",
+            "comment",
+            paimonPgConf);
+
+    Map<String, String> paimonMysqlConf = Maps.newHashMap();
+
+    paimonMysqlConf.put(PaimonCatalogPropertiesMetadata.GRAVITINO_CATALOG_BACKEND, "jdbc");
+    paimonMysqlConf.put(
+        PaimonCatalogPropertiesMetadata.URI, mySQLContainer.getJdbcUrl(TEST_DB_NAME));
+    paimonMysqlConf.put(
+        PaimonCatalogPropertiesMetadata.WAREHOUSE, "file:///tmp/paimon-mysql-warehouse");
+    paimonMysqlConf.put(
+        PaimonCatalogPropertiesMetadata.GRAVITINO_JDBC_USER, mySQLContainer.getUsername());
+    paimonMysqlConf.put(
+        PaimonCatalogPropertiesMetadata.GRAVITINO_JDBC_PASSWORD, mySQLContainer.getPassword());
+    paimonMysqlConf.put(
+        PaimonCatalogPropertiesMetadata.GRAVITINO_JDBC_DRIVER,
+        mySQLContainer.getDriverClassName(TEST_DB_NAME));
+    String mysqlCatalogName = RandomNameUtils.genRandomName("it_paimon_mysql");
+    Catalog mysqlCatalog =
+        metalake.createCatalog(
+            mysqlCatalogName,
+            Catalog.Type.RELATIONAL,
+            "lakehouse-paimon",
+            "comment",
+            paimonMysqlConf);
+
+    String[] nameIdentifiers = mysqlCatalog.asSchemas().listSchemas();
+    Assertions.assertEquals(0, nameIdentifiers.length);
+    nameIdentifiers = postgreSqlCatalog.asSchemas().listSchemas();
+    Assertions.assertEquals(0, nameIdentifiers.length);
+
+    String schemaName = RandomNameUtils.genRandomName("it_schema");
+    mysqlCatalog.asSchemas().createSchema(schemaName, null, Collections.emptyMap());
+
+    postgreSqlCatalog.asSchemas().createSchema(schemaName, null, Collections.emptyMap());
+
+    String tableName = RandomNameUtils.genRandomName("it_table");
+
+    Column col1 = Column.of("col_1", Types.IntegerType.get(), "col_1_comment");
+    String comment = "test";
+    mysqlCatalog
+        .asTableCatalog()
+        .createTable(
+            NameIdentifier.of(schemaName, tableName),
+            new Column[] {col1},
+            comment,
+            Collections.emptyMap());
+
+    postgreSqlCatalog
+        .asTableCatalog()
+        .createTable(
+            NameIdentifier.of(schemaName, tableName),
+            new Column[] {col1},
+            comment,
+            Collections.emptyMap());
+
+    Assertions.assertTrue(
+        mysqlCatalog.asTableCatalog().tableExists(NameIdentifier.of(schemaName, tableName)));
+    Assertions.assertTrue(
+        postgreSqlCatalog.asTableCatalog().tableExists(NameIdentifier.of(schemaName, tableName)));
+  }
+}

catalogs/catalog-lakehouse-paimon/src/test/java/org/apache/gravitino/catalog/lakehouse/paimon/utils/TestCatalogUtils.java

@@ -77,7 +77,9 @@ private void assertCatalog(String metastore, Consumer<Catalog> consumer) throws
                         PaimonConfig.CATALOG_JDBC_USER.getKey(),
                         "user",
                         PaimonConfig.CATALOG_JDBC_PASSWORD.getKey(),
-                        "password")))
+                        "password",
+                        PaimonConfig.CATALOG_JDBC_DRIVER.getKey(),
+                        "org.h2.Driver")))
             .getCatalog()) {
       consumer.accept(catalog);
     }

docs/lakehouse-paimon-catalog.md

@@ -58,12 +58,13 @@ Any properties not defined by Gravitino with `gravitino.bypass.` prefix will pas
 
 #### JDBC backend
 
-If you are using JDBC backend, you must specify the properties like `jdbc-user` and `jdbc-password`.
+If you are using JDBC backend, you must specify the properties like `jdbc-user`, `jdbc-password` and `jdbc-driver`.
 
-| Property name   | Description                                                                                                                                                                                                 | Default value          | Required                                                        | Since Version      |
-|-----------------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|------------------------|-----------------------------------------------------------------|--------------------|
-| `jdbc-user`     | Jdbc user of Gravitino Paimon catalog for `jdbc` backend.                                                                                                                                                   | (none)                 | required if the value of `catalog-backend` is `jdbc`.           | 0.7.0-incubating   |
-| `jdbc-password` | Jdbc password of Gravitino Paimon catalog for `jdbc` backend.                                                                                                                                               | (none)                 | required if the value of `catalog-backend` is `jdbc`.           | 0.7.0-incubating   |
+| Property name   | Description                                                                                               | Default value   | Required                                              | Since Version    |
+|-----------------|-----------------------------------------------------------------------------------------------------------|-----------------|-------------------------------------------------------|------------------|
+| `jdbc-user`     | Jdbc user of Gravitino Paimon catalog for `jdbc` backend.                                                 | (none)          | required if the value of `catalog-backend` is `jdbc`. | 0.7.0-incubating |
+| `jdbc-password` | Jdbc password of Gravitino Paimon catalog for `jdbc` backend.                                             | (none)          | required if the value of `catalog-backend` is `jdbc`. | 0.7.0-incubating |
+| `jdbc-driver`   | `com.mysql.jdbc.Driver` or `com.mysql.cj.jdbc.Driver` for MySQL, `org.postgresql.Driver` for PostgreSQL   | (none)          | required if the value of `catalog-backend` is `jdbc`. | 0.7.0-incubating |
 
 :::caution
 You must download the corresponding JDBC driver and place it to the `catalogs/lakehouse-paimon/libs` directory If you are using JDBC backend.

integration-test-common/src/test/java/org/apache/gravitino/integration/test/util/TestDatabaseName.java

@@ -43,6 +43,8 @@
  *       postgresql.integration.test.TestMultipleJDBCLoad.
  *   <li>{@link #PG_TEST_ICEBERG_CATALOG_MULTIPLE_JDBC_LOAD}: Represents the PostgreSQL database for
  *       lakehouse.iceberg.integration.test.TestMultipleJDBCLoad.
+ *   <li>{@link #PG_TEST_PAIMON_CATALOG_MULTIPLE_JDBC_LOAD}: Represents the PostgreSQL database for
+ *       lakehouse.paimon.integration.test.CatalogPaimonMultipleJDBCLoadIT.
  * </ul>
  */
 public enum TestDatabaseName {
@@ -88,4 +90,16 @@ public String toString() {
       return this.name().toLowerCase();
     }
   },
+
+  /**
+   * Represents the PostgreSQL database for
+   * lakehouse.paimon.integration.test.CatalogPaimonMultipleJDBCLoadIT.
+   */
+  PG_TEST_PAIMON_CATALOG_MULTIPLE_JDBC_LOAD {
+    /** PostgreSQL only accept lowercase database name */
+    @Override
+    public String toString() {
+      return this.name().toLowerCase();
+    }
+  },
 }