// Copyright 2026 TiKV Project Authors. Licensed under Apache-2.0.

use tantivy::schema::{IndexRecordOption, OwnedValue};

use super::*;

fn exact_match_leaf(
    score_mode: &str,
    query: impl Into<serde_json::Value>,
    fields: Vec<serde_json::Value>,
) -> tipb::Expr {
    let mut expr = tipb::Expr::new();
    expr.set_tp(tipb::ExprType::ScalarFunc);
    expr.set_sig(tipb::ScalarFuncSig::FtsMatchWord);
    let query = query.into();
    let metadata = if score_mode == "inherit" {
        serde_json::json!({
            "version": FTS_MATCH_EXPR_METADATA_VERSION,
            "query": query,
            "fields": fields,
        })
    } else {
        serde_json::json!({
            "version": FTS_MATCH_EXPR_METADATA_VERSION,
            "score_mode": score_mode,
            "query": query,
            "fields": fields,
        })
    };
    expr.set_val(serde_json::to_vec(&metadata).unwrap());
    expr
}

fn logical_or(left: tipb::Expr, right: tipb::Expr) -> tipb::Expr {
    let mut expr = tipb::Expr::new();
    expr.set_tp(tipb::ExprType::ScalarFunc);
    expr.set_sig(tipb::ScalarFuncSig::LogicalOr);
    expr.set_val(
        serde_json::to_vec(&serde_json::json!({
            "version": FTS_MATCH_EXPR_METADATA_VERSION,
        }))
        .unwrap(),
    );
    expr.mut_children().push(left);
    expr.mut_children().push(right);
    expr
}

fn path_index_field(col_id: i64, delimiter: &str, enable_bm25: bool) -> FullTextIndexFieldDef {
    let mut field = FullTextIndexFieldDef::new();
    field.set_col_id(col_id);
    let option = field.mut_index_field_option();
    option.set_analyzer_type(FullTextAnalyzerType::FtsAnalyzerPathHierarchyV1);
    option.set_enable_bm25(enable_bm25);
    option
        .mut_path_hierarchy()
        .set_delimiter(delimiter.to_string());
    field
}

fn exact_index_field(col_id: i64, enable_bm25: bool) -> FullTextIndexFieldDef {
    let mut field = FullTextIndexFieldDef::new();
    field.set_col_id(col_id);
    let mut option = FullTextIndexFieldOption::new();
    option.set_analyzer_type(FullTextAnalyzerType::FtsAnalyzerExactValueV1);
    option.set_enable_bm25(enable_bm25);
    field.set_index_field_option(option);
    field
}

#[test]
fn test_match_expr_resolves_same_column_different_index_fields() -> Result<()> {
    let layout = crate::IndexLayout::new(vec![
        crate::SingleFieldLayout::new_with_field_and_bm25(Field::from_field_id(0), "0", false),
        crate::SingleFieldLayout::new_with_field_and_bm25(Field::from_field_id(1), "1", false),
    ]);
    let index_fields = vec![
        exact_index_field(202, false),
        path_index_field(202, "/", false),
    ];
    // The same character column carries two text fields with different analyzers.
    let mut writer =
        crate::TantivyIndexWriter::new_in_memory_with_index_fields(&index_fields, &layout)?;
    for path in ["/src/pkg/main.go", "/tmp/file.txt"] {
        writer.add_document_values([
            (Field::from_field_id(0), OwnedValue::Str(path.to_owned())),
            (Field::from_field_id(1), OwnedValue::Str(path.to_owned())),
        ])?;
    }
    let physical_layout = writer.physical_schema().index_layout().clone();
    assert_eq!(
        physical_layout.fields()[0].field_type(),
        crate::PhysicalFieldType::Text
    );
    let reader = crate::IndexReader::from_tantivy_index(writer.finalize()?)?;

    let mut info = PlainFtsQueryInfo::default().as_query_info();
    info.set_query_type(tipb::FtsQueryType::FtsQueryTypeNoScore);
    info.mut_match_expr().push(match_leaf(
        tipb::ScalarFuncSig::FtsMatchWord,
        "inherit",
        "/src/",
        vec![field_ref(202, "PATH_HIERARCHY_V1")],
    ));
    info.mut_match_expr().push(exact_match_leaf(
        "inherit",
        "/src/pkg/main.go",
        vec![field_ref(202, "EXACT_VALUE_V1")],
    ));
    let query = Query::new_with_index_layout(info, &index_fields, &physical_layout)?;

    let mut results = Vec::new();
    reader.search(&query, &mut results)?;
    assert_eq!(doc_ids(&results), vec![0]);
    assert_eq!(query.document_fields().len(), 2);
    assert!(
        query
            .document_fields()
            .iter()
            .all(|field| field.column_id() == 202)
    );
    assert!(
        query
            .document_tokenizers()
            .get("PATH_HIERARCHY_V1_FIELD_1")
            .is_some()
    );

    let mut document = tantivy::TantivyDocument::new();
    document.add_text(Field::from_field_id(0), "/src/pkg/main.go");
    document.add_text(Field::from_field_id(1), "/src/pkg/main.go");
    let mut brute = crate::BruteScoredSearcher::new(&query)?;
    assert_eq!(
        brute.evaluate_document(&document)?,
        DocumentEvaluation::Match(1.0)
    );
    Ok(())
}

#[test]
fn test_scored_exact_text_uses_raw_tokenizer_and_basic_postings() -> Result<()> {
    let layout = crate::IndexLayout::from_field_names(["0"]);
    let index_fields = vec![exact_index_field(101, true)];
    let mut writer =
        crate::TantivyIndexWriter::new_in_memory_with_index_fields(&index_fields, &layout)?;
    writer.add_document("hello world")?;
    writer.add_document("hello")?;
    writer.add_document("hello world")?;
    let physical_layout = writer.physical_schema().index_layout().clone();
    let index = writer.finalize()?;
    let schema = index.schema();
    assert_eq!(
        schema
            .get_field_entry(Field::from_field_id(0))
            .field_type()
            .get_index_record_option(),
        Some(IndexRecordOption::Basic),
    );
    layout.validate_schema(&schema)?;
    let reader = crate::IndexReader::from_tantivy_index(index)?;

    let mut info = PlainFtsQueryInfo::default().as_query_info();
    info.set_query_type(tipb::FtsQueryType::FtsQueryTypeWithScore);
    info.mut_match_expr().push(exact_match_leaf(
        "inherit",
        "hello world",
        vec![field_ref(101, "EXACT_VALUE_V1")],
    ));
    let query = Query::new_with_index_layout(info, &index_fields, &physical_layout)?;
    assert_eq!(query.bm25_terms().len(), 1);

    let mut stats = Bm25Stats::empty(&query);
    reader.accumulate_bm25_stats(&query, &mut stats)?;
    assert!(query.prepare_bm25_once(stats)?);

    let mut results = Vec::new();
    reader.search(&query, &mut results)?;
    assert_eq!(doc_ids(&results), vec![0, 2]);
    assert!(results[0].score > 0.0);
    assert!((results[0].score - results[1].score).abs() < 1e-6);
    Ok(())
}

#[test]
fn test_scored_exact_filter_prepares_filter_only_field_token_totals() -> Result<()> {
    let layout = crate::IndexLayout::new(vec![crate::SingleFieldLayout::new_with_field_and_bm25(
        Field::from_field_id(0),
        "0",
        false,
    )]);
    let index_fields = vec![exact_index_field(101, false)];
    let mut writer =
        crate::TantivyIndexWriter::new_in_memory_with_index_fields(&index_fields, &layout)?;
    writer.add_document("hello world")?;
    writer.add_document("other")?;
    let physical_layout = writer.physical_schema().index_layout().clone();
    let reader = crate::IndexReader::from_tantivy_index(writer.finalize()?)?;

    let mut info = PlainFtsQueryInfo::default().as_query_info();
    info.set_query_type(tipb::FtsQueryType::FtsQueryTypeWithScore);
    info.mut_match_expr().push(exact_match_leaf(
        "filter",
        "hello world",
        vec![field_ref(101, "EXACT_VALUE_V1")],
    ));
    let query = Query::new_with_index_layout(info, &index_fields, &physical_layout)?;
    assert!(query.bm25_terms().is_empty());
    assert_eq!(query.bm25_fields(), &[Field::from_field_id(0)]);

    let mut stats = Bm25Stats::empty(&query);
    reader.accumulate_bm25_stats(&query, &mut stats)?;
    assert!(query.prepare_bm25_once(stats)?);

    let mut results = Vec::new();
    reader.search(&query, &mut results)?;
    assert_eq!(doc_ids(&results), vec![0]);
    assert_eq!(results[0].score, 0.0);
    Ok(())
}

#[test]
fn test_legacy_query_text_rejects_native_exact_field() -> Result<()> {
    let layout = crate::IndexLayout::new(vec![
        crate::SingleFieldLayout::new_with_field_type_and_bm25(
            Field::from_field_id(0),
            "0",
            crate::PhysicalFieldType::Bytes,
            false,
        ),
    ]);
    let info = PlainFtsQueryInfo {
        query: "hello world".to_string(),
        tokenizer: "STANDARD_V1".to_string(),
        ..Default::default()
    }
    .as_query_info();

    assert!(Query::new_with_index_layout(info, &[exact_index_field(101, false)], &layout).is_err());
    Ok(())
}

#[test]
fn test_final_exact_integer_values_use_native_terms() -> Result<()> {
    let layout = crate::IndexLayout::new(vec![
        crate::SingleFieldLayout::new_with_field_type_and_bm25(
            Field::from_field_id(0),
            "0",
            crate::PhysicalFieldType::I64,
            false,
        ),
    ]);
    let index_fields = vec![exact_index_field(101, false)];
    let mut writer =
        crate::TantivyIndexWriter::new_in_memory_with_index_fields(&index_fields, &layout)?;
    writer.add_document_values([(Field::from_field_id(0), OwnedValue::I64(7))])?;
    writer.add_document_values([(Field::from_field_id(0), OwnedValue::I64(-9))])?;
    writer.add_document_values([(Field::from_field_id(0), OwnedValue::I64(7))])?;
    let physical_layout = writer.physical_schema().index_layout().clone();
    let reader = crate::IndexReader::from_tantivy_index(writer.finalize()?)?;

    let mut info = PlainFtsQueryInfo::default().as_query_info();
    info.set_query_type(tipb::FtsQueryType::FtsQueryTypeNoScore);
    info.mut_match_expr().push(logical_or(
        exact_match_leaf("filter", 7i64, vec![field_ref(101, "EXACT_VALUE_V1")]),
        exact_match_leaf("filter", 11i64, vec![field_ref(101, "EXACT_VALUE_V1")]),
    ));
    let query = Query::new_with_index_layout(info, &index_fields, &physical_layout)?;

    let mut results = Vec::new();
    reader.search(&query, &mut results)?;
    assert_eq!(doc_ids(&results), vec![0, 2]);
    Ok(())
}

#[test]
fn test_exact_u64_query_preserves_full_range() -> Result<()> {
    let field = Field::from_field_id(0);
    let layout = crate::IndexLayout::new(vec![
        crate::SingleFieldLayout::new_with_field_type_and_bm25(
            field,
            "0",
            crate::PhysicalFieldType::U64,
            false,
        ),
    ]);
    let index_fields = vec![exact_index_field(101, false)];
    let mut writer =
        crate::TantivyIndexWriter::new_in_memory_with_index_fields(&index_fields, &layout)?;
    writer.add_document_values([(field, OwnedValue::U64(u64::MAX))])?;
    let physical_layout = writer.physical_schema().index_layout().clone();
    let reader = crate::IndexReader::from_tantivy_index(writer.finalize()?)?;

    let mut info = PlainFtsQueryInfo::default().as_query_info();
    info.set_query_type(tipb::FtsQueryType::FtsQueryTypeNoScore);
    info.mut_match_expr().push(exact_match_leaf(
        "filter",
        u64::MAX,
        vec![field_ref(101, "EXACT_VALUE_V1")],
    ));
    let query = Query::new_with_index_layout(info, &index_fields, &physical_layout)?;

    let mut results = Vec::new();
    reader.search(&query, &mut results)?;
    assert_eq!(doc_ids(&results), vec![0]);
    Ok(())
}

#[test]
fn test_scored_exact_integer_uses_native_bm25_term() -> Result<()> {
    let layout = crate::IndexLayout::new(vec![
        crate::SingleFieldLayout::new_with_field_type_and_bm25(
            Field::from_field_id(0),
            "0",
            crate::PhysicalFieldType::I64,
            true,
        ),
    ]);
    let index_fields = vec![exact_index_field(101, true)];
    let mut writer =
        crate::TantivyIndexWriter::new_in_memory_with_index_fields(&index_fields, &layout)?;
    writer.add_document_values([(Field::from_field_id(0), OwnedValue::I64(7))])?;
    writer.add_document_values([(Field::from_field_id(0), OwnedValue::I64(-9))])?;
    writer.add_document_values([(Field::from_field_id(0), OwnedValue::I64(7))])?;
    let physical_layout = writer.physical_schema().index_layout().clone();
    assert_eq!(
        physical_layout.fields()[0].field_type(),
        crate::PhysicalFieldType::I64
    );
    assert!(physical_layout.fields()[0].enable_bm25());
    let reader = crate::IndexReader::from_tantivy_index(writer.finalize()?)?;

    let mut info = PlainFtsQueryInfo::default().as_query_info();
    info.set_query_type(tipb::FtsQueryType::FtsQueryTypeWithScore);
    info.mut_match_expr().push(exact_match_leaf(
        "inherit",
        7i64,
        vec![field_ref(101, "EXACT_VALUE_V1")],
    ));
    let query = Query::new_with_index_layout(info, &index_fields, &physical_layout)?;
    assert_eq!(query.bm25_terms().len(), 1);
    assert_eq!(query.bm25_fields(), &[Field::from_field_id(0)]);

    let mut stats = Bm25Stats::empty(&query);
    reader.accumulate_bm25_stats(&query, &mut stats)?;
    assert!(query.prepare_bm25_once(stats)?);

    let mut results = Vec::new();
    reader.search(&query, &mut results)?;
    assert_eq!(doc_ids(&results), vec![0, 2]);
    assert!(results[0].score > 0.0);
    assert!((results[0].score - results[1].score).abs() < 1e-6);
    Ok(())
}

#[test]
fn test_native_exact_leaf_validation_happens_during_construction() -> Result<()> {
    let one_field_layout = crate::IndexLayout::new(vec![
        crate::SingleFieldLayout::new_with_field_type_and_bm25(
            Field::from_field_id(0),
            "0",
            crate::PhysicalFieldType::I64,
            false,
        ),
    ]);
    let one_index_field = vec![exact_index_field(101, false)];

    let mut invalid_number = PlainFtsQueryInfo::default().as_query_info();
    invalid_number.set_query_type(tipb::FtsQueryType::FtsQueryTypeNoScore);
    invalid_number.mut_match_expr().push(match_leaf(
        tipb::ScalarFuncSig::FtsMatchWord,
        "filter",
        "not-a-number",
        vec![field_ref(101, "EXACT_VALUE_V1")],
    ));
    let err =
        match Query::new_with_index_layout(invalid_number, &one_index_field, &one_field_layout) {
            Ok(_) => panic!("expected invalid exact numeric query"),
            Err(err) => err,
        };
    assert!(
        err.to_string()
            .contains("requires a JSON number query payload")
    );

    let mut prefix = PlainFtsQueryInfo::default().as_query_info();
    prefix.set_query_type(tipb::FtsQueryType::FtsQueryTypeNoScore);
    let mut prefix_leaf = exact_match_leaf("filter", 7i64, vec![field_ref(101, "EXACT_VALUE_V1")]);
    prefix_leaf.set_sig(tipb::ScalarFuncSig::FtsMatchPrefix);
    prefix.mut_match_expr().push(prefix_leaf);
    let err = match Query::new_with_index_layout(prefix, &one_index_field, &one_field_layout) {
        Ok(_) => panic!("expected native exact prefix rejection"),
        Err(err) => err,
    };
    assert!(
        err.to_string()
            .contains("exact fields only support exact word matching")
    );

    let two_field_layout = crate::IndexLayout::new(vec![
        crate::SingleFieldLayout::new_with_field_type_and_bm25(
            Field::from_field_id(0),
            "0",
            crate::PhysicalFieldType::I64,
            false,
        ),
        crate::SingleFieldLayout::new_with_field_type_and_bm25(
            Field::from_field_id(1),
            "1",
            crate::PhysicalFieldType::I64,
            false,
        ),
    ]);
    let two_index_fields = vec![exact_index_field(101, false), exact_index_field(202, false)];
    let mut multiple_fields = PlainFtsQueryInfo::default().as_query_info();
    multiple_fields.set_query_type(tipb::FtsQueryType::FtsQueryTypeNoScore);
    multiple_fields.mut_match_expr().push(exact_match_leaf(
        "filter",
        7i64,
        vec![
            field_ref(101, "EXACT_VALUE_V1"),
            field_ref(202, "EXACT_VALUE_V1"),
        ],
    ));
    let err =
        match Query::new_with_index_layout(multiple_fields, &two_index_fields, &two_field_layout) {
            Ok(_) => panic!("expected multiple native exact field rejection"),
            Err(err) => err,
        };
    assert!(
        err.to_string()
            .contains("exact field for column 101 requires exactly one field")
    );

    let bytes_layout = crate::IndexLayout::new(vec![
        crate::SingleFieldLayout::new_with_field_type_and_bm25(
            Field::from_field_id(0),
            "0",
            crate::PhysicalFieldType::Bytes,
            false,
        ),
    ]);
    let mut invalid_base64 = PlainFtsQueryInfo::default().as_query_info();
    invalid_base64.set_query_type(tipb::FtsQueryType::FtsQueryTypeNoScore);
    invalid_base64.mut_match_expr().push(exact_match_leaf(
        "filter",
        "not-base64",
        vec![field_ref(101, "EXACT_VALUE_V1")],
    ));
    let err = match Query::new_with_index_layout(invalid_base64, &one_index_field, &bytes_layout) {
        Ok(_) => panic!("expected invalid exact query base64"),
        Err(err) => err,
    };
    assert!(err.to_string().contains("invalid exact query base64"));
    Ok(())
}

#[test]
fn test_non_exact_analyzer_rejects_native_field() -> Result<()> {
    let layout = crate::IndexLayout::new(vec![
        crate::SingleFieldLayout::new_with_field_type_and_bm25(
            Field::from_field_id(0),
            "0",
            crate::PhysicalFieldType::I64,
            false,
        ),
    ]);
    let index_fields = vec![standard_index_field(101, false)];
    let mut info = PlainFtsQueryInfo::default().as_query_info();
    info.set_query_type(tipb::FtsQueryType::FtsQueryTypeNoScore);
    info.mut_match_expr().push(match_leaf(
        tipb::ScalarFuncSig::FtsMatchWord,
        "filter",
        "7",
        vec![field_ref(101, "STANDARD_V1")],
    ));

    let err = match Query::new_with_index_layout(info, &index_fields, &layout) {
        Ok(_) => panic!("expected non-exact analyzer on native field to fail"),
        Err(err) => err,
    };
    assert!(format!("{err:#}").contains("requires a text physical field, got I64"));
    Ok(())
}

#[test]
fn test_text_field_rejects_numeric_query_payload() -> Result<()> {
    let layout = crate::IndexLayout::new(vec![crate::SingleFieldLayout::new_with_field_and_bm25(
        Field::from_field_id(0),
        "0",
        false,
    )]);
    let index_fields = vec![standard_index_field(101, false)];
    let mut info = PlainFtsQueryInfo::default().as_query_info();
    info.set_query_type(tipb::FtsQueryType::FtsQueryTypeNoScore);
    info.mut_match_expr().push(exact_match_leaf(
        "filter",
        7u64,
        vec![field_ref(101, "STANDARD_V1")],
    ));

    let err = match Query::new_with_index_layout(info, &index_fields, &layout) {
        Ok(_) => panic!("expected numeric text query rejection"),
        Err(err) => err,
    };
    assert!(format!("{err:#}").contains("requires a string query payload"));
    Ok(())
}

#[test]
fn test_final_exact_bytes_query_uses_base64_payload() -> Result<()> {
    let layout = crate::IndexLayout::new(vec![
        crate::SingleFieldLayout::new_with_field_type_and_bm25(
            Field::from_field_id(0),
            "0",
            crate::PhysicalFieldType::Bytes,
            false,
        ),
    ]);
    let index_fields = vec![exact_index_field(101, false)];
    let exact_bytes = vec![0, 1, 2, 0xff];
    let mut writer =
        crate::TantivyIndexWriter::new_in_memory_with_index_fields(&index_fields, &layout)?;
    writer.add_document_values([(
        Field::from_field_id(0),
        OwnedValue::Bytes(exact_bytes.clone()),
    )])?;
    let physical_layout = writer.physical_schema().index_layout().clone();
    let reader = crate::IndexReader::from_tantivy_index(writer.finalize()?)?;

    let mut info = PlainFtsQueryInfo::default().as_query_info();
    info.set_query_type(tipb::FtsQueryType::FtsQueryTypeNoScore);
    info.mut_match_expr().push(exact_match_leaf(
        "filter",
        base64::encode(&exact_bytes),
        vec![field_ref(101, "EXACT_VALUE_V1")],
    ));
    let query = Query::new_with_index_layout(info, &index_fields, &physical_layout)?;

    let mut results = Vec::new();
    reader.search(&query, &mut results)?;
    assert_eq!(doc_ids(&results), vec![0]);
    Ok(())
}