/**
 * Copyright (c) 2021 OceanBase
 * OceanBase CE is licensed under Mulan PubL v2.
 * You can use this software according to the terms and conditions of the Mulan PubL v2.
 * You may obtain a copy of Mulan PubL v2 at:
 *          http://license.coscl.org.cn/MulanPubL-2.0
 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
 * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
 * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 * See the Mulan PubL v2 for more details.
 */

#define USING_LOG_PREFIX SQL_DAS
#include "sql/das/ob_das_task.h"
#include "sql/das/ob_das_scan_op.h"
#include "sql/ob_sql_trans_control.h"
#include "sql/das/ob_das_rpc_processor.h"
#include "sql/engine/px/ob_px_util.h"

namespace oceanbase
{
namespace common
{
namespace serialization
{
template <>
struct EnumEncoder<false, const sql::ObDASBaseCtDef*> : sql::DASCtEncoder<sql::ObDASBaseCtDef>
{
};

template <>
struct EnumEncoder<false, sql::ObDASBaseRtDef*> : sql::DASRtEncoder<sql::ObDASBaseRtDef>
{
};
} // end namespace serialization
} // end namespace common

using namespace common;
namespace sql
{
OB_DEF_SERIALIZE(ObDASRemoteInfo)
{
  int ret = OB_SUCCESS;
  ObSQLSessionInfo *session = exec_ctx_->get_my_session();
  OB_UNIS_ENCODE(flags_);
  if (OB_SUCC(ret) && need_tx_) {
    OB_UNIS_ENCODE(*trans_desc_);
  }
  if (OB_SUCC(ret)) {
    OB_UNIS_ENCODE(snapshot_);
  }
  if (OB_SUCC(ret) && (need_calc_expr_ || need_calc_udf_)) {
    OB_UNIS_ENCODE(session->get_effective_tenant_id());
    OB_UNIS_ENCODE(*session);
  }
  if (OB_SUCC(ret) && has_expr_) {
    OZ(ObPxTreeSerializer::serialize_expr_frame_info(
        buf, buf_len, pos, *exec_ctx_, const_cast<ObExprFrameInfo &>(*frame_info_)));
  }
  OB_UNIS_ENCODE(ctdefs_.count());
  for (int64_t i = 0; OB_SUCC(ret) && i < ctdefs_.count(); ++i) {
    const ObDASBaseCtDef *ctdef = ctdefs_.at(i);
    OB_UNIS_ENCODE(ctdef->op_type_);
    OB_UNIS_ENCODE(*ctdef);
  }
  OB_UNIS_ENCODE(rtdefs_.count());
  for (int64_t i = 0; OB_SUCC(ret) && i < rtdefs_.count(); ++i) {
    ObDASBaseRtDef *rtdef = rtdefs_.at(i);
    OB_UNIS_ENCODE(rtdef->op_type_);
    OB_UNIS_ENCODE(*rtdef);
  }
  return ret;
}

OB_DEF_DESERIALIZE(ObDASRemoteInfo)
{
  int ret = OB_SUCCESS;
  bool need_session = false;
  int ctdef_cnt = 0;
  int rtdef_cnt = 0;
  ObEvalCtx *eval_ctx = nullptr;
  ObDASTaskFactory *das_factory = ObDASSyncAccessP::get_das_factory();
#if !defined(NDEBUG)
  CK(typeid(*exec_ctx_) == typeid(ObDesExecContext));
#endif
  OB_UNIS_DECODE(flags_);
  if (OB_SUCC(ret) && need_tx_) {
    auto txs = MTL(transaction::ObTransService*);
    if (OB_FAIL(txs->acquire_tx(buf, data_len, pos, trans_desc_))) {
      LOG_WARN("acquire tx by deserialized failed", K(data_len), K(pos), K(ret));
    }
  }
  if (OB_SUCC(ret)) {
    OB_UNIS_DECODE(snapshot_);
  }
  if (OB_SUCC(ret) && (need_calc_expr_ || need_calc_udf_)) {
    uint64_t tenant_id = OB_INVALID_TENANT_ID;
    ObDesExecContext *des_exec_ctx = static_cast<ObDesExecContext*>(exec_ctx_);
    OB_UNIS_DECODE(tenant_id);
    OZ(des_exec_ctx->create_my_session(tenant_id));
    OB_UNIS_DECODE(*des_exec_ctx->get_my_session());
    if (OB_SUCC(ret)) {
      //notice: can't unlink exec context and session info here
      typedef ObSQLSessionInfo::ExecCtxSessionRegister MyExecCtxSessionRegister;
      des_exec_ctx->get_my_session()->set_is_remote(true);
      MyExecCtxSessionRegister ctx_register(*des_exec_ctx->get_my_session(), *des_exec_ctx);
    }
  }
  OZ(exec_ctx_->create_physical_plan_ctx());
  if (OB_SUCC(ret) && has_expr_) {
    OZ(ObPxTreeSerializer::deserialize_expr_frame_info(
        buf, data_len, pos, *exec_ctx_, const_cast<ObExprFrameInfo &>(*frame_info_)));
    OZ(exec_ctx_->init_expr_op(frame_info_->rt_exprs_.count()));
    if (OB_SUCC(ret)) {
      eval_ctx = OB_NEWx(ObEvalCtx, (&exec_ctx_->get_allocator()), (*exec_ctx_));
      if (OB_ISNULL(eval_ctx)) {
        ret = OB_ALLOCATE_MEMORY_FAILED;
        LOG_WARN("allocate eval ctx failed", K(ret));
      }
    }
  }
  OB_UNIS_DECODE(ctdef_cnt);
  for (int64_t i = 0; OB_SUCC(ret) && i < ctdef_cnt; ++i) {
    ObDASBaseCtDef *ctdef = nullptr;
    ObDASOpType op_type = DAS_OP_INVALID;
    OB_UNIS_DECODE(op_type);
    OZ(das_factory->create_das_ctdef(op_type, ctdef));
    OB_UNIS_DECODE(*ctdef);
    OZ(ctdefs_.push_back(ctdef));
  }
  OB_UNIS_DECODE(rtdef_cnt);
  for (int64_t i = 0; OB_SUCC(ret) && i < rtdef_cnt; ++i) {
    ObDASBaseRtDef *rtdef = nullptr;
    ObDASOpType op_type = DAS_OP_INVALID;
    OB_UNIS_DECODE(op_type);
    OZ(das_factory->create_das_rtdef(op_type, rtdef));
    OB_UNIS_DECODE(*rtdef);
    OX(rtdef->eval_ctx_ = eval_ctx);
    OZ(rtdefs_.push_back(rtdef));
  }
  return ret;
}

OB_DEF_SERIALIZE_SIZE(ObDASRemoteInfo)
{
  int64_t len = 0;
  ObSQLSessionInfo *session = exec_ctx_->get_my_session();
  OB_UNIS_ADD_LEN(flags_);
  if (need_tx_) {
    OB_UNIS_ADD_LEN(*trans_desc_);
  }
  OB_UNIS_ADD_LEN(snapshot_);
  if (need_calc_expr_ || need_calc_udf_) {
    OB_UNIS_ADD_LEN(session->get_effective_tenant_id());
    OB_UNIS_ADD_LEN(*session);
  }
  if (has_expr_) {
    len += ObPxTreeSerializer::get_serialize_expr_frame_info_size(*exec_ctx_,
             const_cast<ObExprFrameInfo&>(*frame_info_));
  }
  OB_UNIS_ADD_LEN(ctdefs_.count());
  for (int64_t i = 0; i < ctdefs_.count(); ++i) {
    const ObDASBaseCtDef *ctdef = ctdefs_.at(i);
    OB_UNIS_ADD_LEN(ctdef->op_type_);
    OB_UNIS_ADD_LEN(*ctdef);
  }
  OB_UNIS_ADD_LEN(rtdefs_.count());
  for (int64_t i = 0; i < rtdefs_.count(); ++i) {
    ObDASBaseRtDef *rtdef = rtdefs_.at(i);
    OB_UNIS_ADD_LEN(rtdef->op_type_);
    OB_UNIS_ADD_LEN(*rtdef);
  }
  return len;
}

int ObIDASTaskOp::start_das_task()
{
  int &ret = errcode_;
  int simulate_error = EVENT_CALL(EventTable::EN_DAS_SIMULATE_OPEN_ERROR);
  if (OB_UNLIKELY(!is_in_retry() && OB_SUCCESS != simulate_error)) {
    ret = simulate_error;
  } else {
    task_started_ = true;
    if (OB_FAIL(open_op())) {
      LOG_WARN("open das task op failed", K(ret));
      if (OB_ERR_DEFENSIVE_CHECK == ret) {
        //dump das task data to help analysis defensive bug
        dump_data();
      }
    }
  }
  return ret;
}

int ObIDASTaskOp::end_das_task()
{
  int ret = OB_SUCCESS;
  int tmp_ret = OB_SUCCESS;
  //release op，then rollback transcation
  if (task_started_) {
    if (OB_SUCCESS != (tmp_ret = release_op())) {
      LOG_WARN("release das task op failed", K(ret), K_(errcode));
    }
    ret = COVER_SUCC(tmp_ret);
  }
  task_started_ = false;
  ret = COVER_SUCC(tmp_ret);
  errcode_ = OB_SUCCESS;
  return ret;
}

OB_SERIALIZE_MEMBER(ObIDASTaskOp,
                    tenant_id_,
                    task_id_,
                    task_flag_,
                    tablet_id_,
                    ls_id_,
                    related_ctdefs_,
                    related_rtdefs_,
                    related_tablet_ids_);

ObDASTaskArg::ObDASTaskArg()
  : timeout_ts_(0),
    ctrl_svr_(),
    runner_svr_(),
    task_ops_(),
    remote_info_(nullptr)
{
}

OB_DEF_SERIALIZE(ObDASTaskArg)
{
  int ret = OB_SUCCESS;
  LST_DO_CODE(OB_UNIS_ENCODE,
              timeout_ts_,
              ctrl_svr_,
              runner_svr_,
              *remote_info_); //remote info must be serialized before task op
  if (OB_SUCC(ret) && OB_FAIL(serialization::encode_vi64(buf, buf_len, pos, task_ops_.count()))) {
    LOG_WARN("fail to encode ob array count", K(ret));
  }
  for (int64_t i = 0; OB_SUCC(ret) && i < task_ops_.count(); i ++) {
    OB_UNIS_ENCODE(task_ops_.at(i)->get_type());
    OB_UNIS_ENCODE(*task_ops_.at(i));
  }
  return ret;
}

OB_DEF_DESERIALIZE(ObDASTaskArg)
{
  int ret = OB_SUCCESS;
  ObDASOpType op_type = DAS_OP_INVALID;
  int64_t count = 0;
  ObIDASTaskOp *task_op = nullptr;
  ObDASTaskFactory *das_factory = ObDASSyncAccessP::get_das_factory();
  CK(OB_NOT_NULL(das_factory));
  LST_DO_CODE(OB_UNIS_DECODE,
              timeout_ts_,
              ctrl_svr_,
              runner_svr_,
              *remote_info_); //remote info must be deserialized before task op
  task_ops_.reset();
  if (OB_FAIL(ret)) {
  } else if (OB_FAIL(serialization::decode_vi64(buf, data_len, pos, &count))) {
    LOG_WARN("fail to decode ob array count", K(ret));
  } else if (OB_FAIL(task_ops_.prepare_allocate(count))) {
    LOG_WARN("fail to allocate space", K(ret), K(count));
  }
  for (int64_t i = 0; OB_SUCC(ret) && i < count; i ++) {
    OB_UNIS_DECODE(op_type);
    OZ(das_factory->create_das_task_op(op_type, task_op));
    OZ(task_op->init_task_info());
    if (OB_SUCC(ret)) {
      task_ops_.at(i) = task_op;
    }
    OB_UNIS_DECODE(*task_op);
    OZ(task_op->swizzling_remote_task(remote_info_));
  }
  if (OB_SUCC(ret)) {
    (void)ObSQLUtils::adjust_time_by_ntp_offset(timeout_ts_);
  }
  return ret;
}

OB_DEF_SERIALIZE_SIZE(ObDASTaskArg)
{
  int64_t len = 0;
  if (task_ops_.count() != 0) {
    LST_DO_CODE(OB_UNIS_ADD_LEN,
                timeout_ts_,
                ctrl_svr_,
                runner_svr_,
                *remote_info_);
    len += serialization::encoded_length_vi64(task_ops_.count());
    for (int i = 0; i < task_ops_.count(); i++) {
      len += serialization::encoded_length(task_ops_.at(i)->get_type());
      len += serialization::encoded_length(*task_ops_.at(i));
    }
  }
  return len;
}

int ObDASTaskArg::add_task_op(ObIDASTaskOp *task_op)
{
  // we currently only support single task per ObDASTaskArg.
  // TODO(roland.qk):remove this assert check after we enable DAS task aggregation.
  OB_ASSERT(task_ops_.count() == 0);
  return task_ops_.push_back(task_op);
}

ObIDASTaskOp *ObDASTaskArg::get_task_op()
{
  OB_ASSERT(task_ops_.count() == 1);
  return task_ops_.at(0);
}

ObDASTaskResp::ObDASTaskResp()
  : has_more_(false),
    ctrl_svr_(),
    runner_svr_(),
    op_results_()
{
}

void ObDASTaskResp::store_err_msg(const ObString &msg)
{
  int ret = OB_SUCCESS;
  if (!msg.empty()) {
    //这里使用databuff_printf的原因是databuff_printf在遇到buffer溢出时会保证buffer以'\0'结束，确保print的安全性
    if (OB_FAIL(databuff_printf(rcode_.msg_, OB_MAX_ERROR_MSG_LEN, "%.*s", msg.length(), msg.ptr()))) {
      SHARE_LOG(WARN, "store err msg failed", K(ret), K(msg));
      if (OB_SIZE_OVERFLOW == ret) {
        rcode_.msg_[OB_MAX_ERROR_MSG_LEN - 1] = '\0';
      }
    }
  } else {
    rcode_.msg_[0] = '\0';
  }
}

int ObDASTaskResp::store_warning_msg(const ObWarningBuffer &wb)
{
  int ret = OB_SUCCESS;
  bool not_null = true;
  for (uint32_t idx = 0; OB_SUCC(ret) && not_null && idx < wb.get_readable_warning_count(); idx++) {
    const ObWarningBuffer::WarningItem *item = wb.get_warning_item(idx);
    if (item != NULL) {
      if (OB_FAIL(rcode_.warnings_.push_back(*item))) {
        RPC_OBRPC_LOG(WARN, "Failed to add warning", K(ret));
      }
    } else {
      not_null = false;
    }
  }
  return ret;
}

OB_DEF_SERIALIZE(ObDASTaskResp)
{
  int ret = OB_SUCCESS;
  LST_DO_CODE(OB_UNIS_ENCODE,
              has_more_,
              ctrl_svr_,
              runner_svr_);
  if (OB_SUCC(ret) && OB_FAIL(serialization::encode_vi64(
      buf, buf_len, pos, op_results_.count()))) {
    LOG_WARN("fail to encode ob array count", K(ret));
  }
  for (int64_t i = 0; OB_SUCC(ret) && i < op_results_.count(); i ++) {
    if (OB_FAIL(serialization::encode(buf, buf_len, pos, *op_results_.at(i)))) {
      LOG_WARN("fail to encode item", K(i), K(ret));
    }
  }
  LST_DO_CODE(OB_UNIS_ENCODE,
              rcode_,
              trans_result_);
  return ret;
}

OB_DEF_DESERIALIZE(ObDASTaskResp)
{
  int ret = OB_SUCCESS;
  int64_t count = 0;
  LST_DO_CODE(OB_UNIS_DECODE,
              has_more_,
              ctrl_svr_,
              runner_svr_);
  if (OB_SUCC(ret) && OB_FAIL(serialization::decode_vi64(buf, data_len, pos, &count))) {
    LOG_WARN("fail to decode ob array count", K(ret));
  } else if (count != op_results_.count()) {
    ret = OB_ERR_UNEXPECTED;
    LOG_WARN("receive das task response count mismatch", K(count), K(op_results_.count()));
  }
  for (int64_t i = 0; OB_SUCC(ret) && i < count; i ++) {
    if (OB_ISNULL(op_results_.at(i))) {
      ret = OB_ERR_UNEXPECTED;
      LOG_WARN("op result is nullptr", K(ret), K_(op_results));
    } else if (OB_FAIL(serialization::decode(buf, data_len, pos, *op_results_.at(i)))) {
      LOG_WARN("fail to decode array item", K(ret), K(i), K(count));
    }
  }
  LST_DO_CODE(OB_UNIS_DECODE,
              rcode_,
              trans_result_);
  return ret;
}

OB_DEF_SERIALIZE_SIZE(ObDASTaskResp)
{
  int64_t len = 0;
  LST_DO_CODE(OB_UNIS_ADD_LEN,
              has_more_,
              ctrl_svr_,
              runner_svr_);
  len += serialization::encoded_length_vi64(op_results_.count());
  for (int i = 0; i < op_results_.count(); i++) {
    len += serialization::encoded_length(*op_results_.at(i));
  }
  LST_DO_CODE(OB_UNIS_ADD_LEN,
              rcode_,
              trans_result_);
  return len;
}

int ObDASTaskResp::add_op_result(ObIDASTaskResult *op_result)
{
  // we currently only support single task per ObDASTaskResp.
  // TODO(roland.qk):remove this assert check after we enable DAS task aggregation.
  OB_ASSERT(op_results_.count() == 0);
  return op_results_.push_back(op_result);
}

ObIDASTaskResult *ObDASTaskResp::get_op_result()
{
  OB_ASSERT(op_results_.count() == 1);
  return op_results_.at(0);
}

OB_SERIALIZE_MEMBER(ObIDASTaskResult, task_id_);

OB_SERIALIZE_MEMBER(ObDASDataFetchReq, tenant_id_, task_id_);

int ObDASDataFetchReq::init(const uint64_t tenant_id, const int64_t task_id)
{
  tenant_id_ = tenant_id;
  task_id_ = task_id;
  return OB_SUCCESS;
}

OB_SERIALIZE_MEMBER(ObDASDataEraseReq, tenant_id_, task_id_);

int ObDASDataEraseReq::init(const uint64_t tenant_id, const int64_t task_id)
{
  tenant_id_ = tenant_id;
  task_id_ = task_id;
  return OB_SUCCESS;
}

OB_SERIALIZE_MEMBER(ObDASDataFetchRes,
                    datum_store_,
                    tenant_id_, task_id_, has_more_);

ObDASDataFetchRes::ObDASDataFetchRes()
        : datum_store_(),
          tenant_id_(0),
          task_id_(0),
          has_more_(false)
{
}

int ObDASDataFetchRes::init(const uint64_t tenant_id, const int64_t task_id)
{
  int ret = OB_SUCCESS;
  tenant_id_ = tenant_id;
  task_id_ = task_id;
  return ret;
}

int DASOpResultIter::get_next_row()
{
  int ret = OB_SUCCESS;
  if (!task_iter_.is_end()) {
    ObDASScanOp *scan_op = DAS_SCAN_OP(*task_iter_);
    if (OB_ISNULL(scan_op)) {
      ret = OB_ERR_UNEXPECTED;
      LOG_WARN("unexpected das task op type", K(ret), KPC(*task_iter_));
    } else {
      ret = scan_op->get_output_result_iter()->get_next_row();
    }
  } else {
    ret = OB_ITER_END;
  }
  return ret;
}

int DASOpResultIter::get_next_rows(int64_t &count, int64_t capacity)
{
  int ret = OB_SUCCESS;
  if (!task_iter_.is_end()) {
    ObDASScanOp *scan_op = DAS_SCAN_OP(*task_iter_);
    if (OB_ISNULL(scan_op)) {
      ret = OB_ERR_UNEXPECTED;
      LOG_WARN("unexpected das task op type", K(ret), KPC(scan_op));
    } else {
      if (scan_op->is_local_task()) {
        //output row from remote DAS task
        //maybe change the expr datum ptr to its RPC datum store
        //if we need to fetch next row from the next das task,
        //must reset the datum ptr to expr preallocate frame buffer
        //otherwise, get_next_row in the local das task maybe touch a wild datum ptr
        reset_wild_datums_ptr();
      }
      ret = scan_op->get_output_result_iter()->get_next_rows(count, capacity);
      if (!scan_op->is_local_task()) {
        //remote task will change datum ptr, need to mark this flag
        //in order to let the next local task reset datum ptr before get_next_rows
        wild_datum_info_->exprs_ = &scan_op->get_result_outputs();
        wild_datum_info_->max_output_rows_ = max(count, wild_datum_info_->max_output_rows_);
      }
    }
  } else {
    ret = OB_ITER_END;
  }
  return ret;
}

int DASOpResultIter::next_result()
{
  int ret = OB_SUCCESS;
  if (!task_iter_.is_end()) {
    ++task_iter_;
  }
  if (OB_UNLIKELY(task_iter_.is_end())) {
    ret = OB_ITER_END;
    LOG_DEBUG("fetch next das task end", K(ret));
  }
  return ret;
}

int DASOpResultIter::reset_wild_datums_ptr()
{
  int ret = OB_SUCCESS;
  if (wild_datum_info_ != nullptr) {
    if (wild_datum_info_->exprs_ != nullptr &&
        wild_datum_info_->max_output_rows_ > 0) {
      FOREACH_CNT(e, *wild_datum_info_->exprs_) {
        (*e)->locate_datums_for_update(wild_datum_info_->eval_ctx_,
                                       wild_datum_info_->max_output_rows_);
        ObEvalInfo &info = (*e)->get_eval_info(wild_datum_info_->eval_ctx_);
        info.point_to_frame_ = true;
      }
      wild_datum_info_->exprs_ = nullptr;
      wild_datum_info_->max_output_rows_ = 0;
    }
    //global index scan and its lookup maybe share some expr,
    //so remote lookup task change its datum ptr,
    //and also lead index scan to touch the wild datum ptr
    //so need to associate the result iterator of scan and lookup
    //resetting the index scan result datum ptr will also reset the lookup result datum ptr
    if (wild_datum_info_->lookup_iter_ != nullptr) {
      wild_datum_info_->lookup_iter_->reset_wild_datums_ptr();
    }
  }
  return ret;
}
}  // namespace sql
}  // namespace oceanbase