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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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}