feat: remove useless optional fields in response of query_state_handler.

Author: youngsofun

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

@@ -447,7 +447,7 @@ impl<E> HTTPSessionEndpoint<E> {
             let sid = s.header.id.clone();
             session.set_client_session_id(sid.clone());
             login_history.session_id = sid.clone();
-            if !s.is_new_session {
+            if !s.is_new_session || is_worksheet {
                 // if session enabled by client:
                 //     log for the first request of the session.
                 // else:

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

@@ -165,34 +165,6 @@ pub struct QueryResponse {
 }
 
 impl QueryResponse {
-    pub(crate) fn closed(query_id: &str, close_reason: CloseReason) -> impl IntoResponse {
-        let id = query_id.to_string();
-        let state = match close_reason {
-            CloseReason::Finalized => ExecuteStateKind::Succeeded,
-            _ => ExecuteStateKind::Failed,
-        };
-        Json(QueryResponse {
-            id: query_id.to_string(),
-            session_id: None,
-            node_id: GlobalConfig::instance().query.node_id.clone(),
-            state,
-            session: None,
-            error: None,
-            warnings: vec![],
-            has_result_set: None,
-            schema: vec![],
-            data: Arc::new(BlocksSerializer::empty()),
-            affect: None,
-            result_timeout_secs: None,
-            stats: Default::default(),
-            stats_uri: None,
-            final_uri: None,
-            next_uri: None,
-            kill_uri: None,
-        })
-        .with_header(HEADER_QUERY_ID, id.clone())
-        .with_header(HEADER_QUERY_STATE, state.to_string())
-    }
     pub(crate) fn from_internal(
         id: String,
         r: HttpQueryResponseInternal,
@@ -272,6 +244,56 @@ impl QueryResponse {
     }
 }
 
+#[derive(Serialize, Debug, Clone)]
+pub struct StateResponse {
+    pub state: ExecuteStateKind,
+    pub error: Option<QueryError>,
+    pub warnings: Vec<String>,
+    pub stats: QueryStats,
+}
+
+impl StateResponse {
+    pub(crate) fn from_internal(id: String, r: HttpQueryResponseInternal) -> impl IntoResponse {
+        let state = r.state.clone();
+
+        if let Some(err) = &r.state.error {
+            metrics_incr_http_response_errors_count(err.name(), err.code());
+        }
+
+        let stats = QueryStats {
+            progresses: state.progresses.clone(),
+            running_time_ms: state.running_time_ms,
+        };
+        let rows = r.data.map(|d| d.page.data.num_rows()).unwrap_or_default();
+
+        Json(StateResponse {
+            state: state.state,
+            stats,
+            warnings: r.state.warnings,
+            error: r.state.error.map(QueryError::from_error_code),
+        })
+        .with_header(HEADER_QUERY_ID, id.clone())
+        .with_header(HEADER_QUERY_STATE, state.state.to_string())
+        .with_header(HEADER_QUERY_PAGE_ROWS, rows)
+    }
+
+    pub(crate) fn closed(query_id: &str, close_reason: CloseReason) -> impl IntoResponse {
+        let id = query_id.to_string();
+        let state = match close_reason {
+            CloseReason::Finalized => ExecuteStateKind::Succeeded,
+            _ => ExecuteStateKind::Failed,
+        };
+        Json(StateResponse {
+            state,
+            error: None,
+            warnings: vec![],
+            stats: Default::default(),
+        })
+        .with_header(HEADER_QUERY_ID, id.clone())
+        .with_header(HEADER_QUERY_STATE, state.to_string())
+    }
+}
+
 /// final is not ACKed by client, so client should not depend on the final response,
 ///
 /// for server:
@@ -368,12 +390,12 @@ async fn query_state_handler(
         match http_query_manager.get_query(&query_id) {
             Some(query) => {
                 if let Some(reason) = query.check_closed() {
-                    Ok(QueryResponse::closed(&query_id, reason.reason).into_response())
+                    Ok(StateResponse::closed(&query_id, reason.reason).into_response())
                 } else {
                     let response = query
                         .get_response_state_only()
                         .map_err(HttpErrorCode::server_error)?;
-                    Ok(QueryResponse::from_internal(query_id, response, false).into_response())
+                    Ok(StateResponse::from_internal(query_id, response).into_response())
                 }
             }
             None => Err(query_id_not_found(&query_id, &ctx.node_id)),

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

@@ -937,7 +937,7 @@ impl HttpQuery {
             }
             ClientState::Closed(st) => {
                 let to = match st.reason {
-                    CloseReason::Finalized => 30,
+                    CloseReason::Finalized => self.result_timeout_secs.min(30),
                     _ => self.result_timeout_secs,
                 };
                 let expire_at = st.ts + Duration::from_secs(to);

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

@@ -127,7 +127,7 @@ impl PageManager {
                 .clone())
         } else {
             let message = format!(
-                "[HTTP-QUERY] Invalid page number: requested {}, current page is {}",
+                "Invalid page number: requested {}, current page is {}",
                 page_no, next_no
             );
             Err(ErrorCode::HttpNotFound(message))

src/query/service/tests/it/servers/http/http_query_handlers.rs

@@ -348,16 +348,6 @@ async fn test_simple_sql() -> Result<()> {
     assert_eq!(result.data.len(), 10, "{:?}", result);
     assert_eq!(result.schema.len(), 31, "{:?}", result);
 
-    // get state
-    let uri = result.stats_uri.unwrap();
-    let (status, result) = get_uri_checked(&ep, &uri).await?;
-    assert_eq!(status, StatusCode::OK, "{:?}", result);
-    assert!(result.error.is_none(), "{:?}", result);
-    assert_eq!(result.data.len(), 0, "{:?}", result);
-    assert_eq!(result.next_uri, Some(final_uri.clone()), "{:?}", result);
-    // assert!(result.schema.is_empty(), "{:?}", result);
-    assert_eq!(result.state, ExecuteStateKind::Succeeded, "{:?}", result);
-
     // get page, support retry
     let page_0_uri = make_page_uri(query_id, 0);
     for _ in 1..3 {
@@ -385,7 +375,7 @@ async fn test_simple_sql() -> Result<()> {
     let body = response.into_body().into_string().await.unwrap();
     assert_eq!(
         body,
-        r#"{"error":{"code":404,"message":"[HTTP-QUERY] [HTTP-QUERY] Invalid page number: requested 2, current page is 1"}}"#
+        r#"{"error":{"code":404,"message":"[HTTP-QUERY] Invalid page number: requested 2, current page is 1"}}"#
     );
 
     // final
@@ -667,7 +657,7 @@ async fn test_pagination() -> Result<()> {
     let body = response.into_body().into_string().await.unwrap();
     assert_eq!(
         body,
-        r#"{"error":{"code":404,"message":"[HTTP-QUERY] [HTTP-QUERY] Invalid page number: requested 6, current page is 1"}}"#
+        r#"{"error":{"code":404,"message":"[HTTP-QUERY] Invalid page number: requested 6, current page is 1"}}"#
     );
 
     let mut next_uri = result.next_uri.clone().unwrap();
@@ -679,11 +669,6 @@ async fn test_pagination() -> Result<()> {
         assert!(result.error.is_none(), "{:?}", msg());
         assert!(!result.schema.is_empty(), "{:?}", result);
         if page == 5 {
-            // get state
-            let uri = result.stats_uri.clone().unwrap();
-            let (status, _state_result) = get_uri_checked(&ep, &uri).await?;
-            assert_eq!(status, StatusCode::OK);
-
             expect_end(&ep, result).await?;
         } else {
             assert_eq!(result.data.len(), 2, "{:?}", msg());