use anyhow::bail;
use tici_common::{
    codec::codec_v2::{KEYSPACE_PREFIX_LEN, remove_keyspace_prefix},
    encode_binary_for_canal_json,
    keyspace::NULLSPACEID,
};
use tici_grpc::{
    IndexInfo as GrpcIndexInfo, TableInfo as GrpcTableInfo, tidb_def, tidb_types,
    util::tipb_column_info_to_field_type,
};
use tici_shard::{
    utils::{key::ShardingKey, schema::IndexMeta, tantivy_doc::RawDoc},
    writer::cdc_log::OperationType,
};
use tidb_query_datatype::{
    codec::{
        Datum,
        batch::{LazyBatchColumn, LazyBatchColumnVec},
        data_type::ScalarValueRef,
        datum::{self, DatumDecoder},
        table,
    },
    expr::EvalContext,
};
use tipb::{ColumnInfo, FieldType, IndexInfo};

use crate::index_scan_executor::{
    DecodeHandleStrategy, ImportDecodeFlags, IndexScanExecutorImpl, ScanExecutorImpl,
};

type DecodedColumns = Vec<(i64, Option<String>)>;

fn is_binary_type(col_info: &ColumnInfo) -> bool {
    (col_info.get_flag() as u32) & tidb_def::BINARY_FLAG != 0
        && matches!(
            col_info.get_tp(),
            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
        )
}

// Currently, data is encoded in the same format as CDC canal json here, and the
// logic for converting to Tantivy documents is reused via RawDoc. A more
// performant implementation might be to implement a custom trait RowDataAccess.
fn encode_value_for_canal_json(value: &Datum, col_info: &ColumnInfo) -> anyhow::Result<String> {
    match value {
        Datum::Bytes(bs) if is_binary_type(col_info) => Ok(encode_binary_for_canal_json(bs)),
        _ => value.to_string().map_err(anyhow::Error::msg),
    }
}

fn decode_column_value(
    ctx: &mut EvalContext,
    column: &LazyBatchColumn,
    col_info: &ColumnInfo,
) -> anyhow::Result<Option<String>> {
    if column.is_empty() {
        return Err(anyhow::anyhow!("Column is empty"));
    }
    if column.is_raw() {
        let raw = column
            .raw()
            .last()
            .ok_or_else(|| anyhow::anyhow!("Column is empty"))?;
        if raw == datum::DATUM_DATA_NULL {
            return Ok(None);
        }
        let mut datum_value = raw;
        let value = table::decode_col_value(&mut datum_value, ctx, col_info)?;
        Ok(Some(encode_value_for_canal_json(&value, col_info)?))
    } else {
        debug_assert!(column.is_decoded());
        let idx = column.len() - 1;
        let scalar = column.decoded().get_scalar_ref(idx);
        let owned = scalar.to_owned();
        if owned.is_none() {
            return Ok(None);
        }
        let field_type = tipb_column_info_to_field_type(col_info);
        let mut encoded = Vec::new();
        scalar.encode(&field_type, ctx, &mut encoded)?;
        let mut datum_value = encoded.as_slice();
        let value = table::decode_col_value(&mut datum_value, ctx, col_info)?;
        Ok(Some(encode_value_for_canal_json(&value, col_info)?))
    }
}

fn infer_pk_index_info(
    table_info: &GrpcTableInfo,
    index_info: &GrpcIndexInfo,
) -> Option<IndexInfo> {
    let pk_columns = table_info.pk_columns();
    if pk_columns.is_empty() {
        return None;
    }

    let mut pk_index_info = IndexInfo::default();
    pk_index_info.set_table_id(index_info.table_id);
    pk_index_info.set_index_id(-1);
    pk_index_info.set_columns(
        pk_columns
            .iter()
            .map(crate::file::grpc_column_info_to_tipb)
            .collect::<anyhow::Result<Vec<_>>>()
            .ok()?
            .into(),
    );
    Some(pk_index_info)
}

fn infer_handle_mode(table_info: &GrpcTableInfo) -> (bool, bool) {
    let pk_columns = table_info.pk_columns();
    let is_int_handle = pk_columns.len() == 1
        && pk_columns
            .first()
            .is_some_and(|column| column.is_signed_int_type() || column.is_unsigned_int_type());
    let is_common_handle = !pk_columns.is_empty() && !is_int_handle;
    (is_int_handle, is_common_handle)
}

pub fn build_sharding_key_from_import_key(
    keyspace_id: u32,
    table_info: &GrpcTableInfo,
    index_info: &GrpcIndexInfo,
    key: &[u8],
) -> anyhow::Result<Vec<u8>> {
    let index_meta = IndexMeta::new(table_info, index_info)?;
    let pk_index_info = infer_pk_index_info(table_info, index_info);
    let (pk_is_handle, is_common_handle) = infer_handle_mode(table_info);
    let tipb_index_info = crate::file::grpc_index_info_to_tipb(index_info)?;
    let options = DecoderOptions::from_table_info(
        keyspace_id,
        0,
        &index_meta,
        tipb_index_info,
        index_info.index_type,
        pk_index_info,
        pk_is_handle,
        is_common_handle,
    )?;
    let mut decoder = Decoder::new(EvalContext::default(), options);
    decoder.build_presplit_sharding_key(key)
}

/// Immutable inputs to construct a `Decoder` from table/index metadata.
pub struct DecoderOptions {
    /// Keyspace ID of the import file.
    pub keyspace_id: u32,
    /// Commit timestamp for all KV pairs in this file.
    pub commit_ts: u64,
    /// Index info for the target index.
    pub index_info: IndexInfo,
    /// Index type for the target index. Distinguishes between fulltext index
    /// and hybrid index.
    pub index_type: i32,
    /// Optional PK index info for handle decoding.
    pub pk_index_info: Option<IndexInfo>,
    /// Whether PK is int handle.
    pub pk_is_handle: bool,
    /// Whether table uses common handle.
    pub is_common_handle: bool,
    /// Number of columns encoded in the index key.
    pub key_columns_len: usize,
}

impl DecoderOptions {
    /// Build decoder options using index meta to derive key column count.
    #[allow(clippy::too_many_arguments)]
    pub fn from_table_info(
        keyspace_id: u32,
        commit_ts: u64,
        index_meta: &IndexMeta,
        index_info: IndexInfo,
        index_type: i32,
        pk_index_info: Option<IndexInfo>,
        pk_is_handle: bool,
        is_common_handle: bool,
    ) -> anyhow::Result<Self> {
        let key_columns_len = {
            if index_type == tidb_types::IndexType::Fulltext as i32 {
                // For fulltext index, the index key contains the primary key columns
                pk_index_info
                    .as_ref()
                    .map_or(0, |pk| pk.get_columns().len())
            } else if index_type == tidb_types::IndexType::Hybrid as i32 {
                // Hybrid indexes with sharding column encode only sharding columns values in
                // the index key. Hybrid indexes without sharding column is not
                // yet supported. See docs/design/tici_import/
                // hybrid_index_kv_format.md if you need details
                // about the hybrid index key format.
                let sharding_columns_len = index_meta.get_sharding_columns().len();
                if sharding_columns_len == 0 {
                    bail!(
                        "Hybrid index without sharding columns is not supported for import: keyspace_id={} table_id={} index_id={} index_type={}",
                        keyspace_id,
                        index_info.get_table_id(),
                        index_info.get_index_id(),
                        index_type,
                    )
                } else {
                    sharding_columns_len
                }
            } else {
                bail!(
                    "Unsupported index type for import: keyspace_id={} table_id={} index_id={} index_type={}",
                    keyspace_id,
                    index_info.get_table_id(),
                    index_info.get_index_id(),
                    index_type,
                )
            }
        };
        Ok(Self {
            keyspace_id,
            commit_ts,
            index_info,
            index_type,
            pk_index_info,
            pk_is_handle,
            is_common_handle,
            key_columns_len,
        })
    }
}

/// Caches decoding metadata and reuses buffers to decode KV pairs into RawDoc.
pub struct Decoder {
    /// Keyspace ID of the import file.
    keyspace_id: u32,
    /// Commit timestamp for all KV pairs in this file.
    commit_ts: u64,
    /// Index info for the target index.
    index_info: IndexInfo,
    /// Index type for the target index. Distinguishes between fulltext index
    /// and hybrid index.
    index_type: i32,
    /// Whether table uses common handle.
    is_common_handle: bool,
    /// Optional PK index info for handle decoding.
    pk_index_info: Option<IndexInfo>,
    /// Handle decoding strategy derived from table metadata.
    decode_handle_strategy: DecodeHandleStrategy,
    /// Schema for decoding columns.
    schema: Vec<FieldType>,
    /// Column IDs for index columns.
    columns_id_without_handle: Vec<i64>,
    /// Column IDs for common-handle columns.
    columns_id_for_common_handle: Vec<i64>,
    /// TiKV index scan executor used for decoding.
    index_scan_exec: IndexScanExecutorImpl,
    /// Reused column buffers for decoding.
    columns: LazyBatchColumnVec,
}

impl std::fmt::Debug for Decoder {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        // Skip the index_scan_exec and columns fields to avoid verbose output
        f.debug_struct("Decoder")
            .field("keyspace_id", &self.keyspace_id)
            .field("commit_ts", &self.commit_ts)
            .field("is_common_handle", &self.is_common_handle)
            .field("decode_handle_strategy", &self.decode_handle_strategy)
            .field("schema", &self.schema)
            .field("columns_id_without_handle", &self.columns_id_without_handle)
            .field(
                "columns_id_for_common_handle",
                &self.columns_id_for_common_handle,
            )
            .finish()
    }
}

impl Decoder {
    /// Create a decoder from precomputed options and evaluation context.
    pub fn new(ctx: EvalContext, options: DecoderOptions) -> Self {
        let DecoderOptions {
            keyspace_id,
            commit_ts,
            index_info,
            index_type,
            pk_index_info,
            pk_is_handle,
            is_common_handle,
            key_columns_len,
        } = options;
        let decode_handle_strategy = if pk_index_info.is_some() && is_common_handle {
            DecodeHandleStrategy::DecodeCommonHandle
        } else if pk_index_info.is_some() && pk_is_handle {
            DecodeHandleStrategy::DecodeIntHandle
        } else {
            DecodeHandleStrategy::NoDecode
        };

        let mut schema = Vec::with_capacity(
            index_info.get_columns().len()
                + pk_index_info
                    .as_ref()
                    .map_or(0, |pk| pk.get_columns().len()),
        );
        let mut columns_id_without_handle = Vec::with_capacity(index_info.get_columns().len());
        for col in index_info.get_columns().iter() {
            schema.push(tipb_column_info_to_field_type(col));
            columns_id_without_handle.push(col.get_column_id());
        }

        let mut columns_id_for_common_handle = Vec::new();
        if decode_handle_strategy != DecodeHandleStrategy::NoDecode {
            if let Some(pk_index) = pk_index_info.as_ref() {
                columns_id_for_common_handle = Vec::with_capacity(pk_index.get_columns().len());
                for col in pk_index.get_columns().iter() {
                    schema.push(tipb_column_info_to_field_type(col));
                    columns_id_for_common_handle.push(col.get_column_id());
                }
            }
        }

        log::info!(
            "keyspace_id={} commit_ts={} schema: {:?} handle_strategy {:?} key_columns_len {:?} columns_id_without_handle: {:?} columns_id_for_common_handle: {:?}",
            keyspace_id,
            commit_ts,
            schema,
            decode_handle_strategy,
            key_columns_len,
            columns_id_without_handle,
            columns_id_for_common_handle
        );

        // Force decoding index columns from restore data for fulltext/hybrid indexes.
        let force_restore_index_columns = index_type == tidb_types::IndexType::Fulltext as i32
            || index_type == tidb_types::IndexType::Hybrid as i32;
        // For fulltext/hybrid, missing restore data should fallback to key-decoded
        // values (especially for common-handle PK columns).
        let fallback_on_missing_restore_data = force_restore_index_columns;
        // Create the index scan executor for decoding
        // The init index_version is -1 and will be set according to the first KV pair.
        // Now we ignore the case that index version changes in the same import file
        // because the import file should be generated by a consistent snapshot in DXF
        // framework.
        let index_scan_exec = IndexScanExecutorImpl::new_for_import(
            ctx,
            schema.clone(),
            columns_id_without_handle.clone(),
            columns_id_for_common_handle.clone(),
            decode_handle_strategy,
            ImportDecodeFlags {
                key_columns_len,
                force_restore_index_columns,
                fallback_on_missing_restore_data,
            },
        );
        let columns = index_scan_exec.build_column_vec(1);
        Decoder {
            keyspace_id,
            commit_ts,
            index_info,
            index_type,
            is_common_handle,
            pk_index_info,
            decode_handle_strategy,
            schema,
            columns_id_without_handle,
            columns_id_for_common_handle,
            index_scan_exec,
            columns,
        }
    }

    /// Return the commit timestamp associated with this import file.
    pub fn commit_ts(&self) -> u64 {
        self.commit_ts
    }

    /// Return the index metadata used by this decoder.
    pub fn index_info(&self) -> &IndexInfo {
        &self.index_info
    }

    /// Return the optional PK index metadata used for handle decoding.
    pub fn pk_index_info(&self) -> Option<&IndexInfo> {
        self.pk_index_info.as_ref()
    }

    /// Decode one KV pair into a `RawDoc`.
    pub fn decode(&mut self, key: &[u8], value: &[u8]) -> anyhow::Result<RawDoc> {
        let key_payload = self.key_payload(key)?;
        self.assert_invariants();
        // clear the column buffers before decoding the next KV pair
        for column in self.columns.as_mut_slice() {
            column.clear();
        }
        self.columns.take_extra_common_handle_keys();

        self.index_scan_exec.process_kv_pair(
            key_payload,
            value,
            &mut self.columns,
            Some(self.commit_ts),
        )?;

        let cols = self.decode_index_columns()?;
        // Get the handle bytes according to the decoding strategy.
        let handle_bytes = self.get_handle_bytes()?;
        log::trace!("Final decoded columns with handle if any: {:?}", cols);

        // ShardingKey format:
        // - Fulltext index
        //   - [keyspace]t[tableID]_r[handleID]
        // - Hybrid index with sharding key
        //   - [keyspace]t[tableID]_i[indexID][shardingColumns...][handleID]
        let raw_sharding_key = {
            let index_key = tici_common::index::IndexKey::new(
                self.keyspace_id,
                self.index_info.get_table_id(),
                self.index_info.get_index_id(),
            );
            if self.index_type == tidb_types::IndexType::Fulltext as i32 {
                let mut raw = "_r".as_bytes().to_vec();
                let handle_bytes = handle_bytes
                    .as_ref()
                    .ok_or_else(|| anyhow::anyhow!("Fulltext handle is missing"))?;
                raw.extend_from_slice(handle_bytes);
                tici_common::codec::table_raw_key(&index_key, &raw)
            } else if self.index_type == tidb_types::IndexType::Hybrid as i32 {
                // the key format of hybrid index is the same as encoded key
                key.to_vec()
            } else {
                bail!(
                    "Unsupported index type for import: table_id={} index_id={} index_type={}",
                    self.index_info.get_table_id(),
                    self.index_info.get_index_id(),
                    self.index_type,
                );
            }
        };
        let sharding_key = ShardingKey::from_raw(raw_sharding_key);
        Ok(RawDoc {
            handle: sharding_key,
            version: self.commit_ts,
            op_type: OperationType::Insert,
            cols,
            is_local_timezone: false, // Timestamps of `import into` are in utc timezone
        })
    }

    fn build_presplit_sharding_key(&mut self, key: &[u8]) -> anyhow::Result<Vec<u8>> {
        let index_key = tici_common::index::IndexKey::new(
            self.keyspace_id,
            self.index_info.get_table_id(),
            self.index_info.get_index_id(),
        );
        if self.index_type == tidb_types::IndexType::Fulltext as i32 {
            let handle_bytes = self.fulltext_handle_bytes_from_key(key)?;
            let mut raw = b"_r".to_vec();
            raw.extend_from_slice(&handle_bytes);
            Ok(tici_common::codec::table_raw_key(&index_key, &raw))
        } else if self.index_type == tidb_types::IndexType::Hybrid as i32 {
            Ok(key.to_vec())
        } else {
            bail!(
                "Unsupported index type for import: table_id={} index_id={} index_type={}",
                self.index_info.get_table_id(),
                self.index_info.get_index_id(),
                self.index_type,
            );
        }
    }

    fn fulltext_handle_bytes_from_key(&self, key: &[u8]) -> anyhow::Result<Vec<u8>> {
        let key = self.key_payload(key)?;
        table::check_index_key(key)?;
        let mut key_payload = &key[table::PREFIX_LEN + table::ID_LEN..];
        let handle_column_count = match self.decode_handle_strategy {
            DecodeHandleStrategy::NoDecode => {
                bail!(
                    "Fulltext handle is missing: table_id={} index_id={}",
                    self.index_info.get_table_id(),
                    self.index_info.get_index_id(),
                )
            }
            DecodeHandleStrategy::DecodeIntHandle => 1,
            DecodeHandleStrategy::DecodeCommonHandle => self.columns_id_for_common_handle.len(),
        };

        let mut datum_slices = Vec::new();
        while !key_payload.is_empty() {
            let (datum_value, remaining) = datum::split_datum(key_payload, false)?;
            datum_slices.push(datum_value);
            key_payload = remaining;
        }
        if datum_slices.len() < handle_column_count {
            bail!(
                "Fulltext key does not contain enough handle columns: table_id={} index_id={} handle_columns={} total_datums={}",
                self.index_info.get_table_id(),
                self.index_info.get_index_id(),
                handle_column_count,
                datum_slices.len(),
            );
        }
        let handle_datums = &datum_slices[datum_slices.len() - handle_column_count..];
        match self.decode_handle_strategy {
            DecodeHandleStrategy::NoDecode => unreachable!(),
            DecodeHandleStrategy::DecodeIntHandle => {
                let mut raw_handle = handle_datums[0];
                let handle = match raw_handle.read_datum()? {
                    Datum::I64(value) => value,
                    Datum::U64(value) => value as i64,
                    value => bail!("Unexpected int handle datum {:?}", value),
                };
                if !raw_handle.is_empty() {
                    bail!(
                        "Unexpected extra bytes in int handle: {}",
                        tikv_log_wrappers::Value(raw_handle)
                    );
                }
                let mut encoded = Vec::with_capacity(std::mem::size_of::<i64>());
                tici_shard::data_types::codec::encode_int_to_cmpable(&mut encoded, handle);
                Ok(encoded)
            }
            DecodeHandleStrategy::DecodeCommonHandle => {
                let total_len = handle_datums.iter().map(|datum| datum.len()).sum();
                let mut encoded = Vec::with_capacity(total_len);
                for datum in handle_datums {
                    encoded.extend_from_slice(datum);
                }
                Ok(encoded)
            }
        }
    }

    #[inline]
    fn key_payload<'a>(&self, key: &'a [u8]) -> anyhow::Result<&'a [u8]> {
        if self.keyspace_id == NULLSPACEID {
            return Ok(key);
        }
        if key.len() < KEYSPACE_PREFIX_LEN {
            return Err(anyhow::anyhow!(
                "Key too short for keyspace prefix: len={} prefix_len={}",
                key.len(),
                KEYSPACE_PREFIX_LEN
            ));
        }
        let payload = remove_keyspace_prefix(key);
        if payload.len() == key.len() {
            return Err(anyhow::anyhow!(
                "Expected keyspace prefix for keyspace_id={}, but key does not have the expected prefix",
                self.keyspace_id
            ));
        }
        Ok(payload)
    }

    #[inline]
    fn assert_invariants(&self) {
        debug_assert_eq!(self.schema.len(), self.columns.columns_len());
        debug_assert_eq!(
            self.columns_id_without_handle.len(),
            self.index_info.get_columns().len()
        );
    }

    fn decode_index_columns(&mut self) -> anyhow::Result<DecodedColumns> {
        let cap = self.index_info.get_columns().len()
            + self
                .pk_index_info
                .as_ref()
                .map_or(0, |pk| pk.get_columns().len());
        let mut cols: DecodedColumns = Vec::with_capacity(cap);
        // Decode index columns according to the index schema.
        for (idx, col) in self.index_info.get_columns().iter().enumerate() {
            let value =
                decode_column_value(self.index_scan_exec.mut_context(), &self.columns[idx], col)?;
            cols.push((col.get_column_id(), value));
        }
        if self.decode_handle_strategy == DecodeHandleStrategy::NoDecode {
            return Ok(cols);
        }

        // Then decode handle columns according to the PK index schema and fill in
        // missing values in cols.
        if let Some(pk_index) = self.pk_index_info.as_ref() {
            // Ensure PK column metadata aligns with the decoded handle columns.
            debug_assert_eq!(
                self.columns_id_for_common_handle.len(),
                pk_index.get_columns().len()
            );
            // PK values, if decoded, are appended after index columns in `self.columns`.
            let offset = self.index_info.get_columns().len();
            for (col_idx, info) in pk_index.get_columns().iter().enumerate() {
                // Decode the PK column value from the cached column buffer.
                let value = decode_column_value(
                    self.index_scan_exec.mut_context(),
                    &self.columns[offset + col_idx],
                    info,
                )?;
                match cols
                    .iter_mut()
                    .find(|(cid, _)| *cid == info.get_column_id())
                {
                    Some((_, slot)) => {
                        // If the index column already exists, only fill missing values.
                        if slot.is_none() {
                            *slot = value;
                        }
                    }
                    // Append PK columns that are not already present in the index columns.
                    None => cols.push((info.get_column_id(), value)),
                }
            }
        }
        Ok(cols)
    }

    fn get_handle_bytes(&self) -> anyhow::Result<Option<Vec<u8>>> {
        let handle_bytes = match self.decode_handle_strategy {
            DecodeHandleStrategy::NoDecode => None,
            DecodeHandleStrategy::DecodeCommonHandle => {
                let common_handle = self
                    .columns
                    .get_extra_common_handle_key(0)
                    .ok_or_else(|| anyhow::anyhow!("Common handle key is missing"))?;
                Some(common_handle.to_vec())
            }
            DecodeHandleStrategy::DecodeIntHandle => {
                let int_handle_val = self.int_handle_from_columns()?;
                let mut buf = Vec::with_capacity(std::mem::size_of::<i64>());
                tici_shard::data_types::codec::encode_int_to_cmpable(&mut buf, int_handle_val);
                Some(buf)
            }
        };
        Ok(handle_bytes)
    }

    // A helper for `handle_bytes` to handle the situation that get the int handle
    // value
    fn int_handle_from_columns(&self) -> anyhow::Result<i64> {
        if self.decode_handle_strategy != DecodeHandleStrategy::DecodeIntHandle {
            return Err(anyhow::anyhow!("Decode handle strategy is not int handle"));
        }

        let handle_idx = self.columns_id_without_handle.len();
        let column = self
            .columns
            .as_slice()
            .get(handle_idx)
            .ok_or_else(|| anyhow::anyhow!("Handle column is missing"))?;
        if column.is_empty() {
            return Err(anyhow::anyhow!("Handle column is empty"));
        }

        let idx = column.len() - 1;
        if column.is_decoded() {
            match column.decoded().get_scalar_ref(idx) {
                ScalarValueRef::Int(Some(value)) => Ok(*value),
                ScalarValueRef::Int(None) => Err(anyhow::anyhow!("Handle value is null")),
                other => Err(anyhow::anyhow!(
                    "Unexpected handle scalar type: {:?}",
                    other
                )),
            }
        } else if column.is_raw() {
            let raw = column
                .raw()
                .last()
                .ok_or_else(|| anyhow::anyhow!("Handle column is empty"))?;
            if raw == datum::DATUM_DATA_NULL {
                return Err(anyhow::anyhow!("Handle value is null"));
            }
            let mut datum_value = raw;
            match datum_value.read_datum()? {
                Datum::I64(value) => Ok(value),
                Datum::U64(value) => Ok(value as i64),
                other => Err(anyhow::anyhow!("Unexpected handle datum type: {:?}", other)),
            }
        } else {
            Err(anyhow::anyhow!("Handle column is neither decoded nor raw"))
        }
    }
}

#[cfg(test)]
mod tests {
    use tici_common::{
        codec::codec_v2::add_keyspace_prefix, index::IndexKey, keyspace::NULLSPACEID,
    };
    use tici_shard::{
        data_types::codec::encode_int_to_cmpable,
        test_utils::index_meta::TEST_FT_INDEX_INFO_STD_STR,
        utils::schema::{IndexMetaStorage, IndexMetaVersion},
    };
    use tidb_query_datatype::{
        FieldTypeTp,
        codec::{Datum, datum, mysql::Decimal, table},
    };
    use tipb::ColumnInfo;

    use super::*;

    const TEST_TABLE_ID: i64 = 1;
    const TEST_INDEX_ID: i64 = 1;
    const TEST_INDEX_VALUE: &[u8] = b"0";

    fn get_test_grpc_table_info() -> GrpcTableInfo {
        let mut table_info = GrpcTableInfo {
            keyspace_id: NULLSPACEID,
            table_id: TEST_TABLE_ID,
            is_clustered: true,
            ..Default::default()
        };
        table_info.columns.push(tici_grpc::ColumnInfo {
            column_id: 1,
            r#type: tidb_def::COLTYPE_LONG,
            collation: "utf8mb4_bin".to_string(),
            is_primary_key: true,
            ..Default::default()
        });
        table_info.columns.push(tici_grpc::ColumnInfo {
            column_id: 2,
            r#type: tidb_def::COLTYPE_VARCHAR,
            collation: "utf8mb4_bin".to_string(),
            ..Default::default()
        });
        table_info
    }

    fn get_test_common_handle_grpc_table_info() -> GrpcTableInfo {
        let mut table_info = get_test_grpc_table_info();
        table_info.columns.push(tici_grpc::ColumnInfo {
            column_id: 3,
            r#type: tidb_def::COLTYPE_LONG,
            collation: "utf8mb4_bin".to_string(),
            is_primary_key: true,
            ..Default::default()
        });
        table_info
    }

    fn get_test_multi_column_grpc_table_info() -> GrpcTableInfo {
        let mut table_info = get_test_grpc_table_info();
        table_info.columns.push(tici_grpc::ColumnInfo {
            column_id: 3,
            r#type: tidb_def::COLTYPE_VARCHAR,
            collation: "utf8mb4_bin".to_string(),
            ..Default::default()
        });
        table_info.columns.push(tici_grpc::ColumnInfo {
            column_id: 4,
            r#type: tidb_def::COLTYPE_VARCHAR,
            collation: "utf8mb4_bin".to_string(),
            ..Default::default()
        });
        table_info
    }

    fn get_test_grpc_index_info() -> GrpcIndexInfo {
        let mut index_info = GrpcIndexInfo {
            keyspace_id: NULLSPACEID,
            table_id: TEST_TABLE_ID,
            index_id: TEST_INDEX_ID,
            index_type: tidb_types::IndexType::Fulltext as i32,
            fulltext_index: TEST_FT_INDEX_INFO_STD_STR.to_string(),
            ..Default::default()
        };
        index_info.columns.push(tici_grpc::ColumnInfo {
            column_id: 2,
            r#type: tidb_def::COLTYPE_VARCHAR,
            collation: "utf8mb4_bin".to_string(),
            ..Default::default()
        });
        index_info
    }

    fn get_test_multi_column_grpc_index_info() -> GrpcIndexInfo {
        let mut index_info = get_test_grpc_index_info();
        index_info.columns.push(tici_grpc::ColumnInfo {
            column_id: 3,
            r#type: tidb_def::COLTYPE_VARCHAR,
            collation: "utf8mb4_bin".to_string(),
            ..Default::default()
        });
        index_info.columns.push(tici_grpc::ColumnInfo {
            column_id: 4,
            r#type: tidb_def::COLTYPE_VARCHAR,
            collation: "utf8mb4_bin".to_string(),
            ..Default::default()
        });
        index_info
    }

    fn get_test_index_info() -> IndexInfo {
        let mut index_info = IndexInfo::default();
        index_info.set_table_id(TEST_TABLE_ID);
        index_info.set_index_id(TEST_INDEX_ID);
        let mut col: ColumnInfo = FieldTypeTp::VarChar.into();
        col.set_column_id(2);
        index_info.mut_columns().push(col);
        index_info
    }

    fn get_test_pk_index_info() -> Option<IndexInfo> {
        let mut index_info = IndexInfo::default();
        index_info.set_table_id(TEST_TABLE_ID);
        index_info.set_index_id(TEST_INDEX_ID);
        let mut col: ColumnInfo = FieldTypeTp::LongLong.into();
        col.set_column_id(1);
        index_info.mut_columns().push(col);
        Some(index_info)
    }

    fn get_test_decimal_pk_index_info() -> Option<IndexInfo> {
        let mut index_info = IndexInfo::default();
        index_info.set_table_id(TEST_TABLE_ID);
        index_info.set_index_id(TEST_INDEX_ID);
        let mut col: ColumnInfo = FieldTypeTp::NewDecimal.into();
        col.set_column_id(3);
        index_info.mut_columns().push(col);
        Some(index_info)
    }

    fn build_index_meta() -> IndexMeta {
        let storage = IndexMetaStorage {
            version: IndexMetaVersion::LATEST,
            index_key: IndexKey::new(NULLSPACEID, TEST_TABLE_ID, TEST_INDEX_ID),
            created_ts: 0,
            columns: Vec::new(),
            index_column_count: 0,
            sort_columns: Vec::new(),
            inverted_columns: Vec::new(),
            fulltext_fields: Vec::new(),
            sharding_columns: Vec::new(),
            analyzer_defs: Vec::new(),
            is_partition: false,
        };
        IndexMeta::try_from(storage).unwrap()
    }

    fn encode_index_key(values: &[Datum]) -> Vec<u8> {
        let encoded = datum::encode_key(&mut EvalContext::default(), values).unwrap();
        table::encode_index_seek_key(TEST_TABLE_ID, TEST_INDEX_ID, &encoded)
    }

    fn encode_int_handle_bytes(handle: i64) -> Vec<u8> {
        let mut buf = Vec::new();
        encode_int_to_cmpable(&mut buf, handle);
        buf
    }

    fn build_fulltext_sharding_key(keyspace_id: u32, handle_bytes: &[u8]) -> Vec<u8> {
        let index_key = IndexKey::new(keyspace_id, TEST_TABLE_ID, TEST_INDEX_ID);
        let mut suffix = b"_r".to_vec();
        suffix.extend_from_slice(handle_bytes);
        tici_common::codec::table_raw_key(&index_key, &suffix)
    }

    #[test]
    fn build_sharding_key_from_import_key_decodes_fulltext_handle() {
        let table_info = get_test_grpc_table_info();
        let index_info = get_test_grpc_index_info();
        let key = encode_index_key(&[Datum::Bytes(b"abc".to_vec()), Datum::I64(100)]);

        let sharding_key =
            build_sharding_key_from_import_key(NULLSPACEID, &table_info, &index_info, &key)
                .unwrap();

        let expected_key = build_fulltext_sharding_key(NULLSPACEID, &encode_int_handle_bytes(100));
        assert_eq!(sharding_key, expected_key);
    }

    #[test]
    fn build_sharding_key_from_import_key_decodes_common_handle() {
        let table_info = get_test_common_handle_grpc_table_info();
        let index_info = get_test_grpc_index_info();
        let key = encode_index_key(&[
            Datum::Bytes(b"abc".to_vec()),
            Datum::I64(100),
            Datum::I64(200),
        ]);

        let sharding_key =
            build_sharding_key_from_import_key(NULLSPACEID, &table_info, &index_info, &key)
                .unwrap();

        let handle_bytes = datum::encode_key(
            &mut EvalContext::default(),
            &[Datum::I64(100), Datum::I64(200)],
        )
        .unwrap();
        let expected_key = build_fulltext_sharding_key(NULLSPACEID, &handle_bytes);
        assert_eq!(sharding_key, expected_key);
    }

    #[test]
    fn build_sharding_key_from_import_key_skips_fulltext_column_decode_for_presplit() {
        let table_info = get_test_multi_column_grpc_table_info();
        let index_info = get_test_multi_column_grpc_index_info();
        let key = encode_index_key(&[Datum::Bytes(b"abc".to_vec()), Datum::I64(100)]);

        let index_meta = build_index_meta();
        let tipb_index_info = crate::file::grpc_index_info_to_tipb(&index_info).unwrap();
        let options = DecoderOptions::from_table_info(
            NULLSPACEID,
            0,
            &index_meta,
            tipb_index_info,
            tidb_types::IndexType::Fulltext as i32,
            get_test_pk_index_info(),
            true,
            false,
        )
        .unwrap();
        let mut decoder = Decoder::new(EvalContext::default(), options);
        let err = decoder.decode(&key, TEST_INDEX_VALUE).unwrap_err();
        assert!(
            err.to_string().contains("2th column is missing value"),
            "unexpected error: {err:#}"
        );

        let sharding_key =
            build_sharding_key_from_import_key(NULLSPACEID, &table_info, &index_info, &key)
                .unwrap();

        let expected_key = build_fulltext_sharding_key(NULLSPACEID, &encode_int_handle_bytes(100));
        assert_eq!(sharding_key, expected_key);
    }

    #[test]
    fn decode_state_basic() {
        let ctx = EvalContext::default();
        let index_info = get_test_index_info();
        let pk_index_info = get_test_pk_index_info();
        let index_meta = build_index_meta();
        let pk_column_id = pk_index_info.as_ref().unwrap().get_columns()[0].get_column_id();
        let commit_ts = 12345;

        let index_vals = vec![Datum::Bytes(b"abc".to_vec()), Datum::I64(100)];
        let key = encode_index_key(&index_vals);
        let mut decoder = Decoder::new(
            ctx,
            DecoderOptions::from_table_info(
                NULLSPACEID,
                commit_ts,
                &index_meta,
                index_info,
                tidb_types::IndexType::Fulltext as i32,
                pk_index_info,
                true,
                false,
            )
            .unwrap(),
        );

        let doc = decoder.decode(&key, TEST_INDEX_VALUE).unwrap();
        let expected_key = build_fulltext_sharding_key(NULLSPACEID, &encode_int_handle_bytes(100));
        assert_eq!(doc.handle, ShardingKey::from_raw(expected_key));
        assert_eq!(doc.version, commit_ts);
        assert_eq!(doc.op_type, OperationType::Insert);
        assert_eq!(doc.cols.len(), 2);
        assert_eq!(doc.cols[0], (2, Some("abc".to_string())));
        assert_eq!(doc.cols[1], (pk_column_id, Some("100".to_string())));
    }

    #[test]
    fn decode_state_decimal_pk() {
        let ctx = EvalContext::default();
        let index_info = get_test_index_info();
        let pk_index_info = get_test_decimal_pk_index_info();
        let index_meta = build_index_meta();
        let pk_column_id = pk_index_info.as_ref().unwrap().get_columns()[0].get_column_id();
        let commit_ts = 12345;

        let index_vals = vec![Datum::Bytes(b"abc".to_vec()), Datum::Dec(Decimal::zero())];
        let key = encode_index_key(&index_vals);
        let mut decoder = Decoder::new(
            ctx,
            DecoderOptions::from_table_info(
                NULLSPACEID,
                commit_ts,
                &index_meta,
                index_info,
                tidb_types::IndexType::Fulltext as i32,
                pk_index_info,
                false,
                true,
            )
            .unwrap(),
        );

        let doc = decoder.decode(&key, TEST_INDEX_VALUE).unwrap();
        let handle_bytes =
            datum::encode_key(&mut EvalContext::default(), &[Datum::Dec(Decimal::zero())]).unwrap();
        let expected_key = build_fulltext_sharding_key(NULLSPACEID, &handle_bytes);
        assert_eq!(doc.handle, ShardingKey::from_raw(expected_key));
        assert_eq!(doc.version, commit_ts);
        assert_eq!(doc.op_type, OperationType::Insert);
        assert_eq!(doc.cols.len(), 2);
        assert_eq!(doc.cols[0], (2, Some("abc".to_string())));
        assert_eq!(doc.cols[1], (pk_column_id, Some("0".to_string())));
    }

    #[test]
    fn decode_state_empty_index_info() {
        let ctx = EvalContext::default();
        let mut index_info = IndexInfo::default();
        index_info.set_table_id(TEST_TABLE_ID);
        index_info.set_index_id(TEST_INDEX_ID);
        let pk_index_info = get_test_pk_index_info();
        let index_meta = build_index_meta();
        let pk_column_id = pk_index_info.as_ref().unwrap().get_columns()[0].get_column_id();
        let commit_ts = 12345;

        let index_vals = vec![Datum::I64(100)];
        let key = encode_index_key(&index_vals);
        let mut decoder = Decoder::new(
            ctx,
            DecoderOptions::from_table_info(
                NULLSPACEID,
                commit_ts,
                &index_meta,
                index_info,
                tidb_types::IndexType::Fulltext as i32,
                pk_index_info,
                true,
                false,
            )
            .unwrap(),
        );

        let doc = decoder.decode(&key, TEST_INDEX_VALUE).unwrap();
        let expected_key = build_fulltext_sharding_key(NULLSPACEID, &encode_int_handle_bytes(100));
        assert_eq!(doc.handle, ShardingKey::from_raw(expected_key));
        assert_eq!(doc.version, commit_ts);
        assert_eq!(doc.op_type, OperationType::Insert);
        assert_eq!(doc.cols.len(), 1);
        assert_eq!(doc.cols[0], (pk_column_id, Some("100".to_string())));
    }

    #[test]
    fn decode_state_reuse_clears_columns() {
        let ctx = EvalContext::default();
        let index_info = get_test_index_info();
        let pk_index_info = get_test_pk_index_info();
        let index_meta = build_index_meta();
        let mut decoder = Decoder::new(
            ctx,
            DecoderOptions::from_table_info(
                NULLSPACEID,
                1,
                &index_meta,
                index_info,
                tidb_types::IndexType::Fulltext as i32,
                pk_index_info,
                true,
                false,
            )
            .unwrap(),
        );

        let key1 = encode_index_key(&[Datum::Bytes(b"first".to_vec()), Datum::I64(1)]);
        let key2 = encode_index_key(&[Datum::Bytes(b"second".to_vec()), Datum::I64(2)]);

        let doc1 = decoder.decode(&key1, TEST_INDEX_VALUE).unwrap();
        let doc2 = decoder.decode(&key2, TEST_INDEX_VALUE).unwrap();

        assert_eq!(doc1.cols[0], (2, Some("first".to_string())));
        assert_eq!(doc2.cols[0], (2, Some("second".to_string())));
        for column in decoder.columns.as_slice() {
            assert_eq!(1, column.len());
        }
    }

    #[test]
    fn decode_state_keyspace_prefix() {
        let index_info = get_test_index_info();
        let pk_index_info = get_test_pk_index_info();
        let index_meta = build_index_meta();
        let mut decode_state_plain = Decoder::new(
            EvalContext::default(),
            DecoderOptions::from_table_info(
                NULLSPACEID,
                10,
                &index_meta,
                index_info.clone(),
                tidb_types::IndexType::Fulltext as i32,
                pk_index_info.clone(),
                true,
                false,
            )
            .unwrap(),
        );
        let mut decoder = Decoder::new(
            EvalContext::default(),
            DecoderOptions::from_table_info(
                1,
                10,
                &index_meta,
                index_info,
                tidb_types::IndexType::Fulltext as i32,
                pk_index_info,
                true,
                false,
            )
            .unwrap(),
        );

        let key = encode_index_key(&[Datum::Bytes(b"abc".to_vec()), Datum::I64(100)]);
        let prefixed_key = add_keyspace_prefix(1, &key);

        let doc_plain = decode_state_plain.decode(&key, TEST_INDEX_VALUE).unwrap();
        let doc_prefixed = decoder.decode(&prefixed_key, TEST_INDEX_VALUE).unwrap();

        assert_eq!(doc_plain.cols, doc_prefixed.cols);
        let expected_key = build_fulltext_sharding_key(1, &encode_int_handle_bytes(100));
        assert_eq!(doc_prefixed.handle, ShardingKey::from_raw(expected_key));
    }
}