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 |