This is an automated email from the ASF dual-hosted git repository.
gavinchou 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 a6751eb3ac6 [improvement](compaction) Support lazy chunked row ID
conversion (#67333)
a6751eb3ac6 is described below
commit a6751eb3ac6b4667c3291544c5cfb1fae48e1556
Author: meiyi <[email protected]>
AuthorDate: Fri Sep 11 12:22:30 2026 +0800
[improvement](compaction) Support lazy chunked row ID conversion (#67333)
Problem Summary:
Later, we will support parallel compaction, which means one compaction
sub task only handles somes rows of the input rowsets.
Now, Row ID conversion allocated a dense destination entry for every
source row, which wastes memory when a compaction reads only sparse
source ranges.
Add an opt-in lazy chunked mode that allocates mappings only for touched
rows.
---
be/src/storage/merger.cpp | 8 +-
be/src/storage/rowid_conversion.h | 140 +++++++++++++++++++++++-------
be/test/storage/rowid_conversion_test.cpp | 77 +++++++++++++++-
3 files changed, 189 insertions(+), 36 deletions(-)
diff --git a/be/src/storage/merger.cpp b/be/src/storage/merger.cpp
index 5348aa3af4a..97f8e12090e 100644
--- a/be/src/storage/merger.cpp
+++ b/be/src/storage/merger.cpp
@@ -139,8 +139,8 @@ Status Merger::vmerge_rowsets(BaseTabletSPtr tablet,
ReaderType reader_type,
if (reader_params.record_rowids && block.rows() > 0) {
std::vector<uint32_t> segment_num_rows;
RETURN_IF_ERROR(dst_rowset_writer->get_segment_num_rows(&segment_num_rows));
-
stats_output->rowid_conversion->add(reader.current_block_row_locations(),
- segment_num_rows);
+ RETURN_IF_ERROR(stats_output->rowid_conversion->add(
+ reader.current_block_row_locations(), segment_num_rows));
}
output_rows += block.rows();
@@ -329,8 +329,8 @@ Status Merger::vertical_compact_one_group(
if (is_key && reader_params.record_rowids && block.rows() > 0) {
std::vector<uint32_t> segment_num_rows;
RETURN_IF_ERROR(dst_rowset_writer->get_segment_num_rows(&segment_num_rows));
-
stats_output->rowid_conversion->add(reader.current_block_row_locations(),
- segment_num_rows);
+ RETURN_IF_ERROR(stats_output->rowid_conversion->add(
+ reader.current_block_row_locations(), segment_num_rows));
}
output_rows += block.rows();
block.clear_column_data();
diff --git a/be/src/storage/rowid_conversion.h
b/be/src/storage/rowid_conversion.h
index e26acd8ce86..d6fd51d7367 100644
--- a/be/src/storage/rowid_conversion.h
+++ b/be/src/storage/rowid_conversion.h
@@ -17,7 +17,10 @@
#pragma once
+#include <algorithm>
#include <map>
+#include <memory>
+#include <utility>
#include <vector>
#include "common/cast_set.h"
@@ -35,12 +38,14 @@ namespace doris {
// destination rowset.
class RowIdConversion {
public:
+ enum class Mode { DENSE, LAZY_CHUNKED };
+
struct DestinationRowId {
uint32_t segment_pos;
uint32_t row_id;
};
- RowIdConversion() = default;
+ explicit RowIdConversion(Mode mode = Mode::DENSE) : _mode(mode) {}
~RowIdConversion() { RELEASE_THREAD_MEM_TRACKER(_seg_rowid_map_mem_used); }
Status init_segment_map(const RowsetId& src_rowset_id, const
std::vector<uint32_t>& segment_ids,
@@ -49,36 +54,30 @@ public:
for (size_t i = 0; i < num_rows.size(); i++) {
auto src_segment = std::pair<RowsetId, uint32_t> {src_rowset_id,
segment_ids[i]};
auto iter = _segment_to_id_map.find(src_segment);
- // Each segment-group reader initializes all source segments, so
reuse existing maps.
+ // A segment-group reader can be reopened, so reuse existing
source-segment maps.
if (iter != _segment_to_id_map.end()) {
- DORIS_CHECK_LT(iter->second, _segments_rowid_map.size());
- DORIS_CHECK_EQ(_segments_rowid_map[iter->second].size(),
num_rows[i]);
+ DORIS_CHECK_LT(iter->second, _segment_num_rows.size());
+ DORIS_CHECK_EQ(_segment_num_rows[iter->second], num_rows[i]);
continue;
}
constexpr size_t RESERVED_MEMORY = 10 * 1024 * 1024; // 10M
- if
(doris::GlobalMemoryArbitrator::is_exceed_hard_mem_limit(RESERVED_MEMORY)) {
- return Status::MemoryLimitExceeded(fmt::format(
- "RowIdConversion init_segment_map failed, process
memory exceed limit or "
- "sys available memory less than low water mark , {}, "
- "consuming "
- "tracker:<{}>, peak used {}, current used {}.",
- doris::GlobalMemoryArbitrator::process_mem_log_str(),
- doris::thread_context()
- ->thread_mem_tracker_mgr->limiter_mem_tracker()
- ->label(),
- doris::thread_context()
- ->thread_mem_tracker_mgr->limiter_mem_tracker()
- ->peak_consumption(),
- doris::thread_context()
- ->thread_mem_tracker_mgr->limiter_mem_tracker()
- ->consumption()));
- }
+ RETURN_IF_ERROR(check_memory_limit(RESERVED_MEMORY));
- uint32_t id = cast_set<uint32_t>(_segments_rowid_map.size());
+ uint32_t id = cast_set<uint32_t>(_segment_num_rows.size());
auto insert_result = _segment_to_id_map.emplace(src_segment, id);
DORIS_CHECK(insert_result.second);
_id_to_segment_map.push_back(src_segment);
+ _segment_num_rows.push_back(num_rows[i]);
+ if (_mode == Mode::LAZY_CHUNKED) {
+ _lazy_segments_rowid_map.emplace_back();
+ auto& chunks = _lazy_segments_rowid_map.back();
+ chunks.resize((cast_set<size_t>(num_rows[i]) + ROWS_PER_CHUNK
- 1) /
+ ROWS_PER_CHUNK);
+ track_lazy_mem_usage(chunks.capacity() *
sizeof(std::unique_ptr<RowIdPair[]>));
+ continue;
+ }
+
std::vector<std::pair<uint32_t, uint32_t>> vec(
num_rows[i], std::pair<uint32_t, uint32_t>(UINT32_MAX,
UINT32_MAX));
@@ -96,8 +95,8 @@ public:
const RowsetId& get_dst_rowset_id() const { return _dst_rowst_id; }
// add row id to the map
- void add(const std::vector<RowLocation>& rss_row_ids,
- const std::vector<uint32_t>& dst_segments_num_row) {
+ Status add(const std::vector<RowLocation>& rss_row_ids,
+ const std::vector<uint32_t>& dst_segments_num_row) {
for (auto& item : rss_row_ids) {
if (item.row_id == -1) {
continue;
@@ -109,9 +108,16 @@ public:
_cur_dst_segment_pos++;
_cur_dst_segment_rowid = 0;
}
- _segments_rowid_map[id][item.row_id] =
- std::pair<uint32_t, uint32_t> {_cur_dst_segment_pos,
_cur_dst_segment_rowid++};
+ if (_mode == Mode::DENSE) {
+ _segments_rowid_map[id][item.row_id] = std::pair<uint32_t,
uint32_t> {
+ _cur_dst_segment_pos, _cur_dst_segment_rowid++};
+ continue;
+ }
+ RowIdPair* destination = nullptr;
+ RETURN_IF_ERROR(get_or_create_lazy_destination(id, item.row_id,
&destination));
+ *destination = {_cur_dst_segment_pos, _cur_dst_segment_rowid++};
}
+ return Status::OK();
}
// Get the destination segment position and row id. The physical
destination segment id is
@@ -122,11 +128,24 @@ public:
if (iter == _segment_to_id_map.end()) {
return -1;
}
- const auto& rowid_map = _segments_rowid_map[iter->second];
- if (src.row_id >= rowid_map.size()) {
+ const RowIdPair* destination = nullptr;
+ if (_mode == Mode::DENSE) {
+ const auto& rowid_map = _segments_rowid_map[iter->second];
+ if (src.row_id >= rowid_map.size()) {
+ return -1;
+ }
+ destination = &rowid_map[src.row_id];
+ } else {
+ const auto id = iter->second;
+ if (src.row_id >= _segment_num_rows[id]) {
+ return -1;
+ }
+ destination = get_lazy_destination(id, src.row_id);
+ }
+ if (destination == nullptr) {
return -1;
}
- auto& [dst_segment_pos, dst_rowid] = rowid_map[src.row_id];
+ const auto& [dst_segment_pos, dst_rowid] = *destination;
if (dst_segment_pos == UINT32_MAX && dst_rowid == UINT32_MAX) {
return -1;
}
@@ -138,10 +157,13 @@ public:
const std::vector<std::vector<std::pair<uint32_t, uint32_t>>>&
get_rowid_conversion_map()
const {
+ DORIS_CHECK(_mode == Mode::DENSE);
return _segments_rowid_map;
}
- const std::map<std::pair<RowsetId, uint32_t>, uint32_t>&
get_src_segment_to_id_map() {
+ size_t memory_usage() const { return _seg_rowid_map_mem_used; }
+
+ const std::map<std::pair<RowsetId, uint32_t>, uint32_t>&
get_src_segment_to_id_map() const {
return _segment_to_id_map;
}
@@ -155,6 +177,52 @@ public:
}
private:
+ using RowIdPair = std::pair<uint32_t, uint32_t>;
+ using LazySegmentRowIdMap = std::vector<std::unique_ptr<RowIdPair[]>>;
+ // A 4096-row chunk uses 32 KiB, balancing sparse-range waste and
allocation overhead.
+ static constexpr uint32_t ROWS_PER_CHUNK = 4096;
+
+ Status check_memory_limit(size_t reserved_memory) const {
+ if
(!doris::GlobalMemoryArbitrator::is_exceed_hard_mem_limit(reserved_memory)) {
+ return Status::OK();
+ }
+ return Status::MemoryLimitExceeded(fmt::format(
+ "RowIdConversion allocation failed, process memory exceed
limit or sys available "
+ "memory less than low water mark, {}, consuming tracker:<{}>,
peak used {}, "
+ "current used {}.",
+ doris::GlobalMemoryArbitrator::process_mem_log_str(),
+
doris::thread_context()->thread_mem_tracker_mgr->limiter_mem_tracker()->label(),
+ doris::thread_context()
+ ->thread_mem_tracker_mgr->limiter_mem_tracker()
+ ->peak_consumption(),
+ doris::thread_context()
+ ->thread_mem_tracker_mgr->limiter_mem_tracker()
+ ->consumption()));
+ }
+
+ Status get_or_create_lazy_destination(uint32_t segment_id, uint32_t row_id,
+ RowIdPair** destination) {
+ DORIS_CHECK_LT(segment_id, _segment_num_rows.size());
+ DORIS_CHECK_LT(row_id, _segment_num_rows[segment_id]);
+ auto& chunks = _lazy_segments_rowid_map[segment_id];
+ auto& chunk = chunks[row_id / ROWS_PER_CHUNK];
+ if (chunk == nullptr) {
+ constexpr size_t CHUNK_BYTES = ROWS_PER_CHUNK * sizeof(RowIdPair);
+ RETURN_IF_ERROR(check_memory_limit(CHUNK_BYTES));
+ chunk = std::make_unique<RowIdPair[]>(ROWS_PER_CHUNK);
+ std::fill_n(chunk.get(), ROWS_PER_CHUNK, RowIdPair {UINT32_MAX,
UINT32_MAX});
+ track_lazy_mem_usage(CHUNK_BYTES);
+ }
+ *destination = &chunk[row_id % ROWS_PER_CHUNK];
+ return Status::OK();
+ }
+
+ const RowIdPair* get_lazy_destination(uint32_t segment_id, uint32_t
row_id) const {
+ const auto& chunks = _lazy_segments_rowid_map[segment_id];
+ const auto& chunk = chunks[row_id / ROWS_PER_CHUNK];
+ return chunk == nullptr ? nullptr : &chunk[row_id % ROWS_PER_CHUNK];
+ }
+
void track_mem_usage(size_t delta_std_pair_cap) {
_std_pair_cap += delta_std_pair_cap;
@@ -165,14 +233,26 @@ private:
_seg_rowid_map_mem_used = new_size;
}
+ void track_lazy_mem_usage(size_t bytes) {
+ CONSUME_THREAD_MEM_TRACKER(bytes);
+ _seg_rowid_map_mem_used += bytes;
+ }
+
private:
// the first level vector: index indicates src segment.
// the second level vector: index indicates row id of source segment,
// value indicates destination segment position and row id.
// <UINT32_MAX, UINT32_MAX> indicates current row not exist.
std::vector<std::vector<std::pair<uint32_t, uint32_t>>>
_segments_rowid_map;
+ // The first-level index indicates the internal source segment id.
+ // The second-level index is source row_id / ROWS_PER_CHUNK and selects a
lazy chunk.
+ // The chunk offset is source row_id % ROWS_PER_CHUNK.
+ // The value indicates destination segment position and row id.
+ std::vector<LazySegmentRowIdMap> _lazy_segments_rowid_map;
+ std::vector<uint32_t> _segment_num_rows;
size_t _seg_rowid_map_mem_used {0};
size_t _std_pair_cap {0};
+ Mode _mode;
// Map source segment to 0 to n
std::map<std::pair<RowsetId, uint32_t>, uint32_t> _segment_to_id_map;
diff --git a/be/test/storage/rowid_conversion_test.cpp
b/be/test/storage/rowid_conversion_test.cpp
index 661381c33fa..8c5fa1ab8f6 100644
--- a/be/test/storage/rowid_conversion_test.cpp
+++ b/be/test/storage/rowid_conversion_test.cpp
@@ -534,7 +534,7 @@ TEST_F(TestRowIdConversion, Basic) {
rowid_conversion.set_dst_rowset_id(dst_rowset);
std::vector<uint32_t> dst_segment_num_rows = {4, 3, 4};
- rowid_conversion.add(rss_row_ids, dst_segment_num_rows);
+ ASSERT_TRUE(rowid_conversion.add(rss_row_ids, dst_segment_num_rows).ok());
int res = 0;
src_rowset.init(0);
@@ -601,7 +601,7 @@ TEST_F(TestRowIdConversion,
ConvertDestinationPositionToPhysicalSegmentId) {
RowIdConversion rowid_conversion;
ASSERT_TRUE(rowid_conversion.init_segment_map(input_rowset_id, {10},
{1}).ok());
rowid_conversion.set_dst_rowset_id(output_rowset_id);
- rowid_conversion.add({RowLocation(input_rowset_id, 10, 0)}, {1});
+ ASSERT_TRUE(rowid_conversion.add({RowLocation(input_rowset_id, 10, 0)},
{1}).ok());
DeleteBitmap input_delete_bitmap(1);
input_delete_bitmap.add({input_rowset_id, 10, 5}, 0);
@@ -622,6 +622,79 @@ TEST_F(TestRowIdConversion,
ConvertDestinationPositionToPhysicalSegmentId) {
EXPECT_EQ(dst.segment_id, 100);
}
+TEST_F(TestRowIdConversion, LazyChunkedOnlyAllocatesTouchedRows) {
+ constexpr uint32_t NUM_ROWS = 1'000'000;
+ RowsetId src_rowset;
+ src_rowset.init(1);
+ RowsetId dst_rowset;
+ dst_rowset.init(2);
+
+ RowIdConversion rowid_conversion(RowIdConversion::Mode::LAZY_CHUNKED);
+ ASSERT_TRUE(rowid_conversion.init_segment_map(src_rowset, {10},
{NUM_ROWS}).ok());
+ rowid_conversion.set_dst_rowset_id(dst_rowset);
+ ASSERT_TRUE(rowid_conversion
+ .add({RowLocation(src_rowset, 10, 0),
+ RowLocation(src_rowset, 10, NUM_ROWS - 1)},
+ {1, 1})
+ .ok());
+
+ RowIdConversion::DestinationRowId dst;
+ ASSERT_EQ(rowid_conversion.get(RowLocation(src_rowset, 10, 0), &dst), 0);
+ EXPECT_EQ(dst.segment_pos, 0);
+ EXPECT_EQ(dst.row_id, 0);
+ ASSERT_EQ(rowid_conversion.get(RowLocation(src_rowset, 10, NUM_ROWS - 1),
&dst), 0);
+ EXPECT_EQ(dst.segment_pos, 1);
+ EXPECT_EQ(dst.row_id, 0);
+ EXPECT_EQ(rowid_conversion.get(RowLocation(src_rowset, 10, NUM_ROWS / 2),
&dst), -1);
+ EXPECT_LT(rowid_conversion.memory_usage(), 1024 * 1024);
+}
+
+TEST_F(TestRowIdConversion, LazyChunkedHandlesMultipleSegmentsAndBoundaries) {
+ constexpr uint32_t CHUNK_SIZE = 4096;
+ RowsetId first_rowset;
+ first_rowset.init(1);
+ RowsetId second_rowset;
+ second_rowset.init(2);
+
+ RowIdConversion rowid_conversion(RowIdConversion::Mode::LAZY_CHUNKED);
+ ASSERT_TRUE(
+ rowid_conversion.init_segment_map(first_rowset, {10, 11},
{CHUNK_SIZE + 1, 2}).ok());
+ ASSERT_TRUE(rowid_conversion.init_segment_map(second_rowset, {10},
{CHUNK_SIZE * 2 + 1}).ok());
+ const size_t memory_usage = rowid_conversion.memory_usage();
+ ASSERT_TRUE(
+ rowid_conversion.init_segment_map(first_rowset, {10, 11},
{CHUNK_SIZE + 1, 2}).ok());
+ EXPECT_EQ(rowid_conversion.get_src_segment_to_id_map().size(), 3);
+ EXPECT_EQ(rowid_conversion.memory_usage(), memory_usage);
+
+ ASSERT_TRUE(rowid_conversion
+ .add({RowLocation(first_rowset, 10, CHUNK_SIZE - 1),
+ RowLocation(first_rowset, 10, CHUNK_SIZE),
+ RowLocation(first_rowset, 11, 1),
+ RowLocation(second_rowset, 10, CHUNK_SIZE * 2)},
+ {2, 2})
+ .ok());
+
+ auto expect_destination = [&](const RowsetId& rowset_id, uint32_t
segment_id, uint32_t row_id,
+ uint32_t segment_pos, uint32_t
destination_row_id) {
+ RowIdConversion::DestinationRowId destination;
+ ASSERT_EQ(rowid_conversion.get(RowLocation(rowset_id, segment_id,
row_id), &destination),
+ 0);
+ EXPECT_EQ(destination.segment_pos, segment_pos);
+ EXPECT_EQ(destination.row_id, destination_row_id);
+ };
+ expect_destination(first_rowset, 10, CHUNK_SIZE - 1, 0, 0);
+ expect_destination(first_rowset, 10, CHUNK_SIZE, 0, 1);
+ expect_destination(first_rowset, 11, 1, 1, 0);
+ expect_destination(second_rowset, 10, CHUNK_SIZE * 2, 1, 1);
+
+ RowIdConversion::DestinationRowId destination;
+ EXPECT_EQ(rowid_conversion.get(RowLocation(first_rowset, 10, 0),
&destination), -1);
+ EXPECT_EQ(rowid_conversion.get(RowLocation(second_rowset, 10, CHUNK_SIZE),
&destination), -1);
+ EXPECT_EQ(rowid_conversion.get(RowLocation(first_rowset, 10, CHUNK_SIZE +
1), &destination),
+ -1);
+ EXPECT_EQ(rowid_conversion.get(RowLocation(first_rowset, 12, 0),
&destination), -1);
+}
+
TEST_F(TestRowIdConversion,
SingleRowsetGroupedCompactionRowIdConversionIsComplete) {
constexpr int64_t num_segments = 5;
constexpr int64_t rows_per_segment = 1500;
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
