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