/**

* 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_EXE

#include "sql/executor/ob_remote_scheduler.h"

#include "sql/executor/ob_remote_job_control.h"

#include "sql/executor/ob_task_spliter_factory.h"

#include "sql/executor/ob_remote_job_executor.h"

#include "sql/executor/ob_remote_task_executor.h"

#include "sql/executor/ob_local_job_executor.h"

#include "sql/executor/ob_local_task_executor.h"

#include "sql/executor/ob_job.h"

#include "share/partition_table/ob_partition_location.h"

#include "sql/executor/ob_job_parser.h"

#include "sql/executor/ob_task_executor_ctx.h"

#include "sql/engine/ob_physical_plan_ctx.h"

#include "share/ob_define.h"

#include "lib/utility/utility.h"

#include "sql/engine/ob_exec_context.h"

#include "rpc/obrpc/ob_rpc_net_handler.h"

namespace oceanbase

{

using namespace oceanbase::common;

namespace sql

{

ObRemoteScheduler::ObRemoteScheduler()

{

}

ObRemoteScheduler::~ObRemoteScheduler()

{

}

int ObRemoteScheduler::schedule(ObExecContext &ctx, ObPhysicalPlan *phy_plan)

{

int ret = OB_SUCCESS;

if (ctx.use_remote_sql()) {

if (OB_ISNULL(ctx.get_sql_ctx())) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("session is null", K(ret), K(ctx.get_sql_ctx()));

} else if (OB_FAIL(execute_with_sql(ctx, phy_plan))) {

LOG_WARN("execute with sql failed", K(ret));

}

} else {

if (OB_FAIL(execute_with_plan(ctx, phy_plan))) {

LOG_WARN("execute with plan failed", K(ret));

}

}

LOG_TRACE("remote_scheduler", K(ctx.use_remote_sql()), KPC(phy_plan));

return ret;

}

int ObRemoteScheduler::execute_with_plan(ObExecContext &ctx, ObPhysicalPlan *phy_plan)

{

// 1. Split and construct task using ObJobConf info

// 2. Call job.schedule()

int ret = OB_SUCCESS;

ObJobParser parser;

ObLocalTaskExecutor local_task_executor;

ObLocalJobExecutor local_job_executor;

ObRemoteTaskExecutor remote_task_executor;

ObRemoteJobExecutor remote_job_executor;

ObSEArray<ObJob *, 2> jobs;

ObJob *root_job = NULL;

ObJob *remote_job = NULL;

ObRemoteJobControl jc;

ObTaskSpliterFactory task_factory;

ObPhysicalPlanCtx *plan_ctx = NULL;

ObTaskExecutorCtx &task_exec_ctx = ctx.get_task_exec_ctx();

if (OB_ISNULL(phy_plan)) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("not init", K(phy_plan), K(ret));

} else if (OB_FAIL(task_exec_ctx.reset_and_init_stream_handler())) {

LOG_WARN("reset and init stream handler failed", K(ret));

} else if (OB_UNLIKELY(NULL == (plan_ctx = ctx.get_physical_plan_ctx()))) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("plan ctx is NULL", K(ret));

} else {

ObExecutionID ob_execution_id;

ob_execution_id.set_server(task_exec_ctx.get_self_addr());

ob_execution_id.set_execution_id(OB_INVALID_ID);

if (OB_FAIL(parser.parse_job(ctx,

phy_plan,

ob_execution_id,

task_factory,

jc))) {

LOG_WARN("fail parse job for scheduler.", K(ret));

}

}

if (OB_SUCC(ret)) {

if (OB_FAIL(jc.get_ready_jobs(jobs))) {

LOG_WARN("fail get jobs.", K(ret));

} else if (OB_UNLIKELY(2 != jobs.count())) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("unexpected job count. expect 2, actual", "job_count", jobs.count());

} else if (OB_UNLIKELY(NULL == (root_job = jobs.at(0)))) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("unexpected null job", K(ret));

} else if (OB_UNLIKELY(NULL == (remote_job = jobs.at(1)))) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("unexpected null job", K(ret));

} else {

local_job_executor.set_task_executor(local_task_executor);

local_job_executor.set_job(*root_job);

remote_job_executor.set_task_executor(remote_task_executor);

remote_job_executor.set_job(*remote_job);

// 说明：本函数阻塞地将RemoteJob发送到远端，发送成功后本函数返回

// 最终控制权会进入到LocalJob中的ObDirectReceive中,

// 它通过get_stream_handler()来获得当前handler

// 然后阻塞等待在handler上收取返回结果

if (OB_FAIL(remote_job_executor.execute(ctx))) {

LOG_WARN("fail execute remote job", K(ret));

} else if (OB_FAIL(local_job_executor.execute(ctx))) {

LOG_WARN("fail execute local job", K(ret));

}

}

}

if (OB_FAIL(ret)) {

// 出错了，打印job的状态，包括它的位置，rpc是否已经发出去等等

int print_ret = OB_SUCCESS;

const static int64_t MAX_JC_STATUS_BUF_LEN = 4096;

char jc_status_buf[MAX_JC_STATUS_BUF_LEN];

if (OB_SUCCESS != (print_ret = jc.print_status(jc_status_buf, MAX_JC_STATUS_BUF_LEN))) {

LOG_WARN("fail to print job control status",

K(ret), K(print_ret), LITERAL_K(MAX_JC_STATUS_BUF_LEN));

} else {

LOG_WARN("fail to schedule, print job's status",

K(ret), K(print_ret), "job_status", jc_status_buf);

}

}

return ret;

}

int ObRemoteScheduler::build_remote_task(ObExecContext &ctx,

ObRemoteTask &remote_task,

const DependenyTableStore &dependency_tables)

{

int ret = OB_SUCCESS;

ObPhysicalPlanCtx *plan_ctx = ctx.get_physical_plan_ctx();

ObTaskExecutorCtx &task_exec_ctx = ctx.get_task_exec_ctx();

ObSQLSessionInfo *session = nullptr;

if (OB_FAIL(remote_task.assign_dependency_tables(dependency_tables))) {

LOG_WARN("fail to assign dependency_tables", K(ret));

}

remote_task.set_ctrl_server(ctx.get_addr());

remote_task.set_session(ctx.get_my_session());

remote_task.set_query_schema_version(task_exec_ctx.get_query_tenant_begin_schema_version(),

task_exec_ctx.get_query_sys_begin_schema_version());

LOG_TRACE("print schema_version", K(task_exec_ctx.get_query_tenant_begin_schema_version()),

K(task_exec_ctx.get_query_sys_begin_schema_version()));

remote_task.set_remote_sql_info(&plan_ctx->get_remote_sql_info());

ObDASTabletLoc *first_tablet_loc = DAS_CTX(ctx).get_table_loc_list().get_first()->get_first_tablet_loc();

if (OB_ISNULL(session = ctx.get_my_session())) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("session is null", K(ret));

} else {

remote_task.set_runner_svr(first_tablet_loc->server_);

ObTaskID task_id;

task_id.set_execution_id(session->get_current_execution_id());

task_id.set_server(ctx.get_addr());

task_id.set_task_id(0);

remote_task.set_task_id(task_id);

remote_task.set_snapshot(ctx.get_das_ctx().get_snapshot());

}

return ret;

}

int ObRemoteScheduler::execute_with_sql(ObExecContext &ctx, ObPhysicalPlan *phy_plan)

{

int ret = OB_SUCCESS;

RemoteExecuteStreamHandle *handler = NULL;

ObSQLSessionInfo *session = ctx.get_my_session();

ObPhysicalPlanCtx *plan_ctx = ctx.get_physical_plan_ctx();

ObTaskExecutorCtx &task_exec_ctx = ctx.get_task_exec_ctx();

ObExecuteResult &exec_result = task_exec_ctx.get_execute_result();

ObExecutorRpcImpl *rpc = NULL;

ObQueryRetryInfo *retry_info = NULL;

ObRemoteTask task;

bool has_sent_task = false;

bool has_transfer_err = false;

if (OB_ISNULL(phy_plan) || OB_ISNULL(session) || OB_ISNULL(plan_ctx)) {

ret = OB_INVALID_ARGUMENT;

LOG_WARN("invalid argument", K(ret), K(phy_plan), K(session), K(plan_ctx));

} else if (OB_FAIL(task_exec_ctx.reset_and_init_stream_handler())) {

LOG_WARN("reset and init stream handler failed", K(ret));

} else if (OB_FAIL(ObTaskExecutorCtxUtil::get_stream_handler(ctx, handler))) {

LOG_WARN("fail get task response handler", K(ret));

} else if (OB_FAIL(ObTaskExecutorCtxUtil::get_task_executor_rpc(ctx, rpc))) {

LOG_WARN("fail get executor rpc", K(ret));

} else if (OB_ISNULL(session)

|| OB_ISNULL(plan_ctx)

|| OB_ISNULL(handler)

|| OB_ISNULL(rpc)

|| OB_ISNULL(retry_info = &session->get_retry_info_for_update())) {

ret = OB_ERR_UNEXPECTED;

LOG_ERROR("unexpected null ptr",

K(ret), K(session), K(plan_ctx), K(handler), K(rpc), K(retry_info));

} else if (FALSE_IT(handler->set_use_remote_protocol_v2())) {

//使用新的remote sync execute协议

} else if (OB_FAIL(handler->reset_and_init_result())) {

LOG_WARN("fail to reset and init result", K(ret));

} else if (OB_FAIL(build_remote_task(ctx, task, phy_plan->get_dependency_table()))) {

LOG_WARN("build remote task failed", K(ret), K(task));

} else if (OB_FAIL(session->add_changed_package_info(ctx))) {

LOG_WARN("failed to add changed package info to session", K(ret));

} else {

session->reset_all_package_changed_info();

LOG_DEBUG("execute remote task", K(task));

ObOperator *op = NULL;

if (OB_FAIL(phy_plan->get_root_op_spec()->create_operator(ctx, op))) {

LOG_WARN("create operator from spec failed", K(ret));

} else if (OB_ISNULL(op)) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("created operator is NULL", K(ret));

} else {

exec_result.set_static_engine_root(op);

}

}

if (OB_SUCC(ret)) {

ObScanner *scanner = NULL;

const int32_t group_id = OB_INVALID_ID == session->get_expect_group_id() ? 0 : session->get_expect_group_id();

ObExecutorRpcCtx rpc_ctx(session->get_rpc_tenant_id(),

plan_ctx->get_timeout_timestamp(),

ctx.get_task_exec_ctx().get_min_cluster_version(),

retry_info,

ctx.get_my_session(),

plan_ctx->is_plain_select_stmt(),

group_id);

if (OB_FAIL(rpc->task_execute_v2(rpc_ctx,

task,

task.get_runner_svr(),

*handler,

has_sent_task,

has_transfer_err))) {

LOG_WARN("task execute failed", K(ret));

}

// handle tx relative info if plan involved in transaction

int tmp_ret = ObRemoteTaskExecutor::handle_tx_after_rpc(handler->get_result(),

session,

has_sent_task,

has_transfer_err,

phy_plan,

ctx);

NG_TRACE_EXT(remote_task_completed, OB_ID(ret), ret,

OB_ID(runner_svr), task.get_runner_svr(), OB_ID(task), task);

// 说明：本函数返回后，最终控制权会进入到ObDirectReceive中,

// 它通过get_stream_handler()来获得当前handler

// 然后阻塞等待在handler上收取返回结果

}

return ret;

}

} /* ns sql */

} /* ns oceanbase */