/**

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

#include "sql/session/ob_sql_session_info.h"

#include "lib/trace/ob_trace_event.h"

#include "lib/alloc/alloc_func.h"

#include "lib/string/ob_sql_string.h"

#include "lib/rc/ob_rc.h"

#include "sql/ob_sql_utils.h"

#include "sql/ob_sql_trans_control.h"

#include "sql/session/ob_sql_session_mgr.h"

#include "io/easy_io.h"

#include "rpc/ob_rpc_define.h"

#include "observer/omt/ob_tenant_config_mgr.h"

#include "observer/ob_server_struct.h"

#include "pl/ob_pl.h"

#include "pl/ob_pl_package.h"

#include "pl/sys_package/ob_dbms_sql.h"

#include "observer/mysql/ob_mysql_request_manager.h"

#include "observer/mysql/obmp_stmt_send_piece_data.h"

#include "observer/mysql/ob_query_driver.h"

#include "observer/ob_server.h"

#include "share/rc/ob_context.h"

#include "share/rc/ob_tenant_base.h"

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

#include "sql/resolver/ddl/ob_ddl_stmt.h"

#include "observer/omt/ob_tenant_config_mgr.h"

#include "share/schema/ob_schema_struct.h"

#include "sql/resolver/ddl/ob_create_synonym_stmt.h"

#include "sql/resolver/ddl/ob_drop_synonym_stmt.h"

#include "sql/engine/expr/ob_datum_cast.h"

#include "lib/checksum/ob_crc64.h"

#include "lib/alloc/alloc_assist.h"

#include "lib/string/ob_string.h"

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

#include "lib/utility/utility.h"

#include "lib/utility/ob_proto_trans_util.h"

#ifdef OB_BUILD_ORACLE_PL

#include "pl/debug/ob_pl_debugger_manager.h"

#include "pl/sys_package/ob_pl_utl_file.h"

#endif

#include "lib/allocator/ob_mod_define.h"

#include "lib/string/ob_hex_utils_base.h"

#include "share/stat/ob_opt_stat_manager.h"

#include "sql/plan_cache/ob_ps_cache.h"

#include "observer/ob_sql_client_decorator.h"

#include "ob_sess_info_verify.h"

#include "share/schema/ob_schema_utils.h"

using namespace oceanbase::sql;

using namespace oceanbase::common;

using namespace oceanbase::share::schema;

using namespace oceanbase::share;

using namespace oceanbase::pl;

using namespace oceanbase::obmysql;

static const int64_t DEFAULT_XA_END_TIMEOUT_SECONDS = 60;/*60s*/

const char *state_str[] =

{

"INIT",

"SLEEP",

"ACTIVE",

"QUERY_KILLED",

"SESSION_KILLED",

};

void ObTenantCachedSchemaGuardInfo::reset()

{

schema_guard_.reset();

ref_ts_ = 0;

tenant_id_ = 0;

schema_version_ = 0;

}

int ObTenantCachedSchemaGuardInfo::refresh_tenant_schema_guard(const uint64_t tenant_id)

{

int ret = OB_SUCCESS;

if (OB_FAIL(OBSERVER.get_gctx().schema_service_->get_tenant_schema_guard(tenant_id, schema_guard_))) {

LOG_WARN("get schema guard failed", K(ret), K(tenant_id));

} else if (OB_FAIL(schema_guard_.get_schema_version(tenant_id, schema_version_))) {

LOG_WARN("fail get schema version", K(ret), K(tenant_id));

} else {

ref_ts_ = ObClockGenerator::getClock();

tenant_id_ = tenant_id;

}

return ret;

}

void ObTenantCachedSchemaGuardInfo::try_revert_schema_guard()

{

if (schema_guard_.is_inited()) {

const int64_t MAX_SCHEMA_GUARD_CACHED_TIME = 10 * 1000 * 1000;

if (ObClockGenerator::getClock() - ref_ts_ > MAX_SCHEMA_GUARD_CACHED_TIME) {

LOG_DEBUG("revert schema guard success by sql",

"session_id", schema_guard_.get_session_id(),

K_(tenant_id),

K_(schema_version));

reset();

}

}

}

ObSQLSessionInfo::ObSQLSessionInfo(const uint64_t tenant_id) :

ObVersionProvider(),

ObBasicSessionInfo(tenant_id),

is_inited_(false),

warnings_buf_(),

show_warnings_buf_(),

end_trans_cb_(),

user_priv_set_(),

db_priv_set_(),

curr_trans_start_time_(0),

curr_trans_last_stmt_time_(0),

sess_create_time_(0),

last_refresh_temp_table_time_(0),

has_temp_table_flag_(false),

has_accessed_session_level_temp_table_(false),

enable_early_lock_release_(false),

is_for_trigger_package_(false),

trans_type_(transaction::ObTxClass::USER),

version_provider_(NULL),

config_provider_(NULL),

request_manager_(NULL),

flt_span_mgr_(NULL),

plan_cache_(NULL),

ps_cache_(NULL),

found_rows_(1),

affected_rows_(-1),

global_sessid_(0),

read_uncommited_(false),

trace_recorder_(NULL),

inner_flag_(false),

is_max_availability_mode_(false),

next_client_ps_stmt_id_(0),

is_remote_session_(false),

session_type_(INVALID_TYPE),

curr_session_context_size_(0),

pl_context_(NULL),

pl_can_retry_(true),

#ifdef OB_BUILD_ORACLE_PL

pl_debugger_(NULL),

#endif

#ifdef OB_BUILD_SPM

select_plan_type_(ObSpmCacheCtx::INVALID_TYPE),

#endif

pl_attach_session_id_(0),

pl_query_sender_(NULL),

pl_ps_protocol_(false),

is_ob20_protocol_(false),

is_session_var_sync_(false),

pl_sync_pkg_vars_(NULL),

inner_conn_(NULL),

encrypt_info_(),

enable_role_array_(),

in_definer_named_proc_(false),

priv_user_id_(OB_INVALID_ID),

xa_end_timeout_seconds_(transaction::ObXADefault::OB_XA_TIMEOUT_SECONDS),

xa_last_result_(OB_SUCCESS),

cached_tenant_config_info_(this),

prelock_(false),

proxy_version_(0),

min_proxy_version_ps_(0),

is_ignore_stmt_(false),

ddl_info_(),

is_table_name_hidden_(false),

piece_cache_(NULL),

is_load_data_exec_session_(false),

pl_exact_err_msg_(),

is_varparams_sql_prepare_(false),

got_tenant_conn_res_(false),

got_user_conn_res_(false),

conn_res_user_id_(OB_INVALID_ID),

mem_context_(nullptr),

has_query_executed_(false),

is_latest_sess_info_(false),

cur_exec_ctx_(nullptr),

restore_auto_commit_(false),

dblink_context_(this),

sql_req_level_(0),

expect_group_id_(OB_INVALID_ID),

group_id_not_expected_(false),

gtt_session_scope_unique_id_(0),

gtt_trans_scope_unique_id_(0),

vid_(OB_INVALID_ID),

vport_(0),

in_bytes_(0),

out_bytes_(0),

current_dblink_sequence_id_(0),

client_non_standard_(false)

{

MEMSET(tenant_buff_, 0, sizeof(share::ObTenantSpaceFetcher));

MEMSET(vip_buf_, 0, sizeof(vip_buf_));

}

ObSQLSessionInfo::~ObSQLSessionInfo()

{

plan_cache_ = NULL;

destroy(false);

}

int ObSQLSessionInfo::init(uint32_t sessid, uint64_t proxy_sessid,

common::ObIAllocator *bucket_allocator, const ObTZInfoMap *tz_info, int64_t sess_create_time,

uint64_t tenant_id)

{

UNUSED(tenant_id);

int ret = OB_SUCCESS;

static const int64_t PS_BUCKET_NUM = 64;

if (OB_FAIL(ObBasicSessionInfo::init(sessid, proxy_sessid, bucket_allocator, tz_info))) {

LOG_WARN("fail to init basic session info", K(ret));

} else if (FALSE_IT(txn_free_route_ctx_.set_sessid(sessid))) {

} else if (!is_acquire_from_pool() &&

OB_FAIL(package_state_map_.create(hash::cal_next_prime(4),

ObMemAttr(orig_tenant_id_, "PackStateMap")))) {

LOG_WARN("create package state map failed", K(ret));

} else if (!is_acquire_from_pool() &&

OB_FAIL(sequence_currval_map_.create(hash::cal_next_prime(32),

ObMemAttr(orig_tenant_id_, "SequenceMap")))) {

LOG_WARN("create sequence current value map failed", K(ret));

} else if (!is_acquire_from_pool() &&

OB_FAIL(dblink_sequence_id_map_.create(hash::cal_next_prime(32),

ObMemAttr(orig_tenant_id_, "SequenceIdMap")))) {

LOG_WARN("create dblink sequence id map failed", K(ret));

} else if (!is_acquire_from_pool() &&

OB_FAIL(contexts_map_.create(hash::cal_next_prime(32),

ObMemAttr(orig_tenant_id_, "ContextsMap")))) {

LOG_WARN("create contexts map failed", K(ret));

} else {

curr_session_context_size_ = 0;

if (is_obproxy_mode()) {

sess_create_time_ = sess_create_time;

} else {

sess_create_time_ = ObTimeUtility::current_time();

}

const char *sup_proxy_min_version = "1.8.4";

min_proxy_version_ps_ = 0;

if (OB_FAIL(ObClusterVersion::get_version(sup_proxy_min_version, min_proxy_version_ps_))) {

LOG_WARN("failed to get version", K(ret));

} else {

is_inited_ = true;

refresh_temp_tables_sess_active_time();

}

}

return ret;

}

//for test

int ObSQLSessionInfo::test_init(uint32_t version, uint32_t sessid, uint64_t proxy_sessid,

common::ObIAllocator *bucket_allocator)

{

int ret = OB_SUCCESS;

UNUSED(version);

if (OB_FAIL(ObBasicSessionInfo::test_init(sessid, proxy_sessid, bucket_allocator))) {

LOG_WARN("fail to init basic session info", K(ret));

} else if (FALSE_IT(txn_free_route_ctx_.set_sessid(sessid))) {

} else {

is_inited_ = true;

}

return ret;

}

void ObSQLSessionInfo::reset(bool skip_sys_var)

{

if (is_inited_) {

// ObVersionProvider::reset();

reset_all_package_changed_info();

warnings_buf_.reset();

show_warnings_buf_.reset();

end_trans_cb_.reset(),

audit_record_.reset();

user_priv_set_ = 0;

db_priv_set_ = 0;

curr_trans_start_time_ = 0;

curr_trans_last_stmt_time_ = 0;

sess_create_time_ = 0;

last_refresh_temp_table_time_ = 0;

has_temp_table_flag_ = false;

has_accessed_session_level_temp_table_ = false;

is_for_trigger_package_ = false;

trans_type_ = transaction::ObTxClass::USER;

version_provider_ = NULL;

config_provider_ = NULL;

request_manager_ = NULL;

flt_span_mgr_ = NULL;

MEMSET(tenant_buff_, 0, sizeof(share::ObTenantSpaceFetcher));

ps_cache_ = NULL;

found_rows_ = 1;

affected_rows_ = -1;

global_sessid_ = 0;

read_uncommited_ = false;

trace_recorder_ = NULL;

inner_flag_ = false;

is_max_availability_mode_ = false;

enable_early_lock_release_ = false;

ps_session_info_map_.reuse();

ps_name_id_map_.reuse();

next_client_ps_stmt_id_ = 0;

is_remote_session_ = false;

session_type_ = INVALID_TYPE;

package_state_map_.reuse();

sequence_currval_map_.reuse();

dblink_sequence_id_map_.reuse();

curr_session_context_size_ = 0;

pl_context_ = NULL;

pl_can_retry_ = true;

#ifdef OB_BUILD_ORACLE_PL

pl_debugger_ = NULL;

#endif

pl_attach_session_id_ = 0;

pl_query_sender_ = NULL;

pl_ps_protocol_ = false;

if (pl_cursor_cache_.is_inited()) {

// when select GV$OPEN_CURSOR, we will add get_thread_data_lock to fetch pl_cursor_map_

// so we need get_thread_data_lock there

ObSQLSessionInfo::LockGuard lock_guard(get_thread_data_lock());

pl_cursor_cache_.reset();

}

inner_conn_ = NULL;

session_stat_.reset();

pl_sync_pkg_vars_ = NULL;

//encrypt_info_.reset();

cached_schema_guard_info_.reset();

encrypt_info_.reset();

enable_role_array_.reset();

in_definer_named_proc_ = false;

priv_user_id_ = OB_INVALID_ID;

xa_end_timeout_seconds_ = transaction::ObXADefault::OB_XA_TIMEOUT_SECONDS;

xa_last_result_ = OB_SUCCESS;

prelock_ = false;

proxy_version_ = 0;

min_proxy_version_ps_ = 0;

if (OB_NOT_NULL(mem_context_)) {

destroy_contexts_map(contexts_map_, mem_context_->get_malloc_allocator());

DESTROY_CONTEXT(mem_context_);

mem_context_ = NULL;

}

contexts_map_.reuse();

cur_exec_ctx_ = nullptr;

plan_cache_ = NULL;

client_app_info_.reset();

has_query_executed_ = false;

flt_control_info_.reset();

is_send_control_info_ = false;

trace_enable_ = false;

auto_flush_trace_ = false;

coninfo_set_by_sess_ = false;

is_ob20_protocol_ = false;

is_session_var_sync_ = false;

is_latest_sess_info_ = false;

int temp_ret = OB_SUCCESS;

sql_req_level_ = 0;

optimizer_tracer_.reset();

expect_group_id_ = OB_INVALID_ID;

flt_control_info_.reset();

group_id_not_expected_ = false;

//call at last time

dblink_context_.reset(); // need reset before ObBasicSessionInfo::reset(skip_sys_var);

ObBasicSessionInfo::reset(skip_sys_var);

txn_free_route_ctx_.reset();

client_non_standard_ = false;

}

gtt_session_scope_unique_id_ = 0;

gtt_trans_scope_unique_id_ = 0;

gtt_session_scope_ids_.reset();

gtt_trans_scope_ids_.reset();

vid_ = OB_INVALID_ID;

vport_ = 0;

in_bytes_ = 0;

out_bytes_ = 0;

MEMSET(vip_buf_, 0, sizeof(vip_buf_));

current_dblink_sequence_id_ = 0;

dblink_sequence_schemas_.reset();

}

void ObSQLSessionInfo::clean_status()

{

reset_all_package_changed_info();

ObBasicSessionInfo::clean_status();

}

bool ObSQLSessionInfo::is_encrypt_tenant()

{

bool ret = false;

#ifdef OB_BUILD_TDE_SECURITY

uint64_t cur_time = ObClockGenerator::getClock();

int64_t tenant_id = get_effective_tenant_id();

if (cur_time - encrypt_info_.last_modify_time_ > 10 * 1000 * 1000L) {

if (OB_UNLIKELY(OB_INVALID_TENANT_ID == tenant_id)) {

LOG_WARN("Invalid tenant id to init fast freeze checker", K(tenant_id));

} else {

encrypt_info_.last_modify_time_ = cur_time;

omt::ObTenantConfigGuard tenant_config(TENANT_CONF(tenant_id));

if (tenant_config.is_valid()) {

ObString method_str(tenant_config->tde_method.get_value());

if (ObTdeMethodUtil::is_kms(method_str)) {

encrypt_info_.is_encrypt_ = true;

ret = true;

} else {

encrypt_info_.is_encrypt_ = false;

ret = false;

}

}

}

} else {

ret = encrypt_info_.is_encrypt_;

}

#endif

return ret;

}

int ObSQLSessionInfo::is_force_temp_table_inline(bool &force_inline) const

{

int ret = OB_SUCCESS;

int64_t with_subquery_policy = 0;

force_inline = false;

int64_t tenant_id = get_effective_tenant_id();

omt::ObTenantConfigGuard tenant_config(TENANT_CONF(tenant_id));

if (tenant_config.is_valid()) {

int64_t with_subquery_policy = tenant_config->_with_subquery;

if (2 == with_subquery_policy) {

force_inline = true;

}

}

return ret;

}

int ObSQLSessionInfo::is_force_temp_table_materialize(bool &force_materialize) const

{

int ret = OB_SUCCESS;

int64_t with_subquery_policy = 0;

force_materialize = false;

int64_t tenant_id = get_effective_tenant_id();

omt::ObTenantConfigGuard tenant_config(TENANT_CONF(tenant_id));

if (tenant_config.is_valid()) {

int64_t with_subquery_policy = tenant_config->_with_subquery;

if (1 == with_subquery_policy) {

force_materialize = true;

}

}

return ret;

}

int ObSQLSessionInfo::is_temp_table_transformation_enabled(bool &transformation_enabled) const

{

int ret = OB_SUCCESS;

transformation_enabled = false;

int64_t tenant_id = get_effective_tenant_id();

omt::ObTenantConfigGuard tenant_config(TENANT_CONF(tenant_id));

if (tenant_config.is_valid()) {

transformation_enabled = tenant_config->_xsolapi_generate_with_clause;

}

return ret;

}

int ObSQLSessionInfo::is_groupby_placement_transformation_enabled(bool &transformation_enabled) const

{

int ret = OB_SUCCESS;

transformation_enabled = false;

int64_t tenant_id = get_effective_tenant_id();

omt::ObTenantConfigGuard tenant_config(TENANT_CONF(tenant_id));

if (tenant_config.is_valid()) {

transformation_enabled = tenant_config->_optimizer_group_by_placement;

}

return ret;

}

bool ObSQLSessionInfo::is_in_range_optimization_enabled() const

{

bool bret = false;

int64_t tenant_id = get_effective_tenant_id();

omt::ObTenantConfigGuard tenant_config(TENANT_CONF(tenant_id));

if (tenant_config.is_valid()) {

bret = tenant_config->_enable_in_range_optimization;

}

return bret;

}

int ObSQLSessionInfo::is_better_inlist_enabled(bool &enabled) const

{

int ret = OB_SUCCESS;

enabled = false;

int64_t tenant_id = get_effective_tenant_id();

omt::ObTenantConfigGuard tenant_config(TENANT_CONF(tenant_id));

if (tenant_config.is_valid()) {

enabled = tenant_config->_optimizer_better_inlist_costing;

}

return ret;

}

bool ObSQLSessionInfo::is_index_skip_scan_enabled() const

{

bool bret = false;

int64_t tenant_id = get_effective_tenant_id();

omt::ObTenantConfigGuard tenant_config(TENANT_CONF(tenant_id));

if (tenant_config.is_valid()) {

bret = tenant_config->_optimizer_skip_scan_enabled;

}

return bret;

}

int ObSQLSessionInfo::is_enable_range_extraction_for_not_in(bool &enabled) const

{

int ret = OB_SUCCESS;

enabled = false;

int64_t tenant_id = get_effective_tenant_id();

omt::ObTenantConfigGuard tenant_config(TENANT_CONF(tenant_id));

if (tenant_config.is_valid()) {

enabled = tenant_config->_enable_range_extraction_for_not_in;

}

return ret;

}

bool ObSQLSessionInfo::is_var_assign_use_das_enabled() const

{

bool bret = false;

int64_t tenant_id = get_effective_tenant_id();

omt::ObTenantConfigGuard tenant_config(TENANT_CONF(tenant_id));

if (tenant_config.is_valid()) {

bret = tenant_config->_enable_var_assign_use_das;

}

return bret;

}

int ObSQLSessionInfo::is_adj_index_cost_enabled(bool &enabled, int64_t &stats_cost_percent) const

{

int ret = OB_SUCCESS;

enabled = false;

stats_cost_percent = 0;

int64_t tenant_id = get_effective_tenant_id();

omt::ObTenantConfigGuard tenant_config(TENANT_CONF(tenant_id));

if (tenant_config.is_valid()) {

stats_cost_percent = tenant_config->optimizer_index_cost_adj;

enabled = (0 != stats_cost_percent);

}

return ret;

}

void ObSQLSessionInfo::destroy(bool skip_sys_var)

{

if (is_inited_) {

int ret = OB_SUCCESS;

if (rpc::is_io_thread()) {

LOG_WARN("free session at IO thread", "sessid", get_sessid(), "proxy_sessid", get_proxy_sessid());

//事务层需要保持在end trans时，既没有阻塞操作，又没有rpc调用。否则会影响server IO性能或产生死锁

}

// 反序列化出来的 session 不应该做 end_trans 等清理工作

// bug:

if (false == get_is_deserialized()) {

if (false == ObSchemaService::g_liboblog_mode_) {

//session断开时调用ObTransService::end_trans回滚事务，

// 此处stmt_timeout ＝ 当前时间 ＋语句query超时时间,而不是最后一条sql的start_time, 相关bug_id : 7961445

set_query_start_time(ObTimeUtility::current_time());

// 这里调用end_trans无需上锁，因为调用reclaim_value时意味着已经没有query并发使用session

// 调用这个函数之前会调session.set_session_state(SESSION_KILLED)，

bool need_disconnect = false;

if (is_in_transaction() && !is_txn_free_route_temp()) {

transaction::ObTransID tx_id = get_tx_id();

MAKE_TENANT_SWITCH_SCOPE_GUARD(guard);

// inner session skip check switch tenant, because the inner connection was shared between tenant

if (OB_SUCC(guard.switch_to(get_effective_tenant_id(), !is_inner()))) {

if (OB_FAIL(ObSqlTransControl::rollback_trans(this, need_disconnect))) {

LOG_WARN("fail to rollback transaction", K(get_sessid()),

"proxy_sessid", get_proxy_sessid(), K(ret));

} else if (false == inner_flag_ && false == is_remote_session_) {

LOG_INFO("end trans successfully",

"sessid", get_sessid(),

"proxy_sessid", get_proxy_sessid(),

"trans id", tx_id);

}

} else {

LOG_WARN("fail to switch tenant", K(get_effective_tenant_id()), K(ret));

}

}

}

}

// 临时表在 slave session 析构时不能清理

if (false == get_is_deserialized()) {

int temp_ret = drop_temp_tables();

if (OB_UNLIKELY(OB_SUCCESS != temp_ret)) {

LOG_WARN("fail to drop temp tables", K(temp_ret));

}

refresh_temp_tables_sess_active_time();

}

// slave session 上 ps_session_info_map_ 为空，调用 close 也不会有副作用

if (OB_SUCC(ret)) {

if (OB_FAIL(close_all_ps_stmt())) {

LOG_WARN("failed to close all stmt", K(ret));

}

}

//close all cursor

if (OB_SUCC(ret) && pl_cursor_cache_.is_inited()) {

if (OB_FAIL(pl_cursor_cache_.close_all(*this))) {

LOG_WARN("failed to close all cursor", K(ret));

}

}

if (OB_SUCC(ret) && NULL != piece_cache_) {

if (OB_FAIL((static_cast<observer::ObPieceCache*>(piece_cache_))

->close_all(*this))) {

LOG_WARN("failed to close all piece", K(ret));

}

static_cast<observer::ObPieceCache*>(piece_cache_)->~ObPieceCache();

get_session_allocator().free(piece_cache_);

piece_cache_ = NULL;

}

#ifdef OB_BUILD_ORACLE_PL

if (OB_SUCC(ret)) {

const int64_t session_id = get_sessid();

// utl file should close all fd when user session exits,

// so we should check session type here to avoid fd closing

// unexpectedly when inner session exists

if (is_user_session() && OB_FAIL(ObPLUtlFile::close_all(session_id))) {

LOG_WARN("failed to close all fd in utl file", K(ret), K(session_id));

}

}

#endif

#ifdef OB_BUILD_ORACLE_PL

// pl debug 功能, pl debug不支持分布式调试，但调用也不会有副作用

reset_pl_debugger_resource();

#endif

// 非分布式需要的话，分布式也需要，用于清理package的全局变量值

reset_all_package_state();

reset(skip_sys_var);

is_inited_ = false;

sql_req_level_ = 0;

}

}

int ObSQLSessionInfo::close_ps_stmt(ObPsStmtId client_stmt_id)

{

int ret = OB_SUCCESS;

ObPsSessionInfo *ps_sess_info = NULL;

if (OB_FAIL(get_ps_session_info(client_stmt_id, ps_sess_info))) {

LOG_WARN("fail to get ps session info", K(client_stmt_id), "session_id", get_sessid(), K(ret));

} else if (OB_ISNULL(ps_sess_info)) {

ret = OB_INVALID_ARGUMENT;

LOG_WARN("ps session info is null", K(client_stmt_id), "session_id", get_sessid(), K(ret));

} else {

ObPsStmtId inner_stmt_id = ps_sess_info->get_inner_stmt_id();

ps_sess_info->dec_ref_count();

if (ps_sess_info->need_erase()) {

if (OB_ISNULL(ps_cache_)) {

ret = OB_INVALID_ARGUMENT;

LOG_WARN("ps cache is null", K(ret));

} else if (OB_FAIL(ps_cache_->deref_ps_stmt(inner_stmt_id))) {

LOG_WARN("close ps stmt failed", K(ret), "session_id", get_sessid(), K(ret));

}

//无论上面是否成功, 都需要将session info资源释放

int tmp_ret = OB_SUCCESS;

if (OB_SUCCESS != (tmp_ret = remove_ps_session_info(client_stmt_id))) {

ret = tmp_ret;

LOG_WARN("remove ps session info failed", K(client_stmt_id),

"session_id", get_sessid(), K(ret));

}

LOG_TRACE("close ps stmt", K(ret), K(client_stmt_id), K(inner_stmt_id), K(lbt()));

}

}

return ret;

}

int ObSQLSessionInfo::close_all_ps_stmt()

{

int ret = OB_SUCCESS;

if (OB_ISNULL(ps_cache_)) {

// do nothing, session no ps

} else if (!ps_session_info_map_.created()) {

// do nothing, no ps added to map

} else {

PsSessionInfoMap::iterator iter = ps_session_info_map_.begin();

ObPsStmtId inner_stmt_id = OB_INVALID_ID;

for (; iter != ps_session_info_map_.end(); ++iter) { //ignore ret

const ObPsStmtId client_stmt_id = iter->first;

if (OB_FAIL(get_inner_ps_stmt_id(client_stmt_id, inner_stmt_id))) {

LOG_WARN("get_inner_ps_stmt_id failed", K(ret), K(client_stmt_id), K(inner_stmt_id));

} else if (OB_FAIL(ps_cache_->deref_ps_stmt(inner_stmt_id))) {

LOG_WARN("close ps stmt failed", K(ret), K(client_stmt_id), K(inner_stmt_id));

} else if (OB_ISNULL(iter->second)) {

// do nothing

} else {

iter->second->~ObPsSessionInfo();

ps_session_info_allocator_.free(iter->second);

iter->second = NULL;

}

}

ps_session_info_allocator_.reset();

ps_session_info_map_.reuse();

}

return ret;

}

//用于oracle临时表数据的清理, 在session断开(会话级&事务级)和commit时(事务级)调用

int ObSQLSessionInfo::delete_from_oracle_temp_tables(const obrpc::ObDropTableArg &const_drop_table_arg)

{

int ret = OB_SUCCESS;

common::ObSqlString sql;

common::ObMySQLProxy *sql_proxy = GCTX.sql_proxy_;

common::ObCommonSqlProxy *user_sql_proxy;

common::ObOracleSqlProxy oracle_sql_proxy;

ObSchemaGetterGuard schema_guard;

const ObDatabaseSchema *database_schema = NULL;

//ObSEArray<const ObSimpleTableSchemaV2 *, 512> table_schemas;

obrpc::ObDropTableArg &drop_table_arg = const_cast<obrpc::ObDropTableArg &>(const_drop_table_arg);

const share::schema::ObTableType table_type = drop_table_arg.table_type_;

const uint64_t tenant_id = drop_table_arg.tenant_id_;

const ObTableSchema *table_schema = NULL;

user_sql_proxy = &oracle_sql_proxy;

if (OB_FAIL(GCTX.schema_service_->get_tenant_schema_guard(

tenant_id,

schema_guard))) {

LOG_WARN("get schema guard failed.", K(ret), K(tenant_id));

} else if (OB_ISNULL(sql_proxy)) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("sql proxy is null", K(ret));

} else if (OB_FAIL(oracle_sql_proxy.init(sql_proxy->get_pool()))) {

LOG_WARN("init oracle sql proxy failed", K(ret));

} else if (TMP_TABLE_ORA_SESS == table_type || TMP_TABLE_ORA_TRX == table_type) {

ObIArray<uint64_t> &table_ids = table_type == share::schema::TMP_TABLE_ORA_TRX ?

get_gtt_trans_scope_ids() : get_gtt_session_scope_ids();

uint64_t unique_id = table_type == share::schema::TMP_TABLE_ORA_TRX ?

get_gtt_trans_scope_unique_id() : get_gtt_session_scope_unique_id();

LOG_DEBUG("delete temp table", K(table_ids), K(unique_id));

for (int64_t i = 0; OB_SUCC(ret) && i < table_ids.count(); i++) {

if (OB_FAIL(schema_guard.get_table_schema(tenant_id, table_ids.at(i), table_schema))) {

LOG_WARN("fail to get table schema", K(ret));

} else if (OB_ISNULL(table_schema)) {

//table may be dropped, ignore

} else if (tenant_id != table_schema->get_tenant_id()) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("tenant_id not match", K(ret), K(tenant_id), "table_id", table_schema->get_table_id());

} else if (((TMP_TABLE_ORA_SESS == table_type && table_schema->is_oracle_tmp_table())

|| (TMP_TABLE_ORA_TRX == table_type && table_schema->is_oracle_trx_tmp_table()))

&& table_schema->is_normal_schema()) {

database_schema = NULL;

if (OB_FAIL(schema_guard.get_database_schema(table_schema->get_tenant_id(),

table_schema->get_database_id(), database_schema))) {

LOG_WARN("failed to get database schema", K(ret));

} else if (OB_ISNULL(database_schema)) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("database schema is null", K(ret));

} else if (database_schema->is_in_recyclebin() || table_schema->is_in_recyclebin()) {

LOG_DEBUG("skip table schema in recyclebin", K(*table_schema));

} else {

const int64_t limit = 1000;

ret = sql.assign_fmt("DELETE FROM \"%.*s\".\"%.*s\" WHERE %s = %ld AND ROWNUM <= %ld",

database_schema->get_database_name_str().length(),

database_schema->get_database_name_str().ptr(),

table_schema->get_table_name_str().length(),

table_schema->get_table_name_str().ptr(),

OB_HIDDEN_SESSION_ID_COLUMN_NAME, unique_id,

limit);

if (OB_SUCC(ret)) {

int64_t affect_rows = 0;

int64_t last_batch_affect_rows = limit;

int64_t cur_time = ObTimeUtility::current_time();

int64_t cur_timeout_backup = THIS_WORKER.get_timeout_ts();

THIS_WORKER.set_timeout_ts(ObTimeUtility::current_time() + OB_MAX_USER_SPECIFIED_TIMEOUT);

while (OB_SUCC(ret) && last_batch_affect_rows > 0) {

if (OB_FAIL(user_sql_proxy->write(tenant_id, sql.ptr(), last_batch_affect_rows))) {

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

} else {

affect_rows += last_batch_affect_rows;

}

}

if (OB_SUCC(ret)) {

LOG_DEBUG("succeed to delete rows in oracle temporary table", K(sql), K(affect_rows));

//delete relation temp table stats.

if (OB_FAIL(ObOptStatManager::get_instance().delete_table_stat(tenant_id,

table_schema->get_table_id(), affect_rows))) {

LOG_WARN("failed to delete table stats", K(ret));

}

} else {

LOG_WARN("failed to delete rows in oracle temporary table", K(ret), K(sql));

}

LOG_INFO("delete rows in oracle temporary table", K(sql), K(affect_rows),

"clean_time", ObTimeUtility::current_time() - cur_time);

THIS_WORKER.set_timeout_ts(cur_timeout_backup);

}

}

}

}

if (TMP_TABLE_ORA_TRX == table_type && !get_is_deserialized()) {

gtt_trans_scope_ids_.reuse();

gen_gtt_trans_scope_unique_id();

if (gtt_session_scope_ids_.count() == 0) {

if (OB_FAIL(set_session_temp_table_used(false))) {

LOG_WARN("fail to set session temp table unused", K(ret));

}

}

}

}

return ret;

}

//mysql租户: 如果session创建过临时表, 直连模式: session断开时drop temp table;

//oracle租户, commit时为了清空数据也会调用此接口, 但仅清除事务级别的临时表;

// session断开时则清理掉事务级和会话级的临时表;

//由于oracle临时表仅仅是清理本session数据, 为避免rs拥塞,不发往rs由sql proxy执行

//对于分布式计划, 除非ac=1否则交给master session清理, 反序列化得到的session不做事情

int ObSQLSessionInfo::drop_temp_tables(const bool is_disconn, const bool is_xa_trans)

{

int ret = OB_SUCCESS;

bool ac = false;

bool is_sess_disconn = is_disconn;

obrpc::ObCommonRpcProxy *common_rpc_proxy = NULL;

if (OB_FAIL(get_autocommit(ac))) {

LOG_WARN("get autocommit error", K(ret), K(ac));

} else if (!(is_inner() && !is_user_session())

&& (get_has_temp_table_flag()

|| has_accessed_session_level_temp_table()

|| has_tx_level_temp_table()

|| is_xa_trans)

&& (!get_is_deserialized() || ac)) {

bool need_drop_temp_table = false;

//mysql: 仅直连 & sess 断开时

//oracle: commit; 或者 直连时的断session;

if (!is_oracle_mode()) {

if (false == is_obproxy_mode() && is_sess_disconn) {

need_drop_temp_table = true;

}

} else {

if (false == is_sess_disconn || false == is_obproxy_mode()) {

need_drop_temp_table = true;

//ac=1, 反序列化session断开时只是任务结束, 并不是真的sess断开, 视作trx commit

if (is_sess_disconn && get_is_deserialized() && ac) {

is_sess_disconn = false;

}

}

}

if (need_drop_temp_table) {

LOG_DEBUG("need_drop_temp_table",

K(is_oracle_mode()),

K(get_current_query_string()),

K(get_login_tenant_id()),

K(get_effective_tenant_id()),

K(lbt()));

obrpc::ObDDLRes res;

obrpc::ObDropTableArg drop_table_arg;

drop_table_arg.if_exist_ = true;

drop_table_arg.to_recyclebin_ = false;

if (false == is_sess_disconn) {

drop_table_arg.table_type_ = share::schema::TMP_TABLE_ORA_TRX;

} else if (is_oracle_mode()) {

drop_table_arg.table_type_ = share::schema::TMP_TABLE_ORA_SESS;

} else {

drop_table_arg.table_type_ = share::schema::TMP_TABLE;

}

drop_table_arg.session_id_ = get_sessid_for_table();

drop_table_arg.tenant_id_ = get_effective_tenant_id();

drop_table_arg.exec_tenant_id_ = get_effective_tenant_id();

common_rpc_proxy = GCTX.rs_rpc_proxy_;

if (OB_ISNULL(common_rpc_proxy)) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("rpc proxy is null", K(ret));

} else if (OB_FAIL(delete_from_oracle_temp_tables(drop_table_arg))) {

LOG_WARN("failed to delete from oracle temporary table", K(drop_table_arg), K(ret));

}/* else if (!is_oracle_mode() && OB_FALSE_IT(drop_table_arg.compat_mode_ = lib::Worker::CompatMode::MYSQL)) {

} else if (!is_oracle_mode() && OB_FAIL(common_rpc_proxy->drop_table(drop_table_arg, res))) {

LOG_WARN("failed to drop temporary table", K(drop_table_arg), K(ret));

}*/ else {

LOG_INFO("temporary tables dropped due to connection disconnected", K(is_sess_disconn), K(drop_table_arg));

}

}

}

if (OB_FAIL(ret)) {

LOG_WARN("fail to drop temp tables", K(ret),

K(get_effective_tenant_id()), K(get_sessid()),

K(has_accessed_session_level_temp_table()),

K(is_xa_trans),

K(lbt()));

}

return ret;

}

//清理oracle临时表中数据来源和当前session id相同, 但属于被重用的旧的session数据

int ObSQLSessionInfo::drop_reused_oracle_temp_tables()

{

int ret = OB_SUCCESS;

//obrpc::ObCommonRpcProxy *common_rpc_proxy = NULL;

if (false == get_is_deserialized()

&& !is_inner()

&& !GCTX.is_standby_cluster()) {

obrpc::ObDropTableArg drop_table_arg;

drop_table_arg.if_exist_ = true;

drop_table_arg.to_recyclebin_ = false;

drop_table_arg.table_type_ = share::schema::TMP_TABLE_ORA_SESS;

drop_table_arg.session_id_ = get_sessid_for_table();

drop_table_arg.tenant_id_ = get_effective_tenant_id();

drop_table_arg.sess_create_time_ = get_sess_create_time();

//common_rpc_proxy = GCTX.rs_rpc_proxy_;

if (OB_FAIL(delete_from_oracle_temp_tables(drop_table_arg))) {

//if (OB_FAIL(common_rpc_proxy->drop_table(drop_table_arg))) {

LOG_WARN("failed to drop reused temporary table", K(drop_table_arg), K(ret));

} else {

LOG_DEBUG("succeed to delete old rows for oracle temporary table", K(drop_table_arg));

}

}

return ret;

}

//proxy方式下session创建、断开和后台定时task检查:

//如果距离上次更新此session->last_refresh_temp_table_time_ 超过1hr

//则更新session创建的临时表最后活动时间SESSION_ACTIVE_TIME

//oracle临时表依赖附加的__sess_create_time判断重用并清理, 不需要更新

void ObSQLSessionInfo::refresh_temp_tables_sess_active_time()

{

int ret = OB_SUCCESS;

const int64_t REFRESH_INTERVAL = 60L * 60L * 1000L * 1000L; // 1hr

obrpc::ObCommonRpcProxy *common_rpc_proxy = NULL;

if (get_has_temp_table_flag() && is_obproxy_mode()

&& !is_oracle_mode()) {

int64_t now = ObTimeUtility::current_time();

obrpc::ObAlterTableRes res;

if (now - get_last_refresh_temp_table_time() >= REFRESH_INTERVAL) {

SMART_VAR(obrpc::ObAlterTableArg, alter_table_arg) {

AlterTableSchema *alter_table_schema = &alter_table_arg.alter_table_schema_;

alter_table_arg.session_id_ = get_sessid_for_table();

alter_table_schema->alter_type_ = OB_DDL_ALTER_TABLE;

common_rpc_proxy = GCTX.rs_rpc_proxy_;

alter_table_arg.nls_formats_[ObNLSFormatEnum::NLS_DATE] = ObTimeConverter::COMPAT_OLD_NLS_DATE_FORMAT;

alter_table_arg.nls_formats_[ObNLSFormatEnum::NLS_TIMESTAMP] = ObTimeConverter::COMPAT_OLD_NLS_TIMESTAMP_FORMAT;

alter_table_arg.nls_formats_[ObNLSFormatEnum::NLS_TIMESTAMP_TZ] = ObTimeConverter::COMPAT_OLD_NLS_TIMESTAMP_TZ_FORMAT;

alter_table_arg.compat_mode_ = lib::Worker::CompatMode::MYSQL;

if (OB_FAIL(alter_table_schema->alter_option_bitset_.add_member(obrpc::ObAlterTableArg::SESSION_ACTIVE_TIME))) {

LOG_WARN("failed to add member SESSION_ACTIVE_TIME for alter table schema", K(ret));

} else if (OB_FAIL(alter_table_arg.tz_info_wrap_.deep_copy(get_tz_info_wrap()))) {

LOG_WARN("failed to deep copy tz_info_wrap", K(ret));

} else if (OB_FAIL(common_rpc_proxy->alter_table(alter_table_arg, res))) {

LOG_WARN("failed to alter temporary table session active time", K(alter_table_arg), K(ret), K(is_obproxy_mode()));

} else {

LOG_DEBUG("session active time of temporary tables refreshed", K(ret), "last refresh time", get_last_refresh_temp_table_time());

set_last_refresh_temp_table_time(now);

}

}

} else {

LOG_DEBUG("no need to refresh session active time of temporary tables", "last refresh time", get_last_refresh_temp_table_time());

}

}

}

ObMySQLRequestManager* ObSQLSessionInfo::get_request_manager()

{

int ret = OB_SUCCESS;

if (NULL == request_manager_) {

MTL_SWITCH(get_effective_tenant_id()) {

request_manager_ = MTL(obmysql::ObMySQLRequestManager*);

}

}

return request_manager_;

}

ObFLTSpanMgr* ObSQLSessionInfo::get_flt_span_manager()

{

int ret = OB_SUCCESS;

if (NULL == flt_span_mgr_) {

MTL_SWITCH(get_priv_tenant_id()) {

flt_span_mgr_ = MTL(sql::ObFLTSpanMgr*);

}

}

return flt_span_mgr_;

}

void ObSQLSessionInfo::set_show_warnings_buf(int error_code)

{

// if error message didn't insert into THREAD warning buffer,

// insert it into SESSION warning buffer

// if no error at all,

// clear err.

if (OB_SUCCESS != error_code && strlen(warnings_buf_.get_err_msg()) <= 0) {

warnings_buf_.set_error(ob_errpkt_strerror(error_code, lib::is_oracle_mode()), error_code);

warnings_buf_.reset_warning();

} else if (OB_SUCCESS == error_code) {

warnings_buf_.reset_err();

}

show_warnings_buf_ = warnings_buf_; // show_warnings_buf_用于show warnings

}