velesdb-core 1.7.2

//! Query plan explanation for `VelesQL`.
//!
//! This module provides EXPLAIN functionality to display query execution plans.
//!
//! # Example
//!
//! ```ignore
//! use velesdb_core::velesql::{Parser, QueryPlan};
//!
//! let query = Parser::parse("SELECT * FROM docs WHERE vector NEAR $v LIMIT 10")?;
//! let plan = QueryPlan::from_select(&query.select);
//! println!("{}", plan.to_tree());
//! ```

mod formatter;

use serde::{Deserialize, Serialize};
use std::collections::HashSet;

use super::ast::{Condition, SelectStatement};
use crate::collection::search::query::match_planner::{
    CollectionStats, MatchExecutionStrategy, MatchQueryPlanner,
};
use crate::velesql::MatchClause;

/// Query execution plan generated by EXPLAIN.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct QueryPlan {
    /// Root node of the plan tree.
    pub root: PlanNode,
    /// Estimated execution cost in milliseconds.
    pub estimated_cost_ms: f64,
    /// Index type used (if any).
    pub index_used: Option<IndexType>,
    /// Filter strategy.
    pub filter_strategy: FilterStrategy,
    /// Whether this plan was served from the compiled plan cache (CACHE-02).
    #[serde(skip_serializing_if = "Option::is_none")]
    pub cache_hit: Option<bool>,
    /// How many times this cached plan has been reused (CACHE-02).
    ///
    /// The value is the `reuse_count` atomically incremented inside the cache on every
    /// successful `get()`. A value of 1 means the plan was found in cache and this is
    /// its first reuse. The count includes the current call to `explain_query`.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub plan_reuse_count: Option<u64>,
}

/// A node in the query execution plan tree.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub enum PlanNode {
    /// Vector similarity search operation.
    VectorSearch(VectorSearchPlan),
    /// Metadata filter operation.
    Filter(FilterPlan),
    /// Limit results.
    Limit(LimitPlan),
    /// Offset skip.
    Offset(OffsetPlan),
    /// Table scan (no index).
    TableScan(TableScanPlan),
    /// Property index lookup (O(1) instead of scan).
    IndexLookup(IndexLookupPlan),
    /// Sequential operations.
    Sequence(Vec<PlanNode>),
    /// MATCH graph traversal (EPIC-046 US-004).
    MatchTraversal(MatchTraversalPlan),
}

/// Vector search plan details.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct VectorSearchPlan {
    /// Collection name.
    pub collection: String,
    /// `ef_search` parameter (for HNSW).
    pub ef_search: u32,
    /// Number of candidates to retrieve.
    pub candidates: u32,
}

/// Filter plan details.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct FilterPlan {
    /// Filter conditions as string representation.
    pub conditions: String,
    /// Estimated selectivity (0.0 - 1.0).
    pub selectivity: f64,
}

/// Limit plan details.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct LimitPlan {
    /// Maximum number of results.
    pub count: u64,
}

/// Offset plan details.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct OffsetPlan {
    /// Number of results to skip.
    pub count: u64,
}

/// Table scan plan (no index used).
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct TableScanPlan {
    /// Collection name.
    pub collection: String,
}

/// Property index lookup plan (O(1) lookup).
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct IndexLookupPlan {
    /// Label being queried.
    pub label: String,
    /// Property name with index.
    pub property: String,
    /// Value being looked up (as string representation).
    pub value: String,
}

/// MATCH traversal plan (EPIC-046 US-004).
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct MatchTraversalPlan {
    /// Execution strategy chosen by planner.
    pub strategy: String,
    /// Start node labels.
    pub start_labels: Vec<String>,
    /// Maximum traversal depth.
    pub max_depth: u32,
    /// Number of relationships in pattern.
    pub relationship_count: usize,
    /// Has similarity condition.
    pub has_similarity: bool,
    /// Similarity threshold (if any).
    pub similarity_threshold: Option<f32>,
}

/// EXPLAIN output with optional ANALYZE stats (EPIC-046 US-004).
#[allow(dead_code)] // Used by API consumers
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ExplainOutput {
    /// The query plan.
    pub plan: QueryPlan,
    /// Actual execution statistics (only with ANALYZE).
    #[serde(skip_serializing_if = "Option::is_none")]
    pub actual_stats: Option<ActualStats>,
}

/// Actual execution statistics for EXPLAIN ANALYZE.
#[allow(dead_code)] // Used by API consumers
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ActualStats {
    /// Actual number of rows returned.
    pub actual_rows: u64,
    /// Actual execution time in milliseconds.
    pub actual_time_ms: f64,
    /// Number of loop iterations.
    pub loops: u64,
    /// Number of nodes visited (for graph traversal).
    pub nodes_visited: u64,
    /// Number of edges traversed.
    pub edges_traversed: u64,
}

/// Type of index used in the query.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub enum IndexType {
    /// HNSW index for vector search.
    Hnsw,
    /// Flat index (brute force).
    Flat,
    /// Binary quantization index.
    BinaryQuantization,
    /// Property index for equality lookups.
    Property,
}

/// Strategy for applying filters.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)]
pub enum FilterStrategy {
    /// No filter.
    #[default]
    None,
    /// Pre-filtering: filter before vector search (high selectivity).
    PreFilter,
    /// Post-filtering: filter after vector search (low selectivity).
    PostFilter,
}

impl QueryPlan {
    /// Creates a new query plan from a SELECT statement.
    #[must_use]
    pub fn from_select(stmt: &SelectStatement) -> Self {
        Self::from_select_with_indexed_fields(stmt, &HashSet::new())
    }

    /// Creates a new query plan from SELECT with known indexed metadata fields.
    #[must_use]
    pub fn from_select_with_indexed_fields(
        stmt: &SelectStatement,
        indexed_fields: &HashSet<String>,
    ) -> Self {
        let mut has_vector_search = false;
        let mut filter_conditions = Vec::new();
        let mut index_lookup = None;

        if let Some(ref condition) = stmt.where_clause {
            Self::analyze_condition(condition, &mut has_vector_search, &mut filter_conditions);
            index_lookup = Self::extract_index_lookup(condition, indexed_fields);
        }

        let (mut nodes, index_used) = Self::build_scan_node(stmt, has_vector_search, index_lookup);
        let filter_strategy = Self::append_filter_nodes(&mut nodes, &filter_conditions, stmt);

        Self::assemble_plan(nodes, index_used, filter_strategy, has_vector_search)
    }

    /// Collapses a `Vec<PlanNode>` into a single root, estimates cost, and builds the plan.
    fn assemble_plan(
        mut nodes: Vec<PlanNode>,
        index_used: Option<IndexType>,
        filter_strategy: FilterStrategy,
        has_vector_search: bool,
    ) -> Self {
        let root = if nodes.len() == 1 {
            nodes.swap_remove(0)
        } else {
            PlanNode::Sequence(nodes)
        };
        let estimated_cost_ms = Self::estimate_cost(&root, has_vector_search);
        Self {
            root,
            estimated_cost_ms,
            index_used,
            filter_strategy,
            cache_hit: None,
            plan_reuse_count: None,
        }
    }

    /// Builds the primary scan node based on search type.
    fn build_scan_node(
        stmt: &SelectStatement,
        has_vector_search: bool,
        index_lookup: Option<(String, String)>,
    ) -> (Vec<PlanNode>, Option<IndexType>) {
        let mut nodes = Vec::new();
        let index_used;

        if has_vector_search {
            index_used = Some(IndexType::Hnsw);
            let candidates = u32::try_from(stmt.limit.unwrap_or(50)).unwrap_or(u32::MAX);
            nodes.push(PlanNode::VectorSearch(VectorSearchPlan {
                collection: stmt.from.clone(),
                ef_search: 100,
                candidates,
            }));
        } else if let Some((property, value)) = index_lookup {
            index_used = Some(IndexType::Property);
            nodes.push(PlanNode::IndexLookup(IndexLookupPlan {
                label: stmt.from.clone(),
                property,
                value,
            }));
        } else {
            index_used = None;
            nodes.push(PlanNode::TableScan(TableScanPlan {
                collection: stmt.from.clone(),
            }));
        }

        (nodes, index_used)
    }

    /// Appends filter, offset, and limit nodes; returns the filter strategy.
    fn append_filter_nodes(
        nodes: &mut Vec<PlanNode>,
        filter_conditions: &[String],
        stmt: &SelectStatement,
    ) -> FilterStrategy {
        let mut filter_strategy = FilterStrategy::None;

        if !filter_conditions.is_empty() {
            let selectivity = Self::estimate_selectivity(filter_conditions);
            filter_strategy = if selectivity > 0.1 {
                FilterStrategy::PostFilter
            } else {
                FilterStrategy::PreFilter
            };
            nodes.push(PlanNode::Filter(FilterPlan {
                conditions: filter_conditions.join(" AND "),
                selectivity,
            }));
        }

        if let Some(offset) = stmt.offset {
            nodes.push(PlanNode::Offset(OffsetPlan { count: offset }));
        }
        if let Some(limit) = stmt.limit {
            nodes.push(PlanNode::Limit(LimitPlan { count: limit }));
        }

        filter_strategy
    }

    /// Analyzes a condition to extract vector search and filter info.
    fn analyze_condition(
        condition: &Condition,
        has_vector_search: &mut bool,
        filter_conditions: &mut Vec<String>,
    ) {
        match condition {
            Condition::VectorSearch(_)
            | Condition::VectorFusedSearch(_)
            | Condition::SparseVectorSearch(_)
            | Condition::Similarity(_) => {
                *has_vector_search = true;
            }
            Condition::Comparison(cmp) => {
                filter_conditions.push(format!("{} {} ?", cmp.column, cmp.operator.as_str()));
            }
            Condition::In(inc) => {
                let op = if inc.negated { "NOT IN" } else { "IN" };
                filter_conditions.push(format!("{} {op} (...)", inc.column));
            }
            Condition::Between(btw) => {
                filter_conditions.push(format!("{} BETWEEN ? AND ?", btw.column));
            }
            Condition::Like(lk) => {
                filter_conditions.push(format!("{} LIKE ?", lk.column));
            }
            Condition::IsNull(isn) => {
                let op = if isn.is_null {
                    "IS NULL"
                } else {
                    "IS NOT NULL"
                };
                filter_conditions.push(format!("{} {op}", isn.column));
            }
            Condition::Match(m) => {
                filter_conditions.push(format!("{} MATCH ?", m.column));
            }
            Condition::GraphMatch(_) => {
                filter_conditions.push("MATCH (...)".to_string());
            }
            Condition::And(left, right) | Condition::Or(left, right) => {
                Self::analyze_condition(left, has_vector_search, filter_conditions);
                Self::analyze_condition(right, has_vector_search, filter_conditions);
            }
            Condition::Not(inner) | Condition::Group(inner) => {
                Self::analyze_condition(inner, has_vector_search, filter_conditions);
            }
        }
    }

    fn extract_index_lookup(
        condition: &Condition,
        indexed_fields: &HashSet<String>,
    ) -> Option<(String, String)> {
        if let Condition::Comparison(cmp) = condition {
            if cmp.operator == crate::velesql::CompareOp::Eq && indexed_fields.contains(&cmp.column)
            {
                return Some((cmp.column.clone(), format!("{:?}", cmp.value)));
            }
        }
        None
    }

    /// Estimates selectivity (placeholder - would need statistics in production).
    pub(crate) fn estimate_selectivity(conditions: &[String]) -> f64 {
        // Heuristic: more conditions = lower selectivity
        let base = 0.5_f64;
        base.powi(i32::try_from(conditions.len()).unwrap_or(i32::MAX))
    }

    /// Estimates execution cost in milliseconds.
    fn estimate_cost(root: &PlanNode, has_vector_search: bool) -> f64 {
        let base_cost = if has_vector_search { 0.05 } else { 1.0 };

        match root {
            PlanNode::Sequence(nodes) => nodes
                .iter()
                .fold(base_cost, |acc, node| acc + Self::node_cost(node)),
            _ => base_cost + Self::node_cost(root),
        }
    }

    pub(crate) fn node_cost(node: &PlanNode) -> f64 {
        match node {
            PlanNode::VectorSearch(_) => 0.05,
            PlanNode::Filter(f) => 0.01 * (1.0 - f.selectivity),
            PlanNode::Limit(_) | PlanNode::Offset(_) => 0.001,
            PlanNode::TableScan(_) => 1.0,
            PlanNode::IndexLookup(_) => 0.0001, // O(1) lookup is very fast
            PlanNode::Sequence(nodes) => nodes.iter().map(Self::node_cost).sum(),
            PlanNode::MatchTraversal(mt) => {
                // Cost depends on depth and strategy
                let base = 0.1;
                let depth_factor = f64::from(mt.max_depth) * 0.05;
                let similarity_factor = if mt.has_similarity { 0.05 } else { 0.0 };
                base + depth_factor + similarity_factor
            }
        }
    }

    /// Creates a new query plan from a MATCH clause (EPIC-046 US-004).
    #[must_use]
    pub fn from_match(match_clause: &MatchClause, stats: &CollectionStats) -> Self {
        let strategy = MatchQueryPlanner::plan(match_clause, stats);
        let strategy_explanation = MatchQueryPlanner::explain(&strategy);

        let (start_labels, max_depth, has_similarity, similarity_threshold) =
            Self::extract_strategy_info(&strategy);

        let relationship_count = match_clause
            .patterns
            .first()
            .map_or(0, |p| p.relationships.len());

        let traversal = PlanNode::MatchTraversal(MatchTraversalPlan {
            strategy: strategy_explanation,
            start_labels,
            max_depth,
            relationship_count,
            has_similarity,
            similarity_threshold,
        });

        let mut nodes = vec![traversal];
        if let Some(limit) = match_clause.return_clause.limit {
            nodes.push(PlanNode::Limit(LimitPlan { count: limit }));
        }

        let index_used = if has_similarity {
            Some(IndexType::Hnsw)
        } else {
            None
        };

        Self::assemble_plan(nodes, index_used, FilterStrategy::None, has_similarity)
    }

    /// Extracts traversal parameters from a `MatchExecutionStrategy`.
    fn extract_strategy_info(
        strategy: &MatchExecutionStrategy,
    ) -> (Vec<String>, u32, bool, Option<f32>) {
        match strategy {
            MatchExecutionStrategy::GraphFirst {
                start_labels,
                max_depth,
            } => (start_labels.clone(), *max_depth, false, None),
            MatchExecutionStrategy::VectorFirst { threshold, .. } => {
                (Vec::new(), 1, true, Some(*threshold))
            }
            MatchExecutionStrategy::Parallel {
                graph_hint,
                vector_hint,
            } => {
                let (labels, depth) = match graph_hint.as_ref() {
                    MatchExecutionStrategy::GraphFirst {
                        start_labels,
                        max_depth,
                    } => (start_labels.clone(), *max_depth),
                    _ => (Vec::new(), 1),
                };
                let threshold = match vector_hint.as_ref() {
                    MatchExecutionStrategy::VectorFirst { threshold, .. } => Some(*threshold),
                    _ => None,
                };
                (labels, depth, true, threshold)
            }
        }
    }
}

// Rendering, formatting, Display impl, and as_str() methods moved to explain/formatter.rs
// Tests in explain_tests.rs per project rules (tests in separate files)