hcq/oceanbase
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

[FEAT MERGE] implement SQL level resource management based on pattern match

Browse Source
This commit is contained in:
obdev
2023-01-04 12:39:02 +00:00
committed by ob-robot
parent cca9f7c2d2
commit 7c991b5da5
58 changed files with 2613 additions and 168 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
src/sql/plan_cache/ob_pcv_set.cpp
View File

@ -493,6 +493,23 @@ int ObPCVSet::check_raw_param_for_dup_col(ObPlanCacheCtx &pc_ctx, bool &contain_
}
return ret;
}
int ObPCVSet::check_contains_table(uint64_t db_id, common::ObString tab_name, bool &contains)
{
int ret = OB_SUCCESS;
DLIST_FOREACH(pcv, pcv_list_) {
if (OB_ISNULL(pcv)) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("invalid argument", K(pcv), K(ret));
} else if (OB_FAIL(pcv->check_contains_table(db_id, tab_name, contains))) {
LOG_WARN("fail to check table name", K(ret), K(db_id), K(tab_name));
} else if (!contains) {
// continue find
} else {
break;
}
}
return ret;
}
}
}

1
src/sql/plan_cache/ob_pcv_set.h
View File

@ -95,6 +95,7 @@ public:
ObPlanCacheKey &get_plan_cache_key() { return pc_key_; }
const ObString &get_sql() { return sql_; }
int deep_copy_sql(const common::ObString &sql);
int check_contains_table(uint64_t db_id, common::ObString tab_name, bool &contains);
TO_STRING_KV(K_(is_inited));

49
src/sql/plan_cache/ob_plan_cache.cpp
View File

@ -217,6 +217,40 @@ struct ObGetKVEntryBySQLIDOp : public ObKVEntryTraverseOp
};
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_;
};
struct ObGetTableIdOp
{
explicit ObGetTableIdOp(uint64_t table_id)
@ -1127,6 +1161,21 @@ int ObPlanCache::cache_evict_plan_by_sql_id(uint64_t db_id, common::ObString sql
}
int ObPlanCache::evict_plan_by_table_name(uint64_t tenant_id, uint64_t database_id, ObString tab_name)
{
int ret = OB_SUCCESS;
ObGlobalReqTimeService::check_req_timeinfo();
SQL_PC_LOG(DEBUG, "cache evict plan by table name start");
LCKeyValueArray to_evict_keys;
ObGetPcvSetByTabNameOp get_ids_op(database_id, tab_name, &to_evict_keys, PCV_GET_PLAN_KEY_HANDLE);
if (OB_FAIL(foreach_cache_evict(get_ids_op))) {
SQL_PC_LOG(WARN, "failed to foreach cache evict", K(ret));
}
SQL_PC_LOG(DEBUG, "cache evict plan baseline by sql id end");
return ret;
}
// Delete the cache according to the evict mechanism
// 1. calc evict_num : (mem_used - mem_lwm) / (mem_used / cache_value_count)
// 2. get evict_sql_id from calc_cache_evict_keys

1
src/sql/plan_cache/ob_plan_cache.h
View File

@ -201,6 +201,7 @@ public:
int add_exists_cache_obj_by_sql(ObILibCacheCtx &ctx,
ObILibCacheObject *cache_obj);
int evict_plan(uint64_t table_id);
int evict_plan_by_table_name(uint64_t tenant_id, uint64_t database_id, ObString tab_name);
/**
* memory related

14
src/sql/plan_cache/ob_plan_cache_manager.cpp
View File

@ -464,6 +464,20 @@ void ObPlanCacheManager::ObPlanCacheEliminationTask::run_free_cache_obj_task()
}
}
int ObPlanCacheManager::evict_plan_by_table_name(uint64_t tenant_id, uint64_t database_id, ObString tab_name)
{
int ret = OB_SUCCESS;
observer::ObReqTimeGuard req_timeinfo_guard;
ObPlanCache *plan_cache = get_plan_cache(tenant_id);
if (NULL != plan_cache) {
if (OB_FAIL(plan_cache->evict_plan_by_table_name(tenant_id, database_id, tab_name))) {
SQL_PC_LOG(WARN, "fail to evict plan by table name", K(ret));
}
plan_cache->dec_ref_count();
}
return ret;
}
int ObPlanCacheManager::flush_plan_cache_by_sql_id(uint64_t tenant_id,
uint64_t db_id,
common::ObString sql_id) {

1
src/sql/plan_cache/ob_plan_cache_manager.h
View File

@ -104,6 +104,7 @@ public:
int flush_ps_cache(const uint64_t tenant_id);
int flush_lib_cache(const uint64_t tenant_id);
int flush_lib_cache_by_ns(const uint64_t tenant_id, const ObLibCacheNameSpace ns);
int evict_plan_by_table_name(uint64_t tenant_id, uint64_t database_id, common::ObString tab_name);
PlanCacheMap &get_plan_cache_map() {return pcm_;}

97
src/sql/plan_cache/ob_plan_cache_value.cpp
View File

@ -28,6 +28,7 @@
#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"
using namespace oceanbase::share::schema;
using namespace oceanbase::common;
using namespace oceanbase::pl;
@ -519,7 +520,63 @@ int ObPlanCacheValue::choose_plan(ObPlanCacheCtx &pc_ctx,
} else {
SQL_PC_LOG(TRACE, "failed to select plan in plan set", K(ret));
}
} else {
} 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循环的宏；
}
}
@ -1029,8 +1086,28 @@ int ObPlanCacheValue::add_plan(ObPlanCacheObject &plan,
} else if (is_old_version) {
ret = OB_OLD_SCHEMA_VERSION;
SQL_PC_LOG(DEBUG, "view or table is old version", K(ret));
/* Consider this concurrent scene:
1. No mapping is defined on t.c1 at first.
2. Thread1 resolve select * from t where c1 = 1; and generate a plan with rule_id = INVALID
3. User define a mapping rule on t.c1.
4. Thread2 load the new mapping rule on t.c1 into cache and evict all plans related with t
5. Thread1 add the plan into plan cache. The plan is marked without a mapping rule
but there is actually a mapping rule on t.c1 now
Solution:
1. When start to resolve a sql, record the current version of mapping rule.
2. Before adding a plan into plan cache, check whether the recorded version is same as current version,
and not add into plan cache if not same.
THERE IS A FLAW of this solution. If step 4 accurs right in the gap between check version and add plan in plan cache,
a stale plan will be added into plan cache. Since the gap is quite small, we think the flaw is acceptable.
*/
} else if (pc_ctx.sql_ctx_.res_map_rule_version_ != 0) {
int64_t latest_rule_version = G_RES_MGR.get_col_mapping_rule_mgr().get_column_mapping_version(MTL_ID());
if (pc_ctx.sql_ctx_.res_map_rule_version_ != latest_rule_version) {
ret = OB_OLD_SCHEMA_VERSION;
SQL_PC_LOG(TRACE, "resource map rule version is outdated, not add to plan cache.", K(ret),
K(pc_ctx.sql_ctx_.res_map_rule_version_), K(latest_rule_version));
}
}
if (OB_FAIL(ret)) {//do nothing
} else if (OB_FAIL(get_outline_param_index(pc_ctx.exec_ctx_, outline_param_idx))) {
LOG_WARN("fail to judge concurrent limit sql", K(ret));
@ -1980,5 +2057,21 @@ int ObPlanCacheValue::rm_space_for_neg_num(ParseNode *param_node, ObIAllocator &
return ret;
}
int ObPlanCacheValue::check_contains_table(uint64_t db_id, common::ObString tab_name, bool &contains)
{
int ret = OB_SUCCESS;
for (int64_t i = 0; !contains && i < stored_schema_objs_.count(); i++) {
if (OB_ISNULL(stored_schema_objs_.at(i))) {
// do nothing
} else {
if ((stored_schema_objs_.at(i)->database_id_ == db_id) &&
(stored_schema_objs_.at(i)->table_name_ == tab_name)) {
contains = true;
}
}
}
return ret;
}
}//end of namespace sql
}//end of namespace oceanbase

1
src/sql/plan_cache/ob_plan_cache_value.h
View File

@ -261,6 +261,7 @@ public:
common::ObIArray<PCVSchemaObj> &schema_array);
int lift_tenant_schema_version(int64_t new_schema_version);
int check_contains_table(uint64_t db_id, common::ObString tab_name, bool &contains);
private:
//used for add plan
//check table version, view table version, merged version

5
src/sql/plan_cache/ob_plan_set.cpp
View File

@ -32,6 +32,7 @@
#include "sql/plan_cache/ob_cache_object_factory.h"
#include "pl/ob_pl.h"
#include "ob_plan_set.h"
#include "share/resource_manager/ob_resource_manager.h"
using namespace oceanbase;
using namespace common;
@ -590,6 +591,10 @@ int ObPlanSet::init_new_set(const ObPlanCacheCtx &pc_ctx,
is_ignore_stmt_ = plan.is_ignore();
//add param info
params_info_.reset();
// set variables for resource map rule
// if rule changed, plan cache will be flush.
res_map_rule_id_ = pc_ctx.sql_ctx_.res_map_rule_id_;
res_map_rule_param_idx_ = pc_ctx.sql_ctx_.res_map_rule_param_idx_;
if (OB_FAIL(init_pre_calc_exprs(plan, pc_alloc_))) {
LOG_WARN("failed to init pre calc exprs", K(ret));

9
src/sql/plan_cache/ob_plan_set.h
View File

@ -109,7 +109,9 @@ public:
all_plan_const_param_constraints_(alloc_),
all_pre_calc_constraints_(),
multi_stmt_rowkey_pos_(alloc_),
pre_cal_expr_handler_(NULL)
pre_cal_expr_handler_(NULL),
res_map_rule_id_(common::OB_INVALID_ID),
res_map_rule_param_idx_(common::OB_INVALID_INDEX)
{}
virtual ~ObPlanSet();
@ -224,6 +226,11 @@ protected:
common::ObFixedArray<int64_t, common::ObIAllocator> multi_stmt_rowkey_pos_;
// pre calculable expression list handler.
PreCalcExprHandler* pre_cal_expr_handler_;
public:
//variables for resource map rule
uint64_t res_map_rule_id_;
int64_t res_map_rule_param_idx_;
};
class ObSqlPlanSet : public ObPlanSet