/**

* 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_basic_session_info.h"

#include "lib/string/ob_sql_string.h"

#include "lib/timezone/ob_oracle_format_models.h"

#include "common/sql_mode/ob_sql_mode_utils.h"

#include "common/ob_smart_call.h"

#include "share/config/ob_server_config.h"

#include "share/schema/ob_schema_getter_guard.h"

#include "share/schema/ob_sys_variable_mgr.h"

#include "sql/engine/ob_exec_context.h"

#include "sql/plan_cache/ob_plan_cache_util.h"

#include "sql/plan_cache/ob_prepare_stmt_struct.h"

#include "sql/ob_sql_trans_util.h"

#include "observer/ob_server_struct.h"

#include "sql/engine/ob_physical_plan.h"

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

#include "storage/tx/ob_trans_service.h"

#include "observer/omt/ob_tenant_timezone_mgr.h"

#include "observer/omt/ob_tenant_config_mgr.h"

#include "share/ob_label_security.h"

#include "share/system_variable/ob_nls_system_variable.h"

#include "share/rc/ob_tenant_base.h"

#include "pl/sys_package/ob_dbms_sql.h"

#include "pl/ob_pl_package_state.h"

#include "rpc/obmysql/ob_sql_sock_session.h"

using namespace oceanbase::common;

using namespace oceanbase::share;

using namespace oceanbase::share::schema;

using namespace oceanbase::transaction;

using oceanbase::omt::ObTenantConfigMgr;

namespace oceanbase

{

namespace sql

{

ObBasicSessionInfo::SysVarsCacheData ObBasicSessionInfo::SysVarsCache::base_data_;

ObBasicSessionInfo::ObBasicSessionInfo(const uint64_t tenant_id)

: orig_tenant_id_(tenant_id),

tenant_session_mgr_(NULL),

query_mutex_(common::ObLatchIds::SESSION_QUERY_LOCK),

thread_data_mutex_(common::ObLatchIds::SESSION_THREAD_DATA_LOCK),

is_valid_(true),

is_deserialized_(false),

tenant_id_(OB_INVALID_ID),

effective_tenant_id_(OB_INVALID_ID),

rpc_tenant_id_(0),

is_changed_to_temp_tenant_(false),

user_id_(OB_INVALID_ID),

client_version_(),

driver_version_(),

sessid_(0),

master_sessid_(INVALID_SESSID),

proxy_sessid_(VALID_PROXY_SESSID),

global_vars_version_(0),

sys_var_base_version_(OB_INVALID_VERSION),

tx_desc_(NULL),

tx_result_(),

reserved_read_snapshot_version_(),

xid_(),

associated_xa_(false),

cached_tenant_config_version_(0),

sess_bt_buff_pos_(0),

sess_ref_cnt_(0),

sess_ref_seq_(0),

block_allocator_(SMALL_BLOCK_SIZE, common::OB_MALLOC_NORMAL_BLOCK_SIZE - 32,

//这里减32是为了适配ObMalloc对齐规则, 防止超8k的内存分配

ObMalloc(lib::ObMemAttr(orig_tenant_id_, ObModIds::OB_SQL_SESSION_SBLOCK))),

ps_session_info_allocator_(sizeof(ObPsSessionInfo), common::OB_MALLOC_NORMAL_BLOCK_SIZE - 32,

//这里减32是为了适配ObMalloc对齐规则, 防止超8k的内存分配

ObMalloc(lib::ObMemAttr(orig_tenant_id_, "PsSessionInfo"))),

cursor_info_allocator_(sizeof(pl::ObDbmsCursorInfo), common::OB_MALLOC_NORMAL_BLOCK_SIZE - 32,

ObMalloc(lib::ObMemAttr(orig_tenant_id_, "SessCursorInfo"))),

package_info_allocator_(sizeof(pl::ObPLPackageState), common::OB_MALLOC_NORMAL_BLOCK_SIZE - 32,

ObMalloc(lib::ObMemAttr(orig_tenant_id_, "SessPackageInfo"))),

name_pool_(lib::ObMemAttr(orig_tenant_id_, ObModIds::OB_SQL_SESSION), OB_MALLOC_NORMAL_BLOCK_SIZE),

trans_flags_(),

sql_scope_flags_(),

need_reset_package_(false),

base_sys_var_alloc_(ObMemAttr(orig_tenant_id_, ObModIds::OB_SQL_SESSION), OB_MALLOC_NORMAL_BLOCK_SIZE),

inc_sys_var_alloc1_(ObMemAttr(orig_tenant_id_, ObModIds::OB_SQL_SESSION), OB_MALLOC_NORMAL_BLOCK_SIZE),

inc_sys_var_alloc2_(ObMemAttr(orig_tenant_id_, ObModIds::OB_SQL_SESSION), OB_MALLOC_NORMAL_BLOCK_SIZE),

current_buf_index_(0),

bucket_allocator_wrapper_(&block_allocator_),

user_var_val_map_(SMALL_BLOCK_SIZE, ObWrapperAllocator(&block_allocator_), orig_tenant_id_),

influence_plan_var_indexs_(),

is_first_gen_(true),

is_first_gen_config_(true),

sys_var_fac_(orig_tenant_id_),

next_frag_mem_point_(OB_MALLOC_NORMAL_BLOCK_SIZE), // 8KB

sys_vars_encode_max_size_(0),

consistency_level_(INVALID_CONSISTENCY),

tz_info_wrap_(),

next_tx_read_only_(-1),

next_tx_isolation_(transaction::ObTxIsolationLevel::INVALID),

log_id_level_map_valid_(false),

cur_phy_plan_(NULL),

plan_id_(0),

last_plan_id_(0),

flt_vars_(),

capability_(),

proxy_capability_(),

client_mode_(OB_MIN_CLIENT_MODE),

changed_sys_vars_(),

changed_user_vars_(),

changed_var_pool_(ObMemAttr(orig_tenant_id_, ObModIds::OB_SQL_SESSION), OB_MALLOC_NORMAL_BLOCK_SIZE),

extra_info_allocator_(ObMemAttr(orig_tenant_id_, ObModIds::OB_SQL_SESSION), OB_MALLOC_NORMAL_BLOCK_SIZE),

is_database_changed_(false),

feedback_manager_(),

trans_spec_status_(TRANS_SPEC_NOT_SET),

debug_sync_actions_(),

partition_hit_(),

magic_num_(0x13572468),

current_execution_id_(-1),

database_id_(OB_INVALID_ID),

retry_info_(),

last_query_trace_id_(),

nested_count_(-1),

inf_pc_configs_(),

curr_trans_last_stmt_end_time_(0),

check_sys_variable_(true),

is_foreign_key_cascade_(false),

is_foreign_key_check_exist_(false),

acquire_from_pool_(false),

release_to_pool_(true),

is_tenant_killed_(0),

reused_count_(0),

first_need_txn_stmt_type_(stmt::T_NONE),

need_recheck_txn_readonly_(false),

exec_min_cluster_version_(GET_MIN_CLUSTER_VERSION()),

stmt_type_(stmt::T_NONE),

labels_(),

thread_id_(0),

is_password_expired_(false),

process_query_time_(0),

last_update_tz_time_(0),

is_client_sessid_support_(false)

{

thread_data_.reset();

MEMSET(sys_vars_, 0, sizeof(sys_vars_));

log_id_level_map_.reset_level();

CHAR_CARRAY_INIT(tenant_);

CHAR_CARRAY_INIT(effective_tenant_);

CHAR_CARRAY_INIT(trace_id_buff_);

sql_id_[0] = '\0';

ssl_cipher_buff_[0] = '\0';

sess_bt_buff_[0] = '\0';

inc_sys_var_alloc_[0] = &inc_sys_var_alloc1_;

inc_sys_var_alloc_[1] = &inc_sys_var_alloc2_;

influence_plan_var_indexs_.set_attr(ObMemAttr(orig_tenant_id_, "PlanVaIdx"));

}

ObBasicSessionInfo::~ObBasicSessionInfo()

{

destroy();

}

bool ObBasicSessionInfo::is_server_status_in_transaction() const

{

bool in_txn = OB_NOT_NULL(tx_desc_) && tx_desc_->in_tx_for_free_route();

LOG_TRACE("decide flag: server in transaction", K(in_txn));

return in_txn;

}

//for test

int ObBasicSessionInfo::test_init(uint32_t sessid, uint64_t proxy_sessid,

common::ObIAllocator *bucket_allocator)

{

int ret = OB_SUCCESS;

if (NULL != bucket_allocator) {

bucket_allocator_wrapper_.set_alloc(bucket_allocator);

}

ret = user_var_val_map_.init(1024 * 1024 * 2,

256, // # of user variables

(NULL == bucket_allocator ? NULL : &bucket_allocator_wrapper_));

if (OB_FAIL(ret)) {

LOG_WARN("fail to init user_var_val_map", K(ret));

} else if (OB_FAIL(load_default_configs_in_pc())) {

LOG_WARN("fail to load default config influence plan cache", K(ret));

} else if (OB_FAIL(debug_sync_actions_.init(SMALL_BLOCK_SIZE, bucket_allocator_wrapper_))) {

LOG_WARN("fail to init debug sync actions", K(ret));

} else if (OB_FAIL(set_session_state(SESSION_INIT))) {

LOG_WARN("fail to set session stat", K(ret));

} else if (OB_FAIL(set_time_zone(ObString("+8:00"), is_oracle_compatible(),

true/* check_timezone_valid */))) {

LOG_WARN("fail to set time zone", K(ret));

} else {

// tz_info_wrap_.set_tz_info_map(GCTX.tz_info_mgr_->get_tz_info_map());

sessid_ = sessid;

proxy_sessid_ = proxy_sessid;

}

return ret;

}

bool ObBasicSessionInfo::is_use_inner_allocator() const

{

return bucket_allocator_wrapper_.get_alloc() == &block_allocator_;

}

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

common::ObIAllocator *bucket_allocator, const ObTZInfoMap *tz_info)

{

int ret = OB_SUCCESS;

ObWrapperAllocator *user_var_allocator_wrapper = NULL;

if (is_acquire_from_pool()) {

reused_count_++;

if (OB_NOT_NULL(bucket_allocator) || !is_use_inner_allocator()) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("session from pool must use inner allocator", K(ret));

}

} else {

if (NULL != bucket_allocator) {

bucket_allocator_wrapper_.set_alloc(bucket_allocator);

user_var_allocator_wrapper = &bucket_allocator_wrapper_;

}

}

if (OB_FAIL(ret)) {

} else if (!is_acquire_from_pool() &&

OB_FAIL(user_var_val_map_.init(1024 * 1024 * 2, 256, user_var_allocator_wrapper))) {

LOG_WARN("fail to init user_var_val_map", K(ret));

} else if (!is_acquire_from_pool() &&

OB_FAIL(debug_sync_actions_.init(SMALL_BLOCK_SIZE, bucket_allocator_wrapper_))) {

LOG_WARN("fail to init debug sync actions", K(ret));

} else if (OB_FAIL(set_session_state(SESSION_INIT))) {

LOG_WARN("fail to set session stat", K(ret));

/* } else if (FALSE_IT(tx_result_.set_trans_desc(&trans_desc_))) { */

} else {

sessid_ = sessid;

proxy_sessid_ = proxy_sessid;

uint64_t tenant_id = tenant_id_;

if (OB_ISNULL(tz_info)) {

ObTZMapWrap tz_map_wrap;

if (OB_FAIL(OTTZ_MGR.get_tenant_tz(tenant_id, tz_map_wrap))) {

LOG_WARN("get tenant timezone map failed", K(ret));

} else {

tz_info_wrap_.set_tz_info_map(tz_map_wrap.get_tz_map());

}

} else {

tz_info_wrap_.set_tz_info_map(tz_info);

}

}

return ret;

}

void ObBasicSessionInfo::destroy()

{

if (magic_num_ != 0x13572468) {

LOG_ERROR_RET(OB_ERROR, "ObBasicSessionInfo may be double free!!!", K(magic_num_));

}

if (OB_NOT_NULL(tx_desc_)) {

LOG_ERROR_RET(OB_ERR_UNEXPECTED, "tx_desc != NULL", KPC(this), KPC_(tx_desc));

}

tx_desc_ = NULL;

tx_result_.reset();

xid_.reset();

associated_xa_ = false;

cached_tenant_config_version_ = 0;

magic_num_ = 0x86427531;

if (thread_data_.cur_query_ != nullptr) {

ob_free(thread_data_.cur_query_);

thread_data_.cur_query_ = nullptr;

thread_data_.cur_query_buf_len_ = 0;

}

total_stmt_tables_.reset();

cur_stmt_tables_.reset();

}

void ObBasicSessionInfo::clean_status()

{

trans_flags_.reset();

sql_scope_flags_.reset();

trans_spec_status_ = TRANS_SPEC_NOT_SET;

if (OB_NOT_NULL(tx_desc_)) {

LockGuard lock_guard(thread_data_mutex_);

int ret = OB_SUCCESS;

MAKE_TENANT_SWITCH_SCOPE_GUARD(guard);

if (OB_SUCC(guard.switch_to(tx_desc_->get_tenant_id(), false))) {

MTL(transaction::ObTransService*)->release_tx(*tx_desc_);

}

tx_desc_ = NULL;

}

xid_.reset();

associated_xa_ = false;

cached_tenant_config_version_ = 0;

set_valid(true);

thread_data_.cur_query_start_time_ = 0;

thread_data_.cur_query_len_ = 0;

thread_data_.last_active_time_ = ObTimeUtility::current_time();

reset_session_changed_info();

}

void ObBasicSessionInfo::reset_user_var()

{

user_var_val_map_.reuse();

}

void ObBasicSessionInfo::reset(bool skip_sys_var)

{

set_valid(false);

if (OB_NOT_NULL(tx_desc_)) {

int ret = OB_SUCCESS;

MAKE_TENANT_SWITCH_SCOPE_GUARD(guard);

if (OB_SUCC(guard.switch_to(tx_desc_->get_tenant_id(), false))) {

MTL(transaction::ObTransService*)->release_tx(*tx_desc_);

}

tx_desc_ = NULL;

}

xid_.reset();

associated_xa_ = false;

cached_tenant_config_version_ = 0;

is_deserialized_ = false;

CHAR_CARRAY_INIT(tenant_);

tenant_id_ = OB_INVALID_ID;

CHAR_CARRAY_INIT(effective_tenant_);

effective_tenant_id_ = OB_INVALID_ID;

is_changed_to_temp_tenant_ = false;

user_id_ = OB_INVALID_ID;

client_version_.reset();

driver_version_.reset();

sessid_ = 0;

master_sessid_ = INVALID_SESSID;

proxy_sessid_ = VALID_PROXY_SESSID;

global_vars_version_ = 0;

tx_result_.reset();

total_stmt_tables_.reset();

cur_stmt_tables_.reset();

// reset() of user_var_val_map_ and debug_sync_actions_ will keep some memory

// allocated from block_allocator_ / bucket_allocator_wrapper_, so we skip

// reset() of block_allocator_ and bucket_allocator_wrapper_.

//block_allocator_.reset();

ps_session_info_allocator_.reset();

cursor_info_allocator_.reset();

package_info_allocator_.reset();

trans_flags_.reset();

sql_scope_flags_.reset();

need_reset_package_ = false;

//bucket_allocator_wrapper_.reset();

user_var_val_map_.reuse();

if (!skip_sys_var) {

memset(sys_vars_, 0, sizeof(sys_vars_));

influence_plan_var_indexs_.reset();

} else {

const SysVarIds &all_sys_var_ids = sys_var_inc_info_.get_all_sys_var_ids();

for (int i = 0; i < all_sys_var_ids.count(); i++) {

int ret = OB_SUCCESS;

int64_t store_idx = -1;

ObSysVarClassType sys_var_id = all_sys_var_ids.at(i);

OZ (ObSysVarFactory::calc_sys_var_store_idx(sys_var_id, store_idx));

OV (0 <= store_idx && store_idx < ObSysVarFactory::ALL_SYS_VARS_COUNT);

OV (OB_NOT_NULL(sys_vars_[store_idx]));

OX (sys_vars_[store_idx]->clean_inc_value());

}

// I don't see any reason why we should reset timezone

// reset_timezone();

}

sys_var_inc_info_.reset();

sys_var_in_pc_str_.reset();

config_in_pc_str_.reset();

flt_vars_.reset();

is_first_gen_ = true;

is_first_gen_config_ = true;

CHAR_CARRAY_INIT(trace_id_buff_);

CHAR_CARRAY_INIT(sql_id_);

char *sql_id = sql_id_;

sql_id = NULL;

//consistency_level_ = INVALID_CONSISTENCY;

next_tx_read_only_ = -1;

next_tx_isolation_ = transaction::ObTxIsolationLevel::INVALID;

log_id_level_map_valid_ = false;

log_id_level_map_.reset_level();

cur_phy_plan_ = NULL;

plan_id_ = 0;

last_plan_id_ = 0;

capability_.capability_ = 0;

proxy_capability_.capability_ = 0;

client_attribute_capability_.capability_ = 0;

client_mode_ = OB_MIN_CLIENT_MODE;

reset_session_changed_info();

extra_info_allocator_.reset();

trans_spec_status_ = TRANS_SPEC_NOT_SET;

debug_sync_actions_.reset();

partition_hit_.reset();

//magic_num_ = 0x86427531;

current_execution_id_ = -1;

last_trace_id_.reset();

curr_trace_id_.reset();

app_trace_id_.reset();

database_id_ = OB_INVALID_ID;

retry_info_.reset();

last_query_trace_id_.reset();

thread_data_.reset();

nested_count_ = -1;

// session 缓存的场景下，只保留 base 值，清理 inc 值

// 去掉 sys var schema version 后，base 值即为 hardcode 值

if (!skip_sys_var) {

sys_vars_cache_.reset();

sys_var_base_version_ = OB_INVALID_VERSION;

} else {

sys_vars_cache_.clean_inc();

sys_var_base_version_ = CACHED_SYS_VAR_VERSION;

}

curr_trans_last_stmt_end_time_ = 0;

reserved_read_snapshot_version_.reset();

check_sys_variable_ = true;

is_foreign_key_cascade_ = false;

is_foreign_key_check_exist_ = false;

acquire_from_pool_ = false;

// 不要重置release_to_pool_，原因见属性声明位置的注释。

is_tenant_killed_ = 0;

first_need_txn_stmt_type_ = stmt::T_NONE;

need_recheck_txn_readonly_ = false;

exec_min_cluster_version_ = GET_MIN_CLUSTER_VERSION();

labels_.reuse();

thread_id_ = 0;

is_password_expired_ = false;

process_query_time_ = 0;

last_update_tz_time_ = 0;

is_client_sessid_support_ = false;

sess_bt_buff_pos_ = 0;

ATOMIC_SET(&sess_ref_cnt_ , 0);

// 最后再重置所有allocator

// 否则thread_data_.user_name_之类的属性会有野指针，在session_mgr的foreach接口遍历时可能core掉。

name_pool_.reset();

inc_sys_var_alloc1_.reset();

inc_sys_var_alloc2_.reset();

current_buf_index_ = 0;

if (!skip_sys_var) {

base_sys_var_alloc_.reset();

sys_var_fac_.destroy();

}

client_identifier_.reset();

}

int ObBasicSessionInfo::reset_timezone()

{

int ret = OB_SUCCESS;

ObObj tmp_obj1;

if (OB_FAIL(get_sys_variable(SYS_VAR_TIME_ZONE, tmp_obj1))) {

LOG_WARN("get sys var failed", K(ret));

} else if (OB_FAIL(process_session_time_zone_value(tmp_obj1, false))) {

LOG_WARN("set time zone failed", K(ret));

}

ObObj tmp_obj2;

if (OB_FAIL(get_sys_variable(SYS_VAR_ERROR_ON_OVERLAP_TIME, tmp_obj2))) {

LOG_WARN("get sys var failed", K(ret));

} else if (OB_FAIL(process_session_overlap_time_value(tmp_obj2))) {

LOG_WARN("process session overlap time value failed", K(ret), K(tmp_obj2));

}

return ret;

}

int ObBasicSessionInfo::init_tenant(const ObString &tenant_name, const uint64_t tenant_id)

{

int ret = OB_SUCCESS;

if (OB_UNLIKELY(!is_valid_tenant_id(tenant_id))) {

ret = OB_INVALID_ARGUMENT;

LOG_WARN("invalid tenant id", K(tenant_id), K(ret));

} else if (tenant_name.empty()) {

ret = OB_INVALID_ARGUMENT;

LOG_WARN("tenant name is empty", K(tenant_name), K(ret));

} else if (tenant_name.length() > OB_MAX_TENANT_NAME_LENGTH) {

ret = OB_INVALID_ARGUMENT;

LOG_WARN("tenant name too long", K(tenant_name), K(ret));

} else if (OB_FAIL(ob_cstrcopy(tenant_, sizeof(tenant_), tenant_name))) {

LOG_WARN("failed to copy tenant name", K(tenant_name), K(ret));

} else if (OB_FAIL(ob_cstrcopy(effective_tenant_, sizeof(effective_tenant_), tenant_name))) {

LOG_WARN("failed to copy effective tenant name", K(tenant_name), K(ret));

} else {

ObTZMapWrap tz_map_wrap;

if (OB_FAIL(OTTZ_MGR.get_tenant_tz(tenant_id, tz_map_wrap))) {

LOG_WARN("get tenant timezone map failed", K(ret));

} else {

tz_info_wrap_.set_tz_info_map(tz_map_wrap.get_tz_map());

tenant_id_ = tenant_id;

effective_tenant_id_ = tenant_id;

LOG_DEBUG("init session tenant", K(tenant_name), K(tenant_id));

}

}

return ret;

}

int ObBasicSessionInfo::set_tenant(const common::ObString &tenant_name, const uint64_t tenant_id)

{

int ret = OB_SUCCESS;

if (!is_valid_tenant_id(tenant_id)) {

ret = OB_INVALID_ARGUMENT;

LOG_WARN("invalid tenant id", K(tenant_id), K(ret));

} else if (tenant_name.empty()) {

ret = OB_INVALID_ARGUMENT;

LOG_WARN("tenant name is empty", K(tenant_name), K(ret));

} else if (tenant_name.length() > OB_MAX_TENANT_NAME_LENGTH) {

ret = OB_INVALID_ARGUMENT;

LOG_WARN("tenant name too long", K(tenant_name), K(ret));

} else if (OB_FAIL(ob_cstrcopy(tenant_, sizeof(tenant_), tenant_name))) {

LOG_WARN("tenant name too long", K(tenant_name));

} else {

tenant_id_ = tenant_id;

LOG_TRACE("set tenant", K(tenant_name), K(tenant_id));

}

return ret;

}

int ObBasicSessionInfo::set_tenant(const common::ObString &tenant_name,

const uint64_t tenant_id,

char *ori_tenant_name,

const uint64_t length,

uint64_t &ori_tenant_id)

{

int ret = OB_SUCCESS;

if (!is_valid_tenant_id(tenant_id)) {

ret = OB_INVALID_ARGUMENT;

LOG_WARN("invalid tenant id", K(tenant_id), K(ret));

} else if (OB_ISNULL(ori_tenant_name)) {

ret = OB_INVALID_ARGUMENT;

LOG_WARN("ori_tenant_name is NULL", K(ret));

} else if (length < OB_MAX_TENANT_NAME_LENGTH + 1) {

ret = OB_INVALID_ARGUMENT_FOR_LENGTH;

LOG_WARN("tenant_name length is not enough ");

} else if (tenant_name.empty()) {

ret = OB_INVALID_ARGUMENT;

LOG_WARN("tenant name is empty", K(tenant_name), K(ret));

} else if (tenant_name.length() > OB_MAX_TENANT_NAME_LENGTH) {

ret = OB_INVALID_ARGUMENT;

LOG_WARN("tenant name too long", K(tenant_name), K(ret));

} else if (OB_FAIL(ob_cstrcopy(ori_tenant_name, length, tenant_, strlen(tenant_)))) {

LOG_ERROR("tenant_name is longer than ori_tenant_name", K(length), K(strlen(tenant_)));

} else if (OB_FAIL(ob_cstrcopy(tenant_, sizeof(tenant_), tenant_name))) {

LOG_ERROR("tenant name too long", K(tenant_name));

} else {

ori_tenant_id = tenant_id_;

tenant_id_ = tenant_id;

}

return ret;

}

int ObBasicSessionInfo::switch_tenant_with_name(

uint64_t effective_tenant_id, const common::ObString &tenant_name)

{

int ret = OB_SUCCESS;

if (!is_valid_tenant_id(effective_tenant_id)) {

ret = OB_INVALID_ARGUMENT;

LOG_WARN("invalid tenant id", K(ret), K(effective_tenant_id));

} else if (tenant_name.empty()) {

ret = OB_INVALID_ARGUMENT;

LOG_WARN("tenant name is empty", K(ret), K(tenant_name));

} else if (tenant_name.length() > OB_MAX_TENANT_NAME_LENGTH) {

ret = OB_INVALID_ARGUMENT;

LOG_WARN("tenant name too long", K(ret), K(tenant_name));

} else if (OB_FAIL(switch_tenant(effective_tenant_id))) {

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

} else if (OB_FAIL(ob_cstrcopy(effective_tenant_, sizeof(effective_tenant_), tenant_name))) {

LOG_WARN("tenant name too long", K(ret), K(tenant_name));

}

return ret;

}

int ObBasicSessionInfo::switch_tenant(uint64_t effective_tenant_id)

{

int ret = OB_SUCCESS;

if (OB_SYS_TENANT_ID != tenant_id_) {

ret = OB_NOT_SUPPORTED;

LOG_USER_ERROR(OB_NOT_SUPPORTED, "switching tenant from a non-sys tenant");

LOG_WARN("only support sys tenant switch tenant", K(ret), K(tenant_id_), K(effective_tenant_id_));

} else if (effective_tenant_id <= 0) {

ret = OB_INVALID_ARGUMENT;

LOG_WARN("invalid tenant_id", K(ret), K(effective_tenant_id));

} else if (OB_NOT_NULL(tx_desc_) && effective_tenant_id != effective_tenant_id_) {

if (tx_desc_->in_tx_or_has_extra_state()) {

ret = OB_NOT_SUPPORTED;

// only inner-SQL goes switch_tenant and may fall into such state

// print out error to easy trouble-shot

LOG_USER_ERROR(OB_NOT_SUPPORTED,

"try to switch to another tenant without commit/rollback in a transaction");

LOG_ERROR("try to switch another tenant while session has active txn,"

" you must commit/rollback first", K(ret),

"current_effective_tenant_id", effective_tenant_id_,

"target_effective_tenant_id", effective_tenant_id,

KPC(tx_desc_), KPC(this));

} else if (OB_FAIL(ObSqlTransControl::reset_session_tx_state(this))) {

LOG_WARN("reset session tx state fail", K(ret), KPC(this));

}

}

if (OB_SUCC(ret)) {

#ifndef NDEBUG

if (effective_tenant_id_ != effective_tenant_id) {

LOG_INFO("switch tenant",

"target_effective_tenant_id", effective_tenant_id,

"current_effective_tenant_id", effective_tenant_id_,

"priv_tenant_id", tenant_id_,

K(lbt()));

}

#endif

effective_tenant_id_ = effective_tenant_id;

}

return ret;

}

const common::ObString ObBasicSessionInfo::get_tenant_name() const

{

return ObString::make_string(tenant_);

}

const common::ObString ObBasicSessionInfo::get_effective_tenant_name() const

{

return ObString::make_string(effective_tenant_);

}

int ObBasicSessionInfo::set_user(const ObString &user_name, const ObString &host_name, const uint64_t user_id)

{

int ret = OB_SUCCESS;

if (OB_UNLIKELY(user_name.length() > common::OB_MAX_USER_NAME_LENGTH)

|| OB_UNLIKELY(host_name.length() > common::OB_MAX_HOST_NAME_LENGTH)) {

ret = OB_INVALID_ARGUMENT;

LOG_WARN("name length invalid_", K(user_name), K(host_name), K(ret));

} else {

char tmp_buf[common::OB_MAX_USER_NAME_LENGTH + common::OB_MAX_HOST_NAME_LENGTH + 2] = {};

snprintf(tmp_buf, sizeof(tmp_buf), "%.*s@%.*s", user_name.length(), user_name.ptr(),

host_name.length(), host_name.ptr());

ObString tmp_string(tmp_buf);

LockGuard lock_guard(thread_data_mutex_);

if (OB_FAIL(name_pool_.write_string(user_name, &thread_data_.user_name_))) {

LOG_WARN("fail to write username to string_buf_", K(user_name), K(ret));

} else if (OB_FAIL(name_pool_.write_string(host_name, &thread_data_.host_name_))) {

LOG_WARN("fail to write hostname to string_buf_", K(host_name), K(ret));

} else if (OB_FAIL(name_pool_.write_string(tmp_string, &thread_data_.user_at_host_name_))) {

LOG_WARN("fail to write user_at_host_name to string_buf_", K(tmp_string), K(ret));

} else {

user_id_ = user_id;

}

}

return ret;

}

int ObBasicSessionInfo::set_real_client_ip(const common::ObString &client_ip)

{

int ret = OB_SUCCESS;

char tmp_buf[common::OB_MAX_USER_NAME_LENGTH + common::OB_MAX_HOST_NAME_LENGTH + 2] = {};

snprintf(tmp_buf, sizeof(tmp_buf), "%.*s@%.*s", thread_data_.user_name_.length(),

thread_data_.user_name_.ptr(),

client_ip.length(),

client_ip.ptr());

ObString tmp_string(tmp_buf);

LockGuard lock_guard(thread_data_mutex_);

if (OB_FAIL(name_pool_.write_string(client_ip, &thread_data_.client_ip_))) {

LOG_WARN("fail to write client_ip to string_buf_", K(client_ip), K(ret));

} else if (OB_FAIL(name_pool_.write_string(tmp_string, &thread_data_.user_at_client_ip_))) {

LOG_WARN("fail to write user_at_host_name to string_buf_", K(tmp_string), K(ret));

} else {

thread_data_.user_client_addr_.set_ip_addr(client_ip, 0);

}

return ret;

}

int ObBasicSessionInfo::set_client_identifier(const common::ObString &client_identifier)

{

int ret = OB_SUCCESS;

int max_size = OB_MAX_CONTEXT_CLIENT_IDENTIFIER_LENGTH_IN_SESSION;

if (OB_FAIL(init_client_identifier())) {

LOG_WARN("failed to init client identifier", K(ret));

} else {

//reset curr identifier

client_identifier_.set_length(0);

//write new string

int64_t write_len = std::min(client_identifier_.size(), client_identifier.length());

if (write_len != client_identifier_.write(client_identifier.ptr(), write_len)) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("failed to write client identifier", K(ret), K(client_identifier_), K(write_len));

}

}

return ret;

}

int ObBasicSessionInfo::init_client_identifier()

{

int ret = OB_SUCCESS;

int max_size = OB_MAX_CONTEXT_CLIENT_IDENTIFIER_LENGTH_IN_SESSION;

if (OB_ISNULL(client_identifier_.ptr())) {

char *ptr = nullptr;

if (OB_ISNULL(ptr = static_cast<char *> (get_session_allocator().alloc(max_size)))) {

ret = OB_ALLOCATE_MEMORY_FAILED;

LOG_WARN("failed to alloc mem for client identifier", K(ret));

} else {

client_identifier_.assign_buffer(ptr, max_size);

}

} else if (max_size != client_identifier_.size()) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("get wrong client identifier", K(ret), K(client_identifier_.size()));

}

return ret;

}

int ObBasicSessionInfo::check_and_init_retry_info(const ObCurTraceId::TraceId &cur_trace_id,

const ObString &sql)

{

int ret = OB_SUCCESS;

// 下面这段判断影响太大，害怕出问题，所以打了ERROR之后还是继续执行，仅仅影响retry info，还是可以正常执行的

if (last_query_trace_id_.equals(cur_trace_id)) { // 是重试query的包

if (OB_UNLIKELY(!retry_info_.is_inited())) {

LOG_ERROR("is retry packet, but retry info is not inited, will init it",

K(last_query_trace_id_), K(cur_trace_id), K(retry_info_), K(get_sessid()), K(sql));

if (OB_FAIL(retry_info_.init())) {

LOG_WARN("fail to init retry info", K(ret), K(retry_info_), K(sql));

}

}

} else {

//@TODO: 不是重试query的包，在正常语句结束的时候，不做清空重试信息逻辑，后续这个地方会做成按需初始化,

//减少正常语句的执行开销，所以如果是上一条语句初始化了重试信息，这里直接清空掉，

//因为对于异步执行而言，在控制端线程结束的时候判断无法判断其重试状态，所以不能做清空重试信息操作

if (OB_UNLIKELY(retry_info_.is_inited())) {

retry_info_.reset();

}

// 不是重试query的包都要init retry info

if (OB_FAIL(retry_info_.init())) {

LOG_WARN("fail to init retry info", K(ret), K(retry_info_), K(sql));

} else {

last_query_trace_id_.set(cur_trace_id);

}

}

return ret;

}

const ObLogIdLevelMap *ObBasicSessionInfo::get_log_id_level_map() const

{

return (log_id_level_map_valid_ ? (&log_id_level_map_) : NULL);

}

int ObBasicSessionInfo::set_client_version(const ObString &client_version)

{

int ret = OB_SUCCESS;

if (client_version.empty()) {

ret = OB_INVALID_ARGUMENT;

LOG_WARN("client version is empty", K(client_version), K(ret));

} else if (OB_FAIL(name_pool_.write_string(client_version, &client_version_))) {

LOG_WARN("failed to write client_version to string_buf_", K(ret));

} else {}

return ret;

}

int ObBasicSessionInfo::set_driver_version(const ObString &driver_version)

{

int ret = OB_SUCCESS;

if (driver_version.empty()) {

ret = OB_INVALID_ARGUMENT;

LOG_WARN("driver version is empty", K(driver_version), K(ret));

} else if (OB_FAIL(name_pool_.write_string(driver_version, &driver_version_))) {

LOG_WARN("failed to write driver_version to string_buf_", K(ret));

} else {}

return ret;

}

int ObBasicSessionInfo::set_default_database(const ObString &database_name,

const ObCollationType coll_type/*= CS_TYPE_INVALID */)

{

int ret = OB_SUCCESS;

if (database_name.length() > OB_MAX_DATABASE_NAME_LENGTH * OB_MAX_CHAR_LEN) {

ret = OB_INVALID_ARGUMENT_FOR_LENGTH;

LOG_WARN("invalid length for database_name", K(database_name), K(ret));

} else {

if (CS_TYPE_INVALID != coll_type) {

const int64_t coll_val = static_cast<int64_t>(coll_type);

if (OB_FAIL(update_sys_variable(SYS_VAR_CHARACTER_SET_DATABASE, coll_val))) {

LOG_WARN("failed to update variable", K(ret));

} else if (OB_FAIL(update_sys_variable(SYS_VAR_COLLATION_DATABASE, coll_val))) {

LOG_WARN("failed to update variable", K(ret));

} else {}

}

if (OB_SUCC(ret)) {

LockGuard lock_guard(thread_data_mutex_);

MEMCPY(thread_data_.database_name_, database_name.ptr(), database_name.length());

thread_data_.database_name_[database_name.length()] = '\0';

if (is_track_session_info()) {

is_database_changed_ = true;

}

}

}

return ret;

}

int ObBasicSessionInfo::update_database_variables(ObSchemaGetterGuard *schema_guard)

{

int ret = OB_SUCCESS;

if (OB_ISNULL(schema_guard)) {

ret = OB_INVALID_ARGUMENT;

LOG_WARN("invalid schema guard is NULL", K(ret));

} else {

if ('\0' == thread_data_.database_name_[0]) {

// no default database

ObObj val;

if (OB_FAIL(get_sys_variable(SYS_VAR_CHARACTER_SET_SERVER, val))) {

LOG_WARN("failed to get sys variable", K(ret));

} else if (OB_FAIL(update_sys_variable(SYS_VAR_CHARACTER_SET_DATABASE, val))) {

LOG_WARN("failed to update sys variable", K(ret));

} else if (OB_FAIL(get_sys_variable(SYS_VAR_COLLATION_SERVER, val))) {

LOG_WARN("failed to get sys variable", K(ret));

} else if (OB_FAIL(update_sys_variable(SYS_VAR_COLLATION_DATABASE, val))) {

LOG_WARN("failed to update sys variable", K(ret));

} else {}

} else {

const share::schema::ObDatabaseSchema *db_schema = NULL;

ObString db_name(thread_data_.database_name_);

if (OB_FAIL(schema_guard->get_database_schema(effective_tenant_id_, db_name,

db_schema))) {

LOG_WARN("get database schema failed",

K(effective_tenant_id_), K(db_name), K(ret));

} else if (NULL == db_schema) {

ret = OB_ERR_BAD_DATABASE;

LOG_WARN("database not exist",

K(effective_tenant_id_), K(db_name), K(ret));

LOG_USER_ERROR(OB_ERR_BAD_DATABASE, db_name.length(), db_name.ptr());

} else {

const int64_t db_coll = static_cast<int64_t>(db_schema->get_collation_type());

if (OB_FAIL(update_sys_variable(SYS_VAR_CHARACTER_SET_DATABASE, db_coll))) {

LOG_WARN("failed to update sys variable", K(ret));

} else if (OB_FAIL(update_sys_variable(SYS_VAR_COLLATION_DATABASE, db_coll))) {

LOG_WARN("failed to update sys variable", K(ret));

} else {}

}

}

}

return ret;

}

int ObBasicSessionInfo::update_max_packet_size()

{

int ret = OB_SUCCESS;

int64_t max_allowed_pkt = 0;

int64_t net_buffer_len = 0;

if (OB_FAIL(get_max_allowed_packet(max_allowed_pkt))) {

LOG_WARN("fail to get_max_allowed_packet", K(ret));

} else if (OB_FAIL(get_net_buffer_length(net_buffer_len))) {

LOG_WARN("fail to get_net_buffer_length", K(ret));

} else {

thread_data_.max_packet_size_ = std::max(max_allowed_pkt, net_buffer_len);

}

return ret;

}

const ObString ObBasicSessionInfo::get_database_name() const

{

ObString str_ret;

str_ret.assign_ptr(const_cast<char*>(thread_data_.database_name_),

static_cast<int32_t>(strlen(thread_data_.database_name_)));

return str_ret;

}

//// FIXME: xiyu

//int ObBasicSessionInfo::get_database_id(

// ObSchemaGetterGuard *schema_guard,

// uint64_t &db_id) const

//{

// int ret = OB_SUCCESS;

// db_id = OB_INVALID_ID;

// if (get_database_name().empty()) {

// //do nothing

// } else if (OB_UNLIKELY(NULL == schema_guard)) {

// ret = OB_INVALID_ARGUMENT;

// LOG_WARN("Schema guard should not be NULL", K(ret));

// } else if (OB_FAIL(schema_guard->get_database_id(get_effective_tenant_id(), get_database_name(),

// db_id))) {

// db_id = OB_INVALID_ID;

// LOG_WARN("failed to get database id", K(db_id), K(ret));

// } else { }//do nothing

// return ret;

//}

////////////////////////////////////////////////////////////////

int ObBasicSessionInfo::get_global_sys_variable(const ObBasicSessionInfo *session,

ObIAllocator &calc_buf,

const ObString &var_name,

ObObj &val)

{

int ret = OB_SUCCESS;

if (OB_ISNULL(session)) {

ret = OB_INVALID_ARGUMENT;

LOG_WARN("session is NULL", K(ret), K(var_name));

} else {

//const ObDataTypeCastParams dtc_params(session->get_timezone_info(),

// session->get_local_nls_formats(),

// session->get_nls_collation(),

// session->get_nls_collation_nation());

ObDataTypeCastParams dtc_params = session->get_dtc_params();

if (OB_FAIL(get_global_sys_variable(session->get_effective_tenant_id(),

calc_buf, dtc_params, var_name, val))) {

LOG_WARN("fail to get global sys variable", K(ret), K(var_name));

}

}

return ret;

}

int ObBasicSessionInfo::get_global_sys_variable(const uint64_t actual_tenant_id, // 为了处理租户已经切掉的情况

ObIAllocator &calc_buf,

const ObDataTypeCastParams &dtc_params,

const ObString &var_name,

ObObj &val)

{

int ret = OB_SUCCESS;

ObSchemaGetterGuard schema_guard;

const ObTenantSchema *tenant_schema = NULL;

const ObSysVarSchema *sysvar_schema = NULL;

const ObSysVariableSchema *sys_variable_schema = NULL;

if (OB_UNLIKELY(!is_valid_tenant_id(actual_tenant_id))) {

ret = OB_INVALID_ARGUMENT;

LOG_WARN("invalid tenant id", K(actual_tenant_id), K(ret));

} else if (OB_ISNULL(GCTX.schema_service_)) {

ret = OB_INVALID_ARGUMENT;

OB_LOG(WARN,"invalid argument", K(GCTX.schema_service_));

} else if (OB_FAIL(GCTX.schema_service_->get_tenant_schema_guard(

actual_tenant_id,

schema_guard))) {

ret = OB_SCHEMA_ERROR;

OB_LOG(WARN,"fail get schema guard", K(ret));

} else if (OB_FAIL(schema_guard.get_tenant_info(actual_tenant_id, tenant_schema))) {

LOG_WARN("get tenant info failed", K(ret), K(actual_tenant_id));

} else if (OB_ISNULL(tenant_schema)) {

ret = OB_SCHEMA_ERROR;

LOG_WARN("tenant_schema is NULL", K(ret));

} else if (OB_FAIL(schema_guard.get_sys_variable_schema(actual_tenant_id, sys_variable_schema))) {

LOG_WARN("get sys variable schema failed", K(ret));

} else if (OB_ISNULL(sys_variable_schema)) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("sys variable schema is null", K(ret));

} else if (OB_FAIL(sys_variable_schema->get_sysvar_schema(var_name, sysvar_schema))) {

ret = OB_ERR_SYS_VARIABLE_UNKNOWN;

LOG_WARN("failed to get sysvar", K(ret), K(var_name));

} else if (OB_ISNULL(sysvar_schema)) {

ret = OB_SCHEMA_ERROR;

LOG_WARN("tenant_schema is NULL", K(ret));

} else if (OB_FAIL(sysvar_schema->get_value(&calc_buf, dtc_params, val))) {

LOG_WARN("failed to get value", K(ret), K(var_name));

} else if (OB_FAIL(ObBasicSessionInfo::change_value_for_special_sys_var(

var_name, val, val))) {

LOG_ERROR("fail to change value for special sys var", K(ret), K(var_name), K(val));

} else {

LOG_DEBUG("get global sysvar", K(var_name), K(val));

}

return ret;

}

const ObBasicSysVar *ObBasicSessionInfo::get_sys_var(const int64_t idx) const

{

const ObBasicSysVar *var = NULL;

if (idx >= 0 && idx < ObSysVarFactory::ALL_SYS_VARS_COUNT) {

var = sys_vars_[idx];

}

return var;

}

ObBasicSysVar *ObBasicSessionInfo::get_sys_var(const int64_t idx)

{

ObBasicSysVar *var = NULL;

if (idx >= 0 && idx < ObSysVarFactory::ALL_SYS_VARS_COUNT) {

var = sys_vars_[idx];

}

return var;

}

int ObBasicSessionInfo::init_system_variables(const bool print_info_log, const bool is_sys_tenant,

bool is_deserialized)

{

int ret = OB_SUCCESS;

ObString name;

ObObj type;

ObObj value;

ObObj min_val;

ObObj max_val;

ObObjType var_type = ObNullType;

int64_t var_flag = ObSysVarFlag::NONE;

int64_t var_amount = ObSysVariables::get_amount();

ObObj casted_value;

ObArenaAllocator calc_buf(ObModIds::OB_SQL_SESSION);

ObCastCtx cast_ctx(&calc_buf, NULL, CM_NONE, ObCharset::get_system_collation());

for (int64_t i = 0; OB_SUCC(ret) && i < var_amount; ++i) {

name.assign_ptr(const_cast<char*>(ObSysVariables::get_name(i).ptr()),

static_cast<ObString::obstr_size_t>(strlen(ObSysVariables::get_name(i).ptr())));

bool is_exist = false;

if (OB_FAIL(sys_variable_exists(name, is_exist))) {

LOG_WARN("failed to check if sys variable exists", K(name), K(ret));

} else if (!is_exist) {