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