1#include <Storages/MergeTree/MergeTreeReadPool.h>
2#include <ext/range.h>
3#include <Storages/MergeTree/MergeTreeBaseSelectProcessor.h>
4
5
6namespace ProfileEvents
7{
8 extern const Event SlowRead;
9 extern const Event ReadBackoff;
10}
11
12namespace ErrorCodes
13{
14 extern const int LOGICAL_ERROR;
15}
16
17namespace DB
18{
19
20
21MergeTreeReadPool::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
42MergeTreeReadTaskPtr 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
132MarkRanges 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
156Block MergeTreeReadPool::getHeader() const
157{
158 return data.getSampleBlockForColumns(column_names);
159}
160
161void 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
202std::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
248void 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