270 lines
11 KiB
C++
270 lines
11 KiB
C++
/**
|
|
* 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 */
|