use std::sync::Arc;

use anyhow::{Context, Result, anyhow, bail, ensure};
use log::info;
use storage_backend::storage::BackendStorage;
use tici_shard::utils::{schema::IndexMeta, tantivy_doc::RawDoc};
use tidb_query_datatype::expr::EvalContext;
use tokio::io::{AsyncRead, AsyncReadExt};

use crate::decoder::{Decoder, DecoderOptions};

// SELECT COLLATION_NAME, ID FROM information_schema.COLLATIONS
pub fn collation_name_to_id(name: &str) -> Result<i32> {
    match name.to_ascii_lowercase().as_str() {
        "gbk_chinese_ci" => Ok(28),
        "utf8_general_ci" => Ok(33),
        "utf8mb4_general_ci" => Ok(45),
        "utf8mb4_bin" => Ok(46),
        "latin1_bin" => Ok(47),
        "binary" => Ok(63),
        "ascii_bin" => Ok(65),
        "utf8_bin" => Ok(83),
        "gbk_bin" => Ok(87),
        "utf8_unicode_ci" => Ok(192),
        "utf8mb4_unicode_ci" => Ok(224),
        "gb18030_chinese_ci" => Ok(248),
        "gb18030_bin" => Ok(249),
        "utf8mb4_0900_ai_ci" => Ok(255),
        "utf8mb4_0900_bin" => Ok(309),
        _ => Err(anyhow!("unknown collation name: {}", name)),
    }
}

pub(crate) fn grpc_column_info_to_tipb(
    col_info: &tici_grpc::ColumnInfo,
) -> Result<tipb::ColumnInfo> {
    let mut new_col_info = tipb::ColumnInfo::default();
    new_col_info.set_column_id(col_info.column_id);
    new_col_info.set_tp(col_info.r#type);
    new_col_info.set_collation(collation_name_to_id(col_info.collation.as_str())?);
    new_col_info.set_column_len(col_info.column_length);
    new_col_info.set_decimal(col_info.decimal);
    new_col_info.set_flag(col_info.flag as i32);
    new_col_info.set_elems(col_info.elems.clone().into());
    new_col_info.set_default_val(col_info.default_val.clone());
    new_col_info.set_pk_handle(col_info.is_primary_key);
    new_col_info.set_array(col_info.is_array);
    Ok(new_col_info)
}

pub(crate) fn grpc_index_info_to_tipb(
    index_info: &tici_grpc::IndexInfo,
) -> Result<tipb::IndexInfo> {
    let mut new_index_info = tipb::IndexInfo::default();
    new_index_info.set_table_id(index_info.table_id);
    new_index_info.set_index_id(index_info.index_id);
    let mut columns = Vec::new();
    for col in index_info.columns.iter() {
        columns.push(grpc_column_info_to_tipb(col)?);
    }
    new_index_info.set_columns(columns.into());
    new_index_info.set_unique(index_info.is_unique);
    Ok(new_index_info)
}

const CURRENT_VERSION: u8 = 1;
/// Default limit for a single KV pair size in bytes.
/// Refer to [tidb_txn_entry_size_limit](https://docs.pingcap.com/tidb/stable/system-variables/#tidb_txn_entry_size_limit-new-in-v760)
/// The default value is the same as `txn-entry-size-limit` in tidb config file,
/// which is 6MiB. And the max value is 120MiB.
/// TODO: Consider add warning log when the size exceed default value limit.
pub const DEFAULT_MAX_KV_BYTES: usize = 128 * 1024 * 1024;

// The data file that DXF writes to the S3.
pub struct File {
    read_buffer: Box<dyn AsyncRead + Unpin>,
    index_meta: IndexMeta,
    decoder: Decoder,
    key_buf: Vec<u8>,
    val_buf: Vec<u8>,
    // max_kv_bytes == 0 means no limit is enforced on the combined key/value size.
    max_kv_bytes: usize,
    bytes_processed: u64,
    input_file_size: u64,
}

impl std::fmt::Debug for File {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        // Skip the read buffer in debug output.
        f.debug_struct("File")
            .field("decoder", &self.decoder)
            .field("index_meta", &self.index_meta)
            .finish()
    }
}

impl File {
    /// Open a file from backend storage with a configurable KV size limit.
    #[allow(clippy::too_many_arguments)]
    pub async fn open(
        s3_bucket: String,
        s3_path: String,
        backend_storage: Arc<dyn BackendStorage>,
        keyspace_id: u32,
        index_id: i64,
        max_kv_bytes: usize,
        parser_info: Option<tici_grpc::ParserInfo>,
        index_created_ts: u64,
    ) -> Result<Self> {
        let (read_buf, meta) = backend_storage
            .get_object(&s3_bucket, &s3_path, None)
            .await
            .with_context(|| {
                format!(
                    "Failed to get_object: bucket={}, key={}",
                    s3_bucket, s3_path
                )
            })?;
        let read_buf = Box::new(read_buf);
        let input_file_size = meta.content_length.unwrap_or(0).max(0) as u64;
        File::new(
            read_buf,
            keyspace_id,
            index_id,
            max_kv_bytes,
            parser_info,
            index_created_ts,
            input_file_size,
        )
        .await
    }

    /// Open a local file with a configurable KV size limit.
    pub async fn open_local(
        file_path: String,
        keyspace_id: u32,
        index_id: i64,
        max_kv_bytes: usize,
        parser_info: Option<tici_grpc::ParserInfo>,
        index_created_ts: u64,
    ) -> Result<Self> {
        let file = tokio::fs::File::open(file_path).await?;
        let input_file_size = file.metadata().await?.len();
        let read_buf = Box::new(file);
        File::new(
            read_buf,
            keyspace_id,
            index_id,
            max_kv_bytes,
            parser_info,
            index_created_ts,
            input_file_size,
        )
        .await
    }

    async fn new(
        mut read_buffer: Box<dyn AsyncRead + Unpin>,
        keyspace_id: u32,
        index_id: i64,
        max_kv_bytes: usize,
        parser_info: Option<tici_grpc::ParserInfo>,
        index_created_ts: u64,
        input_file_size: u64,
    ) -> Result<Self> {
        // HeaderFormat: [Version(1)][TableInfoLen(8)][TableInfo]
        let version = read_buffer.read_u8().await?;
        if version != CURRENT_VERSION {
            return Err(anyhow!(
                "Unknown format version: {version}, expected: {CURRENT_VERSION}"
            ));
        }
        let table_info_len = read_buffer.read_u64().await?;
        let table_info_len =
            usize::try_from(table_info_len).context("table info length overflow")?;
        let mut table_info_buf = vec![0u8; table_info_len];
        read_buffer.read_exact(&mut table_info_buf).await?;
        let commit_ts = read_buffer.read_u64().await?;
        // TiCI allows the import snapshot and CDC to overlap in commit_ts, but
        // they must not leave a gap in coverage:
        // - CDC consumes changes whose commit_ts >= index_created_ts.
        // - The import snapshot must therefore be taken at ts >= index_created_ts, so
        //   it covers all rows committed before that point and may also include rows
        //   that CDC will replay later.
        // - TiDB guarantees the snapshot ts used to generate this import file is >= the
        //   header commit_ts, so validating header commit_ts >= index_created_ts is
        //   sufficient to ensure the import snapshot does not stop before CDC starts.
        //   Any overlap is handled by downstream deduplication.
        ensure!(
            commit_ts >= index_created_ts,
            "import file commit_ts must be >= index created_ts, keyspace_id={} index_id={} commit_ts={} index_created_ts={}",
            keyspace_id,
            index_id,
            commit_ts,
            index_created_ts
        );

        info!(
            "open new import file, version={} table_info_buf={}",
            version,
            String::from_utf8_lossy(&table_info_buf)
        );

        let table_info_from_json: tici_grpc::TiDBTableInfo =
            serde_json::from_slice(&table_info_buf)?;

        // Generate the index info and PK index info from the table info.
        // TODO: let IndexMeta support directly construct from TiDBTableInfo and
        // simplify the code here.
        let table_info = table_info_from_json.get_grpc_table_info(keyspace_id);
        let index_info =
            table_info_from_json.get_grpc_index_info(index_id, keyspace_id, parser_info);
        let (index_meta, pb_index_info, index_type) = match index_info.as_ref() {
            Some(idx_info) => {
                (
                    IndexMeta::new(&table_info, idx_info)?,
                    // transform to tipb struct for decoding
                    grpc_index_info_to_tipb(idx_info)?,
                    idx_info.index_type,
                )
            }
            None => {
                bail!("Index not found in table info, index_id={}", index_id);
            }
        };
        // TODO: In the early stage of TiCI, it does not support parsing the JSON format
        // to TiDBTableInfo, so we have to transform it to tici_grpc::IndexInfo
        // for decoding. But now we actually need to transform the
        // tici_grpc::IndexInfo to tipb::IndexInfo and then do the decoding.
        // We'd better refactor the code to directly get a `tipb::IndexInfo` from
        // `TiDBTableInfo`.
        let pk_index_info =
            if let Some(pk_index_info) = table_info_from_json.get_grpc_pk_index_info(keyspace_id) {
                Some(grpc_index_info_to_tipb(&pk_index_info)?)
            } else {
                None
            };
        let decoder = Decoder::new(
            EvalContext::default(),
            DecoderOptions::from_table_info(
                keyspace_id,
                commit_ts,
                &index_meta,
                pb_index_info,
                index_type,
                pk_index_info,
                table_info_from_json.pk_is_handle,
                table_info_from_json.is_common_handle,
            )?,
        );

        Ok(File {
            read_buffer,
            index_meta,
            decoder,
            key_buf: Vec::new(),
            val_buf: Vec::new(),
            max_kv_bytes,
            bytes_processed: 0,
            input_file_size,
        })
    }

    async fn read_u64_or_eof(&mut self, field: &'static str) -> Result<Option<u64>> {
        let mut buf = [0u8; 8];
        let mut read = 0;
        while read < buf.len() {
            let n = self.read_buffer.read(&mut buf[read..]).await?;
            if n == 0 {
                if read == 0 {
                    return Ok(None);
                }
                bail!(
                    "unexpected EOF while reading {}, read {} of {} bytes",
                    field,
                    read,
                    buf.len()
                );
            }
            read += n;
        }
        Ok(Some(u64::from_be_bytes(buf)))
    }

    async fn read_u64_required(&mut self, field: &'static str) -> Result<u64> {
        self.read_u64_or_eof(field)
            .await?
            .ok_or_else(|| anyhow!("unexpected EOF while reading {}", field))
    }

    // Returns Ok(None) on EOF so callers can stop without treating it as an error.
    pub async fn next(&mut self) -> Result<Option<RawDoc>> {
        // [KV][KV]...[KV]
        // For each KV, its format is [keyLen(8)][valLen(8)][key][value]
        let Some(key_len) = self.read_u64_or_eof("key length").await? else {
            return Ok(None);
        };
        let key_len = key_len as usize;
        let val_len = self.read_u64_required("value length").await? as usize;
        let total_len = key_len.checked_add(val_len).ok_or_else(|| {
            anyhow!(
                "kv length overflow, key_len={} val_len={}",
                key_len,
                val_len
            )
        })?;
        if self.max_kv_bytes > 0 && total_len > self.max_kv_bytes {
            bail!(
                "kv pair too large: key_len={} val_len={} total_len={} max_len={}",
                key_len,
                val_len,
                total_len,
                self.max_kv_bytes
            );
        }

        self.key_buf.resize(key_len, 0);
        self.read_buffer.read_exact(&mut self.key_buf).await?;
        self.val_buf.resize(val_len, 0);
        self.read_buffer.read_exact(&mut self.val_buf).await?;
        self.bytes_processed = self
            .bytes_processed
            .checked_add(total_len as u64)
            .ok_or_else(|| anyhow!("bytes_processed overflow"))?;
        log::trace!(
            "read kv pair: k={} v={}",
            hex::encode(&self.key_buf),
            hex::encode(&self.val_buf)
        );

        let doc = self
            .decoder
            .decode(&self.key_buf, &self.val_buf)
            .with_context(|| {
                format!(
                    "decode error, key={} value={}",
                    hex::encode(&self.key_buf),
                    hex::encode(&self.val_buf)
                )
            })?;
        Ok(Some(doc))
    }

    pub fn get_index_meta(&self) -> &IndexMeta {
        &self.index_meta
    }

    pub fn commit_ts(&self) -> u64 {
        self.decoder.commit_ts()
    }

    pub fn index_info(&self) -> &tipb::IndexInfo {
        self.decoder.index_info()
    }

    pub fn pk_index_info(&self) -> Option<&tipb::IndexInfo> {
        self.decoder.pk_index_info()
    }

    pub fn bytes_processed(&self) -> u64 {
        self.bytes_processed
    }

    pub fn input_file_size(&self) -> u64 {
        self.input_file_size
    }
}

#[cfg(test)]
mod tests {

    use std::path::Path;

    use anyhow::Context;
    use hex::decode;
    use tici_common::{
        codec::{codec_v2::add_keyspace_prefix, table_raw_key},
        index::IndexKey,
        keyspace::NULLSPACEID,
    };
    use tici_grpc::{DEFAULT_PARSER_INFO, tidb_def};
    use tici_test_utils::test_utils::init_logger_for_ut;
    use tidb_query_datatype::codec::{Datum, datum, table};
    use tokio::io::AsyncWriteExt;

    use super::*;

    fn load_table_info(name: &str) -> anyhow::Result<String> {
        let path = Path::new(env!("CARGO_MANIFEST_DIR"))
            .join("testdata")
            .join(name);
        std::fs::read_to_string(&path)
            .with_context(|| format!("read table info from {}", path.display()))
    }

    #[tokio::test]
    async fn reject_import_file_older_than_index_created_ts() -> anyhow::Result<()> {
        init_logger_for_ut();

        let keyspace_id = NULLSPACEID;
        let index_id = 2;
        let commit_ts = 463706994161483785_u64;
        let index_created_ts = commit_ts + 1;
        let table_info_buf = load_table_info("fulltext_index_standard.json")?;

        let mut buffer = Vec::new();
        buffer.push(CURRENT_VERSION);
        buffer.extend_from_slice(&(table_info_buf.len() as u64).to_be_bytes());
        buffer.extend_from_slice(table_info_buf.as_bytes());
        buffer.extend_from_slice(&commit_ts.to_be_bytes());

        let (mut writer, reader) = tokio::io::duplex(buffer.len());
        writer.write_all(&buffer).await?;
        drop(writer);

        let err = File::new(
            Box::new(reader),
            keyspace_id,
            index_id,
            DEFAULT_MAX_KV_BYTES,
            Some(DEFAULT_PARSER_INFO.clone()),
            index_created_ts,
            0,
        )
        .await
        .expect_err("commit_ts older than index_created_ts must be rejected");
        assert!(
            err.to_string()
                .contains("commit_ts must be >= index created_ts")
        );

        Ok(())
    }

    // The test case is based on the following SQLs:
    // CREATE TABLE IF NOT EXISTS test.hdfs_10w (
    //   id BIGINT AUTO_INCREMENT,
    //   timestamp BIGINT,
    //   severity_text VARCHAR(50),
    //   body TEXT,
    //   tenant_id INT,
    //   PRIMARY KEY (tenant_id, id)
    // ) AUTO_INCREMENT = 1000;
    //
    // ALTER TABLE test.hdfs_10w
    // ADD FULLTEXT INDEX ft_body_index (body);
    #[tokio::test]
    async fn parse_fulltext_index_with_clustered_index() -> anyhow::Result<()> {
        init_logger_for_ut();

        let keyspace_id = 1;
        let table_id = 7;
        let index_id = 2;

        let table_info_buf = load_table_info("fulltext_index_with_clustered_index.json")?;

        let commit_ts = 462345502398611563_u64;
        let mut buffer = Vec::new();
        buffer.push(CURRENT_VERSION);
        buffer.extend_from_slice(&(table_info_buf.len() as u64).to_be_bytes());
        buffer.extend_from_slice(table_info_buf.as_bytes());
        buffer.extend_from_slice(&commit_ts.to_be_bytes());

        let mut encode_ctx = EvalContext::default();
        for i in 0..1000_i64 {
            let body = format!("doc_{i}").into_bytes();
            let index_values = vec![Datum::Bytes(body), Datum::I64(1), Datum::I64(i + 1)];
            let encoded = datum::encode_key(&mut encode_ctx, &index_values)?;
            let key = table::encode_index_seek_key(table_id, index_id, &encoded);
            let key = add_keyspace_prefix(keyspace_id, &key);
            let value = b"0";
            buffer.extend_from_slice(&(key.len() as u64).to_be_bytes());
            buffer.extend_from_slice(&(value.len() as u64).to_be_bytes());
            buffer.extend_from_slice(&key);
            buffer.extend_from_slice(value);
        }

        // Write to duplex stream then try to parse it.
        let (mut writer, reader) = tokio::io::duplex(buffer.len());
        writer.write_all(&buffer).await?;
        drop(writer);

        // Verify File reading and parsing.
        let read_buffer = Box::new(reader);
        let mut file = File::new(
            read_buffer,
            keyspace_id,
            index_id,
            DEFAULT_MAX_KV_BYTES,
            Some(DEFAULT_PARSER_INFO.clone()),
            0,
            0,
        )
        .await?;
        assert_eq!(commit_ts, file.commit_ts());
        {
            // index_info
            assert_eq!(table_id, file.index_info().get_table_id());
            assert_eq!(index_id, file.index_info().get_index_id());
            assert_eq!(1, file.index_info().get_columns().len());
            let col = file.index_info().get_columns().first().unwrap();
            assert_eq!(4, col.get_column_id());
            assert_eq!(tidb_def::COLTYPE_BLOB, col.get_tp());
            assert_eq!(65535, col.get_column_len());
        }
        {
            // index_meta
            {
                assert_eq!(1, file.index_meta.index_columns().len());
                let idx_col = file.index_meta.index_columns().first().unwrap();
                assert_eq!(4, idx_col.column_id);
                assert_eq!("body", idx_col.column_name);
                assert_eq!(tidb_def::COLTYPE_BLOB, idx_col.r#type);
                assert_eq!(65535, idx_col.column_length);
            }
            {
                assert_eq!(2, file.index_meta.pk_column_count());
                let pk_col_1 = file.index_meta.pk_columns().first().unwrap().clone();
                assert_eq!(1, pk_col_1.column_id);
                assert_eq!("id", pk_col_1.column_name);
                assert_eq!(tidb_def::COLTYPE_LONGLONG, pk_col_1.r#type);
                assert_eq!(20, pk_col_1.column_length);
                let pk_col_2 = file.index_meta.pk_columns().last().unwrap().clone();
                assert_eq!(5, pk_col_2.column_id);
                assert_eq!("tenant_id", pk_col_2.column_name);
                assert_eq!(tidb_def::COLTYPE_LONG, pk_col_2.r#type);
                assert_eq!(11, pk_col_2.column_length);
            }
        }

        let mut docs = Vec::new();
        let mut expected_ctx = EvalContext::default();
        for i in 0..1000_i64 {
            let doc = file.next().await?.context("expected doc")?;
            let handle_bytes =
                datum::encode_key(&mut expected_ctx, &[Datum::I64(1), Datum::I64(i + 1)])?;
            let mut suffix = b"_r".to_vec();
            suffix.extend_from_slice(&handle_bytes);
            let index_key = IndexKey::new(keyspace_id, table_id, index_id);
            let expected_handle = table_raw_key(&index_key, &suffix);
            assert_eq!(doc.handle.raw(), expected_handle.as_slice());
            docs.push(doc);
        }
        assert_eq!(1000, docs.len());

        // No more docs.
        assert!(file.next().await?.is_none());
        Ok(())
    }

    // The test case is based on the following SQLs:
    // CREATE TABLE test.mysql_standard (
    //   a int default 0,
    //   b int default 0,
    //   content TEXT,
    //   PRIMARY KEY(a, b)
    // );
    //
    // INSERT INTO test.mysql_standard (a, b, content) VALUES
    //   (1, 1, 'xy'),
    //   (2, 2, 'xyz'),
    //   (3, 3, REPEAT('a', 84)),
    //   (4, 4, REPEAT('b', 85));
    // INSERT INTO test.mysql_standard (a, b, content) VALUES
    //   (5, 5, 'Hello World'),
    //   (6, 6, 'MySQL Database'),
    //   (7, 7, 'Apple'),
    //   (8, 8, 'This is a pen');
    //
    // ALTER TABLE test.mysql_standard ADD FULLTEXT INDEX
    // ft_std (content) WITH PARSER STANDARD;
    #[tokio::test]
    async fn parse_fulltext_index_standard() -> anyhow::Result<()> {
        init_logger_for_ut();
        let table_info_buf = load_table_info("fulltext_index_standard.json")?;
        let kvpairs = [
            (
                "7480000000000000895f698000000000000002038000000000000001038000000000000001038000000000000001038000000000000001",
                "007d018000010000000302007879",
            ),
            (
                "7480000000000000895f698000000000000002038000000000000002038000000000000002038000000000000002038000000000000002",
                "007d0180000100000003030078797a",
            ),
            (
                "7480000000000000895f698000000000000002038000000000000003038000000000000003038000000000000003038000000000000003",
                "007d01800001000000035400616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161",
            ),
            (
                "7480000000000000895f698000000000000002038000000000000004038000000000000004038000000000000004038000000000000004",
                "007d0180000100000003550062626262626262626262626262626262626262626262626262626262626262626262626262626262626262626262626262626262626262626262626262626262626262626262626262626262626262626262626262",
            ),
            (
                "7480000000000000895f698000000000000002038000000000000005038000000000000005038000000000000005038000000000000005",
                "007d01800001000000030b0048656c6c6f20576f726c64",
            ),
            (
                "7480000000000000895f698000000000000002038000000000000006038000000000000006038000000000000006038000000000000006",
                "007d01800001000000030e004d7953514c204461746162617365",
            ),
            (
                "7480000000000000895f698000000000000002038000000000000007038000000000000007038000000000000007038000000000000007",
                "007d018000010000000305004170706c65",
            ),
            (
                "7480000000000000895f698000000000000002038000000000000008038000000000000008038000000000000008038000000000000008",
                "007d01800001000000030d005468697320697320612070656e",
            ),
        ];
        let keyspace_id = NULLSPACEID;
        let index_id = 2;
        let commit_ts = 463706994161483785_u64;

        let mut buffer = Vec::new();
        buffer.push(CURRENT_VERSION);
        buffer.extend_from_slice(&(table_info_buf.len() as u64).to_be_bytes());
        buffer.extend_from_slice(table_info_buf.as_bytes());
        buffer.extend_from_slice(&commit_ts.to_be_bytes());
        for (k, v) in kvpairs.iter() {
            let key = decode(k).unwrap();
            let value = decode(v).unwrap();
            buffer.extend_from_slice(&(key.len() as u64).to_be_bytes());
            buffer.extend_from_slice(&(value.len() as u64).to_be_bytes());
            buffer.extend_from_slice(&key);
            buffer.extend_from_slice(&value);
        }

        // Write to duplex stream then try to parse it.
        let (mut writer, reader) = tokio::io::duplex(buffer.len());
        writer.write_all(&buffer).await?;
        drop(writer);

        // Verify File reading and parsing.
        let read_buffer = Box::new(reader);
        let mut file = File::new(
            read_buffer,
            keyspace_id,
            index_id,
            DEFAULT_MAX_KV_BYTES,
            Some(DEFAULT_PARSER_INFO.clone()),
            0,
            0,
        )
        .await?;

        let mut docs = Vec::new();
        for _ in 0..kvpairs.len() {
            docs.push(file.next().await?.context("expected doc")?);
        }
        assert_eq!(8, docs.len());

        let content_3_3 = "a".repeat(84);
        let content_4_4 = "b".repeat(85);
        let expected_docs = [
            vec![(3, Some("xy")), (1, Some("1")), (2, Some("1"))],
            vec![(3, Some("xyz")), (1, Some("2")), (2, Some("2"))],
            vec![(3, Some(&content_3_3)), (1, Some("3")), (2, Some("3"))],
            vec![(3, Some(&content_4_4)), (1, Some("4")), (2, Some("4"))],
            vec![(3, Some("Hello World")), (1, Some("5")), (2, Some("5"))],
            vec![(3, Some("MySQL Database")), (1, Some("6")), (2, Some("6"))],
            vec![(3, Some("Apple")), (1, Some("7")), (2, Some("7"))],
            vec![(3, Some("This is a pen")), (1, Some("8")), (2, Some("8"))],
        ];
        let expected_handles = [
            "7480000000000000895f72038000000000000001038000000000000001",
            "7480000000000000895f72038000000000000002038000000000000002",
            "7480000000000000895f72038000000000000003038000000000000003",
            "7480000000000000895f72038000000000000004038000000000000004",
            "7480000000000000895f72038000000000000005038000000000000005",
            "7480000000000000895f72038000000000000006038000000000000006",
            "7480000000000000895f72038000000000000007038000000000000007",
            "7480000000000000895f72038000000000000008038000000000000008",
        ];
        for (doc, expected_cols, expected_handle) in
            itertools::izip!(docs.iter(), expected_docs.iter(), expected_handles.iter())
        {
            assert_eq!(expected_handle, &hex::encode(doc.handle.raw()));
            let actual_cols: Vec<(i64, Option<&str>)> = doc
                .cols
                .iter()
                .map(|(cid, val)| (*cid, val.as_deref()))
                .collect();
            assert_eq!(expected_cols, &actual_cols);
        }

        Ok(())
    }

    // The test case is based on the following SQLs:
    // CREATE TABLE test.mysql_ngram (
    //   a int default 0,
    //   b int default 0,
    //   content TEXT,
    //   PRIMARY KEY(a, b)
    // );
    //
    // INSERT INTO test.mysql_ngram (a, b, content) VALUES
    //   (1, 1, 'x'),
    //   (2, 2, 'xy'),
    //   (3, 3, 'z'),
    //   (4, 4, 'mars'),
    //   (5, 5, 'abcde'),
    //   (6, 6, 'my sql'),
    //   (7, 7, 'hi,world'),
    //   (8, 8, 'one.two'),
    //   (10, 10, 'Hi MySQL');
    //
    // ALTER TABLE test.mysql_ngram ADD FULLTEXT INDEX
    // ft_ngram (content) WITH PARSER ngram;
    #[tokio::test]
    async fn parse_fulltext_index_ngram() -> anyhow::Result<()> {
        init_logger_for_ut();
        let table_info_buf = load_table_info("fulltext_index_ngram.json")?;
        let kvpairs = [
            (
                "74800000000000008d5f698000000000000002038000000000000001038000000000000001038000000000000001038000000000000001",
                "007d0180000100000003010078",
            ),
            (
                "74800000000000008d5f698000000000000002038000000000000002038000000000000002038000000000000002038000000000000002",
                "007d018000010000000302007879",
            ),
            (
                "74800000000000008d5f698000000000000002038000000000000003038000000000000003038000000000000003038000000000000003",
                "007d018000010000000301007a",
            ),
            (
                "74800000000000008d5f698000000000000002038000000000000004038000000000000004038000000000000004038000000000000004",
                "007d018000010000000304006d617273",
            ),
            (
                "74800000000000008d5f698000000000000002038000000000000005038000000000000005038000000000000005038000000000000005",
                "007d018000010000000305006162636465",
            ),
            (
                "74800000000000008d5f698000000000000002038000000000000006038000000000000006038000000000000006038000000000000006",
                "007d018000010000000306006d792073716c",
            ),
            (
                "74800000000000008d5f698000000000000002038000000000000007038000000000000007038000000000000007038000000000000007",
                "007d0180000100000003080068692c776f726c64",
            ),
            (
                "74800000000000008d5f698000000000000002038000000000000008038000000000000008038000000000000008038000000000000008",
                "007d018000010000000307006f6e652e74776f",
            ),
            (
                "74800000000000008d5f69800000000000000203800000000000000a03800000000000000a03800000000000000a03800000000000000a",
                "007d018000010000000308004869204d7953514c",
            ),
        ];
        let keyspace_id = NULLSPACEID;
        let index_id = 2;
        let commit_ts = 463706994161483785_u64;
        let mut buffer = Vec::new();
        buffer.push(CURRENT_VERSION);
        buffer.extend_from_slice(&(table_info_buf.len() as u64).to_be_bytes());
        buffer.extend_from_slice(table_info_buf.as_bytes());
        buffer.extend_from_slice(&commit_ts.to_be_bytes());
        for (k, v) in kvpairs.iter() {
            let key = decode(k).unwrap();
            let value = decode(v).unwrap();
            buffer.extend_from_slice(&(key.len() as u64).to_be_bytes());
            buffer.extend_from_slice(&(value.len() as u64).to_be_bytes());
            buffer.extend_from_slice(&key);
            buffer.extend_from_slice(&value);
        }

        // Write to duplex stream then try to parse it.
        let (mut writer, reader) = tokio::io::duplex(buffer.len());
        writer.write_all(&buffer).await?;
        drop(writer);

        // Verify File reading and parsing.
        let read_buffer = Box::new(reader);
        let mut file = File::new(
            read_buffer,
            keyspace_id,
            index_id,
            DEFAULT_MAX_KV_BYTES,
            Some(DEFAULT_PARSER_INFO.clone()),
            0,
            0,
        )
        .await?;

        let mut docs = Vec::new();
        for _ in 0..kvpairs.len() {
            docs.push(file.next().await?.context("expected doc")?);
        }
        assert_eq!(9, docs.len());

        let expected_docs = [
            vec![(3, Some("x")), (1, Some("1")), (2, Some("1"))],
            vec![(3, Some("xy")), (1, Some("2")), (2, Some("2"))],
            vec![(3, Some("z")), (1, Some("3")), (2, Some("3"))],
            vec![(3, Some("mars")), (1, Some("4")), (2, Some("4"))],
            vec![(3, Some("abcde")), (1, Some("5")), (2, Some("5"))],
            vec![(3, Some("my sql")), (1, Some("6")), (2, Some("6"))],
            vec![(3, Some("hi,world")), (1, Some("7")), (2, Some("7"))],
            vec![(3, Some("one.two")), (1, Some("8")), (2, Some("8"))],
            vec![(3, Some("Hi MySQL")), (1, Some("10")), (2, Some("10"))],
        ];
        let expected_handles = [
            "74800000000000008d5f72038000000000000001038000000000000001",
            "74800000000000008d5f72038000000000000002038000000000000002",
            "74800000000000008d5f72038000000000000003038000000000000003",
            "74800000000000008d5f72038000000000000004038000000000000004",
            "74800000000000008d5f72038000000000000005038000000000000005",
            "74800000000000008d5f72038000000000000006038000000000000006",
            "74800000000000008d5f72038000000000000007038000000000000007",
            "74800000000000008d5f72038000000000000008038000000000000008",
            "74800000000000008d5f7203800000000000000a03800000000000000a",
        ];
        for (doc, expected_cols, expected_handle) in
            itertools::izip!(docs.iter(), expected_docs.iter(), expected_handles.iter())
        {
            assert_eq!(expected_handle, &hex::encode(doc.handle.raw()));
            let actual_cols: Vec<(i64, Option<&str>)> = doc
                .cols
                .iter()
                .map(|(cid, val)| (*cid, val.as_deref()))
                .collect();
            assert_eq!(expected_cols, &actual_cols);
        }

        Ok(())
    }

    // The test case is based on the following SQLs:
    // CREATE TABLE t_h (
    //   id BIGINT PRIMARY KEY AUTO_INCREMENT,
    //   a  INT NOT NULL,
    //   b  VARCHAR(20) NOT NULL,
    //   c  VARCHAR(20) NOT NULL
    // );
    //
    // INSERT INTO t_h(a,b,c) VALUES
    // (10,'foo','x'),
    // ( 9,'foo','y'),
    // ( 8,'bar','x'),
    // ( 7,'baz','z'),
    // ( 6,'bar','y');
    //
    // CREATE HYBRID INDEX idx_h
    // ON t_h(a, b, c)
    // PARAMETER '{
    //   "inverted":    { "columns": ["b"] },
    //   "sort":        { "columns": ["a"], "order": ["asc"] },
    //   "sharding_key":{ "columns": ["a"] }
    // }';
    #[tokio::test]
    async fn parse_hybrid_index_with_sharding_key_on_non_pk_col() -> anyhow::Result<()> {
        init_logger_for_ut();

        let keyspace_id = NULLSPACEID;
        let index_id = 1;
        let commit_ts = 463706994161483785_u64;
        let table_info_buf = load_table_info("hybrid_index_sharding_key_non_pk_col.json")?;

        let kvpairs = [
            (
                "7480000000000000c85f698000000000000001038000000000000006038000000000000005",
                "008000030000000203040100040005000662617279",
            ),
            (
                "7480000000000000c85f698000000000000001038000000000000007038000000000000004",
                "008000030000000203040100040005000762617a7a",
            ),
            (
                "7480000000000000c85f698000000000000001038000000000000008038000000000000003",
                "008000030000000203040100040005000862617278",
            ),
            (
                "7480000000000000c85f698000000000000001038000000000000009038000000000000002",
                "0080000300000002030401000400050009666f6f79",
            ),
            (
                "7480000000000000c85f69800000000000000103800000000000000a038000000000000001",
                "008000030000000203040100040005000a666f6f78",
            ),
        ];

        let mut buffer = Vec::new();
        buffer.push(CURRENT_VERSION);
        buffer.extend_from_slice(&(table_info_buf.len() as u64).to_be_bytes());
        buffer.extend_from_slice(table_info_buf.as_bytes());
        buffer.extend_from_slice(&commit_ts.to_be_bytes());
        for (k, v) in kvpairs.iter() {
            let key = decode(k).unwrap();
            let value = decode(v).unwrap();
            buffer.extend_from_slice(&(key.len() as u64).to_be_bytes());
            buffer.extend_from_slice(&(value.len() as u64).to_be_bytes());
            buffer.extend_from_slice(&key);
            buffer.extend_from_slice(&value);
        }

        // Write to duplex stream then try to parse it.
        let (mut writer, reader) = tokio::io::duplex(buffer.len());
        writer.write_all(&buffer).await?;
        drop(writer);

        // Verify File reading and parsing.
        let read_buffer = Box::new(reader);
        let mut file = File::new(
            read_buffer,
            keyspace_id,
            index_id,
            DEFAULT_MAX_KV_BYTES,
            Some(DEFAULT_PARSER_INFO.clone()),
            0,
            0,
        )
        .await?;
        assert_eq!(commit_ts.to_string(), file.commit_ts().to_string());
        assert_eq!(188, file.index_info().get_table_id());
        assert_eq!(index_id, file.index_info().get_index_id());
        assert_eq!(3, file.index_info().get_columns().len());
        assert!(file.pk_index_info().is_some());

        // Should be able to read all 5 docs.
        let mut docs = Vec::new();
        for _ in 0..kvpairs.len() {
            docs.push(file.next().await?.context("expected doc")?);
        }

        assert_eq!(5, docs.len());
        let expected_docs = [
            vec![
                (2, Some("6")),
                (3, Some("bar")),
                (4, Some("y")),
                (1, Some("5")),
            ],
            vec![
                (2, Some("7")),
                (3, Some("baz")),
                (4, Some("z")),
                (1, Some("4")),
            ],
            vec![
                (2, Some("8")),
                (3, Some("bar")),
                (4, Some("x")),
                (1, Some("3")),
            ],
            vec![
                (2, Some("9")),
                (3, Some("foo")),
                (4, Some("y")),
                (1, Some("2")),
            ],
            vec![
                (2, Some("10")),
                (3, Some("foo")),
                (4, Some("x")),
                (1, Some("1")),
            ],
        ];

        for (doc, expected_cols, (expected_key_hex, _)) in
            itertools::izip!(docs.iter(), expected_docs.iter(), kvpairs.iter())
        {
            // For hybrid index with sharding key, doc.handle is the same format as the
            // index key.
            assert_eq!(expected_key_hex, &hex::encode(doc.handle.raw()));
            let actual_cols: Vec<(i64, Option<&str>)> = doc
                .cols
                .iter()
                .map(|(cid, val)| (*cid, val.as_deref()))
                .collect();
            assert_eq!(expected_cols, &actual_cols);
        }

        // No more docs.
        assert!(file.next().await?.is_none());
        Ok(())
    }

    #[tokio::test]
    async fn next_returns_decode_error() -> anyhow::Result<()> {
        init_logger_for_ut();

        // The following kvpairs are based on the same table/index as
        // `parse_hybrid_index_with_sharding_key_non_pk_col` test, but the keyspace_id
        // is different, which will cause the decoder to fail to decode the key/value
        // into a doc.
        let keyspace_id = 1234;
        let index_id = 1;
        let commit_ts = 463706994161483785_u64;
        let table_info_buf = load_table_info("hybrid_index_sharding_key_non_pk_col.json")?;

        let kvpairs = [
            (
                "7480000000000000c85f698000000000000001038000000000000006038000000000000005",
                "008000030000000203040100040005000662617279",
            ),
            (
                "7480000000000000c85f698000000000000001038000000000000007038000000000000004",
                "008000030000000203040100040005000762617a7a",
            ),
            (
                "7480000000000000c85f698000000000000001038000000000000008038000000000000003",
                "008000030000000203040100040005000862617278",
            ),
            (
                "7480000000000000c85f698000000000000001038000000000000009038000000000000002",
                "0080000300000002030401000400050009666f6f79",
            ),
            (
                "7480000000000000c85f69800000000000000103800000000000000a038000000000000001",
                "008000030000000203040100040005000a666f6f78",
            ),
        ];

        let mut buffer = Vec::new();
        buffer.push(CURRENT_VERSION);
        buffer.extend_from_slice(&(table_info_buf.len() as u64).to_be_bytes());
        buffer.extend_from_slice(table_info_buf.as_bytes());
        buffer.extend_from_slice(&commit_ts.to_be_bytes());
        for (k, v) in kvpairs.iter() {
            let key = decode(k).unwrap();
            let value = decode(v).unwrap();
            buffer.extend_from_slice(&(key.len() as u64).to_be_bytes());
            buffer.extend_from_slice(&(value.len() as u64).to_be_bytes());
            buffer.extend_from_slice(&key);
            buffer.extend_from_slice(&value);
        }

        // Write to duplex stream then try to parse it.
        let (mut writer, reader) = tokio::io::duplex(buffer.len());
        writer.write_all(&buffer).await?;
        drop(writer);

        // Verify File reading and parsing.
        let read_buffer = Box::new(reader);
        let mut file = File::new(
            read_buffer,
            keyspace_id,
            index_id,
            DEFAULT_MAX_KV_BYTES,
            Some(DEFAULT_PARSER_INFO.clone()),
            0,
            0,
        )
        .await?;
        assert_eq!(commit_ts.to_string(), file.commit_ts().to_string());
        assert_eq!(188, file.index_info().get_table_id());
        assert_eq!(index_id, file.index_info().get_index_id());
        assert_eq!(3, file.index_info().get_columns().len());
        assert!(file.pk_index_info().is_some());

        // Should return an error when the key/value cannot be decoded into a doc.
        assert!(file.next().await.is_err());
        Ok(())
    }

    // The test case is based on the following SQLs:
    // CREATE TABLE t_h (
    //   id BIGINT PRIMARY KEY AUTO_INCREMENT,
    //   sk INT NOT NULL,           -- sharding/sort key (NOT the PK)
    //   b  VARCHAR(20) NOT NULL,   -- inverted column
    //   c  VARCHAR(20) NOT NULL
    // );
    //
    // INSERT INTO t_h(sk,b,c) VALUES
    //   (10,'foo','x'),
    //   ( 9,'foo','y'),
    //   ( 8,'bar','x'),
    //   ( 7,'baz','z'),
    //   ( 6,'bar','y');
    //
    // CREATE HYBRID INDEX idx_h
    // ON t_h(sk, b, c)
    // PARAMETER '{
    //   "inverted":    { "columns": ["b"] },
    //   "sort":        { "columns": ["sk"], "order": ["asc"] },
    //   "sharding_key":{ "columns": ["sk"] }
    // }';
    //
    // CREATE HYBRID INDEX idx_h2
    // ON t_h(sk, b, c)
    // PARAMETER '{
    //   "inverted":    { "columns": ["b"] },
    //   "sort":        { "columns": ["sk","c"], "order": ["asc","desc"] },
    //   "sharding_key":{ "columns": ["sk","c"] }
    // }';
    #[tokio::test]
    async fn parse_hybrid_index_with_sharding_key_on_explicit_sort_key() -> anyhow::Result<()> {
        init_logger_for_ut();
        let keyspace_id = NULLSPACEID;
        let table_info_buf = load_table_info("hybrid_index_sharding_key_explicit_sort_key.json")?;

        let kvpairs_idx_h = vec![
            (
                "7480000000000000d05f698000000000000001038000000000000006038000000000000005",
                "008000030000000203040100040005000662617279",
            ),
            (
                "7480000000000000d05f698000000000000001038000000000000007038000000000000004",
                "008000030000000203040100040005000762617a7a",
            ),
            (
                "7480000000000000d05f698000000000000001038000000000000008038000000000000003",
                "008000030000000203040100040005000862617278",
            ),
            (
                "7480000000000000d05f698000000000000001038000000000000009038000000000000002",
                "0080000300000002030401000400050009666f6f79",
            ),
            (
                "7480000000000000d05f69800000000000000103800000000000000a038000000000000001",
                "008000030000000203040100040005000a666f6f78",
            ),
        ];

        let kvpairs_idx_h2 = vec![
            (
                "7480000000000000d05f698000000000000002038000000000000006017900000000000000f8038000000000000005",
                "008000030000000203040100040005000662617279",
            ),
            (
                "7480000000000000d05f698000000000000002038000000000000007017a00000000000000f8038000000000000004",
                "008000030000000203040100040005000762617a7a",
            ),
            (
                "7480000000000000d05f698000000000000002038000000000000008017800000000000000f8038000000000000003",
                "008000030000000203040100040005000862617278",
            ),
            (
                "7480000000000000d05f698000000000000002038000000000000009017900000000000000f8038000000000000002",
                "0080000300000002030401000400050009666f6f79",
            ),
            (
                "7480000000000000d05f69800000000000000203800000000000000a017800000000000000f8038000000000000001",
                "008000030000000203040100040005000a666f6f78",
            ),
        ];

        let expected_docs = [
            vec![
                (2, Some("6")),
                (3, Some("bar")),
                (4, Some("y")),
                (1, Some("5")),
            ],
            vec![
                (2, Some("7")),
                (3, Some("baz")),
                (4, Some("z")),
                (1, Some("4")),
            ],
            vec![
                (2, Some("8")),
                (3, Some("bar")),
                (4, Some("x")),
                (1, Some("3")),
            ],
            vec![
                (2, Some("9")),
                (3, Some("foo")),
                (4, Some("y")),
                (1, Some("2")),
            ],
            vec![
                (2, Some("10")),
                (3, Some("foo")),
                (4, Some("x")),
                (1, Some("1")),
            ],
        ];

        let cases = vec![
            (1, 463736206020575238_u64, &kvpairs_idx_h, 1_usize),
            (2, 463736263036895238_u64, &kvpairs_idx_h2, 2_usize),
        ];

        for (index_id, commit_ts, kvpairs, sharding_len) in cases {
            let mut buffer = Vec::new();
            buffer.push(CURRENT_VERSION);
            buffer.extend_from_slice(&(table_info_buf.len() as u64).to_be_bytes());
            buffer.extend_from_slice(table_info_buf.as_bytes());
            buffer.extend_from_slice(&commit_ts.to_be_bytes());
            for (k, v) in kvpairs.iter() {
                let key = decode(k).unwrap();
                let value = decode(v).unwrap();
                buffer.extend_from_slice(&(key.len() as u64).to_be_bytes());
                buffer.extend_from_slice(&(value.len() as u64).to_be_bytes());
                buffer.extend_from_slice(&key);
                buffer.extend_from_slice(&value);
            }

            // Write to duplex stream then try to parse it.
            let (mut writer, reader) = tokio::io::duplex(buffer.len());
            writer.write_all(&buffer).await?;
            drop(writer);

            // Verify File reading and parsing.
            let read_buffer = Box::new(reader);
            let mut file = File::new(
                read_buffer,
                keyspace_id,
                index_id,
                DEFAULT_MAX_KV_BYTES,
                Some(DEFAULT_PARSER_INFO.clone()),
                0,
                0,
            )
            .await?;
            assert_eq!(commit_ts, file.commit_ts());
            assert_eq!(208, file.index_info().get_table_id());
            assert_eq!(index_id, file.index_info().get_index_id());
            assert_eq!(3, file.index_info().get_columns().len());
            assert!(file.pk_index_info().is_some());
            assert_eq!(sharding_len, file.index_meta.get_sharding_columns().len());

            // Should be able to read all docs.
            let mut docs = Vec::new();
            for _ in 0..kvpairs.len() {
                docs.push(file.next().await?.context("expected doc")?);
            }

            assert_eq!(kvpairs.len(), docs.len());
            for (doc, expected_cols, (expected_key_hex, _)) in
                itertools::izip!(docs.iter(), expected_docs.iter(), kvpairs.iter())
            {
                // For hybrid index with sharding key, doc.handle is the same format as the
                // index key.
                assert_eq!(expected_key_hex, &hex::encode(doc.handle.raw()));
                let actual_cols: Vec<(i64, Option<&str>)> = doc
                    .cols
                    .iter()
                    .map(|(cid, val)| (*cid, val.as_deref()))
                    .collect();
                assert_eq!(expected_cols, &actual_cols);
            }

            // No more docs.
            assert!(file.next().await?.is_none());
        }

        Ok(())
    }
}