Skip to main content

linkrs_storage/edge/mutable_csr/
read.rs

1use super::super::csr_shared::decode_endpoint_pair;
2use super::super::{ColdStamps, EdgeId, HotNbr, Nbr, Timestamp, VertexId, INVALID_EDGE_ID};
3use super::write::EdgePosition;
4use super::MutableCsr;
5
6impl MutableCsr {
7    /// Clamped primary window of one row as `(start, end)` list indices.
8    ///
9    /// Single bounds check per row: every row scan below goes through this
10    /// instead of probing each slot with `get`, so per-slot branches and
11    /// `Option` unwraps disappear from the hot paths.
12    pub(super) fn primary_window(&self, src_idx: usize) -> (usize, usize) {
13        if src_idx >= self.vertex_capacity() {
14            return (0, 0);
15        }
16        let start = self.rows.adj_offsets[src_idx] as usize;
17        let end = start
18            .saturating_add(self.rows.degrees[src_idx] as usize)
19            .min(self.hot_list.len())
20            .min(self.cold_list.len());
21        (start.min(end), end)
22    }
23
24    /// Primary hot slice of one row; empty when the row is missing or empty.
25    pub(super) fn primary_hot(&self, src_idx: usize) -> &[HotNbr] {
26        let (start, end) = self.primary_window(src_idx);
27        &self.hot_list[start..end]
28    }
29
30    /// Primary hot plus cold slices of one row, locked to the same window.
31    pub(super) fn primary_pair(&self, src_idx: usize) -> (&[HotNbr], &[ColdStamps]) {
32        let (start, end) = self.primary_window(src_idx);
33        (&self.hot_list[start..end], &self.cold_list[start..end])
34    }
35
36    /// Whether the primary block of one row arrives in key order.
37    ///
38    /// Ground-truth order probe over the primary window: threshold scans
39    /// consult the cached per-row flag instead of paying this walk on every
40    /// query. After a maintenance sort the primary block is ordered while
41    /// overflow chunks stay in insertion order as the unsorted suffix.
42    pub fn is_primary_sorted(&self, src_vid: u32) -> bool {
43        let src_idx = src_vid as usize;
44        if src_idx >= self.vertex_capacity() {
45            return true;
46        }
47        let (hot, _) = self.primary_pair(src_idx);
48        hot.windows(2).all(|w| {
49            (w[0].endpoint, w[0].rank, w[0].edge_id.0) <= (w[1].endpoint, w[1].rank, w[1].edge_id.0)
50        })
51    }
52
53    /// Whether the live entries of one row arrive in key order.
54    ///
55    /// Mutable rows are insertion-ordered, so this usually reports false on
56    /// multi-edge rows and true on empty or single-entry rows. Only frozen,
57    /// mapped and single-slot rows promise order and may use bisection;
58    /// this observation must not be cached across restarts (the underlying
59    /// flag is memory-only and rebuilt on load). Rows with an unsorted
60    /// overflow suffix report false even when the primary prefix is sorted;
61    /// threshold scans still bisect the prefix via `is_primary_sorted`.
62    pub fn is_row_sorted(&self, src_vid: u32) -> bool {
63        let src_idx = src_vid as usize;
64        if src_idx >= self.vertex_capacity() {
65            return true;
66        }
67        let mut last: Option<(u32, i64)> = None;
68        let mut ordered = true;
69        self.visit_physical(src_vid, |nbr| {
70            if nbr.delete_ts != Timestamp::MAX {
71                return true;
72            }
73            let key = (nbr.endpoint, nbr.rank);
74            if let Some(prev) = last {
75                if key < prev {
76                    ordered = false;
77                    return false;
78                }
79            }
80            last = Some(key);
81            true
82        });
83        ordered
84    }
85
86    /// Assembled slot copy at a wide-row index position.
87    ///
88    /// Row-relative addressing shared by the endpoint location fast paths:
89    /// primary positions index from the row start, overflow positions address
90    /// a chunk and a slot inside it. Returns `None` for out-of-range
91    /// positions so stale entries fall back to the row scan.
92    fn slot_at_position(
93        &self,
94        src_vid: u32,
95        src_idx: usize,
96        position: EdgePosition,
97    ) -> Option<Nbr> {
98        match position {
99            EdgePosition::Primary { slot } => {
100                let base = *self.rows.adj_offsets.get(src_idx)? as usize;
101                self.slot_at(base + slot as usize)
102            }
103            EdgePosition::Overflow { chunk, slot } => self
104                .overflow_chunks
105                .get(src_vid)?
106                .get(chunk as usize)?
107                .slot_at(slot as usize),
108        }
109    }
110
111    /// Read-only view of one primary slot without mutating state.
112    ///
113    /// Used to verify that a caller-supplied offset still addresses the
114    /// expected edge before a destructive offset write runs.
115    pub fn nbr_at_offset(&self, src_vid: u32, offset: i32) -> Option<Nbr> {
116        if offset < 0 {
117            return None;
118        }
119        let src_idx = src_vid as usize;
120        if src_idx >= self.vertex_capacity() || self.rows.primary_capacities[src_idx] == 0 {
121            return None;
122        }
123        if offset as usize >= self.rows.degrees[src_idx] as usize {
124            return None;
125        }
126        let idx = self.rows.adj_offsets[src_idx] as usize + offset as usize;
127        self.slot_at(idx)
128    }
129
130    /// Locate the first live edge by endpoint without consulting snapshots.
131    ///
132    /// Physical addressing only; tombstoned and gap slots are skipped.
133    /// Snapshot visibility is decided by the version authority above this
134    /// layer, while tombstone access uses the full physical views.
135    ///
136    /// Wide rows answer from the endpoint location index: a present key
137    /// addresses its slot directly, an absent key returns without scanning.
138    /// Narrow rows without an index fall through to the linear walk.
139    pub fn get_edge_physical(&self, src_vid: u32, dst: VertexId) -> Option<Nbr> {
140        use super::super::INVALID_EDGE_ID;
141        let (decoded_endpoint, decoded_rank) = decode_endpoint_pair(dst)?;
142        let src_idx = src_vid as usize;
143        if src_idx >= self.vertex_capacity() {
144            return None;
145        }
146        // Narrow rows never carry a live-set index; skip the HashMap
147        // probe and go straight to the row scan.
148        if src_idx < self.live_counts.len()
149            && self.live_counts[src_idx] as usize <= super::live_set::LIVE_SET_WIDTH_BOUND
150        {
151            return self.get_edge_physical_via_scan(src_idx, decoded_endpoint, decoded_rank);
152        }
153        if let Some(set) = self.live_sets.get(&src_vid) {
154            let position = set.position(&(decoded_endpoint, decoded_rank))?;
155            if let Some(nbr) = self.slot_at_position(src_vid, src_idx, position) {
156                if nbr.endpoint == decoded_endpoint
157                    && nbr.rank == decoded_rank
158                    && nbr.edge_id != INVALID_EDGE_ID
159                    && nbr.delete_ts == Timestamp::MAX
160                {
161                    return Some(nbr);
162                }
163            }
164        }
165        self.get_edge_physical_via_scan(src_idx, decoded_endpoint, decoded_rank)
166    }
167
168    /// Primary-then-overflow physical scan without index dispatch.
169    fn get_edge_physical_via_scan(&self, src_idx: usize, endpoint: u32, rank: i64) -> Option<Nbr> {
170        self.scan_row_for_key(src_idx, endpoint, rank, |hot, cold| {
171            hot.edge_id != INVALID_EDGE_ID && cold.is_live()
172        })
173    }
174
175    /// Every physically stored entry of one vertex without timestamp filtering.
176    ///
177    /// Visibility is decided by the version authority above this layer.
178    /// Test and offline use; production scans use [`Self::fill_physical_into`]
179    /// or the visitor paths instead of this allocating accessor.
180    pub fn physical_edges_of(&self, src_vid: u32) -> Vec<Nbr> {
181        let mut out = Vec::new();
182        self.fill_physical_into(src_vid, &mut out);
183        out
184    }
185
186    /// Fill a caller buffer with every physically stored entry of one vertex.
187    ///
188    /// Same content as the allocating accessor above, without the per-vertex
189    /// allocation. Batch scans reuse one buffer across vertices and slice it
190    /// per vertex instead of collecting one vector per vertex. Gap sentinels
191    /// are excluded; tombstones are included for reclaim and audit paths.
192    pub fn fill_physical_into(&self, src_vid: u32, out: &mut Vec<Nbr>) {
193        out.clear();
194        let src_idx = src_vid as usize;
195        if src_idx >= self.vertex_capacity() {
196            return;
197        }
198        let (hot, cold) = self.primary_pair(src_idx);
199        out.reserve(hot.len());
200        out.extend(
201            hot.iter()
202                .zip(cold.iter())
203                .map(|(h, c)| Nbr::from_parts(*h, *c))
204                .filter(|nbr| nbr.edge_id != INVALID_EDGE_ID),
205        );
206        if let Some(single) = self.overflow_chunks.single_chunk(src_vid) {
207            out.reserve(single.len());
208            for i in 0..single.len() {
209                if let Some(nbr) = single.slot_at(i) {
210                    out.push(nbr);
211                }
212            }
213            return;
214        }
215        if let Some(chunks) = self.overflow_chunks.get(src_vid) {
216            for chunk in chunks {
217                out.reserve(chunk.len());
218                for i in 0..chunk.len() {
219                    if let Some(nbr) = chunk.slot_at(i) {
220                        out.push(nbr);
221                    }
222                }
223            }
224        }
225    }
226
227    /// Fill a caller buffer with the physical entries of many vertices.
228    ///
229    /// Batched counterpart of [`Self::fill_physical_into`] for fan-out scans:
230    /// one clear plus one reservation serves every listed vertex, and `starts`
231    /// records the boundary of each vertex so the caller slices per vertex
232    /// without a second lookup. Gap sentinels are excluded; tombstones are
233    /// included, matching the single-vertex walk exactly.
234    pub fn fill_physical_into_batched(
235        &self,
236        src_vids: &[u32],
237        out: &mut Vec<Nbr>,
238        starts: &mut Vec<usize>,
239    ) {
240        out.clear();
241        starts.clear();
242        starts.reserve(src_vids.len().saturating_add(1));
243        let mut total = 0usize;
244        for vid in src_vids {
245            let src_idx = *vid as usize;
246            if src_idx < self.vertex_capacity() {
247                let (start, end) = self.primary_window(src_idx);
248                total = total.saturating_add(end.saturating_sub(start));
249                if let Some(single) = self.overflow_chunks.single_chunk(*vid) {
250                    total = total.saturating_add(single.len());
251                } else if let Some(chunks) = self.overflow_chunks.get(*vid) {
252                    for chunk in chunks {
253                        total = total.saturating_add(chunk.len());
254                    }
255                }
256            }
257        }
258        out.reserve(total);
259        for vid in src_vids {
260            starts.push(out.len());
261            let src_idx = *vid as usize;
262            if src_idx >= self.vertex_capacity() {
263                continue;
264            }
265            let (hot, cold) = self.primary_pair(src_idx);
266            out.extend(
267                hot.iter()
268                    .zip(cold.iter())
269                    .map(|(h, c)| Nbr::from_parts(*h, *c))
270                    .filter(|nbr| nbr.edge_id != INVALID_EDGE_ID),
271            );
272            if let Some(single) = self.overflow_chunks.single_chunk(*vid) {
273                for i in 0..single.len() {
274                    if let Some(nbr) = single.slot_at(i) {
275                        out.push(nbr);
276                    }
277                }
278                continue;
279            }
280            if let Some(chunks) = self.overflow_chunks.get(*vid) {
281                for chunk in chunks {
282                    for i in 0..chunk.len() {
283                        if let Some(nbr) = chunk.slot_at(i) {
284                            out.push(nbr);
285                        }
286                    }
287                }
288            }
289        }
290        starts.push(out.len());
291    }
292
293    /// Visit every physically stored hot half of many vertices without
294    /// allocating and without touching the stamp lines.
295    ///
296    /// Batched counterpart of [`Self::visit_hot`] for fan-out traversals that
297    /// resolve visibility through the version authority by `edge_id`.
298    pub fn visit_hot_batched<F>(&self, src_vids: &[u32], mut f: F)
299    where
300        F: FnMut(u32, HotNbr) -> bool,
301    {
302        for vid in src_vids {
303            let src_idx = *vid as usize;
304            if src_idx >= self.vertex_capacity() {
305                continue;
306            }
307            for h in self.primary_hot(src_idx) {
308                if h.edge_id == INVALID_EDGE_ID {
309                    continue;
310                }
311                if !f(*vid, *h) {
312                    return;
313                }
314            }
315            if let Some(single) = self.overflow_chunks.single_chunk(*vid) {
316                for h in single.hot_slice() {
317                    if h.edge_id == INVALID_EDGE_ID {
318                        continue;
319                    }
320                    if !f(*vid, *h) {
321                        return;
322                    }
323                }
324                continue;
325            }
326            if let Some(chunks) = self.overflow_chunks.get(*vid) {
327                for chunk in chunks {
328                    for h in chunk.hot_slice() {
329                        if h.edge_id == INVALID_EDGE_ID {
330                            continue;
331                        }
332                        if !f(*vid, *h) {
333                            return;
334                        }
335                    }
336                }
337            }
338        }
339    }
340
341    /// Visit every physically stored hot half of one vertex without
342    /// allocating and without touching the stamp lines.
343    ///
344    /// Hot-only counterpart of [`Self::visit_physical`] for traversals that
345    /// resolve visibility through the version authority by `edge_id`. Stamps
346    /// stay out of cache on this walk. Consolidated rows read their single
347    /// overflow block without the chain loop. Gap sentinels are skipped;
348    /// tombstones are visited.
349    pub fn visit_hot<F>(&self, src_vid: u32, mut f: F)
350    where
351        F: FnMut(HotNbr) -> bool,
352    {
353        let src_idx = src_vid as usize;
354        if src_idx >= self.vertex_capacity() {
355            return;
356        }
357        for h in self.primary_hot(src_idx) {
358            if h.edge_id == INVALID_EDGE_ID {
359                continue;
360            }
361            if !f(*h) {
362                return;
363            }
364        }
365        if let Some(single) = self.overflow_chunks.single_chunk(src_vid) {
366            for h in single.hot_slice() {
367                if h.edge_id == INVALID_EDGE_ID {
368                    continue;
369                }
370                if !f(*h) {
371                    return;
372                }
373            }
374            return;
375        }
376        if let Some(chunks) = self.overflow_chunks.get(src_vid) {
377            for chunk in chunks {
378                for h in chunk.hot_slice() {
379                    if h.edge_id == INVALID_EDGE_ID {
380                        continue;
381                    }
382                    if !f(*h) {
383                        return;
384                    }
385                }
386            }
387        }
388    }
389
390    /// Visit every physically stored entry of one vertex without allocating.
391    ///
392    /// The visitor returns false to stop early. Visibility is decided by the
393    /// version authority above this layer. Consolidated rows read their
394    /// single overflow block without the chain loop. Gap sentinels are
395    /// skipped; tombstones are visited.
396    pub fn visit_physical<F>(&self, src_vid: u32, mut f: F)
397    where
398        F: FnMut(Nbr) -> bool,
399    {
400        let src_idx = src_vid as usize;
401        if src_idx >= self.vertex_capacity() {
402            return;
403        }
404        let (hot, cold) = self.primary_pair(src_idx);
405        for (h, c) in hot.iter().zip(cold.iter()) {
406            if h.edge_id == INVALID_EDGE_ID {
407                continue;
408            }
409            if !f(Nbr::from_parts(*h, *c)) {
410                return;
411            }
412        }
413        if let Some(single) = self.overflow_chunks.single_chunk(src_vid) {
414            for i in 0..single.len() {
415                if let Some(nbr) = single.slot_at(i) {
416                    if !f(nbr) {
417                        return;
418                    }
419                }
420            }
421            return;
422        }
423        if let Some(chunks) = self.overflow_chunks.get(src_vid) {
424            for chunk in chunks {
425                for i in 0..chunk.len() {
426                    if let Some(nbr) = chunk.slot_at(i) {
427                        if !f(nbr) {
428                            return;
429                        }
430                    }
431                }
432            }
433        }
434    }
435
436    /// Visit every physically stored entry with its row position.
437    ///
438    /// Same walk as [`Self::visit_physical`] but the visitor also receives
439    /// the [`EdgePosition`] of each entry, so a later delete or revert can
440    /// address the slot directly instead of rescanning the row. Positions
441    /// stay valid only until the next compaction, rebalance, repack or
442    /// removal of this row; positional writes revalidate the edge id. Gap
443    /// sentinels are skipped; tombstones are visited.
444    pub fn visit_physical_with_position<F>(&self, src_vid: u32, mut f: F)
445    where
446        F: FnMut(EdgePosition, Nbr) -> bool,
447    {
448        let src_idx = src_vid as usize;
449        if src_idx >= self.vertex_capacity() {
450            return;
451        }
452        let (hot, cold) = self.primary_pair(src_idx);
453        for (i, (h, c)) in hot.iter().zip(cold.iter()).enumerate() {
454            if h.edge_id == INVALID_EDGE_ID {
455                continue;
456            }
457            if !f(
458                EdgePosition::Primary { slot: i as u32 },
459                Nbr::from_parts(*h, *c),
460            ) {
461                return;
462            }
463        }
464        if let Some(chunks) = self.overflow_chunks.get(src_vid) {
465            for (chunk_idx, chunk) in chunks.iter().enumerate() {
466                for slot_idx in 0..chunk.len() {
467                    if let Some(nbr) = chunk.slot_at(slot_idx) {
468                        if !f(
469                            EdgePosition::Overflow {
470                                chunk: chunk_idx as u32,
471                                slot: slot_idx as u32,
472                            },
473                            nbr,
474                        ) {
475                            return;
476                        }
477                    }
478                }
479            }
480        }
481    }
482
483    /// Locate the first entry with `edge_id`, returning its position and copy.
484    ///
485    /// Single scan shared by delete and revert callers that then act on the
486    /// position directly.
487    pub fn locate_edge(&self, src_vid: u32, edge_id: EdgeId) -> Option<(EdgePosition, Nbr)> {
488        let mut found = None;
489        self.visit_physical_with_position(src_vid, |position, nbr| {
490            if nbr.edge_id == edge_id {
491                found = Some((position, nbr));
492                false
493            } else {
494                true
495            }
496        });
497        found
498    }
499
500    /// Whether the primary row of one vertex holds `edge_id`.
501    ///
502    /// Used to distinguish a stale offset (edge lives in primary at a
503    /// different offset, must fail) from an overflow row (no offset can
504    /// address it, may fall back to the edge-id path).
505    pub fn primary_contains(&self, src_vid: u32, edge_id: EdgeId) -> bool {
506        let src_idx = src_vid as usize;
507        if src_idx >= self.vertex_capacity() {
508            return false;
509        }
510        self.primary_hot(src_idx)
511            .iter()
512            .any(|hot| hot.edge_id == edge_id)
513    }
514
515    /// Whether one vertex holds any physically stored entry.
516    pub fn has_physical_entries(&self, vid: u32) -> bool {
517        let idx = vid as usize;
518        if idx >= self.vertex_capacity() {
519            return false;
520        }
521        if self.rows.degrees[idx] > 0 {
522            return true;
523        }
524        self.overflow_chunks
525            .get(vid)
526            .is_some_and(|chunks| chunks.iter().any(|c| !c.is_empty()))
527    }
528
529    /// Get edges of a vertex at a given timestamp.
530    ///
531    /// UNFILTERED single-predicate read for tests and offline tools only:
532    /// filters by row delete stamp without consulting the version authority
533    /// (no creation-stamp or authority verdict). Production traversals must
534    /// use the visitor or caller-buffer fill paths with the authority gate.
535    pub fn edges_of(&self, src_vid: u32, ts: Timestamp) -> Vec<Nbr> {
536        let src_idx = src_vid as usize;
537        if src_idx >= self.vertex_capacity() {
538            return Vec::new();
539        }
540
541        let (hot, cold) = self.primary_pair(src_idx);
542        let overflow_len = self
543            .overflow_chunks
544            .get(src_vid)
545            .map(|chunks| chunks.iter().map(|chunk| chunk.len()).sum::<usize>())
546            .unwrap_or(0);
547        let mut result = Vec::with_capacity(hot.len() + overflow_len);
548
549        result.extend(hot.iter().zip(cold.iter()).filter_map(|(h, c)| {
550            let nbr = Nbr::from_parts(*h, *c);
551            nbr.is_alive_at(ts).then_some(nbr)
552        }));
553
554        if let Some(chunks) = self.overflow_chunks.get(src_vid) {
555            for chunk in chunks {
556                for i in 0..chunk.len() {
557                    if let Some(nbr) = chunk.slot_at(i) {
558                        if nbr.is_alive_at(ts) {
559                            result.push(nbr);
560                        }
561                    }
562                }
563            }
564        }
565
566        result
567    }
568
569    /// Get a specific edge
570    ///
571    /// UNFILTERED single-predicate read for tests and offline tools only.
572    /// Wide rows consult the endpoint location index first: a present key
573    /// addresses its slot directly when the live entry covers `ts`. Absent
574    /// keys at the maximum timestamp return without scanning; any other
575    /// timestamp falls through to the row walk so historically visible
576    /// tombstoned versions are still found.
577    pub fn get_edge(&self, src_vid: u32, dst: VertexId, ts: Timestamp) -> Option<Nbr> {
578        let (decoded_endpoint, decoded_rank) = decode_endpoint_pair(dst)?;
579        let src_idx = src_vid as usize;
580        if src_idx >= self.vertex_capacity() {
581            return None;
582        }
583
584        // Narrow rows never carry a live-set index, so the HashMap lookup
585        // is guaranteed to miss.  Skip it and go straight to the row scan
586        // to save one hash probe on the hot path.
587        if src_idx < self.live_counts.len()
588            && self.live_counts[src_idx] as usize <= super::live_set::LIVE_SET_WIDTH_BOUND
589        {
590            return self.get_edge_via_scan(src_idx, decoded_endpoint, decoded_rank, ts);
591        }
592
593        if let Some(set) = self.live_sets.get(&src_vid) {
594            match set.position(&(decoded_endpoint, decoded_rank)) {
595                Some(position) => {
596                    if let Some(nbr) = self.slot_at_position(src_vid, src_idx, position) {
597                        if nbr.endpoint == decoded_endpoint
598                            && nbr.rank == decoded_rank
599                            && nbr.delete_ts == Timestamp::MAX
600                        {
601                            return Some(nbr);
602                        }
603                    }
604                }
605                None if ts == Timestamp::MAX => return None,
606                None => {}
607            }
608        }
609        self.get_edge_via_scan(src_idx, decoded_endpoint, decoded_rank, ts)
610    }
611
612    /// Primary-then-overflow scan for one key without index dispatch.
613    ///
614    /// Extracted so the narrow-row fast path in [`Self::get_edge`] can
615    /// skip the live-set probe and jump directly here.
616    fn get_edge_via_scan(
617        &self,
618        src_idx: usize,
619        endpoint: u32,
620        rank: i64,
621        ts: Timestamp,
622    ) -> Option<Nbr> {
623        self.scan_row_for_key(src_idx, endpoint, rank, |hot, cold| {
624            Nbr::from_parts(hot, cold).is_alive_at(ts)
625        })
626    }
627
628    /// Shared primary-then-overflow key scan behind the point lookups.
629    ///
630    /// Single traversal for the physical and timestamped queries; only the
631    /// acceptance predicate differs, so the consolidated-row fast path and
632    /// the chain fallback cannot drift apart.
633    fn scan_row_for_key(
634        &self,
635        src_idx: usize,
636        endpoint: u32,
637        rank: i64,
638        mut accept: impl FnMut(HotNbr, ColdStamps) -> bool,
639    ) -> Option<Nbr> {
640        let (hot, cold) = self.primary_pair(src_idx);
641        for (h, c) in hot.iter().zip(cold.iter()) {
642            if h.endpoint == endpoint && h.rank == rank && accept(*h, *c) {
643                return Some(Nbr::from_parts(*h, *c));
644            }
645        }
646        let src_vid = src_idx as u32;
647        if let Some(single) = self.overflow_chunks.single_chunk(src_vid) {
648            for i in 0..single.len() {
649                if let Some(nbr) = single.slot_at(i) {
650                    if nbr.endpoint == endpoint && nbr.rank == rank && accept(nbr.hot(), nbr.cold())
651                    {
652                        return Some(nbr);
653                    }
654                }
655            }
656            return None;
657        }
658        if let Some(chunks) = self.overflow_chunks.get(src_vid) {
659            for chunk in chunks {
660                for i in 0..chunk.len() {
661                    if let Some(nbr) = chunk.slot_at(i) {
662                        if nbr.endpoint == endpoint
663                            && nbr.rank == rank
664                            && accept(nbr.hot(), nbr.cold())
665                        {
666                            return Some(nbr);
667                        }
668                    }
669                }
670            }
671        }
672        None
673    }
674
675    /// Visit live entries whose `(endpoint, rank)` key falls in the inclusive
676    /// `[lower, upper]` range (`None` means unbounded).
677    ///
678    /// Sorted primary prefixes bisect to the key window and filter for
679    /// liveness inside it, then the overflow suffix always scans linearly.
680    /// Unsorted primaries scan linearly with the same key predicate. The
681    /// prefix probe is the cached per-row order flag, so hybrid rows (sorted
682    /// primary plus unsorted overflow) still bisect the prefix without
683    /// paying a full sortedness walk on every query. Only live entries
684    /// (`delete_ts == MAX`) are visited; snapshot visibility stays above.
685    pub fn visit_threshold<F>(
686        &self,
687        src_vid: u32,
688        lower: Option<(u32, i64)>,
689        upper: Option<(u32, i64)>,
690        mut f: F,
691    ) where
692        F: FnMut(Nbr) -> bool,
693    {
694        let src_idx = src_vid as usize;
695        if src_idx >= self.vertex_capacity() {
696            return;
697        }
698        let in_range = |endpoint: u32, rank: i64| -> bool {
699            if let Some((lo_ep, lo_rank)) = lower {
700                if (endpoint, rank) < (lo_ep, lo_rank) {
701                    return false;
702                }
703            }
704            if let Some((hi_ep, hi_rank)) = upper {
705                if (endpoint, rank) > (hi_ep, hi_rank) {
706                    return false;
707                }
708            }
709            true
710        };
711        let (hot, cold) = self.primary_pair(src_idx);
712        if self.primary_sorted_flag(src_idx) && hot.len() > 1 {
713            let lo = match lower {
714                Some((lo_ep, lo_rank)) => {
715                    hot.partition_point(|h| (h.endpoint, h.rank) < (lo_ep, lo_rank))
716                }
717                None => 0,
718            };
719            let hi = match upper {
720                Some((hi_ep, hi_rank)) => {
721                    hot.partition_point(|h| (h.endpoint, h.rank) <= (hi_ep, hi_rank))
722                }
723                None => hot.len(),
724            };
725            for (h, c) in hot[lo..hi].iter().zip(&cold[lo..hi]) {
726                if c.is_live() {
727                    let nbr = Nbr::from_parts(*h, *c);
728                    if !f(nbr) {
729                        return;
730                    }
731                }
732            }
733        } else {
734            for (h, c) in hot.iter().zip(cold.iter()) {
735                if c.is_live() && in_range(h.endpoint, h.rank) {
736                    let nbr = Nbr::from_parts(*h, *c);
737                    if !f(nbr) {
738                        return;
739                    }
740                }
741            }
742        }
743        if let Some(single) = self.overflow_chunks.single_chunk(src_vid) {
744            for i in 0..single.len() {
745                if let Some(nbr) = single.slot_at(i) {
746                    if nbr.delete_ts == Timestamp::MAX
747                        && in_range(nbr.endpoint, nbr.rank)
748                        && !f(nbr)
749                    {
750                        return;
751                    }
752                }
753            }
754            return;
755        }
756        if let Some(chunks) = self.overflow_chunks.get(src_vid) {
757            for chunk in chunks {
758                for i in 0..chunk.len() {
759                    if let Some(nbr) = chunk.slot_at(i) {
760                        if nbr.delete_ts == Timestamp::MAX
761                            && in_range(nbr.endpoint, nbr.rank)
762                            && !f(nbr)
763                        {
764                            return;
765                        }
766                    }
767                }
768            }
769        }
770    }
771
772    /// Fill a caller buffer with the same range content as `visit_threshold`.
773    pub fn fill_threshold_into(
774        &self,
775        src_vid: u32,
776        lower: Option<(u32, i64)>,
777        upper: Option<(u32, i64)>,
778        out: &mut Vec<Nbr>,
779    ) {
780        out.clear();
781        self.visit_threshold(src_vid, lower, upper, |nbr| {
782            out.push(nbr);
783            true
784        });
785    }
786}
787
788#[cfg(test)]
789mod tests {
790    use super::*;
791
792    fn packed_endpoint(endpoint: u32, rank: i64) -> VertexId {
793        VertexId::edge_endpoint_key(endpoint, rank)
794    }
795
796    #[test]
797    fn batched_physical_fill_matches_single_vertex_walks() {
798        let mut csr = MutableCsr::with_capacity(4, 32);
799        for (row, base) in [(0u32, 10u64), (1, 20), (2, 30)] {
800            for i in 0..3 {
801                csr.insert_edge(
802                    row,
803                    packed_endpoint(base as u32 + i as u32, 0),
804                    EdgeId(base + i),
805                    1,
806                )
807                .expect("insert builds row");
808            }
809        }
810        csr.delete_edge(1, EdgeId(21), 5)
811            .expect("delete leaves tombstone");
812        let vids = [0u32, 1, 2, 9];
813        let mut batched = Vec::new();
814        let mut starts = Vec::new();
815        csr.fill_physical_into_batched(&vids, &mut batched, &mut starts);
816        assert_eq!(starts.len(), vids.len().saturating_add(1));
817        let mut single = Vec::new();
818        for (pos, vid) in vids.iter().enumerate() {
819            csr.fill_physical_into(*vid, &mut single);
820            assert_eq!(&batched[starts[pos]..starts[pos + 1]], single.as_slice());
821        }
822        let mut visited = Vec::new();
823        csr.visit_hot_batched(&vids, |vid, hot| {
824            visited.push((vid, hot.edge_id));
825            true
826        });
827        let mut expected = Vec::new();
828        for vid in vids {
829            csr.visit_hot(vid, |hot| {
830                expected.push((vid, hot.edge_id));
831                true
832            });
833        }
834        assert_eq!(visited, expected);
835    }
836}