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 /// Creates a new `GraphCollection`, optionally with explicit HNSW
71 /// parameters for the node-embedding index.
72 ///
73 /// Only meaningful together with `dimension = Some(d)`: that is when a
74 /// graph collection carries node embeddings and builds an HNSW index over
75 /// them. Passing `hnsw_params = None` is identical to
76 /// [`GraphCollection::create`].
77 ///
78 /// # Errors
79 ///
80 /// Returns an error if the directory cannot be created or storage fails.
81 pub fn create_with_hnsw_params(
82 path: PathBuf,
83 name: &str,
84 dimension: Option<usize>,
85 metric: DistanceMetric,
86 schema: GraphSchema,
87 hnsw_params: Option<crate::index::hnsw::HnswParams>,
88 ) -> Result<Self> {
89 Ok(Self {
90 inner: Collection::create_graph_collection_with_hnsw_params(
91 path,
92 name,
93 schema,
94 dimension,
95 metric,
96 hnsw_params,
97 )?,
98 })
99 }
100
101 /// Opens an existing `GraphCollection` from disk.
102 ///
103 /// # Errors
104 ///
105 /// Returns an error if config or storage cannot be opened.
106 pub fn open(path: PathBuf) -> Result<Self> {
107 Ok(Self {
108 inner: Collection::open(path)?,
109 })
110 }
111
112 /// Consumes `self` and returns a [`VectorCollection`](super::VectorCollection)
113 /// **structural view** over this graph collection's shared `inner` store.
114 ///
115 /// This is a purely structural re-wrap of the identical `inner: Collection`
116 /// backing store into the `VectorCollection` newtype — an ordinary value
117 /// move, **not** a `transmute` and not memory-unsafe. It does **not** assert
118 /// that the collection is vector-kind: invoking vector-specific methods on
119 /// the result returns empty or misleading state.
120 ///
121 /// It exists solely for the **Python binding**, whose single user-facing
122 /// `Collection` type is backed by a `VectorCollection`: the binding holds a
123 /// graph collection behind that type while gating vector-only operations on
124 /// the real kind it tracks separately (the Python
125 /// `Collection::ensure_vector` guard). The Mobile and Tauri bindings do
126 /// *not* use this view — they go through the variant-checked
127 /// [`AnyCollection::into_vector`](super::AnyCollection::into_vector) and
128 /// reject non-vector collections. Callers that need the graph surface must
129 /// use the graph API, not this view.
130 ///
131 /// [`MetadataCollection::into_vector_view`](super::MetadataCollection::into_vector_view)
132 /// is the exact mirror for metadata collections.
133 #[must_use]
134 pub fn into_vector_view(self) -> super::VectorCollection {
135 super::VectorCollection { inner: self.inner }
136 }
137
138 /// This graph collection's operational metrics.
139 ///
140 /// Reads what the edge write and traversal paths already record, so an
141 /// exporter can publish it; the counters are bumped regardless of whether
142 /// anything reads them.
143 #[must_use]
144 pub fn metrics(&self) -> &crate::collection::graph::GraphMetrics {
145 // Reaches the edge store's counters directly: `Collection::graph` is
146 // `pub(crate)`, and graph_api.rs — where a `Collection` accessor would
147 // naturally sit — is over its frozen line budget and may only shrink.
148 self.inner.graph.edge_store.metrics()
149 }
150
151 /// Flushes all state to disk.
152 ///
153 /// Issue #423: This fast-path flush skips `vectors.idx` serialization.
154 /// The WAL provides crash recovery for the vector index.
155 ///
156 /// # Errors
157 ///
158 /// Returns an error if any flush operation fails.
159 pub fn flush(&self) -> Result<()> {
160 self.inner.flush()
161 }
162
163 /// Full durability flush including `vectors.idx` serialization.
164 ///
165 /// Issue #423: Use on graceful shutdown to avoid a full WAL replay
166 /// on the next startup.
167 ///
168 /// # Errors
169 ///
170 /// Returns an error if any flush operation fails.
171 pub fn flush_full(&self) -> Result<()> {
172 self.inner.flush_full()
173 }
174
175 // -------------------------------------------------------------------------
176 // Metadata
177 // -------------------------------------------------------------------------
178
179 /// Returns the collection name.
180 #[must_use]
181 pub fn name(&self) -> String {
182 self.inner.config().name
183 }
184
185 /// Returns the graph schema stored in config.
186 ///
187 /// Returns `GraphSchema::schemaless()` for collections that have no schema set.
188 #[must_use]
189 pub fn schema(&self) -> GraphSchema {
190 self.inner
191 .graph_schema()
192 .unwrap_or_else(GraphSchema::schemaless)
193 }
194
195 /// Returns `true` if this collection stores node embeddings.
196 #[must_use]
197 pub fn has_embeddings(&self) -> bool {
198 self.inner.has_embeddings()
199 }
200
201 // -------------------------------------------------------------------------
202 // Graph operations — delegate to Collection graph API
203 // -------------------------------------------------------------------------
204
205 /// Adds an edge between two nodes.
206 ///
207 /// # Errors
208 ///
209 /// - Returns `Error::EdgeExists` if an edge with the same ID already exists.
210 ///
211 /// # Examples
212 ///
213 /// ```rust,no_run
214 /// # use velesdb_core::{GraphCollection, GraphSchema, GraphEdge, DistanceMetric};
215 /// # let coll = GraphCollection::create("./data/kg".into(), "kg", None, DistanceMetric::Cosine, GraphSchema::schemaless())?;
216 /// let edge = GraphEdge::new(1, 100, 200, "KNOWS")?;
217 /// coll.add_edge(edge)?;
218 /// # Ok::<(), velesdb_core::Error>(())
219 /// ```
220 pub fn add_edge(&self, edge: GraphEdge) -> Result<()> {
221 self.inner.add_edge(edge)
222 }
223
224 /// Adds multiple edges in batch (much faster than calling add_edge in a loop).
225 ///
226 /// Acquires locks once for the entire batch and rebuilds the CSR snapshot
227 /// once at the end. Duplicate edge IDs are silently skipped.
228 ///
229 /// # Returns
230 ///
231 /// Number of edges successfully added.
232 ///
233 /// # Errors
234 ///
235 /// Returns an error if WAL durability logging fails for graph
236 /// collections (fail-closed: the in-memory store is not mutated).
237 pub fn add_edges_batch(&self, edges: Vec<GraphEdge>) -> Result<usize> {
238 self.inner.add_edges_batch(edges)
239 }
240
241 /// Returns edges, optionally filtered by label.
242 #[must_use]
243 pub fn get_edges(&self, label: Option<&str>) -> Vec<GraphEdge> {
244 match label {
245 Some(lbl) => self.inner.get_edges_by_label(lbl),
246 None => self.inner.get_all_edges(),
247 }
248 }
249
250 /// Returns all outgoing edges from a node.
251 #[must_use]
252 pub fn get_outgoing(&self, node_id: u64) -> Vec<GraphEdge> {
253 self.inner.get_outgoing_edges(node_id)
254 }
255
256 /// Returns all incoming edges to a node.
257 #[must_use]
258 pub fn get_incoming(&self, node_id: u64) -> Vec<GraphEdge> {
259 self.inner.get_incoming_edges(node_id)
260 }
261
262 /// Returns the total number of edges in the graph without materializing them.
263 #[must_use]
264 pub fn edge_count(&self) -> usize {
265 self.inner.edge_count()
266 }
267
268 /// Returns `(in_degree, out_degree)` for a node.
269 #[must_use]
270 pub fn node_degree(&self, node_id: u64) -> (usize, usize) {
271 self.inner.get_node_degree(node_id)
272 }
273
274 /// Returns the IDs of all nodes that have a stored payload.
275 ///
276 /// Nodes that appear only as edge endpoints without a stored payload
277 /// are not included. Use [`GraphCollection::get_edges`] to discover
278 /// all referenced node IDs.
279 #[must_use]
280 pub fn all_node_ids(&self) -> Vec<u64> {
281 self.inner.all_ids()
282 }
283
284 /// Returns the next batch of points for scroll iteration.
285 ///
286 /// Delegates to the inner collection's `scroll_batch` (parallel
287 /// implementation to [`VectorCollection::scroll_batch`](crate::VectorCollection::scroll_batch)).
288 ///
289 /// # Errors
290 ///
291 /// Returns an error if `batch_size` is 0.
292 pub fn scroll_batch(
293 &self,
294 cursor: Option<u64>,
295 batch_size: usize,
296 filter: Option<&crate::filter::Filter>,
297 ) -> Result<crate::collection::ScrollBatch> {
298 self.inner.scroll_batch(cursor, batch_size, filter)
299 }
300
301 /// Returns the number of nodes (points) stored in this collection.
302 #[must_use]
303 pub fn len(&self) -> usize {
304 self.inner.len()
305 }
306
307 /// Returns `true` if the collection contains no nodes.
308 #[must_use]
309 pub fn is_empty(&self) -> bool {
310 self.inner.is_empty()
311 }
312
313 /// Retrieves nodes by IDs, returning `None` for missing entries.
314 #[must_use]
315 pub fn get(&self, ids: &[u64]) -> Vec<Option<Point>> {
316 self.inner.get(ids)
317 }
318
319 /// Deletes nodes by IDs.
320 ///
321 /// Missing IDs are silently ignored.
322 ///
323 /// # Errors
324 ///
325 /// Returns an error if storage operations fail.
326 pub fn delete(&self, ids: &[u64]) -> Result<()> {
327 self.inner.delete(ids)
328 }
329
330 /// Removes an edge from the graph by ID.
331 ///
332 /// Returns `true` if the edge existed and was removed, `false` otherwise —
333 /// including on the genuine failure paths. Use
334 /// [`Self::remove_edge_detailed`] when a failure must not pass for
335 /// "already gone".
336 #[must_use]
337 pub fn remove_edge(&self, edge_id: u64) -> bool {
338 self.inner.remove_edge(edge_id)
339 }
340
341 /// Removes an edge from the graph by ID, reporting WHY when it does not
342 /// happen.
343 pub(crate) fn remove_edge_detailed(&self, edge_id: u64) -> EdgeRemoval {
344 self.inner.remove_edge_detailed(edge_id)
345 }
346
347 /// Test-only fault injection — makes the edge write-ahead log unwritable so
348 /// edge removals fail while the edges themselves stay healthy.
349 #[cfg(all(test, feature = "persistence"))]
350 pub(crate) fn break_edge_wal_for_test(&self) -> std::io::Result<()> {
351 self.inner.break_edge_wal_for_test()
352 }
353
354 /// Returns `true` if an edge with `edge_id` exists in the graph.
355 #[must_use]
356 pub fn has_edge(&self, edge_id: u64) -> bool {
357 self.inner.edge_exists(edge_id)
358 }
359
360 /// Performs BFS traversal from a source node.
361 ///
362 /// # Examples
363 ///
364 /// ```rust,no_run
365 /// # use velesdb_core::{GraphCollection, GraphSchema, GraphEdge, DistanceMetric};
366 /// # use velesdb_core::collection::graph::TraversalConfig;
367 /// # let coll = GraphCollection::create("./data/kg".into(), "kg", None, DistanceMetric::Cosine, GraphSchema::schemaless())?;
368 /// let config = TraversalConfig { max_depth: 3, ..TraversalConfig::default() };
369 /// let results = coll.traverse_bfs(100, &config);
370 /// for r in &results {
371 /// println!("node={} depth={}", r.target_id, r.depth);
372 /// }
373 /// # Ok::<(), velesdb_core::Error>(())
374 /// ```
375 #[must_use]
376 pub fn traverse_bfs(&self, source_id: u64, config: &TraversalConfig) -> Vec<TraversalResult> {
377 self.inner.traverse_bfs_config(source_id, config)
378 }
379
380 /// Performs DFS traversal from a source node.
381 #[must_use]
382 pub fn traverse_dfs(&self, source_id: u64, config: &TraversalConfig) -> Vec<TraversalResult> {
383 self.inner.traverse_dfs_config(source_id, config)
384 }
385
386 /// Performs parallel BFS traversal from multiple start nodes.
387 ///
388 /// When `start_nodes` exceeds the parallel threshold (100 nodes), rayon
389 /// distributes independent per-start-node BFS traversals across CPU cores.
390 /// Results are deduplicated by path signature and truncated to `config.limit`.
391 ///
392 /// # Examples
393 ///
394 /// ```rust,no_run
395 /// # use velesdb_core::{GraphCollection, GraphSchema, DistanceMetric};
396 /// # use velesdb_core::collection::graph::TraversalConfig;
397 /// # let coll = GraphCollection::create("./data/kg".into(), "kg", None, DistanceMetric::Cosine, GraphSchema::schemaless())?;
398 /// let config = TraversalConfig { max_depth: 3, ..TraversalConfig::default() };
399 /// let results = coll.traverse_bfs_parallel(&[100, 200, 300], &config);
400 /// for r in &results {
401 /// println!("node={} depth={}", r.target_id, r.depth);
402 /// }
403 /// # Ok::<(), velesdb_core::Error>(())
404 /// ```
405 #[must_use]
406 pub fn traverse_bfs_parallel(
407 &self,
408 start_nodes: &[u64],
409 config: &TraversalConfig,
410 ) -> Vec<TraversalResult> {
411 self.inner.traverse_bfs_parallel(start_nodes, config)
412 }
413
414 // -------------------------------------------------------------------------
415 // Payload / node properties
416 // -------------------------------------------------------------------------
417
418 /// Inserts or updates node payload (properties).
419 ///
420 /// # Errors
421 ///
422 /// Returns an error if storage fails.
423 pub fn upsert_node_payload(&self, node_id: u64, payload: &serde_json::Value) -> Result<()> {
424 self.inner.store_node_payload(node_id, payload)
425 }
426
427 /// Inserts or updates many node payloads under one durability barrier.
428 ///
429 /// Batched counterpart of [`upsert_node_payload`](Self::upsert_node_payload),
430 /// which pays a barrier per call: looping it cost one fsync per node
431 /// (#2153). Duplicate ids resolve last-wins, and the batch is validated in
432 /// full before anything is written, so a rejected batch commits nothing.
433 ///
434 /// # Errors
435 ///
436 /// Returns an error if storage fails or any payload fails validation.
437 pub fn upsert_node_payloads(&self, entries: &[(u64, &serde_json::Value)]) -> Result<()> {
438 self.inner.store_node_payloads(entries)
439 }
440
441 /// Inserts or updates a node payload, optionally with an embedding vector.
442 ///
443 /// # Errors
444 ///
445 /// Returns an error if storage fails, the vector dimension is invalid, or
446 /// an embedding is supplied for a graph collection without embeddings.
447 pub fn upsert_node(
448 &self,
449 node_id: u64,
450 payload: &serde_json::Value,
451 vector: Option<Vec<f32>>,
452 ) -> Result<()> {
453 match vector {
454 Some(vector) => self
455 .inner
456 .upsert([Point::new(node_id, vector, Some(payload.clone()))]),
457 None => self.upsert_node_payload(node_id, payload),
458 }
459 }
460
461 /// Inserts or updates node payload (properties).
462 ///
463 /// # Errors
464 ///
465 /// Returns an error if storage fails.
466 #[deprecated(since = "1.6.0", note = "Use upsert_node_payload() instead")]
467 pub fn store_node_payload(&self, node_id: u64, payload: &serde_json::Value) -> Result<()> {
468 self.upsert_node_payload(node_id, payload)
469 }
470
471 /// Retrieves node payload.
472 ///
473 /// # Errors
474 ///
475 /// Returns an error if retrieval fails.
476 pub fn get_node_payload(&self, node_id: u64) -> Result<Option<serde_json::Value>> {
477 self.inner.get_node_payload(node_id)
478 }
479
480 // -------------------------------------------------------------------------
481 // Optional embedding search
482 // -------------------------------------------------------------------------
483
484 /// Searches for similar nodes by embedding (only available if `has_embeddings()`).
485 ///
486 /// # Errors
487 ///
488 /// Returns `Error::VectorNotAllowed` if this collection has no embeddings,
489 /// or `Error::DimensionMismatch` if the query dimension is wrong.
490 pub fn search_by_embedding(&self, query: &[f32], k: usize) -> Result<Vec<SearchResult>> {
491 self.inner.search_by_embedding(query, k)
492 }
493
494 /// Alias for [`search_by_embedding`](Self::search_by_embedding).
495 ///
496 /// Provided for API parity with [`crate::VectorCollection::search`].
497 ///
498 /// # Errors
499 ///
500 /// Returns `Error::VectorNotAllowed` if this collection has no embeddings,
501 /// or `Error::DimensionMismatch` if the query dimension is wrong.
502 pub fn search(&self, query: &[f32], k: usize) -> Result<Vec<SearchResult>> {
503 self.search_by_embedding(query, k)
504 }
505}
506
507#[cfg(test)]
508#[path = "graph_collection_tests.rs"]
509mod tests;