/root/doris/be/src/olap/full_compaction.cpp
Line | Count | Source (jump to first uncovered line) |
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 | | #include "olap/full_compaction.h" |
19 | | |
20 | | #include <glog/logging.h> |
21 | | #include <time.h> |
22 | | |
23 | | #include <memory> |
24 | | #include <mutex> |
25 | | #include <ostream> |
26 | | #include <shared_mutex> |
27 | | |
28 | | #include "common/config.h" |
29 | | #include "common/status.h" |
30 | | #include "olap/compaction.h" |
31 | | #include "olap/cumulative_compaction_policy.h" |
32 | | #include "olap/olap_common.h" |
33 | | #include "olap/olap_define.h" |
34 | | #include "olap/rowset/beta_rowset.h" |
35 | | #include "olap/rowset/rowset.h" |
36 | | #include "olap/schema_change.h" |
37 | | #include "olap/tablet_meta.h" |
38 | | #include "runtime/thread_context.h" |
39 | | #include "util/thread.h" |
40 | | #include "util/trace.h" |
41 | | |
42 | | namespace doris { |
43 | | using namespace ErrorCode; |
44 | | |
45 | | FullCompaction::FullCompaction(StorageEngine& engine, const TabletSharedPtr& tablet) |
46 | 0 | : CompactionMixin(engine, tablet, "FullCompaction:" + std::to_string(tablet->tablet_id())) { |
47 | 0 | } |
48 | | |
49 | 0 | FullCompaction::~FullCompaction() { |
50 | 0 | tablet()->set_is_full_compaction_running(false); |
51 | 0 | } |
52 | | |
53 | 0 | Status FullCompaction::prepare_compact() { |
54 | 0 | if (!tablet()->init_succeeded()) { |
55 | 0 | return Status::Error<INVALID_ARGUMENT, false>("Full compaction init failed"); |
56 | 0 | } |
57 | | |
58 | 0 | std::unique_lock base_lock(tablet()->get_base_compaction_lock()); |
59 | 0 | std::unique_lock cumu_lock(tablet()->get_cumulative_compaction_lock()); |
60 | 0 | tablet()->set_is_full_compaction_running(true); |
61 | |
|
62 | 0 | DBUG_EXECUTE_IF("FullCompaction.prepare_compact.set_cumu_point", |
63 | 0 | { tablet()->set_cumulative_layer_point(tablet()->max_version_unlocked() + 1); }) |
64 | | |
65 | | // 1. pick rowsets to compact |
66 | 0 | RETURN_IF_ERROR(pick_rowsets_to_compact()); |
67 | | |
68 | 0 | return Status::OK(); |
69 | 0 | } |
70 | | |
71 | 0 | Status FullCompaction::execute_compact() { |
72 | 0 | std::unique_lock base_lock(tablet()->get_base_compaction_lock()); |
73 | 0 | std::unique_lock cumu_lock(tablet()->get_cumulative_compaction_lock()); |
74 | |
|
75 | 0 | SCOPED_ATTACH_TASK(_mem_tracker); |
76 | |
|
77 | 0 | RETURN_IF_ERROR(CompactionMixin::execute_compact()); |
78 | | |
79 | 0 | tablet()->cumulative_compaction_policy()->update_compaction_level(tablet(), _input_rowsets, |
80 | 0 | _output_rowset); |
81 | |
|
82 | 0 | Version last_version = _input_rowsets.back()->version(); |
83 | 0 | tablet()->cumulative_compaction_policy()->update_cumulative_point(tablet(), _input_rowsets, |
84 | 0 | _output_rowset, last_version); |
85 | 0 | VLOG_CRITICAL << "after cumulative compaction, current cumulative point is " |
86 | 0 | << tablet()->cumulative_layer_point() << ", tablet=" << _tablet->tablet_id(); |
87 | |
|
88 | 0 | tablet()->set_last_full_compaction_success_time(UnixMillis()); |
89 | |
|
90 | 0 | return Status::OK(); |
91 | 0 | } |
92 | | |
93 | 0 | Status FullCompaction::pick_rowsets_to_compact() { |
94 | 0 | _input_rowsets = tablet()->pick_candidate_rowsets_to_full_compaction(); |
95 | 0 | RETURN_IF_ERROR(check_version_continuity(_input_rowsets)); |
96 | 0 | RETURN_IF_ERROR(_check_all_version(_input_rowsets)); |
97 | 0 | if (_input_rowsets.size() <= 1) { |
98 | 0 | return Status::Error<FULL_NO_SUITABLE_VERSION>("There is no suitable version"); |
99 | 0 | } |
100 | | |
101 | 0 | if (_input_rowsets.size() == 2 && _input_rowsets[0]->end_version() == 1) { |
102 | | // the tablet is with rowset: [0-1], [2-y] |
103 | | // and [0-1] has no data. in this situation, no need to do full compaction. |
104 | 0 | return Status::Error<FULL_NO_SUITABLE_VERSION>("There is no suitable version"); |
105 | 0 | } |
106 | | |
107 | 0 | return Status::OK(); |
108 | 0 | } |
109 | | |
110 | 0 | Status FullCompaction::modify_rowsets() { |
111 | 0 | if (_tablet->keys_type() == KeysType::UNIQUE_KEYS && |
112 | 0 | _tablet->enable_unique_key_merge_on_write()) { |
113 | 0 | RETURN_IF_ERROR( |
114 | 0 | _full_compaction_update_delete_bitmap(_output_rowset, _output_rs_writer.get())); |
115 | 0 | } |
116 | 0 | std::vector<RowsetSharedPtr> output_rowsets(1, _output_rowset); |
117 | 0 | { |
118 | 0 | std::lock_guard<std::mutex> rowset_update_wlock(tablet()->get_rowset_update_lock()); |
119 | 0 | std::lock_guard<std::shared_mutex> meta_wlock(_tablet->get_header_lock()); |
120 | 0 | RETURN_IF_ERROR(tablet()->modify_rowsets(output_rowsets, _input_rowsets, true)); |
121 | 0 | DBUG_EXECUTE_IF("FullCompaction.modify_rowsets.sleep", { sleep(5); }) |
122 | 0 | tablet()->save_meta(); |
123 | 0 | } |
124 | 0 | return Status::OK(); |
125 | 0 | } |
126 | | |
127 | 0 | Status FullCompaction::_check_all_version(const std::vector<RowsetSharedPtr>& rowsets) { |
128 | 0 | if (rowsets.empty()) { |
129 | 0 | return Status::Error<FULL_MISS_VERSION, false>( |
130 | 0 | "There is no input rowset when do full compaction"); |
131 | 0 | } |
132 | 0 | const RowsetSharedPtr& last_rowset = rowsets.back(); |
133 | 0 | const RowsetSharedPtr& first_rowset = rowsets.front(); |
134 | 0 | auto max_version = tablet()->max_version(); |
135 | 0 | if (last_rowset->version() != max_version || first_rowset->version().first != 0) { |
136 | 0 | return Status::Error<FULL_MISS_VERSION, false>( |
137 | 0 | "Full compaction rowsets' versions not equal to all exist rowsets' versions. " |
138 | 0 | "full compaction rowsets max version={}-{}" |
139 | 0 | ", current rowsets max version={}-{}" |
140 | 0 | ", full compaction rowsets min version={}-{}, current rowsets min version=0-1", |
141 | 0 | last_rowset->start_version(), last_rowset->end_version(), max_version.first, |
142 | 0 | max_version.second, first_rowset->start_version(), first_rowset->end_version()); |
143 | 0 | } |
144 | 0 | return Status::OK(); |
145 | 0 | } |
146 | | |
147 | | Status FullCompaction::_full_compaction_update_delete_bitmap(const RowsetSharedPtr& rowset, |
148 | 0 | RowsetWriter* rowset_writer) { |
149 | 0 | std::vector<RowsetSharedPtr> tmp_rowsets {}; |
150 | | |
151 | | // tablet is under alter process. The delete bitmap will be calculated after conversion. |
152 | 0 | if (_tablet->tablet_state() == TABLET_NOTREADY) { |
153 | 0 | LOG(INFO) << "tablet is under alter process, update delete bitmap later, tablet_id=" |
154 | 0 | << _tablet->tablet_id(); |
155 | 0 | return Status::OK(); |
156 | 0 | } |
157 | | |
158 | 0 | int64_t max_version = tablet()->max_version().second; |
159 | 0 | DCHECK(max_version >= rowset->version().second); |
160 | 0 | if (max_version > rowset->version().second) { |
161 | 0 | RETURN_IF_ERROR(_tablet->capture_consistent_rowsets_unlocked( |
162 | 0 | {rowset->version().second + 1, max_version}, &tmp_rowsets)); |
163 | 0 | } |
164 | | |
165 | 0 | for (const auto& it : tmp_rowsets) { |
166 | 0 | const int64_t& cur_version = it->rowset_meta()->start_version(); |
167 | 0 | RETURN_IF_ERROR( |
168 | 0 | _full_compaction_calc_delete_bitmap(it, rowset, cur_version, rowset_writer)); |
169 | 0 | } |
170 | | |
171 | 0 | std::lock_guard rowset_update_lock(tablet()->get_rowset_update_lock()); |
172 | 0 | std::lock_guard header_lock(_tablet->get_header_lock()); |
173 | 0 | SCOPED_SIMPLE_TRACE_IF_TIMEOUT(TRACE_TABLET_LOCK_THRESHOLD); |
174 | 0 | for (const auto& it : tablet()->rowset_map()) { |
175 | 0 | const int64_t& cur_version = it.first.first; |
176 | 0 | const RowsetSharedPtr& published_rowset = it.second; |
177 | 0 | if (cur_version > max_version) { |
178 | 0 | RETURN_IF_ERROR(_full_compaction_calc_delete_bitmap(published_rowset, rowset, |
179 | 0 | cur_version, rowset_writer)); |
180 | 0 | } |
181 | 0 | } |
182 | | |
183 | 0 | return Status::OK(); |
184 | 0 | } |
185 | | |
186 | | Status FullCompaction::_full_compaction_calc_delete_bitmap(const RowsetSharedPtr& published_rowset, |
187 | | const RowsetSharedPtr& rowset, |
188 | | const int64_t& cur_version, |
189 | 0 | RowsetWriter* rowset_writer) { |
190 | 0 | std::vector<segment_v2::SegmentSharedPtr> segments; |
191 | 0 | auto beta_rowset = reinterpret_cast<BetaRowset*>(published_rowset.get()); |
192 | 0 | RETURN_IF_ERROR(beta_rowset->load_segments(&segments)); |
193 | 0 | DeleteBitmapPtr delete_bitmap = |
194 | 0 | std::make_shared<DeleteBitmap>(_tablet->tablet_meta()->tablet_id()); |
195 | 0 | std::vector<RowsetSharedPtr> specified_rowsets(1, rowset); |
196 | |
|
197 | 0 | OlapStopWatch watch; |
198 | 0 | RETURN_IF_ERROR(BaseTablet::calc_delete_bitmap(_tablet, published_rowset, segments, |
199 | 0 | specified_rowsets, delete_bitmap, cur_version, |
200 | 0 | nullptr, rowset_writer)); |
201 | 0 | size_t total_rows = std::accumulate( |
202 | 0 | segments.begin(), segments.end(), 0, |
203 | 0 | [](size_t sum, const segment_v2::SegmentSharedPtr& s) { return sum += s->num_rows(); }); |
204 | 0 | VLOG_DEBUG << "[Full compaction] construct delete bitmap tablet: " << _tablet->tablet_id() |
205 | 0 | << ", published rowset version: [" << published_rowset->version().first << "-" |
206 | 0 | << published_rowset->version().second << "]" |
207 | 0 | << ", full compaction rowset version: [" << rowset->version().first << "-" |
208 | 0 | << rowset->version().second << "]" |
209 | 0 | << ", cost: " << watch.get_elapse_time_us() << "(us), total rows: " << total_rows; |
210 | |
|
211 | 0 | for (const auto& [k, v] : delete_bitmap->delete_bitmap) { |
212 | 0 | _tablet->tablet_meta()->delete_bitmap().merge({std::get<0>(k), std::get<1>(k), cur_version}, |
213 | 0 | v); |
214 | 0 | } |
215 | |
|
216 | 0 | return Status::OK(); |
217 | 0 | } |
218 | | |
219 | | } // namespace doris |