table/
requests.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//! Table and TableEngine requests
16
17use std::collections::{HashMap, HashSet};
18use std::fmt;
19use std::str::FromStr;
20
21use common_base::readable_size::ReadableSize;
22use common_datasource::object_store::oss::is_supported_in_oss;
23use common_datasource::object_store::s3::is_supported_in_s3;
24use common_query::AddColumnLocation;
25use common_time::TimeToLive;
26use common_time::range::TimestampRange;
27use datatypes::data_type::ConcreteDataType;
28use datatypes::prelude::VectorRef;
29use datatypes::schema::{
30    ColumnDefaultConstraint, ColumnSchema, FulltextOptions, SkippingIndexOptions,
31};
32use greptime_proto::v1::region::compact_request;
33use once_cell::sync::Lazy;
34use serde::{Deserialize, Serialize};
35use store_api::metric_engine_consts::{
36    LOGICAL_TABLE_METADATA_KEY, PHYSICAL_TABLE_METADATA_KEY, is_metric_engine_option_key,
37};
38use store_api::mito_engine_options::{
39    APPEND_MODE_KEY, COMPACTION_TYPE, MEMTABLE_BULK_ENCODE_BYTES_THRESHOLD,
40    MEMTABLE_BULK_ENCODE_ROW_THRESHOLD, MEMTABLE_BULK_MAX_MERGE_GROUPS,
41    MEMTABLE_BULK_MERGE_THRESHOLD, MEMTABLE_TYPE, MERGE_MODE_KEY, SST_FORMAT_KEY,
42    TWCS_FALLBACK_TO_LOCAL, TWCS_MAX_OUTPUT_FILE_SIZE, TWCS_TIME_WINDOW, TWCS_TRIGGER_FILE_NUM,
43    is_mito_engine_option_key,
44};
45use store_api::region_request::{SetRegionOption, UnsetRegionOption};
46
47use crate::error::{ParseTableOptionSnafu, Result};
48use crate::metadata::{TableId, TableVersion};
49use crate::table_reference::TableReference;
50
51mod semantic;
52pub use semantic::*;
53
54pub const FILE_TABLE_META_KEY: &str = "__private.file_table_meta";
55pub const FILE_TABLE_LOCATION_KEY: &str = "location";
56pub const FILE_TABLE_PATTERN_KEY: &str = "pattern";
57pub const FILE_TABLE_FORMAT_KEY: &str = "format";
58
59pub const TABLE_DATA_MODEL: &str = "table_data_model";
60pub const TABLE_DATA_MODEL_TRACE_V1: &str = "greptime_trace_v1";
61
62pub const OTLP_METRIC_COMPAT_KEY: &str = "otlp_metric_compat";
63pub const OTLP_METRIC_COMPAT_PROM: &str = "prom";
64
65pub const VALID_TABLE_OPTION_KEYS: [&str; 14] = [
66    // common keys:
67    WRITE_BUFFER_SIZE_KEY,
68    TTL_KEY,
69    STORAGE_KEY,
70    COMMENT_KEY,
71    SKIP_WAL_KEY,
72    SST_FORMAT_KEY,
73    // file engine keys:
74    FILE_TABLE_LOCATION_KEY,
75    FILE_TABLE_FORMAT_KEY,
76    FILE_TABLE_PATTERN_KEY,
77    // metric engine keys:
78    PHYSICAL_TABLE_METADATA_KEY,
79    LOGICAL_TABLE_METADATA_KEY,
80    // table model info
81    TABLE_DATA_MODEL,
82    OTLP_METRIC_COMPAT_KEY,
83    REPARTITION_COLUMN_HINT_KEY,
84];
85
86pub const DDL_TIMEOUT: &str = "timeout";
87pub const DDL_WAIT: &str = "wait";
88
89pub const VALID_DDL_OPTION_KEYS: [&str; 2] = [DDL_TIMEOUT, DDL_WAIT];
90
91// Valid option keys when creating a db.
92static VALID_DB_OPT_KEYS: Lazy<HashSet<&str>> = Lazy::new(|| {
93    let mut set = HashSet::new();
94    set.insert(TTL_KEY);
95    set.insert(STORAGE_KEY);
96    set.insert(MEMTABLE_TYPE);
97    set.insert(MEMTABLE_BULK_MERGE_THRESHOLD);
98    set.insert(MEMTABLE_BULK_ENCODE_ROW_THRESHOLD);
99    set.insert(MEMTABLE_BULK_ENCODE_BYTES_THRESHOLD);
100    set.insert(MEMTABLE_BULK_MAX_MERGE_GROUPS);
101    set.insert(APPEND_MODE_KEY);
102    set.insert(MERGE_MODE_KEY);
103    set.insert(SKIP_WAL_KEY);
104    set.insert(COMPACTION_TYPE);
105    set.insert(TWCS_FALLBACK_TO_LOCAL);
106    set.insert(TWCS_TIME_WINDOW);
107    set.insert(TWCS_TRIGGER_FILE_NUM);
108    set.insert(TWCS_MAX_OUTPUT_FILE_SIZE);
109    set.insert(SST_FORMAT_KEY);
110    set
111});
112
113/// Returns true if the `key` is a valid key for database.
114pub fn validate_database_option(key: &str) -> bool {
115    VALID_DB_OPT_KEYS.contains(&key)
116}
117
118/// Returns true if the `key` is a valid key for any engine or storage.
119pub fn validate_table_option(key: &str) -> bool {
120    if is_supported_in_s3(key) {
121        return true;
122    }
123
124    if is_supported_in_oss(key) {
125        return true;
126    }
127
128    if is_mito_engine_option_key(key) {
129        return true;
130    }
131
132    if is_metric_engine_option_key(key) {
133        return true;
134    }
135
136    // Semantic-layer keys share a reserved prefix instead of a fixed allowlist so
137    // the vocabulary can grow without touching this gate. See `semantic` module.
138    if is_semantic_option_key(key) {
139        return true;
140    }
141
142    VALID_TABLE_OPTION_KEYS.contains(&key) || VALID_DDL_OPTION_KEYS.contains(&key)
143}
144
145#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq)]
146#[serde(default)]
147pub struct TableOptions {
148    /// Per-region write buffer stall threshold. Writes are rejected at twice this size.
149    pub write_buffer_size: Option<ReadableSize>,
150    /// Time-to-live of table. Expired data will be automatically purged.
151    pub ttl: Option<TimeToLive>,
152    /// Skip wal write for this table.
153    pub skip_wal: bool,
154    /// Extra options that may not applicable to all table engines.
155    pub extra_options: HashMap<String, String>,
156}
157
158pub const WRITE_BUFFER_SIZE_KEY: &str = store_api::mito_engine_options::WRITE_BUFFER_SIZE_KEY;
159pub const TTL_KEY: &str = store_api::mito_engine_options::TTL_KEY;
160pub const STORAGE_KEY: &str = "storage";
161pub const COMMENT_KEY: &str = "comment";
162pub const AUTO_CREATE_TABLE_KEY: &str = "auto_create_table";
163pub const SKIP_WAL_KEY: &str = store_api::mito_engine_options::SKIP_WAL_KEY;
164pub const TRACE_TABLE_PARTITIONS_HINT_KEY: &str = "trace_table_partitions";
165pub const REPARTITION_COLUMN_HINT_KEY: &str = "repartition.column.hint";
166
167impl TableOptions {
168    pub fn try_from_iter<T: ToString, U: IntoIterator<Item = (T, T)>>(
169        iter: U,
170    ) -> Result<TableOptions> {
171        let mut options = TableOptions::default();
172
173        let kvs: HashMap<String, String> = iter
174            .into_iter()
175            .map(|(k, v)| (k.to_string(), v.to_string()))
176            .collect();
177
178        if let Some(write_buffer_size) = kvs.get(WRITE_BUFFER_SIZE_KEY) {
179            let size = ReadableSize::from_str(write_buffer_size).map_err(|_| {
180                ParseTableOptionSnafu {
181                    key: WRITE_BUFFER_SIZE_KEY,
182                    value: write_buffer_size,
183                }
184                .build()
185            })?;
186            options.write_buffer_size = Some(size)
187        }
188
189        if let Some(ttl) = kvs.get(TTL_KEY) {
190            let ttl_value = TimeToLive::from_humantime_or_str(ttl).map_err(|_| {
191                ParseTableOptionSnafu {
192                    key: TTL_KEY,
193                    value: ttl,
194                }
195                .build()
196            })?;
197            options.ttl = Some(ttl_value);
198        }
199
200        if let Some(skip_wal) = kvs.get(SKIP_WAL_KEY) {
201            options.skip_wal = skip_wal.parse().map_err(|_| {
202                ParseTableOptionSnafu {
203                    key: SKIP_WAL_KEY,
204                    value: skip_wal,
205                }
206                .build()
207            })?;
208        }
209
210        options.extra_options = HashMap::from_iter(
211            kvs.into_iter()
212                .filter(|(k, _)| k != WRITE_BUFFER_SIZE_KEY && k != TTL_KEY),
213        );
214
215        Ok(options)
216    }
217}
218
219impl fmt::Display for TableOptions {
220    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
221        let mut key_vals = vec![];
222        if let Some(size) = self.write_buffer_size {
223            key_vals.push(format!("{}={}", WRITE_BUFFER_SIZE_KEY, size));
224        }
225
226        if let Some(ttl) = self.ttl.map(|ttl| ttl.to_string()) {
227            key_vals.push(format!("{}={}", TTL_KEY, ttl));
228        }
229
230        if self.skip_wal && !self.extra_options.contains_key(SKIP_WAL_KEY) {
231            key_vals.push(format!("{}={}", SKIP_WAL_KEY, self.skip_wal));
232        }
233
234        for (k, v) in &self.extra_options {
235            key_vals.push(format!("{}={}", k, v));
236        }
237
238        write!(f, "{}", key_vals.join(" "))
239    }
240}
241
242impl From<&TableOptions> for HashMap<String, String> {
243    fn from(opts: &TableOptions) -> Self {
244        let mut res = HashMap::with_capacity(3 + opts.extra_options.len());
245        if let Some(write_buffer_size) = opts.write_buffer_size {
246            let _ = res.insert(
247                WRITE_BUFFER_SIZE_KEY.to_string(),
248                write_buffer_size.to_string(),
249            );
250        }
251        if let Some(ttl_str) = opts.ttl.map(|ttl| ttl.to_string()) {
252            let _ = res.insert(TTL_KEY.to_string(), ttl_str);
253        }
254        if opts.skip_wal {
255            let _ = res.insert(SKIP_WAL_KEY.to_string(), true.to_string());
256        }
257        res.extend(opts.extra_options.clone());
258        res
259    }
260}
261
262/// Alter table request
263#[derive(Debug, Clone, Serialize, Deserialize)]
264pub struct AlterTableRequest {
265    pub catalog_name: String,
266    pub schema_name: String,
267    pub table_name: String,
268    pub table_id: TableId,
269    pub alter_kind: AlterKind,
270    // None in standalone.
271    pub table_version: Option<TableVersion>,
272}
273
274/// Add column request
275#[derive(Debug, Clone, Serialize, Deserialize)]
276pub struct AddColumnRequest {
277    pub column_schema: ColumnSchema,
278    pub is_key: bool,
279    pub location: Option<AddColumnLocation>,
280    /// Add column if not exists.
281    pub add_if_not_exists: bool,
282}
283
284/// Change column datatype request
285#[derive(Debug, Clone, Serialize, Deserialize)]
286pub struct ModifyColumnTypeRequest {
287    pub column_name: String,
288    pub target_type: ConcreteDataType,
289}
290
291#[derive(Debug, Clone, Serialize, Deserialize)]
292pub enum AlterKind {
293    AddColumns {
294        columns: Vec<AddColumnRequest>,
295    },
296    DropColumns {
297        names: Vec<String>,
298    },
299    ModifyColumnTypes {
300        columns: Vec<ModifyColumnTypeRequest>,
301    },
302    RenameTable {
303        new_table_name: String,
304    },
305    SetTableOptions {
306        options: Vec<SetRegionOption>,
307    },
308    UnsetTableOptions {
309        keys: Vec<UnsetRegionOption>,
310    },
311    SetRepartitionColumnHint {
312        column_name: String,
313    },
314    UnsetRepartitionColumnHint,
315    SetIndexes {
316        options: Vec<SetIndexOption>,
317    },
318    UnsetIndexes {
319        options: Vec<UnsetIndexOption>,
320    },
321    DropDefaults {
322        names: Vec<String>,
323    },
324    SetDefaults {
325        defaults: Vec<SetDefaultRequest>,
326    },
327}
328
329#[derive(Debug, Clone, Serialize, Deserialize)]
330pub struct SetDefaultRequest {
331    pub column_name: String,
332    pub default_constraint: Option<ColumnDefaultConstraint>,
333}
334
335#[derive(Debug, Clone, Serialize, Deserialize)]
336pub enum SetIndexOption {
337    Fulltext {
338        column_name: String,
339        options: FulltextOptions,
340    },
341    Inverted {
342        column_name: String,
343    },
344    Skipping {
345        column_name: String,
346        options: SkippingIndexOptions,
347    },
348}
349
350impl SetIndexOption {
351    /// Returns the column name of the index option.
352    pub fn column_name(&self) -> &str {
353        match self {
354            SetIndexOption::Fulltext { column_name, .. } => column_name,
355            SetIndexOption::Inverted { column_name, .. } => column_name,
356            SetIndexOption::Skipping { column_name, .. } => column_name,
357        }
358    }
359}
360
361#[derive(Debug, Clone, Serialize, Deserialize)]
362pub enum UnsetIndexOption {
363    Fulltext { column_name: String },
364    Inverted { column_name: String },
365    Skipping { column_name: String },
366}
367
368impl UnsetIndexOption {
369    /// Returns the column name of the index option.
370    pub fn column_name(&self) -> &str {
371        match self {
372            UnsetIndexOption::Fulltext { column_name, .. } => column_name,
373            UnsetIndexOption::Inverted { column_name, .. } => column_name,
374            UnsetIndexOption::Skipping { column_name, .. } => column_name,
375        }
376    }
377}
378
379#[derive(Debug)]
380pub struct InsertRequest {
381    pub catalog_name: String,
382    pub schema_name: String,
383    pub table_name: String,
384    pub columns_values: HashMap<String, VectorRef>,
385}
386
387/// Delete (by primary key) request
388#[derive(Debug)]
389pub struct DeleteRequest {
390    pub catalog_name: String,
391    pub schema_name: String,
392    pub table_name: String,
393    /// Values of each column in this table's primary key and time index.
394    ///
395    /// The key is the column name, and the value is the column value.
396    pub key_column_values: HashMap<String, VectorRef>,
397}
398
399#[derive(Debug)]
400pub enum CopyDirection {
401    Export,
402    Import,
403}
404
405/// Copy table request
406#[derive(Debug)]
407pub struct CopyTableRequest {
408    pub catalog_name: String,
409    pub schema_name: String,
410    pub table_name: String,
411    pub location: String,
412    pub with: HashMap<String, String>,
413    pub connection: HashMap<String, String>,
414    pub pattern: Option<String>,
415    pub direction: CopyDirection,
416    pub timestamp_range: Option<TimestampRange>,
417    pub limit: Option<u64>,
418}
419
420#[derive(Debug, Clone, Default)]
421pub struct FlushTableRequest {
422    pub catalog_name: String,
423    pub schema_name: String,
424    pub table_name: String,
425}
426
427#[derive(Debug, Clone, Default)]
428pub struct BuildIndexTableRequest {
429    pub catalog_name: String,
430    pub schema_name: String,
431    pub table_name: String,
432}
433
434#[derive(Debug, Clone, PartialEq)]
435pub struct CompactTableRequest {
436    pub catalog_name: String,
437    pub schema_name: String,
438    pub table_name: String,
439    pub compact_options: compact_request::Options,
440    pub parallelism: u32,
441    pub time_range: Option<TimestampRange>,
442}
443
444impl Default for CompactTableRequest {
445    fn default() -> Self {
446        Self {
447            catalog_name: Default::default(),
448            schema_name: Default::default(),
449            table_name: Default::default(),
450            compact_options: compact_request::Options::Regular(Default::default()),
451            parallelism: 1,
452            time_range: None,
453        }
454    }
455}
456
457/// Truncate table request
458#[derive(Debug, Clone, Serialize, Deserialize)]
459pub struct TruncateTableRequest {
460    pub catalog_name: String,
461    pub schema_name: String,
462    pub table_name: String,
463    pub table_id: TableId,
464}
465
466impl TruncateTableRequest {
467    pub fn table_ref(&self) -> TableReference<'_> {
468        TableReference {
469            catalog: &self.catalog_name,
470            schema: &self.schema_name,
471            table: &self.table_name,
472        }
473    }
474}
475
476#[derive(Debug, Clone, Default, Deserialize, Serialize)]
477pub struct CopyDatabaseRequest {
478    pub catalog_name: String,
479    pub schema_name: String,
480    pub location: String,
481    pub with: HashMap<String, String>,
482    pub connection: HashMap<String, String>,
483    pub time_range: Option<TimestampRange>,
484}
485
486#[derive(Debug, Clone, Default, Deserialize, Serialize)]
487pub struct CopyQueryToRequest {
488    pub location: String,
489    pub with: HashMap<String, String>,
490    pub connection: HashMap<String, String>,
491}
492
493#[cfg(test)]
494mod tests {
495    use std::time::Duration;
496
497    use super::*;
498
499    #[test]
500    fn test_validate_table_option() {
501        assert!(validate_table_option(FILE_TABLE_LOCATION_KEY));
502        assert!(validate_table_option(FILE_TABLE_FORMAT_KEY));
503        assert!(validate_table_option(FILE_TABLE_PATTERN_KEY));
504        assert!(validate_table_option(TTL_KEY));
505        assert!(validate_table_option(WRITE_BUFFER_SIZE_KEY));
506        assert!(validate_table_option(STORAGE_KEY));
507        assert!(validate_table_option(MEMTABLE_BULK_MERGE_THRESHOLD));
508        assert!(validate_table_option(REPARTITION_COLUMN_HINT_KEY));
509        assert!(!validate_table_option("foo"));
510
511        // Only whitelisted semantic keys are accepted.
512        assert!(validate_table_option(SEMANTIC_SIGNAL_TYPE));
513        assert!(validate_table_option(SEMANTIC_METRIC_TYPE));
514        // Unknown semantic key, near-miss, and the internal transport key are rejected.
515        assert!(!validate_table_option("greptime.semantic.future.key"));
516        assert!(!validate_table_option("greptime.semanticx"));
517        assert!(!validate_table_option(SEMANTIC_PER_TABLE_INDEX_KEY));
518    }
519
520    #[test]
521    fn test_validate_database_option() {
522        assert!(validate_database_option(MEMTABLE_TYPE));
523        assert!(validate_database_option(MEMTABLE_BULK_MERGE_THRESHOLD));
524        assert!(validate_database_option(MEMTABLE_BULK_ENCODE_ROW_THRESHOLD));
525        assert!(validate_database_option(
526            MEMTABLE_BULK_ENCODE_BYTES_THRESHOLD
527        ));
528        assert!(validate_database_option(MEMTABLE_BULK_MAX_MERGE_GROUPS));
529        assert!(!validate_database_option("foo"));
530    }
531
532    #[test]
533    fn test_serialize_table_options() {
534        let options = TableOptions {
535            write_buffer_size: None,
536            ttl: Some(Duration::from_secs(1000).into()),
537            extra_options: HashMap::new(),
538            skip_wal: false,
539        };
540        let serialized = serde_json::to_string(&options).unwrap();
541        let deserialized: TableOptions = serde_json::from_str(&serialized).unwrap();
542        assert_eq!(options, deserialized);
543    }
544
545    #[test]
546    fn test_convert_hashmap_between_table_options() {
547        let options = TableOptions {
548            write_buffer_size: Some(ReadableSize::mb(128)),
549            ttl: Some(Duration::from_secs(1000).into()),
550            extra_options: HashMap::new(),
551            skip_wal: false,
552        };
553        let serialized_map = HashMap::from(&options);
554        let serialized = TableOptions::try_from_iter(&serialized_map).unwrap();
555        assert_eq!(options, serialized);
556
557        let options = TableOptions {
558            write_buffer_size: None,
559            ttl: None,
560            extra_options: HashMap::from([(SKIP_WAL_KEY.to_string(), true.to_string())]),
561            skip_wal: true,
562        };
563        let serialized_map = HashMap::from(&options);
564        assert_eq!(
565            Some("true"),
566            serialized_map.get(SKIP_WAL_KEY).map(String::as_str)
567        );
568        let serialized = TableOptions::try_from_iter(&serialized_map).unwrap();
569        assert_eq!(options, serialized);
570
571        let options = TableOptions {
572            write_buffer_size: None,
573            ttl: Default::default(),
574            extra_options: HashMap::new(),
575            skip_wal: false,
576        };
577        let serialized_map = HashMap::from(&options);
578        let serialized = TableOptions::try_from_iter(&serialized_map).unwrap();
579        assert_eq!(options, serialized);
580
581        let options = TableOptions {
582            write_buffer_size: Some(ReadableSize::mb(128)),
583            ttl: Some(Duration::from_secs(1000).into()),
584            extra_options: HashMap::from([("a".to_string(), "A".to_string())]),
585            skip_wal: false,
586        };
587        let serialized_map = HashMap::from(&options);
588        let serialized = TableOptions::try_from_iter(&serialized_map).unwrap();
589        assert_eq!(options, serialized);
590
591        let options = TableOptions {
592            extra_options: HashMap::from([(SKIP_WAL_KEY.to_string(), false.to_string())]),
593            skip_wal: false,
594            ..Default::default()
595        };
596        let serialized_map = HashMap::from(&options);
597        let serialized = TableOptions::try_from_iter(&serialized_map).unwrap();
598        assert_eq!(options, serialized);
599    }
600
601    #[test]
602    fn test_table_options_to_string() {
603        let options = TableOptions {
604            write_buffer_size: Some(ReadableSize::mb(128)),
605            ttl: Some(Duration::from_secs(1000).into()),
606            extra_options: HashMap::new(),
607            skip_wal: false,
608        };
609
610        assert_eq!(
611            "write_buffer_size=128.0MiB ttl=16m 40s",
612            options.to_string()
613        );
614
615        let options = TableOptions {
616            write_buffer_size: Some(ReadableSize::mb(128)),
617            ttl: Some(Duration::from_secs(1000).into()),
618            extra_options: HashMap::from([("a".to_string(), "A".to_string())]),
619            skip_wal: false,
620        };
621
622        assert_eq!(
623            "write_buffer_size=128.0MiB ttl=16m 40s a=A",
624            options.to_string()
625        );
626
627        let options = TableOptions {
628            write_buffer_size: Some(ReadableSize::mb(128)),
629            ttl: Some(Duration::from_secs(1000).into()),
630            extra_options: HashMap::new(),
631            skip_wal: true,
632        };
633        assert_eq!(
634            "write_buffer_size=128.0MiB ttl=16m 40s skip_wal=true",
635            options.to_string()
636        );
637
638        let options = TableOptions {
639            write_buffer_size: None,
640            ttl: None,
641            extra_options: HashMap::from([(SKIP_WAL_KEY.to_string(), "false".to_string())]),
642            skip_wal: false,
643        };
644        assert_eq!("skip_wal=false", options.to_string());
645    }
646}
table/requests.rs

table/
requests.rs
