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

sql/statements/
create.rs
