This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 72e57d2e9 [core][spark] Support format table create, and implement
format table for SparkCatalog (#4223)
72e57d2e9 is described below
commit 72e57d2e957b76d26870fc7cc9d94aadd71ff35b
Author: Daoyuan Wang <[email protected]>
AuthorDate: Wed Oct 9 10:54:48 2024 +0800
[core][spark] Support format table create, and implement format table for
SparkCatalog (#4223)
---
.../org/apache/paimon/catalog/AbstractCatalog.java | 17 +++-
.../java/org/apache/paimon/hive/HiveCatalog.java | 110 ++++++++++++++++++---
.../apache/paimon/hive/HiveFormatTableUtils.java | 33 ++++---
.../java/org/apache/paimon/spark/SparkCatalog.java | 110 +++++++++++++++++----
.../org/apache/paimon/spark/SparkSource.scala | 2 +
.../paimon/spark/SparkCatalogWithHiveTest.java | 110 +++++++++++++++++++++
.../org/apache/paimon/spark/sql/DDLTestBase.scala | 4 +-
7 files changed, 338 insertions(+), 48 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java
b/paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java
index dab950828..38212c6ab 100644
--- a/paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java
+++ b/paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java
@@ -244,7 +244,11 @@ public abstract class AbstractCatalog implements Catalog {
copyTableDefaultOptions(schema.options());
- createTableImpl(identifier, schema);
+ if (schema.options().getOrDefault("type",
"table").equalsIgnoreCase("format-table")) {
+ createFormatTable(identifier, schema);
+ } else {
+ createTableImpl(identifier, schema);
+ }
}
protected abstract void createTableImpl(Identifier identifier, Schema
schema);
@@ -371,6 +375,17 @@ public abstract class AbstractCatalog implements Catalog {
throw new Catalog.TableNotExistException(identifier);
}
+ /**
+ * Create a {@link FormatTable} identified by the given {@link Identifier}.
+ *
+ * @param identifier Path of the table
+ * @param schema Schema of the table
+ */
+ public void createFormatTable(Identifier identifier, Schema schema) {
+ throw new UnsupportedOperationException(
+ this.getClass().getName() + " currently does not support
format table");
+ }
+
/**
* Get warehouse path for specified database. If a catalog would like to
provide individual path
* for each database, this method can be `Override` in that catalog.
diff --git
a/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveCatalog.java
b/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveCatalog.java
index 2fe2cbf26..cdc6dd42b 100644
---
a/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveCatalog.java
+++
b/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveCatalog.java
@@ -45,6 +45,7 @@ import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.FormatTable;
import org.apache.paimon.types.DataField;
import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
import org.apache.flink.table.hive.LegacyHiveClasses;
import org.apache.hadoop.conf.Configuration;
@@ -491,6 +492,35 @@ public class HiveCatalog extends AbstractCatalog {
}
}
+ public void createFormatTable(Identifier identifier, Schema schema) {
+ if (!formatTableEnabled()) {
+ throw new UnsupportedOperationException(
+ "Format table is not enabled for " + identifier);
+ }
+ List<DataField> fields = schema.fields();
+ List<String> partitionKeys = schema.partitionKeys();
+ List<String> primaryKeys = schema.primaryKeys();
+ Map<String, String> options = schema.options();
+ int highestFieldId = RowType.currentHighestFieldId(fields);
+
+ TableSchema newSchema =
+ new TableSchema(
+ 0,
+ fields,
+ highestFieldId,
+ partitionKeys,
+ primaryKeys,
+ options,
+ schema.comment());
+ try {
+ Table hiveTable = createHiveTable(identifier, newSchema);
+ clients.execute(client -> client.createTable(hiveTable));
+ } catch (Exception e) {
+ // we don't need to delete directories since HMS will roll back db
and fs if failed.
+ throw new RuntimeException("Failed to create table " +
identifier.getFullName(), e);
+ }
+ }
+
private boolean usingExternalTable() {
CatalogTableType tableType =
OptionsUtils.convertToEnum(
@@ -567,7 +597,11 @@ public class HiveCatalog extends AbstractCatalog {
private Table createHiveTable(Identifier identifier, TableSchema
tableSchema) {
Map<String, String> tblProperties;
- if (syncAllProperties()) {
+ String provider = "paimon";
+ if (tableSchema.options().getOrDefault("type",
"table").equalsIgnoreCase("format-table")) {
+ provider = tableSchema.options().get("file.format");
+ }
+ if (syncAllProperties() || !provider.equals("paimon")) {
tblProperties = new HashMap<>(tableSchema.options());
// add primary-key, partition-key to tblproperties
@@ -576,8 +610,8 @@ public class HiveCatalog extends AbstractCatalog {
tblProperties =
convertToPropertiesPrefixKey(tableSchema.options(), HIVE_PREFIX);
}
- Table table = newHmsTable(identifier, tblProperties);
- updateHmsTable(table, identifier, tableSchema);
+ Table table = newHmsTable(identifier, tblProperties, provider);
+ updateHmsTable(table, identifier, tableSchema, provider);
return table;
}
@@ -649,7 +683,7 @@ public class HiveCatalog extends AbstractCatalog {
private void alterTableToHms(Table table, Identifier identifier,
TableSchema newSchema)
throws TException, InterruptedException {
updateHmsTablePars(table, newSchema);
- updateHmsTable(table, identifier, newSchema);
+ updateHmsTable(table, identifier, newSchema,
newSchema.options().get("provider"));
clients.execute(
client ->
client.alter_table(
@@ -767,12 +801,16 @@ public class HiveCatalog extends AbstractCatalog {
return warehouse;
}
- private Table newHmsTable(Identifier identifier, Map<String, String>
tableParameters) {
+ private Table newHmsTable(
+ Identifier identifier, Map<String, String> tableParameters, String
provider) {
long currentTimeMillis = System.currentTimeMillis();
CatalogTableType tableType =
OptionsUtils.convertToEnum(
hiveConf.get(TABLE_TYPE.key(),
CatalogTableType.MANAGED.toString()),
CatalogTableType.class);
+ if (provider == null) {
+ provider = "paimon";
+ }
Table table =
new Table(
identifier.getTableName(),
@@ -788,24 +826,72 @@ public class HiveCatalog extends AbstractCatalog {
null,
null,
tableType.toString().toUpperCase(Locale.ROOT) +
"_TABLE");
- table.getParameters().put(TABLE_TYPE_PROP,
PAIMON_TABLE_TYPE_VALUE.toUpperCase());
- table.getParameters()
- .put(hive_metastoreConstants.META_TABLE_STORAGE,
STORAGE_HANDLER_CLASS_NAME);
+ table.getParameters().put(TABLE_TYPE_PROP, provider.toUpperCase());
+ if ("paimon".equalsIgnoreCase(provider)) {
+ table.getParameters()
+ .put(hive_metastoreConstants.META_TABLE_STORAGE,
STORAGE_HANDLER_CLASS_NAME);
+ } else {
+ table.getParameters().put("file.format", provider.toLowerCase());
+ table.getParameters().put("type", "format-table");
+ }
if (CatalogTableType.EXTERNAL.equals(tableType)) {
table.getParameters().put("EXTERNAL", "TRUE");
}
return table;
}
- private void updateHmsTable(Table table, Identifier identifier,
TableSchema schema) {
+ private String getSerdeClassName(String provider) {
+ if (provider == null || provider.equalsIgnoreCase("paimon")) {
+ return SERDE_CLASS_NAME;
+ } else if (provider.equalsIgnoreCase("csv")) {
+ return "org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe";
+ } else if (provider.equalsIgnoreCase("parquet")) {
+ return
"org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe";
+ } else if (provider.equalsIgnoreCase("orc")) {
+ return "org.apache.hadoop.hive.ql.io.orc.OrcSerde";
+ } else {
+ return SERDE_CLASS_NAME;
+ }
+ }
+
+ private String getInputFormatName(String provider) {
+ if (provider == null || provider.equalsIgnoreCase("paimon")) {
+ return INPUT_FORMAT_CLASS_NAME;
+ } else if (provider.equalsIgnoreCase("csv")) {
+ return "org.apache.hadoop.mapred.TextInputFormat";
+ } else if (provider.equalsIgnoreCase("parquet")) {
+ return
"org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat";
+ } else if (provider.equalsIgnoreCase("orc")) {
+ return "org.apache.hadoop.hive.ql.io.orc.OrcInputFormat";
+ } else {
+ return INPUT_FORMAT_CLASS_NAME;
+ }
+ }
+
+ private String getOutputFormatClassName(String provider) {
+ if (provider == null || provider.equalsIgnoreCase("paimon")) {
+ return OUTPUT_FORMAT_CLASS_NAME;
+ } else if (provider.equalsIgnoreCase("csv")) {
+ return
"org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat";
+ } else if (provider.equalsIgnoreCase("parquet")) {
+ return
"org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat";
+ } else if (provider.equalsIgnoreCase("orc")) {
+ return "org.apache.hadoop.hive.ql.io.orc.OrcOutputFormat";
+ } else {
+ return OUTPUT_FORMAT_CLASS_NAME;
+ }
+ }
+
+ private void updateHmsTable(
+ Table table, Identifier identifier, TableSchema schema, String
provider) {
StorageDescriptor sd = table.getSd() != null ? table.getSd() : new
StorageDescriptor();
- sd.setInputFormat(INPUT_FORMAT_CLASS_NAME);
- sd.setOutputFormat(OUTPUT_FORMAT_CLASS_NAME);
+ sd.setInputFormat(getInputFormatName(provider));
+ sd.setOutputFormat(getOutputFormatClassName(provider));
SerDeInfo serDeInfo = sd.getSerdeInfo() != null ? sd.getSerdeInfo() :
new SerDeInfo();
serDeInfo.setParameters(new HashMap<>());
- serDeInfo.setSerializationLib(SERDE_CLASS_NAME);
+ serDeInfo.setSerializationLib(getSerdeClassName(provider));
sd.setSerdeInfo(serDeInfo);
CoreOptions options = new CoreOptions(schema.options());
diff --git
a/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveFormatTableUtils.java
b/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveFormatTableUtils.java
index dcb7c7ee7..4866bd428 100644
---
a/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveFormatTableUtils.java
+++
b/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveFormatTableUtils.java
@@ -54,21 +54,26 @@ class HiveFormatTableUtils {
String comment = options.remove(COMMENT_PROP);
String location = hiveTable.getSd().getLocation();
Format format;
- SerDeInfo serdeInfo = hiveTable.getSd().getSerdeInfo();
- String serLib = serdeInfo.getSerializationLib().toLowerCase();
- String inputFormat = hiveTable.getSd().getInputFormat();
- if (serLib.contains("parquet")) {
- format = Format.PARQUET;
- } else if (serLib.contains("orc")) {
- format = Format.ORC;
- } else if (inputFormat.contains("Text")) {
- format = Format.CSV;
- // hive default field delimiter is '\u0001'
- options.put(
- FIELD_DELIMITER.key(),
- serdeInfo.getParameters().getOrDefault(FIELD_DELIM,
"\u0001"));
+ if (options.getOrDefault("type", "table").equals("format-table")) {
+ format = Format.valueOf(options.get("file.format").toUpperCase());
+ // field delimiter for csv leaves untouched
} else {
- throw new UnsupportedOperationException("Unsupported table: " +
hiveTable);
+ SerDeInfo serdeInfo = hiveTable.getSd().getSerdeInfo();
+ String serLib = serdeInfo.getSerializationLib().toLowerCase();
+ String inputFormat = hiveTable.getSd().getInputFormat();
+ if (serLib.contains("parquet")) {
+ format = Format.PARQUET;
+ } else if (serLib.contains("orc")) {
+ format = Format.ORC;
+ } else if (inputFormat.contains("Text")) {
+ format = Format.CSV;
+ // hive default field delimiter is '\u0001'
+ options.put(
+ FIELD_DELIMITER.key(),
+ serdeInfo.getParameters().getOrDefault(FIELD_DELIM,
"\u0001"));
+ } else {
+ throw new UnsupportedOperationException("Unsupported table: "
+ hiveTable);
+ }
}
return FormatTable.builder()
.identifier(identifier)
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkCatalog.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkCatalog.java
index 9e05ee649..04a9c6aba 100644
---
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkCatalog.java
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkCatalog.java
@@ -27,6 +27,7 @@ import org.apache.paimon.schema.Schema;
import org.apache.paimon.schema.SchemaChange;
import org.apache.paimon.spark.catalog.SparkBaseCatalog;
import org.apache.paimon.spark.catalog.SupportFunction;
+import org.apache.paimon.table.FormatTable;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.catalyst.analysis.NamespaceAlreadyExistsException;
@@ -41,6 +42,12 @@ import
org.apache.spark.sql.connector.expressions.FieldReference;
import org.apache.spark.sql.connector.expressions.IdentityTransform;
import org.apache.spark.sql.connector.expressions.NamedReference;
import org.apache.spark.sql.connector.expressions.Transform;
+import org.apache.spark.sql.execution.datasources.csv.CSVFileFormat;
+import org.apache.spark.sql.execution.datasources.orc.OrcFileFormat;
+import org.apache.spark.sql.execution.datasources.parquet.ParquetFileFormat;
+import org.apache.spark.sql.execution.datasources.v2.csv.CSVTable;
+import org.apache.spark.sql.execution.datasources.v2.orc.OrcTable;
+import org.apache.spark.sql.execution.datasources.v2.parquet.ParquetTable;
import org.apache.spark.sql.internal.SessionState;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;
@@ -230,7 +237,8 @@ public class SparkCatalog extends SparkBaseCatalog
implements SupportFunction {
}
@Override
- public SparkTable loadTable(Identifier ident) throws NoSuchTableException {
+ public org.apache.spark.sql.connector.catalog.Table loadTable(Identifier
ident)
+ throws NoSuchTableException {
return loadSparkTable(ident, Collections.emptyMap());
}
@@ -239,8 +247,14 @@ public class SparkCatalog extends SparkBaseCatalog
implements SupportFunction {
*/
public SparkTable loadTable(Identifier ident, String version) throws
NoSuchTableException {
LOG.info("Time travel to version '{}'.", version);
- return loadSparkTable(
- ident,
Collections.singletonMap(CoreOptions.SCAN_VERSION.key(), version));
+ org.apache.spark.sql.connector.catalog.Table table =
+ loadSparkTable(
+ ident,
Collections.singletonMap(CoreOptions.SCAN_VERSION.key(), version));
+ if (table instanceof SparkTable) {
+ return (SparkTable) table;
+ } else {
+ throw new NoSuchTableException(ident);
+ }
}
/**
@@ -253,10 +267,17 @@ public class SparkCatalog extends SparkBaseCatalog
implements SupportFunction {
// Paimon's timestamp use millisecond
timestamp = timestamp / 1000;
LOG.info("Time travel target timestamp is {} milliseconds.",
timestamp);
- return loadSparkTable(
- ident,
- Collections.singletonMap(
- CoreOptions.SCAN_TIMESTAMP_MILLIS.key(),
String.valueOf(timestamp)));
+ org.apache.spark.sql.connector.catalog.Table table =
+ loadSparkTable(
+ ident,
+ Collections.singletonMap(
+ CoreOptions.SCAN_TIMESTAMP_MILLIS.key(),
+ String.valueOf(timestamp)));
+ if (table instanceof SparkTable) {
+ return (SparkTable) table;
+ } else {
+ throw new NoSuchTableException(ident);
+ }
}
@Override
@@ -284,7 +305,7 @@ public class SparkCatalog extends SparkBaseCatalog
implements SupportFunction {
}
@Override
- public SparkTable createTable(
+ public org.apache.spark.sql.connector.catalog.Table createTable(
Identifier ident,
StructType schema,
Transform[] partitions,
@@ -292,12 +313,25 @@ public class SparkCatalog extends SparkBaseCatalog
implements SupportFunction {
throws TableAlreadyExistsException, NoSuchNamespaceException {
try {
String provider = properties.get(TableCatalog.PROP_PROVIDER);
- checkArgument(
- usePaimon(provider),
- "SparkCatalog can only create paimon table, but current
provider is %s",
- provider);
- catalog.createTable(
- toIdentifier(ident), toInitialSchema(schema, partitions,
properties), false);
+ if ((!usePaimon(provider))
+ &&
SparkSource.FORMAT_NAMES().contains(provider.toLowerCase())) {
+ Map<String, String> newProperties = new HashMap<String,
String>(properties);
+ newProperties.put("type", "format-table");
+ newProperties.put("file.format", provider.toLowerCase());
+ catalog.createTable(
+ toIdentifier(ident),
+ toInitialSchema(schema, partitions, newProperties),
+ false);
+ } else {
+ checkArgument(
+ usePaimon(provider),
+ "SparkCatalog can only create paimon table, but
current provider is %s",
+ provider);
+ catalog.createTable(
+ toIdentifier(ident),
+ toInitialSchema(schema, partitions, properties),
+ false);
+ }
return loadTable(ident);
} catch (Catalog.TableAlreadyExistException e) {
throw new TableAlreadyExistsException(ident);
@@ -458,11 +492,51 @@ public class SparkCatalog extends SparkBaseCatalog
implements SupportFunction {
return new org.apache.paimon.catalog.Identifier(ident.namespace()[0],
ident.name());
}
- protected SparkTable loadSparkTable(Identifier ident, Map<String, String>
extraOptions)
- throws NoSuchTableException {
+ protected org.apache.spark.sql.connector.catalog.Table loadSparkTable(
+ Identifier ident, Map<String, String> extraOptions) throws
NoSuchTableException {
try {
- return new SparkTable(
- copyWithSQLConf(catalog.getTable(toIdentifier(ident)),
extraOptions));
+ org.apache.paimon.table.Table paimonTable =
catalog.getTable(toIdentifier(ident));
+ if (paimonTable instanceof FormatTable) {
+ FormatTable formatTable = (FormatTable) paimonTable;
+ StructType schema =
SparkTypeUtils.fromPaimonRowType(formatTable.rowType());
+ List<String> pathList = new ArrayList<>();
+ pathList.add(formatTable.location());
+ CaseInsensitiveStringMap dsOptions =
+ new CaseInsensitiveStringMap(formatTable.options());
+ if (formatTable.format() == FormatTable.Format.CSV) {
+ return new CSVTable(
+ ident.name(),
+ SparkSession.active(),
+ dsOptions,
+
scala.collection.JavaConverters.asScalaBuffer(pathList).toSeq(),
+ scala.Option.apply(schema),
+ CSVFileFormat.class);
+ } else if (formatTable.format() == FormatTable.Format.ORC) {
+ return new OrcTable(
+ ident.name(),
+ SparkSession.active(),
+ dsOptions,
+
scala.collection.JavaConverters.asScalaBuffer(pathList).toSeq(),
+ scala.Option.apply(schema),
+ OrcFileFormat.class);
+ } else if (formatTable.format() == FormatTable.Format.PARQUET)
{
+ return new ParquetTable(
+ ident.name(),
+ SparkSession.active(),
+ dsOptions,
+
scala.collection.JavaConverters.asScalaBuffer(pathList).toSeq(),
+ scala.Option.apply(schema),
+ ParquetFileFormat.class);
+ } else {
+ throw new UnsupportedOperationException(
+ "Unsupported format table "
+ + ident.name()
+ + " format "
+ + formatTable.format().name());
+ }
+ } else {
+ return new SparkTable(copyWithSQLConf(paimonTable,
extraOptions));
+ }
} catch (Catalog.TableNotExistException e) {
throw new NoSuchTableException(ident);
}
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/SparkSource.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/SparkSource.scala
index 8ea2c31bc..67ab1312f 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/SparkSource.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/SparkSource.scala
@@ -110,6 +110,8 @@ object SparkSource {
val NAME = "paimon"
+ val FORMAT_NAMES = Seq("csv", "orc", "parquet")
+
def toBaseRelation(table: FileStoreTable, _sqlContext: SQLContext):
BaseRelation = {
new BaseRelation {
override def sqlContext: SQLContext = _sqlContext
diff --git
a/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/SparkCatalogWithHiveTest.java
b/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/SparkCatalogWithHiveTest.java
new file mode 100644
index 000000000..22811af39
--- /dev/null
+++
b/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/SparkCatalogWithHiveTest.java
@@ -0,0 +1,110 @@
+/*
+ * 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.paimon.spark;
+
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.hive.TestHiveMetastore;
+
+import org.apache.spark.sql.SparkSession;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Base tests for spark read. */
+public class SparkCatalogWithHiveTest {
+ private static TestHiveMetastore testHiveMetastore;
+
+ private static final int PORT = 9087;
+
+ @BeforeAll
+ public static void startMetastore() {
+ testHiveMetastore = new TestHiveMetastore();
+ testHiveMetastore.start(PORT);
+ }
+
+ @AfterAll
+ public static void closeMetastore() throws Exception {
+ testHiveMetastore.stop();
+ }
+
+ @Test
+ public void testCreateFormatTable(@TempDir java.nio.file.Path tempDir) {
+ // firstly, we use hive metastore to creata table, and check the
result.
+ Path warehousePath = new Path("file:" + tempDir.toString());
+ SparkSession spark =
+ SparkSession.builder()
+ .config("spark.sql.warehouse.dir",
warehousePath.toString())
+ // with hive metastore
+ .config("spark.sql.catalogImplementation", "hive")
+ .config("hive.metastore.uris", "thrift://localhost:" +
PORT)
+ .config("spark.sql.catalog.spark_catalog",
SparkCatalog.class.getName())
+ .config("spark.sql.catalog.spark_catalog.metastore",
"hive")
+ .config(
+
"spark.sql.catalog.spark_catalog.hive.metastore.uris",
+ "thrift://localhost:" + PORT)
+
.config("spark.sql.catalog.spark_catalog.format-table.enabled", "true")
+ .config(
+ "spark.sql.catalog.spark_catalog.warehouse",
+ warehousePath.toString())
+ .master("local[2]")
+ .getOrCreate();
+
+ spark.sql("CREATE DATABASE IF NOT EXISTS my_db1");
+ spark.sql("USE spark_catalog.my_db1");
+ spark.sql("CREATE TABLE IF NOT EXISTS table_orc (a INT, bb INT, c
STRING) USING orc");
+
+ assertThat(
+ spark.sql("SHOW TABLES").collectAsList().stream()
+ .map(s -> s.get(1))
+ .map(Object::toString))
+ .containsExactlyInAnyOrder("table_orc");
+ spark.close();
+
+ SparkSession spark1 =
+ SparkSession.builder()
+ .config("spark.sql.warehouse.dir",
warehousePath.toString())
+ // with hive metastore
+ .config("spark.sql.catalogImplementation", "hive")
+ .config("hive.metastore.uris", "thrift://localhost:" +
PORT)
+ .config("spark.sql.catalog.spark_catalog",
SparkCatalog.class.getName())
+ .config("spark.sql.catalog.spark_catalog.metastore",
"hive")
+ .config(
+
"spark.sql.catalog.spark_catalog.hive.metastore.uris",
+ "thrift://localhost:" + PORT)
+
.config("spark.sql.catalog.spark_catalog.format-table.enabled", "true")
+ .config(
+ "spark.sql.catalog.spark_catalog.warehouse",
+ warehousePath.toString())
+ .master("local[2]")
+ .getOrCreate();
+ spark1.sql("USE spark_catalog.my_db1");
+ assertThat(
+ spark1.sql("EXPLAIN EXTENDED SELECT * from
table_orc").collectAsList()
+ .stream()
+ .map(s -> s.get(0))
+ .map(Object::toString)
+ .filter(s -> s.contains("OrcScan"))
+ .count())
+ .isGreaterThan(0);
+ spark1.close();
+ }
+}
diff --git
a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/DDLTestBase.scala
b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/DDLTestBase.scala
index f99bff2af..f1215ede4 100644
---
a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/DDLTestBase.scala
+++
b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/DDLTestBase.scala
@@ -154,9 +154,7 @@ abstract class DDLTestBase extends PaimonSparkTestBase {
val error = intercept[Exception] {
spark.sql(s"CREATE TABLE parquet_tbl (id int) USING parquet")
}.getMessage
- assert(
- error.contains(
- "SparkCatalog can only create paimon table, but current provider is
parquet"))
+ assert(error.contains("does not support format table"))
}
}
