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::{BTreeMap, 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
319/// Object store credential keys accepted by [`validate_table_option`].
320const CREDENTIAL_OPTION_KEYS: [&str; 4] = [
321    "access_key_id",
322    "secret_access_key",
323    "session_token",
324    "access_key_secret",
325];
326
327/// Placeholder rendered instead of a credential value.
328pub const REDACTED_OPTION_VALUE: &str = "******";
329
330pub fn is_credential_option(key: &str) -> bool {
331    CREDENTIAL_OPTION_KEYS.contains(&key)
332}
333
334/// Renders the options for users, e.g. `information_schema.tables.create_options`,
335/// so credential values are masked.
336impl fmt::Display for TableOptions {
337    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
338        let mut key_vals = vec![];
339        if let Some(size) = self.write_buffer_size {
340            key_vals.push(format!("{}={}", WRITE_BUFFER_SIZE_KEY, size));
341        }
342
343        if let Some(ttl) = self.ttl.map(|ttl| ttl.to_string()) {
344            key_vals.push(format!("{}={}", TTL_KEY, ttl));
345        }
346
347        if self.skip_wal && !self.extra_options.contains_key(SKIP_WAL_KEY) {
348            key_vals.push(format!("{}={}", SKIP_WAL_KEY, self.skip_wal));
349        }
350
351        // Sort by key so `create_options` is stable across queries; `HashMap` order is not.
352        let extra_options: BTreeMap<_, _> = self.extra_options.iter().collect();
353        for (k, v) in extra_options {
354            let v = if is_credential_option(k) {
355                REDACTED_OPTION_VALUE
356            } else {
357                v
358            };
359            key_vals.push(format!("{}={}", k, v));
360        }
361
362        write!(f, "{}", key_vals.join(" "))
363    }
364}
365
366impl From<&TableOptions> for HashMap<String, String> {
367    fn from(opts: &TableOptions) -> Self {
368        let mut res = HashMap::with_capacity(3 + opts.extra_options.len());
369        if let Some(write_buffer_size) = opts.write_buffer_size {
370            let _ = res.insert(
371                WRITE_BUFFER_SIZE_KEY.to_string(),
372                write_buffer_size.to_string(),
373            );
374        }
375        if let Some(ttl_str) = opts.ttl.map(|ttl| ttl.to_string()) {
376            let _ = res.insert(TTL_KEY.to_string(), ttl_str);
377        }
378        if opts.skip_wal {
379            let _ = res.insert(SKIP_WAL_KEY.to_string(), true.to_string());
380        }
381        res.extend(opts.extra_options.clone());
382        res
383    }
384}
385
386/// Alter table request
387#[derive(Debug, Clone, Serialize, Deserialize)]
388pub struct AlterTableRequest {
389    pub catalog_name: String,
390    pub schema_name: String,
391    pub table_name: String,
392    pub table_id: TableId,
393    pub alter_kind: AlterKind,
394    // None in standalone.
395    pub table_version: Option<TableVersion>,
396}
397
398/// Add column request
399#[derive(Debug, Clone, Serialize, Deserialize)]
400pub struct AddColumnRequest {
401    pub column_schema: ColumnSchema,
402    pub is_key: bool,
403    pub location: Option<AddColumnLocation>,
404    /// Add column if not exists.
405    pub add_if_not_exists: bool,
406}
407
408/// Change column datatype request
409#[derive(Debug, Clone, Serialize, Deserialize)]
410pub struct ModifyColumnTypeRequest {
411    pub column_name: String,
412    pub target_type: ConcreteDataType,
413}
414
415/// Set JSON2 settings request.
416#[derive(Debug, Clone, Serialize, Deserialize)]
417pub struct SetJsonSettingsRequest {
418    pub column_name: String,
419    pub settings: JsonSettings,
420}
421
422/// A family of annotation table options: pure metadata markers that no region
423/// consumes. Setting or unsetting them only rewrites the table's
424/// `extra_options`, so the alter skips region dispatch entirely.
425#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
426pub enum AnnotationFamily {
427    /// `greptime.semantic.*` options (see the [`semantic`] module).
428    Semantic,
429    /// Column and partition count hints consumed by the auto-repartition planner.
430    RepartitionHint,
431}
432
433impl AnnotationFamily {
434    /// The key namespace used in diagnostics; accepted keys are classified by [`Self::of_key`].
435    pub fn namespace(self) -> &'static str {
436        match self {
437            Self::Semantic => SEMANTIC_PREFIX,
438            Self::RepartitionHint => "repartition.",
439        }
440    }
441
442    pub fn of_key(key: &str) -> Option<Self> {
443        if key.starts_with(SEMANTIC_PREFIX) {
444            Some(Self::Semantic)
445        } else if matches!(
446            key,
447            REPARTITION_COLUMN_HINT_KEY | REPARTITION_PARTITION_NUM_HINT_KEY
448        ) {
449            Some(Self::RepartitionHint)
450        } else {
451            None
452        }
453    }
454
455    /// Whether this family may be altered on logical metric tables. Only
456    /// families whose values nothing on the physical side consumes qualify;
457    /// the repartition hint drives physical region repartitioning.
458    pub fn allows_logical_tables(self) -> bool {
459        match self {
460            Self::Semantic => true,
461            Self::RepartitionHint => false,
462        }
463    }
464
465    /// Whether duplicate keys are rejected in this family's SET/UNSET batch.
466    pub fn requires_unique_keys(self) -> bool {
467        matches!(self, Self::RepartitionHint)
468    }
469
470    /// The error for a SET/UNSET batch mixing this family with other options.
471    pub fn mixed_batch_error(self) -> String {
472        match self {
473            Self::Semantic => format!(
474                "`{SEMANTIC_PREFIX}*` options must be altered separately from other table options"
475            ),
476            Self::RepartitionHint => {
477                "repartition hints must be altered separately from other table options".to_string()
478            }
479        }
480    }
481}
482
483/// Why an annotation SET/UNSET key batch was rejected.
484#[derive(Debug, Clone, PartialEq, Eq)]
485pub enum AnnotationKeyError {
486    MixedFamilies { family: AnnotationFamily },
487    DuplicateKey,
488}
489
490impl fmt::Display for AnnotationKeyError {
491    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
492        match self {
493            Self::MixedFamilies { family } => f.write_str(&family.mixed_batch_error()),
494            Self::DuplicateKey => f.write_str("duplicate repartition hint keys"),
495        }
496    }
497}
498
499/// Validates a SET/UNSET batch and returns its annotation family.
500///
501/// Empty batches and batches containing only non-annotation keys return `None`.
502/// Annotation keys cannot share a batch with another family or region options.
503/// Duplicate keys are rejected only for families that require unique keys.
504pub fn validate_annotation_keys<'a>(
505    keys: impl IntoIterator<Item = &'a str>,
506) -> std::result::Result<Option<AnnotationFamily>, AnnotationKeyError> {
507    let mut keys = keys.into_iter();
508    let Some(first) = keys.next() else {
509        return Ok(None);
510    };
511    let family = AnnotationFamily::of_key(first);
512    let reject_duplicates = family.is_some_and(|family| family.requires_unique_keys());
513    let mut seen = HashSet::new();
514    if reject_duplicates {
515        seen.insert(first);
516    }
517    for key in keys {
518        let this = AnnotationFamily::of_key(key);
519        if this != family
520            && let Some(family) = family.or(this)
521        {
522            return Err(AnnotationKeyError::MixedFamilies { family });
523        }
524        if reject_duplicates && !seen.insert(key) {
525            return Err(AnnotationKeyError::DuplicateKey);
526        }
527    }
528    Ok(family)
529}
530
531/// Table shape an annotation option is validated against.
532pub struct AnnotationContext<'a> {
533    pub data_model: Option<&'a str>,
534    pub schema: &'a Schema,
535    pub partition_key_indices: &'a [usize],
536}
537
538/// Why an annotation option was rejected. Typed so each DDL entry point maps
539/// rules onto its existing error variants and status codes: ALTER keeps
540/// missing columns as `TableColumnNotFound` (4002), CREATE keeps its
541/// `InvalidArguments` family — the rules converge, the contracts do not.
542#[derive(Debug, Clone, PartialEq, Eq)]
543pub enum AnnotationValidationError {
544    UnknownKey {
545        key: String,
546    },
547    InvalidValue {
548        key: String,
549        value: String,
550    },
551    ColumnNotFound {
552        column: String,
553    },
554    ColumnNotStringForm {
555        key: String,
556        column: String,
557        ty: ConcreteDataType,
558    },
559    InvalidPartitionNumHint {
560        value: String,
561    },
562    NotSingleColumn,
563    PartitionMetadataConflict,
564    TimeIndexConflict,
565}
566
567impl fmt::Display for AnnotationValidationError {
568    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
569        match self {
570            Self::UnknownKey { key } => write!(f, "unknown semantic option `{key}`"),
571            Self::InvalidValue { key, value } => {
572                write!(f, "invalid value `{value}` for semantic option `{key}`")
573            }
574            Self::ColumnNotFound { column } => write!(f, "column `{column}` not found"),
575            Self::ColumnNotStringForm { key, column, ty } => write!(
576                f,
577                "entity column `{column}` (option `{key}`) has type `{ty}`, \
578                 which cannot render as a string"
579            ),
580            Self::InvalidPartitionNumHint { value } => write!(
581                f,
582                "{REPARTITION_PARTITION_NUM_HINT_KEY} expects a positive integer within u32 range, got `{value}`"
583            ),
584            Self::NotSingleColumn => write!(
585                f,
586                "{REPARTITION_COLUMN_HINT_KEY} expects exactly one column name"
587            ),
588            Self::PartitionMetadataConflict => write!(
589                f,
590                "cannot set {REPARTITION_COLUMN_HINT_KEY} on a table with partition metadata"
591            ),
592            Self::TimeIndexConflict => write!(
593                f,
594                "cannot set {REPARTITION_COLUMN_HINT_KEY} to the time index column"
595            ),
596        }
597    }
598}
599
600/// Validates one annotation option and returns the value to store — the
601/// repartition hints are trimmed, semantic values pass
602/// through unchanged.
603pub(crate) fn validate_and_normalize_annotation(
604    family: AnnotationFamily,
605    cx: &AnnotationContext<'_>,
606    key: &str,
607    value: &str,
608) -> std::result::Result<String, AnnotationValidationError> {
609    match family {
610        AnnotationFamily::Semantic => {
611            if !is_semantic_option_key(key) {
612                return Err(AnnotationValidationError::UnknownKey {
613                    key: key.to_string(),
614                });
615            }
616            if !validate_semantic_option(key, value) {
617                return Err(AnnotationValidationError::InvalidValue {
618                    key: key.to_string(),
619                    value: value.to_string(),
620                });
621            }
622            if parse_entity_option_key(key).is_some() {
623                for column in parse_entity_columns(value) {
624                    if trace_v2_attribute(cx.schema, cx.data_model, &column).is_some() {
625                        continue;
626                    }
627                    let schema = cx.schema.column_schema_by_name(&column).ok_or_else(|| {
628                        AnnotationValidationError::ColumnNotFound {
629                            column: column.clone(),
630                        }
631                    })?;
632                    if !has_stable_string_form(&schema.data_type) {
633                        return Err(AnnotationValidationError::ColumnNotStringForm {
634                            key: key.to_string(),
635                            column,
636                            ty: schema.data_type.clone(),
637                        });
638                    }
639                }
640            }
641            Ok(value.to_string())
642        }
643        AnnotationFamily::RepartitionHint if key == REPARTITION_PARTITION_NUM_HINT_KEY => {
644            let value = value.trim();
645            if !matches!(value.parse::<u32>(), Ok(1..)) {
646                return Err(AnnotationValidationError::InvalidPartitionNumHint {
647                    value: value.to_string(),
648                });
649            }
650            Ok(value.to_string())
651        }
652        AnnotationFamily::RepartitionHint => {
653            let column_name = value.trim();
654            if column_name.is_empty() || column_name.contains(',') {
655                return Err(AnnotationValidationError::NotSingleColumn);
656            }
657            if !cx.partition_key_indices.is_empty() {
658                return Err(AnnotationValidationError::PartitionMetadataConflict);
659            }
660            let column_index = cx.schema.column_index_by_name(column_name).ok_or_else(|| {
661                AnnotationValidationError::ColumnNotFound {
662                    column: column_name.to_string(),
663                }
664            })?;
665            if cx.schema.timestamp_index() == Some(column_index) {
666                return Err(AnnotationValidationError::TimeIndexConflict);
667            }
668            Ok(column_name.to_string())
669        }
670    }
671}
672
673/// CREATE-side entry: validates every annotation option present in `options`
674/// and writes normalized values back in place.
675pub fn validate_and_normalize_annotation_options(
676    options: &mut TableOptions,
677    schema: &Schema,
678    partition_key_indices: &[usize],
679) -> std::result::Result<(), AnnotationValidationError> {
680    let cx = AnnotationContext {
681        data_model: options.data_model(),
682        schema,
683        partition_key_indices,
684    };
685    let mut normalized = Vec::new();
686    for (key, value) in &options.extra_options {
687        let Some(family) = AnnotationFamily::of_key(key) else {
688            continue;
689        };
690        let checked = validate_and_normalize_annotation(family, &cx, key, value)?;
691        if checked != *value {
692            normalized.push((key.clone(), checked));
693        }
694    }
695    for (key, value) in normalized {
696        options.extra_options.insert(key, value);
697    }
698    Ok(())
699}
700
701#[derive(Debug, Clone, Serialize, Deserialize)]
702pub enum AlterKind {
703    AddColumns {
704        columns: Vec<AddColumnRequest>,
705    },
706    DropColumns {
707        names: Vec<String>,
708    },
709    ModifyColumnTypes {
710        columns: Vec<ModifyColumnTypeRequest>,
711    },
712    SetJsonSettings {
713        request: SetJsonSettingsRequest,
714    },
715    RenameTable {
716        new_table_name: String,
717    },
718    SetTableOptions {
719        options: Vec<SetRegionOption>,
720    },
721    UnsetTableOptions {
722        keys: Vec<UnsetRegionOption>,
723    },
724    SetAnnotations {
725        family: AnnotationFamily,
726        options: Vec<(String, String)>,
727    },
728    UnsetAnnotations {
729        family: AnnotationFamily,
730        keys: Vec<String>,
731    },
732    SetIndexes {
733        options: Vec<SetIndexOption>,
734    },
735    UnsetIndexes {
736        options: Vec<UnsetIndexOption>,
737    },
738    DropDefaults {
739        names: Vec<String>,
740    },
741    SetDefaults {
742        defaults: Vec<SetDefaultRequest>,
743    },
744}
745
746#[derive(Debug, Clone, Serialize, Deserialize)]
747pub struct SetDefaultRequest {
748    pub column_name: String,
749    pub default_constraint: Option<ColumnDefaultConstraint>,
750}
751
752#[derive(Debug, Clone, Serialize, Deserialize)]
753pub enum SetIndexOption {
754    Fulltext {
755        column_name: String,
756        options: FulltextOptions,
757    },
758    Inverted {
759        column_name: String,
760    },
761    Skipping {
762        column_name: String,
763        options: SkippingIndexOptions,
764    },
765}
766
767impl SetIndexOption {
768    /// Returns the column name of the index option.
769    pub fn column_name(&self) -> &str {
770        match self {
771            SetIndexOption::Fulltext { column_name, .. } => column_name,
772            SetIndexOption::Inverted { column_name, .. } => column_name,
773            SetIndexOption::Skipping { column_name, .. } => column_name,
774        }
775    }
776}
777
778#[derive(Debug, Clone, Serialize, Deserialize)]
779pub enum UnsetIndexOption {
780    Fulltext { column_name: String },
781    Inverted { column_name: String },
782    Skipping { column_name: String },
783}
784
785impl UnsetIndexOption {
786    /// Returns the column name of the index option.
787    pub fn column_name(&self) -> &str {
788        match self {
789            UnsetIndexOption::Fulltext { column_name, .. } => column_name,
790            UnsetIndexOption::Inverted { column_name, .. } => column_name,
791            UnsetIndexOption::Skipping { column_name, .. } => column_name,
792        }
793    }
794}
795
796#[derive(Debug)]
797pub struct InsertRequest {
798    pub catalog_name: String,
799    pub schema_name: String,
800    pub table_name: String,
801    pub columns_values: HashMap<String, VectorRef>,
802    /// Whether this insert should skip WAL.
803    pub skip_wal: bool,
804}
805
806/// Delete (by primary key) request
807#[derive(Debug)]
808pub struct DeleteRequest {
809    pub catalog_name: String,
810    pub schema_name: String,
811    pub table_name: String,
812    /// Values of each column in this table's primary key and time index.
813    ///
814    /// The key is the column name, and the value is the column value.
815    pub key_column_values: HashMap<String, VectorRef>,
816}
817
818#[derive(Debug)]
819pub enum CopyDirection {
820    Export,
821    Import,
822}
823
824/// Copy table request
825#[derive(Debug)]
826pub struct CopyTableRequest {
827    pub catalog_name: String,
828    pub schema_name: String,
829    pub table_name: String,
830    pub location: String,
831    pub with: HashMap<String, String>,
832    pub connection: HashMap<String, String>,
833    pub pattern: Option<String>,
834    pub direction: CopyDirection,
835    pub timestamp_range: Option<TimestampRange>,
836    pub limit: Option<u64>,
837}
838
839#[derive(Debug, Clone, Default)]
840pub struct FlushTableRequest {
841    pub catalog_name: String,
842    pub schema_name: String,
843    pub table_name: String,
844}
845
846#[derive(Debug, Clone, Default)]
847pub struct BuildIndexTableRequest {
848    /// The index build mode. Absent options select SST indexes.
849    pub options: Option<build_index_request::Options>,
850    pub catalog_name: String,
851    pub schema_name: String,
852    pub table_name: String,
853}
854
855#[derive(Debug, Clone, PartialEq)]
856pub struct CompactTableRequest {
857    pub catalog_name: String,
858    pub schema_name: String,
859    pub table_name: String,
860    pub compact_options: compact_request::Options,
861    pub parallelism: u32,
862    pub time_range: Option<TimestampRange>,
863}
864
865impl Default for CompactTableRequest {
866    fn default() -> Self {
867        Self {
868            catalog_name: Default::default(),
869            schema_name: Default::default(),
870            table_name: Default::default(),
871            compact_options: compact_request::Options::Regular(Default::default()),
872            parallelism: 1,
873            time_range: None,
874        }
875    }
876}
877
878#[derive(Debug, Clone, Default, Deserialize, Serialize)]
879pub struct CopyDatabaseRequest {
880    pub catalog_name: String,
881    pub schema_name: String,
882    pub location: String,
883    pub with: HashMap<String, String>,
884    pub connection: HashMap<String, String>,
885    pub time_range: Option<TimestampRange>,
886}
887
888#[derive(Debug, Clone, Default, Deserialize, Serialize)]
889pub struct CopyQueryToRequest {
890    pub location: String,
891    pub with: HashMap<String, String>,
892    pub connection: HashMap<String, String>,
893}
894
895#[cfg(test)]
896mod tests {
897    use std::time::Duration;
898
899    use common_error::ext::ErrorExt;
900    use common_error::status_code::StatusCode;
901
902    use super::*;
903
904    #[test]
905    fn test_validate_annotation_keys() {
906        let column = REPARTITION_COLUMN_HINT_KEY;
907        let count = REPARTITION_PARTITION_NUM_HINT_KEY;
908        for (keys, expected) in [
909            (vec![], None),
910            (vec![TTL_KEY, TTL_KEY], None),
911            (vec!["repartition.unknown.hint"], None),
912            (vec![column], Some(AnnotationFamily::RepartitionHint)),
913            (vec![count], Some(AnnotationFamily::RepartitionHint)),
914            (vec![column, count], Some(AnnotationFamily::RepartitionHint)),
915            (vec![count, column], Some(AnnotationFamily::RepartitionHint)),
916            (
917                vec!["greptime.semantic.source", "greptime.semantic.source"],
918                Some(AnnotationFamily::Semantic),
919            ),
920        ] {
921            assert_eq!(validate_annotation_keys(keys), Ok(expected));
922        }
923        for keys in [
924            vec![column, column],
925            vec![count, count],
926            vec![column, count, column],
927        ] {
928            assert_eq!(
929                validate_annotation_keys(keys),
930                Err(AnnotationKeyError::DuplicateKey)
931            );
932        }
933        for keys in [
934            vec![column, TTL_KEY],
935            vec![TTL_KEY, count],
936            vec![column, "greptime.semantic.source"],
937        ] {
938            assert_eq!(
939                validate_annotation_keys(keys),
940                Err(AnnotationKeyError::MixedFamilies {
941                    family: AnnotationFamily::RepartitionHint,
942                })
943            );
944        }
945    }
946
947    #[test]
948    fn test_validate_table_option() {
949        assert!(validate_table_option(FILE_TABLE_LOCATION_KEY));
950        assert!(validate_table_option(FILE_TABLE_FORMAT_KEY));
951        assert!(validate_table_option(FILE_TABLE_PATTERN_KEY));
952        assert!(validate_table_option(TTL_KEY));
953        assert!(validate_table_option(WRITE_BUFFER_SIZE_KEY));
954        assert!(validate_table_option(STORAGE_KEY));
955        assert!(validate_table_option(MEMTABLE_BULK_MERGE_THRESHOLD));
956        assert!(validate_table_option(EXPERIMENTAL_SST_FLOAT_FIELD_ENCODING));
957        assert!(validate_table_option(REPARTITION_COLUMN_HINT_KEY));
958        assert!(validate_table_option(REPARTITION_PARTITION_NUM_HINT_KEY));
959        assert_eq!(AnnotationFamily::of_key("repartition.unknown.hint"), None);
960        assert!(!validate_table_option("foo"));
961
962        // Only whitelisted semantic keys are accepted.
963        assert!(validate_table_option(SEMANTIC_SIGNAL_TYPE));
964        assert!(validate_table_option(SEMANTIC_METRIC_TYPE));
965        // Unknown semantic key, near-miss, and the internal transport key are rejected.
966        assert!(!validate_table_option("greptime.semantic.future.key"));
967        assert!(!validate_table_option("greptime.semanticx"));
968        assert!(!validate_table_option(SEMANTIC_PER_TABLE_INDEX_KEY));
969    }
970
971    #[test]
972    fn test_validate_database_option() {
973        assert!(validate_database_option(MEMTABLE_TYPE));
974        assert!(validate_database_option(MEMTABLE_BULK_MERGE_THRESHOLD));
975        assert!(validate_database_option(MEMTABLE_BULK_ENCODE_ROW_THRESHOLD));
976        assert!(validate_database_option(
977            MEMTABLE_BULK_ENCODE_BYTES_THRESHOLD
978        ));
979        assert!(validate_database_option(MEMTABLE_BULK_MAX_MERGE_GROUPS));
980        assert!(validate_database_option(
981            "compaction.twcs.active_window.trigger_file_num"
982        ));
983        assert!(validate_database_option(
984            "compaction.twcs.active_window.l1_merge_trigger"
985        ));
986        assert!(validate_database_option(
987            "compaction.twcs.inactive_window.trigger_file_num"
988        ));
989        assert!(validate_database_option(
990            "compaction.twcs.inactive_window.l1_merge_trigger"
991        ));
992        assert!(validate_database_option(INGEST_ROWS_RATE_LIMIT_KEY));
993        assert!(validate_database_option("ingest_rows_rate_limit"));
994        assert!(!validate_database_option("foo"));
995    }
996
997    #[test]
998    fn test_parse_float_field_encoding_option() {
999        let options = TableOptions::try_from_iter([(
1000            EXPERIMENTAL_SST_FLOAT_FIELD_ENCODING,
1001            "byte_stream_split",
1002        )])
1003        .unwrap();
1004        assert_eq!(
1005            options
1006                .extra_options
1007                .get(EXPERIMENTAL_SST_FLOAT_FIELD_ENCODING),
1008            Some(&"byte_stream_split".to_string())
1009        );
1010        assert!(
1011            TableOptions::try_from_iter([(EXPERIMENTAL_SST_FLOAT_FIELD_ENCODING, "invalid",)])
1012                .is_err()
1013        );
1014    }
1015
1016    #[test]
1017    fn test_database_trigger_value_boundaries() {
1018        let maximum = usize::MAX.to_string();
1019        let overflow = format!("{maximum}0");
1020        for (key, minimum) in [
1021            (TWCS_TRIGGER_FILE_NUM, 0),
1022            (TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM, 0),
1023            (TWCS_INACTIVE_WINDOW_TRIGGER_FILE_NUM, 0),
1024            (TWCS_ACTIVE_WINDOW_L1_MERGE_TRIGGER, 2),
1025            (TWCS_INACTIVE_WINDOW_L1_MERGE_TRIGGER, 2),
1026        ] {
1027            for invalid in [
1028                None,
1029                Some(""),
1030                Some("invalid"),
1031                Some("-1"),
1032                Some(overflow.as_str()),
1033            ] {
1034                assert!(
1035                    validate_database_option_value(key, invalid).is_err(),
1036                    "{key}: {invalid:?}"
1037                );
1038            }
1039            for valid in ["2", maximum.as_str()] {
1040                assert!(
1041                    validate_database_option_value(key, Some(valid)).is_ok(),
1042                    "{key}: {valid}"
1043                );
1044            }
1045            for boundary in ["0", "1"] {
1046                assert_eq!(
1047                    validate_database_option_value(key, Some(boundary)).is_ok(),
1048                    minimum == 0,
1049                    "{key}: {boundary}"
1050                );
1051            }
1052        }
1053    }
1054
1055    #[test]
1056    fn test_database_ingest_rate_limit_value_boundaries() {
1057        for invalid in [
1058            None,
1059            Some(""),
1060            Some("abc"),
1061            Some("1000/s"),
1062            Some("-1"),
1063            Some("1.5"),
1064            Some("18446744073709551616"),
1065        ] {
1066            assert!(validate_database_option_value(INGEST_ROWS_RATE_LIMIT_KEY, invalid).is_err());
1067        }
1068        let maximum = u64::MAX.to_string();
1069        for valid in ["0", "1", maximum.as_str()] {
1070            assert!(
1071                validate_database_option_value(INGEST_ROWS_RATE_LIMIT_KEY, Some(valid)).is_ok()
1072            );
1073        }
1074    }
1075
1076    #[test]
1077    fn test_serialize_table_options() {
1078        let options = TableOptions {
1079            write_buffer_size: None,
1080            ttl: Some(Duration::from_secs(1000).into()),
1081            extra_options: HashMap::new(),
1082            skip_wal: false,
1083        };
1084        let serialized = serde_json::to_string(&options).unwrap();
1085        let deserialized: TableOptions = serde_json::from_str(&serialized).unwrap();
1086        assert_eq!(options, deserialized);
1087    }
1088
1089    #[test]
1090    fn test_convert_hashmap_between_table_options() {
1091        let options = TableOptions {
1092            write_buffer_size: Some(ReadableSize::mb(128)),
1093            ttl: Some(Duration::from_secs(1000).into()),
1094            extra_options: HashMap::new(),
1095            skip_wal: 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: None,
1103            ttl: None,
1104            extra_options: HashMap::from([(SKIP_WAL_KEY.to_string(), true.to_string())]),
1105            skip_wal: true,
1106        };
1107        let serialized_map = HashMap::from(&options);
1108        assert_eq!(
1109            Some("true"),
1110            serialized_map.get(SKIP_WAL_KEY).map(String::as_str)
1111        );
1112        let serialized = TableOptions::try_from_iter(&serialized_map).unwrap();
1113        assert_eq!(options, serialized);
1114
1115        let options = TableOptions {
1116            write_buffer_size: None,
1117            ttl: Default::default(),
1118            extra_options: HashMap::new(),
1119            skip_wal: false,
1120        };
1121        let serialized_map = HashMap::from(&options);
1122        let serialized = TableOptions::try_from_iter(&serialized_map).unwrap();
1123        assert_eq!(options, serialized);
1124
1125        let options = TableOptions {
1126            write_buffer_size: Some(ReadableSize::mb(128)),
1127            ttl: Some(Duration::from_secs(1000).into()),
1128            extra_options: HashMap::from([("a".to_string(), "A".to_string())]),
1129            skip_wal: false,
1130        };
1131        let serialized_map = HashMap::from(&options);
1132        let serialized = TableOptions::try_from_iter(&serialized_map).unwrap();
1133        assert_eq!(options, serialized);
1134
1135        let options = TableOptions {
1136            extra_options: HashMap::from([(SKIP_WAL_KEY.to_string(), false.to_string())]),
1137            skip_wal: false,
1138            ..Default::default()
1139        };
1140        let serialized_map = HashMap::from(&options);
1141        let serialized = TableOptions::try_from_iter(&serialized_map).unwrap();
1142        assert_eq!(options, serialized);
1143    }
1144
1145    #[test]
1146    fn test_table_options_normalizes_twcs_trigger_aliases() {
1147        for options in [
1148            vec![(TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM, "4")],
1149            vec![
1150                (TWCS_TRIGGER_FILE_NUM, "4"),
1151                (TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM, "4"),
1152            ],
1153        ] {
1154            let table_options = TableOptions::try_from_iter(options).unwrap();
1155            assert_eq!(
1156                HashMap::from([(TWCS_TRIGGER_FILE_NUM.to_string(), "4".to_string())]),
1157                table_options.extra_options
1158            );
1159        }
1160    }
1161
1162    #[test]
1163    fn test_table_options_rejects_conflicting_twcs_trigger_aliases() {
1164        let error = TableOptions::try_from_iter([
1165            (TWCS_TRIGGER_FILE_NUM, "4"),
1166            (TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM, "8"),
1167        ])
1168        .unwrap_err();
1169        assert_eq!(StatusCode::InvalidArguments, error.status_code());
1170        assert_eq!(
1171            "Conflicting table options: compaction.twcs.trigger_file_num=4 and compaction.twcs.active_window.trigger_file_num=8",
1172            error.to_string()
1173        );
1174    }
1175
1176    #[test]
1177    fn test_table_options_to_string() {
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: false,
1183        };
1184
1185        assert_eq!(
1186            "write_buffer_size=128.0MiB ttl=16m 40s",
1187            options.to_string()
1188        );
1189
1190        let options = TableOptions {
1191            write_buffer_size: Some(ReadableSize::mb(128)),
1192            ttl: Some(Duration::from_secs(1000).into()),
1193            extra_options: HashMap::from([("a".to_string(), "A".to_string())]),
1194            skip_wal: false,
1195        };
1196
1197        assert_eq!(
1198            "write_buffer_size=128.0MiB ttl=16m 40s a=A",
1199            options.to_string()
1200        );
1201
1202        let options = TableOptions {
1203            write_buffer_size: Some(ReadableSize::mb(128)),
1204            ttl: Some(Duration::from_secs(1000).into()),
1205            extra_options: HashMap::new(),
1206            skip_wal: true,
1207        };
1208        assert_eq!(
1209            "write_buffer_size=128.0MiB ttl=16m 40s skip_wal=true",
1210            options.to_string()
1211        );
1212
1213        let options = TableOptions {
1214            write_buffer_size: None,
1215            ttl: None,
1216            extra_options: HashMap::from([(SKIP_WAL_KEY.to_string(), "false".to_string())]),
1217            skip_wal: false,
1218        };
1219        assert_eq!("skip_wal=false", options.to_string());
1220    }
1221}
table/requests.rs

table/
requests.rs
