1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
// SPDX-License-Identifier: Apache-2.0
//! Edge insert / remove paths and node-edge cleanup.
use super::types::CsrIndex;
impl CsrIndex {
/// Incrementally add an unweighted edge (goes into mutable buffer).
/// Uses weight 1.0 if the graph already has weighted edges.
///
/// Returns `Err(GraphError::LabelOverflow)` if the label id space is
/// exhausted. Production callers should always surface this to the
/// client; silently ignoring it reproduces the silent-wrap bug the
/// `u32` widening was meant to fix.
pub fn add_edge(&mut self, src: &str, label: &str, dst: &str) -> Result<(), crate::GraphError> {
self.add_edge_internal(src, label, dst, "", 1.0, false)
}
/// Add an unweighted edge scoped to `collection`.
///
/// The collection tag is what lets a collection-scoped MATCH / RAG read
/// (`... IN '<collection>'`) see only its own edges even though every
/// collection's edges share one per-`(database, tenant)` CSR partition
/// (nodes, node-labels and surrogates are shared across collections; only
/// edges carry the collection axis).
pub fn add_edge_in_collection(
&mut self,
src: &str,
label: &str,
dst: &str,
collection: &str,
) -> Result<(), crate::GraphError> {
self.add_edge_internal(src, label, dst, collection, 1.0, false)
}
/// Add a weighted edge scoped to `collection`. See
/// [`Self::add_edge_in_collection`] and [`Self::add_edge_weighted`].
pub fn add_edge_weighted_in_collection(
&mut self,
src: &str,
label: &str,
dst: &str,
collection: &str,
weight: f64,
) -> Result<(), crate::GraphError> {
self.add_edge_internal(src, label, dst, collection, weight, weight != 1.0)
}
/// Incrementally add a weighted edge (goes into mutable buffer).
///
/// If this is the first weighted edge (weight != 1.0), initializes
/// the weight tracking infrastructure (backfills existing buffer
/// entries with 1.0).
pub fn add_edge_weighted(
&mut self,
src: &str,
label: &str,
dst: &str,
weight: f64,
) -> Result<(), crate::GraphError> {
self.add_edge_internal(src, label, dst, "", weight, weight != 1.0)
}
fn add_edge_internal(
&mut self,
src: &str,
label: &str,
dst: &str,
collection: &str,
weight: f64,
force_weights: bool,
) -> Result<(), crate::GraphError> {
let src_id = self.ensure_node(src)?;
let dst_id = self.ensure_node(dst)?;
let label_id = self.ensure_label(label)?;
let collection_id = self.ensure_collection(collection);
// Check for duplicates in buffer. Edge identity is collection-aware:
// the same `(label, dst)` under a DIFFERENT collection is a distinct
// edge and must NOT be deduplicated away.
let out = &self.buffer_out[src_id as usize];
let out_colls = &self.buffer_out_collections[src_id as usize];
if out
.iter()
.zip(out_colls.iter())
.any(|(&(l, d), &c)| l == label_id && d == dst_id && c == collection_id)
{
return Ok(());
}
// Check for duplicates in dense CSR (collection-aware).
if self.dense_has_edge(src_id, label_id, dst_id, collection_id) {
return Ok(());
}
// Initialize weight tracking on first non-default weight.
if force_weights && !self.has_weights {
self.enable_weights();
}
self.buffer_out[src_id as usize].push((label_id, dst_id));
self.buffer_in[dst_id as usize].push((label_id, src_id));
self.buffer_out_collections[src_id as usize].push(collection_id);
self.buffer_in_collections[dst_id as usize].push(collection_id);
if self.has_weights {
self.buffer_out_weights[src_id as usize].push(weight);
self.buffer_in_weights[dst_id as usize].push(weight);
}
// If this exact `(src, label, dst, collection)` copy was previously
// deleted, un-delete it.
self.deleted_edges
.remove(&(src_id, label_id, dst_id, collection_id));
Ok(())
}
/// Incrementally remove an edge under the unscoped (`""`) collection.
///
/// Kept for callers that operate on a single-collection CSR (NodeDB-Lite
/// keys one `CsrIndex` per collection) or add edges via the collection-less
/// [`Self::add_edge`]. Origin's collection-scoped delete path uses
/// [`Self::remove_edge_in_collection`].
pub fn remove_edge(&mut self, src: &str, label: &str, dst: &str) {
self.remove_edge_in_collection(src, label, dst, "");
}
/// Incrementally remove the `(src, label, dst, collection)` edge.
///
/// Only the copy tagged with `collection` is removed — an identical triple
/// under a different collection is left intact.
pub fn remove_edge_in_collection(
&mut self,
src: &str,
label: &str,
dst: &str,
collection: &str,
) {
let (Some(&src_id), Some(&dst_id)) = (self.node_to_id.get(src), self.node_to_id.get(dst))
else {
return;
};
let Some(&label_id) = self.label_to_id.get(label) else {
return;
};
let Some(&collection_id) = self.collection_to_id.get(collection) else {
return;
};
// Remove from buffer if present (keep parallel buffers in sync).
let out_buf = &self.buffer_out[src_id as usize];
let out_colls = &self.buffer_out_collections[src_id as usize];
if let Some(pos) = out_buf
.iter()
.zip(out_colls.iter())
.position(|(&(l, d), &c)| l == label_id && d == dst_id && c == collection_id)
{
self.buffer_out[src_id as usize].swap_remove(pos);
self.buffer_out_collections[src_id as usize].swap_remove(pos);
if self.has_weights {
self.buffer_out_weights[src_id as usize].swap_remove(pos);
}
}
let in_buf = &self.buffer_in[dst_id as usize];
let in_colls = &self.buffer_in_collections[dst_id as usize];
if let Some(pos) = in_buf
.iter()
.zip(in_colls.iter())
.position(|(&(l, s), &c)| l == label_id && s == src_id && c == collection_id)
{
self.buffer_in[dst_id as usize].swap_remove(pos);
self.buffer_in_collections[dst_id as usize].swap_remove(pos);
if self.has_weights {
self.buffer_in_weights[dst_id as usize].swap_remove(pos);
}
}
// Mark as deleted in dense CSR.
if self.dense_has_edge(src_id, label_id, dst_id, collection_id) {
self.deleted_edges
.insert((src_id, label_id, dst_id, collection_id));
}
}
/// Remove ALL edges touching a node. Returns the number of edges removed.
pub fn remove_node_edges(&mut self, node: &str) -> usize {
let Some(&node_id) = self.node_to_id.get(node) else {
return 0;
};
let mut removed = 0;
// Collect outgoing edges (collection-tagged) then remove reverse
// references. Every collection's copy of an edge is removed — node
// deletion cascades across all collections.
let out_edges = self.dense_iter_out_coll(node_id);
for (label_id, dst_id, coll_id) in &out_edges {
let in_buf = &self.buffer_in[*dst_id as usize];
let in_colls = &self.buffer_in_collections[*dst_id as usize];
if let Some(pos) = in_buf
.iter()
.zip(in_colls.iter())
.position(|(&(l, s), &c)| l == *label_id && s == node_id && c == *coll_id)
{
self.buffer_in[*dst_id as usize].swap_remove(pos);
self.buffer_in_collections[*dst_id as usize].swap_remove(pos);
if self.has_weights {
self.buffer_in_weights[*dst_id as usize].swap_remove(pos);
}
}
self.deleted_edges
.insert((node_id, *label_id, *dst_id, *coll_id));
removed += 1;
}
self.buffer_out[node_id as usize].clear();
self.buffer_out_collections[node_id as usize].clear();
if self.has_weights {
self.buffer_out_weights[node_id as usize].clear();
}
// Collect incoming edges (collection-tagged) then remove reverse
// references.
let in_edges = self.dense_iter_in_coll(node_id);
for (label_id, src_id, coll_id) in &in_edges {
let out_buf = &self.buffer_out[*src_id as usize];
let out_colls = &self.buffer_out_collections[*src_id as usize];
if let Some(pos) = out_buf
.iter()
.zip(out_colls.iter())
.position(|(&(l, d), &c)| l == *label_id && d == node_id && c == *coll_id)
{
self.buffer_out[*src_id as usize].swap_remove(pos);
self.buffer_out_collections[*src_id as usize].swap_remove(pos);
if self.has_weights {
self.buffer_out_weights[*src_id as usize].swap_remove(pos);
}
}
self.deleted_edges
.insert((*src_id, *label_id, node_id, *coll_id));
removed += 1;
}
self.buffer_in[node_id as usize].clear();
self.buffer_in_collections[node_id as usize].clear();
if self.has_weights {
self.buffer_in_weights[node_id as usize].clear();
}
removed
}
/// Remove all edges touching any node whose ID starts with `prefix`.
///
/// Used for tenant purge: `prefix = "{tenant_id}:"` removes all
/// edges belonging to that tenant.
pub fn remove_nodes_with_prefix(&mut self, prefix: &str) {
let matching_nodes: Vec<String> = self
.node_to_id
.keys()
.filter(|k| k.starts_with(prefix))
.cloned()
.collect();
for node in &matching_nodes {
self.remove_node_edges(node);
}
}
}