This is an automated email from the ASF dual-hosted git repository.
CalvinKirs 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 6f02e20 [Test] Add Doris regression test programs (#370)
6f02e20 is described below
commit 6f02e208fa2b2132ad3b1c604e877f8216da61f3
Author: wudi <[email protected]>
AuthorDate: Thu Aug 13 11:39:21 2026 +0800
[Test] Add Doris regression test programs (#370)
* [Test] Add Doris regression test programs
* [Fix] Correct regression build configuration
* [Test] Rename Doris regression case jar
* [Fix] Use append mode for Arrow regression writes
* [Fix] Fix Arrow regression stream load
* [Fix] Stabilize Arrow regression datetime values
---
doris-regression/README.md | 58 ++++++++++
doris-regression/pom.xml | 121 +++++++++++++++++++++
.../src/main/resources/log4j2.properties | 26 +++++
.../scala/org/apache/doris/DorisArguments.scala | 80 ++++++++++++++
.../apache/doris/DorisSparkReadWriterDemo.scala | 67 ++++++++++++
.../testcase/TestStreamLoadForArrowType.scala | 33 ++++--
.../spark/client/write/StreamLoadProcessor.java | 6 +-
7 files changed, 380 insertions(+), 11 deletions(-)
diff --git a/doris-regression/README.md b/doris-regression/README.md
new file mode 100644
index 0000000..d2cfaf4
--- /dev/null
+++ b/doris-regression/README.md
@@ -0,0 +1,58 @@
+<!--
+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.
+-->
+
+# Doris Regression Cases
+
+This standalone Maven project packages the Spark connector programs used by
the Apache Doris
+regression suite. It is intentionally kept outside the connector Maven reactor
and pins the
+connector version in `pom.xml`.
+
+## Build
+
+```shell
+mvn clean package
+```
+
+The shaded artifact is generated at `target/spark-doris-regression-case.jar`.
+
+## Entry Points
+
+| Main class | Apache Doris regression case | Case-specific arguments |
+| --- | --- | --- |
+| `org.apache.doris.DorisSparkReadWriterDemo` |
`regression-test/suites/connector_p0/spark_connector/spark_connector_read_type.groovy`
| `--doris-read-table-identifier <database.table>` and
`--doris-write-table-identifier <database.table>` |
+| `org.apache.doris.spark.testcase.TestStreamLoadForArrowType` |
`regression-test/suites/connector_p0/spark_connector/spark_connector_arrow.groovy`
| `--doris-database <database>` |
+
+All arguments are passed as name-value pairs. The common required arguments
are:
+
+```text
+--doris-fe-address <host:port>
+--doris-user <user>
+--doris-password <password>
+```
+
+The optional TLS arguments are shared by both entry points:
+
+```text
+--doris-enable-tls <true|false>
+--doris-tls-ca-certificate-path <path>
+--doris-tls-skip-hostname-verification <true|false>
+--doris-tls-excluded-protocols <http,mysql,thrift,arrowflight>
+```
+
+An empty value for `--doris-tls-excluded-protocols` enables TLS for every
supported protocol.
diff --git a/doris-regression/pom.xml b/doris-regression/pom.xml
new file mode 100644
index 0000000..4042a05
--- /dev/null
+++ b/doris-regression/pom.xml
@@ -0,0 +1,121 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0";
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
https://maven.apache.org/xsd/maven-4.0.0.xsd";>
+ <modelVersion>4.0.0</modelVersion>
+
+ <groupId>org.apache.doris</groupId>
+ <artifactId>spark-doris-regression-case</artifactId>
+ <version>1.0-SNAPSHOT</version>
+
+ <properties>
+ <doris.connector.version>26.1.0-SNAPSHOT</doris.connector.version>
+ <java.version>1.8</java.version>
+ <maven.compiler.source>${java.version}</maven.compiler.source>
+ <maven.compiler.target>${java.version}</maven.compiler.target>
+ <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+ <scala.major.version>2.12</scala.major.version>
+ <scala.version>2.12.17</scala.version>
+ <spark.version>3.4.2</spark.version>
+ </properties>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.doris</groupId>
+ <artifactId>spark-doris-connector-spark-3.4</artifactId>
+ <version>${doris.connector.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.spark</groupId>
+ <artifactId>spark-core_${scala.major.version}</artifactId>
+ <version>${spark.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.spark</groupId>
+ <artifactId>spark-sql_${scala.major.version}</artifactId>
+ <version>${spark.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.scala-lang</groupId>
+ <artifactId>scala-library</artifactId>
+ <version>${scala.version}</version>
+ </dependency>
+ </dependencies>
+
+ <build>
+ <finalName>spark-doris-regression-case</finalName>
+ <plugins>
+ <plugin>
+ <artifactId>maven-compiler-plugin</artifactId>
+ <version>3.6.2</version>
+ <configuration>
+ <source>${java.version}</source>
+ <target>${java.version}</target>
+ </configuration>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-shade-plugin</artifactId>
+ <version>3.6.0</version>
+ <executions>
+ <execution>
+ <phase>package</phase>
+ <goals>
+ <goal>shade</goal>
+ </goals>
+ <configuration>
+
<createDependencyReducedPom>false</createDependencyReducedPom>
+ <filters>
+ <filter>
+ <artifact>*:*</artifact>
+ <excludes>
+ <exclude>META-INF/*.SF</exclude>
+ <exclude>META-INF/*.DSA</exclude>
+ <exclude>META-INF/*.RSA</exclude>
+ </excludes>
+ </filter>
+ </filters>
+ <transformers>
+ <transformer
implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
+ <transformer
implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
+
<mainClass>org.apache.doris.DorisSparkReadWriterDemo</mainClass>
+ </transformer>
+ </transformers>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
+ <groupId>org.scala-tools</groupId>
+ <artifactId>maven-scala-plugin</artifactId>
+ <version>2.15.2</version>
+ <executions>
+ <execution>
+ <goals>
+ <goal>compile</goal>
+ <goal>testCompile</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+</project>
diff --git a/doris-regression/src/main/resources/log4j2.properties
b/doris-regression/src/main/resources/log4j2.properties
new file mode 100644
index 0000000..de6bfd5
--- /dev/null
+++ b/doris-regression/src/main/resources/log4j2.properties
@@ -0,0 +1,26 @@
+################################################################################
+# 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.
+################################################################################
+
+rootLogger.level = info
+rootLogger.appenderRef.stdout.ref = console
+
+appender.console.type = Console
+appender.console.name = console
+appender.console.target = SYSTEM_ERR
+appender.console.layout.type = PatternLayout
+appender.console.layout.pattern = %d{yy/MM/dd HH:mm:ss} %p [%t] %c{1}: %m%n%ex
diff --git
a/doris-regression/src/main/scala/org/apache/doris/DorisArguments.scala
b/doris-regression/src/main/scala/org/apache/doris/DorisArguments.scala
new file mode 100644
index 0000000..27b6551
--- /dev/null
+++ b/doris-regression/src/main/scala/org/apache/doris/DorisArguments.scala
@@ -0,0 +1,80 @@
+// 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
+
+final class DorisArguments private (options: Map[String, String]) {
+ import DorisArguments._
+
+ def feAddress: String = required(FE_ADDRESS)
+
+ def readTableIdentifier: String = required(READ_TABLE_IDENTIFIER)
+
+ def writeTableIdentifier: String = required(WRITE_TABLE_IDENTIFIER)
+
+ def database: String = required(DATABASE)
+
+ def user: String = required(USER)
+
+ def password: String = required(PASSWORD)
+
+ def tlsOptions: Map[String, String] = options.collect {
+ case (name, value) if TLS_ARGUMENT_NAMES.contains(name) =>
+ TLS_ARGUMENT_NAMES(name) -> value
+ }
+
+ private def required(name: String): String = {
+ options.getOrElse(name, throw new IllegalArgumentException(s"Missing
required argument: $name"))
+ }
+}
+
+object DorisArguments {
+ private val FE_ADDRESS = "--doris-fe-address"
+ private val READ_TABLE_IDENTIFIER = "--doris-read-table-identifier"
+ private val WRITE_TABLE_IDENTIFIER = "--doris-write-table-identifier"
+ private val DATABASE = "--doris-database"
+ private val USER = "--doris-user"
+ private val PASSWORD = "--doris-password"
+
+ private val TLS_ARGUMENT_NAMES = Map(
+ "--doris-enable-tls" -> "doris.enable.tls",
+ "--doris-tls-ca-certificate-path" -> "doris.tls.ca-certificate-path",
+ "--doris-tls-skip-hostname-verification" ->
"doris.tls.skip-hostname-verification",
+ "--doris-tls-excluded-protocols" -> "doris.tls.excluded-protocols"
+ )
+
+ private val ARGUMENT_NAMES = TLS_ARGUMENT_NAMES.keySet ++ Set(
+ FE_ADDRESS,
+ READ_TABLE_IDENTIFIER,
+ WRITE_TABLE_IDENTIFIER,
+ DATABASE,
+ USER,
+ PASSWORD
+ )
+
+ def parse(args: Array[String]): DorisArguments = {
+ require(args.length % 2 == 0, "Arguments must be name-value pairs")
+
+ val options = args.grouped(2).foldLeft(Map.empty[String, String]) {
(parsed, argument) =>
+ val name = argument(0)
+ require(ARGUMENT_NAMES.contains(name), s"Unknown argument: $name")
+ require(!parsed.contains(name), s"Duplicate argument: $name")
+ parsed.updated(name, argument(1))
+ }
+ new DorisArguments(options)
+ }
+}
diff --git
a/doris-regression/src/main/scala/org/apache/doris/DorisSparkReadWriterDemo.scala
b/doris-regression/src/main/scala/org/apache/doris/DorisSparkReadWriterDemo.scala
new file mode 100644
index 0000000..c5988ee
--- /dev/null
+++
b/doris-regression/src/main/scala/org/apache/doris/DorisSparkReadWriterDemo.scala
@@ -0,0 +1,67 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris
+
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.SparkSession
+
+object DorisSparkReadWriterDemo {
+
+ def main(args: Array[String]): Unit = {
+ println(s"Input arguments: ${args.mkString(" ")}")
+ val arguments = DorisArguments.parse(args)
+ val dorisTlsOptions = arguments.tlsOptions
+
+ val sparkConf: SparkConf = new SparkConf().setMaster("local[1]")
+
+ val spark = SparkSession.builder().config(sparkConf).getOrCreate()
+
+ val dorisReader = spark.read
+ .format("doris")
+ .option("doris.fenodes", arguments.feAddress)
+ .option("doris.table.identifier", arguments.readTableIdentifier)
+ .option("user", arguments.user)
+ .option("password", arguments.password)
+
+ val dorisSparkDF = dorisTlsOptions
+ .foldLeft(dorisReader) { case (reader, (name, value)) =>
+ reader.option(name, value)
+ }
+ .load()
+
+ dorisSparkDF.show()
+
+ val dorisWriter = dorisSparkDF.write
+ .format("doris")
+ .option("doris.fenodes", arguments.feAddress)
+ .option("doris.table.identifier", arguments.writeTableIdentifier)
+ .option("user", arguments.user)
+ .option("password", arguments.password)
+ .option("sink.batch.size",3)
+ .option("sink.max-retries",2)
+
+ dorisTlsOptions
+ .foldLeft(dorisWriter) { case (writer, (name, value)) =>
+ writer.option(name, value)
+ }
+ .mode("append")
+ .save()
+
+ spark.stop()
+ }
+}
diff --git
a/spark-doris-connector/spark-doris-connector-base/src/main/scala/org/apache/doris/spark/testcase/TestStreamLoadForArrowType.scala
b/doris-regression/src/main/scala/org/apache/doris/spark/testcase/TestStreamLoadForArrowType.scala
similarity index 95%
rename from
spark-doris-connector/spark-doris-connector-base/src/main/scala/org/apache/doris/spark/testcase/TestStreamLoadForArrowType.scala
rename to
doris-regression/src/main/scala/org/apache/doris/spark/testcase/TestStreamLoadForArrowType.scala
index c6b8e03..e17a248 100644
---
a/spark-doris-connector/spark-doris-connector-base/src/main/scala/org/apache/doris/spark/testcase/TestStreamLoadForArrowType.scala
+++
b/doris-regression/src/main/scala/org/apache/doris/spark/testcase/TestStreamLoadForArrowType.scala
@@ -17,6 +17,7 @@
package org.apache.doris.spark.testcase
+import org.apache.doris.DorisArguments
import org.apache.spark.sql.types._
import org.apache.spark.sql.{Row, SparkSession}
@@ -29,14 +30,18 @@ object TestStreamLoadForArrowType {
val spark: SparkSession =
SparkSession.builder().master("local[1]").getOrCreate()
var dorisFeNodes = "127.0.0.1:8030"
var dorisUser = "root"
- val dorisPwd = ""
+ var dorisPwd = ""
var databaseName = ""
+ var dorisTlsOptions = Map.empty[String, String]
def main(args: Array[String]): Unit = {
-
- dorisFeNodes = args(0)
- dorisUser = args(1)
- databaseName = args(2)
+ println(s"Input arguments: ${args.mkString(" ")}")
+ val arguments = DorisArguments.parse(args)
+ dorisFeNodes = arguments.feAddress
+ dorisUser = arguments.user
+ dorisPwd = arguments.password
+ databaseName = arguments.database
+ dorisTlsOptions = arguments.tlsOptions
testDataframeWritePrimitiveType()
testDataframeWriteArrayTypes()
@@ -145,10 +150,13 @@ object TestStreamLoadForArrowType {
.option("doris.fenodes", dorisFeNodes)
.option("user", dorisUser)
.option("password", dorisPwd)
+ .options(dorisTlsOptions)
.option("doris.table.identifier",
s"$databaseName.spark_connector_primitive")
.option("doris.sink.batch.size", 3)
.option("doris.sink.properties.format", "arrow")
+ .option("doris.write.schemaless", "true")
.option("doris.sink.max-retries", 0)
+ .mode("append")
.save()
}
@@ -227,7 +235,7 @@ object TestStreamLoadForArrowType {
Array(4.toLong, 5.toLong, 6.toLong),
Array("123456789", "987654321", "123789456"),
Array(6.6.floatValue(), 6.7.floatValue(), 7.8.floatValue()),
- Array(7.7.doubleValue(), 8.8.doubleValue(), 8.9.floatValue()),
+ Array(7.5, 8.5, 9.25),
Array(Decimal.apply(3.12), Decimal.apply(1.12345)),
Array(Date.valueOf("2023-09-08"), Date.valueOf("2027-10-28")),
Array(Timestamp.valueOf("2023-09-08 17:12:34.123456"),
Timestamp.valueOf("2024-09-08 18:12:34.123456")),
@@ -250,10 +258,13 @@ object TestStreamLoadForArrowType {
.option("doris.fenodes", dorisFeNodes)
.option("user", dorisUser)
.option("password", dorisPwd)
+ .options(dorisTlsOptions)
.option("doris.table.identifier", s"$databaseName.spark_connector_array")
.option("doris.sink.batch.size", 30)
.option("doris.sink.properties.format", "arrow")
+ .option("doris.write.schemaless", "true")
.option("doris.sink.max-retries", 0)
+ .mode("append")
.save()
}
@@ -334,8 +345,8 @@ object TestStreamLoadForArrowType {
Map(6.6.floatValue() -> 8.8.floatValue(), 9.9.floatValue() ->
10.1.floatValue()),
Map(7.7.doubleValue() -> 1.1.doubleValue(), 2.2 -> 3.3.doubleValue()),
Map(Decimal.apply(3.12) -> Decimal.apply(1.23), Decimal.apply(2.34) ->
Decimal.apply(5.67)),
- Map(Date.valueOf("2023-09-08") -> Date.valueOf("2024-09-08"),
Date.valueOf("1023-09-08") -> Date.valueOf("2023-09-08")),
- Map(Timestamp.valueOf("1023-09-08 17:12:34.123456") ->
Timestamp.valueOf("2023-09-08 17:12:34.123456"), Timestamp.valueOf("3023-09-08
17:12:34.123456") -> Timestamp.valueOf("4023-09-08 17:12:34.123456")),
+ Map(Date.valueOf("2023-09-08") -> Date.valueOf("2024-09-08"),
Date.valueOf("2022-09-08") -> Date.valueOf("2023-09-08")),
+ Map(Timestamp.valueOf("2022-09-08 17:12:34.123456") ->
Timestamp.valueOf("2023-09-08 17:12:34.123456"), Timestamp.valueOf("2024-09-08
17:12:34.123456") -> Timestamp.valueOf("2025-09-08 17:12:34.123456")),
Map("char" -> "char2", "char2" -> "char3"),
Map("varchar" -> "varchar2", "varchar3" -> "varchar4"),
Map("string" -> "string2", "string3" -> "string4")
@@ -355,10 +366,13 @@ object TestStreamLoadForArrowType {
.option("doris.fenodes", dorisFeNodes)
.option("user", dorisUser)
.option("password", dorisPwd)
+ .options(dorisTlsOptions)
.option("doris.table.identifier", s"$databaseName.spark_connector_map")
.option("doris.sink.batch.size", 3)
.option("doris.sink.properties.format", "arrow")
+ .option("doris.write.schemaless", "true")
.option("doris.sink.max-retries", 0)
+ .mode("append")
.save()
}
@@ -452,10 +466,13 @@ CREATE TABLE `spark_connector_struct` (
.option("doris.fenodes", dorisFeNodes)
.option("user", dorisUser)
.option("password", dorisPwd)
+ .options(dorisTlsOptions)
.option("doris.table.identifier",
s"$databaseName.spark_connector_struct")
.option("doris.sink.batch.size", 3)
.option("doris.sink.properties.format", "arrow")
+ .option("doris.write.schemaless", "true")
.option("doris.sink.max-retries", 0)
+ .mode("append")
.save()
}
}
diff --git
a/spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/client/write/StreamLoadProcessor.java
b/spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/client/write/StreamLoadProcessor.java
index e109ac8..ce8eb8f 100644
---
a/spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/client/write/StreamLoadProcessor.java
+++
b/spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/client/write/StreamLoadProcessor.java
@@ -30,7 +30,7 @@ import org.apache.arrow.vector.ipc.ArrowStreamWriter;
import org.apache.arrow.vector.types.pojo.Schema;
import org.apache.spark.TaskContext;
import org.apache.spark.sql.catalyst.InternalRow;
-import org.apache.spark.sql.execution.arrow.ArrowWriter;
+import org.apache.spark.sql.execution.arrow.DorisArrowWriter;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;
import org.apache.spark.sql.util.DorisArrowUtils;
@@ -67,7 +67,7 @@ public class StreamLoadProcessor extends
AbstractStreamLoadProcessor<InternalRow
public byte[] toArrowFormat(List<InternalRow> rowArray) throws IOException
{
Schema arrowSchema = DorisArrowUtils.toArrowSchema(schema, "UTC");
VectorSchemaRoot root = VectorSchemaRoot.create(arrowSchema, new
RootAllocator(Integer.MAX_VALUE));
- ArrowWriter arrowWriter = ArrowWriter.create(root);
+ DorisArrowWriter arrowWriter = DorisArrowWriter.create(root);
for (InternalRow row : rowArray) {
arrowWriter.write(row);
}
@@ -132,4 +132,4 @@ public class StreamLoadProcessor extends
AbstractStreamLoadProcessor<InternalRow
protected InternalRow copy(InternalRow row) {
return row.copy();
}
-}
\ No newline at end of file
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
