velesdb_core/collection/graph/

traversal.rs

//! Graph traversal algorithms for multi-hop queries.
//!
//! This module provides BFS-based traversal for variable-length path patterns
//! like `(a)-[*1..3]->(b)` in MATCH clauses.
//!
//! # Streaming Mode (EPIC-019 US-005)
//!
//! For large graphs, the module provides streaming iterators that yield results
//! lazily without loading all visited nodes into memory at once.

use super::EdgeStore;
use rustc_hash::{FxHashMap, FxHashSet};
use smallvec::SmallVec;
use std::collections::VecDeque;
use std::time::Instant;

/// Number of queue pops between wall-clock deadline checks.
///
/// Gating `Instant::now()` behind this counter keeps the per-node clock cost
/// negligible while still aborting a runaway dense-graph traversal within a
/// bounded number of expansions.
pub(crate) const DEADLINE_CHECK_INTERVAL: u32 = 1024;

/// Returns `true` when a wall-clock `deadline` is set and has been reached.
///
/// Callers gate this behind a per-node counter (every
/// [`DEADLINE_CHECK_INTERVAL`] pops) so `Instant::now()` is not called per
/// node; the `None` case is a single branch with no clock read.
#[inline]
pub(super) fn deadline_exceeded(deadline: Option<Instant>) -> bool {
    deadline.is_some_and(|d| Instant::now() >= d)
}

/// Periodic wall-clock deadline gate for a BFS/DFS loop head.
///
/// Increments `nodes_since_check`; only every [`DEADLINE_CHECK_INTERVAL`] pops
/// does it read the clock via [`deadline_exceeded`]. Returns `true` when the
/// deadline has been reached (caller should `break`). When `deadline` is
/// `None` this is a single branch with no clock read or counter cost.
#[inline]
pub(crate) fn deadline_reached(deadline: Option<Instant>, nodes_since_check: &mut u32) -> bool {
    if deadline.is_none() {
        return false;
    }
    *nodes_since_check += 1;
    if *nodes_since_check < DEADLINE_CHECK_INTERVAL {
        return false;
    }
    *nodes_since_check = 0;
    deadline_exceeded(deadline)
}

/// Default maximum depth for unbounded traversals.
pub const DEFAULT_MAX_DEPTH: u32 = 3;

/// Safety cap for maximum depth to prevent runaway traversals.
/// Only applied when user requests unbounded traversal (*).
///
/// Note: Neo4j and ArangoDB do NOT impose hard limits.
/// 100 is chosen to cover most real-world use cases:
/// - Social networks (6 degrees of separation)
/// - Dependency graphs (deep npm/cargo trees)
/// - Organizational hierarchies
/// - Knowledge graphs
pub const SAFETY_MAX_DEPTH: u32 = 100;

/// Stack-allocated path type used internally by BFS/DFS traversal.
///
/// Most graph queries are depth 1-3 (social networks, knowledge graphs).
/// `SmallVec<[u64; 4]>` stores up to 4 edge IDs inline (32 bytes on stack)
/// and only heap-allocates for deeper traversals, eliminating per-path
/// allocation overhead in the common case.
///
/// Note: [`TraversalResult::path`] uses `Vec<u64>` at the public API
/// boundary. This type is exposed for advanced internal use only.
pub type TraversalPath = SmallVec<[u64; 4]>;

/// Result of a graph traversal operation.
///
/// The `path` field is `Vec<u64>` at the public API boundary. Internally,
/// BFS state uses `SmallVec<[u64; 4]>` to avoid per-path heap allocation
/// for typical depth 1-3 traversals, but this is converted to `Vec` when
/// building the result.
#[derive(Debug, Clone)]
pub struct TraversalResult {
    /// The target node ID reached.
    pub target_id: u64,
    /// The path taken (list of edge IDs).
    pub path: Vec<u64>,
    /// Depth of the traversal (number of hops).
    pub depth: u32,
}

impl TraversalResult {
    /// Creates a new traversal result.
    #[must_use]
    pub fn new(target_id: u64, path: Vec<u64>, depth: u32) -> Self {
        Self {
            target_id,
            path,
            depth,
        }
    }

    /// Creates a traversal result from an internal `SmallVec` path.
    ///
    /// Converts the stack-allocated path to a heap-allocated `Vec` at the
    /// public API boundary.
    #[must_use]
    #[allow(clippy::needless_pass_by_value)]
    #[allow(dead_code)] // Reason: Constructor used by MATCH clause traversal (Wave 6 wiring)
    pub(crate) fn from_smallvec(target_id: u64, path: TraversalPath, depth: u32) -> Self {
        Self {
            target_id,
            path: path.to_vec(),
            depth,
        }
    }
}

/// Configuration for graph traversal.
#[derive(Debug, Clone)]
pub struct TraversalConfig {
    /// Minimum number of hops (inclusive).
    pub min_depth: u32,
    /// Maximum number of hops (inclusive).
    pub max_depth: u32,
    /// Maximum number of results to return.
    pub limit: usize,
    /// Filter by relationship types (empty = all types).
    pub rel_types: Vec<String>,
    /// Optional wall-clock deadline. When set and reached, traversal stops
    /// expanding and returns the partial result accumulated so far (same
    /// convention as the `limit` / `MAX_VISITED_SIZE` guards). `None` (the
    /// default) disables the time bound entirely.
    pub deadline: Option<Instant>,
}

impl Default for TraversalConfig {
    fn default() -> Self {
        Self {
            min_depth: 1,
            max_depth: DEFAULT_MAX_DEPTH,
            limit: 100,
            rel_types: Vec::new(),
            deadline: None,
        }
    }
}

impl TraversalConfig {
    /// Creates a config for a specific range (e.g., *1..3).
    ///
    /// Respects the caller's max_depth without artificial capping.
    /// For unbounded traversals, use `with_unbounded_range()` instead.
    #[must_use]
    pub fn with_range(min: u32, max: u32) -> Self {
        Self {
            min_depth: min,
            max_depth: max,
            ..Self::default()
        }
    }

    /// Creates a config for unbounded traversal (e.g., *1..).
    ///
    /// Applies SAFETY_MAX_DEPTH cap to prevent runaway traversals.
    #[must_use]
    pub fn with_unbounded_range(min: u32) -> Self {
        Self {
            min_depth: min,
            max_depth: SAFETY_MAX_DEPTH,
            ..Self::default()
        }
    }

    /// Sets the result limit.
    #[must_use]
    pub fn with_limit(mut self, limit: usize) -> Self {
        self.limit = limit;
        self
    }

    /// Filters by relationship types.
    #[must_use]
    pub fn with_rel_types(mut self, types: Vec<String>) -> Self {
        self.rel_types = types;
        self
    }

    /// Sets a custom max depth (for advanced use cases).
    #[must_use]
    pub fn with_max_depth(mut self, max_depth: u32) -> Self {
        self.max_depth = max_depth;
        self
    }

    /// Sets a wall-clock deadline after which traversal returns its partial
    /// result instead of continuing to expand.
    #[must_use]
    pub fn with_deadline(mut self, deadline: Instant) -> Self {
        self.deadline = Some(deadline);
        self
    }
}

/// BFS state for traversal (parent-pointer variant).
///
/// Path reconstruction is deferred to result collection via a shared
/// `FxHashMap<u64, (u64, u64)>` (target -> (parent, edge_id)).
/// This eliminates per-edge `SmallVec` clones during frontier expansion.
#[derive(Debug)]
pub(super) struct BfsState {
    /// Current node ID.
    pub(super) node_id: u64,
    /// Current depth.
    pub(super) depth: u32,
}

/// Reconstructs the edge-ID path from `target` back to `source` using parent pointers.
///
/// Walks the parent chain `target -> parent -> ... -> source` collecting edge IDs,
/// then reverses to produce source-to-target order.  Returns an empty `Vec` if
/// `target == source` (the traversal root has no parent entry).
#[must_use]
pub(super) fn reconstruct_path(
    target: u64,
    source: u64,
    parent_map: &FxHashMap<u64, (u64, u64)>,
) -> Vec<u64> {
    let mut path = Vec::new();
    let mut current = target;
    while current != source {
        if let Some(&(parent, edge_id)) = parent_map.get(&current) {
            path.push(edge_id);
            current = parent;
        } else {
            // Unreachable in a correctly-built parent_map; defensive break.
            break;
        }
    }
    path.reverse();
    path
}

/// Direction of edge traversal used by the shared BFS helper.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum BfsDirection {
    /// Follow outgoing edges (source -> target).
    Forward,
    /// Follow incoming edges (target -> source).
    Reverse,
}

/// Core BFS loop shared by forward and reverse traversal.
///
/// Uses parent-pointer map instead of per-state path cloning.
/// Paths are reconstructed lazily only for nodes that qualify as results
/// (depth >= min_depth), eliminating O(visited * avg_depth) clone overhead.
///
/// For forward direction, uses CSR zero-copy path when snapshot exists,
/// avoiding `GraphEdge` cloning. Reverse direction always uses legacy path
/// since the CSR snapshot only covers outgoing edges.
#[must_use]
fn bfs_traverse_directed(
    edge_store: &EdgeStore,
    source_id: u64,
    config: &TraversalConfig,
    direction: BfsDirection,
) -> Vec<TraversalResult> {
    let mut results = Vec::new();
    let mut visited = FxHashSet::default();
    let mut queue = VecDeque::new();
    // Parent-pointer map: target_node -> (parent_node, edge_id).
    // O(visited_nodes) memory vs O(visited_nodes * avg_depth) for path cloning.
    let mut parent_map: FxHashMap<u64, (u64, u64)> = FxHashMap::default();

    // Pre-build a FxHashSet<&str> once for the entire traversal, not per-node.
    let rel_filter: FxHashSet<&str> = config.rel_types.iter().map(String::as_str).collect();

    // CRITICAL FIX: Mark source node as visited before traversal
    // to prevent cycles back to source causing duplicate work
    visited.insert(source_id);

    queue.push_back(BfsState {
        node_id: source_id,
        depth: 0,
    });

    // Use CSR zero-copy path for forward traversal when snapshot exists.
    let use_csr = direction == BfsDirection::Forward && edge_store.has_csr_snapshot();

    // Start at the threshold so an already-expired deadline aborts on the
    // first pop; otherwise the clock is only read every N pops.
    let mut nodes_since_check = DEADLINE_CHECK_INTERVAL;
    while let Some(state) = queue.pop_front() {
        if results.len() >= config.limit {
            break;
        }
        if deadline_reached(config.deadline, &mut nodes_since_check) {
            break;
        }
        if use_csr {
            process_bfs_csr(
                edge_store,
                &state,
                config,
                source_id,
                &rel_filter,
                &mut results,
                &mut visited,
                &mut queue,
                &mut parent_map,
            );
        } else {
            let edges = match direction {
                BfsDirection::Forward => edge_store.get_outgoing(state.node_id),
                BfsDirection::Reverse => edge_store.get_incoming(state.node_id),
            };
            process_bfs_neighbors(
                &edges,
                &state,
                config,
                source_id,
                &rel_filter,
                direction,
                &mut results,
                &mut visited,
                &mut queue,
                &mut parent_map,
            );
        }
    }

    results
}

/// CSR zero-copy BFS expansion for forward traversal.
///
/// Reads target IDs, edge IDs, and interned labels from contiguous memory
/// instead of cloning full `GraphEdge` objects (96+ bytes each).
/// Uses parent-pointer insertion instead of path cloning.
/// Only emits results for newly-discovered nodes (no duplicates).
#[inline]
#[allow(clippy::too_many_arguments)] // Reason: BFS helper passes parent_map alongside traversal state; private fn
fn process_bfs_csr(
    edge_store: &EdgeStore,
    state: &BfsState,
    config: &TraversalConfig,
    source_id: u64,
    rel_filter: &FxHashSet<&str>,
    results: &mut Vec<TraversalResult>,
    visited: &mut FxHashSet<u64>,
    queue: &mut VecDeque<BfsState>,
    parent_map: &mut FxHashMap<u64, (u64, u64)>,
) {
    let Some(snapshot) = edge_store.csr_snapshot() else {
        return;
    };
    let targets = snapshot.neighbors(state.node_id);
    let edge_ids = snapshot.edge_ids(state.node_id);

    for (i, (&target, &eid)) in targets.iter().zip(edge_ids.iter()).enumerate() {
        if results.len() >= config.limit {
            break;
        }
        if let Some(label) = snapshot.label_at(state.node_id, i) {
            if !rel_filter.is_empty() && !rel_filter.contains(label) {
                continue;
            }
        } else if !rel_filter.is_empty() {
            // Edge with unresolvable label excluded when filter is active
            continue;
        }
        process_bfs_candidate(
            target,
            eid,
            state.node_id,
            state.depth,
            config,
            source_id,
            results,
            visited,
            queue,
            parent_map,
        );
    }
}

/// Processes neighbors for a single BFS level: filters edges, records results,
/// and enqueues unvisited nodes for the next hop.
/// Uses parent-pointer insertion instead of path cloning.
/// Only emits results for newly-discovered nodes (no duplicates).
#[inline]
#[allow(clippy::too_many_arguments)] // Reason: BFS helper passes parent_map alongside traversal state; private fn
fn process_bfs_neighbors(
    edges: &[&super::GraphEdge],
    state: &BfsState,
    config: &TraversalConfig,
    source_id: u64,
    rel_filter: &FxHashSet<&str>,
    direction: BfsDirection,
    results: &mut Vec<TraversalResult>,
    visited: &mut FxHashSet<u64>,
    queue: &mut VecDeque<BfsState>,
    parent_map: &mut FxHashMap<u64, (u64, u64)>,
) {
    for edge in edges {
        if results.len() >= config.limit {
            break;
        }
        if !rel_filter.is_empty() && !rel_filter.contains(edge.label()) {
            continue;
        }
        let next_node = match direction {
            BfsDirection::Forward => edge.target(),
            BfsDirection::Reverse => edge.source(),
        };
        process_bfs_candidate(
            next_node,
            edge.id(),
            state.node_id,
            state.depth,
            config,
            source_id,
            results,
            visited,
            queue,
            parent_map,
        );
    }
}

/// Processes a single BFS candidate: checks depth, visited status, records
/// parent pointer, emits result if within depth range, and enqueues for
/// further expansion.
#[inline]
#[allow(clippy::too_many_arguments)]
fn process_bfs_candidate(
    target: u64,
    edge_id: u64,
    parent_node: u64,
    current_depth: u32,
    config: &TraversalConfig,
    source_id: u64,
    results: &mut Vec<TraversalResult>,
    visited: &mut FxHashSet<u64>,
    queue: &mut VecDeque<BfsState>,
    parent_map: &mut FxHashMap<u64, (u64, u64)>,
) {
    let new_depth = current_depth + 1;
    if new_depth > config.max_depth {
        return;
    }
    let is_new = visited.insert(target);
    if is_new {
        parent_map.insert(target, (parent_node, edge_id));

        if new_depth >= config.min_depth {
            let path = reconstruct_path(target, source_id, parent_map);
            results.push(TraversalResult::new(target, path, new_depth));
        }
        if new_depth < config.max_depth {
            queue.push_back(BfsState {
                node_id: target,
                depth: new_depth,
            });
        }
    }
}

/// Performs BFS traversal from a source node.
///
/// Finds all paths from `source_id` within the configured depth range.
/// Uses iterative BFS with `VecDeque` for better cache locality.
///
/// # Arguments
///
/// * `edge_store` - The edge storage to traverse.
/// * `source_id` - Starting node ID.
/// * `config` - Traversal configuration.
///
/// # Returns
///
/// Vector of traversal results, limited by `config.limit`.
#[must_use]
pub fn bfs_traverse(
    edge_store: &EdgeStore,
    source_id: u64,
    config: &TraversalConfig,
) -> Vec<TraversalResult> {
    bfs_traverse_directed(edge_store, source_id, config, BfsDirection::Forward)
}

/// Performs BFS traversal in the reverse direction (following incoming edges).
#[must_use]
pub fn bfs_traverse_reverse(
    edge_store: &EdgeStore,
    source_id: u64,
    config: &TraversalConfig,
) -> Vec<TraversalResult> {
    bfs_traverse_directed(edge_store, source_id, config, BfsDirection::Reverse)
}

// Tests moved to traversal_tests.rs per project rules