/root/doris/be/src/cloud/cloud_tablet.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 "cloud/cloud_tablet.h" |
19 | | |
20 | | #include <bvar/bvar.h> |
21 | | #include <gen_cpp/olap_file.pb.h> |
22 | | #include <rapidjson/document.h> |
23 | | #include <rapidjson/encodings.h> |
24 | | #include <rapidjson/prettywriter.h> |
25 | | #include <rapidjson/rapidjson.h> |
26 | | #include <rapidjson/stringbuffer.h> |
27 | | |
28 | | #include <atomic> |
29 | | #include <memory> |
30 | | #include <shared_mutex> |
31 | | #include <unordered_map> |
32 | | #include <vector> |
33 | | |
34 | | #include "cloud/cloud_meta_mgr.h" |
35 | | #include "cloud/cloud_storage_engine.h" |
36 | | #include "cloud/cloud_tablet_mgr.h" |
37 | | #include "common/cast_set.h" |
38 | | #include "common/config.h" |
39 | | #include "common/logging.h" |
40 | | #include "io/cache/block_file_cache_downloader.h" |
41 | | #include "io/cache/block_file_cache_factory.h" |
42 | | #include "olap/compaction.h" |
43 | | #include "olap/cumulative_compaction_time_series_policy.h" |
44 | | #include "olap/olap_define.h" |
45 | | #include "olap/rowset/beta_rowset.h" |
46 | | #include "olap/rowset/rowset.h" |
47 | | #include "olap/rowset/rowset_factory.h" |
48 | | #include "olap/rowset/rowset_fwd.h" |
49 | | #include "olap/rowset/rowset_writer.h" |
50 | | #include "olap/rowset/segment_v2/inverted_index_desc.h" |
51 | | #include "olap/storage_policy.h" |
52 | | #include "olap/tablet_schema.h" |
53 | | #include "olap/txn_manager.h" |
54 | | #include "util/debug_points.h" |
55 | | #include "vec/common/schema_util.h" |
56 | | |
57 | | namespace doris { |
58 | | #include "common/compile_check_begin.h" |
59 | | using namespace ErrorCode; |
60 | | |
61 | | bvar::LatencyRecorder g_cu_compaction_get_delete_bitmap_lock_time_ms( |
62 | | "cu_compaction_get_delete_bitmap_lock_time_ms"); |
63 | | bvar::LatencyRecorder g_base_compaction_get_delete_bitmap_lock_time_ms( |
64 | | "base_compaction_get_delete_bitmap_lock_time_ms"); |
65 | | |
66 | | static constexpr int LOAD_INITIATOR_ID = -1; |
67 | | |
68 | | CloudTablet::CloudTablet(CloudStorageEngine& engine, TabletMetaSharedPtr tablet_meta) |
69 | 3 | : BaseTablet(std::move(tablet_meta)), _engine(engine) {} |
70 | | |
71 | 3 | CloudTablet::~CloudTablet() = default; |
72 | | |
73 | 0 | bool CloudTablet::exceed_version_limit(int32_t limit) { |
74 | 0 | return _approximate_num_rowsets.load(std::memory_order_relaxed) > limit; |
75 | 0 | } |
76 | | |
77 | | Status CloudTablet::capture_consistent_rowsets_unlocked( |
78 | 0 | const Version& spec_version, std::vector<RowsetSharedPtr>* rowsets) const { |
79 | 0 | Versions version_path; |
80 | 0 | auto st = _timestamped_version_tracker.capture_consistent_versions(spec_version, &version_path); |
81 | 0 | if (!st.ok()) { |
82 | | // Check no missed versions or req version is merged |
83 | 0 | auto missed_versions = get_missed_versions(spec_version.second); |
84 | 0 | if (missed_versions.empty()) { |
85 | 0 | st.set_code(VERSION_ALREADY_MERGED); // Reset error code |
86 | 0 | } |
87 | 0 | st.append(" tablet_id=" + std::to_string(tablet_id())); |
88 | 0 | return st; |
89 | 0 | } |
90 | 0 | VLOG_DEBUG << "capture consitent versions: " << version_path; |
91 | 0 | return _capture_consistent_rowsets_unlocked(version_path, rowsets); |
92 | 0 | } |
93 | | |
94 | | Status CloudTablet::capture_rs_readers(const Version& spec_version, |
95 | | std::vector<RowSetSplits>* rs_splits, |
96 | 0 | bool skip_missing_version) { |
97 | 0 | DBUG_EXECUTE_IF("CloudTablet.capture_rs_readers.return.e-230", { |
98 | 0 | LOG_WARNING("CloudTablet.capture_rs_readers.return e-230").tag("tablet_id", tablet_id()); |
99 | 0 | return Status::Error<false>(-230, "injected error"); |
100 | 0 | }); |
101 | 0 | Versions version_path; |
102 | 0 | std::shared_lock rlock(_meta_lock); |
103 | 0 | auto st = _timestamped_version_tracker.capture_consistent_versions(spec_version, &version_path); |
104 | 0 | if (!st.ok()) { |
105 | 0 | rlock.unlock(); // avoid logging in lock range |
106 | | // Check no missed versions or req version is merged |
107 | 0 | auto missed_versions = get_missed_versions(spec_version.second); |
108 | 0 | if (missed_versions.empty()) { |
109 | 0 | st.set_code(VERSION_ALREADY_MERGED); // Reset error code |
110 | 0 | st.append(" versions are already compacted, "); |
111 | 0 | } |
112 | 0 | st.append(" tablet_id=" + std::to_string(tablet_id())); |
113 | | // clang-format off |
114 | 0 | LOG(WARNING) << st << '\n' << [this]() { std::string json; get_compaction_status(&json); return json; }(); |
115 | | // clang-format on |
116 | 0 | return st; |
117 | 0 | } |
118 | 0 | VLOG_DEBUG << "capture consitent versions: " << version_path; |
119 | 0 | return capture_rs_readers_unlocked(version_path, rs_splits); |
120 | 0 | } |
121 | | |
122 | 0 | Status CloudTablet::merge_rowsets_schema() { |
123 | | // Find the rowset with the max version |
124 | 0 | auto max_version_rowset = |
125 | 0 | std::max_element( |
126 | 0 | _rs_version_map.begin(), _rs_version_map.end(), |
127 | 0 | [](const auto& a, const auto& b) { |
128 | 0 | return !a.second->tablet_schema() |
129 | 0 | ? true |
130 | 0 | : (!b.second->tablet_schema() |
131 | 0 | ? false |
132 | 0 | : a.second->tablet_schema()->schema_version() < |
133 | 0 | b.second->tablet_schema() |
134 | 0 | ->schema_version()); |
135 | 0 | }) |
136 | 0 | ->second; |
137 | 0 | TabletSchemaSPtr max_version_schema = max_version_rowset->tablet_schema(); |
138 | | // If the schema has variant columns, perform a merge to create a wide tablet schema |
139 | 0 | if (max_version_schema->num_variant_columns() > 0) { |
140 | 0 | std::vector<TabletSchemaSPtr> schemas; |
141 | 0 | std::transform(_rs_version_map.begin(), _rs_version_map.end(), std::back_inserter(schemas), |
142 | 0 | [](const auto& rs_meta) { return rs_meta.second->tablet_schema(); }); |
143 | | // Merge the collected schemas to obtain the least common schema |
144 | 0 | RETURN_IF_ERROR(vectorized::schema_util::get_least_common_schema(schemas, nullptr, |
145 | 0 | max_version_schema)); |
146 | 0 | VLOG_DEBUG << "dump schema: " << max_version_schema->dump_full_schema(); |
147 | 0 | _merged_tablet_schema = max_version_schema; |
148 | 0 | } |
149 | 0 | return Status::OK(); |
150 | 0 | } |
151 | | |
152 | | // There are only two tablet_states RUNNING and NOT_READY in cloud mode |
153 | | // This function will erase the tablet from `CloudTabletMgr` when it can't find this tablet in MS. |
154 | 0 | Status CloudTablet::sync_rowsets(const SyncOptions& options, SyncRowsetStats* stats) { |
155 | 0 | RETURN_IF_ERROR(sync_if_not_running(stats)); |
156 | | |
157 | 0 | if (options.query_version > 0) { |
158 | 0 | std::shared_lock rlock(_meta_lock); |
159 | 0 | if (_max_version >= options.query_version) { |
160 | 0 | return Status::OK(); |
161 | 0 | } |
162 | 0 | } |
163 | | |
164 | | // serially execute sync to reduce unnecessary network overhead |
165 | 0 | std::unique_lock lock(_sync_meta_lock); |
166 | 0 | if (options.query_version > 0) { |
167 | 0 | std::shared_lock rlock(_meta_lock); |
168 | 0 | if (_max_version >= options.query_version) { |
169 | 0 | return Status::OK(); |
170 | 0 | } |
171 | 0 | } |
172 | | |
173 | 0 | auto st = _engine.meta_mgr().sync_tablet_rowsets_unlocked(this, lock, options, stats); |
174 | 0 | if (st.is<ErrorCode::NOT_FOUND>()) { |
175 | 0 | clear_cache(); |
176 | 0 | } |
177 | |
|
178 | 0 | return st; |
179 | 0 | } |
180 | | |
181 | | // Sync tablet meta and all rowset meta if not running. |
182 | | // This could happen when BE didn't finish schema change job and another BE committed this schema change job. |
183 | | // It should be a quite rare situation. |
184 | 0 | Status CloudTablet::sync_if_not_running(SyncRowsetStats* stats) { |
185 | 0 | if (tablet_state() == TABLET_RUNNING) { |
186 | 0 | return Status::OK(); |
187 | 0 | } |
188 | | |
189 | | // Serially execute sync to reduce unnecessary network overhead |
190 | 0 | std::unique_lock lock(_sync_meta_lock); |
191 | |
|
192 | 0 | { |
193 | 0 | std::shared_lock rlock(_meta_lock); |
194 | 0 | if (tablet_state() == TABLET_RUNNING) { |
195 | 0 | return Status::OK(); |
196 | 0 | } |
197 | 0 | } |
198 | | |
199 | 0 | TabletMetaSharedPtr tablet_meta; |
200 | 0 | auto st = _engine.meta_mgr().get_tablet_meta(tablet_id(), &tablet_meta); |
201 | 0 | if (!st.ok()) { |
202 | 0 | if (st.is<ErrorCode::NOT_FOUND>()) { |
203 | 0 | clear_cache(); |
204 | 0 | } |
205 | 0 | return st; |
206 | 0 | } |
207 | | |
208 | 0 | if (tablet_meta->tablet_state() != TABLET_RUNNING) [[unlikely]] { |
209 | | // MoW may go to here when load while schema change |
210 | 0 | return Status::OK(); |
211 | 0 | } |
212 | | |
213 | 0 | TimestampedVersionTracker empty_tracker; |
214 | 0 | { |
215 | 0 | std::lock_guard wlock(_meta_lock); |
216 | 0 | RETURN_IF_ERROR(set_tablet_state(TABLET_RUNNING)); |
217 | 0 | _rs_version_map.clear(); |
218 | 0 | _stale_rs_version_map.clear(); |
219 | 0 | std::swap(_timestamped_version_tracker, empty_tracker); |
220 | 0 | _tablet_meta->clear_rowsets(); |
221 | 0 | _tablet_meta->clear_stale_rowset(); |
222 | 0 | _max_version = -1; |
223 | 0 | } |
224 | | |
225 | 0 | st = _engine.meta_mgr().sync_tablet_rowsets_unlocked(this, lock, {}, stats); |
226 | 0 | if (st.is<ErrorCode::NOT_FOUND>()) { |
227 | 0 | clear_cache(); |
228 | 0 | } |
229 | 0 | return st; |
230 | 0 | } |
231 | | |
232 | 0 | TabletSchemaSPtr CloudTablet::merged_tablet_schema() const { |
233 | 0 | std::shared_lock rlock(_meta_lock); |
234 | 0 | return _merged_tablet_schema; |
235 | 0 | } |
236 | | |
237 | | void CloudTablet::add_rowsets(std::vector<RowsetSharedPtr> to_add, bool version_overlap, |
238 | | std::unique_lock<std::shared_mutex>& meta_lock, |
239 | 0 | bool warmup_delta_data) { |
240 | 0 | if (to_add.empty()) { |
241 | 0 | return; |
242 | 0 | } |
243 | | |
244 | 0 | auto add_rowsets_directly = [=, this](std::vector<RowsetSharedPtr>& rowsets) { |
245 | 0 | for (auto& rs : rowsets) { |
246 | 0 | if (version_overlap || warmup_delta_data) { |
247 | | #ifndef BE_TEST |
248 | | // Warmup rowset data in background |
249 | | for (int seg_id = 0; seg_id < rs->num_segments(); ++seg_id) { |
250 | | const auto& rowset_meta = rs->rowset_meta(); |
251 | | constexpr int64_t interval = 600; // 10 mins |
252 | | // When BE restart and receive the `load_sync` rpc, it will sync all historical rowsets first time. |
253 | | // So we need to filter out the old rowsets avoid to download the whole table. |
254 | | if (warmup_delta_data && |
255 | | ::time(nullptr) - rowset_meta->newest_write_timestamp() >= interval) { |
256 | | continue; |
257 | | } |
258 | | |
259 | | auto storage_resource = rowset_meta->remote_storage_resource(); |
260 | | if (!storage_resource) { |
261 | | LOG(WARNING) << storage_resource.error(); |
262 | | continue; |
263 | | } |
264 | | |
265 | | int64_t expiration_time = |
266 | | _tablet_meta->ttl_seconds() == 0 || |
267 | | rowset_meta->newest_write_timestamp() <= 0 |
268 | | ? 0 |
269 | | : rowset_meta->newest_write_timestamp() + |
270 | | _tablet_meta->ttl_seconds(); |
271 | | // clang-format off |
272 | | _engine.file_cache_block_downloader().submit_download_task(io::DownloadFileMeta { |
273 | | .path = storage_resource.value()->remote_segment_path(*rowset_meta, seg_id), |
274 | | .file_size = rs->rowset_meta()->segment_file_size(seg_id), |
275 | | .file_system = storage_resource.value()->fs, |
276 | | .ctx = |
277 | | { |
278 | | .expiration_time = expiration_time, |
279 | | .is_dryrun = config::enable_reader_dryrun_when_download_file_cache, |
280 | | }, |
281 | | .download_done {}, |
282 | | }); |
283 | | |
284 | | auto download_idx_file = [&](const io::Path& idx_path) { |
285 | | io::DownloadFileMeta meta { |
286 | | .path = idx_path, |
287 | | .file_size = -1, |
288 | | .file_system = storage_resource.value()->fs, |
289 | | .ctx = |
290 | | { |
291 | | .expiration_time = expiration_time, |
292 | | .is_dryrun = config::enable_reader_dryrun_when_download_file_cache, |
293 | | }, |
294 | | .download_done {}, |
295 | | }; |
296 | | _engine.file_cache_block_downloader().submit_download_task(std::move(meta)); |
297 | | }; |
298 | | // clang-format on |
299 | | auto schema_ptr = rowset_meta->tablet_schema(); |
300 | | auto idx_version = schema_ptr->get_inverted_index_storage_format(); |
301 | | if (idx_version == InvertedIndexStorageFormatPB::V1) { |
302 | | for (const auto& index : schema_ptr->inverted_indexes()) { |
303 | | auto idx_path = storage_resource.value()->remote_idx_v1_path( |
304 | | *rowset_meta, seg_id, index->index_id(), |
305 | | index->get_index_suffix()); |
306 | | download_idx_file(idx_path); |
307 | | } |
308 | | } else { |
309 | | if (schema_ptr->has_inverted_index()) { |
310 | | auto idx_path = storage_resource.value()->remote_idx_v2_path( |
311 | | *rowset_meta, seg_id); |
312 | | download_idx_file(idx_path); |
313 | | } |
314 | | } |
315 | | } |
316 | | #endif |
317 | 0 | } |
318 | 0 | _rs_version_map.emplace(rs->version(), rs); |
319 | 0 | _timestamped_version_tracker.add_version(rs->version()); |
320 | 0 | _max_version = std::max(rs->end_version(), _max_version); |
321 | 0 | update_base_size(*rs); |
322 | 0 | } |
323 | 0 | _tablet_meta->add_rowsets_unchecked(rowsets); |
324 | 0 | }; |
325 | |
|
326 | 0 | if (!version_overlap) { |
327 | 0 | add_rowsets_directly(to_add); |
328 | 0 | return; |
329 | 0 | } |
330 | | |
331 | | // Filter out existed rowsets |
332 | 0 | auto remove_it = |
333 | 0 | std::remove_if(to_add.begin(), to_add.end(), [this](const RowsetSharedPtr& rs) { |
334 | 0 | if (auto find_it = _rs_version_map.find(rs->version()); |
335 | 0 | find_it == _rs_version_map.end()) { |
336 | 0 | return false; |
337 | 0 | } else if (find_it->second->rowset_id() == rs->rowset_id()) { |
338 | 0 | return true; // Same rowset |
339 | 0 | } |
340 | | |
341 | | // If version of rowset in `to_add` is equal to rowset in tablet but rowset_id is not equal, |
342 | | // replace existed rowset with `to_add` rowset. This may occur when: |
343 | | // 1. schema change converts rowsets which have been double written to new tablet |
344 | | // 2. cumu compaction picks single overlapping input rowset to perform compaction |
345 | 0 | _tablet_meta->delete_rs_meta_by_version(rs->version(), nullptr); |
346 | 0 | _rs_version_map[rs->version()] = rs; |
347 | 0 | _tablet_meta->add_rowsets_unchecked({rs}); |
348 | 0 | update_base_size(*rs); |
349 | 0 | return true; |
350 | 0 | }); |
351 | |
|
352 | 0 | to_add.erase(remove_it, to_add.end()); |
353 | | |
354 | | // delete rowsets with overlapped version |
355 | 0 | std::vector<RowsetSharedPtr> to_add_directly; |
356 | 0 | for (auto& to_add_rs : to_add) { |
357 | | // delete rowsets with overlapped version |
358 | 0 | std::vector<RowsetSharedPtr> to_delete; |
359 | 0 | Version to_add_v = to_add_rs->version(); |
360 | | // if start_version > max_version, we can skip checking overlap here. |
361 | 0 | if (to_add_v.first > _max_version) { |
362 | | // if start_version > max_version, we can skip checking overlap here. |
363 | 0 | to_add_directly.push_back(to_add_rs); |
364 | 0 | } else { |
365 | 0 | to_add_directly.push_back(to_add_rs); |
366 | 0 | for (auto& [v, rs] : _rs_version_map) { |
367 | 0 | if (to_add_v.contains(v)) { |
368 | 0 | to_delete.push_back(rs); |
369 | 0 | } |
370 | 0 | } |
371 | 0 | delete_rowsets(to_delete, meta_lock); |
372 | 0 | } |
373 | 0 | } |
374 | |
|
375 | 0 | add_rowsets_directly(to_add_directly); |
376 | 0 | } |
377 | | |
378 | | void CloudTablet::delete_rowsets(const std::vector<RowsetSharedPtr>& to_delete, |
379 | 0 | std::unique_lock<std::shared_mutex>&) { |
380 | 0 | if (to_delete.empty()) { |
381 | 0 | return; |
382 | 0 | } |
383 | 0 | std::vector<RowsetMetaSharedPtr> rs_metas; |
384 | 0 | rs_metas.reserve(to_delete.size()); |
385 | 0 | for (auto&& rs : to_delete) { |
386 | 0 | rs_metas.push_back(rs->rowset_meta()); |
387 | 0 | _stale_rs_version_map[rs->version()] = rs; |
388 | 0 | } |
389 | 0 | _timestamped_version_tracker.add_stale_path_version(rs_metas); |
390 | 0 | for (auto&& rs : to_delete) { |
391 | 0 | _rs_version_map.erase(rs->version()); |
392 | 0 | } |
393 | |
|
394 | 0 | _tablet_meta->modify_rs_metas({}, rs_metas, false); |
395 | 0 | } |
396 | | |
397 | 0 | uint64_t CloudTablet::delete_expired_stale_rowsets() { |
398 | 0 | if (config::enable_mow_verbose_log) { |
399 | 0 | LOG_INFO("begin delete_expired_stale_rowset for tablet={}", tablet_id()); |
400 | 0 | } |
401 | 0 | std::vector<RowsetSharedPtr> expired_rowsets; |
402 | | // ATTN: trick, Use stale_rowsets to temporarily increase the reference count of the rowset shared pointer in _stale_rs_version_map so that in the recycle_cached_data function, it checks if the reference count is 2. |
403 | 0 | std::vector<std::pair<Version, std::vector<RowsetSharedPtr>>> deleted_stale_rowsets; |
404 | 0 | int64_t expired_stale_sweep_endtime = |
405 | 0 | ::time(nullptr) - config::tablet_rowset_stale_sweep_time_sec; |
406 | 0 | std::vector<std::string> version_to_delete; |
407 | 0 | { |
408 | 0 | std::unique_lock wlock(_meta_lock); |
409 | |
|
410 | 0 | std::vector<int64_t> path_ids; |
411 | | // capture the path version to delete |
412 | 0 | _timestamped_version_tracker.capture_expired_paths(expired_stale_sweep_endtime, &path_ids); |
413 | |
|
414 | 0 | if (path_ids.empty()) { |
415 | 0 | return 0; |
416 | 0 | } |
417 | | |
418 | 0 | for (int64_t path_id : path_ids) { |
419 | 0 | int64_t start_version = -1; |
420 | 0 | int64_t end_version = -1; |
421 | 0 | std::vector<RowsetSharedPtr> stale_rowsets; |
422 | | // delete stale versions in version graph |
423 | 0 | auto version_path = _timestamped_version_tracker.fetch_and_delete_path_by_id(path_id); |
424 | 0 | for (auto& v_ts : version_path->timestamped_versions()) { |
425 | 0 | auto rs_it = _stale_rs_version_map.find(v_ts->version()); |
426 | 0 | if (rs_it != _stale_rs_version_map.end()) { |
427 | 0 | expired_rowsets.push_back(rs_it->second); |
428 | 0 | stale_rowsets.push_back(rs_it->second); |
429 | 0 | LOG(INFO) << "erase stale rowset, tablet_id=" << tablet_id() |
430 | 0 | << " rowset_id=" << rs_it->second->rowset_id().to_string() |
431 | 0 | << " version=" << rs_it->first.to_string(); |
432 | 0 | _stale_rs_version_map.erase(rs_it); |
433 | 0 | } else { |
434 | 0 | LOG(WARNING) << "cannot find stale rowset " << v_ts->version() << " in tablet " |
435 | 0 | << tablet_id(); |
436 | | // clang-format off |
437 | 0 | DCHECK(false) << [this, &wlock]() { wlock.unlock(); std::string json; get_compaction_status(&json); return json; }(); |
438 | | // clang-format on |
439 | 0 | } |
440 | 0 | if (start_version < 0) { |
441 | 0 | start_version = v_ts->version().first; |
442 | 0 | } |
443 | 0 | end_version = v_ts->version().second; |
444 | 0 | _tablet_meta->delete_stale_rs_meta_by_version(v_ts->version()); |
445 | 0 | } |
446 | 0 | Version version(start_version, end_version); |
447 | 0 | version_to_delete.emplace_back(version.to_string()); |
448 | 0 | if (!stale_rowsets.empty()) { |
449 | 0 | deleted_stale_rowsets.emplace_back(version, std::move(stale_rowsets)); |
450 | 0 | } |
451 | 0 | } |
452 | 0 | _reconstruct_version_tracker_if_necessary(); |
453 | 0 | } |
454 | 0 | _tablet_meta->delete_bitmap().remove_stale_delete_bitmap_from_queue(version_to_delete); |
455 | 0 | recycle_cached_data(expired_rowsets); |
456 | 0 | if (config::enable_mow_verbose_log) { |
457 | 0 | LOG_INFO("finish delete_expired_stale_rowset for tablet={}", tablet_id()); |
458 | 0 | } |
459 | |
|
460 | 0 | if (keys_type() == UNIQUE_KEYS && enable_unique_key_merge_on_write() && |
461 | 0 | !deleted_stale_rowsets.empty()) { |
462 | | // agg delete bitmap for pre rowsets; record unused delete bitmap key ranges |
463 | 0 | OlapStopWatch watch; |
464 | 0 | for (const auto& [version, unused_rowsets] : deleted_stale_rowsets) { |
465 | | // agg delete bitmap for pre rowset |
466 | 0 | DeleteBitmapKeyRanges remove_delete_bitmap_key_ranges; |
467 | 0 | agg_delete_bitmap_for_stale_rowsets(version, remove_delete_bitmap_key_ranges); |
468 | | // add remove delete bitmap |
469 | 0 | if (!remove_delete_bitmap_key_ranges.empty()) { |
470 | 0 | std::lock_guard<std::mutex> lock(_gc_mutex); |
471 | 0 | _unused_delete_bitmap.push_back( |
472 | 0 | std::make_pair(unused_rowsets, remove_delete_bitmap_key_ranges)); |
473 | 0 | } |
474 | 0 | } |
475 | 0 | LOG(INFO) << "agg pre rowsets delete bitmap. tablet_id=" << tablet_id() |
476 | 0 | << ", size=" << deleted_stale_rowsets.size() |
477 | 0 | << ", cost(us)=" << watch.get_elapse_time_us(); |
478 | 0 | } |
479 | 0 | return expired_rowsets.size(); |
480 | 0 | } |
481 | | |
482 | 0 | bool CloudTablet::need_remove_pre_rowset_delete_bitmap() { |
483 | 0 | std::lock_guard<std::mutex> lock(_gc_mutex); |
484 | 0 | return !_unused_delete_bitmap.empty(); |
485 | 0 | } |
486 | | |
487 | 0 | void CloudTablet::remove_pre_rowset_delete_bitmap() { |
488 | 0 | std::lock_guard<std::mutex> lock(_gc_mutex); |
489 | 0 | for (auto it = _unused_delete_bitmap.begin(); it != _unused_delete_bitmap.end();) { |
490 | 0 | auto& rowsets = std::get<0>(*it); |
491 | 0 | bool find_unused_rowset = false; |
492 | 0 | for (const auto& rowset : rowsets) { |
493 | 0 | if (rowset.use_count() > 1) { |
494 | 0 | LOG(INFO) << "can not remove pre rowset delete bitmap because rowset is in use" |
495 | 0 | << ", tablet_id=" << tablet_id() |
496 | 0 | << ", rowset_id=" << rowset->rowset_id().to_string() |
497 | 0 | << ", version=" << rowset->version().to_string() |
498 | 0 | << ", use_count=" << rowset.use_count(); |
499 | 0 | find_unused_rowset = true; |
500 | 0 | break; |
501 | 0 | } |
502 | 0 | } |
503 | 0 | if (find_unused_rowset) { |
504 | 0 | ++it; |
505 | 0 | continue; |
506 | 0 | } |
507 | 0 | auto& key_ranges = std::get<1>(*it); |
508 | 0 | tablet_meta()->delete_bitmap().remove(key_ranges); |
509 | 0 | it = _unused_delete_bitmap.erase(it); |
510 | 0 | } |
511 | 0 | if (!_unused_delete_bitmap.empty()) { |
512 | 0 | LOG(INFO) << "tablet_id=" << tablet_id() |
513 | 0 | << ", unused_delete_bitmap size=" << _unused_delete_bitmap.size(); |
514 | 0 | } |
515 | 0 | } |
516 | | |
517 | 0 | void CloudTablet::update_base_size(const Rowset& rs) { |
518 | | // Define base rowset as the rowset of version [2-x] |
519 | 0 | if (rs.start_version() == 2) { |
520 | 0 | _base_size = rs.total_disk_size(); |
521 | 0 | } |
522 | 0 | } |
523 | | |
524 | 0 | void CloudTablet::clear_cache() { |
525 | 0 | CloudTablet::recycle_cached_data(get_snapshot_rowset(true)); |
526 | 0 | _engine.tablet_mgr().erase_tablet(tablet_id()); |
527 | 0 | } |
528 | | |
529 | 0 | void CloudTablet::recycle_cached_data(const std::vector<RowsetSharedPtr>& rowsets) { |
530 | 0 | for (const auto& rs : rowsets) { |
531 | | // rowsets and tablet._rs_version_map each hold a rowset shared_ptr, so at this point, the reference count of the shared_ptr is at least 2. |
532 | 0 | if (rs.use_count() > 2) { |
533 | 0 | LOG(WARNING) << "Rowset " << rs->rowset_id().to_string() << " has " << rs.use_count() |
534 | 0 | << " references. File Cache won't be recycled when query is using it."; |
535 | 0 | return; |
536 | 0 | } |
537 | 0 | rs->clear_cache(); |
538 | 0 | } |
539 | 0 | } |
540 | | |
541 | | void CloudTablet::reset_approximate_stats(int64_t num_rowsets, int64_t num_segments, |
542 | 0 | int64_t num_rows, int64_t data_size) { |
543 | 0 | _approximate_num_rowsets.store(num_rowsets, std::memory_order_relaxed); |
544 | 0 | _approximate_num_segments.store(num_segments, std::memory_order_relaxed); |
545 | 0 | _approximate_num_rows.store(num_rows, std::memory_order_relaxed); |
546 | 0 | _approximate_data_size.store(data_size, std::memory_order_relaxed); |
547 | 0 | int64_t cumu_num_deltas = 0; |
548 | 0 | int64_t cumu_num_rowsets = 0; |
549 | 0 | auto cp = _cumulative_point.load(std::memory_order_relaxed); |
550 | 0 | for (auto& [v, r] : _rs_version_map) { |
551 | 0 | if (v.second < cp) { |
552 | 0 | continue; |
553 | 0 | } |
554 | | |
555 | 0 | cumu_num_deltas += r->is_segments_overlapping() ? r->num_segments() : 1; |
556 | 0 | ++cumu_num_rowsets; |
557 | 0 | } |
558 | 0 | _approximate_cumu_num_rowsets.store(cumu_num_rowsets, std::memory_order_relaxed); |
559 | 0 | _approximate_cumu_num_deltas.store(cumu_num_deltas, std::memory_order_relaxed); |
560 | 0 | } |
561 | | |
562 | | Result<std::unique_ptr<RowsetWriter>> CloudTablet::create_rowset_writer( |
563 | 0 | RowsetWriterContext& context, bool vertical) { |
564 | 0 | context.rowset_id = _engine.next_rowset_id(); |
565 | | // FIXME(plat1ko): Seems `tablet_id` and `index_id` has been set repeatedly |
566 | 0 | context.tablet_id = tablet_id(); |
567 | 0 | context.index_id = index_id(); |
568 | 0 | context.partition_id = partition_id(); |
569 | 0 | context.enable_unique_key_merge_on_write = enable_unique_key_merge_on_write(); |
570 | 0 | return RowsetFactory::create_rowset_writer(_engine, context, vertical); |
571 | 0 | } |
572 | | |
573 | | // create a rowset writer with rowset_id and seg_id |
574 | | // after writer, merge this transient rowset with original rowset |
575 | | Result<std::unique_ptr<RowsetWriter>> CloudTablet::create_transient_rowset_writer( |
576 | | const Rowset& rowset, std::shared_ptr<PartialUpdateInfo> partial_update_info, |
577 | 0 | int64_t txn_expiration) { |
578 | 0 | if (rowset.rowset_meta_state() != RowsetStatePB::BEGIN_PARTIAL_UPDATE && |
579 | 0 | rowset.rowset_meta_state() != RowsetStatePB::COMMITTED) [[unlikely]] { |
580 | 0 | auto msg = fmt::format( |
581 | 0 | "wrong rowset state when create_transient_rowset_writer, rowset state should be " |
582 | 0 | "BEGIN_PARTIAL_UPDATE or COMMITTED, but found {}, rowset_id={}, tablet_id={}", |
583 | 0 | RowsetStatePB_Name(rowset.rowset_meta_state()), rowset.rowset_id().to_string(), |
584 | 0 | tablet_id()); |
585 | | // see `CloudRowsetWriter::build` for detail. |
586 | | // if this is in a retry task, the rowset state may have been changed to RowsetStatePB::COMMITTED |
587 | | // in `RowsetMeta::merge_rowset_meta()` in previous trials. |
588 | 0 | LOG(WARNING) << msg; |
589 | 0 | DCHECK(false) << msg; |
590 | 0 | } |
591 | 0 | RowsetWriterContext context; |
592 | 0 | context.rowset_state = PREPARED; |
593 | 0 | context.segments_overlap = OVERLAPPING; |
594 | | // During a partial update, the extracted columns of a variant should not be included in the tablet schema. |
595 | | // This is because the partial update for a variant needs to ignore the extracted columns. |
596 | | // Otherwise, the schema types in different rowsets might be inconsistent. When performing a partial update, |
597 | | // the complete variant is constructed by reading all the sub-columns of the variant. |
598 | 0 | context.tablet_schema = rowset.tablet_schema()->copy_without_variant_extracted_columns(); |
599 | 0 | context.newest_write_timestamp = UnixSeconds(); |
600 | 0 | context.tablet_id = table_id(); |
601 | 0 | context.enable_segcompaction = false; |
602 | 0 | context.write_type = DataWriteType::TYPE_DIRECT; |
603 | 0 | context.partial_update_info = std::move(partial_update_info); |
604 | 0 | context.is_transient_rowset_writer = true; |
605 | 0 | context.rowset_id = rowset.rowset_id(); |
606 | 0 | context.tablet_id = tablet_id(); |
607 | 0 | context.index_id = index_id(); |
608 | 0 | context.partition_id = partition_id(); |
609 | 0 | context.enable_unique_key_merge_on_write = enable_unique_key_merge_on_write(); |
610 | 0 | context.txn_expiration = txn_expiration; |
611 | |
|
612 | 0 | auto storage_resource = rowset.rowset_meta()->remote_storage_resource(); |
613 | 0 | if (!storage_resource) { |
614 | 0 | return ResultError(std::move(storage_resource.error())); |
615 | 0 | } |
616 | | |
617 | 0 | context.storage_resource = *storage_resource.value(); |
618 | |
|
619 | 0 | return RowsetFactory::create_rowset_writer(_engine, context, false) |
620 | 0 | .transform([&](auto&& writer) { |
621 | 0 | writer->set_segment_start_id(cast_set<int32_t>(rowset.num_segments())); |
622 | 0 | return writer; |
623 | 0 | }); |
624 | 0 | } |
625 | | |
626 | 3 | int64_t CloudTablet::get_cloud_base_compaction_score() const { |
627 | 3 | if (_tablet_meta->compaction_policy() == CUMULATIVE_TIME_SERIES_POLICY) { |
628 | 0 | bool has_delete = false; |
629 | 0 | int64_t point = cumulative_layer_point(); |
630 | 0 | std::shared_lock<std::shared_mutex> rlock(_meta_lock); |
631 | 0 | for (const auto& rs_meta : _tablet_meta->all_rs_metas()) { |
632 | 0 | if (rs_meta->start_version() >= point) { |
633 | 0 | continue; |
634 | 0 | } |
635 | 0 | if (rs_meta->has_delete_predicate()) { |
636 | 0 | has_delete = true; |
637 | 0 | break; |
638 | 0 | } |
639 | 0 | } |
640 | 0 | if (!has_delete) { |
641 | 0 | return 0; |
642 | 0 | } |
643 | 0 | } |
644 | | |
645 | 3 | return _approximate_num_rowsets.load(std::memory_order_relaxed) - |
646 | 3 | _approximate_cumu_num_rowsets.load(std::memory_order_relaxed); |
647 | 3 | } |
648 | | |
649 | 1 | int64_t CloudTablet::get_cloud_cumu_compaction_score() const { |
650 | | // TODO(plat1ko): Propose an algorithm that considers tablet's key type, number of delete rowsets, |
651 | | // number of tablet versions simultaneously. |
652 | 1 | return _approximate_cumu_num_deltas.load(std::memory_order_relaxed); |
653 | 1 | } |
654 | | |
655 | | // return a json string to show the compaction status of this tablet |
656 | 0 | void CloudTablet::get_compaction_status(std::string* json_result) { |
657 | 0 | rapidjson::Document root; |
658 | 0 | root.SetObject(); |
659 | |
|
660 | 0 | rapidjson::Document path_arr; |
661 | 0 | path_arr.SetArray(); |
662 | |
|
663 | 0 | std::vector<RowsetSharedPtr> rowsets; |
664 | 0 | std::vector<RowsetSharedPtr> stale_rowsets; |
665 | 0 | { |
666 | 0 | std::shared_lock rdlock(_meta_lock); |
667 | 0 | rowsets.reserve(_rs_version_map.size()); |
668 | 0 | for (auto& it : _rs_version_map) { |
669 | 0 | rowsets.push_back(it.second); |
670 | 0 | } |
671 | 0 | stale_rowsets.reserve(_stale_rs_version_map.size()); |
672 | 0 | for (auto& it : _stale_rs_version_map) { |
673 | 0 | stale_rowsets.push_back(it.second); |
674 | 0 | } |
675 | 0 | } |
676 | 0 | std::sort(rowsets.begin(), rowsets.end(), Rowset::comparator); |
677 | 0 | std::sort(stale_rowsets.begin(), stale_rowsets.end(), Rowset::comparator); |
678 | | |
679 | | // get snapshot version path json_doc |
680 | 0 | _timestamped_version_tracker.get_stale_version_path_json_doc(path_arr); |
681 | 0 | root.AddMember("cumulative point", _cumulative_point.load(), root.GetAllocator()); |
682 | 0 | rapidjson::Value cumu_value; |
683 | 0 | std::string format_str = ToStringFromUnixMillis(_last_cumu_compaction_failure_millis.load()); |
684 | 0 | cumu_value.SetString(format_str.c_str(), cast_set<uint>(format_str.length()), |
685 | 0 | root.GetAllocator()); |
686 | 0 | root.AddMember("last cumulative failure time", cumu_value, root.GetAllocator()); |
687 | 0 | rapidjson::Value base_value; |
688 | 0 | format_str = ToStringFromUnixMillis(_last_base_compaction_failure_millis.load()); |
689 | 0 | base_value.SetString(format_str.c_str(), cast_set<uint>(format_str.length()), |
690 | 0 | root.GetAllocator()); |
691 | 0 | root.AddMember("last base failure time", base_value, root.GetAllocator()); |
692 | 0 | rapidjson::Value full_value; |
693 | 0 | format_str = ToStringFromUnixMillis(_last_full_compaction_failure_millis.load()); |
694 | 0 | full_value.SetString(format_str.c_str(), cast_set<uint>(format_str.length()), |
695 | 0 | root.GetAllocator()); |
696 | 0 | root.AddMember("last full failure time", full_value, root.GetAllocator()); |
697 | 0 | rapidjson::Value cumu_success_value; |
698 | 0 | format_str = ToStringFromUnixMillis(_last_cumu_compaction_success_millis.load()); |
699 | 0 | cumu_success_value.SetString(format_str.c_str(), cast_set<uint>(format_str.length()), |
700 | 0 | root.GetAllocator()); |
701 | 0 | root.AddMember("last cumulative success time", cumu_success_value, root.GetAllocator()); |
702 | 0 | rapidjson::Value base_success_value; |
703 | 0 | format_str = ToStringFromUnixMillis(_last_base_compaction_success_millis.load()); |
704 | 0 | base_success_value.SetString(format_str.c_str(), cast_set<uint>(format_str.length()), |
705 | 0 | root.GetAllocator()); |
706 | 0 | root.AddMember("last base success time", base_success_value, root.GetAllocator()); |
707 | 0 | rapidjson::Value full_success_value; |
708 | 0 | format_str = ToStringFromUnixMillis(_last_full_compaction_success_millis.load()); |
709 | 0 | full_success_value.SetString(format_str.c_str(), cast_set<uint>(format_str.length()), |
710 | 0 | root.GetAllocator()); |
711 | 0 | root.AddMember("last full success time", full_success_value, root.GetAllocator()); |
712 | 0 | rapidjson::Value cumu_schedule_value; |
713 | 0 | format_str = ToStringFromUnixMillis(_last_cumu_compaction_schedule_millis.load()); |
714 | 0 | cumu_schedule_value.SetString(format_str.c_str(), cast_set<uint>(format_str.length()), |
715 | 0 | root.GetAllocator()); |
716 | 0 | root.AddMember("last cumulative schedule time", cumu_schedule_value, root.GetAllocator()); |
717 | 0 | rapidjson::Value base_schedule_value; |
718 | 0 | format_str = ToStringFromUnixMillis(_last_base_compaction_schedule_millis.load()); |
719 | 0 | base_schedule_value.SetString(format_str.c_str(), cast_set<uint>(format_str.length()), |
720 | 0 | root.GetAllocator()); |
721 | 0 | root.AddMember("last base schedule time", base_schedule_value, root.GetAllocator()); |
722 | 0 | rapidjson::Value full_schedule_value; |
723 | 0 | format_str = ToStringFromUnixMillis(_last_full_compaction_schedule_millis.load()); |
724 | 0 | full_schedule_value.SetString(format_str.c_str(), cast_set<uint>(format_str.length()), |
725 | 0 | root.GetAllocator()); |
726 | 0 | root.AddMember("last full schedule time", full_schedule_value, root.GetAllocator()); |
727 | 0 | rapidjson::Value cumu_compaction_status_value; |
728 | 0 | cumu_compaction_status_value.SetString(_last_cumu_compaction_status.c_str(), |
729 | 0 | cast_set<uint>(_last_cumu_compaction_status.length()), |
730 | 0 | root.GetAllocator()); |
731 | 0 | root.AddMember("last cumulative status", cumu_compaction_status_value, root.GetAllocator()); |
732 | 0 | rapidjson::Value base_compaction_status_value; |
733 | 0 | base_compaction_status_value.SetString(_last_base_compaction_status.c_str(), |
734 | 0 | cast_set<uint>(_last_base_compaction_status.length()), |
735 | 0 | root.GetAllocator()); |
736 | 0 | root.AddMember("last base status", base_compaction_status_value, root.GetAllocator()); |
737 | 0 | rapidjson::Value full_compaction_status_value; |
738 | 0 | full_compaction_status_value.SetString(_last_full_compaction_status.c_str(), |
739 | 0 | cast_set<uint>(_last_full_compaction_status.length()), |
740 | 0 | root.GetAllocator()); |
741 | 0 | root.AddMember("last full status", full_compaction_status_value, root.GetAllocator()); |
742 | | |
743 | | // print all rowsets' version as an array |
744 | 0 | rapidjson::Document versions_arr; |
745 | 0 | rapidjson::Document missing_versions_arr; |
746 | 0 | versions_arr.SetArray(); |
747 | 0 | missing_versions_arr.SetArray(); |
748 | 0 | int64_t last_version = -1; |
749 | 0 | for (auto& rowset : rowsets) { |
750 | 0 | const Version& ver = rowset->version(); |
751 | 0 | if (ver.first != last_version + 1) { |
752 | 0 | rapidjson::Value miss_value; |
753 | 0 | miss_value.SetString(fmt::format("[{}-{}]", last_version + 1, ver.first - 1).c_str(), |
754 | 0 | missing_versions_arr.GetAllocator()); |
755 | 0 | missing_versions_arr.PushBack(miss_value, missing_versions_arr.GetAllocator()); |
756 | 0 | } |
757 | 0 | rapidjson::Value value; |
758 | 0 | std::string version_str = rowset->get_rowset_info_str(); |
759 | 0 | value.SetString(version_str.c_str(), cast_set<uint32_t>(version_str.length()), |
760 | 0 | versions_arr.GetAllocator()); |
761 | 0 | versions_arr.PushBack(value, versions_arr.GetAllocator()); |
762 | 0 | last_version = ver.second; |
763 | 0 | } |
764 | 0 | root.AddMember("rowsets", versions_arr, root.GetAllocator()); |
765 | 0 | root.AddMember("missing_rowsets", missing_versions_arr, root.GetAllocator()); |
766 | | |
767 | | // print all stale rowsets' version as an array |
768 | 0 | rapidjson::Document stale_versions_arr; |
769 | 0 | stale_versions_arr.SetArray(); |
770 | 0 | for (auto& rowset : stale_rowsets) { |
771 | 0 | rapidjson::Value value; |
772 | 0 | std::string version_str = rowset->get_rowset_info_str(); |
773 | 0 | value.SetString(version_str.c_str(), cast_set<uint32_t>(version_str.length()), |
774 | 0 | stale_versions_arr.GetAllocator()); |
775 | 0 | stale_versions_arr.PushBack(value, stale_versions_arr.GetAllocator()); |
776 | 0 | } |
777 | 0 | root.AddMember("stale_rowsets", stale_versions_arr, root.GetAllocator()); |
778 | | |
779 | | // add stale version rowsets |
780 | 0 | root.AddMember("stale version path", path_arr, root.GetAllocator()); |
781 | | |
782 | | // to json string |
783 | 0 | rapidjson::StringBuffer strbuf; |
784 | 0 | rapidjson::PrettyWriter<rapidjson::StringBuffer> writer(strbuf); |
785 | 0 | root.Accept(writer); |
786 | 0 | *json_result = std::string(strbuf.GetString()); |
787 | 0 | } |
788 | | |
789 | 0 | void CloudTablet::set_cumulative_layer_point(int64_t new_point) { |
790 | 0 | if (new_point == Tablet::K_INVALID_CUMULATIVE_POINT || new_point >= _cumulative_point) { |
791 | 0 | _cumulative_point = new_point; |
792 | 0 | return; |
793 | 0 | } |
794 | | // cumulative point should only be reset to -1, or be increased |
795 | | // FIXME: could happen in currently unresolved race conditions |
796 | 0 | LOG(WARNING) << "Unexpected cumulative point: " << new_point |
797 | 0 | << ", origin: " << _cumulative_point.load(); |
798 | 0 | } |
799 | | |
800 | 0 | std::vector<RowsetSharedPtr> CloudTablet::pick_candidate_rowsets_to_base_compaction() { |
801 | 0 | std::vector<RowsetSharedPtr> candidate_rowsets; |
802 | 0 | { |
803 | 0 | std::shared_lock rlock(_meta_lock); |
804 | 0 | for (const auto& [version, rs] : _rs_version_map) { |
805 | 0 | if (version.first != 0 && version.first < _cumulative_point && |
806 | 0 | (_alter_version == -1 || version.second <= _alter_version)) { |
807 | 0 | candidate_rowsets.push_back(rs); |
808 | 0 | } |
809 | 0 | } |
810 | 0 | } |
811 | 0 | std::sort(candidate_rowsets.begin(), candidate_rowsets.end(), Rowset::comparator); |
812 | 0 | return candidate_rowsets; |
813 | 0 | } |
814 | | |
815 | 0 | std::vector<RowsetSharedPtr> CloudTablet::pick_candidate_rowsets_to_full_compaction() { |
816 | 0 | std::vector<RowsetSharedPtr> candidate_rowsets; |
817 | 0 | { |
818 | 0 | std::shared_lock rlock(_meta_lock); |
819 | 0 | for (auto& [v, rs] : _rs_version_map) { |
820 | | // MUST NOT compact rowset [0-1] for some historical reasons (see cloud_schema_change) |
821 | 0 | if (v.first != 0) { |
822 | 0 | candidate_rowsets.push_back(rs); |
823 | 0 | } |
824 | 0 | } |
825 | 0 | } |
826 | 0 | std::sort(candidate_rowsets.begin(), candidate_rowsets.end(), Rowset::comparator); |
827 | 0 | return candidate_rowsets; |
828 | 0 | } |
829 | | |
830 | 0 | CalcDeleteBitmapExecutor* CloudTablet::calc_delete_bitmap_executor() { |
831 | 0 | return _engine.calc_delete_bitmap_executor(); |
832 | 0 | } |
833 | | |
834 | | Status CloudTablet::save_delete_bitmap(const TabletTxnInfo* txn_info, int64_t txn_id, |
835 | | DeleteBitmapPtr delete_bitmap, RowsetWriter* rowset_writer, |
836 | | const RowsetIdUnorderedSet& cur_rowset_ids, int64_t lock_id, |
837 | 0 | int64_t next_visible_version) { |
838 | 0 | RowsetSharedPtr rowset = txn_info->rowset; |
839 | 0 | int64_t cur_version = rowset->start_version(); |
840 | | // update delete bitmap info, in order to avoid recalculation when trying again |
841 | 0 | RETURN_IF_ERROR(_engine.txn_delete_bitmap_cache().update_tablet_txn_info( |
842 | 0 | txn_id, tablet_id(), delete_bitmap, cur_rowset_ids, PublishStatus::PREPARE)); |
843 | | |
844 | 0 | if (txn_info->partial_update_info && txn_info->partial_update_info->is_partial_update() && |
845 | 0 | rowset_writer->num_rows() > 0) { |
846 | 0 | DBUG_EXECUTE_IF("CloudTablet::save_delete_bitmap.update_tmp_rowset.error", { |
847 | 0 | return Status::InternalError<false>("injected update_tmp_rowset error."); |
848 | 0 | }); |
849 | 0 | const auto& rowset_meta = rowset->rowset_meta(); |
850 | 0 | RETURN_IF_ERROR(_engine.meta_mgr().update_tmp_rowset(*rowset_meta)); |
851 | 0 | } |
852 | | |
853 | 0 | RETURN_IF_ERROR(save_delete_bitmap_to_ms(cur_version, txn_id, delete_bitmap, lock_id, |
854 | 0 | next_visible_version)); |
855 | | |
856 | | // store the delete bitmap with sentinel marks in txn_delete_bitmap_cache because if the txn is retried for some reason, |
857 | | // it will use the delete bitmap from txn_delete_bitmap_cache when re-calculating the delete bitmap, during which it will do |
858 | | // delete bitmap correctness check. If we store the new_delete_bitmap, the delete bitmap correctness check will fail |
859 | 0 | RETURN_IF_ERROR(_engine.txn_delete_bitmap_cache().update_tablet_txn_info( |
860 | 0 | txn_id, tablet_id(), delete_bitmap, cur_rowset_ids, PublishStatus::SUCCEED, |
861 | 0 | txn_info->publish_info)); |
862 | | |
863 | 0 | DBUG_EXECUTE_IF("CloudTablet::save_delete_bitmap.enable_sleep", { |
864 | 0 | auto sleep_sec = dp->param<int>("sleep", 5); |
865 | 0 | std::this_thread::sleep_for(std::chrono::seconds(sleep_sec)); |
866 | 0 | }); |
867 | |
|
868 | 0 | DBUG_EXECUTE_IF("CloudTablet::save_delete_bitmap.injected_error", { |
869 | 0 | auto retry = dp->param<bool>("retry", false); |
870 | 0 | auto sleep_sec = dp->param<int>("sleep", 0); |
871 | 0 | std::this_thread::sleep_for(std::chrono::seconds(sleep_sec)); |
872 | 0 | if (retry) { // return DELETE_BITMAP_LOCK_ERROR to let it retry |
873 | 0 | return Status::Error<ErrorCode::DELETE_BITMAP_LOCK_ERROR>( |
874 | 0 | "injected DELETE_BITMAP_LOCK_ERROR"); |
875 | 0 | } else { |
876 | 0 | return Status::InternalError<false>("injected non-retryable error"); |
877 | 0 | } |
878 | 0 | }); |
879 | |
|
880 | 0 | return Status::OK(); |
881 | 0 | } |
882 | | |
883 | | Status CloudTablet::save_delete_bitmap_to_ms(int64_t cur_version, int64_t txn_id, |
884 | | DeleteBitmapPtr delete_bitmap, int64_t lock_id, |
885 | 0 | int64_t next_visible_version) { |
886 | 0 | DeleteBitmapPtr new_delete_bitmap = std::make_shared<DeleteBitmap>(tablet_id()); |
887 | 0 | for (auto iter = delete_bitmap->delete_bitmap.begin(); |
888 | 0 | iter != delete_bitmap->delete_bitmap.end(); ++iter) { |
889 | | // skip sentinel mark, which is used for delete bitmap correctness check |
890 | 0 | if (std::get<1>(iter->first) != DeleteBitmap::INVALID_SEGMENT_ID) { |
891 | 0 | new_delete_bitmap->merge( |
892 | 0 | {std::get<0>(iter->first), std::get<1>(iter->first), cur_version}, |
893 | 0 | iter->second); |
894 | 0 | } |
895 | 0 | } |
896 | | // lock_id != -1 means this is in an explict txn |
897 | 0 | bool is_explicit_txn = (lock_id != -1); |
898 | 0 | auto ms_lock_id = !is_explicit_txn ? txn_id : lock_id; |
899 | |
|
900 | 0 | RETURN_IF_ERROR(_engine.meta_mgr().update_delete_bitmap(*this, ms_lock_id, LOAD_INITIATOR_ID, |
901 | 0 | new_delete_bitmap.get(), txn_id, |
902 | 0 | is_explicit_txn, next_visible_version)); |
903 | 0 | return Status::OK(); |
904 | 0 | } |
905 | | |
906 | 0 | Versions CloudTablet::calc_missed_versions(int64_t spec_version, Versions existing_versions) const { |
907 | 0 | DCHECK(spec_version > 0) << "invalid spec_version: " << spec_version; |
908 | | |
909 | | // sort the existing versions in ascending order |
910 | 0 | std::sort(existing_versions.begin(), existing_versions.end(), |
911 | 0 | [](const Version& a, const Version& b) { |
912 | | // simple because 2 versions are certainly not overlapping |
913 | 0 | return a.first < b.first; |
914 | 0 | }); |
915 | | |
916 | | // From the first version(=0), find the missing version until spec_version |
917 | 0 | int64_t last_version = -1; |
918 | 0 | Versions missed_versions; |
919 | 0 | for (const Version& version : existing_versions) { |
920 | 0 | if (version.first > last_version + 1) { |
921 | | // there is a hole between versions |
922 | 0 | missed_versions.emplace_back(last_version + 1, std::min(version.first, spec_version)); |
923 | 0 | } |
924 | 0 | last_version = version.second; |
925 | 0 | if (last_version >= spec_version) { |
926 | 0 | break; |
927 | 0 | } |
928 | 0 | } |
929 | 0 | if (last_version < spec_version) { |
930 | | // there is a hole between the last version and the specificed version. |
931 | 0 | missed_versions.emplace_back(last_version + 1, spec_version); |
932 | 0 | } |
933 | 0 | return missed_versions; |
934 | 0 | } |
935 | | |
936 | | Status CloudTablet::calc_delete_bitmap_for_compaction( |
937 | | const std::vector<RowsetSharedPtr>& input_rowsets, const RowsetSharedPtr& output_rowset, |
938 | | const RowIdConversion& rowid_conversion, ReaderType compaction_type, int64_t merged_rows, |
939 | | int64_t filtered_rows, int64_t initiator, DeleteBitmapPtr& output_rowset_delete_bitmap, |
940 | 0 | bool allow_delete_in_cumu_compaction, int64_t& get_delete_bitmap_lock_start_time) { |
941 | 0 | output_rowset_delete_bitmap = std::make_shared<DeleteBitmap>(tablet_id()); |
942 | 0 | std::unique_ptr<RowLocationSet> missed_rows; |
943 | 0 | if ((config::enable_missing_rows_correctness_check || |
944 | 0 | config::enable_mow_compaction_correctness_check_core || |
945 | 0 | config::enable_mow_compaction_correctness_check_fail) && |
946 | 0 | !allow_delete_in_cumu_compaction && |
947 | 0 | compaction_type == ReaderType::READER_CUMULATIVE_COMPACTION) { |
948 | 0 | missed_rows = std::make_unique<RowLocationSet>(); |
949 | 0 | LOG(INFO) << "RowLocation Set inited succ for tablet:" << tablet_id(); |
950 | 0 | } |
951 | |
|
952 | 0 | std::unique_ptr<std::map<RowsetSharedPtr, RowLocationPairList>> location_map; |
953 | 0 | if (config::enable_rowid_conversion_correctness_check && |
954 | 0 | tablet_schema()->cluster_key_uids().empty()) { |
955 | 0 | location_map = std::make_unique<std::map<RowsetSharedPtr, RowLocationPairList>>(); |
956 | 0 | LOG(INFO) << "Location Map inited succ for tablet:" << tablet_id(); |
957 | 0 | } |
958 | | |
959 | | // 1. calc delete bitmap for historical data |
960 | 0 | RETURN_IF_ERROR(_engine.meta_mgr().sync_tablet_rowsets(this)); |
961 | 0 | Version version = max_version(); |
962 | 0 | std::size_t missed_rows_size = 0; |
963 | 0 | calc_compaction_output_rowset_delete_bitmap( |
964 | 0 | input_rowsets, rowid_conversion, 0, version.second + 1, missed_rows.get(), |
965 | 0 | location_map.get(), tablet_meta()->delete_bitmap(), output_rowset_delete_bitmap.get()); |
966 | 0 | if (missed_rows) { |
967 | 0 | missed_rows_size = missed_rows->size(); |
968 | 0 | if (!allow_delete_in_cumu_compaction) { |
969 | 0 | if (compaction_type == ReaderType::READER_CUMULATIVE_COMPACTION && |
970 | 0 | tablet_state() == TABLET_RUNNING) { |
971 | 0 | if (merged_rows + filtered_rows >= 0 && |
972 | 0 | merged_rows + filtered_rows != missed_rows_size) { |
973 | 0 | std::string err_msg = fmt::format( |
974 | 0 | "cumulative compaction: the merged rows({}), the filtered rows({}) is " |
975 | 0 | "not equal to missed rows({}) in rowid conversion, tablet_id: {}, " |
976 | 0 | "table_id:{}", |
977 | 0 | merged_rows, filtered_rows, missed_rows_size, tablet_id(), table_id()); |
978 | 0 | LOG(WARNING) << err_msg; |
979 | 0 | if (config::enable_mow_compaction_correctness_check_core) { |
980 | 0 | CHECK(false) << err_msg; |
981 | 0 | } else if (config::enable_mow_compaction_correctness_check_fail) { |
982 | 0 | return Status::InternalError<false>(err_msg); |
983 | 0 | } else { |
984 | 0 | DCHECK(false) << err_msg; |
985 | 0 | } |
986 | 0 | } |
987 | 0 | } |
988 | 0 | } |
989 | 0 | } |
990 | 0 | if (location_map) { |
991 | 0 | RETURN_IF_ERROR(check_rowid_conversion(output_rowset, *location_map)); |
992 | 0 | location_map->clear(); |
993 | 0 | } |
994 | | |
995 | | // 2. calc delete bitmap for incremental data |
996 | 0 | int64_t t1 = MonotonicMicros(); |
997 | 0 | RETURN_IF_ERROR(_engine.meta_mgr().get_delete_bitmap_update_lock( |
998 | 0 | *this, COMPACTION_DELETE_BITMAP_LOCK_ID, initiator)); |
999 | 0 | int64_t t2 = MonotonicMicros(); |
1000 | 0 | if (compaction_type == ReaderType::READER_CUMULATIVE_COMPACTION) { |
1001 | 0 | g_cu_compaction_get_delete_bitmap_lock_time_ms << (t2 - t1) / 1000; |
1002 | 0 | } else if (compaction_type == ReaderType::READER_BASE_COMPACTION) { |
1003 | 0 | g_base_compaction_get_delete_bitmap_lock_time_ms << (t2 - t1) / 1000; |
1004 | 0 | } |
1005 | 0 | get_delete_bitmap_lock_start_time = t2; |
1006 | 0 | RETURN_IF_ERROR(_engine.meta_mgr().sync_tablet_rowsets(this)); |
1007 | 0 | int64_t t3 = MonotonicMicros(); |
1008 | |
|
1009 | 0 | calc_compaction_output_rowset_delete_bitmap( |
1010 | 0 | input_rowsets, rowid_conversion, version.second, UINT64_MAX, missed_rows.get(), |
1011 | 0 | location_map.get(), tablet_meta()->delete_bitmap(), output_rowset_delete_bitmap.get()); |
1012 | 0 | int64_t t4 = MonotonicMicros(); |
1013 | 0 | if (location_map) { |
1014 | 0 | RETURN_IF_ERROR(check_rowid_conversion(output_rowset, *location_map)); |
1015 | 0 | } |
1016 | 0 | int64_t t5 = MonotonicMicros(); |
1017 | 0 | if (missed_rows) { |
1018 | 0 | DCHECK_EQ(missed_rows->size(), missed_rows_size); |
1019 | 0 | if (missed_rows->size() != missed_rows_size) { |
1020 | 0 | LOG(WARNING) << "missed rows don't match, before: " << missed_rows_size |
1021 | 0 | << " after: " << missed_rows->size(); |
1022 | 0 | } |
1023 | 0 | } |
1024 | | |
1025 | | // 3. store delete bitmap |
1026 | 0 | auto st = _engine.meta_mgr().update_delete_bitmap(*this, -1, initiator, |
1027 | 0 | output_rowset_delete_bitmap.get()); |
1028 | 0 | int64_t t6 = MonotonicMicros(); |
1029 | 0 | LOG(INFO) << "calc_delete_bitmap_for_compaction, tablet_id=" << tablet_id() |
1030 | 0 | << ", get lock cost " << (t2 - t1) << " us, sync rowsets cost " << (t3 - t2) |
1031 | 0 | << " us, calc delete bitmap cost " << (t4 - t3) << " us, check rowid conversion cost " |
1032 | 0 | << (t5 - t4) << " us, store delete bitmap cost " << (t6 - t5) |
1033 | 0 | << " us, st=" << st.to_string(); |
1034 | 0 | return st; |
1035 | 0 | } |
1036 | | |
1037 | | void CloudTablet::agg_delete_bitmap_for_compaction( |
1038 | | int64_t start_version, int64_t end_version, const std::vector<RowsetSharedPtr>& pre_rowsets, |
1039 | | DeleteBitmapPtr& new_delete_bitmap, |
1040 | 0 | std::map<std::string, int64_t>& pre_rowset_to_versions) { |
1041 | 0 | for (auto& rowset : pre_rowsets) { |
1042 | 0 | for (uint32_t seg_id = 0; seg_id < rowset->num_segments(); ++seg_id) { |
1043 | 0 | auto d = tablet_meta()->delete_bitmap().get_agg_without_cache( |
1044 | 0 | {rowset->rowset_id(), seg_id, end_version}, start_version); |
1045 | 0 | if (d->isEmpty()) { |
1046 | 0 | continue; |
1047 | 0 | } |
1048 | 0 | VLOG_DEBUG << "agg delete bitmap for tablet_id=" << tablet_id() |
1049 | 0 | << ", rowset_id=" << rowset->rowset_id() << ", seg_id=" << seg_id |
1050 | 0 | << ", rowset_version=" << rowset->version().to_string() |
1051 | 0 | << ". compaction start_version=" << start_version |
1052 | 0 | << ", end_version=" << end_version |
1053 | 0 | << ". delete_bitmap cardinality=" << d->cardinality(); |
1054 | 0 | DeleteBitmap::BitmapKey end_key {rowset->rowset_id(), seg_id, end_version}; |
1055 | 0 | new_delete_bitmap->set(end_key, *d); |
1056 | 0 | pre_rowset_to_versions[rowset->rowset_id().to_string()] = rowset->version().second; |
1057 | 0 | } |
1058 | 0 | } |
1059 | 0 | } |
1060 | | |
1061 | 0 | Status CloudTablet::sync_meta() { |
1062 | 0 | if (!config::enable_file_cache) { |
1063 | 0 | return Status::OK(); |
1064 | 0 | } |
1065 | | |
1066 | 0 | TabletMetaSharedPtr tablet_meta; |
1067 | 0 | auto st = _engine.meta_mgr().get_tablet_meta(tablet_id(), &tablet_meta); |
1068 | 0 | if (!st.ok()) { |
1069 | 0 | if (st.is<ErrorCode::NOT_FOUND>()) { |
1070 | 0 | clear_cache(); |
1071 | 0 | } |
1072 | 0 | return st; |
1073 | 0 | } |
1074 | | |
1075 | 0 | auto new_ttl_seconds = tablet_meta->ttl_seconds(); |
1076 | 0 | if (_tablet_meta->ttl_seconds() != new_ttl_seconds) { |
1077 | 0 | _tablet_meta->set_ttl_seconds(new_ttl_seconds); |
1078 | 0 | int64_t cur_time = UnixSeconds(); |
1079 | 0 | std::shared_lock rlock(_meta_lock); |
1080 | 0 | for (auto& [_, rs] : _rs_version_map) { |
1081 | 0 | for (int seg_id = 0; seg_id < rs->num_segments(); ++seg_id) { |
1082 | 0 | int64_t new_expiration_time = |
1083 | 0 | new_ttl_seconds + rs->rowset_meta()->newest_write_timestamp(); |
1084 | 0 | new_expiration_time = new_expiration_time > cur_time ? new_expiration_time : 0; |
1085 | 0 | auto file_key = Segment::file_cache_key(rs->rowset_id().to_string(), seg_id); |
1086 | 0 | auto* file_cache = io::FileCacheFactory::instance()->get_by_path(file_key); |
1087 | 0 | file_cache->modify_expiration_time(file_key, new_expiration_time); |
1088 | 0 | } |
1089 | 0 | } |
1090 | 0 | } |
1091 | |
|
1092 | 0 | auto new_compaction_policy = tablet_meta->compaction_policy(); |
1093 | 0 | if (_tablet_meta->compaction_policy() != new_compaction_policy) { |
1094 | 0 | _tablet_meta->set_compaction_policy(new_compaction_policy); |
1095 | 0 | } |
1096 | 0 | auto new_time_series_compaction_goal_size_mbytes = |
1097 | 0 | tablet_meta->time_series_compaction_goal_size_mbytes(); |
1098 | 0 | if (_tablet_meta->time_series_compaction_goal_size_mbytes() != |
1099 | 0 | new_time_series_compaction_goal_size_mbytes) { |
1100 | 0 | _tablet_meta->set_time_series_compaction_goal_size_mbytes( |
1101 | 0 | new_time_series_compaction_goal_size_mbytes); |
1102 | 0 | } |
1103 | 0 | auto new_time_series_compaction_file_count_threshold = |
1104 | 0 | tablet_meta->time_series_compaction_file_count_threshold(); |
1105 | 0 | if (_tablet_meta->time_series_compaction_file_count_threshold() != |
1106 | 0 | new_time_series_compaction_file_count_threshold) { |
1107 | 0 | _tablet_meta->set_time_series_compaction_file_count_threshold( |
1108 | 0 | new_time_series_compaction_file_count_threshold); |
1109 | 0 | } |
1110 | 0 | auto new_time_series_compaction_time_threshold_seconds = |
1111 | 0 | tablet_meta->time_series_compaction_time_threshold_seconds(); |
1112 | 0 | if (_tablet_meta->time_series_compaction_time_threshold_seconds() != |
1113 | 0 | new_time_series_compaction_time_threshold_seconds) { |
1114 | 0 | _tablet_meta->set_time_series_compaction_time_threshold_seconds( |
1115 | 0 | new_time_series_compaction_time_threshold_seconds); |
1116 | 0 | } |
1117 | 0 | auto new_time_series_compaction_empty_rowsets_threshold = |
1118 | 0 | tablet_meta->time_series_compaction_empty_rowsets_threshold(); |
1119 | 0 | if (_tablet_meta->time_series_compaction_empty_rowsets_threshold() != |
1120 | 0 | new_time_series_compaction_empty_rowsets_threshold) { |
1121 | 0 | _tablet_meta->set_time_series_compaction_empty_rowsets_threshold( |
1122 | 0 | new_time_series_compaction_empty_rowsets_threshold); |
1123 | 0 | } |
1124 | 0 | auto new_time_series_compaction_level_threshold = |
1125 | 0 | tablet_meta->time_series_compaction_level_threshold(); |
1126 | 0 | if (_tablet_meta->time_series_compaction_level_threshold() != |
1127 | 0 | new_time_series_compaction_level_threshold) { |
1128 | 0 | _tablet_meta->set_time_series_compaction_level_threshold( |
1129 | 0 | new_time_series_compaction_level_threshold); |
1130 | 0 | } |
1131 | |
|
1132 | 0 | return Status::OK(); |
1133 | 0 | } |
1134 | | |
1135 | 0 | void CloudTablet::build_tablet_report_info(TTabletInfo* tablet_info) { |
1136 | 0 | std::shared_lock rdlock(_meta_lock); |
1137 | 0 | tablet_info->__set_total_version_count(_tablet_meta->version_count()); |
1138 | 0 | tablet_info->__set_tablet_id(_tablet_meta->tablet_id()); |
1139 | | // Currently, this information will not be used by the cloud report, |
1140 | | // but it may be used in the future. |
1141 | 0 | } |
1142 | | |
1143 | | Status CloudTablet::check_delete_bitmap_cache(int64_t txn_id, |
1144 | 0 | DeleteBitmap* expected_delete_bitmap) { |
1145 | 0 | DeleteBitmapPtr cached_delete_bitmap; |
1146 | 0 | CloudStorageEngine& engine = ExecEnv::GetInstance()->storage_engine().to_cloud(); |
1147 | 0 | Status st = engine.txn_delete_bitmap_cache().get_delete_bitmap( |
1148 | 0 | txn_id, tablet_id(), &cached_delete_bitmap, nullptr, nullptr); |
1149 | 0 | if (st.ok()) { |
1150 | 0 | bool res = (expected_delete_bitmap->cardinality() == cached_delete_bitmap->cardinality()); |
1151 | 0 | auto msg = fmt::format( |
1152 | 0 | "delete bitmap cache check failed, cur_cardinality={}, cached_cardinality={}" |
1153 | 0 | "txn_id={}, tablet_id={}", |
1154 | 0 | expected_delete_bitmap->cardinality(), cached_delete_bitmap->cardinality(), txn_id, |
1155 | 0 | tablet_id()); |
1156 | 0 | if (!res) { |
1157 | 0 | DCHECK(res) << msg; |
1158 | 0 | return Status::InternalError<false>(msg); |
1159 | 0 | } |
1160 | 0 | } |
1161 | 0 | return Status::OK(); |
1162 | 0 | } |
1163 | | |
1164 | | #include "common/compile_check_end.h" |
1165 | | } // namespace doris |