try fix sqllogic test cluster
@@ -124,7 +124,7 @@ impl SessionManager {
let user_api = UserApiProvider::instance();
let session_settings = Settings::try_create(&config, user_api, tenant).await?;
let session_ctx = SessionContext::try_create(config.clone(), session_settings)?;
- let session = Session::try_create(id, typ, session_ctx, mysql_conn_id)?;
+ let session = Session::try_create(id, typ.clone(), session_ctx, mysql_conn_id)?;
let mut sessions = self.active_sessions.write();
if sessions.len() < self.max_sessions {
@@ -134,7 +134,12 @@ impl SessionManager {
&config.query.cluster_id,
);
+ match typ {
+ SessionType::FlightRPC => {}
+ _ => {
sessions.insert(session.get_id(), session.clone());
+ }
+ }
Ok(session)
} else {
@@ -144,47 +149,6 @@ impl SessionManager {
}
}
- pub async fn create_rpc_session(
- self: &Arc<Self>,
- id: String,
- aborted: bool,
- ) -> Result<Arc<Session>> {
- // TODO: maybe deadlock?
- let config = self.get_conf();
- {
- let sessions = self.active_sessions.read();
- let v = sessions.get(&id);
- if v.is_some() {
- return Ok(v.unwrap().clone());
- }
- }
-
- let tenant = config.query.tenant_id.clone();
- let user_api = UserApiProvider::instance();
- let session_settings = Settings::try_create(&config, user_api, tenant).await?;
- let session_ctx = SessionContext::try_create(config.clone(), session_settings)?;
- let session = Session::try_create(id.clone(), SessionType::FlightRPC, session_ctx, None)?;
-
- let mut sessions = self.active_sessions.write();
- let v = sessions.get(&id);
- if v.is_none() {
- if aborted {
- return Err(ErrorCode::AbortedSession("Aborting server."));
- }
-
- label_counter(
- super::metrics::METRIC_SESSION_CONNECT_NUMBERS,
- &config.query.tenant_id,
- &config.query.cluster_id,
- );
-
- sessions.insert(id, session.clone());
- Ok(session)
- } else {
- Ok(v.unwrap().clone())
- }
- }
-
#[allow(clippy::ptr_arg)]
pub async fn get_session_by_id(self: &Arc<Self>, id: &str) -> Option<Arc<Session>> {
let sessions = self.active_sessions.read();
