1 | #include <Storages/MergeTree/MergeTreeReadPool.h> |
2 | #include <ext/range.h> |
3 | #include <Storages/MergeTree/MergeTreeBaseSelectProcessor.h> |
4 | |
5 | |
6 | namespace ProfileEvents |
7 | { |
8 | extern const Event SlowRead; |
9 | extern const Event ReadBackoff; |
10 | } |
11 | |
12 | namespace ErrorCodes |
13 | { |
14 | extern const int LOGICAL_ERROR; |
15 | } |
16 | |
17 | namespace DB |
18 | { |
19 | |
20 | |
21 | MergeTreeReadPool::MergeTreeReadPool( |
22 | const size_t threads_, const size_t sum_marks_, const size_t min_marks_for_concurrent_read_, |
23 | RangesInDataParts parts_, const MergeTreeData & data_, const PrewhereInfoPtr & prewhere_info_, |
24 | const bool check_columns_, const Names & column_names_, |
25 | const BackoffSettings & backoff_settings_, size_t preferred_block_size_bytes_, |
26 | const bool do_not_steal_tasks_) |
27 | : backoff_settings{backoff_settings_}, backoff_state{threads_}, data{data_}, |
28 | column_names{column_names_}, do_not_steal_tasks{do_not_steal_tasks_}, |
29 | predict_block_size_bytes{preferred_block_size_bytes_ > 0}, prewhere_info{prewhere_info_}, parts_ranges{parts_} |
30 | { |
31 | /// reverse from right-to-left to left-to-right |
32 | /// because 'reverse' was done in MergeTreeDataSelectExecutor |
33 | for (auto & part_ranges : parts_ranges) |
34 | std::reverse(std::begin(part_ranges.ranges), std::end(part_ranges.ranges)); |
35 | |
36 | /// parts don't contain duplicate MergeTreeDataPart's. |
37 | const auto per_part_sum_marks = fillPerPartInfo(parts_, check_columns_); |
38 | fillPerThreadInfo(threads_, sum_marks_, per_part_sum_marks, parts_, min_marks_for_concurrent_read_); |
39 | } |
40 | |
41 | |
42 | MergeTreeReadTaskPtr MergeTreeReadPool::getTask(const size_t min_marks_to_read, const size_t thread, const Names & ordered_names) |
43 | { |
44 | const std::lock_guard lock{mutex}; |
45 | |
46 | /// If number of threads was lowered due to backoff, then will assign work only for maximum 'backoff_state.current_threads' threads. |
47 | if (thread >= backoff_state.current_threads) |
48 | return nullptr; |
49 | |
50 | if (remaining_thread_tasks.empty()) |
51 | return nullptr; |
52 | |
53 | const auto tasks_remaining_for_this_thread = !threads_tasks[thread].sum_marks_in_parts.empty(); |
54 | if (!tasks_remaining_for_this_thread && do_not_steal_tasks) |
55 | return nullptr; |
56 | |
57 | /// Steal task if nothing to do and it's not prohibited |
58 | const auto thread_idx = tasks_remaining_for_this_thread ? thread : *std::begin(remaining_thread_tasks); |
59 | auto & thread_tasks = threads_tasks[thread_idx]; |
60 | |
61 | auto & thread_task = thread_tasks.parts_and_ranges.back(); |
62 | const auto part_idx = thread_task.part_idx; |
63 | |
64 | auto & part = parts_with_idx[part_idx]; |
65 | auto & marks_in_part = thread_tasks.sum_marks_in_parts.back(); |
66 | |
67 | /// Get whole part to read if it is small enough. |
68 | auto need_marks = std::min(marks_in_part, min_marks_to_read); |
69 | |
70 | /// Do not leave too little rows in part for next time. |
71 | if (marks_in_part > need_marks && |
72 | marks_in_part - need_marks < min_marks_to_read) |
73 | need_marks = marks_in_part; |
74 | |
75 | MarkRanges ranges_to_get_from_part; |
76 | |
77 | /// Get whole part to read if it is small enough. |
78 | if (marks_in_part <= need_marks) |
79 | { |
80 | const auto marks_to_get_from_range = marks_in_part; |
81 | |
82 | /** Ranges are in right-to-left order, because 'reverse' was done in MergeTreeDataSelectExecutor |
83 | * and that order is supported in 'fillPerThreadInfo'. |
84 | */ |
85 | ranges_to_get_from_part = thread_task.ranges; |
86 | |
87 | marks_in_part -= marks_to_get_from_range; |
88 | |
89 | thread_tasks.parts_and_ranges.pop_back(); |
90 | thread_tasks.sum_marks_in_parts.pop_back(); |
91 | |
92 | if (thread_tasks.sum_marks_in_parts.empty()) |
93 | remaining_thread_tasks.erase(thread_idx); |
94 | } |
95 | else |
96 | { |
97 | /// Loop through part ranges. |
98 | while (need_marks > 0 && !thread_task.ranges.empty()) |
99 | { |
100 | auto & range = thread_task.ranges.back(); |
101 | |
102 | const size_t marks_in_range = range.end - range.begin; |
103 | const size_t marks_to_get_from_range = std::min(marks_in_range, need_marks); |
104 | |
105 | ranges_to_get_from_part.emplace_back(range.begin, range.begin + marks_to_get_from_range); |
106 | range.begin += marks_to_get_from_range; |
107 | if (range.begin == range.end) |
108 | { |
109 | std::swap(range, thread_task.ranges.back()); |
110 | thread_task.ranges.pop_back(); |
111 | } |
112 | |
113 | marks_in_part -= marks_to_get_from_range; |
114 | need_marks -= marks_to_get_from_range; |
115 | } |
116 | |
117 | /** Change order to right-to-left, for MergeTreeThreadSelectBlockInputStream to get ranges with .pop_back() |
118 | * (order was changed to left-to-right due to .pop_back() above). |
119 | */ |
120 | std::reverse(std::begin(ranges_to_get_from_part), std::end(ranges_to_get_from_part)); |
121 | } |
122 | |
123 | auto curr_task_size_predictor = !per_part_size_predictor[part_idx] ? nullptr |
124 | : std::make_unique<MergeTreeBlockSizePredictor>(*per_part_size_predictor[part_idx]); /// make a copy |
125 | |
126 | return std::make_unique<MergeTreeReadTask>( |
127 | part.data_part, ranges_to_get_from_part, part.part_index_in_query, ordered_names, |
128 | per_part_column_name_set[part_idx], per_part_columns[part_idx], per_part_pre_columns[part_idx], |
129 | prewhere_info && prewhere_info->remove_prewhere_column, per_part_should_reorder[part_idx], std::move(curr_task_size_predictor)); |
130 | } |
131 | |
132 | MarkRanges MergeTreeReadPool::getRestMarks(const MergeTreeDataPart & part, const MarkRange & from) const |
133 | { |
134 | MarkRanges all_part_ranges; |
135 | |
136 | /// Inefficient in presence of large number of data parts. |
137 | for (const auto & part_ranges : parts_ranges) |
138 | { |
139 | if (part_ranges.data_part.get() == &part) |
140 | { |
141 | all_part_ranges = part_ranges.ranges; |
142 | break; |
143 | } |
144 | } |
145 | if (all_part_ranges.empty()) |
146 | throw Exception("Trying to read marks range [" + std::to_string(from.begin) + ", " + std::to_string(from.end) + "] from part '" |
147 | + part.getFullPath() + "' which has no ranges in this query" , ErrorCodes::LOGICAL_ERROR); |
148 | |
149 | auto begin = std::lower_bound(all_part_ranges.begin(), all_part_ranges.end(), from, [] (const auto & f, const auto & s) { return f.begin < s.begin; }); |
150 | if (begin == all_part_ranges.end()) |
151 | begin = std::prev(all_part_ranges.end()); |
152 | begin->begin = from.begin; |
153 | return MarkRanges(begin, all_part_ranges.end()); |
154 | } |
155 | |
156 | Block MergeTreeReadPool::() const |
157 | { |
158 | return data.getSampleBlockForColumns(column_names); |
159 | } |
160 | |
161 | void MergeTreeReadPool::profileFeedback(const ReadBufferFromFileBase::ProfileInfo info) |
162 | { |
163 | if (backoff_settings.min_read_latency_ms == 0 || do_not_steal_tasks) |
164 | return; |
165 | |
166 | if (info.nanoseconds < backoff_settings.min_read_latency_ms * 1000000) |
167 | return; |
168 | |
169 | std::lock_guard lock(mutex); |
170 | |
171 | if (backoff_state.current_threads <= 1) |
172 | return; |
173 | |
174 | size_t throughput = info.bytes_read * 1000000000 / info.nanoseconds; |
175 | |
176 | if (throughput >= backoff_settings.max_throughput) |
177 | return; |
178 | |
179 | if (backoff_state.time_since_prev_event.elapsed() < backoff_settings.min_interval_between_events_ms * 1000000) |
180 | return; |
181 | |
182 | backoff_state.time_since_prev_event.restart(); |
183 | ++backoff_state.num_events; |
184 | |
185 | ProfileEvents::increment(ProfileEvents::SlowRead); |
186 | LOG_DEBUG(log, std::fixed << std::setprecision(3) |
187 | << "Slow read, event №" << backoff_state.num_events |
188 | << ": read " << info.bytes_read << " bytes in " << info.nanoseconds / 1000000000.0 << " sec., " |
189 | << info.bytes_read * 1000.0 / info.nanoseconds << " MB/s." ); |
190 | |
191 | if (backoff_state.num_events < backoff_settings.min_events) |
192 | return; |
193 | |
194 | backoff_state.num_events = 0; |
195 | --backoff_state.current_threads; |
196 | |
197 | ProfileEvents::increment(ProfileEvents::ReadBackoff); |
198 | LOG_DEBUG(log, "Will lower number of threads to " << backoff_state.current_threads); |
199 | } |
200 | |
201 | |
202 | std::vector<size_t> MergeTreeReadPool::fillPerPartInfo( |
203 | RangesInDataParts & parts, const bool check_columns) |
204 | { |
205 | std::vector<size_t> per_part_sum_marks; |
206 | Block sample_block = data.getSampleBlock(); |
207 | |
208 | for (const auto i : ext::range(0, parts.size())) |
209 | { |
210 | auto & part = parts[i]; |
211 | |
212 | /// Read marks for every data part. |
213 | size_t sum_marks = 0; |
214 | /// Ranges are in right-to-left order, due to 'reverse' in MergeTreeDataSelectExecutor. |
215 | for (const auto & range : part.ranges) |
216 | sum_marks += range.end - range.begin; |
217 | |
218 | per_part_sum_marks.push_back(sum_marks); |
219 | |
220 | per_part_columns_lock.emplace_back(part.data_part->columns_lock); |
221 | |
222 | auto [required_columns, required_pre_columns, should_reorder] = |
223 | getReadTaskColumns(data, part.data_part, column_names, prewhere_info, check_columns); |
224 | |
225 | /// will be used to distinguish between PREWHERE and WHERE columns when applying filter |
226 | const auto & required_column_names = required_columns.getNames(); |
227 | per_part_column_name_set.emplace_back(required_column_names.begin(), required_column_names.end()); |
228 | |
229 | per_part_pre_columns.push_back(std::move(required_pre_columns)); |
230 | per_part_columns.push_back(std::move(required_columns)); |
231 | per_part_should_reorder.push_back(should_reorder); |
232 | |
233 | parts_with_idx.push_back({ part.data_part, part.part_index_in_query }); |
234 | |
235 | if (predict_block_size_bytes) |
236 | { |
237 | per_part_size_predictor.emplace_back(std::make_unique<MergeTreeBlockSizePredictor>( |
238 | part.data_part, column_names, sample_block)); |
239 | } |
240 | else |
241 | per_part_size_predictor.emplace_back(nullptr); |
242 | } |
243 | |
244 | return per_part_sum_marks; |
245 | } |
246 | |
247 | |
248 | void MergeTreeReadPool::fillPerThreadInfo( |
249 | const size_t threads, const size_t sum_marks, std::vector<size_t> per_part_sum_marks, |
250 | RangesInDataParts & parts, const size_t min_marks_for_concurrent_read) |
251 | { |
252 | threads_tasks.resize(threads); |
253 | |
254 | const size_t min_marks_per_thread = (sum_marks - 1) / threads + 1; |
255 | |
256 | for (size_t i = 0; i < threads && !parts.empty(); ++i) |
257 | { |
258 | auto need_marks = min_marks_per_thread; |
259 | |
260 | while (need_marks > 0 && !parts.empty()) |
261 | { |
262 | const auto part_idx = parts.size() - 1; |
263 | RangesInDataPart & part = parts.back(); |
264 | size_t & marks_in_part = per_part_sum_marks.back(); |
265 | |
266 | /// Do not get too few rows from part. |
267 | if (marks_in_part >= min_marks_for_concurrent_read && |
268 | need_marks < min_marks_for_concurrent_read) |
269 | need_marks = min_marks_for_concurrent_read; |
270 | |
271 | /// Do not leave too few rows in part for next time. |
272 | if (marks_in_part > need_marks && |
273 | marks_in_part - need_marks < min_marks_for_concurrent_read) |
274 | need_marks = marks_in_part; |
275 | |
276 | MarkRanges ranges_to_get_from_part; |
277 | size_t marks_in_ranges = need_marks; |
278 | |
279 | /// Get whole part to read if it is small enough. |
280 | if (marks_in_part <= need_marks) |
281 | { |
282 | /// Leave ranges in right-to-left order for convenience to use .pop_back() in .getTask() |
283 | ranges_to_get_from_part = part.ranges; |
284 | marks_in_ranges = marks_in_part; |
285 | |
286 | need_marks -= marks_in_part; |
287 | parts.pop_back(); |
288 | per_part_sum_marks.pop_back(); |
289 | } |
290 | else |
291 | { |
292 | /// Loop through part ranges. |
293 | while (need_marks > 0) |
294 | { |
295 | if (part.ranges.empty()) |
296 | throw Exception("Unexpected end of ranges while spreading marks among threads" , ErrorCodes::LOGICAL_ERROR); |
297 | |
298 | MarkRange & range = part.ranges.back(); |
299 | |
300 | const size_t marks_in_range = range.end - range.begin; |
301 | const size_t marks_to_get_from_range = std::min(marks_in_range, need_marks); |
302 | |
303 | ranges_to_get_from_part.emplace_back(range.begin, range.begin + marks_to_get_from_range); |
304 | range.begin += marks_to_get_from_range; |
305 | marks_in_part -= marks_to_get_from_range; |
306 | need_marks -= marks_to_get_from_range; |
307 | if (range.begin == range.end) |
308 | part.ranges.pop_back(); |
309 | } |
310 | |
311 | /** Change order to right-to-left, for getTask() to get ranges with .pop_back() |
312 | * (order was changed to left-to-right due to .pop_back() above). |
313 | */ |
314 | std::reverse(std::begin(ranges_to_get_from_part), std::end(ranges_to_get_from_part)); |
315 | } |
316 | |
317 | threads_tasks[i].parts_and_ranges.push_back({ part_idx, ranges_to_get_from_part }); |
318 | threads_tasks[i].sum_marks_in_parts.push_back(marks_in_ranges); |
319 | if (marks_in_ranges != 0) |
320 | remaining_thread_tasks.insert(i); |
321 | } |
322 | } |
323 | } |
324 | |
325 | |
326 | } |
327 | |