Coverage Report

Created: 2025-04-28 18:24

/root/doris/be/src/olap/tablet_manager.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 "olap/tablet_manager.h"
19
20
#include <fmt/format.h>
21
#include <gen_cpp/AgentService_types.h>
22
#include <gen_cpp/BackendService_types.h>
23
#include <gen_cpp/Descriptors_types.h>
24
#include <gen_cpp/MasterService_types.h>
25
#include <gen_cpp/Types_types.h>
26
#include <gen_cpp/olap_file.pb.h>
27
#include <re2/re2.h>
28
#include <unistd.h>
29
30
#include <algorithm>
31
#include <list>
32
#include <mutex>
33
#include <ostream>
34
35
#include "bvar/bvar.h"
36
#include "common/compiler_util.h" // IWYU pragma: keep
37
#include "common/config.h"
38
#include "common/logging.h"
39
#include "gutil/integral_types.h"
40
#include "gutil/strings/strcat.h"
41
#include "gutil/strings/substitute.h"
42
#include "io/fs/local_file_system.h"
43
#include "olap/cumulative_compaction_time_series_policy.h"
44
#include "olap/data_dir.h"
45
#include "olap/olap_common.h"
46
#include "olap/olap_define.h"
47
#include "olap/olap_meta.h"
48
#include "olap/pb_helper.h"
49
#include "olap/rowset/beta_rowset.h"
50
#include "olap/rowset/rowset.h"
51
#include "olap/rowset/rowset_meta_manager.h"
52
#include "olap/storage_engine.h"
53
#include "olap/tablet.h"
54
#include "olap/tablet_meta.h"
55
#include "olap/tablet_meta_manager.h"
56
#include "olap/tablet_schema.h"
57
#include "olap/txn_manager.h"
58
#include "runtime/exec_env.h"
59
#include "runtime/memory/mem_tracker.h"
60
#include "runtime/thread_context.h"
61
#include "service/backend_options.h"
62
#include "util/defer_op.h"
63
#include "util/doris_metrics.h"
64
#include "util/histogram.h"
65
#include "util/metrics.h"
66
#include "util/path_util.h"
67
#include "util/scoped_cleanup.h"
68
#include "util/stopwatch.hpp"
69
#include "util/time.h"
70
#include "util/trace.h"
71
#include "util/uid_util.h"
72
73
namespace doris {
74
class CumulativeCompactionPolicy;
75
} // namespace doris
76
77
using std::map;
78
using std::set;
79
using std::string;
80
using std::vector;
81
82
namespace doris {
83
using namespace ErrorCode;
84
85
DEFINE_GAUGE_METRIC_PROTOTYPE_5ARG(tablet_meta_mem_consumption, MetricUnit::BYTES, "",
86
                                   mem_consumption, Labels({{"type", "tablet_meta"}}));
87
88
bvar::Adder<int64_t> g_tablet_meta_schema_columns_count("tablet_meta_schema_columns_count");
89
90
TabletManager::TabletManager(StorageEngine& engine, int32_t tablet_map_lock_shard_size)
91
        : _engine(engine),
92
          _tablet_meta_mem_tracker(std::make_shared<MemTracker>(
93
                  "TabletMeta(experimental)", ExecEnv::GetInstance()->details_mem_tracker_set())),
94
          _tablets_shards_size(tablet_map_lock_shard_size),
95
136
          _tablets_shards_mask(tablet_map_lock_shard_size - 1) {
96
136
    CHECK_GT(_tablets_shards_size, 0);
97
136
    CHECK_EQ(_tablets_shards_size & _tablets_shards_mask, 0);
98
136
    _tablets_shards.resize(_tablets_shards_size);
99
136
    REGISTER_HOOK_METRIC(tablet_meta_mem_consumption,
100
136
                         [this]() { return _tablet_meta_mem_tracker->consumption(); });
101
136
}
102
103
134
TabletManager::~TabletManager() {
104
134
    DEREGISTER_HOOK_METRIC(tablet_meta_mem_consumption);
105
134
}
106
107
Status TabletManager::_add_tablet_unlocked(TTabletId tablet_id, const TabletSharedPtr& tablet,
108
297
                                           bool update_meta, bool force, RuntimeProfile* profile) {
109
297
    if (profile->get_counter("AddTablet") == nullptr) {
110
2
        ADD_TIMER(profile, "AddTablet");
111
2
    }
112
297
    Status res = Status::OK();
113
297
    VLOG_NOTICE << "begin to add tablet to TabletManager. "
114
0
                << "tablet_id=" << tablet_id << ", force=" << force;
115
116
297
    TabletSharedPtr existed_tablet = nullptr;
117
297
    tablet_map_t& tablet_map = _get_tablet_map(tablet_id);
118
297
    const auto& iter = tablet_map.find(tablet_id);
119
297
    if (iter != tablet_map.end()) {
120
2
        existed_tablet = iter->second;
121
2
    }
122
123
297
    if (existed_tablet == nullptr) {
124
295
        return _add_tablet_to_map_unlocked(tablet_id, tablet, update_meta, false /*keep_files*/,
125
295
                                           false /*drop_old*/, profile);
126
295
    }
127
    // During restore process, the tablet is exist and snapshot loader will replace the tablet's rowsets
128
    // and then reload the tablet, the tablet's path will the same
129
2
    if (!force) {
130
2
        if (existed_tablet->tablet_path() == tablet->tablet_path()) {
131
0
            return Status::Error<ENGINE_INSERT_EXISTS_TABLE>(
132
0
                    "add the same tablet twice! tablet_id={}, tablet_path={}", tablet_id,
133
0
                    tablet->tablet_path());
134
0
        }
135
2
        if (existed_tablet->data_dir() == tablet->data_dir()) {
136
0
            return Status::Error<ENGINE_INSERT_EXISTS_TABLE>(
137
0
                    "add tablet with same data dir twice! tablet_id={}", tablet_id);
138
0
        }
139
2
    }
140
141
2
    MonotonicStopWatch watch;
142
2
    watch.start();
143
144
    // During storage migration, the tablet is moved to another disk, have to check
145
    // if the new tablet's rowset version is larger than the old one to prevent losting data during
146
    // migration
147
2
    int64_t old_time, new_time;
148
2
    int32_t old_version, new_version;
149
2
    {
150
2
        std::shared_lock rdlock(existed_tablet->get_header_lock());
151
2
        const RowsetSharedPtr old_rowset = existed_tablet->rowset_with_max_version();
152
2
        const RowsetSharedPtr new_rowset = tablet->rowset_with_max_version();
153
        // If new tablet is empty, it is a newly created schema change tablet.
154
        // the old tablet is dropped before add tablet. it should not exist old tablet
155
2
        if (new_rowset == nullptr) {
156
            // it seems useless to call unlock and return here.
157
            // it could prevent error when log level is changed in the future.
158
0
            return Status::Error<ENGINE_INSERT_EXISTS_TABLE>(
159
0
                    "new tablet is empty and old tablet exists. it should not happen. tablet_id={}",
160
0
                    tablet_id);
161
0
        }
162
2
        old_time = old_rowset == nullptr ? -1 : old_rowset->creation_time();
163
2
        new_time = new_rowset->creation_time();
164
2
        old_version = old_rowset == nullptr ? -1 : old_rowset->end_version();
165
2
        new_version = new_rowset->end_version();
166
2
    }
167
2
    COUNTER_UPDATE(ADD_CHILD_TIMER(profile, "GetExistTabletVersion", "AddTablet"),
168
2
                   static_cast<int64_t>(watch.reset()));
169
170
    // In restore process, we replace all origin files in tablet dir with
171
    // the downloaded snapshot files. Then we try to reload tablet header.
172
    // force == true means we forcibly replace the Tablet in tablet_map
173
    // with the new one. But if we do so, the files in the tablet dir will be
174
    // dropped when the origin Tablet deconstruct.
175
    // So we set keep_files == true to not delete files when the
176
    // origin Tablet deconstruct.
177
    // During restore process, snapshot loader
178
    // replaced the old tablet's rowset with new rowsets, but the tablet path is reused, if drop files
179
    // here, the new rowset's file will also be dropped, so use keep files here
180
2
    bool keep_files = force;
181
2
    if (force ||
182
2
        (new_version > old_version || (new_version == old_version && new_time >= old_time))) {
183
        // check if new tablet's meta is in store and add new tablet's meta to meta store
184
2
        res = _add_tablet_to_map_unlocked(tablet_id, tablet, update_meta, keep_files,
185
2
                                          true /*drop_old*/, profile);
186
2
    } else {
187
0
        RETURN_IF_ERROR(tablet->set_tablet_state(TABLET_SHUTDOWN));
188
0
        tablet->save_meta();
189
0
        COUNTER_UPDATE(ADD_CHILD_TIMER(profile, "SaveMeta", "AddTablet"),
190
0
                       static_cast<int64_t>(watch.reset()));
191
0
        {
192
0
            std::lock_guard<std::shared_mutex> shutdown_tablets_wrlock(_shutdown_tablets_lock);
193
0
            _shutdown_tablets.push_back(tablet);
194
0
        }
195
196
0
        res = Status::Error<ENGINE_INSERT_OLD_TABLET>(
197
0
                "set tablet to shutdown state. tablet_id={}, tablet_path={}", tablet->tablet_id(),
198
0
                tablet->tablet_path());
199
0
    }
200
2
    LOG(WARNING) << "add duplicated tablet. force=" << force << ", res=" << res
201
2
                 << ", tablet_id=" << tablet_id << ", old_version=" << old_version
202
2
                 << ", new_version=" << new_version << ", old_time=" << old_time
203
2
                 << ", new_time=" << new_time
204
2
                 << ", old_tablet_path=" << existed_tablet->tablet_path()
205
2
                 << ", new_tablet_path=" << tablet->tablet_path();
206
207
2
    return res;
208
2
}
209
210
Status TabletManager::_add_tablet_to_map_unlocked(TTabletId tablet_id,
211
                                                  const TabletSharedPtr& tablet, bool update_meta,
212
                                                  bool keep_files, bool drop_old,
213
297
                                                  RuntimeProfile* profile) {
214
    // check if new tablet's meta is in store and add new tablet's meta to meta store
215
297
    Status res = Status::OK();
216
297
    MonotonicStopWatch watch;
217
297
    watch.start();
218
297
    if (update_meta) {
219
        // call tablet save meta in order to valid the meta
220
297
        tablet->save_meta();
221
297
        COUNTER_UPDATE(ADD_CHILD_TIMER(profile, "SaveMeta", "AddTablet"),
222
297
                       static_cast<int64_t>(watch.reset()));
223
297
    }
224
297
    if (drop_old) {
225
        // If the new tablet is fresher than the existing one, then replace
226
        // the existing tablet with the new one.
227
        // Use default replica_id to ignore whether replica_id is match when drop tablet.
228
2
        Status status = _drop_tablet_unlocked(tablet_id, /* replica_id */ 0, keep_files, false);
229
2
        COUNTER_UPDATE(ADD_CHILD_TIMER(profile, "DropOldTablet", "AddTablet"),
230
2
                       static_cast<int64_t>(watch.reset()));
231
2
        RETURN_NOT_OK_STATUS_WITH_WARN(
232
2
                status, strings::Substitute("failed to drop old tablet when add new tablet. "
233
2
                                            "tablet_id=$0",
234
2
                                            tablet_id));
235
2
    }
236
    // Register tablet into DataDir, so that we can manage tablet from
237
    // the perspective of root path.
238
    // Example: unregister all tables when a bad disk found.
239
297
    tablet->register_tablet_into_dir();
240
297
    tablet_map_t& tablet_map = _get_tablet_map(tablet_id);
241
297
    tablet_map[tablet_id] = tablet;
242
297
    _add_tablet_to_partition(tablet);
243
    // TODO: remove multiply 2 of tablet meta mem size
244
    // Because table schema will copy in tablet, there will be double mem cost
245
    // so here multiply 2
246
297
    _tablet_meta_mem_tracker->consume(tablet->tablet_meta()->mem_size() * 2);
247
297
    g_tablet_meta_schema_columns_count << tablet->tablet_meta()->tablet_columns_num();
248
297
    COUNTER_UPDATE(ADD_CHILD_TIMER(profile, "RegisterTabletInfo", "AddTablet"),
249
297
                   static_cast<int64_t>(watch.reset()));
250
251
297
    VLOG_NOTICE << "add tablet to map successfully."
252
0
                << " tablet_id=" << tablet_id;
253
254
297
    return res;
255
297
}
256
257
0
bool TabletManager::check_tablet_id_exist(TTabletId tablet_id) {
258
0
    std::shared_lock rdlock(_get_tablets_shard_lock(tablet_id));
259
0
    return _check_tablet_id_exist_unlocked(tablet_id);
260
0
}
261
262
0
bool TabletManager::_check_tablet_id_exist_unlocked(TTabletId tablet_id) {
263
0
    tablet_map_t& tablet_map = _get_tablet_map(tablet_id);
264
0
    return tablet_map.find(tablet_id) != tablet_map.end();
265
0
}
266
267
Status TabletManager::create_tablet(const TCreateTabletReq& request, std::vector<DataDir*> stores,
268
298
                                    RuntimeProfile* profile) {
269
298
    DorisMetrics::instance()->create_tablet_requests_total->increment(1);
270
271
298
    int64_t tablet_id = request.tablet_id;
272
298
    LOG(INFO) << "begin to create tablet. tablet_id=" << tablet_id
273
298
              << ", table_id=" << request.table_id << ", partition_id=" << request.partition_id
274
298
              << ", replica_id=" << request.replica_id << ", stores.size=" << stores.size()
275
298
              << ", first store=" << stores[0]->path();
276
277
    // when we create rollup tablet A(assume on shard-1) from tablet B(assume on shard-2)
278
    // we need use write lock on shard-1 and then use read lock on shard-2
279
    // if there have create rollup tablet C(assume on shard-2) from tablet D(assume on shard-1) at the same time, we will meet deadlock
280
298
    std::unique_lock two_tablet_lock(_two_tablet_mtx, std::defer_lock);
281
298
    bool in_restore_mode = request.__isset.in_restore_mode && request.in_restore_mode;
282
298
    bool is_schema_change_or_atomic_restore =
283
298
            request.__isset.base_tablet_id && request.base_tablet_id > 0;
284
298
    bool need_two_lock =
285
298
            is_schema_change_or_atomic_restore &&
286
298
            ((_tablets_shards_mask & request.base_tablet_id) != (_tablets_shards_mask & tablet_id));
287
298
    if (need_two_lock) {
288
0
        SCOPED_TIMER(ADD_TIMER(profile, "GetTwoTableLock"));
289
0
        two_tablet_lock.lock();
290
0
    }
291
292
298
    MonotonicStopWatch shard_lock_watch;
293
298
    shard_lock_watch.start();
294
298
    std::lock_guard wrlock(_get_tablets_shard_lock(tablet_id));
295
298
    shard_lock_watch.stop();
296
298
    COUNTER_UPDATE(ADD_TIMER(profile, "GetShardLock"),
297
298
                   static_cast<int64_t>(shard_lock_watch.elapsed_time()));
298
    // Make create_tablet operation to be idempotent:
299
    // 1. Return true if tablet with same tablet_id and schema_hash exist;
300
    //           false if tablet with same tablet_id but different schema_hash exist.
301
    // 2. When this is an alter task, if the tablet(both tablet_id and schema_hash are
302
    // same) already exist, then just return true(an duplicate request). But if
303
    // tablet_id exist but with different schema_hash, return an error(report task will
304
    // eventually trigger its deletion).
305
298
    {
306
298
        SCOPED_TIMER(ADD_TIMER(profile, "GetTabletUnlocked"));
307
298
        if (_get_tablet_unlocked(tablet_id) != nullptr) {
308
3
            LOG(INFO) << "success to create tablet. tablet already exist. tablet_id=" << tablet_id;
309
3
            return Status::OK();
310
3
        }
311
298
    }
312
313
295
    TabletSharedPtr base_tablet = nullptr;
314
    // If the CreateTabletReq has base_tablet_id then it is a alter-tablet request
315
295
    if (is_schema_change_or_atomic_restore) {
316
        // if base_tablet_id's lock diffrent with new_tablet_id, we need lock it.
317
0
        if (need_two_lock) {
318
0
            SCOPED_TIMER(ADD_TIMER(profile, "GetBaseTablet"));
319
0
            base_tablet = get_tablet(request.base_tablet_id);
320
0
            two_tablet_lock.unlock();
321
0
        } else {
322
0
            SCOPED_TIMER(ADD_TIMER(profile, "GetBaseTabletUnlocked"));
323
0
            base_tablet = _get_tablet_unlocked(request.base_tablet_id);
324
0
        }
325
0
        if (base_tablet == nullptr) {
326
0
            DorisMetrics::instance()->create_tablet_requests_failed->increment(1);
327
0
            return Status::Error<TABLE_CREATE_META_ERROR>(
328
0
                    "fail to create tablet(change schema/atomic restore), base tablet does not "
329
0
                    "exist. new_tablet_id={}, base_tablet_id={}",
330
0
                    tablet_id, request.base_tablet_id);
331
0
        }
332
        // If we are doing schema-change or atomic-restore, we should use the same data dir
333
        // TODO(lingbin): A litter trick here, the directory should be determined before
334
        // entering this method
335
        //
336
        // ATTN: Since all restored replicas will be saved to HDD, so no storage_medium check here.
337
0
        if (in_restore_mode ||
338
0
            request.storage_medium == base_tablet->data_dir()->storage_medium()) {
339
0
            LOG(INFO) << "create tablet use the base tablet data dir. tablet_id=" << tablet_id
340
0
                      << ", base tablet_id=" << request.base_tablet_id
341
0
                      << ", data dir=" << base_tablet->data_dir()->path();
342
0
            stores.clear();
343
0
            stores.push_back(base_tablet->data_dir());
344
0
        }
345
0
    }
346
347
    // set alter type to schema-change. it is useless
348
295
    TabletSharedPtr tablet = _internal_create_tablet_unlocked(
349
295
            request, is_schema_change_or_atomic_restore, base_tablet.get(), stores, profile);
350
295
    if (tablet == nullptr) {
351
0
        DorisMetrics::instance()->create_tablet_requests_failed->increment(1);
352
0
        return Status::Error<CE_CMD_PARAMS_ERROR>("fail to create tablet. tablet_id={}",
353
0
                                                  request.tablet_id);
354
0
    }
355
356
295
    LOG(INFO) << "success to create tablet. tablet_id=" << tablet_id
357
295
              << ", tablet_path=" << tablet->tablet_path();
358
295
    return Status::OK();
359
295
}
360
361
TabletSharedPtr TabletManager::_internal_create_tablet_unlocked(
362
        const TCreateTabletReq& request, const bool is_schema_change, const Tablet* base_tablet,
363
295
        const std::vector<DataDir*>& data_dirs, RuntimeProfile* profile) {
364
    // If in schema-change state, base_tablet must also be provided.
365
    // i.e., is_schema_change and base_tablet are either assigned or not assigned
366
295
    DCHECK((is_schema_change && base_tablet) || (!is_schema_change && !base_tablet));
367
368
    // NOTE: The existence of tablet_id and schema_hash has already been checked,
369
    // no need check again here.
370
371
295
    const std::string parent_timer_name = "InternalCreateTablet";
372
295
    SCOPED_TIMER(ADD_TIMER(profile, parent_timer_name));
373
374
295
    MonotonicStopWatch watch;
375
295
    watch.start();
376
295
    auto create_meta_timer = ADD_CHILD_TIMER(profile, "CreateMeta", parent_timer_name);
377
295
    auto tablet = _create_tablet_meta_and_dir_unlocked(request, is_schema_change, base_tablet,
378
295
                                                       data_dirs, profile);
379
295
    COUNTER_UPDATE(create_meta_timer, static_cast<int64_t>(watch.reset()));
380
295
    if (tablet == nullptr) {
381
0
        return nullptr;
382
0
    }
383
384
295
    int64_t new_tablet_id = request.tablet_id;
385
295
    int32_t new_schema_hash = request.tablet_schema.schema_hash;
386
387
    // should remove the tablet's pending_id no matter create-tablet success or not
388
295
    DataDir* data_dir = tablet->data_dir();
389
390
    // TODO(yiguolei)
391
    // the following code is very difficult to understand because it mixed alter tablet v2
392
    // and alter tablet v1 should remove alter tablet v1 code after v0.12
393
295
    Status res = Status::OK();
394
295
    bool is_tablet_added = false;
395
295
    do {
396
295
        res = tablet->init();
397
295
        COUNTER_UPDATE(ADD_CHILD_TIMER(profile, "TabletInit", parent_timer_name),
398
295
                       static_cast<int64_t>(watch.reset()));
399
295
        if (!res.ok()) {
400
0
            LOG(WARNING) << "tablet init failed. tablet:" << tablet->tablet_id();
401
0
            break;
402
0
        }
403
404
        // Create init version if this is not a restore mode replica and request.version is set
405
        // bool in_restore_mode = request.__isset.in_restore_mode && request.in_restore_mode;
406
        // if (!in_restore_mode && request.__isset.version) {
407
        // create initial rowset before add it to storage engine could omit many locks
408
295
        res = tablet->create_initial_rowset(request.version);
409
295
        COUNTER_UPDATE(ADD_CHILD_TIMER(profile, "InitRowset", parent_timer_name),
410
295
                       static_cast<int64_t>(watch.reset()));
411
295
        if (!res.ok()) {
412
0
            LOG(WARNING) << "fail to create initial version for tablet. res=" << res;
413
0
            break;
414
0
        }
415
416
295
        if (is_schema_change) {
417
            // if this is a new alter tablet, has to set its state to not ready
418
            // because schema change handler depends on it to check whether history data
419
            // convert finished
420
0
            static_cast<void>(tablet->set_tablet_state(TabletState::TABLET_NOTREADY));
421
0
        }
422
        // Add tablet to StorageEngine will make it visible to user
423
        // Will persist tablet meta
424
295
        auto add_tablet_timer = ADD_CHILD_TIMER(profile, "AddTablet", parent_timer_name);
425
295
        res = _add_tablet_unlocked(new_tablet_id, tablet, /*update_meta*/ true, false, profile);
426
295
        COUNTER_UPDATE(add_tablet_timer, static_cast<int64_t>(watch.reset()));
427
295
        if (!res.ok()) {
428
0
            LOG(WARNING) << "fail to add tablet to StorageEngine. res=" << res;
429
0
            break;
430
0
        }
431
295
        is_tablet_added = true;
432
433
        // TODO(lingbin): The following logic seems useless, can be removed?
434
        // Because if _add_tablet_unlocked() return OK, we must can get it from map.
435
295
        TabletSharedPtr tablet_ptr = _get_tablet_unlocked(new_tablet_id);
436
295
        COUNTER_UPDATE(ADD_CHILD_TIMER(profile, "GetTablet", parent_timer_name),
437
295
                       static_cast<int64_t>(watch.reset()));
438
295
        if (tablet_ptr == nullptr) {
439
0
            res = Status::Error<TABLE_NOT_FOUND>("fail to get tablet. res={}", res);
440
0
            break;
441
0
        }
442
295
    } while (false);
443
444
295
    if (res.ok()) {
445
295
        return tablet;
446
295
    }
447
    // something is wrong, we need clear environment
448
0
    if (is_tablet_added) {
449
0
        Status status = _drop_tablet_unlocked(new_tablet_id, request.replica_id, false, false);
450
0
        COUNTER_UPDATE(ADD_CHILD_TIMER(profile, "DropTablet", parent_timer_name),
451
0
                       static_cast<int64_t>(watch.reset()));
452
0
        if (!status.ok()) {
453
0
            LOG(WARNING) << "fail to drop tablet when create tablet failed. res=" << res;
454
0
        }
455
0
    } else {
456
0
        tablet->delete_all_files();
457
0
        static_cast<void>(TabletMetaManager::remove(data_dir, new_tablet_id, new_schema_hash));
458
0
        COUNTER_UPDATE(ADD_CHILD_TIMER(profile, "RemoveTabletFiles", parent_timer_name),
459
0
                       static_cast<int64_t>(watch.reset()));
460
0
    }
461
0
    return nullptr;
462
295
}
463
464
295
static string _gen_tablet_dir(const string& dir, int16_t shard_id, int64_t tablet_id) {
465
295
    string path = dir;
466
295
    path = path_util::join_path_segments(path, DATA_PREFIX);
467
295
    path = path_util::join_path_segments(path, std::to_string(shard_id));
468
295
    path = path_util::join_path_segments(path, std::to_string(tablet_id));
469
295
    return path;
470
295
}
471
472
TabletSharedPtr TabletManager::_create_tablet_meta_and_dir_unlocked(
473
        const TCreateTabletReq& request, const bool is_schema_change, const Tablet* base_tablet,
474
295
        const std::vector<DataDir*>& data_dirs, RuntimeProfile* profile) {
475
295
    string pending_id = StrCat(TABLET_ID_PREFIX, request.tablet_id);
476
    // Many attempts are made here in the hope that even if a disk fails, it can still continue.
477
295
    std::string parent_timer_name = "CreateMeta";
478
295
    MonotonicStopWatch watch;
479
295
    watch.start();
480
295
    for (auto& data_dir : data_dirs) {
481
295
        COUNTER_UPDATE(ADD_CHILD_TIMER(profile, "RemovePendingIds", parent_timer_name),
482
295
                       static_cast<int64_t>(watch.reset()));
483
484
295
        TabletMetaSharedPtr tablet_meta;
485
        // if create meta failed, do not need to clean dir, because it is only in memory
486
295
        Status res = _create_tablet_meta_unlocked(request, data_dir, is_schema_change, base_tablet,
487
295
                                                  &tablet_meta);
488
295
        COUNTER_UPDATE(ADD_CHILD_TIMER(profile, "CreateMetaUnlock", parent_timer_name),
489
295
                       static_cast<int64_t>(watch.reset()));
490
295
        if (!res.ok()) {
491
0
            LOG(WARNING) << "fail to create tablet meta. res=" << res
492
0
                         << ", root=" << data_dir->path();
493
0
            continue;
494
0
        }
495
496
295
        string tablet_dir =
497
295
                _gen_tablet_dir(data_dir->path(), tablet_meta->shard_id(), request.tablet_id);
498
295
        string schema_hash_dir = path_util::join_path_segments(
499
295
                tablet_dir, std::to_string(request.tablet_schema.schema_hash));
500
501
        // Because the tablet is removed asynchronously, so that the dir may still exist when BE
502
        // receive create-tablet request again, For example retried schema-change request
503
295
        bool exists = true;
504
295
        res = io::global_local_filesystem()->exists(schema_hash_dir, &exists);
505
295
        if (!res.ok()) {
506
0
            continue;
507
0
        }
508
295
        if (exists) {
509
0
            LOG(WARNING) << "skip this dir because tablet path exist, path=" << schema_hash_dir;
510
0
            continue;
511
295
        } else {
512
295
            Status st = io::global_local_filesystem()->create_directory(schema_hash_dir);
513
295
            if (!st.ok()) {
514
0
                continue;
515
0
            }
516
295
        }
517
518
295
        if (tablet_meta->partition_id() <= 0) {
519
231
            LOG(WARNING) << "invalid partition id " << tablet_meta->partition_id() << ", tablet "
520
231
                         << tablet_meta->tablet_id();
521
231
        }
522
295
        TabletSharedPtr new_tablet =
523
295
                std::make_shared<Tablet>(_engine, std::move(tablet_meta), data_dir);
524
295
        COUNTER_UPDATE(ADD_CHILD_TIMER(profile, "CreateTabletFromMeta", parent_timer_name),
525
295
                       static_cast<int64_t>(watch.reset()));
526
295
        return new_tablet;
527
295
    }
528
0
    return nullptr;
529
295
}
530
531
Status TabletManager::drop_tablet(TTabletId tablet_id, TReplicaId replica_id,
532
251
                                  bool is_drop_table_or_partition) {
533
251
    auto& shard = _get_tablets_shard(tablet_id);
534
251
    std::lock_guard wrlock(shard.lock);
535
251
    return _drop_tablet_unlocked(tablet_id, replica_id, false, is_drop_table_or_partition);
536
251
}
537
538
// Drop specified tablet.
539
Status TabletManager::_drop_tablet_unlocked(TTabletId tablet_id, TReplicaId replica_id,
540
253
                                            bool keep_files, bool is_drop_table_or_partition) {
541
253
    LOG(INFO) << "begin drop tablet. tablet_id=" << tablet_id << ", replica_id=" << replica_id
542
253
              << ", is_drop_table_or_partition=" << is_drop_table_or_partition;
543
253
    DorisMetrics::instance()->drop_tablet_requests_total->increment(1);
544
545
253
    RETURN_IF_ERROR(register_transition_tablet(tablet_id, "drop tablet"));
546
253
    Defer defer {[&]() { unregister_transition_tablet(tablet_id, "drop tablet"); }};
547
548
    // Fetch tablet which need to be dropped
549
253
    TabletSharedPtr to_drop_tablet = _get_tablet_unlocked(tablet_id);
550
253
    if (to_drop_tablet == nullptr) {
551
2
        LOG(WARNING) << "fail to drop tablet because it does not exist. "
552
2
                     << "tablet_id=" << tablet_id;
553
2
        return Status::OK();
554
2
    }
555
556
    // We should compare replica id to avoid dropping new cloned tablet.
557
    // Iff request replica id is 0, FE may be an older release, then we drop this tablet as before.
558
251
    if (to_drop_tablet->replica_id() != replica_id && replica_id != 0) {
559
0
        return Status::Aborted("replica_id not match({} vs {})", to_drop_tablet->replica_id(),
560
0
                               replica_id);
561
0
    }
562
563
251
    _remove_tablet_from_partition(to_drop_tablet);
564
251
    tablet_map_t& tablet_map = _get_tablet_map(tablet_id);
565
251
    tablet_map.erase(tablet_id);
566
567
251
    to_drop_tablet->clear_cache();
568
569
251
    if (!keep_files) {
570
        // drop tablet will update tablet meta, should lock
571
251
        std::lock_guard<std::shared_mutex> wrlock(to_drop_tablet->get_header_lock());
572
251
        SCOPED_SIMPLE_TRACE_IF_TIMEOUT(TRACE_TABLET_LOCK_THRESHOLD);
573
251
        LOG(INFO) << "set tablet to shutdown state and remove it from memory. "
574
251
                  << "tablet_id=" << tablet_id << ", tablet_path=" << to_drop_tablet->tablet_path();
575
        // NOTE: has to update tablet here, but must not update tablet meta directly.
576
        // because other thread may hold the tablet object, they may save meta too.
577
        // If update meta directly here, other thread may override the meta
578
        // and the tablet will be loaded at restart time.
579
        // To avoid this exception, we first set the state of the tablet to `SHUTDOWN`.
580
251
        RETURN_IF_ERROR(to_drop_tablet->set_tablet_state(TABLET_SHUTDOWN));
581
        // We must record unused remote rowsets path info to OlapMeta before tablet state is marked as TABLET_SHUTDOWN in OlapMeta,
582
        // otherwise if BE shutdown after saving tablet state, these remote rowsets path info will lost.
583
251
        if (is_drop_table_or_partition) {
584
0
            RETURN_IF_ERROR(to_drop_tablet->remove_all_remote_rowsets());
585
0
        }
586
251
        to_drop_tablet->save_meta();
587
251
        {
588
251
            std::lock_guard<std::shared_mutex> wrdlock(_shutdown_tablets_lock);
589
251
            _shutdown_tablets.push_back(to_drop_tablet);
590
251
        }
591
251
    }
592
593
251
    to_drop_tablet->deregister_tablet_from_dir();
594
251
    _tablet_meta_mem_tracker->release(to_drop_tablet->tablet_meta()->mem_size() * 2);
595
251
    g_tablet_meta_schema_columns_count << -to_drop_tablet->tablet_meta()->tablet_columns_num();
596
251
    return Status::OK();
597
251
}
598
599
2.71k
TabletSharedPtr TabletManager::get_tablet(TTabletId tablet_id, bool include_deleted, string* err) {
600
2.71k
    std::shared_lock rdlock(_get_tablets_shard_lock(tablet_id));
601
2.71k
    return _get_tablet_unlocked(tablet_id, include_deleted, err);
602
2.71k
}
603
604
35
std::vector<TabletSharedPtr> TabletManager::get_all_tablet(std::function<bool(Tablet*)>&& filter) {
605
35
    std::vector<TabletSharedPtr> res;
606
35
    for_each_tablet([&](const TabletSharedPtr& tablet) { res.emplace_back(tablet); },
607
35
                    std::move(filter));
608
35
    return res;
609
35
}
610
611
void TabletManager::for_each_tablet(std::function<void(const TabletSharedPtr&)>&& handler,
612
136
                                    std::function<bool(Tablet*)>&& filter) {
613
136
    std::vector<TabletSharedPtr> tablets;
614
136
    for (const auto& tablets_shard : _tablets_shards) {
615
136
        tablets.clear();
616
136
        {
617
136
            std::shared_lock rdlock(tablets_shard.lock);
618
256
            for (const auto& [id, tablet] : tablets_shard.tablet_map) {
619
256
                if (filter(tablet.get())) {
620
256
                    tablets.emplace_back(tablet);
621
256
                }
622
256
            }
623
136
        }
624
256
        for (const auto& tablet : tablets) {
625
256
            handler(tablet);
626
256
        }
627
136
    }
628
136
}
629
630
TabletSharedPtr TabletManager::_get_tablet_unlocked(TTabletId tablet_id, bool include_deleted,
631
2.71k
                                                    string* err) {
632
2.71k
    TabletSharedPtr tablet;
633
2.71k
    tablet = _get_tablet_unlocked(tablet_id);
634
2.71k
    if (tablet == nullptr && include_deleted) {
635
4
        std::shared_lock rdlock(_shutdown_tablets_lock);
636
4
        for (auto& deleted_tablet : _shutdown_tablets) {
637
2
            CHECK(deleted_tablet != nullptr) << "deleted tablet is nullptr";
638
2
            if (deleted_tablet->tablet_id() == tablet_id) {
639
2
                tablet = deleted_tablet;
640
2
                break;
641
2
            }
642
2
        }
643
4
    }
644
645
2.71k
    if (tablet == nullptr) {
646
239
        if (err != nullptr) {
647
1
            *err = "tablet does not exist. " + BackendOptions::get_localhost();
648
1
        }
649
239
        return nullptr;
650
239
    }
651
#ifndef BE_TEST
652
    if (!tablet->is_used()) {
653
        LOG(WARNING) << "tablet cannot be used. tablet=" << tablet_id;
654
        if (err != nullptr) {
655
            *err = "tablet cannot be used. " + BackendOptions::get_localhost();
656
        }
657
        return nullptr;
658
    }
659
#endif
660
661
2.47k
    return tablet;
662
2.71k
}
663
664
TabletSharedPtr TabletManager::get_tablet(TTabletId tablet_id, TabletUid tablet_uid,
665
0
                                          bool include_deleted, string* err) {
666
0
    std::shared_lock rdlock(_get_tablets_shard_lock(tablet_id));
667
0
    TabletSharedPtr tablet = _get_tablet_unlocked(tablet_id, include_deleted, err);
668
0
    if (tablet != nullptr && tablet->tablet_uid() == tablet_uid) {
669
0
        return tablet;
670
0
    }
671
0
    return nullptr;
672
0
}
673
674
0
uint64_t TabletManager::get_rowset_nums() {
675
0
    uint64_t rowset_nums = 0;
676
0
    for_each_tablet([&](const TabletSharedPtr& tablet) { rowset_nums += tablet->version_count(); },
677
0
                    filter_all_tablets);
678
0
    return rowset_nums;
679
0
}
680
681
0
uint64_t TabletManager::get_segment_nums() {
682
0
    uint64_t segment_nums = 0;
683
0
    for_each_tablet([&](const TabletSharedPtr& tablet) { segment_nums += tablet->segment_count(); },
684
0
                    filter_all_tablets);
685
0
    return segment_nums;
686
0
}
687
688
bool TabletManager::get_tablet_id_and_schema_hash_from_path(const string& path,
689
                                                            TTabletId* tablet_id,
690
74
                                                            TSchemaHash* schema_hash) {
691
    // the path like: /data/14/10080/964828783/
692
74
    static re2::RE2 normal_re("/data/\\d+/(\\d+)/(\\d+)($|/)");
693
    // match tablet schema hash data path, for example, the path is /data/1/16791/29998
694
    // 1 is shard id , 16791 is tablet id, 29998 is schema hash
695
74
    if (RE2::PartialMatch(path, normal_re, tablet_id, schema_hash)) {
696
70
        return true;
697
70
    }
698
699
    // If we can't match normal path pattern, this may be a path which is a empty tablet
700
    // directory. Use this pattern to match empty tablet directory. In this case schema_hash
701
    // will be set to zero.
702
4
    static re2::RE2 empty_tablet_re("/data/\\d+/(\\d+)($|/$)");
703
4
    if (!RE2::PartialMatch(path, empty_tablet_re, tablet_id)) {
704
2
        return false;
705
2
    }
706
2
    *schema_hash = 0;
707
2
    return true;
708
4
}
709
710
3
bool TabletManager::get_rowset_id_from_path(const string& path, RowsetId* rowset_id) {
711
    // the path like: /data/14/10080/964828783/02000000000000969144d8725cb62765f9af6cd3125d5a91_0.dat
712
3
    static re2::RE2 re("/data/\\d+/\\d+/\\d+/([A-Fa-f0-9]+)_.*");
713
3
    string id_str;
714
3
    bool ret = RE2::PartialMatch(path, re, &id_str);
715
3
    if (ret) {
716
1
        rowset_id->init(id_str);
717
1
        return true;
718
1
    }
719
2
    return false;
720
3
}
721
722
0
void TabletManager::get_tablet_stat(TTabletStatResult* result) {
723
0
    std::shared_ptr<std::vector<TTabletStat>> local_cache;
724
0
    {
725
0
        std::lock_guard<std::mutex> guard(_tablet_stat_cache_mutex);
726
0
        local_cache = _tablet_stat_list_cache;
727
0
    }
728
0
    result->__set_tablet_stat_list(*local_cache);
729
0
}
730
731
struct TabletScore {
732
    TabletSharedPtr tablet_ptr;
733
    int score;
734
};
735
736
std::vector<TabletSharedPtr> TabletManager::find_best_tablets_to_compaction(
737
        CompactionType compaction_type, DataDir* data_dir,
738
        const std::unordered_set<TabletSharedPtr>& tablet_submitted_compaction, uint32_t* score,
739
        const std::unordered_map<std::string_view, std::shared_ptr<CumulativeCompactionPolicy>>&
740
43
                all_cumulative_compaction_policies) {
741
43
    int64_t now_ms = UnixMillis();
742
43
    const string& compaction_type_str =
743
43
            compaction_type == CompactionType::BASE_COMPACTION ? "base" : "cumulative";
744
43
    uint32_t highest_score = 0;
745
43
    uint32_t compaction_score = 0;
746
43
    TabletSharedPtr best_tablet;
747
92
    auto cmp = [](TabletScore left, TabletScore right) { return left.score > right.score; };
748
43
    std::priority_queue<TabletScore, std::vector<TabletScore>, decltype(cmp)> top_tablets(cmp);
749
750
256
    auto handler = [&](const TabletSharedPtr& tablet_ptr) {
751
256
        if (tablet_ptr->tablet_meta()->tablet_schema()->disable_auto_compaction()) {
752
0
            LOG_EVERY_N(INFO, 500) << "Tablet " << tablet_ptr->tablet_id()
753
0
                                   << " will be ignored by automatic compaction tasks since it's "
754
0
                                   << "set to disabled automatic compaction.";
755
0
            return;
756
0
        }
757
758
256
        if (config::enable_skip_tablet_compaction &&
759
256
            tablet_ptr->should_skip_compaction(compaction_type, UnixSeconds())) {
760
2
            return;
761
2
        }
762
254
        if (!tablet_ptr->can_do_compaction(data_dir->path_hash(), compaction_type)) {
763
0
            return;
764
0
        }
765
766
254
        auto search = tablet_submitted_compaction.find(tablet_ptr);
767
254
        if (search != tablet_submitted_compaction.end()) {
768
0
            return;
769
0
        }
770
771
254
        int64_t last_failure_ms = tablet_ptr->last_cumu_compaction_failure_time();
772
254
        if (compaction_type == CompactionType::BASE_COMPACTION) {
773
0
            last_failure_ms = tablet_ptr->last_base_compaction_failure_time();
774
0
        }
775
254
        if (now_ms - last_failure_ms <= 5000) {
776
0
            VLOG_DEBUG << "Too often to check compaction, skip it. "
777
0
                       << "compaction_type=" << compaction_type_str
778
0
                       << ", last_failure_time_ms=" << last_failure_ms
779
0
                       << ", tablet_id=" << tablet_ptr->tablet_id();
780
0
            return;
781
0
        }
782
783
254
        if (compaction_type == CompactionType::BASE_COMPACTION) {
784
0
            std::unique_lock<std::mutex> lock(tablet_ptr->get_base_compaction_lock(),
785
0
                                              std::try_to_lock);
786
0
            if (!lock.owns_lock()) {
787
0
                LOG(INFO) << "can not get base lock: " << tablet_ptr->tablet_id();
788
0
                return;
789
0
            }
790
254
        } else {
791
254
            std::unique_lock<std::mutex> lock(tablet_ptr->get_cumulative_compaction_lock(),
792
254
                                              std::try_to_lock);
793
254
            if (!lock.owns_lock()) {
794
0
                LOG(INFO) << "can not get cumu lock: " << tablet_ptr->tablet_id();
795
0
                return;
796
0
            }
797
254
        }
798
254
        auto cumulative_compaction_policy = all_cumulative_compaction_policies.at(
799
254
                tablet_ptr->tablet_meta()->compaction_policy());
800
254
        uint32_t current_compaction_score =
801
254
                tablet_ptr->calc_compaction_score(compaction_type, cumulative_compaction_policy);
802
254
        if (current_compaction_score < 5) {
803
11
            tablet_ptr->set_skip_compaction(true, compaction_type, UnixSeconds());
804
11
        }
805
806
254
        if (config::compaction_num_per_round > 1) {
807
205
            TabletScore ts;
808
205
            ts.score = current_compaction_score;
809
205
            ts.tablet_ptr = tablet_ptr;
810
205
            if ((top_tablets.size() >= config::compaction_num_per_round &&
811
205
                 current_compaction_score > top_tablets.top().score) ||
812
205
                top_tablets.size() < config::compaction_num_per_round) {
813
25
                top_tablets.push(ts);
814
25
                if (top_tablets.size() > config::compaction_num_per_round) {
815
0
                    top_tablets.pop();
816
0
                }
817
25
                if (current_compaction_score > highest_score) {
818
3
                    highest_score = current_compaction_score;
819
3
                    compaction_score = current_compaction_score;
820
3
                }
821
25
            }
822
205
        } else {
823
49
            if (current_compaction_score > highest_score) {
824
3
                highest_score = current_compaction_score;
825
3
                compaction_score = current_compaction_score;
826
3
                best_tablet = tablet_ptr;
827
3
            }
828
49
        }
829
254
    };
830
831
43
    for_each_tablet(handler, filter_all_tablets);
832
43
    std::vector<TabletSharedPtr> picked_tablet;
833
43
    if (best_tablet != nullptr) {
834
3
        VLOG_CRITICAL << "Found the best tablet for compaction. "
835
0
                      << "compaction_type=" << compaction_type_str
836
0
                      << ", tablet_id=" << best_tablet->tablet_id() << ", path=" << data_dir->path()
837
0
                      << ", compaction_score=" << compaction_score
838
0
                      << ", highest_score=" << highest_score;
839
3
        picked_tablet.emplace_back(std::move(best_tablet));
840
3
    }
841
842
43
    std::vector<TabletSharedPtr> reverse_top_tablets;
843
68
    while (!top_tablets.empty()) {
844
25
        reverse_top_tablets.emplace_back(top_tablets.top().tablet_ptr);
845
25
        top_tablets.pop();
846
25
    }
847
848
68
    for (auto it = reverse_top_tablets.rbegin(); it != reverse_top_tablets.rend(); ++it) {
849
25
        picked_tablet.emplace_back(*it);
850
25
    }
851
852
43
    *score = highest_score;
853
43
    return picked_tablet;
854
43
}
855
856
Status TabletManager::load_tablet_from_meta(DataDir* data_dir, TTabletId tablet_id,
857
                                            TSchemaHash schema_hash, const string& meta_binary,
858
                                            bool update_meta, bool force, bool restore,
859
2
                                            bool check_path) {
860
2
    TabletMetaSharedPtr tablet_meta(new TabletMeta());
861
2
    Status status = tablet_meta->deserialize(meta_binary);
862
2
    if (!status.ok()) {
863
0
        return Status::Error<HEADER_PB_PARSE_FAILED>(
864
0
                "fail to load tablet because can not parse meta_binary string. tablet_id={}, "
865
0
                "schema_hash={}, path={}, status={}",
866
0
                tablet_id, schema_hash, data_dir->path(), status);
867
0
    }
868
2
    tablet_meta->init_rs_metas_fs(data_dir->fs());
869
870
    // check if tablet meta is valid
871
2
    if (tablet_meta->tablet_id() != tablet_id || tablet_meta->schema_hash() != schema_hash) {
872
0
        return Status::Error<HEADER_PB_PARSE_FAILED>(
873
0
                "fail to load tablet because meet invalid tablet meta. trying to load "
874
0
                "tablet(tablet_id={}, schema_hash={}), but meet tablet={}, path={}",
875
0
                tablet_id, schema_hash, tablet_meta->tablet_id(), data_dir->path());
876
0
    }
877
2
    if (tablet_meta->tablet_uid().hi == 0 && tablet_meta->tablet_uid().lo == 0) {
878
0
        return Status::Error<HEADER_PB_PARSE_FAILED>(
879
0
                "fail to load tablet because its uid == 0. tablet={}, path={}",
880
0
                tablet_meta->tablet_id(), data_dir->path());
881
0
    }
882
883
2
    if (restore) {
884
        // we're restoring tablet from trash, tablet state should be changed from shutdown back to running
885
0
        tablet_meta->set_tablet_state(TABLET_RUNNING);
886
0
    }
887
888
2
    if (tablet_meta->partition_id() == 0) {
889
0
        LOG(WARNING) << "tablet=" << tablet_id << " load from meta but partition id eq 0";
890
0
    }
891
892
2
    TabletSharedPtr tablet = std::make_shared<Tablet>(_engine, std::move(tablet_meta), data_dir);
893
894
    // NOTE: method load_tablet_from_meta could be called by two cases as below
895
    // case 1: BE start;
896
    // case 2: Clone Task/Restore
897
    // For case 1 doesn't need path check because BE is just starting and not ready,
898
    // just check tablet meta status to judge whether tablet is delete is enough.
899
    // For case 2, If a tablet has just been copied to local BE,
900
    // it may be cleared by gc-thread(see perform_tablet_gc) because the tablet meta may not be loaded to memory.
901
    // So clone task should check path and then failed and retry in this case.
902
2
    if (check_path) {
903
2
        bool exists = true;
904
2
        RETURN_IF_ERROR(io::global_local_filesystem()->exists(tablet->tablet_path(), &exists));
905
2
        if (!exists) {
906
0
            return Status::Error<TABLE_ALREADY_DELETED_ERROR>(
907
0
                    "tablet path not exists, create tablet failed, path={}", tablet->tablet_path());
908
0
        }
909
2
    }
910
911
2
    if (tablet->tablet_meta()->tablet_state() == TABLET_SHUTDOWN) {
912
0
        {
913
0
            std::lock_guard<std::shared_mutex> shutdown_tablets_wrlock(_shutdown_tablets_lock);
914
0
            _shutdown_tablets.push_back(tablet);
915
0
        }
916
0
        return Status::Error<TABLE_ALREADY_DELETED_ERROR>(
917
0
                "fail to load tablet because it is to be deleted. tablet_id={}, schema_hash={}, "
918
0
                "path={}",
919
0
                tablet_id, schema_hash, data_dir->path());
920
0
    }
921
    // NOTE: We do not check tablet's initial version here, because if BE restarts when
922
    // one tablet is doing schema-change, we may meet empty tablet.
923
2
    if (tablet->max_version().first == -1 && tablet->tablet_state() == TABLET_RUNNING) {
924
        // tablet state is invalid, drop tablet
925
0
        return Status::Error<TABLE_INDEX_VALIDATE_ERROR>(
926
0
                "fail to load tablet. it is in running state but without delta. tablet={}, path={}",
927
0
                tablet->tablet_id(), data_dir->path());
928
0
    }
929
930
2
    RETURN_NOT_OK_STATUS_WITH_WARN(
931
2
            tablet->init(),
932
2
            strings::Substitute("tablet init failed. tablet=$0", tablet->tablet_id()));
933
934
2
    RuntimeProfile profile("CreateTablet");
935
2
    std::lock_guard<std::shared_mutex> wrlock(_get_tablets_shard_lock(tablet_id));
936
2
    RETURN_NOT_OK_STATUS_WITH_WARN(
937
2
            _add_tablet_unlocked(tablet_id, tablet, update_meta, force, &profile),
938
2
            strings::Substitute("fail to add tablet. tablet=$0", tablet->tablet_id()));
939
940
2
    return Status::OK();
941
2
}
942
943
Status TabletManager::load_tablet_from_dir(DataDir* store, TTabletId tablet_id,
944
                                           SchemaHash schema_hash, const string& schema_hash_path,
945
2
                                           bool force, bool restore) {
946
2
    LOG(INFO) << "begin to load tablet from dir. "
947
2
              << " tablet_id=" << tablet_id << " schema_hash=" << schema_hash
948
2
              << " path = " << schema_hash_path << " force = " << force << " restore = " << restore;
949
    // not add lock here, because load_tablet_from_meta already add lock
950
2
    std::string header_path = TabletMeta::construct_header_file_path(schema_hash_path, tablet_id);
951
    // should change shard id before load tablet
952
2
    std::string shard_path =
953
2
            path_util::dir_name(path_util::dir_name(path_util::dir_name(header_path)));
954
2
    std::string shard_str = shard_path.substr(shard_path.find_last_of('/') + 1);
955
2
    int32_t shard = stol(shard_str);
956
957
2
    bool exists = false;
958
2
    RETURN_IF_ERROR(io::global_local_filesystem()->exists(header_path, &exists));
959
2
    if (!exists) {
960
0
        return Status::Error<FILE_NOT_EXIST>("fail to find header file. [header_path={}]",
961
0
                                             header_path);
962
0
    }
963
964
2
    TabletMetaSharedPtr tablet_meta(new TabletMeta());
965
2
    if (!tablet_meta->create_from_file(header_path).ok()) {
966
0
        return Status::Error<ENGINE_LOAD_INDEX_TABLE_ERROR>(
967
0
                "fail to load tablet_meta. file_path={}", header_path);
968
0
    }
969
2
    TabletUid tablet_uid = TabletUid::gen_uid();
970
971
    // remove rowset binlog metas
972
2
    auto binlog_metas_file = fmt::format("{}/rowset_binlog_metas.pb", schema_hash_path);
973
2
    bool binlog_metas_file_exists = false;
974
2
    auto file_exists_status =
975
2
            io::global_local_filesystem()->exists(binlog_metas_file, &binlog_metas_file_exists);
976
2
    if (!file_exists_status.ok()) {
977
0
        return file_exists_status;
978
0
    }
979
2
    bool contain_binlog = false;
980
2
    RowsetBinlogMetasPB rowset_binlog_metas_pb;
981
2
    if (binlog_metas_file_exists) {
982
0
        auto binlog_meta_filesize = std::filesystem::file_size(binlog_metas_file);
983
0
        if (binlog_meta_filesize > 0) {
984
0
            contain_binlog = true;
985
0
            RETURN_IF_ERROR(read_pb(binlog_metas_file, &rowset_binlog_metas_pb));
986
0
            VLOG_DEBUG << "load rowset binlog metas from file. file_path=" << binlog_metas_file;
987
0
        }
988
0
        RETURN_IF_ERROR(io::global_local_filesystem()->delete_file(binlog_metas_file));
989
0
    }
990
2
    if (contain_binlog) {
991
0
        auto binlog_dir = fmt::format("{}/_binlog", schema_hash_path);
992
0
        RETURN_IF_ERROR(io::global_local_filesystem()->create_directory(binlog_dir));
993
994
0
        std::vector<io::FileInfo> files;
995
0
        RETURN_IF_ERROR(
996
0
                io::global_local_filesystem()->list(schema_hash_path, true, &files, &exists));
997
0
        for (auto& file : files) {
998
0
            auto& filename = file.file_name;
999
0
            std::string new_suffix;
1000
0
            std::string old_suffix;
1001
1002
0
            if (filename.ends_with(".binlog")) {
1003
0
                old_suffix = ".binlog";
1004
0
                new_suffix = ".dat";
1005
0
            } else if (filename.ends_with(".binlog-index")) {
1006
0
                old_suffix = ".binlog-index";
1007
0
                new_suffix = ".idx";
1008
0
            } else {
1009
0
                continue;
1010
0
            }
1011
1012
0
            std::string new_filename = filename;
1013
0
            new_filename.replace(filename.size() - old_suffix.size(), old_suffix.size(),
1014
0
                                 new_suffix);
1015
0
            auto from = fmt::format("{}/{}", schema_hash_path, filename);
1016
0
            auto to = fmt::format("{}/_binlog/{}", schema_hash_path, new_filename);
1017
0
            RETURN_IF_ERROR(io::global_local_filesystem()->rename(from, to));
1018
0
        }
1019
1020
0
        auto* meta = store->get_meta();
1021
        // if ingest binlog metas error, it will be gc in gc_unused_binlog_metas
1022
0
        RETURN_IF_ERROR(
1023
0
                RowsetMetaManager::ingest_binlog_metas(meta, tablet_uid, &rowset_binlog_metas_pb));
1024
0
    }
1025
1026
    // has to change shard id here, because meta file maybe copied from other source
1027
    // its shard is different from local shard
1028
2
    tablet_meta->set_shard_id(shard);
1029
    // load dir is called by clone, restore, storage migration
1030
    // should change tablet uid when tablet object changed
1031
2
    tablet_meta->set_tablet_uid(std::move(tablet_uid));
1032
2
    std::string meta_binary;
1033
2
    tablet_meta->serialize(&meta_binary);
1034
2
    RETURN_NOT_OK_STATUS_WITH_WARN(
1035
2
            load_tablet_from_meta(store, tablet_id, schema_hash, meta_binary, true, force, restore,
1036
2
                                  true),
1037
2
            strings::Substitute("fail to load tablet. header_path=$0", header_path));
1038
1039
2
    return Status::OK();
1040
2
}
1041
1042
0
Status TabletManager::report_tablet_info(TTabletInfo* tablet_info) {
1043
0
    LOG(INFO) << "begin to process report tablet info."
1044
0
              << "tablet_id=" << tablet_info->tablet_id;
1045
1046
0
    Status res = Status::OK();
1047
1048
0
    TabletSharedPtr tablet = get_tablet(tablet_info->tablet_id);
1049
0
    if (tablet == nullptr) {
1050
0
        return Status::Error<TABLE_NOT_FOUND>("can't find tablet={}", tablet_info->tablet_id);
1051
0
    }
1052
1053
0
    tablet->build_tablet_report_info(tablet_info);
1054
0
    VLOG_TRACE << "success to process report tablet info.";
1055
0
    return res;
1056
0
}
1057
1058
0
void TabletManager::build_all_report_tablets_info(std::map<TTabletId, TTablet>* tablets_info) {
1059
0
    DCHECK(tablets_info != nullptr);
1060
0
    VLOG_NOTICE << "begin to build all report tablets info";
1061
1062
    // build the expired txn map first, outside the tablet map lock
1063
0
    std::map<TabletInfo, std::vector<int64_t>> expire_txn_map;
1064
0
    _engine.txn_manager()->build_expire_txn_map(&expire_txn_map);
1065
0
    LOG(INFO) << "find expired transactions for " << expire_txn_map.size() << " tablets";
1066
1067
0
    HistogramStat tablet_version_num_hist;
1068
0
    auto local_cache = std::make_shared<std::vector<TTabletStat>>();
1069
0
    auto handler = [&](const TabletSharedPtr& tablet) {
1070
0
        auto& t_tablet = (*tablets_info)[tablet->tablet_id()];
1071
0
        TTabletInfo& tablet_info = t_tablet.tablet_infos.emplace_back();
1072
0
        tablet->build_tablet_report_info(&tablet_info, true, true);
1073
        // find expired transaction corresponding to this tablet
1074
0
        TabletInfo tinfo(tablet->tablet_id(), tablet->tablet_uid());
1075
0
        auto find = expire_txn_map.find(tinfo);
1076
0
        if (find != expire_txn_map.end()) {
1077
0
            tablet_info.__set_transaction_ids(find->second);
1078
0
            expire_txn_map.erase(find);
1079
0
        }
1080
0
        tablet_version_num_hist.add(tablet_info.total_version_count);
1081
0
        auto& t_tablet_stat = local_cache->emplace_back();
1082
0
        t_tablet_stat.__set_tablet_id(tablet_info.tablet_id);
1083
0
        t_tablet_stat.__set_data_size(tablet_info.data_size);
1084
0
        t_tablet_stat.__set_remote_data_size(tablet_info.remote_data_size);
1085
0
        t_tablet_stat.__set_row_count(tablet_info.row_count);
1086
0
        t_tablet_stat.__set_total_version_count(tablet_info.total_version_count);
1087
0
        t_tablet_stat.__set_visible_version_count(tablet_info.visible_version_count);
1088
0
        t_tablet_stat.__set_visible_version(tablet_info.version);
1089
0
    };
1090
0
    for_each_tablet(handler, filter_all_tablets);
1091
1092
0
    {
1093
0
        std::lock_guard<std::mutex> guard(_tablet_stat_cache_mutex);
1094
0
        _tablet_stat_list_cache.swap(local_cache);
1095
0
    }
1096
0
    DorisMetrics::instance()->tablet_version_num_distribution->set_histogram(
1097
0
            tablet_version_num_hist);
1098
0
    LOG(INFO) << "success to build all report tablets info. tablet_count=" << tablets_info->size();
1099
0
}
1100
1101
22
Status TabletManager::start_trash_sweep() {
1102
22
    DBUG_EXECUTE_IF("TabletManager.start_trash_sweep.sleep", DBUG_BLOCK);
1103
22
    std::unique_lock<std::mutex> lock(_gc_tablets_lock, std::defer_lock);
1104
22
    if (!lock.try_lock()) {
1105
0
        return Status::OK();
1106
0
    }
1107
1108
22
    for_each_tablet([](const TabletSharedPtr& tablet) { tablet->delete_expired_stale_rowset(); },
1109
22
                    filter_all_tablets);
1110
1111
22
    std::list<TabletSharedPtr>::iterator last_it;
1112
22
    {
1113
22
        std::shared_lock rdlock(_shutdown_tablets_lock);
1114
22
        last_it = _shutdown_tablets.begin();
1115
22
        if (last_it == _shutdown_tablets.end()) {
1116
17
            return Status::OK();
1117
17
        }
1118
22
    }
1119
1120
10
    auto get_batch_tablets = [this, &last_it](int limit) {
1121
10
        std::vector<TabletSharedPtr> batch_tablets;
1122
10
        std::lock_guard<std::shared_mutex> wrdlock(_shutdown_tablets_lock);
1123
240
        while (last_it != _shutdown_tablets.end() && batch_tablets.size() < limit) {
1124
            // it means current tablet is referenced by other thread
1125
230
            if (last_it->use_count() > 1) {
1126
6
                last_it++;
1127
224
            } else {
1128
224
                batch_tablets.push_back(*last_it);
1129
224
                last_it = _shutdown_tablets.erase(last_it);
1130
224
            }
1131
230
        }
1132
1133
10
        return batch_tablets;
1134
10
    };
1135
1136
5
    std::list<TabletSharedPtr> failed_tablets;
1137
    // return true if need continue delete
1138
6
    auto delete_one_batch = [this, get_batch_tablets, &failed_tablets]() -> bool {
1139
6
        int limit = 200;
1140
10
        for (;;) {
1141
10
            auto batch_tablets = get_batch_tablets(limit);
1142
224
            for (const auto& tablet : batch_tablets) {
1143
224
                if (_move_tablet_to_trash(tablet)) {
1144
224
                    limit--;
1145
224
                } else {
1146
0
                    failed_tablets.push_back(tablet);
1147
0
                }
1148
224
            }
1149
10
            if (limit <= 0) {
1150
1
                return true;
1151
1
            }
1152
9
            if (batch_tablets.empty()) {
1153
5
                return false;
1154
5
            }
1155
9
        }
1156
1157
0
        return false;
1158
6
    };
1159
1160
6
    while (delete_one_batch()) {
1161
#ifndef BE_TEST
1162
        sleep(1);
1163
#endif
1164
1
    }
1165
1166
5
    if (!failed_tablets.empty()) {
1167
0
        std::lock_guard<std::shared_mutex> wrlock(_shutdown_tablets_lock);
1168
0
        _shutdown_tablets.splice(_shutdown_tablets.end(), failed_tablets);
1169
0
    }
1170
1171
5
    return Status::OK();
1172
22
}
1173
1174
224
bool TabletManager::_move_tablet_to_trash(const TabletSharedPtr& tablet) {
1175
224
    RETURN_IF_ERROR(register_transition_tablet(tablet->tablet_id(), "move to trash"));
1176
224
    Defer defer {[&]() { unregister_transition_tablet(tablet->tablet_id(), "move to trash"); }};
1177
1178
224
    TabletSharedPtr tablet_in_not_shutdown = get_tablet(tablet->tablet_id());
1179
224
    if (tablet_in_not_shutdown) {
1180
0
        TSchemaHash schema_hash_not_shutdown = tablet_in_not_shutdown->schema_hash();
1181
0
        size_t path_hash_not_shutdown = tablet_in_not_shutdown->data_dir()->path_hash();
1182
0
        if (tablet->schema_hash() == schema_hash_not_shutdown &&
1183
0
            tablet->data_dir()->path_hash() == path_hash_not_shutdown) {
1184
0
            tablet->clear_cache();
1185
            // shard_id in memory not eq shard_id in shutdown
1186
0
            if (tablet_in_not_shutdown->tablet_path() != tablet->tablet_path()) {
1187
0
                LOG(INFO) << "tablet path not eq shutdown tablet path, move it to trash, tablet_id="
1188
0
                          << tablet_in_not_shutdown->tablet_id()
1189
0
                          << ", mem manager tablet path=" << tablet_in_not_shutdown->tablet_path()
1190
0
                          << ", shutdown tablet path=" << tablet->tablet_path();
1191
0
                return tablet->data_dir()->move_to_trash(tablet->tablet_path());
1192
0
            } else {
1193
0
                LOG(INFO) << "tablet path eq shutdown tablet path, not move to trash, tablet_id="
1194
0
                          << tablet_in_not_shutdown->tablet_id()
1195
0
                          << ", mem manager tablet path=" << tablet_in_not_shutdown->tablet_path()
1196
0
                          << ", shutdown tablet path=" << tablet->tablet_path();
1197
0
                return true;
1198
0
            }
1199
0
        }
1200
0
    }
1201
1202
224
    TabletMetaSharedPtr tablet_meta(new TabletMeta());
1203
224
    int64_t get_meta_ts = MonotonicMicros();
1204
224
    Status check_st = TabletMetaManager::get_meta(tablet->data_dir(), tablet->tablet_id(),
1205
224
                                                  tablet->schema_hash(), tablet_meta);
1206
224
    if (check_st.ok()) {
1207
224
        if (tablet_meta->tablet_state() != TABLET_SHUTDOWN ||
1208
224
            tablet_meta->tablet_uid() != tablet->tablet_uid()) {
1209
0
            LOG(WARNING) << "tablet's state changed to normal, skip remove dirs"
1210
0
                         << " tablet id = " << tablet_meta->tablet_id()
1211
0
                         << " schema hash = " << tablet_meta->schema_hash()
1212
0
                         << " old tablet_uid=" << tablet->tablet_uid()
1213
0
                         << " cur tablet_uid=" << tablet_meta->tablet_uid();
1214
0
            return true;
1215
0
        }
1216
1217
224
        tablet->clear_cache();
1218
1219
        // move data to trash
1220
224
        const auto& tablet_path = tablet->tablet_path();
1221
224
        bool exists = false;
1222
224
        Status exists_st = io::global_local_filesystem()->exists(tablet_path, &exists);
1223
224
        if (!exists_st) {
1224
0
            return false;
1225
0
        }
1226
224
        if (exists) {
1227
            // take snapshot of tablet meta
1228
224
            auto meta_file_path = fmt::format("{}/{}.hdr", tablet_path, tablet->tablet_id());
1229
224
            int64_t save_meta_ts = MonotonicMicros();
1230
224
            auto save_st = tablet->tablet_meta()->save(meta_file_path);
1231
224
            if (!save_st.ok()) {
1232
0
                LOG(WARNING) << "failed to save meta, tablet_id=" << tablet_meta->tablet_id()
1233
0
                             << ", tablet_uid=" << tablet_meta->tablet_uid()
1234
0
                             << ", error=" << save_st;
1235
0
                return false;
1236
0
            }
1237
224
            int64_t now = MonotonicMicros();
1238
224
            LOG(INFO) << "start to move tablet to trash. " << tablet_path
1239
224
                      << ". rocksdb get meta cost " << (save_meta_ts - get_meta_ts)
1240
224
                      << " us, rocksdb save meta cost " << (now - save_meta_ts) << " us";
1241
224
            Status rm_st = tablet->data_dir()->move_to_trash(tablet_path);
1242
224
            if (!rm_st.ok()) {
1243
0
                LOG(WARNING) << "fail to move dir to trash. " << tablet_path;
1244
0
                return false;
1245
0
            }
1246
224
        }
1247
        // remove tablet meta
1248
224
        auto remove_st = TabletMetaManager::remove(tablet->data_dir(), tablet->tablet_id(),
1249
224
                                                   tablet->schema_hash());
1250
224
        if (!remove_st.ok()) {
1251
0
            LOG(WARNING) << "failed to remove meta, tablet_id=" << tablet_meta->tablet_id()
1252
0
                         << ", tablet_uid=" << tablet_meta->tablet_uid() << ", error=" << remove_st;
1253
0
            return false;
1254
0
        }
1255
224
        LOG(INFO) << "successfully move tablet to trash. "
1256
224
                  << "tablet_id=" << tablet->tablet_id()
1257
224
                  << ", schema_hash=" << tablet->schema_hash() << ", tablet_path=" << tablet_path;
1258
224
        return true;
1259
224
    } else {
1260
0
        tablet->clear_cache();
1261
        // if could not find tablet info in meta store, then check if dir existed
1262
0
        const auto& tablet_path = tablet->tablet_path();
1263
0
        bool exists = false;
1264
0
        Status exists_st = io::global_local_filesystem()->exists(tablet_path, &exists);
1265
0
        if (!exists_st) {
1266
0
            return false;
1267
0
        }
1268
0
        if (exists) {
1269
0
            if (check_st.is<META_KEY_NOT_FOUND>()) {
1270
0
                LOG(INFO) << "could not find tablet meta in rocksdb, so just delete it path "
1271
0
                          << "tablet_id=" << tablet->tablet_id()
1272
0
                          << ", schema_hash=" << tablet->schema_hash()
1273
0
                          << ", delete tablet_path=" << tablet_path;
1274
0
                RETURN_IF_ERROR(io::global_local_filesystem()->delete_directory(tablet_path));
1275
0
                RETURN_IF_ERROR(DataDir::delete_tablet_parent_path_if_empty(tablet_path));
1276
0
                return true;
1277
0
            }
1278
0
            LOG(WARNING) << "errors while load meta from store, skip this tablet. "
1279
0
                         << "tablet_id=" << tablet->tablet_id()
1280
0
                         << ", schema_hash=" << tablet->schema_hash();
1281
0
            return false;
1282
0
        } else {
1283
0
            LOG(INFO) << "could not find tablet dir, skip it and remove it from gc-queue. "
1284
0
                      << "tablet_id=" << tablet->tablet_id()
1285
0
                      << ", schema_hash=" << tablet->schema_hash()
1286
0
                      << ", tablet_path=" << tablet_path;
1287
0
            return true;
1288
0
        }
1289
0
    }
1290
224
}
1291
1292
489
Status TabletManager::register_transition_tablet(int64_t tablet_id, std::string reason) {
1293
489
    tablets_shard& shard = _get_tablets_shard(tablet_id);
1294
489
    std::thread::id thread_id = std::this_thread::get_id();
1295
489
    std::lock_guard<std::mutex> lk(shard.lock_for_transition);
1296
489
    if (auto search = shard.tablets_under_transition.find(tablet_id);
1297
489
        search == shard.tablets_under_transition.end()) {
1298
        // not found
1299
487
        shard.tablets_under_transition[tablet_id] = std::make_tuple(reason, thread_id, 1);
1300
487
        LOG(INFO) << "add tablet_id= " << tablet_id << " to map, reason=" << reason
1301
487
                  << ", lock times=1, thread_id_in_map=" << thread_id;
1302
487
        return Status::OK();
1303
487
    } else {
1304
        // found
1305
2
        auto& [r, thread_id_in_map, lock_times] = search->second;
1306
2
        if (thread_id != thread_id_in_map) {
1307
            // other thread, failed
1308
0
            LOG(INFO) << "tablet_id = " << tablet_id << " is doing " << r
1309
0
                      << ", thread_id_in_map=" << thread_id_in_map << " , add reason=" << reason
1310
0
                      << ", thread_id=" << thread_id;
1311
0
            return Status::InternalError<false>("{} failed try later, tablet_id={}", reason,
1312
0
                                                tablet_id);
1313
0
        }
1314
        // add lock times
1315
2
        ++lock_times;
1316
2
        LOG(INFO) << "add tablet_id= " << tablet_id << " to map, reason=" << reason
1317
2
                  << ", lock times=" << lock_times << ", thread_id_in_map=" << thread_id_in_map;
1318
2
        return Status::OK();
1319
2
    }
1320
489
}
1321
1322
489
void TabletManager::unregister_transition_tablet(int64_t tablet_id, std::string reason) {
1323
489
    tablets_shard& shard = _get_tablets_shard(tablet_id);
1324
489
    std::thread::id thread_id = std::this_thread::get_id();
1325
489
    std::lock_guard<std::mutex> lk(shard.lock_for_transition);
1326
489
    if (auto search = shard.tablets_under_transition.find(tablet_id);
1327
489
        search == shard.tablets_under_transition.end()) {
1328
        // impossible, bug
1329
0
        DCHECK(false) << "tablet " << tablet_id
1330
0
                      << " must be found, before unreg must have been reg";
1331
489
    } else {
1332
489
        auto& [r, thread_id_in_map, lock_times] = search->second;
1333
489
        if (thread_id_in_map != thread_id) {
1334
            // impossible, bug
1335
0
            DCHECK(false) << "tablet " << tablet_id << " unreg thread must same reg thread";
1336
0
        }
1337
        // sub lock times
1338
489
        --lock_times;
1339
489
        if (lock_times != 0) {
1340
2
            LOG(INFO) << "erase tablet_id= " << tablet_id << " from map, reason=" << reason
1341
2
                      << ", left=" << lock_times << ", thread_id_in_map=" << thread_id_in_map;
1342
487
        } else {
1343
487
            LOG(INFO) << "erase tablet_id= " << tablet_id << " from map, reason=" << reason
1344
487
                      << ", thread_id_in_map=" << thread_id_in_map;
1345
487
            shard.tablets_under_transition.erase(tablet_id);
1346
487
        }
1347
489
    }
1348
489
}
1349
1350
void TabletManager::try_delete_unused_tablet_path(DataDir* data_dir, TTabletId tablet_id,
1351
                                                  SchemaHash schema_hash,
1352
                                                  const string& schema_hash_path,
1353
11
                                                  int16_t shard_id) {
1354
    // acquire the read lock, so that there is no creating tablet or load tablet from meta tasks
1355
    // create tablet and load tablet task should check whether the dir exists
1356
11
    tablets_shard& shard = _get_tablets_shard(tablet_id);
1357
11
    std::shared_lock rdlock(shard.lock);
1358
1359
    // check if meta already exists
1360
11
    TabletMetaSharedPtr tablet_meta(new TabletMeta());
1361
11
    Status check_st = TabletMetaManager::get_meta(data_dir, tablet_id, schema_hash, tablet_meta);
1362
11
    if (check_st.ok() && tablet_meta->shard_id() == shard_id) {
1363
1
        return;
1364
1
    }
1365
1366
10
    LOG(INFO) << "tablet meta not exists, try delete tablet path " << schema_hash_path;
1367
1368
10
    bool succ = register_transition_tablet(tablet_id, "path gc");
1369
10
    if (!succ) {
1370
0
        return;
1371
0
    }
1372
10
    Defer defer {[&]() { unregister_transition_tablet(tablet_id, "path gc"); }};
1373
1374
10
    TabletSharedPtr tablet = _get_tablet_unlocked(tablet_id);
1375
10
    if (tablet != nullptr && tablet->tablet_path() == schema_hash_path) {
1376
0
        LOG(INFO) << "tablet , skip delete the path " << schema_hash_path;
1377
0
        return;
1378
0
    }
1379
1380
    // TODO(ygl): may do other checks in the future
1381
10
    bool exists = false;
1382
10
    Status exists_st = io::global_local_filesystem()->exists(schema_hash_path, &exists);
1383
10
    if (exists_st && exists) {
1384
10
        LOG(INFO) << "start to move tablet to trash. tablet_path = " << schema_hash_path;
1385
10
        Status rm_st = data_dir->move_to_trash(schema_hash_path);
1386
10
        if (!rm_st.ok()) {
1387
0
            LOG(WARNING) << "fail to move dir to trash. dir=" << schema_hash_path;
1388
10
        } else {
1389
10
            LOG(INFO) << "move path " << schema_hash_path << " to trash successfully";
1390
10
        }
1391
10
    }
1392
10
}
1393
1394
void TabletManager::update_root_path_info(std::map<string, DataDirInfo>* path_map,
1395
17
                                          size_t* tablet_count) {
1396
17
    DCHECK(tablet_count);
1397
17
    *tablet_count = 0;
1398
17
    auto filter = [path_map, tablet_count](Tablet* t) -> bool {
1399
0
        ++(*tablet_count);
1400
0
        auto iter = path_map->find(t->data_dir()->path());
1401
0
        return iter != path_map->end() && iter->second.is_used;
1402
0
    };
1403
1404
17
    auto handler = [&](const TabletSharedPtr& tablet) {
1405
0
        auto& data_dir_info = (*path_map)[tablet->data_dir()->path()];
1406
0
        data_dir_info.local_used_capacity += tablet->tablet_local_size();
1407
0
        data_dir_info.remote_used_capacity += tablet->tablet_remote_size();
1408
0
    };
1409
1410
17
    for_each_tablet(handler, filter);
1411
17
}
1412
1413
void TabletManager::get_partition_related_tablets(int64_t partition_id,
1414
0
                                                  std::set<TabletInfo>* tablet_infos) {
1415
0
    std::shared_lock rdlock(_partitions_lock);
1416
0
    auto it = _partitions.find(partition_id);
1417
0
    if (it != _partitions.end()) {
1418
0
        *tablet_infos = it->second.tablets;
1419
0
    }
1420
0
}
1421
1422
0
void TabletManager::get_partitions_visible_version(std::map<int64_t, int64_t>* partitions_version) {
1423
0
    std::shared_lock rdlock(_partitions_lock);
1424
0
    for (const auto& [partition_id, partition] : _partitions) {
1425
0
        partitions_version->insert(
1426
0
                {partition_id, partition.visible_version->version.load(std::memory_order_relaxed)});
1427
0
    }
1428
0
}
1429
1430
void TabletManager::update_partitions_visible_version(
1431
0
        const std::map<int64_t, int64_t>& partitions_version) {
1432
0
    std::shared_lock rdlock(_partitions_lock);
1433
0
    for (auto [partition_id, version] : partitions_version) {
1434
0
        auto it = _partitions.find(partition_id);
1435
0
        if (it != _partitions.end()) {
1436
0
            it->second.visible_version->update_version_monoto(version);
1437
0
        }
1438
0
    }
1439
0
}
1440
1441
18
void TabletManager::do_tablet_meta_checkpoint(DataDir* data_dir) {
1442
18
    auto filter = [data_dir](Tablet* tablet) -> bool {
1443
0
        return tablet->tablet_state() == TABLET_RUNNING &&
1444
0
               tablet->data_dir()->path_hash() == data_dir->path_hash() && tablet->is_used() &&
1445
0
               tablet->init_succeeded();
1446
0
    };
1447
1448
18
    std::vector<TabletSharedPtr> related_tablets = get_all_tablet(filter);
1449
18
    int counter = 0;
1450
18
    MonotonicStopWatch watch;
1451
18
    watch.start();
1452
18
    for (TabletSharedPtr tablet : related_tablets) {
1453
0
        if (tablet->do_tablet_meta_checkpoint()) {
1454
0
            ++counter;
1455
0
        }
1456
0
    }
1457
18
    int64_t cost = watch.elapsed_time() / 1000 / 1000;
1458
18
    LOG(INFO) << "finish to do meta checkpoint on dir: " << data_dir->path()
1459
18
              << ", number: " << counter << ", cost(ms): " << cost;
1460
18
}
1461
1462
Status TabletManager::_create_tablet_meta_unlocked(const TCreateTabletReq& request, DataDir* store,
1463
                                                   const bool is_schema_change,
1464
                                                   const Tablet* base_tablet,
1465
295
                                                   TabletMetaSharedPtr* tablet_meta) {
1466
295
    uint32_t next_unique_id = 0;
1467
295
    std::unordered_map<uint32_t, uint32_t> col_idx_to_unique_id;
1468
295
    if (!is_schema_change) {
1469
2.13k
        for (uint32_t col_idx = 0; col_idx < request.tablet_schema.columns.size(); ++col_idx) {
1470
1.84k
            col_idx_to_unique_id[col_idx] = col_idx;
1471
1.84k
        }
1472
295
        next_unique_id = request.tablet_schema.columns.size();
1473
295
    } else {
1474
0
        next_unique_id = base_tablet->next_unique_id();
1475
0
        auto& new_columns = request.tablet_schema.columns;
1476
0
        for (uint32_t new_col_idx = 0; new_col_idx < new_columns.size(); ++new_col_idx) {
1477
0
            const TColumn& column = new_columns[new_col_idx];
1478
            // For schema change, compare old_tablet and new_tablet:
1479
            // 1. if column exist in both new_tablet and old_tablet, choose the column's
1480
            //    unique_id in old_tablet to be the column's ordinal number in new_tablet
1481
            // 2. if column exists only in new_tablet, assign next_unique_id of old_tablet
1482
            //    to the new column
1483
0
            int32_t old_col_idx = base_tablet->tablet_schema()->field_index(column.column_name);
1484
0
            if (old_col_idx != -1) {
1485
0
                uint32_t old_unique_id =
1486
0
                        base_tablet->tablet_schema()->column(old_col_idx).unique_id();
1487
0
                col_idx_to_unique_id[new_col_idx] = old_unique_id;
1488
0
            } else {
1489
                // Not exist in old tablet, it is a new added column
1490
0
                col_idx_to_unique_id[new_col_idx] = next_unique_id++;
1491
0
            }
1492
0
        }
1493
0
    }
1494
295
    VLOG_NOTICE << "creating tablet meta. next_unique_id=" << next_unique_id;
1495
1496
    // We generate a new tablet_uid for this new tablet.
1497
295
    uint64_t shard_id = store->get_shard();
1498
295
    *tablet_meta = TabletMeta::create(request, TabletUid::gen_uid(), shard_id, next_unique_id,
1499
295
                                      col_idx_to_unique_id);
1500
295
    if (request.__isset.storage_format) {
1501
51
        if (request.storage_format == TStorageFormat::DEFAULT) {
1502
0
            (*tablet_meta)->set_preferred_rowset_type(_engine.default_rowset_type());
1503
51
        } else if (request.storage_format == TStorageFormat::V1) {
1504
0
            (*tablet_meta)->set_preferred_rowset_type(ALPHA_ROWSET);
1505
51
        } else if (request.storage_format == TStorageFormat::V2) {
1506
51
            (*tablet_meta)->set_preferred_rowset_type(BETA_ROWSET);
1507
51
        } else {
1508
0
            return Status::Error<CE_CMD_PARAMS_ERROR>("invalid TStorageFormat: {}",
1509
0
                                                      request.storage_format);
1510
0
        }
1511
51
    }
1512
295
    return Status::OK();
1513
295
}
1514
1515
3.56k
TabletSharedPtr TabletManager::_get_tablet_unlocked(TTabletId tablet_id) {
1516
3.56k
    VLOG_NOTICE << "begin to get tablet. tablet_id=" << tablet_id;
1517
3.56k
    tablet_map_t& tablet_map = _get_tablet_map(tablet_id);
1518
3.56k
    const auto& iter = tablet_map.find(tablet_id);
1519
3.56k
    if (iter != tablet_map.end()) {
1520
3.01k
        return iter->second;
1521
3.01k
    }
1522
548
    return nullptr;
1523
3.56k
}
1524
1525
297
void TabletManager::_add_tablet_to_partition(const TabletSharedPtr& tablet) {
1526
297
    std::lock_guard<std::shared_mutex> wrlock(_partitions_lock);
1527
297
    auto& partition = _partitions[tablet->partition_id()];
1528
297
    partition.tablets.insert(tablet->get_tablet_info());
1529
297
    tablet->set_visible_version(
1530
297
            std::static_pointer_cast<const VersionWithTime>(partition.visible_version));
1531
297
}
1532
1533
251
void TabletManager::_remove_tablet_from_partition(const TabletSharedPtr& tablet) {
1534
251
    tablet->set_visible_version(nullptr);
1535
251
    std::lock_guard<std::shared_mutex> wrlock(_partitions_lock);
1536
251
    auto it = _partitions.find(tablet->partition_id());
1537
251
    if (it == _partitions.end()) {
1538
0
        return;
1539
0
    }
1540
1541
251
    auto& tablets = it->second.tablets;
1542
251
    tablets.erase(tablet->get_tablet_info());
1543
251
    if (tablets.empty()) {
1544
29
        _partitions.erase(it);
1545
29
    }
1546
251
}
1547
1548
void TabletManager::obtain_specific_quantity_tablets(vector<TabletInfo>& tablets_info,
1549
0
                                                     int64_t num) {
1550
0
    for (const auto& tablets_shard : _tablets_shards) {
1551
0
        std::shared_lock rdlock(tablets_shard.lock);
1552
0
        for (const auto& item : tablets_shard.tablet_map) {
1553
0
            TabletSharedPtr tablet = item.second;
1554
0
            if (tablets_info.size() >= num) {
1555
0
                return;
1556
0
            }
1557
0
            if (tablet == nullptr) {
1558
0
                continue;
1559
0
            }
1560
0
            tablets_info.push_back(tablet->get_tablet_info());
1561
0
        }
1562
0
    }
1563
0
}
1564
1565
3.01k
std::shared_mutex& TabletManager::_get_tablets_shard_lock(TTabletId tabletId) {
1566
3.01k
    return _get_tablets_shard(tabletId).lock;
1567
3.01k
}
1568
1569
4.43k
TabletManager::tablet_map_t& TabletManager::_get_tablet_map(TTabletId tabletId) {
1570
4.43k
    return _get_tablets_shard(tabletId).tablet_map;
1571
4.43k
}
1572
1573
8.68k
TabletManager::tablets_shard& TabletManager::_get_tablets_shard(TTabletId tabletId) {
1574
8.68k
    return _tablets_shards[tabletId & _tablets_shards_mask];
1575
8.68k
}
1576
1577
void TabletManager::get_tablets_distribution_on_different_disks(
1578
        std::map<int64_t, std::map<DataDir*, int64_t>>& tablets_num_on_disk,
1579
0
        std::map<int64_t, std::map<DataDir*, std::vector<TabletSize>>>& tablets_info_on_disk) {
1580
0
    std::vector<DataDir*> data_dirs = _engine.get_stores();
1581
0
    std::map<int64_t, Partition> partitions;
1582
0
    {
1583
        // When drop tablet, '_partitions_lock' is locked in 'tablet_shard_lock'.
1584
        // To avoid locking 'tablet_shard_lock' in '_partitions_lock', we lock and
1585
        // copy _partitions here.
1586
0
        std::shared_lock rdlock(_partitions_lock);
1587
0
        partitions = _partitions;
1588
0
    }
1589
1590
0
    for (const auto& [partition_id, partition] : partitions) {
1591
0
        std::map<DataDir*, int64_t> tablets_num;
1592
0
        std::map<DataDir*, std::vector<TabletSize>> tablets_info;
1593
0
        for (auto* data_dir : data_dirs) {
1594
0
            tablets_num[data_dir] = 0;
1595
0
        }
1596
1597
0
        for (const auto& tablet_info : partition.tablets) {
1598
            // get_tablet() will hold 'tablet_shard_lock'
1599
0
            TabletSharedPtr tablet = get_tablet(tablet_info.tablet_id);
1600
0
            if (tablet == nullptr) {
1601
0
                continue;
1602
0
            }
1603
0
            DataDir* data_dir = tablet->data_dir();
1604
0
            size_t tablet_footprint = tablet->tablet_footprint();
1605
0
            tablets_num[data_dir]++;
1606
0
            TabletSize tablet_size(tablet_info.tablet_id, tablet_footprint);
1607
0
            tablets_info[data_dir].push_back(tablet_size);
1608
0
        }
1609
0
        tablets_num_on_disk[partition_id] = tablets_num;
1610
0
        tablets_info_on_disk[partition_id] = tablets_info;
1611
0
    }
1612
0
}
1613
1614
struct SortCtx {
1615
    SortCtx(TabletSharedPtr tablet, RowsetSharedPtr rowset, int64_t cooldown_timestamp,
1616
            int64_t file_size)
1617
0
            : tablet(tablet), cooldown_timestamp(cooldown_timestamp), file_size(file_size) {}
1618
    TabletSharedPtr tablet;
1619
    RowsetSharedPtr rowset;
1620
    // to ensure the tablet with -1 would always be greater than other
1621
    uint64_t cooldown_timestamp;
1622
    int64_t file_size;
1623
0
    bool operator<(const SortCtx& other) const {
1624
0
        if (this->cooldown_timestamp == other.cooldown_timestamp) {
1625
0
            return this->file_size > other.file_size;
1626
0
        }
1627
0
        return this->cooldown_timestamp < other.cooldown_timestamp;
1628
0
    }
1629
};
1630
1631
void TabletManager::get_cooldown_tablets(std::vector<TabletSharedPtr>* tablets,
1632
                                         std::vector<RowsetSharedPtr>* rowsets,
1633
19
                                         std::function<bool(const TabletSharedPtr&)> skip_tablet) {
1634
19
    std::vector<SortCtx> sort_ctx_vec;
1635
19
    std::vector<std::weak_ptr<Tablet>> candidates;
1636
19
    for_each_tablet([&](const TabletSharedPtr& tablet) { candidates.emplace_back(tablet); },
1637
19
                    filter_all_tablets);
1638
19
    auto get_cooldown_tablet = [&sort_ctx_vec, &skip_tablet](std::weak_ptr<Tablet>& t) {
1639
0
        const TabletSharedPtr& tablet = t.lock();
1640
0
        RowsetSharedPtr rowset = nullptr;
1641
0
        if (UNLIKELY(nullptr == tablet)) {
1642
0
            return;
1643
0
        }
1644
0
        std::shared_lock rdlock(tablet->get_header_lock());
1645
0
        int64_t cooldown_timestamp = -1;
1646
0
        size_t file_size = -1;
1647
0
        if (!skip_tablet(tablet) &&
1648
0
            (rowset = tablet->need_cooldown(&cooldown_timestamp, &file_size))) {
1649
0
            sort_ctx_vec.emplace_back(tablet, rowset, cooldown_timestamp, file_size);
1650
0
        }
1651
0
    };
1652
19
    std::for_each(candidates.begin(), candidates.end(), get_cooldown_tablet);
1653
1654
19
    std::sort(sort_ctx_vec.begin(), sort_ctx_vec.end());
1655
1656
19
    for (SortCtx& ctx : sort_ctx_vec) {
1657
0
        VLOG_DEBUG << "get cooldown tablet: " << ctx.tablet->tablet_id();
1658
0
        tablets->push_back(std::move(ctx.tablet));
1659
0
        rowsets->push_back(std::move(ctx.rowset));
1660
0
    }
1661
19
}
1662
1663
0
void TabletManager::get_all_tablets_storage_format(TCheckStorageFormatResult* result) {
1664
0
    DCHECK(result != nullptr);
1665
0
    auto handler = [result](const TabletSharedPtr& tablet) {
1666
0
        if (tablet->all_beta()) {
1667
0
            result->v2_tablets.push_back(tablet->tablet_id());
1668
0
        } else {
1669
0
            result->v1_tablets.push_back(tablet->tablet_id());
1670
0
        }
1671
0
    };
1672
1673
0
    for_each_tablet(handler, filter_all_tablets);
1674
0
    result->__isset.v1_tablets = true;
1675
0
    result->__isset.v2_tablets = true;
1676
0
}
1677
1678
0
std::set<int64_t> TabletManager::check_all_tablet_segment(bool repair) {
1679
0
    std::set<int64_t> bad_tablets;
1680
0
    std::map<int64_t, std::vector<int64_t>> repair_shard_bad_tablets;
1681
0
    auto handler = [&](const TabletSharedPtr& tablet) {
1682
0
        if (!tablet->check_all_rowset_segment()) {
1683
0
            int64_t tablet_id = tablet->tablet_id();
1684
0
            bad_tablets.insert(tablet_id);
1685
0
            if (repair) {
1686
0
                repair_shard_bad_tablets[tablet_id & _tablets_shards_mask].push_back(tablet_id);
1687
0
            }
1688
0
        }
1689
0
    };
1690
0
    for_each_tablet(handler, filter_all_tablets);
1691
1692
0
    for (const auto& [shard_index, shard_tablets] : repair_shard_bad_tablets) {
1693
0
        auto& tablets_shard = _tablets_shards[shard_index];
1694
0
        auto& tablet_map = tablets_shard.tablet_map;
1695
0
        std::lock_guard<std::shared_mutex> wrlock(tablets_shard.lock);
1696
0
        for (auto tablet_id : shard_tablets) {
1697
0
            auto it = tablet_map.find(tablet_id);
1698
0
            if (it == tablet_map.end()) {
1699
0
                bad_tablets.erase(tablet_id);
1700
0
                LOG(WARNING) << "Bad tablet has be removed. tablet_id=" << tablet_id;
1701
0
            } else {
1702
0
                const auto& tablet = it->second;
1703
0
                static_cast<void>(tablet->set_tablet_state(TABLET_SHUTDOWN));
1704
0
                tablet->save_meta();
1705
0
                {
1706
0
                    std::lock_guard<std::shared_mutex> shutdown_tablets_wrlock(
1707
0
                            _shutdown_tablets_lock);
1708
0
                    _shutdown_tablets.push_back(tablet);
1709
0
                }
1710
0
                LOG(WARNING) << "There are some segments lost, set tablet to shutdown state."
1711
0
                             << "tablet_id=" << tablet->tablet_id()
1712
0
                             << ", tablet_path=" << tablet->tablet_path();
1713
0
            }
1714
0
        }
1715
0
    }
1716
1717
0
    return bad_tablets;
1718
0
}
1719
1720
bool TabletManager::update_tablet_partition_id(::doris::TPartitionId partition_id,
1721
0
                                               ::doris::TTabletId tablet_id) {
1722
0
    std::shared_lock rdlock(_get_tablets_shard_lock(tablet_id));
1723
0
    TabletSharedPtr tablet = _get_tablet_unlocked(tablet_id);
1724
0
    if (tablet == nullptr) {
1725
0
        LOG(WARNING) << "get tablet err partition_id: " << partition_id
1726
0
                     << " tablet_id:" << tablet_id;
1727
0
        return false;
1728
0
    }
1729
0
    _remove_tablet_from_partition(tablet);
1730
0
    auto st = tablet->tablet_meta()->set_partition_id(partition_id);
1731
0
    if (!st.ok()) {
1732
0
        LOG(WARNING) << "set partition id err partition_id: " << partition_id
1733
0
                     << " tablet_id:" << tablet_id;
1734
0
        return false;
1735
0
    }
1736
0
    _add_tablet_to_partition(tablet);
1737
0
    return true;
1738
0
}
1739
1740
} // end namespace doris