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

table/
requests.rs
