chore: log more detail for temp table. (#18153)

Author: youngsofun

src/query/service/src/servers/http/middleware/session.rs

@@ -365,6 +365,7 @@ impl<E> HTTPSessionEndpoint<E> {
         req: &Request,
         query_id: String,
         login_history: &mut LoginHistory,
+        is_sticky_node: bool,
     ) -> Result<HttpQueryContext> {
         let client_host = get_client_ip(req);
         let node_id = GlobalConfig::instance().query.node_id.clone();
@@ -506,6 +507,7 @@ impl<E> HTTPSessionEndpoint<E> {
             client_host,
             client_session_id,
             user_name,
+            is_sticky_node,
         })
     }
 }
@@ -583,6 +585,8 @@ impl<E: Endpoint> Endpoint for HTTPSessionEndpoint<E> {
     async fn call(&self, mut req: Request) -> PoemResult<Self::Output> {
         let headers = req.headers().clone();
 
+        let mut is_sticky_node = false;
+
         if self.endpoint_kind.may_need_sticky() {
             if let Some(sticky_node_id) = headers.get(HEADER_STICKY) {
                 let sticky_node_id = sticky_node_id
@@ -593,6 +597,7 @@ impl<E: Endpoint> Endpoint for HTTPSessionEndpoint<E> {
                         )))
                     })?
                     .to_string();
+                is_sticky_node = true;
                 let local_id = GlobalConfig::instance().query.node_id.clone();
                 if local_id != sticky_node_id {
                     return if let Some(node) = ClusterDiscovery::instance()
@@ -664,7 +669,7 @@ impl<E: Endpoint> Endpoint for HTTPSessionEndpoint<E> {
 
                 log::warn!("Ignore header ({HEADER_WAREHOUSE}: {warehouse:?})");
             }
-        }
+        };
 
         let method = req.method().clone();
         let uri = req.uri().clone();
@@ -680,7 +685,10 @@ impl<E: Endpoint> Endpoint for HTTPSessionEndpoint<E> {
         login_history.connection_uri = uri.to_string();
 
         ThreadTracker::tracking_future(async move {
-            match self.auth(&req, query_id, &mut login_history).await {
+            match self
+                .auth(&req, query_id, &mut login_history, is_sticky_node)
+                .await
+            {
                 Ok(ctx) => {
                     login_history.event_type = LoginEventType::LoginSuccess;
                     login_history.write_to_log();

src/query/service/src/servers/http/v1/query/http_query.rs

@@ -39,6 +39,7 @@ use databend_common_metrics::http::metrics_incr_http_response_errors_count;
 use databend_common_settings::ScopeLevel;
 use databend_storages_common_session::TxnState;
 use http::StatusCode;
+use log::error;
 use log::info;
 use log::warn;
 use parking_lot::Mutex;
@@ -434,11 +435,19 @@ fn try_set_txn(
 impl HttpQuery {
     #[async_backtrace::framed]
     #[fastrace::trace]
-    pub async fn try_create(ctx: &HttpQueryContext, req: HttpQueryRequest) -> Result<HttpQuery> {
+    pub async fn try_create(
+        http_ctx: &HttpQueryContext,
+        req: HttpQueryRequest,
+    ) -> Result<HttpQuery> {
         let http_query_manager = HttpQueryManager::instance();
-        let session = ctx.upgrade_session(SessionType::HTTPQuery).map_err(|err| {
-            ErrorCode::Internal(format!("[HTTP-QUERY] Failed to upgrade session: {err}"))
-        })?;
+        let client_session_id = http_ctx.client_session_id.as_deref().unwrap_or("None");
+        let query_id = http_ctx.query_id.clone();
+        let session = http_ctx
+            .upgrade_session(SessionType::HTTPQuery)
+            .map_err(|err| {
+                ErrorCode::Internal(format!("[HTTP-QUERY] Failed to upgrade session: {err}"))
+            })?;
+        let session_id = session.get_id();
 
         // Read the session variables in the request, and set them to the current session.
         // the session variables includes:
@@ -492,24 +501,36 @@ impl HttpQuery {
                 }
             }
 
-            try_set_txn(&ctx.query_id, &session, session_conf, &http_query_manager)?;
+            try_set_txn(
+                &http_ctx.query_id,
+                &session,
+                session_conf,
+                &http_query_manager,
+            )?;
 
             if session_conf.need_sticky
                 && matches!(session_conf.txn_state, None | Some(TxnState::AutoCommit))
             {
-                http_query_manager.check_sticky_for_temp_table(&session_conf.last_server_info)?;
+                http_query_manager
+                    .check_sticky_for_temp_table(&session_conf.last_server_info)
+                    .map_err(|e| {
+                        let msg = format!(
+                            "[TEMP TABLE] invalid session, session_id={} query_id={}, error={e}, is_sticky_node={}",
+                           client_session_id, query_id, http_ctx.is_sticky_node,
+                        );
+                        error!("{}, session_state={:?}", msg, session_conf);
+                        ErrorCode::InvalidSessionState(msg)
+                    })?;
             }
         };
 
         let settings = session.get_settings();
         let result_timeout_secs = settings.get_http_handler_result_timeout_secs()?;
-        let deduplicate_label = &ctx.deduplicate_label;
-        let user_agent = &ctx.user_agent;
-        let query_id = ctx.query_id.clone();
+        let deduplicate_label = &http_ctx.deduplicate_label;
+        let user_agent = &http_ctx.user_agent;
 
-        session.set_client_host(ctx.client_host.clone());
+        session.set_client_host(http_ctx.client_host.clone());
 
-        let http_ctx = ctx;
         let ctx = session.create_query_context().await?;
 
         // Deduplicate label is used on the DML queries which may be retried by the client.
@@ -526,10 +547,9 @@ impl HttpQuery {
         // TODO: validate the query_id to be uuid format
         ctx.update_init_query_id(query_id.clone());
 
-        let session_id = session.get_id().clone();
         let node_id = ctx.get_cluster().local_id.clone();
         let sql = &req.sql;
-        info!(query_id = query_id, session_id = session_id, node_id = node_id, sql = sql; "[HTTP-QUERY] Creating new query");
+        info!(query_id = query_id, session_id = client_session_id, node_id = node_id, sql = sql; "[HTTP-QUERY] Creating new query");
 
         // Stage attachment is used to carry the data payload to the INSERT/REPLACE statements.
         // When stage attachment is specified, the query may looks like `INSERT INTO mytbl VALUES;`,

src/query/service/src/servers/http/v1/query/http_query_context.rs

@@ -44,6 +44,7 @@ pub struct HttpQueryContext {
     pub client_host: Option<String>,
     pub client_session_id: Option<String>,
     pub user_name: String,
+    pub is_sticky_node: bool,
 }
 
 impl HttpQueryContext {

src/query/service/src/servers/http/v1/query/http_query_manager.rs

@@ -246,25 +246,22 @@ impl HttpQueryManager {
     pub(crate) fn check_sticky_for_temp_table(
         &self,
         last_server_info: &Option<ServerInfo>,
-    ) -> Result<()> {
+    ) -> std::result::Result<(), String> {
         if let Some(ServerInfo { id, start_time }) = last_server_info {
             if self.server_info.id != *id {
-                return Err(ErrorCode::InvalidSessionState(format!(
-                    "[HTTP-QUERY] Session contains temporary tables but request was routed to the wrong server: current server is {}, expected server is {}",
+                return Err(format!(
+                    "contains temporary tables but request was routed to the wrong server: current server is {}, expected server is {}",
                     self.server_info.id, id
-                )));
+                ));
             }
             if self.server_info.start_time != *start_time {
-                return Err(ErrorCode::InvalidSessionState(format!(
-                    "[HTTP-QUERY] Temporary tables were lost because server restarted at {}",
+                return Err(format!(
+                    "temporary tables were lost because server restarted at {}",
                     start_time
-                )));
+                ));
             }
         } else {
-            return Err(ErrorCode::InvalidSessionState(
-                "[HTTP-QUERY] Session contains temporary tables but missing server_info field"
-                    .to_string(),
-            ));
+            return Err("contains temporary tables but missing last_server_info field".to_string());
         }
         Ok(())
     }