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 c7b9dd4 [Improve] Reduce S3 TVF upload copying and log load timings
(#380)
c7b9dd4 is described below
commit c7b9dd4594d90aa1ed3b99018c987d5e3bfb9627
Author: wudi <[email protected]>
AuthorDate: Fri Sep 11 14:15:09 2026 +0800
[Improve] Reduce S3 TVF upload copying and log load timings (#380)
---
.../client/write/tvf/S3ClientObjectStore.java | 6 +-
.../spark/client/write/tvf/S3TvfCommitter.java | 18 ++++-
.../doris/spark/client/write/tvf/S3TvfWriter.java | 20 +++++-
.../client/write/tvf/S3ClientObjectStoreTest.java | 78 ++++++++++++++++++++++
4 files changed, 119 insertions(+), 3 deletions(-)
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
index 2edb360..882adf6 100644
---
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
@@ -27,6 +27,7 @@ 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.ByteArrayInputStream;
import java.io.IOException;
import java.net.URI;
@@ -70,7 +71,10 @@ public final class S3ClientObjectStore implements
S3ObjectStore {
.contentType(JSON_LINES_CONTENT_TYPE)
.build();
try {
- client.putObject(request, RequestBody.fromBytes(content));
+ client.putObject(request, RequestBody.fromContentProvider(
+ () -> new ByteArrayInputStream(content),
+ content.length,
+ JSON_LINES_CONTENT_TYPE));
} catch (RuntimeException e) {
throw new IOException("Failed to upload S3 TVF object: " +
objectKey, e);
}
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
index f709a77..540972b 100644
---
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
@@ -19,15 +19,19 @@ package org.apache.doris.spark.client.write.tvf;
import org.apache.doris.spark.config.DorisConfig;
import org.apache.doris.spark.config.S3TvfOptions;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.sql.SQLException;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.TreeMap;
+import java.util.concurrent.TimeUnit;
/** Commits one Spark partition with one Doris INSERT. */
public final class S3TvfCommitter implements AutoCloseable {
+ private static final Logger LOG =
LoggerFactory.getLogger(S3TvfCommitter.class);
private static final String COLUMNS = "columns";
private static final String PARTIAL_COLUMNS = "partial_columns";
private static final String FORMAT = "format";
@@ -59,11 +63,23 @@ public final class S3TvfCommitter implements AutoCloseable {
if (committable.isEmpty()) {
return;
}
+ String insertSql = sqlBuilder.buildInsertSql(committable);
+ long insertStartedAtNanos = System.nanoTime();
try {
loadClient.executeInsert(
- sqlBuilder.buildInsertSql(committable),
+ insertSql,
sessionVariables);
+ LOG.info("TVF insert completed, label={}, objectCount={},
insertTimeMs={}.",
+ committable.getLabel(),
+ committable.getObjectKeys().size(),
+ TimeUnit.NANOSECONDS.toMillis(System.nanoTime() -
insertStartedAtNanos));
} catch (SQLException e) {
+ LOG.warn("TVF insert failed, label={}, objectCount={},
insertTimeMs={}, SQLState={}, errorCode={}.",
+ committable.getLabel(),
+ committable.getObjectKeys().size(),
+ TimeUnit.NANOSECONDS.toMillis(System.nanoTime() -
insertStartedAtNanos),
+ e.getSQLState(),
+ e.getErrorCode());
throw new IOException(
"Doris INSERT failed for S3 TVF label " +
committable.getLabel(), e);
}
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
index 9ef17dd..135f08e 100644
---
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
@@ -23,15 +23,19 @@ 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 org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;
+import java.util.concurrent.TimeUnit;
/** Writes one logical Spark partition as deterministic JSON Lines objects. */
public final class S3TvfWriter implements AutoCloseable {
+ private static final Logger LOG =
LoggerFactory.getLogger(S3TvfWriter.class);
private static final byte NEW_LINE = '\n';
private final S3TvfOptions options;
@@ -110,7 +114,21 @@ public final class S3TvfWriter implements AutoCloseable {
currentFileNumber);
String prefix = options.getPrefix();
String objectKey = prefix + (prefix.endsWith("/") ? "" : "/") +
fileName;
- objectStore.put(objectKey, content);
+ long uploadStartedAtNanos = System.nanoTime();
+ try {
+ objectStore.put(objectKey, content);
+ LOG.info("S3 TVF object upload completed, objectKey={},
sizeBytes={}, uploadTimeMs={}.",
+ objectKey,
+ content.length,
+ TimeUnit.NANOSECONDS.toMillis(System.nanoTime() -
uploadStartedAtNanos));
+ } catch (IOException | RuntimeException e) {
+ LOG.warn("S3 TVF object upload failed, objectKey={}, sizeBytes={},
uploadTimeMs={}.",
+ objectKey,
+ content.length,
+ TimeUnit.NANOSECONDS.toMillis(System.nanoTime() -
uploadStartedAtNanos),
+ e);
+ throw e;
+ }
objectKeys.add(objectKey);
buffer.reset();
recordCount = 0;
diff --git
a/spark-doris-connector/spark-doris-connector-base/src/test/java/org/apache/doris/spark/client/write/tvf/S3ClientObjectStoreTest.java
b/spark-doris-connector/spark-doris-connector-base/src/test/java/org/apache/doris/spark/client/write/tvf/S3ClientObjectStoreTest.java
new file mode 100644
index 0000000..ae84933
--- /dev/null
+++
b/spark-doris-connector/spark-doris-connector-base/src/test/java/org/apache/doris/spark/client/write/tvf/S3ClientObjectStoreTest.java
@@ -0,0 +1,78 @@
+// 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.junit.Assert;
+import org.junit.Test;
+import software.amazon.awssdk.core.sync.RequestBody;
+import software.amazon.awssdk.services.s3.S3Client;
+import software.amazon.awssdk.services.s3.model.PutObjectRequest;
+import software.amazon.awssdk.services.s3.model.PutObjectResponse;
+
+import java.io.DataInputStream;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+
+public class S3ClientObjectStoreTest {
+ @Test
+ public void uploadBodyIsRepeatableWithoutCopying() throws Exception {
+ byte[] content = "{\"id\":1}\n".getBytes(StandardCharsets.UTF_8);
+ RequestBody[] captured = new RequestBody[1];
+ S3Client client = new S3Client() {
+ @Override
+ public PutObjectResponse putObject(PutObjectRequest request,
RequestBody body) {
+ Assert.assertEquals("bucket", request.bucket());
+ Assert.assertEquals("prefix/file.json", request.key());
+ Assert.assertEquals("application/x-ndjson",
request.contentType());
+ captured[0] = body;
+ return PutObjectResponse.builder().build();
+ }
+
+ @Override
+ public String serviceName() {
+ return "s3";
+ }
+
+ @Override
+ public void close() {}
+ };
+ try (S3ClientObjectStore store = new S3ClientObjectStore(client,
"bucket")) {
+ store.put("prefix/file.json", content);
+ RequestBody body = captured[0];
+ Assert.assertEquals(content.length,
body.optionalContentLength().get().longValue());
+ // Mutate only in this test to detect a defensive copy in the
request body.
+ content[0] = '[';
+ try (DataInputStream first = new
DataInputStream(body.contentStreamProvider().newStream());
+ DataInputStream retry = new
DataInputStream(body.contentStreamProvider().newStream())) {
+ Assert.assertEquals('[', first.read());
+ assertContent(retry, content);
+ byte[] remaining = new byte[content.length - 1];
+ System.arraycopy(content, 1, remaining, 0, remaining.length);
+ assertContent(first, remaining);
+ }
+ Assert.assertEquals("application/x-ndjson", body.contentType());
+ }
+ }
+
+ private static void assertContent(DataInputStream input, byte[] expected)
throws IOException {
+ byte[] actual = new byte[expected.length];
+ input.readFully(actual);
+ Assert.assertArrayEquals(expected, actual);
+ Assert.assertEquals(-1, input.read());
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
