diff --git a/docs/spark-connector/spark-catalog-jdbc.md b/docs/spark-connector/spark-catalog-jdbc.md new file mode 100644 index 00000000000..7805d80266f --- /dev/null +++ b/docs/spark-connector/spark-catalog-jdbc.md @@ -0,0 +1,72 @@ +--- +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 + +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` +- `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: + +- `UPDATE` +- `DELETE` +- `TRUNCATE` + +## SQL example + +```sql +-- Suppose mysql_a is the mysql catalog name managed by Gravitino +USE mysql_a; + +CREATE DATABASE IF NOT EXISTS mydatabase; +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. + 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/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..3f36b9a2a84 --- /dev/null +++ b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/jdbc/GravitinoJdbcCatalog.java @@ -0,0 +1,108 @@ +/* + * 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.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.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; + +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 new SparkJdbcTable( + identifier, + gravitinoTable, + (JDBCTable) sparkTable, + (JDBCTableCatalog) sparkCatalog, + propertiesConverter, + sparkTransformConverter, + sparkTypeConverter); + } + + @Override + protected PropertiesConverter getPropertiesConverter() { + return JdbcPropertiesConverter.getInstance(); + } + + @Override + protected SparkTransformConverter getSparkTransformConverter() { + return new SparkTransformConverter(false); + } + + @Override + protected SparkTypeConverter getSparkTypeConverter() { + return new SparkJdbcTypeConverter(); + } + + @Override + public void createNamespace(String[] namespace, Map metadata) + throws NamespaceAlreadyExistsException { + 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/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..7516646e343 --- /dev/null +++ b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/jdbc/JdbcPropertiesConverter.java @@ -0,0 +1,73 @@ +/* + * 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 com.google.common.collect.ImmutableMap; +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; + } + + 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<>(); + properties.forEach( + (key, value) -> { + if (GRAVITINO_CONFIG_TO_JDBC.containsKey(key)) { + jdbcProperties.put(GRAVITINO_CONFIG_TO_JDBC.get(key), value); + } + }); + 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/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..56e2734a7f4 --- /dev/null +++ b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/jdbc/SparkJdbcTypeConverter.java @@ -0,0 +1,40 @@ +/* + * 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 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 { + 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 9392feac2f1..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,11 +46,23 @@ 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 jdbcCatalogNames.get(String.format("%d.%d", majorVersion, minorVersion)); + } 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/integration/test/SparkCommonIT.java b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/SparkCommonIT.java index c7517a3bf82..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,6 +119,14 @@ private static String getRowLevelDeleteTableSql( protected abstract boolean supportsReplaceColumns(); + protected abstract boolean supportsSchemaAndTableProperties(); + + 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("supportsSchemaAndTableProperties") protected void testCreateAndLoadSchema() { String testDatabaseName = "t_create1"; dropDatabaseIfExists(testDatabaseName); @@ -218,6 +227,7 @@ protected void testCreateAndLoadSchema() { } @Test + @EnabledIf("supportsSchemaAndTableProperties") 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("supportsSchemaAndTableProperties") 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("supportsSchemaAndTableProperties") 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..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 @@ -82,6 +82,7 @@ void startUp() throws Exception { if (lakeHouseIcebergProvider.equalsIgnoreCase(getProvider())) { initIcebergRestServiceEnv(); } + initCatalogEnv(); // Start Gravitino server super.startIntegrationTest(); initHdfsFileSystem(); @@ -151,6 +152,8 @@ private void initHiveEnv() { HiveContainer.HDFS_DEFAULTFS_PORT); } + protected void initCatalogEnv() throws Exception {} + 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..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 @@ -84,6 +84,16 @@ protected boolean supportsReplaceColumns() { return true; } + @Override + protected boolean supportsSchemaAndTableProperties() { + 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..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 @@ -109,6 +109,16 @@ protected boolean supportsReplaceColumns() { return true; } + @Override + protected boolean supportsSchemaAndTableProperties() { + 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/SparkJdbcMysqlCatalogIT.java b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/jdbc/SparkJdbcMysqlCatalogIT.java new file mode 100644 index 00000000000..1b77047fa8e --- /dev/null +++ b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/jdbc/SparkJdbcMysqlCatalogIT.java @@ -0,0 +1,108 @@ +/* + * 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 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; +import org.junit.jupiter.api.Tag; + +@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; + } + + @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 supportsSchemaAndTableProperties() { + 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 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(); + 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..32a66923cbe --- /dev/null +++ b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/jdbc/SparkJdbcTableInfoChecker.java @@ -0,0 +1,55 @@ +/* + * 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(); + } + + // Spark jdbc table cannot distinguish between comment=null and comment="" + @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..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 @@ -63,6 +63,16 @@ protected boolean supportsSchemaEvolution() { return true; } + @Override + protected boolean supportsSchemaAndTableProperties() { + 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/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)); 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); + } +} diff --git a/spark-connector/v3.3/spark/build.gradle.kts b/spark-connector/v3.3/spark/build.gradle.kts index 66c65f863b9..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("*") } @@ -163,6 +166,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.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/SparkJdbcMysqlCatalogIT33.java b/spark-connector/v3.3/spark/src/test/java/org/apache/gravitino/spark/connector/integration/test/jdbc/SparkJdbcMysqlCatalogIT33.java new file mode 100644 index 00000000000..cf190cfd4fb --- /dev/null +++ b/spark-connector/v3.3/spark/src/test/java/org/apache/gravitino/spark/connector/integration/test/jdbc/SparkJdbcMysqlCatalogIT33.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 SparkJdbcMysqlCatalogIT33 extends SparkJdbcMysqlCatalogIT { + @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/build.gradle.kts b/spark-connector/v3.4/spark/build.gradle.kts index aa4134a3c71..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("*") } @@ -163,6 +166,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/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..bbd32e0225d --- /dev/null +++ b/spark-connector/v3.4/spark/src/main/java/org/apache/gravitino/spark/connector/jdbc/SparkJdbcTypeConverter34.java @@ -0,0 +1,39 @@ +/* + * 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 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 { + return super.toSparkType(gravitinoType); + } + } +} diff --git a/spark-connector/v3.4/spark/src/test/java/org/apache/gravitino/spark/connector/integration/test/jdbc/SparkJdbcMysqlCatalogIT34.java b/spark-connector/v3.4/spark/src/test/java/org/apache/gravitino/spark/connector/integration/test/jdbc/SparkJdbcMysqlCatalogIT34.java new file mode 100644 index 00000000000..9a4038404d8 --- /dev/null +++ b/spark-connector/v3.4/spark/src/test/java/org/apache/gravitino/spark/connector/integration/test/jdbc/SparkJdbcMysqlCatalogIT34.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 SparkJdbcMysqlCatalogIT34 extends SparkJdbcMysqlCatalogIT { + @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/build.gradle.kts b/spark-connector/v3.5/spark/build.gradle.kts index 15aa018081d..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("*") } @@ -165,6 +168,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/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..1b10d63fa09 --- /dev/null +++ b/spark-connector/v3.5/spark/src/main/java/org/apache/gravitino/spark/connector/jdbc/GravitinoJdbcCatalogSpark35.java @@ -0,0 +1,39 @@ +/* + * 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; +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/SparkJdbcMysqlCatalogIT35.java b/spark-connector/v3.5/spark/src/test/java/org/apache/gravitino/spark/connector/integration/test/jdbc/SparkJdbcMysqlCatalogIT35.java new file mode 100644 index 00000000000..00c14e40d2c --- /dev/null +++ b/spark-connector/v3.5/spark/src/test/java/org/apache/gravitino/spark/connector/integration/test/jdbc/SparkJdbcMysqlCatalogIT35.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 SparkJdbcMysqlCatalogIT35 extends SparkJdbcMysqlCatalogIT { + @Test + void testCatalogClassName() { + String catalogClass = + getSparkSession() + .sessionState() + .conf() + .getConfString("spark.sql.catalog." + getCatalogName()); + Assertions.assertEquals(GravitinoJdbcCatalogSpark35.class.getName(), catalogClass); + } +}