ruvector_graph/
graph.rs

1//! Graph database implementation with concurrent access and indexing
2
3use crate::edge::Edge;
4use crate::error::Result;
5use crate::hyperedge::{Hyperedge, HyperedgeId};
6use crate::index::{AdjacencyIndex, EdgeTypeIndex, HyperedgeNodeIndex, LabelIndex, PropertyIndex};
7use crate::node::Node;
8#[cfg(feature = "storage")]
9use crate::storage::GraphStorage;
10use crate::types::{EdgeId, NodeId, PropertyValue};
11use dashmap::DashMap;
12#[cfg(feature = "storage")]
13use std::path::Path;
14use std::sync::Arc;
15
16/// High-performance graph database with concurrent access
17pub struct GraphDB {
18    /// In-memory node storage (DashMap for lock-free concurrent reads)
19    nodes: Arc<DashMap<NodeId, Node>>,
20    /// In-memory edge storage
21    edges: Arc<DashMap<EdgeId, Edge>>,
22    /// In-memory hyperedge storage
23    hyperedges: Arc<DashMap<HyperedgeId, Hyperedge>>,
24    /// Label index for fast label-based lookups
25    label_index: LabelIndex,
26    /// Property index for fast property-based lookups
27    property_index: PropertyIndex,
28    /// Edge type index
29    edge_type_index: EdgeTypeIndex,
30    /// Adjacency index for neighbor lookups
31    adjacency_index: AdjacencyIndex,
32    /// Hyperedge node index
33    hyperedge_node_index: HyperedgeNodeIndex,
34    /// Optional persistent storage
35    #[cfg(feature = "storage")]
36    storage: Option<GraphStorage>,
37}
38
39impl GraphDB {
40    /// Create a new in-memory graph database
41    pub fn new() -> Self {
42        Self {
43            nodes: Arc::new(DashMap::new()),
44            edges: Arc::new(DashMap::new()),
45            hyperedges: Arc::new(DashMap::new()),
46            label_index: LabelIndex::new(),
47            property_index: PropertyIndex::new(),
48            edge_type_index: EdgeTypeIndex::new(),
49            adjacency_index: AdjacencyIndex::new(),
50            hyperedge_node_index: HyperedgeNodeIndex::new(),
51            #[cfg(feature = "storage")]
52            storage: None,
53        }
54    }
55
56    /// Create a new graph database with persistent storage
57    #[cfg(feature = "storage")]
58    pub fn with_storage<P: AsRef<Path>>(path: P) -> anyhow::Result<Self> {
59        let storage = GraphStorage::new(path)?;
60
61        let mut db = Self::new();
62        db.storage = Some(storage);
63
64        // Load existing data from storage
65        db.load_from_storage()?;
66
67        Ok(db)
68    }
69
70    /// Load all data from storage into memory
71    #[cfg(feature = "storage")]
72    fn load_from_storage(&mut self) -> anyhow::Result<()> {
73        if let Some(storage) = &self.storage {
74            // Load nodes
75            for node_id in storage.all_node_ids()? {
76                if let Some(node) = storage.get_node(&node_id)? {
77                    self.nodes.insert(node_id.clone(), node.clone());
78                    self.label_index.add_node(&node);
79                    self.property_index.add_node(&node);
80                }
81            }
82
83            // Load edges
84            for edge_id in storage.all_edge_ids()? {
85                if let Some(edge) = storage.get_edge(&edge_id)? {
86                    self.edges.insert(edge_id.clone(), edge.clone());
87                    self.edge_type_index.add_edge(&edge);
88                    self.adjacency_index.add_edge(&edge);
89                }
90            }
91
92            // Load hyperedges
93            for hyperedge_id in storage.all_hyperedge_ids()? {
94                if let Some(hyperedge) = storage.get_hyperedge(&hyperedge_id)? {
95                    self.hyperedges
96                        .insert(hyperedge_id.clone(), hyperedge.clone());
97                    self.hyperedge_node_index.add_hyperedge(&hyperedge);
98                }
99            }
100        }
101        Ok(())
102    }
103
104    // Node operations
105
106    /// Create a node
107    pub fn create_node(&self, node: Node) -> Result<NodeId> {
108        let id = node.id.clone();
109
110        // Update indexes
111        self.label_index.add_node(&node);
112        self.property_index.add_node(&node);
113
114        // Insert into memory
115        self.nodes.insert(id.clone(), node.clone());
116
117        // Persist to storage if available
118        #[cfg(feature = "storage")]
119        if let Some(storage) = &self.storage {
120            storage.insert_node(&node)?;
121        }
122
123        Ok(id)
124    }
125
126    /// Get a node by ID
127    pub fn get_node(&self, id: impl AsRef<str>) -> Option<Node> {
128        self.nodes.get(id.as_ref()).map(|entry| entry.clone())
129    }
130
131    /// Delete a node
132    pub fn delete_node(&self, id: impl AsRef<str>) -> Result<bool> {
133        if let Some((_, node)) = self.nodes.remove(id.as_ref()) {
134            // Update indexes
135            self.label_index.remove_node(&node);
136            self.property_index.remove_node(&node);
137
138            // Delete from storage if available
139            #[cfg(feature = "storage")]
140            if let Some(storage) = &self.storage {
141                storage.delete_node(id.as_ref())?;
142            }
143
144            Ok(true)
145        } else {
146            Ok(false)
147        }
148    }
149
150    /// Get nodes by label
151    pub fn get_nodes_by_label(&self, label: &str) -> Vec<Node> {
152        self.label_index
153            .get_nodes_by_label(label)
154            .into_iter()
155            .filter_map(|id| self.get_node(&id))
156            .collect()
157    }
158
159    /// Get nodes by property
160    pub fn get_nodes_by_property(&self, key: &str, value: &PropertyValue) -> Vec<Node> {
161        self.property_index
162            .get_nodes_by_property(key, value)
163            .into_iter()
164            .filter_map(|id| self.get_node(&id))
165            .collect()
166    }
167
168    // Edge operations
169
170    /// Create an edge
171    pub fn create_edge(&self, edge: Edge) -> Result<EdgeId> {
172        let id = edge.id.clone();
173
174        // Verify nodes exist
175        if !self.nodes.contains_key(&edge.from) || !self.nodes.contains_key(&edge.to) {
176            return Err(crate::error::GraphError::NodeNotFound(
177                "Source or target node not found".to_string(),
178            ));
179        }
180
181        // Update indexes
182        self.edge_type_index.add_edge(&edge);
183        self.adjacency_index.add_edge(&edge);
184
185        // Insert into memory
186        self.edges.insert(id.clone(), edge.clone());
187
188        // Persist to storage if available
189        #[cfg(feature = "storage")]
190        if let Some(storage) = &self.storage {
191            storage.insert_edge(&edge)?;
192        }
193
194        Ok(id)
195    }
196
197    /// Get an edge by ID
198    pub fn get_edge(&self, id: impl AsRef<str>) -> Option<Edge> {
199        self.edges.get(id.as_ref()).map(|entry| entry.clone())
200    }
201
202    /// Delete an edge
203    pub fn delete_edge(&self, id: impl AsRef<str>) -> Result<bool> {
204        if let Some((_, edge)) = self.edges.remove(id.as_ref()) {
205            // Update indexes
206            self.edge_type_index.remove_edge(&edge);
207            self.adjacency_index.remove_edge(&edge);
208
209            // Delete from storage if available
210            #[cfg(feature = "storage")]
211            if let Some(storage) = &self.storage {
212                storage.delete_edge(id.as_ref())?;
213            }
214
215            Ok(true)
216        } else {
217            Ok(false)
218        }
219    }
220
221    /// Get edges by type
222    pub fn get_edges_by_type(&self, edge_type: &str) -> Vec<Edge> {
223        self.edge_type_index
224            .get_edges_by_type(edge_type)
225            .into_iter()
226            .filter_map(|id| self.get_edge(&id))
227            .collect()
228    }
229
230    /// Get outgoing edges from a node
231    pub fn get_outgoing_edges(&self, node_id: &NodeId) -> Vec<Edge> {
232        self.adjacency_index
233            .get_outgoing_edges(node_id)
234            .into_iter()
235            .filter_map(|id| self.get_edge(&id))
236            .collect()
237    }
238
239    /// Get incoming edges to a node
240    pub fn get_incoming_edges(&self, node_id: &NodeId) -> Vec<Edge> {
241        self.adjacency_index
242            .get_incoming_edges(node_id)
243            .into_iter()
244            .filter_map(|id| self.get_edge(&id))
245            .collect()
246    }
247
248    // Hyperedge operations
249
250    /// Create a hyperedge
251    pub fn create_hyperedge(&self, hyperedge: Hyperedge) -> Result<HyperedgeId> {
252        let id = hyperedge.id.clone();
253
254        // Verify all nodes exist
255        for node_id in &hyperedge.nodes {
256            if !self.nodes.contains_key(node_id) {
257                return Err(crate::error::GraphError::NodeNotFound(format!(
258                    "Node {} not found",
259                    node_id
260                )));
261            }
262        }
263
264        // Update index
265        self.hyperedge_node_index.add_hyperedge(&hyperedge);
266
267        // Insert into memory
268        self.hyperedges.insert(id.clone(), hyperedge.clone());
269
270        // Persist to storage if available
271        #[cfg(feature = "storage")]
272        if let Some(storage) = &self.storage {
273            storage.insert_hyperedge(&hyperedge)?;
274        }
275
276        Ok(id)
277    }
278
279    /// Get a hyperedge by ID
280    pub fn get_hyperedge(&self, id: &HyperedgeId) -> Option<Hyperedge> {
281        self.hyperedges.get(id).map(|entry| entry.clone())
282    }
283
284    /// Get hyperedges containing a node
285    pub fn get_hyperedges_by_node(&self, node_id: &NodeId) -> Vec<Hyperedge> {
286        self.hyperedge_node_index
287            .get_hyperedges_by_node(node_id)
288            .into_iter()
289            .filter_map(|id| self.get_hyperedge(&id))
290            .collect()
291    }
292
293    // Statistics
294
295    /// Get the number of nodes
296    pub fn node_count(&self) -> usize {
297        self.nodes.len()
298    }
299
300    /// Get the number of edges
301    pub fn edge_count(&self) -> usize {
302        self.edges.len()
303    }
304
305    /// Get the number of hyperedges
306    pub fn hyperedge_count(&self) -> usize {
307        self.hyperedges.len()
308    }
309}
310
311impl Default for GraphDB {
312    fn default() -> Self {
313        Self::new()
314    }
315}
316
317#[cfg(test)]
318mod tests {
319    use super::*;
320    use crate::edge::EdgeBuilder;
321    use crate::hyperedge::HyperedgeBuilder;
322    use crate::node::NodeBuilder;
323
324    #[test]
325    fn test_graph_creation() {
326        let db = GraphDB::new();
327        assert_eq!(db.node_count(), 0);
328        assert_eq!(db.edge_count(), 0);
329    }
330
331    #[test]
332    fn test_node_operations() {
333        let db = GraphDB::new();
334
335        let node = NodeBuilder::new()
336            .label("Person")
337            .property("name", "Alice")
338            .build();
339
340        let id = db.create_node(node.clone()).unwrap();
341        assert_eq!(db.node_count(), 1);
342
343        let retrieved = db.get_node(&id);
344        assert!(retrieved.is_some());
345
346        let deleted = db.delete_node(&id).unwrap();
347        assert!(deleted);
348        assert_eq!(db.node_count(), 0);
349    }
350
351    #[test]
352    fn test_edge_operations() {
353        let db = GraphDB::new();
354
355        let node1 = NodeBuilder::new().build();
356        let node2 = NodeBuilder::new().build();
357
358        let id1 = db.create_node(node1.clone()).unwrap();
359        let id2 = db.create_node(node2.clone()).unwrap();
360
361        let edge = EdgeBuilder::new(id1.clone(), id2.clone(), "KNOWS")
362            .property("since", 2020i64)
363            .build();
364
365        let edge_id = db.create_edge(edge).unwrap();
366        assert_eq!(db.edge_count(), 1);
367
368        let retrieved = db.get_edge(&edge_id);
369        assert!(retrieved.is_some());
370    }
371
372    #[test]
373    fn test_label_index() {
374        let db = GraphDB::new();
375
376        let node1 = NodeBuilder::new().label("Person").build();
377        let node2 = NodeBuilder::new().label("Person").build();
378        let node3 = NodeBuilder::new().label("Organization").build();
379
380        db.create_node(node1).unwrap();
381        db.create_node(node2).unwrap();
382        db.create_node(node3).unwrap();
383
384        let people = db.get_nodes_by_label("Person");
385        assert_eq!(people.len(), 2);
386
387        let orgs = db.get_nodes_by_label("Organization");
388        assert_eq!(orgs.len(), 1);
389    }
390
391    #[test]
392    fn test_hyperedge_operations() {
393        let db = GraphDB::new();
394
395        let node1 = NodeBuilder::new().build();
396        let node2 = NodeBuilder::new().build();
397        let node3 = NodeBuilder::new().build();
398
399        let id1 = db.create_node(node1).unwrap();
400        let id2 = db.create_node(node2).unwrap();
401        let id3 = db.create_node(node3).unwrap();
402
403        let hyperedge =
404            HyperedgeBuilder::new(vec![id1.clone(), id2.clone(), id3.clone()], "MEETING")
405                .description("Team meeting")
406                .build();
407
408        let hedge_id = db.create_hyperedge(hyperedge).unwrap();
409        assert_eq!(db.hyperedge_count(), 1);
410
411        let hedges = db.get_hyperedges_by_node(&id1);
412        assert_eq!(hedges.len(), 1);
413    }
414}
ruvector_graph/graph.rs

ruvector_graph/
graph.rs
