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

#include <fmt/format.h>

#include "exprs/rpc_fn_comm.h"
#include "runtime/fragment_mgr.h"
#include "util/brpc_client_cache.h"
#include "vec/columns/column.h"
#include "vec/columns/column_vector.h"
#include "vec/core/block.h"
#include "vec/core/column_numbers.h"
#include "vec/data_types/data_type_decimal.h"
#include "vec/data_types/data_type_nullable.h"

namespace doris {

RPCFn::RPCFn(RuntimeState* state, const TFunction& fn, int fn_ctx_id, bool is_agg)
        : _state(state), _fn(fn), _fn_ctx_id(fn_ctx_id), _is_agg(is_agg) {
    _client = ExecEnv::GetInstance()->brpc_function_client_cache()->get_client(_server_addr);
    if (!_is_agg) {
        _function_name = _fn.scalar_fn.symbol;
        _server_addr = _fn.hdfs_location;
        _signature = fmt::format("{}: [{}/{}]", _fn.name.function_name, _fn.hdfs_location,
                                 _fn.scalar_fn.symbol);
    }
}

RPCFn::RPCFn(const TFunction& fn, bool is_agg) : RPCFn(nullptr, fn, -1, is_agg) {}

RPCFn::RPCFn(RuntimeState* state, const TFunction& fn, AggregationStep step, bool is_agg)
        : RPCFn(nullptr, fn, -1, is_agg) {
    _step = step;
    DCHECK(is_agg) << "Only used for agg fns";
    switch (_step) {
    case INIT: {
        _function_name = _fn.aggregate_fn.init_fn_symbol;
        _server_addr = _fn.hdfs_location;
        _signature = fmt::format("{}: [{}/{}]", _fn.name.function_name, _fn.hdfs_location,
                                 _fn.aggregate_fn.init_fn_symbol);
        break;
    }
    case UPDATE: {
        _function_name = _fn.aggregate_fn.init_fn_symbol;
        break;
    }
    case MERGE: {
        _function_name = _fn.aggregate_fn.merge_fn_symbol;
        break;
    }
    case SERIALIZE: {
        _function_name = _fn.aggregate_fn.serialize_fn_symbol;
        break;
    }
    case GET_VALUE: {
        _function_name = _fn.aggregate_fn.get_value_fn_symbol;
        break;
    }
    case FINALIZE: {
        _function_name = _fn.aggregate_fn.finalize_fn_symbol;
        break;
    }
    case REMOVE: {
        _function_name = _fn.aggregate_fn.remove_fn_symbol;
        break;
    }

    default:
        CHECK(false) << "invalid AggregationStep: " << _step;
        break;
    }
    _server_addr = _fn.hdfs_location;
    _signature = fmt::format("{}: [{}/{}]", _fn.name.function_name, _server_addr, _function_name);
}

Status RPCFn::call_internal(ExprContext* context, TupleRow* row, PFunctionCallResponse* response,
                            const std::vector<Expr*>& exprs) {
    FunctionContext* fn_ctx = context->fn_context(_fn_ctx_id);
    PFunctionCallRequest request;
    request.set_function_name(_function_name);
    for (int i = 0; i < exprs.size(); ++i) {
        PValues* arg = request.add_args();
        void* src_slot = context->get_value(exprs[i], row);
        PGenericType* ptype = arg->mutable_type();
        if (src_slot == nullptr) {
            arg->set_has_null(true);
            arg->add_null_map(true);
        } else {
            arg->set_has_null(false);
        }
        switch (exprs[i]->type().type) {
        case TYPE_BOOLEAN: {
            ptype->set_id(PGenericType::BOOLEAN);
            arg->add_bool_value(*(bool*)src_slot);
            break;
        }
        case TYPE_TINYINT: {
            ptype->set_id(PGenericType::INT8);
            arg->add_int32_value(*(int8_t*)src_slot);
            break;
        }
        case TYPE_SMALLINT: {
            ptype->set_id(PGenericType::INT16);
            arg->add_int32_value(*(int16_t*)src_slot);
            break;
        }
        case TYPE_INT: {
            ptype->set_id(PGenericType::INT32);
            arg->add_int32_value(*(int*)src_slot);
            break;
        }
        case TYPE_BIGINT: {
            ptype->set_id(PGenericType::INT64);
            arg->add_int64_value(*(int64_t*)src_slot);
            break;
        }
        case TYPE_LARGEINT: {
            ptype->set_id(PGenericType::INT128);
            char buffer[sizeof(__int128)];
            memcpy(buffer, src_slot, sizeof(__int128));
            arg->add_bytes_value(buffer, sizeof(__int128));
            break;
        }
        case TYPE_DOUBLE: {
            ptype->set_id(PGenericType::DOUBLE);
            arg->add_double_value(*(double*)src_slot);
            break;
        }
        case TYPE_FLOAT: {
            ptype->set_id(PGenericType::FLOAT);
            arg->add_float_value(*(float*)src_slot);
            break;
        }
        case TYPE_VARCHAR:
        case TYPE_STRING:
        case TYPE_CHAR: {
            ptype->set_id(PGenericType::STRING);
            StringValue value = *reinterpret_cast<StringValue*>(src_slot);
            arg->add_string_value(value.ptr, value.len);
            break;
        }
        case TYPE_HLL: {
            ptype->set_id(PGenericType::HLL);
            StringValue value = *reinterpret_cast<StringValue*>(src_slot);
            arg->add_string_value(value.ptr, value.len);
            break;
        }
        case TYPE_OBJECT: {
            ptype->set_id(PGenericType::BITMAP);
            StringValue value = *reinterpret_cast<StringValue*>(src_slot);
            arg->add_string_value(value.ptr, value.len);
            break;
        }
        case TYPE_DECIMALV2: {
            ptype->set_id(PGenericType::DECIMAL128);
            ptype->mutable_decimal_type()->set_precision(exprs[i]->type().precision);
            ptype->mutable_decimal_type()->set_scale(exprs[i]->type().scale);
            char buffer[sizeof(__int128)];
            memcpy(buffer, src_slot, sizeof(__int128));
            arg->add_bytes_value(buffer, sizeof(__int128));
            break;
        }
        case TYPE_DATE: {
            ptype->set_id(PGenericType::DATE);
            const auto* time_val = (const DateTimeValue*)(src_slot);
            PDateTime* date_time = arg->add_datetime_value();
            date_time->set_day(time_val->day());
            date_time->set_month(time_val->month());
            date_time->set_year(time_val->year());
            break;
        }
        case TYPE_DATETIME: {
            ptype->set_id(PGenericType::DATETIME);
            const auto* time_val = (const DateTimeValue*)(src_slot);
            PDateTime* date_time = arg->add_datetime_value();
            date_time->set_day(time_val->day());
            date_time->set_month(time_val->month());
            date_time->set_year(time_val->year());
            date_time->set_hour(time_val->hour());
            date_time->set_minute(time_val->minute());
            date_time->set_second(time_val->second());
            date_time->set_microsecond(time_val->microsecond());
            break;
        }
        case TYPE_TIME: {
            ptype->set_id(PGenericType::DATETIME);
            const auto* time_val = (const DateTimeValue*)(src_slot);
            PDateTime* date_time = arg->add_datetime_value();
            date_time->set_hour(time_val->hour());
            date_time->set_minute(time_val->minute());
            date_time->set_second(time_val->second());
            date_time->set_microsecond(time_val->microsecond());
            break;
        }
        default: {
            std::string error_msg =
                    fmt::format("data time not supported: {}", exprs[i]->type().type);
            fn_ctx->set_error(error_msg.c_str());
            cancel(error_msg);
            break;
        }
        }
    }

    brpc::Controller cntl;
    _client->fn_call(&cntl, &request, response, nullptr);
    if (cntl.Failed()) {
        std::string error_msg =
                fmt::format("call rpc function {} failed: {}", _signature, cntl.ErrorText());
        fn_ctx->set_error(error_msg.c_str());
        cancel(error_msg);
        return Status::InternalError(error_msg);
    }
    if (!response->has_status() || response->result_size() == 0) {
        std::string error_msg =
                fmt::format("call rpc function {} failed: status or result is not set: {}",
                            _signature, response->status().DebugString());
        fn_ctx->set_error(error_msg.c_str());
        cancel(error_msg);
        return Status::InternalError(error_msg);
    }
    if (response->status().status_code() != 0) {
        std::string error_msg = fmt::format("call rpc function {} failed: {}", _signature,
                                            response->status().DebugString());
        fn_ctx->set_error(error_msg.c_str());
        cancel(error_msg);
        return Status::InternalError(error_msg);
    }
    return Status::OK();
}

void RPCFn::cancel(const std::string& msg) {
    _state->exec_env()->fragment_mgr()->cancel(_state->fragment_instance_id(),
                                               PPlanFragmentCancelReason::CALL_RPC_ERROR, msg);
}

Status RPCFn::vec_call(FunctionContext* context, vectorized::Block& block,
                       const vectorized::ColumnNumbers& arguments, size_t result,
                       size_t input_rows_count) {
    PFunctionCallRequest request;
    PFunctionCallResponse response;
    request.set_function_name(_function_name);
    convert_block_to_proto(block, arguments, input_rows_count, &request);
    brpc::Controller cntl;
    _client->fn_call(&cntl, &request, &response, nullptr);
    if (cntl.Failed()) {
        return Status::InternalError("call to rpc function {} failed: {}", _signature,
                                     cntl.ErrorText());
    }
    if (!response.has_status() || response.result_size() == 0) {
        return Status::InternalError("call rpc function {} failed: status or result is not set.",
                                     _signature);
    }
    if (response.status().status_code() != 0) {
        return Status::InternalError("call to rpc function {} failed: {}", _signature,
                                     response.status().DebugString());
    }
    convert_to_block(block, response.result(0), result);
    return Status::OK();
}
} // namespace doris