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

Refactor inner sql safe time mechanism

Browse Source
This commit is contained in:
obdev 2022-11-11 10:09:49 +00:00 committed by wangzelin.wzl
parent f5fa404be1
commit aa75d00d5c
6 changed files with 24 additions and 102 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

67
src/observer/ob_inner_sql_connection.cpp
View File

@ -68,12 +68,6 @@ public:
virtual int execute(sql::ObSql &engine, sql::ObSqlCtx &ctx, sql::ObResultSet &res) virtual int execute(sql::ObSql &engine, sql::ObSqlCtx &ctx, sql::ObResultSet &res)
{ {
/* !!!
* 内部sql，req_timeinfo_guard一定要在进入sql引擎定义
* !!!
*/
bool need_res_update_endtime = true;
observer::ObReqTimeGuard req_timeinfo_guard(!need_res_update_endtime);
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
// Deep copy sql, because sql may be destroyed before result iteration. // Deep copy sql, because sql may be destroyed before result iteration.
const int64_t alloc_size = sizeof(ObString) + sql_.length() + 1; // 1 for C terminate char const int64_t alloc_size = sizeof(ObString) + sql_.length() + 1; // 1 for C terminate char
@ -87,7 +81,6 @@ public:
MEMCPY(dup_sql->ptr(), sql_.ptr(), sql_.length()); MEMCPY(dup_sql->ptr(), sql_.ptr(), sql_.length());
dup_sql->ptr()[sql_.length()] = '\0'; dup_sql->ptr()[sql_.length()] = '\0';
res.get_session().store_query_string(*dup_sql); res.get_session().store_query_string(*dup_sql);
res.inc_need_update_endtime();
ret = engine.stmt_query(*dup_sql, ctx, res); ret = engine.stmt_query(*dup_sql, ctx, res);
} }
return ret; return ret;
@ -1195,12 +1188,6 @@ int ObInnerSQLConnection::start_transaction_inner(
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
ObString sql; ObString sql;
/* !!!
* 内部sql，req_timeinfo_guard一定要在进入sql引擎定义
* !!!
*/
bool need_res_update_endtime = true;
observer::ObReqTimeGuard req_timeinfo_guard(!need_res_update_endtime);
bool has_tenant_resource = false; bool has_tenant_resource = false;
if (with_snap_shot) { if (with_snap_shot) {
sql = ObString::make_string("START TRANSACTION WITH CONSISTENT SNAPSHOT"); sql = ObString::make_string("START TRANSACTION WITH CONSISTENT SNAPSHOT");
@ -1293,7 +1280,6 @@ int ObInnerSQLConnection::start_transaction_inner(
if (OB_SUCC(ret)) { if (OB_SUCC(ret)) {
set_is_in_trans(true); set_is_in_trans(true);
} }
res.inc_need_update_endtime();
} }
} }
@ -1307,9 +1293,6 @@ int ObInnerSQLConnection::register_multi_data_source(const uint64_t &tenant_id,
const int64_t buf_len) const int64_t buf_len)
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
bool need_res_update_endtime = true;
observer::ObReqTimeGuard req_timeinfo_guard(!need_res_update_endtime);
const bool local_execute = is_local_execute(GCONF.cluster_id, tenant_id); const bool local_execute = is_local_execute(GCONF.cluster_id, tenant_id);
transaction::ObTxDesc *tx_desc = nullptr; transaction::ObTxDesc *tx_desc = nullptr;
@ -1393,7 +1376,6 @@ int ObInnerSQLConnection::register_multi_data_source(const uint64_t &tenant_id,
} }
} }
} }
res.inc_need_update_endtime();
} }
@ -1413,9 +1395,6 @@ int ObInnerSQLConnection::lock_table(const uint64_t tenant_id,
const int64_t timeout_us) const int64_t timeout_us)
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
bool need_res_update_endtime = true;
observer::ObReqTimeGuard req_timeinfo_guard(!need_res_update_endtime);
bool has_tenant_resource = false; bool has_tenant_resource = false;
transaction::ObTxDesc *tx_desc = nullptr; transaction::ObTxDesc *tx_desc = nullptr;
@ -1490,7 +1469,6 @@ int ObInnerSQLConnection::lock_table(const uint64_t tenant_id,
} }
} }
} }
res.inc_need_update_endtime();
} }
return ret; return ret;
@ -1503,9 +1481,6 @@ int ObInnerSQLConnection::lock_tablet(const uint64_t tenant_id,
const int64_t timeout_us) const int64_t timeout_us)
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
bool need_res_update_endtime = true;
observer::ObReqTimeGuard req_timeinfo_guard(!need_res_update_endtime);
bool has_tenant_resource = false; bool has_tenant_resource = false;
transaction::ObTxDesc *tx_desc = nullptr; transaction::ObTxDesc *tx_desc = nullptr;
@ -1582,7 +1557,6 @@ int ObInnerSQLConnection::lock_tablet(const uint64_t tenant_id,
} }
} }
} }
res.inc_need_update_endtime();
} }
return ret; return ret;
@ -1653,12 +1627,6 @@ int ObInnerSQLConnection::rollback()
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
FLTSpanGuard(inner_rollback); FLTSpanGuard(inner_rollback);
/* !!!
* 内部sql，req_timeinfo_guard一定要在进入sql引擎定义
* !!!
*/
bool need_res_update_endtime = true;
observer::ObReqTimeGuard req_timeinfo_guard(!need_res_update_endtime);
ObSqlQueryExecutor executor("ROLLBACK"); ObSqlQueryExecutor executor("ROLLBACK");
bool has_tenant_resource = is_resource_conn() || OB_INVALID_ID == get_resource_conn_id(); bool has_tenant_resource = is_resource_conn() || OB_INVALID_ID == get_resource_conn_id();
if (!is_in_trans()) { if (!is_in_trans()) {
@ -1720,7 +1688,6 @@ int ObInnerSQLConnection::rollback()
} }
} }
} }
res.inc_need_update_endtime();
} }
} }
set_is_in_trans(false); set_is_in_trans(false);
@ -1732,12 +1699,6 @@ int ObInnerSQLConnection::commit()
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
FLTSpanGuard(inner_commit); FLTSpanGuard(inner_commit);
DEBUG_SYNC(BEFORE_INNER_SQL_COMMIT); DEBUG_SYNC(BEFORE_INNER_SQL_COMMIT);
/* !!!
* 内部sql，req_timeinfo_guard一定要在进入sql引擎定义
* !!!
*/
bool need_res_update_endtime = true;
observer::ObReqTimeGuard req_timeinfo_guard(!need_res_update_endtime);
ObSqlQueryExecutor executor("COMMIT"); ObSqlQueryExecutor executor("COMMIT");
bool has_tenant_resource = is_resource_conn() || OB_INVALID_ID == get_resource_conn_id(); bool has_tenant_resource = is_resource_conn() || OB_INVALID_ID == get_resource_conn_id();
if (!is_in_trans()) { if (!is_in_trans()) {
@ -1798,7 +1759,6 @@ int ObInnerSQLConnection::commit()
} }
} }
} }
res.inc_need_update_endtime();
} }
} }
set_is_in_trans(false); set_is_in_trans(false);
@ -1831,12 +1791,6 @@ int ObInnerSQLConnection::execute_write_inner(const uint64_t tenant_id, const Ob
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
FLTSpanGuard(inner_execute_write); FLTSpanGuard(inner_execute_write);
/* !!!
* 内部sql，req_timeinfo_guard一定要在进入sql引擎定义
* !!!
*/
bool need_res_update_endtime = true;
observer::ObReqTimeGuard req_timeinfo_guard(!need_res_update_endtime);
ObSqlQueryExecutor executor(sql); ObSqlQueryExecutor executor(sql);
const bool local_execute = is_local_execute(GCONF.cluster_id, tenant_id); const bool local_execute = is_local_execute(GCONF.cluster_id, tenant_id);
SMART_VAR(ObInnerSQLResult, res, get_session()) { SMART_VAR(ObInnerSQLResult, res, get_session()) {
@ -1943,7 +1897,6 @@ int ObInnerSQLConnection::execute_write_inner(const uint64_t tenant_id, const Ob
if (tenant_id < OB_MAX_RESERVED_TENANT_ID) { //only print log for sys table if (tenant_id < OB_MAX_RESERVED_TENANT_ID) { //only print log for sys table
LOG_INFO("execute write sql", K(ret), K(tenant_id), K(affected_rows), K(sql)); LOG_INFO("execute write sql", K(ret), K(tenant_id), K(affected_rows), K(sql));
} }
res.inc_need_update_endtime();
} }
return ret; return ret;
@ -2020,12 +1973,6 @@ int ObInnerSQLConnection::execute_read_inner(const int64_t cluster_id,
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
FLTSpanGuard(inner_execute_read); FLTSpanGuard(inner_execute_read);
ObInnerSQLReadContext *read_ctx = NULL; ObInnerSQLReadContext *read_ctx = NULL;
/* !!!
* 内部sql，req_timeinfo_guard一定要在进入sql引擎定义
* !!!
*/
bool need_res_update_endtime = true;
observer::ObReqTimeGuard req_timeinfo_guard(!need_res_update_endtime);
const static int64_t ctx_size = sizeof(ObInnerSQLReadContext); const static int64_t ctx_size = sizeof(ObInnerSQLReadContext);
static_assert(ctx_size <= ObISQLClient::ReadResult::BUF_SIZE, "buffer not enough"); static_assert(ctx_size <= ObISQLClient::ReadResult::BUF_SIZE, "buffer not enough");
ObSqlQueryExecutor executor(sql); ObSqlQueryExecutor executor(sql);
@ -2183,12 +2130,6 @@ int ObInnerSQLConnection::execute(
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
FLTSpanGuard(inner_execute); FLTSpanGuard(inner_execute);
/* !!!
* 内部sql，req_timeinfo_guard一定要在进入sql引擎定义
* !!!
*/
bool need_res_update_endtime = true;
observer::ObReqTimeGuard req_timeinfo_guard(!need_res_update_endtime);
SMART_VAR(ObInnerSQLResult, res, get_session()) { SMART_VAR(ObInnerSQLResult, res, get_session()) {
if (OB_FAIL(res.init())) { if (OB_FAIL(res.init())) {
LOG_WARN("init result set", K(ret)); LOG_WARN("init result set", K(ret));
@ -2217,7 +2158,6 @@ int ObInnerSQLConnection::execute(
} }
} }
LOG_INFO("execute executor", K(ret), K(tenant_id), K(executor)); LOG_INFO("execute executor", K(ret), K(tenant_id), K(executor));
res.inc_need_update_endtime();
} }
return ret; return ret;
} }
@ -2273,12 +2213,6 @@ int ObInnerSQLConnection::execute(const uint64_t tenant_id,
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
FLTSpanGuard(inner_execute); FLTSpanGuard(inner_execute);
ObInnerSQLReadContext *read_ctx = NULL; ObInnerSQLReadContext *read_ctx = NULL;
/* !!!
* 内部sql，req_timeinfo_guard一定要在进入sql引擎定义
* !!!
*/
bool need_res_update_endtime = true;
observer::ObReqTimeGuard req_timeinfo_guard(!need_res_update_endtime);
ObPsStmtInfoGuard ps_guard; ObPsStmtInfoGuard ps_guard;
ObPsStmtInfo *ps_info = NULL; ObPsStmtInfo *ps_info = NULL;
ObPsCache *ps_cache = get_session().get_ps_cache(); ObPsCache *ps_cache = get_session().get_ps_cache();
@ -2306,7 +2240,6 @@ int ObInnerSQLConnection::execute(const uint64_t tenant_id,
read_ctx->get_result().result_set().set_ps_protocol(); read_ctx->get_result().result_set().set_ps_protocol();
read_ctx->get_result().result_set().set_statement_id(stmt_id); read_ctx->get_result().result_set().set_statement_id(stmt_id);
read_ctx->get_result().result_set().set_stmt_type(stmt_type); read_ctx->get_result().result_set().set_stmt_type(stmt_type);
read_ctx->get_result().result_set().inc_need_update_endtime();
get_session().store_query_string(ps_info->get_ps_sql()); get_session().store_query_string(ps_info->get_ps_sql());
if (OB_FAIL(execute(params, read_ctx->get_result(), &read_ctx->get_vt_iter_factory(), if (OB_FAIL(execute(params, read_ctx->get_result(), &read_ctx->get_vt_iter_factory(),

11
src/observer/ob_inner_sql_result.cpp
View File

@ -58,7 +58,6 @@ ObInnerSQLResult::ObInnerSQLResult(ObSQLSessionInfo &session)
iter_end_(false), iter_end_(false),
is_read_(true), is_read_(true),
has_tenant_resource_(true), has_tenant_resource_(true),
need_update_cnt_(0),
tenant_(nullptr) tenant_(nullptr)
{ {
@ -111,16 +110,6 @@ ObInnerSQLResult::~ObInnerSQLResult()
if (remote_result_set_ != nullptr) { if (remote_result_set_ != nullptr) {
remote_result_set_->~ObRemoteResultSet(); remote_result_set_->~ObRemoteResultSet();
} }
if (need_update_cnt_ > 0) {
oceanbase::observer::ObReqTimeInfo *req_timeinfo = GET_TSI_MULT(observer::ObReqTimeInfo,
observer::ObReqTimeInfo::REQ_TIMEINFO_IDENTIFIER);
OB_ASSERT(NULL != req_timeinfo);
for (int i=0; i < need_update_cnt_; i++) {
req_timeinfo->update_end_time();
}
}
if (tenant_ != nullptr) { if (tenant_ != nullptr) {
tenant_->unlock(handle_); tenant_->unlock(handle_);
tenant_ = nullptr; tenant_ = nullptr;

2
src/observer/ob_inner_sql_result.h
View File

@ -177,7 +177,6 @@ private:
int find_idx(const char *col_name, int64_t &idx) const; int find_idx(const char *col_name, int64_t &idx) const;
int build_column_map() const; int build_column_map() const;
int inner_close(); int inner_close();
void inc_need_update_endtime() { need_update_cnt_++; }
static inline int check_extend_value(const common::ObObj &obj); static inline int check_extend_value(const common::ObObj &obj);
private: private:
@ -222,7 +221,6 @@ private:
bool iter_end_; bool iter_end_;
bool is_read_; //for some write sql , do not need prefetch 1 row in open bool is_read_; //for some write sql , do not need prefetch 1 row in open
bool has_tenant_resource_; bool has_tenant_resource_;
int64_t need_update_cnt_; // used to mark wether need to update endtime for timereqguard
omt::ObTenant *tenant_; omt::ObTenant *tenant_;
ObLDHandle handle_; ObLDHandle handle_;

8
src/observer/ob_req_time_service.h
View File

@ -133,13 +133,12 @@ private:
struct ObReqTimeGuard struct ObReqTimeGuard
{ {
ObReqTimeGuard(bool need_end = true) ObReqTimeGuard()
{ {
ObReqTimeInfo *req_timeinfo = GET_TSI_MULT(ObReqTimeInfo, ObReqTimeInfo *req_timeinfo = GET_TSI_MULT(ObReqTimeInfo,
ObReqTimeInfo::REQ_TIMEINFO_IDENTIFIER); ObReqTimeInfo::REQ_TIMEINFO_IDENTIFIER);
OB_ASSERT(NULL != req_timeinfo); OB_ASSERT(NULL != req_timeinfo);
req_timeinfo->update_start_time(); req_timeinfo->update_start_time();
need_end_ = need_end;
} }
~ObReqTimeGuard() ~ObReqTimeGuard()
@ -147,11 +146,8 @@ struct ObReqTimeGuard
ObReqTimeInfo *req_timeinfo = GET_TSI_MULT(ObReqTimeInfo, ObReqTimeInfo *req_timeinfo = GET_TSI_MULT(ObReqTimeInfo,
ObReqTimeInfo::REQ_TIMEINFO_IDENTIFIER); ObReqTimeInfo::REQ_TIMEINFO_IDENTIFIER);
OB_ASSERT(NULL != req_timeinfo); OB_ASSERT(NULL != req_timeinfo);
if (need_end_) { req_timeinfo->update_end_time();
req_timeinfo->update_end_time();
}
} }
bool need_end_;
}; };
} // end namespace observer } // end namespace observer
} // end namespace oceanbase } // end namespace oceanbase

11
src/sql/ob_result_set.cpp
View File

@ -75,15 +75,8 @@ ObResultSet::~ObResultSet()
if (OB_NOT_NULL(pc)) { if (OB_NOT_NULL(pc)) {
cache_obj_guard_.force_early_release(pc); cache_obj_guard_.force_early_release(pc);
} }
// Always called at the end of the ObResultSet destructor
if (need_update_cnt_ > 0) { update_end_time();
observer::ObReqTimeInfo *req_timeinfo = GET_TSI_MULT(observer::ObReqTimeInfo,
observer::ObReqTimeInfo::REQ_TIMEINFO_IDENTIFIER);
OB_ASSERT(NULL != req_timeinfo);
for (int i=0; i < need_update_cnt_; i++) {
req_timeinfo->update_end_time();
}
}
} }
int ObResultSet::open_cmd() int ObResultSet::open_cmd()

27
src/sql/ob_result_set.h
View File

@ -40,6 +40,7 @@
#include "sql/ob_sql_trans_control.h" #include "sql/ob_sql_trans_control.h"
#include "sql/plan_cache/ob_cache_object_factory.h" #include "sql/plan_cache/ob_cache_object_factory.h"
#include "observer/ob_inner_sql_rpc_proxy.h" #include "observer/ob_inner_sql_rpc_proxy.h"
#include "observer/ob_req_time_service.h"
namespace oceanbase namespace oceanbase
{ {
@ -278,8 +279,6 @@ public:
bool is_returning() const { return is_returning_; } bool is_returning() const { return is_returning_; }
void set_user_sql(bool is_user_sql) { is_user_sql_ = is_user_sql; } void set_user_sql(bool is_user_sql) { is_user_sql_ = is_user_sql; }
bool is_user_sql() const { return is_user_sql_; } bool is_user_sql() const { return is_user_sql_; }
// used to mark wether need to update endtime for timereqguard
void inc_need_update_endtime() { need_update_cnt_++; }
// 往带？的field name填充参数信息 // 往带？的field name填充参数信息
// 要注意的是，除了cname_以外，其余的string都是从计划里面浅拷出来的， // 要注意的是，除了cname_以外，其余的string都是从计划里面浅拷出来的，
@ -345,6 +344,22 @@ private:
int make_final_field_name(char *src, int64_t len, common::ObString &field_name); int make_final_field_name(char *src, int64_t len, common::ObString &field_name);
// 删除ParseNode中raw_text的多余的空格 // 删除ParseNode中raw_text的多余的空格
static int64_t remove_extra_space(char *buff, int64_t len); static int64_t remove_extra_space(char *buff, int64_t len);
// Always called in the ObResultSet constructor
void update_start_time() const
{
oceanbase::observer::ObReqTimeInfo *req_timeinfo = GET_TSI_MULT(observer::ObReqTimeInfo,
observer::ObReqTimeInfo::REQ_TIMEINFO_IDENTIFIER);
OB_ASSERT(NULL != req_timeinfo);
req_timeinfo->update_start_time();
}
// Always called at the end of the ObResultSet destructor
void update_end_time() const
{
oceanbase::observer::ObReqTimeInfo *req_timeinfo = GET_TSI_MULT(observer::ObReqTimeInfo,
observer::ObReqTimeInfo::REQ_TIMEINFO_IDENTIFIER);
OB_ASSERT(NULL != req_timeinfo);
req_timeinfo->update_end_time();
}
protected: protected:
// 区分本ResultSet是为User还是Inner SQL服务, 服务于EndTrans异步回调 // 区分本ResultSet是为User还是Inner SQL服务, 服务于EndTrans异步回调
@ -405,9 +420,6 @@ private:
bool is_returning_; bool is_returning_;
bool is_com_filed_list_; //used to mark COM_FIELD_LIST bool is_com_filed_list_; //used to mark COM_FIELD_LIST
common::ObString wild_str_;//uesd to save filed wildcard in COM_FIELD_LIST; common::ObString wild_str_;//uesd to save filed wildcard in COM_FIELD_LIST;
int64_t need_update_cnt_; // used to mark wether need to update endtime for timereqguard
}; };
@ -478,10 +490,11 @@ inline ObResultSet::ObResultSet(ObSQLSessionInfo &session, common::ObIAllocator
executor_(), executor_(),
is_returning_(false), is_returning_(false),
is_com_filed_list_(false), is_com_filed_list_(false),
wild_str_(), wild_str_()
need_update_cnt_(0)
{ {
message_[0] = '\0'; message_[0] = '\0';
// Always called in the ObResultSet constructor
update_start_time();
} }
inline int64_t ObResultSet::get_affected_rows() const inline int64_t ObResultSet::get_affected_rows() const