Coverage Report

Created: 2026-08-07 10:13

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
be/src/storage/snapshot/snapshot_manager.cpp
Line
Count
Source
1
// Licensed to the Apache Software Foundation (ASF) under one
2
// or more contributor license agreements.  See the NOTICE file
3
// distributed with this work for additional information
4
// regarding copyright ownership.  The ASF licenses this file
5
// to you under the Apache License, Version 2.0 (the
6
// "License"); you may not use this file except in compliance
7
// with the License.  You may obtain a copy of the License at
8
//
9
//   http://www.apache.org/licenses/LICENSE-2.0
10
//
11
// Unless required by applicable law or agreed to in writing,
12
// software distributed under the License is distributed on an
13
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14
// KIND, either express or implied.  See the License for the
15
// specific language governing permissions and limitations
16
// under the License.
17
18
#include "storage/snapshot/snapshot_manager.h"
19
20
#include <fmt/format.h>
21
#include <gen_cpp/AgentService_types.h>
22
#include <gen_cpp/Types_constants.h>
23
#include <gen_cpp/olap_file.pb.h>
24
#include <thrift/protocol/TDebugProtocol.h>
25
26
#include <algorithm>
27
#include <atomic>
28
#include <ctime>
29
#include <filesystem>
30
#include <list>
31
#include <map>
32
#include <new>
33
#include <ostream>
34
#include <set>
35
#include <shared_mutex>
36
#include <unordered_map>
37
#include <utility>
38
39
#include "common/config.h"
40
#include "common/logging.h"
41
#include "common/status.h"
42
#include "io/fs/local_file_system.h"
43
#include "runtime/memory/mem_tracker_limiter.h"
44
#include "runtime/thread_context.h"
45
#include "storage/data_dir.h"
46
#include "storage/olap_common.h"
47
#include "storage/olap_define.h"
48
#include "storage/pb_helper.h"
49
#include "storage/rowset/rowset.h"
50
#include "storage/rowset/rowset_factory.h"
51
#include "storage/rowset/rowset_meta.h"
52
#include "storage/rowset/rowset_writer.h"
53
#include "storage/rowset/rowset_writer_context.h"
54
#include "storage/storage_engine.h"
55
#include "storage/tablet/tablet.h"
56
#include "storage/tablet/tablet_manager.h"
57
#include "storage/tablet/tablet_meta.h"
58
#include "storage/tablet/tablet_schema.h"
59
#include "storage/tablet/tablet_schema_cache.h"
60
#include "storage/utils.h"
61
#include "util/uid_util.h"
62
63
using std::nothrow;
64
using std::string;
65
using std::stringstream;
66
using std::vector;
67
68
namespace doris {
69
using namespace ErrorCode;
70
71
816
LocalSnapshotLockGuard LocalSnapshotLock::acquire(const std::string& path) {
72
816
    std::unique_lock<std::mutex> l(_lock);
73
816
    auto& ctx = _local_snapshot_contexts[path];
74
816
    while (ctx._is_locked) {
75
0
        ctx._waiting_count++;
76
0
        ctx._cv.wait(l);
77
0
        ctx._waiting_count--;
78
0
    }
79
80
816
    ctx._is_locked = true;
81
816
    return {path};
82
816
}
83
84
816
void LocalSnapshotLock::release(const std::string& path) {
85
816
    std::lock_guard<std::mutex> l(_lock);
86
816
    auto iter = _local_snapshot_contexts.find(path);
87
816
    if (iter == _local_snapshot_contexts.end()) {
88
0
        return;
89
0
    }
90
91
816
    auto& ctx = iter->second;
92
816
    ctx._is_locked = false;
93
816
    if (ctx._waiting_count > 0) {
94
0
        ctx._cv.notify_one();
95
816
    } else {
96
816
        _local_snapshot_contexts.erase(iter);
97
816
    }
98
816
}
99
100
514
SnapshotManager::SnapshotManager(StorageEngine& engine) : _engine(engine) {
101
514
    _mem_tracker =
102
514
            MemTrackerLimiter::create_shared(MemTrackerLimiter::Type::OTHER, "SnapshotManager");
103
514
}
104
105
511
SnapshotManager::~SnapshotManager() = default;
106
107
Status SnapshotManager::make_snapshot(const TSnapshotRequest& request, string* snapshot_path,
108
323
                                      bool* allow_incremental_clone) {
109
323
    SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(_mem_tracker);
110
323
    Status res = Status::OK();
111
323
    if (snapshot_path == nullptr) {
112
1
        return Status::Error<INVALID_ARGUMENT>("output parameter cannot be null");
113
1
    }
114
115
322
    TabletSharedPtr target_tablet = _engine.tablet_manager()->get_tablet(request.tablet_id);
116
117
322
    DBUG_EXECUTE_IF("SnapshotManager::make_snapshot.inject_failure", { target_tablet = nullptr; })
118
119
322
    if (target_tablet == nullptr) {
120
0
        return Status::Error<TABLE_NOT_FOUND>("failed to get tablet. tablet={}", request.tablet_id);
121
0
    }
122
123
322
    TabletSharedPtr ref_tablet = target_tablet;
124
322
    if (request.__isset.ref_tablet_id) {
125
0
        int64_t ref_tablet_id = request.ref_tablet_id;
126
0
        TabletSharedPtr base_tablet = _engine.tablet_manager()->get_tablet(ref_tablet_id);
127
128
        // Some tasks, like medium migration, cause the target tablet and base tablet to stay on
129
        // different disks. In this case, we fall through to the normal restore path.
130
        //
131
        // Otherwise, we can directly link the rowset files from the base tablet to the target tablet.
132
0
        if (base_tablet != nullptr &&
133
0
            base_tablet->data_dir()->path() == target_tablet->data_dir()->path()) {
134
0
            ref_tablet = std::move(base_tablet);
135
0
        }
136
0
    }
137
138
322
    res = _create_snapshot_files(ref_tablet, target_tablet, request, snapshot_path,
139
322
                                 allow_incremental_clone);
140
141
322
    if (!res.ok()) {
142
0
        LOG(WARNING) << "failed to make snapshot. res=" << res << " tablet=" << request.tablet_id;
143
0
        return res;
144
0
    }
145
146
322
    LOG(INFO) << "success to make snapshot. path=['" << *snapshot_path << "']";
147
322
    return res;
148
322
}
149
150
321
Status SnapshotManager::release_snapshot(const string& snapshot_path) {
151
321
    auto local_snapshot_guard = LocalSnapshotLock::instance().acquire(snapshot_path);
152
153
    // If the requested snapshot_path is located in the root/snapshot folder, it is considered legal and can be deleted.
154
    // Otherwise, it is considered an illegal request and returns an error result.
155
321
    SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(_mem_tracker);
156
321
    auto stores = _engine.get_stores();
157
321
    for (auto* store : stores) {
158
320
        std::string abs_path;
159
320
        RETURN_IF_ERROR(io::global_local_filesystem()->canonicalize(store->path(), &abs_path));
160
320
        if (snapshot_path.compare(0, abs_path.size(), abs_path) == 0 &&
161
320
            snapshot_path.compare(abs_path.size() + 1, SNAPSHOT_PREFIX.size(), SNAPSHOT_PREFIX) ==
162
320
                    0) {
163
320
            RETURN_IF_ERROR(io::global_local_filesystem()->delete_directory(snapshot_path));
164
320
            LOG(INFO) << "success to release snapshot path. [path='" << snapshot_path << "']";
165
320
            return Status::OK();
166
320
        }
167
320
    }
168
169
1
    return Status::Error<CE_CMD_PARAMS_ERROR>("released snapshot path illegal. [path='{}']",
170
1
                                              snapshot_path);
171
321
}
172
173
Result<std::vector<PendingRowsetGuard>> SnapshotManager::convert_rowset_ids(
174
        const std::string& clone_dir, int64_t tablet_id, int64_t replica_id, int64_t table_id,
175
178
        int64_t partition_id, int32_t schema_hash) {
176
178
    SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(_mem_tracker);
177
178
    std::vector<PendingRowsetGuard> guards;
178
    // check clone dir existed
179
178
    bool exists = true;
180
178
    RETURN_IF_ERROR_RESULT(io::global_local_filesystem()->exists(clone_dir, &exists));
181
178
    if (!exists) {
182
1
        return unexpected(Status::Error<DIR_NOT_EXIST>(
183
1
                "clone dir not existed when convert rowsetids. clone_dir={}", clone_dir));
184
1
    }
185
186
177
    TabletSharedPtr target_tablet = _engine.tablet_manager()->get_tablet(tablet_id);
187
177
    TabletSchemaSPtr target_tablet_schema = nullptr;
188
177
    if (target_tablet != nullptr) {
189
176
        target_tablet_schema = std::make_shared<TabletSchema>();
190
176
        target_tablet_schema->copy_from(*target_tablet->tablet_schema());
191
176
    }
192
193
    // load original tablet meta
194
177
    auto cloned_meta_file = fmt::format("{}/{}.hdr", clone_dir, tablet_id);
195
177
    TabletMetaPB cloned_tablet_meta_pb;
196
177
    RETURN_IF_ERROR_RESULT(TabletMeta::load_from_file(cloned_meta_file, &cloned_tablet_meta_pb));
197
198
177
    TabletMetaPB new_tablet_meta_pb;
199
177
    new_tablet_meta_pb = cloned_tablet_meta_pb;
200
177
    new_tablet_meta_pb.clear_rs_metas();
201
    // inc_rs_meta is deprecated since 0.13.
202
    // keep this just for safety
203
177
    new_tablet_meta_pb.clear_inc_rs_metas();
204
177
    new_tablet_meta_pb.clear_stale_rs_metas();
205
206
    // should modify tablet id and schema hash because in restore process the tablet id is not
207
    // equal to tablet id in meta
208
177
    new_tablet_meta_pb.set_tablet_id(tablet_id);
209
177
    *new_tablet_meta_pb.mutable_tablet_uid() = TabletUid::gen_uid().to_proto();
210
177
    new_tablet_meta_pb.set_replica_id(replica_id);
211
177
    if (table_id > 0) {
212
173
        new_tablet_meta_pb.set_table_id(table_id);
213
173
    }
214
177
    if (partition_id != -1) {
215
177
        new_tablet_meta_pb.set_partition_id(partition_id);
216
177
    }
217
177
    new_tablet_meta_pb.set_schema_hash(schema_hash);
218
177
    TabletSchemaSPtr tablet_schema = std::make_shared<TabletSchema>();
219
177
    tablet_schema->init_from_pb(new_tablet_meta_pb.schema());
220
221
177
    std::unordered_map<Version, RowsetMetaPB*, HashOfVersion> rs_version_map;
222
177
    std::unordered_map<RowsetId, RowsetId> rowset_id_mapping;
223
177
    guards.reserve(cloned_tablet_meta_pb.rs_metas_size() +
224
177
                   cloned_tablet_meta_pb.stale_rs_metas_size());
225
481
    for (auto&& visible_rowset : cloned_tablet_meta_pb.rs_metas()) {
226
481
        RowsetMetaPB* rowset_meta = new_tablet_meta_pb.add_rs_metas();
227
481
        if (!visible_rowset.has_resource_id()) {
228
            // src be local rowset
229
481
            RowsetId rowset_id = _engine.next_rowset_id();
230
481
            guards.push_back(_engine.pending_local_rowsets().add(rowset_id));
231
481
            RETURN_IF_ERROR_RESULT(_rename_rowset_id(
232
481
                    visible_rowset, clone_dir, tablet_schema, rowset_id, rowset_meta,
233
481
                    new_tablet_meta_pb.enable_unique_key_merge_on_write()));
234
481
            RowsetId src_rs_id;
235
481
            if (visible_rowset.rowset_id() > 0) {
236
1
                src_rs_id.init(visible_rowset.rowset_id());
237
480
            } else {
238
480
                src_rs_id.init(visible_rowset.rowset_id_v2());
239
480
            }
240
481
            rowset_id_mapping[src_rs_id] = rowset_id;
241
481
            rowset_meta->set_source_rowset_id(visible_rowset.rowset_id_v2());
242
481
            rowset_meta->set_source_tablet_id(cloned_tablet_meta_pb.tablet_id());
243
481
        } else {
244
            // remote rowset
245
0
            *rowset_meta = visible_rowset;
246
0
        }
247
248
481
        rowset_meta->set_tablet_id(tablet_id);
249
481
        if (partition_id != -1) {
250
481
            rowset_meta->set_partition_id(partition_id);
251
481
        }
252
253
481
        if (rowset_meta->has_tablet_schema() && rowset_meta->tablet_schema().index_size() > 0) {
254
1
            RETURN_IF_ERROR_RESULT(
255
1
                    _rename_index_ids(*rowset_meta->mutable_tablet_schema(), target_tablet_schema));
256
1
        }
257
258
481
        Version rowset_version = {visible_rowset.start_version(), visible_rowset.end_version()};
259
481
        rs_version_map[rowset_version] = rowset_meta;
260
481
    }
261
262
177
    for (auto&& stale_rowset : cloned_tablet_meta_pb.stale_rs_metas()) {
263
0
        Version rowset_version = {stale_rowset.start_version(), stale_rowset.end_version()};
264
0
        auto exist_rs = rs_version_map.find(rowset_version);
265
0
        if (exist_rs != rs_version_map.end()) {
266
0
            continue;
267
0
        }
268
0
        RowsetMetaPB* rowset_meta = new_tablet_meta_pb.add_stale_rs_metas();
269
270
0
        if (!stale_rowset.has_resource_id()) {
271
            // src be local rowset
272
0
            RowsetId rowset_id = _engine.next_rowset_id();
273
0
            guards.push_back(_engine.pending_local_rowsets().add(rowset_id));
274
0
            RETURN_IF_ERROR_RESULT(_rename_rowset_id(
275
0
                    stale_rowset, clone_dir, tablet_schema, rowset_id, rowset_meta,
276
0
                    new_tablet_meta_pb.enable_unique_key_merge_on_write()));
277
0
            RowsetId src_rs_id;
278
0
            if (stale_rowset.rowset_id() > 0) {
279
0
                src_rs_id.init(stale_rowset.rowset_id());
280
0
            } else {
281
0
                src_rs_id.init(stale_rowset.rowset_id_v2());
282
0
            }
283
0
            rowset_id_mapping[src_rs_id] = rowset_id;
284
0
            rowset_meta->set_source_rowset_id(stale_rowset.rowset_id_v2());
285
0
            rowset_meta->set_source_tablet_id(cloned_tablet_meta_pb.tablet_id());
286
0
        } else {
287
            // remote rowset
288
0
            *rowset_meta = stale_rowset;
289
0
        }
290
291
0
        rowset_meta->set_tablet_id(tablet_id);
292
0
        if (partition_id != -1) {
293
0
            rowset_meta->set_partition_id(partition_id);
294
0
        }
295
296
0
        if (rowset_meta->has_tablet_schema() && rowset_meta->tablet_schema().index_size() > 0) {
297
0
            RETURN_IF_ERROR_RESULT(
298
0
                    _rename_index_ids(*rowset_meta->mutable_tablet_schema(), target_tablet_schema));
299
0
        }
300
0
    }
301
302
177
    if (new_tablet_meta_pb.schema().index_size() > 0) {
303
1
        RETURN_IF_ERROR_RESULT(
304
1
                _rename_index_ids(*new_tablet_meta_pb.mutable_schema(), target_tablet_schema));
305
1
    }
306
307
177
    if (!rowset_id_mapping.empty() && cloned_tablet_meta_pb.has_delete_bitmap()) {
308
32
        const auto& cloned_del_bitmap_pb = cloned_tablet_meta_pb.delete_bitmap();
309
32
        DeleteBitmapPB* new_del_bitmap_pb = new_tablet_meta_pb.mutable_delete_bitmap();
310
32
        int rst_ids_size = cloned_del_bitmap_pb.rowset_ids_size();
311
32
        for (int i = 0; i < rst_ids_size; ++i) {
312
0
            RowsetId rst_id;
313
0
            rst_id.init(cloned_del_bitmap_pb.rowset_ids(i));
314
            // It should not happen, if we can't convert some rowid in delete bitmap, the
315
            // data might be inconsist.
316
0
            CHECK(rowset_id_mapping.find(rst_id) != rowset_id_mapping.end())
317
0
                    << "can't find rowset_id " << rst_id.to_string() << " in convert_rowset_ids";
318
0
            new_del_bitmap_pb->set_rowset_ids(i, rowset_id_mapping[rst_id].to_string());
319
0
        }
320
32
    }
321
322
177
    RETURN_IF_ERROR_RESULT(TabletMeta::save(cloned_meta_file, new_tablet_meta_pb));
323
324
177
    return guards;
325
177
}
326
327
Status SnapshotManager::_rename_rowset_id(const RowsetMetaPB& rs_meta_pb,
328
                                          const std::string& new_tablet_path,
329
                                          TabletSchemaSPtr tablet_schema, const RowsetId& rowset_id,
330
                                          RowsetMetaPB* new_rs_meta_pb,
331
481
                                          bool enable_unique_key_merge_on_write) {
332
481
    Status st = Status::OK();
333
481
    RowsetMetaSharedPtr rowset_meta(new RowsetMeta());
334
481
    rowset_meta->init_from_pb(rs_meta_pb);
335
481
    RowsetSharedPtr org_rowset;
336
481
    RETURN_IF_ERROR(
337
481
            RowsetFactory::create_rowset(tablet_schema, new_tablet_path, rowset_meta, &org_rowset));
338
    // do not use cache to load index
339
    // because the index file may conflict
340
    // and the cached fd may be invalid
341
481
    RETURN_IF_ERROR(org_rowset->load(false));
342
481
    RowsetMetaSharedPtr org_rowset_meta = org_rowset->rowset_meta();
343
481
    RowsetWriterContext context;
344
481
    context.rowset_id = rowset_id;
345
481
    context.tablet_id = org_rowset_meta->tablet_id();
346
481
    context.partition_id = org_rowset_meta->partition_id();
347
481
    context.tablet_schema_hash = org_rowset_meta->tablet_schema_hash();
348
481
    context.rowset_type = org_rowset_meta->rowset_type();
349
481
    context.tablet_path = new_tablet_path;
350
481
    context.tablet_schema =
351
481
            org_rowset_meta->tablet_schema() ? org_rowset_meta->tablet_schema() : tablet_schema;
352
481
    context.rowset_state = org_rowset_meta->rowset_state();
353
481
    context.version = org_rowset_meta->version();
354
481
    context.newest_write_timestamp = org_rowset_meta->newest_write_timestamp();
355
    // keep segments_overlap same as origin rowset
356
481
    context.segments_overlap = rowset_meta->segments_overlap();
357
481
    context.compaction_level = org_rowset_meta->compaction_level();
358
481
    if (org_rowset_meta->is_row_binlog()) {
359
0
        context.write_binlog_opt().enable = true;
360
0
    }
361
    // propagate MOW flag so that non-MOW key-bounds aggregation is not applied
362
    // when restoring a MOW tablet's rowset
363
481
    context.enable_unique_key_merge_on_write = enable_unique_key_merge_on_write;
364
365
481
    auto rs_writer = DORIS_TRY(RowsetFactory::create_rowset_writer(_engine, context, false));
366
367
481
    st = rs_writer->add_rowset(org_rowset);
368
481
    if (!st.ok()) {
369
0
        LOG(WARNING) << "failed to add rowset "
370
0
                     << " id = " << org_rowset->rowset_id() << " to rowset " << rowset_id;
371
0
        return st;
372
0
    }
373
481
    RowsetSharedPtr new_rowset;
374
481
    RETURN_NOT_OK_STATUS_WITH_WARN(rs_writer->build(new_rowset),
375
481
                                   "failed to build rowset when rename rowset id");
376
481
    RETURN_IF_ERROR(new_rowset->load(false));
377
481
    new_rowset->rowset_meta()->to_rowset_pb(new_rs_meta_pb);
378
481
    RETURN_IF_ERROR(org_rowset->remove());
379
481
    return Status::OK();
380
481
}
381
382
Status SnapshotManager::_rename_index_ids(TabletSchemaPB& schema_pb,
383
2
                                          const TabletSchemaSPtr& tablet_schema) const {
384
2
    if (tablet_schema == nullptr) {
385
0
        return Status::OK();
386
0
    }
387
388
4
    for (int i = 0; i < schema_pb.index_size(); ++i) {
389
2
        TabletIndexPB* index_pb = schema_pb.mutable_index(i);
390
2
        for (int32_t col_unique_id : index_pb->col_unique_id()) {
391
2
            auto local_index = tablet_schema->get_index(col_unique_id, index_pb->index_type(),
392
2
                                                        index_pb->index_suffix_name());
393
2
            if (local_index) {
394
2
                if (index_pb->index_id() != local_index->index_id()) {
395
2
                    index_pb->set_index_id(local_index->index_id());
396
2
                }
397
2
                break;
398
2
            }
399
2
        }
400
2
    }
401
2
    return Status::OK();
402
2
}
403
404
// get snapshot path: curtime.seq.timeout
405
// eg: 20190819221234.3.86400
406
Status SnapshotManager::_calc_snapshot_id_path(const TabletSharedPtr& tablet, int64_t timeout_s,
407
322
                                               std::string* out_path) {
408
322
    Status res = Status::OK();
409
322
    if (out_path == nullptr) {
410
0
        return Status::Error<INVALID_ARGUMENT>("output parameter cannot be null");
411
0
    }
412
413
    // get current timestamp string
414
322
    string time_str;
415
322
    if ((res = gen_timestamp_string(&time_str)) != Status::OK()) {
416
0
        LOG(WARNING) << "failed to generate time_string when move file to trash."
417
0
                     << "err code=" << res;
418
0
        return res;
419
0
    }
420
421
322
    uint64_t sid = _snapshot_base_id.fetch_add(1, std::memory_order_relaxed) - 1;
422
322
    *out_path = fmt::format("{}/{}/{}.{}.{}", tablet->data_dir()->path(), SNAPSHOT_PREFIX, time_str,
423
322
                            sid, timeout_s);
424
322
    return res;
425
322
}
426
427
// prefix: /path/to/data/DATA_PREFIX/shard_id
428
// return: /path/to/data/DATA_PREFIX/shard_id/tablet_id/schema_hash
429
std::string SnapshotManager::get_schema_hash_full_path(const TabletSharedPtr& ref_tablet,
430
324
                                                       const std::string& prefix) {
431
324
    return fmt::format("{}/{}/{}", prefix, ref_tablet->tablet_id(), ref_tablet->schema_hash());
432
324
}
433
434
std::string SnapshotManager::_get_header_full_path(const TabletSharedPtr& ref_tablet,
435
320
                                                   const std::string& schema_hash_path) const {
436
320
    return fmt::format("{}/{}.hdr", schema_hash_path, ref_tablet->tablet_id());
437
320
}
438
439
std::string SnapshotManager::_get_json_header_full_path(const TabletSharedPtr& ref_tablet,
440
320
                                                        const std::string& schema_hash_path) const {
441
320
    return fmt::format("{}/{}.hdr.json", schema_hash_path, ref_tablet->tablet_id());
442
320
}
443
444
Status SnapshotManager::_link_index_and_data_files(
445
        const std::string& schema_hash_path, const TabletSharedPtr& ref_tablet,
446
0
        const std::vector<RowsetSharedPtr>& consistent_rowsets) {
447
0
    Status res = Status::OK();
448
0
    for (auto& rs : consistent_rowsets) {
449
0
        RETURN_IF_ERROR(rs->link_files_to(schema_hash_path, rs->rowset_id()));
450
0
    }
451
0
    return res;
452
0
}
453
454
Status SnapshotManager::_create_snapshot_files(const TabletSharedPtr& ref_tablet,
455
                                               const TabletSharedPtr& target_tablet,
456
                                               const TSnapshotRequest& request,
457
                                               string* snapshot_path,
458
322
                                               bool* allow_incremental_clone) {
459
322
    int32_t snapshot_version = request.preferred_snapshot_version;
460
322
    LOG(INFO) << "receive a make snapshot request"
461
322
              << ", request detail is " << apache::thrift::ThriftDebugString(request)
462
322
              << ", snapshot_version is " << snapshot_version;
463
322
    Status res = Status::OK();
464
322
    if (snapshot_path == nullptr) {
465
0
        return Status::Error<INVALID_ARGUMENT>("output parameter cannot be null");
466
0
    }
467
468
    // snapshot_id_path:
469
    //      /data/shard_id/tablet_id/snapshot/time_str/id.timeout/
470
322
    int64_t timeout_s = config::snapshot_expire_time_sec;
471
322
    if (request.__isset.timeout) {
472
320
        timeout_s = request.timeout;
473
320
    }
474
322
    std::string snapshot_id_path;
475
322
    res = _calc_snapshot_id_path(target_tablet, timeout_s, &snapshot_id_path);
476
322
    if (!res.ok()) {
477
0
        LOG(WARNING) << "failed to calc snapshot_id_path, tablet="
478
0
                     << target_tablet->data_dir()->path();
479
0
        return res;
480
0
    }
481
482
322
    bool need_ccr_binlog = request.__isset.is_copy_binlog ? request.is_copy_binlog : false;
483
484
    // schema_full_path_desc.filepath:
485
    //      /snapshot_id_path/tablet_id/schema_hash/
486
322
    auto schema_full_path = get_schema_hash_full_path(target_tablet, snapshot_id_path);
487
    // header_path:
488
    //      /schema_full_path/tablet_id.hdr
489
322
    auto header_path = _get_header_full_path(target_tablet, schema_full_path);
490
    //      /schema_full_path/tablet_id.hdr.json
491
322
    auto json_header_path = _get_json_header_full_path(target_tablet, schema_full_path);
492
322
    bool exists = true;
493
322
    RETURN_IF_ERROR(io::global_local_filesystem()->exists(schema_full_path, &exists));
494
322
    if (exists) {
495
0
        VLOG_TRACE << "remove the old schema_full_path." << schema_full_path;
496
0
        RETURN_IF_ERROR(io::global_local_filesystem()->delete_directory(schema_full_path));
497
0
    }
498
499
322
    RETURN_IF_ERROR(io::global_local_filesystem()->create_directory(schema_full_path));
500
322
    string snapshot_id;
501
322
    RETURN_IF_ERROR(io::global_local_filesystem()->canonicalize(snapshot_id_path, &snapshot_id));
502
503
322
    std::vector<RowsetSharedPtr> consistent_rowsets;
504
322
    do {
505
322
        TabletMetaSharedPtr new_tablet_meta(new (nothrow) TabletMeta());
506
322
        if (new_tablet_meta == nullptr) {
507
0
            res = Status::Error<MEM_ALLOC_FAILED>("fail to malloc TabletMeta.");
508
0
            break;
509
0
        }
510
322
        DeleteBitmap delete_bitmap_snapshot(new_tablet_meta->tablet_id());
511
512
        /// If set missing_version, try to get all missing version.
513
        /// If some of them not exist in tablet, we will fall back to
514
        /// make the full snapshot of the tablet.
515
322
        {
516
322
            std::shared_lock rdlock(ref_tablet->get_header_lock());
517
322
            if (ref_tablet->tablet_state() == TABLET_SHUTDOWN) {
518
0
                return Status::Aborted("tablet has shutdown");
519
0
            }
520
322
            bool is_single_rowset_clone =
521
322
                    (request.__isset.start_version && request.__isset.end_version);
522
322
            if (is_single_rowset_clone) {
523
0
                LOG(INFO) << "handle compaction clone make snapshot, tablet_id: "
524
0
                          << ref_tablet->tablet_id();
525
0
                Version version(request.start_version, request.end_version);
526
0
                const RowsetSharedPtr rowset = ref_tablet->get_rowset_by_version(version, false);
527
0
                if (rowset && rowset->is_local()) {
528
0
                    consistent_rowsets.push_back(rowset);
529
0
                } else {
530
0
                    LOG(WARNING) << "failed to find local version when do compaction snapshot. "
531
0
                                 << " tablet=" << request.tablet_id
532
0
                                 << " schema_hash=" << request.schema_hash
533
0
                                 << " version=" << version;
534
0
                    res = Status::InternalError(
535
0
                            "failed to find version when do compaction snapshot");
536
0
                    break;
537
0
                }
538
0
            }
539
            // be would definitely set it as true no matter has missed version or not
540
            // but it would take no effects on the following range loop
541
322
            if (!is_single_rowset_clone && request.__isset.missing_version) {
542
0
                for (int64_t missed_version : request.missing_version) {
543
0
                    Version version = {missed_version, missed_version};
544
                    // find rowset in both rs_meta and stale_rs_meta
545
0
                    const RowsetSharedPtr rowset = ref_tablet->get_rowset_by_version(version, true);
546
0
                    if (rowset != nullptr) {
547
0
                        if (!rowset->is_local()) {
548
                            // MUST make full snapshot to ensure `cooldown_meta_id` is consistent with the cooldowned rowsets after clone.
549
0
                            res = Status::Error<ErrorCode::INTERNAL_ERROR>(
550
0
                                    "missed version is a cooldowned rowset, must make full "
551
0
                                    "snapshot. missed_version={}, tablet_id={}",
552
0
                                    missed_version, ref_tablet->tablet_id());
553
0
                            break;
554
0
                        }
555
0
                        consistent_rowsets.push_back(rowset);
556
0
                    } else {
557
0
                        res = Status::InternalError(
558
0
                                "failed to find missed version when snapshot. tablet={}, "
559
0
                                "schema_hash={}, version={}",
560
0
                                request.tablet_id, request.schema_hash, version.to_string());
561
0
                        break;
562
0
                    }
563
0
                }
564
0
            }
565
566
322
            DBUG_EXECUTE_IF("SnapshotManager.create_snapshot_files.allow_inc_clone", {
567
322
                auto tablet_id = dp->param("tablet_id", 0);
568
322
                auto is_full_clone = dp->param("is_full_clone", false);
569
322
                if (ref_tablet->tablet_id() == tablet_id && is_full_clone) {
570
322
                    LOG(INFO) << "injected full clone for tabelt: " << tablet_id;
571
322
                    res = Status::InternalError("fault injection error");
572
322
                }
573
322
            });
574
575
            // be would definitely set it as true no matter has missed version or not, we could
576
            // just check whether the missed version is empty or not
577
322
            int64_t version = -1;
578
322
            if (!is_single_rowset_clone && (!res.ok() || request.missing_version.empty())) {
579
320
                if (!request.__isset.missing_version &&
580
320
                    ref_tablet->tablet_meta()->cooldown_meta_id().initialized()) {
581
0
                    LOG(WARNING) << "currently not support backup tablet with cooldowned remote "
582
0
                                    "data. tablet="
583
0
                                 << request.tablet_id;
584
0
                    return Status::NotSupported(
585
0
                            "currently not support backup tablet with cooldowned remote data");
586
0
                }
587
                /// not all missing versions are found, fall back to full snapshot.
588
320
                res = Status::OK();         // reset res
589
320
                consistent_rowsets.clear(); // reset vector
590
591
                // get latest version
592
320
                const RowsetSharedPtr last_version = ref_tablet->get_rowset_with_max_version();
593
320
                if (last_version == nullptr) {
594
0
                    res = Status::InternalError("tablet has not any version. path={}",
595
0
                                                ref_tablet->tablet_id());
596
0
                    break;
597
0
                }
598
                // get snapshot version, use request.version if specified
599
320
                version = last_version->end_version();
600
320
                if (request.__isset.version) {
601
148
                    if (last_version->end_version() < request.version) {
602
0
                        res = Status::Error<INVALID_ARGUMENT>(
603
0
                                "invalid make snapshot request. version={}, req_version={}",
604
0
                                last_version->version().to_string(), request.version);
605
0
                        break;
606
0
                    }
607
148
                    version = request.version;
608
148
                }
609
320
                if (ref_tablet->tablet_meta()->cooldown_meta_id().initialized()) {
610
                    // Tablet has cooldowned data, MUST pick consistent rowsets with continuous cooldowned version
611
                    // Get max cooldowned version
612
0
                    int64_t max_cooldowned_version = -1;
613
0
                    for (auto& [v, rs] : ref_tablet->rowset_map()) {
614
0
                        if (rs->is_local()) {
615
0
                            continue;
616
0
                        }
617
0
                        consistent_rowsets.push_back(rs);
618
0
                        max_cooldowned_version = std::max(max_cooldowned_version, v.second);
619
0
                    }
620
0
                    DCHECK_GE(max_cooldowned_version, 1) << "tablet_id=" << ref_tablet->tablet_id();
621
0
                    std::sort(consistent_rowsets.begin(), consistent_rowsets.end(),
622
0
                              Rowset::comparator);
623
0
                    res = check_version_continuity(consistent_rowsets);
624
0
                    if (res.ok() && max_cooldowned_version < version) {
625
                        // Pick consistent rowsets of remaining required version
626
0
                        auto ret = ref_tablet->capture_consistent_rowsets_unlocked(
627
0
                                {max_cooldowned_version + 1, version}, CaptureRowsetOps {});
628
0
                        if (ret) {
629
0
                            consistent_rowsets = std::move(ret->rowsets);
630
0
                        } else {
631
0
                            res = std::move(ret.error());
632
0
                        }
633
0
                    }
634
320
                } else {
635
320
                    auto ret = ref_tablet->capture_consistent_rowsets_unlocked(Version(0, version),
636
320
                                                                               CaptureRowsetOps {});
637
320
                    if (ret) {
638
320
                        consistent_rowsets = std::move(ret->rowsets);
639
320
                    } else {
640
0
                        res = std::move(ret.error());
641
0
                    }
642
320
                }
643
320
                if (!res.ok()) {
644
0
                    LOG(WARNING) << "fail to select versions to span. res=" << res;
645
0
                    break;
646
0
                }
647
320
                *allow_incremental_clone = false;
648
320
            } else {
649
2
                version = ref_tablet->max_version_unlocked();
650
2
                *allow_incremental_clone = true;
651
2
            }
652
653
            // copy the tablet meta to new_tablet_meta inside header lock
654
322
            CHECK(res.ok()) << res;
655
322
            ref_tablet->generate_tablet_meta_copy_unlocked(*new_tablet_meta, false);
656
            // The delete bitmap update operation and the add_inc_rowset operation is not atomic,
657
            // so delete bitmap may contains some data generated by invisible rowset, we should
658
            // get rid of these useless bitmaps when doing snapshot.
659
322
            if (ref_tablet->keys_type() == UNIQUE_KEYS &&
660
322
                ref_tablet->enable_unique_key_merge_on_write()) {
661
40
                delete_bitmap_snapshot =
662
40
                        ref_tablet->tablet_meta()->delete_bitmap().snapshot(version);
663
40
            }
664
322
        }
665
666
0
        std::vector<RowsetMetaSharedPtr> rs_metas;
667
504
        for (auto& rs : consistent_rowsets) {
668
504
            if (rs->is_local()) {
669
                // local rowset
670
504
                res = rs->link_files_to(schema_full_path, rs->rowset_id());
671
504
                if (!res.ok()) {
672
0
                    break;
673
0
                }
674
504
            }
675
504
            rs_metas.push_back(rs->rowset_meta());
676
504
            VLOG_NOTICE << "add rowset meta to clone list. "
677
2
                        << " start version " << rs->rowset_meta()->start_version()
678
2
                        << " end version " << rs->rowset_meta()->end_version() << " empty "
679
2
                        << rs->rowset_meta()->empty();
680
504
        }
681
322
        if (!res.ok()) {
682
0
            LOG(WARNING) << "fail to create hard link. path=" << snapshot_id_path
683
0
                         << " tablet=" << target_tablet->tablet_id()
684
0
                         << " ref tablet=" << ref_tablet->tablet_id();
685
0
            break;
686
0
        }
687
        // The inc_rs_metas is deprecated since Doris version 0.13.
688
        // Clear it for safety reason.
689
        // Whether it is incremental or full snapshot, rowset information is stored in rs_meta.
690
322
        new_tablet_meta->revise_rs_metas(std::move(rs_metas));
691
322
        if (ref_tablet->keys_type() == UNIQUE_KEYS &&
692
322
            ref_tablet->enable_unique_key_merge_on_write()) {
693
40
            new_tablet_meta->revise_delete_bitmap_unlocked(delete_bitmap_snapshot);
694
40
        }
695
696
322
        if (snapshot_version == g_Types_constants.TSNAPSHOT_REQ_VERSION2) {
697
320
            res = new_tablet_meta->save(header_path);
698
320
            if (res.ok() && request.__isset.is_copy_tablet_task && request.is_copy_tablet_task) {
699
0
                res = new_tablet_meta->save_as_json(json_header_path);
700
0
            }
701
320
        } else {
702
2
            res = Status::Error<INVALID_SNAPSHOT_VERSION>(
703
2
                    "snapshot_version not equal to g_Types_constants.TSNAPSHOT_REQ_VERSION2");
704
2
        }
705
706
322
        if (!res.ok()) {
707
0
            LOG(WARNING) << "convert rowset failed, res:" << res
708
0
                         << ", tablet:" << new_tablet_meta->tablet_id()
709
0
                         << ", schema hash:" << new_tablet_meta->schema_hash()
710
0
                         << ", snapshot_version:" << snapshot_version
711
0
                         << ", is incremental:" << request.__isset.missing_version;
712
0
            break;
713
0
        }
714
715
322
    } while (false);
716
717
    // link all binlog files to snapshot path
718
322
    do {
719
322
        if (!res.ok()) {
720
0
            break;
721
0
        }
722
723
322
        if (!need_ccr_binlog || !target_tablet->enable_ccr_binlog()) {
724
314
            break;
725
314
        }
726
727
8
        RowsetBinlogMetasPB rowset_binlog_metas_pb;
728
8
        for (auto& rs : consistent_rowsets) {
729
0
            if (!rs->is_local()) {
730
0
                continue;
731
0
            }
732
0
            res = ref_tablet->get_rowset_binlog_metas(rs->version(), &rowset_binlog_metas_pb);
733
0
            if (!res.ok()) {
734
0
                break;
735
0
            }
736
0
        }
737
8
        if (!res.ok() || rowset_binlog_metas_pb.rowset_binlog_metas_size() == 0) {
738
0
            break;
739
0
        }
740
741
        // write to pb file
742
8
        auto rowset_binlog_metas_pb_filename =
743
8
                fmt::format("{}/rowset_binlog_metas.pb", schema_full_path);
744
8
        res = write_pb(rowset_binlog_metas_pb_filename, rowset_binlog_metas_pb);
745
8
        if (!res.ok()) {
746
0
            break;
747
0
        }
748
749
8
        for (const auto& rowset_binlog_meta : rowset_binlog_metas_pb.rowset_binlog_metas()) {
750
0
            std::string segment_file_path;
751
0
            auto num_segments = rowset_binlog_meta.num_segments();
752
0
            std::string_view rowset_id = rowset_binlog_meta.rowset_id();
753
754
0
            RowsetMetaPB rowset_meta_pb;
755
0
            if (!rowset_meta_pb.ParseFromString(rowset_binlog_meta.data())) {
756
0
                auto err_msg = fmt::format("fail to parse binlog meta data value:{}",
757
0
                                           rowset_binlog_meta.data());
758
0
                res = Status::InternalError(err_msg);
759
0
                LOG(WARNING) << err_msg;
760
0
                return res;
761
0
            }
762
0
            const auto& tablet_schema_pb = rowset_meta_pb.tablet_schema();
763
0
            TabletSchema tablet_schema;
764
0
            tablet_schema.init_from_pb(tablet_schema_pb);
765
766
0
            std::vector<string> linked_success_files;
767
0
            Defer remove_linked_files {[&]() { // clear linked files if errors happen
768
0
                if (!res.ok()) {
769
0
                    LOG(WARNING) << "will delete linked success files due to error " << res;
770
0
                    std::vector<io::Path> paths;
771
0
                    for (auto& file : linked_success_files) {
772
0
                        paths.emplace_back(file);
773
0
                        LOG(WARNING)
774
0
                                << "will delete linked success file " << file << " due to error";
775
0
                    }
776
0
                    static_cast<void>(io::global_local_filesystem()->batch_delete(paths));
777
0
                    LOG(WARNING) << "done delete linked success files due to error " << res;
778
0
                }
779
0
            }};
780
781
            // link segment files and index files
782
0
            for (int64_t segment_index = 0; segment_index < num_segments; ++segment_index) {
783
0
                segment_file_path = ref_tablet->get_segment_filepath(rowset_id, segment_index);
784
0
                auto snapshot_segment_file_path =
785
0
                        fmt::format("{}/{}_{}.binlog", schema_full_path, rowset_id, segment_index);
786
787
0
                res = io::global_local_filesystem()->link_file(segment_file_path,
788
0
                                                               snapshot_segment_file_path);
789
0
                if (!res.ok()) {
790
0
                    LOG(WARNING) << "fail to link binlog file. [src=" << segment_file_path
791
0
                                 << ", dest=" << snapshot_segment_file_path << "]";
792
0
                    break;
793
0
                }
794
0
                linked_success_files.push_back(snapshot_segment_file_path);
795
796
0
                if (tablet_schema.get_inverted_index_storage_format() ==
797
0
                    InvertedIndexStorageFormatPB::V1) {
798
0
                    for (const auto& index : tablet_schema.inverted_indexes()) {
799
0
                        auto index_id = index->index_id();
800
0
                        auto index_file = InvertedIndexDescriptor::get_index_file_path_v1(
801
0
                                InvertedIndexDescriptor::get_index_file_path_prefix(
802
0
                                        segment_file_path),
803
0
                                index_id, index->get_index_suffix());
804
0
                        auto snapshot_segment_index_file_path =
805
0
                                fmt::format("{}/{}_{}_{}.binlog-index", schema_full_path, rowset_id,
806
0
                                            segment_index, index_id);
807
0
                        VLOG_DEBUG << "link " << index_file << " to "
808
0
                                   << snapshot_segment_index_file_path;
809
0
                        res = io::global_local_filesystem()->link_file(
810
0
                                index_file, snapshot_segment_index_file_path);
811
0
                        if (!res.ok()) {
812
0
                            LOG(WARNING) << "fail to link binlog index file. [src=" << index_file
813
0
                                         << ", dest=" << snapshot_segment_index_file_path << "]";
814
0
                            break;
815
0
                        }
816
0
                        linked_success_files.push_back(snapshot_segment_index_file_path);
817
0
                    }
818
0
                } else {
819
0
                    if (tablet_schema.has_inverted_index() || tablet_schema.has_ann_index()) {
820
0
                        auto index_file = InvertedIndexDescriptor::get_index_file_path_v2(
821
0
                                InvertedIndexDescriptor::get_index_file_path_prefix(
822
0
                                        segment_file_path));
823
0
                        auto snapshot_segment_index_file_path =
824
0
                                fmt::format("{}/{}_{}.binlog-index", schema_full_path, rowset_id,
825
0
                                            segment_index);
826
0
                        VLOG_DEBUG << "link " << index_file << " to "
827
0
                                   << snapshot_segment_index_file_path;
828
0
                        res = io::global_local_filesystem()->link_file(
829
0
                                index_file, snapshot_segment_index_file_path);
830
0
                        if (!res.ok()) {
831
0
                            LOG(WARNING) << "fail to link binlog index file. [src=" << index_file
832
0
                                         << ", dest=" << snapshot_segment_index_file_path << "]";
833
0
                            break;
834
0
                        }
835
0
                        linked_success_files.push_back(snapshot_segment_index_file_path);
836
0
                    }
837
0
                }
838
0
            }
839
840
0
            if (!res.ok()) {
841
0
                break;
842
0
            }
843
0
        }
844
8
    } while (false);
845
846
322
    if (!res.ok()) {
847
0
        LOG(WARNING) << "fail to make snapshot, try to delete the snapshot path. path="
848
0
                     << snapshot_id_path.c_str();
849
850
0
        bool exist = true;
851
0
        RETURN_IF_ERROR(io::global_local_filesystem()->exists(snapshot_id_path, &exist));
852
0
        if (exist) {
853
0
            VLOG_NOTICE << "remove snapshot path. [path=" << snapshot_id_path << "]";
854
0
            RETURN_IF_ERROR(io::global_local_filesystem()->delete_directory(snapshot_id_path));
855
0
        }
856
322
    } else {
857
322
        *snapshot_path = snapshot_id;
858
322
    }
859
860
322
    return res;
861
322
}
862
} // namespace doris