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

sql/statements/
create.rs
