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