From 116d5cf82b02c1b1ff54fb809dc4fe060dd1f09a Mon Sep 17 00:00:00 2001 From: codephage <39155160+codephage2020@users.noreply.github.com> Date: Tue, 24 Jun 2025 11:50:16 +0800 Subject: [PATCH] feat: support mysql flavor show processlist shortcut (#6328) (#6379) * feat: support mysql flavor show processlist shortcut (#6328) Signed-off-by: codephage. <381510760@qq.com> * Refactor SHOW PROCESSLIST handling and add tests Signed-off-by: codephage. <381510760@qq.com> * add sqlness test Signed-off-by: codephage. <381510760@qq.com> * add sqlness test result Signed-off-by: codephage. <381510760@qq.com> * fix sqlness test show_processList Signed-off-by: codephage. <381510760@qq.com> --------- Signed-off-by: codephage. <381510760@qq.com> --- .../src/system_schema/information_schema.rs | 2 +- .../information_schema/process_list.rs | 16 +++--- src/frontend/src/instance.rs | 2 + src/operator/src/statement.rs | 1 + src/operator/src/statement/show.rs | 17 ++++-- src/query/src/sql.rs | 52 +++++++++++++++++-- src/sql/src/parsers/show_parser.rs | 42 ++++++++++++++- src/sql/src/statements/show.rs | 17 ++++++ src/sql/src/statements/statement.rs | 7 ++- .../common/show/show_processList.result | 23 ++++++++ .../common/show/show_processList.sql | 10 ++++ 11 files changed, 169 insertions(+), 20 deletions(-) create mode 100644 tests/cases/standalone/common/show/show_processList.result create mode 100644 tests/cases/standalone/common/show/show_processList.sql diff --git a/src/catalog/src/system_schema/information_schema.rs b/src/catalog/src/system_schema/information_schema.rs index 5e24809852..28f1120b67 100644 --- a/src/catalog/src/system_schema/information_schema.rs +++ b/src/catalog/src/system_schema/information_schema.rs @@ -19,7 +19,7 @@ mod information_memory_table; pub mod key_column_usage; mod partitions; mod procedure_info; -mod process_list; +pub mod process_list; pub mod region_peers; mod region_statistics; mod runtime_metrics; diff --git a/src/catalog/src/system_schema/information_schema/process_list.rs b/src/catalog/src/system_schema/information_schema/process_list.rs index 46a94bd458..7c32f34759 100644 --- a/src/catalog/src/system_schema/information_schema/process_list.rs +++ b/src/catalog/src/system_schema/information_schema/process_list.rs @@ -39,14 +39,14 @@ use crate::process_manager::ProcessManagerRef; use crate::system_schema::information_schema::InformationTable; /// Column names of `information_schema.process_list` -const ID: &str = "id"; -const CATALOG: &str = "catalog"; -const SCHEMAS: &str = "schemas"; -const QUERY: &str = "query"; -const CLIENT: &str = "client"; -const FRONTEND: &str = "frontend"; -const START_TIMESTAMP: &str = "start_timestamp"; -const ELAPSED_TIME: &str = "elapsed_time"; +pub const ID: &str = "id"; +pub const CATALOG: &str = "catalog"; +pub const SCHEMAS: &str = "schemas"; +pub const QUERY: &str = "query"; +pub const CLIENT: &str = "client"; +pub const FRONTEND: &str = "frontend"; +pub const START_TIMESTAMP: &str = "start_timestamp"; +pub const ELAPSED_TIME: &str = "elapsed_time"; /// `information_schema.process_list` table implementation that tracks running /// queries in current cluster. diff --git a/src/frontend/src/instance.rs b/src/frontend/src/instance.rs index d17257892e..f10ad17a41 100644 --- a/src/frontend/src/instance.rs +++ b/src/frontend/src/instance.rs @@ -616,6 +616,8 @@ pub fn check_permission( Statement::FetchCursor(_) | Statement::CloseCursor(_) => {} // User can only kill process in their own catalog. Statement::Kill(_) => {} + // SHOW PROCESSLIST + Statement::ShowProcesslist(_) => {} } Ok(()) } diff --git a/src/operator/src/statement.rs b/src/operator/src/statement.rs index acc9265347..988f0846dc 100644 --- a/src/operator/src/statement.rs +++ b/src/operator/src/statement.rs @@ -370,6 +370,7 @@ impl StatementExecutor { Statement::Use(db) => self.use_database(db, query_ctx).await, Statement::Admin(admin) => self.execute_admin_command(admin, query_ctx).await, Statement::Kill(kill) => self.execute_kill(query_ctx, kill).await, + Statement::ShowProcesslist(show) => self.show_processlist(show, query_ctx).await, } } diff --git a/src/operator/src/statement/show.rs b/src/operator/src/statement/show.rs index bf6069d49d..f2f05345a7 100644 --- a/src/operator/src/statement/show.rs +++ b/src/operator/src/statement/show.rs @@ -25,7 +25,7 @@ use sql::ast::Ident; use sql::statements::create::Partitions; use sql::statements::show::{ ShowColumns, ShowCreateFlow, ShowCreateView, ShowDatabases, ShowFlows, ShowIndex, ShowKind, - ShowRegion, ShowTableStatus, ShowTables, ShowVariables, ShowViews, + ShowProcessList, ShowRegion, ShowTableStatus, ShowTables, ShowVariables, ShowViews, }; use sql::statements::OptionMap; use table::metadata::TableType; @@ -33,8 +33,9 @@ use table::table_name::TableName; use table::TableRef; use crate::error::{ - self, CatalogSnafu, ExecuteStatementSnafu, ExternalSnafu, FindViewInfoSnafu, InvalidSqlSnafu, - Result, TableMetadataManagerSnafu, ViewInfoNotFoundSnafu, ViewNotFoundSnafu, + self, CatalogSnafu, ExecLogicalPlanSnafu, ExecuteStatementSnafu, ExternalSnafu, + FindViewInfoSnafu, InvalidSqlSnafu, Result, TableMetadataManagerSnafu, ViewInfoNotFoundSnafu, + ViewNotFoundSnafu, }; use crate::statement::StatementExecutor; @@ -314,6 +315,16 @@ impl StatementExecutor { .await .context(error::ExecuteStatementSnafu) } + + pub async fn show_processlist( + &self, + stmt: ShowProcessList, + query_ctx: QueryContextRef, + ) -> Result { + query::sql::show_processlist(stmt, &self.query_engine, &self.catalog_manager, query_ctx) + .await + .context(ExecLogicalPlanSnafu) + } } pub(crate) fn create_partitions_stmt(partitions: Vec) -> Result> { diff --git a/src/query/src/sql.rs b/src/query/src/sql.rs index c5a77595e4..a30ef071ac 100644 --- a/src/query/src/sql.rs +++ b/src/query/src/sql.rs @@ -18,8 +18,8 @@ use std::collections::HashMap; use std::sync::Arc; use catalog::information_schema::{ - columns, flows, key_column_usage, region_peers, schemata, tables, CHARACTER_SETS, COLLATIONS, - COLUMNS, FLOWS, KEY_COLUMN_USAGE, REGION_PEERS, SCHEMATA, TABLES, VIEWS, + columns, flows, key_column_usage, process_list, region_peers, schemata, tables, CHARACTER_SETS, + COLLATIONS, COLUMNS, FLOWS, KEY_COLUMN_USAGE, REGION_PEERS, SCHEMATA, TABLES, VIEWS, }; use catalog::CatalogManagerRef; use common_catalog::consts::{ @@ -57,8 +57,8 @@ use sql::ast::Ident; use sql::parser::ParserContext; use sql::statements::create::{CreateDatabase, CreateFlow, CreateView, Partitions, SqlOrTql}; use sql::statements::show::{ - ShowColumns, ShowDatabases, ShowFlows, ShowIndex, ShowKind, ShowRegion, ShowTableStatus, - ShowTables, ShowVariables, ShowViews, + ShowColumns, ShowDatabases, ShowFlows, ShowIndex, ShowKind, ShowProcessList, ShowRegion, + ShowTableStatus, ShowTables, ShowVariables, ShowViews, }; use sql::statements::statement::Statement; use sql::statements::OptionMap; @@ -1233,6 +1233,50 @@ fn parse_file_table_format(options: &HashMap) -> Result Result { + let projects = if stmt.full { + vec![ + (process_list::ID, "Id"), + (process_list::CATALOG, "Catalog"), + (process_list::SCHEMAS, "Schema"), + (process_list::CLIENT, "Client"), + (process_list::FRONTEND, "Frontend"), + (process_list::START_TIMESTAMP, "Start Time"), + (process_list::ELAPSED_TIME, "Elapsed Time"), + (process_list::QUERY, "Query"), + ] + } else { + vec![ + (process_list::ID, "Id"), + (process_list::CATALOG, "Catalog"), + (process_list::QUERY, "Query"), + (process_list::ELAPSED_TIME, "Elapsed Time"), + ] + }; + + let filters = vec![]; + let like_field = None; + let sort = vec![col("id").sort(true, true)]; + query_from_information_schema_table( + query_engine, + catalog_manager, + query_ctx.clone(), + "process_list", + vec![], + projects.clone(), + filters, + like_field, + sort, + ShowKind::All, + ) + .await +} + #[cfg(test)] mod test { use std::sync::Arc; diff --git a/src/sql/src/parsers/show_parser.rs b/src/sql/src/parsers/show_parser.rs index f7a32385f4..3fc1ca3da3 100644 --- a/src/sql/src/parsers/show_parser.rs +++ b/src/sql/src/parsers/show_parser.rs @@ -25,8 +25,8 @@ use crate::error::{ use crate::parser::ParserContext; use crate::statements::show::{ ShowColumns, ShowCreateDatabase, ShowCreateFlow, ShowCreateTable, ShowCreateTableVariant, - ShowCreateView, ShowDatabases, ShowFlows, ShowIndex, ShowKind, ShowRegion, ShowSearchPath, - ShowStatus, ShowTableStatus, ShowTables, ShowVariables, ShowViews, + ShowCreateView, ShowDatabases, ShowFlows, ShowIndex, ShowKind, ShowProcessList, ShowRegion, + ShowSearchPath, ShowStatus, ShowTableStatus, ShowTables, ShowVariables, ShowViews, }; use crate::statements::statement::Statement; @@ -104,6 +104,8 @@ impl ParserContext<'_> { self.parse_show_columns(true) } else if self.consume_token("DATABASES") || self.consume_token("SCHEMAS") { self.parse_show_databases(true) + } else if self.consume_token("PROCESSLIST") { + self.parse_show_processlist(true) } else { self.unsupported(self.peek_token_as_string()) } @@ -119,6 +121,8 @@ impl ParserContext<'_> { Ok(Statement::ShowStatus(ShowStatus {})) } else if self.consume_token("SEARCH_PATH") { Ok(Statement::ShowSearchPath(ShowSearchPath {})) + } else if self.consume_token("PROCESSLIST") { + self.parse_show_processlist(false) } else { self.unsupported(self.peek_token_as_string()) } @@ -571,6 +575,15 @@ impl ParserContext<'_> { Ok(Statement::ShowFlows(ShowFlows { kind, database })) } + + fn parse_show_processlist(&mut self, full: bool) -> Result { + match self.parser.next_token().token { + Token::EOF | Token::SemiColon => { + Ok(Statement::ShowProcesslist(ShowProcessList { full })) + } + _ => self.unsupported(self.peek_token_as_string()), + } + } } #[cfg(test)] @@ -1212,4 +1225,29 @@ mod tests { ); assert_eq!(sql, stmts[0].to_string()); } + + #[test] + pub fn test_show_processlist() { + let sql = "SHOW PROCESSLIST"; + let result = + ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default()); + let stmts = result.unwrap(); + assert_eq!(1, stmts.len()); + assert_eq!( + stmts[0], + Statement::ShowProcesslist(ShowProcessList { full: false }) + ); + assert_eq!(sql, stmts[0].to_string()); + + let sql = "SHOW FULL PROCESSLIST"; + let result = + ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default()); + let stmts = result.unwrap(); + assert_eq!(1, stmts.len()); + assert_eq!( + stmts[0], + Statement::ShowProcesslist(ShowProcessList { full: true }) + ); + assert_eq!(sql, stmts[0].to_string()); + } } diff --git a/src/sql/src/statements/show.rs b/src/sql/src/statements/show.rs index 2fbf9db728..e18528a6fb 100644 --- a/src/sql/src/statements/show.rs +++ b/src/sql/src/statements/show.rs @@ -321,6 +321,23 @@ impl Display for ShowSearchPath { } } +/// SQL structure for `SHOW PROCESSLIST`. +#[derive(Debug, Clone, PartialEq, Eq, Visit, VisitMut, Serialize)] +pub struct ShowProcessList { + pub full: bool, +} +impl Display for ShowProcessList { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + if self.full { + write!(f, "SHOW FULL PROCESSLIST")?; + } else { + write!(f, "SHOW PROCESSLIST")?; + } + + Ok(()) + } +} + #[cfg(test)] mod tests { use std::assert_matches::assert_matches; diff --git a/src/sql/src/statements/statement.rs b/src/sql/src/statements/statement.rs index 030a2334c1..7ec2779278 100644 --- a/src/sql/src/statements/statement.rs +++ b/src/sql/src/statements/statement.rs @@ -37,8 +37,8 @@ use crate::statements::query::Query; use crate::statements::set_variables::SetVariables; use crate::statements::show::{ ShowColumns, ShowCreateDatabase, ShowCreateFlow, ShowCreateTable, ShowCreateView, - ShowDatabases, ShowFlows, ShowIndex, ShowKind, ShowRegion, ShowSearchPath, ShowStatus, - ShowTableStatus, ShowTables, ShowVariables, ShowViews, + ShowDatabases, ShowFlows, ShowIndex, ShowKind, ShowProcessList, ShowRegion, ShowSearchPath, + ShowStatus, ShowTableStatus, ShowTables, ShowVariables, ShowViews, }; use crate::statements::tql::Tql; use crate::statements::truncate::TruncateTable; @@ -141,6 +141,8 @@ pub enum Statement { CloseCursor(CloseCursor), // KILL Kill(Kill), + // SHOW PROCESSLIST + ShowProcesslist(ShowProcessList), } impl Display for Statement { @@ -198,6 +200,7 @@ impl Display for Statement { Statement::FetchCursor(s) => s.fmt(f), Statement::CloseCursor(s) => s.fmt(f), Statement::Kill(k) => k.fmt(f), + Statement::ShowProcesslist(s) => s.fmt(f), } } } diff --git a/tests/cases/standalone/common/show/show_processList.result b/tests/cases/standalone/common/show/show_processList.result new file mode 100644 index 0000000000..cec1aad7f9 --- /dev/null +++ b/tests/cases/standalone/common/show/show_processList.result @@ -0,0 +1,23 @@ +-- SQLNESS REPLACE \d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}:\d+/\d+ PROCESS_ID +-- SQLNESS REPLACE PT\d+\.\d+S ELAPSED_TIME +-- SQLNESS REPLACE [\u0020\-]+ +SHOW PROCESSLIST; + ++++++ +|Id|Catalog|Query|ElapsedTime| ++++++ +|PROCESS_ID|greptime|SHOWPROCESSLIST|ELAPSED_TIME| ++++++ + +-- SQLNESS REPLACE \d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}:\d+/\d+ PROCESS_ID +-- SQLNESS REPLACE PT\d+\.\d+S ELAPSED_TIME +-- SQLNESS REPLACE (\s[\-0-9T:\.]{15,}) DATETIME +-- SQLNESS REPLACE [\u0020\-]+ +SHOW FULL PROCESSLIST; + ++++++++++ +|Id|Catalog|Schema|Client|Frontend|StartTime|ElapsedTime|Query| ++++++++++ +|PROCESS_ID|greptime|public|unknown[unknownclientaddr]|DATETIME|DATETIME|ELAPSED_TIME|SHOWFULLPROCESSLIST| ++++++++++ + diff --git a/tests/cases/standalone/common/show/show_processList.sql b/tests/cases/standalone/common/show/show_processList.sql new file mode 100644 index 0000000000..77b9f0b49b --- /dev/null +++ b/tests/cases/standalone/common/show/show_processList.sql @@ -0,0 +1,10 @@ +-- SQLNESS REPLACE \d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}:\d+/\d+ PROCESS_ID +-- SQLNESS REPLACE PT\d+\.\d+S ELAPSED_TIME +-- SQLNESS REPLACE [\u0020\-]+ +SHOW PROCESSLIST; + +-- SQLNESS REPLACE \d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}:\d+/\d+ PROCESS_ID +-- SQLNESS REPLACE PT\d+\.\d+S ELAPSED_TIME +-- SQLNESS REPLACE (\s[\-0-9T:\.]{15,}) DATETIME +-- SQLNESS REPLACE [\u0020\-]+ +SHOW FULL PROCESSLIST;