be/src/cloud/cloud_tablet.h
Line | Count | Source |
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 | | #pragma once |
19 | | |
20 | | #include <memory> |
21 | | |
22 | | #include "storage/partial_update_info.h" |
23 | | #include "storage/rowset/rowset.h" |
24 | | #include "storage/tablet/base_tablet.h" |
25 | | |
26 | | namespace doris { |
27 | | |
28 | | class CloudStorageEngine; |
29 | | |
30 | | enum class WarmUpTriggerSource : int { NONE, SYNC_ROWSET, EVENT_DRIVEN, JOB }; |
31 | | |
32 | | enum class WarmUpProgress : int { NONE, DOING, DONE }; |
33 | | |
34 | | struct WarmUpState { |
35 | | WarmUpTriggerSource trigger_source {WarmUpTriggerSource::NONE}; |
36 | | WarmUpProgress progress {WarmUpProgress::NONE}; |
37 | | |
38 | | bool operator==(const WarmUpState& other) const { |
39 | | return trigger_source == other.trigger_source && progress == other.progress; |
40 | | } |
41 | | }; |
42 | | |
43 | | struct SyncRowsetStats { |
44 | | int64_t get_remote_rowsets_num {0}; |
45 | | int64_t get_remote_rowsets_rpc_ns {0}; |
46 | | |
47 | | int64_t get_local_delete_bitmap_rowsets_num {0}; |
48 | | int64_t get_remote_delete_bitmap_rowsets_num {0}; |
49 | | int64_t get_remote_delete_bitmap_key_count {0}; |
50 | | int64_t get_remote_delete_bitmap_bytes {0}; |
51 | | int64_t get_remote_delete_bitmap_rpc_ns {0}; |
52 | | |
53 | | int64_t get_remote_tablet_meta_rpc_ns {0}; |
54 | | int64_t tablet_meta_cache_hit {0}; |
55 | | int64_t tablet_meta_cache_miss {0}; |
56 | | |
57 | | int64_t bthread_schedule_delay_ns {0}; |
58 | | int64_t meta_lock_wait_ns {0}; // _meta_lock (std::shared_mutex) wait across all acquisitions |
59 | | int64_t sync_meta_lock_wait_ns { |
60 | | 0}; // _sync_meta_lock (bthread::Mutex) wait across all acquisitions |
61 | | }; |
62 | | |
63 | | struct SyncOptions { |
64 | | bool warmup_delta_data = false; |
65 | | bool sync_delete_bitmap = true; |
66 | | bool full_sync = false; |
67 | | bool merge_schema = false; |
68 | | int64_t query_version = -1; |
69 | | }; |
70 | | |
71 | | struct RecycledRowsets { |
72 | | RowsetId rowset_id; |
73 | | int64_t num_segments; |
74 | | std::vector<std::string> index_file_names; |
75 | | }; |
76 | | |
77 | | class CloudTablet final : public BaseTablet { |
78 | | public: |
79 | | CloudTablet(CloudStorageEngine& engine, TabletMetaSharedPtr tablet_meta); |
80 | | |
81 | | ~CloudTablet() override; |
82 | | |
83 | | bool exceed_version_limit(int32_t limit) override; |
84 | | |
85 | | Result<std::unique_ptr<RowsetWriter>> create_rowset_writer(RowsetWriterContext& context, |
86 | | bool vertical) override; |
87 | | |
88 | | Status capture_rs_readers(const Version& spec_version, std::vector<RowSetSplits>* rs_splits, |
89 | | const CaptureRowsetOps& opts) override; |
90 | | |
91 | | [[nodiscard]] Result<std::vector<Version>> capture_consistent_versions_unlocked( |
92 | | const Version& version_range, const CaptureRowsetOps& options) const override; |
93 | | |
94 | | // Capture versions with cache preference optimization. |
95 | | // This method prioritizes using cached/warmed-up rowsets when building version paths, |
96 | | // avoiding cold data reads when possible. It uses capture_consistent_versions_prefer_cache |
97 | | // to find a consistent version path that prefers already warmed-up rowsets. |
98 | | Result<std::vector<Version>> capture_versions_prefer_cache(const Version& spec_version) const; |
99 | | |
100 | | // Capture versions with query freshness tolerance. |
101 | | // This method finds a consistent version path where all rowsets are warmed up, |
102 | | // but allows fallback to normal capture if there are newer rowsets that should be |
103 | | // visible (based on freshness tolerance) but haven't been warmed up yet. |
104 | | // For merge-on-write tables, uses special validation to ensure data correctness. |
105 | | // |
106 | | // IMPORTANT: The returned version may be smaller than the requested version if newer |
107 | | // data hasn't been warmed up yet. This can cause different tablets in the same query |
108 | | // to read from different versions, potentially leading to inconsistent query results. |
109 | | // |
110 | | // @param options.query_freshness_tolerance_ms: Time tolerance in milliseconds. Rowsets that |
111 | | // became visible within this time range (after current_time - query_freshness_tolerance_ms) |
112 | | // can be skipped if not warmed up. However, if older rowsets (before this time point) |
113 | | // are not warmed up, the method will fallback to normal capture. |
114 | | Result<std::vector<Version>> capture_versions_with_freshness_tolerance( |
115 | | const Version& spec_version, const CaptureRowsetOps& options) const; |
116 | | |
117 | 548k | size_t tablet_footprint() override { |
118 | 548k | return _approximate_data_size.load(std::memory_order_relaxed); |
119 | 548k | } |
120 | | |
121 | | std::string tablet_path() const override; |
122 | | |
123 | | // clang-format off |
124 | 2.23M | int64_t fetch_add_approximate_num_rowsets (int64_t x) { return _approximate_num_rowsets .fetch_add(x, std::memory_order_relaxed); } |
125 | 172k | int64_t fetch_add_approximate_num_segments(int64_t x) { return _approximate_num_segments.fetch_add(x, std::memory_order_relaxed); } |
126 | 172k | int64_t fetch_add_approximate_num_rows (int64_t x) { return _approximate_num_rows .fetch_add(x, std::memory_order_relaxed); } |
127 | 172k | int64_t fetch_add_approximate_data_size (int64_t x) { return _approximate_data_size .fetch_add(x, std::memory_order_relaxed); } |
128 | 189k | int64_t fetch_add_approximate_cumu_num_rowsets (int64_t x) { return _approximate_cumu_num_rowsets.fetch_add(x, std::memory_order_relaxed); } |
129 | 172k | int64_t fetch_add_approximate_cumu_num_deltas (int64_t x) { return _approximate_cumu_num_deltas.fetch_add(x, std::memory_order_relaxed); } |
130 | | // clang-format on |
131 | | |
132 | | // meta lock must be held when calling this function |
133 | | void reset_approximate_stats(int64_t num_rowsets, int64_t num_segments, int64_t num_rows, |
134 | | int64_t data_size); |
135 | | |
136 | | // return a json string to show the compaction status of this tablet |
137 | | void get_compaction_status(std::string* json_result); |
138 | | |
139 | | // Synchronize the rowsets from meta service. |
140 | | // If tablet state is not `TABLET_RUNNING`, sync tablet meta and all visible rowsets. |
141 | | // If `query_version` > 0 and local max_version of the tablet >= `query_version`, do nothing. |
142 | | // If 'need_download_data_async' is true, it means that we need to download the new version |
143 | | // rowsets datum async. |
144 | | Status sync_rowsets(const SyncOptions& options = {}, SyncRowsetStats* stats = nullptr); |
145 | | |
146 | | // Synchronize the tablet meta from meta service. |
147 | | Status sync_meta(); |
148 | | |
149 | | // If `version_overlap` is true, function will delete rowsets with overlapped version in this tablet. |
150 | | // If 'warmup_delta_data' is true, download the new version rowset data in background. |
151 | | // MUST hold EXCLUSIVE `_meta_lock`. |
152 | | // If 'need_download_data_async' is true, it means that we need to download the new version |
153 | | // rowsets datum async. |
154 | | void add_rowsets(std::vector<RowsetSharedPtr> to_add, bool version_overlap, |
155 | | std::unique_lock<std::shared_mutex>& meta_lock, |
156 | | bool warmup_delta_data = false); |
157 | | |
158 | | // MUST hold EXCLUSIVE `_meta_lock`. |
159 | | void delete_rowsets(const std::vector<RowsetSharedPtr>& to_delete, |
160 | | std::unique_lock<std::shared_mutex>& meta_lock); |
161 | | |
162 | | // Like delete_rowsets, but also removes edges from the version graph. |
163 | | // Used by schema change to prevent the greedy capture algorithm from |
164 | | // preferring stale compaction rowsets over individual SC output rowsets. |
165 | | // MUST hold EXCLUSIVE `_meta_lock`. |
166 | | void delete_rowsets_for_schema_change(const std::vector<RowsetSharedPtr>& to_delete, |
167 | | std::unique_lock<std::shared_mutex>& meta_lock, |
168 | | bool recycle_deleted_rowsets = true); |
169 | | |
170 | | // Replace local rowsets in [2, alter_version] with schema change output rowsets. |
171 | | // Existing SC output rowsets are kept; other local/double-write/compaction rowsets |
172 | | // in this version range are removed from both _rs_version_map and version graph. |
173 | | // recycle_deleted_rowsets should only be true for the real tablet; temporary |
174 | | // schema-change delete-bitmap tablets only need to normalize their local graph. |
175 | | // MUST hold EXCLUSIVE `_meta_lock`. |
176 | | void replace_rowsets_with_schema_change_output( |
177 | | const std::vector<RowsetSharedPtr>& output_rowsets, int64_t alter_version, |
178 | | std::unique_lock<std::shared_mutex>& meta_lock, const char* stage, |
179 | | bool recycle_deleted_rowsets); |
180 | | |
181 | | // When the tablet is dropped, we need to recycle cached data: |
182 | | // 1. The data in file cache |
183 | | // 2. The memory in tablet cache |
184 | | void clear_cache() override; |
185 | | |
186 | | // Return number of deleted stale rowsets |
187 | | uint64_t delete_expired_stale_rowsets(); |
188 | | |
189 | 577k | bool has_stale_rowsets() const { return !_stale_rs_version_map.empty(); } |
190 | | |
191 | | int64_t get_cloud_base_compaction_score() const; |
192 | | int64_t get_cloud_cumu_compaction_score() const; |
193 | | |
194 | 484k | int64_t max_version_unlocked() const override { return _max_version; } |
195 | 487k | int64_t base_compaction_cnt() const { return _base_compaction_cnt; } |
196 | 487k | int64_t cumulative_compaction_cnt() const { return _cumulative_compaction_cnt; } |
197 | 162k | int64_t full_compaction_cnt() const { return _full_compaction_cnt; } |
198 | 372k | int64_t cumulative_layer_point() const { |
199 | 372k | return _cumulative_point.load(std::memory_order_relaxed); |
200 | 372k | } |
201 | | |
202 | 163k | void set_base_compaction_cnt(int64_t cnt) { _base_compaction_cnt = cnt; } |
203 | 169k | void set_cumulative_compaction_cnt(int64_t cnt) { _cumulative_compaction_cnt = cnt; } |
204 | 162k | void set_full_compaction_cnt(int64_t cnt) { _full_compaction_cnt = cnt; } |
205 | | void set_cumulative_layer_point(int64_t new_point); |
206 | | |
207 | 204M | int64_t last_cumu_compaction_failure_time() { return _last_cumu_compaction_failure_millis; } |
208 | 391 | void set_last_cumu_compaction_failure_time(int64_t millis) { |
209 | 391 | _last_cumu_compaction_failure_millis = millis; |
210 | 391 | } |
211 | | |
212 | 55.0M | int64_t last_base_compaction_failure_time() { return _last_base_compaction_failure_millis; } |
213 | 6.22k | void set_last_base_compaction_failure_time(int64_t millis) { |
214 | 6.22k | _last_base_compaction_failure_millis = millis; |
215 | 6.22k | } |
216 | | |
217 | 0 | int64_t last_full_compaction_failure_time() { return _last_full_compaction_failure_millis; } |
218 | 2 | void set_last_full_compaction_failure_time(int64_t millis) { |
219 | 2 | _last_full_compaction_failure_millis = millis; |
220 | 2 | } |
221 | | |
222 | 6.84k | int64_t last_cumu_compaction_success_time() { return _last_cumu_compaction_success_millis; } |
223 | 12.6k | void set_last_cumu_compaction_success_time(int64_t millis) { |
224 | 12.6k | _last_cumu_compaction_success_millis = millis; |
225 | 12.6k | } |
226 | | |
227 | 6.84k | int64_t last_base_compaction_success_time() { return _last_base_compaction_success_millis; } |
228 | 6.89k | void set_last_base_compaction_success_time(int64_t millis) { |
229 | 6.89k | _last_base_compaction_success_millis = millis; |
230 | 6.89k | } |
231 | | |
232 | 6.80k | int64_t last_full_compaction_success_time() { return _last_full_compaction_success_millis; } |
233 | 6.90k | void set_last_full_compaction_success_time(int64_t millis) { |
234 | 6.90k | _last_full_compaction_success_millis = millis; |
235 | 6.90k | } |
236 | | |
237 | 0 | int64_t last_cumu_compaction_schedule_time() { return _last_cumu_compaction_schedule_millis; } |
238 | 393 | void set_last_cumu_compaction_schedule_time(int64_t millis) { |
239 | 393 | _last_cumu_compaction_schedule_millis = millis; |
240 | 393 | } |
241 | | |
242 | 0 | int64_t last_base_compaction_schedule_time() { return _last_base_compaction_schedule_millis; } |
243 | 2 | void set_last_base_compaction_schedule_time(int64_t millis) { |
244 | 2 | _last_base_compaction_schedule_millis = millis; |
245 | 2 | } |
246 | | |
247 | 0 | int64_t last_full_compaction_schedule_time() { return _last_full_compaction_schedule_millis; } |
248 | 95 | void set_last_full_compaction_schedule_time(int64_t millis) { |
249 | 95 | _last_full_compaction_schedule_millis = millis; |
250 | 95 | } |
251 | | |
252 | 101k | void set_last_cumu_compaction_status(std::string status) { |
253 | 101k | _last_cumu_compaction_status = std::move(status); |
254 | 101k | } |
255 | | |
256 | 0 | std::string get_last_cumu_compaction_status() { return _last_cumu_compaction_status; } |
257 | | |
258 | 3.22k | void set_last_base_compaction_status(std::string status) { |
259 | 3.22k | _last_base_compaction_status = std::move(status); |
260 | 3.22k | } |
261 | | |
262 | 0 | std::string get_last_base_compaction_status() { return _last_base_compaction_status; } |
263 | | |
264 | 189 | void set_last_full_compaction_status(std::string status) { |
265 | 189 | _last_full_compaction_status = std::move(status); |
266 | 189 | } |
267 | | |
268 | 0 | std::string get_last_full_compaction_status() { return _last_full_compaction_status; } |
269 | | |
270 | 153k | int64_t alter_version() const { return _alter_version; } |
271 | 80.7k | void set_alter_version(int64_t alter_version) { _alter_version = alter_version; } |
272 | | |
273 | | // Last active cluster info for compaction read-write separation |
274 | 259M | std::string last_active_cluster_id() const { |
275 | 259M | std::shared_lock lock(_cluster_info_mutex); |
276 | 259M | return _last_active_cluster_id; |
277 | 259M | } |
278 | | int64_t last_active_time_ms() const { |
279 | | std::shared_lock lock(_cluster_info_mutex); |
280 | | return _last_active_time_ms; |
281 | | } |
282 | 13.3k | void set_last_active_cluster_info(const std::string& cluster_id, int64_t time_ms) { |
283 | 13.3k | std::unique_lock lock(_cluster_info_mutex); |
284 | 13.3k | _last_active_cluster_id = cluster_id; |
285 | 13.3k | _last_active_time_ms = time_ms; |
286 | 13.3k | } |
287 | | |
288 | | std::vector<RowsetSharedPtr> pick_candidate_rowsets_to_base_compaction(); |
289 | | |
290 | 806k | inline Version max_version() const { |
291 | 806k | std::shared_lock rdlock(_meta_lock); |
292 | 806k | return _tablet_meta->max_version(); |
293 | 806k | } |
294 | | |
295 | 106k | int64_t base_size() const { return _base_size; } |
296 | | |
297 | | std::vector<RowsetSharedPtr> pick_candidate_rowsets_to_full_compaction(); |
298 | | Result<RowsetSharedPtr> pick_a_rowset_for_index_change(int schema_version, |
299 | | bool& is_base_rowset); |
300 | | Status check_rowset_schema_for_build_index(std::vector<TColumn>& columns, int schema_version); |
301 | | |
302 | 0 | std::mutex& get_base_compaction_lock() { return _base_compaction_lock; } |
303 | 0 | std::mutex& get_cumulative_compaction_lock() { return _cumulative_compaction_lock; } |
304 | | |
305 | | Result<std::unique_ptr<RowsetWriter>> create_transient_rowset_writer( |
306 | | const Rowset& rowset, std::shared_ptr<PartialUpdateInfo> partial_update_info, |
307 | | int64_t txn_expiration = 0) override; |
308 | | |
309 | | CalcDeleteBitmapExecutor* calc_delete_bitmap_executor() override; |
310 | | |
311 | | Status save_delete_bitmap(const TabletTxnInfo* txn_info, int64_t txn_id, |
312 | | DeleteBitmapPtr delete_bitmap, RowsetWriter* rowset_writer, |
313 | | const RowsetIdUnorderedSet& cur_rowset_ids, int64_t lock_id = -1, |
314 | | int64_t next_visible_version = -1) override; |
315 | | |
316 | | Status save_delete_bitmap_to_ms(int64_t cur_version, int64_t txn_id, |
317 | | DeleteBitmapPtr delete_bitmap, int64_t lock_id, |
318 | | int64_t next_visible_version, RowsetSharedPtr rowset); |
319 | | |
320 | | Status calc_delete_bitmap_for_compaction(const std::vector<RowsetSharedPtr>& input_rowsets, |
321 | | const RowsetSharedPtr& output_rowset, |
322 | | const RowIdConversion& rowid_conversion, |
323 | | ReaderType compaction_type, int64_t merged_rows, |
324 | | int64_t filtered_rows, int64_t initiator, |
325 | | DeleteBitmapPtr& output_rowset_delete_bitmap, |
326 | | bool allow_delete_in_cumu_compaction, |
327 | | int64_t& get_delete_bitmap_lock_start_time); |
328 | | |
329 | | // Find the missed versions until the spec_version. |
330 | | // |
331 | | // for example: |
332 | | // [0-4][5-5][8-8][9-9][14-14] |
333 | | // if spec_version = 12, it will return [6-7],[10-12] |
334 | | Versions calc_missed_versions(int64_t spec_version, Versions existing_versions) const override; |
335 | | |
336 | 46.7k | std::mutex& get_rowset_update_lock() { return _rowset_update_lock; } |
337 | | |
338 | 157k | bthread::Mutex& get_sync_meta_lock() { return _sync_meta_lock; } |
339 | | |
340 | 121k | const auto& rowset_map() const { return _rs_version_map; } |
341 | | |
342 | | int64_t last_sync_time_s = 0; |
343 | | int64_t last_load_time_ms = 0; |
344 | | int64_t last_base_compaction_success_time_ms = 0; |
345 | | int64_t last_cumu_compaction_success_time_ms = 0; |
346 | | int64_t last_cumu_no_suitable_version_ms = 0; |
347 | | int64_t last_access_time_ms = 0; |
348 | | |
349 | | std::atomic<int64_t> local_read_time_us = 0; |
350 | | std::atomic<int64_t> remote_read_time_us = 0; |
351 | | std::atomic<int64_t> exec_compaction_time_us = 0; |
352 | | |
353 | | void build_tablet_report_info(TTabletInfo* tablet_info); |
354 | | |
355 | | // check that if the delete bitmap in delete bitmap cache has the same cardinality with the expected_delete_bitmap's |
356 | | Status check_delete_bitmap_cache(int64_t txn_id, DeleteBitmap* expected_delete_bitmap) override; |
357 | | |
358 | | void agg_delete_bitmap_for_compaction(int64_t start_version, int64_t end_version, |
359 | | const std::vector<RowsetSharedPtr>& pre_rowsets, |
360 | | DeleteBitmapPtr& new_delete_bitmap, |
361 | | std::map<std::string, int64_t>& pre_rowset_to_versions); |
362 | | |
363 | | bool need_remove_unused_rowsets(); |
364 | | |
365 | | void add_unused_rowsets(const std::vector<RowsetSharedPtr>& rowsets); |
366 | | void remove_unused_rowsets(); |
367 | | |
368 | | // For each given rowset not in active use, clears its file cache and returns its |
369 | | // ID, segment count, and index file names as RecycledRowsets entries. |
370 | | static std::vector<RecycledRowsets> recycle_cached_data( |
371 | | const std::vector<RowsetSharedPtr>& rowsets); |
372 | | |
373 | | // Add warmup state management |
374 | | WarmUpState get_rowset_warmup_state(RowsetId rowset_id); |
375 | | bool add_rowset_warmup_state( |
376 | | const RowsetMeta& rowset, WarmUpTriggerSource source, |
377 | | std::chrono::steady_clock::time_point start_tp = std::chrono::steady_clock::now()); |
378 | | bool update_rowset_warmup_state_inverted_idx_num(WarmUpTriggerSource source, RowsetId rowset_id, |
379 | | int64_t delta); |
380 | | bool update_rowset_warmup_state_inverted_idx_num_unlocked(WarmUpTriggerSource source, |
381 | | RowsetId rowset_id, int64_t delta); |
382 | | WarmUpState complete_rowset_segment_warmup(WarmUpTriggerSource trigger_source, |
383 | | RowsetId rowset_id, Status status, |
384 | | int64_t segment_num, int64_t inverted_idx_num); |
385 | | |
386 | | bool is_rowset_warmed_up(const RowsetId& rowset_id) const; |
387 | | |
388 | | void add_warmed_up_rowset(const RowsetId& rowset_id); |
389 | | // Test helper: add a rowset to the warmup state map with DOING progress, |
390 | | // so that is_rowset_warmed_up() returns false for it. |
391 | | void add_not_warmed_up_rowset(const RowsetId& rowset_id); |
392 | | |
393 | | // Try to apply visible pending rowsets to tablet meta in version order |
394 | | // This should be called after receiving FE notification or when new rowsets are added |
395 | | // @return Status::OK() if successfully applied, error otherwise |
396 | | void apply_visible_pending_rowsets(); |
397 | | |
398 | | void try_make_committed_rs_visible(int64_t txn_id, int64_t visible_version, |
399 | | int64_t version_update_time_ms); |
400 | | void try_make_committed_rs_visible_for_mow(int64_t txn_id, int64_t visible_version, |
401 | | int64_t version_update_time_ms); |
402 | | |
403 | | void clear_unused_visible_pending_rowsets(); |
404 | | |
405 | 5 | std::string rowset_warmup_digest() const { |
406 | 5 | std::string res; |
407 | 104 | auto add_log = [&](const RowsetSharedPtr& rs) { |
408 | 104 | auto tmp = fmt::format("{}{}", rs->rowset_id().to_string(), rs->version().to_string()); |
409 | 104 | if (_rowset_warm_up_states.contains(rs->rowset_id())) { |
410 | 104 | tmp += fmt::format( |
411 | 104 | ", progress={}, segments_warmed_up={}/{}, inverted_idx_warmed_up={}/{}", |
412 | 104 | _rowset_warm_up_states.at(rs->rowset_id()).state.progress, |
413 | 104 | _rowset_warm_up_states.at(rs->rowset_id()).num_segments_warmed_up, |
414 | 104 | _rowset_warm_up_states.at(rs->rowset_id()).num_segments, |
415 | 104 | _rowset_warm_up_states.at(rs->rowset_id()).num_inverted_idx_warmed_up, |
416 | 104 | _rowset_warm_up_states.at(rs->rowset_id()).num_inverted_idx); |
417 | 104 | } |
418 | 104 | res += fmt::format("[{}],", tmp); |
419 | 104 | }; |
420 | 5 | traverse_rowsets_unlocked(add_log, true); |
421 | 5 | return res; |
422 | 5 | } |
423 | | |
424 | | private: |
425 | | // FIXME(plat1ko): No need to record base size if rowsets are ordered by version |
426 | | void update_base_size(const Rowset& rs); |
427 | | |
428 | | Status sync_if_not_running(SyncRowsetStats* stats = nullptr); |
429 | | |
430 | | bool add_rowset_warmup_state_unlocked( |
431 | | const RowsetMeta& rowset, WarmUpTriggerSource source, |
432 | | std::chrono::steady_clock::time_point start_tp = std::chrono::steady_clock::now()); |
433 | | |
434 | | // used by capture_rs_reader_xxx functions |
435 | | bool rowset_is_warmed_up_unlocked(int64_t start_version, int64_t end_version) const; |
436 | | |
437 | | // Check if a rowset should be visible but not warmed up within freshness tolerance |
438 | | bool _check_rowset_should_be_visible_but_not_warmed_up( |
439 | | const RowsetMetaSharedPtr& rs_meta, int64_t path_max_version, |
440 | | std::chrono::system_clock::time_point freshness_limit_tp) const; |
441 | | |
442 | | // Submit a segment download task for warming up |
443 | | void _submit_segment_download_task(const RowsetSharedPtr& rs, |
444 | | const StorageResource* storage_resource, int seg_id, |
445 | | int64_t expiration_time); |
446 | | |
447 | | // Submit an inverted index download task for warming up |
448 | | void _submit_inverted_index_download_task(const RowsetSharedPtr& rs, |
449 | | const StorageResource* storage_resource, |
450 | | const io::Path& idx_path, int64_t idx_size, |
451 | | int64_t expiration_time); |
452 | | |
453 | | // Add rowsets directly with warmup |
454 | | void _add_rowsets_directly(std::vector<RowsetSharedPtr>& rowsets, bool warmup_delta_data); |
455 | | |
456 | | CloudStorageEngine& _engine; |
457 | | |
458 | | // this mutex MUST ONLY be used when sync meta |
459 | | bthread::Mutex _sync_meta_lock; |
460 | | // ATTENTION: lock order should be: _sync_meta_lock -> _meta_lock |
461 | | |
462 | | std::atomic<int64_t> _cumulative_point {-1}; |
463 | | std::atomic<int64_t> _approximate_num_rowsets {-1}; |
464 | | std::atomic<int64_t> _approximate_num_segments {-1}; |
465 | | std::atomic<int64_t> _approximate_num_rows {-1}; |
466 | | std::atomic<int64_t> _approximate_data_size {-1}; |
467 | | std::atomic<int64_t> _approximate_cumu_num_rowsets {-1}; |
468 | | // Number of sorted arrays (e.g. for rowset with N segments, if rowset is overlapping, delta is N, otherwise 1) after cumu point |
469 | | std::atomic<int64_t> _approximate_cumu_num_deltas {-1}; |
470 | | |
471 | | // timestamp of last cumu compaction failure |
472 | | std::atomic<int64_t> _last_cumu_compaction_failure_millis; |
473 | | // timestamp of last base compaction failure |
474 | | std::atomic<int64_t> _last_base_compaction_failure_millis; |
475 | | // timestamp of last full compaction failure |
476 | | std::atomic<int64_t> _last_full_compaction_failure_millis; |
477 | | // timestamp of last cumu compaction success |
478 | | std::atomic<int64_t> _last_cumu_compaction_success_millis; |
479 | | // timestamp of last base compaction success |
480 | | std::atomic<int64_t> _last_base_compaction_success_millis; |
481 | | // timestamp of last full compaction success |
482 | | std::atomic<int64_t> _last_full_compaction_success_millis; |
483 | | // timestamp of last cumu compaction schedule time |
484 | | std::atomic<int64_t> _last_cumu_compaction_schedule_millis; |
485 | | // timestamp of last base compaction schedule time |
486 | | std::atomic<int64_t> _last_base_compaction_schedule_millis; |
487 | | // timestamp of last full compaction schedule time |
488 | | std::atomic<int64_t> _last_full_compaction_schedule_millis; |
489 | | |
490 | | std::string _last_cumu_compaction_status; |
491 | | std::string _last_base_compaction_status; |
492 | | std::string _last_full_compaction_status; |
493 | | |
494 | | int64_t _base_compaction_cnt = 0; |
495 | | int64_t _cumulative_compaction_cnt = 0; |
496 | | int64_t _full_compaction_cnt = 0; |
497 | | int64_t _max_version = -1; |
498 | | int64_t _base_size = 0; |
499 | | int64_t _alter_version = -1; |
500 | | |
501 | | std::mutex _base_compaction_lock; |
502 | | std::mutex _cumulative_compaction_lock; |
503 | | |
504 | | // To avoid multiple calc delete bitmap tasks on same (txn_id, tablet_id) with different |
505 | | // signatures being executed concurrently, we use _rowset_update_lock to serialize them |
506 | | mutable std::mutex _rowset_update_lock; |
507 | | |
508 | | // unused_rowsets, [start_version, end_version] |
509 | | std::mutex _gc_mutex; |
510 | | std::unordered_map<RowsetId, RowsetSharedPtr> _unused_rowsets; |
511 | | std::vector<std::pair<std::vector<RowsetId>, DeleteBitmapKeyRanges>> _unused_delete_bitmap; |
512 | | |
513 | | // for warm up states management |
514 | | struct RowsetWarmUpInfo { |
515 | | WarmUpState state; |
516 | | int64_t num_segments = 0; |
517 | | int64_t num_inverted_idx = 0; |
518 | | int64_t num_segments_warmed_up = 0; |
519 | | int64_t num_inverted_idx_warmed_up = 0; |
520 | | std::chrono::steady_clock::time_point start_tp; |
521 | | |
522 | 57.3k | void done(int64_t input_num_segments, int64_t input_num_inverted_idx) { |
523 | 57.3k | num_segments_warmed_up += input_num_segments; |
524 | 57.3k | num_inverted_idx_warmed_up += input_num_inverted_idx; |
525 | 57.3k | update_state(); |
526 | 57.3k | } |
527 | | |
528 | 57.4k | bool has_finished() const { |
529 | 57.4k | return (num_segments_warmed_up >= num_segments) && |
530 | 57.4k | (num_inverted_idx_warmed_up >= num_inverted_idx); |
531 | 57.4k | } |
532 | | |
533 | | void update_state(); |
534 | | }; |
535 | | std::unordered_map<RowsetId, RowsetWarmUpInfo> _rowset_warm_up_states; |
536 | | |
537 | | mutable std::shared_mutex _warmed_up_rowsets_mutex; |
538 | | std::unordered_set<RowsetId> _warmed_up_rowsets; |
539 | | |
540 | | // Cluster info for compaction read-write separation |
541 | | mutable std::shared_mutex _cluster_info_mutex; |
542 | | std::string _last_active_cluster_id; |
543 | | int64_t _last_active_time_ms {0}; |
544 | | |
545 | | // Map: version -> <rowset_meta, expiration_time> |
546 | | // Stores rowsets that have been notified by FE but not yet added to tablet meta |
547 | | // due to out-of-order notification or version discontinuity |
548 | | struct VisiblePendingRowset { |
549 | | const bool is_empty_rowset; |
550 | | const int64_t expiration_time; // seconds since epoch |
551 | | RowsetMetaSharedPtr rowset_meta; |
552 | | |
553 | | VisiblePendingRowset(RowsetMetaSharedPtr rowset_meta_, int64_t expiration_time_, |
554 | | bool is_empty_rowset_ = false) |
555 | 121k | : is_empty_rowset(is_empty_rowset_), |
556 | 121k | expiration_time(expiration_time_), |
557 | 121k | rowset_meta(std::move(rowset_meta_)) {} |
558 | | }; |
559 | | mutable std::mutex _visible_pending_rs_lock; |
560 | | std::map<int64_t, VisiblePendingRowset> _visible_pending_rs_map; |
561 | | }; |
562 | | |
563 | | using CloudTabletSPtr = std::shared_ptr<CloudTablet>; |
564 | | |
565 | | } // namespace doris |