1mod json;
16#[cfg(feature = "enterprise")]
17pub mod trigger;
18
19use std::collections::HashMap;
20
21use arrow_buffer::IntervalMonthDayNano;
22use common_catalog::consts::default_engine;
23use datafusion_common::ScalarValue;
24use datatypes::arrow::datatypes::{DataType as ArrowDataType, IntervalUnit};
25use datatypes::data_type::ConcreteDataType;
26use itertools::Itertools;
27pub(crate) use json::parse_json2_type_and_options;
28pub use json::parse_json2_type_hint_path;
29use snafu::{OptionExt, ResultExt, ensure};
30use sqlparser::ast::{
31 ColumnOption, ColumnOptionDef, DataType, Expr, KeyOrIndexDisplay, NullsDistinctOption,
32 PrimaryKeyConstraint, UniqueConstraint,
33};
34use sqlparser::dialect::keywords::Keyword;
35use sqlparser::keywords::ALL_KEYWORDS;
36use sqlparser::parser::IsOptional::Mandatory;
37use sqlparser::parser::{Parser, ParserError};
38use sqlparser::tokenizer::{Token, TokenWithSpan, Word};
39use table::requests::{validate_database_option, validate_database_option_value};
40
41use crate::ast::{ColumnDef, Ident, ObjectNamePartExt};
42use crate::error::{
43 self, InvalidColumnOptionSnafu, InvalidDatabaseOptionSnafu, InvalidDatabaseOptionValueSnafu,
44 InvalidFlowQuerySnafu, InvalidIntervalSnafu, InvalidSqlSnafu, InvalidTimeIndexSnafu,
45 MissingTimeIndexSnafu, Result, SyntaxSnafu, UnexpectedSnafu, UnsupportedSnafu,
46};
47use crate::parser::{FLOW, ParserContext};
48use crate::parsers::tql_parser;
49use crate::parsers::utils::{
50 self, parse_with_options, validate_column_fulltext_create_option,
51 validate_column_skipping_index_create_option,
52};
53use crate::statements::create::{
54 Column, ColumnExtensions, CreateDatabase, CreateExternalTable, CreateFlow, CreateTable,
55 CreateTableLike, CreateView, Partitions, SqlOrTql, TableConstraint, VECTOR_OPT_DIM,
56};
57use crate::statements::statement::Statement;
58use crate::statements::transform::type_alias::get_data_type_by_alias_name;
59use crate::statements::{OptionMap, sql_data_type_to_concrete_data_type};
60use crate::util::{OptionValue, location_to_index, parse_option_string};
61
62pub const ENGINE: &str = "ENGINE";
63pub const MAXVALUE: &str = "MAXVALUE";
64pub const SINK: &str = "SINK";
65pub const EXPIRE: &str = "EXPIRE";
66pub const AFTER: &str = "AFTER";
67pub const INVERTED: &str = "INVERTED";
68pub const SKIPPING: &str = "SKIPPING";
69
70pub type RawIntervalExpr = String;
71
72fn flow_option_map(options: HashMap<String, OptionValue>) -> OptionMap {
76 let mut flow_options = OptionMap::default();
77 for (key, value) in options {
78 flow_options.insert_options(&key, value);
79 }
80 flow_options
81}
82
83impl<'a> ParserContext<'a> {
85 pub(crate) fn parse_create(&mut self) -> Result<Statement> {
86 match self.parser.peek_token().token {
87 Token::Word(w) => match w.keyword {
88 Keyword::TABLE => self.parse_create_table(),
89
90 Keyword::SCHEMA | Keyword::DATABASE => self.parse_create_database(),
91
92 Keyword::EXTERNAL => self.parse_create_external_table(),
93
94 Keyword::OR => {
95 let _ = self.parser.next_token();
96 self.parser
97 .expect_keyword(Keyword::REPLACE)
98 .context(SyntaxSnafu)?;
99 match self.parser.next_token().token {
100 Token::Word(w) => match w.keyword {
101 Keyword::VIEW => self.parse_create_view(true),
102 Keyword::NoKeyword => {
103 let uppercase = w.value.to_uppercase();
104 match uppercase.as_str() {
105 FLOW => self.parse_create_flow(true),
106 _ => self.unsupported(w.to_string()),
107 }
108 }
109 _ => self.unsupported(w.to_string()),
110 },
111 _ => self.unsupported(w.to_string()),
112 }
113 }
114
115 Keyword::VIEW => {
116 let _ = self.parser.next_token();
117 self.parse_create_view(false)
118 }
119
120 #[cfg(feature = "enterprise")]
121 Keyword::TRIGGER => {
122 let _ = self.parser.next_token();
123 self.parse_create_trigger()
124 }
125
126 Keyword::NoKeyword => {
127 let _ = self.parser.next_token();
128 let uppercase = w.value.to_uppercase();
129 match uppercase.as_str() {
130 FLOW => self.parse_create_flow(false),
131 _ => self.unsupported(w.to_string()),
132 }
133 }
134 _ => self.unsupported(w.to_string()),
135 },
136 unexpected => self.unsupported(unexpected.to_string()),
137 }
138 }
139
140 fn parse_create_view(&mut self, or_replace: bool) -> Result<Statement> {
142 let if_not_exists = self.parse_if_not_exist()?;
143 let view_name = self.intern_parse_table_name()?;
144
145 let columns = self.parse_view_columns()?;
146
147 self.parser
148 .expect_keyword(Keyword::AS)
149 .context(SyntaxSnafu)?;
150
151 let query = self.parse_query()?;
152
153 Ok(Statement::CreateView(CreateView {
154 name: view_name,
155 columns,
156 or_replace,
157 query: Box::new(query),
158 if_not_exists,
159 }))
160 }
161
162 fn parse_view_columns(&mut self) -> Result<Vec<Ident>> {
163 let mut columns = vec![];
164 if !self.parser.consume_token(&Token::LParen) || self.parser.consume_token(&Token::RParen) {
165 return Ok(columns);
166 }
167
168 loop {
169 let name = self.parse_column_name().context(SyntaxSnafu)?;
170
171 columns.push(name);
172
173 let comma = self.parser.consume_token(&Token::Comma);
174 if self.parser.consume_token(&Token::RParen) {
175 break;
177 } else if !comma {
178 return self.expected("',' or ')' after column name", self.parser.peek_token());
179 }
180 }
181
182 Ok(columns)
183 }
184
185 fn parse_create_external_table(&mut self) -> Result<Statement> {
186 let _ = self.parser.next_token();
187 self.parser
188 .expect_keyword(Keyword::TABLE)
189 .context(SyntaxSnafu)?;
190 let if_not_exists = self.parse_if_not_exist()?;
191 let table_name = self.intern_parse_table_name()?;
192 let (columns, constraints) = self.parse_columns()?;
193 if !columns.is_empty() {
194 validate_time_index(&columns, &constraints)?;
195 }
196
197 let engine = self.parse_table_engine(common_catalog::consts::FILE_ENGINE)?;
198 let options = self.parse_create_table_options()?;
199 Ok(Statement::CreateExternalTable(CreateExternalTable {
200 name: table_name,
201 columns,
202 constraints,
203 options,
204 if_not_exists,
205 engine,
206 }))
207 }
208
209 fn parse_create_database(&mut self) -> Result<Statement> {
210 let _ = self.parser.next_token();
211 let if_not_exists = self.parse_if_not_exist()?;
212 let database_name = self.parse_object_name().context(error::UnexpectedSnafu {
213 expected: "a database name",
214 actual: self.peek_token_as_string(),
215 })?;
216 let database_name = Self::canonicalize_object_name(database_name)?;
217
218 let options = self
219 .parser
220 .parse_options(Keyword::WITH)
221 .context(SyntaxSnafu)?
222 .into_iter()
223 .map(parse_option_string)
224 .collect::<Result<HashMap<String, OptionValue>>>()?;
225
226 for (key, option_value) in &options {
227 ensure!(
228 validate_database_option(key),
229 InvalidDatabaseOptionSnafu { key: key.clone() }
230 );
231 let option_value_str = option_value.as_string();
232 validate_database_option_value(key, option_value_str).map_err(|reason| {
233 InvalidDatabaseOptionValueSnafu {
234 key: key.clone(),
235 value: option_value_str
236 .map(str::to_owned)
237 .unwrap_or_else(|| option_value.to_string()),
238 reason: reason.to_string(),
239 }
240 .build()
241 })?;
242 }
243 if let Some(append_mode) = options.get("append_mode").and_then(|x| x.as_string())
244 && append_mode == "true"
245 && options.contains_key("merge_mode")
246 {
247 return InvalidDatabaseOptionSnafu {
248 key: "merge_mode".to_string(),
249 }
250 .fail();
251 }
252
253 Ok(Statement::CreateDatabase(CreateDatabase {
254 name: database_name,
255 if_not_exists,
256 options: OptionMap::new(options),
257 }))
258 }
259
260 fn parse_create_table(&mut self) -> Result<Statement> {
261 let _ = self.parser.next_token();
262
263 let if_not_exists = self.parse_if_not_exist()?;
264
265 let table_name = self.intern_parse_table_name()?;
266
267 if self.parser.parse_keyword(Keyword::LIKE) {
268 let source_name = self.intern_parse_table_name()?;
269
270 return Ok(Statement::CreateTableLike(CreateTableLike {
271 table_name,
272 source_name,
273 }));
274 }
275
276 let (columns, constraints) = self.parse_columns()?;
277 validate_time_index(&columns, &constraints)?;
278
279 let partitions = self.parse_partitions()?;
280 if let Some(partitions) = &partitions {
281 validate_partitions(&columns, partitions)?;
282 }
283
284 let engine = self.parse_table_engine(default_engine())?;
285 let options = self.parse_create_table_options()?;
286 let create_table = CreateTable {
287 if_not_exists,
288 name: table_name,
289 columns,
290 engine,
291 constraints,
292 options,
293 table_id: 0, partitions,
295 };
296
297 Ok(Statement::CreateTable(create_table))
298 }
299
300 fn parse_create_flow(&mut self, or_replace: bool) -> Result<Statement> {
302 let if_not_exists = self.parse_if_not_exist()?;
303
304 let flow_name = self.intern_parse_table_name()?;
305
306 if let Token::Word(word) = self.parser.peek_token().token
308 && word.value.eq_ignore_ascii_case(SINK)
309 {
310 self.parser.next_token();
311 } else {
312 Err(ParserError::ParserError(
313 "Expect `SINK` keyword".to_string(),
314 ))
315 .context(SyntaxSnafu)?
316 }
317 self.parser
318 .expect_keyword(Keyword::TO)
319 .context(SyntaxSnafu)?;
320
321 let output_table_name = self.intern_parse_table_name()?;
322
323 let expire_after = if let Token::Word(w1) = &self.parser.peek_token().token
324 && w1.value.eq_ignore_ascii_case(EXPIRE)
325 {
326 self.parser.next_token();
327 if let Token::Word(w2) = &self.parser.peek_token().token
328 && w2.value.eq_ignore_ascii_case(AFTER)
329 {
330 self.parser.next_token();
331 Some(self.parse_interval_no_month("EXPIRE AFTER")?)
332 } else {
333 None
334 }
335 } else {
336 None
337 };
338
339 let (eval_interval, eval_interval_has_fractional_secs) =
340 if self.consume_eval_pair("INTERVAL") {
341 let (secs, has_fractional) =
342 self.parse_interval_no_month_whole_secs("EVAL INTERVAL")?;
343 (Some(secs), has_fractional)
344 } else {
345 (None, false)
346 };
347
348 let eval_offset = if self.consume_eval_pair("OFFSET") {
352 let Some(eval_interval) = eval_interval else {
353 return InvalidIntervalSnafu {
354 reason: "EVAL OFFSET requires EVAL INTERVAL to be specified".to_string(),
355 }
356 .fail();
357 };
358 let (offset_secs, has_fractional) =
359 self.parse_interval_no_month_whole_secs("EVAL OFFSET")?;
360 if has_fractional {
361 return InvalidIntervalSnafu {
362 reason: "EVAL OFFSET must be a whole number of seconds".to_string(),
363 }
364 .fail();
365 }
366 if eval_interval_has_fractional_secs {
367 return InvalidIntervalSnafu {
368 reason: "EVAL INTERVAL must be a whole number of seconds when EVAL OFFSET is specified"
369 .to_string(),
370 }
371 .fail();
372 }
373 if !(0..eval_interval).contains(&offset_secs) {
374 return InvalidIntervalSnafu {
375 reason: format!(
376 "EVAL OFFSET must be in range [0, EVAL INTERVAL), got {offset_secs} seconds with EVAL INTERVAL {eval_interval} seconds"
377 ),
378 }
379 .fail();
380 }
381 if offset_secs == 0 {
384 None
385 } else {
386 Some(offset_secs)
387 }
388 } else {
389 None
390 };
391
392 let comment = if self.parser.parse_keyword(Keyword::COMMENT) {
393 match self.parser.next_token() {
394 TokenWithSpan {
395 token: Token::SingleQuotedString(value, ..),
396 ..
397 } => Some(value),
398 unexpected => {
399 return self
400 .parser
401 .expected("string", unexpected)
402 .context(SyntaxSnafu);
403 }
404 }
405 } else {
406 None
407 };
408
409 let flow_options = self
410 .parser
411 .parse_options(Keyword::WITH)
412 .context(SyntaxSnafu)?
413 .into_iter()
414 .map(parse_option_string)
415 .collect::<Result<HashMap<String, OptionValue>>>()?;
416
417 self.parser
418 .expect_keyword(Keyword::AS)
419 .context(SyntaxSnafu)?;
420
421 let query = Box::new(self.parse_flow_sql_or_tql(true)?);
422
423 Ok(Statement::CreateFlow(CreateFlow {
424 flow_name,
425 sink_table_name: output_table_name,
426 or_replace,
427 if_not_exists,
428 expire_after,
429 eval_interval,
430 eval_offset,
431 comment,
432 flow_options: flow_option_map(flow_options),
433 query,
434 }))
435 }
436
437 fn parse_flow_sql_or_tql(&mut self, require_now_expr: bool) -> Result<SqlOrTql> {
438 let start_loc = self.parser.peek_token().span.start;
439 let start_index = location_to_index(self.sql, &start_loc);
440
441 let starts_with_with = matches!(
442 self.parser.peek_token().token,
443 Token::Word(w) if w.keyword == Keyword::WITH
444 );
445
446 let query = match self.parser.peek_token().token {
448 Token::Word(w) => match w.keyword {
449 Keyword::SELECT => self.parse_query(),
450 Keyword::WITH => self.parse_with_tql_with_now(require_now_expr),
451 Keyword::NoKeyword
452 if w.quote_style.is_none() && w.value.to_uppercase() == tql_parser::TQL =>
453 {
454 self.parse_tql(require_now_expr)
455 }
456
457 _ => self.unsupported(self.peek_token_as_string()),
458 },
459 _ => self.unsupported(self.peek_token_as_string()),
460 }?;
461
462 if starts_with_with {
463 let Statement::Query(query) = &query else {
464 return InvalidFlowQuerySnafu {
465 reason: "Expect a query after WITH".to_string(),
466 }
467 .fail();
468 };
469
470 if utils::has_tql_cte(query) && !utils::is_simple_tql_cte_query(query) {
471 return InvalidFlowQuerySnafu {
472 reason: "WITH is only supported for the simplest TQL CTE in CREATE FLOW"
473 .to_string(),
474 }
475 .fail();
476 }
477 }
478
479 let end_token = self.parser.peek_token();
480
481 let raw_query = if end_token == Token::EOF {
482 &self.sql[start_index..]
483 } else {
484 let end_loc = end_token.span.end;
485 let end_index = location_to_index(self.sql, &end_loc);
486 &self.sql[start_index..end_index.min(self.sql.len())]
487 };
488 let raw_query = raw_query.trim_end_matches(";");
489
490 let query = SqlOrTql::try_from_statement(query, raw_query)?;
491 Ok(query)
492 }
493
494 fn parse_interval_no_month(&mut self, context: &str) -> Result<i64> {
496 Ok(self.parse_interval_no_month_whole_secs(context)?.0)
497 }
498
499 fn parse_interval_no_month_whole_secs(&mut self, context: &str) -> Result<(i64, bool)> {
505 let interval = self.parse_interval_month_day_nano()?.0;
506 if interval.months != 0 {
507 return InvalidIntervalSnafu {
508 reason: format!("Interval with months is not allowed in {context}"),
509 }
510 .fail();
511 }
512 let has_fractional_secs = interval.nanoseconds % 1_000_000_000 != 0;
513 let whole_secs = interval.nanoseconds / 1_000_000_000 + interval.days as i64 * 60 * 60 * 24;
514 Ok((whole_secs, has_fractional_secs))
515 }
516
517 fn consume_eval_pair(&mut self, second: &str) -> bool {
523 let matches = matches!(
524 (
525 &self.parser.peek_token().token,
526 &self.parser.peek_nth_token(1).token,
527 ),
528 (Token::Word(w1), Token::Word(w2))
529 if w1.value.eq_ignore_ascii_case("EVAL") && w2.value.eq_ignore_ascii_case(second)
530 );
531 if matches {
532 self.parser.next_token();
533 self.parser.next_token();
534 }
535 matches
536 }
537
538 fn parse_interval_month_day_nano(&mut self) -> Result<(IntervalMonthDayNano, RawIntervalExpr)> {
540 let interval_expr = self.parser.parse_expr().context(error::SyntaxSnafu)?;
541 let raw_interval_expr = interval_expr.to_string();
542 let interval = utils::parser_expr_to_scalar_value_literal(interval_expr.clone(), false)?
543 .cast_to(&ArrowDataType::Interval(IntervalUnit::MonthDayNano))
544 .ok()
545 .with_context(|| InvalidIntervalSnafu {
546 reason: format!("cannot cast {} to interval type", interval_expr),
547 })?;
548 if let ScalarValue::IntervalMonthDayNano(Some(interval)) = interval {
549 Ok((interval, raw_interval_expr))
550 } else {
551 unreachable!()
552 }
553 }
554
555 fn parse_if_not_exist(&mut self) -> Result<bool> {
556 match self.parser.peek_token().token {
557 Token::Word(w) if Keyword::IF != w.keyword => return Ok(false),
558 _ => {}
559 }
560
561 if self.parser.parse_keywords(&[Keyword::IF, Keyword::NOT]) {
562 return self
563 .parser
564 .expect_keyword(Keyword::EXISTS)
565 .map(|_| true)
566 .context(UnexpectedSnafu {
567 expected: "EXISTS",
568 actual: self.peek_token_as_string(),
569 });
570 }
571
572 if self.parser.parse_keywords(&[Keyword::IF, Keyword::EXISTS]) {
573 return UnsupportedSnafu { keyword: "EXISTS" }.fail();
574 }
575
576 Ok(false)
577 }
578
579 fn parse_create_table_options(&mut self) -> Result<OptionMap> {
580 parse_with_options(&mut self.parser)
581 }
582
583 fn parse_partitions(&mut self) -> Result<Option<Partitions>> {
585 if !self.parser.parse_keyword(Keyword::PARTITION) {
586 return Ok(None);
587 }
588
589 self.parse_partition_on_columns().map(Some)
590 }
591
592 pub(crate) fn parse_partition_on_columns(&mut self) -> Result<Partitions> {
594 self.parser
595 .expect_keywords(&[Keyword::ON, Keyword::COLUMNS])
596 .context(error::UnexpectedSnafu {
597 expected: "ON, COLUMNS",
598 actual: self.peek_token_as_string(),
599 })?;
600
601 let raw_column_list = self
602 .parser
603 .parse_parenthesized_column_list(Mandatory, false)
604 .context(error::SyntaxSnafu)?;
605 let column_list = raw_column_list
606 .into_iter()
607 .map(Self::canonicalize_identifier)
608 .collect();
609
610 let exprs = self.parse_comma_separated(Self::parse_partition_entry)?;
611
612 Ok(Partitions { column_list, exprs })
613 }
614
615 fn parse_partition_entry(&mut self) -> Result<Expr> {
616 self.parser.parse_expr().context(error::SyntaxSnafu)
617 }
618
619 fn parse_comma_separated<T, F>(&mut self, mut f: F) -> Result<Vec<T>>
621 where
622 F: FnMut(&mut ParserContext<'a>) -> Result<T>,
623 {
624 self.parser
625 .expect_token(&Token::LParen)
626 .context(error::UnexpectedSnafu {
627 expected: "(",
628 actual: self.peek_token_as_string(),
629 })?;
630
631 let mut values = vec![];
632 while self.parser.peek_token() != Token::RParen {
633 values.push(f(self)?);
634 if !self.parser.consume_token(&Token::Comma) {
635 break;
636 }
637 }
638
639 self.parser
640 .expect_token(&Token::RParen)
641 .context(error::UnexpectedSnafu {
642 expected: ")",
643 actual: self.peek_token_as_string(),
644 })?;
645
646 Ok(values)
647 }
648
649 fn parse_columns(&mut self) -> Result<(Vec<Column>, Vec<TableConstraint>)> {
651 let mut columns = vec![];
652 let mut constraints = vec![];
653 if !self.parser.consume_token(&Token::LParen) || self.parser.consume_token(&Token::RParen) {
654 return Ok((columns, constraints));
655 }
656
657 loop {
658 if let Some(constraint) = self.parse_optional_table_constraint()? {
659 constraints.push(constraint);
660 } else if let Token::Word(_) = self.parser.peek_token().token {
661 self.parse_column(&mut columns, &mut constraints)?;
662 } else {
663 return self.expected(
664 "column name or constraint definition",
665 self.parser.peek_token(),
666 );
667 }
668 let comma = self.parser.consume_token(&Token::Comma);
669 if self.parser.consume_token(&Token::RParen) {
670 break;
672 } else if !comma {
673 return self.expected(
674 "',' or ')' after column definition",
675 self.parser.peek_token(),
676 );
677 }
678 }
679
680 Ok((columns, constraints))
681 }
682
683 fn parse_column(
684 &mut self,
685 columns: &mut Vec<Column>,
686 constraints: &mut Vec<TableConstraint>,
687 ) -> Result<()> {
688 let mut column = self.parse_column_def()?;
689
690 let mut time_index_opt_idx = None;
691 for (index, opt) in column.options().iter().enumerate() {
692 if let ColumnOption::DialectSpecific(tokens) = &opt.option
693 && matches!(
694 &tokens[..],
695 [
696 Token::Word(Word {
697 keyword: Keyword::TIME,
698 ..
699 }),
700 Token::Word(Word {
701 keyword: Keyword::INDEX,
702 ..
703 })
704 ]
705 )
706 {
707 ensure!(
708 time_index_opt_idx.is_none(),
709 InvalidColumnOptionSnafu {
710 name: column.name().to_string(),
711 msg: "duplicated time index",
712 }
713 );
714 time_index_opt_idx = Some(index);
715
716 let constraint = TableConstraint::TimeIndex {
717 column: Ident::new(column.name().value.clone()),
718 };
719 constraints.push(constraint);
720 }
721 }
722
723 if let Some(index) = time_index_opt_idx {
724 ensure!(
725 !column.options().contains(&ColumnOptionDef {
726 option: ColumnOption::Null,
727 name: None,
728 }),
729 InvalidColumnOptionSnafu {
730 name: column.name().to_string(),
731 msg: "time index column can't be null",
732 }
733 );
734
735 let data_type = get_unalias_type(column.data_type());
737 ensure!(
738 matches!(data_type, DataType::Timestamp(_, _)),
739 InvalidColumnOptionSnafu {
740 name: column.name().to_string(),
741 msg: "time index column data type should be timestamp",
742 }
743 );
744
745 let not_null_opt = ColumnOptionDef {
746 option: ColumnOption::NotNull,
747 name: None,
748 };
749
750 if !column.options().contains(¬_null_opt) {
751 column.mut_options().push(not_null_opt);
752 }
753
754 let _ = column.mut_options().remove(index);
755 }
756
757 columns.push(column);
758
759 Ok(())
760 }
761
762 fn parse_column_name(&mut self) -> std::result::Result<Ident, ParserError> {
764 let name = self.parser.parse_identifier()?;
765 if name.quote_style.is_none() &&
766 ALL_KEYWORDS.binary_search(&name.value.to_uppercase().as_str()).is_ok()
768 {
769 return Err(ParserError::ParserError(format!(
770 "Cannot use keyword '{}' as column name. Hint: add quotes to the name.",
771 &name.value
772 )));
773 }
774
775 Ok(name)
776 }
777
778 pub fn parse_column_def(&mut self) -> Result<Column> {
779 let name = self.parse_column_name().context(SyntaxSnafu)?;
780 let parser = &mut self.parser;
781
782 ensure!(
783 !(name.quote_style.is_none() &&
784 ALL_KEYWORDS.binary_search(&name.value.to_uppercase().as_str()).is_ok()),
786 InvalidSqlSnafu {
787 msg: format!(
788 "Cannot use keyword '{}' as column name. Hint: add quotes to the name.",
789 &name.value
790 ),
791 }
792 );
793
794 let mut extensions = ColumnExtensions::default();
795
796 let data_type =
797 if let Some((data_type, options)) = json::parse_json2_type_and_options(parser)? {
798 extensions.json2_options = options;
799 data_type
800 } else {
801 parser.parse_data_type().context(SyntaxSnafu)?
802 };
803
804 let mut options = vec![];
805 loop {
806 if parser.parse_keyword(Keyword::CONSTRAINT) {
807 let name = Some(parser.parse_identifier().context(SyntaxSnafu)?);
808 if let Some(option) = Self::parse_optional_column_option(parser)? {
809 options.push(ColumnOptionDef { name, option });
810 } else {
811 return parser
812 .expected(
813 "constraint details after CONSTRAINT <name>",
814 parser.peek_token(),
815 )
816 .context(SyntaxSnafu);
817 }
818 } else if let Some(option) = Self::parse_optional_column_option(parser)? {
819 options.push(ColumnOptionDef { name: None, option });
820 } else if !Self::parse_column_extensions(parser, &name, &data_type, &mut extensions)? {
821 break;
822 };
823 }
824
825 Ok(Column {
826 column_def: ColumnDef {
827 name: Self::canonicalize_identifier(name),
828 data_type,
829 options,
830 },
831 extensions,
832 })
833 }
834
835 fn parse_optional_column_option(parser: &mut Parser<'_>) -> Result<Option<ColumnOption>> {
836 if parser.parse_keywords(&[Keyword::CHARACTER, Keyword::SET]) {
837 Ok(Some(ColumnOption::CharacterSet(
838 parser.parse_object_name(false).context(SyntaxSnafu)?,
839 )))
840 } else if parser.parse_keywords(&[Keyword::NOT, Keyword::NULL]) {
841 Ok(Some(ColumnOption::NotNull))
842 } else if parser.parse_keywords(&[Keyword::COMMENT]) {
843 match parser.next_token() {
844 TokenWithSpan {
845 token: Token::SingleQuotedString(value, ..),
846 ..
847 } => Ok(Some(ColumnOption::Comment(value))),
848 unexpected => parser.expected("string", unexpected).context(SyntaxSnafu),
849 }
850 } else if parser.parse_keyword(Keyword::NULL) {
851 Ok(Some(ColumnOption::Null))
852 } else if parser.parse_keyword(Keyword::DEFAULT) {
853 Ok(Some(ColumnOption::Default(
854 parser.parse_expr().context(SyntaxSnafu)?,
855 )))
856 } else if parser.parse_keywords(&[Keyword::PRIMARY, Keyword::KEY]) {
857 Ok(Some(ColumnOption::PrimaryKey(PrimaryKeyConstraint {
858 name: None,
859 index_name: None,
860 index_type: None,
861 columns: vec![],
862 index_options: vec![],
863 characteristics: None,
864 })))
865 } else if parser.parse_keyword(Keyword::UNIQUE) {
866 Ok(Some(ColumnOption::Unique(UniqueConstraint {
867 name: None,
868 index_name: None,
869 index_type_display: KeyOrIndexDisplay::None,
870 index_type: None,
871 columns: vec![],
872 index_options: vec![],
873 characteristics: None,
874 nulls_distinct: NullsDistinctOption::None,
875 })))
876 } else if parser.parse_keywords(&[Keyword::TIME, Keyword::INDEX]) {
877 Ok(Some(ColumnOption::DialectSpecific(vec![
879 Token::Word(Word {
880 value: "TIME".to_string(),
881 quote_style: None,
882 keyword: Keyword::TIME,
883 }),
884 Token::Word(Word {
885 value: "INDEX".to_string(),
886 quote_style: None,
887 keyword: Keyword::INDEX,
888 }),
889 ])))
890 } else {
891 Ok(None)
892 }
893 }
894
895 fn parse_column_extensions(
901 parser: &mut Parser<'_>,
902 column_name: &Ident,
903 column_type: &DataType,
904 column_extensions: &mut ColumnExtensions,
905 ) -> Result<bool> {
906 if let DataType::Custom(name, tokens) = column_type
907 && name.0.len() == 1
908 && &name.0[0].to_string_unquoted().to_uppercase() == "VECTOR"
909 {
910 ensure!(
911 tokens.len() == 1,
912 InvalidColumnOptionSnafu {
913 name: column_name.to_string(),
914 msg: "VECTOR type should have dimension",
915 }
916 );
917
918 let dimension =
919 tokens[0]
920 .parse::<u32>()
921 .ok()
922 .with_context(|| InvalidColumnOptionSnafu {
923 name: column_name.to_string(),
924 msg: "dimension should be a positive integer",
925 })?;
926
927 let options = OptionMap::from([(VECTOR_OPT_DIM.to_string(), dimension.to_string())]);
928 column_extensions.vector_options = Some(options);
929 }
930
931 let mut is_index_declared = false;
933
934 if let Token::Word(word) = parser.peek_token().token
936 && word.value.eq_ignore_ascii_case(SKIPPING)
937 {
938 parser.next_token();
939 ensure!(
941 parser.parse_keyword(Keyword::INDEX),
942 InvalidColumnOptionSnafu {
943 name: column_name.to_string(),
944 msg: "expect INDEX after SKIPPING keyword",
945 }
946 );
947 ensure!(
948 column_extensions.skipping_index_options.is_none(),
949 InvalidColumnOptionSnafu {
950 name: column_name.to_string(),
951 msg: "duplicated SKIPPING index option",
952 }
953 );
954
955 let options = parser
956 .parse_options(Keyword::WITH)
957 .context(error::SyntaxSnafu)?
958 .into_iter()
959 .map(parse_option_string)
960 .collect::<Result<Vec<_>>>()?;
961
962 for (key, _) in options.iter() {
963 ensure!(
964 validate_column_skipping_index_create_option(key),
965 InvalidColumnOptionSnafu {
966 name: column_name.to_string(),
967 msg: format!("invalid SKIPPING INDEX option: {key}"),
968 }
969 );
970 }
971
972 let options = OptionMap::new(options);
973 column_extensions.skipping_index_options = Some(options);
974 is_index_declared |= true;
975 }
976
977 if parser.parse_keyword(Keyword::FULLTEXT) {
979 ensure!(
981 parser.parse_keyword(Keyword::INDEX),
982 InvalidColumnOptionSnafu {
983 name: column_name.to_string(),
984 msg: "expect INDEX after FULLTEXT keyword",
985 }
986 );
987
988 ensure!(
989 column_extensions.fulltext_index_options.is_none(),
990 InvalidColumnOptionSnafu {
991 name: column_name.to_string(),
992 msg: "duplicated FULLTEXT INDEX option",
993 }
994 );
995
996 let column_type = get_unalias_type(column_type);
997 let data_type = sql_data_type_to_concrete_data_type(&column_type)?;
998 ensure!(
999 data_type == ConcreteDataType::string_datatype(),
1000 InvalidColumnOptionSnafu {
1001 name: column_name.to_string(),
1002 msg: "FULLTEXT index only supports string type",
1003 }
1004 );
1005
1006 let options = parser
1007 .parse_options(Keyword::WITH)
1008 .context(error::SyntaxSnafu)?
1009 .into_iter()
1010 .map(parse_option_string)
1011 .collect::<Result<Vec<_>>>()?;
1012
1013 for (key, _) in options.iter() {
1014 ensure!(
1015 validate_column_fulltext_create_option(key),
1016 InvalidColumnOptionSnafu {
1017 name: column_name.to_string(),
1018 msg: format!("invalid FULLTEXT INDEX option: {key}"),
1019 }
1020 );
1021 }
1022
1023 let options = OptionMap::new(options);
1024 column_extensions.fulltext_index_options = Some(options);
1025 is_index_declared |= true;
1026 }
1027
1028 if let Token::Word(word) = parser.peek_token().token
1030 && word.value.eq_ignore_ascii_case(INVERTED)
1031 {
1032 parser.next_token();
1033 ensure!(
1035 parser.parse_keyword(Keyword::INDEX),
1036 InvalidColumnOptionSnafu {
1037 name: column_name.to_string(),
1038 msg: "expect INDEX after INVERTED keyword",
1039 }
1040 );
1041
1042 ensure!(
1043 column_extensions.inverted_index_options.is_none(),
1044 InvalidColumnOptionSnafu {
1045 name: column_name.to_string(),
1046 msg: "duplicated INVERTED index option",
1047 }
1048 );
1049
1050 let with_token = parser.peek_token();
1053 ensure!(
1054 with_token.token
1055 != Token::Word(Word {
1056 value: "WITH".to_string(),
1057 keyword: Keyword::WITH,
1058 quote_style: None,
1059 }),
1060 InvalidColumnOptionSnafu {
1061 name: column_name.to_string(),
1062 msg: "INVERTED index doesn't support options",
1063 }
1064 );
1065
1066 column_extensions.inverted_index_options = Some(OptionMap::default());
1067 is_index_declared |= true;
1068 }
1069
1070 Ok(is_index_declared)
1071 }
1072
1073 fn parse_optional_table_constraint(&mut self) -> Result<Option<TableConstraint>> {
1074 match self.parser.next_token() {
1075 TokenWithSpan {
1076 token: Token::Word(w),
1077 ..
1078 } if w.keyword == Keyword::PRIMARY => {
1079 self.parser
1080 .expect_keyword(Keyword::KEY)
1081 .context(error::UnexpectedSnafu {
1082 expected: "KEY",
1083 actual: self.peek_token_as_string(),
1084 })?;
1085 let raw_columns = self
1086 .parser
1087 .parse_parenthesized_column_list(Mandatory, false)
1088 .context(error::SyntaxSnafu)?;
1089 let columns = raw_columns
1090 .into_iter()
1091 .map(Self::canonicalize_identifier)
1092 .collect();
1093 Ok(Some(TableConstraint::PrimaryKey { columns }))
1094 }
1095 TokenWithSpan {
1096 token: Token::Word(w),
1097 ..
1098 } if w.keyword == Keyword::TIME => {
1099 self.parser
1100 .expect_keyword(Keyword::INDEX)
1101 .context(error::UnexpectedSnafu {
1102 expected: "INDEX",
1103 actual: self.peek_token_as_string(),
1104 })?;
1105
1106 let raw_columns = self
1107 .parser
1108 .parse_parenthesized_column_list(Mandatory, false)
1109 .context(error::SyntaxSnafu)?;
1110 let mut columns = raw_columns
1111 .into_iter()
1112 .map(Self::canonicalize_identifier)
1113 .collect::<Vec<_>>();
1114
1115 ensure!(
1116 columns.len() == 1,
1117 InvalidTimeIndexSnafu {
1118 msg: "it should contain only one column in time index",
1119 }
1120 );
1121
1122 Ok(Some(TableConstraint::TimeIndex {
1123 column: columns.pop().unwrap(),
1124 }))
1125 }
1126 _ => {
1127 self.parser.prev_token();
1128 Ok(None)
1129 }
1130 }
1131 }
1132
1133 fn parse_table_engine(&mut self, default: &str) -> Result<String> {
1135 if !self.consume_token(ENGINE) {
1136 return Ok(default.to_string());
1137 }
1138
1139 self.parser
1140 .expect_token(&Token::Eq)
1141 .context(error::UnexpectedSnafu {
1142 expected: "=",
1143 actual: self.peek_token_as_string(),
1144 })?;
1145
1146 let token = self.parser.next_token();
1147 if let Token::Word(w) = token.token {
1148 Ok(w.value)
1149 } else {
1150 self.expected("'Engine' is missing", token)
1151 }
1152 }
1153}
1154
1155fn validate_time_index(columns: &[Column], constraints: &[TableConstraint]) -> Result<()> {
1156 let time_index_constraints: Vec<_> = constraints
1157 .iter()
1158 .filter_map(|c| match c {
1159 TableConstraint::TimeIndex { column } => Some(column),
1160 _ => None,
1161 })
1162 .unique()
1163 .collect();
1164
1165 ensure!(!time_index_constraints.is_empty(), MissingTimeIndexSnafu);
1166 ensure!(
1167 time_index_constraints.len() == 1,
1168 InvalidTimeIndexSnafu {
1169 msg: format!(
1170 "expected only one time index constraint but actual {}",
1171 time_index_constraints.len()
1172 ),
1173 }
1174 );
1175
1176 let time_index_column_ident = &time_index_constraints[0];
1179 let time_index_column = columns
1180 .iter()
1181 .find(|c| c.name().value == *time_index_column_ident.value)
1182 .with_context(|| InvalidTimeIndexSnafu {
1183 msg: format!(
1184 "time index column {} not found in columns",
1185 time_index_column_ident
1186 ),
1187 })?;
1188
1189 let time_index_data_type = get_unalias_type(time_index_column.data_type());
1190 ensure!(
1191 matches!(time_index_data_type, DataType::Timestamp(_, _)),
1192 InvalidColumnOptionSnafu {
1193 name: time_index_column.name().to_string(),
1194 msg: "time index column data type should be timestamp",
1195 }
1196 );
1197
1198 Ok(())
1199}
1200
1201fn get_unalias_type(data_type: &DataType) -> DataType {
1202 match data_type {
1203 DataType::Custom(name, tokens) if name.0.len() == 1 && tokens.is_empty() => {
1204 if let Some(real_type) =
1205 get_data_type_by_alias_name(name.0[0].to_string_unquoted().as_str())
1206 {
1207 real_type
1208 } else {
1209 data_type.clone()
1210 }
1211 }
1212 _ => data_type.clone(),
1213 }
1214}
1215
1216fn validate_partitions(columns: &[Column], partitions: &Partitions) -> Result<()> {
1217 let partition_columns = ensure_partition_columns_defined(columns, partitions)?;
1218
1219 ensure_exprs_are_binary(&partitions.exprs, &partition_columns)?;
1220
1221 Ok(())
1222}
1223
1224fn ensure_exprs_are_binary(exprs: &[Expr], columns: &[&Column]) -> Result<()> {
1226 for expr in exprs {
1227 if let Expr::BinaryOp { left, op: _, right } = expr {
1229 ensure_one_expr(left, columns)?;
1230 ensure_one_expr(right, columns)?;
1231 } else {
1232 return error::InvalidSqlSnafu {
1233 msg: format!("Partition rule expr {:?} is not a binary expr", expr),
1234 }
1235 .fail();
1236 }
1237 }
1238 Ok(())
1239}
1240
1241fn ensure_one_expr(expr: &Expr, columns: &[&Column]) -> Result<()> {
1245 match expr {
1246 Expr::BinaryOp { left, op: _, right } => {
1247 ensure_one_expr(left, columns)?;
1248 ensure_one_expr(right, columns)?;
1249 Ok(())
1250 }
1251 Expr::Identifier(ident) => {
1252 let column_name = &ident.value;
1253 ensure!(
1254 columns.iter().any(|c| &c.name().value == column_name),
1255 error::InvalidSqlSnafu {
1256 msg: format!(
1257 "Column {:?} in rule expr is not referenced in PARTITION ON",
1258 column_name
1259 ),
1260 }
1261 );
1262 Ok(())
1263 }
1264 Expr::Value(_) => Ok(()),
1265 Expr::UnaryOp { expr, .. } => {
1266 ensure_one_expr(expr, columns)?;
1267 Ok(())
1268 }
1269 _ => error::InvalidSqlSnafu {
1270 msg: format!("Partition rule expr {:?} is not a binary expr", expr),
1271 }
1272 .fail(),
1273 }
1274}
1275
1276fn ensure_partition_columns_defined<'a>(
1278 columns: &'a [Column],
1279 partitions: &'a Partitions,
1280) -> Result<Vec<&'a Column>> {
1281 partitions
1282 .column_list
1283 .iter()
1284 .map(|x| {
1285 let x = ParserContext::canonicalize_identifier(x.clone());
1286 columns
1289 .iter()
1290 .find(|c| *c.name().value == x.value)
1291 .context(error::InvalidSqlSnafu {
1292 msg: format!("Partition column {:?} not defined", x.value),
1293 })
1294 })
1295 .collect::<Result<Vec<&Column>>>()
1296}
1297
1298#[cfg(test)]
1299mod tests {
1300 use std::assert_matches;
1301 use std::collections::HashMap;
1302
1303 use common_catalog::consts::FILE_ENGINE;
1304 use common_error::ext::ErrorExt;
1305 use sqlparser::ast::ColumnOption::NotNull;
1306 use sqlparser::ast::{BinaryOperator, Expr, ObjectName, ObjectNamePart, Value};
1307 use sqlparser::dialect::GenericDialect;
1308 use sqlparser::tokenizer::Tokenizer;
1309
1310 use super::*;
1311 use crate::dialect::GreptimeDbDialect;
1312 use crate::parser::ParseOptions;
1313
1314 fn string_option_map(
1315 entries: impl IntoIterator<Item = (&'static str, &'static str)>,
1316 ) -> OptionMap {
1317 OptionMap::new(entries.into_iter().map(|(key, value)| {
1318 (
1319 key.to_string(),
1320 OptionValue::try_new(Expr::Value(
1321 Value::SingleQuotedString(value.to_string()).into(),
1322 ))
1323 .unwrap(),
1324 )
1325 }))
1326 }
1327
1328 #[test]
1329 fn test_parse_create_table_like() {
1330 let sql = "CREATE TABLE t1 LIKE t2";
1331 let stmts =
1332 ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
1333 .unwrap();
1334
1335 assert_eq!(1, stmts.len());
1336 match &stmts[0] {
1337 Statement::CreateTableLike(c) => {
1338 assert_eq!(c.table_name.to_string(), "t1");
1339 assert_eq!(c.source_name.to_string(), "t2");
1340 }
1341 _ => unreachable!(),
1342 }
1343 }
1344
1345 #[test]
1346 fn test_validate_external_table_options() {
1347 let sql = "CREATE EXTERNAL TABLE city (
1348 host string,
1349 ts timestamp,
1350 cpu float64 default 0,
1351 memory float64,
1352 TIME INDEX (ts),
1353 PRIMARY KEY(ts, host)
1354 ) with(location='/var/data/city.csv',format='csv',foo='bar');";
1355
1356 let result =
1357 ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default());
1358 assert!(matches!(
1359 result,
1360 Err(error::Error::InvalidTableOption { .. })
1361 ));
1362 }
1363
1364 #[test]
1365 fn test_parse_create_external_table() {
1366 struct Test<'a> {
1367 sql: &'a str,
1368 expected_table_name: &'a str,
1369 expected_options: HashMap<&'a str, &'a str>,
1370 expected_engine: &'a str,
1371 expected_if_not_exist: bool,
1372 }
1373
1374 let tests = [
1375 Test {
1376 sql: "CREATE EXTERNAL TABLE city with(location='/var/data/city.csv',format='csv');",
1377 expected_table_name: "city",
1378 expected_options: HashMap::from([
1379 ("location", "/var/data/city.csv"),
1380 ("format", "csv"),
1381 ]),
1382 expected_engine: FILE_ENGINE,
1383 expected_if_not_exist: false,
1384 },
1385 Test {
1386 sql: "CREATE EXTERNAL TABLE IF NOT EXISTS city ENGINE=foo with(location='/var/data/city.csv',format='csv');",
1387 expected_table_name: "city",
1388 expected_options: HashMap::from([
1389 ("location", "/var/data/city.csv"),
1390 ("format", "csv"),
1391 ]),
1392 expected_engine: "foo",
1393 expected_if_not_exist: true,
1394 },
1395 Test {
1396 sql: "CREATE EXTERNAL TABLE IF NOT EXISTS city ENGINE=foo with(location='/var/data/city.csv',format='csv','compaction.type'='bar');",
1397 expected_table_name: "city",
1398 expected_options: HashMap::from([
1399 ("location", "/var/data/city.csv"),
1400 ("format", "csv"),
1401 ("compaction.type", "bar"),
1402 ]),
1403 expected_engine: "foo",
1404 expected_if_not_exist: true,
1405 },
1406 ];
1407
1408 for test in tests {
1409 let stmts = ParserContext::create_with_dialect(
1410 test.sql,
1411 &GreptimeDbDialect {},
1412 ParseOptions::default(),
1413 )
1414 .unwrap();
1415 assert_eq!(1, stmts.len());
1416 match &stmts[0] {
1417 Statement::CreateExternalTable(c) => {
1418 assert_eq!(c.name.to_string(), test.expected_table_name.to_string());
1419 assert_eq!(c.options.to_str_map(), test.expected_options);
1420 assert_eq!(c.if_not_exists, test.expected_if_not_exist);
1421 assert_eq!(c.engine, test.expected_engine);
1422 }
1423 _ => unreachable!(),
1424 }
1425 }
1426 }
1427
1428 #[test]
1429 fn test_parse_create_external_table_with_schema() {
1430 let sql = "CREATE EXTERNAL TABLE city (
1431 host string,
1432 ts timestamp,
1433 cpu float32 default 0,
1434 memory float64,
1435 TIME INDEX (ts),
1436 PRIMARY KEY(ts, host),
1437 ) with(location='/var/data/city.csv',format='csv');";
1438
1439 let options = HashMap::from([("location", "/var/data/city.csv"), ("format", "csv")]);
1440
1441 let stmts =
1442 ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
1443 .unwrap();
1444 assert_eq!(1, stmts.len());
1445 match &stmts[0] {
1446 Statement::CreateExternalTable(c) => {
1447 assert_eq!(c.name.to_string(), "city");
1448 assert_eq!(c.options.to_str_map(), options);
1449
1450 let columns = &c.columns;
1451 assert_column_def(&columns[0].column_def, "host", "STRING");
1452 assert_column_def(&columns[1].column_def, "ts", "TIMESTAMP");
1453 assert_column_def(&columns[2].column_def, "cpu", "FLOAT");
1454 assert_column_def(&columns[3].column_def, "memory", "DOUBLE");
1455
1456 let constraints = &c.constraints;
1457 assert_eq!(
1458 &constraints[0],
1459 &TableConstraint::TimeIndex {
1460 column: Ident::new("ts"),
1461 }
1462 );
1463 assert_eq!(
1464 &constraints[1],
1465 &TableConstraint::PrimaryKey {
1466 columns: vec![Ident::new("ts"), Ident::new("host")]
1467 }
1468 );
1469 }
1470 _ => unreachable!(),
1471 }
1472 }
1473
1474 #[test]
1475 fn test_parse_create_database() {
1476 let sql = "create database";
1477 let result =
1478 ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default());
1479 assert!(
1480 result
1481 .unwrap_err()
1482 .to_string()
1483 .contains("Unexpected token while parsing SQL statement")
1484 );
1485
1486 let sql = "create database prometheus";
1487 let stmts =
1488 ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
1489 .unwrap();
1490
1491 assert_eq!(1, stmts.len());
1492 match &stmts[0] {
1493 Statement::CreateDatabase(c) => {
1494 assert_eq!(c.name.to_string(), "prometheus");
1495 assert!(!c.if_not_exists);
1496 }
1497 _ => unreachable!(),
1498 }
1499
1500 let sql = "create database if not exists prometheus";
1501 let stmts =
1502 ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
1503 .unwrap();
1504
1505 assert_eq!(1, stmts.len());
1506 match &stmts[0] {
1507 Statement::CreateDatabase(c) => {
1508 assert_eq!(c.name.to_string(), "prometheus");
1509 assert!(c.if_not_exists);
1510 }
1511 _ => unreachable!(),
1512 }
1513
1514 let sql = "CREATE DATABASE `fOo`";
1515 let result =
1516 ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default());
1517 let stmts = result.unwrap();
1518 match &stmts.last().unwrap() {
1519 Statement::CreateDatabase(c) => {
1520 assert_eq!(c.name, vec![Ident::with_quote('`', "fOo")].into());
1521 assert!(!c.if_not_exists);
1522 }
1523 _ => unreachable!(),
1524 }
1525
1526 let sql = "CREATE DATABASE prometheus with (ttl='1h');";
1527 let result =
1528 ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default());
1529 let stmts = result.unwrap();
1530 match &stmts[0] {
1531 Statement::CreateDatabase(c) => {
1532 assert_eq!(c.name.to_string(), "prometheus");
1533 assert!(!c.if_not_exists);
1534 assert_eq!(c.options.get("ttl").unwrap(), "1h");
1535 }
1536 _ => unreachable!(),
1537 }
1538
1539 let sql = "CREATE DATABASE prometheus with ('ingest_rows_rate_limit'='1000');";
1540 let result =
1541 ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default());
1542 let stmts = result.unwrap();
1543 match &stmts[0] {
1544 Statement::CreateDatabase(c) => {
1545 assert_eq!(c.name.to_string(), "prometheus");
1546 assert_eq!(c.options.get("ingest_rows_rate_limit").unwrap(), "1000");
1547 }
1548 _ => unreachable!(),
1549 }
1550 }
1551
1552 #[test]
1553 fn test_parse_create_database_option_validation() {
1554 let overflow = format!("{}0", usize::MAX);
1555 for key in [
1556 "compaction.twcs.trigger_file_num",
1557 "compaction.twcs.active_window.trigger_file_num",
1558 "compaction.twcs.inactive_window.trigger_file_num",
1559 ] {
1560 for invalid in ["invalid", "-1", overflow.as_str()] {
1561 let sql = format!("CREATE DATABASE invalid WITH ('{key}'='{invalid}')");
1562 let err = ParserContext::create_with_dialect(
1563 &sql,
1564 &GreptimeDbDialect {},
1565 ParseOptions::default(),
1566 )
1567 .unwrap_err();
1568 assert_eq!(
1569 err.to_string(),
1570 format!(
1571 "Invalid database option value for {key}: {invalid}, expected a non-negative integer fitting in usize"
1572 )
1573 );
1574 }
1575 for valid in ["0", "1"] {
1576 let sql = format!("CREATE DATABASE valid WITH ('{key}'='{valid}')");
1577 ParserContext::create_with_dialect(
1578 &sql,
1579 &GreptimeDbDialect {},
1580 ParseOptions::default(),
1581 )
1582 .unwrap();
1583 }
1584 }
1585 for key in [
1586 "compaction.twcs.active_window.l1_merge_trigger",
1587 "compaction.twcs.inactive_window.l1_merge_trigger",
1588 ] {
1589 let sql = format!("CREATE DATABASE invalid WITH ('{key}'='1')");
1590 let err = ParserContext::create_with_dialect(
1591 &sql,
1592 &GreptimeDbDialect {},
1593 ParseOptions::default(),
1594 )
1595 .unwrap_err();
1596 assert_eq!(
1597 format!(
1598 "Invalid database option value for {key}: 1, expected an integer greater than or equal to 2"
1599 ),
1600 err.to_string()
1601 );
1602 }
1603
1604 let sql =
1605 "CREATE DATABASE valid WITH ('compaction.twcs.active_window.l1_merge_trigger'='2')";
1606 ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
1607 .unwrap();
1608
1609 let sql = "CREATE DATABASE invalid WITH ('unknown'='1')";
1610 let err =
1611 ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
1612 .unwrap_err();
1613 assert_eq!("Unrecognized database option key: unknown", err.to_string());
1614 }
1615
1616 #[test]
1617 fn test_parse_create_flow_more_testcases() {
1618 use pretty_assertions::assert_eq;
1619 fn parse_create_flow(sql: &str) -> CreateFlow {
1620 let stmts = ParserContext::create_with_dialect(
1621 sql,
1622 &GreptimeDbDialect {},
1623 ParseOptions::default(),
1624 )
1625 .unwrap();
1626 assert_eq!(1, stmts.len());
1627 match &stmts[0] {
1628 Statement::CreateFlow(c) => c.clone(),
1629 _ => unreachable!(),
1630 }
1631 }
1632 struct CreateFlowWoutQuery {
1633 pub flow_name: ObjectName,
1635 pub sink_table_name: ObjectName,
1637 pub or_replace: bool,
1639 pub if_not_exists: bool,
1641 pub expire_after: Option<i64>,
1644 pub comment: Option<String>,
1646 pub flow_options: OptionMap,
1648 }
1649 let testcases = vec![
1650 (
1651 r"
1652CREATE OR REPLACE FLOW IF NOT EXISTS task_1
1653SINK TO schema_1.table_1
1654EXPIRE AFTER INTERVAL '5 minutes'
1655COMMENT 'test comment'
1656AS
1657SELECT max(c1), min(c2) FROM schema_2.table_2;",
1658 CreateFlowWoutQuery {
1659 flow_name: ObjectName::from(vec![Ident::new("task_1")]),
1660 sink_table_name: ObjectName::from(vec![
1661 Ident::new("schema_1"),
1662 Ident::new("table_1"),
1663 ]),
1664 or_replace: true,
1665 if_not_exists: true,
1666 expire_after: Some(300),
1667 comment: Some("test comment".to_string()),
1668 flow_options: OptionMap::default(),
1669 },
1670 ),
1671 (
1672 r"
1673CREATE OR REPLACE FLOW IF NOT EXISTS task_1
1674SINK TO schema_1.table_1
1675EXPIRE AFTER INTERVAL '300 s'
1676COMMENT 'test comment'
1677AS
1678SELECT max(c1), min(c2) FROM schema_2.table_2;",
1679 CreateFlowWoutQuery {
1680 flow_name: ObjectName::from(vec![Ident::new("task_1")]),
1681 sink_table_name: ObjectName::from(vec![
1682 Ident::new("schema_1"),
1683 Ident::new("table_1"),
1684 ]),
1685 or_replace: true,
1686 if_not_exists: true,
1687 expire_after: Some(300),
1688 comment: Some("test comment".to_string()),
1689 flow_options: OptionMap::default(),
1690 },
1691 ),
1692 (
1693 r"
1694CREATE OR REPLACE FLOW IF NOT EXISTS task_1
1695SINK TO schema_1.table_1
1696EXPIRE AFTER '5 minutes'
1697COMMENT 'test comment'
1698AS
1699SELECT max(c1), min(c2) FROM schema_2.table_2;",
1700 CreateFlowWoutQuery {
1701 flow_name: ObjectName::from(vec![Ident::new("task_1")]),
1702 sink_table_name: ObjectName::from(vec![
1703 Ident::new("schema_1"),
1704 Ident::new("table_1"),
1705 ]),
1706 or_replace: true,
1707 if_not_exists: true,
1708 expire_after: Some(300),
1709 comment: Some("test comment".to_string()),
1710 flow_options: OptionMap::default(),
1711 },
1712 ),
1713 (
1714 r"
1715CREATE OR REPLACE FLOW IF NOT EXISTS task_1
1716SINK TO schema_1.table_1
1717EXPIRE AFTER '300 s'
1718COMMENT 'test comment'
1719AS
1720SELECT max(c1), min(c2) FROM schema_2.table_2;",
1721 CreateFlowWoutQuery {
1722 flow_name: ObjectName::from(vec![Ident::new("task_1")]),
1723 sink_table_name: ObjectName::from(vec![
1724 Ident::new("schema_1"),
1725 Ident::new("table_1"),
1726 ]),
1727 or_replace: true,
1728 if_not_exists: true,
1729 expire_after: Some(300),
1730 comment: Some("test comment".to_string()),
1731 flow_options: OptionMap::default(),
1732 },
1733 ),
1734 (
1735 r"
1736CREATE FLOW `task_2`
1737SINK TO schema_1.table_1
1738EXPIRE AFTER '2 days 1h 2 min'
1739AS
1740SELECT max(c1), min(c2) FROM schema_2.table_2;",
1741 CreateFlowWoutQuery {
1742 flow_name: ObjectName::from(vec![Ident::with_quote('`', "task_2")]),
1743 sink_table_name: ObjectName::from(vec![
1744 Ident::new("schema_1"),
1745 Ident::new("table_1"),
1746 ]),
1747 or_replace: false,
1748 if_not_exists: false,
1749 expire_after: Some(2 * 86400 + 3600 + 2 * 60),
1750 comment: None,
1751 flow_options: OptionMap::default(),
1752 },
1753 ),
1754 (
1755 r"
1756create flow `task_3`
1757sink to schema_1.table_1
1758expire after '10 minutes'
1759as
1760select max(c1), min(c2) from schema_2.table_2;",
1761 CreateFlowWoutQuery {
1762 flow_name: ObjectName::from(vec![Ident::with_quote('`', "task_3")]),
1763 sink_table_name: ObjectName::from(vec![
1764 Ident::new("schema_1"),
1765 Ident::new("table_1"),
1766 ]),
1767 or_replace: false,
1768 if_not_exists: false,
1769 expire_after: Some(600), comment: None,
1771 flow_options: OptionMap::default(),
1772 },
1773 ),
1774 (
1775 r"
1776create or replace flow if not exists task_4
1777sink to schema_1.table_1
1778expire after interval '2 hours'
1779comment 'lowercase test'
1780as
1781select max(c1), min(c2) from schema_2.table_2;",
1782 CreateFlowWoutQuery {
1783 flow_name: ObjectName::from(vec![Ident::new("task_4")]),
1784 sink_table_name: ObjectName::from(vec![
1785 Ident::new("schema_1"),
1786 Ident::new("table_1"),
1787 ]),
1788 or_replace: true,
1789 if_not_exists: true,
1790 expire_after: Some(7200), comment: Some("lowercase test".to_string()),
1792 flow_options: OptionMap::default(),
1793 },
1794 ),
1795 (
1796 r"
1797CREATE FLOW task_5
1798SINK TO schema_1.table_1
1799WITH (defer_on_missing_source = 'true')
1800AS
1801SELECT max(c1), min(c2) FROM schema_2.table_2;",
1802 CreateFlowWoutQuery {
1803 flow_name: ObjectName::from(vec![Ident::new("task_5")]),
1804 sink_table_name: ObjectName::from(vec![
1805 Ident::new("schema_1"),
1806 Ident::new("table_1"),
1807 ]),
1808 or_replace: false,
1809 if_not_exists: false,
1810 expire_after: None,
1811 comment: None,
1812 flow_options: string_option_map([("defer_on_missing_source", "true")]),
1813 },
1814 ),
1815 ];
1816
1817 for (sql, expected) in testcases {
1818 let create_task = parse_create_flow(sql);
1819
1820 let expected = CreateFlow {
1821 flow_name: expected.flow_name,
1822 sink_table_name: expected.sink_table_name,
1823 or_replace: expected.or_replace,
1824 if_not_exists: expected.if_not_exists,
1825 expire_after: expected.expire_after,
1826 eval_interval: None,
1827 eval_offset: None,
1828 comment: expected.comment,
1829 flow_options: expected.flow_options,
1830 query: create_task.query.clone(),
1832 };
1833
1834 assert_eq!(create_task, expected, "input sql is:\n{sql}");
1835 let show_create = create_task.to_string();
1836 let recreated = parse_create_flow(&show_create);
1837 assert_eq!(recreated, expected, "input sql is:\n{show_create}");
1838 }
1839 }
1840
1841 #[test]
1842 fn test_parse_create_flow() {
1843 use pretty_assertions::assert_eq;
1844 fn parse_create_flow(sql: &str) -> CreateFlow {
1845 let stmts = ParserContext::create_with_dialect(
1846 sql,
1847 &GreptimeDbDialect {},
1848 ParseOptions::default(),
1849 )
1850 .unwrap();
1851 assert_eq!(1, stmts.len());
1852 match &stmts[0] {
1853 Statement::CreateFlow(c) => c.clone(),
1854 _ => panic!("{:?}", stmts[0]),
1855 }
1856 }
1857 struct CreateFlowWoutQuery {
1858 pub flow_name: ObjectName,
1860 pub sink_table_name: ObjectName,
1862 pub or_replace: bool,
1864 pub if_not_exists: bool,
1866 pub expire_after: Option<i64>,
1869 pub eval_interval: Option<i64>,
1873 pub eval_offset: Option<i64>,
1876 pub comment: Option<String>,
1878 pub flow_options: OptionMap,
1880 }
1881
1882 let testcases = vec![
1884 (
1885 r"
1886CREATE OR REPLACE FLOW IF NOT EXISTS task_1
1887SINK TO schema_1.table_1
1888EXPIRE AFTER INTERVAL '5 minutes'
1889COMMENT 'test comment'
1890AS
1891SELECT max(c1), min(c2) FROM schema_2.table_2;",
1892 CreateFlowWoutQuery {
1893 flow_name: ObjectName(vec![ObjectNamePart::Identifier(Ident::new("task_1"))]),
1894 sink_table_name: ObjectName(vec![
1895 ObjectNamePart::Identifier(Ident::new("schema_1")),
1896 ObjectNamePart::Identifier(Ident::new("table_1")),
1897 ]),
1898 or_replace: true,
1899 if_not_exists: true,
1900 expire_after: Some(300),
1901 eval_interval: None,
1902 eval_offset: None,
1903 comment: Some("test comment".to_string()),
1904 flow_options: OptionMap::default(),
1905 },
1906 ),
1907 (
1908 r"
1909CREATE OR REPLACE FLOW IF NOT EXISTS task_1
1910SINK TO schema_1.table_1
1911EXPIRE AFTER INTERVAL '300 s'
1912COMMENT 'test comment'
1913AS
1914SELECT max(c1), min(c2) FROM schema_2.table_2;",
1915 CreateFlowWoutQuery {
1916 flow_name: ObjectName(vec![ObjectNamePart::Identifier(Ident::new("task_1"))]),
1917 sink_table_name: ObjectName(vec![
1918 ObjectNamePart::Identifier(Ident::new("schema_1")),
1919 ObjectNamePart::Identifier(Ident::new("table_1")),
1920 ]),
1921 or_replace: true,
1922 if_not_exists: true,
1923 expire_after: Some(300),
1924 eval_interval: None,
1925 eval_offset: None,
1926 comment: Some("test comment".to_string()),
1927 flow_options: OptionMap::default(),
1928 },
1929 ),
1930 (
1931 r"
1932CREATE OR REPLACE FLOW IF NOT EXISTS task_1
1933SINK TO schema_1.table_1
1934EXPIRE AFTER '5 minutes'
1935EVAL INTERVAL '10 seconds'
1936COMMENT 'test comment'
1937AS
1938SELECT max(c1), min(c2) FROM schema_2.table_2;",
1939 CreateFlowWoutQuery {
1940 flow_name: ObjectName(vec![ObjectNamePart::Identifier(Ident::new("task_1"))]),
1941 sink_table_name: ObjectName(vec![
1942 ObjectNamePart::Identifier(Ident::new("schema_1")),
1943 ObjectNamePart::Identifier(Ident::new("table_1")),
1944 ]),
1945 or_replace: true,
1946 if_not_exists: true,
1947 expire_after: Some(300),
1948 eval_interval: Some(10),
1949 eval_offset: None,
1950 comment: Some("test comment".to_string()),
1951 flow_options: OptionMap::default(),
1952 },
1953 ),
1954 (
1955 r"
1956CREATE OR REPLACE FLOW IF NOT EXISTS task_1
1957SINK TO schema_1.table_1
1958EXPIRE AFTER '5 minutes'
1959EVAL INTERVAL INTERVAL '10 seconds'
1960COMMENT 'test comment'
1961AS
1962SELECT max(c1), min(c2) FROM schema_2.table_2;",
1963 CreateFlowWoutQuery {
1964 flow_name: ObjectName(vec![ObjectNamePart::Identifier(Ident::new("task_1"))]),
1965 sink_table_name: ObjectName(vec![
1966 ObjectNamePart::Identifier(Ident::new("schema_1")),
1967 ObjectNamePart::Identifier(Ident::new("table_1")),
1968 ]),
1969 or_replace: true,
1970 if_not_exists: true,
1971 expire_after: Some(300),
1972 eval_interval: Some(10),
1973 eval_offset: None,
1974 comment: Some("test comment".to_string()),
1975 flow_options: OptionMap::default(),
1976 },
1977 ),
1978 (
1979 r"
1980CREATE FLOW `task_2`
1981SINK TO schema_1.table_1
1982EXPIRE AFTER '2 days 1h 2 min'
1983AS
1984SELECT max(c1), min(c2) FROM schema_2.table_2;",
1985 CreateFlowWoutQuery {
1986 flow_name: ObjectName(vec![ObjectNamePart::Identifier(Ident::with_quote(
1987 '`', "task_2",
1988 ))]),
1989 sink_table_name: ObjectName(vec![
1990 ObjectNamePart::Identifier(Ident::new("schema_1")),
1991 ObjectNamePart::Identifier(Ident::new("table_1")),
1992 ]),
1993 or_replace: false,
1994 if_not_exists: false,
1995 expire_after: Some(2 * 86400 + 3600 + 2 * 60),
1996 eval_interval: None,
1997 eval_offset: None,
1998 comment: None,
1999 flow_options: OptionMap::default(),
2000 },
2001 ),
2002 (
2003 r"
2004CREATE FLOW task_3
2005SINK TO schema_1.table_1
2006EVAL INTERVAL '10 seconds'
2007WITH (defer_on_missing_source = 'true', foo = 'bar')
2008AS
2009SELECT max(c1), min(c2) FROM schema_2.table_2;",
2010 CreateFlowWoutQuery {
2011 flow_name: ObjectName(vec![ObjectNamePart::Identifier(Ident::new("task_3"))]),
2012 sink_table_name: ObjectName(vec![
2013 ObjectNamePart::Identifier(Ident::new("schema_1")),
2014 ObjectNamePart::Identifier(Ident::new("table_1")),
2015 ]),
2016 or_replace: false,
2017 if_not_exists: false,
2018 expire_after: None,
2019 eval_interval: Some(10),
2020 eval_offset: None,
2021 comment: None,
2022 flow_options: string_option_map([
2023 ("defer_on_missing_source", "true"),
2024 ("foo", "bar"),
2025 ]),
2026 },
2027 ),
2028 ];
2029
2030 for (sql, expected) in testcases {
2031 let create_task = parse_create_flow(sql);
2032
2033 let expected = CreateFlow {
2034 flow_name: expected.flow_name,
2035 sink_table_name: expected.sink_table_name,
2036 or_replace: expected.or_replace,
2037 if_not_exists: expected.if_not_exists,
2038 expire_after: expected.expire_after,
2039 eval_interval: expected.eval_interval,
2040 eval_offset: expected.eval_offset,
2041 comment: expected.comment,
2042 flow_options: expected.flow_options,
2043 query: create_task.query.clone(),
2045 };
2046
2047 assert_eq!(create_task, expected, "input sql is:\n{sql}");
2048 let show_create = create_task.to_string();
2049 let recreated = parse_create_flow(&show_create);
2050 assert_eq!(recreated, expected, "input sql is:\n{show_create}");
2051 }
2052 }
2053
2054 #[test]
2055 fn test_parse_create_flow_with_eval_offset() {
2056 use pretty_assertions::assert_eq;
2057 fn parse_create_flow(sql: &str) -> CreateFlow {
2058 let stmts = ParserContext::create_with_dialect(
2059 sql,
2060 &GreptimeDbDialect {},
2061 ParseOptions::default(),
2062 )
2063 .unwrap();
2064 assert_eq!(1, stmts.len());
2065 match &stmts[0] {
2066 Statement::CreateFlow(c) => c.clone(),
2067 _ => panic!("{:?}", stmts[0]),
2068 }
2069 }
2070 let sql = r#"
2071CREATE FLOW task_1
2072SINK TO schema_1.table_1
2073EVAL INTERVAL '1 hour'
2074EVAL OFFSET '2 minutes'
2075AS
2076SELECT max(c1), min(c2) FROM schema_2.table_2;"#;
2077 let create_task = parse_create_flow(sql);
2078 assert_eq!(create_task.eval_interval, Some(3600));
2079 assert_eq!(create_task.eval_offset, Some(120));
2080 let show_create = create_task.to_string();
2081 assert!(
2082 show_create.contains("EVAL OFFSET '120 s'"),
2083 "unexpected display:\n{show_create}"
2084 );
2085 let recreated = parse_create_flow(&show_create);
2086 assert_eq!(recreated, create_task, "input sql is:\n{show_create}");
2087
2088 let sql = r#"
2089create flow task_2
2090sink to schema_1.table_1
2091eval interval '1h'
2092eval offset '2m'
2093as
2094select max(c1), min(c2) from schema_2.table_2;"#;
2095 let create_task = parse_create_flow(sql);
2096 assert_eq!(create_task.eval_interval, Some(3600));
2097 assert_eq!(create_task.eval_offset, Some(120));
2098
2099 let sql = r#"
2101CREATE FLOW task_3
2102SINK TO schema_1.table_1
2103EVAL INTERVAL '1 hour'
2104EVAL OFFSET '0 seconds'
2105AS
2106SELECT max(c1), min(c2) FROM schema_2.table_2;"#;
2107 let create_task = parse_create_flow(sql);
2108 assert_eq!(create_task.eval_interval, Some(3600));
2109 assert_eq!(create_task.eval_offset, None);
2110 assert!(
2111 !create_task.to_string().contains("EVAL OFFSET"),
2112 "zero offset should be omitted on display"
2113 );
2114
2115 let sql = r#"
2116CREATE FLOW task_4
2117SINK TO schema_1.table_1
2118EVAL OFFSET '2 minutes'
2119AS
2120SELECT max(c1), min(c2) FROM schema_2.table_2;"#;
2121 let err =
2122 ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
2123 .unwrap_err()
2124 .to_string();
2125 assert!(
2126 err.contains("EVAL OFFSET requires EVAL INTERVAL"),
2127 "unexpected error: {err}"
2128 );
2129
2130 for (offset, interval) in [
2131 ("-1 seconds", "1 hour"),
2132 ("1 hour", "1 hour"),
2133 ("2 hours", "1 hour"),
2134 ] {
2135 let sql = format!(
2136 r#"
2137CREATE FLOW task_invalid
2138SINK TO schema_1.table_1
2139EVAL INTERVAL '{interval}'
2140EVAL OFFSET '{offset}'
2141AS
2142SELECT max(c1), min(c2) FROM schema_2.table_2;"#
2143 );
2144 let err = ParserContext::create_with_dialect(
2145 &sql,
2146 &GreptimeDbDialect {},
2147 ParseOptions::default(),
2148 )
2149 .unwrap_err()
2150 .to_string();
2151 assert!(
2152 err.contains("EVAL OFFSET must be in range"),
2153 "unexpected error for offset {offset}: {err}"
2154 );
2155 }
2156
2157 let sql = r#"
2158CREATE FLOW task_fractional_offset
2159SINK TO schema_1.table_1
2160EVAL INTERVAL '1 hour'
2161EVAL OFFSET '1.5 seconds'
2162AS
2163SELECT max(c1), min(c2) FROM schema_2.table_2;"#;
2164 let err =
2165 ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
2166 .unwrap_err()
2167 .to_string();
2168 assert!(
2169 err.contains("EVAL OFFSET must be a whole number of seconds"),
2170 "unexpected error: {err}"
2171 );
2172
2173 let sql = r#"
2174CREATE FLOW task_fractional_interval
2175SINK TO schema_1.table_1
2176EVAL INTERVAL '1.5 seconds'
2177EVAL OFFSET '1 second'
2178AS
2179SELECT max(c1), min(c2) FROM schema_2.table_2;"#;
2180 let err =
2181 ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
2182 .unwrap_err()
2183 .to_string();
2184 assert!(
2185 err.contains("EVAL INTERVAL must be a whole number of seconds"),
2186 "unexpected error: {err}"
2187 );
2188 }
2189
2190 #[test]
2191 fn test_parse_create_flow_with_tql_cte_query() {
2192 let sql = r#"
2193CREATE FLOW calc_reqs_cte
2194SINK TO cnt_reqs_cte
2195EVAL INTERVAL '1m'
2196AS
2197WITH tql(the_timestamp, the_value) AS (
2198 TQL EVAL (now() - '1m'::interval, now(), '5s') metric
2199)
2200SELECT * FROM tql;
2201"#;
2202
2203 let stmts =
2204 ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
2205 .unwrap();
2206 assert_eq!(1, stmts.len());
2207 let Statement::CreateFlow(create_flow) = &stmts[0] else {
2208 panic!("unexpected stmt: {:?}", stmts[0]);
2209 };
2210
2211 let query = create_flow.query.to_string();
2212 assert!(query.to_uppercase().contains("WITH"));
2213 assert!(query.to_uppercase().contains("TQL EVAL"));
2214 }
2215
2216 #[test]
2217 fn test_parse_create_flow_with_sql_cte_is_supported() {
2218 let sql = r#"
2219CREATE FLOW f
2220SINK TO s
2221AS
2222WITH cte AS (SELECT 1) SELECT * FROM cte;
2223"#;
2224
2225 let stmts =
2226 ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
2227 .unwrap();
2228 assert_eq!(1, stmts.len());
2229 let Statement::CreateFlow(create_flow) = &stmts[0] else {
2230 panic!("unexpected stmt: {:?}", stmts[0]);
2231 };
2232 assert_eq!(
2233 "WITH cte AS (SELECT 1) SELECT * FROM cte",
2234 create_flow.query.to_string()
2235 );
2236 }
2237
2238 #[test]
2239 fn test_parse_create_flow_with_tql_cte_requires_now_expr() {
2240 let sql = r#"
2241CREATE FLOW f
2242SINK TO s
2243EVAL INTERVAL '1m'
2244AS
2245WITH tql(ts, val) AS (
2246 TQL EVAL (0, 15, '5s') metric
2247)
2248SELECT * FROM tql;
2249"#;
2250
2251 let err =
2252 ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
2253 .unwrap_err();
2254
2255 let msg = format!("{err:?}");
2256 assert!(
2257 msg.contains("Expected expression containing `now()`"),
2258 "unexpected err: {msg}"
2259 );
2260 }
2261
2262 #[test]
2263 fn test_parse_create_flow_with_tql_cte_non_select_star_is_unsupported() {
2264 let sql = r#"
2265CREATE FLOW f
2266SINK TO s
2267AS
2268WITH tql(ts, val) AS (
2269 TQL EVAL (now() - '1m'::interval, now(), '5s') metric
2270)
2271SELECT ts FROM tql;
2272"#;
2273
2274 let err =
2275 ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
2276 .unwrap_err();
2277 assert!(err.to_string().contains("simplest TQL CTE"), "err: {err}");
2278 }
2279
2280 #[test]
2281 fn test_parse_create_flow_with_tql_cte_filter_is_unsupported() {
2282 let sql = r#"
2283CREATE FLOW f
2284SINK TO s
2285AS
2286WITH tql(ts, val) AS (
2287 TQL EVAL (now() - '1m'::interval, now(), '5s') metric
2288)
2289SELECT * FROM tql WHERE ts > 0;
2290"#;
2291
2292 let err =
2293 ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
2294 .unwrap_err();
2295 assert!(err.to_string().contains("simplest TQL CTE"), "err: {err}");
2296 }
2297
2298 #[test]
2299 fn test_parse_create_flow_with_mixed_sql_tql_cte_is_unsupported() {
2300 let sql = r#"
2301CREATE FLOW f
2302SINK TO s
2303AS
2304WITH s1 AS (SELECT 1),
2305 tql(ts, val) AS (TQL EVAL (now() - '1m'::interval, now(), '5s') metric)
2306SELECT * FROM tql;
2307"#;
2308
2309 let err =
2310 ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
2311 .unwrap_err();
2312 assert!(err.to_string().contains("simplest TQL CTE"), "err: {err}");
2313 }
2314
2315 #[test]
2316 fn test_create_flow_no_month() {
2317 let sql = r"
2318CREATE FLOW `task_2`
2319SINK TO schema_1.table_1
2320EXPIRE AFTER '1 month 2 days 1h 2 min'
2321AS
2322SELECT max(c1), min(c2) FROM schema_2.table_2;";
2323 let stmts =
2324 ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default());
2325
2326 assert!(
2327 stmts.is_err()
2328 && stmts
2329 .unwrap_err()
2330 .to_string()
2331 .contains("Interval with months is not allowed")
2332 );
2333 }
2334
2335 #[test]
2336 fn test_validate_create() {
2337 let sql = r"
2338CREATE TABLE rcx ( a INT, b STRING, c INT, ts timestamp TIME INDEX)
2339PARTITION ON COLUMNS(c, a) (
2340 a < 10,
2341 a > 10 AND a < 20,
2342 a > 20 AND c < 100,
2343 a > 20 AND c >= 100
2344)
2345ENGINE=mito";
2346 let result =
2347 ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default());
2348 let _ = result.unwrap();
2349
2350 let sql = r"
2351CREATE TABLE rcx ( ts TIMESTAMP TIME INDEX, a INT, b STRING, c INT )
2352PARTITION ON COLUMNS(x) ()
2353ENGINE=mito";
2354 let result =
2355 ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default());
2356 assert!(
2357 result
2358 .unwrap_err()
2359 .to_string()
2360 .contains("Partition column \"x\" not defined")
2361 );
2362 }
2363
2364 #[test]
2365 fn test_parse_create_table_with_partitions() {
2366 let sql = r"
2367CREATE TABLE monitor (
2368 host_id INT,
2369 idc STRING,
2370 ts TIMESTAMP,
2371 cpu DOUBLE DEFAULT 0,
2372 memory DOUBLE,
2373 TIME INDEX (ts),
2374 PRIMARY KEY (host),
2375)
2376PARTITION ON COLUMNS(idc, host_id) (
2377 idc <= 'hz' AND host_id < 1000,
2378 idc > 'hz' AND idc <= 'sh' AND host_id < 2000,
2379 idc > 'sh' AND host_id < 3000,
2380 idc > 'sh' AND host_id >= 3000,
2381)
2382ENGINE=mito";
2383 let result =
2384 ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
2385 .unwrap();
2386 assert_eq!(result.len(), 1);
2387 match &result[0] {
2388 Statement::CreateTable(c) => {
2389 assert!(c.partitions.is_some());
2390
2391 let partitions = c.partitions.as_ref().unwrap();
2392 let column_list = partitions
2393 .column_list
2394 .iter()
2395 .map(|x| &x.value)
2396 .collect::<Vec<&String>>();
2397 assert_eq!(column_list, vec!["idc", "host_id"]);
2398
2399 let exprs = &partitions.exprs;
2400
2401 assert_eq!(
2402 exprs[0],
2403 Expr::BinaryOp {
2404 left: Box::new(Expr::BinaryOp {
2405 left: Box::new(Expr::Identifier("idc".into())),
2406 op: BinaryOperator::LtEq,
2407 right: Box::new(Expr::Value(
2408 Value::SingleQuotedString("hz".to_string()).into()
2409 ))
2410 }),
2411 op: BinaryOperator::And,
2412 right: Box::new(Expr::BinaryOp {
2413 left: Box::new(Expr::Identifier("host_id".into())),
2414 op: BinaryOperator::Lt,
2415 right: Box::new(Expr::Value(
2416 Value::Number("1000".to_string(), false).into()
2417 ))
2418 })
2419 }
2420 );
2421 assert_eq!(
2422 exprs[1],
2423 Expr::BinaryOp {
2424 left: Box::new(Expr::BinaryOp {
2425 left: Box::new(Expr::BinaryOp {
2426 left: Box::new(Expr::Identifier("idc".into())),
2427 op: BinaryOperator::Gt,
2428 right: Box::new(Expr::Value(
2429 Value::SingleQuotedString("hz".to_string()).into()
2430 ))
2431 }),
2432 op: BinaryOperator::And,
2433 right: Box::new(Expr::BinaryOp {
2434 left: Box::new(Expr::Identifier("idc".into())),
2435 op: BinaryOperator::LtEq,
2436 right: Box::new(Expr::Value(
2437 Value::SingleQuotedString("sh".to_string()).into()
2438 ))
2439 })
2440 }),
2441 op: BinaryOperator::And,
2442 right: Box::new(Expr::BinaryOp {
2443 left: Box::new(Expr::Identifier("host_id".into())),
2444 op: BinaryOperator::Lt,
2445 right: Box::new(Expr::Value(
2446 Value::Number("2000".to_string(), false).into()
2447 ))
2448 })
2449 }
2450 );
2451 assert_eq!(
2452 exprs[2],
2453 Expr::BinaryOp {
2454 left: Box::new(Expr::BinaryOp {
2455 left: Box::new(Expr::Identifier("idc".into())),
2456 op: BinaryOperator::Gt,
2457 right: Box::new(Expr::Value(
2458 Value::SingleQuotedString("sh".to_string()).into()
2459 ))
2460 }),
2461 op: BinaryOperator::And,
2462 right: Box::new(Expr::BinaryOp {
2463 left: Box::new(Expr::Identifier("host_id".into())),
2464 op: BinaryOperator::Lt,
2465 right: Box::new(Expr::Value(
2466 Value::Number("3000".to_string(), false).into()
2467 ))
2468 })
2469 }
2470 );
2471 assert_eq!(
2472 exprs[3],
2473 Expr::BinaryOp {
2474 left: Box::new(Expr::BinaryOp {
2475 left: Box::new(Expr::Identifier("idc".into())),
2476 op: BinaryOperator::Gt,
2477 right: Box::new(Expr::Value(
2478 Value::SingleQuotedString("sh".to_string()).into()
2479 ))
2480 }),
2481 op: BinaryOperator::And,
2482 right: Box::new(Expr::BinaryOp {
2483 left: Box::new(Expr::Identifier("host_id".into())),
2484 op: BinaryOperator::GtEq,
2485 right: Box::new(Expr::Value(
2486 Value::Number("3000".to_string(), false).into()
2487 ))
2488 })
2489 }
2490 );
2491 }
2492 _ => unreachable!(),
2493 }
2494 }
2495
2496 #[test]
2497 fn test_parse_create_table_with_quoted_partitions() {
2498 let sql = r"
2499CREATE TABLE monitor (
2500 `host_id` INT,
2501 idc STRING,
2502 ts TIMESTAMP,
2503 cpu DOUBLE DEFAULT 0,
2504 memory DOUBLE,
2505 TIME INDEX (ts),
2506 PRIMARY KEY (host),
2507)
2508PARTITION ON COLUMNS(IdC, host_id) (
2509 idc <= 'hz' AND host_id < 1000,
2510 idc > 'hz' AND idc <= 'sh' AND host_id < 2000,
2511 idc > 'sh' AND host_id < 3000,
2512 idc > 'sh' AND host_id >= 3000,
2513)";
2514 let result =
2515 ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
2516 .unwrap();
2517 assert_eq!(result.len(), 1);
2518 }
2519
2520 #[test]
2521 fn test_parse_create_table_with_timestamp_index() {
2522 let sql1 = r"
2523CREATE TABLE monitor (
2524 host_id INT,
2525 idc STRING,
2526 ts TIMESTAMP TIME INDEX,
2527 cpu DOUBLE DEFAULT 0,
2528 memory DOUBLE,
2529 PRIMARY KEY (host),
2530)
2531ENGINE=mito";
2532 let result1 = ParserContext::create_with_dialect(
2533 sql1,
2534 &GreptimeDbDialect {},
2535 ParseOptions::default(),
2536 )
2537 .unwrap();
2538
2539 if let Statement::CreateTable(c) = &result1[0] {
2540 assert_eq!(c.constraints.len(), 2);
2541 let tc = c.constraints[0].clone();
2542 match tc {
2543 TableConstraint::TimeIndex { column } => {
2544 assert_eq!(&column.value, "ts");
2545 }
2546 _ => panic!("should be time index constraint"),
2547 };
2548 } else {
2549 panic!("should be create_table statement");
2550 }
2551
2552 let sql2 = r"
2555CREATE TABLE monitor (
2556 host_id INT,
2557 idc STRING,
2558 ts TIMESTAMP NOT NULL,
2559 cpu DOUBLE DEFAULT 0,
2560 memory DOUBLE,
2561 TIME INDEX (ts),
2562 PRIMARY KEY (host),
2563)
2564ENGINE=mito";
2565 let result2 = ParserContext::create_with_dialect(
2566 sql2,
2567 &GreptimeDbDialect {},
2568 ParseOptions::default(),
2569 )
2570 .unwrap();
2571
2572 assert_eq!(result1, result2);
2573
2574 let sql3 = r"
2576CREATE TABLE monitor (
2577 host_id INT,
2578 idc STRING,
2579 ts TIMESTAMP,
2580 cpu DOUBLE DEFAULT 0,
2581 memory DOUBLE,
2582 TIME INDEX (ts),
2583 PRIMARY KEY (host),
2584)
2585ENGINE=mito";
2586
2587 let result3 = ParserContext::create_with_dialect(
2588 sql3,
2589 &GreptimeDbDialect {},
2590 ParseOptions::default(),
2591 )
2592 .unwrap();
2593
2594 assert_ne!(result1, result3);
2595
2596 let sql1 = r"
2598CREATE TABLE monitor (
2599 host_id INT,
2600 idc STRING,
2601 b bigint TIME INDEX,
2602 cpu DOUBLE DEFAULT 0,
2603 memory DOUBLE,
2604 PRIMARY KEY (host),
2605)
2606ENGINE=mito";
2607 let result1 = ParserContext::create_with_dialect(
2608 sql1,
2609 &GreptimeDbDialect {},
2610 ParseOptions::default(),
2611 );
2612
2613 assert!(
2614 result1
2615 .unwrap_err()
2616 .to_string()
2617 .contains("time index column data type should be timestamp")
2618 );
2619 }
2620
2621 #[test]
2622 fn test_parse_create_table_with_timestamp_index_not_null() {
2623 let sql = r"
2624CREATE TABLE monitor (
2625 host_id INT,
2626 idc STRING,
2627 ts TIMESTAMP TIME INDEX,
2628 cpu DOUBLE DEFAULT 0,
2629 memory DOUBLE,
2630 TIME INDEX (ts),
2631 PRIMARY KEY (host),
2632)
2633ENGINE=mito";
2634 let result =
2635 ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
2636 .unwrap();
2637
2638 assert_eq!(result.len(), 1);
2639 if let Statement::CreateTable(c) = &result[0] {
2640 let ts = c.columns[2].clone();
2641 assert_eq!(ts.name().to_string(), "ts");
2642 assert_eq!(ts.options()[0].option, NotNull);
2643 } else {
2644 panic!("should be create table statement");
2645 }
2646
2647 let sql1 = r"
2648CREATE TABLE monitor (
2649 host_id INT,
2650 idc STRING,
2651 ts TIMESTAMP NOT NULL TIME INDEX,
2652 cpu DOUBLE DEFAULT 0,
2653 memory DOUBLE,
2654 TIME INDEX (ts),
2655 PRIMARY KEY (host),
2656)
2657ENGINE=mito";
2658
2659 let result1 = ParserContext::create_with_dialect(
2660 sql1,
2661 &GreptimeDbDialect {},
2662 ParseOptions::default(),
2663 )
2664 .unwrap();
2665 assert_eq!(result, result1);
2666
2667 let sql2 = r"
2668CREATE TABLE monitor (
2669 host_id INT,
2670 idc STRING,
2671 ts TIMESTAMP TIME INDEX NOT NULL,
2672 cpu DOUBLE DEFAULT 0,
2673 memory DOUBLE,
2674 TIME INDEX (ts),
2675 PRIMARY KEY (host),
2676)
2677ENGINE=mito";
2678
2679 let result2 = ParserContext::create_with_dialect(
2680 sql2,
2681 &GreptimeDbDialect {},
2682 ParseOptions::default(),
2683 )
2684 .unwrap();
2685 assert_eq!(result, result2);
2686
2687 let sql3 = r"
2688CREATE TABLE monitor (
2689 host_id INT,
2690 idc STRING,
2691 ts TIMESTAMP TIME INDEX NULL NOT,
2692 cpu DOUBLE DEFAULT 0,
2693 memory DOUBLE,
2694 TIME INDEX (ts),
2695 PRIMARY KEY (host),
2696)
2697ENGINE=mito";
2698
2699 let result3 = ParserContext::create_with_dialect(
2700 sql3,
2701 &GreptimeDbDialect {},
2702 ParseOptions::default(),
2703 );
2704 assert!(result3.is_err());
2705
2706 let sql4 = r"
2707CREATE TABLE monitor (
2708 host_id INT,
2709 idc STRING,
2710 ts TIMESTAMP TIME INDEX NOT NULL NULL,
2711 cpu DOUBLE DEFAULT 0,
2712 memory DOUBLE,
2713 TIME INDEX (ts),
2714 PRIMARY KEY (host),
2715)
2716ENGINE=mito";
2717
2718 let result4 = ParserContext::create_with_dialect(
2719 sql4,
2720 &GreptimeDbDialect {},
2721 ParseOptions::default(),
2722 );
2723 assert!(result4.is_err());
2724
2725 let sql = r"
2726CREATE TABLE monitor (
2727 host_id INT,
2728 idc STRING,
2729 ts TIMESTAMP TIME INDEX DEFAULT CURRENT_TIMESTAMP,
2730 cpu DOUBLE DEFAULT 0,
2731 memory DOUBLE,
2732 TIME INDEX (ts),
2733 PRIMARY KEY (host),
2734)
2735ENGINE=mito";
2736
2737 let result =
2738 ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
2739 .unwrap();
2740
2741 if let Statement::CreateTable(c) = &result[0] {
2742 let tc = c.constraints[0].clone();
2743 match tc {
2744 TableConstraint::TimeIndex { column } => {
2745 assert_eq!(&column.value, "ts");
2746 }
2747 _ => panic!("should be time index constraint"),
2748 }
2749 let ts = c.columns[2].clone();
2750 assert_eq!(ts.name().to_string(), "ts");
2751 assert!(matches!(ts.options()[0].option, ColumnOption::Default(..)));
2752 assert_eq!(ts.options()[1].option, NotNull);
2753 } else {
2754 unreachable!("should be create table statement");
2755 }
2756 }
2757
2758 #[test]
2759 fn test_parse_partitions_with_error_syntax() {
2760 let sql = r"
2761CREATE TABLE rcx ( ts TIMESTAMP TIME INDEX, a INT, b STRING, c INT )
2762PARTITION COLUMNS(c, a) (
2763 a < 10,
2764 a > 10 AND a < 20,
2765 a > 20 AND c < 100,
2766 a > 20 AND c >= 100
2767)
2768ENGINE=mito";
2769 let result =
2770 ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default());
2771 assert!(
2772 result
2773 .unwrap_err()
2774 .output_msg()
2775 .contains("sql parser error: Expected: ON, found: COLUMNS")
2776 );
2777 }
2778
2779 #[test]
2780 fn test_parse_partitions_without_rule() {
2781 let sql = r"
2782CREATE TABLE rcx ( a INT, b STRING, c INT, d TIMESTAMP TIME INDEX )
2783PARTITION ON COLUMNS(c, a) ()
2784ENGINE=mito";
2785 ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
2786 .unwrap();
2787 }
2788
2789 #[test]
2790 fn test_parse_partitions_unreferenced_column() {
2791 let sql = r"
2792CREATE TABLE rcx ( ts TIMESTAMP TIME INDEX, a INT, b STRING, c INT )
2793PARTITION ON COLUMNS(c, a) (
2794 b = 'foo'
2795)
2796ENGINE=mito";
2797 let result =
2798 ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default());
2799 assert_eq!(
2800 result.unwrap_err().output_msg(),
2801 "Invalid SQL, error: Column \"b\" in rule expr is not referenced in PARTITION ON"
2802 );
2803 }
2804
2805 #[test]
2806 fn test_parse_partitions_not_binary_expr() {
2807 let sql = r"
2808CREATE TABLE rcx ( ts TIMESTAMP TIME INDEX, a INT, b STRING, c INT )
2809PARTITION ON COLUMNS(c, a) (
2810 b
2811)
2812ENGINE=mito";
2813 let result =
2814 ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default());
2815 assert_eq!(
2816 result.unwrap_err().output_msg(),
2817 r#"Invalid SQL, error: Partition rule expr Identifier(Ident { value: "b", quote_style: None, span: Span(Location(4,5)..Location(4,6)) }) is not a binary expr"#
2818 );
2819 }
2820
2821 fn assert_column_def(column: &ColumnDef, name: &str, data_type: &str) {
2822 assert_eq!(column.name.to_string(), name);
2823 assert_eq!(column.data_type.to_string(), data_type);
2824 }
2825
2826 #[test]
2827 pub fn test_parse_create_table() {
2828 let sql = r"create table demo(
2829 host string,
2830 ts timestamp,
2831 cpu float32 default 0,
2832 memory float64,
2833 TIME INDEX (ts),
2834 PRIMARY KEY(ts, host),
2835 ) engine=mito
2836 with(ttl='10s');
2837 ";
2838 let result =
2839 ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
2840 .unwrap();
2841 assert_eq!(1, result.len());
2842 match &result[0] {
2843 Statement::CreateTable(c) => {
2844 assert!(!c.if_not_exists);
2845 assert_eq!("demo", c.name.to_string());
2846 assert_eq!("mito", c.engine);
2847 assert_eq!(4, c.columns.len());
2848 let columns = &c.columns;
2849 assert_column_def(&columns[0].column_def, "host", "STRING");
2850 assert_column_def(&columns[1].column_def, "ts", "TIMESTAMP");
2851 assert_column_def(&columns[2].column_def, "cpu", "FLOAT");
2852 assert_column_def(&columns[3].column_def, "memory", "DOUBLE");
2853
2854 let constraints = &c.constraints;
2855 assert_eq!(
2856 &constraints[0],
2857 &TableConstraint::TimeIndex {
2858 column: Ident::new("ts"),
2859 }
2860 );
2861 assert_eq!(
2862 &constraints[1],
2863 &TableConstraint::PrimaryKey {
2864 columns: vec![Ident::new("ts"), Ident::new("host")]
2865 }
2866 );
2867 assert_eq!(1, c.options.len());
2869 assert_eq!(
2870 [("ttl", "10s")].into_iter().collect::<HashMap<_, _>>(),
2871 c.options.to_str_map()
2872 );
2873 }
2874 _ => unreachable!(),
2875 }
2876 }
2877
2878 #[test]
2879 fn test_invalid_index_keys() {
2880 let sql = r"create table demo(
2881 host string,
2882 ts int64,
2883 cpu float64 default 0,
2884 memory float64,
2885 TIME INDEX (ts, host),
2886 PRIMARY KEY(ts, host)) engine=mito;
2887 ";
2888 let result =
2889 ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default());
2890 assert!(result.is_err());
2891 assert_matches!(result, Err(crate::error::Error::InvalidTimeIndex { .. }));
2892 }
2893
2894 #[test]
2895 fn test_duplicated_time_index() {
2896 let sql = r"create table demo(
2897 host string,
2898 ts timestamp time index,
2899 t timestamp time index,
2900 cpu float64 default 0,
2901 memory float64,
2902 TIME INDEX (ts, host),
2903 PRIMARY KEY(ts, host)) engine=mito;
2904 ";
2905 let result =
2906 ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default());
2907 assert!(result.is_err());
2908 assert_matches!(result, Err(crate::error::Error::InvalidTimeIndex { .. }));
2909
2910 let sql = r"create table demo(
2911 host string,
2912 ts timestamp time index,
2913 cpu float64 default 0,
2914 t timestamp,
2915 memory float64,
2916 TIME INDEX (t),
2917 PRIMARY KEY(ts, host)) engine=mito;
2918 ";
2919 let result =
2920 ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default());
2921 assert!(result.is_err());
2922 assert_matches!(result, Err(crate::error::Error::InvalidTimeIndex { .. }));
2923 }
2924
2925 #[test]
2926 fn test_invalid_column_name() {
2927 let sql = "create table foo(user string, i timestamp time index)";
2928 let result =
2929 ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default());
2930 let err = result.unwrap_err().output_msg();
2931 assert!(err.contains("Cannot use keyword 'user' as column name"));
2932
2933 let sql = r#"
2935 create table foo("user" string, i timestamp time index)
2936 "#;
2937 let result =
2938 ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default());
2939 let _ = result.unwrap();
2940 }
2941
2942 #[test]
2943 fn test_incorrect_default_value_issue_3479() {
2944 let sql = r#"CREATE TABLE `ExcePTuRi`(
2945non TIMESTAMP(6) TIME INDEX,
2946`iUSTO` DOUBLE DEFAULT 0.047318541668048164
2947)"#;
2948 let result =
2949 ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
2950 .unwrap();
2951 assert_eq!(1, result.len());
2952 match &result[0] {
2953 Statement::CreateTable(c) => {
2954 assert_eq!(
2955 "`iUSTO` DOUBLE DEFAULT 0.047318541668048164",
2956 c.columns[1].to_string()
2957 );
2958 }
2959 _ => unreachable!(),
2960 }
2961 }
2962
2963 #[test]
2964 fn test_parse_create_view() {
2965 let sql = "CREATE VIEW test AS SELECT * FROM NUMBERS";
2966
2967 let result =
2968 ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
2969 .unwrap();
2970 match &result[0] {
2971 Statement::CreateView(c) => {
2972 assert_eq!(c.to_string(), sql);
2973 assert!(!c.or_replace);
2974 assert!(!c.if_not_exists);
2975 assert_eq!("test", c.name.to_string());
2976 }
2977 _ => unreachable!(),
2978 }
2979
2980 let sql = "CREATE OR REPLACE VIEW IF NOT EXISTS test AS SELECT * FROM NUMBERS";
2981
2982 let result =
2983 ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
2984 .unwrap();
2985 match &result[0] {
2986 Statement::CreateView(c) => {
2987 assert_eq!(c.to_string(), sql);
2988 assert!(c.or_replace);
2989 assert!(c.if_not_exists);
2990 assert_eq!("test", c.name.to_string());
2991 }
2992 _ => unreachable!(),
2993 }
2994 }
2995
2996 #[test]
2997 fn test_parse_create_view_invalid_query() {
2998 let sql = "CREATE VIEW test AS DELETE from demo";
2999 let result =
3000 ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default());
3001 assert!(result.is_ok_and(|x| x.len() == 1));
3002 }
3003
3004 #[test]
3005 fn test_parse_create_table_fulltext_options() {
3006 let sql1 = r"
3007CREATE TABLE log (
3008 ts TIMESTAMP TIME INDEX,
3009 msg TEXT FULLTEXT INDEX,
3010)";
3011 let result1 = ParserContext::create_with_dialect(
3012 sql1,
3013 &GreptimeDbDialect {},
3014 ParseOptions::default(),
3015 )
3016 .unwrap();
3017
3018 if let Statement::CreateTable(c) = &result1[0] {
3019 c.columns.iter().for_each(|col| {
3020 if col.name().value == "msg" {
3021 assert!(
3022 col.extensions
3023 .fulltext_index_options
3024 .as_ref()
3025 .unwrap()
3026 .is_empty()
3027 );
3028 }
3029 });
3030 } else {
3031 panic!("should be create_table statement");
3032 }
3033
3034 let sql2 = r"
3035CREATE TABLE log (
3036 ts TIMESTAMP TIME INDEX,
3037 msg STRING FULLTEXT INDEX WITH (analyzer='English', case_sensitive='false')
3038)";
3039 let result2 = ParserContext::create_with_dialect(
3040 sql2,
3041 &GreptimeDbDialect {},
3042 ParseOptions::default(),
3043 )
3044 .unwrap();
3045
3046 if let Statement::CreateTable(c) = &result2[0] {
3047 c.columns.iter().for_each(|col| {
3048 if col.name().value == "msg" {
3049 let options = col.extensions.fulltext_index_options.as_ref().unwrap();
3050 assert_eq!(options.len(), 2);
3051 assert_eq!(options.get("analyzer").unwrap(), "English");
3052 assert_eq!(options.get("case_sensitive").unwrap(), "false");
3053 }
3054 });
3055 } else {
3056 panic!("should be create_table statement");
3057 }
3058
3059 let sql3 = r"
3060CREATE TABLE log (
3061 ts TIMESTAMP TIME INDEX,
3062 msg1 TINYTEXT FULLTEXT INDEX WITH (analyzer='English', case_sensitive='false'),
3063 msg2 CHAR(20) FULLTEXT INDEX WITH (analyzer='Chinese', case_sensitive='true')
3064)";
3065 let result3 = ParserContext::create_with_dialect(
3066 sql3,
3067 &GreptimeDbDialect {},
3068 ParseOptions::default(),
3069 )
3070 .unwrap();
3071
3072 if let Statement::CreateTable(c) = &result3[0] {
3073 c.columns.iter().for_each(|col| {
3074 if col.name().value == "msg1" {
3075 let options = col.extensions.fulltext_index_options.as_ref().unwrap();
3076 assert_eq!(options.len(), 2);
3077 assert_eq!(options.get("analyzer").unwrap(), "English");
3078 assert_eq!(options.get("case_sensitive").unwrap(), "false");
3079 } else if col.name().value == "msg2" {
3080 let options = col.extensions.fulltext_index_options.as_ref().unwrap();
3081 assert_eq!(options.len(), 2);
3082 assert_eq!(options.get("analyzer").unwrap(), "Chinese");
3083 assert_eq!(options.get("case_sensitive").unwrap(), "true");
3084 }
3085 });
3086 } else {
3087 panic!("should be create_table statement");
3088 }
3089 }
3090
3091 #[test]
3092 fn test_parse_create_table_fulltext_options_invalid_type() {
3093 let sql = r"
3094CREATE TABLE log (
3095 ts TIMESTAMP TIME INDEX,
3096 msg INT FULLTEXT INDEX,
3097)";
3098 let result =
3099 ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default());
3100 assert!(result.is_err());
3101 assert!(
3102 result
3103 .unwrap_err()
3104 .to_string()
3105 .contains("FULLTEXT index only supports string type")
3106 );
3107 }
3108
3109 #[test]
3110 fn test_parse_create_table_fulltext_options_duplicate() {
3111 let sql = r"
3112CREATE TABLE log (
3113 ts TIMESTAMP TIME INDEX,
3114 msg STRING FULLTEXT INDEX WITH (analyzer='English', analyzer='Chinese') FULLTEXT INDEX WITH (case_sensitive='false')
3115)";
3116 let result =
3117 ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default());
3118 assert!(result.is_err());
3119 assert!(
3120 result
3121 .unwrap_err()
3122 .to_string()
3123 .contains("duplicated FULLTEXT INDEX option")
3124 );
3125 }
3126
3127 #[test]
3128 fn test_parse_create_table_fulltext_options_invalid_option() {
3129 let sql = r"
3130CREATE TABLE log (
3131 ts TIMESTAMP TIME INDEX,
3132 msg STRING FULLTEXT INDEX WITH (analyzer='English', invalid_option='Chinese')
3133)";
3134 let result =
3135 ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default());
3136 assert!(result.is_err());
3137 assert!(
3138 result
3139 .unwrap_err()
3140 .to_string()
3141 .contains("invalid FULLTEXT INDEX option")
3142 );
3143 }
3144
3145 #[test]
3146 fn test_parse_create_table_skip_options() {
3147 let sql = r"
3148CREATE TABLE log (
3149 ts TIMESTAMP TIME INDEX,
3150 msg INT SKIPPING INDEX WITH (granularity='8192', type='bloom'),
3151)";
3152 let result =
3153 ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
3154 .unwrap();
3155
3156 if let Statement::CreateTable(c) = &result[0] {
3157 c.columns.iter().for_each(|col| {
3158 if col.name().value == "msg" {
3159 assert!(
3160 !col.extensions
3161 .skipping_index_options
3162 .as_ref()
3163 .unwrap()
3164 .is_empty()
3165 );
3166 }
3167 });
3168 } else {
3169 panic!("should be create_table statement");
3170 }
3171
3172 let sql = r"
3173 CREATE TABLE log (
3174 ts TIMESTAMP TIME INDEX,
3175 msg INT SKIPPING INDEX,
3176 )";
3177 let result =
3178 ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
3179 .unwrap();
3180
3181 if let Statement::CreateTable(c) = &result[0] {
3182 c.columns.iter().for_each(|col| {
3183 if col.name().value == "msg" {
3184 assert!(
3185 col.extensions
3186 .skipping_index_options
3187 .as_ref()
3188 .unwrap()
3189 .is_empty()
3190 );
3191 }
3192 });
3193 } else {
3194 panic!("should be create_table statement");
3195 }
3196 }
3197
3198 #[test]
3199 fn test_parse_create_view_with_columns() {
3200 let sql = "CREATE VIEW test () AS SELECT * FROM NUMBERS";
3201 let result =
3202 ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
3203 .unwrap();
3204
3205 match &result[0] {
3206 Statement::CreateView(c) => {
3207 assert_eq!(c.to_string(), "CREATE VIEW test AS SELECT * FROM NUMBERS");
3208 assert!(!c.or_replace);
3209 assert!(!c.if_not_exists);
3210 assert_eq!("test", c.name.to_string());
3211 }
3212 _ => unreachable!(),
3213 }
3214 assert_eq!(
3215 "CREATE VIEW test AS SELECT * FROM NUMBERS",
3216 result[0].to_string()
3217 );
3218
3219 let sql = "CREATE VIEW test (n1) AS SELECT * FROM NUMBERS";
3220 let result =
3221 ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
3222 .unwrap();
3223
3224 match &result[0] {
3225 Statement::CreateView(c) => {
3226 assert_eq!(c.to_string(), sql);
3227 assert!(!c.or_replace);
3228 assert!(!c.if_not_exists);
3229 assert_eq!("test", c.name.to_string());
3230 }
3231 _ => unreachable!(),
3232 }
3233 assert_eq!(sql, result[0].to_string());
3234
3235 let sql = "CREATE VIEW test (n1, n2) AS SELECT * FROM NUMBERS";
3236 let result =
3237 ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
3238 .unwrap();
3239
3240 match &result[0] {
3241 Statement::CreateView(c) => {
3242 assert_eq!(c.to_string(), sql);
3243 assert!(!c.or_replace);
3244 assert!(!c.if_not_exists);
3245 assert_eq!("test", c.name.to_string());
3246 }
3247 _ => unreachable!(),
3248 }
3249 assert_eq!(sql, result[0].to_string());
3250
3251 let sql = "CREATE VIEW test (n1 AS select * from demo";
3253 let result =
3254 ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default());
3255 assert!(result.is_err());
3256
3257 let sql = "CREATE VIEW test (n1, AS select * from demo";
3258 let result =
3259 ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default());
3260 assert!(result.is_err());
3261
3262 let sql = "CREATE VIEW test n1,n2) AS select * from demo";
3263 let result =
3264 ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default());
3265 assert!(result.is_err());
3266
3267 let sql = "CREATE VIEW test (1) AS select * from demo";
3268 let result =
3269 ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default());
3270 assert!(result.is_err());
3271
3272 let sql = "CREATE VIEW test (n1, select) AS select * from demo";
3274 let result =
3275 ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default());
3276 assert!(result.is_err());
3277 }
3278
3279 #[test]
3280 fn test_parse_column_extensions_vector() {
3281 let sql = "";
3283 let dialect = GenericDialect {};
3284 let mut tokenizer = Tokenizer::new(&dialect, sql);
3285 let tokens = tokenizer.tokenize().unwrap();
3286 let mut parser = Parser::new(&dialect).with_tokens(tokens);
3287 let name = Ident::new("vec_col");
3288 let data_type =
3289 DataType::Custom(vec![Ident::new("VECTOR")].into(), vec!["128".to_string()]);
3290 let mut extensions = ColumnExtensions::default();
3291
3292 let result =
3293 ParserContext::parse_column_extensions(&mut parser, &name, &data_type, &mut extensions);
3294 assert!(result.is_ok());
3295 assert!(extensions.vector_options.is_some());
3296 let vector_options = extensions.vector_options.unwrap();
3297 assert_eq!(vector_options.get(VECTOR_OPT_DIM), Some("128"));
3298 }
3299
3300 #[test]
3301 fn test_parse_column_extensions_vector_invalid() {
3302 let sql = "";
3304 let dialect = GenericDialect {};
3305 let mut tokenizer = Tokenizer::new(&dialect, sql);
3306 let tokens = tokenizer.tokenize().unwrap();
3307 let mut parser = Parser::new(&dialect).with_tokens(tokens);
3308 let name = Ident::new("vec_col");
3309 let data_type = DataType::Custom(vec![Ident::new("VECTOR")].into(), vec![]);
3310 let mut extensions = ColumnExtensions::default();
3311
3312 let result =
3313 ParserContext::parse_column_extensions(&mut parser, &name, &data_type, &mut extensions);
3314 assert!(result.is_err());
3315 }
3316
3317 #[test]
3318 fn test_parse_column_extensions_indices() {
3319 {
3321 let sql = "SKIPPING INDEX";
3322 let dialect = GenericDialect {};
3323 let mut tokenizer = Tokenizer::new(&dialect, sql);
3324 let tokens = tokenizer.tokenize().unwrap();
3325 let mut parser = Parser::new(&dialect).with_tokens(tokens);
3326 let name = Ident::new("col");
3327 let data_type = DataType::String(None);
3328 let mut extensions = ColumnExtensions::default();
3329 let result = ParserContext::parse_column_extensions(
3330 &mut parser,
3331 &name,
3332 &data_type,
3333 &mut extensions,
3334 );
3335 assert!(result.is_ok());
3336 assert!(extensions.skipping_index_options.is_some());
3337 }
3338
3339 {
3341 let sql = "FULLTEXT INDEX WITH (analyzer = 'English', case_sensitive = 'true')";
3342 let dialect = GenericDialect {};
3343 let mut tokenizer = Tokenizer::new(&dialect, sql);
3344 let tokens = tokenizer.tokenize().unwrap();
3345 let mut parser = Parser::new(&dialect).with_tokens(tokens);
3346 let name = Ident::new("text_col");
3347 let data_type = DataType::String(None);
3348 let mut extensions = ColumnExtensions::default();
3349 let result = ParserContext::parse_column_extensions(
3350 &mut parser,
3351 &name,
3352 &data_type,
3353 &mut extensions,
3354 );
3355 assert!(result.unwrap());
3356 assert!(extensions.fulltext_index_options.is_some());
3357 let fulltext_options = extensions.fulltext_index_options.unwrap();
3358 assert_eq!(fulltext_options.get("analyzer"), Some("English"));
3359 assert_eq!(fulltext_options.get("case_sensitive"), Some("true"));
3360 }
3361
3362 {
3364 let sql = "FULLTEXT INDEX WITH (analyzer = 'English')";
3365 let dialect = GenericDialect {};
3366 let mut tokenizer = Tokenizer::new(&dialect, sql);
3367 let tokens = tokenizer.tokenize().unwrap();
3368 let mut parser = Parser::new(&dialect).with_tokens(tokens);
3369 let name = Ident::new("num_col");
3370 let data_type = DataType::Int(None); let mut extensions = ColumnExtensions::default();
3372 let result = ParserContext::parse_column_extensions(
3373 &mut parser,
3374 &name,
3375 &data_type,
3376 &mut extensions,
3377 );
3378 assert!(result.is_err());
3379 assert!(
3380 result
3381 .unwrap_err()
3382 .to_string()
3383 .contains("FULLTEXT index only supports string type")
3384 );
3385 }
3386
3387 {
3389 let sql = "FULLTEXT INDEX WITH (analyzer = 'Invalid', case_sensitive = 'true')";
3390 let dialect = GenericDialect {};
3391 let mut tokenizer = Tokenizer::new(&dialect, sql);
3392 let tokens = tokenizer.tokenize().unwrap();
3393 let mut parser = Parser::new(&dialect).with_tokens(tokens);
3394 let name = Ident::new("text_col");
3395 let data_type = DataType::String(None);
3396 let mut extensions = ColumnExtensions::default();
3397 let result = ParserContext::parse_column_extensions(
3398 &mut parser,
3399 &name,
3400 &data_type,
3401 &mut extensions,
3402 );
3403 assert!(result.unwrap());
3404 }
3405
3406 {
3408 let sql = "INVERTED INDEX";
3409 let dialect = GenericDialect {};
3410 let mut tokenizer = Tokenizer::new(&dialect, sql);
3411 let tokens = tokenizer.tokenize().unwrap();
3412 let mut parser = Parser::new(&dialect).with_tokens(tokens);
3413 let name = Ident::new("col");
3414 let data_type = DataType::String(None);
3415 let mut extensions = ColumnExtensions::default();
3416 let result = ParserContext::parse_column_extensions(
3417 &mut parser,
3418 &name,
3419 &data_type,
3420 &mut extensions,
3421 );
3422 assert!(result.is_ok());
3423 assert!(extensions.inverted_index_options.is_some());
3424 }
3425
3426 {
3428 let sql = "INVERTED INDEX WITH (analyzer = 'English')";
3429 let dialect = GenericDialect {};
3430 let mut tokenizer = Tokenizer::new(&dialect, sql);
3431 let tokens = tokenizer.tokenize().unwrap();
3432 let mut parser = Parser::new(&dialect).with_tokens(tokens);
3433 let name = Ident::new("col");
3434 let data_type = DataType::String(None);
3435 let mut extensions = ColumnExtensions::default();
3436 let result = ParserContext::parse_column_extensions(
3437 &mut parser,
3438 &name,
3439 &data_type,
3440 &mut extensions,
3441 );
3442 assert!(result.is_err());
3443 assert!(
3444 result
3445 .unwrap_err()
3446 .to_string()
3447 .contains("INVERTED index doesn't support options")
3448 );
3449 }
3450
3451 {
3453 let sql = "SKIPPING INDEX FULLTEXT INDEX";
3454 let dialect = GenericDialect {};
3455 let mut tokenizer = Tokenizer::new(&dialect, sql);
3456 let tokens = tokenizer.tokenize().unwrap();
3457 let mut parser = Parser::new(&dialect).with_tokens(tokens);
3458 let name = Ident::new("col");
3459 let data_type = DataType::String(None);
3460 let mut extensions = ColumnExtensions::default();
3461 let result = ParserContext::parse_column_extensions(
3462 &mut parser,
3463 &name,
3464 &data_type,
3465 &mut extensions,
3466 );
3467 assert!(result.unwrap());
3468 assert!(extensions.skipping_index_options.is_some());
3469 assert!(extensions.fulltext_index_options.is_some());
3470 }
3471 }
3472
3473 #[test]
3474 fn test_reject_vector_index() {
3475 for index in ["VECTOR INDEX", "VECTOR INDEX WITH (metric = 'cosine')"] {
3476 let sql = format!(
3477 "CREATE TABLE vectors (ts TIMESTAMP TIME INDEX, embedding VECTOR(3) {index})"
3478 );
3479 assert!(
3480 ParserContext::create_with_dialect(
3481 &sql,
3482 &GreptimeDbDialect {},
3483 ParseOptions::default()
3484 )
3485 .is_err()
3486 );
3487 }
3488 }
3489
3490 #[test]
3491 fn test_parse_interval_cast() {
3492 let s = "select '10s'::INTERVAL";
3493 let stmts =
3494 ParserContext::create_with_dialect(s, &GreptimeDbDialect {}, ParseOptions::default())
3495 .unwrap();
3496 assert_eq!("SELECT '10 seconds'::INTERVAL", &stmts[0].to_string());
3497 }
3498}