//! # Connected Components — Label each node with its component ID
//!
//! Computes connected components using union-find (disjoint-set) over the CSR.
//! O(V + E × α(V)) where α is the inverse Ackermann function (effectively O(V+E)).
//!
//! This is a global graph algorithm — it touches every node and every edge.
//! Runtime scales linearly with graph size.
//!
//! ## Design
//!
//! We use union-find with path compression and union-by-rank for near-linear
//! performance. The component ID for each node is the union-find root after all
//! unions; callers should treat it as a stable label for the current
//! computation, not as a persistent node identifier.
//!
//! See: `docs/contributor_guide/traversal-search-paths.mdx`

#[cfg(test)]
use crate::edge_store::EdgeStore;
use crate::node_store::NodeStore;
#[cfg(test)]
use crate::projection::neighbors::CsrNeighbors;
use crate::projection::neighbors::NeighborSource;
use crate::safety::{GraphError, GraphResult};
use crate::types::TableOid;
use std::collections::HashMap;

/// Result of connected components computation.
#[derive(Debug)]
pub struct ComponentResult {
    /// Component ID for each node. `component[i]` is the component root for node `i`.
    pub component: Vec<u32>,
    /// Number of distinct components.
    pub num_components: u32,
    /// Size of the largest component.
    pub largest_component_size: u32,
    /// Active node count by component ID.
    pub component_sizes: HashMap<u32, u32>,
}

impl ComponentResult {
    pub fn component_size(&self, component_id: u32) -> u32 {
        self.component_sizes
            .get(&component_id)
            .copied()
            .unwrap_or(0)
    }
}

/// Union-Find (Disjoint-Set) data structure.
struct UnionFind {
    parent: Vec<u32>,
    rank: Vec<u8>,
}

impl UnionFind {
    fn try_new(n: usize) -> GraphResult<Self> {
        let node_count = u32::try_from(n).map_err(|_| allocation_error())?;
        let mut parent = Vec::new();
        parent
            .try_reserve_exact(n)
            .map_err(|_| allocation_error())?;
        parent.extend(0..node_count);
        let mut rank = Vec::new();
        rank.try_reserve_exact(n).map_err(|_| allocation_error())?;
        rank.resize(n, 0);
        Ok(Self { parent, rank })
    }

    /// Find the root of node `x` with path compression.
    fn find(&mut self, x: u32) -> u32 {
        if self.parent[x as usize] != x {
            self.parent[x as usize] = self.find(self.parent[x as usize]);
        }
        self.parent[x as usize]
    }

    /// Union the sets containing `x` and `y`. Uses union-by-rank.
    fn union(&mut self, x: u32, y: u32) {
        let rx = self.find(x);
        let ry = self.find(y);
        if rx == ry {
            return;
        }
        // Union by rank keeps trees shallow; equal ranks choose `rx`, which is
        // deterministic for this edge iteration order.
        match self.rank[rx as usize].cmp(&self.rank[ry as usize]) {
            std::cmp::Ordering::Less => {
                self.parent[rx as usize] = ry;
            }
            std::cmp::Ordering::Greater => {
                self.parent[ry as usize] = rx;
            }
            std::cmp::Ordering::Equal => {
                self.parent[ry as usize] = rx;
                self.rank[rx as usize] += 1;
            }
        }
    }
}

/// Compute connected components for all active nodes in the graph.
///
/// # Arguments
/// * `node_store` — to check which nodes are active
/// * `edge_store` — CSR edge data
///
/// # Returns
/// ComponentResult with per-node component labels and summary stats.
#[cfg(test)]
pub fn compute_components(node_store: &NodeStore, edge_store: &EdgeStore) -> ComponentResult {
    let neighbors = CsrNeighbors::new(edge_store);
    compute_components_with_neighbors(node_store, &neighbors)
}

/// Compute connected components over a supplied neighbor source.
#[cfg(test)]
pub(crate) fn compute_components_with_neighbors(
    node_store: &NodeStore,
    neighbors: &impl NeighborSource,
) -> ComponentResult {
    compute_components_with_neighbors_inner(
        node_store,
        neighbors,
        None,
        crate::visibility::VisibilityCoordinator::unrestricted_for_test_or_benchmark().scope(),
    )
    .expect("unbounded component accounting should not fail")
}

/// Compute components with hard work and elapsed-time accounting.
#[cfg(test)]
#[allow(dead_code, reason = "legacy test compatibility entry point")]
pub(crate) fn compute_components_with_neighbors_governed(
    node_store: &NodeStore,
    neighbors: &impl NeighborSource,
    governor: &crate::resource::ResourceGovernor,
) -> GraphResult<ComponentResult> {
    let coordinator =
        crate::visibility::VisibilityCoordinator::unrestricted_for_test_or_benchmark();
    let context = coordinator.context(governor);
    compute_components_with_neighbors_in_context(node_store, neighbors, &context)
}

/// Compute caller-visible components with hard work and elapsed-time accounting.
pub(crate) fn compute_components_with_neighbors_in_context(
    node_store: &NodeStore,
    neighbors: &impl NeighborSource,
    context: &crate::visibility::QueryExecutionContext<'_>,
) -> GraphResult<ComponentResult> {
    compute_components_with_neighbors_inner(
        node_store,
        neighbors,
        Some(context.governor),
        context.visibility,
    )
}

fn compute_components_with_neighbors_inner(
    node_store: &NodeStore,
    neighbors: &impl NeighborSource,
    governor: Option<&crate::resource::ResourceGovernor>,
    visibility: &crate::visibility::VisibilityScope,
) -> GraphResult<ComponentResult> {
    let node_count = node_store.node_count() as usize;

    if node_count == 0 {
        return Ok(ComponentResult {
            component: vec![],
            num_components: 0,
            largest_component_size: 0,
            component_sizes: HashMap::new(),
        });
    }

    let mut uf = UnionFind::try_new(node_count)?;

    // Iterate through all edges in the CSR — sequential, cache-friendly
    for node in 0..node_count as u32 {
        if !node_store.is_active(node)
            || crate::projection::tx_delta::node_deleted(node)
            || !visibility.allows_node(node)
        {
            continue;
        }

        for edge in neighbors.neighbors(node) {
            consume_analytics_work(governor)?;
            if visibility.allows_relationship(edge.type_id, edge.relationship_id)?
                && visibility.allows_node(edge.target)
                && node_store.is_active(edge.target)
                && !crate::projection::tx_delta::node_deleted(edge.target)
            {
                uf.union(node, edge.target);
            }
        }
    }

    // Finalize: compress all paths and compute stats
    let mut component = Vec::new();
    component
        .try_reserve_exact(node_count)
        .map_err(|_| allocation_error())?;
    component.resize(node_count, 0);
    let mut component_sizes = HashMap::new();
    component_sizes
        .try_reserve(node_count)
        .map_err(|_| allocation_error())?;

    for node in 0..node_count as u32 {
        consume_analytics_work(governor)?;
        if !node_store.is_active(node)
            || crate::projection::tx_delta::node_deleted(node)
            || !visibility.allows_node(node)
        {
            component[node as usize] = u32::MAX; // inactive
            continue;
        }
        let root = uf.find(node);
        component[node as usize] = root;
        *component_sizes.entry(root).or_insert(0u32) += 1;
    }

    let num_components = component_sizes.len() as u32;
    let largest = component_sizes.values().copied().max().unwrap_or(0);

    Ok(ComponentResult {
        component,
        num_components,
        largest_component_size: largest,
        component_sizes,
    })
}

fn consume_analytics_work(governor: Option<&crate::resource::ResourceGovernor>) -> GraphResult<()> {
    let Some(governor) = governor else {
        return Ok(());
    };
    governor
        .consume_work(
            crate::resource::ResourcePhase::AnalyticsWorkspace,
            crate::resource::WorkUnits::new(1),
        )
        .map_err(crate::safety::resource_limit_error)?;
    if governor.work_used().as_u64().is_multiple_of(1_024) {
        crate::resource::check_postgres_interrupts();
        governor
            .check_elapsed(crate::resource::ResourcePhase::AnalyticsWorkspace)
            .map_err(crate::safety::resource_limit_error)?;
    }
    Ok(())
}

fn allocation_error() -> GraphError {
    GraphError::Oom {
        used_mb: 0,
        need_mb: 1,
        limit_mb: 0,
    }
}

/// Output row for the SQL function.
#[derive(Debug)]
pub struct ComponentRow {
    pub node_table: TableOid,
    pub node_id: String,
    pub component_id: u32,
    pub component_size: u32,
}

/// Convert [`ComponentResult`] into SQL output rows.
///
/// # Errors
///
/// Returns [`GraphError::CorruptFile`] when an active component node has no
/// corresponding node metadata.
pub fn to_component_rows(
    result: &ComponentResult,
    node_store: &NodeStore,
) -> GraphResult<Vec<ComponentRow>> {
    let node_count = node_store.node_count() as usize;

    let mut rows = Vec::with_capacity(node_count);
    for node_idx in 0..node_count as u32 {
        if result.component[node_idx as usize] == u32::MAX {
            continue; // inactive
        }
        let comp_id = result.component[node_idx as usize];
        let (node_table, node_id) = node_coordinate(node_store, node_idx)?;
        rows.push(ComponentRow {
            node_table,
            node_id: node_id.to_string(),
            component_id: comp_id,
            component_size: result.component_size(comp_id),
        });
    }

    Ok(rows)
}

pub fn component_size_rows(result: &ComponentResult) -> Vec<(u32, u32)> {
    let mut rows = result
        .component_sizes
        .iter()
        .map(|(&component_id, &component_size)| (component_id, component_size))
        .collect::<Vec<_>>();
    rows.sort_by(|left, right| right.1.cmp(&left.1).then_with(|| left.0.cmp(&right.0)));
    rows
}

/// Return one page of nodes belonging to `component_id`.
///
/// # Errors
///
/// Returns [`GraphError::CorruptFile`] when a selected component node has no
/// corresponding node metadata.
pub fn component_rows_page(
    result: &ComponentResult,
    node_store: &NodeStore,
    component_id: u32,
    offset: usize,
    limit: usize,
) -> GraphResult<Vec<ComponentRow>> {
    let mut matching_nodes = result
        .component
        .iter()
        .enumerate()
        .filter_map(|(node_idx, &node_component_id)| {
            (node_component_id == component_id).then_some(node_idx as u32)
        })
        .collect::<Vec<_>>();
    matching_nodes.sort_by(|&left, &right| {
        node_store
            .table_oid(left)
            .cmp(&node_store.table_oid(right))
            .then_with(|| {
                node_store
                    .primary_key(left)
                    .cmp(&node_store.primary_key(right))
            })
    });

    let component_size = result.component_size(component_id);
    matching_nodes
        .into_iter()
        .skip(offset)
        .take(limit)
        .map(|node_idx| {
            let (node_table, node_id) = node_coordinate(node_store, node_idx)?;
            Ok(ComponentRow {
                node_table,
                node_id: node_id.to_string(),
                component_id,
                component_size,
            })
        })
        .collect()
}

/// Return one page of single-node components.
///
/// # Errors
///
/// Returns [`GraphError::CorruptFile`] when a selected isolated node has no
/// corresponding node metadata.
pub fn isolated_rows_page(
    result: &ComponentResult,
    node_store: &NodeStore,
    offset: usize,
    limit: usize,
) -> GraphResult<Vec<ComponentRow>> {
    let mut isolated_nodes = result
        .component
        .iter()
        .enumerate()
        .filter_map(|(node_idx, &component_id)| {
            (component_id != u32::MAX && result.component_size(component_id) == 1)
                .then_some((component_id, node_idx as u32))
        })
        .collect::<Vec<_>>();
    isolated_nodes.sort_by(|(left_component_id, left), (right_component_id, right)| {
        left_component_id
            .cmp(right_component_id)
            .then_with(|| {
                node_store
                    .table_oid(*left)
                    .cmp(&node_store.table_oid(*right))
            })
            .then_with(|| {
                node_store
                    .primary_key(*left)
                    .cmp(&node_store.primary_key(*right))
            })
    });

    isolated_nodes
        .into_iter()
        .skip(offset)
        .take(limit)
        .map(|(component_id, node_idx)| {
            let (node_table, node_id) = node_coordinate(node_store, node_idx)?;
            Ok(ComponentRow {
                node_table,
                node_id: node_id.to_string(),
                component_id,
                component_size: 1,
            })
        })
        .collect()
}

fn node_coordinate(node_store: &NodeStore, node_idx: u32) -> GraphResult<(TableOid, &str)> {
    let table_oid = node_store
        .table_oid(node_idx)
        .ok_or_else(|| GraphError::CorruptFile {
            reason: format!("node index {node_idx} has no table OID metadata"),
        })?;
    let node_id = node_store
        .primary_key(node_idx)
        .ok_or_else(|| GraphError::CorruptFile {
            reason: format!("node index {node_idx} has no primary-key metadata"),
        })?;
    Ok((TableOid(table_oid), node_id))
}

#[cfg(test)]
mod tests {
    //! Covers weakly connected component computation and protects component
    //! cardinality, isolation, and representative-selection invariants.

    use super::*;
    use crate::edge_store::{IdentifiedRawEdge, RawEdge, SortedEdgeStoreBuilder};
    use crate::resource::{
        ByteCount, DiskBudget, ElapsedBudget, MemoryBudget, ResourceGovernor, ResourceLimits,
        RowCount, WorkUnits,
    };
    use std::time::Duration;

    fn component_governor() -> ResourceGovernor {
        ResourceGovernor::new(ResourceLimits::bounded(
            MemoryBudget::new(ByteCount::from_bytes(1_024 * 1_024)),
            DiskBudget::UNLIMITED,
            RowCount::UNLIMITED,
            WorkUnits::new(10_000),
            ElapsedBudget::new(Duration::from_secs(1)),
        ))
    }

    #[test]
    fn single_component() {
        // 0 — 1 — 2 (all connected)
        let mut ns = NodeStore::new();
        for i in 0..3u32 {
            ns.add_node(100, format!("N{}", i));
        }
        let edges = vec![
            RawEdge {
                source: 0,
                target: 1,
                type_id: crate::types::EdgeTypeId::test_v6(1),
                weight: None,
                schema_reversed: false,
            },
            RawEdge {
                source: 1,
                target: 0,
                type_id: crate::types::EdgeTypeId::test_v6(1),
                weight: None,
                schema_reversed: false,
            },
            RawEdge {
                source: 1,
                target: 2,
                type_id: crate::types::EdgeTypeId::test_v6(1),
                weight: None,
                schema_reversed: false,
            },
            RawEdge {
                source: 2,
                target: 1,
                type_id: crate::types::EdgeTypeId::test_v6(1),
                weight: None,
                schema_reversed: false,
            },
        ];
        let es = EdgeStore::from_edges(3, edges, false);

        let result = compute_components(&ns, &es);
        assert_eq!(result.num_components, 1);
        assert_eq!(result.largest_component_size, 3);
        // All nodes should have the same component
        let c0 = result.component[0];
        assert_eq!(result.component[1], c0);
        assert_eq!(result.component[2], c0);
    }

    #[test]
    fn caller_hidden_bridge_splits_components_before_union_and_counting() {
        let mut nodes = NodeStore::new();
        for id in ["A", "hidden", "C"] {
            nodes.add_node(100, id.to_string());
        }
        let edges = EdgeStore::from_edges(
            3,
            vec![
                RawEdge {
                    source: 0,
                    target: 1,
                    type_id: crate::types::EdgeTypeId::test_v6(1),
                    weight: None,
                    schema_reversed: false,
                },
                RawEdge {
                    source: 1,
                    target: 2,
                    type_id: crate::types::EdgeTypeId::test_v6(1),
                    weight: None,
                    schema_reversed: false,
                },
            ],
            false,
        );
        let neighbors = CsrNeighbors::new(&edges);
        let mut hidden_nodes = roaring::RoaringBitmap::new();
        hidden_nodes.insert(1);
        let visibility = crate::visibility::VisibilityScope::enforced_for_test(
            hidden_nodes,
            roaring::RoaringBitmap::new(),
            roaring::RoaringBitmap::new(),
        );
        let governor = component_governor();
        let context = crate::visibility::QueryExecutionContext::new(&governor, &visibility);

        let result =
            compute_components_with_neighbors_in_context(&nodes, &neighbors, &context).unwrap();

        assert_eq!(result.component[1], u32::MAX);
        assert_eq!(result.num_components, 2);
        assert_eq!(result.largest_component_size, 1);
        assert_eq!(result.component_sizes.values().copied().sum::<u32>(), 2);
    }

    #[test]
    fn caller_hidden_relationship_is_not_unioned() {
        let mut nodes = NodeStore::new();
        nodes.add_node(100, "A".to_string());
        nodes.add_node(100, "B".to_string());
        let mut builder = SortedEdgeStoreBuilder::new(2, false);
        builder
            .try_push_identified(IdentifiedRawEdge {
                edge: RawEdge {
                    source: 0,
                    target: 1,
                    type_id: crate::types::EdgeTypeId::test_v6(1),
                    weight: None,
                    schema_reversed: false,
                },
                relationship_id: 7,
            })
            .unwrap();
        let edges = builder.finish();
        let neighbors = CsrNeighbors::new(&edges);
        let mut hidden_relationships = roaring::RoaringBitmap::new();
        hidden_relationships.insert(7);
        let mut relationship_rls_edge_types = roaring::RoaringBitmap::new();
        relationship_rls_edge_types.insert(1);
        let visibility = crate::visibility::VisibilityScope::enforced_for_test(
            roaring::RoaringBitmap::new(),
            hidden_relationships,
            relationship_rls_edge_types,
        );
        let governor = component_governor();
        let context = crate::visibility::QueryExecutionContext::new(&governor, &visibility);

        let result =
            compute_components_with_neighbors_in_context(&nodes, &neighbors, &context).unwrap();

        assert_eq!(result.num_components, 2);
        assert_eq!(result.largest_component_size, 1);
    }

    #[test]
    fn two_components() {
        // 0 — 1   2 — 3 (two separate pairs)
        let mut ns = NodeStore::new();
        for i in 0..4u32 {
            ns.add_node(100, format!("N{}", i));
        }
        let edges = vec![
            RawEdge {
                source: 0,
                target: 1,
                type_id: crate::types::EdgeTypeId::test_v6(1),
                weight: None,
                schema_reversed: false,
            },
            RawEdge {
                source: 1,
                target: 0,
                type_id: crate::types::EdgeTypeId::test_v6(1),
                weight: None,
                schema_reversed: false,
            },
            RawEdge {
                source: 2,
                target: 3,
                type_id: crate::types::EdgeTypeId::test_v6(1),
                weight: None,
                schema_reversed: false,
            },
            RawEdge {
                source: 3,
                target: 2,
                type_id: crate::types::EdgeTypeId::test_v6(1),
                weight: None,
                schema_reversed: false,
            },
        ];
        let es = EdgeStore::from_edges(4, edges, false);

        let result = compute_components(&ns, &es);
        assert_eq!(result.num_components, 2);
        assert_eq!(result.largest_component_size, 2);
        // 0 and 1 share a component, 2 and 3 share a different one
        assert_eq!(result.component[0], result.component[1]);
        assert_eq!(result.component[2], result.component[3]);
        assert_ne!(result.component[0], result.component[2]);
    }

    #[test]
    fn isolated_nodes() {
        let mut ns = NodeStore::new();
        for i in 0..5u32 {
            ns.add_node(100, format!("N{}", i));
        }
        let es = EdgeStore::from_edges(5, vec![], false);

        let result = compute_components(&ns, &es);
        assert_eq!(result.num_components, 5);
        assert_eq!(result.largest_component_size, 1);
    }

    #[test]
    fn tombstoned_nodes_excluded() {
        let mut ns = NodeStore::new();
        for i in 0..3u32 {
            ns.add_node(100, format!("N{}", i));
        }
        ns.deactivate(1); // tombstone node 1

        let edges = vec![
            RawEdge {
                source: 0,
                target: 1,
                type_id: crate::types::EdgeTypeId::test_v6(1),
                weight: None,
                schema_reversed: false,
            },
            RawEdge {
                source: 1,
                target: 0,
                type_id: crate::types::EdgeTypeId::test_v6(1),
                weight: None,
                schema_reversed: false,
            },
            RawEdge {
                source: 1,
                target: 2,
                type_id: crate::types::EdgeTypeId::test_v6(1),
                weight: None,
                schema_reversed: false,
            },
            RawEdge {
                source: 2,
                target: 1,
                type_id: crate::types::EdgeTypeId::test_v6(1),
                weight: None,
                schema_reversed: false,
            },
        ];
        let es = EdgeStore::from_edges(3, edges, false);

        let result = compute_components(&ns, &es);
        assert_eq!(result.num_components, 2); // 0 and 2 are separate (1 is tombstoned)
        assert_eq!(result.component[1], u32::MAX); // tombstoned
    }

    #[test]
    fn empty_graph() {
        let ns = NodeStore::new();
        let es = EdgeStore::from_edges(0, vec![], false);
        let result = compute_components(&ns, &es);
        assert_eq!(result.num_components, 0);
        assert_eq!(result.largest_component_size, 0);
        assert!(result.component.is_empty());
    }

    #[test]
    fn star_topology_single_component() {
        // Hub node 0 connected to 1,2,3,4
        let mut ns = NodeStore::new();
        for i in 0..5u32 {
            ns.add_node(100, format!("N{}", i));
        }
        let mut edges = Vec::new();
        for i in 1..5u32 {
            edges.push(RawEdge {
                source: 0,
                target: i,
                type_id: crate::types::EdgeTypeId::test_v6(1),
                weight: None,
                schema_reversed: false,
            });
            edges.push(RawEdge {
                source: i,
                target: 0,
                type_id: crate::types::EdgeTypeId::test_v6(1),
                weight: None,
                schema_reversed: false,
            });
        }
        let es = EdgeStore::from_edges(5, edges, false);
        let result = compute_components(&ns, &es);
        assert_eq!(result.num_components, 1);
        assert_eq!(result.largest_component_size, 5);
    }

    #[test]
    fn self_loop_single_component() {
        let mut ns = NodeStore::new();
        ns.add_node(100, "self".to_string());
        let edges = vec![RawEdge {
            source: 0,
            target: 0,
            type_id: crate::types::EdgeTypeId::test_v6(1),
            weight: None,
            schema_reversed: false,
        }];
        let es = EdgeStore::from_edges(1, edges, false);
        let result = compute_components(&ns, &es);
        assert_eq!(result.num_components, 1);
        assert_eq!(result.largest_component_size, 1);
    }

    #[test]
    fn to_component_rows_output_matches_components() {
        let mut ns = NodeStore::new();
        ns.add_node(100, "A".to_string());
        ns.add_node(100, "B".to_string());
        ns.add_node(200, "C".to_string());
        let edges = vec![
            RawEdge {
                source: 0,
                target: 1,
                type_id: crate::types::EdgeTypeId::test_v6(1),
                weight: None,
                schema_reversed: false,
            },
            RawEdge {
                source: 1,
                target: 0,
                type_id: crate::types::EdgeTypeId::test_v6(1),
                weight: None,
                schema_reversed: false,
            },
        ];
        let es = EdgeStore::from_edges(3, edges, false);
        let result = compute_components(&ns, &es);
        let rows = to_component_rows(&result, &ns).unwrap();

        assert_eq!(rows.len(), 3);
        // A and B share a component
        let a_comp = rows.iter().find(|r| r.node_id == "A").unwrap().component_id;
        let b_comp = rows.iter().find(|r| r.node_id == "B").unwrap().component_id;
        let c_comp = rows.iter().find(|r| r.node_id == "C").unwrap().component_id;
        assert_eq!(a_comp, b_comp);
        assert_ne!(a_comp, c_comp);
        // Component sizes
        assert_eq!(
            rows.iter()
                .find(|r| r.node_id == "A")
                .unwrap()
                .component_size,
            2
        );
        assert_eq!(
            rows.iter()
                .find(|r| r.node_id == "C")
                .unwrap()
                .component_size,
            1
        );
    }

    #[test]
    fn component_size_rows_reuse_computed_sizes() {
        let mut ns = NodeStore::new();
        for id in ["A", "B", "C", "D"] {
            ns.add_node(100, id.to_string());
        }
        let edges = vec![
            RawEdge {
                source: 0,
                target: 1,
                type_id: crate::types::EdgeTypeId::test_v6(1),
                weight: None,
                schema_reversed: false,
            },
            RawEdge {
                source: 1,
                target: 0,
                type_id: crate::types::EdgeTypeId::test_v6(1),
                weight: None,
                schema_reversed: false,
            },
        ];
        let es = EdgeStore::from_edges(4, edges, false);
        let result = compute_components(&ns, &es);
        let rows = component_size_rows(&result);

        assert_eq!(rows[0].1, 2);
        assert_eq!(rows.iter().filter(|(_, size)| *size == 1).count(), 2);
    }

    #[test]
    fn component_rows_page_filters_before_materializing_rows() {
        let mut ns = NodeStore::new();
        ns.add_node(200, "B".to_string());
        ns.add_node(100, "A".to_string());
        ns.add_node(300, "C".to_string());
        let edges = vec![
            RawEdge {
                source: 0,
                target: 1,
                type_id: crate::types::EdgeTypeId::test_v6(1),
                weight: None,
                schema_reversed: false,
            },
            RawEdge {
                source: 1,
                target: 0,
                type_id: crate::types::EdgeTypeId::test_v6(1),
                weight: None,
                schema_reversed: false,
            },
        ];
        let es = EdgeStore::from_edges(3, edges, false);
        let result = compute_components(&ns, &es);
        let component_id = result.component[0];
        let rows = component_rows_page(&result, &ns, component_id, 0, 1).unwrap();

        assert_eq!(rows.len(), 1);
        assert_eq!(rows[0].node_id, "A");
        assert_eq!(rows[0].component_size, 2);
    }

    #[test]
    fn isolated_rows_page_filters_and_sorts_before_materializing_rows() {
        let mut ns = NodeStore::new();
        ns.add_node(300, "C".to_string());
        ns.add_node(100, "A".to_string());
        ns.add_node(200, "B".to_string());
        let es = EdgeStore::from_edges(3, vec![], false);
        let result = compute_components(&ns, &es);
        let rows = isolated_rows_page(&result, &ns, 1, 1).unwrap();

        assert_eq!(rows.len(), 1);
        assert_eq!(rows[0].node_table.0, 100);
        assert_eq!(rows[0].node_id, "A");
        assert_eq!(rows[0].component_size, 1);
    }
}
