// Copyright (c) 2023-2026 ParadeDB, Inc.
//
// This file is part of ParadeDB - Postgres for Search and Analytics
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.

use crate::api::version::{Version, VersionInfo};
use crate::api::FieldName;
use crate::postgres::datetime::PostgresDateTime;
use crate::postgres::pdb_owned_value::PdbOwnedValue;
use crate::postgres::types::is_pgoid_datetime_type;
use crate::query::numeric::{
    convert_value_for_field, convert_value_for_range_field, map_bound, numeric_bound_to_bytes,
    scale_numeric_bound, string_to_f64, string_to_i64, string_to_json_numeric, string_to_u64,
};
use crate::query::pdb_query::pdb::{FuzzyData, ScoreAdjustStyle, SlopData};
use crate::query::proximity::query::ProximityQuery;
use crate::query::proximity::{ProximityClause, ProximityDistance};
use crate::query::range::{Comparison, RangeField};
use crate::query::{
    check_range_bounds, coerce_bound_to_field_type, value_to_term, QueryError, SearchQueryInput,
};
use crate::schema::{IndexRecordOption, SearchField, SearchFieldType, SearchIndexSchema};
use pgrx::pg_sys::BuiltinOid;
use pgrx::PgOid;
use pgrx::{pg_extern, pg_schema, InOutFuncs, StringInfo};
use serde_json::Value;
use std::collections::Bound;
use std::ffi::CStr;
use tantivy::query::{
    AllQuery, BooleanQuery, BoostQuery, ConstScoreQuery, EmptyQuery, ExistsQuery,
    FastFieldRangeQuery, FuzzyTermQuery, Occur, PhrasePrefixQuery, PhraseQuery,
    Query as TantivyQuery, Query, QueryParser, RangeQuery, RegexPhraseQuery, RegexQuery, TermQuery,
    TermSetQuery,
};
use tantivy::query_grammar::{self, UserInputAst, UserInputBound, UserInputLeaf};
use tantivy::schema::FieldType;
use tantivy::{Searcher, Term};
use tokenizers::SearchTokenizer;

use super::dot_path_to_facet;

#[pg_extern(immutable, parallel_safe)]
pub fn to_search_query_input(field: FieldName, query: pdb::Query) -> SearchQueryInput {
    SearchQueryInput::FieldedQuery { field, query }
}

#[pg_schema]
pub mod pdb {
    use crate::postgres::pdb_owned_value::PdbOwnedValue;
    use crate::query::proximity::{ProximityClause, ProximityDistance};
    use crate::query::range::{deserialize_bound, serialize_bound};
    use pgrx::PostgresType;
    use serde::{Deserialize, Serialize};
    use std::collections::Bound;
    use std::fmt::{Display, Formatter};

    #[derive(Debug, Deserialize, Serialize, Clone, PartialEq)]
    #[serde(rename_all = "snake_case")]
    pub struct FuzzyData {
        pub distance: u8,
        pub prefix: bool,
        pub transposition_cost_one: bool,
    }

    impl Display for FuzzyData {
        #[rustfmt::skip]
        fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
            write!(
                f,
                "{},{},{}",
                self.distance,
                if self.prefix { "t" } else { "f" },
                if self.transposition_cost_one { "t" } else { "f" }
            )
        }
    }

    impl From<i32> for FuzzyData {
        fn from(value: i32) -> Self {
            let distance = (value >> 2) as u8;
            let prefix = (value & 2) != 0;
            let transposition_cost_one = (value & 1) != 0;
            FuzzyData {
                distance,
                prefix,
                transposition_cost_one,
            }
        }
    }

    impl From<FuzzyData> for i32 {
        fn from(data: FuzzyData) -> Self {
            ((data.distance as i32) << 2)
                | ((data.prefix as i32) << 1)
                | (data.transposition_cost_one as i32)
        }
    }

    #[derive(Debug, Deserialize, Serialize, Clone, PartialEq)]
    #[serde(rename_all = "snake_case")]
    pub struct SlopData {
        pub slop: u32,
    }

    impl Display for SlopData {
        #[rustfmt::skip]
        fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
            write!(
                f,
                "{}",
                self.slop,
            )
        }
    }

    impl From<i32> for SlopData {
        fn from(value: i32) -> Self {
            SlopData { slop: value as u32 }
        }
    }

    impl From<SlopData> for i32 {
        fn from(data: SlopData) -> Self {
            data.slop as i32
        }
    }

    #[derive(Debug, Deserialize, Serialize, Clone, PartialEq)]
    pub enum ScoreAdjustStyle {
        Boost(tantivy::Score),
        Const(tantivy::Score),
    }

    #[derive(Debug, PostgresType, Deserialize, Serialize, Clone, PartialEq)]
    #[inoutfuncs]
    #[serde(rename_all = "snake_case")]
    pub enum Query {
        All,
        Empty,

        /// This is instantiated in places where a string literal is used
        /// as the right-hand-side of one of our operators.  For example, in
        ///
        /// ```sql
        /// SELECT * FROM t WHERE f @@@ 'some string'
        /// ```
        ///
        /// This variant is constructed first, then the "SUPPORT" function for our various operators
        /// will rewrite it to the [`Query`] variant that is correct for its usage.
        ///
        /// For example, the `===` operator will rewrite it to a [`Query::Term`] and `@@@` to
        /// a [`Query::ParseWithField`].
        UnclassifiedString {
            string: String,
            #[serde(skip_serializing_if = "Option::is_none")]
            fuzzy_data: Option<FuzzyData>,
            #[serde(skip_serializing_if = "Option::is_none")]
            slop_data: Option<SlopData>,
        },
        /// This is instantiated in places where a text array is used
        /// as the right-hand-side of one of our operators.  For example, in
        ///
        /// ```sql
        /// SELECT * FROM t WHERE f @@@ ARRAY['some', 'terms']
        /// ```
        ///
        /// This variant is constructed first, then the "SUPPORT" function for our various operators
        /// will rewrite it to the [`Query`] variant that is correct for its usage.
        ///
        /// For example, the `===` operator will rewrite it to a [`Query::TermSet`] and `@@@` to
        /// a [`Query::ParseWithField`].
        UnclassifiedArray {
            array: Vec<String>,
            #[serde(skip_serializing_if = "Option::is_none")]
            fuzzy_data: Option<FuzzyData>,
            #[serde(skip_serializing_if = "Option::is_none")]
            slop_data: Option<SlopData>,
        },
        ScoreAdjusted {
            query: Box<Query>,
            score: Option<ScoreAdjustStyle>,
        },
        Exists,
        FastFieldRangeWeight {
            #[serde(
                serialize_with = "serialize_bound",
                deserialize_with = "deserialize_bound"
            )]
            lower_bound: Bound<u64>,
            #[serde(
                serialize_with = "serialize_bound",
                deserialize_with = "deserialize_bound"
            )]
            upper_bound: Bound<u64>,
        },
        FuzzyTerm {
            value: String,
            distance: Option<u8>,
            transposition_cost_one: Option<bool>,
            prefix: Option<bool>,
        },
        Match {
            value: String,
            tokenizer: Option<serde_json::Value>,
            distance: Option<u8>,
            transposition_cost_one: Option<bool>,
            prefix: Option<bool>,
            conjunction_mode: Option<bool>,
        },
        MatchArray {
            tokens: Vec<String>,
            distance: Option<u8>,
            transposition_cost_one: Option<bool>,
            prefix: Option<bool>,
            conjunction_mode: Option<bool>,
        },
        Parse {
            query_string: String,
            lenient: Option<bool>,
            conjunction_mode: Option<bool>,
        },
        ParseWithField {
            query_string: String,
            lenient: Option<bool>,
            conjunction_mode: Option<bool>,
            #[serde(skip_serializing_if = "Option::is_none")]
            fuzzy_data: Option<FuzzyData>,
        },
        Phrase {
            phrases: Vec<String>,
            slop: Option<u32>,
        },
        PhraseArray {
            tokens: Vec<String>,
            slop: Option<u32>,
        },
        PhrasePrefix {
            phrases: Vec<String>,
            max_expansions: Option<u32>,
        },
        Proximity {
            left: ProximityClause,
            distance: ProximityDistance,
            right: ProximityClause,
        },
        TokenizedPhrase {
            phrase: String,
            slop: Option<u32>,
        },
        Range {
            #[serde(
                serialize_with = "serialize_bound",
                deserialize_with = "deserialize_bound"
            )]
            lower_bound: Bound<PdbOwnedValue>,
            #[serde(
                serialize_with = "serialize_bound",
                deserialize_with = "deserialize_bound"
            )]
            upper_bound: Bound<PdbOwnedValue>,
        },
        RangeContains {
            #[serde(
                serialize_with = "serialize_bound",
                deserialize_with = "deserialize_bound"
            )]
            lower_bound: Bound<PdbOwnedValue>,
            #[serde(
                serialize_with = "serialize_bound",
                deserialize_with = "deserialize_bound"
            )]
            upper_bound: Bound<PdbOwnedValue>,
        },
        RangeIntersects {
            #[serde(
                serialize_with = "serialize_bound",
                deserialize_with = "deserialize_bound"
            )]
            lower_bound: Bound<PdbOwnedValue>,
            #[serde(
                serialize_with = "serialize_bound",
                deserialize_with = "deserialize_bound"
            )]
            upper_bound: Bound<PdbOwnedValue>,
        },
        RangeTerm {
            value: PdbOwnedValue,
        },
        RangeWithin {
            #[serde(
                serialize_with = "serialize_bound",
                deserialize_with = "deserialize_bound"
            )]
            lower_bound: Bound<PdbOwnedValue>,
            #[serde(
                serialize_with = "serialize_bound",
                deserialize_with = "deserialize_bound"
            )]
            upper_bound: Bound<PdbOwnedValue>,
        },
        Regex {
            pattern: String,
        },
        RegexPhrase {
            regexes: Vec<String>,
            slop: Option<u32>,
            max_expansions: Option<u32>,
        },
        Term {
            value: PdbOwnedValue,
        },
        TermSet {
            terms: Vec<PdbOwnedValue>,
        },
    }
}

impl pdb::Query {
    pub fn unclassified_string(s: &str) -> pdb::Query {
        pdb::Query::UnclassifiedString {
            string: s.to_string(),
            fuzzy_data: None,
            slop_data: None,
        }
    }

    pub fn unclassified_string_with_fuzz(s: &str, fuzz: FuzzyData) -> pdb::Query {
        pdb::Query::UnclassifiedString {
            string: s.to_string(),
            fuzzy_data: Some(fuzz),
            slop_data: None,
        }
    }

    pub fn unclassified_string_with_slop(s: &str, slop: SlopData) -> pdb::Query {
        pdb::Query::UnclassifiedString {
            string: s.to_string(),
            fuzzy_data: None,
            slop_data: Some(slop),
        }
    }

    pub fn apply_fuzzy_data(&mut self, new_fuzzy_data: Option<FuzzyData>) {
        if new_fuzzy_data.is_none() {
            return;
        }
        let new_fuzzy_data = new_fuzzy_data.unwrap();
        match self {
            pdb::Query::UnclassifiedString { fuzzy_data, .. } => {
                *fuzzy_data = Some(new_fuzzy_data);
            }
            pdb::Query::UnclassifiedArray { fuzzy_data, .. } => {
                *fuzzy_data = Some(new_fuzzy_data);
            }

            pdb::Query::Term {
                value: PdbOwnedValue::Str(value),
            } => {
                *self = pdb::Query::FuzzyTerm {
                    value: value.to_string(),
                    distance: Some(new_fuzzy_data.distance),
                    transposition_cost_one: Some(new_fuzzy_data.transposition_cost_one),
                    prefix: Some(new_fuzzy_data.prefix),
                }
            }

            pdb::Query::TermSet { terms } => {
                // must convert to an OR'd set of FuzzyTerms
                let mut fuzzy_terms = Vec::with_capacity(terms.len());
                for term in terms {
                    let PdbOwnedValue::Str(term) = term else {
                        continue;
                    };
                    fuzzy_terms.push(term.clone());
                }

                *self = pdb::Query::MatchArray {
                    tokens: fuzzy_terms,
                    distance: Some(new_fuzzy_data.distance),
                    transposition_cost_one: Some(new_fuzzy_data.transposition_cost_one),
                    prefix: Some(new_fuzzy_data.prefix),
                    conjunction_mode: Some(false),
                }
            }

            pdb::Query::FuzzyTerm {
                distance,
                transposition_cost_one,
                prefix,
                ..
            } => {
                *distance = Some(new_fuzzy_data.distance);
                *transposition_cost_one = Some(new_fuzzy_data.transposition_cost_one);
                *prefix = Some(new_fuzzy_data.prefix);
            }

            pdb::Query::Match {
                distance,
                transposition_cost_one,
                prefix,
                ..
            }
            | pdb::Query::MatchArray {
                distance,
                transposition_cost_one,
                prefix,
                ..
            } => {
                *distance = Some(new_fuzzy_data.distance);
                *transposition_cost_one = Some(new_fuzzy_data.transposition_cost_one);
                *prefix = Some(new_fuzzy_data.prefix);
            }

            pdb::Query::ParseWithField { fuzzy_data, .. } => *fuzzy_data = Some(new_fuzzy_data),

            pdb::Query::ScoreAdjusted { query, .. } => {
                query.apply_fuzzy_data(Some(new_fuzzy_data));
            }

            _ => panic!("query type is not compatible with fuzzy"),
        }
    }

    pub fn apply_slop_data(&mut self, new_slop_data: Option<SlopData>) {
        if new_slop_data.is_none() {
            return;
        }
        let new_slop_data = new_slop_data.unwrap();
        match self {
            pdb::Query::UnclassifiedString { slop_data, .. } => {
                *slop_data = Some(new_slop_data);
            }
            pdb::Query::UnclassifiedArray { slop_data, .. } => {
                *slop_data = Some(new_slop_data);
            }
            pdb::Query::Phrase { slop, .. } => *slop = Some(new_slop_data.slop),

            pdb::Query::PhraseArray { slop, .. } => *slop = Some(new_slop_data.slop),

            pdb::Query::TokenizedPhrase { slop, .. } => *slop = Some(new_slop_data.slop),

            _ => panic!("query type is not compatible with slop"),
        }
    }

    pub fn into_tantivy_query<QueryParserCtor: Fn() -> QueryParser>(
        self,
        field: FieldName,
        schema: &SearchIndexSchema,
        index_created_by_version: Option<Version>,
        parser: &QueryParserCtor,
        searcher: &Searcher,
    ) -> anyhow::Result<Box<dyn TantivyQuery>> {
        let query: Box<dyn TantivyQuery> = match self {
            pdb::Query::All => Box::new(AllQuery),
            pdb::Query::Empty => Box::new(EmptyQuery),

            pdb::Query::UnclassifiedString { .. } => {
                // this would indicate a problem with the various operator SUPPORT functions failing
                // to convert the UnclassifiedString into the pdb::Query variant they require
                unreachable!(
                    "pdb::Query::UnclassifiedString cannot be converted into a TantivyQuery"
                )
            }
            pdb::Query::UnclassifiedArray { .. } => {
                // this would indicate a problem with the various operator SUPPORT functions failing
                // to convert the UnclassifiedArray into the pdb::Query variant they require
                unreachable!(
                    "pdb::Query::UnclassifiedArray cannot be converted into a TantivyQuery"
                )
            }
            pdb::Query::Exists => exists(field, searcher),
            pdb::Query::ScoreAdjusted { query, score } => score_adjust_query(
                field,
                schema,
                index_created_by_version,
                parser,
                searcher,
                *query,
                score.expect("score adjustment value should have been set"),
            )?,
            pdb::Query::FastFieldRangeWeight {
                lower_bound,
                upper_bound,
            } => fast_field_range_weight(&field, schema, lower_bound, upper_bound),
            pdb::Query::FuzzyTerm {
                value,
                distance,
                transposition_cost_one,
                prefix,
            } => fuzzy_term(
                &field,
                schema,
                index_created_by_version,
                value,
                distance,
                transposition_cost_one,
                prefix,
            )?,
            pdb::Query::Match {
                value,
                tokenizer,
                distance,
                transposition_cost_one,
                prefix,
                conjunction_mode,
            } => match_query(
                &field,
                schema,
                index_created_by_version,
                searcher,
                &value,
                tokenizer,
                distance,
                transposition_cost_one,
                prefix,
                conjunction_mode,
            )?,
            pdb::Query::MatchArray {
                tokens: value,
                distance,
                transposition_cost_one,
                prefix,
                conjunction_mode,
            } => match_array_query(
                &field,
                schema,
                index_created_by_version,
                value,
                distance,
                transposition_cost_one,
                prefix,
                conjunction_mode,
            )?,
            pdb::Query::Parse {
                query_string,
                lenient,
                conjunction_mode,
            } => parse(
                parser,
                schema,
                index_created_by_version,
                query_string,
                lenient,
                conjunction_mode,
            )?,
            pdb::Query::ParseWithField {
                query_string,
                lenient,
                conjunction_mode,
                fuzzy_data,
            } => parse_with_field(
                &field,
                parser,
                schema,
                index_created_by_version,
                query_string,
                lenient,
                conjunction_mode,
                fuzzy_data,
            )?,

            pdb::Query::Phrase { phrases, slop } => phrase(
                &field,
                schema,
                index_created_by_version,
                searcher,
                phrases,
                slop,
            )?,
            pdb::Query::PhraseArray { tokens, slop } => {
                phrase_array(&field, schema, index_created_by_version, tokens, slop)?
            }

            pdb::Query::PhrasePrefix {
                phrases,
                max_expansions,
            } => phrase_prefix(
                &field,
                schema,
                index_created_by_version,
                phrases,
                max_expansions,
            )?,
            pdb::Query::Proximity {
                left,
                distance,
                right,
            } => proximity(&field, schema, left, distance, right)?,
            pdb::Query::TokenizedPhrase { phrase, slop } => tokenized_phrase(
                &field,
                schema,
                index_created_by_version,
                searcher,
                &phrase,
                slop,
            )?,
            pdb::Query::Range {
                lower_bound,
                upper_bound,
            } => range(
                &field,
                schema,
                index_created_by_version,
                lower_bound,
                upper_bound,
            )?,
            pdb::Query::RangeContains {
                lower_bound,
                upper_bound,
            } => range_contains(
                &field,
                schema,
                index_created_by_version,
                lower_bound,
                upper_bound,
            )?,
            pdb::Query::RangeIntersects {
                lower_bound,
                upper_bound,
            } => range_intersects(
                &field,
                schema,
                index_created_by_version,
                lower_bound,
                upper_bound,
            )?,
            pdb::Query::RangeTerm { value } => {
                range_term(&field, schema, index_created_by_version, &value)?
            }
            pdb::Query::RangeWithin {
                lower_bound,
                upper_bound,
            } => range_within(
                &field,
                schema,
                index_created_by_version,
                lower_bound,
                upper_bound,
            )?,
            pdb::Query::Regex { pattern } => regex(&field, schema, &pattern)?,
            pdb::Query::RegexPhrase {
                regexes,
                slop,
                max_expansions,
            } => regex_phrase(&field, schema, regexes, slop, max_expansions)?,
            pdb::Query::Term { value } => term(field, schema, index_created_by_version, &value)?,
            pdb::Query::TermSet { terms } => {
                term_set(field, schema, index_created_by_version, terms)?
            }
        };

        Ok(query)
    }

    pub fn needs_tokenizer(&self) -> bool {
        match self {
            pdb::Query::All
            | pdb::Query::Empty
            | pdb::Query::Exists
            | pdb::Query::Term { .. }
            | pdb::Query::TermSet { .. }
            | pdb::Query::FuzzyTerm { .. }
            | pdb::Query::Range { .. }
            | pdb::Query::RangeContains { .. }
            | pdb::Query::RangeIntersects { .. }
            | pdb::Query::RangeTerm { .. }
            | pdb::Query::RangeWithin { .. }
            | pdb::Query::FastFieldRangeWeight { .. }
            | pdb::Query::Regex { .. }
            | pdb::Query::RegexPhrase { .. }
            | pdb::Query::PhrasePrefix { .. }
            | pdb::Query::Proximity { .. }
            | pdb::Query::MatchArray { .. }
            | pdb::Query::PhraseArray { .. } => false,

            pdb::Query::Parse { .. }
            | pdb::Query::ParseWithField { .. }
            | pdb::Query::Match { .. }
            | pdb::Query::Phrase { .. }
            | pdb::Query::TokenizedPhrase { .. }
            | pdb::Query::UnclassifiedString { .. }
            | pdb::Query::UnclassifiedArray { .. } => true,

            pdb::Query::ScoreAdjusted { query, .. } => query.needs_tokenizer(),
        }
    }

    /// Returns `true` if constructing a Tantivy Scorer for this query type is expensive.
    /// Fuzzy term and regex queries require building DFAs/automata and scanning the term
    /// dictionary during scorer construction, which can be too costly for planner selectivity
    /// estimation.
    ///
    /// Range queries and Match-with-distance are intentionally excluded: range queries on
    /// numeric fast fields are cheap to score, and their selectivity is highly data-dependent
    /// making heuristics unreliable. Inaccurate heuristics for these types cause plan
    /// regressions (e.g. wrong join strategies, wrong append methods).
    pub fn is_expensive_to_estimate(&self) -> bool {
        match self {
            pdb::Query::FuzzyTerm { .. }
            | pdb::Query::Regex { .. }
            | pdb::Query::RegexPhrase { .. } => true,

            pdb::Query::ParseWithField { fuzzy_data, .. } => fuzzy_data.is_some(),

            pdb::Query::ScoreAdjusted { query, .. } => query.is_expensive_to_estimate(),

            _ => false,
        }
    }

    /// Returns a heuristic selectivity for this query type, avoiding expensive scorer construction.
    pub fn selectivity_heuristic(&self) -> f64 {
        use crate::{FUZZY_HIGH_SELECTIVITY, FUZZY_LOW_SELECTIVITY, REGEX_SELECTIVITY};

        match self {
            pdb::Query::FuzzyTerm { distance, .. } => {
                let dist = distance.unwrap_or(1);
                if dist <= 1 {
                    FUZZY_LOW_SELECTIVITY
                } else {
                    FUZZY_HIGH_SELECTIVITY
                }
            }

            pdb::Query::ParseWithField { .. } => FUZZY_LOW_SELECTIVITY,

            pdb::Query::Regex { .. } | pdb::Query::RegexPhrase { .. } => REGEX_SELECTIVITY,

            pdb::Query::ScoreAdjusted { query, .. } => query.selectivity_heuristic(),

            _ => crate::UNKNOWN_SELECTIVITY,
        }
    }
}

impl InOutFuncs for pdb::Query {
    fn input(input: &CStr) -> Self
    where
        Self: Sized,
    {
        if let Ok(from_json) = serde_json::from_slice::<pdb::Query>(input.to_bytes()) {
            from_json
        } else {
            // assume it's just a string and write it as a "match"
            pdb::Query::UnclassifiedString {
                string: input
                    .to_str()
                    .expect("input should be valid UTF8")
                    .to_string(),
                fuzzy_data: None,
                slop_data: None,
            }
        }
    }

    fn output(&self, buffer: &mut StringInfo) {
        serde_json::to_writer(buffer, self).unwrap();
    }
}

fn score_adjust_query<QueryParserCtor: Fn() -> QueryParser>(
    field: FieldName,
    schema: &SearchIndexSchema,
    index_created_by_version: Option<Version>,
    parser: &QueryParserCtor,
    searcher: &Searcher,
    query: pdb::Query,
    score: ScoreAdjustStyle,
) -> anyhow::Result<Box<dyn TantivyQuery>> {
    let query =
        query.into_tantivy_query(field, schema, index_created_by_version, parser, searcher)?;
    match score {
        ScoreAdjustStyle::Boost(boost) => Ok(Box::new(BoostQuery::new(query, boost))),
        ScoreAdjustStyle::Const(score) => Ok(Box::new(ConstScoreQuery::new(query, score))),
    }
}

fn proximity(
    field: &FieldName,
    schema: &SearchIndexSchema,
    left: ProximityClause,
    distance: ProximityDistance,
    right: ProximityClause,
) -> anyhow::Result<Box<dyn Query>> {
    if left.is_empty() || right.is_empty() {
        return Ok(Box::new(EmptyQuery));
    }

    let search_field = schema
        .search_field(field.root())
        .ok_or(QueryError::NonIndexedField(field.clone()))?
        .with_positions()?;

    let prox = ProximityQuery::new(search_field.field(), left, distance, right);
    Ok(Box::new(prox))
}

fn term_set(
    field: FieldName,
    schema: &SearchIndexSchema,
    index_created_by_version: Option<Version>,
    terms: Vec<PdbOwnedValue>,
) -> anyhow::Result<Box<dyn TantivyQuery>> {
    let search_field = schema
        .search_field(&field)
        .expect("field should exist in schema");
    let field_type = search_field.field_entry().field_type();
    let tantivy_field = search_field.field();
    let search_field_type = search_field.field_type();

    // Convert terms based on field type (uses same logic as term())
    let converted_terms: Vec<PdbOwnedValue> = terms
        .into_iter()
        .map(|term| {
            convert_value_for_field(term, &search_field_type).unwrap_or(PdbOwnedValue::Null)
        })
        .collect();

    Ok(Box::new(TermSetQuery::new(
        converted_terms.into_iter().map(|term| {
            value_to_term(
                tantivy_field,
                &term,
                field_type,
                field.path().as_deref(),
                index_created_by_version,
            )
            .expect("could not convert argument to search term")
        }),
    )))
}

fn term(
    field: FieldName,
    schema: &SearchIndexSchema,
    index_created_by_version: Option<Version>,
    value: &PdbOwnedValue,
) -> anyhow::Result<Box<dyn TantivyQuery>> {
    let record_option = IndexRecordOption::WithFreqsAndPositions;
    let search_field = schema
        .search_field(field.root())
        .ok_or(QueryError::NonIndexedField(field.clone()))?;
    let field_type = search_field.field_entry().field_type();
    let search_field_type = search_field.field_type();

    // Convert value based on field type (handles NUMERIC scaling, JSON types, etc.)
    let value = convert_value_for_field(value.clone(), &search_field_type)?;

    let term = value_to_term(
        search_field.field(),
        &value,
        field_type,
        field.path().as_deref(),
        index_created_by_version,
    )?;

    Ok(Box::new(TermQuery::new(term, record_option.into())))
}

fn regex_phrase(
    field: &FieldName,
    schema: &SearchIndexSchema,
    regexes: Vec<String>,
    slop: Option<u32>,
    max_expansions: Option<u32>,
) -> anyhow::Result<Box<dyn TantivyQuery>> {
    let search_field = schema
        .search_field(field.root())
        .ok_or(QueryError::NonIndexedField(field.clone()))?
        .with_positions()?;

    let mut query = RegexPhraseQuery::new(search_field.field(), regexes);

    if let Some(slop) = slop {
        query.set_slop(slop)
    }
    if let Some(max_expansions) = max_expansions {
        query.set_max_expansions(max_expansions)
    }
    Ok(Box::new(query))
}

fn regex(
    field: &FieldName,
    schema: &SearchIndexSchema,
    pattern: &str,
) -> anyhow::Result<Box<dyn TantivyQuery>> {
    let search_field = schema
        .search_field(field.root())
        .ok_or(QueryError::NonIndexedField(field.clone()))?;

    Ok(Box::new(
        RegexQuery::from_pattern(pattern, search_field.field())
            .map_err(|err| QueryError::RegexError(err, pattern.to_string()))?,
    ))
}

/// Note: For JSON numeric fast field limitations, see documentation on [`range`].
fn range_within(
    field: &FieldName,
    schema: &SearchIndexSchema,
    index_created_by_version: Option<Version>,
    lower_bound: Bound<PdbOwnedValue>,
    upper_bound: Bound<PdbOwnedValue>,
) -> anyhow::Result<Box<dyn TantivyQuery>> {
    let search_field = schema
        .search_field(field.root())
        .ok_or(QueryError::NonIndexedField(field.clone()))?;
    let typeoid = search_field.field_type().typeoid();
    let (lower_bound, upper_bound) = check_range_bounds(typeoid, lower_bound, upper_bound)?;

    let range_field = RangeField::new(search_field.field());

    let mut satisfies_lower_bound: Vec<(Occur, Box<dyn TantivyQuery>)> = vec![];
    let mut satisfies_upper_bound: Vec<(Occur, Box<dyn TantivyQuery>)> = vec![];

    match lower_bound {
        Bound::Excluded(ref lower) => {
            satisfies_lower_bound.push((
                Occur::Must,
                Box::new(BooleanQuery::new(vec![(
                    Occur::Must,
                    Box::new(range_field.compare_lower_bound(
                        lower,
                        Comparison::GreaterThanOrEqual,
                        index_created_by_version,
                    )?),
                )])),
            ));
        }
        Bound::Included(ref lower) => satisfies_lower_bound.push((
            Occur::Must,
            (Box::new(BooleanQuery::new(vec![
                (
                    Occur::Should,
                    Box::new(BooleanQuery::new(vec![
                        (
                            Occur::Must,
                            Box::new(range_field.compare_lower_bound(
                                lower,
                                Comparison::GreaterThan,
                                index_created_by_version,
                            )?),
                        ),
                        (
                            Occur::Must,
                            Box::new(range_field.lower_bound_inclusive(false)?),
                        ),
                    ])),
                ),
                (
                    Occur::Should,
                    Box::new(BooleanQuery::new(vec![
                        (
                            Occur::Must,
                            Box::new(range_field.compare_lower_bound(
                                lower,
                                Comparison::GreaterThanOrEqual,
                                index_created_by_version,
                            )?),
                        ),
                        (
                            Occur::Must,
                            Box::new(range_field.lower_bound_inclusive(true)?),
                        ),
                    ])),
                ),
            ]))),
        )),
        _ => {}
    }

    match upper_bound {
        Bound::Excluded(ref upper) => {
            satisfies_upper_bound.push((
                Occur::Must,
                Box::new(BooleanQuery::new(vec![(
                    Occur::Must,
                    Box::new(range_field.compare_upper_bound(
                        upper,
                        Comparison::LessThanOrEqual,
                        index_created_by_version,
                    )?),
                )])),
            ));
        }
        Bound::Included(ref upper) => satisfies_upper_bound.push((
            Occur::Must,
            (Box::new(BooleanQuery::new(vec![
                (
                    Occur::Should,
                    Box::new(BooleanQuery::new(vec![
                        (
                            Occur::Must,
                            Box::new(range_field.compare_upper_bound(
                                upper,
                                Comparison::LessThan,
                                index_created_by_version,
                            )?),
                        ),
                        (
                            Occur::Must,
                            Box::new(range_field.upper_bound_inclusive(false)?),
                        ),
                    ])),
                ),
                (
                    Occur::Should,
                    Box::new(BooleanQuery::new(vec![
                        (
                            Occur::Must,
                            Box::new(range_field.compare_upper_bound(
                                upper,
                                Comparison::LessThanOrEqual,
                                index_created_by_version,
                            )?),
                        ),
                        (
                            Occur::Must,
                            Box::new(range_field.upper_bound_inclusive(true)?),
                        ),
                    ])),
                ),
            ]))),
        )),
        _ => {}
    }

    let satisfies_lower_bound = BooleanQuery::new(vec![
        (
            Occur::Should,
            Box::new(range_field.lower_bound_unbounded(true)?),
        ),
        (
            Occur::Should,
            Box::new(BooleanQuery::new(satisfies_lower_bound)),
        ),
    ]);

    let satisfies_upper_bound = BooleanQuery::new(vec![
        (
            Occur::Should,
            Box::new(range_field.upper_bound_unbounded(true)?),
        ),
        (
            Occur::Should,
            Box::new(BooleanQuery::new(satisfies_upper_bound)),
        ),
    ]);

    let is_empty = match (lower_bound, upper_bound) {
        (Bound::Included(lower), Bound::Excluded(upper)) => lower == upper,
        _ => false,
    };

    Ok(if is_empty {
        Box::new(range_field.exists()?)
    } else {
        Box::new(BooleanQuery::new(vec![
            (Occur::Must, Box::new(satisfies_lower_bound)),
            (Occur::Must, Box::new(satisfies_upper_bound)),
        ]))
    })
}

fn range_term(
    field: &FieldName,
    schema: &SearchIndexSchema,
    index_created_by_version: Option<Version>,
    value: &PdbOwnedValue,
) -> anyhow::Result<Box<dyn TantivyQuery>> {
    let search_field = schema
        .search_field(field.root())
        .ok_or(QueryError::NonIndexedField(field.clone()))?;

    // Convert numeric values to appropriate types based on range field type.
    // Range fields are indexed as JSON with specific element types:
    // - INT4RANGEOID, INT8RANGEOID: indexed as i32/i64 → convert to I64
    // - NUMRANGEOID: indexed as hex-encoded sortable bytes (see SortableDecimal) → convert to hex string
    // - Date/time ranges: handling is determined by examining the schema
    let value = convert_value_for_range_field(value.clone(), &search_field.field_type());

    let range_field = RangeField::new(search_field.field());

    let satisfies_lower_bound = BooleanQuery::new(vec![
        (
            Occur::Should,
            Box::new(range_field.lower_bound_unbounded(true)?),
        ),
        (
            Occur::Should,
            Box::new(BooleanQuery::new(vec![
                (
                    Occur::Should,
                    Box::new(BooleanQuery::new(vec![
                        (
                            Occur::Must,
                            Box::new(range_field.lower_bound_inclusive(true)?),
                        ),
                        (
                            Occur::Must,
                            Box::new(range_field.compare_lower_bound(
                                &value,
                                Comparison::GreaterThanOrEqual,
                                index_created_by_version,
                            )?),
                        ),
                    ])),
                ),
                (
                    Occur::Should,
                    Box::new(BooleanQuery::new(vec![
                        (
                            Occur::Must,
                            Box::new(range_field.lower_bound_inclusive(false)?),
                        ),
                        (
                            Occur::Must,
                            Box::new(range_field.compare_lower_bound(
                                &value,
                                Comparison::GreaterThan,
                                index_created_by_version,
                            )?),
                        ),
                    ])),
                ),
            ])),
        ),
    ]);

    let satisfies_upper_bound = BooleanQuery::new(vec![
        (
            Occur::Should,
            Box::new(range_field.upper_bound_unbounded(true)?),
        ),
        (
            Occur::Should,
            Box::new(BooleanQuery::new(vec![
                (
                    Occur::Should,
                    Box::new(BooleanQuery::new(vec![
                        (
                            Occur::Must,
                            Box::new(range_field.upper_bound_inclusive(true)?),
                        ),
                        (
                            Occur::Must,
                            Box::new(range_field.compare_upper_bound(
                                &value,
                                Comparison::LessThanOrEqual,
                                index_created_by_version,
                            )?),
                        ),
                    ])),
                ),
                (
                    Occur::Should,
                    Box::new(BooleanQuery::new(vec![
                        (
                            Occur::Must,
                            Box::new(range_field.upper_bound_inclusive(false)?),
                        ),
                        (
                            Occur::Must,
                            Box::new(range_field.compare_upper_bound(
                                &value,
                                Comparison::LessThan,
                                index_created_by_version,
                            )?),
                        ),
                    ])),
                ),
            ])),
        ),
    ]);

    Ok(Box::new(BooleanQuery::new(vec![
        (Occur::Must, Box::new(satisfies_lower_bound)),
        (Occur::Must, Box::new(satisfies_upper_bound)),
    ])))
}

fn range_intersects(
    field: &FieldName,
    schema: &SearchIndexSchema,
    index_created_by_version: Option<Version>,
    lower_bound: Bound<PdbOwnedValue>,
    upper_bound: Bound<PdbOwnedValue>,
) -> anyhow::Result<Box<dyn TantivyQuery>> {
    let search_field = schema
        .search_field(field.root())
        .ok_or(QueryError::NonIndexedField(field.clone()))?;
    let typeoid = search_field.field_type().typeoid();

    let (lower_bound, upper_bound) = check_range_bounds(typeoid, lower_bound, upper_bound)?;
    let range_field = RangeField::new(search_field.field());

    let mut satisfies_lower_bound: Vec<(Occur, Box<dyn TantivyQuery>)> = vec![];
    let mut satisfies_upper_bound: Vec<(Occur, Box<dyn TantivyQuery>)> = vec![];

    match lower_bound {
        Bound::Excluded(ref lower) => {
            satisfies_lower_bound.push((
                Occur::Must,
                Box::new(BooleanQuery::new(vec![(
                    Occur::Must,
                    Box::new(range_field.compare_upper_bound(
                        lower,
                        Comparison::LessThan,
                        index_created_by_version,
                    )?),
                )])),
            ));
        }
        Bound::Included(ref lower) => satisfies_lower_bound.push((
            Occur::Must,
            (Box::new(BooleanQuery::new(vec![
                (
                    Occur::Should,
                    Box::new(BooleanQuery::new(vec![
                        (
                            Occur::Must,
                            Box::new(range_field.compare_upper_bound(
                                lower,
                                Comparison::LessThanOrEqual,
                                index_created_by_version,
                            )?),
                        ),
                        (
                            Occur::Must,
                            Box::new(range_field.upper_bound_inclusive(true)?),
                        ),
                    ])),
                ),
                (
                    Occur::Should,
                    Box::new(BooleanQuery::new(vec![
                        (
                            Occur::Must,
                            Box::new(range_field.compare_upper_bound(
                                lower,
                                Comparison::LessThan,
                                index_created_by_version,
                            )?),
                        ),
                        (
                            Occur::Must,
                            Box::new(range_field.upper_bound_inclusive(false)?),
                        ),
                    ])),
                ),
            ]))),
        )),
        Bound::Unbounded => {
            satisfies_lower_bound.push((Occur::Should, Box::new(range_field.exists()?)))
        }
    }

    match upper_bound {
        Bound::Excluded(ref upper) => {
            satisfies_upper_bound.push((
                Occur::Must,
                Box::new(BooleanQuery::new(vec![(
                    Occur::Must,
                    Box::new(range_field.compare_lower_bound(
                        upper,
                        Comparison::GreaterThan,
                        index_created_by_version,
                    )?),
                )])),
            ));
        }
        Bound::Included(ref upper) => satisfies_upper_bound.push((
            Occur::Must,
            (Box::new(BooleanQuery::new(vec![
                (
                    Occur::Should,
                    Box::new(BooleanQuery::new(vec![
                        (
                            Occur::Must,
                            Box::new(range_field.compare_lower_bound(
                                upper,
                                Comparison::GreaterThanOrEqual,
                                index_created_by_version,
                            )?),
                        ),
                        (
                            Occur::Must,
                            Box::new(range_field.lower_bound_inclusive(true)?),
                        ),
                    ])),
                ),
                (
                    Occur::Should,
                    Box::new(BooleanQuery::new(vec![
                        (
                            Occur::Must,
                            Box::new(range_field.compare_lower_bound(
                                upper,
                                Comparison::GreaterThan,
                                index_created_by_version,
                            )?),
                        ),
                        (
                            Occur::Must,
                            Box::new(range_field.lower_bound_inclusive(false)?),
                        ),
                    ])),
                ),
            ]))),
        )),
        Bound::Unbounded => {
            satisfies_upper_bound.push((Occur::Should, Box::new(range_field.exists()?)))
        }
    }

    let satisfies_lower_bound = BooleanQuery::new(vec![
        (
            Occur::Should,
            Box::new(range_field.upper_bound_unbounded(true)?),
        ),
        (
            Occur::Should,
            Box::new(BooleanQuery::new(satisfies_lower_bound)),
        ),
    ]);

    let satisfies_upper_bound = BooleanQuery::new(vec![
        (
            Occur::Should,
            Box::new(range_field.lower_bound_unbounded(true)?),
        ),
        (
            Occur::Should,
            Box::new(BooleanQuery::new(satisfies_upper_bound)),
        ),
    ]);

    let is_empty = match (lower_bound, upper_bound) {
        (Bound::Included(lower), Bound::Excluded(upper)) => lower == upper,
        _ => false,
    };

    Ok(if is_empty {
        Box::new(EmptyQuery)
    } else {
        Box::new(BooleanQuery::new(vec![
            (Occur::Must, Box::new(satisfies_lower_bound)),
            (Occur::Must, Box::new(satisfies_upper_bound)),
            (Occur::Must, Box::new(range_field.empty(false)?)),
        ]))
    })
}

fn range_contains(
    field: &FieldName,
    schema: &SearchIndexSchema,
    index_created_by_version: Option<Version>,
    lower_bound: Bound<PdbOwnedValue>,
    upper_bound: Bound<PdbOwnedValue>,
) -> anyhow::Result<Box<dyn TantivyQuery>> {
    let search_field = schema
        .search_field(field.root())
        .ok_or(QueryError::NonIndexedField(field.clone()))?;
    let typeoid = search_field.field_type().typeoid();
    let (lower_bound, upper_bound) = check_range_bounds(typeoid, lower_bound, upper_bound)?;
    let range_field = RangeField::new(search_field.field());

    let mut satisfies_lower_bound: Vec<(Occur, Box<dyn TantivyQuery>)> = vec![];
    let mut satisfies_upper_bound: Vec<(Occur, Box<dyn TantivyQuery>)> = vec![];

    match lower_bound {
        Bound::Included(lower) => {
            satisfies_lower_bound.push((
                Occur::Must,
                Box::new(BooleanQuery::new(vec![(
                    Occur::Must,
                    Box::new(range_field.compare_lower_bound(
                        &lower,
                        Comparison::LessThanOrEqual,
                        index_created_by_version,
                    )?),
                )])),
            ));
        }
        Bound::Excluded(lower) => satisfies_lower_bound.push((
            Occur::Must,
            (Box::new(BooleanQuery::new(vec![
                (
                    Occur::Should,
                    Box::new(BooleanQuery::new(vec![
                        (
                            Occur::Must,
                            Box::new(range_field.compare_lower_bound(
                                &lower,
                                Comparison::LessThan,
                                index_created_by_version,
                            )?),
                        ),
                        (
                            Occur::Must,
                            Box::new(range_field.lower_bound_inclusive(true)?),
                        ),
                    ])),
                ),
                (
                    Occur::Should,
                    Box::new(BooleanQuery::new(vec![
                        (
                            Occur::Must,
                            Box::new(range_field.compare_lower_bound(
                                &lower,
                                Comparison::LessThanOrEqual,
                                index_created_by_version,
                            )?),
                        ),
                        (
                            Occur::Must,
                            Box::new(range_field.lower_bound_inclusive(false)?),
                        ),
                    ])),
                ),
            ]))),
        )),
        _ => satisfies_lower_bound.push((Occur::Should, Box::new(range_field.exists()?))),
    }

    match upper_bound {
        Bound::Included(upper) => {
            satisfies_upper_bound.push((
                Occur::Must,
                Box::new(BooleanQuery::new(vec![(
                    Occur::Must,
                    Box::new(range_field.compare_upper_bound(
                        &upper,
                        Comparison::GreaterThanOrEqual,
                        index_created_by_version,
                    )?),
                )])),
            ));
        }
        Bound::Excluded(upper) => satisfies_upper_bound.push((
            Occur::Must,
            (Box::new(BooleanQuery::new(vec![
                (
                    Occur::Should,
                    Box::new(BooleanQuery::new(vec![
                        (
                            Occur::Must,
                            Box::new(range_field.compare_upper_bound(
                                &upper,
                                Comparison::GreaterThan,
                                index_created_by_version,
                            )?),
                        ),
                        (
                            Occur::Must,
                            Box::new(range_field.upper_bound_inclusive(true)?),
                        ),
                    ])),
                ),
                (
                    Occur::Should,
                    Box::new(BooleanQuery::new(vec![
                        (
                            Occur::Must,
                            Box::new(range_field.compare_upper_bound(
                                &upper,
                                Comparison::GreaterThanOrEqual,
                                index_created_by_version,
                            )?),
                        ),
                        (
                            Occur::Must,
                            Box::new(range_field.upper_bound_inclusive(false)?),
                        ),
                    ])),
                ),
            ]))),
        )),
        _ => satisfies_upper_bound.push((Occur::Should, Box::new(range_field.exists()?))),
    }

    let satisfies_lower_bound = BooleanQuery::new(vec![
        (Occur::Should, Box::new(range_field.empty(true)?)),
        (
            Occur::Should,
            Box::new(BooleanQuery::new(satisfies_lower_bound)),
        ),
    ]);

    let satisfies_upper_bound = BooleanQuery::new(vec![
        (Occur::Should, Box::new(range_field.empty(true)?)),
        (
            Occur::Should,
            Box::new(BooleanQuery::new(satisfies_upper_bound)),
        ),
    ]);

    Ok(Box::new(BooleanQuery::new(vec![
        (Occur::Must, Box::new(satisfies_lower_bound)),
        (Occur::Must, Box::new(satisfies_upper_bound)),
    ])))
}

/// Creates a range query for the given field and bounds.
///
/// # JSON Numeric Range Queries and Fast Fields
///
/// For JSON fields, Tantivy requires fast fields for range queries (returns error otherwise).
/// Fast field storage has important limitations for JSON numeric values:
///
/// - Each JSON path gets ONE fast field column with ONE numeric type (I64, U64, or F64)
/// - Column type is determined at index time based on values stored:
///   - All integers that fit in i64 → I64 column
///   - All non-negative integers, some exceeding i64::MAX → U64 column
///   - Any float value OR mix of negative + large positive (≥ 2^63) → F64 column
/// - When column is F64, integers > 2^53 lose precision (e.g., 9007199254740993 → 9007199254740992.0)
///
/// At query time, Tantivy discovers the actual column type and converts query bounds accordingly
/// (see `search_on_json_numerical_field` in tantivy's range_query_fastfield.rs).
///
/// References:
/// - Fast field column type selection: tantivy columnar/src/columnar/writer/column_writers.rs
fn range(
    field: &FieldName,
    schema: &SearchIndexSchema,
    index_created_by_version: Option<Version>,
    lower_bound: Bound<PdbOwnedValue>,
    upper_bound: Bound<PdbOwnedValue>,
) -> anyhow::Result<Box<dyn TantivyQuery>> {
    let search_field = schema
        .search_field(field.root())
        .ok_or(QueryError::NonIndexedField(field.clone()))?;
    let field_type = search_field.field_entry().field_type();
    let typeoid = search_field.field_type().typeoid();
    let search_field_type = search_field.field_type();

    // Handle NUMERIC field types with special storage strategies
    // Also handle string-encoded numeric values for JSON and other numeric fields
    let (lower_bound, upper_bound) = match search_field_type {
        SearchFieldType::Numeric64(_, scale) => {
            // Scale bounds for I64 fixed-point storage
            let lower = scale_numeric_bound(lower_bound, scale)?;
            let upper = scale_numeric_bound(upper_bound, scale)?;
            (lower, upper)
        }
        SearchFieldType::NumericBytes(..) => {
            // Convert bounds to lexicographically sortable bytes
            let lower = numeric_bound_to_bytes(lower_bound)?;
            let upper = numeric_bound_to_bytes(upper_bound)?;
            (lower, upper)
        }
        SearchFieldType::Json(_) => {
            // For JSON fields, convert string numeric values to appropriate JSON types
            let lower = map_bound(lower_bound, string_to_json_numeric);
            let upper = map_bound(upper_bound, string_to_json_numeric);
            check_range_bounds(typeoid, lower, upper)?
        }
        SearchFieldType::I64(_) => {
            let lower = map_bound(lower_bound, string_to_i64);
            let upper = map_bound(upper_bound, string_to_i64);
            check_range_bounds(typeoid, lower, upper)?
        }
        SearchFieldType::U64(_) => {
            let lower = map_bound(lower_bound, string_to_u64);
            let upper = map_bound(upper_bound, string_to_u64);
            check_range_bounds(typeoid, lower, upper)?
        }
        SearchFieldType::F64(_) => {
            let lower = map_bound(lower_bound, string_to_f64);
            let upper = map_bound(upper_bound, string_to_f64);
            check_range_bounds(typeoid, lower, upper)?
        }
        _ => {
            // Standard path for other field types
            let lower_bound = coerce_bound_to_field_type(lower_bound, field_type);
            let upper_bound = coerce_bound_to_field_type(upper_bound, field_type);
            check_range_bounds(typeoid, lower_bound, upper_bound)?
        }
    };

    let lower_bound = match lower_bound {
        Bound::Included(value) => Bound::Included(value_to_term(
            search_field.field(),
            &value,
            field_type,
            field.path().as_deref(),
            index_created_by_version,
        )?),
        Bound::Excluded(value) => Bound::Excluded(value_to_term(
            search_field.field(),
            &value,
            field_type,
            field.path().as_deref(),
            index_created_by_version,
        )?),
        Bound::Unbounded => Bound::Unbounded,
    };

    let upper_bound = match upper_bound {
        Bound::Included(value) => Bound::Included(value_to_term(
            search_field.field(),
            &value,
            field_type,
            field.path().as_deref(),
            index_created_by_version,
        )?),
        Bound::Excluded(value) => Bound::Excluded(value_to_term(
            search_field.field(),
            &value,
            field_type,
            field.path().as_deref(),
            index_created_by_version,
        )?),
        Bound::Unbounded => Bound::Unbounded,
    };

    Ok(Box::new(RangeQuery::new(lower_bound, upper_bound)))
}

fn resolve_search_tokenizer(
    search_field: &SearchField,
    schema: &SearchIndexSchema,
    searcher: &Searcher,
) -> anyhow::Result<tantivy::tokenizer::TextAnalyzer> {
    if let Some(st) = search_field.field_config().search_tokenizer() {
        return Ok(st
            .to_tantivy_tokenizer()
            .expect("field-level search_tokenizer should be a valid tantivy tokenizer"));
    }
    if let Some(ref st) = schema.index_search_tokenizer() {
        return Ok(st
            .to_tantivy_tokenizer()
            .expect("index-level search_tokenizer should be a valid tantivy tokenizer"));
    }
    Ok(searcher.index().tokenizer_for_field(search_field.field())?)
}

fn tokenized_phrase(
    field: &FieldName,
    schema: &SearchIndexSchema,
    index_created_by_version: Option<Version>,
    searcher: &Searcher,
    phrase: &str,
    slop: Option<u32>,
) -> anyhow::Result<Box<dyn TantivyQuery>> {
    let search_field = schema
        .search_field(field.root())
        .ok_or(QueryError::NonIndexedField(field.clone()))?
        .with_positions()?;

    let field_type = search_field.field_entry().field_type();
    let mut tokenizer = resolve_search_tokenizer(&search_field, schema, searcher)?;
    let mut stream = tokenizer.token_stream(phrase);
    let path = field.path();

    let mut tokens = Vec::new();
    while let Some(token) = stream.next() {
        let value = PdbOwnedValue::Str(token.text.clone());
        let term = value_to_term(
            search_field.field(),
            &value,
            field_type,
            path.as_deref(),
            index_created_by_version,
        )?;
        tokens.push(term);
    }
    Ok(if tokens.is_empty() {
        Box::new(EmptyQuery)
    } else if tokens.len() == 1 {
        Box::new(TermQuery::new(
            tokens.remove(0),
            IndexRecordOption::WithFreqs.into(),
        ))
    } else {
        let mut query = PhraseQuery::new(tokens);
        query.set_slop(slop.unwrap_or(0));
        Box::new(query)
    })
}

fn phrase_prefix(
    field: &FieldName,
    schema: &SearchIndexSchema,
    index_created_by_version: Option<Version>,
    phrases: Vec<String>,
    max_expansions: Option<u32>,
) -> anyhow::Result<Box<dyn TantivyQuery>> {
    let search_field = schema
        .search_field(field.root())
        .ok_or(QueryError::NonIndexedField(field.clone()))?
        .with_positions()?;

    let field_type = search_field.field_entry().field_type();
    let terms = phrases.clone().into_iter().map(|phrase| {
        value_to_term(
            search_field.field(),
            &PdbOwnedValue::Str(phrase),
            field_type,
            field.path().as_deref(),
            index_created_by_version,
        )
        .unwrap()
    });
    let mut query = PhrasePrefixQuery::new(terms.collect());
    if let Some(max_expansions) = max_expansions {
        query.set_max_expansions(max_expansions)
    }
    Ok(Box::new(query))
}

fn phrase(
    field: &FieldName,
    schema: &SearchIndexSchema,
    index_created_by_version: Option<Version>,
    searcher: &Searcher,
    phrases: Vec<String>,
    slop: Option<u32>,
) -> anyhow::Result<Box<dyn TantivyQuery>> {
    let search_field = schema
        .search_field(field.root())
        .ok_or(QueryError::NonIndexedField(field.clone()))?
        .with_positions()?;
    let field_type = search_field.field_entry().field_type();

    let mut terms = Vec::new();
    let mut analyzer = resolve_search_tokenizer(&search_field, schema, searcher)?;
    let mut should_warn = false;

    for phrase in phrases.into_iter() {
        let mut stream = analyzer.token_stream(&phrase);
        let len_before = terms.len();

        while stream.advance() {
            let token = stream.token().text.clone();
            let term = value_to_term(
                search_field.field(),
                &PdbOwnedValue::Str(token),
                field_type,
                field.path().as_deref(),
                index_created_by_version,
            )?;

            terms.push(term);
        }

        if len_before + 1 < terms.len() {
            should_warn = true;
        }
    }

    // When tokenizers produce more than one token per phrase, their position may not
    // correctly represent the original query.
    // For example, NgramTokenizer can produce many tokens per word and all of them will
    // have position=0 which won't be correctly interpreted when processing slop
    if should_warn {
        pgrx::warning!("Phrase query with multiple tokens per phrase may not be correctly interpreted. Consider using a different tokenizer or switch to parse/match");
    }

    let mut query = PhraseQuery::new(terms);
    if let Some(slop) = slop {
        query.set_slop(slop)
    }
    Ok(Box::new(query))
}

fn phrase_array(
    field: &FieldName,
    schema: &SearchIndexSchema,
    index_created_by_version: Option<Version>,
    mut tokens: Vec<String>,
    slop: Option<u32>,
) -> anyhow::Result<Box<dyn TantivyQuery>> {
    let search_field = schema
        .search_field(field.root())
        .ok_or(QueryError::NonIndexedField(field.clone()))?
        .with_positions()?;
    let field_type = search_field.field_entry().field_type();

    let mut terms = Vec::with_capacity(tokens.len());

    if tokens.len() == 1 {
        let term = value_to_term(
            search_field.field(),
            &PdbOwnedValue::Str(tokens.pop().unwrap()),
            field_type,
            field.path().as_deref(),
            index_created_by_version,
        )?;
        Ok(Box::new(TermQuery::new(
            term,
            IndexRecordOption::WithFreqsAndPositions.into(),
        )))
    } else {
        for token in tokens {
            let term = value_to_term(
                search_field.field(),
                &PdbOwnedValue::Str(token),
                field_type,
                field.path().as_deref(),
                index_created_by_version,
            )?;

            terms.push(term);
        }

        let mut query = PhraseQuery::new(terms);
        if let Some(slop) = slop {
            query.set_slop(slop)
        }
        Ok(Box::new(query))
    }
}

fn parse<QueryParserCtor: Fn() -> QueryParser>(
    parser: &QueryParserCtor,
    schema: &SearchIndexSchema,
    index_created_by_version: Option<Version>,
    query_string: String,
    lenient: Option<bool>,
    conjunction_mode: Option<bool>,
) -> anyhow::Result<Box<dyn TantivyQuery>> {
    let mut parser = parser();
    if let Some(true) = conjunction_mode {
        parser.set_conjunction_by_default();
    }

    let lenient = lenient.unwrap_or(false);
    parse_tantivy_query(
        &mut parser,
        &query_string,
        lenient,
        schema,
        index_created_by_version,
    )
}

#[allow(clippy::too_many_arguments)]
fn parse_with_field<QueryParserCtor: Fn() -> QueryParser>(
    field: &FieldName,
    parser: &QueryParserCtor,
    schema: &SearchIndexSchema,
    index_created_by_version: Option<Version>,
    query_string: String,
    lenient: Option<bool>,
    conjunction_mode: Option<bool>,
    fuzzy_data: Option<FuzzyData>,
) -> anyhow::Result<Box<dyn TantivyQuery>> {
    let search_field = schema
        .search_field(field)
        .ok_or(QueryError::NonIndexedField(field.clone()))?;
    let field_type = search_field.field_type();

    // Handle Numeric64 and NumericBytes fields specially
    // Tantivy's QueryParser can't parse decimal strings directly for these types
    if matches!(
        field_type,
        SearchFieldType::Numeric64(_, _) | SearchFieldType::NumericBytes(..)
    ) {
        // Convert the query string to the appropriate numeric format
        let value = PdbOwnedValue::Str(query_string.trim().to_string());
        if let Ok(converted) = convert_value_for_field(value, &field_type) {
            let tantivy_field_type = search_field.field_entry().field_type();
            let term = value_to_term(
                search_field.field(),
                &converted,
                tantivy_field_type,
                field.path().as_deref(),
                index_created_by_version,
            )?;
            return Ok(Box::new(TermQuery::new(
                term,
                IndexRecordOption::WithFreqsAndPositions.into(),
            )));
        }
        // If conversion fails, fall through to standard parsing (will likely error)
    }

    if matches!(search_field.field_entry().field_type(), FieldType::Facet(_)) {
        let facet = dot_path_to_facet(query_string.trim());
        return Ok(Box::new(TermQuery::new(
            Term::from_facet(search_field.field(), &facet),
            IndexRecordOption::Basic.into(),
        )));
    }

    let mut parser = parser();
    if let Some(true) = conjunction_mode {
        parser.set_conjunction_by_default();
    }
    let tantivy_field = search_field.field();

    if let Some(fuzzy_data) = fuzzy_data {
        parser.set_field_fuzzy(
            tantivy_field,
            fuzzy_data.prefix,
            fuzzy_data.distance,
            fuzzy_data.transposition_cost_one,
        );
    }

    let query_string = format!("{field}:({query_string})");
    let lenient = lenient.unwrap_or(false);
    parse_tantivy_query(
        &mut parser,
        &query_string,
        lenient,
        schema,
        index_created_by_version,
    )
}

/// Clauses destined for a [`BooleanQuery`].
///
/// Collapses on build: a boolean of one positive clause is that clause, so
/// unwrapping skips the union scorer and preserves `Weight::count` fast
/// paths (e.g. doc_freq for a single term). A lone `MustNot` is not
/// equivalent to its inner query and stays wrapped.
struct BooleanClauses(Vec<(Occur, Box<dyn TantivyQuery>)>);

impl FromIterator<(Occur, Box<dyn TantivyQuery>)> for BooleanClauses {
    fn from_iter<I: IntoIterator<Item = (Occur, Box<dyn TantivyQuery>)>>(iter: I) -> Self {
        Self(iter.into_iter().collect())
    }
}

impl BooleanClauses {
    fn into_query(mut self) -> Box<dyn TantivyQuery> {
        if self.0.len() == 1 && !matches!(self.0[0].0, Occur::MustNot) {
            return self.0.pop().unwrap().1;
        }
        Box::new(BooleanQuery::new(self.0))
    }
}

#[allow(clippy::too_many_arguments)]
fn match_query(
    field: &FieldName,
    schema: &SearchIndexSchema,
    index_created_by_version: Option<Version>,
    searcher: &Searcher,
    value: &str,
    tokenizer: Option<Value>,
    distance: Option<u8>,
    transposition_cost_one: Option<bool>,
    prefix: Option<bool>,
    conjunction_mode: Option<bool>,
) -> anyhow::Result<Box<dyn TantivyQuery>> {
    let distance = distance.unwrap_or(0);
    let transposition_cost_one = transposition_cost_one.unwrap_or(true);
    let conjunction_mode = conjunction_mode.unwrap_or(false);
    let prefix = prefix.unwrap_or(false);

    let search_field = schema
        .search_field(field.root())
        .ok_or(QueryError::NonIndexedField(field.clone()))?;
    let field_type = search_field.field_entry().field_type();
    let mut analyzer = match tokenizer {
        Some(ref tokenizer) => SearchTokenizer::from_json_value(tokenizer)?
            .to_tantivy_tokenizer()
            .expect("tantivy should support tokenizer {tokenizer:?}"),
        None => resolve_search_tokenizer(&search_field, schema, searcher)?,
    };
    let mut stream = analyzer.token_stream(value);
    let mut terms = Vec::new();

    while stream.advance() {
        let token = stream.token().text.clone();
        terms.push(value_to_term(
            search_field.field(),
            &PdbOwnedValue::Str(token),
            field_type,
            field.path().as_deref(),
            index_created_by_version,
        )?);
    }

    // For conjunction mode, duplicate terms produce redundant Must clauses.
    // This can happen with ngram tokenizers on strings with repeated substrings.
    if conjunction_mode {
        let mut seen = crate::api::HashSet::default();
        terms.retain(|term| seen.insert(term.clone()));
    }

    let occur = if conjunction_mode {
        Occur::Must
    } else {
        Occur::Should
    };

    let clauses: BooleanClauses = terms
        .into_iter()
        .map(|term| {
            let query: Box<dyn TantivyQuery> = match (distance, prefix) {
                (0, _) => Box::new(TermQuery::new(term, IndexRecordOption::WithFreqs.into())),
                (d, true) => Box::new(FuzzyTermQuery::new_prefix(term, d, transposition_cost_one)),
                (d, false) => Box::new(FuzzyTermQuery::new(term, d, transposition_cost_one)),
            };
            (occur, query)
        })
        .collect();

    Ok(clauses.into_query())
}
#[allow(clippy::too_many_arguments)]
fn match_array_query(
    field: &FieldName,
    schema: &SearchIndexSchema,
    index_created_by_version: Option<Version>,
    tokens: Vec<String>,
    distance: Option<u8>,
    transposition_cost_one: Option<bool>,
    prefix: Option<bool>,
    conjunction_mode: Option<bool>,
) -> anyhow::Result<Box<dyn TantivyQuery>> {
    let distance = distance.unwrap_or(0);
    let transposition_cost_one = transposition_cost_one.unwrap_or(true);
    let conjunction_mode = conjunction_mode.unwrap_or(false);
    let prefix = prefix.unwrap_or(false);

    let search_field = schema
        .search_field(field.root())
        .ok_or(QueryError::NonIndexedField(field.clone()))?;
    let field_type = search_field.field_entry().field_type();
    let mut terms = Vec::with_capacity(tokens.len());

    for token in tokens {
        let term = value_to_term(
            search_field.field(),
            &PdbOwnedValue::Str(token),
            field_type,
            field.path().as_deref(),
            index_created_by_version,
        )?;
        let term_query: Box<dyn TantivyQuery> = match (distance, prefix) {
            (0, _) => Box::new(TermQuery::new(
                term,
                IndexRecordOption::WithFreqsAndPositions.into(),
            )),
            (distance, true) => Box::new(FuzzyTermQuery::new_prefix(
                term,
                distance,
                transposition_cost_one,
            )),
            (distance, false) => {
                Box::new(FuzzyTermQuery::new(term, distance, transposition_cost_one))
            }
        };

        let occur = if conjunction_mode {
            Occur::Must
        } else {
            Occur::Should
        };

        terms.push((occur, term_query));
    }

    Ok(BooleanClauses(terms).into_query())
}

fn fuzzy_term(
    field: &FieldName,
    schema: &SearchIndexSchema,
    index_created_by_version: Option<Version>,
    value: String,
    distance: Option<u8>,
    transposition_cost_one: Option<bool>,
    prefix: Option<bool>,
) -> anyhow::Result<Box<dyn TantivyQuery>> {
    let search_field = schema
        .search_field(field.root())
        .ok_or(QueryError::NonIndexedField(field.clone()))?;
    let field_type = search_field.field_entry().field_type();
    let term = value_to_term(
        search_field.field(),
        &PdbOwnedValue::Str(value),
        field_type,
        field.path().as_deref(),
        index_created_by_version,
    )?;
    let distance = distance.unwrap_or(2);
    let transposition_cost_one = transposition_cost_one.unwrap_or(true);
    let query: Box<dyn TantivyQuery> = match prefix {
        Some(true) => Box::new(FuzzyTermQuery::new_prefix(
            term,
            distance,
            transposition_cost_one,
        )),
        Some(false) | None => Box::new(FuzzyTermQuery::new(term, distance, transposition_cost_one)),
    };

    Ok(query)
}

fn fast_field_range_weight(
    field: &FieldName,
    schema: &SearchIndexSchema,
    lower_bound: Bound<u64>,
    upper_bound: Bound<u64>,
) -> Box<FastFieldRangeQuery> {
    let field = schema.search_field(field.root()).unwrap().field();
    let new_lower_bound = match lower_bound {
        Bound::Excluded(v) => Bound::Excluded(Term::from_field_u64(field, v)),
        Bound::Included(v) => Bound::Included(Term::from_field_u64(field, v)),
        Bound::Unbounded => Bound::Unbounded,
    };

    let new_upper_bound = match upper_bound {
        Bound::Excluded(v) => Bound::Excluded(Term::from_field_u64(field, v)),
        Bound::Included(v) => Bound::Included(Term::from_field_u64(field, v)),
        Bound::Unbounded => Bound::Unbounded,
    };

    Box::new(FastFieldRangeQuery::new(new_lower_bound, new_upper_bound))
}

fn exists(field: FieldName, searcher: &Searcher) -> Box<ExistsQuery> {
    let schema_field = searcher.schema().get_field(&field.root()).unwrap();
    let is_json = searcher
        .schema()
        .get_field_entry(schema_field)
        .field_type()
        .is_json();
    Box::new(ExistsQuery::new(field.into_inner(), is_json))
}

pub(super) fn parse_tantivy_query(
    parser: &mut QueryParser,
    query_string: &str,
    lenient: bool,
    schema: &SearchIndexSchema,
    index_created_by_version: Option<Version>,
) -> anyhow::Result<Box<dyn Query>> {
    if lenient {
        let (mut ast, _) = query_grammar::parse_query_lenient(query_string);
        if index_created_by_version.stores_datetimes_in_i64() {
            rewrite_timestamp_literals(&mut ast, schema);
        }
        let (parsed_query, _) = parser.build_query_from_user_input_ast_lenient(ast);
        Ok(parsed_query)
    } else {
        let mut ast = query_grammar::parse_query(query_string)
            .map_err(|_| QueryError::GrammarParseError(query_string.to_string()))?;
        if index_created_by_version.stores_datetimes_in_i64() {
            rewrite_timestamp_literals(&mut ast, schema);
        }
        let parsed_query = parser
            .build_query_from_user_input_ast(ast)
            .map_err(|err| QueryError::ParseError(err, query_string.to_string()))?;
        Ok(parsed_query)
    }
}

/// Walks the parsed user query AST and rewrites date/timestamp-string phrases as i64s.
/// Best-effort: phrases that fail to parse as datetimes are left untouched, which
/// preserves the original tantivy error path for genuinely malformed input.
pub(super) fn rewrite_timestamp_literals(ast: &mut UserInputAst, schema: &SearchIndexSchema) {
    match ast {
        UserInputAst::Clause(children) => {
            for (_, child) in children {
                rewrite_timestamp_literals(child, schema);
            }
        }
        UserInputAst::Boost(inner, _) => rewrite_timestamp_literals(inner, schema),
        UserInputAst::Leaf(leaf) => rewrite_leaf(leaf, schema),
    }
}

fn rewrite_leaf(leaf: &mut UserInputLeaf, schema: &SearchIndexSchema) {
    match leaf {
        UserInputLeaf::Literal(lit) => {
            if let Some(field_name) = &lit.field_name {
                if let Some(oid) = oid_might_require_timestamp_rewriting(schema, field_name) {
                    if let Some(replacement) = phrase_to_pg_micros_string(&lit.phrase, oid) {
                        lit.phrase = replacement;
                    }
                }
            }
        }
        UserInputLeaf::Range {
            field: Some(name),
            lower,
            upper,
        } => {
            if let Some(oid) = oid_might_require_timestamp_rewriting(schema, name) {
                rewrite_bound(lower, oid);
                rewrite_bound(upper, oid);
            }
        }
        UserInputLeaf::Set {
            field: Some(name),
            elements,
        } => {
            if let Some(oid) = oid_might_require_timestamp_rewriting(schema, name) {
                for element in elements.iter_mut() {
                    if let Some(replacement) = phrase_to_pg_micros_string(element, oid) {
                        *element = replacement;
                    }
                }
            }
        }
        // All, Exists, Regex carry no date phrase.
        // Range/Set without an explicit field can't be resolved here; leave them alone.
        _ => (),
    }
}

fn rewrite_bound(bound: &mut UserInputBound, oid: PgOid) {
    match bound {
        UserInputBound::Inclusive(phrase) | UserInputBound::Exclusive(phrase) => {
            if let Some(replacement) = phrase_to_pg_micros_string(phrase, oid) {
                *phrase = replacement;
            }
        }
        UserInputBound::Unbounded => (),
    }
}

fn oid_might_require_timestamp_rewriting(
    schema: &SearchIndexSchema,
    field_name: &str,
) -> Option<PgOid> {
    let oid = schema.search_field(field_name)?.field_type().typeoid();
    if let PgOid::BuiltIn(built_in) = oid {
        match built_in {
            BuiltinOid::JSONOID | BuiltinOid::JSONBOID => Some(oid),
            _ if is_pgoid_datetime_type(oid) => Some(oid),
            _ => None,
        }
    } else {
        None
    }
}

/// Parse `phrase` as a postgres date string and return the corresponding
/// PG-epoch microseconds formatted as a decimal i64 string. Returns `None`
/// if the phrase can't be parsed — caller leaves the phrase untouched.
fn phrase_to_pg_micros_string(phrase: &str, oid: PgOid) -> Option<String> {
    match oid {
        PgOid::BuiltIn(BuiltinOid::TIMESTAMPOID) => {
            let micros = PostgresDateTime::try_from_timestamp_str(phrase)
                .ok()?
                .into_inner();
            Some(micros.to_string())
        }
        PgOid::BuiltIn(BuiltinOid::TIMESTAMPTZOID) => {
            let micros = PostgresDateTime::try_from_timestamptz_str(phrase)
                .ok()?
                .into_inner();
            Some(micros.to_string())
        }
        PgOid::BuiltIn(BuiltinOid::DATEOID) => {
            let micros = PostgresDateTime::try_from_date_str(phrase)
                .ok()?
                .into_inner();
            Some(micros.to_string())
        }
        PgOid::BuiltIn(BuiltinOid::TIMEOID) => {
            let micros = PostgresDateTime::try_from_time_str(phrase)
                .ok()?
                .into_inner();
            Some(micros.to_string())
        }
        PgOid::BuiltIn(BuiltinOid::TIMETZOID) => {
            let micros = PostgresDateTime::try_from_timetz_str(phrase)
                .ok()?
                .into_inner();
            Some(micros.to_string())
        }
        PgOid::BuiltIn(BuiltinOid::JSONOID) | PgOid::BuiltIn(BuiltinOid::JSONBOID) => {
            // Best-effort: if the phrase parses as a timestamp, rewrite to i64 micros.
            // Otherwise leave alone — could be any non-datetime JSON value.
            PostgresDateTime::try_from(phrase)
                .ok()
                .map(|dt| dt.into_inner().to_string())
        }
        _ => None,
    }
}

#[cfg(any(test, feature = "pg_test"))]
#[pgrx::pg_schema]
mod tests {
    use super::pdb::FuzzyData;
    use pgrx::prelude::*;

    #[pg_test]
    fn fuzzy_data_roundtrip() {
        proptest::proptest!(|(distance in 0u8..=255u8, prefix in 0..=1, transposition_cost_one in 0..=1)| {
            let original = FuzzyData {
                distance,
                prefix: prefix == 1,
                transposition_cost_one: transposition_cost_one == 1,
            };

            let typmod_repr: i32 = original.clone().into();
            assert!(typmod_repr >= 0); // can't be negative
            let from_typmod: FuzzyData = typmod_repr.into();
            assert_eq!(original, from_typmod);
        })
    }

    use super::pdb::Query;
    use std::collections::Bound;

    fn match_query(value: &str) -> Query {
        Query::Match {
            value: value.to_string(),
            tokenizer: None,
            distance: None,
            transposition_cost_one: None,
            prefix: None,
            conjunction_mode: None,
        }
    }

    // Wrap a bare pdb::Query into a SearchQueryInput so the single `is_topk_prunable`
    // impl (on SearchQueryInput) can evaluate it. The field name is irrelevant to
    // prunability.
    fn fielded(query: Query) -> crate::query::SearchQueryInput {
        crate::query::SearchQueryInput::FieldedQuery {
            field: crate::api::FieldName::from("field"),
            query,
        }
    }

    #[pg_test]
    fn topk_prunable_only_for_single_posting_list() {
        // A 1-token match is a single posting list: Block-WAND prunes it. (Score-ASC /
        // field-sort gating is the caller's responsibility, not this method's.)
        assert!(fielded(match_query("help")).is_topk_prunable());

        // Multi-term unions are not prunable (Tantivy's cost() weights them).
        assert!(!fielded(match_query("a b c d e")).is_topk_prunable());

        // Non-trivial docsets never prune.
        let range = Query::Range {
            lower_bound: Bound::Unbounded,
            upper_bound: Bound::Unbounded,
        };
        let regex = Query::Regex {
            pattern: "he.*".into(),
        };
        let phrase = Query::Phrase {
            phrases: vec!["help".into(), "common".into()],
            slop: None,
        };
        assert!(!fielded(range).is_topk_prunable());
        assert!(!fielded(regex).is_topk_prunable());
        assert!(!fielded(phrase).is_topk_prunable());

        // A single-token Match with a modifier that expands it past one posting list is
        // NOT prunable: fuzzy (`distance`), `prefix`, or a custom `tokenizer` whose split
        // a whitespace count can't predict.
        let with = |f: fn(&mut Query)| {
            let mut q = match_query("help");
            f(&mut q);
            q
        };
        let fuzzy = with(|q| {
            if let Query::Match { distance, .. } = q {
                *distance = Some(1)
            }
        });
        let prefix = with(|q| {
            if let Query::Match { prefix, .. } = q {
                *prefix = Some(true)
            }
        });
        let custom_tok = with(|q| {
            if let Query::Match { tokenizer, .. } = q {
                *tokenizer = Some(serde_json::Value::Null)
            }
        });
        assert!(!fielded(fuzzy).is_topk_prunable());
        assert!(!fielded(prefix).is_topk_prunable());
        assert!(!fielded(custom_tok).is_topk_prunable());

        // All / Exists have no Block-WAND-pruning weight (AllWeight / ExistsWeight use
        // Tantivy's default full scan), so they route to the cost model, never serial.
        assert!(!fielded(Query::All).is_topk_prunable());
        assert!(!fielded(Query::Exists).is_topk_prunable());

        // Parse strings: a plain token prunes; a parser metacharacter (wildcard/regex)
        // does not -- it can expand past a single posting list.
        let parse = |s: &str| Query::Parse {
            query_string: s.to_string(),
            lenient: None,
            conjunction_mode: None,
        };
        assert!(fielded(parse("alpha")).is_topk_prunable());
        assert!(fielded(parse("alpha_2")).is_topk_prunable());
        assert!(!fielded(parse("alpha*")).is_topk_prunable());
        assert!(!fielded(parse("/al.*/")).is_topk_prunable());
        assert!(!fielded(parse("-foo")).is_topk_prunable()); // negation, not a single posting list
        assert!(!fielded(parse("foo:bar")).is_topk_prunable()); // field qualifier
        assert!(!fielded(parse("(foo)")).is_topk_prunable()); // grouping
    }

    #[pg_test]
    fn topk_prunable_searchqueryinput_top_level_variants() {
        use crate::query::SearchQueryInput;
        // Top-level (non-fielded) variants have no reliably-pruning weight.
        assert!(!SearchQueryInput::All.is_topk_prunable());
        assert!(!SearchQueryInput::TermSet { terms: vec![] }.is_topk_prunable());
        // A top-level plain parser term prunes; a metacharacter / operator does not.
        let parse = |s: &str| SearchQueryInput::Parse {
            query_string: s.to_string(),
            lenient: None,
            conjunction_mode: None,
        };
        assert!(parse("alpha").is_topk_prunable());
        assert!(!parse("alpha*").is_topk_prunable());
        assert!(!parse("-foo").is_topk_prunable());
    }
}
