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

velesdb_core/collection/
graph_collection.rs
