// Copyright 2026 PingCAP, Inc.

use std::collections::HashSet;

use anyhow::{Result, anyhow, bail};
use serde::Deserialize;

/// Version of the JSON metadata stored in an outermost match-expression node's
/// `Expr.val`.
///
/// Each outermost root in `FtsQueryInfo.match_expr` must carry a versioned
/// metadata object. Match leaves must also provide `query` and `fields`.
/// Non-root nodes may omit `Expr.val` entirely when they only use default
/// values. A canonical SQL path-prefix filter leaf can use:
///
/// ```json
/// {
///   "version": 1,
///   "score_mode": "filter",
///   "query": "/src",
///   "fields": [
///     {
///       "col_id": 4,
///       "index_field": "PATH_HIERARCHY_V1"
///     }
///   ]
/// }
/// ```
///
/// Match-expression children may remain for TiCI compatibility, but CSE does
/// not read query text or fields from them. Unknown fields are rejected rather
/// than silently ignored.
pub const FTS_MATCH_EXPR_METADATA_VERSION: u32 = 1;

const MAX_FTS_MATCH_EXPR_DEPTH: usize = 64;
const MAX_FTS_MATCH_EXPR_NODES: usize = 1024;
const MAX_FTS_MATCH_EXPR_METADATA_BYTES: usize = 64 * 1024;

/// Controls whether a match-expression subtree may contribute to the result
/// score.
///
/// `FtsQueryTypeNoScore` takes precedence over every node. For a scored query,
/// `Filter` is inherited by the complete subtree and cannot be overridden.
#[derive(Clone, Copy, Debug, Default, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub enum FtsScoreMode {
    /// Inherit the nearest ancestor mode, or the query-level mode at the root.
    #[default]
    Inherit,
    /// Match without contributing to the result score.
    Filter,
}

#[derive(Default, Deserialize)]
#[serde(deny_unknown_fields)]
struct FtsExprMetadataV1 {
    #[serde(default)]
    version: u32,
    #[serde(default)]
    score_mode: FtsScoreMode,
    #[serde(default)]
    query: Option<serde_json::Value>,
    #[serde(default)]
    fields: Option<Vec<FtsMatchFieldRef>>,
}

/// The FTS matching operation represented by a match-expression leaf.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum FtsMatchType {
    /// Match one analyzer-specific query term exactly.
    Word,
    /// Match one analyzer-specific query-term prefix.
    Prefix,
}

/// One logical query target identified by a column ID and index field name.
#[derive(Clone, Debug, Deserialize, PartialEq, Eq)]
#[serde(deny_unknown_fields)]
pub struct FtsMatchFieldRef {
    /// Logical TiDB column ID.
    pub col_id: i64,
    /// Name used to resolve the physical index field for this column.
    pub index_field: String,
}

/// A leaf predicate in an FTS match-expression tree.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct FtsMatchPredicate {
    /// The matching operation requested by TiDB.
    pub match_type: FtsMatchType,
    /// Query payload supplied to the matching operation.
    ///
    /// Text fields interpret a string as literal UTF-8 analyzer input. Native
    /// numeric exact fields require a JSON number; exact bytes fields require
    /// a Base64 string containing the final term key.
    pub query: serde_json::Value,
    /// Logical fields, in producer order and without duplicate
    /// `(col_id, index_field)` pairs.
    pub fields: Vec<FtsMatchFieldRef>,
}

/// A storage-independent node in `FtsQueryInfo.match_expr`.
///
/// Physical field resolution, text analysis, and conversion into an execution
/// engine query are intentionally handled by later stages. Effective scoring is
/// determined by the query-level mode and the nearest `Filter` ancestor.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct FtsMatchExpr {
    /// The node-local score mode decoded from `tipb::Expr.val`.
    ///
    /// `Not` nodes are normalized to `Filter` because prohibited clauses never
    /// contribute to a score. Descendants keep their node-local modes, but the
    /// ancestor `Filter` remains dominant.
    pub score_mode: FtsScoreMode,
    /// The Boolean or leaf operation represented by this node.
    pub kind: FtsMatchExprKind,
}

/// The supported, storage-independent FTS expression operations.
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum FtsMatchExprKind {
    /// An FTS matching predicate.
    Match(FtsMatchPredicate),
    /// A conjunction. All children must match.
    And(Vec<FtsMatchExpr>),
    /// A disjunction. At least one child must match.
    Or(Vec<FtsMatchExpr>),
    /// A prohibited subtree. It never contributes to the score.
    Not(Box<FtsMatchExpr>),
}

impl FtsMatchExpr {
    fn new(score_mode: FtsScoreMode, kind: FtsMatchExprKind) -> Self {
        Self { score_mode, kind }
    }
}

/// Parses the roots in `FtsQueryInfo.match_expr` into one expression tree.
///
/// Multiple roots are treated as an implicit conjunction. The parser limits
/// expression depth, node count, and cumulative JSON metadata before
/// allocating the corresponding AST values. Query text and field references
/// are included in the metadata budget.
///
/// This function does not receive `FtsQueryInfo.query_type`. Execution stages
/// must make query-level `NoScore` override all node modes. They must also
/// reject a scored pure-negative tree until indexed and brute-force search
/// share an explicit pure-negative scoring and ordering policy.
pub fn parse_match_exprs(expressions: &[tipb::Expr]) -> Result<FtsMatchExpr> {
    if expressions.is_empty() {
        bail!("FTS match_expr must not be empty");
    }
    if expressions.len() > MAX_FTS_MATCH_EXPR_NODES {
        bail!(
            "FTS match_expr node count {} exceeds limit {}",
            expressions.len(),
            MAX_FTS_MATCH_EXPR_NODES
        );
    }

    let mut budget = ParseBudget::default();
    let mut roots = Vec::with_capacity(expressions.len());
    for expression in expressions {
        roots.push(parse_match_expr(expression, &mut budget, 1, true)?);
    }
    Ok(and_or_single(roots))
}

#[derive(Default)]
struct ParseBudget {
    nodes: usize,
    metadata_bytes: usize,
}

impl ParseBudget {
    fn enter_node(&mut self, expr: &tipb::Expr, depth: usize) -> Result<()> {
        if depth > MAX_FTS_MATCH_EXPR_DEPTH {
            bail!(
                "FTS match_expr depth {} exceeds limit {}",
                depth,
                MAX_FTS_MATCH_EXPR_DEPTH
            );
        }
        charge_budget(&mut self.nodes, 1, MAX_FTS_MATCH_EXPR_NODES, "node count")?;
        charge_budget(
            &mut self.metadata_bytes,
            expr.get_val().len(),
            MAX_FTS_MATCH_EXPR_METADATA_BYTES,
            "metadata bytes",
        )
    }
}

fn charge_budget(total: &mut usize, amount: usize, limit: usize, name: &str) -> Result<()> {
    let next = total
        .checked_add(amount)
        .ok_or_else(|| anyhow!("FTS match_expr {} overflow", name))?;
    if next > limit {
        bail!("FTS match_expr {} {} exceeds limit {}", name, next, limit);
    }
    *total = next;
    Ok(())
}

fn parse_match_expr(
    expr: &tipb::Expr,
    budget: &mut ParseBudget,
    depth: usize,
    is_root: bool,
) -> Result<FtsMatchExpr> {
    budget.enter_node(expr, depth)?;
    if expr.get_tp() != tipb::ExprType::ScalarFunc {
        bail!("unsupported FTS match_expr type {:?}", expr.get_tp());
    }

    let metadata = parse_expr_metadata(expr)?;
    validate_expr_metadata_version(expr, &metadata, is_root)?;
    let score_mode = metadata.score_mode;
    let (score_mode, kind) = match expr.get_sig() {
        tipb::ScalarFuncSig::UnaryNotInt | tipb::ScalarFuncSig::UnaryNotReal => {
            validate_logical_node_metadata(expr, &metadata)?;
            require_children(expr, 1, "UnaryNot")?;
            (
                FtsScoreMode::Filter,
                FtsMatchExprKind::Not(Box::new(parse_match_expr(
                    &expr.get_children()[0],
                    budget,
                    depth + 1,
                    false,
                )?)),
            )
        }
        tipb::ScalarFuncSig::LogicalAnd => {
            validate_logical_node_metadata(expr, &metadata)?;
            require_children(expr, 2, "LogicalAnd")?;
            (
                score_mode,
                FtsMatchExprKind::And(vec![
                    parse_match_expr(&expr.get_children()[0], budget, depth + 1, false)?,
                    parse_match_expr(&expr.get_children()[1], budget, depth + 1, false)?,
                ]),
            )
        }
        tipb::ScalarFuncSig::LogicalOr => {
            validate_logical_node_metadata(expr, &metadata)?;
            require_children(expr, 2, "LogicalOr")?;
            (
                score_mode,
                FtsMatchExprKind::Or(vec![
                    parse_match_expr(&expr.get_children()[0], budget, depth + 1, false)?,
                    parse_match_expr(&expr.get_children()[1], budget, depth + 1, false)?,
                ]),
            )
        }
        tipb::ScalarFuncSig::FtsMatchWord => (
            score_mode,
            FtsMatchExprKind::Match(parse_match_predicate(
                FtsMatchType::Word,
                "FtsMatchWord",
                metadata,
            )?),
        ),
        tipb::ScalarFuncSig::FtsMatchPrefix => (
            score_mode,
            FtsMatchExprKind::Match(parse_match_predicate(
                FtsMatchType::Prefix,
                "FtsMatchPrefix",
                metadata,
            )?),
        ),
        sig => bail!("unsupported FTS match_expr scalar function {:?}", sig),
    };
    Ok(FtsMatchExpr::new(score_mode, kind))
}

fn validate_logical_node_metadata(expr: &tipb::Expr, metadata: &FtsExprMetadataV1) -> Result<()> {
    if metadata.query.is_some() || metadata.fields.is_some() {
        bail!(
            "FTS match_expr node {:?} contains leaf-only metadata",
            expr.get_sig()
        );
    }
    Ok(())
}

fn parse_match_predicate(
    match_type: FtsMatchType,
    function_name: &str,
    metadata: FtsExprMetadataV1,
) -> Result<FtsMatchPredicate> {
    let FtsExprMetadataV1 { query, fields, .. } = metadata;
    let query = query.ok_or_else(|| anyhow!("{} requires query metadata", function_name))?;
    let fields = fields.ok_or_else(|| anyhow!("{} requires fields metadata", function_name))?;
    if fields.is_empty() {
        bail!("{} metadata fields must not be empty", function_name);
    }

    let mut seen_fields = HashSet::with_capacity(fields.len());
    for field in &fields {
        if field.index_field.is_empty() {
            bail!(
                "{} metadata index_field must not be empty for column {}",
                function_name,
                field.col_id
            );
        }
        if !seen_fields.insert((field.col_id, field.index_field.clone())) {
            bail!(
                "{} contains duplicate field for column {}",
                function_name,
                field.col_id
            );
        }
    }

    Ok(FtsMatchPredicate {
        match_type,
        query,
        fields,
    })
}

fn parse_expr_metadata(expr: &tipb::Expr) -> Result<FtsExprMetadataV1> {
    let value = expr.get_val();
    if value.is_empty() {
        return Ok(FtsExprMetadataV1::default());
    }

    serde_json::from_slice::<FtsExprMetadataV1>(value).map_err(|err| {
        anyhow!(
            "invalid FTS match_expr metadata on {:?}: {}",
            expr.get_sig(),
            err
        )
    })
}

fn validate_expr_metadata_version(
    expr: &tipb::Expr,
    metadata: &FtsExprMetadataV1,
    is_root: bool,
) -> Result<()> {
    match metadata.version {
        FTS_MATCH_EXPR_METADATA_VERSION => Ok(()),
        0 if is_root => bail!(
            "missing FTS match_expr metadata version on {:?}",
            expr.get_sig()
        ),
        0 => Ok(()),
        version => bail!(
            "unsupported FTS match_expr metadata version {} on {:?}",
            version,
            expr.get_sig()
        ),
    }
}

fn require_children(expr: &tipb::Expr, expected: usize, function_name: &str) -> Result<()> {
    if expr.get_children().len() != expected {
        bail!(
            "{} requires {} children, got {}",
            function_name,
            expected,
            expr.get_children().len()
        );
    }
    Ok(())
}

fn and_or_single(mut expressions: Vec<FtsMatchExpr>) -> FtsMatchExpr {
    if expressions.len() == 1 {
        expressions.pop().unwrap()
    } else {
        FtsMatchExpr::new(FtsScoreMode::Inherit, FtsMatchExprKind::And(expressions))
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    const TITLE_COL_ID: i64 = 101;
    const BODY_COL_ID: i64 = 202;
    const ROOT_METADATA: &str = r#"{"version":1}"#;

    fn string_expr(value: impl Into<Vec<u8>>) -> tipb::Expr {
        let mut expr = tipb::Expr::new();
        expr.set_tp(tipb::ExprType::String);
        expr.set_val(value.into());
        expr
    }

    fn default_index_field(col_id: i64) -> &'static str {
        if col_id == BODY_COL_ID {
            "MULTILINGUAL_V1"
        } else {
            "STANDARD_V1"
        }
    }

    fn match_expr(sig: tipb::ScalarFuncSig, query: &str, column_ids: &[i64]) -> tipb::Expr {
        let mut expr = tipb::Expr::new();
        expr.set_tp(tipb::ExprType::ScalarFunc);
        expr.set_sig(sig);
        expr.mut_children().push(string_expr(query.as_bytes()));
        for column_id in column_ids {
            expr.mut_children()
                .push(string_expr(format!("column_{column_id}").into_bytes()));
        }
        let fields = column_ids
            .iter()
            .map(|col_id| {
                serde_json::json!({
                    "col_id": col_id,
                    "index_field": default_index_field(*col_id),
                })
            })
            .collect::<Vec<_>>();
        expr.set_val(
            serde_json::to_vec(&serde_json::json!({
                "version": FTS_MATCH_EXPR_METADATA_VERSION,
                "query": query,
                "fields": fields,
            }))
            .unwrap(),
        );
        expr
    }

    fn boolean(sig: tipb::ScalarFuncSig, left: tipb::Expr, right: tipb::Expr) -> tipb::Expr {
        let mut expr = tipb::Expr::new();
        expr.set_tp(tipb::ExprType::ScalarFunc);
        expr.set_sig(sig);
        expr.mut_children().push(left);
        expr.mut_children().push(right);
        expr
    }

    fn unary_not(child: tipb::Expr) -> tipb::Expr {
        let mut expr = tipb::Expr::new();
        expr.set_tp(tipb::ExprType::ScalarFunc);
        expr.set_sig(tipb::ScalarFuncSig::UnaryNotInt);
        expr.mut_children().push(child);
        expr
    }

    fn with_metadata(mut expr: tipb::Expr, metadata: &str) -> tipb::Expr {
        expr.set_val(metadata.as_bytes().to_vec());
        expr
    }

    fn with_root_version(expr: tipb::Expr) -> tipb::Expr {
        with_metadata(expr, ROOT_METADATA)
    }

    fn node(kind: FtsMatchExprKind) -> FtsMatchExpr {
        FtsMatchExpr::new(FtsScoreMode::Inherit, kind)
    }

    fn filter_node(kind: FtsMatchExprKind) -> FtsMatchExpr {
        FtsMatchExpr::new(FtsScoreMode::Filter, kind)
    }

    fn field_ref(col_id: i64, index_field: impl Into<String>) -> FtsMatchFieldRef {
        FtsMatchFieldRef {
            col_id,
            index_field: index_field.into(),
        }
    }

    fn predicate(match_type: FtsMatchType, query: &str, column_ids: Vec<i64>) -> FtsMatchExpr {
        node(FtsMatchExprKind::Match(FtsMatchPredicate {
            match_type,
            query: serde_json::Value::String(query.to_owned()),
            fields: column_ids
                .into_iter()
                .map(|column_id| field_ref(column_id, default_index_field(column_id)))
                .collect(),
        }))
    }

    #[test]
    fn parses_nested_boolean_expression() {
        let expression = with_root_version(boolean(
            tipb::ScalarFuncSig::LogicalAnd,
            match_expr(tipb::ScalarFuncSig::FtsMatchWord, "rust", &[TITLE_COL_ID]),
            boolean(
                tipb::ScalarFuncSig::LogicalOr,
                match_expr(tipb::ScalarFuncSig::FtsMatchPrefix, "data", &[BODY_COL_ID]),
                unary_not(match_expr(
                    tipb::ScalarFuncSig::FtsMatchWord,
                    "draft",
                    &[TITLE_COL_ID, BODY_COL_ID],
                )),
            ),
        ));

        assert_eq!(
            parse_match_exprs(&[expression]).unwrap(),
            node(FtsMatchExprKind::And(vec![
                predicate(FtsMatchType::Word, "rust", vec![TITLE_COL_ID]),
                node(FtsMatchExprKind::Or(vec![
                    predicate(FtsMatchType::Prefix, "data", vec![BODY_COL_ID]),
                    filter_node(FtsMatchExprKind::Not(Box::new(predicate(
                        FtsMatchType::Word,
                        "draft",
                        vec![TITLE_COL_ID, BODY_COL_ID],
                    )))),
                ])),
            ]))
        );
    }

    #[test]
    fn combines_multiple_roots_with_and() {
        let first = match_expr(tipb::ScalarFuncSig::FtsMatchWord, "rust", &[TITLE_COL_ID]);
        let second = match_expr(tipb::ScalarFuncSig::FtsMatchPrefix, "data", &[BODY_COL_ID]);

        assert_eq!(
            parse_match_exprs(&[first, second]).unwrap(),
            node(FtsMatchExprKind::And(vec![
                predicate(FtsMatchType::Word, "rust", vec![TITLE_COL_ID]),
                predicate(FtsMatchType::Prefix, "data", vec![BODY_COL_ID]),
            ]))
        );
    }

    #[test]
    fn parses_versioned_filter_metadata_on_any_subtree() {
        let inherited = match_expr(tipb::ScalarFuncSig::FtsMatchWord, "rust", &[TITLE_COL_ID]);
        let filtered = with_metadata(
            boolean(
                tipb::ScalarFuncSig::LogicalOr,
                match_expr(tipb::ScalarFuncSig::FtsMatchWord, "/src/", &[BODY_COL_ID]),
                match_expr(tipb::ScalarFuncSig::FtsMatchPrefix, "/src/", &[BODY_COL_ID]),
            ),
            r#"{"version":1,"score_mode":"filter"}"#,
        );

        let parsed = parse_match_exprs(&[with_root_version(boolean(
            tipb::ScalarFuncSig::LogicalAnd,
            inherited,
            filtered,
        ))])
        .unwrap();
        assert_eq!(parsed.score_mode, FtsScoreMode::Inherit);
        let FtsMatchExprKind::And(children) = parsed.kind else {
            panic!("expected AND expression");
        };
        assert_eq!(children[0].score_mode, FtsScoreMode::Inherit);
        assert_eq!(children[1].score_mode, FtsScoreMode::Filter);
        let FtsMatchExprKind::Or(filter_children) = &children[1].kind else {
            panic!("expected filter OR expression");
        };
        assert!(
            filter_children
                .iter()
                .all(|child| child.score_mode == FtsScoreMode::Inherit)
        );
    }

    #[test]
    fn parses_explicit_fields_for_sql_filter_leaves() {
        let exact = with_metadata(
            match_expr(
                tipb::ScalarFuncSig::FtsMatchWord,
                "README.md",
                &[TITLE_COL_ID],
            ),
            r#"{"version":1,"score_mode":"filter","query":"README.md","fields":[{"col_id":101,"index_field":"EXACT_VALUE_V1"}]}"#,
        );
        let path_prefix = with_metadata(
            match_expr(tipb::ScalarFuncSig::FtsMatchWord, "/src", &[BODY_COL_ID]),
            r#"{"version":1,"score_mode":"filter","query":"/src","fields":[{"col_id":202,"index_field":"PATH_HIERARCHY_V1"}]}"#,
        );

        assert_eq!(
            parse_match_exprs(&[with_root_version(boolean(
                tipb::ScalarFuncSig::LogicalAnd,
                exact,
                path_prefix,
            ))])
            .unwrap(),
            node(FtsMatchExprKind::And(vec![
                filter_node(FtsMatchExprKind::Match(FtsMatchPredicate {
                    match_type: FtsMatchType::Word,
                    query: serde_json::Value::String("README.md".to_owned()),
                    fields: vec![field_ref(TITLE_COL_ID, "EXACT_VALUE_V1")],
                })),
                filter_node(FtsMatchExprKind::Match(FtsMatchPredicate {
                    match_type: FtsMatchType::Word,
                    query: serde_json::Value::String("/src".to_owned()),
                    fields: vec![field_ref(BODY_COL_ID, "PATH_HIERARCHY_V1")],
                })),
            ]))
        );
    }

    #[test]
    fn parses_an_explicit_index_field_name() {
        let expression = with_metadata(
            match_expr(
                tipb::ScalarFuncSig::FtsMatchWord,
                "database systems",
                &[BODY_COL_ID],
            ),
            r#"{"version":1,"query":"database systems","fields":[{"col_id":202,"index_field":"NGRAM_V1"}]}"#,
        );

        assert_eq!(
            parse_match_exprs(&[expression]).unwrap(),
            node(FtsMatchExprKind::Match(FtsMatchPredicate {
                match_type: FtsMatchType::Word,
                query: serde_json::Value::String("database systems".to_owned()),
                fields: vec![field_ref(BODY_COL_ID, "NGRAM_V1")],
            }))
        );
    }

    #[test]
    fn allows_one_column_to_reference_different_index_fields() {
        let expression = with_metadata(
            match_expr(
                tipb::ScalarFuncSig::FtsMatchWord,
                "/src",
                &[BODY_COL_ID, BODY_COL_ID],
            ),
            r#"{"version":1,"query":"/src","fields":[{"col_id":202,"index_field":"EXACT_VALUE_V1"},{"col_id":202,"index_field":"PATH_HIERARCHY_V1"}]}"#,
        );

        assert_eq!(
            parse_match_exprs(&[expression]).unwrap(),
            node(FtsMatchExprKind::Match(FtsMatchPredicate {
                match_type: FtsMatchType::Word,
                query: serde_json::Value::String("/src".to_owned()),
                fields: vec![
                    field_ref(BODY_COL_ID, "EXACT_VALUE_V1"),
                    field_ref(BODY_COL_ID, "PATH_HIERARCHY_V1"),
                ],
            }))
        );
    }

    #[test]
    fn metadata_is_authoritative_over_match_children() {
        let expression = with_metadata(
            match_expr(
                tipb::ScalarFuncSig::FtsMatchWord,
                "ignored child query",
                &[BODY_COL_ID],
            ),
            r#"{"version":1,"query":"rust","fields":[{"col_id":101,"index_field":"STANDARD_V1"}]}"#,
        );

        assert_eq!(
            parse_match_exprs(&[expression]).unwrap(),
            node(FtsMatchExprKind::Match(FtsMatchPredicate {
                match_type: FtsMatchType::Word,
                query: serde_json::Value::String("rust".to_owned()),
                fields: vec![field_ref(TITLE_COL_ID, "STANDARD_V1")],
            }))
        );
    }

    #[test]
    fn parses_match_leaf_without_argument_children() {
        let mut expression = match_expr(tipb::ScalarFuncSig::FtsMatchWord, "rust", &[TITLE_COL_ID]);
        expression.clear_children();

        assert_eq!(
            parse_match_exprs(&[expression]).unwrap(),
            predicate(FtsMatchType::Word, "rust", vec![TITLE_COL_ID])
        );
    }

    #[test]
    fn allows_sparse_non_root_metadata() {
        let sparse_leaf = with_metadata(
            match_expr(tipb::ScalarFuncSig::FtsMatchWord, "rust", &[TITLE_COL_ID]),
            r#"{"query":"rust","fields":[{"col_id":101,"index_field":"STANDARD_V1"}]}"#,
        );
        let sparse_prefix = with_metadata(
            match_expr(tipb::ScalarFuncSig::FtsMatchPrefix, "data", &[BODY_COL_ID]),
            r#"{"query":"data","fields":[{"col_id":202,"index_field":"MULTILINGUAL_V1"}]}"#,
        );

        assert_eq!(
            parse_match_exprs(&[with_root_version(boolean(
                tipb::ScalarFuncSig::LogicalAnd,
                sparse_leaf,
                boolean(
                    tipb::ScalarFuncSig::LogicalOr,
                    sparse_prefix,
                    unary_not(match_expr(
                        tipb::ScalarFuncSig::FtsMatchWord,
                        "draft",
                        &[TITLE_COL_ID],
                    ))
                ),
            ))])
            .unwrap(),
            node(FtsMatchExprKind::And(vec![
                predicate(FtsMatchType::Word, "rust", vec![TITLE_COL_ID]),
                node(FtsMatchExprKind::Or(vec![
                    predicate(FtsMatchType::Prefix, "data", vec![BODY_COL_ID]),
                    filter_node(FtsMatchExprKind::Not(Box::new(predicate(
                        FtsMatchType::Word,
                        "draft",
                        vec![TITLE_COL_ID],
                    )))),
                ])),
            ]))
        );
    }

    #[test]
    fn rejects_invalid_field_metadata() {
        let field_on_logical_node = with_metadata(
            boolean(
                tipb::ScalarFuncSig::LogicalAnd,
                match_expr(tipb::ScalarFuncSig::FtsMatchWord, "rust", &[TITLE_COL_ID]),
                match_expr(tipb::ScalarFuncSig::FtsMatchWord, "search", &[BODY_COL_ID]),
            ),
            r#"{"version":1,"fields":[{"col_id":101,"index_field":"EXACT_VALUE_V1"}]}"#,
        );
        assert!(
            parse_match_exprs(&[field_on_logical_node])
                .unwrap_err()
                .to_string()
                .contains("contains leaf-only metadata")
        );

        let empty_fields = with_metadata(
            match_expr(tipb::ScalarFuncSig::FtsMatchWord, "rust", &[TITLE_COL_ID]),
            r#"{"version":1,"query":"rust","fields":[]}"#,
        );
        assert!(
            parse_match_exprs(&[empty_fields])
                .unwrap_err()
                .to_string()
                .contains("metadata fields must not be empty")
        );

        let empty_index_field = with_metadata(
            match_expr(tipb::ScalarFuncSig::FtsMatchWord, "/src/", &[BODY_COL_ID]),
            r#"{"version":1,"query":"/src/","fields":[{"col_id":202,"index_field":""}]}"#,
        );
        assert!(
            parse_match_exprs(&[empty_index_field])
                .unwrap_err()
                .to_string()
                .contains("index_field must not be empty")
        );
    }

    #[test]
    fn rejects_invalid_or_unsupported_metadata() {
        let make_expression = |metadata: &str| {
            with_metadata(
                match_expr(tipb::ScalarFuncSig::FtsMatchWord, "rust", &[TITLE_COL_ID]),
                metadata,
            )
        };

        let err = parse_match_exprs(&[make_expression("{")])
            .unwrap_err()
            .to_string();
        assert!(err.contains("invalid FTS match_expr metadata"));

        let err = parse_match_exprs(&[make_expression(
            r#"{"version":1,"query":"rust","fields":null}"#,
        )])
        .unwrap_err()
        .to_string();
        assert!(err.contains("requires fields metadata"));

        let err = parse_match_exprs(&[make_expression(
            r#"{"version":1,"fields":[{"col_id":101,"index_field":"STANDARD_V1"}]}"#,
        )])
        .unwrap_err()
        .to_string();
        assert!(err.contains("requires query metadata"));

        let err = parse_match_exprs(&[make_expression(r#"{"version":1,"query":"rust"}"#)])
            .unwrap_err()
            .to_string();
        assert!(err.contains("requires fields metadata"));

        let err = parse_match_exprs(&[with_metadata(
            match_expr(tipb::ScalarFuncSig::FtsMatchWord, "rust", &[TITLE_COL_ID]),
            r#"{"query":"rust","fields":[{"col_id":101,"index_field":"STANDARD_V1"}]}"#,
        )])
        .unwrap_err()
        .to_string();
        assert!(err.contains("missing FTS match_expr metadata version"));

        let err = parse_match_exprs(&[make_expression(r#"{"version":2,"score_mode":"filter"}"#)])
            .unwrap_err()
            .to_string();
        assert!(err.contains("unsupported FTS match_expr metadata version 2"));

        let err = parse_match_exprs(&[make_expression(
            r#"{"version":1,"score_mode":"filter","query":"rust","fields":[{"col_id":101,"index_field":"STANDARD_V1"}],"boost":2}"#,
        )])
        .unwrap_err()
        .to_string();
        assert!(err.contains("unknown field `boost`"));

        let err = parse_match_exprs(&[make_expression(
            r#"{"version":1,"score_mode":"constant","query":"rust","fields":[{"col_id":101,"index_field":"STANDARD_V1"}]}"#,
        )])
        .unwrap_err()
        .to_string();
        assert!(err.contains("unknown variant `constant`"));

        let err = parse_match_exprs(&[make_expression(
            r#"{"version":1,"score_mode":"score","query":"rust","fields":[{"col_id":101,"index_field":"STANDARD_V1"}]}"#,
        )])
        .unwrap_err()
        .to_string();
        assert!(err.contains("unknown variant `score`"));

        let err = parse_match_exprs(&[make_expression(
            r#"{"version":1,"query":"rust","fields":[{"col_id":101,"index_field":{"analyzer_type":"EXACT_VALUE_V1"}}]}"#,
        )])
        .unwrap_err()
        .to_string();
        assert!(err.contains("expected a string"));
    }

    #[test]
    fn keeps_query_payload_uninterpreted_until_field_resolution() {
        let encoded_query = base64::encode([1, 2, 3]);
        let expression = with_metadata(
            match_expr(tipb::ScalarFuncSig::FtsMatchWord, "", &[TITLE_COL_ID]),
            &serde_json::json!({
                "version": 1,
                "score_mode": "filter",
                "query": encoded_query,
                "fields": [{"col_id": TITLE_COL_ID, "index_field": "EXACT_VALUE_V1"}],
            })
            .to_string(),
        );

        let parsed = parse_match_exprs(&[expression]).unwrap();
        let FtsMatchExprKind::Match(predicate) = parsed.kind else {
            panic!("expected match predicate");
        };
        assert_eq!(predicate.query, serde_json::Value::String(encoded_query));
    }

    #[test]
    fn preserves_u64_query_payload_until_field_resolution() {
        let expression = with_metadata(
            match_expr(tipb::ScalarFuncSig::FtsMatchWord, "", &[TITLE_COL_ID]),
            &serde_json::json!({
                "version": 1,
                "score_mode": "filter",
                "query": u64::MAX,
                "fields": [{"col_id": TITLE_COL_ID, "index_field": "EXACT_VALUE_V1"}],
            })
            .to_string(),
        );

        let parsed = parse_match_exprs(&[expression]).unwrap();
        let FtsMatchExprKind::Match(predicate) = parsed.kind else {
            panic!("expected match predicate");
        };
        assert_eq!(
            predicate.query,
            serde_json::Value::Number(serde_json::Number::from(u64::MAX))
        );
    }

    #[test]
    fn leaves_exact_constraints_for_physical_lowering() {
        let expression = with_metadata(
            match_expr(
                tipb::ScalarFuncSig::FtsMatchPrefix,
                "not-base64",
                &[TITLE_COL_ID, BODY_COL_ID],
            ),
            r#"{"version":1,"query":"not-base64","fields":[{"col_id":101,"index_field":"EXACT_VALUE_V1"},{"col_id":202,"index_field":"EXACT_VALUE_V1"}]}"#,
        );

        let parsed = parse_match_exprs(&[expression]).unwrap();
        let FtsMatchExprKind::Match(predicate) = parsed.kind else {
            panic!("expected match predicate");
        };
        assert_eq!(predicate.match_type, FtsMatchType::Prefix);
        assert_eq!(
            predicate.query,
            serde_json::Value::String("not-base64".to_owned())
        );
        assert_eq!(predicate.fields.len(), 2);
    }

    #[test]
    fn normalizes_not_to_filter() {
        let parsed = parse_match_exprs(&[with_root_version(unary_not(match_expr(
            tipb::ScalarFuncSig::FtsMatchWord,
            "draft",
            &[BODY_COL_ID],
        )))])
        .unwrap();

        assert_eq!(parsed.score_mode, FtsScoreMode::Filter);
        let FtsMatchExprKind::Not(child) = parsed.kind else {
            panic!("expected NOT expression");
        };
        assert_eq!(child.score_mode, FtsScoreMode::Inherit);
    }

    #[test]
    fn rejects_empty_and_malformed_expressions() {
        assert_eq!(
            parse_match_exprs(&[]).unwrap_err().to_string(),
            "FTS match_expr must not be empty"
        );

        let mut non_scalar = tipb::Expr::new();
        non_scalar.set_tp(tipb::ExprType::String);
        assert!(
            parse_match_exprs(&[non_scalar])
                .unwrap_err()
                .to_string()
                .contains("unsupported FTS match_expr type")
        );

        let mut missing_leaf_metadata =
            match_expr(tipb::ScalarFuncSig::FtsMatchWord, "rust", &[TITLE_COL_ID]);
        missing_leaf_metadata.set_val(Vec::new());
        assert!(
            parse_match_exprs(&[missing_leaf_metadata])
                .unwrap_err()
                .to_string()
                .contains("missing FTS match_expr metadata version")
        );

        let mut missing_logical_metadata = boolean(
            tipb::ScalarFuncSig::LogicalAnd,
            match_expr(tipb::ScalarFuncSig::FtsMatchWord, "rust", &[TITLE_COL_ID]),
            match_expr(tipb::ScalarFuncSig::FtsMatchWord, "search", &[BODY_COL_ID]),
        );
        missing_logical_metadata.set_val(Vec::new());
        assert!(
            parse_match_exprs(&[missing_logical_metadata])
                .unwrap_err()
                .to_string()
                .contains("missing FTS match_expr metadata version")
        );

        let invalid_arity = with_root_version(boolean(
            tipb::ScalarFuncSig::LogicalAnd,
            match_expr(tipb::ScalarFuncSig::FtsMatchWord, "rust", &[TITLE_COL_ID]),
            match_expr(tipb::ScalarFuncSig::FtsMatchWord, "search", &[BODY_COL_ID]),
        ));
        let mut invalid_arity = invalid_arity;
        invalid_arity.mut_children().pop();
        assert_eq!(
            parse_match_exprs(&[invalid_arity]).unwrap_err().to_string(),
            "LogicalAnd requires 2 children, got 1"
        );
    }

    #[test]
    fn rejects_invalid_match_arguments() {
        let no_column = match_expr(tipb::ScalarFuncSig::FtsMatchWord, "rust", &[]);
        assert!(
            parse_match_exprs(&[no_column])
                .unwrap_err()
                .to_string()
                .contains("metadata fields must not be empty")
        );

        let mut ignored_column_child =
            match_expr(tipb::ScalarFuncSig::FtsMatchWord, "rust", &[TITLE_COL_ID]);
        ignored_column_child.mut_children()[1] = string_expr("not_a_column_id".as_bytes());
        assert_eq!(
            parse_match_exprs(&[ignored_column_child]).unwrap(),
            predicate(FtsMatchType::Word, "rust", vec![TITLE_COL_ID])
        );

        let duplicate_columns = match_expr(
            tipb::ScalarFuncSig::FtsMatchPrefix,
            "data",
            &[BODY_COL_ID, BODY_COL_ID],
        );
        assert_eq!(
            parse_match_exprs(&[duplicate_columns])
                .unwrap_err()
                .to_string(),
            format!("FtsMatchPrefix contains duplicate field for column {BODY_COL_ID}")
        );
    }

    #[test]
    fn rejects_unsupported_scalar_function() {
        let expression = with_root_version(boolean(
            tipb::ScalarFuncSig::PlusInt,
            string_expr("left".as_bytes()),
            string_expr("right".as_bytes()),
        ));
        assert!(
            parse_match_exprs(&[expression])
                .unwrap_err()
                .to_string()
                .contains("unsupported FTS match_expr scalar function")
        );
    }

    #[test]
    fn rejects_expression_over_depth_limit() {
        let mut expression = match_expr(tipb::ScalarFuncSig::FtsMatchWord, "rust", &[TITLE_COL_ID]);
        for _ in 0..MAX_FTS_MATCH_EXPR_DEPTH {
            expression = unary_not(expression);
        }
        let expression = with_root_version(expression);

        let err = parse_match_exprs(&[expression]).unwrap_err().to_string();
        assert!(err.contains("depth 65 exceeds limit 64"));
    }

    #[test]
    fn rejects_expression_over_node_limit() {
        let expressions = (0..=MAX_FTS_MATCH_EXPR_NODES)
            .map(|_| match_expr(tipb::ScalarFuncSig::FtsMatchWord, "rust", &[TITLE_COL_ID]))
            .collect::<Vec<_>>();

        let err = parse_match_exprs(&expressions).unwrap_err().to_string();
        assert!(err.contains("node count 1025 exceeds limit 1024"));
    }

    #[test]
    fn rejects_expression_over_metadata_limit() {
        let mut metadata = r#"{"version":1,"score_mode":"filter"}"#.to_owned();
        metadata.push_str(&" ".repeat(MAX_FTS_MATCH_EXPR_METADATA_BYTES));
        let expression = with_metadata(
            match_expr(tipb::ScalarFuncSig::FtsMatchWord, "rust", &[TITLE_COL_ID]),
            &metadata,
        );

        let err = parse_match_exprs(&[expression]).unwrap_err().to_string();
        assert!(err.contains("metadata bytes"));
        assert!(err.contains("exceeds limit 65536"));
    }
}