blob: f4c698d76d7a795767d545e0ac4c7597b8f2d9a4 (
plain) (
blame)
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
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
|
#include <Storages/MergeTree/TTLMergeSelector.h>
#include <Storages/MergeTree/MergeTreeData.h>
#include <Parsers/queryToString.h>
#include <algorithm>
#include <cmath>
namespace DB
{
const String & getPartitionIdForPart(const ITTLMergeSelector::Part & part_info)
{
const MergeTreeData::DataPartPtr & part = part_info.getDataPartPtr();
return part->info.partition_id;
}
IMergeSelector::PartsRange ITTLMergeSelector::select(
const PartsRanges & parts_ranges,
size_t max_total_size_to_merge)
{
using Iterator = IMergeSelector::PartsRange::const_iterator;
Iterator best_begin;
ssize_t partition_to_merge_index = -1;
time_t partition_to_merge_min_ttl = 0;
/// Find most old TTL.
for (size_t i = 0; i < parts_ranges.size(); ++i)
{
const auto & mergeable_parts_in_partition = parts_ranges[i];
if (mergeable_parts_in_partition.empty())
continue;
const auto & partition_id = getPartitionIdForPart(mergeable_parts_in_partition.front());
const auto & next_merge_time_for_partition = merge_due_times[partition_id];
if (next_merge_time_for_partition > current_time)
continue;
for (Iterator part_it = mergeable_parts_in_partition.cbegin(); part_it != mergeable_parts_in_partition.cend(); ++part_it)
{
time_t ttl = getTTLForPart(*part_it);
if (ttl && !isTTLAlreadySatisfied(*part_it) && (partition_to_merge_index == -1 || ttl < partition_to_merge_min_ttl))
{
partition_to_merge_min_ttl = ttl;
partition_to_merge_index = i;
best_begin = part_it;
}
}
}
if (partition_to_merge_index == -1 || partition_to_merge_min_ttl > current_time)
return {};
const auto & best_partition = parts_ranges[partition_to_merge_index];
Iterator best_end = best_begin + 1;
size_t total_size = 0;
/// Find begin of range with most old TTL.
while (true)
{
time_t ttl = getTTLForPart(*best_begin);
if (!ttl || isTTLAlreadySatisfied(*best_begin) || ttl > current_time
|| (max_total_size_to_merge && total_size > max_total_size_to_merge))
{
/// This condition can not be satisfied on first iteration.
++best_begin;
break;
}
total_size += best_begin->size;
if (best_begin == best_partition.begin())
break;
--best_begin;
}
/// Find end of range with most old TTL.
while (best_end != best_partition.end())
{
time_t ttl = getTTLForPart(*best_end);
if (!ttl || isTTLAlreadySatisfied(*best_end) || ttl > current_time
|| (max_total_size_to_merge && total_size > max_total_size_to_merge))
break;
total_size += best_end->size;
++best_end;
}
if (!dry_run)
{
const auto & best_partition_id = getPartitionIdForPart(best_partition.front());
merge_due_times[best_partition_id] = current_time + merge_cooldown_time;
}
return PartsRange(best_begin, best_end);
}
time_t TTLDeleteMergeSelector::getTTLForPart(const IMergeSelector::Part & part) const
{
return only_drop_parts ? part.ttl_infos->part_max_ttl : part.ttl_infos->part_min_ttl;
}
bool TTLDeleteMergeSelector::isTTLAlreadySatisfied(const IMergeSelector::Part & part) const
{
/// N.B. Satisfied TTL means that TTL is NOT expired.
/// return true -- this part can not be selected
/// return false -- this part can be selected
/// Dropping whole part is an exception to `shall_participate_in_merges` logic.
if (only_drop_parts)
return false;
/// All TTL satisfied
if (!part.ttl_infos->hasAnyNonFinishedTTLs())
return true;
return !part.shall_participate_in_merges;
}
time_t TTLRecompressMergeSelector::getTTLForPart(const IMergeSelector::Part & part) const
{
return part.ttl_infos->getMinimalMaxRecompressionTTL();
}
bool TTLRecompressMergeSelector::isTTLAlreadySatisfied(const IMergeSelector::Part & part) const
{
/// N.B. Satisfied TTL means that TTL is NOT expired.
/// return true -- this part can not be selected
/// return false -- this part can be selected
if (!part.shall_participate_in_merges)
return true;
if (recompression_ttls.empty())
return false;
auto ttl_description = selectTTLDescriptionForTTLInfos(recompression_ttls, part.ttl_infos->recompression_ttl, current_time, true);
if (!ttl_description)
return true;
auto ast_to_str = [](ASTPtr query) -> String
{
if (!query)
return "";
return queryToString(query);
};
return ast_to_str(ttl_description->recompression_codec) == ast_to_str(part.compression_codec_desc);
}
}
|
