/**

* 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

#include "sql/ob_result_set.h"

#include "lib/oblog/ob_trace_log.h"

#include "lib/charset/ob_charset.h"

#include "lib/utility/ob_macro_utils.h"

#include "rpc/obmysql/ob_mysql_global.h"

#include "rpc/obmysql/ob_mysql_field.h"

#include "lib/oblog/ob_log_module.h"

#include "engine/ob_physical_plan.h"

#include "sql/parser/parse_malloc.h"

#include "share/system_variable/ob_system_variable.h"

#include "share/system_variable/ob_system_variable_alias.h"

#include "sql/session/ob_sql_session_info.h"

#include "sql/resolver/ob_cmd.h"

#include "sql/engine/px/ob_px_admission.h"

#include "sql/engine/cmd/ob_table_direct_insert_service.h"

#include "sql/executor/ob_executor.h"

#include "sql/executor/ob_cmd_executor.h"

#include "sql/resolver/dml/ob_select_stmt.h"

#include "sql/resolver/cmd/ob_call_procedure_stmt.h"

#include "sql/optimizer/ob_optimizer_util.h"

#include "sql/optimizer/ob_log_plan_factory.h"

#include "sql/ob_sql_trans_util.h"

#include "sql/ob_end_trans_callback.h"

#include "sql/session/ob_sql_session_info.h"

#include "lib/profile/ob_perf_event.h"

#include "sql/plan_cache/ob_cache_object_factory.h"

#include "share/ob_cluster_version.h"

#include "storage/tx/ob_trans_define.h"

#include "pl/ob_pl_user_type.h"

#include "pl/ob_pl_stmt.h"

#include "observer/ob_server_struct.h"

#include "storage/tx/wrs/ob_weak_read_service.h" // ObWeakReadService

#include "storage/tx/wrs/ob_i_weak_read_service.h" // WRS_LEVEL_SERVER

#include "storage/tx/wrs/ob_weak_read_util.h" // ObWeakReadUtil

#include "observer/ob_req_time_service.h"

#include "sql/dblink/ob_dblink_utils.h"

#include "sql/dblink/ob_tm_service.h"

#include <cctype>

using namespace oceanbase::sql;

using namespace oceanbase::common;

using namespace oceanbase::share;

using namespace oceanbase::share::schema;

using namespace oceanbase::storage;

using namespace oceanbase::transaction;

ObResultSet::~ObResultSet()

{

bool is_remote_sql = false;

if (OB_NOT_NULL(get_exec_context().get_sql_ctx())) {

is_remote_sql = get_exec_context().get_sql_ctx()->is_remote_sql_;

}

ObPhysicalPlan* physical_plan = get_physical_plan();

if (OB_NOT_NULL(physical_plan) && !is_remote_sql

&& OB_UNLIKELY(physical_plan->is_limited_concurrent_num())) {

physical_plan->dec_concurrent_num();

}

// when ObExecContext is destroyed, it also depends on the physical plan, so need to ensure

// that inner_exec_ctx_ is destroyed before cache_obj_guard_

if (NULL != inner_exec_ctx_) {

inner_exec_ctx_->~ObExecContext();

inner_exec_ctx_ = NULL;

}

ObPlanCache *pc = my_session_.get_plan_cache_directly();

if (OB_NOT_NULL(pc)) {

cache_obj_guard_.force_early_release(pc);

}

// Always called at the end of the ObResultSet destructor

update_end_time();

is_init_ = false;

}

int ObResultSet::open_cmd()

{

int ret = OB_SUCCESS;

FLTSpanGuard(cmd_open);

if (OB_ISNULL(cmd_)) {

LOG_ERROR("cmd and physical_plan both not init", K(stmt_type_));

ret = common::OB_NOT_INIT;

} else if (OB_FAIL(init_cmd_exec_context(get_exec_context()))) {

LOG_WARN("fail init exec context", K(ret), K_(stmt_type));

} else if (OB_FAIL(on_cmd_execute())) {

LOG_WARN("fail start cmd trans", K(ret), K_(stmt_type));

} else if (OB_FAIL(ObCmdExecutor::execute(get_exec_context(), *cmd_))) {

SQL_LOG(WARN, "execute cmd failed", K(ret));

}

return ret;

}

OB_INLINE int ObResultSet::open_plan()

{

int ret = OB_SUCCESS;

//ObLimit *limit_opt = NULL;

ObPhysicalPlan* physical_plan_ = static_cast<ObPhysicalPlan*>(cache_obj_guard_.get_cache_obj());

if (OB_ISNULL(physical_plan_)) {

ret = OB_ERR_UNEXPECTED;

LOG_ERROR("invalid physical plan", K(physical_plan_));

} else {

has_top_limit_ = physical_plan_->has_top_limit();

if (OB_SUCC(ret)) {

if (OB_FAIL(ObPxAdmission::enter_query_admission(my_session_,

get_exec_context(),

get_stmt_type(),

*get_physical_plan()))) {

// query is not admitted to run

// Note: explain statement's phy plan is target query's plan, don't enable admission test

LOG_DEBUG("Query is not admitted to run, try again", K(ret));

} else if (THIS_WORKER.is_timeout()) {

// packet有可能在队列里面呆的时间过长，到这里已经超时，

// 如果这里不检查，则会继续运行，很可能进入到其他模块里面，

// 其他模块检测到超时，引起其他模块的疑惑，因此在这里加上超时检查。

// 之所以在这个位置才检查，是为了考虑hint中的超时时间，

// 因为在init_plan_exec_context函数里面才将hint的超时时间设进THIS_WORKER中。

ret = OB_TIMEOUT;

LOG_WARN("query is timeout", K(ret),

"timeout_ts", THIS_WORKER.get_timeout_ts(),

"start_time", my_session_.get_query_start_time());

} else if (stmt::T_PREPARE != stmt_type_) {

int64_t retry = 0;

if (OB_SUCC(ret)) {

do {

ret = do_open_plan(get_exec_context());

} while (transaction_set_violation_and_retry(ret, retry));

}

}

}

}

return ret;

}

int ObResultSet::open()

{

int ret = OB_SUCCESS;

my_session_.set_process_query_time(ObTimeUtility::current_time());

LinkExecCtxGuard link_guard(my_session_, get_exec_context());

FLTSpanGuard(open);

if (lib::is_oracle_mode() &&

get_exec_context().get_nested_level() >= OB_MAX_RECURSIVE_SQL_LEVELS) {

ret = OB_ERR_RECURSIVE_SQL_LEVELS_EXCEEDED;

LOG_ORACLE_USER_ERROR(OB_ERR_RECURSIVE_SQL_LEVELS_EXCEEDED, OB_MAX_RECURSIVE_SQL_LEVELS);

} else if (OB_FAIL(execute())) {

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

} else if (OB_FAIL(open_result())) {

LOG_WARN("open result set failed", K(ret));

}

if (OB_SUCC(ret)) {

ObPhysicalPlan* physical_plan_ = static_cast<ObPhysicalPlan*>(cache_obj_guard_.get_cache_obj());

if (OB_NOT_NULL(cmd_)) {

// cmd not set

} else if (ret != OB_NOT_INIT &&

OB_ISNULL(physical_plan_)) {

LOG_WARN("empty physical plan", K(ret));

} else if (OB_FAIL(ret)) {

physical_plan_->set_is_last_exec_succ(false);

} else {

physical_plan_->set_is_last_exec_succ(true);

}

}

return ret;

}

int ObResultSet::execute()

{

int ret = common::OB_SUCCESS;

ObPhysicalPlan* physical_plan_ = static_cast<ObPhysicalPlan*>(cache_obj_guard_.get_cache_obj());

// if (stmt::T_PREPARE != stmt_type_

// && stmt::T_DEALLOCATE != stmt_type_) {

if (NULL != physical_plan_) {

ret = open_plan();

} else if (NULL != cmd_) {

ret = open_cmd();

} else {

// inner sql executor, no plan or cmd. do nothing.

}

// } else {

// //T_PREPARE//T_DEALLOCATE do nothing

// }

set_errcode(ret);

return ret;

}

int ObResultSet::open_result()

{

int ret = OB_SUCCESS;

ObPhysicalPlan* physical_plan_ = static_cast<ObPhysicalPlan*>(cache_obj_guard_.get_cache_obj());

if (NULL != physical_plan_) {

if (OB_ISNULL(exec_result_)) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("exec result is null", K(ret));

} else if (OB_FAIL(exec_result_->open(get_exec_context()))) {

if (OB_TRANSACTION_SET_VIOLATION != ret && OB_TRY_LOCK_ROW_CONFLICT != ret) {

SQL_LOG(WARN, "fail open main query", K(ret));

}

} else if (OB_FAIL(drive_dml_query())) {

LOG_WARN("fail to drive dml query", K(ret));

} else if ((stmt::T_INSERT == get_stmt_type())

&& (ObTableDirectInsertService::is_direct_insert(*physical_plan_))) {

// for insert /*+ append */ into select clause

if (OB_FAIL(ObTableDirectInsertService::commit_direct_insert(get_exec_context(), *physical_plan_))) {

LOG_WARN("fail to commit direct insert", KR(ret));

}

}

}

if (OB_SUCC(ret)) {

if (!is_inner_result_set_ && OB_FAIL(set_mysql_info())) {

SQL_LOG(WARN, "fail to get mysql info", K(ret));

} else if (NULL != get_exec_context().get_physical_plan_ctx()) {

SQL_LOG(DEBUG, "get affected row", K(get_stmt_type()),

K(get_exec_context().get_physical_plan_ctx()->get_affected_rows()));

set_affected_rows(get_exec_context().get_physical_plan_ctx()->get_affected_rows());

}

if (OB_SUCC(ret) && get_stmt_type() == stmt::T_ANONYMOUS_BLOCK) {

// Compatible with oracle anonymous block affect rows setting

set_affected_rows(1);

}

}

set_errcode(ret);

return ret;

}

/* on_cmd_execute - prepare before start execute cmd

*

* some command implicit commit current transaction state,

* include release savepoints

*/

int ObResultSet::on_cmd_execute()

{

int ret = OB_SUCCESS;

bool ac = true;

if (OB_ISNULL(cmd_)) {

ret = OB_ERR_UNEXPECTED;

LOG_ERROR("invalid inner state", K(cmd_));

} else if (cmd_->cause_implicit_commit()) {

if (my_session_.is_in_transaction() && my_session_.associated_xa()) {

int tmp_ret = OB_SUCCESS;

transaction::ObTxDesc *tx_desc = my_session_.get_tx_desc();

const transaction::ObXATransID xid = my_session_.get_xid();

const transaction::ObGlobalTxType global_tx_type = tx_desc->get_global_tx_type(xid);

if (transaction::ObGlobalTxType::XA_TRANS == global_tx_type) {

// commit is not allowed in xa trans

ret = OB_TRANS_XA_ERR_COMMIT;

LOG_WARN("COMMIT is not allowed in a xa trans", K(ret), K(xid), K(global_tx_type),

KPC(tx_desc));

} else if (transaction::ObGlobalTxType::DBLINK_TRANS == global_tx_type) {

transaction::ObTransID tx_id;

if (OB_FAIL(ObTMService::tm_commit(get_exec_context(), tx_id))) {

LOG_WARN("fail to do commit for dblink trans", K(ret), K(tx_id), K(xid),

K(global_tx_type));

}

my_session_.restore_auto_commit();

const bool force_disconnect = false;

if (OB_UNLIKELY(OB_SUCCESS != (tmp_ret = my_session_.get_dblink_context().clean_dblink_conn(force_disconnect)))) {

LOG_WARN("dblink transaction failed to release dblink connections", K(tmp_ret), K(tx_id), K(xid));

}

} else {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("unexpected global trans type", K(ret), K(xid), K(global_tx_type), KPC(tx_desc));

}

get_exec_context().set_need_disconnect(false);

} else {

// implicit end transaction and start transaction will not clear next scope transaction settings by:

// a. set by `set transaction read only`

// b. set by `set transaction isolation level XXX`

const int cmd_type = cmd_->get_cmd_type();

bool keep_trans_variable = (cmd_type == stmt::T_START_TRANS);

if (OB_FAIL(ObSqlTransControl::implicit_end_trans(get_exec_context(), false, NULL, !keep_trans_variable))) {

LOG_WARN("fail end implicit trans on cmd execute", K(ret));

} else if (my_session_.need_recheck_txn_readonly() && my_session_.get_tx_read_only()) {

ret = OB_ERR_CANT_EXECUTE_IN_READ_ONLY_TRANSACTION;

LOG_WARN("cmd can not execute because txn is read only", K(ret), K(cmd_type));

}

}

}

return ret;

}

// open transaction if need (eg. ac=1 DML)

int ObResultSet::start_stmt()

{

NG_TRACE(sql_start_stmt_begin);

int ret = OB_SUCCESS;

bool ac = true;

ObPhysicalPlan* phy_plan = static_cast<ObPhysicalPlan*>(cache_obj_guard_.get_cache_obj());

if (OB_ISNULL(phy_plan)) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("invalid inner state", K(phy_plan));

} else if (OB_ISNULL(phy_plan->get_root_op_spec())) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("root_op_spec of phy_plan is NULL", K(phy_plan), K(ret));

} else if (OB_FAIL(my_session_.get_autocommit(ac))) {

LOG_WARN("fail to get autocommit", K(ret));

} else if (phy_plan->is_link_dml_plan() || phy_plan->has_link_sfd()) {

if (ac) {

my_session_.set_autocommit(false);

my_session_.set_restore_auto_commit(); // auto commit will be restored at tm_rollback or tm_commit;

}

LOG_DEBUG("dblink xa trascaction need skip start_stmt()", K(PHY_LINK_DML == phy_plan->get_root_op_spec()->type_), K(my_session_.get_dblink_context().is_dblink_xa_tras()));

} else {

if (-1 != phy_plan->tx_id_) {

const transaction::ObTransID tx_id(phy_plan->tx_id_);

if (OB_FAIL(sql::ObTMService::recover_tx_for_callback(tx_id, get_exec_context()))) {

LOG_WARN("failed to recover dblink xa transaction", K(ret), K(tx_id));

} else {

need_revert_tx_ = true;

LOG_DEBUG("succ to recover dblink xa transaction", K(tx_id));

}

} else {

LOG_DEBUG("recover dblink xa skip", K(phy_plan->tx_id_));

}

bool in_trans = my_session_.get_in_transaction();

// 1. 无论是否处于事务中，只要不是select并且plan为REMOTE的，就反馈给客户端不命中

// 2. feedback this misshit to obproxy (bug#6255177)

// 3. 对于multi-stmt，只反馈首个partition hit信息给客户端

// 4. 需要考虑到重试的情况，需要反馈给客户端的是首个重试成功的partition hit。

if (OB_SUCC(ret) && stmt::T_SELECT != stmt_type_) {

my_session_.partition_hit().try_set_bool(

OB_PHY_PLAN_REMOTE != phy_plan->get_plan_type());

}

if (OB_FAIL(ret)) {

// do nothing

} else if (ObSqlTransUtil::is_remote_trans(

ac, in_trans, phy_plan->get_plan_type())) {

// pass

} else if (OB_LIKELY(phy_plan->is_need_trans())) {

if (get_trans_state().is_start_stmt_executed()) {

ret = OB_ERR_UNEXPECTED;

LOG_ERROR("invalid transaction state", K(get_trans_state().is_start_stmt_executed()));

} else if (OB_FAIL(ObSqlTransControl::start_stmt(get_exec_context()))) {

SQL_LOG(WARN, "fail to start stmt", K(ret),

K(phy_plan->get_dependency_table()));

} else {

auto literal_stmt_type = literal_stmt_type_ != stmt::T_NONE ? literal_stmt_type_ : stmt_type_;

my_session_.set_first_need_txn_stmt_type(literal_stmt_type);

}

get_trans_state().set_start_stmt_executed(OB_SUCC(ret));

}

}

NG_TRACE(sql_start_stmt_end);

return ret;

}

int ObResultSet::end_stmt(const bool is_rollback)

{

int ret = OB_SUCCESS;

NG_TRACE(start_end_stmt);

if (get_trans_state().is_start_stmt_executed()

&& get_trans_state().is_start_stmt_success()) {

ObPhysicalPlan* physical_plan_ = static_cast<ObPhysicalPlan*>(cache_obj_guard_.get_cache_obj());

if (OB_ISNULL(physical_plan_)) {

ret = OB_ERR_UNEXPECTED;

LOG_ERROR("invalid inner state", K(physical_plan_));

} else if (physical_plan_->is_need_trans()) {

if (OB_FAIL(ObSqlTransControl::end_stmt(get_exec_context(), is_rollback))) {

SQL_LOG(WARN, "fail to end stmt", K(ret), K(is_rollback));

}

}

get_trans_state().clear_start_stmt_executed();

} else {

// do nothing

}

if (need_revert_tx_) { // ignore ret

int tmp_ret = sql::ObTMService::revert_tx_for_callback(get_exec_context());

need_revert_tx_ = false;

LOG_DEBUG("revert tx for callback", K(tmp_ret));

}

NG_TRACE(end_stmt);

return ret;

}

//@notice: this interface can not be used in the interface of ObResultSet

//otherwise, will cause the exec context reference mistake in session info

//see the call reference in LinkExecCtxGuard

int ObResultSet::get_next_row(const common::ObNewRow *&row)

{

LinkExecCtxGuard link_guard(my_session_, get_exec_context());

return inner_get_next_row(row);

}

OB_INLINE int ObResultSet::inner_get_next_row(const common::ObNewRow *&row)

{

int &ret = errcode_;

ObPhysicalPlan* physical_plan_ = static_cast<ObPhysicalPlan*>(cache_obj_guard_.get_cache_obj());

// last_exec_succ default values is true

if (OB_LIKELY(NULL != physical_plan_)) { // take this branch more frequently

if (OB_ISNULL(exec_result_)) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("exec result is null", K(ret));

} else if (OB_FAIL(exec_result_->get_next_row(get_exec_context(), row))) {

if (OB_ITER_END != ret) {

LOG_WARN("get next row from exec result failed", K(ret));

// marked last execute status

physical_plan_->set_is_last_exec_succ(false);

}

} else {

return_rows_++;

}

} else if (NULL != cmd_) {

if (is_pl_stmt(static_cast<stmt::StmtType>(cmd_->get_cmd_type()))) {

if (return_rows_ > 0) {

errcode_ = OB_ITER_END;

} else {

row = get_exec_context().get_output_row();

return_rows_++;

}

} else {

ret = OB_ERR_UNEXPECTED;

_OB_LOG(ERROR, "should not call get_next_row in CMD SQL");

}

} else {

_OB_LOG(ERROR, "phy_plan not init");

ret = OB_NOT_INIT;

}

//Save the current execution state to determine whether to refresh location

//and perform other necessary cleanup operations when the statement exits.

DAS_CTX(get_exec_context()).get_location_router().save_cur_exec_status(ret);

return ret;

}

// 触发本错误的条件： A、B两个SQL，同时修改了某几行数据（修改内容有交集）。

// 微观上，修改操作要先读出符合条件的行，然后再更新。在读的时候，会记录一个版本号，

// 更新的时候，会检查版本号是否有变化。如果有变化，则说明在读之后、写之前，数据被其它

// 更新操作修改了。这个时候如果强行更新，则会违反一致性。举个例子：

// tbl里包含一行数据 0，希望通过A、B两个并发的操作让它变成2：

// A: update tbl set a = a + 1;

// B: update tbl set a = a + 1;

// 如果在A读a之后、更新a之前，B完成了整个更新操作，则B的更新会丢失，最终结果为a=1

//

// 解决方案：在更新数据之前检查版本号，发现版本号不一致，则抛出OB_TRANSACTION_SET_VIOLATION

// 错误，由SQL层在这里重试，保证读写版本号一致。

//

// Note: 由于本重试不涉及到刷新Schema或更新Location Cache，所以无需做整个语句级重试，

// 执行级别的重试即可。

bool ObResultSet::transaction_set_violation_and_retry(int &err, int64_t &retry_times)

{

bool bret = false;

ObSqlCtx *sql_ctx = get_exec_context().get_sql_ctx();

bool is_batched_stmt = false;

if (sql_ctx != nullptr) {

is_batched_stmt = sql_ctx->is_batch_params_execute();

}

if ((OB_SNAPSHOT_DISCARDED == err

|| OB_TRANSACTION_SET_VIOLATION == err)

&& retry_times < TRANSACTION_SET_VIOLATION_MAX_RETRY

&& !is_batched_stmt) {

//batched stmt本身是一种优化，其中cursor状态在快速重路径中无法维护，不走快速重试路径

ObTxIsolationLevel isolation = my_session_.get_tx_isolation();

bool is_isolation_RR_or_SE = (isolation == ObTxIsolationLevel::RR

|| isolation == ObTxIsolationLevel::SERIAL);

// bug#6361189 pass err to force rollback stmt in do_close_plan()

if (OB_TRANSACTION_SET_VIOLATION == err && 0 == retry_times && !is_isolation_RR_or_SE) {

// TSC错误重试时，只在第一次打印WARN日志

LOG_WARN_RET(err, "transaction set consistency violation, will retry");

}

int ret = do_close_plan(err, get_exec_context());

ObPhysicalPlanCtx *plan_ctx = get_exec_context().get_physical_plan_ctx();

if (OB_NOT_NULL(plan_ctx)) {

plan_ctx->reset_for_quick_retry();

}

if (OB_SUCCESS != ret) {

// 注意：如果do_close失败，会影响到TSC冲突重试，导致本该重试却实际没有重试的问题

// 这里如果出现bug，则需要看每个涉及到的Operator的close实现

LOG_WARN("failed to close plan", K(err), K(ret));

} else {

// OB_SNAPSHOT_DISCARDED should not retry now, see:

//

// so we remove this condition: OB_TRANSACTION_SET_VIOLATION == err

if (/*OB_TRANSACTION_SET_VIOLATION == err &&*/ is_isolation_RR_or_SE) {

// rewrite err in ObQueryRetryCtrl::test_and_save_retry_state().

// err = OB_TRANS_CANNOT_SERIALIZE;

bret = false;

} else {

++retry_times;

bret = true;

}

}

LOG_DEBUG("transaction set consistency violation and retry",

"retry", bret, K(retry_times), K(err));

}

return bret;

}

OB_INLINE int ObResultSet::do_open_plan(ObExecContext &ctx)

{

ObPhysicalPlan* physical_plan_ = static_cast<ObPhysicalPlan*>(cache_obj_guard_.get_cache_obj());

NG_TRACE_EXT(do_open_plan_begin, OB_ID(plan_id), physical_plan_->get_plan_id());

int ret = OB_SUCCESS;

ctx.reset_op_env();

exec_result_ = &(ctx.get_task_exec_ctx().get_execute_result());

if (stmt::T_PREPARE != stmt_type_) {

if (OB_FAIL(ctx.init_phy_op(physical_plan_->get_phy_operator_size()))) {

LOG_WARN("fail init exec phy op ctx", K(ret));

} else if (OB_FAIL(ctx.init_expr_op(physical_plan_->get_expr_operator_size()))) {

LOG_WARN("fail init exec expr op ctx", K(ret));

}

}

if (OB_SUCC(ret) && my_session_.get_ddl_info().is_ddl() && stmt::T_INSERT == get_stmt_type()) {

if (OB_FAIL(ObDDLUtil::clear_ddl_checksum(physical_plan_))) {

LOG_WARN("fail to clear ddl checksum", K(ret));

}

}

// for insert /*+ append */ into select clause

if (OB_SUCC(ret)

&& (stmt::T_INSERT == get_stmt_type())

&& (ObTableDirectInsertService::is_direct_insert(*physical_plan_))) {

if (OB_FAIL(ObTableDirectInsertService::start_direct_insert(ctx, *physical_plan_))) {

LOG_WARN("fail to start direct insert", KR(ret));

}

}

if (OB_FAIL(ret)) {

} else if (OB_FAIL(start_stmt())) {

LOG_WARN("fail start stmt", K(ret));

} else {

/* 将exec_result_设置到executor的运行时环境中，用于返回数据 */

/* 执行plan,

* 无论是本地、远程、还是分布式plan，除RootJob外，其余都会在execute_plan函数返回之前执行完毕

* exec_result_负责执行最后一个Job: RootJob

**/

if (OB_FAIL(executor_.init(physical_plan_))) {

SQL_LOG(WARN, "fail to init executor", K(ret), K(physical_plan_));

} else if (OB_FAIL(executor_.execute_plan(ctx))) {

SQL_LOG(WARN, "fail execute plan", K(ret));

}

}

NG_TRACE(do_open_plan_end);

return ret;

}

int ObResultSet::set_mysql_info()

{

int ret = OB_SUCCESS;

ObPhysicalPlanCtx *plan_ctx = get_exec_context().get_physical_plan_ctx();

int64_t pos = 0;

if (OB_ISNULL(plan_ctx)) {

ret = OB_ERR_UNEXPECTED;

SQL_LOG(WARN, "fail to get physical plan ctx");

} else if (stmt::T_UPDATE == get_stmt_type()) {

int result_len = snprintf(message_ + pos, MSG_SIZE - pos, OB_UPDATE_MSG_FMT, plan_ctx->get_row_matched_count(),

plan_ctx->get_row_duplicated_count(), warning_count_);

if (OB_UNLIKELY(result_len < 0) || OB_UNLIKELY(result_len >= MSG_SIZE - pos)) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("fail to snprintf to buff", K(ret));

}

} else if (stmt::T_REPLACE == get_stmt_type()

|| stmt::T_INSERT == get_stmt_type()) {

if (plan_ctx->get_row_matched_count() <= 1) {

//nothing to do

} else {

int result_len = snprintf(message_ + pos, MSG_SIZE - pos, OB_INSERT_MSG_FMT, plan_ctx->get_row_matched_count(),

plan_ctx->get_row_duplicated_count(), warning_count_);

if (OB_UNLIKELY(result_len < 0) || OB_UNLIKELY(result_len >= MSG_SIZE - pos)) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("fail to snprintf to buff", K(ret));

}

}

} else if (stmt::T_LOAD_DATA == get_stmt_type()) {

int result_len = snprintf(message_ + pos, MSG_SIZE - pos, OB_LOAD_DATA_MSG_FMT, plan_ctx->get_row_matched_count(),

plan_ctx->get_row_deleted_count(), plan_ctx->get_row_duplicated_count(), warning_count_);

if (OB_UNLIKELY(result_len < 0) || OB_UNLIKELY(result_len >= MSG_SIZE - pos)) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("fail to snprintf to buff", K(ret));

}

} else {

//nothing to do

}

return ret;

}

OB_INLINE void ObResultSet::store_affected_rows(ObPhysicalPlanCtx &plan_ctx)

{

int64_t affected_row = 0;

if (!ObStmt::is_dml_stmt(get_stmt_type())

&& (lib::is_oracle_mode() || !is_pl_stmt(get_stmt_type()))) {

affected_row = 0;

} else if (stmt::T_SELECT == get_stmt_type()) {

affected_row = plan_ctx.get_affected_rows();

if (lib::is_mysql_mode() && 0 == affected_row) {

affected_row = -1;

}

} else {

affected_row = get_affected_rows();

}

NG_TRACE_EXT(affected_rows, OB_ID(affected_rows), affected_row);

my_session_.set_affected_rows(affected_row);

}

OB_INLINE void ObResultSet::store_found_rows(ObPhysicalPlanCtx &plan_ctx)

{

int64_t rows = 1;

//如果是execute语句的话，则需要判断对应prepare语句的类型，如果是select的话，则会影响found_rows的值；

//to do by rongxuan.lc 20151127

if (plan_ctx.is_affect_found_row()) {

if (OB_UNLIKELY(stmt::T_EXPLAIN == get_stmt_type())) {

rows = 0;

my_session_.set_found_rows(rows);

} else {

int64_t found_rows = -1;

found_rows = plan_ctx.get_found_rows();

rows = found_rows == 0 ? return_rows_ : found_rows;

my_session_.set_found_rows(rows);

NG_TRACE_EXT(store_found_rows,

OB_ID(found_rows), found_rows,

OB_ID(return_rows), return_rows_);

}

}

return;

}

int ObResultSet::update_last_insert_id()

{

return store_last_insert_id(get_exec_context());

}

int ObResultSet::update_last_insert_id_to_client()

{

int ret = OB_SUCCESS;

ObPhysicalPlanCtx *plan_ctx = get_exec_context().get_physical_plan_ctx();

if (OB_ISNULL(plan_ctx)) {

ret = OB_ERR_UNEXPECTED;

SQL_LOG(WARN, "get plan ctx is NULL", K(ret));

} else {

set_last_insert_id_to_client(plan_ctx->calc_last_insert_id_to_client());

}

return ret;

}

int ObResultSet::update_is_result_accurate()

{

int ret = OB_SUCCESS;

if (stmt::T_SELECT == stmt_type_) {

ObPhysicalPlanCtx *plan_ctx = get_exec_context().get_physical_plan_ctx();

bool old_is_result_accurate = true;

if (OB_ISNULL(plan_ctx)) {

ret = OB_ERR_UNEXPECTED;

SQL_LOG(WARN, "get plan ctx is NULL", K(ret));

} else if (OB_FAIL(my_session_.get_is_result_accurate(old_is_result_accurate))) {

SQL_LOG(WARN, "faile to get is_result_accurate", K(ret));

} else {

bool is_result_accurate = plan_ctx->is_result_accurate();

SQL_LOG(DEBUG, "debug is_result_accurate for session", K(is_result_accurate),

K(old_is_result_accurate), K(ret));

if (is_result_accurate != old_is_result_accurate) {

// FIXME @qianfu 暂时写成update_sys_variable函数，以后再实现一个可以传入ObSysVarClassType并且

// 和set系统变量语句走同一套逻辑的函数，在这里调用

if (OB_FAIL(my_session_.update_sys_variable(SYS_VAR_IS_RESULT_ACCURATE, is_result_accurate ? 1 : 0))) {

LOG_WARN("fail to update result accurate", K(ret));

}

}

}

}

return ret;

}

int ObResultSet::store_last_insert_id(ObExecContext &ctx)

{

int ret = OB_SUCCESS;

ObPhysicalPlan* physical_plan_ = static_cast<ObPhysicalPlan*>(cache_obj_guard_.get_cache_obj());

if (ObStmt::is_dml_stmt(stmt_type_)

&& OB_LIKELY(NULL != physical_plan_)

&& OB_LIKELY(!physical_plan_->is_affected_last_insert_id())) {

//nothing to do

} else {

ObPhysicalPlanCtx *plan_ctx = ctx.get_physical_plan_ctx();

if (OB_ISNULL(plan_ctx)) {

ret = OB_ERR_UNEXPECTED;

SQL_LOG(WARN, "get plan ctx is NULL", K(plan_ctx));

} else {

uint64_t last_insert_id_session = plan_ctx->calc_last_insert_id_session();

SQL_LOG(DEBUG, "debug last_insert_id for session", K(last_insert_id_session), K(ret));

SQL_LOG(DEBUG, "debug last_insert_id changed", K(ret),

"last_insert_id_changed", plan_ctx->get_last_insert_id_changed());

if (plan_ctx->get_last_insert_id_changed()) {

ObObj last_insert_id;

last_insert_id.set_uint64(last_insert_id_session);

// FIXME @qianfu 暂时写成update_sys_variable函数，以后再实现一个可以传入ObSysVarClassType并且

// 和set系统变量语句走同一套逻辑的函数，在这里调用

if (OB_FAIL(my_session_.update_sys_variable(SYS_VAR_LAST_INSERT_ID, last_insert_id))) {

LOG_WARN("fail to update last_insert_id", K(ret));

} else if (OB_FAIL(my_session_.update_sys_variable(SYS_VAR_IDENTITY, last_insert_id))) {

LOG_WARN("succ update last_insert_id, but fail to update identity", K(ret));

} else {

NG_TRACE_EXT(last_insert_id, OB_ID(last_insert_id), last_insert_id_session);

}

}

if (OB_SUCC(ret)) {

// TODO when does observer should return last_insert_id to client?

set_last_insert_id_to_client(plan_ctx->calc_last_insert_id_to_client());

SQL_LOG(DEBUG, "zixu debug", K(ret),

"last_insert_id_to_client", get_last_insert_id_to_client());

}

}

}

return ret;

}

bool ObResultSet::need_rollback(int ret, int errcode, bool is_error_ignored) const

{

bool bret = false;

if (OB_SUCCESS != ret //当前close执行语句出错

|| (errcode != OB_SUCCESS && errcode != OB_ITER_END) //errcode是open阶段特意保存下来的错误

|| is_error_ignored) { //曾经出错，但是被忽略了；

bret = true;

}

return bret;

}

OB_INLINE int ObResultSet::do_close_plan(int errcode, ObExecContext &ctx)

{

int ret = common::OB_SUCCESS;

int pret = OB_SUCCESS;

int sret = OB_SUCCESS;

NG_TRACE(close_plan_begin);

ObPhysicalPlan* physical_plan_ = static_cast<ObPhysicalPlan*>(cache_obj_guard_.get_cache_obj());

if (OB_LIKELY(NULL != physical_plan_)) {

// executor_.close要在exec_result_.close之后，因为主线程get_next_row的时候会占着锁，

// exec_result_.close会释放锁，而executor_.close需要拿写锁来删掉该scheduler。

// FIXME qianfu.zpf 这里本来应该要调度线程结束才调用exec_result_.close的。调度线程给主线程push完最后一个

// task结果之后主线程就会往下走，有可能在调度线程结束之前主线程就调用了exec_result_.close，

// 但是由于目前调度线程给主线程push完最后一个task结果之后, 调度线程不会用到exec_result_.close中释放的变量，

// 因此目前这样写暂时是没有问题的。

// 以后要修正这里的逻辑。

int old_errcode = ctx.get_errcode();

if (OB_SUCCESS == old_errcode) {

// record error code generated in open-phase for after stmt trigger

ctx.set_errcode(errcode);

}

if (OB_ISNULL(exec_result_)) {

ret = OB_NOT_INIT;

LOG_WARN("exec result is null", K(ret));

} else if (OB_FAIL(exec_result_->close(ctx))) {

SQL_LOG(WARN, "fail close main query", K(ret));

}

// whether `close` is successful or not, restore ctx.errcode_

ctx.set_errcode(old_errcode);

// 通知该plan的所有task删掉对应的中间结果

int close_ret = OB_SUCCESS;

ObPhysicalPlanCtx *plan_ctx = NULL;

if (OB_ISNULL(plan_ctx = get_exec_context().get_physical_plan_ctx())) {

close_ret = OB_ERR_UNEXPECTED;

LOG_WARN("physical plan ctx is null");

} else if (OB_SUCCESS != (close_ret = executor_.close(ctx))) { // executor_.close里面会等到调度线程结束才返回。

SQL_LOG(WARN, "fail to close executor", K(ret), K(close_ret));

}

ObPxAdmission::exit_query_admission(my_session_, get_exec_context(), get_stmt_type(), *get_physical_plan());

// Finishing direct-insert must be executed after ObPxTargetMgr::release_target()

if ((OB_SUCCESS == close_ret)

&& (OB_SUCCESS == errcode || OB_ITER_END == errcode)

&& (stmt::T_INSERT == get_stmt_type())

&& (ObTableDirectInsertService::is_direct_insert(*physical_plan_))) {

// for insert /*+ append */ into select clause

int tmp_ret = OB_SUCCESS;

if (OB_TMP_FAIL(ObTableDirectInsertService::finish_direct_insert(ctx, *physical_plan_))) {

errcode_ = tmp_ret; // record error code

errcode = tmp_ret;

LOG_WARN("fail to finish direct insert", KR(tmp_ret));

}

}

// // 必须要在executor_.execute_plan运行之后再调用exec_result_的一系列函数。

// if (OB_FAIL(exec_result_.close(ctx))) {

// SQL_LOG(WARN, "fail close main query", K(ret));

// }

// 无论如何都reset_op_ctx

bool err_ignored = false;

if (OB_ISNULL(plan_ctx)) {

LOG_ERROR("plan is not NULL, but plan ctx is NULL", K(ret), K(errcode));

} else {

err_ignored = plan_ctx->is_error_ignored();

}

bool rollback = need_rollback(ret, errcode, err_ignored);

get_exec_context().set_errcode(errcode);

sret = end_stmt(rollback || OB_SUCCESS != pret);

// SQL_LOG(INFO, "end_stmt err code", K_(errcode), K(ret), K(pret), K(sret));

// if branch fail is returned from end_stmt, then return it first

if (OB_TRANS_XA_BRANCH_FAIL == sret) {

ret = OB_TRANS_XA_BRANCH_FAIL;

} else if (OB_FAIL(ret)) {

// nop

} else if (OB_SUCCESS != pret) {

ret = pret;

} else if (OB_SUCCESS != sret) {

ret = sret;

}

if (OB_SUCC(ret)) {

if (physical_plan_->need_record_plan_info()) {

if (ctx.get_feedback_info().is_valid() &&

physical_plan_->get_logical_plan().is_valid() &&

OB_FAIL(physical_plan_->set_feedback_info(ctx))) {

LOG_WARN("failed to set feedback info", K(ret));

} else {

physical_plan_->set_record_plan_info(false);

}

}

}

} else {

ret = OB_ERR_UNEXPECTED;

}

// 无论如何都reset掉executor_，否则前面调executor_.init的时候可能会报init twice

executor_.reset();

NG_TRACE(close_plan_end);

return ret;

}

int ObResultSet::close(int &client_ret)

{

int ret = OB_SUCCESS;

LinkExecCtxGuard link_guard(my_session_, get_exec_context());

FLTSpanGuard(close);

int do_close_plan_ret = OB_SUCCESS;

ObPhysicalPlan* physical_plan_ = static_cast<ObPhysicalPlan*>(cache_obj_guard_.get_cache_obj());

if (OB_LIKELY(NULL != physical_plan_)) {

if (OB_FAIL(my_session_.reset_tx_variable_if_remote_trans(

physical_plan_->get_plan_type()))) {

LOG_WARN("fail to reset tx_read_only if it is remote trans", K(ret));

} else {

my_session_.set_last_plan_id(physical_plan_->get_plan_id());

}

// 无论如何必须执行do_close_plan

if (OB_UNLIKELY(OB_SUCCESS != (do_close_plan_ret = do_close_plan(errcode_,

get_exec_context())))) {

SQL_LOG(WARN, "fail close main query", K(ret), K(do_close_plan_ret));

}

if (OB_SUCC(ret)) {

ret = do_close_plan_ret;

}

} else if (NULL != cmd_) {

ret = OB_SUCCESS; // cmd mode always return true in close phase

} else {

// inner sql executor, no plan or cmd. do nothing.

}

// open, get_next_row无论是否成功，close总是要调用

// 下面的逻辑保证对外吐出的是首次出现的错误码。 by xiaochu.yh

// 保存close之前的错误码，用来判断本次stmt是否需要回滚； by rongxuan.lc

if (OB_SUCCESS != errcode_ && OB_ITER_END != errcode_) {

ret = errcode_;

}

if (OB_SUCC(ret)) {

ObPhysicalPlanCtx *plan_ctx = get_exec_context().get_physical_plan_ctx();

if (OB_ISNULL(plan_ctx)) {

ret = OB_NOT_INIT;

LOG_WARN("result set isn't init", K(ret));

} else {

store_affected_rows(*plan_ctx);

store_found_rows(*plan_ctx);

}

}

if (OB_SUCC(ret) && get_stmt_type() == stmt::T_SELECT) {

if (OB_FAIL(update_is_result_accurate())) {

SQL_LOG(WARN, "failed to update is result_accurate", K(ret));

}

}

// set last_insert_id

int ins_ret = OB_SUCCESS;

if (OB_SUCCESS != ret

&& get_stmt_type() != stmt::T_INSERT

&& get_stmt_type() != stmt::T_REPLACE) {

// ignore when OB_SUCCESS != ret and stmt like select/update/delete... executed

} else if (OB_SUCCESS != (ins_ret = store_last_insert_id(get_exec_context()))) {

SQL_LOG(WARN, "failed to store last_insert_id", K(ret), K(ins_ret));

}

if (OB_SUCC(ret)) {

ret = ins_ret;

}

if (OB_SUCC(ret)) {

if (!get_exec_context().get_das_ctx().is_partition_hit()) {

my_session_.partition_hit().try_set_bool(false);

}

}

int prev_ret = ret;

bool async = false; // for debug purpose

if (OB_TRANS_XA_BRANCH_FAIL == ret) {

if (my_session_.associated_xa()) {

//兼容oracle，这里需要重置session状态

LOG_WARN("branch fail in global transaction", KPC(my_session_.get_tx_desc()));

ObSqlTransControl::clear_xa_branch(my_session_.get_xid(), my_session_.get_tx_desc());

my_session_.reset_tx_variable();

my_session_.disassociate_xa();

}

} else if (OB_NOT_NULL(physical_plan_)) {

//Because of the async close result we need set the partition_hit flag

//to the call back param, than close the result.

//But the das framwork set the partition_hit after result is closed.

//So we need to set the partition info at here.

if (is_end_trans_async()) {

ObCurTraceId::TraceId *cur_trace_id = NULL;

if (OB_ISNULL(cur_trace_id = ObCurTraceId::get_trace_id())) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("current trace id is NULL", K(ret));

set_end_trans_async(false);

} else {

observer::ObSqlEndTransCb &sql_end_cb = my_session_.get_mysql_end_trans_cb();

ObEndTransCbPacketParam pkt_param;

int fill_ret = OB_SUCCESS;

fill_ret = sql_end_cb.set_packet_param(pkt_param.fill(*this, my_session_, *cur_trace_id));

if (OB_SUCCESS != fill_ret) {

LOG_WARN("fail set packet param", K(ret));

set_end_trans_async(false);

}

}

}

ret = auto_end_plan_trans(*physical_plan_, ret, async);

}

if (is_user_sql_ && my_session_.need_reset_package()) {

// need_reset_package is set, it must be reset package, wether exec succ or not.

int tmp_ret = my_session_.reset_all_package_state_by_dbms_session(true);

if (OB_SUCCESS != tmp_ret) {

LOG_WARN("reset all package fail. ", K(tmp_ret), K(ret));

ret = OB_SUCCESS == ret ? tmp_ret : ret;

}

}

// notify close fail to listener

int err = OB_SUCCESS != do_close_plan_ret ? do_close_plan_ret : ret;

if (OB_SUCCESS != err && err != errcode_ && close_fail_cb_.is_valid()) {

close_fail_cb_(err, client_ret);

}

//Save the current execution state to determine whether to refresh location

//and perform other necessary cleanup operations when the statement exits.

DAS_CTX(get_exec_context()).get_location_router().save_cur_exec_status(ret);

//NG_TRACE_EXT(result_set_close, OB_ID(ret), ret, OB_ID(arg1), prev_ret,

//OB_ID(arg2), ins_ret, OB_ID(arg3), errcode_, OB_ID(async), async);

return ret; // 后面所有的操作都通过callback来完成

}

// 异步调用end_trans

// 1. 仅仅针对AC=1的phy plan，不针对cmd (除了还没实现的commit/rollback)

// 2. TODO:对于commit/rollback这个cmd，后面也需要走这个路径。现在还是走同步Callback。

OB_INLINE int ObResultSet::auto_end_plan_trans(ObPhysicalPlan& plan,

int ret,

bool &async)

{

NG_TRACE(auto_end_plan_begin);

bool in_trans = my_session_.is_in_transaction();

bool ac = true;

bool explicit_trans = my_session_.has_explicit_start_trans();

bool is_rollback = false;

my_session_.get_autocommit(ac);

async = false;

LOG_TRACE("auto_end_plan_trans.start", K(ret),

K(in_trans), K(ac), K(explicit_trans),

K(is_async_end_trans_submitted()));

// explicit start trans will disable auto-commit

if (!explicit_trans && ac) {

// Query like `select 1` will keep next scope set transaction xxx valid

// for example:

// set session transaction read only;

// set @@session.autocommit=1;

// set transaction read write;

// select 1;

// insert into t values(1); -- this will be success

//

// so, can not reset these transaction variables

//

// must always commit/rollback the transactional state in `ObTxDesc`

// for example:

// set session transaction isolation level SERIALIZABLE

// -- UDF with: select count(1) from t1;

// select UDF1() from dual; -- PL will remain transctional state after run UDF1

//

// after execute UDF1, snapshot is kept in ObTxDesc, must cleanup before run

// other Query

bool reset_tx_variable = plan.is_need_trans();

ObPhysicalPlanCtx *plan_ctx = NULL;

if (OB_ISNULL(plan_ctx = get_exec_context().get_physical_plan_ctx())) {

ret = OB_ERR_UNEXPECTED;

LOG_ERROR("physical plan ctx is null, won't call end trans!!!", K(ret));