From 579c4a9d86191b9cba25d69098d1b4a9b8f538b0 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Sun, 9 Aug 2026 23:55:06 +0000 Subject: [PATCH 1/3] feat(exec): consume persisted CSR without HashMap expansion Closes #340 Introduce CSR-native adjacency views with O(1) row lookup, bounded delta overlays that retain the base CSR, and CSR-style analyst projection storage so index hits no longer expand the complete valid CSR into per-node hash maps. Co-authored-by: David Spencer --- crates/graphforge-exec/src/adjacency.rs | 564 ++++++++++++++---- crates/graphforge-exec/src/algorithm_graph.rs | 179 ++++-- crates/graphforge-exec/src/lib.rs | 8 +- crates/graphforge-storage/src/adjacency.rs | 89 +++ .../graphforge-storage/src/adjacency_delta.rs | 191 +++++- 5 files changed, 873 insertions(+), 158 deletions(-) diff --git a/crates/graphforge-exec/src/adjacency.rs b/crates/graphforge-exec/src/adjacency.rs index 57ec5d5df..7d5e3e077 100644 --- a/crates/graphforge-exec/src/adjacency.rs +++ b/crates/graphforge-exec/src/adjacency.rs @@ -28,9 +28,11 @@ use arrow::record_batch::RecordBatch; use graphforge_core::{GfError, OntologyMode}; use graphforge_ir::Direction; use graphforge_storage::adjacency::{ - self as csr, ALL_RELATIONS_STEM, AdjacencyManifestRow, CsrIndex, build_adjacency_index, + self as csr, ALL_RELATIONS_STEM, AdjacencyManifestRow, CsrIndex, CsrRow, build_adjacency_index, +}; +use graphforge_storage::adjacency_delta::{ + CsrDeltaOverlay, DeltaSegment, overlay_delta_segments, read_delta_chain, }; -use graphforge_storage::adjacency_delta::{DeltaSegment, apply_delta_segments, read_delta_chain}; use graphforge_storage::generation::read_topology_generation; use crate::ValueAt; @@ -63,16 +65,161 @@ impl AdjacencyStatus { } } -/// Surrogate-keyed adjacency view consumed by traversal. +/// How a view is physically backed — used for structural assertions (#340). +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum AdjacencyBacking { + /// Scan-built `HashMap` of per-node vectors (fallback / oracle path). + ScanHashMap, + /// Directed CSR served with O(1) row lookup; no O(E) expansion. + CsrNative, + /// Directed CSR plus a bounded delta-key overlay; base CSR not copied. + CsrOverlay, + /// Out+in CSR pair merged per row on access (no full merged hash map). + CsrUndirected, +} + +/// Surrogate-keyed adjacency view consumed by traversal (#340). /// -/// The representation is private. Today it is the scan-built map the retired -/// `build_adjacency` produced; #761 may back the same contract with a -/// disk-loaded CSR (offsets + interleaved entries) — both serve -/// [`neighbors`](Self::neighbors) as a contiguous slice, so consumers never -/// change. -#[derive(Clone, Debug, Default, PartialEq, Eq)] +/// Persisted-index hits keep the validated CSR (and optional bounded overlay) +/// without expanding into `HashMap>`. Scan-build fallback retains +/// the historical map representation for oracle parity. +#[derive(Clone, Debug, Default)] pub struct Adjacency { - map: HashMap>, + inner: AdjacencyInner, +} + +#[derive(Clone, Debug, Default)] +enum AdjacencyInner { + #[default] + Empty, + Map(HashMap>), + Csr(Arc), + Overlay(CsrDeltaOverlay), + Undirected { + out: Arc, + inbound: Arc, + }, +} + +/// Borrowed or owned neighbor row for one node. +#[derive(Clone, Debug)] +pub struct NeighborRow<'a> { + kind: NeighborRowKind<'a>, +} + +#[derive(Clone, Debug)] +enum NeighborRowKind<'a> { + Pairs(&'a [(u64, u64)]), + Csr(CsrRow<'a>), + Owned(Vec<(u64, u64)>), +} + +impl<'a> NeighborRow<'a> { + fn pairs(entries: &'a [(u64, u64)]) -> Self { + Self { + kind: NeighborRowKind::Pairs(entries), + } + } + + fn csr(row: CsrRow<'a>) -> Self { + Self { + kind: NeighborRowKind::Csr(row), + } + } + + fn owned(entries: Vec<(u64, u64)>) -> Self { + Self { + kind: NeighborRowKind::Owned(entries), + } + } + + /// Number of `(edge_id, neighbor_id)` entries. + #[must_use] + pub fn len(&self) -> usize { + match &self.kind { + NeighborRowKind::Pairs(entries) => entries.len(), + NeighborRowKind::Csr(row) => row.len(), + NeighborRowKind::Owned(entries) => entries.len(), + } + } + + /// Whether the row is empty. + #[must_use] + pub fn is_empty(&self) -> bool { + self.len() == 0 + } + + /// Entry at `index`, if present. + #[must_use] + pub fn get(&self, index: usize) -> Option<(u64, u64)> { + match &self.kind { + NeighborRowKind::Pairs(entries) => entries.get(index).copied(), + NeighborRowKind::Csr(row) => row.get(index), + NeighborRowKind::Owned(entries) => entries.get(index).copied(), + } + } + + /// Materialize entries (tests and rare owned-row consumers). + #[must_use] + pub fn to_vec(&self) -> Vec<(u64, u64)> { + match &self.kind { + NeighborRowKind::Pairs(entries) => entries.to_vec(), + NeighborRowKind::Csr(row) => row.iter().collect(), + NeighborRowKind::Owned(entries) => entries.clone(), + } + } + + /// Iterate `(edge_id, neighbor_id)` pairs. + pub fn iter(&self) -> NeighborRowIter<'_> { + NeighborRowIter { + row: self, + index: 0, + } + } +} + +impl PartialEq<[(u64, u64)]> for NeighborRow<'_> { + fn eq(&self, other: &[(u64, u64)]) -> bool { + self.len() == other.len() && (0..self.len()).all(|i| self.get(i) == Some(other[i])) + } +} + +impl PartialEq<&[(u64, u64)]> for NeighborRow<'_> { + fn eq(&self, other: &&[(u64, u64)]) -> bool { + self == *other + } +} + +/// Iterator over [`NeighborRow`] entries. +pub struct NeighborRowIter<'a> { + row: &'a NeighborRow<'a>, + index: usize, +} + +impl Iterator for NeighborRowIter<'_> { + type Item = (u64, u64); + + fn next(&mut self) -> Option { + let item = self.row.get(self.index)?; + self.index += 1; + Some(item) + } + + fn size_hint(&self) -> (usize, Option) { + let remaining = self.row.len().saturating_sub(self.index); + (remaining, Some(remaining)) + } +} + +impl ExactSizeIterator for NeighborRowIter<'_> {} + +impl<'a> IntoIterator for &'a NeighborRow<'a> { + type Item = (u64, u64); + type IntoIter = NeighborRowIter<'a>; + + fn into_iter(self) -> Self::IntoIter { + self.iter() + } } impl Adjacency { @@ -81,30 +228,196 @@ impl Adjacency { /// and dst-keyed entries in that order). BFS emission order observably /// depends on this ordering. /// - /// Unknown, isolated, or deleted ids yield `&[]` — never a panic. (#761's - /// dense CSR backing must preserve this for out-of-range ids.) + /// Unknown, isolated, or deleted ids yield an empty row — never a panic. + #[must_use] + pub fn neighbors(&self, node_id: u64) -> NeighborRow<'_> { + self.inner.neighbors(node_id) + } + + /// Physical backing representation (#340 structural counter surface). + #[must_use] + pub fn backing(&self) -> AdjacencyBacking { + self.inner.backing() + } + + /// Number of base-CSR adjacency entries that were expanded into a hash map + /// or per-node heap vectors while constructing this view. + /// + /// Persisted CSR hits (directed, undirected, and bounded overlays) report + /// `0`. Scan-build fallback reports the number of map entries inserted. + #[must_use] + pub fn base_csr_entries_expanded(&self) -> u64 { + self.inner.base_csr_entries_expanded() + } + + /// Overlay replacement-row count (0 unless [`AdjacencyBacking::CsrOverlay`] + /// or an undirected pair that includes an overlay). #[must_use] - pub fn neighbors(&self, node_id: u64) -> &[(u64, u64)] { - self.map.get(&node_id).map_or(&[], Vec::as_slice) + pub fn overlay_row_count(&self) -> u64 { + self.inner.overlay_row_count() } - /// Whether the view contains no entries at all. + /// Whether the view contains no adjacency entries at all. #[must_use] pub fn is_empty(&self) -> bool { - self.map.is_empty() + self.inner.is_empty() } - /// Iterate the surrogate-keyed rows in unspecified map order. + /// Visit every node that may have adjacency entries. /// - /// Consumers that expose deterministic behavior must sort the yielded - /// node ids. The entries within each row retain provider order. - pub(crate) fn rows(&self) -> impl Iterator { - self.map - .iter() - .map(|(&node_id, entries)| (node_id, entries.as_slice())) + /// CSR-native backings yield dense `0..node_extent` ids (empty rows + /// included only when the callback needs them — callers that skip empty + /// rows should check [`NeighborRow::is_empty`]). Map backings yield only + /// keys present in the map. + pub(crate) fn for_each_row(&self, mut visit: impl FnMut(u64, NeighborRow<'_>)) { + self.inner.for_each_row(&mut visit); + } +} + +impl AdjacencyInner { + fn is_empty(&self) -> bool { + match self { + Self::Empty => true, + Self::Map(map) => map.is_empty(), + Self::Csr(csr) => csr.edge_count() == 0, + Self::Overlay(overlay) => { + overlay.base.edge_count() == 0 + && overlay.replaced.values().all(|row| row.is_empty()) + } + Self::Undirected { out, inbound } => out.is_empty() && inbound.is_empty(), + } + } + + fn neighbors(&self, node_id: u64) -> NeighborRow<'_> { + match self { + Self::Empty => NeighborRow::pairs(&[]), + Self::Map(map) => NeighborRow::pairs(map.get(&node_id).map_or(&[], Vec::as_slice)), + Self::Csr(csr) => NeighborRow::csr(csr.row(node_id)), + Self::Overlay(overlay) => match overlay.row(node_id) { + graphforge_storage::adjacency_delta::OverlayRow::Base(row) => NeighborRow::csr(row), + graphforge_storage::adjacency_delta::OverlayRow::Replaced(entries) => { + NeighborRow::pairs(entries) + } + }, + Self::Undirected { out, inbound } => { + merge_undirected_row(out.neighbors(node_id), inbound.neighbors(node_id)) + } + } + } + + fn backing(&self) -> AdjacencyBacking { + match self { + Self::Empty | Self::Map(_) => AdjacencyBacking::ScanHashMap, + Self::Csr(_) => AdjacencyBacking::CsrNative, + Self::Overlay(_) => AdjacencyBacking::CsrOverlay, + Self::Undirected { .. } => AdjacencyBacking::CsrUndirected, + } + } + + fn base_csr_entries_expanded(&self) -> u64 { + match self { + Self::Empty | Self::Csr(_) | Self::Overlay(_) => 0, + Self::Map(map) => u64::try_from(map.values().map(Vec::len).sum::()).unwrap_or(0), + Self::Undirected { out, inbound } => out + .base_csr_entries_expanded() + .saturating_add(inbound.base_csr_entries_expanded()), + } + } + + fn overlay_row_count(&self) -> u64 { + match self { + Self::Overlay(overlay) => overlay.overlay_row_count(), + Self::Undirected { out, inbound } => out + .overlay_row_count() + .saturating_add(inbound.overlay_row_count()), + _ => 0, + } + } + + fn node_extent(&self) -> u64 { + match self { + Self::Empty => 0, + Self::Map(map) => map.keys().next().map_or(0, |_| { + map.keys().copied().max().map_or(0, |m| m.saturating_add(1)) + }), + Self::Csr(csr) => csr.node_count(), + Self::Overlay(overlay) => overlay.node_extent, + Self::Undirected { out, inbound } => out.node_extent().max(inbound.node_extent()), + } + } + + fn for_each_row(&self, visit: &mut dyn FnMut(u64, NeighborRow<'_>)) { + match self { + Self::Empty => {} + Self::Map(map) => { + for (&node_id, entries) in map { + visit(node_id, NeighborRow::pairs(entries.as_slice())); + } + } + Self::Csr(csr) => { + for node_id in 0..csr.node_count() { + visit(node_id, NeighborRow::csr(csr.row(node_id))); + } + } + Self::Overlay(overlay) => { + for node_id in 0..overlay.node_extent { + visit(node_id, self.neighbors(node_id)); + } + } + Self::Undirected { out, inbound } => { + let extent = out.node_extent().max(inbound.node_extent()); + for node_id in 0..extent { + visit(node_id, self.neighbors(node_id)); + } + } + } + } +} + +impl PartialEq for Adjacency { + fn eq(&self, other: &Self) -> bool { + fn collect(view: &Adjacency) -> Vec<(u64, Vec<(u64, u64)>)> { + let mut rows = Vec::new(); + view.for_each_row(|node_id, row| { + if !row.is_empty() { + rows.push((node_id, row.to_vec())); + } + }); + rows.sort_unstable_by_key(|(node_id, _)| *node_id); + rows + } + collect(self) == collect(other) } } +impl Eq for Adjacency {} + +/// Merge out and in rows with ascending `edge_id` and **out before in on ties**. +fn merge_undirected_row<'a>(out: NeighborRow<'a>, inbound: NeighborRow<'a>) -> NeighborRow<'a> { + if out.is_empty() { + return NeighborRow::owned(inbound.to_vec()); + } + if inbound.is_empty() { + return NeighborRow::owned(out.to_vec()); + } + let mut entries = Vec::with_capacity(out.len() + inbound.len()); + let mut o = 0usize; + let mut i = 0usize; + while o < out.len() || i < inbound.len() { + let take_out = i >= inbound.len() + || (o < out.len() + && out.get(o).map(|(e, _)| e) <= inbound.get(i).map(|(e, _)| e)); + if take_out { + entries.push(out.get(o).expect("out index in range")); + o += 1; + } else { + entries.push(inbound.get(i).expect("in index in range")); + i += 1; + } + } + NeighborRow::owned(entries) +} + /// Single adjacency abstraction (ADR 0005): implementations decide *how* a /// view is produced (scan-build now; disk-loaded CSR with scan fallback in /// #761) — consumers only see [`Adjacency`]. @@ -218,7 +531,13 @@ fn build_from_edge_batches( } } } - Ok(Adjacency { map: adj }) + Ok(Adjacency { + inner: if adj.is_empty() { + AdjacencyInner::Empty + } else { + AdjacencyInner::Map(adj) + }, + }) } // --------------------------------------------------------------------------- @@ -360,6 +679,10 @@ impl PersistentAdjacencyProvider { /// Load the view for (`stem`, `direction`) from the CSR file(s), overlaying /// the delta chain (#765) when one is present (`deltas` non-empty). + /// + /// A fresh base CSR is retained as a CSR-native view (#340): no O(E) + /// HashMap expansion. Non-empty delta chains attach a bounded overlay that + /// replaces only touched keys. fn load( &self, stem: &str, @@ -367,10 +690,10 @@ impl PersistentAdjacencyProvider { rows: &[AdjacencyManifestRow], deltas: &[DeltaSegment], ) -> Result { - let overlaid = |d: csr::Direction| -> Result { - let base = csr::read_csr(&csr::csr_path(&self.dir, stem, d))?; + let directed = |d: csr::Direction| -> Result { + let base = Arc::new(csr::read_csr(&csr::csr_path(&self.dir, stem, d))?); if deltas.is_empty() { - return Ok(base); + return Ok(AdjacencyInner::Csr(base)); } // Torn-read count guard (#765): the base CSR must match the manifest // row it was loaded against. If a concurrent compaction rewrote the @@ -385,15 +708,26 @@ impl PersistentAdjacencyProvider { "adjacency base CSR disagrees with manifest counts (torn read)".into(), )); } - Ok(apply_delta_segments(&base, stem, d, deltas)) + let overlay = overlay_delta_segments(base, stem, d, deltas); + if overlay.replaced.is_empty() { + Ok(AdjacencyInner::Csr(overlay.base)) + } else { + Ok(AdjacencyInner::Overlay(overlay)) + } }; match direction { - Direction::Out => Ok(adjacency_from_csr(&overlaid(csr::Direction::Out)?)), - Direction::In => Ok(adjacency_from_csr(&overlaid(csr::Direction::In)?)), - Direction::Undirected => Ok(merge_undirected( - &overlaid(csr::Direction::Out)?, - &overlaid(csr::Direction::In)?, - )), + Direction::Out => Ok(Adjacency { + inner: directed(csr::Direction::Out)?, + }), + Direction::In => Ok(Adjacency { + inner: directed(csr::Direction::In)?, + }), + Direction::Undirected => Ok(Adjacency { + inner: AdjacencyInner::Undirected { + out: Arc::new(directed(csr::Direction::Out)?), + inbound: Arc::new(directed(csr::Direction::In)?), + }, + }), } } @@ -589,79 +923,34 @@ impl AdjacencyProvider for PersistentAdjacencyProvider { } } -/// Convert a directed CSR into the traversal view. Empty rows are skipped so -/// the result is `PartialEq`-identical to a scan-built view (scan-build never -/// inserts empty entry vectors). +/// Convert a directed CSR into the traversal view without O(E) HashMap +/// expansion (#340). Empty CSR becomes [`AdjacencyInner::Empty`]. /// /// Bounds safety: callers only pass [`CsrIndex`]es produced by /// `graphforge_storage::adjacency::read_csr`, which validates the invariants /// (offsets `[0]`-rooted, monotone, final offset == target lengths) before -/// returning — so every slice below is in bounds and the `usize` conversions -/// cannot fail (offsets never exceed `Vec::len`, which is a `usize`). A -/// malformed file errors inside `read_csr`, taking the rebuild/fallback path -/// in the provider, never a panic here. -fn adjacency_from_csr(index: &CsrIndex) -> Adjacency { - let mut map: HashMap> = HashMap::new(); - for (node, window) in index.offsets.windows(2).enumerate() { - let (start, end) = ( - usize::try_from(window[0]).expect("CSR offset exceeds usize"), - usize::try_from(window[1]).expect("CSR offset exceeds usize"), - ); - if start == end { - continue; +/// returning. A malformed file errors inside `read_csr`, taking the +/// rebuild/fallback path in the provider, never a panic here. +#[cfg(test)] +fn adjacency_from_csr(index: CsrIndex) -> Adjacency { + if index.edge_count() == 0 { + Adjacency::default() + } else { + Adjacency { + inner: AdjacencyInner::Csr(Arc::new(index)), } - map.insert( - node as u64, - index.edge_ids[start..end] - .iter() - .copied() - .zip(index.neighbor_ids[start..end].iter().copied()) - .collect(), - ); } - Adjacency { map } } -/// Merge a node's out and in CSR rows into the undirected view, interleaving -/// ascending by `edge_id` with **out before in on ties** — the exact order -/// scan-build produces (edge files are edge_id-ascending; each row pushes its -/// src-keyed entry, then for a self-loop its dst-keyed entry). -/// -/// Bounds safety: same `read_csr`-validated invariants as -/// [`adjacency_from_csr`]; `row()` additionally guards `node` against the -/// CSR's own `node_count`. -fn merge_undirected(out: &CsrIndex, inbound: &CsrIndex) -> Adjacency { - let row = |index: &CsrIndex, node: u64| -> (usize, usize) { - if node >= index.node_count() { - return (0, 0); - } - let i = usize::try_from(node).expect("node id exceeds usize"); - ( - usize::try_from(index.offsets[i]).expect("CSR offset exceeds usize"), - usize::try_from(index.offsets[i + 1]).expect("CSR offset exceeds usize"), - ) - }; - let mut map: HashMap> = HashMap::new(); - for node in 0..out.node_count().max(inbound.node_count()) { - let (mut o, o_end) = row(out, node); - let (mut i, i_end) = row(inbound, node); - if o == o_end && i == i_end { - continue; - } - let mut entries = Vec::with_capacity((o_end - o) + (i_end - i)); - while o < o_end || i < i_end { - let take_out = i >= i_end || (o < o_end && out.edge_ids[o] <= inbound.edge_ids[i]); - if take_out { - entries.push((out.edge_ids[o], out.neighbor_ids[o])); - o += 1; - } else { - entries.push((inbound.edge_ids[i], inbound.neighbor_ids[i])); - i += 1; - } - } - map.insert(node, entries); +/// Undirected CSR pair without materializing a merged hash map (#340). +#[cfg(test)] +fn merge_undirected(out: CsrIndex, inbound: CsrIndex) -> Adjacency { + Adjacency { + inner: AdjacencyInner::Undirected { + out: Arc::new(AdjacencyInner::Csr(Arc::new(out))), + inbound: Arc::new(AdjacencyInner::Csr(Arc::new(inbound))), + }, } - Adjacency { map } } // --------------------------------------------------------------------------- @@ -710,14 +999,14 @@ mod tests { // a has three outgoing entries: a→b, a→c, then the parallel a→b. assert_eq!(out.neighbors(a).len(), 3); // The self-loop is one Out entry under d... - assert_eq!(out.neighbors(d), &[(6, d)]); + assert_eq!(out.neighbors(d).to_vec(), vec![(6, d)]); // ...and two Undirected entries under d (src-keyed + dst-keyed), after // the two incoming diamond edges b→d, c→d. let undirected = provider.adjacency("KNOWS", Direction::Undirected).unwrap(); assert_eq!(undirected.neighbors(d).len(), 4); let in_view = provider.adjacency("KNOWS", Direction::In).unwrap(); // b's incoming: a→b and the parallel a→b. - assert_eq!(in_view.neighbors(b), &[(1, a), (5, a)]); + assert_eq!(in_view.neighbors(b).to_vec(), vec![(1, a), (5, a)]); } /// `KNOWS` and `OWNS` rows share `_exploratory.parquet`; the rel filter @@ -738,7 +1027,11 @@ mod tests { let provider = ScanBuildAdjacencyProvider::new(dir.path().to_path_buf(), OntologyMode::Exploratory); let view = provider.adjacency("KNOWS", Direction::Out).unwrap(); - assert_eq!(view.neighbors(ids[0]), &[(1, ids[1])], "OWNS row excluded"); + assert_eq!( + view.neighbors(ids[0]).to_vec(), + vec![(1, ids[1])], + "OWNS row excluded" + ); } #[test] @@ -758,8 +1051,8 @@ mod tests { ScanBuildAdjacencyProvider::new(dir.path().to_path_buf(), OntologyMode::Exploratory); let view = provider.adjacency("*", Direction::Out).unwrap(); assert_eq!( - view.neighbors(ids[0]), - &[(1, ids[1]), (2, ids[2])], + view.neighbors(ids[0]).to_vec(), + vec![(1, ids[1]), (2, ids[2])], "wildcard skips the rel filter" ); } @@ -824,8 +1117,15 @@ mod tests { let provider = ScanBuildAdjacencyProvider::new(dir.path().to_path_buf(), OntologyMode::Strict); let view = provider.adjacency("KNOWS", Direction::Out).unwrap(); - assert_eq!(view.neighbors(ids[1]), &[], "deleted id yields empty"); - assert_eq!(view.neighbors(ids[2]), &[(3, ids[3])], "survivor intact"); + assert!( + view.neighbors(ids[1]).is_empty(), + "deleted id yields empty" + ); + assert_eq!( + view.neighbors(ids[2]).to_vec(), + vec![(3, ids[3])], + "survivor intact" + ); } #[test] @@ -835,7 +1135,7 @@ mod tests { ScanBuildAdjacencyProvider::new(dir.path().to_path_buf(), OntologyMode::Strict); let view = provider.adjacency("KNOWS", Direction::Out).unwrap(); assert!(view.is_empty()); - assert_eq!(view.neighbors(1), &[]); + assert!(view.neighbors(1).is_empty()); } /// Neighbor order is edge-file row order — BFS emission order depends on @@ -860,12 +1160,60 @@ mod tests { ScanBuildAdjacencyProvider::new(dir.path().to_path_buf(), OntologyMode::Strict); let view = provider.adjacency("KNOWS", Direction::Out).unwrap(); assert_eq!( - view.neighbors(hub_id), - &[(1, spoke_ids[0]), (2, spoke_ids[1]), (3, spoke_ids[2])], + view.neighbors(hub_id).to_vec(), + vec![(1, spoke_ids[0]), (2, spoke_ids[1]), (3, spoke_ids[2])], "entries in edge-file row order" ); } + #[test] + fn persisted_csr_hit_does_not_expand_base_into_hash_map() { + let dir = TempDir::new().unwrap(); + let _ids = write_diamond(dir.path()); + graphforge_storage::adjacency::build_adjacency_index(dir.path(), TS).unwrap(); + let provider = + PersistentAdjacencyProvider::new(dir.path().to_path_buf(), OntologyMode::Strict); + assert_eq!( + provider.status("KNOWS", Direction::Out), + AdjacencyStatus::Hit + ); + let out = provider.adjacency("KNOWS", Direction::Out).unwrap(); + assert_eq!(out.backing(), AdjacencyBacking::CsrNative); + assert_eq!(out.base_csr_entries_expanded(), 0); + assert_eq!(out.overlay_row_count(), 0); + + let undirected = provider.adjacency("KNOWS", Direction::Undirected).unwrap(); + assert_eq!(undirected.backing(), AdjacencyBacking::CsrUndirected); + assert_eq!(undirected.base_csr_entries_expanded(), 0); + + let scanned = ScanBuildAdjacencyProvider::new( + dir.path().to_path_buf(), + OntologyMode::Strict, + ) + .adjacency("KNOWS", Direction::Out) + .unwrap(); + assert_eq!(out.as_ref(), scanned.as_ref(), "CSR-native matches scan oracle"); + assert!(scanned.base_csr_entries_expanded() > 0); + assert_eq!(scanned.backing(), AdjacencyBacking::ScanHashMap); + } + + #[test] + fn undirected_csr_merge_preserves_out_before_in_ties() { + let out = CsrIndex { + offsets: vec![0, 1], + edge_ids: vec![7], + neighbor_ids: vec![1], + }; + let inbound = CsrIndex { + offsets: vec![0, 1], + edge_ids: vec![7], + neighbor_ids: vec![2], + }; + let view = merge_undirected(out, inbound); + assert_eq!(view.neighbors(0).to_vec(), vec![(7, 1), (7, 2)]); + assert_eq!(view.base_csr_entries_expanded(), 0); + } + #[test] fn scan_build_status_is_building() { let dir = TempDir::new().unwrap(); diff --git a/crates/graphforge-exec/src/algorithm_graph.rs b/crates/graphforge-exec/src/algorithm_graph.rs index 2f884e81c..207b70127 100644 --- a/crates/graphforge-exec/src/algorithm_graph.rs +++ b/crates/graphforge-exec/src/algorithm_graph.rs @@ -91,16 +91,68 @@ struct LogicalProjectionEdge { } /// Algorithm graph with deterministic surrogate ordering and UUID round trips. +/// +/// Neighbor storage is CSR-style (#340): one flat `AlgorithmEdge` vector plus +/// per-node offset ranges. UUID identity keeps a compact by-id map and the +/// reverse UUID map only where lookup requires them — not a second full-graph +/// edge hash map. #[derive(Clone, Debug, Default)] pub(crate) struct AdjacencyGraph { directed: bool, node_ids: Vec, node_uuid_by_id: HashMap, node_id_by_uuid: HashMap<[u8; 16], u64>, - neighbors: HashMap>, + /// Maps node surrogate → row index into [`Self::neighbor_offsets`]. + neighbor_row: HashMap, + /// CSR offsets over [`Self::neighbor_edges`], length `row_count + 1`. + neighbor_offsets: Vec, + /// Flat adjacency entries in CSR order. + neighbor_edges: Vec, node_vectors: HashMap>, } +/// Build CSR neighbor arrays from `(node_id, edge)` pairs already sorted by +/// `(node_id, edge_id, neighbor_id)`. +fn neighbor_csr_from_sorted( + sorted: &[(u64, AlgorithmEdge)], +) -> (HashMap, Vec, Vec) { + let mut neighbor_row = HashMap::new(); + let mut neighbor_offsets = Vec::new(); + let mut neighbor_edges = Vec::with_capacity(sorted.len()); + neighbor_offsets.push(0); + let mut row = 0_u32; + let mut index = 0usize; + while index < sorted.len() { + let node_id = sorted[index].0; + neighbor_row.insert(node_id, row); + while index < sorted.len() && sorted[index].0 == node_id { + neighbor_edges.push(sorted[index].1); + index += 1; + } + neighbor_offsets.push( + u32::try_from(neighbor_edges.len()).expect("algorithm neighbor count fits u32"), + ); + row = row.saturating_add(1); + } + (neighbor_row, neighbor_offsets, neighbor_edges) +} + +fn neighbor_csr_from_map( + mut neighbors: HashMap>, +) -> (HashMap, Vec, Vec) { + let mut sorted = Vec::new(); + let mut keys = neighbors.keys().copied().collect::>(); + keys.sort_unstable(); + for node_id in keys { + let mut entries = neighbors.remove(&node_id).unwrap_or_default(); + entries.sort_by_key(|edge| (edge.edge_id, edge.neighbor_id)); + for edge in entries { + sorted.push((node_id, edge)); + } + } + neighbor_csr_from_sorted(&sorted) +} + impl AdjacencyGraph { #[cfg(test)] pub(crate) fn malformed_for_defensive_tests( @@ -113,12 +165,15 @@ impl AdjacencyGraph { .iter() .map(|(node_id, uuid)| (*uuid, *node_id)) .collect(); + let (neighbor_row, neighbor_offsets, neighbor_edges) = neighbor_csr_from_map(neighbors); Self { directed, node_ids, node_uuid_by_id, node_id_by_uuid, - neighbors, + neighbor_row, + neighbor_offsets, + neighbor_edges, node_vectors: HashMap::new(), } } @@ -343,12 +398,15 @@ impl AdjacencyGraph { for entries in neighbors.values_mut() { entries.sort_by_key(|edge| (edge.edge_id, edge.neighbor_id)); } + let (neighbor_row, neighbor_offsets, neighbor_edges) = neighbor_csr_from_map(neighbors); Ok(Self { directed: projection.directed, node_ids, node_uuid_by_id, node_id_by_uuid, - neighbors, + neighbor_row, + neighbor_offsets, + neighbor_edges, node_vectors: HashMap::new(), }) } @@ -375,7 +433,18 @@ impl AdjacencyGraph { /// Adjacency entries for `node_id`, ordered by `(edge_id, neighbor_id)`. #[must_use] pub(crate) fn neighbors(&self, node_id: u64) -> &[AlgorithmEdge] { - self.neighbors.get(&node_id).map_or(&[], Vec::as_slice) + let Some(&row) = self.neighbor_row.get(&node_id) else { + return &[]; + }; + let row = usize::try_from(row).unwrap_or(usize::MAX); + if row + 1 >= self.neighbor_offsets.len() { + return &[]; + } + let start = usize::try_from(self.neighbor_offsets[row]).unwrap_or(0); + let end = usize::try_from(self.neighbor_offsets[row + 1]).unwrap_or(start); + let end = end.min(self.neighbor_edges.len()); + let start = start.min(end); + &self.neighbor_edges[start..end] } /// Resolve an internal node surrogate to its public UUID. @@ -423,10 +492,7 @@ impl AdjacencyGraph { /// Number of direction-expanded adjacency entries. pub(crate) fn edge_entry_count(&self) -> u64 { - self.neighbors - .values() - .map(|entries| u64::try_from(entries.len()).unwrap_or(u64::MAX)) - .fold(0_u64, u64::saturating_add) + u64::try_from(self.neighbor_edges.len()).unwrap_or(u64::MAX) } #[cfg(test)] @@ -455,12 +521,15 @@ impl AdjacencyGraph { .collect(), )]) }; + let (neighbor_row, neighbor_offsets, neighbor_edges) = neighbor_csr_from_map(neighbors); Self { directed: false, node_ids, node_uuid_by_id, node_id_by_uuid, - neighbors, + neighbor_row, + neighbor_offsets, + neighbor_edges, node_vectors: HashMap::new(), } } @@ -486,12 +555,15 @@ impl AdjacencyGraph { weight: 1.0, }); } + let (neighbor_row, neighbor_offsets, neighbor_edges) = neighbor_csr_from_map(neighbors); Self { directed: false, node_ids, node_uuid_by_id, node_id_by_uuid, - neighbors, + neighbor_row, + neighbor_offsets, + neighbor_edges, node_vectors: HashMap::new(), } } @@ -548,19 +620,22 @@ impl AdjacencyGraph { neighbors.entry(target).or_default().push(edge(source)); } } + let (neighbor_row, neighbor_offsets, neighbor_edges) = neighbor_csr_from_map(neighbors); Self { directed: false, node_ids, node_uuid_by_id, node_id_by_uuid, - neighbors, + neighbor_row, + neighbor_offsets, + neighbor_edges, node_vectors: HashMap::new(), } } #[cfg(test)] pub(crate) fn with_test_edge_weights(mut self, weights: &[f64]) -> Self { - let mut edges = self.neighbors.values_mut().flatten().collect::>(); + let mut edges = self.neighbor_edges.iter_mut().collect::>(); edges.sort_unstable_by_key(|edge| edge.edge_id); assert_eq!(edges.len(), weights.len()); for (edge, &weight) in edges.into_iter().zip(weights) { @@ -585,7 +660,9 @@ pub(crate) fn export_node_selection( node_ids, node_uuid_by_id, node_id_by_uuid, - neighbors: HashMap::new(), + neighbor_row: HashMap::new(), + neighbor_offsets: vec![0], + neighbor_edges: Vec::new(), node_vectors: HashMap::new(), }) } @@ -610,13 +687,12 @@ pub(crate) fn export_adjacency( let (node_ids, node_uuid_by_id) = selected_nodes(dir, selection.label)?; let selected: HashSet = node_ids.iter().copied().collect(); + // Bound projection work by the selected node set (#340): look up each + // selected node's CSR/map row instead of scanning the complete graph. let mut raw = Vec::new(); let mut edge_ids = HashSet::new(); - for (node_id, entries) in adjacency.rows() { - if !selected.contains(&node_id) { - continue; - } - for &(edge_id, neighbor_id) in entries { + for &node_id in &node_ids { + for (edge_id, neighbor_id) in adjacency.neighbors(node_id) { if selected.contains(&neighbor_id) { raw.push((node_id, edge_id, neighbor_id)); edge_ids.insert(edge_id); @@ -631,7 +707,7 @@ pub(crate) fn export_adjacency( None => HashMap::new(), }; - let mut neighbors: HashMap> = HashMap::new(); + let mut sorted = Vec::with_capacity(raw.len()); for (node_id, edge_id, neighbor_id) in raw { let edge_uuid = edge_uuids.get(&edge_id).copied().ok_or_else(|| { GfError::Execution(format!("adjacency edge {edge_id} has no topology UUID")) @@ -645,16 +721,18 @@ pub(crate) fn export_adjacency( uuid_text(&edge_uuid) ))); } - neighbors.entry(node_id).or_default().push(AlgorithmEdge { - edge_id, - edge_uuid, - neighbor_id, - weight, - }); - } - for entries in neighbors.values_mut() { - entries.sort_by_key(|edge| (edge.edge_id, edge.neighbor_id)); + sorted.push(( + node_id, + AlgorithmEdge { + edge_id, + edge_uuid, + neighbor_id, + weight, + }, + )); } + sorted.sort_by_key(|(node_id, edge)| (*node_id, edge.edge_id, edge.neighbor_id)); + let (neighbor_row, neighbor_offsets, neighbor_edges) = neighbor_csr_from_sorted(&sorted); let node_id_by_uuid = node_uuid_by_id .iter() .map(|(&id, &uuid)| (uuid, id)) @@ -664,7 +742,9 @@ pub(crate) fn export_adjacency( node_ids, node_uuid_by_id, node_id_by_uuid, - neighbors, + neighbor_row, + neighbor_offsets, + neighbor_edges, node_vectors: HashMap::new(), }) } @@ -2151,28 +2231,35 @@ mod tests { let uuid0 = u128::from(10_u8).to_be_bytes(); let uuid1 = u128::from(11_u8).to_be_bytes(); let edge_uuid = u128::from(12_u8).to_be_bytes(); - let base = || AdjacencyGraph { - directed: true, - node_ids: vec![0, 1], - node_uuid_by_id: HashMap::from([(0, uuid0), (1, uuid1)]), - node_id_by_uuid: HashMap::from([(uuid0, 0), (uuid1, 1)]), - neighbors: HashMap::from([( - 0, - vec![AlgorithmEdge { - edge_id: 0, - neighbor_id: 1, - edge_uuid, - weight: 2.5, - }], - )]), - node_vectors: HashMap::new(), + let base = || { + let (neighbor_row, neighbor_offsets, neighbor_edges) = neighbor_csr_from_map( + HashMap::from([( + 0, + vec![AlgorithmEdge { + edge_id: 0, + neighbor_id: 1, + edge_uuid, + weight: 2.5, + }], + )]), + ); + AdjacencyGraph { + directed: true, + node_ids: vec![0, 1], + node_uuid_by_id: HashMap::from([(0, uuid0), (1, uuid1)]), + node_id_by_uuid: HashMap::from([(uuid0, 0), (uuid1, 1)]), + neighbor_row, + neighbor_offsets, + neighbor_edges, + node_vectors: HashMap::new(), + } }; let valid = base().projection_fingerprint().expect("valid projection"); assert_ne!(valid.as_bytes(), &[0; 32]); let mut missing_target = base(); - missing_target.neighbors.get_mut(&0).unwrap()[0].neighbor_id = 9; + missing_target.neighbor_edges[0].neighbor_id = 9; assert_eq!( missing_target .projection_fingerprint() @@ -2182,7 +2269,7 @@ mod tests { ); let mut non_finite = base(); - non_finite.neighbors.get_mut(&0).unwrap()[0].weight = f64::NAN; + non_finite.neighbor_edges[0].weight = f64::NAN; assert_eq!( non_finite.projection_fingerprint().unwrap_err().to_string(), "execution error: algorithm projection edge weight is not finite" diff --git a/crates/graphforge-exec/src/lib.rs b/crates/graphforge-exec/src/lib.rs index 2436b9391..279320bc0 100644 --- a/crates/graphforge-exec/src/lib.rs +++ b/crates/graphforge-exec/src/lib.rs @@ -189,7 +189,7 @@ pub(crate) mod algorithm_weighted_undirected; #[doc(hidden)] pub mod demand; pub use adjacency::{ - Adjacency, AdjacencyProvider, AdjacencyStatus, PersistentAdjacencyProvider, + Adjacency, AdjacencyBacking, AdjacencyProvider, AdjacencyStatus, PersistentAdjacencyProvider, ScanBuildAdjacencyProvider, }; pub use algorithm_analyze::{ @@ -2722,7 +2722,7 @@ fn bfs_emit( if cfg.max_hops.is_some_and(|m| p.hops >= m) { continue; } - for &(edge_id, next) in adjacency.neighbors(p.node) { + for (edge_id, next) in adjacency.neighbors(p.node) { if p.visited_edges.contains(&edge_id) { continue; // relationship isomorphism: no edge twice per path } @@ -3484,7 +3484,9 @@ fn expand_single_hop_chunk( }; let neighbors = adjacency.neighbors(src); while position.neighbor_offset < neighbors.len() && triples.len() < max_output { - let (edge_id, neighbor) = neighbors[position.neighbor_offset]; + let (edge_id, neighbor) = neighbors + .get(position.neighbor_offset) + .expect("neighbor_offset < len"); position.neighbor_offset += 1; if matches!(cfg.direction, Direction::Undirected) && !position.seen_edges.insert(edge_id) diff --git a/crates/graphforge-storage/src/adjacency.rs b/crates/graphforge-storage/src/adjacency.rs index bcc1d1c10..ae3e7f32b 100644 --- a/crates/graphforge-storage/src/adjacency.rs +++ b/crates/graphforge-storage/src/adjacency.rs @@ -139,6 +139,36 @@ impl CsrIndex { self.edge_ids.len() as u64 } + /// Checked O(1) row lookup: parallel `(edge_id, neighbor_id)` slices for + /// `node_id`, or empty slices when the id is out of range / isolated. + /// + /// Callers that loaded this CSR via [`read_csr`] already validated offsets, + /// so the returned ranges are always in bounds. + #[must_use] + pub fn row(&self, node_id: u64) -> CsrRow<'_> { + if node_id >= self.node_count() { + return CsrRow { + edge_ids: &[], + neighbor_ids: &[], + }; + } + let i = usize::try_from(node_id).unwrap_or(usize::MAX); + if i >= self.offsets.len().saturating_sub(1) { + return CsrRow { + edge_ids: &[], + neighbor_ids: &[], + }; + } + let start = usize::try_from(self.offsets[i]).unwrap_or(0); + let end = usize::try_from(self.offsets[i + 1]).unwrap_or(start); + let end = end.min(self.edge_ids.len()).min(self.neighbor_ids.len()); + let start = start.min(end); + CsrRow { + edge_ids: &self.edge_ids[start..end], + neighbor_ids: &self.neighbor_ids[start..end], + } + } + /// Check the structural invariants listed on the type. fn validate(&self) -> Result<(), GfError> { if self.offsets.first() != Some(&0) { @@ -165,6 +195,47 @@ impl CsrIndex { } } +/// Borrowed CSR row: parallel edge-id and neighbor-id slices (same length). +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct CsrRow<'a> { + /// Edge surrogates for this row, in CSR order. + pub edge_ids: &'a [u64], + /// Neighbor node surrogates aligned with [`Self::edge_ids`]. + pub neighbor_ids: &'a [u64], +} + +impl<'a> CsrRow<'a> { + /// Number of `(edge, neighbor)` entries in this row. + #[must_use] + pub fn len(&self) -> usize { + self.edge_ids.len().min(self.neighbor_ids.len()) + } + + /// Whether the row has no adjacency entries. + #[must_use] + pub fn is_empty(&self) -> bool { + self.len() == 0 + } + + /// Entry at `index`, if in range. + #[must_use] + pub fn get(&self, index: usize) -> Option<(u64, u64)> { + if index < self.len() { + Some((self.edge_ids[index], self.neighbor_ids[index])) + } else { + None + } + } + + /// Iterate `(edge_id, neighbor_id)` pairs without allocating. + pub fn iter(self) -> impl Iterator + 'a { + self.edge_ids + .iter() + .copied() + .zip(self.neighbor_ids.iter().copied()) + } +} + /// One row of `index_manifest.parquet` — the build record for a single CSR /// file. See [`ADJACENCY_MANIFEST_SCHEMA`]. #[derive(Clone, Debug, PartialEq, Eq)] @@ -1828,6 +1899,24 @@ mod tests { assert_eq!(read_csr(&path).unwrap(), csr); } + #[test] + fn csr_row_lookup_is_o1_and_handles_empty_boundary_and_oor() { + let csr = sample_csr(); + let row0 = csr.row(0); + assert_eq!(row0.len(), 2); + assert_eq!(row0.get(0), Some((csr.edge_ids[0], csr.neighbor_ids[0]))); + assert_eq!(row0.get(1), Some((csr.edge_ids[1], csr.neighbor_ids[1]))); + assert!(csr.row(1).is_empty(), "empty interior row"); + assert_eq!(csr.row(2).len(), 2); + assert!(csr.row(3).is_empty(), "out of range"); + assert!(csr.row(u64::MAX).is_empty()); + let empty = CsrIndex { + offsets: vec![0], + ..CsrIndex::default() + }; + assert!(empty.row(0).is_empty()); + } + #[test] fn empty_graph_round_trips_as_offsets_zero() { let dir = TempDir::new().unwrap(); diff --git a/crates/graphforge-storage/src/adjacency_delta.rs b/crates/graphforge-storage/src/adjacency_delta.rs index c613b3e2b..af5a77608 100644 --- a/crates/graphforge-storage/src/adjacency_delta.rs +++ b/crates/graphforge-storage/src/adjacency_delta.rs @@ -15,7 +15,12 @@ //! `edge_id` is a unique surrogate, the result is byte-identical to a full //! rebuild from `topology/` for the same edges — that equivalence is the //! correctness contract (acceptance criterion 1), pinned by the tests below. +//! +//! Execution (#340) prefers [`overlay_delta_segments`]: a borrowable base CSR +//! plus a **bounded** replacement map over only the keys touched by the delta +//! chain. That path never copies the complete valid base CSR. +use std::collections::HashMap; use std::path::{Path, PathBuf}; use std::sync::Arc; @@ -25,7 +30,7 @@ use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder; use graphforge_core::GfError; use crate::adjacency::{ - ALL_RELATIONS_STEM, BuildEntry, CsrIndex, Direction, adjacency_dir, csr_from_entries, + ALL_RELATIONS_STEM, BuildEntry, CsrIndex, CsrRow, Direction, adjacency_dir, csr_from_entries, usable_stem, }; use crate::schemas::ADJACENCY_DELTA_SCHEMA; @@ -259,6 +264,135 @@ pub fn apply_delta_segments( csr_from_entries(&entries, direction) } +/// Bounded delta overlay over a borrowed base CSR (#340). +/// +/// Untouched rows are served directly from `base` (no copy). Only keys that +/// appear in the filtered delta chain allocate replacement rows — size is +/// proportional to overlay cardinality, never to the complete base edge count. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct CsrDeltaOverlay { + /// Validated base CSR; retained by reference-counted ownership. + pub base: Arc, + /// Complete replacement rows for keys touched by the delta chain. + pub replaced: HashMap>, + /// Exclusive upper bound on node ids that may have entries + /// (`max(base.node_count, max_replaced_key + 1)`). + pub node_extent: u64, +} + +impl CsrDeltaOverlay { + /// Checked row lookup: overlay replacement when present, else base CSR row. + #[must_use] + pub fn row(&self, node_id: u64) -> OverlayRow<'_> { + if let Some(entries) = self.replaced.get(&node_id) { + return OverlayRow::Replaced(entries.as_slice()); + } + OverlayRow::Base(self.base.row(node_id)) + } + + /// Number of keys with an allocated replacement row. + #[must_use] + pub fn overlay_row_count(&self) -> u64 { + u64::try_from(self.replaced.len()).unwrap_or(u64::MAX) + } +} + +/// A row resolved through [`CsrDeltaOverlay::row`]. +#[derive(Clone, Copy, Debug)] +pub enum OverlayRow<'a> { + /// Untouched base CSR row. + Base(CsrRow<'a>), + /// Delta-touched replacement (base row ∪ filtered delta entries). + Replaced(&'a [(u64, u64)]), +} + +impl OverlayRow<'_> { + /// Number of `(edge, neighbor)` entries. + #[must_use] + pub fn len(&self) -> usize { + match self { + Self::Base(row) => row.len(), + Self::Replaced(entries) => entries.len(), + } + } + + /// Whether the row is empty. + #[must_use] + pub fn is_empty(&self) -> bool { + self.len() == 0 + } + + /// Entry at `index`, if in range. + #[must_use] + pub fn get(&self, index: usize) -> Option<(u64, u64)> { + match self { + Self::Base(row) => row.get(index), + Self::Replaced(entries) => entries.get(index).copied(), + } + } +} + +/// Build a bounded overlay that preserves the base CSR without copying it. +/// +/// Semantic parity with [`apply_delta_segments`]: every key's row matches the +/// fully rebuilt CSR for that key. Empty chains return an overlay with no +/// replacements (base served as-is). +#[must_use] +pub fn overlay_delta_segments( + base: Arc, + stem: &str, + direction: Direction, + chain: &[DeltaSegment], +) -> CsrDeltaOverlay { + let take_all = stem == ALL_RELATIONS_STEM; + let mut delta_by_key: HashMap> = HashMap::new(); + let mut max_key = 0_u64; + let mut saw_key = false; + for seg in chain { + for e in &seg.edges { + if take_all || (e.rel_type_name == stem && usable_stem(&e.rel_type_name)) { + let (key, neighbor) = match direction { + Direction::Out => (e.src_id, e.dst_id), + Direction::In => (e.dst_id, e.src_id), + }; + delta_by_key + .entry(key) + .or_default() + .push((e.edge_id, neighbor)); + max_key = if saw_key { max_key.max(key) } else { key }; + saw_key = true; + } + } + } + + let mut replaced = HashMap::with_capacity(delta_by_key.len()); + for (key, mut delta_entries) in delta_by_key { + let mut row = Vec::new(); + if key < base.node_count() { + let base_row = base.row(key); + row.reserve(base_row.len() + delta_entries.len()); + row.extend(base_row.iter()); + } else { + row.reserve(delta_entries.len()); + } + row.append(&mut delta_entries); + row.sort_unstable_by_key(|&(edge_id, _)| edge_id); + replaced.insert(key, row); + } + + let node_extent = if !saw_key { + base.node_count() + } else { + max_key.saturating_add(1).max(base.node_count()) + }; + + CsrDeltaOverlay { + base, + replaced, + node_extent, + } +} + /// Reconstruct the `(src, edge, dst)` build entries a CSR for `direction` was /// built from: an `out` CSR is keyed by `src` with `dst` neighbors, an `in` CSR /// by `dst` with `src` neighbors. @@ -355,6 +489,61 @@ mod tests { apply_delta_segments(&base, ALL_RELATIONS_STEM, Direction::Out, &[]), base ); + let overlay = overlay_delta_segments( + Arc::new(base.clone()), + ALL_RELATIONS_STEM, + Direction::Out, + &[], + ); + assert!(overlay.replaced.is_empty()); + assert_eq!(overlay.node_extent, base.node_count()); + assert_eq!(overlay.base.as_ref(), &base); + } + + #[test] + fn overlay_matches_full_rebuild_without_copying_untouched_base_rows() { + let base_edges: Vec = vec![ + (0, 1, 1), + (0, 2, 1), + (1, 3, 2), + (2, 4, 0), + (2, 5, 2), + ]; + let chain = vec![ + seg(8, &[("KNOWS", 6, 1, 3), ("KNOWS", 7, 3, 0)]), + seg(9, &[("OWNS", 8, 0, 2), ("../evil", 9, 3, 1)]), + ]; + for direction in [Direction::Out, Direction::In] { + let base = Arc::new(csr_from_entries(&base_edges, direction)); + let rebuilt = apply_delta_segments(&base, ALL_RELATIONS_STEM, direction, &chain); + let overlay = overlay_delta_segments( + Arc::clone(&base), + ALL_RELATIONS_STEM, + direction, + &chain, + ); + // Overlay keys are only those touched by filtered deltas — never |E|. + assert!(overlay.overlay_row_count() < base.edge_count()); + assert!( + Arc::ptr_eq(&overlay.base, &base), + "overlay must retain the original base CSR allocation" + ); + let extent = rebuilt.node_count().max(overlay.node_extent); + for node in 0..extent { + let expected = rebuilt.row(node); + let got = overlay.row(node); + assert_eq!(got.len(), expected.len(), "node {node} {direction:?}"); + for i in 0..expected.len() { + assert_eq!(got.get(i), expected.get(i), "node {node} idx {i}"); + } + } + // Untouched base key 2 (self-loop + reverse) keeps a Base borrow when + // no delta edge touches key 2 under this direction. + let key_two_touched = overlay.replaced.contains_key(&2); + if !key_two_touched { + assert!(matches!(overlay.row(2), OverlayRow::Base(_))); + } + } } #[test] From 4005466d9a0786a24351095365612df7d37926ab Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Mon, 10 Aug 2026 00:04:11 +0000 Subject: [PATCH 2/3] fix(exec): clippy cleanups and CSR-native docs for #340 Closes #340 Document CSR-native execution honestly in scale-limits and architecture, and apply clippy fixes after the adjacency rewrite. Co-authored-by: David Spencer --- crates/graphforge-exec/src/adjacency.rs | 51 ++++++------------- crates/graphforge-exec/src/algorithm_graph.rs | 14 +++-- crates/graphforge-exec/src/lib.rs | 2 +- .../graphforge-storage/src/adjacency_delta.rs | 23 +++------ docs/book/architecture/execution-model.md | 10 ++++ docs/book/architecture/storage.md | 9 ++-- docs/reference/scale-limits.md | 28 ++++++++++ 7 files changed, 72 insertions(+), 65 deletions(-) diff --git a/crates/graphforge-exec/src/adjacency.rs b/crates/graphforge-exec/src/adjacency.rs index 7d5e3e077..93ffd1742 100644 --- a/crates/graphforge-exec/src/adjacency.rs +++ b/crates/graphforge-exec/src/adjacency.rs @@ -170,6 +170,7 @@ impl<'a> NeighborRow<'a> { } /// Iterate `(edge_id, neighbor_id)` pairs. + #[must_use] pub fn iter(&self) -> NeighborRowIter<'_> { NeighborRowIter { row: self, @@ -281,8 +282,7 @@ impl AdjacencyInner { Self::Map(map) => map.is_empty(), Self::Csr(csr) => csr.edge_count() == 0, Self::Overlay(overlay) => { - overlay.base.edge_count() == 0 - && overlay.replaced.values().all(|row| row.is_empty()) + overlay.base.edge_count() == 0 && overlay.replaced.values().all(Vec::is_empty) } Self::Undirected { out, inbound } => out.is_empty() && inbound.is_empty(), } @@ -300,7 +300,7 @@ impl AdjacencyInner { } }, Self::Undirected { out, inbound } => { - merge_undirected_row(out.neighbors(node_id), inbound.neighbors(node_id)) + merge_undirected_row(&out.neighbors(node_id), &inbound.neighbors(node_id)) } } } @@ -393,7 +393,7 @@ impl PartialEq for Adjacency { impl Eq for Adjacency {} /// Merge out and in rows with ascending `edge_id` and **out before in on ties**. -fn merge_undirected_row<'a>(out: NeighborRow<'a>, inbound: NeighborRow<'a>) -> NeighborRow<'a> { +fn merge_undirected_row<'a>(out: &NeighborRow<'a>, inbound: &NeighborRow<'a>) -> NeighborRow<'a> { if out.is_empty() { return NeighborRow::owned(inbound.to_vec()); } @@ -405,8 +405,7 @@ fn merge_undirected_row<'a>(out: NeighborRow<'a>, inbound: NeighborRow<'a>) -> N let mut i = 0usize; while o < out.len() || i < inbound.len() { let take_out = i >= inbound.len() - || (o < out.len() - && out.get(o).map(|(e, _)| e) <= inbound.get(i).map(|(e, _)| e)); + || (o < out.len() && out.get(o).map(|(e, _)| e) <= inbound.get(i).map(|(e, _)| e)); if take_out { entries.push(out.get(o).expect("out index in range")); o += 1; @@ -923,25 +922,6 @@ impl AdjacencyProvider for PersistentAdjacencyProvider { } } -/// Convert a directed CSR into the traversal view without O(E) HashMap -/// expansion (#340). Empty CSR becomes [`AdjacencyInner::Empty`]. -/// -/// Bounds safety: callers only pass [`CsrIndex`]es produced by -/// `graphforge_storage::adjacency::read_csr`, which validates the invariants -/// (offsets `[0]`-rooted, monotone, final offset == target lengths) before -/// returning. A malformed file errors inside `read_csr`, taking the -/// rebuild/fallback path in the provider, never a panic here. -#[cfg(test)] -fn adjacency_from_csr(index: CsrIndex) -> Adjacency { - if index.edge_count() == 0 { - Adjacency::default() - } else { - Adjacency { - inner: AdjacencyInner::Csr(Arc::new(index)), - } - } -} - /// Undirected CSR pair without materializing a merged hash map (#340). #[cfg(test)] fn merge_undirected(out: CsrIndex, inbound: CsrIndex) -> Adjacency { @@ -1117,10 +1097,7 @@ mod tests { let provider = ScanBuildAdjacencyProvider::new(dir.path().to_path_buf(), OntologyMode::Strict); let view = provider.adjacency("KNOWS", Direction::Out).unwrap(); - assert!( - view.neighbors(ids[1]).is_empty(), - "deleted id yields empty" - ); + assert!(view.neighbors(ids[1]).is_empty(), "deleted id yields empty"); assert_eq!( view.neighbors(ids[2]).to_vec(), vec![(3, ids[3])], @@ -1186,13 +1163,15 @@ mod tests { assert_eq!(undirected.backing(), AdjacencyBacking::CsrUndirected); assert_eq!(undirected.base_csr_entries_expanded(), 0); - let scanned = ScanBuildAdjacencyProvider::new( - dir.path().to_path_buf(), - OntologyMode::Strict, - ) - .adjacency("KNOWS", Direction::Out) - .unwrap(); - assert_eq!(out.as_ref(), scanned.as_ref(), "CSR-native matches scan oracle"); + let scanned = + ScanBuildAdjacencyProvider::new(dir.path().to_path_buf(), OntologyMode::Strict) + .adjacency("KNOWS", Direction::Out) + .unwrap(); + assert_eq!( + out.as_ref(), + scanned.as_ref(), + "CSR-native matches scan oracle" + ); assert!(scanned.base_csr_entries_expanded() > 0); assert_eq!(scanned.backing(), AdjacencyBacking::ScanHashMap); } diff --git a/crates/graphforge-exec/src/algorithm_graph.rs b/crates/graphforge-exec/src/algorithm_graph.rs index 207b70127..c26c19e21 100644 --- a/crates/graphforge-exec/src/algorithm_graph.rs +++ b/crates/graphforge-exec/src/algorithm_graph.rs @@ -129,9 +129,8 @@ fn neighbor_csr_from_sorted( neighbor_edges.push(sorted[index].1); index += 1; } - neighbor_offsets.push( - u32::try_from(neighbor_edges.len()).expect("algorithm neighbor count fits u32"), - ); + neighbor_offsets + .push(u32::try_from(neighbor_edges.len()).expect("algorithm neighbor count fits u32")); row = row.saturating_add(1); } (neighbor_row, neighbor_offsets, neighbor_edges) @@ -692,7 +691,7 @@ pub(crate) fn export_adjacency( let mut raw = Vec::new(); let mut edge_ids = HashSet::new(); for &node_id in &node_ids { - for (edge_id, neighbor_id) in adjacency.neighbors(node_id) { + for (edge_id, neighbor_id) in adjacency.neighbors(node_id).iter() { if selected.contains(&neighbor_id) { raw.push((node_id, edge_id, neighbor_id)); edge_ids.insert(edge_id); @@ -2232,8 +2231,8 @@ mod tests { let uuid1 = u128::from(11_u8).to_be_bytes(); let edge_uuid = u128::from(12_u8).to_be_bytes(); let base = || { - let (neighbor_row, neighbor_offsets, neighbor_edges) = neighbor_csr_from_map( - HashMap::from([( + let (neighbor_row, neighbor_offsets, neighbor_edges) = + neighbor_csr_from_map(HashMap::from([( 0, vec![AlgorithmEdge { edge_id: 0, @@ -2241,8 +2240,7 @@ mod tests { edge_uuid, weight: 2.5, }], - )]), - ); + )])); AdjacencyGraph { directed: true, node_ids: vec![0, 1], diff --git a/crates/graphforge-exec/src/lib.rs b/crates/graphforge-exec/src/lib.rs index 279320bc0..736e0f830 100644 --- a/crates/graphforge-exec/src/lib.rs +++ b/crates/graphforge-exec/src/lib.rs @@ -2722,7 +2722,7 @@ fn bfs_emit( if cfg.max_hops.is_some_and(|m| p.hops >= m) { continue; } - for (edge_id, next) in adjacency.neighbors(p.node) { + for (edge_id, next) in adjacency.neighbors(p.node).iter() { if p.visited_edges.contains(&edge_id) { continue; // relationship isomorphism: no edge twice per path } diff --git a/crates/graphforge-storage/src/adjacency_delta.rs b/crates/graphforge-storage/src/adjacency_delta.rs index af5a77608..6c7032350 100644 --- a/crates/graphforge-storage/src/adjacency_delta.rs +++ b/crates/graphforge-storage/src/adjacency_delta.rs @@ -380,10 +380,10 @@ pub fn overlay_delta_segments( replaced.insert(key, row); } - let node_extent = if !saw_key { - base.node_count() - } else { + let node_extent = if saw_key { max_key.saturating_add(1).max(base.node_count()) + } else { + base.node_count() }; CsrDeltaOverlay { @@ -502,13 +502,8 @@ mod tests { #[test] fn overlay_matches_full_rebuild_without_copying_untouched_base_rows() { - let base_edges: Vec = vec![ - (0, 1, 1), - (0, 2, 1), - (1, 3, 2), - (2, 4, 0), - (2, 5, 2), - ]; + let base_edges: Vec = + vec![(0, 1, 1), (0, 2, 1), (1, 3, 2), (2, 4, 0), (2, 5, 2)]; let chain = vec![ seg(8, &[("KNOWS", 6, 1, 3), ("KNOWS", 7, 3, 0)]), seg(9, &[("OWNS", 8, 0, 2), ("../evil", 9, 3, 1)]), @@ -516,12 +511,8 @@ mod tests { for direction in [Direction::Out, Direction::In] { let base = Arc::new(csr_from_entries(&base_edges, direction)); let rebuilt = apply_delta_segments(&base, ALL_RELATIONS_STEM, direction, &chain); - let overlay = overlay_delta_segments( - Arc::clone(&base), - ALL_RELATIONS_STEM, - direction, - &chain, - ); + let overlay = + overlay_delta_segments(Arc::clone(&base), ALL_RELATIONS_STEM, direction, &chain); // Overlay keys are only those touched by filtered deltas — never |E|. assert!(overlay.overlay_row_count() < base.edge_count()); assert!( diff --git a/docs/book/architecture/execution-model.md b/docs/book/architecture/execution-model.md index 784a9ee21..25b4d778c 100644 --- a/docs/book/architecture/execution-model.md +++ b/docs/book/architecture/execution-model.md @@ -124,6 +124,16 @@ falls back to scan-and-build with identical output. The pinned v0.5 project-generation UUID and graph source fingerprint detect staleness; the committed Parquet graph participant is always the source of truth. +Fresh index hits serve a **CSR-native** adjacency view (#340): validated +offsets with parallel edge/neighbor columns and O(1) row lookup. Undirected +requests keep separate out/in CSRs and merge per accessed row (out before in +on equal `edge_id`) without materializing a full merged hash map. Delta +overlays attach a bounded replacement map over touched keys only — they do +not copy the complete valid base CSR. Scan-built fallback retains the +historical hash-map representation for oracle parity. Analyst +`export_adjacency` projects selected nodes into a flat CSR of algorithm +edges rather than duplicating the full graph into per-node heap vectors. + `explain` annotates each traversal node with `adjacency=hit | miss | building`, so plan inspection shows whether the accelerator was used (`ExecutionSession::explain_physical` renders the physical plan without executing it). One `PersistentAdjacencyProvider` lives per diff --git a/docs/book/architecture/storage.md b/docs/book/architecture/storage.md index 9a70c2e07..66a912da7 100644 --- a/docs/book/architecture/storage.md +++ b/docs/book/architecture/storage.md @@ -252,10 +252,11 @@ Conventions: - **Node with no neighbors**: an empty list (`offsets[i] == offsets[i+1]`). - CSR rows cover exactly `node_id ∈ 0..node_count`; surrogates beyond `node_count` simply have no entries. -- In-memory consumers (`graphforge_exec::AdjacencyProvider` — scan-build via - `ScanBuildAdjacencyProvider` until the persistent loader) see the logical - `offsets`/`targets` model; the list encoding is a file-format detail - (`graphforge_storage::adjacency::CsrIndex` on the storage side). +- In-memory consumers (`graphforge_exec::AdjacencyProvider`) keep the logical + `offsets` / parallel `edge_ids`+`neighbor_ids` model on a persisted hit + (#340 CSR-native views); the list encoding remains a file-format detail + (`graphforge_storage::adjacency::CsrIndex`). Scan-build fallback still + materializes a hash map for oracle parity. ### Rebuild and versioning semantics diff --git a/docs/reference/scale-limits.md b/docs/reference/scale-limits.md index 7bf2dcc2f..6bcf5a02a 100644 --- a/docs/reference/scale-limits.md +++ b/docs/reference/scale-limits.md @@ -140,6 +140,34 @@ boundary as a GraphForge maximum graph size. Manual/scheduled 8M/128M reproduction (not CI): build or point at the measured fixture, publish through `GraphForge`, reopen, and record RSS/storage/fingerprint via the #334 evidence emitter. + +## CSR-native execution (#340) + +Persisted-index **hits** no longer expand the validated base CSR into +`HashMap>` for traversal or analyst +projection. Execution keeps: + +- directed CSR with checked O(1) row lookup over offsets + parallel + edge/neighbor columns; +- undirected views as an out+in CSR pair merged **per accessed row** + (out-before-in on equal `edge_id`), without a full merged hash map; +- delta overlays as a bounded replacement map over only keys touched by the + delta chain — the complete valid base CSR is retained, not recopied. + +Scan-build / missing / stale / corrupt index paths still use the historical +hash-map oracle (or rebuild then serve CSR-native). Structural counters on +`Adjacency` (`backing()`, `base_csr_entries_expanded()`, `overlay_row_count()`) +assert zero base-CSR expansion on a fresh hit. Analyst export builds a +selection-bounded flat CSR of `AlgorithmEdge` entries rather than per-node +heap vectors for every graph edge. + +| Claim | Status | +|---|---| +| Fresh index hit: no O(E) HashMap / per-node Vec expansion | Covered by unit structural counter + parity vs scan | +| Out / in / undirected / typed / wildcard semantics preserved | Covered by adjacency + persistent provider tests | +| Bounded delta overlay without full base copy | Covered by storage overlay parity tests | +| Selected-subgraph projection bounded by selection | Covered by export path iterating selected node ids | +| Peak RSS / cold-warm first-use on #334 fixtures | Pending M4 scale evidence (timing remains hardware-specific) | --- ## Why Edge Count, Not Node Count From 017ec94d4f2e3fe073a10d40f0b87e4aa1d9320b Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Mon, 10 Aug 2026 01:30:27 +0000 Subject: [PATCH 3/3] ci: retrigger Bazel Bootstrap after local storage test pass Local `bazelisk test //crates/graphforge-storage:graphforge_storage_test` passed; prior CI WriterBusy failure looks load-flake under parallel suite. Co-authored-by: David Spencer