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

[CP] Fix: session sync related bug

Browse Source
This commit is contained in:
yaojing624
2023-09-26 06:10:06 +00:00
committed by ob-robot
parent 0be5791bd5
commit 8f6d380ccd
4 changed files with 68 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
src/observer/mysql/obmp_utils.cpp
View File

@ -236,7 +236,7 @@ int ObMPUtils::append_modfied_sess_info(common::ObIAllocator &allocator,
char *buf = NULL;
int64_t size = 0;
int32_t sess_size[SESSION_SYNC_MAX_TYPE];
int64_t sess_size[SESSION_SYNC_MAX_TYPE];
for (int64_t i=0; OB_SUCC(ret) && i < SESSION_SYNC_MAX_TYPE; i++) {
oceanbase::sql::SessionSyncInfoType info_type = (oceanbase::sql::SessionSyncInfoType)(i);
sess_size[i] = 0;
@ -246,9 +246,12 @@ int ObMPUtils::append_modfied_sess_info(common::ObIAllocator &allocator,
if (info_type == SESSION_SYNC_SYS_VAR && !need_sync_sys_var) {
// do nothing.
} else if (encoder->is_changed_) {
sess_size[i] = encoder->get_serialize_size(sess);
if (OB_FAIL(encoder->get_serialize_size(sess, sess_size[i]))) {
LOG_WARN("fail to get serialize size", K(info_type), K(ret));
} else {
size += ObProtoTransUtil::get_serialize_size(sess_size[i]);
LOG_DEBUG("get seri size", K(sess_size[i]), K(encoder->get_serialize_size(sess)));
LOG_DEBUG("get seri size", K(sess_size[i]));
}
} else {
// encoder->is_changed_ = false;
}

13
src/sql/session/ob_basic_session_info.cpp
View File

@ -3787,6 +3787,7 @@ bool ObBasicSessionInfo::is_sync_sys_var(share::ObSysVarClassType sys_var_id) co
case SYS_VAR_OB_PROXY_PARTITION_HIT:
case SYS_VAR_VERSION_COMMENT:
case SYS_VAR_OB_LAST_SCHEMA_VERSION:
case SYS_VAR__OB_PROXY_WEAKREAD_FEEDBACK:
not_need_serialize = true;
break;
default:
@ -3977,15 +3978,13 @@ int ObBasicSessionInfo::deserialize_sync_sys_vars(int64_t &deserialize_sys_var_c
LOG_ERROR("process system variable error", K(ret), K(*sys_var));
} else if (sys_var->is_influence_plan()
&& FALSE_IT(is_influence_plan_cache_sys_var = true)) {
// do nothing.
// add all deserialize sys_var id.
} else if (!is_error_sync && OB_FAIL(tmp_sys_var_inc_info.add_sys_var_id(sys_var_id))) {
LOG_WARN("fail to add sys var id", K(sys_var_id), K(ret));
} else {
LOG_TRACE("deserialize sync sys var", K(sys_var_id), K(*sys_var),
K(sessid_), K(proxy_sessid_));
}
// add all deserialize sys_var id.
if (OB_SUCC(ret) && !is_error_sync && OB_FAIL(tmp_sys_var_inc_info.add_sys_var_id(sys_var_id))) {
LOG_WARN("fail to add sys var id", K(sys_var_id), K(ret));
}
}
if (OB_SUCC(ret) && !is_error_sync) {
if (OB_FAIL(sync_default_sys_vars(sys_var_inc_info_, tmp_sys_var_inc_info,
@ -4475,8 +4474,6 @@ OB_DEF_DESERIALIZE(ObBasicSessionInfo)
if (OB_FAIL(serialization::decode(buf, data_len, pos, tmp_sys_var_id))) {
LOG_WARN("fail to deserialize sys var id", K(data_len), K(pos), K(ret));
} else if (FALSE_IT(sys_var_id = static_cast<ObSysVarClassType>(tmp_sys_var_id))) {
} else if (OB_FAIL(sys_var_inc_info_.add_sys_var_id(sys_var_id))) {
LOG_WARN("fail to add sys var id", K(sys_var_id), K(ret));
} else if (OB_FAIL(ObSysVarFactory::calc_sys_var_store_idx(sys_var_id, store_idx))) {
if (OB_SYS_VARS_MAYBE_DIFF_VERSION == ret) {
// 可能是版本不同，为了兼容，跳过这段数据，并继续循环
@ -4492,6 +4489,8 @@ OB_DEF_DESERIALIZE(ObBasicSessionInfo)
} else {
LOG_ERROR("invalid sys var id", K(sys_var_id), K(ret));
}
} else if (OB_FAIL(sys_var_inc_info_.add_sys_var_id(sys_var_id))) {
LOG_WARN("fail to add sys var id", K(sys_var_id), K(ret));
} else if (OB_FAIL(create_sys_var(sys_var_id, store_idx, sys_var))) {
LOG_WARN("fail to create sys var", K(sys_var_id), K(ret));
} else if (OB_ISNULL(sys_var)) {

75
src/sql/session/ob_sql_session_info.cpp
View File

@ -3052,9 +3052,8 @@ int ObErrorSyncSysVarEncoder::deserialize(ObSQLSessionInfo &sess, const char *bu
return ret;
}
int64_t ObErrorSyncSysVarEncoder::get_serialize_size(ObSQLSessionInfo& sess) const {
int ObErrorSyncSysVarEncoder::get_serialize_size(ObSQLSessionInfo& sess, int64_t &len) const {
int ret = OB_SUCCESS;
int64_t len = 0;
ObSEArray<ObSysVarClassType, ObSysVarFactory::ALL_SYS_VARS_COUNT> sys_var_delta_ids;
if (OB_FAIL(sess.get_error_sync_sys_vars(sys_var_delta_ids))) {
LOG_WARN("failed to calc need serialize vars", K(ret));
@ -3063,7 +3062,7 @@ int64_t ObErrorSyncSysVarEncoder::get_serialize_size(ObSQLSessionInfo& sess) con
} else {
LOG_DEBUG("success serialize size sys var delta", K(ret), K(sys_var_delta_ids.count()), K(len));
}
return len;
return ret;
}
int ObErrorSyncSysVarEncoder::fetch_sess_info(ObSQLSessionInfo &sess, char *buf, const int64_t length, int64_t &pos)
@ -3198,9 +3197,8 @@ int ObSysVarEncoder::deserialize(ObSQLSessionInfo &sess, const char *buf,
return ret;
}
int64_t ObSysVarEncoder::get_serialize_size(ObSQLSessionInfo& sess) const {
int ObSysVarEncoder::get_serialize_size(ObSQLSessionInfo& sess, int64_t &len) const {
int ret = OB_SUCCESS;
int64_t len = 0;
ObSEArray<ObSysVarClassType, ObSysVarFactory::ALL_SYS_VARS_COUNT> sys_var_delta_ids;
if (OB_FAIL(sess.get_sync_sys_vars(sys_var_delta_ids))) {
LOG_WARN("failed to calc need serialize vars", K(ret));
@ -3209,7 +3207,7 @@ int64_t ObSysVarEncoder::get_serialize_size(ObSQLSessionInfo& sess) const {
} else {
LOG_DEBUG("success serialize size sys var delta", K(ret), K(sys_var_delta_ids.count()), K(len));
}
return len;
return ret;
}
int ObSysVarEncoder::fetch_sess_info(ObSQLSessionInfo &sess, char *buf, const int64_t length, int64_t &pos)
@ -3224,7 +3222,8 @@ int ObSysVarEncoder::fetch_sess_info(ObSQLSessionInfo &sess, char *buf, const in
if (ObSysVariables::get_sys_var_id(j) == SYS_VAR_SERVER_UUID ||
ObSysVariables::get_sys_var_id(j) == SYS_VAR_OB_PROXY_PARTITION_HIT ||
ObSysVariables::get_sys_var_id(j) == SYS_VAR_OB_STATEMENT_TRACE_ID ||
ObSysVariables::get_sys_var_id(j) == SYS_VAR_VERSION_COMMENT) {
ObSysVariables::get_sys_var_id(j) == SYS_VAR_VERSION_COMMENT ||
ObSysVariables::get_sys_var_id(j) == SYS_VAR__OB_PROXY_WEAKREAD_FEEDBACK) {
// no need sync sys var
continue;
}
@ -3245,7 +3244,8 @@ int64_t ObSysVarEncoder::get_fetch_sess_info_size(ObSQLSessionInfo& sess)
if (ObSysVariables::get_sys_var_id(j) == SYS_VAR_SERVER_UUID ||
ObSysVariables::get_sys_var_id(j) == SYS_VAR_OB_PROXY_PARTITION_HIT ||
ObSysVariables::get_sys_var_id(j) == SYS_VAR_OB_STATEMENT_TRACE_ID ||
ObSysVariables::get_sys_var_id(j) == SYS_VAR_VERSION_COMMENT) {
ObSysVariables::get_sys_var_id(j) == SYS_VAR_VERSION_COMMENT ||
ObSysVariables::get_sys_var_id(j) == SYS_VAR__OB_PROXY_WEAKREAD_FEEDBACK) {
// no need sync sys var
continue;
}
@ -3297,7 +3297,8 @@ int ObSysVarEncoder::display_sess_info(ObSQLSessionInfo &sess, const char* curre
if (ObSysVariables::get_sys_var_id(j) == SYS_VAR_SERVER_UUID ||
ObSysVariables::get_sys_var_id(j) == SYS_VAR_OB_PROXY_PARTITION_HIT ||
ObSysVariables::get_sys_var_id(j) == SYS_VAR_OB_STATEMENT_TRACE_ID ||
ObSysVariables::get_sys_var_id(j) == SYS_VAR_VERSION_COMMENT) {
ObSysVariables::get_sys_var_id(j) == SYS_VAR_VERSION_COMMENT ||
ObSysVariables::get_sys_var_id(j) == SYS_VAR__OB_PROXY_WEAKREAD_FEEDBACK) {
// no need sync sys var
continue;
}
@ -3391,8 +3392,10 @@ int ObAppInfoEncoder::deserialize(ObSQLSessionInfo &sess, const char *buf, const
return ret;
}
int64_t ObAppInfoEncoder::get_serialize_size(ObSQLSessionInfo& sess) const {
return sess.get_client_app_info().get_serialize_size();
int ObAppInfoEncoder::get_serialize_size(ObSQLSessionInfo& sess, int64_t &len) const {
int ret = OB_SUCCESS;
len = sess.get_client_app_info().get_serialize_size();
return ret;
}
int ObAppInfoEncoder::fetch_sess_info(ObSQLSessionInfo &sess, char *buf, const int64_t length, int64_t &pos)
@ -3406,7 +3409,9 @@ int ObAppInfoEncoder::fetch_sess_info(ObSQLSessionInfo &sess, char *buf, const i
int64_t ObAppInfoEncoder::get_fetch_sess_info_size(ObSQLSessionInfo& sess)
{
return get_serialize_size(sess);
int64_t len = 0;
get_serialize_size(sess, len);
return len;
}
int ObAppInfoEncoder::compare_sess_info(const char* current_sess_buf, int64_t current_sess_length,
@ -3523,11 +3528,11 @@ int ObClientIdInfoEncoder::deserialize(ObSQLSessionInfo &sess, const char *buf,
}
return ret;
}
int64_t ObClientIdInfoEncoder::get_serialize_size(ObSQLSessionInfo& sess) const
int ObClientIdInfoEncoder::get_serialize_size(ObSQLSessionInfo& sess, int64_t &len) const
{
int64_t len = 0;
int ret = OB_SUCCESS;
OB_UNIS_ADD_LEN(sess.get_client_identifier());
return len;
return ret;
}
int ObClientIdInfoEncoder::fetch_sess_info(ObSQLSessionInfo &sess, char *buf, const int64_t length, int64_t &pos)
@ -3541,7 +3546,9 @@ int ObClientIdInfoEncoder::fetch_sess_info(ObSQLSessionInfo &sess, char *buf, co
int64_t ObClientIdInfoEncoder::get_fetch_sess_info_size(ObSQLSessionInfo& sess)
{
return get_serialize_size(sess);
int64_t len = 0;
get_serialize_size(sess, len);
return len;
}
int ObClientIdInfoEncoder::compare_sess_info(const char* current_sess_buf, int64_t current_sess_length,
@ -3622,15 +3629,15 @@ int ObAppCtxInfoEncoder::deserialize(ObSQLSessionInfo &sess, const char *buf, co
}
return ret;
}
int64_t ObAppCtxInfoEncoder::get_serialize_size(ObSQLSessionInfo& sess) const
int ObAppCtxInfoEncoder::get_serialize_size(ObSQLSessionInfo& sess, int64_t &len) const
{
int64_t len = 0;
int ret = OB_SUCCESS;
ObContextsMap &map = sess.get_contexts_map();
OB_UNIS_ADD_LEN(map.size());
for (auto it = map.begin(); it != map.end(); ++it) {
OB_UNIS_ADD_LEN(*it->second);
}
return len;
return ret;
}
int ObAppCtxInfoEncoder::fetch_sess_info(ObSQLSessionInfo &sess, char *buf, const int64_t length, int64_t &pos)
@ -3644,7 +3651,9 @@ int ObAppCtxInfoEncoder::fetch_sess_info(ObSQLSessionInfo &sess, char *buf, cons
int64_t ObAppCtxInfoEncoder::get_fetch_sess_info_size(ObSQLSessionInfo& sess)
{
return get_serialize_size(sess);
int64_t len = 0;
get_serialize_size(sess, len);
return len;
}
int ObAppCtxInfoEncoder::compare_sess_info(const char* current_sess_buf, int64_t current_sess_length,
@ -3772,9 +3781,9 @@ int ObSequenceCurrvalEncoder::deserialize(ObSQLSessionInfo &sess, const char *bu
return ret;
}
int64_t ObSequenceCurrvalEncoder::get_serialize_size(ObSQLSessionInfo& sess) const
int ObSequenceCurrvalEncoder::get_serialize_size(ObSQLSessionInfo& sess, int64_t &len) const
{
int64_t len = 0;
int ret = OB_SUCCESS;
ObSequenceCurrvalMap &map = sess.get_sequence_currval_map();
OB_UNIS_ADD_LEN(map.size());
for (auto it = map.begin(); it != map.end(); ++it) {
@ -3789,7 +3798,7 @@ int64_t ObSequenceCurrvalEncoder::get_serialize_size(ObSQLSessionInfo& sess) con
OB_UNIS_ADD_LEN(it->first.dblink_id_);
OB_UNIS_ADD_LEN(it->second);
}
return len;
return ret;
}
int ObSequenceCurrvalEncoder::fetch_sess_info(ObSQLSessionInfo &sess, char *buf,
@ -3804,7 +3813,9 @@ int ObSequenceCurrvalEncoder::fetch_sess_info(ObSQLSessionInfo &sess, char *buf,
int64_t ObSequenceCurrvalEncoder::get_fetch_sess_info_size(ObSQLSessionInfo& sess)
{
return get_serialize_size(sess);
int64_t len = 0;
get_serialize_size(sess, len);
return len;
}
int ObSequenceCurrvalEncoder::compare_sess_info(const char* current_sess_buf,
@ -4040,9 +4051,11 @@ int ObControlInfoEncoder::deserialize(ObSQLSessionInfo &sess, const char *buf, c
return ret;
}
int64_t ObControlInfoEncoder::get_serialize_size(ObSQLSessionInfo& sess) const
int ObControlInfoEncoder::get_serialize_size(ObSQLSessionInfo& sess, int64_t &len) const
{
return sess.get_control_info().get_serialize_size() + 6 + sizeof(bool);
int ret = OB_SUCCESS;
len = sess.get_control_info().get_serialize_size() + 6 + sizeof(bool);
return ret;
}
int ObControlInfoEncoder::fetch_sess_info(ObSQLSessionInfo &sess, char *buf, const int64_t length, int64_t &pos)
@ -4056,7 +4069,9 @@ int ObControlInfoEncoder::fetch_sess_info(ObSQLSessionInfo &sess, char *buf, con
int64_t ObControlInfoEncoder::get_fetch_sess_info_size(ObSQLSessionInfo& sess)
{
return get_serialize_size(sess);
int64_t len = 0;
get_serialize_size(sess, len);
return len;
}
int ObControlInfoEncoder::compare_sess_info(const char* current_sess_buf, int64_t current_sess_length,
@ -4123,9 +4138,11 @@ int CLS::deserialize(ObSQLSessionInfo &sess, const char *buf, const int64_t leng
{ \
return ObSqlTransControl::update_txn_##func##_state(sess, buf, length, pos); \
} \
int64_t CLS::get_serialize_size(ObSQLSessionInfo &sess) const \
int CLS::get_serialize_size(ObSQLSessionInfo &sess, int64_t &len) const \
{ \
return ObSqlTransControl::get_txn_##func##_state_serialize_size(sess); \
int ret = OB_SUCCESS; \
len = ObSqlTransControl::get_txn_##func##_state_serialize_size(sess); \
return ret; \
} \
int CLS::fetch_sess_info(ObSQLSessionInfo &sess, char *buf, const int64_t data_len, int64_t &pos) \
{ \

18
src/sql/session/ob_sql_session_info.h
View File

@ -249,7 +249,7 @@ public:
virtual ~ObSessInfoEncoder() {}
virtual int serialize(ObSQLSessionInfo &sess, char *buf, const int64_t length, int64_t &pos) = 0;
virtual int deserialize(ObSQLSessionInfo &sess, const char *buf, const int64_t length, int64_t &pos) = 0;
virtual int64_t get_serialize_size(ObSQLSessionInfo& sess) const = 0;
virtual int get_serialize_size(ObSQLSessionInfo& sess, int64_t &length) const = 0;
// When implementing new information synchronization,
// it is necessary to add a self-verification interface
// include fetch, compare, and error display.
@ -269,7 +269,7 @@ public:
~ObSysVarEncoder() {}
int serialize(ObSQLSessionInfo &sess, char *buf, const int64_t length, int64_t &pos);
int deserialize(ObSQLSessionInfo &sess, const char *buf, const int64_t length, int64_t &pos);
int64_t get_serialize_size(ObSQLSessionInfo& sess) const;
int get_serialize_size(ObSQLSessionInfo& sess, int64_t &length) const;
int fetch_sess_info(ObSQLSessionInfo &sess, char *buf, const int64_t length, int64_t &pos);
int64_t get_fetch_sess_info_size(ObSQLSessionInfo& sess);
int compare_sess_info(const char* current_sess_buf, int64_t current_sess_length,
@ -284,7 +284,7 @@ public:
~ObAppInfoEncoder() {}
int serialize(ObSQLSessionInfo &sess, char *buf, const int64_t length, int64_t &pos);
int deserialize(ObSQLSessionInfo &sess, const char *buf, const int64_t length, int64_t &pos);
int64_t get_serialize_size(ObSQLSessionInfo& sess) const;
int get_serialize_size(ObSQLSessionInfo& sess, int64_t &length) const;
int fetch_sess_info(ObSQLSessionInfo &sess, char *buf, const int64_t length, int64_t &pos);
int64_t get_fetch_sess_info_size(ObSQLSessionInfo& sess);
int compare_sess_info(const char* current_sess_buf, int64_t current_sess_length,
@ -312,7 +312,7 @@ public:
virtual ~ObAppCtxInfoEncoder() {}
virtual int serialize(ObSQLSessionInfo &sess, char *buf, const int64_t length, int64_t &pos) override;
virtual int deserialize(ObSQLSessionInfo &sess, const char *buf, const int64_t length, int64_t &pos) override;
virtual int64_t get_serialize_size(ObSQLSessionInfo& sess) const override;
virtual int get_serialize_size(ObSQLSessionInfo& sess, int64_t &length) const override;
virtual int fetch_sess_info(ObSQLSessionInfo &sess, char *buf,
const int64_t length, int64_t &pos);
virtual int64_t get_fetch_sess_info_size(ObSQLSessionInfo& sess);
@ -327,7 +327,7 @@ public:
virtual ~ObClientIdInfoEncoder() {}
virtual int serialize(ObSQLSessionInfo &sess, char *buf, const int64_t length, int64_t &pos) override;
virtual int deserialize(ObSQLSessionInfo &sess, const char *buf, const int64_t length, int64_t &pos) override;
virtual int64_t get_serialize_size(ObSQLSessionInfo &sess) const override;
virtual int get_serialize_size(ObSQLSessionInfo &sess, int64_t &length) const override;
virtual int fetch_sess_info(ObSQLSessionInfo &sess, char *buf,
const int64_t length, int64_t &pos);
virtual int64_t get_fetch_sess_info_size(ObSQLSessionInfo& sess);
@ -343,7 +343,7 @@ public:
virtual ~ObSequenceCurrvalEncoder() {}
virtual int serialize(ObSQLSessionInfo &sess, char *buf, const int64_t length, int64_t &pos) override;
virtual int deserialize(ObSQLSessionInfo &sess, const char *buf, const int64_t length, int64_t &pos) override;
virtual int64_t get_serialize_size(ObSQLSessionInfo &sess) const override;
virtual int get_serialize_size(ObSQLSessionInfo &sess, int64_t &length) const override;
virtual int fetch_sess_info(ObSQLSessionInfo &sess, char *buf, const int64_t length, int64_t &pos) override;
virtual int64_t get_fetch_sess_info_size(ObSQLSessionInfo& sess) override;
virtual int compare_sess_info(const char* current_sess_buf, int64_t current_sess_length,
@ -358,7 +358,7 @@ public:
virtual ~ObControlInfoEncoder() {}
virtual int serialize(ObSQLSessionInfo &sess, char *buf, const int64_t length, int64_t &pos) override;
virtual int deserialize(ObSQLSessionInfo &sess, const char *buf, const int64_t length, int64_t &pos) override;
virtual int64_t get_serialize_size(ObSQLSessionInfo &sess) const override;
virtual int get_serialize_size(ObSQLSessionInfo &sess, int64_t &length) const override;
virtual int fetch_sess_info(ObSQLSessionInfo &sess, char *buf,
const int64_t length, int64_t &pos);
virtual int64_t get_fetch_sess_info_size(ObSQLSessionInfo& sess);
@ -378,7 +378,7 @@ public:
virtual ~ObErrorSyncSysVarEncoder() {}
virtual int serialize(ObSQLSessionInfo &sess, char *buf, const int64_t length, int64_t &pos) override;
virtual int deserialize(ObSQLSessionInfo &sess, const char *buf, const int64_t length, int64_t &pos) override;
virtual int64_t get_serialize_size(ObSQLSessionInfo &sess) const override;
virtual int get_serialize_size(ObSQLSessionInfo &sess, int64_t &length) const override;
virtual int fetch_sess_info(ObSQLSessionInfo &sess, char *buf, const int64_t length, int64_t &pos) override;
virtual int64_t get_fetch_sess_info_size(ObSQLSessionInfo& sess) override;
virtual int compare_sess_info(const char* current_sess_buf, int64_t current_sess_length,
@ -392,7 +392,7 @@ class CLS final : public ObSessInfoEncoder { \
public: \
int serialize(ObSQLSessionInfo &sess, char *buf, const int64_t length, int64_t &pos) override; \
int deserialize(ObSQLSessionInfo &sess, const char *buf, const int64_t length, int64_t &pos) override; \
int64_t get_serialize_size(ObSQLSessionInfo &sess) const override; \
int get_serialize_size(ObSQLSessionInfo &sess, int64_t &length) const override; \
int fetch_sess_info(ObSQLSessionInfo &sess, char *buf, const int64_t length, int64_t &pos) override; \
int64_t get_fetch_sess_info_size(ObSQLSessionInfo& sess) override; \
int compare_sess_info(const char* current_sess_buf, int64_t current_sess_length, const char* last_sess_buf, int64_t last_sess_length) override; \