/**

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

*/

#include "common/ob_clock_generator.h"

#include "lib/ob_errno.h"

#include "share/rc/ob_tenant_base.h"

#define USING_LOG_PREFIX SQL_EXE

#include "share/ob_schema_status_proxy.h" // ObSchemaStatusProxy

#include "share/schema/ob_tenant_schema_service.h"

#include "sql/ob_sql_trans_control.h"

#include "sql/engine/ob_physical_plan.h"

#include "sql/engine/ob_physical_plan_ctx.h"

#include "sql/parser/parse_malloc.h"

#include "sql/resolver/ob_stmt.h"

#include "sql/session/ob_sql_session_info.h"

#include "sql/ob_sql_trans_util.h"

#include "sql/ob_end_trans_callback.h"

#include "lib/oblog/ob_trace_log.h"

#include "storage/tx/ob_trans_service.h"

#include "storage/tx/ob_xa_service.h"

#include "storage/tx/ob_trans_define.h"

#include "storage/tablelock/ob_table_lock_service.h"

#include "sql/engine/ob_exec_context.h"

#include "sql/executor/ob_task_spliter.h"

#include "lib/profile/ob_perf_event.h"

#include "observer/ob_server_struct.h"

#include "observer/ob_server.h"

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

#include "storage/tx_storage/ob_ls_service.h"

#include "storage/ls/ob_ls_get_mod.h"

#include "storage/tablet/ob_tablet.h"

#include "sql/das/ob_das_dml_ctx_define.h"

#include "share/deadlock/ob_deadlock_detector_mgr.h"

#ifdef CHECK_SESSION

#error "redefine macro CHECK_SESSION"

#else

#define CHECK_SESSION(session) \

if (OB_SUCC(ret) && session->is_zombie()) { \

ret = OB_ERR_SESSION_INTERRUPTED; \

LOG_WARN("session has been killed", KR(ret), KPC(session)); \

}

#endif

#define CHECK_TX_FREE_ROUTE(exec_ctx, session, ...) \

if (OB_SUCC(ret) && session->is_txn_free_route_temp()) { \

__VA_ARGS__; \

ret = OB_ERR_UNEXPECTED; \

exec_ctx.set_need_disconnect(true); \

TRANS_LOG(ERROR, "trans act on txn temporary node", KR(ret), \

K(session->get_txn_free_route_ctx()), \

K(session->get_tx_id()), KPC(session)); \

if (session->get_tx_desc()) { \

session->get_tx_desc()->dump_and_print_trace(); \

} \

}

namespace oceanbase

{

using namespace common;

using namespace transaction;

using namespace share;

using namespace share::schema;

using namespace share::detector;

namespace sql

{

static int get_tx_service(ObBasicSessionInfo *session,

transaction::ObTransService *&txs)

{

int ret = OB_SUCCESS;

auto effective_tenant_id = session->get_effective_tenant_id();

if (OB_NOT_NULL(session->get_tx_desc())) {

auto tx_tenant_id = session->get_tx_desc()->get_tenant_id();

if (effective_tenant_id != tx_tenant_id) {

ret = OB_TENANT_ID_NOT_MATCH;

LOG_ERROR("effective_tenant_id not equals to tx_tenant_id", K(ret), K(effective_tenant_id), K(tx_tenant_id), KPC(session));

}

}

if (OB_SUCC(ret)) {

if (OB_ISNULL(txs = MTL_WITH_CHECK_TENANT(transaction::ObTransService*, effective_tenant_id))) {

ret = OB_ERR_UNEXPECTED;

LOG_ERROR("get_tx_service", K(ret), K(effective_tenant_id), K(MTL_ID()));

}

}

return ret;

}

static inline int get_lock_service(uint64_t tenant_id,

transaction::tablelock::ObTableLockService *&lock_service)

{

int ret = OB_SUCCESS;

lock_service = MTL_WITH_CHECK_TENANT(transaction::tablelock::ObTableLockService*,

tenant_id);

if (OB_ISNULL(lock_service)) {

ret = OB_ERR_UNEXPECTED;

LOG_ERROR("get_lock_service", K(ret), K(tenant_id), K(MTL_ID()));

}

return ret;

}

static int get_org_cluster_id_(ObSQLSessionInfo*, int64_t &);

static inline int build_tx_param_(ObSQLSessionInfo *session, ObTxParam &p, const bool *readonly = nullptr)

{

int ret = OB_SUCCESS;

int64_t org_cluster_id = OB_INVALID_ORG_CLUSTER_ID;

OZ (get_org_cluster_id_(session, org_cluster_id));

int64_t tx_timeout_us = 0;

session->get_tx_timeout(tx_timeout_us);

p.timeout_us_ = tx_timeout_us;

p.lock_timeout_us_ = session->get_trx_lock_timeout();

bool ro = OB_NOT_NULL(readonly) ? *readonly : session->get_tx_read_only();

p.access_mode_ = ro ? ObTxAccessMode::RD_ONLY : ObTxAccessMode::RW;

p.isolation_ = session->get_tx_isolation();

p.cluster_id_ = org_cluster_id;

return ret;

}

int ObSqlTransControl::create_stash_savepoint(ObExecContext &ctx, const ObString &name)

{

int ret = OB_SUCCESS;

transaction::ObTransService *txs = NULL;

ObSQLSessionInfo *session = GET_MY_SESSION(ctx);

CK (OB_NOT_NULL(session));

OZ (get_tx_service(session, txs));

OZ (acquire_tx_if_need_(txs, *session));

OZ (txs->create_stash_savepoint(*session->get_tx_desc(), name));

return ret;

}

int ObSqlTransControl::explicit_start_trans(ObExecContext &ctx, const bool read_only, const ObString hint)

{

int ret = OB_SUCCESS;

ObPhysicalPlanCtx *plan_ctx = GET_PHY_PLAN_CTX(ctx);

ObSQLSessionInfo *session = GET_MY_SESSION(ctx);

transaction::ObTransService *txs = NULL;

uint64_t tenant_id = 0;

ObTransID tx_id;

bool cleanup = true;

CK (OB_NOT_NULL(plan_ctx), OB_NOT_NULL(session));

CHECK_SESSION(session);

CHECK_TX_FREE_ROUTE(ctx, session, cleanup = false);

if (OB_SUCC(ret) && session->is_in_transaction()) {

ret = OB_ERR_UNEXPECTED;

cleanup = false;

LOG_ERROR("nested start transaction not allowed", KR(ret), K(ctx));

}

OX (tenant_id = session->get_effective_tenant_id());

OZ (get_tx_service(session, txs), tenant_id);

if (OB_SUCC(ret) && OB_NOT_NULL(session->get_tx_desc())) {

ObSQLSessionInfo::LockGuard data_lock_guard(session->get_thread_data_lock());

auto *tx_desc = session->get_tx_desc();

if (tx_desc->get_tenant_id() != tenant_id) {

LOG_ERROR("switch tenant but hold tx_desc", K(tenant_id), KPC(tx_desc));

}

txs->release_tx(*tx_desc);

session->get_tx_desc() = NULL;

}

ObTxParam &tx_param = plan_ctx->get_trans_param();

OZ (build_tx_param_(session, tx_param, &read_only));

OZ (txs->acquire_tx(session->get_tx_desc(), session->get_sessid()));

OZ (txs->start_tx(*session->get_tx_desc(), tx_param), tx_param);

OX (tx_id = session->get_tx_desc()->get_tx_id());

if (OB_FAIL(ret) && cleanup && OB_NOT_NULL(txs) && OB_NOT_NULL(session->get_tx_desc())) {

ObSQLSessionInfo::LockGuard data_lock_guard(session->get_thread_data_lock());

txs->release_tx(*session->get_tx_desc());

session->get_tx_desc() = NULL;

}

OX (session->get_raw_audit_record().trans_id_ = session->get_tx_id());

NG_TRACE_EXT(start_trans, OB_ID(ret), ret,

OB_ID(trans_id), tx_id.get_id(),

OB_ID(timeout), tx_param.timeout_us_,

OB_ID(start_time), session ? session->get_query_start_time() : 0);

if (hint.length()) {

LOG_INFO("explicit start trans with hint", "trans_id", tx_id,

K(ret), K(hint), K(read_only), "session_id", (session ? session->get_sessid() : 0));

}

#ifndef NDEBUG

LOG_INFO("start_trans", K(ret), K(tx_id), KPC(session), K(read_only), K(ctx.get_execution_id()));

#endif

return ret;

}

int ObSqlTransControl::implicit_end_trans(ObExecContext &exec_ctx,

const bool is_rollback,

ObEndTransAsyncCallback *callback,

bool reset_trans_variable)

{

int ret = OB_SUCCESS;

#ifndef NDEBUG

LOG_INFO("implicit end trans", K(is_rollback), K(exec_ctx.get_execution_id()), KP(callback));

#endif

ObSQLSessionInfo *session = GET_MY_SESSION(exec_ctx);

CK (OB_NOT_NULL(session));

int64_t tx_id = 0;

OX (tx_id = session->get_tx_id().get_id());

CHECK_TX_FREE_ROUTE(exec_ctx, session);

FLTSpanGuard(end_transaction);

OZ(end_trans(exec_ctx, is_rollback, false, callback, reset_trans_variable));

FLT_SET_TAG(trans_id, tx_id);

return ret;

}

int ObSqlTransControl::explicit_end_trans(ObExecContext &exec_ctx, const bool is_rollback, const ObString hint)

{

int ret = OB_SUCCESS;

#ifndef NDEBUG

LOG_INFO("explicit end trans", K(is_rollback), K(exec_ctx.get_execution_id()));

#endif

FLTSpanGuard(end_transaction);

ObTransID txn_id;

ObEndTransAsyncCallback *callback = NULL;

ObSQLSessionInfo *session = GET_MY_SESSION(exec_ctx);

CK (OB_NOT_NULL(session));

if (OB_SUCC(ret) && session->get_tx_desc()) {

txn_id = session->get_tx_desc()->tid();

}

CHECK_TX_FREE_ROUTE(exec_ctx, session);

if (exec_ctx.is_end_trans_async()) {

CK (OB_NOT_NULL(callback = &session->get_end_trans_cb()));

}

OZ (end_trans(exec_ctx, is_rollback, true, callback));

FLT_SET_TAG(trans_id, txn_id.get_id());

if (hint.length()) {

LOG_INFO("explicit end trans with hint",

"trans_id", txn_id, "action", (is_rollback ? "ROLLBACK" : "COMMIT"),

K(ret), K(hint), "session_id", session->get_sessid());

}

return ret;

}

int ObSqlTransControl::end_trans(ObExecContext &exec_ctx,

const bool is_rollback,

const bool is_explicit,

ObEndTransAsyncCallback *callback,

bool reset_trans_variable)

{

int ret = OB_SUCCESS;

bool sync = false;

ObSQLSessionInfo *session = GET_MY_SESSION(exec_ctx);

ObPhysicalPlanCtx *plan_ctx = GET_PHY_PLAN_CTX(exec_ctx);

#ifndef NDEBUG

LOG_INFO("end_trans", K(session->is_in_transaction()),

K(session->has_explicit_start_trans()),

K(exec_ctx.get_execution_id()),

KP(callback));

#endif

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

ret = OB_INVALID_ARGUMENT;

LOG_ERROR("invalid argument", K(ret), KPC(session), KP(plan_ctx));

} else if (OB_NOT_NULL(callback)) {

callback->set_is_need_rollback(is_rollback);

callback->set_end_trans_type(is_explicit ?

ObExclusiveEndTransCallback::END_TRANS_TYPE_EXPLICIT :

ObExclusiveEndTransCallback::END_TRANS_TYPE_IMPLICIT);

}

if (OB_FAIL(ret)) {

} else if (!session->is_in_transaction()) {

if (!is_rollback && OB_NOT_NULL(callback)) {

if (OB_FAIL(inc_session_ref(session))) {

LOG_WARN("fail to inc session ref", K(ret));

} else {

callback->handout();

}

callback->callback(OB_SUCCESS);

} else {

reset_session_tx_state(session, true, reset_trans_variable);

exec_ctx.set_need_disconnect(false);

}

} else {

// add tx id to AuditRecord

session->get_raw_audit_record().trans_id_ = session->get_tx_id();

int64_t expire_ts = get_stmt_expire_ts(plan_ctx, *session);

if (OB_FAIL(do_end_trans_(session,

is_rollback,

is_explicit,

expire_ts,

callback))) {

}

bool need_disconnect = false;

ObSQLUtils::check_if_need_disconnect_after_end_trans(ret,

is_rollback,

is_explicit,

need_disconnect);

exec_ctx.set_need_disconnect(need_disconnect);

if (is_rollback || OB_FAIL(ret) || !callback) {

bool reuse_tx = OB_SUCCESS == ret

|| OB_TRANS_COMMITED == ret

|| OB_TRANS_ROLLBACKED == ret;

reset_session_tx_state(session, reuse_tx, reset_trans_variable);

}

}

if (callback && !is_rollback) {

exec_ctx.get_trans_state().set_end_trans_executed(OB_SUCC(ret));

}

return ret;

}

int ObSqlTransControl::kill_query_session(ObSQLSessionInfo &session,

const ObSQLSessionState &status)

{

int ret = OB_SUCCESS;

if (session.get_in_transaction()) {

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

auto tx_tenant_id = tx_desc->get_tenant_id();

MTL_SWITCH(tx_tenant_id) {

transaction::ObTransService *txs = NULL;

CK(OB_NOT_NULL(txs = MTL_WITH_CHECK_TENANT(transaction::ObTransService*,

tx_tenant_id)));

OZ(txs->interrupt(*tx_desc, OB_ERR_QUERY_INTERRUPTED),

tx_desc->get_tx_id(), status);

LOG_INFO("kill_query_session", K(ret), K(session), K(tx_desc->get_tx_id()),

"session_status", status);

}

}

return ret;

}

int ObSqlTransControl::kill_idle_timeout_tx(ObSQLSessionInfo *session)

{

int ret = OB_SUCCESS;

using namespace oceanbase::transaction;

if (!session->can_txn_free_route()) {

ret = kill_tx(session, OB_TRANS_IDLE_TIMEOUT);

}

return ret;

}

int ObSqlTransControl::kill_tx(ObSQLSessionInfo *session, int cause)

{

int ret = OB_SUCCESS;

if (!session->get_is_deserialized() && session->is_in_transaction()) {

auto session_id = session->get_sessid();

LOG_INFO("begin to kill tx", K(cause), K(session_id), KPC(session));

transaction::ObTxDesc *tx_desc = session->get_tx_desc();

auto tx_tenant_id = tx_desc->get_tenant_id();

const ObTransID tx_id = tx_desc->get_tx_id();

auto tx_free_route_tmp = session->is_txn_free_route_temp();

MTL_SWITCH(tx_tenant_id) {

ObSQLSessionInfo::LockGuard data_lock_guard(session->get_thread_data_lock());

if (tx_free_route_tmp) {

// if XA-txn is on this server, we have acquired its ref, release ref

// and disassocate with session

if (tx_desc->is_xa_trans() && tx_desc->get_addr() == GCONF.self_addr_) {

auto txs = MTL(transaction::ObTransService*);

CK (OB_NOT_NULL(txs), session_id, tx_id);

OZ (txs->release_tx_ref(*tx_desc), session_id, tx_id);

session->get_tx_desc() = NULL;

}

} else if (tx_desc->is_xa_trans()) {

const transaction::ObXATransID xid = session->get_xid();

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

auto xas = MTL(transaction::ObXAService *);

CK (OB_NOT_NULL(xas));

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

OZ (xas->handle_terminate_for_xa_branch(session->get_xid(), tx_desc, session->get_xa_end_timeout_seconds()),

xid, global_tx_type, session_id, tx_id);

// currently, tx_desc is NULL

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

OZ (xas->rollback_for_dblink_trans(tx_desc), ret, xid, global_tx_type, tx_id);

// currently, tx_desc is NULL

} else {

ret = OB_ERR_UNEXPECTED;

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

}

session->get_tx_desc() = NULL;

} else {

transaction::ObTransService *txs = NULL;

CK(OB_NOT_NULL(txs = MTL_WITH_CHECK_TENANT(transaction::ObTransService*, tx_tenant_id)));

OZ(txs->abort_tx(*tx_desc, cause), *session, tx_desc->get_tx_id());

}

// NOTE that the tx_desc is set to NULL in xa case, DO NOT print anything in tx_desc

LOG_INFO("kill tx done", K(ret), K(cause), K(session_id), K(tx_id), K(tx_free_route_tmp));

}

}

return ret;

}

int ObSqlTransControl::rollback_trans(ObSQLSessionInfo *session,

bool &need_disconnect)

{

int ret = OB_SUCCESS;

if (OB_ISNULL(session)) {

ret = OB_INVALID_ARGUMENT;

LOG_ERROR("", K(ret), K(session));

} else if (OB_NOT_NULL(session->get_tx_desc())) {

need_disconnect = false;

if (OB_FAIL(do_end_trans_(session, true, false, INT64_MAX, NULL))) {

LOG_WARN("fail rollback trans", K(ret), KPC(session->get_tx_desc()));

ObSQLUtils::check_if_need_disconnect_after_end_trans(

ret, true, false, need_disconnect);

}

reset_session_tx_state(session);

} else {

reset_session_tx_state(session);

}

return ret;

}

int ObSqlTransControl::do_end_trans_(ObSQLSessionInfo *session,

const bool is_rollback,

const bool is_explicit,

const int64_t expire_ts,

ObEndTransAsyncCallback *callback)

{

int ret = OB_SUCCESS;

transaction::ObTxDesc *&tx_ptr = session->get_tx_desc();

bool is_detector_exist = false;

int tmp_ret = OB_SUCCESS;

if (OB_ISNULL(MTL(share::detector::ObDeadLockDetectorMgr*))) {

tmp_ret = OB_BAD_NULL_ERROR;

DETECT_LOG(WARN, "MTL ObDeadLockDetectorMgr is NULL", K(tmp_ret), K(tx_ptr->tid()));

} else if (OB_TMP_FAIL(MTL(share::detector::ObDeadLockDetectorMgr*)->

check_detector_exist(tx_ptr->tid(), is_detector_exist))) {

DETECT_LOG(WARN, "fail to check detector exist, may causing detector leak", K(tmp_ret),

K(tx_ptr->tid()));

} else if (is_detector_exist) {

ObTransDeadlockDetectorAdapter::unregister_from_deadlock_detector(tx_ptr->tid(),

ObTransDeadlockDetectorAdapter::UnregisterPath::DO_END_TRANS);

}

if (session->associated_xa() && !is_explicit) {

ret = OB_TRANS_XA_RMFAIL;

LOG_ERROR("executing do end trans in xa", K(ret), K(session->get_xid()), KPC(tx_ptr));

} else {

/*

* normal transaction control

*

* call convention:

* if trans_service.end_trans failed:

* 1) tx will be aborted (if tx exist and not terminated)

* 2) the callback will not been called

*/

transaction::ObTransService *txs = NULL;

uint64_t tenant_id = session->get_effective_tenant_id();

auto &trace_info = session->get_ob_trace_info();

if (OB_FAIL(get_tx_service(session, txs))) {

LOG_ERROR("fail to get trans service", K(ret), K(tenant_id));

} else if (is_rollback) {

ret = txs->rollback_tx(*tx_ptr);

} else if (callback) {

if (OB_FAIL(inc_session_ref(session))) {

LOG_WARN("fail to inc session ref", K(ret));

} else {

callback->handout();

// Add ASH flags to async commit of transactions

// In the end of async commit in func named ` ObEndTransAsyncCallback::callback() `,

// set the ash flag named `in_committing_` to false.

ObActiveSessionGuard::get_stat().in_committing_ = true;

if(OB_FAIL(txs->submit_commit_tx(*tx_ptr, expire_ts, *callback, &trace_info))) {

LOG_WARN("submit commit tx fail", K(ret), KP(callback), K(expire_ts), KPC(tx_ptr));

GCTX.session_mgr_->revert_session(session);

callback->handin();

}

}

} else {

ACTIVE_SESSION_FLAG_SETTER_GUARD(in_committing);

if (OB_FAIL(txs->commit_tx(*tx_ptr, expire_ts, &trace_info))) {

LOG_WARN("sync commit tx fail", K(ret), K(expire_ts), KPC(tx_ptr));

}

}

}

bool print_log = OB_FAIL(ret);

#ifndef NDEBUG

print_log = true;

#endif

if (print_log) {

LOG_INFO("do_end_trans", K(ret),

KPC(tx_ptr),

K(is_rollback),

K(expire_ts),

K(is_explicit),

KP(callback));

}

return ret;

}

int ObSqlTransControl::decide_trans_read_interface_specs(

const ObConsistencyLevel &sql_consistency_level,

ObTxConsistencyType &trans_consistency_type)

{

int ret = OB_SUCCESS;

if (sql_consistency_level == STRONG) {

trans_consistency_type = ObTxConsistencyType::CURRENT_READ;

} else if (sql_consistency_level == WEAK || sql_consistency_level == FROZEN){

trans_consistency_type = ObTxConsistencyType::BOUNDED_STALENESS_READ;

} else {

ret = OB_INVALID_ARGUMENT;

SQL_LOG(ERROR, "invalid consistency_level", K(sql_consistency_level));

}

return ret;

}

int ObSqlTransControl::start_stmt(ObExecContext &exec_ctx)

{

int ret = OB_SUCCESS;

ObSQLSessionInfo *session = GET_MY_SESSION(exec_ctx);

ObPhysicalPlanCtx *plan_ctx = GET_PHY_PLAN_CTX(exec_ctx);

const ObPhysicalPlan *plan = plan_ctx->get_phy_plan();

ObDASCtx &das_ctx = DAS_CTX(exec_ctx);

transaction::ObTransService *txs = NULL;

uint64_t tenant_id = 0;

CK (OB_NOT_NULL(session), OB_NOT_NULL(plan_ctx), OB_NOT_NULL(plan));

OX (tenant_id = session->get_effective_tenant_id());

OX (session->get_trans_result().reset());

OZ (get_tx_service(session, txs), tenant_id);

OZ (acquire_tx_if_need_(txs, *session));

OZ (stmt_sanity_check_(session, plan, plan_ctx));

bool start_hook = false;

if (!ObSQLUtils::is_nested_sql(&exec_ctx)) {

OZ (txs->sql_stmt_start_hook(session->get_xid(), *session->get_tx_desc(), session->get_sessid(), get_real_session_id(*session)));

if (OB_SUCC(ret)) {

start_hook = true;

OX (session->get_tx_desc()->clear_interrupt());

}

}

if (OB_SUCC(ret)

&& txs->get_tx_elr_util().check_and_update_tx_elr_info(*session->get_tx_desc())) {

LOG_WARN("check and update tx elr info", K(ret), KPC(session->get_tx_desc()));

}

uint32_t session_id = 0;

ObTxDesc *tx_desc = NULL;

bool is_plain_select = false;

int64_t nested_level = 0;

OX (nested_level = exec_ctx.get_nested_level());

OX (session_id = session->get_sessid());

OX (tx_desc = session->get_tx_desc());

OX (is_plain_select = plan->is_plain_select());

OX (tx_desc->clear_interrupt());

if (OB_SUCC(ret) && !is_plain_select) {

OZ (stmt_setup_savepoint_(session, das_ctx, plan_ctx, txs, nested_level), session_id, *tx_desc);

}

OZ (stmt_setup_snapshot_(session, das_ctx, plan, plan_ctx, txs), session_id, *tx_desc);

// add tx id to AuditRecord

OX (session->get_raw_audit_record().trans_id_ = session->get_tx_id());

// add snapshot info to AuditRecord

if (OB_SUCC(ret)) {

ObAuditRecordData &audit_record = session->get_raw_audit_record();

auto &snapshot = das_ctx.get_snapshot();

auto &ar_snapshot = audit_record.snapshot_;

ar_snapshot.version_ = snapshot.core_.version_;

ar_snapshot.tx_id_ = snapshot.core_.tx_id_.get_id();

ar_snapshot.scn_ = snapshot.core_.scn_.cast_to_int();

ar_snapshot.source_ = snapshot.get_source_name().ptr();

}

if (OB_SUCC(ret) && !session->has_start_stmt()) {

OZ (session->set_start_stmt());

}

if (plan->is_contain_oracle_trx_level_temporary_table()) {

OX (tx_desc->set_with_temporary_table());

}

if (OB_FAIL(ret) && start_hook) {

int tmp_ret = txs->sql_stmt_end_hook(session->get_xid(), *session->get_tx_desc());

if (OB_SUCCESS != tmp_ret) {

LOG_WARN("call sql stmt end hook fail", K(tmp_ret));

}

}

if (OB_SUCC(ret)

&& !ObSQLUtils::is_nested_sql(&exec_ctx)

&& das_ctx.get_snapshot().core_.version_.is_valid()) {

// maintain the read snapshot version on session for multi-version garbage

// colloecor. It is maintained for all cases except remote exection with ac

// = 1. So we need carefully design the version for the corner case.

session->set_reserved_snapshot_version(das_ctx.get_snapshot().core_.version_);

}

bool print_log = false;

#ifndef NDEBUG

print_log = true;

#else

if (OB_FAIL(ret)) { print_log = true; }

#endif

if (print_log) {

bool auto_commit = false;

session->get_autocommit(auto_commit);

auto plan_type = plan->get_location_type();

auto stmt_type = plan->get_stmt_type();

auto has_for_update = plan->has_for_update();

auto use_das = plan->use_das();

auto &trans_result = session->get_trans_result();

auto query_start_time = session->get_query_start_time();

auto &snapshot = das_ctx.get_snapshot();

auto savepoint = das_ctx.get_savepoint();

LOG_INFO("start stmt", K(ret),

K(auto_commit),

K(session_id),

K(snapshot),

K(savepoint),

KPC(tx_desc),

K(plan_type),

K(stmt_type),

K(has_for_update),

K(query_start_time),

K(use_das),

K(nested_level),

KPC(session),

K(plan),

"consistency_level_in_plan_ctx", plan_ctx->get_consistency_level(),

K(trans_result));

}

return ret;

}

int ObSqlTransControl::stmt_sanity_check_(ObSQLSessionInfo *session,

const ObPhysicalPlan *plan,

ObPhysicalPlanCtx *plan_ctx)

{

int ret = OB_SUCCESS;

auto current_consist_level = plan_ctx->get_consistency_level();

CK (current_consist_level != ObConsistencyLevel::INVALID_CONSISTENCY);

bool is_plain_select = plan->is_plain_select();

// adjust stmt's consistency level

if (OB_SUCC(ret)) {

// Weak read statement with inner table should be converted to strong read.

// For example, schema refresh statement;

if (plan->is_contain_inner_table() ||

(!is_plain_select && current_consist_level != ObConsistencyLevel::STRONG)) {

plan_ctx->set_consistency_level(ObConsistencyLevel::STRONG);

}

}

// check isolation with consistency type

if (OB_SUCC(ret) && session->is_in_transaction()) {

auto iso = session->get_tx_desc()->get_isolation_level();

auto cl = plan_ctx->get_consistency_level();

if (ObConsistencyLevel::WEAK == cl && (iso == ObTxIsolationLevel::SERIAL || iso == ObTxIsolationLevel::RR)) {

ret = OB_NOT_SUPPORTED;

TRANS_LOG(ERROR, "statement of weak consistency is not allowed under SERIALIZABLE isolation",

KR(ret), "trans_id", session->get_tx_id(), "consistency_level", cl);

LOG_USER_ERROR(OB_NOT_SUPPORTED, "weak consistency under SERIALIZABLE and REPEATABLE-READ isolation level");

}

}

return ret;

}

int ObSqlTransControl::stmt_setup_snapshot_(ObSQLSessionInfo *session,

ObDASCtx &das_ctx,

const ObPhysicalPlan *plan,

const ObPhysicalPlanCtx *plan_ctx,

transaction::ObTransService *txs)

{

int ret = OB_SUCCESS;

auto cl = plan_ctx->get_consistency_level();

auto &snapshot = das_ctx.get_snapshot();

if (cl == ObConsistencyLevel::WEAK || cl == ObConsistencyLevel::FROZEN) {

SCN snapshot_version = SCN::min_scn();

const bool local_single_ls = plan->is_local_plan() &&

OB_PHY_PLAN_LOCAL == plan->get_location_type();

if (OB_FAIL(txs->get_weak_read_snapshot_version(session->get_ob_max_read_stale_time(),

local_single_ls,

snapshot_version))) {

TRANS_LOG(WARN, "get weak read snapshot fail", KPC(txs));

int64_t stale_time = session->get_ob_max_read_stale_time();

int64_t refresh_interval = GCONF.weak_read_version_refresh_interval;

if (stale_time > 0 && refresh_interval > stale_time) {

TRANS_LOG(WARN, "weak_read_version_refresh_interval is larger than ob_max_read_stale_time ",

K(refresh_interval), K(stale_time), KPC(txs));

}

} else {

snapshot.init_weak_read(snapshot_version);

}

// 1) acquire snapshot version when insert operator is executed

// 2) don't resolve RR and SERIALIZABLE isolation scenario temporarily, because of remote stmt plan

} else if (plan->is_plain_insert()

&& session->get_tx_isolation() != ObTxIsolationLevel::SERIAL

&& session->get_tx_isolation() != ObTxIsolationLevel::RR) {

auto &tx_desc = *session->get_tx_desc();

snapshot.init_none_read();

snapshot.core_.tx_id_ = tx_desc.get_tx_id();

snapshot.core_.scn_ = tx_desc.get_tx_seq();

} else {

auto &tx_desc = *session->get_tx_desc();

int64_t stmt_expire_ts = get_stmt_expire_ts(plan_ctx, *session);

share::ObLSID first_ls_id;

bool local_single_ls_plan = false;

const bool local_single_ls_plan_maybe = plan->is_local_plan() &&

OB_PHY_PLAN_LOCAL == plan->get_location_type();

if (local_single_ls_plan_maybe) {

if (OB_FAIL(get_first_lsid(das_ctx, first_ls_id))) {

} else if (!first_ls_id.is_valid()) {

// do nothing

} else if (OB_FAIL(txs->get_ls_read_snapshot(tx_desc,

session->get_tx_isolation(),

first_ls_id,

stmt_expire_ts,

snapshot))) {

} else {

local_single_ls_plan = has_same_lsid(das_ctx, snapshot, first_ls_id);

}

}

if (OB_SUCC(ret) && !local_single_ls_plan) {

ret = txs->get_read_snapshot(tx_desc,

session->get_tx_isolation(),

stmt_expire_ts,

snapshot);

}

if (OB_FAIL(ret)) {

LOG_WARN("fail to get snapshot", K(ret), K(local_single_ls_plan), K(first_ls_id), KPC(session));

}

}

return ret;

}

int ObSqlTransControl::stmt_refresh_snapshot(ObExecContext &exec_ctx) {

int ret = OB_SUCCESS;

ObSQLSessionInfo *session = GET_MY_SESSION(exec_ctx);

ObDASCtx &das_ctx = DAS_CTX(exec_ctx);

ObPhysicalPlanCtx *plan_ctx = GET_PHY_PLAN_CTX(exec_ctx);

const ObPhysicalPlan *plan = plan_ctx->get_phy_plan();

transaction::ObTransService *txs = NULL;

if (sql::stmt::T_INSERT == plan->get_stmt_type() || sql::stmt::T_INSERT_ALL == plan->get_stmt_type()) {

//NOTE: oracle insert and insert all stmt can't see the evaluated results of before stmt trigger, no need to refresh snapshot

} else if (OB_FAIL(get_tx_service(session, txs))) {

LOG_WARN("failed to get transaction service", K(ret));

} else if (OB_FAIL(stmt_setup_snapshot_(session, das_ctx, plan, plan_ctx, txs))) {

LOG_WARN("failed to set snapshot", K(ret));

}

return ret;

}

int ObSqlTransControl::set_fk_check_snapshot(ObExecContext &exec_ctx)

{

int ret = OB_SUCCESS;

ObSQLSessionInfo *session = GET_MY_SESSION(exec_ctx);

ObDASCtx &das_ctx = DAS_CTX(exec_ctx);

ObPhysicalPlanCtx *plan_ctx = GET_PHY_PLAN_CTX(exec_ctx);

const ObPhysicalPlan *plan = plan_ctx->get_phy_plan();

// insert stmt does not set snapshot by default, set snapshopt for foreign key check induced by insert heres

if (plan->is_plain_insert()) {

transaction::ObTransService *txs = NULL;

auto &snapshot = das_ctx.get_snapshot();

auto &tx_desc = *session->get_tx_desc();

int64_t stmt_expire_ts = get_stmt_expire_ts(plan_ctx, *session);

share::ObLSID local_ls_id;

bool local_single_ls_plan = plan->is_local_plan()

&& OB_PHY_PLAN_LOCAL == plan->get_location_type()

&& has_same_lsid(das_ctx, snapshot, local_ls_id);

if (OB_FAIL(get_tx_service(session, txs))) {

LOG_WARN("failed to get transaction service", K(ret));

} else {

if (local_single_ls_plan) {

ret = txs->get_ls_read_snapshot(tx_desc,

session->get_tx_isolation(),

local_ls_id,

stmt_expire_ts,

snapshot);

} else {

ret = txs->get_read_snapshot(tx_desc,

session->get_tx_isolation(),

stmt_expire_ts,

snapshot);

}

if (OB_FAIL(ret)) {

LOG_WARN("fail to get snapshot", K(ret), K(local_ls_id), KPC(session));

}

}

}

return ret;

}

int ObSqlTransControl::stmt_setup_savepoint_(ObSQLSessionInfo *session,

ObDASCtx &das_ctx,

ObPhysicalPlanCtx *plan_ctx,

transaction::ObTransService* txs,

const int64_t nested_level)

{

int ret = OB_SUCCESS;

ObTxParam &tx_param = plan_ctx->get_trans_param();

OZ (build_tx_param_(session, tx_param));

auto &tx = *session->get_tx_desc();

transaction::ObTxSEQ savepoint;

OZ (txs->create_implicit_savepoint(tx, tx_param, savepoint, nested_level == 0), tx, tx_param);

OX (das_ctx.set_savepoint(savepoint));

return ret;

}

#define CHECK_TXN_FREE_ROUTE_ALLOWED() \

if (OB_SUCC(ret) && !session->is_inner() && session->is_txn_free_route_temp()) { \

ret = OB_TRANS_FREE_ROUTE_NOT_SUPPORTED; \

LOG_WARN("current stmt is not allowed executed on txn tmp node", K(ret), \

K(session->get_txn_free_route_ctx()), KPC(session)); \

}

int ObSqlTransControl::create_savepoint(ObExecContext &exec_ctx,

const ObString &sp_name,

const bool user_create)

{

int ret = OB_SUCCESS;

ObSQLSessionInfo *session = GET_MY_SESSION(exec_ctx);

transaction::ObTransService *txs = NULL;

CK (OB_NOT_NULL(session));

CHECK_SESSION (session);

CHECK_TXN_FREE_ROUTE_ALLOWED();

OZ (get_tx_service(session, txs));

OZ (acquire_tx_if_need_(txs, *session));

bool start_hook = false;

OZ(start_hook_if_need_(*session, txs, start_hook));

OZ (txs->create_explicit_savepoint(*session->get_tx_desc(), sp_name, get_real_session_id(*session), user_create), sp_name);

if (start_hook) {

int tmp_ret = txs->sql_stmt_end_hook(session->get_xid(), *session->get_tx_desc());

if (OB_SUCCESS != tmp_ret) {

LOG_WARN("call sql stmt end hook fail", K(tmp_ret));

ret = COVER_SUCC(tmp_ret);

}

}

return ret;

}

uint32_t ObSqlTransControl::get_real_session_id(ObSQLSessionInfo &session)

{

return session.get_xid().empty() ? 0 : (session.get_proxy_sessid() != 0 ? session.get_proxy_sessid() : session.get_sessid());

}

int ObSqlTransControl::get_first_lsid(const ObDASCtx &das_ctx, share::ObLSID &first_lsid)

{

int ret = OB_SUCCESS;

const DASTableLocList &table_locs = das_ctx.get_table_loc_list();

if (!table_locs.empty()) {

const ObDASTableLoc *first_table_loc = table_locs.get_first();

const DASTabletLocList &tablet_locs = first_table_loc->get_tablet_locs();

if (!tablet_locs.empty()) {

const ObDASTabletLoc *tablet_loc = tablet_locs.get_first();

first_lsid = tablet_loc->ls_id_;

}

}

return ret;

}

bool ObSqlTransControl::has_same_lsid(const ObDASCtx &das_ctx,

const transaction::ObTxReadSnapshot &snapshot,

share::ObLSID &first_lsid)

{

int ret = OB_SUCCESS;

bool bret = true;

ObLSHandle ls_handle;

const share::SCN snapshot_version = snapshot.core_.version_;

const DASTableLocList &table_locs = das_ctx.get_table_loc_list();

FOREACH_X(table_node, table_locs, bret) {

ObDASTableLoc *table_loc = *table_node;

for (DASTabletLocListIter tablet_node = table_loc->tablet_locs_begin();

bret && tablet_node != table_loc->tablet_locs_end(); ++tablet_node) {

ObDASTabletLoc *tablet_loc = *tablet_node;

const ObTabletID tablet_id = tablet_loc->tablet_id_;

if (first_lsid != tablet_loc->ls_id_) {

bret = false;

}

if (bret && !ls_handle.is_valid()) {

ObLSService *ls_svr = NULL;

if (OB_ISNULL(ls_svr = MTL(ObLSService *))) {

bret = false;

} else if (OB_FAIL(ls_svr->get_ls(first_lsid, ls_handle, ObLSGetMod::TRANS_MOD))) {

bret = false;

} else {

// do nothing

}

}

if (bret) {

ObLS *ls = NULL;

ObLSTabletService *ls_tablet_service = NULL;

if (OB_ISNULL(ls = ls_handle.get_ls())) {

bret = false;

} else if (OB_ISNULL(ls_tablet_service = ls->get_tablet_svr())) {

bret = false;

} else {

ObTablet *tablet = NULL;

ObTabletHandle tablet_handle;

if (OB_FAIL(ls_tablet_service->get_tablet(tablet_id, tablet_handle, 100 * 1000))) {

bret = false;

} else if (OB_ISNULL(tablet = tablet_handle.get_obj())) {

bret = false;

} else if (!tablet->get_tablet_meta().transfer_info_.is_valid()) {

bret = false;

} else if (tablet->get_tablet_meta().transfer_info_.transfer_start_scn_ >= snapshot_version) {

bret = false;

} else {

// do nothing

}

}

}

if (bret && common::ObRole::FOLLOWER == snapshot.snapshot_ls_role_) {

ObLS *ls = NULL;

if (OB_ISNULL(ls = ls_handle.get_ls())) {

bret = false;

LOG_WARN("invalid ls", K(bret), K(first_lsid), K(snapshot));

} else if (!(ls->get_dup_table_ls_handler()->is_dup_tablet(tablet_id))) {

bret = false;

LOG_WARN("There is a normal tablet, retry to acquire snapshot with gts", K(bret), K(first_lsid),

K(snapshot), K(tablet_loc));

}

}

}

}

return bret;

}

int ObSqlTransControl::start_hook_if_need_(ObSQLSessionInfo &session,

transaction::ObTransService *txs,

bool &start_hook)

{

int ret = OB_SUCCESS;

if (!session.get_tx_desc()->is_shadow() && !session.has_start_stmt() &&

OB_SUCC(txs->sql_stmt_start_hook(session.get_xid(), *session.get_tx_desc(), session.get_sessid(), get_real_session_id(session)))) {

start_hook = true;

}

return ret;

}

int ObSqlTransControl::rollback_savepoint(ObExecContext &exec_ctx,

const ObString &sp_name)

{

int ret = OB_SUCCESS;

ObSQLSessionInfo *session = GET_MY_SESSION(exec_ctx);

const ObPhysicalPlanCtx *plan_ctx = GET_PHY_PLAN_CTX(exec_ctx);

transaction::ObTransService *txs = NULL;

int64_t stmt_expire_ts = 0;

CK (OB_NOT_NULL(session), OB_NOT_NULL(plan_ctx));

CHECK_SESSION (session);

CHECK_TXN_FREE_ROUTE_ALLOWED();

OZ (get_tx_service(session, txs));

OZ (acquire_tx_if_need_(txs, *session));

OX (stmt_expire_ts = get_stmt_expire_ts(plan_ctx, *session));

bool start_hook = false;

OZ(start_hook_if_need_(*session, txs, start_hook));

OZ (txs->rollback_to_explicit_savepoint(*session->get_tx_desc(), sp_name, stmt_expire_ts, get_real_session_id(*session)), sp_name);

if (start_hook) {

int tmp_ret = txs->sql_stmt_end_hook(session->get_xid(), *session->get_tx_desc());

if (OB_SUCCESS != tmp_ret) {

LOG_WARN("call sql stmt end hook fail", K(tmp_ret));

ret = COVER_SUCC(tmp_ret);

}

}

return ret;

}

int ObSqlTransControl::release_stash_savepoint(ObExecContext &exec_ctx,

const ObString &sp_name)

{

int ret = OB_SUCCESS;

ObSQLSessionInfo *session = GET_MY_SESSION(exec_ctx);

transaction::ObTransService *txs = NULL;

CK (OB_NOT_NULL(session));

// NOTE: should _NOT_ check session is zombie, because the stash savepoint

// should be release before query quit

// CHECK_SESSION (session);

OZ (get_tx_service(session, txs), *session);

OZ (acquire_tx_if_need_(txs, *session));

OZ (txs->release_explicit_savepoint(*session->get_tx_desc(), sp_name, get_real_session_id(*session)), *session, sp_name);

return ret;

}

int ObSqlTransControl::release_savepoint(ObExecContext &exec_ctx,

const ObString &sp_name)

{

int ret = OB_SUCCESS;

ObSQLSessionInfo *session = GET_MY_SESSION(exec_ctx);

transaction::ObTransService *txs = NULL;

CK (OB_NOT_NULL(session));

CHECK_SESSION (session);

CHECK_TXN_FREE_ROUTE_ALLOWED();

OZ (get_tx_service(session, txs), *session);

OZ (acquire_tx_if_need_(txs, *session));

bool start_hook = false;

OZ(start_hook_if_need_(*session, txs, start_hook));

OZ (txs->release_explicit_savepoint(*session->get_tx_desc(), sp_name, get_real_session_id(*session)), *session, sp_name);

if (start_hook) {

int tmp_ret = txs->sql_stmt_end_hook(session->get_xid(), *session->get_tx_desc());

if (OB_SUCCESS != tmp_ret) {

LOG_WARN("call sql stmt end hook fail", K(tmp_ret));

ret = COVER_SUCC(tmp_ret);

}

}

return ret;

}

int ObSqlTransControl::xa_rollback_all_changes(ObExecContext &exec_ctx)

{

int ret = OB_SUCCESS;

ObSQLSessionInfo *session = GET_MY_SESSION(exec_ctx);

const ObPhysicalPlanCtx *plan_ctx = GET_PHY_PLAN_CTX(exec_ctx);