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, HashSet};
16use std::fmt::{Display, Formatter};
17use std::sync::Arc;
18
19use common_catalog::consts::FILE_ENGINE;
20use datatypes::data_type::ConcreteDataType;
21use datatypes::json::JsonStructureSettings;
22use datatypes::schema::{
23    FulltextOptions, SkippingIndexOptions, VectorDistanceMetric, VectorIndexEngineType,
24    VectorIndexOptions,
25};
26use datatypes::types::StructType;
27use itertools::Itertools;
28use serde::Serialize;
29use snafu::{OptionExt, ResultExt};
30use sqlparser::ast::{ColumnOptionDef, DataType, Expr};
31use sqlparser_derive::{Visit, VisitMut};
32
33use crate::ast::{ColumnDef, Ident, ObjectName, Value as SqlValue};
34use crate::error::{
35    InvalidFlowQuerySnafu, InvalidJsonStructureSettingSnafu, InvalidSqlSnafu, Result,
36    SetFulltextOptionSnafu, SetSkippingIndexOptionSnafu,
37};
38use crate::statements::query::Query as GtQuery;
39use crate::statements::statement::Statement;
40use crate::statements::tql::Tql;
41use crate::statements::{OptionMap, sql_data_type_to_concrete_data_type};
42use crate::util::OptionValue;
43
44const LINE_SEP: &str = ",\n";
45const COMMA_SEP: &str = ", ";
46const INDENT: usize = 2;
47pub const VECTOR_OPT_DIM: &str = "dim";
48
49pub const JSON_OPT_UNSTRUCTURED_KEYS: &str = "unstructured_keys";
50pub const JSON_OPT_FORMAT: &str = "format";
51pub(crate) const JSON_OPT_FIELDS: &str = "fields";
52pub const JSON_FORMAT_FULL_STRUCTURED: &str = "structured";
53pub const JSON_FORMAT_RAW: &str = "raw";
54pub const JSON_FORMAT_PARTIAL: &str = "partial";
55
56macro_rules! format_indent {
57    ($fmt: expr, $arg: expr) => {
58        format!($fmt, format_args!("{: >1$}", "", INDENT), $arg)
59    };
60    ($arg: expr) => {
61        format_indent!("{}{}", $arg)
62    };
63}
64
65macro_rules! format_list_indent {
66    ($list: expr) => {
67        $list.iter().map(|e| format_indent!(e)).join(LINE_SEP)
68    };
69}
70
71macro_rules! format_list_comma {
72    ($list: expr) => {
73        $list.iter().map(|e| format!("{}", e)).join(COMMA_SEP)
74    };
75}
76
77#[cfg(feature = "enterprise")]
78pub mod trigger;
79
80fn format_table_constraint(constraints: &[TableConstraint]) -> String {
81    constraints.iter().map(|c| format_indent!(c)).join(LINE_SEP)
82}
83
84/// Table constraint for create table statement.
85#[derive(Debug, PartialEq, Eq, Clone, Visit, VisitMut, Serialize)]
86pub enum TableConstraint {
87    /// Primary key constraint.
88    PrimaryKey { columns: Vec<Ident> },
89    /// Time index constraint.
90    TimeIndex { column: Ident },
91}
92
93impl Display for TableConstraint {
94    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
95        match self {
96            TableConstraint::PrimaryKey { columns } => {
97                write!(f, "PRIMARY KEY ({})", format_list_comma!(columns))
98            }
99            TableConstraint::TimeIndex { column } => {
100                write!(f, "TIME INDEX ({})", column)
101            }
102        }
103    }
104}
105
106#[derive(Debug, PartialEq, Eq, Clone, Visit, VisitMut, Serialize)]
107pub struct CreateTable {
108    /// Create if not exists
109    pub if_not_exists: bool,
110    pub table_id: u32,
111    /// Table name
112    pub name: ObjectName,
113    pub columns: Vec<Column>,
114    pub engine: String,
115    pub constraints: Vec<TableConstraint>,
116    /// Table options in `WITH`. All keys are lowercase.
117    pub options: OptionMap,
118    pub partitions: Option<Partitions>,
119}
120
121/// Column definition in `CREATE TABLE` statement.
122#[derive(Debug, PartialEq, Eq, Clone, Visit, VisitMut, Serialize)]
123pub struct Column {
124    /// `ColumnDef` from `sqlparser::ast`
125    pub column_def: ColumnDef,
126    /// Column extensions for greptimedb dialect.
127    pub extensions: ColumnExtensions,
128}
129
130/// Column extensions for greptimedb dialect.
131#[derive(Debug, PartialEq, Eq, Clone, Visit, VisitMut, Default, Serialize)]
132pub struct ColumnExtensions {
133    /// Vector type options.
134    pub vector_options: Option<OptionMap>,
135
136    /// Fulltext index options.
137    pub fulltext_index_options: Option<OptionMap>,
138    /// Skipping index options.
139    pub skipping_index_options: Option<OptionMap>,
140    /// Inverted index options.
141    ///
142    /// Inverted index doesn't have options at present. There won't be any options in that map.
143    pub inverted_index_options: Option<OptionMap>,
144    /// Vector index options for HNSW-based vector similarity search.
145    pub vector_index_options: Option<OptionMap>,
146    pub json_datatype_options: Option<OptionMap>,
147}
148
149impl Column {
150    pub fn name(&self) -> &Ident {
151        &self.column_def.name
152    }
153
154    pub fn data_type(&self) -> &DataType {
155        &self.column_def.data_type
156    }
157
158    pub fn mut_data_type(&mut self) -> &mut DataType {
159        &mut self.column_def.data_type
160    }
161
162    pub fn options(&self) -> &[ColumnOptionDef] {
163        &self.column_def.options
164    }
165
166    pub fn mut_options(&mut self) -> &mut Vec<ColumnOptionDef> {
167        &mut self.column_def.options
168    }
169}
170
171impl Display for Column {
172    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
173        if let Some(vector_options) = &self.extensions.vector_options
174            && let Some(dim) = vector_options.get(VECTOR_OPT_DIM)
175        {
176            write!(f, "{} VECTOR({})", self.column_def.name, dim)?;
177            return Ok(());
178        }
179
180        write!(f, "{} {}", self.column_def.name, self.column_def.data_type)?;
181        if let Some(options) = &self.extensions.json_datatype_options {
182            write!(
183                f,
184                "({})",
185                options
186                    .entries()
187                    .map(|(k, v)| format!("{k} = {v}"))
188                    .join(COMMA_SEP)
189            )?;
190        }
191        for option in &self.column_def.options {
192            write!(f, " {option}")?;
193        }
194
195        if let Some(fulltext_options) = &self.extensions.fulltext_index_options {
196            if !fulltext_options.is_empty() {
197                let options = fulltext_options.kv_pairs();
198                write!(f, " FULLTEXT INDEX WITH({})", format_list_comma!(options))?;
199            } else {
200                write!(f, " FULLTEXT INDEX")?;
201            }
202        }
203
204        if let Some(skipping_index_options) = &self.extensions.skipping_index_options {
205            if !skipping_index_options.is_empty() {
206                let options = skipping_index_options.kv_pairs();
207                write!(f, " SKIPPING INDEX WITH({})", format_list_comma!(options))?;
208            } else {
209                write!(f, " SKIPPING INDEX")?;
210            }
211        }
212
213        if let Some(inverted_index_options) = &self.extensions.inverted_index_options {
214            if !inverted_index_options.is_empty() {
215                let options = inverted_index_options.kv_pairs();
216                write!(f, " INVERTED INDEX WITH({})", format_list_comma!(options))?;
217            } else {
218                write!(f, " INVERTED INDEX")?;
219            }
220        }
221
222        if let Some(vector_index_options) = &self.extensions.vector_index_options {
223            if !vector_index_options.is_empty() {
224                let options = vector_index_options.kv_pairs();
225                write!(f, " VECTOR INDEX WITH({})", format_list_comma!(options))?;
226            } else {
227                write!(f, " VECTOR INDEX")?;
228            }
229        }
230        Ok(())
231    }
232}
233
234impl ColumnExtensions {
235    pub fn build_fulltext_options(&self) -> Result<Option<FulltextOptions>> {
236        let Some(options) = self.fulltext_index_options.as_ref() else {
237            return Ok(None);
238        };
239
240        let options: HashMap<String, String> = options.clone().into_map();
241        Ok(Some(options.try_into().context(SetFulltextOptionSnafu)?))
242    }
243
244    pub fn build_skipping_index_options(&self) -> Result<Option<SkippingIndexOptions>> {
245        let Some(options) = self.skipping_index_options.as_ref() else {
246            return Ok(None);
247        };
248
249        let options: HashMap<String, String> = options.clone().into_map();
250        Ok(Some(
251            options.try_into().context(SetSkippingIndexOptionSnafu)?,
252        ))
253    }
254
255    pub fn build_vector_index_options(&self) -> Result<Option<VectorIndexOptions>> {
256        let Some(options) = self.vector_index_options.as_ref() else {
257            return Ok(None);
258        };
259
260        let options_map: HashMap<String, String> = options.clone().into_map();
261        let mut result = VectorIndexOptions::default();
262
263        if let Some(s) = options_map.get("engine") {
264            result.engine = s.parse::<VectorIndexEngineType>().map_err(|e| {
265                InvalidSqlSnafu {
266                    msg: format!("invalid VECTOR INDEX engine: {e}"),
267                }
268                .build()
269            })?;
270        }
271
272        if let Some(s) = options_map.get("metric") {
273            result.metric = s.parse::<VectorDistanceMetric>().map_err(|e| {
274                InvalidSqlSnafu {
275                    msg: format!("invalid VECTOR INDEX metric: {e}"),
276                }
277                .build()
278            })?;
279        }
280
281        if let Some(s) = options_map.get("connectivity") {
282            let value = s.parse::<u32>().map_err(|_| {
283                InvalidSqlSnafu {
284                    msg: format!(
285                        "invalid VECTOR INDEX connectivity: {s}, expected positive integer"
286                    ),
287                }
288                .build()
289            })?;
290            if !(2..=2048).contains(&value) {
291                return InvalidSqlSnafu {
292                    msg: "VECTOR INDEX connectivity must be in the range [2, 2048].".to_string(),
293                }
294                .fail();
295            }
296            result.connectivity = value;
297        }
298
299        if let Some(s) = options_map.get("expansion_add") {
300            let value = s.parse::<u32>().map_err(|_| {
301                InvalidSqlSnafu {
302                    msg: format!(
303                        "invalid VECTOR INDEX expansion_add: {s}, expected positive integer"
304                    ),
305                }
306                .build()
307            })?;
308            if value == 0 {
309                return InvalidSqlSnafu {
310                    msg: "VECTOR INDEX expansion_add must be greater than 0".to_string(),
311                }
312                .fail();
313            }
314            result.expansion_add = value;
315        }
316
317        if let Some(s) = options_map.get("expansion_search") {
318            let value = s.parse::<u32>().map_err(|_| {
319                InvalidSqlSnafu {
320                    msg: format!(
321                        "invalid VECTOR INDEX expansion_search: {s}, expected positive integer"
322                    ),
323                }
324                .build()
325            })?;
326            if value == 0 {
327                return InvalidSqlSnafu {
328                    msg: "VECTOR INDEX expansion_search must be greater than 0".to_string(),
329                }
330                .fail();
331            }
332            result.expansion_search = value;
333        }
334
335        Ok(Some(result))
336    }
337
338    pub fn build_json_structure_settings(&self) -> Result<Option<JsonStructureSettings>> {
339        let Some(options) = self.json_datatype_options.as_ref() else {
340            return Ok(None);
341        };
342
343        let unstructured_keys = options
344            .value(JSON_OPT_UNSTRUCTURED_KEYS)
345            .and_then(|v| {
346                v.as_list().map(|x| {
347                    x.into_iter()
348                        .map(|x| x.to_string())
349                        .collect::<HashSet<String>>()
350                })
351            })
352            .unwrap_or_default();
353
354        let fields = if let Some(value) = options.value(JSON_OPT_FIELDS) {
355            let fields = value
356                .as_struct_fields()
357                .context(InvalidJsonStructureSettingSnafu {
358                    reason: format!(r#"expect "{JSON_OPT_FIELDS}" a struct, actual: "{value}""#,),
359                })?;
360            let fields = fields
361                .iter()
362                .map(|field| {
363                    let name = field.field_name.as_ref().map(|x| x.value.clone()).context(
364                        InvalidJsonStructureSettingSnafu {
365                            reason: format!(r#"missing field name in "{field}""#),
366                        },
367                    )?;
368                    let datatype = sql_data_type_to_concrete_data_type(
369                        &field.field_type,
370                        &Default::default(),
371                    )?;
372                    Ok(datatypes::types::StructField::new(name, datatype, true))
373                })
374                .collect::<Result<_>>()?;
375            Some(StructType::new(Arc::new(fields)))
376        } else {
377            None
378        };
379
380        let format = options
381            .get(JSON_OPT_FORMAT)
382            .unwrap_or(JSON_FORMAT_FULL_STRUCTURED);
383        let settings = match format {
384            JSON_FORMAT_FULL_STRUCTURED => JsonStructureSettings::Structured(fields),
385            JSON_FORMAT_PARTIAL => {
386                let fields = fields.map(|fields| {
387                    let mut fields = Arc::unwrap_or_clone(fields.fields());
388                    fields.push(datatypes::types::StructField::new(
389                        JsonStructureSettings::RAW_FIELD.to_string(),
390                        ConcreteDataType::string_datatype(),
391                        true,
392                    ));
393                    StructType::new(Arc::new(fields))
394                });
395                JsonStructureSettings::PartialUnstructuredByKey {
396                    fields,
397                    unstructured_keys,
398                }
399            }
400            JSON_FORMAT_RAW => JsonStructureSettings::UnstructuredRaw,
401            _ => {
402                return InvalidSqlSnafu {
403                    msg: format!("unknown JSON datatype 'format': {format}"),
404                }
405                .fail();
406            }
407        };
408        Ok(Some(settings))
409    }
410
411    pub fn set_json_structure_settings(&mut self, settings: JsonStructureSettings) {
412        let mut map = OptionMap::default();
413
414        let format = match settings {
415            JsonStructureSettings::Structured(_) => JSON_FORMAT_FULL_STRUCTURED,
416            JsonStructureSettings::PartialUnstructuredByKey { .. } => JSON_FORMAT_PARTIAL,
417            JsonStructureSettings::UnstructuredRaw => JSON_FORMAT_RAW,
418        };
419        map.insert(JSON_OPT_FORMAT.to_string(), format.to_string());
420
421        if let JsonStructureSettings::PartialUnstructuredByKey {
422            fields: _,
423            unstructured_keys,
424        } = settings
425        {
426            let value = OptionValue::from(
427                unstructured_keys
428                    .iter()
429                    .map(|x| x.as_str())
430                    .sorted()
431                    .collect::<Vec<_>>(),
432            );
433            map.insert_options(JSON_OPT_UNSTRUCTURED_KEYS, value);
434        }
435
436        self.json_datatype_options = Some(map);
437    }
438}
439
440/// Partition on columns or values.
441///
442/// - `column_list` is the list of columns in `PARTITION ON COLUMNS` clause.
443/// - `exprs` is the list of expressions in `PARTITION ON VALUES` clause, like
444///   `host <= 'host1'`, `host > 'host1' and host <= 'host2'` or `host > 'host2'`.
445///   Each expression stands for a partition.
446#[derive(Debug, PartialEq, Eq, Clone, Visit, VisitMut, Serialize)]
447pub struct Partitions {
448    pub column_list: Vec<Ident>,
449    pub exprs: Vec<Expr>,
450}
451
452impl Partitions {
453    /// set quotes to all [Ident]s from column list
454    pub fn set_quote(&mut self, quote_style: char) {
455        self.column_list
456            .iter_mut()
457            .for_each(|c| c.quote_style = Some(quote_style));
458    }
459}
460
461#[derive(Debug, PartialEq, Eq, Clone, Visit, VisitMut)]
462pub struct PartitionEntry {
463    pub name: Ident,
464    pub value_list: Vec<SqlValue>,
465}
466
467impl Display for PartitionEntry {
468    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
469        write!(
470            f,
471            "PARTITION {} VALUES LESS THAN ({})",
472            self.name,
473            format_list_comma!(self.value_list),
474        )
475    }
476}
477
478impl Display for Partitions {
479    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
480        if !self.column_list.is_empty() {
481            write!(
482                f,
483                "PARTITION ON COLUMNS ({}) (\n{}\n)",
484                format_list_comma!(self.column_list),
485                format_list_indent!(self.exprs),
486            )?;
487        }
488        Ok(())
489    }
490}
491
492impl Display for CreateTable {
493    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
494        write!(f, "CREATE ")?;
495        if self.engine == FILE_ENGINE {
496            write!(f, "EXTERNAL ")?;
497        }
498        write!(f, "TABLE ")?;
499        if self.if_not_exists {
500            write!(f, "IF NOT EXISTS ")?;
501        }
502        writeln!(f, "{} (", &self.name)?;
503        writeln!(f, "{},", format_list_indent!(self.columns))?;
504        writeln!(f, "{}", format_table_constraint(&self.constraints))?;
505        writeln!(f, ")")?;
506        if let Some(partitions) = &self.partitions {
507            writeln!(f, "{partitions}")?;
508        }
509        writeln!(f, "ENGINE={}", &self.engine)?;
510        if !self.options.is_empty() {
511            let options = self.options.kv_pairs();
512            write!(f, "WITH(\n{}\n)", format_list_indent!(options))?;
513        }
514        Ok(())
515    }
516}
517
518#[derive(Debug, PartialEq, Eq, Clone, Visit, VisitMut, Serialize)]
519pub struct CreateDatabase {
520    pub name: ObjectName,
521    /// Create if not exists
522    pub if_not_exists: bool,
523    pub options: OptionMap,
524}
525
526impl CreateDatabase {
527    /// Creates a statement for `CREATE DATABASE`
528    pub fn new(name: ObjectName, if_not_exists: bool, options: OptionMap) -> Self {
529        Self {
530            name,
531            if_not_exists,
532            options,
533        }
534    }
535}
536
537impl Display for CreateDatabase {
538    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
539        write!(f, "CREATE DATABASE ")?;
540        if self.if_not_exists {
541            write!(f, "IF NOT EXISTS ")?;
542        }
543        write!(f, "{}", &self.name)?;
544        if !self.options.is_empty() {
545            let options = self.options.kv_pairs();
546            write!(f, "\nWITH(\n{}\n)", format_list_indent!(options))?;
547        }
548        Ok(())
549    }
550}
551
552#[derive(Debug, PartialEq, Eq, Clone, Visit, VisitMut, Serialize)]
553pub struct CreateExternalTable {
554    /// Table name
555    pub name: ObjectName,
556    pub columns: Vec<Column>,
557    pub constraints: Vec<TableConstraint>,
558    /// Table options in `WITH`. All keys are lowercase.
559    pub options: OptionMap,
560    pub if_not_exists: bool,
561    pub engine: String,
562}
563
564impl Display for CreateExternalTable {
565    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
566        write!(f, "CREATE EXTERNAL TABLE ")?;
567        if self.if_not_exists {
568            write!(f, "IF NOT EXISTS ")?;
569        }
570        writeln!(f, "{} (", &self.name)?;
571        writeln!(f, "{},", format_list_indent!(self.columns))?;
572        writeln!(f, "{}", format_table_constraint(&self.constraints))?;
573        writeln!(f, ")")?;
574        writeln!(f, "ENGINE={}", &self.engine)?;
575        if !self.options.is_empty() {
576            let options = self.options.kv_pairs();
577            write!(f, "WITH(\n{}\n)", format_list_indent!(options))?;
578        }
579        Ok(())
580    }
581}
582
583#[derive(Debug, PartialEq, Eq, Clone, Visit, VisitMut, Serialize)]
584pub struct CreateTableLike {
585    /// Table name
586    pub table_name: ObjectName,
587    /// The table that is designated to be imitated by `Like`
588    pub source_name: ObjectName,
589}
590
591impl Display for CreateTableLike {
592    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
593        let table_name = &self.table_name;
594        let source_name = &self.source_name;
595        write!(f, r#"CREATE TABLE {table_name} LIKE {source_name}"#)
596    }
597}
598
599#[derive(Debug, PartialEq, Eq, Clone, Visit, VisitMut, Serialize)]
600pub struct CreateFlow {
601    /// Flow name
602    pub flow_name: ObjectName,
603    /// Output (sink) table name
604    pub sink_table_name: ObjectName,
605    /// Whether to replace existing task
606    pub or_replace: bool,
607    /// Create if not exist
608    pub if_not_exists: bool,
609    /// `EXPIRE AFTER`
610    /// Duration in second as `i64`
611    pub expire_after: Option<i64>,
612    /// Duration for flow evaluation interval
613    /// Duration in seconds as `i64`
614    /// If not set, flow will be evaluated based on time window size and other args.
615    pub eval_interval: Option<i64>,
616    /// Comment string
617    pub comment: Option<String>,
618    /// SQL statement
619    pub query: Box<SqlOrTql>,
620}
621
622/// Either a sql query or a tql query
623#[derive(Debug, PartialEq, Eq, Clone, Visit, VisitMut, Serialize)]
624pub enum SqlOrTql {
625    Sql(GtQuery, String),
626    Tql(Tql, String),
627}
628
629impl std::fmt::Display for SqlOrTql {
630    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
631        match self {
632            Self::Sql(_, s) => write!(f, "{}", s),
633            Self::Tql(_, s) => write!(f, "{}", s),
634        }
635    }
636}
637
638impl SqlOrTql {
639    pub fn try_from_statement(
640        value: Statement,
641        original_query: &str,
642    ) -> std::result::Result<Self, crate::error::Error> {
643        match value {
644            Statement::Query(query) => Ok(Self::Sql(*query, original_query.to_string())),
645            Statement::Tql(tql) => Ok(Self::Tql(tql, original_query.to_string())),
646            _ => InvalidFlowQuerySnafu {
647                reason: format!("Expect either sql query or promql query, found {:?}", value),
648            }
649            .fail(),
650        }
651    }
652}
653
654impl Display for CreateFlow {
655    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
656        write!(f, "CREATE ")?;
657        if self.or_replace {
658            write!(f, "OR REPLACE ")?;
659        }
660        write!(f, "FLOW ")?;
661        if self.if_not_exists {
662            write!(f, "IF NOT EXISTS ")?;
663        }
664        writeln!(f, "{}", &self.flow_name)?;
665        writeln!(f, "SINK TO {}", &self.sink_table_name)?;
666        if let Some(expire_after) = &self.expire_after {
667            writeln!(f, "EXPIRE AFTER '{} s'", expire_after)?;
668        }
669        if let Some(eval_interval) = &self.eval_interval {
670            writeln!(f, "EVAL INTERVAL '{} s'", eval_interval)?;
671        }
672        if let Some(comment) = &self.comment {
673            writeln!(f, "COMMENT '{}'", comment)?;
674        }
675        write!(f, "AS {}", &self.query)
676    }
677}
678
679/// Create SQL view statement.
680#[derive(Debug, PartialEq, Eq, Clone, Visit, VisitMut, Serialize)]
681pub struct CreateView {
682    /// View name
683    pub name: ObjectName,
684    /// An optional list of names to be used for columns of the view
685    pub columns: Vec<Ident>,
686    /// The clause after `As` that defines the VIEW.
687    /// Can only be either [Statement::Query] or [Statement::Tql].
688    pub query: Box<Statement>,
689    /// Whether to replace existing VIEW
690    pub or_replace: bool,
691    /// Create VIEW only when it doesn't exists
692    pub if_not_exists: bool,
693}
694
695impl Display for CreateView {
696    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
697        write!(f, "CREATE ")?;
698        if self.or_replace {
699            write!(f, "OR REPLACE ")?;
700        }
701        write!(f, "VIEW ")?;
702        if self.if_not_exists {
703            write!(f, "IF NOT EXISTS ")?;
704        }
705        write!(f, "{} ", &self.name)?;
706        if !self.columns.is_empty() {
707            write!(f, "({}) ", format_list_comma!(self.columns))?;
708        }
709        write!(f, "AS {}", &self.query)
710    }
711}
712
713#[cfg(test)]
714mod tests {
715    use std::assert_matches;
716
717    use crate::dialect::GreptimeDbDialect;
718    use crate::error::Error;
719    use crate::parser::{ParseOptions, ParserContext};
720    use crate::statements::statement::Statement;
721
722    #[test]
723    fn test_display_create_table() {
724        let sql = r"create table if not exists demo(
725                             host string,
726                             ts timestamp,
727                             cpu double default 0,
728                             memory double,
729                             TIME INDEX (ts),
730                             PRIMARY KEY(host)
731                       )
732                       PARTITION ON COLUMNS (host) (
733                            host = 'a',
734                            host > 'a',
735                       )
736                       engine=mito
737                       with(ttl='7d', storage='File');
738         ";
739        let result =
740            ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
741                .unwrap();
742        assert_eq!(1, result.len());
743
744        match &result[0] {
745            Statement::CreateTable(c) => {
746                let new_sql = format!("\n{}", c);
747                assert_eq!(
748                    r#"
749CREATE TABLE IF NOT EXISTS demo (
750  host STRING,
751  ts TIMESTAMP,
752  cpu DOUBLE DEFAULT 0,
753  memory DOUBLE,
754  TIME INDEX (ts),
755  PRIMARY KEY (host)
756)
757PARTITION ON COLUMNS (host) (
758  host = 'a',
759  host > 'a'
760)
761ENGINE=mito
762WITH(
763  storage = 'File',
764  ttl = '7d'
765)"#,
766                    &new_sql
767                );
768
769                let new_result = ParserContext::create_with_dialect(
770                    &new_sql,
771                    &GreptimeDbDialect {},
772                    ParseOptions::default(),
773                )
774                .unwrap();
775                assert_eq!(result, new_result);
776            }
777            _ => unreachable!(),
778        }
779    }
780
781    #[test]
782    fn test_display_empty_partition_column() {
783        let sql = r"create table if not exists demo(
784            host string,
785            ts timestamp,
786            cpu double default 0,
787            memory double,
788            TIME INDEX (ts),
789            PRIMARY KEY(ts, host)
790            );
791        ";
792        let result =
793            ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
794                .unwrap();
795        assert_eq!(1, result.len());
796
797        match &result[0] {
798            Statement::CreateTable(c) => {
799                let new_sql = format!("\n{}", c);
800                assert_eq!(
801                    r#"
802CREATE TABLE IF NOT EXISTS demo (
803  host STRING,
804  ts TIMESTAMP,
805  cpu DOUBLE DEFAULT 0,
806  memory DOUBLE,
807  TIME INDEX (ts),
808  PRIMARY KEY (ts, host)
809)
810ENGINE=mito
811"#,
812                    &new_sql
813                );
814
815                let new_result = ParserContext::create_with_dialect(
816                    &new_sql,
817                    &GreptimeDbDialect {},
818                    ParseOptions::default(),
819                )
820                .unwrap();
821                assert_eq!(result, new_result);
822            }
823            _ => unreachable!(),
824        }
825    }
826
827    #[test]
828    fn test_validate_table_options() {
829        let sql = r"create table if not exists demo(
830            host string,
831            ts timestamp,
832            cpu double default 0,
833            memory double,
834            TIME INDEX (ts),
835            PRIMARY KEY(host)
836      )
837      PARTITION ON COLUMNS (host) ()
838      engine=mito
839      with(ttl='7d', 'compaction.type'='world');
840";
841        let result =
842            ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
843                .unwrap();
844        match &result[0] {
845            Statement::CreateTable(c) => {
846                assert_eq!(2, c.options.len());
847            }
848            _ => unreachable!(),
849        }
850
851        let sql = r"create table if not exists demo(
852            host string,
853            ts timestamp,
854            cpu double default 0,
855            memory double,
856            TIME INDEX (ts),
857            PRIMARY KEY(host)
858      )
859      PARTITION ON COLUMNS (host) ()
860      engine=mito
861      with(ttl='7d', hello='world');
862";
863        let result =
864            ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default());
865        assert_matches!(result, Err(Error::InvalidTableOption { .. }))
866    }
867
868    #[test]
869    fn test_display_create_database() {
870        let sql = r"create database test;";
871        let stmts =
872            ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
873                .unwrap();
874        assert_eq!(1, stmts.len());
875        assert_matches!(&stmts[0], Statement::CreateDatabase { .. });
876
877        match &stmts[0] {
878            Statement::CreateDatabase(set) => {
879                let new_sql = format!("\n{}", set);
880                assert_eq!(
881                    r#"
882CREATE DATABASE test"#,
883                    &new_sql
884                );
885            }
886            _ => {
887                unreachable!();
888            }
889        }
890
891        let sql = r"create database if not exists test;";
892        let stmts =
893            ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
894                .unwrap();
895        assert_eq!(1, stmts.len());
896        assert_matches!(&stmts[0], Statement::CreateDatabase { .. });
897
898        match &stmts[0] {
899            Statement::CreateDatabase(set) => {
900                let new_sql = format!("\n{}", set);
901                assert_eq!(
902                    r#"
903CREATE DATABASE IF NOT EXISTS test"#,
904                    &new_sql
905                );
906            }
907            _ => {
908                unreachable!();
909            }
910        }
911
912        let sql = r#"CREATE DATABASE IF NOT EXISTS test WITH (ttl='1h');"#;
913        let stmts =
914            ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
915                .unwrap();
916        assert_eq!(1, stmts.len());
917        assert_matches!(&stmts[0], Statement::CreateDatabase { .. });
918
919        match &stmts[0] {
920            Statement::CreateDatabase(set) => {
921                let new_sql = format!("\n{}", set);
922                assert_eq!(
923                    r#"
924CREATE DATABASE IF NOT EXISTS test
925WITH(
926  ttl = '1h'
927)"#,
928                    &new_sql
929                );
930            }
931            _ => {
932                unreachable!();
933            }
934        }
935    }
936
937    #[test]
938    fn test_display_create_table_like() {
939        let sql = r"create table t2 like t1;";
940        let stmts =
941            ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
942                .unwrap();
943        assert_eq!(1, stmts.len());
944        assert_matches!(&stmts[0], Statement::CreateTableLike { .. });
945
946        match &stmts[0] {
947            Statement::CreateTableLike(create) => {
948                let new_sql = format!("\n{}", create);
949                assert_eq!(
950                    r#"
951CREATE TABLE t2 LIKE t1"#,
952                    &new_sql
953                );
954            }
955            _ => {
956                unreachable!();
957            }
958        }
959    }
960
961    #[test]
962    fn test_display_create_external_table() {
963        let sql = r#"CREATE EXTERNAL TABLE city (
964            host string,
965            ts timestamp,
966            cpu float64 default 0,
967            memory float64,
968            TIME INDEX (ts),
969            PRIMARY KEY(host)
970) WITH (location='/var/data/city.csv', format='csv');"#;
971        let stmts =
972            ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
973                .unwrap();
974        assert_eq!(1, stmts.len());
975        assert_matches!(&stmts[0], Statement::CreateExternalTable { .. });
976
977        match &stmts[0] {
978            Statement::CreateExternalTable(create) => {
979                let new_sql = format!("\n{}", create);
980                assert_eq!(
981                    r#"
982CREATE EXTERNAL TABLE city (
983  host STRING,
984  ts TIMESTAMP,
985  cpu DOUBLE DEFAULT 0,
986  memory DOUBLE,
987  TIME INDEX (ts),
988  PRIMARY KEY (host)
989)
990ENGINE=file
991WITH(
992  format = 'csv',
993  location = '/var/data/city.csv'
994)"#,
995                    &new_sql
996                );
997            }
998            _ => {
999                unreachable!();
1000            }
1001        }
1002    }
1003
1004    #[test]
1005    fn test_display_create_flow() {
1006        let sql = r"CREATE FLOW filter_numbers
1007            SINK TO out_num_cnt
1008            AS SELECT number FROM numbers_input where number > 10;";
1009        let result =
1010            ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
1011                .unwrap();
1012        assert_eq!(1, result.len());
1013
1014        match &result[0] {
1015            Statement::CreateFlow(c) => {
1016                let new_sql = format!("\n{}", c);
1017                assert_eq!(
1018                    r#"
1019CREATE FLOW filter_numbers
1020SINK TO out_num_cnt
1021AS SELECT number FROM numbers_input where number > 10"#,
1022                    &new_sql
1023                );
1024
1025                let new_result = ParserContext::create_with_dialect(
1026                    &new_sql,
1027                    &GreptimeDbDialect {},
1028                    ParseOptions::default(),
1029                )
1030                .unwrap();
1031                assert_eq!(result, new_result);
1032            }
1033            _ => unreachable!(),
1034        }
1035    }
1036
1037    #[test]
1038    fn test_vector_index_options_validation() {
1039        use super::{ColumnExtensions, OptionMap};
1040
1041        // Test zero connectivity should fail
1042        let extensions = ColumnExtensions {
1043            fulltext_index_options: None,
1044            vector_options: None,
1045            skipping_index_options: None,
1046            inverted_index_options: None,
1047            json_datatype_options: None,
1048            vector_index_options: Some(OptionMap::from([(
1049                "connectivity".to_string(),
1050                "0".to_string(),
1051            )])),
1052        };
1053        let result = extensions.build_vector_index_options();
1054        assert!(result.is_err());
1055        assert!(
1056            result
1057                .unwrap_err()
1058                .to_string()
1059                .contains("connectivity must be in the range [2, 2048]")
1060        );
1061
1062        // Test zero expansion_add should fail
1063        let extensions = ColumnExtensions {
1064            fulltext_index_options: None,
1065            vector_options: None,
1066            skipping_index_options: None,
1067            inverted_index_options: None,
1068            json_datatype_options: None,
1069            vector_index_options: Some(OptionMap::from([(
1070                "expansion_add".to_string(),
1071                "0".to_string(),
1072            )])),
1073        };
1074        let result = extensions.build_vector_index_options();
1075        assert!(result.is_err());
1076        assert!(
1077            result
1078                .unwrap_err()
1079                .to_string()
1080                .contains("expansion_add must be greater than 0")
1081        );
1082
1083        // Test zero expansion_search should fail
1084        let extensions = ColumnExtensions {
1085            fulltext_index_options: None,
1086            vector_options: None,
1087            skipping_index_options: None,
1088            inverted_index_options: None,
1089            json_datatype_options: None,
1090            vector_index_options: Some(OptionMap::from([(
1091                "expansion_search".to_string(),
1092                "0".to_string(),
1093            )])),
1094        };
1095        let result = extensions.build_vector_index_options();
1096        assert!(result.is_err());
1097        assert!(
1098            result
1099                .unwrap_err()
1100                .to_string()
1101                .contains("expansion_search must be greater than 0")
1102        );
1103
1104        // Test valid values should succeed
1105        let extensions = ColumnExtensions {
1106            fulltext_index_options: None,
1107            vector_options: None,
1108            skipping_index_options: None,
1109            inverted_index_options: None,
1110            json_datatype_options: None,
1111            vector_index_options: Some(OptionMap::from([
1112                ("connectivity".to_string(), "32".to_string()),
1113                ("expansion_add".to_string(), "200".to_string()),
1114                ("expansion_search".to_string(), "100".to_string()),
1115            ])),
1116        };
1117        let result = extensions.build_vector_index_options();
1118        assert!(result.is_ok());
1119        let options = result.unwrap().unwrap();
1120        assert_eq!(options.connectivity, 32);
1121        assert_eq!(options.expansion_add, 200);
1122        assert_eq!(options.expansion_search, 100);
1123    }
1124}
sql/statements/create.rs

sql/statements/
create.rs
