chore(query): add node column for processes table (databendlabs#13364)

* chore(query) add node column for processes table * chore(query) add node column for processes table
BohuTANG · Oct 21, 2023 · 46d2e86 · 46d2e86
1 parent a057943
commit 46d2e86
Show file tree

Hide file tree

Showing 2 changed files with 7 additions and 0 deletions.
diff --git a/src/query/service/tests/it/storages/testdata/columns_table.txt b/src/query/service/tests/it/storages/testdata/columns_table.txt
@@ -196,6 +196,7 @@ DB.Table: 'system'.'columns', Table: columns-table_id:1, ver:0, Engine: SystemCo
 | 'node'                          | 'system'             | 'backtrace'           | 'String'              | 'VARCHAR'           | ''       | ''       | 'NO'     | ''       |
 | 'node'                          | 'system'             | 'caches'              | 'String'              | 'VARCHAR'           | ''       | ''       | 'NO'     | ''       |
 | 'node'                          | 'system'             | 'metrics'             | 'String'              | 'VARCHAR'           | ''       | ''       | 'NO'     | ''       |
+| 'node'                          | 'system'             | 'processes'           | 'String'              | 'VARCHAR'           | ''       | ''       | 'NO'     | ''       |
 | 'node_id'                       | 'system'             | 'query_log'           | 'String'              | 'VARCHAR'           | ''       | ''       | 'NO'     | ''       |
 | 'non_unique'                    | 'information_schema' | 'statistics'          | 'NULL'                | 'NULL'              | ''       | ''       | 'NO'     | ''       |
 | 'nullable'                      | 'information_schema' | 'columns'             | 'Nullable(UInt8)'     | 'TINYINT UNSIGNED'  | ''       | ''       | 'YES'    | ''       |

diff --git a/src/query/storages/system/src/processes_table.rs b/src/query/storages/system/src/processes_table.rs
@@ -54,6 +54,9 @@ impl SyncSystemTable for ProcessesTable {
     fn get_full_data(&self, ctx: Arc<dyn TableContext>) -> Result<DataBlock> {
         let processes_info = ctx.get_processes_info();
 
+        let local_node = ctx.get_cluster().local_id.clone();
+
+        let mut nodes: Vec<Vec<u8>> = Vec::with_capacity(processes_info.len());
         let mut processes_id = Vec::with_capacity(processes_info.len());
         let mut processes_type = Vec::with_capacity(processes_info.len());
         let mut processes_host = Vec::with_capacity(processes_info.len());
@@ -79,6 +82,7 @@ impl SyncSystemTable for ProcessesTable {
                 .unwrap_or(Duration::from_secs(0))
                 .as_secs();
 
+            nodes.push(local_node.clone().into_bytes());
             processes_id.push(process_info.id.clone().into_bytes());
             processes_type.push(process_info.typ.clone().into_bytes());
             processes_state.push(process_info.state.to_string().into_bytes());
@@ -118,6 +122,7 @@ impl SyncSystemTable for ProcessesTable {
         }
 
         Ok(DataBlock::new_from_columns(vec![
+            StringType::from_data(nodes),
             StringType::from_data(processes_id),
             StringType::from_data(processes_type),
             StringType::from_opt_data(processes_host),
@@ -140,6 +145,7 @@ impl SyncSystemTable for ProcessesTable {
 impl ProcessesTable {
     pub fn create(table_id: u64) -> Arc<dyn Table> {
         let schema = TableSchemaRefExt::create(vec![
+            TableField::new("node", TableDataType::String),
             TableField::new("id", TableDataType::String),
             TableField::new("type", TableDataType::String),
             TableField::new(