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: sql_data_type_to_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_sql_data_type(data_type: &ConcreteDataType) -> Result<DataType> {
400    let sql_type = match data_type {
401        ConcreteDataType::String(_) => DataType::String(None),
402        ConcreteDataType::Int64(_) => DataType::BigInt(None),
403        ConcreteDataType::UInt64(_) => DataType::BigIntUnsigned(None),
404        ConcreteDataType::Float64(_) => DataType::Double(sqlparser::ast::ExactNumberInfo::None),
405        ConcreteDataType::Boolean(_) => DataType::Boolean,
406        _ => {
407            return InvalidSqlSnafu {
408                msg: format!("unsupported JSON2 type hint data type: {data_type}"),
409            }
410            .fail();
411        }
412    };
413    Ok(sql_type)
414}
415
416fn format_json_type_hint(hint: &JsonTypeHint) -> String {
417    let path = hint
418        .path
419        .iter()
420        .map(|segment| format_json_path_segment(segment))
421        .join(".");
422    let inverted_index = if hint.inverted_index {
423        " INVERTED INDEX"
424    } else {
425        ""
426    };
427    format!("{} {}{}", path, hint.data_type, inverted_index)
428}
429
430fn format_json_path_segment(segment: &str) -> String {
431    format!("\"{}\"", segment.replace('"', "\"\""))
432}
433
434/// Partition on columns or values.
435///
436/// - `column_list` is the list of columns in `PARTITION ON COLUMNS` clause.
437/// - `exprs` is the list of expressions in `PARTITION ON VALUES` clause, like
438///   `host <= 'host1'`, `host > 'host1' and host <= 'host2'` or `host > 'host2'`.
439///   Each expression stands for a partition.
440#[derive(Debug, PartialEq, Eq, Clone, Visit, VisitMut, Serialize)]
441pub struct Partitions {
442    pub column_list: Vec<Ident>,
443    pub exprs: Vec<Expr>,
444}
445
446impl Partitions {
447    /// set quotes to all [Ident]s from column list
448    pub fn set_quote(&mut self, quote_style: char) {
449        self.column_list
450            .iter_mut()
451            .for_each(|c| c.quote_style = Some(quote_style));
452    }
453}
454
455#[derive(Debug, PartialEq, Eq, Clone, Visit, VisitMut)]
456pub struct PartitionEntry {
457    pub name: Ident,
458    pub value_list: Vec<SqlValue>,
459}
460
461impl Display for PartitionEntry {
462    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
463        write!(
464            f,
465            "PARTITION {} VALUES LESS THAN ({})",
466            self.name,
467            format_list_comma!(self.value_list),
468        )
469    }
470}
471
472impl Display for Partitions {
473    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
474        if !self.column_list.is_empty() {
475            write!(
476                f,
477                "PARTITION ON COLUMNS ({}) (\n{}\n)",
478                format_list_comma!(self.column_list),
479                format_list_indent!(self.exprs),
480            )?;
481        }
482        Ok(())
483    }
484}
485
486impl Display for CreateTable {
487    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
488        write!(f, "CREATE ")?;
489        if self.engine == FILE_ENGINE {
490            write!(f, "EXTERNAL ")?;
491        }
492        write!(f, "TABLE ")?;
493        if self.if_not_exists {
494            write!(f, "IF NOT EXISTS ")?;
495        }
496        writeln!(f, "{} (", &self.name)?;
497        writeln!(f, "{},", format_list_indent!(self.columns))?;
498        writeln!(f, "{}", format_table_constraint(&self.constraints))?;
499        writeln!(f, ")")?;
500        if let Some(partitions) = &self.partitions {
501            writeln!(f, "{partitions}")?;
502        }
503        writeln!(f, "ENGINE={}", &self.engine)?;
504        if !self.options.is_empty() {
505            let options = self.options.kv_pairs();
506            write!(f, "WITH(\n{}\n)", format_list_indent!(options))?;
507        }
508        Ok(())
509    }
510}
511
512#[derive(Debug, PartialEq, Eq, Clone, Visit, VisitMut, Serialize)]
513pub struct CreateDatabase {
514    pub name: ObjectName,
515    /// Create if not exists
516    pub if_not_exists: bool,
517    pub options: OptionMap,
518}
519
520impl CreateDatabase {
521    /// Creates a statement for `CREATE DATABASE`
522    pub fn new(name: ObjectName, if_not_exists: bool, options: OptionMap) -> Self {
523        Self {
524            name,
525            if_not_exists,
526            options,
527        }
528    }
529}
530
531impl Display for CreateDatabase {
532    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
533        write!(f, "CREATE DATABASE ")?;
534        if self.if_not_exists {
535            write!(f, "IF NOT EXISTS ")?;
536        }
537        write!(f, "{}", &self.name)?;
538        if !self.options.is_empty() {
539            let options = self.options.kv_pairs();
540            write!(f, "\nWITH(\n{}\n)", format_list_indent!(options))?;
541        }
542        Ok(())
543    }
544}
545
546#[derive(Debug, PartialEq, Eq, Clone, Visit, VisitMut, Serialize)]
547pub struct CreateExternalTable {
548    /// Table name
549    pub name: ObjectName,
550    pub columns: Vec<Column>,
551    pub constraints: Vec<TableConstraint>,
552    /// Table options in `WITH`. All keys are lowercase.
553    pub options: OptionMap,
554    pub if_not_exists: bool,
555    pub engine: String,
556}
557
558impl Display for CreateExternalTable {
559    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
560        write!(f, "CREATE EXTERNAL TABLE ")?;
561        if self.if_not_exists {
562            write!(f, "IF NOT EXISTS ")?;
563        }
564        writeln!(f, "{} (", &self.name)?;
565        writeln!(f, "{},", format_list_indent!(self.columns))?;
566        writeln!(f, "{}", format_table_constraint(&self.constraints))?;
567        writeln!(f, ")")?;
568        writeln!(f, "ENGINE={}", &self.engine)?;
569        if !self.options.is_empty() {
570            let options = self.options.kv_pairs();
571            write!(f, "WITH(\n{}\n)", format_list_indent!(options))?;
572        }
573        Ok(())
574    }
575}
576
577#[derive(Debug, PartialEq, Eq, Clone, Visit, VisitMut, Serialize)]
578pub struct CreateTableLike {
579    /// Table name
580    pub table_name: ObjectName,
581    /// The table that is designated to be imitated by `Like`
582    pub source_name: ObjectName,
583}
584
585impl Display for CreateTableLike {
586    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
587        let table_name = &self.table_name;
588        let source_name = &self.source_name;
589        write!(f, r#"CREATE TABLE {table_name} LIKE {source_name}"#)
590    }
591}
592
593#[derive(Debug, PartialEq, Eq, Clone, Visit, VisitMut, Serialize)]
594pub struct CreateFlow {
595    /// Flow name
596    pub flow_name: ObjectName,
597    /// Output (sink) table name
598    pub sink_table_name: ObjectName,
599    /// Whether to replace existing task
600    pub or_replace: bool,
601    /// Create if not exist
602    pub if_not_exists: bool,
603    /// `EXPIRE AFTER`
604    /// Duration in second as `i64`
605    pub expire_after: Option<i64>,
606    /// Duration for flow evaluation interval
607    /// Duration in seconds as `i64`
608    /// If not set, flow will be evaluated based on time window size and other args.
609    pub eval_interval: Option<i64>,
610    /// Phase offset of the flow evaluation schedule within `eval_interval`.
611    /// Duration in seconds as `i64`.
612    /// Must be in range `[0, eval_interval)`. Only legal together with
613    /// `eval_interval`. A value of zero (the default) means the schedule is
614    /// anchored to the Unix epoch, i.e. phases at `k * eval_interval`.
615    pub eval_offset: Option<i64>,
616    /// Comment string
617    pub comment: Option<String>,
618    /// Flow creation options from `WITH (...)`
619    pub flow_options: OptionMap,
620    /// SQL statement
621    pub query: Box<SqlOrTql>,
622}
623
624/// Either a sql query or a tql query
625#[derive(Debug, PartialEq, Eq, Clone, Visit, VisitMut, Serialize)]
626pub enum SqlOrTql {
627    Sql(GtQuery, String),
628    Tql(Tql, String),
629}
630
631impl std::fmt::Display for SqlOrTql {
632    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
633        match self {
634            Self::Sql(_, s) => write!(f, "{}", s),
635            Self::Tql(_, s) => write!(f, "{}", s),
636        }
637    }
638}
639
640impl SqlOrTql {
641    pub fn try_from_statement(
642        value: Statement,
643        original_query: &str,
644    ) -> std::result::Result<Self, crate::error::Error> {
645        match value {
646            Statement::Query(query) => Ok(Self::Sql(*query, original_query.to_string())),
647            Statement::Tql(tql) => Ok(Self::Tql(tql, original_query.to_string())),
648            _ => InvalidFlowQuerySnafu {
649                reason: format!("Expect either sql query or promql query, found {:?}", value),
650            }
651            .fail(),
652        }
653    }
654}
655
656impl Display for CreateFlow {
657    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
658        write!(f, "CREATE ")?;
659        if self.or_replace {
660            write!(f, "OR REPLACE ")?;
661        }
662        write!(f, "FLOW ")?;
663        if self.if_not_exists {
664            write!(f, "IF NOT EXISTS ")?;
665        }
666        writeln!(f, "{}", &self.flow_name)?;
667        writeln!(f, "SINK TO {}", &self.sink_table_name)?;
668        if let Some(expire_after) = &self.expire_after {
669            writeln!(f, "EXPIRE AFTER '{} s'", expire_after)?;
670        }
671        if let Some(eval_interval) = &self.eval_interval {
672            writeln!(f, "EVAL INTERVAL '{} s'", eval_interval)?;
673        }
674        // Canonical display: omit a zero offset (equivalent to the default
675        // epoch-anchored schedule). Non-zero offsets are always emitted.
676        if let Some(eval_offset) = &self.eval_offset
677            && *eval_offset != 0
678        {
679            writeln!(f, "EVAL OFFSET '{} s'", eval_offset)?;
680        }
681        if let Some(comment) = &self.comment {
682            writeln!(f, "COMMENT '{}'", comment)?;
683        }
684        if !self.flow_options.is_empty() {
685            let options = self.flow_options.kv_pairs();
686            writeln!(f, "WITH ({})", format_list_comma!(options))?;
687        }
688        write!(f, "AS {}", &self.query)
689    }
690}
691
692/// Create SQL view statement.
693#[derive(Debug, PartialEq, Eq, Clone, Visit, VisitMut, Serialize)]
694pub struct CreateView {
695    /// View name
696    pub name: ObjectName,
697    /// An optional list of names to be used for columns of the view
698    pub columns: Vec<Ident>,
699    /// The clause after `As` that defines the VIEW.
700    /// Can only be either [Statement::Query] or [Statement::Tql].
701    pub query: Box<Statement>,
702    /// Whether to replace existing VIEW
703    pub or_replace: bool,
704    /// Create VIEW only when it doesn't exists
705    pub if_not_exists: bool,
706}
707
708impl Display for CreateView {
709    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
710        write!(f, "CREATE ")?;
711        if self.or_replace {
712            write!(f, "OR REPLACE ")?;
713        }
714        write!(f, "VIEW ")?;
715        if self.if_not_exists {
716            write!(f, "IF NOT EXISTS ")?;
717        }
718        write!(f, "{} ", &self.name)?;
719        if !self.columns.is_empty() {
720            write!(f, "({}) ", format_list_comma!(self.columns))?;
721        }
722        write!(f, "AS {}", &self.query)
723    }
724}
725
726#[cfg(test)]
727mod tests {
728    use std::assert_matches;
729
730    use datatypes::json::{JsonSettings, JsonTypeHint as DatatypeJsonTypeHint};
731    use datatypes::prelude::ConcreteDataType;
732
733    use super::*;
734    use crate::dialect::GreptimeDbDialect;
735    use crate::error::Error;
736    use crate::parser::{ParseOptions, ParserContext};
737    use crate::statements::statement::Statement;
738
739    #[test]
740    fn test_display_create_table() {
741        let sql = r"create table if not exists demo(
742                             host string,
743                             ts timestamp,
744                             cpu double default 0,
745                             memory double,
746                             TIME INDEX (ts),
747                             PRIMARY KEY(host)
748                       )
749                       PARTITION ON COLUMNS (host) (
750                            host = 'a',
751                            host > 'a',
752                       )
753                       engine=mito
754                       with(ttl='7d', storage='File');
755         ";
756        let result =
757            ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
758                .unwrap();
759        assert_eq!(1, result.len());
760
761        match &result[0] {
762            Statement::CreateTable(c) => {
763                let new_sql = format!("\n{}", c);
764                assert_eq!(
765                    r#"
766CREATE TABLE IF NOT EXISTS demo (
767  host STRING,
768  ts TIMESTAMP,
769  cpu DOUBLE DEFAULT 0,
770  memory DOUBLE,
771  TIME INDEX (ts),
772  PRIMARY KEY (host)
773)
774PARTITION ON COLUMNS (host) (
775  host = 'a',
776  host > 'a'
777)
778ENGINE=mito
779WITH(
780  storage = 'File',
781  ttl = '7d'
782)"#,
783                    &new_sql
784                );
785
786                let new_result = ParserContext::create_with_dialect(
787                    &new_sql,
788                    &GreptimeDbDialect {},
789                    ParseOptions::default(),
790                )
791                .unwrap();
792                assert_eq!(result, new_result);
793            }
794            _ => unreachable!(),
795        }
796    }
797
798    #[test]
799    fn test_display_empty_partition_column() {
800        let sql = r"create table if not exists demo(
801            host string,
802            ts timestamp,
803            cpu double default 0,
804            memory double,
805            TIME INDEX (ts),
806            PRIMARY KEY(ts, host)
807            );
808        ";
809        let result =
810            ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
811                .unwrap();
812        assert_eq!(1, result.len());
813
814        match &result[0] {
815            Statement::CreateTable(c) => {
816                let new_sql = format!("\n{}", c);
817                assert_eq!(
818                    r#"
819CREATE TABLE IF NOT EXISTS demo (
820  host STRING,
821  ts TIMESTAMP,
822  cpu DOUBLE DEFAULT 0,
823  memory DOUBLE,
824  TIME INDEX (ts),
825  PRIMARY KEY (ts, host)
826)
827ENGINE=mito
828"#,
829                    &new_sql
830                );
831
832                let new_result = ParserContext::create_with_dialect(
833                    &new_sql,
834                    &GreptimeDbDialect {},
835                    ParseOptions::default(),
836                )
837                .unwrap();
838                assert_eq!(result, new_result);
839            }
840            _ => unreachable!(),
841        }
842    }
843
844    #[test]
845    fn test_validate_table_options() {
846        let sql = r"create table if not exists demo(
847            host string,
848            ts timestamp,
849            cpu double default 0,
850            memory double,
851            TIME INDEX (ts),
852            PRIMARY KEY(host)
853      )
854      PARTITION ON COLUMNS (host) ()
855      engine=mito
856      with(ttl='7d', 'compaction.type'='world');
857";
858        let result =
859            ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
860                .unwrap();
861        match &result[0] {
862            Statement::CreateTable(c) => {
863                assert_eq!(2, c.options.len());
864            }
865            _ => unreachable!(),
866        }
867
868        let sql = r"create table if not exists demo(
869            host string,
870            ts timestamp,
871            cpu double default 0,
872            memory double,
873            TIME INDEX (ts),
874            PRIMARY KEY(host)
875      )
876      PARTITION ON COLUMNS (host) ()
877      engine=mito
878      with(ttl='7d', hello='world');
879";
880        let result =
881            ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default());
882        assert_matches!(result, Err(Error::InvalidTableOption { .. }));
883
884        // A whitelisted semantic key with an in-domain value is accepted.
885        let semantic = |with: &str| {
886            let sql =
887                format!("create table demo(host string, ts timestamp time index) with({with});");
888            ParserContext::create_with_dialect(&sql, &GreptimeDbDialect {}, ParseOptions::default())
889        };
890        assert!(semantic("'greptime.semantic.signal_type'='metric'").is_ok());
891        // An out-of-domain value is rejected.
892        assert_matches!(
893            semantic("'greptime.semantic.signal_type'='spans'"),
894            Err(Error::InvalidTableOption { .. })
895        );
896        // An unknown key under the semantic prefix is rejected.
897        assert_matches!(
898            semantic("'greptime.semantic.bogus'='x'"),
899            Err(Error::InvalidTableOption { .. })
900        );
901    }
902
903    #[test]
904    fn test_display_json2_type_hints_quotes_path_segments() {
905        let sql = r#"CREATE TABLE traces (
906            log_json_data JSON2 (
907                "service.name" STRING,
908                "a.b"."c" BIGINT,
909                a."b.c" STRING
910            ),
911            ts TIMESTAMP TIME INDEX
912        )"#;
913        let result =
914            ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
915                .unwrap();
916
917        match &result[0] {
918            Statement::CreateTable(c) => {
919                let new_sql = format!("\n{}", c);
920                assert_eq!(
921                    r#"
922CREATE TABLE traces (
923  log_json_data JSON2(
924    "service.name" STRING,
925    "a.b"."c" BIGINT,
926    "a"."b.c" STRING
927  ),
928  ts TIMESTAMP NOT NULL,
929  TIME INDEX (ts)
930)
931ENGINE=mito
932"#,
933                    &new_sql
934                );
935
936                let new_result = ParserContext::create_with_dialect(
937                    &new_sql,
938                    &GreptimeDbDialect {},
939                    ParseOptions::default(),
940                )
941                .unwrap();
942                assert_eq!(result, new_result);
943            }
944            _ => unreachable!(),
945        }
946    }
947
948    #[test]
949    fn test_parse_json2_max_auto_expanded_paths_option() -> Result<()> {
950        let sql = r#"CREATE TABLE traces (
951            log_json_data JSON2 (
952                status_code BIGINT,
953                max_auto_expanded_paths = 1
954            ),
955            ts TIMESTAMP TIME INDEX
956        )"#;
957        let result = ParserContext::create_with_dialect(
958            sql,
959            &GreptimeDbDialect {},
960            ParseOptions::default(),
961        )?;
962        let Statement::CreateTable(create_table) = &result[0] else {
963            unreachable!()
964        };
965        let settings = create_table.columns[0]
966            .extensions
967            .build_json_settings()?
968            .unwrap();
969        assert_eq!(settings.max_auto_expanded_paths(), Some(1));
970        Ok(())
971    }
972
973    #[test]
974    fn test_display_json2_type_hints_quotes_numeric_segments() {
975        let sql = r#"CREATE TABLE traces (
976            log_json_data JSON2 (
977                "1abc" STRING,
978                a."2b" BIGINT
979            ),
980            ts TIMESTAMP TIME INDEX
981        )"#;
982        let result =
983            ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
984                .unwrap();
985
986        match &result[0] {
987            Statement::CreateTable(c) => {
988                let new_sql = format!("\n{}", c);
989                assert_eq!(
990                    r#"
991CREATE TABLE traces (
992  log_json_data JSON2(
993    "1abc" STRING,
994    "a"."2b" BIGINT
995  ),
996  ts TIMESTAMP NOT NULL,
997  TIME INDEX (ts)
998)
999ENGINE=mito
1000"#,
1001                    &new_sql
1002                );
1003
1004                let new_result = ParserContext::create_with_dialect(
1005                    &new_sql,
1006                    &GreptimeDbDialect {},
1007                    ParseOptions::default(),
1008                )
1009                .unwrap();
1010                assert_eq!(result, new_result);
1011            }
1012            _ => unreachable!(),
1013        }
1014    }
1015
1016    #[test]
1017    fn test_json2_type_hint_rejects_default() {
1018        let sql = r#"CREATE TABLE traces (
1019            log_json_data JSON2 (
1020                status_code BIGINT DEFAULT -5,
1021                duration DOUBLE DEFAULT +1.5,
1022                error BOOLEAN DEFAULT false,
1023                message STRING DEFAULT 'unknown'
1024            ),
1025            ts TIMESTAMP TIME INDEX
1026        )"#;
1027        let err =
1028            ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
1029                .unwrap_err();
1030        assert!(err.to_string().contains("DEFAULT is not supported"));
1031    }
1032
1033    #[test]
1034    fn test_json2_type_hint_rejects_not_null() {
1035        let sql = r#"CREATE TABLE traces (
1036            log_json_data JSON2 (
1037                status_code BIGINT NOT NULL DEFAULT NULL
1038            ),
1039            ts TIMESTAMP TIME INDEX
1040        )"#;
1041        let err =
1042            ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
1043                .unwrap_err();
1044        assert!(err.to_string().contains("NULL/NOT NULL is not supported"));
1045    }
1046
1047    #[test]
1048    fn test_set_json_settings_preserves_type_hint_sql_types() -> Result<()> {
1049        let mut extensions = super::ColumnExtensions::default();
1050        let settings = JsonSettings::try_new(
1051            vec![
1052                DatatypeJsonTypeHint {
1053                    path: vec!["i".to_string()],
1054                    data_type: ConcreteDataType::int64_datatype(),
1055                    inverted_index: false,
1056                },
1057                DatatypeJsonTypeHint {
1058                    path: vec!["f".to_string()],
1059                    data_type: ConcreteDataType::float64_datatype(),
1060                    inverted_index: false,
1061                },
1062                DatatypeJsonTypeHint {
1063                    path: vec!["u".to_string()],
1064                    data_type: ConcreteDataType::uint64_datatype(),
1065                    inverted_index: false,
1066                },
1067                DatatypeJsonTypeHint {
1068                    path: vec!["s".to_string()],
1069                    data_type: ConcreteDataType::string_datatype(),
1070                    inverted_index: false,
1071                },
1072                DatatypeJsonTypeHint {
1073                    path: vec!["b".to_string()],
1074                    data_type: ConcreteDataType::boolean_datatype(),
1075                    inverted_index: false,
1076                },
1077            ],
1078            None,
1079        )?;
1080        extensions.set_json_settings(settings)?;
1081
1082        assert_eq!(
1083            extensions
1084                .json2_options
1085                .unwrap()
1086                .type_hints
1087                .iter()
1088                .map(|hint| hint.data_type.to_string())
1089                .collect::<Vec<_>>(),
1090            vec!["BIGINT", "DOUBLE", "BIGINT UNSIGNED", "STRING", "BOOLEAN"]
1091        );
1092        Ok(())
1093    }
1094
1095    #[test]
1096    fn test_set_json_settings_rejects_unsupported_type_hint_type() -> Result<()> {
1097        let err = JsonSettings::try_new(
1098            vec![DatatypeJsonTypeHint {
1099                path: vec!["u".to_string()],
1100                data_type: ConcreteDataType::date_datatype(),
1101                inverted_index: false,
1102            }],
1103            None,
1104        )
1105        .unwrap_err();
1106
1107        assert!(
1108            err.to_string()
1109                .contains("unsupported JSON2 type hint data type")
1110        );
1111        Ok(())
1112    }
1113
1114    #[test]
1115    fn test_set_empty_json_settings_omits_json2_options() -> Result<()> {
1116        let mut extensions = ColumnExtensions::default();
1117        extensions.set_json_settings(JsonSettings::default())?;
1118        assert!(extensions.json2_options.is_none());
1119        Ok(())
1120    }
1121
1122    #[test]
1123    fn test_display_create_database() {
1124        let sql = r"create database test;";
1125        let stmts =
1126            ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
1127                .unwrap();
1128        assert_eq!(1, stmts.len());
1129        assert_matches!(&stmts[0], Statement::CreateDatabase { .. });
1130
1131        match &stmts[0] {
1132            Statement::CreateDatabase(set) => {
1133                let new_sql = format!("\n{}", set);
1134                assert_eq!(
1135                    r#"
1136CREATE DATABASE test"#,
1137                    &new_sql
1138                );
1139            }
1140            _ => {
1141                unreachable!();
1142            }
1143        }
1144
1145        let sql = r"create database if not exists test;";
1146        let stmts =
1147            ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
1148                .unwrap();
1149        assert_eq!(1, stmts.len());
1150        assert_matches!(&stmts[0], Statement::CreateDatabase { .. });
1151
1152        match &stmts[0] {
1153            Statement::CreateDatabase(set) => {
1154                let new_sql = format!("\n{}", set);
1155                assert_eq!(
1156                    r#"
1157CREATE DATABASE IF NOT EXISTS test"#,
1158                    &new_sql
1159                );
1160            }
1161            _ => {
1162                unreachable!();
1163            }
1164        }
1165
1166        let sql = r#"CREATE DATABASE IF NOT EXISTS test WITH (ttl='1h');"#;
1167        let stmts =
1168            ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
1169                .unwrap();
1170        assert_eq!(1, stmts.len());
1171        assert_matches!(&stmts[0], Statement::CreateDatabase { .. });
1172
1173        match &stmts[0] {
1174            Statement::CreateDatabase(set) => {
1175                let new_sql = format!("\n{}", set);
1176                assert_eq!(
1177                    r#"
1178CREATE DATABASE IF NOT EXISTS test
1179WITH(
1180  ttl = '1h'
1181)"#,
1182                    &new_sql
1183                );
1184            }
1185            _ => {
1186                unreachable!();
1187            }
1188        }
1189    }
1190
1191    #[test]
1192    fn test_display_create_table_like() {
1193        let sql = r"create table t2 like t1;";
1194        let stmts =
1195            ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
1196                .unwrap();
1197        assert_eq!(1, stmts.len());
1198        assert_matches!(&stmts[0], Statement::CreateTableLike { .. });
1199
1200        match &stmts[0] {
1201            Statement::CreateTableLike(create) => {
1202                let new_sql = format!("\n{}", create);
1203                assert_eq!(
1204                    r#"
1205CREATE TABLE t2 LIKE t1"#,
1206                    &new_sql
1207                );
1208            }
1209            _ => {
1210                unreachable!();
1211            }
1212        }
1213    }
1214
1215    #[test]
1216    fn test_display_create_external_table() {
1217        let sql = r#"CREATE EXTERNAL TABLE city (
1218            host string,
1219            ts timestamp,
1220            cpu float64 default 0,
1221            memory float64,
1222            TIME INDEX (ts),
1223            PRIMARY KEY(host)
1224) WITH (location='/var/data/city.csv', format='csv');"#;
1225        let stmts =
1226            ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
1227                .unwrap();
1228        assert_eq!(1, stmts.len());
1229        assert_matches!(&stmts[0], Statement::CreateExternalTable { .. });
1230
1231        match &stmts[0] {
1232            Statement::CreateExternalTable(create) => {
1233                let new_sql = format!("\n{}", create);
1234                assert_eq!(
1235                    r#"
1236CREATE EXTERNAL TABLE city (
1237  host STRING,
1238  ts TIMESTAMP,
1239  cpu DOUBLE DEFAULT 0,
1240  memory DOUBLE,
1241  TIME INDEX (ts),
1242  PRIMARY KEY (host)
1243)
1244ENGINE=file
1245WITH(
1246  format = 'csv',
1247  location = '/var/data/city.csv'
1248)"#,
1249                    &new_sql
1250                );
1251            }
1252            _ => {
1253                unreachable!();
1254            }
1255        }
1256    }
1257
1258    #[test]
1259    fn test_display_create_flow() {
1260        let sql = r"CREATE FLOW filter_numbers
1261            SINK TO out_num_cnt
1262            AS SELECT number FROM numbers_input where number > 10;";
1263        let result =
1264            ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
1265                .unwrap();
1266        assert_eq!(1, result.len());
1267
1268        match &result[0] {
1269            Statement::CreateFlow(c) => {
1270                let new_sql = format!("\n{}", c);
1271                assert_eq!(
1272                    r#"
1273CREATE FLOW filter_numbers
1274SINK TO out_num_cnt
1275AS SELECT number FROM numbers_input where number > 10"#,
1276                    &new_sql
1277                );
1278
1279                let new_result = ParserContext::create_with_dialect(
1280                    &new_sql,
1281                    &GreptimeDbDialect {},
1282                    ParseOptions::default(),
1283                )
1284                .unwrap();
1285                assert_eq!(result, new_result);
1286            }
1287            _ => unreachable!(),
1288        }
1289    }
1290
1291    #[test]
1292    fn test_vector_index_options_validation() {
1293        use super::{ColumnExtensions, OptionMap};
1294
1295        // Test zero connectivity should fail
1296        let extensions = ColumnExtensions {
1297            vector_index_options: Some(OptionMap::from([(
1298                "connectivity".to_string(),
1299                "0".to_string(),
1300            )])),
1301            ..Default::default()
1302        };
1303        let result = extensions.build_vector_index_options();
1304        assert!(result.is_err());
1305        assert!(
1306            result
1307                .unwrap_err()
1308                .to_string()
1309                .contains("connectivity must be in the range [2, 2048]")
1310        );
1311
1312        // Test zero expansion_add should fail
1313        let extensions = ColumnExtensions {
1314            vector_index_options: Some(OptionMap::from([(
1315                "expansion_add".to_string(),
1316                "0".to_string(),
1317            )])),
1318            ..Default::default()
1319        };
1320        let result = extensions.build_vector_index_options();
1321        assert!(result.is_err());
1322        assert!(
1323            result
1324                .unwrap_err()
1325                .to_string()
1326                .contains("expansion_add must be greater than 0")
1327        );
1328
1329        // Test zero expansion_search should fail
1330        let extensions = ColumnExtensions {
1331            vector_index_options: Some(OptionMap::from([(
1332                "expansion_search".to_string(),
1333                "0".to_string(),
1334            )])),
1335            ..Default::default()
1336        };
1337        let result = extensions.build_vector_index_options();
1338        assert!(result.is_err());
1339        assert!(
1340            result
1341                .unwrap_err()
1342                .to_string()
1343                .contains("expansion_search must be greater than 0")
1344        );
1345
1346        // Test valid values should succeed
1347        let extensions = ColumnExtensions {
1348            vector_index_options: Some(OptionMap::from([
1349                ("connectivity".to_string(), "32".to_string()),
1350                ("expansion_add".to_string(), "200".to_string()),
1351                ("expansion_search".to_string(), "100".to_string()),
1352            ])),
1353            ..Default::default()
1354        };
1355        let result = extensions.build_vector_index_options();
1356        assert!(result.is_ok());
1357        let options = result.unwrap().unwrap();
1358        assert_eq!(options.connectivity, 32);
1359        assert_eq!(options.expansion_add, 200);
1360        assert_eq!(options.expansion_search, 100);
1361    }
1362}
sql/statements/create.rs

sql/statements/
create.rs
