You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by yi...@apache.org on 2026/04/20 04:35:01 UTC
(doris) branch adaptive_batch_size updated: [opt](maxcompute)Allocate write block ids from FE and add catalog-level write_max_block_bytes prop. (#62615)
This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch adaptive_batch_size
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/adaptive_batch_size by this push:
new e76a88673ff [opt](maxcompute)Allocate write block ids from FE and add catalog-level write_max_block_bytes prop. (#62615)
e76a88673ff is described below
commit e76a88673ff3a5501ce7a9d095ab921b5fb359c6
Author: daidai <ch...@selectdb.com>
AuthorDate: Mon Apr 20 12:34:55 2026 +0800
[opt](maxcompute)Allocate write block ids from FE and add catalog-level write_max_block_bytes prop. (#62615)
### What problem does this PR solve?
Issue Number: close #xxx
Related PR: #xxx
Problem Summary:
### Release note
None
### Check List (For Author)
- Test <!-- At least one of them must be included. -->
- [ ] Regression test
- [ ] Unit Test
- [ ] Manual test (add detailed scripts or steps below)
- [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
- [ ] Previous test can cover this change.
- [ ] No code files have been changed.
- [ ] Other reason <!-- Add your reason? -->
- Behavior changed:
- [ ] No.
- [ ] Yes. <!-- Explain the behavior change -->
- Does this need documentation?
- [ ] No.
- [ ] Yes. <!-- Add document PR link here. eg:
https://github.com/apache/doris-website/pull/1214 -->
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label <!-- Add branch pick label that this PR
should merge into -->
---
.../sink/writer/maxcompute/vmc_table_writer.cpp | 54 +--
.../exec/sink/writer/maxcompute/vmc_table_writer.h | 13 -
be/src/util/jni-util.cpp | 4 +
be/src/util/jni_native_method.cpp | 108 +++++
be/src/util/jni_native_method.h | 4 +
.../doris/common/jni/utils/JNINativeMethod.java | 5 +
.../doris/maxcompute/MaxComputeJniScanner.java | 162 +------
.../doris/maxcompute/MaxComputeJniWriter.java | 472 +++++++++------------
.../doris/common/maxcompute/MCProperties.java | 7 +-
.../doris/datasource/maxcompute/MCTransaction.java | 69 ++-
.../maxcompute/MaxComputeExternalCatalog.java | 8 -
.../plans/commands/insert/MCInsertExecutor.java | 4 +-
.../apache/doris/planner/MaxComputeTableSink.java | 6 +-
.../apache/doris/service/FrontendServiceImpl.java | 48 +++
.../doris/service/FrontendServiceImplTest.java | 49 +++
gensrc/thrift/DataSinks.thrift | 3 +-
gensrc/thrift/FrontendService.thrift | 14 +
.../write/test_mc_write_catalog_block_bytes.groovy | 167 ++++++++
.../write/test_mc_write_large_data.groovy | 49 ---
19 files changed, 704 insertions(+), 542 deletions(-)
diff --git a/be/src/exec/sink/writer/maxcompute/vmc_table_writer.cpp b/be/src/exec/sink/writer/maxcompute/vmc_table_writer.cpp
index 3abb836d87d..a53bf5edada 100644
--- a/be/src/exec/sink/writer/maxcompute/vmc_table_writer.cpp
+++ b/be/src/exec/sink/writer/maxcompute/vmc_table_writer.cpp
@@ -42,13 +42,11 @@ Status VMCTableWriter::init_properties(ObjectPool* pool) {
Status VMCTableWriter::open(RuntimeState* state, RuntimeProfile* profile) {
_state = state;
- _next_block_id.store(state->per_fragment_instance_idx() * BLOCK_ID_STRIDE);
LOG(INFO) << "VMCTableWriter::open"
<< ", fragment_instance_id=" << print_id(state->fragment_instance_id())
<< ", per_fragment_instance_idx=" << state->per_fragment_instance_idx()
- << ", write_session_id=" << _mc_sink.write_session_id
- << ", next_block_id_start=" << _next_block_id.load();
+ << ", write_session_id=" << _mc_sink.write_session_id;
_written_rows_counter = ADD_COUNTER(_operator_profile, "WrittenRows", TUnit::UNIT);
_send_data_timer = ADD_TIMER(_operator_profile, "SendDataTime");
@@ -105,6 +103,9 @@ std::map<std::string, std::string> VMCTableWriter::_build_base_writer_params() {
if (_mc_sink.__isset.project) params["project"] = _mc_sink.project;
if (_mc_sink.__isset.table_name) params["table"] = _mc_sink.table_name;
if (_mc_sink.__isset.quota) params["quota"] = _mc_sink.quota;
+ if (_mc_sink.__isset.txn_id) {
+ params["txn_id"] = std::to_string(_mc_sink.txn_id);
+ }
if (_mc_sink.__isset.write_session_id) {
params["write_session_id"] = _mc_sink.write_session_id;
}
@@ -117,9 +118,6 @@ std::map<std::string, std::string> VMCTableWriter::_build_base_writer_params() {
if (_mc_sink.__isset.retry_count) {
params["retry_count"] = std::to_string(_mc_sink.retry_count);
}
- if (_mc_sink.__isset.max_write_batch_rows) {
- params["max_write_batch_rows"] = std::to_string(_mc_sink.max_write_batch_rows);
- }
return params;
}
@@ -127,11 +125,9 @@ std::shared_ptr<VMCPartitionWriter> VMCTableWriter::_create_partition_writer(
const std::string& partition_spec) {
auto params = _build_base_writer_params();
params["partition_spec"] = partition_spec;
- // Each partition writer gets a unique block_id from the atomic counter
- params["block_id"] = std::to_string(_next_block_id.fetch_add(1));
LOG(INFO) << "VMCTableWriter::_create_partition_writer"
<< ", fragment_instance_id=" << print_id(_state->fragment_instance_id())
- << ", partition_spec=" << partition_spec << ", block_id=" << params["block_id"];
+ << ", partition_spec=" << partition_spec;
return std::make_shared<VMCPartitionWriter>(_state, _write_output_vexpr_ctxs, partition_spec,
std::move(params));
}
@@ -159,10 +155,13 @@ Status VMCTableWriter::write(RuntimeState* state, Block& block) {
it = _partitions_to_writers.find(_static_partition_spec);
}
output_block.erase(_non_write_columns_indices);
- return _write_block_in_chunks(it->second, output_block);
+ return it->second->write(output_block);
}
// Case 2: Dynamic partition or non-partitioned table
+ // For dynamic partitions, MaxCompute Storage API (with DynamicPartitionOptions) expects
+ // partition column values in the Arrow data and handles routing internally.
+ // So we send the full block including partition columns to a single writer.
std::string partition_key = "";
auto it = _partitions_to_writers.find(partition_key);
if (it == _partitions_to_writers.end()) {
@@ -171,40 +170,7 @@ Status VMCTableWriter::write(RuntimeState* state, Block& block) {
_partitions_to_writers.insert({partition_key, writer});
it = _partitions_to_writers.find(partition_key);
}
- return _write_block_in_chunks(it->second, output_block);
-}
-
-Status VMCTableWriter::_write_block_in_chunks(const std::shared_ptr<VMCPartitionWriter>& writer,
- Block& output_block) {
- // Limit per-JNI data to MAX_WRITE_BLOCK_BYTES. When data source is not MC scanner
- // (e.g. Doris internal table, Hive, JDBC), the upstream batch_size controls Block
- // row count but not byte size. With large rows (585KB/row), a 4096-row Block is
- // ~2.4GB. Splitting ensures each JNI call processes bounded data, limiting Arrow
- // and SDK native memory per call.
- static constexpr size_t MAX_WRITE_BLOCK_BYTES = 256 * 1024 * 1024; // 256MB
-
- const size_t block_bytes = output_block.allocated_bytes();
- const size_t rows = output_block.rows();
-
- if (block_bytes <= MAX_WRITE_BLOCK_BYTES || rows <= 1) {
- return writer->write(output_block);
- }
-
- const size_t bytes_per_row = block_bytes / rows;
- const size_t max_rows = std::max(size_t(1), MAX_WRITE_BLOCK_BYTES / bytes_per_row);
-
- for (size_t offset = 0; offset < rows; offset += max_rows) {
- const size_t num_rows = std::min(max_rows, rows - offset);
- Block sub_block = output_block.clone_empty();
- auto columns = sub_block.mutate_columns();
- for (size_t i = 0; i < columns.size(); i++) {
- columns[i]->insert_range_from(*output_block.get_by_position(i).column, offset,
- num_rows);
- }
- sub_block.set_columns(std::move(columns));
- RETURN_IF_ERROR(writer->write(sub_block));
- }
- return Status::OK();
+ return it->second->write(output_block);
}
Status VMCTableWriter::close(Status status) {
diff --git a/be/src/exec/sink/writer/maxcompute/vmc_table_writer.h b/be/src/exec/sink/writer/maxcompute/vmc_table_writer.h
index 643b28f1cc0..720da155381 100644
--- a/be/src/exec/sink/writer/maxcompute/vmc_table_writer.h
+++ b/be/src/exec/sink/writer/maxcompute/vmc_table_writer.h
@@ -19,7 +19,6 @@
#include <gen_cpp/DataSinks_types.h>
-#include <atomic>
#include <set>
#include <unordered_map>
@@ -54,12 +53,6 @@ public:
private:
std::shared_ptr<VMCPartitionWriter> _create_partition_writer(const std::string& partition_spec);
- // Split large blocks into sub-blocks before JNI to limit Arrow and SDK
- // native memory. Needed when data source is not MC scanner and blocks
- // may exceed 256MB (e.g. batch_size=4096 with 585KB/row = 2.4GB).
- Status _write_block_in_chunks(const std::shared_ptr<VMCPartitionWriter>& writer,
- Block& output_block);
-
std::map<std::string, std::string> _build_base_writer_params();
TDataSink _t_sink;
@@ -82,12 +75,6 @@ private:
// Write output expr contexts (after removing partition columns)
VExprContextSPtrs _write_output_vexpr_ctxs;
- // Atomic block_id counter: each partition writer gets a unique block_id
- // Initialized with offset based on per_fragment_instance_idx to avoid collisions
- // across pipeline instances sharing the same write session.
- std::atomic<int64_t> _next_block_id {0};
- static constexpr int64_t BLOCK_ID_STRIDE = 100;
-
size_t _row_count = 0;
int64_t _send_data_ns = 0;
int64_t _close_ns = 0;
diff --git a/be/src/util/jni-util.cpp b/be/src/util/jni-util.cpp
index 9f5f3999c17..6aad1010903 100644
--- a/be/src/util/jni-util.cpp
+++ b/be/src/util/jni-util.cpp
@@ -406,6 +406,8 @@ Status Util::_init_register_natives() {
static char memory_alloc_batch_sign[] = "([I)[J";
static char memory_free_batch_name[] = "memoryTrackerFreeBatch";
static char memory_free_batch_sign[] = "([J)V";
+ static char request_mc_block_id_name[] = "requestMaxComputeBlockId";
+ static char request_mc_block_id_sign[] = "(JLjava/lang/String;)J";
static JNINativeMethod java_native_methods[] = {
{memory_alloc_name, memory_alloc_sign, (void*)&JavaNativeMethods::memoryMalloc},
{memory_free_name, memory_free_sign, (void*)&JavaNativeMethods::memoryFree},
@@ -413,6 +415,8 @@ Status Util::_init_register_natives() {
(void*)&JavaNativeMethods::memoryMallocBatch},
{memory_free_batch_name, memory_free_batch_sign,
(void*)&JavaNativeMethods::memoryFreeBatch},
+ {request_mc_block_id_name, request_mc_block_id_sign,
+ (void*)&JavaNativeMethods::requestMaxComputeBlockId},
};
int res = env->RegisterNatives(local_jni_native_exc_cl, java_native_methods,
diff --git a/be/src/util/jni_native_method.cpp b/be/src/util/jni_native_method.cpp
index 77099e269dd..47e105a1300 100644
--- a/be/src/util/jni_native_method.cpp
+++ b/be/src/util/jni_native_method.cpp
@@ -17,16 +17,100 @@
#include "util/jni_native_method.h"
+#include <gen_cpp/FrontendService.h>
#include <glog/logging.h>
+#include <chrono>
#include <cstdlib>
+#include <thread>
#include <vector>
+#include "common/status.h"
#include "jni.h"
+#include "runtime/exec_env.h"
+#include "util/client_cache.h"
#include "util/defer_op.h"
+#include "util/thrift_rpc_helper.h"
namespace doris {
+namespace {
+
+void throw_java_runtime_exception(JNIEnv* env, const std::string& message) {
+ jclass exception_cl = env->FindClass("java/lang/IllegalStateException");
+ if (exception_cl != nullptr) {
+ env->ThrowNew(exception_cl, message.c_str());
+ env->DeleteLocalRef(exception_cl);
+ }
+}
+
+Result<int64_t> request_maxcompute_block_id_from_fe(int64_t txn_id,
+ const std::string& write_session_id) {
+ if (txn_id <= 0) {
+ return ResultError(Status::InvalidArgument(
+ "invalid MaxCompute txn_id for block_id allocation: {}", txn_id));
+ }
+ if (write_session_id.empty()) {
+ return ResultError(Status::InvalidArgument(
+ "empty MaxCompute write_session_id for block_id allocation"));
+ }
+
+ constexpr uint32_t FETCH_BLOCK_ID_MAX_RETRY_TIMES = 3;
+ TNetworkAddress master_addr = ExecEnv::GetInstance()->cluster_info()->master_fe_addr;
+ for (uint32_t retry_times = 0; retry_times < FETCH_BLOCK_ID_MAX_RETRY_TIMES; retry_times++) {
+ TMaxComputeBlockIdRequest request;
+ TMaxComputeBlockIdResult result;
+ request.__set_txn_id(txn_id);
+ request.__set_write_session_id(write_session_id);
+ request.__set_length(1);
+
+ Status rpc_status = ThriftRpcHelper::rpc<FrontendServiceClient>(
+ master_addr.hostname, master_addr.port,
+ [&request, &result](FrontendServiceConnection& client) {
+ client->getMaxComputeBlockIdRange(result, request);
+ });
+
+ if (rpc_status.is<ErrorCode::NOT_MASTER>()) {
+ LOG(WARNING) << "Failed to allocate MaxCompute block_id, requested non-master FE@"
+ << master_addr.hostname << ":" << master_addr.port << ", switch to FE@"
+ << result.master_address.hostname << ":" << result.master_address.port
+ << ", retry_time=" << retry_times << ", txn_id=" << txn_id
+ << ", write_session_id=" << write_session_id;
+ master_addr = result.master_address;
+ std::this_thread::sleep_for(std::chrono::milliseconds(10));
+ continue;
+ }
+
+ if (!rpc_status.ok()) {
+ LOG(WARNING) << "Failed to allocate MaxCompute block_id, rpc failure, retry_time="
+ << retry_times << ", txn_id=" << txn_id
+ << ", write_session_id=" << write_session_id << ", status=" << rpc_status;
+ std::this_thread::sleep_for(std::chrono::milliseconds(10));
+ continue;
+ }
+
+ if (result.length != 1) {
+ auto msg = "Failed to allocate MaxCompute block_id, expected length=1 but got " +
+ std::to_string(result.length) + " for txn_id=" + std::to_string(txn_id) +
+ ", write_session_id=" + write_session_id;
+ LOG(WARNING) << msg;
+ std::this_thread::sleep_for(std::chrono::milliseconds(10));
+ continue;
+ }
+
+ LOG(INFO) << "Allocated MaxCompute block_id from FE@" << master_addr.hostname << ":"
+ << master_addr.port << ", txn_id=" << txn_id
+ << ", write_session_id=" << write_session_id << ", block_id=" << result.start;
+ return result.start;
+ }
+
+ return ResultError(Status::RpcError(
+ "failed to allocate MaxCompute block_id from FE, txn_id={}, write_session_id={}",
+ txn_id, write_session_id));
+}
+
+} // namespace
+
jlong JavaNativeMethods::memoryMalloc(JNIEnv* env, jclass clazz, jlong bytes) {
return reinterpret_cast<long>(malloc(bytes));
}
@@ -106,4 +190,28 @@ void JavaNativeMethods::memoryFreeBatch(JNIEnv* env, jclass clazz, jlongArray ad
env->ReleaseLongArrayElements(addrs, elems, JNI_ABORT);
}
+jlong JavaNativeMethods::requestMaxComputeBlockId(JNIEnv* env, jclass clazz, jlong txn_id,
+ jstring write_session_id) {
+ if (write_session_id == nullptr) {
+ throw_java_runtime_exception(
+ env, "MaxCompute write_session_id is null when requesting block_id");
+ return 0;
+ }
+
+ const char* write_session_id_chars = env->GetStringUTFChars(write_session_id, nullptr);
+ if (write_session_id_chars == nullptr) {
+ throw_java_runtime_exception(env, "Failed to read MaxCompute write_session_id from Java");
+ return 0;
+ }
+ std::string write_session_id_str(write_session_id_chars);
+ env->ReleaseStringUTFChars(write_session_id, write_session_id_chars);
+
+ auto block_id = request_maxcompute_block_id_from_fe(txn_id, write_session_id_str);
+ if (!block_id.has_value()) {
+ throw_java_runtime_exception(env, block_id.error().to_string());
+ return 0;
+ }
+ return static_cast<jlong>(block_id.value());
+}
+
} // namespace doris
diff --git a/be/src/util/jni_native_method.h b/be/src/util/jni_native_method.h
index 48c74d91d67..23429c3500a 100644
--- a/be/src/util/jni_native_method.h
+++ b/be/src/util/jni_native_method.h
@@ -42,6 +42,10 @@ struct JavaNativeMethods {
// Batch free multiple addresses; addrs is a long[]
static void memoryFreeBatch(JNIEnv* env, jclass clazz, jlongArray addrs);
+
+ // Request a MaxCompute block id from FE via BE.
+ static jlong requestMaxComputeBlockId(JNIEnv* env, jclass clazz, jlong txn_id,
+ jstring write_session_id);
};
} // namespace doris
diff --git a/fe/be-java-extensions/java-common/src/main/java/org/apache/doris/common/jni/utils/JNINativeMethod.java b/fe/be-java-extensions/java-common/src/main/java/org/apache/doris/common/jni/utils/JNINativeMethod.java
index 9ba171e5c53..1104a5fa934 100644
--- a/fe/be-java-extensions/java-common/src/main/java/org/apache/doris/common/jni/utils/JNINativeMethod.java
+++ b/fe/be-java-extensions/java-common/src/main/java/org/apache/doris/common/jni/utils/JNINativeMethod.java
@@ -42,4 +42,9 @@ public class JNINativeMethod {
* corresponds to one deallocation.
*/
public static native void memoryTrackerFreeBatch(long[] addrs);
+
+ /**
+ * Request a MaxCompute block id from BE, which will forward the request to FE.
+ */
+ public static native long requestMaxComputeBlockId(long txnId, String writeSessionId);
}
diff --git a/fe/be-java-extensions/max-compute-connector/src/main/java/org/apache/doris/maxcompute/MaxComputeJniScanner.java b/fe/be-java-extensions/max-compute-connector/src/main/java/org/apache/doris/maxcompute/MaxComputeJniScanner.java
index 336991f3802..ce1b3941dca 100644
--- a/fe/be-java-extensions/max-compute-connector/src/main/java/org/apache/doris/maxcompute/MaxComputeJniScanner.java
+++ b/fe/be-java-extensions/max-compute-connector/src/main/java/org/apache/doris/maxcompute/MaxComputeJniScanner.java
@@ -33,7 +33,6 @@ import com.aliyun.odps.table.read.split.InputSplit;
import com.aliyun.odps.table.read.split.impl.IndexedInputSplit;
import com.aliyun.odps.table.read.split.impl.RowRangeInputSplit;
import com.google.common.base.Strings;
-import org.apache.arrow.vector.BaseVariableWidthVector;
import org.apache.arrow.vector.FieldVector;
import org.apache.arrow.vector.VectorSchemaRoot;
import org.apache.log4j.Logger;
@@ -61,13 +60,6 @@ public class MaxComputeJniScanner extends JniScanner {
private static final Logger LOG = Logger.getLogger(MaxComputeJniScanner.class);
- // 256MB byte budget per scanner batch — limits the C++ Block size at the source.
- // With large rows (e.g. 585KB/row STRING), batch_size=4096 would create ~2.4GB Blocks.
- // The pipeline's AsyncResultWriter queues up to 3 Blocks per instance, and with
- // parallel_pipeline_task_num instances, total queue memory = instances * 3 * block_size.
- // 256MB keeps queue memory manageable: 5 instances * 3 * 256MB = 3.8GB.
- private static final long MAX_BATCH_BYTES = 256 * 1024 * 1024L;
-
private static final String ACCESS_KEY = "access_key";
private static final String SECRET_KEY = "secret_key";
private static final String ENDPOINT = "endpoint";
@@ -98,8 +90,6 @@ public class MaxComputeJniScanner extends JniScanner {
private String table;
private SplitReader<VectorSchemaRoot> currentSplitReader;
- private VectorSchemaRoot currentBatch = null;
- private int currentBatchRowOffset = 0;
private MaxComputeColumnValue columnValue;
private Map<String, Integer> readColumnsToId;
@@ -225,17 +215,8 @@ public class MaxComputeJniScanner extends JniScanner {
@Override
public void close() throws IOException {
- if (currentSplitReader != null) {
- try {
- currentSplitReader.close();
- } catch (Exception e) {
- LOG.warn("Failed to close MaxCompute split reader for table " + project + "." + table, e);
- }
- }
startOffset = -1;
splitSize = -1;
- currentBatch = null;
- currentBatchRowOffset = 0;
currentSplitReader = null;
settings = null;
scan = null;
@@ -253,74 +234,45 @@ public class MaxComputeJniScanner extends JniScanner {
return readVectors(expectedRows);
}
- private VectorSchemaRoot getNextBatch() throws IOException {
- try {
- if (!currentSplitReader.hasNext()) {
- currentSplitReader.close();
- currentSplitReader = null;
- return null;
- }
- return currentSplitReader.get();
- } catch (Exception e) {
- String errorMsg = "MaxComputeJniScanner readVectors get batch fail";
- LOG.warn(errorMsg, e);
- throw new IOException(e.getMessage(), e);
- }
- }
-
private int readVectors(int expectedRows) throws IOException {
int curReadRows = 0;
- long accumulatedBytes = 0;
while (curReadRows < expectedRows) {
- // Stop early if accumulated variable-width bytes approach int32 limit
- if (accumulatedBytes >= MAX_BATCH_BYTES) {
- break;
- }
- if (currentBatch == null) {
- currentBatch = getNextBatch();
- if (currentBatch == null || currentBatch.getRowCount() == 0) {
- currentBatch = null;
+ try {
+ if (!currentSplitReader.hasNext()) {
+ currentSplitReader.close();
+ currentSplitReader = null;
break;
}
- currentBatchRowOffset = 0;
+ } catch (Exception e) {
+ String errorMsg = "MaxComputeJniScanner readVectors hasNext fail";
+ LOG.warn(errorMsg, e);
+ throw new IOException(e.getMessage(), e);
}
+
try {
- int rowsToAppend = Math.min(expectedRows - curReadRows,
- currentBatch.getRowCount() - currentBatchRowOffset);
- List<FieldVector> fieldVectors = currentBatch.getFieldVectors();
-
- // Limit rows to avoid int32 overflow in VectorColumn's String byte buffer
- rowsToAppend = limitRowsByVarWidthBytes(
- fieldVectors, currentBatchRowOffset, rowsToAppend,
- MAX_BATCH_BYTES - accumulatedBytes);
- if (rowsToAppend <= 0) {
+ VectorSchemaRoot data = currentSplitReader.get();
+ if (data.getRowCount() == 0) {
break;
}
+ List<FieldVector> fieldVectors = data.getFieldVectors();
+ int batchRows = 0;
long startTime = System.nanoTime();
for (FieldVector column : fieldVectors) {
Integer readColumnId = readColumnsToId.get(column.getName());
+ batchRows = column.getValueCount();
if (readColumnId == null) {
continue;
}
columnValue.reset(column);
- for (int j = currentBatchRowOffset; j < currentBatchRowOffset + rowsToAppend; j++) {
+ for (int j = 0; j < batchRows; j++) {
columnValue.setColumnIdx(j);
appendData(readColumnId, columnValue);
}
}
appendDataTime += System.nanoTime() - startTime;
- // Track bytes for the rows just appended
- accumulatedBytes += estimateVarWidthBytes(
- fieldVectors, currentBatchRowOffset, rowsToAppend);
-
- currentBatchRowOffset += rowsToAppend;
- curReadRows += rowsToAppend;
- if (currentBatchRowOffset >= currentBatch.getRowCount()) {
- currentBatch = null;
- currentBatchRowOffset = 0;
- }
+ curReadRows += batchRows;
} catch (Exception e) {
String errorMsg = String.format("MaxComputeJniScanner Fail to read arrow data. "
+ "curReadRows = {}, expectedRows = {}", curReadRows, expectedRows);
@@ -328,91 +280,9 @@ public class MaxComputeJniScanner extends JniScanner {
throw new RuntimeException(errorMsg, e);
}
}
- if (LOG.isDebugEnabled() && curReadRows > 0 && curReadRows < expectedRows) {
- LOG.debug("readVectors: returning " + curReadRows + " rows (limited by byte budget)"
- + ", totalVarWidthBytes=" + accumulatedBytes
- + ", expectedRows=" + expectedRows);
- }
return curReadRows;
}
- /**
- * Limit the number of rows to append so that no single variable-width column
- * exceeds the remaining byte budget. This prevents int32 overflow in
- * VectorColumn's appendIndex for String/Binary child byte arrays.
- *
- * Uses Arrow's offset buffer for O(1)-per-row byte size calculation —
- * no data copying involved.
- */
- private int limitRowsByVarWidthBytes(List<FieldVector> fieldVectors,
- int offset, int maxRows, long remainingBudget) {
- if (remainingBudget <= 0) {
- return 0;
- }
- int safeRows = maxRows;
- for (FieldVector fv : fieldVectors) {
- if (fv instanceof BaseVariableWidthVector) {
- BaseVariableWidthVector vec = (BaseVariableWidthVector) fv;
- // Find how many rows fit within the budget for THIS column
- int rows = findMaxRowsWithinBudget(vec, offset, maxRows, remainingBudget);
- safeRows = Math.min(safeRows, rows);
- }
- }
- // Always allow at least 1 row to make progress, even if it exceeds budget
- return Math.max(1, safeRows);
- }
-
- /**
- * Binary search for the maximum number of rows starting at 'offset'
- * whose total bytes in the variable-width vector fit within 'budget'.
- */
- private int findMaxRowsWithinBudget(BaseVariableWidthVector vec,
- int offset, int maxRows, long budget) {
- if (maxRows <= 0) {
- return 0;
- }
- // Total bytes for all maxRows
- long totalBytes = (long) vec.getOffsetBuffer().getInt((long) (offset + maxRows) * 4)
- - (long) vec.getOffsetBuffer().getInt((long) offset * 4);
- if (totalBytes <= budget) {
- return maxRows;
- }
- // Binary search for the cutoff point
- int lo = 1;
- int hi = maxRows - 1;
- int startOff = vec.getOffsetBuffer().getInt((long) offset * 4);
- while (lo <= hi) {
- int mid = lo + (hi - lo) / 2;
- long bytes = (long) vec.getOffsetBuffer().getInt((long) (offset + mid) * 4) - startOff;
- if (bytes <= budget) {
- lo = mid + 1;
- } else {
- hi = mid - 1;
- }
- }
- // 'hi' is the largest count whose bytes <= budget (could be 0)
- return hi;
- }
-
- /**
- * Estimate total variable-width bytes for the given row range across all columns.
- * Returns the max bytes of any single column (since each column has its own
- * VectorColumn child buffer and the overflow is per-column).
- */
- private long estimateVarWidthBytes(List<FieldVector> fieldVectors,
- int offset, int rows) {
- long maxColumnBytes = 0;
- for (FieldVector fv : fieldVectors) {
- if (fv instanceof BaseVariableWidthVector) {
- BaseVariableWidthVector vec = (BaseVariableWidthVector) fv;
- long bytes = (long) vec.getOffsetBuffer().getInt((long) (offset + rows) * 4)
- - (long) vec.getOffsetBuffer().getInt((long) offset * 4);
- maxColumnBytes = Math.max(maxColumnBytes, bytes);
- }
- }
- return maxColumnBytes;
- }
-
private static Object deserialize(String serializedString) throws IOException, ClassNotFoundException {
byte[] serializedBytes = Base64.getDecoder().decode(serializedString);
ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(serializedBytes);
diff --git a/fe/be-java-extensions/max-compute-connector/src/main/java/org/apache/doris/maxcompute/MaxComputeJniWriter.java b/fe/be-java-extensions/max-compute-connector/src/main/java/org/apache/doris/maxcompute/MaxComputeJniWriter.java
index 4e0f2570a82..ba3c4328fe9 100644
--- a/fe/be-java-extensions/max-compute-connector/src/main/java/org/apache/doris/maxcompute/MaxComputeJniWriter.java
+++ b/fe/be-java-extensions/max-compute-connector/src/main/java/org/apache/doris/maxcompute/MaxComputeJniWriter.java
@@ -18,12 +18,16 @@
package org.apache.doris.maxcompute;
import org.apache.doris.common.jni.JniWriter;
+import org.apache.doris.common.jni.utils.JNINativeMethod;
import org.apache.doris.common.jni.vec.VectorColumn;
import org.apache.doris.common.jni.vec.VectorTable;
+import org.apache.doris.common.maxcompute.MCProperties;
import org.apache.doris.common.maxcompute.MCUtils;
import com.aliyun.odps.Odps;
import com.aliyun.odps.OdpsType;
+import com.aliyun.odps.table.arrow.ArrowWriter;
+import com.aliyun.odps.table.arrow.ArrowWriterFactory;
import com.aliyun.odps.table.configuration.ArrowOptions;
import com.aliyun.odps.table.configuration.ArrowOptions.TimestampUnit;
import com.aliyun.odps.table.configuration.CompressionCodec;
@@ -64,11 +68,13 @@ import org.apache.log4j.Logger;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.ObjectOutputStream;
+import java.io.OutputStream;
import java.math.BigDecimal;
import java.nio.charset.StandardCharsets;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.ZoneId;
+import java.util.ArrayList;
import java.util.Base64;
import java.util.HashMap;
import java.util.List;
@@ -88,56 +94,38 @@ public class MaxComputeJniWriter extends JniWriter {
private static final String QUOTA = "quota";
private static final String PROJECT = "project";
private static final String TABLE = "table";
+ private static final String TXN_ID = "txn_id";
private static final String WRITE_SESSION_ID = "write_session_id";
private static final String BLOCK_ID = "block_id";
private static final String PARTITION_SPEC = "partition_spec";
private static final String CONNECT_TIMEOUT = "connect_timeout";
private static final String READ_TIMEOUT = "read_timeout";
private static final String RETRY_COUNT = "retry_count";
- private static final String MAX_WRITE_BATCH_ROWS = "max_write_batch_rows";
private final Map<String, String> params;
-
- // 128MB batch threshold — controls peak Arrow native memory per batch.
- // Arrow uses sun.misc.Unsafe.allocateMemory() which is invisible to JVM metrics.
- // Each batch temporarily holds ~batchDataBytes of native memory.
- // With 3 concurrent writers, total Arrow native = 3 * 128MB = ~384MB.
- // Using 1GB was too large: 3 writers * 1GB = 3GB invisible native memory.
- private static final long MAX_ARROW_BATCH_BYTES = 128 * 1024 * 1024L;
-
- // Segmented commit: commit and recreate batchWriter every N rows to prevent
- // MaxCompute SDK native memory accumulation. Without this, the SDK buffers
- // all written data internally, causing process RSS to grow linearly with
- // total data volume until SIGSEGV.
- private static final long ROWS_PER_SEGMENT = 5000;
-
private final String endpoint;
private final String project;
private final String tableName;
private final String quota;
- private String writeSessionId;
- private long blockId;
- private long nextBlockId; // For creating new segments with unique blockIds
- private String partitionSpec;
- private int connectTimeout;
- private int readTimeout;
- private int retryCount;
- private int maxWriteBatchRows;
+ private final long txnId;
+ private final String writeSessionId;
+ private final Long preallocatedBlockId;
+ private final String partitionSpec;
+ private final int connectTimeout;
+ private final int readTimeout;
+ private final int retryCount;
+ private final long maxBlockBytes;
// Storage API objects
private TableBatchWriteSession writeSession;
private BatchWriter<VectorSchemaRoot> batchWriter;
private BufferAllocator allocator;
+ private WriterOptions writerOptions;
private List<TypeInfo> columnTypeInfos;
private List<String> columnNames;
- // Collect commit messages from all segments (each batchWriter commit produces one)
- private final List<WriterCommitMessage> commitMessages = new java.util.ArrayList<>();
-
- // Per-segment row counter (resets after each segment commit)
- private long segmentRows = 0;
-
- // Writer options cached for creating new batchWriters
- private WriterOptions writerOptions;
+ private long currentBlockId = -1L;
+ private long currentBlockWrittenBytes = 0L;
+ private final List<WriterCommitMessage> commitMessages = new ArrayList<>();
// Statistics
private long writtenRows = 0;
@@ -150,15 +138,19 @@ public class MaxComputeJniWriter extends JniWriter {
this.project = Objects.requireNonNull(params.get(PROJECT), "required property '" + PROJECT + "'.");
this.tableName = Objects.requireNonNull(params.get(TABLE), "required property '" + TABLE + "'.");
this.quota = params.getOrDefault(QUOTA, "");
+ this.txnId = Long.parseLong(Objects.requireNonNull(params.get(TXN_ID),
+ "required property '" + TXN_ID + "'."));
this.writeSessionId = Objects.requireNonNull(params.get(WRITE_SESSION_ID),
"required property '" + WRITE_SESSION_ID + "'.");
- this.blockId = Long.parseLong(params.getOrDefault(BLOCK_ID, "0"));
- this.nextBlockId = this.blockId + 1; // Reserve blockId for first writer, increment for segments
+ this.preallocatedBlockId = params.containsKey(BLOCK_ID)
+ ? Long.parseLong(params.get(BLOCK_ID)) : null;
this.partitionSpec = params.getOrDefault(PARTITION_SPEC, "");
this.connectTimeout = Integer.parseInt(params.getOrDefault(CONNECT_TIMEOUT, "10"));
this.readTimeout = Integer.parseInt(params.getOrDefault(READ_TIMEOUT, "120"));
this.retryCount = Integer.parseInt(params.getOrDefault(RETRY_COUNT, "4"));
- this.maxWriteBatchRows = Integer.parseInt(params.getOrDefault(MAX_WRITE_BATCH_ROWS, "4096"));
+ this.maxBlockBytes = Long.parseLong(
+ params.getOrDefault(MCProperties.WRITE_MAX_BLOCK_BYTES,
+ MCProperties.DEFAULT_WRITE_MAX_BLOCK_BYTES));
}
@Override
@@ -212,17 +204,15 @@ public class MaxComputeJniWriter extends JniWriter {
allocator = new RootAllocator(Long.MAX_VALUE);
- // Cache writer options for creating new batchWriters in segments
writerOptions = WriterOptions.newBuilder()
.withSettings(settings)
.withCompressionCodec(CompressionCodec.ZSTD)
.build();
- batchWriter = writeSession.createArrowWriter(blockId,
- WriterAttemptId.of(0), writerOptions);
+ openBatchWriter(resolveInitialBlockId());
LOG.info("MaxComputeJniWriter opened: project=" + project + ", table=" + tableName
+ ", writeSessionId=" + writeSessionId + ", partitionSpec=" + partitionSpec
- + ", blockId=" + blockId);
+ + ", blockId=" + currentBlockId);
} catch (Exception e) {
String errorMsg = "Failed to open MaxCompute write session for table " + project + "." + tableName;
LOG.error(errorMsg, e);
@@ -239,44 +229,7 @@ public class MaxComputeJniWriter extends JniWriter {
}
try {
- // Stream data directly from off-heap VectorColumn to Arrow vectors.
- // Unlike the previous getMaterializedData() approach that created
- // Object[][] (with String objects for STRING columns causing 3x memory
- // amplification), this reads bytes directly from VectorColumn and writes
- // to Arrow, keeping peak heap usage per batch to O(batch_rows * row_size)
- // instead of O(2 * batch_rows * row_size).
- int rowOffset = 0;
- while (rowOffset < numRows) {
- int batchRows = Math.min(maxWriteBatchRows, numRows - rowOffset);
-
- // For variable-width columns, check byte budget to avoid Arrow int32 overflow
- batchRows = limitWriteBatchByBytesStreaming(inputTable, numCols,
- rowOffset, batchRows);
-
- VectorSchemaRoot root = batchWriter.newElement();
- try {
- root.setRowCount(batchRows);
-
- for (int col = 0; col < numCols && col < columnTypeInfos.size(); col++) {
- OdpsType odpsType = columnTypeInfos.get(col).getOdpsType();
- fillArrowVectorStreaming(root, col, odpsType,
- inputTable.getColumn(col), rowOffset, batchRows);
- }
-
- batchWriter.write(root);
- } finally {
- root.close();
- }
-
- writtenRows += batchRows;
- segmentRows += batchRows;
- rowOffset += batchRows;
-
- // Segmented commit: rotate batchWriter to release SDK native memory
- if (segmentRows >= ROWS_PER_SEGMENT) {
- rotateBatchWriter();
- }
- }
+ writeRowsWithRowChecks(inputTable, numRows, numCols);
} catch (Exception e) {
String errorMsg = "Failed to write data to MaxCompute table " + project + "." + tableName;
LOG.error(errorMsg, e);
@@ -284,116 +237,120 @@ public class MaxComputeJniWriter extends JniWriter {
}
}
- /**
- * Commit current batchWriter and create a new one with a fresh blockId.
- * This forces the MaxCompute SDK to flush and release internal native memory
- * buffers that accumulate during writes. Without rotation, the SDK holds all
- * serialized Arrow data in native memory until close(), causing process RSS
- * to grow linearly with total data volume.
- */
- private void rotateBatchWriter() throws IOException {
- try {
- // 1. Commit current batchWriter and save its commit message
- WriterCommitMessage msg = batchWriter.commit();
- commitMessages.add(msg);
- batchWriter = null;
-
- // 2. Close current allocator to release Arrow native memory
- allocator.close();
- allocator = null;
+ private long resolveInitialBlockId() {
+ return preallocatedBlockId != null ? preallocatedBlockId : requestBlockId();
+ }
- // 3. Create new allocator and batchWriter with a new blockId
- long newBlockId = nextBlockId++;
- allocator = new RootAllocator(Long.MAX_VALUE);
- batchWriter = writeSession.createArrowWriter(newBlockId,
- WriterAttemptId.of(0), writerOptions);
+ private long requestBlockId() {
+ return JNINativeMethod.requestMaxComputeBlockId(txnId, writeSessionId);
+ }
- LOG.info("Rotated batchWriter: oldBlockId=" + blockId + ", newBlockId=" + newBlockId
- + ", totalCommitMessages=" + commitMessages.size()
- + ", totalWrittenRows=" + writtenRows);
+ private void openBatchWriter(long blockId) throws IOException {
+ currentBlockId = blockId;
+ currentBlockWrittenBytes = 0L;
+ batchWriter = writeSession.createArrowWriter(blockId, WriterAttemptId.of(0), writerOptions);
+ }
- blockId = newBlockId;
- segmentRows = 0;
- } catch (Exception e) {
- throw new IOException("Failed to rotate batchWriter for table "
- + project + "." + tableName, e);
+ private void closeCurrentBatchWriterAndCollectCommit() throws IOException {
+ if (batchWriter == null) {
+ return;
+ }
+ WriterCommitMessage commitMessage = batchWriter.commit();
+ if (commitMessage != null) {
+ commitMessages.add(commitMessage);
}
+ batchWriter = null;
}
-
- private boolean isVariableWidthType(OdpsType type) {
- return type == OdpsType.STRING || type == OdpsType.VARCHAR
- || type == OdpsType.CHAR || type == OdpsType.BINARY;
+ private void rotateCurrentBatchWriter() throws IOException {
+ closeCurrentBatchWriterAndCollectCommit();
+ openBatchWriter(requestBlockId());
}
- /**
- * Limit write batch size by estimating variable-width column bytes directly
- * from the off-heap VectorColumn, without materializing data to Java heap.
- */
- private int limitWriteBatchByBytesStreaming(VectorTable inputTable, int numCols,
- int rowOffset, int batchRows) {
- for (int col = 0; col < numCols && col < columnTypeInfos.size(); col++) {
- OdpsType odpsType = columnTypeInfos.get(col).getOdpsType();
- if (!isVariableWidthType(odpsType)) {
- continue;
+ private void writeRowsWithRowChecks(VectorTable inputTable, int numRows, int numCols) throws IOException {
+ int rowStart = 0;
+ while (rowStart < numRows) {
+ int rowEnd = rowStart;
+ long batchEstimatedBytes = 0L;
+ boolean rotateAfterWrite = false;
+ while (rowEnd < numRows) {
+ long rowEstimatedBytes = estimateSingleRowPayloadBytes(inputTable, numCols, rowEnd);
+ boolean exceedsHardLimit = currentBlockWrittenBytes + batchEstimatedBytes
+ + rowEstimatedBytes > maxBlockBytes;
+ if (exceedsHardLimit) {
+ if (rowEnd == rowStart) {
+ if (currentBlockWrittenBytes > 0) {
+ rotateCurrentBatchWriter();
+ continue;
+ }
+ batchEstimatedBytes += rowEstimatedBytes;
+ rowEnd++;
+ rotateAfterWrite = true;
+ }
+ break;
+ }
+ batchEstimatedBytes += rowEstimatedBytes;
+ rowEnd++;
+ if (currentBlockWrittenBytes + batchEstimatedBytes >= maxBlockBytes) {
+ rotateAfterWrite = true;
+ break;
+ }
}
- VectorColumn vc = inputTable.getColumn(col);
- batchRows = findMaxRowsForColumnStreaming(vc, rowOffset, batchRows, MAX_ARROW_BATCH_BYTES);
- if (batchRows <= 1) {
- return Math.max(1, batchRows);
+
+ if (rowEnd == rowStart) {
+ long rowEstimatedBytes = estimateSingleRowPayloadBytes(inputTable, numCols, rowStart);
+ batchEstimatedBytes = rowEstimatedBytes;
+ rowEnd = rowStart + 1;
+ rotateAfterWrite = true;
+ }
+
+ try (VectorSchemaRoot root = buildRowRangeRoot(inputTable, numCols, rowStart, rowEnd)) {
+ batchWriter.write(root);
+ }
+ batchWriter.flush();
+ int rowsWrittenNow = rowEnd - rowStart;
+ writtenRows += rowsWrittenNow;
+ currentBlockWrittenBytes += batchEstimatedBytes;
+ writtenBytes += batchEstimatedBytes;
+ rowStart = rowEnd;
+
+ if (rotateAfterWrite && rowStart < numRows) {
+ rotateCurrentBatchWriter();
}
}
- return batchRows;
}
- /**
- * Find the maximum number of rows (from rowOffset) whose total byte size
- * fits within budget, by reading offset metadata directly from VectorColumn.
- */
- private int findMaxRowsForColumnStreaming(VectorColumn vc, int rowOffset, int maxRows, long budget) {
- long totalBytes = estimateColumnBytesStreaming(vc, rowOffset, maxRows);
- if (totalBytes <= budget) {
- return maxRows;
+ private static class CountingDiscardOutputStream extends OutputStream {
+ @Override
+ public void write(int b) {
+ // Discard bytes while allowing WriteChannel to track payload size.
}
- int rows = maxRows;
- while (rows > 1) {
- rows = rows / 2;
- totalBytes = estimateColumnBytesStreaming(vc, rowOffset, rows);
- if (totalBytes <= budget) {
- int lo = rows;
- int hi = Math.min(rows * 2, maxRows);
- while (lo < hi) {
- int mid = lo + (hi - lo + 1) / 2;
- if (estimateColumnBytesStreaming(vc, rowOffset, mid) <= budget) {
- lo = mid;
- } else {
- hi = mid - 1;
- }
- }
- return lo;
- }
+
+ @Override
+ public void write(byte[] b, int off, int len) {
+ // Discard bytes while allowing WriteChannel to track payload size.
}
- return 1;
}
- /**
- * Estimate total bytes for a range of rows in a VectorColumn by reading
- * the offset array directly from off-heap memory, without creating any
- * byte[] objects. This is O(1) per row (just offset subtraction).
- */
- private long estimateColumnBytesStreaming(VectorColumn vc, int rowOffset, int rows) {
- long total = 0;
- long offsetAddr = vc.offsetAddress();
- for (int i = rowOffset; i < rowOffset + rows; i++) {
- if (!vc.isNullAt(i)) {
- // String offsets are stored as int32 in VectorColumn
- int startOff = i == 0 ? 0
- : org.apache.doris.common.jni.utils.OffHeap.getInt(null, offsetAddr + 4L * (i - 1));
- int endOff = org.apache.doris.common.jni.utils.OffHeap.getInt(null, offsetAddr + 4L * i);
- total += (endOff - startOff);
- }
+ private long estimateSingleRowPayloadBytes(VectorTable inputTable, int numCols, int rowIndex)
+ throws IOException {
+ try (VectorSchemaRoot root = buildRowRangeRoot(inputTable, numCols, rowIndex, rowIndex + 1);
+ ArrowWriter estimator = ArrowWriterFactory.getRecordBatchWriter(
+ new CountingDiscardOutputStream(), writerOptions)) {
+ estimator.writeBatch(root);
+ return estimator.bytesWritten();
+ }
+ }
+
+ private VectorSchemaRoot buildRowRangeRoot(VectorTable inputTable, int numCols, int rowStart, int rowEnd) {
+ int rowCount = rowEnd - rowStart;
+ VectorSchemaRoot root = batchWriter.newElement();
+ root.setRowCount(rowCount);
+ for (int col = 0; col < numCols && col < columnTypeInfos.size(); col++) {
+ fillArrowVectorStreaming(root, col, columnTypeInfos.get(col).getOdpsType(),
+ inputTable.getColumn(col), rowStart, rowCount);
}
- return total;
+ return root;
}
/**
@@ -512,17 +469,13 @@ public class MaxComputeJniWriter extends JniWriter {
case STRING:
case VARCHAR:
case CHAR: {
- // KEY FIX: Read bytes directly from off-heap, no String creation.
- // Previously: getMaterializedData -> String[] -> toString().getBytes() -> Arrow
- // Now: getBytesWithOffset() -> byte[] -> Arrow (1 copy instead of 3)
VarCharVector vec = (VarCharVector) root.getVector(colIdx);
vec.allocateNew(numRows);
for (int i = 0; i < numRows; i++) {
if (vc.isNullAt(rowOffset + i)) {
vec.setNull(i);
} else {
- byte[] bytes = vc.getBytesWithOffset(rowOffset + i);
- vec.setSafe(i, bytes);
+ vec.setSafe(i, vc.getBytesWithOffset(rowOffset + i));
}
}
vec.setValueCount(numRows);
@@ -535,8 +488,7 @@ public class MaxComputeJniWriter extends JniWriter {
if (vc.isNullAt(rowOffset + i)) {
vec.setNull(i);
} else {
- LocalDate date = vc.getDate(rowOffset + i);
- vec.set(i, (int) date.toEpochDay());
+ vec.set(i, (int) vc.getDate(rowOffset + i).toEpochDay());
}
}
vec.setValueCount(numRows);
@@ -550,8 +502,8 @@ public class MaxComputeJniWriter extends JniWriter {
if (vc.isNullAt(rowOffset + i)) {
vec.setNull(i);
} else {
- LocalDateTime dt = vc.getDateTime(rowOffset + i);
- long millis = dt.atZone(ZoneId.systemDefault()).toInstant().toEpochMilli();
+ long millis = vc.getDateTime(rowOffset + i)
+ .atZone(ZoneId.systemDefault()).toInstant().toEpochMilli();
vec.set(i, millis);
}
}
@@ -565,16 +517,15 @@ public class MaxComputeJniWriter extends JniWriter {
if (vc.isNullAt(rowOffset + i)) {
vec.setNull(i);
} else {
- byte[] bytes = vc.getBytesWithOffset(rowOffset + i);
- vec.setSafe(i, bytes);
+ vec.setSafe(i, vc.getBytesWithOffset(rowOffset + i));
}
}
vec.setValueCount(numRows);
break;
}
default: {
- // For complex types (ARRAY, MAP, STRUCT) and other types,
- // fall back to object-based materialization for this column only.
+ // Complex types still fall back to object materialization; the
+ // optimization target here is large variable-width primitive columns.
Object[] colData = vc.getObjectColumn(rowOffset, rowOffset + numRows);
fillArrowVector(root, colIdx, odpsType, colData, 0, numRows);
break;
@@ -583,16 +534,17 @@ public class MaxComputeJniWriter extends JniWriter {
}
private void fillArrowVector(VectorSchemaRoot root, int colIdx, OdpsType odpsType,
- Object[] colData, int rowOffset, int numRows) {
+ Object[] colData, int startRow, int numRows) {
switch (odpsType) {
case BOOLEAN: {
BitVector vec = (BitVector) root.getVector(colIdx);
vec.allocateNew(numRows);
for (int i = 0; i < numRows; i++) {
- if (colData[rowOffset + i] == null) {
+ Object value = colData[startRow + i];
+ if (value == null) {
vec.setNull(i);
} else {
- vec.set(i, (Boolean) colData[rowOffset + i] ? 1 : 0);
+ vec.set(i, (Boolean) value ? 1 : 0);
}
}
vec.setValueCount(numRows);
@@ -602,10 +554,11 @@ public class MaxComputeJniWriter extends JniWriter {
TinyIntVector vec = (TinyIntVector) root.getVector(colIdx);
vec.allocateNew(numRows);
for (int i = 0; i < numRows; i++) {
- if (colData[rowOffset + i] == null) {
+ Object value = colData[startRow + i];
+ if (value == null) {
vec.setNull(i);
} else {
- vec.set(i, ((Number) colData[rowOffset + i]).byteValue());
+ vec.set(i, ((Number) value).byteValue());
}
}
vec.setValueCount(numRows);
@@ -615,10 +568,11 @@ public class MaxComputeJniWriter extends JniWriter {
SmallIntVector vec = (SmallIntVector) root.getVector(colIdx);
vec.allocateNew(numRows);
for (int i = 0; i < numRows; i++) {
- if (colData[rowOffset + i] == null) {
+ Object value = colData[startRow + i];
+ if (value == null) {
vec.setNull(i);
} else {
- vec.set(i, ((Number) colData[rowOffset + i]).shortValue());
+ vec.set(i, ((Number) value).shortValue());
}
}
vec.setValueCount(numRows);
@@ -628,10 +582,11 @@ public class MaxComputeJniWriter extends JniWriter {
IntVector vec = (IntVector) root.getVector(colIdx);
vec.allocateNew(numRows);
for (int i = 0; i < numRows; i++) {
- if (colData[rowOffset + i] == null) {
+ Object value = colData[startRow + i];
+ if (value == null) {
vec.setNull(i);
} else {
- vec.set(i, ((Number) colData[rowOffset + i]).intValue());
+ vec.set(i, ((Number) value).intValue());
}
}
vec.setValueCount(numRows);
@@ -641,10 +596,11 @@ public class MaxComputeJniWriter extends JniWriter {
BigIntVector vec = (BigIntVector) root.getVector(colIdx);
vec.allocateNew(numRows);
for (int i = 0; i < numRows; i++) {
- if (colData[rowOffset + i] == null) {
+ Object value = colData[startRow + i];
+ if (value == null) {
vec.setNull(i);
} else {
- vec.set(i, ((Number) colData[rowOffset + i]).longValue());
+ vec.set(i, ((Number) value).longValue());
}
}
vec.setValueCount(numRows);
@@ -654,10 +610,11 @@ public class MaxComputeJniWriter extends JniWriter {
Float4Vector vec = (Float4Vector) root.getVector(colIdx);
vec.allocateNew(numRows);
for (int i = 0; i < numRows; i++) {
- if (colData[rowOffset + i] == null) {
+ Object value = colData[startRow + i];
+ if (value == null) {
vec.setNull(i);
} else {
- vec.set(i, ((Number) colData[rowOffset + i]).floatValue());
+ vec.set(i, ((Number) value).floatValue());
}
}
vec.setValueCount(numRows);
@@ -667,10 +624,11 @@ public class MaxComputeJniWriter extends JniWriter {
Float8Vector vec = (Float8Vector) root.getVector(colIdx);
vec.allocateNew(numRows);
for (int i = 0; i < numRows; i++) {
- if (colData[rowOffset + i] == null) {
+ Object value = colData[startRow + i];
+ if (value == null) {
vec.setNull(i);
} else {
- vec.set(i, ((Number) colData[rowOffset + i]).doubleValue());
+ vec.set(i, ((Number) value).doubleValue());
}
}
vec.setValueCount(numRows);
@@ -680,12 +638,13 @@ public class MaxComputeJniWriter extends JniWriter {
DecimalVector vec = (DecimalVector) root.getVector(colIdx);
vec.allocateNew(numRows);
for (int i = 0; i < numRows; i++) {
- if (colData[rowOffset + i] == null) {
+ Object value = colData[startRow + i];
+ if (value == null) {
vec.setNull(i);
} else {
- BigDecimal bd = (colData[rowOffset + i] instanceof BigDecimal)
- ? (BigDecimal) colData[rowOffset + i]
- : new BigDecimal(colData[rowOffset + i].toString());
+ BigDecimal bd = (value instanceof BigDecimal)
+ ? (BigDecimal) value
+ : new BigDecimal(value.toString());
vec.set(i, bd);
}
}
@@ -698,14 +657,15 @@ public class MaxComputeJniWriter extends JniWriter {
VarCharVector vec = (VarCharVector) root.getVector(colIdx);
vec.allocateNew(numRows);
for (int i = 0; i < numRows; i++) {
- if (colData[rowOffset + i] == null) {
+ Object value = colData[startRow + i];
+ if (value == null) {
vec.setNull(i);
} else {
byte[] bytes;
- if (colData[rowOffset + i] instanceof byte[]) {
- bytes = (byte[]) colData[rowOffset + i];
+ if (value instanceof byte[]) {
+ bytes = (byte[]) value;
} else {
- bytes = colData[rowOffset + i].toString().getBytes(StandardCharsets.UTF_8);
+ bytes = value.toString().getBytes(StandardCharsets.UTF_8);
}
vec.setSafe(i, bytes);
}
@@ -717,12 +677,13 @@ public class MaxComputeJniWriter extends JniWriter {
DateDayVector vec = (DateDayVector) root.getVector(colIdx);
vec.allocateNew(numRows);
for (int i = 0; i < numRows; i++) {
- if (colData[rowOffset + i] == null) {
+ Object value = colData[startRow + i];
+ if (value == null) {
vec.setNull(i);
- } else if (colData[rowOffset + i] instanceof LocalDate) {
- vec.set(i, (int) ((LocalDate) colData[rowOffset + i]).toEpochDay());
+ } else if (value instanceof LocalDate) {
+ vec.set(i, (int) ((LocalDate) value).toEpochDay());
} else {
- vec.set(i, (int) LocalDate.parse(colData[rowOffset + i].toString()).toEpochDay());
+ vec.set(i, (int) LocalDate.parse(value.toString()).toEpochDay());
}
}
vec.setValueCount(numRows);
@@ -733,16 +694,17 @@ public class MaxComputeJniWriter extends JniWriter {
TimeStampMilliVector vec = (TimeStampMilliVector) root.getVector(colIdx);
vec.allocateNew(numRows);
for (int i = 0; i < numRows; i++) {
- if (colData[rowOffset + i] == null) {
+ Object value = colData[startRow + i];
+ if (value == null) {
vec.setNull(i);
- } else if (colData[rowOffset + i] instanceof LocalDateTime) {
- long millis = ((LocalDateTime) colData[rowOffset + i])
+ } else if (value instanceof LocalDateTime) {
+ long millis = ((LocalDateTime) value)
.atZone(ZoneId.systemDefault()).toInstant().toEpochMilli();
vec.set(i, millis);
- } else if (colData[rowOffset + i] instanceof java.sql.Timestamp) {
- vec.set(i, ((java.sql.Timestamp) colData[rowOffset + i]).getTime());
+ } else if (value instanceof java.sql.Timestamp) {
+ vec.set(i, ((java.sql.Timestamp) value).getTime());
} else {
- long millis = LocalDateTime.parse(colData[rowOffset + i].toString())
+ long millis = LocalDateTime.parse(value.toString())
.atZone(ZoneId.systemDefault()).toInstant().toEpochMilli();
vec.set(i, millis);
}
@@ -754,12 +716,13 @@ public class MaxComputeJniWriter extends JniWriter {
VarBinaryVector vec = (VarBinaryVector) root.getVector(colIdx);
vec.allocateNew(numRows);
for (int i = 0; i < numRows; i++) {
- if (colData[rowOffset + i] == null) {
+ Object value = colData[startRow + i];
+ if (value == null) {
vec.setNull(i);
- } else if (colData[rowOffset + i] instanceof byte[]) {
- vec.setSafe(i, (byte[]) colData[rowOffset + i]);
+ } else if (value instanceof byte[]) {
+ vec.setSafe(i, (byte[]) value);
} else {
- vec.setSafe(i, colData[rowOffset + i].toString().getBytes(StandardCharsets.UTF_8));
+ vec.setSafe(i, value.toString().getBytes(StandardCharsets.UTF_8));
}
}
vec.setValueCount(numRows);
@@ -771,10 +734,11 @@ public class MaxComputeJniWriter extends JniWriter {
FieldVector dataVec = listVec.getDataVector();
int elemIdx = 0;
for (int i = 0; i < numRows; i++) {
- if (colData[rowOffset + i] == null) {
+ Object value = colData[startRow + i];
+ if (value == null) {
listVec.setNull(i);
} else {
- List<?> list = (List<?>) colData[rowOffset + i];
+ List<?> list = (List<?>) value;
listVec.startNewValue(i);
for (Object elem : list) {
writeListElement(dataVec, elemIdx++, elem);
@@ -794,10 +758,11 @@ public class MaxComputeJniWriter extends JniWriter {
FieldVector valVec = structVec.getChildrenFromFields().get(1);
int elemIdx = 0;
for (int i = 0; i < numRows; i++) {
- if (colData[rowOffset + i] == null) {
+ Object value = colData[startRow + i];
+ if (value == null) {
mapVec.setNull(i);
} else {
- Map<?, ?> map = (Map<?, ?>) colData[rowOffset + i];
+ Map<?, ?> map = (Map<?, ?>) value;
mapVec.startNewValue(i);
for (Map.Entry<?, ?> entry : map.entrySet()) {
structVec.setIndexDefined(elemIdx);
@@ -818,11 +783,12 @@ public class MaxComputeJniWriter extends JniWriter {
StructVector structVec = (StructVector) root.getVector(colIdx);
structVec.allocateNew();
for (int i = 0; i < numRows; i++) {
- if (colData[rowOffset + i] == null) {
+ Object value = colData[startRow + i];
+ if (value == null) {
structVec.setNull(i);
} else {
structVec.setIndexDefined(i);
- Map<?, ?> struct = (Map<?, ?>) colData[rowOffset + i];
+ Map<?, ?> struct = (Map<?, ?>) value;
for (FieldVector childVec : structVec.getChildrenFromFields()) {
Object val = struct.get(childVec.getName());
writeListElement(childVec, i, val);
@@ -840,10 +806,11 @@ public class MaxComputeJniWriter extends JniWriter {
VarCharVector vec = (VarCharVector) root.getVector(colIdx);
vec.allocateNew(numRows);
for (int i = 0; i < numRows; i++) {
- if (colData[rowOffset + i] == null) {
+ Object value = colData[startRow + i];
+ if (value == null) {
vec.setNull(i);
} else {
- vec.setSafe(i, colData[rowOffset + i].toString().getBytes(StandardCharsets.UTF_8));
+ vec.setSafe(i, value.toString().getBytes(StandardCharsets.UTF_8));
}
}
vec.setValueCount(numRows);
@@ -932,41 +899,21 @@ public class MaxComputeJniWriter extends JniWriter {
@Override
public void close() throws IOException {
- Exception firstException = null;
try {
- // Commit the final segment's batchWriter
- if (batchWriter != null) {
- try {
- WriterCommitMessage msg = batchWriter.commit();
- commitMessages.add(msg);
- } catch (Exception e) {
- firstException = e;
- LOG.warn("Failed to commit batch writer for table " + project + "." + tableName, e);
- } finally {
- batchWriter = null;
- }
- }
- } finally {
+ closeCurrentBatchWriterAndCollectCommit();
if (allocator != null) {
- try {
- allocator.close();
- } catch (Exception e) {
- LOG.warn("Failed to close Arrow allocator (possible memory leak)", e);
- if (firstException == null) {
- firstException = e;
- }
- } finally {
- allocator = null;
- }
- }
- }
- LOG.info("MaxComputeJniWriter closed: writeSessionId=" + writeSessionId
- + ", partitionSpec=" + partitionSpec
- + ", writtenRows=" + writtenRows
- + ", totalSegments=" + commitMessages.size()
- + ", blockId=" + blockId);
- if (firstException != null) {
- throw new IOException("Failed to close MaxCompute arrow writer", firstException);
+ allocator.close();
+ allocator = null;
+ }
+ LOG.info("MaxComputeJniWriter closed: writeSessionId=" + writeSessionId
+ + ", partitionSpec=" + partitionSpec
+ + ", writtenRows=" + writtenRows
+ + ", lastBlockId=" + currentBlockId
+ + ", commitMessageCount=" + commitMessages.size());
+ } catch (Exception e) {
+ String errorMsg = "Failed to close MaxCompute arrow writer";
+ LOG.error(errorMsg, e);
+ throw new IOException(errorMsg, e);
}
}
@@ -975,14 +922,12 @@ public class MaxComputeJniWriter extends JniWriter {
Map<String, String> stats = new HashMap<>();
stats.put("mc_partition_spec", partitionSpec != null ? partitionSpec : "");
- // Serialize all WriterCommitMessages (one per segment) as a List object.
+ // Serialize commit messages to Base64.
if (!commitMessages.isEmpty()) {
try {
ByteArrayOutputStream baos = new ByteArrayOutputStream();
ObjectOutputStream oos = new ObjectOutputStream(baos);
- // Serialize the entire list as one object to avoid mixing
- // writeInt/writeObject which causes OptionalDataException
- oos.writeObject(new java.util.ArrayList<>(commitMessages));
+ oos.writeObject(commitMessages);
oos.close();
stats.put("mc_commit_message", Base64.getEncoder().encodeToString(baos.toByteArray()));
} catch (IOException e) {
@@ -997,4 +942,3 @@ public class MaxComputeJniWriter extends JniWriter {
return stats;
}
}
-
diff --git a/fe/fe-common/src/main/java/org/apache/doris/common/maxcompute/MCProperties.java b/fe/fe-common/src/main/java/org/apache/doris/common/maxcompute/MCProperties.java
index 9ca45b19ea8..717871825c6 100644
--- a/fe/fe-common/src/main/java/org/apache/doris/common/maxcompute/MCProperties.java
+++ b/fe/fe-common/src/main/java/org/apache/doris/common/maxcompute/MCProperties.java
@@ -60,8 +60,11 @@ public class MCProperties {
public static final String MAX_FIELD_SIZE = "mc.max_field_size_bytes";
public static final String DEFAULT_MAX_FIELD_SIZE = "8388608"; // 8 * 1024 * 1024 = 8MB
- public static final String MAX_WRITE_BATCH_ROWS = "mc.max_write_batch_rows";
- public static final String DEFAULT_MAX_WRITE_BATCH_ROWS = "4096";
+ // Approximate upper bound in bytes for a single MaxCompute write block.
+ // Once a writer block is estimated to exceed this threshold, Doris will
+ // switch to a new block_id before continuing the write.
+ public static final String WRITE_MAX_BLOCK_BYTES = "mc.write_max_block_bytes";
+ public static final String DEFAULT_WRITE_MAX_BLOCK_BYTES = "67108864"; // 64 * 1024 * 1024
//withCrossPartition(true):
// Very friendly to scenarios where there are many partitions but each partition is very small.
diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/maxcompute/MCTransaction.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/maxcompute/MCTransaction.java
index 77200d47cec..76a3c84ebb7 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/datasource/maxcompute/MCTransaction.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/maxcompute/MCTransaction.java
@@ -43,11 +43,13 @@ import java.util.Base64;
import java.util.List;
import java.util.Map;
import java.util.Optional;
+import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;
public class MCTransaction implements Transaction {
private static final Logger LOG = LogManager.getLogger(MCTransaction.class);
+ private static final long MAX_BLOCK_COUNT = 20000L;
private final MaxComputeExternalCatalog catalog;
private MaxComputeExternalTable table;
@@ -55,6 +57,7 @@ public class MCTransaction implements Transaction {
// Storage API write session ID (created in beginInsert, used in finishInsert)
private String writeSessionId;
+ private final AtomicLong nextBlockId = new AtomicLong(0);
public MCTransaction(MaxComputeExternalCatalog catalog) {
this.catalog = catalog;
@@ -113,6 +116,7 @@ public class MCTransaction implements Transaction {
TableBatchWriteSession writeSession = builder.buildBatchWriteSession();
writeSessionId = writeSession.getId();
+ nextBlockId.set(0);
LOG.info("Created MC Storage API write session: {} for table {}.{}",
writeSessionId, catalog.getDefaultProject(), table.getName());
@@ -126,6 +130,60 @@ public class MCTransaction implements Transaction {
return writeSessionId;
}
+ public long allocateBlockIdRange(String requestWriteSessionId, long length) throws UserException {
+ if (length <= 0) {
+ throw new UserException("MaxCompute block_id allocation length must be positive: " + length);
+ }
+ if (writeSessionId == null || writeSessionId.isEmpty()) {
+ throw new UserException("MaxCompute write session has not been initialized");
+ }
+ if (!writeSessionId.equals(requestWriteSessionId)) {
+ throw new UserException("MaxCompute write session mismatch, expected=" + writeSessionId
+ + ", actual=" + requestWriteSessionId);
+ }
+
+ long start;
+ long endExclusive;
+ do {
+ start = nextBlockId.get();
+ endExclusive = start + length;
+ if (endExclusive > MAX_BLOCK_COUNT) {
+ throw new UserException("MaxCompute block_id exceeds limit, start="
+ + start + ", length=" + length + ", maxBlockCount=" + MAX_BLOCK_COUNT);
+ }
+ } while (!nextBlockId.compareAndSet(start, endExclusive));
+
+ LOG.info("Allocated MaxCompute block_id range: sessionId={}, start={}, length={}",
+ writeSessionId, start, length);
+ return start;
+ }
+
+ private void appendCommitMessages(List<WriterCommitMessage> allMessages, String encodedCommitMessage)
+ throws Exception {
+ byte[] bytes = Base64.getDecoder().decode(encodedCommitMessage);
+ ByteArrayInputStream bais = new ByteArrayInputStream(bytes);
+ ObjectInputStream ois = new ObjectInputStream(bais);
+ Object payload = ois.readObject();
+ ois.close();
+
+ if (payload instanceof WriterCommitMessage) {
+ allMessages.add((WriterCommitMessage) payload);
+ return;
+ }
+ if (payload instanceof List<?>) {
+ for (Object item : (List<?>) payload) {
+ if (!(item instanceof WriterCommitMessage)) {
+ throw new UserException("Unexpected MaxCompute commit payload item type: "
+ + (item == null ? "null" : item.getClass().getName()));
+ }
+ allMessages.add((WriterCommitMessage) item);
+ }
+ return;
+ }
+ throw new UserException("Unexpected MaxCompute commit payload type: "
+ + (payload == null ? "null" : payload.getClass().getName()));
+ }
+
public void finishInsert() throws UserException {
try {
long t0 = System.currentTimeMillis();
@@ -134,16 +192,7 @@ public class MCTransaction implements Transaction {
synchronized (this) {
for (TMCCommitData data : commitDataList) {
if (data.isSetCommitMessage() && !data.getCommitMessage().isEmpty()) {
- byte[] bytes = Base64.getDecoder().decode(data.getCommitMessage());
- ByteArrayInputStream bais = new ByteArrayInputStream(bytes);
- ObjectInputStream ois = new ObjectInputStream(bais);
- // Deserialized as List<WriterCommitMessage> — supports segmented
- // commit where one writer produces multiple commit messages
- @SuppressWarnings("unchecked")
- List<WriterCommitMessage> msgs =
- (List<WriterCommitMessage>) ois.readObject();
- allMessages.addAll(msgs);
- ois.close();
+ appendCommitMessages(allMessages, data.getCommitMessage());
}
}
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/maxcompute/MaxComputeExternalCatalog.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/maxcompute/MaxComputeExternalCatalog.java
index b3a15355b14..1292746cc25 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/datasource/maxcompute/MaxComputeExternalCatalog.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/maxcompute/MaxComputeExternalCatalog.java
@@ -70,7 +70,6 @@ public class MaxComputeExternalCatalog extends ExternalCatalog {
private int readTimeout;
private int retryTimes;
private long maxFieldSize;
- private int maxWriteBatchRows;
public boolean dateTimePredicatePushDown;
@@ -195,8 +194,6 @@ public class MaxComputeExternalCatalog extends ExternalCatalog {
props.getOrDefault(MCProperties.RETRY_COUNT, MCProperties.DEFAULT_RETRY_COUNT));
maxFieldSize = Long.parseLong(
props.getOrDefault(MCProperties.MAX_FIELD_SIZE, MCProperties.DEFAULT_MAX_FIELD_SIZE));
- maxWriteBatchRows = Integer.parseInt(
- props.getOrDefault(MCProperties.MAX_WRITE_BATCH_ROWS, MCProperties.DEFAULT_MAX_WRITE_BATCH_ROWS));
RestOptions restOptions = RestOptions.newBuilder()
.withConnectTimeout(connectTimeout)
@@ -331,11 +328,6 @@ public class MaxComputeExternalCatalog extends ExternalCatalog {
return maxFieldSize;
}
- public int getMaxWriteBatchRows() {
- makeSureInitialized();
- return maxWriteBatchRows;
- }
-
public boolean getDateTimePredicatePushDown() {
return dateTimePredicatePushDown;
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/MCInsertExecutor.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/MCInsertExecutor.java
index d490f96419e..47df06485e7 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/MCInsertExecutor.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/MCInsertExecutor.java
@@ -64,9 +64,9 @@ public class MCInsertExecutor extends BaseExternalTableInsertExecutor {
MCTransaction transaction = (MCTransaction) transactionManager.getTransaction(txnId);
transaction.beginInsert((MaxComputeExternalTable) table, insertCtx);
- // 2. Inject writeSessionId into the Thrift sink before fragments are sent to BE
+ // 2. Inject write context into the Thrift sink before fragments are sent to BE
if (mcTableSink != null) {
- mcTableSink.setWriteSessionId(transaction.getWriteSessionId());
+ mcTableSink.setWriteContext(txnId, transaction.getWriteSessionId());
}
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/MaxComputeTableSink.java b/fe/fe-core/src/main/java/org/apache/doris/planner/MaxComputeTableSink.java
index bf52e894628..98537fa0307 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/MaxComputeTableSink.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/MaxComputeTableSink.java
@@ -74,7 +74,6 @@ public class MaxComputeTableSink extends BaseExternalTableDataSink {
tSink.setConnectTimeout(catalog.getConnectTimeout());
tSink.setReadTimeout(catalog.getReadTimeout());
tSink.setRetryCount(catalog.getRetryTimes());
- tSink.setMaxWriteBatchRows(catalog.getMaxWriteBatchRows());
// Partition columns
List<String> partitionColumnNames = targetTable.getPartitionColumns().stream()
@@ -101,12 +100,13 @@ public class MaxComputeTableSink extends BaseExternalTableDataSink {
}
/**
- * Called by MCInsertExecutor.beforeExec() to inject the writeSessionId
+ * Called by MCInsertExecutor.beforeExec() to inject runtime write context
* after MCTransaction.beginInsert() creates the Storage API session.
* This must be called before fragments are sent to BE (i.e., before execImpl).
*/
- public void setWriteSessionId(String writeSessionId) {
+ public void setWriteContext(long txnId, String writeSessionId) {
if (tDataSink != null && tDataSink.isSetMaxComputeTableSink()) {
+ tDataSink.getMaxComputeTableSink().setTxnId(txnId);
tDataSink.getMaxComputeTableSink().setWriteSessionId(writeSessionId);
}
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java b/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java
index 6d5cbaab063..3991b9c471e 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java
@@ -82,6 +82,7 @@ import org.apache.doris.datasource.ExternalCatalog;
import org.apache.doris.datasource.ExternalDatabase;
import org.apache.doris.datasource.InternalCatalog;
import org.apache.doris.datasource.SplitSource;
+import org.apache.doris.datasource.maxcompute.MCTransaction;
import org.apache.doris.encryption.EncryptionKey;
import org.apache.doris.info.TableNameInfo;
import org.apache.doris.info.TableRefInfo;
@@ -240,6 +241,8 @@ import org.apache.doris.thrift.TMasterAddressResult;
import org.apache.doris.thrift.TMasterOpRequest;
import org.apache.doris.thrift.TMasterOpResult;
import org.apache.doris.thrift.TMasterResult;
+import org.apache.doris.thrift.TMaxComputeBlockIdRequest;
+import org.apache.doris.thrift.TMaxComputeBlockIdResult;
import org.apache.doris.thrift.TMySqlLoadAcquireTokenResult;
import org.apache.doris.thrift.TNetworkAddress;
import org.apache.doris.thrift.TNodeInfo;
@@ -305,6 +308,7 @@ import org.apache.doris.thrift.TWaitingTxnStatusResult;
import org.apache.doris.transaction.BeginTransactionException;
import org.apache.doris.transaction.SubTransactionState;
import org.apache.doris.transaction.TabletCommitInfo;
+import org.apache.doris.transaction.Transaction;
import org.apache.doris.transaction.TransactionState;
import org.apache.doris.transaction.TransactionState.TxnCoordinator;
import org.apache.doris.transaction.TransactionState.TxnSourceType;
@@ -3592,6 +3596,50 @@ public class FrontendServiceImpl implements FrontendService.Iface {
return result;
}
+ @Override
+ public TMaxComputeBlockIdResult getMaxComputeBlockIdRange(TMaxComputeBlockIdRequest request) {
+ String clientAddr = getClientAddrAsString();
+ LOG.info("receive getMaxComputeBlockIdRange request: {}, backend: {}", request, clientAddr);
+
+ TMaxComputeBlockIdResult result = new TMaxComputeBlockIdResult();
+ TStatus status = checkMaster();
+ result.setStatus(status);
+
+ if (status.getStatusCode() != TStatusCode.OK) {
+ result.setMasterAddress(getMasterAddress());
+ return result;
+ }
+
+ try {
+ Transaction transaction = Env.getCurrentEnv().getGlobalExternalTransactionInfoMgr()
+ .getTxnById(request.getTxnId());
+ if (!(transaction instanceof MCTransaction)) {
+ throw new UserException("Transaction " + request.getTxnId()
+ + " is not a MaxCompute transaction");
+ }
+
+ long start = ((MCTransaction) transaction).allocateBlockIdRange(
+ request.getWriteSessionId(), request.getLength());
+ result.setStart(start);
+ result.setLength(request.getLength());
+ } catch (UserException e) {
+ LOG.warn("failed to allocate MaxCompute block_id, txnId={}, sessionId={}, errmsg={}",
+ request.getTxnId(), request.getWriteSessionId(), e.getMessage());
+ status.setStatusCode(TStatusCode.ANALYSIS_ERROR);
+ status.addToErrorMsgs(e.getMessage());
+ } catch (RuntimeException e) {
+ LOG.warn("failed to allocate MaxCompute block_id, txnId={}, sessionId={}, errmsg={}",
+ request.getTxnId(), request.getWriteSessionId(), e.getMessage(), e);
+ status.setStatusCode(TStatusCode.ANALYSIS_ERROR);
+ status.addToErrorMsgs(Strings.nullToEmpty(e.getMessage()));
+ } catch (Throwable e) {
+ LOG.warn("catch unknown result when allocating MaxCompute block_id.", e);
+ status.setStatusCode(TStatusCode.INTERNAL_ERROR);
+ status.addToErrorMsgs(e.getClass().getSimpleName() + ": " + Strings.nullToEmpty(e.getMessage()));
+ }
+ return result;
+ }
+
public TGetBinlogResult getBinlog(TGetBinlogRequest request) throws TException {
String clientAddr = getClientAddrAsString();
if (LOG.isDebugEnabled()) {
diff --git a/fe/fe-core/src/test/java/org/apache/doris/service/FrontendServiceImplTest.java b/fe/fe-core/src/test/java/org/apache/doris/service/FrontendServiceImplTest.java
index 00903892b23..0e136ace69a 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/service/FrontendServiceImplTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/service/FrontendServiceImplTest.java
@@ -24,6 +24,8 @@ import org.apache.doris.catalog.Partition;
import org.apache.doris.common.Config;
import org.apache.doris.common.FeConstants;
import org.apache.doris.common.util.DatasourcePrintableMap;
+import org.apache.doris.datasource.maxcompute.MCTransaction;
+import org.apache.doris.datasource.maxcompute.MaxComputeExternalCatalog;
import org.apache.doris.nereids.parser.NereidsParser;
import org.apache.doris.nereids.trees.plans.commands.CreateDatabaseCommand;
import org.apache.doris.nereids.trees.plans.commands.CreateTableCommand;
@@ -38,6 +40,8 @@ import org.apache.doris.thrift.TFetchSchemaTableDataRequest;
import org.apache.doris.thrift.TFetchSchemaTableDataResult;
import org.apache.doris.thrift.TGetDbsParams;
import org.apache.doris.thrift.TGetDbsResult;
+import org.apache.doris.thrift.TMaxComputeBlockIdRequest;
+import org.apache.doris.thrift.TMaxComputeBlockIdResult;
import org.apache.doris.thrift.TMetadataTableRequestParams;
import org.apache.doris.thrift.TMetadataType;
import org.apache.doris.thrift.TNullableStringLiteral;
@@ -56,6 +60,7 @@ import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.ExpectedException;
+import java.lang.reflect.Field;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
@@ -105,6 +110,22 @@ public class FrontendServiceImplTest {
}
}
+ private static void executeCommand(String sql) throws Exception {
+ NereidsParser nereidsParser = new NereidsParser();
+ LogicalPlan parsed = nereidsParser.parseSingle(sql);
+ StmtExecutor stmtExecutor = new StmtExecutor(connectContext, sql);
+ if (parsed instanceof Command) {
+ ((Command) parsed).run(connectContext, stmtExecutor);
+ return;
+ }
+ throw new IllegalArgumentException("Unsupported command in test: " + sql);
+ }
+
+ private static void setPrivateField(Object target, String fieldName, Object value) throws Exception {
+ Field field = target.getClass().getDeclaredField(fieldName);
+ field.setAccessible(true);
+ field.set(target, value);
+ }
@Test
public void testCreatePartitionRange() throws Exception {
@@ -251,6 +272,34 @@ public class FrontendServiceImplTest {
System.out.println(result);
}
+ @Test
+ public void testGetMaxComputeBlockIdRange() throws Exception {
+ FrontendServiceImpl impl = new FrontendServiceImpl(exeEnv);
+ long txnId = Env.getCurrentEnv().getNextId();
+ MCTransaction transaction = new MCTransaction(Mockito.mock(MaxComputeExternalCatalog.class));
+ setPrivateField(transaction, "writeSessionId", "session-1");
+ Env.getCurrentEnv().getGlobalExternalTransactionInfoMgr().putTxnById(txnId, transaction);
+
+ try {
+ TMaxComputeBlockIdRequest request = new TMaxComputeBlockIdRequest();
+ request.setTxnId(txnId);
+ request.setWriteSessionId("session-1");
+ request.setLength(1);
+
+ TMaxComputeBlockIdResult first = impl.getMaxComputeBlockIdRange(request);
+ Assert.assertEquals(TStatusCode.OK, first.getStatus().getStatusCode());
+ Assert.assertEquals(0L, first.getStart());
+ Assert.assertEquals(1L, first.getLength());
+
+ TMaxComputeBlockIdResult second = impl.getMaxComputeBlockIdRange(request);
+ Assert.assertEquals(TStatusCode.OK, second.getStatus().getStatusCode());
+ Assert.assertEquals(1L, second.getStart());
+ Assert.assertEquals(1L, second.getLength());
+ } finally {
+ Env.getCurrentEnv().getGlobalExternalTransactionInfoMgr().removeTxnById(txnId);
+ }
+ }
+
@Test
public void testFetchAuthenticationIntegrationsSchemaTableData() throws Exception {
String integrationName = "test_authentication_integration";
diff --git a/gensrc/thrift/DataSinks.thrift b/gensrc/thrift/DataSinks.thrift
index 89897fd4cf0..0d6d9126a06 100644
--- a/gensrc/thrift/DataSinks.thrift
+++ b/gensrc/thrift/DataSinks.thrift
@@ -599,7 +599,8 @@ struct TMaxComputeTableSink {
14: optional list<string> partition_columns // partition column names for dynamic partition
15: optional string write_session_id // Storage API write session ID
16: optional map<string, string> properties // contains authentication properties
- 17: optional i32 max_write_batch_rows // max rows per Arrow batch for write
+ 17: optional i32 max_write_batch_rows // max rows per Arrow batch for write, deprecated.
+ 18: optional i64 txn_id // FE external transaction ID for runtime block_id allocation
}
struct TDataSink {
diff --git a/gensrc/thrift/FrontendService.thrift b/gensrc/thrift/FrontendService.thrift
index 34f56fe2510..f639297d9a4 100644
--- a/gensrc/thrift/FrontendService.thrift
+++ b/gensrc/thrift/FrontendService.thrift
@@ -1343,6 +1343,19 @@ struct TAutoIncrementRangeResult {
4: optional Types.TNetworkAddress master_address
}
+struct TMaxComputeBlockIdRequest {
+ 1: optional i64 txn_id
+ 2: optional string write_session_id
+ 3: optional i64 length
+}
+
+struct TMaxComputeBlockIdResult {
+ 1: optional Status.TStatus status
+ 2: optional i64 start
+ 3: optional i64 length
+ 4: optional Types.TNetworkAddress master_address
+}
+
struct TCreatePartitionRequest {
1: optional i64 txn_id
2: optional i64 db_id
@@ -1956,6 +1969,7 @@ service FrontendService {
Status.TStatus updatePlanStatsCache(1: TUpdatePlanStatsCacheRequest request)
TAutoIncrementRangeResult getAutoIncrementRange(1: TAutoIncrementRangeRequest request)
+ TMaxComputeBlockIdResult getMaxComputeBlockIdRange(1: TMaxComputeBlockIdRequest request)
TCreatePartitionResult createPartition(1: TCreatePartitionRequest request)
// insert overwrite partition(*)
diff --git a/regression-test/suites/external_table_p2/maxcompute/write/test_mc_write_catalog_block_bytes.groovy b/regression-test/suites/external_table_p2/maxcompute/write/test_mc_write_catalog_block_bytes.groovy
new file mode 100644
index 00000000000..5557cae9091
--- /dev/null
+++ b/regression-test/suites/external_table_p2/maxcompute/write/test_mc_write_catalog_block_bytes.groovy
@@ -0,0 +1,167 @@
+// 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.
+
+suite("test_mc_write_catalog_block_bytes", "p2,external") {
+ String enabled = context.config.otherConfigs.get("enableMaxComputeTest")
+ if (enabled == null || !enabled.equalsIgnoreCase("true")) {
+ logger.info("disable MaxCompute test.")
+ return
+ }
+
+ String ak = context.config.otherConfigs.get("ak")
+ String sk = context.config.otherConfigs.get("sk")
+ String endpoint = "http://service.cn-beijing-vpc.maxcompute.aliyun-inc.com/api"
+ String defaultProject = "mc_doris_test_write"
+ String uuid = UUID.randomUUID().toString().replace("-", "").substring(0, 8)
+
+ int rowCount = 128
+ String internalDb = "mc_block_bytes_internal_${uuid}"
+ String internalTb = "block_bytes_src_${uuid}"
+
+ def normalizeRows = { rows ->
+ rows.collect { row ->
+ row.collect { value -> value == null ? "null" : value.toString() }
+ }
+ }
+
+ def expectedSummary = [["128", "0", "127", "2058", "2060"]]
+ def expectedDistribution = [
+ ["20250101", "32"],
+ ["20250102", "32"],
+ ["20250103", "32"],
+ ["20250104", "32"]
+ ]
+ def expectedPreview = [
+ ["0", "2058", "20250101"],
+ ["1", "2058", "20250102"],
+ ["2", "2058", "20250103"],
+ ["3", "2058", "20250104"],
+ ["4", "2058", "20250101"]
+ ]
+ def catalogConfigs = [
+ [
+ name: "test_mc_write_catalog_block_bytes_4k_${uuid}",
+ maxBlockBytes: "4",
+ db: "mc_block_bytes_4k_${uuid}",
+ tb: "block_bytes_4k_${uuid}"
+ ],
+ [
+ name: "test_mc_write_catalog_block_bytes_8k_${uuid}",
+ maxBlockBytes: "8192",
+ db: "mc_block_bytes_8k_${uuid}",
+ tb: "block_bytes_8k_${uuid}"
+ ],
+ [
+ name: "test_mc_write_catalog_block_bytes_16k_${uuid}",
+ maxBlockBytes: "163840",
+ db: "mc_block_bytes_16k_${uuid}",
+ tb: "block_bytes_16k_${uuid}"
+ ]
+ ]
+ def createdCatalogConfigs = []
+
+ try {
+ sql """CREATE DATABASE IF NOT EXISTS internal.${internalDb}"""
+ sql """DROP TABLE IF EXISTS internal.${internalDb}.${internalTb}"""
+ sql """
+ CREATE TABLE internal.${internalDb}.${internalTb} (
+ id INT,
+ payload STRING,
+ ds STRING
+ )
+ DISTRIBUTED BY HASH(id) BUCKETS 1
+ PROPERTIES ("replication_allocation" = "tag.location.default: 1")
+ """
+ sql """
+ INSERT INTO internal.${internalDb}.${internalTb}
+ SELECT
+ number AS id,
+ concat('payload_', cast(number AS STRING), '_', repeat('x', 2048)) AS payload,
+ concat('2025010', cast((number % 4 + 1) AS STRING)) AS ds
+ FROM numbers("number"="${rowCount}")
+ """
+
+ for (def catalogConfig in catalogConfigs) {
+ String catalogName = catalogConfig.name
+ String maxBlockBytes = catalogConfig.maxBlockBytes
+ String db = catalogConfig.db
+ String tb = catalogConfig.tb
+
+ sql """DROP CATALOG IF EXISTS ${catalogName}"""
+ sql """
+ CREATE CATALOG IF NOT EXISTS ${catalogName} PROPERTIES (
+ "type" = "max_compute",
+ "mc.default.project" = "${defaultProject}",
+ "mc.access_key" = "${ak}",
+ "mc.secret_key" = "${sk}",
+ "mc.endpoint" = "${endpoint}",
+ "mc.quota" = "pay-as-you-go",
+ "mc.enable.namespace.schema" = "true",
+ "mc.write_max_block_bytes" = "${maxBlockBytes}"
+ );
+ """
+ createdCatalogConfigs.add(catalogConfig)
+
+ def showCreate = sql """SHOW CREATE CATALOG ${catalogName}"""
+ assert showCreate.size() == 1
+ assert showCreate[0][1].contains("\"mc.write_max_block_bytes\" = \"${maxBlockBytes}\"")
+
+ sql """SWITCH ${catalogName}"""
+ sql """DROP DATABASE IF EXISTS ${db}"""
+ sql """CREATE DATABASE ${db}"""
+ sql """USE ${db}"""
+ sql """DROP TABLE IF EXISTS ${tb}"""
+ sql """
+ CREATE TABLE ${tb} (
+ id INT,
+ payload STRING,
+ ds STRING
+ )
+ """
+
+ sql """INSERT INTO ${tb} SELECT * FROM internal.${internalDb}.${internalTb}"""
+
+ def summary = normalizeRows(sql """
+ SELECT count(*), min(id), max(id), min(length(payload)), max(length(payload))
+ FROM ${tb}
+ """)
+ assert summary == expectedSummary : "${catalogName} summary mismatch: ${summary}"
+
+ def distribution = normalizeRows(sql """
+ SELECT ds, count(*)
+ FROM ${tb}
+ GROUP BY ds
+ ORDER BY ds
+ """)
+ assert distribution == expectedDistribution : "${catalogName} distribution mismatch: ${distribution}"
+
+ def preview = normalizeRows(sql """
+ SELECT id, length(payload), ds
+ FROM ${tb}
+ ORDER BY id
+ LIMIT 5
+ """)
+ assert preview == expectedPreview : "${catalogName} preview mismatch: ${preview}"
+ }
+ } finally {
+ for (int i = createdCatalogConfigs.size() - 1; i >= 0; i--) {
+ def catalogConfig = createdCatalogConfigs[i]
+ sql """DROP DATABASE IF EXISTS ${catalogConfig.name}.${catalogConfig.db}"""
+ sql """DROP CATALOG IF EXISTS ${catalogConfig.name}"""
+ }
+ }
+}
diff --git a/regression-test/suites/external_table_p2/maxcompute/write/test_mc_write_large_data.groovy b/regression-test/suites/external_table_p2/maxcompute/write/test_mc_write_large_data.groovy
index d45389e9ee0..6897a01125c 100644
--- a/regression-test/suites/external_table_p2/maxcompute/write/test_mc_write_large_data.groovy
+++ b/regression-test/suites/external_table_p2/maxcompute/write/test_mc_write_large_data.groovy
@@ -140,53 +140,4 @@ suite("test_mc_write_large_data", "p2,external") {
sql """DROP TABLE IF EXISTS internal.${internal_db}.${internal_tb}"""
sql """DROP DATABASE IF EXISTS internal.${internal_db}"""
}
-
- // Test: mc.max_write_batch_rows and mc.max_field_size_bytes catalog properties
- String mc_catalog_props = "test_mc_write_large_data_props"
- sql """drop catalog if exists ${mc_catalog_props}"""
- sql """
- CREATE CATALOG IF NOT EXISTS ${mc_catalog_props} PROPERTIES (
- "type" = "max_compute",
- "mc.default.project" = "${defaultProject}",
- "mc.access_key" = "${ak}",
- "mc.secret_key" = "${sk}",
- "mc.endpoint" = "http://service.cn-beijing-vpc.maxcompute.aliyun-inc.com/api",
- "mc.quota" = "pay-as-you-go",
- "mc.enable.namespace.schema" = "true",
- "mc.max_write_batch_rows" = "512",
- "mc.max_field_size_bytes" = "4194304"
- );
- """
-
- sql """switch ${mc_catalog_props}"""
- String db_props = "mc_props_test_${uuid}"
- sql """drop database if exists ${db_props}"""
- sql """create database ${db_props}"""
- sql """use ${db_props}"""
-
- try {
- String tb_props = "props_verify_${uuid}"
- sql """DROP TABLE IF EXISTS ${tb_props}"""
- sql """
- CREATE TABLE ${tb_props} (
- id INT,
- name STRING
- )
- """
-
- // Insert 2000 rows to exceed max_write_batch_rows=512 (will be split into 4 batches)
- sql """
- INSERT INTO ${tb_props}
- SELECT
- number AS id,
- concat('name_', cast(number AS STRING)) AS name
- FROM numbers("number"="2000")
- """
-
- qt_props_count """ SELECT count(*) FROM ${tb_props} """
- order_qt_props_top5 """ SELECT * FROM ${tb_props} ORDER BY id LIMIT 5 """
- } finally {
- sql """drop database if exists ${mc_catalog_props}.${db_props}"""
- sql """drop catalog if exists ${mc_catalog_props}"""
- }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org