velesdb-core 5.0.0

use super::{Collection, HashSet, QuerySearchOptions, Result, SearchResult, MAX_LIMIT};

impl Collection {
    // Metadata index query strategy is in metadata_query.rs

    pub(in crate::collection::search::query) fn evaluate_graph_match_anchor_ids(
        &self,
        predicate: &crate::velesql::GraphMatchPredicate,
        params: &std::collections::HashMap<String, serde_json::Value>,
        from_aliases: &[String],
        guards: Option<&super::match_exec::MatchStorageGuards<'_>>,
    ) -> Result<HashSet<u64>> {
        let anchor_alias = Self::resolve_anchor_alias(predicate, from_aliases)?;
        let clause = Self::build_anchor_match_clause(predicate);

        // Anchor evaluation needs the full unordered match set (it collects bound
        // ids into a set), so use the raw primitive rather than the ORDER BY/LIMIT
        // entry point. A caller that already holds the vector/payload guards
        // (the aggregation runtime WHERE loop evaluating MATCH-in-WHERE) passes
        // them through so the nested MATCH does not re-acquire either lock —
        // a nested read acquisition on the same thread deadlocks once a writer
        // queues (parking_lot task-fair semantics).
        let matches = match guards {
            Some(g) => self.execute_match_with_guards(&clause, params, None, g)?,
            None => self.execute_match_with_context(&clause, params, None)?,
        };
        let mut ids = HashSet::with_capacity(matches.len());
        for m in matches {
            if let Some(id) = m.bindings.get(&anchor_alias) {
                ids.insert(*id);
            }
        }
        Ok(ids)
    }

    /// Extracts and validates the anchor alias from the first node in a MATCH predicate.
    ///
    /// Mirrors the V011 anchor rule (`validation_anchor.rs`): an anchor alias
    /// declared in FROM/JOIN binds explicitly; otherwise the leftmost node
    /// binds implicitly to the FROM rows, guarded by G1 (a declared alias in
    /// a non-anchor position) and G3 (no `@collection` override on the
    /// anchor). G2 (implicit anchor shared across MATCH predicates) is
    /// validation-only: it needs the whole WHERE tree, and every execution
    /// path runs `QueryValidator::validate` before reaching this point.
    fn resolve_anchor_alias(
        predicate: &crate::velesql::GraphMatchPredicate,
        from_aliases: &[String],
    ) -> Result<String> {
        let first_node = predicate.pattern.nodes.first().ok_or_else(|| {
            crate::error::Error::Query("MATCH predicate requires at least one node".to_string())
        })?;

        let anchor_alias = first_node.alias.clone().ok_or_else(|| {
            crate::error::Error::Query(
                "MATCH predicate in SELECT WHERE requires an alias on the first node, \
                 e.g. MATCH (d:Doc)-[:REL]->(x)"
                    .to_string(),
            )
        })?;

        // BUG-8: explicit anchor — or a bare FROM, where any anchor is accepted.
        if from_aliases.is_empty() || from_aliases.iter().any(|a| a == &anchor_alias) {
            return Ok(anchor_alias);
        }
        Self::check_implicit_anchor_guards(predicate, &anchor_alias, from_aliases)?;
        // Implicit anchor: the leftmost node binds to the FROM rows.
        Ok(anchor_alias)
    }

    /// Runtime G1/G3 guards for an anchor alias not declared in FROM/JOIN
    /// (implicit binding candidate).
    fn check_implicit_anchor_guards(
        predicate: &crate::velesql::GraphMatchPredicate,
        anchor_alias: &str,
        from_aliases: &[String],
    ) -> Result<()> {
        // G1: a declared alias elsewhere in the pattern means the anchor must
        // be that alias (the pattern direction is likely inverted).
        let declared = predicate
            .pattern
            .nodes
            .iter()
            .skip(1)
            .filter_map(|node| node.alias.as_deref())
            .find(|alias| from_aliases.iter().any(|f| f == alias));
        if let Some(declared) = declared {
            return Err(crate::error::Error::Query(format!(
                "MATCH predicate anchor alias '{anchor_alias}' must be the declared \
                 FROM/JOIN alias '{declared}' used elsewhere in the pattern"
            )));
        }
        // G3: a @collection anchor resolves outside the FROM collection and
        // cannot bind implicitly to its rows.
        if predicate
            .pattern
            .nodes
            .first()
            .is_some_and(|node| node.collection.is_some())
        {
            return Err(crate::error::Error::Query(format!(
                "MATCH predicate anchor alias '{anchor_alias}' has a @collection \
                 override; anchor on one of the FROM/JOIN aliases: {from_aliases:?}"
            )));
        }
        Ok(())
    }

    /// Builds a `MatchClause` that returns all bindings for anchor evaluation.
    fn build_anchor_match_clause(
        predicate: &crate::velesql::GraphMatchPredicate,
    ) -> crate::velesql::MatchClause {
        crate::velesql::MatchClause {
            patterns: vec![predicate.pattern.clone()],
            where_clause: None,
            return_clause: crate::velesql::ReturnClause {
                items: vec![crate::velesql::ReturnItem {
                    expression: "*".to_string(),
                    alias: None,
                }],
                order_by: None,
                // Internal anchor evaluation must not silently cap MATCH results.
                limit: Some(u64::MAX),
            },
        }
    }

    /// Dispatches the core vector / similarity / metadata query based on extracted components.
    ///
    /// Called from `execute_query_with_client` after query extraction and CBO planning.
    /// Handles all combinations of NEAR, similarity(), and metadata-only queries.
    /// Applies optional metadata post-filter to an already similarity-filtered result set.
    fn apply_optional_metadata_filter(
        filtered: Vec<SearchResult>,
        filter_cond: Option<&crate::velesql::Condition>,
        skip_metadata_prefilter_for_graph_or: bool,
        execution_limit: usize,
    ) -> Vec<SearchResult> {
        let Some(cond) = filter_cond else {
            return filtered;
        };
        if skip_metadata_prefilter_for_graph_or {
            return filtered;
        }
        let Some(metadata_cond) = Self::extract_metadata_filter(cond) else {
            return filtered;
        };
        let filter = crate::filter::Filter::new(crate::filter::Condition::from(metadata_cond));
        filtered
            .into_iter()
            .filter(|r| match r.point.payload.as_ref() {
                Some(p) => filter.matches(p),
                None => filter.matches(&serde_json::Value::Null),
            })
            .take(execution_limit)
            .collect()
    }

    /// Applies all similarity cascade filters sequentially.
    fn apply_similarity_cascade(
        &self,
        candidates: Vec<SearchResult>,
        first_similarity: &(String, Vec<f32>, crate::velesql::CompareOp, f64),
        similarity_conditions: &[(String, Vec<f32>, crate::velesql::CompareOp, f64)],
        filter_k: usize,
    ) -> Vec<SearchResult> {
        let (field, vec, op, threshold) = first_similarity;
        let mut filtered =
            self.filter_by_similarity(candidates, field, vec, *op, *threshold, filter_k);
        for (sim_field, sim_vec, sim_op, sim_threshold) in similarity_conditions.iter().skip(1) {
            filtered = self.filter_by_similarity(
                filtered,
                sim_field,
                sim_vec,
                *sim_op,
                *sim_threshold,
                filter_k,
            );
        }
        filtered
    }

    /// Handles the `(NEAR vector, no similarity(), optional metadata filter)` path.
    #[allow(clippy::too_many_arguments)] // All arguments come from dispatch_vector_query.
    fn dispatch_near_with_filter(
        &self,
        vector: &[f32],
        cond: &crate::velesql::Condition,
        execution_limit: usize,
        skip_metadata_prefilter_for_graph_or: bool,
        search_opts: &QuerySearchOptions,
        cbo_strategy: crate::velesql::ExecutionStrategy,
        cbo_over_fetch: usize,
    ) -> Result<Vec<SearchResult>> {
        if let Some(text_query) = Self::extract_match_query(cond) {
            let fusion = search_opts.fusion_clause.as_ref();
            // Bug #474: Extract co-occurring metadata filters (e.g. `category = 'tech'`)
            // before fusing. Without this, metadata conditions alongside MATCH
            // are silently dropped.
            // Bug #6: route through hybrid_search_with_clause so the FUSION
            // strategy / graph_weight take effect instead of always running RRF.
            let filter = Self::extract_metadata_filter(cond)
                .map(|c| crate::filter::Filter::new(crate::filter::Condition::from(c)));
            return self.hybrid_search_with_clause(
                vector,
                &text_query,
                execution_limit,
                fusion,
                filter.as_ref(),
            );
        }
        let cbo_search_k = execution_limit
            .saturating_mul(cbo_over_fetch)
            .min(MAX_LIMIT);
        if skip_metadata_prefilter_for_graph_or {
            return self.search_with_opts(vector, execution_limit, search_opts);
        }
        if let Some(metadata_cond) = Self::extract_metadata_filter(cond) {
            let filter = crate::filter::Filter::new(crate::filter::Condition::from(metadata_cond));
            return self.dispatch_vector_with_strategy(
                vector,
                &filter,
                cbo_strategy,
                cbo_search_k,
                execution_limit,
                search_opts,
            );
        }
        self.search_with_opts(vector, execution_limit, search_opts)
    }

    /// Dispatches a filtered vector query according to the CBO strategy
    /// (GraphFirst, Parallel, or the default VectorFirst path).
    fn dispatch_vector_with_strategy(
        &self,
        vector: &[f32],
        filter: &crate::filter::Filter,
        cbo_strategy: crate::velesql::ExecutionStrategy,
        cbo_search_k: usize,
        execution_limit: usize,
        search_opts: &QuerySearchOptions,
    ) -> Result<Vec<SearchResult>> {
        match cbo_strategy {
            crate::velesql::ExecutionStrategy::GraphFirst => {
                Ok(self.scan_and_score_by_vector(filter, vector, execution_limit))
            }
            crate::velesql::ExecutionStrategy::Parallel => {
                // R2 (#1390): run the GraphFirst scan and the VectorFirst HNSW
                // branch CONCURRENTLY via `rayon::join`. Both legs are read-only
                // over immutable collection data and the merge below is a
                // best-score-by-id union (order-insensitive), so the concurrent
                // result set is byte-for-byte identical to the former sequential
                // execution — only wall-clock latency changes. The two closures
                // each take read locks only (no writer during a query), so there
                // is no lock-ordering hazard, mirroring the concurrent hybrid
                // dense/sparse path in `hybrid_sparse::execute_both_branches`.
                #[cfg(feature = "persistence")]
                let (graph_results, vector_results) = rayon::join(
                    || self.scan_and_score_by_vector(filter, vector, execution_limit),
                    || self.search_with_filter_and_opts(vector, cbo_search_k, filter, search_opts),
                );
                #[cfg(not(feature = "persistence"))]
                let (graph_results, vector_results) = (
                    self.scan_and_score_by_vector(filter, vector, execution_limit),
                    self.search_with_filter_and_opts(vector, cbo_search_k, filter, search_opts),
                );
                let vector_results = vector_results?;
                let higher = self.storage.config.read().metric.higher_is_better();
                Ok(merge_select_parallel_results(
                    graph_results,
                    vector_results,
                    higher,
                    execution_limit,
                ))
            }
            _ => self.search_with_filter_and_opts(vector, cbo_search_k, filter, search_opts),
        }
    }

    /// Handles the metadata-only (`(None, None, Some(cond))`) query path.
    fn dispatch_metadata_only(
        &self,
        cond: &crate::velesql::Condition,
        execution_limit: usize,
        skip_metadata_prefilter_for_graph_or: bool,
    ) -> Result<Vec<SearchResult>> {
        if let crate::velesql::Condition::Match(ref m) = cond {
            return self.text_search(&m.query, execution_limit);
        }
        let empty_filter =
            || crate::filter::Filter::new(crate::filter::Condition::And { conditions: vec![] });
        if skip_metadata_prefilter_for_graph_or {
            return Ok(self.execute_scan_query(&empty_filter(), execution_limit, None));
        }
        let Some(metadata_cond) = Self::extract_metadata_filter(cond) else {
            return Ok(self.execute_scan_query(&empty_filter(), execution_limit, None));
        };
        Ok(self.dispatch_metadata_filter(cond, &metadata_cond, execution_limit))
    }

    /// Resolves a metadata filter by probing bitmap → indexed → BM25 → scan paths.
    ///
    /// Extracted from `dispatch_metadata_only` to keep cyclomatic complexity ≤ 8.
    fn dispatch_metadata_filter(
        &self,
        cond: &crate::velesql::Condition,
        metadata_cond: &crate::velesql::Condition,
        execution_limit: usize,
    ) -> Vec<SearchResult> {
        // Fast path: use bitmap from secondary indexes (same mechanism as
        // search_with_filter). This handles AND conditions, Eq lookups, and
        // range queries via the bitmap infrastructure.
        let filter =
            crate::filter::Filter::new(crate::filter::Condition::from(metadata_cond.clone()));
        if let Some(bitmap_results) =
            self.try_bitmap_prefilter(&filter, metadata_cond, execution_limit)
        {
            return bitmap_results;
        }

        tracing::debug!("dispatch_metadata_only: trying indexed path");
        if let Some(indexed) = self.execute_indexed_metadata_query(metadata_cond, execution_limit) {
            tracing::debug!("dispatch_metadata_only: indexed path succeeded");
            return indexed;
        }
        tracing::debug!("dispatch_metadata_only: indexed path returned None, trying mirror");

        // ColumnStore payload mirror: typed columnar bitmap scan when no
        // secondary index covers the condition. Built adaptively once enough
        // full-scan debt accumulates (see collection/payload_mirror).
        if let Some(mirror_results) = self.try_mirror_filter(&filter, execution_limit) {
            return mirror_results;
        }

        // Try BM25 text search for LIKE conditions before falling back to full scan.
        // When a LIKE pattern contains a word-like substring (e.g. `%google%`),
        // BM25 can narrow candidates significantly faster than a sequential scan.
        if let Some(like_results) = self.try_like_via_text_index(cond, execution_limit) {
            return like_results;
        }

        let filter =
            crate::filter::Filter::new(crate::filter::Condition::from(metadata_cond.clone()));
        self.execute_scan_query(&filter, execution_limit, Some(metadata_cond))
    }

    /// Cost-model-driven fallback decision (audit F-4.7, issue #1391).
    ///
    /// Replaces the arbitrary `execution_limit * 50 .max(1000)` candidate budget
    /// that decided "too many index/bitmap candidates → sequential full scan".
    /// Compares the estimated cost of hydrating `candidate_count` index/bitmap
    /// candidates against a full sequential scan that early-exits once
    /// `exec_limit` matches are produced, using the same [`CostEstimator`](crate::velesql::CostEstimator)
    /// the ORDER BY router (`ordered_index_scan.rs`) already relies on.
    ///
    /// Returns `true` when the candidate scan is the cheaper (or equal) path.
    ///
    /// # Guardrails
    /// - `.max(1000)` floor: candidate sets of ≤ 1000 always take the candidate
    ///   path — a noisy estimate must never penalise a trivially small set.
    /// - `cond == None` (SELECT * / empty filter — no indexed Eq exists on those
    ///   paths anyway): falls back to the historical
    ///   `execution_limit * 50 .max(1000)` budget.
    /// - Empty / un-analysed collection (`total == 0`): candidate path.
    ///
    /// The choice **never** affects the result set — both branches post-filter
    /// the same predicate and yield identical rows; only the physical path
    /// differs, so switching paths is functionally safe.
    pub(super) fn prefer_candidate_scan(
        &self,
        candidate_count: usize,
        exec_limit: usize,
        cond: Option<&crate::velesql::Condition>,
    ) -> bool {
        Self::candidate_scan_preferred(&self.get_stats(), candidate_count, exec_limit, cond)
    }

    /// Pure cost comparison behind [`Self::prefer_candidate_scan`], split out so
    /// the routing decision can be unit-tested against hand-built
    /// [`CollectionStats`](crate::collection::stats::CollectionStats) without a
    /// live collection.
    // Reason: usize/u64 → f64 for cardinality ratios; ±1 ULP has no operational
    // impact on a routing decision that is guarded by a `.max(1000)` floor.
    #[allow(clippy::cast_precision_loss)]
    pub(super) fn candidate_scan_preferred(
        stats: &crate::collection::stats::CollectionStats,
        candidate_count: usize,
        exec_limit: usize,
        cond: Option<&crate::velesql::Condition>,
    ) -> bool {
        /// Floor guard: tiny candidate sets are always cheaper to hydrate
        /// directly than to scan the whole collection.
        const CANDIDATE_SCAN_FLOOR: usize = 1000;

        if candidate_count <= CANDIDATE_SCAN_FLOOR {
            return true;
        }

        // No condition tree available (SELECT * / empty filter): preserve the
        // historical budget so behaviour on those paths is unchanged.
        let Some(cond) = cond else {
            return candidate_count <= exec_limit.saturating_mul(50).max(CANDIDATE_SCAN_FLOOR);
        };

        let total = stats.total_points.max(stats.row_count);
        if total == 0 {
            return true;
        }
        let total_f = total as f64;

        let estimator = crate::velesql::CostEstimator::new(stats);
        // Clamp selectivity away from 0 so the early-exit row estimate stays
        // finite; a single-row lower bound is the tightest meaningful floor.
        let selectivity = estimator
            .estimate_condition_selectivity(cond)
            .clamp(1.0 / total_f, 1.0);

        // A full sequential scan with early exit visits ≈ exec_limit /
        // selectivity rows before it collects `exec_limit` matches (bounded by
        // the collection size).
        let full_scan_rows = ((exec_limit.max(1) as f64) / selectivity)
            .min(total_f)
            .max(1.0);

        // Cost both paths through the calibrated filter-cost model. Expressing
        // each cardinality as a selectivity (`rows / total`) reuses
        // `estimate_filter_cost_from_selectivity`, so calibrated I/O and CPU
        // factors weight both sides identically; the decision is driven by the
        // histogram-/cardinality-calibrated selectivity feeding `full_scan_rows`.
        let candidate_cost = estimator
            .estimate_filter_cost_from_selectivity(candidate_count as f64 / total_f)
            .total();
        let full_scan_cost = estimator
            .estimate_filter_cost_from_selectivity(full_scan_rows / total_f)
            .total();

        candidate_cost <= full_scan_cost
    }

    /// Attempts a bitmap-prefiltered scan when the candidate set is bounded.
    ///
    /// Returns `Some(results)` when the bitmap path is viable (empty result or
    /// a candidate count the cost model deems cheaper to scan than the full
    /// collection). Returns `None` to let the caller fall through to
    /// indexed/scan paths.
    fn try_bitmap_prefilter(
        &self,
        filter: &crate::filter::Filter,
        cond: &crate::velesql::Condition,
        execution_limit: usize,
    ) -> Option<Vec<SearchResult>> {
        let bitmap = self.build_prefilter_bitmap(filter)?;
        if bitmap.is_empty() {
            return Some(Vec::new());
        }
        let candidate_ids: Vec<u64> = bitmap.iter().map(u64::from).collect();
        if self.prefer_candidate_scan(candidate_ids.len(), execution_limit, Some(cond)) {
            return Some(self.scan_ids_with_filter(&candidate_ids, filter, execution_limit));
        }
        // Too many bitmap hits — fall through to scan with early exit
        None
    }

    /// Attempts a `ColumnStore` payload-mirror scan for the filter.
    ///
    /// The mirror returns a candidate id superset from typed columnar
    /// bitmaps; `scan_ids_with_filter` post-filters with the JSON filter,
    /// so results are exactly those of the sequential scan path. Candidates
    /// are hydrated in chunks so broad matches (e.g. `!=` over most rows)
    /// stay memory-bounded and benefit from early exit at the limit.
    ///
    /// Returns `None` when the mirror is not built (insufficient scan debt),
    /// or the condition is not answerable from columnar data.
    fn try_mirror_filter(
        &self,
        filter: &crate::filter::Filter,
        execution_limit: usize,
    ) -> Option<Vec<SearchResult>> {
        const HYDRATION_CHUNK: usize = 1024;
        let candidate_ids = self.mirror_candidate_ids(&filter.condition)?;
        let mut results = Vec::new();
        for chunk in candidate_ids.chunks(HYDRATION_CHUNK) {
            let remaining = execution_limit.saturating_sub(results.len());
            if remaining == 0 {
                break;
            }
            results.extend(self.scan_ids_with_filter(chunk, filter, remaining));
        }
        Some(results)
    }

    /// Attempts to accelerate a LIKE condition using the BM25 text index.
    ///
    /// Extracts the word-like core from a `%word%` pattern and queries BM25
    /// for candidate document IDs. The full condition is then post-filtered
    /// over those candidates instead of scanning the entire collection.
    ///
    /// Returns `Some(results)` only when BM25 found enough candidates to
    /// fill the limit. When BM25 returns fewer matches than requested, the
    /// result set may be incomplete (BM25 tokenization differs from LIKE
    /// substring matching), so we return `None` to let the caller fall
    /// through to a full sequential scan.
    ///
    /// Returns `None` when:
    /// - No LIKE condition is found in the condition tree
    /// - The extracted word is too short (< 3 chars) for meaningful BM25 lookup
    /// - BM25 returns no candidates (fall through to sequential scan)
    /// - BM25 candidates yield fewer than `limit` matches (incomplete set)
    fn try_like_via_text_index(
        &self,
        cond: &crate::velesql::Condition,
        limit: usize,
    ) -> Option<Vec<SearchResult>> {
        let candidate_ids = self.bm25_candidates_for_like(cond, limit)?;
        let filter = crate::filter::Filter::new(crate::filter::Condition::from(cond.clone()));
        let results = self.collect_matching_points(&candidate_ids, &filter, limit);

        // Only return BM25 results when we filled the limit — otherwise the
        // result set may be incomplete because BM25 tokenization differs from
        // LIKE substring matching (e.g., "analytics.google.com" won't match
        // BM25 for "google" but should match LIKE '%google%').
        if results.len() >= limit {
            Some(results)
        } else {
            None // Fall through to full sequential scan
        }
    }

    /// Extracts BM25 candidate IDs for a LIKE condition, if the pattern yields
    /// a meaningful word and BM25 returns any match.
    fn bm25_candidates_for_like(
        &self,
        cond: &crate::velesql::Condition,
        limit: usize,
    ) -> Option<Vec<u64>> {
        let pattern = Self::extract_like_pattern(cond)?;

        // Extract the word-like core from the pattern (strip leading/trailing %).
        let word = pattern.trim_matches('%');
        if word.len() < 3 {
            return None;
        }

        // Use BM25 text index to find candidates (over-fetch 10× for post-filter headroom).
        let text_results = self
            .storage
            .text_index
            .search(word, limit.saturating_mul(10));
        if text_results.is_empty() {
            return None;
        }

        Some(text_results.iter().map(|(id, _)| *id).collect())
    }

    /// Scans a candidate ID list, returning up to `limit` points that match the filter.
    fn collect_matching_points(
        &self,
        candidate_ids: &[u64],
        filter: &crate::filter::Filter,
        limit: usize,
    ) -> Vec<SearchResult> {
        let mut results = Vec::new();
        for point in self.get(candidate_ids).into_iter().flatten() {
            let payload = point.payload.clone().unwrap_or(serde_json::Value::Null);
            if filter.matches(&payload) {
                results.push(SearchResult::new(point, 1.0));
                if results.len() >= limit {
                    break;
                }
            }
        }
        results
    }

    /// Recursively extracts the first LIKE pattern from a condition tree.
    fn extract_like_pattern(cond: &crate::velesql::Condition) -> Option<String> {
        match cond {
            crate::velesql::Condition::Like(like) => Some(like.pattern.clone()),
            crate::velesql::Condition::And(left, right) => {
                Self::extract_like_pattern(left).or_else(|| Self::extract_like_pattern(right))
            }
            crate::velesql::Condition::Group(inner) => Self::extract_like_pattern(inner),
            _ => None,
        }
    }

    #[allow(clippy::too_many_arguments)] // All arguments come from query extraction in the caller.
    pub(super) fn dispatch_vector_query(
        &self,
        vector_search: Option<&Vec<f32>>,
        first_similarity: Option<&(String, Vec<f32>, crate::velesql::CompareOp, f64)>,
        similarity_conditions: &[(String, Vec<f32>, crate::velesql::CompareOp, f64)],
        filter_condition: Option<&crate::velesql::Condition>,
        execution_limit: usize,
        skip_metadata_prefilter_for_graph_or: bool,
        search_opts: &QuerySearchOptions,
        cbo_strategy: crate::velesql::ExecutionStrategy,
        cbo_over_fetch: usize,
    ) -> Result<Vec<SearchResult>> {
        // `cbo_strategy` (VectorFirst / GraphFirst / Parallel) only has a
        // physically distinct realization on the NEAR + metadata-filter arm
        // below, which routes through `dispatch_vector_with_strategy`. Every
        // other arm has exactly one sensible physical plan, so it deliberately
        // ignores `cbo_strategy` (documented per-arm). This is intentional, not
        // an oversight: forcing a graph/parallel shape onto these arms would add
        // machinery with no cost benefit (audit F-2.15, #1390).
        match (vector_search, first_similarity, filter_condition) {
            // similarity() with optional NEAR vector and optional metadata filter.
            // Strategy IGNORED (deliberate): a similarity() threshold is
            // intrinsically VectorFirst — it must score vector candidates before
            // it can apply the threshold, so GraphFirst/Parallel have no
            // meaningful realization here. The optional metadata filter is
            // applied as a post-filter over the scored candidates.
            (search_vec, Some(sim), filter_cond) => self.dispatch_similarity_query(
                search_vec.map(Vec::as_slice),
                sim,
                similarity_conditions,
                filter_cond,
                execution_limit,
                skip_metadata_prefilter_for_graph_or,
                search_opts,
            ),
            // NEAR + metadata filter (no similarity threshold). This is the ONLY
            // arm that HONORS `cbo_strategy`: it dispatches to
            // `dispatch_vector_with_strategy`, which realizes VectorFirst,
            // GraphFirst, and Parallel as three physically distinct plans.
            (Some(vector), None, Some(cond)) => self.dispatch_near_with_filter(
                vector,
                cond,
                execution_limit,
                skip_metadata_prefilter_for_graph_or,
                search_opts,
                cbo_strategy,
                cbo_over_fetch,
            ),
            // Pure NEAR (no filter, no similarity threshold).
            // Strategy IGNORED (deliberate): with no metadata/graph predicate
            // there is no second leg to run first or in parallel — only the
            // VectorFirst HNSW search exists, so the strategy is moot.
            (Some(vector), None, None) => {
                self.dispatch_pure_near(vector, execution_limit, search_opts)
            }
            // Metadata-only (no vector query at all).
            // Strategy IGNORED (deliberate): `ExecutionStrategy` orders a vector
            // search relative to a filter; with no vector query there is nothing
            // to order — the path is a pure index/bitmap/scan resolution.
            (None, None, Some(cond)) => self.dispatch_metadata_only(
                cond,
                execution_limit,
                skip_metadata_prefilter_for_graph_or,
            ),
            // SELECT * (no WHERE, no vector).
            // Strategy IGNORED (deliberate): no predicate and no vector query —
            // a full sequential scan is the only possible plan.
            (None, None, None) => Ok(self.execute_scan_query(
                &crate::filter::Filter::new(crate::filter::Condition::And { conditions: vec![] }),
                execution_limit,
                None,
            )),
        }
    }

    /// Handles the similarity() path with optional NEAR vector and optional metadata filter.
    #[allow(clippy::too_many_arguments)] // All arguments come from dispatch_vector_query.
    fn dispatch_similarity_query(
        &self,
        search_vector: Option<&[f32]>,
        sim: &(String, Vec<f32>, crate::velesql::CompareOp, f64),
        similarity_conditions: &[(String, Vec<f32>, crate::velesql::CompareOp, f64)],
        filter_cond: Option<&crate::velesql::Condition>,
        execution_limit: usize,
        skip_metadata_prefilter_for_graph_or: bool,
        search_opts: &QuerySearchOptions,
    ) -> Result<Vec<SearchResult>> {
        let k = execution_limit
            .saturating_mul(10 * similarity_conditions.len().max(1))
            .min(MAX_LIMIT);
        let search_vec = search_vector.unwrap_or(&sim.1);
        let candidates = self.search_with_opts(search_vec, k, search_opts)?;
        let filtered = self.apply_similarity_cascade(
            candidates,
            sim,
            similarity_conditions,
            execution_limit.saturating_mul(2),
        );
        Ok(Self::apply_optional_metadata_filter(
            filtered,
            filter_cond,
            skip_metadata_prefilter_for_graph_or,
            execution_limit,
        ))
    }

    /// Handles the pure NEAR path (no similarity threshold, no metadata filter).
    fn dispatch_pure_near(
        &self,
        vector: &[f32],
        execution_limit: usize,
        search_opts: &QuerySearchOptions,
    ) -> Result<Vec<SearchResult>> {
        self.search_with_opts(vector, execution_limit, search_opts)
    }
}

#[cfg(test)]
#[path = "execution_paths_tests.rs"]
mod execution_paths_tests;

/// Merges GraphFirst and VectorFirst `SearchResult` sets for the SELECT Parallel
/// path (sequential execution, union semantics — best score wins per ID).
fn merge_select_parallel_results(
    graph: Vec<SearchResult>,
    vector: Vec<SearchResult>,
    higher_is_better: bool,
    limit: usize,
) -> Vec<SearchResult> {
    let mut by_id: rustc_hash::FxHashMap<u64, SearchResult> =
        rustc_hash::FxHashMap::with_capacity_and_hasher(
            graph.len() + vector.len(),
            rustc_hash::FxBuildHasher,
        );
    for r in graph.into_iter().chain(vector) {
        by_id
            .entry(r.point.id)
            .and_modify(|existing| {
                let better = if higher_is_better {
                    r.score > existing.score
                } else {
                    r.score < existing.score
                };
                if better {
                    *existing = r.clone();
                }
            })
            .or_insert(r);
    }
    let mut merged: Vec<SearchResult> = by_id.into_values().collect();
    if higher_is_better {
        merged.sort_unstable_by(|a, b| b.score.total_cmp(&a.score));
    } else {
        merged.sort_unstable_by(|a, b| a.score.total_cmp(&b.score));
    }
    merged.truncate(limit);
    merged
}