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

[to #48895314]prexecute support packet retry

Browse Source
This commit is contained in:
LiuYoung00
2023-07-11 04:12:37 +00:00
committed by ob-robot
parent 622e2b81d1
commit 1d86161bf1
10 changed files with 352 additions and 386 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
src/observer/mysql/ob_async_cmd_driver.cpp
View File

@ -18,6 +18,7 @@
#include "obsm_row.h"
#include "sql/resolver/cmd/ob_variable_set_stmt.h"
#include "observer/mysql/obmp_query.h"
#include "observer/mysql/obmp_stmt_prexecute.h"
namespace oceanbase
{
@ -55,7 +56,8 @@ int ObAsyncCmdDriver::response_result(ObMySQLResultSet &result)
if (OB_ISNULL(cur_trace_id = ObCurTraceId::get_trace_id())) {
ret = OB_ERR_UNEXPECTED;
LOG_ERROR("current trace id is NULL", K(ret));
} else if (is_prexecute_ && OB_FAIL(sender_.flush_buffer(false))) {
} else if (is_prexecute_
&& OB_FAIL(response_query_header(result, false, false, true))) {
LOG_WARN("flush buffer fail before send async ok packet.", K(ret));
} else if (OB_FAIL(sql_end_cb.set_packet_param(pkt_param.fill(result, session_, *cur_trace_id)))) {
LOG_ERROR("fail to set packet param", K(ret));

36
src/observer/mysql/ob_async_plan_driver.cpp
View File

@ -24,6 +24,7 @@
#include "obsm_row.h"
#include "observer/mysql/obmp_query.h"
#include "observer/mysql/ob_mysql_end_trans_cb.h"
#include "observer/mysql/obmp_stmt_prexecute.h"
namespace oceanbase
{
@ -63,18 +64,15 @@ int ObAsyncPlanDriver::response_result(ObMySQLResultSet &result)
} else if (OB_FAIL(result.update_last_insert_id_to_client())) {
LOG_WARN("failed to update last insert id after open", K(ret));
} else {
if (is_prexecute_ && OB_FAIL(sender_.flush_buffer(false))) {
LOG_WARN("flush buffer fail before send async ok packet.", K(ret));
} else {
// open 成功，允许异步回包
result.set_end_trans_async(true);
}
}
if (OB_SUCCESS != ret) {
// 如果try_again为true，说明这条SQL需要重做。考虑到重做之前我们需要回滚整个事务，会调用EndTransCb
// 所以这里设置一个标记，告诉EndTransCb这种情况下不要给客户端回包。
int cli_ret = OB_SUCCESS;
retry_ctrl_.test_and_save_retry_state(gctx_, ctx_, result, ret, cli_ret, is_prexecute_);
retry_ctrl_.test_and_save_retry_state(gctx_, ctx_, result, ret, cli_ret);
if (retry_ctrl_.need_retry()) {
result.set_end_trans_async(false);
}
@ -93,11 +91,39 @@ int ObAsyncPlanDriver::response_result(ObMySQLResultSet &result)
K(ret), K(cli_ret), K(cret), K(retry_ctrl_.need_retry()));
}
ret = cli_ret;
} else if (is_prexecute_ && OB_FAIL(response_query_header(result, false, false, true))) {
/*
* prexecute 仅在执行成功时候发送 header 包
* 执行失败时候有两种表现
* 1. 只返回一个 error 包， 这个时候需要注意 ps stmt 的泄漏问题
* 2. 重试， local 重试直接交给上层做， package 重试需要 注意 ps stmt 的泄漏问题
* 3. response_query_header & flush_buffer 不会产生需要重试的报错，此位置是 ObAsyncPlanDriver 重试前的一步，中间不会有别的可能重试的操作
* 4. ps stmt 清理要注意异步回包的情况，可能需要在异步包里面做清理
*/
LOG_WARN("prexecute response query head fail. ", K(ret));
} else if (result.is_with_rows()) {
ret = OB_ERR_UNEXPECTED;
LOG_ERROR("SELECT should not use async method. wrong code!!!", K(ret));
} else if (OB_FAIL(result.close())) {
LOG_WARN("result close failed, let's leave process(). EndTransCb will clean this mess", K(ret));
} else {
// async 并没有调用 ObSqlEndTransCb 回复 OK 包
// 所以 二合一协议的 OK 包也要放在这里处理
if (is_prexecute_) {
if (stmt::T_SELECT == result.get_stmt_type()) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("select stmt do not use async plan in prexecute.", K(ret));
} else {
ObOKPParam ok_param;
ok_param.affected_rows_ = result.get_affected_rows();
ok_param.is_partition_hit_ = session_.partition_hit().get_bool();
ok_param.has_more_result_ = result.has_more_result();
ok_param.send_last_row_ = true;
if (OB_FAIL(sender_.send_ok_packet(session_, ok_param))) {
LOG_WARN("fail to send ok packt", K(ok_param), K(ret));
}
}
}
}
// 仅在end_trans执行后（无论成功与否）才会设置，意味着一定会回调

105
src/observer/mysql/ob_query_driver.cpp
View File

@ -23,6 +23,7 @@
#include <string.h>
#include "share/ob_lob_access_utils.h"
#include "lib/charset/ob_charset.h"
#include "observer/mysql/obmp_stmt_prexecute.h"
namespace oceanbase
@ -34,37 +35,71 @@ namespace observer
{
int ObQueryDriver::response_query_header(ObResultSet &result,
bool has_nore_result,
bool has_more_result,
bool need_set_ps_out_flag,
bool is_prexecute)
bool need_flush_buffer)
{
int ret = OB_SUCCESS;
if (is_prexecute_) {
// 二合一协议发送 header 包, 不单独发送 column
if (OB_FAIL(static_cast<ObMPStmtPrexecute&>(sender_).response_query_header(session_, result, need_flush_buffer))) {
LOG_WARN("prexecute response query head fail. ", K(ret));
}
} else {
if (NULL == result.get_field_columns()) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("response field is null. ", K(ret));
} else if (OB_FAIL(response_query_header(*result.get_field_columns(),
has_more_result,
need_set_ps_out_flag,
false,
&result))) {
LOG_WARN("response query head fail. ", K(ret));
}
}
if (OB_FAIL(ret)) {
result.set_errcode(ret);
}
return ret;
}
int ObQueryDriver::response_query_header(const ColumnsFieldIArray &fields,
bool has_more_result,
bool need_set_ps_out_flag,
bool ps_cursor_execute,
ObResultSet *result)
{
int ret = OB_SUCCESS;
bool ac = true;
if (result.get_field_cnt() <= 0) {
LOG_WARN("result set column", K(result.get_field_cnt()));
// result == null means ps cursor in execute or fetch .
if (NULL != result && (&fields != result->get_field_columns())) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("filed is not from result in non ps cursor mode. ", K(ret));
} else if (fields.count() <= 0) {
LOG_WARN("column cnt is null ", K(fields.count()));
ret = OB_ERR_BAD_FIELD_ERROR;
} else if (OB_FAIL(session_.get_autocommit(ac))) {
LOG_WARN("fail to get autocommit", K(ret));
}
if (OB_SUCC(ret) && !result.get_is_com_filed_list() && !is_prexecute) {
} else if (!(NULL != result && result->get_is_com_filed_list())) {
// 普通协议发送 cnt 值
OMPKResheader rhp;
rhp.set_field_count(result.get_field_cnt());
if (OB_FAIL(sender_.response_packet(rhp, &result.get_session()))) {
rhp.set_field_count(fields.count());
if (OB_FAIL(sender_.response_packet(rhp, &session_))) {
LOG_WARN("response packet fail", K(ret));
}
} else {
// com field 协议在这里什么都不发，直接发送 field 信息
}
// 发送 field 信息
if (OB_SUCC(ret)) {
const ColumnsFieldIArray *fields = result.get_field_columns();
if (OB_ISNULL(fields)) {
ret = OB_INVALID_ARGUMENT;
result.set_errcode(ret);
}
for (int64_t i = 0; OB_SUCC(ret) && i < result.get_field_cnt(); ++i) {
for (int64_t i = 0; OB_SUCC(ret) && i < fields.count(); ++i) {
bool is_not_match = false;
ObMySQLField field;
const ObField &ob_field = result.get_field_columns()->at(i);
if (OB_FAIL(is_com_filed_list_match_wildcard_str(
result,
const ObField &ob_field = fields.at(i);
if (NULL != result && result->get_is_com_filed_list()
&& OB_FAIL(is_com_filed_list_match_wildcard_str(
*result,
static_cast<ObCollationType>(ob_field.charsetnr_),
ob_field.org_cname_,
is_not_match))) {
@ -73,14 +108,13 @@ int ObQueryDriver::response_query_header(ObResultSet &result,
/*do nothing*/
} else {
ret = ObMySQLResultSet::to_mysql_field(ob_field, field);
result.replace_lob_type(result.get_session(), ob_field, field);
if (result.get_is_com_filed_list()) {
if (OB_SUCC(ret)) {
ObMySQLResultSet::replace_lob_type(session_, ob_field, field);
if (NULL != result && result->get_is_com_filed_list()) {
field.default_value_ = static_cast<EMySQLFieldType>(ob_field.default_value_.get_ext());
}
result.set_errcode(ret);
if (OB_SUCC(ret)) {
OMPKField fp(field);
if (OB_FAIL(sender_.response_packet(fp, &result.get_session()))) {
if (OB_FAIL(sender_.response_packet(fp, &session_))) {
LOG_WARN("response packet fail", K(ret));
}
}
@ -95,18 +129,37 @@ int ObQueryDriver::response_query_header(ObResultSet &result,
flags.status_flags_.OB_SERVER_STATUS_IN_TRANS
= (session_.is_server_status_in_transaction() ? 1 : 0);
flags.status_flags_.OB_SERVER_STATUS_AUTOCOMMIT = (ac ? 1 : 0);
flags.status_flags_.OB_SERVER_MORE_RESULTS_EXISTS = has_nore_result;
flags.status_flags_.OB_SERVER_MORE_RESULTS_EXISTS = has_more_result;
flags.status_flags_.OB_SERVER_PS_OUT_PARAMS = need_set_ps_out_flag ? 1 : 0;
// NULL == result 说明是老协议 ps cursor execute 回包，或者fetch 协议回包， cursor_exit = true
flags.status_flags_.OB_SERVER_STATUS_CURSOR_EXISTS = NULL == result ? 1 : 0;
if (!session_.is_obproxy_mode()) {
// in java client or others, use slow query bit to indicate partition hit or not
flags.status_flags_.OB_SERVER_QUERY_WAS_SLOW = !session_.partition_hit().get_bool();
}
eofp.set_server_status(flags);
if (OB_FAIL(sender_.response_packet(eofp, &result.get_session()))) {
if (ps_cursor_execute && sender_.need_send_extra_ok_packet()) {
// 老协议 ps cursor execute 回包， 只回 field 信息， 所以对于 proxy ， 需要额外回一个 OK包
// 但是由于 2.0 协议需要在回 OK 包的同时了解 EOF 包的情况，所以这个 OK 包没办法抽到 execute 协议层处理
if (OB_FAIL(sender_.update_last_pkt_pos())) {
LOG_WARN("failed to update last packet pos", K(ret));
} else {
// in multi-stmt, send extra ok packet in the last stmt(has no more result)
ObOKPParam ok_param;
ok_param.affected_rows_ = 0;
ok_param.is_partition_hit_ = session_.partition_hit().get_bool();
ok_param.has_more_result_ = false;
if (OB_FAIL(sender_.send_ok_packet(session_, ok_param, &eofp))) {
LOG_WARN("fail to send ok packt", K(ok_param), K(ret));
}
}
} else {
if (OB_FAIL(sender_.response_packet(eofp, &session_))) {
LOG_WARN("response packet fail", K(ret));
}
}
}
return ret;
}
@ -152,7 +205,7 @@ int ObQueryDriver::response_query_result(ObResultSet &result,
if (is_first_row) {
is_first_row = false;
can_retry = false; // 已经获取到第一行数据，不再重试了
if (OB_FAIL(response_query_header(result, has_more_result, false, is_prexecute_))) {
if (OB_FAIL(response_query_header(result, has_more_result, false))) {
LOG_WARN("fail to response query header", K(ret), K(row_num), K(can_retry));
}
}
@ -235,7 +288,7 @@ int ObQueryDriver::response_query_result(ObResultSet &result,
if (OB_SUCC(ret) && 0 == row_num) {
// 如果是一行数据也没有，则还是要给客户端回复field等信息，并且不再重试了
can_retry = false;
if (OB_FAIL(response_query_header(result, has_more_result, false, is_prexecute_))) {
if (OB_FAIL(response_query_header(result, has_more_result, false))) {
LOG_WARN("fail to response query header", K(ret), K(row_num), K(can_retry));
}
}

12
src/observer/mysql/ob_query_driver.h
View File

@ -16,6 +16,7 @@
#include "share/ob_define.h"
#include "lib/charset/ob_charset.h"
#include "lib/string/ob_string.h"
#include "deps/oblib/src/common/ob_field.h"
namespace oceanbase
{
@ -60,14 +61,19 @@ public:
virtual int response_result(ObMySQLResultSet &result) = 0;
virtual int response_query_header(sql::ObResultSet &result,
bool has_nore_result = false,
bool need_set_ps_out = false,
bool is_prexecute = false);
bool has_more_result,
bool need_set_ps_out_flag,
bool need_flush_buffer = false);
virtual int response_query_result(sql::ObResultSet &result,
bool is_ps_protocol,
bool has_more_result,
bool &can_retry,
int64_t fetch_limit = common::OB_INVALID_COUNT);
int response_query_header(const ColumnsFieldIArray &fields,
bool has_more_result = false,
bool need_set_ps_out = false,
bool ps_cursor_execute = false,
sql::ObResultSet *result = NULL);
int convert_string_value_charset(common::ObObj& value, sql::ObResultSet &result);
int convert_string_value_charset(common::ObObj& value,
common::ObCharsetType charset_type,

15
src/observer/mysql/ob_sync_cmd_driver.cpp
View File

@ -21,6 +21,7 @@
#include "rpc/obmysql/packet/ompk_row.h"
#include "rpc/obmysql/packet/ompk_eof.h"
#include "share/ob_lob_access_utils.h"
#include "observer/mysql/obmp_stmt_prexecute.h"
namespace oceanbase
{
@ -127,7 +128,7 @@ int ObSyncCmdDriver::response_result(ObMySQLResultSet &result)
}
// open失败，决定是否需要重试
retry_ctrl_.test_and_save_retry_state(gctx_, ctx_, result, ret, cli_ret, is_prexecute_);
retry_ctrl_.test_and_save_retry_state(gctx_, ctx_, result, ret, cli_ret);
LOG_WARN("result set open failed, check if need retry",
K(ret), K(cli_ret), K(retry_ctrl_.need_retry()));
ret = cli_ret;
@ -148,7 +149,12 @@ int ObSyncCmdDriver::response_result(ObMySQLResultSet &result)
need_send_eof = true;
}
}
} else { /*do nothing*/ }
} else if (is_prexecute_) {
if (OB_FAIL(response_query_header(result, false, false , // in prexecute , has_more_result and has_ps out is no matter, it will be recalc
true))) {
LOG_WARN("prexecute response query head fail. ", K(ret));
}
}
if (OB_SUCC(ret)) {
// for CRUD SQL
@ -189,7 +195,7 @@ int ObSyncCmdDriver::response_result(ObMySQLResultSet &result)
LOG_WARN("response packet fail", K(ret));
}
}
}
} else { /*do nothing*/ }
if (!OB_SUCC(ret) && !process_ok && !retry_ctrl_.need_retry()) {
int sret = OB_SUCCESS;
@ -286,8 +292,7 @@ int ObSyncCmdDriver::response_query_result(ObMySQLResultSet &result)
const common::ObNewRow *row = NULL;
if (OB_FAIL(result.next_row(row)) ) {
LOG_WARN("fail to get next row", K(ret));
} else if ((!is_prexecute_ || stmt::T_ANONYMOUS_BLOCK == result.get_stmt_type()) &&
OB_FAIL(response_query_header(result, result.has_more_result(), true, is_prexecute_))) {
} else if (OB_FAIL(response_query_header(result, result.has_more_result(), true))) {
LOG_WARN("fail to response query header", K(ret));
} else if (OB_ISNULL(ctx_.session_info_)) {
ret = OB_ERR_UNEXPECTED;

27
src/observer/mysql/ob_sync_plan_driver.cpp
View File

@ -25,6 +25,7 @@
#include "sql/engine/px/ob_px_admission.h"
#include "sql/ob_spi.h"
#include "share/object/ob_obj_cast.h"
#include "observer/mysql/obmp_stmt_prexecute.h"
namespace oceanbase
{
@ -72,8 +73,7 @@ int ObSyncPlanDriver::response_result(ObMySQLResultSet &result)
ctx_,
result,
ret,
cli_ret,
is_prexecute_);
cli_ret);
if (OB_TRANSACTION_SET_VIOLATION != ret && OB_REPLICA_NOT_READABLE != ret) {
if (OB_TRY_LOCK_ROW_CONFLICT == ret && retry_ctrl_.need_retry()) {
//锁冲突重试不打印日志，避免刷屏
@ -108,8 +108,7 @@ int ObSyncPlanDriver::response_result(ObMySQLResultSet &result)
ctx_,
result,
ret,
cli_ret,
is_prexecute_);
cli_ret);
LOG_WARN("result response failed, check if need retry",
K(ret), K(cli_ret), K(retry_ctrl_.need_retry()));
ret = cli_ret;
@ -158,15 +157,19 @@ int ObSyncPlanDriver::response_result(ObMySQLResultSet &result)
if (OB_SUCC(ret) && !result.get_is_com_filed_list()) {
need_send_eof = true;
}
// for obproxy
if (OB_SUCC(ret)) {
if (OB_FAIL(ret)) {
// do nothing
} else if ((is_prexecute_ && stmt::T_SELECT != result.get_stmt_type())
|| (!is_prexecute_ && sender_.need_send_extra_ok_packet() && !result.has_more_result())) {
// 二合一协议 select 语句的 OK 包全部放在协议层发送
// sync plan 此时需要为 二合一协议单独发送 OK 包 ， for obproxy，
// in multi-stmt, send extra ok packet in the last stmt(has no more result)
if (!is_prexecute_ && sender_.need_send_extra_ok_packet() && !result.has_more_result()) {
ObOKPParam ok_param;
ok_param.affected_rows_ = 0;
ok_param.affected_rows_ = result.get_affected_rows();
ok_param.is_partition_hit_ = session_.partition_hit().get_bool();
ok_param.has_more_result_ = result.has_more_result();
ok_param.send_last_row_ = is_prexecute_ ? true : false;
if (need_send_eof) {
if (OB_FAIL(sender_.send_ok_packet(session_, ok_param, &eofp))) {
LOG_WARN("fail to send ok packt", K(ok_param), K(ret));
@ -177,14 +180,17 @@ int ObSyncPlanDriver::response_result(ObMySQLResultSet &result)
}
}
} else {
// 二合一协议 select 语句结果集后的 EOF 包都要在这里发送
// 非 二合一协议， 不需要额外 OK 包的 EOF 包需要在这里发送
if (need_send_eof && OB_FAIL(sender_.response_packet(eofp, &result.get_session()))) {
LOG_WARN("response packet fail", K(ret));
}
}
}
}
} else {
if (OB_FAIL(result.close())) {
if (is_prexecute_ && OB_FAIL(response_query_header(result, false, false, true))) {
LOG_WARN("prexecute response query head fail. ", K(ret));
} else if (OB_FAIL(result.close())) {
LOG_WARN("close result set fail", K(ret));
} else {
if (!result.has_implicit_cursor()) {
@ -233,6 +239,7 @@ int ObSyncPlanDriver::response_result(ObMySQLResultSet &result)
if (OB_TIMEOUT == ret) {
LOG_USER_ERROR(OB_TIMEOUT, THIS_WORKER.get_timeout_ts() - session_.get_query_start_time());
}
if (OB_FAIL(ret) &&
!process_ok &&
!retry_ctrl_.need_retry() &&

83
src/observer/mysql/obmp_stmt_execute.cpp
View File

@ -2952,81 +2952,20 @@ void ObMPStmtExecute::record_stat(const stmt::StmtType type, const int64_t end_t
int ObMPStmtExecute::response_query_header(ObSQLSessionInfo &session, pl::ObDbmsCursorInfo &cursor)
{
// TODO: 增加com类型的处理
int ret = OB_SUCCESS;
bool ac = true;
const ColumnsFieldArray &fields = cursor.get_field_columns();
int64_t fields_count = fields.count();
if (fields_count <= 0) {
LOG_WARN("cursor set column", K(fields_count));
ret = OB_ERR_BAD_FIELD_ERROR;
} else if (OB_FAIL(session.get_autocommit(ac))) {
ObSyncPlanDriver drv(gctx_,
ctx_,
session,
retry_ctrl_,
*this,
false,
OB_INVALID_COUNT);
if (OB_FAIL(drv.response_query_header(cursor.get_field_columns(),
false,
false,
true))) {
LOG_WARN("fail to get autocommit", K(ret));
}
if (OB_SUCC(ret)) {
OMPKResheader rhp;
rhp.set_field_count(fields_count);
if (OB_FAIL(response_packet(rhp, &session))) {
LOG_WARN("response packet fail", K(ret));
}
}
if (OB_SUCC(ret)) {
for (int64_t i = 0; OB_SUCC(ret) && i < fields_count; ++i) {
ObMySQLField field;
const ObField &ob_field = fields.at(i);
ret = ObMySQLResultSet::to_mysql_field(ob_field, field);
if (OB_SUCC(ret)) {
ObMySQLResultSet::replace_lob_type(session, ob_field, field);
OMPKField fp(field);
if (OB_FAIL(response_packet(fp, &session))) {
LOG_WARN("response packet fail", K(ret));
}
}
}
}
if (OB_SUCC(ret)) {
OMPKEOF eofp;
eofp.set_warning_count(0);
ObServerStatusFlags flags = eofp.get_server_status();
flags.status_flags_.OB_SERVER_STATUS_IN_TRANS
= (session.is_server_status_in_transaction() ? 1 : 0);
flags.status_flags_.OB_SERVER_STATUS_AUTOCOMMIT = (ac ? 1 : 0);
flags.status_flags_.OB_SERVER_MORE_RESULTS_EXISTS = false;
flags.status_flags_.OB_SERVER_PS_OUT_PARAMS = false;
flags.status_flags_.OB_SERVER_STATUS_CURSOR_EXISTS = true;
if (!session.is_obproxy_mode()) {
// in java client or others, use slow query bit to indicate partition hit or not
flags.status_flags_.OB_SERVER_QUERY_WAS_SLOW = !session.partition_hit().get_bool();
}
eofp.set_server_status(flags);
// for obproxy, 最后一次要把 eof和OK包放一起
if (OB_SUCC(ret)) {
if (OB_FAIL(update_last_pkt_pos())) {
LOG_WARN("failed to update last packet pos", K(ret));
} else {
// do nothing
}
}
// for obproxy
if (OB_SUCC(ret)) {
// in multi-stmt, send extra ok packet in the last stmt(has no more result)
if (need_send_extra_ok_packet()) {
ObOKPParam ok_param;
ok_param.affected_rows_ = 0;
ok_param.is_partition_hit_ = session.partition_hit().get_bool();
ok_param.has_more_result_ = false;
if (OB_FAIL(send_ok_packet(session, ok_param, &eofp))) {
LOG_WARN("fail to send ok packt", K(ok_param), K(ret));
}
} else {
if (OB_FAIL(response_packet(eofp, &session))) {
LOG_WARN("response packet fail", K(ret));
}
}
}
}
return ret;
}

64
src/observer/mysql/obmp_stmt_fetch.cpp
View File

@ -266,58 +266,22 @@ int ObMPStmtFetch::response_query_header(ObSQLSessionInfo &session,
// TODO: 增加com类型的处理
int ret = OB_SUCCESS;
bool ac = true;
CK (OB_NOT_NULL(fields));
if (OB_SUCC(ret)) {
int64_t fields_count = fields->count();
if (fields_count <= 0) {
ret = OB_ERR_BAD_FIELD_ERROR;
LOG_WARN("cursor set column", K(ret), K(fields_count));
} else if (OB_FAIL(session.get_autocommit(ac))) {
ObSqlCtx ctx;
ObQueryRetryCtrl retry_ctrl;
ObSyncPlanDriver drv(gctx_,
ctx,
session,
retry_ctrl,
*this,
false,
OB_INVALID_COUNT);
if (NULL == fields) {
ret = OB_ERR_UNEXPECTED;
} else if (OB_FAIL(drv.response_query_header(*fields,
has_ok_packet(),
false))) {
LOG_WARN("fail to get autocommit", K(ret));
}
if (OB_SUCC(ret)) {
OMPKResheader rhp;
rhp.set_field_count(fields_count);
if (OB_FAIL(response_packet(rhp, &session))) {
LOG_WARN("response packet fail", K(ret));
}
}
if (OB_SUCC(ret)) {
for (int64_t i = 0; OB_SUCC(ret) && i < fields_count; ++i) {
ObMySQLField field;
const ObField &ob_field = fields->at(i);
ret = ObMySQLResultSet::to_mysql_field(ob_field, field);
if (OB_SUCC(ret)) {
ObMySQLResultSet::replace_lob_type(session, ob_field, field);
OMPKField fp(field);
if (OB_FAIL(response_packet(fp, &session))) {
LOG_WARN("response packet fail", K(ret));
}
}
}
}
}
if (OB_SUCC(ret)) {
OMPKEOF eofp;
eofp.set_warning_count(0);
ObServerStatusFlags flags = eofp.get_server_status();
flags.status_flags_.OB_SERVER_STATUS_IN_TRANS
= (session.is_server_status_in_transaction() ? 1 : 0);
flags.status_flags_.OB_SERVER_STATUS_AUTOCOMMIT = (ac ? 1 : 0);
flags.status_flags_.OB_SERVER_MORE_RESULTS_EXISTS = has_ok_packet() ? true : false;
flags.status_flags_.OB_SERVER_PS_OUT_PARAMS = false;
flags.status_flags_.OB_SERVER_STATUS_CURSOR_EXISTS = true;
if (!session.is_obproxy_mode()) {
// in java client or others, use slow query bit to indicate partition hit or not
flags.status_flags_.OB_SERVER_QUERY_WAS_SLOW = !session.partition_hit().get_bool();
}
eofp.set_server_status(flags);
if (OB_FAIL(response_packet(eofp, &session))) {
LOG_WARN("response packet fail", K(ret));
}
}
return ret;
}

223
src/observer/mysql/obmp_stmt_prexecute.cpp
View File

@ -352,32 +352,6 @@ int ObMPStmtPrexecute::before_process()
}
}
}
if (OB_FAIL(ret)) {
// do nothing
} else if (is_send_long_data()) {
// response prepare packet
if (!first_time_) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("send long data need stmt_id is 0.", K(ret));
} else {
OMPKPrepare prepare_packet;
prepare_packet.set_statement_id(static_cast<uint32_t>(result.get_statement_id()));
prepare_packet.set_column_num(static_cast<uint16_t>(result.get_field_cnt()));
prepare_packet.set_warning_count(static_cast<uint16_t>(result.get_warning_count()));
if (OB_ISNULL(result.get_param_fields())) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("invalid argument", K(ret), K(result.get_param_fields()));
} else {
prepare_packet.set_param_num(
static_cast<uint16_t>(result.get_param_fields()->count()));
}
if (OB_SUCC(ret) && OB_FAIL(response_packet(prepare_packet, &result.get_session()))) {
LOG_WARN("response packet failed", K(ret));
}
}
}
LOG_DEBUG("after get packet in prexecute protocol.", K(close_stmt_count_),
K(exec_mode_), K(extend_flag_));
}
}
}
@ -401,6 +375,28 @@ int ObMPStmtPrexecute::before_process()
return ret;
}
int ObMPStmtPrexecute::clean_ps_stmt(ObSQLSessionInfo &session,
const bool is_local_retry,
const bool is_batch)
{
int ret = OB_SUCCESS;
if (first_time_
&& ((!is_batch && !is_local_retry)
|| (is_batch && THIS_WORKER.need_retry()))) {
/* 清理 ps stmt 的时机
* 1. 第一次执行时，也就是在 before_process 中执行 prepare 生成 stmt 时
* 2. 第一次执行且 非batch 模式， 非 local retry 的报错都需要清理
* 3. 第一次执行且 batch 模式， 参考 try_batch_multi_stmt_optimization 的实现，
* 只有 THIS_WORKER.need_retry() 的时候队列重试需要清理，
* 其他时候都退化成了 local 重试，不需要清理
*/
if (OB_FAIL(session.close_ps_stmt(stmt_id_))) {
LOG_WARN("close cursor failed.", K(ret), K(stmt_id_));
}
}
return ret;
}
int ObMPStmtPrexecute::execute_response(ObSQLSessionInfo &session,
ParamStore &params,
sql::ObSqlCtx &ctx,
@ -633,28 +629,14 @@ int ObMPStmtPrexecute::execute_response(ObSQLSessionInfo &session,
if (OB_SUCCESS != close_ret) {
LOG_WARN("fail to close result", K(close_ret));
}
int tmp_ret = clean_ps_stmt(session,
RETRY_TYPE_LOCAL == retry_ctrl.get_retry_type(),
ctx.multi_stmt_item_.is_batched_multi_stmt());
if (OB_SUCCESS != tmp_ret) {
LOG_WARN("prexecute clean ps stmt fail. ", K(ret), K(tmp_ret));
}
} else {
if (first_time_
&& stmt::T_ANONYMOUS_BLOCK == stmt_type_
&& !is_send_long_data()
&& !get_arraybounding()) {
int8_t has_result = 0;
if (OB_NOT_NULL(result.get_field_columns())
&& result.get_field_columns()->count() > 0) {
has_result = 1;
}
if (OB_FAIL(response_query_header(session,
result.get_field_columns(),
result.get_param_fields(),
NULL,
has_result,
result.get_warning_count(),
has_result /*ps_out*/))) {
LOG_WARN("send header packet faild.", K(ret));
}
}
int8_t has_result = 0;
if (OB_SUCC(ret)) {
//监控项统计开始
set_exec_start_timestamp(ObTimeUtility::current_time());
// 本分支内如果出错, 全部会在response_result内部处理妥当, 无需再额外处理回复错误包
@ -663,25 +645,11 @@ int ObMPStmtPrexecute::execute_response(ObSQLSessionInfo &session,
ctx.is_show_trace_stmt_ = ObStmt::is_show_trace_stmt(result.get_literal_stmt_type());
session.set_current_execution_id(execution_id);
result.set_statement_id(stmt_id_);
check_has_result(result, has_result);
}
if (OB_FAIL(ret)) {
} else if (get_arraybounding()) {
if (get_arraybounding()) {
if (OB_FAIL(after_do_process_for_arraybinding(session, result))) {
LOG_WARN("failed to process arraybinding sql", K(ret));
}
} else {
if (OB_FAIL(response_query_header(session,
result.get_field_columns(),
result.get_param_fields(),
result.get_returning_param_fields(),
has_result,
result.get_warning_count(),
1 == has_result ? true : false))) {
LOG_WARN("send header packet faild.", K(ret));
}
if (OB_SUCC(ret)) {
if (OB_FAIL(response_result(result, session, force_sync_resp, async_resp_used))) {
} else if (OB_FAIL(response_result(result, session, force_sync_resp, async_resp_used))) {
ObPhysicalPlanCtx *plan_ctx = result.get_exec_context().get_physical_plan_ctx();
if (OB_ISNULL(plan_ctx)) {
LOG_ERROR("execute query fail, and plan_ctx is NULL", K(ret));
@ -689,6 +657,13 @@ int ObMPStmtPrexecute::execute_response(ObSQLSessionInfo &session,
LOG_WARN("execute query fail",
K(ret), "timeout_timestamp", plan_ctx->get_timeout_timestamp());
}
int tmp_ret = clean_ps_stmt(session,
RETRY_TYPE_LOCAL == retry_ctrl.get_retry_type(),
ctx.multi_stmt_item_.is_batched_multi_stmt());
if (OB_SUCCESS != tmp_ret) {
LOG_WARN("prexecute clean ps stmt fail. ", K(ret), K(tmp_ret));
}
} else if (OB_OCI_EXACT_FETCH == exec_mode_ && stmt::T_SELECT == stmt_type_) {
int exact_fetch_ret = OB_SUCCESS;
if (0 == iteration_count_) {
@ -720,7 +695,7 @@ int ObMPStmtPrexecute::execute_response(ObSQLSessionInfo &session,
K(exact_fetch_ret), K(ret));
}
}
} else if (sql::ObDMLStmt::is_dml_write_stmt(stmt_type_) && result.is_with_rows()) {
} /*else if (sql::ObDMLStmt::is_dml_write_stmt(stmt_type_) && result.is_with_rows()) {
if (OB_FAIL(send_ok_packet(session,
result.get_return_rows(),
session.partition_hit().get_bool(),
@ -730,9 +705,7 @@ int ObMPStmtPrexecute::execute_response(ObSQLSessionInfo &session,
LOG_WARN("fail to send ok packt", "affect_rows", result.get_return_rows(),
"is_partition_hit" ,session.partition_hit().get_bool(), K(ret));
}
}
}
}
}*/
if ((OB_SUCC(ret) && is_diagnostics_stmt) || async_resp_used) {
// if diagnostic stmt succeed, no need to clear warning buf.
// or async resp is used, it will be cleared in callback thread.
@ -744,93 +717,80 @@ int ObMPStmtPrexecute::execute_response(ObSQLSessionInfo &session,
return ret;
}
void ObMPStmtPrexecute::check_has_result(ObMySQLResultSet &result, int8_t &has_result)
{
if ((0 != iteration_count_
&& OB_NOT_NULL(result.get_field_columns())
&& result.get_field_columns()->count() > 0)
|| (OB_OCI_EXACT_FETCH == exec_mode_ && stmt::T_SELECT == stmt_type_)) {
has_result = 1;
}
}
int ObMPStmtPrexecute::response_query_header(ObSQLSessionInfo &session,
const ColumnsFieldIArray *fields,
const ParamsFieldIArray *inout_params,
const ParamsFieldIArray *returning_params_field,
int8_t has_result,
int64_t warning_count,
bool ps_out)
ObResultSet &result,
bool need_flush_buffer)
{
// TODO: 增加com类型的处理
int ret = OB_SUCCESS;
bool need_send_eof = false;
LOG_DEBUG("before response_query_header",KPC(fields), KPC(inout_params), KPC(returning_params_field));
if (!prepare_packet_sent_) {
uint64_t params_cnt = 0;
const ColumnsFieldIArray *fields = result.get_field_columns();
const ParamsFieldIArray *param_fields = result.get_param_fields();
const ParamsFieldIArray *returning_params_field = result.get_returning_param_fields();
uint64_t params_cnt = OB_NOT_NULL(param_fields) ? param_fields->count() : 0;
int64_t fields_count = OB_NOT_NULL(fields) ? fields->count() : 0;
uint64_t returning_params_cnt = 0;
int64_t fields_count = 0;
bool has_arraybinding_result = false;
bool has_ps_out = false;
if (OB_NOT_NULL(inout_params)) {
params_cnt = inout_params->count();
}
if (OB_NOT_NULL(fields)) {
fields_count = fields->count();
}
int8_t has_result = 0;
bool ps_out = false;
LOG_DEBUG("before response_query_header",KPC(fields), KPC(param_fields), KPC(returning_params_field));
// check has arraybinding result
if (OB_NOT_NULL(returning_params_field) && is_arraybinding_has_result_type(stmt_type_)) {
/*
* 1. arraybinding 带结果集的语句类型 包含了 DML 语句 + 匿名块 + CALL
* 1. returning_params_field 不为空 且语句类型满足 1 的情况，认为 arraybinding 有结果集返回
* 2. param 的个数包含了 returning 的个数
*/
returning_params_cnt = returning_params_field->count();
params_cnt = params_cnt + returning_params_cnt;
has_arraybinding_result = returning_params_cnt > 0 ? true : false;
}
if (returning_params_cnt > 0) {
has_arraybinding_result = true;
// check has result
if (((0 != iteration_count_ || stmt::T_ANONYMOUS_BLOCK == stmt_type_) && fields_count > 0)
|| (OB_OCI_EXACT_FETCH == exec_mode_ && stmt::T_SELECT == stmt_type_)) {
/* has result 的几种情况：
* 1. 预取且有结果集： iteration_count_ > 0 & fields_count > 0
* 2. 匿名块且有结果集 ： T_ANONYMOUS_BLOCK == stmt_type_ && fields_count > 0.
* 匿名块情况下无论 iteration_count_ 是多少，是否有预取， 都需要设置 has_result
* 3. exact_fetch 模式 ： OB_OCI_EXACT_FETCH == exec_mode_ && stmt::T_SELECT == stmt_type_
*/
has_result = 1;
}
if ((stmt::T_ANONYMOUS_BLOCK == stmt_type_ || stmt::T_CALL_PROCEDURE == stmt_type_)
&& ps_out && get_arraybounding()) {
has_ps_out = true;
// check ps out
if ((stmt::T_ANONYMOUS_BLOCK == stmt_type_ || stmt::T_CALL_PROCEDURE == stmt_type_) && has_result) {
// PL 语句 + has_result
ps_out = true;
}
if (OB_FAIL(send_prepare_packet(stmt_id_,
// send packet
if (OB_FAIL(ret)) {
// do nothing
} else if (OB_FAIL(send_prepare_packet(stmt_id_,
fields_count,
params_cnt,
warning_count,
result.get_warning_count(),
has_result,
has_arraybinding_result,
has_ps_out))) {
ps_out && get_arraybounding()))) { // 只有 arraybinding + PL + 有结果集返回， prepare 中的 ps_out 才设置为 true
LOG_WARN("packet send prepare infomation fail", K(ret), K(stmt_id_));
}
if (OB_SUCC(ret) && params_cnt > 0) {
need_send_eof = true;
if (OB_FAIL(send_param_field_packet(session, inout_params))) {
} else if (params_cnt > 0 && OB_FAIL(send_param_field_packet(session, param_fields))) {
LOG_WARN("response param packet fail", K(ret));
}
}
if (OB_SUCC(ret) && returning_params_cnt > 0 && is_arraybinding_has_result_type(stmt_type_)) {
need_send_eof = true;
if (OB_FAIL(send_param_field_packet(session, returning_params_field))) {
} else if (returning_params_cnt > 0 && is_arraybinding_has_result_type(stmt_type_)
&& OB_FAIL(send_param_field_packet(session, returning_params_field))) {
LOG_WARN("response param packet fail", K(ret));
}
}
if (OB_SUCC(ret) && need_send_eof) {
if (OB_FAIL(send_eof_packet(session, 0, false, true, false))) {
} else if (params_cnt > 0 && OB_FAIL(send_eof_packet(session, 0, false, true, false))) {
LOG_WARN("send eof field failed", K(ret));
}
}
if (OB_SUCC(ret) && fields_count > 0) {
if (stmt::T_ANONYMOUS_BLOCK == stmt_type_
|| (OB_OCI_EXACT_FETCH == exec_mode_ && stmt::T_SELECT == stmt_type_)) {
// send column at sync_cmd_driver or sync_plan_driver
// do nothing here.
} else if (is_arraybinding_has_result_type(stmt_type_) && returning_params_field->count() > 0) {
// insert, update, delete returning stmt will send column at sync_plan_driver
// do nothing here.
} else if (OB_FAIL(send_column_packet(session, fields, ps_out))) {
} else if (fields_count > 0 && OB_FAIL(send_column_packet(session, fields, ps_out))) {
LOG_WARN("response column packet fail", K(ret));
}
}
if (OB_SUCC(ret)) {
} else if (need_flush_buffer && OB_FAIL(flush_buffer(false))) {
LOG_WARN("flush buffer fail before send async ok packet.", K(ret), K(stmt_id_));
} else {
prepare_packet_sent_ = true;
}
}
@ -1303,7 +1263,7 @@ int ObMPStmtPrexecute::send_column_packet(ObSQLSessionInfo &session,
}
}
if (OB_SUCC(ret) && fields_count > 0) {
if (OB_FAIL(send_eof_packet(session, 0, false, true, false, ps_out))) {
if (OB_FAIL(send_eof_packet(session, 0, false, is_ps_cursor(), false, ps_out))) {
LOG_WARN("column send eof fail", K(ret));
}
}
@ -1340,7 +1300,8 @@ int ObMPStmtPrexecute::send_eof_packet(ObSQLSessionInfo &session,
if(OB_FAIL(response_packet(eofp, &session))) {
LOG_WARN("response packet fail", K(ret));
} else {
LOG_DEBUG("send eof packet in prepare-execute protocol.");
LOG_DEBUG("send eof packet in prepare-execute protocol.", K(warning_count),
K(has_result), K(cursor_exist), K(last_row), K(ps_out));
}
}
}

17
src/observer/mysql/obmp_stmt_prexecute.h
View File

@ -72,12 +72,8 @@ public:
bool &async_resp_used,
ObPsStmtId &inner_stmt_id);
int response_query_header(sql::ObSQLSessionInfo &session,
const ColumnsFieldIArray *fields,
const ParamsFieldIArray *inout_params,
const ParamsFieldIArray *returning_params_field,
int8_t has_result,
int64_t warning_count = 0,
bool ps_out = false);
ObResultSet &result,
bool need_flush_buffer = false);
int response_param_query_header(sql::ObSQLSessionInfo &session,
const ColumnsFieldIArray *fields,
ParamStore *params,
@ -87,8 +83,8 @@ public:
bool ps_out = false);
inline int32_t get_iteration_count() { return iteration_count_; }
virtual bool is_send_long_data() { return SEND_LONG_DATA & extend_flag_;}
void check_has_result(ObMySQLResultSet &result, int8_t &has_result);
inline ObIAllocator *get_alloc() { return allocator_;}
int clean_ps_stmt(sql::ObSQLSessionInfo &session, const bool is_local_retry, const bool is_batch);
protected:
virtual int deserialize() { return common::OB_SUCCESS; }
@ -150,6 +146,13 @@ private:
private:
common::ObString sql_;
uint64_t sql_len_;
/*
* iteration_count_ 的含义
* 1. DML 语句 + iteration_count_ > 1 表示当前是 arraybinding 模式
* 2. arraybinding 模式下， 此值代表了 array 的大小
* 3. exact_fetch + select 模式下， 此值代表了返回结果集的大小
* 4. 其余场景，此值 > 0 表示需要有结果集返回
**/
int32_t iteration_count_;
uint32_t exec_mode_;
uint32_t close_stmt_count_;