Coverage Report

Created: 2025-06-06 13:44

/root/doris/be/src/cloud/cloud_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 "cloud/cloud_tablet.h"
19
20
#include <bvar/bvar.h>
21
#include <gen_cpp/olap_file.pb.h>
22
#include <rapidjson/document.h>
23
#include <rapidjson/encodings.h>
24
#include <rapidjson/prettywriter.h>
25
#include <rapidjson/rapidjson.h>
26
#include <rapidjson/stringbuffer.h>
27
28
#include <atomic>
29
#include <memory>
30
#include <shared_mutex>
31
#include <unordered_map>
32
#include <vector>
33
34
#include "cloud/cloud_meta_mgr.h"
35
#include "cloud/cloud_storage_engine.h"
36
#include "cloud/cloud_tablet_mgr.h"
37
#include "common/cast_set.h"
38
#include "common/config.h"
39
#include "common/logging.h"
40
#include "io/cache/block_file_cache_downloader.h"
41
#include "io/cache/block_file_cache_factory.h"
42
#include "olap/compaction.h"
43
#include "olap/cumulative_compaction_time_series_policy.h"
44
#include "olap/olap_define.h"
45
#include "olap/rowset/beta_rowset.h"
46
#include "olap/rowset/rowset.h"
47
#include "olap/rowset/rowset_factory.h"
48
#include "olap/rowset/rowset_fwd.h"
49
#include "olap/rowset/rowset_writer.h"
50
#include "olap/rowset/segment_v2/inverted_index_desc.h"
51
#include "olap/storage_policy.h"
52
#include "olap/tablet_schema.h"
53
#include "olap/txn_manager.h"
54
#include "util/debug_points.h"
55
#include "vec/common/schema_util.h"
56
57
namespace doris {
58
#include "common/compile_check_begin.h"
59
using namespace ErrorCode;
60
61
bvar::LatencyRecorder g_cu_compaction_get_delete_bitmap_lock_time_ms(
62
        "cu_compaction_get_delete_bitmap_lock_time_ms");
63
bvar::LatencyRecorder g_base_compaction_get_delete_bitmap_lock_time_ms(
64
        "base_compaction_get_delete_bitmap_lock_time_ms");
65
66
static constexpr int LOAD_INITIATOR_ID = -1;
67
68
CloudTablet::CloudTablet(CloudStorageEngine& engine, TabletMetaSharedPtr tablet_meta)
69
3
        : BaseTablet(std::move(tablet_meta)), _engine(engine) {}
70
71
3
CloudTablet::~CloudTablet() = default;
72
73
0
bool CloudTablet::exceed_version_limit(int32_t limit) {
74
0
    return _approximate_num_rowsets.load(std::memory_order_relaxed) > limit;
75
0
}
76
77
Status CloudTablet::capture_consistent_rowsets_unlocked(
78
0
        const Version& spec_version, std::vector<RowsetSharedPtr>* rowsets) const {
79
0
    Versions version_path;
80
0
    auto st = _timestamped_version_tracker.capture_consistent_versions(spec_version, &version_path);
81
0
    if (!st.ok()) {
82
        // Check no missed versions or req version is merged
83
0
        auto missed_versions = get_missed_versions(spec_version.second);
84
0
        if (missed_versions.empty()) {
85
0
            st.set_code(VERSION_ALREADY_MERGED); // Reset error code
86
0
        }
87
0
        st.append(" tablet_id=" + std::to_string(tablet_id()));
88
0
        return st;
89
0
    }
90
0
    VLOG_DEBUG << "capture consitent versions: " << version_path;
91
0
    return _capture_consistent_rowsets_unlocked(version_path, rowsets);
92
0
}
93
94
Status CloudTablet::capture_rs_readers(const Version& spec_version,
95
                                       std::vector<RowSetSplits>* rs_splits,
96
0
                                       bool skip_missing_version) {
97
0
    DBUG_EXECUTE_IF("CloudTablet.capture_rs_readers.return.e-230", {
98
0
        LOG_WARNING("CloudTablet.capture_rs_readers.return e-230").tag("tablet_id", tablet_id());
99
0
        return Status::Error<false>(-230, "injected error");
100
0
    });
101
0
    Versions version_path;
102
0
    std::shared_lock rlock(_meta_lock);
103
0
    auto st = _timestamped_version_tracker.capture_consistent_versions(spec_version, &version_path);
104
0
    if (!st.ok()) {
105
0
        rlock.unlock(); // avoid logging in lock range
106
        // Check no missed versions or req version is merged
107
0
        auto missed_versions = get_missed_versions(spec_version.second);
108
0
        if (missed_versions.empty()) {
109
0
            st.set_code(VERSION_ALREADY_MERGED); // Reset error code
110
0
            st.append(" versions are already compacted, ");
111
0
        }
112
0
        st.append(" tablet_id=" + std::to_string(tablet_id()));
113
        // clang-format off
114
0
        LOG(WARNING) << st << '\n' << [this]() { std::string json; get_compaction_status(&json); return json; }();
115
        // clang-format on
116
0
        return st;
117
0
    }
118
0
    VLOG_DEBUG << "capture consitent versions: " << version_path;
119
0
    return capture_rs_readers_unlocked(version_path, rs_splits);
120
0
}
121
122
0
Status CloudTablet::merge_rowsets_schema() {
123
    // Find the rowset with the max version
124
0
    auto max_version_rowset =
125
0
            std::max_element(
126
0
                    _rs_version_map.begin(), _rs_version_map.end(),
127
0
                    [](const auto& a, const auto& b) {
128
0
                        return !a.second->tablet_schema()
129
0
                                       ? true
130
0
                                       : (!b.second->tablet_schema()
131
0
                                                  ? false
132
0
                                                  : a.second->tablet_schema()->schema_version() <
133
0
                                                            b.second->tablet_schema()
134
0
                                                                    ->schema_version());
135
0
                    })
136
0
                    ->second;
137
0
    TabletSchemaSPtr max_version_schema = max_version_rowset->tablet_schema();
138
    // If the schema has variant columns, perform a merge to create a wide tablet schema
139
0
    if (max_version_schema->num_variant_columns() > 0) {
140
0
        std::vector<TabletSchemaSPtr> schemas;
141
0
        std::transform(_rs_version_map.begin(), _rs_version_map.end(), std::back_inserter(schemas),
142
0
                       [](const auto& rs_meta) { return rs_meta.second->tablet_schema(); });
143
        // Merge the collected schemas to obtain the least common schema
144
0
        RETURN_IF_ERROR(vectorized::schema_util::get_least_common_schema(schemas, nullptr,
145
0
                                                                         max_version_schema));
146
0
        VLOG_DEBUG << "dump schema: " << max_version_schema->dump_full_schema();
147
0
        _merged_tablet_schema = max_version_schema;
148
0
    }
149
0
    return Status::OK();
150
0
}
151
152
// There are only two tablet_states RUNNING and NOT_READY in cloud mode
153
// This function will erase the tablet from `CloudTabletMgr` when it can't find this tablet in MS.
154
0
Status CloudTablet::sync_rowsets(const SyncOptions& options, SyncRowsetStats* stats) {
155
0
    RETURN_IF_ERROR(sync_if_not_running(stats));
156
157
0
    if (options.query_version > 0) {
158
0
        std::shared_lock rlock(_meta_lock);
159
0
        if (_max_version >= options.query_version) {
160
0
            return Status::OK();
161
0
        }
162
0
    }
163
164
    // serially execute sync to reduce unnecessary network overhead
165
0
    std::unique_lock lock(_sync_meta_lock);
166
0
    if (options.query_version > 0) {
167
0
        std::shared_lock rlock(_meta_lock);
168
0
        if (_max_version >= options.query_version) {
169
0
            return Status::OK();
170
0
        }
171
0
    }
172
173
0
    auto st = _engine.meta_mgr().sync_tablet_rowsets_unlocked(this, lock, options, stats);
174
0
    if (st.is<ErrorCode::NOT_FOUND>()) {
175
0
        clear_cache();
176
0
    }
177
178
0
    return st;
179
0
}
180
181
// Sync tablet meta and all rowset meta if not running.
182
// This could happen when BE didn't finish schema change job and another BE committed this schema change job.
183
// It should be a quite rare situation.
184
0
Status CloudTablet::sync_if_not_running(SyncRowsetStats* stats) {
185
0
    if (tablet_state() == TABLET_RUNNING) {
186
0
        return Status::OK();
187
0
    }
188
189
    // Serially execute sync to reduce unnecessary network overhead
190
0
    std::unique_lock lock(_sync_meta_lock);
191
192
0
    {
193
0
        std::shared_lock rlock(_meta_lock);
194
0
        if (tablet_state() == TABLET_RUNNING) {
195
0
            return Status::OK();
196
0
        }
197
0
    }
198
199
0
    TabletMetaSharedPtr tablet_meta;
200
0
    auto st = _engine.meta_mgr().get_tablet_meta(tablet_id(), &tablet_meta);
201
0
    if (!st.ok()) {
202
0
        if (st.is<ErrorCode::NOT_FOUND>()) {
203
0
            clear_cache();
204
0
        }
205
0
        return st;
206
0
    }
207
208
0
    if (tablet_meta->tablet_state() != TABLET_RUNNING) [[unlikely]] {
209
        // MoW may go to here when load while schema change
210
0
        return Status::OK();
211
0
    }
212
213
0
    TimestampedVersionTracker empty_tracker;
214
0
    {
215
0
        std::lock_guard wlock(_meta_lock);
216
0
        RETURN_IF_ERROR(set_tablet_state(TABLET_RUNNING));
217
0
        _rs_version_map.clear();
218
0
        _stale_rs_version_map.clear();
219
0
        std::swap(_timestamped_version_tracker, empty_tracker);
220
0
        _tablet_meta->clear_rowsets();
221
0
        _tablet_meta->clear_stale_rowset();
222
0
        _max_version = -1;
223
0
    }
224
225
0
    st = _engine.meta_mgr().sync_tablet_rowsets_unlocked(this, lock, {}, stats);
226
0
    if (st.is<ErrorCode::NOT_FOUND>()) {
227
0
        clear_cache();
228
0
    }
229
0
    return st;
230
0
}
231
232
0
TabletSchemaSPtr CloudTablet::merged_tablet_schema() const {
233
0
    std::shared_lock rlock(_meta_lock);
234
0
    return _merged_tablet_schema;
235
0
}
236
237
void CloudTablet::add_rowsets(std::vector<RowsetSharedPtr> to_add, bool version_overlap,
238
                              std::unique_lock<std::shared_mutex>& meta_lock,
239
0
                              bool warmup_delta_data) {
240
0
    if (to_add.empty()) {
241
0
        return;
242
0
    }
243
244
0
    auto add_rowsets_directly = [=, this](std::vector<RowsetSharedPtr>& rowsets) {
245
0
        for (auto& rs : rowsets) {
246
0
            if (version_overlap || warmup_delta_data) {
247
#ifndef BE_TEST
248
                // Warmup rowset data in background
249
                for (int seg_id = 0; seg_id < rs->num_segments(); ++seg_id) {
250
                    const auto& rowset_meta = rs->rowset_meta();
251
                    constexpr int64_t interval = 600; // 10 mins
252
                    // When BE restart and receive the `load_sync` rpc, it will sync all historical rowsets first time.
253
                    // So we need to filter out the old rowsets avoid to download the whole table.
254
                    if (warmup_delta_data &&
255
                        ::time(nullptr) - rowset_meta->newest_write_timestamp() >= interval) {
256
                        continue;
257
                    }
258
259
                    auto storage_resource = rowset_meta->remote_storage_resource();
260
                    if (!storage_resource) {
261
                        LOG(WARNING) << storage_resource.error();
262
                        continue;
263
                    }
264
265
                    int64_t expiration_time =
266
                            _tablet_meta->ttl_seconds() == 0 ||
267
                                            rowset_meta->newest_write_timestamp() <= 0
268
                                    ? 0
269
                                    : rowset_meta->newest_write_timestamp() +
270
                                              _tablet_meta->ttl_seconds();
271
                    // clang-format off
272
                    _engine.file_cache_block_downloader().submit_download_task(io::DownloadFileMeta {
273
                            .path = storage_resource.value()->remote_segment_path(*rowset_meta, seg_id),
274
                            .file_size = rs->rowset_meta()->segment_file_size(seg_id),
275
                            .file_system = storage_resource.value()->fs,
276
                            .ctx =
277
                                    {
278
                                            .expiration_time = expiration_time,
279
                                            .is_dryrun = config::enable_reader_dryrun_when_download_file_cache,
280
                                    },
281
                            .download_done {},
282
                    });
283
284
                    auto download_idx_file = [&](const io::Path& idx_path) {
285
                        io::DownloadFileMeta meta {
286
                                .path = idx_path,
287
                                .file_size = -1,
288
                                .file_system = storage_resource.value()->fs,
289
                                .ctx =
290
                                        {
291
                                                .expiration_time = expiration_time,
292
                                                .is_dryrun = config::enable_reader_dryrun_when_download_file_cache,
293
                                        },
294
                                .download_done {},
295
                        };
296
                        _engine.file_cache_block_downloader().submit_download_task(std::move(meta));
297
                    };
298
                    // clang-format on
299
                    auto schema_ptr = rowset_meta->tablet_schema();
300
                    auto idx_version = schema_ptr->get_inverted_index_storage_format();
301
                    if (idx_version == InvertedIndexStorageFormatPB::V1) {
302
                        for (const auto& index : schema_ptr->inverted_indexes()) {
303
                            auto idx_path = storage_resource.value()->remote_idx_v1_path(
304
                                    *rowset_meta, seg_id, index->index_id(),
305
                                    index->get_index_suffix());
306
                            download_idx_file(idx_path);
307
                        }
308
                    } else {
309
                        if (schema_ptr->has_inverted_index()) {
310
                            auto idx_path = storage_resource.value()->remote_idx_v2_path(
311
                                    *rowset_meta, seg_id);
312
                            download_idx_file(idx_path);
313
                        }
314
                    }
315
                }
316
#endif
317
0
            }
318
0
            _rs_version_map.emplace(rs->version(), rs);
319
0
            _timestamped_version_tracker.add_version(rs->version());
320
0
            _max_version = std::max(rs->end_version(), _max_version);
321
0
            update_base_size(*rs);
322
0
        }
323
0
        _tablet_meta->add_rowsets_unchecked(rowsets);
324
0
    };
325
326
0
    if (!version_overlap) {
327
0
        add_rowsets_directly(to_add);
328
0
        return;
329
0
    }
330
331
    // Filter out existed rowsets
332
0
    auto remove_it =
333
0
            std::remove_if(to_add.begin(), to_add.end(), [this](const RowsetSharedPtr& rs) {
334
0
                if (auto find_it = _rs_version_map.find(rs->version());
335
0
                    find_it == _rs_version_map.end()) {
336
0
                    return false;
337
0
                } else if (find_it->second->rowset_id() == rs->rowset_id()) {
338
0
                    return true; // Same rowset
339
0
                }
340
341
                // If version of rowset in `to_add` is equal to rowset in tablet but rowset_id is not equal,
342
                // replace existed rowset with `to_add` rowset. This may occur when:
343
                //  1. schema change converts rowsets which have been double written to new tablet
344
                //  2. cumu compaction picks single overlapping input rowset to perform compaction
345
0
                _tablet_meta->delete_rs_meta_by_version(rs->version(), nullptr);
346
0
                _rs_version_map[rs->version()] = rs;
347
0
                _tablet_meta->add_rowsets_unchecked({rs});
348
0
                update_base_size(*rs);
349
0
                return true;
350
0
            });
351
352
0
    to_add.erase(remove_it, to_add.end());
353
354
    // delete rowsets with overlapped version
355
0
    std::vector<RowsetSharedPtr> to_add_directly;
356
0
    for (auto& to_add_rs : to_add) {
357
        // delete rowsets with overlapped version
358
0
        std::vector<RowsetSharedPtr> to_delete;
359
0
        Version to_add_v = to_add_rs->version();
360
        // if start_version  > max_version, we can skip checking overlap here.
361
0
        if (to_add_v.first > _max_version) {
362
            // if start_version  > max_version, we can skip checking overlap here.
363
0
            to_add_directly.push_back(to_add_rs);
364
0
        } else {
365
0
            to_add_directly.push_back(to_add_rs);
366
0
            for (auto& [v, rs] : _rs_version_map) {
367
0
                if (to_add_v.contains(v)) {
368
0
                    to_delete.push_back(rs);
369
0
                }
370
0
            }
371
0
            delete_rowsets(to_delete, meta_lock);
372
0
        }
373
0
    }
374
375
0
    add_rowsets_directly(to_add_directly);
376
0
}
377
378
void CloudTablet::delete_rowsets(const std::vector<RowsetSharedPtr>& to_delete,
379
0
                                 std::unique_lock<std::shared_mutex>&) {
380
0
    if (to_delete.empty()) {
381
0
        return;
382
0
    }
383
0
    std::vector<RowsetMetaSharedPtr> rs_metas;
384
0
    rs_metas.reserve(to_delete.size());
385
0
    for (auto&& rs : to_delete) {
386
0
        rs_metas.push_back(rs->rowset_meta());
387
0
        _stale_rs_version_map[rs->version()] = rs;
388
0
    }
389
0
    _timestamped_version_tracker.add_stale_path_version(rs_metas);
390
0
    for (auto&& rs : to_delete) {
391
0
        _rs_version_map.erase(rs->version());
392
0
    }
393
394
0
    _tablet_meta->modify_rs_metas({}, rs_metas, false);
395
0
}
396
397
0
uint64_t CloudTablet::delete_expired_stale_rowsets() {
398
0
    if (config::enable_mow_verbose_log) {
399
0
        LOG_INFO("begin delete_expired_stale_rowset for tablet={}", tablet_id());
400
0
    }
401
0
    std::vector<RowsetSharedPtr> expired_rowsets;
402
    // ATTN: trick, Use stale_rowsets to temporarily increase the reference count of the rowset shared pointer in _stale_rs_version_map so that in the recycle_cached_data function, it checks if the reference count is 2.
403
0
    std::vector<std::pair<Version, std::vector<RowsetSharedPtr>>> deleted_stale_rowsets;
404
0
    int64_t expired_stale_sweep_endtime =
405
0
            ::time(nullptr) - config::tablet_rowset_stale_sweep_time_sec;
406
0
    std::vector<std::string> version_to_delete;
407
0
    {
408
0
        std::unique_lock wlock(_meta_lock);
409
410
0
        std::vector<int64_t> path_ids;
411
        // capture the path version to delete
412
0
        _timestamped_version_tracker.capture_expired_paths(expired_stale_sweep_endtime, &path_ids);
413
414
0
        if (path_ids.empty()) {
415
0
            return 0;
416
0
        }
417
418
0
        for (int64_t path_id : path_ids) {
419
0
            int64_t start_version = -1;
420
0
            int64_t end_version = -1;
421
0
            std::vector<RowsetSharedPtr> stale_rowsets;
422
            // delete stale versions in version graph
423
0
            auto version_path = _timestamped_version_tracker.fetch_and_delete_path_by_id(path_id);
424
0
            for (auto& v_ts : version_path->timestamped_versions()) {
425
0
                auto rs_it = _stale_rs_version_map.find(v_ts->version());
426
0
                if (rs_it != _stale_rs_version_map.end()) {
427
0
                    expired_rowsets.push_back(rs_it->second);
428
0
                    stale_rowsets.push_back(rs_it->second);
429
0
                    LOG(INFO) << "erase stale rowset, tablet_id=" << tablet_id()
430
0
                              << " rowset_id=" << rs_it->second->rowset_id().to_string()
431
0
                              << " version=" << rs_it->first.to_string();
432
0
                    _stale_rs_version_map.erase(rs_it);
433
0
                } else {
434
0
                    LOG(WARNING) << "cannot find stale rowset " << v_ts->version() << " in tablet "
435
0
                                 << tablet_id();
436
                    // clang-format off
437
0
                    DCHECK(false) << [this, &wlock]() { wlock.unlock(); std::string json; get_compaction_status(&json); return json; }();
438
                    // clang-format on
439
0
                }
440
0
                if (start_version < 0) {
441
0
                    start_version = v_ts->version().first;
442
0
                }
443
0
                end_version = v_ts->version().second;
444
0
                _tablet_meta->delete_stale_rs_meta_by_version(v_ts->version());
445
0
            }
446
0
            Version version(start_version, end_version);
447
0
            version_to_delete.emplace_back(version.to_string());
448
0
            if (!stale_rowsets.empty()) {
449
0
                deleted_stale_rowsets.emplace_back(version, std::move(stale_rowsets));
450
0
            }
451
0
        }
452
0
        _reconstruct_version_tracker_if_necessary();
453
0
    }
454
0
    _tablet_meta->delete_bitmap().remove_stale_delete_bitmap_from_queue(version_to_delete);
455
0
    recycle_cached_data(expired_rowsets);
456
0
    if (config::enable_mow_verbose_log) {
457
0
        LOG_INFO("finish delete_expired_stale_rowset for tablet={}", tablet_id());
458
0
    }
459
460
0
    if (keys_type() == UNIQUE_KEYS && enable_unique_key_merge_on_write() &&
461
0
        !deleted_stale_rowsets.empty()) {
462
        // agg delete bitmap for pre rowsets; record unused delete bitmap key ranges
463
0
        OlapStopWatch watch;
464
0
        for (const auto& [version, unused_rowsets] : deleted_stale_rowsets) {
465
            // agg delete bitmap for pre rowset
466
0
            DeleteBitmapKeyRanges remove_delete_bitmap_key_ranges;
467
0
            agg_delete_bitmap_for_stale_rowsets(version, remove_delete_bitmap_key_ranges);
468
            // add remove delete bitmap
469
0
            if (!remove_delete_bitmap_key_ranges.empty()) {
470
0
                std::lock_guard<std::mutex> lock(_gc_mutex);
471
0
                _unused_delete_bitmap.push_back(
472
0
                        std::make_pair(unused_rowsets, remove_delete_bitmap_key_ranges));
473
0
            }
474
0
        }
475
0
        LOG(INFO) << "agg pre rowsets delete bitmap. tablet_id=" << tablet_id()
476
0
                  << ", size=" << deleted_stale_rowsets.size()
477
0
                  << ", cost(us)=" << watch.get_elapse_time_us();
478
0
    }
479
0
    return expired_rowsets.size();
480
0
}
481
482
0
bool CloudTablet::need_remove_pre_rowset_delete_bitmap() {
483
0
    std::lock_guard<std::mutex> lock(_gc_mutex);
484
0
    return !_unused_delete_bitmap.empty();
485
0
}
486
487
0
void CloudTablet::remove_pre_rowset_delete_bitmap() {
488
0
    std::lock_guard<std::mutex> lock(_gc_mutex);
489
0
    for (auto it = _unused_delete_bitmap.begin(); it != _unused_delete_bitmap.end();) {
490
0
        auto& rowsets = std::get<0>(*it);
491
0
        bool find_unused_rowset = false;
492
0
        for (const auto& rowset : rowsets) {
493
0
            if (rowset.use_count() > 1) {
494
0
                LOG(INFO) << "can not remove pre rowset delete bitmap because rowset is in use"
495
0
                          << ", tablet_id=" << tablet_id()
496
0
                          << ", rowset_id=" << rowset->rowset_id().to_string()
497
0
                          << ", version=" << rowset->version().to_string()
498
0
                          << ", use_count=" << rowset.use_count();
499
0
                find_unused_rowset = true;
500
0
                break;
501
0
            }
502
0
        }
503
0
        if (find_unused_rowset) {
504
0
            ++it;
505
0
            continue;
506
0
        }
507
0
        auto& key_ranges = std::get<1>(*it);
508
0
        tablet_meta()->delete_bitmap().remove(key_ranges);
509
0
        it = _unused_delete_bitmap.erase(it);
510
0
    }
511
0
    if (!_unused_delete_bitmap.empty()) {
512
0
        LOG(INFO) << "tablet_id=" << tablet_id()
513
0
                  << ", unused_delete_bitmap size=" << _unused_delete_bitmap.size();
514
0
    }
515
0
}
516
517
0
void CloudTablet::update_base_size(const Rowset& rs) {
518
    // Define base rowset as the rowset of version [2-x]
519
0
    if (rs.start_version() == 2) {
520
0
        _base_size = rs.total_disk_size();
521
0
    }
522
0
}
523
524
0
void CloudTablet::clear_cache() {
525
0
    CloudTablet::recycle_cached_data(get_snapshot_rowset(true));
526
0
    _engine.tablet_mgr().erase_tablet(tablet_id());
527
0
}
528
529
0
void CloudTablet::recycle_cached_data(const std::vector<RowsetSharedPtr>& rowsets) {
530
0
    for (const auto& rs : rowsets) {
531
        // rowsets and tablet._rs_version_map each hold a rowset shared_ptr, so at this point, the reference count of the shared_ptr is at least 2.
532
0
        if (rs.use_count() > 2) {
533
0
            LOG(WARNING) << "Rowset " << rs->rowset_id().to_string() << " has " << rs.use_count()
534
0
                         << " references. File Cache won't be recycled when query is using it.";
535
0
            return;
536
0
        }
537
0
        rs->clear_cache();
538
0
    }
539
0
}
540
541
void CloudTablet::reset_approximate_stats(int64_t num_rowsets, int64_t num_segments,
542
0
                                          int64_t num_rows, int64_t data_size) {
543
0
    _approximate_num_rowsets.store(num_rowsets, std::memory_order_relaxed);
544
0
    _approximate_num_segments.store(num_segments, std::memory_order_relaxed);
545
0
    _approximate_num_rows.store(num_rows, std::memory_order_relaxed);
546
0
    _approximate_data_size.store(data_size, std::memory_order_relaxed);
547
0
    int64_t cumu_num_deltas = 0;
548
0
    int64_t cumu_num_rowsets = 0;
549
0
    auto cp = _cumulative_point.load(std::memory_order_relaxed);
550
0
    for (auto& [v, r] : _rs_version_map) {
551
0
        if (v.second < cp) {
552
0
            continue;
553
0
        }
554
555
0
        cumu_num_deltas += r->is_segments_overlapping() ? r->num_segments() : 1;
556
0
        ++cumu_num_rowsets;
557
0
    }
558
0
    _approximate_cumu_num_rowsets.store(cumu_num_rowsets, std::memory_order_relaxed);
559
0
    _approximate_cumu_num_deltas.store(cumu_num_deltas, std::memory_order_relaxed);
560
0
}
561
562
Result<std::unique_ptr<RowsetWriter>> CloudTablet::create_rowset_writer(
563
0
        RowsetWriterContext& context, bool vertical) {
564
0
    context.rowset_id = _engine.next_rowset_id();
565
    // FIXME(plat1ko): Seems `tablet_id` and `index_id` has been set repeatedly
566
0
    context.tablet_id = tablet_id();
567
0
    context.index_id = index_id();
568
0
    context.partition_id = partition_id();
569
0
    context.enable_unique_key_merge_on_write = enable_unique_key_merge_on_write();
570
0
    return RowsetFactory::create_rowset_writer(_engine, context, vertical);
571
0
}
572
573
// create a rowset writer with rowset_id and seg_id
574
// after writer, merge this transient rowset with original rowset
575
Result<std::unique_ptr<RowsetWriter>> CloudTablet::create_transient_rowset_writer(
576
        const Rowset& rowset, std::shared_ptr<PartialUpdateInfo> partial_update_info,
577
0
        int64_t txn_expiration) {
578
0
    if (rowset.rowset_meta_state() != RowsetStatePB::BEGIN_PARTIAL_UPDATE &&
579
0
        rowset.rowset_meta_state() != RowsetStatePB::COMMITTED) [[unlikely]] {
580
0
        auto msg = fmt::format(
581
0
                "wrong rowset state when create_transient_rowset_writer, rowset state should be "
582
0
                "BEGIN_PARTIAL_UPDATE or COMMITTED, but found {}, rowset_id={}, tablet_id={}",
583
0
                RowsetStatePB_Name(rowset.rowset_meta_state()), rowset.rowset_id().to_string(),
584
0
                tablet_id());
585
        // see `CloudRowsetWriter::build` for detail.
586
        // if this is in a retry task, the rowset state may have been changed to RowsetStatePB::COMMITTED
587
        // in `RowsetMeta::merge_rowset_meta()` in previous trials.
588
0
        LOG(WARNING) << msg;
589
0
        DCHECK(false) << msg;
590
0
    }
591
0
    RowsetWriterContext context;
592
0
    context.rowset_state = PREPARED;
593
0
    context.segments_overlap = OVERLAPPING;
594
    // During a partial update, the extracted columns of a variant should not be included in the tablet schema.
595
    // This is because the partial update for a variant needs to ignore the extracted columns.
596
    // Otherwise, the schema types in different rowsets might be inconsistent. When performing a partial update,
597
    // the complete variant is constructed by reading all the sub-columns of the variant.
598
0
    context.tablet_schema = rowset.tablet_schema()->copy_without_variant_extracted_columns();
599
0
    context.newest_write_timestamp = UnixSeconds();
600
0
    context.tablet_id = table_id();
601
0
    context.enable_segcompaction = false;
602
0
    context.write_type = DataWriteType::TYPE_DIRECT;
603
0
    context.partial_update_info = std::move(partial_update_info);
604
0
    context.is_transient_rowset_writer = true;
605
0
    context.rowset_id = rowset.rowset_id();
606
0
    context.tablet_id = tablet_id();
607
0
    context.index_id = index_id();
608
0
    context.partition_id = partition_id();
609
0
    context.enable_unique_key_merge_on_write = enable_unique_key_merge_on_write();
610
0
    context.txn_expiration = txn_expiration;
611
612
0
    auto storage_resource = rowset.rowset_meta()->remote_storage_resource();
613
0
    if (!storage_resource) {
614
0
        return ResultError(std::move(storage_resource.error()));
615
0
    }
616
617
0
    context.storage_resource = *storage_resource.value();
618
619
0
    return RowsetFactory::create_rowset_writer(_engine, context, false)
620
0
            .transform([&](auto&& writer) {
621
0
                writer->set_segment_start_id(cast_set<int32_t>(rowset.num_segments()));
622
0
                return writer;
623
0
            });
624
0
}
625
626
3
int64_t CloudTablet::get_cloud_base_compaction_score() const {
627
3
    if (_tablet_meta->compaction_policy() == CUMULATIVE_TIME_SERIES_POLICY) {
628
0
        bool has_delete = false;
629
0
        int64_t point = cumulative_layer_point();
630
0
        std::shared_lock<std::shared_mutex> rlock(_meta_lock);
631
0
        for (const auto& rs_meta : _tablet_meta->all_rs_metas()) {
632
0
            if (rs_meta->start_version() >= point) {
633
0
                continue;
634
0
            }
635
0
            if (rs_meta->has_delete_predicate()) {
636
0
                has_delete = true;
637
0
                break;
638
0
            }
639
0
        }
640
0
        if (!has_delete) {
641
0
            return 0;
642
0
        }
643
0
    }
644
645
3
    return _approximate_num_rowsets.load(std::memory_order_relaxed) -
646
3
           _approximate_cumu_num_rowsets.load(std::memory_order_relaxed);
647
3
}
648
649
1
int64_t CloudTablet::get_cloud_cumu_compaction_score() const {
650
    // TODO(plat1ko): Propose an algorithm that considers tablet's key type, number of delete rowsets,
651
    //  number of tablet versions simultaneously.
652
1
    return _approximate_cumu_num_deltas.load(std::memory_order_relaxed);
653
1
}
654
655
// return a json string to show the compaction status of this tablet
656
0
void CloudTablet::get_compaction_status(std::string* json_result) {
657
0
    rapidjson::Document root;
658
0
    root.SetObject();
659
660
0
    rapidjson::Document path_arr;
661
0
    path_arr.SetArray();
662
663
0
    std::vector<RowsetSharedPtr> rowsets;
664
0
    std::vector<RowsetSharedPtr> stale_rowsets;
665
0
    {
666
0
        std::shared_lock rdlock(_meta_lock);
667
0
        rowsets.reserve(_rs_version_map.size());
668
0
        for (auto& it : _rs_version_map) {
669
0
            rowsets.push_back(it.second);
670
0
        }
671
0
        stale_rowsets.reserve(_stale_rs_version_map.size());
672
0
        for (auto& it : _stale_rs_version_map) {
673
0
            stale_rowsets.push_back(it.second);
674
0
        }
675
0
    }
676
0
    std::sort(rowsets.begin(), rowsets.end(), Rowset::comparator);
677
0
    std::sort(stale_rowsets.begin(), stale_rowsets.end(), Rowset::comparator);
678
679
    // get snapshot version path json_doc
680
0
    _timestamped_version_tracker.get_stale_version_path_json_doc(path_arr);
681
0
    root.AddMember("cumulative point", _cumulative_point.load(), root.GetAllocator());
682
0
    rapidjson::Value cumu_value;
683
0
    std::string format_str = ToStringFromUnixMillis(_last_cumu_compaction_failure_millis.load());
684
0
    cumu_value.SetString(format_str.c_str(), cast_set<uint>(format_str.length()),
685
0
                         root.GetAllocator());
686
0
    root.AddMember("last cumulative failure time", cumu_value, root.GetAllocator());
687
0
    rapidjson::Value base_value;
688
0
    format_str = ToStringFromUnixMillis(_last_base_compaction_failure_millis.load());
689
0
    base_value.SetString(format_str.c_str(), cast_set<uint>(format_str.length()),
690
0
                         root.GetAllocator());
691
0
    root.AddMember("last base failure time", base_value, root.GetAllocator());
692
0
    rapidjson::Value full_value;
693
0
    format_str = ToStringFromUnixMillis(_last_full_compaction_failure_millis.load());
694
0
    full_value.SetString(format_str.c_str(), cast_set<uint>(format_str.length()),
695
0
                         root.GetAllocator());
696
0
    root.AddMember("last full failure time", full_value, root.GetAllocator());
697
0
    rapidjson::Value cumu_success_value;
698
0
    format_str = ToStringFromUnixMillis(_last_cumu_compaction_success_millis.load());
699
0
    cumu_success_value.SetString(format_str.c_str(), cast_set<uint>(format_str.length()),
700
0
                                 root.GetAllocator());
701
0
    root.AddMember("last cumulative success time", cumu_success_value, root.GetAllocator());
702
0
    rapidjson::Value base_success_value;
703
0
    format_str = ToStringFromUnixMillis(_last_base_compaction_success_millis.load());
704
0
    base_success_value.SetString(format_str.c_str(), cast_set<uint>(format_str.length()),
705
0
                                 root.GetAllocator());
706
0
    root.AddMember("last base success time", base_success_value, root.GetAllocator());
707
0
    rapidjson::Value full_success_value;
708
0
    format_str = ToStringFromUnixMillis(_last_full_compaction_success_millis.load());
709
0
    full_success_value.SetString(format_str.c_str(), cast_set<uint>(format_str.length()),
710
0
                                 root.GetAllocator());
711
0
    root.AddMember("last full success time", full_success_value, root.GetAllocator());
712
0
    rapidjson::Value cumu_schedule_value;
713
0
    format_str = ToStringFromUnixMillis(_last_cumu_compaction_schedule_millis.load());
714
0
    cumu_schedule_value.SetString(format_str.c_str(), cast_set<uint>(format_str.length()),
715
0
                                  root.GetAllocator());
716
0
    root.AddMember("last cumulative schedule time", cumu_schedule_value, root.GetAllocator());
717
0
    rapidjson::Value base_schedule_value;
718
0
    format_str = ToStringFromUnixMillis(_last_base_compaction_schedule_millis.load());
719
0
    base_schedule_value.SetString(format_str.c_str(), cast_set<uint>(format_str.length()),
720
0
                                  root.GetAllocator());
721
0
    root.AddMember("last base schedule time", base_schedule_value, root.GetAllocator());
722
0
    rapidjson::Value full_schedule_value;
723
0
    format_str = ToStringFromUnixMillis(_last_full_compaction_schedule_millis.load());
724
0
    full_schedule_value.SetString(format_str.c_str(), cast_set<uint>(format_str.length()),
725
0
                                  root.GetAllocator());
726
0
    root.AddMember("last full schedule time", full_schedule_value, root.GetAllocator());
727
0
    rapidjson::Value cumu_compaction_status_value;
728
0
    cumu_compaction_status_value.SetString(_last_cumu_compaction_status.c_str(),
729
0
                                           cast_set<uint>(_last_cumu_compaction_status.length()),
730
0
                                           root.GetAllocator());
731
0
    root.AddMember("last cumulative status", cumu_compaction_status_value, root.GetAllocator());
732
0
    rapidjson::Value base_compaction_status_value;
733
0
    base_compaction_status_value.SetString(_last_base_compaction_status.c_str(),
734
0
                                           cast_set<uint>(_last_base_compaction_status.length()),
735
0
                                           root.GetAllocator());
736
0
    root.AddMember("last base status", base_compaction_status_value, root.GetAllocator());
737
0
    rapidjson::Value full_compaction_status_value;
738
0
    full_compaction_status_value.SetString(_last_full_compaction_status.c_str(),
739
0
                                           cast_set<uint>(_last_full_compaction_status.length()),
740
0
                                           root.GetAllocator());
741
0
    root.AddMember("last full status", full_compaction_status_value, root.GetAllocator());
742
743
    // print all rowsets' version as an array
744
0
    rapidjson::Document versions_arr;
745
0
    rapidjson::Document missing_versions_arr;
746
0
    versions_arr.SetArray();
747
0
    missing_versions_arr.SetArray();
748
0
    int64_t last_version = -1;
749
0
    for (auto& rowset : rowsets) {
750
0
        const Version& ver = rowset->version();
751
0
        if (ver.first != last_version + 1) {
752
0
            rapidjson::Value miss_value;
753
0
            miss_value.SetString(fmt::format("[{}-{}]", last_version + 1, ver.first - 1).c_str(),
754
0
                                 missing_versions_arr.GetAllocator());
755
0
            missing_versions_arr.PushBack(miss_value, missing_versions_arr.GetAllocator());
756
0
        }
757
0
        rapidjson::Value value;
758
0
        std::string version_str = rowset->get_rowset_info_str();
759
0
        value.SetString(version_str.c_str(), cast_set<uint32_t>(version_str.length()),
760
0
                        versions_arr.GetAllocator());
761
0
        versions_arr.PushBack(value, versions_arr.GetAllocator());
762
0
        last_version = ver.second;
763
0
    }
764
0
    root.AddMember("rowsets", versions_arr, root.GetAllocator());
765
0
    root.AddMember("missing_rowsets", missing_versions_arr, root.GetAllocator());
766
767
    // print all stale rowsets' version as an array
768
0
    rapidjson::Document stale_versions_arr;
769
0
    stale_versions_arr.SetArray();
770
0
    for (auto& rowset : stale_rowsets) {
771
0
        rapidjson::Value value;
772
0
        std::string version_str = rowset->get_rowset_info_str();
773
0
        value.SetString(version_str.c_str(), cast_set<uint32_t>(version_str.length()),
774
0
                        stale_versions_arr.GetAllocator());
775
0
        stale_versions_arr.PushBack(value, stale_versions_arr.GetAllocator());
776
0
    }
777
0
    root.AddMember("stale_rowsets", stale_versions_arr, root.GetAllocator());
778
779
    // add stale version rowsets
780
0
    root.AddMember("stale version path", path_arr, root.GetAllocator());
781
782
    // to json string
783
0
    rapidjson::StringBuffer strbuf;
784
0
    rapidjson::PrettyWriter<rapidjson::StringBuffer> writer(strbuf);
785
0
    root.Accept(writer);
786
0
    *json_result = std::string(strbuf.GetString());
787
0
}
788
789
0
void CloudTablet::set_cumulative_layer_point(int64_t new_point) {
790
0
    if (new_point == Tablet::K_INVALID_CUMULATIVE_POINT || new_point >= _cumulative_point) {
791
0
        _cumulative_point = new_point;
792
0
        return;
793
0
    }
794
    // cumulative point should only be reset to -1, or be increased
795
    // FIXME: could happen in currently unresolved race conditions
796
0
    LOG(WARNING) << "Unexpected cumulative point: " << new_point
797
0
                 << ", origin: " << _cumulative_point.load();
798
0
}
799
800
0
std::vector<RowsetSharedPtr> CloudTablet::pick_candidate_rowsets_to_base_compaction() {
801
0
    std::vector<RowsetSharedPtr> candidate_rowsets;
802
0
    {
803
0
        std::shared_lock rlock(_meta_lock);
804
0
        for (const auto& [version, rs] : _rs_version_map) {
805
0
            if (version.first != 0 && version.first < _cumulative_point &&
806
0
                (_alter_version == -1 || version.second <= _alter_version)) {
807
0
                candidate_rowsets.push_back(rs);
808
0
            }
809
0
        }
810
0
    }
811
0
    std::sort(candidate_rowsets.begin(), candidate_rowsets.end(), Rowset::comparator);
812
0
    return candidate_rowsets;
813
0
}
814
815
0
std::vector<RowsetSharedPtr> CloudTablet::pick_candidate_rowsets_to_full_compaction() {
816
0
    std::vector<RowsetSharedPtr> candidate_rowsets;
817
0
    {
818
0
        std::shared_lock rlock(_meta_lock);
819
0
        for (auto& [v, rs] : _rs_version_map) {
820
            // MUST NOT compact rowset [0-1] for some historical reasons (see cloud_schema_change)
821
0
            if (v.first != 0) {
822
0
                candidate_rowsets.push_back(rs);
823
0
            }
824
0
        }
825
0
    }
826
0
    std::sort(candidate_rowsets.begin(), candidate_rowsets.end(), Rowset::comparator);
827
0
    return candidate_rowsets;
828
0
}
829
830
0
CalcDeleteBitmapExecutor* CloudTablet::calc_delete_bitmap_executor() {
831
0
    return _engine.calc_delete_bitmap_executor();
832
0
}
833
834
Status CloudTablet::save_delete_bitmap(const TabletTxnInfo* txn_info, int64_t txn_id,
835
                                       DeleteBitmapPtr delete_bitmap, RowsetWriter* rowset_writer,
836
                                       const RowsetIdUnorderedSet& cur_rowset_ids, int64_t lock_id,
837
0
                                       int64_t next_visible_version) {
838
0
    RowsetSharedPtr rowset = txn_info->rowset;
839
0
    int64_t cur_version = rowset->start_version();
840
    // update delete bitmap info, in order to avoid recalculation when trying again
841
0
    RETURN_IF_ERROR(_engine.txn_delete_bitmap_cache().update_tablet_txn_info(
842
0
            txn_id, tablet_id(), delete_bitmap, cur_rowset_ids, PublishStatus::PREPARE));
843
844
0
    if (txn_info->partial_update_info && txn_info->partial_update_info->is_partial_update() &&
845
0
        rowset_writer->num_rows() > 0) {
846
0
        DBUG_EXECUTE_IF("CloudTablet::save_delete_bitmap.update_tmp_rowset.error", {
847
0
            return Status::InternalError<false>("injected update_tmp_rowset error.");
848
0
        });
849
0
        const auto& rowset_meta = rowset->rowset_meta();
850
0
        RETURN_IF_ERROR(_engine.meta_mgr().update_tmp_rowset(*rowset_meta));
851
0
    }
852
853
0
    RETURN_IF_ERROR(save_delete_bitmap_to_ms(cur_version, txn_id, delete_bitmap, lock_id,
854
0
                                             next_visible_version));
855
856
    // store the delete bitmap with sentinel marks in txn_delete_bitmap_cache because if the txn is retried for some reason,
857
    // it will use the delete bitmap from txn_delete_bitmap_cache when re-calculating the delete bitmap, during which it will do
858
    // delete bitmap correctness check. If we store the new_delete_bitmap, the delete bitmap correctness check will fail
859
0
    RETURN_IF_ERROR(_engine.txn_delete_bitmap_cache().update_tablet_txn_info(
860
0
            txn_id, tablet_id(), delete_bitmap, cur_rowset_ids, PublishStatus::SUCCEED,
861
0
            txn_info->publish_info));
862
863
0
    DBUG_EXECUTE_IF("CloudTablet::save_delete_bitmap.enable_sleep", {
864
0
        auto sleep_sec = dp->param<int>("sleep", 5);
865
0
        std::this_thread::sleep_for(std::chrono::seconds(sleep_sec));
866
0
    });
867
868
0
    DBUG_EXECUTE_IF("CloudTablet::save_delete_bitmap.injected_error", {
869
0
        auto retry = dp->param<bool>("retry", false);
870
0
        auto sleep_sec = dp->param<int>("sleep", 0);
871
0
        std::this_thread::sleep_for(std::chrono::seconds(sleep_sec));
872
0
        if (retry) { // return DELETE_BITMAP_LOCK_ERROR to let it retry
873
0
            return Status::Error<ErrorCode::DELETE_BITMAP_LOCK_ERROR>(
874
0
                    "injected DELETE_BITMAP_LOCK_ERROR");
875
0
        } else {
876
0
            return Status::InternalError<false>("injected non-retryable error");
877
0
        }
878
0
    });
879
880
0
    return Status::OK();
881
0
}
882
883
Status CloudTablet::save_delete_bitmap_to_ms(int64_t cur_version, int64_t txn_id,
884
                                             DeleteBitmapPtr delete_bitmap, int64_t lock_id,
885
0
                                             int64_t next_visible_version) {
886
0
    DeleteBitmapPtr new_delete_bitmap = std::make_shared<DeleteBitmap>(tablet_id());
887
0
    for (auto iter = delete_bitmap->delete_bitmap.begin();
888
0
         iter != delete_bitmap->delete_bitmap.end(); ++iter) {
889
        // skip sentinel mark, which is used for delete bitmap correctness check
890
0
        if (std::get<1>(iter->first) != DeleteBitmap::INVALID_SEGMENT_ID) {
891
0
            new_delete_bitmap->merge(
892
0
                    {std::get<0>(iter->first), std::get<1>(iter->first), cur_version},
893
0
                    iter->second);
894
0
        }
895
0
    }
896
    // lock_id != -1 means this is in an explict txn
897
0
    bool is_explicit_txn = (lock_id != -1);
898
0
    auto ms_lock_id = !is_explicit_txn ? txn_id : lock_id;
899
900
0
    RETURN_IF_ERROR(_engine.meta_mgr().update_delete_bitmap(*this, ms_lock_id, LOAD_INITIATOR_ID,
901
0
                                                            new_delete_bitmap.get(), txn_id,
902
0
                                                            is_explicit_txn, next_visible_version));
903
0
    return Status::OK();
904
0
}
905
906
0
Versions CloudTablet::calc_missed_versions(int64_t spec_version, Versions existing_versions) const {
907
0
    DCHECK(spec_version > 0) << "invalid spec_version: " << spec_version;
908
909
    // sort the existing versions in ascending order
910
0
    std::sort(existing_versions.begin(), existing_versions.end(),
911
0
              [](const Version& a, const Version& b) {
912
                  // simple because 2 versions are certainly not overlapping
913
0
                  return a.first < b.first;
914
0
              });
915
916
    // From the first version(=0), find the missing version until spec_version
917
0
    int64_t last_version = -1;
918
0
    Versions missed_versions;
919
0
    for (const Version& version : existing_versions) {
920
0
        if (version.first > last_version + 1) {
921
            // there is a hole between versions
922
0
            missed_versions.emplace_back(last_version + 1, std::min(version.first, spec_version));
923
0
        }
924
0
        last_version = version.second;
925
0
        if (last_version >= spec_version) {
926
0
            break;
927
0
        }
928
0
    }
929
0
    if (last_version < spec_version) {
930
        // there is a hole between the last version and the specificed version.
931
0
        missed_versions.emplace_back(last_version + 1, spec_version);
932
0
    }
933
0
    return missed_versions;
934
0
}
935
936
Status CloudTablet::calc_delete_bitmap_for_compaction(
937
        const std::vector<RowsetSharedPtr>& input_rowsets, const RowsetSharedPtr& output_rowset,
938
        const RowIdConversion& rowid_conversion, ReaderType compaction_type, int64_t merged_rows,
939
        int64_t filtered_rows, int64_t initiator, DeleteBitmapPtr& output_rowset_delete_bitmap,
940
0
        bool allow_delete_in_cumu_compaction, int64_t& get_delete_bitmap_lock_start_time) {
941
0
    output_rowset_delete_bitmap = std::make_shared<DeleteBitmap>(tablet_id());
942
0
    std::unique_ptr<RowLocationSet> missed_rows;
943
0
    if ((config::enable_missing_rows_correctness_check ||
944
0
         config::enable_mow_compaction_correctness_check_core ||
945
0
         config::enable_mow_compaction_correctness_check_fail) &&
946
0
        !allow_delete_in_cumu_compaction &&
947
0
        compaction_type == ReaderType::READER_CUMULATIVE_COMPACTION) {
948
0
        missed_rows = std::make_unique<RowLocationSet>();
949
0
        LOG(INFO) << "RowLocation Set inited succ for tablet:" << tablet_id();
950
0
    }
951
952
0
    std::unique_ptr<std::map<RowsetSharedPtr, RowLocationPairList>> location_map;
953
0
    if (config::enable_rowid_conversion_correctness_check &&
954
0
        tablet_schema()->cluster_key_uids().empty()) {
955
0
        location_map = std::make_unique<std::map<RowsetSharedPtr, RowLocationPairList>>();
956
0
        LOG(INFO) << "Location Map inited succ for tablet:" << tablet_id();
957
0
    }
958
959
    // 1. calc delete bitmap for historical data
960
0
    RETURN_IF_ERROR(_engine.meta_mgr().sync_tablet_rowsets(this));
961
0
    Version version = max_version();
962
0
    std::size_t missed_rows_size = 0;
963
0
    calc_compaction_output_rowset_delete_bitmap(
964
0
            input_rowsets, rowid_conversion, 0, version.second + 1, missed_rows.get(),
965
0
            location_map.get(), tablet_meta()->delete_bitmap(), output_rowset_delete_bitmap.get());
966
0
    if (missed_rows) {
967
0
        missed_rows_size = missed_rows->size();
968
0
        if (!allow_delete_in_cumu_compaction) {
969
0
            if (compaction_type == ReaderType::READER_CUMULATIVE_COMPACTION &&
970
0
                tablet_state() == TABLET_RUNNING) {
971
0
                if (merged_rows + filtered_rows >= 0 &&
972
0
                    merged_rows + filtered_rows != missed_rows_size) {
973
0
                    std::string err_msg = fmt::format(
974
0
                            "cumulative compaction: the merged rows({}), the filtered rows({}) is "
975
0
                            "not equal to missed rows({}) in rowid conversion, tablet_id: {}, "
976
0
                            "table_id:{}",
977
0
                            merged_rows, filtered_rows, missed_rows_size, tablet_id(), table_id());
978
0
                    LOG(WARNING) << err_msg;
979
0
                    if (config::enable_mow_compaction_correctness_check_core) {
980
0
                        CHECK(false) << err_msg;
981
0
                    } else if (config::enable_mow_compaction_correctness_check_fail) {
982
0
                        return Status::InternalError<false>(err_msg);
983
0
                    } else {
984
0
                        DCHECK(false) << err_msg;
985
0
                    }
986
0
                }
987
0
            }
988
0
        }
989
0
    }
990
0
    if (location_map) {
991
0
        RETURN_IF_ERROR(check_rowid_conversion(output_rowset, *location_map));
992
0
        location_map->clear();
993
0
    }
994
995
    // 2. calc delete bitmap for incremental data
996
0
    int64_t t1 = MonotonicMicros();
997
0
    RETURN_IF_ERROR(_engine.meta_mgr().get_delete_bitmap_update_lock(
998
0
            *this, COMPACTION_DELETE_BITMAP_LOCK_ID, initiator));
999
0
    int64_t t2 = MonotonicMicros();
1000
0
    if (compaction_type == ReaderType::READER_CUMULATIVE_COMPACTION) {
1001
0
        g_cu_compaction_get_delete_bitmap_lock_time_ms << (t2 - t1) / 1000;
1002
0
    } else if (compaction_type == ReaderType::READER_BASE_COMPACTION) {
1003
0
        g_base_compaction_get_delete_bitmap_lock_time_ms << (t2 - t1) / 1000;
1004
0
    }
1005
0
    get_delete_bitmap_lock_start_time = t2;
1006
0
    RETURN_IF_ERROR(_engine.meta_mgr().sync_tablet_rowsets(this));
1007
0
    int64_t t3 = MonotonicMicros();
1008
1009
0
    calc_compaction_output_rowset_delete_bitmap(
1010
0
            input_rowsets, rowid_conversion, version.second, UINT64_MAX, missed_rows.get(),
1011
0
            location_map.get(), tablet_meta()->delete_bitmap(), output_rowset_delete_bitmap.get());
1012
0
    int64_t t4 = MonotonicMicros();
1013
0
    if (location_map) {
1014
0
        RETURN_IF_ERROR(check_rowid_conversion(output_rowset, *location_map));
1015
0
    }
1016
0
    int64_t t5 = MonotonicMicros();
1017
0
    if (missed_rows) {
1018
0
        DCHECK_EQ(missed_rows->size(), missed_rows_size);
1019
0
        if (missed_rows->size() != missed_rows_size) {
1020
0
            LOG(WARNING) << "missed rows don't match, before: " << missed_rows_size
1021
0
                         << " after: " << missed_rows->size();
1022
0
        }
1023
0
    }
1024
1025
    // 3. store delete bitmap
1026
0
    auto st = _engine.meta_mgr().update_delete_bitmap(*this, -1, initiator,
1027
0
                                                      output_rowset_delete_bitmap.get());
1028
0
    int64_t t6 = MonotonicMicros();
1029
0
    LOG(INFO) << "calc_delete_bitmap_for_compaction, tablet_id=" << tablet_id()
1030
0
              << ", get lock cost " << (t2 - t1) << " us, sync rowsets cost " << (t3 - t2)
1031
0
              << " us, calc delete bitmap cost " << (t4 - t3) << " us, check rowid conversion cost "
1032
0
              << (t5 - t4) << " us, store delete bitmap cost " << (t6 - t5)
1033
0
              << " us, st=" << st.to_string();
1034
0
    return st;
1035
0
}
1036
1037
void CloudTablet::agg_delete_bitmap_for_compaction(
1038
        int64_t start_version, int64_t end_version, const std::vector<RowsetSharedPtr>& pre_rowsets,
1039
        DeleteBitmapPtr& new_delete_bitmap,
1040
0
        std::map<std::string, int64_t>& pre_rowset_to_versions) {
1041
0
    for (auto& rowset : pre_rowsets) {
1042
0
        for (uint32_t seg_id = 0; seg_id < rowset->num_segments(); ++seg_id) {
1043
0
            auto d = tablet_meta()->delete_bitmap().get_agg_without_cache(
1044
0
                    {rowset->rowset_id(), seg_id, end_version}, start_version);
1045
0
            if (d->isEmpty()) {
1046
0
                continue;
1047
0
            }
1048
0
            VLOG_DEBUG << "agg delete bitmap for tablet_id=" << tablet_id()
1049
0
                       << ", rowset_id=" << rowset->rowset_id() << ", seg_id=" << seg_id
1050
0
                       << ", rowset_version=" << rowset->version().to_string()
1051
0
                       << ". compaction start_version=" << start_version
1052
0
                       << ", end_version=" << end_version
1053
0
                       << ". delete_bitmap cardinality=" << d->cardinality();
1054
0
            DeleteBitmap::BitmapKey end_key {rowset->rowset_id(), seg_id, end_version};
1055
0
            new_delete_bitmap->set(end_key, *d);
1056
0
            pre_rowset_to_versions[rowset->rowset_id().to_string()] = rowset->version().second;
1057
0
        }
1058
0
    }
1059
0
}
1060
1061
0
Status CloudTablet::sync_meta() {
1062
0
    if (!config::enable_file_cache) {
1063
0
        return Status::OK();
1064
0
    }
1065
1066
0
    TabletMetaSharedPtr tablet_meta;
1067
0
    auto st = _engine.meta_mgr().get_tablet_meta(tablet_id(), &tablet_meta);
1068
0
    if (!st.ok()) {
1069
0
        if (st.is<ErrorCode::NOT_FOUND>()) {
1070
0
            clear_cache();
1071
0
        }
1072
0
        return st;
1073
0
    }
1074
1075
0
    auto new_ttl_seconds = tablet_meta->ttl_seconds();
1076
0
    if (_tablet_meta->ttl_seconds() != new_ttl_seconds) {
1077
0
        _tablet_meta->set_ttl_seconds(new_ttl_seconds);
1078
0
        int64_t cur_time = UnixSeconds();
1079
0
        std::shared_lock rlock(_meta_lock);
1080
0
        for (auto& [_, rs] : _rs_version_map) {
1081
0
            for (int seg_id = 0; seg_id < rs->num_segments(); ++seg_id) {
1082
0
                int64_t new_expiration_time =
1083
0
                        new_ttl_seconds + rs->rowset_meta()->newest_write_timestamp();
1084
0
                new_expiration_time = new_expiration_time > cur_time ? new_expiration_time : 0;
1085
0
                auto file_key = Segment::file_cache_key(rs->rowset_id().to_string(), seg_id);
1086
0
                auto* file_cache = io::FileCacheFactory::instance()->get_by_path(file_key);
1087
0
                file_cache->modify_expiration_time(file_key, new_expiration_time);
1088
0
            }
1089
0
        }
1090
0
    }
1091
1092
0
    auto new_compaction_policy = tablet_meta->compaction_policy();
1093
0
    if (_tablet_meta->compaction_policy() != new_compaction_policy) {
1094
0
        _tablet_meta->set_compaction_policy(new_compaction_policy);
1095
0
    }
1096
0
    auto new_time_series_compaction_goal_size_mbytes =
1097
0
            tablet_meta->time_series_compaction_goal_size_mbytes();
1098
0
    if (_tablet_meta->time_series_compaction_goal_size_mbytes() !=
1099
0
        new_time_series_compaction_goal_size_mbytes) {
1100
0
        _tablet_meta->set_time_series_compaction_goal_size_mbytes(
1101
0
                new_time_series_compaction_goal_size_mbytes);
1102
0
    }
1103
0
    auto new_time_series_compaction_file_count_threshold =
1104
0
            tablet_meta->time_series_compaction_file_count_threshold();
1105
0
    if (_tablet_meta->time_series_compaction_file_count_threshold() !=
1106
0
        new_time_series_compaction_file_count_threshold) {
1107
0
        _tablet_meta->set_time_series_compaction_file_count_threshold(
1108
0
                new_time_series_compaction_file_count_threshold);
1109
0
    }
1110
0
    auto new_time_series_compaction_time_threshold_seconds =
1111
0
            tablet_meta->time_series_compaction_time_threshold_seconds();
1112
0
    if (_tablet_meta->time_series_compaction_time_threshold_seconds() !=
1113
0
        new_time_series_compaction_time_threshold_seconds) {
1114
0
        _tablet_meta->set_time_series_compaction_time_threshold_seconds(
1115
0
                new_time_series_compaction_time_threshold_seconds);
1116
0
    }
1117
0
    auto new_time_series_compaction_empty_rowsets_threshold =
1118
0
            tablet_meta->time_series_compaction_empty_rowsets_threshold();
1119
0
    if (_tablet_meta->time_series_compaction_empty_rowsets_threshold() !=
1120
0
        new_time_series_compaction_empty_rowsets_threshold) {
1121
0
        _tablet_meta->set_time_series_compaction_empty_rowsets_threshold(
1122
0
                new_time_series_compaction_empty_rowsets_threshold);
1123
0
    }
1124
0
    auto new_time_series_compaction_level_threshold =
1125
0
            tablet_meta->time_series_compaction_level_threshold();
1126
0
    if (_tablet_meta->time_series_compaction_level_threshold() !=
1127
0
        new_time_series_compaction_level_threshold) {
1128
0
        _tablet_meta->set_time_series_compaction_level_threshold(
1129
0
                new_time_series_compaction_level_threshold);
1130
0
    }
1131
1132
0
    return Status::OK();
1133
0
}
1134
1135
0
void CloudTablet::build_tablet_report_info(TTabletInfo* tablet_info) {
1136
0
    std::shared_lock rdlock(_meta_lock);
1137
0
    tablet_info->__set_total_version_count(_tablet_meta->version_count());
1138
0
    tablet_info->__set_tablet_id(_tablet_meta->tablet_id());
1139
    // Currently, this information will not be used by the cloud report,
1140
    // but it may be used in the future.
1141
0
}
1142
1143
Status CloudTablet::check_delete_bitmap_cache(int64_t txn_id,
1144
0
                                              DeleteBitmap* expected_delete_bitmap) {
1145
0
    DeleteBitmapPtr cached_delete_bitmap;
1146
0
    CloudStorageEngine& engine = ExecEnv::GetInstance()->storage_engine().to_cloud();
1147
0
    Status st = engine.txn_delete_bitmap_cache().get_delete_bitmap(
1148
0
            txn_id, tablet_id(), &cached_delete_bitmap, nullptr, nullptr);
1149
0
    if (st.ok()) {
1150
0
        bool res = (expected_delete_bitmap->cardinality() == cached_delete_bitmap->cardinality());
1151
0
        auto msg = fmt::format(
1152
0
                "delete bitmap cache check failed, cur_cardinality={}, cached_cardinality={}"
1153
0
                "txn_id={}, tablet_id={}",
1154
0
                expected_delete_bitmap->cardinality(), cached_delete_bitmap->cardinality(), txn_id,
1155
0
                tablet_id());
1156
0
        if (!res) {
1157
0
            DCHECK(res) << msg;
1158
0
            return Status::InternalError<false>(msg);
1159
0
        }
1160
0
    }
1161
0
    return Status::OK();
1162
0
}
1163
1164
#include "common/compile_check_end.h"
1165
} // namespace doris