1use ahash::HashMap;
4use mentedb_core::edge::{EdgeType, MemoryEdge};
5use mentedb_core::error::{MenteError, MenteResult};
6use mentedb_core::types::{MemoryId, Timestamp};
7use serde::{Deserialize, Serialize};
8
9#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
11pub struct StoredEdge {
12 pub edge_type: EdgeType,
14 pub weight: f32,
16 pub created_at: Timestamp,
18 #[serde(default, skip_serializing_if = "Option::is_none")]
20 pub valid_from: Option<Timestamp>,
21 #[serde(default, skip_serializing_if = "Option::is_none")]
23 pub valid_until: Option<Timestamp>,
24 #[serde(default, skip_serializing_if = "Option::is_none")]
26 pub label: Option<String>,
27}
28
29impl StoredEdge {
30 pub fn from_memory_edge(edge: &MemoryEdge) -> Self {
32 Self {
33 edge_type: edge.edge_type,
34 weight: edge.weight,
35 created_at: edge.created_at,
36 valid_from: edge.valid_from,
37 valid_until: edge.valid_until,
38 label: edge.label.clone(),
39 }
40 }
41
42 pub fn is_valid_at(&self, at: Timestamp) -> bool {
44 let from = self.valid_from.unwrap_or(0);
45 match self.valid_until {
46 Some(until) => at >= from && at < until,
47 None => at >= from,
48 }
49 }
50
51 pub fn is_invalidated(&self) -> bool {
53 self.valid_until.is_some()
54 }
55}
56
57#[derive(Debug, Clone, Serialize, Deserialize)]
59struct DeltaEdge {
60 source_idx: u32,
61 target_idx: u32,
62 data: StoredEdge,
63}
64
65#[derive(Debug, Clone, Default, Serialize, Deserialize)]
67struct CompressedStorage {
68 row_offsets: Vec<u32>,
70 col_indices: Vec<u32>,
72 edge_data: Vec<StoredEdge>,
74}
75
76impl CompressedStorage {
77 #[allow(dead_code)]
78 fn new(num_nodes: usize) -> Self {
79 Self {
80 row_offsets: vec![0; num_nodes + 1],
81 col_indices: Vec::new(),
82 edge_data: Vec::new(),
83 }
84 }
85
86 fn neighbors(&self, row: u32) -> &[u32] {
88 let row = row as usize;
89 if row + 1 >= self.row_offsets.len() {
90 return &[];
91 }
92 let start = self.row_offsets[row] as usize;
93 let end = self.row_offsets[row + 1] as usize;
94 &self.col_indices[start..end]
95 }
96
97 fn edge_data_for(&self, row: u32) -> &[StoredEdge] {
98 let row = row as usize;
99 if row + 1 >= self.row_offsets.len() {
100 return &[];
101 }
102 let start = self.row_offsets[row] as usize;
103 let end = self.row_offsets[row + 1] as usize;
104 &self.edge_data[start..end]
105 }
106}
107
108#[derive(Debug, Clone, Serialize, Deserialize)]
110pub struct CsrGraph {
111 id_to_idx: HashMap<MemoryId, u32>,
113 idx_to_id: Vec<MemoryId>,
115
116 csr: CompressedStorage,
118 csc: CompressedStorage,
120
121 delta_edges: Vec<DeltaEdge>,
123 removed_edges: Vec<(u32, u32)>,
125}
126
127impl CsrGraph {
128 pub fn new() -> Self {
130 Self {
131 id_to_idx: HashMap::default(),
132 idx_to_id: Vec::new(),
133 csr: CompressedStorage::default(),
134 csc: CompressedStorage::default(),
135 delta_edges: Vec::new(),
136 removed_edges: Vec::new(),
137 }
138 }
139
140 pub fn add_node(&mut self, id: MemoryId) -> u32 {
142 if let Some(&idx) = self.id_to_idx.get(&id) {
143 return idx;
144 }
145 let idx = self.idx_to_id.len() as u32;
146 self.id_to_idx.insert(id, idx);
147 self.idx_to_id.push(id);
148 idx
149 }
150
151 pub fn remove_node(&mut self, id: MemoryId) {
153 let Some(&idx) = self.id_to_idx.get(&id) else {
154 return;
155 };
156 for &neighbor in self.csr.neighbors(idx) {
158 self.removed_edges.push((idx, neighbor));
159 }
160 for &neighbor in self.csc.neighbors(idx) {
161 self.removed_edges.push((neighbor, idx));
162 }
163 self.delta_edges
165 .retain(|e| e.source_idx != idx && e.target_idx != idx);
166 self.id_to_idx.remove(&id);
167 }
168
169 pub fn add_edge(&mut self, edge: &MemoryEdge) {
176 let source_idx = self.add_node(edge.source);
177 let target_idx = self.add_node(edge.target);
178
179 let duplicate = self.outgoing_by_idx(source_idx).into_iter().any(|(t, e)| {
180 t == edge.target && e.edge_type == edge.edge_type && e.valid_until.is_none()
181 });
182 if duplicate {
183 return;
184 }
185
186 self.delta_edges.push(DeltaEdge {
187 source_idx,
188 target_idx,
189 data: StoredEdge::from_memory_edge(edge),
190 });
191 }
192
193 pub fn strengthen_edge(&mut self, source: MemoryId, target: MemoryId, delta: f32) {
200 let (Some(&source_idx), Some(&target_idx)) =
201 (self.id_to_idx.get(&source), self.id_to_idx.get(&target))
202 else {
203 return;
204 };
205
206 if let Some(existing) = self
208 .delta_edges
209 .iter_mut()
210 .find(|e| e.source_idx == source_idx && e.target_idx == target_idx)
211 {
212 existing.data.weight = (existing.data.weight + delta).min(1.0);
213 return;
214 }
215
216 if let Some((_, stored)) = self
220 .outgoing_by_idx(source_idx)
221 .into_iter()
222 .find(|(id, _)| *id == target)
223 {
224 let new_weight = (stored.weight + delta).min(1.0);
225 self.removed_edges.push((source_idx, target_idx));
226 self.delta_edges.push(DeltaEdge {
227 source_idx,
228 target_idx,
229 data: StoredEdge {
230 weight: new_weight,
231 ..stored
232 },
233 });
234 }
235 }
236
237 pub fn remove_edge(&mut self, source: MemoryId, target: MemoryId) {
239 let (Some(&src_idx), Some(&tgt_idx)) =
240 (self.id_to_idx.get(&source), self.id_to_idx.get(&target))
241 else {
242 return;
243 };
244 self.removed_edges.push((src_idx, tgt_idx));
245 self.delta_edges
246 .retain(|e| !(e.source_idx == src_idx && e.target_idx == tgt_idx));
247 }
248
249 pub fn remove_edges_of_types(&mut self, types: &[EdgeType]) -> usize {
260 let matches = |et: EdgeType| types.contains(&et);
261 let mut removed = 0usize;
262
263 let before = self.delta_edges.len();
265 self.delta_edges.retain(|e| !matches(e.data.edge_type));
266 removed += before - self.delta_edges.len();
267
268 let mut restore: Vec<DeltaEdge> = Vec::new();
271 let n = self.idx_to_id.len() as u32;
272 for src_idx in 0..n {
273 let row: Vec<(u32, StoredEdge)> = {
274 let neighbors = self.csr.neighbors(src_idx);
275 let edges = self.csr.edge_data_for(src_idx);
276 let mut v = Vec::new();
277 for (i, &t) in neighbors.iter().enumerate() {
278 if !self.is_removed(src_idx, t) {
279 v.push((t, edges[i].clone()));
280 }
281 }
282 v
283 };
284 let matched_pairs: std::collections::HashSet<u32> = row
285 .iter()
286 .filter(|(_, e)| matches(e.edge_type))
287 .map(|(t, _)| *t)
288 .collect();
289 if matched_pairs.is_empty() {
290 continue;
291 }
292 for &tgt_idx in &matched_pairs {
293 self.removed_edges.push((src_idx, tgt_idx));
294 }
295 for (tgt_idx, e) in row {
296 if !matched_pairs.contains(&tgt_idx) {
297 continue;
298 }
299 if matches(e.edge_type) {
300 removed += 1;
301 } else {
302 restore.push(DeltaEdge {
303 source_idx: src_idx,
304 target_idx: tgt_idx,
305 data: e,
306 });
307 }
308 }
309 }
310 self.delta_edges.extend(restore);
311 removed
312 }
313
314 pub fn outgoing(&self, id: MemoryId) -> Vec<(MemoryId, StoredEdge)> {
316 let Some(&idx) = self.id_to_idx.get(&id) else {
317 return Vec::new();
318 };
319 self.outgoing_by_idx(idx)
320 }
321
322 pub fn outgoing_valid_at(&self, id: MemoryId, at: Timestamp) -> Vec<(MemoryId, StoredEdge)> {
324 self.outgoing(id)
325 .into_iter()
326 .filter(|(_, e)| e.is_valid_at(at))
327 .collect()
328 }
329
330 pub(crate) fn outgoing_by_idx(&self, idx: u32) -> Vec<(MemoryId, StoredEdge)> {
331 let mut results = Vec::new();
332
333 let neighbors = self.csr.neighbors(idx);
335 let edges = self.csr.edge_data_for(idx);
336 for (i, &neighbor) in neighbors.iter().enumerate() {
337 if !self.is_removed(idx, neighbor)
338 && let Some(&id) = self.idx_to_id.get(neighbor as usize)
339 {
340 results.push((id, edges[i].clone()));
341 }
342 }
343
344 for delta in &self.delta_edges {
346 if delta.source_idx == idx
347 && let Some(&id) = self.idx_to_id.get(delta.target_idx as usize)
348 {
349 results.push((id, delta.data.clone()));
350 }
351 }
352
353 results
354 }
355
356 pub fn conflict_edge_counts(&self) -> (usize, usize) {
363 let norm = |a: u32, b: u32| if a <= b { (a, b) } else { (b, a) };
364 let mut contradicts: std::collections::HashSet<(u32, u32)> =
365 std::collections::HashSet::new();
366 let mut supersedes = 0usize;
367
368 let node_count = self.idx_to_id.len() as u32;
371 for idx in 0..node_count {
372 let neighbors = self.csr.neighbors(idx);
373 let data = self.csr.edge_data_for(idx);
374 for (i, &nbr) in neighbors.iter().enumerate() {
375 if self.is_removed(idx, nbr) {
376 continue;
377 }
378 match data[i].edge_type {
379 EdgeType::Contradicts => {
380 contradicts.insert(norm(idx, nbr));
381 }
382 EdgeType::Supersedes => supersedes += 1,
383 _ => {}
384 }
385 }
386 }
387
388 for d in &self.delta_edges {
390 if self.is_removed(d.source_idx, d.target_idx) {
391 continue;
392 }
393 match d.data.edge_type {
394 EdgeType::Contradicts => {
395 contradicts.insert(norm(d.source_idx, d.target_idx));
396 }
397 EdgeType::Supersedes => supersedes += 1,
398 _ => {}
399 }
400 }
401
402 (contradicts.len(), supersedes)
403 }
404
405 pub fn incoming(&self, id: MemoryId) -> Vec<(MemoryId, StoredEdge)> {
407 let Some(&idx) = self.id_to_idx.get(&id) else {
408 return Vec::new();
409 };
410 self.incoming_by_idx(idx)
411 }
412
413 pub fn incoming_valid_at(&self, id: MemoryId, at: Timestamp) -> Vec<(MemoryId, StoredEdge)> {
415 self.incoming(id)
416 .into_iter()
417 .filter(|(_, e)| e.is_valid_at(at))
418 .collect()
419 }
420
421 pub(crate) fn incoming_by_idx(&self, idx: u32) -> Vec<(MemoryId, StoredEdge)> {
422 let mut results = Vec::new();
423
424 let neighbors = self.csc.neighbors(idx);
426 let edges = self.csc.edge_data_for(idx);
427 for (i, &neighbor) in neighbors.iter().enumerate() {
428 if !self.is_removed(neighbor, idx)
429 && let Some(&id) = self.idx_to_id.get(neighbor as usize)
430 {
431 results.push((id, edges[i].clone()));
432 }
433 }
434
435 for delta in &self.delta_edges {
437 if delta.target_idx == idx
438 && let Some(&id) = self.idx_to_id.get(delta.source_idx as usize)
439 {
440 results.push((id, delta.data.clone()));
441 }
442 }
443
444 results
445 }
446
447 pub fn contains_node(&self, id: MemoryId) -> bool {
449 self.id_to_idx.contains_key(&id)
450 }
451
452 pub fn node_count(&self) -> usize {
454 self.idx_to_id.len()
455 }
456
457 pub(crate) fn get_idx(&self, id: MemoryId) -> Option<u32> {
459 self.id_to_idx.get(&id).copied()
460 }
461
462 #[allow(dead_code)]
464 pub(crate) fn get_id(&self, idx: u32) -> Option<MemoryId> {
465 self.idx_to_id.get(idx as usize).copied()
466 }
467
468 pub fn node_ids(&self) -> &[MemoryId] {
470 &self.idx_to_id
471 }
472
473 fn is_removed(&self, source: u32, target: u32) -> bool {
474 self.removed_edges
475 .iter()
476 .any(|&(s, t)| s == source && t == target)
477 }
478
479 pub fn compact(&mut self) {
481 let num_nodes = self.idx_to_id.len();
482
483 let mut all_edges: Vec<(u32, u32, StoredEdge)> = Vec::new();
485
486 for row in 0..num_nodes {
488 let row = row as u32;
489 let neighbors = self.csr.neighbors(row);
490 let edges = self.csr.edge_data_for(row);
491 for (i, &col) in neighbors.iter().enumerate() {
492 if !self.is_removed(row, col) {
493 all_edges.push((row, col, edges[i].clone()));
494 }
495 }
496 }
497
498 for delta in &self.delta_edges {
500 all_edges.push((delta.source_idx, delta.target_idx, delta.data.clone()));
501 }
502
503 self.csr = Self::build_compressed(&all_edges, num_nodes, false);
505
506 self.csc = Self::build_compressed(&all_edges, num_nodes, true);
508
509 self.delta_edges.clear();
510 self.removed_edges.clear();
511 }
512
513 fn build_compressed(
514 edges: &[(u32, u32, StoredEdge)],
515 num_nodes: usize,
516 transpose: bool,
517 ) -> CompressedStorage {
518 let mut counts = vec![0u32; num_nodes];
520 for &(src, tgt, ref _data) in edges {
521 let row = if transpose { tgt } else { src };
522 if (row as usize) < num_nodes {
523 counts[row as usize] += 1;
524 }
525 }
526
527 let mut row_offsets = vec![0u32; num_nodes + 1];
529 for i in 0..num_nodes {
530 row_offsets[i + 1] = row_offsets[i] + counts[i];
531 }
532
533 let total = row_offsets[num_nodes] as usize;
534 let mut col_indices = vec![0u32; total];
535 let mut edge_data = vec![
536 StoredEdge {
537 edge_type: EdgeType::Related,
538 weight: 0.0,
539 created_at: 0,
540 valid_from: None,
541 valid_until: None,
542 label: None,
543 };
544 total
545 ];
546
547 let mut cursors = row_offsets[..num_nodes].to_vec();
549 for &(src, tgt, ref data) in edges {
550 let (row, col) = if transpose { (tgt, src) } else { (src, tgt) };
551 if (row as usize) < num_nodes {
552 let pos = cursors[row as usize] as usize;
553 col_indices[pos] = col;
554 edge_data[pos] = data.clone();
555 cursors[row as usize] += 1;
556 }
557 }
558
559 CompressedStorage {
560 row_offsets,
561 col_indices,
562 edge_data,
563 }
564 }
565 pub fn save(&self, path: &std::path::Path) -> MenteResult<()> {
567 let data =
568 serde_json::to_vec(self).map_err(|e| MenteError::Serialization(e.to_string()))?;
569 let tmp_path = path.with_extension("json.tmp");
572 {
573 use std::io::Write;
574 let mut file = std::fs::File::create(&tmp_path)?;
575 file.write_all(&data)?;
576 file.sync_data()?;
577 }
578 std::fs::rename(&tmp_path, path)?;
579 Ok(())
580 }
581
582 pub fn load(path: &std::path::Path) -> MenteResult<Self> {
584 let data = std::fs::read(path)?;
585 let graph: Self =
586 serde_json::from_slice(&data).map_err(|e| MenteError::Serialization(e.to_string()))?;
587 Ok(graph)
588 }
589}
590
591impl Default for CsrGraph {
592 fn default() -> Self {
593 Self::new()
594 }
595}
596
597#[cfg(test)]
598mod tests {
599 use super::*;
600
601 fn make_edge(src: MemoryId, tgt: MemoryId, etype: EdgeType) -> MemoryEdge {
602 MemoryEdge {
603 source: src,
604 target: tgt,
605 edge_type: etype,
606 weight: 0.8,
607 created_at: 1000,
608 valid_from: None,
609 valid_until: None,
610 label: None,
611 }
612 }
613
614 #[test]
615 fn test_add_node_idempotent() {
616 let mut g = CsrGraph::new();
617 let id = MemoryId::new();
618 let idx1 = g.add_node(id);
619 let idx2 = g.add_node(id);
620 assert_eq!(idx1, idx2);
621 assert_eq!(g.node_count(), 1);
622 }
623
624 #[test]
625 fn conflict_edge_counts_span_compressed_and_delta_and_dedup() {
626 let mut g = CsrGraph::new();
627 let (a, b, c, d) = (
628 MemoryId::new(),
629 MemoryId::new(),
630 MemoryId::new(),
631 MemoryId::new(),
632 );
633
634 g.add_edge(&make_edge(a, b, EdgeType::Contradicts));
637 g.add_edge(&make_edge(a, c, EdgeType::Related));
638 g.compact();
639
640 g.add_edge(&make_edge(d, c, EdgeType::Supersedes));
644 g.add_edge(&make_edge(b, a, EdgeType::Contradicts));
645
646 let (contradicts, supersedes) = g.conflict_edge_counts();
647 assert_eq!(contradicts, 1, "A<->B is one undirected contradiction");
648 assert_eq!(supersedes, 1);
649 }
650
651 #[test]
652 fn test_add_and_query_edges() {
653 let mut g = CsrGraph::new();
654 let a = MemoryId::new();
655 let b = MemoryId::new();
656 let c = MemoryId::new();
657
658 g.add_edge(&make_edge(a, b, EdgeType::Caused));
659 g.add_edge(&make_edge(a, c, EdgeType::Related));
660
661 let out = g.outgoing(a);
662 assert_eq!(out.len(), 2);
663
664 let inc_b = g.incoming(b);
665 assert_eq!(inc_b.len(), 1);
666 assert_eq!(inc_b[0].0, a);
667 }
668
669 #[test]
670 fn test_remove_edge() {
671 let mut g = CsrGraph::new();
672 let a = MemoryId::new();
673 let b = MemoryId::new();
674
675 g.add_edge(&make_edge(a, b, EdgeType::Caused));
676 assert_eq!(g.outgoing(a).len(), 1);
677
678 g.remove_edge(a, b);
679 assert_eq!(g.outgoing(a).len(), 0);
680 }
681
682 #[test]
683 fn test_remove_edges_of_types_preserves_siblings() {
684 let mut g = CsrGraph::new();
685 let a = MemoryId::new();
686 let b = MemoryId::new();
687 let c = MemoryId::new();
688
689 g.add_edge(&make_edge(a, b, EdgeType::Supersedes));
692 g.add_edge(&make_edge(a, b, EdgeType::Related));
693 g.add_edge(&make_edge(a, c, EdgeType::Contradicts));
695 g.add_edge(&make_edge(b, c, EdgeType::Caused));
697
698 g.compact();
701
702 let removed = g.remove_edges_of_types(&[EdgeType::Contradicts, EdgeType::Supersedes]);
703 g.compact();
704 assert_eq!(removed, 2, "one Supersedes + one Contradicts");
705
706 let ab: Vec<_> = g.outgoing(a).into_iter().filter(|(t, _)| *t == b).collect();
708 assert_eq!(ab.len(), 1);
709 assert_eq!(ab[0].1.edge_type, EdgeType::Related);
710
711 assert!(g.outgoing(a).into_iter().all(|(t, _)| t != c));
713
714 assert_eq!(g.outgoing(b).len(), 1);
716 let inc_c = g.incoming(c);
717 assert_eq!(inc_c.len(), 1);
718 assert_eq!(inc_c[0].0, b);
719 }
720
721 #[test]
722 fn test_compact() {
723 let mut g = CsrGraph::new();
724 let a = MemoryId::new();
725 let b = MemoryId::new();
726 let c = MemoryId::new();
727
728 g.add_edge(&make_edge(a, b, EdgeType::Caused));
729 g.add_edge(&make_edge(b, c, EdgeType::Before));
730 g.compact();
731
732 let out_a = g.outgoing(a);
733 assert_eq!(out_a.len(), 1);
734 assert_eq!(out_a[0].0, b);
735
736 let inc_c = g.incoming(c);
737 assert_eq!(inc_c.len(), 1);
738 assert_eq!(inc_c[0].0, b);
739 }
740
741 #[test]
742 fn test_compact_with_removals() {
743 let mut g = CsrGraph::new();
744 let a = MemoryId::new();
745 let b = MemoryId::new();
746 let c = MemoryId::new();
747
748 g.add_edge(&make_edge(a, b, EdgeType::Caused));
749 g.add_edge(&make_edge(a, c, EdgeType::Related));
750 g.compact();
751
752 g.remove_edge(a, b);
753 g.compact();
754
755 let out = g.outgoing(a);
756 assert_eq!(out.len(), 1);
757 assert_eq!(out[0].0, c);
758 }
759
760 #[test]
761 fn test_remove_node_cleans_id_to_idx() {
762 let mut g = CsrGraph::new();
763 let a = MemoryId::new();
764 let b = MemoryId::new();
765
766 g.add_edge(&make_edge(a, b, EdgeType::Caused));
767 assert!(g.contains_node(a));
768 assert!(g.contains_node(b));
769
770 g.remove_node(a);
771 assert!(
772 !g.contains_node(a),
773 "removed node should not be in id_to_idx"
774 );
775 assert!(g.contains_node(b), "unrelated node should still exist");
776
777 assert!(g.outgoing(a).is_empty());
779 assert!(g.incoming(b).is_empty());
780 }
781
782 #[test]
783 fn test_remove_node_then_readd() {
784 let mut g = CsrGraph::new();
785 let a = MemoryId::new();
786 let b = MemoryId::new();
787 let c = MemoryId::new();
788
789 g.add_edge(&make_edge(a, b, EdgeType::Caused));
790 g.remove_node(a);
791
792 g.add_edge(&make_edge(a, c, EdgeType::Related));
794 assert!(g.contains_node(a));
795 let out = g.outgoing(a);
796 assert_eq!(out.len(), 1);
797 assert_eq!(out[0].0, c);
798 }
799}