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

table/
requests.rs
