oceanbase

1
/**
2
 * Copyright (c) 2021 OceanBase
3
 * OceanBase CE is licensed under Mulan PubL v2.
4
 * You can use this software according to the terms and conditions of the Mulan PubL v2.
5
 * You may obtain a copy of Mulan PubL v2 at:
6
 *          http://license.coscl.org.cn/MulanPubL-2.0
7
 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
8
 * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
9
 * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
10
 * See the Mulan PubL v2 for more details.
11
 */
12

13
#define USING_LOG_PREFIX SERVER
14
#include "obmp_stmt_close.h"
15
#include "rpc/ob_request.h"
16
#include "rpc/obmysql/ob_mysql_util.h"
17
#include "rpc/obmysql/ob_mysql_packet.h"
18
#include "sql/plan_cache/ob_ps_cache.h"
19
#include "sql/plan_cache/ob_prepare_stmt_struct.h"
20
#include "sql/session/ob_sql_session_info.h"
21
#include "observer/mysql/obmp_utils.h"
22
#include "observer/omt/ob_tenant.h"
23

24
namespace oceanbase
25
{
26
using namespace common;
27
using namespace rpc;
28
using namespace obmysql;
29
using namespace sql;
30

31
namespace observer
32
{
33

34
int ObMPStmtClose::deserialize()
35
{
36
  int ret = OB_SUCCESS;
37
  if (OB_ISNULL(req_)) {
38
    ret = OB_ERR_UNEXPECTED;
39
    LOG_WARN("invalid packet", K(ret), K_(req));
40
  } else if (OB_UNLIKELY(req_->get_type() != ObRequest::OB_MYSQL)) {
41
    ret = OB_ERR_UNEXPECTED;
42
    LOG_WARN("invalid packet", K(ret), K_(req), K(req_->get_type()));
43
  } else {
44
    const ObMySQLRawPacket &pkt = reinterpret_cast<const ObMySQLRawPacket&>(req_->get_packet());
45
    const char* pos = pkt.get_cdata();
46
    uint32_t stmt_id = -1; //INVALID_STMT_ID
47
    ObMySQLUtil::get_uint4(pos, stmt_id);
48
    stmt_id_ = stmt_id;
49
  }
50
  return ret;
51
}
52

53
int ObMPStmtClose::process()
54
{
55
  int ret = OB_SUCCESS;
56
  sql::ObSQLSessionInfo *session = NULL;
57
  trace::UUID ps_close_span_id;
58
  if (OB_ISNULL(req_) || OB_ISNULL(get_conn())) {
59
    ret = OB_INVALID_ARGUMENT;
60
    LOG_WARN("invalid packet", K(ret), KP(req_));
61
  } else if (OB_INVALID_STMT_ID == stmt_id_) {
62
    ret = OB_INVALID_ARGUMENT;
63
    LOG_WARN("stmt_id is invalid", K(ret));
64
  } else if (OB_FAIL(get_session(session))) {
65
    LOG_WARN("get session failed");
66
  } else if (OB_ISNULL(session)) {
67
    ret = OB_ERR_UNEXPECTED;
68
    LOG_WARN("session is NULL or invalid", K(ret), K(session));
69
  } else if (OB_FAIL(update_transmission_checksum_flag(*session))) {
70
    LOG_WARN("update transmisson checksum flag failed", K(ret));
71
  } else {
72
    const ObMySQLRawPacket &pkt = reinterpret_cast<const ObMySQLRawPacket&>(req_->get_packet());
73
    ObSQLSessionInfo::LockGuard lock_guard(session->get_query_lock());
74
    ObSessionStatEstGuard stat_est_guard(get_conn()->tenant_->id(), session->get_sessid());
75
    session->init_use_rich_format();
76
    LOG_TRACE("close ps stmt or cursor", K_(stmt_id), K(session->get_sessid()));
77
    if (OB_FAIL(sql::ObFLTUtils::init_flt_info(pkt.get_extra_info(), *session,
78
                     get_conn()->proxy_cap_flags_.is_full_link_trace_support()))) {
79
      LOG_WARN("failed to init flt extra info", K(ret));
80
    }
81
    FLTSpanGuard(ps_close);
82
    if (OB_FAIL(ret)) {
83
    } else if (is_cursor_close()) {
84
      if (OB_FAIL(session->close_cursor(stmt_id_))) {
85
        LOG_WARN("fail to close cursor", K(ret), K_(stmt_id), K(session->get_sessid()));
86
      }
87
    } else {
88
      int tmp_ret = OB_SUCCESS;
89
      if (OB_NOT_NULL(session->get_cursor(stmt_id_))) {
90
        if (OB_FAIL(session->close_cursor(stmt_id_))) {
91
          tmp_ret = ret;
92
          LOG_WARN("fail to close cursor", K(ret), K_(stmt_id), K(session->get_sessid()));
93
        }
94
      }
95
      if (OB_FAIL(session->close_ps_stmt(stmt_id_))) {
96
        LOG_WARN("fail to close ps stmt", K(ret), K_(stmt_id), K(session->get_sessid()));
97
      }
98
      if (OB_SUCCESS != tmp_ret) {
99
        // close_cursor 失败时错误码的优先级比 close_ps_stmt 高，此处进行覆盖
100
        ret = tmp_ret;
101
      }
102
    }
103
    if (OB_SUCC(ret)) {
104
      if (pkt.exist_trace_info()
105
          && OB_FAIL(session->update_sys_variable(share::SYS_VAR_OB_TRACE_INFO,
106
                                                  pkt.get_trace_info()))) {
107
        LOG_WARN("fail to update trace info", K(ret));
108
      }
109
    }
110
  }
111
  if (lib::is_diagnose_info_enabled()) {
112
    int64_t exec_end = ObTimeUtility::current_time();
113
    const int64_t time_cost = exec_end - get_receive_timestamp();
114
    EVENT_INC(SQL_PS_CLOSE_COUNT);
115
    EVENT_ADD(SQL_PS_CLOSE_TIME, time_cost);
116
  }
117
  if (NULL != session) {
118
    revert_session(session);
119
  }
120
  return ret;
121
}
122

123
} //end of namespace sql
124
} //end of namespace oceanbase
125