Coverage Report

Created: 2025-06-09 07:11

/root/doris/be/src/olap/tablet.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.h"
19
20
#include <butil/logging.h>
21
#include <bvar/reducer.h>
22
#include <bvar/window.h>
23
#include <fmt/format.h>
24
#include <gen_cpp/FrontendService_types.h>
25
#include <gen_cpp/MasterService_types.h>
26
#include <gen_cpp/Metrics_types.h>
27
#include <gen_cpp/olap_file.pb.h>
28
#include <gen_cpp/types.pb.h>
29
#include <glog/logging.h>
30
#include <rapidjson/document.h>
31
#include <rapidjson/encodings.h>
32
#include <rapidjson/prettywriter.h>
33
#include <rapidjson/rapidjson.h>
34
#include <rapidjson/stringbuffer.h>
35
36
#include <algorithm>
37
#include <atomic>
38
#include <boost/container/detail/std_fwd.hpp>
39
#include <cstdint>
40
#include <roaring/roaring.hh>
41
42
#include "common/compiler_util.h" // IWYU pragma: keep
43
// IWYU pragma: no_include <bits/chrono.h>
44
#include <chrono> // IWYU pragma: keep
45
#include <filesystem>
46
#include <iterator>
47
#include <limits>
48
#include <map>
49
#include <memory>
50
#include <mutex>
51
#include <set>
52
#include <shared_mutex>
53
#include <string>
54
#include <tuple>
55
#include <type_traits>
56
#include <unordered_map>
57
#include <unordered_set>
58
59
#include "agent/utils.h"
60
#include "common/config.h"
61
#include "common/consts.h"
62
#include "common/logging.h"
63
#include "common/signal_handler.h"
64
#include "common/status.h"
65
#include "io/fs/file_reader.h"
66
#include "io/fs/file_reader_writer_fwd.h"
67
#include "io/fs/file_system.h"
68
#include "io/fs/file_writer.h"
69
#include "io/fs/path.h"
70
#include "io/fs/remote_file_system.h"
71
#include "io/io_common.h"
72
#include "olap/base_compaction.h"
73
#include "olap/base_tablet.h"
74
#include "olap/binlog.h"
75
#include "olap/cumulative_compaction.h"
76
#include "olap/cumulative_compaction_policy.h"
77
#include "olap/cumulative_compaction_time_series_policy.h"
78
#include "olap/delete_bitmap_calculator.h"
79
#include "olap/full_compaction.h"
80
#include "olap/memtable.h"
81
#include "olap/olap_common.h"
82
#include "olap/olap_define.h"
83
#include "olap/olap_meta.h"
84
#include "olap/primary_key_index.h"
85
#include "olap/rowset/beta_rowset.h"
86
#include "olap/rowset/rowset.h"
87
#include "olap/rowset/rowset_factory.h"
88
#include "olap/rowset/rowset_fwd.h"
89
#include "olap/rowset/rowset_meta.h"
90
#include "olap/rowset/rowset_meta_manager.h"
91
#include "olap/rowset/rowset_writer.h"
92
#include "olap/rowset/rowset_writer_context.h"
93
#include "olap/rowset/segment_v2/column_reader.h"
94
#include "olap/rowset/segment_v2/common.h"
95
#include "olap/rowset/segment_v2/indexed_column_reader.h"
96
#include "olap/rowset/vertical_beta_rowset_writer.h"
97
#include "olap/schema_change.h"
98
#include "olap/single_replica_compaction.h"
99
#include "olap/storage_engine.h"
100
#include "olap/storage_policy.h"
101
#include "olap/tablet_manager.h"
102
#include "olap/tablet_meta.h"
103
#include "olap/tablet_meta_manager.h"
104
#include "olap/tablet_schema.h"
105
#include "olap/txn_manager.h"
106
#include "olap/types.h"
107
#include "olap/utils.h"
108
#include "segment_loader.h"
109
#include "service/point_query_executor.h"
110
#include "tablet.h"
111
#include "util/bvar_helper.h"
112
#include "util/debug_points.h"
113
#include "util/defer_op.h"
114
#include "util/doris_metrics.h"
115
#include "util/pretty_printer.h"
116
#include "util/scoped_cleanup.h"
117
#include "util/stopwatch.hpp"
118
#include "util/threadpool.h"
119
#include "util/time.h"
120
#include "util/trace.h"
121
#include "util/uid_util.h"
122
#include "util/work_thread_pool.hpp"
123
#include "vec/columns/column.h"
124
#include "vec/columns/column_string.h"
125
#include "vec/common/schema_util.h"
126
#include "vec/common/string_ref.h"
127
#include "vec/data_types/data_type.h"
128
#include "vec/data_types/data_type_factory.hpp"
129
#include "vec/data_types/serde/data_type_serde.h"
130
#include "vec/jsonb/serialize.h"
131
132
namespace doris {
133
class TupleDescriptor;
134
135
namespace vectorized {
136
class Block;
137
} // namespace vectorized
138
139
using namespace ErrorCode;
140
using namespace std::chrono_literals;
141
142
using std::pair;
143
using std::string;
144
using std::vector;
145
using io::FileSystemSPtr;
146
147
namespace {
148
149
bvar::Adder<uint64_t> exceed_version_limit_counter;
150
bvar::Window<bvar::Adder<uint64_t>> exceed_version_limit_counter_minute(
151
        &exceed_version_limit_counter, 60);
152
bvar::Adder<uint64_t> cooldown_pending_task("cooldown_pending_task");
153
bvar::Adder<uint64_t> cooldown_processing_task("cooldown_processing_task");
154
155
14.4k
void set_last_failure_time(Tablet* tablet, const Compaction& compaction, int64_t ms) {
156
14.4k
    switch (compaction.compaction_type()) {
157
14.3k
    case ReaderType::READER_CUMULATIVE_COMPACTION:
158
14.3k
        tablet->set_last_cumu_compaction_failure_time(ms);
159
14.3k
        return;
160
90
    case ReaderType::READER_BASE_COMPACTION:
161
90
        tablet->set_last_base_compaction_failure_time(ms);
162
90
        return;
163
10
    case ReaderType::READER_FULL_COMPACTION:
164
10
        tablet->set_last_full_compaction_failure_time(ms);
165
10
        return;
166
0
    default:
167
0
        LOG(FATAL) << "invalid compaction type " << compaction.compaction_name()
168
0
                   << " tablet_id: " << tablet->tablet_id();
169
14.4k
    }
170
14.4k
};
171
172
} // namespace
173
174
bvar::Adder<uint64_t> unused_remote_rowset_num("unused_remote_rowset_num");
175
176
WriteCooldownMetaExecutors::WriteCooldownMetaExecutors(size_t executor_nums)
177
4
        : _executor_nums(executor_nums) {
178
24
    for (size_t i = 0; i < _executor_nums; i++) {
179
20
        std::unique_ptr<PriorityThreadPool> pool;
180
20
        static_cast<void>(ThreadPoolBuilder("WriteCooldownMetaExecutor")
181
20
                                  .set_min_threads(1)
182
20
                                  .set_max_threads(1)
183
20
                                  .set_max_queue_size(std::numeric_limits<int>::max())
184
20
                                  .build(&pool));
185
20
        _executors.emplace_back(std::move(pool));
186
20
    }
187
4
}
188
189
1
void WriteCooldownMetaExecutors::stop() {
190
5
    for (auto& pool_ptr : _executors) {
191
5
        if (pool_ptr) {
192
5
            pool_ptr->shutdown();
193
5
        }
194
5
    }
195
1
}
196
197
14
void WriteCooldownMetaExecutors::WriteCooldownMetaExecutors::submit(TabletSharedPtr tablet) {
198
14
    auto tablet_id = tablet->tablet_id();
199
200
14
    {
201
14
        std::shared_lock rdlock(tablet->get_header_lock());
202
14
        if (!tablet->tablet_meta()->cooldown_meta_id().initialized()) {
203
0
            VLOG_NOTICE << "tablet " << tablet_id << " is not cooldown replica";
204
0
            return;
205
0
        }
206
14
        if (tablet->tablet_state() == TABLET_SHUTDOWN) [[unlikely]] {
207
0
            LOG_INFO("tablet {} has been dropped, don't do cooldown", tablet_id);
208
0
            return;
209
0
        }
210
14
    }
211
14
    {
212
        // one tablet could at most have one cooldown task to be done
213
14
        std::unique_lock<std::mutex> lck {_latch};
214
14
        if (_pending_tablets.count(tablet_id) > 0) {
215
0
            return;
216
0
        }
217
14
        _pending_tablets.insert(tablet_id);
218
14
    }
219
220
14
    auto async_write_task = [this, t = std::move(tablet)]() {
221
14
        {
222
14
            std::unique_lock<std::mutex> lck {_latch};
223
14
            _pending_tablets.erase(t->tablet_id());
224
14
        }
225
14
        SCOPED_ATTACH_TASK(ExecEnv::GetInstance()->orphan_mem_tracker());
226
14
        auto s = t->write_cooldown_meta();
227
14
        if (s.ok()) {
228
14
            return;
229
14
        }
230
0
        if (!s.is<ABORTED>()) {
231
0
            LOG_EVERY_SECOND(WARNING)
232
0
                    << "write tablet " << t->tablet_id() << " cooldown meta failed because: " << s;
233
0
            submit(t);
234
0
            return;
235
0
        }
236
0
        VLOG_DEBUG << "tablet " << t->tablet_id() << " is not cooldown replica";
237
0
    };
238
239
14
    cooldown_pending_task << 1;
240
14
    _executors[_get_executor_pos(tablet_id)]->offer([task = std::move(async_write_task)]() {
241
14
        cooldown_pending_task << -1;
242
14
        cooldown_processing_task << 1;
243
14
        task();
244
14
        cooldown_processing_task << -1;
245
14
    });
246
14
}
247
248
Tablet::Tablet(StorageEngine& engine, TabletMetaSharedPtr tablet_meta, DataDir* data_dir,
249
               const std::string_view& cumulative_compaction_type)
250
        : BaseTablet(std::move(tablet_meta)),
251
          _engine(engine),
252
          _data_dir(data_dir),
253
          _is_bad(false),
254
          _last_cumu_compaction_failure_millis(0),
255
          _last_base_compaction_failure_millis(0),
256
          _last_full_compaction_failure_millis(0),
257
          _last_cumu_compaction_success_millis(0),
258
          _last_base_compaction_success_millis(0),
259
          _last_full_compaction_success_millis(0),
260
          _cumulative_point(K_INVALID_CUMULATIVE_POINT),
261
          _newly_created_rowset_num(0),
262
          _last_checkpoint_time(0),
263
          _cumulative_compaction_type(cumulative_compaction_type),
264
          _is_tablet_path_exists(true),
265
          _last_missed_version(-1),
266
231k
          _last_missed_time_s(0) {
267
231k
    if (_data_dir != nullptr) {
268
231k
        _tablet_path = fmt::format("{}/{}/{}/{}/{}", _data_dir->path(), DATA_PREFIX,
269
231k
                                   _tablet_meta->shard_id(), tablet_id(), schema_hash());
270
231k
    }
271
231k
}
272
273
1.71k
bool Tablet::set_tablet_schema_into_rowset_meta() {
274
1.71k
    bool flag = false;
275
3.16k
    for (auto&& rowset_meta : _tablet_meta->all_mutable_rs_metas()) {
276
3.16k
        if (!rowset_meta->tablet_schema()) {
277
0
            rowset_meta->set_tablet_schema(_tablet_meta->tablet_schema());
278
0
            flag = true;
279
0
        }
280
3.16k
    }
281
1.71k
    return flag;
282
1.71k
}
283
284
231k
Status Tablet::_init_once_action() {
285
231k
    Status res = Status::OK();
286
18.4E
    VLOG_NOTICE << "begin to load tablet. tablet=" << tablet_id()
287
18.4E
                << ", version_size=" << _tablet_meta->version_count();
288
289
#ifdef BE_TEST
290
    // init cumulative compaction policy by type
291
    _cumulative_compaction_policy =
292
            CumulativeCompactionPolicyFactory::create_cumulative_compaction_policy(
293
                    _tablet_meta->compaction_policy());
294
#endif
295
296
231k
    for (const auto& rs_meta : _tablet_meta->all_rs_metas()) {
297
5.08k
        Version version = rs_meta->version();
298
5.08k
        RowsetSharedPtr rowset;
299
5.08k
        res = create_rowset(rs_meta, &rowset);
300
5.08k
        if (!res.ok()) {
301
0
            LOG(WARNING) << "fail to init rowset. tablet_id=" << tablet_id()
302
0
                         << ", schema_hash=" << schema_hash() << ", version=" << version
303
0
                         << ", res=" << res;
304
0
            return res;
305
0
        }
306
5.08k
        _rs_version_map[version] = std::move(rowset);
307
5.08k
    }
308
309
    // init stale rowset
310
231k
    for (const auto& stale_rs_meta : _tablet_meta->all_stale_rs_metas()) {
311
40
        Version version = stale_rs_meta->version();
312
40
        RowsetSharedPtr rowset;
313
40
        res = create_rowset(stale_rs_meta, &rowset);
314
40
        if (!res.ok()) {
315
0
            LOG(WARNING) << "fail to init stale rowset. tablet_id:" << tablet_id()
316
0
                         << ", schema_hash:" << schema_hash() << ", version=" << version
317
0
                         << ", res:" << res;
318
0
            return res;
319
0
        }
320
40
        _stale_rs_version_map[version] = std::move(rowset);
321
40
    }
322
323
231k
    return res;
324
231k
}
325
326
231k
Status Tablet::init() {
327
231k
    return _init_once.call([this] { return _init_once_action(); });
328
231k
}
329
330
// should save tablet meta to remote meta store
331
// if it's a primary replica
332
585k
void Tablet::save_meta() {
333
585k
    check_table_size_correctness();
334
585k
    auto res = _tablet_meta->save_meta(_data_dir);
335
585k
    CHECK_EQ(res, Status::OK()) << "fail to save tablet_meta. res=" << res
336
0
                                << ", root=" << _data_dir->path();
337
585k
}
338
339
// Caller should hold _meta_lock.
340
Status Tablet::revise_tablet_meta(const std::vector<RowsetSharedPtr>& to_add,
341
                                  const std::vector<RowsetSharedPtr>& to_delete,
342
0
                                  bool is_incremental_clone) {
343
0
    LOG(INFO) << "begin to revise tablet. tablet_id=" << tablet_id();
344
    // 1. for incremental clone, we have to add the rowsets first to make it easy to compute
345
    //    all the delete bitmaps, and it's easy to delete them if we end up with a failure
346
    // 2. for full clone, we can calculate delete bitmaps on the cloned rowsets directly.
347
0
    if (is_incremental_clone) {
348
0
        CHECK(to_delete.empty()); // don't need to delete rowsets
349
0
        add_rowsets(to_add);
350
        // reconstruct from tablet meta
351
0
        _timestamped_version_tracker.construct_versioned_tracker(_tablet_meta->all_rs_metas());
352
0
    }
353
354
0
    Status calc_bm_status;
355
0
    std::vector<RowsetSharedPtr> base_rowsets_for_full_clone = to_add; // copy vector
356
0
    while (keys_type() == UNIQUE_KEYS && enable_unique_key_merge_on_write()) {
357
0
        std::vector<RowsetSharedPtr> calc_delete_bitmap_rowsets;
358
0
        int64_t to_add_min_version = INT64_MAX;
359
0
        int64_t to_add_max_version = INT64_MIN;
360
0
        for (auto& rs : to_add) {
361
0
            if (to_add_min_version > rs->start_version()) {
362
0
                to_add_min_version = rs->start_version();
363
0
            }
364
0
            if (to_add_max_version < rs->end_version()) {
365
0
                to_add_max_version = rs->end_version();
366
0
            }
367
0
        }
368
0
        Version calc_delete_bitmap_ver;
369
0
        if (is_incremental_clone) {
370
            // From the rowset of to_add with smallest version, all other rowsets
371
            // need to recalculate the delete bitmap
372
            // For example:
373
            // local tablet: [0-1] [2-5] [6-6] [9-10]
374
            // clone tablet: [7-7] [8-8]
375
            // new tablet:   [0-1] [2-5] [6-6] [7-7] [8-8] [9-10]
376
            // [7-7] [8-8] [9-10] need to recalculate delete bitmap
377
0
            calc_delete_bitmap_ver = Version(to_add_min_version, max_version_unlocked());
378
0
        } else {
379
            // the delete bitmap of to_add's rowsets has clone from remote when full clone.
380
            // only other rowsets in local need to recalculate the delete bitmap.
381
            // For example:
382
            // local tablet: [0-1]x [2-5]x [6-6]x [7-7]x [9-10]
383
            // clone tablet: [0-1]  [2-4]  [5-6]  [7-8]
384
            // new tablet:   [0-1]  [2-4]  [5-6]  [7-8] [9-10]
385
            // only [9-10] need to recalculate delete bitmap
386
0
            CHECK_EQ(to_add_min_version, 0) << "to_add_min_version is: " << to_add_min_version;
387
0
            calc_delete_bitmap_ver = Version(to_add_max_version + 1, max_version_unlocked());
388
0
        }
389
390
0
        if (calc_delete_bitmap_ver.first <= calc_delete_bitmap_ver.second) {
391
0
            calc_bm_status = capture_consistent_rowsets_unlocked(calc_delete_bitmap_ver,
392
0
                                                                 &calc_delete_bitmap_rowsets);
393
0
            if (!calc_bm_status.ok()) {
394
0
                LOG(WARNING) << "fail to capture_consistent_rowsets, res: " << calc_bm_status;
395
0
                break;
396
0
            }
397
            // FIXME(plat1ko): Use `const TabletSharedPtr&` as parameter
398
0
            auto self = _engine.tablet_manager()->get_tablet(tablet_id());
399
0
            CHECK(self);
400
0
            for (auto rs : calc_delete_bitmap_rowsets) {
401
0
                if (is_incremental_clone) {
402
0
                    calc_bm_status = update_delete_bitmap_without_lock(self, rs);
403
0
                } else {
404
0
                    calc_bm_status = update_delete_bitmap_without_lock(
405
0
                            self, rs, &base_rowsets_for_full_clone);
406
0
                    base_rowsets_for_full_clone.push_back(rs);
407
0
                }
408
0
                if (!calc_bm_status.ok()) {
409
0
                    LOG(WARNING) << "fail to update_delete_bitmap_without_lock, res: "
410
0
                                 << calc_bm_status;
411
0
                    break;
412
0
                }
413
0
            }
414
0
        }
415
0
        break; // while (keys_type() == UNIQUE_KEYS && enable_unique_key_merge_on_write())
416
0
    }
417
418
0
    DBUG_EXECUTE_IF("Tablet.revise_tablet_meta_fail", {
419
0
        auto ptablet_id = dp->param("tablet_id", 0);
420
0
        if (tablet_id() == ptablet_id) {
421
0
            LOG(INFO) << "injected revies_tablet_meta failure for tabelt: " << ptablet_id;
422
0
            calc_bm_status = Status::InternalError("fault injection error");
423
0
        }
424
0
    });
425
426
    // error handling
427
0
    if (!calc_bm_status.ok()) {
428
0
        if (is_incremental_clone) {
429
0
            RETURN_IF_ERROR(delete_rowsets(to_add, false));
430
0
            LOG(WARNING) << "incremental clone on tablet: " << tablet_id() << " failed due to "
431
0
                         << calc_bm_status.msg() << ", revert " << to_add.size()
432
0
                         << " rowsets added before.";
433
0
        } else {
434
0
            LOG(WARNING) << "full clone on tablet: " << tablet_id() << " failed due to "
435
0
                         << calc_bm_status.msg() << ", will not update tablet meta.";
436
0
        }
437
0
        return calc_bm_status;
438
0
    }
439
440
    // full clone, calculate delete bitmap succeeded, update rowset
441
0
    if (!is_incremental_clone) {
442
0
        RETURN_IF_ERROR(delete_rowsets(to_delete, false));
443
0
        add_rowsets(to_add);
444
        // reconstruct from tablet meta
445
0
        _timestamped_version_tracker.construct_versioned_tracker(_tablet_meta->all_rs_metas());
446
447
        // check the rowsets used for delete bitmap calculation is equal to the rowsets
448
        // that we can capture by version
449
0
        if (keys_type() == UNIQUE_KEYS && enable_unique_key_merge_on_write()) {
450
0
            Version full_version = Version(0, max_version_unlocked());
451
0
            std::vector<RowsetSharedPtr> expected_rowsets;
452
0
            auto st = capture_consistent_rowsets_unlocked(full_version, &expected_rowsets);
453
0
            DCHECK(st.ok()) << st;
454
0
            DCHECK_EQ(base_rowsets_for_full_clone.size(), expected_rowsets.size());
455
0
            if (st.ok() && base_rowsets_for_full_clone.size() != expected_rowsets.size())
456
0
                    [[unlikely]] {
457
0
                LOG(WARNING) << "full clone succeeded, but the count("
458
0
                             << base_rowsets_for_full_clone.size()
459
0
                             << ") of base rowsets used for delete bitmap calculation is not match "
460
0
                                "expect count("
461
0
                             << expected_rowsets.size() << ") we capture from tablet meta";
462
0
            }
463
0
        }
464
0
    }
465
466
    // clear stale rowset
467
0
    for (auto& [v, rs] : _stale_rs_version_map) {
468
0
        _engine.add_unused_rowset(rs);
469
0
    }
470
0
    _stale_rs_version_map.clear();
471
0
    _tablet_meta->clear_stale_rowset();
472
0
    save_meta();
473
474
0
    LOG(INFO) << "finish to revise tablet. tablet_id=" << tablet_id();
475
0
    return Status::OK();
476
0
}
477
478
245k
Status Tablet::add_rowset(RowsetSharedPtr rowset) {
479
245k
    DCHECK(rowset != nullptr);
480
245k
    std::lock_guard<std::shared_mutex> wrlock(_meta_lock);
481
245k
    SCOPED_SIMPLE_TRACE_IF_TIMEOUT(TRACE_TABLET_LOCK_THRESHOLD);
482
    // If the rowset already exist, just return directly.  The rowset_id is an unique-id,
483
    // we can use it to check this situation.
484
245k
    if (_contains_rowset(rowset->rowset_id())) {
485
12
        return Status::OK();
486
12
    }
487
    // Otherwise, the version should be not contained in any existing rowset.
488
245k
    RETURN_IF_ERROR(_contains_version(rowset->version()));
489
490
245k
    RETURN_IF_ERROR(_tablet_meta->add_rs_meta(rowset->rowset_meta()));
491
245k
    _rs_version_map[rowset->version()] = rowset;
492
245k
    _timestamped_version_tracker.add_version(rowset->version());
493
245k
    add_compaction_score(rowset->rowset_meta()->get_compaction_score());
494
495
245k
    std::vector<RowsetSharedPtr> rowsets_to_delete;
496
    // yiguolei: temp code, should remove the rowset contains by this rowset
497
    // but it should be removed in multi path version
498
264k
    for (auto& it : _rs_version_map) {
499
264k
        if (rowset->version().contains(it.first) && rowset->version() != it.first) {
500
0
            CHECK(it.second != nullptr)
501
0
                    << "there exist a version=" << it.first
502
0
                    << " contains the input rs with version=" << rowset->version()
503
0
                    << ", but the related rs is null";
504
0
            rowsets_to_delete.push_back(it.second);
505
0
        }
506
264k
    }
507
245k
    std::vector<RowsetSharedPtr> empty_vec;
508
245k
    RETURN_IF_ERROR(modify_rowsets(empty_vec, rowsets_to_delete));
509
245k
    ++_newly_created_rowset_num;
510
245k
    return Status::OK();
511
245k
}
512
513
48.9k
bool Tablet::rowset_exists_unlocked(const RowsetSharedPtr& rowset) {
514
48.9k
    if (auto it = _rs_version_map.find(rowset->version()); it == _rs_version_map.end()) {
515
0
        return false;
516
48.9k
    } else if (rowset->rowset_id() != it->second->rowset_id()) {
517
0
        return false;
518
0
    }
519
48.9k
    return true;
520
48.9k
}
521
522
Status Tablet::modify_rowsets(std::vector<RowsetSharedPtr>& to_add,
523
261k
                              std::vector<RowsetSharedPtr>& to_delete, bool check_delete) {
524
    // the compaction process allow to compact the single version, eg: version[4-4].
525
    // this kind of "single version compaction" has same "input version" and "output version".
526
    // which means "to_add->version()" equals to "to_delete->version()".
527
    // So we should delete the "to_delete" before adding the "to_add",
528
    // otherwise, the "to_add" will be deleted from _rs_version_map, eventually.
529
    //
530
    // And if the version of "to_add" and "to_delete" are exactly same. eg:
531
    // to_add:      [7-7]
532
    // to_delete:   [7-7]
533
    // In this case, we no longer need to add the rowset in "to_delete" to
534
    // _stale_rs_version_map, but can delete it directly.
535
536
261k
    if (to_add.empty() && to_delete.empty()) {
537
245k
        return Status::OK();
538
245k
    }
539
540
16.0k
    if (check_delete) {
541
131k
        for (auto&& rs : to_delete) {
542
131k
            if (auto it = _rs_version_map.find(rs->version()); it == _rs_version_map.end()) {
543
14
                return Status::Error<DELETE_VERSION_ERROR>(
544
14
                        "try to delete not exist version {} from {}", rs->version().to_string(),
545
14
                        tablet_id());
546
131k
            } else if (rs->rowset_id() != it->second->rowset_id()) {
547
0
                return Status::Error<DELETE_VERSION_ERROR>(
548
0
                        "try to delete version {} from {}, but rowset id changed, delete rowset id "
549
0
                        "is {}, exists rowsetid is {}",
550
0
                        rs->version().to_string(), tablet_id(), rs->rowset_id().to_string(),
551
0
                        it->second->rowset_id().to_string());
552
0
            }
553
131k
        }
554
16.0k
    }
555
556
15.8k
    bool same_version = true;
557
15.8k
    std::sort(to_add.begin(), to_add.end(), Rowset::comparator);
558
15.8k
    std::sort(to_delete.begin(), to_delete.end(), Rowset::comparator);
559
15.8k
    if (to_add.size() == to_delete.size()) {
560
4.27k
        for (int i = 0; i < to_add.size(); ++i) {
561
3.08k
            if (to_add[i]->version() != to_delete[i]->version()) {
562
0
                same_version = false;
563
0
                break;
564
0
            }
565
3.08k
        }
566
14.6k
    } else {
567
14.6k
        same_version = false;
568
14.6k
    }
569
570
15.8k
    std::vector<RowsetMetaSharedPtr> rs_metas_to_delete;
571
131k
    for (auto& rs : to_delete) {
572
131k
        rs_metas_to_delete.push_back(rs->rowset_meta());
573
131k
        _rs_version_map.erase(rs->version());
574
575
131k
        if (!same_version) {
576
            // put compaction rowsets in _stale_rs_version_map.
577
128k
            _stale_rs_version_map[rs->version()] = rs;
578
128k
        }
579
131k
    }
580
581
15.8k
    std::vector<RowsetMetaSharedPtr> rs_metas_to_add;
582
17.8k
    for (auto& rs : to_add) {
583
17.8k
        rs_metas_to_add.push_back(rs->rowset_meta());
584
17.8k
        _rs_version_map[rs->version()] = rs;
585
586
17.8k
        if (!same_version) {
587
            // If version are same, then _timestamped_version_tracker
588
            // already has this version, no need to add again.
589
14.8k
            _timestamped_version_tracker.add_version(rs->version());
590
14.8k
        }
591
17.8k
        ++_newly_created_rowset_num;
592
17.8k
    }
593
594
15.8k
    _tablet_meta->modify_rs_metas(rs_metas_to_add, rs_metas_to_delete, same_version);
595
596
15.8k
    if (!same_version) {
597
        // add rs_metas_to_delete to tracker
598
14.8k
        _timestamped_version_tracker.add_stale_path_version(rs_metas_to_delete);
599
14.8k
    } else {
600
        // delete rowset in "to_delete" directly
601
3.08k
        for (auto& rs : to_delete) {
602
3.08k
            LOG(INFO) << "add unused rowset " << rs->rowset_id() << " because of same version";
603
3.08k
            if (rs->is_local()) {
604
3.08k
                _engine.add_unused_rowset(rs);
605
3.08k
            }
606
3.08k
        }
607
1.03k
    }
608
609
15.8k
    int32_t add_score = 0;
610
17.8k
    for (auto rs : to_add) {
611
17.8k
        add_score += rs->rowset_meta()->get_compaction_score();
612
17.8k
    }
613
15.8k
    int32_t sub_score = 0;
614
131k
    for (auto rs : to_delete) {
615
131k
        sub_score += rs->rowset_meta()->get_compaction_score();
616
131k
    }
617
15.8k
    add_compaction_score(add_score - sub_score);
618
619
15.8k
    return Status::OK();
620
15.8k
}
621
622
14
void Tablet::add_rowsets(const std::vector<RowsetSharedPtr>& to_add) {
623
14
    if (to_add.empty()) {
624
0
        return;
625
0
    }
626
14
    std::vector<RowsetMetaSharedPtr> rs_metas;
627
14
    rs_metas.reserve(to_add.size());
628
14
    for (auto& rs : to_add) {
629
14
        _rs_version_map.emplace(rs->version(), rs);
630
14
        _timestamped_version_tracker.add_version(rs->version());
631
14
        rs_metas.push_back(rs->rowset_meta());
632
14
    }
633
14
    _tablet_meta->modify_rs_metas(rs_metas, {});
634
14
}
635
636
9.16k
Status Tablet::delete_rowsets(const std::vector<RowsetSharedPtr>& to_delete, bool move_to_stale) {
637
9.16k
    if (to_delete.empty()) {
638
0
        return Status::OK();
639
0
    }
640
9.16k
    std::vector<RowsetMetaSharedPtr> rs_metas;
641
9.16k
    rs_metas.reserve(to_delete.size());
642
9.17k
    for (const auto& rs : to_delete) {
643
9.17k
        rs_metas.push_back(rs->rowset_meta());
644
9.17k
        _rs_version_map.erase(rs->version());
645
9.17k
    }
646
9.16k
    _tablet_meta->modify_rs_metas({}, rs_metas, !move_to_stale);
647
9.16k
    if (move_to_stale) {
648
0
        for (const auto& rs : to_delete) {
649
0
            _stale_rs_version_map[rs->version()] = rs;
650
0
        }
651
0
        _timestamped_version_tracker.add_stale_path_version(rs_metas);
652
9.16k
    } else {
653
9.17k
        for (const auto& rs : to_delete) {
654
9.17k
            _timestamped_version_tracker.delete_version(rs->version());
655
9.17k
            if (rs->is_local()) {
656
9.17k
                _engine.add_unused_rowset(rs);
657
9.17k
                RETURN_IF_ERROR(RowsetMetaManager::remove(_data_dir->get_meta(), tablet_uid(),
658
9.17k
                                                          rs->rowset_meta()->rowset_id()));
659
9.17k
            }
660
9.17k
        }
661
9.16k
    }
662
9.16k
    return Status::OK();
663
9.16k
}
664
665
0
RowsetSharedPtr Tablet::_rowset_with_largest_size() {
666
0
    RowsetSharedPtr largest_rowset = nullptr;
667
0
    for (auto& it : _rs_version_map) {
668
0
        if (it.second->empty() || it.second->zero_num_rows()) {
669
0
            continue;
670
0
        }
671
0
        if (largest_rowset == nullptr || it.second->rowset_meta()->index_disk_size() >
672
0
                                                 largest_rowset->rowset_meta()->index_disk_size()) {
673
0
            largest_rowset = it.second;
674
0
        }
675
0
    }
676
677
0
    return largest_rowset;
678
0
}
679
680
// add inc rowset should not persist tablet meta, because it will be persisted when publish txn.
681
269k
Status Tablet::add_inc_rowset(const RowsetSharedPtr& rowset) {
682
269k
    DCHECK(rowset != nullptr);
683
269k
    std::lock_guard<std::shared_mutex> wrlock(_meta_lock);
684
269k
    SCOPED_SIMPLE_TRACE_IF_TIMEOUT(TRACE_TABLET_LOCK_THRESHOLD);
685
269k
    if (_contains_rowset(rowset->rowset_id())) {
686
0
        return Status::OK();
687
0
    }
688
269k
    RETURN_IF_ERROR(_contains_version(rowset->version()));
689
690
269k
    RETURN_IF_ERROR(_tablet_meta->add_rs_meta(rowset->rowset_meta()));
691
269k
    _rs_version_map[rowset->version()] = rowset;
692
693
269k
    _timestamped_version_tracker.add_version(rowset->version());
694
695
269k
    ++_newly_created_rowset_num;
696
697
269k
    add_compaction_score(rowset->rowset_meta()->get_compaction_score());
698
699
269k
    return Status::OK();
700
269k
}
701
702
106k
void Tablet::_delete_stale_rowset_by_version(const Version& version) {
703
106k
    RowsetMetaSharedPtr rowset_meta = _tablet_meta->acquire_stale_rs_meta_by_version(version);
704
106k
    if (rowset_meta == nullptr) {
705
8
        return;
706
8
    }
707
106k
    _tablet_meta->delete_stale_rs_meta_by_version(version);
708
106k
    VLOG_NOTICE << "delete stale rowset. tablet=" << tablet_id() << ", version=" << version;
709
106k
}
710
711
1.45M
void Tablet::delete_expired_stale_rowset() {
712
1.45M
    if (config::enable_mow_verbose_log) {
713
0
        LOG_INFO("begin delete_expired_stale_rowset for tablet={}", tablet_id());
714
0
    }
715
1.45M
    int64_t now = UnixSeconds();
716
1.45M
    std::vector<std::pair<Version, std::vector<RowsetId>>> deleted_stale_rowsets;
717
    // hold write lock while processing stable rowset
718
1.45M
    {
719
1.45M
        std::lock_guard<std::shared_mutex> wrlock(_meta_lock);
720
1.45M
        SCOPED_SIMPLE_TRACE_IF_TIMEOUT(TRACE_TABLET_LOCK_THRESHOLD);
721
        // Compute the end time to delete rowsets, when a expired rowset createtime less then this time, it will be deleted.
722
1.45M
        double expired_stale_sweep_endtime =
723
1.45M
                ::difftime(now, config::tablet_rowset_stale_sweep_time_sec);
724
1.45M
        if (config::tablet_rowset_stale_sweep_by_size) {
725
0
            expired_stale_sweep_endtime = now;
726
0
        }
727
728
1.45M
        std::vector<int64_t> path_id_vec;
729
        // capture the path version to delete
730
1.45M
        _timestamped_version_tracker.capture_expired_paths(
731
1.45M
                static_cast<int64_t>(expired_stale_sweep_endtime), &path_id_vec);
732
733
1.45M
        if (path_id_vec.empty()) {
734
1.44M
            return;
735
1.44M
        }
736
737
8.00k
        const RowsetSharedPtr lastest_delta = get_rowset_with_max_version();
738
8.00k
        if (lastest_delta == nullptr) {
739
0
            LOG(WARNING) << "lastest_delta is null " << tablet_id();
740
0
            return;
741
0
        }
742
743
        // fetch missing version before delete
744
8.00k
        Versions missed_versions = get_missed_versions_unlocked(lastest_delta->end_version());
745
8.00k
        if (!missed_versions.empty()) {
746
0
            LOG(WARNING) << "tablet:" << tablet_id()
747
0
                         << ", missed version for version:" << lastest_delta->end_version();
748
0
            _print_missed_versions(missed_versions);
749
0
            return;
750
0
        }
751
752
        // do check consistent operation
753
8.00k
        auto path_id_iter = path_id_vec.begin();
754
755
8.00k
        std::map<int64_t, PathVersionListSharedPtr> stale_version_path_map;
756
19.7k
        while (path_id_iter != path_id_vec.end()) {
757
11.7k
            PathVersionListSharedPtr version_path =
758
11.7k
                    _timestamped_version_tracker.fetch_and_delete_path_by_id(*path_id_iter);
759
760
11.7k
            Version test_version = Version(0, lastest_delta->end_version());
761
11.7k
            stale_version_path_map[*path_id_iter] = version_path;
762
763
11.7k
            Status status =
764
11.7k
                    capture_consistent_versions_unlocked(test_version, nullptr, false, false);
765
            // 1. When there is no consistent versions, we must reconstruct the tracker.
766
11.7k
            if (!status.ok()) {
767
                // 2. fetch missing version after delete
768
0
                Versions after_missed_versions =
769
0
                        get_missed_versions_unlocked(lastest_delta->end_version());
770
771
                // 2.1 check whether missed_versions and after_missed_versions are the same.
772
                // when they are the same, it means we can delete the path securely.
773
0
                bool is_missing = missed_versions.size() != after_missed_versions.size();
774
775
0
                if (!is_missing) {
776
0
                    for (int ver_index = 0; ver_index < missed_versions.size(); ver_index++) {
777
0
                        if (missed_versions[ver_index] != after_missed_versions[ver_index]) {
778
0
                            is_missing = true;
779
0
                            break;
780
0
                        }
781
0
                    }
782
0
                }
783
784
0
                if (is_missing) {
785
0
                    LOG(WARNING) << "The consistent version check fails, there are bugs. "
786
0
                                 << "Reconstruct the tracker to recover versions in tablet="
787
0
                                 << tablet_id();
788
789
                    // 3. try to recover
790
0
                    _timestamped_version_tracker.recover_versioned_tracker(stale_version_path_map);
791
792
                    // 4. double check the consistent versions
793
                    // fetch missing version after recover
794
0
                    Versions recover_missed_versions =
795
0
                            get_missed_versions_unlocked(lastest_delta->end_version());
796
797
                    // 4.1 check whether missed_versions and recover_missed_versions are the same.
798
                    // when they are the same, it means we recover successfully.
799
0
                    bool is_recover_missing =
800
0
                            missed_versions.size() != recover_missed_versions.size();
801
802
0
                    if (!is_recover_missing) {
803
0
                        for (int ver_index = 0; ver_index < missed_versions.size(); ver_index++) {
804
0
                            if (missed_versions[ver_index] != recover_missed_versions[ver_index]) {
805
0
                                is_recover_missing = true;
806
0
                                break;
807
0
                            }
808
0
                        }
809
0
                    }
810
811
                    // 5. check recover fail, version is mission
812
0
                    if (is_recover_missing) {
813
0
                        if (!config::ignore_rowset_stale_unconsistent_delete) {
814
0
                            LOG(FATAL)
815
0
                                    << "rowset stale unconsistent delete. tablet= " << tablet_id();
816
0
                        } else {
817
0
                            LOG(WARNING)
818
0
                                    << "rowset stale unconsistent delete. tablet= " << tablet_id();
819
0
                        }
820
0
                    }
821
0
                }
822
0
                return;
823
0
            }
824
11.7k
            path_id_iter++;
825
11.7k
        }
826
827
8.00k
        auto old_size = _stale_rs_version_map.size();
828
8.00k
        auto old_meta_size = _tablet_meta->all_stale_rs_metas().size();
829
830
        // do delete operation
831
8.00k
        auto to_delete_iter = stale_version_path_map.begin();
832
19.7k
        while (to_delete_iter != stale_version_path_map.end()) {
833
11.7k
            std::vector<TimestampedVersionSharedPtr>& to_delete_version =
834
11.7k
                    to_delete_iter->second->timestamped_versions();
835
11.7k
            int64_t start_version = -1;
836
11.7k
            int64_t end_version = -1;
837
11.7k
            std::vector<RowsetId> remove_rowset_ids;
838
106k
            for (auto& timestampedVersion : to_delete_version) {
839
106k
                auto it = _stale_rs_version_map.find(timestampedVersion->version());
840
106k
                if (it != _stale_rs_version_map.end()) {
841
106k
                    it->second->clear_cache();
842
                    // delete rowset
843
106k
                    if (it->second->is_local()) {
844
106k
                        _engine.add_unused_rowset(it->second);
845
106k
                        if (keys_type() == UNIQUE_KEYS && enable_unique_key_merge_on_write()) {
846
                            // mow does not support cold data in object storage
847
44.8k
                            remove_rowset_ids.emplace_back(it->second->rowset_id());
848
44.8k
                        }
849
106k
                    }
850
106k
                    _stale_rs_version_map.erase(it);
851
106k
                    VLOG_NOTICE << "delete stale rowset tablet=" << tablet_id() << " version["
852
0
                                << timestampedVersion->version().first << ","
853
0
                                << timestampedVersion->version().second
854
0
                                << "] move to unused_rowset success " << std::fixed
855
0
                                << expired_stale_sweep_endtime;
856
106k
                } else {
857
8
                    LOG(WARNING) << "delete stale rowset tablet=" << tablet_id() << " version["
858
8
                                 << timestampedVersion->version().first << ","
859
8
                                 << timestampedVersion->version().second
860
8
                                 << "] not find in stale rs version map";
861
8
                }
862
106k
                if (start_version < 0) {
863
11.7k
                    start_version = timestampedVersion->version().first;
864
11.7k
                }
865
106k
                end_version = timestampedVersion->version().second;
866
106k
                _delete_stale_rowset_by_version(timestampedVersion->version());
867
106k
            }
868
11.7k
            Version version(start_version, end_version);
869
11.7k
            to_delete_iter++;
870
11.7k
            if (!remove_rowset_ids.empty()) {
871
3.10k
                deleted_stale_rowsets.emplace_back(version, remove_rowset_ids);
872
3.10k
            }
873
11.7k
        }
874
875
8.00k
        bool reconstructed = _reconstruct_version_tracker_if_necessary();
876
877
8.00k
        VLOG_NOTICE << "delete stale rowset _stale_rs_version_map tablet=" << tablet_id()
878
0
                    << " current_size=" << _stale_rs_version_map.size() << " old_size=" << old_size
879
0
                    << " current_meta_size=" << _tablet_meta->all_stale_rs_metas().size()
880
0
                    << " old_meta_size=" << old_meta_size << " sweep endtime " << std::fixed
881
0
                    << expired_stale_sweep_endtime << ", reconstructed=" << reconstructed;
882
8.00k
    }
883
8.00k
    if (!deleted_stale_rowsets.empty()) {
884
        // agg delete bitmap for pre rowsets; record unused delete bitmap key ranges
885
458
        OlapStopWatch watch;
886
3.10k
        for (const auto& [version, remove_rowset_ids] : deleted_stale_rowsets) {
887
            // agg delete bitmap for pre rowset
888
3.10k
            DeleteBitmapKeyRanges remove_delete_bitmap_key_ranges;
889
3.10k
            agg_delete_bitmap_for_stale_rowsets(version, remove_delete_bitmap_key_ranges);
890
            // add remove delete bitmap
891
3.10k
            if (!remove_delete_bitmap_key_ranges.empty()) {
892
98
                _engine.add_unused_delete_bitmap_key_ranges(tablet_id(), remove_rowset_ids,
893
98
                                                            remove_delete_bitmap_key_ranges);
894
98
            }
895
3.10k
        }
896
458
        LOG(INFO) << "agg pre rowsets delete bitmap. tablet_id=" << tablet_id()
897
458
                  << ", size=" << deleted_stale_rowsets.size()
898
458
                  << ", cost(us)=" << watch.get_elapse_time_us();
899
458
    }
900
8.00k
#ifndef BE_TEST
901
8.00k
    {
902
8.00k
        std::shared_lock<std::shared_mutex> rlock(_meta_lock);
903
8.00k
        save_meta();
904
8.00k
    }
905
8.00k
#endif
906
8.00k
    if (config::enable_mow_verbose_log) {
907
0
        LOG_INFO("finish delete_expired_stale_rowset for tablet={}", tablet_id());
908
0
    }
909
8.00k
    DBUG_EXECUTE_IF("Tablet.delete_expired_stale_rowset.start_delete_unused_rowset",
910
8.00k
                    { _engine.start_delete_unused_rowset(); });
911
8.00k
}
912
913
Status Tablet::capture_consistent_versions_unlocked(const Version& spec_version,
914
                                                    Versions* version_path,
915
3.19M
                                                    bool skip_missing_version, bool quiet) const {
916
3.19M
    Status status =
917
3.19M
            _timestamped_version_tracker.capture_consistent_versions(spec_version, version_path);
918
3.19M
    if (!status.ok() && !quiet) {
919
1
        Versions missed_versions = get_missed_versions_unlocked(spec_version.second);
920
1
        if (missed_versions.empty()) {
921
            // if version_path is null, it may be a compaction check logic.
922
            // so to avoid print too many logs.
923
0
            if (version_path != nullptr) {
924
0
                LOG(WARNING) << "tablet:" << tablet_id()
925
0
                             << ", version already has been merged. spec_version: " << spec_version
926
0
                             << ", max_version: " << max_version_unlocked();
927
0
            }
928
0
            status = Status::Error<VERSION_ALREADY_MERGED, false>(
929
0
                    "versions are already compacted, spec_version "
930
0
                    "{}, max_version {}, tablet_id {}",
931
0
                    spec_version.second, max_version_unlocked(), tablet_id());
932
1
        } else {
933
1
            if (version_path != nullptr) {
934
1
                LOG(WARNING) << "status:" << status << ", tablet:" << tablet_id()
935
1
                             << ", missed version for version:" << spec_version;
936
1
                _print_missed_versions(missed_versions);
937
1
                if (skip_missing_version) {
938
0
                    LOG(WARNING) << "force skipping missing version for tablet:" << tablet_id();
939
0
                    return Status::OK();
940
0
                }
941
1
            }
942
1
        }
943
1
    }
944
945
3.19M
    DBUG_EXECUTE_IF("TTablet::capture_consistent_versions.inject_failure", {
946
3.19M
        auto tablet_id = dp->param<int64>("tablet_id", -1);
947
3.19M
        if (tablet_id != -1 && tablet_id == _tablet_meta->tablet_id()) {
948
3.19M
            status = Status::Error<VERSION_ALREADY_MERGED>("version already merged");
949
3.19M
        }
950
3.19M
    });
951
952
3.19M
    return status;
953
3.19M
}
954
955
9.06k
Status Tablet::check_version_integrity(const Version& version, bool quiet) {
956
9.06k
    std::shared_lock rdlock(_meta_lock);
957
9.06k
    return capture_consistent_versions_unlocked(version, nullptr, false, quiet);
958
9.06k
}
959
960
515k
bool Tablet::exceed_version_limit(int32_t limit) {
961
515k
    if (_tablet_meta->version_count() > limit) {
962
0
        exceed_version_limit_counter << 1;
963
0
        return true;
964
0
    }
965
515k
    return false;
966
515k
}
967
968
// If any rowset contains the specific version, it means the version already exist
969
1.94M
bool Tablet::check_version_exist(const Version& version) const {
970
1.94M
    std::shared_lock rdlock(_meta_lock);
971
7.71M
    for (auto& it : _rs_version_map) {
972
7.71M
        if (it.first.contains(version)) {
973
1.49M
            return true;
974
1.49M
        }
975
7.71M
    }
976
446k
    return false;
977
1.94M
}
978
979
// The meta read lock should be held before calling
980
void Tablet::acquire_version_and_rowsets(
981
269k
        std::vector<std::pair<Version, RowsetSharedPtr>>* version_rowsets) const {
982
426k
    for (const auto& it : _rs_version_map) {
983
426k
        version_rowsets->emplace_back(it.first, it.second);
984
426k
    }
985
269k
}
986
987
Status Tablet::capture_consistent_rowsets_unlocked(const Version& spec_version,
988
1.95k
                                                   std::vector<RowsetSharedPtr>* rowsets) const {
989
1.95k
    std::vector<Version> version_path;
990
1.95k
    RETURN_IF_ERROR(
991
1.95k
            capture_consistent_versions_unlocked(spec_version, &version_path, false, false));
992
1.95k
    RETURN_IF_ERROR(_capture_consistent_rowsets_unlocked(version_path, rowsets));
993
1.95k
    return Status::OK();
994
1.95k
}
995
996
Status Tablet::capture_rs_readers(const Version& spec_version, std::vector<RowSetSplits>* rs_splits,
997
3.17M
                                  bool skip_missing_version) {
998
3.17M
    std::shared_lock rlock(_meta_lock);
999
3.17M
    std::vector<Version> version_path;
1000
3.17M
    RETURN_IF_ERROR(capture_consistent_versions_unlocked(spec_version, &version_path,
1001
3.17M
                                                         skip_missing_version, false));
1002
3.17M
    RETURN_IF_ERROR(capture_rs_readers_unlocked(version_path, rs_splits));
1003
3.17M
    return Status::OK();
1004
3.17M
}
1005
1006
8.01k
Versions Tablet::calc_missed_versions(int64_t spec_version, Versions existing_versions) const {
1007
8.01k
    DCHECK(spec_version > 0) << "invalid spec_version: " << spec_version;
1008
1009
    // sort the existing versions in ascending order
1010
8.01k
    std::sort(existing_versions.begin(), existing_versions.end(),
1011
49.1k
              [](const Version& a, const Version& b) {
1012
                  // simple because 2 versions are certainly not overlapping
1013
49.1k
                  return a.first < b.first;
1014
49.1k
              });
1015
1016
    // From the first version(=0),  find the missing version until spec_version
1017
8.01k
    int64_t last_version = -1;
1018
8.01k
    Versions missed_versions;
1019
22.0k
    for (const Version& version : existing_versions) {
1020
22.0k
        if (version.first > last_version + 1) {
1021
4
            for (int64_t i = last_version + 1; i < version.first && i <= spec_version; ++i) {
1022
                // Don't merge missed_versions because clone & snapshot use single version.
1023
                // For example, if miss 4 ~ 6, clone need [4, 4], [5, 5], [6, 6], but not [4, 6].
1024
2
                missed_versions.emplace_back(i, i);
1025
2
            }
1026
2
        }
1027
22.0k
        last_version = version.second;
1028
22.0k
        if (last_version >= spec_version) {
1029
8.01k
            break;
1030
8.01k
        }
1031
22.0k
    }
1032
8.01k
    for (int64_t i = last_version + 1; i <= spec_version; ++i) {
1033
0
        missed_versions.emplace_back(i, i);
1034
0
    }
1035
1036
8.01k
    return missed_versions;
1037
8.01k
}
1038
1039
126M
bool Tablet::can_do_compaction(size_t path_hash, CompactionType compaction_type) {
1040
126M
    if (compaction_type == CompactionType::BASE_COMPACTION && tablet_state() != TABLET_RUNNING) {
1041
        // base compaction can only be done for tablet in TABLET_RUNNING state.
1042
        // but cumulative compaction can be done for TABLET_NOTREADY, such as tablet under alter process.
1043
34.3k
        return false;
1044
34.3k
    }
1045
1046
126M
    if (data_dir()->path_hash() != path_hash || !is_used() || !init_succeeded()) {
1047
46.3M
        return false;
1048
46.3M
    }
1049
1050
    // In TABLET_NOTREADY, we keep last 10 versions in new tablet so base tablet max_version
1051
    // not merged in new tablet and then we can do compaction
1052
80.3M
    return tablet_state() == TABLET_RUNNING || tablet_state() == TABLET_NOTREADY;
1053
126M
}
1054
1055
79.1M
uint32_t Tablet::calc_compaction_score() {
1056
79.1M
    if (_score_check_cnt++ % config::check_score_rounds_num != 0) {
1057
78.9M
        std::shared_lock rdlock(_meta_lock);
1058
78.9M
        if (_compaction_score > 0) {
1059
78.9M
            return _compaction_score;
1060
78.9M
        }
1061
78.9M
    }
1062
1063
240k
    {
1064
        // Need meta lock, because it will iterator "all_rs_metas" of tablet meta.
1065
240k
        std::shared_lock rdlock(_meta_lock);
1066
240k
        int32_t score = get_real_compaction_score();
1067
240k
        if (_compaction_score > 0 && _compaction_score != score) {
1068
259
            LOG(WARNING) << "cumu cache score not equal real score, cache score; "
1069
259
                         << _compaction_score << ", real score: " << score
1070
259
                         << ", tablet: " << tablet_id();
1071
259
        }
1072
240k
        _compaction_score = score;
1073
240k
        return score;
1074
79.1M
    }
1075
79.1M
}
1076
1077
bool Tablet::suitable_for_compaction(
1078
        CompactionType compaction_type,
1079
215k
        std::shared_ptr<CumulativeCompactionPolicy> cumulative_compaction_policy) {
1080
    // Need meta lock, because it will iterator "all_rs_metas" of tablet meta.
1081
215k
    std::shared_lock rdlock(_meta_lock);
1082
215k
    int32_t score = -1;
1083
215k
    if (compaction_type == CompactionType::CUMULATIVE_COMPACTION) {
1084
180k
        score = _calc_cumulative_compaction_score(cumulative_compaction_policy);
1085
180k
    } else {
1086
34.4k
        DCHECK_EQ(compaction_type, CompactionType::BASE_COMPACTION);
1087
34.4k
        score = _calc_base_compaction_score();
1088
34.4k
    }
1089
215k
    return score > 0;
1090
215k
}
1091
1092
74
uint32_t Tablet::calc_cold_data_compaction_score() const {
1093
74
    uint32_t score = 0;
1094
74
    std::vector<RowsetMetaSharedPtr> cooldowned_rowsets;
1095
74
    int64_t max_delete_version = 0;
1096
74
    {
1097
74
        std::shared_lock rlock(_meta_lock);
1098
151
        for (auto& rs_meta : _tablet_meta->all_rs_metas()) {
1099
151
            if (!rs_meta->is_local()) {
1100
150
                cooldowned_rowsets.push_back(rs_meta);
1101
150
                if (rs_meta->has_delete_predicate() &&
1102
150
                    rs_meta->end_version() > max_delete_version) {
1103
0
                    max_delete_version = rs_meta->end_version();
1104
0
                }
1105
150
            }
1106
151
        }
1107
74
    }
1108
150
    for (auto& rs_meta : cooldowned_rowsets) {
1109
150
        if (rs_meta->end_version() < max_delete_version) {
1110
0
            score += rs_meta->num_segments();
1111
150
        } else {
1112
150
            score += rs_meta->get_compaction_score();
1113
150
        }
1114
150
    }
1115
74
    return (keys_type() != KeysType::DUP_KEYS) ? score * 2 : score;
1116
74
}
1117
1118
uint32_t Tablet::_calc_cumulative_compaction_score(
1119
180k
        std::shared_ptr<CumulativeCompactionPolicy> cumulative_compaction_policy) {
1120
180k
    if (cumulative_compaction_policy == nullptr) [[unlikely]] {
1121
0
        return 0;
1122
0
    }
1123
180k
#ifndef BE_TEST
1124
180k
    if (_cumulative_compaction_policy == nullptr ||
1125
180k
        _cumulative_compaction_policy->name() != cumulative_compaction_policy->name()) {
1126
21.9k
        _cumulative_compaction_policy = cumulative_compaction_policy;
1127
21.9k
    }
1128
180k
#endif
1129
180k
    DBUG_EXECUTE_IF("Tablet._calc_cumulative_compaction_score.return", {
1130
180k
        LOG_WARNING("Tablet._calc_cumulative_compaction_score.return")
1131
180k
                .tag("tablet id", tablet_id());
1132
180k
        return 0;
1133
180k
    });
1134
180k
    return _cumulative_compaction_policy->calc_cumulative_compaction_score(this);
1135
180k
}
1136
1137
34.4k
uint32_t Tablet::_calc_base_compaction_score() const {
1138
34.4k
    uint32_t score = 0;
1139
34.4k
    const int64_t point = cumulative_layer_point();
1140
34.4k
    bool base_rowset_exist = false;
1141
34.4k
    bool has_delete = false;
1142
304k
    for (auto& rs_meta : _tablet_meta->all_rs_metas()) {
1143
304k
        if (rs_meta->start_version() == 0) {
1144
34.4k
            base_rowset_exist = true;
1145
34.4k
        }
1146
304k
        if (rs_meta->start_version() >= point || !rs_meta->is_local()) {
1147
            // all_rs_metas() is not sorted, so we use _continue_ other than _break_ here.
1148
279k
            continue;
1149
279k
        }
1150
24.5k
        if (rs_meta->has_delete_predicate()) {
1151
336
            has_delete = true;
1152
336
        }
1153
24.5k
        score += rs_meta->get_compaction_score();
1154
24.5k
    }
1155
1156
    // In the time series compaction policy, we want the base compaction to be triggered
1157
    // when there are delete versions present.
1158
34.4k
    if (_tablet_meta->compaction_policy() == CUMULATIVE_TIME_SERIES_POLICY) {
1159
0
        return (base_rowset_exist && has_delete) ? score : 0;
1160
0
    }
1161
1162
    // base不存在可能是tablet正在做alter table,先不选它,设score=0
1163
34.4k
    return base_rowset_exist ? score : 0;
1164
34.4k
}
1165
1166
9.54k
void Tablet::max_continuous_version_from_beginning(Version* version, Version* max_version) {
1167
9.54k
    bool has_version_cross;
1168
9.54k
    std::shared_lock rdlock(_meta_lock);
1169
9.54k
    _max_continuous_version_from_beginning_unlocked(version, max_version, &has_version_cross);
1170
9.54k
}
1171
1172
void Tablet::_max_continuous_version_from_beginning_unlocked(Version* version, Version* max_version,
1173
19.8M
                                                             bool* has_version_cross) const {
1174
19.8M
    std::vector<Version> existing_versions;
1175
19.8M
    *has_version_cross = false;
1176
34.3M
    for (auto& rs : _tablet_meta->all_rs_metas()) {
1177
34.3M
        existing_versions.emplace_back(rs->version());
1178
34.3M
    }
1179
1180
    // sort the existing versions in ascending order
1181
19.8M
    std::sort(existing_versions.begin(), existing_versions.end(),
1182
33.9M
              [](const Version& left, const Version& right) {
1183
                  // simple because 2 versions are certainly not overlapping
1184
33.9M
                  return left.first < right.first;
1185
33.9M
              });
1186
1187
19.8M
    Version max_continuous_version = {-1, -1};
1188
54.1M
    for (int i = 0; i < existing_versions.size(); ++i) {
1189
34.3M
        if (existing_versions[i].first > max_continuous_version.second + 1) {
1190
5
            break;
1191
34.3M
        } else if (existing_versions[i].first <= max_continuous_version.second) {
1192
0
            *has_version_cross = true;
1193
0
        }
1194
34.3M
        max_continuous_version = existing_versions[i];
1195
34.3M
    }
1196
19.8M
    *version = max_continuous_version;
1197
    // tablet may not has rowset, eg, tablet has just been clear for restore.
1198
19.8M
    if (max_version != nullptr && !existing_versions.empty()) {
1199
19.8M
        *max_version = existing_versions.back();
1200
19.8M
    }
1201
19.8M
}
1202
1203
47.9k
void Tablet::calculate_cumulative_point() {
1204
47.9k
    std::lock_guard<std::shared_mutex> wrlock(_meta_lock);
1205
47.9k
    SCOPED_SIMPLE_TRACE_IF_TIMEOUT(TRACE_TABLET_LOCK_THRESHOLD);
1206
47.9k
    int64_t ret_cumulative_point;
1207
47.9k
    _cumulative_compaction_policy->calculate_cumulative_point(
1208
47.9k
            this, _tablet_meta->all_rs_metas(), _cumulative_point, &ret_cumulative_point);
1209
1210
47.9k
    if (ret_cumulative_point == K_INVALID_CUMULATIVE_POINT) {
1211
46.5k
        return;
1212
46.5k
    }
1213
1.38k
    set_cumulative_layer_point(ret_cumulative_point);
1214
1.38k
}
1215
1216
// NOTE: only used when create_table, so it is sure that there is no concurrent reader and writer.
1217
0
void Tablet::delete_all_files() {
1218
    // Release resources like memory and disk space.
1219
0
    std::shared_lock rdlock(_meta_lock);
1220
0
    for (auto it : _rs_version_map) {
1221
0
        static_cast<void>(it.second->remove());
1222
0
    }
1223
0
    _rs_version_map.clear();
1224
1225
0
    for (auto it : _stale_rs_version_map) {
1226
0
        static_cast<void>(it.second->remove());
1227
0
    }
1228
0
    _stale_rs_version_map.clear();
1229
0
}
1230
1231
0
void Tablet::check_tablet_path_exists() {
1232
0
    if (!tablet_path().empty()) {
1233
0
        std::error_code ec;
1234
0
        if (std::filesystem::is_directory(tablet_path(), ec)) {
1235
0
            _is_tablet_path_exists.store(true, std::memory_order_relaxed);
1236
0
        } else if (ec.value() == ENOENT || ec.value() == 0) {
1237
0
            _is_tablet_path_exists.store(false, std::memory_order_relaxed);
1238
0
        }
1239
0
    }
1240
0
}
1241
1242
515k
Status Tablet::_contains_version(const Version& version) {
1243
    // check if there exist a rowset contains the added rowset
1244
2.64M
    for (auto& it : _rs_version_map) {
1245
2.64M
        if (it.first.contains(version)) {
1246
            // TODO(lingbin): Is this check unnecessary?
1247
            // because the value type is std::shared_ptr, when will it be nullptr?
1248
            // In addition, in this class, there are many places that do not make this judgment
1249
            // when access _rs_version_map's value.
1250
292
            CHECK(it.second != nullptr) << "there exist a version=" << it.first
1251
0
                                        << " contains the input rs with version=" << version
1252
0
                                        << ", but the related rs is null";
1253
292
            return Status::Error<PUSH_VERSION_ALREADY_EXIST>("Tablet push duplicate version {}",
1254
292
                                                             version.to_string());
1255
292
        }
1256
2.64M
    }
1257
1258
515k
    return Status::OK();
1259
515k
}
1260
1261
47.9k
std::vector<RowsetSharedPtr> Tablet::pick_candidate_rowsets_to_cumulative_compaction() {
1262
47.9k
    std::vector<RowsetSharedPtr> candidate_rowsets;
1263
47.9k
    if (_cumulative_point == K_INVALID_CUMULATIVE_POINT) {
1264
1
        return candidate_rowsets;
1265
1
    }
1266
47.9k
    return _pick_visible_rowsets_to_compaction(_cumulative_point,
1267
47.9k
                                               std::numeric_limits<int64_t>::max());
1268
47.9k
}
1269
1270
5.78k
std::vector<RowsetSharedPtr> Tablet::pick_candidate_rowsets_to_base_compaction() {
1271
5.78k
    return _pick_visible_rowsets_to_compaction(std::numeric_limits<int64_t>::min(),
1272
5.78k
                                               _cumulative_point - 1);
1273
5.78k
}
1274
1275
std::vector<RowsetSharedPtr> Tablet::_pick_visible_rowsets_to_compaction(
1276
53.7k
        int64_t min_start_version, int64_t max_start_version) {
1277
53.7k
    auto [visible_version, update_ts] = get_visible_version_and_time();
1278
53.7k
    bool update_time_long = MonotonicMillis() - update_ts >
1279
53.7k
                            config::compaction_keep_invisible_version_timeout_sec * 1000L;
1280
53.7k
    int32_t keep_invisible_version_limit =
1281
53.7k
            update_time_long ? config::compaction_keep_invisible_version_min_count
1282
53.7k
                             : config::compaction_keep_invisible_version_max_count;
1283
1284
53.7k
    std::vector<RowsetSharedPtr> candidate_rowsets;
1285
53.7k
    {
1286
53.7k
        std::shared_lock rlock(_meta_lock);
1287
470k
        for (const auto& [version, rs] : _rs_version_map) {
1288
470k
            int64_t version_start = version.first;
1289
            // rowset is remote or rowset is not in given range
1290
470k
            if (!rs->is_local() || version_start < min_start_version ||
1291
470k
                version_start > max_start_version) {
1292
193k
                continue;
1293
193k
            }
1294
1295
            // can compact, met one of the conditions:
1296
            // 1. had been visible;
1297
            // 2. exceeds the limit of keep invisible versions.
1298
277k
            int64_t version_end = version.second;
1299
277k
            if (version_end <= visible_version ||
1300
277k
                version_end > visible_version + keep_invisible_version_limit) {
1301
268k
                candidate_rowsets.push_back(rs);
1302
268k
            }
1303
277k
        }
1304
53.7k
    }
1305
53.7k
    std::sort(candidate_rowsets.begin(), candidate_rowsets.end(), Rowset::comparator);
1306
53.7k
    return candidate_rowsets;
1307
53.7k
}
1308
1309
155
std::vector<RowsetSharedPtr> Tablet::pick_candidate_rowsets_to_full_compaction() {
1310
155
    std::vector<RowsetSharedPtr> candidate_rowsets;
1311
628
    traverse_rowsets([&candidate_rowsets](const auto& rs) {
1312
        // Do full compaction on all local rowsets.
1313
628
        if (rs->is_local()) {
1314
628
            candidate_rowsets.emplace_back(rs);
1315
628
        }
1316
628
    });
1317
155
    std::sort(candidate_rowsets.begin(), candidate_rowsets.end(), Rowset::comparator);
1318
155
    return candidate_rowsets;
1319
155
}
1320
1321
std::vector<RowsetSharedPtr> Tablet::pick_candidate_rowsets_to_build_inverted_index(
1322
1.34k
        const std::set<int64_t>& alter_index_uids, bool is_drop_op) {
1323
1.34k
    std::vector<RowsetSharedPtr> candidate_rowsets;
1324
1.34k
    {
1325
1.34k
        std::shared_lock rlock(_meta_lock);
1326
6.50k
        auto has_alter_inverted_index = [&](RowsetSharedPtr rowset) -> bool {
1327
6.50k
            for (const auto& index_id : alter_index_uids) {
1328
6.50k
                if (rowset->tablet_schema()->has_inverted_index_with_index_id(index_id)) {
1329
3.47k
                    return true;
1330
3.47k
                }
1331
6.50k
            }
1332
3.03k
            return false;
1333
6.50k
        };
1334
1335
3.35k
        for (const auto& [version, rs] : _rs_version_map) {
1336
3.35k
            if (!has_alter_inverted_index(rs) && is_drop_op) {
1337
198
                continue;
1338
198
            }
1339
3.15k
            if (has_alter_inverted_index(rs) && !is_drop_op) {
1340
72
                continue;
1341
72
            }
1342
1343
3.08k
            if (rs->is_local()) {
1344
3.08k
                candidate_rowsets.push_back(rs);
1345
3.08k
            }
1346
3.08k
        }
1347
1.34k
    }
1348
1.34k
    std::sort(candidate_rowsets.begin(), candidate_rowsets.end(), Rowset::comparator);
1349
1.34k
    return candidate_rowsets;
1350
1.34k
}
1351
1352
19.8M
std::tuple<int64_t, int64_t> Tablet::get_visible_version_and_time() const {
1353
    // some old tablet has bug, its partition_id is 0, fe couldn't update its visible version.
1354
    // so let this tablet's visible version become int64 max.
1355
19.8M
    auto version_info = std::atomic_load_explicit(&_visible_version, std::memory_order_relaxed);
1356
19.8M
    if (version_info != nullptr && partition_id() != 0) {
1357
19.8M
        return std::make_tuple(version_info->version.load(std::memory_order_relaxed),
1358
19.8M
                               version_info->update_ts);
1359
19.8M
    } else {
1360
34
        return std::make_tuple(std::numeric_limits<int64_t>::max(),
1361
34
                               std::numeric_limits<int64_t>::max());
1362
34
    }
1363
19.8M
}
1364
1365
// For http compaction action
1366
2.31k
void Tablet::get_compaction_status(std::string* json_result) {
1367
2.31k
    rapidjson::Document root;
1368
2.31k
    root.SetObject();
1369
1370
2.31k
    rapidjson::Document path_arr;
1371
2.31k
    path_arr.SetArray();
1372
1373
2.31k
    std::vector<RowsetSharedPtr> rowsets;
1374
2.31k
    std::vector<RowsetSharedPtr> stale_rowsets;
1375
2.31k
    std::vector<bool> delete_flags;
1376
2.31k
    {
1377
2.31k
        std::shared_lock rdlock(_meta_lock);
1378
2.31k
        rowsets.reserve(_rs_version_map.size());
1379
10.8k
        for (auto& it : _rs_version_map) {
1380
10.8k
            rowsets.push_back(it.second);
1381
10.8k
        }
1382
2.31k
        std::sort(rowsets.begin(), rowsets.end(), Rowset::comparator);
1383
1384
2.31k
        stale_rowsets.reserve(_stale_rs_version_map.size());
1385
7.12k
        for (auto& it : _stale_rs_version_map) {
1386
7.12k
            stale_rowsets.push_back(it.second);
1387
7.12k
        }
1388
2.31k
        std::sort(stale_rowsets.begin(), stale_rowsets.end(), Rowset::comparator);
1389
1390
2.31k
        delete_flags.reserve(rowsets.size());
1391
10.8k
        for (auto& rs : rowsets) {
1392
10.8k
            delete_flags.push_back(rs->rowset_meta()->has_delete_predicate());
1393
10.8k
        }
1394
        // get snapshot version path json_doc
1395
2.31k
        _timestamped_version_tracker.get_stale_version_path_json_doc(path_arr);
1396
2.31k
    }
1397
2.31k
    rapidjson::Value cumulative_policy_type;
1398
2.31k
    std::string policy_type_str = "cumulative compaction policy not initializied";
1399
2.31k
    if (_cumulative_compaction_policy != nullptr) {
1400
754
        policy_type_str = _cumulative_compaction_policy->name();
1401
754
    }
1402
2.31k
    cumulative_policy_type.SetString(policy_type_str.c_str(), policy_type_str.length(),
1403
2.31k
                                     root.GetAllocator());
1404
2.31k
    root.AddMember("cumulative policy type", cumulative_policy_type, root.GetAllocator());
1405
2.31k
    root.AddMember("cumulative point", _cumulative_point.load(), root.GetAllocator());
1406
1407
2.31k
#define FORMAT_UNIXMILLIS_ADD_JSON_NODE(root, key, unixmillis_value)                   \
1408
20.8k
    {                                                                                  \
1409
20.8k
        rapidjson::Value value;                                                        \
1410
20.8k
        std::string format_str = ToStringFromUnixMillis(unixmillis_value.load());      \
1411
20.8k
        value.SetString(format_str.c_str(), format_str.length(), root.GetAllocator()); \
1412
20.8k
        root.AddMember(key, value, root.GetAllocator());                               \
1413
20.8k
    }
1414
2.31k
#define FORMAT_STRING_ADD_JSON_NODE(root, key, str_value)                            \
1415
6.94k
    {                                                                                \
1416
6.94k
        rapidjson::Value value;                                                      \
1417
6.94k
        value.SetString(str_value.c_str(), str_value.length(), root.GetAllocator()); \
1418
6.94k
        root.AddMember(key, value, root.GetAllocator());                             \
1419
6.94k
    }
1420
1421
2.31k
    FORMAT_UNIXMILLIS_ADD_JSON_NODE(root, "last cumulative failure time",
1422
2.31k
                                    _last_cumu_compaction_failure_millis)
1423
2.31k
    FORMAT_UNIXMILLIS_ADD_JSON_NODE(root, "last base failure time",
1424
2.31k
                                    _last_base_compaction_failure_millis)
1425
2.31k
    FORMAT_UNIXMILLIS_ADD_JSON_NODE(root, "last full failure time",
1426
2.31k
                                    _last_full_compaction_failure_millis)
1427
2.31k
    FORMAT_UNIXMILLIS_ADD_JSON_NODE(root, "last cumulative success time",
1428
2.31k
                                    _last_cumu_compaction_success_millis)
1429
2.31k
    FORMAT_UNIXMILLIS_ADD_JSON_NODE(root, "last base success time",
1430
2.31k
                                    _last_base_compaction_success_millis)
1431
2.31k
    FORMAT_UNIXMILLIS_ADD_JSON_NODE(root, "last full success time",
1432
2.31k
                                    _last_full_compaction_success_millis)
1433
2.31k
    FORMAT_UNIXMILLIS_ADD_JSON_NODE(root, "last cumulative schedule time",
1434
2.31k
                                    _last_cumu_compaction_schedule_millis)
1435
2.31k
    FORMAT_UNIXMILLIS_ADD_JSON_NODE(root, "last base schedule time",
1436
2.31k
                                    _last_base_compaction_schedule_millis)
1437
2.31k
    FORMAT_UNIXMILLIS_ADD_JSON_NODE(root, "last full schedule time",
1438
2.31k
                                    _last_full_compaction_schedule_millis)
1439
2.31k
    FORMAT_STRING_ADD_JSON_NODE(root, "last cumulative status", _last_cumu_compaction_status)
1440
2.31k
    FORMAT_STRING_ADD_JSON_NODE(root, "last base status", _last_base_compaction_status)
1441
2.31k
    FORMAT_STRING_ADD_JSON_NODE(root, "last full status", _last_full_compaction_status)
1442
1443
    // last single replica compaction status
1444
    // "single replica compaction status": {
1445
    //     "remote peer": "172.100.1.0:10875",
1446
    //     "last failure status": "",
1447
    //     "last fetched rowset": "[8-10]"
1448
    // }
1449
2.31k
    rapidjson::Document status;
1450
2.31k
    status.SetObject();
1451
2.31k
    TReplicaInfo replica_info;
1452
2.31k
    std::string dummp_token;
1453
2.31k
    if (tablet_meta()->tablet_schema()->enable_single_replica_compaction() &&
1454
2.31k
        _engine.get_peer_replica_info(tablet_id(), &replica_info, &dummp_token)) {
1455
        // remote peer
1456
0
        rapidjson::Value peer_addr;
1457
0
        std::string addr = replica_info.host + ":" + std::to_string(replica_info.brpc_port);
1458
0
        peer_addr.SetString(addr.c_str(), addr.length(), status.GetAllocator());
1459
0
        status.AddMember("remote peer", peer_addr, status.GetAllocator());
1460
        // last failure status
1461
0
        rapidjson::Value compaction_status;
1462
0
        compaction_status.SetString(_last_single_compaction_failure_status.c_str(),
1463
0
                                    _last_single_compaction_failure_status.length(),
1464
0
                                    status.GetAllocator());
1465
0
        status.AddMember("last failure status", compaction_status, status.GetAllocator());
1466
        // last fetched rowset
1467
0
        rapidjson::Value version;
1468
0
        std::string fetched_version = _last_fetched_version.to_string();
1469
0
        version.SetString(fetched_version.c_str(), fetched_version.length(), status.GetAllocator());
1470
0
        status.AddMember("last fetched rowset", version, status.GetAllocator());
1471
0
        root.AddMember("single replica compaction status", status, root.GetAllocator());
1472
0
    }
1473
1474
    // print all rowsets' version as an array
1475
2.31k
    rapidjson::Document versions_arr;
1476
2.31k
    rapidjson::Document missing_versions_arr;
1477
2.31k
    versions_arr.SetArray();
1478
2.31k
    missing_versions_arr.SetArray();
1479
2.31k
    int64_t last_version = -1;
1480
10.8k
    for (auto& rowset : rowsets) {
1481
10.8k
        const Version& ver = rowset->version();
1482
10.8k
        if (ver.first != last_version + 1) {
1483
0
            rapidjson::Value miss_value;
1484
0
            miss_value.SetString(fmt::format("[{}-{}]", last_version + 1, ver.first - 1).c_str(),
1485
0
                                 missing_versions_arr.GetAllocator());
1486
0
            missing_versions_arr.PushBack(miss_value, missing_versions_arr.GetAllocator());
1487
0
        }
1488
10.8k
        rapidjson::Value value;
1489
10.8k
        std::string version_str = rowset->get_rowset_info_str();
1490
10.8k
        value.SetString(version_str.c_str(), version_str.length(), versions_arr.GetAllocator());
1491
10.8k
        versions_arr.PushBack(value, versions_arr.GetAllocator());
1492
10.8k
        last_version = ver.second;
1493
10.8k
    }
1494
2.31k
    root.AddMember("rowsets", versions_arr, root.GetAllocator());
1495
2.31k
    root.AddMember("missing_rowsets", missing_versions_arr, root.GetAllocator());
1496
1497
    // print all stale rowsets' version as an array
1498
2.31k
    rapidjson::Document stale_versions_arr;
1499
2.31k
    stale_versions_arr.SetArray();
1500
7.12k
    for (auto& rowset : stale_rowsets) {
1501
7.12k
        rapidjson::Value value;
1502
7.12k
        std::string version_str = rowset->get_rowset_info_str();
1503
7.12k
        value.SetString(version_str.c_str(), version_str.length(),
1504
7.12k
                        stale_versions_arr.GetAllocator());
1505
7.12k
        stale_versions_arr.PushBack(value, stale_versions_arr.GetAllocator());
1506
7.12k
    }
1507
2.31k
    root.AddMember("stale_rowsets", stale_versions_arr, root.GetAllocator());
1508
1509
    // add stale version rowsets
1510
2.31k
    root.AddMember("stale version path", path_arr, root.GetAllocator());
1511
1512
    // to json string
1513
2.31k
    rapidjson::StringBuffer strbuf;
1514
2.31k
    rapidjson::PrettyWriter<rapidjson::StringBuffer> writer(strbuf);
1515
2.31k
    root.Accept(writer);
1516
2.31k
    *json_result = std::string(strbuf.GetString());
1517
2.31k
}
1518
1519
666k
bool Tablet::do_tablet_meta_checkpoint() {
1520
666k
    std::lock_guard<std::shared_mutex> store_lock(_meta_store_lock);
1521
666k
    if (_newly_created_rowset_num == 0) {
1522
449k
        return false;
1523
449k
    }
1524
216k
    if (UnixMillis() - _last_checkpoint_time <
1525
216k
                config::tablet_meta_checkpoint_min_interval_secs * 1000 &&
1526
216k
        _newly_created_rowset_num < config::tablet_meta_checkpoint_min_new_rowsets_num) {
1527
2.23k
        return false;
1528
2.23k
    }
1529
    // hold read-lock other than write-lock, because it will not modify meta structure
1530
214k
    std::shared_lock rdlock(_meta_lock);
1531
214k
    if (tablet_state() != TABLET_RUNNING) {
1532
2.04k
        LOG(INFO) << "tablet is under state=" << tablet_state()
1533
2.04k
                  << ", not running, skip do checkpoint"
1534
2.04k
                  << ", tablet=" << tablet_id();
1535
2.04k
        return false;
1536
2.04k
    }
1537
212k
    VLOG_NOTICE << "start to do tablet meta checkpoint, tablet=" << tablet_id();
1538
212k
    save_meta();
1539
    // if save meta successfully, then should remove the rowset meta existing in tablet
1540
    // meta from rowset meta store
1541
377k
    for (auto& rs_meta : _tablet_meta->all_rs_metas()) {
1542
        // If we delete it from rowset manager's meta explicitly in previous checkpoint, just skip.
1543
377k
        if (rs_meta->is_remove_from_rowset_meta()) {
1544
6.06k
            continue;
1545
6.06k
        }
1546
371k
        if (RowsetMetaManager::check_rowset_meta(_data_dir->get_meta(), tablet_uid(),
1547
371k
                                                 rs_meta->rowset_id())) {
1548
370k
            RETURN_FALSE_IF_ERROR(RowsetMetaManager::remove(_data_dir->get_meta(), tablet_uid(),
1549
370k
                                                            rs_meta->rowset_id()));
1550
370k
            VLOG_NOTICE << "remove rowset id from meta store because it is already persistent with "
1551
0
                        << "tablet meta, rowset_id=" << rs_meta->rowset_id();
1552
370k
        }
1553
371k
        rs_meta->set_remove_from_rowset_meta();
1554
371k
    }
1555
1556
    // check _stale_rs_version_map to remove meta from rowset meta store
1557
212k
    for (auto& rs_meta : _tablet_meta->all_stale_rs_metas()) {
1558
        // If we delete it from rowset manager's meta explicitly in previous checkpoint, just skip.
1559
114k
        if (rs_meta->is_remove_from_rowset_meta()) {
1560
14.3k
            continue;
1561
14.3k
        }
1562
99.7k
        if (RowsetMetaManager::check_rowset_meta(_data_dir->get_meta(), tablet_uid(),
1563
99.7k
                                                 rs_meta->rowset_id())) {
1564
99.6k
            RETURN_FALSE_IF_ERROR(RowsetMetaManager::remove(_data_dir->get_meta(), tablet_uid(),
1565
99.6k
                                                            rs_meta->rowset_id()));
1566
99.6k
            VLOG_NOTICE << "remove rowset id from meta store because it is already persistent with "
1567
0
                        << "tablet meta, rowset_id=" << rs_meta->rowset_id();
1568
99.6k
        }
1569
99.7k
        rs_meta->set_remove_from_rowset_meta();
1570
99.7k
    }
1571
1572
212k
    if (keys_type() == UNIQUE_KEYS && enable_unique_key_merge_on_write()) {
1573
39.9k
        RETURN_FALSE_IF_ERROR(TabletMetaManager::remove_old_version_delete_bitmap(
1574
39.9k
                _data_dir, tablet_id(), max_version_unlocked()));
1575
39.9k
    }
1576
1577
212k
    _newly_created_rowset_num = 0;
1578
212k
    _last_checkpoint_time = UnixMillis();
1579
212k
    return true;
1580
212k
}
1581
1582
294k
bool Tablet::rowset_meta_is_useful(RowsetMetaSharedPtr rowset_meta) {
1583
294k
    std::shared_lock rdlock(_meta_lock);
1584
294k
    bool find_version = false;
1585
1.29M
    for (auto& version_rowset : _rs_version_map) {
1586
1.29M
        if (version_rowset.second->rowset_id() == rowset_meta->rowset_id()) {
1587
170k
            return true;
1588
170k
        }
1589
1.12M
        if (version_rowset.second->contains_version(rowset_meta->version())) {
1590
123k
            find_version = true;
1591
123k
        }
1592
1.12M
    }
1593
29.4M
    for (auto& stale_version_rowset : _stale_rs_version_map) {
1594
29.4M
        if (stale_version_rowset.second->rowset_id() == rowset_meta->rowset_id()) {
1595
117k
            return true;
1596
117k
        }
1597
29.3M
        if (stale_version_rowset.second->contains_version(rowset_meta->version())) {
1598
466k
            find_version = true;
1599
466k
        }
1600
29.3M
    }
1601
5.42k
    return !find_version;
1602
123k
}
1603
1604
515k
bool Tablet::_contains_rowset(const RowsetId rowset_id) {
1605
2.64M
    for (auto& version_rowset : _rs_version_map) {
1606
2.64M
        if (version_rowset.second->rowset_id() == rowset_id) {
1607
8
            return true;
1608
8
        }
1609
2.64M
    }
1610
32.4M
    for (auto& stale_version_rowset : _stale_rs_version_map) {
1611
32.4M
        if (stale_version_rowset.second->rowset_id() == rowset_id) {
1612
4
            return true;
1613
4
        }
1614
32.4M
    }
1615
515k
    return false;
1616
515k
}
1617
1618
// need check if consecutive version missing in full report
1619
// alter tablet will ignore this check
1620
void Tablet::build_tablet_report_info(TTabletInfo* tablet_info,
1621
                                      bool enable_consecutive_missing_check,
1622
19.8M
                                      bool enable_path_check) {
1623
19.8M
    std::shared_lock rdlock(_meta_lock);
1624
19.8M
    tablet_info->__set_tablet_id(_tablet_meta->tablet_id());
1625
19.8M
    tablet_info->__set_schema_hash(_tablet_meta->schema_hash());
1626
19.8M
    tablet_info->__set_row_count(_tablet_meta->num_rows());
1627
19.8M
    tablet_info->__set_data_size(_tablet_meta->tablet_local_size());
1628
1629
    // Here we need to report to FE if there are any missing versions of tablet.
1630
    // We start from the initial version and traverse backwards until we meet a discontinuous version.
1631
19.8M
    Version cversion;
1632
19.8M
    Version max_version;
1633
19.8M
    bool has_version_cross;
1634
19.8M
    _max_continuous_version_from_beginning_unlocked(&cversion, &max_version, &has_version_cross);
1635
    // cause publish version task runs concurrently, version may be flying
1636
    // so we add a consecutive miss check to solve this problem:
1637
    // if publish version 5 arrives but version 4 flying, we may judge replica miss version
1638
    // and set version miss in tablet_info, which makes fe treat this replica as unhealth
1639
    // and lead to other problems
1640
19.8M
    if (enable_consecutive_missing_check) {
1641
19.8M
        if (cversion.second < max_version.second) {
1642
17
            if (_last_missed_version == cversion.second + 1) {
1643
4
                if (MonotonicSeconds() - _last_missed_time_s >= 60) {
1644
                    // version missed for over 60 seconds
1645
0
                    tablet_info->__set_version_miss(true);
1646
0
                    _last_missed_version = -1;
1647
0
                    _last_missed_time_s = 0;
1648
0
                }
1649
13
            } else {
1650
13
                _last_missed_version = cversion.second + 1;
1651
13
                _last_missed_time_s = MonotonicSeconds();
1652
13
            }
1653
17
        }
1654
19.8M
    } else {
1655
12.3k
        tablet_info->__set_version_miss(cversion.second < max_version.second);
1656
12.3k
    }
1657
1658
19.8M
    DBUG_EXECUTE_IF("Tablet.build_tablet_report_info.version_miss", {
1659
19.8M
        auto tablet_id = dp->param<int64>("tablet_id", -1);
1660
19.8M
        if (tablet_id != -1 && tablet_id == _tablet_meta->tablet_id()) {
1661
19.8M
            auto miss = dp->param<bool>("version_miss", true);
1662
19.8M
            tablet_info->__set_version_miss(miss);
1663
19.8M
        }
1664
19.8M
    });
1665
1666
    // find rowset with max version
1667
19.8M
    auto iter = _rs_version_map.find(max_version);
1668
19.8M
    if (iter == _rs_version_map.end()) {
1669
        // If the tablet is in running state, it must not be doing schema-change. so if we can not
1670
        // access its rowsets, it means that the tablet is bad and needs to be reported to the FE
1671
        // for subsequent repairs (through the cloning task)
1672
12
        if (tablet_state() == TABLET_RUNNING) {
1673
0
            tablet_info->__set_used(false);
1674
0
        }
1675
        // For other states, FE knows that the tablet is in a certain change process, so here
1676
        // still sets the state to normal when reporting. Note that every task has an timeout,
1677
        // so if the task corresponding to this change hangs, when the task timeout, FE will know
1678
        // and perform state modification operations.
1679
12
    }
1680
1681
19.8M
    if (tablet_state() == TABLET_RUNNING) {
1682
19.8M
        if (has_version_cross || is_io_error_too_times() || !data_dir()->is_used()) {
1683
0
            LOG(INFO) << "report " << tablet_id() << " as bad, version_cross=" << has_version_cross
1684
0
                      << ", ioe times=" << get_io_error_times() << ", data_dir used "
1685
0
                      << data_dir()->is_used();
1686
0
            tablet_info->__set_used(false);
1687
0
        }
1688
1689
19.8M
        if (enable_path_check) {
1690
19.8M
            if (!_is_tablet_path_exists.exchange(true, std::memory_order_relaxed)) {
1691
0
                LOG(INFO) << "report " << tablet_id() << " as bad, tablet directory not found";
1692
0
                tablet_info->__set_used(false);
1693
0
            }
1694
19.8M
        }
1695
19.8M
    }
1696
1697
    // There are two cases when tablet state is TABLET_NOTREADY
1698
    // case 1: tablet is doing schema change. Fe knows it's state, doing nothing.
1699
    // case 2: tablet has finished schema change, but failed. Fe will perform recovery.
1700
19.8M
    if (tablet_state() == TABLET_NOTREADY && is_alter_failed()) {
1701
854
        tablet_info->__set_used(false);
1702
854
    }
1703
1704
19.8M
    if (tablet_state() == TABLET_SHUTDOWN) {
1705
10
        tablet_info->__set_used(false);
1706
10
    }
1707
1708
19.8M
    DBUG_EXECUTE_IF("Tablet.build_tablet_report_info.used", {
1709
19.8M
        auto tablet_id = dp->param<int64>("tablet_id", -1);
1710
19.8M
        if (tablet_id != -1 && tablet_id == _tablet_meta->tablet_id()) {
1711
19.8M
            auto used = dp->param<bool>("used", true);
1712
19.8M
            LOG_WARNING("Tablet.build_tablet_report_info.used")
1713
19.8M
                    .tag("tablet id", tablet_id)
1714
19.8M
                    .tag("used", used);
1715
19.8M
            tablet_info->__set_used(used);
1716
19.8M
        } else {
1717
19.8M
            LOG_WARNING("Tablet.build_tablet_report_info.used").tag("tablet id", tablet_id);
1718
19.8M
        }
1719
19.8M
    });
1720
1721
19.8M
    int64_t total_version_count = _tablet_meta->version_count();
1722
1723
    // For compatibility.
1724
    // For old fe, it wouldn't send visible version request to be, then be's visible version is always 0.
1725
    // Let visible_version_count set to total_version_count in be's report.
1726
19.8M
    int64_t visible_version_count = total_version_count;
1727
19.8M
    if (auto [visible_version, _] = get_visible_version_and_time(); visible_version > 0) {
1728
16.5M
        visible_version_count = _tablet_meta->version_count_cross_with_range({0, visible_version});
1729
16.5M
    }
1730
    // the report version is the largest continuous version, same logic as in FE side
1731
19.8M
    tablet_info->__set_version(cversion.second);
1732
    // Useless but it is a required filed in TTabletInfo
1733
19.8M
    tablet_info->__set_version_hash(0);
1734
19.8M
    tablet_info->__set_partition_id(_tablet_meta->partition_id());
1735
19.8M
    tablet_info->__set_storage_medium(_data_dir->storage_medium());
1736
19.8M
    tablet_info->__set_total_version_count(total_version_count);
1737
19.8M
    tablet_info->__set_visible_version_count(visible_version_count);
1738
19.8M
    tablet_info->__set_path_hash(_data_dir->path_hash());
1739
19.8M
    tablet_info->__set_is_in_memory(_tablet_meta->tablet_schema()->is_in_memory());
1740
19.8M
    tablet_info->__set_replica_id(replica_id());
1741
19.8M
    tablet_info->__set_remote_data_size(_tablet_meta->tablet_remote_size());
1742
19.8M
    if (_tablet_meta->cooldown_meta_id().initialized()) { // has cooldowned data
1743
236
        tablet_info->__set_cooldown_term(_cooldown_conf.term);
1744
236
        tablet_info->__set_cooldown_meta_id(_tablet_meta->cooldown_meta_id().to_thrift());
1745
236
    }
1746
19.8M
    if (tablet_state() == TABLET_RUNNING && _tablet_meta->storage_policy_id() > 0) {
1747
        // tablet may not have cooldowned data, but the storage policy is set
1748
32.8k
        tablet_info->__set_cooldown_term(_cooldown_conf.term);
1749
32.8k
    }
1750
19.8M
    tablet_info->__set_local_index_size(_tablet_meta->tablet_local_index_size());
1751
19.8M
    tablet_info->__set_local_segment_size(_tablet_meta->tablet_local_segment_size());
1752
19.8M
    tablet_info->__set_remote_index_size(_tablet_meta->tablet_remote_index_size());
1753
19.8M
    tablet_info->__set_remote_segment_size(_tablet_meta->tablet_remote_segment_size());
1754
19.8M
}
1755
1756
560
void Tablet::report_error(const Status& st) {
1757
560
    if (st.is<ErrorCode::IO_ERROR>()) {
1758
0
        ++_io_error_times;
1759
560
    } else if (st.is<ErrorCode::CORRUPTION>()) {
1760
1
        _io_error_times = config::max_tablet_io_errors + 1;
1761
559
    } else if (st.is<ErrorCode::NOT_FOUND>()) {
1762
0
        check_tablet_path_exists();
1763
0
        if (!_is_tablet_path_exists.load(std::memory_order_relaxed)) {
1764
0
            _io_error_times = config::max_tablet_io_errors + 1;
1765
0
        }
1766
0
    }
1767
560
}
1768
1769
Status Tablet::prepare_compaction_and_calculate_permits(
1770
        CompactionType compaction_type, const TabletSharedPtr& tablet,
1771
53.4k
        std::shared_ptr<CompactionMixin>& compaction, int64_t& permits) {
1772
53.4k
    if (compaction_type == CompactionType::CUMULATIVE_COMPACTION) {
1773
47.5k
        MonotonicStopWatch watch;
1774
47.5k
        watch.start();
1775
1776
47.5k
        compaction = std::make_shared<CumulativeCompaction>(tablet->_engine, tablet);
1777
47.5k
        DorisMetrics::instance()->cumulative_compaction_request_total->increment(1);
1778
47.5k
        Status res = compaction->prepare_compact();
1779
47.5k
        if (!config::disable_compaction_trace_log &&
1780
47.5k
            watch.elapsed_time() / 1e9 > config::cumulative_compaction_trace_threshold) {
1781
0
            std::stringstream ss;
1782
0
            compaction->runtime_profile()->pretty_print(&ss);
1783
0
            LOG(WARNING) << "prepare cumulative compaction cost " << watch.elapsed_time() / 1e9
1784
0
                         << std::endl
1785
0
                         << ss.str();
1786
0
        }
1787
1788
47.5k
        if (!res.ok()) {
1789
33.2k
            permits = 0;
1790
            // if we meet a delete version, should increase the cumulative point to let base compaction handle the delete version.
1791
            // no need to wait 5s.
1792
33.2k
            if (!res.is<ErrorCode::CUMULATIVE_MEET_DELETE_VERSION>() ||
1793
33.2k
                config::enable_sleep_between_delete_cumu_compaction) {
1794
33.1k
                tablet->set_last_cumu_compaction_failure_time(UnixMillis());
1795
33.1k
            }
1796
33.2k
            if (!res.is<CUMULATIVE_NO_SUITABLE_VERSION>() &&
1797
33.2k
                !res.is<ErrorCode::CUMULATIVE_MEET_DELETE_VERSION>()) {
1798
0
                DorisMetrics::instance()->cumulative_compaction_request_failed->increment(1);
1799
0
                return Status::InternalError("prepare cumulative compaction with err: {}",
1800
0
                                             res.to_string());
1801
0
            }
1802
            // return OK if OLAP_ERR_CUMULATIVE_NO_SUITABLE_VERSION, so that we don't need to
1803
            // print too much useless logs.
1804
            // And because we set permits to 0, so even if we return OK here, nothing will be done.
1805
33.2k
            return Status::OK();
1806
33.2k
        }
1807
47.5k
    } else if (compaction_type == CompactionType::BASE_COMPACTION) {
1808
5.78k
        MonotonicStopWatch watch;
1809
5.78k
        watch.start();
1810
1811
5.78k
        compaction = std::make_shared<BaseCompaction>(tablet->_engine, tablet);
1812
5.78k
        DorisMetrics::instance()->base_compaction_request_total->increment(1);
1813
5.78k
        Status res = compaction->prepare_compact();
1814
5.78k
        if (!config::disable_compaction_trace_log &&
1815
5.78k
            watch.elapsed_time() / 1e9 > config::base_compaction_trace_threshold) {
1816
0
            std::stringstream ss;
1817
0
            compaction->runtime_profile()->pretty_print(&ss);
1818
0
            LOG(WARNING) << "prepare base compaction cost " << watch.elapsed_time() / 1e9
1819
0
                         << std::endl
1820
0
                         << ss.str();
1821
0
        }
1822
1823
5.78k
        tablet->set_last_base_compaction_status(res.to_string());
1824
5.78k
        if (!res.ok()) {
1825
5.69k
            tablet->set_last_base_compaction_failure_time(UnixMillis());
1826
5.69k
            permits = 0;
1827
5.69k
            if (!res.is<BE_NO_SUITABLE_VERSION>()) {
1828
0
                DorisMetrics::instance()->base_compaction_request_failed->increment(1);
1829
0
                return Status::InternalError("prepare base compaction with err: {}",
1830
0
                                             res.to_string());
1831
0
            }
1832
            // return OK if OLAP_ERR_BE_NO_SUITABLE_VERSION, so that we don't need to
1833
            // print too much useless logs.
1834
            // And because we set permits to 0, so even if we return OK here, nothing will be done.
1835
5.69k
            return Status::OK();
1836
5.69k
        }
1837
5.78k
    } else {
1838
112
        DCHECK_EQ(compaction_type, CompactionType::FULL_COMPACTION);
1839
1840
112
        compaction = std::make_shared<FullCompaction>(tablet->_engine, tablet);
1841
112
        Status res = compaction->prepare_compact();
1842
112
        if (!res.ok()) {
1843
102
            tablet->set_last_full_compaction_failure_time(UnixMillis());
1844
102
            permits = 0;
1845
102
            if (!res.is<FULL_NO_SUITABLE_VERSION>()) {
1846
0
                return Status::InternalError("prepare full compaction with err: {}",
1847
0
                                             res.to_string());
1848
0
            }
1849
            // return OK if OLAP_ERR_BE_NO_SUITABLE_VERSION, so that we don't need to
1850
            // print too much useless logs.
1851
            // And because we set permits to 0, so even if we return OK here, nothing will be done.
1852
102
            return Status::OK();
1853
102
        }
1854
112
    }
1855
1856
    // Time series policy does not rely on permits, it uses goal size to control memory
1857
14.4k
    if (tablet->tablet_meta()->compaction_policy() == CUMULATIVE_TIME_SERIES_POLICY) {
1858
        // permits = 0 means that prepare_compaction failed
1859
0
        permits = 1;
1860
14.4k
    } else {
1861
14.4k
        permits = compaction->get_compaction_permits();
1862
14.4k
    }
1863
14.4k
    return Status::OK();
1864
53.4k
}
1865
1866
0
void Tablet::execute_single_replica_compaction(SingleReplicaCompaction& compaction) {
1867
0
    Status res = compaction.execute_compact();
1868
0
    if (!res.ok()) {
1869
0
        set_last_failure_time(this, compaction, UnixMillis());
1870
0
        set_last_single_compaction_failure_status(res.to_string());
1871
0
        if (res.is<CANCELLED>()) {
1872
0
            DorisMetrics::instance()->single_compaction_request_cancelled->increment(1);
1873
            // "CANCELLED" indicates that the peer has not performed compaction,
1874
            // wait for the peer to perform compaction
1875
0
            set_skip_compaction(true, compaction.real_compact_type(), UnixSeconds());
1876
0
            VLOG_CRITICAL << "Cannel fetching from the remote peer. res=" << res
1877
0
                          << ", tablet=" << tablet_id();
1878
0
        } else {
1879
0
            DorisMetrics::instance()->single_compaction_request_failed->increment(1);
1880
0
            LOG(WARNING) << "failed to do single replica compaction. res=" << res
1881
0
                         << ", tablet=" << tablet_id();
1882
0
        }
1883
0
        return;
1884
0
    }
1885
0
    set_last_failure_time(this, compaction, 0);
1886
0
}
1887
1888
158M
bool Tablet::should_fetch_from_peer() {
1889
158M
    return tablet_meta()->tablet_schema()->enable_single_replica_compaction() &&
1890
158M
           _engine.should_fetch_from_peer(tablet_id());
1891
158M
}
1892
1893
3
std::vector<Version> Tablet::get_all_local_versions() {
1894
3
    std::vector<Version> local_versions;
1895
3
    {
1896
3
        std::shared_lock rlock(_meta_lock);
1897
63
        for (const auto& [version, rs] : _rs_version_map) {
1898
63
            if (rs->is_local()) {
1899
53
                local_versions.emplace_back(version);
1900
53
            }
1901
63
        }
1902
3
    }
1903
3
    std::sort(local_versions.begin(), local_versions.end(),
1904
318
              [](const Version& left, const Version& right) { return left.first < right.first; });
1905
3
    return local_versions;
1906
3
}
1907
1908
14.4k
void Tablet::execute_compaction(CompactionMixin& compaction) {
1909
14.4k
    signal::tablet_id = tablet_id();
1910
1911
14.4k
    MonotonicStopWatch watch;
1912
14.4k
    watch.start();
1913
1914
14.4k
    Status res = [&]() { RETURN_IF_CATCH_EXCEPTION({ return compaction.execute_compact(); }); }();
1915
1916
14.4k
    if (!res.ok()) [[unlikely]] {
1917
15
        set_last_failure_time(this, compaction, UnixMillis());
1918
15
        LOG(WARNING) << "failed to do " << compaction.compaction_name()
1919
15
                     << ", tablet=" << tablet_id() << " : " << res;
1920
14.4k
    } else {
1921
14.4k
        set_last_failure_time(this, compaction, 0);
1922
14.4k
    }
1923
1924
14.4k
    if (!config::disable_compaction_trace_log) {
1925
0
        auto need_trace = [&compaction, &watch] {
1926
0
            return compaction.compaction_type() == ReaderType::READER_CUMULATIVE_COMPACTION
1927
0
                           ? watch.elapsed_time() / 1e9 >
1928
0
                                     config::cumulative_compaction_trace_threshold
1929
0
                   : compaction.compaction_type() == ReaderType::READER_BASE_COMPACTION
1930
0
                           ? watch.elapsed_time() / 1e9 > config::base_compaction_trace_threshold
1931
0
                           : false;
1932
0
        };
1933
0
        if (need_trace()) {
1934
0
            std::stringstream ss;
1935
0
            compaction.runtime_profile()->pretty_print(&ss);
1936
0
            LOG(WARNING) << "execute " << compaction.compaction_name() << " cost "
1937
0
                         << watch.elapsed_time() / 1e9 << std::endl
1938
0
                         << ss.str();
1939
0
        }
1940
0
    }
1941
14.4k
}
1942
1943
228k
Status Tablet::create_initial_rowset(const int64_t req_version) {
1944
228k
    if (req_version < 1) {
1945
0
        return Status::Error<CE_CMD_PARAMS_ERROR>(
1946
0
                "init version of tablet should at least 1. req.ver={}", req_version);
1947
0
    }
1948
228k
    Version version(0, req_version);
1949
228k
    RowsetSharedPtr new_rowset;
1950
    // there is no data in init rowset, so overlapping info is unknown.
1951
228k
    RowsetWriterContext context;
1952
228k
    context.version = version;
1953
228k
    context.rowset_state = VISIBLE;
1954
228k
    context.segments_overlap = OVERLAP_UNKNOWN;
1955
228k
    context.tablet_schema = tablet_schema();
1956
228k
    context.newest_write_timestamp = UnixSeconds();
1957
228k
    auto rs_writer = DORIS_TRY(create_rowset_writer(context, false));
1958
228k
    RETURN_IF_ERROR(rs_writer->flush());
1959
228k
    RETURN_IF_ERROR(rs_writer->build(new_rowset));
1960
228k
    RETURN_IF_ERROR(add_rowset(std::move(new_rowset)));
1961
228k
    set_cumulative_layer_point(req_version + 1);
1962
228k
    return Status::OK();
1963
228k
}
1964
1965
Result<std::unique_ptr<RowsetWriter>> Tablet::create_rowset_writer(RowsetWriterContext& context,
1966
526k
                                                                   bool vertical) {
1967
526k
    context.rowset_id = _engine.next_rowset_id();
1968
526k
    _init_context_common_fields(context);
1969
526k
    return RowsetFactory::create_rowset_writer(_engine, context, vertical);
1970
526k
}
1971
1972
// create a rowset writer with rowset_id and seg_id
1973
// after writer, merge this transient rowset with original rowset
1974
Result<std::unique_ptr<RowsetWriter>> Tablet::create_transient_rowset_writer(
1975
        const Rowset& rowset, std::shared_ptr<PartialUpdateInfo> partial_update_info,
1976
2.53k
        int64_t txn_expiration) {
1977
2.53k
    RowsetWriterContext context;
1978
2.53k
    context.rowset_state = PREPARED;
1979
2.53k
    context.segments_overlap = OVERLAPPING;
1980
2.53k
    context.tablet_schema = std::make_shared<TabletSchema>();
1981
    // During a partial update, the extracted columns of a variant should not be included in the tablet schema.
1982
    // This is because the partial update for a variant needs to ignore the extracted columns.
1983
    // Otherwise, the schema types in different rowsets might be inconsistent. When performing a partial update,
1984
    // the complete variant is constructed by reading all the sub-columns of the variant.
1985
2.53k
    context.tablet_schema = rowset.tablet_schema()->copy_without_variant_extracted_columns();
1986
2.53k
    context.newest_write_timestamp = UnixSeconds();
1987
2.53k
    context.tablet_id = table_id();
1988
2.53k
    context.enable_segcompaction = false;
1989
    // ATTN: context.tablet is a shared_ptr, can't simply set it's value to `this`. We should
1990
    // get the shared_ptr from tablet_manager.
1991
2.53k
    auto tablet = _engine.tablet_manager()->get_tablet(tablet_id());
1992
2.53k
    if (!tablet) {
1993
0
        LOG(WARNING) << "cant find tablet by tablet_id=" << tablet_id();
1994
0
        return ResultError(Status::NotFound("cant find tablet by tablet_id={}", tablet_id()));
1995
0
    }
1996
2.53k
    context.tablet = tablet;
1997
2.53k
    context.write_type = DataWriteType::TYPE_DIRECT;
1998
2.53k
    context.partial_update_info = std::move(partial_update_info);
1999
2.53k
    context.is_transient_rowset_writer = true;
2000
2.53k
    return create_transient_rowset_writer(context, rowset.rowset_id())
2001
2.53k
            .transform([&](auto&& writer) {
2002
2.53k
                writer->set_segment_start_id(rowset.num_segments());
2003
2.53k
                return writer;
2004
2.53k
            });
2005
2.53k
}
2006
2007
Result<std::unique_ptr<RowsetWriter>> Tablet::create_transient_rowset_writer(
2008
2.53k
        RowsetWriterContext& context, const RowsetId& rowset_id) {
2009
2.53k
    context.rowset_id = rowset_id;
2010
2.53k
    _init_context_common_fields(context);
2011
2.53k
    return RowsetFactory::create_rowset_writer(_engine, context, false);
2012
2.53k
}
2013
2014
528k
void Tablet::_init_context_common_fields(RowsetWriterContext& context) {
2015
528k
    context.tablet_uid = tablet_uid();
2016
528k
    context.tablet_id = tablet_id();
2017
528k
    context.partition_id = partition_id();
2018
528k
    context.tablet_schema_hash = schema_hash();
2019
528k
    context.rowset_type = tablet_meta()->preferred_rowset_type();
2020
    // Alpha Rowset will be removed in the future, so that if the tablet's default rowset type is
2021
    // alpha rowset, then set the newly created rowset to storage engine's default rowset.
2022
528k
    if (context.rowset_type == ALPHA_ROWSET) {
2023
0
        context.rowset_type = _engine.default_rowset_type();
2024
0
    }
2025
2026
529k
    if (context.is_local_rowset()) {
2027
529k
        context.tablet_path = _tablet_path;
2028
529k
    }
2029
2030
528k
    context.data_dir = data_dir();
2031
528k
    context.enable_unique_key_merge_on_write = enable_unique_key_merge_on_write();
2032
528k
}
2033
2034
5.45k
Status Tablet::create_rowset(const RowsetMetaSharedPtr& rowset_meta, RowsetSharedPtr* rowset) {
2035
5.45k
    return RowsetFactory::create_rowset(_tablet_meta->tablet_schema(),
2036
5.45k
                                        rowset_meta->is_local() ? _tablet_path : "", rowset_meta,
2037
5.45k
                                        rowset);
2038
5.45k
}
2039
2040
25
Status Tablet::cooldown(RowsetSharedPtr rowset) {
2041
25
    std::unique_lock schema_change_lock(_schema_change_lock, std::try_to_lock);
2042
25
    if (!schema_change_lock.owns_lock()) {
2043
0
        return Status::Error<TRY_LOCK_FAILED>(
2044
0
                "try schema_change_lock failed, schema change running or inverted index built on "
2045
0
                "this tablet={}",
2046
0
                tablet_id());
2047
0
    }
2048
    // Check executing serially with compaction task.
2049
25
    std::unique_lock base_compaction_lock(_base_compaction_lock, std::try_to_lock);
2050
25
    if (!base_compaction_lock.owns_lock()) {
2051
0
        return Status::Error<TRY_LOCK_FAILED>("try base_compaction_lock failed");
2052
0
    }
2053
25
    std::unique_lock cumu_compaction_lock(_cumulative_compaction_lock, std::try_to_lock);
2054
25
    if (!cumu_compaction_lock.owns_lock()) {
2055
0
        return Status::Error<TRY_LOCK_FAILED>("try cumu_compaction_lock failed");
2056
0
    }
2057
25
    std::shared_lock cooldown_conf_rlock(_cooldown_conf_lock);
2058
25
    if (_cooldown_conf.cooldown_replica_id <= 0) { // wait for FE to push cooldown conf
2059
12
        return Status::InternalError("invalid cooldown_replica_id");
2060
12
    }
2061
2062
13
    auto mem_tracker = MemTrackerLimiter::create_shared(
2063
13
            MemTrackerLimiter::Type::OTHER,
2064
13
            fmt::format("Tablet::cooldown#tableId={}:replicaId={}", std::to_string(tablet_id()),
2065
13
                        std::to_string(replica_id())));
2066
13
    SCOPED_ATTACH_TASK(mem_tracker);
2067
2068
13
    if (_cooldown_conf.cooldown_replica_id == replica_id()) {
2069
        // this replica is cooldown replica
2070
13
        RETURN_IF_ERROR(_cooldown_data(std::move(rowset)));
2071
13
    } else {
2072
0
        Status st = _follow_cooldowned_data();
2073
0
        if (UNLIKELY(!st.ok())) {
2074
0
            _last_failed_follow_cooldown_time = time(nullptr);
2075
0
            return st;
2076
0
        }
2077
0
        _last_failed_follow_cooldown_time = 0;
2078
0
    }
2079
13
    return Status::OK();
2080
13
}
2081
2082
// hold SHARED `cooldown_conf_lock`
2083
13
Status Tablet::_cooldown_data(RowsetSharedPtr rowset) {
2084
13
    DCHECK(_cooldown_conf.cooldown_replica_id == replica_id());
2085
2086
13
    auto storage_resource = DORIS_TRY(get_resource_by_storage_policy_id(storage_policy_id()));
2087
13
    RowsetSharedPtr old_rowset = nullptr;
2088
2089
13
    if (rowset) {
2090
0
        const auto& rowset_id = rowset->rowset_id();
2091
0
        const auto& rowset_version = rowset->version();
2092
0
        std::shared_lock meta_rlock(_meta_lock);
2093
0
        auto iter = _rs_version_map.find(rowset_version);
2094
0
        if (iter != _rs_version_map.end() && iter->second->rowset_id() == rowset_id) {
2095
0
            old_rowset = rowset;
2096
0
        }
2097
0
    }
2098
2099
13
    if (!old_rowset) {
2100
13
        old_rowset = pick_cooldown_rowset();
2101
13
    }
2102
2103
13
    if (!old_rowset) {
2104
0
        LOG(INFO) << "cannot pick cooldown rowset in tablet " << tablet_id();
2105
0
        return Status::OK();
2106
0
    }
2107
2108
13
    RowsetId new_rowset_id = _engine.next_rowset_id();
2109
13
    auto pending_rs_guard = _engine.pending_remote_rowsets().add(new_rowset_id);
2110
13
    Status st;
2111
13
    Defer defer {[&] {
2112
13
        if (!st.ok()) {
2113
            // reclaim the incomplete rowset data in remote storage
2114
0
            record_unused_remote_rowset(new_rowset_id, storage_resource.fs->id(),
2115
0
                                        old_rowset->num_segments());
2116
0
        }
2117
13
    }};
2118
13
    auto start = std::chrono::steady_clock::now();
2119
13
    if (st = old_rowset->upload_to(storage_resource, new_rowset_id); !st.ok()) {
2120
0
        return st;
2121
0
    }
2122
2123
13
    auto duration = std::chrono::duration<float>(std::chrono::steady_clock::now() - start);
2124
13
    LOG(INFO) << "Upload rowset " << old_rowset->version() << " " << new_rowset_id.to_string()
2125
13
              << " to " << storage_resource.fs->root_path().native()
2126
13
              << ", tablet_id=" << tablet_id() << ", duration=" << duration.count()
2127
13
              << ", capacity=" << old_rowset->total_disk_size()
2128
13
              << ", tp=" << old_rowset->total_disk_size() / duration.count()
2129
13
              << ", old rowset_id=" << old_rowset->rowset_id().to_string();
2130
2131
    // gen a new rowset
2132
13
    auto new_rowset_meta = std::make_shared<RowsetMeta>();
2133
13
    new_rowset_meta->init(old_rowset->rowset_meta().get());
2134
13
    new_rowset_meta->set_rowset_id(new_rowset_id);
2135
13
    new_rowset_meta->set_remote_storage_resource(std::move(storage_resource));
2136
13
    new_rowset_meta->set_creation_time(time(nullptr));
2137
13
    UniqueId cooldown_meta_id = UniqueId::gen_uid();
2138
13
    RowsetSharedPtr new_rowset;
2139
13
    RETURN_IF_ERROR(RowsetFactory::create_rowset(_tablet_meta->tablet_schema(), "", new_rowset_meta,
2140
13
                                                 &new_rowset));
2141
2142
13
    {
2143
13
        std::unique_lock meta_wlock(_meta_lock);
2144
13
        SCOPED_SIMPLE_TRACE_IF_TIMEOUT(TRACE_TABLET_LOCK_THRESHOLD);
2145
13
        if (tablet_state() == TABLET_RUNNING) {
2146
13
            RETURN_IF_ERROR(delete_rowsets({std::move(old_rowset)}, false));
2147
13
            add_rowsets({std::move(new_rowset)});
2148
            // TODO(plat1ko): process primary key
2149
13
            _tablet_meta->set_cooldown_meta_id(cooldown_meta_id);
2150
13
        }
2151
13
    }
2152
13
    {
2153
13
        std::shared_lock meta_rlock(_meta_lock);
2154
13
        SCOPED_SIMPLE_TRACE_IF_TIMEOUT(TRACE_TABLET_LOCK_THRESHOLD);
2155
13
        save_meta();
2156
13
    }
2157
    // Upload cooldowned rowset meta to remote fs
2158
    // ATTN: Even if it is an empty rowset, in order for the followers to synchronize, the coolown meta must be
2159
    // uploaded, otherwise followers may never completely cooldown.
2160
13
    if (auto t = _engine.tablet_manager()->get_tablet(tablet_id());
2161
13
        t != nullptr) { // `t` can be nullptr if it has been dropped
2162
13
        async_write_cooldown_meta(std::move(t));
2163
13
    }
2164
13
    return Status::OK();
2165
13
}
2166
2167
// hold SHARED `cooldown_conf_lock`
2168
Status Tablet::_read_cooldown_meta(const StorageResource& storage_resource,
2169
0
                                   TabletMetaPB* tablet_meta_pb) {
2170
0
    std::string remote_meta_path = storage_resource.cooldown_tablet_meta_path(
2171
0
            tablet_id(), _cooldown_conf.cooldown_replica_id, _cooldown_conf.term);
2172
0
    io::FileReaderSPtr tablet_meta_reader;
2173
0
    RETURN_IF_ERROR(storage_resource.fs->open_file(remote_meta_path, &tablet_meta_reader));
2174
0
    auto file_size = tablet_meta_reader->size();
2175
0
    size_t bytes_read;
2176
0
    auto buf = std::unique_ptr<uint8_t[]>(new uint8_t[file_size]);
2177
0
    RETURN_IF_ERROR(tablet_meta_reader->read_at(0, {buf.get(), file_size}, &bytes_read));
2178
0
    RETURN_IF_ERROR(tablet_meta_reader->close());
2179
0
    if (!tablet_meta_pb->ParseFromArray(buf.get(), file_size)) {
2180
0
        return Status::InternalError("malformed tablet meta, path={}/{}",
2181
0
                                     storage_resource.fs->root_path().native(), remote_meta_path);
2182
0
    }
2183
0
    return Status::OK();
2184
0
}
2185
2186
// `rs_metas` MUST already be sorted by `RowsetMeta::comparator`
2187
14
Status check_version_continuity(const std::vector<RowsetMetaSharedPtr>& rs_metas) {
2188
14
    if (rs_metas.size() < 2) {
2189
5
        return Status::OK();
2190
5
    }
2191
9
    auto prev = rs_metas.begin();
2192
29
    for (auto it = rs_metas.begin() + 1; it != rs_metas.end(); ++it) {
2193
20
        if ((*prev)->end_version() + 1 != (*it)->start_version()) {
2194
0
            return Status::InternalError("versions are not continuity: prev={} cur={}",
2195
0
                                         (*prev)->version().to_string(),
2196
0
                                         (*it)->version().to_string());
2197
0
        }
2198
20
        prev = it;
2199
20
    }
2200
9
    return Status::OK();
2201
9
}
2202
2203
// It's guaranteed the write cooldown meta task would be invoked at the end unless BE crashes
2204
// one tablet would at most have one async task to be done
2205
14
void Tablet::async_write_cooldown_meta(TabletSharedPtr tablet) {
2206
14
    ExecEnv::GetInstance()->write_cooldown_meta_executors()->submit(std::move(tablet));
2207
14
}
2208
2209
166
bool Tablet::update_cooldown_conf(int64_t cooldown_term, int64_t cooldown_replica_id) {
2210
166
    std::unique_lock wlock(_cooldown_conf_lock, std::try_to_lock);
2211
166
    if (!wlock.owns_lock()) {
2212
0
        LOG(INFO) << "try cooldown_conf_lock failed, tablet_id=" << tablet_id();
2213
0
        return false;
2214
0
    }
2215
166
    if (cooldown_term <= _cooldown_conf.term) {
2216
0
        return false;
2217
0
    }
2218
166
    LOG(INFO) << "update cooldown conf. tablet_id=" << tablet_id()
2219
166
              << " cooldown_replica_id: " << _cooldown_conf.cooldown_replica_id << " -> "
2220
166
              << cooldown_replica_id << ", cooldown_term: " << _cooldown_conf.term << " -> "
2221
166
              << cooldown_term;
2222
166
    _cooldown_conf.cooldown_replica_id = cooldown_replica_id;
2223
166
    _cooldown_conf.term = cooldown_term;
2224
166
    return true;
2225
166
}
2226
2227
14
Status Tablet::write_cooldown_meta() {
2228
14
    std::shared_lock rlock(_cooldown_conf_lock);
2229
14
    if (_cooldown_conf.cooldown_replica_id != _tablet_meta->replica_id()) {
2230
0
        return Status::Aborted<false>("not cooldown replica({} vs {}) tablet_id={}",
2231
0
                                      _tablet_meta->replica_id(),
2232
0
                                      _cooldown_conf.cooldown_replica_id, tablet_id());
2233
0
    }
2234
2235
14
    auto storage_resource = DORIS_TRY(get_resource_by_storage_policy_id(storage_policy_id()));
2236
2237
14
    std::vector<RowsetMetaSharedPtr> cooldowned_rs_metas;
2238
14
    UniqueId cooldown_meta_id;
2239
14
    {
2240
14
        std::shared_lock meta_rlock(_meta_lock);
2241
52
        for (auto& rs_meta : _tablet_meta->all_rs_metas()) {
2242
52
            if (!rs_meta->is_local()) {
2243
34
                cooldowned_rs_metas.push_back(rs_meta);
2244
34
            }
2245
52
        }
2246
14
        cooldown_meta_id = _tablet_meta->cooldown_meta_id();
2247
14
    }
2248
14
    if (cooldowned_rs_metas.empty()) {
2249
0
        LOG(INFO) << "no cooldown meta to write, tablet_id=" << tablet_id();
2250
0
        return Status::OK();
2251
0
    }
2252
14
    std::sort(cooldowned_rs_metas.begin(), cooldowned_rs_metas.end(), RowsetMeta::comparator);
2253
14
    DCHECK(cooldowned_rs_metas.front()->start_version() == 0);
2254
    // If version not continuous, it must be a bug
2255
14
    if (auto st = check_version_continuity(cooldowned_rs_metas); !st.ok()) {
2256
0
        DCHECK(st.ok()) << st << " tablet_id=" << tablet_id();
2257
0
        st.set_code(ABORTED);
2258
0
        return st;
2259
0
    }
2260
2261
14
    TabletMetaPB tablet_meta_pb;
2262
14
    auto* rs_metas = tablet_meta_pb.mutable_rs_metas();
2263
14
    rs_metas->Reserve(cooldowned_rs_metas.size());
2264
34
    for (auto& rs_meta : cooldowned_rs_metas) {
2265
34
        rs_metas->Add(rs_meta->get_rowset_pb());
2266
34
    }
2267
14
    tablet_meta_pb.mutable_cooldown_meta_id()->set_hi(cooldown_meta_id.hi);
2268
14
    tablet_meta_pb.mutable_cooldown_meta_id()->set_lo(cooldown_meta_id.lo);
2269
2270
14
    std::string remote_meta_path = storage_resource.cooldown_tablet_meta_path(
2271
14
            tablet_id(), _cooldown_conf.cooldown_replica_id, _cooldown_conf.term);
2272
14
    io::FileWriterPtr tablet_meta_writer;
2273
    // FIXME(plat1ko): What if object store permanently unavailable?
2274
14
    RETURN_IF_ERROR(storage_resource.fs->create_file(remote_meta_path, &tablet_meta_writer));
2275
14
    auto val = tablet_meta_pb.SerializeAsString();
2276
14
    RETURN_IF_ERROR(tablet_meta_writer->append({val.data(), val.size()}));
2277
14
    return tablet_meta_writer->close();
2278
14
}
2279
2280
// hold SHARED `cooldown_conf_lock`
2281
0
Status Tablet::_follow_cooldowned_data() {
2282
0
    DCHECK(_cooldown_conf.cooldown_replica_id != replica_id());
2283
0
    LOG(INFO) << "try to follow cooldowned data. tablet_id=" << tablet_id()
2284
0
              << " cooldown_replica_id=" << _cooldown_conf.cooldown_replica_id
2285
0
              << " local replica=" << replica_id();
2286
2287
0
    auto storage_resource = DORIS_TRY(get_resource_by_storage_policy_id(storage_policy_id()));
2288
    // MUST executing serially with cold data compaction, because compaction input rowsets may be deleted by this function
2289
0
    std::unique_lock cold_compaction_lock(_cold_compaction_lock, std::try_to_lock);
2290
0
    if (!cold_compaction_lock.owns_lock()) {
2291
0
        return Status::Error<TRY_LOCK_FAILED>("try cold_compaction_lock failed");
2292
0
    }
2293
2294
0
    TabletMetaPB cooldown_meta_pb;
2295
0
    auto st = _read_cooldown_meta(storage_resource, &cooldown_meta_pb);
2296
0
    if (!st.ok()) {
2297
0
        LOG(INFO) << "cannot read cooldown meta: " << st;
2298
0
        return Status::InternalError<false>("cannot read cooldown meta");
2299
0
    }
2300
0
    DCHECK(cooldown_meta_pb.rs_metas_size() > 0);
2301
0
    if (_tablet_meta->cooldown_meta_id() == cooldown_meta_pb.cooldown_meta_id()) {
2302
        // cooldowned rowsets are same, no need to follow
2303
0
        return Status::OK();
2304
0
    }
2305
2306
0
    int64_t cooldowned_version = cooldown_meta_pb.rs_metas().rbegin()->end_version();
2307
2308
0
    std::vector<RowsetSharedPtr> overlap_rowsets;
2309
0
    bool version_aligned = false;
2310
2311
    // Holding these to delete rowsets' shared ptr until save meta can avoid trash sweeping thread
2312
    // deleting these rowsets' files before rowset meta has been removed from disk, which may cause
2313
    // data loss when BE reboot before save meta to disk.
2314
0
    std::vector<RowsetSharedPtr> to_delete;
2315
0
    std::vector<RowsetSharedPtr> to_add;
2316
2317
0
    {
2318
0
        std::lock_guard wlock(_meta_lock);
2319
0
        SCOPED_SIMPLE_TRACE_IF_TIMEOUT(TRACE_TABLET_LOCK_THRESHOLD);
2320
0
        if (tablet_state() != TABLET_RUNNING) {
2321
0
            return Status::InternalError<false>("tablet not running");
2322
0
        }
2323
2324
0
        for (auto& [v, rs] : _rs_version_map) {
2325
0
            if (v.second <= cooldowned_version) {
2326
0
                overlap_rowsets.push_back(rs);
2327
0
                if (!version_aligned && v.second == cooldowned_version) {
2328
0
                    version_aligned = true;
2329
0
                }
2330
0
            } else if (!rs->is_local()) {
2331
0
                return Status::InternalError<false>(
2332
0
                        "cooldowned version larger than that to follow with cooldown version {}",
2333
0
                        cooldowned_version);
2334
0
            }
2335
0
        }
2336
2337
0
        if (!version_aligned) {
2338
0
            return Status::InternalError<false>("cooldowned version is not aligned with version {}",
2339
0
                                                cooldowned_version);
2340
0
        }
2341
2342
0
        std::sort(overlap_rowsets.begin(), overlap_rowsets.end(), Rowset::comparator);
2343
2344
        // Find different rowset in `overlap_rowsets` and `cooldown_meta_pb.rs_metas`
2345
0
        auto rs_pb_it = cooldown_meta_pb.rs_metas().begin();
2346
0
        auto rs_it = overlap_rowsets.begin();
2347
0
        for (; rs_pb_it != cooldown_meta_pb.rs_metas().end() && rs_it != overlap_rowsets.end();
2348
0
             ++rs_pb_it, ++rs_it) {
2349
0
            if (rs_pb_it->rowset_id_v2() != (*rs_it)->rowset_id().to_string()) {
2350
0
                break;
2351
0
            }
2352
0
        }
2353
2354
0
        to_delete.assign(rs_it, overlap_rowsets.end());
2355
0
        to_add.reserve(cooldown_meta_pb.rs_metas().end() - rs_pb_it);
2356
0
        for (; rs_pb_it != cooldown_meta_pb.rs_metas().end(); ++rs_pb_it) {
2357
0
            auto rs_meta = std::make_shared<RowsetMeta>();
2358
0
            rs_meta->init_from_pb(*rs_pb_it);
2359
0
            RowsetSharedPtr rs;
2360
0
            RETURN_IF_ERROR(
2361
0
                    RowsetFactory::create_rowset(_tablet_meta->tablet_schema(), "", rs_meta, &rs));
2362
0
            to_add.push_back(std::move(rs));
2363
0
        }
2364
        // Note: We CANNOT call `modify_rowsets` here because `modify_rowsets` cannot process version graph correctly.
2365
0
        RETURN_IF_ERROR(delete_rowsets(to_delete, false));
2366
0
        add_rowsets(to_add);
2367
        // TODO(plat1ko): process primary key
2368
0
        _tablet_meta->set_cooldown_meta_id(cooldown_meta_pb.cooldown_meta_id());
2369
0
    }
2370
2371
0
    {
2372
0
        std::lock_guard rlock(_meta_lock);
2373
0
        SCOPED_SIMPLE_TRACE_IF_TIMEOUT(TRACE_TABLET_LOCK_THRESHOLD);
2374
0
        save_meta();
2375
0
    }
2376
2377
0
    if (!to_add.empty()) {
2378
0
        LOG(INFO) << "modify rowsets when follow cooldowned data, tablet_id=" << tablet_id()
2379
0
                  << [&]() {
2380
0
                         std::stringstream ss;
2381
0
                         ss << " delete rowsets:\n";
2382
0
                         for (auto&& rs : to_delete) {
2383
0
                             ss << rs->version() << ' ' << rs->rowset_id() << '\n';
2384
0
                         }
2385
0
                         ss << "add rowsets:\n";
2386
0
                         for (auto&& rs : to_add) {
2387
0
                             ss << rs->version() << ' ' << rs->rowset_id() << '\n';
2388
0
                         }
2389
0
                         return ss.str();
2390
0
                     }();
2391
0
    }
2392
2393
0
    return Status::OK();
2394
0
}
2395
2396
19.4M
bool Tablet::_has_data_to_cooldown() {
2397
19.4M
    int64_t min_local_version = std::numeric_limits<int64_t>::max();
2398
19.4M
    RowsetSharedPtr rowset;
2399
19.4M
    std::shared_lock meta_rlock(_meta_lock);
2400
    // Ususally once the tablet has done cooldown successfully then the first
2401
    // rowset would always be remote rowset
2402
19.4M
    bool has_cooldowned = false;
2403
33.6M
    for (const auto& [_, rs] : _rs_version_map) {
2404
33.6M
        if (!rs->is_local()) {
2405
228
            has_cooldowned = true;
2406
228
            break;
2407
228
        }
2408
33.6M
    }
2409
33.6M
    for (auto& [v, rs] : _rs_version_map) {
2410
33.6M
        auto predicate = rs->is_local() && v.first < min_local_version;
2411
33.6M
        if (!has_cooldowned) {
2412
33.6M
            predicate = predicate && (rs->data_disk_size() > 0);
2413
33.6M
        }
2414
33.6M
        if (predicate) {
2415
            // this is a local rowset and has data
2416
3.27M
            min_local_version = v.first;
2417
3.27M
            rowset = rs;
2418
3.27M
        }
2419
33.6M
    }
2420
2421
19.4M
    int64_t newest_cooldown_time = 0;
2422
19.4M
    if (rowset != nullptr) {
2423
2.96M
        newest_cooldown_time = _get_newest_cooldown_time(rowset);
2424
2.96M
    }
2425
2426
19.4M
    return (newest_cooldown_time != 0) && (newest_cooldown_time < UnixSeconds());
2427
19.4M
}
2428
2429
19.4M
RowsetSharedPtr Tablet::pick_cooldown_rowset() {
2430
19.4M
    RowsetSharedPtr rowset;
2431
2432
19.4M
    if (!_has_data_to_cooldown()) {
2433
19.4M
        return nullptr;
2434
19.4M
    }
2435
2436
    // TODO(plat1ko): should we maintain `cooldowned_version` in `Tablet`?
2437
34
    int64_t cooldowned_version = -1;
2438
    // We pick the rowset with smallest start version in local.
2439
34
    int64_t min_local_version = std::numeric_limits<int64_t>::max();
2440
34
    {
2441
34
        std::shared_lock meta_rlock(_meta_lock);
2442
154
        for (auto& [v, rs] : _rs_version_map) {
2443
154
            if (!rs->is_local()) {
2444
36
                cooldowned_version = std::max(cooldowned_version, v.second);
2445
118
            } else if (v.first < min_local_version) { // this is a local rowset
2446
105
                min_local_version = v.first;
2447
105
                rowset = rs;
2448
105
            }
2449
154
        }
2450
34
    }
2451
34
    if (!rowset) {
2452
0
        return nullptr;
2453
0
    }
2454
34
    if (tablet_footprint() == 0) {
2455
0
        VLOG_DEBUG << "skip cooldown due to empty tablet_id = " << tablet_id();
2456
0
        return nullptr;
2457
0
    }
2458
34
    if (min_local_version != cooldowned_version + 1) { // ensure version continuity
2459
0
        if (UNLIKELY(cooldowned_version != -1)) {
2460
0
            LOG(WARNING) << "version not continuous. tablet_id=" << tablet_id()
2461
0
                         << " cooldowned_version=" << cooldowned_version
2462
0
                         << " min_local_version=" << min_local_version;
2463
0
        }
2464
0
        return nullptr;
2465
0
    }
2466
34
    return rowset;
2467
34
}
2468
2469
2.96M
int64_t Tablet::_get_newest_cooldown_time(const RowsetSharedPtr& rowset) {
2470
2.96M
    int64_t id = storage_policy_id();
2471
2.96M
    if (id <= 0) {
2472
2.96M
        VLOG_DEBUG << "tablet does not need cooldown, tablet id: " << tablet_id();
2473
2.96M
        return 0;
2474
2.96M
    }
2475
100
    auto storage_policy = get_storage_policy(id);
2476
100
    if (!storage_policy) {
2477
44
        LOG(WARNING) << "Cannot get storage policy: " << id;
2478
44
        return 0;
2479
44
    }
2480
56
    auto cooldown_ttl_sec = storage_policy->cooldown_ttl;
2481
56
    auto cooldown_datetime = storage_policy->cooldown_datetime;
2482
56
    int64_t newest_cooldown_time = std::numeric_limits<int64_t>::max();
2483
2484
56
    if (cooldown_ttl_sec >= 0) {
2485
53
        newest_cooldown_time = rowset->newest_write_timestamp() + cooldown_ttl_sec;
2486
53
    }
2487
56
    if (cooldown_datetime > 0) {
2488
10
        newest_cooldown_time = std::min(newest_cooldown_time, cooldown_datetime);
2489
10
    }
2490
2491
56
    return newest_cooldown_time;
2492
100
}
2493
2494
19.4M
RowsetSharedPtr Tablet::need_cooldown(int64_t* cooldown_timestamp, size_t* file_size) {
2495
19.4M
    RowsetSharedPtr rowset = pick_cooldown_rowset();
2496
19.4M
    if (!rowset) {
2497
19.4M
        VLOG_DEBUG << "pick cooldown rowset, get null, tablet id: " << tablet_id();
2498
19.4M
        return nullptr;
2499
19.4M
    }
2500
2501
21
    auto newest_cooldown_time = _get_newest_cooldown_time(rowset);
2502
2503
    // the rowset should do cooldown job only if it's cooldown ttl plus newest write time is less than
2504
    // current time or it's datatime is less than current time
2505
21
    if (newest_cooldown_time != 0 && newest_cooldown_time < UnixSeconds()) {
2506
21
        *cooldown_timestamp = newest_cooldown_time;
2507
21
        *file_size = rowset->total_disk_size();
2508
21
        VLOG_DEBUG << "tablet need cooldown, tablet id: " << tablet_id()
2509
0
                   << " file_size: " << *file_size;
2510
21
        return rowset;
2511
21
    }
2512
2513
0
    VLOG_DEBUG << "tablet does not need cooldown, tablet id: " << tablet_id()
2514
0
               << " newest write time: " << rowset->newest_write_timestamp();
2515
0
    return nullptr;
2516
21
}
2517
2518
void Tablet::record_unused_remote_rowset(const RowsetId& rowset_id, const std::string& resource,
2519
0
                                         int64_t num_segments) {
2520
0
    auto gc_key = REMOTE_ROWSET_GC_PREFIX + rowset_id.to_string();
2521
0
    RemoteRowsetGcPB gc_pb;
2522
0
    gc_pb.set_resource_id(resource);
2523
0
    gc_pb.set_tablet_id(tablet_id());
2524
0
    gc_pb.set_num_segments(num_segments);
2525
0
    auto st =
2526
0
            _data_dir->get_meta()->put(META_COLUMN_FAMILY_INDEX, gc_key, gc_pb.SerializeAsString());
2527
0
    if (!st.ok()) {
2528
0
        LOG(WARNING) << "failed to record unused remote rowset. tablet_id=" << tablet_id()
2529
0
                     << " rowset_id=" << rowset_id << " resource_id=" << resource;
2530
0
    }
2531
0
    unused_remote_rowset_num << 1;
2532
0
}
2533
2534
91.9k
Status Tablet::remove_all_remote_rowsets() {
2535
91.9k
    DCHECK(tablet_state() == TABLET_SHUTDOWN);
2536
91.9k
    std::set<std::string> resource_ids;
2537
139k
    for (auto& rs_meta : _tablet_meta->all_rs_metas()) {
2538
139k
        if (!rs_meta->is_local()) {
2539
1
            resource_ids.insert(rs_meta->resource_id());
2540
1
        }
2541
139k
    }
2542
91.9k
    if (resource_ids.empty()) {
2543
91.9k
        return Status::OK();
2544
91.9k
    }
2545
3
    auto tablet_gc_key = REMOTE_TABLET_GC_PREFIX + std::to_string(tablet_id());
2546
3
    RemoteTabletGcPB gc_pb;
2547
3
    for (auto& resource_id : resource_ids) {
2548
1
        gc_pb.add_resource_ids(resource_id);
2549
1
    }
2550
3
    return _data_dir->get_meta()->put(META_COLUMN_FAMILY_INDEX, tablet_gc_key,
2551
3
                                      gc_pb.SerializeAsString());
2552
91.9k
}
2553
2554
0
void Tablet::update_max_version_schema(const TabletSchemaSPtr& tablet_schema) {
2555
0
    std::lock_guard wrlock(_meta_lock);
2556
0
    SCOPED_SIMPLE_TRACE_IF_TIMEOUT(TRACE_TABLET_LOCK_THRESHOLD);
2557
    // Double Check for concurrent update
2558
0
    if (!_max_version_schema ||
2559
0
        tablet_schema->schema_version() > _max_version_schema->schema_version()) {
2560
0
        _max_version_schema = tablet_schema;
2561
0
    }
2562
0
}
2563
2564
22
CalcDeleteBitmapExecutor* Tablet::calc_delete_bitmap_executor() {
2565
22
    return _engine.calc_delete_bitmap_executor();
2566
22
}
2567
2568
Status Tablet::save_delete_bitmap(const TabletTxnInfo* txn_info, int64_t txn_id,
2569
                                  DeleteBitmapPtr delete_bitmap, RowsetWriter* rowset_writer,
2570
                                  const RowsetIdUnorderedSet& cur_rowset_ids, int64_t lock_id,
2571
65.9k
                                  int64_t next_visible_version) {
2572
65.9k
    RowsetSharedPtr rowset = txn_info->rowset;
2573
65.9k
    int64_t cur_version = rowset->start_version();
2574
2575
    // update version without write lock, compaction and publish_txn
2576
    // will update delete bitmap, handle compaction with _rowset_update_lock
2577
    // and publish_txn runs sequential so no need to lock here
2578
553k
    for (auto& [key, bitmap] : delete_bitmap->delete_bitmap) {
2579
        // skip sentinel mark, which is used for delete bitmap correctness check
2580
553k
        if (std::get<1>(key) != DeleteBitmap::INVALID_SEGMENT_ID) {
2581
7.79k
            _tablet_meta->delete_bitmap().merge({std::get<0>(key), std::get<1>(key), cur_version},
2582
7.79k
                                                bitmap);
2583
7.79k
        }
2584
553k
    }
2585
2586
65.9k
    return Status::OK();
2587
65.9k
}
2588
2589
3.61k
void Tablet::merge_delete_bitmap(const DeleteBitmap& delete_bitmap) {
2590
3.61k
    _tablet_meta->delete_bitmap().merge(delete_bitmap);
2591
3.61k
}
2592
2593
0
bool Tablet::check_all_rowset_segment() {
2594
0
    std::shared_lock rdlock(_meta_lock);
2595
0
    for (auto& version_rowset : _rs_version_map) {
2596
0
        RowsetSharedPtr rowset = version_rowset.second;
2597
0
        if (!rowset->check_rowset_segment()) {
2598
0
            LOG(WARNING) << "Tablet Segment Check. find a bad tablet, tablet_id=" << tablet_id();
2599
0
            return false;
2600
0
        }
2601
0
    }
2602
0
    return true;
2603
0
}
2604
2605
72.7M
void Tablet::set_skip_compaction(bool skip, CompactionType compaction_type, int64_t start) {
2606
72.7M
    if (!skip) {
2607
115
        _skip_cumu_compaction = false;
2608
115
        _skip_base_compaction = false;
2609
115
        return;
2610
115
    }
2611
72.7M
    if (compaction_type == CompactionType::CUMULATIVE_COMPACTION) {
2612
38.9M
        _skip_cumu_compaction = true;
2613
38.9M
        _skip_cumu_compaction_ts = start;
2614
38.9M
    } else {
2615
33.7M
        DCHECK(compaction_type == CompactionType::BASE_COMPACTION);
2616
33.7M
        _skip_base_compaction = true;
2617
33.7M
        _skip_base_compaction_ts = start;
2618
33.7M
    }
2619
72.7M
}
2620
2621
1.40G
bool Tablet::should_skip_compaction(CompactionType compaction_type, int64_t now) {
2622
1.40G
    if (compaction_type == CompactionType::CUMULATIVE_COMPACTION && _skip_cumu_compaction &&
2623
1.40G
        now < _skip_cumu_compaction_ts + config::skip_tablet_compaction_second) {
2624
1.18G
        return true;
2625
1.18G
    } else if (compaction_type == CompactionType::BASE_COMPACTION && _skip_base_compaction &&
2626
214M
               now < _skip_base_compaction_ts + config::skip_tablet_compaction_second) {
2627
87.6M
        return true;
2628
87.6M
    }
2629
126M
    return false;
2630
1.40G
}
2631
2632
240
std::pair<std::string, int64_t> Tablet::get_binlog_info(std::string_view binlog_version) const {
2633
240
    return RowsetMetaManager::get_binlog_info(_data_dir->get_meta(), tablet_uid(), binlog_version);
2634
240
}
2635
2636
std::string Tablet::get_rowset_binlog_meta(std::string_view binlog_version,
2637
222
                                           std::string_view rowset_id) const {
2638
222
    return RowsetMetaManager::get_rowset_binlog_meta(_data_dir->get_meta(), tablet_uid(),
2639
222
                                                     binlog_version, rowset_id);
2640
222
}
2641
2642
Status Tablet::get_rowset_binlog_metas(const std::vector<int64_t>& binlog_versions,
2643
0
                                       RowsetBinlogMetasPB* metas_pb) {
2644
0
    return RowsetMetaManager::get_rowset_binlog_metas(_data_dir->get_meta(), tablet_uid(),
2645
0
                                                      binlog_versions, metas_pb);
2646
0
}
2647
2648
4
Status Tablet::get_rowset_binlog_metas(Version binlog_versions, RowsetBinlogMetasPB* metas_pb) {
2649
4
    return RowsetMetaManager::get_rowset_binlog_metas(_data_dir->get_meta(), tablet_uid(),
2650
4
                                                      binlog_versions, metas_pb);
2651
4
}
2652
2653
std::string Tablet::get_segment_filepath(std::string_view rowset_id,
2654
436
                                         std::string_view segment_index) const {
2655
436
    return fmt::format("{}/_binlog/{}_{}.dat", _tablet_path, rowset_id, segment_index);
2656
436
}
2657
2658
1
std::string Tablet::get_segment_filepath(std::string_view rowset_id, int64_t segment_index) const {
2659
1
    return fmt::format("{}/_binlog/{}_{}.dat", _tablet_path, rowset_id, segment_index);
2660
1
}
2661
2662
0
std::vector<std::string> Tablet::get_binlog_filepath(std::string_view binlog_version) const {
2663
0
    const auto& [rowset_id, num_segments] = get_binlog_info(binlog_version);
2664
0
    std::vector<std::string> binlog_filepath;
2665
0
    for (int i = 0; i < num_segments; ++i) {
2666
        // TODO(Drogon): rewrite by filesystem path
2667
0
        auto segment_file = fmt::format("{}_{}.dat", rowset_id, i);
2668
0
        binlog_filepath.emplace_back(fmt::format("{}/_binlog/{}", _tablet_path, segment_file));
2669
0
    }
2670
0
    return binlog_filepath;
2671
0
}
2672
2673
444
bool Tablet::can_add_binlog(uint64_t total_binlog_size) const {
2674
444
    return !_data_dir->reach_capacity_limit(total_binlog_size);
2675
444
}
2676
2677
259k
bool Tablet::is_enable_binlog() {
2678
259k
    return config::enable_feature_binlog && tablet_meta()->binlog_config().is_enable();
2679
259k
}
2680
2681
44
void Tablet::set_binlog_config(BinlogConfig binlog_config) {
2682
44
    tablet_meta()->set_binlog_config(binlog_config);
2683
44
}
2684
2685
2
void Tablet::gc_binlogs(int64_t version) {
2686
2
    auto meta = _data_dir->get_meta();
2687
2
    DCHECK(meta != nullptr);
2688
2689
2
    const auto& tablet_uid = this->tablet_uid();
2690
2
    const auto tablet_id = this->tablet_id();
2691
2
    std::string begin_key = make_binlog_meta_key_prefix(tablet_uid);
2692
2
    std::string end_key = make_binlog_meta_key_prefix(tablet_uid, version + 1);
2693
2
    LOG(INFO) << fmt::format("gc binlog meta, tablet_id:{}, begin_key:{}, end_key:{}", tablet_id,
2694
2
                             begin_key, end_key);
2695
2696
2
    std::vector<std::string> wait_for_deleted_binlog_keys;
2697
2
    std::vector<std::string> wait_for_deleted_binlog_files;
2698
2
    auto add_to_wait_for_deleted = [&](std::string_view key, std::string_view rowset_id,
2699
2
                                       int64_t num_segments) {
2700
        // add binlog meta key and binlog data key
2701
0
        wait_for_deleted_binlog_keys.emplace_back(key);
2702
0
        wait_for_deleted_binlog_keys.push_back(get_binlog_data_key_from_meta_key(key));
2703
2704
        // add binlog segment files and index files
2705
0
        for (int64_t i = 0; i < num_segments; ++i) {
2706
0
            auto segment_file_path = get_segment_filepath(rowset_id, i);
2707
0
            wait_for_deleted_binlog_files.emplace_back(segment_file_path);
2708
2709
            // index files
2710
0
            if (tablet_schema()->has_inverted_index()) {
2711
0
                if (tablet_schema()->get_inverted_index_storage_format() ==
2712
0
                    InvertedIndexStorageFormatPB::V1) {
2713
0
                    for (const auto& index : tablet_schema()->inverted_indexes()) {
2714
0
                        auto index_file = InvertedIndexDescriptor::get_index_file_path_v1(
2715
0
                                InvertedIndexDescriptor::get_index_file_path_prefix(
2716
0
                                        segment_file_path),
2717
0
                                index->index_id(), index->get_index_suffix());
2718
0
                        wait_for_deleted_binlog_files.emplace_back(index_file);
2719
0
                    }
2720
0
                } else {
2721
0
                    auto index_file = InvertedIndexDescriptor::get_index_file_path_v2(
2722
0
                            InvertedIndexDescriptor::get_index_file_path_prefix(segment_file_path));
2723
0
                    wait_for_deleted_binlog_files.emplace_back(index_file);
2724
0
                }
2725
0
            }
2726
0
        }
2727
0
    };
2728
2729
2
    auto check_binlog_ttl = [&](std::string_view key, std::string_view value) mutable -> bool {
2730
0
        if (key >= end_key) {
2731
0
            return false;
2732
0
        }
2733
2734
0
        BinlogMetaEntryPB binlog_meta_entry_pb;
2735
0
        if (!binlog_meta_entry_pb.ParseFromArray(value.data(), value.size())) {
2736
0
            LOG(WARNING) << "failed to parse binlog meta entry, key:" << key;
2737
0
            return true;
2738
0
        }
2739
2740
0
        auto num_segments = binlog_meta_entry_pb.num_segments();
2741
0
        std::string rowset_id;
2742
0
        if (binlog_meta_entry_pb.has_rowset_id_v2()) {
2743
0
            rowset_id = binlog_meta_entry_pb.rowset_id_v2();
2744
0
        } else {
2745
            // key is 'binlog_meta_6943f1585fe834b5-e542c2b83a21d0b7_00000000000000000069_020000000000000135449d7cd7eadfe672aa0f928fa99593', extract last part '020000000000000135449d7cd7eadfe672aa0f928fa99593'
2746
0
            auto pos = key.rfind('_');
2747
0
            if (pos == std::string::npos) {
2748
0
                LOG(WARNING) << fmt::format("invalid binlog meta key:{}", key);
2749
0
                return false;
2750
0
            }
2751
0
            rowset_id = key.substr(pos + 1);
2752
0
        }
2753
0
        add_to_wait_for_deleted(key, rowset_id, num_segments);
2754
2755
0
        return true;
2756
0
    };
2757
2758
2
    auto status = meta->iterate(META_COLUMN_FAMILY_INDEX, begin_key, check_binlog_ttl);
2759
2
    if (!status.ok()) {
2760
0
        LOG(WARNING) << "failed to iterate binlog meta, status:" << status;
2761
0
        return;
2762
0
    }
2763
2764
    // first remove binlog files, if failed, just break, then retry next time
2765
    // this keep binlog meta in meta store, so that binlog can be removed next time
2766
2
    bool remove_binlog_files_failed = false;
2767
2
    for (auto& file : wait_for_deleted_binlog_files) {
2768
2
        if (unlink(file.c_str()) != 0) {
2769
            // file not exist, continue
2770
0
            if (errno == ENOENT) {
2771
0
                continue;
2772
0
            }
2773
2774
0
            remove_binlog_files_failed = true;
2775
0
            LOG(WARNING) << "failed to remove binlog file:" << file << ", errno:" << errno;
2776
0
            break;
2777
0
        }
2778
2
    }
2779
2
    if (!remove_binlog_files_failed) {
2780
2
        static_cast<void>(meta->remove(META_COLUMN_FAMILY_INDEX, wait_for_deleted_binlog_keys));
2781
2
    }
2782
2
}
2783
2784
0
Status Tablet::ingest_binlog_metas(RowsetBinlogMetasPB* metas_pb) {
2785
0
    return RowsetMetaManager::ingest_binlog_metas(_data_dir->get_meta(), tablet_uid(), metas_pb);
2786
0
}
2787
2788
184k
void Tablet::clear_cache() {
2789
184k
    std::vector<RowsetSharedPtr> rowsets;
2790
184k
    {
2791
184k
        std::shared_lock rlock(get_header_lock());
2792
184k
        SCOPED_SIMPLE_TRACE_IF_TIMEOUT(TRACE_TABLET_LOCK_THRESHOLD);
2793
2794
298k
        for (auto& [_, rowset] : rowset_map()) {
2795
298k
            rowsets.push_back(rowset);
2796
298k
        }
2797
184k
        for (auto& [_, rowset] : stale_rowset_map()) {
2798
32.7k
            rowsets.push_back(rowset);
2799
32.7k
        }
2800
184k
    }
2801
331k
    for (auto& rowset : rowsets) {
2802
331k
        rowset->clear_cache();
2803
331k
    }
2804
184k
}
2805
2806
585k
void Tablet::check_table_size_correctness() {
2807
585k
    if (!config::enable_table_size_correctness_check) {
2808
585k
        return;
2809
585k
    }
2810
18.4E
    const std::vector<RowsetMetaSharedPtr>& all_rs_metas = _tablet_meta->all_rs_metas();
2811
18.4E
    for (const auto& rs_meta : all_rs_metas) {
2812
0
        int64_t total_segment_size = get_segment_file_size(rs_meta);
2813
0
        int64_t total_inverted_index_size = get_inverted_index_file_size(rs_meta);
2814
0
        if (rs_meta->data_disk_size() != total_segment_size ||
2815
0
            rs_meta->index_disk_size() != total_inverted_index_size ||
2816
0
            rs_meta->data_disk_size() + rs_meta->index_disk_size() != rs_meta->total_disk_size()) {
2817
0
            LOG(WARNING) << "[Local table table size check failed]:"
2818
0
                         << " tablet id: " << rs_meta->tablet_id()
2819
0
                         << ", rowset id:" << rs_meta->rowset_id()
2820
0
                         << ", rowset data disk size:" << rs_meta->data_disk_size()
2821
0
                         << ", rowset real data disk size:" << total_segment_size
2822
0
                         << ", rowset index disk size:" << rs_meta->index_disk_size()
2823
0
                         << ", rowset real index disk size:" << total_inverted_index_size
2824
0
                         << ", rowset total disk size:" << rs_meta->total_disk_size()
2825
0
                         << ", rowset segment path:"
2826
0
                         << StorageResource().remote_segment_path(
2827
0
                                    rs_meta->tablet_id(), rs_meta->rowset_id().to_string(), 0);
2828
0
            DCHECK(false);
2829
0
        }
2830
0
    }
2831
18.4E
}
2832
2833
0
std::string Tablet::get_segment_path(const RowsetMetaSharedPtr& rs_meta, int64_t seg_id) {
2834
0
    std::string segment_path;
2835
0
    if (rs_meta->is_local()) {
2836
0
        segment_path = local_segment_path(_tablet_path, rs_meta->rowset_id().to_string(), seg_id);
2837
0
    } else {
2838
0
        segment_path = rs_meta->remote_storage_resource().value()->remote_segment_path(
2839
0
                rs_meta->tablet_id(), rs_meta->rowset_id().to_string(), seg_id);
2840
0
    }
2841
0
    return segment_path;
2842
0
}
2843
2844
0
int64_t Tablet::get_segment_file_size(const RowsetMetaSharedPtr& rs_meta) {
2845
0
    const auto& fs = rs_meta->fs();
2846
0
    if (!fs) {
2847
0
        LOG(WARNING) << "get fs failed, resource_id={}" << rs_meta->resource_id();
2848
0
    }
2849
0
    int64_t total_segment_size = 0;
2850
0
    for (int64_t seg_id = 0; seg_id < rs_meta->num_segments(); seg_id++) {
2851
0
        std::string segment_path = get_segment_path(rs_meta, seg_id);
2852
0
        int64_t segment_file_size = 0;
2853
0
        auto st = fs->file_size(segment_path, &segment_file_size);
2854
0
        if (!st.ok()) {
2855
0
            segment_file_size = 0;
2856
0
            LOG(WARNING) << "table size correctness check get segment size failed! msg:"
2857
0
                         << st.to_string() << ", segment path:" << segment_path;
2858
0
        }
2859
0
        total_segment_size += segment_file_size;
2860
0
    }
2861
0
    return total_segment_size;
2862
0
}
2863
2864
0
int64_t Tablet::get_inverted_index_file_size(const RowsetMetaSharedPtr& rs_meta) {
2865
0
    const auto& fs = rs_meta->fs();
2866
0
    if (!fs) {
2867
0
        LOG(WARNING) << "get fs failed, resource_id={}" << rs_meta->resource_id();
2868
0
    }
2869
0
    int64_t total_inverted_index_size = 0;
2870
2871
0
    if (rs_meta->tablet_schema()->get_inverted_index_storage_format() ==
2872
0
        InvertedIndexStorageFormatPB::V1) {
2873
0
        const auto& indices = rs_meta->tablet_schema()->inverted_indexes();
2874
0
        for (auto& index : indices) {
2875
0
            for (int seg_id = 0; seg_id < rs_meta->num_segments(); ++seg_id) {
2876
0
                std::string segment_path = get_segment_path(rs_meta, seg_id);
2877
0
                int64_t file_size = 0;
2878
2879
0
                std::string inverted_index_file_path =
2880
0
                        InvertedIndexDescriptor::get_index_file_path_v1(
2881
0
                                InvertedIndexDescriptor::get_index_file_path_prefix(segment_path),
2882
0
                                index->index_id(), index->get_index_suffix());
2883
0
                auto st = fs->file_size(inverted_index_file_path, &file_size);
2884
0
                if (!st.ok()) {
2885
0
                    file_size = 0;
2886
0
                    LOG(WARNING) << " tablet id: " << get_tablet_info().tablet_id
2887
0
                                 << ", rowset id:" << rs_meta->rowset_id()
2888
0
                                 << ", table size correctness check get inverted index v1 "
2889
0
                                    "size failed! msg:"
2890
0
                                 << st.to_string()
2891
0
                                 << ", inverted index path:" << inverted_index_file_path;
2892
0
                }
2893
0
                total_inverted_index_size += file_size;
2894
0
            }
2895
0
        }
2896
0
    } else {
2897
0
        for (int seg_id = 0; seg_id < rs_meta->num_segments(); ++seg_id) {
2898
0
            int64_t file_size = 0;
2899
0
            std::string segment_path = get_segment_path(rs_meta, seg_id);
2900
0
            std::string inverted_index_file_path = InvertedIndexDescriptor::get_index_file_path_v2(
2901
0
                    InvertedIndexDescriptor::get_index_file_path_prefix(segment_path));
2902
0
            auto st = fs->file_size(inverted_index_file_path, &file_size);
2903
0
            if (!st.ok()) {
2904
0
                file_size = 0;
2905
0
                if (st.is<NOT_FOUND>()) {
2906
0
                    LOG(INFO) << " tablet id: " << get_tablet_info().tablet_id
2907
0
                              << ", rowset id:" << rs_meta->rowset_id()
2908
0
                              << ", table size correctness check get inverted index v2 failed "
2909
0
                                 "because file not exist:"
2910
0
                              << inverted_index_file_path;
2911
0
                } else {
2912
0
                    LOG(WARNING) << " tablet id: " << get_tablet_info().tablet_id
2913
0
                                 << ", rowset id:" << rs_meta->rowset_id()
2914
0
                                 << ", table size correctness check get inverted index v2 "
2915
0
                                    "size failed! msg:"
2916
0
                                 << st.to_string()
2917
0
                                 << ", inverted index path:" << inverted_index_file_path;
2918
0
                }
2919
0
            }
2920
0
            total_inverted_index_size += file_size;
2921
0
        }
2922
0
    }
2923
0
    return total_inverted_index_size;
2924
0
}
2925
2926
} // namespace doris