hcq/oceanbase
1
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

[FEAT MERGE] 4.1 PL/SQL enhence & development

Browse Source
This commit is contained in:
obdev
2023-01-09 08:47:31 +00:00
committed by ob-robot
parent 322032b9ca
commit 08802d94f6
98 changed files with 5260 additions and 2676 deletions
Download Patch File Download Diff File Expand all files Collapse all files

560
src/observer/ob_inner_sql_connection.cpp
View File

@ -412,7 +412,6 @@ int ObInnerSQLConnection::init_result(ObInnerSQLResult &res,
pl::ObPLBlockNS *secondary_namespace,
bool is_prepare_protocol,
bool is_prepare_stage,
bool is_from_pl,
bool is_dynamic_sql,
bool is_dbms_sql,
bool is_cursor)
@ -427,8 +426,7 @@ int ObInnerSQLConnection::init_result(ObInnerSQLResult &res,
res.sql_ctx().retry_times_ = retry_cnt;
res.sql_ctx().session_info_ = &get_session();
res.sql_ctx().disable_privilege_check_ = OB_SYS_TENANT_ID == res.sql_ctx().session_info_->get_priv_tenant_id()
? PRIV_CHECK_FLAG_DISABLE
: (is_from_pl ? PRIV_CHECK_FLAG_IN_PL : PRIV_CHECK_FLAG_DISABLE);
? PRIV_CHECK_FLAG_DISABLE : PRIV_CHECK_FLAG_DISABLE;
res.sql_ctx().secondary_namespace_ = secondary_namespace;
res.sql_ctx().is_prepare_protocol_ = is_prepare_protocol;
res.sql_ctx().is_prepare_stage_ = is_prepare_stage;
@ -450,8 +448,7 @@ int ObInnerSQLConnection::process_retry(ObInnerSQLResult &res,
int last_ret,
int64_t abs_timeout_us,
bool &need_retry,
int64_t retry_cnt,
bool is_from_pl)
int64_t retry_cnt)
{
UNUSED(abs_timeout_us);
UNUSED(retry_cnt);
@ -460,7 +457,7 @@ int ObInnerSQLConnection::process_retry(ObInnerSQLResult &res,
bool is_inner_sql = true;
retry_ctrl_.test_and_save_retry_state(GCTX, res.sql_ctx(), res.result_set(),
last_ret, client_ret,
force_local_retry, is_inner_sql, is_from_pl);
force_local_retry, is_inner_sql);
need_retry = (ObQueryRetryType::RETRY_TYPE_LOCAL == retry_ctrl_.get_retry_type());
return client_ret;
}
@ -491,26 +488,27 @@ private:
int64_t execute_end_timestamp_;
};
int ObInnerSQLConnection::process_record(ObInnerSQLResult &res,
int ObInnerSQLConnection::process_record(sql::ObResultSet &result_set,
sql::ObSqlCtx &sql_ctx,
sql::ObSQLSessionInfo &session,
ObITimeRecord &time_record,
int last_ret,
int64_t execution_id,
int64_t ps_stmt_id,
int64_t routine_id,
ObWaitEventDesc &max_wait_desc,
ObWaitEventStat &total_wait_desc,
ObExecRecord &exec_record,
ObExecTimestamp &exec_timestamp,
bool has_tenant_resource,
const ObString &ps_sql,
bool is_from_pl)
{
int ret = OB_SUCCESS;
UNUSED(routine_id);
if (res.has_tenant_resource()) {
sql::ObResultSet &result_set = res.result_set();
if (has_tenant_resource) {
ObAuditRecordData &audit_record = session.get_raw_audit_record();
audit_record.try_cnt_++;
ObPhysicalPlan *plan = res.result_set().get_physical_plan();
ObPhysicalPlan *plan = result_set.get_physical_plan();
audit_record.seq_ = 0; //don't use now
audit_record.status_ = (0 == last_ret || OB_ITER_END == last_ret)
? obmysql::REQUEST_SUCC : last_ret;
@ -520,11 +518,15 @@ int ObInnerSQLConnection::process_record(ObInnerSQLResult &res,
audit_record.user_group_ = THIS_WORKER.get_group_id();
audit_record.execution_id_ = execution_id;
audit_record.ps_stmt_id_ = ps_stmt_id;
MEMCPY(audit_record.sql_id_, res.sql_ctx().sql_id_, (int32_t)sizeof(audit_record.sql_id_));
audit_record.affected_rows_ = res.result_set().get_affected_rows();
audit_record.return_rows_ = res.result_set().get_return_rows();
if (NULL != res.result_set().get_exec_context().get_task_executor_ctx()) {
audit_record.partition_cnt_ = res.result_set().get_exec_context()
if (ps_sql.length() != 0) {
audit_record.sql_ = const_cast<char *>(ps_sql.ptr());
audit_record.sql_len_ = min(ps_sql.length(), OB_MAX_SQL_LENGTH);
}
MEMCPY(audit_record.sql_id_, sql_ctx.sql_id_, (int32_t)sizeof(audit_record.sql_id_));
audit_record.affected_rows_ = result_set.get_affected_rows();
audit_record.return_rows_ = result_set.get_return_rows();
if (NULL != result_set.get_exec_context().get_task_executor_ctx()) {
audit_record.partition_cnt_ = result_set.get_exec_context()
.get_das_ctx()
.get_related_tablet_cnt();
}
@ -534,16 +536,16 @@ int ObInnerSQLConnection::process_record(ObInnerSQLResult &res,
exec_record.wait_time_end_ = total_wait_desc.time_waited_;
exec_record.wait_count_end_ = total_wait_desc.total_waits_;
if (NULL != res.result_set().get_physical_plan()) {
audit_record.plan_type_ = res.result_set().get_physical_plan()->get_plan_type();
audit_record.table_scan_ = res.result_set().get_physical_plan()->contain_table_scan();
audit_record.plan_id_ = res.result_set().get_physical_plan()->get_plan_id();
audit_record.plan_hash_ = res.result_set().get_physical_plan()->get_plan_hash_value();
if (NULL != result_set.get_physical_plan()) {
audit_record.plan_type_ = result_set.get_physical_plan()->get_plan_type();
audit_record.table_scan_ = result_set.get_physical_plan()->contain_table_scan();
audit_record.plan_id_ = result_set.get_physical_plan()->get_plan_id();
audit_record.plan_hash_ = result_set.get_physical_plan()->get_plan_hash_value();
}
audit_record.is_executor_rpc_ = false;
audit_record.is_inner_sql_ = true;
audit_record.is_hit_plan_cache_ = res.result_set().get_is_from_plan_cache();
audit_record.is_inner_sql_ = !is_from_pl;
audit_record.is_hit_plan_cache_ = result_set.get_is_from_plan_cache();
audit_record.is_multi_stmt_ = false; //是否是multi sql
bool first_record = (1 == audit_record.try_cnt_);
@ -552,7 +554,7 @@ int ObInnerSQLConnection::process_record(ObInnerSQLResult &res,
audit_record.exec_record_ = exec_record;
ObIArray<ObTableRowCount> *table_row_count_list = NULL;
ObPhysicalPlanCtx *plan_ctx = GET_PHY_PLAN_CTX(res.result_set().get_exec_context());
ObPhysicalPlanCtx *plan_ctx = GET_PHY_PLAN_CTX(result_set.get_exec_context());
if (NULL != plan_ctx) {
audit_record.consistency_level_ = plan_ctx->get_consistency_level();
audit_record.table_scan_stat_ = plan_ctx->get_table_scan_stat();
@ -565,23 +567,25 @@ int ObInnerSQLConnection::process_record(ObInnerSQLResult &res,
//update v$sql statistics
if (OB_SUCC(last_ret) && session.get_local_ob_enable_plan_cache()) {
if (NULL != plan) {
if (!(res.sql_ctx().self_add_plan_) && res.sql_ctx().plan_cache_hit_) {
if (!(sql_ctx.self_add_plan_) && sql_ctx.plan_cache_hit_) {
plan->update_plan_stat(audit_record,
false, // false mean not first update plan stat
res.result_set().get_exec_context().get_is_evolution(),
table_row_count_list);
} else if (res.sql_ctx().self_add_plan_ && !res.sql_ctx().plan_cache_hit_) {
false, // false mean not first update plan stat
result_set.get_exec_context().get_is_evolution(),
table_row_count_list);
} else if (sql_ctx.self_add_plan_ && !sql_ctx.plan_cache_hit_) {
plan->update_plan_stat(audit_record,
true,
res.result_set().get_exec_context().get_is_evolution(),
table_row_count_list);
true,
result_set.get_exec_context().get_is_evolution(),
table_row_count_list);
}
}
}
record_stat(session, result_set.get_stmt_type(), is_from_pl);
ObSQLUtils::handle_audit_record(false, sql::PSCursor == exec_timestamp.exec_type_ ? EXECUTE_PS_EXECUTE
: is_from_pl ? EXECUTE_PL_EXECUTE : EXECUTE_INNER, session);
ObSQLUtils::handle_audit_record(false, sql::PSCursor == exec_timestamp.exec_type_
? EXECUTE_PS_EXECUTE :
(is_from_pl ? EXECUTE_PL_EXECUTE : EXECUTE_INNER),
session);
}
return ret;
}
@ -645,8 +649,7 @@ int ObInnerSQLConnection::do_query(sqlclient::ObIExecutor &executor, ObInnerSQLR
int ObInnerSQLConnection::query(sqlclient::ObIExecutor &executor,
ObInnerSQLResult &res,
ObVirtualTableIteratorFactory *vt_iter_factory,
bool is_from_pl)
ObVirtualTableIteratorFactory *vt_iter_factory)
{
int ret = OB_SUCCESS;
lib::CompatModeGuard g(get_compat_mode());
@ -656,9 +659,7 @@ int ObInnerSQLConnection::query(sqlclient::ObIExecutor &executor,
exec_timestamp.exec_type_ = sql::InnerSql;
const ObGlobalContext &gctx = ObServer::get_instance().get_gctx();
int64_t start_time = ObTimeUtility::current_time();
if (!is_from_pl) {
get_session().set_query_start_time(start_time); //FIXME 暂时写成这样
}
get_session().set_query_start_time(start_time); //FIXME 暂时写成这样
get_session().set_trans_type(transaction::ObTxClass::SYS);
int64_t abs_timeout_us = 0;
int64_t execution_id = 0;
@ -680,7 +681,7 @@ int ObInnerSQLConnection::query(sqlclient::ObIExecutor &executor,
ret = OB_REF_NUM_NOT_ZERO;
LOG_ERROR("connection still be referred by previous sql result, can not execute sql now",
K(ret), K(executor));
} else if (OB_FAIL(set_timeout(abs_timeout_us, is_from_pl))) {
} else if (OB_FAIL(set_timeout(abs_timeout_us))) {
LOG_WARN("set timeout failed", K(ret));
} else if (OB_ISNULL(ob_sql_)) {
ret = OB_ERR_UNEXPECTED;
@ -706,7 +707,7 @@ int ObInnerSQLConnection::query(sqlclient::ObIExecutor &executor,
for (int64_t retry_cnt = 0; need_retry; ++retry_cnt) {
need_retry = false;
retry_info.clear_state_before_each_retry();
res.set_is_read((is_from_pl && lib::is_mysql_mode()) ? false : true);
res.set_is_read(true);
if (retry_cnt > 0) { // reset result set
bool is_user_sql = res.result_set().is_user_sql();
res.~ObInnerSQLResult();
@ -715,7 +716,7 @@ int ObInnerSQLConnection::query(sqlclient::ObIExecutor &executor,
LOG_WARN("fail to init result set", K(ret));
} else {
res.result_set().set_user_sql(is_user_sql);
res.set_is_read((is_from_pl && lib::is_mysql_mode()) ? false : true);
res.set_is_read(true);
}
}
get_session().get_raw_audit_record().request_memory_used_ = 0;
@ -728,9 +729,7 @@ int ObInnerSQLConnection::query(sqlclient::ObIExecutor &executor,
ObWaitEventStat total_wait_desc;
const bool enable_perf_event = lib::is_diagnose_info_enabled();
const bool enable_sql_audit =
GCONF.enable_sql_audit
&& get_session().get_local_ob_enable_sql_audit()
&& !is_from_pl; // PL的audit在SPI中记录
GCONF.enable_sql_audit && get_session().get_local_ob_enable_sql_audit();
{
ObMaxWaitGuard max_wait_guard(enable_perf_event ? &max_wait_desc : NULL);
ObTotalWaitGuard total_wait_guard(enable_perf_event ? &total_wait_desc : NULL);
@ -747,7 +746,7 @@ int ObInnerSQLConnection::query(sqlclient::ObIExecutor &executor,
} else if (OB_FAIL(gctx.schema_service_->get_tenant_schema_guard(tenant_id, res.schema_guard_))) {
LOG_WARN("get schema guard failed", K(ret));
} else if (OB_FAIL(init_result(res, vt_iter_factory, retry_cnt,
res.schema_guard_, NULL, false, false, is_from_pl))) {
res.schema_guard_, NULL, false, false))) {
LOG_WARN("failed to init result", K(ret));
} else if (OB_FAIL(res.schema_guard_.get_schema_version(tenant_id, local_tenant_schema_version))) {
LOG_WARN("get tenant schema version failed", K(ret), K(ob_sql_));
@ -764,7 +763,7 @@ int ObInnerSQLConnection::query(sqlclient::ObIExecutor &executor,
} else if (OB_FAIL(SMART_CALL(do_query(executor, res)))) {
ret_code = ret;
LOG_WARN("execute failed", K(ret), K(tenant_id), K(executor), K(retry_cnt));
ret = process_retry(res, ret, abs_timeout_us, need_retry, retry_cnt, is_from_pl);
ret = process_retry(res, ret, abs_timeout_us, need_retry, retry_cnt);
// moved here from ObInnerSQLConnection::do_query() -> ObInnerSQLResult::open().
int close_ret = res.force_close();
if (OB_SUCCESS != close_ret) {
@ -787,11 +786,13 @@ int ObInnerSQLConnection::query(sqlclient::ObIExecutor &executor,
if (enable_sql_audit && res.is_inited()) {
ObInnerSQLTimeRecord time_record(get_session());
ObString dummy_ps_sql;
time_record.set_execute_start_timestamp(execute_start_timestamp_);
time_record.set_execute_end_timestamp(execute_end_timestamp_);
int record_ret = process_record(res, get_session(), time_record, ret,
execution_id, OB_INVALID_ID, OB_INVALID_ID,
max_wait_desc, total_wait_desc, exec_record, exec_timestamp);
int record_ret = process_record(res.result_set(), res.sql_ctx(), get_session(), time_record, ret,
execution_id, OB_INVALID_ID,
max_wait_desc, total_wait_desc, exec_record, exec_timestamp,
res.has_tenant_resource(), dummy_ps_sql);
if (OB_SUCCESS != record_ret) {
LOG_WARN("failed to process record", K(executor), K(record_ret), K(ret));
}
@ -823,333 +824,6 @@ int ObInnerSQLConnection::query(sqlclient::ObIExecutor &executor,
return ret;
}
int ObInnerSQLConnection::do_prepare(const common::ObString &sql, ObInnerSQLResult &res)
{
int ret = OB_SUCCESS;
WITH_CONTEXT(res.mem_context_) {
if (!inited_) {
ret = OB_NOT_INIT;
LOG_WARN("not init", K(ret));
} else if (sql.empty()) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("invalid argument", K(ret), K(sql));
} else if (OB_FAIL(ob_sql_->stmt_prepare(sql, res.sql_ctx(), res.result_set()))) {
LOG_WARN("sql execute failed", K(ret), K(sql));
}
}
return ret;
}
int ObInnerSQLConnection::prepare(const ObString &sql,
pl::ObPLBlockNS *secondary_namespace,
bool is_dynamic_sql,
bool is_dbms_sql,
bool is_cursor,
ObInnerSQLResult &res,
ObVirtualTableIteratorFactory *vt_iter_factory)
{
int ret = OB_SUCCESS;
FLTSpanGuard(inner_prepare);
ObExecRecord exec_record;
ObExecTimestamp exec_timestamp;
int64_t execution_id = 0;
exec_timestamp.exec_type_ = sql::InnerSql;
const ObGlobalContext &gctx = ObServer::get_instance().get_gctx();
int64_t old_query_start_time = get_session().get_query_start_time();
get_session().set_query_start_time(ObTimeUtility::current_time()); //FIXME 暂时写成这样
get_session().set_trans_type(transaction::ObTxClass::SYS);
// get_session().store_query_string(sql);
int64_t abs_timeout_us = 0;
const uint64_t* trace_id_val = ObCurTraceId::get();
bool is_trace_id_init = true;
ObQueryRetryInfo &retry_info = get_session().get_retry_info_for_update();
if (0 == trace_id_val[0]) {
is_trace_id_init = false;
common::ObCurTraceId::init(observer::ObServer::get_instance().get_self());
}
// backup && restore worker/session timeout.
TimeoutGuard timeout_guard(*this);
// %vt_iter_factory may be NULL
if (!inited_) {
ret = OB_NOT_INIT;
LOG_WARN("not init", K(ret));
} else if (sql.empty()) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("invalid argument", K(ret), K(sql));
} else if (NULL != ref_ctx_) {
ret = OB_REF_NUM_NOT_ZERO;
LOG_ERROR("connection still be referred by previous sql result, can not execute sql now",
K(ret), K(sql));
} else if (OB_FAIL(set_timeout(abs_timeout_us, true))) {
LOG_WARN("set timeout failed", K(ret));
} else if (OB_ISNULL(ob_sql_)) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("invalid sql engine", K(ret), K(ob_sql_));
} else if (OB_UNLIKELY(retry_info.is_inited())) {
if (is_inner_session()) {
ret = OB_ERR_UNEXPECTED;
LOG_ERROR("retry info is inited", K(ret), K(retry_info), K(sql));
}
} else if (OB_FAIL(retry_info.init())) {
LOG_WARN("fail to init retry info", K(ret), K(retry_info), K(sql));
}
if (OB_SUCC(ret)) {
execution_id = ob_sql_->get_execution_id();
retry_ctrl_.clear_state_before_each_retry(get_session().get_retry_info_for_update());
retry_ctrl_.reset_retry_times();
bool need_retry = true;
for (int64_t retry_cnt = 0; need_retry; ++retry_cnt) {
need_retry = false;
retry_info.clear_state_before_each_retry();
if (retry_cnt > 0) { // reset result set
res.~ObInnerSQLResult();
new (&res) ObInnerSQLResult(get_session());
ret = res.init();
}
const uint64_t tenant_id = get_session().get_effective_tenant_id();
if (OB_FAIL(gctx.schema_service_->get_tenant_schema_guard(tenant_id, res.schema_guard_))) {
LOG_WARN("get schema guard failed", K(ret));
} else if (OB_FAIL(init_result(res, vt_iter_factory, retry_cnt,
res.schema_guard_, secondary_namespace, true, true, true,
is_dynamic_sql, is_dbms_sql, is_cursor))) {
LOG_WARN("failed to init result", K(ret));
} else if (OB_FAIL(do_prepare(sql, res))) {
LOG_WARN("execute sql failed", K(ret), K(sql), K(retry_cnt));
ret = process_retry(res, ret, abs_timeout_us, need_retry, retry_cnt, true);
}
}
}
if (res.is_inited()) {
int aret = process_final(sql, res, ret);
if (OB_SUCCESS != aret) {
LOG_WARN("failed to process final", K(sql), K(aret), K(ret));
}
}
if (false == is_trace_id_init) {
common::ObCurTraceId::reset();
}
if (is_inner_session()) {
retry_info.reset();
}
get_session().set_query_start_time(old_query_start_time);
return ret;
}
int ObInnerSQLConnection::do_execute(const ParamStore &params, ObInnerSQLResult &res)
{
int ret = OB_SUCCESS;
WITH_CONTEXT(res.mem_context_) {
if (!inited_) {
ret = OB_NOT_INIT;
LOG_WARN("not init", K(ret));
} else if (OB_ISNULL(ob_sql_)) {
ret = OB_ERR_UNEXPECTED;
LOG_ERROR("ob_sql_ is NULL", K(ret));
} else if (OB_FAIL(ob_sql_->stmt_execute(res.result_set().get_statement_id(),
res.result_set().get_stmt_type(),
params,
res.sql_ctx(),
res.result_set(),
true /* is_inner_sql */))) {
LOG_WARN("sql execute failed", K(res.result_set().get_statement_id()), K(ret));
} else {
ObSQLSessionInfo &session = res.result_set().get_session();
if (OB_ISNULL(res.sql_ctx().schema_guard_)) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("schema guard is null");
} else if (OB_FAIL(session.update_query_sensitive_system_variable(*(res.sql_ctx().schema_guard_)))) {
LOG_WARN("update query affacted system variable failed", K(ret));
} else if (OB_UNLIKELY(NULL != sql_modifier_)
&& OB_FAIL(sql_modifier_->modify(res.result_set()))) {
LOG_WARN("fail modify sql", K(res.result_set().get_statement_name()), K(ret));
} else if (OB_FAIL(res.open())) {
LOG_WARN("result set open failed", K(res.result_set().get_statement_id()), K(ret));
} else { /*do nothing*/ }
}
}
return ret;
}
int ObInnerSQLConnection::execute(ParamStore &params,
ObInnerSQLResult &res,
ObVirtualTableIteratorFactory *vt_iter_factory,
bool is_from_pl,
bool is_dynamic,
bool is_forall,
int64_t array_binding_count)
{
int ret = OB_SUCCESS;
ObExecRecord exec_record;
ObExecTimestamp exec_timestamp;
exec_timestamp.exec_type_ = sql::InnerSql;
const ObGlobalContext &gctx = ObServer::get_instance().get_gctx();
const ObString &sql = res.result_set().get_statement_name();
int64_t start_time = ObTimeUtility::current_time();
if (!is_from_pl) {
get_session().set_query_start_time(start_time); //FIXME 暂时写成这样
}
get_session().set_trans_type(transaction::ObTxClass::SYS);
int64_t abs_timeout_us = 0;
int64_t execution_id = 0;
uint64_t stmt_id = res.result_set().get_statement_id();
sql::stmt::StmtType stmt_type = res.result_set().get_stmt_type();
const uint64_t* trace_id_val = ObCurTraceId::get();
bool is_trace_id_init = true;
ObQueryRetryInfo &retry_info = get_session().get_retry_info_for_update();
if (0 == trace_id_val[0]) {
is_trace_id_init = false;
common::ObCurTraceId::init(observer::ObServer::get_instance().get_self());
}
// backup && restore worker/session timeout.
TimeoutGuard timeout_guard(*this);
// %vt_iter_factory may be NULL
if (!inited_) {
ret = OB_NOT_INIT;
LOG_WARN("not init", K(ret));
} else if (NULL != ref_ctx_) {
ret = OB_REF_NUM_NOT_ZERO;
LOG_ERROR("connection still be referred by previous sql result, can not execute sql now",
K(ret), K(sql));
} else if (OB_FAIL(set_timeout(abs_timeout_us, is_from_pl))) {
LOG_WARN("set timeout failed", K(ret));
} else if (OB_ISNULL(ob_sql_)) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("invalid sql engine", K(ret), K(ob_sql_));
} else if (OB_UNLIKELY(retry_info.is_inited())) {
if (is_inner_session()) {
ret = OB_ERR_UNEXPECTED;
LOG_ERROR("retry info is inited", K(ret), K(retry_info), K(sql));
}
} else if (OB_FAIL(retry_info.init())) {
LOG_WARN("fail to init retry info", K(ret), K(retry_info), K(sql));
}
if (OB_SUCC(ret)) {
execution_id = ob_sql_->get_execution_id();
bool need_retry = true;
retry_ctrl_.clear_state_before_each_retry(get_session().get_retry_info_for_update());
retry_ctrl_.reset_retry_times();
for (int64_t retry_cnt = 0; need_retry; ++retry_cnt) {
need_retry = false;
retry_info.clear_state_before_each_retry();
res.set_is_read((is_from_pl && lib::is_mysql_mode()) ? false : true);
if (retry_cnt > 0) { // reset result set
res.~ObInnerSQLResult();
new (&res) ObInnerSQLResult(get_session());
ret = res.init();
if (OB_SUCC(ret)) {
res.result_set().set_ps_protocol();
res.result_set().set_statement_id(stmt_id);
res.result_set().set_stmt_type(stmt_type);
res.set_is_read((is_from_pl && lib::is_mysql_mode()) ? false : true);
}
}
get_session().get_raw_audit_record().request_memory_used_ = 0;
observer::ObProcessMallocCallback pmcb(0,
get_session().get_raw_audit_record().request_memory_used_);
ObMallocCallbackGuard guard(pmcb);
ObWaitEventDesc max_wait_desc;
ObWaitEventStat total_wait_desc;
const bool enable_perf_event = lib::is_diagnose_info_enabled();
const bool enable_sql_audit =
GCONF.enable_sql_audit
&& get_session().get_local_ob_enable_sql_audit()
&& !is_from_pl; // PL中的SQL语句Audit在SPI中记录
{
ObMaxWaitGuard max_wait_guard(enable_perf_event ? &max_wait_desc : NULL);
ObTotalWaitGuard total_wait_guard(enable_perf_event ? &total_wait_desc : NULL);
//监控项统计开始
if (enable_sql_audit) {
exec_record.record_start();
}
if (OB_SUCC(ret) && is_forall) {
res.sql_ctx().multi_stmt_item_.set_ps_mode(true);
res.sql_ctx().multi_stmt_item_.set_ab_cnt(array_binding_count);
}
const uint64_t tenant_id = get_session().get_effective_tenant_id();
int ret_code = OB_SUCCESS;
if (OB_FAIL(ret)) {
} else if (OB_FAIL(gctx.schema_service_->get_tenant_schema_guard(tenant_id, res.schema_guard_))) {
LOG_WARN("get schema guard failed", K(ret));
} else if (OB_FAIL(init_result(res, vt_iter_factory, retry_cnt, res.schema_guard_, NULL,
false, false, is_from_pl, is_dynamic))) {
LOG_WARN("failed to init result", K(ret));
} else if (OB_FAIL(do_execute(params, res))) {
LOG_WARN("execute sql failed", K(ret), K(sql), K(retry_cnt));
ret_code = ret;
ret = process_retry(res, ret, abs_timeout_us, need_retry, retry_cnt, is_from_pl);
// moved here from ObInnerSQLConnection::do_execute() -> ObInnerSQLResult::open().
int close_ret = res.force_close();
if (OB_SUCCESS != close_ret) {
LOG_WARN("failed to close result", K(close_ret), K(ret), K(sql));
}
}
if (OB_NOT_NULL(res.get_result_set())) {
ObSQLSessionInfo &session = res.result_set().get_session();
session.set_session_in_retry(need_retry, ret_code);
LOG_DEBUG("after process_retry", K(retry_cnt), K(ret), K(need_retry),
K(inner_session_), KP(&inner_session_),
K(inner_session_.get_is_in_retry()),
K(session.get_is_in_retry_for_dup_tbl()),
K(session), K(&session), K(session.get_is_in_retry()));
}
execute_start_timestamp_ = (res.get_execute_start_ts() > 0)
? res.get_execute_start_ts()
: ObTimeUtility::current_time();
execute_end_timestamp_ = (res.get_execute_end_ts() > 0)
? res.get_execute_end_ts()
: ObTimeUtility::current_time();
//监控项统计结束
if (enable_sql_audit) {
exec_record.record_end();
}
}
if (enable_sql_audit && res.is_inited()) {
ObInnerSQLTimeRecord time_record(get_session());
time_record.set_execute_start_timestamp(execute_start_timestamp_);
time_record.set_execute_end_timestamp(execute_end_timestamp_);
int record_ret = process_record(res, get_session(), time_record, ret,
execution_id, stmt_id, OB_INVALID_ID,
max_wait_desc, total_wait_desc, exec_record, exec_timestamp);
if (OB_SUCCESS != record_ret) {
LOG_WARN("failed to process record", K(sql), K(record_ret), K(ret));
}
}
if (get_session().get_in_transaction()) {
if (ObStmt::is_dml_write_stmt(stmt_type)) {
get_session().set_has_inner_dml_write(true);
}
}
}
}
if (res.is_inited()) {
int aret = process_final(sql, res, ret);
if (OB_SUCCESS != aret) {
LOG_WARN("failed to process final", K(sql), K(aret), K(ret));
}
}
if (false == is_trace_id_init) {
common::ObCurTraceId::reset();
}
if (is_inner_session()) {
retry_info.reset();
}
return ret;
}
common::sqlclient::ObCommonServerConnectionPool *ObInnerSQLConnection::get_common_server_pool()
{
return NULL;
@ -1170,7 +844,7 @@ int ObInnerSQLConnection::retry_while_no_tenant_resource(const int64_t cluster_i
get_session().set_query_start_time(start_time);
TimeoutGuard timeout_guard(*this); // backup && restore worker/session timeout.
if (OB_FAIL(set_timeout(abs_timeout_us, false))) {
if (OB_FAIL(set_timeout(abs_timeout_us))) {
LOG_WARN("set timeout failed", K(ret));
} else {
do {
@ -1948,7 +1622,7 @@ int ObInnerSQLConnection::get_session_timeout_for_rpc(int64_t &query_timeout, in
int64_t abs_timeout_us = 0;
int64_t start_time = ObTimeUtility::current_time();
get_session().set_query_start_time(start_time);
if (OB_FAIL(set_timeout(abs_timeout_us, false))) {
if (OB_FAIL(set_timeout(abs_timeout_us))) {
LOG_WARN("set timeout failed", K(ret));
} else if (OB_FAIL(get_session().get_query_timeout(query_timeout))
|| OB_FAIL(get_session().get_tx_timeout(trx_timeout))) {
@ -1961,10 +1635,9 @@ int ObInnerSQLConnection::execute_read(const uint64_t tenant_id,
const char *sql,
ObISQLClient::ReadResult &res,
bool is_user_sql,
bool is_from_pl,
const common::ObAddr *sql_exec_addr)
{
return execute_read(GCONF.cluster_id, tenant_id, sql, res, is_user_sql, is_from_pl, sql_exec_addr);
return execute_read(GCONF.cluster_id, tenant_id, sql, res, is_user_sql, sql_exec_addr);
}
int ObInnerSQLConnection::execute_read(const int64_t cluster_id,
@ -1972,12 +1645,11 @@ int ObInnerSQLConnection::execute_read(const int64_t cluster_id,
const ObString &sql,
ObISQLClient::ReadResult &res,
bool is_user_sql,
bool is_from_pl,
const common::ObAddr *sql_exec_addr)
{
int ret = OB_SUCCESS;
auto function = [&]() { return execute_read_inner(cluster_id, tenant_id, sql, res, is_user_sql, is_from_pl, sql_exec_addr); };
auto function = [&]() { return execute_read_inner(cluster_id, tenant_id, sql, res, is_user_sql, sql_exec_addr); };
if (OB_FAIL(retry_while_no_tenant_resource(cluster_id, tenant_id, function))) {
LOG_WARN("execute_read failed", K(ret), K(cluster_id), K(tenant_id));
}
@ -2009,7 +1681,6 @@ int ObInnerSQLConnection::execute_read_inner(const int64_t cluster_id,
const ObString &sql,
ObISQLClient::ReadResult &res,
bool is_user_sql,
bool is_from_pl,
const common::ObAddr *sql_exec_addr)
{
int ret = OB_SUCCESS;
@ -2039,7 +1710,7 @@ int ObInnerSQLConnection::execute_read_inner(const int64_t cluster_id,
} else if (local_execute) {
read_ctx->get_result().result_set().set_user_sql(is_user_sql);
if (OB_FAIL(query(executor, read_ctx->get_result(),
&read_ctx->get_vt_iter_factory(), is_from_pl))) {
&read_ctx->get_vt_iter_factory()))) {
LOG_WARN("execute sql failed", K(ret), K(tenant_id), K(sql));
}
} else if (is_resource_conn()) {
@ -2117,8 +1788,7 @@ int ObInnerSQLConnection::execute_read_inner(const int64_t cluster_id,
handler->get_result()->get_stmt_type()))) {
} else if (OB_FAIL(read_ctx->get_result().open())) {
LOG_WARN("result set open failed", K(ret));
} else if (FALSE_IT(read_ctx->get_result().set_is_read(
(is_from_pl && lib::is_mysql_mode()) ? false : true))) {
} else if (FALSE_IT(read_ctx->get_result().set_is_read(true))) {
} else if (FALSE_IT(get_session().set_trans_type(transaction::ObTxClass::SYS))) {
}
}
@ -2148,7 +1818,7 @@ int ObInnerSQLConnection::nonblock_get_leader(
if (OB_ISNULL(GCTX.location_service_)) {
ret = OB_NOT_INIT;
LOG_WARN("location cache is NULL", K(ret));
} else if (OB_FAIL(set_timeout(abs_timeout_us, false))) {
} else if (OB_FAIL(set_timeout(abs_timeout_us))) {
LOG_WARN("set timeout failed", K(ret));
} else {
const int64_t retry_interval_us = 200 * 1000;
@ -2209,97 +1879,6 @@ int ObInnerSQLConnection::execute(
return ret;
}
int ObInnerSQLConnection::prepare(const uint64_t tenant_id,
const ObString &sql,
pl::ObPLBlockNS *secondary_namespace,
bool is_dynamic_sql,
bool is_dbms_sql,
bool is_cursor,
ObISQLClient::ReadResult &res)
{
int ret = OB_SUCCESS;
ObInnerSQLReadContext *read_ctx = NULL;
if (!inited_) {
ret = OB_NOT_INIT;
LOG_WARN("connection not inited", K(ret));
} else if (NULL == sql || '\0' == *sql || OB_INVALID_ID == tenant_id) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("invalid argument", K(ret), K(sql), K(tenant_id));
} else if (OB_FAIL(switch_tenant(tenant_id))) {
LOG_WARN("switch tenant_id failed", K(ret), K(tenant_id));
} else if (OB_FAIL(res.create_handler(read_ctx, *this))) {
LOG_WARN("create result handler failed", K(ret));
} else if (OB_FAIL(read_ctx->get_result().init())) {
LOG_WARN("init result set", K(ret));
} else if (OB_FAIL(prepare(sql,
secondary_namespace,
is_dynamic_sql,
is_dbms_sql,
is_cursor,
read_ctx->get_result(),
&read_ctx->get_vt_iter_factory()))) {
LOG_WARN("execute sql failed", K(ret), K(tenant_id), K(sql));
}
if (OB_SUCC(ret)) {
ref_ctx_ = read_ctx;
}
return ret;
}
int ObInnerSQLConnection::execute(const uint64_t tenant_id,
const ObPsStmtId stmt_id,
const stmt::StmtType stmt_type,
ParamStore &params,
ObISQLClient::ReadResult &res,
bool is_from_pl,
bool is_dynamic,
bool is_forall,
int64_t array_binding_count)
{
int ret = OB_SUCCESS;
FLTSpanGuard(inner_execute);
ObInnerSQLReadContext *read_ctx = NULL;
ObPsStmtInfoGuard ps_guard;
ObPsStmtInfo *ps_info = NULL;
ObPsCache *ps_cache = get_session().get_ps_cache();
if (!inited_) {
ret = OB_NOT_INIT;
LOG_WARN("connection not inited", K(ret));
} else if (OB_INVALID_ID == tenant_id) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("invalid argument", K(ret), K(stmt_id), K(tenant_id));
} else if (OB_FAIL(switch_tenant(tenant_id))) {
LOG_WARN("switch tenant_id failed", K(ret), K(tenant_id));
} else if (OB_FAIL(res.create_handler(read_ctx, *this))) {
LOG_WARN("create result handler failed", K(ret));
} else if (OB_FAIL(read_ctx->get_result().init())) {
LOG_WARN("init result set", K(ret));
} else if (OB_ISNULL(ps_cache)) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("ps cache is null", K(ret), K(ps_cache));
} else if (OB_FAIL(ps_cache->get_stmt_info_guard(stmt_id, ps_guard))) {
LOG_WARN("get stmt info guard failed", K(ret), K(stmt_id));
} else if (OB_ISNULL(ps_info = ps_guard.get_stmt_info())) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("get stmt info is null", K(ret), K(ps_info), K(stmt_id));
} else {
read_ctx->get_result().result_set().set_ps_protocol();
read_ctx->get_result().result_set().set_statement_id(stmt_id);
read_ctx->get_result().result_set().set_stmt_type(stmt_type);
get_session().store_query_string(ps_info->get_ps_sql());
if (OB_FAIL(execute(params, read_ctx->get_result(), &read_ctx->get_vt_iter_factory(),
is_from_pl, is_dynamic, is_forall, array_binding_count))) {
LOG_WARN("execute sql failed", K(ret), K(tenant_id), K(stmt_id));
}
}
if (OB_SUCC(ret)) {
ref_ctx_ = read_ctx;
}
return ret;
}
int ObInnerSQLConnection::switch_tenant(const uint64_t tenant_id)
{
int ret = OB_SUCCESS;
@ -2319,7 +1898,7 @@ int ObInnerSQLConnection::switch_tenant(const uint64_t tenant_id)
return ret;
}
int ObInnerSQLConnection::set_timeout(int64_t &abs_timeout_us, bool is_from_pl)
int ObInnerSQLConnection::set_timeout(int64_t &abs_timeout_us)
{
int ret = OB_SUCCESS;
const ObTimeoutCtx &ctx = ObTimeoutCtx::get_ctx();
@ -2377,22 +1956,11 @@ int ObInnerSQLConnection::set_timeout(int64_t &abs_timeout_us, bool is_from_pl)
// no need to set session timeout for outer session if no timeout ctx
if (OB_SUCC(ret)
&& (is_inner_session() || ctx.is_timeout_set() || ctx.is_trx_timeout_set() || is_from_pl)) {
if (!is_from_pl) {
if (OB_FAIL(set_session_timeout(timeout, trx_timeout))) {
LOG_WARN("set session timeout failed", K(timeout), K(trx_timeout), K(ret));
} else {
THIS_WORKER.set_timeout_ts(get_session().get_query_start_time() + timeout);
}
&& (is_inner_session() || ctx.is_timeout_set() || ctx.is_trx_timeout_set())) {
if (OB_FAIL(set_session_timeout(timeout, trx_timeout))) {
LOG_WARN("set session timeout failed", K(timeout), K(trx_timeout), K(ret));
} else {
int64_t query_timeout;
OZ (get_session().get_query_timeout(query_timeout));
OX (abs_timeout_us = get_session().get_query_start_time() > 0
? get_session().get_query_start_time() + query_timeout
: ObTimeUtility::current_time() + query_timeout);
if (OB_SUCC(ret) && THIS_WORKER.get_timeout_ts() > abs_timeout_us) {
OX (THIS_WORKER.set_timeout_ts(abs_timeout_us));
}
THIS_WORKER.set_timeout_ts(get_session().get_query_start_time() + timeout);
}
}
return ret;