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

table/
requests.rs
