fix: adhoc fix session leak

query/src/servers/clickhouse/interactive_worker.rs

@@ -50,12 +50,10 @@ impl ClickHouseSession for InteractiveWorker {
 
         let session = self.session.clone();
         let get_query_result = InteractiveWorkerBase::do_query(ctx, session);
-        let query_ctx = self
+        let format = self
             .session
-            .get_shared_query_context()
-            .await
+            .get_format_settings()
             .map_err(to_clickhouse_err)?;
-        let format = query_ctx.get_format_settings().map_err(to_clickhouse_err)?;
         if let Err(cause) = query_writer.write(get_query_result.await, &format).await {
             let new_error = cause.add_message(&ctx.state.query);
             return Err(to_clickhouse_err(new_error));

query/src/servers/mysql/mysql_interactive_worker.rs

@@ -187,11 +187,7 @@ impl<W: std::io::Write + Send + Sync> AsyncMysqlShim<W> for InteractiveWorker<W>
         let instant = Instant::now();
         let blocks = self.base.do_query(query).await;
 
-        let format = self
-            .session
-            .get_shared_query_context()
-            .await?
-            .get_format_settings()?;
+        let format = self.session.get_format_settings()?;
         let mut write_result = writer.write(blocks, &format);
 
         if let Err(cause) = write_result {

query/src/sessions/query_ctx.rs

@@ -62,7 +62,6 @@ use crate::clusters::Cluster;
 use crate::servers::http::v1::HttpQueryHandle;
 use crate::sessions::ProcessInfo;
 use crate::sessions::QueryContextShared;
-use crate::sessions::Session;
 use crate::sessions::SessionRef;
 use crate::sessions::Settings;
 use crate::storages::cache::CacheManager;
@@ -338,7 +337,7 @@ impl QueryContext {
     }
 
     pub fn get_format_settings(&self) -> Result<FormatSettings> {
-        self.shared.get_format_settings()
+        self.shared.session.get_format_settings()
     }
 
     pub fn get_config(&self) -> Config {
@@ -372,8 +371,8 @@ impl QueryContext {
     }
 
     // Get the current session.
-    pub fn get_current_session(self: &Arc<Self>) -> Arc<Session> {
-        self.shared.session.clone()
+    pub fn get_current_session(self: &Arc<Self>) -> SessionRef {
+        SessionRef::create(self.shared.session.clone())
     }
 
     // Get one session by session id.

query/src/sessions/query_ctx_shared.rs

@@ -19,13 +19,11 @@ use std::sync::atomic::AtomicUsize;
 use std::sync::atomic::Ordering;
 use std::sync::Arc;
 
-use chrono_tz::Tz;
 use common_base::base::Progress;
 use common_base::base::Runtime;
 use common_contexts::DalContext;
 use common_exception::ErrorCode;
 use common_exception::Result;
-use common_io::prelude::FormatSettings;
 use common_meta_types::UserInfo;
 use common_planners::PlanNode;
 use common_users::RoleCacheMgr;
@@ -40,7 +38,6 @@ use crate::catalogs::CatalogManager;
 use crate::clusters::Cluster;
 use crate::servers::http::v1::HttpQueryHandle;
 use crate::sessions::Session;
-use crate::sessions::SessionType;
 use crate::sessions::Settings;
 use crate::sql::SQLCommon;
 use crate::storages::Table;
@@ -290,34 +287,6 @@ impl QueryContextShared {
         self.session.get_config()
     }
 
-    pub fn get_format_settings(&self) -> Result<FormatSettings> {
-        let settings = self.get_settings();
-        let mut format = FormatSettings::default();
-        if let SessionType::HTTPQuery = self.session.get_type() {
-            format.false_bytes = vec![b'f', b'a', b'l', b's', b'e'];
-            format.true_bytes = vec![b't', b'r', b'u', b'e'];
-        }
-        {
-            format.record_delimiter = settings.get_record_delimiter()?;
-            format.field_delimiter = settings.get_field_delimiter()?;
-            format.empty_as_default = settings.get_empty_as_default()? > 0;
-            format.skip_header = settings.get_skip_header()? > 0;
-
-            let tz = String::from_utf8(settings.get_timezone()?).map_err(|_| {
-                ErrorCode::LogicalError("Timezone has been checked and should be valid.")
-            })?;
-            format.timezone = tz.parse::<Tz>().map_err(|_| {
-                ErrorCode::InvalidTimezone("Timezone has been checked and should be valid")
-            })?;
-
-            let compress = String::from_utf8(settings.get_compression()?).map_err(|_| {
-                ErrorCode::UnknownCompressionType("Compress type must be valid utf-8")
-            })?;
-            format.compression = compress.parse()?
-        }
-        Ok(format)
-    }
-
     pub fn get_connection_id(&self) -> String {
         self.session.get_id()
     }

query/src/sessions/session.rs

@@ -16,9 +16,11 @@ use std::net::SocketAddr;
 use std::sync::atomic::AtomicUsize;
 use std::sync::Arc;
 
+use chrono_tz::Tz;
 use common_base::mem_allocator::malloc_size;
 use common_exception::ErrorCode;
 use common_exception::Result;
+use common_io::prelude::FormatSettings;
 use common_macros::MallocSizeOf;
 use common_meta_types::GrantObject;
 use common_meta_types::UserInfo;
@@ -153,6 +155,34 @@ impl Session {
         Ok(shared)
     }
 
+    pub fn get_format_settings(&self) -> Result<FormatSettings> {
+        let settings = &self.session_settings;
+        let mut format = FormatSettings::default();
+        if let SessionType::HTTPQuery = self.get_type() {
+            format.false_bytes = vec![b'f', b'a', b'l', b's', b'e'];
+            format.true_bytes = vec![b't', b'r', b'u', b'e'];
+        }
+        {
+            format.record_delimiter = settings.get_record_delimiter()?;
+            format.field_delimiter = settings.get_field_delimiter()?;
+            format.empty_as_default = settings.get_empty_as_default()? > 0;
+            format.skip_header = settings.get_skip_header()? > 0;
+
+            let tz = String::from_utf8(settings.get_timezone()?).map_err(|_| {
+                ErrorCode::LogicalError("Timezone has been checked and should be valid.")
+            })?;
+            format.timezone = tz.parse::<Tz>().map_err(|_| {
+                ErrorCode::InvalidTimezone("Timezone has been checked and should be valid")
+            })?;
+
+            let compress = String::from_utf8(settings.get_compression()?).map_err(|_| {
+                ErrorCode::UnknownCompressionType("Compress type must be valid utf-8")
+            })?;
+            format.compression = compress.parse()?
+        }
+        Ok(format)
+    }
+
     pub fn get_current_query_id(&self) -> Option<String> {
         self.session_ctx.get_current_query_id()
     }