Skip to main content

sql/parsers/
create_parser.rs

1// Copyright 2023 Greptime Team
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15mod 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
72// Preserve raw CREATE FLOW option entries until operator-side validation.
73// Do not use `OptionMap::new()` here: it can drop non-string values for
74// redacted keys before the flow option allowlist rejects them.
75fn 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
83/// Parses create [table] statement
84impl<'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    /// Parse `CREAVE VIEW` statement.
141    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                // allow a trailing comma, even though it's not in standard
176                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, // table id is assigned by catalog manager
294            partitions,
295        };
296
297        Ok(Statement::CreateTable(create_table))
298    }
299
300    /// "CREATE FLOW" clause
301    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        // make `SINK` case in-sensitive
307        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        // `EVAL OFFSET` is only legal together with `EVAL INTERVAL`. The phase
349        // offset must be a whole number of seconds; when an offset is present
350        // the interval must also be whole seconds (never silently truncated).
351        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            // Canonicalize a zero offset to `None` (the default epoch-anchored
382            // schedule) so that parse/display/reparse round-trips are stable.
383            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        // only accept sql or tql
447        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    /// Parse the interval expr to duration in seconds.
495    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    /// Parses an interval that must not contain months and returns the total
500    /// whole seconds together with whether the interval contains a sub-second
501    /// fraction. Whole seconds are computed by truncating the nanosecond part;
502    /// callers that require exact whole-second precision (e.g. `EVAL OFFSET`)
503    /// must reject `has_fractional_secs`.
504    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    /// Consumes an `EVAL <keyword>` token pair case-insensitively.
518    ///
519    /// `EVAL` is not a sqlparser keyword, so a plain
520    /// `consume_tokens([Token::make_keyword("EVAL"), ...])` comparison would be
521    /// case-sensitive on the word value and reject lower-case `eval interval`.
522    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    /// Parse interval expr to [`IntervalMonthDayNano`].
539    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    /// "PARTITION ON COLUMNS (...)" clause
584    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    /// Parses the "ON COLUMNS (...) (...)" part after "PARTITION".
593    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    /// Parse a comma-separated list wrapped by "()", and of which all items accepted by `F`
620    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    /// Parse the columns and constraints.
650    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                // allow a trailing comma, even though it's not in standard
671                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            // The timestamp type may be an alias type, we have to retrieve the actual type.
736            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(&not_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    /// Parse the column name and check if it's valid.
763    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" are sorted.
767            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" are sorted.
785            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            // Use a DialectSpecific option for time index
878            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    /// Parse a column option extensions.
896    ///
897    /// This function will handle:
898    /// - Vector type
899    /// - Indexes
900    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        // parse index options in column definition
932        let mut is_index_declared = false;
933
934        // skipping index
935        if let Token::Word(word) = parser.peek_token().token
936            && word.value.eq_ignore_ascii_case(SKIPPING)
937        {
938            parser.next_token();
939            // Consume `INDEX` keyword
940            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        // fulltext index
978        if parser.parse_keyword(Keyword::FULLTEXT) {
979            // Consume `INDEX` keyword
980            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        // inverted index
1029        if let Token::Word(word) = parser.peek_token().token
1030            && word.value.eq_ignore_ascii_case(INVERTED)
1031        {
1032            parser.next_token();
1033            // Consume `INDEX` keyword
1034            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            // inverted index doesn't have options, skipping `WITH`
1051            // try cache `WITH` and throw error
1052            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        // vector index
1071        if let Token::Word(word) = parser.peek_token().token
1072            && word.value.eq_ignore_ascii_case(VECTOR)
1073        {
1074            parser.next_token();
1075            // Consume `INDEX` keyword
1076            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            // Check that column is a vector type
1093            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    /// Parses the set of valid formats
1189    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    // It's safe to use time_index_constraints[0][0],
1232    // we already check the bound above.
1233    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
1279/// Ensure all exprs are binary expr and all the columns are defined in the column list.
1280fn ensure_exprs_are_binary(exprs: &[Expr], columns: &[&Column]) -> Result<()> {
1281    for expr in exprs {
1282        // The first level must be binary expr
1283        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
1296/// Check if the expr is a binary expr, an ident or a literal value.
1297/// If is ident, then check it is in the column list.
1298/// This recursive function is intended to be used by [ensure_exprs_are_binary].
1299fn 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
1331/// Ensure that all columns used in "PARTITION ON COLUMNS" are defined in create table.
1332fn 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            // Normally the columns in "create table" won't be too many,
1342            // a linear search to find the target every time is fine.
1343            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            /// Flow name
1677            pub flow_name: ObjectName,
1678            /// Output (sink) table name
1679            pub sink_table_name: ObjectName,
1680            /// Whether to replace existing task
1681            pub or_replace: bool,
1682            /// Create if not exist
1683            pub if_not_exists: bool,
1684            /// `EXPIRE AFTER`
1685            /// Duration in second as `i64`
1686            pub expire_after: Option<i64>,
1687            /// Comment string
1688            pub comment: Option<String>,
1689            /// Flow creation options
1690            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), // 10 minutes in seconds
1813                    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), // 2 hours in seconds
1834                    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                // ignore query parse result
1874                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            /// Flow name
1902            pub flow_name: ObjectName,
1903            /// Output (sink) table name
1904            pub sink_table_name: ObjectName,
1905            /// Whether to replace existing task
1906            pub or_replace: bool,
1907            /// Create if not exist
1908            pub if_not_exists: bool,
1909            /// `EXPIRE AFTER`
1910            /// Duration in second as `i64`
1911            pub expire_after: Option<i64>,
1912            /// Duration for flow evaluation interval
1913            /// Duration in seconds as `i64`
1914            /// If not set, flow will be evaluated based on time window size and other args.
1915            pub eval_interval: Option<i64>,
1916            /// Phase offset of the flow evaluation schedule within `eval_interval`.
1917            /// Duration in seconds as `i64`.
1918            pub eval_offset: Option<i64>,
1919            /// Comment string
1920            pub comment: Option<String>,
1921            /// Flow creation options
1922            pub flow_options: OptionMap,
1923        }
1924
1925        // create flow without `OR REPLACE`, `IF NOT EXISTS`, `EXPIRE AFTER` and `COMMENT`
1926        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                // ignore query parse result
2087                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        // zero offset is canonicalized to `None` and omitted on display
2143        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        // `TIME INDEX` should be in front of `PRIMARY KEY`
2596        // in order to equal the `TIMESTAMP TIME INDEX` constraint options vector
2597        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        // TIMESTAMP can be NULL which is not equal to above
2618        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        // BIGINT can't be time index any more
2640        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                // inverted index is merged into column options
2911                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        // If column name is quoted, it's valid even same with keyword.
2977        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        // Some invalid syntax cases
3295        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        // keyword
3316        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        // Test that vector options are parsed from data_type (no additional SQL needed)
3325        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        // Test that vector with no dimension fails
3346        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        // Test skipping index
3363        {
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        // Test fulltext index with options
3383        {
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        // Test fulltext index with invalid type (should fail)
3406        {
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); // Non-string type
3414            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        // Test fulltext index with invalid option (won't fail, the parser doesn't check the option's content)
3431        {
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        // Test inverted index
3450        {
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        // Test inverted index with options (should fail)
3470        {
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        // Test multiple indices
3495        {
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        // Test basic vector index
3528        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        // Test vector index with options
3554        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        // Test vector index on non-vector type (should fail)
3582        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        // Test duplicate vector index (should fail)
3601        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        // Test invalid option key (should fail)
3620        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        // Test vector index on vector type
3639        {
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            // First, parse the vector type to set vector_options
3649            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        // Test vector index on non-vector type (should fail)
3670        {
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); // Non-vector type
3678            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}