Coverage Report

Created: 2026-10-21 13:42

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
be/src/storage/rowid_conversion.h
Line
Count
Source
1
// Licensed to the Apache Software Foundation (ASF) under one
2
// or more contributor license agreements.  See the NOTICE file
3
// distributed with this work for additional information
4
// regarding copyright ownership.  The ASF licenses this file
5
// to you under the Apache License, Version 2.0 (the
6
// "License"); you may not use this file except in compliance
7
// with the License.  You may obtain a copy of the License at
8
//
9
//   http://www.apache.org/licenses/LICENSE-2.0
10
//
11
// Unless required by applicable law or agreed to in writing,
12
// software distributed under the License is distributed on an
13
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14
// KIND, either express or implied.  See the License for the
15
// specific language governing permissions and limitations
16
// under the License.
17
18
#pragma once
19
20
#include <algorithm>
21
#include <map>
22
#include <memory>
23
#include <set>
24
#include <utility>
25
#include <vector>
26
27
#include "common/cast_set.h"
28
#include "common/check.h"
29
#include "core/custom_allocator.h"
30
#include "runtime/thread_context.h"
31
#include "storage/olap_common.h"
32
#include "storage/utils.h"
33
34
namespace doris {
35
36
// For unique key merge on write table, we should update delete bitmap
37
// of destination rowset when compaction finished.
38
// Through the row id correspondence between the source rowset and the
39
// destination rowset, we can quickly update the delete bitmap of the
40
// destination rowset.
41
class RowIdConversion {
42
public:
43
    enum class Mode { DENSE, LAZY_CHUNKED };
44
    using VisibilityFilteredRows =
45
            std::set<RowLocation, std::less<>, CustomStdAllocator<RowLocation>>;
46
47
    struct DestinationRowId {
48
        uint32_t segment_pos;
49
        uint32_t row_id;
50
    };
51
52
356k
    explicit RowIdConversion(Mode mode = Mode::DENSE) : _mode(mode) {}
53
356k
    ~RowIdConversion() { RELEASE_THREAD_MEM_TRACKER(_seg_rowid_map_mem_used); }
54
55
    Status init_segment_map(const RowsetId& src_rowset_id, const std::vector<uint32_t>& segment_ids,
56
56.2k
                            const std::vector<uint32_t>& num_rows) {
57
56.2k
        DCHECK_EQ(segment_ids.size(), num_rows.size());
58
82.7k
        for (size_t i = 0; i < num_rows.size(); i++) {
59
26.6k
            auto src_segment = std::pair<RowsetId, uint32_t> {src_rowset_id, segment_ids[i]};
60
26.6k
            auto iter = _segment_to_id_map.find(src_segment);
61
            // A segment-group reader can be reopened, so reuse existing source-segment maps.
62
26.6k
            if (iter != _segment_to_id_map.end()) {
63
22
                DORIS_CHECK_LT(iter->second, _segment_num_rows.size());
64
22
                DORIS_CHECK_EQ(_segment_num_rows[iter->second], num_rows[i]);
65
22
                continue;
66
22
            }
67
68
26.6k
            constexpr size_t RESERVED_MEMORY = 10 * 1024 * 1024; // 10M
69
26.6k
            RETURN_IF_ERROR(check_memory_limit(RESERVED_MEMORY));
70
71
26.4k
            uint32_t id = cast_set<uint32_t>(_segment_num_rows.size());
72
26.4k
            auto insert_result = _segment_to_id_map.emplace(src_segment, id);
73
26.4k
            DORIS_CHECK(insert_result.second);
74
26.4k
            _id_to_segment_map.push_back(src_segment);
75
26.4k
            _segment_num_rows.push_back(num_rows[i]);
76
26.4k
            if (_mode == Mode::LAZY_CHUNKED) {
77
5
                _lazy_segments_rowid_map.emplace_back();
78
5
                auto& chunks = _lazy_segments_rowid_map.back();
79
5
                chunks.resize((cast_set<size_t>(num_rows[i]) + ROWS_PER_CHUNK - 1) /
80
5
                              ROWS_PER_CHUNK);
81
5
                track_lazy_mem_usage(chunks.capacity() * sizeof(std::unique_ptr<RowIdPair[]>));
82
5
                continue;
83
5
            }
84
85
26.4k
            std::vector<std::pair<uint32_t, uint32_t>> vec(
86
26.4k
                    num_rows[i], std::pair<uint32_t, uint32_t>(UINT32_MAX, UINT32_MAX));
87
88
            //NOTE: manually count _segments_rowid_map's memory here, because _segments_rowid_map could be used by indexCompaction.
89
            // indexCompaction is a thridparty code, it's too complex to modify it.
90
            // refer compact_column.
91
26.4k
            track_mem_usage(vec.capacity());
92
26.4k
            _segments_rowid_map.emplace_back(std::move(vec));
93
26.4k
        }
94
56.0k
        return Status::OK();
95
56.2k
    }
96
97
    // set dst rowset id
98
5.91k
    void set_dst_rowset_id(const RowsetId& dst_rowset_id) { _dst_rowst_id = dst_rowset_id; }
99
10.8k
    const RowsetId& get_dst_rowset_id() const { return _dst_rowst_id; }
100
101
    // Debug correctness checks distinguish rows discarded by visibility filtering from
102
    // missing mappings. Such rows can acquire delete-bitmap entries during concurrent loads.
103
122k
    void enable_visibility_filter_tracking() {
104
122k
        _visibility_filtered_rows = std::make_unique<VisibilityFilteredRows>();
105
122k
    }
106
1.90k
    VisibilityFilteredRows* visibility_filtered_rows() { return _visibility_filtered_rows.get(); }
107
660
    bool is_visibility_filtered(const RowLocation& src) const {
108
660
        return _visibility_filtered_rows != nullptr && _visibility_filtered_rows->contains(src);
109
660
    }
110
111
    // add row id to the map
112
    Status add(const std::vector<RowLocation>& rss_row_ids,
113
6.47k
               const std::vector<uint32_t>& dst_segments_num_row) {
114
6.48M
        for (auto& item : rss_row_ids) {
115
6.48M
            if (item.row_id == -1) {
116
0
                continue;
117
0
            }
118
6.48M
            uint32_t id = _segment_to_id_map.at(
119
6.48M
                    std::pair<RowsetId, uint32_t> {item.rowset_id, item.segment_id});
120
6.48M
            if (_cur_dst_segment_pos < dst_segments_num_row.size() &&
121
6.48M
                _cur_dst_segment_rowid >= dst_segments_num_row[_cur_dst_segment_pos]) {
122
1.15k
                _cur_dst_segment_pos++;
123
1.15k
                _cur_dst_segment_rowid = 0;
124
1.15k
            }
125
6.48M
            if (_mode == Mode::DENSE) {
126
6.48M
                _segments_rowid_map[id][item.row_id] = std::pair<uint32_t, uint32_t> {
127
6.48M
                        _cur_dst_segment_pos, _cur_dst_segment_rowid++};
128
6.48M
                continue;
129
6.48M
            }
130
18.4E
            RowIdPair* destination = nullptr;
131
18.4E
            RETURN_IF_ERROR(get_or_create_lazy_destination(id, item.row_id, &destination));
132
18.4E
            *destination = {_cur_dst_segment_pos, _cur_dst_segment_rowid++};
133
18.4E
        }
134
6.47k
        return Status::OK();
135
6.47k
    }
136
137
    // Get the destination segment position and row id. The physical destination segment id is
138
    // resolved only after the output rowset is built.
139
    // return non-zero if the src RowLocation does not exist
140
5.71M
    int get(const RowLocation& src, DestinationRowId* dst) const {
141
5.71M
        auto iter = _segment_to_id_map.find({src.rowset_id, src.segment_id});
142
5.71M
        if (iter == _segment_to_id_map.end()) {
143
2
            return -1;
144
2
        }
145
5.71M
        const RowIdPair* destination = nullptr;
146
5.71M
        if (_mode == Mode::DENSE) {
147
5.71M
            const auto& rowid_map = _segments_rowid_map[iter->second];
148
5.71M
            if (src.row_id >= rowid_map.size()) {
149
1
                return -1;
150
1
            }
151
5.71M
            destination = &rowid_map[src.row_id];
152
5.71M
        } else {
153
12
            const auto id = iter->second;
154
12
            if (src.row_id >= _segment_num_rows[id]) {
155
1
                return -1;
156
1
            }
157
11
            destination = get_lazy_destination(id, src.row_id);
158
11
        }
159
5.71M
        if (destination == nullptr) {
160
2
            return -1;
161
2
        }
162
5.71M
        const auto& [dst_segment_pos, dst_rowid] = *destination;
163
5.71M
        if (dst_segment_pos == UINT32_MAX && dst_rowid == UINT32_MAX) {
164
4.54M
            return -1;
165
4.54M
        }
166
167
1.16M
        dst->segment_pos = dst_segment_pos;
168
1.16M
        dst->row_id = dst_rowid;
169
1.16M
        return 0;
170
5.71M
    }
171
172
    const std::vector<std::vector<std::pair<uint32_t, uint32_t>>>& get_rowid_conversion_map()
173
202
            const {
174
202
        DORIS_CHECK(_mode == Mode::DENSE);
175
202
        return _segments_rowid_map;
176
202
    }
177
178
    size_t memory_usage() const { return _seg_rowid_map_mem_used; }
179
180
203
    const std::map<std::pair<RowsetId, uint32_t>, uint32_t>& get_src_segment_to_id_map() const {
181
203
        return _segment_to_id_map;
182
203
    }
183
184
0
    std::pair<RowsetId, uint32_t> get_segment_by_id(uint32_t id) const {
185
0
        DCHECK_GT(_id_to_segment_map.size(), id);
186
0
        return _id_to_segment_map.at(id);
187
0
    }
188
189
0
    uint32_t get_id_by_segment(const std::pair<RowsetId, uint32_t>& segment) const {
190
0
        return _segment_to_id_map.at(segment);
191
0
    }
192
193
private:
194
    using RowIdPair = std::pair<uint32_t, uint32_t>;
195
    using LazySegmentRowIdMap = std::vector<std::unique_ptr<RowIdPair[]>>;
196
    // A 4096-row chunk uses 32 KiB, balancing sparse-range waste and allocation overhead.
197
    static constexpr uint32_t ROWS_PER_CHUNK = 4096;
198
199
26.6k
    Status check_memory_limit(size_t reserved_memory) const {
200
26.6k
        if (!doris::GlobalMemoryArbitrator::is_exceed_hard_mem_limit(reserved_memory)) {
201
26.4k
            return Status::OK();
202
26.4k
        }
203
176
        return Status::MemoryLimitExceeded(fmt::format(
204
176
                "RowIdConversion allocation failed, process memory exceed limit or sys available "
205
176
                "memory less than low water mark, {}, consuming tracker:<{}>, peak used {}, "
206
176
                "current used {}.",
207
176
                doris::GlobalMemoryArbitrator::process_mem_log_str(),
208
176
                doris::thread_context()->thread_mem_tracker_mgr->limiter_mem_tracker()->label(),
209
176
                doris::thread_context()
210
176
                        ->thread_mem_tracker_mgr->limiter_mem_tracker()
211
176
                        ->peak_consumption(),
212
176
                doris::thread_context()
213
176
                        ->thread_mem_tracker_mgr->limiter_mem_tracker()
214
176
                        ->consumption()));
215
26.6k
    }
216
217
    Status get_or_create_lazy_destination(uint32_t segment_id, uint32_t row_id,
218
7
                                          RowIdPair** destination) {
219
7
        DORIS_CHECK_LT(segment_id, _segment_num_rows.size());
220
7
        DORIS_CHECK_LT(row_id, _segment_num_rows[segment_id]);
221
7
        auto& chunks = _lazy_segments_rowid_map[segment_id];
222
7
        auto& chunk = chunks[row_id / ROWS_PER_CHUNK];
223
7
        if (chunk == nullptr) {
224
7
            constexpr size_t CHUNK_BYTES = ROWS_PER_CHUNK * sizeof(RowIdPair);
225
7
            RETURN_IF_ERROR(check_memory_limit(CHUNK_BYTES));
226
7
            chunk = std::make_unique<RowIdPair[]>(ROWS_PER_CHUNK);
227
7
            std::fill_n(chunk.get(), ROWS_PER_CHUNK, RowIdPair {UINT32_MAX, UINT32_MAX});
228
7
            track_lazy_mem_usage(CHUNK_BYTES);
229
7
        }
230
7
        *destination = &chunk[row_id % ROWS_PER_CHUNK];
231
7
        return Status::OK();
232
7
    }
233
234
12
    const RowIdPair* get_lazy_destination(uint32_t segment_id, uint32_t row_id) const {
235
12
        const auto& chunks = _lazy_segments_rowid_map[segment_id];
236
12
        const auto& chunk = chunks[row_id / ROWS_PER_CHUNK];
237
12
        return chunk == nullptr ? nullptr : &chunk[row_id % ROWS_PER_CHUNK];
238
12
    }
239
240
26.4k
    void track_mem_usage(size_t delta_std_pair_cap) {
241
26.4k
        _std_pair_cap += delta_std_pair_cap;
242
243
26.4k
        size_t new_size =
244
26.4k
                _std_pair_cap * sizeof(std::pair<uint32_t, uint32_t>) +
245
26.4k
                _segments_rowid_map.capacity() * sizeof(std::vector<std::pair<uint32_t, uint32_t>>);
246
26.4k
        CONSUME_THREAD_MEM_TRACKER(new_size - _seg_rowid_map_mem_used);
247
0
        _seg_rowid_map_mem_used = new_size;
248
26.4k
    }
249
250
12
    void track_lazy_mem_usage(size_t bytes) {
251
12
        CONSUME_THREAD_MEM_TRACKER(bytes);
252
0
        _seg_rowid_map_mem_used += bytes;
253
12
    }
254
255
private:
256
    // the first level vector: index indicates src segment.
257
    // the second level vector: index indicates row id of source segment,
258
    // value indicates destination segment position and row id.
259
    // <UINT32_MAX, UINT32_MAX> indicates current row not exist.
260
    std::vector<std::vector<std::pair<uint32_t, uint32_t>>> _segments_rowid_map;
261
    // The first-level index indicates the internal source segment id.
262
    // The second-level index is source row_id / ROWS_PER_CHUNK and selects a lazy chunk.
263
    // The chunk offset is source row_id % ROWS_PER_CHUNK.
264
    // The value indicates destination segment position and row id.
265
    std::vector<LazySegmentRowIdMap> _lazy_segments_rowid_map;
266
    std::vector<uint32_t> _segment_num_rows;
267
    size_t _seg_rowid_map_mem_used {0};
268
    size_t _std_pair_cap {0};
269
    Mode _mode;
270
    std::unique_ptr<VisibilityFilteredRows> _visibility_filtered_rows;
271
272
    // Map source segment to 0 to n
273
    std::map<std::pair<RowsetId, uint32_t>, uint32_t> _segment_to_id_map;
274
275
    // Map 0 to n to source segment
276
    std::vector<std::pair<RowsetId, uint32_t>> _id_to_segment_map;
277
278
    // dst rowset id
279
    RowsetId _dst_rowst_id;
280
281
    // current dst segment position
282
    std::uint32_t _cur_dst_segment_pos = 0;
283
284
    // current rowid of dst segment
285
    std::uint32_t _cur_dst_segment_rowid = 0;
286
};
287
288
} // namespace doris