hermes_core/query/vector/

sparse.rs

//! Sparse vector queries for similarity search (MaxScore-based)

use crate::dsl::Field;
use crate::segment::SegmentReader;
use crate::{DocId, Score, TERMINATED};

use super::combiner::MultiValueCombiner;
use crate::query::ScoredPosition;
use crate::query::traits::{CountFuture, MatchedPositions, Query, Scorer, ScorerFuture};

const DEFAULT_SPARSE_OVER_FETCH_FACTOR: f32 = crate::query::MAX_CANDIDATE_OVERSUBSCRIPTION as f32;

enum SparseQueryInfos {
    Local(Vec<crate::query::SparseTermQueryInfo>),
    Shared(std::sync::Arc<[crate::query::SparseTermQueryInfo]>),
}

impl SparseQueryInfos {
    fn as_slice(&self) -> &[crate::query::SparseTermQueryInfo] {
        match self {
            Self::Local(infos) => infos,
            Self::Shared(infos) => infos,
        }
    }
}

/// Sparse vector query for similarity search
#[derive(Debug, Clone)]
pub struct SparseVectorQuery {
    /// Field containing the sparse vectors
    pub field: Field,
    /// Query vector as (dimension_id, weight) pairs
    pub vector: Vec<(u32, f32)>,
    /// How to combine scores for multi-valued documents
    pub combiner: MultiValueCombiner,
    /// Approximate search factor (1.0 = exact, lower values = faster but approximate)
    /// Controls MaxScore pruning aggressiveness in block-max scoring
    pub heap_factor: f32,
    /// Minimum abs(weight) for query dimensions (0.0 = no filtering)
    /// Dimensions below this threshold are dropped from candidate generation.
    /// BMP still uses the bounded full query when scoring visited candidates.
    pub weight_threshold: f32,
    /// Maximum candidate-generation dimensions (None = implementation cap).
    /// Keeps only the top-k dimensions by abs(weight); BMP final scoring uses
    /// up to `MAX_QUERY_TERMS` dimensions from the full query.
    pub max_query_dims: Option<usize>,
    /// Fraction of query dimensions to keep (0.0-1.0), same semantics as
    /// indexing-time `pruning`: sort by abs(weight) descending,
    /// keep top fraction. BMP applies it to candidate generation and scores
    /// visited candidates with the bounded full query. None or 1.0 = no pruning.
    pub pruning: Option<f32>,
    /// Minimum number of query dimensions before pruning and weight_threshold
    /// filtering are applied. Protects short queries from losing signal.
    /// Default: 4. Set to 0 to always apply.
    pub min_query_dims: usize,
    /// Multiplier on executor limit for ordinal deduplication (1.0 = no over-fetch)
    pub over_fetch_factor: f32,
    /// LSP/0 γ. None is depth-derived; Some(0) is exhaustive.
    pub lsp_gamma: Option<usize>,
    /// Cached pruned vector; None = use `vector` as-is (no pruning applied)
    pruned: Option<Vec<(u32, f32)>>,
}

impl std::fmt::Display for SparseVectorQuery {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let dims = self.pruned_dims();
        write!(f, "Sparse({}, dims={}", self.field.0, dims.len())?;
        if self.heap_factor < 1.0 {
            write!(f, ", heap={}", self.heap_factor)?;
        }
        if self.vector.len() != dims.len() {
            write!(f, ", orig={}", self.vector.len())?;
        }
        write!(f, ")")
    }
}

impl SparseVectorQuery {
    /// Create a new sparse vector query
    ///
    /// Default combiner is `LogSumExp { temperature: 0.7 }` which provides
    /// saturation for documents with many sparse vectors (e.g., 100+ ordinals).
    /// This prevents over-weighting from multiple matches while still allowing
    /// additional matches to contribute to the score.
    pub fn new(field: Field, vector: Vec<(u32, f32)>) -> Self {
        let mut q = Self {
            field,
            vector,
            combiner: MultiValueCombiner::LogSumExp { temperature: 0.7 },
            heap_factor: 1.0,
            weight_threshold: 0.0,
            max_query_dims: Some(crate::query::MAX_QUERY_TERMS),
            pruning: None,
            min_query_dims: 4,
            over_fetch_factor: DEFAULT_SPARSE_OVER_FETCH_FACTOR,
            lsp_gamma: None,
            pruned: None,
        };
        q.pruned = q.compute_pruned_vector();
        q
    }

    /// Effective query dimensions after pruning. Returns `vector` if no pruning is configured.
    pub(crate) fn pruned_dims(&self) -> &[(u32, f32)] {
        self.pruned.as_deref().unwrap_or(&self.vector)
    }

    fn validate(&self, reader: &SegmentReader) -> crate::Result<()> {
        let entry = reader
            .schema()
            .get_field_entry(self.field)
            .ok_or_else(|| crate::Error::FieldNotFound(self.field.0.to_string()))?;
        if entry.field_type != crate::dsl::FieldType::SparseVector {
            return Err(crate::Error::InvalidFieldType {
                expected: "sparse_vector".to_string(),
                got: format!("{:?}", entry.field_type),
            });
        }
        if self.vector.iter().any(|(_, weight)| !weight.is_finite()) {
            return Err(crate::Error::Query(
                "sparse query contains a non-finite weight".to_string(),
            ));
        }
        if self.pruned_dims().len() > crate::query::MAX_QUERY_TERMS {
            return Err(crate::Error::Query(format!(
                "sparse query contains more than {} effective dimensions",
                crate::query::MAX_QUERY_TERMS
            )));
        }
        if !self.heap_factor.is_finite() || !(0.0..=1.0).contains(&self.heap_factor) {
            return Err(crate::Error::Query(format!(
                "sparse heap_factor must be finite and in [0, 1], got {}",
                self.heap_factor
            )));
        }
        if !self.over_fetch_factor.is_finite()
            || !(1.0..=DEFAULT_SPARSE_OVER_FETCH_FACTOR).contains(&self.over_fetch_factor)
        {
            return Err(crate::Error::Query(format!(
                "sparse over_fetch_factor must be finite and in [1, {DEFAULT_SPARSE_OVER_FETCH_FACTOR}], got {}",
                self.over_fetch_factor
            )));
        }
        self.combiner.validate().map_err(crate::Error::Query)
    }

    /// Set the multi-value score combiner
    pub fn with_combiner(mut self, combiner: MultiValueCombiner) -> Self {
        self.combiner = combiner;
        self
    }

    /// Set executor over-fetch factor for multi-valued fields.
    /// After MaxScore execution, ordinal combining may reduce result count;
    /// this multiplier compensates by fetching more from the executor.
    /// (1.0 = no over-fetch, 2.0 = fetch 2x then combine down)
    pub fn with_over_fetch_factor(mut self, factor: f32) -> Self {
        self.over_fetch_factor = factor.clamp(1.0, DEFAULT_SPARSE_OVER_FETCH_FACTOR);
        self
    }

    /// Set the heap factor for approximate search
    ///
    /// Controls the trade-off between speed and recall:
    /// - 1.0 = exact search (default)
    /// - 0.8-0.9 = ~20-40% faster with minimal recall loss
    /// - Lower values = more aggressive pruning, faster but lower recall
    pub fn with_heap_factor(mut self, heap_factor: f32) -> Self {
        self.heap_factor = heap_factor.clamp(0.0, 1.0);
        self
    }

    /// Set minimum weight threshold for query dimensions
    /// Dimensions with abs(weight) below this are dropped before search.
    pub fn with_weight_threshold(mut self, threshold: f32) -> Self {
        self.weight_threshold = threshold;
        self.pruned = self.compute_pruned_vector();
        self
    }

    /// Set maximum number of query dimensions (top-k by weight)
    pub fn with_max_query_dims(mut self, max_dims: usize) -> Self {
        // MaxScore and BMP use a u64 query-term mask in their hot paths.  Keep
        // this invariant here even when an SDL or RPC override asks for more.
        self.max_query_dims = Some(max_dims.min(crate::query::MAX_QUERY_TERMS));
        self.pruned = self.compute_pruned_vector();
        self
    }

    /// Set pruning fraction (0.0-1.0): keep top fraction of query dims by weight.
    /// Same semantics as indexing-time `pruning`.
    pub fn with_pruning(mut self, fraction: f32) -> Self {
        self.pruning = Some(fraction.clamp(0.0, 1.0));
        self.pruned = self.compute_pruned_vector();
        self
    }

    /// Set minimum query dimensions before pruning/filtering are applied.
    /// Queries with fewer dimensions than this skip weight_threshold and pruning.
    pub fn with_min_query_dims(mut self, min_dims: usize) -> Self {
        self.min_query_dims = min_dims;
        self.pruned = self.compute_pruned_vector();
        self
    }

    /// Select at most the top-γ superblocks by SBMax using LSP/0.
    /// Zero retains exhaustive SBMax-ordered traversal.
    pub fn with_lsp_gamma(mut self, gamma: usize) -> Self {
        self.lsp_gamma = Some(gamma);
        self
    }

    /// Apply weight_threshold, pruning, and max_query_dims. `None` aliases the
    /// original query vector and avoids a second allocation on the default
    /// unpruned path.
    fn compute_pruned_vector(&self) -> Option<Vec<(u32, f32)>> {
        let original_len = self.vector.len();
        let max_dims = self
            .max_query_dims
            .unwrap_or(crate::query::MAX_QUERY_TERMS)
            .min(crate::query::MAX_QUERY_TERMS);
        let filtering_enabled = self.weight_threshold > 0.0 && original_len > self.min_query_dims;
        let pruning_enabled = self
            .pruning
            .is_some_and(|fraction| fraction < 1.0 && original_len > self.min_query_dims);
        if !filtering_enabled && !pruning_enabled && original_len <= max_dims {
            return None;
        }

        // Step 1: weight_threshold — drop dimensions below minimum weight
        // Skip when query has fewer than min_query_dims dimensions
        let mut v: Vec<(u32, f32)> = if filtering_enabled {
            self.vector
                .iter()
                .copied()
                .filter(|(_, w)| w.abs() >= self.weight_threshold)
                .collect()
        } else {
            self.vector.clone()
        };
        let after_threshold = v.len();

        // Step 2: pruning — keep top fraction by abs(weight), same as indexing
        // Skip when query has fewer than min_query_dims dimensions
        let mut sorted_by_weight = false;
        if let Some(fraction) = self.pruning
            && fraction < 1.0
            && v.len() > self.min_query_dims
        {
            v.sort_unstable_by(|a, b| b.1.abs().total_cmp(&a.1.abs()).then_with(|| a.0.cmp(&b.0)));
            sorted_by_weight = true;
            let keep = ((v.len() as f64 * fraction as f64).ceil() as usize).max(1);
            v.truncate(keep);
        }
        let after_pruning = v.len();

        // Step 3: max_query_dims — absolute cap on dimensions.  The hard
        // MAX_QUERY_TERMS bound is a correctness requirement, not merely a
        // tuning default: both sparse executors represent query terms in u64.
        if v.len() > max_dims {
            if !sorted_by_weight {
                v.sort_unstable_by(|a, b| {
                    b.1.abs().total_cmp(&a.1.abs()).then_with(|| a.0.cmp(&b.0))
                });
            }
            v.truncate(max_dims);
        }

        if v.len() < original_len && log::log_enabled!(log::Level::Debug) {
            let src: Vec<_> = self
                .vector
                .iter()
                .map(|(d, w)| format!("({},{:.4})", d, w))
                .collect();
            let pruned_fmt: Vec<_> = v.iter().map(|(d, w)| format!("({},{:.4})", d, w)).collect();
            log::debug!(
                "[sparse query] field={}: pruned {}->{} dims \
                 (threshold: {}->{}, pruning: {}->{}, max_dims: {}->{}), \
                 source=[{}], pruned=[{}]",
                self.field.0,
                original_len,
                v.len(),
                original_len,
                after_threshold,
                after_threshold,
                after_pruning,
                after_pruning,
                v.len(),
                src.join(", "),
                pruned_fmt.join(", "),
            );
        }

        Some(v)
    }

    /// Create from separate indices and weights vectors
    pub fn from_indices_weights(field: Field, indices: Vec<u32>, weights: Vec<f32>) -> Self {
        let vector: Vec<(u32, f32)> = indices.into_iter().zip(weights).collect();
        Self::new(field, vector)
    }

    /// Create from raw text using a HuggingFace tokenizer (single segment)
    ///
    /// This method tokenizes the text and creates a sparse vector query.
    /// For multi-segment indexes, use `from_text_with_stats` instead.
    ///
    /// # Arguments
    /// * `field` - The sparse vector field to search
    /// * `text` - Raw text to tokenize
    /// * `tokenizer_name` - HuggingFace tokenizer path (e.g., "bert-base-uncased")
    /// * `weighting` - Weighting strategy for tokens
    /// * `sparse_index` - Optional sparse index for IDF lookup (required for IDF weighting)
    #[cfg(feature = "native")]
    pub fn from_text(
        field: Field,
        text: &str,
        tokenizer_name: &str,
        weighting: crate::structures::QueryWeighting,
        sparse_index: Option<&crate::segment::SparseIndex>,
    ) -> crate::Result<Self> {
        use crate::structures::QueryWeighting;
        use crate::tokenizer::tokenizer_cache;

        let tokenizer = tokenizer_cache().get_or_load(tokenizer_name)?;
        let token_ids = tokenizer.tokenize_unique(text)?;

        let weights: Vec<f32> = match weighting {
            QueryWeighting::One => vec![1.0f32; token_ids.len()],
            QueryWeighting::Idf => {
                if let Some(index) = sparse_index {
                    index.idf_weights(&token_ids)
                } else {
                    vec![1.0f32; token_ids.len()]
                }
            }
            QueryWeighting::IdfFile => {
                use crate::tokenizer::idf_weights_cache;
                if let Some(idf) = idf_weights_cache().get_or_load(tokenizer_name, None) {
                    token_ids.iter().map(|&id| idf.get(id)).collect()
                } else {
                    vec![1.0f32; token_ids.len()]
                }
            }
        };

        let vector: Vec<(u32, f32)> = token_ids.into_iter().zip(weights).collect();
        Ok(Self::new(field, vector))
    }

    /// Create from raw text using global statistics (multi-segment)
    ///
    /// This is the recommended method for multi-segment indexes as it uses
    /// aggregated IDF values across all segments for consistent ranking.
    ///
    /// # Arguments
    /// * `field` - The sparse vector field to search
    /// * `text` - Raw text to tokenize
    /// * `tokenizer` - Pre-loaded HuggingFace tokenizer
    /// * `weighting` - Weighting strategy for tokens
    /// * `global_stats` - Global statistics for IDF computation
    #[cfg(feature = "native")]
    pub fn from_text_with_stats(
        field: Field,
        text: &str,
        tokenizer: &crate::tokenizer::HfTokenizer,
        weighting: crate::structures::QueryWeighting,
        global_stats: Option<&crate::query::GlobalStats>,
    ) -> crate::Result<Self> {
        use crate::structures::QueryWeighting;

        let token_ids = tokenizer.tokenize_unique(text)?;

        let weights: Vec<f32> = match weighting {
            QueryWeighting::One => vec![1.0f32; token_ids.len()],
            QueryWeighting::Idf => {
                if let Some(stats) = global_stats {
                    // Clamp to zero: negative weights don't make sense for IDF
                    stats
                        .sparse_idf_weights(field, &token_ids)
                        .into_iter()
                        .map(|w| w.max(0.0))
                        .collect()
                } else {
                    vec![1.0f32; token_ids.len()]
                }
            }
            QueryWeighting::IdfFile => {
                // IdfFile requires a tokenizer name for HF model lookup;
                // this code path doesn't have one, so fall back to 1.0
                vec![1.0f32; token_ids.len()]
            }
        };

        let vector: Vec<(u32, f32)> = token_ids.into_iter().zip(weights).collect();
        Ok(Self::new(field, vector))
    }

    /// Create from raw text, loading tokenizer from index directory
    ///
    /// This method supports the `index://` prefix for tokenizer paths,
    /// loading tokenizer.json from the index directory.
    ///
    /// # Arguments
    /// * `field` - The sparse vector field to search
    /// * `text` - Raw text to tokenize
    /// * `tokenizer_bytes` - Tokenizer JSON bytes (pre-loaded from directory)
    /// * `weighting` - Weighting strategy for tokens
    /// * `global_stats` - Global statistics for IDF computation
    #[cfg(feature = "native")]
    pub fn from_text_with_tokenizer_bytes(
        field: Field,
        text: &str,
        tokenizer_bytes: &[u8],
        weighting: crate::structures::QueryWeighting,
        global_stats: Option<&crate::query::GlobalStats>,
    ) -> crate::Result<Self> {
        use crate::structures::QueryWeighting;
        use crate::tokenizer::HfTokenizer;

        let tokenizer = HfTokenizer::from_bytes(tokenizer_bytes)?;
        let token_ids = tokenizer.tokenize_unique(text)?;

        let weights: Vec<f32> = match weighting {
            QueryWeighting::One => vec![1.0f32; token_ids.len()],
            QueryWeighting::Idf => {
                if let Some(stats) = global_stats {
                    // Clamp to zero: negative weights don't make sense for IDF
                    stats
                        .sparse_idf_weights(field, &token_ids)
                        .into_iter()
                        .map(|w| w.max(0.0))
                        .collect()
                } else {
                    vec![1.0f32; token_ids.len()]
                }
            }
            QueryWeighting::IdfFile => {
                // IdfFile requires a tokenizer name for HF model lookup;
                // this code path doesn't have one, so fall back to 1.0
                vec![1.0f32; token_ids.len()]
            }
        };

        let vector: Vec<(u32, f32)> = token_ids.into_iter().zip(weights).collect();
        Ok(Self::new(field, vector))
    }
}

impl SparseVectorQuery {
    /// Build a bounded full-query decomposition and mark the pruned terms used
    /// for candidate generation. LSP/BMP scores visited documents with the
    /// full list; MaxScore continues to consume only marked candidate terms.
    fn sparse_infos(&self) -> Vec<crate::query::SparseTermQueryInfo> {
        let candidate_dims: Option<rustc_hash::FxHashSet<u32>> = self
            .pruned
            .as_ref()
            .map(|dimensions| dimensions.iter().map(|&(dimension, _)| dimension).collect());
        let make_info = |(dim_id, weight)| crate::query::SparseTermQueryInfo {
            field: self.field,
            dim_id,
            weight,
            candidate: candidate_dims
                .as_ref()
                .is_none_or(|dimensions| dimensions.contains(&dim_id)),
            heap_factor: self.heap_factor,
            combiner: self.combiner,
            over_fetch_factor: self.over_fetch_factor,
            lsp_gamma: self.lsp_gamma,
        };
        if self.vector.len() <= crate::query::MAX_QUERY_TERMS {
            return self.vector.iter().copied().map(make_info).collect();
        }

        let mut scoring_dims = self.vector.clone();
        scoring_dims.sort_unstable_by(|left, right| {
            right
                .1
                .abs()
                .total_cmp(&left.1.abs())
                .then_with(|| left.0.cmp(&right.0))
        });
        scoring_dims.truncate(crate::query::MAX_QUERY_TERMS);
        scoring_dims.into_iter().map(make_info).collect()
    }

    fn sparse_infos_for_plan(
        &self,
        plan: Option<&std::sync::Arc<crate::query::bmp::LspSegmentPlan>>,
    ) -> SparseQueryInfos {
        match plan {
            Some(plan) => SparseQueryInfos::Shared(std::sync::Arc::clone(&plan.infos)),
            None => SparseQueryInfos::Local(self.sparse_infos()),
        }
    }
}

impl Query for SparseVectorQuery {
    fn scorer<'a>(&self, reader: &'a SegmentReader, limit: usize) -> ScorerFuture<'a> {
        self.scorer_with_options(reader, limit, crate::query::ScorerOptions::with_positions())
    }

    fn scorer_with_options<'a>(
        &self,
        reader: &'a SegmentReader,
        limit: usize,
        options: crate::query::ScorerOptions,
    ) -> ScorerFuture<'a> {
        let validation = self.validate(reader);
        let infos = self.sparse_infos_for_plan(options.lsp_plan.as_ref());

        Box::pin(async move {
            validation?;
            let infos = infos.as_slice();
            if infos.is_empty() {
                return Ok(Box::new(crate::query::EmptyScorer) as Box<dyn Scorer>);
            }

            // Auto-detect: try BMP executor first (coupled to index format)
            if let Some((raw, info)) =
                crate::query::planner::build_sparse_bmp_results(infos, reader, limit, &options)?
            {
                return Ok(crate::query::planner::combine_sparse_results(
                    raw,
                    info.combiner,
                    info.field,
                    limit,
                ));
            }

            // Fall back to MaxScore execution
            if let Some((executor, info)) =
                crate::query::planner::build_sparse_maxscore_executor(infos, reader, limit, None)
            {
                let raw = executor.execute().await?;
                return Ok(crate::query::planner::combine_sparse_results(
                    raw,
                    info.combiner,
                    info.field,
                    limit,
                ));
            }

            Ok(Box::new(crate::query::EmptyScorer) as Box<dyn Scorer>)
        })
    }

    #[cfg(feature = "sync")]
    fn scorer_sync<'a>(
        &self,
        reader: &'a SegmentReader,
        limit: usize,
    ) -> crate::Result<Box<dyn Scorer + 'a>> {
        self.scorer_sync_with_options(reader, limit, crate::query::ScorerOptions::with_positions())
    }

    #[cfg(feature = "sync")]
    fn scorer_sync_with_options<'a>(
        &self,
        reader: &'a SegmentReader,
        limit: usize,
        options: crate::query::ScorerOptions,
    ) -> crate::Result<Box<dyn Scorer + 'a>> {
        self.validate(reader)?;
        let infos = self.sparse_infos_for_plan(options.lsp_plan.as_ref());
        let infos = infos.as_slice();
        if infos.is_empty() {
            return Ok(Box::new(crate::query::EmptyScorer) as Box<dyn Scorer + 'a>);
        }

        // Auto-detect: try BMP executor first (coupled to index format)
        if let Some((raw, info)) =
            crate::query::planner::build_sparse_bmp_results(infos, reader, limit, &options)?
        {
            return Ok(crate::query::planner::combine_sparse_results(
                raw,
                info.combiner,
                info.field,
                limit,
            ));
        }

        // Fall back to MaxScore execution
        if let Some((executor, info)) =
            crate::query::planner::build_sparse_maxscore_executor(infos, reader, limit, None)
        {
            let raw = executor.execute_sync()?;
            return Ok(crate::query::planner::combine_sparse_results(
                raw,
                info.combiner,
                info.field,
                limit,
            ));
        }

        Ok(Box::new(crate::query::EmptyScorer) as Box<dyn Scorer + 'a>)
    }

    fn count_estimate<'a>(&self, _reader: &'a SegmentReader) -> CountFuture<'a> {
        Box::pin(async move { Ok(u32::MAX) })
    }

    fn decompose(&self) -> crate::query::QueryDecomposition {
        let infos = self.sparse_infos();
        if infos.is_empty() {
            crate::query::QueryDecomposition::Opaque
        } else {
            crate::query::QueryDecomposition::SparseTerms(infos)
        }
    }
}

// ── SparseTermQuery: single sparse dimension query (like TermQuery for text) ──

/// Query for a single sparse vector dimension.
///
/// Analogous to `TermQuery` for text: searches one dimension's posting list
/// with a given weight. Multiple `SparseTermQuery` instances are combined as
/// `BooleanQuery` SHOULD clauses to form a full sparse vector search.
#[derive(Debug, Clone)]
pub struct SparseTermQuery {
    pub field: Field,
    pub dim_id: u32,
    pub weight: f32,
    /// MaxScore heap factor (1.0 = exact, lower = approximate)
    pub heap_factor: f32,
    /// Multi-value combiner for ordinal deduplication
    pub combiner: MultiValueCombiner,
    /// Multiplier on executor limit to compensate for ordinal deduplication
    pub over_fetch_factor: f32,
}

impl std::fmt::Display for SparseTermQuery {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(
            f,
            "SparseTerm({}, dim={}, w={:.3})",
            self.field.0, self.dim_id, self.weight
        )
    }
}

impl SparseTermQuery {
    pub fn new(field: Field, dim_id: u32, weight: f32) -> Self {
        Self {
            field,
            dim_id,
            weight,
            heap_factor: 1.0,
            combiner: MultiValueCombiner::default(),
            over_fetch_factor: DEFAULT_SPARSE_OVER_FETCH_FACTOR,
        }
    }

    pub fn with_heap_factor(mut self, heap_factor: f32) -> Self {
        self.heap_factor = heap_factor;
        self
    }

    pub fn with_combiner(mut self, combiner: MultiValueCombiner) -> Self {
        self.combiner = combiner;
        self
    }

    pub fn with_over_fetch_factor(mut self, factor: f32) -> Self {
        self.over_fetch_factor = factor.clamp(1.0, DEFAULT_SPARSE_OVER_FETCH_FACTOR);
        self
    }

    fn validate(&self, reader: &SegmentReader) -> crate::Result<()> {
        let entry = reader
            .schema()
            .get_field_entry(self.field)
            .ok_or_else(|| crate::Error::FieldNotFound(self.field.0.to_string()))?;
        if entry.field_type != crate::dsl::FieldType::SparseVector {
            return Err(crate::Error::InvalidFieldType {
                expected: "sparse_vector".to_string(),
                got: format!("{:?}", entry.field_type),
            });
        }
        if !self.weight.is_finite() {
            return Err(crate::Error::Query(
                "sparse term query weight must be finite".to_string(),
            ));
        }
        if !self.heap_factor.is_finite() || !(0.0..=1.0).contains(&self.heap_factor) {
            return Err(crate::Error::Query(format!(
                "sparse heap_factor must be finite and in [0, 1], got {}",
                self.heap_factor
            )));
        }
        if !self.over_fetch_factor.is_finite()
            || !(1.0..=DEFAULT_SPARSE_OVER_FETCH_FACTOR).contains(&self.over_fetch_factor)
        {
            return Err(crate::Error::Query(format!(
                "sparse over_fetch_factor must be finite and in [1, {DEFAULT_SPARSE_OVER_FETCH_FACTOR}], got {}",
                self.over_fetch_factor
            )));
        }
        self.combiner.validate().map_err(crate::Error::Query)
    }

    /// BMP fallback: execute BMP for this single dimension and wrap in a TopK scorer.
    fn bmp_fallback_scorer<'a>(
        &self,
        reader: &'a SegmentReader,
        limit: usize,
        options: &crate::query::ScorerOptions,
    ) -> crate::Result<Box<dyn Scorer + 'a>> {
        let infos = [crate::query::SparseTermQueryInfo {
            field: self.field,
            dim_id: self.dim_id,
            weight: self.weight,
            candidate: true,
            heap_factor: self.heap_factor,
            combiner: self.combiner,
            over_fetch_factor: self.over_fetch_factor,
            lsp_gamma: None,
        }];
        if let Some((raw, info)) =
            crate::query::planner::build_sparse_bmp_results(&infos, reader, limit, options)?
        {
            return Ok(crate::query::planner::combine_sparse_results(
                raw,
                info.combiner,
                info.field,
                limit,
            ));
        }
        Ok(Box::new(crate::query::EmptyScorer))
    }

    /// Create a SparseTermScorer from this query's config against a segment.
    /// Returns EmptyScorer if the dimension doesn't exist.
    fn make_scorer<'a>(
        &self,
        reader: &'a SegmentReader,
    ) -> crate::Result<Option<SparseTermScorer<'a>>> {
        let si = match reader.sparse_index(self.field) {
            Some(si) => si,
            None => return Ok(None),
        };
        let (skip_start, skip_count, global_max, block_data_offset) =
            match si.get_skip_range_full(self.dim_id) {
                Some(v) => v,
                None => return Ok(None),
            };
        let cursor = crate::query::TermCursor::sparse(
            si,
            self.weight,
            skip_start,
            skip_count,
            global_max,
            block_data_offset,
        );
        Ok(Some(SparseTermScorer {
            cursor,
            field_id: self.field.0,
        }))
    }
}

impl Query for SparseTermQuery {
    fn scorer<'a>(&self, reader: &'a SegmentReader, limit: usize) -> ScorerFuture<'a> {
        self.scorer_with_options(reader, limit, crate::query::ScorerOptions::with_positions())
    }

    fn scorer_with_options<'a>(
        &self,
        reader: &'a SegmentReader,
        limit: usize,
        options: crate::query::ScorerOptions,
    ) -> ScorerFuture<'a> {
        let query = self.clone();
        Box::pin(async move {
            query.validate(reader)?;
            let mut scorer = match query.make_scorer(reader)? {
                Some(s) => s,
                None => return query.bmp_fallback_scorer(reader, limit, &options),
            };
            scorer.cursor.ensure_block_loaded().await.ok();
            Ok(Box::new(scorer) as Box<dyn Scorer + 'a>)
        })
    }

    #[cfg(feature = "sync")]
    fn scorer_sync<'a>(
        &self,
        reader: &'a SegmentReader,
        limit: usize,
    ) -> crate::Result<Box<dyn Scorer + 'a>> {
        self.scorer_sync_with_options(reader, limit, crate::query::ScorerOptions::with_positions())
    }

    #[cfg(feature = "sync")]
    fn scorer_sync_with_options<'a>(
        &self,
        reader: &'a SegmentReader,
        limit: usize,
        options: crate::query::ScorerOptions,
    ) -> crate::Result<Box<dyn Scorer + 'a>> {
        self.validate(reader)?;
        let mut scorer = match self.make_scorer(reader)? {
            Some(s) => s,
            None => return self.bmp_fallback_scorer(reader, limit, &options),
        };
        scorer.cursor.ensure_block_loaded_sync().ok();
        Ok(Box::new(scorer) as Box<dyn Scorer + 'a>)
    }

    fn count_estimate<'a>(&self, reader: &'a SegmentReader) -> CountFuture<'a> {
        let field = self.field;
        let dim_id = self.dim_id;
        Box::pin(async move {
            let si = match reader.sparse_index(field) {
                Some(si) => si,
                None => return Ok(0),
            };
            match si.get_skip_range_full(dim_id) {
                Some((_, skip_count, _, _)) => Ok((skip_count * 256) as u32),
                None => Ok(0),
            }
        })
    }

    fn decompose(&self) -> crate::query::QueryDecomposition {
        crate::query::QueryDecomposition::SparseTerms(vec![crate::query::SparseTermQueryInfo {
            field: self.field,
            dim_id: self.dim_id,
            weight: self.weight,
            candidate: true,
            heap_factor: self.heap_factor,
            combiner: self.combiner,
            over_fetch_factor: self.over_fetch_factor,
            lsp_gamma: None,
        }])
    }
}

/// Lazy scorer for a single sparse dimension, backed by `TermCursor::Sparse`.
///
/// Iterates through the posting list block-by-block using sync I/O.
/// Score for each doc = `query_weight * quantized_stored_weight`.
struct SparseTermScorer<'a> {
    cursor: crate::query::TermCursor<'a>,
    field_id: u32,
}

impl crate::query::docset::DocSet for SparseTermScorer<'_> {
    fn doc(&self) -> DocId {
        let d = self.cursor.doc();
        if d == u32::MAX { TERMINATED } else { d }
    }

    fn advance(&mut self) -> DocId {
        match self.cursor.advance_sync() {
            Ok(d) if d == u32::MAX => TERMINATED,
            Ok(d) => d,
            Err(_) => TERMINATED,
        }
    }

    fn seek(&mut self, target: DocId) -> DocId {
        match self.cursor.seek_sync(target) {
            Ok(d) if d == u32::MAX => TERMINATED,
            Ok(d) => d,
            Err(_) => TERMINATED,
        }
    }

    fn size_hint(&self) -> u32 {
        0
    }
}

impl Scorer for SparseTermScorer<'_> {
    fn score(&self) -> Score {
        self.cursor.score()
    }

    fn matched_positions(&self) -> Option<MatchedPositions> {
        let ordinal = self.cursor.ordinal();
        let score = self.cursor.score();
        if score == 0.0 {
            return None;
        }
        Some(vec![(
            self.field_id,
            vec![ScoredPosition::new(ordinal as u32, score)],
        )])
    }
}

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

    #[test]
    fn test_sparse_vector_query_new() {
        let sparse = vec![(1, 0.5), (5, 0.3), (10, 0.2)];
        let query = SparseVectorQuery::new(Field(0), sparse.clone());

        assert_eq!(query.field, Field(0));
        assert_eq!(query.vector, sparse);
        assert!(
            query.pruned.is_none(),
            "the default path must alias the source vector instead of cloning it"
        );
    }

    #[test]
    fn test_sparse_vector_query_from_indices_weights() {
        let query =
            SparseVectorQuery::from_indices_weights(Field(0), vec![1, 5, 10], vec![0.5, 0.3, 0.2]);

        assert_eq!(query.vector, vec![(1, 0.5), (5, 0.3), (10, 0.2)]);
    }

    #[test]
    fn max_query_dims_cannot_exceed_executor_mask_width() {
        let vector: Vec<(u32, f32)> = (0..100).map(|dim| (dim, dim as f32 + 1.0)).collect();
        let query = SparseVectorQuery::new(Field(0), vector).with_max_query_dims(usize::MAX);

        assert_eq!(query.pruned_dims().len(), crate::query::MAX_QUERY_TERMS);
        // Pruning retains the dimensions with the largest absolute weights.
        assert!(query.pruned_dims().iter().all(|(dim, _)| *dim >= 36));
    }

    #[test]
    fn decomposition_keeps_full_scores_and_marks_pruned_candidates() {
        let query = SparseVectorQuery::new(Field(0), vec![(3, 1.0), (7, 0.8), (11, 0.2)])
            .with_min_query_dims(0)
            .with_pruning(0.34);
        let infos = query.sparse_infos();

        assert_eq!(infos.len(), 3);
        assert_eq!(
            infos
                .iter()
                .filter(|info| info.candidate)
                .map(|info| info.dim_id)
                .collect::<Vec<_>>(),
            vec![3, 7]
        );
        assert_eq!(
            infos
                .iter()
                .map(|info| (info.dim_id, info.weight))
                .collect::<Vec<_>>(),
            vec![(3, 1.0), (7, 0.8), (11, 0.2)]
        );
    }

    #[test]
    fn sparse_over_fetch_factor_uses_shared_candidate_bound() {
        let query = SparseVectorQuery::new(Field(0), vec![]).with_over_fetch_factor(99.0);
        let term = SparseTermQuery::new(Field(0), 1, 1.0).with_over_fetch_factor(99.0);

        assert_eq!(
            query.over_fetch_factor,
            crate::query::MAX_CANDIDATE_OVERSUBSCRIPTION as f32
        );
        assert_eq!(
            term.over_fetch_factor,
            crate::query::MAX_CANDIDATE_OVERSUBSCRIPTION as f32
        );
    }
}