1use std::collections::HashMap;
18
19use arrow_schema::extension::ExtensionType;
20use common_meta::SchemaOptions;
21use datatypes::extension::json::{Json2ExtensionType, parse_legacy_json2_settings};
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,
27 COLUMN_VECTOR_INDEX_OPT_KEY_CONNECTIVITY, COLUMN_VECTOR_INDEX_OPT_KEY_ENGINE,
28 COLUMN_VECTOR_INDEX_OPT_KEY_EXPANSION_ADD, COLUMN_VECTOR_INDEX_OPT_KEY_EXPANSION_SEARCH,
29 COLUMN_VECTOR_INDEX_OPT_KEY_METRIC, COMMENT_KEY, ColumnDefaultConstraint, ColumnSchema,
30 FulltextBackend, SchemaRef,
31};
32use datatypes::types::JsonFormat;
33use snafu::ResultExt;
34use sql::ast::{ColumnDef, ColumnOption, ColumnOptionDef, DataType, Expr, Ident, ObjectName};
35use sql::dialect::GreptimeDbDialect;
36use sql::parser::ParserContext;
37use sql::statements::create::{Column, ColumnExtensions, CreateTable, TableConstraint};
38use sql::statements::{self, OptionMap, concrete_data_type_to_sql_data_type};
39use store_api::metric_engine_consts::{is_metric_engine, is_metric_engine_internal_column};
40use table::metadata::{TableInfoRef, TableMeta};
41use table::requests::{
42 COMMENT_KEY as TABLE_COMMENT_KEY, FILE_TABLE_META_KEY, SKIP_WAL_KEY, TTL_KEY,
43 WRITE_BUFFER_SIZE_KEY,
44};
45
46use crate::error::{
47 ConvertSqlTypeSnafu, ConvertSqlValueSnafu, GetFulltextOptionsSnafu,
48 GetSkippingIndexOptionsSnafu, GetVectorIndexOptionsSnafu, Result, SqlSnafu,
49};
50
51fn create_sql_options(table_meta: &TableMeta, schema_options: Option<SchemaOptions>) -> OptionMap {
53 let table_opts = &table_meta.options;
54 let mut options = OptionMap::default();
55 if let Some(write_buffer_size) = table_opts.write_buffer_size {
56 options.insert(
57 WRITE_BUFFER_SIZE_KEY.to_string(),
58 write_buffer_size.to_string(),
59 );
60 }
61 if let Some(ttl) = table_opts.ttl.map(|t| t.to_string()) {
62 options.insert(TTL_KEY.to_string(), ttl);
63 } else if let Some(database_ttl) = schema_options
64 .as_ref()
65 .and_then(|o| o.ttl)
66 .map(|ttl| ttl.to_string())
67 {
68 options.insert(TTL_KEY.to_string(), database_ttl);
69 };
70 for (k, v) in table_opts
71 .extra_options
72 .iter()
73 .filter(|(k, _)| k != &FILE_TABLE_META_KEY)
74 {
75 options.insert(k.clone(), v.clone());
76 }
77 if table_opts.skip_wal {
78 options.insert(SKIP_WAL_KEY.to_string(), true.to_string());
79 }
80 options
81}
82
83#[inline]
84fn column_option_def(option: ColumnOption) -> ColumnOptionDef {
85 ColumnOptionDef { name: None, option }
86}
87
88fn create_column(column_schema: &ColumnSchema, quote_style: char) -> Result<Column> {
89 let name = &column_schema.name;
90 let mut options = Vec::with_capacity(2);
91 let mut extensions = ColumnExtensions::default();
92
93 if column_schema.is_nullable() {
94 options.push(column_option_def(ColumnOption::Null));
95 } else {
96 options.push(column_option_def(ColumnOption::NotNull));
97 }
98
99 if let Some(c) = column_schema.default_constraint() {
100 let expr = match c {
101 ColumnDefaultConstraint::Value(v) => Expr::Value(
102 statements::value_to_sql_value(v)
103 .with_context(|_| ConvertSqlValueSnafu { value: v.clone() })?
104 .into(),
105 ),
106 ColumnDefaultConstraint::Function(expr) => {
107 ParserContext::parse_function(expr, &GreptimeDbDialect {}).context(SqlSnafu)?
108 }
109 };
110
111 options.push(column_option_def(ColumnOption::Default(expr)));
112 }
113
114 if let Some(c) = column_schema.metadata().get(COMMENT_KEY) {
115 options.push(column_option_def(ColumnOption::Comment(c.clone())));
116 }
117
118 if let Some(opt) = column_schema
119 .fulltext_options()
120 .context(GetFulltextOptionsSnafu)?
121 && opt.enable
122 {
123 let mut map = HashMap::from([
124 (
125 COLUMN_FULLTEXT_OPT_KEY_ANALYZER.to_string(),
126 opt.analyzer.to_string(),
127 ),
128 (
129 COLUMN_FULLTEXT_OPT_KEY_CASE_SENSITIVE.to_string(),
130 opt.case_sensitive.to_string(),
131 ),
132 (
133 COLUMN_FULLTEXT_OPT_KEY_BACKEND.to_string(),
134 opt.backend.to_string(),
135 ),
136 ]);
137 if opt.backend == FulltextBackend::Bloom {
138 map.insert(
139 COLUMN_FULLTEXT_OPT_KEY_GRANULARITY.to_string(),
140 opt.granularity.to_string(),
141 );
142 map.insert(
143 COLUMN_FULLTEXT_OPT_KEY_FALSE_POSITIVE_RATE.to_string(),
144 opt.false_positive_rate().to_string(),
145 );
146 }
147 extensions.fulltext_index_options = Some(map.into());
148 }
149
150 if let Some(opt) = column_schema
151 .skipping_index_options()
152 .context(GetSkippingIndexOptionsSnafu)?
153 {
154 let map = HashMap::from([
155 (
156 COLUMN_SKIPPING_INDEX_OPT_KEY_GRANULARITY.to_string(),
157 opt.granularity.to_string(),
158 ),
159 (
160 COLUMN_SKIPPING_INDEX_OPT_KEY_FALSE_POSITIVE_RATE.to_string(),
161 opt.false_positive_rate().to_string(),
162 ),
163 (
164 COLUMN_SKIPPING_INDEX_OPT_KEY_TYPE.to_string(),
165 opt.index_type.to_string(),
166 ),
167 ]);
168 extensions.skipping_index_options = Some(map.into());
169 }
170
171 if let Some(opt) = column_schema
172 .vector_index_options()
173 .context(GetVectorIndexOptionsSnafu)?
174 {
175 let map = HashMap::from([
176 (
177 COLUMN_VECTOR_INDEX_OPT_KEY_ENGINE.to_string(),
178 opt.engine.to_string(),
179 ),
180 (
181 COLUMN_VECTOR_INDEX_OPT_KEY_METRIC.to_string(),
182 opt.metric.to_string(),
183 ),
184 (
185 COLUMN_VECTOR_INDEX_OPT_KEY_CONNECTIVITY.to_string(),
186 opt.connectivity.to_string(),
187 ),
188 (
189 COLUMN_VECTOR_INDEX_OPT_KEY_EXPANSION_ADD.to_string(),
190 opt.expansion_add.to_string(),
191 ),
192 (
193 COLUMN_VECTOR_INDEX_OPT_KEY_EXPANSION_SEARCH.to_string(),
194 opt.expansion_search.to_string(),
195 ),
196 ]);
197 extensions.vector_index_options = Some(map.into());
198 }
199
200 if column_schema.is_inverted_indexed() {
201 extensions.inverted_index_options = Some(HashMap::new().into());
202 }
203
204 let mut data_type = concrete_data_type_to_sql_data_type(&column_schema.data_type)
205 .with_context(|_| ConvertSqlTypeSnafu {
206 datatype: column_schema.data_type.clone(),
207 })?;
208
209 if matches!(
210 &column_schema.data_type,
211 datatypes::data_type::ConcreteDataType::Json(json_type)
212 if matches!(json_type.format, JsonFormat::Json2(_))
213 ) {
214 data_type = DataType::Custom(ObjectName::from(vec![Ident::new("JSON2")]), vec![]);
215 }
216
217 let settings = if let Some(extension) = column_schema.extension_type::<Json2ExtensionType>()? {
218 Some(extension.metadata().json_settings().clone())
219 } else {
220 parse_legacy_json2_settings(column_schema.metadata())?
221 };
222 if let Some(settings) = settings {
223 extensions.set_json_settings(settings).context(SqlSnafu)?;
224 }
225
226 Ok(Column {
227 column_def: ColumnDef {
228 name: Ident::with_quote(quote_style, name),
229 data_type,
230 options,
231 },
232 extensions,
233 })
234}
235
236fn primary_key_columns_for_show_create<'a>(
240 table_meta: &'a TableMeta,
241 engine: &str,
242) -> Vec<&'a String> {
243 let is_metric_engine = is_metric_engine(engine);
244 if is_metric_engine {
245 table_meta
246 .row_key_column_names()
247 .filter(|name| !is_metric_engine_internal_column(name))
248 .collect()
249 } else {
250 table_meta.row_key_column_names().collect()
251 }
252}
253
254fn create_table_constraints(
255 engine: &str,
256 schema: &SchemaRef,
257 table_meta: &TableMeta,
258 quote_style: char,
259) -> Vec<TableConstraint> {
260 let mut constraints = Vec::with_capacity(2);
261 if let Some(timestamp_column) = schema.timestamp_column() {
262 let column_name = ×tamp_column.name;
263 constraints.push(TableConstraint::TimeIndex {
264 column: Ident::with_quote(quote_style, column_name),
265 });
266 }
267 if !table_meta.primary_key_indices.is_empty() {
268 let columns = primary_key_columns_for_show_create(table_meta, engine)
269 .into_iter()
270 .map(|name| Ident::with_quote(quote_style, name))
271 .collect();
272 constraints.push(TableConstraint::PrimaryKey { columns });
273 }
274
275 constraints
276}
277
278pub fn create_table_stmt(
280 table_info: &TableInfoRef,
281 schema_options: Option<SchemaOptions>,
282 quote_style: char,
283) -> Result<CreateTable> {
284 let table_meta = &table_info.meta;
285 let table_name = &table_info.name;
286 let schema = &table_info.meta.schema;
287 let is_metric_engine = is_metric_engine(&table_meta.engine);
288 let columns = schema
289 .column_schemas()
290 .iter()
291 .filter_map(|c| {
292 if is_metric_engine && is_metric_engine_internal_column(&c.name) {
293 None
294 } else {
295 Some(create_column(c, quote_style))
296 }
297 })
298 .collect::<Result<Vec<_>>>()?;
299
300 let constraints = create_table_constraints(&table_meta.engine, schema, table_meta, quote_style);
301
302 let mut options = create_sql_options(table_meta, schema_options);
303 if let Some(comment) = &table_info.desc
304 && options.get(TABLE_COMMENT_KEY).is_none()
305 {
306 options.insert(format!("'{TABLE_COMMENT_KEY}'"), comment.clone());
307 }
308
309 Ok(CreateTable {
310 if_not_exists: true,
311 table_id: table_info.ident.table_id,
312 name: ObjectName::from(vec![Ident::with_quote(quote_style, table_name)]),
313 columns,
314 engine: table_meta.engine.clone(),
315 constraints,
316 options,
317 partitions: None,
318 })
319}
320
321#[cfg(test)]
322mod tests {
323 use std::sync::Arc;
324 use std::time::Duration;
325
326 use common_time::timestamp::TimeUnit;
327 use datatypes::extension::json::JsonExtensionType;
328 use datatypes::prelude::ConcreteDataType;
329 use datatypes::schema::{
330 FulltextOptions, Schema, SchemaRef, SkippingIndexOptions, VectorIndexOptions,
331 };
332 use table::metadata::*;
333 use table::requests::{
334 FILE_TABLE_FORMAT_KEY, FILE_TABLE_LOCATION_KEY, FILE_TABLE_META_KEY, TableOptions,
335 };
336
337 use super::*;
338
339 #[test]
340 fn test_show_create_table_sql() {
341 let schema = vec![
342 ColumnSchema::new("id", ConcreteDataType::uint32_datatype(), true)
343 .with_skipping_options(SkippingIndexOptions {
344 granularity: 4096,
345 ..Default::default()
346 })
347 .unwrap(),
348 ColumnSchema::new("host", ConcreteDataType::string_datatype(), true)
349 .with_inverted_index(true),
350 ColumnSchema::new("cpu", ConcreteDataType::float64_datatype(), true),
351 ColumnSchema::new("disk", ConcreteDataType::float32_datatype(), true),
352 ColumnSchema::new("msg", ConcreteDataType::string_datatype(), true)
353 .with_fulltext_options(FulltextOptions {
354 enable: true,
355 ..Default::default()
356 })
357 .unwrap(),
358 ColumnSchema::new("embedding", ConcreteDataType::vector_datatype(4), true)
359 .with_vector_index_options(&VectorIndexOptions::default())
360 .unwrap(),
361 ColumnSchema::new(
362 "ts",
363 ConcreteDataType::timestamp_datatype(TimeUnit::Millisecond),
364 false,
365 )
366 .with_default_constraint(Some(ColumnDefaultConstraint::Function(String::from(
367 "current_timestamp()",
368 ))))
369 .unwrap()
370 .with_time_index(true),
371 ];
372
373 let table_schema = SchemaRef::new(Schema::new(schema));
374 let table_name = "system_metrics";
375 let schema_name = "public".to_string();
376 let catalog_name = "greptime".to_string();
377
378 let mut options = table::requests::TableOptions {
379 ttl: Some(Duration::from_secs(30).into()),
380 skip_wal: true,
381 ..Default::default()
382 };
383
384 let _ = options
385 .extra_options
386 .insert("compaction.type".to_string(), "twcs".to_string());
387
388 let meta = TableMetaBuilder::empty()
389 .schema(table_schema)
390 .primary_key_indices(vec![0, 1])
391 .value_indices(vec![2, 3])
392 .engine("mito".to_string())
393 .next_column_id(0)
394 .options(options)
395 .created_on(Default::default())
396 .build()
397 .unwrap();
398
399 let info = Arc::new(
400 TableInfoBuilder::default()
401 .table_id(1024)
402 .table_version(0 as TableVersion)
403 .name(table_name)
404 .schema_name(schema_name)
405 .catalog_name(catalog_name)
406 .desc(None)
407 .table_type(TableType::Base)
408 .meta(meta)
409 .build()
410 .unwrap(),
411 );
412
413 let stmt = create_table_stmt(&info, None, '"').unwrap();
414
415 let sql = format!("\n{}", stmt);
416 assert_eq!(
417 r#"
418CREATE TABLE IF NOT EXISTS "system_metrics" (
419 "id" INT UNSIGNED NULL SKIPPING INDEX WITH(false_positive_rate = '0.01', granularity = '4096', type = 'BLOOM'),
420 "host" STRING NULL INVERTED INDEX,
421 "cpu" DOUBLE NULL,
422 "disk" FLOAT NULL,
423 "msg" STRING NULL FULLTEXT INDEX WITH(analyzer = 'English', backend = 'bloom', case_sensitive = 'false', false_positive_rate = '0.01', granularity = '10240'),
424 "embedding" VECTOR(4) NULL VECTOR INDEX WITH(connectivity = '16', engine = 'usearch', expansion_add = '128', expansion_search = '64', metric = 'l2sq'),
425 "ts" TIMESTAMP(3) NOT NULL DEFAULT current_timestamp(),
426 TIME INDEX ("ts"),
427 PRIMARY KEY ("id", "host")
428)
429ENGINE=mito
430WITH(
431 'compaction.type' = 'twcs',
432 skip_wal = 'true',
433 ttl = '30s'
434)"#,
435 sql
436 );
437
438 let mut table_meta = info.meta.clone();
439 table_meta.options.skip_wal = false;
440 table_meta
441 .options
442 .extra_options
443 .insert(SKIP_WAL_KEY.to_string(), false.to_string());
444 assert_eq!(
445 Some("false"),
446 create_sql_options(&table_meta, None).get(SKIP_WAL_KEY)
447 );
448
449 let mut schema_options = SchemaOptions::default();
450 schema_options
451 .extra_options
452 .insert(SKIP_WAL_KEY.to_string(), true.to_string());
453 assert_eq!(
454 Some("false"),
455 create_sql_options(&table_meta, Some(schema_options)).get(SKIP_WAL_KEY)
456 );
457 }
458
459 #[test]
460 fn test_show_create_legacy_json_with_json_extension() {
461 let mut json_column = ColumnSchema::new("j", ConcreteDataType::json_datatype(), true);
462 json_column.with_extension_type(&JsonExtensionType);
463
464 let table_schema = SchemaRef::new(Schema::new(vec![
465 json_column,
466 ColumnSchema::new(
467 "ts",
468 ConcreteDataType::timestamp_datatype(TimeUnit::Millisecond),
469 false,
470 )
471 .with_time_index(true),
472 ]));
473 let table_name = "legacy_json";
474 let meta = TableMetaBuilder::empty()
475 .schema(table_schema)
476 .primary_key_indices(vec![])
477 .value_indices(vec![0])
478 .engine("mito".to_string())
479 .next_column_id(0)
480 .options(Default::default())
481 .created_on(Default::default())
482 .build()
483 .unwrap();
484
485 let info = Arc::new(
486 TableInfoBuilder::default()
487 .table_id(1024)
488 .table_version(0 as TableVersion)
489 .name(table_name)
490 .schema_name("public")
491 .catalog_name("greptime")
492 .desc(None)
493 .table_type(TableType::Base)
494 .meta(meta)
495 .build()
496 .unwrap(),
497 );
498
499 let stmt = create_table_stmt(&info, None, '"').unwrap();
500 let sql = format!("\n{}", stmt);
501 assert_eq!(
502 r#"
503CREATE TABLE IF NOT EXISTS "legacy_json" (
504 "j" JSON NULL,
505 "ts" TIMESTAMP(3) NOT NULL,
506 TIME INDEX ("ts")
507)
508ENGINE=mito
509"#,
510 sql
511 );
512 }
513
514 #[test]
515 fn test_show_create_external_table_sql() {
516 let schema = vec![
517 ColumnSchema::new("host", ConcreteDataType::string_datatype(), true),
518 ColumnSchema::new("cpu", ConcreteDataType::float64_datatype(), true),
519 ];
520 let table_schema = SchemaRef::new(Schema::new(schema));
521 let table_name = "system_metrics";
522 let schema_name = "public".to_string();
523 let catalog_name = "greptime".to_string();
524 let mut options: TableOptions = Default::default();
525 let _ = options
526 .extra_options
527 .insert(FILE_TABLE_LOCATION_KEY.to_string(), "foo.csv".to_string());
528 let _ = options.extra_options.insert(
529 FILE_TABLE_META_KEY.to_string(),
530 "{{\"files\":[\"foo.csv\"]}}".to_string(),
531 );
532 let _ = options
533 .extra_options
534 .insert(FILE_TABLE_FORMAT_KEY.to_string(), "csv".to_string());
535 let meta = TableMetaBuilder::empty()
536 .schema(table_schema)
537 .primary_key_indices(vec![])
538 .engine("file".to_string())
539 .next_column_id(0)
540 .options(options)
541 .created_on(Default::default())
542 .build()
543 .unwrap();
544
545 let info = Arc::new(
546 TableInfoBuilder::default()
547 .table_id(1024)
548 .table_version(0 as TableVersion)
549 .name(table_name)
550 .schema_name(schema_name)
551 .catalog_name(catalog_name)
552 .desc(None)
553 .table_type(TableType::Base)
554 .meta(meta)
555 .build()
556 .unwrap(),
557 );
558
559 let stmt = create_table_stmt(&info, None, '"').unwrap();
560
561 let sql = format!("\n{}", stmt);
562 assert_eq!(
563 r#"
564CREATE EXTERNAL TABLE IF NOT EXISTS "system_metrics" (
565 "host" STRING NULL,
566 "cpu" DOUBLE NULL,
567
568)
569ENGINE=file
570WITH(
571 format = 'csv',
572 location = 'foo.csv'
573)"#,
574 sql
575 );
576 }
577}