[CP] add this_work session set

src/observer/mysql/obmp_reset_connection.cpp

@@ -50,6 +50,7 @@ int ObMPResetConnection::process()
     ret = OB_ERR_UNEXPECTED;
     LOG_ERROR("fail to get session info", K(ret), K(session));
   } else {
+    THIS_WORKER.set_session(session);
     ObSQLSessionInfo::LockGuard lock_guard(session->get_query_lock());
     session->update_last_active_time();
 
@@ -83,6 +84,7 @@ int ObMPResetConnection::process()
     force_disconnect();
   }
 
+  THIS_WORKER.set_session(NULL);
   if (session != NULL) {
     revert_session(session);
   }

src/observer/mysql/obmp_stmt_execute.cpp

@@ -1683,6 +1683,7 @@ int ObMPStmtExecute::process()
     ObSQLSessionInfo &session = *sess;
     int64_t tenant_version = 0;
     int64_t sys_version = 0;
+    THIS_WORKER.set_session(sess);
     ObSQLSessionInfo::LockGuard lock_guard(session.get_query_lock());
     session.set_current_trace_id(ObCurTraceId::get_trace_id());
     session.get_raw_audit_record().request_memory_used_ = 0;
@@ -1811,6 +1812,7 @@ int ObMPStmtExecute::process()
     need_retry_ = true;
   }
 
+  THIS_WORKER.set_session(NULL);
   if (sess != NULL) {
     revert_session(sess); //current ignore revert session ret
   }

src/observer/mysql/obmp_stmt_fetch.cpp

@@ -719,6 +719,7 @@ int ObMPStmtFetch::process()
     ObSQLSessionInfo &session = *sess;
     int64_t tenant_version = 0;
     int64_t sys_version = 0;
+    THIS_WORKER.set_session(sess);
     ObSQLSessionInfo::LockGuard lock_guard(session.get_query_lock());
     session.set_current_trace_id(ObCurTraceId::get_trace_id());
     session.get_raw_audit_record().request_memory_used_ = 0;
@@ -780,6 +781,7 @@ int ObMPStmtFetch::process()
   if (!THIS_WORKER.need_retry()) {
     flush_ret = flush_buffer(true);
   }
+  THIS_WORKER.set_session(NULL);
   if (sess != NULL) {
     revert_session(sess); //current ignore revert session ret
   }

src/observer/mysql/obmp_stmt_get_piece_data.cpp

@@ -109,6 +109,7 @@ int ObMPStmtGetPieceData::process()
     LOG_WARN("update transmisson checksum flag failed", K(ret), K(stmt_id_));
   } else {
     ObSQLSessionInfo &session = *sess;
+    THIS_WORKER.set_session(sess);
     ObSQLSessionInfo::LockGuard lock_guard(session.get_query_lock());
     session.set_current_trace_id(ObCurTraceId::get_trace_id());
     session.get_raw_audit_record().request_memory_used_ = 0;
@@ -150,6 +151,7 @@ int ObMPStmtGetPieceData::process()
     }
 
     session.set_last_trace_id(ObCurTraceId::get_trace_id());
+    THIS_WORKER.set_session(NULL);
     revert_session(sess); //current ignore revert session ret
   }
   return ret;

src/observer/mysql/obmp_stmt_prepare.cpp

@@ -175,6 +175,7 @@ int ObMPStmtPrepare::process()
     LOG_WARN("update transmisson checksum flag failed", K(ret));
   } else {
     ObSQLSessionInfo &session = *sess;
+    THIS_WORKER.set_session(sess);
     ObSQLSessionInfo::LockGuard lock_guard(session.get_query_lock());
     session.set_current_trace_id(ObCurTraceId::get_trace_id());
     session.get_raw_audit_record().request_memory_used_ = 0;
@@ -265,6 +266,7 @@ int ObMPStmtPrepare::process()
     }
 
     session.set_last_trace_id(ObCurTraceId::get_trace_id());
+    THIS_WORKER.set_session(NULL);
     revert_session(sess); //current ignore revert session ret
   }
   return ret;

src/observer/mysql/obmp_stmt_reset.cpp

@@ -70,6 +70,7 @@ int ObMPStmtReset::process()
     const ObMySQLRawPacket &pkt = reinterpret_cast<const ObMySQLRawPacket&>(req_->get_packet());
     ObPieceCache *piece_cache = static_cast<ObPieceCache*>(session->get_piece_cache());
     int64_t param_num = 0;
+    THIS_WORKER.set_session(session);
     ObSQLSessionInfo::LockGuard lock_guard(session->get_query_lock());
     LOG_TRACE("close ps stmt or cursor", K_(stmt_id), K(session->get_sessid()));
 
@@ -138,6 +139,7 @@ int ObMPStmtReset::process()
   }
   flush_buffer(true);
 
+  THIS_WORKER.set_session(NULL);
   if (NULL != session) {
     revert_session(session);
   }

src/observer/mysql/obmp_stmt_send_long_data.cpp

@@ -115,6 +115,7 @@ int ObMPStmtSendLongData::process()
     LOG_WARN("update transmisson checksum flag failed", K(ret));
   } else {
     ObSQLSessionInfo &session = *sess;
+    THIS_WORKER.set_session(sess);
     ObSQLSessionInfo::LockGuard lock_guard(session.get_query_lock());
     session.set_current_trace_id(ObCurTraceId::get_trace_id());
     session.get_raw_audit_record().request_memory_used_ = 0;
@@ -162,6 +163,7 @@ int ObMPStmtSendLongData::process()
     }
 
     session.set_last_trace_id(ObCurTraceId::get_trace_id());
+    THIS_WORKER.set_session(NULL);
     revert_session(sess); //current ignore revert session ret
   }
   return ret;

src/observer/mysql/obmp_stmt_send_piece_data.cpp

@@ -121,6 +121,7 @@ int ObMPStmtSendPieceData::process()
     LOG_WARN("update transmisson checksum flag failed", K(ret));
   } else {
     ObSQLSessionInfo &session = *sess;
+    THIS_WORKER.set_session(sess);
     ObSQLSessionInfo::LockGuard lock_guard(session.get_query_lock());
     session.set_current_trace_id(ObCurTraceId::get_trace_id());
     session.get_raw_audit_record().request_memory_used_ = 0;
@@ -168,6 +169,7 @@ int ObMPStmtSendPieceData::process()
     }
 
     session.set_last_trace_id(ObCurTraceId::get_trace_id());
+    THIS_WORKER.set_session(NULL);
     revert_session(sess); //current ignore revert session ret
   }
   return ret;