reddb-io-server 1.24.2

//! Graph inline-TVF & analytics materialization.
//!
//! Extracted verbatim from `impl_core.rs` (impl_core slice 4/10, issue #1625).
//! Houses the graph table-valued-function family that PRD #1619 keeps out of
//! the central dispatch file:
//!
//! - **Inline-TVF free fns** — `is_graph_tvf_name`, `analytics_view_algorithm`,
//!   argument parsing/validation, node/edge inlining, and the small
//!   `Value`-to-node-id / weight coercions.
//! - **Analytics dispatch methods** — `execute_inline_graph_function`,
//!   `materialize_whole_graph_abstract`, `try_resolve_analytics_view`,
//!   `dispatch_graph_algorithm`, the per-algorithm TVF executors, and
//!   `materialize_graph_with_rls` (which routes every entity through the RLS
//!   gate — see `super::rls_injection`).
//!
//! Names, signatures and visibility are preserved so the central dispatch and
//! sibling-file callers need no edits; a few formerly-private free fns are
//! bumped to `pub(crate)` and re-exported from `impl_core` for the call sites
//! that still live there.
use super::authz::policy_columns::parse_positive_iterations;
use super::execution_context::{capture_current_snapshot, current_auth_identity};
use super::rls_injection::{edge_passes_rls, node_passes_rls};
use super::*;
use crate::storage::unified::segment::GraphEntityKind;

/// The graph-analytics table-valued functions recognized in FROM position.
/// Both the graph-collection form and the inline `nodes => / edges =>` form
/// (issue #799) accept these names.
pub(crate) fn is_graph_tvf_name(name: &str) -> bool {
    name.eq_ignore_ascii_case("components")
        || name.eq_ignore_ascii_case("louvain")
        || name.eq_ignore_ascii_case("degree_centrality")
        || name.eq_ignore_ascii_case("shortest_path")
        || name.eq_ignore_ascii_case("betweenness")
        || name.eq_ignore_ascii_case("eigenvector")
        || name.eq_ignore_ascii_case("pagerank")
}

/// Map a declared `WITH ANALYTICS` view to the concrete graph algorithm name
/// and named-argument list that [`RedDBRuntime::dispatch_graph_algorithm`]
/// consumes (issue #800). The `using` option selects the algorithm inside the
/// output family; unsupported algorithms and the options that do not apply to
/// the chosen algorithm are rejected so a view never silently ignores a
/// declared parameter.
fn analytics_view_algorithm(
    graph: &str,
    view: &crate::catalog::AnalyticsViewDescriptor,
) -> RedDBResult<(String, Vec<(String, f64)>)> {
    use crate::catalog::AnalyticsOutput;

    let mut named_args: Vec<(String, f64)> = Vec::new();
    let algorithm = match view.output {
        AnalyticsOutput::Communities => {
            let algo = view.algorithm.as_deref().unwrap_or("louvain");
            if !algo.eq_ignore_ascii_case("louvain") {
                return Err(RedDBError::Query(format!(
                    "analytics output 'communities' on graph '{graph}' has unsupported algorithm '{algo}' (expected louvain)"
                )));
            }
            if let Some(resolution) = view.resolution {
                named_args.push(("resolution".to_string(), resolution));
            }
            "louvain".to_string()
        }
        AnalyticsOutput::Components => {
            if let Some(algo) = view.algorithm.as_deref() {
                if !algo.eq_ignore_ascii_case("components")
                    && !algo.eq_ignore_ascii_case("connected_components")
                {
                    return Err(RedDBError::Query(format!(
                        "analytics output 'components' on graph '{graph}' has unsupported algorithm '{algo}' (expected connected_components)"
                    )));
                }
            }
            "components".to_string()
        }
        AnalyticsOutput::Centrality => {
            let algo = view
                .algorithm
                .as_deref()
                .unwrap_or("pagerank")
                .to_ascii_lowercase();
            match algo.as_str() {
                "pagerank" => {
                    if let Some(max_iterations) = view.max_iterations {
                        named_args.push(("max_iterations".to_string(), max_iterations as f64));
                    }
                }
                "eigenvector" => {
                    if let Some(max_iterations) = view.max_iterations {
                        named_args.push(("max_iterations".to_string(), max_iterations as f64));
                    }
                    if let Some(tolerance) = view.tolerance {
                        named_args.push(("tolerance".to_string(), tolerance));
                    }
                }
                "betweenness" => {}
                other => {
                    return Err(RedDBError::Query(format!(
                        "analytics output 'centrality' on graph '{graph}' has unsupported algorithm '{other}' (expected pagerank, betweenness, or eigenvector)"
                    )));
                }
            }
            algo
        }
    };
    Ok((algorithm, named_args))
}

/// Reject any named arguments for a TVF that accepts none.
fn reject_named_args(name: &str, named_args: &[(String, f64)]) -> RedDBResult<()> {
    if let Some((key, _)) = named_args.first() {
        return Err(RedDBError::Query(format!(
            "table function '{name}' has no named argument '{key}'"
        )));
    }
    Ok(())
}

/// Resolve louvain's optional `resolution` named arg (γ, default 1.0). Any
/// other named key, or a non-finite / non-positive resolution, is rejected.
fn louvain_resolution(named_args: &[(String, f64)]) -> RedDBResult<f64> {
    let mut resolution = 1.0_f64;
    for (key, value) in named_args {
        if key.eq_ignore_ascii_case("resolution") {
            if !value.is_finite() || *value <= 0.0 {
                return Err(RedDBError::Query(format!(
                    "table function 'louvain' resolution must be > 0, got {value}"
                )));
            }
            resolution = *value;
        } else {
            return Err(RedDBError::Query(format!(
                "table function 'louvain' has no named argument '{key}' (expected 'resolution')"
            )));
        }
    }
    Ok(resolution)
}

/// Undirected degree centrality over abstract inputs: each edge contributes
/// 1 to both of its endpoints. Returns `(node_id, degree)` deterministically
/// in ascending node-id order, so identical input always yields identical
/// rows.
pub(crate) fn abstract_degree_centrality(
    nodes: &[String],
    edges: &[(
        String,
        String,
        crate::storage::engine::graph_algorithms::Weight,
    )],
) -> Vec<(String, usize)> {
    let mut degree: std::collections::BTreeMap<String, usize> = std::collections::BTreeMap::new();
    for n in nodes {
        degree.entry(n.clone()).or_insert(0);
    }
    for (a, b, _w) in edges {
        *degree.entry(a.clone()).or_insert(0) += 1;
        *degree.entry(b.clone()).or_insert(0) += 1;
    }
    degree.into_iter().collect()
}

/// Ordered column names for a materialized subquery result: the projection
/// columns when present, else the first record's field order.
fn ordered_result_columns(result: &crate::storage::query::unified::UnifiedResult) -> Vec<String> {
    if !result.columns.is_empty() {
        return result.columns.clone();
    }
    result
        .records
        .first()
        .map(|record| {
            record
                .column_names()
                .iter()
                .map(|column| column.to_string())
                .collect()
        })
        .unwrap_or_default()
}

/// Canonical node-id string for a cell value, so the node universe (from the
/// `nodes` subquery) and the edge endpoints (from the `edges` subquery)
/// compare equal regardless of integer-vs-text typing. `Null` is not a node.
fn value_to_node_id(value: &reddb_types::Value) -> Option<String> {
    use reddb_types::Value;
    match value {
        Value::Null => None,
        Value::Text(s) => Some(s.to_string()),
        Value::Integer(n) => Some(n.to_string()),
        Value::UnsignedInteger(n) => Some(n.to_string()),
        Value::NodeRef(s) => Some(s.clone()),
        other => Some(other.to_string()),
    }
}

/// Numeric edge weight from a cell value (the optional third `edges` column).
fn value_to_weight(value: &reddb_types::Value) -> Option<f32> {
    use reddb_types::Value;
    match value {
        Value::Float(f) => Some(*f as f32),
        Value::Integer(n) => Some(*n as f32),
        Value::UnsignedInteger(n) => Some(*n as f32),
        _ => None,
    }
}

/// Build the node universe from a materialized `nodes` subquery result: the
/// first projected column of each row is the node id (issue #799). Zero rows
/// is a valid empty node set; a row set with no columns is a shape error.
fn inline_node_ids(
    name: &str,
    result: &crate::storage::query::unified::UnifiedResult,
) -> RedDBResult<Vec<String>> {
    if result.records.is_empty() {
        return Ok(Vec::new());
    }
    let columns = ordered_result_columns(result);
    let Some(first_col) = columns.first() else {
        return Err(RedDBError::Query(format!(
            "table function '{name}' inline form: `nodes` subquery must project at least one column (the node id)"
        )));
    };
    let mut ids = Vec::with_capacity(result.records.len());
    for record in &result.records {
        if let Some(id) = record.get(first_col).and_then(value_to_node_id) {
            ids.push(id);
        }
    }
    Ok(ids)
}

/// Build the edge list from a materialized `edges` subquery result: the first
/// two projected columns are `(source, target)` and an optional third column
/// is the numeric weight (defaulting to 1.0). Fewer than two columns is a
/// shape error (issue #799).
fn inline_edges(
    name: &str,
    result: &crate::storage::query::unified::UnifiedResult,
) -> RedDBResult<
    Vec<(
        String,
        String,
        crate::storage::engine::graph_algorithms::Weight,
    )>,
> {
    if result.records.is_empty() {
        return Ok(Vec::new());
    }
    let columns = ordered_result_columns(result);
    if columns.len() < 2 {
        return Err(RedDBError::Query(format!(
            "table function '{name}' inline form: `edges` subquery must project at least two columns (source, target), got {}",
            columns.len()
        )));
    }
    let src_col = &columns[0];
    let dst_col = &columns[1];
    let weight_col = columns.get(2);
    let mut edges = Vec::with_capacity(result.records.len());
    for record in &result.records {
        let (Some(src), Some(dst)) = (
            record.get(src_col).and_then(value_to_node_id),
            record.get(dst_col).and_then(value_to_node_id),
        ) else {
            // A null/absent endpoint is not a valid edge; skip it.
            continue;
        };
        let weight = match weight_col {
            Some(col) => match record.get(col) {
                None | Some(reddb_types::Value::Null) => 1.0,
                Some(value) => value_to_weight(value).ok_or_else(|| {
                    RedDBError::Query(format!(
                        "table function '{name}' inline form: `edges` weight column must be numeric"
                    ))
                })?,
            },
            None => 1.0,
        };
        edges.push((src, dst, weight));
    }
    Ok(edges)
}

impl RedDBRuntime {
    /// Dispatch an inline-graph table-valued function call in FROM position
    /// (e.g. `SELECT * FROM components(nodes => (…), edges => (…))`, issue
    /// #799).
    ///
    /// Materializes the two subqueries through the normal read path (so RLS,
    /// column authz, and MVCC visibility all apply), constructs the abstract
    /// graph — the first column of `nodes` is the node id; the first two-or-
    /// three columns of `edges` are `(source, target [, weight])` — then runs
    /// the same algorithm path used by the graph-collection form. Read-only.
    pub(crate) fn execute_inline_graph_function(
        &self,
        name: &str,
        nodes_query: &QueryExpr,
        edges_query: &QueryExpr,
        named_args: &[(String, f64)],
    ) -> RedDBResult<crate::storage::query::unified::UnifiedResult> {
        if !is_graph_tvf_name(name) {
            return Err(RedDBError::Query(format!("unknown table function: {name}")));
        }

        let node_result = self.execute_query_expr(nodes_query.clone())?.result;
        let nodes = inline_node_ids(name, &node_result)?;

        let edge_result = self.execute_query_expr(edges_query.clone())?.result;
        let edges = inline_edges(name, &edge_result)?;

        self.dispatch_graph_algorithm(name, nodes, edges, named_args)
    }

    /// Materialize the whole active graph read-only into the abstract
    /// `(nodes, edges)` inputs the pure graph algorithms consume.
    pub(crate) fn materialize_whole_graph_abstract(
        &self,
    ) -> RedDBResult<(
        Vec<String>,
        Vec<(
            String,
            String,
            crate::storage::engine::graph_algorithms::Weight,
        )>,
    )> {
        use crate::storage::engine::graph_algorithms;

        let graph = super::graph_dsl::materialize_graph_with_projection(self, None)?;
        let nodes: Vec<String> = graph.iter_nodes().map(|n| n.id.clone()).collect();
        let edges: Vec<(String, String, graph_algorithms::Weight)> = graph
            .iter_all_edges()
            .into_iter()
            .map(|e| (e.source_id, e.target_id, e.weight))
            .collect();
        Ok((nodes, edges))
    }

    /// Resolve a `<graph>.<output>` analytics virtual view (issue #800).
    ///
    /// Returns `Ok(None)` when `table` is not an analytics view — either the
    /// name is not dotted, a real collection of that exact name exists (a real
    /// collection always wins; no shadowing), the suffix is not a recognised
    /// analytics output, or the parent is not a graph. Returns `Ok(Some(_))`
    /// with the freshly computed result when it does resolve, and an error when
    /// the parent graph exists but the output is not enabled, a declared
    /// algorithm is unsupported, or the parent collection's policy denies the
    /// read.
    ///
    /// The view is recomputed on every call (no result-cache write) so it
    /// always reflects the current graph data, satisfying the on-demand
    /// recompute contract for this slice.
    pub(crate) fn try_resolve_analytics_view(
        &self,
        table: &TableQuery,
        frame: &dyn super::statement_frame::ReadFrame,
    ) -> RedDBResult<Option<crate::storage::query::unified::UnifiedResult>> {
        let full = table.table.as_str();
        let Some(dot) = full.rfind('.') else {
            return Ok(None);
        };
        // A real collection literally named `g.communities` always wins.
        if self.inner.db.store().get_collection(full).is_some() {
            return Ok(None);
        }
        let graph_name = &full[..dot];
        let output_name = &full[dot + 1..];
        let Some(output) = crate::catalog::AnalyticsOutput::from_str(output_name) else {
            return Ok(None);
        };

        let contracts = self.inner.db.collection_contracts();
        let Some(contract) = contracts.iter().find(|c| c.name == graph_name) else {
            return Ok(None);
        };
        if contract.declared_model != crate::catalog::CollectionModel::Graph {
            return Ok(None);
        }
        let Some(view) = contract
            .analytics_config
            .iter()
            .find(|view| view.output == output)
        else {
            // The parent graph exists but this output was not declared — a
            // clear error beats the misleading "collection not found".
            return Err(RedDBError::Query(format!(
                "analytics output '{output_name}' is not enabled on graph '{graph_name}'; declare it with WITH ANALYTICS (...)"
            )));
        };

        // Policy inheritance (AC5): route through the parent graph collection's
        // read authorization. A policy or RLS rule that denies the parent
        // denies its analytics views transitively.
        let parent_query = TableQuery::new(graph_name);
        if self
            .authorize_relational_table_select(parent_query, frame)?
            .is_none()
        {
            return Err(RedDBError::Query(format!(
                "permission denied: policy on graph '{graph_name}' denies analytics view '{output_name}'"
            )));
        }

        let (algorithm, named_args) = analytics_view_algorithm(graph_name, view)?;
        let (nodes, edges) = self.materialize_whole_graph_abstract()?;
        let result = self.dispatch_graph_algorithm(&algorithm, nodes, edges, &named_args)?;
        Ok(Some(result))
    }

    /// Shared algorithm dispatch over abstract `(nodes, edges)` inputs.
    ///
    /// Both the graph-collection form and the inline-graph form route here so
    /// named-argument validation and the projected row shape stay identical
    /// across the two signatures (issue #799). Projects each algorithm's
    /// native output shape.
    pub(crate) fn dispatch_graph_algorithm(
        &self,
        name: &str,
        nodes: Vec<String>,
        edges: Vec<(
            String,
            String,
            crate::storage::engine::graph_algorithms::Weight,
        )>,
        named_args: &[(String, f64)],
    ) -> RedDBResult<crate::storage::query::unified::UnifiedResult> {
        use crate::storage::engine::graph_algorithms;
        use crate::storage::query::unified::UnifiedResult;
        use reddb_types::Value;

        if name.eq_ignore_ascii_case("components") {
            reject_named_args(name, named_args)?;
            let assignment = graph_algorithms::connected_components(&nodes, &edges);
            let mut result =
                UnifiedResult::with_columns(vec!["node_id".into(), "island_id".into()]);
            for (node_id, island_id) in assignment {
                let mut record = UnifiedRecord::new();
                record.set("node_id", Value::text(node_id));
                record.set("island_id", Value::Integer(island_id as i64));
                result.push(record);
            }
            return Ok(result);
        }

        if name.eq_ignore_ascii_case("louvain") {
            // The only supported named argument is `resolution` (γ). It
            // defaults to 1.0 (classic modularity) and must be a finite,
            // strictly positive number — a non-positive (or NaN/inf)
            // resolution has no sensible meaning.
            let resolution = louvain_resolution(named_args)?;
            let assignment = graph_algorithms::louvain(&nodes, &edges, resolution);
            let mut result =
                UnifiedResult::with_columns(vec!["node_id".into(), "community_id".into()]);
            for (node_id, community_id) in assignment {
                let mut record = UnifiedRecord::new();
                record.set("node_id", Value::text(node_id));
                record.set("community_id", Value::Integer(community_id as i64));
                result.push(record);
            }
            return Ok(result);
        }

        if name.eq_ignore_ascii_case("degree_centrality") {
            reject_named_args(name, named_args)?;
            let assignment = abstract_degree_centrality(&nodes, &edges);
            let mut result = UnifiedResult::with_columns(vec!["node_id".into(), "degree".into()]);
            for (node_id, degree) in assignment {
                let mut record = UnifiedRecord::new();
                record.set("node_id", Value::text(node_id));
                record.set("degree", Value::Integer(degree as i64));
                result.push(record);
            }
            return Ok(result);
        }

        if name.eq_ignore_ascii_case("shortest_path") {
            // Scalar named arguments: `src` and `dst` are required node ids,
            // `max_hops` is an optional non-negative edge-count cap. Node ids
            // in the graph store are integer entity ids rendered as strings, so
            // each id arg must be a non-negative whole number; reject anything
            // else (fractional, negative, NaN/inf) with a clear message.
            let mut src: Option<String> = None;
            let mut dst: Option<String> = None;
            let mut max_hops: Option<usize> = None;
            let as_node_id = |key: &str, value: f64| -> RedDBResult<String> {
                if !value.is_finite() || value < 0.0 || value.fract() != 0.0 {
                    return Err(RedDBError::Query(format!(
                        "table function 'shortest_path' argument '{key}' must be a non-negative integer node id, got {value}"
                    )));
                }
                Ok((value as i64).to_string())
            };
            for (key, value) in named_args {
                if key.eq_ignore_ascii_case("src") {
                    src = Some(as_node_id("src", *value)?);
                } else if key.eq_ignore_ascii_case("dst") {
                    dst = Some(as_node_id("dst", *value)?);
                } else if key.eq_ignore_ascii_case("max_hops") {
                    if !value.is_finite() || *value < 0.0 || value.fract() != 0.0 {
                        return Err(RedDBError::Query(format!(
                            "table function 'shortest_path' max_hops must be a non-negative integer, got {value}"
                        )));
                    }
                    max_hops = Some(*value as usize);
                } else {
                    return Err(RedDBError::Query(format!(
                        "table function 'shortest_path' has no named argument '{key}' (expected 'src', 'dst', 'max_hops')"
                    )));
                }
            }
            let src = src.ok_or_else(|| {
                RedDBError::Query(
                    "table function 'shortest_path' requires named argument 'src'".to_string(),
                )
            })?;
            let dst = dst.ok_or_else(|| {
                RedDBError::Query(
                    "table function 'shortest_path' requires named argument 'dst'".to_string(),
                )
            })?;

            // Columns are always present; an unreachable pair (within the
            // optional `max_hops` budget) simply yields zero rows — never an
            // error. `hop` is the 0-based index from the source;
            // `cumulative_weight` is the running path weight (0 at the source,
            // the total at the destination). Edges are treated as undirected,
            // consistent with `components` / `louvain`.
            let mut result = UnifiedResult::with_columns(vec![
                "hop".into(),
                "node_id".into(),
                "cumulative_weight".into(),
            ]);
            if let Some(path) =
                graph_algorithms::shortest_path(&nodes, &edges, &src, &dst, max_hops)
            {
                for (hop, (node_id, cumulative_weight)) in path.into_iter().enumerate() {
                    let mut record = UnifiedRecord::new();
                    record.set("hop", Value::Integer(hop as i64));
                    record.set("node_id", Value::text(node_id));
                    record.set("cumulative_weight", Value::Float(cumulative_weight));
                    result.push(record);
                }
            }
            return Ok(result);
        }
        // ── Centrality family (issue #797): each returns rows `(node_id,
        // score)` over the abstract `(nodes, edges)` graph. Like the other
        // graph TVFs the graph is treated as undirected and scores are
        // deterministic; the inline-graph form shares this dispatch. ──
        if name.eq_ignore_ascii_case("betweenness") {
            reject_named_args(name, named_args)?;
            return Ok(Self::centrality_result(graph_algorithms::betweenness(
                &nodes, &edges,
            )));
        }
        if name.eq_ignore_ascii_case("eigenvector") {
            // Optional `max_iterations` (positive integer, default 100) and
            // `tolerance` (finite, strictly positive, default 1e-6).
            let mut max_iterations = 100_usize;
            let mut tolerance = 1e-6_f64;
            for (key, value) in named_args {
                if key.eq_ignore_ascii_case("max_iterations") {
                    max_iterations = parse_positive_iterations("eigenvector", value)?;
                } else if key.eq_ignore_ascii_case("tolerance") {
                    if !value.is_finite() || *value <= 0.0 {
                        return Err(RedDBError::Query(format!(
                            "table function 'eigenvector' tolerance must be > 0, got {value}"
                        )));
                    }
                    tolerance = *value;
                } else {
                    return Err(RedDBError::Query(format!(
                        "table function 'eigenvector' has no named argument '{key}' (expected 'max_iterations' or 'tolerance')"
                    )));
                }
            }
            return Ok(Self::centrality_result(graph_algorithms::eigenvector(
                &nodes,
                &edges,
                max_iterations,
                tolerance,
            )));
        }
        if name.eq_ignore_ascii_case("pagerank") {
            // Optional `damping` (in (0, 1), default 0.85) and `max_iterations`
            // (positive integer, default 100).
            let mut damping = 0.85_f64;
            let mut max_iterations = 100_usize;
            for (key, value) in named_args {
                if key.eq_ignore_ascii_case("damping") {
                    if !value.is_finite() || *value <= 0.0 || *value >= 1.0 {
                        return Err(RedDBError::Query(format!(
                            "table function 'pagerank' damping must be in (0, 1), got {value}"
                        )));
                    }
                    damping = *value;
                } else if key.eq_ignore_ascii_case("max_iterations") {
                    max_iterations = parse_positive_iterations("pagerank", value)?;
                } else {
                    return Err(RedDBError::Query(format!(
                        "table function 'pagerank' has no named argument '{key}' (expected 'damping' or 'max_iterations')"
                    )));
                }
            }
            return Ok(Self::centrality_result(graph_algorithms::pagerank(
                &nodes,
                &edges,
                damping,
                max_iterations,
            )));
        }
        Err(RedDBError::Query(format!("unknown table function: {name}")))
    }

    /// `components(<graph_collection>)` — returns rows `(node_id, island_id)`.
    ///
    /// Materializes the active graph (nodes + weighted edges) read-only and
    /// runs the pure `graph_algorithms::connected_components`. Edges are
    /// treated as undirected; island ids are deterministic (ascending order of
    /// each component's smallest node).
    fn execute_components_tvf(
        &self,
        _collection: &str,
    ) -> RedDBResult<crate::storage::query::unified::UnifiedResult> {
        use crate::storage::engine::graph_algorithms;
        use crate::storage::query::unified::UnifiedResult;
        use reddb_types::Value;

        // Read-only materialization of the full active graph. The named
        // collection identifies the active graph scope; passing `None` for the
        // projection uses the full graph store (the same result
        // `active_graph_projection` yields when no projection is registered).
        // Materialization never mutates any store.
        let graph = super::graph_dsl::materialize_graph_with_projection(self, None)?;

        // Materialize abstract inputs for the pure algorithm.
        let nodes: Vec<String> = graph.iter_nodes().map(|n| n.id.clone()).collect();
        let edges: Vec<(String, String, graph_algorithms::Weight)> = graph
            .iter_all_edges()
            .into_iter()
            .map(|e| (e.source_id, e.target_id, e.weight))
            .collect();

        let assignment = graph_algorithms::connected_components(&nodes, &edges);

        // Project into a UnifiedResult with columns ["node_id", "island_id"].
        let mut result = UnifiedResult::with_columns(vec!["node_id".into(), "island_id".into()]);
        for (node_id, island_id) in assignment {
            let mut record = UnifiedRecord::new();
            record.set("node_id", Value::text(node_id));
            record.set("island_id", Value::Integer(island_id as i64));
            result.push(record);
        }
        Ok(result)
    }

    /// `louvain(<graph> [, resolution => <f64>])` — returns rows
    /// `(node_id, community_id)` (issue #796).
    ///
    /// Materializes the active graph (nodes + weighted edges) read-only and
    /// runs the pure, deterministic `graph_algorithms::louvain`. Edges are
    /// treated as undirected; community ids are assigned in ascending order of
    /// each community's smallest node, so identical input + resolution always
    /// yields identical rows. Like `components`, the v0 form runs over the
    /// whole graph store regardless of the collection argument value.
    fn execute_louvain_tvf(
        &self,
        _collection: &str,
        resolution: f64,
    ) -> RedDBResult<crate::storage::query::unified::UnifiedResult> {
        use crate::storage::engine::graph_algorithms;
        use crate::storage::query::unified::UnifiedResult;
        use reddb_types::Value;

        let graph = super::graph_dsl::materialize_graph_with_projection(self, None)?;

        let nodes: Vec<String> = graph.iter_nodes().map(|n| n.id.clone()).collect();
        let edges: Vec<(String, String, graph_algorithms::Weight)> = graph
            .iter_all_edges()
            .into_iter()
            .map(|e| (e.source_id, e.target_id, e.weight))
            .collect();

        let assignment = graph_algorithms::louvain(&nodes, &edges, resolution);

        // Project into a UnifiedResult with columns ["node_id", "community_id"].
        let mut result = UnifiedResult::with_columns(vec!["node_id".into(), "community_id".into()]);
        for (node_id, community_id) in assignment {
            let mut record = UnifiedRecord::new();
            record.set("node_id", Value::text(node_id));
            record.set("community_id", Value::Integer(community_id as i64));
            result.push(record);
        }
        Ok(result)
    }

    /// Project `(node_id, score)` centrality rows into a `UnifiedResult` with
    /// columns `["node_id", "score"]`; scores are `Value::Float`.
    fn centrality_result(
        rows: Vec<(String, f64)>,
    ) -> crate::storage::query::unified::UnifiedResult {
        use crate::storage::query::unified::UnifiedResult;
        use reddb_types::Value;
        let mut result = UnifiedResult::with_columns(vec!["node_id".into(), "score".into()]);
        for (node_id, score) in rows {
            let mut record = UnifiedRecord::new();
            record.set("node_id", Value::text(node_id));
            record.set("score", Value::Float(score));
            result.push(record);
        }
        result
    }

    /// Materialise the entire graph store while applying MVCC visibility
    /// AND per-collection RLS to each candidate node and edge. Mirrors
    /// `materialize_graph` but routes every entity through the same
    /// gate the SELECT path uses, with the correct `PolicyTargetKind`
    /// per entity kind (`Nodes` for graph nodes, `Edges` for graph
    /// edges). Returns the filtered `GraphStore` plus the
    /// `node_id → properties` map the executor needs for `RETURN n.*`
    /// projections.
    pub(crate) fn materialize_graph_with_rls(
        &self,
    ) -> RedDBResult<(
        crate::storage::engine::GraphStore,
        std::collections::HashMap<String, std::collections::HashMap<String, reddb_types::Value>>,
        crate::storage::query::unified::EdgeProperties,
    )> {
        self.materialize_graph_filtered(None, true)
    }

    pub(super) fn materialize_graph_filtered(
        &self,
        projection: Option<&RuntimeGraphProjection>,
        collect_properties: bool,
    ) -> RedDBResult<(
        crate::storage::engine::GraphStore,
        std::collections::HashMap<String, std::collections::HashMap<String, reddb_types::Value>>,
        crate::storage::query::unified::EdgeProperties,
    )> {
        use crate::storage::engine::GraphStore;
        use crate::storage::unified::entity::{EntityData, EntityKind};
        use reddb_rql::ast::{PolicyAction, PolicyTargetKind};
        use std::collections::{HashMap, HashSet};

        let store = self.inner.db.store();
        let snap_ctx = capture_current_snapshot();
        let role = current_auth_identity().map(|(_, r)| r.as_str().to_string());

        let graph = GraphStore::new();
        let mut node_properties: HashMap<String, HashMap<String, reddb_types::Value>> =
            HashMap::new();
        let mut edge_properties: crate::storage::query::unified::EdgeProperties = HashMap::new();
        let mut allowed_nodes: HashSet<String> = HashSet::new();
        let mut endpoint_aliases: HashMap<String, Option<String>> = HashMap::new();

        // Per-collection cached compiled filters — Nodes-kind for
        // first pass, Edges-kind for the second. None entries mean
        // "RLS enabled, zero matching policy → deny all of this kind".
        let mut node_rls: HashMap<String, Option<reddb_rql::ast::Filter>> = HashMap::new();
        let mut edge_rls: HashMap<String, Option<reddb_rql::ast::Filter>> = HashMap::new();

        let collections = store.list_collections();

        let node_labels =
            projection.and_then(|p| normalize_token_filter_list(p.node_labels.clone()));
        let node_types = projection.and_then(|p| normalize_token_filter_list(p.node_types.clone()));
        let edge_labels = projection.and_then(|p| normalize_edge_filters(p.edge_labels.clone()));

        // First pass — gather nodes.
        for collection in &collections {
            let Some(manager) = store.get_collection(collection) else {
                continue;
            };
            visit_graph_materialization_entities(
                &manager,
                snap_ctx.as_ref(),
                GraphEntityKind::Node,
                |entity| {
                    let EntityKind::GraphNode(ref node) = entity.kind else {
                        return Ok(());
                    };
                    if !matches_graph_node_projection(
                        &node.label,
                        &node.node_type,
                        node_labels.as_ref(),
                        node_types.as_ref(),
                    ) {
                        return Ok(());
                    }
                    if !node_passes_rls(self, collection, role.as_deref(), &mut node_rls, &entity) {
                        return Ok(());
                    }
                    let id_str = entity.logical_id().raw().to_string();
                    graph
                        .add_node_with_label(
                            &id_str,
                            &node.label,
                            &super::graph_node_label(&node.node_type),
                        )
                        .map_err(|err| RedDBError::Query(err.to_string()))?;
                    allowed_nodes.insert(id_str.clone());
                    if collect_properties {
                        if let EntityData::Node(node_data) = &entity.data {
                            node_properties.insert(id_str, node_data.properties.clone());
                        }
                    }
                    Ok(())
                },
            )?;
        }

        // Second pass — gather edges. An edge appears only when both
        // endpoint nodes survived the RLS pass AND the edge itself
        // passes its own RLS gate.
        for collection in &collections {
            let Some(manager) = store.get_collection(collection) else {
                continue;
            };
            visit_graph_materialization_entities(
                &manager,
                snap_ctx.as_ref(),
                GraphEntityKind::Edge,
                |entity| {
                    let EntityKind::GraphEdge(ref edge) = entity.kind else {
                        return Ok(());
                    };
                    if !matches_graph_edge_projection(&edge.label, edge_labels.as_ref()) {
                        return Ok(());
                    }
                    if !edge_passes_rls(self, collection, role.as_deref(), &mut edge_rls, &entity) {
                        return Ok(());
                    }
                    let Some(from_node) = resolve_materialized_graph_endpoint(
                        &edge.from_node,
                        &allowed_nodes,
                        &mut endpoint_aliases,
                        &store,
                        &collections,
                    )?
                    else {
                        return Ok(());
                    };
                    let Some(to_node) = resolve_materialized_graph_endpoint(
                        &edge.to_node,
                        &allowed_nodes,
                        &mut endpoint_aliases,
                        &store,
                        &collections,
                    )?
                    else {
                        return Ok(());
                    };
                    let weight = match &entity.data {
                        EntityData::Edge(e) => e.weight,
                        _ => edge.weight as f32 / 1000.0,
                    };
                    let edge_label = super::graph_edge_label(&edge.label);
                    graph
                        .add_edge_with_label(from_node, to_node, &edge_label, weight)
                        .map_err(|err| RedDBError::Query(err.to_string()))?;
                    if collect_properties {
                        if let EntityData::Edge(edge_data) = &entity.data {
                            edge_properties.insert(
                                (from_node.to_string(), edge_label, to_node.to_string()),
                                edge_data.properties.clone(),
                            );
                        }
                    }
                    Ok(())
                },
            )?;
        }

        // Suppress unused-PolicyAction/PolicyTargetKind warnings — both
        // are used inside the helper closures via the per-kind helpers
        // declared at the bottom of this file.
        let _ = (PolicyAction::Select, PolicyTargetKind::Nodes);

        Ok((graph, node_properties, edge_properties))
    }
}

/// Legacy edges can reference any retained physical version. Identity lookup
/// does not grant visibility: the caller must gate endpoints by snapshot/RLS
/// before exposing or traversing them. Cache misses too, so dangling edges do
/// not repeat probes.
pub(super) fn resolve_materialized_graph_endpoint<'a>(
    endpoint: &str,
    allowed_nodes: &'a std::collections::HashSet<String>,
    aliases: &mut std::collections::HashMap<String, Option<String>>,
    store: &crate::storage::unified::UnifiedStore,
    collections: &[String],
) -> RedDBResult<Option<&'a str>> {
    if let Some(node) = allowed_nodes.get(endpoint) {
        return Ok(Some(node.as_str()));
    }
    if !aliases.contains_key(endpoint) {
        let mut logical = None;
        if let Ok(physical) = endpoint.parse::<u64>() {
            for collection in collections {
                super::function_budget::charge(1)?;
                let Some(manager) = store.get_collection(collection) else {
                    continue;
                };
                let identity = manager.graph_node_logical_id(EntityId::new(physical), || {
                    super::function_budget::charge(1).is_ok()
                });
                // A stopped probe must propagate the sticky budget error.
                super::function_budget::charge(0)?;
                if let Some(identity) = identity {
                    logical = Some(identity.raw().to_string());
                    break;
                }
            }
        }
        aliases.insert(endpoint.to_string(), logical);
    }
    Ok(aliases
        .get(endpoint)
        .and_then(Option::as_ref)
        .and_then(|identity| allowed_nodes.get(identity))
        .map(String::as_str))
}

/// RLS may re-enter storage, so only collect IDs while holding segment locks.
/// Filter by entity kind before cloning payloads or collecting IDs. Collections
/// can contain mixed kinds. Ordinary queries retain the parallel scan path.
pub(super) fn visit_graph_materialization_entities(
    manager: &crate::storage::unified::manager::SegmentManager,
    snapshot: Option<&super::impl_core::SnapshotContext>,
    kind: GraphEntityKind,
    mut visit: impl FnMut(crate::storage::unified::entity::UnifiedEntity) -> RedDBResult<()>,
) -> RedDBResult<()> {
    if !super::function_budget::active() {
        for entity in manager.scan_graph_kind(snapshot, kind) {
            visit(entity)?;
        }
        return Ok(());
    }
    let mut ids = Vec::new();
    let mut segment_failure = None;
    super::function_budget::scan(
        |callback| {
            manager.scan_graph_candidates_for_each(
                snapshot,
                kind,
                || {
                    if let Err(cause) = super::function_budget::charge(1) {
                        segment_failure = Some(cause);
                        return false;
                    }
                    true
                },
                callback,
            )
        },
        |entity| {
            // The scan charges every visible candidate, including rejected kinds.
            // Only the current graph pass's payloads need to leave the segment lock.
            if kind.matches(&entity.kind) {
                ids.push(entity.id);
            }
            true
        },
    )?;
    if let Some(cause) = segment_failure {
        return Err(cause);
    }
    for batch in ids.chunks(256) {
        super::function_budget::charge(0)?;
        for entity in manager.get_many(batch).into_iter().flatten() {
            super::function_budget::charge(1)?;
            // Re-check the captured MVCC view after releasing the scan lock.
            if super::impl_core::entity_visible_with_context(snapshot, &entity) {
                visit(entity)?;
            }
        }
    }
    Ok(())
}

#[cfg(test)]
mod pruning_tests {
    use super::*;
    use crate::storage::{EntityId, SegmentManager, UnifiedEntity};

    #[test]
    fn historical_endpoint_probes_consume_call_work() {
        let runtime = RedDBRuntime::in_memory().expect("runtime");
        let store = runtime.db().store();
        let manager = store.get_or_create_collection("history");
        for id in 1..=3 {
            manager
                .insert(UnifiedEntity::vector(EntityId::new(id), "history", vec![]))
                .expect("vector");
            manager.force_seal().expect("seal");
        }
        let mut historical =
            UnifiedEntity::graph_node(EntityId::new(100), "history", "node", Default::default());
        historical.set_logical_id(EntityId::new(99));
        historical.set_xmax(5);
        manager.insert(historical).expect("historical node");
        manager.force_seal().expect("seal node");
        let allowed = std::collections::HashSet::from(["99".to_string()]);
        let collections = vec!["history".to_string()];
        let mut aliases = std::collections::HashMap::new();
        {
            let _budget = super::super::function_budget::Scope::enter(2, 60_000).expect("budget");
            let error = resolve_materialized_graph_endpoint(
                "100",
                &allowed,
                &mut aliases,
                &store,
                &collections,
            )
            .expect_err("collection and segment probes exhaust budget");
            assert!(
                error.to_string().contains("execution work_max exceeded"),
                "{error}"
            );
            assert!(
                aliases.is_empty(),
                "failed lookup cannot cache partial resolution"
            );
        }
        assert_eq!(
            resolve_materialized_graph_endpoint(
                "100",
                &allowed,
                &mut aliases,
                &store,
                &collections,
            )
            .expect("retained identity"),
            Some("99")
        );
        for invalid in ["1", "missing", "101"] {
            assert_eq!(
                resolve_materialized_graph_endpoint(
                    invalid,
                    &allowed,
                    &mut aliases,
                    &store,
                    &collections,
                )
                .expect("absent node"),
                None
            );
        }
    }

    #[test]
    fn skipped_graph_segments_still_consume_call_work() {
        let manager = SegmentManager::new("vectors");
        for id in 1..=3 {
            manager
                .bulk_insert(vec![UnifiedEntity::vector(
                    EntityId::new(id),
                    "vectors",
                    vec![],
                )])
                .expect("vector segment");
            manager.force_seal().expect("seal vectors");
        }
        let _budget = super::super::function_budget::Scope::enter(2, 60_000).expect("budget");
        let error =
            visit_graph_materialization_entities(&manager, None, GraphEntityKind::Node, |_| {
                panic!("vector payload must not be visited")
            })
            .expect_err("three segments exceed two work units");
        assert!(
            error.to_string().contains("execution work_max exceeded"),
            "{error}"
        );
    }
}