query/sql/
show_create_table.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
15//! Implementation of `SHOW CREATE TABLE` statement.
16
17use std::collections::HashMap;
18
19use arrow_schema::extension::ExtensionType;
20use common_meta::SchemaOptions;
21use datatypes::extension::json::JsonExtensionType;
22use datatypes::schema::{
23    COLUMN_FULLTEXT_OPT_KEY_ANALYZER, COLUMN_FULLTEXT_OPT_KEY_BACKEND,
24    COLUMN_FULLTEXT_OPT_KEY_CASE_SENSITIVE, COLUMN_FULLTEXT_OPT_KEY_FALSE_POSITIVE_RATE,
25    COLUMN_FULLTEXT_OPT_KEY_GRANULARITY, COLUMN_SKIPPING_INDEX_OPT_KEY_FALSE_POSITIVE_RATE,
26    COLUMN_SKIPPING_INDEX_OPT_KEY_GRANULARITY, COLUMN_SKIPPING_INDEX_OPT_KEY_TYPE, COMMENT_KEY,
27    ColumnDefaultConstraint, ColumnSchema, FulltextBackend, SchemaRef,
28};
29use snafu::ResultExt;
30use sql::ast::{ColumnDef, ColumnOption, ColumnOptionDef, Expr, Ident, ObjectName};
31use sql::dialect::GreptimeDbDialect;
32use sql::parser::ParserContext;
33use sql::statements::create::{Column, ColumnExtensions, CreateTable, TableConstraint};
34use sql::statements::{self, OptionMap};
35use store_api::metric_engine_consts::{is_metric_engine, is_metric_engine_internal_column};
36use table::metadata::{TableInfoRef, TableMeta};
37use table::requests::{
38    COMMENT_KEY as TABLE_COMMENT_KEY, FILE_TABLE_META_KEY, TTL_KEY, WRITE_BUFFER_SIZE_KEY,
39};
40
41use crate::error::{
42    ConvertSqlTypeSnafu, ConvertSqlValueSnafu, GetFulltextOptionsSnafu,
43    GetSkippingIndexOptionsSnafu, Result, SqlSnafu,
44};
45
46/// Generates CREATE TABLE options from given table metadata and schema-level options.
47fn create_sql_options(table_meta: &TableMeta, schema_options: Option<SchemaOptions>) -> OptionMap {
48    let table_opts = &table_meta.options;
49    let mut options = OptionMap::default();
50    if let Some(write_buffer_size) = table_opts.write_buffer_size {
51        options.insert(
52            WRITE_BUFFER_SIZE_KEY.to_string(),
53            write_buffer_size.to_string(),
54        );
55    }
56    if let Some(ttl) = table_opts.ttl.map(|t| t.to_string()) {
57        options.insert(TTL_KEY.to_string(), ttl);
58    } else if let Some(database_ttl) = schema_options
59        .as_ref()
60        .and_then(|o| o.ttl)
61        .map(|ttl| ttl.to_string())
62    {
63        options.insert(TTL_KEY.to_string(), database_ttl);
64    };
65
66    for (k, v) in table_opts
67        .extra_options
68        .iter()
69        .filter(|(k, _)| k != &FILE_TABLE_META_KEY)
70    {
71        options.insert(k.clone(), v.clone());
72    }
73    options
74}
75
76#[inline]
77fn column_option_def(option: ColumnOption) -> ColumnOptionDef {
78    ColumnOptionDef { name: None, option }
79}
80
81fn create_column(column_schema: &ColumnSchema, quote_style: char) -> Result<Column> {
82    let name = &column_schema.name;
83    let mut options = Vec::with_capacity(2);
84    let mut extensions = ColumnExtensions::default();
85
86    if column_schema.is_nullable() {
87        options.push(column_option_def(ColumnOption::Null));
88    } else {
89        options.push(column_option_def(ColumnOption::NotNull));
90    }
91
92    if let Some(c) = column_schema.default_constraint() {
93        let expr = match c {
94            ColumnDefaultConstraint::Value(v) => Expr::Value(
95                statements::value_to_sql_value(v)
96                    .with_context(|_| ConvertSqlValueSnafu { value: v.clone() })?
97                    .into(),
98            ),
99            ColumnDefaultConstraint::Function(expr) => {
100                ParserContext::parse_function(expr, &GreptimeDbDialect {}).context(SqlSnafu)?
101            }
102        };
103
104        options.push(column_option_def(ColumnOption::Default(expr)));
105    }
106
107    if let Some(c) = column_schema.metadata().get(COMMENT_KEY) {
108        options.push(column_option_def(ColumnOption::Comment(c.clone())));
109    }
110
111    if let Some(opt) = column_schema
112        .fulltext_options()
113        .context(GetFulltextOptionsSnafu)?
114        && opt.enable
115    {
116        let mut map = HashMap::from([
117            (
118                COLUMN_FULLTEXT_OPT_KEY_ANALYZER.to_string(),
119                opt.analyzer.to_string(),
120            ),
121            (
122                COLUMN_FULLTEXT_OPT_KEY_CASE_SENSITIVE.to_string(),
123                opt.case_sensitive.to_string(),
124            ),
125            (
126                COLUMN_FULLTEXT_OPT_KEY_BACKEND.to_string(),
127                opt.backend.to_string(),
128            ),
129        ]);
130        if opt.backend == FulltextBackend::Bloom {
131            map.insert(
132                COLUMN_FULLTEXT_OPT_KEY_GRANULARITY.to_string(),
133                opt.granularity.to_string(),
134            );
135            map.insert(
136                COLUMN_FULLTEXT_OPT_KEY_FALSE_POSITIVE_RATE.to_string(),
137                opt.false_positive_rate().to_string(),
138            );
139        }
140        extensions.fulltext_index_options = Some(map.into());
141    }
142
143    if let Some(opt) = column_schema
144        .skipping_index_options()
145        .context(GetSkippingIndexOptionsSnafu)?
146    {
147        let map = HashMap::from([
148            (
149                COLUMN_SKIPPING_INDEX_OPT_KEY_GRANULARITY.to_string(),
150                opt.granularity.to_string(),
151            ),
152            (
153                COLUMN_SKIPPING_INDEX_OPT_KEY_FALSE_POSITIVE_RATE.to_string(),
154                opt.false_positive_rate().to_string(),
155            ),
156            (
157                COLUMN_SKIPPING_INDEX_OPT_KEY_TYPE.to_string(),
158                opt.index_type.to_string(),
159            ),
160        ]);
161        extensions.skipping_index_options = Some(map.into());
162    }
163
164    if column_schema.is_inverted_indexed() {
165        extensions.inverted_index_options = Some(HashMap::new().into());
166    }
167
168    if let Some(json_extension) = column_schema.extension_type::<JsonExtensionType>()? {
169        let settings = json_extension
170            .metadata()
171            .json_structure_settings
172            .clone()
173            .unwrap_or_default();
174        extensions.set_json_structure_settings(settings);
175    }
176
177    Ok(Column {
178        column_def: ColumnDef {
179            name: Ident::with_quote(quote_style, name),
180            data_type: statements::concrete_data_type_to_sql_data_type(&column_schema.data_type)
181                .with_context(|_| ConvertSqlTypeSnafu {
182                    datatype: column_schema.data_type.clone(),
183                })?,
184            options,
185        },
186        extensions,
187    })
188}
189
190/// Returns the primary key columns for `SHOW CREATE TABLE` statement.
191///
192/// For metric engine, it will only return the primary key columns that are not internal columns.
193fn primary_key_columns_for_show_create<'a>(
194    table_meta: &'a TableMeta,
195    engine: &str,
196) -> Vec<&'a String> {
197    let is_metric_engine = is_metric_engine(engine);
198    if is_metric_engine {
199        table_meta
200            .row_key_column_names()
201            .filter(|name| !is_metric_engine_internal_column(name))
202            .collect()
203    } else {
204        table_meta.row_key_column_names().collect()
205    }
206}
207
208fn create_table_constraints(
209    engine: &str,
210    schema: &SchemaRef,
211    table_meta: &TableMeta,
212    quote_style: char,
213) -> Vec<TableConstraint> {
214    let mut constraints = Vec::with_capacity(2);
215    if let Some(timestamp_column) = schema.timestamp_column() {
216        let column_name = &timestamp_column.name;
217        constraints.push(TableConstraint::TimeIndex {
218            column: Ident::with_quote(quote_style, column_name),
219        });
220    }
221    if !table_meta.primary_key_indices.is_empty() {
222        let columns = primary_key_columns_for_show_create(table_meta, engine)
223            .into_iter()
224            .map(|name| Ident::with_quote(quote_style, name))
225            .collect();
226        constraints.push(TableConstraint::PrimaryKey { columns });
227    }
228
229    constraints
230}
231
232/// Create a CreateTable statement from table info.
233pub fn create_table_stmt(
234    table_info: &TableInfoRef,
235    schema_options: Option<SchemaOptions>,
236    quote_style: char,
237) -> Result<CreateTable> {
238    let table_meta = &table_info.meta;
239    let table_name = &table_info.name;
240    let schema = &table_info.meta.schema;
241    let is_metric_engine = is_metric_engine(&table_meta.engine);
242    let columns = schema
243        .column_schemas()
244        .iter()
245        .filter_map(|c| {
246            if is_metric_engine && is_metric_engine_internal_column(&c.name) {
247                None
248            } else {
249                Some(create_column(c, quote_style))
250            }
251        })
252        .collect::<Result<Vec<_>>>()?;
253
254    let constraints = create_table_constraints(&table_meta.engine, schema, table_meta, quote_style);
255
256    let mut options = create_sql_options(table_meta, schema_options);
257    if let Some(comment) = &table_info.desc
258        && options.get(TABLE_COMMENT_KEY).is_none()
259    {
260        options.insert(format!("'{TABLE_COMMENT_KEY}'"), comment.clone());
261    }
262
263    Ok(CreateTable {
264        if_not_exists: true,
265        table_id: table_info.ident.table_id,
266        name: ObjectName::from(vec![Ident::with_quote(quote_style, table_name)]),
267        columns,
268        engine: table_meta.engine.clone(),
269        constraints,
270        options,
271        partitions: None,
272    })
273}
274
275#[cfg(test)]
276mod tests {
277    use std::sync::Arc;
278    use std::time::Duration;
279
280    use common_time::timestamp::TimeUnit;
281    use datatypes::prelude::ConcreteDataType;
282    use datatypes::schema::{FulltextOptions, Schema, SchemaRef, SkippingIndexOptions};
283    use table::metadata::*;
284    use table::requests::{
285        FILE_TABLE_FORMAT_KEY, FILE_TABLE_LOCATION_KEY, FILE_TABLE_META_KEY, TableOptions,
286    };
287
288    use super::*;
289
290    #[test]
291    fn test_show_create_table_sql() {
292        let schema = vec![
293            ColumnSchema::new("id", ConcreteDataType::uint32_datatype(), true)
294                .with_skipping_options(SkippingIndexOptions {
295                    granularity: 4096,
296                    ..Default::default()
297                })
298                .unwrap(),
299            ColumnSchema::new("host", ConcreteDataType::string_datatype(), true)
300                .with_inverted_index(true),
301            ColumnSchema::new("cpu", ConcreteDataType::float64_datatype(), true),
302            ColumnSchema::new("disk", ConcreteDataType::float32_datatype(), true),
303            ColumnSchema::new("msg", ConcreteDataType::string_datatype(), true)
304                .with_fulltext_options(FulltextOptions {
305                    enable: true,
306                    ..Default::default()
307                })
308                .unwrap(),
309            ColumnSchema::new(
310                "ts",
311                ConcreteDataType::timestamp_datatype(TimeUnit::Millisecond),
312                false,
313            )
314            .with_default_constraint(Some(ColumnDefaultConstraint::Function(String::from(
315                "current_timestamp()",
316            ))))
317            .unwrap()
318            .with_time_index(true),
319        ];
320
321        let table_schema = SchemaRef::new(Schema::new(schema));
322        let table_name = "system_metrics";
323        let schema_name = "public".to_string();
324        let catalog_name = "greptime".to_string();
325
326        let mut options = table::requests::TableOptions {
327            ttl: Some(Duration::from_secs(30).into()),
328            ..Default::default()
329        };
330
331        let _ = options
332            .extra_options
333            .insert("compaction.type".to_string(), "twcs".to_string());
334
335        let meta = TableMetaBuilder::empty()
336            .schema(table_schema)
337            .primary_key_indices(vec![0, 1])
338            .value_indices(vec![2, 3])
339            .engine("mito".to_string())
340            .next_column_id(0)
341            .options(options)
342            .created_on(Default::default())
343            .build()
344            .unwrap();
345
346        let info = Arc::new(
347            TableInfoBuilder::default()
348                .table_id(1024)
349                .table_version(0 as TableVersion)
350                .name(table_name)
351                .schema_name(schema_name)
352                .catalog_name(catalog_name)
353                .desc(None)
354                .table_type(TableType::Base)
355                .meta(meta)
356                .build()
357                .unwrap(),
358        );
359
360        let stmt = create_table_stmt(&info, None, '"').unwrap();
361
362        let sql = format!("\n{}", stmt);
363        assert_eq!(
364            r#"
365CREATE TABLE IF NOT EXISTS "system_metrics" (
366  "id" INT UNSIGNED NULL SKIPPING INDEX WITH(false_positive_rate = '0.01', granularity = '4096', type = 'BLOOM'),
367  "host" STRING NULL INVERTED INDEX,
368  "cpu" DOUBLE NULL,
369  "disk" FLOAT NULL,
370  "msg" STRING NULL FULLTEXT INDEX WITH(analyzer = 'English', backend = 'bloom', case_sensitive = 'false', false_positive_rate = '0.01', granularity = '10240'),
371  "ts" TIMESTAMP(3) NOT NULL DEFAULT current_timestamp(),
372  TIME INDEX ("ts"),
373  PRIMARY KEY ("id", "host")
374)
375ENGINE=mito
376WITH(
377  'compaction.type' = 'twcs',
378  ttl = '30s'
379)"#,
380            sql
381        );
382    }
383
384    #[test]
385    fn test_show_create_external_table_sql() {
386        let schema = vec![
387            ColumnSchema::new("host", ConcreteDataType::string_datatype(), true),
388            ColumnSchema::new("cpu", ConcreteDataType::float64_datatype(), true),
389        ];
390        let table_schema = SchemaRef::new(Schema::new(schema));
391        let table_name = "system_metrics";
392        let schema_name = "public".to_string();
393        let catalog_name = "greptime".to_string();
394        let mut options: TableOptions = Default::default();
395        let _ = options
396            .extra_options
397            .insert(FILE_TABLE_LOCATION_KEY.to_string(), "foo.csv".to_string());
398        let _ = options.extra_options.insert(
399            FILE_TABLE_META_KEY.to_string(),
400            "{{\"files\":[\"foo.csv\"]}}".to_string(),
401        );
402        let _ = options
403            .extra_options
404            .insert(FILE_TABLE_FORMAT_KEY.to_string(), "csv".to_string());
405        let meta = TableMetaBuilder::empty()
406            .schema(table_schema)
407            .primary_key_indices(vec![])
408            .engine("file".to_string())
409            .next_column_id(0)
410            .options(options)
411            .created_on(Default::default())
412            .build()
413            .unwrap();
414
415        let info = Arc::new(
416            TableInfoBuilder::default()
417                .table_id(1024)
418                .table_version(0 as TableVersion)
419                .name(table_name)
420                .schema_name(schema_name)
421                .catalog_name(catalog_name)
422                .desc(None)
423                .table_type(TableType::Base)
424                .meta(meta)
425                .build()
426                .unwrap(),
427        );
428
429        let stmt = create_table_stmt(&info, None, '"').unwrap();
430
431        let sql = format!("\n{}", stmt);
432        assert_eq!(
433            r#"
434CREATE EXTERNAL TABLE IF NOT EXISTS "system_metrics" (
435  "host" STRING NULL,
436  "cpu" DOUBLE NULL,
437
438)
439ENGINE=file
440WITH(
441  format = 'csv',
442  location = 'foo.csv'
443)"#,
444            sql
445        );
446    }
447}
query/sql/show_create_table.rs

query/sql/
show_create_table.rs
