use std::collections::HashMap;

use anyhow::{Context, Result, anyhow, bail, ensure};
use serde::Deserialize;
use tici_common::{
    cdc_file_name::CDCFileName,
    codec::{codec_v2::get_txn_keyspace_prefix, parse_table_id_from_sharding_key},
    decode_binary_from_canal_json,
    keyspace::NULLSPACEID,
};
use tici_grpc::ColumnInfo;

use crate::{
    data_types::{
        codec::{Datum, INT_HANDLE_FLAG, encode_decimal_val, encode_index_key},
        my_time::MyTime,
    },
    utils::{
        key::ShardingKey, metrics::inc_cdc_event_count, schema::IndexMeta, tantivy_doc::TantivyDoc,
    },
};

// Length of the raw table record-key prefix "t{table_id}_r"
// 1 byte ('t') + 8 bytes (table_id, comparable int) + 2 bytes ('_r') = 11
const RAW_KEY_PREFIX_LEN: usize = 11;

#[derive(Clone)]
pub enum CDCLogEntry {
    CDCData(TantivyDoc),
    Filename(CDCFileName),
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum OperationType {
    Insert,
    Update,
    Delete,
    Watermark,
}

impl OperationType {
    pub fn is_dml(&self) -> bool {
        matches!(
            self,
            OperationType::Insert | OperationType::Update | OperationType::Delete
        )
    }
}

pub trait RowDataAccess {
    fn get_sharding_key(&self, index_meta: &IndexMeta) -> Result<ShardingKey>;
    fn get_version(&self) -> Result<u64>;
    fn get_dml_type(&self) -> Result<OperationType>;

    // For CDC logs, timestamp values are in local timezone.
    // For Import into, timestamp values are in UTC timezone.
    fn is_local_timezone(&self) -> bool;

    fn get_column_value(&self, col: &ColumnInfo) -> Result<Option<&str>>;

    fn get_str(&self, col: &ColumnInfo) -> Result<Option<&str>> {
        if !col.is_text_type() {
            bail!("Column {:?} is not text type", col);
        }
        self.get_column_value(col)
    }

    fn get_binary(&self, col: &ColumnInfo) -> Result<Option<Vec<u8>>> {
        if !col.is_binary_type() {
            bail!("Column {:?} is not binary type", col);
        }

        let val_str = match self.get_column_value(col)? {
            Some(s) => s,
            None => return Ok(None),
        };

        let res = decode_binary_from_canal_json(val_str)?;
        Ok(Some(res))
    }

    fn get_u64(&self, col: &ColumnInfo) -> Result<Option<u64>> {
        if !col.is_unsigned_int_type() {
            bail!("Column {:?} is not unsigned int type", col);
        }
        self.parse_column::<u64>(col)
    }

    fn get_i64(&self, col: &ColumnInfo) -> Result<Option<i64>> {
        if !col.is_signed_int_type() {
            bail!("Column {:?} is not int type", col);
        }
        self.parse_column::<i64>(col)
    }

    fn get_f64(&self, col: &ColumnInfo) -> Result<Option<f64>> {
        if !col.is_float_type() {
            bail!("Column {:?} is not float type", col);
        }
        self.parse_column::<f64>(col)
    }

    fn get_tidb_packed_time(&self, col: &ColumnInfo) -> Result<Option<u64>> {
        if !col.is_date_time_type() {
            bail!("Column {:?} is not date time type", col);
        }
        match self.get_column_value(col)? {
            Some(datetime_str) => Ok(Some(
                MyTime::new(datetime_str, col.r#type, self.is_local_timezone())?.packed_time,
            )),
            None => Ok(None),
        }
    }

    fn get_decimal_bytes(&self, col: &ColumnInfo) -> Result<Option<Vec<u8>>> {
        if !col.is_decimal_type() {
            bail!("Column {:?} is not decimal type", col);
        }
        match self.get_column_value(col)? {
            Some(decimal_str) => {
                let mut buf = vec![];
                encode_decimal_val(
                    &mut buf,
                    decimal_str,
                    col.column_length as u8,
                    col.decimal as u8,
                )?;
                Ok(Some(buf))
            }
            None => Ok(None),
        }
    }

    fn parse_column<T: std::str::FromStr>(&self, col: &ColumnInfo) -> Result<Option<T>>
    where
        T::Err: std::fmt::Display,
    {
        match self.get_column_value(col)? {
            Some(s) => {
                let parsed = s.parse::<T>().map_err(|e| {
                    anyhow!(format!(
                        "Failed to parse column '{}', value '{}' as {}: {}",
                        col.column_name,
                        s,
                        std::any::type_name::<T>(),
                        e
                    ))
                })?;
                Ok(Some(parsed))
            }
            None => Ok(None),
        }
    }
}

pub struct ShardingKeyPair {
    pub data_sharding_key: ShardingKey,
    pub old_sharding_key: Option<ShardingKey>,
}

pub struct CDCLog {
    event: CDCEvent,
}

impl RowDataAccess for CDCLog {
    fn get_sharding_key(&self, index_meta: &IndexMeta) -> Result<ShardingKey> {
        Ok(self.create_sharding_key_pair(index_meta)?.data_sharding_key)
    }

    fn get_version(&self) -> Result<u64> {
        self.event.get_version()
    }

    fn get_dml_type(&self) -> Result<OperationType> {
        let op_type = self.event.get_operation_type()?;
        if !op_type.is_dml() {
            bail!("{:?} is not a DML operation", op_type);
        }
        Ok(op_type)
    }

    fn get_column_value(&self, col: &ColumnInfo) -> Result<Option<&str>> {
        let data = self.event.get_data()?;
        CDCEvent::get_column_value(data, col)
    }

    fn is_local_timezone(&self) -> bool {
        true
    }
}

impl CDCLog {
    // Creates a new CDCLog from a JSON string.
    // Returns:
    // - Ok(Some(CDCLog)) if the JSON represents a valid DML CDC event.
    // - Ok(None) if the JSON does not represent a valid DML CDC event.
    // - Err(anyhow::Error) if there is an error parsing the JSON or processing the
    //   event.
    pub fn new(json: &str) -> Result<Option<Self>> {
        // Validate input
        if json.is_empty() {
            bail!("CDC log JSON string is empty");
        }

        let event = CDCEvent::from_json(json)?;

        // Ignore DDL events and non-DML events, e.g., WATERMARK
        if event.is_ddl() {
            return Ok(None);
        }
        let op_type = event.get_operation_type()?;
        if !op_type.is_dml() {
            return Ok(None);
        }

        let _data = event.get_data()?;
        let old = event.get_old()?;
        if op_type == OperationType::Update && old.is_none() {
            bail!("Old data is required for UPDATE operation");
        } else if op_type != OperationType::Update && old.is_some() {
            bail!("Old data should be None for non-UPDATE operation");
        }

        Ok(Some(Self { event }))
    }

    fn create_sharding_key_pair(&self, index_meta: &IndexMeta) -> Result<ShardingKeyPair> {
        let handle = self.event.get_handle()?;
        let table_id = parse_table_id_from_sharding_key(handle.raw()).ok_or_else(|| {
            anyhow!(
                "Failed to parse table_id from sharding_key, handle={}",
                handle
            )
        })?;

        let index_key = index_meta.index_key();
        ensure!(
            table_id == index_key.table_id,
            " CDC log table_id={} != index_meta table_id={}",
            table_id,
            index_key.table_id
        );

        let data = self.event.get_data()?;
        let old = self.event.get_old()?;
        let sharding_columns = index_meta.get_sharding_columns();
        if sharding_columns.is_empty() {
            // Use handle as sharding key for non-sharding columns
            return Ok(ShardingKeyPair {
                data_sharding_key: CDCEvent::build_default_sharding_key(
                    index_key.keyspace_id,
                    handle,
                ),
                old_sharding_key: None,
            });
        }

        let data_sharding_key = CDCEvent::build_sharding_key(index_meta, data, handle)?;

        // If sharding column values are changed, we need to delete the old sharding key
        // in another shard.
        let old_sharding_key = if let Some(old) = &old {
            if CDCEvent::is_sharding_values_changed(data, old, sharding_columns)? {
                Some(CDCEvent::build_sharding_key(index_meta, old, handle)?)
            } else {
                None
            }
        } else {
            None
        };

        Ok(ShardingKeyPair {
            data_sharding_key,
            old_sharding_key,
        })
    }

    pub fn get_sharding_key_pair(&self, index_meta: &IndexMeta) -> Result<ShardingKeyPair> {
        self.create_sharding_key_pair(index_meta)
    }
}

#[derive(Debug, Clone, Deserialize)]
struct TiDBExt {
    #[serde(rename = "commitTs")]
    version: Option<u64>,

    //#[serde(rename = "watermarkTs")]
    // watermark_ts: Option<u64>,
    #[serde(rename = "rowkey")]
    handle: Option<ShardingKey>,
}

type MapVecOpt = Option<Vec<HashMap<String, Option<String>>>>;

fn deserialize_lowercase_map_vec_opt<'de, D>(
    deserializer: D,
) -> std::result::Result<MapVecOpt, D::Error>
where
    D: serde::Deserializer<'de>,
{
    let opt = MapVecOpt::deserialize(deserializer)?;
    Ok(opt.map(|vec_maps| {
        vec_maps
            .into_iter()
            .map(|map| {
                map.into_iter()
                    .map(|(k, v)| (k.to_lowercase(), v))
                    .collect()
            })
            .collect()
    }))
}

// Reference: https://docs.pingcap.com/tidb/stable/ticdc-canal-json/
#[derive(Debug, Clone, Deserialize)]
pub struct CDCEvent {
    pub id: i64,
    pub database: String,
    pub table: String,

    //#[serde(rename = "pkNames")]
    // pk_names: Option<Vec<String>>,
    #[serde(rename = "isDdl")]
    is_ddl: bool,

    // Operation type (INSERT, UPDATE, DELETE, WATERMARK, etc.)
    #[serde(rename = "type")]
    operation_type: String,

    // Entry timestamp (epoch milliseconds)
    // es: i64,

    // Transaction timestamp (epoch milliseconds)
    // ts: i64,

    // SQL statement (if available)
    // sql: String,

    // SQL data types for columns
    //#[serde(rename = "sqlType")]
    // sql_type: Option<HashMap<String, i32>>,

    // MySQL data types for columns
    //#[serde(rename = "mysqlType")]
    // mysql_type: Option<HashMap<String, String>>,

    // Previous values (for UPDATE operations)
    #[serde(default, deserialize_with = "deserialize_lowercase_map_vec_opt")]
    old: MapVecOpt,

    // Changed data rows
    #[serde(default, deserialize_with = "deserialize_lowercase_map_vec_opt")]
    data: MapVecOpt,

    #[serde(rename = "_tidb")]
    tidb_ext: TiDBExt,
}

impl CDCEvent {
    fn from_json(json_str: &str) -> Result<Self> {
        let cdc_log: CDCEvent = serde_json::from_str(json_str)?;
        inc_cdc_event_count(cdc_log.operation_type.as_str());
        Ok(cdc_log)
    }

    fn get_version(&self) -> Result<u64> {
        self.tidb_ext.version.context("Version not found")
    }

    fn get_handle(&self) -> Result<&ShardingKey> {
        self.tidb_ext.handle.as_ref().context("Handle not found")
    }

    fn is_ddl(&self) -> bool {
        self.is_ddl
    }

    fn get_operation_type(&self) -> Result<OperationType> {
        match self.operation_type.as_str() {
            "INSERT" => Ok(OperationType::Insert),
            "UPDATE" => Ok(OperationType::Update),
            "DELETE" => Ok(OperationType::Delete),
            "TIDB_WATERMARK" => Ok(OperationType::Watermark),
            _ => bail!(
                "Invalid or unsupported operation type: '{}'. Expected one of: INSERT, UPDATE, DELETE, TIDB_WATERMARK",
                self.operation_type
            ),
        }
    }

    fn get_datum<'a>(
        data: &'a HashMap<String, Option<String>>,
        col: &ColumnInfo,
    ) -> Result<Datum<'a>> {
        if col.is_text_type() {
            match Self::get_str(data, col)? {
                Some(v) => Ok(Datum::String(v)),
                None => Ok(Datum::Null),
            }
        } else if col.is_unsigned_int_type() {
            match Self::get_u64(data, col)? {
                Some(v) => Ok(Datum::Uint64(v)),
                None => Ok(Datum::Null),
            }
        } else if col.is_signed_int_type() {
            match Self::get_i64(data, col)? {
                Some(v) => Ok(Datum::Int64(v)),
                None => Ok(Datum::Null),
            }
        } else if col.is_float_type() {
            match Self::get_f64(data, col)? {
                Some(v) => Ok(Datum::Float64(v)),
                None => Ok(Datum::Null),
            }
        } else if col.is_date_time_type() {
            match Self::get_mytime(data, col)? {
                Some(v) => Ok(Datum::Time(v)),
                None => Ok(Datum::Null),
            }
        } else if col.is_decimal_type() {
            match Self::get_decimal(data, col)? {
                Some(v) => Ok(Datum::Decimal(
                    v,
                    col.column_length as u8,
                    col.decimal as u8,
                )),
                None => Ok(Datum::Null),
            }
        } else {
            bail!("Unsupported column={:?}", col);
        }
    }

    fn get_data(&self) -> Result<&HashMap<String, Option<String>>> {
        let rows = self.data.as_ref().context("Data not found in CDC event")?;

        if rows.is_empty() {
            bail!("Expected exactly one row in data, found 0 (empty data array)");
        }
        if rows.len() != 1 {
            bail!(
                "Expected exactly one row in data, found {} (CDC events should contain single row changes)",
                rows.len()
            );
        }
        rows.first().context("Data row missing")
    }

    fn get_old(&self) -> Result<Option<&HashMap<String, Option<String>>>> {
        if let Some(rows) = self.old.as_ref() {
            if rows.is_empty() {
                bail!("Expected exactly one row in old data, found 0 (empty old data array)");
            }
            if rows.len() != 1 {
                bail!(
                    "Expected exactly one row in old data, found {} (CDC events should contain single row changes)",
                    rows.len()
                );
            }
            Ok(rows.first())
        } else {
            Ok(None)
        }
    }

    fn get_column_value<'a>(
        data: &'a HashMap<String, Option<String>>,
        col: &ColumnInfo,
    ) -> Result<Option<&'a str>> {
        let col_value = data.get(&col.column_name).ok_or_else(|| {
            anyhow!(
                "Column '{}' not found in data row. Available columns: {:?}",
                col.column_name,
                data.keys().collect::<Vec<_>>()
            )
        })?;

        match col_value {
            Some(s) => Ok(Some(s.as_str())),
            None if col.is_nullable() => Ok(None),
            None => bail!(
                "Column '{}' is not nullable, but received NULL value",
                col.column_name
            ),
        }
    }

    fn parse_column<T: std::str::FromStr>(
        data: &HashMap<String, Option<String>>,
        col: &ColumnInfo,
    ) -> Result<Option<T>>
    where
        T::Err: std::fmt::Display,
    {
        match Self::get_column_value(data, col)? {
            Some(s) => {
                let parsed = s.parse::<T>().map_err(|e| {
                    anyhow!(
                        "Failed to parse column '{}', value '{}' as {}: {}",
                        col.column_name,
                        s,
                        std::any::type_name::<T>(),
                        e
                    )
                })?;
                Ok(Some(parsed))
            }
            None => Ok(None),
        }
    }

    fn get_str<'a>(
        data: &'a HashMap<String, Option<String>>,
        col: &ColumnInfo,
    ) -> Result<Option<&'a str>> {
        if !col.is_text_type() {
            bail!("Column {:?} is not text type", col);
        }
        Self::get_column_value(data, col)
    }

    fn get_u64(data: &HashMap<String, Option<String>>, col: &ColumnInfo) -> Result<Option<u64>> {
        if !col.is_unsigned_int_type() {
            bail!("Column {:?} is not unsigned int type", col);
        }
        Self::parse_column::<u64>(data, col)
    }

    fn get_i64(data: &HashMap<String, Option<String>>, col: &ColumnInfo) -> Result<Option<i64>> {
        if !col.is_signed_int_type() {
            bail!("Column {:?} is not int type", col);
        }
        Self::parse_column::<i64>(data, col)
    }

    fn get_f64(data: &HashMap<String, Option<String>>, col: &ColumnInfo) -> Result<Option<f64>> {
        if !col.is_float_type() {
            bail!("Column {:?} is not float type", col);
        }
        Self::parse_column::<f64>(data, col)
    }

    fn get_mytime(
        data: &HashMap<String, Option<String>>,
        col: &ColumnInfo,
    ) -> Result<Option<MyTime>> {
        if !col.is_date_time_type() {
            bail!("Column {:?} is not date time type", col);
        }
        match Self::get_column_value(data, col)? {
            Some(datetime_str) => Ok(Some(
                // CDC logs are in local timezone
                MyTime::new(datetime_str, col.r#type, true)?,
            )),
            None => Ok(None),
        }
    }

    fn get_decimal(
        data: &HashMap<String, Option<String>>,
        col: &ColumnInfo,
    ) -> Result<Option<Vec<u8>>> {
        if !col.is_decimal_type() {
            bail!("Column {:?} is not decimal type", col);
        }
        match Self::get_column_value(data, col)? {
            Some(decimal_str) => {
                let mut buf = vec![];
                let precision = col.column_length as u8;
                let scale = col.decimal as u8;
                encode_decimal_val(&mut buf, decimal_str, precision, scale)?;
                Ok(Some(buf))
            }
            None => Ok(None),
        }
    }

    fn build_sharding_key(
        index_meta: &IndexMeta,
        data_row: &HashMap<String, Option<String>>,
        handle: &ShardingKey,
    ) -> Result<ShardingKey> {
        let keyspace_id = index_meta.index_key().keyspace_id;
        let table_id = index_meta.index_key().table_id;
        let index_id = index_meta.index_key().index_id;
        let sharding_columns = index_meta.get_sharding_columns();
        ensure!(
            !sharding_columns.is_empty(),
            "sharding columns are empty when building sharding key"
        );
        let mut sharding_key_parts = Vec::with_capacity(sharding_columns.len());
        for col in sharding_columns {
            let datum = CDCEvent::get_datum(data_row, col)?;
            sharding_key_parts.push(datum);
        }

        // Build the sharding key
        let mut key = Vec::new();
        // Prepend keyspace prefix if it's not default keyspace
        if keyspace_id != NULLSPACEID {
            key.extend_from_slice(get_txn_keyspace_prefix(keyspace_id).as_slice());
        }
        encode_index_key(&mut key, table_id, index_id, &sharding_key_parts)?;

        ensure!(
            handle.raw().len() >= RAW_KEY_PREFIX_LEN,
            "invalid handle length: expected at least {} bytes for prefix, got {}",
            RAW_KEY_PREFIX_LEN,
            handle.raw().len()
        );
        let h = &handle.raw()[RAW_KEY_PREFIX_LEN..]; // Remove t{table_id}_r prefix

        // Refer to TiDB tablecodec encoding logic:
        // https://github.com/pingcap/tidb/blob/02fe040a0f184f2ea8f450b989678da9ac21a0de/pkg/tablecodec/tablecodec.go#L1228-L1236
        if index_meta.is_int_handle() {
            ensure!(
                h.len() == std::mem::size_of::<i64>(),
                "handle len must be 8 bytes, but got {}",
                h.len()
            );
            key.push(INT_HANDLE_FLAG);
            key.extend_from_slice(h);
        } else {
            key.extend_from_slice(h);
        }
        Ok(ShardingKey::from_raw(key))
    }

    fn build_default_sharding_key(keyspace_id: u32, handle: &ShardingKey) -> ShardingKey {
        let mut key = Vec::new();
        if keyspace_id != NULLSPACEID {
            key.extend_from_slice(get_txn_keyspace_prefix(keyspace_id).as_slice());
        }
        key.extend_from_slice(handle.raw());
        ShardingKey::from_raw(key)
    }

    fn is_sharding_values_changed(
        data: &HashMap<String, Option<String>>,
        old: &HashMap<String, Option<String>>,
        sharding_columns: &[ColumnInfo],
    ) -> Result<bool> {
        for col in sharding_columns {
            let data_value = CDCEvent::get_column_value(data, col)?;
            let old_value = CDCEvent::get_column_value(old, col)?;
            if data_value != old_value {
                return Ok(true);
            }
        }
        Ok(false)
    }
}

#[cfg(test)]
mod tests {
    use std::collections::HashMap;

    use tici_common::index::IndexKey;
    use tici_grpc::tidb_def::{BINARY_FLAG, COLTYPE_BLOB, COLTYPE_STRING, NOT_NULL_FLAG};

    use super::*;
    use crate::test_utils::index_meta::{
        mock_custom_hdfs_logs_index_meta, mock_custom_hdfs_logs_index_meta_int_handle,
    };

    // CREATE TABLE ticdc_binary_test (
    // id INT PRIMARY KEY,
    // name VARCHAR(20),
    // bin_data BINARY(16),
    // blob_data BLOB,
    // long_blob_data LONGBLOB
    // );
    //
    // -- 1. Mixed test: NULL bytes, plain letters, escape chars, and HTML-sensitive
    // chars -- Corresponding bytes: 0x00, 'A', 0x09 (TAB), '&', '<', '>', '"',
    // '\\' INSERT INTO ticdc_binary_test (id, name, bin_data, blob_data)
    // VALUES (1, 'special_chars',
    // X'004109263C3E225C',
    // X'004109263C3E225C');
    //
    // -- 2. High-byte test: non-ASCII range (128-255)
    // -- Corresponding bytes: 0x80, 0xAB, 0xF1, 0xFF
    // INSERT INTO ticdc_binary_test (id, name, bin_data, blob_data)
    // VALUES (2, 'high_bit_bytes',
    // X'80ABF1FF',
    // X'80ABF1FF');
    //
    // -- 3. Boundary length test: padding (BINARY pads with 0x00 when shorter)
    // INSERT INTO ticdc_binary_test (id, name, bin_data)
    // VALUES (3, 'padding_test', X'414243'); -- 'ABC' with remaining bytes padded
    // as \u0000

    const CDC_JSON_SPECIAL_CHARS: &str = r#"
    {
        "id": 0,
        "database": "test",
        "table": "ticdc_binary_test",
        "pkNames": ["id"],
        "isDdl": false,
        "type": "INSERT",
        "es": 1770536785133,
        "ts": 1770536790152,
        "sql": "",
        "sqlType": {
            "id": 4,
            "name": 12,
            "bin_data": 2004,
            "blob_data": 2004,
            "long_blob_data": 2004
        },
        "mysqlType": {
            "bin_data": "binary",
            "blob_data": "blob",
            "long_blob_data": "longblob",
            "id": "int",
            "name": "varchar"
        },
        "old": null,
        "data": [{
            "id": "1",
            "name": "special_chars",
            "bin_data": "\u0000A\t\u0026\u003c\u003e\"\\\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000",
            "blob_data": "\u0000A\t\u0026\u003c\u003e\"\\",
            "long_blob_data": null
        }],
        "_tidb": {
            "commitTs": 464135595001905265,
            "rowkey": "dIAAAAAAAAAZX3KAAAAAAAAAAQ=="
        }
    }
    "#;

    const CDC_JSON_HIGH_BIT_BYTES: &str = r#"
    {
        "id": 0,
        "database": "test",
        "table": "ticdc_binary_test",
        "pkNames": ["id"],
        "isDdl": false,
        "type": "INSERT",
        "es": 1770536814582,
        "ts": 1770536818175,
        "sql": "",
        "sqlType": {
            "id": 4,
            "name": 12,
            "bin_data": 2004,
            "blob_data": 2004,
            "long_blob_data": 2004
        },
        "mysqlType": {
            "id": "int",
            "name": "varchar",
            "bin_data": "binary",
            "blob_data": "blob",
            "long_blob_data": "longblob"
        },
        "old": null,
        "data": [{
            "id": "2",
            "name": "high_bit_bytes",
            "bin_data": "^?«ñÿ\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000",
            "blob_data": "^?«ñÿ",
            "long_blob_data": null
        }],
        "_tidb": {
            "commitTs": 464135602721783816,
            "rowkey": "dIAAAAAAAAAZX3KAAAAAAAAAAg=="
        }
    }
    "#;

    const CDC_JSON_PADDING: &str = r#"
    {
        "id": 0,
        "database": "test",
        "table": "ticdc_binary_test",
        "pkNames": ["id"],
        "isDdl": false,
        "type": "INSERT",
        "es": 1770536821382,
        "ts": 1770536826184,
        "sql": "",
        "sqlType": {
            "id": 4,
            "name": 12,
            "bin_data": 2004,
            "blob_data": 2004,
            "long_blob_data": 2004
        },
        "mysqlType": {
            "id": "int",
            "name": "varchar",
            "bin_data": "binary",
            "blob_data": "blob",
            "long_blob_data": "longblob"
        },
        "old": null,
        "data": [{
            "id": "3",
            "name": "padding_test",
            "bin_data": "ABC\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000",
            "blob_data": null,
            "long_blob_data": null
        }],
        "_tidb": {
            "commitTs": 464135604504363014,
            "rowkey": "dIAAAAAAAAAZX3KAAAAAAAAAAw=="
        }
    }
    "#;

    fn mock_binary_column(column_name: &str, column_type: i32, is_nullable: bool) -> ColumnInfo {
        let mut flag = BINARY_FLAG;
        if !is_nullable {
            flag |= NOT_NULL_FLAG;
        }

        ColumnInfo {
            column_name: column_name.to_string(),
            r#type: column_type,
            flag,
            column_length: 16,
            decimal: 0,
            is_primary_key: false,
            ..Default::default()
        }
    }

    fn mock_cdc_log_from_json(event_json: &str) -> Result<CDCLog> {
        let event = CDCEvent::from_json(event_json)?;
        Ok(CDCLog { event })
    }

    #[test]
    fn test_ddl_event() {
        let json = r#"
        {
            "id": 0,
            "database": "test",
            "table": "",
            "pkNames": null,
            "isDdl": true,
            "type": "QUERY",
            "es": 1639633094670,
            "ts": 1639633095489,
            "sql": "drop database if exists test",
            "sqlType": null,
            "mysqlType": null,
            "data": null,
            "old": null,
            "_tidb": {
                "commitTs": 429918007904436226
            }
        }"#;
        let log = CDCEvent::from_json(json).unwrap();
        assert_eq!(log.id, 0);
        assert_eq!(log.database, "test");
        assert_eq!(log.table, "");
        assert_eq!(log.operation_type, "QUERY");
        assert!(log.data.is_none());
        assert_eq!(log.tidb_ext.version.unwrap(), 429918007904436226);
    }

    #[test]
    fn test_dml_event() {
        let json = r#"
        {
            "id": 0,
            "database": "test",
            "table": "tp_int",
            "pkNames": [
                "id"
            ],
            "isDdl": false,
            "type": "INSERT",
            "es": 1639633141221,
            "ts": 1639633142960,
            "sql": "",
            "sqlType": {
                "c_bigint": -5,
                "c_int": 4,
                "c_mediumint": 4,
                "c_smallint": 5,
                "c_tinyint": -6,
                "id": 4
            },
            "mysqlType": {
                "c_bigint": "bigint",
                "c_int": "int",
                "c_mediumint": "mediumint",
                "c_smallint": "smallint",
                "c_tinyint": "tinyint",
                "id": "int"
            },
            "data": [
                {
                    "c_bigint": "9223372036854775807",
                    "c_int": "2147483647",
                    "c_mediumint": "8388607",
                    "c_smallint": "32767",
                    "c_tinyint": "127",
                    "id": "2"
                }
            ],
            "old": null,
            "_tidb": {
                "commitTs": 429918007904436226,
                "rowkey": "dIAAAAAAAAB2X3KAAAAAAAAD7w=="
            }
        }"#;
        let log = CDCEvent::from_json(json).unwrap();
        assert_eq!(log.id, 0);
        assert_eq!(log.database, "test");
        assert_eq!(log.table, "tp_int");
        assert_eq!(log.operation_type, "INSERT");
        assert!(log.data.is_some());
        assert_eq!(log.tidb_ext.version.unwrap(), 429918007904436226);

        let handle_nonclustered = log.tidb_ext.handle.unwrap();
        assert_eq!(handle_nonclustered.base64(), "dIAAAAAAAAB2X3KAAAAAAAAD7w==");
        assert_eq!(
            handle_nonclustered.hex(),
            "7480000000000000765f7280000000000003ef"
        );
        assert_eq!(
            format!("{}", handle_nonclustered),
            "7480000000000000765f7280000000000003ef"
        );
        assert_eq!(
            format!("{:?}", handle_nonclustered),
            "7480000000000000765f7280000000000003ef"
        );
        // mysql> select tidb_decode_key('7480000000000000765f7280000000000003ef');
        // +-----------------------------------------------------------+
        // | tidb_decode_key('7480000000000000765f7280000000000003ef') |
        // +-----------------------------------------------------------+
        // | {"_tidb_rowid":1007,"table_id":"118"}                     |
        // +-----------------------------------------------------------+

        let handle_clustered =
            ShardingKey::from_base64("dIAAAAAAAABuX3IDgAAAAAAAAAEDgAAAAAAABFA=").unwrap();
        assert_eq!(
            handle_clustered.hex(),
            "74800000000000006e5f72038000000000000001038000000000000450"
        );
        assert_eq!(
            format!("{}", handle_clustered),
            "74800000000000006e5f72038000000000000001038000000000000450"
        );
        assert_eq!(
            format!("{:?}", handle_clustered),
            "74800000000000006e5f72038000000000000001038000000000000450"
        );
        // mysql> select
        // tidb_decode_key('
        // 74800000000000006e5f72038000000000000001038000000000000450');
        // +-------------------------------------------------------------------------------+
        // | tidb_decode_key('74800000000000006e5f72038000000000000001038000000000000450') |
        // +-------------------------------------------------------------------------------+
        // | {"handle":{"id":"1104","tenant_id":"1"},"table_id":110}                       |
        // +-------------------------------------------------------------------------------+
    }

    #[test]
    fn test_watermark_event() {
        let json = r#"
        {
            "id": 0,
            "database": "",
            "table": "",
            "pkNames": null,
            "isDdl": false,
            "type": "TIDB_WATERMARK",
            "es": 1640007049196,
            "ts": 1640007050284,
            "sql": "",
            "sqlType": null,
            "mysqlType": null,
            "data": null,
            "old": null,
            "_tidb": {
                "watermarkTs": 429918007904436226
            }
        }"#;
        let log = CDCEvent::from_json(json).unwrap();
        assert_eq!(log.id, 0);
        assert_eq!(log.database, "");
        assert_eq!(log.table, "");
        assert_eq!(log.operation_type, "TIDB_WATERMARK");
        assert!(log.data.is_none());
    }

    #[test]
    fn test_operation_type_is_dml() -> Result<()> {
        assert!(OperationType::Insert.is_dml());
        assert!(OperationType::Update.is_dml());
        assert!(OperationType::Delete.is_dml());
        assert!(!OperationType::Watermark.is_dml());
        Ok(())
    }

    #[test]
    fn test_get_operation_type() -> Result<()> {
        let json_insert = r#"{"id":1,"database":"test","table":"t","isDdl":false,"type":"INSERT","data":[{"id":"1"}],"_tidb":{"commitTs":123,"rowkey":"dIAAAAAAAAB2X3KAAAAAAAAD7w=="}}"#;
        let event = CDCEvent::from_json(json_insert)?;
        assert_eq!(event.get_operation_type()?, OperationType::Insert);

        let json_update = r#"{"id":1,"database":"test","table":"t","isDdl":false,"type":"UPDATE","data":[{"id":"1"}],"old":[{"id":"2"}],"_tidb":{"commitTs":123,"rowkey":"dIAAAAAAAAB2X3KAAAAAAAAD7w=="}}"#;
        let event = CDCEvent::from_json(json_update)?;
        assert_eq!(event.get_operation_type()?, OperationType::Update);

        let json_delete = r#"{"id":1,"database":"test","table":"t","isDdl":false,"type":"DELETE","data":[{"id":"1"}],"_tidb":{"commitTs":123,"rowkey":"dIAAAAAAAAB2X3KAAAAAAAAD7w=="}}"#;
        let event = CDCEvent::from_json(json_delete)?;
        assert_eq!(event.get_operation_type()?, OperationType::Delete);

        let json_watermark = r#"{"id":1,"database":"","table":"","isDdl":false,"type":"TIDB_WATERMARK","_tidb":{"watermarkTs":123}}"#;
        let event = CDCEvent::from_json(json_watermark)?;
        assert_eq!(event.get_operation_type()?, OperationType::Watermark);
        Ok(())
    }

    #[test]
    fn test_invalid_operation_type() -> Result<()> {
        let json = r#"{"id":1,"database":"test","table":"t","isDdl":false,"type":"INVALID","_tidb":{"commitTs":123}}"#;
        let event = CDCEvent::from_json(json)?;
        assert!(event.get_operation_type().is_err());
        Ok(())
    }

    #[test]
    fn test_get_data_success() -> Result<()> {
        let json = r#"{"id":1,"database":"test","table":"t","isDdl":false,"type":"INSERT","data":[{"id":"1","name":"test"}],"_tidb":{"commitTs":123,"rowkey":"dIAAAAAAAAB2X3KAAAAAAAAD7w=="}}"#;
        let event = CDCEvent::from_json(json)?;
        let data = event.get_data()?;
        assert_eq!(data.get("id").unwrap().as_ref().unwrap(), "1");
        assert_eq!(data.get("name").unwrap().as_ref().unwrap(), "test");
        Ok(())
    }

    /// Asserts that `old` and `data` HashMap keys are lowercased during
    /// deserialization.
    #[test]
    fn test_cdc_event_deserialize_lowercase_keys() -> Result<()> {
        // data with uppercase and mixed-case keys
        let json_insert = r#"{"id":1,"database":"test","table":"t","isDdl":false,"type":"INSERT","data":[{"ID":"1","Name":"Alice","Col_ABC":"value"}],"_tidb":{"commitTs":123,"rowkey":"dIAAAAAAAAB2X3KAAAAAAAAD7w=="}}"#;
        let event = CDCEvent::from_json(json_insert)?;
        let data = event.get_data()?;
        assert_eq!(
            data.get("id")
                .context("id key")?
                .as_ref()
                .context("id value")?,
            "1"
        );
        assert_eq!(
            data.get("name")
                .context("name key")?
                .as_ref()
                .context("name value")?,
            "Alice"
        );
        assert_eq!(
            data.get("col_abc")
                .context("col_abc key")?
                .as_ref()
                .context("col_abc value")?,
            "value"
        );
        assert!(data.get("ID").is_none());
        assert!(data.get("Name").is_none());

        // UPDATE with uppercase keys in both old and data
        let json_update = r#"{"id":2,"database":"test","table":"t","isDdl":false,"type":"UPDATE","data":[{"ID":"2","Name":"Bob"}],"old":[{"ID":"2","Name":"Alice"}],"_tidb":{"commitTs":124,"rowkey":"dIAAAAAAAAB2X3KAAAAAAAAD7w=="}}"#;
        let event = CDCEvent::from_json(json_update)?;
        let data = event.get_data()?;
        let old = event.get_old()?.context("old should be some")?;
        assert_eq!(
            data.get("id")
                .context("id key")?
                .as_ref()
                .context("id value")?,
            "2"
        );
        assert_eq!(
            data.get("name")
                .context("name key")?
                .as_ref()
                .context("name value")?,
            "Bob"
        );
        assert_eq!(
            old.get("id")
                .context("id key")?
                .as_ref()
                .context("id value")?,
            "2"
        );
        assert_eq!(
            old.get("name")
                .context("name key")?
                .as_ref()
                .context("name value")?,
            "Alice"
        );
        assert!(old.get("ID").is_none());
        Ok(())
    }

    #[test]
    fn test_get_data_missing() -> Result<()> {
        let json = r#"{"id":1,"database":"test","table":"t","isDdl":false,"type":"INSERT","_tidb":{"commitTs":123}}"#;
        let event = CDCEvent::from_json(json)?;
        assert!(event.get_data().is_err());
        Ok(())
    }

    #[test]
    fn test_get_data_multiple_rows() -> Result<()> {
        let json = r#"{"id":1,"database":"test","table":"t","isDdl":false,"type":"INSERT","data":[{"id":"1"},{"id":"2"}],"_tidb":{"commitTs":123}}"#;
        let event = CDCEvent::from_json(json)?;
        let result = event.get_data();
        assert!(result.is_err());
        assert!(
            result
                .unwrap_err()
                .to_string()
                .contains("Expected exactly one row")
        );
        Ok(())
    }

    #[test]
    fn test_get_old_success() -> Result<()> {
        let json = r#"{"id":1,"database":"test","table":"t","isDdl":false,"type":"UPDATE","data":[{"id":"1"}],"old":[{"id":"2"}],"_tidb":{"commitTs":123,"rowkey":"dIAAAAAAAAB2X3KAAAAAAAAD7w=="}}"#;
        let event = CDCEvent::from_json(json)?;
        let old = event.get_old()?;
        assert!(old.is_some());
        assert_eq!(old.unwrap().get("id").unwrap().as_ref().unwrap(), "2");
        Ok(())
    }

    #[test]
    fn test_get_old_none() -> Result<()> {
        let json = r#"{"id":1,"database":"test","table":"t","isDdl":false,"type":"INSERT","data":[{"id":"1"}],"_tidb":{"commitTs":123,"rowkey":"dIAAAAAAAAB2X3KAAAAAAAAD7w=="}}"#;
        let event = CDCEvent::from_json(json)?;
        assert!(event.get_old()?.is_none());
        Ok(())
    }

    #[test]
    fn test_get_old_multiple_rows() -> Result<()> {
        let json = r#"{"id":1,"database":"test","table":"t","isDdl":false,"type":"UPDATE","data":[{"id":"1"}],"old":[{"id":"2"},{"id":"3"}],"_tidb":{"commitTs":123}}"#;
        let event = CDCEvent::from_json(json)?;
        let result = event.get_old();
        assert!(result.is_err());
        assert!(
            result
                .unwrap_err()
                .to_string()
                .contains("Expected exactly one row")
        );
        Ok(())
    }

    #[test]
    fn test_get_version() -> Result<()> {
        let json = r#"{"id":1,"database":"test","table":"t","isDdl":false,"type":"INSERT","data":[{"id":"1"}],"_tidb":{"commitTs":429918007904436226,"rowkey":"dIAAAAAAAAB2X3KAAAAAAAAD7w=="}}"#;
        let event = CDCEvent::from_json(json)?;
        assert_eq!(event.get_version()?, 429918007904436226);
        Ok(())
    }

    #[test]
    fn test_get_version_missing() -> Result<()> {
        let json = r#"{"id":1,"database":"test","table":"t","isDdl":false,"type":"INSERT","data":[{"id":"1"}],"_tidb":{"rowkey":"dIAAAAAAAAB2X3KAAAAAAAAD7w=="}}"#;
        let event = CDCEvent::from_json(json)?;
        assert!(event.get_version().is_err());
        Ok(())
    }

    #[test]
    fn test_get_handle() -> Result<()> {
        let json = r#"{"id":1,"database":"test","table":"t","isDdl":false,"type":"INSERT","data":[{"id":"1"}],"_tidb":{"commitTs":123,"rowkey":"dIAAAAAAAAB2X3KAAAAAAAAD7w=="}}"#;
        let event = CDCEvent::from_json(json)?;
        let handle = event.get_handle()?;
        assert_eq!(handle.base64(), "dIAAAAAAAAB2X3KAAAAAAAAD7w==");
        Ok(())
    }

    #[test]
    fn test_get_handle_missing() -> Result<()> {
        let json = r#"{"id":1,"database":"test","table":"t","isDdl":false,"type":"INSERT","data":[{"id":"1"}],"_tidb":{"commitTs":123}}"#;
        let event = CDCEvent::from_json(json)?;
        assert!(event.get_handle().is_err());
        Ok(())
    }

    #[test]
    fn test_is_ddl() -> Result<()> {
        let json_ddl = r#"{"id":1,"database":"test","table":"","isDdl":true,"type":"QUERY","_tidb":{"commitTs":123}}"#;
        let event = CDCEvent::from_json(json_ddl)?;
        assert!(event.is_ddl());

        let json_dml = r#"{"id":1,"database":"test","table":"t","isDdl":false,"type":"INSERT","data":[{"id":"1"}],"_tidb":{"commitTs":123,"rowkey":"dIAAAAAAAAB2X3KAAAAAAAAD7w=="}}"#;
        let event = CDCEvent::from_json(json_dml)?;
        assert!(!event.is_ddl());
        Ok(())
    }

    #[test]
    fn test_json_deserialization_error() -> Result<()> {
        let invalid_json = r#"{"invalid": json}"#;
        assert!(CDCEvent::from_json(invalid_json).is_err());
        Ok(())
    }

    #[test]
    fn test_parse_column_error() -> Result<()> {
        let mut data = HashMap::new();
        data.insert("col1".to_string(), Some("not_a_number".to_string()));

        let col = ColumnInfo {
            column_name: "col1".to_string(),
            r#type: 3, // INT type
            flag: NOT_NULL_FLAG,
            column_length: 11,
            decimal: 0,
            is_primary_key: false,
            ..Default::default()
        };

        let result = CDCEvent::parse_column::<i64>(&data, &col);
        assert!(result.is_err());
        let err_msg = result.unwrap_err().to_string();
        assert!(err_msg.contains("Failed to parse"));
        assert!(err_msg.contains("col1"));
        assert!(err_msg.contains("not_a_number"));
        Ok(())
    }

    #[test]
    fn test_get_binary_decodes_special_chars_and_padding() -> Result<()> {
        let expected_blob_data = vec![0x00, b'A', b'\t', b'&', b'<', b'>', b'"', b'\\'];
        let mut expected_bin_data = expected_blob_data.clone();
        expected_bin_data.resize(16, 0);

        let log = mock_cdc_log_from_json(CDC_JSON_SPECIAL_CHARS)?;

        let bin_col = mock_binary_column("bin_data", COLTYPE_STRING, true);
        let blob_col = mock_binary_column("blob_data", COLTYPE_BLOB, true);

        assert_eq!(log.get_binary(&bin_col)?, Some(expected_bin_data));
        assert_eq!(log.get_binary(&blob_col)?, Some(expected_blob_data));
        Ok(())
    }

    #[test]
    fn test_get_binary_decodes_high_bit_bytes() -> Result<()> {
        let expected_blob_data = vec![b'^', b'?', 0xAB, 0xF1, 0xFF];
        let mut expected_bin_data = expected_blob_data.clone();
        expected_bin_data.resize(17, 0);

        let log = mock_cdc_log_from_json(CDC_JSON_HIGH_BIT_BYTES)?;

        let bin_col = mock_binary_column("bin_data", COLTYPE_STRING, true);
        let blob_col = mock_binary_column("blob_data", COLTYPE_BLOB, true);

        assert_eq!(log.get_binary(&bin_col)?, Some(expected_bin_data));
        assert_eq!(log.get_binary(&blob_col)?, Some(expected_blob_data));
        Ok(())
    }

    #[test]
    fn test_get_binary_binary_padding_and_nullable_blob() -> Result<()> {
        let mut expected_bin_data = b"ABC".to_vec();
        expected_bin_data.resize(16, 0);

        let log = mock_cdc_log_from_json(CDC_JSON_PADDING)?;

        let bin_col = mock_binary_column("bin_data", COLTYPE_STRING, true);
        let blob_col = mock_binary_column("blob_data", COLTYPE_BLOB, true);

        assert_eq!(log.get_binary(&bin_col)?, Some(expected_bin_data));
        assert_eq!(log.get_binary(&blob_col)?, None);
        Ok(())
    }

    #[test]
    fn test_is_sharding_values_changed() -> Result<()> {
        let mut data = HashMap::new();
        data.insert("id".to_string(), Some("1".to_string()));
        data.insert("name".to_string(), Some("alice".to_string()));

        let mut old = HashMap::new();
        old.insert("id".to_string(), Some("1".to_string()));
        old.insert("name".to_string(), Some("bob".to_string()));

        let sharding_col = ColumnInfo {
            column_name: "name".to_string(),
            r#type: 15, // VARCHAR
            flag: NOT_NULL_FLAG,
            column_length: 100,
            decimal: 0,
            is_primary_key: false,
            ..Default::default()
        };

        // Test when sharding column value changed
        let changed = CDCEvent::is_sharding_values_changed(&data, &old, &[sharding_col.clone()])?;
        assert!(changed);

        // Test when sharding column value not changed
        let id_col = ColumnInfo {
            column_name: "id".to_string(),
            r#type: 3,
            flag: NOT_NULL_FLAG,
            column_length: 11,
            decimal: 0,
            is_primary_key: true,
            ..Default::default()
        };

        let not_changed = CDCEvent::is_sharding_values_changed(&data, &old, &[id_col])?;
        assert!(!not_changed);
        Ok(())
    }

    #[test]
    fn test_get_column_value_null_non_nullable() -> Result<()> {
        let mut data = HashMap::new();
        data.insert("col1".to_string(), None);

        let col = ColumnInfo {
            column_name: "col1".to_string(),
            r#type: 3,
            flag: NOT_NULL_FLAG,
            column_length: 11,
            decimal: 0,
            is_primary_key: false,
            ..Default::default()
        };

        let result = CDCEvent::get_column_value(&data, &col);
        assert!(result.is_err());
        assert!(result.unwrap_err().to_string().contains("is not nullable"));
        Ok(())
    }

    #[test]
    fn test_get_column_value_null_nullable() -> Result<()> {
        let mut data = HashMap::new();
        data.insert("col1".to_string(), None);

        let col = ColumnInfo {
            column_name: "col1".to_string(),
            r#type: 3,
            flag: 0,
            column_length: 11,
            decimal: 0,
            is_primary_key: false,
            ..Default::default()
        };

        let result = CDCEvent::get_column_value(&data, &col)?;
        assert!(result.is_none());
        Ok(())
    }

    #[test]
    fn test_get_column_value_missing() -> Result<()> {
        let data = HashMap::new();

        let col = ColumnInfo {
            column_name: "missing_col".to_string(),
            r#type: 3,
            flag: 0,
            column_length: 11,
            decimal: 0,
            is_primary_key: false,
            ..Default::default()
        };

        let result = CDCEvent::get_column_value(&data, &col);
        assert!(result.is_err());
        assert!(result.unwrap_err().to_string().contains("not found"));
        Ok(())
    }

    #[test]
    fn test_build_sharding_key_is_int_handle() -> Result<()> {
        let json = r#"{"id":0,"database":"test","table":"t2","pkNames":["id"],"isDdl":false,"type":"INSERT","es":1767508360474,"ts":1767508365178,"sql":"","sqlType":{"id":-5,"timestamp":-5,"severity_text":12,"body":2005,"tenant_id":4},"mysqlType":{"id":"bigint","timestamp":"bigint","severity_text":"varchar","body":"text","tenant_id":"int"},"old":null,"data":[{"id":"804","timestamp":"1440670493","severity_text":"INFO","body":"src: /10.10.34.30:48744, dest: /10.10.34.13:50010, bytes: 147355, op: HDFS_WRITE, cliID: DFSClient_NONMAPREDUCE_-1482587964_1, offset: 0, srvID: 95a87747-ff24-41e2-8453-0127a1675627, blockid: BP-108841162-10.10.34.11-1440074360971:blk_1073836179_95355, duration: 3403794","tenant_id":"52"}],"_tidb":{"commitTs":463341711648096283,"rowkey":"dIAAAAAAAAASX3KAAAAAAAADJA=="}}"#;
        let event = CDCEvent::from_json(json)?;
        let data = event.get_data()?;
        let handle = event.get_handle()?;
        let index_key = IndexKey {
            keyspace_id: 1,
            table_id: 18,
            index_id: 1,
        };
        let index_meta = mock_custom_hdfs_logs_index_meta_int_handle(index_key, &[4])?;
        let sharding_key = CDCEvent::build_sharding_key(&index_meta, data, handle)?;

        // Verify the sharding key structure
        let key_bytes = sharding_key.raw();
        assert!(key_bytes.len() >= 9, "Key should be at least 9 bytes");
        let last_9_bytes = &key_bytes[key_bytes.len() - 9..];
        assert_eq!(
            last_9_bytes[0], INT_HANDLE_FLAG,
            "Should have INT_HANDLE_FLAG"
        );

        Ok(())
    }

    #[test]
    fn test_build_sharding_key_not_int_handle() -> Result<()> {
        let json = r#"{"id":0,"database":"test","table":"t1","pkNames":["id","tenant_id"],"isDdl":false,"type":"INSERT","es":1767163127105,"ts":1767163130233,"sql":"","sqlType":{"id":-5,"timestamp":-5,"severity_text":12,"body":2005,"tenant_id":4},"mysqlType":{"id":"bigint","timestamp":"bigint","severity_text":"varchar","body":"text","tenant_id":"int"},"old":null,"data":[{"id":"1780","timestamp":"1440670441","severity_text":"INFO","body":"src: /10.10.34.12:51966, dest: /10.10.34.42:50010, bytes: 218525, op: HDFS_WRITE, cliID: DFSClient_NONMAPREDUCE_-1482587964_1, offset: 0, srvID: ac6cb715-a2bc-4644-aaa4-10fcbd1c390e, blockid: BP-108841162-10.10.34.11-1440074360971:blk_1073833144_92320, duration: 4531499","tenant_id":"10"}],"_tidb":{"commitTs":463251210791813123,"rowkey":"dIAAAAAAAAAHX3IDgAAAAAAAAAoDgAAAAAAABvQ="}}"#;
        let event = CDCEvent::from_json(json)?;
        let data = event.get_data()?;
        let handle = event.get_handle()?;
        let index_key = IndexKey {
            keyspace_id: 1,
            table_id: 7,
            index_id: 1,
        };
        let index_meta = mock_custom_hdfs_logs_index_meta(index_key, &[4])?;
        let sharding_key = CDCEvent::build_sharding_key(&index_meta, data, handle)?;

        // Verify the sharding key structure
        let key_bytes = sharding_key.raw();
        let handle_bytes = handle.raw();
        // The sharding key should contain the handle portion (after HANDLE_PREFIX_LEN)
        let handle_suffix = &handle_bytes[RAW_KEY_PREFIX_LEN..];
        assert!(
            key_bytes.ends_with(handle_suffix),
            "Key should end with handle suffix for not int handle"
        );
        Ok(())
    }

    #[test]
    fn test_build_sharding_key_with_multiple_sharding_columns() -> Result<()> {
        let json = r#"{"id":0,"database":"test","table":"t1","pkNames":["id","tenant_id"],"isDdl":false,"type":"INSERT","es":1767163127105,"ts":1767163130233,"sql":"","sqlType":{"id":-5,"timestamp":-5,"severity_text":12,"body":2005,"tenant_id":4},"mysqlType":{"id":"bigint","timestamp":"bigint","severity_text":"varchar","body":"text","tenant_id":"int"},"old":null,"data":[{"id":"1780","timestamp":"1440670441","severity_text":"INFO","body":"test body","tenant_id":"10"}],"_tidb":{"commitTs":463251210791813123,"rowkey":"dIAAAAAAAAAHX3IDgAAAAAAAAAoDgAAAAAAABvQ="}}"#;
        let event = CDCEvent::from_json(json)?;
        let data = event.get_data()?;
        let handle = event.get_handle()?;
        let index_key = IndexKey {
            keyspace_id: 1,
            table_id: 7,
            index_id: 1,
        };
        let index_meta = mock_custom_hdfs_logs_index_meta(index_key, &[4, 1])?;
        let sharding_key = CDCEvent::build_sharding_key(&index_meta, data, handle)?;

        // Verify key was built successfully
        assert!(
            !sharding_key.raw().is_empty(),
            "Sharding key should not be empty"
        );

        Ok(())
    }
}