// 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.

//! Table and TableEngine requests

use std::collections::{HashMap, HashSet};
use std::fmt;
use std::str::FromStr;

use common_base::readable_size::ReadableSize;
use common_datasource::object_store::oss::is_supported_in_oss;
use common_datasource::object_store::s3::is_supported_in_s3;
use common_query::AddColumnLocation;
use common_time::TimeToLive;
use common_time::range::TimestampRange;
use datatypes::data_type::ConcreteDataType;
use datatypes::prelude::VectorRef;
use datatypes::schema::{
    ColumnDefaultConstraint, ColumnSchema, FulltextOptions, Schema, SkippingIndexOptions,
};
use greptime_proto::v1::region::compact_request;
use once_cell::sync::Lazy;
use serde::{Deserialize, Serialize};
use store_api::metric_engine_consts::{
    LOGICAL_TABLE_METADATA_KEY, PHYSICAL_TABLE_METADATA_KEY, is_metric_engine_option_key,
};
use store_api::mito_engine_options::{
    APPEND_MODE_KEY, COMPACTION_TYPE, MEMTABLE_BULK_ENCODE_BYTES_THRESHOLD,
    MEMTABLE_BULK_ENCODE_ROW_THRESHOLD, MEMTABLE_BULK_MAX_MERGE_GROUPS,
    MEMTABLE_BULK_MERGE_THRESHOLD, MEMTABLE_TYPE, MERGE_MODE_KEY, SST_FORMAT_KEY,
    TWCS_FALLBACK_TO_LOCAL, TWCS_MAX_OUTPUT_FILE_SIZE, TWCS_TIME_WINDOW, TWCS_TRIGGER_FILE_NUM,
    is_mito_engine_option_key,
};
use store_api::region_request::{SetRegionOption, UnsetRegionOption};

use crate::error::{ParseTableOptionSnafu, Result};
use crate::metadata::{TableId, TableVersion};
use crate::table_reference::TableReference;

mod semantic;
pub use semantic::*;

pub const FILE_TABLE_META_KEY: &str = "__private.file_table_meta";
pub const FILE_TABLE_LOCATION_KEY: &str = "location";
pub const FILE_TABLE_PATTERN_KEY: &str = "pattern";
pub const FILE_TABLE_FORMAT_KEY: &str = "format";

pub const TABLE_DATA_MODEL: &str = "table_data_model";
pub const TABLE_DATA_MODEL_TRACE_V1: &str = "greptime_trace_v1";

/// Returns true if the table stores spans in the `greptime_trace_v1` data model
/// (fixed span columns), the shape the Jaeger query path and the entity-graph
/// derivation rely on.
pub fn is_trace_v1_table(table_info: &crate::metadata::TableInfo) -> bool {
    table_info
        .meta
        .options
        .extra_options
        .get(TABLE_DATA_MODEL)
        .map(|v| v == TABLE_DATA_MODEL_TRACE_V1)
        .unwrap_or(false)
}

pub const OTLP_METRIC_COMPAT_KEY: &str = "otlp_metric_compat";
pub const OTLP_METRIC_COMPAT_PROM: &str = "prom";

pub const VALID_TABLE_OPTION_KEYS: [&str; 14] = [
    // common keys:
    WRITE_BUFFER_SIZE_KEY,
    TTL_KEY,
    STORAGE_KEY,
    COMMENT_KEY,
    SKIP_WAL_KEY,
    SST_FORMAT_KEY,
    // file engine keys:
    FILE_TABLE_LOCATION_KEY,
    FILE_TABLE_FORMAT_KEY,
    FILE_TABLE_PATTERN_KEY,
    // metric engine keys:
    PHYSICAL_TABLE_METADATA_KEY,
    LOGICAL_TABLE_METADATA_KEY,
    // table model info
    TABLE_DATA_MODEL,
    OTLP_METRIC_COMPAT_KEY,
    REPARTITION_COLUMN_HINT_KEY,
];

pub const DDL_TIMEOUT: &str = "timeout";
pub const DDL_WAIT: &str = "wait";

pub const VALID_DDL_OPTION_KEYS: [&str; 2] = [DDL_TIMEOUT, DDL_WAIT];

// Valid option keys when creating a db.
static VALID_DB_OPT_KEYS: Lazy<HashSet<&str>> = Lazy::new(|| {
    let mut set = HashSet::new();
    set.insert(TTL_KEY);
    set.insert(STORAGE_KEY);
    set.insert(MEMTABLE_TYPE);
    set.insert(MEMTABLE_BULK_MERGE_THRESHOLD);
    set.insert(MEMTABLE_BULK_ENCODE_ROW_THRESHOLD);
    set.insert(MEMTABLE_BULK_ENCODE_BYTES_THRESHOLD);
    set.insert(MEMTABLE_BULK_MAX_MERGE_GROUPS);
    set.insert(APPEND_MODE_KEY);
    set.insert(MERGE_MODE_KEY);
    set.insert(SKIP_WAL_KEY);
    set.insert(COMPACTION_TYPE);
    set.insert(TWCS_FALLBACK_TO_LOCAL);
    set.insert(TWCS_TIME_WINDOW);
    set.insert(TWCS_TRIGGER_FILE_NUM);
    set.insert(TWCS_MAX_OUTPUT_FILE_SIZE);
    set.insert(SST_FORMAT_KEY);
    set
});

/// Returns true if the `key` is a valid key for database.
pub fn validate_database_option(key: &str) -> bool {
    VALID_DB_OPT_KEYS.contains(&key)
}

/// Returns true if the `key` is a valid key for any engine or storage.
pub fn validate_table_option(key: &str) -> bool {
    if is_supported_in_s3(key) {
        return true;
    }

    if is_supported_in_oss(key) {
        return true;
    }

    if is_mito_engine_option_key(key) {
        return true;
    }

    if is_metric_engine_option_key(key) {
        return true;
    }

    // Semantic-layer keys share a reserved prefix instead of a fixed allowlist so
    // the vocabulary can grow without touching this gate. See `semantic` module.
    if is_semantic_option_key(key) {
        return true;
    }

    VALID_TABLE_OPTION_KEYS.contains(&key) || VALID_DDL_OPTION_KEYS.contains(&key)
}

#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq)]
#[serde(default)]
pub struct TableOptions {
    /// Per-region write buffer stall threshold. Writes are rejected at twice this size.
    pub write_buffer_size: Option<ReadableSize>,
    /// Time-to-live of table. Expired data will be automatically purged.
    pub ttl: Option<TimeToLive>,
    /// Skip wal write for this table.
    pub skip_wal: bool,
    /// Extra options that may not applicable to all table engines.
    pub extra_options: HashMap<String, String>,
}

pub const WRITE_BUFFER_SIZE_KEY: &str = store_api::mito_engine_options::WRITE_BUFFER_SIZE_KEY;
pub const TTL_KEY: &str = store_api::mito_engine_options::TTL_KEY;
pub const STORAGE_KEY: &str = "storage";
pub const COMMENT_KEY: &str = "comment";
pub const AUTO_CREATE_TABLE_KEY: &str = "auto_create_table";
pub const SKIP_WAL_KEY: &str = store_api::mito_engine_options::SKIP_WAL_KEY;
pub const TRACE_TABLE_PARTITIONS_HINT_KEY: &str = "trace_table_partitions";
pub const REPARTITION_COLUMN_HINT_KEY: &str = "repartition.column.hint";

impl TableOptions {
    pub fn try_from_iter<T: ToString, U: IntoIterator<Item = (T, T)>>(
        iter: U,
    ) -> Result<TableOptions> {
        let mut options = TableOptions::default();

        let kvs: HashMap<String, String> = iter
            .into_iter()
            .map(|(k, v)| (k.to_string(), v.to_string()))
            .collect();

        if let Some(write_buffer_size) = kvs.get(WRITE_BUFFER_SIZE_KEY) {
            let size = ReadableSize::from_str(write_buffer_size).map_err(|_| {
                ParseTableOptionSnafu {
                    key: WRITE_BUFFER_SIZE_KEY,
                    value: write_buffer_size,
                }
                .build()
            })?;
            options.write_buffer_size = Some(size)
        }

        if let Some(ttl) = kvs.get(TTL_KEY) {
            let ttl_value = TimeToLive::from_humantime_or_str(ttl).map_err(|_| {
                ParseTableOptionSnafu {
                    key: TTL_KEY,
                    value: ttl,
                }
                .build()
            })?;
            options.ttl = Some(ttl_value);
        }

        if let Some(skip_wal) = kvs.get(SKIP_WAL_KEY) {
            options.skip_wal = skip_wal.parse().map_err(|_| {
                ParseTableOptionSnafu {
                    key: SKIP_WAL_KEY,
                    value: skip_wal,
                }
                .build()
            })?;
        }

        options.extra_options = HashMap::from_iter(
            kvs.into_iter()
                .filter(|(k, _)| k != WRITE_BUFFER_SIZE_KEY && k != TTL_KEY),
        );

        Ok(options)
    }
}

impl fmt::Display for TableOptions {
    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
        let mut key_vals = vec![];
        if let Some(size) = self.write_buffer_size {
            key_vals.push(format!("{}={}", WRITE_BUFFER_SIZE_KEY, size));
        }

        if let Some(ttl) = self.ttl.map(|ttl| ttl.to_string()) {
            key_vals.push(format!("{}={}", TTL_KEY, ttl));
        }

        if self.skip_wal && !self.extra_options.contains_key(SKIP_WAL_KEY) {
            key_vals.push(format!("{}={}", SKIP_WAL_KEY, self.skip_wal));
        }

        for (k, v) in &self.extra_options {
            key_vals.push(format!("{}={}", k, v));
        }

        write!(f, "{}", key_vals.join(" "))
    }
}

impl From<&TableOptions> for HashMap<String, String> {
    fn from(opts: &TableOptions) -> Self {
        let mut res = HashMap::with_capacity(3 + opts.extra_options.len());
        if let Some(write_buffer_size) = opts.write_buffer_size {
            let _ = res.insert(
                WRITE_BUFFER_SIZE_KEY.to_string(),
                write_buffer_size.to_string(),
            );
        }
        if let Some(ttl_str) = opts.ttl.map(|ttl| ttl.to_string()) {
            let _ = res.insert(TTL_KEY.to_string(), ttl_str);
        }
        if opts.skip_wal {
            let _ = res.insert(SKIP_WAL_KEY.to_string(), true.to_string());
        }
        res.extend(opts.extra_options.clone());
        res
    }
}

/// Alter table request
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct AlterTableRequest {
    pub catalog_name: String,
    pub schema_name: String,
    pub table_name: String,
    pub table_id: TableId,
    pub alter_kind: AlterKind,
    // None in standalone.
    pub table_version: Option<TableVersion>,
}

/// Add column request
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct AddColumnRequest {
    pub column_schema: ColumnSchema,
    pub is_key: bool,
    pub location: Option<AddColumnLocation>,
    /// Add column if not exists.
    pub add_if_not_exists: bool,
}

/// Change column datatype request
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ModifyColumnTypeRequest {
    pub column_name: String,
    pub target_type: ConcreteDataType,
}

/// A family of annotation table options: pure metadata markers that no region
/// consumes. Setting or unsetting them only rewrites the table's
/// `extra_options`, so the alter skips region dispatch entirely.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub enum AnnotationFamily {
    /// `greptime.semantic.*` options (see the [`semantic`] module).
    Semantic,
    /// `repartition.column.hint`, consumed by the auto-repartition planner.
    RepartitionHint,
}

impl AnnotationFamily {
    /// The key namespace: a prefix for [`Self::Semantic`], the exact key for
    /// [`Self::RepartitionHint`].
    pub fn namespace(self) -> &'static str {
        match self {
            Self::Semantic => SEMANTIC_PREFIX,
            Self::RepartitionHint => REPARTITION_COLUMN_HINT_KEY,
        }
    }

    pub fn of_key(key: &str) -> Option<Self> {
        if key.starts_with(SEMANTIC_PREFIX) {
            Some(Self::Semantic)
        } else if key == REPARTITION_COLUMN_HINT_KEY {
            Some(Self::RepartitionHint)
        } else {
            None
        }
    }

    /// Whether this family may be altered on logical metric tables. Only
    /// families whose values nothing on the physical side consumes qualify;
    /// the repartition hint drives physical region repartitioning.
    pub fn allows_logical_tables(self) -> bool {
        match self {
            Self::Semantic => true,
            Self::RepartitionHint => false,
        }
    }

    /// Whether this family's SET/UNSET batch must contain exactly one key.
    /// The repartition hint is a single marker; a batch with several hint
    /// entries (duplicates included) has no meaningful order.
    pub fn requires_single_key(self) -> bool {
        matches!(self, Self::RepartitionHint)
    }

    /// The error for a SET/UNSET batch mixing this family with other options.
    pub fn mixed_batch_error(self) -> String {
        match self {
            Self::Semantic => format!(
                "`{SEMANTIC_PREFIX}*` options must be altered separately from other table options"
            ),
            Self::RepartitionHint => {
                format!("{REPARTITION_COLUMN_HINT_KEY} must be altered separately")
            }
        }
    }
}

/// Table shape an annotation option is validated against.
pub struct AnnotationContext<'a> {
    pub schema: &'a Schema,
    pub partition_key_indices: &'a [usize],
}

/// Why an annotation option was rejected. Typed so each DDL entry point maps
/// rules onto its existing error variants and status codes: ALTER keeps
/// missing columns as `TableColumnNotFound` (4002), CREATE keeps its
/// `InvalidArguments` family — the rules converge, the contracts do not.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum AnnotationValidationError {
    UnknownKey {
        key: String,
    },
    InvalidValue {
        key: String,
        value: String,
    },
    ColumnNotFound {
        column: String,
    },
    ColumnNotStringForm {
        key: String,
        column: String,
        ty: ConcreteDataType,
    },
    NotSingleColumn,
    PartitionMetadataConflict,
    TimeIndexConflict,
}

impl fmt::Display for AnnotationValidationError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::UnknownKey { key } => write!(f, "unknown semantic option `{key}`"),
            Self::InvalidValue { key, value } => {
                write!(f, "invalid value `{value}` for semantic option `{key}`")
            }
            Self::ColumnNotFound { column } => write!(f, "column `{column}` not found"),
            Self::ColumnNotStringForm { key, column, ty } => write!(
                f,
                "entity column `{column}` (option `{key}`) has type `{ty}`, \
                 which cannot render as a string"
            ),
            Self::NotSingleColumn => write!(
                f,
                "{REPARTITION_COLUMN_HINT_KEY} expects exactly one column name"
            ),
            Self::PartitionMetadataConflict => write!(
                f,
                "cannot set {REPARTITION_COLUMN_HINT_KEY} on a table with partition metadata"
            ),
            Self::TimeIndexConflict => write!(
                f,
                "cannot set {REPARTITION_COLUMN_HINT_KEY} to the time index column"
            ),
        }
    }
}

/// Validates one annotation option and returns the value to store — the
/// repartition hint is trimmed to the bare column name, semantic values pass
/// through unchanged.
pub(crate) fn validate_and_normalize_annotation(
    family: AnnotationFamily,
    cx: &AnnotationContext<'_>,
    key: &str,
    value: &str,
) -> std::result::Result<String, AnnotationValidationError> {
    match family {
        AnnotationFamily::Semantic => {
            if !is_semantic_option_key(key) {
                return Err(AnnotationValidationError::UnknownKey {
                    key: key.to_string(),
                });
            }
            if !validate_semantic_option(key, value) {
                return Err(AnnotationValidationError::InvalidValue {
                    key: key.to_string(),
                    value: value.to_string(),
                });
            }
            if parse_entity_option_key(key).is_some() {
                for column in parse_entity_columns(value) {
                    let schema = cx.schema.column_schema_by_name(&column).ok_or_else(|| {
                        AnnotationValidationError::ColumnNotFound {
                            column: column.clone(),
                        }
                    })?;
                    if !has_stable_string_form(&schema.data_type) {
                        return Err(AnnotationValidationError::ColumnNotStringForm {
                            key: key.to_string(),
                            column,
                            ty: schema.data_type.clone(),
                        });
                    }
                }
            }
            Ok(value.to_string())
        }
        AnnotationFamily::RepartitionHint => {
            let column_name = value.trim();
            if column_name.is_empty() || column_name.contains(',') {
                return Err(AnnotationValidationError::NotSingleColumn);
            }
            if !cx.partition_key_indices.is_empty() {
                return Err(AnnotationValidationError::PartitionMetadataConflict);
            }
            let column_index = cx.schema.column_index_by_name(column_name).ok_or_else(|| {
                AnnotationValidationError::ColumnNotFound {
                    column: column_name.to_string(),
                }
            })?;
            if cx.schema.timestamp_index() == Some(column_index) {
                return Err(AnnotationValidationError::TimeIndexConflict);
            }
            Ok(column_name.to_string())
        }
    }
}

/// CREATE-side entry: validates every annotation option present in `options`
/// and writes normalized values back in place.
pub fn validate_and_normalize_annotation_options(
    options: &mut TableOptions,
    cx: &AnnotationContext<'_>,
) -> std::result::Result<(), AnnotationValidationError> {
    let mut normalized = Vec::new();
    for (key, value) in &options.extra_options {
        let Some(family) = AnnotationFamily::of_key(key) else {
            continue;
        };
        let checked = validate_and_normalize_annotation(family, cx, key, value)?;
        if checked != *value {
            normalized.push((key.clone(), checked));
        }
    }
    for (key, value) in normalized {
        options.extra_options.insert(key, value);
    }
    Ok(())
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum AlterKind {
    AddColumns {
        columns: Vec<AddColumnRequest>,
    },
    DropColumns {
        names: Vec<String>,
    },
    ModifyColumnTypes {
        columns: Vec<ModifyColumnTypeRequest>,
    },
    RenameTable {
        new_table_name: String,
    },
    SetTableOptions {
        options: Vec<SetRegionOption>,
    },
    UnsetTableOptions {
        keys: Vec<UnsetRegionOption>,
    },
    SetAnnotations {
        family: AnnotationFamily,
        options: Vec<(String, String)>,
    },
    UnsetAnnotations {
        family: AnnotationFamily,
        keys: Vec<String>,
    },
    SetIndexes {
        options: Vec<SetIndexOption>,
    },
    UnsetIndexes {
        options: Vec<UnsetIndexOption>,
    },
    DropDefaults {
        names: Vec<String>,
    },
    SetDefaults {
        defaults: Vec<SetDefaultRequest>,
    },
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SetDefaultRequest {
    pub column_name: String,
    pub default_constraint: Option<ColumnDefaultConstraint>,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
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) -> &str {
        match self {
            SetIndexOption::Fulltext { column_name, .. } => column_name,
            SetIndexOption::Inverted { column_name, .. } => column_name,
            SetIndexOption::Skipping { column_name, .. } => column_name,
        }
    }
}

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

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

#[derive(Debug)]
pub struct InsertRequest {
    pub catalog_name: String,
    pub schema_name: String,
    pub table_name: String,
    pub columns_values: HashMap<String, VectorRef>,
}

/// Delete (by primary key) request
#[derive(Debug)]
pub struct DeleteRequest {
    pub catalog_name: String,
    pub schema_name: String,
    pub table_name: String,
    /// Values of each column in this table's primary key and time index.
    ///
    /// The key is the column name, and the value is the column value.
    pub key_column_values: HashMap<String, VectorRef>,
}

#[derive(Debug)]
pub enum CopyDirection {
    Export,
    Import,
}

/// Copy table request
#[derive(Debug)]
pub struct CopyTableRequest {
    pub catalog_name: String,
    pub schema_name: String,
    pub table_name: String,
    pub location: String,
    pub with: HashMap<String, String>,
    pub connection: HashMap<String, String>,
    pub pattern: Option<String>,
    pub direction: CopyDirection,
    pub timestamp_range: Option<TimestampRange>,
    pub limit: Option<u64>,
}

#[derive(Debug, Clone, Default)]
pub struct FlushTableRequest {
    pub catalog_name: String,
    pub schema_name: String,
    pub table_name: String,
}

#[derive(Debug, Clone, Default)]
pub struct BuildIndexTableRequest {
    pub catalog_name: String,
    pub schema_name: String,
    pub table_name: String,
}

#[derive(Debug, Clone, PartialEq)]
pub struct CompactTableRequest {
    pub catalog_name: String,
    pub schema_name: String,
    pub table_name: String,
    pub compact_options: compact_request::Options,
    pub parallelism: u32,
    pub time_range: Option<TimestampRange>,
}

impl Default for CompactTableRequest {
    fn default() -> Self {
        Self {
            catalog_name: Default::default(),
            schema_name: Default::default(),
            table_name: Default::default(),
            compact_options: compact_request::Options::Regular(Default::default()),
            parallelism: 1,
            time_range: None,
        }
    }
}

/// Truncate table request
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TruncateTableRequest {
    pub catalog_name: String,
    pub schema_name: String,
    pub table_name: String,
    pub table_id: TableId,
}

impl TruncateTableRequest {
    pub fn table_ref(&self) -> TableReference<'_> {
        TableReference {
            catalog: &self.catalog_name,
            schema: &self.schema_name,
            table: &self.table_name,
        }
    }
}

#[derive(Debug, Clone, Default, Deserialize, Serialize)]
pub struct CopyDatabaseRequest {
    pub catalog_name: String,
    pub schema_name: String,
    pub location: String,
    pub with: HashMap<String, String>,
    pub connection: HashMap<String, String>,
    pub time_range: Option<TimestampRange>,
}

#[derive(Debug, Clone, Default, Deserialize, Serialize)]
pub struct CopyQueryToRequest {
    pub location: String,
    pub with: HashMap<String, String>,
    pub connection: HashMap<String, String>,
}

#[cfg(test)]
mod tests {
    use std::time::Duration;

    use super::*;

    #[test]
    fn test_validate_table_option() {
        assert!(validate_table_option(FILE_TABLE_LOCATION_KEY));
        assert!(validate_table_option(FILE_TABLE_FORMAT_KEY));
        assert!(validate_table_option(FILE_TABLE_PATTERN_KEY));
        assert!(validate_table_option(TTL_KEY));
        assert!(validate_table_option(WRITE_BUFFER_SIZE_KEY));
        assert!(validate_table_option(STORAGE_KEY));
        assert!(validate_table_option(MEMTABLE_BULK_MERGE_THRESHOLD));
        assert!(validate_table_option(REPARTITION_COLUMN_HINT_KEY));
        assert!(!validate_table_option("foo"));

        // Only whitelisted semantic keys are accepted.
        assert!(validate_table_option(SEMANTIC_SIGNAL_TYPE));
        assert!(validate_table_option(SEMANTIC_METRIC_TYPE));
        // Unknown semantic key, near-miss, and the internal transport key are rejected.
        assert!(!validate_table_option("greptime.semantic.future.key"));
        assert!(!validate_table_option("greptime.semanticx"));
        assert!(!validate_table_option(SEMANTIC_PER_TABLE_INDEX_KEY));
    }

    #[test]
    fn test_validate_database_option() {
        assert!(validate_database_option(MEMTABLE_TYPE));
        assert!(validate_database_option(MEMTABLE_BULK_MERGE_THRESHOLD));
        assert!(validate_database_option(MEMTABLE_BULK_ENCODE_ROW_THRESHOLD));
        assert!(validate_database_option(
            MEMTABLE_BULK_ENCODE_BYTES_THRESHOLD
        ));
        assert!(validate_database_option(MEMTABLE_BULK_MAX_MERGE_GROUPS));
        assert!(!validate_database_option("foo"));
    }

    #[test]
    fn test_serialize_table_options() {
        let options = TableOptions {
            write_buffer_size: None,
            ttl: Some(Duration::from_secs(1000).into()),
            extra_options: HashMap::new(),
            skip_wal: false,
        };
        let serialized = serde_json::to_string(&options).unwrap();
        let deserialized: TableOptions = serde_json::from_str(&serialized).unwrap();
        assert_eq!(options, deserialized);
    }

    #[test]
    fn test_convert_hashmap_between_table_options() {
        let options = TableOptions {
            write_buffer_size: Some(ReadableSize::mb(128)),
            ttl: Some(Duration::from_secs(1000).into()),
            extra_options: HashMap::new(),
            skip_wal: false,
        };
        let serialized_map = HashMap::from(&options);
        let serialized = TableOptions::try_from_iter(&serialized_map).unwrap();
        assert_eq!(options, serialized);

        let options = TableOptions {
            write_buffer_size: None,
            ttl: None,
            extra_options: HashMap::from([(SKIP_WAL_KEY.to_string(), true.to_string())]),
            skip_wal: true,
        };
        let serialized_map = HashMap::from(&options);
        assert_eq!(
            Some("true"),
            serialized_map.get(SKIP_WAL_KEY).map(String::as_str)
        );
        let serialized = TableOptions::try_from_iter(&serialized_map).unwrap();
        assert_eq!(options, serialized);

        let options = TableOptions {
            write_buffer_size: None,
            ttl: Default::default(),
            extra_options: HashMap::new(),
            skip_wal: false,
        };
        let serialized_map = HashMap::from(&options);
        let serialized = TableOptions::try_from_iter(&serialized_map).unwrap();
        assert_eq!(options, serialized);

        let options = TableOptions {
            write_buffer_size: Some(ReadableSize::mb(128)),
            ttl: Some(Duration::from_secs(1000).into()),
            extra_options: HashMap::from([("a".to_string(), "A".to_string())]),
            skip_wal: false,
        };
        let serialized_map = HashMap::from(&options);
        let serialized = TableOptions::try_from_iter(&serialized_map).unwrap();
        assert_eq!(options, serialized);

        let options = TableOptions {
            extra_options: HashMap::from([(SKIP_WAL_KEY.to_string(), false.to_string())]),
            skip_wal: false,
            ..Default::default()
        };
        let serialized_map = HashMap::from(&options);
        let serialized = TableOptions::try_from_iter(&serialized_map).unwrap();
        assert_eq!(options, serialized);
    }

    #[test]
    fn test_table_options_to_string() {
        let options = TableOptions {
            write_buffer_size: Some(ReadableSize::mb(128)),
            ttl: Some(Duration::from_secs(1000).into()),
            extra_options: HashMap::new(),
            skip_wal: false,
        };

        assert_eq!(
            "write_buffer_size=128.0MiB ttl=16m 40s",
            options.to_string()
        );

        let options = TableOptions {
            write_buffer_size: Some(ReadableSize::mb(128)),
            ttl: Some(Duration::from_secs(1000).into()),
            extra_options: HashMap::from([("a".to_string(), "A".to_string())]),
            skip_wal: false,
        };

        assert_eq!(
            "write_buffer_size=128.0MiB ttl=16m 40s a=A",
            options.to_string()
        );

        let options = TableOptions {
            write_buffer_size: Some(ReadableSize::mb(128)),
            ttl: Some(Duration::from_secs(1000).into()),
            extra_options: HashMap::new(),
            skip_wal: true,
        };
        assert_eq!(
            "write_buffer_size=128.0MiB ttl=16m 40s skip_wal=true",
            options.to_string()
        );

        let options = TableOptions {
            write_buffer_size: None,
            ttl: None,
            extra_options: HashMap::from([(SKIP_WAL_KEY.to_string(), "false".to_string())]),
            skip_wal: false,
        };
        assert_eq!("skip_wal=false", options.to_string());
    }
}
