This is an automated email from the ASF dual-hosted git repository.
liaoxin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 91d2bc2566c [opt](memory) limit max block bytes per batch in the write
path (#61821)
91d2bc2566c is described below
commit 91d2bc2566c1d5b3620327168fe6ba263a799117
Author: hui lai <[email protected]>
AuthorDate: Thu Apr 9 11:33:48 2026 +0800
[opt](memory) limit max block bytes per batch in the write path (#61821)
## Problem
During import/load, file format readers accumulate rows into a block
until `batch_size`
(row count) is reached. When individual rows are large (e.g., wide
schemas, complex nested
types), a single batch can exceed hundreds of MBs, causing excessive
memory pressure.
## Solution
Add a new BE config `load_reader_max_block_bytes` (default **64MB**, set
to `0` to disable)
and enforce it across all load file format readers:
| Reader | Mechanism |
|---|---|
| `CsvReader` | Track cumulative line bytes in the read loop; break
early when limit reached |
| `NewJsonReader` | Check `block->bytes()` in the loop guard before
reading each JSON object |
| `ParquetReader` | After each `next_batch()`, adaptively reduce
`_batch_size` based on observed bytes/row |
| `OrcReader` | Same adaptive reduction + recreate `_batch` with the new
size for subsequent reads |
## Config
// be.conf
load_reader_max_block_bytes=67108864 # 64MB (default)
load_reader_max_block_bytes=0 # disable limit
## Memory Test
TVF load 5TB data with big block.
before: It was canceled as soon as it was loaed
<img width="1398" height="990" alt="image"
src="https://github.com/user-attachments/assets/bdd67fd5-c3af-4ca1-97fa-1efff9053c60";
/>
after: Stable load with over 20G of memory
<img width="1410" height="912" alt="image"
src="https://github.com/user-attachments/assets/0f647690-8fea-407c-96a1-1dff2f5ef033";
/>
---
be/src/common/config.cpp | 4 ++++
be/src/common/config.h | 4 ++++
be/src/exec/scan/scanner.cpp | 11 +++++++++++
be/src/format/csv/csv_reader.cpp | 14 ++++++++++++--
be/src/format/json/new_json_reader.cpp | 7 ++++++-
be/src/format/orc/vorc_reader.cpp | 23 +++++++++++++++++++++++
be/src/format/orc/vorc_reader.h | 4 ++++
be/src/format/parquet/vparquet_reader.cpp | 18 ++++++++++++++++++
be/src/format/parquet/vparquet_reader.h | 4 ++++
9 files changed, 86 insertions(+), 3 deletions(-)
diff --git a/be/src/common/config.cpp b/be/src/common/config.cpp
index 1a7ac0cea48..1d7def9c66f 100644
--- a/be/src/common/config.cpp
+++ b/be/src/common/config.cpp
@@ -1727,6 +1727,10 @@
DEFINE_mBool(enable_auto_clone_on_mow_publish_missing_version, "false");
// The maximum csv line reader output buffer size
DEFINE_mInt64(max_csv_line_reader_output_buffer_size, "4294967296");
+// The maximum bytes of a single block returned by load file readers
(CsvReader, NewJsonReader,
+// ParquetReader, OrcReader). Default is 64MB. Set to 0 to disable the limit.
+DEFINE_mInt64(load_reader_max_block_bytes, "67108864");
+
// Maximum number of OpenMP threads allowed for concurrent vector index builds.
// -1 means auto: use 80% of the available CPU cores.
DEFINE_Int32(omp_threads_limit, "-1");
diff --git a/be/src/common/config.h b/be/src/common/config.h
index c5430a3ed0a..417212ed724 100644
--- a/be/src/common/config.h
+++ b/be/src/common/config.h
@@ -1785,6 +1785,10 @@ DECLARE_String(fuzzy_test_type);
// The maximum csv line reader output buffer size
DECLARE_mInt64(max_csv_line_reader_output_buffer_size);
+// The maximum bytes of a single block returned by load file readers
(CsvReader, NewJsonReader,
+// ParquetReader, OrcReader). Default is 200MB. Set to 0 to disable the limit.
+DECLARE_mInt64(load_reader_max_block_bytes);
+
// Maximum number of OpenMP threads available for concurrent index builds.
// -1 means auto: use 80% of detected CPU cores.
DECLARE_Int32(omp_threads_limit);
diff --git a/be/src/exec/scan/scanner.cpp b/be/src/exec/scan/scanner.cpp
index d60ed9b5017..fb6507c33ac 100644
--- a/be/src/exec/scan/scanner.cpp
+++ b/be/src/exec/scan/scanner.cpp
@@ -87,7 +87,18 @@ Status Scanner::get_block_after_projects(RuntimeState*
state, Block* block, bool
} else {
_origin_block.clear_column_data(row_descriptor.num_materialized_slots());
const auto min_batch_size = std::max(state->batch_size() / 2, 1);
+ // For LOAD queries, use load_reader_max_block_bytes as the
byte-based
+ // upper bound for block accumulation, so small blocks can still be
+ // merged to a reasonable size without exceeding the memory limit.
+ const int64_t load_max_bytes = (state->query_type() ==
TQueryType::LOAD &&
+
config::load_reader_max_block_bytes > 0)
+ ?
config::load_reader_max_block_bytes
+ : 0;
while (_padding_block.rows() < min_batch_size && !*eos) {
+ if (load_max_bytes > 0 &&
+ static_cast<int64_t>(_padding_block.bytes()) >=
load_max_bytes) {
+ break;
+ }
RETURN_IF_ERROR(get_block(state, &_origin_block, eos));
if (_origin_block.rows() >= min_batch_size) {
break;
diff --git a/be/src/format/csv/csv_reader.cpp b/be/src/format/csv/csv_reader.cpp
index 63045f8bfeb..b3aed7d9468 100644
--- a/be/src/format/csv/csv_reader.cpp
+++ b/be/src/format/csv/csv_reader.cpp
@@ -31,6 +31,7 @@
#include <utility>
#include "common/compiler_util.h" // IWYU pragma: keep
+#include "common/config.h"
#include "common/consts.h"
#include "common/status.h"
#include "core/block/block.h"
@@ -315,12 +316,18 @@ Status CsvReader::get_next_block(Block* block, size_t*
read_rows, bool* eof) {
}
const int batch_size = std::max(_state->batch_size(),
(int)_MIN_BATCH_SIZE);
+ const int64_t max_block_bytes =
+ (_state->query_type() == TQueryType::LOAD &&
config::load_reader_max_block_bytes > 0)
+ ? config::load_reader_max_block_bytes
+ : 0;
size_t rows = 0;
+ size_t block_bytes = 0;
bool success = false;
bool is_remove_bom = false;
if (_push_down_agg_type == TPushAggOp::type::COUNT) {
- while (rows < batch_size && !_line_reader_eof) {
+ while (rows < batch_size && !_line_reader_eof &&
+ (max_block_bytes <= 0 || (int64_t)block_bytes <
max_block_bytes)) {
const uint8_t* ptr = nullptr;
size_t size = 0;
RETURN_IF_ERROR(_line_reader->read_line(&ptr, &size,
&_line_reader_eof, _io_ctx));
@@ -348,6 +355,7 @@ Status CsvReader::get_next_block(Block* block, size_t*
read_rows, bool* eof) {
RETURN_IF_ERROR(_validate_line(Slice(ptr, size), &success));
++rows;
+ block_bytes += size;
}
auto mutate_columns = block->mutate_columns();
for (auto& col : mutate_columns) {
@@ -356,7 +364,8 @@ Status CsvReader::get_next_block(Block* block, size_t*
read_rows, bool* eof) {
block->set_columns(std::move(mutate_columns));
} else {
auto columns = block->mutate_columns();
- while (rows < batch_size && !_line_reader_eof) {
+ while (rows < batch_size && !_line_reader_eof &&
+ (max_block_bytes <= 0 || (int64_t)block_bytes <
max_block_bytes)) {
const uint8_t* ptr = nullptr;
size_t size = 0;
RETURN_IF_ERROR(_line_reader->read_line(&ptr, &size,
&_line_reader_eof, _io_ctx));
@@ -387,6 +396,7 @@ Status CsvReader::get_next_block(Block* block, size_t*
read_rows, bool* eof) {
continue;
}
RETURN_IF_ERROR(_fill_dest_columns(Slice(ptr, size), block,
columns, &rows));
+ block_bytes += size;
}
block->set_columns(std::move(columns));
}
diff --git a/be/src/format/json/new_json_reader.cpp
b/be/src/format/json/new_json_reader.cpp
index cecfcf3f0dc..f889a962c1c 100644
--- a/be/src/format/json/new_json_reader.cpp
+++ b/be/src/format/json/new_json_reader.cpp
@@ -204,8 +204,13 @@ Status NewJsonReader::get_next_block(Block* block, size_t*
read_rows, bool* eof)
}
const int batch_size = std::max(_state->batch_size(),
(int)_MIN_BATCH_SIZE);
+ const int64_t max_block_bytes =
+ (_state->query_type() == TQueryType::LOAD &&
config::load_reader_max_block_bytes > 0)
+ ? config::load_reader_max_block_bytes
+ : 0;
- while (block->rows() < batch_size && !_reader_eof) {
+ while (block->rows() < batch_size && !_reader_eof &&
+ (max_block_bytes <= 0 || (int64_t)block->bytes() <
max_block_bytes)) {
if (UNLIKELY(_read_json_by_line && _skip_first_line)) {
size_t size = 0;
const uint8_t* line_ptr = nullptr;
diff --git a/be/src/format/orc/vorc_reader.cpp
b/be/src/format/orc/vorc_reader.cpp
index a1a46d85652..782783b9ac7 100644
--- a/be/src/format/orc/vorc_reader.cpp
+++ b/be/src/format/orc/vorc_reader.cpp
@@ -48,6 +48,7 @@
#include "absl/strings/substitute.h"
#include "cctz/civil_time.h"
#include "cctz/time_zone.h"
+#include "common/config.h"
#include "common/consts.h"
#include "common/exception.h"
#include "core/block/block.h"
@@ -2435,6 +2436,15 @@ Status OrcReader::_get_next_block_impl(Block* block,
size_t* read_rows, bool* eo
*read_rows = 0;
return Status::OK();
}
+
+ // Limit memory per batch for load paths: pre-shrink _batch_size using the
bytes-per-row
+ // estimate from the previous batch so the current batch stays within
load_reader_max_block_bytes
+ // (effective from call 2 onward; first batch is capped on the next call).
+ const int64_t max_block_bytes =
+ (_state != nullptr && _state->query_type() == TQueryType::LOAD &&
+ config::load_reader_max_block_bytes > 0)
+ ? config::load_reader_max_block_bytes
+ : 0;
if (_push_down_agg_type == TPushAggOp::type::COUNT) {
auto rows = std::min(get_remaining_rows(), (int64_t)_batch_size);
@@ -2451,6 +2461,15 @@ Status OrcReader::_get_next_block_impl(Block* block,
size_t* read_rows, bool* eo
return Status::OK();
}
+ if (max_block_bytes > 0 && _load_bytes_per_row > 0 && _row_reader) {
+ size_t new_batch_size = std::max(
+ (size_t)1, (size_t)((int64_t)max_block_bytes /
(int64_t)_load_bytes_per_row));
+ if (new_batch_size != _batch_size) {
+ _batch_size = new_batch_size;
+ _batch = _row_reader->createRowBatch(_batch_size);
+ }
+ }
+
if (!_seek_to_read_one_line()) {
*eof = true;
return Status::OK();
@@ -2792,6 +2811,10 @@ Status OrcReader::_get_next_block_impl(Block* block,
size_t* read_rows, bool* eo
#endif
*read_rows = block->rows();
}
+
+ if (max_block_bytes > 0 && *read_rows > 0) {
+ _load_bytes_per_row = block->bytes() / *read_rows;
+ }
return Status::OK();
}
diff --git a/be/src/format/orc/vorc_reader.h b/be/src/format/orc/vorc_reader.h
index 4d7a73d19f6..86e6611c934 100644
--- a/be/src/format/orc/vorc_reader.h
+++ b/be/src/format/orc/vorc_reader.h
@@ -692,6 +692,10 @@ private:
io::FileSystemProperties _system_properties;
io::FileDescription _file_description;
size_t _batch_size;
+ // Bytes-per-row estimate from the previous batch, used to pre-shrink
_batch_size
+ // before reading so that oversized blocks are prevented from the current
call onward.
+ // Zero means no prior data (first batch).
+ size_t _load_bytes_per_row = 0;
int64_t _range_start_offset;
int64_t _range_size;
std::string _ctz;
diff --git a/be/src/format/parquet/vparquet_reader.cpp
b/be/src/format/parquet/vparquet_reader.cpp
index 6de215608f7..106b6124282 100644
--- a/be/src/format/parquet/vparquet_reader.cpp
+++ b/be/src/format/parquet/vparquet_reader.cpp
@@ -26,6 +26,7 @@
#include <functional>
#include <utility>
+#include "common/config.h"
#include "common/status.h"
#include "core/block/block.h"
#include "core/block/column_with_type_and_name.h"
@@ -719,6 +720,19 @@ Status ParquetReader::get_next_block(Block* block, size_t*
read_rows, bool* eof)
return Status::OK();
}
+ // Limit memory per batch for load paths.
+ // _load_bytes_per_row is updated after each batch so the *next* call
pre-shrinks _batch_size
+ // before reading, ensuring the current batch is already within the limit
(from call 2 onward).
+ const int64_t max_block_bytes =
+ (_state != nullptr && _state->query_type() == TQueryType::LOAD &&
+ config::load_reader_max_block_bytes > 0)
+ ? config::load_reader_max_block_bytes
+ : 0;
+ if (max_block_bytes > 0 && _load_bytes_per_row > 0) {
+ _batch_size = std::max((size_t)1,
+ (size_t)((int64_t)max_block_bytes /
(int64_t)_load_bytes_per_row));
+ }
+
SCOPED_RAW_TIMER(&_reader_statistics.column_read_time);
Status batch_st =
_current_group_reader->next_batch(block, _batch_size, read_rows,
&_row_group_eof);
@@ -735,6 +749,10 @@ Status ParquetReader::get_next_block(Block* block, size_t*
read_rows, bool* eof)
batch_st.to_string());
}
+ if (max_block_bytes > 0 && *read_rows > 0) {
+ _load_bytes_per_row = block->bytes() / *read_rows;
+ }
+
if (_row_group_eof) {
auto column_st = _current_group_reader->merged_column_statistics();
_column_statistics.merge(column_st);
diff --git a/be/src/format/parquet/vparquet_reader.h
b/be/src/format/parquet/vparquet_reader.h
index 5172c8efdb9..bbdbcf23d82 100644
--- a/be/src/format/parquet/vparquet_reader.h
+++ b/be/src/format/parquet/vparquet_reader.h
@@ -341,6 +341,10 @@ private:
// parquet file reader object
size_t _batch_size;
+ // Bytes-per-row estimate from the previous batch, used to pre-shrink
_batch_size
+ // before reading so that oversized blocks are prevented from the current
call onward.
+ // Zero means no prior data (first batch).
+ size_t _load_bytes_per_row = 0;
int64_t _range_start_offset;
int64_t _range_size;
const cctz::time_zone* _ctz = nullptr;
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
