velesdb_core/collection/graph/edge_concurrent/
persistence.rs

1//! `ConcurrentEdgeStore` persistence (serialization/deserialization).
2//!
3//! Extracted from `edge_concurrent/mod.rs` to reduce NLOC below the 500 threshold.
4
5use super::super::edge::EdgeStore;
6use super::ConcurrentEdgeStore;
7
8impl ConcurrentEdgeStore {
9    /// Builds a `ConcurrentEdgeStore` from a persisted `EdgeStore`.
10    ///
11    /// Re-distributes edges across shards based on source node ID.
12    ///
13    /// Issue #905: uses [`add_edges_batch`](Self::add_edges_batch) (one
14    /// `edge_ids` write-lock acquisition for the whole set, snapshot
15    /// invalidated once) instead of a per-edge `add_edge` loop (one lock
16    /// cycle + one snapshot-dirty flip per edge). The single
17    /// [`build_read_snapshot`](Self::build_read_snapshot) at the end performs
18    /// exactly one O(N+E) CSR build for the whole reconstruction.
19    #[must_use]
20    pub fn from_edge_store(store: &EdgeStore) -> Self {
21        let edges: Vec<_> = store.all_edges().into_iter().cloned().collect();
22        let concurrent = Self::with_estimated_edges(edges.len());
23
24        let expected = edges.len();
25        let added = concurrent.add_edges_batch(edges);
26        if added != expected {
27            tracing::warn!(
28                "skipped {} duplicate edge(s) during CES reconstruction",
29                expected - added
30            );
31        }
32
33        concurrent.build_read_snapshot();
34        concurrent
35    }
36
37    /// Saves the concurrent edge store to a file.
38    ///
39    /// # Errors
40    ///
41    /// Returns an error if serialization or file I/O fails.
42    pub fn save_to_file(&self, path: &std::path::Path) -> std::io::Result<()> {
43        self.to_merged_edge_store().save_to_file(path)
44    }
45
46    /// Loads a concurrent edge store from a persisted file.
47    ///
48    /// # Errors
49    ///
50    /// Returns an error if file I/O or deserialization fails.
51    pub fn load_from_file(path: &std::path::Path) -> std::io::Result<Self> {
52        let store = EdgeStore::load_from_file(path)?;
53        Ok(Self::from_edge_store(&store))
54    }
55
56    /// Merges all shards into a single `EdgeStore` for serialization.
57    fn to_merged_edge_store(&self) -> EdgeStore {
58        let ids = self.edge_ids.read();
59        let mut merged = EdgeStore::with_capacity(ids.len(), ids.len());
60
61        for (&edge_id, &source_id) in ids.iter() {
62            let shard_idx = self.shard_index(source_id);
63            let guard = self.shards[shard_idx].read();
64            if let Some(edge) = guard.get_edge(edge_id) {
65                let _ = merged.add_edge(edge.clone());
66            }
67        }
68        merged
69    }
70}
velesdb_core/collection/graph/edge_concurrent/persistence.rs

velesdb_core/collection/graph/edge_concurrent/
persistence.rs
