#![feature(debug_closure_helpers)]

pub mod generated;
pub mod metrics;
pub mod test_utils;
pub mod tidb_def;
pub mod util;

pub mod tidb_types;

use std::collections::{HashMap, HashSet};

use anyhow::{Context, Result, anyhow, bail, ensure};
use lazy_static::lazy_static;

mod tici {
    include!("generated/tici.rs");
}

mod tici_internal {
    include!("generated/tici_internal.rs");
}

mod reader {
    include!("generated/reader.rs");
}

mod import {
    include!("generated/import.rs");
}

mod schema {
    include!("generated/schema.rs");
}

pub use import::{ImportWriteRequest, ImportWriteResponse, ImportWriteStatus};
pub use reader::{
    GetDebugCacheInfoRequest, GetDebugCacheInfoResponse, WarmupShardRequest, WarmupShardResponse,
    reader_service_client::ReaderServiceClient as GrpcReaderServiceClient,
    reader_service_server::{
        ReaderService as GrpcReaderService, ReaderServiceServer as GrpcReaderServiceServer,
    },
};
pub use schema::{
    ColumnInfo, IndexInfo, ParserInfo, ParserType, PartitionDefinition, PartitionTableInfo,
    TableInfo,
};
pub mod prelude_import {
    pub use super::tici::{
        FinishImportIndexUploadRequest, FinishImportPartitionUploadRequest, FinishImportResponse,
        GetImportStoragePrefixRequest, GetImportStoragePrefixResponse, PreSplitImportIndexResult,
        PreSplitImportShardMeta, PreSplitImportShardsRequest, PreSplitImportShardsResponse,
    };
}
pub use tici::{
    CreateIndexRequest, CreateIndexResponse, DropIndexRequest, DropIndexResponse, ErrorCode,
    GetIndexProgressRequest, GetIndexProgressResponse, GetShardLocalCacheRequest,
    GetShardLocalCacheResponse, KeyRange, PreSplitImportIndexResult, PreSplitImportShardMeta,
    PreSplitImportShardsRequest, PreSplitImportShardsResponse, S3Location, ShardLocalCacheInfo,
    ShardManifestHeader,
    get_index_progress_response::State as WriterProgressState,
    meta_service_client::MetaServiceClient as GrpcMetaServiceClient,
    meta_service_server::{
        MetaService as GrpcMetaService, MetaServiceServer as GrpcMetaServiceServer,
    },
};
use tici_common::index::IndexKey;
pub use tici_internal::{
    AddShardRequest, AddShardResponse, AppendBaseFragMetaRequest, AppendDeltaFragMetaRequest,
    AppendFragMetaResponse, CompactFragRequest, CompactFragResponse, CompactType,
    DebugGetShardInfoRequest, DebugGetShardInfoResponse, DebugGetShardManifestRequest,
    DebugGetShardManifestResponse, FinishCompactFragRequest, FinishCompactFragResponse,
    FinishImportPartitionBuildRequest, FragMeta, FragProperty, FragStorageRef, GetIndexMetaRequest,
    GetIndexMetaResponse, GetShardManifestRequest, GetShardManifestResponse, ReaderCachedShard,
    ReaderNodeHeartbeatRequest, ReaderNodeHeartbeatResponse, ReaderNodeStatus,
    ShardInfo as GrpcInternalShardInfo, ShardProgress, WorkerNodeCompactionStatus,
    WorkerNodeHeartbeatRequest, WorkerNodeHeartbeatResponse, WorkerNodeShardStatus,
    WorkerNodeStatus,
    meta_service_internal_client::MetaServiceInternalClient as GrpcMetaServiceInternalClient,
    meta_service_internal_server::{
        MetaServiceInternal as GrpcMetaServiceInternal,
        MetaServiceInternalServer as GrpcMetaServiceInternalServer,
    },
    worker_service_client::WorkerServiceClient as GrpcWorkerServiceClient,
    worker_service_server::{
        WorkerService as GrpcWorkerService, WorkerServiceServer as GrpcWorkerServiceServer,
    },
};
pub use tidb_types::{
    CIStr, ColumnInfo as TiDBColumnInfo, FieldType as TiDBFieldType, FullTextIndexInfo,
    FullTextParserType, HybridIndexInfo, HybridInverted, HybridSharding, HybridSort,
    IndexColumn as TiDBIndexColumn, IndexInfo as TiDBIndexInfo, IndexType as TiDBIndexType,
    IndexType, TableInfo as TiDBTableInfo,
};

impl FragMeta {
    pub fn new(
        format_version: u64,
        frag_path: impl Into<String>,
        property: FragProperty,
        storage_refs: Vec<FragStorageRef>,
        seg_count: u32,
    ) -> Result<Self> {
        let frag_path = frag_path.into();
        Self::validate_frag_path(&frag_path)?;
        Self::validate_property(&property)?;
        Self::validate_storage_refs_with_property(&property, &storage_refs)?;
        // Do not require seg_count > 0: delete-only fragments persist row entries
        // for delete marks, but Tantivy does not create a searchable segment when
        // no live documents are indexed.

        Ok(Self {
            format_version,
            frag_path,
            property: Some(property),
            storage_refs,
            seg_count,
        })
    }

    pub fn validate(&self) -> Result<&FragProperty> {
        Self::validate_frag_path(&self.frag_path)?;
        let property = self.property.as_ref().context("property is empty")?;
        Self::validate_property(property)?;
        Self::validate_storage_refs_with_property(property, &self.storage_refs)?;
        Ok(property)
    }

    fn validate_frag_path(frag_path: &str) -> Result<()> {
        let frag_name = frag_path.rsplit('/').next().unwrap_or_default();
        let tso = frag_name
            .strip_prefix("f_")
            .with_context(|| format!("invalid frag_path format: {frag_path}"))?;
        ensure!(!tso.is_empty(), "frag_path tso is empty: {frag_path}");
        let tso = tso
            .parse::<u64>()
            .with_context(|| format!("invalid frag_path tso: {frag_path}"))?;
        ensure!(tso > 0, "frag_path tso must be > 0: {frag_path}");
        Ok(())
    }

    fn validate_property(property: &FragProperty) -> Result<()> {
        ensure!(
            !property.split_handles.is_empty(),
            "split_handles must not be empty"
        );

        let mut previous = property.min_handle.as_slice();
        for split_handle in &property.split_handles {
            ensure!(
                previous <= split_handle.as_slice(),
                "split_handles must be sorted between min_handle and max_handle"
            );
            previous = split_handle.as_slice();
        }
        ensure!(
            previous <= property.max_handle.as_slice(),
            "max_handle must be >= the last split handle"
        );
        Ok(())
    }

    fn validate_storage_refs_with_property(
        property: &FragProperty,
        storage_refs: &[FragStorageRef],
    ) -> Result<()> {
        Self::validate_storage_refs_slice(storage_refs)?;
        Self::validate_storage_ref_sizes_sum_matches_property(property, storage_refs)
    }

    fn validate_storage_ref_sizes_sum_matches_property(
        property: &FragProperty,
        storage_refs: &[FragStorageRef],
    ) -> Result<()> {
        let frag_size = storage_refs
            .iter()
            .try_fold(0u64, |sum, sr| sum.checked_add(sr.size))
            .with_context(|| format!(
                "sum of FragStorageRef::size overflows u64, storage_refs={:?}, property.size={}, property.count={}",
                storage_refs,
                property.size, property.count
            ))?;
        ensure!(
            frag_size == property.size,
            "FragProperty::size ({}) must equal sum of FragStorageRef::size ({}), property.count={}",
            property.size,
            frag_size,
            property.count
        );
        Ok(())
    }

    fn validate_storage_refs_slice(storage_refs: &[FragStorageRef]) -> Result<()> {
        ensure!(
            !storage_refs.is_empty(),
            "FragMeta::storage_refs must not be empty"
        );
        let mut seen_keys = HashSet::with_capacity(storage_refs.len());
        for storage_ref in storage_refs {
            ensure!(
                !storage_ref.sub_key.is_empty(),
                "FragStorageRef::sub_key must not be empty"
            );
            ensure!(
                !storage_ref.sub_key.starts_with('/') && !storage_ref.sub_key.ends_with('/'),
                "FragStorageRef::sub_key must be a relative path, got: {}. Must not start or end with '/'",
                storage_ref.sub_key
            );
            ensure!(
                storage_ref
                    .sub_key
                    .split('/')
                    .all(|part| !part.is_empty() && part != "." && part != ".."),
                "FragStorageRef::sub_key must not contain empty, '.' or '..' path components, got: {}",
                storage_ref.sub_key
            );
            ensure!(
                storage_ref.size > 0,
                "FragStorageRef::size must be > 0, got sub_key={}",
                storage_ref.sub_key
            );
            ensure!(
                storage_ref.offset.checked_add(storage_ref.size).is_some(),
                "FragStorageRef::offset + size must not overflow u64, sub_key={}, offset={}, size={}",
                storage_ref.sub_key,
                storage_ref.offset,
                storage_ref.size,
            );
            ensure!(
                seen_keys.insert(storage_ref.sub_key.as_str()),
                "FragStorageRef::sub_key must be unique, duplicated sub_key={}",
                storage_ref.sub_key
            );
        }
        Ok(())
    }

    pub fn bytes(&self) -> u64 {
        self.property.as_ref().unwrap().size
    }

    pub fn bytes_mb_f64(&self) -> f64 {
        self.bytes() as f64 / 1024.0 / 1024.0
    }

    pub fn rows(&self) -> u64 {
        self.property.as_ref().unwrap().count
    }

    pub fn seg_count(&self) -> usize {
        self.seg_count as usize
    }
}

impl heapsize::HeapSizeOf for FragProperty {
    fn heap_size_of_children(&self) -> usize {
        self.min_handle.capacity()
            + self.split_handles.heap_size_of_children()
            + self.max_handle.capacity()
    }
}

impl heapsize::HeapSizeOf for FragStorageRef {
    fn heap_size_of_children(&self) -> usize {
        self.sub_key.capacity()
            + std::mem::size_of_val(&self.offset)
            + std::mem::size_of_val(&self.size)
    }
}

impl heapsize::HeapSizeOf for FragMeta {
    fn heap_size_of_children(&self) -> usize {
        self.frag_path.capacity()
            + self.storage_refs.heap_size_of_children()
            + self.property.heap_size_of_children()
    }
}

impl FragProperty {
    pub fn format_split_handles(&self) -> String {
        format!(
            "[{}]{}",
            self.split_handles.len(),
            self.split_handles
                .iter()
                .map(hex::encode)
                .collect::<Vec<_>>()
                .join("|")
        )
    }
}

#[cfg(test)]
mod tests {
    use super::{FragMeta, FragProperty, FragStorageRef};

    fn property(min_handle: &[u8], split_handles: Vec<Vec<u8>>, max_handle: &[u8]) -> FragProperty {
        FragProperty {
            size: 1,
            count: 1,
            min_handle: min_handle.to_vec(),
            split_handles,
            max_handle: max_handle.to_vec(),
            delete_count: 0,
        }
    }

    fn storage_refs() -> Vec<FragStorageRef> {
        vec![FragStorageRef {
            sub_key: "object".to_string(),
            offset: 0,
            size: 1,
        }]
    }

    #[test]
    fn test_frag_meta_new_accepts_valid_args() {
        let meta = FragMeta::new(
            1,
            "parent/f_42",
            property(b"a", vec![b"b".to_vec()], b"c"),
            storage_refs(),
            1,
        )
        .unwrap();
        assert_eq!(meta.frag_path, "parent/f_42");
        assert_eq!(meta.validate().unwrap().count, 1);

        assert!(
            FragMeta::new(
                1,
                "f_42",
                property(b"a", vec![b"b".to_vec()], b"c"),
                storage_refs(),
                1
            )
            .is_ok()
        );
    }

    #[test]
    fn test_frag_meta_new_rejects_invalid_frag_path() {
        for frag_path in ["", "foo", "foo/f_0", "foo/f_x", "foo/f_1/"] {
            assert!(
                FragMeta::new(
                    1,
                    frag_path,
                    property(b"a", vec![b"b".to_vec()], b"c"),
                    storage_refs(),
                    1
                )
                .is_err(),
                "frag_path should be rejected: {frag_path}"
            );
        }
    }

    #[test]
    fn test_frag_meta_new_rejects_invalid_property_range() {
        assert!(FragMeta::new(1, "f_1", property(b"a", vec![], b"c"), storage_refs(), 1).is_err());
        assert!(
            FragMeta::new(
                1,
                "f_1",
                property(b"b", vec![b"a".to_vec()], b"c"),
                storage_refs(),
                1
            )
            .is_err()
        );
        assert!(
            FragMeta::new(
                1,
                "f_1",
                property(b"a", vec![b"c".to_vec(), b"b".to_vec()], b"d"),
                storage_refs(),
                1
            )
            .is_err()
        );
        assert!(
            FragMeta::new(
                1,
                "f_1",
                property(b"a", vec![b"d".to_vec()], b"c"),
                storage_refs(),
                1
            )
            .is_err()
        );
    }

    #[test]
    fn test_frag_meta_new_rejects_empty_storage_refs() {
        assert!(
            FragMeta::new(
                1,
                "f_1",
                property(b"a", vec![b"b".to_vec()], b"c"),
                Vec::new(),
                1
            )
            .is_err()
        );
    }

    #[test]
    fn test_frag_meta_new_rejects_invalid_storage_ref_elements() {
        let make = |refs: Vec<FragStorageRef>, property_size: u64| {
            let mut p = property(b"a", vec![b"b".to_vec()], b"c");
            p.size = property_size;
            FragMeta::new(1, "f_1", p, refs, 1)
        };

        assert!(
            make(
                vec![FragStorageRef {
                    sub_key: String::new(),
                    offset: 0,
                    size: 1,
                }],
                1,
            )
            .is_err(),
            "empty sub_key should be rejected"
        );

        assert!(
            make(
                vec![FragStorageRef {
                    sub_key: "/abs/path".to_string(),
                    offset: 0,
                    size: 1,
                }],
                1,
            )
            .is_err(),
            "absolute path sub_key should be rejected"
        );

        assert!(
            make(
                vec![FragStorageRef {
                    sub_key: "frag/../multi.0".to_string(),
                    offset: 0,
                    size: 1,
                }],
                1,
            )
            .is_err(),
            "parent path component should be rejected"
        );

        assert!(
            make(
                vec![FragStorageRef {
                    sub_key: "object".to_string(),
                    offset: 0,
                    size: 0,
                }],
                1,
            )
            .is_err(),
            "size == 0 should be rejected"
        );

        assert!(
            make(
                vec![FragStorageRef {
                    sub_key: "object".to_string(),
                    offset: u64::MAX,
                    size: 1,
                }],
                1,
            )
            .is_err(),
            "offset + size overflow should be rejected"
        );

        assert!(
            make(
                vec![
                    FragStorageRef {
                        sub_key: "object".to_string(),
                        offset: 0,
                        size: 1,
                    },
                    FragStorageRef {
                        sub_key: "object".to_string(),
                        offset: 1,
                        size: 1,
                    },
                ],
                2,
            )
            .is_err(),
            "duplicated sub_key should be rejected"
        );

        assert!(
            make(
                vec![
                    FragStorageRef {
                        sub_key: "a".to_string(),
                        offset: 0,
                        size: 1,
                    },
                    FragStorageRef {
                        sub_key: "b".to_string(),
                        offset: 0,
                        size: 2,
                    },
                ],
                3,
            )
            .is_ok(),
            "distinct relative keys with positive size should be accepted when property.size matches ref sum"
        );

        assert!(
            make(
                vec![FragStorageRef {
                    sub_key: "only".to_string(),
                    offset: 0,
                    size: 1,
                }],
                2,
            )
            .is_err(),
            "property.size must equal sum(storage_refs.size)"
        );

        assert!(
            make(
                vec![
                    FragStorageRef {
                        sub_key: "x".to_string(),
                        offset: 0,
                        size: u64::MAX,
                    },
                    FragStorageRef {
                        sub_key: "y".to_string(),
                        offset: 0,
                        size: 2,
                    },
                ],
                3,
            )
            .is_err(),
            "sum(storage_refs.size) must not overflow u64"
        );
    }

    #[test]
    fn test_frag_meta_new_accepts_zero_seg_count() {
        assert!(
            FragMeta::new(
                1,
                "f_1",
                property(b"a", vec![b"b".to_vec()], b"c"),
                storage_refs(),
                0
            )
            .is_ok()
        );
    }
}

impl std::fmt::Debug for FragProperty {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("FragProp")
            .field_with("c/s", |f| {
                f.write_fmt(format_args!("{}/{}", self.count, self.size))
            })
            .field_with("range", |f| {
                f.write_fmt(format_args!(
                    "{}..{}",
                    &hex::encode(&self.min_handle),
                    &hex::encode(&self.max_handle)
                ))
            })
            .field_with("mid", |f| {
                f.write_fmt(format_args!("{}", &self.format_split_handles()))
            })
            .finish()
    }
}

impl std::fmt::Debug for ShardManifestHeader {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_fmt(format_args!(
            "SMH(shard_id={}/range={}/epoch={})",
            self.shard_id,
            &format!(
                "{}..{}",
                &hex::encode(&self.start_key),
                &hex::encode(&self.end_key)
            ),
            self.epoch,
        ))
    }
}

impl std::fmt::Debug for KeyRange {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_fmt(format_args!(
            "Range({}..{})",
            &hex::encode(&self.start_key),
            &hex::encode(&self.end_key)
        ))
    }
}

impl std::fmt::Debug for S3Location {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_fmt(format_args!("{}/{}", self.bucket, self.prefix))
    }
}

impl heapsize::HeapSizeOf for S3Location {
    fn heap_size_of_children(&self) -> usize {
        self.bucket.capacity() + self.prefix.capacity()
    }
}

impl ErrorCode {
    pub fn to_i32(&self) -> i32 {
        <ErrorCode as Into<i32>>::into(*self)
    }
}

impl ColumnInfo {
    pub fn is_text_type(&self) -> bool {
        self.flag & tidb_def::BINARY_FLAG == 0
            && (matches!(
                self.column_type(),
                tidb_def::COLTYPE_VARCHAR
                    | tidb_def::COLTYPE_STRING
                    | tidb_def::COLTYPE_VARSTRING
                    | tidb_def::COLTYPE_TINYBLOB
                    | tidb_def::COLTYPE_MEDIUMBLOB
                    | tidb_def::COLTYPE_LONGBLOB
                    | tidb_def::COLTYPE_BLOB
            ))
    }

    pub fn is_binary_type(&self) -> bool {
        self.flag & tidb_def::BINARY_FLAG != 0
            && (matches!(
                self.column_type(),
                tidb_def::COLTYPE_VARCHAR
                    | tidb_def::COLTYPE_STRING
                    | tidb_def::COLTYPE_VARSTRING
                    | tidb_def::COLTYPE_TINYBLOB
                    | tidb_def::COLTYPE_MEDIUMBLOB
                    | tidb_def::COLTYPE_LONGBLOB
                    | tidb_def::COLTYPE_BLOB
            ))
    }

    pub fn is_signed_int_type(&self) -> bool {
        self.flag & tidb_def::UNSIGNED_FLAG == 0
            && matches!(
                self.column_type(),
                tidb_def::COLTYPE_TINY
                    | tidb_def::COLTYPE_SHORT
                    | tidb_def::COLTYPE_LONG
                    | tidb_def::COLTYPE_LONGLONG
                    | tidb_def::COLTYPE_INT24
            )
    }

    pub fn is_unsigned_int_type(&self) -> bool {
        self.flag & tidb_def::UNSIGNED_FLAG != 0
            && matches!(
                self.column_type(),
                tidb_def::COLTYPE_TINY
                    | tidb_def::COLTYPE_SHORT
                    | tidb_def::COLTYPE_LONG
                    | tidb_def::COLTYPE_LONGLONG
                    | tidb_def::COLTYPE_INT24
            )
    }

    pub fn is_float_type(&self) -> bool {
        matches!(
            self.column_type(),
            tidb_def::COLTYPE_FLOAT | tidb_def::COLTYPE_DOUBLE
        )
    }

    pub fn is_date_time_type(&self) -> bool {
        matches!(
            self.column_type(),
            tidb_def::COLTYPE_DATETIME | tidb_def::COLTYPE_TIMESTAMP | tidb_def::COLTYPE_DATE
        )
    }

    pub fn is_decimal_type(&self) -> bool {
        matches!(self.column_type(), tidb_def::COLTYPE_NEWDECIMAL)
    }

    pub fn column_type(&self) -> i32 {
        self.r#type
    }

    pub fn is_supported_fulltext_index_type(&self) -> bool {
        self.is_text_type()
    }

    // If more index type is supported, add here.
    pub fn is_supported_hybrid_index_type(&self) -> bool {
        self.is_text_type()
            || self.is_signed_int_type()
            || self.is_unsigned_int_type()
            || self.is_float_type()
            || self.is_date_time_type()
            || self.is_decimal_type()
    }

    // If more pk type is supported, add here.
    pub fn is_supported_pk_type(&self) -> bool {
        self.is_text_type()
            || self.is_signed_int_type()
            || self.is_unsigned_int_type()
            || self.is_date_time_type()
            || self.is_binary_type()
    }

    pub fn is_nullable(&self) -> bool {
        self.flag & tidb_def::NOT_NULL_FLAG == 0
    }
}

const SUPPORTED_COLLATIONS: &[&str] = &[
    "utf8mb4_bin",
    "utf8_bin",
    "utf8mb3_bin",
    "ascii_bin",
    "latin1_bin",
];
impl TableInfo {
    pub fn validate(&self) -> anyhow::Result<()> {
        if !self.is_clustered {
            bail!("Only clustered index is supported.");
        }
        let mut has_primary_key = false;
        for col in self.columns.iter() {
            if col.is_primary_key {
                has_primary_key = true;
                if !col.is_supported_pk_type() {
                    bail!("Primary key column type not supported.");
                }
            }
        }
        if !has_primary_key {
            bail!("Primary key must not be empty.");
        }

        Ok(())
    }

    pub fn pk_columns(&self) -> Vec<ColumnInfo> {
        self.columns
            .iter()
            .filter(|col| col.is_primary_key)
            .cloned()
            .collect()
    }
}

impl IndexInfo {
    pub fn validate(&self, table_info: &TableInfo) -> anyhow::Result<()> {
        ensure!(
            self.is_fulltext_index() || self.is_hybrid_index(),
            "Unsupported index type: {}. Only 'Fulltext' and 'Hybrid' indexes are supported.",
            self.index_type
        );

        ensure!(
            !self.columns.is_empty(),
            "Index definition must include at least one column."
        );

        self.ensure_no_duplicate_columns()?;

        if self.is_hybrid_index() {
            self.ensure_supported_hybrid_index_columns()?;
            self.ensure_supported_collations()?;
        }
        if self.is_fulltext_index() {
            ensure!(
                !table_info.is_partition,
                "Fulltext index is not supported for partition tables."
            );
            self.ensure_supported_fulltext_index_columns()?;
        }

        Ok(())
    }

    fn ensure_no_duplicate_columns(&self) -> anyhow::Result<()> {
        let mut seen_column_ids = HashSet::new();
        for col in self.columns.iter() {
            if !seen_column_ids.insert(col.column_id) {
                bail!(
                    "Index definition contains duplicate column '{}'.",
                    col.column_name
                );
            }
        }
        Ok(())
    }

    fn ensure_supported_hybrid_index_columns(&self) -> anyhow::Result<()> {
        for col in self.columns.iter() {
            if !col.is_supported_hybrid_index_type() {
                bail!(
                    "Unsupported data type for index column '{}'.",
                    col.column_name
                );
            }
        }
        Ok(())
    }

    fn ensure_supported_fulltext_index_columns(&self) -> anyhow::Result<()> {
        for col in self.columns.iter() {
            if !col.is_supported_fulltext_index_type() {
                bail!(
                    "Unsupported data type for index column '{}'.",
                    col.column_name
                );
            }
        }
        Ok(())
    }

    fn ensure_supported_collations(&self) -> anyhow::Result<()> {
        for col in self.columns.iter() {
            if col.is_text_type() && !SUPPORTED_COLLATIONS.contains(&col.collation.as_str()) {
                bail!(
                    "Unsupported collation '{}' for column '{}'. Only {} are supported.",
                    col.collation,
                    col.column_name,
                    SUPPORTED_COLLATIONS.join(", ")
                );
            }
        }
        Ok(())
    }

    fn is_fulltext_index(&self) -> bool {
        self.index_type == tidb_types::IndexType::Fulltext as i32
    }

    fn is_hybrid_index(&self) -> bool {
        self.index_type == tidb_types::IndexType::Hybrid as i32
    }
}

impl std::fmt::Debug for FragMeta {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let mut part = f.debug_struct("Frag");
        part.field("fv", &self.format_version)
            .field_with("P", |f| f.write_fmt(format_args!("{}", &self.frag_path)))
            .field("seg_count", &self.seg_count)
            .field_with("sr", |f| {
                f.write_fmt(format_args!("{:?}", self.storage_refs))
            });
        if let Some(x) = &self.property {
            part.field("prop", x);
        }
        part.finish()
    }
}

pub trait CreateIndexRequestExt {
    fn get_grpc_table_info_and_index_info(
        &self,
    ) -> Result<(
        crate::TableInfo,
        crate::IndexInfo,
        crate::PartitionTableInfo,
    )>;
}

impl CreateIndexRequestExt for CreateIndexRequest {
    fn get_grpc_table_info_and_index_info(
        &self,
    ) -> Result<(
        crate::TableInfo,
        crate::IndexInfo,
        crate::PartitionTableInfo,
    )> {
        let full_table_info: crate::tidb_types::TableInfo =
            serde_json::from_slice(&self.table_info).context("failed to parse table info")?;
        let keyspace_id = self.keyspace_id;

        let mut table_info = full_table_info.get_grpc_table_info(keyspace_id);
        table_info.database_name = self.database_name.clone();

        let mut index_info = full_table_info
            .get_grpc_index_info(self.index_id, keyspace_id, self.parser_info.clone())
            .ok_or_else(|| anyhow!("index_id {} not found", self.index_id))?;
        index_info.keyspace_id = keyspace_id;

        let partition_table_info = full_table_info.get_grpc_partition_table_info();

        Ok((table_info, index_info, partition_table_info))
    }
}

impl std::fmt::Debug for CreateIndexRequest {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        if let Ok(x) = std::str::from_utf8(&self.table_info) {
            f.debug_struct("CreateIndexRequest")
                .field("keyspace_id", &self.keyspace_id)
                .field("database_name", &self.database_name)
                .field("index_id", &self.index_id)
                .field("table_info", &x)
                .field("parse_info", &self.parser_info)
                .finish()
        } else {
            f.debug_struct("CreateIndexRequest")
                .field("keyspace_id", &self.keyspace_id)
                .field("database_name", &self.database_name)
                .field("index_id", &self.index_id)
                .field("encoded_table_info", &self.table_info)
                .field("parse_info", &self.parser_info)
                .finish()
        }
    }
}

impl std::fmt::Debug for ShardProgress {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_fmt(format_args!("{}", self.last_cdc_file))
    }
}

impl ShardManifestHeader {
    pub fn to_long_ident(&self) -> String {
        format!(
            "(shard_id={}, start_key={})",
            self.shard_id,
            hex::encode(self.start_key.as_slice()),
        )
    }
}

impl From<&IndexInfo> for IndexKey {
    fn from(index_info: &IndexInfo) -> Self {
        Self {
            keyspace_id: index_info.keyspace_id,
            table_id: index_info.table_id,
            index_id: index_info.index_id,
        }
    }
}

impl FinishCompactFragRequest {
    pub fn get_index_key(&self) -> IndexKey {
        IndexKey::new(self.keyspace_id, self.table_id, self.index_id)
    }
}

impl CompactFragRequest {
    pub fn get_index_key(&self) -> IndexKey {
        IndexKey::new(self.keyspace_id, self.table_id, self.index_id)
    }
}

impl GetShardLocalCacheRequest {
    pub fn get_index_key(&self) -> IndexKey {
        IndexKey::new(self.keyspace_id, self.table_id, self.index_id)
    }
}

impl std::fmt::Debug for AppendDeltaFragMetaRequest {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("AppendDeltaFragMetaRequest")
            .field(
                "index",
                &IndexKey::new(self.keyspace_id, self.table_id, self.index_id),
            )
            .field("shard_id", &self.shard_id)
            .field("frag_metas", &self.frag_metas)
            .field("worker_node_addr", &self.worker_node_addr)
            .field("progress", &self.progress)
            .finish()
    }
}

impl std::fmt::Debug for ShardLocalCacheInfo {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_fmt(format_args!(
            "SLCI({:?}=>[{}])",
            self.shard,
            self.local_cache_addrs.to_vec().join(";")
        ))
    }
}

impl ParserInfo {
    pub fn get_innodb_ft_enable_stopword(&self) -> Result<bool> {
        let x = self
            .parser_params
            .get("innodb_ft_enable_stopword")
            .context("innodb_ft_enable_stopword not found")?;
        match x.to_ascii_lowercase().as_str() {
            "on" => Ok(true),
            "off" => Ok(false),
            _ => bail!(
                "invalid value for innodb_ft_enable_stopword: {}. Must be 'ON' or 'OFF'.",
                x
            ),
        }
    }

    pub fn get_stopwords(&self) -> Result<Option<&[String]>> {
        let is_custom_stopwords = !self.get_innodb_ft_server_stopword_table()?.is_empty()
            || !self.get_innodb_ft_user_stopword_table()?.is_empty();
        if is_custom_stopwords {
            Ok(Some(self.stop_words.as_slice()))
        } else {
            Ok(None)
        }
    }

    fn get_innodb_ft_server_stopword_table(&self) -> Result<&String> {
        self.parser_params
            .get("innodb_ft_server_stopword_table")
            .context("innodb_ft_server_stopword_table not found")
    }

    fn get_innodb_ft_user_stopword_table(&self) -> Result<&String> {
        self.parser_params
            .get("innodb_ft_user_stopword_table")
            .context("innodb_ft_user_stopword_table not found")
    }

    pub fn get_innodb_ft_max_token_size(&self) -> Result<usize> {
        self.parse_usize("innodb_ft_max_token_size")
    }

    pub fn get_innodb_ft_min_token_size(&self) -> Result<usize> {
        self.parse_usize("innodb_ft_min_token_size")
    }

    pub fn get_ngram_token_size(&self) -> Result<usize> {
        self.parse_usize("ngram_token_size")
    }

    fn parse_usize(&self, key: &str) -> Result<usize> {
        let x = self
            .parser_params
            .get(key)
            .with_context(|| format!("{} not found", key))?;
        x.parse::<usize>()
            .with_context(|| format!("failed to parse {}: {}", key, x))
    }
}

lazy_static! {
    pub static ref DEFAULT_PARSER_INFO: ParserInfo = {
        // https://dev.mysql.com/doc/refman/8.0/en/innodb-parameters.html#sysvar_innodb_ft_min_token_size
        let my_default_min_token_length = 3;
        // https://dev.mysql.com/doc/refman/8.0/en/innodb-parameters.html#sysvar_innodb_ft_max_token_size
        let my_default_max_token_length = 84;
        // https://dev.mysql.com/doc/refman/8.0/en/server-system-variables.html#sysvar_ngram_token_size
        let my_default_ngram_token_size = 2;

        let mut parser_params = HashMap::new();
        parser_params.insert("innodb_ft_enable_stopword".to_string(), "on".to_string());
        parser_params.insert(
            "innodb_ft_min_token_size".to_string(),
            my_default_min_token_length.to_string(),
        );
        parser_params.insert(
            "innodb_ft_max_token_size".to_string(),
            my_default_max_token_length.to_string(),
        );
        parser_params.insert(
            "ngram_token_size".to_string(),
            my_default_ngram_token_size.to_string(),
        );
        parser_params.insert(
            "innodb_ft_server_stopword_table".to_string(),
            "".to_string(),
        );
        parser_params.insert(
            "innodb_ft_user_stopword_table".to_string(),
            "".to_string(),
        );
        ParserInfo {
            parser_type: ParserType::DefaultParser as i32,
            parser_params,
            stop_words: vec![],
        }
    };
}