velesdb_core/collection/graph_collection.rs
1//! `GraphCollection`: knowledge graph with optional node embeddings.
2//!
3//! # Design
4//!
5//! `GraphCollection` is a pure newtype over `Collection` (C-02).
6//! All graph state (edge store, property/range indexes, node payloads, optional
7//! HNSW for node embeddings) lives inside the single `inner: Collection`.
8//! The graph schema and embedding dimension are persisted in `config.json`.
9//! There are no separate engine fields — no dual-storage risk.
10
11use std::path::PathBuf;
12
13use crate::collection::graph::{
14 EdgeRemoval, GraphEdge, GraphSchema, TraversalConfig, TraversalResult,
15};
16use crate::collection::types::Collection;
17use crate::distance::DistanceMetric;
18use crate::error::Result;
19use crate::point::{Point, SearchResult};
20
21/// A graph collection storing typed relationships between nodes.
22///
23/// Node embeddings are optional: if `dimension` is `None`, no vector index is created.
24///
25/// # Examples
26///
27/// ```rust,no_run
28/// use velesdb_core::{GraphCollection, GraphSchema, GraphEdge, DistanceMetric};
29///
30/// let coll = GraphCollection::create(
31/// "./data/kg".into(),
32/// "knowledge",
33/// None, // no embeddings
34/// DistanceMetric::Cosine, // unused when no embeddings
35/// GraphSchema::schemaless(),
36/// )?;
37///
38/// let edge = GraphEdge::new(1, 100, 200, "KNOWS")?;
39/// coll.add_edge(edge)?;
40/// # Ok::<(), velesdb_core::Error>(())
41/// ```
42#[derive(Clone)]
43pub struct GraphCollection {
44 /// Single source of truth — all graph state lives here (C-02 pure newtype).
45 pub(crate) inner: Collection,
46}
47
48impl GraphCollection {
49 // -------------------------------------------------------------------------
50 // Lifecycle
51 // -------------------------------------------------------------------------
52
53 /// Creates a new `GraphCollection`.
54 ///
55 /// # Errors
56 ///
57 /// Returns an error if the directory cannot be created or storage fails.
58 pub fn create(
59 path: PathBuf,
60 name: &str,
61 dimension: Option<usize>,
62 metric: DistanceMetric,
63 schema: GraphSchema,
64 ) -> Result<Self> {
65 Ok(Self {
66 inner: Collection::create_graph_collection(path, name, schema, dimension, metric)?,
67 })
68 }
69
70 /// Opens an existing `GraphCollection` from disk.
71 ///
72 /// # Errors
73 ///
74 /// Returns an error if config or storage cannot be opened.
75 pub fn open(path: PathBuf) -> Result<Self> {
76 Ok(Self {
77 inner: Collection::open(path)?,
78 })
79 }
80
81 /// Consumes `self` and returns a [`VectorCollection`](super::VectorCollection)
82 /// **structural view** over this graph collection's shared `inner` store.
83 ///
84 /// This is a purely structural re-wrap of the identical `inner: Collection`
85 /// backing store into the `VectorCollection` newtype — an ordinary value
86 /// move, **not** a `transmute` and not memory-unsafe. It does **not** assert
87 /// that the collection is vector-kind: invoking vector-specific methods on
88 /// the result returns empty or misleading state.
89 ///
90 /// It exists solely for the **Python binding**, whose single user-facing
91 /// `Collection` type is backed by a `VectorCollection`: the binding holds a
92 /// graph collection behind that type while gating vector-only operations on
93 /// the real kind it tracks separately (the Python
94 /// `Collection::ensure_vector` guard). The Mobile and Tauri bindings do
95 /// *not* use this view — they go through the variant-checked
96 /// [`AnyCollection::into_vector`](super::AnyCollection::into_vector) and
97 /// reject non-vector collections. Callers that need the graph surface must
98 /// use the graph API, not this view.
99 ///
100 /// [`MetadataCollection::into_vector_view`](super::MetadataCollection::into_vector_view)
101 /// is the exact mirror for metadata collections.
102 #[must_use]
103 pub fn into_vector_view(self) -> super::VectorCollection {
104 super::VectorCollection { inner: self.inner }
105 }
106
107 /// Flushes all state to disk.
108 ///
109 /// Issue #423: This fast-path flush skips `vectors.idx` serialization.
110 /// The WAL provides crash recovery for the vector index.
111 ///
112 /// # Errors
113 ///
114 /// Returns an error if any flush operation fails.
115 pub fn flush(&self) -> Result<()> {
116 self.inner.flush()
117 }
118
119 /// Full durability flush including `vectors.idx` serialization.
120 ///
121 /// Issue #423: Use on graceful shutdown to avoid a full WAL replay
122 /// on the next startup.
123 ///
124 /// # Errors
125 ///
126 /// Returns an error if any flush operation fails.
127 pub fn flush_full(&self) -> Result<()> {
128 self.inner.flush_full()
129 }
130
131 // -------------------------------------------------------------------------
132 // Metadata
133 // -------------------------------------------------------------------------
134
135 /// Returns the collection name.
136 #[must_use]
137 pub fn name(&self) -> String {
138 self.inner.config().name
139 }
140
141 /// Returns the graph schema stored in config.
142 ///
143 /// Returns `GraphSchema::schemaless()` for collections that have no schema set.
144 #[must_use]
145 pub fn schema(&self) -> GraphSchema {
146 self.inner
147 .graph_schema()
148 .unwrap_or_else(GraphSchema::schemaless)
149 }
150
151 /// Returns `true` if this collection stores node embeddings.
152 #[must_use]
153 pub fn has_embeddings(&self) -> bool {
154 self.inner.has_embeddings()
155 }
156
157 // -------------------------------------------------------------------------
158 // Graph operations — delegate to Collection graph API
159 // -------------------------------------------------------------------------
160
161 /// Adds an edge between two nodes.
162 ///
163 /// # Errors
164 ///
165 /// - Returns `Error::EdgeExists` if an edge with the same ID already exists.
166 ///
167 /// # Examples
168 ///
169 /// ```rust,no_run
170 /// # use velesdb_core::{GraphCollection, GraphSchema, GraphEdge, DistanceMetric};
171 /// # let coll = GraphCollection::create("./data/kg".into(), "kg", None, DistanceMetric::Cosine, GraphSchema::schemaless())?;
172 /// let edge = GraphEdge::new(1, 100, 200, "KNOWS")?;
173 /// coll.add_edge(edge)?;
174 /// # Ok::<(), velesdb_core::Error>(())
175 /// ```
176 pub fn add_edge(&self, edge: GraphEdge) -> Result<()> {
177 self.inner.add_edge(edge)
178 }
179
180 /// Adds multiple edges in batch (much faster than calling add_edge in a loop).
181 ///
182 /// Acquires locks once for the entire batch and rebuilds the CSR snapshot
183 /// once at the end. Duplicate edge IDs are silently skipped.
184 ///
185 /// # Returns
186 ///
187 /// Number of edges successfully added.
188 ///
189 /// # Errors
190 ///
191 /// Returns an error if WAL durability logging fails for graph
192 /// collections (fail-closed: the in-memory store is not mutated).
193 pub fn add_edges_batch(&self, edges: Vec<GraphEdge>) -> Result<usize> {
194 self.inner.add_edges_batch(edges)
195 }
196
197 /// Returns edges, optionally filtered by label.
198 #[must_use]
199 pub fn get_edges(&self, label: Option<&str>) -> Vec<GraphEdge> {
200 match label {
201 Some(lbl) => self.inner.get_edges_by_label(lbl),
202 None => self.inner.get_all_edges(),
203 }
204 }
205
206 /// Returns all outgoing edges from a node.
207 #[must_use]
208 pub fn get_outgoing(&self, node_id: u64) -> Vec<GraphEdge> {
209 self.inner.get_outgoing_edges(node_id)
210 }
211
212 /// Returns all incoming edges to a node.
213 #[must_use]
214 pub fn get_incoming(&self, node_id: u64) -> Vec<GraphEdge> {
215 self.inner.get_incoming_edges(node_id)
216 }
217
218 /// Returns the total number of edges in the graph without materializing them.
219 #[must_use]
220 pub fn edge_count(&self) -> usize {
221 self.inner.edge_count()
222 }
223
224 /// Returns `(in_degree, out_degree)` for a node.
225 #[must_use]
226 pub fn node_degree(&self, node_id: u64) -> (usize, usize) {
227 self.inner.get_node_degree(node_id)
228 }
229
230 /// Returns the IDs of all nodes that have a stored payload.
231 ///
232 /// Nodes that appear only as edge endpoints without a stored payload
233 /// are not included. Use [`GraphCollection::get_edges`] to discover
234 /// all referenced node IDs.
235 #[must_use]
236 pub fn all_node_ids(&self) -> Vec<u64> {
237 self.inner.all_ids()
238 }
239
240 /// Returns the next batch of points for scroll iteration.
241 ///
242 /// Delegates to the inner collection's `scroll_batch` (parallel
243 /// implementation to [`VectorCollection::scroll_batch`](crate::VectorCollection::scroll_batch)).
244 ///
245 /// # Errors
246 ///
247 /// Returns an error if `batch_size` is 0.
248 pub fn scroll_batch(
249 &self,
250 cursor: Option<u64>,
251 batch_size: usize,
252 filter: Option<&crate::filter::Filter>,
253 ) -> Result<crate::collection::ScrollBatch> {
254 self.inner.scroll_batch(cursor, batch_size, filter)
255 }
256
257 /// Returns the number of nodes (points) stored in this collection.
258 #[must_use]
259 pub fn len(&self) -> usize {
260 self.inner.len()
261 }
262
263 /// Returns `true` if the collection contains no nodes.
264 #[must_use]
265 pub fn is_empty(&self) -> bool {
266 self.inner.is_empty()
267 }
268
269 /// Retrieves nodes by IDs, returning `None` for missing entries.
270 #[must_use]
271 pub fn get(&self, ids: &[u64]) -> Vec<Option<Point>> {
272 self.inner.get(ids)
273 }
274
275 /// Deletes nodes by IDs.
276 ///
277 /// Missing IDs are silently ignored.
278 ///
279 /// # Errors
280 ///
281 /// Returns an error if storage operations fail.
282 pub fn delete(&self, ids: &[u64]) -> Result<()> {
283 self.inner.delete(ids)
284 }
285
286 /// Removes an edge from the graph by ID.
287 ///
288 /// Returns `true` if the edge existed and was removed, `false` otherwise —
289 /// including on the genuine failure paths. Use
290 /// [`Self::remove_edge_detailed`] when a failure must not pass for
291 /// "already gone".
292 #[must_use]
293 pub fn remove_edge(&self, edge_id: u64) -> bool {
294 self.inner.remove_edge(edge_id)
295 }
296
297 /// Removes an edge from the graph by ID, reporting WHY when it does not
298 /// happen.
299 pub(crate) fn remove_edge_detailed(&self, edge_id: u64) -> EdgeRemoval {
300 self.inner.remove_edge_detailed(edge_id)
301 }
302
303 /// Test-only fault injection — makes the edge write-ahead log unwritable so
304 /// edge removals fail while the edges themselves stay healthy.
305 #[cfg(all(test, feature = "persistence"))]
306 pub(crate) fn break_edge_wal_for_test(&self) -> std::io::Result<()> {
307 self.inner.break_edge_wal_for_test()
308 }
309
310 /// Returns `true` if an edge with `edge_id` exists in the graph.
311 #[must_use]
312 pub fn has_edge(&self, edge_id: u64) -> bool {
313 self.inner.edge_exists(edge_id)
314 }
315
316 /// Performs BFS traversal from a source node.
317 ///
318 /// # Examples
319 ///
320 /// ```rust,no_run
321 /// # use velesdb_core::{GraphCollection, GraphSchema, GraphEdge, DistanceMetric};
322 /// # use velesdb_core::collection::graph::TraversalConfig;
323 /// # let coll = GraphCollection::create("./data/kg".into(), "kg", None, DistanceMetric::Cosine, GraphSchema::schemaless())?;
324 /// let config = TraversalConfig { max_depth: 3, ..TraversalConfig::default() };
325 /// let results = coll.traverse_bfs(100, &config);
326 /// for r in &results {
327 /// println!("node={} depth={}", r.target_id, r.depth);
328 /// }
329 /// # Ok::<(), velesdb_core::Error>(())
330 /// ```
331 #[must_use]
332 pub fn traverse_bfs(&self, source_id: u64, config: &TraversalConfig) -> Vec<TraversalResult> {
333 self.inner.traverse_bfs_config(source_id, config)
334 }
335
336 /// Performs DFS traversal from a source node.
337 #[must_use]
338 pub fn traverse_dfs(&self, source_id: u64, config: &TraversalConfig) -> Vec<TraversalResult> {
339 self.inner.traverse_dfs_config(source_id, config)
340 }
341
342 /// Performs parallel BFS traversal from multiple start nodes.
343 ///
344 /// When `start_nodes` exceeds the parallel threshold (100 nodes), rayon
345 /// distributes independent per-start-node BFS traversals across CPU cores.
346 /// Results are deduplicated by path signature and truncated to `config.limit`.
347 ///
348 /// # Examples
349 ///
350 /// ```rust,no_run
351 /// # use velesdb_core::{GraphCollection, GraphSchema, DistanceMetric};
352 /// # use velesdb_core::collection::graph::TraversalConfig;
353 /// # let coll = GraphCollection::create("./data/kg".into(), "kg", None, DistanceMetric::Cosine, GraphSchema::schemaless())?;
354 /// let config = TraversalConfig { max_depth: 3, ..TraversalConfig::default() };
355 /// let results = coll.traverse_bfs_parallel(&[100, 200, 300], &config);
356 /// for r in &results {
357 /// println!("node={} depth={}", r.target_id, r.depth);
358 /// }
359 /// # Ok::<(), velesdb_core::Error>(())
360 /// ```
361 #[must_use]
362 pub fn traverse_bfs_parallel(
363 &self,
364 start_nodes: &[u64],
365 config: &TraversalConfig,
366 ) -> Vec<TraversalResult> {
367 self.inner.traverse_bfs_parallel(start_nodes, config)
368 }
369
370 // -------------------------------------------------------------------------
371 // Payload / node properties
372 // -------------------------------------------------------------------------
373
374 /// Inserts or updates node payload (properties).
375 ///
376 /// # Errors
377 ///
378 /// Returns an error if storage fails.
379 pub fn upsert_node_payload(&self, node_id: u64, payload: &serde_json::Value) -> Result<()> {
380 self.inner.store_node_payload(node_id, payload)
381 }
382
383 /// Inserts or updates a node payload, optionally with an embedding vector.
384 ///
385 /// # Errors
386 ///
387 /// Returns an error if storage fails, the vector dimension is invalid, or
388 /// an embedding is supplied for a graph collection without embeddings.
389 pub fn upsert_node(
390 &self,
391 node_id: u64,
392 payload: &serde_json::Value,
393 vector: Option<Vec<f32>>,
394 ) -> Result<()> {
395 match vector {
396 Some(vector) => self
397 .inner
398 .upsert([Point::new(node_id, vector, Some(payload.clone()))]),
399 None => self.upsert_node_payload(node_id, payload),
400 }
401 }
402
403 /// Inserts or updates node payload (properties).
404 ///
405 /// # Errors
406 ///
407 /// Returns an error if storage fails.
408 #[deprecated(since = "1.6.0", note = "Use upsert_node_payload() instead")]
409 pub fn store_node_payload(&self, node_id: u64, payload: &serde_json::Value) -> Result<()> {
410 self.upsert_node_payload(node_id, payload)
411 }
412
413 /// Retrieves node payload.
414 ///
415 /// # Errors
416 ///
417 /// Returns an error if retrieval fails.
418 pub fn get_node_payload(&self, node_id: u64) -> Result<Option<serde_json::Value>> {
419 self.inner.get_node_payload(node_id)
420 }
421
422 // -------------------------------------------------------------------------
423 // Optional embedding search
424 // -------------------------------------------------------------------------
425
426 /// Searches for similar nodes by embedding (only available if `has_embeddings()`).
427 ///
428 /// # Errors
429 ///
430 /// Returns `Error::VectorNotAllowed` if this collection has no embeddings,
431 /// or `Error::DimensionMismatch` if the query dimension is wrong.
432 pub fn search_by_embedding(&self, query: &[f32], k: usize) -> Result<Vec<SearchResult>> {
433 self.inner.search_by_embedding(query, k)
434 }
435
436 /// Alias for [`search_by_embedding`](Self::search_by_embedding).
437 ///
438 /// Provided for API parity with [`crate::VectorCollection::search`].
439 ///
440 /// # Errors
441 ///
442 /// Returns `Error::VectorNotAllowed` if this collection has no embeddings,
443 /// or `Error::DimensionMismatch` if the query dimension is wrong.
444 pub fn search(&self, query: &[f32], k: usize) -> Result<Vec<SearchResult>> {
445 self.search_by_embedding(query, k)
446 }
447}
448
449#[cfg(test)]
450mod tests {
451 use super::*;
452 use crate::collection::graph::GraphSchema;
453 use crate::distance::DistanceMetric;
454 use std::collections::HashMap;
455 use tempfile::{tempdir, TempDir};
456
457 /// Creates a schemaless cosine `GraphCollection` in a fresh temp dir.
458 ///
459 /// Returns the `TempDir` guard alongside the collection so the backing
460 /// directory outlives the test. `dimension` controls embedding support
461 /// (`None` for payload/edge-only collections, `Some(n)` for searchable ones).
462 fn make_test_collection(dimension: Option<usize>) -> (TempDir, GraphCollection) {
463 let dir = tempdir().unwrap();
464 let col = GraphCollection::create(
465 dir.path().to_path_buf(),
466 "kg",
467 dimension,
468 DistanceMetric::Cosine,
469 GraphSchema::schemaless(),
470 )
471 .unwrap();
472 (dir, col)
473 }
474
475 #[test]
476 fn test_all_node_ids_returns_ids_with_payload() {
477 let (_dir, col) = make_test_collection(None);
478
479 // Store payloads on two nodes
480 col.upsert_node_payload(10, &serde_json::json!({"name": "Alice"}))
481 .unwrap();
482 col.upsert_node_payload(20, &serde_json::json!({"name": "Bob"}))
483 .unwrap();
484
485 let ids = col.all_node_ids();
486 assert!(ids.contains(&10), "node 10 should be present");
487 assert!(ids.contains(&20), "node 20 should be present");
488 assert_eq!(ids.len(), 2);
489 }
490
491 #[test]
492 fn test_upsert_node_with_embedding_is_searchable() {
493 let (_dir, col) = make_test_collection(Some(4));
494
495 col.upsert_node(
496 10,
497 &serde_json::json!({"name": "Alice"}),
498 Some(vec![1.0, 0.0, 0.0, 0.0]),
499 )
500 .unwrap();
501
502 assert_eq!(
503 col.get_node_payload(10).unwrap(),
504 Some(serde_json::json!({"name": "Alice"}))
505 );
506 let results = col.search_by_embedding(&[1.0, 0.0, 0.0, 0.0], 1).unwrap();
507 assert_eq!(results[0].point.id, 10);
508 }
509
510 #[test]
511 fn test_edge_count_returns_correct_count() {
512 let (_dir, col) = make_test_collection(None);
513
514 assert_eq!(col.edge_count(), 0);
515 for id in [10, 20, 30] {
516 col.upsert_node_payload(id, &serde_json::json!({})).unwrap();
517 }
518
519 let edge1 = crate::collection::graph::GraphEdge::new(1, 10, 20, "knows").unwrap();
520 col.add_edge(edge1).unwrap();
521 assert_eq!(col.edge_count(), 1);
522
523 let edge2 = crate::collection::graph::GraphEdge::new(2, 20, 30, "likes").unwrap();
524 col.add_edge(edge2).unwrap();
525 assert_eq!(col.edge_count(), 2);
526 }
527
528 #[test]
529 fn test_traverse_bfs_parallel_through_graph_collection() {
530 let (_dir, col) = make_test_collection(None);
531
532 // Build chain: 1->2->3
533 for id in [1, 2, 3] {
534 col.upsert_node_payload(id, &serde_json::json!({})).unwrap();
535 }
536 col.add_edge(GraphEdge::new(1, 1, 2, "NEXT").unwrap())
537 .unwrap();
538 col.add_edge(GraphEdge::new(2, 2, 3, "NEXT").unwrap())
539 .unwrap();
540
541 let config = TraversalConfig {
542 max_depth: 3,
543 min_depth: 1,
544 ..TraversalConfig::default()
545 };
546 let results = col.traverse_bfs_parallel(&[1], &config);
547 let target_ids: std::collections::HashSet<u64> =
548 results.iter().map(|r| r.target_id).collect();
549 assert!(target_ids.contains(&2), "should reach node 2");
550 assert!(target_ids.contains(&3), "should reach node 3");
551 }
552
553 #[test]
554 fn test_execute_match_finds_edges() {
555 let (_dir, col) = make_test_collection(None);
556
557 // Store node payloads with labels
558 col.upsert_node_payload(
559 10,
560 &serde_json::json!({"_labels": ["Person"], "name": "Alice"}),
561 )
562 .unwrap();
563 col.upsert_node_payload(
564 20,
565 &serde_json::json!({"_labels": ["Person"], "name": "Bob"}),
566 )
567 .unwrap();
568
569 // Add edge: Alice -> Bob
570 let edge = crate::collection::graph::GraphEdge::new(1, 10, 20, "KNOWS").unwrap();
571 col.add_edge(edge).unwrap();
572
573 // MATCH query through the GraphCollection delegate
574 let match_clause = crate::velesql::MatchClause {
575 patterns: vec![crate::velesql::GraphPattern {
576 name: None,
577 nodes: vec![
578 crate::velesql::NodePattern::new().with_alias("a"),
579 crate::velesql::NodePattern::new().with_alias("b"),
580 ],
581 relationships: vec![crate::velesql::RelationshipPattern::new(
582 crate::velesql::Direction::Outgoing,
583 )],
584 }],
585 where_clause: None,
586 return_clause: crate::velesql::ReturnClause {
587 items: vec![],
588 order_by: None,
589 limit: Some(10),
590 },
591 };
592
593 let params = HashMap::new();
594 let results = col.execute_match(&match_clause, ¶ms).unwrap();
595 assert!(
596 !results.is_empty(),
597 "execute_match should find the KNOWS edge"
598 );
599 assert_eq!(results[0].node_id, 20, "target should be Bob (id=20)");
600 }
601
602 #[test]
603 fn test_has_edge_and_remove_edge() {
604 let (_dir, col) = make_test_collection(None);
605 assert!(!col.has_edge(7), "unknown edge id is absent");
606 for id in [10, 20] {
607 col.upsert_node_payload(id, &serde_json::json!({})).unwrap();
608 }
609
610 col.add_edge(GraphEdge::new(7, 10, 20, "KNOWS").unwrap())
611 .unwrap();
612 assert!(col.has_edge(7), "edge present after add");
613
614 assert!(col.remove_edge(7), "removing an existing edge returns true");
615 assert!(!col.has_edge(7), "edge gone after remove");
616 assert!(!col.remove_edge(7), "removing a missing edge returns false");
617 }
618
619 #[test]
620 fn test_upsert_node_without_vector_stores_payload_only() {
621 // No embeddings: the `None`-vector branch delegates to upsert_node_payload.
622 let (_dir, col) = make_test_collection(None);
623 col.upsert_node(42, &serde_json::json!({"name": "Carol"}), None)
624 .unwrap();
625 assert_eq!(
626 col.get_node_payload(42).unwrap(),
627 Some(serde_json::json!({"name": "Carol"}))
628 );
629 assert!(col.all_node_ids().contains(&42));
630 assert!(!col.has_embeddings(), "no embeddings without a dimension");
631 }
632
633 #[test]
634 fn test_get_edges_filtered_by_label() {
635 let (_dir, col) = make_test_collection(None);
636 for id in [10, 20, 30, 40] {
637 col.upsert_node_payload(id, &serde_json::json!({})).unwrap();
638 }
639 col.add_edge(GraphEdge::new(1, 10, 20, "KNOWS").unwrap())
640 .unwrap();
641 col.add_edge(GraphEdge::new(2, 20, 30, "LIKES").unwrap())
642 .unwrap();
643 col.add_edge(GraphEdge::new(3, 30, 40, "KNOWS").unwrap())
644 .unwrap();
645
646 let knows = col.get_edges(Some("KNOWS"));
647 assert_eq!(knows.len(), 2, "two KNOWS edges");
648 assert!(knows.iter().all(|e| e.label() == "KNOWS"));
649
650 let all = col.get_edges(None);
651 assert_eq!(all.len(), 3, "three edges total");
652 }
653
654 #[test]
655 fn test_node_degree_and_directional_edges() {
656 let (_dir, col) = make_test_collection(None);
657 for id in [10, 20, 30, 40] {
658 col.upsert_node_payload(id, &serde_json::json!({})).unwrap();
659 }
660 col.add_edge(GraphEdge::new(1, 10, 20, "NEXT").unwrap())
661 .unwrap();
662 col.add_edge(GraphEdge::new(2, 30, 20, "NEXT").unwrap())
663 .unwrap();
664 col.add_edge(GraphEdge::new(3, 20, 40, "NEXT").unwrap())
665 .unwrap();
666
667 // Node 20: 2 incoming (from 10, 30), 1 outgoing (to 40).
668 assert_eq!(col.node_degree(20), (2, 1));
669 assert_eq!(col.get_incoming(20).len(), 2);
670 let outgoing = col.get_outgoing(20);
671 assert_eq!(outgoing.len(), 1);
672 assert_eq!(outgoing[0].target(), 40);
673 }
674
675 #[test]
676 fn test_delete_removes_node_payload() {
677 let (_dir, col) = make_test_collection(None);
678 col.upsert_node_payload(10, &serde_json::json!({"k": 1}))
679 .unwrap();
680 col.upsert_node_payload(20, &serde_json::json!({"k": 2}))
681 .unwrap();
682 assert_eq!(col.all_node_ids().len(), 2);
683
684 col.delete(&[10]).unwrap();
685 assert!(col.get(&[10])[0].is_none(), "deleted node is gone");
686 assert!(
687 !col.all_node_ids().contains(&10),
688 "deleted node leaves the id set"
689 );
690 assert!(col.get_node_payload(20).unwrap().is_some(), "node 20 stays");
691 }
692
693 #[test]
694 fn test_scroll_batch_paginates_embedded_nodes() {
695 // scroll_batch iterates the point (vector) store, so use embeddings.
696 let (_dir, col) = make_test_collection(Some(2));
697 for id in [1u64, 2, 3] {
698 col.upsert_node(id, &serde_json::json!({"id": id}), Some(vec![1.0, 0.0]))
699 .unwrap();
700 }
701 assert!(!col.is_empty());
702 assert_eq!(col.len(), 3);
703
704 let first = col.scroll_batch(None, 2, None).unwrap();
705 assert_eq!(first.points.len(), 2, "first page has 2 of 3 nodes");
706 let cursor = first.next_cursor.expect("non-empty page yields a cursor");
707 let second = col.scroll_batch(Some(cursor), 2, None).unwrap();
708 assert_eq!(second.points.len(), 1, "second page has the last node");
709 // A page past the end returns no points (and therefore no cursor).
710 let tail_cursor = second.next_cursor.expect("page yields a cursor");
711 let third = col.scroll_batch(Some(tail_cursor), 2, None).unwrap();
712 assert!(third.points.is_empty(), "no points past the end");
713 assert!(third.next_cursor.is_none(), "empty page yields no cursor");
714
715 // batch_size 0 is rejected.
716 assert!(col.scroll_batch(None, 0, None).is_err());
717 }
718
719 #[test]
720 fn test_flush_and_flush_full_succeed() {
721 let (_dir, col) = make_test_collection(None);
722 col.upsert_node_payload(1, &serde_json::json!({"k": 1}))
723 .unwrap();
724 col.upsert_node_payload(2, &serde_json::json!({})).unwrap();
725 col.add_edge(GraphEdge::new(1, 1, 2, "NEXT").unwrap())
726 .unwrap();
727 col.flush().expect("fast-path flush succeeds");
728 col.flush_full().expect("full durability flush succeeds");
729 }
730
731 #[test]
732 fn test_reopen_recovers_edges_and_payloads() {
733 let dir = tempdir().unwrap();
734 let path = dir.path().to_path_buf();
735 {
736 let col = GraphCollection::create(
737 path.clone(),
738 "kg",
739 None,
740 DistanceMetric::Cosine,
741 GraphSchema::schemaless(),
742 )
743 .unwrap();
744 col.upsert_node_payload(1, &serde_json::json!({"name": "A"}))
745 .unwrap();
746 col.upsert_node_payload(2, &serde_json::json!({})).unwrap();
747 col.add_edge(GraphEdge::new(5, 1, 2, "NEXT").unwrap())
748 .unwrap();
749 col.flush_full().unwrap();
750 }
751 let reopened = GraphCollection::open(path).unwrap();
752 assert_eq!(reopened.name(), "kg");
753 assert!(reopened.has_edge(5), "edge survives reopen");
754 assert_eq!(
755 reopened.get_node_payload(1).unwrap(),
756 Some(serde_json::json!({"name": "A"}))
757 );
758 }
759}