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

fix dblink transaction bug

Browse Source
This commit is contained in:
cqliang1995
2024-03-15 17:15:43 +00:00
committed by ob-robot
parent 4de78f3ba5
commit 43c435fe20
50 changed files with 1287 additions and 453 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
src/sql/engine/dml/ob_link_dml_op.cpp
View File

@ -114,13 +114,11 @@ int ObLinkDmlOp::send_reverse_link_info(transaction::ObTransID &tx_id)
int ObLinkDmlOp::inner_execute_link_stmt(const char *link_stmt)
{
int ret = OB_SUCCESS;
transaction::ObTransID tx_id;
if (OB_ISNULL(dblink_proxy_) || OB_ISNULL(dblink_conn_) || OB_ISNULL(link_stmt)) {
ObSQLSessionInfo *session = ctx_.get_my_session();
if (OB_ISNULL(dblink_proxy_) || OB_ISNULL(dblink_conn_) || OB_ISNULL(link_stmt) || OB_ISNULL(session)) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("dblink_proxy or link_stmt is NULL", K(ret), KP(dblink_proxy_), KP(link_stmt), KP(dblink_conn_));
} else if (OB_FAIL(ObTMService::tm_rm_start(ctx_, link_type_, dblink_conn_, tx_id))) {
LOG_WARN("failed to tm_rm_start", K(ret), K(dblink_id_), K(dblink_conn_));
} else if (MY_SPEC.is_reverse_link_ && OB_FAIL(send_reverse_link_info(tx_id))) {
LOG_WARN("dblink_proxy or link_stmt is NULL", K(ret), KP(dblink_proxy_), KP(link_stmt), KP(dblink_conn_), KP(session));
} else if (MY_SPEC.is_reverse_link_ && OB_FAIL(send_reverse_link_info(session->get_dblink_context().get_tx_id()))) {
LOG_WARN("failed to send reverse link info", K(ret), K(link_stmt));
} else if (OB_FAIL(dblink_proxy_->dblink_write(dblink_conn_, affected_rows_, link_stmt))) {
LOG_WARN("write failed", K(ret), K(link_stmt));
@ -133,7 +131,8 @@ int ObLinkDmlOp::inner_execute_link_stmt(const char *link_stmt)
void ObLinkDmlOp::reset_dblink()
{
int tmp_ret = OB_SUCCESS;
// when it is oci connection, does need terminate stmt like read, because write will terminate stmt after execute_update immediately
// when it is oci connection, does not need terminate stmt like read, because write will terminate stmt after execute_update immediately
ObDblinkCtxInSession::revert_dblink_conn(dblink_conn_); // release rlock locked by get_dblink_conn
tenant_id_ = OB_INVALID_ID;
dblink_id_ = OB_INVALID_ID;
dblink_proxy_ = NULL;
@ -146,19 +145,22 @@ int ObLinkDmlOp::inner_open()
ObSQLSessionInfo *session = ctx_.get_my_session();
ObPhysicalPlanCtx *plan_ctx = ctx_.get_physical_plan_ctx();
stmt_buf_len_ += head_comment_length_;
if (OB_ISNULL(session) || OB_ISNULL(plan_ctx)) {
in_xa_transaction_ = true; // link dml aways in xa trasaction
dblink_id_ = MY_SPEC.dblink_id_;
dblink_proxy_ = GCTX.dblink_proxy_;
if (OB_ISNULL(session) || OB_ISNULL(plan_ctx) || OB_ISNULL(dblink_proxy_)) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("session or plan_ctx is NULL", K(ret), KP(session), KP(plan_ctx));
LOG_WARN("session or plan_ctx or dblink_proxy_ is NULL", K(ret), KP(session), KP(plan_ctx), KP(dblink_proxy_));
} else if (FALSE_IT(tenant_id_ = session->get_effective_tenant_id())) {
} else if (FALSE_IT(sessid_ = session->get_sessid())) {
} else if (OB_FAIL(set_next_sql_req_level())) {
LOG_WARN("failed to set next sql req level", K(ret));
} else if (OB_FAIL(init_dblink(MY_SPEC.dblink_id_, GCTX.dblink_proxy_, true))) {
LOG_WARN("failed to init dblink", K(ret), K(MY_SPEC.dblink_id_));
} else if (OB_FAIL(init_dblink())) {
LOG_WARN("failed to init dblink", K(ret), K(dblink_id_));
} else if (OB_ISNULL(stmt_buf_ = static_cast<char *>(allocator_.alloc(stmt_buf_len_)))) {
ret = OB_ALLOCATE_MEMORY_FAILED;
LOG_WARN("failed to init stmt buf", K(ret), K(stmt_buf_len_));
} else {
LOG_DEBUG("succ to open link dml", K(lbt()));
}
return ret;

141
src/sql/engine/dml/ob_link_op.cpp
View File

@ -108,79 +108,67 @@ ObLinkOp::ObLinkOp(ObExecContext &exec_ctx, const ObOpSpec &spec, ObOpInput *inp
stmt_buf_len_(STMT_BUF_BLOCK),
next_sql_req_level_(0),
link_type_(DBLINK_DRV_OB),
in_xa_trascaction_(false)
in_xa_transaction_(false),
tm_sessid_(0)
{}
int ObLinkOp::init_dblink(uint64_t dblink_id, ObDbLinkProxy *dblink_proxy, bool in_xa_trascaction)
int ObLinkOp::init_dblink()
{
int ret = OB_SUCCESS;
ObSchemaGetterGuard schema_guard;
dblink_param_ctx param_ctx;
ObSQLSessionInfo * my_session = NULL;
common::sqlclient::ObISQLConnection *dblink_conn = NULL;
my_session = ctx_.get_my_session();
ObPhysicalPlanCtx *plan_ctx = GET_PHY_PLAN_CTX(ctx_);
in_xa_trascaction_ = in_xa_trascaction;
dblink_id_ = dblink_id;
if (OB_NOT_NULL(dblink_proxy_)) {
ret = OB_INIT_TWICE;
LOG_WARN("link scan ctx already inited", K(ret));
} else if (OB_ISNULL(dblink_proxy)) {
if (OB_FAIL(ret)) {
// do nothing
} else if (OB_ISNULL(dblink_proxy_)) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("dblink_proxy is NULL", K(ret));
LOG_WARN("dblink_proxy_ is NULL", K(ret));
} else if (OB_ISNULL(my_session) || OB_ISNULL(plan_ctx)) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("my_session or plan_ctx is NULL", K(my_session), K(plan_ctx), K(ret));
} else if (FALSE_IT(sessid_ = my_session->get_sessid())) {
LOG_WARN("my_session or plan_ctx is NULL", KP(my_session), KP(plan_ctx), K(ret));
} else if (OB_FAIL(GCTX.schema_service_->get_tenant_schema_guard(tenant_id_, schema_guard))) {
LOG_WARN("failed to get schema guard", K(ret), K(tenant_id_));
} else if (OB_FAIL(schema_guard.get_dblink_schema(tenant_id_, dblink_id, dblink_schema_))) {
LOG_WARN("failed to get dblink schema", K(ret), K(tenant_id_), K(dblink_id));
} else if (OB_FAIL(schema_guard.get_dblink_schema(tenant_id_, dblink_id_, dblink_schema_))) {
LOG_WARN("failed to get dblink schema", K(ret), K(tenant_id_), K(dblink_id_));
} else if (OB_ISNULL(dblink_schema_)) {
ret = OB_DBLINK_NOT_EXIST_TO_ACCESS;
LOG_WARN("dblink schema is NULL", K(ret), K(dblink_id));
} else if (FALSE_IT(set_link_driver_proto(static_cast<DblinkDriverProto>(dblink_schema_->get_driver_proto())))) {
// do nothing
} else if (OB_FAIL(ObLinkOp::init_dblink_param_ctx(ctx_,
param_ctx,
link_type_,
tenant_id_,
dblink_id_,
sessid_,
next_sql_req_level_))) {
LOG_WARN("failed to init dblink param ctx", K(ret));
} else if (OB_FAIL(dblink_proxy->create_dblink_pool(param_ctx,
dblink_schema_->get_host_addr(),
dblink_schema_->get_tenant_name(),
dblink_schema_->get_user_name(),
dblink_schema_->get_plain_password(),
dblink_schema_->get_database_name(),
dblink_schema_->get_conn_string(),
dblink_schema_->get_cluster_name()))) {
LOG_WARN("failed to create dblink pool", K(ret));
} else if (OB_FAIL(my_session->get_dblink_context().get_dblink_conn(dblink_id, dblink_conn))) {
LOG_WARN("dblink schema is NULL", K(ret), K(dblink_id_));
} else if (FALSE_IT(link_type_ = static_cast<DblinkDriverProto>(dblink_schema_->get_driver_proto()))) {
} else if (OB_FAIL(my_session->get_dblink_context().get_dblink_conn(dblink_id_, dblink_conn, tm_sessid_))) {
LOG_WARN("failed to get dblink connection from session", K(my_session), K(sessid_), K(ret));
} else {
if (NULL == dblink_conn) {
if (OB_FAIL(ObDblinkService::get_local_session_vars(my_session, allocator_, param_ctx))) {
LOG_WARN("failed to get local session vars", K(ret));
} else if (OB_FAIL(dblink_proxy->acquire_dblink(param_ctx, dblink_conn_))) {
LOG_WARN("failed to acquire dblink", K(ret), K(param_ctx));
if (NULL == dblink_conn) { // nothing about transaction
if (OB_FAIL(ObDblinkService::init_dblink_param_ctx(dblink_param_ctx_,
my_session,
allocator_, // useless in oracle mode
dblink_id_,
link_type_))) {
LOG_WARN("failed to init dblink param ctx", K(ret), K(dblink_param_ctx_), K(dblink_id_), K(link_type_));
} else if (OB_FAIL(dblink_proxy_->create_dblink_pool(dblink_param_ctx_,
dblink_schema_->get_host_addr(),
dblink_schema_->get_tenant_name(),
dblink_schema_->get_user_name(),
dblink_schema_->get_plain_password(),
dblink_schema_->get_database_name(),
dblink_schema_->get_conn_string(),
dblink_schema_->get_cluster_name()))) {
LOG_WARN("failed to create dblink pool", K(ret));
} else if (OB_FAIL(dblink_proxy_->acquire_dblink(dblink_param_ctx_, dblink_conn_))) {
LOG_WARN("failed to acquire dblink", K(ret), K(dblink_param_ctx_));
} else if (OB_FAIL(my_session->get_dblink_context().register_dblink_conn_pool(dblink_conn_->get_common_server_pool()))) {
LOG_WARN("failed to register dblink conn pool to current session", K(ret));
} else if (in_xa_trascaction_ && lib::is_oracle_mode() &&
OB_FAIL(my_session->get_dblink_context().set_dblink_conn(dblink_conn_))) {
LOG_WARN("failed to set dblink connection to session", K(in_xa_trascaction_), K(my_session), K(sessid_), K(ret));
} else {
LOG_TRACE("link op get connection from dblink pool", K(in_xa_trascaction_), KP(dblink_conn_), K(lbt()));
LOG_TRACE("link op get connection from dblink pool", K(in_xa_transaction_), KP(dblink_conn_), K(lbt()));
}
} else {
} else if (dblink_conn->get_dblink_driver_proto() != link_type_) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("wrong driver proto", K(ret), K(dblink_conn->get_dblink_driver_proto()), K(link_type_), K(next_sql_req_level_));
} else { // about transaction
dblink_conn_ = dblink_conn;
in_xa_trascaction_ = true; //to tell link scan op don't release dblink_conn_
LOG_TRACE("link op get connection from xa transaction", K(dblink_id), KP(dblink_conn_));
}
if (OB_SUCC(ret)) {
dblink_proxy_ = dblink_proxy;
in_xa_transaction_ = true; //to tell link scan op don't release dblink_conn_
LOG_TRACE("link op get connection from xa transaction", K(dblink_id_), KP(dblink_conn_));
}
}
return ret;
@ -385,59 +373,6 @@ int ObLinkSpec::set_param_infos(const ObIArray<ObParamPosIdx> &param_infos)
return ret;
}
int ObLinkOp::init_dblink_param_ctx(ObExecContext &exec_ctx,
common::sqlclient::dblink_param_ctx &param_ctx,
common::sqlclient::DblinkDriverProto link_type,
uint64_t tenant_id,
uint64_t dblink_id,
uint32_t session_id,
int64_t next_sql_req_level)
{
int ret = OB_SUCCESS;
uint16_t charset_id = 0;
uint16_t ncharset_id = 0;
if (OB_FAIL(get_charset_id(exec_ctx, charset_id, ncharset_id))) {
LOG_WARN("failed to get session charset id", K(ret));
} else {
param_ctx.charset_id_ = charset_id;
param_ctx.ncharset_id_ = ncharset_id;
param_ctx.pool_type_ = DblinkPoolType::DBLINK_POOL_DEF;
param_ctx.tenant_id_ = tenant_id;
param_ctx.dblink_id_ = dblink_id;
param_ctx.link_type_ = link_type;
param_ctx.sessid_ = session_id;
param_ctx.sql_request_level_ = next_sql_req_level;
}
return ret;
}
int ObLinkOp::get_charset_id(ObExecContext &exec_ctx,
uint16_t &charset_id,
uint16_t &ncharset_id)
{
int ret = OB_SUCCESS;
ObSQLSessionInfo *sess_info = NULL;
if (OB_ISNULL(sess_info = exec_ctx.get_my_session())) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("null session info", K(ret));
} else {
ObCollationType coll_type = sess_info->get_nls_collation();
ObCollationType ncoll_type = sess_info->get_nls_collation_nation();
ObCharsetType cs_type = ObCharset::charset_type_by_coll(coll_type);
ObCharsetType ncs_type = ObCharset::charset_type_by_coll(ncoll_type);
if (CHARSET_INVALID == cs_type || CHARSET_INVALID == ncs_type) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("failed to get charset id", K(ret), K(coll_type));
} else {
charset_id = static_cast<uint16_t>(ObCharset::charset_type_to_ora_charset_id(cs_type));
ncharset_id = static_cast<uint16_t>(ObCharset::charset_type_to_ora_charset_id(ncs_type));
LOG_DEBUG("get charset id", K(ret), K(charset_id), K(ncharset_id),
K(cs_type), K(ncs_type), K(coll_type), K(ncoll_type));
}
}
return ret;
}
void ObLinkOp::reset_link_sql()
{
if (OB_NOT_NULL(stmt_buf_)) {

16
src/sql/engine/dml/ob_link_op.h
View File

@ -51,23 +51,13 @@ public:
explicit ObLinkOp(ObExecContext &exec_ctx, const ObOpSpec &spec, ObOpInput *input);
virtual ~ObLinkOp() { destroy(); }
virtual void destroy() { reset(); }
inline void set_link_driver_proto(common::sqlclient::DblinkDriverProto type) { link_type_ = type; }
virtual void reset() = 0;
int init_dblink(uint64_t dblink_id, common::ObDbLinkProxy *dblink_proxy, bool in_xa_trascaction = false);
int init_dblink();
int execute_link_stmt(const common::ObString &link_stmt_fmt,
const common::ObIArray<ObParamPosIdx> &param_infos,
const ObParamStore &param_store,
ObReverseLink *reverse_link = NULL);
virtual int inner_execute_link_stmt(const char *link_stmt) = 0;
static int init_dblink_param_ctx(ObExecContext &exec_ctx,
common::sqlclient::dblink_param_ctx &param_ctx,
common::sqlclient::DblinkDriverProto link_type,
uint64_t tenant_id,
uint64_t dblink_id,
uint32_t session_id,
int64_t next_sql_req_level);
static int get_charset_id(ObExecContext &exec_ctx, uint16_t &charset_id, uint16_t &ncharset_id);
protected:
int combine_link_stmt(const common::ObString &link_stmt_fmt,
const common::ObIArray<ObParamPosIdx> &param_infos,
@ -90,7 +80,9 @@ protected:
int64_t next_sql_req_level_;
static const int64_t STMT_BUF_BLOCK;
common::sqlclient::DblinkDriverProto link_type_;
bool in_xa_trascaction_; // is dblink write/read remote database in xa transaction
bool in_xa_transaction_; // is dblink write/read remote database in xa transaction
common::sqlclient::dblink_param_ctx dblink_param_ctx_;
uint32_t tm_sessid_; // only used by link scan
static const char * head_comment_fmt_;
static const int64_t head_comment_length_;
static const char *proxy_route_info_fmt_;

6
src/sql/engine/ob_physical_plan.cpp
View File

@ -97,8 +97,6 @@ ObPhysicalPlan::ObPhysicalPlan(MemoryContext &mem_context /* = CURRENT_CONTEXT *
stat_(),
op_stats_(),
need_drive_dml_query_(false),
tx_id_(-1),
tm_sessid_(-1),
var_init_exprs_(allocator_),
is_returning_(false),
is_late_materialized_(false),
@ -228,9 +226,6 @@ void ObPhysicalPlan::reset()
append_table_id_ = 0;
stat_.expected_worker_map_.destroy();
stat_.minimal_worker_map_.destroy();
tx_id_ = -1;
tm_sessid_ = -1;
var_init_exprs_.reset();
need_record_plan_info_ = false;
logical_plan_.reset();
is_enable_px_fast_reclaim_ = false;
@ -239,7 +234,6 @@ void ObPhysicalPlan::reset()
udf_has_dml_stmt_ = false;
mview_ids_.reset();
}
void ObPhysicalPlan::destroy()
{
#ifndef NDEBUG

4
src/sql/engine/ob_physical_plan.h
View File

@ -453,7 +453,7 @@ public:
inline const ObExprFrameInfo &get_expr_frame_info() const { return expr_frame_info_; }
const ObOpSpec *get_root_op_spec() const { return root_op_spec_; }
inline bool is_link_dml_plan() {
inline bool is_link_dml_plan() const {
bool is_link_dml = false;
if (NULL != get_root_op_spec()) {
is_link_dml = oceanbase::sql::ObPhyOperatorType::PHY_LINK_DML == get_root_op_spec()->type_;
@ -619,8 +619,6 @@ public:
//@todo: yuchen.wyc add a temporary member to mark whether
//the DML statement needs to be executed through get_next_row
bool need_drive_dml_query_;
int64_t tx_id_; //for dblink recover xa tx
int64_t tm_sessid_; //for dblink get connection attached on tm session
ExprFixedArray var_init_exprs_;
private:
bool is_returning_; //是否设置了returning

6
src/sql/engine/ob_physical_plan_ctx.cpp
View File

@ -116,7 +116,11 @@ ObPhysicalPlanCtx::ObPhysicalPlanCtx(common::ObIAllocator &allocator)
enable_rich_format_(false),
all_local_session_vars_(allocator),
mview_ids_(allocator),
last_refresh_scns_(allocator)
last_refresh_scns_(allocator),
tx_id_(0),
tm_sessid_(0),
hint_xa_trans_stop_check_lock_(false),
main_xa_trans_branch_(false)
{
}

15
src/sql/engine/ob_physical_plan_ctx.h
View File

@ -465,6 +465,16 @@ public:
ObIArray<ObArrayParamGroup> &get_array_param_groups() { return array_param_groups_; }
int set_all_local_session_vars(ObIArray<ObLocalSessionVar> &all_local_session_vars);
int get_local_session_vars(int64_t idx, const ObLocalSessionVar *&local_vars);
void set_tx_id(int64_t tx_id) { tx_id_ = tx_id; }
int64_t get_tx_id() const { return tx_id_; }
void set_tm_sessid(int64_t tm_sessid) { tm_sessid_ = tm_sessid; }
int64_t get_tm_sessid() const { return tm_sessid_; }
void set_hint_xa_trans_stop_check_lock(int64_t hint_xa_trans_stop_check_lock) { hint_xa_trans_stop_check_lock_ = hint_xa_trans_stop_check_lock; }
int64_t get_hint_xa_trans_stop_check_lock() const { return hint_xa_trans_stop_check_lock_; }
void set_main_xa_trans_branch(int64_t main_xa_trans_branch) { main_xa_trans_branch_ = main_xa_trans_branch; }
int64_t get_main_xa_trans_branch() const { return main_xa_trans_branch_; }
ObIArray<uint64_t> &get_dblink_ids() { return dblink_ids_; }
inline int keep_dblink_id(uint64_t dblink_id) { return add_var_to_array_no_dup(dblink_ids_, dblink_id); }
private:
int init_param_store_after_deserialize();
void reset_datum_frame(char *frame, int64_t expr_cnt);
@ -610,6 +620,11 @@ private:
// for last_refresh_scn expr to get last_refresh_scn for rt mview used in query
common::ObFixedArray<uint64_t, common::ObIAllocator> mview_ids_;
common::ObFixedArray<uint64_t, common::ObIAllocator> last_refresh_scns_;
int64_t tx_id_; //for dblink recover xa tx
uint32_t tm_sessid_; //for dblink get connection attached on tm session
bool hint_xa_trans_stop_check_lock_; // for dblink to stop check stmt lock in xa trans
bool main_xa_trans_branch_; // for dblink to indicate weather this sql is executed in main_xa_trans_branch
ObSEArray<uint64_t, 8> dblink_ids_;
};
}

16
src/sql/engine/sequence/ob_sequence_op.cpp
View File

@ -230,13 +230,11 @@ int ObRemoteSequenceExecutor::init_dblink_connection(ObExecContext &ctx)
LOG_WARN("dblink schema is NULL", K(ret), K(dblink_id_));
} else if (FALSE_IT(link_type_ = static_cast<DblinkDriverProto>(dblink_schema->get_driver_proto()))) {
// do nothing
} else if (OB_FAIL(ObLinkOp::init_dblink_param_ctx(ctx,
param_ctx,
link_type_,
tenant_id,
} else if (OB_FAIL(ObDblinkService::init_dblink_param_ctx(param_ctx,
my_session,
ctx.get_allocator(), // uselees in oracle mode
dblink_id_,
sessid_,
my_session->get_next_sql_request_level()))) {
link_type_))) {
LOG_WARN("failed to init dblink param ctx", K(ret));
} else if (OB_FAIL(dblink_proxy->create_dblink_pool(param_ctx,
dblink_schema->get_host_addr(),
@ -258,6 +256,8 @@ int ObRemoteSequenceExecutor::init_dblink_connection(ObExecContext &ctx)
LOG_WARN("failed to register dblink conn pool to current session", K(ret));
} else if (OB_FAIL(my_session->get_dblink_context().set_dblink_conn(dblink_conn_))) {
LOG_WARN("failed to set dblink connection to session", K(my_session), K(sessid_), K(ret));
} else if (OB_FAIL(my_session->get_dblink_context().get_dblink_conn(param_ctx.dblink_id_, dblink_conn_))) { // will add a rlock on dblink conn, means this dblink_conn_ is inuse
LOG_WARN("failed to get dblink connection from session", K(ret), K(param_ctx.dblink_id_));
} else {
LOG_TRACE("link op get connection from dblink pool", KP(dblink_conn_), K(lbt()));
}
@ -316,6 +316,10 @@ void ObRemoteSequenceExecutor::destroy()
LOG_WARN("failed to close oci result", K(ret));
}
#endif
// release rlock on dblink_conn
if (OB_SUCCESS != (ret = ObDblinkCtxInSession::revert_dblink_conn(dblink_conn_))) {
LOG_WARN("failed to revert dblink conn", K(ret), KP(dblink_conn_));
}
//release dblink connection by session
sessid_ = 0;
dblink_conn_ = NULL;

99
src/sql/engine/table/ob_link_scan_op.cpp
View File

@ -50,7 +50,6 @@ ObLinkScanOp::ObLinkScanOp(ObExecContext &exec_ctx, const ObOpSpec &spec, ObOpIn
iter_end_(false),
row_allocator_(),
iterated_rows_(-1),
tm_session_(NULL),
tm_rm_connection_(NULL),
reverse_link_(NULL),
conn_type_(sql::DblinkGetConnType::DBLINK_POOL),
@ -62,6 +61,7 @@ void ObLinkScanOp::reset()
{
tz_info_ = NULL;
dblink_schema_ = NULL;
tm_sessid_ = 0;
reset_result();
reset_link_sql();
reset_dblink();
@ -152,7 +152,6 @@ int ObLinkScanOp::inner_execute_link_stmt(const char *link_stmt)
uint16_t ncharset_id = 0;
ObSQLSessionInfo * my_session = NULL;
my_session = ctx_.get_my_session();
transaction::ObTransID tx_id;
bool have_lob = false;
res_.set_enable_use_result(true);
bool new_snapshot = false;
@ -174,8 +173,6 @@ int ObLinkScanOp::inner_execute_link_stmt(const char *link_stmt)
} else if (OB_ISNULL(dblink_proxy_) || OB_ISNULL(my_session)) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("unexpected NULL", K(ret), KP(dblink_proxy_), KP(my_session));
} else if (need_tx(my_session) && OB_FAIL(ObTMService::tm_rm_start(ctx_, link_type_, dblink_conn_, tx_id))) {
LOG_WARN("failed to tm_rm_start", K(ret), K(dblink_id_), K(dblink_conn_), K(sessid_));
#ifdef OB_BUILD_DBLINK
} else if (OB_FAIL(init_conn_snapshot(new_snapshot))) {
LOG_WARN("init conn snapshot failed", K(ret));
@ -207,7 +204,7 @@ int ObLinkScanOp::inner_execute_link_stmt(const char *link_stmt)
ret = OB_NOT_SUPPORTED;
LOG_WARN("dblink not support lob type", K(ret));
LOG_USER_ERROR(OB_NOT_SUPPORTED, "dblink fetch lob type data");
} else if (OB_FAIL(ObLinkOp::get_charset_id(ctx_, charset_id, ncharset_id))) {
} else if (OB_FAIL(ObDblinkService::get_charset_id(my_session, charset_id, ncharset_id))) {
LOG_WARN("failed to get charset id", K(ret));
} else if (OB_FAIL(result_->set_expected_charset_id(charset_id, ncharset_id))) {// for oci dblink set expected result charset, actually useless...
LOG_WARN("failed to set result set expected charset", K(ret), K(charset_id), K(ncharset_id));
@ -245,13 +242,20 @@ void ObLinkScanOp::reset_dblink()
LOG_WARN_RET(tmp_ret, "free dblink snapshot failed");
}
#endif
if (OB_NOT_NULL(dblink_proxy_) && OB_NOT_NULL(dblink_conn_) && !in_xa_trascaction_ &&
if (OB_NOT_NULL(dblink_proxy_) && OB_NOT_NULL(dblink_conn_) && !in_xa_transaction_ &&
OB_SUCCESS != (tmp_ret = dblink_proxy_->release_dblink(link_type_, dblink_conn_))) {
LOG_WARN_RET(tmp_ret, "failed to release connection", K(tmp_ret));
}
if (OB_NOT_NULL(reverse_link_)) {
reverse_link_->close();
}
if (in_xa_transaction_ && OB_NOT_NULL(dblink_conn_)) {
ObDblinkCtxInSession::revert_dblink_conn(dblink_conn_); // release rlock locked by get_dblink_conn
}
if (OB_NOT_NULL(tm_rm_connection_)) {
ObDblinkCtxInSession::revert_dblink_conn(tm_rm_connection_); // release rlock locked by get_dblink_conn
}
tenant_id_ = OB_INVALID_ID;
dblink_id_ = OB_INVALID_ID;
dblink_proxy_ = NULL;
@ -282,21 +286,22 @@ int ObLinkScanOp::inner_open()
ObSQLSessionInfo *session = ctx_.get_my_session();
ObPhysicalPlanCtx *plan_ctx = ctx_.get_physical_plan_ctx();
const ObPhysicalPlan *phy_plan = MY_SPEC.get_phy_plan();
int64_t tm_sessid = -1;
reverse_link_ = NULL;
stmt_buf_len_ += head_comment_length_;
if (NULL != phy_plan) {
tm_sessid = phy_plan->tm_sessid_;
}
if (OB_ISNULL(session) || OB_ISNULL(plan_ctx)) {
dblink_id_ = MY_SPEC.dblink_id_;
dblink_proxy_ = GCTX.dblink_proxy_;
if (OB_ISNULL(session) || OB_ISNULL(plan_ctx) || OB_ISNULL(phy_plan) || OB_ISNULL(dblink_proxy_)) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("session or plan_ctx is NULL", K(ret), KP(session), KP(plan_ctx));
LOG_WARN("session or plan_ctx or dblink_proxy_ or dblink_proxy_ is NULL", K(ret), KP(session), KP(plan_ctx), KP(phy_plan), KP(dblink_proxy_));
} else if (FALSE_IT(tenant_id_ = session->get_effective_tenant_id())) {
} else if (FALSE_IT(sessid_ = session->get_sessid())) {
} else if (FALSE_IT(tm_sessid_ = plan_ctx->get_tm_sessid())) {
} else if (OB_FAIL(set_next_sql_req_level())) {
LOG_WARN("failed to set next sql req level", K(ret));
} else if (FALSE_IT(tenant_id_ = session->get_effective_tenant_id())) {
} else if (MY_SPEC.is_reverse_link_) {
// RM process sql within @! and @xxxx! send by TM
LOG_DEBUG("link scan op, RM process sql within @! and @xxxx! send by TM");
LOG_TRACE("link scan op, RM process sql within @! and @xxxx! send by TM");
conn_type_ = sql::DblinkGetConnType::TEMP_CONN;
ObPhysicalPlanCtx *plan_ctx = GET_PHY_PLAN_CTX(ctx_);
if (OB_ISNULL(plan_ctx)) {
@ -310,46 +315,38 @@ int ObLinkScanOp::inner_open()
} else if (OB_FAIL(reverse_link_->open(next_sql_req_level_))) {
LOG_WARN("failed to open reverse_link", K(ret));
}
} else if (-1 != tm_sessid) { // TM process sql within @xxxx send by RM
LOG_DEBUG("link scan op, TM process sql within @xxxx send by RM", K(tm_sessid));
sql::ObSQLSessionMgr *session_mgr = GCTX.session_mgr_;
if (OB_ISNULL(session_mgr)) {
} else if (0 != tm_sessid_) { // TM process sql within @xxxx send by RM
LOG_TRACE("link scan op, TM process sql within @xxxx send by RM", K(tm_sessid_));
if (OB_FAIL(session->get_dblink_context().get_dblink_conn(MY_SPEC.dblink_id_, tm_rm_connection_, tm_sessid_))) {
LOG_TRACE("link scan op failed to get tm_tm_connection", K(tm_sessid_), K(MY_SPEC.dblink_id_), K(ret));
} else if (OB_ISNULL(tm_rm_connection_)) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("unexpected null", K(ret), KP(session_mgr));
} else if (OB_FAIL(session_mgr->get_session(static_cast<uint32_t>(tm_sessid), tm_session_))) {
LOG_WARN("failed to get session", K(ret), K(tm_sessid));
LOG_WARN("unexcepted null ptr", K(tm_sessid_), K(MY_SPEC.dblink_id_), K(ret));
} else {
if (NULL != tm_session_ &&
tm_session_->get_dblink_context().get_dblink_conn(MY_SPEC.dblink_id_, tm_rm_connection_)) {
LOG_WARN("failed to get dblink connection from session", KP(tm_session_), K(ret));
} else if (NULL != tm_rm_connection_){
conn_type_ = sql::DblinkGetConnType::TM_CONN;
LOG_DEBUG("get tm sesseion and connection", KP(tm_session_), KP(tm_rm_connection_));
}
session_mgr->revert_session(tm_session_);
tm_session_ = NULL;
conn_type_ = sql::DblinkGetConnType::TM_CONN;
link_type_ = tm_rm_connection_->get_dblink_driver_proto();
LOG_TRACE("get tm sesseion and connection", KP(tm_rm_connection_), K(link_type_), K(tm_sessid_), K(sessid_), K(ret));
}
} else if (OB_FAIL(init_dblink())) {
LOG_WARN("failed to init dblink", K(ret), K(dblink_id_), K(MY_SPEC.is_reverse_link_));
}
if (OB_FAIL(ret)) {
// do nothing
} else if (sql::DblinkGetConnType::DBLINK_POOL == conn_type_ &&
OB_FAIL(init_dblink(MY_SPEC.dblink_id_, GCTX.dblink_proxy_, MY_SPEC.has_for_update_))) {
LOG_WARN("failed to init dblink", K(ret), K(MY_SPEC.dblink_id_), K(MY_SPEC.is_reverse_link_));
} else if (OB_FAIL(init_tz_info(TZ_INFO(session)))) {
if (OB_SUCC(ret)) {
if (OB_FAIL(init_tz_info(TZ_INFO(session)))) {
LOG_WARN("failed to tz info", K(ret), KP(session));
} else {
row_allocator_.set_tenant_id(tenant_id_);
row_allocator_.set_label("linkoprow");
row_allocator_.set_ctx_id(ObCtxIds::WORK_AREA);
}
if (OB_SUCC(ret) && OB_ISNULL(stmt_buf_ = static_cast<char *>(allocator_.alloc(stmt_buf_len_)))) {
ret = OB_ALLOCATE_MEMORY_FAILED;
LOG_WARN("failed to init stmt buf", K(ret), K(stmt_buf_len_));
} else if (OB_ISNULL(stmt_buf_ = static_cast<char *>(allocator_.alloc(stmt_buf_len_)))) {
ret = OB_ALLOCATE_MEMORY_FAILED;
LOG_WARN("failed to init stmt buf", K(ret), K(stmt_buf_len_));
} else {
row_allocator_.set_tenant_id(tenant_id_);
row_allocator_.set_label("linkoprow");
row_allocator_.set_ctx_id(ObCtxIds::WORK_AREA);
}
LOG_TRACE("succ to open link scan op", K(dblink_id_), K(tm_sessid_), K(MY_SPEC.is_reverse_link_), K(conn_type_), K(ret));
}
// close reverse_link
if (NULL != reverse_link_ && OB_FAIL(ret)) {
reverse_link_->close();
LOG_DEBUG("close reverse link", KP(reverse_link_), K(ret));
LOG_TRACE("close reverse link", KP(reverse_link_), K(ret));
}
return ret;
}
@ -544,19 +541,5 @@ int ObLinkScanOp::inner_get_next_batch(const int64_t max_row_cnt)
return ret;
}
// PLEASE check the input parameter in public interface
bool ObLinkScanOp::need_tx(const ObSQLSessionInfo *my_session) const
{
bool ret_bool = false;
if (MY_SPEC.has_for_update_) {
// case 1, if select for update, tm_rm_start is required to be called.
ret_bool = true;
} else if (false && my_session->is_in_transaction()) {
// case 2, if select in transaction, tm_rm_start is required to be called.
ret_bool = true;
}
return ret_bool;
}
} // end namespace sql
} // end namespace oceanbase

1
src/sql/engine/table/ob_link_scan_op.h
View File

@ -64,7 +64,6 @@ private:
bool iter_end_;
common::ObArenaAllocator row_allocator_;
int64_t iterated_rows_;
ObSQLSessionInfo *tm_session_;
common::sqlclient::ObISQLConnection *tm_rm_connection_;
ObReverseLink *reverse_link_;
sql::DblinkGetConnType conn_type_;