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 2025/10/22 02:24:34 UTC
(doris) branch branch-4.0 updated: branch-4.0: [Opt](MergeIO) make read slice size is configurable #55936 (#57160)
This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.0 by this push:
new 7bc7db07f43 branch-4.0: [Opt](MergeIO) make read slice size is configurable #55936 (#57160)
7bc7db07f43 is described below
commit 7bc7db07f4384ec85796d8238c51fae45bb7559c
Author: github-actions[bot] <41...@users.noreply.github.com>
AuthorDate: Wed Oct 22 10:24:28 2025 +0800
branch-4.0: [Opt](MergeIO) make read slice size is configurable #55936 (#57160)
Cherry-picked from #55936
Co-authored-by: shee <13...@users.noreply.github.com>
Co-authored-by: garenshi <ga...@tencnet.com>
---
be/src/io/fs/buffered_reader.cpp | 4 ++--
be/src/io/fs/buffered_reader.h | 11 ++++++++++-
be/src/vec/exec/format/parquet/vparquet_reader.cpp | 13 +++++++++----
.../src/main/java/org/apache/doris/qe/SessionVariable.java | 9 ++++++++-
gensrc/thrift/PaloInternalService.thrift | 1 +
5 files changed, 30 insertions(+), 8 deletions(-)
diff --git a/be/src/io/fs/buffered_reader.cpp b/be/src/io/fs/buffered_reader.cpp
index a22b08c0659..cae08d28417 100644
--- a/be/src/io/fs/buffered_reader.cpp
+++ b/be/src/io/fs/buffered_reader.cpp
@@ -106,7 +106,7 @@ Status MergeRangeFileReader::read_at_impl(size_t offset, Slice result, size_t* b
// merge small IO
size_t merge_start = offset + has_read;
- const size_t merge_end = merge_start + READ_SLICE_SIZE;
+ const size_t merge_end = merge_start + _merged_read_slice_size;
// <slice_size, is_content>
std::vector<std::pair<size_t, bool>> merged_slice;
size_t content_size = 0;
@@ -315,7 +315,7 @@ void MergeRangeFileReader::_read_in_box(RangeCachedData& cached_data, size_t off
Status MergeRangeFileReader::_fill_box(int range_index, size_t start_offset, size_t to_read,
size_t* bytes_read, const IOContext* io_ctx) {
if (!_read_slice) {
- _read_slice = std::make_unique<OwnedSlice>(READ_SLICE_SIZE);
+ _read_slice = std::make_unique<OwnedSlice>(_merged_read_slice_size);
}
*bytes_read = 0;
diff --git a/be/src/io/fs/buffered_reader.h b/be/src/io/fs/buffered_reader.h
index 394c7eb9ad4..6bcf634aef3 100644
--- a/be/src/io/fs/buffered_reader.h
+++ b/be/src/io/fs/buffered_reader.h
@@ -278,7 +278,8 @@ public:
static constexpr size_t NUM_BOX = TOTAL_BUFFER_SIZE / BOX_SIZE; // 128
MergeRangeFileReader(RuntimeProfile* profile, io::FileReaderSPtr reader,
- const std::vector<PrefetchRange>& random_access_ranges)
+ const std::vector<PrefetchRange>& random_access_ranges,
+ int64_t merge_read_slice_size = READ_SLICE_SIZE)
: _profile(profile),
_reader(std::move(reader)),
_random_access_ranges(random_access_ranges) {
@@ -291,6 +292,13 @@ public:
// 1MB for oss, 8KB for hdfs
_equivalent_io_size =
_is_oss ? config::merged_oss_min_io_size : config::merged_hdfs_min_io_size;
+
+ _merged_read_slice_size = merge_read_slice_size;
+
+ if (_merged_read_slice_size < 0) {
+ _merged_read_slice_size = READ_SLICE_SIZE;
+ }
+
for (const PrefetchRange& range : _random_access_ranges) {
_statistics.apply_bytes += range.end_offset - range.start_offset;
}
@@ -391,6 +399,7 @@ private:
bool _is_oss;
double _max_amplified_ratio;
size_t _equivalent_io_size;
+ int64_t _merged_read_slice_size;
Statistics _statistics;
};
diff --git a/be/src/vec/exec/format/parquet/vparquet_reader.cpp b/be/src/vec/exec/format/parquet/vparquet_reader.cpp
index f48ad48d0ba..787161efc22 100644
--- a/be/src/vec/exec/format/parquet/vparquet_reader.cpp
+++ b/be/src/vec/exec/format/parquet/vparquet_reader.cpp
@@ -866,12 +866,17 @@ Status ParquetReader::_next_row_group_reader() {
size_t avg_io_size = 0;
const std::vector<io::PrefetchRange> io_ranges =
_generate_random_access_ranges(row_group_index, &avg_io_size);
+ int64_t merged_read_slice_size = -1;
+ if (_state != nullptr && _state->query_options().__isset.merge_read_slice_size) {
+ merged_read_slice_size = _state->query_options().merge_read_slice_size;
+ }
// The underlying page reader will prefetch data in column.
// Using both MergeRangeFileReader and BufferedStreamReader simultaneously would waste a lot of memory.
- group_file_reader = avg_io_size < io::MergeRangeFileReader::SMALL_IO
- ? std::make_shared<io::MergeRangeFileReader>(
- _profile, _file_reader, io_ranges)
- : _file_reader;
+ group_file_reader =
+ avg_io_size < io::MergeRangeFileReader::SMALL_IO
+ ? std::make_shared<io::MergeRangeFileReader>(
+ _profile, _file_reader, io_ranges, merged_read_slice_size)
+ : _file_reader;
}
_current_group_reader.reset(
new RowGroupReader(_io_ctx ? std::make_shared<io::TracingFileReader>(
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
index bcab9dbea77..eb32fe14fca 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
@@ -868,6 +868,8 @@ public class SessionVariable implements Serializable, Writable {
public static final String MULTI_DISTINCT_STRATEGY = "multi_distinct_strategy";
public static final String AGG_PHASE = "agg_phase";
+ public static final String MERGE_IO_READ_SLICE_SIZE = "merge_io_read_slice_size";
+
public static final String ENABLE_PREFER_CACHED_ROWSET = "enable_prefer_cached_rowset";
public static final String QUERY_FRESHNESS_TOLERANCE_MS = "query_freshness_tolerance_ms";
@@ -2586,6 +2588,11 @@ public class SessionVariable implements Serializable, Writable {
checker = "checkAggPhase")
public int aggPhase = 0;
+
+ @VariableMgr.VarAttr(name = MERGE_IO_READ_SLICE_SIZE, description = {"调整 READ_SLICE_SIZE 大小，降低 Merge IO 读放大影响",
+ "Make the READ_SLICE_SIZE variable configurable to reduce the impact caused by read amplification."})
+ public int mergeReadSliceSize = 8388608;
+
public void setAggPhase(int phase) {
aggPhase = phase;
}
@@ -4744,7 +4751,7 @@ public class SessionVariable implements Serializable, Writable {
tResult.setHnswEfSearch(hnswEFSearch);
tResult.setHnswCheckRelativeDistance(hnswCheckRelativeDistance);
tResult.setHnswBoundedQueue(hnswBoundedQueue);
-
+ tResult.setMergeReadSliceSize(mergeReadSliceSize);
return tResult;
}
diff --git a/gensrc/thrift/PaloInternalService.thrift b/gensrc/thrift/PaloInternalService.thrift
index 7f2ac6a8c1c..f9490fce799 100644
--- a/gensrc/thrift/PaloInternalService.thrift
+++ b/gensrc/thrift/PaloInternalService.thrift
@@ -406,6 +406,7 @@ struct TQueryOptions {
172: optional bool enable_prefer_cached_rowset
173: optional i64 query_freshness_tolerance_ms
+ 174: optional i64 merge_read_slice_size = 8388608;
175: optional bool enable_fuzzy_blockable_task = false;
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org