From d22084e90c5087fb443f6051ebd0d05caddc1043 Mon Sep 17 00:00:00 2001 From: fys <40801205+fengys1996@users.noreply.github.com> Date: Thu, 29 May 2025 12:14:27 +0800 Subject: [PATCH] feat: support parsing trigger create sql (#6197) * feat: support parsing trigger create sql * chore: add context for TryFromInt error * chore: refine error msg about int convert * avoid clone of token --- licenserc.toml | 3 + src/frontend/Cargo.toml | 1 + src/frontend/src/instance.rs | 4 + src/operator/Cargo.toml | 1 + src/operator/src/error.rs | 9 + src/operator/src/statement.rs | 2 + src/operator/src/statement/ddl.rs | 10 + src/sql/Cargo.toml | 3 + src/sql/src/error.rs | 40 ++ src/sql/src/parsers/create_parser.rs | 51 +- src/sql/src/parsers/create_parser/trigger.rs | 679 +++++++++++++++++++ src/sql/src/statements/create.rs | 3 + src/sql/src/statements/create/trigger.rs | 120 ++++ src/sql/src/statements/statement.rs | 5 + 14 files changed, 912 insertions(+), 19 deletions(-) create mode 100644 src/sql/src/parsers/create_parser/trigger.rs create mode 100644 src/sql/src/statements/create/trigger.rs diff --git a/licenserc.toml b/licenserc.toml index f18ed6b2eac..6783bc98feb 100644 --- a/licenserc.toml +++ b/licenserc.toml @@ -26,6 +26,9 @@ excludes = [ "src/common/base/src/secrets.rs", "src/servers/src/repeated_field.rs", "src/servers/src/http/test_helpers.rs", + # enterprise + "src/sql/src/statements/create/trigger.rs", + "src/sql/src/parsers/create_parser/trigger.rs", ] [properties] diff --git a/src/frontend/Cargo.toml b/src/frontend/Cargo.toml index b6ae383c328..9d54845f21b 100644 --- a/src/frontend/Cargo.toml +++ b/src/frontend/Cargo.toml @@ -6,6 +6,7 @@ license.workspace = true [features] testing = [] +enterprise = ["operator/enterprise", "sql/enterprise"] [lints] workspace = true diff --git a/src/frontend/src/instance.rs b/src/frontend/src/instance.rs index c527d270076..3887379cad6 100644 --- a/src/frontend/src/instance.rs +++ b/src/frontend/src/instance.rs @@ -495,6 +495,10 @@ pub fn check_permission( // TODO: should also validate source table name here? validate_param(&stmt.sink_table_name, query_ctx)?; } + #[cfg(feature = "enterprise")] + Statement::CreateTrigger(stmt) => { + validate_param(&stmt.trigger_name, query_ctx)?; + } Statement::CreateView(stmt) => { validate_param(&stmt.name, query_ctx)?; } diff --git a/src/operator/Cargo.toml b/src/operator/Cargo.toml index f6d56b1e048..8ce93ff8278 100644 --- a/src/operator/Cargo.toml +++ b/src/operator/Cargo.toml @@ -6,6 +6,7 @@ license.workspace = true [features] testing = [] +enterprise = ["sql/enterprise"] [lints] workspace = true diff --git a/src/operator/src/error.rs b/src/operator/src/error.rs index 8b80977b8b8..36b8cda6853 100644 --- a/src/operator/src/error.rs +++ b/src/operator/src/error.rs @@ -822,6 +822,13 @@ pub enum Error { #[snafu(implicit)] location: Location, }, + + #[cfg(feature = "enterprise")] + #[snafu(display("Trigger related operations are not currently supported"))] + UnsupportedTrigger { + #[snafu(implicit)] + location: Location, + }, } pub type Result = std::result::Result; @@ -871,6 +878,8 @@ impl ErrorExt for Error { Error::NotSupported { .. } | Error::ShowCreateTableBaseOnly { .. } | Error::SchemaReadOnly { .. } => StatusCode::Unsupported, + #[cfg(feature = "enterprise")] + Error::UnsupportedTrigger { .. } => StatusCode::Unsupported, Error::TableMetadataManager { source, .. } => source.status_code(), Error::ParseSql { source, .. } => source.status_code(), Error::InvalidateTableCache { source, .. } => source.status_code(), diff --git a/src/operator/src/statement.rs b/src/operator/src/statement.rs index d786396e87e..8224308fb94 100644 --- a/src/operator/src/statement.rs +++ b/src/operator/src/statement.rs @@ -224,6 +224,8 @@ impl StatementExecutor { Ok(Output::new_with_affected_rows(0)) } Statement::CreateFlow(stmt) => self.create_flow(stmt, query_ctx).await, + #[cfg(feature = "enterprise")] + Statement::CreateTrigger(stmt) => self.create_trigger(stmt, query_ctx).await, Statement::DropFlow(stmt) => { self.drop_flow( query_ctx.current_catalog().to_string(), diff --git a/src/operator/src/statement/ddl.rs b/src/operator/src/statement/ddl.rs index 19a971c2735..ccaa1bfa89a 100644 --- a/src/operator/src/statement/ddl.rs +++ b/src/operator/src/statement/ddl.rs @@ -343,6 +343,16 @@ impl StatementExecutor { .collect()) } + #[cfg(feature = "enterprise")] + #[tracing::instrument(skip_all)] + pub async fn create_trigger( + &self, + _stmt: sql::statements::create::trigger::CreateTrigger, + _query_context: QueryContextRef, + ) -> Result { + crate::error::UnsupportedTriggerSnafu {}.fail() + } + #[tracing::instrument(skip_all)] pub async fn create_flow( &self, diff --git a/src/sql/Cargo.toml b/src/sql/Cargo.toml index 812fe42709b..0cd54aba314 100644 --- a/src/sql/Cargo.toml +++ b/src/sql/Cargo.toml @@ -7,6 +7,9 @@ license.workspace = true [lints] workspace = true +[features] +enterprise = [] + [dependencies] api.workspace = true chrono.workspace = true diff --git a/src/sql/src/error.rs b/src/sql/src/error.rs index 6d503f9541e..b46233fe61b 100644 --- a/src/sql/src/error.rs +++ b/src/sql/src/error.rs @@ -370,6 +370,38 @@ pub enum Error { #[snafu(implicit)] location: Location, }, + + #[cfg(feature = "enterprise")] + #[snafu(display("Missing `{}` clause", name))] + MissingClause { + name: String, + #[snafu(implicit)] + location: Location, + }, + + #[cfg(feature = "enterprise")] + #[snafu(display("Unrecognized trigger webhook option key: {}", key))] + InvalidTriggerWebhookOption { + key: String, + #[snafu(implicit)] + location: Location, + }, + + #[cfg(feature = "enterprise")] + #[snafu(display("The execution interval cannot be negative"))] + NegativeInterval { + #[snafu(source)] + error: std::num::TryFromIntError, + #[snafu(implicit)] + location: Location, + }, + + #[cfg(feature = "enterprise")] + #[snafu(display("Must specify at least one notify channel"))] + MissingNotifyChannel { + #[snafu(implicit)] + location: Location, + }, } impl ErrorExt for Error { @@ -390,6 +422,9 @@ impl ErrorExt for Error { | UnexpectedToken { .. } | InvalidDefault { .. } => StatusCode::InvalidSyntax, + #[cfg(feature = "enterprise")] + MissingClause { .. } | MissingNotifyChannel { .. } => StatusCode::InvalidSyntax, + InvalidColumnOption { .. } | InvalidTableOptionValue { .. } | InvalidDatabaseName { .. } @@ -410,6 +445,11 @@ impl ErrorExt for Error { | UnsupportedUnaryOp { .. } | ConvertStr { .. } => StatusCode::InvalidArguments, + #[cfg(feature = "enterprise")] + InvalidTriggerWebhookOption { .. } | NegativeInterval { .. } => { + StatusCode::InvalidArguments + } + SerializeColumnDefaultConstraint { source, .. } => source.status_code(), ConvertToGrpcDataType { source, .. } => source.status_code(), Datatype { source, .. } => source.status_code(), diff --git a/src/sql/src/parsers/create_parser.rs b/src/sql/src/parsers/create_parser.rs index 5aa78832b37..a61d38f0999 100644 --- a/src/sql/src/parsers/create_parser.rs +++ b/src/sql/src/parsers/create_parser.rs @@ -12,6 +12,9 @@ // See the License for the specific language governing permissions and // limitations under the License. +#[cfg(feature = "enterprise")] +pub mod trigger; + use std::collections::HashMap; use common_catalog::consts::default_engine; @@ -98,6 +101,12 @@ impl<'a> ParserContext<'a> { self.parse_create_view(false) } + #[cfg(feature = "enterprise")] + Keyword::TRIGGER => { + let _ = self.parser.next_token(); + self.parse_create_trigger() + } + Keyword::NoKeyword => { let _ = self.parser.next_token(); let uppercase = w.value.to_uppercase(); @@ -281,25 +290,7 @@ impl<'a> ParserContext<'a> { .parser .consume_tokens(&[Token::make_keyword(EXPIRE), Token::make_keyword(AFTER)]) { - let expire_after_expr = self.parser.parse_expr().context(error::SyntaxSnafu)?; - let expire_after_lit = - utils::parser_expr_to_scalar_value_literal(expire_after_expr.clone())? - .cast_to(&ArrowDataType::Interval(IntervalUnit::MonthDayNano)) - .ok() - .with_context(|| InvalidIntervalSnafu { - reason: format!("cannot cast {} to interval type", expire_after_expr), - })?; - if let ScalarValue::IntervalMonthDayNano(Some(interval)) = expire_after_lit { - Some( - interval.nanoseconds / 1_000_000_000 - + interval.days as i64 * 60 * 60 * 24 - + interval.months as i64 * 60 * 60 * 24 * 3044 / 1000, // 1 month=365.25/12=30.44 days - // this is to keep the same as https://docs.rs/humantime/latest/humantime/fn.parse_duration.html - // which we use in database to parse i.e. ttl interval and many other intervals - ) - } else { - unreachable!() - } + Some(self.parse_interval()?) } else { None }; @@ -353,6 +344,28 @@ impl<'a> ParserContext<'a> { })) } + /// Parse the interval expr to duration in seconds. + fn parse_interval(&mut self) -> Result { + let interval_expr = self.parser.parse_expr().context(error::SyntaxSnafu)?; + let interval = utils::parser_expr_to_scalar_value_literal(interval_expr.clone())? + .cast_to(&ArrowDataType::Interval(IntervalUnit::MonthDayNano)) + .ok() + .with_context(|| InvalidIntervalSnafu { + reason: format!("cannot cast {} to interval type", interval_expr), + })?; + if let ScalarValue::IntervalMonthDayNano(Some(interval)) = interval { + Ok( + interval.nanoseconds / 1_000_000_000 + + interval.days as i64 * 60 * 60 * 24 + + interval.months as i64 * 60 * 60 * 24 * 3044 / 1000, // 1 month=365.25/12=30.44 days + // this is to keep the same as https://docs.rs/humantime/latest/humantime/fn.parse_duration.html + // which we use in database to parse i.e. ttl interval and many other intervals + ) + } else { + unreachable!() + } + } + fn parse_if_not_exist(&mut self) -> Result { match self.parser.peek_token().token { Token::Word(w) if Keyword::IF != w.keyword => return Ok(false), diff --git a/src/sql/src/parsers/create_parser/trigger.rs b/src/sql/src/parsers/create_parser/trigger.rs new file mode 100644 index 00000000000..24d9b27d762 --- /dev/null +++ b/src/sql/src/parsers/create_parser/trigger.rs @@ -0,0 +1,679 @@ +use std::collections::HashMap; + +use snafu::{ensure, OptionExt, ResultExt}; +use sqlparser::ast::Query; +use sqlparser::keywords::Keyword; +use sqlparser::parser::Parser; +use sqlparser::tokenizer::Token; + +use crate::error; +use crate::error::Result; +use crate::parser::ParserContext; +use crate::statements::create::trigger::{ + AlertManagerWebhook, ChannelType, CreateTrigger, NotifyChannel, +}; +use crate::statements::statement::Statement; +use crate::statements::OptionMap; +use crate::util::parse_option_string; + +/// Some keywords about trigger. +pub const ON: &str = "ON"; +pub const EVERY: &str = "EVERY"; +pub const LABELS: &str = "LABELS"; +pub const ANNOTATIONS: &str = "ANNOTATIONS"; +pub const NOTIFY: &str = "NOTIFY"; +pub const WEBHOOK: &str = "WEBHOOK"; +pub const URL: &str = "URL"; + +const TIMEOUT: &str = "timeout"; + +fn validate_webhook_option(key: &str) -> bool { + [TIMEOUT].contains(&key) +} + +impl<'a> ParserContext<'a> { + /// The SQL format as follows: + /// + /// ```sql + /// -- CREATE TRIGGER + /// [IF NOT EXISTS] + /// ON () + /// EVERY + /// [LABELS (=, ...)] + /// [ANNOTATIONS (=, ...)] + /// NOTIFY( + /// WEBHOOK URL '' [WITH (=, ...)], + /// WEBHOOK URL '' [WITH (=, ...)] + /// ) + /// ``` + /// + /// Note: This method does not parse the `CREATE TRIGGER` keywords. + pub(super) fn parse_create_trigger(&mut self) -> Result { + let if_not_exists = self.parse_if_not_exist()?; + let trigger_name = self.intern_parse_table_name()?; + + let mut may_query = None; + let mut may_interval = None; + let mut may_labels = None; + let mut may_annotations = None; + let mut notify_channels = vec![]; + + loop { + let next_token = self.parser.peek_token(); + match next_token.token { + Token::Word(w) if w.value.eq_ignore_ascii_case(ON) => { + self.parser.next_token(); + let (query, interval) = self.parse_trigger_on(true)?; + may_query.replace(query); + may_interval.replace(interval); + } + Token::Word(w) if w.value.eq_ignore_ascii_case(LABELS) => { + self.parser.next_token(); + let labels = self.parse_trigger_labels(true)?; + may_labels.replace(labels); + } + Token::Word(w) if w.value.eq_ignore_ascii_case(ANNOTATIONS) => { + self.parser.next_token(); + let annotations = self.parse_trigger_annotations(true)?; + may_annotations.replace(annotations); + } + Token::Word(w) if w.value.eq_ignore_ascii_case(NOTIFY) => { + self.parser.next_token(); + let channels = self.parse_trigger_notify(true)?; + notify_channels.extend(channels); + } + Token::EOF => { + break; + } + _ => { + return self.expected( + "`ON` or `LABELS` or `ANNOTATIONS` or `NOTIFY` keyword", + next_token, + ); + } + } + } + + let query = may_query.context(error::MissingClauseSnafu { name: ON })?; + let interval = may_interval.context(error::MissingClauseSnafu { name: ON })?; + let labels = may_labels.unwrap_or_default(); + let annotations = may_annotations.unwrap_or_default(); + + ensure!( + !notify_channels.is_empty(), + error::MissingClauseSnafu { name: NOTIFY } + ); + + let create_trigger = CreateTrigger { + trigger_name, + if_not_exists, + query, + interval, + labels, + annotations, + channels: notify_channels, + }; + let create_trigger = Statement::CreateTrigger(create_trigger); + Ok(create_trigger) + } + + /// The SQL format as follows: + /// + /// ```sql + /// ON () EVERY + /// ``` + /// + /// ## Parameters + /// + /// - `is_first_keyword_matched`: indicates whether the first keyword `ON` + /// has been matched. + fn parse_trigger_on(&mut self, is_first_keyword_matched: bool) -> Result<(Box, u64)> { + if !is_first_keyword_matched { + if let Token::Word(w) = self.parser.peek_token().token + && w.value.eq_ignore_ascii_case(ON) + { + self.parser.next_token(); + } else { + return self.expected("`ON` keyword", self.parser.peek_token()); + } + } + + let query = self.parser.parse_query().context(error::SyntaxSnafu)?; + + if let Token::Word(w) = self.parser.peek_token().token + && w.value.eq_ignore_ascii_case(EVERY) + { + self.parser.next_token(); + } else { + return self.expected("`EVERY` keyword", self.parser.peek_token()); + } + + let interval = self + .parse_interval()? + .try_into() + .context(error::NegativeIntervalSnafu)?; + + Ok((query, interval)) + } + + /// The SQL format as follows: + /// + /// ```sql + /// LABELS (key1 = 'value1', key2 = value2) + /// ``` + /// + /// ## Parameters + /// + /// - `is_first_keyword_matched`: indicates whether the first keyword `LABELS` + /// has been matched. + fn parse_trigger_labels(&mut self, is_first_keyword_matched: bool) -> Result { + if !is_first_keyword_matched { + if let Token::Word(w) = self.parser.peek_token().token + && w.value.eq_ignore_ascii_case(LABELS) + { + self.parser.next_token(); + } else { + return self.expected("`LABELS` keyword", self.parser.peek_token()); + } + } + if let Token::LParen = self.parser.peek_token().token { + self.parser.next_token(); + } else { + return self.expected("`(` after `LABELS`", self.parser.peek_token()); + } + let options = self + .parser + .parse_comma_separated0(Parser::parse_sql_option, Token::RParen) + .context(error::SyntaxSnafu)? + .into_iter() + .map(parse_option_string) + .collect::>>()?; + self.parser + .expect_token(&Token::RParen) + .context(error::SyntaxSnafu)?; + Ok(options.into()) + } + + /// The SQL format as follows: + /// + /// ```sql + /// ANNOTATIONS (key1 = 'value1', key2 = value2) + /// ``` + /// + /// ## Parameters + /// + /// - `is_first_keyword_matched`: indicates whether the first keyword + /// `ANNOTATIONS` has been matched. + fn parse_trigger_annotations(&mut self, is_first_keyword_matched: bool) -> Result { + if !is_first_keyword_matched { + if let Token::Word(w) = self.parser.peek_token().token + && w.value.eq_ignore_ascii_case(ANNOTATIONS) + { + self.parser.next_token(); + } else { + return self.expected("`ANNOTATIONS` keyword", self.parser.peek_token()); + } + } + if let Token::LParen = self.parser.peek_token().token { + self.parser.next_token(); + } else { + return self.expected("`(` after `LABELS`", self.parser.peek_token()); + } + let options = self + .parser + .parse_comma_separated0(Parser::parse_sql_option, Token::RParen) + .context(error::SyntaxSnafu)? + .into_iter() + .map(parse_option_string) + .collect::>>()?; + self.parser + .expect_token(&Token::RParen) + .context(error::SyntaxSnafu)?; + Ok(options.into()) + } + + /// The SQL format as follows: + /// + /// ```sql + // NOTIFY( + // WEBHOOK URL '' [WITH (=, ...)], + // WEBHOOK URL '' [WITH (=, ...)] + // ) + /// ``` + /// + /// ## Parameters + /// + /// - `is_first_keyword_matched`: indicates whether the first keyword `NOTIFY` + /// has been matched. + fn parse_trigger_notify( + &mut self, + is_first_keyword_matched: bool, + ) -> Result> { + if !is_first_keyword_matched { + if let Token::Word(w) = self.parser.peek_token().token + && w.value.eq_ignore_ascii_case(NOTIFY) + { + self.parser.next_token(); + } else { + return self.expected("`NOTIFY` keyword", self.parser.peek_token()); + } + } + + self.parser + .expect_token(&Token::LParen) + .context(error::SyntaxSnafu)?; + + let mut notify_channels = vec![]; + + loop { + let next_token = self.parser.peek_token(); + match next_token.token { + Token::Word(w) if w.value.eq_ignore_ascii_case(WEBHOOK) => { + self.parser.next_token(); + let notify_channel = self.parse_trigger_notify_webhook(true)?; + notify_channels.push(notify_channel); + } + Token::RParen => { + self.parser.next_token(); + break; + } + _ => { + return self.expected("`WEBHOOK` keyword", next_token); + } + } + + let next_token = self.parser.peek_token(); + if next_token.token == Token::RParen { + self.parser.next_token(); + break; + } else if next_token.token == Token::Comma { + self.parser.next_token(); + if self.parser.peek_token().token == Token::RParen { + self.parser.next_token(); + break; + } + } else { + return self.expected("`,` or `)`", next_token); + } + } + + ensure!( + !notify_channels.is_empty(), + error::MissingNotifyChannelSnafu + ); + + Ok(notify_channels) + } + + /// The SQL format as follows: + /// + /// ```sql + /// WEBHOOK URL '' [WITH (=, ...)] + /// ``` + /// + /// ## Parameters + /// + /// - `is_first_keyword_matched`: indicates whether the first keyword `WEBHOOK` + /// has been matched. + fn parse_trigger_notify_webhook( + &mut self, + is_first_keyword_matched: bool, + ) -> Result { + if !is_first_keyword_matched { + if let Token::Word(w) = self.parser.peek_token().token + && w.value.eq_ignore_ascii_case(WEBHOOK) + { + self.parser.next_token(); + } else { + return self.expected("`WEBHOOK` keyword", self.parser.peek_token()); + } + } + + let notify_ident = self.parser.parse_identifier().context(error::SyntaxSnafu)?; + + if let Token::Word(w) = self.parser.peek_token().token + && w.value.eq_ignore_ascii_case(URL) + { + self.parser.next_token(); + } else { + return self.expected("`URL` keyword", self.parser.peek_token()); + } + + let url = self.parser.parse_identifier().context(error::SyntaxSnafu)?; + + let options = self + .parser + .parse_options(Keyword::WITH) + .context(error::SyntaxSnafu)? + .into_iter() + .map(parse_option_string) + .collect::>>()?; + + for key in options.keys() { + ensure!( + validate_webhook_option(key), + error::InvalidTriggerWebhookOptionSnafu { + key: key.to_string() + } + ); + } + + let webhook = AlertManagerWebhook { + url, + options: options.into(), + }; + + Ok(NotifyChannel { + name: notify_ident, + channel_type: ChannelType::Webhook(webhook), + }) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::dialect::GreptimeDbDialect; + use crate::statements::create::trigger::ChannelType; + + #[test] + fn test_parse_create_trigger() { + // Normal, with all clauses. + let sql = create_trigger_sql_1(); + let mut ctx = ParserContext::new(&GreptimeDbDialect {}, sql).unwrap(); + let statement = ctx.parse_create_trigger().unwrap(); + let Statement::CreateTrigger(create_trigger) = statement else { + panic!("Expected CreateTrigger statement"); + }; + verify_create_trigger(&create_trigger); + + // Valid, but shuffle the order of clauses. + let sql = create_trigger_sql_2(); + let mut ctx = ParserContext::new(&GreptimeDbDialect {}, sql).unwrap(); + let statement = ctx.parse_create_trigger().unwrap(); + let Statement::CreateTrigger(create_trigger) = statement else { + panic!("Expected CreateTrigger statement"); + }; + verify_create_trigger(&create_trigger); + + // Invalid, since missing notify channels. + let sql = create_trigger_sql_3(); + let mut ctx = ParserContext::new(&GreptimeDbDialect {}, sql).unwrap(); + let result = ctx.parse_create_trigger(); + assert!(result.is_err()); + } + + fn create_trigger_sql_1() -> &'static str { + r#" +IF NOT EXISTS cpu_monitor + ON (SELECT host AS host_label, cpu, memory FROM machine_monitor WHERE cpu > 1) + EVERY '5 minute'::INTERVAL + LABELS (label_name=label_val) + ANNOTATIONS (annotation_name=annotation_val) + NOTIFY( + WEBHOOK alert_manager_1 URL 'http://127.0.0.1:9093' WITH (timeout='1m'), + WEBHOOK alert_manager_2 URL 'http://127.0.0.1:9094' WITH (timeout='2m') + )"# + } + + fn create_trigger_sql_2() -> &'static str { + r#" +IF NOT EXISTS cpu_monitor + NOTIFY( + WEBHOOK alert_manager_1 URL 'http://127.0.0.1:9093' WITH (timeout='1m'), + WEBHOOK alert_manager_2 URL 'http://127.0.0.1:9094' WITH (timeout='2m') + ) + LABELS (label_name=label_val) + ANNOTATIONS (annotation_name=annotation_val) + ON (SELECT host AS host_label, cpu, memory FROM machine_monitor WHERE cpu > 1) + EVERY '5 minute'::INTERVAL + "# + } + + fn create_trigger_sql_3() -> &'static str { + r#" +IF NOT EXISTS cpu_monitor + NOTIFY( + ) + ON (SELECT host AS host_label, cpu, memory FROM machine_monitor WHERE cpu > 1) + EVERY '5 minute'::INTERVAL + "# + } + + fn verify_create_trigger(create_trigger: &CreateTrigger) { + assert!(create_trigger.if_not_exists); + assert_eq!(create_trigger.trigger_name.to_string(), "cpu_monitor"); + assert_eq!( + create_trigger.query.to_string(), + "(SELECT host AS host_label, cpu, memory FROM machine_monitor WHERE cpu > 1)" + ); + assert_eq!(create_trigger.interval, 300); + assert_eq!(create_trigger.labels.len(), 1); + assert_eq!( + create_trigger.labels.get("label_name").unwrap(), + "label_val" + ); + assert_eq!(create_trigger.annotations.len(), 1); + assert_eq!( + create_trigger.annotations.get("annotation_name").unwrap(), + "annotation_val" + ); + assert_eq!(create_trigger.channels.len(), 2); + + let channel1 = &create_trigger.channels[0]; + let ChannelType::Webhook(webhook1) = &channel1.channel_type; + assert_eq!(webhook1.url.to_string(), "'http://127.0.0.1:9093'"); + assert_eq!(webhook1.options.len(), 1); + assert_eq!(webhook1.options.get("timeout").unwrap(), "1m"); + + let channel2 = &create_trigger.channels[1]; + assert_eq!(channel2.name.to_string(), "alert_manager_2"); + let ChannelType::Webhook(webhook2) = &channel2.channel_type; + assert_eq!(webhook2.url.to_string(), "'http://127.0.0.1:9094'"); + assert_eq!(webhook2.options.len(), 1); + assert_eq!(webhook2.options.get("timeout").unwrap(), "2m"); + } + + #[test] + fn test_parse_trigger_on() { + // Normal. + let sql = "ON (SELECT * FROM cpu_usage) EVERY '5 minute'::INTERVAL"; + let mut ctx = ParserContext::new(&GreptimeDbDialect {}, sql).unwrap(); + let (query, interval) = ctx.parse_trigger_on(false).unwrap(); + assert_eq!(query.to_string(), "(SELECT * FROM cpu_usage)"); + assert_eq!(interval, 300); + + // Invalid, since missing `ON` keyword. + let sql = "SELECT * FROM cpu_usage EVERY '5 minute'::INTERVAL"; + let mut ctx = ParserContext::new(&GreptimeDbDialect {}, sql).unwrap(); + assert!(ctx.parse_trigger_on(false).is_err()); + + // Invalid, since missing `EVERY` keyword. + let sql = "ON (SELECT * FROM cpu_usage) '5 minute'::INTERVAL"; + let mut ctx = ParserContext::new(&GreptimeDbDialect {}, sql).unwrap(); + assert!(ctx.parse_trigger_on(false).is_err()); + + // Invalid, since invalid query expression. + let sql = "ON (SELECT * cpu_usage) EVERY '5 minute'::INTERVAL"; + let mut ctx = ParserContext::new(&GreptimeDbDialect {}, sql).unwrap(); + assert!(ctx.parse_trigger_on(false).is_err()); + } + + #[test] + fn test_parse_trigger_labels() { + // Normal. + let sql = labels_sql_clause(); + let mut ctx = ParserContext::new(&GreptimeDbDialect {}, sql).unwrap(); + let labels = ctx.parse_trigger_labels(false).unwrap(); + verify_labels(&labels); + + // Empty labels clause. It's valid, since empty labels is allowed. + let sql = "LABELS ()"; + let mut ctx = ParserContext::new(&GreptimeDbDialect {}, sql).unwrap(); + let labels = ctx.parse_trigger_labels(false).unwrap(); + assert!(labels.is_empty()); + + // Invalid, since missing ')' + let sql = "LABELS ("; + let mut ctx = ParserContext::new(&GreptimeDbDialect {}, sql).unwrap(); + assert!(ctx.parse_trigger_labels(false).is_err()); + + // Invalid SQL clause, since the value of `name2` is not specified. + let sql = "LABELS (name1 = val1, name2)"; + let mut ctx = ParserContext::new(&GreptimeDbDialect {}, sql).unwrap(); + assert!(ctx.parse_trigger_labels(false).is_err()); + } + + fn labels_sql_clause() -> &'static str { + r#"LABELS (name1 = val1, name2 = 'val2')"# + } + + fn verify_labels(labels: &OptionMap) { + assert_eq!(labels.len(), 2); + assert_eq!(labels.get("name1").unwrap(), "val1"); + assert_eq!(labels.get("name2").unwrap(), "val2"); + } + + #[test] + fn test_parse_trigger_annotations() { + // Normal. + let sql = annotations_sql_clause(); + let mut ctx = ParserContext::new(&GreptimeDbDialect {}, sql).unwrap(); + let annotations = ctx.parse_trigger_annotations(false).unwrap(); + verify_annotations(&annotations); + + // Empty annotations clause. It's valid, since empty annotations is allowed. + let sql = "ANNOTATIONS ()"; + let mut ctx = ParserContext::new(&GreptimeDbDialect {}, sql).unwrap(); + let annotations = ctx.parse_trigger_annotations(false).unwrap(); + assert!(annotations.is_empty()); + + // Invalid, since missing ')' + let sql = "ANNOTATIONS ("; + let mut ctx = ParserContext::new(&GreptimeDbDialect {}, sql).unwrap(); + assert!(ctx.parse_trigger_annotations(false).is_err()); + + // Invalid, since the value of `name2` is not specified. + let sql = "ANNOTATIONS (name1 = val1, name2)"; + let mut ctx = ParserContext::new(&GreptimeDbDialect {}, sql).unwrap(); + assert!(ctx.parse_trigger_annotations(false).is_err()); + } + + fn annotations_sql_clause() -> &'static str { + r#"ANNOTATIONS (name1 = val1, name2 = 'val2')"# + } + + fn verify_annotations(annotations: &OptionMap) { + assert_eq!(annotations.len(), 2); + assert_eq!(annotations.get("name1").unwrap(), "val1"); + assert_eq!(annotations.get("name2").unwrap(), "val2"); + } + + #[test] + fn test_parse_trigger_notify() { + let sql = notify_sql_clause_1(); + let mut ctx = ParserContext::new(&GreptimeDbDialect {}, sql).unwrap(); + let notify_channels = ctx.parse_trigger_notify(false).unwrap(); + verify_notify_channels(¬ify_channels); + + let sql = notify_sql_clause_2(); + let mut ctx = ParserContext::new(&GreptimeDbDialect {}, sql).unwrap(); + let notify_channels = ctx.parse_trigger_notify(false).unwrap(); + verify_notify_channels(¬ify_channels); + + // Invalid, since must specify at least one notify channel. + let sql = "NOTIFY()"; + let mut ctx = ParserContext::new(&GreptimeDbDialect {}, sql).unwrap(); + let result = ctx.parse_trigger_notify(false); + assert!(result.is_err()); + } + + fn notify_sql_clause_1() -> &'static str { + r#"NOTIFY( + WEBHOOK notify_name1 URL 'http://example.com/1' WITH (timeout = '1m'), + WEBHOOK notify_name2 URL 'http://example.com/2' WITH (timeout = '2m') + )"# + } + + fn notify_sql_clause_2() -> &'static str { + r#"NOTIFY( + WEBHOOK 'notify_name1' URL 'http://example.com/1' WITH (timeout = '1m'), + WEBHOOK 'notify_name2' URL 'http://example.com/2' WITH (timeout = '2m'), + )"# + } + + fn verify_notify_channels(channels: &[NotifyChannel]) { + assert_eq!(channels.len(), 2); + let channel1 = &channels[0]; + assert_eq!(channel1.name.value, "notify_name1"); + let ChannelType::Webhook(webhook1) = &channel1.channel_type; + assert_eq!(webhook1.url.value, "http://example.com/1"); + assert_eq!(webhook1.options.get("timeout").unwrap(), "1m"); + let channel2 = &channels[1]; + assert_eq!(channel2.name.value, "notify_name2"); + let ChannelType::Webhook(webhook2) = &channel2.channel_type; + assert_eq!(webhook2.url.value, "http://example.com/2"); + assert_eq!(webhook2.options.get("timeout").unwrap(), "2m"); + } + + #[test] + fn test_parse_trigger_notify_webhook() { + // Normal. + let sql = webhook_sql_clause(); + let mut ctx = ParserContext::new(&GreptimeDbDialect {}, sql).unwrap(); + let channel = ctx.parse_trigger_notify_webhook(false).unwrap(); + verify_notify_webhook(channel); + + // Valid, but without `WITH` clause. + let sql = "WEBHOOK am URL 'http://example.com'"; + let mut ctx = ParserContext::new(&GreptimeDbDialect {}, sql).unwrap(); + let channel = ctx.parse_trigger_notify_webhook(false).unwrap(); + assert_eq!(channel.name.value, "am"); + let ChannelType::Webhook(webhook) = channel.channel_type; + assert_eq!(webhook.url.value, "http://example.com"); + assert!(webhook.options.is_empty()); + + // Valid, but with empty options. + let sql = "WEBHOOK am URL 'http://example.com' WITH ()"; + let mut ctx = ParserContext::new(&GreptimeDbDialect {}, sql).unwrap(); + let channel = ctx.parse_trigger_notify_webhook(false).unwrap(); + assert_eq!(channel.name.value, "am"); + let ChannelType::Webhook(webhook) = channel.channel_type; + assert_eq!(webhook.url.value, "http://example.com"); + assert!(webhook.options.is_empty()); + + // Invalid, since missing URL. + let sql = "WEBHOOK am"; + let mut ctx = ParserContext::new(&GreptimeDbDialect {}, sql).unwrap(); + assert!(ctx.parse_trigger_notify_webhook(false).is_err()); + + // Invalid, since the value of `timeout` is not specified. + let sql = "WEBHOOK am URL 'http://example.com' WITH (timeout)"; + let mut ctx = ParserContext::new(&GreptimeDbDialect {}, sql).unwrap(); + assert!(ctx.parse_trigger_notify_webhook(false).is_err()); + + // Invalid, missing ')' keyword. + let sql = "WEBHOOK am URL 'http://example.com' WITH ("; + let mut ctx = ParserContext::new(&GreptimeDbDialect {}, sql).unwrap(); + assert!(ctx.parse_trigger_notify_webhook(false).is_err()); + } + + fn webhook_sql_clause() -> &'static str { + "WEBHOOK am URL 'http://example.com' WITH (timeout = '1m')" + } + + fn verify_notify_webhook(channel: NotifyChannel) { + let ChannelType::Webhook(webhook) = channel.channel_type; + assert_eq!(channel.name.value, "am"); + assert_eq!(webhook.url.value, "http://example.com"); + let options = webhook.options; + assert_eq!(options.len(), 1); + assert_eq!(options.get("timeout").unwrap(), "1m"); + } + + #[test] + fn test_validate_notify_option() { + assert!(validate_webhook_option(TIMEOUT)); + assert!(!validate_webhook_option("invalid_option")); + } +} diff --git a/src/sql/src/statements/create.rs b/src/sql/src/statements/create.rs index 2784e59459c..d43985f6806 100644 --- a/src/sql/src/statements/create.rs +++ b/src/sql/src/statements/create.rs @@ -57,6 +57,9 @@ macro_rules! format_list_comma { }; } +#[cfg(feature = "enterprise")] +pub mod trigger; + fn format_table_constraint(constraints: &[TableConstraint]) -> String { constraints.iter().map(|c| format_indent!(c)).join(LINE_SEP) } diff --git a/src/sql/src/statements/create/trigger.rs b/src/sql/src/statements/create/trigger.rs new file mode 100644 index 00000000000..999b2a30afa --- /dev/null +++ b/src/sql/src/statements/create/trigger.rs @@ -0,0 +1,120 @@ +use std::fmt::{Display, Formatter}; + +use itertools::Itertools; +use serde::Serialize; +use sqlparser::ast::Query; +use sqlparser_derive::{Visit, VisitMut}; + +use crate::ast::{Ident, ObjectName}; +use crate::statements::create::{COMMA_SEP, INDENT, LINE_SEP}; +use crate::statements::OptionMap; + +#[derive(Debug, PartialEq, Eq, Clone, Visit, VisitMut, Serialize)] +pub struct CreateTrigger { + pub trigger_name: ObjectName, + pub if_not_exists: bool, + /// SQL statement executed periodically. + pub query: Box, + /// The interval of exec query. Unit is second. + pub interval: u64, + pub labels: OptionMap, + pub annotations: OptionMap, + pub channels: Vec, +} + +impl Display for CreateTrigger { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + write!(f, "CREATE TRIGGER ")?; + if self.if_not_exists { + write!(f, "IF NOT EXISTS ")?; + } + writeln!(f, "{}", self.trigger_name)?; + write!(f, "ON {} ", self.query)?; + writeln!(f, "EVERY {} SECONDS", self.interval)?; + + if !self.labels.is_empty() { + let labels = self.labels.kv_pairs(); + writeln!(f, "LABELS ({})", format_list_comma!(labels))?; + } + + if !self.annotations.is_empty() { + let annotations = self.annotations.kv_pairs(); + writeln!(f, "ANNOTATIONS ({})", format_list_comma!(annotations))?; + } + + if !self.channels.is_empty() { + writeln!(f, "NOTIFY(")?; + write!(f, "{}", format_list_indent!(self.channels))?; + write!(f, ")")?; + } + + Ok(()) + } +} + +#[derive(Debug, PartialEq, Eq, Clone, Visit, VisitMut, Serialize)] +pub struct NotifyChannel { + pub name: Ident, + pub channel_type: ChannelType, +} + +impl Display for NotifyChannel { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + match &self.channel_type { + ChannelType::Webhook(webhook) => { + write!(f, "WEBHOOK {} URL {}", self.name, webhook.url)?; + if !webhook.options.is_empty() { + let options = webhook.options.kv_pairs(); + write!(f, " WITH ({})", format_list_comma!(options))?; + } + } + } + Ok(()) + } +} + +#[derive(Debug, PartialEq, Eq, Clone, Visit, VisitMut, Serialize)] +pub enum ChannelType { + /// Alert manager webhook options. + Webhook(AlertManagerWebhook), +} + +#[derive(Debug, PartialEq, Eq, Clone, Visit, VisitMut, Serialize)] +pub struct AlertManagerWebhook { + pub url: Ident, + pub options: OptionMap, +} + +#[cfg(test)] +mod tests { + use crate::dialect::GreptimeDbDialect; + use crate::parser::{ParseOptions, ParserContext}; + use crate::statements::statement::Statement; + + #[test] + fn test_display_create_trigger() { + let sql = r#"CREATE TRIGGER IF NOT EXISTS cpu_monitor +ON (SELECT host AS host_label, cpu, memory FROM machine_monitor WHERE cpu > 2) EVERY '5 minute'::INTERVAL +LABELS (label_name=label_val) +ANNOTATIONS (annotation_name=annotation_val) +NOTIFY +(WEBHOOK alert_manager URL 'http://127.0.0.1:9093' WITH (timeout='1m'))"#; + + let result = + ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default()) + .unwrap(); + assert_eq!(1, result.len()); + + let Statement::CreateTrigger(trigger) = &result[0] else { + panic!("Expected CreateTrigger statement"); + }; + let formatted = format!("{}", trigger); + let expected = r#"CREATE TRIGGER IF NOT EXISTS cpu_monitor +ON (SELECT host AS host_label, cpu, memory FROM machine_monitor WHERE cpu > 2) EVERY 300 SECONDS +LABELS (label_name = 'label_val') +ANNOTATIONS (annotation_name = 'annotation_val') +NOTIFY( + WEBHOOK alert_manager URL 'http://127.0.0.1:9093' WITH (timeout = '1m'))"#; + assert_eq!(expected, formatted); + } +} diff --git a/src/sql/src/statements/statement.rs b/src/sql/src/statements/statement.rs index 6bae0419d68..bc2491db676 100644 --- a/src/sql/src/statements/statement.rs +++ b/src/sql/src/statements/statement.rs @@ -62,6 +62,9 @@ pub enum Statement { CreateFlow(CreateFlow), // CREATE VIEW ... AS CreateView(CreateView), + // CREATE TRIGGER + #[cfg(feature = "enterprise")] + CreateTrigger(crate::statements::create::trigger::CreateTrigger), // DROP TABLE DropTable(DropTable), // DROP DATABASE @@ -144,6 +147,8 @@ impl Display for Statement { Statement::CreateExternalTable(s) => s.fmt(f), Statement::CreateTableLike(s) => s.fmt(f), Statement::CreateFlow(s) => s.fmt(f), + #[cfg(feature = "enterprise")] + Statement::CreateTrigger(s) => s.fmt(f), Statement::DropFlow(s) => s.fmt(f), Statement::DropTable(s) => s.fmt(f), Statement::DropDatabase(s) => s.fmt(f),