sql/statements/
create.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
15use std::collections::HashMap;
16use std::fmt::{Display, Formatter};
17
18use common_catalog::consts::FILE_ENGINE;
19use common_sql::default_constraint::parse_column_default_constraint;
20use datatypes::json::{JSON2_DEFAULT_MAX_AUTO_EXPANDED_PATHS, JsonSettings};
21use datatypes::prelude::ConcreteDataType;
22use datatypes::schema::{
23    ColumnDefaultConstraint, FulltextOptions, SkippingIndexOptions, VectorDistanceMetric,
24    VectorIndexEngineType, VectorIndexOptions,
25};
26use itertools::Itertools;
27use serde::Serialize;
28use snafu::ResultExt;
29use sqlparser::ast::{ColumnOption, ColumnOptionDef, DataType, Expr};
30use sqlparser_derive::{Visit, VisitMut};
31
32use crate::ast::{ColumnDef, Ident, ObjectName, Value as SqlValue};
33use crate::dialect::GreptimeDbDialect;
34use crate::error::{
35    InvalidFlowQuerySnafu, InvalidSqlSnafu, Result, SetFulltextOptionSnafu,
36    SetSkippingIndexOptionSnafu,
37};
38use crate::parser::ParserContext;
39use crate::statements::query::Query as GtQuery;
40use crate::statements::statement::Statement;
41use crate::statements::tql::Tql;
42use crate::statements::{OptionMap, sql_data_type_to_concrete_data_type, value_to_sql_value};
43
44const LINE_SEP: &str = ",\n";
45const COMMA_SEP: &str = ", ";
46const INDENT: usize = 2;
47pub const VECTOR_OPT_DIM: &str = "dim";
48
49macro_rules! format_indent {
50    ($fmt: expr, $arg: expr) => {
51        format!($fmt, format_args!("{: >1$}", "", INDENT), $arg)
52    };
53    ($arg: expr) => {
54        format_indent!("{}{}", $arg)
55    };
56}
57
58macro_rules! format_list_indent {
59    ($list: expr) => {
60        $list.iter().map(|e| format_indent!(e)).join(LINE_SEP)
61    };
62}
63
64macro_rules! format_list_comma {
65    ($list: expr) => {
66        $list.iter().map(|e| format!("{}", e)).join(COMMA_SEP)
67    };
68}
69
70#[cfg(feature = "enterprise")]
71pub mod trigger;
72
73fn format_table_constraint(constraints: &[TableConstraint]) -> String {
74    constraints.iter().map(|c| format_indent!(c)).join(LINE_SEP)
75}
76
77/// Table constraint for create table statement.
78#[derive(Debug, PartialEq, Eq, Clone, Visit, VisitMut, Serialize)]
79pub enum TableConstraint {
80    /// Primary key constraint.
81    PrimaryKey { columns: Vec<Ident> },
82    /// Time index constraint.
83    TimeIndex { column: Ident },
84}
85
86impl Display for TableConstraint {
87    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
88        match self {
89            TableConstraint::PrimaryKey { columns } => {
90                write!(f, "PRIMARY KEY ({})", format_list_comma!(columns))
91            }
92            TableConstraint::TimeIndex { column } => {
93                write!(f, "TIME INDEX ({})", column)
94            }
95        }
96    }
97}
98
99#[derive(Debug, PartialEq, Eq, Clone, Visit, VisitMut, Serialize)]
100pub struct CreateTable {
101    /// Create if not exists
102    pub if_not_exists: bool,
103    pub table_id: u32,
104    /// Table name
105    pub name: ObjectName,
106    pub columns: Vec<Column>,
107    pub engine: String,
108    pub constraints: Vec<TableConstraint>,
109    /// Table options in `WITH`. All keys are lowercase.
110    pub options: OptionMap,
111    pub partitions: Option<Partitions>,
112}
113
114/// Column definition in `CREATE TABLE` statement.
115#[derive(Debug, PartialEq, Eq, Clone, Visit, VisitMut, Serialize)]
116pub struct Column {
117    /// `ColumnDef` from `sqlparser::ast`
118    pub column_def: ColumnDef,
119    /// Column extensions for greptimedb dialect.
120    pub extensions: ColumnExtensions,
121}
122
123/// Column extensions for greptimedb dialect.
124#[derive(Debug, PartialEq, Eq, Clone, Visit, VisitMut, Default, Serialize)]
125pub struct ColumnExtensions {
126    /// Vector type options.
127    pub vector_options: Option<OptionMap>,
128
129    /// Fulltext index options.
130    pub fulltext_index_options: Option<OptionMap>,
131    /// Skipping index options.
132    pub skipping_index_options: Option<OptionMap>,
133    /// Inverted index options.
134    ///
135    /// Inverted index doesn't have options at present. There won't be any options in that map.
136    pub inverted_index_options: Option<OptionMap>,
137    /// Vector index options for HNSW-based vector similarity search.
138    pub vector_index_options: Option<OptionMap>,
139    /// JSON2-specific column options.
140    pub json2_options: Option<Json2Options>,
141}
142
143/// JSON2-specific options represented in the SQL AST.
144#[derive(Debug, PartialEq, Eq, Clone, Visit, VisitMut, Default, Serialize)]
145pub struct Json2Options {
146    /// Maximum number of unhinted JSON2 paths expanded into Arrow fields.
147    pub(crate) max_auto_expanded_paths: Option<u32>,
148    /// Paths stored as explicitly typed JSON2 fields.
149    pub(crate) type_hints: Vec<JsonTypeHint>,
150}
151
152impl Display for Json2Options {
153    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
154        let mut options = Vec::with_capacity(self.type_hints.len() + 1);
155        if let Some(max) = self.max_auto_expanded_paths {
156            options.push(format!("max_auto_expanded_paths = {max}"));
157        }
158        options.extend(self.type_hints.iter().map(format_json_type_hint));
159        write!(f, "(\n    {}\n  )", options.iter().join(",\n    "))
160    }
161}
162
163#[derive(Debug, PartialEq, Eq, Clone, Visit, VisitMut, Serialize)]
164pub struct JsonTypeHint {
165    pub path: Vec<String>,
166    pub data_type: DataType,
167    pub nullable: bool,
168    pub default: Option<Expr>,
169    pub inverted_index: bool,
170}
171
172impl Column {
173    pub fn name(&self) -> &Ident {
174        &self.column_def.name
175    }
176
177    pub fn data_type(&self) -> &DataType {
178        &self.column_def.data_type
179    }
180
181    pub fn mut_data_type(&mut self) -> &mut DataType {
182        &mut self.column_def.data_type
183    }
184
185    pub fn options(&self) -> &[ColumnOptionDef] {
186        &self.column_def.options
187    }
188
189    pub fn mut_options(&mut self) -> &mut Vec<ColumnOptionDef> {
190        &mut self.column_def.options
191    }
192}
193
194impl Display for Column {
195    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
196        if let Some(vector_options) = &self.extensions.vector_options
197            && let Some(dim) = vector_options.get(VECTOR_OPT_DIM)
198        {
199            write!(f, "{} VECTOR({})", self.column_def.name, dim)?;
200            return Ok(());
201        }
202
203        write!(f, "{} {}", self.column_def.name, self.column_def.data_type)?;
204        if let Some(options) = &self.extensions.json2_options {
205            write!(f, "{options}")?;
206        }
207        for option in &self.column_def.options {
208            write!(f, " {option}")?;
209        }
210
211        if let Some(fulltext_options) = &self.extensions.fulltext_index_options {
212            if !fulltext_options.is_empty() {
213                let options = fulltext_options.kv_pairs();
214                write!(f, " FULLTEXT INDEX WITH({})", format_list_comma!(options))?;
215            } else {
216                write!(f, " FULLTEXT INDEX")?;
217            }
218        }
219
220        if let Some(skipping_index_options) = &self.extensions.skipping_index_options {
221            if !skipping_index_options.is_empty() {
222                let options = skipping_index_options.kv_pairs();
223                write!(f, " SKIPPING INDEX WITH({})", format_list_comma!(options))?;
224            } else {
225                write!(f, " SKIPPING INDEX")?;
226            }
227        }
228
229        if let Some(inverted_index_options) = &self.extensions.inverted_index_options {
230            if !inverted_index_options.is_empty() {
231                let options = inverted_index_options.kv_pairs();
232                write!(f, " INVERTED INDEX WITH({})", format_list_comma!(options))?;
233            } else {
234                write!(f, " INVERTED INDEX")?;
235            }
236        }
237
238        if let Some(vector_index_options) = &self.extensions.vector_index_options {
239            if !vector_index_options.is_empty() {
240                let options = vector_index_options.kv_pairs();
241                write!(f, " VECTOR INDEX WITH({})", format_list_comma!(options))?;
242            } else {
243                write!(f, " VECTOR INDEX")?;
244            }
245        }
246        Ok(())
247    }
248}
249
250impl ColumnExtensions {
251    pub fn build_fulltext_options(&self) -> Result<Option<FulltextOptions>> {
252        let Some(options) = self.fulltext_index_options.as_ref() else {
253            return Ok(None);
254        };
255
256        let options: HashMap<String, String> = options.clone().into_map();
257        Ok(Some(options.try_into().context(SetFulltextOptionSnafu)?))
258    }
259
260    pub fn build_skipping_index_options(&self) -> Result<Option<SkippingIndexOptions>> {
261        let Some(options) = self.skipping_index_options.as_ref() else {
262            return Ok(None);
263        };
264
265        let options: HashMap<String, String> = options.clone().into_map();
266        Ok(Some(
267            options.try_into().context(SetSkippingIndexOptionSnafu)?,
268        ))
269    }
270
271    pub fn build_vector_index_options(&self) -> Result<Option<VectorIndexOptions>> {
272        let Some(options) = self.vector_index_options.as_ref() else {
273            return Ok(None);
274        };
275
276        let options_map: HashMap<String, String> = options.clone().into_map();
277        let mut result = VectorIndexOptions::default();
278
279        if let Some(s) = options_map.get("engine") {
280            result.engine = s.parse::<VectorIndexEngineType>().map_err(|e| {
281                InvalidSqlSnafu {
282                    msg: format!("invalid VECTOR INDEX engine: {e}"),
283                }
284                .build()
285            })?;
286        }
287
288        if let Some(s) = options_map.get("metric") {
289            result.metric = s.parse::<VectorDistanceMetric>().map_err(|e| {
290                InvalidSqlSnafu {
291                    msg: format!("invalid VECTOR INDEX metric: {e}"),
292                }
293                .build()
294            })?;
295        }
296
297        if let Some(s) = options_map.get("connectivity") {
298            let value = s.parse::<u32>().map_err(|_| {
299                InvalidSqlSnafu {
300                    msg: format!(
301                        "invalid VECTOR INDEX connectivity: {s}, expected positive integer"
302                    ),
303                }
304                .build()
305            })?;
306            if !(2..=2048).contains(&value) {
307                return InvalidSqlSnafu {
308                    msg: "VECTOR INDEX connectivity must be in the range [2, 2048].".to_string(),
309                }
310                .fail();
311            }
312            result.connectivity = value;
313        }
314
315        if let Some(s) = options_map.get("expansion_add") {
316            let value = s.parse::<u32>().map_err(|_| {
317                InvalidSqlSnafu {
318                    msg: format!(
319                        "invalid VECTOR INDEX expansion_add: {s}, expected positive integer"
320                    ),
321                }
322                .build()
323            })?;
324            if value == 0 {
325                return InvalidSqlSnafu {
326                    msg: "VECTOR INDEX expansion_add must be greater than 0".to_string(),
327                }
328                .fail();
329            }
330            result.expansion_add = value;
331        }
332
333        if let Some(s) = options_map.get("expansion_search") {
334            let value = s.parse::<u32>().map_err(|_| {
335                InvalidSqlSnafu {
336                    msg: format!(
337                        "invalid VECTOR INDEX expansion_search: {s}, expected positive integer"
338                    ),
339                }
340                .build()
341            })?;
342            if value == 0 {
343                return InvalidSqlSnafu {
344                    msg: "VECTOR INDEX expansion_search must be greater than 0".to_string(),
345                }
346                .fail();
347            }
348            result.expansion_search = value;
349        }
350
351        Ok(Some(result))
352    }
353
354    pub fn build_json_settings(&self) -> Result<Option<JsonSettings>> {
355        let Some(options) = &self.json2_options else {
356            return Ok(None);
357        };
358
359        let type_hints = options
360            .type_hints
361            .iter()
362            .map(|hint| {
363                Ok(datatypes::json::JsonTypeHint {
364                    path: hint.path.clone(),
365                    data_type: json_type_hint_concrete_data_type(&hint.data_type)?,
366                    nullable: hint.nullable,
367                    default_constraint: build_json_type_hint_default_constraint(hint)?,
368                    inverted_index: hint.inverted_index,
369                })
370            })
371            .collect::<Result<Vec<_>>>()?;
372        let settings = JsonSettings::try_new(
373            type_hints,
374            options
375                .max_auto_expanded_paths
376                .or(Some(JSON2_DEFAULT_MAX_AUTO_EXPANDED_PATHS)),
377        )?;
378        Ok(Some(settings))
379    }
380
381    pub fn set_json_settings(&mut self, settings: JsonSettings) -> Result<()> {
382        let (type_hints, max_auto_expanded_paths) = settings.into_parts();
383        let type_hints = type_hints
384            .into_iter()
385            .map(|hint| {
386                let data_type = json_type_hint_sql_data_type(&hint.data_type)?;
387                let default = hint
388                    .default_constraint
389                    .map(|constraint| column_default_constraint_to_expr(&constraint))
390                    .transpose()?;
391                Ok(JsonTypeHint {
392                    path: hint.path,
393                    data_type,
394                    nullable: hint.nullable,
395                    default,
396                    inverted_index: hint.inverted_index,
397                })
398            })
399            .collect::<Result<Vec<_>>>()?;
400        self.json2_options = (max_auto_expanded_paths.is_some() || !type_hints.is_empty())
401            .then_some(Json2Options {
402                max_auto_expanded_paths,
403                type_hints,
404            });
405        Ok(())
406    }
407}
408
409fn build_json_type_hint_default_constraint(
410    hint: &JsonTypeHint,
411) -> Result<Option<ColumnDefaultConstraint>> {
412    let Some(default) = &hint.default else {
413        return Ok(None);
414    };
415
416    let data_type = json_type_hint_concrete_data_type(&hint.data_type)?;
417    let opts = [ColumnOptionDef {
418        name: None,
419        option: ColumnOption::Default(default.clone()),
420    }];
421
422    // Use the JSON path as the column name context for default value parsing errors.
423    let json_path = hint.path.join(".");
424    let default_constraint = parse_column_default_constraint(&json_path, &data_type, &opts, None)
425        .context(crate::error::SqlCommonSnafu)?;
426
427    if let Some(constraint) = &default_constraint {
428        constraint
429            .validate(&data_type, hint.nullable)
430            .map_err(|e| {
431                InvalidSqlSnafu {
432                    msg: format!("invalid DEFAULT for JSON2 type hint '{}': {e}", json_path),
433                }
434                .build()
435            })?;
436    }
437
438    Ok(default_constraint)
439}
440
441fn json_type_hint_concrete_data_type(data_type: &DataType) -> Result<ConcreteDataType> {
442    let data_type = sql_data_type_to_concrete_data_type(data_type)?;
443    normalize_json_type_hint_concrete_data_type(&data_type)
444}
445
446fn normalize_json_type_hint_concrete_data_type(
447    data_type: &ConcreteDataType,
448) -> Result<ConcreteDataType> {
449    let normalized = match data_type {
450        ConcreteDataType::String(_) => ConcreteDataType::string_datatype(),
451        ConcreteDataType::Int8(_)
452        | ConcreteDataType::Int16(_)
453        | ConcreteDataType::Int32(_)
454        | ConcreteDataType::Int64(_) => ConcreteDataType::int64_datatype(),
455        ConcreteDataType::UInt8(_)
456        | ConcreteDataType::UInt16(_)
457        | ConcreteDataType::UInt32(_)
458        | ConcreteDataType::UInt64(_) => ConcreteDataType::uint64_datatype(),
459        ConcreteDataType::Float32(_) | ConcreteDataType::Float64(_) => {
460            ConcreteDataType::float64_datatype()
461        }
462        ConcreteDataType::Boolean(_) => ConcreteDataType::boolean_datatype(),
463        _ => {
464            return InvalidSqlSnafu {
465                msg: format!("unsupported JSON2 type hint data type: {data_type}"),
466            }
467            .fail();
468        }
469    };
470    Ok(normalized)
471}
472
473fn json_type_hint_sql_data_type(data_type: &ConcreteDataType) -> Result<DataType> {
474    let data_type = normalize_json_type_hint_concrete_data_type(data_type)?;
475    let sql_type = match data_type {
476        ConcreteDataType::String(_) => DataType::String(None),
477        ConcreteDataType::Int64(_) => DataType::BigInt(None),
478        ConcreteDataType::UInt64(_) => DataType::BigIntUnsigned(None),
479        ConcreteDataType::Float64(_) => DataType::Double(sqlparser::ast::ExactNumberInfo::None),
480        ConcreteDataType::Boolean(_) => DataType::Boolean,
481        _ => unreachable!("JSON2 type hint data type should have been normalized"),
482    };
483    Ok(sql_type)
484}
485
486fn column_default_constraint_to_expr(constraint: &ColumnDefaultConstraint) -> Result<Expr> {
487    match constraint {
488        ColumnDefaultConstraint::Value(value) => Ok(Expr::Value(value_to_sql_value(value)?.into())),
489        ColumnDefaultConstraint::Function(function) => {
490            ParserContext::parse_function(function, &GreptimeDbDialect {})
491        }
492    }
493}
494
495fn format_json_type_hint(hint: &JsonTypeHint) -> String {
496    let path = hint
497        .path
498        .iter()
499        .map(|segment| format_json_path_segment(segment))
500        .join(".");
501    let nullability = if hint.nullable { " NULL" } else { " NOT NULL" };
502    let default = hint
503        .default
504        .as_ref()
505        .map(|expr| format!(" DEFAULT {expr}"))
506        .unwrap_or_default();
507    let inverted_index = if hint.inverted_index {
508        " INVERTED INDEX"
509    } else {
510        ""
511    };
512    format!(
513        "{} {}{}{}{}",
514        path, hint.data_type, nullability, default, inverted_index
515    )
516}
517
518fn format_json_path_segment(segment: &str) -> String {
519    format!("\"{}\"", segment.replace('"', "\"\""))
520}
521
522/// Partition on columns or values.
523///
524/// - `column_list` is the list of columns in `PARTITION ON COLUMNS` clause.
525/// - `exprs` is the list of expressions in `PARTITION ON VALUES` clause, like
526///   `host <= 'host1'`, `host > 'host1' and host <= 'host2'` or `host > 'host2'`.
527///   Each expression stands for a partition.
528#[derive(Debug, PartialEq, Eq, Clone, Visit, VisitMut, Serialize)]
529pub struct Partitions {
530    pub column_list: Vec<Ident>,
531    pub exprs: Vec<Expr>,
532}
533
534impl Partitions {
535    /// set quotes to all [Ident]s from column list
536    pub fn set_quote(&mut self, quote_style: char) {
537        self.column_list
538            .iter_mut()
539            .for_each(|c| c.quote_style = Some(quote_style));
540    }
541}
542
543#[derive(Debug, PartialEq, Eq, Clone, Visit, VisitMut)]
544pub struct PartitionEntry {
545    pub name: Ident,
546    pub value_list: Vec<SqlValue>,
547}
548
549impl Display for PartitionEntry {
550    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
551        write!(
552            f,
553            "PARTITION {} VALUES LESS THAN ({})",
554            self.name,
555            format_list_comma!(self.value_list),
556        )
557    }
558}
559
560impl Display for Partitions {
561    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
562        if !self.column_list.is_empty() {
563            write!(
564                f,
565                "PARTITION ON COLUMNS ({}) (\n{}\n)",
566                format_list_comma!(self.column_list),
567                format_list_indent!(self.exprs),
568            )?;
569        }
570        Ok(())
571    }
572}
573
574impl Display for CreateTable {
575    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
576        write!(f, "CREATE ")?;
577        if self.engine == FILE_ENGINE {
578            write!(f, "EXTERNAL ")?;
579        }
580        write!(f, "TABLE ")?;
581        if self.if_not_exists {
582            write!(f, "IF NOT EXISTS ")?;
583        }
584        writeln!(f, "{} (", &self.name)?;
585        writeln!(f, "{},", format_list_indent!(self.columns))?;
586        writeln!(f, "{}", format_table_constraint(&self.constraints))?;
587        writeln!(f, ")")?;
588        if let Some(partitions) = &self.partitions {
589            writeln!(f, "{partitions}")?;
590        }
591        writeln!(f, "ENGINE={}", &self.engine)?;
592        if !self.options.is_empty() {
593            let options = self.options.kv_pairs();
594            write!(f, "WITH(\n{}\n)", format_list_indent!(options))?;
595        }
596        Ok(())
597    }
598}
599
600#[derive(Debug, PartialEq, Eq, Clone, Visit, VisitMut, Serialize)]
601pub struct CreateDatabase {
602    pub name: ObjectName,
603    /// Create if not exists
604    pub if_not_exists: bool,
605    pub options: OptionMap,
606}
607
608impl CreateDatabase {
609    /// Creates a statement for `CREATE DATABASE`
610    pub fn new(name: ObjectName, if_not_exists: bool, options: OptionMap) -> Self {
611        Self {
612            name,
613            if_not_exists,
614            options,
615        }
616    }
617}
618
619impl Display for CreateDatabase {
620    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
621        write!(f, "CREATE DATABASE ")?;
622        if self.if_not_exists {
623            write!(f, "IF NOT EXISTS ")?;
624        }
625        write!(f, "{}", &self.name)?;
626        if !self.options.is_empty() {
627            let options = self.options.kv_pairs();
628            write!(f, "\nWITH(\n{}\n)", format_list_indent!(options))?;
629        }
630        Ok(())
631    }
632}
633
634#[derive(Debug, PartialEq, Eq, Clone, Visit, VisitMut, Serialize)]
635pub struct CreateExternalTable {
636    /// Table name
637    pub name: ObjectName,
638    pub columns: Vec<Column>,
639    pub constraints: Vec<TableConstraint>,
640    /// Table options in `WITH`. All keys are lowercase.
641    pub options: OptionMap,
642    pub if_not_exists: bool,
643    pub engine: String,
644}
645
646impl Display for CreateExternalTable {
647    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
648        write!(f, "CREATE EXTERNAL TABLE ")?;
649        if self.if_not_exists {
650            write!(f, "IF NOT EXISTS ")?;
651        }
652        writeln!(f, "{} (", &self.name)?;
653        writeln!(f, "{},", format_list_indent!(self.columns))?;
654        writeln!(f, "{}", format_table_constraint(&self.constraints))?;
655        writeln!(f, ")")?;
656        writeln!(f, "ENGINE={}", &self.engine)?;
657        if !self.options.is_empty() {
658            let options = self.options.kv_pairs();
659            write!(f, "WITH(\n{}\n)", format_list_indent!(options))?;
660        }
661        Ok(())
662    }
663}
664
665#[derive(Debug, PartialEq, Eq, Clone, Visit, VisitMut, Serialize)]
666pub struct CreateTableLike {
667    /// Table name
668    pub table_name: ObjectName,
669    /// The table that is designated to be imitated by `Like`
670    pub source_name: ObjectName,
671}
672
673impl Display for CreateTableLike {
674    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
675        let table_name = &self.table_name;
676        let source_name = &self.source_name;
677        write!(f, r#"CREATE TABLE {table_name} LIKE {source_name}"#)
678    }
679}
680
681#[derive(Debug, PartialEq, Eq, Clone, Visit, VisitMut, Serialize)]
682pub struct CreateFlow {
683    /// Flow name
684    pub flow_name: ObjectName,
685    /// Output (sink) table name
686    pub sink_table_name: ObjectName,
687    /// Whether to replace existing task
688    pub or_replace: bool,
689    /// Create if not exist
690    pub if_not_exists: bool,
691    /// `EXPIRE AFTER`
692    /// Duration in second as `i64`
693    pub expire_after: Option<i64>,
694    /// Duration for flow evaluation interval
695    /// Duration in seconds as `i64`
696    /// If not set, flow will be evaluated based on time window size and other args.
697    pub eval_interval: Option<i64>,
698    /// Phase offset of the flow evaluation schedule within `eval_interval`.
699    /// Duration in seconds as `i64`.
700    /// Must be in range `[0, eval_interval)`. Only legal together with
701    /// `eval_interval`. A value of zero (the default) means the schedule is
702    /// anchored to the Unix epoch, i.e. phases at `k * eval_interval`.
703    pub eval_offset: Option<i64>,
704    /// Comment string
705    pub comment: Option<String>,
706    /// Flow creation options from `WITH (...)`
707    pub flow_options: OptionMap,
708    /// SQL statement
709    pub query: Box<SqlOrTql>,
710}
711
712/// Either a sql query or a tql query
713#[derive(Debug, PartialEq, Eq, Clone, Visit, VisitMut, Serialize)]
714pub enum SqlOrTql {
715    Sql(GtQuery, String),
716    Tql(Tql, String),
717}
718
719impl std::fmt::Display for SqlOrTql {
720    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
721        match self {
722            Self::Sql(_, s) => write!(f, "{}", s),
723            Self::Tql(_, s) => write!(f, "{}", s),
724        }
725    }
726}
727
728impl SqlOrTql {
729    pub fn try_from_statement(
730        value: Statement,
731        original_query: &str,
732    ) -> std::result::Result<Self, crate::error::Error> {
733        match value {
734            Statement::Query(query) => Ok(Self::Sql(*query, original_query.to_string())),
735            Statement::Tql(tql) => Ok(Self::Tql(tql, original_query.to_string())),
736            _ => InvalidFlowQuerySnafu {
737                reason: format!("Expect either sql query or promql query, found {:?}", value),
738            }
739            .fail(),
740        }
741    }
742}
743
744impl Display for CreateFlow {
745    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
746        write!(f, "CREATE ")?;
747        if self.or_replace {
748            write!(f, "OR REPLACE ")?;
749        }
750        write!(f, "FLOW ")?;
751        if self.if_not_exists {
752            write!(f, "IF NOT EXISTS ")?;
753        }
754        writeln!(f, "{}", &self.flow_name)?;
755        writeln!(f, "SINK TO {}", &self.sink_table_name)?;
756        if let Some(expire_after) = &self.expire_after {
757            writeln!(f, "EXPIRE AFTER '{} s'", expire_after)?;
758        }
759        if let Some(eval_interval) = &self.eval_interval {
760            writeln!(f, "EVAL INTERVAL '{} s'", eval_interval)?;
761        }
762        // Canonical display: omit a zero offset (equivalent to the default
763        // epoch-anchored schedule). Non-zero offsets are always emitted.
764        if let Some(eval_offset) = &self.eval_offset
765            && *eval_offset != 0
766        {
767            writeln!(f, "EVAL OFFSET '{} s'", eval_offset)?;
768        }
769        if let Some(comment) = &self.comment {
770            writeln!(f, "COMMENT '{}'", comment)?;
771        }
772        if !self.flow_options.is_empty() {
773            let options = self.flow_options.kv_pairs();
774            writeln!(f, "WITH ({})", format_list_comma!(options))?;
775        }
776        write!(f, "AS {}", &self.query)
777    }
778}
779
780/// Create SQL view statement.
781#[derive(Debug, PartialEq, Eq, Clone, Visit, VisitMut, Serialize)]
782pub struct CreateView {
783    /// View name
784    pub name: ObjectName,
785    /// An optional list of names to be used for columns of the view
786    pub columns: Vec<Ident>,
787    /// The clause after `As` that defines the VIEW.
788    /// Can only be either [Statement::Query] or [Statement::Tql].
789    pub query: Box<Statement>,
790    /// Whether to replace existing VIEW
791    pub or_replace: bool,
792    /// Create VIEW only when it doesn't exists
793    pub if_not_exists: bool,
794}
795
796impl Display for CreateView {
797    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
798        write!(f, "CREATE ")?;
799        if self.or_replace {
800            write!(f, "OR REPLACE ")?;
801        }
802        write!(f, "VIEW ")?;
803        if self.if_not_exists {
804            write!(f, "IF NOT EXISTS ")?;
805        }
806        write!(f, "{} ", &self.name)?;
807        if !self.columns.is_empty() {
808            write!(f, "({}) ", format_list_comma!(self.columns))?;
809        }
810        write!(f, "AS {}", &self.query)
811    }
812}
813
814#[cfg(test)]
815mod tests {
816    use std::assert_matches;
817
818    use datatypes::json::{JsonSettings, JsonTypeHint as DatatypeJsonTypeHint};
819    use datatypes::prelude::ConcreteDataType;
820    use datatypes::schema::ColumnDefaultConstraint;
821    use datatypes::value::Value;
822
823    use super::*;
824    use crate::dialect::GreptimeDbDialect;
825    use crate::error::Error;
826    use crate::parser::{ParseOptions, ParserContext};
827    use crate::statements::statement::Statement;
828
829    #[test]
830    fn test_display_create_table() {
831        let sql = r"create table if not exists demo(
832                             host string,
833                             ts timestamp,
834                             cpu double default 0,
835                             memory double,
836                             TIME INDEX (ts),
837                             PRIMARY KEY(host)
838                       )
839                       PARTITION ON COLUMNS (host) (
840                            host = 'a',
841                            host > 'a',
842                       )
843                       engine=mito
844                       with(ttl='7d', storage='File');
845         ";
846        let result =
847            ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
848                .unwrap();
849        assert_eq!(1, result.len());
850
851        match &result[0] {
852            Statement::CreateTable(c) => {
853                let new_sql = format!("\n{}", c);
854                assert_eq!(
855                    r#"
856CREATE TABLE IF NOT EXISTS demo (
857  host STRING,
858  ts TIMESTAMP,
859  cpu DOUBLE DEFAULT 0,
860  memory DOUBLE,
861  TIME INDEX (ts),
862  PRIMARY KEY (host)
863)
864PARTITION ON COLUMNS (host) (
865  host = 'a',
866  host > 'a'
867)
868ENGINE=mito
869WITH(
870  storage = 'File',
871  ttl = '7d'
872)"#,
873                    &new_sql
874                );
875
876                let new_result = ParserContext::create_with_dialect(
877                    &new_sql,
878                    &GreptimeDbDialect {},
879                    ParseOptions::default(),
880                )
881                .unwrap();
882                assert_eq!(result, new_result);
883            }
884            _ => unreachable!(),
885        }
886    }
887
888    #[test]
889    fn test_display_empty_partition_column() {
890        let sql = r"create table if not exists demo(
891            host string,
892            ts timestamp,
893            cpu double default 0,
894            memory double,
895            TIME INDEX (ts),
896            PRIMARY KEY(ts, host)
897            );
898        ";
899        let result =
900            ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
901                .unwrap();
902        assert_eq!(1, result.len());
903
904        match &result[0] {
905            Statement::CreateTable(c) => {
906                let new_sql = format!("\n{}", c);
907                assert_eq!(
908                    r#"
909CREATE TABLE IF NOT EXISTS demo (
910  host STRING,
911  ts TIMESTAMP,
912  cpu DOUBLE DEFAULT 0,
913  memory DOUBLE,
914  TIME INDEX (ts),
915  PRIMARY KEY (ts, host)
916)
917ENGINE=mito
918"#,
919                    &new_sql
920                );
921
922                let new_result = ParserContext::create_with_dialect(
923                    &new_sql,
924                    &GreptimeDbDialect {},
925                    ParseOptions::default(),
926                )
927                .unwrap();
928                assert_eq!(result, new_result);
929            }
930            _ => unreachable!(),
931        }
932    }
933
934    #[test]
935    fn test_validate_table_options() {
936        let sql = r"create table if not exists demo(
937            host string,
938            ts timestamp,
939            cpu double default 0,
940            memory double,
941            TIME INDEX (ts),
942            PRIMARY KEY(host)
943      )
944      PARTITION ON COLUMNS (host) ()
945      engine=mito
946      with(ttl='7d', 'compaction.type'='world');
947";
948        let result =
949            ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
950                .unwrap();
951        match &result[0] {
952            Statement::CreateTable(c) => {
953                assert_eq!(2, c.options.len());
954            }
955            _ => unreachable!(),
956        }
957
958        let sql = r"create table if not exists demo(
959            host string,
960            ts timestamp,
961            cpu double default 0,
962            memory double,
963            TIME INDEX (ts),
964            PRIMARY KEY(host)
965      )
966      PARTITION ON COLUMNS (host) ()
967      engine=mito
968      with(ttl='7d', hello='world');
969";
970        let result =
971            ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default());
972        assert_matches!(result, Err(Error::InvalidTableOption { .. }));
973
974        // A whitelisted semantic key with an in-domain value is accepted.
975        let semantic = |with: &str| {
976            let sql =
977                format!("create table demo(host string, ts timestamp time index) with({with});");
978            ParserContext::create_with_dialect(&sql, &GreptimeDbDialect {}, ParseOptions::default())
979        };
980        assert!(semantic("'greptime.semantic.signal_type'='metric'").is_ok());
981        // An out-of-domain value is rejected.
982        assert_matches!(
983            semantic("'greptime.semantic.signal_type'='spans'"),
984            Err(Error::InvalidTableOption { .. })
985        );
986        // An unknown key under the semantic prefix is rejected.
987        assert_matches!(
988            semantic("'greptime.semantic.bogus'='x'"),
989            Err(Error::InvalidTableOption { .. })
990        );
991    }
992
993    #[test]
994    fn test_display_json2_type_hints_quotes_path_segments() {
995        let sql = r#"CREATE TABLE traces (
996            log_json_data JSON2 (
997                "service.name" STRING,
998                "a.b"."c" INT64 NOT NULL,
999                a."b.c" STRING
1000            ),
1001            ts TIMESTAMP TIME INDEX
1002        )"#;
1003        let result =
1004            ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
1005                .unwrap();
1006
1007        match &result[0] {
1008            Statement::CreateTable(c) => {
1009                let new_sql = format!("\n{}", c);
1010                assert_eq!(
1011                    r#"
1012CREATE TABLE traces (
1013  log_json_data JSON2(
1014    "service.name" STRING NULL,
1015    "a.b"."c" BIGINT NOT NULL,
1016    "a"."b.c" STRING NULL
1017  ),
1018  ts TIMESTAMP NOT NULL,
1019  TIME INDEX (ts)
1020)
1021ENGINE=mito
1022"#,
1023                    &new_sql
1024                );
1025
1026                let new_result = ParserContext::create_with_dialect(
1027                    &new_sql,
1028                    &GreptimeDbDialect {},
1029                    ParseOptions::default(),
1030                )
1031                .unwrap();
1032                assert_eq!(result, new_result);
1033            }
1034            _ => unreachable!(),
1035        }
1036    }
1037
1038    #[test]
1039    fn test_parse_json2_max_auto_expanded_paths_option() -> Result<()> {
1040        let sql = r#"CREATE TABLE traces (
1041            log_json_data JSON2 (
1042                status_code INT64 NOT NULL,
1043                max_auto_expanded_paths = 1
1044            ),
1045            ts TIMESTAMP TIME INDEX
1046        )"#;
1047        let result = ParserContext::create_with_dialect(
1048            sql,
1049            &GreptimeDbDialect {},
1050            ParseOptions::default(),
1051        )?;
1052        let Statement::CreateTable(create_table) = &result[0] else {
1053            unreachable!()
1054        };
1055        let settings = create_table.columns[0]
1056            .extensions
1057            .build_json_settings()?
1058            .unwrap();
1059        assert_eq!(settings.max_auto_expanded_paths(), Some(1));
1060        Ok(())
1061    }
1062
1063    #[test]
1064    fn test_display_json2_type_hints_quotes_numeric_segments() {
1065        let sql = r#"CREATE TABLE traces (
1066            log_json_data JSON2 (
1067                "1abc" STRING,
1068                a."2b" INT64 NOT NULL
1069            ),
1070            ts TIMESTAMP TIME INDEX
1071        )"#;
1072        let result =
1073            ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
1074                .unwrap();
1075
1076        match &result[0] {
1077            Statement::CreateTable(c) => {
1078                let new_sql = format!("\n{}", c);
1079                assert_eq!(
1080                    r#"
1081CREATE TABLE traces (
1082  log_json_data JSON2(
1083    "1abc" STRING NULL,
1084    "a"."2b" BIGINT NOT NULL
1085  ),
1086  ts TIMESTAMP NOT NULL,
1087  TIME INDEX (ts)
1088)
1089ENGINE=mito
1090"#,
1091                    &new_sql
1092                );
1093
1094                let new_result = ParserContext::create_with_dialect(
1095                    &new_sql,
1096                    &GreptimeDbDialect {},
1097                    ParseOptions::default(),
1098                )
1099                .unwrap();
1100                assert_eq!(result, new_result);
1101            }
1102            _ => unreachable!(),
1103        }
1104    }
1105
1106    #[test]
1107    fn test_json2_type_hint_default_builds_default_constraint() {
1108        let sql = r#"CREATE TABLE traces (
1109            log_json_data JSON2 (
1110                status_code INT64 DEFAULT -5,
1111                duration FLOAT64 DEFAULT +1.5,
1112                error BOOLEAN DEFAULT false,
1113                message STRING DEFAULT 'unknown'
1114            ),
1115            ts TIMESTAMP TIME INDEX
1116        )"#;
1117        let result =
1118            ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
1119                .unwrap();
1120
1121        let Statement::CreateTable(create_table) = &result[0] else {
1122            unreachable!()
1123        };
1124        let settings = create_table.columns[0]
1125            .extensions
1126            .build_json_settings()
1127            .unwrap()
1128            .unwrap();
1129        let hints = settings.type_hints();
1130
1131        assert_eq!(hints[0].data_type, ConcreteDataType::int64_datatype());
1132        assert_eq!(
1133            hints[0].default_constraint,
1134            Some(ColumnDefaultConstraint::Value(Value::Int64(-5)))
1135        );
1136        assert_eq!(hints[1].data_type, ConcreteDataType::float64_datatype());
1137        assert_eq!(
1138            hints[1].default_constraint,
1139            Some(ColumnDefaultConstraint::Value(Value::Float64(1.5.into())))
1140        );
1141        assert_eq!(hints[2].data_type, ConcreteDataType::boolean_datatype());
1142        assert_eq!(
1143            hints[2].default_constraint,
1144            Some(ColumnDefaultConstraint::Value(Value::Boolean(false)))
1145        );
1146        assert_eq!(hints[3].data_type, ConcreteDataType::string_datatype());
1147        assert_eq!(
1148            hints[3].default_constraint,
1149            Some(ColumnDefaultConstraint::Value(Value::String(
1150                "unknown".into()
1151            )))
1152        );
1153    }
1154
1155    #[test]
1156    fn test_json2_type_hint_not_null_default_null_is_rejected() {
1157        let sql = r#"CREATE TABLE traces (
1158            log_json_data JSON2 (
1159                status_code INT64 NOT NULL DEFAULT NULL
1160            ),
1161            ts TIMESTAMP TIME INDEX
1162        )"#;
1163        let result =
1164            ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
1165                .unwrap();
1166
1167        let Statement::CreateTable(create_table) = &result[0] else {
1168            unreachable!()
1169        };
1170        let err = create_table.columns[0]
1171            .extensions
1172            .build_json_settings()
1173            .unwrap_err();
1174        assert!(
1175            err.to_string()
1176                .contains("Default value should not be null for non null column")
1177        );
1178    }
1179
1180    #[test]
1181    fn test_set_json_settings_normalizes_type_hint_sql_types() -> Result<()> {
1182        let mut extensions = super::ColumnExtensions::default();
1183        let settings = JsonSettings::try_new(
1184            vec![
1185                DatatypeJsonTypeHint {
1186                    path: vec!["i".to_string()],
1187                    data_type: ConcreteDataType::int32_datatype(),
1188                    nullable: true,
1189                    default_constraint: None,
1190                    inverted_index: false,
1191                },
1192                DatatypeJsonTypeHint {
1193                    path: vec!["f".to_string()],
1194                    data_type: ConcreteDataType::float32_datatype(),
1195                    nullable: true,
1196                    default_constraint: None,
1197                    inverted_index: false,
1198                },
1199                DatatypeJsonTypeHint {
1200                    path: vec!["u".to_string()],
1201                    data_type: ConcreteDataType::uint32_datatype(),
1202                    nullable: true,
1203                    default_constraint: None,
1204                    inverted_index: false,
1205                },
1206                DatatypeJsonTypeHint {
1207                    path: vec!["s".to_string()],
1208                    data_type: ConcreteDataType::string_datatype(),
1209                    nullable: true,
1210                    default_constraint: None,
1211                    inverted_index: false,
1212                },
1213                DatatypeJsonTypeHint {
1214                    path: vec!["b".to_string()],
1215                    data_type: ConcreteDataType::boolean_datatype(),
1216                    nullable: true,
1217                    default_constraint: None,
1218                    inverted_index: false,
1219                },
1220            ],
1221            None,
1222        )?;
1223        extensions.set_json_settings(settings)?;
1224
1225        assert_eq!(
1226            extensions
1227                .json2_options
1228                .unwrap()
1229                .type_hints
1230                .iter()
1231                .map(|hint| hint.data_type.to_string())
1232                .collect::<Vec<_>>(),
1233            vec!["BIGINT", "DOUBLE", "BIGINT UNSIGNED", "STRING", "BOOLEAN"]
1234        );
1235        Ok(())
1236    }
1237
1238    #[test]
1239    fn test_set_json_settings_rejects_unsupported_type_hint_type() -> Result<()> {
1240        let err = JsonSettings::try_new(
1241            vec![DatatypeJsonTypeHint {
1242                path: vec!["u".to_string()],
1243                data_type: ConcreteDataType::date_datatype(),
1244                nullable: true,
1245                default_constraint: None,
1246                inverted_index: false,
1247            }],
1248            None,
1249        )
1250        .unwrap_err();
1251
1252        assert!(
1253            err.to_string()
1254                .contains("unsupported JSON2 type hint data type")
1255        );
1256        Ok(())
1257    }
1258
1259    #[test]
1260    fn test_set_empty_json_settings_omits_json2_options() -> Result<()> {
1261        let mut extensions = ColumnExtensions::default();
1262        extensions.set_json_settings(JsonSettings::default())?;
1263        assert!(extensions.json2_options.is_none());
1264        Ok(())
1265    }
1266
1267    #[test]
1268    fn test_display_create_database() {
1269        let sql = r"create database test;";
1270        let stmts =
1271            ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
1272                .unwrap();
1273        assert_eq!(1, stmts.len());
1274        assert_matches!(&stmts[0], Statement::CreateDatabase { .. });
1275
1276        match &stmts[0] {
1277            Statement::CreateDatabase(set) => {
1278                let new_sql = format!("\n{}", set);
1279                assert_eq!(
1280                    r#"
1281CREATE DATABASE test"#,
1282                    &new_sql
1283                );
1284            }
1285            _ => {
1286                unreachable!();
1287            }
1288        }
1289
1290        let sql = r"create database if not exists test;";
1291        let stmts =
1292            ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
1293                .unwrap();
1294        assert_eq!(1, stmts.len());
1295        assert_matches!(&stmts[0], Statement::CreateDatabase { .. });
1296
1297        match &stmts[0] {
1298            Statement::CreateDatabase(set) => {
1299                let new_sql = format!("\n{}", set);
1300                assert_eq!(
1301                    r#"
1302CREATE DATABASE IF NOT EXISTS test"#,
1303                    &new_sql
1304                );
1305            }
1306            _ => {
1307                unreachable!();
1308            }
1309        }
1310
1311        let sql = r#"CREATE DATABASE IF NOT EXISTS test WITH (ttl='1h');"#;
1312        let stmts =
1313            ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
1314                .unwrap();
1315        assert_eq!(1, stmts.len());
1316        assert_matches!(&stmts[0], Statement::CreateDatabase { .. });
1317
1318        match &stmts[0] {
1319            Statement::CreateDatabase(set) => {
1320                let new_sql = format!("\n{}", set);
1321                assert_eq!(
1322                    r#"
1323CREATE DATABASE IF NOT EXISTS test
1324WITH(
1325  ttl = '1h'
1326)"#,
1327                    &new_sql
1328                );
1329            }
1330            _ => {
1331                unreachable!();
1332            }
1333        }
1334    }
1335
1336    #[test]
1337    fn test_display_create_table_like() {
1338        let sql = r"create table t2 like t1;";
1339        let stmts =
1340            ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
1341                .unwrap();
1342        assert_eq!(1, stmts.len());
1343        assert_matches!(&stmts[0], Statement::CreateTableLike { .. });
1344
1345        match &stmts[0] {
1346            Statement::CreateTableLike(create) => {
1347                let new_sql = format!("\n{}", create);
1348                assert_eq!(
1349                    r#"
1350CREATE TABLE t2 LIKE t1"#,
1351                    &new_sql
1352                );
1353            }
1354            _ => {
1355                unreachable!();
1356            }
1357        }
1358    }
1359
1360    #[test]
1361    fn test_display_create_external_table() {
1362        let sql = r#"CREATE EXTERNAL TABLE city (
1363            host string,
1364            ts timestamp,
1365            cpu float64 default 0,
1366            memory float64,
1367            TIME INDEX (ts),
1368            PRIMARY KEY(host)
1369) WITH (location='/var/data/city.csv', format='csv');"#;
1370        let stmts =
1371            ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
1372                .unwrap();
1373        assert_eq!(1, stmts.len());
1374        assert_matches!(&stmts[0], Statement::CreateExternalTable { .. });
1375
1376        match &stmts[0] {
1377            Statement::CreateExternalTable(create) => {
1378                let new_sql = format!("\n{}", create);
1379                assert_eq!(
1380                    r#"
1381CREATE EXTERNAL TABLE city (
1382  host STRING,
1383  ts TIMESTAMP,
1384  cpu DOUBLE DEFAULT 0,
1385  memory DOUBLE,
1386  TIME INDEX (ts),
1387  PRIMARY KEY (host)
1388)
1389ENGINE=file
1390WITH(
1391  format = 'csv',
1392  location = '/var/data/city.csv'
1393)"#,
1394                    &new_sql
1395                );
1396            }
1397            _ => {
1398                unreachable!();
1399            }
1400        }
1401    }
1402
1403    #[test]
1404    fn test_display_create_flow() {
1405        let sql = r"CREATE FLOW filter_numbers
1406            SINK TO out_num_cnt
1407            AS SELECT number FROM numbers_input where number > 10;";
1408        let result =
1409            ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
1410                .unwrap();
1411        assert_eq!(1, result.len());
1412
1413        match &result[0] {
1414            Statement::CreateFlow(c) => {
1415                let new_sql = format!("\n{}", c);
1416                assert_eq!(
1417                    r#"
1418CREATE FLOW filter_numbers
1419SINK TO out_num_cnt
1420AS SELECT number FROM numbers_input where number > 10"#,
1421                    &new_sql
1422                );
1423
1424                let new_result = ParserContext::create_with_dialect(
1425                    &new_sql,
1426                    &GreptimeDbDialect {},
1427                    ParseOptions::default(),
1428                )
1429                .unwrap();
1430                assert_eq!(result, new_result);
1431            }
1432            _ => unreachable!(),
1433        }
1434    }
1435
1436    #[test]
1437    fn test_vector_index_options_validation() {
1438        use super::{ColumnExtensions, OptionMap};
1439
1440        // Test zero connectivity should fail
1441        let extensions = ColumnExtensions {
1442            vector_index_options: Some(OptionMap::from([(
1443                "connectivity".to_string(),
1444                "0".to_string(),
1445            )])),
1446            ..Default::default()
1447        };
1448        let result = extensions.build_vector_index_options();
1449        assert!(result.is_err());
1450        assert!(
1451            result
1452                .unwrap_err()
1453                .to_string()
1454                .contains("connectivity must be in the range [2, 2048]")
1455        );
1456
1457        // Test zero expansion_add should fail
1458        let extensions = ColumnExtensions {
1459            vector_index_options: Some(OptionMap::from([(
1460                "expansion_add".to_string(),
1461                "0".to_string(),
1462            )])),
1463            ..Default::default()
1464        };
1465        let result = extensions.build_vector_index_options();
1466        assert!(result.is_err());
1467        assert!(
1468            result
1469                .unwrap_err()
1470                .to_string()
1471                .contains("expansion_add must be greater than 0")
1472        );
1473
1474        // Test zero expansion_search should fail
1475        let extensions = ColumnExtensions {
1476            vector_index_options: Some(OptionMap::from([(
1477                "expansion_search".to_string(),
1478                "0".to_string(),
1479            )])),
1480            ..Default::default()
1481        };
1482        let result = extensions.build_vector_index_options();
1483        assert!(result.is_err());
1484        assert!(
1485            result
1486                .unwrap_err()
1487                .to_string()
1488                .contains("expansion_search must be greater than 0")
1489        );
1490
1491        // Test valid values should succeed
1492        let extensions = ColumnExtensions {
1493            vector_index_options: Some(OptionMap::from([
1494                ("connectivity".to_string(), "32".to_string()),
1495                ("expansion_add".to_string(), "200".to_string()),
1496                ("expansion_search".to_string(), "100".to_string()),
1497            ])),
1498            ..Default::default()
1499        };
1500        let result = extensions.build_vector_index_options();
1501        assert!(result.is_ok());
1502        let options = result.unwrap().unwrap();
1503        assert_eq!(options.connectivity, 32);
1504        assert_eq!(options.expansion_add, 200);
1505        assert_eq!(options.expansion_search, 100);
1506    }
1507}
sql/statements/create.rs

sql/statements/
create.rs
