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

table/
requests.rs
