/**

* 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 "sql/plan_cache/ob_plan_cache.h"

#include "lib/container/ob_se_array_iterator.h"

#include "lib/profile/ob_perf_event.h"

#include "lib/json/ob_json_print_utils.h"

#include "lib/allocator/ob_mod_define.h"

#include "lib/alloc/alloc_func.h"

#include "lib/utility/ob_tracepoint.h"

#include "lib/allocator/page_arena.h"

#include "share/config/ob_server_config.h"

#include "share/ob_rpc_struct.h"

#include "share/ob_truncated_string.h"

#include "share/schema/ob_schema_getter_guard.h"

#include "lib/rc/ob_rc.h"

#include "observer/ob_server_struct.h"

#include "sql/plan_cache/ob_ps_cache_callback.h"

#include "sql/plan_cache/ob_ps_sql_utils.h"

#include "sql/ob_sql_context.h"

#include "sql/engine/ob_exec_context.h"

#include "sql/session/ob_sql_session_info.h"

#include "sql/engine/ob_physical_plan.h"

#include "sql/plan_cache/ob_plan_cache_callback.h"

#include "sql/plan_cache/ob_cache_object_factory.h"

#include "sql/udr/ob_udr_mgr.h"

#include "pl/ob_pl.h"

#include "pl/ob_pl_package.h"

#include "observer/ob_req_time_service.h"

#ifdef OB_BUILD_SPM

#include "sql/spm/ob_spm_define.h"

#include "sql/spm/ob_spm_controller.h"

#include "sql/spm/ob_spm_evolution_plan.h"

#endif

#include "pl/pl_cache/ob_pl_cache_mgr.h"

#include "sql/plan_cache/ob_values_table_compression.h"

using namespace oceanbase::common;

using namespace oceanbase::common::hash;

using namespace oceanbase::share::schema;

using namespace oceanbase::lib;

using namespace oceanbase::pl;

using namespace oceanbase::observer;

namespace oceanbase

{

namespace sql

{

struct ObGetPlanIdBySqlIdOp

{

explicit ObGetPlanIdBySqlIdOp(common::ObIArray<uint64_t> *key_array,

const common::ObString &sql_id,

const bool with_plan_hash,

const uint64_t &plan_hash_value)

: key_array_(key_array), sql_id_(sql_id), with_plan_hash_(with_plan_hash), plan_hash_value_(plan_hash_value)

{

}

int operator()(common::hash::HashMapPair<ObCacheObjID, ObILibCacheObject *> &entry)

{

int ret = common::OB_SUCCESS;

ObPhysicalPlan *plan = NULL;

if (OB_ISNULL(key_array_) || OB_ISNULL(entry.second)) {

ret = common::OB_NOT_INIT;

SQL_PC_LOG(WARN, "invalid argument", K(ret));

} else if (ObLibCacheNameSpace::NS_CRSR != entry.second->get_ns()) {

// not sql plan

// do nothing

} else if (OB_ISNULL(plan = dynamic_cast<ObPhysicalPlan *>(entry.second))) {

ret = OB_ERR_UNEXPECTED;

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

} else if (sql_id_ != plan->stat_.sql_id_) {

// do nothing

} else if (with_plan_hash_ && plan->stat_.plan_hash_value_ != plan_hash_value_) {

// do nothing

} else if (OB_FAIL(key_array_->push_back(entry.first))) {

SQL_PC_LOG(WARN, "fail to push back plan_id", K(ret));

}

return ret;

}

common::ObIArray<uint64_t> *key_array_;

common::ObString sql_id_;

bool with_plan_hash_;

uint64_t plan_hash_value_;

};

struct ObGetKVEntryByNsOp : public ObKVEntryTraverseOp

{

explicit ObGetKVEntryByNsOp(const ObLibCacheNameSpace ns,

LCKeyValueArray *key_val_list,

const CacheRefHandleID ref_handle)

: ObKVEntryTraverseOp(key_val_list, ref_handle),

namespace_(ns)

{

}

virtual int check_entry_match(LibCacheKVEntry &entry, bool &is_match)

{

int ret = OB_SUCCESS;

is_match = false;

if (namespace_ == entry.first->namespace_) {

is_match = true;

}

return ret;

}

ObLibCacheNameSpace namespace_;

};

struct ObGetKVEntryBySQLIDOp : public ObKVEntryTraverseOp

{

explicit ObGetKVEntryBySQLIDOp(uint64_t db_id,

common::ObString sql_id,

LCKeyValueArray *key_val_list,

const CacheRefHandleID ref_handle)

: ObKVEntryTraverseOp(key_val_list, ref_handle),

db_id_(db_id),

sql_id_(sql_id)

{

}

virtual int check_entry_match(LibCacheKVEntry &entry, bool &is_match)

{

int ret = OB_SUCCESS;

is_match = false;

if (entry.first->namespace_ == ObLibCacheNameSpace::NS_CRSR) {

ObPlanCacheKey *key = static_cast<ObPlanCacheKey*>(entry.first);

ObPCVSet *node = static_cast<ObPCVSet*>(entry.second);

if (db_id_ != common::OB_INVALID_ID && db_id_ != key->db_id_) {

// skip entry that has non-matched db_id

} else if (!contain_sql_id(node->get_sql_id())) {

// skip entry which not contains same sql_id

} else {

is_match = true;

}

}

return ret;

}

bool contain_sql_id(common::ObIArray<common::ObString> &sql_ids)

{

bool contains = false;

for (int64_t i = 0; !contains && i < sql_ids.count(); i++) {

if (sql_ids.at(i) == sql_id_) {

contains = true;

}

}

return contains;

}

uint64_t db_id_;

common::ObString sql_id_;

};

#ifdef OB_BUILD_SPM

struct ObGetPlanBaselineBySQLIDOp : public ObKVEntryTraverseOp

{

explicit ObGetPlanBaselineBySQLIDOp(uint64_t db_id,

common::ObString sql_id,

LCKeyValueArray *key_val_list,

const CacheRefHandleID ref_handle)

: ObKVEntryTraverseOp(key_val_list, ref_handle),

db_id_(db_id),

sql_id_(sql_id)

{

}

virtual int check_entry_match(LibCacheKVEntry &entry, bool &is_match)

{

int ret = OB_SUCCESS;

is_match = false;

if (ObLibCacheNameSpace::NS_SPM == entry.first->namespace_) {

ObBaselineKey *key = static_cast<ObBaselineKey*>(entry.first);

if (db_id_ != common::OB_INVALID_ID && db_id_ != key->db_id_) {

// skip entry that has non-matched db_id

} else if (sql_id_ == key->sql_id_) {

is_match = true;

}

}

return ret;

}

uint64_t db_id_;

common::ObString sql_id_;

};

#endif

struct ObGetPcvSetByTabNameOp : public ObKVEntryTraverseOp

{

explicit ObGetPcvSetByTabNameOp(uint64_t db_id, common::ObString tab_name,

LCKeyValueArray *key_val_list,

const CacheRefHandleID ref_handle)

: ObKVEntryTraverseOp(key_val_list, ref_handle),

db_id_(db_id),

tab_name_(tab_name)

{

}

virtual int check_entry_match(LibCacheKVEntry &entry, bool &is_match)

{

int ret = common::OB_SUCCESS;

is_match = false;

if (entry.first->namespace_ >= ObLibCacheNameSpace::NS_CRSR

&& entry.first->namespace_ <= ObLibCacheNameSpace::NS_PKG) {

ObPlanCacheKey *key = static_cast<ObPlanCacheKey*>(entry.first);

ObPCVSet *node = static_cast<ObPCVSet*>(entry.second);

if (db_id_ == common::OB_INVALID_ID) {

// do nothing

} else if (db_id_ != key->db_id_) {

// skip entry that has non-matched db_id

} else if (OB_FAIL(node->check_contains_table(db_id_, tab_name_, is_match))) {

LOG_WARN("fail to check table name", K(ret), K(db_id_), K(tab_name_));

} else {

// do nothing

}

}

return ret;

}

uint64_t db_id_;

common::ObString tab_name_;

};

#ifdef OB_BUILD_SPM

struct ObGetEvolutionTaskPcvSetOp : public ObKVEntryTraverseOp

{

explicit ObGetEvolutionTaskPcvSetOp(EvolutionPlanList *evo_task_list,

LCKeyValueArray *key_val_list,

const CacheRefHandleID ref_handle)

: ObKVEntryTraverseOp(key_val_list, ref_handle),

evo_task_list_(evo_task_list)

{

}

virtual int check_entry_match(LibCacheKVEntry &entry, bool &is_match)

{

int ret = common::OB_SUCCESS;

is_match = false;

if (entry.first->namespace_ == ObLibCacheNameSpace::NS_CRSR) {

ObPCVSet *node = static_cast<ObPCVSet*>(entry.second);

int64_t origin_count = evo_task_list_->count();

if (OB_FAIL(node->get_evolving_evolution_task(*evo_task_list_))) {

LOG_WARN("failed to get evolving evolution task", K(ret));

} else {

is_match = evo_task_list_->count() > origin_count;

}

}

return ret;

}

EvolutionPlanList *evo_task_list_;

};

#endif

struct ObGetTableIdOp

{

explicit ObGetTableIdOp(uint64_t table_id)

: table_id_(table_id)

{}

int operator()(common::hash::HashMapPair<ObCacheObjID, ObILibCacheObject *> &entry)

{

int ret = common::OB_SUCCESS;

ObPhysicalPlan *plan = NULL;

int64_t version = -1;

if (OB_ISNULL(entry.second)) {

ret = common::OB_NOT_INIT;

SQL_PC_LOG(WARN, "invalid argument", K(ret));

} else if (ObLibCacheNameSpace::NS_CRSR != entry.second->get_ns()) {

// not sql plan

// do nothing

} else if (OB_ISNULL(plan = dynamic_cast<ObPhysicalPlan *>(entry.second))) {

ret = OB_ERR_UNEXPECTED;

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

} else if (plan->get_base_table_version(table_id_, version)) {

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

} else if (version > 0) {

plan->set_is_expired(true);

}

return ret;

}

const uint64_t table_id_;

};

// true means entry_left is more active than entry_right

bool stat_compare(const LCKeyValue &left, const LCKeyValue &right)

{

bool cmp_ret = false;

if (OB_ISNULL(left.node_) || OB_ISNULL(right.node_)) {

cmp_ret = false;

SQL_PC_LOG_RET(ERROR, OB_INVALID_ARGUMENT, "invalid argument", KP(left.node_), KP(right.node_), K(cmp_ret));

} else if (OB_ISNULL(left.node_->get_node_stat())

|| OB_ISNULL(right.node_->get_node_stat())) {

cmp_ret = false;

SQL_PC_LOG_RET(ERROR, OB_INVALID_ARGUMENT, "invalid argument", K(left.node_->get_node_stat()),

K(right.node_->get_node_stat()), K(cmp_ret));

} else {

cmp_ret = left.node_->get_node_stat()->weight() < right.node_->get_node_stat()->weight();

}

return cmp_ret;

}

// filter entries satisfy stat_condition

// get 'evict_num' number of sql_ids which weight is lower;

struct ObNodeStatFilterOp : public ObKVEntryTraverseOp

{

ObNodeStatFilterOp(int64_t evict_num,

LCKeyValueArray *key_val_list,

const CacheRefHandleID ref_handle)

: ObKVEntryTraverseOp(key_val_list, ref_handle),

evict_num_(evict_num)

{

}

virtual int operator()(LibCacheKVEntry &entry)

{

int ret = common::OB_SUCCESS;

if (OB_ISNULL(key_value_list_) || OB_ISNULL(entry.first) || OB_ISNULL(entry.second)) {

ret = common::OB_INVALID_ARGUMENT;

SQL_PC_LOG(WARN, "invalid argument",

K(key_value_list_), K(entry.first), K(entry.second), K(ret));

} else {

ObLCKeyValue lc_kv(entry.first, entry.second);

if (key_value_list_->count() < evict_num_) {

if (OB_FAIL(key_value_list_->push_back(lc_kv))) {

SQL_PC_LOG(WARN, "fail to push into evict_array", K(ret));

} else {

lc_kv.node_->inc_ref_count(ref_handle_);

}

} else {

// replace one element

if (stat_compare(lc_kv, key_value_list_->at(0))) {

// pop_heap move the largest value to last

std::pop_heap(key_value_list_->begin(), key_value_list_->end(), stat_compare);

// remove and replace the largest value

LCKeyValue kv;

key_value_list_->pop_back(kv);

if (NULL != kv.node_) {

kv.node_->dec_ref_count(ref_handle_);

}

if (OB_FAIL(key_value_list_->push_back(lc_kv))) {

SQL_PC_LOG(WARN, "fail to push into evict_array", K(ret));

} else {

lc_kv.node_->inc_ref_count(ref_handle_);

}

}

}

// construct max-heap, the most active entry is on top

// std::push_heap arrange the last element, make evict_array max-heap

if (OB_SUCC(ret)) {

std::push_heap(key_value_list_->begin(), key_value_list_->end(), stat_compare);

}

}

return ret;

}

int64_t evict_num_;

};

ObPlanCache::ObPlanCache()

:inited_(false),

tenant_id_(OB_INVALID_TENANT_ID),

mem_limit_pct_(OB_PLAN_CACHE_PERCENTAGE),

mem_high_pct_(OB_PLAN_CACHE_EVICT_HIGH_PERCENTAGE),

mem_low_pct_(OB_PLAN_CACHE_EVICT_LOW_PERCENTAGE),

mem_used_(0),

bucket_num_(0),

inner_allocator_(),

ref_handle_mgr_(),

pcm_(NULL),

destroy_(0),

tg_id_(-1)

{

}

ObPlanCache::~ObPlanCache()

{

destroy();

}

void ObPlanCache::destroy()

{

observer::ObReqTimeGuard req_timeinfo_guard;

if (inited_) {

TG_DESTROY(tg_id_);

if (OB_SUCCESS != (cache_evict_all_obj())) {

SQL_PC_LOG_RET(WARN, OB_ERROR, "fail to evict all lib cache cache");

}

if (root_context_ != NULL) {

DESTROY_CONTEXT(root_context_);

root_context_ = NULL;

}

inited_ = false;

}

}

int ObPlanCache::init(int64_t hash_bucket, uint64_t tenant_id)

{

int ret = OB_SUCCESS;

if (!inited_) {

ObPCMemPctConf default_conf;

ObMemAttr attr(tenant_id, "PlanCache", ObCtxIds::PLAN_CACHE_CTX_ID);

lib::ContextParam param;

param.set_properties(lib::ADD_CHILD_THREAD_SAFE | lib::ALLOC_THREAD_SAFE)

.set_mem_attr(attr);

if (OB_FAIL(ROOT_CONTEXT->CREATE_CONTEXT(root_context_, param))) {

SQL_PC_LOG(WARN, "failed to create context", K(ret));

} else if (OB_FAIL(co_mgr_.init(hash_bucket, tenant_id))) {

SQL_PC_LOG(WARN, "failed to init lib cache manager", K(ret));

} else if (OB_FAIL(cache_key_node_map_.create(hash::cal_next_prime(hash_bucket),

ObModIds::OB_HASH_BUCKET_PLAN_CACHE,

ObModIds::OB_HASH_NODE_PLAN_CACHE,

tenant_id))) {

SQL_PC_LOG(WARN, "failed to init PlanCache", K(ret));

} else if (OB_FAIL(TG_CREATE_TENANT(lib::TGDefIDs::PlanCacheEvict, tg_id_))) {

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

} else if (OB_FAIL(TG_START(tg_id_))) {

LOG_WARN("failed to start tg", K(ret));

} else if (OB_FAIL(TG_SCHEDULE(tg_id_, evict_task_, GCONF.plan_cache_evict_interval, true))) {

LOG_WARN("failed to schedule refresh task", K(ret));

} else if (OB_FAIL(set_mem_conf(default_conf))) {

LOG_WARN("fail to set plan cache memory conf", K(ret));

} else {

evict_task_.plan_cache_ = this;

cn_factory_.set_lib_cache(this);

ObMemAttr attr = get_mem_attr();

attr.tenant_id_ = tenant_id;

inner_allocator_.set_attr(attr);

set_host(const_cast<ObAddr &>(GCTX.self_addr()));

bucket_num_ = hash::cal_next_prime(hash_bucket);

tenant_id_ = tenant_id;

ref_handle_mgr_.set_tenant_id(tenant_id_);

inited_ = true;

}

if (OB_FAIL(ret)) {

if (root_context_ != NULL) {

DESTROY_CONTEXT(root_context_);

root_context_ = NULL;

}

}

}

return ret;

}

int ObPlanCache::get_normalized_pattern_digest(const ObPlanCacheCtx &pc_ctx, uint64_t &pattern_digest)

{

int ret = OB_SUCCESS;

pattern_digest = 0;

if (pc_ctx.mode_ == PC_PS_MODE || pc_ctx.mode_ == PC_PL_MODE || pc_ctx.fp_result_.pc_key_.name_.empty()) {

ObFastParserResult fp_result;

ObSQLMode sql_mode = pc_ctx.sql_ctx_.session_info_->get_sql_mode();

ObCharsets4Parser charsets4parser = pc_ctx.sql_ctx_.session_info_->get_charsets4parser();

FPContext fp_ctx(charsets4parser);

fp_ctx.enable_batched_multi_stmt_ = pc_ctx.sql_ctx_.handle_batched_multi_stmt();

fp_ctx.sql_mode_ = sql_mode;

if (OB_FAIL(ObSqlParameterization::fast_parser(pc_ctx.allocator_,

fp_ctx,

pc_ctx.raw_sql_,

fp_result))) {

LOG_WARN("failed to fast parser", K(ret), K(sql_mode), K(pc_ctx.raw_sql_));

} else {

pattern_digest = fp_result.pc_key_.name_.hash();

}

} else {

pattern_digest = pc_ctx.fp_result_.pc_key_.name_.hash();

}

return ret;

}

int ObPlanCache::check_after_get_plan(int tmp_ret,

ObILibCacheCtx &ctx,

ObILibCacheObject *cache_obj)

{

int ret = tmp_ret;

ObPhysicalPlan *plan = NULL;

bool enable_udr = false;

bool need_late_compilation = false;

ObJITEnableMode jit_mode = ObJITEnableMode::OFF;

ObPlanCacheCtx &pc_ctx = static_cast<ObPlanCacheCtx&>(ctx);

if (cache_obj != NULL && ObLibCacheNameSpace::NS_CRSR == cache_obj->get_ns()) {

plan = static_cast<ObPhysicalPlan *>(cache_obj);

}

if (OB_SUCC(ret)) {

if (OB_ISNULL(pc_ctx.sql_ctx_.session_info_)) {

ret = OB_ERR_UNEXPECTED;

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

} else if (OB_FAIL(pc_ctx.sql_ctx_.session_info_->get_jit_enabled_mode(jit_mode))) {

LOG_WARN("failed to get jit mode");

} else {

enable_udr = pc_ctx.sql_ctx_.session_info_->enable_udr();

}

}

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

bool is_exists = false;

uint64_t pattern_digest = 0;

sql::ObUDRMgr *rule_mgr = MTL(sql::ObUDRMgr*);

// when the global rule version changes or enable_user_defined_rewrite_rules changes

// it is necessary to check whether the physical plan are expired

if ((plan->get_rule_version() != rule_mgr->get_rule_version()

|| plan->is_enable_udr() != enable_udr)) {

if (OB_FAIL(get_normalized_pattern_digest(pc_ctx, pattern_digest))) {

LOG_WARN("failed to calc normalized pattern digest", K(ret));

} else if (OB_FAIL(rule_mgr->fuzzy_check_by_pattern_digest(pattern_digest, is_exists))) {

LOG_WARN("failed to fuzzy check by pattern digest", K(ret));

} else if (is_exists || plan->is_rewrite_sql()) {

ret = OB_OLD_SCHEMA_VERSION;

LOG_TRACE("Obsolete user-defined rewrite rules require eviction plan", K(ret),

K(is_exists), K(pc_ctx.raw_sql_), K(plan->is_enable_udr()), K(enable_udr),

K(plan->is_rewrite_sql()), K(plan->get_rule_version()), K(rule_mgr->get_rule_version()));

} else {

plan->set_rule_version(rule_mgr->get_rule_version());

plan->set_is_enable_udr(enable_udr);

}

}

if (OB_SUCC(ret)) {

if (ObJITEnableMode::AUTO == jit_mode && // only use late compilation when jit_mode is auto

OB_FAIL(need_late_compile(plan, need_late_compilation))) {

LOG_WARN("failed to check for late compilation", K(ret));

} else {

// set context's need_late_compile_ for upper layer to proceed

pc_ctx.sql_ctx_.need_late_compile_ = need_late_compilation;

}

}

}

// if schema expired, update pcv set;

if (OB_OLD_SCHEMA_VERSION == ret

|| (plan != NULL && plan->is_expired())

|| need_late_compilation) {

if (plan != NULL && plan->is_expired()) {

LOG_INFO("the statistics of table is stale and evict plan.", K(plan->stat_));

}

if (OB_FAIL(remove_cache_node(pc_ctx.key_))) {

LOG_WARN("fail to remove pcv set when schema/plan expired", K(ret));

} else {

ret = OB_SQL_PC_NOT_EXIST;

}

}

return ret;

}

//plan cache获取plan入口

//1.fast parser获取param sql及raw params

//2.根据param sql获得pcv set

//3.检查权限信息

int ObPlanCache::get_plan(common::ObIAllocator &allocator,

ObPlanCacheCtx &pc_ctx,

ObCacheObjGuard& guard)

{

int ret = OB_SUCCESS;

FLTSpanGuard(pc_get_plan);

ObGlobalReqTimeService::check_req_timeinfo();

pc_ctx.handle_id_ = guard.ref_handle_;

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_.schema_guard_), K(pc_ctx.exec_ctx_.get_physical_plan_ctx()));

} else if (pc_ctx.sql_ctx_.multi_stmt_item_.is_batched_multi_stmt()) {

if (OB_FAIL(construct_multi_stmt_fast_parser_result(allocator,

pc_ctx))) {

if (OB_BATCHED_MULTI_STMT_ROLLBACK != ret) {

LOG_WARN("failed to construct multi stmt fast parser", K(ret));

}

} else {

pc_ctx.fp_result_ = pc_ctx.multi_stmt_fp_results_.at(0);

}

} else if (OB_FAIL(construct_fast_parser_result(allocator,

pc_ctx,

pc_ctx.raw_sql_,

pc_ctx.fp_result_ ))) {

LOG_WARN("failed to construct fast parser results", K(ret));

}

if (OB_SUCC(ret)) {

if (OB_FAIL(get_plan_cache(pc_ctx, guard))) {

SQL_PC_LOG(TRACE, "failed to get plan", K(ret), K(pc_ctx.fp_result_.pc_key_));

} else if (OB_ISNULL(guard.cache_obj_)

|| ObLibCacheNameSpace::NS_CRSR != guard.cache_obj_->get_ns()) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("cache obj is invalid", K(ret));

} else {

ObPhysicalPlan* plan = NULL;

if (OB_ISNULL(plan = static_cast<ObPhysicalPlan*>(guard.cache_obj_))) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("failed to cast cache obj to physical plan.", K(ret));

} else {

MEMCPY(pc_ctx.sql_ctx_.sql_id_,

plan->stat_.sql_id_.ptr(),

plan->stat_.sql_id_.length());

if (GCONF.enable_perf_event) {

uint64_t tenant_id = pc_ctx.sql_ctx_.session_info_->get_effective_tenant_id();

bool read_only = false;

if ((pc_ctx.sql_ctx_.session_info_->is_inner() && !pc_ctx.sql_ctx_.is_from_pl_)) {

// do nothing

} else if (OB_FAIL(pc_ctx.sql_ctx_.schema_guard_->get_tenant_read_only(tenant_id,

read_only))) {

LOG_WARN("failed to check read_only privilege", K(ret));

} else if (OB_FAIL(pc_ctx.sql_ctx_.session_info_->check_read_only_privilege(

read_only, pc_ctx.sql_traits_))) {

LOG_WARN("failed to check read_only privilege", K(ret));

}

}

}

}

}

if (OB_FAIL(ret) && OB_NOT_NULL(guard.cache_obj_)) {

co_mgr_.free(guard.cache_obj_, guard.ref_handle_);

guard.cache_obj_ = NULL;

}

return ret;

}

int ObPlanCache::construct_multi_stmt_fast_parser_result(common::ObIAllocator &allocator,

ObPlanCacheCtx &pc_ctx)

{

int ret = OB_SUCCESS;

const common::ObIArray<ObString> *queries = NULL;

bool enable_explain_batched_multi_statement = ObSQLUtils::is_enable_explain_batched_multi_statement();

if (OB_ISNULL(queries = pc_ctx.sql_ctx_.multi_stmt_item_.get_queries())) {

ret = OB_ERR_UNEXPECTED;

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

} else if (OB_FAIL(pc_ctx.multi_stmt_fp_results_.reserve(queries->count()))) {

LOG_WARN("failed to reserve array space", K(ret));

} else {

ObFastParserResult parser_result;

ObFastParserResult trimmed_parser_result;

for (int64_t i = 0; OB_SUCC(ret) && i < queries->count(); i++) {

parser_result.reset();

ObString trimed_stmt = const_cast<ObString &>(queries->at(i)).trim();

if (OB_FAIL(construct_fast_parser_result(allocator,

pc_ctx,

trimed_stmt,

parser_result))) {

LOG_WARN("failed to construct fast parser result", K(ret), K(i), K(trimed_stmt));

} else if (OB_FAIL(pc_ctx.multi_stmt_fp_results_.push_back(parser_result))) {

LOG_WARN("failed to push back parser result", K(ret));

} else if (i == 0 && enable_explain_batched_multi_statement) {

const char *p_normal_start = nullptr;

ObString trimmed_query;

if (ObParser::is_explain_stmt(queries->at(0), p_normal_start)) {

trimmed_query = queries->at(0).after(p_normal_start - 1);

if (OB_FAIL(construct_fast_parser_result(allocator,

pc_ctx,

trimmed_query,

trimmed_parser_result))) {

LOG_WARN("failed to construct fast parser result", K(ret));

}

}

} else if (i > 0 ) {

// Example query: 'explain update t1 set b=0 where a=1;explain update t1 set b=0 where a=2;'

// Original first query: 'explain update t1 set b=0 where a=1'

// Trimmed first query: 'update t1 set b=0 where a=1'

//

// If we compare the second query (e.g. 'explain update t1 set b=0 where a=2') with the original

// first query, the check will pass, but we will batch run one 'update' statement with one

// 'explain update' statement which makes no sense. Thus, we should use the trimmed first query instead.

ObFastParserResult &first_parser_result = enable_explain_batched_multi_statement ?

trimmed_parser_result : pc_ctx.multi_stmt_fp_results_.at(0);

if (parser_result.pc_key_.name_ != first_parser_result.pc_key_.name_

|| parser_result.raw_params_.count() != first_parser_result.raw_params_.count()) {

ret = OB_BATCHED_MULTI_STMT_ROLLBACK;

if (REACH_TIME_INTERVAL(10000000)) {

LOG_INFO("batched multi_stmt needs rollback",

K(parser_result.pc_key_),

K(parser_result.raw_params_.count()),

K(first_parser_result.pc_key_),

K(first_parser_result.raw_params_.count()),

K(ret), K(lbt()));

}

}

} else { /*do nothing*/ }

}

}

return ret;

}

int ObPlanCache::construct_fast_parser_result(common::ObIAllocator &allocator,

ObPlanCacheCtx &pc_ctx,

const common::ObString &raw_sql,

ObFastParserResult &fp_result)

{

int ret = OB_SUCCESS;

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

OB_ISNULL(pc_ctx.exec_ctx_.get_physical_plan_ctx())) {

ret = OB_ERR_UNEXPECTED;

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

} else {

ObSQLMode sql_mode = pc_ctx.sql_ctx_.session_info_->get_sql_mode();

ObCharsets4Parser charsets4parser = pc_ctx.sql_ctx_.session_info_->get_charsets4parser();

// if exact mode is on, not needs to do fast parser

bool enable_exact_mode = pc_ctx.sql_ctx_.session_info_->get_enable_exact_mode();

fp_result.cache_params_ =

&(pc_ctx.exec_ctx_.get_physical_plan_ctx()->get_param_store_for_update());

if (OB_FAIL(construct_plan_cache_key(*pc_ctx.sql_ctx_.session_info_,

ObLibCacheNameSpace::NS_CRSR,

fp_result.pc_key_,

pc_ctx.sql_ctx_.is_protocol_weak_read_))) {

LOG_WARN("failed to construct plan cache key", K(ret));

} else if (enable_exact_mode) {

(void)fp_result.pc_key_.name_.assign_ptr(raw_sql.ptr(), raw_sql.length());

} else {

FPContext fp_ctx(charsets4parser);

fp_ctx.enable_batched_multi_stmt_ = pc_ctx.sql_ctx_.handle_batched_multi_stmt();

fp_ctx.sql_mode_ = sql_mode;

bool can_do_batch_insert = false;

ObString first_truncated_sql;

int64_t batch_count = 0;

bool is_insert_values = false;

if (OB_FAIL(ObSqlParameterization::fast_parser(allocator,

fp_ctx,

raw_sql,

fp_result))) {

LOG_WARN("failed to fast parser", K(ret), K(sql_mode), K(pc_ctx.raw_sql_));

} else if (OB_FAIL(check_can_do_insert_opt(allocator,

pc_ctx,

fp_result,

can_do_batch_insert,

batch_count,

first_truncated_sql,

is_insert_values))) {

LOG_WARN("fail to do insert optimization", K(ret));

} else if (can_do_batch_insert) {

if (OB_FAIL(rebuild_raw_params(allocator,

pc_ctx,

fp_result,

batch_count))) {

LOG_WARN("fail to rebuild raw_param", K(ret), K(batch_count));

} else if (pc_ctx.insert_batch_opt_info_.multi_raw_params_.empty()) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("unexpected multi_raw_params, can't do batch insert opt, but not need to return error",

K(batch_count), K(first_truncated_sql), K(pc_ctx.raw_sql_), K(fp_result));

} else if (OB_ISNULL(pc_ctx.insert_batch_opt_info_.multi_raw_params_.at(0))) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("unexpected null ptr, can't do batch insert opt, but not need to return error",

K(batch_count), K(first_truncated_sql), K(pc_ctx.raw_sql_), K(fp_result));

} else {

fp_result.raw_params_.reset();

fp_result.raw_params_.set_allocator(&allocator);

fp_result.raw_params_.set_capacity(pc_ctx.insert_batch_opt_info_.multi_raw_params_.at(0)->count());

if (OB_FAIL(fp_result.raw_params_.assign(*pc_ctx.insert_batch_opt_info_.multi_raw_params_.at(0)))) {

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

} else {

pc_ctx.sql_ctx_.set_is_do_insert_batch_opt(batch_count);

fp_result.pc_key_.name_.assign_ptr(first_truncated_sql.ptr(), first_truncated_sql.length());

LOG_DEBUG("print new fp_result.pc_key_.name_", K(fp_result.pc_key_.name_));

}

}

} else if (!is_insert_values &&

OB_FAIL(ObValuesTableCompression::try_batch_exec_params(allocator, pc_ctx,

*pc_ctx.sql_ctx_.session_info_, fp_result))) {

LOG_WARN("failed to check fold params valid", K(ret));

}

}

}

return ret;

}

// For insert into t1 values(1,1),(2,2),(3,3); After parameterization,

// the SQL will become insert into t1 values(?,?),(?,?),(?,?);

// After inspection, it is found that insert multi-values ​​batch optimization can be done,

// and the SQL is truncated to insert into t1 values(?,?); If the SQL does not hit the plan from plan_cache,

// hard parsing is required, then insert into t1 values(?,?); Revert to insert into t1 values(1,1);

// This function is to complete the parameter reduction in parameterless SQL,

// so that the original SQL can be used as a parser later

// replace into statement and insert_up statement are same

int ObPlanCache::restore_param_to_truncated_sql(ObPlanCacheCtx &pc_ctx)

{

int ret = OB_SUCCESS;

char *buf = NULL;

int32_t pos = 0;

int64_t idx = 0;

const ObIArray<ObPCParam *> *raw_params = nullptr;

int64_t buff_len = pc_ctx.raw_sql_.length();

int64_t ins_params_count = pc_ctx.insert_batch_opt_info_.insert_params_count_;

ObString &no_param_sql = pc_ctx.fp_result_.pc_key_.name_;

if (pc_ctx.insert_batch_opt_info_.multi_raw_params_.empty()) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("unexpected params count", K(ret), K(pc_ctx.insert_batch_opt_info_.multi_raw_params_.count()));

} else if (OB_ISNULL(raw_params = pc_ctx.insert_batch_opt_info_.multi_raw_params_.at(0))) {

ret = OB_ERR_UNEXPECTED;

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

} else if (OB_ISNULL(buf = (char *)pc_ctx.allocator_.alloc(pc_ctx.raw_sql_.length()))) {

ret = OB_ALLOCATE_MEMORY_FAILED;

LOG_WARN("buff is null", K(ret), K(pc_ctx.raw_sql_.length()));

} else if (raw_params->count() < ins_params_count) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("unexpected params count", K(ret), K(raw_params->count()), K(ins_params_count));

}

for (int64_t i = 0; OB_SUCC(ret) && i < raw_params->count(); i++) {

ObPCParam *pc_param = nullptr;

if (OB_ISNULL(pc_param = raw_params->at(i))) {

ret = OB_ERR_UNEXPECTED;

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

} else {

int32_t len = (int32_t)pc_param->node_->pos_ - idx;

LOG_TRACE("print raw_params", K(i), K(buff_len), K(len), K(idx), K(pc_param->node_->pos_),

K(ObString(pc_param->node_->text_len_, pc_param->node_->raw_text_)));

if (len == 0) {

// insert into t1 values(2-1,2-2); becomes insert into t1 values(??,??); after parameterization

// So this scenario len == 0 is needed

if (pc_param->node_->text_len_ > buff_len - pos) {

ret = OB_BUF_NOT_ENOUGH;

LOG_WARN("unexpected len", K(ret), K(i), K(buff_len), K(pc_param->node_->text_len_), K(pos), K(no_param_sql));

} else {

MEMCPY(buf + pos, pc_param->node_->raw_text_, pc_param->node_->text_len_);

pos += (int32_t)pc_param->node_->text_len_;

idx = (int32_t)pc_param->node_->pos_ + 1;

}

} else if (len > 0) {

if (len > buff_len - pos) {

ret = OB_BUF_NOT_ENOUGH;

LOG_WARN("unexpected len", K(ret), K(i), K(buff_len), K(idx), K(pos), K(no_param_sql));

} else if (pc_param->node_->text_len_ > (buff_len - pos - len)) {

ret = OB_BUF_NOT_ENOUGH;

LOG_WARN("unexpected len", K(ret), K(i), K(buff_len), K(idx), K(pc_param->node_->text_len_), K(pos), K(no_param_sql));

} else {

// copy sql text

// insert into t1 values(?,?);

// first times, it copy 'insert into t1 values('

MEMCPY(buf + pos, no_param_sql.ptr() + idx, len);

idx = (int32_t)pc_param->node_->pos_ + 1;

pos += len;

//copy raw param

MEMCPY(buf + pos, pc_param->node_->raw_text_, pc_param->node_->text_len_);

pos += (int32_t)pc_param->node_->text_len_;

}

}

}

}

if (OB_SUCCESS == ret) {

int32_t len = no_param_sql.length() - idx;

if (len > buff_len - pos) {

ret = OB_BUF_NOT_ENOUGH;

LOG_WARN("unexpected len", K(ret), K(buff_len), K(pos), K(idx), K(no_param_sql.length()), K(no_param_sql));

} else if (len > 0) {

MEMCPY(buf + pos, no_param_sql.ptr() + idx, len);

idx += len;

pos += len;

}

}

if (OB_SUCC(ret)) {

buf[pos] = ';';

pos++;

pc_ctx.insert_batch_opt_info_.new_reconstruct_sql_.assign_ptr(buf, pos);

LOG_TRACE("print new_truncated_sql", K(pc_ctx.insert_batch_opt_info_.new_reconstruct_sql_), K(pos));

}

return ret;

}

// For insert into t1 values(1,1),(2,2); after finishing fast_parser,

// all the parameters will be extracted,and the parameter array becomes [1, 1, 2, 2]

// But we will truncate the SQL to insert into t1 values(?,?);

// so the parameter array needs to be changed to a two-dimensional matrix like {[1, 1], [2, 2]}

// This function is to complete the conversion. At the same time, for the insert_up statement,

// insert into t1 values(1,1),(2,2) on duplicate key update c1 = 3, c2 = 4;

// After the SQL is truncated, the two parameters in the update part correspond to the pos_ in the truncated SQL

// that needs to be subtracted from the truncated part

int ObPlanCache::rebuild_raw_params(common::ObIAllocator &allocator,

ObPlanCacheCtx &pc_ctx,

ObFastParserResult &fp_result,

int64_t row_count)

{

int ret = OB_SUCCESS;

int64_t params_idx = 0;

ObSEArray<ObPCParam *, 8> update_raw_params;

int64_t insert_param_count = pc_ctx.insert_batch_opt_info_.insert_params_count_;

int64_t upd_param_count = pc_ctx.insert_batch_opt_info_.update_params_count_;

int64_t one_row_params_cnt = insert_param_count + upd_param_count;

int64_t upd_start_idx = row_count * insert_param_count;

int64_t sql_delta_length = pc_ctx.insert_batch_opt_info_.sql_delta_length_;

if (((row_count * insert_param_count) + upd_param_count) != fp_result.raw_params_.count()) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("unexpected raw_params", K(ret),

K(row_count), K(insert_param_count), K(upd_param_count), K(fp_result.raw_params_.count()));

} else {

pc_ctx.insert_batch_opt_info_.multi_raw_params_.set_capacity(row_count);

}

for (int64_t i = upd_start_idx; OB_SUCC(ret) && i < fp_result.raw_params_.count(); i++) {

// As sql: insert into t1 values(1,1),(2,2) on duplicate key update c1 = 3, c2 = 4;

// After the SQL is truncated, the two parameters in the update part correspond to the pos_ in the truncated SQL

// that needs to be subtracted from the truncated part, sql_delta_length is the length of truncated part.

ObPCParam *pc_param = fp_result.raw_params_.at(i);

if (OB_ISNULL(pc_param)) {

ret = OB_ERR_UNEXPECTED;

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

} else if (FALSE_IT(pc_param->node_->pos_ = pc_param->node_->pos_ - sql_delta_length)) {

// For the parameters of the update part, pos_ needs to subtract the length of the truncated part

} else if (OB_FAIL(update_raw_params.push_back(pc_param))) {

LOG_WARN("fail to push back raw_param", K(ret), K(i));

}

}

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

void *buf = nullptr;

ObRawParams *params_array = nullptr;

if (OB_ISNULL(buf = allocator.alloc(sizeof(ObRawParams)))) {

ret = OB_ALLOCATE_MEMORY_FAILED;

LOG_WARN("fail to alloc memory", K(ret), K(sizeof(ObRawParams)));

} else {

params_array = new(buf) ObRawParams(allocator);

params_array->set_capacity(one_row_params_cnt);

}

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

if (params_idx >= fp_result.raw_params_.count()) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("unexpected params_idx", K(ret), K(i), K(j), K(params_idx), K(fp_result.raw_params_));

} else if (OB_FAIL(params_array->push_back(fp_result.raw_params_.at(params_idx)))) {

LOG_WARN("fail to push back", K(ret), K(i), K(j), K(params_idx));

} else {

params_idx++;

}

}

if (OB_SUCC(ret)) {

if (0 != upd_param_count && OB_FAIL(append(*params_array, update_raw_params))) {

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

} else if (OB_FAIL(pc_ctx.insert_batch_opt_info_.multi_raw_params_.push_back(params_array))) {

LOG_WARN("fail to push params array", K(ret));

}

}

}

return ret;

}

bool ObPlanCache::can_do_insert_batch_opt(ObPlanCacheCtx &pc_ctx)

{

bool bret = false;

ObSQLSessionInfo *session_info = nullptr;

if (OB_NOT_NULL(session_info = pc_ctx.sql_ctx_.session_info_)) {

if (!pc_ctx.sql_ctx_.is_batch_params_execute() &&

GCONF._sql_insert_multi_values_split_opt &&

!pc_ctx.sql_ctx_.get_enable_user_defined_rewrite() &&

!session_info->is_inner() &&

OB_BATCHED_MULTI_STMT_ROLLBACK != session_info->get_retry_info().get_last_query_retry_err()) {

bret = true;

} else {

LOG_TRACE("can't do insert batch optimization",

"is_arraybinding", pc_ctx.sql_ctx_.is_batch_params_execute(),

"is_open_switch", GCONF._sql_insert_multi_values_split_opt,

"is_inner_sql", session_info->is_inner(),

"udr", pc_ctx.sql_ctx_.get_enable_user_defined_rewrite(),

"last_ret", session_info->get_retry_info().get_last_query_retry_err(),

"curr_sql", pc_ctx.raw_sql_);

}

}

return bret;

}

int ObPlanCache::check_can_do_insert_opt(common::ObIAllocator &allocator,

ObPlanCacheCtx &pc_ctx,

ObFastParserResult &fp_result,

bool &can_do_batch,

int64_t &batch_count,

ObString &first_truncated_sql,

bool &is_insert_values)

{

int ret = OB_SUCCESS;

can_do_batch = false;

batch_count = 0;

if (fp_result.values_token_pos_ != 0 &&

can_do_insert_batch_opt(pc_ctx)) {

char *new_param_sql = nullptr;

int64_t new_param_sql_len = 0;

int64_t ins_params_count = 0;

int64_t upd_params_count = 0;

int64_t delta_length = 0;

ObSQLMode sql_mode = pc_ctx.sql_ctx_.session_info_->get_sql_mode();

ObCharsets4Parser charsets4parser = pc_ctx.sql_ctx_.session_info_->get_charsets4parser();

FPContext fp_ctx(charsets4parser);

fp_ctx.enable_batched_multi_stmt_ = pc_ctx.sql_ctx_.handle_batched_multi_stmt();

fp_ctx.sql_mode_ = sql_mode;

ObFastParserMysql fp(allocator, fp_ctx);

if (OB_FAIL(fp.parser_insert_str(allocator,

fp_result.values_token_pos_,

fp_result.pc_key_.name_,

first_truncated_sql,

can_do_batch,

ins_params_count,

upd_params_count,

delta_length,

batch_count))) {

LOG_WARN("fail to parser insert string", K(ret), K(fp_result.pc_key_.name_));

} else if (!can_do_batch || ins_params_count <= 0) {

can_do_batch = false;

// Only the insert ... values ​​... statement will print this,after trying to do insert batch optimization failure

LOG_INFO("can not do batch insert opt", K(ret), K(can_do_batch), K(upd_params_count),

K(ins_params_count), K(batch_count), K(pc_ctx.raw_sql_));

} else if (batch_count <= 1) {

can_do_batch = false;