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, Schema, 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_ACTIVE_WINDOW_L1_MERGE_TRIGGER, TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM,
43    TWCS_FALLBACK_TO_LOCAL, TWCS_INACTIVE_WINDOW_L1_MERGE_TRIGGER,
44    TWCS_INACTIVE_WINDOW_TRIGGER_FILE_NUM, TWCS_MAX_OUTPUT_FILE_SIZE, TWCS_TIME_WINDOW,
45    TWCS_TRIGGER_FILE_NUM, is_mito_engine_option_key, normalize_twcs_trigger_options,
46};
47use store_api::region_request::{SetRegionOption, UnsetRegionOption};
48
49use crate::error::{ConflictingTableOptionsSnafu, ParseTableOptionSnafu, Result};
50use crate::metadata::{TableId, TableVersion};
51use crate::table_reference::TableReference;
52
53mod semantic;
54pub use semantic::*;
55
56pub const FILE_TABLE_META_KEY: &str = "__private.file_table_meta";
57pub const FILE_TABLE_LOCATION_KEY: &str = "location";
58pub const FILE_TABLE_PATTERN_KEY: &str = "pattern";
59pub const FILE_TABLE_FORMAT_KEY: &str = "format";
60
61pub const TABLE_DATA_MODEL: &str = "table_data_model";
62pub const TABLE_DATA_MODEL_TRACE_V1: &str = "greptime_trace_v1";
63
64/// Returns true if the table stores spans in the `greptime_trace_v1` data model
65/// (fixed span columns), the shape the Jaeger query path and the entity-graph
66/// derivation rely on.
67pub fn is_trace_v1_table(table_info: &crate::metadata::TableInfo) -> bool {
68    table_info
69        .meta
70        .options
71        .extra_options
72        .get(TABLE_DATA_MODEL)
73        .map(|v| v == TABLE_DATA_MODEL_TRACE_V1)
74        .unwrap_or(false)
75}
76
77pub const OTLP_METRIC_COMPAT_KEY: &str = "otlp_metric_compat";
78pub const OTLP_METRIC_COMPAT_PROM: &str = "prom";
79
80pub const VALID_TABLE_OPTION_KEYS: [&str; 15] = [
81    // common keys:
82    WRITE_BUFFER_SIZE_KEY,
83    TTL_KEY,
84    STORAGE_KEY,
85    COMMENT_KEY,
86    SKIP_WAL_KEY,
87    SST_FORMAT_KEY,
88    // file engine keys:
89    FILE_TABLE_LOCATION_KEY,
90    FILE_TABLE_FORMAT_KEY,
91    FILE_TABLE_PATTERN_KEY,
92    // metric engine keys:
93    PHYSICAL_TABLE_METADATA_KEY,
94    LOGICAL_TABLE_METADATA_KEY,
95    // table model info
96    TABLE_DATA_MODEL,
97    OTLP_METRIC_COMPAT_KEY,
98    REPARTITION_COLUMN_HINT_KEY,
99    REPARTITION_PARTITION_NUM_HINT_KEY,
100];
101
102pub const DDL_TIMEOUT: &str = "timeout";
103pub const DDL_WAIT: &str = "wait";
104
105pub const VALID_DDL_OPTION_KEYS: [&str; 2] = [DDL_TIMEOUT, DDL_WAIT];
106
107// Valid option keys when creating a db.
108static VALID_DB_OPT_KEYS: Lazy<HashSet<&str>> = Lazy::new(|| {
109    let mut set = HashSet::new();
110    set.insert(TTL_KEY);
111    set.insert(STORAGE_KEY);
112    set.insert(MEMTABLE_TYPE);
113    set.insert(MEMTABLE_BULK_MERGE_THRESHOLD);
114    set.insert(MEMTABLE_BULK_ENCODE_ROW_THRESHOLD);
115    set.insert(MEMTABLE_BULK_ENCODE_BYTES_THRESHOLD);
116    set.insert(MEMTABLE_BULK_MAX_MERGE_GROUPS);
117    set.insert(APPEND_MODE_KEY);
118    set.insert(MERGE_MODE_KEY);
119    set.insert(SKIP_WAL_KEY);
120    set.insert(COMPACTION_TYPE);
121    set.insert(TWCS_FALLBACK_TO_LOCAL);
122    set.insert(TWCS_TIME_WINDOW);
123    set.insert(TWCS_TRIGGER_FILE_NUM);
124    set.insert(TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM);
125    set.insert(TWCS_ACTIVE_WINDOW_L1_MERGE_TRIGGER);
126    set.insert(TWCS_INACTIVE_WINDOW_TRIGGER_FILE_NUM);
127    set.insert(TWCS_INACTIVE_WINDOW_L1_MERGE_TRIGGER);
128    set.insert(TWCS_MAX_OUTPUT_FILE_SIZE);
129    set.insert(SST_FORMAT_KEY);
130    set
131});
132
133/// Returns true if the `key` is a valid key for database.
134pub fn validate_database_option(key: &str) -> bool {
135    VALID_DB_OPT_KEYS.contains(&key)
136}
137
138/// Validates a database option value, returning the violated constraint on error.
139pub fn validate_database_option_value(
140    key: &str,
141    value: Option<&str>,
142) -> std::result::Result<(), &'static str> {
143    let (minimum, constraint) = match key {
144        TWCS_TRIGGER_FILE_NUM
145        | TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM
146        | TWCS_INACTIVE_WINDOW_TRIGGER_FILE_NUM => {
147            (0, "expected a non-negative integer fitting in usize")
148        }
149        TWCS_ACTIVE_WINDOW_L1_MERGE_TRIGGER | TWCS_INACTIVE_WINDOW_L1_MERGE_TRIGGER => {
150            (2, "expected an integer greater than or equal to 2")
151        }
152        _ => return Ok(()),
153    };
154    if value
155        .and_then(|value| value.parse::<usize>().ok())
156        .is_some_and(|files| files >= minimum)
157    {
158        Ok(())
159    } else {
160        Err(constraint)
161    }
162}
163
164/// Returns true if the `key` is a valid key for any engine or storage.
165pub fn validate_table_option(key: &str) -> bool {
166    if is_supported_in_s3(key) {
167        return true;
168    }
169
170    if is_supported_in_oss(key) {
171        return true;
172    }
173
174    if is_mito_engine_option_key(key) {
175        return true;
176    }
177
178    if is_metric_engine_option_key(key) {
179        return true;
180    }
181
182    // Semantic-layer keys share a reserved prefix instead of a fixed allowlist so
183    // the vocabulary can grow without touching this gate. See `semantic` module.
184    if is_semantic_option_key(key) {
185        return true;
186    }
187
188    VALID_TABLE_OPTION_KEYS.contains(&key) || VALID_DDL_OPTION_KEYS.contains(&key)
189}
190
191#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq)]
192#[serde(default)]
193pub struct TableOptions {
194    /// Per-region write buffer stall threshold. Writes are rejected at twice this size.
195    pub write_buffer_size: Option<ReadableSize>,
196    /// Time-to-live of table. Expired data will be automatically purged.
197    pub ttl: Option<TimeToLive>,
198    /// Skip wal write for this table.
199    pub skip_wal: bool,
200    /// Extra options that may not applicable to all table engines.
201    pub extra_options: HashMap<String, String>,
202}
203
204pub const WRITE_BUFFER_SIZE_KEY: &str = store_api::mito_engine_options::WRITE_BUFFER_SIZE_KEY;
205pub const TTL_KEY: &str = store_api::mito_engine_options::TTL_KEY;
206pub const STORAGE_KEY: &str = "storage";
207pub const COMMENT_KEY: &str = "comment";
208pub const AUTO_CREATE_TABLE_KEY: &str = "auto_create_table";
209pub const SKIP_WAL_KEY: &str = store_api::mito_engine_options::SKIP_WAL_KEY;
210pub const TRACE_TABLE_PARTITIONS_HINT_KEY: &str = "trace_table_partitions";
211pub const REPARTITION_COLUMN_HINT_KEY: &str = "repartition.column.hint";
212
213/// Table-level partition count hint consumed by the auto-repartition planner.
214pub const REPARTITION_PARTITION_NUM_HINT_KEY: &str = "repartition.partition.num.hint";
215
216impl TableOptions {
217    pub fn try_from_iter<T: ToString, U: IntoIterator<Item = (T, T)>>(
218        iter: U,
219    ) -> Result<TableOptions> {
220        let mut options = TableOptions::default();
221
222        let mut kvs: HashMap<String, String> = iter
223            .into_iter()
224            .map(|(k, v)| (k.to_string(), v.to_string()))
225            .collect();
226
227        normalize_twcs_trigger_options(&mut kvs).map_err(|conflict| {
228            ConflictingTableOptionsSnafu {
229                first_key: TWCS_TRIGGER_FILE_NUM,
230                first_value: conflict.legacy_value,
231                second_key: TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM,
232                second_value: conflict.canonical_value,
233            }
234            .build()
235        })?;
236
237        if let Some(write_buffer_size) = kvs.get(WRITE_BUFFER_SIZE_KEY) {
238            let size = ReadableSize::from_str(write_buffer_size).map_err(|_| {
239                ParseTableOptionSnafu {
240                    key: WRITE_BUFFER_SIZE_KEY,
241                    value: write_buffer_size,
242                }
243                .build()
244            })?;
245            options.write_buffer_size = Some(size)
246        }
247
248        if let Some(ttl) = kvs.get(TTL_KEY) {
249            let ttl_value = TimeToLive::from_humantime_or_str(ttl).map_err(|_| {
250                ParseTableOptionSnafu {
251                    key: TTL_KEY,
252                    value: ttl,
253                }
254                .build()
255            })?;
256            options.ttl = Some(ttl_value);
257        }
258
259        if let Some(skip_wal) = kvs.get(SKIP_WAL_KEY) {
260            options.skip_wal = skip_wal.parse().map_err(|_| {
261                ParseTableOptionSnafu {
262                    key: SKIP_WAL_KEY,
263                    value: skip_wal,
264                }
265                .build()
266            })?;
267        }
268
269        options.extra_options = HashMap::from_iter(
270            kvs.into_iter()
271                .filter(|(k, _)| k != WRITE_BUFFER_SIZE_KEY && k != TTL_KEY),
272        );
273
274        Ok(options)
275    }
276}
277
278impl fmt::Display for TableOptions {
279    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
280        let mut key_vals = vec![];
281        if let Some(size) = self.write_buffer_size {
282            key_vals.push(format!("{}={}", WRITE_BUFFER_SIZE_KEY, size));
283        }
284
285        if let Some(ttl) = self.ttl.map(|ttl| ttl.to_string()) {
286            key_vals.push(format!("{}={}", TTL_KEY, ttl));
287        }
288
289        if self.skip_wal && !self.extra_options.contains_key(SKIP_WAL_KEY) {
290            key_vals.push(format!("{}={}", SKIP_WAL_KEY, self.skip_wal));
291        }
292
293        for (k, v) in &self.extra_options {
294            key_vals.push(format!("{}={}", k, v));
295        }
296
297        write!(f, "{}", key_vals.join(" "))
298    }
299}
300
301impl From<&TableOptions> for HashMap<String, String> {
302    fn from(opts: &TableOptions) -> Self {
303        let mut res = HashMap::with_capacity(3 + opts.extra_options.len());
304        if let Some(write_buffer_size) = opts.write_buffer_size {
305            let _ = res.insert(
306                WRITE_BUFFER_SIZE_KEY.to_string(),
307                write_buffer_size.to_string(),
308            );
309        }
310        if let Some(ttl_str) = opts.ttl.map(|ttl| ttl.to_string()) {
311            let _ = res.insert(TTL_KEY.to_string(), ttl_str);
312        }
313        if opts.skip_wal {
314            let _ = res.insert(SKIP_WAL_KEY.to_string(), true.to_string());
315        }
316        res.extend(opts.extra_options.clone());
317        res
318    }
319}
320
321/// Alter table request
322#[derive(Debug, Clone, Serialize, Deserialize)]
323pub struct AlterTableRequest {
324    pub catalog_name: String,
325    pub schema_name: String,
326    pub table_name: String,
327    pub table_id: TableId,
328    pub alter_kind: AlterKind,
329    // None in standalone.
330    pub table_version: Option<TableVersion>,
331}
332
333/// Add column request
334#[derive(Debug, Clone, Serialize, Deserialize)]
335pub struct AddColumnRequest {
336    pub column_schema: ColumnSchema,
337    pub is_key: bool,
338    pub location: Option<AddColumnLocation>,
339    /// Add column if not exists.
340    pub add_if_not_exists: bool,
341}
342
343/// Change column datatype request
344#[derive(Debug, Clone, Serialize, Deserialize)]
345pub struct ModifyColumnTypeRequest {
346    pub column_name: String,
347    pub target_type: ConcreteDataType,
348}
349
350/// A family of annotation table options: pure metadata markers that no region
351/// consumes. Setting or unsetting them only rewrites the table's
352/// `extra_options`, so the alter skips region dispatch entirely.
353#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
354pub enum AnnotationFamily {
355    /// `greptime.semantic.*` options (see the [`semantic`] module).
356    Semantic,
357    /// Column and partition count hints consumed by the auto-repartition planner.
358    RepartitionHint,
359}
360
361impl AnnotationFamily {
362    /// The key namespace used in diagnostics; accepted keys are classified by [`Self::of_key`].
363    pub fn namespace(self) -> &'static str {
364        match self {
365            Self::Semantic => SEMANTIC_PREFIX,
366            Self::RepartitionHint => "repartition.",
367        }
368    }
369
370    pub fn of_key(key: &str) -> Option<Self> {
371        if key.starts_with(SEMANTIC_PREFIX) {
372            Some(Self::Semantic)
373        } else if matches!(
374            key,
375            REPARTITION_COLUMN_HINT_KEY | REPARTITION_PARTITION_NUM_HINT_KEY
376        ) {
377            Some(Self::RepartitionHint)
378        } else {
379            None
380        }
381    }
382
383    /// Whether this family may be altered on logical metric tables. Only
384    /// families whose values nothing on the physical side consumes qualify;
385    /// the repartition hint drives physical region repartitioning.
386    pub fn allows_logical_tables(self) -> bool {
387        match self {
388            Self::Semantic => true,
389            Self::RepartitionHint => false,
390        }
391    }
392
393    /// Whether duplicate keys are rejected in this family's SET/UNSET batch.
394    pub fn requires_unique_keys(self) -> bool {
395        matches!(self, Self::RepartitionHint)
396    }
397
398    /// The error for a SET/UNSET batch mixing this family with other options.
399    pub fn mixed_batch_error(self) -> String {
400        match self {
401            Self::Semantic => format!(
402                "`{SEMANTIC_PREFIX}*` options must be altered separately from other table options"
403            ),
404            Self::RepartitionHint => {
405                "repartition hints must be altered separately from other table options".to_string()
406            }
407        }
408    }
409}
410
411/// Why an annotation SET/UNSET key batch was rejected.
412#[derive(Debug, Clone, PartialEq, Eq)]
413pub enum AnnotationKeyError {
414    MixedFamilies { family: AnnotationFamily },
415    DuplicateKey,
416}
417
418impl fmt::Display for AnnotationKeyError {
419    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
420        match self {
421            Self::MixedFamilies { family } => f.write_str(&family.mixed_batch_error()),
422            Self::DuplicateKey => f.write_str("duplicate repartition hint keys"),
423        }
424    }
425}
426
427/// Validates a SET/UNSET batch and returns its annotation family.
428///
429/// Empty batches and batches containing only non-annotation keys return `None`.
430/// Annotation keys cannot share a batch with another family or region options.
431/// Duplicate keys are rejected only for families that require unique keys.
432pub fn validate_annotation_keys<'a>(
433    keys: impl IntoIterator<Item = &'a str>,
434) -> std::result::Result<Option<AnnotationFamily>, AnnotationKeyError> {
435    let mut keys = keys.into_iter();
436    let Some(first) = keys.next() else {
437        return Ok(None);
438    };
439    let family = AnnotationFamily::of_key(first);
440    let reject_duplicates = family.is_some_and(|family| family.requires_unique_keys());
441    let mut seen = HashSet::new();
442    if reject_duplicates {
443        seen.insert(first);
444    }
445    for key in keys {
446        let this = AnnotationFamily::of_key(key);
447        if this != family
448            && let Some(family) = family.or(this)
449        {
450            return Err(AnnotationKeyError::MixedFamilies { family });
451        }
452        if reject_duplicates && !seen.insert(key) {
453            return Err(AnnotationKeyError::DuplicateKey);
454        }
455    }
456    Ok(family)
457}
458
459/// Table shape an annotation option is validated against.
460pub struct AnnotationContext<'a> {
461    pub schema: &'a Schema,
462    pub partition_key_indices: &'a [usize],
463}
464
465/// Why an annotation option was rejected. Typed so each DDL entry point maps
466/// rules onto its existing error variants and status codes: ALTER keeps
467/// missing columns as `TableColumnNotFound` (4002), CREATE keeps its
468/// `InvalidArguments` family — the rules converge, the contracts do not.
469#[derive(Debug, Clone, PartialEq, Eq)]
470pub enum AnnotationValidationError {
471    UnknownKey {
472        key: String,
473    },
474    InvalidValue {
475        key: String,
476        value: String,
477    },
478    ColumnNotFound {
479        column: String,
480    },
481    ColumnNotStringForm {
482        key: String,
483        column: String,
484        ty: ConcreteDataType,
485    },
486    InvalidPartitionNumHint {
487        value: String,
488    },
489    NotSingleColumn,
490    PartitionMetadataConflict,
491    TimeIndexConflict,
492}
493
494impl fmt::Display for AnnotationValidationError {
495    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
496        match self {
497            Self::UnknownKey { key } => write!(f, "unknown semantic option `{key}`"),
498            Self::InvalidValue { key, value } => {
499                write!(f, "invalid value `{value}` for semantic option `{key}`")
500            }
501            Self::ColumnNotFound { column } => write!(f, "column `{column}` not found"),
502            Self::ColumnNotStringForm { key, column, ty } => write!(
503                f,
504                "entity column `{column}` (option `{key}`) has type `{ty}`, \
505                 which cannot render as a string"
506            ),
507            Self::InvalidPartitionNumHint { value } => write!(
508                f,
509                "{REPARTITION_PARTITION_NUM_HINT_KEY} expects a positive integer within u32 range, got `{value}`"
510            ),
511            Self::NotSingleColumn => write!(
512                f,
513                "{REPARTITION_COLUMN_HINT_KEY} expects exactly one column name"
514            ),
515            Self::PartitionMetadataConflict => write!(
516                f,
517                "cannot set {REPARTITION_COLUMN_HINT_KEY} on a table with partition metadata"
518            ),
519            Self::TimeIndexConflict => write!(
520                f,
521                "cannot set {REPARTITION_COLUMN_HINT_KEY} to the time index column"
522            ),
523        }
524    }
525}
526
527/// Validates one annotation option and returns the value to store — the
528/// repartition hints are trimmed, semantic values pass
529/// through unchanged.
530pub(crate) fn validate_and_normalize_annotation(
531    family: AnnotationFamily,
532    cx: &AnnotationContext<'_>,
533    key: &str,
534    value: &str,
535) -> std::result::Result<String, AnnotationValidationError> {
536    match family {
537        AnnotationFamily::Semantic => {
538            if !is_semantic_option_key(key) {
539                return Err(AnnotationValidationError::UnknownKey {
540                    key: key.to_string(),
541                });
542            }
543            if !validate_semantic_option(key, value) {
544                return Err(AnnotationValidationError::InvalidValue {
545                    key: key.to_string(),
546                    value: value.to_string(),
547                });
548            }
549            if parse_entity_option_key(key).is_some() {
550                for column in parse_entity_columns(value) {
551                    let schema = cx.schema.column_schema_by_name(&column).ok_or_else(|| {
552                        AnnotationValidationError::ColumnNotFound {
553                            column: column.clone(),
554                        }
555                    })?;
556                    if !has_stable_string_form(&schema.data_type) {
557                        return Err(AnnotationValidationError::ColumnNotStringForm {
558                            key: key.to_string(),
559                            column,
560                            ty: schema.data_type.clone(),
561                        });
562                    }
563                }
564            }
565            Ok(value.to_string())
566        }
567        AnnotationFamily::RepartitionHint if key == REPARTITION_PARTITION_NUM_HINT_KEY => {
568            let value = value.trim();
569            if !matches!(value.parse::<u32>(), Ok(1..)) {
570                return Err(AnnotationValidationError::InvalidPartitionNumHint {
571                    value: value.to_string(),
572                });
573            }
574            Ok(value.to_string())
575        }
576        AnnotationFamily::RepartitionHint => {
577            let column_name = value.trim();
578            if column_name.is_empty() || column_name.contains(',') {
579                return Err(AnnotationValidationError::NotSingleColumn);
580            }
581            if !cx.partition_key_indices.is_empty() {
582                return Err(AnnotationValidationError::PartitionMetadataConflict);
583            }
584            let column_index = cx.schema.column_index_by_name(column_name).ok_or_else(|| {
585                AnnotationValidationError::ColumnNotFound {
586                    column: column_name.to_string(),
587                }
588            })?;
589            if cx.schema.timestamp_index() == Some(column_index) {
590                return Err(AnnotationValidationError::TimeIndexConflict);
591            }
592            Ok(column_name.to_string())
593        }
594    }
595}
596
597/// CREATE-side entry: validates every annotation option present in `options`
598/// and writes normalized values back in place.
599pub fn validate_and_normalize_annotation_options(
600    options: &mut TableOptions,
601    cx: &AnnotationContext<'_>,
602) -> std::result::Result<(), AnnotationValidationError> {
603    let mut normalized = Vec::new();
604    for (key, value) in &options.extra_options {
605        let Some(family) = AnnotationFamily::of_key(key) else {
606            continue;
607        };
608        let checked = validate_and_normalize_annotation(family, cx, key, value)?;
609        if checked != *value {
610            normalized.push((key.clone(), checked));
611        }
612    }
613    for (key, value) in normalized {
614        options.extra_options.insert(key, value);
615    }
616    Ok(())
617}
618
619#[derive(Debug, Clone, Serialize, Deserialize)]
620pub enum AlterKind {
621    AddColumns {
622        columns: Vec<AddColumnRequest>,
623    },
624    DropColumns {
625        names: Vec<String>,
626    },
627    ModifyColumnTypes {
628        columns: Vec<ModifyColumnTypeRequest>,
629    },
630    RenameTable {
631        new_table_name: String,
632    },
633    SetTableOptions {
634        options: Vec<SetRegionOption>,
635    },
636    UnsetTableOptions {
637        keys: Vec<UnsetRegionOption>,
638    },
639    SetAnnotations {
640        family: AnnotationFamily,
641        options: Vec<(String, String)>,
642    },
643    UnsetAnnotations {
644        family: AnnotationFamily,
645        keys: Vec<String>,
646    },
647    SetIndexes {
648        options: Vec<SetIndexOption>,
649    },
650    UnsetIndexes {
651        options: Vec<UnsetIndexOption>,
652    },
653    DropDefaults {
654        names: Vec<String>,
655    },
656    SetDefaults {
657        defaults: Vec<SetDefaultRequest>,
658    },
659}
660
661#[derive(Debug, Clone, Serialize, Deserialize)]
662pub struct SetDefaultRequest {
663    pub column_name: String,
664    pub default_constraint: Option<ColumnDefaultConstraint>,
665}
666
667#[derive(Debug, Clone, Serialize, Deserialize)]
668pub enum SetIndexOption {
669    Fulltext {
670        column_name: String,
671        options: FulltextOptions,
672    },
673    Inverted {
674        column_name: String,
675    },
676    Skipping {
677        column_name: String,
678        options: SkippingIndexOptions,
679    },
680}
681
682impl SetIndexOption {
683    /// Returns the column name of the index option.
684    pub fn column_name(&self) -> &str {
685        match self {
686            SetIndexOption::Fulltext { column_name, .. } => column_name,
687            SetIndexOption::Inverted { column_name, .. } => column_name,
688            SetIndexOption::Skipping { column_name, .. } => column_name,
689        }
690    }
691}
692
693#[derive(Debug, Clone, Serialize, Deserialize)]
694pub enum UnsetIndexOption {
695    Fulltext { column_name: String },
696    Inverted { column_name: String },
697    Skipping { column_name: String },
698}
699
700impl UnsetIndexOption {
701    /// Returns the column name of the index option.
702    pub fn column_name(&self) -> &str {
703        match self {
704            UnsetIndexOption::Fulltext { column_name, .. } => column_name,
705            UnsetIndexOption::Inverted { column_name, .. } => column_name,
706            UnsetIndexOption::Skipping { column_name, .. } => column_name,
707        }
708    }
709}
710
711#[derive(Debug)]
712pub struct InsertRequest {
713    pub catalog_name: String,
714    pub schema_name: String,
715    pub table_name: String,
716    pub columns_values: HashMap<String, VectorRef>,
717    /// Whether this insert should skip WAL.
718    pub skip_wal: bool,
719}
720
721/// Delete (by primary key) request
722#[derive(Debug)]
723pub struct DeleteRequest {
724    pub catalog_name: String,
725    pub schema_name: String,
726    pub table_name: String,
727    /// Values of each column in this table's primary key and time index.
728    ///
729    /// The key is the column name, and the value is the column value.
730    pub key_column_values: HashMap<String, VectorRef>,
731}
732
733#[derive(Debug)]
734pub enum CopyDirection {
735    Export,
736    Import,
737}
738
739/// Copy table request
740#[derive(Debug)]
741pub struct CopyTableRequest {
742    pub catalog_name: String,
743    pub schema_name: String,
744    pub table_name: String,
745    pub location: String,
746    pub with: HashMap<String, String>,
747    pub connection: HashMap<String, String>,
748    pub pattern: Option<String>,
749    pub direction: CopyDirection,
750    pub timestamp_range: Option<TimestampRange>,
751    pub limit: Option<u64>,
752}
753
754#[derive(Debug, Clone, Default)]
755pub struct FlushTableRequest {
756    pub catalog_name: String,
757    pub schema_name: String,
758    pub table_name: String,
759}
760
761#[derive(Debug, Clone, Default)]
762pub struct BuildIndexTableRequest {
763    pub catalog_name: String,
764    pub schema_name: String,
765    pub table_name: String,
766}
767
768#[derive(Debug, Clone, PartialEq)]
769pub struct CompactTableRequest {
770    pub catalog_name: String,
771    pub schema_name: String,
772    pub table_name: String,
773    pub compact_options: compact_request::Options,
774    pub parallelism: u32,
775    pub time_range: Option<TimestampRange>,
776}
777
778impl Default for CompactTableRequest {
779    fn default() -> Self {
780        Self {
781            catalog_name: Default::default(),
782            schema_name: Default::default(),
783            table_name: Default::default(),
784            compact_options: compact_request::Options::Regular(Default::default()),
785            parallelism: 1,
786            time_range: None,
787        }
788    }
789}
790
791/// Truncate table request
792#[derive(Debug, Clone, Serialize, Deserialize)]
793pub struct TruncateTableRequest {
794    pub catalog_name: String,
795    pub schema_name: String,
796    pub table_name: String,
797    pub table_id: TableId,
798}
799
800impl TruncateTableRequest {
801    pub fn table_ref(&self) -> TableReference<'_> {
802        TableReference {
803            catalog: &self.catalog_name,
804            schema: &self.schema_name,
805            table: &self.table_name,
806        }
807    }
808}
809
810#[derive(Debug, Clone, Default, Deserialize, Serialize)]
811pub struct CopyDatabaseRequest {
812    pub catalog_name: String,
813    pub schema_name: String,
814    pub location: String,
815    pub with: HashMap<String, String>,
816    pub connection: HashMap<String, String>,
817    pub time_range: Option<TimestampRange>,
818}
819
820#[derive(Debug, Clone, Default, Deserialize, Serialize)]
821pub struct CopyQueryToRequest {
822    pub location: String,
823    pub with: HashMap<String, String>,
824    pub connection: HashMap<String, String>,
825}
826
827#[cfg(test)]
828mod tests {
829    use std::time::Duration;
830
831    use common_error::ext::ErrorExt;
832    use common_error::status_code::StatusCode;
833
834    use super::*;
835
836    #[test]
837    fn test_validate_annotation_keys() {
838        let column = REPARTITION_COLUMN_HINT_KEY;
839        let count = REPARTITION_PARTITION_NUM_HINT_KEY;
840        for (keys, expected) in [
841            (vec![], None),
842            (vec![TTL_KEY, TTL_KEY], None),
843            (vec!["repartition.unknown.hint"], None),
844            (vec![column], Some(AnnotationFamily::RepartitionHint)),
845            (vec![count], Some(AnnotationFamily::RepartitionHint)),
846            (vec![column, count], Some(AnnotationFamily::RepartitionHint)),
847            (vec![count, column], Some(AnnotationFamily::RepartitionHint)),
848            (
849                vec!["greptime.semantic.source", "greptime.semantic.source"],
850                Some(AnnotationFamily::Semantic),
851            ),
852        ] {
853            assert_eq!(validate_annotation_keys(keys), Ok(expected));
854        }
855        for keys in [
856            vec![column, column],
857            vec![count, count],
858            vec![column, count, column],
859        ] {
860            assert_eq!(
861                validate_annotation_keys(keys),
862                Err(AnnotationKeyError::DuplicateKey)
863            );
864        }
865        for keys in [
866            vec![column, TTL_KEY],
867            vec![TTL_KEY, count],
868            vec![column, "greptime.semantic.source"],
869        ] {
870            assert_eq!(
871                validate_annotation_keys(keys),
872                Err(AnnotationKeyError::MixedFamilies {
873                    family: AnnotationFamily::RepartitionHint,
874                })
875            );
876        }
877    }
878
879    #[test]
880    fn test_validate_table_option() {
881        assert!(validate_table_option(FILE_TABLE_LOCATION_KEY));
882        assert!(validate_table_option(FILE_TABLE_FORMAT_KEY));
883        assert!(validate_table_option(FILE_TABLE_PATTERN_KEY));
884        assert!(validate_table_option(TTL_KEY));
885        assert!(validate_table_option(WRITE_BUFFER_SIZE_KEY));
886        assert!(validate_table_option(STORAGE_KEY));
887        assert!(validate_table_option(MEMTABLE_BULK_MERGE_THRESHOLD));
888        assert!(validate_table_option(REPARTITION_COLUMN_HINT_KEY));
889        assert!(validate_table_option(REPARTITION_PARTITION_NUM_HINT_KEY));
890        assert_eq!(AnnotationFamily::of_key("repartition.unknown.hint"), None);
891        assert!(!validate_table_option("foo"));
892
893        // Only whitelisted semantic keys are accepted.
894        assert!(validate_table_option(SEMANTIC_SIGNAL_TYPE));
895        assert!(validate_table_option(SEMANTIC_METRIC_TYPE));
896        // Unknown semantic key, near-miss, and the internal transport key are rejected.
897        assert!(!validate_table_option("greptime.semantic.future.key"));
898        assert!(!validate_table_option("greptime.semanticx"));
899        assert!(!validate_table_option(SEMANTIC_PER_TABLE_INDEX_KEY));
900    }
901
902    #[test]
903    fn test_validate_database_option() {
904        assert!(validate_database_option(MEMTABLE_TYPE));
905        assert!(validate_database_option(MEMTABLE_BULK_MERGE_THRESHOLD));
906        assert!(validate_database_option(MEMTABLE_BULK_ENCODE_ROW_THRESHOLD));
907        assert!(validate_database_option(
908            MEMTABLE_BULK_ENCODE_BYTES_THRESHOLD
909        ));
910        assert!(validate_database_option(MEMTABLE_BULK_MAX_MERGE_GROUPS));
911        assert!(validate_database_option(
912            "compaction.twcs.active_window.trigger_file_num"
913        ));
914        assert!(validate_database_option(
915            "compaction.twcs.active_window.l1_merge_trigger"
916        ));
917        assert!(validate_database_option(
918            "compaction.twcs.inactive_window.trigger_file_num"
919        ));
920        assert!(validate_database_option(
921            "compaction.twcs.inactive_window.l1_merge_trigger"
922        ));
923        assert!(!validate_database_option("foo"));
924    }
925
926    #[test]
927    fn test_database_trigger_value_boundaries() {
928        let maximum = usize::MAX.to_string();
929        let overflow = format!("{maximum}0");
930        for (key, minimum) in [
931            (TWCS_TRIGGER_FILE_NUM, 0),
932            (TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM, 0),
933            (TWCS_INACTIVE_WINDOW_TRIGGER_FILE_NUM, 0),
934            (TWCS_ACTIVE_WINDOW_L1_MERGE_TRIGGER, 2),
935            (TWCS_INACTIVE_WINDOW_L1_MERGE_TRIGGER, 2),
936        ] {
937            for invalid in [
938                None,
939                Some(""),
940                Some("invalid"),
941                Some("-1"),
942                Some(overflow.as_str()),
943            ] {
944                assert!(
945                    validate_database_option_value(key, invalid).is_err(),
946                    "{key}: {invalid:?}"
947                );
948            }
949            for valid in ["2", maximum.as_str()] {
950                assert!(
951                    validate_database_option_value(key, Some(valid)).is_ok(),
952                    "{key}: {valid}"
953                );
954            }
955            for boundary in ["0", "1"] {
956                assert_eq!(
957                    validate_database_option_value(key, Some(boundary)).is_ok(),
958                    minimum == 0,
959                    "{key}: {boundary}"
960                );
961            }
962        }
963    }
964
965    #[test]
966    fn test_serialize_table_options() {
967        let options = TableOptions {
968            write_buffer_size: None,
969            ttl: Some(Duration::from_secs(1000).into()),
970            extra_options: HashMap::new(),
971            skip_wal: false,
972        };
973        let serialized = serde_json::to_string(&options).unwrap();
974        let deserialized: TableOptions = serde_json::from_str(&serialized).unwrap();
975        assert_eq!(options, deserialized);
976    }
977
978    #[test]
979    fn test_convert_hashmap_between_table_options() {
980        let options = TableOptions {
981            write_buffer_size: Some(ReadableSize::mb(128)),
982            ttl: Some(Duration::from_secs(1000).into()),
983            extra_options: HashMap::new(),
984            skip_wal: false,
985        };
986        let serialized_map = HashMap::from(&options);
987        let serialized = TableOptions::try_from_iter(&serialized_map).unwrap();
988        assert_eq!(options, serialized);
989
990        let options = TableOptions {
991            write_buffer_size: None,
992            ttl: None,
993            extra_options: HashMap::from([(SKIP_WAL_KEY.to_string(), true.to_string())]),
994            skip_wal: true,
995        };
996        let serialized_map = HashMap::from(&options);
997        assert_eq!(
998            Some("true"),
999            serialized_map.get(SKIP_WAL_KEY).map(String::as_str)
1000        );
1001        let serialized = TableOptions::try_from_iter(&serialized_map).unwrap();
1002        assert_eq!(options, serialized);
1003
1004        let options = TableOptions {
1005            write_buffer_size: None,
1006            ttl: Default::default(),
1007            extra_options: HashMap::new(),
1008            skip_wal: false,
1009        };
1010        let serialized_map = HashMap::from(&options);
1011        let serialized = TableOptions::try_from_iter(&serialized_map).unwrap();
1012        assert_eq!(options, serialized);
1013
1014        let options = TableOptions {
1015            write_buffer_size: Some(ReadableSize::mb(128)),
1016            ttl: Some(Duration::from_secs(1000).into()),
1017            extra_options: HashMap::from([("a".to_string(), "A".to_string())]),
1018            skip_wal: false,
1019        };
1020        let serialized_map = HashMap::from(&options);
1021        let serialized = TableOptions::try_from_iter(&serialized_map).unwrap();
1022        assert_eq!(options, serialized);
1023
1024        let options = TableOptions {
1025            extra_options: HashMap::from([(SKIP_WAL_KEY.to_string(), false.to_string())]),
1026            skip_wal: false,
1027            ..Default::default()
1028        };
1029        let serialized_map = HashMap::from(&options);
1030        let serialized = TableOptions::try_from_iter(&serialized_map).unwrap();
1031        assert_eq!(options, serialized);
1032    }
1033
1034    #[test]
1035    fn test_table_options_normalizes_twcs_trigger_aliases() {
1036        for options in [
1037            vec![(TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM, "4")],
1038            vec![
1039                (TWCS_TRIGGER_FILE_NUM, "4"),
1040                (TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM, "4"),
1041            ],
1042        ] {
1043            let table_options = TableOptions::try_from_iter(options).unwrap();
1044            assert_eq!(
1045                HashMap::from([(TWCS_TRIGGER_FILE_NUM.to_string(), "4".to_string())]),
1046                table_options.extra_options
1047            );
1048        }
1049    }
1050
1051    #[test]
1052    fn test_table_options_rejects_conflicting_twcs_trigger_aliases() {
1053        let error = TableOptions::try_from_iter([
1054            (TWCS_TRIGGER_FILE_NUM, "4"),
1055            (TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM, "8"),
1056        ])
1057        .unwrap_err();
1058        assert_eq!(StatusCode::InvalidArguments, error.status_code());
1059        assert_eq!(
1060            "Conflicting table options: compaction.twcs.trigger_file_num=4 and compaction.twcs.active_window.trigger_file_num=8",
1061            error.to_string()
1062        );
1063    }
1064
1065    #[test]
1066    fn test_table_options_to_string() {
1067        let options = TableOptions {
1068            write_buffer_size: Some(ReadableSize::mb(128)),
1069            ttl: Some(Duration::from_secs(1000).into()),
1070            extra_options: HashMap::new(),
1071            skip_wal: false,
1072        };
1073
1074        assert_eq!(
1075            "write_buffer_size=128.0MiB ttl=16m 40s",
1076            options.to_string()
1077        );
1078
1079        let options = TableOptions {
1080            write_buffer_size: Some(ReadableSize::mb(128)),
1081            ttl: Some(Duration::from_secs(1000).into()),
1082            extra_options: HashMap::from([("a".to_string(), "A".to_string())]),
1083            skip_wal: false,
1084        };
1085
1086        assert_eq!(
1087            "write_buffer_size=128.0MiB ttl=16m 40s a=A",
1088            options.to_string()
1089        );
1090
1091        let options = TableOptions {
1092            write_buffer_size: Some(ReadableSize::mb(128)),
1093            ttl: Some(Duration::from_secs(1000).into()),
1094            extra_options: HashMap::new(),
1095            skip_wal: true,
1096        };
1097        assert_eq!(
1098            "write_buffer_size=128.0MiB ttl=16m 40s skip_wal=true",
1099            options.to_string()
1100        );
1101
1102        let options = TableOptions {
1103            write_buffer_size: None,
1104            ttl: None,
1105            extra_options: HashMap::from([(SKIP_WAL_KEY.to_string(), "false".to_string())]),
1106            skip_wal: false,
1107        };
1108        assert_eq!("skip_wal=false", options.to_string());
1109    }
1110}
table/requests.rs

table/
requests.rs
