Coverage Report

Created: 2026-09-15 08:49

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
be/src/storage/storage_engine.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 <butil/macros.h>
21
#include <bvar/bvar.h>
22
#include <gen_cpp/Types_types.h>
23
#include <gen_cpp/internal_service.pb.h>
24
#include <gen_cpp/olap_file.pb.h>
25
26
#include <atomic>
27
#include <condition_variable>
28
#include <cstdint>
29
#include <ctime>
30
#include <map>
31
#include <memory>
32
#include <mutex>
33
#include <set>
34
#include <shared_mutex>
35
#include <string>
36
#include <unordered_map>
37
#include <unordered_set>
38
#include <vector>
39
40
#include "agent/task_worker_pool.h"
41
#include "common/config.h"
42
#include "common/status.h"
43
#include "runtime/heartbeat_flags.h"
44
#include "storage/adaptive_thread_pool_controller.h"
45
#include "storage/compaction/compaction_permit_limiter.h"
46
#include "storage/delete/calc_delete_bitmap_executor.h"
47
#include "storage/olap_common.h"
48
#include "storage/options.h"
49
#include "storage/rowset/pending_rowset_helper.h"
50
#include "storage/rowset/rowset_fwd.h"
51
#include "storage/segment/segment.h"
52
#include "storage/tablet/tablet_fwd.h"
53
#include "storage/task/index_builder.h"
54
#include "util/countdown_latch.h"
55
56
namespace doris {
57
58
class DataDir;
59
class EngineTask;
60
class MemTableFlushExecutor;
61
class SegcompactionWorker;
62
class BaseCompaction;
63
class CumulativeCompaction;
64
class CumulativeCompactionPolicy;
65
class StreamLoadRecorder;
66
class TCloneReq;
67
class TCreateTabletReq;
68
class TabletManager;
69
class Thread;
70
class ThreadPool;
71
class TxnManager;
72
class ReportWorker;
73
class CreateTabletRRIdxCache;
74
struct DirInfo;
75
class SnapshotManager;
76
class WorkloadGroup;
77
78
using SegCompactionCandidates = std::vector<segment_v2::SegmentSharedPtr>;
79
using SegCompactionCandidatesSharedPtr = std::shared_ptr<SegCompactionCandidates>;
80
using CumuCompactionPolicyTable =
81
        std::unordered_map<std::string_view, std::shared_ptr<CumulativeCompactionPolicy>>;
82
83
class StorageEngine;
84
class CloudStorageEngine;
85
86
extern bvar::Status<int64_t> g_max_rowsets_with_useless_delete_bitmap;
87
extern bvar::Status<int64_t> g_max_rowsets_with_useless_delete_bitmap_version;
88
89
// StorageEngine singleton to manage all Table pointers.
90
// Providing add/drop/get operations.
91
// StorageEngine instance doesn't own the Table resources, just hold the pointer,
92
// allocation/deallocation must be done outside.
93
class BaseStorageEngine {
94
protected:
95
    enum Type : uint8_t {
96
        LOCAL, // Shared-nothing integrated compute and storage architecture
97
        CLOUD, // Separating compute and storage architecture
98
    };
99
    Type _type;
100
101
public:
102
    BaseStorageEngine(Type type, const UniqueId& backend_uid);
103
    virtual ~BaseStorageEngine();
104
105
    StorageEngine& to_local();
106
    CloudStorageEngine& to_cloud();
107
108
    virtual Status open() = 0;
109
    virtual void stop() = 0;
110
    virtual bool stopped() = 0;
111
112
    // start all background threads. This should be call after env is ready.
113
    virtual Status start_bg_threads(std::shared_ptr<WorkloadGroup> wg_sptr = nullptr) = 0;
114
115
    /* Parameters:
116
     * - tablet_id: the id of tablet to get
117
     * - sync_stats: the stats of sync rowset
118
     * - force_use_only_cached: whether only use cached tablet meta
119
     * - cache_on_miss: whether cache the tablet meta when missing in cache
120
     */
121
    virtual Result<BaseTabletSPtr> get_tablet(int64_t tablet_id,
122
                                              SyncRowsetStats* sync_stats = nullptr,
123
                                              bool force_use_only_cached = false,
124
                                              bool cache_on_miss = true) = 0;
125
126
    virtual Status get_tablet_meta(int64_t tablet_id, TabletMetaSharedPtr* tablet_meta,
127
                                   bool force_use_only_cached = false) = 0;
128
129
    void register_row_binlog_tablet(const BaseTabletSPtr& tablet);
130
    // Catalog discovery refreshes cached Cloud metadata too, including tablets with no recent reads.
131
    Status submit_row_binlog_ttl(int64_t tablet_id, bool refresh_cloud_meta = false);
132
133
    void register_report_listener(ReportWorker* listener);
134
    void deregister_report_listener(ReportWorker* listener);
135
    void notify_listeners();
136
    bool notify_listener(std::string_view name);
137
138
7
    void set_heartbeat_flags(HeartbeatFlags* heartbeat_flags) {
139
7
        _heartbeat_flags = heartbeat_flags;
140
7
    }
141
    virtual Status set_cluster_id(int32_t cluster_id) = 0;
142
7
    int32_t effective_cluster_id() const { return _effective_cluster_id; }
143
144
    RowsetId next_rowset_id();
145
146
176k
    MemTableFlushExecutor* memtable_flush_executor() { return _memtable_flush_executor.get(); }
147
56
    AdaptiveThreadPoolController* adaptive_thread_controller() {
148
56
        return &_adaptive_thread_controller;
149
56
    }
150
179k
    CalcDeleteBitmapExecutor* calc_delete_bitmap_executor() {
151
179k
        return _calc_delete_bitmap_executor.get();
152
179k
    }
153
154
58.5k
    CalcDeleteBitmapExecutor* calc_delete_bitmap_executor_for_load() {
155
58.5k
        return _calc_delete_bitmap_executor_for_load.get();
156
58.5k
    }
157
158
    int64_t memory_limitation_bytes_per_thread_for_schema_change() const;
159
160
5.51k
    int get_disk_num() { return _disk_num; }
161
162
    Status init_stream_load_recorder(const std::string& stream_load_record_path);
163
164
2.55k
    const std::shared_ptr<StreamLoadRecorder>& get_stream_load_recorder() {
165
2.55k
        return _stream_load_recorder;
166
2.55k
    }
167
168
protected:
169
    void _start_adaptive_thread_controller();
170
    void _gc_expired_id_file_map();
171
    void _gc_expired_id_file_map_thread_callback();
172
    bool _should_delay_large_task();
173
174
    int32_t _effective_cluster_id = -1;
175
    HeartbeatFlags* _heartbeat_flags = nullptr;
176
177
    // For task, tablet and disk report
178
    std::mutex _report_mtx;
179
    std::vector<ReportWorker*> _report_listeners;
180
181
    std::unique_ptr<RowsetIdGenerator> _rowset_id_generator;
182
    std::unique_ptr<MemTableFlushExecutor> _memtable_flush_executor;
183
    AdaptiveThreadPoolController _adaptive_thread_controller;
184
    std::unique_ptr<CalcDeleteBitmapExecutor> _calc_delete_bitmap_executor;
185
    std::unique_ptr<CalcDeleteBitmapExecutor> _calc_delete_bitmap_executor_for_load;
186
    CountDownLatch _stop_background_threads_latch;
187
188
    Status _start_row_binlog_ttl_scanner();
189
    void _stop_row_binlog_ttl_scanner();
190
    std::mutex _row_binlog_ttl_mutex;
191
    std::map<int64_t, std::weak_ptr<BaseTablet>> _row_binlog_ttl_tablets;
192
    std::unordered_set<int64_t> _row_binlog_ttl_pending;
193
    std::unique_ptr<ThreadPool> _row_binlog_ttl_prepare_pool;
194
    std::shared_ptr<Thread> _row_binlog_ttl_scan_thread;
195
196
    std::shared_ptr<Thread> _id_file_map_gc_thread;
197
198
    int64_t _memory_limitation_bytes_for_schema_change;
199
200
    int _disk_num {-1};
201
202
    std::shared_ptr<StreamLoadRecorder> _stream_load_recorder;
203
204
    std::shared_ptr<bvar::Status<size_t>> _tablet_max_delete_bitmap_score_metrics;
205
    std::shared_ptr<bvar::Status<size_t>> _tablet_max_base_rowset_delete_bitmap_score_metrics;
206
207
    std::unique_ptr<ThreadPool> _base_compaction_thread_pool;
208
    std::unique_ptr<ThreadPool> _cumu_compaction_thread_pool;
209
    std::unique_ptr<ThreadPool> _binlog_compaction_thread_pool;
210
    int _cumu_compaction_thread_pool_used_threads {0};
211
    int _cumu_compaction_thread_pool_small_tasks_running {0};
212
};
213
214
class CompactionSubmitRegistry {
215
    using TabletSet = std::unordered_set<TabletSharedPtr>;
216
    using Registry = std::map<DataDir*, TabletSet>;
217
218
public:
219
17.1k
    CompactionSubmitRegistry() = default;
220
    CompactionSubmitRegistry(CompactionSubmitRegistry&& r);
221
222
    // create a snapshot for current registry, operations to the snapshot can be lock-free.
223
    CompactionSubmitRegistry create_snapshot();
224
225
    void reset(const std::vector<DataDir*>& stores);
226
227
    uint32_t count_executing_compaction(DataDir* dir, CompactionType compaction_type);
228
    uint32_t count_executing_cumu_and_base(DataDir* dir);
229
230
    bool has_compaction_task(DataDir* dir, CompactionType compaction_type);
231
232
    bool insert(TabletSharedPtr tablet, CompactionType compaction_type);
233
234
    void remove(TabletSharedPtr tablet, CompactionType compaction_type,
235
                std::function<void()> wakeup_cb);
236
237
    void jsonfy_compaction_status(std::string* result);
238
239
    std::vector<TabletCompactionContext> pick_topn_tablets_for_compaction(
240
            TabletManager* tablet_mgr, DataDir* data_dir, CompactionType compaction_type,
241
            const CumuCompactionPolicyTable& cumu_compaction_policies,
242
            CompactionScoreStats* disk_score_stats);
243
244
private:
245
    TabletSet& _get_tablet_set(DataDir* dir, CompactionType compaction_type);
246
247
    std::mutex _tablet_submitted_compaction_mutex;
248
    Registry _tablet_submitted_cumu_compaction;
249
    Registry _tablet_submitted_base_compaction;
250
    Registry _tablet_submitted_full_compaction;
251
    Registry _tablet_submitted_binlog_compaction;
252
};
253
254
class StorageEngine final : public BaseStorageEngine {
255
public:
256
    StorageEngine(const EngineOptions& options);
257
    ~StorageEngine() override;
258
259
    Status open() override;
260
261
    Status create_tablet(const TCreateTabletReq& request, RuntimeProfile* profile);
262
263
    /* Parameters:
264
     * - tablet_id: the id of tablet to get
265
     * - sync_stats: the stats of sync rowset
266
     * - force_use_only_cached: whether only use cached tablet meta
267
     * - cache_on_miss: whether cache the tablet meta when missing in cache
268
     */
269
    Result<BaseTabletSPtr> get_tablet(int64_t tablet_id, SyncRowsetStats* sync_stats = nullptr,
270
                                      bool force_use_only_cached = false,
271
                                      bool cache_on_miss = true) override;
272
273
    Status get_tablet_meta(int64_t tablet_id, TabletMetaSharedPtr* tablet_meta,
274
                           bool force_use_only_cached = false) override;
275
276
    void clear_transaction_task(const TTransactionId transaction_id);
277
    void clear_transaction_task(const TTransactionId transaction_id,
278
                                const std::vector<TPartitionId>& partition_ids);
279
280
    std::vector<DataDir*> get_stores(bool include_unused = false);
281
282
    // get all info of root_path
283
    Status get_all_data_dir_info(std::vector<DataDirInfo>* data_dir_infos, bool need_update);
284
285
    static int64_t get_file_or_directory_size(const std::string& file_path);
286
287
    // get root path for creating tablet. The returned vector of root path should be round robin,
288
    // for avoiding that all the tablet would be deployed one disk.
289
    std::vector<DataDir*> get_stores_for_create_tablet(int64_t partition_id,
290
                                                       TStorageMedium::type storage_medium);
291
292
    DataDir* get_store(const std::string& path);
293
294
0
    uint32_t available_storage_medium_type_count() const {
295
0
        return _available_storage_medium_type_count;
296
0
    }
297
298
    Status set_cluster_id(int32_t cluster_id) override;
299
300
    void start_delete_unused_rowset();
301
    void add_unused_rowset(RowsetSharedPtr rowset);
302
    using DeleteBitmapKeyRanges =
303
            std::vector<std::tuple<DeleteBitmap::BitmapKey, DeleteBitmap::BitmapKey>>;
304
    void add_unused_delete_bitmap_key_ranges(int64_t tablet_id,
305
                                             const std::vector<RowsetId>& rowsets,
306
                                             const DeleteBitmapKeyRanges& key_ranges);
307
308
    // Obtain shard path for new tablet.
309
    //
310
    // @param [out] shard_path choose an available root_path to clone new tablet
311
    // @return error code
312
    Status obtain_shard_path(TStorageMedium::type storage_medium, int64_t path_hash,
313
                             std::string* shared_path, DataDir** store, int64_t partition_id);
314
315
    // Load new tablet to make it effective.
316
    //
317
    // @param [in] root_path specify root path of new tablet
318
    // @param [in] request specify new tablet info
319
    // @param [in] restore whether we're restoring a tablet from trash
320
    // @return OK if load tablet success
321
    Status load_header(const std::string& shard_path, const TCloneReq& request,
322
                       bool restore = false);
323
324
1.18M
    TabletManager* tablet_manager() { return _tablet_manager.get(); }
325
5.85k
    TxnManager* txn_manager() { return _txn_manager.get(); }
326
15
    SnapshotManager* snapshot_mgr() { return _snapshot_mgr.get(); }
327
    // Rowset garbage collection helpers
328
    bool check_rowset_id_in_unused_rowsets(const RowsetId& rowset_id);
329
84.0k
    PendingRowsetSet& pending_local_rowsets() { return _pending_local_rowsets; }
330
5
    PendingRowsetSet& pending_remote_rowsets() { return _pending_remote_rowsets; }
331
    PendingRowsetGuard add_pending_rowset(const RowsetWriterContext& ctx);
332
333
0
    RowsetTypePB default_rowset_type() const {
334
0
        if (_heartbeat_flags != nullptr && _heartbeat_flags->is_set_default_rowset_type_to_beta()) {
335
0
            return BETA_ROWSET;
336
0
        }
337
0
        return _default_rowset_type;
338
0
    }
339
340
    Status start_bg_threads(std::shared_ptr<WorkloadGroup> wg_sptr = nullptr) override;
341
342
    // clear trash and snapshot file
343
    // option: update disk usage after sweep
344
    Status start_trash_sweep(double* usage, bool ignore_guard = false);
345
346
    // Must call stop() before storage_engine is deconstructed
347
    void stop() override;
348
349
    void get_tablet_rowset_versions(const PGetTabletVersionsRequest* request,
350
                                    PGetTabletVersionsResponse* response);
351
352
    bool get_peers_replica_backends(int64_t tablet_id, std::vector<TBackend>* backends);
353
354
14
    const std::shared_ptr<StreamLoadRecorder>& get_stream_load_recorder() {
355
14
        return _stream_load_recorder;
356
14
    }
357
358
    void get_compaction_status_json(std::string* result);
359
360
    Status submit_compaction_task(TabletSharedPtr tablet, CompactionType compaction_type,
361
                                  bool force, bool eager = true, int trigger_method = 0);
362
    Status submit_seg_compaction_task(std::shared_ptr<SegcompactionWorker> worker,
363
                                      SegCompactionCandidatesSharedPtr segments);
364
365
104
    ThreadPool* tablet_publish_txn_thread_pool() { return _tablet_publish_txn_thread_pool.get(); }
366
7.86k
    bool stopped() override { return _stopped; }
367
368
    Status process_index_change_task(const TAlterInvertedIndexReq& reqest);
369
370
    void gc_binlogs(const std::unordered_map<int64_t, int64_t>& gc_tablet_infos);
371
372
    void add_async_publish_task(int64_t partition_id, int64_t tablet_id, int64_t publish_version,
373
                                int64_t transaction_id, bool is_recover, int64_t commit_tso);
374
    int64_t get_pending_publish_min_version(int64_t tablet_id);
375
376
    bool add_broken_path(std::string path);
377
    bool remove_broken_path(std::string path);
378
379
11
    std::set<std::string> get_broken_paths() { return _broken_paths; }
380
381
    Status submit_clone_task(Tablet* tablet, int64_t version);
382
383
    std::unordered_map<int64_t, std::unique_ptr<TaskWorkerPoolIf>>* workers;
384
385
25.3k
    int64_t get_compaction_num_per_round() const { return _compaction_num_per_round; }
386
387
#ifdef BE_TEST
388
    std::vector<TabletSharedPtr> generate_compaction_tasks_for_test(
389
            CompactionType compaction_type, std::vector<DataDir*>& data_dirs, bool check_score) {
390
        auto tablet_contexts = _generate_compaction_tasks(compaction_type, data_dirs, check_score);
391
        std::vector<TabletSharedPtr> tablets;
392
        tablets.reserve(tablet_contexts.size());
393
        for (auto& context : tablet_contexts) {
394
            tablets.emplace_back(std::move(context.tablet));
395
        }
396
        return tablets;
397
    }
398
399
    CompactionSubmitRegistry& compaction_submit_registry_for_test() {
400
        return _compaction_submit_registry;
401
    }
402
#endif
403
404
private:
405
    // Instance should be inited from `static open()`
406
    // MUST NOT be called in other circumstances.
407
    Status _open();
408
409
    Status _init_store_map();
410
411
    void _update_storage_medium_type_count();
412
413
    // Some check methods
414
    Status _check_file_descriptor_number();
415
    Status _check_all_root_path_cluster_id();
416
    Status _judge_and_update_effective_cluster_id(int32_t cluster_id);
417
418
    void _exit_if_too_many_disks_are_failed();
419
420
    void _clean_unused_txns();
421
422
    void _clean_unused_rowset_metas();
423
424
    void _clean_unused_binlog_metas();
425
426
    void _clean_unused_delete_bitmap();
427
428
    void _clean_unused_pending_publish_info();
429
430
    void _clean_unused_partial_update_info();
431
432
    Status _do_sweep(const std::string& scan_root, const time_t& local_tm_now,
433
                     const int32_t expire);
434
435
    // All these xxx_callback() functions are for Background threads
436
    // unused rowset monitor thread
437
    void _unused_rowset_monitor_thread_callback();
438
439
    // garbage sweep thread process function. clear snapshot and trash folder
440
    void _garbage_sweeper_thread_callback();
441
442
    // delete tablet with io error process function
443
    void _disk_stat_monitor_thread_callback();
444
445
    // path gc process function
446
    void _path_gc_thread_callback(DataDir* data_dir);
447
448
    void _tablet_path_check_callback();
449
450
    void _tablet_checkpoint_callback(const std::vector<DataDir*>& data_dirs);
451
452
    // parse the default rowset type config to RowsetTypePB
453
    void _parse_default_rowset_type();
454
455
    // Disk status monitoring. Monitoring unused_flag Road King's new corresponding root_path unused flag,
456
    // When the unused mark is detected, the corresponding table information is deleted from the memory, and the disk data does not move.
457
    // When the disk status is unusable, but the unused logo is not _push_tablet_into_submitted_compactiondetected, you need to download it from root_path
458
    // Reload the data.
459
    void _start_disk_stat_monitor();
460
461
    void _compaction_tasks_producer_callback();
462
    void _binlog_compaction_tasks_producer_callback();
463
464
    std::vector<TabletCompactionContext> _generate_compaction_tasks(
465
            CompactionType compaction_type, std::vector<DataDir*>& data_dirs, bool check_score);
466
    void _update_cumulative_compaction_policy();
467
    CumuCompactionPolicyTable _snapshot_cumulative_compaction_policy();
468
    std::shared_ptr<CumulativeCompactionPolicy> _get_cumulative_compaction_policy(
469
            std::string_view compaction_policy);
470
471
    void _pop_tablet_from_submitted_compaction(TabletSharedPtr tablet,
472
                                               CompactionType compaction_type);
473
474
    Status _submit_compaction_task(TabletSharedPtr tablet, CompactionType compaction_type,
475
                                   bool force, int trigger_method = 0);
476
477
    void _handle_compaction(TabletSharedPtr tablet, std::shared_ptr<CompactionMixin> compaction,
478
                            CompactionType compaction_type, int64_t permits, bool force,
479
                            int64_t compaction_id = 0);
480
481
    void _adjust_compaction_thread_num();
482
483
    void _cooldown_tasks_producer_callback();
484
    void _remove_unused_remote_files_callback();
485
    void do_remove_unused_remote_files();
486
    void _cold_data_compaction_producer_callback();
487
    void _handle_cold_data_compaction(TabletSharedPtr tablet);
488
    void _follow_cooldown_meta(TabletSharedPtr tablet);
489
490
    Status _handle_seg_compaction(std::shared_ptr<SegcompactionWorker> worker,
491
                                  SegCompactionCandidatesSharedPtr segments,
492
                                  uint64_t submission_time);
493
494
    Status _handle_index_change(IndexBuilderSharedPtr index_builder);
495
496
    void _gc_binlogs(int64_t tablet_id, int64_t version);
497
498
    void _async_publish_callback();
499
500
    void _process_async_publish();
501
502
    Status _persist_broken_paths();
503
504
    bool _increase_low_priority_task_nums(DataDir* dir);
505
506
    void _decrease_low_priority_task_nums(DataDir* dir);
507
508
    void _get_candidate_stores(TStorageMedium::type storage_medium,
509
                               std::vector<DirInfo>& dir_infos);
510
511
    int _get_and_set_next_disk_index(int64_t partition_id, TStorageMedium::type storage_medium);
512
513
    int32_t _auto_get_interval_by_disk_capacity(DataDir* data_dir);
514
515
    void _check_tablet_delete_bitmap_score_callback();
516
517
private:
518
    EngineOptions _options;
519
    std::mutex _store_lock;
520
    std::mutex _trash_sweep_lock;
521
    std::map<std::string, std::unique_ptr<DataDir>> _store_map;
522
    std::set<std::string> _broken_paths;
523
    std::mutex _broken_paths_mutex;
524
525
    uint32_t _available_storage_medium_type_count;
526
527
    bool _is_all_cluster_id_exist;
528
529
    std::atomic_bool _stopped {false};
530
531
    std::mutex _gc_mutex;
532
    std::unordered_map<RowsetId, RowsetSharedPtr> _unused_rowsets;
533
    // tablet_id, unused_rowsets, [start_version, end_version]
534
    std::vector<std::tuple<int64_t, std::vector<RowsetId>, DeleteBitmapKeyRanges>>
535
            _unused_delete_bitmap;
536
    PendingRowsetSet _pending_local_rowsets;
537
    PendingRowsetSet _pending_remote_rowsets;
538
539
    std::shared_ptr<Thread> _unused_rowset_monitor_thread;
540
    // thread to monitor snapshot expiry
541
    std::shared_ptr<Thread> _garbage_sweeper_thread;
542
    // thread to monitor disk stat
543
    std::shared_ptr<Thread> _disk_stat_monitor_thread;
544
    // thread to produce both base and cumulative compaction tasks
545
    std::shared_ptr<Thread> _compaction_tasks_producer_thread;
546
    std::shared_ptr<Thread> _binlog_compaction_tasks_producer_thread;
547
    std::shared_ptr<Thread> _cache_clean_thread;
548
    // threads to clean all file descriptor not actively in use
549
    std::vector<std::shared_ptr<Thread>> _path_gc_threads;
550
    // thread to produce tablet checkpoint tasks
551
    std::shared_ptr<Thread> _tablet_checkpoint_tasks_producer_thread;
552
    // thread to check tablet path
553
    std::shared_ptr<Thread> _tablet_path_check_thread;
554
    // thread to clean tablet lookup cache
555
    std::shared_ptr<Thread> _lookup_cache_clean_thread;
556
557
    std::mutex _engine_task_mutex;
558
559
    std::unique_ptr<TabletManager> _tablet_manager;
560
    std::unique_ptr<TxnManager> _txn_manager;
561
562
    // Used to control the migration from segment_v1 to segment_v2, can be deleted in futrue.
563
    // Type of new loaded data
564
    RowsetTypePB _default_rowset_type;
565
566
    std::unique_ptr<ThreadPool> _seg_compaction_thread_pool;
567
    std::unique_ptr<ThreadPool> _cold_data_compaction_thread_pool;
568
569
    std::unique_ptr<ThreadPool> _tablet_publish_txn_thread_pool;
570
571
    std::unique_ptr<ThreadPool> _tablet_meta_checkpoint_thread_pool;
572
573
    CompactionPermitLimiter _permit_limiter;
574
575
    CompactionSubmitRegistry _compaction_submit_registry;
576
577
    std::mutex _low_priority_task_nums_mutex;
578
    std::unordered_map<DataDir*, int32_t> _low_priority_task_nums;
579
580
    std::atomic<int32_t> _wakeup_producer_flag {0};
581
582
    std::mutex _compaction_producer_sleep_mutex;
583
    std::condition_variable _compaction_producer_sleep_cv;
584
585
    // we use unordered_map to store all cumulative compaction policy sharded ptr
586
    std::mutex _cumulative_compaction_policy_mtx;
587
    CumuCompactionPolicyTable _cumulative_compaction_policies;
588
589
    std::shared_ptr<Thread> _cooldown_tasks_producer_thread;
590
    std::shared_ptr<Thread> _remove_unused_remote_files_thread;
591
    std::shared_ptr<Thread> _cold_data_compaction_producer_thread;
592
593
    std::shared_ptr<Thread> _cache_file_cleaner_tasks_producer_thread;
594
595
    std::unique_ptr<PriorityThreadPool> _cooldown_thread_pool;
596
597
    std::mutex _running_cooldown_mutex;
598
    std::unordered_set<int64_t> _running_cooldown_tablets;
599
600
    std::mutex _cold_compaction_tablet_submitted_mtx;
601
    std::unordered_set<int64_t> _cold_compaction_tablet_submitted;
602
603
    std::mutex _cumu_compaction_delay_mtx;
604
605
    // tablet_id, publish_version, transaction_id, partition_id, commit_tso
606
    std::map<int64_t, std::map<int64_t, std::tuple<int64_t, int64_t, int64_t>>>
607
            _async_publish_tasks;
608
    // aync publish for discontinuous versions of merge_on_write table
609
    std::shared_ptr<Thread> _async_publish_thread;
610
    std::shared_mutex _async_publish_lock;
611
612
    std::atomic<bool> _need_clean_trash {false};
613
614
    // next index for create tablet
615
    std::map<TStorageMedium::type, int> _last_use_index;
616
617
    std::unique_ptr<CreateTabletRRIdxCache> _create_tablet_idx_lru_cache;
618
619
    std::unique_ptr<SnapshotManager> _snapshot_mgr;
620
621
    // thread to check tablet delete bitmap count tasks
622
    std::shared_ptr<Thread> _check_delete_bitmap_score_thread;
623
624
    int64_t _last_get_peers_replica_backends_time_ms {0};
625
626
    int64_t _compaction_num_per_round {1};
627
};
628
629
// lru cache for create tabelt round robin in disks
630
// key: partitionId_medium
631
// value: index
632
class CreateTabletRRIdxCache : public LRUCachePolicy {
633
public:
634
    // get key, delimiter with DELIMITER '-'
635
139
    static std::string get_key(int64_t partition_id, TStorageMedium::type medium) {
636
139
        return fmt::format("{}-{}", partition_id, medium);
637
139
    }
638
639
    // -1 not found key in lru
640
    int get_index(const std::string& key);
641
642
    void set_index(const std::string& key, int next_idx);
643
644
    class CacheValue : public LRUCacheValueBase {
645
    public:
646
        int idx = 0;
647
    };
648
649
    CreateTabletRRIdxCache(size_t capacity)
650
699
            : LRUCachePolicy(CachePolicy::CacheType::CREATE_TABLET_RR_IDX_CACHE, capacity,
651
699
                             LRUCacheType::NUMBER,
652
699
                             /*stale_sweep_time_s*/ 30 * 60, /*num shards*/ 1,
653
699
                             /*element count capacity */ 0,
654
699
                             /*enable prune*/ true, /*is lru-k*/ false) {}
655
};
656
657
struct DirInfo {
658
    DataDir* data_dir;
659
660
    double usage = 0;
661
    int available_level = 0;
662
663
5
    bool operator<(const DirInfo& other) const {
664
5
        if (available_level != other.available_level) {
665
0
            return available_level < other.available_level;
666
0
        }
667
5
        return data_dir->path_hash() < other.data_dir->path_hash();
668
5
    }
669
};
670
671
} // namespace doris