/**

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

#include "ob_plan_cache_value.h"

#include "share/schema/ob_schema_getter_guard.h"

#include "share/schema/ob_schema_struct.h"

#include "share/stat/ob_opt_stat_manager.h"

#include "sql/parser/parse_malloc.h"

#include "sql/resolver/ob_resolver_utils.h"

#include "sql/ob_sql_context.h"

#include "sql/executor/ob_task_executor_ctx.h"

#include "sql/engine/ob_exec_context.h"

#include "sql/plan_cache/ob_pcv_set.h"

#include "sql/plan_cache/ob_plan_cache.h"

#include "sql/plan_cache/ob_plan_set.h"

#include "sql/session/ob_sql_session_info.h"

#include "sql/udr/ob_udr_mgr.h"

#include "sql/udr/ob_udr_utils.h"

#include "share/ob_duplicate_scope_define.h"

#include "pl/ob_pl_stmt.h"

#include "share/resource_manager/ob_resource_manager.h"

#include "sql/plan_cache/ob_values_table_compression.h"

using namespace oceanbase::share::schema;

using namespace oceanbase::common;

using namespace oceanbase::pl;

namespace oceanbase

{

namespace sql

{

int PCVSchemaObj::init(const ObTableSchema *schema)

{

int ret = OB_SUCCESS;

if (OB_ISNULL(schema) || OB_ISNULL(inner_alloc_)) {

ret = OB_INVALID_ARGUMENT;

LOG_WARN("unexpected null argument", K(ret), K(schema), K(inner_alloc_));

} else {

tenant_id_ = schema->get_tenant_id();

database_id_ = schema->get_database_id();

schema_id_ = schema->get_table_id();

schema_version_ = schema->get_schema_version();

schema_type_ = TABLE_SCHEMA;

table_type_ = schema->get_table_type();

is_tmp_table_ = schema->is_tmp_table();

// copy table name

char *buf = nullptr;

const ObString &tname = schema->get_table_name_str();

if (nullptr == (buf = static_cast<char *>(inner_alloc_->alloc(tname.length())))) {

ret = OB_ALLOCATE_MEMORY_FAILED;

LOG_WARN("failed to allocate memory", K(ret), K(tname.length()));

} else {

MEMCPY(buf, tname.ptr(), tname.length());

table_name_.assign_ptr(buf, tname.length());

}

}

return ret;

}

int PCVSchemaObj::init_with_synonym(const ObSimpleSynonymSchema *schema) {

int ret = OB_SUCCESS;

if (OB_ISNULL(schema) || OB_ISNULL(inner_alloc_)) {

ret = OB_INVALID_ARGUMENT;

LOG_WARN("unexpected null argument", K(ret), K(schema), K(inner_alloc_));

} else {

database_id_ = schema->get_database_id();

// copy table name

char *buf = nullptr;

const ObString &tname = schema->get_synonym_name_str();

if (nullptr == (buf = static_cast<char *>(inner_alloc_->alloc(tname.length())))) {

ret = OB_ALLOCATE_MEMORY_FAILED;

LOG_WARN("failed to allocate memory", K(ret), K(tname.length()));

} else {

MEMCPY(buf, tname.ptr(), tname.length());

table_name_.assign_ptr(buf, tname.length());

}

}

return ret;

}

bool PCVSchemaObj::operator==(const PCVSchemaObj &other) const

{

bool ret = true;

if (schema_type_ != other.schema_type_) {

ret = false;

} else if (TABLE_SCHEMA == other.schema_type_) {

ret = tenant_id_ == other.tenant_id_ &&

database_id_ == other.database_id_ &&

schema_id_ == other.schema_id_ &&

schema_version_ == other.schema_version_ &&

table_type_ == other.table_type_;

} else {

ret = schema_id_ == other.schema_id_ &&

schema_version_ == other.schema_version_;

}

return ret;

}

int PCVSchemaObj::init_without_copy_name(const ObSimpleTableSchemaV2 *schema)

{

int ret = OB_SUCCESS;

if (OB_ISNULL(schema)) {

ret = OB_INVALID_ARGUMENT;

LOG_WARN("unexpected null argument", K(ret), K(schema));

} else {

tenant_id_ = schema->get_tenant_id();

database_id_ = schema->get_database_id();

schema_id_ = schema->get_table_id();

schema_version_ = schema->get_schema_version();

schema_type_ = TABLE_SCHEMA;

table_type_ = schema->get_table_type();

is_tmp_table_ = schema->is_tmp_table();

table_name_ = schema->get_table_name_str();

}

return ret;

}

int PCVSchemaObj::init_with_version_obj(const ObSchemaObjVersion &schema_obj_version)

{

int ret = OB_SUCCESS;

schema_type_ = schema_obj_version.get_schema_type();

schema_id_ = schema_obj_version.object_id_;

schema_version_ = schema_obj_version.version_;

return ret;

}

void PCVSchemaObj::reset()

{

tenant_id_ = common::OB_INVALID_ID;

database_id_ = common::OB_INVALID_ID;

schema_id_ = common::OB_INVALID_ID;

schema_type_ = OB_MAX_SCHEMA;

schema_version_ = 0;

table_type_ = MAX_TABLE_TYPE;

is_tmp_table_ = false;

if (inner_alloc_ != nullptr && table_name_.ptr() != nullptr) {

inner_alloc_->free(table_name_.ptr());

table_name_.reset();

inner_alloc_ = nullptr;

}

}

PCVSchemaObj::~PCVSchemaObj()

{

reset();

}

ObPlanCacheValue::ObPlanCacheValue()

: pcv_set_(NULL),

pc_alloc_(NULL),

last_plan_id_(OB_INVALID_ID),

//use_global_location_cache_(true),

tenant_schema_version_(OB_INVALID_VERSION),

sys_schema_version_(OB_INVALID_VERSION),

outline_state_(),

outline_params_wrapper_(),

sessid_(OB_INVALID_ID),

sess_create_time_(0),

contain_sys_name_table_(false),

#ifdef OB_BUILD_SPM

is_spm_closed_(false),

#endif

need_param_(true),

is_nested_sql_(false),

is_batch_execute_(false),

has_dynamic_values_table_(false),

stored_schema_objs_(pc_alloc_),

stmt_type_(stmt::T_MAX)

{

MEMSET(sql_id_, 0, sizeof(sql_id_));

not_param_index_.set_attr(ObMemAttr(MTL_ID(), "NotParamIdex"));

neg_param_index_.set_attr(ObMemAttr(MTL_ID(), "NegParamIdex"));

must_be_positive_idx_.set_attr(ObMemAttr(MTL_ID(), "MustBePosiIdx"));

not_param_info_.set_attr(ObMemAttr(MTL_ID(), "NotParamInfo"));

not_param_var_.set_attr(ObMemAttr(MTL_ID(), "NotParamVar"));

param_charset_type_.set_attr(ObMemAttr(MTL_ID(), "ParamCharsType"));

}

int ObPlanCacheValue::assign_udr_infos(ObPlanCacheCtx &pc_ctx)

{

int ret = OB_SUCCESS;

if (OB_FAIL(dynamic_param_list_.assign(pc_ctx.dynamic_param_info_list_))) {

LOG_WARN("fail to assign dynamic param info list", K(ret));

} else if (OB_FAIL(tpl_sql_const_cons_.assign(pc_ctx.tpl_sql_const_cons_))) {

LOG_WARN("failed to assign tpl sql const cons", K(ret));

} else {

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

NotParamInfoList &not_param_list = tpl_sql_const_cons_.at(i);

std::sort(not_param_list.begin(), not_param_list.end(),

[](NotParamInfo &l, NotParamInfo &r) { return (l.idx_ < r.idx_); });

for (int64_t j = 0; OB_SUCC(ret) && j < not_param_list.count(); ++j) {

if (OB_FAIL(ob_write_string(*pc_alloc_,

not_param_list.at(j).raw_text_,

not_param_list.at(j).raw_text_))) {

LOG_WARN("deep_copy_obj failed", K(i), K(not_param_list.at(j)));

}

}

}

}

return ret;

}

void ObPlanCacheValue::reset_tpl_sql_const_cons()

{

for (int64_t i = 0; i < tpl_sql_const_cons_.count(); ++i) {

NotParamInfoList &not_param_list = tpl_sql_const_cons_.at(i);

for (int64_t j = 0; j < not_param_list.count(); ++j) {

pc_alloc_->free(not_param_list.at(j).raw_text_.ptr());

not_param_list.at(j).raw_text_.reset();

}

}

tpl_sql_const_cons_.reset();

}

int ObPlanCacheValue::init(ObPCVSet *pcv_set, const ObILibCacheObject *cache_obj, ObPlanCacheCtx &pc_ctx)

{

int ret = OB_SUCCESS;

ObMemAttr mem_attr;

const ObPlanCacheObject *plan = static_cast<const ObPlanCacheObject*>(cache_obj);

if (OB_ISNULL(pcv_set) || OB_ISNULL(plan) || OB_ISNULL(pcv_set->get_plan_cache())) {

ret = OB_INVALID_ARGUMENT;

LOG_WARN("invalid argument", K(pcv_set), K(plan));

} else if (FALSE_IT(mem_attr = pcv_set->get_plan_cache()->get_mem_attr())) {

// do nothing

} else if (OB_ISNULL(pc_alloc_ = pcv_set->get_allocator())) {

LOG_WARN("invalid argument", K(pcv_set->get_allocator()));

} else if (OB_ISNULL(pc_malloc_ = pcv_set->get_pc_allocator())) {

LOG_WARN("invalid argument", K(pcv_set->get_pc_allocator()));

} else if (OB_FAIL(outline_params_wrapper_.set_allocator(pc_alloc_, mem_attr))) {

LOG_WARN("fail to set outline param wrapper allocator", K(ret));

} else if (OB_NOT_NULL(pc_ctx.exec_ctx_.get_outline_params_wrapper())) {

if (OB_FAIL(set_outline_params_wrapper(*pc_ctx.exec_ctx_.get_outline_params_wrapper()))) {

LOG_WARN("fail to set outline params", K(ret));

}

} else if (OB_ISNULL(pc_ctx.sql_ctx_.session_info_) ||

OB_ISNULL(pc_ctx.sql_ctx_.schema_guard_)) {

ret = OB_INVALID_ARGUMENT;

LOG_WARN("invalid argument",

K(ret),

K(pc_ctx.sql_ctx_.session_info_),

K(pc_ctx.sql_ctx_.schema_guard_));

}

if (OB_SUCC(ret)) {

pcv_set_ = pcv_set;

//use_global_location_cache_ = !cache_obj->is_contain_virtual_table();

outline_state_ = plan->get_outline_state();

sys_schema_version_ = plan->get_sys_schema_version();

tenant_schema_version_ = plan->get_tenant_schema_version();

sql_traits_ = pc_ctx.sql_traits_;

#ifdef OB_BUILD_SPM

is_spm_closed_ = pcv_set->get_spm_closed();

#endif

stmt_type_ = plan->get_stmt_type();

need_param_ = plan->need_param();

is_nested_sql_ = ObSQLUtils::is_nested_sql(&pc_ctx.exec_ctx_);

is_batch_execute_ = pc_ctx.sql_ctx_.is_batch_params_execute();

has_dynamic_values_table_ = pc_ctx.exec_ctx_.has_dynamic_values_table();

MEMCPY(sql_id_, pc_ctx.sql_ctx_.sql_id_, sizeof(pc_ctx.sql_ctx_.sql_id_));

if (OB_FAIL(not_param_index_.add_members2(pc_ctx.not_param_index_))) {

LOG_WARN("fail to add not param index members", K(ret));

} else if (OB_FAIL(neg_param_index_.add_members2(pc_ctx.neg_param_index_))) {

LOG_WARN("fail to add neg param index members", K(ret));

} else if (OB_FAIL(param_charset_type_.assign(pc_ctx.param_charset_type_))) {

LOG_WARN("fail to assign param charset type", K(ret));

} else if (OB_FAIL(must_be_positive_idx_.add_members2(pc_ctx.must_be_positive_index_))) {

LOG_WARN("failed to add bitset members", K(ret));

} else if (OB_FAIL(set_stored_schema_objs(plan->get_dependency_table(),

pc_ctx.sql_ctx_.schema_guard_))) {

LOG_WARN("failed to set stored schema objs",

K(ret), K(plan->get_dependency_table()), K(pc_ctx.sql_ctx_.schema_guard_));

} else if (OB_FAIL(assign_udr_infos(pc_ctx))) {

LOG_WARN("failed to assign user-defined rule infos", K(ret));

} else {

//deep copy special param raw text

if (PC_PS_MODE == pc_ctx.mode_ || PC_PL_MODE == pc_ctx.mode_) {

if (OB_FAIL(not_param_var_.assign(pc_ctx.not_param_var_))) {

LOG_WARN("fail to assign not param var", K(ret));

}

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

if (OB_FAIL(deep_copy_obj(*pc_alloc_,

not_param_var_.at(i).ps_param_,

not_param_var_.at(i).ps_param_))) {

LOG_WARN("deep_copy_obj failed", K(i), K(not_param_var_.at(i)));

}

}

} else {

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

if (OB_FAIL(not_param_info_.push_back(pc_ctx.not_param_info_.at(i)))) {

LOG_WARN("fail to push not_param_info", K(ret));

} else if (OB_FAIL(ob_write_string(*pc_alloc_,

not_param_info_.at(i).raw_text_,

not_param_info_.at(i).raw_text_))) {

LOG_WARN("fail to deep copy param raw text", K(ret));

}

} //for end

if (OB_SUCC(ret)) {

std::sort(not_param_info_.begin(), not_param_info_.end(),

[](NotParamInfo &l, NotParamInfo &r) { return (l.idx_ < r.idx_); });

}

}

//deep copy constructed sql

if (OB_SUCC(ret)) {

ObString outline_signature_str;

if (PC_PS_MODE == pc_ctx.mode_ || PC_PL_MODE == pc_ctx.mode_) {

outline_signature_str = pc_ctx.raw_sql_;

} else {

outline_signature_str = pc_ctx.sql_ctx_.spm_ctx_.bl_key_.constructed_sql_;

}

int64_t size = outline_signature_str.get_serialize_size();

if (0 == size) {

ret = OB_ERR_UNEXPECTED;

} else {

char *buf = NULL;

int64_t pos_s = 0;

if (OB_UNLIKELY(NULL == (buf = (char *)pc_alloc_->alloc(size)))) {

ret = OB_ALLOCATE_MEMORY_FAILED;

LOG_WARN("fail to alloc mem", K(ret));

} else if (OB_FAIL(outline_signature_str.serialize(buf, size, pos_s))) {

LOG_WARN("fail to serialize constructed_sql_", K(ret));

} else {

outline_signature_.assign_ptr(buf, static_cast<ObString::obstr_size_t>(pos_s));

}

}

}

// deep copy constructed_sql_, used for baseline;

if (OB_SUCC(ret)) {

if ((PC_PS_MODE == pc_ctx.mode_ || PC_PL_MODE == pc_ctx.mode_)

&& OB_FAIL(ob_write_string(*pc_alloc_, pc_ctx.raw_sql_, constructed_sql_))) {

LOG_WARN("fail to deep copy param raw text", K(ret));

} else if (!(PC_PS_MODE == pc_ctx.mode_ || PC_PL_MODE == pc_ctx.mode_)

&& OB_FAIL(ob_write_string(*pc_alloc_,

pc_ctx.sql_ctx_.spm_ctx_.bl_key_.constructed_sql_,

constructed_sql_))) {

LOG_WARN("failed to write string", K(ret));

}

}

// deep copy raw_sql if not need to param

if (OB_SUCC(ret) && !need_param_ &&

OB_FAIL(ob_write_string(*pc_alloc_, pc_ctx.raw_sql_, raw_sql_))) {

LOG_WARN("failed to write raw sql", K(ret));

}

// set sessid_ if necessary

if (OB_SUCC(ret)) {

if (is_contain_tmp_tbl()) {

//临时表的行为取决于用户创建的session，而对于远程执行而言，远程的session id是一个临时的session_id

//因此这里统一应该使用master session id，来保证匹配计划一直使用的是用户session

sessid_ = pc_ctx.sql_ctx_.session_info_->get_sessid_for_table();

sess_create_time_ = pc_ctx.sql_ctx_.session_info_->get_sess_create_time();

// 获取临时表的表名

pc_ctx.tmp_table_names_.reset();

if (OB_FAIL(get_tmp_depend_tbl_names(pc_ctx.tmp_table_names_))) {

LOG_WARN("failed to get tmp depend tbl ids", K(ret));

} else {

// do nothing

}

} else { /* do nothing */ }

}

}

}

return ret;

}

int ObPlanCacheValue::match_all_params_info(ObPlanSet *batch_plan_set,

ObPlanCacheCtx &pc_ctx,

int64_t outline_param_idx,

bool &is_same)

{

int ret = OB_SUCCESS;

is_same = true;

bool is_batched_multi_stmt = pc_ctx.sql_ctx_.is_batch_params_execute();

ParamStore *params = pc_ctx.fp_result_.cache_params_;

if (is_batched_multi_stmt) {

ObArenaAllocator tmp_alloc;

ParamStore param_store((ObWrapperAllocator(tmp_alloc)));

int64_t query_cnt = pc_ctx.sql_ctx_.get_batch_params_count();

ParamStore *ab_params = pc_ctx.ab_params_;

ObPhysicalPlanCtx *phy_ctx = pc_ctx.exec_ctx_.get_physical_plan_ctx();

if (OB_ISNULL(ab_params) || OB_ISNULL(phy_ctx)) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("null ptr unexpected", K(ret), K(phy_ctx), K(ab_params));

} else if (query_cnt <= 1) {

ret = OB_BATCHED_MULTI_STMT_ROLLBACK;

LOG_TRACE("unexpected query count", K(ret), K(query_cnt));

} else if (batch_plan_set->get_can_skip_params_match()) {

// can_skip_params_match_ is true, we not need match all params,

// only call match_params_info once

param_store.reuse();

phy_ctx->reset_datum_param_store();

phy_ctx->set_original_param_cnt(ab_params->count());

if (OB_FAIL(get_one_group_params(0, *ab_params, param_store))) {

LOG_WARN("fail to get one params", K(ret));

} else if (OB_FAIL(pc_ctx.fp_result_.cache_params_->assign(param_store))) {

LOG_WARN("assign params failed", K(ret), K(param_store));

} else if (OB_FAIL(batch_plan_set->match_params_info(params, pc_ctx, outline_param_idx, is_same))) {

LOG_WARN("fail to match_params_info", K(ret), K(outline_param_idx), KPC(params));

} else {

// not match this plan, try match next plan

}

} else {

// pc_ctx.fp_result_.cache_params_ 在batch场景下，cache_params_不应该为null

// When the first set of parameters cannot match the plan_set parameter requirements,

// -5787 cannot be returned and the current batch optimization cannot be rolled back.

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

param_store.reuse();

phy_ctx->reset_datum_param_store();

phy_ctx->set_original_param_cnt(ab_params->count());

if (OB_FAIL(get_one_group_params(i, *ab_params, param_store))) {

LOG_WARN("fail to get one params", K(ret), K(i));

} else if (OB_FAIL(pc_ctx.fp_result_.cache_params_->assign(param_store))) {

LOG_WARN("assign params failed", K(ret), K(param_store));

} else if (OB_FAIL(phy_ctx->init_datum_param_store())) {

LOG_WARN("init datum_store failed", K(ret), K(param_store));

} else if (OB_FAIL(batch_plan_set->match_params_info(params, pc_ctx, outline_param_idx, is_same))) {

LOG_WARN("fail to match_params_info", K(ret), K(outline_param_idx), KPC(params));

} else if (i != 0 && !is_same) {

ret = OB_BATCHED_MULTI_STMT_ROLLBACK;

LOG_TRACE("params is not same type", K(param_store), K(i));

}

}

}

if (OB_SUCC(ret) && is_same) {

phy_ctx->reset_datum_param_store();

phy_ctx->set_original_param_cnt(ab_params->count());

if (OB_FAIL(batch_plan_set->copy_param_flag_from_param_info(ab_params))) {

LOG_WARN("failed copy param flag", K(ret), KPC(ab_params));

} else if (OB_FAIL(phy_ctx->get_param_store_for_update().assign(*ab_params))) {

LOG_WARN("failed to assign params_store", K(ret), K(*ab_params));

} else if (OB_FAIL(phy_ctx->init_datum_param_store())) {

LOG_WARN("failed to init datum store", K(ret));

}

}

} else {

if (OB_FAIL(batch_plan_set->match_params_info(params, pc_ctx, outline_param_idx, is_same))) {

LOG_WARN("fail to match_params_info", K(ret), K(outline_param_idx));

}

}

return ret;

}

int ObPlanCacheValue::choose_plan(ObPlanCacheCtx &pc_ctx,

const ObIArray<PCVSchemaObj> &schema_array,

ObPlanCacheObject *&plan_out)

{

int ret = OB_SUCCESS;

bool is_old_version = false;

plan_out = NULL;

ObSQLSessionInfo *session = NULL;

uint64_t tenant_id = OB_INVALID_ID;

ObPlanCacheObject *plan = NULL;

int64_t outline_param_idx = 0;

//检查在pcv中缓存的该sql涉及的view 及 table的version，

//如果不为最新的,在plan cache层会删除该value，并重新add plan

//TODO shengle 此处拷贝需要想办法处理掉

bool enable_baseline = false;

bool captrue_baseline = false;

bool need_check_schema = (schema_array.count() != 0);

if (schema_array.count() == 0 && stored_schema_objs_.count() == 0) {

need_check_schema = true;

}

if (stmt::T_NONE == pc_ctx.sql_ctx_.stmt_type_) {

//sql_ctx_.stmt_type_ != stmt::T_NONE means this calling in nested sql,

//can't cover the first stmt type in sql context

pc_ctx.sql_ctx_.stmt_type_ = stmt_type_;

}

if (OB_ISNULL(session = pc_ctx.exec_ctx_.get_my_session())) {

ret = OB_ERR_UNEXPECTED;

SQL_PC_LOG(ERROR, "got session is NULL", K(ret));

} else if (FALSE_IT(session->set_stmt_type(stmt_type_))) {

} else if (OB_FAIL(session->get_use_plan_baseline(enable_baseline))) {

LOG_WARN("fail to get use plan baseline", K(ret));

} else if (OB_FAIL(session->get_capture_plan_baseline(captrue_baseline))) {

LOG_WARN("failed to capture plan baseline", K(ret));

} else if (enable_baseline || captrue_baseline) {

if (OB_FAIL(ob_write_string(pc_ctx.allocator_,

constructed_sql_,

pc_ctx.sql_ctx_.spm_ctx_.bl_key_.constructed_sql_))) {

LOG_WARN("fail to deep copy constructed_sql_", K(ret));

} else if (OB_FAIL(ob_write_string(pc_ctx.allocator_,

common::ObString(OB_MAX_SQL_ID_LENGTH, sql_id_),

pc_ctx.sql_ctx_.spm_ctx_.bl_key_.sql_id_))) {

LOG_WARN("fail to deep copy sql_id_", K(ret));

}

}

if (OB_FAIL(ret)) {

// do nothing

} else if (OB_INVALID_ID == (tenant_id = session->get_effective_tenant_id())) {

ret = OB_ERR_UNEXPECTED;

SQL_PC_LOG(ERROR, "got effective tenant id is invalid", K(ret));

} else if (OB_FAIL(check_value_version_for_get(pc_ctx.sql_ctx_.schema_guard_,

need_check_schema,

schema_array,

tenant_id,

is_old_version))) {

SQL_PC_LOG(WARN, "fail to check table version", K(ret));

} else if (true == is_old_version) {

ret = OB_OLD_SCHEMA_VERSION;

SQL_PC_LOG(TRACE, "view or table is old version", K(ret));

}

if (OB_FAIL(ret)) {

//do nothing

} else {

ParamStore *params = pc_ctx.fp_result_.cache_params_;

//init param store

if (OB_LIKELY(pc_ctx.sql_ctx_.is_batch_params_execute())) {

if (OB_FAIL(resolve_multi_stmt_params(pc_ctx))) {

if (OB_BATCHED_MULTI_STMT_ROLLBACK != ret) {

LOG_WARN("failed to resolver row params", K(ret));

}

}

} else if (OB_UNLIKELY(pc_ctx.exec_ctx_.has_dynamic_values_table())) {

if (OB_FAIL(ObValuesTableCompression::resolve_params_for_values_clause(pc_ctx, stmt_type_,

not_param_info_, param_charset_type_, neg_param_index_, not_param_index_,

must_be_positive_idx_, params))) {

LOG_WARN("failed to resolve_params_for_values_clause ", K(ret));

}

} else if (OB_FAIL(resolver_params(pc_ctx,

stmt_type_,

param_charset_type_,

neg_param_index_,

not_param_index_,

must_be_positive_idx_,

pc_ctx.fp_result_.raw_params_,

params))) {

LOG_WARN("fail to resolver raw params", K(ret));

}

// cons user-defined rule param store

if (OB_SUCC(ret)) {

ParamStore param_store( (ObWrapperAllocator(pc_ctx.allocator_)) );

if (OB_FAIL(ObUDRUtils::cons_udr_param_store(dynamic_param_list_, pc_ctx, param_store))) {

LOG_WARN("failed to construct user-defined rule param store", K(ret));

}

}

if (OB_SUCC(ret)) {

ObPhysicalPlanCtx *phy_ctx = pc_ctx.exec_ctx_.get_physical_plan_ctx();

if (NULL != phy_ctx) {

phy_ctx->set_original_param_cnt(phy_ctx->get_param_store().count());

if (OB_FAIL(phy_ctx->init_datum_param_store())) {

LOG_WARN("fail to init datum param store", K(ret));

}

}

}

if (OB_FAIL(ret)) {

//do nothing

} else if (OB_FAIL(get_outline_param_index(pc_ctx.exec_ctx_,

outline_param_idx))) {//need use param store

LOG_WARN("fail to judge concurrent limit sql", K(ret));

} else {

// 这里记录org_param_count, 用于在匹配计划结束后, 如果没有匹配成功,

// 则将param store中参数恢复到当前状态(plan_set中match会进行可计算表

// 达式计算并将结果加入到param store) 避免本次失败后, 上层进行重试时,

// param store参数变多导致再匹配时, 因参数个数不符合预期报错

int64_t org_param_count = 0;

if (OB_NOT_NULL(params)) {

org_param_count = params->count();

}

DLIST_FOREACH(plan_set, plan_sets_) {

plan = NULL;

bool is_same = false;

if (OB_FAIL(match_all_params_info(plan_set, pc_ctx, outline_param_idx, is_same))) {

SQL_PC_LOG(WARN, "fail to match params info", K(ret));

} else if (!is_same) { //do nothing

LOG_TRACE("params info does not match", KPC(params));

} else {

if (OB_FAIL(plan_set->select_plan(pc_ctx, plan))) {

if (OB_SQL_PC_NOT_EXIST == ret) {

ret = OB_SUCCESS;

} else {

SQL_PC_LOG(TRACE, "failed to select plan in plan set", K(ret));

}

} else if (NULL != params) {

// set res map rule

uint64_t rule_id = plan_set->res_map_rule_id_;

int64_t param_idx = plan_set->res_map_rule_param_idx_;

uint64_t tenant_id = OB_INVALID_ID;

ObString param_text;

ObCollationType cs_type = CS_TYPE_INVALID;

if (rule_id != OB_INVALID_ID && param_idx != OB_INVALID_INDEX

&& pc_ctx.sql_ctx_.enable_sql_resource_manage_) {

if (OB_UNLIKELY(param_idx < 0 || param_idx >= params->count())) {

ret = OB_ERR_UNEXPECTED;

LOG_ERROR("unexpected res map rule param idx", K(ret), K(rule_id), K(param_idx), K(params->count()));

} else if (OB_FAIL(session->get_collation_connection(cs_type))) {

LOG_WARN("get collation connection failed", K(ret));

} else if (OB_INVALID_ID == (tenant_id = session->get_effective_tenant_id())) {

ret = OB_ERR_UNEXPECTED;

SQL_PC_LOG(ERROR, "got effective tenant id is invalid", K(ret));

} else if (OB_FAIL(ObObjCaster::get_obj_param_text(

params->at(plan_set->res_map_rule_param_idx_),

pc_ctx.raw_sql_, pc_ctx.allocator_,

cs_type, param_text))) {

LOG_WARN("get obj param text failed", K(ret));

} else {

uint64_t group_id = G_RES_MGR.get_col_mapping_rule_mgr().get_column_mapping_group_id(

tenant_id,

plan_set->res_map_rule_id_,

session->get_user_name(),

param_text);

if (OB_INVALID_ID == group_id) {

// OB_INVALID_ID means current user+param_value is not defined in mapping rule,

// get group_id according to current user.

if (OB_FAIL(G_RES_MGR.get_mapping_rule_mgr().get_group_id_by_user(

tenant_id, session->get_user_id(), group_id))) {

LOG_WARN("get group id by user failed", K(ret));

} else if (OB_INVALID_ID == group_id) {

// if not set consumer_group for current user, use OTHER_GROUP by default.

group_id = 0;

}

}

if (OB_SUCC(ret)) {

session->set_expect_group_id(group_id);

if (group_id == THIS_WORKER.get_group_id()) {

// do nothing if equals to current group id.

} else if (session->get_is_in_retry()

&& OB_NEED_SWITCH_CONSUMER_GROUP

== session->get_retry_info().get_last_query_retry_err()) {

LOG_ERROR("use unexpected group when retry, maybe set packet retry failed before",

K(group_id), K(THIS_WORKER.get_group_id()), K(rule_id), K(param_idx));

} else {

ret = OB_NEED_SWITCH_CONSUMER_GROUP;

}

LOG_TRACE("get expect rule id", K(ret), K(group_id),

K(THIS_WORKER.get_group_id()), K(session->get_expect_group_id()),

K(pc_ctx.raw_sql_));

}

}

}

break; //这个地方建议保留，如果去掉，需要另外加标记在for()中判断，并且不使用上面的for循环的宏；

}

}

if (NULL == plan

&& OB_NOT_NULL(params)

&& (params->count() > org_param_count)) {

ObPhysicalPlanCtx * phy_ctx = pc_ctx.exec_ctx_.get_physical_plan_ctx();

if (NULL == phy_ctx) {

ret = OB_INVALID_ARGUMENT;

LOG_WARN("physical ctx is null", K(ret));

}

for (int64_t i = params->count(); OB_SUCC(ret) && i > org_param_count; --i) {

params->pop_back();

phy_ctx->get_datum_param_store().pop_back();

// todo: param_frame_ptrs_ and other params in physical_plan_ctx need be reset

}

}

} // end for

}

}

if (NULL == plan && OB_SUCC(ret)) {

ret = OB_SQL_PC_NOT_EXIST;

}

if (OB_SUCC(ret)) {

plan_out = plan;

pc_ctx.sql_traits_ = sql_traits_; //used for check read only

}

return ret;

}

//将fast parser参数化的参数转化为ObObjParam, 用于get plan 时

int ObPlanCacheValue::resolver_params(ObPlanCacheCtx &pc_ctx,

const stmt::StmtType stmt_type,

const ObIArray<ObCharsetType> &param_charset_type,

const ObBitSet<> &neg_param_index,

const ObBitSet<> &not_param_index,

const ObBitSet<> &must_be_positive_idx,

ObIArray<ObPCParam *> &raw_params,

ParamStore *obj_params)

{

int ret = OB_SUCCESS;

ObSQLSessionInfo *session = pc_ctx.exec_ctx_.get_my_session();

ObPhysicalPlanCtx *phy_ctx = pc_ctx.exec_ctx_.get_physical_plan_ctx();

const int64_t raw_param_cnt = raw_params.count();

ObObjParam value;

bool enable_decimal_int = false;

if (OB_ISNULL(session) || OB_ISNULL(phy_ctx)) {

ret = OB_INVALID_ARGUMENT;

SQL_PC_LOG(WARN, "invalid argument", K(ret), KP(session), KP(phy_ctx));

} else if (obj_params == NULL || PC_PS_MODE == pc_ctx.mode_ || PC_PL_MODE == pc_ctx.mode_) {

/* do nothing */

} else if (OB_UNLIKELY(raw_param_cnt != param_charset_type.count())) {

ret = OB_INVALID_ARGUMENT;

SQL_PC_LOG(WARN, "raw_params and param_charset_type count is different", K(ret),

K(raw_param_cnt), K(param_charset_type.count()), K(pc_ctx.raw_sql_));

} else if (OB_FAIL(ObSQLUtils::check_enable_decimalint(session, enable_decimal_int))) {

LOG_WARN("fail to check enable decimal int", K(ret));

} else {

CHECK_COMPATIBILITY_MODE(session);

ObCollationType collation_connection = static_cast<ObCollationType>(session->get_local_collation_connection());

(void)obj_params->reserve(raw_param_cnt);

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

bool is_param = false;

if (OB_FAIL(ObResolverUtils::resolver_param(pc_ctx, *session, phy_ctx->get_param_store_for_update(), stmt_type,

param_charset_type.at(i), neg_param_index, not_param_index, must_be_positive_idx,

raw_params.at(i), i, value, is_param, enable_decimal_int))) {

SQL_PC_LOG(WARN, "failed to resolver param", K(ret), K(i));

} else if (is_param && OB_FAIL(obj_params->push_back(value))) {

SQL_PC_LOG(WARN, "fail to push item to array", K(ret));

} else {/* do nothing */}

}

}

return ret;

}

int ObPlanCacheValue::before_resolve_array_params(ObPlanCacheCtx &pc_ctx, int64_t query_num, int64_t param_num, ParamStore *&ab_params)

{

int ret = OB_SUCCESS;

if (OB_ISNULL(ab_params = static_cast<ParamStore *>(pc_ctx.allocator_.alloc(sizeof(ParamStore))))) {

ret = OB_ALLOCATE_MEMORY_FAILED;

LOG_WARN("failed to allocate memory", K(ret));

} else if (FALSE_IT(ab_params = new(ab_params)ParamStore(ObWrapperAllocator(pc_ctx.allocator_)))) {

// do nothing

} else if (OB_FAIL(ObSQLUtils::create_multi_stmt_param_store(pc_ctx.allocator_,

query_num,

param_num,

*ab_params))) {

LOG_WARN("failed to create multi_stmt_param_store", K(ret));

}

return ret;

}

int ObPlanCacheValue::resolve_multi_stmt_params(ObPlanCacheCtx &pc_ctx)

{

int ret = OB_SUCCESS;

ParamStore *ab_params = NULL;

bool is_valid = true;

if (pc_ctx.sql_ctx_.is_do_insert_batch_opt()) {

int64_t query_num = pc_ctx.sql_ctx_.get_insert_batch_row_cnt();

int64_t param_num = pc_ctx.fp_result_.raw_params_.count() - not_param_info_.count();

if (!not_param_info_.empty() &&

OB_FAIL(check_insert_multi_values_param(pc_ctx, is_valid))) {

LOG_WARN("failed to check multi insert param value", K(ret));

} else if (!is_valid) {

ret = OB_BATCHED_MULTI_STMT_ROLLBACK;

LOG_TRACE("batched multi_stmt needs rollback", K(ret));

} else if (OB_FAIL(before_resolve_array_params(pc_ctx, query_num, param_num, ab_params))) {

LOG_WARN("fail to prepare resolve params", K(ret));

} else if (OB_ISNULL(ab_params)) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("unexpected null", K(ret));

} else if (resolve_insert_multi_values_param(pc_ctx,

stmt_type_,

param_charset_type_,

neg_param_index_,

not_param_index_,

must_be_positive_idx_,

param_num,

*ab_params)) {

} else {

pc_ctx.ab_params_ = ab_params;

}

} else if (!pc_ctx.sql_ctx_.multi_stmt_item_.is_ab_batch_opt()) {

int64_t query_num = pc_ctx.multi_stmt_fp_results_.count();

int64_t param_num = pc_ctx.fp_result_.raw_params_.count() - not_param_info_.count();

// check whether all the values are the same

// 1、创建param_store指针

if (!not_param_info_.empty() &&

OB_FAIL(check_multi_stmt_not_param_value(pc_ctx.multi_stmt_fp_results_,

not_param_info_,

is_valid))) {

LOG_WARN("failed to check multi stmt not param value", K(ret));

} else if (!is_valid) {

ret = OB_BATCHED_MULTI_STMT_ROLLBACK;

LOG_TRACE("batched multi_stmt needs rollback", K(ret));

} else if (OB_FAIL(before_resolve_array_params(pc_ctx, query_num, param_num, ab_params))) {

LOG_WARN("fail to prepare resolve params", K(ret));

} else if (OB_ISNULL(ab_params)) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("unexpected null", K(ret));

} else if (OB_FAIL(check_multi_stmt_param_type(pc_ctx,

stmt_type_,

param_charset_type_,

neg_param_index_,

not_param_index_,

must_be_positive_idx_,

*ab_params))) {

LOG_WARN("failed to check multi stmt param type", K(ret));

} else {

pc_ctx.ab_params_ = ab_params;

}

}

return ret;

}

int ObPlanCacheValue::resolve_insert_multi_values_param(ObPlanCacheCtx &pc_ctx,

const stmt::StmtType stmt_type,

const ObIArray<ObCharsetType> &param_charset_type,

const ObBitSet<> &neg_param_index,

const ObBitSet<> &not_param_index,

const ObBitSet<> &must_be_positive_idx,

int64_t params_num,

ParamStore &param_store)

{

int ret = OB_SUCCESS;

ObArenaAllocator tmp_alloc;

ObRawParams *raw_param_array = nullptr;

ParamStore temp_obj_params((ObWrapperAllocator(tmp_alloc)));

ParamStore first_obj_params((ObWrapperAllocator(tmp_alloc)));

int64_t query_num = pc_ctx.sql_ctx_.get_insert_batch_row_cnt();

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

raw_param_array = nullptr;

temp_obj_params.reuse();

if (OB_ISNULL(raw_param_array = pc_ctx.insert_batch_opt_info_.multi_raw_params_.at(i))) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("unexpected null", K(ret));

} else if (OB_FAIL(resolver_params(pc_ctx,

stmt_type,

param_charset_type,

neg_param_index,

not_param_index,

must_be_positive_idx,

*raw_param_array,

&temp_obj_params))) {

LOG_WARN("failed to resolve parames", K(ret));

} else {

LOG_DEBUG("print one insert temp_obj_params",

K(temp_obj_params), K(params_num), K(query_num), K(pc_ctx.not_param_info_));

}

if (OB_SUCC(ret) && i == 0) {

if (OB_FAIL(first_obj_params.assign(temp_obj_params))) {

LOG_WARN("fail to assign params", K(ret));

}

// set type and external type

for (int64_t j = 0; OB_SUCC(ret) && j < params_num; j++) {

ObSqlArrayObj *array_params = nullptr;

if (OB_UNLIKELY(!param_store.at(j).is_ext_sql_array())) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("param store object is invalid", K(ret), K(param_store.at(j)));

} else if (OB_ISNULL(array_params = reinterpret_cast<ObSqlArrayObj*>(param_store.at(j).get_ext()))) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("get unexpected null", K(ret));

} else {

array_params->element_.set_meta_type(temp_obj_params.at(j).get_meta());

array_params->element_.set_accuracy(temp_obj_params.at(j).get_accuracy());

}

} // end init accuracy

}

// copy data

for (int64_t j = 0; OB_SUCC(ret) && j < params_num; j++) {

ObSqlArrayObj *array_params = nullptr;

if (OB_UNLIKELY(!param_store.at(j).is_ext_sql_array())) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("param object is invalid", K(ret), K(param_store.at(j)));

} else if (OB_ISNULL(array_params =

reinterpret_cast<ObSqlArrayObj*>(param_store.at(j).get_ext()))

|| OB_ISNULL(array_params->data_)) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("get unexpected null", K(ret), KPC(array_params));

} else {

array_params->data_[i] = temp_obj_params.at(j);

}

} // end copy

}

return ret;

}

int ObPlanCacheValue::check_multi_stmt_param_type(ObPlanCacheCtx &pc_ctx,

stmt::StmtType stmt_type,

const ObIArray<ObCharsetType> &param_charset_type,

const ObBitSet<> &neg_param_index,

const ObBitSet<> &not_param_index,

const ObBitSet<> &must_be_positive_idx,

ParamStore &param_store)

{

int ret = OB_SUCCESS;

ObArenaAllocator tmp_alloc;

ParamStore temp_obj_params((ObWrapperAllocator(tmp_alloc)));

ParamStore first_obj_params((ObWrapperAllocator(tmp_alloc)));

int64_t query_num = pc_ctx.multi_stmt_fp_results_.count();

int64_t param_num = pc_ctx.fp_result_.raw_params_.count() - not_param_index.num_members();

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

temp_obj_params.reuse();

if (OB_FAIL(resolver_params(pc_ctx,

stmt_type,

param_charset_type,

neg_param_index,

not_param_index,

must_be_positive_idx,

pc_ctx.multi_stmt_fp_results_.at(i).raw_params_,

&temp_obj_params))) {

LOG_WARN("failed to resolve parames", K(ret));

} else if (i == 0) {

ret = first_obj_params.assign(temp_obj_params);

// set type and external type

for (int64_t j = 0; OB_SUCC(ret) && j < param_num; j++) {

ObSqlArrayObj *array_params = nullptr;

if (OB_UNLIKELY(!param_store.at(j).is_ext_sql_array())) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("param store object is invalid", K(ret), K(param_store.at(j)));

} else if (OB_ISNULL(array_params = reinterpret_cast<ObSqlArrayObj*>(param_store.at(j).get_ext()))) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("get unexpected null", K(ret));

} else {

array_params->element_.set_meta_type(temp_obj_params.at(j).get_meta());

array_params->element_.set_accuracy(temp_obj_params.at(j).get_accuracy());

}

}

}

// copy data

for (int64_t j = 0; OB_SUCC(ret) && j < param_num; j++) {

ObSqlArrayObj *array_params = nullptr;

if (OB_UNLIKELY(!param_store.at(j).is_ext_sql_array())) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("param object is invalid", K(ret), K(param_store.at(j)));

} else if (OB_ISNULL(array_params =

reinterpret_cast<ObSqlArrayObj*>(param_store.at(j).get_ext()))

|| OB_ISNULL(array_params->data_)) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("get unexpected null", K(ret), KPC(array_params));

} else {

array_params->data_[i] = temp_obj_params.at(j);

}

}

}

return ret;

}

int ObPlanCacheValue::check_multi_stmt_not_param_value(

const ObIArray<ObFastParserResult> &multi_stmt_fp_results,

const ObIArray<NotParamInfo> &not_param_info,

bool &is_same)

{

int ret = OB_SUCCESS;

is_same = true;

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

if (OB_FAIL(check_not_param_value(multi_stmt_fp_results.at(i),

not_param_info,

is_same))) {

LOG_WARN("failed to check not param value", K(ret));

} else { /*do nothing*/ }

}

return ret;

}

int ObPlanCacheValue::check_insert_multi_values_param(ObPlanCacheCtx &pc_ctx, bool &is_same)

{

int ret = OB_SUCCESS;

ObRawParams *raw_params = nullptr;

int64_t count = pc_ctx.insert_batch_opt_info_.multi_raw_params_.count();

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

if (OB_ISNULL(raw_params = pc_ctx.insert_batch_opt_info_.multi_raw_params_.at(i))) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("unexpected null ptr", K(ret));

} else if (OB_FAIL(check_not_param_value(*raw_params,

pc_ctx.not_param_info_,

is_same))) {

LOG_WARN("failed to check not param value", K(ret));

} else { /*do nothing*/ }

}

return ret;

}

int ObPlanCacheValue::check_not_param_value(const ObIArray<ObPCParam *> &raw_params,

const ObIArray<NotParamInfo> &not_param_info,

bool &is_same)

{

int ret = OB_SUCCESS;

ParseNode *raw_param = NULL;

ObPCParam *pc_param = NULL;

is_same = true;

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

if (OB_FAIL(raw_params.at(not_param_info.at(i).idx_, pc_param))) {

LOG_WARN("fail to get raw params", K(not_param_info.at(i).idx_), K(ret));

} else if (OB_ISNULL(pc_param)) {

ret = OB_INVALID_ARGUMENT;

LOG_WARN("invalid argument", K(pc_param));

} else if (NULL == (raw_param = pc_param->node_)) {

ret = OB_INVALID_ARGUMENT;

LOG_WARN("invalid argument", K(raw_param));

} else if (0 != not_param_info.at(i).raw_text_.compare(

ObString(raw_param->text_len_, raw_param->raw_text_))) {

is_same = false;

LOG_TRACE("can't match not param info",

"raw value", ObString(raw_param->text_len_, raw_param->raw_text_),

"cached special value", not_param_info.at(i).raw_text_);

} else {