模板站怎么改成营销型网站,怎么从网上找客户,php代理ip访问网站,建设包包网站的目的本专题的前几篇文章已经详细阐述了OceanBase的配置项和系统变量的基础用法#xff0c;并对配置项的源码进行了剖析。但有一些同学可能还对系统变量的实现方式有兴趣#xff0c;并希望能够像自定义配置项那样#xff0c;新增一个系统变量。
本文将围绕“如何新增系统变量”这…本专题的前几篇文章已经详细阐述了OceanBase的配置项和系统变量的基础用法并对配置项的源码进行了剖析。但有一些同学可能还对系统变量的实现方式有兴趣并希望能够像自定义配置项那样新增一个系统变量。
本文将围绕“如何新增系统变量”这一话题展开结合系统变量的源码实现为大家详细讲解系统变量的定义、初始化过程、内部访问方式以及同步机制。帮助大家更好地理解和运用系统变量。 系统变量概述
一个系统变量可以同时具有 global 级别和 session 级别的数据在代码中存在多个副本。这与配置项参数不同一个配置项要么是集群级别要么是租户级别。
系统变量的定义
要增加一个新的变量首先在 src/share/system_variable/ob_system_variable_init.json 文件中按照下面的格式编写一段代码。 ob_query_timeout: {id: 10005,name: ob_query_timeout,default_value: 10000000,base_value: 10000000,data_type: int,info: Query timeout in microsecond(us),flags: GLOBAL | SESSION | NEED_SERIALIZE,on_check_and_convert_func: ObSysVarOnCheckFuncs::check_and_convert_timeout_too_large,publish_version: ,info_cn: ,background_cn: ,ref_url: },
其中每一项的含义如下
2参数说明举例对象名括号外数据结构的名字与括号内的变量名一致ob_query_timeoutid变量的序号按顺序手动分配0, 1, 95, 10001, 10037name变量名与括号外的对象名一致ob_query_timeoutdefault_value默认值0, 1000, disabled, binarybase_value基准值初始与默认值相同用于不同session间的变量同步0, 1000, disabled, binarydata_type数据类型int, varchar, enum, boolinfo描述信息描述变量的主要功能Query timeout in microsecond(us)flags标签包括级别、是否序列化、是否可见等GLOBAL | SESSION | NEED_SERIALIZE | ORACLE_ONLY | INVISIBLEon_check_and_convert_func合法性检查和格式转换函数可选ObSysVarOnCheckFuncs::check_and_convert_timeout_too_largemin_val最小值可选1max_val最大值可选10000enum_names枚举类型的取值列表可选enum_names: [OFF, AUTO, FORCE],publish_version发布版本号227, 420即2.2.74.2.0info_cn中文描述查询超时时间微秒background_cn添加原因/背景避免死锁、一直重试等问题导致的sql阻塞的情况ref_url变量申请链接xxx
然后执行 ./gen_ob_sys_variables.py 脚本新增的变量会出现在 ob_system_variable_init.cpp 文件中。
[] (){ObSysVars[100].default_value_ 10000000 ;ObSysVars[100].info_ Query timeout in microsecond(us) ;ObSysVars[100].name_ ob_query_timeout ;ObSysVars[100].data_type_ ObIntType ;ObSysVars[100].flags_ ObSysVarFlag::GLOBAL_SCOPE | ObSysVarFlag::SESSION_SCOPE | ObSysVarFlag::NEED_SERIALIZE ;ObSysVars[100].on_check_and_convert_func_ ObSysVarOnCheckFuncs::check_and_convert_timeout_too_large ;ObSysVars[100].id_ SYS_VAR_OB_QUERY_TIMEOUT ;cur_max_var_id MAX(cur_max_var_id, static_castint64_t(SYS_VAR_OB_QUERY_TIMEOUT)) ;ObSysVarsIdToArrayIdx[SYS_VAR_OB_QUERY_TIMEOUT] 100 ;ObSysVars[100].base_value_ 10000000 ;ObSysVars[100].alias_ OB_SV_QUERY_TIMEOUT ;}();
这些变量的初始值都会存储在以下数组中。其中 ObSysVars 数组保存了从 json 文件生成的最原始的变量信息ObSysVarDefaultValues 数组存储了系统变量的默认值ObSysVarBaseValues 数组存储了系统变量的基准值。每新增一个系统变量这些数组中就会多一个元素。
namespace oceanbase
{
namespace share
{
// 变量的初始信息
static ObSysVarFromJson ObSysVars[ObSysVarFactory::ALL_SYS_VARS_COUNT];
// 变量的默认值
static ObObj ObSysVarDefaultValues[ObSysVarFactory::ALL_SYS_VARS_COUNT];
// 变量的基准值
static ObObj ObSysVarBaseValues[ObSysVarFactory::ALL_SYS_VARS_COUNT]; 系统变量缓存的定义
在某些场景为满足功能或性能的需求需要额外为系统变量增加一份缓存按需定义不是必须的。系统变量的缓存通过 DEF_SYS_VAR_CACHE_FUNCS 和 DEF_SYS_VAR_CACHE_FUNCS_STR 宏定义。
class ObBasicSessionInfo
{
private:
#define DEF_SYS_VAR_CACHE_FUNCS(SYS_VAR_TYPE, SYS_VAR_NAME) \void set_##SYS_VAR_NAME(SYS_VAR_TYPE value) \{ \inc_data_.SYS_VAR_NAME##_ (value); \inc_##SYS_VAR_NAME##_ true; \} \void set_base_##SYS_VAR_NAME(SYS_VAR_TYPE value) \{ \base_data_.SYS_VAR_NAME##_ (value); \} \const SYS_VAR_TYPE get_##SYS_VAR_NAME() const \{ \return get_##SYS_VAR_NAME(inc_##SYS_VAR_NAME##_); \} \const SYS_VAR_TYPE get_##SYS_VAR_NAME(bool is_inc) const \{ \return is_inc ? inc_data_.SYS_VAR_NAME##_ : base_data_.SYS_VAR_NAME##_; \}class SysVarsCache{public:DEF_SYS_VAR_CACHE_FUNCS(int64_t, ob_max_read_stale_time);DEF_SYS_VAR_CACHE_FUNCS(bool, tx_read_only);DEF_SYS_VAR_CACHE_FUNCS_STR(nls_date_format);}private:static SysVarsCacheData base_data_;SysVarsCacheData inc_data_;union {uint64_t inc_flags_;struct {bool inc_auto_increment_increment_:1;bool inc_sql_throttle_current_priority_:1;......
其中 base_data_ 表示基准数据它是全局唯一的与“global变量”对应inc_data_ 表示最新数据每个session各有一份与“session变量”对应。
以 ob_max_read_stale_time 变量为例上面的宏展开后生成以下代码。 void set_ob_max_read_stale_time(int64_t value){inc_data_.ob_max_read_stale_time_ (value);inc_ob_max_read_stale_time_ true;}void set_base_ob_max_read_stale_time(int64_t value){base_data_.ob_max_read_stale_time_ (value);}const int64_t get_ob_max_read_stale_time() const{return get_ob_max_read_stale_time(inc_ob_max_read_stale_time_);}const int64_t get_ob_max_read_stale_time(bool is_inc) const{return is_inc ? inc_data_.ob_max_read_stale_time_ : base_data_.ob_max_read_stale_time_;}
其中 set_ob_max_read_stale_time() 函数会修改 inc_ob_max_read_stale_time_ 为 true 并更新 inc_data_然后调用 get_ob_max_read_stale_time() 函数会返回修改过的数据 inc_data_.ob_max_read_stale_time_未修改则会返回基准数据 base_data_.ob_max_read_stale_time_。 变量初始化
定义上面的一系列数据结构后接下来在 OBServer 启动时系统变量会进行初始化。
变量基础数据
ObSysVars 数组保存了最原始的变量信息在静态对象的构造函数中进行初始化。
namespace oceanbase
{
namespace share
{
// 变量的初始信息
static ObSysVarFromJson ObSysVars[ObSysVarFactory::ALL_SYS_VARS_COUNT];
// 变量的默认值
static ObObj ObSysVarDefaultValues[ObSysVarFactory::ALL_SYS_VARS_COUNT];
// 变量的基准值
static ObObj ObSysVarBaseValues[ObSysVarFactory::ALL_SYS_VARS_COUNT];static struct VarsInit{VarsInit(){// 保存当前系统变量的最大的idint64_t cur_max_var_id 0;// ObSysVarsIdToArrayIdx数组默认初始值为-1-1表示无效索引memset(ObSysVarsIdToArrayIdx, -1, sizeof(ObSysVarsIdToArrayIdx));[] (){ObSysVars[0].default_value_ 1 ;ObSysVars[0].info_ ;ObSysVars[0].name_ auto_increment_increment ;ObSysVars[0].data_type_ ObUInt64Type ;ObSysVars[0].min_val_ 1 ;ObSysVars[0].max_val_ 65535 ;ObSysVars[0].flags_ ObSysVarFlag::GLOBAL_SCOPE | ObSysVarFlag::SESSION_SCOPE | ObSysVarFlag::NEED_SERIALIZE ;ObSysVars[0].id_ SYS_VAR_AUTO_INCREMENT_INCREMENT ;cur_max_var_id MAX(cur_max_var_id, static_castint64_t(SYS_VAR_AUTO_INCREMENT_INCREMENT)) ;ObSysVarsIdToArrayIdx[SYS_VAR_AUTO_INCREMENT_INCREMENT] 0 ;ObSysVars[0].base_value_ 1 ;ObSysVars[0].alias_ OB_SV_AUTO_INCREMENT_INCREMENT ;}();
ObSysVarDefaultValues 数组保存了系统变量的默认值ObSysVarBaseValues 数组保存了系统变量的基准值。在 OBServer 启动时会初始化这两个数组
ObServer::init()ObPreProcessSysVars::init_sys_var()ObSysVariables::init_default_values()
该函数遍历 ObSysVars 数组将每个变量的 default_value_ 存储到 ObSysVarDefaultValues 数组中将每个变量的 base_value_ 存储到 ObSysVarBaseValues 数组中。
int ObSysVariables::init_default_values()
{int ret OB_SUCCESS;int64_t sys_var_count get_amount();for (int64_t i 0; OB_SUCC(ret) i sys_var_count; i) {const ObString sys_var_val_str ObSysVariables::get_value(i);const ObString base_sys_var_val_str ObSysVariables::get_base_str_value(i);......ObSysVarDefaultValues[i] out_obj;ObSysVarBaseValues[i] base_out_obj; global 变量sysvar_array_
ObSysVariableSchema 是管理系统变量的 schema 对象其中的 sysvar_array_ 数组存储了 global 变量的值因此系统变量也具有 schema 多版本的能力。
class ObSysVariableSchema : public ObSchema
{
private:uint64_t tenant_id_;int64_t schema_version_;ObSysVarSchema *sysvar_array_[ObSysVarFactory::ALL_SYS_VARS_COUNT];bool read_only_;class ObSysVarSchema : public ObSchema
{
private:uint64_t tenant_id_;common::ObString name_;common::ObObjType data_type_;common::ObString value_;common::ObString min_val_;common::ObString max_val_;common::ObString info_;common::ObZone zone_;int64_t schema_version_;int64_t flags_;
sysvar_array_ 数组在 OBServer 启动时进行初始化首先调用的是 init_schema() 函数。
ObServer::init_schema()ObMultiVersionSchemaService::init()ObMultiVersionSchemaService::init_sys_tenant_user_schema()
先调用 load_default_system_variable() 将所有变量的默认值都加载到 sysvar_array_ 数组中再调用 put_schema() 把这个 schema 添加到 schema_cache_ 中。
int ObMultiVersionSchemaService::init_sys_tenant_user_schema()
{ObSysVariableSchema sys_variable;if (OB_FAIL(sys_variable.load_default_system_variable(true))) {LOG_WARN(load sys tenant default system variable failed, K(ret));} else if (OB_FAIL(schema_cache_.put_schema(SYS_VARIABLE_SCHEMA,OB_SYS_TENANT_ID,sys_variable.get_tenant_id(),sys_variable.get_schema_version(),sys_variable))) {
ObSysVariableSchema::load_default_system_variable()
该函数遍历 ObSysVars 数组获取变量的默认值然后调用 add_sysvar_schema() 函数将变量值添加到 sysvar_array_ 数组中。
int ObSysVariableSchema::load_default_system_variable(bool is_sys_tenant)
{int ret OB_SUCCESS;ObString value;ObSysVarSchema sysvar;for (int64_t i 0; OB_SUCC(ret) i ObSysVariables::get_amount(); i) {sysvar.reset();// 从 ObSysVariables 中获取默认数据if (is_sys_tenant ObCharset::case_insensitive_equal(ObSysVariables::get_name(i), OB_SV_LOWER_CASE_TABLE_NAMES)) {value ObString::make_string(2);} else {value ObSysVariables::get_value(i);}} else if (OB_FAIL(sysvar.set_value(value))) {LOG_WARN(set sysvar value failed, K(ret), K(value));} else if (OB_FAIL(add_sysvar_schema(sysvar))) {LOG_WARN(add sysvar schema failed, K(ret));}} session 变量sys_vars_
ObBasicSessionInfo 是管理 session 相关信息的对象其中的 sys_vars_ 数组存储了 session 变量的值在建立 session 连接时进行初始化。
class ObBasicSessionInfo
{
private:share::ObBasicSysVar *sys_vars_[share::ObSysVarFactory::ALL_SYS_VARS_COUNT];
外部 session
外部 session 连接是指从客户端发起的连接一般用于处理用户的 sql 请求。外部 session 的变量缓存值在初始化时会从 schema 中取值同时也依赖 schema 的同步机制。
int ObMPConnect::process()
在建立连接时先调用 create_session() 函数创建 session然后调用 verify_identify() 初始化变量。
ObMPConnect::verify_identify()ObMPConnect::load_privilege_info()
通过租户ID从 schema_cache_ 中获取对应的 sys_variable_schema然后将 sys_variable_schema 作为参数调用 load_all_sys_vars() 函数。
int ObMPConnect::load_privilege_info(ObSQLSessionInfo session)
{const ObSysVariableSchema *sys_variable_schema NULL;} else if (OB_FAIL(schema_guard.get_sys_variable_schema(conn-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(session.init_tenant(tenant_name_, conn-tenant_id_))) {LOG_WARN(failed to init_tenant, K(ret));} else if (OB_FAIL(session.load_all_sys_vars(*sys_variable_schema, false))) {LOG_WARN(load system variables failed, K(ret));
ObBasicSessionInfo::load_all_sys_vars()
该函数每次循环先从 ObSysVars 数组中获取 sys_var_id然后从传入的 sys_var_schema 中获取 sys_var 对象然后把变量的 name 和 value 作为参数调用 load_sys_variable() 函数。
int ObBasicSessionInfo::load_all_sys_vars(const ObSysVariableSchema sys_var_schema, bool sys_var_created)
{int ret OB_SUCCESS;OZ (clean_all_sys_vars());if (!sys_var_created) {OZ (sys_var_fac_.create_all_sys_vars());}OX (influence_plan_var_indexs_.reset());ObArenaAllocator calc_buf(ObModIds::OB_SQL_SESSION);for (int64_t i 0; OB_SUCC(ret) i get_sys_var_count(); i) {ObSysVarClassType sys_var_id ObSysVariables::get_sys_var_id(i);const ObSysVarSchema *sys_var NULL;OZ (sys_var_schema.get_sysvar_schema(sys_var_id, sys_var), sys_var_id, i);OV (OB_NOT_NULL(sys_var));OZ (load_sys_variable(calc_buf, sys_var-get_name(), sys_var-get_data_type(),sys_var-get_value(), sys_var-get_min_val(),sys_var-get_max_val(), sys_var-get_flags(), true));
ObBasicSessionInfo::load_sys_variable()
先调用 find_sys_var_id_by_name() 函数通过 name 获取 var_id然后调用 create_sys_var() 函数获取创建变量指针 sys_var再调用 sys_var-init() 给变量赋值最后调用 process_session_variable() 函数给变量缓存赋值。
int ObBasicSessionInfo::load_sys_variable(ObIAllocator calc_buf,const ObString name,const ObObj type,const ObObj value,const ObObj min_val,const ObObj max_val,const int64_t flags,bool is_from_sys_table)
{if (SYS_VAR_INVALID (var_id ObSysVarFactory::find_sys_var_id_by_name(name, is_from_sys_table))) {......} else if (OB_FAIL(create_sys_var(var_id, store_idx, sys_var))) {LOG_WARN(fail to create sys var, K(name), K(value), K(ret));......} else if (OB_FAIL(sys_var-init(real_val, min_ptr, max_ptr, val_type.get_type(), flags))) {LOG_WARN(fail to init sys var, K(ret), K(sys_var-get_type()),K(real_val), K(name), K(value));} else if (OB_FAIL(process_session_variable(var_id, real_val,false /*check_timezone_valid*/,false /*is_update_sys_var*/))) {
ObBasicSessionInfo::create_sys_var()
初次调用 create_sys_var() 函数时sys_vars_[store_idx] 为空然后调用 sys_var_fac_.create_sys_var() 函数生成 sys_var再把 sys_var 赋给 sys_vars_[store_idx]。session 刚初始化时sys_var 指向的是变量的默认值。
再次调用 create_sys_var() 函数时sys_vars_[store_idx] 不为空将其值赋给 sys_var相当于获取了当前session 的变量。
int ObBasicSessionInfo::create_sys_var(ObSysVarClassType sys_var_id,int64_t store_idx, ObBasicSysVar *sys_var)
{int ret OB_SUCCESS;OV (0 store_idx store_idx ObSysVarFactory::ALL_SYS_VARS_COUNT,OB_ERR_UNEXPECTED, sys_var_id, store_idx);if (OB_NOT_NULL(sys_vars_[store_idx])) {OV (sys_vars_[store_idx]-get_type() sys_var_id,OB_ERR_UNEXPECTED, sys_var_id, store_idx, sys_vars_[store_idx]-get_type());OX (sys_var sys_vars_[store_idx]);} else {OZ (sys_var_fac_.create_sys_var(sys_var_id, sys_var), sys_var_id);OV (OB_NOT_NULL(sys_var), OB_ERR_UNEXPECTED, sys_var_id, store_idx);OX (sys_vars_[store_idx] sys_var);}return ret;
} 内部 session
内部 session 连接是指 OBServer 内部或 OBProxy 向 OBServer 发起的内部请求建立的连接不是用户主动建立的连接。内部 session 的变量总是获取全局默认值作为初值而不是从最新的 schema 中取值。
ObCommonSqlProxy::acquire()ObInnerSQLConnectionPool::acquire()ObInnerSQLConnection::init()ObInnerSQLConnection::init_session()ObInnerSQLConnection::init_session_info()ObBasicSessionInfo::load_default_sys_variable()
int ObBasicSessionInfo::load_default_sys_variable(const bool print_info_log, const bool is_sys_tenant, bool is_deserialized)
{int ret OB_SUCCESS;if (OB_FAIL(sys_var_fac_.create_all_sys_vars())) {LOG_WARN(fail create all sys variables, K(ret));} else if (OB_FAIL(init_system_variables(print_info_log, is_sys_tenant, is_deserialized))) {LOG_WARN(Init system variables failed !, K(ret));}return ret;
}
ObBasicSessionInfo::init_system_variables()
调用 ObSysVariables::get_value(i) 获取 value也就是从全局的 ObSysVars 数组中获取默认值然后调用 load_sys_variable() 函数将 value 的值加载到 session 中。
int ObBasicSessionInfo::init_system_variables(const bool print_info_log, const bool is_sys_tenant,bool is_deserialized)
{for (int64_t i 0; OB_SUCC(ret) i var_amount; i) {name.assign_ptr(const_castchar*(ObSysVariables::get_name(i).ptr()),static_castObString::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) {// Note: 如果已经初始化过 base value则下面的流程不会执行var_type ObSysVariables::get_type(i);var_flag ObSysVariables::get_flags(i);value.set_varchar(is_deserialized ? ObSysVariables::get_base_str_value(i) :ObSysVariables::get_value(i));value.set_collation_type(ObCharset::get_system_collation());min_val.set_varchar(ObSysVariables::get_min(i));min_val.set_collation_type(ObCharset::get_system_collation());max_val.set_varchar(ObSysVariables::get_max(i));max_val.set_collation_type(ObCharset::get_system_collation());type.set_type(var_type);if(is_sys_tenant) {if (OB_FAIL(process_variable_for_tenant(name, value))) {LOG_WARN(process system variable for tenant error, K(name), K(value), K(ret));}}if (OB_SUCC(ret)) {if (OB_FAIL(load_sys_variable(calc_buf, name, type, value, min_val, max_val, var_flag, false))) {
ObBasicSessionInfo::load_sys_variable()
初次调用 create_sys_var() 函数时创建 session 变量 sys_var然后用传入的默认值 value 对其进行初始化。
int ObBasicSessionInfo::load_sys_variable(ObIAllocator calc_buf,const ObString name,const ObObj type,const ObObj value,const ObObj min_val,const ObObj max_val,const int64_t flags,bool is_from_sys_table)
{} else if (OB_FAIL(create_sys_var(var_id, store_idx, sys_var))) {LOG_WARN(fail to create sys var, K(name), K(value), K(ret));} else if (OB_FAIL(ObBasicSessionInfo::change_value_for_special_sys_var(var_id, val_ptr, real_val))) {LOG_WARN(fail to change value for special sys var, K(ret), K(var_id), K(val_ptr));} else if (OB_FAIL(sys_var-init(real_val, min_ptr, max_ptr, val_type.get_type(), flags))) {LOG_WARN(fail to init sys var, K(ret), K(sys_var-get_type()),K(real_val), K(name), K(value));} else if (OB_FAIL(process_session_variable(var_id, real_val,false /*check_timezone_valid*/,false /*is_update_sys_var*/))) {LOG_WARN(process system variable error, K(name), K(type), K(real_val), K(value), K(ret)); 变量缓存基准值base_data_
base_data_ 表示系统变量缓存的基准数据它是 OBServer 全局唯一的。
ObServer::init()ObBasicSessionInfo::init_sys_vars_cache_base_values()
遍历 ObSysVarDefaultValues 数组将变量的默认值写到 ObBasicSessionInfo::base_data_ 中这是一个 static SysVarsCacheData[] 类型的数组。
int ObBasicSessionInfo::init_sys_vars_cache_base_values()
{int ret OB_SUCCESS;int64_t store_idx OB_INVALID_INDEX_INT64;ObBasicSessionInfo::SysVarsCache sys_vars_cache;int64_t var_amount ObSysVariables::get_amount();for (int64_t i 0; OB_SUCC(ret) i var_amount; i) {store_idx ObSysVarsToIdxMap::get_store_idx((int64_t)ObSysVariables::get_sys_var_id(i));OX (fill_sys_vars_cache_base_value(ObSysVariables::get_sys_var_id(i),sys_vars_cache,ObSysVariables::get_default_value(store_idx) ));}return ret;
} 变量缓存增量值inc_data_
inc_data_ 表示 session 变量的缓存值每个 session 各有一份缓存数据。
外部session 的 inc_data_ 变量缓存值初始化为 schema 中的最新值内部 session 的 inc_data_ 初始化为变量的默认值也就是说变量缓存与 session 变量的初值是一致的它们初始化的调用路径也基本相同。 变量查询
外部查询
查询 global 变量
global 变量的值是从 schema 中获取的在执行如“show global variables like xxx”的命令时就会触发以下流程
ObBasicSessionInfo::get_global_sys_variable()
先调用 get_sys_variable_schema() 函数获取 sys_variable_schema 对象然后从该对象中获取 sysvar_schema。
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)
{} 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))) {
ObSysVariableSchema::get_sysvar_schema(const ObString name, const ObSysVarSchema *sysvar_schema)ObSysVariableSchema::get_sysvar_schema(ObSysVarClassType var_id, const ObSysVarSchema *sysvar_schema)ObSysVariableSchema::get_sysvar_schema(int64_t idx)
该函数根据传入的 idx 获取对应的变量最终返回 sysvar_array_[idx]。
const ObSysVarSchema *ObSysVariableSchema::get_sysvar_schema(int64_t idx) const
{const ObSysVarSchema *ret NULL;if (idx 0 idx get_sysvar_count()) {ret sysvar_array_[idx];}return ret;
} 查询 session 变量
session 变量的值是从 session 本地数组 sys_vars_ 中获取的执行如“show variables like xxx”的命令时会触发以下调用流程
ObResultSet::inner_get_next_row()......ObBasicSessionInfo::get_sys_variable_by_name()ObBasicSessionInfo::inner_get_sys_var(oceanbase::common::ObString const, oceanbase::share::ObBasicSysVar*)ObBasicSessionInfo::inner_get_sys_var(oceanbase::common::ObString const, long, oceanbase::share::ObBasicSysVar*)
该函数先调用 find_sys_var_id_by_name() 函数通过 sys_var_name 获取 sys_var_id然后调用 calc_sys_var_store_idx() 函数利用 sys_var_id 计算出 store_idx也就是变量在数组中的下标最后返回 sys_vars_[store_idx] 即可。
int ObBasicSessionInfo::inner_get_sys_var(const ObString sys_var_name,int64_t store_idx,ObBasicSysVar *sys_var) const
{int ret OB_SUCCESS;ObSysVarClassType sys_var_id SYS_VAR_INVALID;if (OB_UNLIKELY(SYS_VAR_INVALID (sys_var_id ObSysVarFactory::find_sys_var_id_by_name(sys_var_name)))) {ret OB_ERR_SYS_VARIABLE_UNKNOWN;LOG_WARN(fail to find sys var id by name, K(ret), K(sys_var_name), K(lbt()));} else if (OB_FAIL(ObSysVarFactory::calc_sys_var_store_idx(sys_var_id, store_idx))) {LOG_WARN(fail to calc sys var store idx, K(ret), K(sys_var_id), K(sys_var_name), K(lbt()));} else if (OB_UNLIKELY(store_idx 0) ||OB_UNLIKELY(store_idx ObSysVarFactory::ALL_SYS_VARS_COUNT)) {ret OB_ERR_UNEXPECTED;LOG_ERROR(got store_idx is invalid, K(ret), K(store_idx));} else if (OB_ISNULL(sys_vars_[store_idx])) {ret OB_ENTRY_NOT_EXIST;LOG_WARN(sys var is NULL, K(ret), K(store_idx), K(sys_var_name));} else {sys_var sys_vars_[store_idx];}return ret;
} 内部获取
获取 global 变量
内部访问 global 变量同样是通过 schema 获取先从全局的 GCTX.schema_service_中获取 schema_guard然后通过变量名获取 var_schema再逐步解析得到最终的基础数据类型。
例如获取 weak_read_version_refresh_interval
int ObRootService::check_weak_read_version_refresh_interval(int64_t refresh_interval, bool valid)
{......} else if (OB_FAIL(GCTX.schema_service_-get_tenant_schema_guard(tenant_id, schema_guard))) {LOG_WARN(get schema guard failed, KR(ret), K(tenant_id));} else if (OB_FAIL(schema_guard.get_tenant_system_variable(tenant_id,OB_SV_MAX_READ_STALE_TIME, var_schema))) {LOG_WARN(get tenant system variable failed, KR(ret), K(tenant_id));} else if (OB_ISNULL(var_schema)) {ret OB_ERR_UNEXPECTED;LOG_WARN(var schema is null, KR(ret), K(tenant_id));} else if (OB_FAIL(var_schema-get_value(NULL, NULL, obj))) {LOG_WARN(get value failed, KR(ret), K(tenant_id), K(obj));} else if (OB_FAIL(obj.get_int(session_max_stale_time))) {LOG_WARN(get int failed, KR(ret), K(tenant_id), K(obj));} else if (session_max_stale_time ! share::ObSysVarFactory::INVALID_MAX_READ_STALE_TIME refresh_interval session_max_stale_time) {
通过这种方式获取变量通常会定义一个变量别名方便代码中进行访问。
namespace oceanbase
{
namespace share
{static const char* const OB_SV_ENABLE_RICH_ERROR_MSG ob_enable_rich_error_msg;static const char* const OB_SV_LOG_ROW_VALUE_OPTIONS log_row_value_options;static const char* const OB_SV_MAX_READ_STALE_TIME ob_max_read_stale_time;...... 获取 session 变量最新值
内部查询调用 get_sys_var() 函数通过下标 idx 获取 session 变量的值。
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;
} 获取 session 变量缓存值
获取变量缓存值就是从 ObBasicSessionInfo::sys_vars_cache_数组中取值通常是封装了一些函数然后再调用 sys_vars_cache_ 的 get_xxx() 接口这些接口就是用 DEF_SYS_VAR_CACHE_FUNCS 宏定义的。
举几个例子
class ObBasicSessionInfo
{int64_t get_trx_lock_timeout() const{return sys_vars_cache_.get_ob_trx_lock_timeout();}int64_t get_ob_max_read_stale_time() {return sys_vars_cache_.get_ob_max_read_stale_time();}int get_sql_throttle_current_priority(int64_t sql_throttle_current_priority){sql_throttle_current_priority sys_vars_cache_.get_sql_throttle_current_priority();return common::OB_SUCCESS;}
只要能够访问 session就可以访问 session 中的变量比如获取 ob_max_read_stale_time
session-get_ob_max_read_stale_time(); 变量更新
外部修改
global 变量修改流程
更新 global 变量就是更新其对应的 schema执行如“set global xxx xx”的命令时触发以下调用流程
ObMPQuery::do_process()ObVariableSetExecutor::execute(ObExecContext ctx, ObVariableSetStmt stmt)ObVariableSetExecutor::update_global_variables()
该函数先对更新数据做合法性检查然后构造一个新的 sysvar_schema 加入到 rpc 参数中最后发送更新系统变量的请求到其他节点。
int ObVariableSetExecutor::update_global_variables(ObExecContext ctx,ObDDLStmt stmt,const ObSetVar set_var,const ObObj val)
{// 合法性检查} else if (set_var.var_name_ OB_SV_MAX_READ_STALE_TIME) {int64_t max_read_stale_time 0;if (OB_FAIL(val.get_int(max_read_stale_time))) {LOG_WARN(fail to get int value, K(ret), K(val));} else if (max_read_stale_time ! ObSysVarFactory::INVALID_MAX_READ_STALE_TIME max_read_stale_time GCONF.weak_read_version_refresh_interval) {ret OB_INVALID_ARGUMENT;LOG_USER_ERROR(OB_INVALID_ARGUMENT,max_read_stale_time is smaller than weak_read_version_refresh_interval);}// 构造新的 sysvar_schema添加到参数 arg 的 sys_var_list_ 列表中if (OB_SUCC(ret)) {......} else if (OB_FAIL(sysvar_schema.set_name(set_var.var_name_))) {LOG_WARN(set sysvar schema name failed, K(ret));} else if (OB_FAIL(sysvar_schema.set_value(val_str))) {LOG_WARN(set sysvar schema value failed, K(ret));} else {sysvar_schema.set_tenant_id(arg.tenant_id_);if (OB_FAIL(arg.sys_var_list_.push_back(sysvar_schema))) {LOG_WARN(store sys var to array failed, K(ret));}}}// 想其他节点发送更新变量的rpc请求if (OB_SUCC(ret)) {if (OB_ISNULL(task_exec_ctx GET_TASK_EXECUTOR_CTX(ctx)) ||OB_ISNULL(common_rpc_proxy task_exec_ctx-get_common_rpc())) {ret OB_NOT_INIT;LOG_WARN(task exec ctx or common rpc proxy is NULL, K(ret), K(task_exec_ctx), K(common_rpc_proxy));} else if (OB_FAIL(common_rpc_proxy-modify_system_variable(arg))) {LOG_WARN(rpc proxy alter system variable failed, K(ret));} else {}}return ret;
} session 变量修改流程
更新session变量值的同时还会更新对应缓存数据的值执行如“set xxx xx”的命令时会触发以下调用流程
ObBasicSysVar::session_update()ObBasicSessionInfo::update_sys_variable_by_name()
通过变量名 var 获取 var_id然后调用 update_sys_variable() 传入 val。
int ObBasicSessionInfo::update_sys_variable_by_name(const ObString var, const ObObj val)
{int ret OB_SUCCESS;ObSysVarClassType var_id SYS_VAR_INVALID;if (var.empty()) {ret OB_INVALID_ARGUMENT;LOG_WARN(invalid variable name, K(var), K(val), K(ret));} else if (SYS_VAR_INVALID (var_id ObSysVarFactory::find_sys_var_id_by_name(var))) {ret OB_ERR_SYS_VARIABLE_UNKNOWN;LOG_WARN(unknown variable, K(var), K(val), K(ret));} else if (OB_FAIL(update_sys_variable(var_id, val))) {LOG_WARN(failed to update sys variable, K(var), K(val), K(ret));} else {}return ret;
}
ObBasicSessionInfo::update_sys_variable()
先记录变量旧值然后更新 session 变量缓存值再更新 sys_var 的值也就是 sys_vars_[store_idx] 的值。
int ObBasicSessionInfo::update_sys_variable(const ObSysVarClassType sys_var_id, const ObObj val)
{} else if (is_track_session_info()) {// 记录修改变量的旧值if (OB_FAIL(track_sys_var(sys_var_id, sys_var-get_value()))) {LOG_WARN(failed to track sys var, K(ret), K(sys_var_id), K(val));if (OB_SUCC(ret)) {if (OB_FAIL(process_session_variable(sys_var_id, val, false /*check_timezone_valid*/,true /*is_update_sys_var*/))) {LOG_WARN(process system variable error, K(sys_var_id), K(val), K(ret));} else {sys_var-set_value(val);
ObBasicSessionInfo::process_session_variable()
基于传入的变量类型 var更新对应 session 变量的缓存值即 inc_data_.xxx 。
OB_INLINE int ObBasicSessionInfo::process_session_variable(ObSysVarClassType var, const ObObj val,const bool check_timezone_valid/*true*/, const bool is_update_sys_var/*false*/)
{int ret OB_SUCCESS;switch (var) {case SYS_VAR_OB_MAX_READ_STALE_TIME: {int64_t max_read_stale_time 0;if (OB_FAIL(val.get_int(max_read_stale_time))) {LOG_WARN(fail to get int value, K(ret), K(val));} else if (max_read_stale_time ! ObSysVarFactory::INVALID_MAX_READ_STALE_TIME max_read_stale_time GCONF.weak_read_version_refresh_interval) {ret OB_INVALID_ARGUMENT;LOG_USER_ERROR(OB_INVALID_ARGUMENT,max_read_stale_time is smaller than weak_read_version_refresh_interval);} else {sys_vars_cache_.set_ob_max_read_stale_time(max_read_stale_time);}break;} 内部更新
内部更新主要是指系统变量在各节点之间的同步机制。当一个节点执行 global 变量更新操作之后其他节点刷新 schema 的流程
ObServerSchemaUpdater::process_refresh_task(const ObServerSchemaTask task)ObMultiVersionSchemaService::refresh_and_add_schema()ObMultiVersionSchemaService::refresh_tenant_schema(const uint64_t tenant_id)ObServerSchemaService::refresh_schema(const ObRefreshSchemaStatus schema_status)
int ObServerSchemaService::refresh_schema(const ObRefreshSchemaStatus schema_status)
{} else if (is_full_schema) {if (OB_FAIL(refresh_full_schema(schema_status))) {LOG_WARN(tenant refresh full schema failed, K(ret), K(schema_status));}} else {if (OB_FAIL(refresh_increment_schema(schema_status))) {LOG_WARN(tenant refresh increment schema failed, K(ret), K(schema_status));}
ObServerSchemaService::refresh_increment_schema()
调用 get_increment_schema_operations() 获取增量 schema 操作 schema_operations然后调用 replay_log() 回放日志再调用 update_schema_mgr() 更新 schema。
int ObServerSchemaService::refresh_increment_schema(const ObRefreshSchemaStatus schema_status)
{if (OB_SUCC(ret) !core_schema_change !sys_schema_change) {const int64_t fetch_version std::max(core_schema_version, schema_version);if (OB_FAIL(schema_mgr_for_cache_map_.get_refactored(tenant_id, schema_mgr_for_cache))) {LOG_WARN(fail to get schema_mgr_for_cache, KR(ret), K(schema_status));} else if (OB_ISNULL(schema_mgr_for_cache)) {ret OB_ERR_UNEXPECTED;LOG_WARN(schema mgr for cache is null, KR(ret));} else if (OB_FAIL(schema_service_-get_increment_schema_operations(schema_status,local_schema_version, fetch_version, sql_client, schema_operations))) {LOG_WARN(get_increment_schema_operations failed, KR(ret), K(schema_status),K(local_schema_version), K(fetch_version));} else if (schema_operations.count() 0) {// new cacheSMART_VAR(AllSchemaKeys, all_keys) {if (OB_FAIL(replay_log(*schema_mgr_for_cache, schema_operations, all_keys))) {LOG_WARN(replay_log failed, KR(ret), K(schema_status), K(schema_operations));} else if (OB_FAIL(update_schema_mgr(sql_client, schema_status,*schema_mgr_for_cache, fetch_version, all_keys))){LOG_WARN(update schema mgr failed, KR(ret), K(schema_status));}}} else {}if (OB_SUCC(ret)) {schema_mgr_for_cache-set_schema_version(fetch_version);if (OB_FAIL(publish_schema(tenant_id))) {LOG_WARN(publish_schema failed, KR(ret), K(schema_status));} else {LOG_INFO(change schema version, K(schema_status), K(schema_version), K(core_schema_version));break;}} 小结
相比前面介绍过的配置项系统变量的内部机制更加复杂一个系统变量不仅同时具有 global 级别和 session 级别的副本还可能在 session 上定义了一份缓存值。在新增系统变量时要根据实际需求去定义变量的标签和缓存在使用过程中也要注意取到的值是不是符合预期。
目前对配置项和系统变量的源码解析基本完成前文中也偶尔提到过“合法性检查”这一概念在下一篇文章中将会介绍它们是如何进行合法性检查的以及相关的函数和规则是怎样定义的欢迎大家持续关注。 参考文档
系统变量总览什么是系统变量如何使用系统变量OceanBase源码 文章转载自： http://www.morning.tbqxh.cn.gov.cn.tbqxh.cn http://www.morning.fgsqz.cn.gov.cn.fgsqz.cn http://www.morning.zxybw.cn.gov.cn.zxybw.cn http://www.morning.heleyo.com.gov.cn.heleyo.com http://www.morning.mtbth.cn.gov.cn.mtbth.cn http://www.morning.ykgp.cn.gov.cn.ykgp.cn http://www.morning.zqkms.cn.gov.cn.zqkms.cn http://www.morning.wjjxr.cn.gov.cn.wjjxr.cn http://www.morning.ktfbl.cn.gov.cn.ktfbl.cn http://www.morning.nfqyk.cn.gov.cn.nfqyk.cn http://www.morning.bxqry.cn.gov.cn.bxqry.cn http://www.morning.qdlr.cn.gov.cn.qdlr.cn http://www.morning.bppml.cn.gov.cn.bppml.cn http://www.morning.pswzc.cn.gov.cn.pswzc.cn http://www.morning.lwqst.cn.gov.cn.lwqst.cn http://www.morning.horihe.com.gov.cn.horihe.com http://www.morning.vehna.com.gov.cn.vehna.com http://www.morning.wfyqn.cn.gov.cn.wfyqn.cn http://www.morning.dnls.cn.gov.cn.dnls.cn http://www.morning.qrzqd.cn.gov.cn.qrzqd.cn http://www.morning.cdlewan.com.gov.cn.cdlewan.com http://www.morning.ydyjf.cn.gov.cn.ydyjf.cn http://www.morning.wgrm.cn.gov.cn.wgrm.cn http://www.morning.pmtky.cn.gov.cn.pmtky.cn http://www.morning.clbgy.cn.gov.cn.clbgy.cn http://www.morning.nwtmy.cn.gov.cn.nwtmy.cn http://www.morning.mwmtk.cn.gov.cn.mwmtk.cn http://www.morning.zkbxx.cn.gov.cn.zkbxx.cn http://www.morning.fkmrj.cn.gov.cn.fkmrj.cn http://www.morning.hlhqs.cn.gov.cn.hlhqs.cn http://www.morning.clpdm.cn.gov.cn.clpdm.cn http://www.morning.tsxg.cn.gov.cn.tsxg.cn http://www.morning.hxxyp.cn.gov.cn.hxxyp.cn http://www.morning.ztnmc.cn.gov.cn.ztnmc.cn http://www.morning.rgmls.cn.gov.cn.rgmls.cn http://www.morning.coffeedelsol.com.gov.cn.coffeedelsol.com http://www.morning.dnycx.cn.gov.cn.dnycx.cn http://www.morning.mnnxt.cn.gov.cn.mnnxt.cn http://www.morning.zsleyuan.cn.gov.cn.zsleyuan.cn http://www.morning.npkrm.cn.gov.cn.npkrm.cn http://www.morning.jjhrj.cn.gov.cn.jjhrj.cn http://www.morning.jkcnq.cn.gov.cn.jkcnq.cn http://www.morning.prsxj.cn.gov.cn.prsxj.cn http://www.morning.mkrjf.cn.gov.cn.mkrjf.cn http://www.morning.rqsnl.cn.gov.cn.rqsnl.cn http://www.morning.bljcb.cn.gov.cn.bljcb.cn http://www.morning.rbylq.cn.gov.cn.rbylq.cn http://www.morning.krjrb.cn.gov.cn.krjrb.cn http://www.morning.nlysd.cn.gov.cn.nlysd.cn http://www.morning.nsppc.cn.gov.cn.nsppc.cn http://www.morning.lqws.cn.gov.cn.lqws.cn http://www.morning.mxgpp.cn.gov.cn.mxgpp.cn http://www.morning.twpq.cn.gov.cn.twpq.cn http://www.morning.fssjw.cn.gov.cn.fssjw.cn http://www.morning.zcckq.cn.gov.cn.zcckq.cn http://www.morning.lxbml.cn.gov.cn.lxbml.cn http://www.morning.jrlgz.cn.gov.cn.jrlgz.cn http://www.morning.ntyks.cn.gov.cn.ntyks.cn http://www.morning.nrzbq.cn.gov.cn.nrzbq.cn http://www.morning.mpflb.cn.gov.cn.mpflb.cn http://www.morning.gyqnc.cn.gov.cn.gyqnc.cn http://www.morning.fbhmn.cn.gov.cn.fbhmn.cn http://www.morning.rxhs.cn.gov.cn.rxhs.cn http://www.morning.hxmqb.cn.gov.cn.hxmqb.cn http://www.morning.prmbn.cn.gov.cn.prmbn.cn http://www.morning.qjlkp.cn.gov.cn.qjlkp.cn http://www.morning.qjmnl.cn.gov.cn.qjmnl.cn http://www.morning.rszbj.cn.gov.cn.rszbj.cn http://www.morning.kxscs.cn.gov.cn.kxscs.cn http://www.morning.srbfp.cn.gov.cn.srbfp.cn http://www.morning.nbybb.cn.gov.cn.nbybb.cn http://www.morning.fzqfb.cn.gov.cn.fzqfb.cn http://www.morning.jwxnr.cn.gov.cn.jwxnr.cn http://www.morning.wzyfk.cn.gov.cn.wzyfk.cn http://www.morning.yqkmd.cn.gov.cn.yqkmd.cn http://www.morning.pcwzb.cn.gov.cn.pcwzb.cn http://www.morning.wknj.cn.gov.cn.wknj.cn http://www.morning.zdnrb.cn.gov.cn.zdnrb.cn http://www.morning.bfjyp.cn.gov.cn.bfjyp.cn http://www.morning.zqfz.cn.gov.cn.zqfz.cn
