// 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_table.h"

#include <absl/strings/str_split.h>

#include <thrift/protocol/TDebugProtocol.h>

#include "io/fs/local_file_system.h"

#include "io/fs/stream_load_pipe.h"

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

#include "runtime/fragment_mgr.h"

#include "service/http/action/http_stream.h"

#include "service/http/action/stream_load.h"

#include "service/http/ev_http_server.h"

#include "service/http/http_common.h"

#include "service/http/http_headers.h"

#include "service/http/utils.h"

#include "util/client_cache.h"

#include "util/path_util.h"

#include "util/thrift_rpc_helper.h"

namespace doris {

bvar::Adder<uint64_t> wal_fail("group_commit_wal_fail");

WalTable::WalTable(ExecEnv* exec_env, int64_t db_id, int64_t table_id)

: _exec_env(exec_env), _db_id(db_id), _table_id(table_id) {

_http_stream_action = std::make_shared<HttpStreamAction>(exec_env);

}

WalTable::~WalTable() {}

void WalTable::add_wal(int64_t wal_id, std::string wal) {

std::lock_guard<std::mutex> lock(_replay_wal_lock);

LOG(INFO) << "add replay wal=" << wal;

auto wal_info = std::make_shared<WalInfo>(wal_id, wal, 0, UnixMillis());

_replay_wal_map.emplace(wal, wal_info);

}

void WalTable::_pick_relay_wals() {

std::lock_guard<std::mutex> lock(_replay_wal_lock);

std::vector<std::string> need_replay_wals;

std::vector<std::string> need_erase_wals;

for (const auto& [wal_path, wal_info] : _replay_wal_map) {

if (config::group_commit_wait_replay_wal_finish &&

wal_info->get_retry_num() >= config::group_commit_replay_wal_retry_num) {

LOG(WARNING) << "failed to replay wal=" << wal_path << " after retry "

<< wal_info->get_retry_num() << " times";

[[maybe_unused]] auto st = _exec_env->wal_mgr()->rename_to_tmp_path(

wal_path, _table_id, wal_info->get_wal_id());

auto notify_st = _exec_env->wal_mgr()->notify_relay_wal(wal_info->get_wal_id());

if (!notify_st.ok()) {

LOG(WARNING) << "notify wal " << wal_info->get_wal_id() << " fail";

}

need_erase_wals.push_back(wal_path);

continue;

}

if (_need_replay(wal_info)) {

need_replay_wals.push_back(wal_path);

}

}

for (const auto& wal : need_erase_wals) {

_replay_wal_map.erase(wal);

}

std::sort(need_replay_wals.begin(), need_replay_wals.end());

for (const auto& wal : need_replay_wals) {

_replaying_queue.emplace_back(_replay_wal_map[wal]);

_replay_wal_map.erase(wal);

}

}

Status WalTable::_relay_wal_one_by_one() {

std::vector<std::shared_ptr<WalInfo>> need_retry_wals;

for (auto wal_info : _replaying_queue) {

wal_info->add_retry_num();

Status st;

int64_t file_size = 0;

std::filesystem::path file_path(wal_info->get_wal_path());

if (!std::filesystem::exists(file_path)) {

st = Status::InternalError("wal file {} does not exist", wal_info->get_wal_path());

} else {

file_size = std::filesystem::file_size(file_path);

st = _replay_wal_internal(wal_info->get_wal_path());

}

auto msg = st.msg();

if (st.ok() || st.is<ErrorCode::PUBLISH_TIMEOUT>() || st.is<ErrorCode::NOT_FOUND>() ||

st.is<ErrorCode::DATA_QUALITY_ERROR>() ||

(msg.find("has already been used") != msg.npos &&

(msg.find("COMMITTED") != msg.npos || msg.find("VISIBLE") != msg.npos))) {

LOG(INFO) << "succeed to replay wal=" << wal_info->get_wal_path()

<< ", st=" << st.to_string() << ", file size=" << file_size;

// delete wal

WARN_IF_ERROR(_exec_env->wal_mgr()->delete_wal(_table_id, wal_info->get_wal_id()),

"failed to delete wal=" + wal_info->get_wal_path());

if (config::group_commit_wait_replay_wal_finish) {

RETURN_IF_ERROR(_exec_env->wal_mgr()->notify_relay_wal(wal_info->get_wal_id()));

}

} else {

doris::wal_fail << 1;

LOG(WARNING) << "failed to replay wal=" << wal_info->get_wal_path()

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

need_retry_wals.push_back(wal_info);

}

}

{

std::lock_guard<std::mutex> lock(_replay_wal_lock);

_replaying_queue.clear();

for (auto retry_wal_info : need_retry_wals) {

_replay_wal_map.emplace(retry_wal_info->get_wal_path(), retry_wal_info);

}

}

return Status::OK();

}

Status WalTable::replay_wals() {

{

std::lock_guard<std::mutex> lock(_replay_wal_lock);

if (_replay_wal_map.empty()) {

LOG(INFO) << "_replay_wal_map is empty, skip relaying for table_id=" << _table_id;

return Status::OK();

}

if (!_replaying_queue.empty()) {

LOG(INFO) << "_replaying_queue is not empty, skip relaying for table_id=" << _table_id;

return Status::OK();

}

}

VLOG_DEBUG << "Start replay wals for db=" << _db_id << ", table=" << _table_id

<< ", wal size=" << _replay_wal_map.size();

_pick_relay_wals();

RETURN_IF_ERROR(_relay_wal_one_by_one());

return Status::OK();

}

bool WalTable::_need_replay(std::shared_ptr<WalInfo> wal_info) {

if (config::group_commit_wait_replay_wal_finish) {

return true;

}

#ifndef BE_TEST

int64_t replay_interval = 0;

if (wal_info->get_retry_num() >= config::group_commit_replay_wal_retry_num) {

replay_interval =

int64_t(pow(2, config::group_commit_replay_wal_retry_num) *

config::group_commit_replay_wal_retry_interval_seconds * 1000 +

(wal_info->get_retry_num() - config::group_commit_replay_wal_retry_num) *

config::group_commit_replay_wal_retry_interval_max_seconds * 1000);

} else {

replay_interval = int64_t(pow(2, wal_info->get_retry_num()) *

config::group_commit_replay_wal_retry_interval_seconds * 1000);

}

return UnixMillis() - wal_info->get_start_time_ms() >= replay_interval;

#else

return true;

#endif

}

Status WalTable::_try_abort_txn(int64_t db_id, std::string& label) {

TLoadTxnRollbackRequest request;

// this is a fake, fe not check it now

// should be removed in 3.1, use token instead

request.__set_auth_code(0);

request.__set_token(_exec_env->cluster_info()->curr_auth_token);

request.__set_db_id(db_id);

request.__set_label(label);

request.__set_reason("relay wal with label " + label);

TLoadTxnRollbackResult result;

TNetworkAddress master_addr = _exec_env->cluster_info()->master_fe_addr;

auto st = ThriftRpcHelper::rpc<FrontendServiceClient>(

master_addr.hostname, master_addr.port,

[&request, &result](FrontendServiceConnection& client) {

client->loadTxnRollback(result, request);

});

if (st.ok()) {

auto result_status = Status::create<false>(result.status);

LOG(INFO) << "abort label " << label << ", result_status:" << result_status;

return result_status;

} else {

LOG(WARNING) << "abort label " << label << ", rpc error:" << st;

return st;

}

}

Status WalTable::_replay_wal_internal(const std::string& wal) {

LOG(INFO) << "start replay wal=" << wal;

int64_t version = -1;

int64_t backend_id = -1;

int64_t wal_id = -1;

std::string label = "";

io::Path wal_path = wal;

auto file_name = wal_path.filename().string();

RETURN_IF_ERROR(WalManager::parse_wal_path(file_name, version, backend_id, wal_id, label));

#ifndef BE_TEST

if (!config::group_commit_wait_replay_wal_finish) {

[[maybe_unused]] auto st = _try_abort_txn(_db_id, label);

}

#endif

DBUG_EXECUTE_IF("WalTable.replay_wals.stop",

{ return Status::InternalError("WalTable.replay_wals.stop"); });

return _replay_one_wal_with_streamload(wal_id, wal, label);

}

Status WalTable::_construct_sql_str(const std::string& wal, const std::string& label,

std::string& sql_str) {

std::string columns;

RETURN_IF_ERROR(_read_wal_header(wal, columns));

std::vector<std::string> column_id_vector =

absl::StrSplit(columns, ",", absl::SkipWhitespace());

std::map<int64_t, std::string> column_info_map;

RETURN_IF_ERROR(_get_column_info(_db_id, _table_id, column_info_map));

std::stringstream ss_name;

for (auto column_id_str : column_id_vector) {

try {

int64_t column_id = std::strtoll(column_id_str.c_str(), NULL, 10);

auto it = column_info_map.find(column_id);

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

ss_name << "`" << it->second << "`,";

column_info_map.erase(column_id);

}

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

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

}

}

auto name = ss_name.str().substr(0, ss_name.str().size() - 1);

std::stringstream ss;

ss << "insert into doris_internal_table_id(" << _table_id << ") WITH LABEL " << label << " ("

<< name << ") select " << name << " from http_stream(\"format\" = \"wal\", \"table_id\" = \""

<< std::to_string(_table_id) << "\")";

sql_str = ss.str().data();

return Status::OK();

}

Status WalTable::_handle_stream_load(int64_t wal_id, const std::string& wal,

const std::string& label) {

std::string sql_str;

RETURN_IF_ERROR(_construct_sql_str(wal, label, sql_str));

std::shared_ptr<StreamLoadContext> ctx = std::make_shared<StreamLoadContext>(_exec_env);

ctx->sql_str = sql_str;

ctx->db_id = _db_id;

ctx->table_id = _table_id;

ctx->wal_id = wal_id;

ctx->label = label;

ctx->need_commit_self = false;

ctx->auth.token = _exec_env->cluster_info()->curr_auth_token;

ctx->auth.user = "admin";

ctx->group_commit = false;

ctx->load_type = TLoadType::MANUL_LOAD;

ctx->load_src_type = TLoadSourceType::RAW;

ctx->max_filter_ratio = 1;

auto st = _http_stream_action->process_put(nullptr, ctx);

DBUG_EXECUTE_IF("WalTable::_handle_stream_load.fail",

{ st = Status::InternalError("WalTable::_handle_stream_load.fail"); });

if (st.ok()) {

// wait stream load finish

RETURN_IF_ERROR(ctx->load_status_future.get());

if (ctx->status.ok()) {

// deprecated and should be removed in 3.1, use token instead.

ctx->auth.auth_code = wal_id;

st = _exec_env->stream_load_executor()->commit_txn(ctx.get());

} else {

st = ctx->status;

}

}

if (!st.ok()) {

_exec_env->stream_load_executor()->rollback_txn(ctx.get());

}

return st;

}

Status WalTable::_replay_one_wal_with_streamload(int64_t wal_id, const std::string& wal,

const std::string& label) {

#ifndef BE_TEST

return _handle_stream_load(wal_id, wal, label);

#else

return Status::OK();

#endif

}

void WalTable::stop() {

do {

{

std::lock_guard<std::mutex> lock(_replay_wal_lock);

if (_replay_wal_map.empty() && _replaying_queue.empty()) {

break;

}

LOG(INFO) << "stopping wal_table,wait for relay wal task done, now "

<< _replay_wal_map.size() << " wals wait to replay, "

<< _replaying_queue.size() << " wals are replaying";

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

}

} while (true);

}

size_t WalTable::size() {

std::lock_guard<std::mutex> lock(_replay_wal_lock);

return _replay_wal_map.size() + _replaying_queue.size();

}

Status WalTable::_get_column_info(int64_t db_id, int64_t tb_id,

std::map<int64_t, std::string>& column_info_map) {

TGetColumnInfoRequest request;

request.__set_db_id(db_id);

request.__set_table_id(tb_id);

TGetColumnInfoResult result;

Status status;

TNetworkAddress master_addr = _exec_env->cluster_info()->master_fe_addr;

if (master_addr.hostname.empty() || master_addr.port == 0) {

status = Status::InternalError<false>("Have not get FE Master heartbeat yet");

} else {

RETURN_IF_ERROR(ThriftRpcHelper::rpc<FrontendServiceClient>(

master_addr.hostname, master_addr.port,

[&request, &result](FrontendServiceConnection& client) {

client->getColumnInfo(result, request);

}));

status = Status::create<false>(result.status);

if (!status.ok()) {

return status;

}

std::vector<TColumnInfo> column_element = result.columns;

for (auto column : column_element) {

auto column_name = column.column_name;

auto column_id = column.column_id;

column_info_map.emplace(column_id, column_name);

}

}

return status;

}

Status WalTable::_read_wal_header(const std::string& wal_path, std::string& columns) {

std::shared_ptr<doris::WalFileReader> wal_reader =

std::make_shared<doris::WalFileReader>(wal_path);

RETURN_IF_ERROR(wal_reader->init());

uint32_t version = 0;

RETURN_IF_ERROR(wal_reader->read_header(version, columns));

VLOG_DEBUG << "wal=" << wal_path << ",version=" << std::to_string(version)

<< ",columns=" << columns;

RETURN_IF_ERROR(wal_reader->finalize());

return Status::OK();

}

} // namespace doris