chore(query): expose query hashes in system processes (#19927)

feat(query): expose query hashes in system processes

Author: dqhl76

src/query/catalog/src/table_context.rs

@@ -129,6 +129,8 @@ pub struct ProcessInfo {
     pub created_time: SystemTime,
     pub status_info: Option<String>,
     pub current_query_id: Option<String>,
+    pub query_hash: String,
+    pub query_parameterized_hash: String,
 }
 
 #[derive(Debug, Clone, Eq, PartialEq)]

src/query/service/src/sessions/session_info.rs

@@ -59,6 +59,14 @@ impl Session {
                 .as_ref()
                 .map(|qry_ctx| qry_ctx.get_status_info()),
             current_query_id: self.get_current_query_id(),
+            query_hash: shared_query_context
+                .as_ref()
+                .map(|qry_ctx| qry_ctx.get_query_text_hash())
+                .unwrap_or_default(),
+            query_parameterized_hash: shared_query_context
+                .as_ref()
+                .map(|qry_ctx| qry_ctx.get_query_parameterized_hash())
+                .unwrap_or_default(),
         }
     }
 

src/query/service/tests/it/storages/testdata/columns_table.txt

@@ -361,11 +361,13 @@ DB.Table: 'system'.'columns', Table: columns-table_id:1, ver:0, Engine: SystemCo
 | 'process_rows'                    | 'system'             | 'query_execution'         | 'UInt64'              | 'BIGINT UNSIGNED'   | ''       | ''       | 'NO'     | ''       |
 | 'process_time_in_micros'          | 'system'             | 'query_execution'         | 'UInt64'              | 'BIGINT UNSIGNED'   | ''       | ''       | 'NO'     | ''       |
 | 'processed'                       | 'system'             | 'notification_history'    | 'Nullable(Timestamp)' | 'TIMESTAMP'         | ''       | ''       | 'YES'    | ''       |
+| 'query_hash'                      | 'system'             | 'processes'               | 'String'              | 'VARCHAR'           | ''       | ''       | 'NO'     | ''       |
 | 'query_id'                        | 'system'             | 'backtrace'               | 'String'              | 'VARCHAR'           | ''       | ''       | 'NO'     | ''       |
 | 'query_id'                        | 'system'             | 'locks'                   | 'String'              | 'VARCHAR'           | ''       | ''       | 'NO'     | ''       |
 | 'query_id'                        | 'system'             | 'query_cache'             | 'String'              | 'VARCHAR'           | ''       | ''       | 'NO'     | ''       |
 | 'query_id'                        | 'system'             | 'query_execution'         | 'String'              | 'VARCHAR'           | ''       | ''       | 'NO'     | ''       |
 | 'query_id'                        | 'system'             | 'task_history'            | 'String'              | 'VARCHAR'           | ''       | ''       | 'NO'     | ''       |
+| 'query_parameterized_hash'        | 'system'             | 'processes'               | 'String'              | 'VARCHAR'           | ''       | ''       | 'NO'     | ''       |
 | 'range'                           | 'system'             | 'settings'                | 'String'              | 'VARCHAR'           | ''       | ''       | 'NO'     | ''       |
 | 'referenced_column_name'          | 'information_schema' | 'key_column_usage'        | 'NULL'                | 'NULL'              | ''       | ''       | 'NO'     | ''       |
 | 'referenced_table_name'           | 'information_schema' | 'key_column_usage'        | 'NULL'                | 'NULL'              | ''       | ''       | 'NO'     | ''       |

src/query/storages/system/src/processes_table.rs

@@ -76,6 +76,8 @@ impl SyncSystemTable for ProcessesTable {
         let mut processes_created_time = Vec::with_capacity(processes_info.len());
         let mut processes_status = Vec::with_capacity(processes_info.len());
         let mut processes_current_query_id = Vec::with_capacity(processes_info.len());
+        let mut processes_query_hash = Vec::with_capacity(processes_info.len());
+        let mut processes_query_parameterized_hash = Vec::with_capacity(processes_info.len());
 
         for process_info in &processes_info {
             let data_metrics = &process_info.data_metrics;
@@ -122,6 +124,8 @@ impl SyncSystemTable for ProcessesTable {
                     .clone()
                     .unwrap_or("".to_owned()),
             );
+            processes_query_hash.push(process_info.query_hash.clone());
+            processes_query_parameterized_hash.push(process_info.query_parameterized_hash.clone());
         }
 
         Ok(DataBlock::new_from_columns(vec![
@@ -143,6 +147,8 @@ impl SyncSystemTable for ProcessesTable {
             TimestampType::from_data(processes_created_time),
             StringType::from_data(processes_status),
             StringType::from_data(processes_current_query_id),
+            StringType::from_data(processes_query_hash),
+            StringType::from_data(processes_query_parameterized_hash),
         ]))
     }
 }
@@ -186,6 +192,8 @@ impl ProcessesTable {
             TableField::new("created_time", TableDataType::Timestamp),
             TableField::new("status", TableDataType::String),
             TableField::new("current_query_id", TableDataType::String),
+            TableField::new("query_hash", TableDataType::String),
+            TableField::new("query_parameterized_hash", TableDataType::String),
         ]);
 
         let table_info = TableInfo {