Coverage Report

Created: 2025-03-11 14:37

/root/doris/be/src/olap/olap_server.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 <gen_cpp/Types_types.h>
19
#include <gen_cpp/olap_file.pb.h>
20
#include <glog/logging.h>
21
#include <rapidjson/prettywriter.h>
22
#include <rapidjson/stringbuffer.h>
23
#include <stdint.h>
24
#include <sys/types.h>
25
26
#include <algorithm>
27
#include <atomic>
28
// IWYU pragma: no_include <bits/chrono.h>
29
#include <chrono> // IWYU pragma: keep
30
#include <cmath>
31
#include <condition_variable>
32
#include <cstdint>
33
#include <ctime>
34
#include <functional>
35
#include <map>
36
#include <memory>
37
#include <mutex>
38
#include <ostream>
39
#include <random>
40
#include <shared_mutex>
41
#include <string>
42
#include <thread>
43
#include <unordered_set>
44
#include <utility>
45
#include <vector>
46
47
#include "agent/utils.h"
48
#include "common/config.h"
49
#include "common/logging.h"
50
#include "common/status.h"
51
#include "cpp/sync_point.h"
52
#include "gen_cpp/FrontendService.h"
53
#include "gen_cpp/internal_service.pb.h"
54
#include "gutil/ref_counted.h"
55
#include "io/fs/file_writer.h" // IWYU pragma: keep
56
#include "io/fs/path.h"
57
#include "olap/base_tablet.h"
58
#include "olap/cold_data_compaction.h"
59
#include "olap/compaction_permit_limiter.h"
60
#include "olap/cumulative_compaction_policy.h"
61
#include "olap/cumulative_compaction_time_series_policy.h"
62
#include "olap/data_dir.h"
63
#include "olap/olap_common.h"
64
#include "olap/olap_define.h"
65
#include "olap/rowset/segcompaction.h"
66
#include "olap/schema_change.h"
67
#include "olap/single_replica_compaction.h"
68
#include "olap/storage_engine.h"
69
#include "olap/storage_policy.h"
70
#include "olap/tablet.h"
71
#include "olap/tablet_manager.h"
72
#include "olap/tablet_meta.h"
73
#include "olap/tablet_meta_manager.h"
74
#include "olap/tablet_schema.h"
75
#include "olap/task/engine_publish_version_task.h"
76
#include "olap/task/index_builder.h"
77
#include "runtime/client_cache.h"
78
#include "runtime/memory/cache_manager.h"
79
#include "runtime/memory/global_memory_arbitrator.h"
80
#include "util/countdown_latch.h"
81
#include "util/debug_points.h"
82
#include "util/doris_metrics.h"
83
#include "util/mem_info.h"
84
#include "util/thread.h"
85
#include "util/threadpool.h"
86
#include "util/thrift_rpc_helper.h"
87
#include "util/time.h"
88
#include "util/uid_util.h"
89
#include "util/work_thread_pool.hpp"
90
91
using std::string;
92
93
namespace doris {
94
95
using io::Path;
96
97
// number of running SCHEMA-CHANGE threads
98
volatile uint32_t g_schema_change_active_threads = 0;
99
100
static const uint64_t DEFAULT_SEED = 104729;
101
static const uint64_t MOD_PRIME = 7652413;
102
103
0
CompactionSubmitRegistry::CompactionSubmitRegistry(CompactionSubmitRegistry&& r) {
104
0
    std::swap(_tablet_submitted_cumu_compaction, r._tablet_submitted_cumu_compaction);
105
0
    std::swap(_tablet_submitted_base_compaction, r._tablet_submitted_base_compaction);
106
0
    std::swap(_tablet_submitted_full_compaction, r._tablet_submitted_full_compaction);
107
0
}
108
109
0
CompactionSubmitRegistry CompactionSubmitRegistry::create_snapshot() {
110
    // full compaction is not engaged in this method
111
0
    std::unique_lock<std::mutex> l(_tablet_submitted_compaction_mutex);
112
0
    CompactionSubmitRegistry registry;
113
0
    registry._tablet_submitted_base_compaction = _tablet_submitted_base_compaction;
114
0
    registry._tablet_submitted_cumu_compaction = _tablet_submitted_cumu_compaction;
115
0
    return registry;
116
0
}
117
118
0
void CompactionSubmitRegistry::reset(const std::vector<DataDir*>& stores) {
119
    // full compaction is not engaged in this method
120
0
    for (const auto& store : stores) {
121
0
        _tablet_submitted_cumu_compaction[store] = {};
122
0
        _tablet_submitted_base_compaction[store] = {};
123
0
    }
124
0
}
125
126
uint32_t CompactionSubmitRegistry::count_executing_compaction(DataDir* dir,
127
2
                                                              CompactionType compaction_type) {
128
    // non-lock, used in snapshot
129
2
    const auto& compaction_tasks = _get_tablet_set(dir, compaction_type);
130
10
    return std::count_if(compaction_tasks.begin(), compaction_tasks.end(), [](const auto& task) {
131
10
        return task->compaction_stage == CompactionStage::EXECUTING;
132
10
    });
133
2
}
134
135
1
uint32_t CompactionSubmitRegistry::count_executing_cumu_and_base(DataDir* dir) {
136
    // non-lock, used in snapshot
137
1
    return count_executing_compaction(dir, CompactionType::BASE_COMPACTION) +
138
1
           count_executing_compaction(dir, CompactionType::CUMULATIVE_COMPACTION);
139
1
}
140
141
0
bool CompactionSubmitRegistry::has_compaction_task(DataDir* dir, CompactionType compaction_type) {
142
    // non-lock, used in snapshot
143
0
    return !_get_tablet_set(dir, compaction_type).empty();
144
0
}
145
146
std::vector<TabletSharedPtr> CompactionSubmitRegistry::pick_topn_tablets_for_compaction(
147
        TabletManager* tablet_mgr, DataDir* data_dir, CompactionType compaction_type,
148
0
        const CumuCompactionPolicyTable& cumu_compaction_policies, uint32_t* disk_max_score) {
149
    // non-lock, used in snapshot
150
0
    return tablet_mgr->find_best_tablets_to_compaction(compaction_type, data_dir,
151
0
                                                       _get_tablet_set(data_dir, compaction_type),
152
0
                                                       disk_max_score, cumu_compaction_policies);
153
0
}
154
155
10
bool CompactionSubmitRegistry::insert(TabletSharedPtr tablet, CompactionType compaction_type) {
156
10
    std::unique_lock<std::mutex> l(_tablet_submitted_compaction_mutex);
157
10
    auto& tablet_set = _get_tablet_set(tablet->data_dir(), compaction_type);
158
10
    bool already_exist = !(tablet_set.insert(tablet).second);
159
10
    return already_exist;
160
10
}
161
162
void CompactionSubmitRegistry::remove(TabletSharedPtr tablet, CompactionType compaction_type,
163
2
                                      std::function<void()> wakeup_cb) {
164
2
    std::unique_lock<std::mutex> l(_tablet_submitted_compaction_mutex);
165
2
    auto& tablet_set = _get_tablet_set(tablet->data_dir(), compaction_type);
166
2
    size_t removed = tablet_set.erase(tablet);
167
2
    if (removed == 1) {
168
2
        wakeup_cb();
169
2
    }
170
2
}
171
172
CompactionSubmitRegistry::TabletSet& CompactionSubmitRegistry::_get_tablet_set(
173
14
        DataDir* dir, CompactionType compaction_type) {
174
14
    switch (compaction_type) {
175
1
    case CompactionType::BASE_COMPACTION:
176
1
        return _tablet_submitted_base_compaction[dir];
177
13
    case CompactionType::CUMULATIVE_COMPACTION:
178
13
        return _tablet_submitted_cumu_compaction[dir];
179
0
    case CompactionType::FULL_COMPACTION:
180
0
        return _tablet_submitted_full_compaction[dir];
181
0
    default:
182
0
        CHECK(false) << "invalid compaction type";
183
14
    }
184
14
}
185
186
0
static int32_t get_cumu_compaction_threads_num(size_t data_dirs_num) {
187
0
    int32_t threads_num = config::max_cumu_compaction_threads;
188
0
    if (threads_num == -1) {
189
0
        threads_num = data_dirs_num;
190
0
    }
191
0
    threads_num = threads_num <= 0 ? 1 : threads_num;
192
0
    return threads_num;
193
0
}
194
195
0
static int32_t get_base_compaction_threads_num(size_t data_dirs_num) {
196
0
    int32_t threads_num = config::max_base_compaction_threads;
197
0
    if (threads_num == -1) {
198
0
        threads_num = data_dirs_num;
199
0
    }
200
0
    threads_num = threads_num <= 0 ? 1 : threads_num;
201
0
    return threads_num;
202
0
}
203
204
0
static int32_t get_single_replica_compaction_threads_num(size_t data_dirs_num) {
205
0
    int32_t threads_num = config::max_single_replica_compaction_threads;
206
0
    if (threads_num == -1) {
207
0
        threads_num = data_dirs_num;
208
0
    }
209
0
    threads_num = threads_num <= 0 ? 1 : threads_num;
210
0
    return threads_num;
211
0
}
212
213
0
Status StorageEngine::start_bg_threads(std::shared_ptr<WorkloadGroup> wg_sptr) {
214
0
    RETURN_IF_ERROR(Thread::create(
215
0
            "StorageEngine", "unused_rowset_monitor_thread",
216
0
            [this]() { this->_unused_rowset_monitor_thread_callback(); },
217
0
            &_unused_rowset_monitor_thread));
218
0
    LOG(INFO) << "unused rowset monitor thread started";
219
220
0
    RETURN_IF_ERROR(Thread::create(
221
0
            "StorageEngine", "evict_querying_rowset_thread",
222
0
            [this]() { this->_evict_quring_rowset_thread_callback(); },
223
0
            &_evict_quering_rowset_thread));
224
0
    LOG(INFO) << "evict quering thread started";
225
226
    // start thread for monitoring the snapshot and trash folder
227
0
    RETURN_IF_ERROR(Thread::create(
228
0
            "StorageEngine", "garbage_sweeper_thread",
229
0
            [this]() { this->_garbage_sweeper_thread_callback(); }, &_garbage_sweeper_thread));
230
0
    LOG(INFO) << "garbage sweeper thread started";
231
232
    // start thread for monitoring the tablet with io error
233
0
    RETURN_IF_ERROR(Thread::create(
234
0
            "StorageEngine", "disk_stat_monitor_thread",
235
0
            [this]() { this->_disk_stat_monitor_thread_callback(); }, &_disk_stat_monitor_thread));
236
0
    LOG(INFO) << "disk stat monitor thread started";
237
238
    // convert store map to vector
239
0
    std::vector<DataDir*> data_dirs = get_stores();
240
241
0
    auto base_compaction_threads = get_base_compaction_threads_num(data_dirs.size());
242
0
    auto cumu_compaction_threads = get_cumu_compaction_threads_num(data_dirs.size());
243
0
    auto single_replica_compaction_threads =
244
0
            get_single_replica_compaction_threads_num(data_dirs.size());
245
246
0
    RETURN_IF_ERROR(ThreadPoolBuilder("BaseCompactionTaskThreadPool")
247
0
                            .set_min_threads(base_compaction_threads)
248
0
                            .set_max_threads(base_compaction_threads)
249
0
                            .build(&_base_compaction_thread_pool));
250
0
    RETURN_IF_ERROR(ThreadPoolBuilder("CumuCompactionTaskThreadPool")
251
0
                            .set_min_threads(cumu_compaction_threads)
252
0
                            .set_max_threads(cumu_compaction_threads)
253
0
                            .build(&_cumu_compaction_thread_pool));
254
0
    RETURN_IF_ERROR(ThreadPoolBuilder("SingleReplicaCompactionTaskThreadPool")
255
0
                            .set_min_threads(single_replica_compaction_threads)
256
0
                            .set_max_threads(single_replica_compaction_threads)
257
0
                            .build(&_single_replica_compaction_thread_pool));
258
259
0
    if (config::enable_segcompaction) {
260
0
        RETURN_IF_ERROR(ThreadPoolBuilder("SegCompactionTaskThreadPool")
261
0
                                .set_min_threads(config::segcompaction_num_threads)
262
0
                                .set_max_threads(config::segcompaction_num_threads)
263
0
                                .build(&_seg_compaction_thread_pool));
264
0
    }
265
0
    RETURN_IF_ERROR(ThreadPoolBuilder("ColdDataCompactionTaskThreadPool")
266
0
                            .set_min_threads(config::cold_data_compaction_thread_num)
267
0
                            .set_max_threads(config::cold_data_compaction_thread_num)
268
0
                            .build(&_cold_data_compaction_thread_pool));
269
270
    // compaction tasks producer thread
271
0
    RETURN_IF_ERROR(Thread::create(
272
0
            "StorageEngine", "compaction_tasks_producer_thread",
273
0
            [this]() { this->_compaction_tasks_producer_callback(); },
274
0
            &_compaction_tasks_producer_thread));
275
0
    LOG(INFO) << "compaction tasks producer thread started";
276
277
0
    RETURN_IF_ERROR(Thread::create(
278
0
            "StorageEngine", "_update_replica_infos_thread",
279
0
            [this]() { this->_update_replica_infos_callback(); }, &_update_replica_infos_thread));
280
0
    LOG(INFO) << "tablet replicas info update thread started";
281
282
0
    int32_t max_checkpoint_thread_num = config::max_meta_checkpoint_threads;
283
0
    if (max_checkpoint_thread_num < 0) {
284
0
        max_checkpoint_thread_num = data_dirs.size();
285
0
    }
286
0
    RETURN_IF_ERROR(ThreadPoolBuilder("TabletMetaCheckpointTaskThreadPool")
287
0
                            .set_max_threads(max_checkpoint_thread_num)
288
0
                            .build(&_tablet_meta_checkpoint_thread_pool));
289
290
0
    RETURN_IF_ERROR(ThreadPoolBuilder("MultiGetTaskThreadPool")
291
0
                            .set_min_threads(config::multi_get_max_threads)
292
0
                            .set_max_threads(config::multi_get_max_threads)
293
0
                            .build(&_bg_multi_get_thread_pool));
294
0
    RETURN_IF_ERROR(Thread::create(
295
0
            "StorageEngine", "tablet_checkpoint_tasks_producer_thread",
296
0
            [this, data_dirs]() { this->_tablet_checkpoint_callback(data_dirs); },
297
0
            &_tablet_checkpoint_tasks_producer_thread));
298
0
    LOG(INFO) << "tablet checkpoint tasks producer thread started";
299
300
0
    RETURN_IF_ERROR(Thread::create(
301
0
            "StorageEngine", "tablet_path_check_thread",
302
0
            [this]() { this->_tablet_path_check_callback(); }, &_tablet_path_check_thread));
303
0
    LOG(INFO) << "tablet path check thread started";
304
305
    // path scan and gc thread
306
0
    if (config::path_gc_check) {
307
0
        for (auto data_dir : get_stores()) {
308
0
            scoped_refptr<Thread> path_gc_thread;
309
0
            RETURN_IF_ERROR(Thread::create(
310
0
                    "StorageEngine", "path_gc_thread",
311
0
                    [this, data_dir]() { this->_path_gc_thread_callback(data_dir); },
312
0
                    &path_gc_thread));
313
0
            _path_gc_threads.emplace_back(path_gc_thread);
314
0
        }
315
0
        LOG(INFO) << "path gc threads started. number:" << get_stores().size();
316
0
    }
317
318
0
    RETURN_IF_ERROR(ThreadPoolBuilder("CooldownTaskThreadPool")
319
0
                            .set_min_threads(config::cooldown_thread_num)
320
0
                            .set_max_threads(config::cooldown_thread_num)
321
0
                            .build(&_cooldown_thread_pool));
322
0
    LOG(INFO) << "cooldown thread pool started";
323
324
0
    RETURN_IF_ERROR(Thread::create(
325
0
            "StorageEngine", "cooldown_tasks_producer_thread",
326
0
            [this]() { this->_cooldown_tasks_producer_callback(); },
327
0
            &_cooldown_tasks_producer_thread));
328
0
    LOG(INFO) << "cooldown tasks producer thread started";
329
330
0
    RETURN_IF_ERROR(Thread::create(
331
0
            "StorageEngine", "remove_unused_remote_files_thread",
332
0
            [this]() { this->_remove_unused_remote_files_callback(); },
333
0
            &_remove_unused_remote_files_thread));
334
0
    LOG(INFO) << "remove unused remote files thread started";
335
336
0
    RETURN_IF_ERROR(Thread::create(
337
0
            "StorageEngine", "cold_data_compaction_producer_thread",
338
0
            [this]() { this->_cold_data_compaction_producer_callback(); },
339
0
            &_cold_data_compaction_producer_thread));
340
0
    LOG(INFO) << "cold data compaction producer thread started";
341
342
    // add tablet publish version thread pool
343
0
    RETURN_IF_ERROR(ThreadPoolBuilder("TabletPublishTxnThreadPool")
344
0
                            .set_min_threads(config::tablet_publish_txn_max_thread)
345
0
                            .set_max_threads(config::tablet_publish_txn_max_thread)
346
0
                            .build(&_tablet_publish_txn_thread_pool));
347
348
0
    RETURN_IF_ERROR(Thread::create(
349
0
            "StorageEngine", "async_publish_version_thread",
350
0
            [this]() { this->_async_publish_callback(); }, &_async_publish_thread));
351
0
    LOG(INFO) << "async publish thread started";
352
353
0
    RETURN_IF_ERROR(Thread::create(
354
0
            "StorageEngine", "check_tablet_delete_bitmap_score_thread",
355
0
            [this]() { this->_check_tablet_delete_bitmap_score_callback(); },
356
0
            &_check_delete_bitmap_score_thread));
357
0
    LOG(INFO) << "check tablet delete bitmap score thread started";
358
359
0
    LOG(INFO) << "all storage engine's background threads are started.";
360
0
    return Status::OK();
361
0
}
362
363
0
void StorageEngine::_garbage_sweeper_thread_callback() {
364
0
    uint32_t max_interval = config::max_garbage_sweep_interval;
365
0
    uint32_t min_interval = config::min_garbage_sweep_interval;
366
367
0
    if (max_interval < min_interval || min_interval <= 0) {
368
0
        LOG(WARNING) << "garbage sweep interval config is illegal: [max=" << max_interval
369
0
                     << " min=" << min_interval << "].";
370
0
        min_interval = 1;
371
0
        max_interval = max_interval >= min_interval ? max_interval : min_interval;
372
0
        LOG(INFO) << "force reset garbage sweep interval. "
373
0
                  << "max_interval=" << max_interval << ", min_interval=" << min_interval;
374
0
    }
375
376
0
    const double pi = M_PI;
377
0
    double usage = 1.0;
378
    // After the program starts, the first round of cleaning starts after min_interval.
379
0
    uint32_t curr_interval = min_interval;
380
0
    do {
381
        // Function properties:
382
        // when usage < 0.6,          ratio close to 1.(interval close to max_interval)
383
        // when usage at [0.6, 0.75], ratio is rapidly decreasing from 0.87 to 0.27.
384
        // when usage > 0.75,         ratio is slowly decreasing.
385
        // when usage > 0.8,          ratio close to min_interval.
386
        // when usage = 0.88,         ratio is approximately 0.0057.
387
0
        double ratio = (1.1 * (pi / 2 - std::atan(usage * 100 / 5 - 14)) - 0.28) / pi;
388
0
        ratio = ratio > 0 ? ratio : 0;
389
0
        auto curr_interval = uint32_t(max_interval * ratio);
390
0
        curr_interval = std::max(curr_interval, min_interval);
391
0
        curr_interval = std::min(curr_interval, max_interval);
392
393
        // start clean trash and update usage.
394
0
        Status res = start_trash_sweep(&usage);
395
0
        if (res.ok() && _need_clean_trash.exchange(false, std::memory_order_relaxed)) {
396
0
            res = start_trash_sweep(&usage, true);
397
0
        }
398
399
0
        if (!res.ok()) {
400
0
            LOG(WARNING) << "one or more errors occur when sweep trash."
401
0
                         << "see previous message for detail. err code=" << res;
402
            // do nothing. continue next loop.
403
0
        }
404
0
    } while (!_stop_background_threads_latch.wait_for(std::chrono::seconds(curr_interval)));
405
0
}
406
407
0
void StorageEngine::_disk_stat_monitor_thread_callback() {
408
0
    int32_t interval = config::disk_stat_monitor_interval;
409
0
    do {
410
0
        _start_disk_stat_monitor();
411
412
0
        interval = config::disk_stat_monitor_interval;
413
0
        if (interval <= 0) {
414
0
            LOG(WARNING) << "disk_stat_monitor_interval config is illegal: " << interval
415
0
                         << ", force set to 1";
416
0
            interval = 1;
417
0
        }
418
0
    } while (!_stop_background_threads_latch.wait_for(std::chrono::seconds(interval)));
419
0
}
420
421
0
void StorageEngine::_unused_rowset_monitor_thread_callback() {
422
0
    int32_t interval = config::unused_rowset_monitor_interval;
423
0
    do {
424
0
        start_delete_unused_rowset();
425
426
0
        interval = config::unused_rowset_monitor_interval;
427
0
        if (interval <= 0) {
428
0
            LOG(WARNING) << "unused_rowset_monitor_interval config is illegal: " << interval
429
0
                         << ", force set to 1";
430
0
            interval = 1;
431
0
        }
432
0
    } while (!_stop_background_threads_latch.wait_for(std::chrono::seconds(interval)));
433
0
}
434
435
0
int32_t StorageEngine::_auto_get_interval_by_disk_capacity(DataDir* data_dir) {
436
0
    double disk_used = data_dir->get_usage(0);
437
0
    double remain_used = 1 - disk_used;
438
0
    DCHECK(remain_used >= 0 && remain_used <= 1);
439
0
    DCHECK(config::path_gc_check_interval_second >= 0);
440
0
    int32_t ret = 0;
441
0
    if (remain_used > 0.9) {
442
        // if config::path_gc_check_interval_second == 24h
443
0
        ret = config::path_gc_check_interval_second;
444
0
    } else if (remain_used > 0.7) {
445
        // 12h
446
0
        ret = config::path_gc_check_interval_second / 2;
447
0
    } else if (remain_used > 0.5) {
448
        // 6h
449
0
        ret = config::path_gc_check_interval_second / 4;
450
0
    } else if (remain_used > 0.3) {
451
        // 4h
452
0
        ret = config::path_gc_check_interval_second / 6;
453
0
    } else {
454
        // 3h
455
0
        ret = config::path_gc_check_interval_second / 8;
456
0
    }
457
0
    return ret;
458
0
}
459
460
0
void StorageEngine::_path_gc_thread_callback(DataDir* data_dir) {
461
0
    LOG(INFO) << "try to start path gc thread!";
462
0
    int32_t last_exec_time = 0;
463
0
    do {
464
0
        int32_t current_time = time(nullptr);
465
466
0
        int32_t interval = _auto_get_interval_by_disk_capacity(data_dir);
467
0
        if (interval <= 0) {
468
0
            LOG(WARNING) << "path gc thread check interval config is illegal:" << interval
469
0
                         << "will be forced set to half hour";
470
0
            interval = 1800; // 0.5 hour
471
0
        }
472
0
        if (current_time - last_exec_time >= interval) {
473
0
            LOG(INFO) << "try to perform path gc! disk remain [" << 1 - data_dir->get_usage(0)
474
0
                      << "] internal [" << interval << "]";
475
0
            data_dir->perform_path_gc();
476
0
            last_exec_time = time(nullptr);
477
0
        }
478
0
    } while (!_stop_background_threads_latch.wait_for(std::chrono::seconds(5)));
479
0
    LOG(INFO) << "stop path gc thread!";
480
0
}
481
482
0
void StorageEngine::_tablet_checkpoint_callback(const std::vector<DataDir*>& data_dirs) {
483
0
    int64_t interval = config::generate_tablet_meta_checkpoint_tasks_interval_secs;
484
0
    do {
485
0
        LOG(INFO) << "begin to produce tablet meta checkpoint tasks.";
486
0
        for (auto data_dir : data_dirs) {
487
0
            auto st = _tablet_meta_checkpoint_thread_pool->submit_func(
488
0
                    [data_dir, this]() { _tablet_manager->do_tablet_meta_checkpoint(data_dir); });
489
0
            if (!st.ok()) {
490
0
                LOG(WARNING) << "submit tablet checkpoint tasks failed.";
491
0
            }
492
0
        }
493
0
        interval = config::generate_tablet_meta_checkpoint_tasks_interval_secs;
494
0
    } while (!_stop_background_threads_latch.wait_for(std::chrono::seconds(interval)));
495
0
}
496
497
0
void StorageEngine::_tablet_path_check_callback() {
498
0
    struct TabletIdComparator {
499
0
        bool operator()(Tablet* a, Tablet* b) { return a->tablet_id() < b->tablet_id(); }
500
0
    };
501
502
0
    using TabletQueue = std::priority_queue<Tablet*, std::vector<Tablet*>, TabletIdComparator>;
503
504
0
    int64_t interval = config::tablet_path_check_interval_seconds;
505
0
    if (interval <= 0) {
506
0
        return;
507
0
    }
508
509
0
    int64_t last_tablet_id = 0;
510
0
    do {
511
0
        int32_t batch_size = config::tablet_path_check_batch_size;
512
0
        if (batch_size <= 0) {
513
0
            if (_stop_background_threads_latch.wait_for(std::chrono::seconds(interval))) {
514
0
                break;
515
0
            }
516
0
            continue;
517
0
        }
518
519
0
        LOG(INFO) << "start to check tablet path";
520
521
0
        auto all_tablets = _tablet_manager->get_all_tablet(
522
0
                [](Tablet* t) { return t->is_used() && t->tablet_state() == TABLET_RUNNING; });
523
524
0
        TabletQueue big_id_tablets;
525
0
        TabletQueue small_id_tablets;
526
0
        for (auto tablet : all_tablets) {
527
0
            auto tablet_id = tablet->tablet_id();
528
0
            TabletQueue* belong_tablets = nullptr;
529
0
            if (tablet_id > last_tablet_id) {
530
0
                if (big_id_tablets.size() < batch_size ||
531
0
                    big_id_tablets.top()->tablet_id() > tablet_id) {
532
0
                    belong_tablets = &big_id_tablets;
533
0
                }
534
0
            } else if (big_id_tablets.size() < batch_size) {
535
0
                if (small_id_tablets.size() < batch_size ||
536
0
                    small_id_tablets.top()->tablet_id() > tablet_id) {
537
0
                    belong_tablets = &small_id_tablets;
538
0
                }
539
0
            }
540
0
            if (belong_tablets != nullptr) {
541
0
                belong_tablets->push(tablet.get());
542
0
                if (belong_tablets->size() > batch_size) {
543
0
                    belong_tablets->pop();
544
0
                }
545
0
            }
546
0
        }
547
548
0
        int32_t need_small_id_tablet_size =
549
0
                batch_size - static_cast<int32_t>(big_id_tablets.size());
550
551
0
        if (!big_id_tablets.empty()) {
552
0
            last_tablet_id = big_id_tablets.top()->tablet_id();
553
0
        }
554
0
        while (!big_id_tablets.empty()) {
555
0
            big_id_tablets.top()->check_tablet_path_exists();
556
0
            big_id_tablets.pop();
557
0
        }
558
559
0
        if (!small_id_tablets.empty() && need_small_id_tablet_size > 0) {
560
0
            while (static_cast<int32_t>(small_id_tablets.size()) > need_small_id_tablet_size) {
561
0
                small_id_tablets.pop();
562
0
            }
563
564
0
            last_tablet_id = small_id_tablets.top()->tablet_id();
565
0
            while (!small_id_tablets.empty()) {
566
0
                small_id_tablets.top()->check_tablet_path_exists();
567
0
                small_id_tablets.pop();
568
0
            }
569
0
        }
570
571
0
    } while (!_stop_background_threads_latch.wait_for(std::chrono::seconds(interval)));
572
0
}
573
574
0
void StorageEngine::_adjust_compaction_thread_num() {
575
0
    auto base_compaction_threads_num = get_base_compaction_threads_num(_store_map.size());
576
0
    if (_base_compaction_thread_pool->max_threads() != base_compaction_threads_num) {
577
0
        int old_max_threads = _base_compaction_thread_pool->max_threads();
578
0
        Status status = _base_compaction_thread_pool->set_max_threads(base_compaction_threads_num);
579
0
        if (status.ok()) {
580
0
            VLOG_NOTICE << "update base compaction thread pool max_threads from " << old_max_threads
581
0
                        << " to " << base_compaction_threads_num;
582
0
        }
583
0
    }
584
0
    if (_base_compaction_thread_pool->min_threads() != base_compaction_threads_num) {
585
0
        int old_min_threads = _base_compaction_thread_pool->min_threads();
586
0
        Status status = _base_compaction_thread_pool->set_min_threads(base_compaction_threads_num);
587
0
        if (status.ok()) {
588
0
            VLOG_NOTICE << "update base compaction thread pool min_threads from " << old_min_threads
589
0
                        << " to " << base_compaction_threads_num;
590
0
        }
591
0
    }
592
593
0
    auto cumu_compaction_threads_num = get_cumu_compaction_threads_num(_store_map.size());
594
0
    if (_cumu_compaction_thread_pool->max_threads() != cumu_compaction_threads_num) {
595
0
        int old_max_threads = _cumu_compaction_thread_pool->max_threads();
596
0
        Status status = _cumu_compaction_thread_pool->set_max_threads(cumu_compaction_threads_num);
597
0
        if (status.ok()) {
598
0
            VLOG_NOTICE << "update cumu compaction thread pool max_threads from " << old_max_threads
599
0
                        << " to " << cumu_compaction_threads_num;
600
0
        }
601
0
    }
602
0
    if (_cumu_compaction_thread_pool->min_threads() != cumu_compaction_threads_num) {
603
0
        int old_min_threads = _cumu_compaction_thread_pool->min_threads();
604
0
        Status status = _cumu_compaction_thread_pool->set_min_threads(cumu_compaction_threads_num);
605
0
        if (status.ok()) {
606
0
            VLOG_NOTICE << "update cumu compaction thread pool min_threads from " << old_min_threads
607
0
                        << " to " << cumu_compaction_threads_num;
608
0
        }
609
0
    }
610
611
0
    auto single_replica_compaction_threads_num =
612
0
            get_single_replica_compaction_threads_num(_store_map.size());
613
0
    if (_single_replica_compaction_thread_pool->max_threads() !=
614
0
        single_replica_compaction_threads_num) {
615
0
        int old_max_threads = _single_replica_compaction_thread_pool->max_threads();
616
0
        Status status = _single_replica_compaction_thread_pool->set_max_threads(
617
0
                single_replica_compaction_threads_num);
618
0
        if (status.ok()) {
619
0
            VLOG_NOTICE << "update single replica compaction thread pool max_threads from "
620
0
                        << old_max_threads << " to " << single_replica_compaction_threads_num;
621
0
        }
622
0
    }
623
0
    if (_single_replica_compaction_thread_pool->min_threads() !=
624
0
        single_replica_compaction_threads_num) {
625
0
        int old_min_threads = _single_replica_compaction_thread_pool->min_threads();
626
0
        Status status = _single_replica_compaction_thread_pool->set_min_threads(
627
0
                single_replica_compaction_threads_num);
628
0
        if (status.ok()) {
629
0
            VLOG_NOTICE << "update single replica compaction thread pool min_threads from "
630
0
                        << old_min_threads << " to " << single_replica_compaction_threads_num;
631
0
        }
632
0
    }
633
0
}
634
635
0
void StorageEngine::_compaction_tasks_producer_callback() {
636
0
    LOG(INFO) << "try to start compaction producer process!";
637
638
0
    std::vector<DataDir*> data_dirs = get_stores();
639
0
    _compaction_submit_registry.reset(data_dirs);
640
641
0
    int round = 0;
642
0
    CompactionType compaction_type;
643
644
    // Used to record the time when the score metric was last updated.
645
    // The update of the score metric is accompanied by the logic of selecting the tablet.
646
    // If there is no slot available, the logic of selecting the tablet will be terminated,
647
    // which causes the score metric update to be terminated.
648
    // In order to avoid this situation, we need to update the score regularly.
649
0
    int64_t last_cumulative_score_update_time = 0;
650
0
    int64_t last_base_score_update_time = 0;
651
0
    static const int64_t check_score_interval_ms = 5000; // 5 secs
652
653
0
    int64_t interval = config::generate_compaction_tasks_interval_ms;
654
0
    do {
655
0
        if (!config::disable_auto_compaction &&
656
0
            !GlobalMemoryArbitrator::is_exceed_soft_mem_limit(GB_EXCHANGE_BYTE)) {
657
0
            _adjust_compaction_thread_num();
658
659
0
            bool check_score = false;
660
0
            int64_t cur_time = UnixMillis();
661
0
            if (round < config::cumulative_compaction_rounds_for_each_base_compaction_round) {
662
0
                compaction_type = CompactionType::CUMULATIVE_COMPACTION;
663
0
                round++;
664
0
                if (cur_time - last_cumulative_score_update_time >= check_score_interval_ms) {
665
0
                    check_score = true;
666
0
                    last_cumulative_score_update_time = cur_time;
667
0
                }
668
0
            } else {
669
0
                compaction_type = CompactionType::BASE_COMPACTION;
670
0
                round = 0;
671
0
                if (cur_time - last_base_score_update_time >= check_score_interval_ms) {
672
0
                    check_score = true;
673
0
                    last_base_score_update_time = cur_time;
674
0
                }
675
0
            }
676
0
            std::vector<TabletSharedPtr> tablets_compaction =
677
0
                    _generate_compaction_tasks(compaction_type, data_dirs, check_score);
678
0
            if (tablets_compaction.size() == 0) {
679
0
                std::unique_lock<std::mutex> lock(_compaction_producer_sleep_mutex);
680
0
                _wakeup_producer_flag = 0;
681
                // It is necessary to wake up the thread on timeout to prevent deadlock
682
                // in case of no running compaction task.
683
0
                _compaction_producer_sleep_cv.wait_for(
684
0
                        lock, std::chrono::milliseconds(2000),
685
0
                        [this] { return _wakeup_producer_flag == 1; });
686
0
                continue;
687
0
            }
688
689
0
            for (const auto& tablet : tablets_compaction) {
690
0
                if (compaction_type == CompactionType::BASE_COMPACTION) {
691
0
                    tablet->set_last_base_compaction_schedule_time(UnixMillis());
692
0
                }
693
0
                Status st = _submit_compaction_task(tablet, compaction_type, false);
694
0
                if (!st.ok()) {
695
0
                    LOG(WARNING) << "failed to submit compaction task for tablet: "
696
0
                                 << tablet->tablet_id() << ", err: " << st;
697
0
                }
698
0
            }
699
0
            interval = config::generate_compaction_tasks_interval_ms;
700
0
        } else {
701
0
            interval = 5000; // 5s to check disable_auto_compaction
702
0
        }
703
0
    } while (!_stop_background_threads_latch.wait_for(std::chrono::milliseconds(interval)));
704
0
}
705
706
0
void StorageEngine::_update_replica_infos_callback() {
707
#ifdef GOOGLE_PROFILER
708
    ProfilerRegisterThread();
709
#endif
710
0
    LOG(INFO) << "start to update replica infos!";
711
712
0
    int64_t interval = config::update_replica_infos_interval_seconds;
713
0
    do {
714
0
        auto all_tablets = _tablet_manager->get_all_tablet([](Tablet* t) {
715
0
            return t->is_used() && t->tablet_state() == TABLET_RUNNING &&
716
0
                   !t->tablet_meta()->tablet_schema()->disable_auto_compaction() &&
717
0
                   t->tablet_meta()->tablet_schema()->enable_single_replica_compaction();
718
0
        });
719
0
        ClusterInfo* cluster_info = ExecEnv::GetInstance()->cluster_info();
720
0
        if (cluster_info == nullptr) {
721
0
            LOG(WARNING) << "Have not get FE Master heartbeat yet";
722
0
            std::this_thread::sleep_for(std::chrono::seconds(2));
723
0
            continue;
724
0
        }
725
0
        TNetworkAddress master_addr = cluster_info->master_fe_addr;
726
0
        if (master_addr.hostname == "" || master_addr.port == 0) {
727
0
            LOG(WARNING) << "Have not get FE Master heartbeat yet";
728
0
            std::this_thread::sleep_for(std::chrono::seconds(2));
729
0
            continue;
730
0
        }
731
732
0
        int start = 0;
733
0
        int tablet_size = all_tablets.size();
734
        // The while loop may take a long time, we should skip it when stop
735
0
        while (start < tablet_size && _stop_background_threads_latch.count() > 0) {
736
0
            int batch_size = std::min(100, tablet_size - start);
737
0
            int end = start + batch_size;
738
0
            TGetTabletReplicaInfosRequest request;
739
0
            TGetTabletReplicaInfosResult result;
740
0
            for (int i = start; i < end; i++) {
741
0
                request.tablet_ids.emplace_back(all_tablets[i]->tablet_id());
742
0
            }
743
0
            Status rpc_st = ThriftRpcHelper::rpc<FrontendServiceClient>(
744
0
                    master_addr.hostname, master_addr.port,
745
0
                    [&request, &result](FrontendServiceConnection& client) {
746
0
                        client->getTabletReplicaInfos(result, request);
747
0
                    });
748
749
0
            if (!rpc_st.ok()) {
750
0
                LOG(WARNING) << "Failed to get tablet replica infos, encounter rpc failure, "
751
0
                                "tablet start: "
752
0
                             << start << " end: " << end;
753
0
                continue;
754
0
            }
755
756
0
            std::unique_lock<std::mutex> lock(_peer_replica_infos_mutex);
757
0
            for (const auto& it : result.tablet_replica_infos) {
758
0
                auto tablet_id = it.first;
759
0
                auto tablet = _tablet_manager->get_tablet(tablet_id);
760
0
                if (tablet == nullptr) {
761
0
                    VLOG_CRITICAL << "tablet ptr is nullptr";
762
0
                    continue;
763
0
                }
764
765
0
                VLOG_NOTICE << tablet_id << " tablet has " << it.second.size() << " replicas";
766
0
                uint64_t min_modulo = MOD_PRIME;
767
0
                TReplicaInfo peer_replica;
768
0
                for (const auto& replica : it.second) {
769
0
                    int64_t peer_replica_id = replica.replica_id;
770
0
                    uint64_t modulo = HashUtil::hash64(&peer_replica_id, sizeof(peer_replica_id),
771
0
                                                       DEFAULT_SEED) %
772
0
                                      MOD_PRIME;
773
0
                    if (modulo < min_modulo) {
774
0
                        peer_replica = replica;
775
0
                        min_modulo = modulo;
776
0
                    }
777
0
                }
778
0
                VLOG_NOTICE << "tablet " << tablet_id << ", peer replica host is "
779
0
                            << peer_replica.host;
780
0
                _peer_replica_infos[tablet_id] = peer_replica;
781
0
            }
782
0
            _token = result.token;
783
0
            VLOG_NOTICE << "get tablet replica infos from fe, size is " << end - start
784
0
                        << " token = " << result.token;
785
0
            start = end;
786
0
        }
787
0
        interval = config::update_replica_infos_interval_seconds;
788
0
    } while (!_stop_background_threads_latch.wait_for(std::chrono::seconds(interval)));
789
0
}
790
791
Status StorageEngine::_submit_single_replica_compaction_task(TabletSharedPtr tablet,
792
0
                                                             CompactionType compaction_type) {
793
    // For single replica compaction, the local version to be merged is determined based on the version fetched from the peer replica.
794
    // Therefore, it is currently not possible to determine whether it should be a base compaction or cumulative compaction.
795
    // As a result, the tablet needs to be pushed to both the _tablet_submitted_cumu_compaction and the _tablet_submitted_base_compaction simultaneously.
796
0
    bool already_exist =
797
0
            _compaction_submit_registry.insert(tablet, CompactionType::CUMULATIVE_COMPACTION);
798
0
    if (already_exist) {
799
0
        return Status::AlreadyExist<false>(
800
0
                "compaction task has already been submitted, tablet_id={}", tablet->tablet_id());
801
0
    }
802
803
0
    already_exist = _compaction_submit_registry.insert(tablet, CompactionType::BASE_COMPACTION);
804
0
    if (already_exist) {
805
0
        _pop_tablet_from_submitted_compaction(tablet, CompactionType::CUMULATIVE_COMPACTION);
806
0
        return Status::AlreadyExist<false>(
807
0
                "compaction task has already been submitted, tablet_id={}", tablet->tablet_id());
808
0
    }
809
810
0
    auto compaction = std::make_shared<SingleReplicaCompaction>(*this, tablet, compaction_type);
811
0
    DorisMetrics::instance()->single_compaction_request_total->increment(1);
812
0
    auto st = compaction->prepare_compact();
813
814
0
    auto clean_single_replica_compaction = [tablet, this]() {
815
0
        _pop_tablet_from_submitted_compaction(tablet, CompactionType::CUMULATIVE_COMPACTION);
816
0
        _pop_tablet_from_submitted_compaction(tablet, CompactionType::BASE_COMPACTION);
817
0
    };
818
819
0
    if (!st.ok()) {
820
0
        clean_single_replica_compaction();
821
0
        if (!st.is<ErrorCode::CUMULATIVE_NO_SUITABLE_VERSION>()) {
822
0
            LOG(WARNING) << "failed to prepare single replica compaction, tablet_id="
823
0
                         << tablet->tablet_id() << " : " << st;
824
0
            return st;
825
0
        }
826
0
        return Status::OK(); // No suitable version, regard as OK
827
0
    }
828
829
0
    auto submit_st = _single_replica_compaction_thread_pool->submit_func(
830
0
            [tablet, compaction = std::move(compaction),
831
0
             clean_single_replica_compaction]() mutable {
832
0
                tablet->execute_single_replica_compaction(*compaction);
833
0
                clean_single_replica_compaction();
834
0
            });
835
0
    if (!submit_st.ok()) {
836
0
        clean_single_replica_compaction();
837
0
        return Status::InternalError(
838
0
                "failed to submit single replica compaction task to thread pool, "
839
0
                "tablet_id={}",
840
0
                tablet->tablet_id());
841
0
    }
842
0
    return Status::OK();
843
0
}
844
845
void StorageEngine::get_tablet_rowset_versions(const PGetTabletVersionsRequest* request,
846
0
                                               PGetTabletVersionsResponse* response) {
847
0
    TabletSharedPtr tablet = _tablet_manager->get_tablet(request->tablet_id());
848
0
    if (tablet == nullptr) {
849
0
        response->mutable_status()->set_status_code(TStatusCode::CANCELLED);
850
0
        return;
851
0
    }
852
0
    std::vector<Version> local_versions = tablet->get_all_local_versions();
853
0
    for (const auto& local_version : local_versions) {
854
0
        auto version = response->add_versions();
855
0
        version->set_first(local_version.first);
856
0
        version->set_second(local_version.second);
857
0
    }
858
0
    response->mutable_status()->set_status_code(0);
859
0
}
860
861
bool need_generate_compaction_tasks(int task_cnt_per_disk, int thread_per_disk,
862
0
                                    CompactionType compaction_type, bool all_base) {
863
    // We need to reserve at least one Slot for cumulative compaction.
864
    // So when there is only one Slot, we have to judge whether there is a cumulative compaction
865
    // in the current submitted tasks.
866
    // If so, the last Slot can be assigned to Base compaction,
867
    // otherwise, this Slot needs to be reserved for cumulative compaction.
868
0
    if (task_cnt_per_disk >= thread_per_disk) {
869
        // Return if no available slot
870
0
        return false;
871
0
    } else if (task_cnt_per_disk >= thread_per_disk - 1) {
872
        // Only one slot left, check if it can be assigned to base compaction task.
873
0
        if (compaction_type == CompactionType::BASE_COMPACTION) {
874
0
            if (all_base) {
875
0
                return false;
876
0
            }
877
0
        }
878
0
    }
879
0
    return true;
880
0
}
881
882
0
int get_concurrent_per_disk(int max_score, int thread_per_disk) {
883
0
    if (!config::enable_compaction_priority_scheduling) {
884
0
        return thread_per_disk;
885
0
    }
886
887
0
    double load_average = 0;
888
0
    if (DorisMetrics::instance()->system_metrics() != nullptr) {
889
0
        load_average = DorisMetrics::instance()->system_metrics()->get_load_average_1_min();
890
0
    }
891
0
    int num_cores = doris::CpuInfo::num_cores();
892
0
    bool cpu_usage_high = load_average > num_cores * 0.8;
893
894
0
    auto process_memory_usage = doris::GlobalMemoryArbitrator::process_memory_usage();
895
0
    bool memory_usage_high = process_memory_usage > MemInfo::soft_mem_limit() * 0.8;
896
897
0
    if (max_score <= config::low_priority_compaction_score_threshold &&
898
0
        (cpu_usage_high || memory_usage_high)) {
899
0
        return config::low_priority_compaction_task_num_per_disk;
900
0
    }
901
902
0
    return thread_per_disk;
903
0
}
904
905
0
int32_t disk_compaction_slot_num(const DataDir& data_dir) {
906
0
    return data_dir.is_ssd_disk() ? config::compaction_task_num_per_fast_disk
907
0
                                  : config::compaction_task_num_per_disk;
908
0
}
909
910
bool has_free_compaction_slot(CompactionSubmitRegistry* registry, DataDir* dir,
911
0
                              CompactionType compaction_type, uint32_t executing_cnt) {
912
0
    int32_t thread_per_disk = disk_compaction_slot_num(*dir);
913
0
    return need_generate_compaction_tasks(
914
0
            executing_cnt, thread_per_disk, compaction_type,
915
0
            !registry->has_compaction_task(dir, CompactionType::CUMULATIVE_COMPACTION));
916
0
}
917
918
std::vector<TabletSharedPtr> StorageEngine::_generate_compaction_tasks(
919
0
        CompactionType compaction_type, std::vector<DataDir*>& data_dirs, bool check_score) {
920
0
    _update_cumulative_compaction_policy();
921
0
    std::vector<TabletSharedPtr> tablets_compaction;
922
0
    uint32_t max_compaction_score = 0;
923
924
0
    std::random_device rd;
925
0
    std::mt19937 g(rd());
926
0
    std::shuffle(data_dirs.begin(), data_dirs.end(), g);
927
928
    // Copy _tablet_submitted_xxx_compaction map so that we don't need to hold _tablet_submitted_compaction_mutex
929
    // when traversing the data dir
930
0
    auto compaction_registry_snapshot = _compaction_submit_registry.create_snapshot();
931
0
    for (auto* data_dir : data_dirs) {
932
0
        bool need_pick_tablet = true;
933
0
        uint32_t executing_task_num =
934
0
                compaction_registry_snapshot.count_executing_cumu_and_base(data_dir);
935
0
        need_pick_tablet = has_free_compaction_slot(&compaction_registry_snapshot, data_dir,
936
0
                                                    compaction_type, executing_task_num);
937
0
        if (!need_pick_tablet && !check_score) {
938
0
            continue;
939
0
        }
940
941
        // Even if need_pick_tablet is false, we still need to call find_best_tablet_to_compaction(),
942
        // So that we can update the max_compaction_score metric.
943
0
        if (!data_dir->reach_capacity_limit(0)) {
944
0
            uint32_t disk_max_score = 0;
945
0
            auto tablets = compaction_registry_snapshot.pick_topn_tablets_for_compaction(
946
0
                    _tablet_manager.get(), data_dir, compaction_type,
947
0
                    _cumulative_compaction_policies, &disk_max_score);
948
0
            int concurrent_num =
949
0
                    get_concurrent_per_disk(disk_max_score, disk_compaction_slot_num(*data_dir));
950
0
            need_pick_tablet = need_generate_compaction_tasks(
951
0
                    executing_task_num, concurrent_num, compaction_type,
952
0
                    !compaction_registry_snapshot.has_compaction_task(
953
0
                            data_dir, CompactionType::CUMULATIVE_COMPACTION));
954
0
            for (const auto& tablet : tablets) {
955
0
                if (tablet != nullptr) {
956
0
                    if (need_pick_tablet) {
957
0
                        tablets_compaction.emplace_back(tablet);
958
0
                    }
959
0
                    max_compaction_score = std::max(max_compaction_score, disk_max_score);
960
0
                }
961
0
            }
962
0
        }
963
0
    }
964
965
0
    if (max_compaction_score > 0) {
966
0
        if (compaction_type == CompactionType::BASE_COMPACTION) {
967
0
            DorisMetrics::instance()->tablet_base_max_compaction_score->set_value(
968
0
                    max_compaction_score);
969
0
        } else {
970
0
            DorisMetrics::instance()->tablet_cumulative_max_compaction_score->set_value(
971
0
                    max_compaction_score);
972
0
        }
973
0
    }
974
0
    return tablets_compaction;
975
0
}
976
977
0
void StorageEngine::_update_cumulative_compaction_policy() {
978
0
    if (_cumulative_compaction_policies.empty()) {
979
0
        _cumulative_compaction_policies[CUMULATIVE_SIZE_BASED_POLICY] =
980
0
                CumulativeCompactionPolicyFactory::create_cumulative_compaction_policy(
981
0
                        CUMULATIVE_SIZE_BASED_POLICY);
982
0
        _cumulative_compaction_policies[CUMULATIVE_TIME_SERIES_POLICY] =
983
0
                CumulativeCompactionPolicyFactory::create_cumulative_compaction_policy(
984
0
                        CUMULATIVE_TIME_SERIES_POLICY);
985
0
    }
986
0
}
987
988
void StorageEngine::_pop_tablet_from_submitted_compaction(TabletSharedPtr tablet,
989
2
                                                          CompactionType compaction_type) {
990
2
    _compaction_submit_registry.remove(tablet, compaction_type, [this]() {
991
2
        std::unique_lock<std::mutex> lock(_compaction_producer_sleep_mutex);
992
2
        _wakeup_producer_flag = 1;
993
2
        _compaction_producer_sleep_cv.notify_one();
994
2
    });
995
2
}
996
997
Status StorageEngine::_submit_compaction_task(TabletSharedPtr tablet,
998
10
                                              CompactionType compaction_type, bool force) {
999
10
    if (tablet->tablet_meta()->tablet_schema()->enable_single_replica_compaction() &&
1000
10
        should_fetch_from_peer(tablet->tablet_id())) {
1001
0
        VLOG_CRITICAL << "start to submit single replica compaction task for tablet: "
1002
0
                      << tablet->tablet_id();
1003
0
        Status st = _submit_single_replica_compaction_task(tablet, compaction_type);
1004
0
        if (!st.ok()) {
1005
0
            LOG(WARNING) << "failed to submit single replica compaction task for tablet: "
1006
0
                         << tablet->tablet_id() << ", err: " << st;
1007
0
        }
1008
1009
0
        return Status::OK();
1010
0
    }
1011
10
    bool already_exist = _compaction_submit_registry.insert(tablet, compaction_type);
1012
10
    if (already_exist) {
1013
0
        return Status::AlreadyExist<false>(
1014
0
                "compaction task has already been submitted, tablet_id={}, compaction_type={}.",
1015
0
                tablet->tablet_id(), compaction_type);
1016
0
    }
1017
10
    tablet->compaction_stage = CompactionStage::PENDING;
1018
10
    std::shared_ptr<CompactionMixin> compaction;
1019
10
    int64_t permits = 0;
1020
10
    Status st = Tablet::prepare_compaction_and_calculate_permits(compaction_type, tablet,
1021
10
                                                                 compaction, permits);
1022
10
    if (st.ok() && permits > 0) {
1023
10
        if (!force) {
1024
10
            _permit_limiter.request(permits);
1025
10
        }
1026
10
        std::unique_ptr<ThreadPool>& thread_pool =
1027
10
                (compaction_type == CompactionType::CUMULATIVE_COMPACTION)
1028
10
                        ? _cumu_compaction_thread_pool
1029
10
                        : _base_compaction_thread_pool;
1030
10
        VLOG_CRITICAL << "compaction thread pool. type: "
1031
0
                      << (compaction_type == CompactionType::CUMULATIVE_COMPACTION ? "CUMU"
1032
0
                                                                                   : "BASE")
1033
0
                      << ", num_threads: " << thread_pool->num_threads()
1034
0
                      << ", num_threads_pending_start: " << thread_pool->num_threads_pending_start()
1035
0
                      << ", num_active_threads: " << thread_pool->num_active_threads()
1036
0
                      << ", max_threads: " << thread_pool->max_threads()
1037
0
                      << ", min_threads: " << thread_pool->min_threads()
1038
0
                      << ", num_total_queued_tasks: " << thread_pool->get_queue_size();
1039
10
        auto st = thread_pool->submit_func([tablet, compaction = std::move(compaction),
1040
10
                                            compaction_type, permits, force, this]() {
1041
2
            Defer defer {[&]() {
1042
2
                if (!force) {
1043
2
                    _permit_limiter.release(permits);
1044
2
                }
1045
2
                _pop_tablet_from_submitted_compaction(tablet, compaction_type);
1046
2
                tablet->compaction_stage = CompactionStage::NOT_SCHEDULED;
1047
2
            }};
1048
2
            if (!tablet->can_do_compaction(tablet->data_dir()->path_hash(), compaction_type)) {
1049
0
                LOG(INFO) << "Tablet state has been changed, no need to begin this compaction "
1050
0
                             "task, tablet_id="
1051
0
                          << tablet->tablet_id() << ", tablet_state=" << tablet->tablet_state();
1052
0
                return;
1053
0
            }
1054
2
            tablet->compaction_stage = CompactionStage::EXECUTING;
1055
2
            TEST_SYNC_POINT_RETURN_WITH_VOID("olap_server::execute_compaction");
1056
0
            tablet->execute_compaction(*compaction);
1057
0
        });
1058
10
        if (!st.ok()) {
1059
0
            if (!force) {
1060
0
                _permit_limiter.release(permits);
1061
0
            }
1062
0
            _pop_tablet_from_submitted_compaction(tablet, compaction_type);
1063
0
            tablet->compaction_stage = CompactionStage::NOT_SCHEDULED;
1064
0
            return Status::InternalError(
1065
0
                    "failed to submit compaction task to thread pool, "
1066
0
                    "tablet_id={}, compaction_type={}.",
1067
0
                    tablet->tablet_id(), compaction_type);
1068
0
        }
1069
10
        return Status::OK();
1070
10
    } else {
1071
0
        _pop_tablet_from_submitted_compaction(tablet, compaction_type);
1072
0
        tablet->compaction_stage = CompactionStage::NOT_SCHEDULED;
1073
0
        if (!st.ok()) {
1074
0
            return Status::InternalError(
1075
0
                    "failed to prepare compaction task and calculate permits, "
1076
0
                    "tablet_id={}, compaction_type={}, "
1077
0
                    "permit={}, current_permit={}, status={}",
1078
0
                    tablet->tablet_id(), compaction_type, permits, _permit_limiter.usage(),
1079
0
                    st.to_string());
1080
0
        }
1081
0
        return st;
1082
0
    }
1083
10
}
1084
1085
Status StorageEngine::submit_compaction_task(TabletSharedPtr tablet, CompactionType compaction_type,
1086
0
                                             bool force, bool eager) {
1087
0
    if (!eager) {
1088
0
        DCHECK(compaction_type == CompactionType::BASE_COMPACTION ||
1089
0
               compaction_type == CompactionType::CUMULATIVE_COMPACTION);
1090
0
        auto compaction_registry_snapshot = _compaction_submit_registry.create_snapshot();
1091
0
        auto stores = get_stores();
1092
1093
0
        bool is_busy = std::none_of(
1094
0
                stores.begin(), stores.end(),
1095
0
                [&compaction_registry_snapshot, compaction_type](auto* data_dir) {
1096
0
                    return has_free_compaction_slot(
1097
0
                            &compaction_registry_snapshot, data_dir, compaction_type,
1098
0
                            compaction_registry_snapshot.count_executing_cumu_and_base(data_dir));
1099
0
                });
1100
0
        if (is_busy) {
1101
0
            LOG_EVERY_N(WARNING, 100)
1102
0
                    << "Too busy to submit a compaction task, tablet=" << tablet->get_table_id();
1103
0
            return Status::OK();
1104
0
        }
1105
0
    }
1106
0
    _update_cumulative_compaction_policy();
1107
    // alter table tableName set ("compaction_policy"="time_series")
1108
    // if atler table's compaction  policy, we need to modify tablet compaction policy shared ptr
1109
0
    if (tablet->get_cumulative_compaction_policy() == nullptr ||
1110
0
        tablet->get_cumulative_compaction_policy()->name() !=
1111
0
                tablet->tablet_meta()->compaction_policy()) {
1112
0
        tablet->set_cumulative_compaction_policy(
1113
0
                _cumulative_compaction_policies.at(tablet->tablet_meta()->compaction_policy()));
1114
0
    }
1115
0
    tablet->set_skip_compaction(false);
1116
0
    return _submit_compaction_task(tablet, compaction_type, force);
1117
0
}
1118
1119
Status StorageEngine::_handle_seg_compaction(std::shared_ptr<SegcompactionWorker> worker,
1120
                                             SegCompactionCandidatesSharedPtr segments,
1121
11
                                             uint64_t submission_time) {
1122
    // note: be aware that worker->_writer maybe released when the task is cancelled
1123
11
    uint64_t exec_queue_time = GetCurrentTimeMicros() - submission_time;
1124
11
    LOG(INFO) << "segcompaction thread pool queue time(ms): " << exec_queue_time / 1000;
1125
11
    worker->compact_segments(segments);
1126
    // return OK here. error will be reported via BetaRowsetWriter::_segcompaction_status
1127
11
    return Status::OK();
1128
11
}
1129
1130
Status StorageEngine::submit_seg_compaction_task(std::shared_ptr<SegcompactionWorker> worker,
1131
11
                                                 SegCompactionCandidatesSharedPtr segments) {
1132
11
    uint64_t submission_time = GetCurrentTimeMicros();
1133
11
    return _seg_compaction_thread_pool->submit_func([this, worker, segments, submission_time] {
1134
11
        static_cast<void>(_handle_seg_compaction(worker, segments, submission_time));
1135
11
    });
1136
11
}
1137
1138
0
Status StorageEngine::process_index_change_task(const TAlterInvertedIndexReq& request) {
1139
0
    auto tablet_id = request.tablet_id;
1140
0
    TabletSharedPtr tablet = _tablet_manager->get_tablet(tablet_id);
1141
0
    DBUG_EXECUTE_IF("StorageEngine::process_index_change_task_tablet_nullptr",
1142
0
                    { tablet = nullptr; })
1143
0
    if (tablet == nullptr) {
1144
0
        LOG(WARNING) << "tablet: " << tablet_id << " not exist";
1145
0
        return Status::InternalError("tablet not exist, tablet_id={}.", tablet_id);
1146
0
    }
1147
1148
0
    IndexBuilderSharedPtr index_builder = std::make_shared<IndexBuilder>(
1149
0
            *this, tablet, request.columns, request.alter_inverted_indexes, request.is_drop_op);
1150
0
    RETURN_IF_ERROR(_handle_index_change(index_builder));
1151
0
    return Status::OK();
1152
0
}
1153
1154
0
Status StorageEngine::_handle_index_change(IndexBuilderSharedPtr index_builder) {
1155
0
    RETURN_IF_ERROR(index_builder->init());
1156
0
    RETURN_IF_ERROR(index_builder->do_build_inverted_index());
1157
0
    return Status::OK();
1158
0
}
1159
1160
0
void StorageEngine::_cooldown_tasks_producer_callback() {
1161
0
    int64_t interval = config::generate_cooldown_task_interval_sec;
1162
    // the cooldown replica may be slow to upload it's meta file, so we should wait
1163
    // until it has done uploaded
1164
0
    int64_t skip_failed_interval = interval * 10;
1165
0
    do {
1166
        // these tables are ordered by priority desc
1167
0
        std::vector<TabletSharedPtr> tablets;
1168
0
        std::vector<RowsetSharedPtr> rowsets;
1169
        // TODO(luwei) : a more efficient way to get cooldown tablets
1170
0
        auto cur_time = time(nullptr);
1171
        // we should skip all the tablets which are not running and those pending to do cooldown
1172
        // also tablets once failed to do follow cooldown
1173
0
        auto skip_tablet = [this, skip_failed_interval,
1174
0
                            cur_time](const TabletSharedPtr& tablet) -> bool {
1175
0
            bool is_skip =
1176
0
                    cur_time - tablet->last_failed_follow_cooldown_time() < skip_failed_interval ||
1177
0
                    TABLET_RUNNING != tablet->tablet_state();
1178
0
            if (is_skip) {
1179
0
                return is_skip;
1180
0
            }
1181
0
            std::lock_guard<std::mutex> lock(_running_cooldown_mutex);
1182
0
            return _running_cooldown_tablets.find(tablet->tablet_id()) !=
1183
0
                   _running_cooldown_tablets.end();
1184
0
        };
1185
0
        _tablet_manager->get_cooldown_tablets(&tablets, &rowsets, std::move(skip_tablet));
1186
0
        LOG(INFO) << "cooldown producer get tablet num: " << tablets.size();
1187
0
        int max_priority = tablets.size();
1188
0
        int index = 0;
1189
0
        for (const auto& tablet : tablets) {
1190
0
            {
1191
0
                std::lock_guard<std::mutex> lock(_running_cooldown_mutex);
1192
0
                _running_cooldown_tablets.insert(tablet->tablet_id());
1193
0
            }
1194
0
            PriorityThreadPool::Task task;
1195
0
            RowsetSharedPtr rowset = std::move(rowsets[index++]);
1196
0
            task.work_function = [tablet, rowset, task_size = tablets.size(), this]() {
1197
0
                Status st = tablet->cooldown(rowset);
1198
0
                {
1199
0
                    std::lock_guard<std::mutex> lock(_running_cooldown_mutex);
1200
0
                    _running_cooldown_tablets.erase(tablet->tablet_id());
1201
0
                }
1202
0
                if (!st.ok()) {
1203
0
                    LOG(WARNING) << "failed to cooldown, tablet: " << tablet->tablet_id()
1204
0
                                 << " err: " << st;
1205
0
                } else {
1206
0
                    LOG(INFO) << "succeed to cooldown, tablet: " << tablet->tablet_id()
1207
0
                              << " cooldown progress ("
1208
0
                              << task_size - _cooldown_thread_pool->get_queue_size() << "/"
1209
0
                              << task_size << ")";
1210
0
                }
1211
0
            };
1212
0
            task.priority = max_priority--;
1213
0
            bool submited = _cooldown_thread_pool->offer(std::move(task));
1214
1215
0
            if (!submited) {
1216
0
                LOG(INFO) << "failed to submit cooldown task";
1217
0
            }
1218
0
        }
1219
0
    } while (!_stop_background_threads_latch.wait_for(std::chrono::seconds(interval)));
1220
0
}
1221
1222
0
void StorageEngine::_remove_unused_remote_files_callback() {
1223
0
    while (!_stop_background_threads_latch.wait_for(
1224
0
            std::chrono::seconds(config::remove_unused_remote_files_interval_sec))) {
1225
0
        LOG(INFO) << "begin to remove unused remote files";
1226
0
        do_remove_unused_remote_files();
1227
0
    }
1228
0
}
1229
1230
0
void StorageEngine::do_remove_unused_remote_files() {
1231
0
    auto tablets = tablet_manager()->get_all_tablet([](Tablet* t) {
1232
0
        return t->tablet_meta()->cooldown_meta_id().initialized() && t->is_used() &&
1233
0
               t->tablet_state() == TABLET_RUNNING &&
1234
0
               t->cooldown_conf_unlocked().cooldown_replica_id == t->replica_id();
1235
0
    });
1236
0
    TConfirmUnusedRemoteFilesRequest req;
1237
0
    req.__isset.confirm_list = true;
1238
    // tablet_id -> [storage_resource, unused_remote_files]
1239
0
    using unused_remote_files_buffer_t =
1240
0
            std::unordered_map<int64_t, std::pair<StorageResource, std::vector<io::FileInfo>>>;
1241
0
    unused_remote_files_buffer_t buffer;
1242
0
    int64_t num_files_in_buffer = 0;
1243
    // assume a filename is 0.1KB, buffer size should not larger than 100MB
1244
0
    constexpr int64_t max_files_in_buffer = 1000000;
1245
1246
0
    auto calc_unused_remote_files = [&req, &buffer, &num_files_in_buffer, this](Tablet* t) {
1247
0
        auto storage_resource = get_resource_by_storage_policy_id(t->storage_policy_id());
1248
0
        if (!storage_resource) {
1249
0
            LOG(WARNING) << "encounter error when remove unused remote files, tablet_id="
1250
0
                         << t->tablet_id() << " : " << storage_resource.error();
1251
0
            return;
1252
0
        }
1253
1254
        // TODO(plat1ko): Support path v1
1255
0
        if (storage_resource->path_version > 0) {
1256
0
            return;
1257
0
        }
1258
1259
0
        std::vector<io::FileInfo> files;
1260
        // FIXME(plat1ko): What if user reset resource in storage policy to another resource?
1261
        //  Maybe we should also list files in previously uploaded resources.
1262
0
        bool exists = true;
1263
0
        auto st = storage_resource->fs->list(storage_resource->remote_tablet_path(t->tablet_id()),
1264
0
                                             true, &files, &exists);
1265
0
        if (!st.ok()) {
1266
0
            LOG(WARNING) << "encounter error when remove unused remote files, tablet_id="
1267
0
                         << t->tablet_id() << " : " << st;
1268
0
            return;
1269
0
        }
1270
0
        if (!exists || files.empty()) {
1271
0
            return;
1272
0
        }
1273
        // get all cooldowned rowsets
1274
0
        RowsetIdUnorderedSet cooldowned_rowsets;
1275
0
        UniqueId cooldown_meta_id;
1276
0
        {
1277
0
            std::shared_lock rlock(t->get_header_lock());
1278
0
            for (auto&& rs_meta : t->tablet_meta()->all_rs_metas()) {
1279
0
                if (!rs_meta->is_local()) {
1280
0
                    cooldowned_rowsets.insert(rs_meta->rowset_id());
1281
0
                }
1282
0
            }
1283
0
            if (cooldowned_rowsets.empty()) {
1284
0
                return;
1285
0
            }
1286
0
            cooldown_meta_id = t->tablet_meta()->cooldown_meta_id();
1287
0
        }
1288
0
        auto [cooldown_term, cooldown_replica_id] = t->cooldown_conf();
1289
0
        if (cooldown_replica_id != t->replica_id()) {
1290
0
            return;
1291
0
        }
1292
        // {cooldown_replica_id}.{cooldown_term}.meta
1293
0
        std::string remote_meta_path =
1294
0
                cooldown_tablet_meta_filename(cooldown_replica_id, cooldown_term);
1295
        // filter out the paths that should be reserved
1296
0
        auto filter = [&, this](io::FileInfo& info) {
1297
0
            std::string_view filename = info.file_name;
1298
0
            if (filename.ends_with(".meta")) {
1299
0
                return filename == remote_meta_path;
1300
0
            }
1301
0
            auto rowset_id = extract_rowset_id(filename);
1302
0
            if (rowset_id.hi == 0) {
1303
0
                return false;
1304
0
            }
1305
0
            return cooldowned_rowsets.contains(rowset_id) ||
1306
0
                   pending_remote_rowsets().contains(rowset_id);
1307
0
        };
1308
0
        files.erase(std::remove_if(files.begin(), files.end(), std::move(filter)), files.end());
1309
0
        if (files.empty()) {
1310
0
            return;
1311
0
        }
1312
0
        files.shrink_to_fit();
1313
0
        num_files_in_buffer += files.size();
1314
0
        buffer.insert({t->tablet_id(), {*storage_resource, std::move(files)}});
1315
0
        auto& info = req.confirm_list.emplace_back();
1316
0
        info.__set_tablet_id(t->tablet_id());
1317
0
        info.__set_cooldown_replica_id(cooldown_replica_id);
1318
0
        info.__set_cooldown_meta_id(cooldown_meta_id.to_thrift());
1319
0
    };
1320
1321
0
    auto confirm_and_remove_files = [&buffer, &req, &num_files_in_buffer]() {
1322
0
        TConfirmUnusedRemoteFilesResult result;
1323
0
        LOG(INFO) << "begin to confirm unused remote files. num_tablets=" << buffer.size()
1324
0
                  << " num_files=" << num_files_in_buffer;
1325
0
        auto st = MasterServerClient::instance()->confirm_unused_remote_files(req, &result);
1326
0
        if (!st.ok()) {
1327
0
            LOG(WARNING) << st;
1328
0
            return;
1329
0
        }
1330
0
        for (auto id : result.confirmed_tablets) {
1331
0
            if (auto it = buffer.find(id); LIKELY(it != buffer.end())) {
1332
0
                auto& storage_resource = it->second.first;
1333
0
                auto& files = it->second.second;
1334
0
                std::vector<io::Path> paths;
1335
0
                paths.reserve(files.size());
1336
                // delete unused files
1337
0
                LOG(INFO) << "delete unused files. root_path=" << storage_resource.fs->root_path()
1338
0
                          << " tablet_id=" << id;
1339
0
                io::Path dir = storage_resource.remote_tablet_path(id);
1340
0
                for (auto& file : files) {
1341
0
                    auto file_path = dir / file.file_name;
1342
0
                    LOG(INFO) << "delete unused file: " << file_path.native();
1343
0
                    paths.push_back(std::move(file_path));
1344
0
                }
1345
0
                st = storage_resource.fs->batch_delete(paths);
1346
0
                if (!st.ok()) {
1347
0
                    LOG(WARNING) << "failed to delete unused files, tablet_id=" << id << " : "
1348
0
                                 << st;
1349
0
                }
1350
0
                buffer.erase(it);
1351
0
            }
1352
0
        }
1353
0
    };
1354
1355
    // batch confirm to reduce FE's overhead
1356
0
    auto next_confirm_time = std::chrono::steady_clock::now() +
1357
0
                             std::chrono::seconds(config::confirm_unused_remote_files_interval_sec);
1358
0
    for (auto& t : tablets) {
1359
0
        if (t.use_count() <= 1 // this means tablet has been dropped
1360
0
            || t->cooldown_conf_unlocked().cooldown_replica_id != t->replica_id() ||
1361
0
            t->tablet_state() != TABLET_RUNNING) {
1362
0
            continue;
1363
0
        }
1364
0
        calc_unused_remote_files(t.get());
1365
0
        if (num_files_in_buffer > 0 && (num_files_in_buffer > max_files_in_buffer ||
1366
0
                                        std::chrono::steady_clock::now() > next_confirm_time)) {
1367
0
            confirm_and_remove_files();
1368
0
            buffer.clear();
1369
0
            req.confirm_list.clear();
1370
0
            num_files_in_buffer = 0;
1371
0
            next_confirm_time =
1372
0
                    std::chrono::steady_clock::now() +
1373
0
                    std::chrono::seconds(config::confirm_unused_remote_files_interval_sec);
1374
0
        }
1375
0
    }
1376
0
    if (num_files_in_buffer > 0) {
1377
0
        confirm_and_remove_files();
1378
0
    }
1379
0
}
1380
1381
0
void StorageEngine::_cold_data_compaction_producer_callback() {
1382
0
    while (!_stop_background_threads_latch.wait_for(
1383
0
            std::chrono::seconds(config::cold_data_compaction_interval_sec))) {
1384
0
        if (config::disable_auto_compaction ||
1385
0
            GlobalMemoryArbitrator::is_exceed_soft_mem_limit(GB_EXCHANGE_BYTE)) {
1386
0
            continue;
1387
0
        }
1388
1389
0
        std::unordered_set<int64_t> copied_tablet_submitted;
1390
0
        {
1391
0
            std::lock_guard lock(_cold_compaction_tablet_submitted_mtx);
1392
0
            copied_tablet_submitted = _cold_compaction_tablet_submitted;
1393
0
        }
1394
0
        int n = config::cold_data_compaction_thread_num - copied_tablet_submitted.size();
1395
0
        if (n <= 0) {
1396
0
            continue;
1397
0
        }
1398
0
        auto tablets = _tablet_manager->get_all_tablet([&copied_tablet_submitted](Tablet* t) {
1399
0
            return t->tablet_meta()->cooldown_meta_id().initialized() && t->is_used() &&
1400
0
                   t->tablet_state() == TABLET_RUNNING &&
1401
0
                   !copied_tablet_submitted.contains(t->tablet_id()) &&
1402
0
                   !t->tablet_meta()->tablet_schema()->disable_auto_compaction();
1403
0
        });
1404
0
        std::vector<std::pair<TabletSharedPtr, int64_t>> tablet_to_compact;
1405
0
        tablet_to_compact.reserve(n + 1);
1406
0
        std::vector<std::pair<TabletSharedPtr, int64_t>> tablet_to_follow;
1407
0
        tablet_to_follow.reserve(n + 1);
1408
1409
0
        for (auto& t : tablets) {
1410
0
            if (t->replica_id() == t->cooldown_conf_unlocked().cooldown_replica_id) {
1411
0
                auto score = t->calc_cold_data_compaction_score();
1412
0
                if (score < 4) {
1413
0
                    continue;
1414
0
                }
1415
0
                tablet_to_compact.emplace_back(t, score);
1416
0
                if (tablet_to_compact.size() > n) {
1417
0
                    std::sort(tablet_to_compact.begin(), tablet_to_compact.end(),
1418
0
                              [](auto& a, auto& b) { return a.second > b.second; });
1419
0
                    tablet_to_compact.pop_back();
1420
0
                }
1421
0
                continue;
1422
0
            }
1423
            // else, need to follow
1424
0
            {
1425
0
                std::lock_guard lock(_running_cooldown_mutex);
1426
0
                if (_running_cooldown_tablets.contains(t->table_id())) {
1427
                    // already in cooldown queue
1428
0
                    continue;
1429
0
                }
1430
0
            }
1431
            // TODO(plat1ko): some avoidance strategy if failed to follow
1432
0
            auto score = t->calc_cold_data_compaction_score();
1433
0
            tablet_to_follow.emplace_back(t, score);
1434
1435
0
            if (tablet_to_follow.size() > n) {
1436
0
                std::sort(tablet_to_follow.begin(), tablet_to_follow.end(),
1437
0
                          [](auto& a, auto& b) { return a.second > b.second; });
1438
0
                tablet_to_follow.pop_back();
1439
0
            }
1440
0
        }
1441
1442
0
        for (auto& [tablet, score] : tablet_to_compact) {
1443
0
            LOG(INFO) << "submit cold data compaction. tablet_id=" << tablet->tablet_id()
1444
0
                      << " score=" << score;
1445
0
            static_cast<void>(_cold_data_compaction_thread_pool->submit_func(
1446
0
                    [&, t = std::move(tablet), this]() {
1447
0
                        auto compaction = std::make_shared<ColdDataCompaction>(*this, t);
1448
0
                        {
1449
0
                            std::lock_guard lock(_cold_compaction_tablet_submitted_mtx);
1450
0
                            _cold_compaction_tablet_submitted.insert(t->tablet_id());
1451
0
                        }
1452
0
                        Defer defer {[&] {
1453
0
                            std::lock_guard lock(_cold_compaction_tablet_submitted_mtx);
1454
0
                            _cold_compaction_tablet_submitted.erase(t->tablet_id());
1455
0
                        }};
1456
0
                        std::unique_lock cold_compaction_lock(t->get_cold_compaction_lock(),
1457
0
                                                              std::try_to_lock);
1458
0
                        if (!cold_compaction_lock.owns_lock()) {
1459
0
                            LOG(WARNING) << "try cold_compaction_lock failed, tablet_id="
1460
0
                                         << t->tablet_id();
1461
0
                            return;
1462
0
                        }
1463
0
                        if (t->get_cumulative_compaction_policy() == nullptr ||
1464
0
                            t->get_cumulative_compaction_policy()->name() !=
1465
0
                                    t->tablet_meta()->compaction_policy()) {
1466
0
                            t->set_cumulative_compaction_policy(_cumulative_compaction_policies.at(
1467
0
                                    t->tablet_meta()->compaction_policy()));
1468
0
                        }
1469
1470
0
                        auto st = compaction->prepare_compact();
1471
0
                        if (!st.ok()) {
1472
0
                            LOG(WARNING) << "failed to prepare cold data compaction. tablet_id="
1473
0
                                         << t->tablet_id() << " err=" << st;
1474
0
                            return;
1475
0
                        }
1476
1477
0
                        st = compaction->execute_compact();
1478
0
                        if (!st.ok()) {
1479
0
                            LOG(WARNING) << "failed to execute cold data compaction. tablet_id="
1480
0
                                         << t->tablet_id() << " err=" << st;
1481
0
                            return;
1482
0
                        }
1483
0
                    }));
1484
0
        }
1485
1486
0
        for (auto& [tablet, score] : tablet_to_follow) {
1487
0
            LOG(INFO) << "submit to follow cooldown meta. tablet_id=" << tablet->tablet_id()
1488
0
                      << " score=" << score;
1489
0
            static_cast<void>(_cold_data_compaction_thread_pool->submit_func([&,
1490
0
                                                                              t = std::move(
1491
0
                                                                                      tablet)]() {
1492
0
                {
1493
0
                    std::lock_guard lock(_cold_compaction_tablet_submitted_mtx);
1494
0
                    _cold_compaction_tablet_submitted.insert(t->tablet_id());
1495
0
                }
1496
0
                auto st = t->cooldown();
1497
0
                {
1498
0
                    std::lock_guard lock(_cold_compaction_tablet_submitted_mtx);
1499
0
                    _cold_compaction_tablet_submitted.erase(t->tablet_id());
1500
0
                }
1501
0
                if (!st.ok()) {
1502
                    // The cooldown of the replica may be relatively slow
1503
                    // resulting in a short period of time where following cannot be successful
1504
0
                    LOG_EVERY_N(WARNING, 5)
1505
0
                            << "failed to cooldown. tablet_id=" << t->tablet_id() << " err=" << st;
1506
0
                }
1507
0
            }));
1508
0
        }
1509
0
    }
1510
0
}
1511
1512
void StorageEngine::add_async_publish_task(int64_t partition_id, int64_t tablet_id,
1513
                                           int64_t publish_version, int64_t transaction_id,
1514
2.05k
                                           bool is_recovery) {
1515
2.05k
    if (!is_recovery) {
1516
2.05k
        bool exists = false;
1517
2.05k
        {
1518
2.05k
            std::shared_lock<std::shared_mutex> rlock(_async_publish_lock);
1519
2.05k
            if (auto tablet_iter = _async_publish_tasks.find(tablet_id);
1520
2.05k
                tablet_iter != _async_publish_tasks.end()) {
1521
2.05k
                if (auto iter = tablet_iter->second.find(publish_version);
1522
2.05k
                    iter != tablet_iter->second.end()) {
1523
20
                    exists = true;
1524
20
                }
1525
2.05k
            }
1526
2.05k
        }
1527
2.05k
        if (exists) {
1528
20
            return;
1529
20
        }
1530
2.03k
        TabletSharedPtr tablet = tablet_manager()->get_tablet(tablet_id);
1531
2.03k
        if (tablet == nullptr) {
1532
0
            LOG(INFO) << "tablet may be dropped when add async publish task, tablet_id: "
1533
0
                      << tablet_id;
1534
0
            return;
1535
0
        }
1536
2.03k
        PendingPublishInfoPB pending_publish_info_pb;
1537
2.03k
        pending_publish_info_pb.set_partition_id(partition_id);
1538
2.03k
        pending_publish_info_pb.set_transaction_id(transaction_id);
1539
2.03k
        static_cast<void>(TabletMetaManager::save_pending_publish_info(
1540
2.03k
                tablet->data_dir(), tablet->tablet_id(), publish_version,
1541
2.03k
                pending_publish_info_pb.SerializeAsString()));
1542
2.03k
    }
1543
2.03k
    LOG(INFO) << "add pending publish task, tablet_id: " << tablet_id
1544
2.03k
              << " version: " << publish_version << " txn_id:" << transaction_id
1545
2.03k
              << " is_recovery: " << is_recovery;
1546
2.03k
    std::unique_lock<std::shared_mutex> wlock(_async_publish_lock);
1547
2.03k
    _async_publish_tasks[tablet_id][publish_version] = {transaction_id, partition_id};
1548
2.03k
}
1549
1550
3
int64_t StorageEngine::get_pending_publish_min_version(int64_t tablet_id) {
1551
3
    std::shared_lock<std::shared_mutex> rlock(_async_publish_lock);
1552
3
    auto iter = _async_publish_tasks.find(tablet_id);
1553
3
    if (iter == _async_publish_tasks.end()) {
1554
0
        return INT64_MAX;
1555
0
    }
1556
3
    if (iter->second.empty()) {
1557
0
        return INT64_MAX;
1558
0
    }
1559
3
    return iter->second.begin()->first;
1560
3
}
1561
1562
10
void StorageEngine::_process_async_publish() {
1563
    // tablet, publish_version
1564
10
    std::vector<std::pair<TabletSharedPtr, int64_t>> need_removed_tasks;
1565
10
    {
1566
10
        std::unique_lock<std::shared_mutex> wlock(_async_publish_lock);
1567
10
        for (auto tablet_iter = _async_publish_tasks.begin();
1568
20
             tablet_iter != _async_publish_tasks.end();) {
1569
10
            if (tablet_iter->second.empty()) {
1570
1
                tablet_iter = _async_publish_tasks.erase(tablet_iter);
1571
1
                continue;
1572
1
            }
1573
9
            int64_t tablet_id = tablet_iter->first;
1574
9
            TabletSharedPtr tablet = tablet_manager()->get_tablet(tablet_id);
1575
9
            if (!tablet) {
1576
1
                LOG(WARNING) << "tablet does not exist when async publush, tablet_id: "
1577
1
                             << tablet_id;
1578
1
                tablet_iter = _async_publish_tasks.erase(tablet_iter);
1579
1
                continue;
1580
1
            }
1581
1582
8
            auto task_iter = tablet_iter->second.begin();
1583
8
            int64_t version = task_iter->first;
1584
8
            int64_t transaction_id = task_iter->second.first;
1585
8
            int64_t partition_id = task_iter->second.second;
1586
8
            int64_t max_version = tablet->max_version().second;
1587
1588
8
            if (version <= max_version) {
1589
6
                need_removed_tasks.emplace_back(tablet, version);
1590
6
                tablet_iter->second.erase(task_iter);
1591
6
                tablet_iter++;
1592
6
                continue;
1593
6
            }
1594
2
            if (version != max_version + 1) {
1595
                // Keep only the most recent versions
1596
31
                while (tablet_iter->second.size() > config::max_tablet_version_num) {
1597
30
                    need_removed_tasks.emplace_back(tablet, version);
1598
30
                    task_iter = tablet_iter->second.erase(task_iter);
1599
30
                    version = task_iter->first;
1600
30
                }
1601
1
                tablet_iter++;
1602
1
                continue;
1603
1
            }
1604
1605
1
            auto async_publish_task = std::make_shared<AsyncTabletPublishTask>(
1606
1
                    *this, tablet, partition_id, transaction_id, version);
1607
1
            static_cast<void>(_tablet_publish_txn_thread_pool->submit_func(
1608
1
                    [=]() { async_publish_task->handle(); }));
1609
1
            tablet_iter->second.erase(task_iter);
1610
1
            need_removed_tasks.emplace_back(tablet, version);
1611
1
            tablet_iter++;
1612
1
        }
1613
10
    }
1614
37
    for (auto& [tablet, publish_version] : need_removed_tasks) {
1615
37
        static_cast<void>(TabletMetaManager::remove_pending_publish_info(
1616
37
                tablet->data_dir(), tablet->tablet_id(), publish_version));
1617
37
    }
1618
10
}
1619
1620
0
void StorageEngine::_async_publish_callback() {
1621
0
    while (!_stop_background_threads_latch.wait_for(std::chrono::milliseconds(30))) {
1622
0
        _process_async_publish();
1623
0
    }
1624
0
}
1625
1626
0
void StorageEngine::_check_tablet_delete_bitmap_score_callback() {
1627
0
    LOG(INFO) << "try to start check tablet delete bitmap score!";
1628
0
    while (!_stop_background_threads_latch.wait_for(
1629
0
            std::chrono::seconds(config::check_tablet_delete_bitmap_interval_seconds))) {
1630
0
        if (!config::enable_check_tablet_delete_bitmap_score) {
1631
0
            return;
1632
0
        }
1633
0
        uint64_t max_delete_bitmap_score = 0;
1634
0
        uint64_t max_base_rowset_delete_bitmap_score = 0;
1635
0
        std::vector<CloudTabletSPtr> tablets;
1636
0
        _tablet_manager.get()->get_topn_tablet_delete_bitmap_score(
1637
0
                &max_delete_bitmap_score, &max_base_rowset_delete_bitmap_score);
1638
0
        if (max_delete_bitmap_score > 0) {
1639
0
            _tablet_max_delete_bitmap_score_metrics->set_value(max_delete_bitmap_score);
1640
0
        }
1641
0
        if (max_base_rowset_delete_bitmap_score > 0) {
1642
0
            _tablet_max_base_rowset_delete_bitmap_score_metrics->set_value(
1643
0
                    max_base_rowset_delete_bitmap_score);
1644
0
        }
1645
0
    }
1646
0
}
1647
1648
} // namespace doris