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

#include <bvar/bvar.h>

#include "cloud/cloud_meta_mgr.h"

#include "cloud/cloud_storage_engine.h"

#include "cloud/config.h"

#include "common/logging.h"

#include "common/status.h"

#include "load/stream_load/stream_load_context.h"

#include "util/debug_points.h"

namespace doris {

bvar::Adder<uint64_t> stream_load_commit_retry_counter("stream_load_commit_retry_counter");

bvar::Window<bvar::Adder<uint64_t>> stream_load_commit_retry_counter_minute(

        "stream_load_commit_retry_counter", "1m", &stream_load_commit_retry_counter, 60);

enum class TxnOpParamType : int {

    ILLEGAL,

    WITH_TXN_ID,

    WITH_LABEL,

};

CloudStreamLoadExecutor::CloudStreamLoadExecutor(ExecEnv* exec_env)

        : StreamLoadExecutor(exec_env) {}

CloudStreamLoadExecutor::~CloudStreamLoadExecutor() = default;

Status CloudStreamLoadExecutor::pre_commit_txn(StreamLoadContext* ctx) {

    auto st = _exec_env->storage_engine().to_cloud().meta_mgr().precommit_txn(*ctx);

    if (!st.ok()) {

        LOG(WARNING) << "Failed to precommit txn: " << st << ", " << ctx->brief();

        return st;

    }

    ctx->need_rollback = false;

    return st;

}

Status CloudStreamLoadExecutor::operate_txn_2pc(StreamLoadContext* ctx) {

    std::stringstream ss;

    ss << "db_id=" << ctx->db_id << " txn_id=" << ctx->txn_id << " label=" << ctx->label

       << " txn_2pc_op=" << ctx->txn_operation;

    std::string op_info = ss.str();

    VLOG_DEBUG << "operate_txn_2pc " << op_info;

    TxnOpParamType topt = ctx->txn_id > 0       ? TxnOpParamType::WITH_TXN_ID

                          : !ctx->label.empty() ? TxnOpParamType::WITH_LABEL

                                                : TxnOpParamType::ILLEGAL;

    Status st = Status::InternalError<false>("impossible branch reached, " + op_info);

    if (ctx->txn_operation.compare("commit") == 0) {

        if (!config::enable_stream_load_commit_txn_on_be) {

            VLOG_DEBUG << "2pc commit stream load txn with FE support: " << op_info;

            st = StreamLoadExecutor::operate_txn_2pc(ctx);

        } else if (topt == TxnOpParamType::WITH_TXN_ID) {

            VLOG_DEBUG << "2pc commit stream load txn directly: " << op_info;

            st = _exec_env->storage_engine().to_cloud().meta_mgr().commit_txn(*ctx, true);

        } else if (topt == TxnOpParamType::WITH_LABEL) {

            VLOG_DEBUG << "2pc commit stream load txn with FE support: " << op_info;

            st = StreamLoadExecutor::operate_txn_2pc(ctx);

        } else {

            st = Status::InternalError<false>(

                    "failed to 2pc commit txn, with TxnOpParamType::illegal input, " + op_info);

        }

    } else if (ctx->txn_operation.compare("abort") == 0) {

        if (topt == TxnOpParamType::WITH_TXN_ID) {

            LOG(INFO) << "2pc abort stream load txn directly: " << op_info;

            st = _exec_env->storage_engine().to_cloud().meta_mgr().abort_txn(*ctx);

            WARN_IF_ERROR(st, "failed to rollback txn " + op_info);

        } else if (topt == TxnOpParamType::WITH_LABEL) { // maybe a label send to FE to abort

            VLOG_DEBUG << "2pc abort stream load txn with FE support: " << op_info;

            StreamLoadExecutor::rollback_txn(ctx);

            st = Status::OK();

        } else {

            st = Status::InternalError<false>("failed abort txn, with illegal input, " + op_info);

        }

    } else {

        std::string msg =

                "failed to operate_txn_2pc, unrecognized operation: " + ctx->txn_operation;

        LOG(WARNING) << msg << " " << op_info;

        st = Status::InternalError<false>(msg + " " + op_info);

    }

    WARN_IF_ERROR(st, "failed to operate_txn_2pc " + op_info)

    return st;

}

Status CloudStreamLoadExecutor::commit_txn(StreamLoadContext* ctx) {

    DBUG_EXECUTE_IF("StreamLoadExecutor.commit_txn.block", DBUG_BLOCK);

    DBUG_EXECUTE_IF("StreamLoadExecutor.commit_txn.crash", {

        LOG(INFO) << "debug point " << DP_NAME << " trigger crash";

        volatile int* p = nullptr;

        *p = 1;

    });

    // forward to fe to excute commit transaction for MoW table

    if (ctx->is_mow_table() || !config::enable_stream_load_commit_txn_on_be ||

        ctx->load_type == TLoadType::ROUTINE_LOAD) {

        Status st;

        int retry_times = 0;

        while (retry_times < config::mow_stream_load_commit_retry_times) {

            st = StreamLoadExecutor::commit_txn(ctx);

            // DELETE_BITMAP_LOCK_ERROR will be retried

            if (st.ok() || !st.is<ErrorCode::DELETE_BITMAP_LOCK_ERROR>()) {

                break;

            }

            LOG_WARNING("Failed to commit txn")

                    .tag("txn_id", ctx->txn_id)

                    .tag("retry_times", retry_times)

                    .error(st);

            retry_times++;

            stream_load_commit_retry_counter << 1;

        }

        return st;

    }

    auto st = _exec_env->storage_engine().to_cloud().meta_mgr().commit_txn(*ctx, false);

    if (!st.ok()) {

        LOG(WARNING) << "Failed to commit txn: " << st << ", " << ctx->brief();

        return st;

    }

    ctx->need_rollback = false;

    return st;

}

void CloudStreamLoadExecutor::rollback_txn(StreamLoadContext* ctx) {

    std::stringstream ss;

    ss << "db_id=" << ctx->db_id << " txn_id=" << ctx->txn_id << " label=" << ctx->label;

    std::string op_info = ss.str();

    LOG(INFO) << "rollback stream load txn " << op_info;

    TxnOpParamType topt = ctx->txn_id > 0       ? TxnOpParamType::WITH_TXN_ID

                          : !ctx->label.empty() ? TxnOpParamType::WITH_LABEL

                                                : TxnOpParamType::ILLEGAL;

    if (topt == TxnOpParamType::WITH_TXN_ID && ctx->load_type != TLoadType::ROUTINE_LOAD) {

        VLOG_DEBUG << "abort stream load txn directly: " << op_info;

        WARN_IF_ERROR(_exec_env->storage_engine().to_cloud().meta_mgr().abort_txn(*ctx),

                      "failed to rollback txn " + op_info);

    } else { // maybe a label send to FE to abort

        // does not care about the return status

        // ctx->db_id > 0 && !ctx->label.empty()

        VLOG_DEBUG << "abort stream load txn with FE support: " << op_info;

        StreamLoadExecutor::rollback_txn(ctx);

    }

}

} // namespace doris