This is an automated email from the ASF dual-hosted git repository.
JNSimba pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris-spark-connector.git
The following commit(s) were added to refs/heads/master by this push:
new 223240b [Feature] Support reading external catalog tables (#383)
223240b is described below
commit 223240b5e88290fec89f4a77e1cdfcc17321b464
Author: wudi <[email protected]>
AuthorDate: Thu Nov 5 09:57:08 2026 +0800
[Feature] Support reading external catalog tables (#383)
---
README.md | 25 ++++
spark-doris-connector/pom.xml | 2 +-
.../doris/spark/client/DorisFrontendClient.java | 6 +-
.../spark/client/read/DorisFlightSqlReader.java | 5 +-
.../spark/client/read/DorisReadModeResolver.java | 10 ++
.../client/read/ReaderPartitionGenerator.java | 46 ++++---
.../org/apache/doris/spark/config/DorisConfig.java | 14 +++
.../apache/doris/spark/rdd/AbstractDorisRDD.scala | 2 +-
.../doris/spark/sql/sources/DorisRelation.scala | 4 +-
.../scala/org/apache/doris/spark/util/URLs.scala | 3 +
.../client/read/DorisReadModeResolverTest.java | 26 ++++
.../spark/client/read/ExternalCatalogReadTest.java | 108 +++++++++++++++++
.../client/read/ReaderPartitionGeneratorTest.java | 11 +-
.../org/apache/doris/spark/util/URLsTest.scala | 8 ++
.../container/AbstractMySqlCatalogTestBase.java | 135 +++++++++++++++++++++
.../spark/container/AbstractS3TvfTestBase.java | 2 +-
.../spark/container/instance/DorisContainer.java | 4 +
.../spark/sql/DorisExternalCatalogITCase.scala | 88 ++++++++++++++
.../doris/spark/sql/DorisSourceProvider.scala | 2 +
.../doris/spark/catalog/DorisTableBase.scala | 6 +-
.../spark/catalog/DorisTableProviderBase.scala | 12 +-
.../doris/spark/catalog/DorisTableBaseTest.scala | 55 +++++++++
.../doris/spark/catalog/DorisTableBase.scala | 6 +-
.../spark/catalog/DorisTableProviderBase.scala | 12 +-
.../doris/spark/catalog/DorisTableBaseTest.scala | 55 +++++++++
25 files changed, 593 insertions(+), 54 deletions(-)
diff --git a/README.md b/README.md
index 9241ca1..f559130 100644
--- a/README.md
+++ b/README.md
@@ -133,6 +133,31 @@ dorisSparkDF = spark.read.format("doris")
dorisSparkDF.show(5)
```
+## Reading external catalog tables
+
+Use a three-part `catalog.database.table` identifier to read tables from Doris
+external catalogs such as Hive, Iceberg, and Paimon:
+
+```scala
+spark.read.format("doris")
+ .option("doris.fenodes", "127.0.0.1:8030")
+ .option("doris.table.identifier", "hive_catalog.database.table")
+ .option("doris.user", "root")
+ .option("doris.password", "")
+ .load()
+```
+
+Arrow Flight SQL must be enabled on Doris. The connector discovers its port;
+when frontend discovery is disabled, set `doris.read.arrow-flight-sql.port`
+explicitly. Three-part identifiers require Arrow mode and do not fall back to
+Thrift. The same identifier works with SQL `USING doris` and the RDD API.
+
+External reads use one Spark partition. Doris plans and executes the external
+scan. Column projection and filter pushdown are retained. Complex types
+use the existing connector conversions: ARRAY as a string (or an array of
+strings with `doris.read.array.native-type=true`), MAP as a map of strings, and
+STRUCT as a string. Writing through a three-part identifier is not supported.
+
## TLS
The connector supports one-way TLS for Doris HTTP APIs and Stream Load,
diff --git a/spark-doris-connector/pom.xml b/spark-doris-connector/pom.xml
index fcb5ce2..bdf2137 100644
--- a/spark-doris-connector/pom.xml
+++ b/spark-doris-connector/pom.xml
@@ -96,7 +96,7 @@
<netty.version>4.1.110.Final</netty.version>
<fasterxml.jackson.version>2.13.5</fasterxml.jackson.version>
<thrift-service.version>1.0.1</thrift-service.version>
- <testcontainers.version>1.17.6</testcontainers.version>
+ <testcontainers.version>1.21.4</testcontainers.version>
<awssdk.version>2.29.52</awssdk.version>
<jmockit.version>1.49</jmockit.version>
<fe_ut_parallel>1</fe_ut_parallel>
diff --git
a/spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/client/DorisFrontendClient.java
b/spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/client/DorisFrontendClient.java
index 3d3fa0b..05eba20 100644
---
a/spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/client/DorisFrontendClient.java
+++
b/spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/client/DorisFrontendClient.java
@@ -332,8 +332,12 @@ public class DorisFrontendClient implements Serializable {
}
public Schema getTableSchema(String db, String table) throws Exception {
+ return getTableSchema(new String[]{db, table});
+ }
+
+ public Schema getTableSchema(String[] tableIdentifier) throws Exception {
return requestFrontends((frontend, httpClient) -> {
- String url = URLs.tableSchema(frontend.getHost(),
frontend.getHttpPort(), db, table, isHttpsEnabled);
+ String url = URLs.tableSchema(frontend.getHost(),
frontend.getHttpPort(), tableIdentifier, isHttpsEnabled);
HttpGet httpGet = new HttpGet(url);
HttpUtils.setAuth(httpGet, username, password);
Schema dorisSchema;
diff --git
a/spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/client/read/DorisFlightSqlReader.java
b/spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/client/read/DorisFlightSqlReader.java
index 71d4b88..7f05048 100644
---
a/spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/client/read/DorisFlightSqlReader.java
+++
b/spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/client/read/DorisFlightSqlReader.java
@@ -222,7 +222,8 @@ public class DorisFlightSqlReader extends DorisReader {
protected String generateQuerySql(DorisReaderPartition partition) throws
OptionRequiredException {
String columns = String.join(",", partition.getReadColumns());
String fullTableName =
config.getValue(DorisOptions.DORIS_TABLE_IDENTIFIER);
- String tablets = String.format("TABLET(%s)",
StringUtils.join(partition.getTablets(), ","));
+ String tablets = partition.getTablets().length == 0 ? ""
+ : String.format("TABLET(%s)",
StringUtils.join(partition.getTablets(), ","));
String predicates = partition.getFilters().length == 0 ? "" : " WHERE
" + String.join(" AND ", partition.getFilters());
String limit = partition.getLimit() > 0 ? " LIMIT " +
partition.getLimit() : "";
return generateQueryPrefix() + String.format("SELECT %s FROM %s
%s%s%s", columns, fullTableName, tablets, predicates, limit);
@@ -235,7 +236,7 @@ public class DorisFlightSqlReader extends DorisReader {
protected Schema processDorisSchema(DorisReaderPartition partition) throws
Exception {
Schema processedSchema = new Schema();
- Schema tableSchema =
frontendClient.getTableSchema(partition.getDatabase(), partition.getTable());
+ Schema tableSchema =
frontendClient.getTableSchema(config.getTableIdentifierParts());
Map<String, Field> fieldTypeMap = tableSchema.getProperties().stream()
.collect(Collectors.toMap(Field::getName,
Function.identity()));
String[] readColumns = partition.getReadColumns();
diff --git
a/spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/client/read/DorisReadModeResolver.java
b/spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/client/read/DorisReadModeResolver.java
index f4766d1..65fc9ab 100644
---
a/spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/client/read/DorisReadModeResolver.java
+++
b/spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/client/read/DorisReadModeResolver.java
@@ -56,7 +56,11 @@ public class DorisReadModeResolver {
static String resolve(DorisConfig config, FlightSqlPortSupplier
portSupplier)
throws OptionRequiredException {
String readMode =
config.getValue(DorisOptions.READ_MODE).toLowerCase(Locale.ROOT);
+ boolean catalogRead = config.getTableIdentifierParts().length == 3;
if (THRIFT.equals(readMode)) {
+ if (catalogRead) {
+ throw new IllegalArgumentException("Reading catalog.db.table
requires Arrow Flight SQL");
+ }
return THRIFT;
}
if (!ARROW.equals(readMode)) {
@@ -71,6 +75,9 @@ public class DorisReadModeResolver {
try {
flightSqlPort = portSupplier.get();
} catch (Exception e) {
+ if (catalogRead) {
+ throw new IllegalArgumentException("Reading catalog.db.table
requires an available Arrow Flight SQL port", e);
+ }
LOG.warn("failed to discover Arrow Flight SQL port, falling back
to Thrift", e);
return THRIFT;
}
@@ -80,6 +87,9 @@ public class DorisReadModeResolver {
return ARROW;
}
+ if (catalogRead) {
+ throw new IllegalArgumentException("Reading catalog.db.table
requires an available Arrow Flight SQL port");
+ }
LOG.warn("Arrow Flight SQL port {} is invalid or unavailable, falling
back to Thrift", flightSqlPort);
return THRIFT;
}
diff --git
a/spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/client/read/ReaderPartitionGenerator.java
b/spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/client/read/ReaderPartitionGenerator.java
index bc6c410..cdc0adb 100644
---
a/spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/client/read/ReaderPartitionGenerator.java
+++
b/spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/client/read/ReaderPartitionGenerator.java
@@ -65,25 +65,34 @@ public class ReaderPartitionGenerator {
public static DorisReaderPartition[] generatePartitions(DorisConfig config,
String[] fields,
String[] filters, Integer limit,
Boolean
datetimeJava8ApiEnabled) throws Exception {
+ String[] tableParts = config.getTableIdentifierParts();
+ String db = tableParts[tableParts.length - 2];
+ String table = tableParts[tableParts.length - 1];
DorisFrontendClient frontend = new DorisFrontendClient(config);
- String fullTableName =
config.getValue(DorisOptions.DORIS_TABLE_IDENTIFIER);
- String[] tableParts = fullTableName.split("\\.");
- String db = tableParts[0].replaceAll("`", "");
- String table = tableParts[1].replaceAll("`", "");
- String[] originReadCols = fields;
- if (fields == null || fields.length == 0) {
- originReadCols = frontend.getTableAllColumns(db, table);
+ try {
+ Schema tableSchema = frontend.getTableSchema(tableParts);
+ String[] originReadCols = fields;
+ if (fields == null || fields.length == 0) {
+ originReadCols =
tableSchema.getProperties().stream().map(Field::getName).toArray(String[]::new);
+ }
+ String[] finalReadColumns = getFinalReadColumns(config,
tableSchema, originReadCols);
+ if (tableParts.length == 3) {
+ // Doris plans catalog scans without connector-side tablet
assignment.
+ return new DorisReaderPartition[]{new DorisReaderPartition(db,
table, null,
+ new Long[0], "", finalReadColumns, filters, limit,
config, datetimeJava8ApiEnabled)};
+ }
+ String finalReadColumnString = String.join(",", finalReadColumns);
+ String finalWhereClauseString = filters.length == 0 ? "" : " WHERE
" + String.join(" AND ", filters);
+ String sql = "SELECT " + finalReadColumnString + " FROM `" + db +
"`.`" + table + "`" + finalWhereClauseString;
+ LOG.info("get query plan for table " + db + "." + table + ", sql:
" + sql);
+ QueryPlan queryPlan = frontend.getQueryPlan(db, table, sql);
+ Map<String, List<Long>> beToTablets =
mappingBeToTablets(queryPlan);
+ int maxTabletSize =
config.getValue(DorisOptions.DORIS_TABLET_SIZE);
+ return distributeTabletsToPartitions(db, table, beToTablets,
queryPlan.getOpaqued_query_plan(), maxTabletSize,
+ finalReadColumns, filters, config, limit,
datetimeJava8ApiEnabled);
+ } finally {
+ frontend.close();
}
- String[] finalReadColumns = getFinalReadColumns(config, frontend, db,
table, originReadCols);
- String finalReadColumnString = String.join(",", finalReadColumns);
- String finalWhereClauseString = filters.length == 0 ? "" : " WHERE " +
String.join(" AND ", filters);
- String sql = "SELECT " + finalReadColumnString + " FROM `" + db +
"`.`" + table + "`" + finalWhereClauseString;
- LOG.info("get query plan for table " + db + "." + table + ", sql: " +
sql);
- QueryPlan queryPlan = frontend.getQueryPlan(db, table, sql);
- Map<String, List<Long>> beToTablets = mappingBeToTablets(queryPlan);
- int maxTabletSize = config.getValue(DorisOptions.DORIS_TABLET_SIZE);
- return distributeTabletsToPartitions(db, table, beToTablets,
queryPlan.getOpaqued_query_plan(), maxTabletSize,
- finalReadColumns, filters, config, limit,
datetimeJava8ApiEnabled);
}
@VisibleForTesting
@@ -130,8 +139,7 @@ public class ReaderPartitionGenerator {
return partitions.toArray(new DorisReaderPartition[0]);
}
- protected static String[] getFinalReadColumns(DorisConfig config,
DorisFrontendClient frontendClient, String db, String table, String[]
readFields) throws Exception {
- Schema tableSchema = frontendClient.getTableSchema(db, table);
+ protected static String[] getFinalReadColumns(DorisConfig config, Schema
tableSchema, String[] readFields) throws Exception {
Map<String, String> fieldTypeMap =
tableSchema.getProperties().stream().collect(
Collectors.toMap(Field::getName, Field::getType));
Boolean bitmapToString =
config.getValue(DorisOptions.DORIS_READ_BITMAP_TO_STRING);
diff --git
a/spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/config/DorisConfig.java
b/spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/config/DorisConfig.java
index 1fa1fcc..6c22a64 100644
---
a/spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/config/DorisConfig.java
+++
b/spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/config/DorisConfig.java
@@ -40,6 +40,20 @@ public class DorisConfig implements Serializable {
checkOptions(this.configOptions);
}
+ public String[] getTableIdentifierParts() throws OptionRequiredException {
+ String[] parts =
getValue(DorisOptions.DORIS_TABLE_IDENTIFIER).split("\\.", -1);
+ if (parts.length != 2 && parts.length != 3) {
+ throw new IllegalArgumentException("doris.table.identifier must be
db.table or catalog.db.table");
+ }
+ for (int i = 0; i < parts.length; i++) {
+ parts[i] = parts[i].replace("`", "");
+ if (parts[i].trim().isEmpty()) {
+ throw new IllegalArgumentException("doris.table.identifier
contains an empty name");
+ }
+ }
+ return parts;
+ }
+
private Map<String, String> processOptions(Map<String, String> options) {
Map<String, String> dottedOptions = new HashMap<>();
for (Map.Entry<String, String> entry : options.entrySet()) {
diff --git
a/spark-doris-connector/spark-doris-connector-base/src/main/scala/org/apache/doris/spark/rdd/AbstractDorisRDD.scala
b/spark-doris-connector/spark-doris-connector-base/src/main/scala/org/apache/doris/spark/rdd/AbstractDorisRDD.scala
index 50f6474..e354da3 100644
---
a/spark-doris-connector/spark-doris-connector-base/src/main/scala/org/apache/doris/spark/rdd/AbstractDorisRDD.scala
+++
b/spark-doris-connector/spark-doris-connector-base/src/main/scala/org/apache/doris/spark/rdd/AbstractDorisRDD.scala
@@ -39,7 +39,7 @@ protected[spark] abstract class AbstractDorisRDD[T: ClassTag](
override def getPreferredLocations(split: Partition): Seq[String] = {
val dorisSplit = split.asInstanceOf[DorisPartition]
- Seq(dorisSplit.dorisPartition.getBackend.hostRpcPortString())
+
Option(dorisSplit.dorisPartition.getBackend).map(_.hostRpcPortString()).toSeq
}
override def checkpoint(): Unit = {
diff --git
a/spark-doris-connector/spark-doris-connector-base/src/main/scala/org/apache/doris/spark/sql/sources/DorisRelation.scala
b/spark-doris-connector/spark-doris-connector-base/src/main/scala/org/apache/doris/spark/sql/sources/DorisRelation.scala
index 084b821..11d311c 100644
---
a/spark-doris-connector/spark-doris-connector-base/src/main/scala/org/apache/doris/spark/sql/sources/DorisRelation.scala
+++
b/spark-doris-connector/spark-doris-connector-base/src/main/scala/org/apache/doris/spark/sql/sources/DorisRelation.scala
@@ -42,9 +42,7 @@ private[sql] class DorisRelation(
private lazy val frontend: DorisFrontendClient = new DorisFrontendClient(cfg)
private lazy val lazySchema = {
- val tableIdentifier = cfg.getValue(DorisOptions.DORIS_TABLE_IDENTIFIER)
- val tableIdentifierArr =
tableIdentifier.split("\\.").map(_.replaceAll("`", ""))
- val dorisSchema = frontend.getTableSchema(tableIdentifierArr(0),
tableIdentifierArr(1))
+ val dorisSchema = frontend.getTableSchema(cfg.getTableIdentifierParts)
val arrayNativeType =
cfg.getValue(DorisOptions.DORIS_READ_ARRAY_NATIVE_TYPE)
StructType(dorisSchema.getProperties.asScala.map(field => {
StructField(field.getName,
SchemaConvertors.toCatalystType(field.getType, field.getPrecision,
field.getScale, arrayNativeType))
diff --git
a/spark-doris-connector/spark-doris-connector-base/src/main/scala/org/apache/doris/spark/util/URLs.scala
b/spark-doris-connector/spark-doris-connector-base/src/main/scala/org/apache/doris/spark/util/URLs.scala
index b4e868a..f5545ee 100644
---
a/spark-doris-connector/spark-doris-connector-base/src/main/scala/org/apache/doris/spark/util/URLs.scala
+++
b/spark-doris-connector/spark-doris-connector-base/src/main/scala/org/apache/doris/spark/util/URLs.scala
@@ -35,6 +35,9 @@ object URLs {
def tableSchema(host: String, port: Int, database: String, table: String,
enableHttps: Boolean): String =
tableSchema(s"${assemblePath(host, port)}", database, table, enableHttps)
+ def tableSchema(host: String, port: Int, tableIdentifier: Array[String],
enableHttps: Boolean): String =
+ s"${schema(enableHttps)}://${assemblePath(host,
port)}/api/${tableIdentifier.mkString("/")}/_schema"
+
def queryPlan(feNode: String, database: String, table: String, enableHttps:
Boolean = false) =
s"${schema(enableHttps)}://$feNode/api/$database/$table/_query_plan"
diff --git
a/spark-doris-connector/spark-doris-connector-base/src/test/java/org/apache/doris/spark/client/read/DorisReadModeResolverTest.java
b/spark-doris-connector/spark-doris-connector-base/src/test/java/org/apache/doris/spark/client/read/DorisReadModeResolverTest.java
index 96d5f70..794a52a 100644
---
a/spark-doris-connector/spark-doris-connector-base/src/test/java/org/apache/doris/spark/client/read/DorisReadModeResolverTest.java
+++
b/spark-doris-connector/spark-doris-connector-base/src/test/java/org/apache/doris/spark/client/read/DorisReadModeResolverTest.java
@@ -103,6 +103,32 @@ public class DorisReadModeResolverTest {
}
}
+ @Test
+ public void catalogReadRejectsThrift() throws Exception {
+ DorisConfig config = createConfig("thrift", null);
+ config.setProperty(DorisOptions.DORIS_TABLE_IDENTIFIER, "hive.db.tbl");
+
org.junit.jupiter.api.Assertions.assertThrows(IllegalArgumentException.class,
+ () -> DorisReadModeResolver.resolve(config, () -> 9040));
+ }
+
+ @Test
+ public void catalogReadDoesNotFallBackWhenFlightPortIsUnavailable() throws
Exception {
+ DorisConfig config = createConfig("arrow", null);
+ config.setProperty(DorisOptions.DORIS_TABLE_IDENTIFIER, "hive.db.tbl");
+
org.junit.jupiter.api.Assertions.assertThrows(IllegalArgumentException.class,
+ () -> DorisReadModeResolver.resolve(config, () -> -1));
+ }
+
+ @Test
+ public void catalogReadDoesNotFallBackWhenDiscoveryFails() throws
Exception {
+ DorisConfig config = createConfig("arrow", null);
+ config.setProperty(DorisOptions.DORIS_TABLE_IDENTIFIER, "hive.db.tbl");
+
org.junit.jupiter.api.Assertions.assertThrows(IllegalArgumentException.class,
+ () -> DorisReadModeResolver.resolve(config, () -> {
+ throw new java.io.IOException("discovery failed");
+ }));
+ }
+
private DorisConfig createConfig(String readMode, Integer flightSqlPort)
throws Exception {
Map<String, String> options = new HashMap<>();
options.put("doris.fenodes", "127.0.0.1:8030");
diff --git
a/spark-doris-connector/spark-doris-connector-base/src/test/java/org/apache/doris/spark/client/read/ExternalCatalogReadTest.java
b/spark-doris-connector/spark-doris-connector-base/src/test/java/org/apache/doris/spark/client/read/ExternalCatalogReadTest.java
new file mode 100644
index 0000000..28151e4
--- /dev/null
+++
b/spark-doris-connector/spark-doris-connector-base/src/test/java/org/apache/doris/spark/client/read/ExternalCatalogReadTest.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.doris.spark.client.read;
+
+import org.apache.doris.spark.client.entity.DorisReaderPartition;
+import org.apache.doris.spark.config.DorisConfig;
+import org.junit.Assert;
+import org.junit.Test;
+
+import com.sun.net.httpserver.HttpServer;
+
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+public class ExternalCatalogReadTest {
+
+ @Test
+ public void catalogScanUsesCatalogSchemaAndOneTabletFreePartition() throws
Exception {
+ List<String> requests = new CopyOnWriteArrayList<>();
+ HttpServer server = HttpServer.create(new
InetSocketAddress("127.0.0.1", 0), 0);
+ server.createContext("/", exchange -> {
+ String path = exchange.getRequestURI().getPath();
+ requests.add(path);
+ String data = path.endsWith("/_schema")
+ ? "{\"properties\":[{\"name\":\"id\",\"type\":\"INT\"},"
+ + "{\"name\":\"name\",\"type\":\"STRING\"}]}"
+ :
"{\"status\":0,\"partitions\":{},\"opaqued_query_plan\":\"\"}";
+ byte[] response = ("{\"code\":0,\"data\":" + data + "}")
+ .getBytes(StandardCharsets.UTF_8);
+ exchange.sendResponseHeaders(200, response.length);
+ try (java.io.OutputStream out = exchange.getResponseBody()) {
+ out.write(response);
+ }
+ });
+ server.start();
+ try {
+ DorisConfig config = createConfig("hive.db.tbl",
server.getAddress().getPort());
+ String[] filters = {"id > 1"};
+ DorisReaderPartition[] partitions =
ReaderPartitionGenerator.generatePartitions(
+ config, new String[0], filters, 5, false);
+
+ Assert.assertEquals(1, partitions.length);
+ Assert.assertEquals("db", partitions[0].getDatabase());
+ Assert.assertEquals("tbl", partitions[0].getTable());
+ Assert.assertNull(partitions[0].getBackend());
+ Assert.assertEquals(0, partitions[0].getTablets().length);
+ Assert.assertArrayEquals(new String[]{"`id`", "`name`"},
partitions[0].getReadColumns());
+ Assert.assertArrayEquals(filters, partitions[0].getFilters());
+ Assert.assertEquals(Integer.valueOf(5), partitions[0].getLimit());
+ Assert.assertTrue(requests.contains("/api/hive/db/tbl/_schema"));
+ Assert.assertFalse(requests.stream().anyMatch(path ->
path.endsWith("/_query_plan")));
+ Assert.assertEquals(1L, requests.stream().filter(path ->
path.endsWith("/_schema")).count());
+
+ DorisReaderPartition[] projected =
ReaderPartitionGenerator.generatePartitions(
+ config, new String[]{"name"}, new String[0], -1, false);
+ Assert.assertArrayEquals(new String[]{"`name`"},
projected[0].getReadColumns());
+ } finally {
+ server.stop(0);
+ }
+ }
+
+ @Test
+ public void identifiersPreserveCatalogAndNormalizeBackticks() throws
Exception {
+ Assert.assertArrayEquals(new String[]{"hive", "db", "tbl"},
+ createConfig("`hive`.`db`.`tbl`",
8030).getTableIdentifierParts());
+ Assert.assertArrayEquals(new String[]{"db", "tbl"},
+ createConfig("db.tbl", 8030).getTableIdentifierParts());
+ }
+
+ @Test
+ public void malformedIdentifiersAreRejected() {
+ for (String identifier : new String[]{"db", ".tbl", "db.", "cat..tbl",
"a.b.c.d", "``.tbl"}) {
+
org.junit.jupiter.api.Assertions.assertThrows(IllegalArgumentException.class,
+ () -> createConfig(identifier,
8030).getTableIdentifierParts());
+ }
+ }
+
+ private DorisConfig createConfig(String tableIdentifier, int httpPort)
throws Exception {
+ Map<String, String> options = new HashMap<>();
+ options.put("doris.fenodes", "127.0.0.1:" + httpPort);
+ options.put("doris.fe.auto.fetch", "false");
+ options.put("doris.table.identifier", tableIdentifier);
+ options.put("doris.user", "root");
+ options.put("doris.password", "");
+ options.put("doris.read.arrow-flight-sql.port", "9040");
+ return DorisConfig.fromMap(options, false);
+ }
+
+}
diff --git
a/spark-doris-connector/spark-doris-connector-base/src/test/java/org/apache/doris/spark/client/read/ReaderPartitionGeneratorTest.java
b/spark-doris-connector/spark-doris-connector-base/src/test/java/org/apache/doris/spark/client/read/ReaderPartitionGeneratorTest.java
index 768ddd6..7cf29be 100644
---
a/spark-doris-connector/spark-doris-connector-base/src/test/java/org/apache/doris/spark/client/read/ReaderPartitionGeneratorTest.java
+++
b/spark-doris-connector/spark-doris-connector-base/src/test/java/org/apache/doris/spark/client/read/ReaderPartitionGeneratorTest.java
@@ -19,7 +19,6 @@ package org.apache.doris.spark.client.read;
import mockit.Expectations;
import mockit.Mocked;
-import org.apache.doris.spark.client.DorisFrontendClient;
import org.apache.doris.spark.config.DorisConfig;
import org.apache.doris.spark.config.DorisOptions;
import org.apache.doris.spark.rest.models.Field;
@@ -70,7 +69,7 @@ public class ReaderPartitionGeneratorTest {
}
@Test
- public void getFinalReadColumnsTest(@Mocked DorisConfig config, @Mocked
DorisFrontendClient frontendClient) throws Exception {
+ public void getFinalReadColumnsTest(@Mocked DorisConfig config) throws
Exception {
String[] cols = new String[]{"c0", "c1", "c2"};
Schema schema = new Schema();
@@ -83,11 +82,9 @@ public class ReaderPartitionGeneratorTest {
result = true;
config.getValue(DorisOptions.DORIS_READ_BITMAP_TO_STRING);
result = false;
- frontendClient.getTableSchema(anyString, anyString);
- result = schema;
}};
- String[] finalReadColumns1 =
ReaderPartitionGenerator.getFinalReadColumns(config, frontendClient, "db",
"tbl", cols);
+ String[] finalReadColumns1 =
ReaderPartitionGenerator.getFinalReadColumns(config, schema, cols);
Assert.assertArrayEquals(new String[]{"`c0`", "bitmap_to_base64(`c1`)
AS `c1`", "'Read unsupported' AS `c2`"}, finalReadColumns1);
new Expectations() {{
@@ -95,11 +92,9 @@ public class ReaderPartitionGeneratorTest {
result = false;
config.getValue(DorisOptions.DORIS_READ_BITMAP_TO_STRING);
result = true;
- frontendClient.getTableSchema(anyString, anyString);
- result = schema;
}};
- String[] finalReadColumns2 =
ReaderPartitionGenerator.getFinalReadColumns(config, frontendClient, "db",
"tbl", cols);
+ String[] finalReadColumns2 =
ReaderPartitionGenerator.getFinalReadColumns(config, schema, cols);
Assert.assertArrayEquals(new String[]{"`c0`", "bitmap_to_string(`c1`)
AS `c1`", "'Read unsupported' AS `c2`"}, finalReadColumns2);
}
diff --git
a/spark-doris-connector/spark-doris-connector-base/src/test/scala/org/apache/doris/spark/util/URLsTest.scala
b/spark-doris-connector/spark-doris-connector-base/src/test/scala/org/apache/doris/spark/util/URLsTest.scala
index ff977c0..1799a63 100644
---
a/spark-doris-connector/spark-doris-connector-base/src/test/scala/org/apache/doris/spark/util/URLsTest.scala
+++
b/spark-doris-connector/spark-doris-connector-base/src/test/scala/org/apache/doris/spark/util/URLsTest.scala
@@ -34,6 +34,14 @@ class URLsTest {
Assert.assertEquals(URLs.tableSchema("127.0.0.1", 8030, "db", "tbl",
enableHttps = true), "https://127.0.0.1:8030/api/db/tbl/_schema";)
}
+ @Test
+ def catalogTableSchemaTest(): Unit = {
+ Assert.assertEquals("http://127.0.0.1:8030/api/hive/db/tbl/_schema";,
+ URLs.tableSchema("127.0.0.1", 8030, Array("hive", "db", "tbl"),
enableHttps = false))
+ Assert.assertEquals("https://127.0.0.1:8030/api/hive/db/tbl/_schema";,
+ URLs.tableSchema("127.0.0.1", 8030, Array("hive", "db", "tbl"),
enableHttps = true))
+ }
+
@Test
def queryPlanTest(): Unit = {
Assert.assertEquals(URLs.queryPlan("127.0.0.1", 8030, "db", "tbl",
enableHttps = false), "http://127.0.0.1:8030/api/db/tbl/_query_plan";)
diff --git
a/spark-doris-connector/spark-doris-connector-it/src/test/java/org/apache/doris/spark/container/AbstractMySqlCatalogTestBase.java
b/spark-doris-connector/spark-doris-connector-it/src/test/java/org/apache/doris/spark/container/AbstractMySqlCatalogTestBase.java
new file mode 100644
index 0000000..ecc91bd
--- /dev/null
+++
b/spark-doris-connector/spark-doris-connector-it/src/test/java/org/apache/doris/spark/container/AbstractMySqlCatalogTestBase.java
@@ -0,0 +1,135 @@
+// 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.doris.spark.container;
+
+import org.apache.doris.spark.container.instance.DorisContainer;
+
+import org.junit.AfterClass;
+import org.junit.Assume;
+import org.junit.BeforeClass;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.wait.strategy.Wait;
+import org.testcontainers.utility.DockerImageName;
+import org.testcontainers.utility.MountableFile;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.Statement;
+import java.time.Duration;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.UUID;
+import java.util.stream.Collectors;
+
+import static org.junit.Assert.assertTrue;
+
+/** Shares a MySQL-backed JDBC Catalog across the external-catalog read tests.
*/
+public abstract class AbstractMySqlCatalogTestBase extends
AbstractContainerTestBase {
+ private static final Logger LOG =
LoggerFactory.getLogger(AbstractMySqlCatalogTestBase.class);
+ private static final String DATABASE = "spark_catalog_source";
+ private static final String MYSQL_PASSWORD = "catalog-test";
+ private static GenericContainer<?> mysql;
+ private static String catalog;
+ private static boolean catalogCreated;
+
+ @BeforeClass
+ public static void startMySqlCatalog() throws Exception {
+ Assume.assumeTrue("This case requires the managed Doris container",
+ dorisContainerService instanceof DorisContainer);
+ GenericContainer<?> doris = ((DorisContainer)
dorisContainerService).getContainer();
+ String suffix = UUID.randomUUID().toString().replace("-",
"").substring(0, 8);
+ catalog = "spark_mysql_catalog_" + suffix;
+ String mysqlHost = "mysql-catalog-" + suffix;
+ mysql = new GenericContainer<>(DockerImageName.parse("mysql:8.0.36"))
+ .withNetwork(doris.getNetwork())
+ .withNetworkAliases(mysqlHost)
+ .withExposedPorts(3306)
+ .withEnv("MYSQL_ROOT_PASSWORD", MYSQL_PASSWORD)
+ .withEnv("MYSQL_ROOT_HOST", "%")
+ .withEnv("MYSQL_DATABASE", DATABASE)
+ .waitingFor(Wait.forLogMessage(".*mysqld: ready for
connections.*port: 3306 .*\\n", 1))
+ .withStartupTimeout(Duration.ofMinutes(2));
+ mysql.start();
+ String mysqlUrl = "jdbc:mysql://" + mysql.getHost() + ":" +
mysql.getMappedPort(3306)
+ + "/" + DATABASE +
"?useSSL=false&allowPublicKeyRetrieval=true&serverTimezone=UTC";
+ try (Connection connection = DriverManager.getConnection(mysqlUrl,
"root", MYSQL_PASSWORD);
+ Statement statement = connection.createStatement()) {
+ statement.execute("CREATE TABLE `order` (id INT PRIMARY KEY, name
VARCHAR(32), score INT)");
+ statement.execute("INSERT INTO `order` VALUES
(1,'alpha',10),(2,'beta',20),(3,NULL,NULL)");
+ statement.execute("CREATE TABLE empty_records (id INT PRIMARY KEY,
name VARCHAR(32), score INT)");
+ }
+
+ Path driverJar = Paths.get(com.mysql.cj.jdbc.Driver.class
+ .getProtectionDomain().getCodeSource().getLocation().toURI());
+ assertTrue("Expected mysql-connector-j JAR on the test classpath: " +
driverJar,
+ Files.isRegularFile(driverJar) &&
driverJar.toString().endsWith(".jar"));
+ String driverPath = "/tmp/spark-catalog-mysql-" + suffix + ".jar";
+
doris.copyFileToContainer(MountableFile.forHostPath(driverJar.toString()),
driverPath);
+ Map<String, String> properties = new LinkedHashMap<>();
+ properties.put("type", "jdbc");
+ properties.put("user", "root");
+ properties.put("password", MYSQL_PASSWORD);
+ properties.put("jdbc_url", "jdbc:mysql://" + mysqlHost + ":3306/" +
DATABASE
+ +
"?useSSL=false&allowPublicKeyRetrieval=true&serverTimezone=UTC");
+ properties.put("driver_url", "file://" + driverPath);
+ properties.put("driver_class", "com.mysql.cj.jdbc.Driver");
+ properties.put("only_specified_database", "true");
+ executeSql("CREATE CATALOG `" + catalog + "` PROPERTIES(" +
renderOptions(properties) + ")");
+ catalogCreated = true;
+ }
+
+ @AfterClass
+ public static void stopMySqlCatalog() throws Exception {
+ try {
+ if (catalogCreated) {
+ executeSql("DROP CATALOG `" + catalog + "`");
+ }
+ } finally {
+ if (mysql != null) {
+ mysql.stop();
+ }
+ }
+ }
+
+ protected Map<String, String> connectorOptions(String table) {
+ Map<String, String> options = new LinkedHashMap<>();
+ options.put("doris.fenodes", getFenodes());
+ options.put("doris.table.identifier", "`" + catalog + "`.`" + DATABASE
+ "`.`" + table + "`");
+ options.put("doris.user", getDorisUsername());
+ options.put("doris.password", getDorisPassword());
+ return options;
+ }
+
+ protected static String renderOptions(Map<String, String> options) {
+ return options.entrySet().stream()
+ .map(entry -> "'" + entry.getKey().replace("'", "''") + "'='"
+ + entry.getValue().replace("'", "''") + "'")
+ .collect(Collectors.joining(","));
+ }
+
+ private static void executeSql(String sql) throws Exception {
+ try (Connection connection = getDorisQueryConnection()) {
+ ContainerUtils.executeSQLStatement(connection, LOG, sql);
+ }
+ }
+}
diff --git
a/spark-doris-connector/spark-doris-connector-it/src/test/java/org/apache/doris/spark/container/AbstractS3TvfTestBase.java
b/spark-doris-connector/spark-doris-connector-it/src/test/java/org/apache/doris/spark/container/AbstractS3TvfTestBase.java
index ab91781..59a1bd6 100644
---
a/spark-doris-connector/spark-doris-connector-it/src/test/java/org/apache/doris/spark/container/AbstractS3TvfTestBase.java
+++
b/spark-doris-connector/spark-doris-connector-it/src/test/java/org/apache/doris/spark/container/AbstractS3TvfTestBase.java
@@ -52,7 +52,7 @@ public abstract class AbstractS3TvfTestBase extends
AbstractContainerTestBase {
private static final int MINIO_PORT = 9000;
private static final String MINIO_IMAGE =
- "quay.io/minio/minio:RELEASE.2024-10-13T13-34-11Z";
+ "doristhirdpartydocker/minio:RELEASE.2024-11-07T00-52-20Z";
private static GenericContainer<?> minio;
private static S3Client s3Client;
diff --git
a/spark-doris-connector/spark-doris-connector-it/src/test/java/org/apache/doris/spark/container/instance/DorisContainer.java
b/spark-doris-connector/spark-doris-connector-it/src/test/java/org/apache/doris/spark/container/instance/DorisContainer.java
index 1e99536..ab8a0f4 100644
---
a/spark-doris-connector/spark-doris-connector-it/src/test/java/org/apache/doris/spark/container/instance/DorisContainer.java
+++
b/spark-doris-connector/spark-doris-connector-it/src/test/java/org/apache/doris/spark/container/instance/DorisContainer.java
@@ -65,6 +65,10 @@ public class DorisContainer implements ContainerService {
dorisContainer = createDorisContainer();
}
+ public GenericContainer<?> getContainer() {
+ return dorisContainer;
+ }
+
public GenericContainer createDorisContainer() {
LOG.info("Will create doris containers.");
GenericContainer container =
diff --git
a/spark-doris-connector/spark-doris-connector-it/src/test/java/org/apache/doris/spark/sql/DorisExternalCatalogITCase.scala
b/spark-doris-connector/spark-doris-connector-it/src/test/java/org/apache/doris/spark/sql/DorisExternalCatalogITCase.scala
new file mode 100644
index 0000000..2b6295c
--- /dev/null
+++
b/spark-doris-connector/spark-doris-connector-it/src/test/java/org/apache/doris/spark/sql/DorisExternalCatalogITCase.scala
@@ -0,0 +1,88 @@
+// 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.doris.spark.sql
+
+import org.apache.doris.spark.container.AbstractMySqlCatalogTestBase
+import
org.apache.doris.spark.container.AbstractMySqlCatalogTestBase.renderOptions
+import org.apache.doris.spark.sparkContextFunctions
+import org.apache.spark.sql.{Row, SparkSession}
+import org.junit.Assert.assertEquals
+import org.junit.{After, Before, Test}
+
+import scala.collection.JavaConverters._
+
+/** Covers each Spark read API against a real MySQL-backed external catalog. */
+class DorisExternalCatalogITCase extends AbstractMySqlCatalogTestBase {
+
+ private var session: SparkSession = _
+
+ @Before
+ def startSpark(): Unit = {
+ session = SparkSession.builder()
+ .appName("doris-external-catalog-it")
+ .master("local[2]")
+ .config("spark.ui.enabled", "false")
+ .config("spark.driver.host", "127.0.0.1")
+ .config("spark.driver.bindAddress", "127.0.0.1")
+ .config("spark.sql.shuffle.partitions", "2")
+ .getOrCreate()
+ }
+
+ @After
+ def stopSpark(): Unit = {
+ if (session != null) session.stop()
+ }
+
+ @Test
+ def testDataFrameRead(): Unit = {
+ // Leave read mode and Flight port unset to exercise the default Arrow
path.
+ val data =
session.read.format("doris").options(connectorOptions("order").asScala).load()
+ assertEquals(1, data.rdd.getNumPartitions)
+ assertEquals(Seq(Row(1, "alpha", 10), Row(2, "beta", 20), Row(3, null,
null)),
+ data.orderBy("id").collect().toSeq)
+ assertEquals(Seq(Row("beta", 2), Row(null, 3)),
+ data.select("name", "id").where("id >= 2").orderBy("id").collect().toSeq)
+ assertEquals(1, data.limit(1).collect().length)
+ }
+
+ @Test
+ def testSparkSqlRead(): Unit = {
+ session.sql("CREATE TEMPORARY VIEW external_catalog_source USING doris
OPTIONS(" +
+ renderOptions(connectorOptions("order")) + ")")
+ assertEquals(Seq(Row("beta", 2), Row(null, 3)), session.sql(
+ "SELECT name,id FROM external_catalog_source WHERE id >= 2 ORDER BY
id").collect().toSeq)
+ }
+
+ @Test
+ def testRddRead(): Unit = {
+ val options = connectorOptions("order").asScala.toMap
+ val rows = session.sparkContext.dorisRDD(
+ tableIdentifier = Some(options("doris.table.identifier")),
+ query = Some("id >= 2"),
+ cfg = Some(options + ("doris.read.fields" -> "name,id")))
+ .collect().map(_.asInstanceOf[Array[AnyRef]].toSeq)
+ .sortBy(_(1).asInstanceOf[java.lang.Number].intValue()).toSeq
+ assertEquals(Seq(Seq("beta", 2), Seq(null, 3)), rows)
+ }
+
+ @Test
+ def testEmptyTableRead(): Unit = {
+ assertEquals(0L, session.read.format("doris")
+ .options(connectorOptions("empty_records").asScala).load().count())
+ }
+}
diff --git
a/spark-doris-connector/spark-doris-connector-spark-2/src/main/scala/org/apache/doris/spark/sql/DorisSourceProvider.scala
b/spark-doris-connector/spark-doris-connector-spark-2/src/main/scala/org/apache/doris/spark/sql/DorisSourceProvider.scala
index b2fb7e5..42f9b74 100644
---
a/spark-doris-connector/spark-doris-connector-spark-2/src/main/scala/org/apache/doris/spark/sql/DorisSourceProvider.scala
+++
b/spark-doris-connector/spark-doris-connector-spark-2/src/main/scala/org/apache/doris/spark/sql/DorisSourceProvider.scala
@@ -56,6 +56,7 @@ private[sql] class DorisSourceProvider extends
DorisSourceRegisterTrait
data: DataFrame): BaseRelation = {
val config =
DorisConfig.fromMap(sqlContext.sparkContext.getConf.getAll.toMap.asJava,
parameters.asJava, false)
+ require(config.getTableIdentifierParts.length == 2, "Writing
catalog.db.table is not supported")
mode match {
case SaveMode.Overwrite =>
@@ -91,6 +92,7 @@ private[sql] class DorisSourceProvider extends
DorisSourceRegisterTrait
override def createSink(sqlContext: SQLContext, parameters: Map[String,
String], partitionColumns: Seq[String], outputMode: OutputMode): Sink = {
val config = DorisConfig.fromMap(Utils.params(parameters, logger).asJava,
false)
+ require(config.getTableIdentifierParts.length == 2, "Writing
catalog.db.table is not supported")
if (config.getValue(DorisOptions.LOAD_MODE) == "tvf") {
throw new UnsupportedOperationException("tvf write mode does not support
Structured Streaming")
}
diff --git
a/spark-doris-connector/spark-doris-connector-spark-3-base/src/main/scala/org/apache/doris/spark/catalog/DorisTableBase.scala
b/spark-doris-connector/spark-doris-connector-spark-3-base/src/main/scala/org/apache/doris/spark/catalog/DorisTableBase.scala
index 214732f..ef0921a 100644
---
a/spark-doris-connector/spark-doris-connector-spark-3-base/src/main/scala/org/apache/doris/spark/catalog/DorisTableBase.scala
+++
b/spark-doris-connector/spark-doris-connector-spark-3-base/src/main/scala/org/apache/doris/spark/catalog/DorisTableBase.scala
@@ -40,11 +40,14 @@ abstract class DorisTableBase(identifier: Identifier,
config: DorisConfig, schem
override def name(): String = identifier.toString
override def schema(): StructType = schema.getOrElse({
- val dorisSchema = frontend.getTableSchema(identifier.namespace()(0),
identifier.name())
+ val dorisSchema = frontend.getTableSchema(identifier.namespace() :+
identifier.name())
dorisSchemaToStructType(dorisSchema)
})
override def capabilities(): util.Set[TableCapability] = {
+ if (identifier.namespace().length == 2) {
+ return Set[TableCapability](BATCH_READ).asJava
+ }
val capabilities = mutable.Set(BATCH_READ,
BATCH_WRITE,
STREAMING_WRITE,
@@ -59,7 +62,6 @@ abstract class DorisTableBase(identifier: Identifier, config:
DorisConfig, schem
}
override def newScanBuilder(caseInsensitiveStringMap:
CaseInsensitiveStringMap): ScanBuilder = {
- config.setProperty(DorisOptions.DORIS_TABLE_IDENTIFIER, name())
createScanBuilder(config, schema())
}
diff --git
a/spark-doris-connector/spark-doris-connector-spark-3-base/src/main/scala/org/apache/doris/spark/catalog/DorisTableProviderBase.scala
b/spark-doris-connector/spark-doris-connector-spark-3-base/src/main/scala/org/apache/doris/spark/catalog/DorisTableProviderBase.scala
index a03435d..39eeaeb 100644
---
a/spark-doris-connector/spark-doris-connector-spark-3-base/src/main/scala/org/apache/doris/spark/catalog/DorisTableProviderBase.scala
+++
b/spark-doris-connector/spark-doris-connector-spark-3-base/src/main/scala/org/apache/doris/spark/catalog/DorisTableProviderBase.scala
@@ -17,7 +17,7 @@
package org.apache.doris.spark.catalog
-import org.apache.doris.spark.config.{DorisConfig, DorisOptions}
+import org.apache.doris.spark.config.DorisConfig
import org.apache.spark.sql.connector.catalog.{Identifier, Table,
TableProvider}
import org.apache.spark.sql.connector.expressions.Transform
import org.apache.spark.sql.types.StructType
@@ -38,9 +38,8 @@ abstract class DorisTableProviderBase extends TableProvider {
if (t != null) t
else {
val dorisConfig = DorisConfig.fromMap(properties, false)
- val tableIdentifier =
dorisConfig.getValue(DorisOptions.DORIS_TABLE_IDENTIFIER)
- val tableIdentifierArr = tableIdentifier.split("\\.")
- newTableInstance(Identifier.of(Array[String](tableIdentifierArr(0)),
tableIdentifierArr(1)), dorisConfig, Some(schema))
+ val tableIdentifierArr = dorisConfig.getTableIdentifierParts
+ newTableInstance(Identifier.of(tableIdentifierArr.dropRight(1),
tableIdentifierArr.last), dorisConfig, Some(schema))
}
}
@@ -48,9 +47,8 @@ abstract class DorisTableProviderBase extends TableProvider {
if (t != null) t
else {
val dorisConfig = DorisConfig.fromMap(options, false)
- val tableIdentifier =
dorisConfig.getValue(DorisOptions.DORIS_TABLE_IDENTIFIER)
- val tableIdentifierArr = tableIdentifier.split("\\.")
- newTableInstance(Identifier.of(Array[String](tableIdentifierArr(0)),
tableIdentifierArr(1)), dorisConfig, None)
+ val tableIdentifierArr = dorisConfig.getTableIdentifierParts
+ newTableInstance(Identifier.of(tableIdentifierArr.dropRight(1),
tableIdentifierArr.last), dorisConfig, None)
}
}
diff --git
a/spark-doris-connector/spark-doris-connector-spark-3-base/src/test/scala/org/apache/doris/spark/catalog/DorisTableBaseTest.scala
b/spark-doris-connector/spark-doris-connector-spark-3-base/src/test/scala/org/apache/doris/spark/catalog/DorisTableBaseTest.scala
new file mode 100644
index 0000000..7fedca5
--- /dev/null
+++
b/spark-doris-connector/spark-doris-connector-spark-3-base/src/test/scala/org/apache/doris/spark/catalog/DorisTableBaseTest.scala
@@ -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.doris.spark.catalog
+
+import org.apache.doris.spark.config.{DorisConfig, DorisOptions}
+import org.apache.spark.sql.connector.catalog.Identifier
+import org.apache.spark.sql.connector.read.ScanBuilder
+import org.apache.spark.sql.connector.write.WriteBuilder
+import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
+import org.junit.Assert.assertEquals
+import org.junit.Test
+
+import java.util
+import scala.collection.JavaConverters._
+
+class DorisTableBaseTest {
+
+ @Test
+ def readPreservesConfiguredIdentifierQuoting(): Unit = {
+ val tableIdentifier = "`hive`.`db`.`order`"
+ val config = DorisConfig.fromMap(Map(
+ "doris.fenodes" -> "127.0.0.1:8030",
+ "doris.table.identifier" -> tableIdentifier,
+ "doris.user" -> "root",
+ "doris.password" -> "").asJava, false)
+ val table = new DorisTableBase(
+ Identifier.of(Array("hive", "db"), "order"), config,
Some(StructType(Nil))) {
+ override protected def createScanBuilder(readConfig: DorisConfig,
schema: StructType): ScanBuilder = {
+ assertEquals(tableIdentifier,
readConfig.getValue(DorisOptions.DORIS_TABLE_IDENTIFIER))
+ null
+ }
+
+ override protected def createWriteBuilder(writeConfig: DorisConfig,
schema: StructType): WriteBuilder =
+ throw new UnsupportedOperationException()
+ }
+
+ table.newScanBuilder(new
CaseInsensitiveStringMap(util.Collections.emptyMap[String, String]()))
+ }
+}
diff --git
a/spark-doris-connector/spark-doris-connector-spark-4-base/src/main/scala/org/apache/doris/spark/catalog/DorisTableBase.scala
b/spark-doris-connector/spark-doris-connector-spark-4-base/src/main/scala/org/apache/doris/spark/catalog/DorisTableBase.scala
index 830aa5e..dea91e0 100644
---
a/spark-doris-connector/spark-doris-connector-spark-4-base/src/main/scala/org/apache/doris/spark/catalog/DorisTableBase.scala
+++
b/spark-doris-connector/spark-doris-connector-spark-4-base/src/main/scala/org/apache/doris/spark/catalog/DorisTableBase.scala
@@ -40,11 +40,14 @@ abstract class DorisTableBase(identifier: Identifier,
config: DorisConfig, schem
override def name(): String = identifier.toString
override def schema(): StructType = schema.getOrElse({
- val dorisSchema = frontend.getTableSchema(identifier.namespace()(0),
identifier.name())
+ val dorisSchema = frontend.getTableSchema(identifier.namespace() :+
identifier.name())
dorisSchemaToStructType(dorisSchema)
})
override def capabilities(): util.Set[TableCapability] = {
+ if (identifier.namespace().length == 2) {
+ return Set[TableCapability](BATCH_READ).asJava
+ }
val capabilities = mutable.Set(BATCH_READ,
BATCH_WRITE,
STREAMING_WRITE,
@@ -59,7 +62,6 @@ abstract class DorisTableBase(identifier: Identifier, config:
DorisConfig, schem
}
override def newScanBuilder(caseInsensitiveStringMap:
CaseInsensitiveStringMap): ScanBuilder = {
- config.setProperty(DorisOptions.DORIS_TABLE_IDENTIFIER, name())
createScanBuilder(config, schema())
}
diff --git
a/spark-doris-connector/spark-doris-connector-spark-4-base/src/main/scala/org/apache/doris/spark/catalog/DorisTableProviderBase.scala
b/spark-doris-connector/spark-doris-connector-spark-4-base/src/main/scala/org/apache/doris/spark/catalog/DorisTableProviderBase.scala
index a03435d..39eeaeb 100644
---
a/spark-doris-connector/spark-doris-connector-spark-4-base/src/main/scala/org/apache/doris/spark/catalog/DorisTableProviderBase.scala
+++
b/spark-doris-connector/spark-doris-connector-spark-4-base/src/main/scala/org/apache/doris/spark/catalog/DorisTableProviderBase.scala
@@ -17,7 +17,7 @@
package org.apache.doris.spark.catalog
-import org.apache.doris.spark.config.{DorisConfig, DorisOptions}
+import org.apache.doris.spark.config.DorisConfig
import org.apache.spark.sql.connector.catalog.{Identifier, Table,
TableProvider}
import org.apache.spark.sql.connector.expressions.Transform
import org.apache.spark.sql.types.StructType
@@ -38,9 +38,8 @@ abstract class DorisTableProviderBase extends TableProvider {
if (t != null) t
else {
val dorisConfig = DorisConfig.fromMap(properties, false)
- val tableIdentifier =
dorisConfig.getValue(DorisOptions.DORIS_TABLE_IDENTIFIER)
- val tableIdentifierArr = tableIdentifier.split("\\.")
- newTableInstance(Identifier.of(Array[String](tableIdentifierArr(0)),
tableIdentifierArr(1)), dorisConfig, Some(schema))
+ val tableIdentifierArr = dorisConfig.getTableIdentifierParts
+ newTableInstance(Identifier.of(tableIdentifierArr.dropRight(1),
tableIdentifierArr.last), dorisConfig, Some(schema))
}
}
@@ -48,9 +47,8 @@ abstract class DorisTableProviderBase extends TableProvider {
if (t != null) t
else {
val dorisConfig = DorisConfig.fromMap(options, false)
- val tableIdentifier =
dorisConfig.getValue(DorisOptions.DORIS_TABLE_IDENTIFIER)
- val tableIdentifierArr = tableIdentifier.split("\\.")
- newTableInstance(Identifier.of(Array[String](tableIdentifierArr(0)),
tableIdentifierArr(1)), dorisConfig, None)
+ val tableIdentifierArr = dorisConfig.getTableIdentifierParts
+ newTableInstance(Identifier.of(tableIdentifierArr.dropRight(1),
tableIdentifierArr.last), dorisConfig, None)
}
}
diff --git
a/spark-doris-connector/spark-doris-connector-spark-4-base/src/test/scala/org/apache/doris/spark/catalog/DorisTableBaseTest.scala
b/spark-doris-connector/spark-doris-connector-spark-4-base/src/test/scala/org/apache/doris/spark/catalog/DorisTableBaseTest.scala
new file mode 100644
index 0000000..7fedca5
--- /dev/null
+++
b/spark-doris-connector/spark-doris-connector-spark-4-base/src/test/scala/org/apache/doris/spark/catalog/DorisTableBaseTest.scala
@@ -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.doris.spark.catalog
+
+import org.apache.doris.spark.config.{DorisConfig, DorisOptions}
+import org.apache.spark.sql.connector.catalog.Identifier
+import org.apache.spark.sql.connector.read.ScanBuilder
+import org.apache.spark.sql.connector.write.WriteBuilder
+import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
+import org.junit.Assert.assertEquals
+import org.junit.Test
+
+import java.util
+import scala.collection.JavaConverters._
+
+class DorisTableBaseTest {
+
+ @Test
+ def readPreservesConfiguredIdentifierQuoting(): Unit = {
+ val tableIdentifier = "`hive`.`db`.`order`"
+ val config = DorisConfig.fromMap(Map(
+ "doris.fenodes" -> "127.0.0.1:8030",
+ "doris.table.identifier" -> tableIdentifier,
+ "doris.user" -> "root",
+ "doris.password" -> "").asJava, false)
+ val table = new DorisTableBase(
+ Identifier.of(Array("hive", "db"), "order"), config,
Some(StructType(Nil))) {
+ override protected def createScanBuilder(readConfig: DorisConfig,
schema: StructType): ScanBuilder = {
+ assertEquals(tableIdentifier,
readConfig.getValue(DorisOptions.DORIS_TABLE_IDENTIFIER))
+ null
+ }
+
+ override protected def createWriteBuilder(writeConfig: DorisConfig,
schema: StructType): WriteBuilder =
+ throw new UnsupportedOperationException()
+ }
+
+ table.newScanBuilder(new
CaseInsensitiveStringMap(util.Collections.emptyMap[String, String]()))
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
