From 2e415a20b2b6501e088c1113b47275a69f20cf67 Mon Sep 17 00:00:00 2001 From: youze Liang <525672876@qq.com> Date: Mon, 13 Jan 2025 16:13:48 +0800 Subject: [PATCH 1/8] feat(spark-connector):support JDBC catalog --- .../connector/jdbc/GravitinoJdbcCatalog.java | 66 ++++++++++++++++++ .../jdbc/JdbcPropertiesConstants.java | 33 +++++++++ .../jdbc/JdbcPropertiesConverter.java | 67 +++++++++++++++++++ .../connector/version/CatalogNameAdaptor.java | 3 + .../jdbc/TestJdbcPropertiesConverter.java | 62 +++++++++++++++++ 5 files changed, 231 insertions(+) create mode 100644 spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/jdbc/GravitinoJdbcCatalog.java create mode 100644 spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/jdbc/JdbcPropertiesConstants.java create mode 100644 spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/jdbc/JdbcPropertiesConverter.java create mode 100644 spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/jdbc/TestJdbcPropertiesConverter.java diff --git a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/jdbc/GravitinoJdbcCatalog.java b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/jdbc/GravitinoJdbcCatalog.java new file mode 100644 index 00000000000..c079d4376a3 --- /dev/null +++ b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/jdbc/GravitinoJdbcCatalog.java @@ -0,0 +1,66 @@ +/* + * 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.spark.connector.jdbc; + +import java.util.Map; +import org.apache.gravitino.spark.connector.PropertiesConverter; +import org.apache.gravitino.spark.connector.SparkTransformConverter; +import org.apache.gravitino.spark.connector.SparkTypeConverter; +import org.apache.gravitino.spark.connector.catalog.BaseCatalog; +import org.apache.spark.sql.connector.catalog.Identifier; +import org.apache.spark.sql.connector.catalog.Table; +import org.apache.spark.sql.connector.catalog.TableCatalog; +import org.apache.spark.sql.execution.datasources.v2.jdbc.JDBCTableCatalog; +import org.apache.spark.sql.util.CaseInsensitiveStringMap; + +public class GravitinoJdbcCatalog extends BaseCatalog { + + @Override + protected TableCatalog createAndInitSparkCatalog( + String name, CaseInsensitiveStringMap options, Map properties) { + JDBCTableCatalog jdbcTableCatalog = new JDBCTableCatalog(); + Map all = + getPropertiesConverter().toSparkCatalogProperties(options, properties); + jdbcTableCatalog.initialize(name, new CaseInsensitiveStringMap(all)); + return jdbcTableCatalog; + } + + @Override + protected Table createSparkTable( + Identifier identifier, + org.apache.gravitino.rel.Table gravitinoTable, + Table sparkTable, + TableCatalog sparkCatalog, + PropertiesConverter propertiesConverter, + SparkTransformConverter sparkTransformConverter, + SparkTypeConverter sparkTypeConverter) { + return sparkTable; + } + + @Override + protected PropertiesConverter getPropertiesConverter() { + return JdbcPropertiesConverter.getInstance(); + } + + @Override + protected SparkTransformConverter getSparkTransformConverter() { + return new SparkTransformConverter(false); + } +} diff --git a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/jdbc/JdbcPropertiesConstants.java b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/jdbc/JdbcPropertiesConstants.java new file mode 100644 index 00000000000..f1cf50f81d3 --- /dev/null +++ b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/jdbc/JdbcPropertiesConstants.java @@ -0,0 +1,33 @@ +/* + * 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.spark.connector.jdbc; + +public class JdbcPropertiesConstants { + + public static final String GRAVITINO_JDBC_USER = "jdbc-user"; + public static final String GRAVITINO_JDBC_PASSWORD = "jdbc-password"; + public static final String GRAVITINO_JDBC_DRIVER = "jdbc-driver"; + public static final String GRAVITINO_JDBC_URL = "jdbc-url"; + + public static final String SPARK_JDBC_URL = "url"; + public static final String SPARK_JDBC_USER = "user"; + public static final String SPARK_JDBC_PASSWORD = "password"; + public static final String SPARK_JDBC_DRIVER = "driver"; +} diff --git a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/jdbc/JdbcPropertiesConverter.java b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/jdbc/JdbcPropertiesConverter.java new file mode 100644 index 00000000000..308e548602c --- /dev/null +++ b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/jdbc/JdbcPropertiesConverter.java @@ -0,0 +1,67 @@ +/* + * 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.spark.connector.jdbc; + +import com.google.common.base.Preconditions; +import java.util.HashMap; +import java.util.Map; +import org.apache.gravitino.spark.connector.PropertiesConverter; + +public class JdbcPropertiesConverter implements PropertiesConverter { + + public static class JdbcPropertiesConverterHolder { + private static final JdbcPropertiesConverter INSTANCE = new JdbcPropertiesConverter(); + } + + private JdbcPropertiesConverter() {} + + public static JdbcPropertiesConverter getInstance() { + return JdbcPropertiesConverterHolder.INSTANCE; + } + + @Override + public Map toSparkCatalogProperties(Map properties) { + Preconditions.checkArgument(properties != null, "Jdbc Catalog properties should not be null"); + HashMap jdbcProperties = new HashMap<>(); + jdbcProperties.put( + JdbcPropertiesConstants.SPARK_JDBC_URL, + properties.get(JdbcPropertiesConstants.GRAVITINO_JDBC_URL)); + jdbcProperties.put( + JdbcPropertiesConstants.SPARK_JDBC_USER, + properties.get(JdbcPropertiesConstants.GRAVITINO_JDBC_USER)); + jdbcProperties.put( + JdbcPropertiesConstants.SPARK_JDBC_PASSWORD, + properties.get(JdbcPropertiesConstants.GRAVITINO_JDBC_PASSWORD)); + jdbcProperties.put( + JdbcPropertiesConstants.SPARK_JDBC_DRIVER, + properties.get(JdbcPropertiesConstants.GRAVITINO_JDBC_DRIVER)); + return jdbcProperties; + } + + @Override + public Map toGravitinoTableProperties(Map properties) { + return new HashMap<>(properties); + } + + @Override + public Map toSparkTableProperties(Map properties) { + return new HashMap<>(properties); + } +} diff --git a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/version/CatalogNameAdaptor.java b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/version/CatalogNameAdaptor.java index 9392feac2f1..41e769ca2b4 100644 --- a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/version/CatalogNameAdaptor.java +++ b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/version/CatalogNameAdaptor.java @@ -51,6 +51,9 @@ private static String sparkVersion() { } private static String getCatalogName(String provider, int majorVersion, int minorVersion) { + if (provider.startsWith("jdbc")) { + return "org.apache.gravitino.spark.connector.jdbc.GravitinoJdbcCatalog"; + } String key = String.format("%s-%d.%d", provider.toLowerCase(Locale.ROOT), majorVersion, minorVersion); return catalogNames.get(key); diff --git a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/jdbc/TestJdbcPropertiesConverter.java b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/jdbc/TestJdbcPropertiesConverter.java new file mode 100644 index 00000000000..5d3e4d065fd --- /dev/null +++ b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/jdbc/TestJdbcPropertiesConverter.java @@ -0,0 +1,62 @@ +/* + * 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.spark.connector.jdbc; + +import com.google.common.collect.ImmutableMap; +import java.util.Map; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +public class TestJdbcPropertiesConverter { + private final JdbcPropertiesConverter jdbcPropertiesConverter = + JdbcPropertiesConverter.getInstance(); + + @Test + void testCatalogProperties() { + String url = "jdbc-url"; + String user = "user1"; + String passwd = "passwd1"; + String driver = "jdbc-driver"; + Map properties = + jdbcPropertiesConverter.toSparkCatalogProperties( + ImmutableMap.of( + JdbcPropertiesConstants.GRAVITINO_JDBC_URL, + url, + JdbcPropertiesConstants.GRAVITINO_JDBC_USER, + user, + JdbcPropertiesConstants.GRAVITINO_JDBC_PASSWORD, + passwd, + JdbcPropertiesConstants.GRAVITINO_JDBC_DRIVER, + driver, + "key1", + "value1")); + Assertions.assertEquals( + ImmutableMap.of( + JdbcPropertiesConstants.SPARK_JDBC_URL, + url, + JdbcPropertiesConstants.SPARK_JDBC_USER, + user, + JdbcPropertiesConstants.SPARK_JDBC_PASSWORD, + passwd, + JdbcPropertiesConstants.SPARK_JDBC_DRIVER, + driver), + properties); + } +} From dc48d0a6e23654a04ece9a948aaba7e0f716707d Mon Sep 17 00:00:00 2001 From: youze Liang <525672876@qq.com> Date: Mon, 13 Jan 2025 18:05:55 +0800 Subject: [PATCH 2/8] add docs --- docs/spark-connector/spark-catalog-jdbc.md | 69 ++++++++++++++++++++++ 1 file changed, 69 insertions(+) create mode 100644 docs/spark-connector/spark-catalog-jdbc.md diff --git a/docs/spark-connector/spark-catalog-jdbc.md b/docs/spark-connector/spark-catalog-jdbc.md new file mode 100644 index 00000000000..6989ff2808f --- /dev/null +++ b/docs/spark-connector/spark-catalog-jdbc.md @@ -0,0 +1,69 @@ +--- +title: "Spark connector JDBC catalog" +slug: /spark-connector/spark-catalog-jdbc +keyword: spark connector jdbc catalog +license: "This software is licensed under the Apache License version 2." +--- + +The Apache Gravitino Spark connector offers the capability to read JDBC tables, with the metadata managed by the Gravitino server. To enable the use of the JDBC catalog within the Spark connector, you must download the jdbc driver jar which you used to Spark classpath. + +## Capabilities + +#### Support DML and DDL operations: + +- `CREATE TABLE` +- `DROP TABLE` +- `ALTER TABLE` +- `SELECT` +- `INSERT` + + :::info + JDBCTable does not support distributed transaction. When writing data to RDBMS, each task is an independent transaction. If some tasks of spark succeed and some tasks fail, dirty data is generated. + ::: + +#### Not supported operations: +- `CREATE DATABASE` +- `UPDATE` +- `DELETE` +- `TRUNCATE` + +## SQL example + +```sql +-- Suppose mysql_a is the mysql catalog name managed by Gravitino +USE mysql_a; +-- Suppose mydatabase is in your mysql +USE mydatabase; + +CREATE TABLE IF NOT EXISTS employee ( + id bigint, + name string, + department string, + hire_date timestamp +) +DESC TABLE EXTENDED employee; + +INSERT INTO employee +VALUES +(1, 'Alice', 'Engineering', TIMESTAMP '2021-01-01 09:00:00'), +(2, 'Bob', 'Marketing', TIMESTAMP '2021-02-01 10:30:00'), +(3, 'Charlie', 'Sales', TIMESTAMP '2021-03-01 08:45:00'); + +SELECT * FROM employee WHERE date(hire_date) = '2021-01-01'; + + +``` + +## Catalog properties + +Gravitino spark connector will transform below property names which are defined in catalog properties to Spark JDBC connector configuration. + +| Gravitino catalog property name | Spark JDBC connector configuration | Description | Since Version | +|---------------------------------|------------------------------------|---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|---------------| +| `jdbc-url` | `url` | JDBC URL for connecting to the database. For example, jdbc:mysql://localhost:3306 | 0.3.0 | +| `jdbc-user` | `jdbc.user` | JDBC user name | 0.3.0 | +| `jdbc-password` | `jdbc.password` | JDBC password | 0.3.0 | +| `jdbc-driver` | `driver` | The driver of the JDBC connection. For example, com.mysql.jdbc.Driver or com.mysql.cj.jdbc.Driver | 0.3.0 | + +Gravitino catalog property names with the prefix `spark.bypass.` are passed to Spark JDBC connector. + From 72e0eadb3a9650dd8cf40edf73552cffd737e928 Mon Sep 17 00:00:00 2001 From: youze Liang <525672876@qq.com> Date: Fri, 17 Jan 2025 09:07:31 +0800 Subject: [PATCH 3/8] add integeration test add integeration test --- docs/spark-connector/spark-catalog-jdbc.md | 4 +- spark-connector/spark-common/build.gradle.kts | 6 ++ .../connector/jdbc/GravitinoJdbcCatalog.java | 25 +++++- .../spark/connector/jdbc/SparkJdbcTable.java | 71 +++++++++++++++ .../jdbc/SparkJdbcTypeConverter.java | 38 ++++++++ .../connector/version/CatalogNameAdaptor.java | 11 ++- .../integration/test/SparkCommonIT.java | 37 +++++--- .../integration/test/SparkEnvIT.java | 15 ++++ .../test/hive/SparkHiveCatalogIT.java | 10 +++ .../test/iceberg/SparkIcebergCatalogIT.java | 12 +++ .../test/jdbc/SparkJdbcCatalogIT.java | 89 +++++++++++++++++++ .../test/jdbc/SparkJdbcTableInfoChecker.java | 54 +++++++++++ .../test/paimon/SparkPaimonCatalogIT.java | 10 +++ .../integration/test/util/SparkTableInfo.java | 3 + .../test/util/SparkTableInfoChecker.java | 6 +- .../jdbc/GravitinoJdbcCatalogSpark33.java | 22 +++++ .../test/jdbc/SparkJdbcCatalogIT33.java | 36 ++++++++ .../jdbc/GravitinoJdbcCatalogSpark34.java | 38 ++++++++ .../jdbc/SparkJdbcTypeConverter34.java | 37 ++++++++ .../test/jdbc/SparkJdbcCatalogIT34.java | 35 ++++++++ .../jdbc/GravitinoJdbcCatalogSpark35.java | 22 +++++ .../test/jdbc/SparkJdbcCatalogIT35.java | 36 ++++++++ 22 files changed, 598 insertions(+), 19 deletions(-) create mode 100644 spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/jdbc/SparkJdbcTable.java create mode 100644 spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/jdbc/SparkJdbcTypeConverter.java create mode 100644 spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/jdbc/SparkJdbcCatalogIT.java create mode 100644 spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/jdbc/SparkJdbcTableInfoChecker.java create mode 100644 spark-connector/v3.3/spark/src/main/java/org/apache/gravitino/spark/connector/jdbc/GravitinoJdbcCatalogSpark33.java create mode 100644 spark-connector/v3.3/spark/src/test/java/org/apache/gravitino/spark/connector/integration/test/jdbc/SparkJdbcCatalogIT33.java create mode 100644 spark-connector/v3.4/spark/src/main/java/org/apache/gravitino/spark/connector/jdbc/GravitinoJdbcCatalogSpark34.java create mode 100644 spark-connector/v3.4/spark/src/main/java/org/apache/gravitino/spark/connector/jdbc/SparkJdbcTypeConverter34.java create mode 100644 spark-connector/v3.4/spark/src/test/java/org/apache/gravitino/spark/connector/integration/test/jdbc/SparkJdbcCatalogIT34.java create mode 100644 spark-connector/v3.5/spark/src/main/java/org/apache/gravitino/spark/connector/jdbc/GravitinoJdbcCatalogSpark35.java create mode 100644 spark-connector/v3.5/spark/src/test/java/org/apache/gravitino/spark/connector/integration/test/jdbc/SparkJdbcCatalogIT35.java diff --git a/docs/spark-connector/spark-catalog-jdbc.md b/docs/spark-connector/spark-catalog-jdbc.md index 6989ff2808f..3b1a3a7616c 100644 --- a/docs/spark-connector/spark-catalog-jdbc.md +++ b/docs/spark-connector/spark-catalog-jdbc.md @@ -22,7 +22,6 @@ The Apache Gravitino Spark connector offers the capability to read JDBC tables, ::: #### Not supported operations: -- `CREATE DATABASE` - `UPDATE` - `DELETE` - `TRUNCATE` @@ -32,7 +31,8 @@ The Apache Gravitino Spark connector offers the capability to read JDBC tables, ```sql -- Suppose mysql_a is the mysql catalog name managed by Gravitino USE mysql_a; --- Suppose mydatabase is in your mysql + +CREATE DATABASE IF NOT EXISTS mydatabase; USE mydatabase; CREATE TABLE IF NOT EXISTS employee ( diff --git a/spark-connector/spark-common/build.gradle.kts b/spark-connector/spark-common/build.gradle.kts index 06e0077d21e..7de83c05791 100644 --- a/spark-connector/spark-common/build.gradle.kts +++ b/spark-connector/spark-common/build.gradle.kts @@ -39,6 +39,9 @@ val scalaCollectionCompatVersion: String = libs.versions.scala.collection.compat dependencies { implementation(project(":catalogs:catalog-common")) + implementation(project(":catalogs:catalog-jdbc-common")) { + exclude("org.apache.logging.log4j") + } implementation(libs.guava) compileOnly(project(":clients:client-java-runtime", configuration = "shadow")) @@ -78,6 +81,9 @@ dependencies { testImplementation(project(":server-common")) { exclude("org.apache.logging.log4j") } + testImplementation(project(":catalogs:catalog-jdbc-mysql")) { + exclude("org.apache.logging.log4j") + } testImplementation(project(":integration-test-common", "testArtifacts")) testImplementation(libs.hive2.common) { diff --git a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/jdbc/GravitinoJdbcCatalog.java b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/jdbc/GravitinoJdbcCatalog.java index c079d4376a3..60fbc6aa83f 100644 --- a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/jdbc/GravitinoJdbcCatalog.java +++ b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/jdbc/GravitinoJdbcCatalog.java @@ -19,14 +19,18 @@ package org.apache.gravitino.spark.connector.jdbc; +import com.google.common.collect.Maps; import java.util.Map; import org.apache.gravitino.spark.connector.PropertiesConverter; import org.apache.gravitino.spark.connector.SparkTransformConverter; import org.apache.gravitino.spark.connector.SparkTypeConverter; import org.apache.gravitino.spark.connector.catalog.BaseCatalog; +import org.apache.spark.sql.catalyst.analysis.NamespaceAlreadyExistsException; import org.apache.spark.sql.connector.catalog.Identifier; +import org.apache.spark.sql.connector.catalog.SupportsNamespaces; import org.apache.spark.sql.connector.catalog.Table; import org.apache.spark.sql.connector.catalog.TableCatalog; +import org.apache.spark.sql.execution.datasources.v2.jdbc.JDBCTable; import org.apache.spark.sql.execution.datasources.v2.jdbc.JDBCTableCatalog; import org.apache.spark.sql.util.CaseInsensitiveStringMap; @@ -51,7 +55,14 @@ protected Table createSparkTable( PropertiesConverter propertiesConverter, SparkTransformConverter sparkTransformConverter, SparkTypeConverter sparkTypeConverter) { - return sparkTable; + return new SparkJdbcTable( + identifier, + gravitinoTable, + (JDBCTable) sparkTable, + (JDBCTableCatalog) sparkCatalog, + propertiesConverter, + sparkTransformConverter, + sparkTypeConverter); } @Override @@ -63,4 +74,16 @@ protected PropertiesConverter getPropertiesConverter() { protected SparkTransformConverter getSparkTransformConverter() { return new SparkTransformConverter(false); } + + @Override + protected SparkTypeConverter getSparkTypeConverter() { + return new SparkJdbcTypeConverter(); + } + + @Override + public void createNamespace(String[] namespace, Map metadata) + throws NamespaceAlreadyExistsException { + super.createNamespace( + namespace, Maps.filterKeys(metadata, key -> key.equals(SupportsNamespaces.PROP_COMMENT))); + } } diff --git a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/jdbc/SparkJdbcTable.java b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/jdbc/SparkJdbcTable.java new file mode 100644 index 00000000000..3de807c3685 --- /dev/null +++ b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/jdbc/SparkJdbcTable.java @@ -0,0 +1,71 @@ +/* + * 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.spark.connector.jdbc; + +import java.util.Map; +import org.apache.gravitino.rel.Table; +import org.apache.gravitino.spark.connector.PropertiesConverter; +import org.apache.gravitino.spark.connector.SparkTransformConverter; +import org.apache.gravitino.spark.connector.SparkTypeConverter; +import org.apache.gravitino.spark.connector.utils.GravitinoTableInfoHelper; +import org.apache.spark.sql.connector.catalog.Identifier; +import org.apache.spark.sql.execution.datasources.v2.jdbc.JDBCTable; +import org.apache.spark.sql.execution.datasources.v2.jdbc.JDBCTableCatalog; +import org.apache.spark.sql.types.StructType; + +public class SparkJdbcTable extends JDBCTable { + + private GravitinoTableInfoHelper gravitinoTableInfoHelper; + + public SparkJdbcTable( + Identifier identifier, + Table gravitinoTable, + JDBCTable jdbcTable, + JDBCTableCatalog jdbcTableCatalog, + PropertiesConverter propertiesConverter, + SparkTransformConverter sparkTransformConverter, + SparkTypeConverter sparkTypeConverter) { + super(identifier, jdbcTable.schema(), jdbcTable.jdbcOptions()); + this.gravitinoTableInfoHelper = + new GravitinoTableInfoHelper( + false, + identifier, + gravitinoTable, + propertiesConverter, + sparkTransformConverter, + sparkTypeConverter); + } + + @Override + public String name() { + return gravitinoTableInfoHelper.name(); + } + + @Override + @SuppressWarnings("deprecation") + public StructType schema() { + return gravitinoTableInfoHelper.schema(); + } + + @Override + public Map properties() { + return gravitinoTableInfoHelper.properties(); + } +} diff --git a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/jdbc/SparkJdbcTypeConverter.java b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/jdbc/SparkJdbcTypeConverter.java new file mode 100644 index 00000000000..a73d592c7d6 --- /dev/null +++ b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/jdbc/SparkJdbcTypeConverter.java @@ -0,0 +1,38 @@ +/* + * 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.spark.connector.jdbc; + +import org.apache.gravitino.rel.types.Type; +import org.apache.gravitino.rel.types.Types; +import org.apache.gravitino.spark.connector.SparkTypeConverter; +import org.apache.spark.sql.types.DataType; +import org.apache.spark.sql.types.DataTypes; + +public class SparkJdbcTypeConverter extends SparkTypeConverter { + + @Override + public DataType toSparkType(Type gravitinoType) { + if (gravitinoType instanceof Types.VarCharType) { + return DataTypes.StringType; + } else { + return super.toSparkType(gravitinoType); + } + } +} diff --git a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/version/CatalogNameAdaptor.java b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/version/CatalogNameAdaptor.java index 41e769ca2b4..9d8594b9124 100644 --- a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/version/CatalogNameAdaptor.java +++ b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/version/CatalogNameAdaptor.java @@ -46,13 +46,22 @@ public class CatalogNameAdaptor { "lakehouse-paimon-3.5", "org.apache.gravitino.spark.connector.paimon.GravitinoPaimonCatalogSpark35"); + private static final Map jdbcCatalogNames = + ImmutableMap.of( + "3.3", + "org.apache.gravitino.spark.connector.jdbc.GravitinoJdbcCatalogSpark33", + "3.4", + "org.apache.gravitino.spark.connector.jdbc.GravitinoJdbcCatalogSpark34", + "3.5", + "org.apache.gravitino.spark.connector.jdbc.GravitinoJdbcCatalogSpark35"); + private static String sparkVersion() { return package$.MODULE$.SPARK_VERSION(); } private static String getCatalogName(String provider, int majorVersion, int minorVersion) { if (provider.startsWith("jdbc")) { - return "org.apache.gravitino.spark.connector.jdbc.GravitinoJdbcCatalog"; + return jdbcCatalogNames.get(String.format("%d.%d", majorVersion, minorVersion)); } String key = String.format("%s-%d.%d", provider.toLowerCase(Locale.ROOT), majorVersion, minorVersion); diff --git a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/SparkCommonIT.java b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/SparkCommonIT.java index c7517a3bf82..0bea3c21910 100644 --- a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/SparkCommonIT.java +++ b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/SparkCommonIT.java @@ -119,6 +119,14 @@ private static String getRowLevelDeleteTableSql( protected abstract boolean supportsReplaceColumns(); + protected abstract boolean supportsProperties(); + + protected abstract boolean supportsComplexType(); + + protected SparkTableInfoChecker getTableInfoChecker() { + return SparkTableInfoChecker.create(); + } + // Use a custom database not the original default database because SparkCommonIT couldn't // read&write data to tables in default database. The main reason is default database location is // determined by `hive.metastore.warehouse.dir` in hive-site.xml which is local HDFS address @@ -189,6 +197,7 @@ void testLoadCatalogs() { } @Test + @EnabledIf("supportsProperties") protected void testCreateAndLoadSchema() { String testDatabaseName = "t_create1"; dropDatabaseIfExists(testDatabaseName); @@ -218,6 +227,7 @@ protected void testCreateAndLoadSchema() { } @Test + @EnabledIf("supportsProperties") protected void testAlterSchema() { String testDatabaseName = "t_alter"; dropDatabaseIfExists(testDatabaseName); @@ -266,7 +276,7 @@ void testCreateSimpleTable() { SparkTableInfo tableInfo = getTableInfo(tableName); SparkTableInfoChecker checker = - SparkTableInfoChecker.create() + getTableInfoChecker() .withName(tableName) .withColumns(getSimpleTableColumn()) .withComment(null); @@ -287,7 +297,7 @@ void testCreateTableWithDatabase() { createSimpleTable(tableIdentifier); SparkTableInfo tableInfo = getTableInfo(tableIdentifier); SparkTableInfoChecker checker = - SparkTableInfoChecker.create().withName(tableName).withColumns(getSimpleTableColumn()); + getTableInfoChecker().withName(tableName).withColumns(getSimpleTableColumn()); checker.check(tableInfo); checkTableReadWrite(tableInfo); @@ -300,8 +310,7 @@ void testCreateTableWithDatabase() { dropTableIfExists(tableName); createSimpleTable(tableName); tableInfo = getTableInfo(tableName); - checker = - SparkTableInfoChecker.create().withName(tableName).withColumns(getSimpleTableColumn()); + checker = getTableInfoChecker().withName(tableName).withColumns(getSimpleTableColumn()); checker.check(tableInfo); checkTableReadWrite(tableInfo); } @@ -317,7 +326,7 @@ void testCreateTableWithComment() { SparkTableInfo tableInfo = getTableInfo(tableName); SparkTableInfoChecker checker = - SparkTableInfoChecker.create() + getTableInfoChecker() .withName(tableName) .withColumns(getSimpleTableColumn()) .withComment(tableComment); @@ -396,6 +405,7 @@ void testListTable() { } @Test + @EnabledIf("supportsProperties") void testAlterTableSetAndRemoveProperty() { String tableName = "test_property"; dropTableIfExists(tableName); @@ -425,8 +435,7 @@ void testAlterTableUpdateComment() { "ALTER TABLE %s SET TBLPROPERTIES('%s'='%s')", tableName, ConnectorConstants.COMMENT, comment)); SparkTableInfo tableInfo = getTableInfo(tableName); - SparkTableInfoChecker checker = - SparkTableInfoChecker.create().withName(tableName).withComment(comment); + SparkTableInfoChecker checker = getTableInfoChecker().withName(tableName).withComment(comment); checker.check(tableInfo); } @@ -593,6 +602,7 @@ protected void testAlterTableReplaceColumns() { } @Test + @EnabledIf("supportsComplexType") void testComplexType() { String tableName = "complex_type_table"; dropTableIfExists(tableName); @@ -632,7 +642,7 @@ void testCreateDatasourceFormatPartitionTable() { sql(createTableSQL); SparkTableInfo tableInfo = getTableInfo(tableName); SparkTableInfoChecker checker = - SparkTableInfoChecker.create() + getTableInfoChecker() .withName(tableName) .withColumns(getSimpleTableColumn()) .withIdentifyPartition(Arrays.asList("name", "age")); @@ -652,7 +662,7 @@ void testCreateBucketTable() { sql(createTableSQL); SparkTableInfo tableInfo = getTableInfo(tableName); SparkTableInfoChecker checker = - SparkTableInfoChecker.create() + getTableInfoChecker() .withName(tableName) .withColumns(getSimpleTableColumn()) .withBucket(4, Arrays.asList("id", "name")); @@ -672,7 +682,7 @@ void testCreateSortBucketTable() { sql(createTableSQL); SparkTableInfo tableInfo = getTableInfo(tableName); SparkTableInfoChecker checker = - SparkTableInfoChecker.create() + getTableInfoChecker() .withName(tableName) .withColumns(getSimpleTableColumn()) .withBucket(4, Arrays.asList("id", "name"), Arrays.asList("name", "id")); @@ -695,7 +705,7 @@ void testCreateTableAsSelect() { SparkTableInfo newTableInfo = getTableInfo(newTableName); SparkTableInfoChecker checker = - SparkTableInfoChecker.create().withName(newTableName).withColumns(getSimpleTableColumn()); + getTableInfoChecker().withName(newTableName).withColumns(getSimpleTableColumn()); checker.check(newTableInfo); List tableData = getTableData(newTableName); @@ -797,6 +807,7 @@ protected void deleteDirIfExists(String path) { } @Test + @EnabledIf("supportsProperties") void testTableOptions() { String tableName = "options_table"; dropTableIfExists(tableName); @@ -806,7 +817,7 @@ void testTableOptions() { SparkTableInfo tableInfo = getTableInfo(tableName); SparkTableInfoChecker checker = - SparkTableInfoChecker.create() + getTableInfoChecker() .withName(tableName) .withTableProperties(ImmutableMap.of(TableCatalog.OPTION_PREFIX + "a", "b")); checker.check(tableInfo); @@ -983,7 +994,7 @@ protected void createSimpleTable(String identifier) { protected void checkTableColumns( String tableName, List columns, SparkTableInfo tableInfo) { - SparkTableInfoChecker.create() + getTableInfoChecker() .withName(tableName) .withColumns(columns) .withComment(null) diff --git a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/SparkEnvIT.java b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/SparkEnvIT.java index b534a9772f7..70c2af2becc 100644 --- a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/SparkEnvIT.java +++ b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/SparkEnvIT.java @@ -19,10 +19,12 @@ package org.apache.gravitino.spark.connector.integration.test; +import static org.apache.gravitino.integration.test.util.TestDatabaseName.MYSQL_CATALOG_MYSQL_IT; import static org.apache.gravitino.spark.connector.PropertiesConverter.SPARK_PROPERTY_PREFIX; import static org.apache.gravitino.spark.connector.iceberg.IcebergPropertiesConstants.ICEBERG_CATALOG_CACHE_ENABLED; import java.io.IOException; +import java.sql.SQLException; import java.util.Collections; import java.util.HashMap; import java.util.Map; @@ -56,6 +58,10 @@ public abstract class SparkEnvIT extends SparkUtilIT { protected String warehouse; protected FileSystem hdfs; protected String icebergRestServiceUri; + protected String mysqlUrl; + protected String mysqlUsername; + protected String mysqlPassword; + protected String mysqlDriver; private final String metalakeName = "test"; private SparkSession sparkSession; @@ -82,6 +88,7 @@ void startUp() throws Exception { if (lakeHouseIcebergProvider.equalsIgnoreCase(getProvider())) { initIcebergRestServiceEnv(); } + initMysqlEnv(); // Start Gravitino server super.startIntegrationTest(); initHdfsFileSystem(); @@ -151,6 +158,14 @@ private void initHiveEnv() { HiveContainer.HDFS_DEFAULTFS_PORT); } + private void initMysqlEnv() throws SQLException { + containerSuite.startMySQLContainer(MYSQL_CATALOG_MYSQL_IT); + mysqlUrl = containerSuite.getMySQLContainer().getJdbcUrl(); + mysqlUsername = containerSuite.getMySQLContainer().getUsername(); + mysqlPassword = containerSuite.getMySQLContainer().getPassword(); + mysqlDriver = containerSuite.getMySQLContainer().getDriverClassName(MYSQL_CATALOG_MYSQL_IT); + } + private void initIcebergRestServiceEnv() { ignoreIcebergRestService = false; Map icebergRestServiceConfigs = new HashMap<>(); diff --git a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/hive/SparkHiveCatalogIT.java b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/hive/SparkHiveCatalogIT.java index b95882a0d01..d472d2a3d09 100644 --- a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/hive/SparkHiveCatalogIT.java +++ b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/hive/SparkHiveCatalogIT.java @@ -84,6 +84,16 @@ protected boolean supportsReplaceColumns() { return true; } + @Override + protected boolean supportsProperties() { + return true; + } + + @Override + protected boolean supportsComplexType() { + return true; + } + @Test void testCreateHiveFormatPartitionTable() { String tableName = "hive_partition_table"; diff --git a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/iceberg/SparkIcebergCatalogIT.java b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/iceberg/SparkIcebergCatalogIT.java index f5fd337a13d..e3c9792fcf8 100644 --- a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/iceberg/SparkIcebergCatalogIT.java +++ b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/iceberg/SparkIcebergCatalogIT.java @@ -61,10 +61,12 @@ import org.apache.spark.sql.types.DataTypes; import org.apache.spark.sql.types.StructField; import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Tag; import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.MethodSource; +@Tag("gravitino-docker-test") public abstract class SparkIcebergCatalogIT extends SparkCommonIT { private static final String ICEBERG_FORMAT_VERSION = "format-version"; @@ -109,6 +111,16 @@ protected boolean supportsReplaceColumns() { return true; } + @Override + protected boolean supportsProperties() { + return true; + } + + @Override + protected boolean supportsComplexType() { + return true; + } + @Override protected String getTableLocation(SparkTableInfo table) { return String.join(File.separator, table.getTableLocation(), "data"); diff --git a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/jdbc/SparkJdbcCatalogIT.java b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/jdbc/SparkJdbcCatalogIT.java new file mode 100644 index 00000000000..3fbf97f291f --- /dev/null +++ b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/jdbc/SparkJdbcCatalogIT.java @@ -0,0 +1,89 @@ +/* + * 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.spark.connector.integration.test.jdbc; + +import com.google.common.collect.Maps; +import java.util.Map; +import org.apache.gravitino.spark.connector.integration.test.SparkCommonIT; +import org.apache.gravitino.spark.connector.integration.test.util.SparkTableInfoChecker; +import org.apache.gravitino.spark.connector.jdbc.JdbcPropertiesConstants; +import org.junit.jupiter.api.Tag; + +@Tag("gravitino-docker-test") +public abstract class SparkJdbcCatalogIT extends SparkCommonIT { + @Override + protected boolean supportsSparkSQLClusteredBy() { + return false; + } + + @Override + protected boolean supportsPartition() { + return false; + } + + @Override + protected boolean supportsDelete() { + return false; + } + + @Override + protected boolean supportsSchemaEvolution() { + return false; + } + + @Override + protected boolean supportsReplaceColumns() { + return false; + } + + @Override + protected boolean supportsProperties() { + return false; + } + + @Override + protected boolean supportsComplexType() { + return false; + } + + @Override + protected String getCatalogName() { + return "jdbc_mysql"; + } + + @Override + protected String getProvider() { + return "jdbc-mysql"; + } + + @Override + protected SparkTableInfoChecker getTableInfoChecker() { + return SparkJdbcTableInfoChecker.create(); + } + + @Override + protected Map getCatalogConfigs() { + Map catalogProperties = Maps.newHashMap(); + catalogProperties.put(JdbcPropertiesConstants.GRAVITINO_JDBC_URL, mysqlUrl); + catalogProperties.put(JdbcPropertiesConstants.GRAVITINO_JDBC_USER, mysqlUsername); + catalogProperties.put(JdbcPropertiesConstants.GRAVITINO_JDBC_PASSWORD, mysqlPassword); + catalogProperties.put(JdbcPropertiesConstants.GRAVITINO_JDBC_DRIVER, mysqlDriver); + return catalogProperties; + } +} diff --git a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/jdbc/SparkJdbcTableInfoChecker.java b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/jdbc/SparkJdbcTableInfoChecker.java new file mode 100644 index 00000000000..9240e051bca --- /dev/null +++ b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/jdbc/SparkJdbcTableInfoChecker.java @@ -0,0 +1,54 @@ +/* + * 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.spark.connector.integration.test.jdbc; + +import java.util.List; +import java.util.stream.Collectors; +import org.apache.commons.lang3.StringUtils; +import org.apache.gravitino.spark.connector.integration.test.util.SparkTableInfo; +import org.apache.gravitino.spark.connector.integration.test.util.SparkTableInfoChecker; + +public class SparkJdbcTableInfoChecker extends SparkTableInfoChecker { + + public static SparkJdbcTableInfoChecker create() { + return new SparkJdbcTableInfoChecker(); + } + + @Override + public SparkTableInfoChecker withColumns(List columns) { + getExpectedTableInfo() + .setColumns( + columns.stream() + .peek( + column -> + column.setComment( + StringUtils.isEmpty(column.getComment()) ? null : column.getComment())) + .collect(Collectors.toList())); + getCheckFields().add(CheckField.COLUMN); + return this; + } + + @Override + public SparkTableInfoChecker withComment(String comment) { + getExpectedTableInfo().setComment(StringUtils.isEmpty(comment) ? "" : comment); + getCheckFields().add(CheckField.COMMENT); + return this; + } +} diff --git a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/paimon/SparkPaimonCatalogIT.java b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/paimon/SparkPaimonCatalogIT.java index 9d036482857..fff31d13efe 100644 --- a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/paimon/SparkPaimonCatalogIT.java +++ b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/paimon/SparkPaimonCatalogIT.java @@ -63,6 +63,16 @@ protected boolean supportsSchemaEvolution() { return true; } + @Override + protected boolean supportsProperties() { + return true; + } + + @Override + protected boolean supportsComplexType() { + return true; + } + @Override protected boolean supportsReplaceColumns() { // Paimon doesn't support replace columns, because it doesn't support drop all fields in table. diff --git a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/util/SparkTableInfo.java b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/util/SparkTableInfo.java index 077936c29c5..74b3ea09685 100644 --- a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/util/SparkTableInfo.java +++ b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/util/SparkTableInfo.java @@ -31,6 +31,7 @@ import org.apache.gravitino.spark.connector.ConnectorConstants; import org.apache.gravitino.spark.connector.hive.SparkHiveTable; import org.apache.gravitino.spark.connector.iceberg.SparkIcebergTable; +import org.apache.gravitino.spark.connector.jdbc.SparkJdbcTable; import org.apache.gravitino.spark.connector.paimon.SparkPaimonTable; import org.apache.spark.sql.connector.catalog.SupportsMetadataColumns; import org.apache.spark.sql.connector.catalog.Table; @@ -193,6 +194,8 @@ private static StructType getSchema(Table baseTable) { return ((SparkIcebergTable) baseTable).schema(); } else if (baseTable instanceof SparkPaimonTable) { return ((SparkPaimonTable) baseTable).schema(); + } else if (baseTable instanceof SparkJdbcTable) { + return ((SparkJdbcTable) baseTable).schema(); } else { throw new IllegalArgumentException( "Doesn't support Spark table: " + baseTable.getClass().getName()); diff --git a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/util/SparkTableInfoChecker.java b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/util/SparkTableInfoChecker.java index 33a6a356828..bd7164af786 100644 --- a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/util/SparkTableInfoChecker.java +++ b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/util/SparkTableInfoChecker.java @@ -23,6 +23,7 @@ import java.util.List; import java.util.Map; import java.util.Set; +import lombok.Data; import org.apache.gravitino.spark.connector.SparkTransformConverter; import org.apache.gravitino.spark.connector.integration.test.util.SparkTableInfo.SparkColumnInfo; import org.apache.spark.sql.connector.expressions.Expressions; @@ -34,17 +35,18 @@ * To create an expected SparkTableInfo for verifying the SQL execution result, only the explicitly * set fields will be checked. */ +@Data public class SparkTableInfoChecker { private SparkTableInfo expectedTableInfo = new SparkTableInfo(); private Set checkFields = new LinkedHashSet<>(); - private SparkTableInfoChecker() {} + protected SparkTableInfoChecker() {} public static SparkTableInfoChecker create() { return new SparkTableInfoChecker(); } - private enum CheckField { + protected enum CheckField { NAME, COLUMN, PARTITION, diff --git a/spark-connector/v3.3/spark/src/main/java/org/apache/gravitino/spark/connector/jdbc/GravitinoJdbcCatalogSpark33.java b/spark-connector/v3.3/spark/src/main/java/org/apache/gravitino/spark/connector/jdbc/GravitinoJdbcCatalogSpark33.java new file mode 100644 index 00000000000..d322cd82ca0 --- /dev/null +++ b/spark-connector/v3.3/spark/src/main/java/org/apache/gravitino/spark/connector/jdbc/GravitinoJdbcCatalogSpark33.java @@ -0,0 +1,22 @@ +/* + * 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.spark.connector.jdbc; + +public class GravitinoJdbcCatalogSpark33 extends GravitinoJdbcCatalog {} diff --git a/spark-connector/v3.3/spark/src/test/java/org/apache/gravitino/spark/connector/integration/test/jdbc/SparkJdbcCatalogIT33.java b/spark-connector/v3.3/spark/src/test/java/org/apache/gravitino/spark/connector/integration/test/jdbc/SparkJdbcCatalogIT33.java new file mode 100644 index 00000000000..acef94e8bf5 --- /dev/null +++ b/spark-connector/v3.3/spark/src/test/java/org/apache/gravitino/spark/connector/integration/test/jdbc/SparkJdbcCatalogIT33.java @@ -0,0 +1,36 @@ +/* + * 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.spark.connector.integration.test.jdbc; + +import org.apache.gravitino.spark.connector.jdbc.GravitinoJdbcCatalogSpark33; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +public class SparkJdbcCatalogIT33 extends SparkJdbcCatalogIT { + @Test + void testCatalogClassName() { + String catalogClass = + getSparkSession() + .sessionState() + .conf() + .getConfString("spark.sql.catalog." + getCatalogName()); + Assertions.assertEquals(GravitinoJdbcCatalogSpark33.class.getName(), catalogClass); + } +} diff --git a/spark-connector/v3.4/spark/src/main/java/org/apache/gravitino/spark/connector/jdbc/GravitinoJdbcCatalogSpark34.java b/spark-connector/v3.4/spark/src/main/java/org/apache/gravitino/spark/connector/jdbc/GravitinoJdbcCatalogSpark34.java new file mode 100644 index 00000000000..e9c091c1882 --- /dev/null +++ b/spark-connector/v3.4/spark/src/main/java/org/apache/gravitino/spark/connector/jdbc/GravitinoJdbcCatalogSpark34.java @@ -0,0 +1,38 @@ +/* + * 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.spark.connector.jdbc; + +import org.apache.gravitino.spark.connector.SparkTableChangeConverter; +import org.apache.gravitino.spark.connector.SparkTableChangeConverter34; +import org.apache.gravitino.spark.connector.SparkTypeConverter; + +public class GravitinoJdbcCatalogSpark34 extends GravitinoJdbcCatalog { + + @Override + protected SparkTypeConverter getSparkTypeConverter() { + return new SparkJdbcTypeConverter34(); + } + + @Override + protected SparkTableChangeConverter getSparkTableChangeConverter( + SparkTypeConverter sparkTypeConverter) { + return new SparkTableChangeConverter34(sparkTypeConverter); + } +} diff --git a/spark-connector/v3.4/spark/src/main/java/org/apache/gravitino/spark/connector/jdbc/SparkJdbcTypeConverter34.java b/spark-connector/v3.4/spark/src/main/java/org/apache/gravitino/spark/connector/jdbc/SparkJdbcTypeConverter34.java new file mode 100644 index 00000000000..a152ba73cae --- /dev/null +++ b/spark-connector/v3.4/spark/src/main/java/org/apache/gravitino/spark/connector/jdbc/SparkJdbcTypeConverter34.java @@ -0,0 +1,37 @@ +/* + * 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.spark.connector.jdbc; + +import org.apache.gravitino.rel.types.Type; +import org.apache.gravitino.rel.types.Types; +import org.apache.gravitino.spark.connector.SparkTypeConverter34; +import org.apache.spark.sql.types.DataType; +import org.apache.spark.sql.types.DataTypes; + +public class SparkJdbcTypeConverter34 extends SparkTypeConverter34 { + @Override + public DataType toSparkType(Type gravitinoType) { + if (gravitinoType instanceof Types.VarCharType) { + return DataTypes.StringType; + } else { + return super.toSparkType(gravitinoType); + } + } +} diff --git a/spark-connector/v3.4/spark/src/test/java/org/apache/gravitino/spark/connector/integration/test/jdbc/SparkJdbcCatalogIT34.java b/spark-connector/v3.4/spark/src/test/java/org/apache/gravitino/spark/connector/integration/test/jdbc/SparkJdbcCatalogIT34.java new file mode 100644 index 00000000000..a105ed9aff2 --- /dev/null +++ b/spark-connector/v3.4/spark/src/test/java/org/apache/gravitino/spark/connector/integration/test/jdbc/SparkJdbcCatalogIT34.java @@ -0,0 +1,35 @@ +/* + * 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.spark.connector.integration.test.jdbc; + +import org.apache.gravitino.spark.connector.jdbc.GravitinoJdbcCatalogSpark34; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +public class SparkJdbcCatalogIT34 extends SparkJdbcCatalogIT { + @Test + void testCatalogClassName() { + String catalogClass = + getSparkSession() + .sessionState() + .conf() + .getConfString("spark.sql.catalog." + getCatalogName()); + Assertions.assertEquals(GravitinoJdbcCatalogSpark34.class.getName(), catalogClass); + } +} diff --git a/spark-connector/v3.5/spark/src/main/java/org/apache/gravitino/spark/connector/jdbc/GravitinoJdbcCatalogSpark35.java b/spark-connector/v3.5/spark/src/main/java/org/apache/gravitino/spark/connector/jdbc/GravitinoJdbcCatalogSpark35.java new file mode 100644 index 00000000000..31b8569f94a --- /dev/null +++ b/spark-connector/v3.5/spark/src/main/java/org/apache/gravitino/spark/connector/jdbc/GravitinoJdbcCatalogSpark35.java @@ -0,0 +1,22 @@ +/* + * 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.spark.connector.jdbc; + +public class GravitinoJdbcCatalogSpark35 extends GravitinoJdbcCatalogSpark34 {} diff --git a/spark-connector/v3.5/spark/src/test/java/org/apache/gravitino/spark/connector/integration/test/jdbc/SparkJdbcCatalogIT35.java b/spark-connector/v3.5/spark/src/test/java/org/apache/gravitino/spark/connector/integration/test/jdbc/SparkJdbcCatalogIT35.java new file mode 100644 index 00000000000..653dc0c297b --- /dev/null +++ b/spark-connector/v3.5/spark/src/test/java/org/apache/gravitino/spark/connector/integration/test/jdbc/SparkJdbcCatalogIT35.java @@ -0,0 +1,36 @@ +/* + * 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.spark.connector.integration.test.jdbc; + +import org.apache.gravitino.spark.connector.jdbc.GravitinoJdbcCatalogSpark35; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +public class SparkJdbcCatalogIT35 extends SparkJdbcCatalogIT { + @Test + void testCatalogClassName() { + String catalogClass = + getSparkSession() + .sessionState() + .conf() + .getConfString("spark.sql.catalog." + getCatalogName()); + Assertions.assertEquals(GravitinoJdbcCatalogSpark35.class.getName(), catalogClass); + } +} From 41076d9d4ecedb4d954aaa05c3c17b273817b71e Mon Sep 17 00:00:00 2001 From: youze Liang <525672876@qq.com> Date: Mon, 20 Jan 2025 18:37:12 +0800 Subject: [PATCH 4/8] fix dependencies --- spark-connector/spark-common/build.gradle.kts | 3 --- spark-connector/v3.3/spark/build.gradle.kts | 1 + spark-connector/v3.4/spark/build.gradle.kts | 1 + spark-connector/v3.5/spark/build.gradle.kts | 1 + 4 files changed, 3 insertions(+), 3 deletions(-) diff --git a/spark-connector/spark-common/build.gradle.kts b/spark-connector/spark-common/build.gradle.kts index 7de83c05791..2d1cccfd19e 100644 --- a/spark-connector/spark-common/build.gradle.kts +++ b/spark-connector/spark-common/build.gradle.kts @@ -81,9 +81,6 @@ dependencies { testImplementation(project(":server-common")) { exclude("org.apache.logging.log4j") } - testImplementation(project(":catalogs:catalog-jdbc-mysql")) { - exclude("org.apache.logging.log4j") - } testImplementation(project(":integration-test-common", "testArtifacts")) testImplementation(libs.hive2.common) { diff --git a/spark-connector/v3.3/spark/build.gradle.kts b/spark-connector/v3.3/spark/build.gradle.kts index 66c65f863b9..5046bf2557e 100644 --- a/spark-connector/v3.3/spark/build.gradle.kts +++ b/spark-connector/v3.3/spark/build.gradle.kts @@ -163,6 +163,7 @@ tasks.test { dependsOn(":catalogs:catalog-hive:jar") dependsOn(":iceberg:iceberg-rest-server:jar") dependsOn(":catalogs:catalog-lakehouse-paimon:jar") + dependsOn(":catalogs:catalog-jdbc-mysql:jar") } } diff --git a/spark-connector/v3.4/spark/build.gradle.kts b/spark-connector/v3.4/spark/build.gradle.kts index aa4134a3c71..77904f6beee 100644 --- a/spark-connector/v3.4/spark/build.gradle.kts +++ b/spark-connector/v3.4/spark/build.gradle.kts @@ -163,6 +163,7 @@ tasks.test { dependsOn(":catalogs:catalog-hive:jar") dependsOn(":iceberg:iceberg-rest-server:jar") dependsOn(":catalogs:catalog-lakehouse-paimon:jar") + dependsOn(":catalogs:catalog-jdbc-mysql:jar") } } diff --git a/spark-connector/v3.5/spark/build.gradle.kts b/spark-connector/v3.5/spark/build.gradle.kts index 15aa018081d..f4c7c128112 100644 --- a/spark-connector/v3.5/spark/build.gradle.kts +++ b/spark-connector/v3.5/spark/build.gradle.kts @@ -165,6 +165,7 @@ tasks.test { dependsOn(":catalogs:catalog-hive:jar") dependsOn(":iceberg:iceberg-rest-server:jar") dependsOn(":catalogs:catalog-lakehouse-paimon:jar") + dependsOn(":catalogs:catalog-jdbc-mysql:jar") } } From 83baa668f837d1ac0bb02ee7f22c7a9f6640be61 Mon Sep 17 00:00:00 2001 From: youze Liang <525672876@qq.com> Date: Wed, 22 Jan 2025 11:37:34 +0800 Subject: [PATCH 5/8] resolve conversation --- docs/spark-connector/spark-catalog-jdbc.md | 3 +++ docs/spark-connector/spark-connector.md | 2 +- spark-connector/spark-common/build.gradle.kts | 3 --- .../jdbc/SparkJdbcTypeConverter.java | 2 ++ .../integration/test/SparkCommonIT.java | 10 +++++----- .../test/hive/SparkHiveCatalogIT.java | 2 +- .../test/iceberg/SparkIcebergCatalogIT.java | 4 +--- ...ogIT.java => SparkJdbcMysqlCatalogIT.java} | 4 ++-- .../test/jdbc/SparkJdbcTableInfoChecker.java | 1 + .../test/paimon/SparkPaimonCatalogIT.java | 2 +- spark-connector/v3.3/spark/build.gradle.kts | 3 +++ ...33.java => SparkJdbcMysqlCatalogIT33.java} | 2 +- spark-connector/v3.4/spark/build.gradle.kts | 3 +++ .../jdbc/SparkJdbcTypeConverter34.java | 2 ++ ...34.java => SparkJdbcMysqlCatalogIT34.java} | 2 +- spark-connector/v3.5/spark/build.gradle.kts | 3 +++ .../jdbc/GravitinoJdbcCatalogSpark35.java | 19 ++++++++++++++++++- ...35.java => SparkJdbcMysqlCatalogIT35.java} | 2 +- 18 files changed, 49 insertions(+), 20 deletions(-) rename spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/jdbc/{SparkJdbcCatalogIT.java => SparkJdbcMysqlCatalogIT.java} (95%) rename spark-connector/v3.3/spark/src/test/java/org/apache/gravitino/spark/connector/integration/test/jdbc/{SparkJdbcCatalogIT33.java => SparkJdbcMysqlCatalogIT33.java} (94%) rename spark-connector/v3.4/spark/src/test/java/org/apache/gravitino/spark/connector/integration/test/jdbc/{SparkJdbcCatalogIT34.java => SparkJdbcMysqlCatalogIT34.java} (94%) rename spark-connector/v3.5/spark/src/test/java/org/apache/gravitino/spark/connector/integration/test/jdbc/{SparkJdbcCatalogIT35.java => SparkJdbcMysqlCatalogIT35.java} (94%) diff --git a/docs/spark-connector/spark-catalog-jdbc.md b/docs/spark-connector/spark-catalog-jdbc.md index 3b1a3a7616c..7805d80266f 100644 --- a/docs/spark-connector/spark-catalog-jdbc.md +++ b/docs/spark-connector/spark-catalog-jdbc.md @@ -9,6 +9,8 @@ The Apache Gravitino Spark connector offers the capability to read JDBC tables, ## Capabilities +Supports MySQL and PostgreSQL. For OceanBase which is compatible with Mysql Dialects could use Mysql driver and Mysql Dialects as a trackoff way. But for Doris which do not support MySQL Dialects, are not currently supported. + #### Support DML and DDL operations: - `CREATE TABLE` @@ -22,6 +24,7 @@ The Apache Gravitino Spark connector offers the capability to read JDBC tables, ::: #### Not supported operations: + - `UPDATE` - `DELETE` - `TRUNCATE` diff --git a/docs/spark-connector/spark-connector.md b/docs/spark-connector/spark-connector.md index a7c47d51f22..a982313433a 100644 --- a/docs/spark-connector/spark-connector.md +++ b/docs/spark-connector/spark-connector.md @@ -11,7 +11,7 @@ The Apache Gravitino Spark connector leverages the Spark DataSourceV2 interface ## Capabilities -1. Supports [Hive catalog](spark-catalog-hive.md), [Iceberg catalog](spark-catalog-iceberg.md) and [Paimon catalog](spark-catalog-paimon.md). +1. Supports [Hive catalog](spark-catalog-hive.md), [Iceberg catalog](spark-catalog-iceberg.md), [Paimon catalog](spark-catalog-paimon.md) and [Jdbc catalog](spark-catalog-jdbc.md). 2. Supports federation query. 3. Supports most DDL and DML SQLs. diff --git a/spark-connector/spark-common/build.gradle.kts b/spark-connector/spark-common/build.gradle.kts index 2d1cccfd19e..06e0077d21e 100644 --- a/spark-connector/spark-common/build.gradle.kts +++ b/spark-connector/spark-common/build.gradle.kts @@ -39,9 +39,6 @@ val scalaCollectionCompatVersion: String = libs.versions.scala.collection.compat dependencies { implementation(project(":catalogs:catalog-common")) - implementation(project(":catalogs:catalog-jdbc-common")) { - exclude("org.apache.logging.log4j") - } implementation(libs.guava) compileOnly(project(":clients:client-java-runtime", configuration = "shadow")) diff --git a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/jdbc/SparkJdbcTypeConverter.java b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/jdbc/SparkJdbcTypeConverter.java index a73d592c7d6..56e2734a7f4 100644 --- a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/jdbc/SparkJdbcTypeConverter.java +++ b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/jdbc/SparkJdbcTypeConverter.java @@ -29,6 +29,8 @@ public class SparkJdbcTypeConverter extends SparkTypeConverter { @Override public DataType toSparkType(Type gravitinoType) { + // if spark version lower than 3.4.4, using VarCharType will throw an exception: Unsupported + // type varchar. if (gravitinoType instanceof Types.VarCharType) { return DataTypes.StringType; } else { diff --git a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/SparkCommonIT.java b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/SparkCommonIT.java index 0bea3c21910..2eb9e7b9b5a 100644 --- a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/SparkCommonIT.java +++ b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/SparkCommonIT.java @@ -119,7 +119,7 @@ private static String getRowLevelDeleteTableSql( protected abstract boolean supportsReplaceColumns(); - protected abstract boolean supportsProperties(); + protected abstract boolean supportsSchemaAndTableProperties(); protected abstract boolean supportsComplexType(); @@ -197,7 +197,7 @@ void testLoadCatalogs() { } @Test - @EnabledIf("supportsProperties") + @EnabledIf("supportsSchemaAndTableProperties") protected void testCreateAndLoadSchema() { String testDatabaseName = "t_create1"; dropDatabaseIfExists(testDatabaseName); @@ -227,7 +227,7 @@ protected void testCreateAndLoadSchema() { } @Test - @EnabledIf("supportsProperties") + @EnabledIf("supportsSchemaAndTableProperties") protected void testAlterSchema() { String testDatabaseName = "t_alter"; dropDatabaseIfExists(testDatabaseName); @@ -405,7 +405,7 @@ void testListTable() { } @Test - @EnabledIf("supportsProperties") + @EnabledIf("supportsSchemaAndTableProperties") void testAlterTableSetAndRemoveProperty() { String tableName = "test_property"; dropTableIfExists(tableName); @@ -807,7 +807,7 @@ protected void deleteDirIfExists(String path) { } @Test - @EnabledIf("supportsProperties") + @EnabledIf("supportsSchemaAndTableProperties") void testTableOptions() { String tableName = "options_table"; dropTableIfExists(tableName); diff --git a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/hive/SparkHiveCatalogIT.java b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/hive/SparkHiveCatalogIT.java index d472d2a3d09..6ed8e12d647 100644 --- a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/hive/SparkHiveCatalogIT.java +++ b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/hive/SparkHiveCatalogIT.java @@ -85,7 +85,7 @@ protected boolean supportsReplaceColumns() { } @Override - protected boolean supportsProperties() { + protected boolean supportsSchemaAndTableProperties() { return true; } diff --git a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/iceberg/SparkIcebergCatalogIT.java b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/iceberg/SparkIcebergCatalogIT.java index e3c9792fcf8..291f8f25dbf 100644 --- a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/iceberg/SparkIcebergCatalogIT.java +++ b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/iceberg/SparkIcebergCatalogIT.java @@ -61,12 +61,10 @@ import org.apache.spark.sql.types.DataTypes; import org.apache.spark.sql.types.StructField; import org.junit.jupiter.api.Assertions; -import org.junit.jupiter.api.Tag; import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.MethodSource; -@Tag("gravitino-docker-test") public abstract class SparkIcebergCatalogIT extends SparkCommonIT { private static final String ICEBERG_FORMAT_VERSION = "format-version"; @@ -112,7 +110,7 @@ protected boolean supportsReplaceColumns() { } @Override - protected boolean supportsProperties() { + protected boolean supportsSchemaAndTableProperties() { return true; } diff --git a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/jdbc/SparkJdbcCatalogIT.java b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/jdbc/SparkJdbcMysqlCatalogIT.java similarity index 95% rename from spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/jdbc/SparkJdbcCatalogIT.java rename to spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/jdbc/SparkJdbcMysqlCatalogIT.java index 3fbf97f291f..65ed937be3a 100644 --- a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/jdbc/SparkJdbcCatalogIT.java +++ b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/jdbc/SparkJdbcMysqlCatalogIT.java @@ -26,7 +26,7 @@ import org.junit.jupiter.api.Tag; @Tag("gravitino-docker-test") -public abstract class SparkJdbcCatalogIT extends SparkCommonIT { +public abstract class SparkJdbcMysqlCatalogIT extends SparkCommonIT { @Override protected boolean supportsSparkSQLClusteredBy() { return false; @@ -53,7 +53,7 @@ protected boolean supportsReplaceColumns() { } @Override - protected boolean supportsProperties() { + protected boolean supportsSchemaAndTableProperties() { return false; } diff --git a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/jdbc/SparkJdbcTableInfoChecker.java b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/jdbc/SparkJdbcTableInfoChecker.java index 9240e051bca..32a66923cbe 100644 --- a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/jdbc/SparkJdbcTableInfoChecker.java +++ b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/jdbc/SparkJdbcTableInfoChecker.java @@ -31,6 +31,7 @@ public static SparkJdbcTableInfoChecker create() { return new SparkJdbcTableInfoChecker(); } + // Spark jdbc table cannot distinguish between comment=null and comment="" @Override public SparkTableInfoChecker withColumns(List columns) { getExpectedTableInfo() diff --git a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/paimon/SparkPaimonCatalogIT.java b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/paimon/SparkPaimonCatalogIT.java index fff31d13efe..40afa060859 100644 --- a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/paimon/SparkPaimonCatalogIT.java +++ b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/paimon/SparkPaimonCatalogIT.java @@ -64,7 +64,7 @@ protected boolean supportsSchemaEvolution() { } @Override - protected boolean supportsProperties() { + protected boolean supportsSchemaAndTableProperties() { return true; } diff --git a/spark-connector/v3.3/spark/build.gradle.kts b/spark-connector/v3.3/spark/build.gradle.kts index 5046bf2557e..6b633434e46 100644 --- a/spark-connector/v3.3/spark/build.gradle.kts +++ b/spark-connector/v3.3/spark/build.gradle.kts @@ -52,6 +52,9 @@ dependencies { exclude("org.apache.logging.log4j") exclude("org.slf4j") } + testImplementation(project(":catalogs:catalog-jdbc-common")) { + exclude("org.apache.logging.log4j") + } testImplementation(project(":catalogs:hive-metastore-common")) { exclude("*") } diff --git a/spark-connector/v3.3/spark/src/test/java/org/apache/gravitino/spark/connector/integration/test/jdbc/SparkJdbcCatalogIT33.java b/spark-connector/v3.3/spark/src/test/java/org/apache/gravitino/spark/connector/integration/test/jdbc/SparkJdbcMysqlCatalogIT33.java similarity index 94% rename from spark-connector/v3.3/spark/src/test/java/org/apache/gravitino/spark/connector/integration/test/jdbc/SparkJdbcCatalogIT33.java rename to spark-connector/v3.3/spark/src/test/java/org/apache/gravitino/spark/connector/integration/test/jdbc/SparkJdbcMysqlCatalogIT33.java index acef94e8bf5..cf190cfd4fb 100644 --- a/spark-connector/v3.3/spark/src/test/java/org/apache/gravitino/spark/connector/integration/test/jdbc/SparkJdbcCatalogIT33.java +++ b/spark-connector/v3.3/spark/src/test/java/org/apache/gravitino/spark/connector/integration/test/jdbc/SparkJdbcMysqlCatalogIT33.java @@ -23,7 +23,7 @@ import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; -public class SparkJdbcCatalogIT33 extends SparkJdbcCatalogIT { +public class SparkJdbcMysqlCatalogIT33 extends SparkJdbcMysqlCatalogIT { @Test void testCatalogClassName() { String catalogClass = diff --git a/spark-connector/v3.4/spark/build.gradle.kts b/spark-connector/v3.4/spark/build.gradle.kts index 77904f6beee..08ab9ca9caf 100644 --- a/spark-connector/v3.4/spark/build.gradle.kts +++ b/spark-connector/v3.4/spark/build.gradle.kts @@ -53,6 +53,9 @@ dependencies { exclude("org.apache.logging.log4j") exclude("org.slf4j") } + testImplementation(project(":catalogs:catalog-jdbc-common")) { + exclude("org.apache.logging.log4j") + } testImplementation(project(":catalogs:hive-metastore-common")) { exclude("*") } diff --git a/spark-connector/v3.4/spark/src/main/java/org/apache/gravitino/spark/connector/jdbc/SparkJdbcTypeConverter34.java b/spark-connector/v3.4/spark/src/main/java/org/apache/gravitino/spark/connector/jdbc/SparkJdbcTypeConverter34.java index a152ba73cae..bbd32e0225d 100644 --- a/spark-connector/v3.4/spark/src/main/java/org/apache/gravitino/spark/connector/jdbc/SparkJdbcTypeConverter34.java +++ b/spark-connector/v3.4/spark/src/main/java/org/apache/gravitino/spark/connector/jdbc/SparkJdbcTypeConverter34.java @@ -28,6 +28,8 @@ public class SparkJdbcTypeConverter34 extends SparkTypeConverter34 { @Override public DataType toSparkType(Type gravitinoType) { + // if spark version lower than 3.4.4, using VarCharType will throw an exception: Unsupported + // type varchar. if (gravitinoType instanceof Types.VarCharType) { return DataTypes.StringType; } else { diff --git a/spark-connector/v3.4/spark/src/test/java/org/apache/gravitino/spark/connector/integration/test/jdbc/SparkJdbcCatalogIT34.java b/spark-connector/v3.4/spark/src/test/java/org/apache/gravitino/spark/connector/integration/test/jdbc/SparkJdbcMysqlCatalogIT34.java similarity index 94% rename from spark-connector/v3.4/spark/src/test/java/org/apache/gravitino/spark/connector/integration/test/jdbc/SparkJdbcCatalogIT34.java rename to spark-connector/v3.4/spark/src/test/java/org/apache/gravitino/spark/connector/integration/test/jdbc/SparkJdbcMysqlCatalogIT34.java index a105ed9aff2..9a4038404d8 100644 --- a/spark-connector/v3.4/spark/src/test/java/org/apache/gravitino/spark/connector/integration/test/jdbc/SparkJdbcCatalogIT34.java +++ b/spark-connector/v3.4/spark/src/test/java/org/apache/gravitino/spark/connector/integration/test/jdbc/SparkJdbcMysqlCatalogIT34.java @@ -22,7 +22,7 @@ import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; -public class SparkJdbcCatalogIT34 extends SparkJdbcCatalogIT { +public class SparkJdbcMysqlCatalogIT34 extends SparkJdbcMysqlCatalogIT { @Test void testCatalogClassName() { String catalogClass = diff --git a/spark-connector/v3.5/spark/build.gradle.kts b/spark-connector/v3.5/spark/build.gradle.kts index f4c7c128112..782d514aed4 100644 --- a/spark-connector/v3.5/spark/build.gradle.kts +++ b/spark-connector/v3.5/spark/build.gradle.kts @@ -53,6 +53,9 @@ dependencies { testImplementation(project(":api")) { exclude("org.apache.logging.log4j") } + testImplementation(project(":catalogs:catalog-jdbc-common")) { + exclude("org.apache.logging.log4j") + } testImplementation(project(":catalogs:hive-metastore-common")) { exclude("*") } diff --git a/spark-connector/v3.5/spark/src/main/java/org/apache/gravitino/spark/connector/jdbc/GravitinoJdbcCatalogSpark35.java b/spark-connector/v3.5/spark/src/main/java/org/apache/gravitino/spark/connector/jdbc/GravitinoJdbcCatalogSpark35.java index 31b8569f94a..1b10d63fa09 100644 --- a/spark-connector/v3.5/spark/src/main/java/org/apache/gravitino/spark/connector/jdbc/GravitinoJdbcCatalogSpark35.java +++ b/spark-connector/v3.5/spark/src/main/java/org/apache/gravitino/spark/connector/jdbc/GravitinoJdbcCatalogSpark35.java @@ -19,4 +19,21 @@ package org.apache.gravitino.spark.connector.jdbc; -public class GravitinoJdbcCatalogSpark35 extends GravitinoJdbcCatalogSpark34 {} +import org.apache.gravitino.spark.connector.SparkTableChangeConverter; +import org.apache.gravitino.spark.connector.SparkTableChangeConverter34; +import org.apache.gravitino.spark.connector.SparkTypeConverter; +import org.apache.gravitino.spark.connector.SparkTypeConverter34; + +public class GravitinoJdbcCatalogSpark35 extends GravitinoJdbcCatalog { + + @Override + protected SparkTypeConverter getSparkTypeConverter() { + return new SparkTypeConverter34(); + } + + @Override + protected SparkTableChangeConverter getSparkTableChangeConverter( + SparkTypeConverter sparkTypeConverter) { + return new SparkTableChangeConverter34(sparkTypeConverter); + } +} diff --git a/spark-connector/v3.5/spark/src/test/java/org/apache/gravitino/spark/connector/integration/test/jdbc/SparkJdbcCatalogIT35.java b/spark-connector/v3.5/spark/src/test/java/org/apache/gravitino/spark/connector/integration/test/jdbc/SparkJdbcMysqlCatalogIT35.java similarity index 94% rename from spark-connector/v3.5/spark/src/test/java/org/apache/gravitino/spark/connector/integration/test/jdbc/SparkJdbcCatalogIT35.java rename to spark-connector/v3.5/spark/src/test/java/org/apache/gravitino/spark/connector/integration/test/jdbc/SparkJdbcMysqlCatalogIT35.java index 653dc0c297b..00c14e40d2c 100644 --- a/spark-connector/v3.5/spark/src/test/java/org/apache/gravitino/spark/connector/integration/test/jdbc/SparkJdbcCatalogIT35.java +++ b/spark-connector/v3.5/spark/src/test/java/org/apache/gravitino/spark/connector/integration/test/jdbc/SparkJdbcMysqlCatalogIT35.java @@ -23,7 +23,7 @@ import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; -public class SparkJdbcCatalogIT35 extends SparkJdbcCatalogIT { +public class SparkJdbcMysqlCatalogIT35 extends SparkJdbcMysqlCatalogIT { @Test void testCatalogClassName() { String catalogClass = From 52bd7126cee49f8623d1348e6bec637824588e46 Mon Sep 17 00:00:00 2001 From: youze Liang <525672876@qq.com> Date: Thu, 23 Jan 2025 13:44:22 +0800 Subject: [PATCH 6/8] resolve conversation --- .../connector/jdbc/GravitinoJdbcCatalog.java | 23 ++++++++++++-- .../jdbc/JdbcPropertiesConverter.java | 30 +++++++++++-------- 2 files changed, 39 insertions(+), 14 deletions(-) diff --git a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/jdbc/GravitinoJdbcCatalog.java b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/jdbc/GravitinoJdbcCatalog.java index 60fbc6aa83f..3f36b9a2a84 100644 --- a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/jdbc/GravitinoJdbcCatalog.java +++ b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/jdbc/GravitinoJdbcCatalog.java @@ -30,6 +30,7 @@ import org.apache.spark.sql.connector.catalog.SupportsNamespaces; import org.apache.spark.sql.connector.catalog.Table; import org.apache.spark.sql.connector.catalog.TableCatalog; +import org.apache.spark.sql.errors.QueryCompilationErrors; import org.apache.spark.sql.execution.datasources.v2.jdbc.JDBCTable; import org.apache.spark.sql.execution.datasources.v2.jdbc.JDBCTableCatalog; import org.apache.spark.sql.util.CaseInsensitiveStringMap; @@ -83,7 +84,25 @@ protected SparkTypeConverter getSparkTypeConverter() { @Override public void createNamespace(String[] namespace, Map metadata) throws NamespaceAlreadyExistsException { - super.createNamespace( - namespace, Maps.filterKeys(metadata, key -> key.equals(SupportsNamespaces.PROP_COMMENT))); + Map properties = Maps.newHashMap(); + if (!metadata.isEmpty()) { + metadata.forEach( + (k, v) -> { + switch (k) { + case SupportsNamespaces.PROP_COMMENT: + properties.put(k, v); + break; + case SupportsNamespaces.PROP_OWNER: + break; + case SupportsNamespaces.PROP_LOCATION: + throw new RuntimeException( + QueryCompilationErrors.cannotCreateJDBCNamespaceUsingProviderError()); + default: + throw new RuntimeException( + QueryCompilationErrors.cannotCreateJDBCNamespaceWithPropertyError(k)); + } + }); + } + super.createNamespace(namespace, properties); } } diff --git a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/jdbc/JdbcPropertiesConverter.java b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/jdbc/JdbcPropertiesConverter.java index 308e548602c..7516646e343 100644 --- a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/jdbc/JdbcPropertiesConverter.java +++ b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/jdbc/JdbcPropertiesConverter.java @@ -20,6 +20,7 @@ package org.apache.gravitino.spark.connector.jdbc; import com.google.common.base.Preconditions; +import com.google.common.collect.ImmutableMap; import java.util.HashMap; import java.util.Map; import org.apache.gravitino.spark.connector.PropertiesConverter; @@ -36,22 +37,27 @@ public static JdbcPropertiesConverter getInstance() { return JdbcPropertiesConverterHolder.INSTANCE; } + private static final Map GRAVITINO_CONFIG_TO_JDBC = + ImmutableMap.of( + JdbcPropertiesConstants.GRAVITINO_JDBC_URL, + JdbcPropertiesConstants.SPARK_JDBC_URL, + JdbcPropertiesConstants.GRAVITINO_JDBC_USER, + JdbcPropertiesConstants.SPARK_JDBC_USER, + JdbcPropertiesConstants.GRAVITINO_JDBC_PASSWORD, + JdbcPropertiesConstants.SPARK_JDBC_PASSWORD, + JdbcPropertiesConstants.GRAVITINO_JDBC_DRIVER, + JdbcPropertiesConstants.SPARK_JDBC_DRIVER); + @Override public Map toSparkCatalogProperties(Map properties) { Preconditions.checkArgument(properties != null, "Jdbc Catalog properties should not be null"); HashMap jdbcProperties = new HashMap<>(); - jdbcProperties.put( - JdbcPropertiesConstants.SPARK_JDBC_URL, - properties.get(JdbcPropertiesConstants.GRAVITINO_JDBC_URL)); - jdbcProperties.put( - JdbcPropertiesConstants.SPARK_JDBC_USER, - properties.get(JdbcPropertiesConstants.GRAVITINO_JDBC_USER)); - jdbcProperties.put( - JdbcPropertiesConstants.SPARK_JDBC_PASSWORD, - properties.get(JdbcPropertiesConstants.GRAVITINO_JDBC_PASSWORD)); - jdbcProperties.put( - JdbcPropertiesConstants.SPARK_JDBC_DRIVER, - properties.get(JdbcPropertiesConstants.GRAVITINO_JDBC_DRIVER)); + properties.forEach( + (key, value) -> { + if (GRAVITINO_CONFIG_TO_JDBC.containsKey(key)) { + jdbcProperties.put(GRAVITINO_CONFIG_TO_JDBC.get(key), value); + } + }); return jdbcProperties; } From ebf2728dda663f21f861498f801b8d76ce10a0aa Mon Sep 17 00:00:00 2001 From: youze Liang <525672876@qq.com> Date: Thu, 23 Jan 2025 15:42:27 +0800 Subject: [PATCH 7/8] fix integration test --- .../spark/connector/integration/test/util/SparkUtilIT.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/util/SparkUtilIT.java b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/util/SparkUtilIT.java index ed7d2085ffd..5c188f58001 100644 --- a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/util/SparkUtilIT.java +++ b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/util/SparkUtilIT.java @@ -77,7 +77,7 @@ protected void dropDatabaseIfExists(String database) { // However, Paimon does not support create a database with a specified location. protected void createDatabaseIfNotExists(String database, String provider) { String locationClause = - "lakehouse-paimon".equalsIgnoreCase(provider) + "lakehouse-paimon".equalsIgnoreCase(provider) || provider.startsWith("jdbc") ? "" : String.format("LOCATION '/user/hive/%s'", database); sql(String.format("CREATE DATABASE IF NOT EXISTS %s %s", database, locationClause)); From 6a7475ba2ecca92568c5453cbf4ac0ff6c062766 Mon Sep 17 00:00:00 2001 From: youze Liang <525672876@qq.com> Date: Fri, 24 Jan 2025 11:54:47 +0800 Subject: [PATCH 8/8] fix integration test --- .../integration/test/SparkEnvIT.java | 16 ++-------------- .../test/jdbc/SparkJdbcMysqlCatalogIT.java | 19 +++++++++++++++++++ 2 files changed, 21 insertions(+), 14 deletions(-) diff --git a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/SparkEnvIT.java b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/SparkEnvIT.java index 70c2af2becc..5bcdc9a2cb6 100644 --- a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/SparkEnvIT.java +++ b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/SparkEnvIT.java @@ -19,12 +19,10 @@ package org.apache.gravitino.spark.connector.integration.test; -import static org.apache.gravitino.integration.test.util.TestDatabaseName.MYSQL_CATALOG_MYSQL_IT; import static org.apache.gravitino.spark.connector.PropertiesConverter.SPARK_PROPERTY_PREFIX; import static org.apache.gravitino.spark.connector.iceberg.IcebergPropertiesConstants.ICEBERG_CATALOG_CACHE_ENABLED; import java.io.IOException; -import java.sql.SQLException; import java.util.Collections; import java.util.HashMap; import java.util.Map; @@ -58,10 +56,6 @@ public abstract class SparkEnvIT extends SparkUtilIT { protected String warehouse; protected FileSystem hdfs; protected String icebergRestServiceUri; - protected String mysqlUrl; - protected String mysqlUsername; - protected String mysqlPassword; - protected String mysqlDriver; private final String metalakeName = "test"; private SparkSession sparkSession; @@ -88,7 +82,7 @@ void startUp() throws Exception { if (lakeHouseIcebergProvider.equalsIgnoreCase(getProvider())) { initIcebergRestServiceEnv(); } - initMysqlEnv(); + initCatalogEnv(); // Start Gravitino server super.startIntegrationTest(); initHdfsFileSystem(); @@ -158,13 +152,7 @@ private void initHiveEnv() { HiveContainer.HDFS_DEFAULTFS_PORT); } - private void initMysqlEnv() throws SQLException { - containerSuite.startMySQLContainer(MYSQL_CATALOG_MYSQL_IT); - mysqlUrl = containerSuite.getMySQLContainer().getJdbcUrl(); - mysqlUsername = containerSuite.getMySQLContainer().getUsername(); - mysqlPassword = containerSuite.getMySQLContainer().getPassword(); - mysqlDriver = containerSuite.getMySQLContainer().getDriverClassName(MYSQL_CATALOG_MYSQL_IT); - } + protected void initCatalogEnv() throws Exception {} private void initIcebergRestServiceEnv() { ignoreIcebergRestService = false; diff --git a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/jdbc/SparkJdbcMysqlCatalogIT.java b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/jdbc/SparkJdbcMysqlCatalogIT.java index 65ed937be3a..1b77047fa8e 100644 --- a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/jdbc/SparkJdbcMysqlCatalogIT.java +++ b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/jdbc/SparkJdbcMysqlCatalogIT.java @@ -18,8 +18,11 @@ */ package org.apache.gravitino.spark.connector.integration.test.jdbc; +import static org.apache.gravitino.integration.test.util.TestDatabaseName.MYSQL_CATALOG_MYSQL_IT; + import com.google.common.collect.Maps; import java.util.Map; +import org.apache.gravitino.integration.test.container.ContainerSuite; import org.apache.gravitino.spark.connector.integration.test.SparkCommonIT; import org.apache.gravitino.spark.connector.integration.test.util.SparkTableInfoChecker; import org.apache.gravitino.spark.connector.jdbc.JdbcPropertiesConstants; @@ -27,6 +30,12 @@ @Tag("gravitino-docker-test") public abstract class SparkJdbcMysqlCatalogIT extends SparkCommonIT { + + protected String mysqlUrl; + protected String mysqlUsername; + protected String mysqlPassword; + protected String mysqlDriver; + @Override protected boolean supportsSparkSQLClusteredBy() { return false; @@ -77,6 +86,16 @@ protected SparkTableInfoChecker getTableInfoChecker() { return SparkJdbcTableInfoChecker.create(); } + @Override + protected void initCatalogEnv() throws Exception { + ContainerSuite containerSuite = ContainerSuite.getInstance(); + containerSuite.startMySQLContainer(MYSQL_CATALOG_MYSQL_IT); + mysqlUrl = containerSuite.getMySQLContainer().getJdbcUrl(); + mysqlUsername = containerSuite.getMySQLContainer().getUsername(); + mysqlPassword = containerSuite.getMySQLContainer().getPassword(); + mysqlDriver = containerSuite.getMySQLContainer().getDriverClassName(MYSQL_CATALOG_MYSQL_IT); + } + @Override protected Map getCatalogConfigs() { Map catalogProperties = Maps.newHashMap();