// Copyright 2026 PingCAP, Inc.

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

use anyhow::Result;
use clara_fts::test_util::{PlainFtsQueryInfo, make_unscored_query, ngram_index_field};
use kvenginepb::fts::{
    FtsFieldLayout, FullTextAnalyzerType, FullTextIndexDef, FullTextIndexFieldDef,
};
use tantivy::{DocSet, postings::Postings, schema::FieldType};
use tidb_query_datatype::FieldTypeTp;

use super::{ColumnarToFtsL0Opts, columnar_to_fts_l0};
use crate::table::{
    SnapVersion,
    fts::{
        IntPk, PackedFile, PkReader, lp_key,
        test_util::{SchemaBuilder, make_exact_match_expr_query, new_columnar, new_schema_file},
    },
};

fn legacy_index_layouts(
    index_pairs: &[(i64, i64)],
) -> HashMap<(i64, i64), crate::table::fts::TrackedIndexLayout> {
    index_pairs
        .iter()
        .map(|&(table_id, index_id)| {
            let meta = crate::table::fts::new_legacy_tracked_index_meta(table_id, index_id);
            (
                (table_id, index_id),
                crate::table::fts::tracked_index_layout_from_tracked_meta(&meta, &[]).unwrap(),
            )
        })
        .collect()
}

fn tracked_field(col_id: i64) -> FullTextIndexFieldDef {
    let mut field = FullTextIndexFieldDef::new();
    field.set_col_id(col_id);
    field
}

fn tracked_field_with_analyzer(
    col_id: i64,
    analyzer_type: FullTextAnalyzerType,
) -> FullTextIndexFieldDef {
    tracked_field_with_option(col_id, analyzer_type, true, None)
}

fn tracked_field_with_option(
    col_id: i64,
    analyzer_type: FullTextAnalyzerType,
    enable_bm25: bool,
    delimiter: Option<&str>,
) -> FullTextIndexFieldDef {
    let mut field = tracked_field(col_id);
    let option = field.mut_index_field_option();
    option.set_analyzer_type(analyzer_type);
    option.set_enable_bm25(enable_bm25);
    if let Some(delimiter) = delimiter {
        option
            .mut_path_hierarchy()
            .set_delimiter(delimiter.to_string());
    }
    field
}

fn ordinal_index_layouts(
    entries: &[((i64, i64), Vec<FullTextIndexFieldDef>)],
    columns: &[tipb::ColumnInfo],
) -> HashMap<(i64, i64), crate::table::fts::TrackedIndexLayout> {
    entries
        .iter()
        .map(|((table_id, index_id), index_fields)| {
            let meta = crate::table::fts::new_tracked_index_meta(
                *table_id,
                *index_id,
                index_fields.clone(),
                FtsFieldLayout::FtsFieldLayoutIndexFieldOrdinalString,
            );
            (
                (*table_id, *index_id),
                crate::table::fts::tracked_index_layout_from_tracked_meta(&meta, columns).unwrap(),
            )
        })
        .collect()
}

#[tokio::test]
async fn test_columnar_l0_to_fts_basic() -> Result<()> {
    let table_id = 30;

    let schema = SchemaBuilder::<IntPk>::new(table_id)
        .column(7, FieldTypeTp::String)
        .fts_index(FullTextIndexDef {
            index_id: 1,
            col_id: 7, // Column 7 is the text column
            parser_type: "STANDARD_V1".to_string(),
            ..Default::default()
        })
        .schema();
    let schema_file = new_schema_file(&[&schema]);

    let columnar_file = new_columnar(1, 0)
        .table::<IntPk>(&schema, |row| {
            row(0, 1000, false, |datum| {
                datum("abc first entry");
            });
            row(1, 950, false, |datum| {
                datum("plain content");
            });
            row(2, 1020, false, |datum| {
                datum("another abc value");
            });
            row(3, 980, false, |datum| {
                datum("quiet note");
            });
        })
        .finish_as_file();

    // Convert to FTS L0
    let index_pairs = &[(table_id, 1)];
    let index_layouts = legacy_index_layouts(index_pairs);
    let opts = ColumnarToFtsL0Opts {
        columnar_files: &[columnar_file],
        schema_file,
        index_pairs, // Specify the explicit index_id from fts_index definition
        index_layouts: &index_layouts,
        encryption_key: None,
        snap_version: SnapVersion::zero(),
        l0_file_max_size: 0,
    };

    let file_data = columnar_to_fts_l0(opts).await?;
    assert!(!file_data.is_empty(), "Should generate FTS data");

    let output = file_data.into_iter().next().unwrap();
    let packed_file = PackedFile::from_buffer(&output.data)?;

    // Test logical partition structure
    assert!(
        packed_file.cached_get_lp(&lp_key(1, 1)).await?.is_none(),
        "Should not find non-existent table"
    );

    let lp = packed_file.cached_get_lp(&lp_key(30, 1)).await?;
    assert!(lp.is_some(), "Should find the correct logical partition");

    assert!(
        packed_file.cached_get_lp(&lp_key(30, 5)).await?.is_none(),
        "Should not find non-existent index"
    );

    // Enhanced tests: Test IndexReader and search functionality
    // Extract the Int variant from the enum
    let lp = lp.unwrap();
    let lp = lp.as_int_lp().unwrap();
    assert!(lp.props().get_n_pk() > 0, "Should have documents");

    // Test Tantivy index reading and searching
    let reader = lp.cached_read_index()?;
    let mut results = Vec::new();

    // Test searching for "abc" - rows above embed "abc" so the query should
    // succeed.
    let query = make_unscored_query(&PlainFtsQueryInfo {
        query: "abc".to_string(),
        ..Default::default()
    });
    reader.search(&query, &mut results)?;
    assert!(
        !results.is_empty(),
        "Should find documents containing 'abc' from generated test data"
    );

    // Test searching for a term that doesn't exist
    let query = make_unscored_query(&PlainFtsQueryInfo {
        query: "nonexistent".to_string(),
        ..Default::default()
    });
    reader.search(&query, &mut results)?;
    assert_eq!(
        results.len(),
        0,
        "Should not find any documents containing 'nonexistent'"
    );

    // Test MVCC functionality with has_newer_version
    // Test that non-existent handles return false
    assert!(
        !lp.has_newer_version(&IntPk::encode(999), 0, u64::MAX)
            .await?,
        "Non-existent handle should return false"
    );

    // Test with version 1000 (row 0 uses version 1000)
    assert!(
        lp.has_newer_version(&IntPk::encode(0), 900, 1100).await?,
        "Handle 0 should have version around 1000"
    );
    assert!(
        !lp.has_newer_version(&IntPk::encode(0), 1000, 1100).await?,
        "Handle 0 version should not be > 1000"
    );

    Ok(())
}

#[tokio::test]
async fn test_columnar_l0_to_fts_multi_column_layout_selects_requested_field() -> Result<()> {
    let table_id = 31;

    let mut index_def = FullTextIndexDef {
        index_id: 2,
        col_id: 7,
        parser_type: "STANDARD_V1".to_string(),
        ..Default::default()
    };
    let index_fields = [7, 8]
        .iter()
        .map(|&col_id| {
            tracked_field_with_analyzer(col_id, FullTextAnalyzerType::FtsAnalyzerStandardV1)
        })
        .collect::<Vec<_>>();
    for index_field in &index_fields {
        index_def.mut_index_fields().push(index_field.clone());
    }

    let schema = SchemaBuilder::<IntPk>::new(table_id)
        .column(7, FieldTypeTp::String)
        .column(8, FieldTypeTp::String)
        .fts_index(index_def)
        .schema();
    let schema_file = new_schema_file(&[&schema]);

    let columnar_file = new_columnar(1, 0)
        .table::<IntPk>(&schema, |row| {
            row(0, 1000, false, |datum| {
                datum("alpha");
                datum("target");
            });
            row(1, 990, false, |datum| {
                datum("target");
                datum("beta");
            });
        })
        .finish_as_file();

    let index_pairs = &[(table_id, 2)];
    let index_layouts = ordinal_index_layouts(&[((table_id, 2), index_fields)], &schema.columns);
    let outputs = columnar_to_fts_l0(ColumnarToFtsL0Opts {
        columnar_files: &[columnar_file],
        schema_file,
        index_pairs,
        index_layouts: &index_layouts,
        encryption_key: None,
        snap_version: SnapVersion::zero(),
        l0_file_max_size: 0,
    })
    .await?;

    let packed_file = PackedFile::from_buffer(&outputs.into_iter().next().unwrap().data)?;
    let lp = packed_file
        .cached_get_lp(&lp_key(table_id, 2))
        .await?
        .unwrap();
    let reader = lp.as_int_lp().unwrap().cached_read_index()?;

    let query = make_unscored_query(&PlainFtsQueryInfo {
        query: "target".to_string(),
        column_id: 8,
        index_id: 2,
        ..Default::default()
    });
    let tracked_index_layout = index_layouts.get(&(table_id, 2)).unwrap();
    let query = clara_fts::Query::new_with_index_layout(
        query.info().clone(),
        &tracked_index_layout.index_fields,
        &tracked_index_layout.index_layout,
    )?;
    let mut results = Vec::new();
    reader.search(&query, &mut results)?;
    assert_eq!(results.len(), 1);
    assert_eq!(results[0].doc_id, 0);

    Ok(())
}

#[tokio::test]
async fn test_columnar_l0_to_fts_multi_column_layout_uses_per_field_tokenizers() -> Result<()> {
    let table_id = 32;

    let mut index_def = FullTextIndexDef {
        index_id: 3,
        ..Default::default()
    };
    let index_fields = vec![
        tracked_field_with_analyzer(7, FullTextAnalyzerType::FtsAnalyzerStandardV1),
        tracked_field_with_analyzer(8, FullTextAnalyzerType::FtsAnalyzerExactValueV1),
    ];
    for index_field in &index_fields {
        index_def.mut_index_fields().push(index_field.clone());
    }

    let schema = SchemaBuilder::<IntPk>::new(table_id)
        .column(7, FieldTypeTp::String)
        .column(8, FieldTypeTp::String)
        .fts_index(index_def)
        .schema();
    let schema_file = new_schema_file(&[&schema]);

    let columnar_file = new_columnar(1, 0)
        .table::<IntPk>(&schema, |row| {
            row(0, 1000, false, |datum| {
                datum("hello world");
                datum("hello world");
            });
            row(1, 990, false, |datum| {
                datum("greetings earth");
                datum("greetings earth");
            });
        })
        .finish_as_file();

    let index_pairs = &[(table_id, 3)];
    let index_layouts = ordinal_index_layouts(&[((table_id, 3), index_fields)], &schema.columns);
    let opts = ColumnarToFtsL0Opts {
        columnar_files: &[columnar_file],
        schema_file,
        index_pairs,
        index_layouts: &index_layouts,
        encryption_key: None,
        snap_version: SnapVersion::zero(),
        l0_file_max_size: 0,
    };

    let outputs = columnar_to_fts_l0(opts).await?;
    let packed_file = PackedFile::from_buffer(&outputs[0].data)?;
    let lp = packed_file
        .cached_get_lp(&lp_key(table_id, 3))
        .await?
        .unwrap();
    let lp = lp.as_int_lp().unwrap();
    let reader = lp.cached_read_index()?;
    let tracked_index_layout = index_layouts.get(&(table_id, 3)).unwrap();

    let query = make_unscored_query(&PlainFtsQueryInfo {
        query: "hello".to_string(),
        column_id: 7,
        index_id: 3,
        tokenizer: "STANDARD_V1".to_string(),
        ..Default::default()
    });
    let query = clara_fts::Query::new_with_index_layout(
        query.info().clone(),
        &tracked_index_layout.index_fields,
        &tracked_index_layout.index_layout,
    )?;
    let mut results = Vec::new();
    reader.search(&query, &mut results)?;
    assert_eq!(results.len(), 1);

    let query = make_unscored_query(&PlainFtsQueryInfo {
        query: "hello world".to_string(),
        column_id: 8,
        index_id: 3,
        tokenizer: "EXACT_VALUE_V1".to_string(),
        ..Default::default()
    });
    let query = clara_fts::Query::new_with_index_layout(
        query.info().clone(),
        &tracked_index_layout.index_fields,
        &tracked_index_layout.index_layout,
    )?;
    results.clear();
    reader.search(&query, &mut results)?;
    assert_eq!(results.len(), 1);

    let query = make_unscored_query(&PlainFtsQueryInfo {
        query: "hello".to_string(),
        column_id: 8,
        index_id: 3,
        tokenizer: "EXACT_VALUE_V1".to_string(),
        ..Default::default()
    });
    let query = clara_fts::Query::new_with_index_layout(
        query.info().clone(),
        &tracked_index_layout.index_fields,
        &tracked_index_layout.index_layout,
    )?;
    results.clear();
    reader.search(&query, &mut results)?;
    assert_eq!(results.len(), 0);

    Ok(())
}

#[tokio::test]
async fn test_build_stable_fts_l0_with_ngram_field_options() -> Result<()> {
    let table_id = 34;
    let index_id = 5;
    let index_fields = vec![ngram_index_field(7, 3, 3, "WORD", false, true)];
    let mut index_def = FullTextIndexDef {
        index_id,
        ..Default::default()
    };
    index_def.set_index_fields(index_fields.clone().into());

    let schema = SchemaBuilder::<IntPk>::new(table_id)
        .column(7, FieldTypeTp::String)
        .fts_index(index_def)
        .schema();
    let schema_file = new_schema_file(&[&schema]);
    let columnar_file = new_columnar(1, 0)
        .table::<IntPk>(&schema, |row| {
            row(0, 1000, false, |datum| {
                datum("Théllo Théllo");
            });
        })
        .finish_as_file();

    let index_layouts =
        ordinal_index_layouts(&[((table_id, index_id), index_fields)], &schema.columns);
    let outputs = columnar_to_fts_l0(ColumnarToFtsL0Opts {
        columnar_files: &[columnar_file],
        schema_file,
        index_pairs: &[(table_id, index_id)],
        index_layouts: &index_layouts,
        encryption_key: None,
        snap_version: SnapVersion::zero(),
        l0_file_max_size: 0,
    })
    .await?;

    let packed_file = PackedFile::from_buffer(&outputs.into_iter().next().unwrap().data)?;
    let lp = packed_file
        .cached_get_lp(&lp_key(table_id, index_id))
        .await?
        .unwrap();
    let reader = lp.as_int_lp().unwrap().cached_read_index()?;
    let layout = index_layouts.get(&(table_id, index_id)).unwrap();
    let field = layout.index_layout.first_field().field();

    let tantivy_schema = reader.tantivy_index().schema();
    let FieldType::Str(text_options) = tantivy_schema.get_field_entry(field).field_type() else {
        panic!("expected text field");
    };
    assert_eq!(
        text_options.get_indexing_options().unwrap().tokenizer(),
        "NGRAM_V1_FIELD_0"
    );

    let searcher = reader.tantivy_reader().searcher();
    assert_eq!(
        searcher.doc_freq(&tantivy::Term::from_field_text(field, "Thé"))?,
        1
    );
    assert_eq!(
        searcher.doc_freq(&tantivy::Term::from_field_text(field, "thé"))?,
        0
    );
    let inverted_index = searcher.segment_readers()[0].inverted_index(field)?;
    let mut postings = inverted_index
        .read_postings(
            &tantivy::Term::from_field_text(field, "Thé"),
            tantivy::schema::IndexRecordOption::WithFreqsAndPositions,
        )?
        .unwrap();
    assert_eq!(postings.doc(), 0);
    assert_eq!(postings.term_freq(), 2);
    let mut positions = Vec::new();
    postings.positions(&mut positions);
    assert_eq!(positions, [0, 7]);

    Ok(())
}

#[tokio::test]
async fn test_build_stable_fts_l0_with_char_lowercase_and_bm25() -> Result<()> {
    let table_id = 35;
    let index_id = 6;
    let index_fields = vec![ngram_index_field(7, 3, 3, "CHAR", true, true)];
    let mut index_def = FullTextIndexDef {
        index_id,
        ..Default::default()
    };
    index_def.set_index_fields(index_fields.clone().into());

    let schema = SchemaBuilder::<IntPk>::new(table_id)
        .column(7, FieldTypeTp::String)
        .fts_index(index_def)
        .schema();
    let schema_file = new_schema_file(&[&schema]);
    let columnar_file = new_columnar(1, 0)
        .table::<IntPk>(&schema, |row| {
            row(0, 1000, false, |datum| {
                datum("Théllo World");
            });
        })
        .finish_as_file();

    let index_layouts =
        ordinal_index_layouts(&[((table_id, index_id), index_fields)], &schema.columns);
    let outputs = columnar_to_fts_l0(ColumnarToFtsL0Opts {
        columnar_files: &[columnar_file],
        schema_file,
        index_pairs: &[(table_id, index_id)],
        index_layouts: &index_layouts,
        encryption_key: None,
        snap_version: SnapVersion::zero(),
        l0_file_max_size: 0,
    })
    .await?;

    let packed_file = PackedFile::from_buffer(&outputs.into_iter().next().unwrap().data)?;
    let lp = packed_file
        .cached_get_lp(&lp_key(table_id, index_id))
        .await?
        .unwrap();
    let reader = lp.as_int_lp().unwrap().cached_read_index()?;
    let layout = index_layouts.get(&(table_id, index_id)).unwrap();
    let field = layout.index_layout.first_field().field();

    let tantivy_schema = reader.tantivy_index().schema();
    let FieldType::Str(text_options) = tantivy_schema.get_field_entry(field).field_type() else {
        panic!("expected text field");
    };
    let indexing_options = text_options.get_indexing_options().unwrap();
    assert_eq!(indexing_options.tokenizer(), "NGRAM_V1_FIELD_0");
    assert!(indexing_options.fieldnorms());
    assert_eq!(
        indexing_options.index_option(),
        tantivy::schema::IndexRecordOption::WithFreqsAndPositions
    );

    let searcher = reader.tantivy_reader().searcher();
    assert_eq!(
        searcher.doc_freq(&tantivy::Term::from_field_text(field, "thé"))?,
        1
    );
    assert_eq!(
        searcher.doc_freq(&tantivy::Term::from_field_text(field, "Thé"))?,
        0
    );
    let inverted_index = searcher.segment_readers()[0].inverted_index(field)?;
    let mut postings = inverted_index
        .read_postings(
            &tantivy::Term::from_field_text(field, "o w"),
            tantivy::schema::IndexRecordOption::WithFreqsAndPositions,
        )?
        .unwrap();
    assert_eq!(postings.doc(), 0);
    assert_eq!(postings.term_freq(), 1);
    let mut positions = Vec::new();
    postings.positions(&mut positions);
    assert_eq!(positions, [5]);

    Ok(())
}

#[tokio::test]
async fn test_columnar_l0_to_fts_exact_and_path_prefix_filters() -> Result<()> {
    let table_id = 33;
    let index_id = 4;
    let index_fields = vec![
        tracked_field_with_option(7, FullTextAnalyzerType::FtsAnalyzerStandardV1, true, None),
        tracked_field_with_option(
            8,
            FullTextAnalyzerType::FtsAnalyzerExactValueV1,
            false,
            None,
        ),
        tracked_field_with_option(
            8,
            FullTextAnalyzerType::FtsAnalyzerPathHierarchyV1,
            false,
            Some("/"),
        ),
        tracked_field_with_option(
            9,
            FullTextAnalyzerType::FtsAnalyzerExactValueV1,
            false,
            None,
        ),
    ];
    let mut index_def = FullTextIndexDef {
        index_id,
        ..Default::default()
    };
    index_def.set_index_fields(index_fields.clone().into());

    let schema = SchemaBuilder::<IntPk>::new(table_id)
        .column(7, FieldTypeTp::String)
        .column(8, FieldTypeTp::String)
        .column(9, FieldTypeTp::LongLong)
        .fts_index(index_def)
        .schema();
    let schema_file = new_schema_file(&[&schema]);
    let columnar_file = new_columnar(1, 0)
        .table::<IntPk>(&schema, |row| {
            row(0, 1000, false, |datum| {
                datum("target");
                datum("/src/pkg/main.go");
                datum(7i64);
            });
            row(1, 990, false, |datum| {
                datum("target");
                datum("/test/main.go");
                datum(-9i64);
            });
            row(2, 980, false, |datum| {
                datum("other");
                datum("/src/lib.rs");
                datum(7i64);
            });
            row(3, 970, false, |datum| {
                datum("target");
                datum("");
                datum(11i64);
            });
        })
        .finish_as_file();

    let index_layouts = ordinal_index_layouts(
        &[((table_id, index_id), index_fields.clone())],
        &schema.columns,
    );
    let outputs = columnar_to_fts_l0(ColumnarToFtsL0Opts {
        columnar_files: &[columnar_file],
        schema_file,
        index_pairs: &[(table_id, index_id)],
        index_layouts: &index_layouts,
        encryption_key: None,
        snap_version: SnapVersion::zero(),
        l0_file_max_size: 0,
    })
    .await?;

    let packed_file = PackedFile::from_buffer(&outputs.into_iter().next().unwrap().data)?;
    let lp = packed_file
        .cached_get_lp(&lp_key(table_id, index_id))
        .await?
        .unwrap();
    let reader = lp.as_int_lp().unwrap().cached_read_index()?;
    let searcher = reader.tantivy_reader().searcher();
    let layout = index_layouts.get(&(table_id, index_id)).unwrap();

    let exact_field = layout.index_layout.fields()[1].field();
    assert_eq!(
        searcher.doc_freq(&tantivy::Term::from_field_text(
            exact_field,
            "/src/pkg/main.go"
        ))?,
        1
    );
    assert_eq!(
        searcher.doc_freq(&tantivy::Term::from_field_text(exact_field, "/src/"))?,
        0
    );
    let query = make_exact_match_expr_query("", 8, index_id, &index_fields, &layout.index_layout)?;
    let mut results = Vec::new();
    reader.search(&query, &mut results)?;
    assert_eq!(results.len(), 1);
    assert_eq!(results[0].doc_id, 3);

    let prefix_field = layout.index_layout.fields()[2].field();
    assert_eq!(
        searcher.doc_freq(&tantivy::Term::from_field_text(prefix_field, "/src"))?,
        2
    );
    assert_eq!(
        searcher.doc_freq(&tantivy::Term::from_field_text(prefix_field, "/src/pkg"))?,
        1
    );
    assert_eq!(
        searcher.doc_freq(&tantivy::Term::from_field_text(
            prefix_field,
            "/src/pkg/main.go"
        ))?,
        1
    );
    assert_eq!(
        searcher.doc_freq(&tantivy::Term::from_field_text(prefix_field, "/src/"))?,
        0
    );

    let integer_field = layout.index_layout.fields()[3].field();
    assert!(matches!(
        reader
            .tantivy_index()
            .schema()
            .get_field_entry(integer_field)
            .field_type(),
        FieldType::I64(_)
    ));
    assert_eq!(
        searcher.doc_freq(&tantivy::Term::from_field_i64(integer_field, 7))?,
        2
    );
    assert_eq!(
        searcher.doc_freq(&tantivy::Term::from_field_i64(integer_field, -9))?,
        1
    );

    Ok(())
}

#[tokio::test]
async fn test_columnar_l0_to_fts_invalid_utf8_should_not_fail() -> Result<()> {
    let table_id = 30;

    let schema = SchemaBuilder::<IntPk>::new(table_id)
        .column(7, FieldTypeTp::String)
        .fts_index(FullTextIndexDef {
            index_id: 1,
            col_id: 7,
            parser_type: "STANDARD_V1".to_string(),
            ..Default::default()
        })
        .schema();
    let schema_file = new_schema_file(&[&schema]);

    // 0x80 is not a valid UTF-8 byte by itself.
    let invalid_utf8: &[u8] = &[b'a', b'b', b'c', 0x80, b'd', b'e', b'f'];

    let columnar_file = new_columnar(1, 0)
        .table::<IntPk>(&schema, |row| {
            row(0, 1000, false, |datum| {
                datum(invalid_utf8);
            });
            row(1, 1000, false, |datum| {
                datum("hello world");
            });
        })
        .finish_as_file();

    let index_pairs = &[(table_id, 1)];
    let index_layouts = legacy_index_layouts(index_pairs);
    let opts = ColumnarToFtsL0Opts {
        columnar_files: &[columnar_file],
        schema_file,
        index_pairs,
        index_layouts: &index_layouts,
        encryption_key: None,
        snap_version: SnapVersion::zero(),
        l0_file_max_size: 0,
    };

    let outputs = columnar_to_fts_l0(opts).await?;
    assert!(!outputs.is_empty(), "Should generate FTS data");

    let packed_file = PackedFile::from_buffer(&outputs[0].data)?;
    let lp = packed_file
        .cached_get_lp(&lp_key(table_id, 1))
        .await?
        .expect("lp must exist");
    let lp = lp.as_int_lp().unwrap();

    let reader = lp.cached_read_index()?;
    let query = make_unscored_query(&PlainFtsQueryInfo {
        query: "hello".to_string(),
        ..Default::default()
    });
    let mut results = Vec::new();
    reader.search(&query, &mut results)?;
    assert!(!results.is_empty(), "Should still index valid UTF-8 rows");

    Ok(())
}

#[tokio::test]
async fn test_columnar_l0_to_fts_not_null_text_column() -> Result<()> {
    let table_id = 31;

    let schema = SchemaBuilder::<IntPk>::new(table_id)
        .column_not_null(7, FieldTypeTp::String)
        .fts_index(FullTextIndexDef {
            index_id: 1,
            col_id: 7,
            parser_type: "STANDARD_V1".to_string(),
            ..Default::default()
        })
        .schema();
    let schema_file = new_schema_file(&[&schema]);

    let columnar_file = new_columnar(2, 0)
        .table::<IntPk>(&schema, |row| {
            row(0, 1000, false, |datum| {
                datum("abc not null");
            });
            row(1, 950, false, |datum| {
                datum("plain content");
            });
        })
        .finish_as_file();

    let index_pairs = &[(table_id, 1)];
    let index_layouts = legacy_index_layouts(index_pairs);
    let opts = ColumnarToFtsL0Opts {
        columnar_files: &[columnar_file],
        schema_file,
        index_pairs,
        index_layouts: &index_layouts,
        encryption_key: None,
        snap_version: SnapVersion::zero(),
        l0_file_max_size: 0,
    };
    let file_data = columnar_to_fts_l0(opts).await?;
    assert!(!file_data.is_empty());

    Ok(())
}

#[tokio::test]
async fn test_edge_cases() -> Result<()> {
    let table_id = 100;

    let schema = SchemaBuilder::<IntPk>::new(table_id)
        .column(7, FieldTypeTp::String)
        .fts_index(FullTextIndexDef {
            index_id: 1,
            col_id: 7,
            parser_type: "STANDARD_V1".to_string(),
            ..Default::default()
        })
        .schema();
    let schema_file = new_schema_file(&[&schema]);

    let columnar_file = new_columnar(11, 0)
        .table::<IntPk>(&schema, |row| {
            row(10, 500, false, |datum| {
                datum("edge alpha");
            });
            row(11, 600, false, |datum| {
                datum("edge beta");
            });
        })
        .finish_as_file();

    // Edge Case 1: Empty index pairs list (no indexes to process)
    let empty_index_pairs = &[];
    let empty_index_layouts = legacy_index_layouts(empty_index_pairs);
    let opts_empty = ColumnarToFtsL0Opts {
        columnar_files: std::slice::from_ref(&columnar_file),
        schema_file: schema_file.clone(),
        index_pairs: empty_index_pairs, // Empty index pairs list
        index_layouts: &empty_index_layouts,
        encryption_key: None,
        snap_version: SnapVersion::zero(),
        l0_file_max_size: 0,
    };
    let file_data_empty = columnar_to_fts_l0(opts_empty).await?;
    assert!(
        file_data_empty.is_empty(),
        "Empty index pairs list should produce no FTS file"
    );

    // Edge Case 2: Index pairs that don't exist in schema
    let nonexistent_index_pairs = &[(999, 1)];
    let nonexistent_index_layouts = legacy_index_layouts(nonexistent_index_pairs);
    let opts_nonexistent = ColumnarToFtsL0Opts {
        columnar_files: &[columnar_file],
        schema_file,
        index_pairs: nonexistent_index_pairs, // Non-existent table ID
        index_layouts: &nonexistent_index_layouts,
        encryption_key: None,
        snap_version: SnapVersion::zero(),
        l0_file_max_size: 0,
    };
    let file_data_nonexistent = columnar_to_fts_l0(opts_nonexistent).await?;
    assert!(
        file_data_nonexistent.is_empty(),
        "Non-existent table should produce no FTS file"
    );

    Ok(())
}

#[tokio::test]
async fn test_multiple_columnar_tables() -> Result<()> {
    // Create two different tables with FTS indexes
    let table_id_1 = 200;
    let table_id_2 = 300;

    let schema_1 = SchemaBuilder::<IntPk>::new(table_id_1)
        .column(7, FieldTypeTp::String)
        .fts_index(FullTextIndexDef {
            index_id: 1,
            col_id: 7,
            parser_type: "STANDARD_V1".to_string(),
            ..Default::default()
        })
        .schema();
    let schema_2 = SchemaBuilder::<IntPk>::new(table_id_2)
        .column(7, FieldTypeTp::String)
        .fts_index(FullTextIndexDef {
            index_id: 2,
            col_id: 7,
            parser_type: "STANDARD_V1".to_string(),
            ..Default::default()
        })
        .schema();
    let schema_file = new_schema_file(&[&schema_1, &schema_2]);

    // Create columnar files for both tables
    let columnar_file_1 = new_columnar(21, 0)
        .table::<IntPk>(&schema_1, |row| {
            row(1, 700, false, |datum| {
                datum("table200 alpha");
            });
            row(2, 720, false, |datum| {
                datum("table200 abc beta");
            });
        })
        .finish_as_file();
    let columnar_file_2 = new_columnar(22, 0)
        .table::<IntPk>(&schema_2, |row| {
            row(1, 800, false, |datum| {
                datum("table300 abc delta");
            });
            row(2, 830, false, |datum| {
                datum("table300 omega");
            });
        })
        .finish_as_file();

    // Convert both tables to FTS L0
    let index_pairs = &[(table_id_1, 1), (table_id_2, 2)];
    let index_layouts = legacy_index_layouts(index_pairs);
    let opts = ColumnarToFtsL0Opts {
        columnar_files: &[columnar_file_1, columnar_file_2],
        schema_file,
        index_pairs, // Specify both (table, index) pairs
        index_layouts: &index_layouts,
        encryption_key: None,
        snap_version: SnapVersion::zero(),
        l0_file_max_size: 0,
    };

    let file_data = columnar_to_fts_l0(opts).await?;
    assert!(
        !file_data.is_empty(),
        "Should generate FTS data for multiple tables"
    );

    let output = file_data.into_iter().next().unwrap();
    let packed_file = PackedFile::from_buffer(&output.data)?;

    // Test both logical partitions exist
    let lp_1 = packed_file.cached_get_lp(&lp_key(200, 1)).await?;
    let lp_2 = packed_file.cached_get_lp(&lp_key(300, 2)).await?;

    assert!(
        lp_1.is_some(),
        "Should find logical partition for table 200"
    );
    assert!(
        lp_2.is_some(),
        "Should find logical partition for table 300"
    );

    // Test cross-table isolation
    assert!(
        packed_file.cached_get_lp(&lp_key(200, 2)).await?.is_none(),
        "Should not find table 200 with index 2"
    );
    assert!(
        packed_file.cached_get_lp(&lp_key(300, 1)).await?.is_none(),
        "Should not find table 300 with index 1"
    );

    // Test documents exist in both partitions
    // Extract Int variants
    let lp_1 = lp_1.unwrap();
    let lp_1 = lp_1.as_int_lp().unwrap();
    let lp_2 = lp_2.unwrap();
    let lp_2 = lp_2.as_int_lp().unwrap();
    assert!(
        lp_1.props().get_n_pk() > 0,
        "Table 200 should have documents"
    );
    assert!(
        lp_2.props().get_n_pk() > 0,
        "Table 300 should have documents"
    );

    Ok(())
}

#[tokio::test]
async fn test_multiple_indexes_per_table() -> Result<()> {
    let table_id = 400;

    // Create multiple FTS indexes on different columns
    let schema = SchemaBuilder::<IntPk>::new(table_id)
        .column(7, FieldTypeTp::String)
        .fts_index(FullTextIndexDef {
            index_id: 1,
            col_id: 7,
            parser_type: "STANDARD_V1".to_string(),
            ..Default::default()
        })
        .fts_index(FullTextIndexDef {
            index_id: 3,
            col_id: 7, // Same column, different index
            parser_type: "STANDARD_V1".to_string(),
            ..Default::default()
        })
        .schema();
    let schema_file = new_schema_file(&[&schema]);

    let columnar_file = new_columnar(31, 0)
        .table::<IntPk>(&schema, |row| {
            row(1, 600, false, |datum| {
                datum("abc multi index one");
            });
            row(2, 620, false, |datum| {
                datum("plain multi index");
            });
            row(3, 640, false, |datum| {
                datum("abc multi index two");
            });
        })
        .finish_as_file();

    // Convert to FTS L0
    let index_pairs = &[(table_id, 1), (table_id, 3)];
    let index_layouts = legacy_index_layouts(index_pairs);
    let opts = ColumnarToFtsL0Opts {
        columnar_files: &[columnar_file],
        schema_file,
        index_pairs, // Specify both index IDs for the same table
        index_layouts: &index_layouts,
        encryption_key: None,
        snap_version: SnapVersion::zero(),
        l0_file_max_size: 0,
    };

    let file_data = columnar_to_fts_l0(opts).await?;
    assert!(
        !file_data.is_empty(),
        "Should generate FTS data for multiple indexes"
    );

    let output = file_data.into_iter().next().unwrap();
    let packed_file = PackedFile::from_buffer(&output.data)?;

    // Test both logical partitions exist for the same table but different indexes
    let lp_index_1 = packed_file.cached_get_lp(&lp_key(400, 1)).await?;
    let lp_index_3 = packed_file.cached_get_lp(&lp_key(400, 3)).await?;

    assert!(
        lp_index_1.is_some(),
        "Should find logical partition for index 1"
    );
    assert!(
        lp_index_3.is_some(),
        "Should find logical partition for index 3"
    );

    // Test non-existent index
    assert!(
        packed_file.cached_get_lp(&lp_key(400, 2)).await?.is_none(),
        "Should not find non-existent index 2"
    );

    // Both indexes should have the same documents (since they index the same
    // column)
    // Extract Int variants
    let lp_index_1 = lp_index_1.unwrap();
    let lp_index_1 = lp_index_1.as_int_lp().unwrap();
    let lp_index_3 = lp_index_3.unwrap();
    let lp_index_3 = lp_index_3.as_int_lp().unwrap();
    assert!(
        lp_index_1.props().get_n_pk() > 0,
        "Index 1 should have documents"
    );
    assert!(
        lp_index_3.props().get_n_pk() > 0,
        "Index 3 should have documents"
    );
    assert_eq!(
        lp_index_1.props().get_n_pk(),
        lp_index_3.props().get_n_pk(),
        "Both indexes should have the same number of documents"
    );

    // Test search isolation between indexes
    let reader_1 = lp_index_1.cached_read_index()?;
    let reader_3 = lp_index_3.cached_read_index()?;

    let mut results_1 = Vec::new();
    let mut results_3 = Vec::new();

    let query = make_unscored_query(&PlainFtsQueryInfo {
        query: "abc".to_string(),
        ..Default::default()
    });

    reader_1.search(&query, &mut results_1)?;
    reader_3.search(&query, &mut results_3)?;

    // Both should find results since they index the same data
    assert!(!results_1.is_empty(), "Index 1 should find search results");
    assert!(!results_3.is_empty(), "Index 3 should find search results");

    Ok(())
}

#[tokio::test]
async fn test_multiple_columnar_files() -> Result<()> {
    let table_id = 500;

    let schema = SchemaBuilder::<IntPk>::new(table_id)
        .column(7, FieldTypeTp::String)
        .fts_index(FullTextIndexDef {
            index_id: 1,
            col_id: 7,
            parser_type: "STANDARD_V1".to_string(),
            ..Default::default()
        })
        .schema();
    let schema_file = new_schema_file(&[&schema]);

    // Create multiple columnar files with explicit overlapping versions
    let columnar_files = vec![
        new_columnar(41, 0)
            .table::<IntPk>(&schema, |row| {
                row(0, 1000, false, |datum| {
                    datum("abc file1 handle0 v1000");
                });
                row(1, 1000, false, |datum| {
                    datum("abc file1 handle1 v1000");
                });
            })
            .finish_as_file(),
        new_columnar(42, 0)
            .table::<IntPk>(&schema, |row| {
                row(2, 2000, false, |datum| {
                    datum("abc file2 handle2 v2000");
                });
                row(3, 2000, false, |datum| {
                    datum("abc file2 handle3 v2000");
                });
            })
            .finish_as_file(),
        new_columnar(43, 0)
            .table::<IntPk>(&schema, |row| {
                row(1, 1500, false, |datum| {
                    datum("abc file3 handle1 v1500");
                });
                row(2, 1500, false, |datum| {
                    datum("abc file3 handle2 v1500");
                });
            })
            .finish_as_file(),
    ];

    // Convert to FTS L0
    let index_pairs = &[(table_id, 1)];
    let index_layouts = legacy_index_layouts(index_pairs);
    let opts = ColumnarToFtsL0Opts {
        columnar_files: &columnar_files,
        schema_file,
        index_pairs, // Specify the explicit index_id
        index_layouts: &index_layouts,
        encryption_key: None,
        snap_version: SnapVersion::zero(),
        l0_file_max_size: 0,
    };

    let file_data = columnar_to_fts_l0(opts).await?;
    assert!(
        !file_data.is_empty(),
        "Should generate FTS data from multiple files"
    );

    let output = file_data.into_iter().next().unwrap();
    let packed_file = PackedFile::from_buffer(&output.data)?;

    let lp = packed_file.cached_get_lp(&lp_key(500, 1)).await?;
    assert!(lp.is_some(), "Should find logical partition");

    // Extract Int variant
    let lp = lp.unwrap();
    let lp = lp.as_int_lp().unwrap();

    // Should have documents from all files, with proper MVCC handling
    // The exact number depends on how the merge handles overlapping handles with
    // different versions
    assert!(
        lp.props().get_n_pk() > 0,
        "Should have documents from merged files"
    );

    // Test MVCC with different versions
    let reader = lp.cached_read_index()?;
    let mut results = Vec::new();

    let query = make_unscored_query(&PlainFtsQueryInfo {
        query: "abc".to_string(),
        ..Default::default()
    });

    reader.search(&query, &mut results)?;
    assert!(!results.is_empty(), "Should find documents in merged index");

    // Test MVCC version checking with overlapping versions
    // MVCC keeps ALL versions for each handle, not just the latest:
    // - Handle 0: version 1000 (from file 1)
    // - Handle 1: versions 1000 (file 1) AND 1500 (file 3)
    // - Handle 2: versions 1500 (file 3) AND 2000 (file 2)
    // - Handle 3: version 2000 (from file 2)

    // Test Handle 0: should have version 1000 only
    assert!(
        lp.has_newer_version(&IntPk::encode(0), 900, 1100).await?,
        "Handle 0 should have version 1000 (from file 1)"
    );
    assert!(
        !lp.has_newer_version(&IntPk::encode(0), 1000, 1100).await?,
        "Handle 0 should not have version > 1000"
    );
    assert!(
        !lp.has_newer_version(&IntPk::encode(0), 1400, 1600).await?,
        "Handle 0 should not have version 1500 (not in file 3)"
    );

    // Test Handle 1: should have BOTH versions 1000 and 1500
    assert!(
        lp.has_newer_version(&IntPk::encode(1), 900, 1100).await?,
        "Handle 1 should have version 1000 (from file 1)"
    );
    assert!(
        lp.has_newer_version(&IntPk::encode(1), 1400, 1600).await?,
        "Handle 1 should also have version 1500 (from file 3)"
    );
    assert!(
        !lp.has_newer_version(&IntPk::encode(1), 1500, 1600).await?,
        "Handle 1 should not have version > 1500"
    );
    assert!(
        !lp.has_newer_version(&IntPk::encode(1), 1900, 2100).await?,
        "Handle 1 should not have version 2000 (not in file 2)"
    );

    // Test Handle 2: should have BOTH versions 1500 and 2000
    assert!(
        lp.has_newer_version(&IntPk::encode(2), 1400, 1600).await?,
        "Handle 2 should have version 1500 (from file 3)"
    );
    assert!(
        lp.has_newer_version(&IntPk::encode(2), 1900, 2100).await?,
        "Handle 2 should also have version 2000 (from file 2)"
    );
    assert!(
        !lp.has_newer_version(&IntPk::encode(2), 2000, 2100).await?,
        "Handle 2 should not have version > 2000"
    );
    assert!(
        !lp.has_newer_version(&IntPk::encode(2), 900, 1100).await?,
        "Handle 2 should not have version 1000 (not in file 1)"
    );

    // Test Handle 3: should have version 2000 only
    assert!(
        lp.has_newer_version(&IntPk::encode(3), 1900, 2100).await?,
        "Handle 3 should have version 2000 (from file 2)"
    );
    assert!(
        !lp.has_newer_version(&IntPk::encode(3), 2000, 2100).await?,
        "Handle 3 should not have version > 2000"
    );
    assert!(
        !lp.has_newer_version(&IntPk::encode(3), 1400, 1600).await?,
        "Handle 3 should not have version 1500 (not in file 3)"
    );

    // Test non-existent handles
    assert!(
        !lp.has_newer_version(&IntPk::encode(4), 0, u64::MAX).await?,
        "Handle 4 should not exist"
    );
    assert!(
        !lp.has_newer_version(&IntPk::encode(999), 0, u64::MAX)
            .await?,
        "Handle 999 should not exist"
    );

    Ok(())
}

#[tokio::test]
async fn test_index_id_filtering() -> Result<()> {
    let table_id = 600;

    // Create multiple FTS indexes on the same column but with different index_ids
    // Create schema with ALL three indexes
    let schema = SchemaBuilder::<IntPk>::new(table_id)
        .column(7, FieldTypeTp::String)
        .fts_index(FullTextIndexDef {
            index_id: 1,
            col_id: 7,
            parser_type: "STANDARD_V1".to_string(),
            ..Default::default()
        })
        .fts_index(FullTextIndexDef {
            index_id: 2,
            col_id: 7, // Same column, different index_id
            parser_type: "STANDARD_V1".to_string(),
            ..Default::default()
        })
        .fts_index(FullTextIndexDef {
            index_id: 3,
            col_id: 7, // Same column, different index_id
            parser_type: "STANDARD_V1".to_string(),
            ..Default::default()
        })
        .schema();
    let schema_file = new_schema_file(&[&schema]);

    let columnar_file = new_columnar(51, 0)
        .table::<IntPk>(&schema, |row| {
            row(1, 1000, false, |datum| {
                datum("abc index filter one");
            });
            row(2, 1010, false, |datum| {
                datum("index filter middle");
            });
            row(3, 1020, false, |datum| {
                datum("abc index filter two");
            });
        })
        .finish_as_file();

    // KEY TEST: Only specify index_id=2, even though schema has indexes 1, 2, 3
    let index_pairs = &[(table_id, 2)];
    let index_layouts = legacy_index_layouts(index_pairs);
    let opts = ColumnarToFtsL0Opts {
        columnar_files: &[columnar_file],
        schema_file,
        index_pairs, // Only index_id=2, NOT 1 or 3
        index_layouts: &index_layouts,
        encryption_key: None,
        snap_version: SnapVersion::zero(),
        l0_file_max_size: 0,
    };

    let file_data = columnar_to_fts_l0(opts).await?;
    assert!(
        !file_data.is_empty(),
        "Should generate FTS data for index 2"
    );

    let output = file_data.into_iter().next().unwrap();
    let packed_file = PackedFile::from_buffer(&output.data)?;

    // Verify: ONLY index_id=2 should exist, others should NOT
    let lp_index_1 = packed_file.cached_get_lp(&lp_key(600, 1)).await?;
    let lp_index_2 = packed_file.cached_get_lp(&lp_key(600, 2)).await?;
    let lp_index_3 = packed_file.cached_get_lp(&lp_key(600, 3)).await?;

    assert!(
        lp_index_1.is_none(),
        "Index 1 should NOT be built (not requested)"
    );
    assert!(
        lp_index_2.is_some(),
        "Index 2 should be built (explicitly requested)"
    );
    assert!(
        lp_index_3.is_none(),
        "Index 3 should NOT be built (not requested)"
    );

    // Verify the built index actually works
    // Extract Int variant
    let lp_2 = lp_index_2.unwrap();
    let lp_2 = lp_2.as_int_lp().unwrap();
    assert!(lp_2.props().get_n_pk() > 0, "Index 2 should have documents");

    // Verify search works on index 2
    let reader = lp_2.cached_read_index()?;
    let mut results = Vec::new();

    let query = make_unscored_query(&PlainFtsQueryInfo {
        query: "abc".to_string(),
        ..Default::default()
    });

    reader.search(&query, &mut results)?;
    assert!(!results.is_empty(), "Search should find results in index 2");

    Ok(())
}

#[tokio::test]
async fn test_columnar_to_fts_splits_l0_files() -> Result<()> {
    let table_id = 700;

    let schema = SchemaBuilder::<IntPk>::new(table_id)
        .column(2, FieldTypeTp::String)
        .fts_index(FullTextIndexDef {
            index_id: 1,
            col_id: 2,
            parser_type: "STANDARD_V1".to_string(),
            ..Default::default()
        })
        .fts_index(FullTextIndexDef {
            index_id: 2,
            col_id: 2,
            parser_type: "STANDARD_V1".to_string(),
            ..Default::default()
        })
        .fts_index(FullTextIndexDef {
            index_id: 3,
            col_id: 2,
            parser_type: "STANDARD_V1".to_string(),
            ..Default::default()
        })
        .schema();
    let schema_file = new_schema_file(&[&schema]);

    let columnar_file = new_columnar(1, 0)
        .table::<IntPk>(&schema, |row| {
            row(0, 1000, false, |datum| {
                datum("abc first entry");
            });
            row(1, 950, false, |datum| {
                datum("plain content");
            });
            row(2, 1020, false, |datum| {
                datum("another abc value");
            });
            row(3, 980, false, |datum| {
                datum("quiet note");
            });
        })
        .finish_as_file();

    let index_pairs = &[(table_id, 1), (table_id, 2), (table_id, 3)];
    let index_layouts = legacy_index_layouts(index_pairs);
    let opts = ColumnarToFtsL0Opts {
        columnar_files: &[columnar_file],
        schema_file,
        index_pairs,
        index_layouts: &index_layouts,
        encryption_key: None,
        snap_version: SnapVersion::zero(),
        l0_file_max_size: 1, // Force split at every LP boundary.
    };

    let outputs = columnar_to_fts_l0(opts).await?;
    assert!(outputs.len() > 1, "Should split into multiple output files");

    let mut built_indexes = HashSet::new();
    for out in outputs {
        let packed_file = PackedFile::from_buffer(&out.data)?;
        for index_id in 1..=3 {
            if packed_file
                .cached_get_lp(&lp_key(table_id, index_id))
                .await?
                .is_some()
            {
                built_indexes.insert(index_id);
            }
        }
    }

    assert_eq!(
        built_indexes.len(),
        3,
        "All requested indexes should be built"
    );
    Ok(())
}