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