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

sql/statements/
create.rs
