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
use super::MappedFrozen;
use crate::edge::csr_shared::decode_endpoint_pair;
use crate::edge::{ColdStamps, EdgePosition, HotNbr, Nbr, INVALID_EDGE_ID};
use linkrs_core::types::{EdgeId, Timestamp, VertexId};
impl MappedFrozen {
/// Row length of one vertex. Out-of-range rows report zero.
pub fn row_degree(&self, src_vid: u32) -> usize {
let idx = src_vid as usize;
if idx >= self.rows {
0
} else {
self.degree_at(idx) as usize
}
}
/// Mapped rows inherit the frozen sort order.
pub fn is_row_sorted(&self, _src_vid: u32) -> bool {
true
}
/// Threshold range inside one mapped row, mirroring the heap bisection.
fn threshold_range(
&self,
start: usize,
end: usize,
lower: Option<(u32, i64)>,
upper: Option<(u32, i64)>,
) -> (usize, usize) {
let key_at = |idx: usize| (self.endpoint_at(idx), self.rank_at(idx));
let lo = match lower {
Some((lo_ep, lo_rank)) => {
let mut lo = start;
let mut hi = end;
while lo < hi {
let mid = lo + (hi - lo) / 2;
if key_at(mid) < (lo_ep, lo_rank) {
lo = mid + 1;
} else {
hi = mid;
}
}
lo
}
None => start,
};
let hi = match upper {
Some((hi_ep, hi_rank)) => {
let mut lo = lo;
let mut hi = end;
while lo < hi {
let mid = lo + (hi - lo) / 2;
if key_at(mid) <= (hi_ep, hi_rank) {
lo = mid + 1;
} else {
hi = mid;
}
}
lo
}
None => end,
};
(lo, hi)
}
/// Visit live entries whose `(endpoint, rank)` key falls in the inclusive
/// `[lower, upper]` range. The mapped row is sorted, so the window is
/// bisected before the liveness filter.
pub fn visit_threshold<F>(
&self,
src_vid: u32,
lower: Option<(u32, i64)>,
upper: Option<(u32, i64)>,
mut f: F,
) where
F: FnMut(Nbr) -> bool,
{
let Some((start, end)) = self.row_window(src_vid) else {
return;
};
if start == end {
return;
}
let (lo, hi) = self.threshold_range(start, end, lower, upper);
for idx in lo..hi {
let hot = HotNbr {
endpoint: self.endpoint_at(idx),
rank: self.rank_at(idx),
edge_id: self.edge_id_at(idx),
};
let cold = ColdStamps {
delete_ts: self.delete_at(idx),
};
if cold.is_live() && hot.edge_id != INVALID_EDGE_ID && !f(Nbr::from_parts(hot, cold)) {
return;
}
}
}
/// 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
});
}
/// Timestamp-filtered read of one row.
///
/// Test and offline use; production traversals use `iter_edges_of` or
/// `fill_physical_into` instead of this allocating accessor.
pub fn edges_of(&self, src_vid: u32, ts: Timestamp) -> Vec<Nbr> {
let mut out = Vec::new();
self.fill_visible_into(src_vid, ts, &mut out);
out
}
/// Fill a caller buffer with the timestamp-visible entries of one row.
///
/// Same filtered content as `edges_of` without the per-row allocation.
/// Columns are sliced once per row and decoded in one pass.
pub fn fill_visible_into(&self, src_vid: u32, ts: Timestamp, out: &mut Vec<Nbr>) {
out.clear();
let Some((start, end)) = self.row_window(src_vid) else {
return;
};
if start == end {
return;
}
out.reserve(end - start);
out.extend(self.row_slots(start, end).filter(|nbr| nbr.is_alive_at(ts)));
}
/// First timestamp-visible entry matching an endpoint key: key-range
/// bisection plus in-range timestamp filter, same rule as the heap form.
pub fn get_edge(&self, src_vid: u32, dst: VertexId, ts: Timestamp) -> Option<Nbr> {
let (decoded_endpoint, decoded_rank) = decode_endpoint_pair(dst)?;
let (start, end) = self.row_window(src_vid)?;
let (lo, hi) = self.key_range(start, end, decoded_endpoint, decoded_rank);
(lo..hi).find_map(|idx| {
let nbr = self.slot_at(idx)?;
nbr.is_alive_at(ts).then_some(nbr)
})
}
/// First live entry matching an endpoint key without consulting snapshots.
pub fn get_edge_physical(&self, src_vid: u32, dst: VertexId) -> Option<Nbr> {
let (decoded_endpoint, decoded_rank) = decode_endpoint_pair(dst)?;
let (start, end) = self.row_window(src_vid)?;
let (lo, hi) = self.key_range(start, end, decoded_endpoint, decoded_rank);
(lo..hi).find_map(|idx| {
let hot = self.hot_at(idx)?;
let cold = self.cold_at(idx)?;
if hot.edge_id != INVALID_EDGE_ID && cold.is_live() {
Some(Nbr::from_parts(hot, cold))
} else {
None
}
})
}
/// Every physically stored entry of one row without timestamp filtering.
///
/// Test and offline use; production scans use `fill_physical_into` or
/// the visitor paths instead of this allocating accessor.
pub fn physical_edges_of(&self, src_vid: u32) -> Vec<Nbr> {
let mut out = Vec::new();
self.fill_physical_into(src_vid, &mut out);
out
}
/// Fill a caller buffer with every physically stored entry of one row.
///
/// Columns are sliced once per row and decoded in one pass instead of
/// one bounds-checked scalar read per slot.
pub fn fill_physical_into(&self, src_vid: u32, out: &mut Vec<Nbr>) {
out.clear();
let Some((start, end)) = self.row_window(src_vid) else {
return;
};
if start == end {
return;
}
out.reserve(end - start);
out.extend(self.row_slots(start, end));
}
/// Visit every physically stored entry of one row without allocating.
///
/// Decodes from one slice per column instead of one bounds-checked scalar
/// read per slot.
pub fn visit_physical<F>(&self, src_vid: u32, mut f: F)
where
F: FnMut(Nbr) -> bool,
{
let Some((start, end)) = self.row_window(src_vid) else {
return;
};
if start == end {
return;
}
for nbr in self.row_slots(start, end) {
if !f(nbr) {
return;
}
}
}
/// Visit every physically stored hot half of one row without allocating
/// and without touching the stamp columns.
///
/// Decodes from one slice per topology column instead of one scalar read
/// per slot.
pub fn visit_hot<F>(&self, src_vid: u32, mut f: F)
where
F: FnMut(HotNbr) -> bool,
{
let Some((start, end)) = self.row_window(src_vid) else {
return;
};
if start == end {
return;
}
for hot in self.row_hot_slots(start, end) {
if !f(hot) {
return;
}
}
}
/// Read-only view of one packed slot without mutating state.
pub fn nbr_at_offset(&self, src_vid: u32, offset: i32) -> Option<Nbr> {
if offset < 0 {
return None;
}
let (start, end) = self.row_window(src_vid)?;
if start.saturating_add(offset as usize) >= end {
return None;
}
self.slot_at(start + offset as usize)
}
/// Whether one row holds any physically stored entry.
pub fn has_physical_entries(&self, vid: u32) -> bool {
self.row_degree(vid) > 0
}
/// Whether one row holds `edge_id`.
pub fn primary_contains(&self, src_vid: u32, edge_id: EdgeId) -> bool {
let Some((start, end)) = self.row_window(src_vid) else {
return false;
};
(start..end).any(|idx| self.edge_id_at(idx) == edge_id)
}
/// Read the inline value of one edge by id within its source row.
///
/// Mapped half of the frozen paired traversal: resolve the edge from
/// the topology walk first, then read its value here. Sidecars without
/// a value column still resolve the edge but report a NULL slot.
pub fn bundled_value_by_edge_id(&self, src_vid: u32, edge_id: EdgeId) -> Option<(u64, bool)> {
let (start, end) = self.row_window(src_vid)?;
for idx in start..end {
if self.edge_id_at(idx) == edge_id {
return Some((self.value_at(idx), self.valid_at(idx)));
}
}
None
}
/// Read the inline value of the first physical entry for one endpoint.
///
/// First-match walk mirroring the heap accessor: tombstones are physical
/// entries and may match.
pub fn bundled_value_by_endpoint(&self, src_vid: u32, endpoint: u32) -> Option<(u64, bool)> {
let (start, end) = self.row_window(src_vid)?;
for idx in start..end {
if self.endpoint_at(idx) == endpoint && self.edge_id_at(idx) != INVALID_EDGE_ID {
return Some((self.value_at(idx), self.valid_at(idx)));
}
}
None
}
/// Visit every physically stored entry of one row with its inline value
/// (`None` for NULL slots).
pub fn visit_physical_with_values<F>(&self, src_vid: u32, mut f: F)
where
F: FnMut(Nbr, Option<u64>) -> bool,
{
let Some((start, end)) = self.row_window(src_vid) else {
return;
};
for idx in start..end {
let Some(nbr) = self.slot_at(idx) else {
continue;
};
let raw = self.value_at(idx);
let value = self.valid_at(idx).then_some(raw);
if !f(nbr, value) {
return;
}
}
}
/// Whether any slot holds a valid inline value.
pub fn any_valid_values(&self) -> bool {
let bytes = self.column_bytes(self.columns.validity);
bytes.iter().any(|b| *b != 0)
}
/// Locate the first entry with `edge_id`, returning its packed slot.
pub fn locate_edge(&self, src_vid: u32, edge_id: EdgeId) -> Option<(EdgePosition, Nbr)> {
let (start, end) = self.row_window(src_vid)?;
for (slot, idx) in (start..end).enumerate() {
if self.edge_id_at(idx) == edge_id {
return Some((
EdgePosition::Primary { slot: slot as u32 },
self.slot_at(idx)?,
));
}
}
None
}
/// Physical entry census of one row: `(live, dead, capacity)`.
pub fn vertex_census(&self, vid: u32) -> (usize, usize, usize) {
let Some((start, end)) = self.row_window(vid) else {
return (0, 0, 0);
};
let mut live = 0usize;
let mut dead = 0usize;
for idx in start..end {
let cold = ColdStamps {
delete_ts: self.delete_at(idx),
};
if cold.is_live() {
live += 1;
} else {
dead += 1;
}
}
(live, dead, end - start)
}
/// Mapped groups are never reclaimed; maintenance skips them like frozen.
pub fn reclaimable_count(&self, _vid: u32, _cutoff: Timestamp) -> usize {
0
}
}