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 ff27c83 [Feature] Support Spark 4.1 (#365)
ff27c83 is described below
commit ff27c8399305148dd665459a58ad0a59aa7fcc6b
Author: YangJie <[email protected]>
AuthorDate: Fri Jul 17 17:10:47 2026 +0800
[Feature] Support Spark 4.1 (#365)
Adds Apache Spark 4.1 support (JDK 17, Scala 2.13.17) with full unit +
integration + E2E coverage, plus the foundational changes that let the shared
base module cross-build Scala 2.12 and 2.13.
Per the maintainer discussion, this targets Spark 4.1 only (4.0 is skipped).
New modules
spark-doris-connector-spark-4-base — shared Spark 4 DataSource V2 base
(fork of spark-3-base)
spark-doris-connector-spark-4.1 — Spark 4.1.0 / Scala 2.13.17 adapter
---
.github/workflows/build-extension.yml | 22 ++++
.github/workflows/run-e2ecase.yml | 21 ++++
.github/workflows/run-itcase.yml | 21 ++++
spark-doris-connector/build.sh | 36 +++++-
spark-doris-connector/pom.xml | 66 ++++++++++-
.../spark-doris-connector-base/pom.xml | 20 ++++
.../spark/client/write/StreamLoadProcessor.java | 13 ++-
.../org/apache/doris/spark/config/DorisConfig.java | 3 +-
.../org/apache/doris/spark/sql/DorisRow.scala | 2 -
.../doris/spark/sql/sources/DorisRelation.scala | 2 +-
.../testcase/TestStreamLoadForArrowType.scala | 8 +-
.../scala/org/apache/doris/spark/util/Retry.scala | 33 +++++-
.../apache/spark/sql/util/DorisArrowUtils.scala | 2 +-
.../spark-doris-connector-it/pom.xml | 31 +++++-
.../spark/sql/DorisWriterFailoverITCase.scala | 5 +-
.../spark-doris-connector-spark-4-base/pom.xml | 51 +++++++++
.../doris/spark/catalog/DorisTableBase.scala | 82 ++++++++++++++
.../spark/catalog/DorisTableCatalogBase.scala | 108 ++++++++++++++++++
.../spark/catalog/DorisTableProviderBase.scala | 59 ++++++++++
.../doris/spark/read/AbstractDorisScan.scala | 61 ++++++++++
.../doris/spark/read/DorisPartitionReader.scala | 75 +++++++++++++
.../spark/read/DorisPartitionReaderFactory.scala | 31 ++++++
.../org/apache/doris/spark/read/DorisScan.scala | 33 ++++++
.../doris/spark/read/DorisScanBuilderBase.scala | 52 +++++++++
.../org/apache/doris/spark/read/ScanMode.java | 22 ++++
.../spark/sql/sources/DorisSourceProvider.scala | 29 +++++
.../apache/doris/spark/write/DorisDataWriter.scala | 123 +++++++++++++++++++++
.../doris/spark/write/DorisDataWriterFactory.scala | 37 +++++++
.../org/apache/doris/spark/write/DorisWrite.scala | 91 +++++++++++++++
.../doris/spark/write/DorisWriteBuilder.scala | 47 ++++++++
.../spark-doris-connector-spark-4.1/pom.xml | 58 ++++++++++
...org.apache.spark.sql.sources.DataSourceRegister | 18 +++
.../apache/doris/spark/catalog/DorisTable.scala | 34 ++++++
.../doris/spark/catalog/DorisTableCatalog.scala | 28 +++++
.../apache/doris/spark/read/DorisScanBuilder.scala | 53 +++++++++
.../org/apache/doris/spark/read/DorisScanV2.scala | 34 ++++++
.../read/expression/V2ExpressionBuilder.scala | 120 ++++++++++++++++++++
.../doris/spark/sql/sources/DorisDataSource.scala | 30 +++++
.../read/expression/V2ExpressionBuilderTest.scala | 65 +++++++++++
.../org/apache/spark/sql/ExpressionUtil.scala | 30 +++++
tools/releasing/deploy_staging_jars.sh | 10 +-
...aging_jars.sh => deploy_staging_jars_spark4.sh} | 55 +++++----
42 files changed, 1670 insertions(+), 51 deletions(-)
diff --git a/.github/workflows/build-extension.yml
b/.github/workflows/build-extension.yml
index 008bc35..1fe1c73 100644
--- a/.github/workflows/build-extension.yml
+++ b/.github/workflows/build-extension.yml
@@ -73,3 +73,25 @@ jobs:
- name: Build spark load
run: |
cd spark-load && mvn clean package ${MVN_OPT} -Pspark3,scala_2.12
+
+ build-extension-spark4:
+ name: "Build Extensions (Spark 4.x)"
+ runs-on: ubuntu-latest
+ defaults:
+ run:
+ shell: bash
+ steps:
+ - name: Checkout
+ uses: actions/checkout@v3
+
+ # Spark 4.x requires JDK 17 and Scala 2.13, so it builds in a separate job.
+ - name: Setup java
+ uses: actions/setup-java@v3
+ with:
+ distribution: temurin
+ java-version: '17'
+ cache: 'maven'
+
+ - name: Build spark connector 4.1
+ run: |
+ cd spark-doris-connector && mvn clean install ${MVN_OPT} -Pspark-4.1
-pl spark-doris-connector-spark-4.1 -am
diff --git a/.github/workflows/run-e2ecase.yml
b/.github/workflows/run-e2ecase.yml
index 1991210..6ce6545 100644
--- a/.github/workflows/run-e2ecase.yml
+++ b/.github/workflows/run-e2ecase.yml
@@ -61,3 +61,24 @@ jobs:
- name: Run E2ECases for spark 3.5
run: |
cd spark-doris-connector && mvn clean test -Pspark-3-it,spark-3.5 -pl
spark-doris-connector-it -am -DfailIfNoTests=false -Dtest="*E2ECase"
-Dimage="apache/doris:doris-all-in-one-2.1.0"
+
+ run-e2ecase-spark4:
+ name: "Run E2ECases (Spark 4.x)"
+ runs-on: ubuntu-latest
+ defaults:
+ run:
+ shell: bash
+ steps:
+ - name: Checkout
+ uses: actions/checkout@master
+
+ # Spark 4.x requires JDK 17.
+ - name: Setup java
+ uses: actions/setup-java@v3
+ with:
+ distribution: temurin
+ java-version: '17'
+
+ - name: Run E2ECases for spark 4.1
+ run: |
+ cd spark-doris-connector && mvn clean test -Pspark-4-it,spark-4.1 -pl
spark-doris-connector-it -am -DfailIfNoTests=false -Dtest="*E2ECase"
-Dimage="apache/doris:doris-all-in-one-2.1.0"
diff --git a/.github/workflows/run-itcase.yml b/.github/workflows/run-itcase.yml
index bdc1225..1d4b8d7 100644
--- a/.github/workflows/run-itcase.yml
+++ b/.github/workflows/run-itcase.yml
@@ -62,3 +62,24 @@ jobs:
run: |
cd spark-doris-connector && mvn clean test -Pspark-3-it,spark-3.5 -pl
spark-doris-connector-it -am -DfailIfNoTests=false -Dtest="*ITCase"
-Dimage="apache/doris:doris-all-in-one-2.1.0"
+ run-itcase-spark4:
+ name: "Run ITCases (Spark 4.x)"
+ runs-on: ubuntu-latest
+ defaults:
+ run:
+ shell: bash
+ steps:
+ - name: Checkout
+ uses: actions/checkout@master
+
+ # Spark 4.x requires JDK 17.
+ - name: Setup java
+ uses: actions/setup-java@v3
+ with:
+ distribution: temurin
+ java-version: '17'
+
+ - name: Run ITCases for spark 4.1
+ run: |
+ cd spark-doris-connector && mvn clean test -Pspark-4-it,spark-4.1 -pl
spark-doris-connector-it -am -DfailIfNoTests=false -Dtest="*ITCase"
-Dimage="apache/doris:doris-all-in-one-2.1.0"
+
diff --git a/spark-doris-connector/build.sh b/spark-doris-connector/build.sh
index c1b1ea0..7cb56aa 100755
--- a/spark-doris-connector/build.sh
+++ b/spark-doris-connector/build.sh
@@ -122,8 +122,8 @@ if [[ -f ${DORIS_HOME}/custom_env.sh ]]; then
fi
selectScala() {
- echo 'Spark-Doris-Connector supports Scala 2.11 and 2.12. Which version do
you need ?'
- select scala in "2.11" "2.12"
+ echo 'Spark-Doris-Connector supports Scala 2.11, 2.12 and 2.13. Which
version do you need ?'
+ select scala in "2.11" "2.12" "2.13"
do
case $scala in
"2.11")
@@ -132,6 +132,9 @@ selectScala() {
"2.12")
return 2
;;
+ "2.13")
+ return 3
+ ;;
*)
echo "invalid selected, exit.."
exit 1
@@ -142,7 +145,7 @@ selectScala() {
selectSpark() {
echo 'Spark-Doris-Connector supports multiple versions of spark. Which
version do you need ?'
- select spark in "2.4" "3.1" "3.2" "3.3" "3.4" "3.5" "other"
+ select spark in "2.4" "3.1" "3.2" "3.3" "3.4" "3.5" "4.1" "other"
do
case $spark in
"2.4")
@@ -163,9 +166,12 @@ selectSpark() {
"3.5")
return 6
;;
- "other")
+ "4.1")
return 7
;;
+ "other")
+ return 8
+ ;;
esac
done
}
@@ -177,6 +183,8 @@ if [ ${ScalaVer} -eq 1 ]; then
SCALA_VERSION="2.11.12"
elif [ ${ScalaVer} -eq 2 ]; then
SCALA_VERSION="2.12.18"
+elif [ ${ScalaVer} -eq 3 ]; then
+ SCALA_VERSION="2.13.17"
fi
@@ -196,6 +204,8 @@ elif [ ${SparkVer} -eq 5 ]; then
elif [ ${SparkVer} -eq 6 ]; then
SPARK_VERSION="3.5.3"
elif [ ${SparkVer} -eq 7 ]; then
+ SPARK_VERSION="4.1.0"
+elif [ ${SparkVer} -eq 8 ]; then
# shellcheck disable=SC2162
read -p 'Which spark version do you need? please input
:' ver
@@ -207,6 +217,22 @@ if [[ $SPARK_VERSION =~ ^3.* && $SCALA_VERSION ==
"2.11.12" ]]; then
exit 1
fi
+if [[ $SPARK_VERSION =~ ^4.* && $SCALA_VERSION != "2.13.17" ]]; then
+ echo_r "Spark 4.x requires scala 2.13, will exit."
+ exit 1
+fi
+
+if [[ ! $SPARK_VERSION =~ ^4.* && $SCALA_VERSION == "2.13.17" ]]; then
+ echo_r "Scala 2.13 is only supported with Spark 4.x, will exit."
+ exit 1
+fi
+
+# Spark 4.x requires JDK 17. JAVA_VER is the class-file major version from
env.sh (52=JDK8, 61=JDK17).
+if [[ $SPARK_VERSION =~ ^4.* && ${JAVA_VER:-0} -lt 61 ]]; then
+ echo_r "Spark 4.x requires JDK 17 or later, will exit."
+ exit 1
+fi
+
# extract major version:
# eg: 3.1.2 -> 3.1
SCALA_MAJOR_VERSION=0
@@ -218,7 +244,7 @@ echo_g " scala version: ${SCALA_VERSION}, major version:
${SCALA_MAJOR_VERSION}"
echo_g " spark version: ${SPARK_VERSION}, major version:
${SPARK_MAJOR_VERSION}"
echo_g " build starting..."
-if [[ $SPARK_VERSION =~ ^3.* ]]; then
+if [[ $SPARK_VERSION =~ ^3.* || $SPARK_VERSION =~ ^4.* ]]; then
profile_name="spark-${SPARK_MAJOR_VERSION}"
module_suffix=${SPARK_MAJOR_VERSION}
else
diff --git a/spark-doris-connector/pom.xml b/spark-doris-connector/pom.xml
index 70edde1..873c3ef 100644
--- a/spark-doris-connector/pom.xml
+++ b/spark-doris-connector/pom.xml
@@ -49,6 +49,8 @@
<module>spark-doris-connector-spark-3.3</module>
<module>spark-doris-connector-spark-3.4</module>
<module>spark-doris-connector-spark-3.5</module>
+ <module>spark-doris-connector-spark-4-base</module>
+ <module>spark-doris-connector-spark-4.1</module>
</modules>
<scm>
@@ -86,6 +88,9 @@
<scala.major.version>2.11</scala.major.version>
<libthrift.version>0.16.0</libthrift.version>
<arrow.version>15.0.2</arrow.version>
+ <adbc.version>0.14.0</adbc.version>
+ <grpc.version>1.60.0</grpc.version>
+ <protobuf.version>3.22.3</protobuf.version>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<project.scm.id>github</project.scm.id>
<netty.version>4.1.110.Final</netty.version>
@@ -139,6 +144,16 @@
<artifactId>spark-doris-connector-spark-3.5</artifactId>
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.doris</groupId>
+ <artifactId>spark-doris-connector-spark-4-base</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.doris</groupId>
+ <artifactId>spark-doris-connector-spark-4.1</artifactId>
+ <version>${project.version}</version>
+ </dependency>
<dependency>
<groupId>org.apache.doris</groupId>
<artifactId>thrift-service</artifactId>
@@ -233,7 +248,7 @@
<dependency>
<groupId>org.apache.arrow.adbc</groupId>
<artifactId>adbc-driver-flight-sql</artifactId>
- <version>0.14.0</version>
+ <version>${adbc.version}</version>
<exclusions>
<exclusion>
<groupId>org.apache.arrow</groupId>
@@ -270,10 +285,22 @@
</exclusions>
</dependency>
+ <!-- Keep the whole gRPC stack on one version. grpc-netty pulls
grpc-api/core/util at
+ ${grpc.version}, but adbc -> arrow-flight pulls
grpc-stub/grpc-protobuf transitively
+ (e.g. flight 18 -> grpc 1.65 while grpc-netty is 1.69 under
Spark 4.1). Importing the
+ BOM aligns every io.grpc artifact, avoiding a mixed-version
gRPC frozen into the
+ shaded jar. The explicit grpc-netty entry below still applies
its netty exclusions. -->
+ <dependency>
+ <groupId>io.grpc</groupId>
+ <artifactId>grpc-bom</artifactId>
+ <version>${grpc.version}</version>
+ <type>pom</type>
+ <scope>import</scope>
+ </dependency>
<dependency>
<groupId>io.grpc</groupId>
<artifactId>grpc-netty</artifactId>
- <version>1.60.0</version>
+ <version>${grpc.version}</version>
<exclusions>
<exclusion>
<artifactId>netty-codec-http2</artifactId>
@@ -313,7 +340,7 @@
<dependency>
<groupId>com.google.protobuf</groupId>
<artifactId>protobuf-java</artifactId>
- <version>3.22.3</version>
+ <version>${protobuf.version}</version>
</dependency>
<dependency>
@@ -403,6 +430,15 @@
<version>${fasterxml.jackson.version}</version>
</dependency>
+ <!-- Pin jackson-datatype-jsr310 to the connector's jackson
version. Spark/Arrow pull a
+ newer jsr310 transitively (e.g. 2.20.0 on Spark 4.1), which
would otherwise be
+ shaded next to the connector's databind and break on
java.time (de)serialization. -->
+ <dependency>
+ <groupId>com.fasterxml.jackson.datatype</groupId>
+ <artifactId>jackson-datatype-jsr310</artifactId>
+ <version>${fasterxml.jackson.version}</version>
+ </dependency>
+
<!--
https://mvnrepository.com/artifact/com.mysql/mysql-connector-j -->
<dependency>
<groupId>com.mysql</groupId>
@@ -509,7 +545,9 @@
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-shade-plugin</artifactId>
- <version>3.4.1</version>
+ <!-- 3.6.0 bundles an ASM that can process Java 21
multi-release classes, e.g.
+ bcprov-jdk18on (pulled transitively under Spark 4.1's
grpc 1.69 / netty 4.2). -->
+ <version>3.6.0</version>
<configuration>
<filters>
<filter>
@@ -879,6 +917,26 @@
<scala.major.version>2.12</scala.major.version>
</properties>
</profile>
+ <profile>
+ <id>spark-4.1</id>
+ <properties>
+ <spark.version>4.1.0</spark.version>
+ <spark.major.version>4.1</spark.major.version>
+ <scala.version>2.13.17</scala.version>
+ <scala.major.version>2.13</scala.major.version>
+ <!-- Match Spark 4.1's Arrow and Netty so the connector and
Spark agree on one copy
+ of each on the (unshaded) integration-test classpath;
bump adbc/flight/grpc in
+ lockstep (grpc 1.69 supports Netty 4.2). -->
+ <arrow.version>18.3.0</arrow.version>
+ <adbc.version>0.18.0</adbc.version>
+ <grpc.version>1.69.0</grpc.version>
+ <protobuf.version>3.25.1</protobuf.version>
+ <netty.version>4.2.7.Final</netty.version>
+ <!-- Spark 4 on JDK 17 needs these module opens (Arrow
off-heap, Spark date/time,
+ Catalyst codegen, etc.) — mirrors Spark's own
JavaModuleOptions. -->
+ <argLine>-Xmx512m --add-opens=java.base/java.lang=ALL-UNNAMED
--add-opens=java.base/java.lang.invoke=ALL-UNNAMED
--add-opens=java.base/java.lang.reflect=ALL-UNNAMED
--add-opens=java.base/java.io=ALL-UNNAMED
--add-opens=java.base/java.net=ALL-UNNAMED
--add-opens=java.base/java.nio=ALL-UNNAMED
--add-opens=java.base/java.util=ALL-UNNAMED
--add-opens=java.base/java.util.concurrent=ALL-UNNAMED
--add-opens=java.base/java.util.concurrent.atomic=ALL-UNNAMED
--add-opens=java.base/ [...]
+ </properties>
+ </profile>
</profiles>
</project>
diff --git a/spark-doris-connector/spark-doris-connector-base/pom.xml
b/spark-doris-connector/spark-doris-connector-base/pom.xml
index 67ab8c3..e3480ca 100644
--- a/spark-doris-connector/spark-doris-connector-base/pom.xml
+++ b/spark-doris-connector/spark-doris-connector-base/pom.xml
@@ -54,6 +54,17 @@
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
<version>${netty.version}</version>
+ <exclusions>
+ <!-- netty-all 4.2.x (Spark 4.1) is an aggregator that drags
BouncyCastle in at
+ compile scope; it is not relocated by the shade plugin,
so it would ship an
+ unrelocated org.bouncycastle crypto provider that can
clash with the one Spark
+ puts on the classpath. The connector does not use
BouncyCastle, so exclude it.
+ (No-op on netty-all 4.1.x, which does not pull it.) -->
+ <exclusion>
+ <groupId>org.bouncycastle</groupId>
+ <artifactId>bcprov-jdk18on</artifactId>
+ </exclusion>
+ </exclusions>
</dependency>
<dependency>
@@ -103,6 +114,15 @@
<scope>runtime</scope>
</dependency>
+ <!-- Pin arrow-memory-core to the connector's Arrow version so the
test classpath stays
+ internally consistent. Spark 4.x ships a newer Arrow (provided),
which otherwise
+ leaks a mismatched arrow-memory-core and breaks the
allocation-manager lookup. -->
+ <dependency>
+ <groupId>org.apache.arrow</groupId>
+ <artifactId>arrow-memory-core</artifactId>
+ <version>${arrow.version}</version>
+ </dependency>
+
<dependency>
<groupId>org.apache.arrow.adbc</groupId>
<artifactId>adbc-driver-flight-sql</artifactId>
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 a67ab28..e109ac8 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
@@ -38,9 +38,19 @@ import org.apache.spark.sql.util.DorisArrowUtils;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.util.List;
+import java.util.concurrent.atomic.AtomicLong;
public class StreamLoadProcessor extends
AbstractStreamLoadProcessor<InternalRow> {
+ /**
+ * Monotonic per-JVM sequence appended to every generated label. The label
already carries
+ * {@code currentTimeMillis()}, but consecutive committed batches of the
same task (e.g.
+ * {@code batch.size=1} with a zero batch interval) can fall in the same
millisecond and collide;
+ * a colliding fresh label is rejected as "Label Already Exists". The
sequence guarantees each
+ * label is distinct so a retry (which mints a new label) never clashes
with a prior batch.
+ */
+ private static final AtomicLong LABEL_SEQ = new AtomicLong();
+
private StructType schema;
public StreamLoadProcessor(DorisConfig config) throws Exception {
@@ -110,7 +120,8 @@ public class StreamLoadProcessor extends
AbstractStreamLoadProcessor<InternalRow
long taskAttemptId = taskContext.taskAttemptId();
int partitionId = taskContext.partitionId();
String prefix = config.getValue(DorisOptions.DORIS_SINK_LABEL_PREFIX);
- return String.format("%s-%d-%d-%d-%d", prefix, stageId, taskAttemptId,
partitionId, System.currentTimeMillis());
+ return String.format("%s-%d-%d-%d-%d-%d", prefix, stageId,
taskAttemptId, partitionId,
+ System.currentTimeMillis(), LABEL_SEQ.getAndIncrement());
}
public void setSchema(StructType schema) {
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 79ed053..fd0bbd8 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
@@ -17,7 +17,6 @@
package org.apache.doris.spark.config;
-import org.apache.commons.collections.MapUtils;
import org.apache.doris.spark.exception.OptionRequiredException;
import java.io.Serializable;
@@ -184,7 +183,7 @@ public class DorisConfig implements Serializable {
public static DorisConfig fromMap(Map<String, String> sparkConfMap,
Map<String, String> options, Boolean ignoreTableCheck) throws
OptionRequiredException {
Map<String, String> map = new HashMap<>(sparkConfMap);
- if (MapUtils.isNotEmpty(options)) {
+ if (options != null && !options.isEmpty()) {
map.putAll(options);
}
return new DorisConfig(map, ignoreTableCheck);
diff --git
a/spark-doris-connector/spark-doris-connector-base/src/main/scala/org/apache/doris/spark/sql/DorisRow.scala
b/spark-doris-connector/spark-doris-connector-base/src/main/scala/org/apache/doris/spark/sql/DorisRow.scala
index 818af05..756574a 100644
---
a/spark-doris-connector/spark-doris-connector-base/src/main/scala/org/apache/doris/spark/sql/DorisRow.scala
+++
b/spark-doris-connector/spark-doris-connector-base/src/main/scala/org/apache/doris/spark/sql/DorisRow.scala
@@ -54,6 +54,4 @@ private[spark] class DorisRow(rowOrder: Seq[String]) extends
Row {
override def getString(i: Int): String = get(i).toString
override def copy(): Row = this
-
- override def toSeq: Seq[Any] = values
}
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 6fd7af5..084b821 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
@@ -48,7 +48,7 @@ private[sql] class DorisRelation(
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))
- }))
+ }).toArray)
}
diff --git
a/spark-doris-connector/spark-doris-connector-base/src/main/scala/org/apache/doris/spark/testcase/TestStreamLoadForArrowType.scala
b/spark-doris-connector/spark-doris-connector-base/src/main/scala/org/apache/doris/spark/testcase/TestStreamLoadForArrowType.scala
index f1e1977..c6b8e03 100644
---
a/spark-doris-connector/spark-doris-connector-base/src/main/scala/org/apache/doris/spark/testcase/TestStreamLoadForArrowType.scala
+++
b/spark-doris-connector/spark-doris-connector-base/src/main/scala/org/apache/doris/spark/testcase/TestStreamLoadForArrowType.scala
@@ -137,7 +137,7 @@ object TestStreamLoadForArrowType {
inputList.append(row)
}
- val rdd = spark.sparkContext.parallelize(inputList, 1)
+ val rdd = spark.sparkContext.parallelize(inputList.toList, 1)
val df = spark.createDataFrame(rdd, schema).toDF()
df.write
@@ -242,7 +242,7 @@ object TestStreamLoadForArrowType {
inputList.append(row)
}
- val rdd = spark.sparkContext.parallelize(inputList, 1)
+ val rdd = spark.sparkContext.parallelize(inputList.toList, 1)
val df = spark.createDataFrame(rdd, schema).toDF()
df.write
@@ -347,7 +347,7 @@ object TestStreamLoadForArrowType {
inputList.append(row)
}
- val rdd = spark.sparkContext.parallelize(inputList, 1)
+ val rdd = spark.sparkContext.parallelize(inputList.toList, 1)
val df = spark.createDataFrame(rdd, schema).toDF()
df.write
@@ -444,7 +444,7 @@ CREATE TABLE `spark_connector_struct` (
inputList.append(row)
}
- val rdd = spark.sparkContext.parallelize(inputList, 1)
+ val rdd = spark.sparkContext.parallelize(inputList.toList, 1)
val df = spark.createDataFrame(rdd, schema).toDF()
df.write
diff --git
a/spark-doris-connector/spark-doris-connector-base/src/main/scala/org/apache/doris/spark/util/Retry.scala
b/spark-doris-connector/spark-doris-connector-base/src/main/scala/org/apache/doris/spark/util/Retry.scala
index 24f8c13..10c5955 100644
---
a/spark-doris-connector/spark-doris-connector-base/src/main/scala/org/apache/doris/spark/util/Retry.scala
+++
b/spark-doris-connector/spark-doris-connector-base/src/main/scala/org/apache/doris/spark/util/Retry.scala
@@ -20,7 +20,7 @@ package org.apache.doris.spark.util
import org.slf4j.Logger
import java.time.Duration
-import java.util.concurrent.locks.LockSupport
+import java.util.concurrent.TimeUnit
import scala.annotation.tailrec
import scala.reflect.ClassTag
import scala.util.{Failure, Success, Try}
@@ -38,11 +38,40 @@ object Retry {
case Failure(exception: T) if retryTimes > 0 =>
logger.warn("Execution failed caused by: {}", exception.getMessage)
logger.warn(s"$retryTimes times retry remaining, the next attempt will
be in ${interval.toMillis} ms")
- LockSupport.parkNanos(interval.toNanos)
+ sleepBeforeRetry(interval)
h
exec(retryTimes - 1, interval, logger)(f)(h)
case Failure(exception) => Failure(exception)
}
}
+ /**
+ * Wait for the full interval before the next retry.
+ *
+ * [[java.util.concurrent.locks.LockSupport.parkNanos]] may return before
the requested time on a
+ * spurious wakeup or when another thread unparks this one. The stream-load
processor interrupts
+ * the task thread on an async load failure (to unblock a blocking pipe
write), and an interrupt
+ * also unparks a parked thread, so a single `parkNanos` for the backoff is
repeatedly cut short
+ * and the configured retry interval collapses to a few milliseconds — most
visibly under Spark
+ * 4.x. That both hammers the backend with retries and breaks failover that
relies on the interval.
+ *
+ * Use [[Thread.sleep]] (via [[TimeUnit]]), which is not affected by
`unpark`, and loop until the
+ * full interval has elapsed. A spurious interrupt is absorbed (the
interrupt status is restored
+ * afterwards) so the interval is still honored.
+ */
+ private def sleepBeforeRetry(interval: Duration): Unit = {
+ val deadlineNanos = System.nanoTime() + interval.toNanos
+ var remainingNanos = interval.toNanos
+ var interrupted = false
+ while (remainingNanos > 0L) {
+ try {
+ TimeUnit.NANOSECONDS.sleep(remainingNanos)
+ } catch {
+ case _: InterruptedException => interrupted = true
+ }
+ remainingNanos = deadlineNanos - System.nanoTime()
+ }
+ if (interrupted) Thread.currentThread().interrupt()
+ }
+
}
diff --git
a/spark-doris-connector/spark-doris-connector-base/src/main/scala/org/apache/spark/sql/util/DorisArrowUtils.scala
b/spark-doris-connector/spark-doris-connector-base/src/main/scala/org/apache/spark/sql/util/DorisArrowUtils.scala
index 3879321..4f81227 100644
---
a/spark-doris-connector/spark-doris-connector-base/src/main/scala/org/apache/spark/sql/util/DorisArrowUtils.scala
+++
b/spark-doris-connector/spark-doris-connector-base/src/main/scala/org/apache/spark/sql/util/DorisArrowUtils.scala
@@ -106,7 +106,7 @@ object DorisArrowUtils {
val dt = fromArrowField(child)
StructField(child.getName, dt, child.isNullable)
}
- StructType(fields)
+ StructType(fields.toArray)
case arrowType => fromArrowType(arrowType)
}
}
diff --git a/spark-doris-connector/spark-doris-connector-it/pom.xml
b/spark-doris-connector/spark-doris-connector-it/pom.xml
index 067a9b2..b61946a 100644
--- a/spark-doris-connector/spark-doris-connector-it/pom.xml
+++ b/spark-doris-connector/spark-doris-connector-it/pom.xml
@@ -82,12 +82,6 @@
<version>8.0.33</version>
<scope>test</scope>
</dependency>
- <dependency>
- <groupId>org.slf4j</groupId>
- <artifactId>slf4j-log4j12</artifactId>
- <version>1.7.25</version>
- <scope>test</scope>
- </dependency>
</dependencies>
<profiles>
@@ -99,12 +93,37 @@
<properties>
<spark.doris.connector.artifactId>spark-doris-connector-spark-2</spark.doris.connector.artifactId>
</properties>
+ <dependencies>
+ <!-- Spark 2.x / 3.x use slf4j 1.7 + log4j 1.2 -->
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-log4j12</artifactId>
+ <version>1.7.25</version>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
</profile>
<profile>
<id>spark-3-it</id>
<properties>
<spark.doris.connector.artifactId>spark-doris-connector-spark-${spark.major.version}</spark.doris.connector.artifactId>
</properties>
+ <dependencies>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-log4j12</artifactId>
+ <version>1.7.25</version>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+ </profile>
+ <profile>
+ <!-- Spark 4.x brings its own slf4j 2.x + log4j 2.x binding; the
slf4j 1.7
+ binding above would clash (NoSuchFieldError: mdc). -->
+ <id>spark-4-it</id>
+ <properties>
+
<spark.doris.connector.artifactId>spark-doris-connector-spark-${spark.major.version}</spark.doris.connector.artifactId>
+ </properties>
</profile>
</profiles>
diff --git
a/spark-doris-connector/spark-doris-connector-it/src/test/java/org/apache/doris/spark/sql/DorisWriterFailoverITCase.scala
b/spark-doris-connector/spark-doris-connector-it/src/test/java/org/apache/doris/spark/sql/DorisWriterFailoverITCase.scala
index 9c38eed..4381d4e 100644
---
a/spark-doris-connector/spark-doris-connector-it/src/test/java/org/apache/doris/spark/sql/DorisWriterFailoverITCase.scala
+++
b/spark-doris-connector/spark-doris-connector-it/src/test/java/org/apache/doris/spark/sql/DorisWriterFailoverITCase.scala
@@ -61,7 +61,10 @@ class DorisWriterFailoverITCase extends
AbstractContainerTestBase {
@Test
def testFailoverForRetry(): Unit = {
LOG.info("start to test testFailoverForRetry.")
- initializeTable(TABLE_WRITE_TBL_RETRY, DataModel.DUPLICATE)
+ // Use a UNIQUE (primary-key) table: without 2PC a retried batch may
re-load rows that already
+ // landed, and the unique key on `name` makes that re-load idempotent. The
three rows have
+ // distinct names, so dedup removes duplicates without dropping any
legitimate row.
+ initializeTable(TABLE_WRITE_TBL_RETRY, DataModel.UNIQUE)
val session = SparkSession.builder().master("local[1]").getOrCreate()
val df = session.createDataFrame(Seq(
("doris", "1234"),
diff --git a/spark-doris-connector/spark-doris-connector-spark-4-base/pom.xml
b/spark-doris-connector/spark-doris-connector-spark-4-base/pom.xml
new file mode 100644
index 0000000..61f1ee5
--- /dev/null
+++ b/spark-doris-connector/spark-doris-connector-spark-4-base/pom.xml
@@ -0,0 +1,51 @@
+<?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
http://maven.apache.org/xsd/maven-4.0.0.xsd";>
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.doris</groupId>
+ <artifactId>spark-doris-connector</artifactId>
+ <version>${revision}</version>
+ </parent>
+
+ <artifactId>spark-doris-connector-spark-4-base</artifactId>
+
+ <properties>
+ <maven.compiler.source>8</maven.compiler.source>
+ <maven.compiler.target>8</maven.compiler.target>
+ <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+ <maven.deploy.skip>true</maven.deploy.skip>
+ </properties>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.doris</groupId>
+ <artifactId>spark-doris-connector-base</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.spark</groupId>
+ <artifactId>spark-sql_${scala.major.version}</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ </dependencies>
+
+</project>
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
new file mode 100644
index 0000000..830aa5e
--- /dev/null
+++
b/spark-doris-connector/spark-doris-connector-spark-4-base/src/main/scala/org/apache/doris/spark/catalog/DorisTableBase.scala
@@ -0,0 +1,82 @@
+// 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.client.DorisFrontendClient
+import org.apache.doris.spark.config.{DorisConfig, DorisOptions}
+import org.apache.doris.spark.rest.models.Schema
+import org.apache.doris.spark.util.SchemaConvertors
+import org.apache.spark.sql.connector.catalog.TableCapability._
+import org.apache.spark.sql.connector.catalog.{Identifier, SupportsRead,
SupportsWrite, Table, TableCapability}
+import org.apache.spark.sql.connector.read.ScanBuilder
+import org.apache.spark.sql.connector.write.{LogicalWriteInfo, WriteBuilder}
+import org.apache.spark.sql.types.{StructField, StructType}
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
+
+import java.util
+import scala.collection.JavaConverters._
+import scala.collection.mutable
+
+
+abstract class DorisTableBase(identifier: Identifier, config: DorisConfig,
schema: Option[StructType]) extends Table with SupportsRead with SupportsWrite {
+
+ private lazy val frontend:DorisFrontendClient = new
DorisFrontendClient(config)
+
+ override def name(): String = identifier.toString
+
+ override def schema(): StructType = schema.getOrElse({
+ val dorisSchema = frontend.getTableSchema(identifier.namespace()(0),
identifier.name())
+ dorisSchemaToStructType(dorisSchema)
+ })
+
+ override def capabilities(): util.Set[TableCapability] = {
+ val capabilities = mutable.Set(BATCH_READ,
+ BATCH_WRITE,
+ STREAMING_WRITE,
+ TRUNCATE)
+ val properties = config.getSinkProperties
+ val partialColumnsEnabled =
properties.containsKey(DorisOptions.PARTIAL_COLUMNS) &&
"true".equalsIgnoreCase(properties.get(DorisOptions.PARTIAL_COLUMNS))
+ val schemaLessEnabled =
config.getValue(DorisOptions.DORIS_WRITE_SCHEMA_LESS)
+ if (partialColumnsEnabled || schemaLessEnabled) {
+ capabilities += ACCEPT_ANY_SCHEMA
+ }
+ capabilities.asJava
+ }
+
+ override def newScanBuilder(caseInsensitiveStringMap:
CaseInsensitiveStringMap): ScanBuilder = {
+ config.setProperty(DorisOptions.DORIS_TABLE_IDENTIFIER, name())
+ createScanBuilder(config, schema())
+ }
+
+ override def newWriteBuilder(logicalWriteInfo: LogicalWriteInfo):
WriteBuilder = {
+ config.setProperty(DorisOptions.DORIS_TABLE_IDENTIFIER, name())
+ createWriteBuilder(config, logicalWriteInfo.schema())
+ }
+
+ private def dorisSchemaToStructType(dorisSchema: Schema): StructType = {
+ val arrayNativeType =
config.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))
+ }).toArray)
+ }
+
+ protected def createScanBuilder(config: DorisConfig, schema: StructType):
ScanBuilder
+
+ protected def createWriteBuilder(config: DorisConfig, schema: StructType):
WriteBuilder
+
+}
diff --git
a/spark-doris-connector/spark-doris-connector-spark-4-base/src/main/scala/org/apache/doris/spark/catalog/DorisTableCatalogBase.scala
b/spark-doris-connector/spark-doris-connector-spark-4-base/src/main/scala/org/apache/doris/spark/catalog/DorisTableCatalogBase.scala
new file mode 100644
index 0000000..2c9ad0f
--- /dev/null
+++
b/spark-doris-connector/spark-doris-connector-spark-4-base/src/main/scala/org/apache/doris/spark/catalog/DorisTableCatalogBase.scala
@@ -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.catalog
+
+import org.apache.doris.spark.client.DorisFrontendClient
+import org.apache.doris.spark.config.{DorisConfig, DorisOptions}
+import org.apache.spark.sql.catalyst.analysis.NoSuchNamespaceException
+import org.apache.spark.sql.connector.catalog.{Identifier, NamespaceChange,
SupportsNamespaces, Table, TableCatalog, TableChange}
+import org.apache.spark.sql.connector.expressions.Transform
+import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
+
+import java.util
+import scala.collection.JavaConverters._
+
+trait DorisTableCatalogBase extends TableCatalog with SupportsNamespaces {
+
+ protected var catalogName: Option[String] = None
+
+ protected var dorisConfig: DorisConfig = _
+
+ protected var frontend: DorisFrontendClient = _
+
+ override def name(): String = {
+ require(catalogName.nonEmpty, "The Doris table catalog is not initialed")
+ catalogName.get
+ }
+
+ override def initialize(name: String, caseInsensitiveStringMap:
CaseInsensitiveStringMap): Unit = {
+ assert(catalogName.isEmpty, "The Doris table catalog is already initialed")
+ catalogName = Some(name)
+ dorisConfig = DorisConfig.fromMap(caseInsensitiveStringMap, true)
+ frontend = new DorisFrontendClient(dorisConfig)
+ }
+
+ override def listTables(namespace: Array[String]): Array[Identifier] = {
+ frontend.listTables(namespace).asScala.map(i => Identifier.of(i.getLeft,
i.getValue)).toArray
+ }
+
+ override def loadTable(identifier: Identifier): Table = {
+ checkIdentifier(identifier)
+ newTableInstance(identifier,
DorisConfig.fromMap((dorisConfig.toMap.asScala +
+ (DorisOptions.DORIS_TABLE_IDENTIFIER.getName ->
getFullTableName(identifier))).asJava, false), None)
+ }
+
+ override def createTable(identifier: Identifier, structType: StructType,
transforms: Array[Transform], map: util.Map[String, String]): Table = throw new
UnsupportedOperationException()
+
+ override def alterTable(identifier: Identifier, tableChanges: TableChange*):
Table = throw new UnsupportedOperationException()
+
+ override def dropTable(identifier: Identifier): Boolean = throw new
UnsupportedOperationException()
+
+ override def renameTable(identifier: Identifier, identifier1: Identifier):
Unit = throw new UnsupportedOperationException()
+
+ override def listNamespaces(): Array[Array[String]] = {
+ frontend.listDatabases().map(Array(_))
+ }
+
+ override def listNamespaces(namespace: Array[String]): Array[Array[String]]
= {
+ namespace match {
+ case Array() => listNamespaces()
+ case Array(_) if frontend.databaseExists(namespace(0)) => Array()
+ case _ => throw new NoSuchNamespaceException(namespace)
+ }
+ }
+
+ override def loadNamespaceMetadata(namespace: Array[String]):
util.Map[String, String] = {
+ namespace match {
+ case Array(database) =>
+ if (!frontend.databaseExists(database)) {
+ throw new NoSuchNamespaceException(namespace)
+ }
+ new util.HashMap[String, String]()
+ case _ => throw new NoSuchNamespaceException(namespace)
+ }
+ }
+
+ override def createNamespace(namespace: Array[String], metadata:
util.Map[String, String]): Unit = throw new UnsupportedOperationException()
+
+ override def alterNamespace(namespace: Array[String], changes:
NamespaceChange*): Unit = throw new UnsupportedOperationException()
+
+ private def checkIdentifier(identifier: Identifier): Unit = {
+ if (identifier.namespace().length > 1) {
+ throw new NoSuchNamespaceException(identifier.namespace())
+ }
+ }
+
+ private def getFullTableName(identifier: Identifier): String = {
+ (identifier.namespace() :+ identifier.name()).map(item =>
s"""`$item`""").mkString(".")
+ }
+
+ def newTableInstance(identifier: Identifier, config: DorisConfig, schema:
Option[StructType]): Table
+
+}
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
new file mode 100644
index 0000000..a03435d
--- /dev/null
+++
b/spark-doris-connector/spark-doris-connector-spark-4-base/src/main/scala/org/apache/doris/spark/catalog/DorisTableProviderBase.scala
@@ -0,0 +1,59 @@
+// 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, Table,
TableProvider}
+import org.apache.spark.sql.connector.expressions.Transform
+import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
+
+import java.util
+
+abstract class DorisTableProviderBase extends TableProvider {
+
+ protected var t: Table = _
+
+ override def inferSchema(options: CaseInsensitiveStringMap): StructType = {
+ if (t == null) t = getTable(options)
+ t.schema()
+ }
+
+ override def getTable(schema: StructType, partitioning: Array[Transform],
properties: util.Map[String, String]): Table = {
+ 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))
+ }
+ }
+
+ private def getTable(options: CaseInsensitiveStringMap): Table = {
+ 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)
+ }
+ }
+
+ def newTableInstance(identifier: Identifier, config: DorisConfig, schema:
Option[StructType]): Table;
+
+}
diff --git
a/spark-doris-connector/spark-doris-connector-spark-4-base/src/main/scala/org/apache/doris/spark/read/AbstractDorisScan.scala
b/spark-doris-connector/spark-doris-connector-spark-4-base/src/main/scala/org/apache/doris/spark/read/AbstractDorisScan.scala
new file mode 100644
index 0000000..4afcc01
--- /dev/null
+++
b/spark-doris-connector/spark-doris-connector-spark-4-base/src/main/scala/org/apache/doris/spark/read/AbstractDorisScan.scala
@@ -0,0 +1,61 @@
+// 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.read
+
+import org.apache.doris.spark.client.entity.{Backend, DorisReaderPartition}
+import org.apache.doris.spark.client.read.ReaderPartitionGenerator
+import org.apache.doris.spark.config.{DorisConfig, DorisOptions}
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.connector.read.{Batch, InputPartition,
PartitionReaderFactory, Scan}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.StructType
+
+import scala.language.implicitConversions
+
+abstract class AbstractDorisScan(config: DorisConfig, schema: StructType)
extends Scan with Batch with Logging {
+
+ private val scanMode =
ScanMode.valueOf(config.getValue(DorisOptions.READ_MODE).toUpperCase)
+
+ override def readSchema(): StructType = schema
+
+ override def toBatch: Batch = this
+
+ override def planInputPartitions(): Array[InputPartition] = {
+ ReaderPartitionGenerator.generatePartitions(config, schema.names,
compiledFilters(), getLimit,
+ SQLConf.get.datetimeJava8ApiEnabled)
+ .map(toInputPartition)
+ }
+
+
+ override def createReaderFactory(): PartitionReaderFactory = {
+ new DorisPartitionReaderFactory(readSchema(), scanMode, config)
+ }
+
+ private def toInputPartition(rp: DorisReaderPartition): DorisInputPartition =
+ DorisInputPartition(rp.getDatabase, rp.getTable, rp.getBackend,
rp.getTablets.map(_.toLong), rp.getOpaquedQueryPlan,
+ rp.getReadColumns, rp.getFilters, rp.getLimit,
rp.getDateTimeJava8APIEnabled)
+
+ protected def compiledFilters(): Array[String]
+
+ protected def getLimit: Int = -1
+
+}
+
+case class DorisInputPartition(database: String, table: String, backend:
Backend, tablets: Array[Long],
+ opaquedQueryPlan: String, readCols:
Array[String], predicates: Array[String],
+ limit: Int = -1, datetimeJava8ApiEnabled:
Boolean) extends InputPartition
diff --git
a/spark-doris-connector/spark-doris-connector-spark-4-base/src/main/scala/org/apache/doris/spark/read/DorisPartitionReader.scala
b/spark-doris-connector/spark-doris-connector-spark-4-base/src/main/scala/org/apache/doris/spark/read/DorisPartitionReader.scala
new file mode 100644
index 0000000..9cf2fd6
--- /dev/null
+++
b/spark-doris-connector/spark-doris-connector-spark-4-base/src/main/scala/org/apache/doris/spark/read/DorisPartitionReader.scala
@@ -0,0 +1,75 @@
+// 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.read
+
+import org.apache.doris.spark.client.entity.DorisReaderPartition
+import org.apache.doris.spark.client.read.{DorisFlightSqlReader, DorisReader,
DorisThriftReader}
+import org.apache.doris.spark.config.DorisConfig
+import org.apache.doris.spark.util.RowConvertors
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.GenericInternalRow
+import org.apache.spark.sql.connector.read.{InputPartition, PartitionReader}
+import org.apache.spark.sql.types.StructType
+
+import scala.language.implicitConversions
+
+class DorisPartitionReader(inputPartition: InputPartition, schema: StructType,
mode: ScanMode, config: DorisConfig)
+ extends PartitionReader[InternalRow] {
+
+ private implicit def toReaderPartition(inputPart: DorisInputPartition):
DorisReaderPartition = {
+ val tablets = inputPart.tablets.map(java.lang.Long.valueOf)
+ new DorisReaderPartition(inputPart.database, inputPart.table,
inputPart.backend, tablets,
+ inputPart.opaquedQueryPlan, inputPart.readCols, inputPart.predicates,
inputPart.limit, config, inputPart.datetimeJava8ApiEnabled)
+ }
+
+ private lazy val reader: DorisReader = {
+ mode match {
+ case ScanMode.THRIFT => new
DorisThriftReader(inputPartition.asInstanceOf[DorisInputPartition])
+ case ScanMode.ARROW => new
DorisFlightSqlReader(inputPartition.asInstanceOf[DorisInputPartition])
+ case _ => throw new UnsupportedOperationException()
+ }
+ }
+
+ private val datetimeJava8ApiEnabled: Boolean =
inputPartition.asInstanceOf[DorisInputPartition].datetimeJava8ApiEnabled
+
+ override def next(): Boolean = reader.hasNext
+
+ override def get(): InternalRow = {
+ val values = reader.next().asInstanceOf[Array[Any]]
+ val row = new GenericInternalRow(schema.length)
+ if (values.nonEmpty) {
+ values.zipWithIndex.foreach {
+ case (value, index) =>
+ if (value == null) row.setNullAt(index)
+ else {
+ val dataType = schema.fields(index).dataType
+ val catalystValue = RowConvertors.convertValue(value, dataType,
datetimeJava8ApiEnabled)
+ row.update(index, catalystValue)
+ }
+ }
+ }
+ row
+ }
+
+ override def close(): Unit = {
+ if (reader != null) {
+ reader.close()
+ }
+ }
+
+}
diff --git
a/spark-doris-connector/spark-doris-connector-spark-4-base/src/main/scala/org/apache/doris/spark/read/DorisPartitionReaderFactory.scala
b/spark-doris-connector/spark-doris-connector-spark-4-base/src/main/scala/org/apache/doris/spark/read/DorisPartitionReaderFactory.scala
new file mode 100644
index 0000000..e91a9c6
--- /dev/null
+++
b/spark-doris-connector/spark-doris-connector-spark-4-base/src/main/scala/org/apache/doris/spark/read/DorisPartitionReaderFactory.scala
@@ -0,0 +1,31 @@
+// 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.read
+
+import org.apache.doris.spark.config.DorisConfig
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.connector.read.{InputPartition, PartitionReader,
PartitionReaderFactory}
+import org.apache.spark.sql.types.StructType
+
+class DorisPartitionReaderFactory(schema: StructType, mode: ScanMode, config:
DorisConfig) extends PartitionReaderFactory {
+
+ override def createReader(inputPartition: InputPartition):
PartitionReader[InternalRow] = {
+ new DorisPartitionReader(inputPartition, schema, mode, config)
+ }
+
+}
\ No newline at end of file
diff --git
a/spark-doris-connector/spark-doris-connector-spark-4-base/src/main/scala/org/apache/doris/spark/read/DorisScan.scala
b/spark-doris-connector/spark-doris-connector-spark-4-base/src/main/scala/org/apache/doris/spark/read/DorisScan.scala
new file mode 100644
index 0000000..d52a82a
--- /dev/null
+++
b/spark-doris-connector/spark-doris-connector-spark-4-base/src/main/scala/org/apache/doris/spark/read/DorisScan.scala
@@ -0,0 +1,33 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.spark.read
+
+import org.apache.doris.spark.config.{DorisConfig, DorisOptions}
+import org.apache.doris.spark.util.DorisDialects
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.sources.Filter
+import org.apache.spark.sql.types.StructType
+
+import scala.language.implicitConversions
+
+class DorisScan(config: DorisConfig, schema: StructType, filters:
Array[Filter]) extends AbstractDorisScan(config, schema) with Logging {
+ override protected def compiledFilters(): Array[String] = {
+ val inValueLengthLimit =
config.getValue(DorisOptions.DORIS_FILTER_QUERY_IN_MAX_COUNT)
+ filters.map(DorisDialects.compileFilter(_,
inValueLengthLimit)).filter(_.isDefined).map(_.get)
+ }
+}
diff --git
a/spark-doris-connector/spark-doris-connector-spark-4-base/src/main/scala/org/apache/doris/spark/read/DorisScanBuilderBase.scala
b/spark-doris-connector/spark-doris-connector-spark-4-base/src/main/scala/org/apache/doris/spark/read/DorisScanBuilderBase.scala
new file mode 100644
index 0000000..94367b2
--- /dev/null
+++
b/spark-doris-connector/spark-doris-connector-spark-4-base/src/main/scala/org/apache/doris/spark/read/DorisScanBuilderBase.scala
@@ -0,0 +1,52 @@
+// 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.read
+
+import org.apache.doris.spark.config.{DorisConfig, DorisOptions}
+import org.apache.spark.sql.connector.read.{ScanBuilder,
SupportsPushDownRequiredColumns}
+import org.apache.spark.sql.types.StructType
+
+protected[spark] abstract class DorisScanBuilderBase(config: DorisConfig,
schema: StructType) extends ScanBuilder
+ with SupportsPushDownRequiredColumns {
+
+ protected var readSchema: StructType = {
+ if (config.contains(DorisOptions.DORIS_READ_FIELDS)) {
+ val dorisReadFields =
config.getValue(DorisOptions.DORIS_READ_FIELDS).split(",").map(_.trim.replaceAll("`",
""))
+ doPruneColumns(schema, dorisReadFields)
+ } else {
+ schema
+ }
+ }
+
+ override def pruneColumns(requiredSchema: StructType): Unit = {
+ readSchema = doPruneColumns(readSchema, requiredSchema.fieldNames)
+ }
+
+ private def doPruneColumns(originSchema: StructType, requiredCols:
Array[String]): StructType = {
+ if (requiredCols.nonEmpty) {
+ val fields = originSchema.fields.filter(
+ field => requiredCols.contains(field.name)
+ )
+ if (fields.isEmpty) {
+ throw new IllegalArgumentException("No required columns found")
+ }
+ StructType(fields)
+ } else originSchema
+ }
+
+}
diff --git
a/spark-doris-connector/spark-doris-connector-spark-4-base/src/main/scala/org/apache/doris/spark/read/ScanMode.java
b/spark-doris-connector/spark-doris-connector-spark-4-base/src/main/scala/org/apache/doris/spark/read/ScanMode.java
new file mode 100644
index 0000000..430b71b
--- /dev/null
+++
b/spark-doris-connector/spark-doris-connector-spark-4-base/src/main/scala/org/apache/doris/spark/read/ScanMode.java
@@ -0,0 +1,22 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.spark.read;
+
+public enum ScanMode {
+ THRIFT,ARROW
+}
diff --git
a/spark-doris-connector/spark-doris-connector-spark-4-base/src/main/scala/org/apache/doris/spark/sql/sources/DorisSourceProvider.scala
b/spark-doris-connector/spark-doris-connector-spark-4-base/src/main/scala/org/apache/doris/spark/sql/sources/DorisSourceProvider.scala
new file mode 100644
index 0000000..5fab64c
--- /dev/null
+++
b/spark-doris-connector/spark-doris-connector-spark-4-base/src/main/scala/org/apache/doris/spark/sql/sources/DorisSourceProvider.scala
@@ -0,0 +1,29 @@
+// 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.sources
+
+import org.apache.spark.sql.SQLContext
+import org.apache.spark.sql.sources.{BaseRelation, RelationProvider}
+
+trait DorisSourceProvider extends RelationProvider {
+
+ override def createRelation(sqlContext: SQLContext, parameters: Map[String,
String]): BaseRelation = {
+ new DorisRelation(sqlContext, parameters)
+ }
+
+}
diff --git
a/spark-doris-connector/spark-doris-connector-spark-4-base/src/main/scala/org/apache/doris/spark/write/DorisDataWriter.scala
b/spark-doris-connector/spark-doris-connector-spark-4-base/src/main/scala/org/apache/doris/spark/write/DorisDataWriter.scala
new file mode 100644
index 0000000..0020c3d
--- /dev/null
+++
b/spark-doris-connector/spark-doris-connector-spark-4-base/src/main/scala/org/apache/doris/spark/write/DorisDataWriter.scala
@@ -0,0 +1,123 @@
+// 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.write
+
+import org.apache.doris.spark.client.write.{CopyIntoProcessor, DorisCommitter,
DorisWriter, StreamLoadProcessor}
+import org.apache.doris.spark.config.{DorisConfig, DorisOptions}
+import org.apache.doris.spark.util.Retry
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.connector.write.{DataWriter, WriterCommitMessage}
+import org.apache.spark.sql.types.StructType
+
+import java.time.Duration
+import java.util.concurrent.locks.LockSupport
+import scala.collection.mutable
+import scala.util.{Failure, Success}
+
+class DorisDataWriter(config: DorisConfig, schema: StructType, partitionId:
Int, taskId: Long, epochId: Long = -1) extends DataWriter[InternalRow] with
Logging {
+
+ private val (writer: DorisWriter[InternalRow], committer: DorisCommitter) =
+ config.getValue(DorisOptions.LOAD_MODE) match {
+ case "stream_load" => (new StreamLoadProcessor(config, schema), new
StreamLoadProcessor(config, schema))
+ case "copy_into" => (new CopyIntoProcessor(config, schema), new
CopyIntoProcessor(config, schema))
+ case mode => throw new IllegalArgumentException("Unsupported load mode:
" + mode)
+ }
+
+ private val batchIntervalMs =
config.getValue(DorisOptions.DORIS_SINK_BATCH_INTERVAL_MS)
+
+ private val retries = config.getValue(DorisOptions.DORIS_SINK_MAX_RETRIES)
+
+ private val retryIntervalMs =
config.getValue(DorisOptions.DORIS_SINK_RETRY_INTERVAL_MS)
+
+ private val twoPhaseCommitEnabled =
config.getValue(DorisOptions.DORIS_SINK_ENABLE_2PC)
+
+ private val committedMessages = mutable.Buffer[String]()
+
+ private lazy val recordBuffer = mutable.Buffer[InternalRow]()
+
+ override def write(record: InternalRow): Unit = loadBatchWithRetries(record)
+
+ override def commit(): WriterCommitMessage = {
+ val txnId = Option(writer.stop())
+ if (twoPhaseCommitEnabled) {
+ if (txnId.isDefined) {
+ committedMessages += txnId.get
+ } else {
+ log.warn("No txn {} to commit batch", txnId)
+ }
+ }
+ DorisWriterCommitMessage(partitionId, taskId, epochId,
committedMessages.toArray)
+ }
+
+ override def abort(): Unit = {
+ if (committedMessages.nonEmpty) {
+ committedMessages.foreach(msg => committer.abort(msg))
+ }
+ close()
+ }
+
+ override def close(): Unit = {
+ if (writer != null) {
+ writer.close()
+ }
+ }
+
+ @throws[Exception]
+ private def loadBatchWithRetries(record: InternalRow): Unit = {
+ var isRetrying = false
+ Retry.exec[Unit, Exception](retries,
Duration.ofMillis(retryIntervalMs.toLong), log) {
+ if (isRetrying) {
+ // retrying, reload data from buffer
+ while (writer.getBatchCount < recordBuffer.size){
+ val idx = writer.getBatchCount
+ writer.load(recordBuffer(idx))
+ }
+ isRetrying = false
+ }
+ if (writer.endOfBatch()) {
+ // end of batch, stop batch write
+ val txnId = Option(writer.stop())
+ if (twoPhaseCommitEnabled) {
+ if (txnId.isDefined) {
+ committedMessages += txnId.get
+ } else {
+ throw new Exception("Failed to end batch write")
+ }
+ }
+ // clear buffer if retry is enabled
+ if (retries > 0) {
+ recordBuffer.clear()
+ }
+ writer.resetBatchCount()
+
LockSupport.parkNanos(Duration.ofMillis(batchIntervalMs.toLong).toNanos)
+ }
+ writer.load(record)
+ } {
+ // batch write failed, set retry flag and reset batch count
+ isRetrying = true
+ writer.resetBatchCount()
+ } match {
+ case Success(_) => if (retries > 0) recordBuffer += record
+ case Failure(exception) => throw new Exception(exception)
+ }
+ }
+
+}
+
+case class DorisWriterCommitMessage(partitionId: Int, taskId: Long, epochId:
Long, commitMessages: Array[String]) extends WriterCommitMessage
diff --git
a/spark-doris-connector/spark-doris-connector-spark-4-base/src/main/scala/org/apache/doris/spark/write/DorisDataWriterFactory.scala
b/spark-doris-connector/spark-doris-connector-spark-4-base/src/main/scala/org/apache/doris/spark/write/DorisDataWriterFactory.scala
new file mode 100644
index 0000000..2afd641
--- /dev/null
+++
b/spark-doris-connector/spark-doris-connector-spark-4-base/src/main/scala/org/apache/doris/spark/write/DorisDataWriterFactory.scala
@@ -0,0 +1,37 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.spark.write
+
+import org.apache.doris.spark.config.DorisConfig
+import org.apache.spark.sql.catalyst.InternalRow
+import
org.apache.spark.sql.connector.write.streaming.StreamingDataWriterFactory
+import org.apache.spark.sql.connector.write.{DataWriter, DataWriterFactory}
+import org.apache.spark.sql.types.StructType
+
+class DorisDataWriterFactory(config: DorisConfig, schema: StructType) extends
DataWriterFactory with StreamingDataWriterFactory {
+
+ // for batch write
+ override def createWriter(partitionId: Int, taskId: Long):
DataWriter[InternalRow] = {
+ new DorisDataWriter(config, schema, partitionId, taskId)
+ }
+
+ // for streaming write
+ override def createWriter(partitionId: Int, taskId: Long, epochId: Long):
DataWriter[InternalRow] = {
+ new DorisDataWriter(config, schema, partitionId, taskId, epochId)
+ }
+}
diff --git
a/spark-doris-connector/spark-doris-connector-spark-4-base/src/main/scala/org/apache/doris/spark/write/DorisWrite.scala
b/spark-doris-connector/spark-doris-connector-spark-4-base/src/main/scala/org/apache/doris/spark/write/DorisWrite.scala
new file mode 100644
index 0000000..52233a7
--- /dev/null
+++
b/spark-doris-connector/spark-doris-connector-spark-4-base/src/main/scala/org/apache/doris/spark/write/DorisWrite.scala
@@ -0,0 +1,91 @@
+// 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.write
+
+import org.apache.doris.spark.client.write.{CopyIntoProcessor, DorisCommitter,
StreamLoadProcessor}
+import org.apache.doris.spark.config.{DorisConfig, DorisOptions}
+import
org.apache.spark.sql.connector.write.streaming.{StreamingDataWriterFactory,
StreamingWrite}
+import org.apache.spark.sql.connector.write.{BatchWrite, DataWriterFactory,
PhysicalWriteInfo, WriterCommitMessage}
+import org.apache.spark.sql.types.StructType
+import org.slf4j.LoggerFactory
+
+class DorisWrite(config: DorisConfig, schema: StructType) extends BatchWrite
with StreamingWrite {
+
+ private val LOG = LoggerFactory.getLogger(classOf[DorisWrite])
+
+ private val committer: DorisCommitter =
config.getValue(DorisOptions.LOAD_MODE) match {
+ case "stream_load" => new StreamLoadProcessor(config, schema)
+ case "copy_into" => new CopyIntoProcessor(config, schema)
+ case _ => throw new IllegalArgumentException()
+ }
+
+ private var lastCommittedEpoch: Option[Long] = None
+
+ private val committedEpochLock = new AnyRef
+
+ override def createBatchWriterFactory(physicalWriteInfo: PhysicalWriteInfo):
DataWriterFactory = {
+ new DorisDataWriterFactory(config, schema)
+ }
+
+ // for batch write
+ override def commit(writerCommitMessages: Array[WriterCommitMessage]): Unit
= {
+ if (writerCommitMessages != null && writerCommitMessages.nonEmpty) {
+ writerCommitMessages.filter(_ != null)
+
.foreach(_.asInstanceOf[DorisWriterCommitMessage].commitMessages.foreach(committer.commit))
+ }
+ }
+
+ // for batch write
+ override def abort(writerCommitMessages: Array[WriterCommitMessage]): Unit =
{
+ if (writerCommitMessages != null && writerCommitMessages.nonEmpty) {
+ LOG.info("writerCommitMessages size: " + writerCommitMessages.length)
+ writerCommitMessages.filter(_ != null)
+
.foreach(_.asInstanceOf[DorisWriterCommitMessage].commitMessages.foreach(committer.abort))
+ }
+ }
+
+ override def useCommitCoordinator(): Boolean = true
+
+ override def createStreamingWriterFactory(physicalWriteInfo:
PhysicalWriteInfo): StreamingDataWriterFactory = {
+ new DorisDataWriterFactory(config, schema)
+ }
+
+ // for streaming write
+ override def commit(epochId: Long, writerCommitMessages:
Array[WriterCommitMessage]): Unit = {
+ committedEpochLock.synchronized {
+ if ((lastCommittedEpoch.isEmpty || epochId > lastCommittedEpoch.get)
+ && writerCommitMessages != null && writerCommitMessages.exists(_ !=
null)) {
+ writerCommitMessages.filter(_ != null)
+
.foreach(_.asInstanceOf[DorisWriterCommitMessage].commitMessages.foreach(committer.commit))
+ lastCommittedEpoch = Some(epochId)
+ }
+ }
+ }
+
+ // for streaming write
+ override def abort(epochId: Long, writerCommitMessages:
Array[WriterCommitMessage]): Unit = {
+ committedEpochLock.synchronized {
+ if ((lastCommittedEpoch.isEmpty || epochId > lastCommittedEpoch.get)
+ && writerCommitMessages != null && writerCommitMessages.exists(_ !=
null)) {
+ writerCommitMessages.filter(_ != null)
+
.foreach(_.asInstanceOf[DorisWriterCommitMessage].commitMessages.foreach(committer.abort))
+ }
+ }
+ }
+
+}
\ No newline at end of file
diff --git
a/spark-doris-connector/spark-doris-connector-spark-4-base/src/main/scala/org/apache/doris/spark/write/DorisWriteBuilder.scala
b/spark-doris-connector/spark-doris-connector-spark-4-base/src/main/scala/org/apache/doris/spark/write/DorisWriteBuilder.scala
new file mode 100644
index 0000000..b74cc45
--- /dev/null
+++
b/spark-doris-connector/spark-doris-connector-spark-4-base/src/main/scala/org/apache/doris/spark/write/DorisWriteBuilder.scala
@@ -0,0 +1,47 @@
+// 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.write
+
+import org.apache.doris.spark.client.DorisFrontendClient
+import org.apache.doris.spark.config.{DorisConfig, DorisOptions}
+import org.apache.spark.sql.connector.write.streaming.StreamingWrite
+import org.apache.spark.sql.connector.write.{BatchWrite, SupportsTruncate,
WriteBuilder}
+import org.apache.spark.sql.types.StructType
+
+class DorisWriteBuilder(config: DorisConfig, schema: StructType) extends
WriteBuilder with SupportsTruncate {
+
+ private var isTruncate = false
+
+ override def buildForBatch(): BatchWrite = {
+ if (isTruncate) {
+ val client = new DorisFrontendClient(config)
+ val tableDb =
config.getValue(DorisOptions.DORIS_TABLE_IDENTIFIER).split("\\.")
+ client.truncateTable(tableDb(0), tableDb(1))
+ }
+ new DorisWrite(config, schema)
+ }
+
+ override def buildForStreaming(): StreamingWrite = {
+ new DorisWrite(config, schema)
+ }
+
+ override def truncate(): WriteBuilder = {
+ isTruncate = true
+ this
+ }
+}
diff --git a/spark-doris-connector/spark-doris-connector-spark-4.1/pom.xml
b/spark-doris-connector/spark-doris-connector-spark-4.1/pom.xml
new file mode 100644
index 0000000..4d8437b
--- /dev/null
+++ b/spark-doris-connector/spark-doris-connector-spark-4.1/pom.xml
@@ -0,0 +1,58 @@
+<?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
http://maven.apache.org/xsd/maven-4.0.0.xsd";>
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.doris</groupId>
+ <artifactId>spark-doris-connector</artifactId>
+ <version>${revision}</version>
+ </parent>
+
+ <artifactId>spark-doris-connector-spark-4.1</artifactId>
+
+ <properties>
+ <maven.compiler.source>17</maven.compiler.source>
+ <maven.compiler.target>17</maven.compiler.target>
+ <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+ <spark.version>4.1.0</spark.version>
+ <spark.major.version>4.1</spark.major.version>
+ <scala.version>2.13.17</scala.version>
+ <scala.major.version>2.13</scala.major.version>
+ </properties>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.doris</groupId>
+ <artifactId>spark-doris-connector-spark-4-base</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.spark</groupId>
+ <artifactId>spark-sql_${scala.major.version}</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.junit.jupiter</groupId>
+ <artifactId>junit-jupiter-api</artifactId>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+
+</project>
\ No newline at end of file
diff --git
a/spark-doris-connector/spark-doris-connector-spark-4.1/src/main/resources/META-INF/services/org.apache.spark.sql.sources.DataSourceRegister
b/spark-doris-connector/spark-doris-connector-spark-4.1/src/main/resources/META-INF/services/org.apache.spark.sql.sources.DataSourceRegister
new file mode 100644
index 0000000..986b253
--- /dev/null
+++
b/spark-doris-connector/spark-doris-connector-spark-4.1/src/main/resources/META-INF/services/org.apache.spark.sql.sources.DataSourceRegister
@@ -0,0 +1,18 @@
+# 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.
+
+org.apache.doris.spark.sql.sources.DorisDataSource
diff --git
a/spark-doris-connector/spark-doris-connector-spark-4.1/src/main/scala/org/apache/doris/spark/catalog/DorisTable.scala
b/spark-doris-connector/spark-doris-connector-spark-4.1/src/main/scala/org/apache/doris/spark/catalog/DorisTable.scala
new file mode 100644
index 0000000..be461af
--- /dev/null
+++
b/spark-doris-connector/spark-doris-connector-spark-4.1/src/main/scala/org/apache/doris/spark/catalog/DorisTable.scala
@@ -0,0 +1,34 @@
+// 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
+import org.apache.doris.spark.read.DorisScanBuilder
+import org.apache.doris.spark.write.DorisWriteBuilder
+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
+
+class DorisTable(identifier: Identifier, config: DorisConfig, schema:
Option[StructType])
+ extends DorisTableBase(identifier, config, schema) {
+
+ override def createScanBuilder(config: DorisConfig, schema: StructType):
ScanBuilder = new DorisScanBuilder(config, schema)
+ override protected def createWriteBuilder(config: DorisConfig, schema:
StructType): WriteBuilder = new DorisWriteBuilder(config, schema)
+
+}
diff --git
a/spark-doris-connector/spark-doris-connector-spark-4.1/src/main/scala/org/apache/doris/spark/catalog/DorisTableCatalog.scala
b/spark-doris-connector/spark-doris-connector-spark-4.1/src/main/scala/org/apache/doris/spark/catalog/DorisTableCatalog.scala
new file mode 100644
index 0000000..e3ffcc9
--- /dev/null
+++
b/spark-doris-connector/spark-doris-connector-spark-4.1/src/main/scala/org/apache/doris/spark/catalog/DorisTableCatalog.scala
@@ -0,0 +1,28 @@
+// 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
+import org.apache.spark.sql.connector.catalog.{Identifier, Table}
+import org.apache.spark.sql.types.StructType
+
+class DorisTableCatalog extends DorisTableCatalogBase {
+ override def dropNamespace(strings: Array[String], b: Boolean): Boolean =
throw new UnsupportedOperationException()
+ override def newTableInstance(identifier: Identifier, config: DorisConfig,
schema: Option[StructType]): Table =
+ new DorisTable(identifier, config, schema)
+}
diff --git
a/spark-doris-connector/spark-doris-connector-spark-4.1/src/main/scala/org/apache/doris/spark/read/DorisScanBuilder.scala
b/spark-doris-connector/spark-doris-connector-spark-4.1/src/main/scala/org/apache/doris/spark/read/DorisScanBuilder.scala
new file mode 100644
index 0000000..6bdd256
--- /dev/null
+++
b/spark-doris-connector/spark-doris-connector-spark-4.1/src/main/scala/org/apache/doris/spark/read/DorisScanBuilder.scala
@@ -0,0 +1,53 @@
+// 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.read
+
+import org.apache.doris.spark.config.{DorisConfig, DorisOptions}
+import org.apache.doris.spark.read.expression.V2ExpressionBuilder
+import org.apache.spark.sql.connector.expressions.filter.Predicate
+import org.apache.spark.sql.connector.read.{Scan, SupportsPushDownLimit,
SupportsPushDownV2Filters}
+import org.apache.spark.sql.types.StructType
+
+class DorisScanBuilder(config: DorisConfig, schema: StructType) extends
DorisScanBuilderBase(config, schema)
+ with SupportsPushDownV2Filters
+ with SupportsPushDownLimit {
+
+ private var pushDownPredicates: Array[Predicate] = Array[Predicate]()
+
+ private val expressionBuilder = new
V2ExpressionBuilder(config.getValue(DorisOptions.DORIS_FILTER_QUERY_IN_MAX_COUNT))
+
+ private var limitSize: Int = -1
+
+ override def build(): Scan = new DorisScanV2(config, readSchema,
pushDownPredicates, limitSize)
+
+ override def pushPredicates(predicates: Array[Predicate]): Array[Predicate]
= {
+ val (pushed, unsupported) = predicates.partition(predicate => {
+ expressionBuilder.buildOpt(predicate).isDefined
+ })
+ this.pushDownPredicates = pushed
+ unsupported
+ }
+
+ override def pushedPredicates(): Array[Predicate] = pushDownPredicates
+
+ override def pushLimit(i: Int): Boolean = {
+ limitSize = i
+ true
+ }
+
+}
diff --git
a/spark-doris-connector/spark-doris-connector-spark-4.1/src/main/scala/org/apache/doris/spark/read/DorisScanV2.scala
b/spark-doris-connector/spark-doris-connector-spark-4.1/src/main/scala/org/apache/doris/spark/read/DorisScanV2.scala
new file mode 100644
index 0000000..5e0104b
--- /dev/null
+++
b/spark-doris-connector/spark-doris-connector-spark-4.1/src/main/scala/org/apache/doris/spark/read/DorisScanV2.scala
@@ -0,0 +1,34 @@
+// 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.read
+
+import org.apache.doris.spark.config.{DorisConfig, DorisOptions}
+import org.apache.doris.spark.read.expression.V2ExpressionBuilder
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.connector.expressions.filter.Predicate
+import org.apache.spark.sql.types.StructType
+
+class DorisScanV2(config: DorisConfig, schema: StructType, filters:
Array[Predicate], limit: Int) extends AbstractDorisScan(config, schema) with
Logging {
+ override protected def compiledFilters(): Array[String] = {
+ val inValueLengthLimit =
config.getValue(DorisOptions.DORIS_FILTER_QUERY_IN_MAX_COUNT)
+ val v2ExpressionBuilder = new V2ExpressionBuilder(inValueLengthLimit)
+ filters.map(e =>
v2ExpressionBuilder.buildOpt(e)).filter(_.isDefined).map(_.get)
+ }
+
+ override protected def getLimit: Int = limit
+}
diff --git
a/spark-doris-connector/spark-doris-connector-spark-4.1/src/main/scala/org/apache/doris/spark/read/expression/V2ExpressionBuilder.scala
b/spark-doris-connector/spark-doris-connector-spark-4.1/src/main/scala/org/apache/doris/spark/read/expression/V2ExpressionBuilder.scala
new file mode 100644
index 0000000..9a86a69
--- /dev/null
+++
b/spark-doris-connector/spark-doris-connector-spark-4.1/src/main/scala/org/apache/doris/spark/read/expression/V2ExpressionBuilder.scala
@@ -0,0 +1,120 @@
+// 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.read.expression
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.catalyst.util.DateTimeUtils
+import org.apache.spark.sql.connector.expressions.filter.{AlwaysFalse,
AlwaysTrue, And, Not, Or}
+import org.apache.spark.sql.connector.expressions.{Expression,
GeneralScalarExpression, Literal, NamedReference}
+import org.apache.spark.sql.types.{DateType, TimestampType}
+
+import scala.util.{Failure, Success, Try}
+
+class V2ExpressionBuilder(inValueLengthLimit: Int) extends Logging {
+
+ def buildOpt(predicate: Expression): Option[String] = {
+ Try {
+ Some(build(predicate))
+ } match {
+ case Success(value) => value
+ case Failure(exception) =>
+ logWarning(s"Failed to build expression: ${predicate.toString}, and
not support predicate push down, errMsg is ${exception.getMessage}")
+ None
+ }
+ }
+
+ def build(predicate: Expression): String = {
+ predicate match {
+ case and: And => s"(${build(and.left())} AND ${build(and.right())})"
+ case or: Or => s"(${build(or.left())} OR ${build(or.right())})"
+ case not: Not =>
+ not.child().name() match {
+ case "IS_NULL" => build(new GeneralScalarExpression("IS_NOT_NULL",
not.children()(0).children()))
+ case "=" => build(new GeneralScalarExpression("!=",
not.children()(0).children()))
+ case _ => s"NOT (${build(not.child())})"
+ }
+ case _: AlwaysTrue => "1=1"
+ case _: AlwaysFalse => "1=0"
+ case expr: Expression =>
+ expr match {
+ case literal: Literal[_] => visitLiteral(literal)
+ case namedRef: NamedReference => s"`${namedRef.toString}`"
+ case e: GeneralScalarExpression => e.name() match {
+ case "IN" =>
+ val expressions = e.children()
+ if (expressions.nonEmpty && expressions.length <=
inValueLengthLimit) {
+ s"""${build(expressions(0))} IN (${expressions.slice(1,
expressions.length).map(build).mkString(",")})"""
+ } else throw new IllegalArgumentException(s"exceeding limit of
IN values: actual size ${expressions.length}, limit size $inValueLengthLimit")
+ case "IS_NULL" => s"${build(e.children()(0))} IS NULL"
+ case "IS_NOT_NULL" => s"${build(e.children()(0))} IS NOT NULL"
+ case "STARTS_WITH" => visitStartWith(build(e.children()(0)),
build(e.children()(1)));
+ case "ENDS_WITH" => visitEndWith(build(e.children()(0)),
build(e.children()(1)));
+ case "CONTAINS" => visitContains(build(e.children()(0)),
build(e.children()(1)));
+ case "=" => s"${build(e.children()(0))} =
${build(e.children()(1))}"
+ case "!=" | "<>" => s"${build(e.children()(0))} !=
${build(e.children()(1))}"
+ case "<" => s"${build(e.children()(0))} <
${build(e.children()(1))}"
+ case "<=" => s"${build(e.children()(0))} <=
${build(e.children()(1))}"
+ case ">" => s"${build(e.children()(0))} >
${build(e.children()(1))}"
+ case ">=" => s"${build(e.children()(0))} >=
${build(e.children()(1))}"
+ case "CASE_WHEN" =>
+ val fragment = new StringBuilder("CASE ")
+ val expressions = e.children()
+
+ for(i<- 0 until expressions.size - 1 by 2){
+ fragment.append(s" WHEN ${build(expressions(i))} THEN
${build(expressions(i+1))} ")
+ }
+
+ if (expressions.length % 2 != 0) {
+ val last = expressions(expressions.length - 1)
+ fragment.append(s" ELSE ${build(last)} ")
+ }
+ fragment.append(" END")
+
+ fragment.mkString
+ case _ => throw new IllegalArgumentException(s"Unsupported
expression: ${e.name()}")
+ }
+ }
+ }
+ }
+
+ def visitLiteral(literal: Literal[_]): String = {
+ if (literal.value() == null) {
+ return "null"
+ }
+ literal.dataType() match {
+ case DateType =>
s"'${DateTimeUtils.toJavaDate(literal.value().asInstanceOf[Int]).toString}'"
+ case TimestampType =>
s"'${DateTimeUtils.toJavaTimestamp(literal.value().asInstanceOf[Long]).toString}'"
+ case _ => literal.toString
+ }
+ }
+ def visitStartWith(l: String, r: String): String = {
+ val value = r.substring(1, r.length - 1)
+ s"$l LIKE '$value%'"
+ }
+
+ def visitEndWith(l: String, r: String): String = {
+ val value = r.substring(1, r.length - 1)
+ s"$l LIKE '%$value'"
+ }
+
+ def visitContains(l: String, r: String): String = {
+ val value = r.substring(1, r.length - 1)
+ s"$l LIKE '%$value%'"
+ }
+
+}
diff --git
a/spark-doris-connector/spark-doris-connector-spark-4.1/src/main/scala/org/apache/doris/spark/sql/sources/DorisDataSource.scala
b/spark-doris-connector/spark-doris-connector-spark-4.1/src/main/scala/org/apache/doris/spark/sql/sources/DorisDataSource.scala
new file mode 100644
index 0000000..beae22b
--- /dev/null
+++
b/spark-doris-connector/spark-doris-connector-spark-4.1/src/main/scala/org/apache/doris/spark/sql/sources/DorisDataSource.scala
@@ -0,0 +1,30 @@
+// 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.sources
+
+import org.apache.doris.spark.catalog.{DorisTable, DorisTableProviderBase}
+import org.apache.doris.spark.config.DorisConfig
+import org.apache.spark.sql.connector.catalog.{Identifier, Table}
+import org.apache.spark.sql.types.StructType
+
+class DorisDataSource extends DorisTableProviderBase with
DorisSourceRegisterTrait with DorisSourceProvider with Serializable {
+
+ override def newTableInstance(identifier: Identifier, config: DorisConfig,
schema: Option[StructType]): Table =
+ new DorisTable(identifier, config, schema)
+
+}
diff --git
a/spark-doris-connector/spark-doris-connector-spark-4.1/src/test/scala/org/apache/doris/spark/read/expression/V2ExpressionBuilderTest.scala
b/spark-doris-connector/spark-doris-connector-spark-4.1/src/test/scala/org/apache/doris/spark/read/expression/V2ExpressionBuilderTest.scala
new file mode 100644
index 0000000..7691b29
--- /dev/null
+++
b/spark-doris-connector/spark-doris-connector-spark-4.1/src/test/scala/org/apache/doris/spark/read/expression/V2ExpressionBuilderTest.scala
@@ -0,0 +1,65 @@
+package org.apache.doris.spark.read.expression
+
+// 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.
+
+import org.apache.spark.sql.ExpressionUtil
+import org.apache.spark.sql.sources._
+import org.junit.jupiter.api.{Assertions, Test}
+
+class V2ExpressionBuilderTest {
+
+ @Test
+ def buildTest(): Unit = {
+
+ val builder = new V2ExpressionBuilder(10)
+ Assertions.assertEquals(builder.build(EqualTo("c0", 1).toV2), "`c0` = 1")
+ Assertions.assertEquals(builder.build(Not(EqualTo("c1", 2)).toV2), "`c1`
!= 2")
+ Assertions.assertEquals(builder.build(GreaterThan("c2", 3.4).toV2), "`c2`
> 3.4")
+ Assertions.assertEquals(builder.build(GreaterThanOrEqual("c3", 5.6).toV2),
"`c3` >= 5.6")
+ Assertions.assertEquals(builder.build(LessThan("c4", 7.8).toV2), "`c4` <
7.8")
+ Assertions.assertEquals(builder.build(LessThanOrEqual("c5",
BigDecimal(9.1011)).toV2), "`c5` <= 9.1011")
+ Assertions.assertEquals(builder.build(StringStartsWith("c6","a").toV2),
"`c6` LIKE 'a%'")
+ Assertions.assertEquals(builder.build(StringEndsWith("c7", "b").toV2),
"`c7` LIKE '%b'")
+ Assertions.assertEquals(builder.build(StringContains("c8", "c").toV2),
"`c8` LIKE '%c%'")
+ Assertions.assertEquals(builder.build(In("c9", Array(12,13,14)).toV2),
"`c9` IN (12,13,14)")
+ Assertions.assertEquals(builder.build(IsNull("c10").toV2), "`c10` IS NULL")
+ Assertions.assertEquals(builder.build(Not(IsNull("c11")).toV2), "`c11` IS
NOT NULL")
+ Assertions.assertEquals(builder.build(And(EqualTo("c12", 15),
EqualTo("c13", 16)).toV2), "(`c12` = 15 AND `c13` = 16)")
+ Assertions.assertEquals(builder.build(Or(EqualTo("c14", 17),
EqualTo("c15", 18)).toV2), "(`c14` = 17 OR `c15` = 18)")
+ Assertions.assertEquals(builder.build(AlwaysTrue.toV2), "1=1")
+ Assertions.assertEquals(builder.build(AlwaysFalse.toV2), "1=0")
+ Assertions.assertEquals(builder.build(In("c19",
Array(19,20,21,22,23,24,25,26)).toV2), "`c19` IN (19,20,21,22,23,24,25,26)")
+ Assertions.assertEquals(builder.build(In("c19", Array("19","20")).toV2),
"`c19` IN ('19','20')")
+ val inException =
Assertions.assertThrows(classOf[IllegalArgumentException], () =>
builder.build(In("c19", Array(19,20,21,22,23,24,25,26,27,28,29)).toV2))
+ Assertions.assertEquals(inException.getMessage, "exceeding limit of IN
values: actual size 12, limit size 10")
+ val exception = Assertions.assertThrows(classOf[IllegalArgumentException],
() => builder.build(ExpressionUtil.buildCoalesceFilter()))
+ Assertions.assertEquals(exception.getMessage, "Unsupported expression:
COALESCE")
+
+ }
+
+ @Test
+ def buildOptTest() : Unit = {
+
+ val builder = new V2ExpressionBuilder(10)
+ Assertions.assertEquals(builder.buildOpt(EqualTo("c0", 1).toV2),
Some("`c0` = 1"))
+ Assertions.assertEquals(builder.buildOpt(Not(EqualTo("c1", 2)).toV2),
Some("`c1` != 2"))
+ Assertions.assertEquals(builder.buildOpt(GreaterThan("c2", 3.4).toV2),
Some("`c2` > 3.4"))
+
Assertions.assertEquals(builder.buildOpt(ExpressionUtil.buildCoalesceFilter()),
None)
+ }
+
+}
diff --git
a/spark-doris-connector/spark-doris-connector-spark-4.1/src/test/scala/org/apache/spark/sql/ExpressionUtil.scala
b/spark-doris-connector/spark-doris-connector-spark-4.1/src/test/scala/org/apache/spark/sql/ExpressionUtil.scala
new file mode 100644
index 0000000..2da931a
--- /dev/null
+++
b/spark-doris-connector/spark-doris-connector-spark-4.1/src/test/scala/org/apache/spark/sql/ExpressionUtil.scala
@@ -0,0 +1,30 @@
+// 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.spark.sql
+
+import org.apache.spark.sql.connector.expressions.filter.Predicate
+import org.apache.spark.sql.connector.expressions.{Expression, FieldReference,
GeneralScalarExpression, LiteralValue}
+import org.apache.spark.sql.types.StringType
+
+object ExpressionUtil {
+
+ def buildCoalesceFilter(): Expression = {
+ val gse = new GeneralScalarExpression("COALESCE",
Array(FieldReference(Seq("A4")), LiteralValue("null", StringType)))
+ new Predicate("=", Array(gse, LiteralValue("1", StringType)))
+ }
+}
diff --git a/tools/releasing/deploy_staging_jars.sh
b/tools/releasing/deploy_staging_jars.sh
index 7d4b4b9..9eee7b9 100644
--- a/tools/releasing/deploy_staging_jars.sh
+++ b/tools/releasing/deploy_staging_jars.sh
@@ -17,7 +17,13 @@
# under the License.
##############################################################
-# This script is deploy stage jars to repository.apache.org
+# This script deploys the JDK 8 / Scala 2.11-2.12 stage jars
+# (Spark 2.4 - 3.5) to repository.apache.org.
+#
+# Spark 4.x requires JDK 17 and Scala 2.13, so it is deployed
+# by a separate script (deploy_staging_jars_spark4.sh). Keeping
+# the two toolchains in separate scripts lets either release
+# flow be rerun on its own if a JDK-switching issue occurs.
##############################################################
MVN=${MVN:-mvn}
@@ -61,5 +67,5 @@ ${MVN} clean deploy -Papache-release -DskipTests
-DretryFailedDeploymentCount=10
echo "Deploying spark3.5..."
${MVN} clean deploy -Papache-release -DskipTests
-DretryFailedDeploymentCount=10 -Pspark-3.5 -pl spark-doris-connector-spark-3.5
-am
-echo "Deploy jar finished."
+echo "Deploy jar finished. Run deploy_staging_jars_spark4.sh (with JDK 17) to
deploy Spark 4.x."
cd ${CURR_DIR}
\ No newline at end of file
diff --git a/tools/releasing/deploy_staging_jars.sh
b/tools/releasing/deploy_staging_jars_spark4.sh
similarity index 52%
copy from tools/releasing/deploy_staging_jars.sh
copy to tools/releasing/deploy_staging_jars_spark4.sh
index 7d4b4b9..97f605f 100644
--- a/tools/releasing/deploy_staging_jars.sh
+++ b/tools/releasing/deploy_staging_jars_spark4.sh
@@ -17,7 +17,15 @@
# under the License.
##############################################################
-# This script is deploy stage jars to repository.apache.org
+# This script deploys the Spark 4.x stage jars (JDK 17,
+# Scala 2.13) to repository.apache.org.
+#
+# It is kept separate from deploy_staging_jars.sh (Spark 2.4 -
+# 3.5, JDK 8) so the two toolchains never share a run: either
+# release flow can be rerun on its own if a JDK-switching issue
+# occurs. This script fails fast when JDK 17 is not in effect
+# rather than silently falling back to JDK 8 and breaking the
+# build.
##############################################################
MVN=${MVN:-mvn}
@@ -27,6 +35,8 @@ CUSTOM_OPTIONS=${CUSTOM_OPTIONS:-}
set -o errexit
set -o nounset
+# PLACEHOLDER_BODY
+
CURR_DIR=`pwd`
BASE_DIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" >/dev/null && pwd )"
PROJECT_ROOT="${BASE_DIR}/../../"
@@ -37,29 +47,34 @@ if [ ! -f ${PROJECT_ROOT}/LICENSE.txt ]; then
exit 1
fi
+# Spark 4.x requires JDK 17. Prefer JAVA17_HOME when set; otherwise the
current JAVA_HOME must be
+# JDK 17. Fail fast on a wrong or missing toolchain instead of silently
building with JDK 8.
+if [ -n "${JAVA17_HOME:-}" ]; then
+ export JAVA_HOME="${JAVA17_HOME}"
+fi
+if [ -z "${JAVA_HOME:-}" ]; then
+ echo "JDK 17 is required to deploy Spark 4.x. Set JAVA17_HOME (or
JAVA_HOME) to a JDK 17 home."
+ exit 1
+fi
+JAVA_VERSION=$("${JAVA_HOME}/bin/java" -version 2>&1 | awk -F'"' '/version/
{print $2}')
+case "${JAVA_VERSION}" in
+ 17.*|18.*|19.*|2[0-9].*)
+ ;;
+ *)
+ echo "JDK 17 or later is required to deploy Spark 4.x, but JAVA_HOME
points to Java ${JAVA_VERSION}."
+ echo "Set JAVA17_HOME (or JAVA_HOME) to a JDK 17 home and rerun this
script."
+ exit 1
+ ;;
+esac
+
###########################
cd ${PROJECT_ROOT}/spark-doris-connector
-echo "Deploying to repository.apache.org"
-
-echo "Deploying spark2.4..."
-${MVN} clean deploy -Papache-release -DskipTests
-DretryFailedDeploymentCount=10 -Pspark-2.4_2.11 -pl
spark-doris-connector-spark-2 -am
-
-echo "Deploying spark3.1..."
-${MVN} clean deploy -Papache-release -DskipTests
-DretryFailedDeploymentCount=10 -Pspark-3.1 -pl spark-doris-connector-spark-3.1
-am
-
-echo "Deploying spark3.2..."
-${MVN} clean deploy -Papache-release -DskipTests
-DretryFailedDeploymentCount=10 -Pspark-3.2 -pl spark-doris-connector-spark-3.2
-am
-
-echo "Deploying spark3.3..."
-${MVN} clean deploy -Papache-release -DskipTests
-DretryFailedDeploymentCount=10 -Pspark-3.3 -pl spark-doris-connector-spark-3.3
-am
-
-echo "Deploying spark3.4..."
-${MVN} clean deploy -Papache-release -DskipTests
-DretryFailedDeploymentCount=10 -Pspark-3.4 -pl spark-doris-connector-spark-3.4
-am
+echo "Deploying to repository.apache.org with JDK ${JAVA_VERSION}
(JAVA_HOME=${JAVA_HOME})"
-echo "Deploying spark3.5..."
-${MVN} clean deploy -Papache-release -DskipTests
-DretryFailedDeploymentCount=10 -Pspark-3.5 -pl spark-doris-connector-spark-3.5
-am
+echo "Deploying spark4.1..."
+${MVN} clean deploy -Papache-release -DskipTests
-DretryFailedDeploymentCount=10 -Pspark-4.1 -pl spark-doris-connector-spark-4.1
-am
echo "Deploy jar finished."
-cd ${CURR_DIR}
\ No newline at end of file
+cd ${CURR_DIR}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
