// Licensed to the Apache Software Foundation (ASF) under one

// or more contributor license agreements. See the NOTICE file

// distributed with this work for additional information

// regarding copyright ownership. The ASF licenses this file

// to you under the Apache License, Version 2.0 (the

// "License"); you may not use this file except in compliance

// with the License. You may obtain a copy of the License at

//

// http://www.apache.org/licenses/LICENSE-2.0

//

// Unless required by applicable law or agreed to in writing,

// software distributed under the License is distributed on an

// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY

// KIND, either express or implied. See the License for the

// specific language governing permissions and limitations

// under the License.

#include "cloud/cloud_snapshot_loader.h"

#include <gen_cpp/Types_types.h>

#include <unordered_map>

#include "cloud/cloud_snapshot_mgr.h"

#include "cloud/cloud_storage_engine.h"

#include "common/logging.h"

#include "io/fs/broker_file_system.h"

#include "io/fs/file_system.h"

#include "io/fs/hdfs_file_system.h"

#include "io/fs/path.h"

#include "io/fs/remote_file_system.h"

#include "io/fs/s3_file_system.h"

#include "storage/olap_common.h"

#include "storage/olap_define.h"

#include "storage/rowset/rowset_factory.h"

#include "storage/rowset/rowset_meta.h"

#include "storage/rowset/rowset_writer.h"

#include "storage/rowset/rowset_writer_context.h"

#include "storage/tablet/tablet.h"

#include "util/slice.h"

namespace doris {

#include "common/compile_check_avoid_begin.h"

namespace {

bool _end_with(std::string_view str, std::string_view match) {

return str.size() >= match.size() &&

str.compare(str.size() - match.size(), match.size(), match) == 0;

}

} // namespace

CloudSnapshotLoader::CloudSnapshotLoader(CloudStorageEngine& engine, ExecEnv* env, int64_t job_id,

int64_t task_id, const TNetworkAddress& broker_addr,

const std::map<std::string, std::string>& broker_prop)

: BaseSnapshotLoader(env, job_id, task_id, broker_addr, broker_prop), _engine(engine) {};

Status CloudSnapshotLoader::init(TStorageBackendType::type type, const std::string& location,

std::string vault_id) {

RETURN_IF_ERROR(BaseSnapshotLoader::init(type, location));

_storage_resource = _engine.get_storage_resource(vault_id);

if (!_storage_resource) {

return Status::InternalError("vault id not found, vault id {}", vault_id);

}

return Status::OK();

}

io::RemoteFileSystemSPtr CloudSnapshotLoader::storage_fs() {

return _storage_resource->fs;

}

Status CloudSnapshotLoader::upload(const std::map<std::string, std::string>& src_to_dest_path,

std::map<int64_t, std::vector<std::string>>* tablet_files) {

return Status::NotSupported("upload not supported");

}

Status CloudSnapshotLoader::download(const std::map<std::string, std::string>& src_to_dest_path,

std::vector<int64_t>* downloaded_tablet_ids) {

if (!_remote_fs || !_storage_resource) {

return Status::InternalError("Storage backend not initialized.");

}

LOG(INFO) << "begin to transfer snapshot files. num: " << src_to_dest_path.size()

<< ", broker addr: " << _broker_addr << ", job: " << _job_id

<< ", task id: " << _task_id;

// check if job has already been cancelled

int tmp_counter = 1;

RETURN_IF_ERROR(_report_every(0, &tmp_counter, 0, 0, TTaskType::type::DOWNLOAD));

Status status = Status::OK();

// 1. for each src path, transfer files to target path

int report_counter = 0;

int total_num = src_to_dest_path.size();

int finished_num = 0;

for (const auto& iter : src_to_dest_path) {

// remote_path eg:

// cos://xxx/__palo_repository_xxx/_ss_xxx/_ss_content/__db_10000/

// __tbl_10001/__part_10002/_idx_10001/__10003

const std::string& remote_path = iter.first;

const std::string& tablet_str = iter.second;

int64_t target_tablet_id = -1;

try {

target_tablet_id = std::stoll(tablet_str);

} catch (std::exception& e) {

return Status::InternalError("failed to parse target tablet id {}, {}", tablet_str,

e.what());

}

const std::string target_path = _storage_resource->remote_tablet_path(target_tablet_id);

// 1.1. check target path not exists

bool target_path_exist = false;

if (!storage_fs()->exists(target_path, &target_path_exist).ok() || target_path_exist) {

std::stringstream ss;

ss << "failed to download snapshot files, target path already exists: " << target_path;

LOG(WARNING) << ss.str();

return Status::InternalError(ss.str());

}

downloaded_tablet_ids->push_back(target_tablet_id);

int64_t remote_tablet_id;

RETURN_IF_ERROR(_get_tablet_id_from_remote_path(remote_path, &remote_tablet_id));

VLOG_CRITICAL << "get target tablet id: " << target_tablet_id

<< ", remote tablet id: " << remote_tablet_id;

// 1.2. get remote files

std::map<std::string, FileStat> remote_files;

RETURN_IF_ERROR(_list_with_checksum(remote_path, &remote_files));

if (remote_files.empty()) {

std::stringstream ss;

ss << "get nothing from remote path: " << remote_path;

LOG(WARNING) << ss.str();

return Status::InternalError(ss.str());

}

auto remote_hdr_file_path = [&remote_files, &remote_path](std::string& full_hdr_path,

size_t* hdr_file_len) {

for (auto iter = remote_files.begin(); iter != remote_files.end();) {

if (_end_with(iter->first, ".hdr")) {

*hdr_file_len = iter->second.size;

full_hdr_path = remote_path + "/" + iter->first + "." + iter->second.md5;

// remove hdr file from remote_files

iter = remote_files.erase(iter);

return true;

} else {

++iter;

}

}

return false;

};

size_t hdr_file_len;

std::string full_remote_hdr_path;

if (!remote_hdr_file_path(full_remote_hdr_path, &hdr_file_len)) {

std::stringstream ss;

ss << "failed to find hdr file from remote_path: " << remote_path;

LOG(WARNING) << ss.str();

return Status::InternalError(ss.str());

}

// 1.3. download hdr file

io::FileReaderOptions reader_options {

.cache_type = io::FileCachePolicy::NO_CACHE,

.is_doris_table = false,

.cache_base_path = "",

.file_size = static_cast<int64_t>(hdr_file_len),

.storage_resource_id {},

};

LOG(INFO) << "download hdr file: " << full_remote_hdr_path;

io::FileReaderSPtr hdr_reader = nullptr;

RETURN_IF_ERROR(_remote_fs->open_file(full_remote_hdr_path, &hdr_reader, &reader_options));

std::unique_ptr<char[]> read_buf = std::make_unique_for_overwrite<char[]>(hdr_file_len);

size_t read_len = 0;

Slice hdr_slice(read_buf.get(), hdr_file_len);

RETURN_IF_ERROR(hdr_reader->read_at(0, hdr_slice, &read_len));

if (read_len != hdr_file_len) {

std::stringstream ss;

ss << "failed to read hdr file: " << full_remote_hdr_path;

LOG(WARNING) << ss.str();

return Status::InternalError(ss.str());

}

RETURN_IF_ERROR(

_report_every(0, &tmp_counter, finished_num, total_num, TTaskType::type::DOWNLOAD));

// 1.4. make snapshot

std::unordered_map<std::string, std::string> file_mapping;

RETURN_IF_ERROR(_engine.cloud_snapshot_mgr().make_snapshot(

target_tablet_id, *_storage_resource, file_mapping, true, &hdr_slice));

LOG(INFO) << "finish to make snapshot for tablet: " << target_tablet_id;

// 1.5. download files

for (auto& nested_iter : remote_files) {

RETURN_IF_ERROR(_report_every(10, &report_counter, finished_num, total_num,

TTaskType::type::DOWNLOAD));

const std::string& remote_file = nested_iter.first;

const FileStat& file_stat = nested_iter.second;

auto find = file_mapping.find(remote_file);

if (find == file_mapping.end()) {

continue;

}

std::string target_file = find->second;

std::string full_remote_file = remote_path + "/" + remote_file + "." + file_stat.md5;

std::string full_target_file = target_path + "/" + target_file;

LOG(INFO) << "begin to download from " << full_remote_file << " to "

<< full_target_file;

io::FileReaderOptions nested_reader_options {

.cache_type = io::FileCachePolicy::NO_CACHE,

.is_doris_table = false,

.cache_base_path = "",

.file_size = static_cast<int64_t>(file_stat.size),

.storage_resource_id {},

};

io::FileReaderSPtr file_reader = nullptr;

RETURN_IF_ERROR(

_remote_fs->open_file(full_remote_file, &file_reader, &nested_reader_options));

io::FileWriterPtr file_writer = nullptr;

RETURN_IF_ERROR(storage_fs()->create_file(full_target_file, &file_writer));

size_t buf_size = config::s3_file_system_local_upload_buffer_size;

std::unique_ptr<char[]> transfer_buffer =

std::make_unique_for_overwrite<char[]>(buf_size);

size_t cur_offset = 0;

// (TODO) Add Verification that the length of the source file is consistent

// with the length of the target file

while (true) {

read_len = 0;

RETURN_IF_ERROR(file_reader->read_at(

cur_offset, Slice {transfer_buffer.get(), buf_size}, &read_len));

cur_offset += read_len;

if (read_len == 0) {

break;

}

RETURN_IF_ERROR(file_writer->append({transfer_buffer.get(), read_len}));

}

RETURN_IF_ERROR(file_writer->close());

}

finished_num++;

// (TODO) Add bvar metrics to track download time

} // end for src_to_dest_path

LOG(INFO) << "finished to download snapshots. job: " << _job_id << ", task id: " << _task_id;

return status;

}

#include "common/compile_check_avoid_end.h"

} // end namespace doris