1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
|
#include <Storages/MergeTree/MergeTreeReverseSelectProcessor.h>
#include <Storages/MergeTree/IntersectionsIndexes.h>
#include "Storages/MergeTree/MergeTreeBaseSelectProcessor.h"
namespace DB
{
namespace ErrorCodes
{
extern const int MEMORY_LIMIT_EXCEEDED;
}
bool MergeTreeReverseSelectAlgorithm::getNewTaskImpl()
try
{
if (pool)
return getNewTaskParallelReplicas();
else
return getNewTaskOrdinaryReading();
}
catch (...)
{
/// Suspicion of the broken part. A part is added to the queue for verification.
if (getCurrentExceptionCode() != ErrorCodes::MEMORY_LIMIT_EXCEEDED)
storage.reportBrokenPart(data_part);
throw;
}
bool MergeTreeReverseSelectAlgorithm::getNewTaskOrdinaryReading()
{
if (chunks.empty() && all_mark_ranges.empty())
return false;
/// We have some blocks to return in buffer.
/// Return true to continue reading, but actually don't create a task.
if (all_mark_ranges.empty())
return true;
if (!reader)
initializeReaders();
/// Read ranges from right to left.
MarkRanges mark_ranges_for_task{std::move(all_mark_ranges.back())};
all_mark_ranges.pop_back();
auto size_predictor = (preferred_block_size_bytes == 0) ? nullptr
: getSizePredictor(data_part, task_columns, sample_block);
task = std::make_unique<MergeTreeReadTask>(
data_part,
alter_conversions,
mark_ranges_for_task,
part_index_in_query,
column_name_set,
task_columns,
std::move(size_predictor));
return true;
}
bool MergeTreeReverseSelectAlgorithm::getNewTaskParallelReplicas()
{
if (chunks.empty() && no_more_tasks)
return false;
/// We have some blocks to return in buffer.
/// Return true to continue reading, but actually don't create a task.
if (no_more_tasks)
return true;
if (!reader)
initializeReaders();
auto description = RangesInDataPartDescription{
.info = data_part->info,
/// We just ignore all the distribution done before
/// Everything will be done on coordinator side
.ranges = {},
};
auto mark_ranges_for_task = pool->getNewTask(description);
if (mark_ranges_for_task.empty())
{
/// If we have chunks in buffer - return true to continue reading from them
return !chunks.empty();
}
auto size_predictor = (preferred_block_size_bytes == 0) ? nullptr
: getSizePredictor(data_part, task_columns, sample_block);
task = std::make_unique<MergeTreeReadTask>(
data_part,
alter_conversions,
mark_ranges_for_task,
part_index_in_query,
column_name_set,
task_columns,
std::move(size_predictor));
return true;
}
MergeTreeReverseSelectAlgorithm::BlockAndProgress MergeTreeReverseSelectAlgorithm::readFromPart()
{
BlockAndProgress res;
if (!chunks.empty())
{
res = std::move(chunks.back());
chunks.pop_back();
return res;
}
if (!task->range_reader.isInitialized())
initializeRangeReaders(*task);
while (!task->isFinished())
{
auto chunk = readFromPartImpl();
chunks.push_back(std::move(chunk));
}
if (chunks.empty())
return {};
res = std::move(chunks.back());
chunks.pop_back();
return res;
}
}
|
