// 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/cluster_info.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();

            {

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

                _last_replay_wal_failed_reason =

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

                        ", st=" + st.to_string().substr(0, 100);

            }

            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);

        }

        if (_replay_wal_map.empty()) {

            _last_replay_wal_failed_reason.clear();

        }

    }

    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();

}

std::string WalTable::get_last_replay_wal_failed_reason() const {

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

    return _last_replay_wal_failed_reason;

}

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