Skip to main content

store_api/
region_request.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
15use std::collections::HashMap;
16use std::fmt::{self, Display};
17use std::time::Duration;
18
19use api::helper::{ColumnDataTypeWrapper, from_pb_time_ranges, from_pb_time_unit};
20use api::v1::add_column_location::LocationType;
21use api::v1::column_def::{
22    as_fulltext_option_analyzer, as_fulltext_option_backend, as_skipping_index_type,
23};
24use api::v1::region::bulk_insert_request::Body;
25use api::v1::region::{
26    AlterRequest, AlterRequests, BuildIndexRequest, BulkInsertRequest,
27    CleanUpRequest as PbCleanUpRequest, CloseRequest, CompactRequest, CreateRequest,
28    CreateRequests, DeleteRequests, DropRequest, DropRequests, FlushRequest, InsertRequests,
29    OpenRequest, TruncateRequest, alter_request, build_index_request, compact_request,
30    region_request, truncate_request,
31};
32use api::v1::{
33    self, Analyzer, ArrowIpc, FulltextBackend as PbFulltextBackend, Option as PbOption, Rows,
34    SemanticType, SkippingIndexType as PbSkippingIndexType, WriteHint,
35};
36use arrow_schema::extension::ExtensionType;
37pub use common_base::AffectedRows;
38use common_base::readable_size::ReadableSize;
39use common_grpc::flight::FlightDecoder;
40use common_recordbatch::DfRecordBatch;
41use common_time::range::TimestampRange;
42use common_time::{TimeToLive, Timestamp};
43use datatypes::error::time_index_not_widening_error;
44use datatypes::extension::json::Json2ExtensionType;
45use datatypes::json::{JsonSettings, JsonTypeHint};
46use datatypes::prelude::ConcreteDataType;
47use datatypes::schema::{FulltextOptions, SkippingIndexOptions};
48use num_enum::TryFromPrimitive;
49use serde::{Deserialize, Deserializer, Serialize, Serializer};
50use snafu::{OptionExt, ResultExt, ensure};
51use strum::{AsRefStr, IntoStaticStr};
52
53use crate::logstore::entry;
54use crate::metadata::{
55    ColumnMetadata, ConvertTimeRangesSnafu, DecodeProtoSnafu, FlightCodecSnafu,
56    InvalidIndexOptionSnafu, InvalidRawRegionRequestSnafu, InvalidRegionRequestSnafu,
57    InvalidSetRegionOptionRequestSnafu, InvalidUnsetRegionOptionRequestSnafu, MetadataError,
58    RegionMetadata, Result, UnexpectedSnafu,
59};
60use crate::metric_engine_consts::PHYSICAL_TABLE_METADATA_KEY;
61use crate::metrics;
62use crate::mito_engine_options::{
63    APPEND_MODE_KEY, AUTO_FLUSH_INTERVAL_KEY, MAX_ROW_GROUP_ROW_COUNT,
64    MAX_ROW_GROUP_ROW_COUNT_LIMIT, PRESERVE_ROW_SEQUENCE, SKIP_WAL_KEY, SST_FORMAT_KEY, TTL_KEY,
65    TWCS_ACTIVE_WINDOW_L1_MERGE_TRIGGER, TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM,
66    TWCS_INACTIVE_WINDOW_L1_MERGE_TRIGGER, TWCS_INACTIVE_WINDOW_TRIGGER_FILE_NUM,
67    TWCS_MAX_OUTPUT_FILE_SIZE, TWCS_TIME_WINDOW, TWCS_TRIGGER_FILE_NUM, WRITE_BUFFER_SIZE_KEY,
68};
69use crate::path_utils::table_dir;
70use crate::storage::{ColumnId, RegionId};
71
72/// The type of path to generate.
73#[derive(Debug, Clone, Copy, PartialEq, TryFromPrimitive)]
74#[repr(u8)]
75pub enum PathType {
76    /// A bare path - the original path of an engine.
77    ///
78    /// The path prefix is `{table_dir}/{table_id}_{region_sequence}/`.
79    Bare,
80    /// A path for the data region of a metric engine table.
81    ///
82    /// The path prefix is `{table_dir}/{table_id}_{region_sequence}/data/`.
83    Data,
84    /// A path for the metadata region of a metric engine table.
85    ///
86    /// The path prefix is `{table_dir}/{table_id}_{region_sequence}/metadata/`.
87    Metadata,
88}
89
90#[derive(Debug, IntoStaticStr)]
91pub enum BatchRegionDdlRequest {
92    Create(Vec<(RegionId, RegionCreateRequest)>),
93    Drop(Vec<(RegionId, RegionDropRequest)>),
94    Alter(Vec<(RegionId, RegionAlterRequest)>),
95}
96
97impl BatchRegionDdlRequest {
98    /// Converts [Body](region_request::Body) to [`BatchRegionDdlRequest`].
99    pub fn try_from_request_body(body: region_request::Body) -> Result<Option<Self>> {
100        match body {
101            region_request::Body::Creates(creates) => {
102                let requests = creates
103                    .requests
104                    .into_iter()
105                    .map(parse_region_create)
106                    .collect::<Result<Vec<_>>>()?;
107                Ok(Some(Self::Create(requests)))
108            }
109            region_request::Body::Drops(drops) => {
110                let requests = drops
111                    .requests
112                    .into_iter()
113                    .map(parse_region_drop)
114                    .collect::<Result<Vec<_>>>()?;
115                Ok(Some(Self::Drop(requests)))
116            }
117            region_request::Body::Alters(alters) => {
118                let requests = alters
119                    .requests
120                    .into_iter()
121                    .map(parse_region_alter)
122                    .collect::<Result<Vec<_>>>()?;
123                Ok(Some(Self::Alter(requests)))
124            }
125            _ => Ok(None),
126        }
127    }
128
129    pub fn request_type(&self) -> &'static str {
130        self.into()
131    }
132
133    pub fn into_region_requests(self) -> Vec<(RegionId, RegionRequest)> {
134        match self {
135            Self::Create(requests) => requests
136                .into_iter()
137                .map(|(region_id, request)| (region_id, RegionRequest::Create(request)))
138                .collect(),
139            Self::Drop(requests) => requests
140                .into_iter()
141                .map(|(region_id, request)| (region_id, RegionRequest::Drop(request)))
142                .collect(),
143            Self::Alter(requests) => requests
144                .into_iter()
145                .map(|(region_id, request)| (region_id, RegionRequest::Alter(request)))
146                .collect(),
147        }
148    }
149}
150
151#[derive(Debug, IntoStaticStr)]
152pub enum RegionRequest {
153    Put(RegionPutRequest),
154    Delete(RegionDeleteRequest),
155    Create(RegionCreateRequest),
156    Drop(RegionDropRequest),
157    Open(RegionOpenRequest),
158    CleanUp(RegionCleanUpRequest),
159    Close(RegionCloseRequest),
160    Alter(RegionAlterRequest),
161    Flush(RegionFlushRequest),
162    Compact(RegionCompactRequest),
163    BuildIndex(RegionBuildIndexRequest),
164    Truncate(RegionTruncateRequest),
165    Catchup(RegionCatchupRequest),
166    BulkInserts(RegionBulkInsertsRequest),
167    EnterStaging(EnterStagingRequest),
168    ApplyStagingManifest(ApplyStagingManifestRequest),
169}
170
171impl RegionRequest {
172    /// Convert [Body](region_request::Body) to a group of [RegionRequest] with region id.
173    /// Inserts/Deletes request might become multiple requests. Others are one-to-one.
174    pub fn try_from_request_body(body: region_request::Body) -> Result<Vec<(RegionId, Self)>> {
175        match body {
176            region_request::Body::Inserts(inserts) => make_region_puts(inserts),
177            region_request::Body::Deletes(deletes) => make_region_deletes(deletes),
178            region_request::Body::Create(create) => make_region_create(create),
179            region_request::Body::Drop(drop) => make_region_drop(drop),
180            region_request::Body::Open(open) => make_region_open(open),
181            region_request::Body::CleanUp(clean_up) => make_region_clean_up(clean_up),
182            region_request::Body::Close(close) => make_region_close(close),
183            region_request::Body::Alter(alter) => make_region_alter(alter),
184            region_request::Body::Flush(flush) => make_region_flush(flush),
185            region_request::Body::Compact(compact) => make_region_compact(compact),
186            region_request::Body::BuildIndex(index) => make_region_build_index(index),
187            region_request::Body::Truncate(truncate) => make_region_truncate(truncate),
188            region_request::Body::Creates(creates) => make_region_creates(creates),
189            region_request::Body::Drops(drops) => make_region_drops(drops),
190            region_request::Body::Alters(alters) => make_region_alters(alters),
191            region_request::Body::BulkInsert(bulk) => make_region_bulk_inserts(bulk),
192            region_request::Body::Sync(_) => UnexpectedSnafu {
193                reason: "Sync request should be handled separately by RegionServer",
194            }
195            .fail(),
196            region_request::Body::ListMetadata(_) => UnexpectedSnafu {
197                reason: "ListMetadata request should be handled separately by RegionServer",
198            }
199            .fail(),
200            region_request::Body::RemoteDynFilter(_) => UnexpectedSnafu {
201                reason: "RemoteDynFilter request should be handled separately by RegionServer",
202            }
203            .fail(),
204            region_request::Body::ApplyStagingManifest(apply) => {
205                make_region_apply_staging_manifest(apply)
206            }
207        }
208    }
209
210    /// Returns the type name of the request.
211    pub fn request_type(&self) -> &'static str {
212        self.into()
213    }
214}
215
216/// Converts one insert, treating absent rows as a no-op like vector inserts.
217pub fn make_region_put(
218    request: api::v1::region::InsertRequest,
219) -> Option<(RegionId, RegionPutRequest)> {
220    request.rows.map(|rows| {
221        (
222            request.region_id.into(),
223            RegionPutRequest {
224                rows,
225                hint: None,
226                skip_wal: request.skip_wal,
227                partition_expr_version: request.partition_expr_version.map(|version| version.value),
228            },
229        )
230    })
231}
232
233fn make_region_puts(inserts: InsertRequests) -> Result<Vec<(RegionId, RegionRequest)>> {
234    Ok(inserts
235        .requests
236        .into_iter()
237        .filter_map(make_region_put)
238        .map(|(region_id, request)| (region_id, RegionRequest::Put(request)))
239        .collect())
240}
241
242/// Converts one delete, treating absent rows as a no-op like vector deletes.
243pub fn make_region_delete(
244    request: api::v1::region::DeleteRequest,
245) -> Option<(RegionId, RegionDeleteRequest)> {
246    request.rows.map(|rows| {
247        (
248            request.region_id.into(),
249            RegionDeleteRequest {
250                rows,
251                hint: None,
252                partition_expr_version: request.partition_expr_version.map(|version| version.value),
253            },
254        )
255    })
256}
257
258fn make_region_deletes(deletes: DeleteRequests) -> Result<Vec<(RegionId, RegionRequest)>> {
259    Ok(deletes
260        .requests
261        .into_iter()
262        .filter_map(make_region_delete)
263        .map(|(region_id, request)| (region_id, RegionRequest::Delete(request)))
264        .collect())
265}
266
267fn parse_region_create(create: CreateRequest) -> Result<(RegionId, RegionCreateRequest)> {
268    let column_metadatas = create
269        .column_defs
270        .into_iter()
271        .map(ColumnMetadata::try_from_column_def)
272        .collect::<Result<Vec<_>>>()?;
273    let region_id = RegionId::from(create.region_id);
274    let table_dir = table_dir(&create.path, region_id.table_id());
275    let partition_expr_json = create.partition.as_ref().map(|p| p.expression.clone());
276    Ok((
277        region_id,
278        RegionCreateRequest {
279            engine: create.engine,
280            column_metadatas,
281            primary_key: create.primary_key,
282            options: create.options,
283            table_dir,
284            path_type: PathType::Bare,
285            partition_expr_json,
286            requirements: create
287                .requirements
288                .map(RegionRequirements::from)
289                .unwrap_or_default(),
290        },
291    ))
292}
293
294fn make_region_create(create: CreateRequest) -> Result<Vec<(RegionId, RegionRequest)>> {
295    let (region_id, request) = parse_region_create(create)?;
296    Ok(vec![(region_id, RegionRequest::Create(request))])
297}
298
299fn make_region_creates(creates: CreateRequests) -> Result<Vec<(RegionId, RegionRequest)>> {
300    let mut requests = Vec::with_capacity(creates.requests.len());
301    for create in creates.requests {
302        requests.extend(make_region_create(create)?);
303    }
304    Ok(requests)
305}
306
307fn parse_region_drop(drop: DropRequest) -> Result<(RegionId, RegionDropRequest)> {
308    let region_id = drop.region_id.into();
309    Ok((
310        region_id,
311        RegionDropRequest {
312            fast_path: drop.fast_path,
313            force: drop.force,
314            partial_drop: drop.partial_drop,
315        },
316    ))
317}
318
319fn make_region_drop(drop: DropRequest) -> Result<Vec<(RegionId, RegionRequest)>> {
320    let (region_id, request) = parse_region_drop(drop)?;
321    Ok(vec![(region_id, RegionRequest::Drop(request))])
322}
323
324fn make_region_drops(drops: DropRequests) -> Result<Vec<(RegionId, RegionRequest)>> {
325    let mut requests = Vec::with_capacity(drops.requests.len());
326    for drop in drops.requests {
327        requests.extend(make_region_drop(drop)?);
328    }
329    Ok(requests)
330}
331
332fn make_region_open(open: OpenRequest) -> Result<Vec<(RegionId, RegionRequest)>> {
333    let region_id = RegionId::from(open.region_id);
334    let table_dir = table_dir(&open.path, region_id.table_id());
335    Ok(vec![(
336        region_id,
337        RegionRequest::Open(RegionOpenRequest {
338            engine: open.engine,
339            table_dir,
340            path_type: PathType::Bare,
341            options: open.options,
342            skip_wal_replay: false,
343            checkpoint: None,
344            requirements: Default::default(),
345        }),
346    )])
347}
348
349fn make_region_clean_up(clean_up: PbCleanUpRequest) -> Result<Vec<(RegionId, RegionRequest)>> {
350    let region_id = RegionId::from(clean_up.region_id);
351    let table_dir = table_dir(&clean_up.path, region_id.table_id());
352    Ok(vec![(
353        region_id,
354        RegionRequest::CleanUp(RegionCleanUpRequest {
355            engine: clean_up.engine,
356            table_dir,
357            path_type: PathType::Bare,
358            options: clean_up.options,
359        }),
360    )])
361}
362
363fn make_region_close(close: CloseRequest) -> Result<Vec<(RegionId, RegionRequest)>> {
364    let region_id = close.region_id.into();
365    Ok(vec![(
366        region_id,
367        RegionRequest::Close(RegionCloseRequest {
368            flush_on_close: close.flush_on_close,
369        }),
370    )])
371}
372
373fn parse_region_alter(alter: AlterRequest) -> Result<(RegionId, RegionAlterRequest)> {
374    let region_id = alter.region_id.into();
375    let request = RegionAlterRequest::try_from(alter)?;
376    Ok((region_id, request))
377}
378
379fn make_region_alter(alter: AlterRequest) -> Result<Vec<(RegionId, RegionRequest)>> {
380    let (region_id, request) = parse_region_alter(alter)?;
381    Ok(vec![(region_id, RegionRequest::Alter(request))])
382}
383
384fn make_region_alters(alters: AlterRequests) -> Result<Vec<(RegionId, RegionRequest)>> {
385    let mut requests = Vec::with_capacity(alters.requests.len());
386    for alter in alters.requests {
387        requests.extend(make_region_alter(alter)?);
388    }
389    Ok(requests)
390}
391
392fn make_region_flush(flush: FlushRequest) -> Result<Vec<(RegionId, RegionRequest)>> {
393    let region_id = flush.region_id.into();
394    Ok(vec![(
395        region_id,
396        RegionRequest::Flush(RegionFlushRequest::default()),
397    )])
398}
399
400fn make_region_compact(compact: CompactRequest) -> Result<Vec<(RegionId, RegionRequest)>> {
401    let region_id = compact.region_id.into();
402    let options = compact
403        .options
404        .unwrap_or(compact_request::Options::Regular(Default::default()));
405    // Convert parallelism: a value of 0 indicates no specific parallelism requested (None)
406    let parallelism = if compact.parallelism == 0 {
407        None
408    } else {
409        Some(compact.parallelism)
410    };
411    let time_range = compact
412        .time_range
413        .map(|range| {
414            let time_unit = v1::TimeUnit::try_from(range.time_unit).map_err(|_| {
415                InvalidRegionRequestSnafu {
416                    region_id,
417                    err: format!("invalid compaction time unit: {}", range.time_unit),
418                }
419                .build()
420            })?;
421            let time_unit = from_pb_time_unit(time_unit);
422            let start = Timestamp::new(range.start, time_unit);
423            let end = Timestamp::new(range.end, time_unit);
424            ensure!(
425                start < end,
426                InvalidRegionRequestSnafu {
427                    region_id,
428                    err: format!(
429                        "compaction start time must be earlier than end time: {} >= {}",
430                        range.start, range.end
431                    ),
432                }
433            );
434            TimestampRange::new(start, end).with_context(|| InvalidRegionRequestSnafu {
435                region_id,
436                err: "invalid compaction time range".to_string(),
437            })
438        })
439        .transpose()?;
440    Ok(vec![(
441        region_id,
442        RegionRequest::Compact(RegionCompactRequest {
443            options,
444            parallelism,
445            time_range,
446        }),
447    )])
448}
449
450fn make_region_build_index(index: BuildIndexRequest) -> Result<Vec<(RegionId, RegionRequest)>> {
451    let region_id = index.region_id.into();
452    Ok(vec![(
453        region_id,
454        RegionRequest::BuildIndex(RegionBuildIndexRequest {
455            options: index.options,
456        }),
457    )])
458}
459
460fn make_region_truncate(truncate: TruncateRequest) -> Result<Vec<(RegionId, RegionRequest)>> {
461    let region_id = truncate.region_id.into();
462    match truncate.kind {
463        None => InvalidRawRegionRequestSnafu {
464            err: "missing kind in TruncateRequest".to_string(),
465        }
466        .fail(),
467        Some(truncate_request::Kind::All(_)) => Ok(vec![(
468            region_id,
469            RegionRequest::Truncate(RegionTruncateRequest::All),
470        )]),
471        Some(truncate_request::Kind::TimeRanges(time_ranges)) => {
472            let time_ranges = from_pb_time_ranges(time_ranges).context(ConvertTimeRangesSnafu)?;
473
474            Ok(vec![(
475                region_id,
476                RegionRequest::Truncate(RegionTruncateRequest::ByTimeRanges { time_ranges }),
477            )])
478        }
479        Some(truncate_request::Kind::Unflushed(_)) => Ok(vec![(
480            region_id,
481            RegionRequest::Truncate(RegionTruncateRequest::Unflushed),
482        )]),
483    }
484}
485
486/// Convert [BulkInsertRequest] to [RegionRequest] and group by [RegionId].
487fn make_region_bulk_inserts(request: BulkInsertRequest) -> Result<Vec<(RegionId, RegionRequest)>> {
488    let region_id = request.region_id.into();
489    let skip_wal = request.skip_wal;
490    let partition_expr_version = request.partition_expr_version.map(|v| v.value);
491    let aligned_schema_version = request.aligned_schema_version.map(|v| v.schema_version);
492    let Some(Body::ArrowIpc(request)) = request.body else {
493        return Ok(vec![]);
494    };
495
496    let decoder_timer = metrics::CONVERT_REGION_BULK_REQUEST
497        .with_label_values(&["decode"])
498        .start_timer();
499    let mut decoder =
500        FlightDecoder::try_from_schema_bytes(&request.schema).context(FlightCodecSnafu)?;
501    let payload = decoder
502        .try_decode_record_batch(&request.data_header, &request.payload)
503        .context(FlightCodecSnafu)?;
504    decoder_timer.observe_duration();
505    Ok(vec![(
506        region_id,
507        RegionRequest::BulkInserts(RegionBulkInsertsRequest {
508            region_id,
509            payload,
510            raw_data: request,
511            skip_wal,
512            partition_expr_version,
513            aligned_schema_version,
514        }),
515    )])
516}
517
518fn make_region_apply_staging_manifest(
519    api::v1::region::ApplyStagingManifestRequest {
520        region_id,
521        partition_expr,
522        central_region_id,
523        manifest_path,
524    }: api::v1::region::ApplyStagingManifestRequest,
525) -> Result<Vec<(RegionId, RegionRequest)>> {
526    let region_id = region_id.into();
527    Ok(vec![(
528        region_id,
529        RegionRequest::ApplyStagingManifest(ApplyStagingManifestRequest {
530            partition_expr,
531            central_region_id: central_region_id.into(),
532            manifest_path,
533        }),
534    )])
535}
536
537/// Request to put data into a region.
538#[derive(Debug)]
539pub struct RegionPutRequest {
540    /// Rows to put.
541    pub rows: Rows,
542    /// Write hint.
543    pub hint: Option<WriteHint>,
544    /// Skip WAL for this insert without changing region options.
545    /// Metadata writes must not inherit this option from user inserts.
546    pub skip_wal: bool,
547    /// Partition expression version for the region.
548    pub partition_expr_version: Option<u64>,
549}
550
551/// Request to delete data from a region.
552#[derive(Debug)]
553pub struct RegionDeleteRequest {
554    /// Keys to rows to delete.
555    ///
556    /// Each row only contains primary key columns and a time index column.
557    pub rows: Rows,
558    /// Write hint.
559    pub hint: Option<WriteHint>,
560    /// Partition expression version for the region.
561    pub partition_expr_version: Option<u64>,
562}
563
564#[derive(Debug, Clone)]
565pub struct RegionCreateRequest {
566    /// Region engine name
567    pub engine: String,
568    /// Columns in this region.
569    pub column_metadatas: Vec<ColumnMetadata>,
570    /// Columns in the primary key.
571    pub primary_key: Vec<ColumnId>,
572    /// Options of the created region.
573    pub options: HashMap<String, String>,
574    /// Directory for table's data home. Usually is composed by catalog and table id
575    pub table_dir: String,
576    /// Path type for generating paths
577    pub path_type: PathType,
578    /// Partition expression JSON from table metadata. Set to empty string for a region without partition.
579    /// `Option` to keep compatibility with old clients.
580    pub partition_expr_json: Option<String>,
581    /// Requirements for creating the region.
582    pub requirements: RegionRequirements,
583}
584
585impl RegionCreateRequest {
586    /// Checks whether the request is valid, returns an error if it is invalid.
587    pub fn validate(&self) -> Result<()> {
588        // time index must exist
589        ensure!(
590            self.column_metadatas
591                .iter()
592                .any(|x| x.semantic_type == SemanticType::Timestamp),
593            InvalidRegionRequestSnafu {
594                region_id: RegionId::new(0, 0),
595                err: "missing timestamp column in create region request".to_string(),
596            }
597        );
598
599        // build column id to indices
600        let mut column_id_to_indices = HashMap::with_capacity(self.column_metadatas.len());
601        for (i, c) in self.column_metadatas.iter().enumerate() {
602            if let Some(previous) = column_id_to_indices.insert(c.column_id, i) {
603                return InvalidRegionRequestSnafu {
604                    region_id: RegionId::new(0, 0),
605                    err: format!(
606                        "duplicate column id {} (at position {} and {}) in create region request",
607                        c.column_id, previous, i
608                    ),
609                }
610                .fail();
611            }
612        }
613
614        // primary key must exist
615        for column_id in &self.primary_key {
616            ensure!(
617                column_id_to_indices.contains_key(column_id),
618                InvalidRegionRequestSnafu {
619                    region_id: RegionId::new(0, 0),
620                    err: format!(
621                        "missing primary key column {} in create region request",
622                        column_id
623                    ),
624                }
625            );
626        }
627
628        Ok(())
629    }
630
631    /// Returns true when the region belongs to the metric engine's physical table.
632    pub fn is_physical_table(&self) -> bool {
633        self.options.contains_key(PHYSICAL_TABLE_METADATA_KEY)
634    }
635}
636
637#[derive(Debug, Clone)]
638pub struct RegionDropRequest {
639    /// Enables fast-path drop optimizations for logical regions.
640    /// Only applicable to the Metric Engine; ignored by others.
641    pub fast_path: bool,
642
643    /// Forces the drop of a physical region and all its associated logical regions.
644    /// Only relevant for physical regions managed by the Metric Engine.
645    pub force: bool,
646
647    /// If true, indicates that only a portion of the region is being dropped, and files may still be referenced by other regions.
648    /// This is used to prevent deletion of files that are still in use by other regions.
649    pub partial_drop: bool,
650}
651
652/// Requirements for a region request.
653#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
654#[serde(default)]
655pub struct RegionRequirements {
656    /// Whether the region data must be backed by object storage.
657    pub object_storage: bool,
658}
659
660impl RegionRequirements {
661    /// Returns empty requirements.
662    pub fn empty() -> Self {
663        Self::default()
664    }
665
666    /// Returns requirements for object storage.
667    pub fn object_storage() -> Self {
668        Self {
669            object_storage: true,
670        }
671    }
672}
673
674impl From<api::v1::region::RegionRequirements> for RegionRequirements {
675    fn from(value: api::v1::region::RegionRequirements) -> Self {
676        Self {
677            object_storage: value.object_storage,
678        }
679    }
680}
681
682impl From<RegionRequirements> for api::v1::region::RegionRequirements {
683    fn from(value: RegionRequirements) -> Self {
684        Self {
685            object_storage: value.object_storage,
686        }
687    }
688}
689
690/// Open region request.
691#[derive(Debug, Clone)]
692pub struct RegionOpenRequest {
693    /// Region engine name
694    pub engine: String,
695    /// Directory for table's data home. Usually is composed by catalog and table id
696    pub table_dir: String,
697    /// Path type for generating paths
698    pub path_type: PathType,
699    /// Options of the opened region.
700    pub options: HashMap<String, String>,
701    /// To skip replaying the WAL.
702    pub skip_wal_replay: bool,
703    /// Replay checkpoint.
704    pub checkpoint: Option<ReplayCheckpoint>,
705    /// Requirements for opening the region.
706    pub requirements: RegionRequirements,
707}
708
709#[derive(Debug, Clone, Copy, PartialEq, Eq)]
710pub struct ReplayCheckpoint {
711    pub entry_id: u64,
712    pub metadata_entry_id: Option<u64>,
713}
714
715impl RegionOpenRequest {
716    /// Returns true when the region belongs to the metric engine's physical table.
717    pub fn is_physical_table(&self) -> bool {
718        self.options.contains_key(PHYSICAL_TABLE_METADATA_KEY)
719    }
720}
721
722/// Offline region cleanup request.
723#[derive(Debug, Clone)]
724pub struct RegionCleanUpRequest {
725    /// Region engine name
726    pub engine: String,
727    /// Directory for table's data home. Usually is composed by catalog and table id
728    pub table_dir: String,
729    /// Path type for generating paths
730    pub path_type: PathType,
731    /// Options of the cleaned region.
732    pub options: HashMap<String, String>,
733}
734
735impl RegionCleanUpRequest {
736    /// Returns true when the region belongs to the metric engine's physical table.
737    pub fn is_physical_table(&self) -> bool {
738        self.options.contains_key(PHYSICAL_TABLE_METADATA_KEY)
739    }
740}
741
742/// Close region request.
743#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
744pub struct RegionCloseRequest {
745    /// Whether to flush the region before closing it.
746    pub flush_on_close: bool,
747}
748
749/// Alter metadata of a region.
750#[derive(Debug, PartialEq, Eq, Clone)]
751pub struct RegionAlterRequest {
752    /// Kind of alteration to do.
753    pub kind: AlterKind,
754}
755
756impl RegionAlterRequest {
757    /// Checks whether the request is valid, returns an error if it is invalid.
758    pub fn validate(&self, metadata: &RegionMetadata) -> Result<()> {
759        self.kind.validate(metadata)?;
760
761        Ok(())
762    }
763
764    /// Returns true if we need to apply the request to the region.
765    ///
766    /// The `request` should be valid.
767    pub fn need_alter(&self, metadata: &RegionMetadata) -> bool {
768        debug_assert!(self.validate(metadata).is_ok());
769        self.kind.need_alter(metadata)
770    }
771}
772
773impl TryFrom<AlterRequest> for RegionAlterRequest {
774    type Error = MetadataError;
775
776    fn try_from(value: AlterRequest) -> Result<Self> {
777        let kind = value.kind.context(InvalidRawRegionRequestSnafu {
778            err: "missing kind in AlterRequest",
779        })?;
780
781        let kind = AlterKind::try_from(kind)?;
782        Ok(RegionAlterRequest { kind })
783    }
784}
785
786/// Kind of the alteration.
787#[derive(Debug, PartialEq, Eq, Clone, AsRefStr)]
788pub enum AlterKind {
789    /// Add columns to the region.
790    AddColumns {
791        /// Columns to add.
792        columns: Vec<AddColumn>,
793    },
794    /// Drop columns from the region, only fields are allowed to drop.
795    DropColumns {
796        /// Name of columns to drop.
797        names: Vec<String>,
798    },
799    /// Change columns datatype of the region. Field columns can change to any
800    /// Arrow-castable type; the time index column only supports widening its
801    /// timestamp unit (e.g. `TimestampMillisecond -> TimestampMicrosecond`),
802    /// which is lossless for values that fit the target unit's `i64` range.
803    ModifyColumnTypes {
804        /// Columns to change.
805        columns: Vec<ModifyColumnType>,
806    },
807    /// Set JSON2 settings of a region column.
808    SetJsonSettings {
809        /// Column name.
810        column_name: String,
811        /// Target JSON2 settings.
812        settings: JsonSettings,
813    },
814    /// Set region options.
815    SetRegionOptions { options: Vec<SetRegionOption> },
816    /// Unset region options.
817    UnsetRegionOptions { keys: Vec<UnsetRegionOption> },
818    /// Set index options.
819    SetIndexes { options: Vec<SetIndexOption> },
820    /// Unset index options.
821    UnsetIndexes { options: Vec<UnsetIndexOption> },
822    /// Drop column default value.
823    DropDefaults {
824        /// Name of columns to drop.
825        names: Vec<String>,
826    },
827    /// Set column default value.
828    SetDefaults {
829        /// Columns to change.
830        columns: Vec<SetDefault>,
831    },
832    /// Sync column metadatas.
833    SyncColumns {
834        column_metadatas: Vec<ColumnMetadata>,
835    },
836}
837#[derive(Debug, PartialEq, Eq, Clone)]
838pub struct SetDefault {
839    pub name: String,
840    pub default_constraint: Vec<u8>,
841}
842
843#[derive(Debug, PartialEq, Eq, Clone)]
844pub enum SetIndexOption {
845    Fulltext {
846        column_name: String,
847        options: FulltextOptions,
848    },
849    Inverted {
850        column_name: String,
851    },
852    Skipping {
853        column_name: String,
854        options: SkippingIndexOptions,
855    },
856}
857
858impl SetIndexOption {
859    /// Returns the column name of the index option.
860    pub fn column_name(&self) -> &String {
861        match self {
862            SetIndexOption::Fulltext { column_name, .. } => column_name,
863            SetIndexOption::Inverted { column_name } => column_name,
864            SetIndexOption::Skipping { column_name, .. } => column_name,
865        }
866    }
867
868    /// Returns true if the index option is fulltext.
869    pub fn is_fulltext(&self) -> bool {
870        match self {
871            SetIndexOption::Fulltext { .. } => true,
872            SetIndexOption::Inverted { .. } => false,
873            SetIndexOption::Skipping { .. } => false,
874        }
875    }
876}
877
878impl TryFrom<v1::SetIndex> for SetIndexOption {
879    type Error = MetadataError;
880
881    fn try_from(value: v1::SetIndex) -> Result<Self> {
882        let option = value.options.context(InvalidRawRegionRequestSnafu {
883            err: "missing options in SetIndex",
884        })?;
885
886        let opt = match option {
887            v1::set_index::Options::Fulltext(x) => SetIndexOption::Fulltext {
888                column_name: x.column_name.clone(),
889                options: FulltextOptions::new(
890                    x.enable,
891                    as_fulltext_option_analyzer(
892                        Analyzer::try_from(x.analyzer).context(DecodeProtoSnafu)?,
893                    ),
894                    x.case_sensitive,
895                    as_fulltext_option_backend(
896                        PbFulltextBackend::try_from(x.backend).context(DecodeProtoSnafu)?,
897                    ),
898                    x.granularity as u32,
899                    x.false_positive_rate,
900                )
901                .context(InvalidIndexOptionSnafu)?,
902            },
903            v1::set_index::Options::Inverted(i) => SetIndexOption::Inverted {
904                column_name: i.column_name,
905            },
906            v1::set_index::Options::Skipping(s) => SetIndexOption::Skipping {
907                column_name: s.column_name,
908                options: SkippingIndexOptions::new(
909                    s.granularity as u32,
910                    s.false_positive_rate,
911                    as_skipping_index_type(
912                        PbSkippingIndexType::try_from(s.skipping_index_type)
913                            .context(DecodeProtoSnafu)?,
914                    ),
915                )
916                .context(InvalidIndexOptionSnafu)?,
917            },
918        };
919
920        Ok(opt)
921    }
922}
923
924#[derive(Debug, PartialEq, Eq, Clone)]
925pub enum UnsetIndexOption {
926    Fulltext { column_name: String },
927    Inverted { column_name: String },
928    Skipping { column_name: String },
929}
930
931impl UnsetIndexOption {
932    pub fn column_name(&self) -> &String {
933        match self {
934            UnsetIndexOption::Fulltext { column_name } => column_name,
935            UnsetIndexOption::Inverted { column_name } => column_name,
936            UnsetIndexOption::Skipping { column_name } => column_name,
937        }
938    }
939
940    pub fn is_fulltext(&self) -> bool {
941        match self {
942            UnsetIndexOption::Fulltext { .. } => true,
943            UnsetIndexOption::Inverted { .. } => false,
944            UnsetIndexOption::Skipping { .. } => false,
945        }
946    }
947}
948
949impl TryFrom<v1::UnsetIndex> for UnsetIndexOption {
950    type Error = MetadataError;
951
952    fn try_from(value: v1::UnsetIndex) -> Result<Self> {
953        let option = value.options.context(InvalidRawRegionRequestSnafu {
954            err: "missing options in UnsetIndex",
955        })?;
956
957        let opt = match option {
958            v1::unset_index::Options::Fulltext(f) => UnsetIndexOption::Fulltext {
959                column_name: f.column_name,
960            },
961            v1::unset_index::Options::Inverted(i) => UnsetIndexOption::Inverted {
962                column_name: i.column_name,
963            },
964            v1::unset_index::Options::Skipping(s) => UnsetIndexOption::Skipping {
965                column_name: s.column_name,
966            },
967        };
968
969        Ok(opt)
970    }
971}
972
973impl AlterKind {
974    /// Returns an error if the alter kind is invalid.
975    ///
976    /// It allows adding column if not exists and dropping column if exists.
977    pub fn validate(&self, metadata: &RegionMetadata) -> Result<()> {
978        match self {
979            AlterKind::AddColumns { columns } => {
980                for col_to_add in columns {
981                    col_to_add.validate(metadata)?;
982                }
983            }
984            AlterKind::DropColumns { names } => {
985                for name in names {
986                    Self::validate_column_to_drop(name, metadata)?;
987                }
988            }
989            AlterKind::ModifyColumnTypes { columns } => {
990                for col_to_change in columns {
991                    col_to_change.validate(metadata)?;
992                }
993            }
994            AlterKind::SetJsonSettings { column_name, .. } => {
995                Self::validate_set_json_settings(column_name, metadata)?
996            }
997            AlterKind::SetRegionOptions { .. } => {}
998            AlterKind::UnsetRegionOptions { .. } => {}
999            AlterKind::SetIndexes { options } => {
1000                for option in options {
1001                    Self::validate_column_alter_index_option(
1002                        option.column_name(),
1003                        metadata,
1004                        option.is_fulltext(),
1005                    )?;
1006                }
1007            }
1008            AlterKind::UnsetIndexes { options } => {
1009                for option in options {
1010                    Self::validate_column_alter_index_option(
1011                        option.column_name(),
1012                        metadata,
1013                        option.is_fulltext(),
1014                    )?;
1015                }
1016            }
1017            AlterKind::DropDefaults { names } => {
1018                names
1019                    .iter()
1020                    .try_for_each(|name| Self::validate_column_existence(name, metadata))?;
1021            }
1022            AlterKind::SetDefaults { columns } => {
1023                columns
1024                    .iter()
1025                    .try_for_each(|col| Self::validate_column_existence(&col.name, metadata))?;
1026            }
1027            AlterKind::SyncColumns { column_metadatas } => {
1028                let new_primary_keys = column_metadatas
1029                    .iter()
1030                    .filter(|c| c.semantic_type == SemanticType::Tag)
1031                    .map(|c| (c.column_schema.name.as_str(), c.column_id))
1032                    .collect::<HashMap<_, _>>();
1033
1034                let old_primary_keys = metadata
1035                    .column_metadatas
1036                    .iter()
1037                    .filter(|c| c.semantic_type == SemanticType::Tag)
1038                    .map(|c| (c.column_schema.name.as_str(), c.column_id));
1039
1040                for (name, id) in old_primary_keys {
1041                    let primary_key =
1042                        new_primary_keys
1043                            .get(name)
1044                            .with_context(|| InvalidRegionRequestSnafu {
1045                                region_id: metadata.region_id,
1046                                err: format!("column {} is not a primary key", name),
1047                            })?;
1048
1049                    ensure!(
1050                        *primary_key == id,
1051                        InvalidRegionRequestSnafu {
1052                            region_id: metadata.region_id,
1053                            err: format!(
1054                                "column with same name {} has different id, existing: {}, got: {}",
1055                                name, id, primary_key
1056                            ),
1057                        }
1058                    );
1059                }
1060
1061                let new_ts_column = column_metadatas
1062                    .iter()
1063                    .find(|c| c.semantic_type == SemanticType::Timestamp)
1064                    .map(|c| (c.column_schema.name.as_str(), c.column_id))
1065                    .context(InvalidRegionRequestSnafu {
1066                        region_id: metadata.region_id,
1067                        err: "timestamp column not found",
1068                    })?;
1069
1070                // Safety: timestamp column must exist.
1071                let old_ts_column = metadata
1072                    .column_metadatas
1073                    .iter()
1074                    .find(|c| c.semantic_type == SemanticType::Timestamp)
1075                    .map(|c| (c.column_schema.name.as_str(), c.column_id))
1076                    .unwrap();
1077
1078                ensure!(
1079                    new_ts_column == old_ts_column,
1080                    InvalidRegionRequestSnafu {
1081                        region_id: metadata.region_id,
1082                        err: format!(
1083                            "timestamp column {} has different id, existing: {}, got: {}",
1084                            old_ts_column.0, old_ts_column.1, new_ts_column.1
1085                        ),
1086                    }
1087                );
1088            }
1089        }
1090        Ok(())
1091    }
1092
1093    /// Returns true if we need to apply the alteration to the region.
1094    pub fn need_alter(&self, metadata: &RegionMetadata) -> bool {
1095        debug_assert!(self.validate(metadata).is_ok());
1096        match self {
1097            AlterKind::AddColumns { columns } => columns
1098                .iter()
1099                .any(|col_to_add| col_to_add.need_alter(metadata)),
1100            AlterKind::DropColumns { names } => names
1101                .iter()
1102                .any(|name| metadata.column_by_name(name).is_some()),
1103            AlterKind::ModifyColumnTypes { columns } => columns
1104                .iter()
1105                .any(|col_to_change| col_to_change.need_alter(metadata)),
1106            AlterKind::SetJsonSettings {
1107                column_name,
1108                settings,
1109            } => metadata.column_by_name(column_name).is_some_and(|col| {
1110                col.column_schema
1111                    .extension_type::<Json2ExtensionType>()
1112                    .ok()
1113                    .flatten()
1114                    .is_none_or(|extension| {
1115                        !extension.metadata().json_settings().equivalent(settings)
1116                    })
1117            }),
1118            AlterKind::SetRegionOptions { .. } => true,
1119            AlterKind::UnsetRegionOptions { .. } => true,
1120            AlterKind::SetIndexes { options, .. } => options
1121                .iter()
1122                .any(|option| metadata.column_by_name(option.column_name()).is_some()),
1123            AlterKind::UnsetIndexes { options } => options
1124                .iter()
1125                .any(|option| metadata.column_by_name(option.column_name()).is_some()),
1126            AlterKind::DropDefaults { names } => names
1127                .iter()
1128                .any(|name| metadata.column_by_name(name).is_some()),
1129
1130            AlterKind::SetDefaults { columns } => columns
1131                .iter()
1132                .any(|x| metadata.column_by_name(&x.name).is_some()),
1133            AlterKind::SyncColumns { column_metadatas } => {
1134                metadata.column_metadatas != *column_metadatas
1135            }
1136        }
1137    }
1138
1139    /// Returns an error if the column to drop is invalid.
1140    fn validate_column_to_drop(name: &str, metadata: &RegionMetadata) -> Result<()> {
1141        let Some(column) = metadata.column_by_name(name) else {
1142            return Ok(());
1143        };
1144        ensure!(
1145            column.semantic_type == SemanticType::Field,
1146            InvalidRegionRequestSnafu {
1147                region_id: metadata.region_id,
1148                err: format!("column {} is not a field and could not be dropped", name),
1149            }
1150        );
1151        Ok(())
1152    }
1153
1154    /// Returns an error if the column's alter index option is invalid.
1155    fn validate_column_alter_index_option(
1156        column_name: &String,
1157        metadata: &RegionMetadata,
1158        is_fulltext: bool,
1159    ) -> Result<()> {
1160        let column = metadata
1161            .column_by_name(column_name)
1162            .context(InvalidRegionRequestSnafu {
1163                region_id: metadata.region_id,
1164                err: format!("column {} not found", column_name),
1165            })?;
1166
1167        if is_fulltext {
1168            ensure!(
1169                column.column_schema.data_type.is_string(),
1170                InvalidRegionRequestSnafu {
1171                    region_id: metadata.region_id,
1172                    err: format!(
1173                        "cannot change alter index options for non-string column {}",
1174                        column_name
1175                    ),
1176                }
1177            );
1178        }
1179
1180        Ok(())
1181    }
1182
1183    /// Returns an error if the column isn't exist.
1184    fn validate_column_existence(column_name: &String, metadata: &RegionMetadata) -> Result<()> {
1185        metadata
1186            .column_by_name(column_name)
1187            .context(InvalidRegionRequestSnafu {
1188                region_id: metadata.region_id,
1189                err: format!("column {} not found", column_name),
1190            })?;
1191
1192        Ok(())
1193    }
1194
1195    fn validate_set_json_settings(col_name: &String, metadata: &RegionMetadata) -> Result<()> {
1196        let region_id = metadata.region_id;
1197
1198        let col = metadata
1199            .column_by_name(col_name)
1200            .with_context(|| InvalidRegionRequestSnafu {
1201                region_id,
1202                err: format!("column {} not found", col_name),
1203            })?;
1204
1205        ensure!(
1206            col.semantic_type == SemanticType::Field,
1207            InvalidRegionRequestSnafu {
1208                region_id,
1209                err: format!("column {} is not a field column", col_name),
1210            }
1211        );
1212        ensure!(
1213            col.column_schema.data_type.is_json2(),
1214            InvalidRegionRequestSnafu {
1215                region_id,
1216                err: format!("column {} is not a JSON2 column", col_name),
1217            }
1218        );
1219
1220        Ok(())
1221    }
1222}
1223
1224impl TryFrom<alter_request::Kind> for AlterKind {
1225    type Error = MetadataError;
1226
1227    fn try_from(kind: alter_request::Kind) -> Result<Self> {
1228        let alter_kind = match kind {
1229            alter_request::Kind::AddColumns(x) => {
1230                let columns = x
1231                    .add_columns
1232                    .into_iter()
1233                    .map(|x| x.try_into())
1234                    .collect::<Result<Vec<_>>>()?;
1235                AlterKind::AddColumns { columns }
1236            }
1237            alter_request::Kind::ModifyColumnTypes(x) => {
1238                let columns = x
1239                    .modify_column_types
1240                    .into_iter()
1241                    .map(|x| x.into())
1242                    .collect::<Vec<_>>();
1243                AlterKind::ModifyColumnTypes { columns }
1244            }
1245            alter_request::Kind::SetJsonSettings(x) => {
1246                let settings = x.settings.context(InvalidRawRegionRequestSnafu {
1247                    err: "missing settings in SetJsonSettings",
1248                })?;
1249                AlterKind::SetJsonSettings {
1250                    column_name: x.column_name,
1251                    settings: json_settings_from_proto(settings)?,
1252                }
1253            }
1254            alter_request::Kind::DropColumns(x) => {
1255                let names = x.drop_columns.into_iter().map(|x| x.name).collect();
1256                AlterKind::DropColumns { names }
1257            }
1258            alter_request::Kind::SetTableOptions(options) => AlterKind::SetRegionOptions {
1259                options: options
1260                    .table_options
1261                    .iter()
1262                    .map(TryFrom::try_from)
1263                    .collect::<Result<Vec<_>>>()?,
1264            },
1265            alter_request::Kind::UnsetTableOptions(options) => AlterKind::UnsetRegionOptions {
1266                keys: options
1267                    .keys
1268                    .iter()
1269                    .map(|key| UnsetRegionOption::try_from(key.as_str()))
1270                    .collect::<Result<Vec<_>>>()?,
1271            },
1272            alter_request::Kind::SetIndex(o) => AlterKind::SetIndexes {
1273                options: vec![SetIndexOption::try_from(o)?],
1274            },
1275            alter_request::Kind::UnsetIndex(o) => AlterKind::UnsetIndexes {
1276                options: vec![UnsetIndexOption::try_from(o)?],
1277            },
1278            alter_request::Kind::SetIndexes(o) => AlterKind::SetIndexes {
1279                options: o
1280                    .set_indexes
1281                    .into_iter()
1282                    .map(SetIndexOption::try_from)
1283                    .collect::<Result<Vec<_>>>()?,
1284            },
1285            alter_request::Kind::UnsetIndexes(o) => AlterKind::UnsetIndexes {
1286                options: o
1287                    .unset_indexes
1288                    .into_iter()
1289                    .map(UnsetIndexOption::try_from)
1290                    .collect::<Result<Vec<_>>>()?,
1291            },
1292            alter_request::Kind::DropDefaults(x) => AlterKind::DropDefaults {
1293                names: x.drop_defaults.into_iter().map(|x| x.column_name).collect(),
1294            },
1295            alter_request::Kind::SetDefaults(x) => AlterKind::SetDefaults {
1296                columns: x
1297                    .set_defaults
1298                    .into_iter()
1299                    .map(|x| {
1300                        Ok(SetDefault {
1301                            name: x.column_name,
1302                            default_constraint: x.default_constraint.clone(),
1303                        })
1304                    })
1305                    .collect::<Result<Vec<_>>>()?,
1306            },
1307            alter_request::Kind::SyncColumns(x) => AlterKind::SyncColumns {
1308                column_metadatas: x
1309                    .column_defs
1310                    .into_iter()
1311                    .map(ColumnMetadata::try_from_column_def)
1312                    .collect::<Result<Vec<_>>>()?,
1313            },
1314        };
1315
1316        Ok(alter_kind)
1317    }
1318}
1319
1320/// Adds a column.
1321#[derive(Debug, PartialEq, Eq, Clone)]
1322pub struct AddColumn {
1323    /// Metadata of the column to add.
1324    pub column_metadata: ColumnMetadata,
1325    /// Location to add the column. If location is None, the region adds
1326    /// the column to the last.
1327    pub location: Option<AddColumnLocation>,
1328}
1329
1330impl AddColumn {
1331    /// Returns an error if the column to add is invalid.
1332    ///
1333    /// It allows adding existing columns. However, the existing column must have the same metadata
1334    /// and the location must be None.
1335    pub fn validate(&self, metadata: &RegionMetadata) -> Result<()> {
1336        ensure!(
1337            self.column_metadata.column_schema.is_nullable()
1338                || self
1339                    .column_metadata
1340                    .column_schema
1341                    .default_constraint()
1342                    .is_some(),
1343            InvalidRegionRequestSnafu {
1344                region_id: metadata.region_id,
1345                err: format!(
1346                    "no default value for column {}",
1347                    self.column_metadata.column_schema.name
1348                ),
1349            }
1350        );
1351
1352        if let Some(existing_column) =
1353            metadata.column_by_name(&self.column_metadata.column_schema.name)
1354        {
1355            // If the column already exists.
1356            ensure!(
1357                *existing_column == self.column_metadata,
1358                InvalidRegionRequestSnafu {
1359                    region_id: metadata.region_id,
1360                    err: format!(
1361                        "column {} already exists with different metadata, existing: {:?}, got: {:?}",
1362                        self.column_metadata.column_schema.name,
1363                        existing_column,
1364                        self.column_metadata,
1365                    ),
1366                }
1367            );
1368            ensure!(
1369                self.location.is_none(),
1370                InvalidRegionRequestSnafu {
1371                    region_id: metadata.region_id,
1372                    err: format!(
1373                        "column {} already exists, but location is specified",
1374                        self.column_metadata.column_schema.name
1375                    ),
1376                }
1377            );
1378        }
1379
1380        if let Some(existing_column) = metadata.column_by_id(self.column_metadata.column_id) {
1381            // Ensures the existing column has the same name.
1382            ensure!(
1383                existing_column.column_schema.name == self.column_metadata.column_schema.name,
1384                InvalidRegionRequestSnafu {
1385                    region_id: metadata.region_id,
1386                    err: format!(
1387                        "column id {} already exists with different name {}",
1388                        self.column_metadata.column_id, existing_column.column_schema.name
1389                    ),
1390                }
1391            );
1392        }
1393
1394        Ok(())
1395    }
1396
1397    /// Returns true if no column to add to the region.
1398    pub fn need_alter(&self, metadata: &RegionMetadata) -> bool {
1399        debug_assert!(self.validate(metadata).is_ok());
1400        metadata
1401            .column_by_name(&self.column_metadata.column_schema.name)
1402            .is_none()
1403    }
1404}
1405
1406impl TryFrom<v1::region::AddColumn> for AddColumn {
1407    type Error = MetadataError;
1408
1409    fn try_from(add_column: v1::region::AddColumn) -> Result<Self> {
1410        let column_def = add_column
1411            .column_def
1412            .context(InvalidRawRegionRequestSnafu {
1413                err: "missing column_def in AddColumn",
1414            })?;
1415
1416        let column_metadata = ColumnMetadata::try_from_column_def(column_def)?;
1417        let location = add_column
1418            .location
1419            .map(AddColumnLocation::try_from)
1420            .transpose()?;
1421
1422        Ok(AddColumn {
1423            column_metadata,
1424            location,
1425        })
1426    }
1427}
1428
1429/// Location to add a column.
1430#[derive(Debug, PartialEq, Eq, Clone)]
1431pub enum AddColumnLocation {
1432    /// Add the column to the first position of columns.
1433    First,
1434    /// Add the column after specific column.
1435    After {
1436        /// Add the column after this column.
1437        column_name: String,
1438    },
1439}
1440
1441impl TryFrom<v1::AddColumnLocation> for AddColumnLocation {
1442    type Error = MetadataError;
1443
1444    fn try_from(location: v1::AddColumnLocation) -> Result<Self> {
1445        let location_type = LocationType::try_from(location.location_type)
1446            .map_err(|e| InvalidRawRegionRequestSnafu { err: e.to_string() }.build())?;
1447        let add_column_location = match location_type {
1448            LocationType::First => AddColumnLocation::First,
1449            LocationType::After => AddColumnLocation::After {
1450                column_name: location.after_column_name,
1451            },
1452        };
1453
1454        Ok(add_column_location)
1455    }
1456}
1457
1458/// Change a column's datatype.
1459#[derive(Debug, PartialEq, Eq, Clone)]
1460pub struct ModifyColumnType {
1461    /// Schema of the column to modify.
1462    pub column_name: String,
1463    /// Column will be changed to this type.
1464    pub target_type: ConcreteDataType,
1465}
1466
1467impl ModifyColumnType {
1468    /// Returns an error if the column's datatype to change is invalid.
1469    pub fn validate(&self, metadata: &RegionMetadata) -> Result<()> {
1470        let column_meta = metadata
1471            .column_by_name(&self.column_name)
1472            .with_context(|| InvalidRegionRequestSnafu {
1473                region_id: metadata.region_id,
1474                err: format!("column {} not found", self.column_name),
1475            })?;
1476
1477        match column_meta.semantic_type {
1478            SemanticType::Field => {
1479                ensure!(
1480                    column_meta
1481                        .column_schema
1482                        .data_type
1483                        .can_arrow_type_cast_to(&self.target_type),
1484                    InvalidRegionRequestSnafu {
1485                        region_id: metadata.region_id,
1486                        err: format!(
1487                            "column '{}' cannot be cast automatically to type '{}'",
1488                            self.column_name, self.target_type
1489                        ),
1490                    }
1491                );
1492            }
1493            // The time index column only supports widening its timestamp
1494            // unit; historical SST data is cast to the new unit on read.
1495            // A same-type change validates so a retried alter procedure is a
1496            // no-op instead of failing the retry forever.
1497            SemanticType::Timestamp => {
1498                ensure!(
1499                    column_meta.column_schema.data_type == self.target_type
1500                        || column_meta
1501                            .column_schema
1502                            .data_type
1503                            .is_timestamp_unit_widening_to(&self.target_type),
1504                    InvalidRegionRequestSnafu {
1505                        region_id: metadata.region_id,
1506                        err: time_index_not_widening_error(
1507                            &column_meta.column_schema.name,
1508                            &column_meta.column_schema.data_type,
1509                            &self.target_type,
1510                        ),
1511                    }
1512                );
1513            }
1514            SemanticType::Tag => {
1515                return InvalidRegionRequestSnafu {
1516                    region_id: metadata.region_id,
1517                    err: format!(
1518                        "tag column '{}' cannot change type, it is part of the primary key",
1519                        self.column_name
1520                    ),
1521                }
1522                .fail();
1523            }
1524        }
1525
1526        Ok(())
1527    }
1528
1529    /// Returns true if no column's datatype to change to the region.
1530    /// A column already in the target type needs no alteration, so a retried
1531    /// alter is a successful no-op.
1532    pub fn need_alter(&self, metadata: &RegionMetadata) -> bool {
1533        debug_assert!(self.validate(metadata).is_ok());
1534        metadata
1535            .column_by_name(&self.column_name)
1536            .is_some_and(|column| column.column_schema.data_type != self.target_type)
1537    }
1538}
1539
1540impl From<v1::ModifyColumnType> for ModifyColumnType {
1541    fn from(modify_column_type: v1::ModifyColumnType) -> Self {
1542        let target_type = ColumnDataTypeWrapper::new(
1543            modify_column_type.target_type(),
1544            modify_column_type.target_type_extension,
1545        )
1546        .into();
1547
1548        ModifyColumnType {
1549            column_name: modify_column_type.column_name,
1550            target_type,
1551        }
1552    }
1553}
1554
1555fn json_settings_from_proto(settings: v1::JsonSettings) -> Result<JsonSettings> {
1556    let type_hints = settings
1557        .type_hints
1558        .into_iter()
1559        .map(|hint| {
1560            let wrapper = ColumnDataTypeWrapper::try_new(hint.data_type, hint.datatype_extension)
1561                .map_err(|err| {
1562                InvalidRawRegionRequestSnafu {
1563                    err: err.to_string(),
1564                }
1565                .build()
1566            })?;
1567            let data_type = ConcreteDataType::from(wrapper);
1568
1569            Ok(JsonTypeHint {
1570                path: hint.path,
1571                data_type,
1572                inverted_index: false,
1573            })
1574        })
1575        .collect::<Result<Vec<_>>>()?;
1576
1577    JsonSettings::try_new(type_hints, settings.max_auto_expanded_paths).map_err(|err| {
1578        InvalidRawRegionRequestSnafu {
1579            err: err.to_string(),
1580        }
1581        .build()
1582    })
1583}
1584
1585/// Region option changes used by ALTER requests.
1586///
1587/// This type is serialized for request persistence. Keep future changes backward
1588/// compatible with previously serialized variants.
1589#[derive(Debug, Eq, PartialEq, Clone)]
1590pub enum SetRegionOption {
1591    WriteBufferSize(Option<ReadableSize>),
1592    Ttl(Option<TimeToLive>),
1593    // Modifying TwscOptions with values as (option name, new value).
1594    Twsc(String, String),
1595    // Modifying the SST format.
1596    Format(String),
1597    // Modifying the append mode.
1598    AppendMode(bool),
1599    // Modifying the per-region auto flush interval override.
1600    AutoFlushInterval(Option<Duration>),
1601    // Modifying the max number of rows in a parquet row group.
1602    MaxRowGroupRowCount(Option<usize>),
1603    PreserveRowSequence(bool),
1604    // Whether to skip writing new WAL entries.
1605    SkipWal(bool),
1606}
1607
1608#[derive(Serialize, Deserialize)]
1609enum SetRegionOptionSerde {
1610    WriteBufferSize(Option<ReadableSize>),
1611    Ttl(Option<TimeToLive>),
1612    Twsc(String, String),
1613    Format(String),
1614    AppendMode(bool),
1615    AutoFlushInterval(Option<Duration>),
1616    MaxRowGroupRowCount(Option<usize>),
1617    PreserveRowSequence(bool),
1618    SkipWal(bool),
1619}
1620
1621impl Serialize for SetRegionOption {
1622    fn serialize<S>(&self, serializer: S) -> std::result::Result<S::Ok, S::Error>
1623    where
1624        S: Serializer,
1625    {
1626        // Older binaries deserialize the disable request as a unit variant.
1627        if matches!(self, Self::SkipWal(true)) {
1628            return serializer.serialize_unit_variant("SetRegionOption", 8, "SkipWal");
1629        }
1630
1631        let option = match self {
1632            Self::WriteBufferSize(value) => SetRegionOptionSerde::WriteBufferSize(*value),
1633            Self::Ttl(value) => SetRegionOptionSerde::Ttl(*value),
1634            Self::Twsc(key, value) => SetRegionOptionSerde::Twsc(key.clone(), value.clone()),
1635            Self::Format(value) => SetRegionOptionSerde::Format(value.clone()),
1636            Self::AppendMode(value) => SetRegionOptionSerde::AppendMode(*value),
1637            Self::AutoFlushInterval(value) => SetRegionOptionSerde::AutoFlushInterval(*value),
1638            Self::MaxRowGroupRowCount(value) => SetRegionOptionSerde::MaxRowGroupRowCount(*value),
1639            Self::PreserveRowSequence(value) => SetRegionOptionSerde::PreserveRowSequence(*value),
1640            Self::SkipWal(value) => SetRegionOptionSerde::SkipWal(*value),
1641        };
1642        option.serialize(serializer)
1643    }
1644}
1645
1646#[derive(Deserialize)]
1647enum LegacySetRegionOption {
1648    SkipWal,
1649}
1650
1651#[derive(Deserialize)]
1652#[serde(untagged)]
1653enum BackwardCompatibleSetRegionOption {
1654    Current(SetRegionOptionSerde),
1655    Legacy(LegacySetRegionOption),
1656}
1657
1658impl From<SetRegionOptionSerde> for SetRegionOption {
1659    fn from(option: SetRegionOptionSerde) -> Self {
1660        match option {
1661            SetRegionOptionSerde::WriteBufferSize(value) => Self::WriteBufferSize(value),
1662            SetRegionOptionSerde::Ttl(value) => Self::Ttl(value),
1663            SetRegionOptionSerde::Twsc(key, value) => Self::Twsc(key, value),
1664            SetRegionOptionSerde::Format(value) => Self::Format(value),
1665            SetRegionOptionSerde::AppendMode(value) => Self::AppendMode(value),
1666            SetRegionOptionSerde::AutoFlushInterval(value) => Self::AutoFlushInterval(value),
1667            SetRegionOptionSerde::MaxRowGroupRowCount(value) => Self::MaxRowGroupRowCount(value),
1668            SetRegionOptionSerde::PreserveRowSequence(value) => Self::PreserveRowSequence(value),
1669            SetRegionOptionSerde::SkipWal(value) => Self::SkipWal(value),
1670        }
1671    }
1672}
1673
1674impl<'de> Deserialize<'de> for SetRegionOption {
1675    fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
1676    where
1677        D: Deserializer<'de>,
1678    {
1679        Ok(
1680            match BackwardCompatibleSetRegionOption::deserialize(deserializer)? {
1681                BackwardCompatibleSetRegionOption::Current(option) => option.into(),
1682                BackwardCompatibleSetRegionOption::Legacy(LegacySetRegionOption::SkipWal) => {
1683                    Self::SkipWal(true)
1684                }
1685            },
1686        )
1687    }
1688}
1689
1690impl TryFrom<&PbOption> for SetRegionOption {
1691    type Error = MetadataError;
1692
1693    fn try_from(value: &PbOption) -> std::result::Result<Self, Self::Error> {
1694        let PbOption { key, value } = value;
1695        match key.as_str() {
1696            WRITE_BUFFER_SIZE_KEY => {
1697                let size = value
1698                    .parse::<ReadableSize>()
1699                    .map_err(|_| InvalidSetRegionOptionRequestSnafu { key, value }.build())?;
1700                Ok(Self::WriteBufferSize(Some(size)))
1701            }
1702            TTL_KEY => {
1703                let ttl = TimeToLive::from_humantime_or_str(value)
1704                    .map_err(|_| InvalidSetRegionOptionRequestSnafu { key, value }.build())?;
1705
1706                Ok(Self::Ttl(Some(ttl)))
1707            }
1708            TWCS_TRIGGER_FILE_NUM
1709            | TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM
1710            | TWCS_ACTIVE_WINDOW_L1_MERGE_TRIGGER
1711            | TWCS_INACTIVE_WINDOW_TRIGGER_FILE_NUM
1712            | TWCS_INACTIVE_WINDOW_L1_MERGE_TRIGGER
1713            | TWCS_MAX_OUTPUT_FILE_SIZE
1714            | TWCS_TIME_WINDOW => Ok(Self::Twsc(key.clone(), value.clone())),
1715            SST_FORMAT_KEY => Ok(Self::Format(value.clone())),
1716            APPEND_MODE_KEY => {
1717                let append_mode = value
1718                    .parse::<bool>()
1719                    .map_err(|_| InvalidSetRegionOptionRequestSnafu { key, value }.build())?;
1720                Ok(Self::AppendMode(append_mode))
1721            }
1722            AUTO_FLUSH_INTERVAL_KEY => {
1723                if value.is_empty() {
1724                    // SET 'auto_flush_interval' = NULL comes through as an empty
1725                    // string; treat it as clearing the override (fall back to
1726                    // the global default), same as Ttl.
1727                    return Ok(Self::AutoFlushInterval(None));
1728                }
1729                let interval = humantime::parse_duration(value)
1730                    .map_err(|_| InvalidSetRegionOptionRequestSnafu { key, value }.build())?;
1731                if interval <= Duration::ZERO {
1732                    return InvalidSetRegionOptionRequestSnafu { key, value }.fail();
1733                }
1734                Ok(Self::AutoFlushInterval(Some(interval)))
1735            }
1736            MAX_ROW_GROUP_ROW_COUNT => {
1737                if value.is_empty() {
1738                    return Ok(Self::MaxRowGroupRowCount(None));
1739                }
1740                let row_count = value
1741                    .parse::<usize>()
1742                    .ok()
1743                    .filter(|row_count| {
1744                        *row_count > 0 && *row_count <= MAX_ROW_GROUP_ROW_COUNT_LIMIT
1745                    })
1746                    .ok_or_else(|| InvalidSetRegionOptionRequestSnafu { key, value }.build())?;
1747                Ok(Self::MaxRowGroupRowCount(Some(row_count)))
1748            }
1749            PRESERVE_ROW_SEQUENCE => {
1750                let preserve = value
1751                    .parse::<bool>()
1752                    .map_err(|_| InvalidSetRegionOptionRequestSnafu { key, value }.build())?;
1753                Ok(Self::PreserveRowSequence(preserve))
1754            }
1755            SKIP_WAL_KEY => {
1756                let skip_wal = value
1757                    .parse::<bool>()
1758                    .map_err(|_| InvalidSetRegionOptionRequestSnafu { key, value }.build())?;
1759                Ok(Self::SkipWal(skip_wal))
1760            }
1761            _ => InvalidSetRegionOptionRequestSnafu { key, value }.fail(),
1762        }
1763    }
1764}
1765
1766impl From<&UnsetRegionOption> for SetRegionOption {
1767    fn from(unset_option: &UnsetRegionOption) -> Self {
1768        match unset_option {
1769            UnsetRegionOption::TwcsTriggerFileNum => {
1770                SetRegionOption::Twsc(unset_option.to_string(), String::new())
1771            }
1772            UnsetRegionOption::TwcsActiveWindowTriggerFileNum => {
1773                SetRegionOption::Twsc(unset_option.to_string(), String::new())
1774            }
1775            UnsetRegionOption::TwcsActiveWindowL1MergeTrigger => {
1776                SetRegionOption::Twsc(unset_option.to_string(), String::new())
1777            }
1778            UnsetRegionOption::TwcsInactiveWindowTriggerFileNum => {
1779                SetRegionOption::Twsc(unset_option.to_string(), String::new())
1780            }
1781            UnsetRegionOption::TwcsInactiveWindowL1MergeTrigger => {
1782                SetRegionOption::Twsc(unset_option.to_string(), String::new())
1783            }
1784            UnsetRegionOption::TwcsMaxOutputFileSize => {
1785                SetRegionOption::Twsc(unset_option.to_string(), String::new())
1786            }
1787            UnsetRegionOption::TwcsTimeWindow => {
1788                SetRegionOption::Twsc(unset_option.to_string(), String::new())
1789            }
1790            UnsetRegionOption::Ttl => SetRegionOption::Ttl(Default::default()),
1791            UnsetRegionOption::MaxRowGroupRowCount => SetRegionOption::MaxRowGroupRowCount(None),
1792            UnsetRegionOption::WriteBufferSize => SetRegionOption::WriteBufferSize(None),
1793            UnsetRegionOption::PreserveRowSequence => SetRegionOption::PreserveRowSequence(false),
1794        }
1795    }
1796}
1797
1798impl TryFrom<&str> for UnsetRegionOption {
1799    type Error = MetadataError;
1800
1801    fn try_from(key: &str) -> Result<Self> {
1802        match key.to_ascii_lowercase().as_str() {
1803            TTL_KEY => Ok(Self::Ttl),
1804            WRITE_BUFFER_SIZE_KEY => Ok(Self::WriteBufferSize),
1805            TWCS_TRIGGER_FILE_NUM => Ok(Self::TwcsTriggerFileNum),
1806            TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM => Ok(Self::TwcsActiveWindowTriggerFileNum),
1807            TWCS_ACTIVE_WINDOW_L1_MERGE_TRIGGER => Ok(Self::TwcsActiveWindowL1MergeTrigger),
1808            TWCS_INACTIVE_WINDOW_TRIGGER_FILE_NUM => Ok(Self::TwcsInactiveWindowTriggerFileNum),
1809            TWCS_INACTIVE_WINDOW_L1_MERGE_TRIGGER => Ok(Self::TwcsInactiveWindowL1MergeTrigger),
1810            TWCS_MAX_OUTPUT_FILE_SIZE => Ok(Self::TwcsMaxOutputFileSize),
1811            TWCS_TIME_WINDOW => Ok(Self::TwcsTimeWindow),
1812            MAX_ROW_GROUP_ROW_COUNT => Ok(Self::MaxRowGroupRowCount),
1813            PRESERVE_ROW_SEQUENCE => Ok(Self::PreserveRowSequence),
1814            _ => InvalidUnsetRegionOptionRequestSnafu { key }.fail(),
1815        }
1816    }
1817}
1818
1819#[derive(Debug, Eq, PartialEq, Clone, Serialize, Deserialize)]
1820pub enum UnsetRegionOption {
1821    TwcsTriggerFileNum,
1822    TwcsActiveWindowTriggerFileNum,
1823    TwcsInactiveWindowTriggerFileNum,
1824    TwcsInactiveWindowL1MergeTrigger,
1825    TwcsMaxOutputFileSize,
1826    TwcsTimeWindow,
1827    Ttl,
1828    MaxRowGroupRowCount,
1829    WriteBufferSize,
1830    PreserveRowSequence,
1831    TwcsActiveWindowL1MergeTrigger,
1832}
1833
1834impl UnsetRegionOption {
1835    pub fn as_str(&self) -> &str {
1836        match self {
1837            Self::Ttl => TTL_KEY,
1838            Self::WriteBufferSize => WRITE_BUFFER_SIZE_KEY,
1839            Self::TwcsTriggerFileNum => TWCS_TRIGGER_FILE_NUM,
1840            Self::TwcsActiveWindowTriggerFileNum => TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM,
1841            Self::TwcsActiveWindowL1MergeTrigger => TWCS_ACTIVE_WINDOW_L1_MERGE_TRIGGER,
1842            Self::TwcsInactiveWindowTriggerFileNum => TWCS_INACTIVE_WINDOW_TRIGGER_FILE_NUM,
1843            Self::TwcsInactiveWindowL1MergeTrigger => TWCS_INACTIVE_WINDOW_L1_MERGE_TRIGGER,
1844            Self::TwcsMaxOutputFileSize => TWCS_MAX_OUTPUT_FILE_SIZE,
1845            Self::TwcsTimeWindow => TWCS_TIME_WINDOW,
1846            Self::MaxRowGroupRowCount => MAX_ROW_GROUP_ROW_COUNT,
1847            Self::PreserveRowSequence => PRESERVE_ROW_SEQUENCE,
1848        }
1849    }
1850}
1851
1852impl Display for UnsetRegionOption {
1853    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1854        write!(f, "{}", self.as_str())
1855    }
1856}
1857
1858#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
1859pub enum RegionFlushReason {
1860    /// Flush triggered before region migration.
1861    RegionMigration,
1862    /// Flush triggered by repartition procedure.
1863    Repartition,
1864    /// Flush triggered by remote WAL pruning.
1865    RemoteWalPrune,
1866    /// Flush region before closing region.
1867    Closing,
1868    /// Flush region before downgrading region.
1869    Downgrading,
1870}
1871
1872#[derive(Debug, Clone, Default)]
1873pub struct RegionFlushRequest {
1874    pub row_group_size: Option<usize>,
1875    pub reason: Option<RegionFlushReason>,
1876}
1877
1878#[derive(Debug)]
1879pub struct RegionCompactRequest {
1880    pub options: compact_request::Options,
1881    pub parallelism: Option<u32>,
1882    pub time_range: Option<TimestampRange>,
1883}
1884
1885impl Default for RegionCompactRequest {
1886    fn default() -> Self {
1887        Self {
1888            // Default to regular compaction.
1889            options: compact_request::Options::Regular(Default::default()),
1890            parallelism: None,
1891            time_range: None,
1892        }
1893    }
1894}
1895
1896#[derive(Debug, Clone, Default)]
1897pub struct RegionBuildIndexRequest {
1898    /// The index build mode. Absent options select SST indexes.
1899    pub options: Option<build_index_request::Options>,
1900}
1901
1902/// Truncate region request.
1903#[derive(Debug)]
1904pub enum RegionTruncateRequest {
1905    /// Truncate all data in the region.
1906    All,
1907    /// Discard all unflushed data while preserving persisted SST files.
1908    ///
1909    /// This destroys the region's in-memory data irreversibly. Persisted SST files and
1910    /// the writable state are preserved, and the WAL is obsoleted up to the discard point
1911    /// so a restart won't replay the discarded data.
1912    ///
1913    /// An error may be returned after the data has already been discarded, because the
1914    /// WAL is obsoleted last. Retrying is safe: a request against a region with nothing
1915    /// left to discard only re-attempts the WAL obsoletion.
1916    Unflushed,
1917    ByTimeRanges {
1918        /// Time ranges to truncate. Both bound are inclusive.
1919        /// only files that are fully contained in the time range will be truncated.
1920        /// so no guarantee that all data in the time range will be truncated.
1921        time_ranges: Vec<(Timestamp, Timestamp)>,
1922    },
1923}
1924
1925/// Catchup region request.
1926///
1927/// Makes a readonly region to catch up to leader region changes.
1928/// There is no effect if it operating on a leader region.
1929#[derive(Debug, Clone, Copy, Default)]
1930pub struct RegionCatchupRequest {
1931    /// Sets it to writable if it's available after it has caught up with all changes.
1932    pub set_writable: bool,
1933    /// The `entry_id` that was expected to reply to.
1934    /// `None` stands replaying to latest.
1935    pub entry_id: Option<entry::Id>,
1936    /// Used for metrics metadata region.
1937    /// The `entry_id` that was expected to reply to.
1938    /// `None` stands replaying to latest.
1939    pub metadata_entry_id: Option<entry::Id>,
1940    /// Replay checkpoint.
1941    pub checkpoint: Option<ReplayCheckpoint>,
1942}
1943
1944#[derive(Debug, Clone)]
1945pub struct RegionBulkInsertsRequest {
1946    /// Whether this request should skip WAL.
1947    pub skip_wal: bool,
1948    pub region_id: RegionId,
1949    pub payload: DfRecordBatch,
1950    pub raw_data: ArrowIpc,
1951    pub partition_expr_version: Option<u64>,
1952    pub aligned_schema_version: Option<u64>,
1953}
1954
1955impl RegionBulkInsertsRequest {
1956    pub fn estimated_size(&self) -> usize {
1957        self.payload.get_array_memory_size()
1958    }
1959}
1960
1961/// Request to stage a region with a new partition directive.
1962///
1963/// This request transitions a region into the staging mode.
1964/// It first flushes the memtable for the old partition expression if it is not
1965/// empty, then enters the staging mode with the new directive.
1966#[derive(Debug, Clone, PartialEq, Eq)]
1967pub enum StagingPartitionDirective {
1968    UpdatePartitionExpr(String),
1969    RejectAllWrites,
1970}
1971
1972impl StagingPartitionDirective {
1973    /// Returns the partition expression carried by this directive, if any.
1974    pub fn partition_expr(&self) -> Option<&str> {
1975        match self {
1976            Self::UpdatePartitionExpr(expr) => Some(expr),
1977            Self::RejectAllWrites => None,
1978        }
1979    }
1980}
1981
1982#[derive(Debug, Clone)]
1983pub struct EnterStagingRequest {
1984    /// The staging partition directive of the region.
1985    pub partition_directive: StagingPartitionDirective,
1986}
1987
1988impl EnterStagingRequest {
1989    /// Builds an enter-staging request with a partition expression directive.
1990    pub fn with_partition_expr(partition_expr: String) -> Self {
1991        Self {
1992            partition_directive: StagingPartitionDirective::UpdatePartitionExpr(partition_expr),
1993        }
1994    }
1995}
1996
1997/// This request is used as part of the region repartition.
1998///
1999/// After a region has entered staging mode with a new partition expression
2000/// expression) and a separate process (for example, `remap_manifests`) has
2001/// generated the new file assignments for the staging region, this request
2002/// applies that generated manifest to the region.
2003///
2004/// In practice, this means:
2005/// - The `partition_expr` identifies the staging partition expression that the manifest
2006///   was generated for.
2007/// - `central_region_id` specifies which region holds the staging blob storage
2008///   where the manifest was written during the `remap_manifests` operation.
2009/// - `manifest_path` is the relative path within the central region's staging
2010///   blob storage to fetch the generated manifest.
2011///
2012/// It should typically be called **after** the staging region has been
2013/// initialized by [`EnterStagingRequest`] and the new file layout has been
2014/// computed, to finalize the repartition operation.
2015#[derive(Debug, Clone)]
2016pub struct ApplyStagingManifestRequest {
2017    /// The partition expression of the staging region.
2018    pub partition_expr: String,
2019    /// The region that stores the staging manifests in its staging blob storage.
2020    pub central_region_id: RegionId,
2021    /// The relative path to the staging manifest within the central region's
2022    /// staging blob storage.
2023    pub manifest_path: String,
2024}
2025
2026impl fmt::Display for RegionRequest {
2027    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
2028        match self {
2029            RegionRequest::Put(_) => write!(f, "Put"),
2030            RegionRequest::Delete(_) => write!(f, "Delete"),
2031            RegionRequest::Create(_) => write!(f, "Create"),
2032            RegionRequest::Drop(_) => write!(f, "Drop"),
2033            RegionRequest::Open(_) => write!(f, "Open"),
2034            RegionRequest::CleanUp(_) => write!(f, "CleanUp"),
2035            RegionRequest::Close(_) => write!(f, "Close"),
2036            RegionRequest::Alter(_) => write!(f, "Alter"),
2037            RegionRequest::Flush(_) => write!(f, "Flush"),
2038            RegionRequest::Compact(_) => write!(f, "Compact"),
2039            RegionRequest::BuildIndex(_) => write!(f, "BuildIndex"),
2040            RegionRequest::Truncate(_) => write!(f, "Truncate"),
2041            RegionRequest::Catchup(_) => write!(f, "Catchup"),
2042            RegionRequest::BulkInserts(_) => write!(f, "BulkInserts"),
2043            RegionRequest::EnterStaging(_) => write!(f, "EnterStaging"),
2044            RegionRequest::ApplyStagingManifest(_) => write!(f, "ApplyStagingManifest"),
2045        }
2046    }
2047}
2048
2049#[cfg(test)]
2050mod tests {
2051
2052    use api::v1::region::RegionColumnDef;
2053    use api::v1::{ColumnDataType, ColumnDef};
2054    use common_time::range::TimestampRange;
2055    use datatypes::extension::json::{Json2ExtensionType, JsonMetadata};
2056    use datatypes::json::JsonSettings;
2057    use datatypes::prelude::ConcreteDataType;
2058    use datatypes::schema::{ColumnSchema, FulltextAnalyzer, FulltextBackend};
2059    use datatypes::types::JsonType;
2060
2061    use super::*;
2062    use crate::metadata::RegionMetadataBuilder;
2063
2064    #[test]
2065    fn test_build_index_options_round_trip() {
2066        use prost::Message;
2067
2068        for options in [
2069            None,
2070            Some(build_index_request::Options::SstIndex(Default::default())),
2071            Some(build_index_request::Options::SeriesIndex(Default::default())),
2072        ] {
2073            let request = BuildIndexRequest {
2074                region_id: 42,
2075                options,
2076            };
2077            let decoded = BuildIndexRequest::decode(request.encode_to_vec().as_slice()).unwrap();
2078            let requests = make_region_build_index(decoded).unwrap();
2079            assert_eq!(1, requests.len());
2080            assert_eq!(RegionId::from_u64(42), requests[0].0);
2081            let RegionRequest::BuildIndex(request) = &requests[0].1 else {
2082                panic!("expected build-index request");
2083            };
2084            assert_eq!(options, request.options);
2085        }
2086        // The legacy wire message contains only region_id (field 1).
2087        let legacy = BuildIndexRequest::decode(&[0x08, 42][..]).unwrap();
2088        assert!(legacy.options.is_none());
2089    }
2090
2091    #[test]
2092    fn test_make_region_puts_preserves_skip_wal() {
2093        let region_id = RegionId::new(42, 3);
2094        let rows = Rows::default();
2095        let requests = make_region_puts(InsertRequests {
2096            requests: [false, true, false]
2097                .into_iter()
2098                .map(|skip_wal| api::v1::region::InsertRequest {
2099                    region_id: region_id.as_u64(),
2100                    rows: Some(rows.clone()),
2101                    partition_expr_version: Some(api::v1::PartitionExprVersion { value: 7 }),
2102                    skip_wal,
2103                })
2104                .collect(),
2105        })
2106        .unwrap();
2107
2108        assert_eq!(3, requests.len());
2109        for ((id, request), skip_wal) in requests.into_iter().zip([false, true, false]) {
2110            assert_eq!(region_id, id);
2111            let RegionRequest::Put(request) = request else {
2112                panic!("expected a put request");
2113            };
2114            assert_eq!(rows, request.rows);
2115            assert_eq!(skip_wal, request.skip_wal);
2116            assert_eq!(Some(7), request.partition_expr_version);
2117            assert!(request.hint.is_none());
2118        }
2119    }
2120
2121    #[test]
2122    fn test_make_region_compact_with_time_range() {
2123        let requests = make_region_compact(CompactRequest {
2124            region_id: 42,
2125            time_range: Some(api::v1::region::CompactionTimeRange {
2126                start: 1_000,
2127                end: 2_000,
2128                time_unit: api::v1::TimeUnit::Microsecond as i32,
2129            }),
2130            ..Default::default()
2131        })
2132        .unwrap();
2133
2134        let RegionRequest::Compact(request) = &requests[0].1 else {
2135            unreachable!();
2136        };
2137        assert_eq!(
2138            Some(
2139                TimestampRange::new(
2140                    Timestamp::new_microsecond(1_000),
2141                    Timestamp::new_microsecond(2_000),
2142                )
2143                .unwrap()
2144            ),
2145            request.time_range
2146        );
2147    }
2148
2149    #[test]
2150    fn test_make_region_truncate_unflushed() {
2151        let region_id = RegionId::new(42, 3);
2152        let requests =
2153            RegionRequest::try_from_request_body(region_request::Body::Truncate(TruncateRequest {
2154                region_id: region_id.as_u64(),
2155                kind: Some(truncate_request::Kind::Unflushed(
2156                    api::v1::region::Unflushed {},
2157                )),
2158            }))
2159            .unwrap();
2160
2161        assert_eq!(region_id, requests[0].0);
2162        assert!(matches!(
2163            requests[0].1,
2164            RegionRequest::Truncate(RegionTruncateRequest::Unflushed)
2165        ));
2166    }
2167
2168    #[test]
2169    fn test_make_region_truncate_requires_kind() {
2170        let error =
2171            RegionRequest::try_from_request_body(region_request::Body::Truncate(TruncateRequest {
2172                region_id: RegionId::new(42, 3).as_u64(),
2173                kind: None,
2174            }))
2175            .unwrap_err();
2176
2177        assert!(
2178            error
2179                .to_string()
2180                .contains("missing kind in TruncateRequest")
2181        );
2182    }
2183
2184    #[test]
2185    fn test_from_proto_location() {
2186        let proto_location = v1::AddColumnLocation {
2187            location_type: LocationType::First as i32,
2188            after_column_name: String::default(),
2189        };
2190        let location = AddColumnLocation::try_from(proto_location).unwrap();
2191        assert_eq!(location, AddColumnLocation::First);
2192
2193        let proto_location = v1::AddColumnLocation {
2194            location_type: 10,
2195            after_column_name: String::default(),
2196        };
2197        AddColumnLocation::try_from(proto_location).unwrap_err();
2198
2199        let proto_location = v1::AddColumnLocation {
2200            location_type: LocationType::After as i32,
2201            after_column_name: "a".to_string(),
2202        };
2203        let location = AddColumnLocation::try_from(proto_location).unwrap();
2204        assert_eq!(
2205            location,
2206            AddColumnLocation::After {
2207                column_name: "a".to_string()
2208            }
2209        );
2210    }
2211
2212    #[test]
2213    fn test_from_none_proto_add_column() {
2214        AddColumn::try_from(v1::region::AddColumn {
2215            column_def: None,
2216            location: None,
2217        })
2218        .unwrap_err();
2219    }
2220
2221    #[test]
2222    fn test_set_region_option_auto_flush_interval_try_from() {
2223        use std::time::Duration;
2224
2225        // Valid duration
2226        let pb = PbOption {
2227            key: "auto_flush_interval".to_string(),
2228            value: "5m".to_string(),
2229        };
2230        let opt = SetRegionOption::try_from(&pb).unwrap();
2231        assert_eq!(
2232            opt,
2233            SetRegionOption::AutoFlushInterval(Some(Duration::from_secs(300)))
2234        );
2235
2236        // Empty value clears the override (mirrors Ttl's behaviour for `SET key = NULL`).
2237        let pb = PbOption {
2238            key: "auto_flush_interval".to_string(),
2239            value: String::new(),
2240        };
2241        let opt = SetRegionOption::try_from(&pb).unwrap();
2242        assert_eq!(opt, SetRegionOption::AutoFlushInterval(None));
2243
2244        // Zero is rejected up front (engine invariant).
2245        let pb = PbOption {
2246            key: "auto_flush_interval".to_string(),
2247            value: "0s".to_string(),
2248        };
2249        assert!(SetRegionOption::try_from(&pb).is_err());
2250
2251        // Garbage value is rejected.
2252        let pb = PbOption {
2253            key: "auto_flush_interval".to_string(),
2254            value: "not_a_duration".to_string(),
2255        };
2256        assert!(SetRegionOption::try_from(&pb).is_err());
2257    }
2258
2259    #[test]
2260    fn test_set_region_option_skip_wal_try_from() {
2261        let pb = PbOption {
2262            key: SKIP_WAL_KEY.to_string(),
2263            value: "true".to_string(),
2264        };
2265        assert_eq!(
2266            SetRegionOption::SkipWal(true),
2267            SetRegionOption::try_from(&pb).unwrap()
2268        );
2269
2270        let pb = PbOption {
2271            key: SKIP_WAL_KEY.to_string(),
2272            value: "false".to_string(),
2273        };
2274        assert_eq!(
2275            SetRegionOption::SkipWal(false),
2276            SetRegionOption::try_from(&pb).unwrap()
2277        );
2278
2279        for value in ["", "invalid"] {
2280            let pb = PbOption {
2281                key: SKIP_WAL_KEY.to_string(),
2282                value: value.to_string(),
2283            };
2284            assert!(SetRegionOption::try_from(&pb).is_err());
2285        }
2286
2287        assert!(UnsetRegionOption::try_from(SKIP_WAL_KEY).is_err());
2288    }
2289
2290    #[test]
2291    fn test_set_region_option_skip_wal_serde_compatibility() {
2292        let legacy = serde_json::from_str::<SetRegionOption>(r#""SkipWal""#).unwrap();
2293        assert_eq!(SetRegionOption::SkipWal(true), legacy);
2294
2295        check_set_region_option_skip_wal_serde_compatibility(false);
2296        check_set_region_option_skip_wal_serde_compatibility(true);
2297    }
2298
2299    fn check_set_region_option_skip_wal_serde_compatibility(skip_wal: bool) {
2300        let option = SetRegionOption::SkipWal(skip_wal);
2301        let serialized = serde_json::to_string(&option).unwrap();
2302        let expected = if skip_wal {
2303            r#""SkipWal""#
2304        } else {
2305            r#"{"SkipWal":false}"#
2306        };
2307        assert_eq!(expected, serialized);
2308        assert_eq!(
2309            option,
2310            serde_json::from_str::<SetRegionOption>(&serialized).unwrap()
2311        );
2312        if skip_wal {
2313            assert!(serde_json::from_str::<LegacySetRegionOption>(&serialized).is_ok());
2314            assert_eq!(
2315                option,
2316                serde_json::from_str::<SetRegionOption>(r#"{"SkipWal":true}"#).unwrap()
2317            );
2318        }
2319    }
2320
2321    #[test]
2322    fn test_set_region_option_max_row_group_row_count_try_from() {
2323        let pb = PbOption {
2324            key: MAX_ROW_GROUP_ROW_COUNT.to_string(),
2325            value: "512".to_string(),
2326        };
2327        assert_eq!(
2328            SetRegionOption::MaxRowGroupRowCount(Some(512)),
2329            SetRegionOption::try_from(&pb).unwrap()
2330        );
2331
2332        let pb = PbOption {
2333            key: MAX_ROW_GROUP_ROW_COUNT.to_string(),
2334            value: String::new(),
2335        };
2336        assert_eq!(
2337            SetRegionOption::MaxRowGroupRowCount(None),
2338            SetRegionOption::try_from(&pb).unwrap()
2339        );
2340
2341        for value in [
2342            "0".to_string(),
2343            (MAX_ROW_GROUP_ROW_COUNT_LIMIT + 1).to_string(),
2344            "invalid".to_string(),
2345        ] {
2346            let pb = PbOption {
2347                key: MAX_ROW_GROUP_ROW_COUNT.to_string(),
2348                value,
2349            };
2350            assert!(SetRegionOption::try_from(&pb).is_err());
2351        }
2352
2353        assert_eq!(
2354            UnsetRegionOption::MaxRowGroupRowCount,
2355            UnsetRegionOption::try_from(MAX_ROW_GROUP_ROW_COUNT).unwrap()
2356        );
2357    }
2358
2359    #[test]
2360    fn test_set_region_option_preserve_row_sequence_try_from() {
2361        for (value, expected) in [("true", true), ("false", false)] {
2362            let pb = PbOption {
2363                key: PRESERVE_ROW_SEQUENCE.to_string(),
2364                value: value.to_string(),
2365            };
2366            assert_eq!(
2367                SetRegionOption::PreserveRowSequence(expected),
2368                SetRegionOption::try_from(&pb).unwrap()
2369            );
2370        }
2371
2372        for value in ["1", "invalid"] {
2373            let pb = PbOption {
2374                key: PRESERVE_ROW_SEQUENCE.to_string(),
2375                value: value.to_string(),
2376            };
2377            assert!(SetRegionOption::try_from(&pb).is_err());
2378        }
2379
2380        assert_eq!(
2381            UnsetRegionOption::PreserveRowSequence,
2382            UnsetRegionOption::try_from(PRESERVE_ROW_SEQUENCE).unwrap()
2383        );
2384        assert_eq!(
2385            SetRegionOption::PreserveRowSequence(false),
2386            (&UnsetRegionOption::PreserveRowSequence).into()
2387        );
2388    }
2389
2390    #[test]
2391    fn test_set_twcs_window_trigger_options_try_from() {
2392        for key in [
2393            "compaction.twcs.active_window.trigger_file_num",
2394            "compaction.twcs.active_window.l1_merge_trigger",
2395            "compaction.twcs.inactive_window.trigger_file_num",
2396            "compaction.twcs.inactive_window.l1_merge_trigger",
2397        ] {
2398            let option = PbOption {
2399                key: key.to_string(),
2400                value: "8".to_string(),
2401            };
2402            assert_eq!(
2403                SetRegionOption::Twsc(key.to_string(), "8".to_string()),
2404                SetRegionOption::try_from(&option).unwrap(),
2405                "{key}"
2406            );
2407        }
2408    }
2409
2410    #[test]
2411    fn test_unset_twcs_window_trigger_options_try_from() {
2412        for (key, expected) in [
2413            (
2414                "compaction.twcs.active_window.trigger_file_num",
2415                UnsetRegionOption::TwcsActiveWindowTriggerFileNum,
2416            ),
2417            (
2418                "compaction.twcs.active_window.l1_merge_trigger",
2419                UnsetRegionOption::TwcsActiveWindowL1MergeTrigger,
2420            ),
2421            (
2422                "compaction.twcs.inactive_window.trigger_file_num",
2423                UnsetRegionOption::TwcsInactiveWindowTriggerFileNum,
2424            ),
2425        ] {
2426            assert_eq!(expected, UnsetRegionOption::try_from(key).unwrap());
2427            assert_eq!(
2428                SetRegionOption::Twsc(key.to_string(), String::new()),
2429                SetRegionOption::from(&expected)
2430            );
2431        }
2432
2433        let key = "compaction.twcs.inactive_window.l1_merge_trigger";
2434        let option = UnsetRegionOption::try_from(key).unwrap();
2435        assert_eq!(key, option.to_string());
2436        assert_eq!(
2437            SetRegionOption::Twsc(key.to_string(), String::new()),
2438            SetRegionOption::from(&option)
2439        );
2440    }
2441
2442    #[test]
2443    fn test_from_proto_alter_request() {
2444        RegionAlterRequest::try_from(AlterRequest {
2445            region_id: 0,
2446            schema_version: 1,
2447            kind: None,
2448        })
2449        .unwrap_err();
2450
2451        let request = RegionAlterRequest::try_from(AlterRequest {
2452            region_id: 0,
2453            schema_version: 1,
2454            kind: Some(alter_request::Kind::AddColumns(v1::region::AddColumns {
2455                add_columns: vec![v1::region::AddColumn {
2456                    column_def: Some(RegionColumnDef {
2457                        column_def: Some(ColumnDef {
2458                            name: "a".to_string(),
2459                            data_type: ColumnDataType::String as i32,
2460                            is_nullable: true,
2461                            default_constraint: vec![],
2462                            semantic_type: SemanticType::Field as i32,
2463                            comment: String::new(),
2464                            ..Default::default()
2465                        }),
2466                        column_id: 1,
2467                    }),
2468                    location: Some(v1::AddColumnLocation {
2469                        location_type: LocationType::First as i32,
2470                        after_column_name: String::default(),
2471                    }),
2472                }],
2473            })),
2474        })
2475        .unwrap();
2476
2477        assert_eq!(
2478            request,
2479            RegionAlterRequest {
2480                kind: AlterKind::AddColumns {
2481                    columns: vec![AddColumn {
2482                        column_metadata: ColumnMetadata {
2483                            column_schema: ColumnSchema::new(
2484                                "a",
2485                                ConcreteDataType::string_datatype(),
2486                                true,
2487                            ),
2488                            semantic_type: SemanticType::Field,
2489                            column_id: 1,
2490                        },
2491                        location: Some(AddColumnLocation::First),
2492                    }]
2493                },
2494            }
2495        );
2496
2497        let request = RegionAlterRequest::try_from(AlterRequest {
2498            region_id: 0,
2499            schema_version: 1,
2500            kind: Some(alter_request::Kind::SetJsonSettings(v1::SetJsonSettings {
2501                column_name: "payload".to_string(),
2502                settings: Some(v1::JsonSettings {
2503                    type_hints: vec![v1::JsonTypeHint {
2504                        path: vec!["service".to_string()],
2505                        data_type: ColumnDataType::String as i32,
2506                        datatype_extension: None,
2507                    }],
2508                    max_auto_expanded_paths: Some(10),
2509                }),
2510            })),
2511        })
2512        .unwrap();
2513
2514        let AlterKind::SetJsonSettings {
2515            column_name,
2516            settings,
2517        } = request.kind
2518        else {
2519            unreachable!()
2520        };
2521        assert_eq!("payload", column_name);
2522        assert_eq!(Some(10), settings.max_auto_expanded_paths());
2523        assert_eq!(1, settings.type_hints().len());
2524    }
2525
2526    #[test]
2527    fn test_write_buffer_size_region_options() {
2528        let option = PbOption {
2529            key: WRITE_BUFFER_SIZE_KEY.to_string(),
2530            value: "128MiB".to_string(),
2531        };
2532        assert_eq!(
2533            SetRegionOption::WriteBufferSize(Some(ReadableSize::mb(128))),
2534            SetRegionOption::try_from(&option).unwrap()
2535        );
2536
2537        let option = PbOption {
2538            key: WRITE_BUFFER_SIZE_KEY.to_string(),
2539            value: "invalid".to_string(),
2540        };
2541        SetRegionOption::try_from(&option).unwrap_err();
2542
2543        // Clearing the option uses UNSET. Empty SET values, including SQL NULL,
2544        // remain invalid because the SQL layer does not distinguish NULL from ''.
2545        let option = PbOption {
2546            key: WRITE_BUFFER_SIZE_KEY.to_string(),
2547            value: String::new(),
2548        };
2549        SetRegionOption::try_from(&option).unwrap_err();
2550
2551        assert_eq!(
2552            UnsetRegionOption::WriteBufferSize,
2553            UnsetRegionOption::try_from(WRITE_BUFFER_SIZE_KEY).unwrap()
2554        );
2555        assert_eq!(
2556            SetRegionOption::WriteBufferSize(None),
2557            SetRegionOption::from(&UnsetRegionOption::WriteBufferSize)
2558        );
2559    }
2560
2561    /// Returns a new region metadata for testing. Metadata:
2562    /// `[(ts, ms, 1), (tag_0, string, 2), (field_0, string, 3), (field_1, bool, 4)]`
2563    fn new_metadata() -> RegionMetadata {
2564        let mut builder = RegionMetadataBuilder::new(RegionId::new(1, 1));
2565        builder
2566            .push_column_metadata(ColumnMetadata {
2567                column_schema: ColumnSchema::new(
2568                    "ts",
2569                    ConcreteDataType::timestamp_millisecond_datatype(),
2570                    false,
2571                ),
2572                semantic_type: SemanticType::Timestamp,
2573                column_id: 1,
2574            })
2575            .push_column_metadata(ColumnMetadata {
2576                column_schema: ColumnSchema::new(
2577                    "tag_0",
2578                    ConcreteDataType::string_datatype(),
2579                    true,
2580                ),
2581                semantic_type: SemanticType::Tag,
2582                column_id: 2,
2583            })
2584            .push_column_metadata(ColumnMetadata {
2585                column_schema: ColumnSchema::new(
2586                    "field_0",
2587                    ConcreteDataType::string_datatype(),
2588                    true,
2589                ),
2590                semantic_type: SemanticType::Field,
2591                column_id: 3,
2592            })
2593            .push_column_metadata(ColumnMetadata {
2594                column_schema: ColumnSchema::new(
2595                    "field_1",
2596                    ConcreteDataType::boolean_datatype(),
2597                    true,
2598                ),
2599                semantic_type: SemanticType::Field,
2600                column_id: 4,
2601            })
2602            .primary_key(vec![2]);
2603        builder.build().unwrap()
2604    }
2605
2606    fn json2_column_schema(name: &str, settings: JsonSettings) -> ColumnSchema {
2607        let mut column_schema =
2608            ColumnSchema::new(name, ConcreteDataType::Json(JsonType::null()), true);
2609        column_schema.with_extension_type(&Json2ExtensionType::new(std::sync::Arc::new(
2610            JsonMetadata::new(settings),
2611        )));
2612        column_schema
2613    }
2614
2615    #[test]
2616    fn test_add_column_validate() {
2617        let metadata = new_metadata();
2618        let add_column = AddColumn {
2619            column_metadata: ColumnMetadata {
2620                column_schema: ColumnSchema::new(
2621                    "tag_1",
2622                    ConcreteDataType::string_datatype(),
2623                    true,
2624                ),
2625                semantic_type: SemanticType::Tag,
2626                column_id: 5,
2627            },
2628            location: None,
2629        };
2630        add_column.validate(&metadata).unwrap();
2631        assert!(add_column.need_alter(&metadata));
2632
2633        // Add not null column.
2634        AddColumn {
2635            column_metadata: ColumnMetadata {
2636                column_schema: ColumnSchema::new(
2637                    "tag_1",
2638                    ConcreteDataType::string_datatype(),
2639                    false,
2640                ),
2641                semantic_type: SemanticType::Tag,
2642                column_id: 5,
2643            },
2644            location: None,
2645        }
2646        .validate(&metadata)
2647        .unwrap_err();
2648
2649        // Add existing column.
2650        let add_column = AddColumn {
2651            column_metadata: ColumnMetadata {
2652                column_schema: ColumnSchema::new(
2653                    "tag_0",
2654                    ConcreteDataType::string_datatype(),
2655                    true,
2656                ),
2657                semantic_type: SemanticType::Tag,
2658                column_id: 2,
2659            },
2660            location: None,
2661        };
2662        add_column.validate(&metadata).unwrap();
2663        assert!(!add_column.need_alter(&metadata));
2664    }
2665
2666    #[test]
2667    fn test_add_duplicate_columns() {
2668        let kind = AlterKind::AddColumns {
2669            columns: vec![
2670                AddColumn {
2671                    column_metadata: ColumnMetadata {
2672                        column_schema: ColumnSchema::new(
2673                            "tag_1",
2674                            ConcreteDataType::string_datatype(),
2675                            true,
2676                        ),
2677                        semantic_type: SemanticType::Tag,
2678                        column_id: 5,
2679                    },
2680                    location: None,
2681                },
2682                AddColumn {
2683                    column_metadata: ColumnMetadata {
2684                        column_schema: ColumnSchema::new(
2685                            "tag_1",
2686                            ConcreteDataType::string_datatype(),
2687                            true,
2688                        ),
2689                        semantic_type: SemanticType::Field,
2690                        column_id: 6,
2691                    },
2692                    location: None,
2693                },
2694            ],
2695        };
2696        let metadata = new_metadata();
2697        kind.validate(&metadata).unwrap();
2698        assert!(kind.need_alter(&metadata));
2699    }
2700
2701    #[test]
2702    fn test_add_existing_column_different_metadata() {
2703        let metadata = new_metadata();
2704
2705        // Add existing column with different id.
2706        let kind = AlterKind::AddColumns {
2707            columns: vec![AddColumn {
2708                column_metadata: ColumnMetadata {
2709                    column_schema: ColumnSchema::new(
2710                        "tag_0",
2711                        ConcreteDataType::string_datatype(),
2712                        true,
2713                    ),
2714                    semantic_type: SemanticType::Tag,
2715                    column_id: 4,
2716                },
2717                location: None,
2718            }],
2719        };
2720        kind.validate(&metadata).unwrap_err();
2721
2722        // Add existing column with different type.
2723        let kind = AlterKind::AddColumns {
2724            columns: vec![AddColumn {
2725                column_metadata: ColumnMetadata {
2726                    column_schema: ColumnSchema::new(
2727                        "tag_0",
2728                        ConcreteDataType::int64_datatype(),
2729                        true,
2730                    ),
2731                    semantic_type: SemanticType::Tag,
2732                    column_id: 2,
2733                },
2734                location: None,
2735            }],
2736        };
2737        kind.validate(&metadata).unwrap_err();
2738
2739        // Add existing column with different name.
2740        let kind = AlterKind::AddColumns {
2741            columns: vec![AddColumn {
2742                column_metadata: ColumnMetadata {
2743                    column_schema: ColumnSchema::new(
2744                        "tag_1",
2745                        ConcreteDataType::string_datatype(),
2746                        true,
2747                    ),
2748                    semantic_type: SemanticType::Tag,
2749                    column_id: 2,
2750                },
2751                location: None,
2752            }],
2753        };
2754        kind.validate(&metadata).unwrap_err();
2755    }
2756
2757    #[test]
2758    fn test_add_existing_column_with_location() {
2759        let metadata = new_metadata();
2760        let kind = AlterKind::AddColumns {
2761            columns: vec![AddColumn {
2762                column_metadata: ColumnMetadata {
2763                    column_schema: ColumnSchema::new(
2764                        "tag_0",
2765                        ConcreteDataType::string_datatype(),
2766                        true,
2767                    ),
2768                    semantic_type: SemanticType::Tag,
2769                    column_id: 2,
2770                },
2771                location: Some(AddColumnLocation::First),
2772            }],
2773        };
2774        kind.validate(&metadata).unwrap_err();
2775    }
2776
2777    #[test]
2778    fn test_validate_drop_column() {
2779        let metadata = new_metadata();
2780        let kind = AlterKind::DropColumns {
2781            names: vec!["xxxx".to_string()],
2782        };
2783        kind.validate(&metadata).unwrap();
2784        assert!(!kind.need_alter(&metadata));
2785
2786        AlterKind::DropColumns {
2787            names: vec!["tag_0".to_string()],
2788        }
2789        .validate(&metadata)
2790        .unwrap_err();
2791
2792        let kind = AlterKind::DropColumns {
2793            names: vec!["field_0".to_string()],
2794        };
2795        kind.validate(&metadata).unwrap();
2796        assert!(kind.need_alter(&metadata));
2797    }
2798
2799    #[test]
2800    fn test_validate_modify_column_type() {
2801        let metadata = new_metadata();
2802        AlterKind::ModifyColumnTypes {
2803            columns: vec![ModifyColumnType {
2804                column_name: "xxxx".to_string(),
2805                target_type: ConcreteDataType::string_datatype(),
2806            }],
2807        }
2808        .validate(&metadata)
2809        .unwrap_err();
2810
2811        AlterKind::ModifyColumnTypes {
2812            columns: vec![ModifyColumnType {
2813                column_name: "field_1".to_string(),
2814                target_type: ConcreteDataType::date_datatype(),
2815            }],
2816        }
2817        .validate(&metadata)
2818        .unwrap_err();
2819
2820        AlterKind::ModifyColumnTypes {
2821            columns: vec![ModifyColumnType {
2822                column_name: "ts".to_string(),
2823                target_type: ConcreteDataType::date_datatype(),
2824            }],
2825        }
2826        .validate(&metadata)
2827        .unwrap_err();
2828
2829        // Time index unit widening is allowed.
2830        let kind = AlterKind::ModifyColumnTypes {
2831            columns: vec![ModifyColumnType {
2832                column_name: "ts".to_string(),
2833                target_type: ConcreteDataType::timestamp_microsecond_datatype(),
2834            }],
2835        };
2836        kind.validate(&metadata).unwrap();
2837        assert!(kind.need_alter(&metadata));
2838
2839        // Narrowing the time index unit is rejected.
2840        let metadata_nano = {
2841            let mut metadata = new_metadata();
2842            for col in metadata.column_metadatas.iter_mut() {
2843                if col.column_schema.name == "ts" {
2844                    col.column_schema.data_type = ConcreteDataType::timestamp_nanosecond_datatype();
2845                }
2846            }
2847            metadata
2848        };
2849        AlterKind::ModifyColumnTypes {
2850            columns: vec![ModifyColumnType {
2851                column_name: "ts".to_string(),
2852                target_type: ConcreteDataType::timestamp_millisecond_datatype(),
2853            }],
2854        }
2855        .validate(&metadata_nano)
2856        .unwrap_err();
2857
2858        // Changing the time index to the same type is a validated no-op, so a
2859        // retried alter procedure (region already altered) succeeds and is
2860        // skipped by `need_alter`.
2861        let same_type = AlterKind::ModifyColumnTypes {
2862            columns: vec![ModifyColumnType {
2863                column_name: "ts".to_string(),
2864                target_type: ConcreteDataType::timestamp_millisecond_datatype(),
2865            }],
2866        };
2867        same_type.validate(&metadata).unwrap();
2868        assert!(!same_type.need_alter(&metadata));
2869
2870        // Changing the time index to a non-timestamp type is rejected.
2871        AlterKind::ModifyColumnTypes {
2872            columns: vec![ModifyColumnType {
2873                column_name: "ts".to_string(),
2874                target_type: ConcreteDataType::string_datatype(),
2875            }],
2876        }
2877        .validate(&metadata)
2878        .unwrap_err();
2879
2880        AlterKind::ModifyColumnTypes {
2881            columns: vec![ModifyColumnType {
2882                column_name: "tag_0".to_string(),
2883                target_type: ConcreteDataType::date_datatype(),
2884            }],
2885        }
2886        .validate(&metadata)
2887        .unwrap_err();
2888
2889        let kind = AlterKind::ModifyColumnTypes {
2890            columns: vec![ModifyColumnType {
2891                column_name: "field_0".to_string(),
2892                target_type: ConcreteDataType::int32_datatype(),
2893            }],
2894        };
2895        kind.validate(&metadata).unwrap();
2896        assert!(kind.need_alter(&metadata));
2897    }
2898
2899    #[test]
2900    fn test_validate_set_json_settings() {
2901        let mut metadata = new_metadata();
2902        let current_schema = json2_column_schema("field_0", JsonSettings::new_v2());
2903        let target_settings = JsonSettings::try_new(vec![], Some(10)).unwrap();
2904        metadata
2905            .column_metadatas
2906            .iter_mut()
2907            .find(|column| column.column_schema.name == "field_0")
2908            .unwrap()
2909            .column_schema = current_schema.clone();
2910
2911        let kind = AlterKind::SetJsonSettings {
2912            column_name: "field_0".to_string(),
2913            settings: target_settings.clone(),
2914        };
2915        kind.validate(&metadata).unwrap();
2916        assert!(kind.need_alter(&metadata));
2917
2918        let no_op = AlterKind::SetJsonSettings {
2919            column_name: "field_0".to_string(),
2920            settings: JsonSettings::new_v2(),
2921        };
2922        no_op.validate(&metadata).unwrap();
2923        assert!(!no_op.need_alter(&metadata));
2924
2925        AlterKind::SetJsonSettings {
2926            column_name: "tag_0".to_string(),
2927            settings: target_settings,
2928        }
2929        .validate(&metadata)
2930        .unwrap_err();
2931    }
2932
2933    #[test]
2934    fn test_validate_add_columns() {
2935        let kind = AlterKind::AddColumns {
2936            columns: vec![
2937                AddColumn {
2938                    column_metadata: ColumnMetadata {
2939                        column_schema: ColumnSchema::new(
2940                            "tag_1",
2941                            ConcreteDataType::string_datatype(),
2942                            true,
2943                        ),
2944                        semantic_type: SemanticType::Tag,
2945                        column_id: 5,
2946                    },
2947                    location: None,
2948                },
2949                AddColumn {
2950                    column_metadata: ColumnMetadata {
2951                        column_schema: ColumnSchema::new(
2952                            "field_2",
2953                            ConcreteDataType::string_datatype(),
2954                            true,
2955                        ),
2956                        semantic_type: SemanticType::Field,
2957                        column_id: 6,
2958                    },
2959                    location: None,
2960                },
2961            ],
2962        };
2963        let request = RegionAlterRequest { kind };
2964        let mut metadata = new_metadata();
2965        metadata.schema_version = 1;
2966        request.validate(&metadata).unwrap();
2967    }
2968
2969    #[test]
2970    fn test_validate_create_region() {
2971        let column_metadatas = vec![
2972            ColumnMetadata {
2973                column_schema: ColumnSchema::new(
2974                    "ts",
2975                    ConcreteDataType::timestamp_millisecond_datatype(),
2976                    false,
2977                ),
2978                semantic_type: SemanticType::Timestamp,
2979                column_id: 1,
2980            },
2981            ColumnMetadata {
2982                column_schema: ColumnSchema::new(
2983                    "tag_0",
2984                    ConcreteDataType::string_datatype(),
2985                    true,
2986                ),
2987                semantic_type: SemanticType::Tag,
2988                column_id: 2,
2989            },
2990            ColumnMetadata {
2991                column_schema: ColumnSchema::new(
2992                    "field_0",
2993                    ConcreteDataType::string_datatype(),
2994                    true,
2995                ),
2996                semantic_type: SemanticType::Field,
2997                column_id: 3,
2998            },
2999        ];
3000        let create = RegionCreateRequest {
3001            engine: "mito".to_string(),
3002            column_metadatas,
3003            primary_key: vec![3, 4],
3004            options: HashMap::new(),
3005            table_dir: "path".to_string(),
3006            path_type: PathType::Bare,
3007            partition_expr_json: Some("".to_string()),
3008            requirements: Default::default(),
3009        };
3010
3011        assert!(create.validate().is_err());
3012    }
3013
3014    #[test]
3015    fn test_parse_create_region_requirements_defaults_to_empty() {
3016        let create = CreateRequest {
3017            region_id: RegionId::new(42, 0).as_u64(),
3018            engine: "mito".to_string(),
3019            column_defs: vec![],
3020            primary_key: vec![],
3021            path: "test".to_string(),
3022            options: HashMap::new(),
3023            partition: None,
3024            requirements: None,
3025        };
3026
3027        let requests =
3028            RegionRequest::try_from_request_body(region_request::Body::Create(create)).unwrap();
3029        let RegionRequest::Create(request) = &requests[0].1 else {
3030            unreachable!()
3031        };
3032
3033        assert_eq!(request.requirements, RegionRequirements::empty());
3034    }
3035
3036    #[test]
3037    fn test_parse_create_region_requirements_from_proto() {
3038        let create = CreateRequest {
3039            region_id: RegionId::new(42, 0).as_u64(),
3040            engine: "mito".to_string(),
3041            column_defs: vec![],
3042            primary_key: vec![],
3043            path: "test".to_string(),
3044            options: HashMap::new(),
3045            partition: None,
3046            requirements: Some(api::v1::region::RegionRequirements {
3047                object_storage: true,
3048            }),
3049        };
3050
3051        let requests =
3052            RegionRequest::try_from_request_body(region_request::Body::Create(create)).unwrap();
3053        let RegionRequest::Create(request) = &requests[0].1 else {
3054            unreachable!()
3055        };
3056
3057        assert_eq!(request.requirements, RegionRequirements::object_storage());
3058    }
3059
3060    #[test]
3061    fn test_parse_region_cleanup_from_proto() {
3062        let clean_up = api::v1::region::CleanUpRequest {
3063            region_id: RegionId::new(42, 3).as_u64(),
3064            engine: "mito".to_string(),
3065            path: "test".to_string(),
3066            options: HashMap::from([("k".to_string(), "v".to_string())]),
3067        };
3068
3069        let requests =
3070            RegionRequest::try_from_request_body(region_request::Body::CleanUp(clean_up)).unwrap();
3071        let RegionRequest::CleanUp(request) = &requests[0].1 else {
3072            unreachable!()
3073        };
3074
3075        assert_eq!(requests[0].0, RegionId::new(42, 3));
3076        assert_eq!(request.engine, "mito");
3077        assert_eq!(request.table_dir, "data/test/42/");
3078        assert_eq!(request.path_type, PathType::Bare);
3079        assert_eq!(request.options.get("k"), Some(&"v".to_string()));
3080    }
3081
3082    #[test]
3083    fn test_validate_modify_column_fulltext_options() {
3084        let kind = AlterKind::SetIndexes {
3085            options: vec![SetIndexOption::Fulltext {
3086                column_name: "tag_0".to_string(),
3087                options: FulltextOptions::new_unchecked(
3088                    true,
3089                    FulltextAnalyzer::Chinese,
3090                    false,
3091                    FulltextBackend::Bloom,
3092                    1000,
3093                    0.01,
3094                ),
3095            }],
3096        };
3097        let request = RegionAlterRequest { kind };
3098        let mut metadata = new_metadata();
3099        metadata.schema_version = 1;
3100        request.validate(&metadata).unwrap();
3101
3102        let kind = AlterKind::UnsetIndexes {
3103            options: vec![UnsetIndexOption::Fulltext {
3104                column_name: "tag_0".to_string(),
3105            }],
3106        };
3107        let request = RegionAlterRequest { kind };
3108        let mut metadata = new_metadata();
3109        metadata.schema_version = 1;
3110        request.validate(&metadata).unwrap();
3111    }
3112
3113    #[test]
3114    fn test_validate_sync_columns() {
3115        let metadata = new_metadata();
3116        let kind = AlterKind::SyncColumns {
3117            column_metadatas: vec![
3118                ColumnMetadata {
3119                    column_schema: ColumnSchema::new(
3120                        "tag_1",
3121                        ConcreteDataType::string_datatype(),
3122                        true,
3123                    ),
3124                    semantic_type: SemanticType::Tag,
3125                    column_id: 5,
3126                },
3127                ColumnMetadata {
3128                    column_schema: ColumnSchema::new(
3129                        "field_2",
3130                        ConcreteDataType::string_datatype(),
3131                        true,
3132                    ),
3133                    semantic_type: SemanticType::Field,
3134                    column_id: 6,
3135                },
3136            ],
3137        };
3138        let err = kind.validate(&metadata).unwrap_err();
3139        assert!(err.to_string().contains("not a primary key"));
3140
3141        // Change the timestamp column name.
3142        let mut column_metadatas_with_different_ts_column = metadata.column_metadatas.clone();
3143        let ts_column = column_metadatas_with_different_ts_column
3144            .iter_mut()
3145            .find(|c| c.semantic_type == SemanticType::Timestamp)
3146            .unwrap();
3147        ts_column.column_schema.name = "ts1".to_string();
3148
3149        let kind = AlterKind::SyncColumns {
3150            column_metadatas: column_metadatas_with_different_ts_column,
3151        };
3152        let err = kind.validate(&metadata).unwrap_err();
3153        assert!(
3154            err.to_string()
3155                .contains("timestamp column ts has different id")
3156        );
3157
3158        // Change the primary key column name.
3159        let mut column_metadatas_with_different_pk_column = metadata.column_metadatas.clone();
3160        let pk_column = column_metadatas_with_different_pk_column
3161            .iter_mut()
3162            .find(|c| c.column_schema.name == "tag_0")
3163            .unwrap();
3164        pk_column.column_id = 100;
3165        let kind = AlterKind::SyncColumns {
3166            column_metadatas: column_metadatas_with_different_pk_column,
3167        };
3168        let err = kind.validate(&metadata).unwrap_err();
3169        assert!(
3170            err.to_string()
3171                .contains("column with same name tag_0 has different id")
3172        );
3173
3174        // Add a new field column.
3175        let mut column_metadatas_with_new_field_column = metadata.column_metadatas.clone();
3176        column_metadatas_with_new_field_column.push(ColumnMetadata {
3177            column_schema: ColumnSchema::new("field_2", ConcreteDataType::string_datatype(), true),
3178            semantic_type: SemanticType::Field,
3179            column_id: 4,
3180        });
3181        let kind = AlterKind::SyncColumns {
3182            column_metadatas: column_metadatas_with_new_field_column,
3183        };
3184        kind.validate(&metadata).unwrap();
3185    }
3186
3187    #[test]
3188    fn test_cast_path_type_to_primitive() {
3189        assert_eq!(PathType::Bare as u8, 0);
3190        assert_eq!(PathType::Data as u8, 1);
3191        assert_eq!(PathType::Metadata as u8, 2);
3192        assert_eq!(
3193            PathType::try_from(PathType::Bare as u8).unwrap(),
3194            PathType::Bare
3195        );
3196        assert_eq!(
3197            PathType::try_from(PathType::Data as u8).unwrap(),
3198            PathType::Data
3199        );
3200        assert_eq!(
3201            PathType::try_from(PathType::Metadata as u8).unwrap(),
3202            PathType::Metadata
3203        );
3204    }
3205}