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

use anyhow::{Result, anyhow, bail};
use kvenginepb::fts::{FullTextAnalyzerType, FullTextIndexFieldDef};
use tantivy::schema::{BytesOptions, NumericOptions, Schema, TextOptions};
use tipb::ColumnInfo;

use crate::{
    IndexLayout, PhysicalFieldType, SingleFieldLayout,
    physical_index_analysis::{text_field_indexing, text_tokenizer_name, validate_index_fields},
};

/// Tantivy schema built from an already-resolved physical field layout.
///
/// Callers resolve that layout from the current table `ColumnInfo` before
/// constructing readers or writers. Clara then uses the same layout for schema
/// creation, value encoding and query lowering.
#[derive(Clone, Debug, PartialEq)]
pub struct PhysicalIndexSchema {
    tantivy_schema: Schema,
    index_layout: IndexLayout,
}

impl PhysicalIndexSchema {
    pub fn build(
        index_fields: &[FullTextIndexFieldDef],
        index_layout: &IndexLayout,
    ) -> Result<Self> {
        validate_index_fields(index_fields, index_layout)?;
        crate::physical_index_analysis::validate_resolved_field_types(index_fields, index_layout)?;
        Self::build_validated(index_fields, index_layout)
    }

    pub(crate) fn build_validated(
        index_fields: &[FullTextIndexFieldDef],
        index_layout: &IndexLayout,
    ) -> Result<Self> {
        let mut schema_builder = Schema::builder();
        for (field_idx, (field_layout, index_field)) in
            index_layout.fields().iter().zip(index_fields).enumerate()
        {
            let field_name = field_layout.field_name();
            let field = match field_layout.field_type() {
                PhysicalFieldType::Text => {
                    let tokenizer_name = text_tokenizer_name(
                        index_field.get_index_field_option().get_analyzer_type(),
                        field_idx,
                    )?;
                    schema_builder.add_text_field(
                        field_name,
                        TextOptions::default().set_indexing_options(text_field_indexing(
                            index_field.get_index_field_option(),
                            &tokenizer_name,
                        )),
                    )
                }
                PhysicalFieldType::I64 => schema_builder.add_i64_field(
                    field_name,
                    native_numeric_options(field_layout.enable_bm25()),
                ),
                PhysicalFieldType::U64 => schema_builder.add_u64_field(
                    field_name,
                    native_numeric_options(field_layout.enable_bm25()),
                ),
                PhysicalFieldType::F64 => schema_builder.add_f64_field(
                    field_name,
                    native_numeric_options(field_layout.enable_bm25()),
                ),
                PhysicalFieldType::Bytes => schema_builder
                    .add_bytes_field(field_name, native_bytes_options(field_layout.enable_bm25())),
            };
            if field != field_layout.field() {
                bail!("unexpected field id for index layout {:?}", field_name);
            }
        }

        Ok(Self {
            tantivy_schema: schema_builder.build(),
            index_layout: index_layout.clone(),
        })
    }

    #[inline]
    pub fn tantivy_schema(&self) -> &Schema {
        &self.tantivy_schema
    }

    #[inline]
    pub fn index_layout(&self) -> &IndexLayout {
        &self.index_layout
    }
}

/// Exact-value encoding and scoring capability are independent. Enabling BM25
/// lets a native exact term contribute to scoring instead of being filter-only.
fn native_numeric_options(enable_bm25: bool) -> NumericOptions {
    let options = NumericOptions::default().set_indexed();
    if enable_bm25 {
        options.set_fieldnorm()
    } else {
        options
    }
}

/// See [`native_numeric_options`].
fn native_bytes_options(enable_bm25: bool) -> BytesOptions {
    let options = BytesOptions::default().set_indexed();
    if enable_bm25 {
        options.set_fieldnorms()
    } else {
        options
    }
}

/// Resolves the physical field types expected by query and reader paths.
///
/// This does not build a Tantivy schema. Writers use [`PhysicalIndexSchema`]
/// when creating a new index; readers only need this layout to lower exact
/// values and validate the schema embedded in each existing index.
pub fn resolve_physical_index_layout(
    index_fields: &[FullTextIndexFieldDef],
    columns: &[ColumnInfo],
    index_layout: &IndexLayout,
) -> Result<IndexLayout> {
    validate_index_fields(index_fields, index_layout)?;
    let fields = index_layout
        .fields()
        .iter()
        .zip(index_fields)
        .map(|(field_layout, index_field)| {
            let field_type = if index_field.get_index_field_option().get_analyzer_type()
                == FullTextAnalyzerType::FtsAnalyzerExactValueV1
            {
                let column = columns
                    .iter()
                    .find(|column| column.get_column_id() == index_field.get_col_id())
                    .ok_or_else(|| {
                        anyhow!(
                            "missing ColumnInfo for exact FTS field with col_id {}",
                            index_field.get_col_id()
                        )
                    })?;
                crate::exact_physical_field_type(column)?
            } else {
                PhysicalFieldType::Text
            };
            Ok(SingleFieldLayout::new_with_field_type_and_bm25(
                field_layout.field(),
                field_layout.field_name(),
                field_type,
                field_layout.enable_bm25(),
            ))
        })
        .collect::<Result<Vec<_>>>()?;
    Ok(IndexLayout::new(fields))
}

#[cfg(test)]
mod tests {
    use kvenginepb::fts::FullTextIndexFieldOption;
    use tantivy::schema::{Field, FieldType};
    use tidb_query_datatype::{Collation, FieldTypeFlag, FieldTypeTp};

    use super::*;

    fn index_field(
        analyzer_type: FullTextAnalyzerType,
        enable_bm25: bool,
    ) -> FullTextIndexFieldDef {
        let mut field = FullTextIndexFieldDef::new();
        let mut option = FullTextIndexFieldOption::new();
        option.set_analyzer_type(analyzer_type);
        option.set_enable_bm25(enable_bm25);
        field.set_index_field_option(option);
        field
    }

    fn exact_field(col_id: i64, enable_bm25: bool) -> FullTextIndexFieldDef {
        let mut field = index_field(FullTextAnalyzerType::FtsAnalyzerExactValueV1, enable_bm25);
        field.set_col_id(col_id);
        field
    }

    fn column(
        col_id: i64,
        field_type: FieldTypeTp,
        unsigned: bool,
        collation: Collation,
    ) -> ColumnInfo {
        let mut column = ColumnInfo::new();
        column.set_column_id(col_id);
        column.set_tp(field_type as i32);
        column.set_collation(collation as i32);
        if unsigned {
            column.set_flag(FieldTypeFlag::UNSIGNED.bits() as i32);
        }
        column
    }

    #[test]
    fn test_exact_physical_type_and_bm25_are_independent() {
        let cases = [
            (
                FieldTypeTp::LongLong,
                false,
                Collation::Binary,
                PhysicalFieldType::I64,
            ),
            (
                FieldTypeTp::LongLong,
                true,
                Collation::Binary,
                PhysicalFieldType::U64,
            ),
            (
                FieldTypeTp::Double,
                false,
                Collation::Binary,
                PhysicalFieldType::F64,
            ),
            (
                FieldTypeTp::Enum,
                false,
                Collation::Utf8Mb4GeneralCi,
                PhysicalFieldType::U64,
            ),
            (
                FieldTypeTp::VarString,
                false,
                Collation::Utf8Mb4GeneralCi,
                PhysicalFieldType::Text,
            ),
        ];

        for enable_bm25 in [false, true] {
            for (field_type, unsigned, collation, expected) in cases {
                let fields = vec![exact_field(1, enable_bm25)];
                let columns = vec![column(1, field_type, unsigned, collation)];
                let layout = IndexLayout::new(vec![SingleFieldLayout::new_with_field_and_bm25(
                    Field::from_field_id(0),
                    "0",
                    enable_bm25,
                )]);

                let layout = resolve_physical_index_layout(&fields, &columns, &layout).unwrap();
                let schema = PhysicalIndexSchema::build(&fields, &layout).unwrap();
                assert_eq!(schema.index_layout().fields()[0].field_type(), expected);
                schema
                    .index_layout()
                    .validate_schema(schema.tantivy_schema())
                    .unwrap();

                let actual = schema
                    .tantivy_schema()
                    .get_field_entry(Field::from_field_id(0))
                    .field_type();
                let fieldnorms = match (expected, actual) {
                    (PhysicalFieldType::Text, FieldType::Str(options)) => options
                        .get_indexing_options()
                        .expect("exact text field must be indexed")
                        .fieldnorms(),
                    (PhysicalFieldType::I64, FieldType::I64(options))
                    | (PhysicalFieldType::U64, FieldType::U64(options))
                    | (PhysicalFieldType::F64, FieldType::F64(options)) => {
                        assert!(options.is_indexed());
                        options.fieldnorms()
                    }
                    (PhysicalFieldType::Bytes, FieldType::Bytes(options)) => {
                        assert!(options.is_indexed());
                        options.fieldnorms()
                    }
                    _ => panic!("unexpected Tantivy field type {actual:?} for {expected:?}"),
                };
                assert_eq!(fieldnorms, enable_bm25);
            }
        }
    }

    #[test]
    fn test_exact_binary_string_is_rejected() {
        let fields = vec![exact_field(1, false)];
        let columns = vec![column(1, FieldTypeTp::VarString, false, Collation::Binary)];
        let layout = IndexLayout::from_field_names(["0"]);

        assert!(resolve_physical_index_layout(&fields, &columns, &layout).is_err());
    }

    #[test]
    fn test_build_typed_physical_schema() -> Result<()> {
        let field_types = [
            PhysicalFieldType::Text,
            PhysicalFieldType::I64,
            PhysicalFieldType::U64,
            PhysicalFieldType::F64,
            PhysicalFieldType::Bytes,
        ];
        let index_fields = field_types
            .iter()
            .map(|field_type| {
                index_field(
                    if *field_type == PhysicalFieldType::Text {
                        FullTextAnalyzerType::FtsAnalyzerStandardV1
                    } else {
                        FullTextAnalyzerType::FtsAnalyzerExactValueV1
                    },
                    true,
                )
            })
            .collect::<Vec<_>>();
        let layout = IndexLayout::new(
            field_types
                .iter()
                .enumerate()
                .map(|(field_idx, field_type)| {
                    SingleFieldLayout::new_with_field_type_and_bm25(
                        Field::from_field_id(field_idx as u32),
                        field_idx.to_string(),
                        *field_type,
                        true,
                    )
                })
                .collect(),
        );

        let schema = PhysicalIndexSchema::build(&index_fields, &layout)?;
        schema
            .index_layout()
            .validate_schema(schema.tantivy_schema())?;

        for (field_idx, expected) in field_types.iter().enumerate() {
            let actual = schema
                .tantivy_schema()
                .get_field_entry(Field::from_field_id(field_idx as u32))
                .field_type();
            assert!(matches!(
                (expected, actual),
                (PhysicalFieldType::Text, FieldType::Str(_))
                    | (PhysicalFieldType::I64, FieldType::I64(_))
                    | (PhysicalFieldType::U64, FieldType::U64(_))
                    | (PhysicalFieldType::F64, FieldType::F64(_))
                    | (PhysicalFieldType::Bytes, FieldType::Bytes(_))
            ));
        }
        Ok(())
    }

    #[test]
    fn test_non_exact_analyzer_rejects_native_field() {
        let index_fields = vec![index_field(
            FullTextAnalyzerType::FtsAnalyzerStandardV1,
            false,
        )];
        let layout = IndexLayout::new(vec![SingleFieldLayout::new_with_field_type_and_bm25(
            Field::from_field_id(0),
            "0",
            PhysicalFieldType::I64,
            false,
        )]);

        assert!(PhysicalIndexSchema::build(&index_fields, &layout).is_err());
    }
}