store_api/

region_request.rs

// Copyright 2023 Greptime Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

use std::collections::HashMap;
use std::fmt::{self, Display};
use std::time::Duration;

use api::helper::{ColumnDataTypeWrapper, from_pb_time_ranges, from_pb_time_unit};
use api::v1::add_column_location::LocationType;
use api::v1::column_def::{
    as_fulltext_option_analyzer, as_fulltext_option_backend, as_skipping_index_type,
};
use api::v1::region::bulk_insert_request::Body;
use api::v1::region::{
    AlterRequest, AlterRequests, BuildIndexRequest, BulkInsertRequest,
    CleanUpRequest as PbCleanUpRequest, CloseRequest, CompactRequest, CreateRequest,
    CreateRequests, DeleteRequests, DropRequest, DropRequests, FlushRequest, InsertRequests,
    OpenRequest, TruncateRequest, alter_request, build_index_request, compact_request,
    region_request, truncate_request,
};
use api::v1::{
    self, Analyzer, ArrowIpc, FulltextBackend as PbFulltextBackend, Option as PbOption, Rows,
    SemanticType, SkippingIndexType as PbSkippingIndexType, WriteHint,
};
use arrow_schema::extension::ExtensionType;
pub use common_base::AffectedRows;
use common_base::readable_size::ReadableSize;
use common_grpc::flight::FlightDecoder;
use common_recordbatch::DfRecordBatch;
use common_time::range::TimestampRange;
use common_time::{TimeToLive, Timestamp};
use datatypes::error::time_index_not_widening_error;
use datatypes::extension::json::Json2ExtensionType;
use datatypes::json::{JsonSettings, JsonTypeHint};
use datatypes::prelude::ConcreteDataType;
use datatypes::schema::{FulltextOptions, SkippingIndexOptions};
use num_enum::TryFromPrimitive;
use serde::{Deserialize, Deserializer, Serialize, Serializer};
use snafu::{OptionExt, ResultExt, ensure};
use strum::{AsRefStr, IntoStaticStr};

use crate::logstore::entry;
use crate::metadata::{
    ColumnMetadata, ConvertTimeRangesSnafu, DecodeProtoSnafu, FlightCodecSnafu,
    InvalidIndexOptionSnafu, InvalidRawRegionRequestSnafu, InvalidRegionRequestSnafu,
    InvalidSetRegionOptionRequestSnafu, InvalidUnsetRegionOptionRequestSnafu, MetadataError,
    RegionMetadata, Result, UnexpectedSnafu,
};
use crate::metric_engine_consts::PHYSICAL_TABLE_METADATA_KEY;
use crate::metrics;
use crate::mito_engine_options::{
    APPEND_MODE_KEY, AUTO_FLUSH_INTERVAL_KEY, COMPACTION_EXPERIMENTAL_OUTPUT_PARALLELISM,
    MAX_ROW_GROUP_ROW_COUNT, MAX_ROW_GROUP_ROW_COUNT_LIMIT, PRESERVE_ROW_SEQUENCE, SKIP_WAL_KEY,
    SST_FORMAT_KEY, TTL_KEY, TWCS_ACTIVE_WINDOW_L1_MERGE_TRIGGER,
    TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM, TWCS_INACTIVE_WINDOW_L1_MERGE_TRIGGER,
    TWCS_INACTIVE_WINDOW_TRIGGER_FILE_NUM, TWCS_MAX_OUTPUT_FILE_SIZE, TWCS_TIME_WINDOW,
    TWCS_TRIGGER_FILE_NUM, WRITE_BUFFER_SIZE_KEY,
};
use crate::path_utils::table_dir;
use crate::storage::{ColumnId, RegionId};

/// The type of path to generate.
#[derive(Debug, Clone, Copy, PartialEq, TryFromPrimitive)]
#[repr(u8)]
pub enum PathType {
    /// A bare path - the original path of an engine.
    ///
    /// The path prefix is `{table_dir}/{table_id}_{region_sequence}/`.
    Bare,
    /// A path for the data region of a metric engine table.
    ///
    /// The path prefix is `{table_dir}/{table_id}_{region_sequence}/data/`.
    Data,
    /// A path for the metadata region of a metric engine table.
    ///
    /// The path prefix is `{table_dir}/{table_id}_{region_sequence}/metadata/`.
    Metadata,
}

#[derive(Debug, IntoStaticStr)]
pub enum BatchRegionDdlRequest {
    Create(Vec<(RegionId, RegionCreateRequest)>),
    Drop(Vec<(RegionId, RegionDropRequest)>),
    Alter(Vec<(RegionId, RegionAlterRequest)>),
}

impl BatchRegionDdlRequest {
    /// Converts [Body](region_request::Body) to [`BatchRegionDdlRequest`].
    pub fn try_from_request_body(body: region_request::Body) -> Result<Option<Self>> {
        match body {
            region_request::Body::Creates(creates) => {
                let requests = creates
                    .requests
                    .into_iter()
                    .map(parse_region_create)
                    .collect::<Result<Vec<_>>>()?;
                Ok(Some(Self::Create(requests)))
            }
            region_request::Body::Drops(drops) => {
                let requests = drops
                    .requests
                    .into_iter()
                    .map(parse_region_drop)
                    .collect::<Result<Vec<_>>>()?;
                Ok(Some(Self::Drop(requests)))
            }
            region_request::Body::Alters(alters) => {
                let requests = alters
                    .requests
                    .into_iter()
                    .map(parse_region_alter)
                    .collect::<Result<Vec<_>>>()?;
                Ok(Some(Self::Alter(requests)))
            }
            _ => Ok(None),
        }
    }

    pub fn request_type(&self) -> &'static str {
        self.into()
    }

    pub fn into_region_requests(self) -> Vec<(RegionId, RegionRequest)> {
        match self {
            Self::Create(requests) => requests
                .into_iter()
                .map(|(region_id, request)| (region_id, RegionRequest::Create(request)))
                .collect(),
            Self::Drop(requests) => requests
                .into_iter()
                .map(|(region_id, request)| (region_id, RegionRequest::Drop(request)))
                .collect(),
            Self::Alter(requests) => requests
                .into_iter()
                .map(|(region_id, request)| (region_id, RegionRequest::Alter(request)))
                .collect(),
        }
    }
}

#[derive(Debug, IntoStaticStr)]
pub enum RegionRequest {
    Put(RegionPutRequest),
    Delete(RegionDeleteRequest),
    Create(RegionCreateRequest),
    Drop(RegionDropRequest),
    Open(RegionOpenRequest),
    CleanUp(RegionCleanUpRequest),
    Close(RegionCloseRequest),
    Alter(RegionAlterRequest),
    Flush(RegionFlushRequest),
    Compact(RegionCompactRequest),
    BuildIndex(RegionBuildIndexRequest),
    Truncate(RegionTruncateRequest),
    Catchup(RegionCatchupRequest),
    BulkInserts(RegionBulkInsertsRequest),
    EnterStaging(EnterStagingRequest),
    ApplyStagingManifest(ApplyStagingManifestRequest),
}

impl RegionRequest {
    /// Convert [Body](region_request::Body) to a group of [RegionRequest] with region id.
    /// Inserts/Deletes request might become multiple requests. Others are one-to-one.
    pub fn try_from_request_body(body: region_request::Body) -> Result<Vec<(RegionId, Self)>> {
        match body {
            region_request::Body::Inserts(inserts) => make_region_puts(inserts),
            region_request::Body::Deletes(deletes) => make_region_deletes(deletes),
            region_request::Body::Create(create) => make_region_create(create),
            region_request::Body::Drop(drop) => make_region_drop(drop),
            region_request::Body::Open(open) => make_region_open(open),
            region_request::Body::CleanUp(clean_up) => make_region_clean_up(clean_up),
            region_request::Body::Close(close) => make_region_close(close),
            region_request::Body::Alter(alter) => make_region_alter(alter),
            region_request::Body::Flush(flush) => make_region_flush(flush),
            region_request::Body::Compact(compact) => make_region_compact(compact),
            region_request::Body::BuildIndex(index) => make_region_build_index(index),
            region_request::Body::Truncate(truncate) => make_region_truncate(truncate),
            region_request::Body::Creates(creates) => make_region_creates(creates),
            region_request::Body::Drops(drops) => make_region_drops(drops),
            region_request::Body::Alters(alters) => make_region_alters(alters),
            region_request::Body::BulkInsert(bulk) => make_region_bulk_inserts(bulk),
            region_request::Body::Sync(_) => UnexpectedSnafu {
                reason: "Sync request should be handled separately by RegionServer",
            }
            .fail(),
            region_request::Body::ListMetadata(_) => UnexpectedSnafu {
                reason: "ListMetadata request should be handled separately by RegionServer",
            }
            .fail(),
            region_request::Body::RemoteDynFilter(_) => UnexpectedSnafu {
                reason: "RemoteDynFilter request should be handled separately by RegionServer",
            }
            .fail(),
            region_request::Body::ApplyStagingManifest(apply) => {
                make_region_apply_staging_manifest(apply)
            }
        }
    }

    /// Returns the type name of the request.
    pub fn request_type(&self) -> &'static str {
        self.into()
    }
}

/// Converts one insert, treating absent rows as a no-op like vector inserts.
pub fn make_region_put(
    request: api::v1::region::InsertRequest,
) -> Option<(RegionId, RegionPutRequest)> {
    request.rows.map(|rows| {
        (
            request.region_id.into(),
            RegionPutRequest {
                rows,
                hint: None,
                skip_wal: request.skip_wal,
                partition_expr_version: request.partition_expr_version.map(|version| version.value),
            },
        )
    })
}

fn make_region_puts(inserts: InsertRequests) -> Result<Vec<(RegionId, RegionRequest)>> {
    Ok(inserts
        .requests
        .into_iter()
        .filter_map(make_region_put)
        .map(|(region_id, request)| (region_id, RegionRequest::Put(request)))
        .collect())
}

/// Converts one delete, treating absent rows as a no-op like vector deletes.
pub fn make_region_delete(
    request: api::v1::region::DeleteRequest,
) -> Option<(RegionId, RegionDeleteRequest)> {
    request.rows.map(|rows| {
        (
            request.region_id.into(),
            RegionDeleteRequest {
                rows,
                hint: None,
                partition_expr_version: request.partition_expr_version.map(|version| version.value),
            },
        )
    })
}

fn make_region_deletes(deletes: DeleteRequests) -> Result<Vec<(RegionId, RegionRequest)>> {
    Ok(deletes
        .requests
        .into_iter()
        .filter_map(make_region_delete)
        .map(|(region_id, request)| (region_id, RegionRequest::Delete(request)))
        .collect())
}

fn parse_region_create(create: CreateRequest) -> Result<(RegionId, RegionCreateRequest)> {
    let column_metadatas = create
        .column_defs
        .into_iter()
        .map(ColumnMetadata::try_from_column_def)
        .collect::<Result<Vec<_>>>()?;
    let region_id = RegionId::from(create.region_id);
    let table_dir = table_dir(&create.path, region_id.table_id());
    let partition_expr_json = create.partition.as_ref().map(|p| p.expression.clone());
    Ok((
        region_id,
        RegionCreateRequest {
            engine: create.engine,
            column_metadatas,
            primary_key: create.primary_key,
            options: create.options,
            table_dir,
            path_type: PathType::Bare,
            partition_expr_json,
            requirements: create
                .requirements
                .map(RegionRequirements::from)
                .unwrap_or_default(),
        },
    ))
}

fn make_region_create(create: CreateRequest) -> Result<Vec<(RegionId, RegionRequest)>> {
    let (region_id, request) = parse_region_create(create)?;
    Ok(vec![(region_id, RegionRequest::Create(request))])
}

fn make_region_creates(creates: CreateRequests) -> Result<Vec<(RegionId, RegionRequest)>> {
    let mut requests = Vec::with_capacity(creates.requests.len());
    for create in creates.requests {
        requests.extend(make_region_create(create)?);
    }
    Ok(requests)
}

fn parse_region_drop(drop: DropRequest) -> Result<(RegionId, RegionDropRequest)> {
    let region_id = drop.region_id.into();
    Ok((
        region_id,
        RegionDropRequest {
            fast_path: drop.fast_path,
            force: drop.force,
            partial_drop: drop.partial_drop,
        },
    ))
}

fn make_region_drop(drop: DropRequest) -> Result<Vec<(RegionId, RegionRequest)>> {
    let (region_id, request) = parse_region_drop(drop)?;
    Ok(vec![(region_id, RegionRequest::Drop(request))])
}

fn make_region_drops(drops: DropRequests) -> Result<Vec<(RegionId, RegionRequest)>> {
    let mut requests = Vec::with_capacity(drops.requests.len());
    for drop in drops.requests {
        requests.extend(make_region_drop(drop)?);
    }
    Ok(requests)
}

fn make_region_open(open: OpenRequest) -> Result<Vec<(RegionId, RegionRequest)>> {
    let region_id = RegionId::from(open.region_id);
    let table_dir = table_dir(&open.path, region_id.table_id());
    Ok(vec![(
        region_id,
        RegionRequest::Open(RegionOpenRequest {
            engine: open.engine,
            table_dir,
            path_type: PathType::Bare,
            options: open.options,
            skip_wal_replay: false,
            checkpoint: None,
            requirements: Default::default(),
        }),
    )])
}

fn make_region_clean_up(clean_up: PbCleanUpRequest) -> Result<Vec<(RegionId, RegionRequest)>> {
    let region_id = RegionId::from(clean_up.region_id);
    let table_dir = table_dir(&clean_up.path, region_id.table_id());
    Ok(vec![(
        region_id,
        RegionRequest::CleanUp(RegionCleanUpRequest {
            engine: clean_up.engine,
            table_dir,
            path_type: PathType::Bare,
            options: clean_up.options,
        }),
    )])
}

fn make_region_close(close: CloseRequest) -> Result<Vec<(RegionId, RegionRequest)>> {
    let region_id = close.region_id.into();
    Ok(vec![(
        region_id,
        RegionRequest::Close(RegionCloseRequest {
            flush_on_close: close.flush_on_close,
        }),
    )])
}

fn parse_region_alter(alter: AlterRequest) -> Result<(RegionId, RegionAlterRequest)> {
    let region_id = alter.region_id.into();
    let request = RegionAlterRequest::try_from(alter)?;
    Ok((region_id, request))
}

fn make_region_alter(alter: AlterRequest) -> Result<Vec<(RegionId, RegionRequest)>> {
    let (region_id, request) = parse_region_alter(alter)?;
    Ok(vec![(region_id, RegionRequest::Alter(request))])
}

fn make_region_alters(alters: AlterRequests) -> Result<Vec<(RegionId, RegionRequest)>> {
    let mut requests = Vec::with_capacity(alters.requests.len());
    for alter in alters.requests {
        requests.extend(make_region_alter(alter)?);
    }
    Ok(requests)
}

fn make_region_flush(flush: FlushRequest) -> Result<Vec<(RegionId, RegionRequest)>> {
    let region_id = flush.region_id.into();
    Ok(vec![(
        region_id,
        RegionRequest::Flush(RegionFlushRequest::default()),
    )])
}

fn make_region_compact(compact: CompactRequest) -> Result<Vec<(RegionId, RegionRequest)>> {
    let region_id = compact.region_id.into();
    let options = compact
        .options
        .unwrap_or(compact_request::Options::Regular(Default::default()));
    // Convert parallelism: a value of 0 indicates no specific parallelism requested (None)
    let parallelism = if compact.parallelism == 0 {
        None
    } else {
        Some(compact.parallelism)
    };
    let time_range = compact
        .time_range
        .map(|range| {
            let time_unit = v1::TimeUnit::try_from(range.time_unit).map_err(|_| {
                InvalidRegionRequestSnafu {
                    region_id,
                    err: format!("invalid compaction time unit: {}", range.time_unit),
                }
                .build()
            })?;
            let time_unit = from_pb_time_unit(time_unit);
            let start = Timestamp::new(range.start, time_unit);
            let end = Timestamp::new(range.end, time_unit);
            ensure!(
                start < end,
                InvalidRegionRequestSnafu {
                    region_id,
                    err: format!(
                        "compaction start time must be earlier than end time: {} >= {}",
                        range.start, range.end
                    ),
                }
            );
            TimestampRange::new(start, end).with_context(|| InvalidRegionRequestSnafu {
                region_id,
                err: "invalid compaction time range".to_string(),
            })
        })
        .transpose()?;
    Ok(vec![(
        region_id,
        RegionRequest::Compact(RegionCompactRequest {
            options,
            parallelism,
            time_range,
        }),
    )])
}

fn make_region_build_index(index: BuildIndexRequest) -> Result<Vec<(RegionId, RegionRequest)>> {
    let region_id = index.region_id.into();
    Ok(vec![(
        region_id,
        RegionRequest::BuildIndex(RegionBuildIndexRequest {
            options: index.options,
        }),
    )])
}

fn make_region_truncate(truncate: TruncateRequest) -> Result<Vec<(RegionId, RegionRequest)>> {
    let region_id = truncate.region_id.into();
    match truncate.kind {
        None => InvalidRawRegionRequestSnafu {
            err: "missing kind in TruncateRequest".to_string(),
        }
        .fail(),
        Some(truncate_request::Kind::All(_)) => Ok(vec![(
            region_id,
            RegionRequest::Truncate(RegionTruncateRequest::All),
        )]),
        Some(truncate_request::Kind::TimeRanges(time_ranges)) => {
            let time_ranges = from_pb_time_ranges(time_ranges).context(ConvertTimeRangesSnafu)?;

            Ok(vec![(
                region_id,
                RegionRequest::Truncate(RegionTruncateRequest::ByTimeRanges { time_ranges }),
            )])
        }
        Some(truncate_request::Kind::Unflushed(_)) => Ok(vec![(
            region_id,
            RegionRequest::Truncate(RegionTruncateRequest::Unflushed),
        )]),
    }
}

/// Convert [BulkInsertRequest] to [RegionRequest] and group by [RegionId].
fn make_region_bulk_inserts(request: BulkInsertRequest) -> Result<Vec<(RegionId, RegionRequest)>> {
    let region_id = request.region_id.into();
    let skip_wal = request.skip_wal;
    let partition_expr_version = request.partition_expr_version.map(|v| v.value);
    let aligned_schema_version = request.aligned_schema_version.map(|v| v.schema_version);
    let Some(Body::ArrowIpc(request)) = request.body else {
        return Ok(vec![]);
    };

    let decoder_timer = metrics::CONVERT_REGION_BULK_REQUEST
        .with_label_values(&["decode"])
        .start_timer();
    let mut decoder =
        FlightDecoder::try_from_schema_bytes(&request.schema).context(FlightCodecSnafu)?;
    let payload = decoder
        .try_decode_record_batch(&request.data_header, &request.payload)
        .context(FlightCodecSnafu)?;
    decoder_timer.observe_duration();
    Ok(vec![(
        region_id,
        RegionRequest::BulkInserts(RegionBulkInsertsRequest {
            region_id,
            payload,
            raw_data: request,
            skip_wal,
            partition_expr_version,
            aligned_schema_version,
        }),
    )])
}

fn make_region_apply_staging_manifest(
    api::v1::region::ApplyStagingManifestRequest {
        region_id,
        partition_expr,
        central_region_id,
        manifest_path,
    }: api::v1::region::ApplyStagingManifestRequest,
) -> Result<Vec<(RegionId, RegionRequest)>> {
    let region_id = region_id.into();
    Ok(vec![(
        region_id,
        RegionRequest::ApplyStagingManifest(ApplyStagingManifestRequest {
            partition_expr,
            central_region_id: central_region_id.into(),
            manifest_path,
        }),
    )])
}

/// Request to put data into a region.
#[derive(Debug)]
pub struct RegionPutRequest {
    /// Rows to put.
    pub rows: Rows,
    /// Write hint.
    pub hint: Option<WriteHint>,
    /// Skip WAL for this insert without changing region options.
    /// Metadata writes must not inherit this option from user inserts.
    pub skip_wal: bool,
    /// Partition expression version for the region.
    pub partition_expr_version: Option<u64>,
}

/// Request to delete data from a region.
#[derive(Debug)]
pub struct RegionDeleteRequest {
    /// Keys to rows to delete.
    ///
    /// Each row only contains primary key columns and a time index column.
    pub rows: Rows,
    /// Write hint.
    pub hint: Option<WriteHint>,
    /// Partition expression version for the region.
    pub partition_expr_version: Option<u64>,
}

#[derive(Debug, Clone)]
pub struct RegionCreateRequest {
    /// Region engine name
    pub engine: String,
    /// Columns in this region.
    pub column_metadatas: Vec<ColumnMetadata>,
    /// Columns in the primary key.
    pub primary_key: Vec<ColumnId>,
    /// Options of the created region.
    pub options: HashMap<String, String>,
    /// Directory for table's data home. Usually is composed by catalog and table id
    pub table_dir: String,
    /// Path type for generating paths
    pub path_type: PathType,
    /// Partition expression JSON from table metadata. Set to empty string for a region without partition.
    /// `Option` to keep compatibility with old clients.
    pub partition_expr_json: Option<String>,
    /// Requirements for creating the region.
    pub requirements: RegionRequirements,
}

impl RegionCreateRequest {
    /// Checks whether the request is valid, returns an error if it is invalid.
    pub fn validate(&self) -> Result<()> {
        // time index must exist
        ensure!(
            self.column_metadatas
                .iter()
                .any(|x| x.semantic_type == SemanticType::Timestamp),
            InvalidRegionRequestSnafu {
                region_id: RegionId::new(0, 0),
                err: "missing timestamp column in create region request".to_string(),
            }
        );

        // build column id to indices
        let mut column_id_to_indices = HashMap::with_capacity(self.column_metadatas.len());
        for (i, c) in self.column_metadatas.iter().enumerate() {
            if let Some(previous) = column_id_to_indices.insert(c.column_id, i) {
                return InvalidRegionRequestSnafu {
                    region_id: RegionId::new(0, 0),
                    err: format!(
                        "duplicate column id {} (at position {} and {}) in create region request",
                        c.column_id, previous, i
                    ),
                }
                .fail();
            }
        }

        // primary key must exist
        for column_id in &self.primary_key {
            ensure!(
                column_id_to_indices.contains_key(column_id),
                InvalidRegionRequestSnafu {
                    region_id: RegionId::new(0, 0),
                    err: format!(
                        "missing primary key column {} in create region request",
                        column_id
                    ),
                }
            );
        }

        Ok(())
    }

    /// Returns true when the region belongs to the metric engine's physical table.
    pub fn is_physical_table(&self) -> bool {
        self.options.contains_key(PHYSICAL_TABLE_METADATA_KEY)
    }
}

#[derive(Debug, Clone)]
pub struct RegionDropRequest {
    /// Enables fast-path drop optimizations for logical regions.
    /// Only applicable to the Metric Engine; ignored by others.
    pub fast_path: bool,

    /// Forces the drop of a physical region and all its associated logical regions.
    /// Only relevant for physical regions managed by the Metric Engine.
    pub force: bool,

    /// If true, indicates that only a portion of the region is being dropped, and files may still be referenced by other regions.
    /// This is used to prevent deletion of files that are still in use by other regions.
    pub partial_drop: bool,
}

/// Requirements for a region request.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(default)]
pub struct RegionRequirements {
    /// Whether the region data must be backed by object storage.
    pub object_storage: bool,
}

impl RegionRequirements {
    /// Returns empty requirements.
    pub fn empty() -> Self {
        Self::default()
    }

    /// Returns requirements for object storage.
    pub fn object_storage() -> Self {
        Self {
            object_storage: true,
        }
    }
}

impl From<api::v1::region::RegionRequirements> for RegionRequirements {
    fn from(value: api::v1::region::RegionRequirements) -> Self {
        Self {
            object_storage: value.object_storage,
        }
    }
}

impl From<RegionRequirements> for api::v1::region::RegionRequirements {
    fn from(value: RegionRequirements) -> Self {
        Self {
            object_storage: value.object_storage,
        }
    }
}

/// Open region request.
#[derive(Debug, Clone)]
pub struct RegionOpenRequest {
    /// Region engine name
    pub engine: String,
    /// Directory for table's data home. Usually is composed by catalog and table id
    pub table_dir: String,
    /// Path type for generating paths
    pub path_type: PathType,
    /// Options of the opened region.
    pub options: HashMap<String, String>,
    /// To skip replaying the WAL.
    pub skip_wal_replay: bool,
    /// Replay checkpoint.
    pub checkpoint: Option<ReplayCheckpoint>,
    /// Requirements for opening the region.
    pub requirements: RegionRequirements,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct ReplayCheckpoint {
    pub entry_id: u64,
    pub metadata_entry_id: Option<u64>,
}

impl RegionOpenRequest {
    /// Returns true when the region belongs to the metric engine's physical table.
    pub fn is_physical_table(&self) -> bool {
        self.options.contains_key(PHYSICAL_TABLE_METADATA_KEY)
    }
}

/// Offline region cleanup request.
#[derive(Debug, Clone)]
pub struct RegionCleanUpRequest {
    /// Region engine name
    pub engine: String,
    /// Directory for table's data home. Usually is composed by catalog and table id
    pub table_dir: String,
    /// Path type for generating paths
    pub path_type: PathType,
    /// Options of the cleaned region.
    pub options: HashMap<String, String>,
}

impl RegionCleanUpRequest {
    /// Returns true when the region belongs to the metric engine's physical table.
    pub fn is_physical_table(&self) -> bool {
        self.options.contains_key(PHYSICAL_TABLE_METADATA_KEY)
    }
}

/// Close region request.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct RegionCloseRequest {
    /// Whether to flush the region before closing it.
    pub flush_on_close: bool,
}

/// Alter metadata of a region.
#[derive(Debug, PartialEq, Eq, Clone)]
pub struct RegionAlterRequest {
    /// Kind of alteration to do.
    pub kind: AlterKind,
}

impl RegionAlterRequest {
    /// Checks whether the request is valid, returns an error if it is invalid.
    pub fn validate(&self, metadata: &RegionMetadata) -> Result<()> {
        self.kind.validate(metadata)?;

        Ok(())
    }

    /// Returns true if we need to apply the request to the region.
    ///
    /// The `request` should be valid.
    pub fn need_alter(&self, metadata: &RegionMetadata) -> bool {
        debug_assert!(self.validate(metadata).is_ok());
        self.kind.need_alter(metadata)
    }
}

impl TryFrom<AlterRequest> for RegionAlterRequest {
    type Error = MetadataError;

    fn try_from(value: AlterRequest) -> Result<Self> {
        let kind = value.kind.context(InvalidRawRegionRequestSnafu {
            err: "missing kind in AlterRequest",
        })?;

        let kind = AlterKind::try_from(kind)?;
        Ok(RegionAlterRequest { kind })
    }
}

/// Kind of the alteration.
#[derive(Debug, PartialEq, Eq, Clone, AsRefStr)]
pub enum AlterKind {
    /// Add columns to the region.
    AddColumns {
        /// Columns to add.
        columns: Vec<AddColumn>,
    },
    /// Drop columns from the region, only fields are allowed to drop.
    DropColumns {
        /// Name of columns to drop.
        names: Vec<String>,
    },
    /// Change columns datatype of the region. Field columns can change to any
    /// Arrow-castable type; the time index column only supports widening its
    /// timestamp unit (e.g. `TimestampMillisecond -> TimestampMicrosecond`),
    /// which is lossless for values that fit the target unit's `i64` range.
    ModifyColumnTypes {
        /// Columns to change.
        columns: Vec<ModifyColumnType>,
    },
    /// Set JSON2 settings of a region column.
    SetJsonSettings {
        /// Column name.
        column_name: String,
        /// Target JSON2 settings.
        settings: JsonSettings,
    },
    /// Set region options.
    SetRegionOptions { options: Vec<SetRegionOption> },
    /// Unset region options.
    UnsetRegionOptions { keys: Vec<UnsetRegionOption> },
    /// Set index options.
    SetIndexes { options: Vec<SetIndexOption> },
    /// Unset index options.
    UnsetIndexes { options: Vec<UnsetIndexOption> },
    /// Drop column default value.
    DropDefaults {
        /// Name of columns to drop.
        names: Vec<String>,
    },
    /// Set column default value.
    SetDefaults {
        /// Columns to change.
        columns: Vec<SetDefault>,
    },
    /// Sync column metadatas.
    SyncColumns {
        column_metadatas: Vec<ColumnMetadata>,
    },
}
#[derive(Debug, PartialEq, Eq, Clone)]
pub struct SetDefault {
    pub name: String,
    pub default_constraint: Vec<u8>,
}

#[derive(Debug, PartialEq, Eq, Clone)]
pub enum SetIndexOption {
    Fulltext {
        column_name: String,
        options: FulltextOptions,
    },
    Inverted {
        column_name: String,
    },
    Skipping {
        column_name: String,
        options: SkippingIndexOptions,
    },
}

impl SetIndexOption {
    /// Returns the column name of the index option.
    pub fn column_name(&self) -> &String {
        match self {
            SetIndexOption::Fulltext { column_name, .. } => column_name,
            SetIndexOption::Inverted { column_name } => column_name,
            SetIndexOption::Skipping { column_name, .. } => column_name,
        }
    }

    /// Returns true if the index option is fulltext.
    pub fn is_fulltext(&self) -> bool {
        match self {
            SetIndexOption::Fulltext { .. } => true,
            SetIndexOption::Inverted { .. } => false,
            SetIndexOption::Skipping { .. } => false,
        }
    }
}

impl TryFrom<v1::SetIndex> for SetIndexOption {
    type Error = MetadataError;

    fn try_from(value: v1::SetIndex) -> Result<Self> {
        let option = value.options.context(InvalidRawRegionRequestSnafu {
            err: "missing options in SetIndex",
        })?;

        let opt = match option {
            v1::set_index::Options::Fulltext(x) => SetIndexOption::Fulltext {
                column_name: x.column_name.clone(),
                options: FulltextOptions::new(
                    x.enable,
                    as_fulltext_option_analyzer(
                        Analyzer::try_from(x.analyzer).context(DecodeProtoSnafu)?,
                    ),
                    x.case_sensitive,
                    as_fulltext_option_backend(
                        PbFulltextBackend::try_from(x.backend).context(DecodeProtoSnafu)?,
                    ),
                    x.granularity as u32,
                    x.false_positive_rate,
                )
                .context(InvalidIndexOptionSnafu)?,
            },
            v1::set_index::Options::Inverted(i) => SetIndexOption::Inverted {
                column_name: i.column_name,
            },
            v1::set_index::Options::Skipping(s) => SetIndexOption::Skipping {
                column_name: s.column_name,
                options: SkippingIndexOptions::new(
                    s.granularity as u32,
                    s.false_positive_rate,
                    as_skipping_index_type(
                        PbSkippingIndexType::try_from(s.skipping_index_type)
                            .context(DecodeProtoSnafu)?,
                    ),
                )
                .context(InvalidIndexOptionSnafu)?,
            },
        };

        Ok(opt)
    }
}

#[derive(Debug, PartialEq, Eq, Clone)]
pub enum UnsetIndexOption {
    Fulltext { column_name: String },
    Inverted { column_name: String },
    Skipping { column_name: String },
}

impl UnsetIndexOption {
    pub fn column_name(&self) -> &String {
        match self {
            UnsetIndexOption::Fulltext { column_name } => column_name,
            UnsetIndexOption::Inverted { column_name } => column_name,
            UnsetIndexOption::Skipping { column_name } => column_name,
        }
    }

    pub fn is_fulltext(&self) -> bool {
        match self {
            UnsetIndexOption::Fulltext { .. } => true,
            UnsetIndexOption::Inverted { .. } => false,
            UnsetIndexOption::Skipping { .. } => false,
        }
    }
}

impl TryFrom<v1::UnsetIndex> for UnsetIndexOption {
    type Error = MetadataError;

    fn try_from(value: v1::UnsetIndex) -> Result<Self> {
        let option = value.options.context(InvalidRawRegionRequestSnafu {
            err: "missing options in UnsetIndex",
        })?;

        let opt = match option {
            v1::unset_index::Options::Fulltext(f) => UnsetIndexOption::Fulltext {
                column_name: f.column_name,
            },
            v1::unset_index::Options::Inverted(i) => UnsetIndexOption::Inverted {
                column_name: i.column_name,
            },
            v1::unset_index::Options::Skipping(s) => UnsetIndexOption::Skipping {
                column_name: s.column_name,
            },
        };

        Ok(opt)
    }
}

impl AlterKind {
    /// Returns an error if the alter kind is invalid.
    ///
    /// It allows adding column if not exists and dropping column if exists.
    pub fn validate(&self, metadata: &RegionMetadata) -> Result<()> {
        match self {
            AlterKind::AddColumns { columns } => {
                for col_to_add in columns {
                    col_to_add.validate(metadata)?;
                }
            }
            AlterKind::DropColumns { names } => {
                for name in names {
                    Self::validate_column_to_drop(name, metadata)?;
                }
            }
            AlterKind::ModifyColumnTypes { columns } => {
                for col_to_change in columns {
                    col_to_change.validate(metadata)?;
                }
            }
            AlterKind::SetJsonSettings { column_name, .. } => {
                Self::validate_set_json_settings(column_name, metadata)?
            }
            AlterKind::SetRegionOptions { .. } => {}
            AlterKind::UnsetRegionOptions { .. } => {}
            AlterKind::SetIndexes { options } => {
                for option in options {
                    Self::validate_column_alter_index_option(
                        option.column_name(),
                        metadata,
                        option.is_fulltext(),
                    )?;
                }
            }
            AlterKind::UnsetIndexes { options } => {
                for option in options {
                    Self::validate_column_alter_index_option(
                        option.column_name(),
                        metadata,
                        option.is_fulltext(),
                    )?;
                }
            }
            AlterKind::DropDefaults { names } => {
                names
                    .iter()
                    .try_for_each(|name| Self::validate_column_existence(name, metadata))?;
            }
            AlterKind::SetDefaults { columns } => {
                columns
                    .iter()
                    .try_for_each(|col| Self::validate_column_existence(&col.name, metadata))?;
            }
            AlterKind::SyncColumns { column_metadatas } => {
                let new_primary_keys = column_metadatas
                    .iter()
                    .filter(|c| c.semantic_type == SemanticType::Tag)
                    .map(|c| (c.column_schema.name.as_str(), c.column_id))
                    .collect::<HashMap<_, _>>();

                let old_primary_keys = metadata
                    .column_metadatas
                    .iter()
                    .filter(|c| c.semantic_type == SemanticType::Tag)
                    .map(|c| (c.column_schema.name.as_str(), c.column_id));

                for (name, id) in old_primary_keys {
                    let primary_key =
                        new_primary_keys
                            .get(name)
                            .with_context(|| InvalidRegionRequestSnafu {
                                region_id: metadata.region_id,
                                err: format!("column {} is not a primary key", name),
                            })?;

                    ensure!(
                        *primary_key == id,
                        InvalidRegionRequestSnafu {
                            region_id: metadata.region_id,
                            err: format!(
                                "column with same name {} has different id, existing: {}, got: {}",
                                name, id, primary_key
                            ),
                        }
                    );
                }

                let new_ts_column = column_metadatas
                    .iter()
                    .find(|c| c.semantic_type == SemanticType::Timestamp)
                    .map(|c| (c.column_schema.name.as_str(), c.column_id))
                    .context(InvalidRegionRequestSnafu {
                        region_id: metadata.region_id,
                        err: "timestamp column not found",
                    })?;

                // Safety: timestamp column must exist.
                let old_ts_column = metadata
                    .column_metadatas
                    .iter()
                    .find(|c| c.semantic_type == SemanticType::Timestamp)
                    .map(|c| (c.column_schema.name.as_str(), c.column_id))
                    .unwrap();

                ensure!(
                    new_ts_column == old_ts_column,
                    InvalidRegionRequestSnafu {
                        region_id: metadata.region_id,
                        err: format!(
                            "timestamp column {} has different id, existing: {}, got: {}",
                            old_ts_column.0, old_ts_column.1, new_ts_column.1
                        ),
                    }
                );
            }
        }
        Ok(())
    }

    /// Returns true if we need to apply the alteration to the region.
    pub fn need_alter(&self, metadata: &RegionMetadata) -> bool {
        debug_assert!(self.validate(metadata).is_ok());
        match self {
            AlterKind::AddColumns { columns } => columns
                .iter()
                .any(|col_to_add| col_to_add.need_alter(metadata)),
            AlterKind::DropColumns { names } => names
                .iter()
                .any(|name| metadata.column_by_name(name).is_some()),
            AlterKind::ModifyColumnTypes { columns } => columns
                .iter()
                .any(|col_to_change| col_to_change.need_alter(metadata)),
            AlterKind::SetJsonSettings {
                column_name,
                settings,
            } => metadata.column_by_name(column_name).is_some_and(|col| {
                col.column_schema
                    .extension_type::<Json2ExtensionType>()
                    .ok()
                    .flatten()
                    .is_none_or(|extension| {
                        !extension.metadata().json_settings().equivalent(settings)
                    })
            }),
            AlterKind::SetRegionOptions { .. } => true,
            AlterKind::UnsetRegionOptions { .. } => true,
            AlterKind::SetIndexes { options, .. } => options
                .iter()
                .any(|option| metadata.column_by_name(option.column_name()).is_some()),
            AlterKind::UnsetIndexes { options } => options
                .iter()
                .any(|option| metadata.column_by_name(option.column_name()).is_some()),
            AlterKind::DropDefaults { names } => names
                .iter()
                .any(|name| metadata.column_by_name(name).is_some()),

            AlterKind::SetDefaults { columns } => columns
                .iter()
                .any(|x| metadata.column_by_name(&x.name).is_some()),
            AlterKind::SyncColumns { column_metadatas } => {
                metadata.column_metadatas != *column_metadatas
            }
        }
    }

    /// Returns an error if the column to drop is invalid.
    fn validate_column_to_drop(name: &str, metadata: &RegionMetadata) -> Result<()> {
        let Some(column) = metadata.column_by_name(name) else {
            return Ok(());
        };
        ensure!(
            column.semantic_type == SemanticType::Field,
            InvalidRegionRequestSnafu {
                region_id: metadata.region_id,
                err: format!("column {} is not a field and could not be dropped", name),
            }
        );
        Ok(())
    }

    /// Returns an error if the column's alter index option is invalid.
    fn validate_column_alter_index_option(
        column_name: &String,
        metadata: &RegionMetadata,
        is_fulltext: bool,
    ) -> Result<()> {
        let column = metadata
            .column_by_name(column_name)
            .context(InvalidRegionRequestSnafu {
                region_id: metadata.region_id,
                err: format!("column {} not found", column_name),
            })?;

        if is_fulltext {
            ensure!(
                column.column_schema.data_type.is_string(),
                InvalidRegionRequestSnafu {
                    region_id: metadata.region_id,
                    err: format!(
                        "cannot change alter index options for non-string column {}",
                        column_name
                    ),
                }
            );
        }

        Ok(())
    }

    /// Returns an error if the column isn't exist.
    fn validate_column_existence(column_name: &String, metadata: &RegionMetadata) -> Result<()> {
        metadata
            .column_by_name(column_name)
            .context(InvalidRegionRequestSnafu {
                region_id: metadata.region_id,
                err: format!("column {} not found", column_name),
            })?;

        Ok(())
    }

    fn validate_set_json_settings(col_name: &String, metadata: &RegionMetadata) -> Result<()> {
        let region_id = metadata.region_id;

        let col = metadata
            .column_by_name(col_name)
            .with_context(|| InvalidRegionRequestSnafu {
                region_id,
                err: format!("column {} not found", col_name),
            })?;

        ensure!(
            col.semantic_type == SemanticType::Field,
            InvalidRegionRequestSnafu {
                region_id,
                err: format!("column {} is not a field column", col_name),
            }
        );
        ensure!(
            col.column_schema.data_type.is_json2(),
            InvalidRegionRequestSnafu {
                region_id,
                err: format!("column {} is not a JSON2 column", col_name),
            }
        );

        Ok(())
    }
}

impl TryFrom<alter_request::Kind> for AlterKind {
    type Error = MetadataError;

    fn try_from(kind: alter_request::Kind) -> Result<Self> {
        let alter_kind = match kind {
            alter_request::Kind::AddColumns(x) => {
                let columns = x
                    .add_columns
                    .into_iter()
                    .map(|x| x.try_into())
                    .collect::<Result<Vec<_>>>()?;
                AlterKind::AddColumns { columns }
            }
            alter_request::Kind::ModifyColumnTypes(x) => {
                let columns = x
                    .modify_column_types
                    .into_iter()
                    .map(|x| x.into())
                    .collect::<Vec<_>>();
                AlterKind::ModifyColumnTypes { columns }
            }
            alter_request::Kind::SetJsonSettings(x) => {
                let settings = x.settings.context(InvalidRawRegionRequestSnafu {
                    err: "missing settings in SetJsonSettings",
                })?;
                AlterKind::SetJsonSettings {
                    column_name: x.column_name,
                    settings: json_settings_from_proto(settings)?,
                }
            }
            alter_request::Kind::DropColumns(x) => {
                let names = x.drop_columns.into_iter().map(|x| x.name).collect();
                AlterKind::DropColumns { names }
            }
            alter_request::Kind::SetTableOptions(options) => AlterKind::SetRegionOptions {
                options: options
                    .table_options
                    .iter()
                    .map(TryFrom::try_from)
                    .collect::<Result<Vec<_>>>()?,
            },
            alter_request::Kind::UnsetTableOptions(options) => AlterKind::UnsetRegionOptions {
                keys: options
                    .keys
                    .iter()
                    .map(|key| UnsetRegionOption::try_from(key.as_str()))
                    .collect::<Result<Vec<_>>>()?,
            },
            alter_request::Kind::SetIndex(o) => AlterKind::SetIndexes {
                options: vec![SetIndexOption::try_from(o)?],
            },
            alter_request::Kind::UnsetIndex(o) => AlterKind::UnsetIndexes {
                options: vec![UnsetIndexOption::try_from(o)?],
            },
            alter_request::Kind::SetIndexes(o) => AlterKind::SetIndexes {
                options: o
                    .set_indexes
                    .into_iter()
                    .map(SetIndexOption::try_from)
                    .collect::<Result<Vec<_>>>()?,
            },
            alter_request::Kind::UnsetIndexes(o) => AlterKind::UnsetIndexes {
                options: o
                    .unset_indexes
                    .into_iter()
                    .map(UnsetIndexOption::try_from)
                    .collect::<Result<Vec<_>>>()?,
            },
            alter_request::Kind::DropDefaults(x) => AlterKind::DropDefaults {
                names: x.drop_defaults.into_iter().map(|x| x.column_name).collect(),
            },
            alter_request::Kind::SetDefaults(x) => AlterKind::SetDefaults {
                columns: x
                    .set_defaults
                    .into_iter()
                    .map(|x| {
                        Ok(SetDefault {
                            name: x.column_name,
                            default_constraint: x.default_constraint.clone(),
                        })
                    })
                    .collect::<Result<Vec<_>>>()?,
            },
            alter_request::Kind::SyncColumns(x) => AlterKind::SyncColumns {
                column_metadatas: x
                    .column_defs
                    .into_iter()
                    .map(ColumnMetadata::try_from_column_def)
                    .collect::<Result<Vec<_>>>()?,
            },
        };

        Ok(alter_kind)
    }
}

/// Adds a column.
#[derive(Debug, PartialEq, Eq, Clone)]
pub struct AddColumn {
    /// Metadata of the column to add.
    pub column_metadata: ColumnMetadata,
    /// Location to add the column. If location is None, the region adds
    /// the column to the last.
    pub location: Option<AddColumnLocation>,
}

impl AddColumn {
    /// Returns an error if the column to add is invalid.
    ///
    /// It allows adding existing columns. However, the existing column must have the same metadata
    /// and the location must be None.
    pub fn validate(&self, metadata: &RegionMetadata) -> Result<()> {
        ensure!(
            self.column_metadata.column_schema.is_nullable()
                || self
                    .column_metadata
                    .column_schema
                    .default_constraint()
                    .is_some(),
            InvalidRegionRequestSnafu {
                region_id: metadata.region_id,
                err: format!(
                    "no default value for column {}",
                    self.column_metadata.column_schema.name
                ),
            }
        );

        if let Some(existing_column) =
            metadata.column_by_name(&self.column_metadata.column_schema.name)
        {
            // If the column already exists.
            ensure!(
                *existing_column == self.column_metadata,
                InvalidRegionRequestSnafu {
                    region_id: metadata.region_id,
                    err: format!(
                        "column {} already exists with different metadata, existing: {:?}, got: {:?}",
                        self.column_metadata.column_schema.name,
                        existing_column,
                        self.column_metadata,
                    ),
                }
            );
            ensure!(
                self.location.is_none(),
                InvalidRegionRequestSnafu {
                    region_id: metadata.region_id,
                    err: format!(
                        "column {} already exists, but location is specified",
                        self.column_metadata.column_schema.name
                    ),
                }
            );
        }

        if let Some(existing_column) = metadata.column_by_id(self.column_metadata.column_id) {
            // Ensures the existing column has the same name.
            ensure!(
                existing_column.column_schema.name == self.column_metadata.column_schema.name,
                InvalidRegionRequestSnafu {
                    region_id: metadata.region_id,
                    err: format!(
                        "column id {} already exists with different name {}",
                        self.column_metadata.column_id, existing_column.column_schema.name
                    ),
                }
            );
        }

        Ok(())
    }

    /// Returns true if no column to add to the region.
    pub fn need_alter(&self, metadata: &RegionMetadata) -> bool {
        debug_assert!(self.validate(metadata).is_ok());
        metadata
            .column_by_name(&self.column_metadata.column_schema.name)
            .is_none()
    }
}

impl TryFrom<v1::region::AddColumn> for AddColumn {
    type Error = MetadataError;

    fn try_from(add_column: v1::region::AddColumn) -> Result<Self> {
        let column_def = add_column
            .column_def
            .context(InvalidRawRegionRequestSnafu {
                err: "missing column_def in AddColumn",
            })?;

        let column_metadata = ColumnMetadata::try_from_column_def(column_def)?;
        let location = add_column
            .location
            .map(AddColumnLocation::try_from)
            .transpose()?;

        Ok(AddColumn {
            column_metadata,
            location,
        })
    }
}

/// Location to add a column.
#[derive(Debug, PartialEq, Eq, Clone)]
pub enum AddColumnLocation {
    /// Add the column to the first position of columns.
    First,
    /// Add the column after specific column.
    After {
        /// Add the column after this column.
        column_name: String,
    },
}

impl TryFrom<v1::AddColumnLocation> for AddColumnLocation {
    type Error = MetadataError;

    fn try_from(location: v1::AddColumnLocation) -> Result<Self> {
        let location_type = LocationType::try_from(location.location_type)
            .map_err(|e| InvalidRawRegionRequestSnafu { err: e.to_string() }.build())?;
        let add_column_location = match location_type {
            LocationType::First => AddColumnLocation::First,
            LocationType::After => AddColumnLocation::After {
                column_name: location.after_column_name,
            },
        };

        Ok(add_column_location)
    }
}

/// Change a column's datatype.
#[derive(Debug, PartialEq, Eq, Clone)]
pub struct ModifyColumnType {
    /// Schema of the column to modify.
    pub column_name: String,
    /// Column will be changed to this type.
    pub target_type: ConcreteDataType,
}

impl ModifyColumnType {
    /// Returns an error if the column's datatype to change is invalid.
    pub fn validate(&self, metadata: &RegionMetadata) -> Result<()> {
        let column_meta = metadata
            .column_by_name(&self.column_name)
            .with_context(|| InvalidRegionRequestSnafu {
                region_id: metadata.region_id,
                err: format!("column {} not found", self.column_name),
            })?;

        match column_meta.semantic_type {
            SemanticType::Field => {
                ensure!(
                    column_meta
                        .column_schema
                        .data_type
                        .can_arrow_type_cast_to(&self.target_type),
                    InvalidRegionRequestSnafu {
                        region_id: metadata.region_id,
                        err: format!(
                            "column '{}' cannot be cast automatically to type '{}'",
                            self.column_name, self.target_type
                        ),
                    }
                );
            }
            // The time index column only supports widening its timestamp
            // unit; historical SST data is cast to the new unit on read.
            // A same-type change validates so a retried alter procedure is a
            // no-op instead of failing the retry forever.
            SemanticType::Timestamp => {
                ensure!(
                    column_meta.column_schema.data_type == self.target_type
                        || column_meta
                            .column_schema
                            .data_type
                            .is_timestamp_unit_widening_to(&self.target_type),
                    InvalidRegionRequestSnafu {
                        region_id: metadata.region_id,
                        err: time_index_not_widening_error(
                            &column_meta.column_schema.name,
                            &column_meta.column_schema.data_type,
                            &self.target_type,
                        ),
                    }
                );
            }
            SemanticType::Tag => {
                return InvalidRegionRequestSnafu {
                    region_id: metadata.region_id,
                    err: format!(
                        "tag column '{}' cannot change type, it is part of the primary key",
                        self.column_name
                    ),
                }
                .fail();
            }
        }

        Ok(())
    }

    /// Returns true if no column's datatype to change to the region.
    /// A column already in the target type needs no alteration, so a retried
    /// alter is a successful no-op.
    pub fn need_alter(&self, metadata: &RegionMetadata) -> bool {
        debug_assert!(self.validate(metadata).is_ok());
        metadata
            .column_by_name(&self.column_name)
            .is_some_and(|column| column.column_schema.data_type != self.target_type)
    }
}

impl From<v1::ModifyColumnType> for ModifyColumnType {
    fn from(modify_column_type: v1::ModifyColumnType) -> Self {
        let target_type = ColumnDataTypeWrapper::new(
            modify_column_type.target_type(),
            modify_column_type.target_type_extension,
        )
        .into();

        ModifyColumnType {
            column_name: modify_column_type.column_name,
            target_type,
        }
    }
}

fn json_settings_from_proto(settings: v1::JsonSettings) -> Result<JsonSettings> {
    let type_hints = settings
        .type_hints
        .into_iter()
        .map(|hint| {
            let wrapper = ColumnDataTypeWrapper::try_new(hint.data_type, hint.datatype_extension)
                .map_err(|err| {
                InvalidRawRegionRequestSnafu {
                    err: err.to_string(),
                }
                .build()
            })?;
            let data_type = ConcreteDataType::from(wrapper);

            Ok(JsonTypeHint {
                path: hint.path,
                data_type,
                inverted_index: hint.inverted_index,
            })
        })
        .collect::<Result<Vec<_>>>()?;

    JsonSettings::try_new(type_hints, settings.max_auto_expanded_paths).map_err(|err| {
        InvalidRawRegionRequestSnafu {
            err: err.to_string(),
        }
        .build()
    })
}

/// Region option changes used by ALTER requests.
///
/// This type is serialized for request persistence. Keep future changes backward
/// compatible with previously serialized variants.
#[derive(Debug, Eq, PartialEq, Clone)]
pub enum SetRegionOption {
    WriteBufferSize(Option<ReadableSize>),
    Ttl(Option<TimeToLive>),
    // Modifying TwscOptions with values as (option name, new value).
    Twsc(String, String),
    // Modifying the SST format.
    Format(String),
    // Modifying the append mode.
    AppendMode(bool),
    // Modifying the per-region auto flush interval override.
    AutoFlushInterval(Option<Duration>),
    // Modifying the max number of rows in a parquet row group.
    MaxRowGroupRowCount(Option<usize>),
    PreserveRowSequence(bool),
    // Whether to skip writing new WAL entries.
    SkipWal(bool),
    /// Overrides per-output compaction parallelism; None restores the engine default.
    CompactionOutputParallelism(Option<usize>),
}

#[derive(Serialize, Deserialize)]
enum SetRegionOptionSerde {
    WriteBufferSize(Option<ReadableSize>),
    Ttl(Option<TimeToLive>),
    Twsc(String, String),
    Format(String),
    AppendMode(bool),
    AutoFlushInterval(Option<Duration>),
    MaxRowGroupRowCount(Option<usize>),
    PreserveRowSequence(bool),
    SkipWal(bool),
    CompactionOutputParallelism(Option<usize>),
}

impl Serialize for SetRegionOption {
    fn serialize<S>(&self, serializer: S) -> std::result::Result<S::Ok, S::Error>
    where
        S: Serializer,
    {
        // Older binaries deserialize the disable request as a unit variant.
        if matches!(self, Self::SkipWal(true)) {
            return serializer.serialize_unit_variant("SetRegionOption", 8, "SkipWal");
        }

        let option = match self {
            Self::WriteBufferSize(value) => SetRegionOptionSerde::WriteBufferSize(*value),
            Self::Ttl(value) => SetRegionOptionSerde::Ttl(*value),
            Self::Twsc(key, value) => SetRegionOptionSerde::Twsc(key.clone(), value.clone()),
            Self::Format(value) => SetRegionOptionSerde::Format(value.clone()),
            Self::AppendMode(value) => SetRegionOptionSerde::AppendMode(*value),
            Self::AutoFlushInterval(value) => SetRegionOptionSerde::AutoFlushInterval(*value),
            Self::MaxRowGroupRowCount(value) => SetRegionOptionSerde::MaxRowGroupRowCount(*value),
            Self::PreserveRowSequence(value) => SetRegionOptionSerde::PreserveRowSequence(*value),
            Self::SkipWal(value) => SetRegionOptionSerde::SkipWal(*value),
            Self::CompactionOutputParallelism(value) => {
                SetRegionOptionSerde::CompactionOutputParallelism(*value)
            }
        };
        option.serialize(serializer)
    }
}

#[derive(Deserialize)]
enum LegacySetRegionOption {
    SkipWal,
}

#[derive(Deserialize)]
#[serde(untagged)]
enum BackwardCompatibleSetRegionOption {
    Current(SetRegionOptionSerde),
    Legacy(LegacySetRegionOption),
}

impl From<SetRegionOptionSerde> for SetRegionOption {
    fn from(option: SetRegionOptionSerde) -> Self {
        match option {
            SetRegionOptionSerde::WriteBufferSize(value) => Self::WriteBufferSize(value),
            SetRegionOptionSerde::Ttl(value) => Self::Ttl(value),
            SetRegionOptionSerde::Twsc(key, value) => Self::Twsc(key, value),
            SetRegionOptionSerde::Format(value) => Self::Format(value),
            SetRegionOptionSerde::AppendMode(value) => Self::AppendMode(value),
            SetRegionOptionSerde::AutoFlushInterval(value) => Self::AutoFlushInterval(value),
            SetRegionOptionSerde::MaxRowGroupRowCount(value) => Self::MaxRowGroupRowCount(value),
            SetRegionOptionSerde::PreserveRowSequence(value) => Self::PreserveRowSequence(value),
            SetRegionOptionSerde::SkipWal(value) => Self::SkipWal(value),
            SetRegionOptionSerde::CompactionOutputParallelism(value) => {
                Self::CompactionOutputParallelism(value)
            }
        }
    }
}

impl<'de> Deserialize<'de> for SetRegionOption {
    fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
    where
        D: Deserializer<'de>,
    {
        Ok(
            match BackwardCompatibleSetRegionOption::deserialize(deserializer)? {
                BackwardCompatibleSetRegionOption::Current(option) => option.into(),
                BackwardCompatibleSetRegionOption::Legacy(LegacySetRegionOption::SkipWal) => {
                    Self::SkipWal(true)
                }
            },
        )
    }
}

impl TryFrom<&PbOption> for SetRegionOption {
    type Error = MetadataError;

    fn try_from(value: &PbOption) -> std::result::Result<Self, Self::Error> {
        let PbOption { key, value } = value;
        match key.as_str() {
            WRITE_BUFFER_SIZE_KEY => {
                let size = value
                    .parse::<ReadableSize>()
                    .map_err(|_| InvalidSetRegionOptionRequestSnafu { key, value }.build())?;
                Ok(Self::WriteBufferSize(Some(size)))
            }
            TTL_KEY => {
                let ttl = TimeToLive::from_humantime_or_str(value)
                    .map_err(|_| InvalidSetRegionOptionRequestSnafu { key, value }.build())?;

                Ok(Self::Ttl(Some(ttl)))
            }
            TWCS_TRIGGER_FILE_NUM
            | TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM
            | TWCS_ACTIVE_WINDOW_L1_MERGE_TRIGGER
            | TWCS_INACTIVE_WINDOW_TRIGGER_FILE_NUM
            | TWCS_INACTIVE_WINDOW_L1_MERGE_TRIGGER
            | TWCS_MAX_OUTPUT_FILE_SIZE
            | TWCS_TIME_WINDOW => Ok(Self::Twsc(key.clone(), value.clone())),
            SST_FORMAT_KEY => Ok(Self::Format(value.clone())),
            APPEND_MODE_KEY => {
                let append_mode = value
                    .parse::<bool>()
                    .map_err(|_| InvalidSetRegionOptionRequestSnafu { key, value }.build())?;
                Ok(Self::AppendMode(append_mode))
            }
            AUTO_FLUSH_INTERVAL_KEY => {
                if value.is_empty() {
                    // SET 'auto_flush_interval' = NULL comes through as an empty
                    // string; treat it as clearing the override (fall back to
                    // the global default), same as Ttl.
                    return Ok(Self::AutoFlushInterval(None));
                }
                let interval = humantime::parse_duration(value)
                    .map_err(|_| InvalidSetRegionOptionRequestSnafu { key, value }.build())?;
                if interval <= Duration::ZERO {
                    return InvalidSetRegionOptionRequestSnafu { key, value }.fail();
                }
                Ok(Self::AutoFlushInterval(Some(interval)))
            }
            COMPACTION_EXPERIMENTAL_OUTPUT_PARALLELISM => {
                if value.is_empty() {
                    return Ok(Self::CompactionOutputParallelism(None));
                }
                let parallelism = value
                    .parse::<usize>()
                    .ok()
                    .filter(|parallelism| *parallelism > 0)
                    .ok_or_else(|| InvalidSetRegionOptionRequestSnafu { key, value }.build())?;
                Ok(Self::CompactionOutputParallelism(Some(parallelism)))
            }
            MAX_ROW_GROUP_ROW_COUNT => {
                if value.is_empty() {
                    return Ok(Self::MaxRowGroupRowCount(None));
                }
                let row_count = value
                    .parse::<usize>()
                    .ok()
                    .filter(|row_count| {
                        *row_count > 0 && *row_count <= MAX_ROW_GROUP_ROW_COUNT_LIMIT
                    })
                    .ok_or_else(|| InvalidSetRegionOptionRequestSnafu { key, value }.build())?;
                Ok(Self::MaxRowGroupRowCount(Some(row_count)))
            }
            PRESERVE_ROW_SEQUENCE => {
                let preserve = value
                    .parse::<bool>()
                    .map_err(|_| InvalidSetRegionOptionRequestSnafu { key, value }.build())?;
                Ok(Self::PreserveRowSequence(preserve))
            }
            SKIP_WAL_KEY => {
                let skip_wal = value
                    .parse::<bool>()
                    .map_err(|_| InvalidSetRegionOptionRequestSnafu { key, value }.build())?;
                Ok(Self::SkipWal(skip_wal))
            }
            _ => InvalidSetRegionOptionRequestSnafu { key, value }.fail(),
        }
    }
}

impl From<&UnsetRegionOption> for SetRegionOption {
    fn from(unset_option: &UnsetRegionOption) -> Self {
        match unset_option {
            UnsetRegionOption::TwcsTriggerFileNum => {
                SetRegionOption::Twsc(unset_option.to_string(), String::new())
            }
            UnsetRegionOption::TwcsActiveWindowTriggerFileNum => {
                SetRegionOption::Twsc(unset_option.to_string(), String::new())
            }
            UnsetRegionOption::TwcsActiveWindowL1MergeTrigger => {
                SetRegionOption::Twsc(unset_option.to_string(), String::new())
            }
            UnsetRegionOption::TwcsInactiveWindowTriggerFileNum => {
                SetRegionOption::Twsc(unset_option.to_string(), String::new())
            }
            UnsetRegionOption::TwcsInactiveWindowL1MergeTrigger => {
                SetRegionOption::Twsc(unset_option.to_string(), String::new())
            }
            UnsetRegionOption::TwcsMaxOutputFileSize => {
                SetRegionOption::Twsc(unset_option.to_string(), String::new())
            }
            UnsetRegionOption::TwcsTimeWindow => {
                SetRegionOption::Twsc(unset_option.to_string(), String::new())
            }
            UnsetRegionOption::Ttl => SetRegionOption::Ttl(Default::default()),
            UnsetRegionOption::MaxRowGroupRowCount => SetRegionOption::MaxRowGroupRowCount(None),
            UnsetRegionOption::WriteBufferSize => SetRegionOption::WriteBufferSize(None),
            UnsetRegionOption::PreserveRowSequence => SetRegionOption::PreserveRowSequence(false),
            UnsetRegionOption::CompactionOutputParallelism => {
                SetRegionOption::CompactionOutputParallelism(None)
            }
        }
    }
}

impl TryFrom<&str> for UnsetRegionOption {
    type Error = MetadataError;

    fn try_from(key: &str) -> Result<Self> {
        match key.to_ascii_lowercase().as_str() {
            TTL_KEY => Ok(Self::Ttl),
            WRITE_BUFFER_SIZE_KEY => Ok(Self::WriteBufferSize),
            TWCS_TRIGGER_FILE_NUM => Ok(Self::TwcsTriggerFileNum),
            TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM => Ok(Self::TwcsActiveWindowTriggerFileNum),
            TWCS_ACTIVE_WINDOW_L1_MERGE_TRIGGER => Ok(Self::TwcsActiveWindowL1MergeTrigger),
            TWCS_INACTIVE_WINDOW_TRIGGER_FILE_NUM => Ok(Self::TwcsInactiveWindowTriggerFileNum),
            TWCS_INACTIVE_WINDOW_L1_MERGE_TRIGGER => Ok(Self::TwcsInactiveWindowL1MergeTrigger),
            TWCS_MAX_OUTPUT_FILE_SIZE => Ok(Self::TwcsMaxOutputFileSize),
            TWCS_TIME_WINDOW => Ok(Self::TwcsTimeWindow),
            MAX_ROW_GROUP_ROW_COUNT => Ok(Self::MaxRowGroupRowCount),
            PRESERVE_ROW_SEQUENCE => Ok(Self::PreserveRowSequence),
            COMPACTION_EXPERIMENTAL_OUTPUT_PARALLELISM => Ok(Self::CompactionOutputParallelism),
            _ => InvalidUnsetRegionOptionRequestSnafu { key }.fail(),
        }
    }
}

#[derive(Debug, Eq, PartialEq, Clone, Serialize, Deserialize)]
pub enum UnsetRegionOption {
    TwcsTriggerFileNum,
    TwcsActiveWindowTriggerFileNum,
    TwcsInactiveWindowTriggerFileNum,
    TwcsInactiveWindowL1MergeTrigger,
    TwcsMaxOutputFileSize,
    TwcsTimeWindow,
    Ttl,
    MaxRowGroupRowCount,
    WriteBufferSize,
    PreserveRowSequence,
    TwcsActiveWindowL1MergeTrigger,
    CompactionOutputParallelism,
}

impl UnsetRegionOption {
    pub fn as_str(&self) -> &str {
        match self {
            Self::Ttl => TTL_KEY,
            Self::WriteBufferSize => WRITE_BUFFER_SIZE_KEY,
            Self::TwcsTriggerFileNum => TWCS_TRIGGER_FILE_NUM,
            Self::TwcsActiveWindowTriggerFileNum => TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM,
            Self::TwcsActiveWindowL1MergeTrigger => TWCS_ACTIVE_WINDOW_L1_MERGE_TRIGGER,
            Self::TwcsInactiveWindowTriggerFileNum => TWCS_INACTIVE_WINDOW_TRIGGER_FILE_NUM,
            Self::TwcsInactiveWindowL1MergeTrigger => TWCS_INACTIVE_WINDOW_L1_MERGE_TRIGGER,
            Self::TwcsMaxOutputFileSize => TWCS_MAX_OUTPUT_FILE_SIZE,
            Self::TwcsTimeWindow => TWCS_TIME_WINDOW,
            Self::MaxRowGroupRowCount => MAX_ROW_GROUP_ROW_COUNT,
            Self::PreserveRowSequence => PRESERVE_ROW_SEQUENCE,
            Self::CompactionOutputParallelism => COMPACTION_EXPERIMENTAL_OUTPUT_PARALLELISM,
        }
    }
}

impl Display for UnsetRegionOption {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "{}", self.as_str())
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub enum RegionFlushReason {
    /// Flush triggered before region migration.
    RegionMigration,
    /// Flush triggered by repartition procedure.
    Repartition,
    /// Flush triggered by remote WAL pruning.
    RemoteWalPrune,
    /// Flush region before closing region.
    Closing,
    /// Flush region before downgrading region.
    Downgrading,
}

#[derive(Debug, Clone, Default)]
pub struct RegionFlushRequest {
    pub row_group_size: Option<usize>,
    pub reason: Option<RegionFlushReason>,
}

#[derive(Debug)]
pub struct RegionCompactRequest {
    pub options: compact_request::Options,
    pub parallelism: Option<u32>,
    pub time_range: Option<TimestampRange>,
}

impl Default for RegionCompactRequest {
    fn default() -> Self {
        Self {
            // Default to regular compaction.
            options: compact_request::Options::Regular(Default::default()),
            parallelism: None,
            time_range: None,
        }
    }
}

#[derive(Debug, Clone, Default)]
pub struct RegionBuildIndexRequest {
    /// The index build mode. Absent options select SST indexes.
    pub options: Option<build_index_request::Options>,
}

/// Truncate region request.
#[derive(Debug)]
pub enum RegionTruncateRequest {
    /// Truncate all data in the region.
    All,
    /// Discard all unflushed data while preserving persisted SST files.
    ///
    /// This destroys the region's in-memory data irreversibly. Persisted SST files and
    /// the writable state are preserved, and the WAL is obsoleted up to the discard point
    /// so a restart won't replay the discarded data.
    ///
    /// An error may be returned after the data has already been discarded, because the
    /// WAL is obsoleted last. Retrying is safe: a request against a region with nothing
    /// left to discard only re-attempts the WAL obsoletion.
    Unflushed,
    ByTimeRanges {
        /// Time ranges to truncate. Both bound are inclusive.
        /// only files that are fully contained in the time range will be truncated.
        /// so no guarantee that all data in the time range will be truncated.
        time_ranges: Vec<(Timestamp, Timestamp)>,
    },
}

/// Catchup region request.
///
/// Makes a readonly region to catch up to leader region changes.
/// There is no effect if it operating on a leader region.
#[derive(Debug, Clone, Copy, Default)]
pub struct RegionCatchupRequest {
    /// Sets it to writable if it's available after it has caught up with all changes.
    pub set_writable: bool,
    /// The `entry_id` that was expected to reply to.
    /// `None` stands replaying to latest.
    pub entry_id: Option<entry::Id>,
    /// Used for metrics metadata region.
    /// The `entry_id` that was expected to reply to.
    /// `None` stands replaying to latest.
    pub metadata_entry_id: Option<entry::Id>,
    /// Replay checkpoint.
    pub checkpoint: Option<ReplayCheckpoint>,
}

#[derive(Debug, Clone)]
pub struct RegionBulkInsertsRequest {
    /// Whether this request should skip WAL.
    pub skip_wal: bool,
    pub region_id: RegionId,
    pub payload: DfRecordBatch,
    pub raw_data: ArrowIpc,
    pub partition_expr_version: Option<u64>,
    pub aligned_schema_version: Option<u64>,
}

impl RegionBulkInsertsRequest {
    pub fn estimated_size(&self) -> usize {
        self.payload.get_array_memory_size()
    }
}

/// Request to stage a region with a new partition directive.
///
/// This request transitions a region into the staging mode.
/// It first flushes the memtable for the old partition expression if it is not
/// empty, then enters the staging mode with the new directive.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum StagingPartitionDirective {
    UpdatePartitionExpr(String),
    RejectAllWrites,
}

impl StagingPartitionDirective {
    /// Returns the partition expression carried by this directive, if any.
    pub fn partition_expr(&self) -> Option<&str> {
        match self {
            Self::UpdatePartitionExpr(expr) => Some(expr),
            Self::RejectAllWrites => None,
        }
    }
}

#[derive(Debug, Clone)]
pub struct EnterStagingRequest {
    /// The staging partition directive of the region.
    pub partition_directive: StagingPartitionDirective,
}

impl EnterStagingRequest {
    /// Builds an enter-staging request with a partition expression directive.
    pub fn with_partition_expr(partition_expr: String) -> Self {
        Self {
            partition_directive: StagingPartitionDirective::UpdatePartitionExpr(partition_expr),
        }
    }
}

/// This request is used as part of the region repartition.
///
/// After a region has entered staging mode with a new partition expression
/// expression) and a separate process (for example, `remap_manifests`) has
/// generated the new file assignments for the staging region, this request
/// applies that generated manifest to the region.
///
/// In practice, this means:
/// - The `partition_expr` identifies the staging partition expression that the manifest
///   was generated for.
/// - `central_region_id` specifies which region holds the staging blob storage
///   where the manifest was written during the `remap_manifests` operation.
/// - `manifest_path` is the relative path within the central region's staging
///   blob storage to fetch the generated manifest.
///
/// It should typically be called **after** the staging region has been
/// initialized by [`EnterStagingRequest`] and the new file layout has been
/// computed, to finalize the repartition operation.
#[derive(Debug, Clone)]
pub struct ApplyStagingManifestRequest {
    /// The partition expression of the staging region.
    pub partition_expr: String,
    /// The region that stores the staging manifests in its staging blob storage.
    pub central_region_id: RegionId,
    /// The relative path to the staging manifest within the central region's
    /// staging blob storage.
    pub manifest_path: String,
}

impl fmt::Display for RegionRequest {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            RegionRequest::Put(_) => write!(f, "Put"),
            RegionRequest::Delete(_) => write!(f, "Delete"),
            RegionRequest::Create(_) => write!(f, "Create"),
            RegionRequest::Drop(_) => write!(f, "Drop"),
            RegionRequest::Open(_) => write!(f, "Open"),
            RegionRequest::CleanUp(_) => write!(f, "CleanUp"),
            RegionRequest::Close(_) => write!(f, "Close"),
            RegionRequest::Alter(_) => write!(f, "Alter"),
            RegionRequest::Flush(_) => write!(f, "Flush"),
            RegionRequest::Compact(_) => write!(f, "Compact"),
            RegionRequest::BuildIndex(_) => write!(f, "BuildIndex"),
            RegionRequest::Truncate(_) => write!(f, "Truncate"),
            RegionRequest::Catchup(_) => write!(f, "Catchup"),
            RegionRequest::BulkInserts(_) => write!(f, "BulkInserts"),
            RegionRequest::EnterStaging(_) => write!(f, "EnterStaging"),
            RegionRequest::ApplyStagingManifest(_) => write!(f, "ApplyStagingManifest"),
        }
    }
}

#[cfg(test)]
mod tests {

    use api::v1::region::RegionColumnDef;
    use api::v1::{ColumnDataType, ColumnDef};
    use common_time::range::TimestampRange;
    use datatypes::extension::json::{Json2ExtensionType, JsonMetadata};
    use datatypes::json::JsonSettings;
    use datatypes::prelude::ConcreteDataType;
    use datatypes::schema::{ColumnSchema, FulltextAnalyzer, FulltextBackend};
    use datatypes::types::JsonType;

    use super::*;
    use crate::metadata::RegionMetadataBuilder;

    #[test]
    fn test_build_index_options_round_trip() {
        use prost::Message;

        for options in [
            None,
            Some(build_index_request::Options::SstIndex(Default::default())),
            Some(build_index_request::Options::SeriesIndex(Default::default())),
        ] {
            let request = BuildIndexRequest {
                region_id: 42,
                options,
            };
            let decoded = BuildIndexRequest::decode(request.encode_to_vec().as_slice()).unwrap();
            let requests = make_region_build_index(decoded).unwrap();
            assert_eq!(1, requests.len());
            assert_eq!(RegionId::from_u64(42), requests[0].0);
            let RegionRequest::BuildIndex(request) = &requests[0].1 else {
                panic!("expected build-index request");
            };
            assert_eq!(options, request.options);
        }
        // The legacy wire message contains only region_id (field 1).
        let legacy = BuildIndexRequest::decode(&[0x08, 42][..]).unwrap();
        assert!(legacy.options.is_none());
    }

    #[test]
    fn test_make_region_puts_preserves_skip_wal() {
        let region_id = RegionId::new(42, 3);
        let rows = Rows::default();
        let requests = make_region_puts(InsertRequests {
            requests: [false, true, false]
                .into_iter()
                .map(|skip_wal| api::v1::region::InsertRequest {
                    region_id: region_id.as_u64(),
                    rows: Some(rows.clone()),
                    partition_expr_version: Some(api::v1::PartitionExprVersion { value: 7 }),
                    skip_wal,
                })
                .collect(),
        })
        .unwrap();

        assert_eq!(3, requests.len());
        for ((id, request), skip_wal) in requests.into_iter().zip([false, true, false]) {
            assert_eq!(region_id, id);
            let RegionRequest::Put(request) = request else {
                panic!("expected a put request");
            };
            assert_eq!(rows, request.rows);
            assert_eq!(skip_wal, request.skip_wal);
            assert_eq!(Some(7), request.partition_expr_version);
            assert!(request.hint.is_none());
        }
    }

    #[test]
    fn test_make_region_compact_with_time_range() {
        let requests = make_region_compact(CompactRequest {
            region_id: 42,
            time_range: Some(api::v1::region::CompactionTimeRange {
                start: 1_000,
                end: 2_000,
                time_unit: api::v1::TimeUnit::Microsecond as i32,
            }),
            ..Default::default()
        })
        .unwrap();

        let RegionRequest::Compact(request) = &requests[0].1 else {
            unreachable!();
        };
        assert_eq!(
            Some(
                TimestampRange::new(
                    Timestamp::new_microsecond(1_000),
                    Timestamp::new_microsecond(2_000),
                )
                .unwrap()
            ),
            request.time_range
        );
    }

    #[test]
    fn test_make_region_truncate_unflushed() {
        let region_id = RegionId::new(42, 3);
        let requests =
            RegionRequest::try_from_request_body(region_request::Body::Truncate(TruncateRequest {
                region_id: region_id.as_u64(),
                kind: Some(truncate_request::Kind::Unflushed(
                    api::v1::region::Unflushed {},
                )),
            }))
            .unwrap();

        assert_eq!(region_id, requests[0].0);
        assert!(matches!(
            requests[0].1,
            RegionRequest::Truncate(RegionTruncateRequest::Unflushed)
        ));
    }

    #[test]
    fn test_make_region_truncate_requires_kind() {
        let error =
            RegionRequest::try_from_request_body(region_request::Body::Truncate(TruncateRequest {
                region_id: RegionId::new(42, 3).as_u64(),
                kind: None,
            }))
            .unwrap_err();

        assert!(
            error
                .to_string()
                .contains("missing kind in TruncateRequest")
        );
    }

    #[test]
    fn test_from_proto_location() {
        let proto_location = v1::AddColumnLocation {
            location_type: LocationType::First as i32,
            after_column_name: String::default(),
        };
        let location = AddColumnLocation::try_from(proto_location).unwrap();
        assert_eq!(location, AddColumnLocation::First);

        let proto_location = v1::AddColumnLocation {
            location_type: 10,
            after_column_name: String::default(),
        };
        AddColumnLocation::try_from(proto_location).unwrap_err();

        let proto_location = v1::AddColumnLocation {
            location_type: LocationType::After as i32,
            after_column_name: "a".to_string(),
        };
        let location = AddColumnLocation::try_from(proto_location).unwrap();
        assert_eq!(
            location,
            AddColumnLocation::After {
                column_name: "a".to_string()
            }
        );
    }

    #[test]
    fn test_from_none_proto_add_column() {
        AddColumn::try_from(v1::region::AddColumn {
            column_def: None,
            location: None,
        })
        .unwrap_err();
    }

    #[test]
    fn test_compaction_output_parallelism_set_unset_and_serde() {
        for (raw, expected) in [("", None), ("1", Some(1)), ("4", Some(4))] {
            let option = SetRegionOption::try_from(&PbOption {
                key: COMPACTION_EXPERIMENTAL_OUTPUT_PARALLELISM.to_string(),
                value: raw.to_string(),
            })
            .unwrap();
            assert_eq!(
                SetRegionOption::CompactionOutputParallelism(expected),
                option
            );
            let json = serde_json::to_string(&option).unwrap();
            assert_eq!(
                option,
                serde_json::from_str::<SetRegionOption>(&json).unwrap()
            );
        }
        for raw in ["0", "-1", "1.5", "invalid", "18446744073709551616"] {
            assert!(
                SetRegionOption::try_from(&PbOption {
                    key: COMPACTION_EXPERIMENTAL_OUTPUT_PARALLELISM.to_string(),
                    value: raw.to_string(),
                })
                .is_err()
            );
        }
        let unset =
            UnsetRegionOption::try_from(COMPACTION_EXPERIMENTAL_OUTPUT_PARALLELISM).unwrap();
        assert_eq!(COMPACTION_EXPERIMENTAL_OUTPUT_PARALLELISM, unset.as_str());
        assert_eq!(
            SetRegionOption::CompactionOutputParallelism(None),
            (&unset).into()
        );
    }

    #[test]
    fn test_set_region_option_auto_flush_interval_try_from() {
        use std::time::Duration;

        // Valid duration
        let pb = PbOption {
            key: "auto_flush_interval".to_string(),
            value: "5m".to_string(),
        };
        let opt = SetRegionOption::try_from(&pb).unwrap();
        assert_eq!(
            opt,
            SetRegionOption::AutoFlushInterval(Some(Duration::from_secs(300)))
        );

        // Empty value clears the override (mirrors Ttl's behaviour for `SET key = NULL`).
        let pb = PbOption {
            key: "auto_flush_interval".to_string(),
            value: String::new(),
        };
        let opt = SetRegionOption::try_from(&pb).unwrap();
        assert_eq!(opt, SetRegionOption::AutoFlushInterval(None));

        // Zero is rejected up front (engine invariant).
        let pb = PbOption {
            key: "auto_flush_interval".to_string(),
            value: "0s".to_string(),
        };
        assert!(SetRegionOption::try_from(&pb).is_err());

        // Garbage value is rejected.
        let pb = PbOption {
            key: "auto_flush_interval".to_string(),
            value: "not_a_duration".to_string(),
        };
        assert!(SetRegionOption::try_from(&pb).is_err());
    }

    #[test]
    fn test_set_region_option_skip_wal_try_from() {
        let pb = PbOption {
            key: SKIP_WAL_KEY.to_string(),
            value: "true".to_string(),
        };
        assert_eq!(
            SetRegionOption::SkipWal(true),
            SetRegionOption::try_from(&pb).unwrap()
        );

        let pb = PbOption {
            key: SKIP_WAL_KEY.to_string(),
            value: "false".to_string(),
        };
        assert_eq!(
            SetRegionOption::SkipWal(false),
            SetRegionOption::try_from(&pb).unwrap()
        );

        for value in ["", "invalid"] {
            let pb = PbOption {
                key: SKIP_WAL_KEY.to_string(),
                value: value.to_string(),
            };
            assert!(SetRegionOption::try_from(&pb).is_err());
        }

        assert!(UnsetRegionOption::try_from(SKIP_WAL_KEY).is_err());
    }

    #[test]
    fn test_set_region_option_skip_wal_serde_compatibility() {
        let legacy = serde_json::from_str::<SetRegionOption>(r#""SkipWal""#).unwrap();
        assert_eq!(SetRegionOption::SkipWal(true), legacy);

        check_set_region_option_skip_wal_serde_compatibility(false);
        check_set_region_option_skip_wal_serde_compatibility(true);
    }

    fn check_set_region_option_skip_wal_serde_compatibility(skip_wal: bool) {
        let option = SetRegionOption::SkipWal(skip_wal);
        let serialized = serde_json::to_string(&option).unwrap();
        let expected = if skip_wal {
            r#""SkipWal""#
        } else {
            r#"{"SkipWal":false}"#
        };
        assert_eq!(expected, serialized);
        assert_eq!(
            option,
            serde_json::from_str::<SetRegionOption>(&serialized).unwrap()
        );
        if skip_wal {
            assert!(serde_json::from_str::<LegacySetRegionOption>(&serialized).is_ok());
            assert_eq!(
                option,
                serde_json::from_str::<SetRegionOption>(r#"{"SkipWal":true}"#).unwrap()
            );
        }
    }

    #[test]
    fn test_set_region_option_max_row_group_row_count_try_from() {
        let pb = PbOption {
            key: MAX_ROW_GROUP_ROW_COUNT.to_string(),
            value: "512".to_string(),
        };
        assert_eq!(
            SetRegionOption::MaxRowGroupRowCount(Some(512)),
            SetRegionOption::try_from(&pb).unwrap()
        );

        let pb = PbOption {
            key: MAX_ROW_GROUP_ROW_COUNT.to_string(),
            value: String::new(),
        };
        assert_eq!(
            SetRegionOption::MaxRowGroupRowCount(None),
            SetRegionOption::try_from(&pb).unwrap()
        );

        for value in [
            "0".to_string(),
            (MAX_ROW_GROUP_ROW_COUNT_LIMIT + 1).to_string(),
            "invalid".to_string(),
        ] {
            let pb = PbOption {
                key: MAX_ROW_GROUP_ROW_COUNT.to_string(),
                value,
            };
            assert!(SetRegionOption::try_from(&pb).is_err());
        }

        assert_eq!(
            UnsetRegionOption::MaxRowGroupRowCount,
            UnsetRegionOption::try_from(MAX_ROW_GROUP_ROW_COUNT).unwrap()
        );
    }

    #[test]
    fn test_set_region_option_preserve_row_sequence_try_from() {
        for (value, expected) in [("true", true), ("false", false)] {
            let pb = PbOption {
                key: PRESERVE_ROW_SEQUENCE.to_string(),
                value: value.to_string(),
            };
            assert_eq!(
                SetRegionOption::PreserveRowSequence(expected),
                SetRegionOption::try_from(&pb).unwrap()
            );
        }

        for value in ["1", "invalid"] {
            let pb = PbOption {
                key: PRESERVE_ROW_SEQUENCE.to_string(),
                value: value.to_string(),
            };
            assert!(SetRegionOption::try_from(&pb).is_err());
        }

        assert_eq!(
            UnsetRegionOption::PreserveRowSequence,
            UnsetRegionOption::try_from(PRESERVE_ROW_SEQUENCE).unwrap()
        );
        assert_eq!(
            SetRegionOption::PreserveRowSequence(false),
            (&UnsetRegionOption::PreserveRowSequence).into()
        );
    }

    #[test]
    fn test_set_twcs_window_trigger_options_try_from() {
        for key in [
            "compaction.twcs.active_window.trigger_file_num",
            "compaction.twcs.active_window.l1_merge_trigger",
            "compaction.twcs.inactive_window.trigger_file_num",
            "compaction.twcs.inactive_window.l1_merge_trigger",
        ] {
            let option = PbOption {
                key: key.to_string(),
                value: "8".to_string(),
            };
            assert_eq!(
                SetRegionOption::Twsc(key.to_string(), "8".to_string()),
                SetRegionOption::try_from(&option).unwrap(),
                "{key}"
            );
        }
    }

    #[test]
    fn test_unset_twcs_window_trigger_options_try_from() {
        for (key, expected) in [
            (
                "compaction.twcs.active_window.trigger_file_num",
                UnsetRegionOption::TwcsActiveWindowTriggerFileNum,
            ),
            (
                "compaction.twcs.active_window.l1_merge_trigger",
                UnsetRegionOption::TwcsActiveWindowL1MergeTrigger,
            ),
            (
                "compaction.twcs.inactive_window.trigger_file_num",
                UnsetRegionOption::TwcsInactiveWindowTriggerFileNum,
            ),
        ] {
            assert_eq!(expected, UnsetRegionOption::try_from(key).unwrap());
            assert_eq!(
                SetRegionOption::Twsc(key.to_string(), String::new()),
                SetRegionOption::from(&expected)
            );
        }

        let key = "compaction.twcs.inactive_window.l1_merge_trigger";
        let option = UnsetRegionOption::try_from(key).unwrap();
        assert_eq!(key, option.to_string());
        assert_eq!(
            SetRegionOption::Twsc(key.to_string(), String::new()),
            SetRegionOption::from(&option)
        );
    }

    #[test]
    fn test_json_settings_path_validation_from_proto() {
        for (path, valid_path) in [
            (vec!["nested", "leaf"], true),
            (vec![""], false),
            (vec!["nested", ""], false),
            (
                vec!["nested", datatypes::json::JSON2_REMAINDER_FIELD_NAME],
                false,
            ),
        ] {
            for inverted_index in [false, true] {
                let settings = api::v1::SetJsonSettings {
                    column_name: "payload".into(),
                    settings: Some(api::v1::JsonSettings {
                        type_hints: vec![api::v1::JsonTypeHint {
                            path: path.iter().map(|part| (*part).to_string()).collect(),
                            data_type: ColumnDataType::String as i32,
                            inverted_index,
                            ..Default::default()
                        }],
                        max_auto_expanded_paths: Some(0),
                    }),
                };
                let result = RegionAlterRequest::try_from(AlterRequest {
                    region_id: 0,
                    schema_version: 1,
                    kind: Some(alter_request::Kind::SetJsonSettings(settings)),
                });
                assert_eq!(
                    result.is_ok(),
                    valid_path,
                    "path: {path:?}, inverted_index: {inverted_index}"
                );
            }
        }
    }

    #[test]
    fn test_from_proto_alter_request() {
        RegionAlterRequest::try_from(AlterRequest {
            region_id: 0,
            schema_version: 1,
            kind: None,
        })
        .unwrap_err();

        let request = RegionAlterRequest::try_from(AlterRequest {
            region_id: 0,
            schema_version: 1,
            kind: Some(alter_request::Kind::AddColumns(v1::region::AddColumns {
                add_columns: vec![v1::region::AddColumn {
                    column_def: Some(RegionColumnDef {
                        column_def: Some(ColumnDef {
                            name: "a".to_string(),
                            data_type: ColumnDataType::String as i32,
                            is_nullable: true,
                            default_constraint: vec![],
                            semantic_type: SemanticType::Field as i32,
                            comment: String::new(),
                            ..Default::default()
                        }),
                        column_id: 1,
                    }),
                    location: Some(v1::AddColumnLocation {
                        location_type: LocationType::First as i32,
                        after_column_name: String::default(),
                    }),
                }],
            })),
        })
        .unwrap();

        assert_eq!(
            request,
            RegionAlterRequest {
                kind: AlterKind::AddColumns {
                    columns: vec![AddColumn {
                        column_metadata: ColumnMetadata {
                            column_schema: ColumnSchema::new(
                                "a",
                                ConcreteDataType::string_datatype(),
                                true,
                            ),
                            semantic_type: SemanticType::Field,
                            column_id: 1,
                        },
                        location: Some(AddColumnLocation::First),
                    }]
                },
            }
        );

        let request = RegionAlterRequest::try_from(AlterRequest {
            region_id: 0,
            schema_version: 1,
            kind: Some(alter_request::Kind::SetJsonSettings(v1::SetJsonSettings {
                column_name: "payload".to_string(),
                settings: Some(v1::JsonSettings {
                    type_hints: vec![v1::JsonTypeHint {
                        path: vec!["service".to_string()],
                        data_type: ColumnDataType::String as i32,
                        datatype_extension: None,
                        inverted_index: true,
                    }],
                    max_auto_expanded_paths: Some(10),
                }),
            })),
        })
        .unwrap();

        let AlterKind::SetJsonSettings {
            column_name,
            settings,
        } = request.kind
        else {
            unreachable!()
        };
        assert_eq!("payload", column_name);
        assert_eq!(Some(10), settings.max_auto_expanded_paths());
        assert_eq!(1, settings.type_hints().len());
        assert!(settings.type_hints()[0].inverted_index);
    }

    #[test]
    fn test_write_buffer_size_region_options() {
        let option = PbOption {
            key: WRITE_BUFFER_SIZE_KEY.to_string(),
            value: "128MiB".to_string(),
        };
        assert_eq!(
            SetRegionOption::WriteBufferSize(Some(ReadableSize::mb(128))),
            SetRegionOption::try_from(&option).unwrap()
        );

        let option = PbOption {
            key: WRITE_BUFFER_SIZE_KEY.to_string(),
            value: "invalid".to_string(),
        };
        SetRegionOption::try_from(&option).unwrap_err();

        // Clearing the option uses UNSET. Empty SET values, including SQL NULL,
        // remain invalid because the SQL layer does not distinguish NULL from ''.
        let option = PbOption {
            key: WRITE_BUFFER_SIZE_KEY.to_string(),
            value: String::new(),
        };
        SetRegionOption::try_from(&option).unwrap_err();

        assert_eq!(
            UnsetRegionOption::WriteBufferSize,
            UnsetRegionOption::try_from(WRITE_BUFFER_SIZE_KEY).unwrap()
        );
        assert_eq!(
            SetRegionOption::WriteBufferSize(None),
            SetRegionOption::from(&UnsetRegionOption::WriteBufferSize)
        );
    }

    /// Returns a new region metadata for testing. Metadata:
    /// `[(ts, ms, 1), (tag_0, string, 2), (field_0, string, 3), (field_1, bool, 4)]`
    fn new_metadata() -> RegionMetadata {
        let mut builder = RegionMetadataBuilder::new(RegionId::new(1, 1));
        builder
            .push_column_metadata(ColumnMetadata {
                column_schema: ColumnSchema::new(
                    "ts",
                    ConcreteDataType::timestamp_millisecond_datatype(),
                    false,
                ),
                semantic_type: SemanticType::Timestamp,
                column_id: 1,
            })
            .push_column_metadata(ColumnMetadata {
                column_schema: ColumnSchema::new(
                    "tag_0",
                    ConcreteDataType::string_datatype(),
                    true,
                ),
                semantic_type: SemanticType::Tag,
                column_id: 2,
            })
            .push_column_metadata(ColumnMetadata {
                column_schema: ColumnSchema::new(
                    "field_0",
                    ConcreteDataType::string_datatype(),
                    true,
                ),
                semantic_type: SemanticType::Field,
                column_id: 3,
            })
            .push_column_metadata(ColumnMetadata {
                column_schema: ColumnSchema::new(
                    "field_1",
                    ConcreteDataType::boolean_datatype(),
                    true,
                ),
                semantic_type: SemanticType::Field,
                column_id: 4,
            })
            .primary_key(vec![2]);
        builder.build().unwrap()
    }

    fn json2_column_schema(name: &str, settings: JsonSettings) -> ColumnSchema {
        let mut column_schema =
            ColumnSchema::new(name, ConcreteDataType::Json(JsonType::null()), true);
        column_schema.with_extension_type(&Json2ExtensionType::new(std::sync::Arc::new(
            JsonMetadata::new(settings),
        )));
        column_schema
    }

    #[test]
    fn test_add_column_validate() {
        let metadata = new_metadata();
        let add_column = AddColumn {
            column_metadata: ColumnMetadata {
                column_schema: ColumnSchema::new(
                    "tag_1",
                    ConcreteDataType::string_datatype(),
                    true,
                ),
                semantic_type: SemanticType::Tag,
                column_id: 5,
            },
            location: None,
        };
        add_column.validate(&metadata).unwrap();
        assert!(add_column.need_alter(&metadata));

        // Add not null column.
        AddColumn {
            column_metadata: ColumnMetadata {
                column_schema: ColumnSchema::new(
                    "tag_1",
                    ConcreteDataType::string_datatype(),
                    false,
                ),
                semantic_type: SemanticType::Tag,
                column_id: 5,
            },
            location: None,
        }
        .validate(&metadata)
        .unwrap_err();

        // Add existing column.
        let add_column = AddColumn {
            column_metadata: ColumnMetadata {
                column_schema: ColumnSchema::new(
                    "tag_0",
                    ConcreteDataType::string_datatype(),
                    true,
                ),
                semantic_type: SemanticType::Tag,
                column_id: 2,
            },
            location: None,
        };
        add_column.validate(&metadata).unwrap();
        assert!(!add_column.need_alter(&metadata));
    }

    #[test]
    fn test_add_duplicate_columns() {
        let kind = AlterKind::AddColumns {
            columns: vec![
                AddColumn {
                    column_metadata: ColumnMetadata {
                        column_schema: ColumnSchema::new(
                            "tag_1",
                            ConcreteDataType::string_datatype(),
                            true,
                        ),
                        semantic_type: SemanticType::Tag,
                        column_id: 5,
                    },
                    location: None,
                },
                AddColumn {
                    column_metadata: ColumnMetadata {
                        column_schema: ColumnSchema::new(
                            "tag_1",
                            ConcreteDataType::string_datatype(),
                            true,
                        ),
                        semantic_type: SemanticType::Field,
                        column_id: 6,
                    },
                    location: None,
                },
            ],
        };
        let metadata = new_metadata();
        kind.validate(&metadata).unwrap();
        assert!(kind.need_alter(&metadata));
    }

    #[test]
    fn test_add_existing_column_different_metadata() {
        let metadata = new_metadata();

        // Add existing column with different id.
        let kind = AlterKind::AddColumns {
            columns: vec![AddColumn {
                column_metadata: ColumnMetadata {
                    column_schema: ColumnSchema::new(
                        "tag_0",
                        ConcreteDataType::string_datatype(),
                        true,
                    ),
                    semantic_type: SemanticType::Tag,
                    column_id: 4,
                },
                location: None,
            }],
        };
        kind.validate(&metadata).unwrap_err();

        // Add existing column with different type.
        let kind = AlterKind::AddColumns {
            columns: vec![AddColumn {
                column_metadata: ColumnMetadata {
                    column_schema: ColumnSchema::new(
                        "tag_0",
                        ConcreteDataType::int64_datatype(),
                        true,
                    ),
                    semantic_type: SemanticType::Tag,
                    column_id: 2,
                },
                location: None,
            }],
        };
        kind.validate(&metadata).unwrap_err();

        // Add existing column with different name.
        let kind = AlterKind::AddColumns {
            columns: vec![AddColumn {
                column_metadata: ColumnMetadata {
                    column_schema: ColumnSchema::new(
                        "tag_1",
                        ConcreteDataType::string_datatype(),
                        true,
                    ),
                    semantic_type: SemanticType::Tag,
                    column_id: 2,
                },
                location: None,
            }],
        };
        kind.validate(&metadata).unwrap_err();
    }

    #[test]
    fn test_add_existing_column_with_location() {
        let metadata = new_metadata();
        let kind = AlterKind::AddColumns {
            columns: vec![AddColumn {
                column_metadata: ColumnMetadata {
                    column_schema: ColumnSchema::new(
                        "tag_0",
                        ConcreteDataType::string_datatype(),
                        true,
                    ),
                    semantic_type: SemanticType::Tag,
                    column_id: 2,
                },
                location: Some(AddColumnLocation::First),
            }],
        };
        kind.validate(&metadata).unwrap_err();
    }

    #[test]
    fn test_validate_drop_column() {
        let metadata = new_metadata();
        let kind = AlterKind::DropColumns {
            names: vec!["xxxx".to_string()],
        };
        kind.validate(&metadata).unwrap();
        assert!(!kind.need_alter(&metadata));

        AlterKind::DropColumns {
            names: vec!["tag_0".to_string()],
        }
        .validate(&metadata)
        .unwrap_err();

        let kind = AlterKind::DropColumns {
            names: vec!["field_0".to_string()],
        };
        kind.validate(&metadata).unwrap();
        assert!(kind.need_alter(&metadata));
    }

    #[test]
    fn test_validate_modify_column_type() {
        let metadata = new_metadata();
        AlterKind::ModifyColumnTypes {
            columns: vec![ModifyColumnType {
                column_name: "xxxx".to_string(),
                target_type: ConcreteDataType::string_datatype(),
            }],
        }
        .validate(&metadata)
        .unwrap_err();

        AlterKind::ModifyColumnTypes {
            columns: vec![ModifyColumnType {
                column_name: "field_1".to_string(),
                target_type: ConcreteDataType::date_datatype(),
            }],
        }
        .validate(&metadata)
        .unwrap_err();

        AlterKind::ModifyColumnTypes {
            columns: vec![ModifyColumnType {
                column_name: "ts".to_string(),
                target_type: ConcreteDataType::date_datatype(),
            }],
        }
        .validate(&metadata)
        .unwrap_err();

        // Time index unit widening is allowed.
        let kind = AlterKind::ModifyColumnTypes {
            columns: vec![ModifyColumnType {
                column_name: "ts".to_string(),
                target_type: ConcreteDataType::timestamp_microsecond_datatype(),
            }],
        };
        kind.validate(&metadata).unwrap();
        assert!(kind.need_alter(&metadata));

        // Narrowing the time index unit is rejected.
        let metadata_nano = {
            let mut metadata = new_metadata();
            for col in metadata.column_metadatas.iter_mut() {
                if col.column_schema.name == "ts" {
                    col.column_schema.data_type = ConcreteDataType::timestamp_nanosecond_datatype();
                }
            }
            metadata
        };
        AlterKind::ModifyColumnTypes {
            columns: vec![ModifyColumnType {
                column_name: "ts".to_string(),
                target_type: ConcreteDataType::timestamp_millisecond_datatype(),
            }],
        }
        .validate(&metadata_nano)
        .unwrap_err();

        // Changing the time index to the same type is a validated no-op, so a
        // retried alter procedure (region already altered) succeeds and is
        // skipped by `need_alter`.
        let same_type = AlterKind::ModifyColumnTypes {
            columns: vec![ModifyColumnType {
                column_name: "ts".to_string(),
                target_type: ConcreteDataType::timestamp_millisecond_datatype(),
            }],
        };
        same_type.validate(&metadata).unwrap();
        assert!(!same_type.need_alter(&metadata));

        // Changing the time index to a non-timestamp type is rejected.
        AlterKind::ModifyColumnTypes {
            columns: vec![ModifyColumnType {
                column_name: "ts".to_string(),
                target_type: ConcreteDataType::string_datatype(),
            }],
        }
        .validate(&metadata)
        .unwrap_err();

        AlterKind::ModifyColumnTypes {
            columns: vec![ModifyColumnType {
                column_name: "tag_0".to_string(),
                target_type: ConcreteDataType::date_datatype(),
            }],
        }
        .validate(&metadata)
        .unwrap_err();

        let kind = AlterKind::ModifyColumnTypes {
            columns: vec![ModifyColumnType {
                column_name: "field_0".to_string(),
                target_type: ConcreteDataType::int32_datatype(),
            }],
        };
        kind.validate(&metadata).unwrap();
        assert!(kind.need_alter(&metadata));
    }

    #[test]
    fn test_validate_set_json_settings() {
        let mut metadata = new_metadata();
        let current_schema = json2_column_schema("field_0", JsonSettings::new_v2());
        let target_settings = JsonSettings::try_new(vec![], Some(10)).unwrap();
        metadata
            .column_metadatas
            .iter_mut()
            .find(|column| column.column_schema.name == "field_0")
            .unwrap()
            .column_schema = current_schema.clone();

        let kind = AlterKind::SetJsonSettings {
            column_name: "field_0".to_string(),
            settings: target_settings.clone(),
        };
        kind.validate(&metadata).unwrap();
        assert!(kind.need_alter(&metadata));

        let no_op = AlterKind::SetJsonSettings {
            column_name: "field_0".to_string(),
            settings: JsonSettings::new_v2(),
        };
        no_op.validate(&metadata).unwrap();
        assert!(!no_op.need_alter(&metadata));

        AlterKind::SetJsonSettings {
            column_name: "tag_0".to_string(),
            settings: target_settings,
        }
        .validate(&metadata)
        .unwrap_err();
    }

    #[test]
    fn test_validate_add_columns() {
        let kind = AlterKind::AddColumns {
            columns: vec![
                AddColumn {
                    column_metadata: ColumnMetadata {
                        column_schema: ColumnSchema::new(
                            "tag_1",
                            ConcreteDataType::string_datatype(),
                            true,
                        ),
                        semantic_type: SemanticType::Tag,
                        column_id: 5,
                    },
                    location: None,
                },
                AddColumn {
                    column_metadata: ColumnMetadata {
                        column_schema: ColumnSchema::new(
                            "field_2",
                            ConcreteDataType::string_datatype(),
                            true,
                        ),
                        semantic_type: SemanticType::Field,
                        column_id: 6,
                    },
                    location: None,
                },
            ],
        };
        let request = RegionAlterRequest { kind };
        let mut metadata = new_metadata();
        metadata.schema_version = 1;
        request.validate(&metadata).unwrap();
    }

    #[test]
    fn test_validate_create_region() {
        let column_metadatas = vec![
            ColumnMetadata {
                column_schema: ColumnSchema::new(
                    "ts",
                    ConcreteDataType::timestamp_millisecond_datatype(),
                    false,
                ),
                semantic_type: SemanticType::Timestamp,
                column_id: 1,
            },
            ColumnMetadata {
                column_schema: ColumnSchema::new(
                    "tag_0",
                    ConcreteDataType::string_datatype(),
                    true,
                ),
                semantic_type: SemanticType::Tag,
                column_id: 2,
            },
            ColumnMetadata {
                column_schema: ColumnSchema::new(
                    "field_0",
                    ConcreteDataType::string_datatype(),
                    true,
                ),
                semantic_type: SemanticType::Field,
                column_id: 3,
            },
        ];
        let create = RegionCreateRequest {
            engine: "mito".to_string(),
            column_metadatas,
            primary_key: vec![3, 4],
            options: HashMap::new(),
            table_dir: "path".to_string(),
            path_type: PathType::Bare,
            partition_expr_json: Some("".to_string()),
            requirements: Default::default(),
        };

        assert!(create.validate().is_err());
    }

    #[test]
    fn test_parse_create_region_requirements_defaults_to_empty() {
        let create = CreateRequest {
            region_id: RegionId::new(42, 0).as_u64(),
            engine: "mito".to_string(),
            column_defs: vec![],
            primary_key: vec![],
            path: "test".to_string(),
            options: HashMap::new(),
            partition: None,
            requirements: None,
        };

        let requests =
            RegionRequest::try_from_request_body(region_request::Body::Create(create)).unwrap();
        let RegionRequest::Create(request) = &requests[0].1 else {
            unreachable!()
        };

        assert_eq!(request.requirements, RegionRequirements::empty());
    }

    #[test]
    fn test_parse_create_region_requirements_from_proto() {
        let create = CreateRequest {
            region_id: RegionId::new(42, 0).as_u64(),
            engine: "mito".to_string(),
            column_defs: vec![],
            primary_key: vec![],
            path: "test".to_string(),
            options: HashMap::new(),
            partition: None,
            requirements: Some(api::v1::region::RegionRequirements {
                object_storage: true,
            }),
        };

        let requests =
            RegionRequest::try_from_request_body(region_request::Body::Create(create)).unwrap();
        let RegionRequest::Create(request) = &requests[0].1 else {
            unreachable!()
        };

        assert_eq!(request.requirements, RegionRequirements::object_storage());
    }

    #[test]
    fn test_parse_region_cleanup_from_proto() {
        let clean_up = api::v1::region::CleanUpRequest {
            region_id: RegionId::new(42, 3).as_u64(),
            engine: "mito".to_string(),
            path: "test".to_string(),
            options: HashMap::from([("k".to_string(), "v".to_string())]),
        };

        let requests =
            RegionRequest::try_from_request_body(region_request::Body::CleanUp(clean_up)).unwrap();
        let RegionRequest::CleanUp(request) = &requests[0].1 else {
            unreachable!()
        };

        assert_eq!(requests[0].0, RegionId::new(42, 3));
        assert_eq!(request.engine, "mito");
        assert_eq!(request.table_dir, "data/test/42/");
        assert_eq!(request.path_type, PathType::Bare);
        assert_eq!(request.options.get("k"), Some(&"v".to_string()));
    }

    #[test]
    fn test_validate_modify_column_fulltext_options() {
        let kind = AlterKind::SetIndexes {
            options: vec![SetIndexOption::Fulltext {
                column_name: "tag_0".to_string(),
                options: FulltextOptions::new_unchecked(
                    true,
                    FulltextAnalyzer::Chinese,
                    false,
                    FulltextBackend::Bloom,
                    1000,
                    0.01,
                ),
            }],
        };
        let request = RegionAlterRequest { kind };
        let mut metadata = new_metadata();
        metadata.schema_version = 1;
        request.validate(&metadata).unwrap();

        let kind = AlterKind::UnsetIndexes {
            options: vec![UnsetIndexOption::Fulltext {
                column_name: "tag_0".to_string(),
            }],
        };
        let request = RegionAlterRequest { kind };
        let mut metadata = new_metadata();
        metadata.schema_version = 1;
        request.validate(&metadata).unwrap();
    }

    #[test]
    fn test_validate_sync_columns() {
        let metadata = new_metadata();
        let kind = AlterKind::SyncColumns {
            column_metadatas: vec![
                ColumnMetadata {
                    column_schema: ColumnSchema::new(
                        "tag_1",
                        ConcreteDataType::string_datatype(),
                        true,
                    ),
                    semantic_type: SemanticType::Tag,
                    column_id: 5,
                },
                ColumnMetadata {
                    column_schema: ColumnSchema::new(
                        "field_2",
                        ConcreteDataType::string_datatype(),
                        true,
                    ),
                    semantic_type: SemanticType::Field,
                    column_id: 6,
                },
            ],
        };
        let err = kind.validate(&metadata).unwrap_err();
        assert!(err.to_string().contains("not a primary key"));

        // Change the timestamp column name.
        let mut column_metadatas_with_different_ts_column = metadata.column_metadatas.clone();
        let ts_column = column_metadatas_with_different_ts_column
            .iter_mut()
            .find(|c| c.semantic_type == SemanticType::Timestamp)
            .unwrap();
        ts_column.column_schema.name = "ts1".to_string();

        let kind = AlterKind::SyncColumns {
            column_metadatas: column_metadatas_with_different_ts_column,
        };
        let err = kind.validate(&metadata).unwrap_err();
        assert!(
            err.to_string()
                .contains("timestamp column ts has different id")
        );

        // Change the primary key column name.
        let mut column_metadatas_with_different_pk_column = metadata.column_metadatas.clone();
        let pk_column = column_metadatas_with_different_pk_column
            .iter_mut()
            .find(|c| c.column_schema.name == "tag_0")
            .unwrap();
        pk_column.column_id = 100;
        let kind = AlterKind::SyncColumns {
            column_metadatas: column_metadatas_with_different_pk_column,
        };
        let err = kind.validate(&metadata).unwrap_err();
        assert!(
            err.to_string()
                .contains("column with same name tag_0 has different id")
        );

        // Add a new field column.
        let mut column_metadatas_with_new_field_column = metadata.column_metadatas.clone();
        column_metadatas_with_new_field_column.push(ColumnMetadata {
            column_schema: ColumnSchema::new("field_2", ConcreteDataType::string_datatype(), true),
            semantic_type: SemanticType::Field,
            column_id: 4,
        });
        let kind = AlterKind::SyncColumns {
            column_metadatas: column_metadatas_with_new_field_column,
        };
        kind.validate(&metadata).unwrap();
    }

    #[test]
    fn test_cast_path_type_to_primitive() {
        assert_eq!(PathType::Bare as u8, 0);
        assert_eq!(PathType::Data as u8, 1);
        assert_eq!(PathType::Metadata as u8, 2);
        assert_eq!(
            PathType::try_from(PathType::Bare as u8).unwrap(),
            PathType::Bare
        );
        assert_eq!(
            PathType::try_from(PathType::Data as u8).unwrap(),
            PathType::Data
        );
        assert_eq!(
            PathType::try_from(PathType::Metadata as u8).unwrap(),
            PathType::Metadata
        );
    }
}