Coverage Report

Created: 2025-03-10 19:30

/root/doris/be/src/runtime/snapshot_loader.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 "runtime/snapshot_loader.h"
19
20
// IWYU pragma: no_include <bthread/errno.h>
21
#include <errno.h> // IWYU pragma: keep
22
#include <fmt/format.h>
23
#include <gen_cpp/FrontendService.h>
24
#include <gen_cpp/FrontendService_types.h>
25
#include <gen_cpp/HeartbeatService_types.h>
26
#include <gen_cpp/PlanNodes_types.h>
27
#include <gen_cpp/Status_types.h>
28
#include <gen_cpp/Types_types.h>
29
#include <stdint.h>
30
#include <stdio.h>
31
#include <unistd.h>
32
33
#include <algorithm>
34
#include <condition_variable>
35
#include <cstring>
36
#include <filesystem>
37
#include <istream>
38
#include <unordered_map>
39
#include <utility>
40
41
#include "common/config.h"
42
#include "common/logging.h"
43
#include "gutil/strings/split.h"
44
#include "http/http_client.h"
45
#include "io/fs/broker_file_system.h"
46
#include "io/fs/file_system.h"
47
#include "io/fs/file_writer.h"
48
#include "io/fs/hdfs_file_system.h"
49
#include "io/fs/local_file_system.h"
50
#include "io/fs/path.h"
51
#include "io/fs/remote_file_system.h"
52
#include "io/fs/s3_file_system.h"
53
#include "io/hdfs_builder.h"
54
#include "olap/data_dir.h"
55
#include "olap/snapshot_manager.h"
56
#include "olap/storage_engine.h"
57
#include "olap/tablet.h"
58
#include "olap/tablet_manager.h"
59
#include "runtime/client_cache.h"
60
#include "runtime/exec_env.h"
61
#include "util/s3_uri.h"
62
#include "util/s3_util.h"
63
#include "util/thrift_rpc_helper.h"
64
65
namespace doris {
66
67
0
static std::string get_loaded_tag_path(const std::string& snapshot_path) {
68
0
    return snapshot_path + "/LOADED";
69
0
}
70
71
0
static Status write_loaded_tag(const std::string& snapshot_path, int64_t tablet_id) {
72
0
    std::unique_ptr<io::FileWriter> writer;
73
0
    std::string file = get_loaded_tag_path(snapshot_path);
74
0
    RETURN_IF_ERROR(io::global_local_filesystem()->create_file(file, &writer));
75
0
    return writer->close();
76
0
}
77
78
Status upload_with_checksum(io::RemoteFileSystem& fs, std::string_view local_path,
79
0
                            std::string_view remote_path, std::string_view checksum) {
80
0
    auto full_remote_path = fmt::format("{}.{}", remote_path, checksum);
81
0
    switch (fs.type()) {
82
0
    case io::FileSystemType::HDFS:
83
0
    case io::FileSystemType::BROKER: {
84
0
        std::string temp = fmt::format("{}.part", remote_path);
85
0
        RETURN_IF_ERROR(fs.upload(local_path, temp));
86
0
        RETURN_IF_ERROR(fs.rename(temp, full_remote_path));
87
0
        break;
88
0
    }
89
0
    case io::FileSystemType::S3:
90
0
        RETURN_IF_ERROR(fs.upload(local_path, full_remote_path));
91
0
        break;
92
0
    default:
93
0
        LOG(FATAL) << "unknown fs type: " << static_cast<int>(fs.type());
94
0
    }
95
0
    return Status::OK();
96
0
}
97
98
SnapshotLoader::SnapshotLoader(ExecEnv* env, int64_t job_id, int64_t task_id)
99
        : _env(env),
100
          _job_id(job_id),
101
          _task_id(task_id),
102
          _broker_addr(TNetworkAddress()),
103
          _prop(std::map<std::string, std::string>()),
104
1
          _remote_fs(nullptr) {}
105
106
SnapshotLoader::SnapshotLoader(ExecEnv* env, int64_t job_id, int64_t task_id,
107
                               const TNetworkAddress& broker_addr,
108
                               const std::map<std::string, std::string>& prop)
109
0
        : _env(env), _job_id(job_id), _task_id(task_id), _broker_addr(broker_addr), _prop(prop) {}
110
111
0
Status SnapshotLoader::init(TStorageBackendType::type type, const std::string& location) {
112
0
    if (TStorageBackendType::type::S3 == type) {
113
0
        S3Conf s3_conf;
114
0
        S3URI s3_uri(location);
115
0
        RETURN_IF_ERROR(s3_uri.parse());
116
0
        RETURN_IF_ERROR(S3ClientFactory::convert_properties_to_s3_conf(_prop, s3_uri, &s3_conf));
117
0
        std::shared_ptr<io::S3FileSystem> fs;
118
0
        RETURN_IF_ERROR(io::S3FileSystem::create(std::move(s3_conf), "", nullptr, &fs));
119
0
        _remote_fs = std::move(fs);
120
0
    } else if (TStorageBackendType::type::HDFS == type) {
121
0
        THdfsParams hdfs_params = parse_properties(_prop);
122
0
        std::shared_ptr<io::HdfsFileSystem> fs;
123
0
        RETURN_IF_ERROR(
124
0
                io::HdfsFileSystem::create(hdfs_params, "", hdfs_params.fs_name, nullptr, &fs));
125
0
        _remote_fs = std::move(fs);
126
0
    } else if (TStorageBackendType::type::BROKER == type) {
127
0
        std::shared_ptr<io::BrokerFileSystem> fs;
128
0
        RETURN_IF_ERROR(io::BrokerFileSystem::create(_broker_addr, _prop, &fs));
129
0
        _remote_fs = std::move(fs);
130
0
    } else {
131
0
        return Status::InternalError("Unknown storage type: {}", type);
132
0
    }
133
0
    return Status::OK();
134
0
}
135
136
1
SnapshotLoader::~SnapshotLoader() = default;
137
138
Status SnapshotLoader::upload(const std::map<std::string, std::string>& src_to_dest_path,
139
0
                              std::map<int64_t, std::vector<std::string>>* tablet_files) {
140
0
    if (!_remote_fs) {
141
0
        return Status::InternalError("Storage backend not initialized.");
142
0
    }
143
0
    LOG(INFO) << "begin to upload snapshot files. num: " << src_to_dest_path.size()
144
0
              << ", broker addr: " << _broker_addr << ", job: " << _job_id << ", task" << _task_id;
145
146
    // check if job has already been cancelled
147
0
    int tmp_counter = 1;
148
0
    RETURN_IF_ERROR(_report_every(0, &tmp_counter, 0, 0, TTaskType::type::UPLOAD));
149
150
0
    Status status = Status::OK();
151
    // 1. validate local tablet snapshot paths
152
0
    RETURN_IF_ERROR(_check_local_snapshot_paths(src_to_dest_path, true));
153
154
    // 2. for each src path, upload it to remote storage
155
    // we report to frontend for every 10 files, and we will cancel the job if
156
    // the job has already been cancelled in frontend.
157
0
    int report_counter = 0;
158
0
    int total_num = src_to_dest_path.size();
159
0
    int finished_num = 0;
160
0
    for (auto iter = src_to_dest_path.begin(); iter != src_to_dest_path.end(); iter++) {
161
0
        const std::string& src_path = iter->first;
162
0
        const std::string& dest_path = iter->second;
163
164
        // Take a lock to protect the local snapshot path.
165
0
        auto local_snapshot_guard = LocalSnapshotLock::instance().acquire(src_path);
166
167
0
        int64_t tablet_id = 0;
168
0
        int32_t schema_hash = 0;
169
0
        RETURN_IF_ERROR(
170
0
                _get_tablet_id_and_schema_hash_from_file_path(src_path, &tablet_id, &schema_hash));
171
172
        // 2.1 get existing files from remote path
173
0
        std::map<std::string, FileStat> remote_files;
174
0
        RETURN_IF_ERROR(_list_with_checksum(dest_path, &remote_files));
175
176
0
        for (auto& tmp : remote_files) {
177
0
            VLOG_CRITICAL << "get remote file: " << tmp.first << ", checksum: " << tmp.second.md5;
178
0
        }
179
180
        // 2.2 list local files
181
0
        std::vector<std::string> local_files;
182
0
        std::vector<std::string> local_files_with_checksum;
183
0
        RETURN_IF_ERROR(_get_existing_files_from_local(src_path, &local_files));
184
185
        // 2.3 iterate local files
186
0
        for (auto it = local_files.begin(); it != local_files.end(); it++) {
187
0
            RETURN_IF_ERROR(_report_every(10, &report_counter, finished_num, total_num,
188
0
                                          TTaskType::type::UPLOAD));
189
190
0
            const std::string& local_file = *it;
191
            // calc md5sum of localfile
192
0
            std::string md5sum;
193
0
            RETURN_IF_ERROR(
194
0
                    io::global_local_filesystem()->md5sum(src_path + "/" + local_file, &md5sum));
195
0
            VLOG_CRITICAL << "get file checksum: " << local_file << ": " << md5sum;
196
0
            local_files_with_checksum.push_back(local_file + "." + md5sum);
197
198
            // check if this local file need upload
199
0
            bool need_upload = false;
200
0
            auto find = remote_files.find(local_file);
201
0
            if (find != remote_files.end()) {
202
0
                if (md5sum != find->second.md5) {
203
                    // remote storage file exist, but with different checksum
204
0
                    LOG(WARNING) << "remote file checksum is invalid. remote: " << find->first
205
0
                                 << ", local: " << md5sum;
206
                    // TODO(cmy): save these files and delete them later
207
0
                    need_upload = true;
208
0
                }
209
0
            } else {
210
0
                need_upload = true;
211
0
            }
212
213
0
            if (!need_upload) {
214
0
                VLOG_CRITICAL << "file exist in remote path, no need to upload: " << local_file;
215
0
                continue;
216
0
            }
217
218
            // upload
219
0
            std::string remote_path = dest_path + '/' + local_file;
220
0
            std::string local_path = src_path + '/' + local_file;
221
0
            RETURN_IF_ERROR(upload_with_checksum(*_remote_fs, local_path, remote_path, md5sum));
222
0
        } // end for each tablet's local files
223
224
0
        tablet_files->emplace(tablet_id, local_files_with_checksum);
225
0
        finished_num++;
226
0
        LOG(INFO) << "finished to write tablet to remote. local path: " << src_path
227
0
                  << ", remote path: " << dest_path;
228
0
    } // end for each tablet path
229
230
0
    LOG(INFO) << "finished to upload snapshots. job: " << _job_id << ", task id: " << _task_id;
231
0
    return status;
232
0
}
233
234
/*
235
 * Download snapshot files from remote.
236
 * After downloaded, the local dir should contains all files existing in remote,
237
 * may also contains several useless files.
238
 */
239
Status SnapshotLoader::download(const std::map<std::string, std::string>& src_to_dest_path,
240
0
                                std::vector<int64_t>* downloaded_tablet_ids) {
241
0
    if (!_remote_fs) {
242
0
        return Status::InternalError("Storage backend not initialized.");
243
0
    }
244
0
    LOG(INFO) << "begin to download snapshot files. num: " << src_to_dest_path.size()
245
0
              << ", broker addr: " << _broker_addr << ", job: " << _job_id
246
0
              << ", task id: " << _task_id;
247
248
    // check if job has already been cancelled
249
0
    int tmp_counter = 1;
250
0
    RETURN_IF_ERROR(_report_every(0, &tmp_counter, 0, 0, TTaskType::type::DOWNLOAD));
251
252
0
    Status status = Status::OK();
253
    // 1. validate local tablet snapshot paths
254
0
    RETURN_IF_ERROR(_check_local_snapshot_paths(src_to_dest_path, false));
255
256
    // 2. for each src path, download it to local storage
257
0
    int report_counter = 0;
258
0
    int total_num = src_to_dest_path.size();
259
0
    int finished_num = 0;
260
0
    for (auto iter = src_to_dest_path.begin(); iter != src_to_dest_path.end(); iter++) {
261
0
        const std::string& remote_path = iter->first;
262
0
        const std::string& local_path = iter->second;
263
264
        // Take a lock to protect the local snapshot path.
265
0
        auto local_snapshot_guard = LocalSnapshotLock::instance().acquire(local_path);
266
267
0
        int64_t local_tablet_id = 0;
268
0
        int32_t schema_hash = 0;
269
0
        RETURN_IF_ERROR(_get_tablet_id_and_schema_hash_from_file_path(local_path, &local_tablet_id,
270
0
                                                                      &schema_hash));
271
0
        downloaded_tablet_ids->push_back(local_tablet_id);
272
273
0
        int64_t remote_tablet_id;
274
0
        RETURN_IF_ERROR(_get_tablet_id_from_remote_path(remote_path, &remote_tablet_id));
275
0
        VLOG_CRITICAL << "get local tablet id: " << local_tablet_id
276
0
                      << ", schema hash: " << schema_hash
277
0
                      << ", remote tablet id: " << remote_tablet_id;
278
279
        // 2.1. get local files
280
0
        std::vector<std::string> local_files;
281
0
        RETURN_IF_ERROR(_get_existing_files_from_local(local_path, &local_files));
282
283
        // 2.2. get remote files
284
0
        std::map<std::string, FileStat> remote_files;
285
0
        RETURN_IF_ERROR(_list_with_checksum(remote_path, &remote_files));
286
0
        if (remote_files.empty()) {
287
0
            std::stringstream ss;
288
0
            ss << "get nothing from remote path: " << remote_path;
289
0
            LOG(WARNING) << ss.str();
290
0
            return Status::InternalError(ss.str());
291
0
        }
292
293
0
        TabletSharedPtr tablet =
294
0
                StorageEngine::instance()->tablet_manager()->get_tablet(local_tablet_id);
295
0
        if (tablet == nullptr) {
296
0
            std::stringstream ss;
297
0
            ss << "failed to get local tablet: " << local_tablet_id;
298
0
            LOG(WARNING) << ss.str();
299
0
            return Status::InternalError(ss.str());
300
0
        }
301
0
        DataDir* data_dir = tablet->data_dir();
302
303
0
        for (auto& iter : remote_files) {
304
0
            RETURN_IF_ERROR(_report_every(10, &report_counter, finished_num, total_num,
305
0
                                          TTaskType::type::DOWNLOAD));
306
307
0
            bool need_download = false;
308
0
            const std::string& remote_file = iter.first;
309
0
            const FileStat& file_stat = iter.second;
310
0
            auto find = std::find(local_files.begin(), local_files.end(), remote_file);
311
0
            if (find == local_files.end()) {
312
                // remote file does not exist in local, download it
313
0
                need_download = true;
314
0
            } else {
315
0
                if (_end_with(remote_file, ".hdr")) {
316
                    // this is a header file, download it.
317
0
                    need_download = true;
318
0
                } else {
319
                    // check checksum
320
0
                    std::string local_md5sum;
321
0
                    Status st = io::global_local_filesystem()->md5sum(
322
0
                            local_path + "/" + remote_file, &local_md5sum);
323
0
                    if (!st.ok()) {
324
0
                        LOG(WARNING) << "failed to get md5sum of local file: " << remote_file
325
0
                                     << ". msg: " << st << ". download it";
326
0
                        need_download = true;
327
0
                    } else {
328
0
                        VLOG_CRITICAL << "get local file checksum: " << remote_file << ": "
329
0
                                      << local_md5sum;
330
0
                        if (file_stat.md5 != local_md5sum) {
331
                            // file's checksum does not equal, download it.
332
0
                            need_download = true;
333
0
                        }
334
0
                    }
335
0
                }
336
0
            }
337
338
0
            if (!need_download) {
339
0
                LOG(INFO) << "remote file already exist in local, no need to download."
340
0
                          << ", file: " << remote_file;
341
0
                continue;
342
0
            }
343
344
            // begin to download
345
0
            std::string full_remote_file = remote_path + "/" + remote_file + "." + file_stat.md5;
346
0
            std::string local_file_name;
347
            // we need to replace the tablet_id in remote file name with local tablet id
348
0
            RETURN_IF_ERROR(_replace_tablet_id(remote_file, local_tablet_id, &local_file_name));
349
0
            std::string full_local_file = local_path + "/" + local_file_name;
350
0
            LOG(INFO) << "begin to download from " << full_remote_file << " to " << full_local_file;
351
0
            size_t file_len = file_stat.size;
352
353
            // check disk capacity
354
0
            if (data_dir->reach_capacity_limit(file_len)) {
355
0
                return Status::Error<ErrorCode::EXCEEDED_LIMIT>(
356
0
                        "reach the capacity limit of path {}, file_size={}", data_dir->path(),
357
0
                        file_len);
358
0
            }
359
            // remove file which will be downloaded now.
360
            // this file will be added to local_files if it be downloaded successfully.
361
0
            if (find != local_files.end()) {
362
0
                local_files.erase(find);
363
0
            }
364
0
            RETURN_IF_ERROR(_remote_fs->download(full_remote_file, full_local_file));
365
366
            // 3. check md5 of the downloaded file
367
0
            std::string downloaded_md5sum;
368
0
            RETURN_IF_ERROR(
369
0
                    io::global_local_filesystem()->md5sum(full_local_file, &downloaded_md5sum));
370
0
            VLOG_CRITICAL << "get downloaded file checksum: " << full_local_file << ": "
371
0
                          << downloaded_md5sum;
372
0
            if (downloaded_md5sum != file_stat.md5) {
373
0
                std::stringstream ss;
374
0
                ss << "invalid md5 of downloaded file: " << full_local_file
375
0
                   << ", expected: " << file_stat.md5 << ", get: " << downloaded_md5sum;
376
0
                LOG(WARNING) << ss.str();
377
0
                return Status::InternalError(ss.str());
378
0
            }
379
380
            // local_files always keep the updated local files
381
0
            local_files.push_back(local_file_name);
382
0
            LOG(INFO) << "finished to download file via broker. file: " << full_local_file
383
0
                      << ", length: " << file_len;
384
0
        } // end for all remote files
385
386
        // finally, delete local files which are not in remote
387
0
        for (const auto& local_file : local_files) {
388
            // replace the tablet id in local file name with the remote tablet id,
389
            // in order to compare the file name.
390
0
            std::string new_name;
391
0
            Status st = _replace_tablet_id(local_file, remote_tablet_id, &new_name);
392
0
            if (!st.ok()) {
393
0
                LOG(WARNING) << "failed to replace tablet id. unknown local file: " << st
394
0
                             << ". ignore it";
395
0
                continue;
396
0
            }
397
0
            VLOG_CRITICAL << "new file name after replace tablet id: " << new_name;
398
0
            const auto& find = remote_files.find(new_name);
399
0
            if (find != remote_files.end()) {
400
0
                continue;
401
0
            }
402
403
            // delete
404
0
            std::string full_local_file = local_path + "/" + local_file;
405
0
            VLOG_CRITICAL << "begin to delete local snapshot file: " << full_local_file
406
0
                          << ", it does not exist in remote";
407
0
            if (remove(full_local_file.c_str()) != 0) {
408
0
                LOG(WARNING) << "failed to delete unknown local file: " << full_local_file
409
0
                             << ", ignore it";
410
0
            }
411
0
        }
412
413
0
        finished_num++;
414
0
    } // end for src_to_dest_path
415
416
0
    LOG(INFO) << "finished to download snapshots. job: " << _job_id << ", task id: " << _task_id;
417
0
    return status;
418
0
}
419
420
Status SnapshotLoader::remote_http_download(
421
        const std::vector<TRemoteTabletSnapshot>& remote_tablet_snapshots,
422
0
        std::vector<int64_t>* downloaded_tablet_ids) {
423
0
    LOG(INFO) << fmt::format("begin to download snapshots via http. job: {}, task id: {}", _job_id,
424
0
                             _task_id);
425
0
    constexpr uint32_t kDownloadFileMaxRetry = 3;
426
427
    // check if job has already been cancelled
428
0
    int tmp_counter = 1;
429
0
    RETURN_IF_ERROR(_report_every(0, &tmp_counter, 0, 0, TTaskType::type::DOWNLOAD));
430
0
    Status status = Status::OK();
431
432
0
    int report_counter = 0;
433
0
    int finished_num = 0;
434
0
    int total_num = remote_tablet_snapshots.size();
435
0
    for (const auto& remote_tablet_snapshot : remote_tablet_snapshots) {
436
0
        const auto& local_path = remote_tablet_snapshot.local_snapshot_path;
437
0
        const auto& remote_path = remote_tablet_snapshot.remote_snapshot_path;
438
0
        LOG(INFO) << fmt::format(
439
0
                "download snapshots via http. job: {}, task id: {}, local dir: {}, remote dir: {}",
440
0
                _job_id, _task_id, local_path, remote_path);
441
442
        // Take a lock to protect the local snapshot path.
443
0
        auto local_snapshot_guard = LocalSnapshotLock::instance().acquire(local_path);
444
445
        // Step 1: Validate local tablet snapshot paths
446
0
        bool res = true;
447
0
        RETURN_IF_ERROR(io::global_local_filesystem()->is_directory(local_path, &res));
448
0
        if (!res) {
449
0
            std::stringstream ss;
450
0
            auto err_msg =
451
0
                    fmt::format("snapshot path is not directory or does not exist: {}", local_path);
452
0
            LOG(WARNING) << err_msg;
453
0
            return Status::RuntimeError(err_msg);
454
0
        }
455
456
        // Step 2: get all local files
457
0
        struct LocalFileStat {
458
0
            uint64_t size;
459
0
            std::string md5;
460
0
        };
461
0
        std::unordered_map<std::string, LocalFileStat> local_files;
462
0
        std::vector<std::string> existing_files;
463
0
        RETURN_IF_ERROR(_get_existing_files_from_local(local_path, &existing_files));
464
0
        for (auto& local_file : existing_files) {
465
            // add file size
466
0
            std::string local_file_path = local_path + "/" + local_file;
467
0
            std::error_code ec;
468
0
            uint64_t local_file_size = std::filesystem::file_size(local_file_path, ec);
469
0
            if (ec) {
470
0
                LOG(WARNING) << "download file error" << ec.message();
471
0
                return Status::IOError("can't retrive file_size of {}, due to {}", local_file_path,
472
0
                                       ec.message());
473
0
            }
474
0
            std::string md5;
475
0
            auto status = io::global_local_filesystem()->md5sum(local_file_path, &md5);
476
0
            if (!status.ok()) {
477
0
                LOG(WARNING) << "download file error, local file " << local_file_path
478
0
                             << " md5sum: " << status.to_string();
479
0
                return status;
480
0
            }
481
0
            local_files[local_file] = {local_file_size, md5};
482
0
        }
483
0
        existing_files.clear();
484
485
        // Step 3: Validate remote tablet snapshot paths && remote files map
486
        // key is remote snapshot paths, value is filelist
487
        // get all these use http download action
488
        // http://172.16.0.14:6781/api/_tablet/_download?token=e804dd27-86da-4072-af58-70724075d2a4&file=/home/ubuntu/doris_master/output/be/storage/snapshot/20230410102306.9.180//2774718/217609978/2774718.hdr
489
0
        struct RemoteFileStat {
490
0
            std::string url;
491
0
            std::string md5;
492
0
            uint64_t size;
493
0
        };
494
0
        std::unordered_map<std::string, RemoteFileStat> remote_files;
495
0
        const auto& token = remote_tablet_snapshot.remote_token;
496
0
        const auto& remote_be_addr = remote_tablet_snapshot.remote_be_addr;
497
498
        // HEAD http://172.16.0.14:6781/api/_tablet/_download?token=e804dd27-86da-4072-af58-70724075d2a4&file=/home/ubuntu/doris_master/output/be/storage/snapshot/20230410102306.9.180/
499
0
        std::string base_url = fmt::format("http://{}:{}/api/_tablet/_download?token={}",
500
0
                                           remote_be_addr.hostname, remote_be_addr.port, token);
501
0
        std::string remote_url_prefix = fmt::format("{}&file={}", base_url, remote_path);
502
503
0
        string file_list_str;
504
0
        auto list_files_cb = [&remote_url_prefix, &file_list_str](HttpClient* client) {
505
0
            RETURN_IF_ERROR(client->init(remote_url_prefix));
506
0
            client->set_timeout_ms(config::download_binlog_meta_timeout_ms);
507
0
            return client->execute(&file_list_str);
508
0
        };
509
0
        RETURN_IF_ERROR(HttpClient::execute_with_retry(kDownloadFileMaxRetry, 1, list_files_cb));
510
0
        std::vector<string> filename_list =
511
0
                strings::Split(file_list_str, "\n", strings::SkipWhitespace());
512
513
0
        for (const auto& filename : filename_list) {
514
0
            std::string remote_file_url =
515
0
                    fmt::format("{}&file={}/{}&channel=ingest_binlog", base_url,
516
0
                                remote_tablet_snapshot.remote_snapshot_path, filename);
517
518
            // get file length
519
0
            uint64_t file_size = 0;
520
0
            std::string file_md5;
521
0
            auto get_file_stat_cb = [&remote_file_url, &file_size, &file_md5](HttpClient* client) {
522
0
                int64_t timeout_ms = config::download_binlog_meta_timeout_ms;
523
0
                std::string url = remote_file_url;
524
0
                if (config::enable_download_md5sum_check) {
525
                    // compute md5sum is time-consuming, so we set a longer timeout
526
0
                    timeout_ms = config::download_binlog_meta_timeout_ms * 3;
527
0
                    url = fmt::format("{}&acquire_md5=true", remote_file_url);
528
0
                }
529
0
                RETURN_IF_ERROR(client->init(url));
530
0
                client->set_timeout_ms(timeout_ms);
531
0
                RETURN_IF_ERROR(client->head());
532
0
                RETURN_IF_ERROR(client->get_content_length(&file_size));
533
0
                if (config::enable_download_md5sum_check) {
534
0
                    RETURN_IF_ERROR(client->get_content_md5(&file_md5));
535
0
                }
536
0
                return Status::OK();
537
0
            };
538
0
            RETURN_IF_ERROR(
539
0
                    HttpClient::execute_with_retry(kDownloadFileMaxRetry, 1, get_file_stat_cb));
540
541
0
            remote_files[filename] = RemoteFileStat {remote_file_url, file_md5, file_size};
542
0
        }
543
544
        // Step 4: Compare local and remote files && get all need download files
545
0
        RETURN_IF_ERROR(_report_every(10, &report_counter, finished_num, total_num,
546
0
                                      TTaskType::type::DOWNLOAD));
547
548
        // get all need download files
549
0
        std::vector<std::string> need_download_files;
550
0
        for (const auto& [remote_file, remote_filestat] : remote_files) {
551
0
            LOG(INFO) << "remote file: " << remote_file << ", size: " << remote_filestat.size
552
0
                      << ", md5: " << remote_filestat.md5;
553
0
            auto it = local_files.find(remote_file);
554
0
            if (it == local_files.end()) {
555
0
                need_download_files.emplace_back(remote_file);
556
0
                continue;
557
0
            }
558
0
            if (_end_with(remote_file, ".hdr")) {
559
0
                need_download_files.emplace_back(remote_file);
560
0
                continue;
561
0
            }
562
563
0
            if (auto& local_filestat = it->second; local_filestat.size != remote_filestat.size) {
564
0
                need_download_files.emplace_back(remote_file);
565
0
                continue;
566
0
            }
567
568
0
            if (auto& local_filestat = it->second; local_filestat.md5 != remote_filestat.md5) {
569
0
                need_download_files.emplace_back(remote_file);
570
0
                continue;
571
0
            }
572
573
0
            LOG(INFO) << fmt::format("file {} already exists, skip download", remote_file);
574
0
        }
575
576
0
        auto local_tablet_id = remote_tablet_snapshot.local_tablet_id;
577
0
        TabletSharedPtr tablet =
578
0
                StorageEngine::instance()->tablet_manager()->get_tablet(local_tablet_id);
579
0
        if (tablet == nullptr) {
580
0
            std::stringstream ss;
581
0
            ss << "failed to get local tablet: " << local_tablet_id;
582
0
            LOG(WARNING) << ss.str();
583
0
            return Status::InternalError(ss.str());
584
0
        }
585
0
        DataDir* data_dir = tablet->data_dir();
586
587
        // download all need download files
588
0
        uint64_t total_file_size = 0;
589
0
        MonotonicStopWatch watch;
590
0
        watch.start();
591
0
        for (auto& filename : need_download_files) {
592
0
            auto& remote_filestat = remote_files[filename];
593
0
            auto file_size = remote_filestat.size;
594
0
            auto& remote_file_url = remote_filestat.url;
595
0
            auto& remote_file_md5 = remote_filestat.md5;
596
597
            // check disk capacity
598
0
            if (data_dir->reach_capacity_limit(file_size)) {
599
0
                return Status::Error<ErrorCode::EXCEEDED_LIMIT>(
600
0
                        "reach the capacity limit of path {}, file_size={}", data_dir->path(),
601
0
                        file_size);
602
0
            }
603
604
0
            total_file_size += file_size;
605
0
            uint64_t estimate_timeout = file_size / config::download_low_speed_limit_kbps / 1024;
606
0
            if (estimate_timeout < config::download_low_speed_time) {
607
0
                estimate_timeout = config::download_low_speed_time;
608
0
            }
609
610
0
            std::string local_filename;
611
0
            RETURN_IF_ERROR(_replace_tablet_id(filename, local_tablet_id, &local_filename));
612
0
            std::string local_file_path = local_path + "/" + local_filename;
613
614
0
            LOG(INFO) << "clone begin to download file from: " << remote_file_url
615
0
                      << " to: " << local_file_path << ". size(B): " << file_size
616
0
                      << ", timeout(s): " << estimate_timeout;
617
618
0
            auto download_cb = [&remote_file_url, &remote_file_md5, estimate_timeout,
619
0
                                &local_file_path, file_size](HttpClient* client) {
620
0
                RETURN_IF_ERROR(client->init(remote_file_url));
621
0
                client->set_timeout_ms(estimate_timeout * 1000);
622
0
                RETURN_IF_ERROR(client->download(local_file_path));
623
624
0
                std::error_code ec;
625
                // Check file length
626
0
                uint64_t local_file_size = std::filesystem::file_size(local_file_path, ec);
627
0
                if (ec) {
628
0
                    LOG(WARNING) << "download file error" << ec.message();
629
0
                    return Status::IOError("can't retrive file_size of {}, due to {}",
630
0
                                           local_file_path, ec.message());
631
0
                }
632
0
                if (local_file_size != file_size) {
633
0
                    LOG(WARNING) << "download file length error"
634
0
                                 << ", remote_path=" << remote_file_url
635
0
                                 << ", file_size=" << file_size
636
0
                                 << ", local_file_size=" << local_file_size;
637
0
                    return Status::InternalError("downloaded file size is not equal");
638
0
                }
639
640
0
                if (!remote_file_md5.empty()) { // keep compatibility
641
0
                    std::string local_file_md5;
642
0
                    RETURN_IF_ERROR(io::global_local_filesystem()->md5sum(local_file_path,
643
0
                                                                          &local_file_md5));
644
0
                    if (local_file_md5 != remote_file_md5) {
645
0
                        LOG(WARNING) << "download file md5 error"
646
0
                                     << ", remote_file_url=" << remote_file_url
647
0
                                     << ", local_file_path=" << local_file_path
648
0
                                     << ", remote_file_md5=" << remote_file_md5
649
0
                                     << ", local_file_md5=" << local_file_md5;
650
0
                        return Status::RuntimeError(
651
0
                                "download file {} md5 is not equal, local={}, remote={}",
652
0
                                remote_file_url, local_file_md5, remote_file_md5);
653
0
                    }
654
0
                }
655
656
0
                return io::global_local_filesystem()->permission(
657
0
                        local_file_path, io::LocalFileSystem::PERMS_OWNER_RW);
658
0
            };
659
0
            auto status = HttpClient::execute_with_retry(kDownloadFileMaxRetry, 1, download_cb);
660
0
            if (!status.ok()) {
661
0
                LOG(WARNING) << "failed to download file from " << remote_file_url
662
0
                             << ", status: " << status.to_string();
663
0
                return status;
664
0
            }
665
666
            // local_files always keep the updated local files
667
0
            local_files[filename] = LocalFileStat {file_size, remote_file_md5};
668
0
        }
669
670
0
        uint64_t total_time_ms = watch.elapsed_time() / 1000 / 1000;
671
0
        total_time_ms = total_time_ms > 0 ? total_time_ms : 0;
672
0
        double copy_rate = 0.0;
673
0
        if (total_time_ms > 0) {
674
0
            copy_rate = total_file_size / ((double)total_time_ms) / 1000;
675
0
        }
676
0
        auto remote_tablet_id = remote_tablet_snapshot.remote_tablet_id;
677
0
        LOG(INFO) << fmt::format(
678
0
                "succeed to copy remote tablet {} to local tablet {}, total file size: {} B, cost: "
679
0
                "{} ms, rate: {} MB/s",
680
0
                remote_tablet_id, local_tablet_id, total_file_size, total_time_ms, copy_rate);
681
682
        // local_files: contain all remote files and local files
683
        // finally, delete local files which are not in remote
684
0
        for (const auto& [local_file, local_filestat] : local_files) {
685
            // replace the tablet id in local file name with the remote tablet id,
686
            // in order to compare the file name.
687
0
            std::string new_name;
688
0
            Status st = _replace_tablet_id(local_file, remote_tablet_id, &new_name);
689
0
            if (!st.ok()) {
690
0
                LOG(WARNING) << "failed to replace tablet id. unknown local file: " << st
691
0
                             << ". ignore it";
692
0
                continue;
693
0
            }
694
0
            VLOG_CRITICAL << "new file name after replace tablet id: " << new_name;
695
0
            const auto& find = remote_files.find(new_name);
696
0
            if (find != remote_files.end()) {
697
0
                continue;
698
0
            }
699
700
            // delete
701
0
            std::string full_local_file = local_path + "/" + local_file;
702
0
            LOG(INFO) << "begin to delete local snapshot file: " << full_local_file
703
0
                      << ", it does not exist in remote";
704
0
            if (remove(full_local_file.c_str()) != 0) {
705
0
                LOG(WARNING) << "failed to delete unknown local file: " << full_local_file
706
0
                             << ", error: " << strerror(errno)
707
0
                             << ", file size: " << local_filestat.size << ", ignore it";
708
0
            }
709
0
        }
710
711
0
        ++finished_num;
712
0
    }
713
714
0
    LOG(INFO) << "finished to download snapshots. job: " << _job_id << ", task id: " << _task_id;
715
0
    return status;
716
0
}
717
718
// move the snapshot files in snapshot_path
719
// to tablet_path
720
// If overwrite, just replace the tablet_path with snapshot_path,
721
// else: (TODO)
722
//
723
// MUST hold tablet's header lock, push lock, cumulative lock and base compaction lock
724
Status SnapshotLoader::move(const std::string& snapshot_path, TabletSharedPtr tablet,
725
0
                            bool overwrite) {
726
    // Take a lock to protect the local snapshot path.
727
0
    auto local_snapshot_guard = LocalSnapshotLock::instance().acquire(snapshot_path);
728
729
0
    auto tablet_path = tablet->tablet_path();
730
0
    auto store_path = tablet->data_dir()->path();
731
0
    LOG(INFO) << "begin to move snapshot files. from: " << snapshot_path << ", to: " << tablet_path
732
0
              << ", store: " << store_path << ", job: " << _job_id << ", task id: " << _task_id;
733
734
0
    Status status = Status::OK();
735
736
    // validate snapshot_path and tablet_path
737
0
    int64_t snapshot_tablet_id = 0;
738
0
    int32_t snapshot_schema_hash = 0;
739
0
    RETURN_IF_ERROR(_get_tablet_id_and_schema_hash_from_file_path(
740
0
            snapshot_path, &snapshot_tablet_id, &snapshot_schema_hash));
741
742
0
    int64_t tablet_id = 0;
743
0
    int32_t schema_hash = 0;
744
0
    RETURN_IF_ERROR(
745
0
            _get_tablet_id_and_schema_hash_from_file_path(tablet_path, &tablet_id, &schema_hash));
746
747
0
    if (tablet_id != snapshot_tablet_id || schema_hash != snapshot_schema_hash) {
748
0
        std::stringstream ss;
749
0
        ss << "path does not match. snapshot: " << snapshot_path
750
0
           << ", tablet path: " << tablet_path;
751
0
        LOG(WARNING) << ss.str();
752
0
        return Status::InternalError(ss.str());
753
0
    }
754
755
0
    DataDir* store = StorageEngine::instance()->get_store(store_path);
756
0
    if (store == nullptr) {
757
0
        std::stringstream ss;
758
0
        ss << "failed to get store by path: " << store_path;
759
0
        LOG(WARNING) << ss.str();
760
0
        return Status::InternalError(ss.str());
761
0
    }
762
763
0
    if (!std::filesystem::exists(tablet_path)) {
764
0
        std::stringstream ss;
765
0
        ss << "tablet path does not exist: " << tablet_path;
766
0
        LOG(WARNING) << ss.str();
767
0
        return Status::InternalError(ss.str());
768
0
    }
769
770
0
    if (!std::filesystem::exists(snapshot_path)) {
771
0
        std::stringstream ss;
772
0
        ss << "snapshot path does not exist: " << snapshot_path;
773
0
        LOG(WARNING) << ss.str();
774
0
        return Status::InternalError(ss.str());
775
0
    }
776
777
0
    std::string loaded_tag_path = get_loaded_tag_path(snapshot_path);
778
0
    bool already_loaded = false;
779
0
    RETURN_IF_ERROR(io::global_local_filesystem()->exists(loaded_tag_path, &already_loaded));
780
0
    if (already_loaded) {
781
0
        LOG(INFO) << "snapshot path already moved: " << snapshot_path;
782
0
        return Status::OK();
783
0
    }
784
785
    // rename the rowset ids and tabletid info in rowset meta
786
0
    auto res = SnapshotManager::instance()->convert_rowset_ids(
787
0
            snapshot_path, tablet_id, tablet->replica_id(), tablet->partition_id(), schema_hash);
788
0
    if (!res.has_value()) [[unlikely]] {
789
0
        auto err_msg =
790
0
                fmt::format("failed to convert rowsetids in snapshot: {}, tablet path: {}, err: {}",
791
0
                            snapshot_path, tablet_path, res.error());
792
0
        LOG(WARNING) << err_msg;
793
0
        return Status::InternalError(err_msg);
794
0
    }
795
796
0
    if (!overwrite) {
797
0
        LOG(FATAL) << "only support overwrite now";
798
0
        __builtin_unreachable();
799
0
    }
800
801
    // Medium migration/clone/checkpoint/compaction may change or check the
802
    // files and tablet meta, so we need to take these locks.
803
0
    std::unique_lock migration_lock(tablet->get_migration_lock(), std::try_to_lock);
804
0
    std::unique_lock base_compact_lock(tablet->get_base_compaction_lock(), std::try_to_lock);
805
0
    std::unique_lock cumu_compact_lock(tablet->get_cumulative_compaction_lock(), std::try_to_lock);
806
0
    std::unique_lock cold_compact_lock(tablet->get_cold_compaction_lock(), std::try_to_lock);
807
0
    std::unique_lock build_idx_lock(tablet->get_build_inverted_index_lock(), std::try_to_lock);
808
0
    std::unique_lock meta_store_lock(tablet->get_meta_store_lock(), std::try_to_lock);
809
0
    if (!migration_lock.owns_lock() || !base_compact_lock.owns_lock() ||
810
0
        !cumu_compact_lock.owns_lock() || !cold_compact_lock.owns_lock() ||
811
0
        !build_idx_lock.owns_lock() || !meta_store_lock.owns_lock()) {
812
        // This error should be retryable
813
0
        auto status = Status::ObtainLockFailed("failed to get tablet locks, tablet: {}", tablet_id);
814
0
        LOG(WARNING) << status << ", snapshot path: " << snapshot_path
815
0
                     << ", tablet path: " << tablet_path;
816
0
        return status;
817
0
    }
818
819
0
    std::vector<std::string> snapshot_files;
820
0
    RETURN_IF_ERROR(_get_existing_files_from_local(snapshot_path, &snapshot_files));
821
822
    // FIXME: the below logic will demage the tablet files if failed in the middle.
823
824
    // 1. simply delete the old dir and replace it with the snapshot dir
825
0
    try {
826
        // This remove seems soft enough, because we already get
827
        // tablet id and schema hash from this path, which
828
        // means this path is a valid path.
829
0
        std::filesystem::remove_all(tablet_path);
830
0
        VLOG_CRITICAL << "remove dir: " << tablet_path;
831
0
        std::filesystem::create_directory(tablet_path);
832
0
        VLOG_CRITICAL << "re-create dir: " << tablet_path;
833
0
    } catch (const std::filesystem::filesystem_error& e) {
834
0
        std::stringstream ss;
835
0
        ss << "failed to move tablet path: " << tablet_path << ". err: " << e.what();
836
0
        LOG(WARNING) << ss.str();
837
0
        return Status::InternalError(ss.str());
838
0
    }
839
840
    // link files one by one
841
    // files in snapshot dir will be moved in snapshot clean process
842
0
    std::vector<std::string> linked_files;
843
0
    for (auto& file : snapshot_files) {
844
0
        auto full_src_path = fmt::format("{}/{}", snapshot_path, file);
845
0
        auto full_dest_path = fmt::format("{}/{}", tablet_path, file);
846
0
        if (link(full_src_path.c_str(), full_dest_path.c_str()) != 0) {
847
0
            LOG(WARNING) << "failed to link file from " << full_src_path << " to " << full_dest_path
848
0
                         << ", err: " << std::strerror(errno);
849
850
            // clean the already linked files
851
0
            for (auto& linked_file : linked_files) {
852
0
                remove(linked_file.c_str());
853
0
            }
854
855
0
            return Status::InternalError("move tablet failed");
856
0
        }
857
0
        linked_files.push_back(full_dest_path);
858
0
        VLOG_CRITICAL << "link file from " << full_src_path << " to " << full_dest_path;
859
0
    }
860
861
    // snapshot loader not need to change tablet uid
862
    // fixme: there is no header now and can not call load_one_tablet here
863
    // reload header
864
0
    Status ost = StorageEngine::instance()->tablet_manager()->load_tablet_from_dir(
865
0
            store, tablet_id, schema_hash, tablet_path, true);
866
0
    if (!ost.ok()) {
867
0
        std::stringstream ss;
868
0
        ss << "failed to reload header of tablet: " << tablet_id;
869
0
        LOG(WARNING) << ss.str();
870
0
        return Status::InternalError(ss.str());
871
0
    }
872
873
    // mark the snapshot path as loaded
874
0
    RETURN_IF_ERROR(write_loaded_tag(snapshot_path, tablet_id));
875
876
0
    LOG(INFO) << "finished to reload header of tablet: " << tablet_id;
877
878
0
    return status;
879
0
}
880
881
11
bool SnapshotLoader::_end_with(const std::string& str, const std::string& match) {
882
11
    if (str.size() >= match.size() &&
883
11
        str.compare(str.size() - match.size(), match.size(), match) == 0) {
884
4
        return true;
885
4
    }
886
7
    return false;
887
11
}
888
889
Status SnapshotLoader::_get_tablet_id_and_schema_hash_from_file_path(const std::string& src_path,
890
                                                                     int64_t* tablet_id,
891
2
                                                                     int32_t* schema_hash) {
892
    // path should be like: /path/.../tablet_id/schema_hash
893
    // we try to extract tablet_id from path
894
2
    size_t pos = src_path.find_last_of("/");
895
2
    if (pos == std::string::npos || pos == src_path.length() - 1) {
896
1
        return Status::InternalError("failed to get tablet id from path: {}", src_path);
897
1
    }
898
899
1
    std::string schema_hash_str = src_path.substr(pos + 1);
900
1
    std::stringstream ss1;
901
1
    ss1 << schema_hash_str;
902
1
    ss1 >> *schema_hash;
903
904
    // skip schema hash part
905
1
    size_t pos2 = src_path.find_last_of("/", pos - 1);
906
1
    if (pos2 == std::string::npos) {
907
0
        return Status::InternalError("failed to get tablet id from path: {}", src_path);
908
0
    }
909
910
1
    std::string tablet_str = src_path.substr(pos2 + 1, pos - pos2);
911
1
    std::stringstream ss2;
912
1
    ss2 << tablet_str;
913
1
    ss2 >> *tablet_id;
914
915
1
    VLOG_CRITICAL << "get tablet id " << *tablet_id << ", schema hash: " << *schema_hash
916
0
                  << " from path: " << src_path;
917
1
    return Status::OK();
918
1
}
919
920
Status SnapshotLoader::_check_local_snapshot_paths(
921
4
        const std::map<std::string, std::string>& src_to_dest_path, bool check_src) {
922
4
    bool res = true;
923
4
    for (const auto& pair : src_to_dest_path) {
924
4
        std::string path;
925
4
        if (check_src) {
926
2
            path = pair.first;
927
2
        } else {
928
2
            path = pair.second;
929
2
        }
930
931
4
        RETURN_IF_ERROR(io::global_local_filesystem()->is_directory(path, &res));
932
2
        if (!res) {
933
0
            std::stringstream ss;
934
0
            ss << "snapshot path is not directory or does not exist: " << path;
935
0
            LOG(WARNING) << ss.str();
936
0
            return Status::RuntimeError(ss.str());
937
0
        }
938
2
    }
939
2
    LOG(INFO) << "all local snapshot paths are existing. num: " << src_to_dest_path.size();
940
2
    return Status::OK();
941
4
}
942
943
Status SnapshotLoader::_get_existing_files_from_local(const std::string& local_path,
944
1
                                                      std::vector<std::string>* local_files) {
945
1
    bool exists = true;
946
1
    std::vector<io::FileInfo> files;
947
1
    RETURN_IF_ERROR(io::global_local_filesystem()->list(local_path, true, &files, &exists));
948
1
    for (auto& file : files) {
949
0
        local_files->push_back(file.file_name);
950
0
    }
951
1
    LOG(INFO) << "finished to list files in local path: " << local_path
952
1
              << ", file num: " << local_files->size();
953
1
    return Status::OK();
954
1
}
955
956
Status SnapshotLoader::_replace_tablet_id(const std::string& file_name, int64_t tablet_id,
957
4
                                          std::string* new_file_name) {
958
    // eg:
959
    // 10007.hdr
960
    // 10007_2_2_0_0.idx
961
    // 10007_2_2_0_0.dat
962
4
    if (_end_with(file_name, ".hdr")) {
963
1
        std::stringstream ss;
964
1
        ss << tablet_id << ".hdr";
965
1
        *new_file_name = ss.str();
966
1
        return Status::OK();
967
3
    } else if (_end_with(file_name, ".idx") || _end_with(file_name, ".dat")) {
968
2
        *new_file_name = file_name;
969
2
        return Status::OK();
970
2
    } else {
971
1
        return Status::InternalError("invalid tablet file name: {}", file_name);
972
1
    }
973
4
}
974
975
Status SnapshotLoader::_get_tablet_id_from_remote_path(const std::string& remote_path,
976
1
                                                       int64_t* tablet_id) {
977
    // eg:
978
    // bos://xxx/../__tbl_10004/__part_10003/__idx_10004/__10005
979
1
    size_t pos = remote_path.find_last_of("_");
980
1
    if (pos == std::string::npos) {
981
0
        return Status::InternalError("invalid remove file path: {}", remote_path);
982
0
    }
983
984
1
    std::string tablet_id_str = remote_path.substr(pos + 1);
985
1
    std::stringstream ss;
986
1
    ss << tablet_id_str;
987
1
    ss >> *tablet_id;
988
989
1
    return Status::OK();
990
1
}
991
992
// only return CANCELLED if FE return that job is cancelled.
993
// otherwise, return OK
994
Status SnapshotLoader::_report_every(int report_threshold, int* counter, int32_t finished_num,
995
0
                                     int32_t total_num, TTaskType::type type) {
996
0
    ++*counter;
997
0
    if (*counter <= report_threshold) {
998
0
        return Status::OK();
999
0
    }
1000
1001
0
    LOG(INFO) << "report to frontend. job id: " << _job_id << ", task id: " << _task_id
1002
0
              << ", finished num: " << finished_num << ", total num:" << total_num;
1003
1004
0
    TNetworkAddress master_addr = _env->master_info()->network_address;
1005
1006
0
    TSnapshotLoaderReportRequest request;
1007
0
    request.job_id = _job_id;
1008
0
    request.task_id = _task_id;
1009
0
    request.task_type = type;
1010
0
    request.__set_finished_num(finished_num);
1011
0
    request.__set_total_num(total_num);
1012
0
    TStatus report_st;
1013
1014
0
    Status rpcStatus = ThriftRpcHelper::rpc<FrontendServiceClient>(
1015
0
            master_addr.hostname, master_addr.port,
1016
0
            [&request, &report_st](FrontendServiceConnection& client) {
1017
0
                client->snapshotLoaderReport(report_st, request);
1018
0
            },
1019
0
            10000);
1020
1021
0
    if (!rpcStatus.ok()) {
1022
        // rpc failed, ignore
1023
0
        return Status::OK();
1024
0
    }
1025
1026
    // reset
1027
0
    *counter = 0;
1028
0
    if (report_st.status_code == TStatusCode::CANCELLED) {
1029
0
        LOG(INFO) << "job is cancelled. job id: " << _job_id << ", task id: " << _task_id;
1030
0
        return Status::Cancelled("Cancelled");
1031
0
    }
1032
0
    return Status::OK();
1033
0
}
1034
1035
Status SnapshotLoader::_list_with_checksum(const std::string& dir,
1036
0
                                           std::map<std::string, FileStat>* md5_files) {
1037
0
    bool exists = true;
1038
0
    std::vector<io::FileInfo> files;
1039
0
    RETURN_IF_ERROR(_remote_fs->list(dir, true, &files, &exists));
1040
0
    for (auto& tmp_file : files) {
1041
0
        io::Path path(tmp_file.file_name);
1042
0
        std::string file_name = path.filename();
1043
0
        size_t pos = file_name.find_last_of(".");
1044
0
        if (pos == std::string::npos || pos == file_name.size() - 1) {
1045
            // Not found checksum separator, ignore this file
1046
0
            continue;
1047
0
        }
1048
0
        FileStat stat = {std::string(file_name, 0, pos), std::string(file_name, pos + 1),
1049
0
                         tmp_file.file_size};
1050
0
        md5_files->emplace(std::string(file_name, 0, pos), stat);
1051
0
    }
1052
1053
0
    return Status::OK();
1054
0
}
1055
1056
} // end namespace doris