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 ed5342a [Feature] Add S3 TVF batch sink support (#371)
ed5342a is described below
commit ed5342a5157d694e640409a5b27496a62ea0a398
Author: wudi <[email protected]>
AuthorDate: Mon Aug 24 11:26:26 2026 +0800
[Feature] Add S3 TVF batch sink support (#371)
Add a TVF batch sink write mode that stages JSON files in S3-compatible
object storage and loads them into Doris through INSERT INTO ... SELECT FROM
S3(...).
Each Spark task uploads one or more files and submits one synchronous
INSERT containing all files produced by that task. A batch UUID and partition
ID are included in object keys and labels so Spark task retries use independent
files and labels. The first version provides at-least-once semantics and does
not support Structured Streaming.
---
.github/workflows/run-e2ecase.yml | 24 +-
.github/workflows/run-itcase.yml | 25 +-
spark-doris-connector/pom.xml | 9 +
.../spark-doris-connector-base/pom.xml | 21 ++
.../doris/spark/client/DorisFrontendClient.java | 35 +++
.../client/write/tvf/JdbcS3TvfLoadClient.java | 76 +++++
.../client/write/tvf/S3ClientObjectStore.java | 83 ++++++
.../spark/client/write/tvf/S3ObjectStore.java | 26 ++
.../spark/client/write/tvf/S3TvfColumnUtils.java | 76 +++++
.../spark/client/write/tvf/S3TvfCommittable.java | 68 +++++
.../spark/client/write/tvf/S3TvfCommitter.java | 94 ++++++
.../spark/client/write/tvf/S3TvfLoadClient.java | 27 ++
.../spark/client/write/tvf/S3TvfSqlBuilder.java | 95 ++++++
.../doris/spark/client/write/tvf/S3TvfWriter.java | 178 ++++++++++++
.../doris/spark/client/write/tvf/TvfSqlUtils.java | 30 ++
.../org/apache/doris/spark/config/DorisConfig.java | 4 +
.../apache/doris/spark/config/DorisOptions.java | 18 +-
.../apache/doris/spark/config/S3TvfOptions.java | 135 +++++++++
.../spark-doris-connector-it/pom.xml | 24 +-
.../spark/container/AbstractS3TvfTestBase.java | 163 +++++++++++
.../spark/container/instance/DorisContainer.java | 34 ++-
.../doris/spark/sql/Doris2DorisE2ECase.scala | 11 +-
.../apache/doris/spark/sql/DorisReaderITCase.scala | 26 +-
.../spark/sql/DorisStreamingWriterITCase.scala | 18 +-
.../spark/sql/DorisWriterFailoverITCase.scala | 140 ++++-----
.../org/apache/doris/spark/sql/S3TvfE2ECase.scala | 122 ++++++++
.../apache/doris/spark/sql/S3TvfSinkITCase.scala | 318 +++++++++++++++++++++
.../src/test/resources/docker/doris/fe.conf | 4 +-
.../src/test/resources/log4j.properties | 24 ++
.../spark-doris-connector-spark-2/pom.xml | 12 +-
.../doris/spark/sql/DorisSourceProvider.scala | 23 +-
.../doris/spark/writer/S3TvfBatchWriter.scala | 86 ++++++
.../spark-doris-connector-spark-3-base/pom.xml | 10 +
.../org/apache/doris/spark/write/DorisWrite.scala | 29 +-
.../doris/spark/write/DorisWriteBuilder.scala | 3 +
.../apache/doris/spark/write/S3TvfDataWriter.scala | 74 +++++
.../doris/spark/write/S3TvfDataWriterFactory.scala | 50 ++++
.../spark-doris-connector-spark-4-base/pom.xml | 10 +
.../org/apache/doris/spark/write/DorisWrite.scala | 27 +-
.../doris/spark/write/DorisWriteBuilder.scala | 3 +
.../apache/doris/spark/write/S3TvfDataWriter.scala | 74 +++++
.../doris/spark/write/S3TvfDataWriterFactory.scala | 50 ++++
42 files changed, 2223 insertions(+), 136 deletions(-)
diff --git a/.github/workflows/run-e2ecase.yml
b/.github/workflows/run-e2ecase.yml
index 6ce6545..769f56d 100644
--- a/.github/workflows/run-e2ecase.yml
+++ b/.github/workflows/run-e2ecase.yml
@@ -32,6 +32,11 @@ jobs:
- name: Checkout
uses: actions/checkout@master
+ - name: Configure Doris prerequisites
+ run: |
+ sudo sysctl -w vm.max_map_count=2000000
+ sudo swapoff -a
+
- name: Setup java
uses: actions/setup-java@v2
with:
@@ -40,27 +45,27 @@ jobs:
- name: Run E2ECases for spark 2
run: |
- cd spark-doris-connector && mvn clean test -Pspark-2-it,spark-2.4_2.11
-pl spark-doris-connector-it -am -DfailIfNoTests=false -Dtest="*E2ECase"
-Dimage="apache/doris:doris-all-in-one-2.1.0"
+ cd spark-doris-connector && mvn clean test -Pspark-2-it,spark-2.4_2.11
-pl spark-doris-connector-it -am -DfailIfNoTests=false -Dtest="*E2ECase"
-Dimage="jnsimba/doris-all-in-one:4.1.3"
- name: Run E2ECases for spark 3.1
run: |
- cd spark-doris-connector && mvn clean test -Pspark-3-it,spark-3.1 -pl
spark-doris-connector-it -am -DfailIfNoTests=false -Dtest="*E2ECase"
-Dimage="apache/doris:doris-all-in-one-2.1.0"
+ cd spark-doris-connector && mvn clean test -Pspark-3-it,spark-3.1 -pl
spark-doris-connector-it -am -DfailIfNoTests=false -Dtest="*E2ECase"
-Dimage="jnsimba/doris-all-in-one:4.1.3"
- name: Run E2ECases for spark 3.2
run: |
- cd spark-doris-connector && mvn clean test -Pspark-3-it,spark-3.2 -pl
spark-doris-connector-it -am -DfailIfNoTests=false -Dtest="*E2ECase"
-Dimage="apache/doris:doris-all-in-one-2.1.0"
+ cd spark-doris-connector && mvn clean test -Pspark-3-it,spark-3.2 -pl
spark-doris-connector-it -am -DfailIfNoTests=false -Dtest="*E2ECase"
-Dimage="jnsimba/doris-all-in-one:4.1.3"
- name: Run E2ECases for spark 3.3
run: |
- cd spark-doris-connector && mvn clean test -Pspark-3-it,spark-3.3 -pl
spark-doris-connector-it -am -DfailIfNoTests=false -Dtest="*E2ECase"
-Dimage="apache/doris:doris-all-in-one-2.1.0"
+ cd spark-doris-connector && mvn clean test -Pspark-3-it,spark-3.3 -pl
spark-doris-connector-it -am -DfailIfNoTests=false -Dtest="*E2ECase"
-Dimage="jnsimba/doris-all-in-one:4.1.3"
- name: Run E2ECases for spark 3.4
run: |
- cd spark-doris-connector && mvn clean test -Pspark-3-it,spark-3.4 -pl
spark-doris-connector-it -am -DfailIfNoTests=false -Dtest="*E2ECase"
-Dimage="apache/doris:doris-all-in-one-2.1.0"
+ cd spark-doris-connector && mvn clean test -Pspark-3-it,spark-3.4 -pl
spark-doris-connector-it -am -DfailIfNoTests=false -Dtest="*E2ECase"
-Dimage="jnsimba/doris-all-in-one:4.1.3"
- 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"
+ cd spark-doris-connector && mvn clean test -Pspark-3-it,spark-3.5 -pl
spark-doris-connector-it -am -DfailIfNoTests=false -Dtest="*E2ECase"
-Dimage="jnsimba/doris-all-in-one:4.1.3"
run-e2ecase-spark4:
name: "Run E2ECases (Spark 4.x)"
@@ -72,6 +77,11 @@ jobs:
- name: Checkout
uses: actions/checkout@master
+ - name: Configure Doris prerequisites
+ run: |
+ sudo sysctl -w vm.max_map_count=2000000
+ sudo swapoff -a
+
# Spark 4.x requires JDK 17.
- name: Setup java
uses: actions/setup-java@v3
@@ -81,4 +91,4 @@ jobs:
- 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"
+ cd spark-doris-connector && mvn clean test -Pspark-4-it,spark-4.1 -pl
spark-doris-connector-it -am -DfailIfNoTests=false -Dtest="*E2ECase"
-Dimage="jnsimba/doris-all-in-one:4.1.3"
diff --git a/.github/workflows/run-itcase.yml b/.github/workflows/run-itcase.yml
index 1d4b8d7..d670259 100644
--- a/.github/workflows/run-itcase.yml
+++ b/.github/workflows/run-itcase.yml
@@ -32,6 +32,11 @@ jobs:
- name: Checkout
uses: actions/checkout@master
+ - name: Configure Doris prerequisites
+ run: |
+ sudo sysctl -w vm.max_map_count=2000000
+ sudo swapoff -a
+
- name: Setup java
uses: actions/setup-java@v2
with:
@@ -40,27 +45,27 @@ jobs:
- name: Run ITCases for spark 2
run: |
- cd spark-doris-connector && mvn clean test -Pspark-2-it,spark-2.4_2.11
-pl spark-doris-connector-it -am -DfailIfNoTests=false -Dtest="*ITCase"
-Dimage="apache/doris:doris-all-in-one-2.1.0"
+ cd spark-doris-connector && mvn clean test -Pspark-2-it,spark-2.4_2.11
-pl spark-doris-connector-it -am -DfailIfNoTests=false -Dtest="*ITCase"
-Dimage="jnsimba/doris-all-in-one:4.1.3"
- name: Run ITCases for spark 3.1
run: |
- cd spark-doris-connector && mvn clean test -Pspark-3-it,spark-3.1 -pl
spark-doris-connector-it -am -DfailIfNoTests=false -Dtest="*ITCase"
-Dimage="apache/doris:doris-all-in-one-2.1.0"
+ cd spark-doris-connector && mvn clean test -Pspark-3-it,spark-3.1 -pl
spark-doris-connector-it -am -DfailIfNoTests=false -Dtest="*ITCase"
-Dimage="jnsimba/doris-all-in-one:4.1.3"
- name: Run ITCases for spark 3.2
run: |
- cd spark-doris-connector && mvn clean test -Pspark-3-it,spark-3.2 -pl
spark-doris-connector-it -am -DfailIfNoTests=false -Dtest="*ITCase"
-Dimage="apache/doris:doris-all-in-one-2.1.0"
+ cd spark-doris-connector && mvn clean test -Pspark-3-it,spark-3.2 -pl
spark-doris-connector-it -am -DfailIfNoTests=false -Dtest="*ITCase"
-Dimage="jnsimba/doris-all-in-one:4.1.3"
- name: Run ITCases for spark 3.3
run: |
- cd spark-doris-connector && mvn clean test -Pspark-3-it,spark-3.3 -pl
spark-doris-connector-it -am -DfailIfNoTests=false -Dtest="*ITCase"
-Dimage="apache/doris:doris-all-in-one-2.1.0"
+ cd spark-doris-connector && mvn clean test -Pspark-3-it,spark-3.3 -pl
spark-doris-connector-it -am -DfailIfNoTests=false -Dtest="*ITCase"
-Dimage="jnsimba/doris-all-in-one:4.1.3"
- name: Run ITCases for spark 3.4
run: |
- cd spark-doris-connector && mvn clean test -Pspark-3-it,spark-3.4 -pl
spark-doris-connector-it -am -DfailIfNoTests=false -Dtest="*ITCase"
-Dimage="apache/doris:doris-all-in-one-2.1.0"
+ cd spark-doris-connector && mvn clean test -Pspark-3-it,spark-3.4 -pl
spark-doris-connector-it -am -DfailIfNoTests=false -Dtest="*ITCase"
-Dimage="jnsimba/doris-all-in-one:4.1.3"
- name: Run ITCases 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="*ITCase"
-Dimage="apache/doris:doris-all-in-one-2.1.0"
+ cd spark-doris-connector && mvn clean test -Pspark-3-it,spark-3.5 -pl
spark-doris-connector-it -am -DfailIfNoTests=false -Dtest="*ITCase"
-Dimage="jnsimba/doris-all-in-one:4.1.3"
run-itcase-spark4:
name: "Run ITCases (Spark 4.x)"
@@ -72,6 +77,11 @@ jobs:
- name: Checkout
uses: actions/checkout@master
+ - name: Configure Doris prerequisites
+ run: |
+ sudo sysctl -w vm.max_map_count=2000000
+ sudo swapoff -a
+
# Spark 4.x requires JDK 17.
- name: Setup java
uses: actions/setup-java@v3
@@ -81,5 +91,4 @@ jobs:
- 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"
-
+ cd spark-doris-connector && mvn clean test -Pspark-4-it,spark-4.1 -pl
spark-doris-connector-it -am -DfailIfNoTests=false -Dtest="*ITCase"
-Dimage="jnsimba/doris-all-in-one:4.1.3"
diff --git a/spark-doris-connector/pom.xml b/spark-doris-connector/pom.xml
index 873c3ef..fcb5ce2 100644
--- a/spark-doris-connector/pom.xml
+++ b/spark-doris-connector/pom.xml
@@ -97,6 +97,7 @@
<fasterxml.jackson.version>2.13.5</fasterxml.jackson.version>
<thrift-service.version>1.0.1</thrift-service.version>
<testcontainers.version>1.17.6</testcontainers.version>
+ <awssdk.version>2.29.52</awssdk.version>
<jmockit.version>1.49</jmockit.version>
<fe_ut_parallel>1</fe_ut_parallel>
<argLine>-Xmx512m</argLine>
@@ -603,6 +604,14 @@
<pattern>com.google</pattern>
<shadedPattern>org.apache.doris.shaded.com.google</shadedPattern>
</relocation>
+ <relocation>
+ <pattern>software.amazon.awssdk</pattern>
+
<shadedPattern>org.apache.doris.shaded.software.amazon.awssdk</shadedPattern>
+ </relocation>
+ <relocation>
+ <pattern>software.amazon.eventstream</pattern>
+
<shadedPattern>org.apache.doris.shaded.software.amazon.eventstream</shadedPattern>
+ </relocation>
</relocations>
<transformers>
<transformer
diff --git a/spark-doris-connector/spark-doris-connector-base/pom.xml
b/spark-doris-connector/spark-doris-connector-base/pom.xml
index e3480ca..4e24e7f 100644
--- a/spark-doris-connector/spark-doris-connector-base/pom.xml
+++ b/spark-doris-connector/spark-doris-connector-base/pom.xml
@@ -208,6 +208,27 @@
<artifactId>guava</artifactId>
</dependency>
+ <dependency>
+ <groupId>software.amazon.awssdk</groupId>
+ <artifactId>s3</artifactId>
+ <version>${awssdk.version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>software.amazon.awssdk</groupId>
+ <artifactId>apache-client</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>software.amazon.awssdk</groupId>
+ <artifactId>netty-nio-client</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>software.amazon.awssdk</groupId>
+ <artifactId>url-connection-client</artifactId>
+ <version>${awssdk.version}</version>
+ </dependency>
+
<!--Test-->
<dependency>
<groupId>org.hamcrest</groupId>
diff --git
a/spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/client/DorisFrontendClient.java
b/spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/client/DorisFrontendClient.java
index 711a769..f0c57e8 100644
---
a/spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/client/DorisFrontendClient.java
+++
b/spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/client/DorisFrontendClient.java
@@ -72,6 +72,11 @@ import java.util.stream.Collectors;
public class DorisFrontendClient implements Serializable {
+ @FunctionalInterface
+ public interface JdbcAction {
+ void execute(Connection connection) throws SQLException;
+ }
+
private static final Logger LOG =
LoggerFactory.getLogger(DorisFrontendClient.class);
private static final ObjectMapper MAPPER = JsonMapper.builder().build();
@@ -218,6 +223,36 @@ public class DorisFrontendClient implements Serializable {
throw ex;
}
+ /**
+ * Executes a JDBC action against one selected frontend without failover.
This is intended for
+ * non-idempotent statements whose outcome may be ambiguous after a
connection failure.
+ */
+ public void executeFrontendOnce(JdbcAction action) throws Exception {
+ if (jdbcTlsAdapter == null) {
+ jdbcTlsAdapter = DorisJdbcTlsAdapter.create(tlsOptions);
+ }
+ Iterator<Frontend> iterator = frontends.iterator();
+ if (!iterator.hasNext()) {
+ throw new DorisException("No frontend is available for JDBC
query.");
+ }
+ Frontend frontEnd = iterator.next();
+ if (frontEnd.getQueryPort() == -1) {
+ throw new
OptionRequiredException(DorisOptions.DORIS_QUERY_PORT.getName());
+ }
+ try {
+ Class.forName("com.mysql.cj.jdbc.Driver");
+ } catch (ClassNotFoundException e) {
+ Class.forName("com.mysql.jdbc.Driver");
+ }
+ String jdbcUrl = "jdbc:mysql://" + frontEnd.getHost() + ":" +
frontEnd.getQueryPort();
+ jdbcTlsAdapter.validateJdbcUrl(jdbcUrl);
+ try (Connection conn = DriverManager.getConnection(
+ jdbcUrl,
+ jdbcTlsAdapter.createConnectionProperties(username,
password))) {
+ action.execute(conn);
+ }
+ }
+
public List<Pair<String[], String>> listTables(String[] databases) throws
Exception {
return queryFrontends(conn -> {
String where = databases.length == 1 ? " WHERE TABLE_SCHEMA = '" +
databases[0] + "'" : "";
diff --git
a/spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/client/write/tvf/JdbcS3TvfLoadClient.java
b/spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/client/write/tvf/JdbcS3TvfLoadClient.java
new file mode 100644
index 0000000..7a95523
--- /dev/null
+++
b/spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/client/write/tvf/JdbcS3TvfLoadClient.java
@@ -0,0 +1,76 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.spark.client.write.tvf;
+
+import org.apache.doris.spark.client.DorisFrontendClient;
+import org.apache.doris.spark.config.DorisConfig;
+
+import java.io.IOException;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.Map;
+import java.util.TreeMap;
+import java.util.regex.Pattern;
+
+/** JDBC implementation that applies session variables and executes one
INSERT. */
+public final class JdbcS3TvfLoadClient implements S3TvfLoadClient {
+ private static final Pattern SESSION_VARIABLE =
Pattern.compile("[A-Za-z_][A-Za-z0-9_]*");
+
+ private final DorisFrontendClient frontendClient;
+
+ public JdbcS3TvfLoadClient(DorisConfig config) throws Exception {
+ this(new DorisFrontendClient(config));
+ }
+
+ JdbcS3TvfLoadClient(DorisFrontendClient frontendClient) {
+ this.frontendClient = frontendClient;
+ }
+
+ @Override
+ public void executeInsert(String sql, Map<String, String> sessionVariables)
+ throws SQLException {
+ try {
+ frontendClient.executeFrontendOnce(connection -> {
+ try (Statement statement = connection.createStatement()) {
+ for (Map.Entry<String, String> entry
+ : new TreeMap<>(sessionVariables).entrySet()) {
+ if
(!SESSION_VARIABLE.matcher(entry.getKey()).matches()) {
+ throw new SQLException(
+ "Invalid Doris session variable: " +
entry.getKey());
+ }
+ statement.execute(
+ "SET SESSION "
+ + entry.getKey()
+ + " = "
+ +
TvfSqlUtils.quoteLiteral(entry.getValue()));
+ }
+ statement.execute(sql);
+ }
+ });
+ } catch (SQLException e) {
+ throw e;
+ } catch (Exception e) {
+ throw new SQLException("Failed to execute Doris S3 TVF INSERT", e);
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ frontendClient.close();
+ }
+}
diff --git
a/spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/client/write/tvf/S3ClientObjectStore.java
b/spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/client/write/tvf/S3ClientObjectStore.java
new file mode 100644
index 0000000..2edb360
--- /dev/null
+++
b/spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/client/write/tvf/S3ClientObjectStore.java
@@ -0,0 +1,83 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.spark.client.write.tvf;
+
+import org.apache.doris.spark.config.S3TvfOptions;
+import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
+import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
+import software.amazon.awssdk.core.sync.RequestBody;
+import software.amazon.awssdk.http.urlconnection.UrlConnectionHttpClient;
+import software.amazon.awssdk.regions.Region;
+import software.amazon.awssdk.services.s3.S3Client;
+import software.amazon.awssdk.services.s3.S3Configuration;
+import software.amazon.awssdk.services.s3.model.PutObjectRequest;
+
+import java.io.IOException;
+import java.net.URI;
+
+/** AWS SDK based implementation for S3-compatible object storage. */
+public final class S3ClientObjectStore implements S3ObjectStore {
+ private static final String JSON_LINES_CONTENT_TYPE =
"application/x-ndjson";
+
+ private final S3Client client;
+ private final String bucket;
+
+ public S3ClientObjectStore(S3TvfOptions options) {
+ this(createClient(options), options.getBucket());
+ }
+
+ S3ClientObjectStore(S3Client client, String bucket) {
+ this.client = client;
+ this.bucket = bucket;
+ }
+
+ private static S3Client createClient(S3TvfOptions options) {
+ return S3Client.builder()
+ .endpointOverride(URI.create(options.getEndpoint()))
+ .region(Region.of(options.getRegion()))
+ .credentialsProvider(
+ StaticCredentialsProvider.create(
+ AwsBasicCredentials.create(
+ options.getAccessKey(),
options.getSecretKey())))
+ .httpClientBuilder(UrlConnectionHttpClient.builder())
+ .serviceConfiguration(
+ S3Configuration.builder()
+
.pathStyleAccessEnabled(options.isPathStyleAccess())
+ .build())
+ .build();
+ }
+
+ @Override
+ public void put(String objectKey, byte[] content) throws IOException {
+ PutObjectRequest request = PutObjectRequest.builder()
+ .bucket(bucket)
+ .key(objectKey)
+ .contentType(JSON_LINES_CONTENT_TYPE)
+ .build();
+ try {
+ client.putObject(request, RequestBody.fromBytes(content));
+ } catch (RuntimeException e) {
+ throw new IOException("Failed to upload S3 TVF object: " +
objectKey, e);
+ }
+ }
+
+ @Override
+ public void close() {
+ client.close();
+ }
+}
diff --git
a/spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/client/write/tvf/S3ObjectStore.java
b/spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/client/write/tvf/S3ObjectStore.java
new file mode 100644
index 0000000..dc9ee09
--- /dev/null
+++
b/spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/client/write/tvf/S3ObjectStore.java
@@ -0,0 +1,26 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.spark.client.write.tvf;
+
+import java.io.Closeable;
+import java.io.IOException;
+
+/** Object storage operations required by the S3 TVF sink. */
+public interface S3ObjectStore extends Closeable {
+ void put(String objectKey, byte[] content) throws IOException;
+}
diff --git
a/spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/client/write/tvf/S3TvfColumnUtils.java
b/spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/client/write/tvf/S3TvfColumnUtils.java
new file mode 100644
index 0000000..8b0ef59
--- /dev/null
+++
b/spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/client/write/tvf/S3TvfColumnUtils.java
@@ -0,0 +1,76 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.spark.client.write.tvf;
+
+import org.apache.spark.sql.types.StructType;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+/** Resolves the fixed physical column list used by the TVF write path. */
+public final class S3TvfColumnUtils {
+ private static final String COLUMNS = "columns";
+
+ private S3TvfColumnUtils() {}
+
+ public static List<String> resolveColumns(
+ Map<String, String> loadProperties, StructType schema) {
+ String configuredColumns = loadProperties.get(COLUMNS);
+ if (configuredColumns == null || configuredColumns.trim().isEmpty()) {
+ return
Collections.unmodifiableList(Arrays.asList(schema.fieldNames()));
+ }
+
+ Set<String> schemaFields = new
HashSet<>(Arrays.asList(schema.fieldNames()));
+ Set<String> seen = new HashSet<>();
+ List<String> columns = new ArrayList<>();
+ for (String rawColumn : configuredColumns.split(",", -1)) {
+ String column = unquoteIdentifier(rawColumn.trim());
+ if (column.isEmpty()) {
+ throw new IllegalArgumentException(
+ "doris.sink.properties.columns must not contain an
empty column");
+ }
+ if (!schemaFields.contains(column)) {
+ throw new IllegalArgumentException(
+ "Column '"
+ + column
+ + "' from doris.sink.properties.columns is not
a Spark input "
+ + "column");
+ }
+ if (!seen.add(column)) {
+ throw new IllegalArgumentException(
+ "Column '"
+ + column
+ + "' is duplicated in
doris.sink.properties.columns");
+ }
+ columns.add(column);
+ }
+ return Collections.unmodifiableList(columns);
+ }
+
+ private static String unquoteIdentifier(String value) {
+ if (value.length() >= 2 && value.startsWith("`") &&
value.endsWith("`")) {
+ return value.substring(1, value.length() - 1).replace("``", "`");
+ }
+ return value;
+ }
+}
diff --git
a/spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/client/write/tvf/S3TvfCommittable.java
b/spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/client/write/tvf/S3TvfCommittable.java
new file mode 100644
index 0000000..4a13df2
--- /dev/null
+++
b/spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/client/write/tvf/S3TvfCommittable.java
@@ -0,0 +1,68 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.spark.client.write.tvf;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+/** Doris target, files, and columns prepared by one Spark task for a Doris
INSERT. */
+public final class S3TvfCommittable {
+ private final String database;
+ private final String table;
+ private final String label;
+ private final List<String> objectKeys;
+ private final List<String> columns;
+
+ public S3TvfCommittable(
+ String database,
+ String table,
+ String label,
+ List<String> objectKeys,
+ List<String> columns) {
+ this.database = database;
+ this.table = table;
+ this.label = label;
+ this.objectKeys = Collections.unmodifiableList(new
ArrayList<>(objectKeys));
+ this.columns = Collections.unmodifiableList(new ArrayList<>(columns));
+ }
+
+ public String getDatabase() {
+ return database;
+ }
+
+ public String getTable() {
+ return table;
+ }
+
+ public String getLabel() {
+ return label;
+ }
+
+ public List<String> getObjectKeys() {
+ return objectKeys;
+ }
+
+ public List<String> getColumns() {
+ return columns;
+ }
+
+ public boolean isEmpty() {
+ return objectKeys.isEmpty();
+ }
+}
diff --git
a/spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/client/write/tvf/S3TvfCommitter.java
b/spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/client/write/tvf/S3TvfCommitter.java
new file mode 100644
index 0000000..f709a77
--- /dev/null
+++
b/spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/client/write/tvf/S3TvfCommitter.java
@@ -0,0 +1,94 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.spark.client.write.tvf;
+
+import org.apache.doris.spark.config.DorisConfig;
+import org.apache.doris.spark.config.S3TvfOptions;
+
+import java.io.IOException;
+import java.sql.SQLException;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.TreeMap;
+
+/** Commits one Spark partition with one Doris INSERT. */
+public final class S3TvfCommitter implements AutoCloseable {
+ private static final String COLUMNS = "columns";
+ private static final String PARTIAL_COLUMNS = "partial_columns";
+ private static final String FORMAT = "format";
+ private static final String READ_JSON_BY_LINE = "read_json_by_line";
+ private static final String ENABLE_UNIQUE_KEY_PARTIAL_UPDATE =
+ "enable_unique_key_partial_update";
+
+ private final Map<String, String> sessionVariables;
+ private final S3TvfLoadClient loadClient;
+ private final S3TvfSqlBuilder sqlBuilder;
+
+ public S3TvfCommitter(DorisConfig config) throws Exception {
+ this(
+ S3TvfOptions.fromConfig(config),
+ config.getSinkProperties(),
+ new JdbcS3TvfLoadClient(config));
+ }
+
+ S3TvfCommitter(
+ S3TvfOptions options,
+ Map<String, String> loadProperties,
+ S3TvfLoadClient loadClient) {
+ this.sessionVariables = toSessionVariables(loadProperties);
+ this.loadClient = loadClient;
+ this.sqlBuilder = new S3TvfSqlBuilder(options);
+ }
+
+ public void commit(S3TvfCommittable committable) throws IOException {
+ if (committable.isEmpty()) {
+ return;
+ }
+ try {
+ loadClient.executeInsert(
+ sqlBuilder.buildInsertSql(committable),
+ sessionVariables);
+ } catch (SQLException e) {
+ throw new IOException(
+ "Doris INSERT failed for S3 TVF label " +
committable.getLabel(), e);
+ }
+ }
+
+ private static Map<String, String> toSessionVariables(Map<String, String>
loadProperties) {
+ Map<String, String> values = new TreeMap<>();
+ for (Map.Entry<String, String> entry : loadProperties.entrySet()) {
+ String name = entry.getKey();
+ if (!COLUMNS.equals(name)
+ && !PARTIAL_COLUMNS.equals(name)
+ && !FORMAT.equals(name)
+ && !READ_JSON_BY_LINE.equals(name)) {
+ values.put(name, entry.getValue());
+ }
+ }
+ if (loadProperties.containsKey(PARTIAL_COLUMNS)) {
+ values.put(
+ ENABLE_UNIQUE_KEY_PARTIAL_UPDATE,
loadProperties.get(PARTIAL_COLUMNS));
+ }
+ return new LinkedHashMap<>(values);
+ }
+
+ @Override
+ public void close() throws IOException {
+ loadClient.close();
+ }
+}
diff --git
a/spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/client/write/tvf/S3TvfLoadClient.java
b/spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/client/write/tvf/S3TvfLoadClient.java
new file mode 100644
index 0000000..eb2556d
--- /dev/null
+++
b/spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/client/write/tvf/S3TvfLoadClient.java
@@ -0,0 +1,27 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.spark.client.write.tvf;
+
+import java.io.Closeable;
+import java.sql.SQLException;
+import java.util.Map;
+
+/** Submits one synchronous Doris INSERT. */
+public interface S3TvfLoadClient extends Closeable {
+ void executeInsert(String sql, Map<String, String> sessionVariables)
throws SQLException;
+}
diff --git
a/spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/client/write/tvf/S3TvfSqlBuilder.java
b/spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/client/write/tvf/S3TvfSqlBuilder.java
new file mode 100644
index 0000000..0fbd025
--- /dev/null
+++
b/spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/client/write/tvf/S3TvfSqlBuilder.java
@@ -0,0 +1,95 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.spark.client.write.tvf;
+
+import org.apache.doris.spark.config.S3TvfOptions;
+
+import java.util.List;
+import java.util.StringJoiner;
+
+/** Builds one S3 TVF INSERT for one Spark partition. */
+public final class S3TvfSqlBuilder {
+ private final S3TvfOptions options;
+
+ public S3TvfSqlBuilder(S3TvfOptions options) {
+ this.options = options;
+ }
+
+ public String buildInsertSql(S3TvfCommittable committable) {
+ List<String> objectKeys = committable.getObjectKeys();
+ List<String> columns = committable.getColumns();
+ if (objectKeys.isEmpty()) {
+ throw new IllegalArgumentException("S3 TVF object keys must not be
empty");
+ }
+ if (columns.isEmpty()) {
+ throw new IllegalArgumentException("S3 TVF columns must not be
empty");
+ }
+ String columnSql = joinIdentifiers(columns);
+ String uri = buildUri(objectKeys);
+ return "INSERT INTO "
+ + TvfSqlUtils.quoteIdentifier(committable.getDatabase())
+ + "."
+ + TvfSqlUtils.quoteIdentifier(committable.getTable())
+ + " WITH LABEL "
+ + TvfSqlUtils.quoteIdentifier(committable.getLabel())
+ + " ("
+ + columnSql
+ + ") SELECT "
+ + columnSql
+ + " FROM S3("
+ + property("uri", uri)
+ + ","
+ + property("format", "json")
+ + ","
+ + property("read_json_by_line", "true")
+ + ","
+ + property("s3.endpoint", options.getEndpoint())
+ + ","
+ + property("s3.region", options.getRegion())
+ + ","
+ + property("s3.access_key", options.getAccessKey())
+ + ","
+ + property("s3.secret_key", options.getSecretKey())
+ + ","
+ + property("use_path_style",
Boolean.toString(options.isPathStyleAccess()))
+ + ")";
+ }
+
+ private String buildUri(List<String> objectKeys) {
+ if (objectKeys.size() == 1) {
+ return "s3://" + options.getBucket() + "/" + objectKeys.get(0);
+ }
+ StringJoiner keys = new StringJoiner(",");
+ for (String objectKey : objectKeys) {
+ keys.add(objectKey);
+ }
+ return "s3://" + options.getBucket() + "/{" + keys + "}";
+ }
+
+ private static String joinIdentifiers(List<String> columns) {
+ StringJoiner joiner = new StringJoiner(",");
+ for (String column : columns) {
+ joiner.add(TvfSqlUtils.quoteIdentifier(column));
+ }
+ return joiner.toString();
+ }
+
+ private static String property(String key, String value) {
+ return TvfSqlUtils.quoteLiteral(key) + " = " +
TvfSqlUtils.quoteLiteral(value);
+ }
+}
diff --git
a/spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/client/write/tvf/S3TvfWriter.java
b/spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/client/write/tvf/S3TvfWriter.java
new file mode 100644
index 0000000..9ef17dd
--- /dev/null
+++
b/spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/client/write/tvf/S3TvfWriter.java
@@ -0,0 +1,178 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.spark.client.write.tvf;
+
+import org.apache.doris.spark.config.S3TvfOptions;
+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.types.StructField;
+import org.apache.spark.sql.types.StructType;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.UUID;
+
+/** Writes one logical Spark partition as deterministic JSON Lines objects. */
+public final class S3TvfWriter implements AutoCloseable {
+ private static final byte NEW_LINE = '\n';
+
+ private final S3TvfOptions options;
+ private final String database;
+ private final String table;
+ private final String normalizedTable;
+ private final String labelPrefix;
+ private final List<String> columns;
+ private final StructType outputSchema;
+ private final int[] selectedIndexes;
+ private final boolean projectionRequired;
+ private final String batchUuid = UUID.randomUUID().toString();
+ private final int partitionId;
+ private final int batchSize;
+ private final S3ObjectStore objectStore;
+ private final ByteArrayOutputStream buffer = new ByteArrayOutputStream();
+ private final List<String> objectKeys = new ArrayList<>();
+
+ private int recordCount;
+ private int fileNumber;
+
+ public S3TvfWriter(
+ S3TvfOptions options,
+ String database,
+ String table,
+ String labelPrefix,
+ StructType inputSchema,
+ List<String> columns,
+ int partitionId,
+ int batchSize,
+ S3ObjectStore objectStore) {
+ if (batchSize <= 0) {
+ throw new IllegalArgumentException("S3 TVF batch size must be
greater than zero");
+ }
+ this.options = options;
+ this.database = database;
+ this.table = table;
+ this.normalizedTable = normalizeTable(table);
+ this.labelPrefix = labelPrefix;
+ this.columns = new ArrayList<>(columns);
+ this.selectedIndexes = resolveIndexes(inputSchema, columns);
+ this.outputSchema = selectSchema(inputSchema, selectedIndexes);
+ this.projectionRequired = requiresProjection(selectedIndexes,
inputSchema.fields().length);
+ this.partitionId = partitionId;
+ this.batchSize = batchSize;
+ this.objectStore = objectStore;
+ }
+
+ public void write(InternalRow row) throws IOException {
+ byte[] json = RowConvertors.convertToJsonBytes(project(row),
outputSchema);
+ buffer.write(json);
+ buffer.write(NEW_LINE);
+ recordCount++;
+ if (recordCount >= batchSize) {
+ uploadBuffer();
+ }
+ }
+
+ public S3TvfCommittable prepareCommit() throws IOException {
+ uploadBuffer();
+ return new S3TvfCommittable(database, table, label(), objectKeys,
columns);
+ }
+
+ private void uploadBuffer() throws IOException {
+ if (recordCount == 0) {
+ return;
+ }
+ byte[] content = buffer.toByteArray();
+ int currentFileNumber = fileNumber++;
+ String fileName = String.format(
+ "%s_%s_%s_%d_%d.json",
+ labelPrefix,
+ normalizedTable,
+ batchUuid,
+ partitionId,
+ currentFileNumber);
+ String prefix = options.getPrefix();
+ String objectKey = prefix + (prefix.endsWith("/") ? "" : "/") +
fileName;
+ objectStore.put(objectKey, content);
+ objectKeys.add(objectKey);
+ buffer.reset();
+ recordCount = 0;
+ }
+
+ private String label() {
+ return labelPrefix
+ + "_"
+ + normalizedTable
+ + "_"
+ + batchUuid
+ + "_"
+ + partitionId;
+ }
+
+ private static String normalizeTable(String table) {
+ return table.replaceAll("[^A-Za-z0-9_-]", "_");
+ }
+
+ private InternalRow project(InternalRow row) {
+ if (!projectionRequired) {
+ return row;
+ }
+ Object[] values = new Object[selectedIndexes.length];
+ StructField[] fields = outputSchema.fields();
+ for (int index = 0; index < selectedIndexes.length; index++) {
+ values[index] = row.get(selectedIndexes[index],
fields[index].dataType());
+ }
+ return new GenericInternalRow(values);
+ }
+
+ private static int[] resolveIndexes(StructType schema, List<String>
columns) {
+ int[] indexes = new int[columns.size()];
+ for (int index = 0; index < columns.size(); index++) {
+ indexes[index] = schema.fieldIndex(columns.get(index));
+ }
+ return indexes;
+ }
+
+ private static StructType selectSchema(StructType schema, int[] indexes) {
+ StructField[] selectedFields = new StructField[indexes.length];
+ StructField[] fields = schema.fields();
+ for (int index = 0; index < indexes.length; index++) {
+ selectedFields[index] = fields[indexes[index]];
+ }
+ return new StructType(selectedFields);
+ }
+
+ private static boolean requiresProjection(int[] indexes, int fieldCount) {
+ if (indexes.length != fieldCount) {
+ return true;
+ }
+ for (int index = 0; index < indexes.length; index++) {
+ if (indexes[index] != index) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ @Override
+ public void close() throws IOException {
+ objectStore.close();
+ }
+}
diff --git
a/spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/client/write/tvf/TvfSqlUtils.java
b/spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/client/write/tvf/TvfSqlUtils.java
new file mode 100644
index 0000000..f32d3e1
--- /dev/null
+++
b/spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/client/write/tvf/TvfSqlUtils.java
@@ -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.client.write.tvf;
+
+final class TvfSqlUtils {
+ private TvfSqlUtils() {}
+
+ static String quoteIdentifier(String value) {
+ return "`" + value.replace("`", "``") + "`";
+ }
+
+ static String quoteLiteral(String value) {
+ return "'" + value.replace("\\", "\\\\").replace("'", "''") + "'";
+ }
+}
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 bfb89ff..1fa1fcc 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
@@ -106,6 +106,10 @@ public class DorisConfig implements Serializable {
if (!options.containsKey(DorisOptions.DORIS_PASSWORD.getName())) {
throw new
OptionRequiredException(DorisOptions.DORIS_PASSWORD.getName());
}
+ if
("tvf".equalsIgnoreCase(options.get(DorisOptions.LOAD_MODE.getName()))
+ &&
!options.containsKey(DorisOptions.DORIS_QUERY_PORT.getName())) {
+ throw new
OptionRequiredException(DorisOptions.DORIS_QUERY_PORT.getName());
+ }
if
("thrift".equalsIgnoreCase(options.get(DorisOptions.READ_MODE.getName()))) {
if
(Boolean.parseBoolean(options.get(DorisOptions.DORIS_READ_BITMAP_TO_STRING.getName())))
{
throw new IllegalArgumentException(String.format("option [%s]
is invalid in thrift read mode",
diff --git
a/spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/config/DorisOptions.java
b/spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/config/DorisOptions.java
index 361da92..10e9972 100644
---
a/spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/config/DorisOptions.java
+++
b/spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/config/DorisOptions.java
@@ -124,7 +124,7 @@ public class DorisOptions {
public static final ConfigOption<String> DORIS_TLS_EXCLUDED_PROTOCOLS =
ConfigOptions.name("doris.tls.excluded-protocols").stringType().defaultValue("").withDescription("Comma-separated
protocols excluded from TLS: http, mysql, thrift, arrowflight.");
- public static final ConfigOption<String> LOAD_MODE =
ConfigOptions.name("doris.sink.mode").stringType().defaultValue("stream_load").withDescription("");
+ public static final ConfigOption<String> LOAD_MODE =
ConfigOptions.name("doris.sink.mode").stringType().defaultValue("stream_load").withDescription("Write
mode, supports stream_load, copy_into and tvf.");
public static final ConfigOption<String> READ_MODE =
ConfigOptions.name("doris.read.mode").stringType().defaultValue("thrift").withDescription("");
@@ -132,7 +132,21 @@ public class DorisOptions {
public static final ConfigOption<Integer> DORIS_READ_FLIGHT_SQL_PORT =
ConfigOptions.name("doris.read.arrow-flight-sql.port").intType().withoutDefaultValue().withDescription("");
- public static final ConfigOption<String> DORIS_SINK_LABEL_PREFIX =
ConfigOptions.name("doris.sink.label.prefix").stringType().defaultValue("spark-doris").withDescription("");
+ public static final ConfigOption<String> DORIS_SINK_LABEL_PREFIX =
ConfigOptions.name("doris.sink.label.prefix").stringType().defaultValue("spark-doris").withDescription("Label
prefix used by Doris sink writes.");
+
+ public static final ConfigOption<String> DORIS_SINK_S3_ENDPOINT =
ConfigOptions.name("doris.sink.s3.endpoint").stringType().withoutDefaultValue().withDescription("Endpoint
of the S3-compatible object storage.");
+
+ public static final ConfigOption<String> DORIS_SINK_S3_REGION =
ConfigOptions.name("doris.sink.s3.region").stringType().withoutDefaultValue().withDescription("Region
of the S3-compatible object storage.");
+
+ public static final ConfigOption<String> DORIS_SINK_S3_BUCKET =
ConfigOptions.name("doris.sink.s3.bucket").stringType().withoutDefaultValue().withDescription("Bucket
used to stage files for the S3 TVF.");
+
+ public static final ConfigOption<String> DORIS_SINK_S3_PREFIX =
ConfigOptions.name("doris.sink.s3.prefix").stringType().withoutDefaultValue().withDescription("Object
key path prefix used to stage TVF files.");
+
+ public static final ConfigOption<String> DORIS_SINK_S3_ACCESS_KEY =
ConfigOptions.name("doris.sink.s3.access-key").stringType().withoutDefaultValue().withDescription("Access
key of the S3-compatible object storage.");
+
+ public static final ConfigOption<String> DORIS_SINK_S3_SECRET_KEY =
ConfigOptions.name("doris.sink.s3.secret-key").stringType().withoutDefaultValue().withDescription("Secret
key of the S3-compatible object storage.");
+
+ public static final ConfigOption<Boolean> DORIS_SINK_S3_PATH_STYLE_ACCESS
=
ConfigOptions.name("doris.sink.s3.path-style-access").booleanType().defaultValue(false).withDescription("Whether
to use path-style access for object storage.");
public static final ConfigOption<Integer> DORIS_THRIFT_MAX_MESSAGE_SIZE =
ConfigOptions.name("doris.thrift.max.message.size").intType().defaultValue(Integer.MAX_VALUE).withDescription("")
;
diff --git
a/spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/config/S3TvfOptions.java
b/spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/config/S3TvfOptions.java
new file mode 100644
index 0000000..399a3ba
--- /dev/null
+++
b/spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/config/S3TvfOptions.java
@@ -0,0 +1,135 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.spark.config;
+
+import org.apache.doris.spark.exception.OptionRequiredException;
+
+import java.io.Serializable;
+import java.util.Map;
+
+/** Immutable configuration for the S3 TVF sink. */
+public final class S3TvfOptions implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+ private static final String FORMAT = "format";
+ private static final String READ_JSON_BY_LINE = "read_json_by_line";
+
+ private final String endpoint;
+ private final String region;
+ private final String bucket;
+ private final String prefix;
+ private final String accessKey;
+ private final String secretKey;
+ private final boolean pathStyleAccess;
+
+ private S3TvfOptions(
+ String endpoint,
+ String region,
+ String bucket,
+ String prefix,
+ String accessKey,
+ String secretKey,
+ boolean pathStyleAccess) {
+ this.endpoint = endpoint;
+ this.region = region;
+ this.bucket = bucket;
+ this.prefix = prefix;
+ this.accessKey = accessKey;
+ this.secretKey = secretKey;
+ this.pathStyleAccess = pathStyleAccess;
+ }
+
+ public static S3TvfOptions fromConfig(DorisConfig config) throws
OptionRequiredException {
+ validateLoadProperties(config.getSinkProperties());
+ String endpoint = required(config,
DorisOptions.DORIS_SINK_S3_ENDPOINT);
+ String region = required(config, DorisOptions.DORIS_SINK_S3_REGION);
+ String bucket = required(config, DorisOptions.DORIS_SINK_S3_BUCKET);
+ String prefix = required(config, DorisOptions.DORIS_SINK_S3_PREFIX);
+ String accessKey = required(config,
DorisOptions.DORIS_SINK_S3_ACCESS_KEY);
+ String secretKey = required(config,
DorisOptions.DORIS_SINK_S3_SECRET_KEY);
+ validatePrefix(prefix);
+
+ return new S3TvfOptions(
+ endpoint,
+ region,
+ bucket,
+ prefix,
+ accessKey,
+ secretKey,
+ config.getValue(DorisOptions.DORIS_SINK_S3_PATH_STYLE_ACCESS));
+ }
+
+ private static String required(DorisConfig config, ConfigOption<String>
option)
+ throws OptionRequiredException {
+ String value = config.getValue(option).trim();
+ if (value.isEmpty()) {
+ throw new IllegalArgumentException(option.getName() + " must not
be empty");
+ }
+ return value;
+ }
+
+ private static void validateLoadProperties(Map<String, String>
loadProperties) {
+ String format = loadProperties.get(FORMAT);
+ if (format != null && !"json".equalsIgnoreCase(format.trim())) {
+ throw new IllegalArgumentException("TVF write mode only supports
JSON format");
+ }
+ String readJsonByLine = loadProperties.get(READ_JSON_BY_LINE);
+ if (readJsonByLine != null &&
!Boolean.parseBoolean(readJsonByLine.trim())) {
+ throw new IllegalArgumentException(
+ "TVF write mode requires
'doris.sink.properties.read_json_by_line' to be true");
+ }
+ }
+
+ private static void validatePrefix(String prefix) {
+ for (char character : "*?[]{},\\".toCharArray()) {
+ if (prefix.indexOf(character) >= 0) {
+ throw new IllegalArgumentException(
+ DorisOptions.DORIS_SINK_S3_PREFIX.getName()
+ + " must not contain glob characters");
+ }
+ }
+ }
+
+ public String getEndpoint() {
+ return endpoint;
+ }
+
+ public String getRegion() {
+ return region;
+ }
+
+ public String getBucket() {
+ return bucket;
+ }
+
+ public String getPrefix() {
+ return prefix;
+ }
+
+ public String getAccessKey() {
+ return accessKey;
+ }
+
+ public String getSecretKey() {
+ return secretKey;
+ }
+
+ public boolean isPathStyleAccess() {
+ return pathStyleAccess;
+ }
+}
diff --git a/spark-doris-connector/spark-doris-connector-it/pom.xml
b/spark-doris-connector/spark-doris-connector-it/pom.xml
index b61946a..586c014 100644
--- a/spark-doris-connector/spark-doris-connector-it/pom.xml
+++ b/spark-doris-connector/spark-doris-connector-it/pom.xml
@@ -50,6 +50,28 @@
<artifactId>testcontainers</artifactId>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>software.amazon.awssdk</groupId>
+ <artifactId>s3</artifactId>
+ <version>${awssdk.version}</version>
+ <scope>test</scope>
+ <exclusions>
+ <exclusion>
+ <groupId>software.amazon.awssdk</groupId>
+ <artifactId>apache-client</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>software.amazon.awssdk</groupId>
+ <artifactId>netty-nio-client</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>software.amazon.awssdk</groupId>
+ <artifactId>url-connection-client</artifactId>
+ <version>${awssdk.version}</version>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
@@ -157,4 +179,4 @@
</plugins>
</build>
-</project>
\ No newline at end of file
+</project>
diff --git
a/spark-doris-connector/spark-doris-connector-it/src/test/java/org/apache/doris/spark/container/AbstractS3TvfTestBase.java
b/spark-doris-connector/spark-doris-connector-it/src/test/java/org/apache/doris/spark/container/AbstractS3TvfTestBase.java
new file mode 100644
index 0000000..b1e0671
--- /dev/null
+++
b/spark-doris-connector/spark-doris-connector-it/src/test/java/org/apache/doris/spark/container/AbstractS3TvfTestBase.java
@@ -0,0 +1,163 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.spark.container;
+
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.testcontainers.DockerClientFactory;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.wait.strategy.Wait;
+import org.testcontainers.utility.DockerImageName;
+import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
+import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
+import software.amazon.awssdk.http.urlconnection.UrlConnectionHttpClient;
+import software.amazon.awssdk.regions.Region;
+import software.amazon.awssdk.services.s3.S3Client;
+import software.amazon.awssdk.services.s3.S3Configuration;
+import software.amazon.awssdk.services.s3.model.CreateBucketRequest;
+import software.amazon.awssdk.services.s3.model.ListObjectsV2Request;
+
+import java.net.Inet4Address;
+import java.net.InetAddress;
+import java.net.NetworkInterface;
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.Enumeration;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+
+/** Shared Doris and MinIO environment for S3 TVF integration tests. */
+public abstract class AbstractS3TvfTestBase extends AbstractContainerTestBase {
+ protected static final String S3_BUCKET = "doris-tvf-it";
+ protected static final String S3_REGION = "us-east-1";
+ protected static final String S3_ACCESS_KEY = "minioadmin";
+ protected static final String S3_SECRET_KEY = "minioadmin";
+
+ private static final int MINIO_PORT = 9000;
+ private static final String MINIO_IMAGE =
+ "minio/minio:RELEASE.2024-10-13T13-34-11Z";
+
+ private static GenericContainer<?> minio;
+ private static S3Client s3Client;
+ private static String s3Endpoint;
+
+ @BeforeClass
+ public static void startObjectStorage() throws Exception {
+ minio =
+ new GenericContainer<>(DockerImageName.parse(MINIO_IMAGE))
+ .withEnv("MINIO_ROOT_USER", S3_ACCESS_KEY)
+ .withEnv("MINIO_ROOT_PASSWORD", S3_SECRET_KEY)
+ .withCommand("server", "/data")
+ .withExposedPorts(MINIO_PORT)
+
.waitingFor(Wait.forHttp("/minio/health/live").forPort(MINIO_PORT));
+ minio.start();
+
+ String dockerHost =
DockerClientFactory.instance().dockerHostIpAddress();
+ String endpointHost = resolveEndpointHost(dockerHost);
+ s3Endpoint = "http://"; + endpointHost + ":" +
minio.getMappedPort(MINIO_PORT);
+ s3Client = createS3Client(s3Endpoint);
+
s3Client.createBucket(CreateBucketRequest.builder().bucket(S3_BUCKET).build());
+ }
+
+ @AfterClass
+ public static void stopObjectStorage() {
+ if (s3Client != null) {
+ s3Client.close();
+ }
+ if (minio != null) {
+ minio.stop();
+ }
+ }
+
+ protected Map<String, String> s3TvfSinkOptions(
+ String tableIdentifier, String objectPrefix, String labelPrefix,
int batchSize) {
+ Map<String, String> options = new LinkedHashMap<>();
+ options.put("doris.fenodes", getFenodes());
+ options.put("doris.query.port", Integer.toString(getQueryPort()));
+ options.put("doris.table.identifier", tableIdentifier);
+ options.put("user", getDorisUsername());
+ options.put("password", getDorisPassword());
+ options.put("doris.sink.mode", "tvf");
+ options.put("doris.sink.label.prefix", labelPrefix);
+ options.put("doris.sink.batch.size", Integer.toString(batchSize));
+ options.put("doris.sink.s3.endpoint", s3Endpoint);
+ options.put("doris.sink.s3.region", S3_REGION);
+ options.put("doris.sink.s3.bucket", S3_BUCKET);
+ options.put("doris.sink.s3.prefix", objectPrefix);
+ options.put("doris.sink.s3.access-key", S3_ACCESS_KEY);
+ options.put("doris.sink.s3.secret-key", S3_SECRET_KEY);
+ options.put("doris.sink.s3.path-style-access", "true");
+ return options;
+ }
+
+ protected List<String> listObjectKeys(String prefix) {
+ List<String> keys = new ArrayList<>();
+ s3Client
+ .listObjectsV2(
+ ListObjectsV2Request.builder()
+ .bucket(S3_BUCKET)
+ .prefix(prefix)
+ .build())
+ .contents()
+ .forEach(object -> keys.add(object.key()));
+ return keys;
+ }
+
+ protected static String uniqueName(String prefix) {
+ return prefix + "_" + UUID.randomUUID().toString().replace("-",
"").substring(0, 8);
+ }
+
+ private static S3Client createS3Client(String endpoint) {
+ return S3Client.builder()
+ .endpointOverride(URI.create(endpoint))
+ .region(Region.of(S3_REGION))
+ .credentialsProvider(
+ StaticCredentialsProvider.create(
+ AwsBasicCredentials.create(S3_ACCESS_KEY,
S3_SECRET_KEY)))
+ .httpClientBuilder(UrlConnectionHttpClient.builder())
+ .serviceConfiguration(
+
S3Configuration.builder().pathStyleAccessEnabled(true).build())
+ .build();
+ }
+
+ private static String resolveEndpointHost(String dockerHost) throws
Exception {
+ InetAddress dockerAddress = InetAddress.getByName(dockerHost);
+ if (!dockerAddress.isLoopbackAddress() &&
!dockerAddress.isAnyLocalAddress()) {
+ return dockerHost;
+ }
+ Enumeration<NetworkInterface> interfaces =
NetworkInterface.getNetworkInterfaces();
+ while (interfaces.hasMoreElements()) {
+ NetworkInterface networkInterface = interfaces.nextElement();
+ if (!networkInterface.isUp() || networkInterface.isLoopback()) {
+ continue;
+ }
+ Enumeration<InetAddress> addresses =
networkInterface.getInetAddresses();
+ while (addresses.hasMoreElements()) {
+ InetAddress address = addresses.nextElement();
+ if (address instanceof Inet4Address
+ && !address.isLoopbackAddress()
+ && !address.isLinkLocalAddress()) {
+ return address.getHostAddress();
+ }
+ }
+ }
+ return dockerHost;
+ }
+}
diff --git
a/spark-doris-connector/spark-doris-connector-it/src/test/java/org/apache/doris/spark/container/instance/DorisContainer.java
b/spark-doris-connector/spark-doris-connector-it/src/test/java/org/apache/doris/spark/container/instance/DorisContainer.java
index 5220876..1e99536 100644
---
a/spark-doris-connector/spark-doris-connector-it/src/test/java/org/apache/doris/spark/container/instance/DorisContainer.java
+++
b/spark-doris-connector/spark-doris-connector-it/src/test/java/org/apache/doris/spark/container/instance/DorisContainer.java
@@ -49,7 +49,7 @@ import java.util.concurrent.locks.LockSupport;
public class DorisContainer implements ContainerService {
private static final Logger LOG =
LoggerFactory.getLogger(DorisContainer.class);
- private static final String DEFAULT_DOCKER_IMAGE =
"apache/doris:doris-all-in-one-2.1.0";
+ private static final String DEFAULT_DOCKER_IMAGE =
"jnsimba/doris-all-in-one:4.1.3";
private static final String DORIS_DOCKER_IMAGE =
System.getProperty("image") == null
? DEFAULT_DOCKER_IMAGE
@@ -208,15 +208,33 @@ public class DorisContainer implements ContainerService {
private void initializeJdbcConnection() throws Exception {
initializeJDBCDriver();
- try (Connection connection = getQueryConnection();
- Statement statement = connection.createStatement()) {
- ResultSet resultSet;
- do {
+ Duration timeout = Duration.ofMinutes(5L);
+ long startNanos = System.nanoTime();
+ Throwable lastFailure = null;
+ boolean frontendReady = false;
+ while (System.nanoTime() - startNanos < timeout.toNanos()) {
+ try (Connection connection = getQueryConnection();
+ Statement statement = connection.createStatement();
+ ResultSet resultSet = statement.executeQuery("show
backends")) {
+ frontendReady = true;
+ lastFailure = null;
LOG.info("Waiting for the Backend to start successfully.");
- resultSet = statement.executeQuery("show backends");
- } while (!isBeReady(resultSet, Duration.ofSeconds(1L)));
+ if (isBeReady(resultSet, Duration.ofSeconds(1L))) {
+ LOG.info("Connected to Doris successfully.");
+ return;
+ }
+ } catch (DorisRuntimeException | SQLException e) {
+ frontendReady = false;
+ lastFailure = e;
+ LOG.info("Waiting for the Frontend to accept connections.");
+ LockSupport.parkNanos(Duration.ofSeconds(1L).toNanos());
+ }
}
- LOG.info("Connected to Doris successfully.");
+ throw new DorisRuntimeException(
+ String.format(
+ "Doris %s did not become ready within %d seconds.",
+ frontendReady ? "Backend" : "Frontend",
timeout.getSeconds()),
+ lastFailure);
}
private boolean isBeReady(ResultSet rs, Duration duration) throws
SQLException {
diff --git
a/spark-doris-connector/spark-doris-connector-it/src/test/java/org/apache/doris/spark/sql/Doris2DorisE2ECase.scala
b/spark-doris-connector/spark-doris-connector-it/src/test/java/org/apache/doris/spark/sql/Doris2DorisE2ECase.scala
index 032195d..76ecccf 100644
---
a/spark-doris-connector/spark-doris-connector-it/src/test/java/org/apache/doris/spark/sql/Doris2DorisE2ECase.scala
+++
b/spark-doris-connector/spark-doris-connector-it/src/test/java/org/apache/doris/spark/sql/Doris2DorisE2ECase.scala
@@ -97,11 +97,16 @@ class Doris2DorisE2ECase(readMode: String, flightSqlPort:
Int) extends AbstractC
|""".stripMargin)
session.stop()
+ // TODO: Remove the legacy Scanner expectations after Doris returns
DATETIME as a
+ // timezone-naive Arrow timestamp.
+ val datetime1 = if (readMode == "arrow") "2025-03-11T12:34:56" else
"2025-03-11T04:34:56"
+ val datetime2 = if (readMode == "arrow") "2024-12-25T23:59:59" else
"2024-12-25T15:59:59"
+ val datetime3 = if (readMode == "arrow") "2023-06-15T08:00" else
"2023-06-15T00:00"
val excepted =
util.Arrays.asList(
-
"1,true,127,32767,2147483647,9223372036854775807,170141183460469231731687303715884105727,3.14,2.71828,12345.6789,2025-03-11,2025-03-11T12:34:56,A,Hello,
Doris!,This is a string,[\"Alice\", \"Bob\"],{\"key1\":\"value1\",
\"key2\":\"value2\"},{\"name\": \"Tom\", \"age\":
30},{\"key\":\"value\"},{\"data\":123,\"type\":\"variant\"}",
-
"2,false,-128,-32768,-2147483648,-9223372036854775808,-170141183460469231731687303715884105728,-1.23,1.0E-4,-9999.9999,2024-12-25,2024-12-25T23:59:59,B,Doris
Test,Another string!,[\"Charlie\", \"David\"],{\"k1\":\"v1\",
\"k2\":\"v2\"},{\"name\": \"Jerry\", \"age\":
25},{\"status\":\"ok\"},{\"data\":[1,2,3]}",
- "3,true,0,0,0,0,0,0.0,0.0,0.0000,2023-06-15,2023-06-15T08:00,C,Test
Doris,Sample text,[\"Eve\", \"Frank\"],{\"alpha\":\"beta\"},{\"name\":
\"Alice\", \"age\":
40},{\"nested\":{\"key\":\"value\"}},{\"variant\":\"test\"}",
+
"1,true,127,32767,2147483647,9223372036854775807,170141183460469231731687303715884105727,3.14,2.71828,12345.6789,2025-03-11,"
+ datetime1 + ",A,Hello, Doris!,This is a string,[\"Alice\",
\"Bob\"],{\"key1\":\"value1\", \"key2\":\"value2\"},{\"name\":\"Tom\",
\"age\":30},{\"key\":\"value\"},{\"data\":123,\"type\":\"variant\"}",
+
"2,false,-128,-32768,-2147483648,-9223372036854775808,-170141183460469231731687303715884105728,-1.23,1.0E-4,-9999.9999,2024-12-25,"
+ datetime2 + ",B,Doris Test,Another string!,[\"Charlie\",
\"David\"],{\"k1\":\"v1\", \"k2\":\"v2\"},{\"name\":\"Jerry\",
\"age\":25},{\"status\":\"ok\"},{\"data\":[1,2,3]}",
+ "3,true,0,0,0,0,0,0.0,0.0,0.0000,2023-06-15," + datetime3 + ",C,Test
Doris,Sample text,[\"Eve\",
\"Frank\"],{\"alpha\":\"beta\"},{\"name\":\"Alice\",
\"age\":40},{\"nested\":{\"key\":\"value\"}},{\"variant\":\"test\"}",
"4,null,null,null,null,null,null,null,null,null,null,null,null,null,null,null,null,null,null,null");
val query = String.format("select * from %s order by id",
TABLE_WRITE_TBL_ALL_TYPES)
diff --git
a/spark-doris-connector/spark-doris-connector-it/src/test/java/org/apache/doris/spark/sql/DorisReaderITCase.scala
b/spark-doris-connector/spark-doris-connector-it/src/test/java/org/apache/doris/spark/sql/DorisReaderITCase.scala
index 8815714..5653dfe 100644
---
a/spark-doris-connector/spark-doris-connector-it/src/test/java/org/apache/doris/spark/sql/DorisReaderITCase.scala
+++
b/spark-doris-connector/spark-doris-connector-it/src/test/java/org/apache/doris/spark/sql/DorisReaderITCase.scala
@@ -235,17 +235,21 @@ class DorisReaderITCase(readMode: String, flightSqlPort:
Int) extends AbstractCo
|select * from test_source order by id
|""".stripMargin).collect()
+ val expectedTimestamp1 = if (readMode == "thrift") "2025-03-11 04:34:56"
else "2025-03-11 12:34:56"
+ val expectedTimestamp2 = if (readMode == "thrift") "2024-12-25 15:59:59"
else "2024-12-25 23:59:59"
+ val expectedTimestamp3 = if (readMode == "thrift") "2023-06-15 00:00:00"
else "2023-06-15 08:00:00"
+
val expectedData = Array(
Row(1, true, 127, 32767, 2147483647, 9223372036854775807L,
"170141183460469231731687303715884105727",
- 3.14f, 2.71828, new java.math.BigDecimal("12345.6789"),
Date.valueOf("2025-03-11"), Timestamp.valueOf("2025-03-11 12:34:56"), "A",
"Hello, Doris!", "This is a string",
+ 3.14f, 2.71828, new java.math.BigDecimal("12345.6789"),
Date.valueOf("2025-03-11"), Timestamp.valueOf(expectedTimestamp1), "A", "Hello,
Doris!", "This is a string",
"""["Alice","Bob"]""", Map("key1" -> "value1", "key2" -> "value2"),
"""{"name":"Tom","age":30}""",
"""{"key":"value"}""", """{"type":"variant","data":123}"""),
Row(2, false, -128, -32768, -2147483648, -9223372036854775808L,
"-170141183460469231731687303715884105728",
- -1.23f, 0.0001, new java.math.BigDecimal("-9999.9999"),
Date.valueOf("2024-12-25"), Timestamp.valueOf("2024-12-25 23:59:59"), "B",
"Doris Test", "Another string!",
+ -1.23f, 0.0001, new java.math.BigDecimal("-9999.9999"),
Date.valueOf("2024-12-25"), Timestamp.valueOf(expectedTimestamp2), "B", "Doris
Test", "Another string!",
"""["Charlie","David"]""", Map("k1" -> "v1", "k2" -> "v2"),
"""{"name":"Jerry","age":25}""",
"""{"status":"ok"}""", """{"data":[1,2,3]}"""),
Row(3, true, 0, 0, 0, 0, "0",
- 0.0f, 0.0, new java.math.BigDecimal("0.0000"),
Date.valueOf("2023-06-15"), Timestamp.valueOf("2023-06-15 08:00:00"), "C",
"Test Doris", "Sample text",
+ 0.0f, 0.0, new java.math.BigDecimal("0.0000"),
Date.valueOf("2023-06-15"), Timestamp.valueOf(expectedTimestamp3), "C", "Test
Doris", "Sample text",
"""["Eve","Frank"]""", Map("alpha" -> "beta"),
"""{"name":"Alice","age":40}""",
"""{"nested":{"key":"value"}}""", """{"variant":"test"}"""),
Row(4, null, null, null, null, null, null,
@@ -366,7 +370,7 @@ class DorisReaderITCase(readMode: String, flightSqlPort:
Int) extends AbstractCo
|select * from test_source order by hour
|""".stripMargin).collect()
- assert("List([20200622,1,AfMAAAA=],
[20200622,2,AjswAQABAAAEAAYAAAABAAEABABvoQ==], [20200622,3,A91yV/pCAAAA])"
+ assert("List([20200622,1,BQHzAAAAAAAAAA==],
[20200622,2,BQYEAAAAAAAAAAEAAAAAAAAABQAAAAAAAAACAAAAAAAAAG+hBgAAAAAAAwAAAAAAAAA=],
[20200622,3,BQHdclf6QgAAAA==])"
.equals(actualData.toList.toString()))
} finally {
session.stop()
@@ -412,14 +416,24 @@ class DorisReaderITCase(readMode: String, flightSqlPort:
Int) extends AbstractCo
|select id,c10,c11 from test_source where c10 = '2025-03-11' and c13
like 'Hello%'
|""".stripMargin).collect()
- assert("List([1,2025-03-11,2025-03-11
12:34:56.0])".equals(dateFilter.toList.toString()))
+ val expectedDateFilter = if (readMode == "thrift") {
+ "List([1,2025-03-11,2025-03-11 04:34:56.0])"
+ } else {
+ "List([1,2025-03-11,2025-03-11 12:34:56.0])"
+ }
+ assert(expectedDateFilter.equals(dateFilter.toList.toString()))
val datetimeFilter = session.sql(
"""
|select id,c11,c12 from test_source where c10 < '2025-03-11' and c11
= '2024-12-25 23:59:59'
|""".stripMargin).collect()
- assert("List([2,2024-12-25
23:59:59.0,B])".equals(datetimeFilter.toList.toString()))
+ val expectedDatetimeFilter = if (readMode == "thrift") {
+ "List([2,2024-12-25 15:59:59.0,B])"
+ } else {
+ "List([2,2024-12-25 23:59:59.0,B])"
+ }
+ assert(expectedDatetimeFilter.equals(datetimeFilter.toList.toString()))
val stringFilter = session.sql(
"""
diff --git
a/spark-doris-connector/spark-doris-connector-it/src/test/java/org/apache/doris/spark/sql/DorisStreamingWriterITCase.scala
b/spark-doris-connector/spark-doris-connector-it/src/test/java/org/apache/doris/spark/sql/DorisStreamingWriterITCase.scala
index 530fda2..d6e1403 100644
---
a/spark-doris-connector/spark-doris-connector-it/src/test/java/org/apache/doris/spark/sql/DorisStreamingWriterITCase.scala
+++
b/spark-doris-connector/spark-doris-connector-it/src/test/java/org/apache/doris/spark/sql/DorisStreamingWriterITCase.scala
@@ -22,9 +22,12 @@ import
org.apache.doris.spark.container.{AbstractContainerTestBase, ContainerUti
import org.apache.doris.spark.rest.models.DataModel
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.streaming.{StreamingQueryListener,
StreamingQueryProgress}
+import org.junit.Assert.assertTrue
import org.junit.{Before, Test}
import org.slf4j.LoggerFactory
+import java.util.concurrent.TimeUnit
+
/**
* Doris Structured Streaming Test Case.
*/
@@ -83,9 +86,18 @@ class DorisStreamingWriterITCase extends
AbstractContainerTestBase {
override def onQueryTerminated(event:
StreamingQueryListener.QueryTerminatedEvent): Unit = {}
})
- dorissink.awaitTermination()
-
- spark.stop()
+ try {
+ assertTrue("Streaming query did not terminate within 60 seconds",
+ dorissink.awaitTermination(TimeUnit.SECONDS.toMillis(60)))
+ } finally {
+ try {
+ if (dorissink.isActive) {
+ dorissink.stop()
+ }
+ } finally {
+ spark.stop()
+ }
+ }
val cnt = ContainerUtils.executeSQLStatement(
getDorisQueryConnection(DATABASE),
LOG,
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 4381d4e..b1edd80 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
@@ -20,7 +20,7 @@ package org.apache.doris.spark.sql
import
org.apache.doris.spark.container.AbstractContainerTestBase.{assertEqualsInAnyOrder,
getDorisQueryConnection}
import org.apache.doris.spark.container.{AbstractContainerTestBase,
ContainerUtils}
import org.apache.doris.spark.rest.models.DataModel
-import org.apache.spark.SparkException
+import org.apache.spark.{SparkException, TaskContext}
import org.apache.spark.sql.SparkSession
import org.hamcrest.{CoreMatchers, Description, Matcher}
import org.junit.rules.ExpectedException
@@ -28,10 +28,8 @@ import org.junit.{Before, Rule, Test}
import org.slf4j.LoggerFactory
import java.util
-import java.util.UUID
-import java.util.concurrent.{Executors, TimeUnit}
+import java.util.concurrent.{Executors, Future, TimeUnit}
import scala.collection.JavaConverters._
-import scala.util.control.Breaks._
/**
* Test DorisWriter failover.
@@ -64,12 +62,12 @@ class DorisWriterFailoverITCase extends
AbstractContainerTestBase {
// 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)
+ initializeTable(TABLE_WRITE_TBL_RETRY, DataModel.UNIQUE, "varchar(256) NOT
NULL")
val session = SparkSession.builder().master("local[1]").getOrCreate()
val df = session.createDataFrame(Seq(
("doris", "1234"),
- ("spark", "123456"),
- ("catalog", "12345678")
+ ("spark", null),
+ ("catalog", null)
)).toDF("name", "address")
df.createTempView("mock_source")
@@ -85,7 +83,8 @@ class DorisWriterFailoverITCase extends
AbstractContainerTestBase {
| "doris.sink.retry.interval.ms"="10000",
| "doris.sink.batch.size"="1",
| "doris.sink.max-retries"="3",
- | "doris.sink.enable-2pc"="false"
+ | "doris.sink.enable-2pc"="false",
+ | "doris.sink.properties.strict_mode"="true"
|)
|""".stripMargin)
@@ -99,36 +98,33 @@ class DorisWriterFailoverITCase extends
AbstractContainerTestBase {
val query = String.format("SELECT * FROM %s.%s", DATABASE,
TABLE_WRITE_TBL_RETRY)
var result: util.List[String] = null
val connection = getDorisQueryConnection(DATABASE)
- breakable {
- while (true) {
+ try {
+ waitForCondition(future, "at least one row to be loaded") {
try {
- // query may be failed
+ // query may fail while the load is being retried
result = ContainerUtils.executeSQLStatement(connection, LOG, query,
2)
} catch {
case ex: Exception =>
LOG.error("Failed to query result, cause " + ex.getMessage)
}
-
- // until insert 1 rows
- if (result.size >= 1){
- Thread.sleep(5000)
- ContainerUtils.executeSQLStatement(
- connection,
- LOG,
- String.format("ALTER TABLE %s.%s MODIFY COLUMN address
varchar(256)", DATABASE, TABLE_WRITE_TBL_RETRY))
- break
- }
+ result != null && result.size >= 1
}
- }
+ ContainerUtils.executeSQLStatement(
+ connection,
+ LOG,
+ String.format("ALTER TABLE %s.%s MODIFY COLUMN address varchar(256)
NULL", DATABASE, TABLE_WRITE_TBL_RETRY))
- future.get(60, TimeUnit.SECONDS)
- session.stop()
+ future.get(60, TimeUnit.SECONDS)
+ } finally {
+ service.shutdownNow()
+ session.stop()
+ }
val actual = ContainerUtils.executeSQLStatement(
getDorisQueryConnection,
LOG,
String.format("select * from %s.%s", DATABASE, TABLE_WRITE_TBL_RETRY),
2)
- val expected = util.Arrays.asList("doris,1234", "spark,123456",
"catalog,12345678");
+ val expected = util.Arrays.asList("doris,1234", "spark,null",
"catalog,null");
checkResultInAnyOrder("testFailoverForRetry", expected.toArray,
actual.toArray)
}
@@ -140,15 +136,31 @@ class DorisWriterFailoverITCase extends
AbstractContainerTestBase {
def testFailoverForTaskRetry(): Unit = {
LOG.info("start to test testFailoverForTaskRetry.")
initializeTable(TABLE_WRITE_TBL_TASK_RETRY, DataModel.DUPLICATE)
- val session = SparkSession.builder().master("local[1,1000]").getOrCreate()
- val df = session.createDataFrame(Seq(
+ val session = SparkSession.builder().master("local[1,2]").getOrCreate()
+ import session.implicits._
+ val df = session.sparkContext.parallelize(Seq(
("doris", "cn"),
("spark", "us"),
("catalog", "uk")
- )).toDF("name", "address")
+ ), 1).mapPartitions { records =>
+ val attempt = TaskContext.get().attemptNumber()
+ var emitted = 0
+ new Iterator[(String, String)] {
+ override def hasNext: Boolean = {
+ if (attempt == 0 && emitted == 1) {
+ throw new RuntimeException("Trigger task retry after the first
row")
+ }
+ records.hasNext
+ }
+
+ override def next(): (String, String) = {
+ emitted += 1
+ records.next()
+ }
+ }
+ }.toDF("name", "address")
df.createTempView("mock_source")
- var uuid = UUID.randomUUID().toString
session.sql(
s"""
|CREATE TEMPORARY VIEW test_sink
@@ -158,50 +170,19 @@ class DorisWriterFailoverITCase extends
AbstractContainerTestBase {
| "fenodes"="${getFenodes}",
| "user"="${getDorisUsername}",
| "password"="${getDorisPassword}",
- | "doris.sink.batch.size"="1",
+ | "doris.sink.batch.size"="100",
| "doris.sink.batch.interval.ms"="1000",
| "doris.sink.max-retries"="0",
- | "doris.sink.enable-2pc"="true",
- | "doris.sink.label.prefix"='${uuid}'
+ | "doris.sink.enable-2pc"="true"
|)
|""".stripMargin)
- val service = Executors.newSingleThreadExecutor()
- val future = service.submit(new Runnable {
- override def run(): Unit = {
- session.sql("INSERT INTO test_sink SELECT * FROM mock_source")
- }
- })
-
- val query = "show transaction from " + DATABASE + " where label like '" +
uuid + "%'"
- var result: List[String] = null
- val connection = getDorisQueryConnection(DATABASE)
- breakable {
- while (true) {
- try {
- // query may be failed
- result = ContainerUtils.executeSQLStatement(connection, LOG, query,
15).asScala.toList
- Thread.sleep(10)
- } catch {
- case ex: Exception =>
- LOG.error("Failed to query result, cause " + ex.getMessage)
- }
-
- // until insert 1 rows
- if (result.size >= 1 && result.forall(s =>
s.contains("PRECOMMITTED"))){
- faultInjectionOpen()
- Thread.sleep(3000)
- faultInjectionClear()
- break
- }
- }
+ try {
+ session.sql("INSERT INTO test_sink SELECT * FROM mock_source")
+ } finally {
+ session.stop()
}
- future.get(60, TimeUnit.SECONDS)
- session.stop()
-
- //make sure publish success
- Thread.sleep(5000)
val actual = ContainerUtils.executeSQLStatement(
getDorisQueryConnection,
LOG,
@@ -211,8 +192,29 @@ class DorisWriterFailoverITCase extends
AbstractContainerTestBase {
checkResultInAnyOrder("testFailoverForTaskRetry", expected.toArray,
actual.toArray)
}
+ private def waitForCondition(future: Future[_], description:
String)(condition: => Boolean): Unit = {
+ val deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(60)
+ while (System.nanoTime() < deadline) {
+ if (future.isDone) {
+ future.get()
+ throw new AssertionError(s"Load completed before observing
$description")
+ }
+ if (condition) {
+ if (future.isDone) {
+ future.get()
+ throw new AssertionError(s"Load completed before observing
$description")
+ }
+ return
+ }
+ Thread.sleep(100)
+ }
+ throw new AssertionError(s"Timed out waiting for $description")
+ }
- private def initializeTable(table: String, dataModel: DataModel): Unit = {
+ private def initializeTable(
+ table: String,
+ dataModel: DataModel,
+ addressType: String = "varchar(4)"): Unit = {
val max = if (DataModel.AGGREGATE == dataModel) "MAX" else ""
val morProps = if (!(DataModel.UNIQUE_MOR == dataModel)) "" else
",\"enable_unique_key_merge_on_write\" = \"false\""
val model = if (dataModel == DataModel.UNIQUE_MOR)
DataModel.UNIQUE.toString else dataModel.toString
@@ -223,12 +225,12 @@ class DorisWriterFailoverITCase extends
AbstractContainerTestBase {
String.format("DROP TABLE IF EXISTS %s.%s", DATABASE, table),
String.format("CREATE TABLE %s.%s ( \n"
+ "`name` varchar(32),\n"
- + "`address` varchar(4) %s\n"
+ + "`address` %s %s\n"
+ ") "
+ " %s KEY(`name`) "
+ " DISTRIBUTED BY HASH(`name`) BUCKETS 1\n"
+ "PROPERTIES ("
- + "\"replication_num\" = \"1\"\n" + morProps + ")", DATABASE, table,
max, model))
+ + "\"replication_num\" = \"1\"\n" + morProps + ")", DATABASE, table,
addressType, max, model))
}
private def checkResultInAnyOrder(testName: String, expected: Array[AnyRef],
actual: Array[AnyRef]): Unit = {
diff --git
a/spark-doris-connector/spark-doris-connector-it/src/test/java/org/apache/doris/spark/sql/S3TvfE2ECase.scala
b/spark-doris-connector/spark-doris-connector-it/src/test/java/org/apache/doris/spark/sql/S3TvfE2ECase.scala
new file mode 100644
index 0000000..709e928
--- /dev/null
+++
b/spark-doris-connector/spark-doris-connector-it/src/test/java/org/apache/doris/spark/sql/S3TvfE2ECase.scala
@@ -0,0 +1,122 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.spark.sql
+
+import org.apache.doris.spark.container.{AbstractS3TvfTestBase, ContainerUtils}
+import
org.apache.doris.spark.container.AbstractContainerTestBase.getDorisQueryConnection
+import org.apache.doris.spark.container.AbstractS3TvfTestBase.uniqueName
+import org.apache.spark.sql.SparkSession
+import org.junit.Assert.assertTrue
+import org.junit.Test
+import org.slf4j.LoggerFactory
+
+import java.util
+import scala.collection.JavaConverters._
+
+/** End-to-end coverage for reading from Doris and writing through the S3 TVF.
*/
+class S3TvfE2ECase extends AbstractS3TvfTestBase {
+
+ private val LOG = LoggerFactory.getLogger(classOf[S3TvfE2ECase])
+
+ @Test
+ def testDorisToDorisThroughS3Tvf(): Unit = {
+ val database = uniqueName("test_s3_tvf_e2e")
+ val sourceTable = "source_table"
+ val targetTable = "target_table"
+ val prefix = uniqueName("e2e_objects")
+ val labelPrefix = uniqueName("e2e_label")
+
+ executeSql(
+ s"CREATE DATABASE $database",
+ createTableSql(database, sourceTable),
+ createTableSql(database, targetTable),
+ s"INSERT INTO $database.$sourceTable VALUES " +
+ "(1, 'alpha', 10.25, '2026-08-18', '中文')," +
+ "(2, 'beta', -2.50, '2026-08-19', 'quote-''and-\"')," +
+ "(3, 'gamma', 0.00, '2026-08-20', NULL)," +
+ "(4, 'delta', 9999.99, '2026-08-21', 'done')")
+
+ val session = SparkSession.builder()
+ .appName("s3-tvf-e2e")
+ .master("local[2]")
+ .config("spark.ui.enabled", "false")
+ .config("spark.driver.host", "127.0.0.1")
+ .config("spark.driver.bindAddress", "127.0.0.1")
+ .getOrCreate()
+ try {
+ val source = session.read
+ .format("doris")
+ .option("doris.fenodes", getFenodes)
+ .option("doris.table.identifier", s"$database.$sourceTable")
+ .option("user", getDorisUsername)
+ .option("password", getDorisPassword)
+ .load()
+ .select("id", "name", "amount", "event_date", "note")
+ .repartition(2)
+ source.createOrReplaceTempView("s3_tvf_source")
+
+ val sinkOptions = s3TvfSinkOptions(
+ s"$database.$targetTable", prefix, labelPrefix, 2)
+ session.sql(
+ s"CREATE TEMPORARY VIEW s3_tvf_sink USING doris OPTIONS(" +
+ renderOptions(sinkOptions) + ")")
+ session.sql(
+ "INSERT INTO s3_tvf_sink " +
+ "SELECT id, name, amount, event_date, note FROM s3_tvf_source")
+ } finally {
+ session.stop()
+ }
+
+ ContainerUtils.checkResult(
+ getDorisQueryConnection,
+ LOG,
+ util.Arrays.asList(
+ "1,alpha,10.25,2026-08-18,中文",
+ "2,beta,-2.50,2026-08-19,quote-'and-\"",
+ "3,gamma,0.00,2026-08-20,null",
+ "4,delta,9999.99,2026-08-21,done"),
+ s"SELECT id,name,amount,event_date,note " +
+ s"FROM $database.$targetTable ORDER BY id",
+ 5,
+ true)
+ assertTrue(listObjectKeys(prefix + "/").size() >= 2)
+ }
+
+ private def createTableSql(database: String, table: String): String = {
+ s"CREATE TABLE $database.$table (" +
+ "`id` INT, `name` VARCHAR(128), `amount` DECIMAL(12,2), " +
+ "`event_date` DATE, `note` VARCHAR(128) NULL) " +
+ "DUPLICATE KEY(`id`) DISTRIBUTED BY HASH(`id`) BUCKETS 1 " +
+ "PROPERTIES (\"replication_num\" = \"1\")"
+ }
+
+ private def renderOptions(options: java.util.Map[String, String]): String = {
+ options.asScala.toSeq
+ .sortBy(_._1)
+ .map { case (key, value) =>
+ s"'${escapeSqlLiteral(key)}'='${escapeSqlLiteral(value)}'"
+ }
+ .mkString(",")
+ }
+
+ private def escapeSqlLiteral(value: String): String = value.replace("'",
"''")
+
+ private def executeSql(sql: String*): Unit = {
+ ContainerUtils.executeSQLStatement(getDorisQueryConnection, LOG, sql: _*)
+ }
+}
diff --git
a/spark-doris-connector/spark-doris-connector-it/src/test/java/org/apache/doris/spark/sql/S3TvfSinkITCase.scala
b/spark-doris-connector/spark-doris-connector-it/src/test/java/org/apache/doris/spark/sql/S3TvfSinkITCase.scala
new file mode 100644
index 0000000..0f5176e
--- /dev/null
+++
b/spark-doris-connector/spark-doris-connector-it/src/test/java/org/apache/doris/spark/sql/S3TvfSinkITCase.scala
@@ -0,0 +1,318 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.spark.sql
+
+import org.apache.doris.spark.container.{AbstractS3TvfTestBase, ContainerUtils}
+import
org.apache.doris.spark.container.AbstractContainerTestBase.getDorisQueryConnection
+import org.apache.doris.spark.container.AbstractS3TvfTestBase.uniqueName
+import org.apache.spark.TaskContext
+import org.apache.spark.sql.{SaveMode, SparkSession}
+import org.junit.Assert.{assertEquals, assertFalse, assertTrue}
+import org.junit.{Before, Test}
+import org.slf4j.LoggerFactory
+
+import java.util
+import java.util.regex.Pattern
+import scala.collection.JavaConverters._
+
+/** Integration coverage for Spark batch writes through the Doris S3 TVF. */
+class S3TvfSinkITCase extends AbstractS3TvfTestBase {
+
+ private val LOG = LoggerFactory.getLogger(classOf[S3TvfSinkITCase])
+ private val database = "test_s3_tvf_sink"
+
+ @Before
+ def createDatabase(): Unit = {
+ executeSql(s"CREATE DATABASE IF NOT EXISTS $database")
+ }
+
+ @Test
+ def testMultiFileBatchWrite(): Unit = {
+ val table = uniqueName("duplicate_multi_file")
+ val prefix = uniqueName("objects") + "/"
+ val labelPrefix = uniqueName("label")
+ createDuplicateTable(
+ table,
+ "`id` INT, `name` VARCHAR(128), `note` VARCHAR(128) NULL")
+
+ withSpark("local[1]") { session =>
+ import session.implicits._
+ val rows = Seq(
+ (1, "doris", "中文"),
+ (2, "spark", "quote-'and-\""),
+ (3, "null-value", null.asInstanceOf[String]))
+ .toDF("id", "name", "note")
+ .coalesce(1)
+
+ rows.write
+ .format("doris")
+ .options(s3TvfSinkOptions(
+ s"$database.$table", prefix, labelPrefix, 2).asScala)
+ .mode(SaveMode.Append)
+ .save()
+ }
+
+ assertResult(
+ table,
+ "id,name,note",
+ util.Arrays.asList(
+ "1,doris,中文",
+ "2,spark,quote-'and-\"",
+ "3,null-value,null"),
+ columnCount = 3)
+
+ val keys = listObjectKeys(prefix).asScala
+ assertEquals(2, keys.size)
+ val keyPattern =
+ Pattern.quote(prefix + labelPrefix + "_" + table + "_") +
+ "[0-9a-f-]{36}_0_[0-9]+\\.json"
+ assertTrue(keys.forall(_.matches(keyPattern)))
+ assertFalse(keys.exists(_.contains("//")))
+ }
+
+ @Test
+ def testParallelTaskWrite(): Unit = {
+ val table = uniqueName("parallel")
+ val prefix = uniqueName("parallel_objects")
+ val labelPrefix = uniqueName("parallel_label")
+ createDuplicateTable(table, "`id` INT, `task_value` VARCHAR(128)")
+
+ withSpark("local[2]") { session =>
+ val rows = session.range(0, 20)
+ .selectExpr("CAST(id AS INT) AS id", "concat('value-', id) AS
task_value")
+ .repartition(2)
+ rows.write
+ .format("doris")
+ .options(s3TvfSinkOptions(
+ s"$database.$table", prefix, labelPrefix, 100).asScala)
+ .mode(SaveMode.Append)
+ .save()
+ }
+
+ val count = querySingleValue(s"SELECT COUNT(*) FROM $database.$table")
+ assertEquals("20", count)
+
+ val keys = listObjectKeys(prefix + "/").asScala
+ assertEquals(2, keys.size)
+ val keyPattern =
+ (Pattern.quote(prefix + "/" + labelPrefix + "_" + table + "_") +
+ "([0-9a-f-]{36})_([0-9]+)_0\\.json").r
+ val taskFiles = keys.map {
+ case keyPattern(uuid, partition) => uuid -> partition
+ case key => throw new AssertionError("Unexpected S3 TVF object key: " +
key)
+ }
+ assertEquals(Set("0", "1"), taskFiles.map(_._2).toSet)
+ assertEquals(2, taskFiles.map(_._1).toSet.size)
+ }
+
+ @Test
+ def testEmptyPartitionDoesNotInsert(): Unit = {
+ val table = uniqueName("empty_partition")
+ val prefix = uniqueName("empty_partition_objects")
+ val labelPrefix = uniqueName("empty_partition_label")
+ createDuplicateTable(table, "`id` INT, `name` VARCHAR(128)")
+
+ withSpark("local[2]") { session =>
+ import session.implicits._
+ val rows = session.sparkContext.parallelize(Seq((1, "only-row")), 2)
+ .toDF("id", "name")
+ assertEquals(2, rows.rdd.getNumPartitions)
+ rows.write
+ .format("doris")
+ .options(s3TvfSinkOptions(
+ s"$database.$table", prefix, labelPrefix, 100).asScala)
+ .mode(SaveMode.Append)
+ .save()
+ }
+
+ assertEquals("1", querySingleValue(s"SELECT COUNT(*) FROM
$database.$table"))
+ assertEquals(1, listObjectKeys(prefix + "/").size())
+ }
+
+ @Test
+ def testPartialColumnUpdate(): Unit = {
+ val table = uniqueName("partial_update")
+ val prefix = uniqueName("partial_update_objects")
+ val labelPrefix = uniqueName("partial_update_label")
+ createUniqueTable(table, "`id` INT, `name` VARCHAR(128), `score` INT")
+ executeSql(s"INSERT INTO $database.$table VALUES (1, 'before', 10)")
+
+ withSpark("local[1]") { session =>
+ import session.implicits._
+ val options = s3TvfSinkOptions(
+ s"$database.$table", prefix, labelPrefix, 100)
+ options.put("sink.properties.columns", "id,name")
+ options.put("sink.properties.partial_columns", "true")
+ Seq((1, "after"))
+ .toDF("id", "name")
+ .write
+ .format("doris")
+ .options(options.asScala)
+ .mode(SaveMode.Append)
+ .save()
+ }
+
+ assertResult(
+ table,
+ "id,name,score",
+ util.Collections.singletonList("1,after,10"),
+ columnCount = 3)
+ }
+
+ @Test
+ def testInsertFailureRetainsObjects(): Unit = {
+ val table = uniqueName("failed_insert")
+ val prefix = uniqueName("failed_objects")
+ val labelPrefix = uniqueName("failed_label")
+ createDuplicateTable(table, "`id` INT, `name` VARCHAR(128)")
+
+ var writeFailed = false
+ withSpark("local[1]") { session =>
+ import session.implicits._
+ val options = s3TvfSinkOptions(
+ s"$database.$table", prefix, labelPrefix, 100)
+ options.put("sink.properties.invalid-variable", "true")
+ try {
+ Seq((1, "failed-row"))
+ .toDF("id", "name")
+ .write
+ .format("doris")
+ .options(options.asScala)
+ .mode(SaveMode.Append)
+ .save()
+ } catch {
+ case _: Throwable => writeFailed = true
+ }
+ }
+ assertTrue("Expected the Doris INSERT to fail", writeFailed)
+
+ val keys = listObjectKeys(prefix + "/").asScala
+ assertTrue("Expected failed task attempts to retain staged objects",
keys.nonEmpty)
+ }
+
+ @Test
+ def testTaskRetryUsesNewUuid(): Unit = {
+ val table = uniqueName("task_retry")
+ val prefix = uniqueName("retry_objects")
+ val labelPrefix = uniqueName("retry_label")
+ createDuplicateTable(table, "`id` INT, `name` VARCHAR(128)")
+
+ withSpark("local[1,2]") { session =>
+ import session.implicits._
+ session.sparkContext.parallelize(Seq(
+ (1, "doris"),
+ (2, "spark"),
+ (3, "catalog")
+ ), 1).mapPartitions { records =>
+ val attempt = TaskContext.get().attemptNumber()
+ var emitted = 0
+ new Iterator[(Int, String)] {
+ override def hasNext: Boolean = {
+ if (attempt == 0 && emitted == 1) {
+ throw new RuntimeException("Trigger task retry after the first
row")
+ }
+ records.hasNext
+ }
+
+ override def next(): (Int, String) = {
+ emitted += 1
+ records.next()
+ }
+ }
+ }.toDF("id", "name")
+ .write
+ .format("doris")
+ .options(s3TvfSinkOptions(
+ s"$database.$table", prefix, labelPrefix, 1).asScala)
+ .mode(SaveMode.Append)
+ .save()
+ }
+
+ assertEquals("3", querySingleValue(s"SELECT COUNT(*) FROM
$database.$table"))
+ val keys = listObjectKeys(prefix + "/").asScala
+ val keyPattern =
+ (Pattern.quote(prefix + "/" + labelPrefix + "_" + table + "_") +
+ "([0-9a-f-]{36})_[0-9]+_[0-9]+\\.json").r
+ val uuids = keys.map {
+ case keyPattern(uuid) => uuid
+ case key => throw new AssertionError("Unexpected S3 TVF object key: " +
key)
+ }.toSet
+ assertTrue("Expected task retry to use a new UUID", uuids.size >= 2)
+ }
+
+ private def createDuplicateTable(table: String, columns: String): Unit = {
+ createTable(table, columns, s"DUPLICATE KEY(`id`)", "")
+ }
+
+ private def createUniqueTable(table: String, columns: String): Unit = {
+ createTable(
+ table,
+ columns,
+ "UNIQUE KEY(`id`)",
+ ", \"enable_unique_key_merge_on_write\" = \"true\"")
+ }
+
+ private def createTable(
+ table: String,
+ columns: String,
+ keyDefinition: String,
+ additionalProperties: String): Unit = {
+ executeSql(
+ s"DROP TABLE IF EXISTS $database.$table",
+ s"CREATE TABLE $database.$table ($columns) $keyDefinition " +
+ "DISTRIBUTED BY HASH(`id`) BUCKETS 1 " +
+ s"""PROPERTIES ("replication_num" = "1"$additionalProperties)""")
+ }
+
+ private def assertResult(
+ table: String,
+ columns: String,
+ expected: util.List[String],
+ columnCount: Int): Unit = {
+ ContainerUtils.checkResult(
+ getDorisQueryConnection,
+ LOG,
+ expected,
+ s"SELECT $columns FROM $database.$table ORDER BY $columns",
+ columnCount,
+ true)
+ }
+
+ private def querySingleValue(sql: String): String = {
+ ContainerUtils.executeSQLStatement(getDorisQueryConnection, LOG, sql,
1).get(0)
+ }
+
+ private def executeSql(sql: String*): Unit = {
+ ContainerUtils.executeSQLStatement(getDorisQueryConnection, LOG, sql: _*)
+ }
+
+ private def withSpark(master: String)(test: SparkSession => Unit): Unit = {
+ val session = SparkSession.builder()
+ .appName("s3-tvf-it")
+ .master(master)
+ .config("spark.ui.enabled", "false")
+ .config("spark.driver.host", "127.0.0.1")
+ .config("spark.driver.bindAddress", "127.0.0.1")
+ .getOrCreate()
+ try {
+ test(session)
+ } finally {
+ session.stop()
+ }
+ }
+}
diff --git
a/spark-doris-connector/spark-doris-connector-it/src/test/resources/docker/doris/fe.conf
b/spark-doris-connector/spark-doris-connector-it/src/test/resources/docker/doris/fe.conf
index a45fb53..012bc9c 100644
---
a/spark-doris-connector/spark-doris-connector-it/src/test/resources/docker/doris/fe.conf
+++
b/spark-doris-connector/spark-doris-connector-it/src/test/resources/docker/doris/fe.conf
@@ -27,7 +27,7 @@ CUR_DATE=`date +%Y%m%d-%H%M%S`
LOG_DIR = ${DORIS_HOME}/log
# For jdk 17, this JAVA_OPTS will be used as default JVM options
-JAVA_OPTS_FOR_JDK_17="-Dfile.encoding=UTF-8
-Djavax.security.auth.useSubjectCredsOnly=false -Xmx8192m -Xms8192m
-XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=$LOG_DIR
-Xlog:gc*,classhisto*=trace:$LOG_DIR/fe.gc.log.$CUR_DATE:time,uptime:filecount=10,filesize=50M
--add-opens=java.base/java.nio=ALL-UNNAMED --add-opens
java.base/jdk.internal.ref=ALL-UNNAMED"
+JAVA_OPTS_FOR_JDK_17="-Dfile.encoding=UTF-8
-Djavax.security.auth.useSubjectCredsOnly=false -Xmx8192m -Xms8192m
-XX:-UseContainerSupport -XX:+HeapDumpOnOutOfMemoryError
-XX:HeapDumpPath=$LOG_DIR
-Xlog:gc*,classhisto*=trace:$LOG_DIR/fe.gc.log.$CUR_DATE:time,uptime:filecount=10,filesize=50M
--add-opens=java.base/java.nio=ALL-UNNAMED --add-opens
java.base/jdk.internal.ref=ALL-UNNAMED"
# Set your own JAVA_HOME
# JAVA_HOME=/path/to/jdk/
@@ -71,4 +71,4 @@ sys_log_mode = ASYNC
# meta_delay_toleration_second = 10
# qe_max_connection = 1024
# qe_query_timeout_second = 300
-# qe_slow_log_ms = 5000
\ No newline at end of file
+# qe_slow_log_ms = 5000
diff --git
a/spark-doris-connector/spark-doris-connector-it/src/test/resources/log4j.properties
b/spark-doris-connector/spark-doris-connector-it/src/test/resources/log4j.properties
new file mode 100644
index 0000000..1fd2558
--- /dev/null
+++
b/spark-doris-connector/spark-doris-connector-it/src/test/resources/log4j.properties
@@ -0,0 +1,24 @@
+################################################################################
+# 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.
+################################################################################
+
+log4j.rootLogger=INFO, console
+
+log4j.appender.console=org.apache.log4j.ConsoleAppender
+log4j.appender.console.target=System.err
+log4j.appender.console.layout=org.apache.log4j.PatternLayout
+log4j.appender.console.layout.ConversionPattern=%d{yy/MM/dd HH:mm:ss} %p [%t]
%c{1}: %m%n
diff --git a/spark-doris-connector/spark-doris-connector-spark-2/pom.xml
b/spark-doris-connector/spark-doris-connector-spark-2/pom.xml
index b402084..b990e27 100644
--- a/spark-doris-connector/spark-doris-connector-spark-2/pom.xml
+++ b/spark-doris-connector/spark-doris-connector-spark-2/pom.xml
@@ -52,6 +52,16 @@
<artifactId>spark-sql_${scala.major.version}</artifactId>
<scope>provided</scope>
</dependency>
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.hamcrest</groupId>
+ <artifactId>hamcrest-core</artifactId>
+ <scope>test</scope>
+ </dependency>
</dependencies>
-</project>
\ No newline at end of file
+</project>
diff --git
a/spark-doris-connector/spark-doris-connector-spark-2/src/main/scala/org/apache/doris/spark/sql/DorisSourceProvider.scala
b/spark-doris-connector/spark-doris-connector-spark-2/src/main/scala/org/apache/doris/spark/sql/DorisSourceProvider.scala
index 86d18dc..b2fb7e5 100644
---
a/spark-doris-connector/spark-doris-connector-spark-2/src/main/scala/org/apache/doris/spark/sql/DorisSourceProvider.scala
+++
b/spark-doris-connector/spark-doris-connector-spark-2/src/main/scala/org/apache/doris/spark/sql/DorisSourceProvider.scala
@@ -23,6 +23,7 @@ import org.apache.doris.spark.load.CommitMessage
import org.apache.doris.spark.sql.DorisSourceProvider.SHORT_NAME
import org.apache.doris.spark.sql.sources.{DorisRelation,
DorisSourceRegisterTrait}
import org.apache.doris.spark.writer.DorisWriter
+import org.apache.doris.spark.writer.S3TvfBatchWriter
import org.apache.spark.sql.execution.streaming.Sink
import org.apache.spark.sql.sources._
import org.apache.spark.sql.streaming.OutputMode
@@ -62,11 +63,15 @@ private[sql] class DorisSourceProvider extends
DorisSourceRegisterTrait
case _: SaveMode => // do nothing
}
- // accumulator for transaction handling
- val acc =
sqlContext.sparkContext.collectionAccumulator[CommitMessage]("BatchTxnAcc")
- // init stream loader
- val writer = new DorisWriter(config, acc, false)
- writer.write(data)
+ if (config.getValue(DorisOptions.LOAD_MODE) == "tvf") {
+ new S3TvfBatchWriter(config).write(data)
+ } else {
+ // accumulator for transaction handling
+ val acc =
sqlContext.sparkContext.collectionAccumulator[CommitMessage]("BatchTxnAcc")
+ // init stream loader
+ val writer = new DorisWriter(config, acc, false)
+ writer.write(data)
+ }
new BaseRelation {
override def sqlContext: SQLContext = unsupportedException
@@ -85,7 +90,11 @@ private[sql] class DorisSourceProvider extends
DorisSourceRegisterTrait
}
override def createSink(sqlContext: SQLContext, parameters: Map[String,
String], partitionColumns: Seq[String], outputMode: OutputMode): Sink = {
- new DorisStreamLoadSink(sqlContext,
DorisConfig.fromMap(Utils.params(parameters, logger).asJava, false))
+ val config = DorisConfig.fromMap(Utils.params(parameters, logger).asJava,
false)
+ if (config.getValue(DorisOptions.LOAD_MODE) == "tvf") {
+ throw new UnsupportedOperationException("tvf write mode does not support
Structured Streaming")
+ }
+ new DorisStreamLoadSink(sqlContext, config)
}
private def truncateTable(config: DorisConfig): Unit = {
@@ -106,4 +115,4 @@ private[sql] class DorisSourceProvider extends
DorisSourceRegisterTrait
object DorisSourceProvider {
val SHORT_NAME: String = "doris"
-}
\ No newline at end of file
+}
diff --git
a/spark-doris-connector/spark-doris-connector-spark-2/src/main/scala/org/apache/doris/spark/writer/S3TvfBatchWriter.scala
b/spark-doris-connector/spark-doris-connector-spark-2/src/main/scala/org/apache/doris/spark/writer/S3TvfBatchWriter.scala
new file mode 100644
index 0000000..5a843cb
--- /dev/null
+++
b/spark-doris-connector/spark-doris-connector-spark-2/src/main/scala/org/apache/doris/spark/writer/S3TvfBatchWriter.scala
@@ -0,0 +1,86 @@
+// 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.writer
+
+import org.apache.doris.spark.client.write.tvf.{
+ S3ClientObjectStore,
+ S3TvfColumnUtils,
+ S3TvfCommitter,
+ S3TvfWriter
+}
+import org.apache.doris.spark.config.{DorisConfig, DorisOptions, S3TvfOptions}
+import org.apache.spark.sql.DataFrame
+
+/** Spark 2 batch adapter for the shared S3 TVF writer and committer. */
+class S3TvfBatchWriter(config: DorisConfig) extends Serializable {
+
+ def write(dataFrame: DataFrame): Unit = {
+ val resultDataFrame = repartition(dataFrame)
+ val schema = resultDataFrame.schema
+ val options = S3TvfOptions.fromConfig(config)
+ val tableIdentifier =
config.getValue(DorisOptions.DORIS_TABLE_IDENTIFIER).split("\\.")
+ val database = tableIdentifier(0).replaceAll("`", "").trim
+ val table = tableIdentifier(1).replaceAll("`", "").trim
+ val labelPrefix =
config.getValue(DorisOptions.DORIS_SINK_LABEL_PREFIX).trim
+ val columns = S3TvfColumnUtils.resolveColumns(config.getSinkProperties,
schema)
+ val batchSize = config.getValue(DorisOptions.DORIS_SINK_BATCH_SIZE)
+
+ resultDataFrame.queryExecution.toRdd
+ .mapPartitionsWithIndex { case (partitionId, records) =>
+ val writer = new S3TvfWriter(
+ options,
+ database,
+ table,
+ labelPrefix,
+ schema,
+ columns,
+ partitionId,
+ batchSize,
+ new S3ClientObjectStore(options))
+ try {
+ records.foreach(writer.write)
+ val committable = writer.prepareCommit()
+ if (!committable.isEmpty) {
+ val committer = new S3TvfCommitter(config)
+ try {
+ committer.commit(committable)
+ } finally {
+ committer.close()
+ }
+ }
+ Iterator.single(())
+ } finally {
+ writer.close()
+ }
+ }
+ .count()
+ }
+
+ private def repartition(dataFrame: DataFrame): DataFrame = {
+ if (!config.contains(DorisOptions.DORIS_SINK_TASK_PARTITION_SIZE)) {
+ dataFrame
+ } else {
+ val partitionSize =
config.getValue(DorisOptions.DORIS_SINK_TASK_PARTITION_SIZE)
+ if (config.getValue(DorisOptions.DORIS_SINK_TASK_USE_REPARTITION)) {
+ dataFrame.repartition(partitionSize)
+ } else {
+ dataFrame.coalesce(partitionSize)
+ }
+ }
+ }
+}
diff --git a/spark-doris-connector/spark-doris-connector-spark-3-base/pom.xml
b/spark-doris-connector/spark-doris-connector-spark-3-base/pom.xml
index 3d7499d..6622765 100644
--- a/spark-doris-connector/spark-doris-connector-spark-3-base/pom.xml
+++ b/spark-doris-connector/spark-doris-connector-spark-3-base/pom.xml
@@ -46,6 +46,16 @@
<artifactId>spark-sql_${scala.major.version}</artifactId>
<scope>provided</scope>
</dependency>
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.hamcrest</groupId>
+ <artifactId>hamcrest-core</artifactId>
+ <scope>test</scope>
+ </dependency>
</dependencies>
</project>
diff --git
a/spark-doris-connector/spark-doris-connector-spark-3-base/src/main/scala/org/apache/doris/spark/write/DorisWrite.scala
b/spark-doris-connector/spark-doris-connector-spark-3-base/src/main/scala/org/apache/doris/spark/write/DorisWrite.scala
index e5fddaf..a40b0a2 100644
---
a/spark-doris-connector/spark-doris-connector-spark-3-base/src/main/scala/org/apache/doris/spark/write/DorisWrite.scala
+++
b/spark-doris-connector/spark-doris-connector-spark-3-base/src/main/scala/org/apache/doris/spark/write/DorisWrite.scala
@@ -28,7 +28,9 @@ class DorisWrite(config: DorisConfig, schema: StructType)
extends BatchWrite wit
private val LOG = LoggerFactory.getLogger(classOf[DorisWrite])
- private val committer: DorisCommitter =
config.getValue(DorisOptions.LOAD_MODE) match {
+ private val loadMode = config.getValue(DorisOptions.LOAD_MODE)
+
+ private lazy val driverCommitter: DorisCommitter = loadMode match {
case "stream_load" => new StreamLoadProcessor(config, schema)
case "copy_into" => new CopyIntoProcessor(config, schema)
case _ => throw new IllegalArgumentException()
@@ -39,29 +41,36 @@ class DorisWrite(config: DorisConfig, schema: StructType)
extends BatchWrite wit
private val committedEpochLock = new AnyRef
override def createBatchWriterFactory(physicalWriteInfo: PhysicalWriteInfo):
DataWriterFactory = {
- new DorisDataWriterFactory(config, schema)
+ if (loadMode == "tvf") {
+ new S3TvfDataWriterFactory(config, schema)
+ } else {
+ new DorisDataWriterFactory(config, schema)
+ }
}
// for batch write
override def commit(writerCommitMessages: Array[WriterCommitMessage]): Unit
= {
- if (writerCommitMessages != null && writerCommitMessages.nonEmpty) {
+ if (loadMode != "tvf" && writerCommitMessages != null &&
writerCommitMessages.nonEmpty) {
writerCommitMessages.filter(_ != null)
-
.foreach(_.asInstanceOf[DorisWriterCommitMessage].commitMessages.foreach(committer.commit))
+
.foreach(_.asInstanceOf[DorisWriterCommitMessage].commitMessages.foreach(driverCommitter.commit))
}
}
// for batch write
override def abort(writerCommitMessages: Array[WriterCommitMessage]): Unit =
{
- LOG.info("writerCommitMessages size: " + writerCommitMessages.length)
- if (writerCommitMessages.exists(_ != null) &&
writerCommitMessages.nonEmpty) {
+ if (loadMode != "tvf" && writerCommitMessages != null &&
writerCommitMessages.nonEmpty) {
+ LOG.info("writerCommitMessages size: " + writerCommitMessages.length)
writerCommitMessages.filter(_ != null)
-
.foreach(_.asInstanceOf[DorisWriterCommitMessage].commitMessages.foreach(committer.abort))
+
.foreach(_.asInstanceOf[DorisWriterCommitMessage].commitMessages.foreach(driverCommitter.abort))
}
}
override def useCommitCoordinator(): Boolean = true
override def createStreamingWriterFactory(physicalWriteInfo:
PhysicalWriteInfo): StreamingDataWriterFactory = {
+ if (loadMode == "tvf") {
+ throw new UnsupportedOperationException("tvf write mode does not support
Structured Streaming")
+ }
new DorisDataWriterFactory(config, schema)
}
@@ -69,7 +78,7 @@ class DorisWrite(config: DorisConfig, schema: StructType)
extends BatchWrite wit
override def commit(epochId: Long, writerCommitMessages:
Array[WriterCommitMessage]): Unit = {
committedEpochLock.synchronized {
if (lastCommittedEpoch.isEmpty || epochId > lastCommittedEpoch.get &&
writerCommitMessages.exists(_ != null)) {
-
writerCommitMessages.foreach(_.asInstanceOf[DorisWriterCommitMessage].commitMessages.foreach(committer.commit))
+
writerCommitMessages.foreach(_.asInstanceOf[DorisWriterCommitMessage].commitMessages.foreach(driverCommitter.commit))
lastCommittedEpoch = Some(epochId)
}
}
@@ -79,9 +88,9 @@ class DorisWrite(config: DorisConfig, schema: StructType)
extends BatchWrite wit
override def abort(epochId: Long, writerCommitMessages:
Array[WriterCommitMessage]): Unit = {
committedEpochLock.synchronized {
if ((lastCommittedEpoch.isEmpty || epochId > lastCommittedEpoch.get) &&
writerCommitMessages.exists(_ != null)) {
-
writerCommitMessages.foreach(_.asInstanceOf[DorisWriterCommitMessage].commitMessages.foreach(committer.abort))
+
writerCommitMessages.foreach(_.asInstanceOf[DorisWriterCommitMessage].commitMessages.foreach(driverCommitter.abort))
}
}
}
-}
\ No newline at end of file
+}
diff --git
a/spark-doris-connector/spark-doris-connector-spark-3-base/src/main/scala/org/apache/doris/spark/write/DorisWriteBuilder.scala
b/spark-doris-connector/spark-doris-connector-spark-3-base/src/main/scala/org/apache/doris/spark/write/DorisWriteBuilder.scala
index b74cc45..c62e674 100644
---
a/spark-doris-connector/spark-doris-connector-spark-3-base/src/main/scala/org/apache/doris/spark/write/DorisWriteBuilder.scala
+++
b/spark-doris-connector/spark-doris-connector-spark-3-base/src/main/scala/org/apache/doris/spark/write/DorisWriteBuilder.scala
@@ -37,6 +37,9 @@ class DorisWriteBuilder(config: DorisConfig, schema:
StructType) extends WriteBu
}
override def buildForStreaming(): StreamingWrite = {
+ if (config.getValue(DorisOptions.LOAD_MODE) == "tvf") {
+ throw new UnsupportedOperationException("tvf write mode does not support
Structured Streaming")
+ }
new DorisWrite(config, schema)
}
diff --git
a/spark-doris-connector/spark-doris-connector-spark-3-base/src/main/scala/org/apache/doris/spark/write/S3TvfDataWriter.scala
b/spark-doris-connector/spark-doris-connector-spark-3-base/src/main/scala/org/apache/doris/spark/write/S3TvfDataWriter.scala
new file mode 100644
index 0000000..07a1a70
--- /dev/null
+++
b/spark-doris-connector/spark-doris-connector-spark-3-base/src/main/scala/org/apache/doris/spark/write/S3TvfDataWriter.scala
@@ -0,0 +1,74 @@
+// 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.tvf.{S3ClientObjectStore,
S3TvfCommitter, S3TvfWriter}
+import org.apache.doris.spark.config.{DorisConfig, S3TvfOptions}
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.connector.write.{DataWriter, WriterCommitMessage}
+import org.apache.spark.sql.types.StructType
+
+private[write] class S3TvfDataWriter(
+ config: DorisConfig,
+ options: S3TvfOptions,
+ database: String,
+ table: String,
+ labelPrefix: String,
+ schema: StructType,
+ columns: java.util.List[String],
+ partitionId: Int,
+ batchSize: Int) extends DataWriter[InternalRow] {
+
+ private val writer = new S3TvfWriter(
+ options,
+ database,
+ table,
+ labelPrefix,
+ schema,
+ columns,
+ partitionId,
+ batchSize,
+ new S3ClientObjectStore(options))
+ private var closed = false
+
+ override def write(record: InternalRow): Unit = writer.write(record)
+
+ override def commit(): WriterCommitMessage = {
+ val committable = writer.prepareCommit()
+ if (!committable.isEmpty) {
+ val committer = new S3TvfCommitter(config)
+ try {
+ committer.commit(committable)
+ } finally {
+ committer.close()
+ }
+ }
+ S3TvfWriterCommitMessage
+ }
+
+ override def abort(): Unit = close()
+
+ override def close(): Unit = {
+ if (!closed) {
+ writer.close()
+ closed = true
+ }
+ }
+}
+
+private[write] case object S3TvfWriterCommitMessage extends WriterCommitMessage
diff --git
a/spark-doris-connector/spark-doris-connector-spark-3-base/src/main/scala/org/apache/doris/spark/write/S3TvfDataWriterFactory.scala
b/spark-doris-connector/spark-doris-connector-spark-3-base/src/main/scala/org/apache/doris/spark/write/S3TvfDataWriterFactory.scala
new file mode 100644
index 0000000..7d2c0e5
--- /dev/null
+++
b/spark-doris-connector/spark-doris-connector-spark-3-base/src/main/scala/org/apache/doris/spark/write/S3TvfDataWriterFactory.scala
@@ -0,0 +1,50 @@
+// 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.tvf.S3TvfColumnUtils
+import org.apache.doris.spark.config.{DorisConfig, DorisOptions, S3TvfOptions}
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.connector.write.{DataWriter, DataWriterFactory}
+import org.apache.spark.sql.types.StructType
+
+private[write] class S3TvfDataWriterFactory(
+ config: DorisConfig,
+ schema: StructType) extends DataWriterFactory {
+
+ private val options = S3TvfOptions.fromConfig(config)
+ private val tableIdentifier =
config.getValue(DorisOptions.DORIS_TABLE_IDENTIFIER).split("\\.")
+ private val database = tableIdentifier(0).replaceAll("`", "").trim
+ private val table = tableIdentifier(1).replaceAll("`", "").trim
+ private val labelPrefix =
config.getValue(DorisOptions.DORIS_SINK_LABEL_PREFIX).trim
+ private val columns =
S3TvfColumnUtils.resolveColumns(config.getSinkProperties, schema)
+ private val batchSize = config.getValue(DorisOptions.DORIS_SINK_BATCH_SIZE)
+
+ override def createWriter(partitionId: Int, taskId: Long):
DataWriter[InternalRow] = {
+ new S3TvfDataWriter(
+ config,
+ options,
+ database,
+ table,
+ labelPrefix,
+ schema,
+ columns,
+ partitionId,
+ batchSize)
+ }
+}
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
index 61f1ee5..570624b 100644
--- a/spark-doris-connector/spark-doris-connector-spark-4-base/pom.xml
+++ b/spark-doris-connector/spark-doris-connector-spark-4-base/pom.xml
@@ -46,6 +46,16 @@
<artifactId>spark-sql_${scala.major.version}</artifactId>
<scope>provided</scope>
</dependency>
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.hamcrest</groupId>
+ <artifactId>hamcrest-core</artifactId>
+ <scope>test</scope>
+ </dependency>
</dependencies>
</project>
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
index 52233a7..c0c80d5 100644
---
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
@@ -28,7 +28,9 @@ class DorisWrite(config: DorisConfig, schema: StructType)
extends BatchWrite wit
private val LOG = LoggerFactory.getLogger(classOf[DorisWrite])
- private val committer: DorisCommitter =
config.getValue(DorisOptions.LOAD_MODE) match {
+ private val loadMode = config.getValue(DorisOptions.LOAD_MODE)
+
+ private lazy val driverCommitter: DorisCommitter = loadMode match {
case "stream_load" => new StreamLoadProcessor(config, schema)
case "copy_into" => new CopyIntoProcessor(config, schema)
case _ => throw new IllegalArgumentException()
@@ -39,29 +41,36 @@ class DorisWrite(config: DorisConfig, schema: StructType)
extends BatchWrite wit
private val committedEpochLock = new AnyRef
override def createBatchWriterFactory(physicalWriteInfo: PhysicalWriteInfo):
DataWriterFactory = {
- new DorisDataWriterFactory(config, schema)
+ if (loadMode == "tvf") {
+ new S3TvfDataWriterFactory(config, schema)
+ } else {
+ new DorisDataWriterFactory(config, schema)
+ }
}
// for batch write
override def commit(writerCommitMessages: Array[WriterCommitMessage]): Unit
= {
- if (writerCommitMessages != null && writerCommitMessages.nonEmpty) {
+ if (loadMode != "tvf" && writerCommitMessages != null &&
writerCommitMessages.nonEmpty) {
writerCommitMessages.filter(_ != null)
-
.foreach(_.asInstanceOf[DorisWriterCommitMessage].commitMessages.foreach(committer.commit))
+
.foreach(_.asInstanceOf[DorisWriterCommitMessage].commitMessages.foreach(driverCommitter.commit))
}
}
// for batch write
override def abort(writerCommitMessages: Array[WriterCommitMessage]): Unit =
{
- if (writerCommitMessages != null && writerCommitMessages.nonEmpty) {
+ if (loadMode != "tvf" && writerCommitMessages != null &&
writerCommitMessages.nonEmpty) {
LOG.info("writerCommitMessages size: " + writerCommitMessages.length)
writerCommitMessages.filter(_ != null)
-
.foreach(_.asInstanceOf[DorisWriterCommitMessage].commitMessages.foreach(committer.abort))
+
.foreach(_.asInstanceOf[DorisWriterCommitMessage].commitMessages.foreach(driverCommitter.abort))
}
}
override def useCommitCoordinator(): Boolean = true
override def createStreamingWriterFactory(physicalWriteInfo:
PhysicalWriteInfo): StreamingDataWriterFactory = {
+ if (loadMode == "tvf") {
+ throw new UnsupportedOperationException("tvf write mode does not support
Structured Streaming")
+ }
new DorisDataWriterFactory(config, schema)
}
@@ -71,7 +80,7 @@ class DorisWrite(config: DorisConfig, schema: StructType)
extends BatchWrite wit
if ((lastCommittedEpoch.isEmpty || epochId > lastCommittedEpoch.get)
&& writerCommitMessages != null && writerCommitMessages.exists(_ !=
null)) {
writerCommitMessages.filter(_ != null)
-
.foreach(_.asInstanceOf[DorisWriterCommitMessage].commitMessages.foreach(committer.commit))
+
.foreach(_.asInstanceOf[DorisWriterCommitMessage].commitMessages.foreach(driverCommitter.commit))
lastCommittedEpoch = Some(epochId)
}
}
@@ -83,9 +92,9 @@ class DorisWrite(config: DorisConfig, schema: StructType)
extends BatchWrite wit
if ((lastCommittedEpoch.isEmpty || epochId > lastCommittedEpoch.get)
&& writerCommitMessages != null && writerCommitMessages.exists(_ !=
null)) {
writerCommitMessages.filter(_ != null)
-
.foreach(_.asInstanceOf[DorisWriterCommitMessage].commitMessages.foreach(committer.abort))
+
.foreach(_.asInstanceOf[DorisWriterCommitMessage].commitMessages.foreach(driverCommitter.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
index b74cc45..c62e674 100644
---
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
@@ -37,6 +37,9 @@ class DorisWriteBuilder(config: DorisConfig, schema:
StructType) extends WriteBu
}
override def buildForStreaming(): StreamingWrite = {
+ if (config.getValue(DorisOptions.LOAD_MODE) == "tvf") {
+ throw new UnsupportedOperationException("tvf write mode does not support
Structured Streaming")
+ }
new DorisWrite(config, schema)
}
diff --git
a/spark-doris-connector/spark-doris-connector-spark-4-base/src/main/scala/org/apache/doris/spark/write/S3TvfDataWriter.scala
b/spark-doris-connector/spark-doris-connector-spark-4-base/src/main/scala/org/apache/doris/spark/write/S3TvfDataWriter.scala
new file mode 100644
index 0000000..07a1a70
--- /dev/null
+++
b/spark-doris-connector/spark-doris-connector-spark-4-base/src/main/scala/org/apache/doris/spark/write/S3TvfDataWriter.scala
@@ -0,0 +1,74 @@
+// 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.tvf.{S3ClientObjectStore,
S3TvfCommitter, S3TvfWriter}
+import org.apache.doris.spark.config.{DorisConfig, S3TvfOptions}
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.connector.write.{DataWriter, WriterCommitMessage}
+import org.apache.spark.sql.types.StructType
+
+private[write] class S3TvfDataWriter(
+ config: DorisConfig,
+ options: S3TvfOptions,
+ database: String,
+ table: String,
+ labelPrefix: String,
+ schema: StructType,
+ columns: java.util.List[String],
+ partitionId: Int,
+ batchSize: Int) extends DataWriter[InternalRow] {
+
+ private val writer = new S3TvfWriter(
+ options,
+ database,
+ table,
+ labelPrefix,
+ schema,
+ columns,
+ partitionId,
+ batchSize,
+ new S3ClientObjectStore(options))
+ private var closed = false
+
+ override def write(record: InternalRow): Unit = writer.write(record)
+
+ override def commit(): WriterCommitMessage = {
+ val committable = writer.prepareCommit()
+ if (!committable.isEmpty) {
+ val committer = new S3TvfCommitter(config)
+ try {
+ committer.commit(committable)
+ } finally {
+ committer.close()
+ }
+ }
+ S3TvfWriterCommitMessage
+ }
+
+ override def abort(): Unit = close()
+
+ override def close(): Unit = {
+ if (!closed) {
+ writer.close()
+ closed = true
+ }
+ }
+}
+
+private[write] case object S3TvfWriterCommitMessage extends WriterCommitMessage
diff --git
a/spark-doris-connector/spark-doris-connector-spark-4-base/src/main/scala/org/apache/doris/spark/write/S3TvfDataWriterFactory.scala
b/spark-doris-connector/spark-doris-connector-spark-4-base/src/main/scala/org/apache/doris/spark/write/S3TvfDataWriterFactory.scala
new file mode 100644
index 0000000..7d2c0e5
--- /dev/null
+++
b/spark-doris-connector/spark-doris-connector-spark-4-base/src/main/scala/org/apache/doris/spark/write/S3TvfDataWriterFactory.scala
@@ -0,0 +1,50 @@
+// 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.tvf.S3TvfColumnUtils
+import org.apache.doris.spark.config.{DorisConfig, DorisOptions, S3TvfOptions}
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.connector.write.{DataWriter, DataWriterFactory}
+import org.apache.spark.sql.types.StructType
+
+private[write] class S3TvfDataWriterFactory(
+ config: DorisConfig,
+ schema: StructType) extends DataWriterFactory {
+
+ private val options = S3TvfOptions.fromConfig(config)
+ private val tableIdentifier =
config.getValue(DorisOptions.DORIS_TABLE_IDENTIFIER).split("\\.")
+ private val database = tableIdentifier(0).replaceAll("`", "").trim
+ private val table = tableIdentifier(1).replaceAll("`", "").trim
+ private val labelPrefix =
config.getValue(DorisOptions.DORIS_SINK_LABEL_PREFIX).trim
+ private val columns =
S3TvfColumnUtils.resolveColumns(config.getSinkProperties, schema)
+ private val batchSize = config.getValue(DorisOptions.DORIS_SINK_BATCH_SIZE)
+
+ override def createWriter(partitionId: Int, taskId: Long):
DataWriter[InternalRow] = {
+ new S3TvfDataWriter(
+ config,
+ options,
+ database,
+ table,
+ labelPrefix,
+ schema,
+ columns,
+ partitionId,
+ batchSize)
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
