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
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
//! Routed physical read helpers over the shard set.
//!
//! Reads resolve rows through `route`, never creating groups: missing groups
//! read as empty.
//!
//! Batch scans group consecutive same-group vertices into runs so each run
//! borrows its shard once, then walks rows through the `CsrVariant`
//! single-row entries. Dispatch itself stays converged in the variant: these
//! callers never match on the form, so adding a form only touches the
//! variant read entries.
use linkrs_core::types::EdgeId;
use super::super::{HotNbr, MutableCsrTrait, Nbr};
use super::CsrShardSet;
impl CsrShardSet {
/// Whether the primary row of one vertex holds `edge_id`.
pub fn primary_contains(&self, src_vid: u32, edge_id: EdgeId) -> bool {
let Some((gid, local)) = self.route(src_vid) else {
return false;
};
self.shards.get(&gid).is_some_and(|shard| {
if let Some(mapped) = &shard.mapped {
return mapped.primary_contains(local, edge_id);
}
shard.variant.primary_contains(local, edge_id)
})
}
/// Owner group of one global vertex id without creating groups.
///
/// Shared row-location entry point for point lookups, adjacency batches
/// and full scans: every read path resolves rows through this routing
/// instead of duplicating the group arithmetic.
pub fn group_of(&self, vid: u32) -> Option<usize> {
self.route(vid).map(|(gid, _)| gid)
}
/// Visit every physically stored entry of one vertex without allocating.
pub fn visit_physical<F>(&self, src_vid: u32, mut f: F)
where
F: FnMut(Nbr) -> bool,
{
let Some((gid, local)) = self.route(src_vid) else {
return;
};
if let Some(shard) = self.shards.get(&gid) {
let mut visited = 0usize;
let wrapped = |nbr: Nbr| {
visited += 1;
f(nbr)
};
if let Some(mapped) = &shard.mapped {
mapped.visit_physical(local, wrapped);
} else {
shard.variant.visit_physical(local, wrapped);
}
if visited > super::super::mutable_csr::row::HIGH_DEGREE_LIVE_THRESHOLD {
log::warn!(
"slow topology traversal at row {} (group {} local {}): degree {} exceeds wide threshold",
src_vid,
gid,
local,
visited
);
}
}
}
/// Visit every physically stored hot half of one vertex without
/// allocating and without touching the stamp lines.
///
/// Hot-only counterpart of [`Self::visit_physical`] for traversals that
/// resolve visibility through the version authority by `edge_id`.
pub fn visit_hot<F>(&self, src_vid: u32, f: F)
where
F: FnMut(HotNbr) -> bool,
{
let Some((gid, local)) = self.route(src_vid) else {
return;
};
if let Some(shard) = self.shards.get(&gid) {
if let Some(mapped) = &shard.mapped {
mapped.visit_hot(local, f);
} else {
shard.variant.visit_hot(local, f);
}
}
}
/// Fill a caller buffer with every physically stored entry of one vertex.
///
/// Same content as the allocating trait accessor, without the per-vertex
/// allocation. Missing groups read as empty and never create groups.
pub fn fill_physical_into(&self, src_vid: u32, out: &mut Vec<Nbr>) {
let Some((gid, local)) = self.route(src_vid) else {
out.clear();
return;
};
match self.shards.get(&gid) {
Some(shard) => {
if let Some(mapped) = &shard.mapped {
mapped.fill_physical_into(local, out);
} else {
shard.variant.fill_physical_into(local, out);
}
}
None => out.clear(),
}
}
/// Fill one shared buffer with the physical entries of many vertices.
///
/// Records one start offset per vertex plus a trailing end offset, so
/// `out[offsets[i]..offsets[i + 1]]` is vertex `vids[i]` in order. One
/// pass over the vertices, no per-vertex allocation; missing groups and
/// empty rows contribute empty slices. Read paths never create groups.
///
/// Dispatch stays converged in `CsrVariant`: each run borrows its shard
/// once (one map lookup per run), then walks rows through the variant
/// single-row entry. The row loop pays one variant match per row, which
/// is noise next to the per-row routing plus the row walk itself; in
/// return there are no per-form arms to keep in sync here, so adding a
/// form only touches the variant read entries.
pub fn fill_physical_batch_into(
&self,
vids: &[u32],
out: &mut Vec<Nbr>,
offsets: &mut Vec<usize>,
) {
out.clear();
offsets.clear();
offsets.reserve(vids.len() + 1);
let mut idx = 0usize;
while idx < vids.len() {
let Some((gid, _)) = self.route(vids[idx]) else {
offsets.push(out.len());
idx += 1;
continue;
};
let mut run_end = idx + 1;
while run_end < vids.len() {
match self.route(vids[run_end]) {
Some((next_gid, _)) if next_gid == gid => run_end += 1,
_ => break,
}
}
let Some(shard) = self.shards.get(&gid) else {
for _ in idx..run_end {
offsets.push(out.len());
}
idx = run_end;
continue;
};
let variant = &shard.variant;
let mapped = shard.mapped.clone();
for vid in &vids[idx..run_end] {
offsets.push(out.len());
let (_, local) = self.route(*vid).expect("run shares one group");
if let Some(mapped) = &mapped {
mapped.visit_physical(local, |nbr| {
out.push(nbr);
true
});
} else {
variant.visit_physical(local, |nbr| {
out.push(nbr);
true
});
}
}
idx = run_end;
}
offsets.push(out.len());
}
/// Fill one shared buffer with paired topology plus inline values.
///
/// Batched counterpart of the single-row paired fill: same offsets
/// contract as the topology batch above, but each entry carries its
/// inline value. Non-bundled forms fill with `None` values through the
/// same walk. Like the topology batch, each run borrows its shard once
/// and walks rows through the variant paired entry, so there are no
/// per-form arms to keep in sync here either.
pub fn fill_physical_with_values_batch_into(
&self,
vids: &[u32],
out: &mut Vec<(Nbr, Option<u64>)>,
offsets: &mut Vec<usize>,
) {
out.clear();
offsets.clear();
offsets.reserve(vids.len() + 1);
let mut idx = 0usize;
while idx < vids.len() {
let Some((gid, _)) = self.route(vids[idx]) else {
offsets.push(out.len());
idx += 1;
continue;
};
let mut run_end = idx + 1;
while run_end < vids.len() {
match self.route(vids[run_end]) {
Some((next_gid, _)) if next_gid == gid => run_end += 1,
_ => break,
}
}
let Some(shard) = self.shards.get(&gid) else {
for _ in idx..run_end {
offsets.push(out.len());
}
idx = run_end;
continue;
};
let variant = &shard.variant;
let mapped = shard.mapped.clone();
for vid in &vids[idx..run_end] {
offsets.push(out.len());
let (_, local) = self.route(*vid).expect("run shares one group");
if let Some(mapped) = &mapped {
mapped.visit_physical_with_values(local, |nbr, value| {
out.push((nbr, value));
true
});
} else {
variant.visit_physical_with_values(local, |nbr, value| {
out.push((nbr, value));
true
});
}
}
idx = run_end;
}
offsets.push(out.len());
}
/// Whether the live entries of one row arrive in key order.
///
/// Only frozen and single-slot rows promise order and may use bisection;
/// other variants report an observation that is memory-only, rebuilt on
/// load and never cached across restarts. Planning paths prefer
/// `should_use_bisection` below so the promise plus the live observation
/// stay in one place. Frozen groups serve the same sorted order from the
/// heap and from a derived mapping alike.
pub fn is_row_sorted(&self, src_vid: u32) -> bool {
let Some((gid, local)) = self.route(src_vid) else {
return true;
};
self.shards
.get(&gid)
.is_some_and(|shard| shard.variant.is_row_sorted(local))
}
/// Whether a range scan should bisect this row.
///
/// Central plan selection combining the order promise with the live
/// sorted observation. Missing groups read as sorted empty. The result
/// is memory-only and must never be cached across restarts.
pub fn should_use_bisection(&self, src_vid: u32) -> bool {
let Some((gid, local)) = self.route(src_vid) else {
return true;
};
self.shards
.get(&gid)
.is_some_and(|shard| shard.variant.should_use_bisection(local))
}
/// Sort one row on the maintenance path. Positions for the row go stale.
pub fn sort_row(&mut self, src_vid: u32) -> bool {
let Some((gid, local)) = self.route(src_vid) else {
return false;
};
self.shards
.get_mut(&gid)
.is_some_and(|shard| shard.variant.sort_row(local))
}
/// Visit live entries of one row whose key falls in the inclusive range.
///
/// Pure and bundled rows ignore the rank halves by contract: callers
/// pass endpoint intervals there.
pub fn visit_threshold<F>(
&self,
src_vid: u32,
lower: Option<(u32, i64)>,
upper: Option<(u32, i64)>,
f: F,
) where
F: FnMut(Nbr) -> bool,
{
let Some((gid, local)) = self.route(src_vid) else {
return;
};
if let Some(shard) = self.shards.get(&gid) {
if let Some(mapped) = &shard.mapped {
mapped.visit_threshold(local, lower, upper, f);
} else {
shard.variant.visit_threshold(local, lower, upper, f);
}
}
}
/// Fill a caller buffer with the same range content as `visit_threshold`.
pub fn fill_threshold_into(
&self,
src_vid: u32,
lower: Option<(u32, i64)>,
upper: Option<(u32, i64)>,
out: &mut Vec<Nbr>,
) {
out.clear();
self.visit_threshold(src_vid, lower, upper, |nbr| {
out.push(nbr);
true
});
}
/// Visit entries whose endpoint falls in the inclusive endpoint range.
///
/// Endpoint-only counterpart of `visit_threshold` sharing one interval
/// across forms. Missing groups read as empty and never create groups.
pub fn visit_threshold_endpoint_only<F>(
&self,
src_vid: u32,
lower: Option<u32>,
upper: Option<u32>,
f: F,
) where
F: FnMut(Nbr) -> bool,
{
let Some((gid, local)) = self.route(src_vid) else {
return;
};
if let Some(shard) = self.shards.get(&gid) {
if let Some(mapped) = &shard.mapped {
mapped.visit_threshold(
local,
lower.map(|endpoint| (endpoint, i64::MIN)),
upper.map(|endpoint| (endpoint, i64::MAX)),
f,
);
} else {
shard
.variant
.visit_threshold_endpoint_only(local, lower, upper, f);
}
}
}
/// Fill a caller buffer with the same endpoint-range content.
pub fn fill_threshold_endpoint_only_into(
&self,
src_vid: u32,
lower: Option<u32>,
upper: Option<u32>,
out: &mut Vec<Nbr>,
) {
out.clear();
self.visit_threshold_endpoint_only(src_vid, lower, upper, |nbr| {
out.push(nbr);
true
});
}
}