// 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 "load/group_commit/wal/wal_manager.h"

#include <absl/strings/str_split.h>

#include <bvar/bvar.h>

#include <glog/logging.h>

#include <chrono>

#include <filesystem>

#include <shared_mutex>

#include <thread>

#include <unordered_map>

#include <vector>

#include "common/config.h"

#include "common/status.h"

#include "io/fs/local_file_system.h"

#include "load/group_commit/wal/wal_dirs_info.h"

#include "load/group_commit/wal/wal_reader.h"

#include "runtime/exec_env.h"

#include "runtime/fragment_mgr.h"

#include "util/parse_util.h"

namespace doris {

bvar::Status<size_t> g_wal_total_count("wal_total_count", 0);

bvar::Status<size_t> g_wal_max_count_per_table("wal_max_count_per_table", 0);

WalManager::WalManager(ExecEnv* exec_env, const std::string& wal_dir_list)

: _exec_env(exec_env),

_stop(false),

_stop_background_threads_latch(1),

_first_replay(true) {

_wal_dirs = absl::StrSplit(wal_dir_list, ";", absl::SkipWhitespace());

static_cast<void>(ThreadPoolBuilder("GroupCommitReplayWalThreadPool")

.set_min_threads(1)

.set_max_threads(config::group_commit_relay_wal_threads)

.build(&_thread_pool));

_wal_dirs_info = WalDirsInfo::create_unique();

}

WalManager::~WalManager() {

LOG(INFO) << "WalManager is destoried";

}

bool WalManager::is_running() {

return !_stop.load();

}

void WalManager::stop() {

if (!this->_stop.load()) {

this->_stop.store(true);

_stop_relay_wal();

_stop_background_threads_latch.count_down();

if (_replay_thread) {

_replay_thread->join();

}

if (_update_wal_dirs_info_thread) {

_update_wal_dirs_info_thread->join();

}

_thread_pool->shutdown();

LOG(INFO) << "WalManager is stopped";

}

}

Status WalManager::init() {

RETURN_IF_ERROR(_init_wal_dirs_conf());

RETURN_IF_ERROR(_init_wal_dirs());

RETURN_IF_ERROR(_init_wal_dirs_info());

return Thread::create(

"WalMgr", "replay_wal", [this]() { static_cast<void>(this->_replay_background()); },

&_replay_thread);

}

Status WalManager::_init_wal_dirs_conf() {

std::vector<std::string> tmp_dirs;

if (_wal_dirs.empty()) {

// default case.

for (const StorePath& path : ExecEnv::GetInstance()->store_paths()) {

tmp_dirs.emplace_back(path.path + "/wal");

}

} else {

// user config must be absolute path.

for (const std::string& wal_dir : _wal_dirs) {

if (std::filesystem::path(wal_dir).is_absolute()) {

tmp_dirs.emplace_back(wal_dir);

} else {

return Status::InternalError(

"BE config group_commit_replay_wal_dir has to be absolute path!");

}

}

}

_wal_dirs = tmp_dirs;

return Status::OK();

}

Status WalManager::_init_wal_dirs() {

bool exists = false;

for (auto wal_dir : _wal_dirs) {

std::string tmp_dir = wal_dir + "/" + _tmp;

LOG(INFO) << "wal_dir:" << wal_dir << ", tmp_dir:" << tmp_dir;

RETURN_IF_ERROR(io::global_local_filesystem()->exists(wal_dir, &exists));

if (!exists) {

RETURN_IF_ERROR(io::global_local_filesystem()->create_directory(wal_dir));

}

RETURN_IF_ERROR(io::global_local_filesystem()->exists(tmp_dir, &exists));

if (!exists) {

RETURN_IF_ERROR(io::global_local_filesystem()->create_directory(tmp_dir));

}

}

return Status::OK();

}

Status WalManager::_init_wal_dirs_info() {

for (const std::string& wal_dir : _wal_dirs) {

size_t available_bytes;

#ifndef BE_TEST

size_t disk_capacity_bytes;

RETURN_IF_ERROR(io::global_local_filesystem()->get_space_info(wal_dir, &disk_capacity_bytes,

&available_bytes));

#else

available_bytes = wal_limit_test_bytes;

#endif

bool is_percent = true;

int64_t wal_disk_limit = ParseUtil::parse_mem_spec(config::group_commit_wal_max_disk_limit,

-1, available_bytes, &is_percent);

if (wal_disk_limit < 0) {

return Status::InternalError(

"group_commit_wal_max_disk_limit config is wrong, please check your config!");

}

// if there are some wal files in wal dir, we need to add it to wal disk limit.

size_t wal_dir_size = 0;

#ifndef BE_TEST

RETURN_IF_ERROR(io::global_local_filesystem()->directory_size(wal_dir, &wal_dir_size));

#endif

if (is_percent) {

wal_disk_limit += wal_dir_size;

}

RETURN_IF_ERROR(_wal_dirs_info->add(wal_dir, wal_disk_limit, wal_dir_size, 0));

#ifdef BE_TEST

wal_limit_test_bytes = wal_disk_limit;

#endif

}

#ifndef BE_TEST

return Thread::create(

"WalMgr", "update_wal_dir_info",

[this]() { static_cast<void>(this->_update_wal_dir_info_thread()); },

&_update_wal_dirs_info_thread);

#else

return Status::OK();

#endif

}

void WalManager::add_wal_queue(int64_t table_id, int64_t wal_id) {

std::lock_guard<std::shared_mutex> wrlock(_wal_queue_lock);

LOG(INFO) << "add wal to queue, table_id: " << table_id << ", wal_id: " << wal_id;

auto it = _wal_queues.find(table_id);

if (it == _wal_queues.end()) {

std::set<int64_t> tmp_set;

tmp_set.insert(wal_id);

_wal_queues.emplace(table_id, tmp_set);

} else {

it->second.insert(wal_id);

}

}

void WalManager::erase_wal_queue(int64_t table_id, int64_t wal_id) {

std::lock_guard<std::shared_mutex> wrlock(_wal_queue_lock);

auto it = _wal_queues.find(table_id);

if (it != _wal_queues.end()) {

LOG(INFO) << "remove wal from queue, table_id: " << table_id << ", wal_id: " << wal_id;

it->second.erase(wal_id);

if (it->second.empty()) {

_wal_queues.erase(table_id);

}

}

}

size_t WalManager::get_wal_queue_size(int64_t table_id) {

std::lock_guard<std::shared_mutex> wrlock(_wal_queue_lock);

size_t count = 0;

if (table_id > 0) {

auto it = _wal_queues.find(table_id);

if (it != _wal_queues.end()) {

return it->second.size();

} else {

return 0;

}

} else {

// table_id is -1 meaning get all table wal size

size_t max_count_per_table = 0;

for (auto& [_, table_wals] : _wal_queues) {

size_t table_wal_count = table_wals.size();

count += table_wal_count;

if (table_wal_count > max_count_per_table) {

max_count_per_table = table_wal_count;

}

}

g_wal_max_count_per_table.set_value(max_count_per_table);

}

return count;

}

Status WalManager::create_wal_path(int64_t db_id, int64_t table_id, int64_t wal_id,

const std::string& label, std::string& base_path,

uint32_t wal_version) {

base_path = _wal_dirs_info->get_available_random_wal_dir();

std::stringstream ss;

ss << base_path << "/" << std::to_string(db_id) << "/" << std::to_string(table_id) << "/"

<< std::to_string(wal_version) << "_" << _exec_env->cluster_info()->backend_id << "_"

<< std::to_string(wal_id) << "_" << label;

{

std::lock_guard<std::shared_mutex> wrlock(_wal_path_lock);

auto it = _wal_path_map.find(wal_id);

if (it != _wal_path_map.end()) {

return Status::InternalError("wal_id {} already in wal_path_map", wal_id);

}

_wal_path_map.emplace(wal_id, ss.str());

}

return Status::OK();

}

Status WalManager::get_wal_path(int64_t wal_id, std::string& wal_path) {

std::shared_lock rdlock(_wal_path_lock);

auto it = _wal_path_map.find(wal_id);

if (it != _wal_path_map.end()) {

wal_path = _wal_path_map[wal_id];

} else {

return Status::InternalError("can not find wal_id {} in wal_path_map", wal_id);

}

return Status::OK();

}

Status WalManager::parse_wal_path(const std::string& file_name, int64_t& version,

int64_t& backend_id, int64_t& wal_id, std::string& label) {

try {

// find version

auto pos = file_name.find("_");

version = std::strtoll(file_name.substr(0, pos).c_str(), NULL, 10);

// find be id

auto substring1 = file_name.substr(pos + 1);

pos = substring1.find("_");

backend_id = std::strtoll(substring1.substr(0, pos).c_str(), NULL, 10);

// find wal id

auto substring2 = substring1.substr(pos + 1);

pos = substring2.find("_");

wal_id = std::strtoll(substring2.substr(0, pos).c_str(), NULL, 10);

// find label

label = substring2.substr(pos + 1);

VLOG_DEBUG << "version:" << version << "backend_id:" << backend_id << ",wal_id:" << wal_id

<< ",label:" << label;

} catch (const std::invalid_argument& e) {

return Status::InvalidArgument("Invalid format, {}", e.what());

}

return Status::OK();

}

Status WalManager::_load_wals() {

std::vector<ScanWalInfo> wals;

for (auto wal_dir : _wal_dirs) {

WARN_IF_ERROR(_scan_wals(wal_dir, wals), fmt::format("fail to scan wal dir={}", wal_dir));

}

for (const auto& wal : wals) {

bool exists = false;

WARN_IF_ERROR(io::global_local_filesystem()->exists(wal.wal_path, &exists),

fmt::format("fail to check exist on wal file={}", wal.wal_path));

if (!exists) {

continue;

}

LOG(INFO) << "find wal: " << wal.wal_path;

{

std::lock_guard<std::shared_mutex> wrlock(_wal_path_lock);

auto it = _wal_path_map.find(wal.wal_id);

if (it != _wal_path_map.end()) {

LOG(INFO) << "wal_id " << wal.wal_id << " already in wal_path_map, skip it";

continue;

}

_wal_path_map.emplace(wal.wal_id, wal.wal_path);

}

// this config is use for test p0 case in pipeline

if (config::group_commit_wait_replay_wal_finish) {

auto lock = std::make_shared<std::mutex>();

auto cv = std::make_shared<std::condition_variable>();

auto add_st = add_wal_cv_map(wal.wal_id, lock, cv);

if (!add_st.ok()) {

LOG(WARNING) << "fail to add wal_id " << wal.wal_id << " to wal_cv_map";

continue;

}

}

_exec_env->wal_mgr()->add_wal_queue(wal.tb_id, wal.wal_id);

WARN_IF_ERROR(add_recover_wal(wal.db_id, wal.tb_id, wal.wal_id, wal.wal_path),

fmt::format("Failed to add recover wal={}", wal.wal_path));

}

return Status::OK();

}

Status WalManager::_scan_wals(const std::string& wal_path, std::vector<ScanWalInfo>& res) {

bool exists = false;

auto last_total_size = res.size();

std::vector<io::FileInfo> dbs;

Status st = io::global_local_filesystem()->list(wal_path, false, &dbs, &exists);

if (!st.ok()) {

LOG(WARNING) << "failed list files for wal_dir=" << wal_path << ", st=" << st.to_string();

return st;

}

for (const auto& database_id : dbs) {

if (database_id.is_file || database_id.file_name == _tmp) {

continue;

}

std::vector<io::FileInfo> tables;

auto db_path = wal_path + "/" + database_id.file_name;

st = io::global_local_filesystem()->list(db_path, false, &tables, &exists);

if (!st.ok()) {

LOG(WARNING) << "failed list files for wal_dir=" << db_path

<< ", st=" << st.to_string();

return st;

}

for (const auto& table_id : tables) {

if (table_id.is_file) {

continue;

}

std::vector<io::FileInfo> wals;

auto table_path = db_path + "/" + table_id.file_name;

st = io::global_local_filesystem()->list(table_path, false, &wals, &exists);

if (!st.ok()) {

LOG(WARNING) << "failed list files for wal_dir=" << table_path

<< ", st=" << st.to_string();

return st;

}

if (wals.empty()) {

continue;

}

int64_t db_id = -1;

int64_t tb_id = -1;

try {

db_id = std::strtoll(database_id.file_name.c_str(), NULL, 10);

tb_id = std::strtoll(table_id.file_name.c_str(), NULL, 10);

} catch (const std::invalid_argument& e) {

return Status::InvalidArgument("Invalid format, {}", e.what());

}

for (const auto& wal : wals) {

int64_t version = -1;

int64_t backend_id = -1;

int64_t wal_id = -1;

std::string label = "";

auto parse_st = parse_wal_path(wal.file_name, version, backend_id, wal_id, label);

if (!parse_st.ok()) {

LOG(WARNING) << "fail to parse file=" << wal.file_name

<< ",st=" << parse_st.to_string();

continue;

}

auto wal_file = table_path + "/" + wal.file_name;

struct ScanWalInfo scan_wal_info;

scan_wal_info.wal_path = wal_file;

scan_wal_info.db_id = db_id;

scan_wal_info.tb_id = tb_id;

scan_wal_info.wal_id = wal_id;

scan_wal_info.be_id = backend_id;

res.emplace_back(scan_wal_info);

}

}

}

LOG(INFO) << "Finish list wal_dir=" << wal_path

<< ", wal count=" << std::to_string(res.size() - last_total_size);

return Status::OK();

}

Status WalManager::_replay_background() {

do {

if (_stop.load()) {

break;

}

// port == 0 means not received heartbeat yet

if (_exec_env->cluster_info() != nullptr &&

_exec_env->cluster_info()->master_fe_addr.port == 0) {

continue;

}

// replay residual wal,only replay once

bool expected = true;

if (_first_replay.compare_exchange_strong(expected, false)) {

RETURN_IF_ERROR(_load_wals());

}

g_wal_total_count.set_value(get_wal_queue_size(-1));

// replay wal of current process

std::vector<int64_t> replay_tables;

{

std::lock_guard<std::shared_mutex> wrlock(_table_lock);

auto it = _table_map.begin();

while (it != _table_map.end()) {

if (it->second->size() > 0) {

replay_tables.push_back(it->first);

}

it++;

}

}

for (const auto& table_id : replay_tables) {

RETURN_IF_ERROR(_thread_pool->submit_func([table_id, this] {

auto st = this->_table_map[table_id]->replay_wals();

if (!st.ok()) {

LOG(WARNING) << "failed to submit replay wal for table=" << table_id;

}

}));

}

} while (!_stop_background_threads_latch.wait_for(

std::chrono::seconds(config::group_commit_replay_wal_retry_interval_seconds)));

return Status::OK();

}

Status WalManager::add_recover_wal(int64_t db_id, int64_t table_id, int64_t wal_id,

std::string wal) {

std::lock_guard<std::shared_mutex> wrlock(_table_lock);

std::shared_ptr<WalTable> table_ptr;

auto it = _table_map.find(table_id);

if (it == _table_map.end()) {

table_ptr = std::make_shared<WalTable>(_exec_env, db_id, table_id);

_table_map.emplace(table_id, table_ptr);

} else {

table_ptr = it->second;

}

table_ptr->add_wal(wal_id, wal);

#ifndef BE_TEST

WARN_IF_ERROR(update_wal_dir_limit(get_base_wal_path(wal)),

"Failed to update wal dir limit while add recover wal!");

WARN_IF_ERROR(update_wal_dir_used(get_base_wal_path(wal)),

"Failed to update wal dir used while add recove wal!");

#endif

return Status::OK();

}

size_t WalManager::get_wal_table_size(int64_t table_id) {

std::shared_lock rdlock(_table_lock);

auto it = _table_map.find(table_id);

if (it != _table_map.end()) {

return it->second->size();

} else {

return 0;

}

}

void WalManager::_stop_relay_wal() {

std::lock_guard<std::shared_mutex> wrlock(_table_lock);

for (auto& [_, wal_table] : _table_map) {

wal_table->stop();

}

}

size_t WalManager::get_max_available_size() {

return _wal_dirs_info->get_max_available_size();

}

std::string WalManager::get_wal_dirs_info_string() {

return _wal_dirs_info->get_wal_dirs_info_string();

}

Status WalManager::update_wal_dir_limit(const std::string& wal_dir, size_t limit) {

return _wal_dirs_info->update_wal_dir_limit(wal_dir, limit);

}

Status WalManager::update_wal_dir_used(const std::string& wal_dir, size_t used) {

return _wal_dirs_info->update_wal_dir_used(wal_dir, used);

}

Status WalManager::update_wal_dir_estimated_wal_bytes(const std::string& wal_dir,

size_t increase_estimated_wal_bytes,

size_t decrease_estimated_wal_bytes) {

return _wal_dirs_info->update_wal_dir_estimated_wal_bytes(wal_dir, increase_estimated_wal_bytes,

decrease_estimated_wal_bytes);

}

Status WalManager::_update_wal_dir_info_thread() {

while (!_stop.load()) {

if (!ExecEnv::ready()) {

VLOG_DEBUG << "Sleep 1s to wait for storage engine init.";

std::this_thread::sleep_for(std::chrono::milliseconds(1000));

continue;

}

static_cast<void>(_wal_dirs_info->update_all_wal_dir_limit());

static_cast<void>(_wal_dirs_info->update_all_wal_dir_used());

LOG_EVERY_N(INFO, 100) << "Scheduled(every 10s) WAL info: " << get_wal_dirs_info_string();

std::this_thread::sleep_for(std::chrono::milliseconds(100));

}

return Status::OK();

}

Status WalManager::get_wal_dir_available_size(const std::string& wal_dir, size_t* available_bytes) {

return _wal_dirs_info->get_wal_dir_available_size(wal_dir, available_bytes);

}

std::string WalManager::get_base_wal_path(const std::string& wal_path_str) {

io::Path wal_path = wal_path_str;

for (int i = 0; i < 3; ++i) {

if (!wal_path.has_parent_path()) {

return "";

}

wal_path = wal_path.parent_path();

}

return wal_path.string();

}

Status WalManager::add_wal_cv_map(int64_t wal_id, std::shared_ptr<std::mutex> lock,

std::shared_ptr<std::condition_variable> cv) {

std::lock_guard<std::shared_mutex> wrlock(_wal_cv_lock);

auto it = _wal_cv_map.find(wal_id);

if (it != _wal_cv_map.end()) {

return Status::InternalError("wal {} is already in _wal_cv_map ", wal_id);

}

auto pair = std::make_pair(lock, cv);

_wal_cv_map.emplace(wal_id, pair);

LOG(INFO) << "add " << wal_id << " to _wal_cv_map";

return Status::OK();

}

Status WalManager::erase_wal_cv_map(int64_t wal_id) {

std::lock_guard<std::shared_mutex> wrlock(_wal_cv_lock);

if (_wal_cv_map.erase(wal_id)) {

LOG(INFO) << "erase " << wal_id << " from _wal_cv_map";

} else {

return Status::InternalError("fail to erase wal {} from wal_cv_map", wal_id);

}

return Status::OK();

}

Status WalManager::wait_replay_wal_finish(int64_t wal_id) {

std::shared_ptr<std::mutex> lock = nullptr;

std::shared_ptr<std::condition_variable> cv = nullptr;

auto st = get_lock_and_cv(wal_id, lock, cv);

if (st.ok()) {

std::unique_lock l(*(lock));

LOG(INFO) << "start wait " << wal_id;

if (cv->wait_for(l, std::chrono::seconds(180)) == std::cv_status::timeout) {

LOG(WARNING) << "wait for " << wal_id << " is time out";

}

LOG(INFO) << "get wal " << wal_id << ",finish wait";

RETURN_IF_ERROR(erase_wal_cv_map(wal_id));

LOG(INFO) << "erase wal " << wal_id;

}

return Status::OK();

}

Status WalManager::notify_relay_wal(int64_t wal_id) {

std::shared_ptr<std::mutex> lock = nullptr;

std::shared_ptr<std::condition_variable> cv = nullptr;

auto st = get_lock_and_cv(wal_id, lock, cv);

if (st.ok()) {

std::unique_lock l(*(lock));

cv->notify_all();

LOG(INFO) << "get wal " << wal_id << ",notify all";

}

return Status::OK();

}

Status WalManager::get_lock_and_cv(int64_t wal_id, std::shared_ptr<std::mutex>& lock,

std::shared_ptr<std::condition_variable>& cv) {

std::lock_guard<std::shared_mutex> wrlock(_wal_cv_lock);

auto it = _wal_cv_map.find(wal_id);

if (it == _wal_cv_map.end()) {

return Status::InternalError("cannot find txn {} in _wal_cv_map", wal_id);

}

lock = it->second.first;

cv = it->second.second;

return Status::OK();

}

Status WalManager::delete_wal(int64_t table_id, int64_t wal_id) {

std::string wal_path;

{

std::lock_guard<std::shared_mutex> wrlock(_wal_path_lock);

auto it = _wal_path_map.find(wal_id);

if (it != _wal_path_map.end()) {

wal_path = it->second;

auto st = io::global_local_filesystem()->delete_file(wal_path);

if (st.ok()) {

LOG(INFO) << "delete wal=" << wal_path;

} else {

LOG(WARNING) << "failed to delete wal=" << wal_path << ", st=" << st.to_string();

}

_wal_path_map.erase(wal_id);

}

}

erase_wal_queue(table_id, wal_id);

return Status::OK();

}

Status WalManager::rename_to_tmp_path(const std::string wal, int64_t table_id, int64_t wal_id) {

io::Path wal_path = wal;

std::list<std::string> path_element;

for (int i = 0; i < 3; ++i) {

if (!wal_path.has_parent_path()) {

return Status::InternalError("parent path is not enough when rename " + wal);

}

path_element.push_front(wal_path.filename().string());

wal_path = wal_path.parent_path();

}

wal_path.append(_tmp);

for (auto path : path_element) {

wal_path.append(path);

}

bool exists = false;

RETURN_IF_ERROR(io::global_local_filesystem()->exists(wal_path.parent_path(), &exists));

if (!exists) {

RETURN_IF_ERROR(io::global_local_filesystem()->create_directory(wal_path.parent_path()));

}

auto res = std::rename(wal.c_str(), wal_path.string().c_str());

if (res < 0) {

LOG(INFO) << "failed to rename wal from " << wal << " to " << wal_path.string();

return Status::InternalError("rename fail on path " + wal);

}

LOG(INFO) << "rename wal from " << wal << " to " << wal_path.string();

{

std::lock_guard<std::shared_mutex> wrlock(_wal_path_lock);

auto it = _wal_path_map.find(wal_id);

if (it != _wal_path_map.end()) {

_wal_path_map.erase(wal_id);

} else {

LOG(WARNING) << "can't find " << wal_id << " in _wal_path_map when trying to rename";

}

}

erase_wal_queue(table_id, wal_id);

return Status::OK();

}

} // namespace doris