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

sql/statements/
create.rs
