1use std::ops::RangeInclusive;
5use std::sync::Arc;
6
7use super::bitmap::{Bitmap, count_ones};
8use super::{RowIdSequence, U64Segment};
9use lance_core::deepsize::DeepSizeOf;
10use lance_core::utils::address::RowAddress;
11use lance_core::utils::deletion::DeletionVector;
12use lance_core::{Error, Result};
13use rangemap::RangeInclusiveMap;
14
15const MAX_PROBE_DEPTH: u64 = 64;
19
20const MERGE_ROWS_BUDGET: u64 = 1 << 20;
23
24const RANK_BLOCK_BYTES: usize = 256;
28
29#[derive(Debug)]
45pub struct RowIdIndex {
46 fragments: Vec<FragmentEntry>,
48 end_tree: Vec<u64>,
51 merged: Option<MergedIndex>,
52}
53
54type MergedIndex = RangeInclusiveMap<u64, (U64Segment, U64Segment)>;
55
56pub struct FragmentRowIdIndex {
57 pub fragment_id: u32,
58 pub row_id_sequence: Arc<RowIdSequence>,
59 pub deletion_vector: Arc<DeletionVector>,
60}
61
62impl RowIdIndex {
63 pub fn new(fragment_indices: &[FragmentRowIdIndex]) -> Result<Self> {
65 let mut fragments: Vec<FragmentEntry> = fragment_indices
66 .iter()
67 .filter_map(FragmentEntry::new)
68 .collect();
69 fragments.sort_unstable_by_key(|entry| entry.start);
70
71 let mut index = Self {
72 end_tree: build_end_tree(&fragments),
73 fragments,
74 merged: None,
75 };
76 if !probing_beats_merging(&index.fragments) {
77 index.merged = Some(index.build_merged()?);
78 }
79 Ok(index)
80 }
81
82 fn build_merged(&self) -> Result<MergedIndex> {
83 let sources: Vec<FragmentRowIdIndex> = self
84 .fragments
85 .iter()
86 .map(|entry| FragmentRowIdIndex {
87 fragment_id: entry.fragment_id,
88 row_id_sequence: entry.sequence.clone(),
89 deletion_vector: entry.deletion_vector.clone(),
90 })
91 .collect();
92 let chunks = sources
93 .iter()
94 .flat_map(decompose_sequence)
95 .collect::<Vec<_>>();
96
97 let mut final_chunks = Vec::new();
98 for processed_chunk in prep_index_chunks(chunks) {
99 match processed_chunk {
100 RawIndexChunk::NonOverlapping(chunk) => {
101 final_chunks.push(chunk);
102 }
103 RawIndexChunk::Overlapping(_range, overlapping_chunks) => {
104 let merged_chunk = merge_overlapping_chunks(overlapping_chunks)?;
108 final_chunks.push(merged_chunk);
109 }
110 }
111 }
112
113 Ok(RangeInclusiveMap::from_iter(final_chunks))
114 }
115
116 pub fn get(&self, row_id: u64) -> Result<Option<RowAddress>> {
125 if let Some(merged) = &self.merged {
126 return Ok(merged_get(merged, row_id));
127 }
128 self.probe(row_id)
129 }
130
131 pub fn get_many(&self, row_ids: &[u64]) -> Result<Vec<Option<RowAddress>>> {
143 let n = row_ids.len();
144 let mut out = vec![None; n];
145 if n == 0 {
146 return Ok(out);
147 }
148
149 let mut sorted: Vec<(u64, usize)> = row_ids.iter().copied().zip(0..n).collect();
150 sorted.sort_unstable_by_key(|&(id, _)| id);
151
152 let Some(merged) = &self.merged else {
153 for (id, orig_idx) in sorted {
155 out[orig_idx] = self.probe(id)?;
156 }
157 return Ok(out);
158 };
159
160 let mut chunks = merged.iter().peekable();
161 for (id, orig_idx) in sorted {
162 while let Some((range, _)) = chunks.peek() {
164 if *range.end() < id {
165 chunks.next();
166 } else {
167 break;
168 }
169 }
170 let Some((range, (row_id_seg, addr_seg))) = chunks.peek() else {
171 break;
172 };
173 if id < *range.start() {
174 continue; }
176 if let Some(pos) = row_id_seg.position(id)
177 && let Some(addr) = addr_seg.get(pos)
178 {
179 out[orig_idx] = Some(RowAddress::from(addr));
180 }
181 }
182 Ok(out)
183 }
184
185 fn probe(&self, row_id: u64) -> Result<Option<RowAddress>> {
191 let fragments = self.fragments.len();
192 if fragments == 0 {
193 return Ok(None);
194 }
195 let upper = self
197 .fragments
198 .partition_point(|entry| entry.start <= row_id);
199 if upper == 0 {
200 return Ok(None);
201 }
202 let mut found: Option<RowAddress> = None;
203 let (mut node, mut lo, mut width) = (1usize, 0usize, self.end_tree.len() / 2);
208 loop {
209 if lo >= upper {
210 break;
212 }
213 if self.end_tree[node] >= row_id {
214 if width > 1 {
215 node *= 2;
216 width /= 2;
217 continue;
218 }
219 if lo < fragments
220 && let Some(candidate) = self.fragments[lo].resolve(row_id)
221 {
222 if found.is_some() {
223 return Err(Error::internal(format!(
224 "row id index corrupt: stable row id {row_id} is \
225 live in multiple fragments",
226 )));
227 }
228 found = Some(candidate);
229 }
230 }
231 while node & 1 == 1 {
232 if node == 1 {
233 return Ok(found);
234 }
235 node /= 2;
236 lo -= width;
237 width *= 2;
238 }
239 node += 1;
240 lo += width;
241 }
242 Ok(found)
243 }
244}
245
246fn merged_get(merged: &MergedIndex, row_id: u64) -> Option<RowAddress> {
247 let (row_id_segment, address_segment) = merged.get(&row_id)?;
248 let pos = row_id_segment.position(row_id)?;
249 let address = address_segment.get(pos)?;
250 Some(RowAddress::from(address))
251}
252
253#[derive(Debug)]
255struct SegmentEntry {
256 seq_idx: usize,
257 range: RangeInclusive<u64>,
258 start_offset: u32,
259 lookup: SegmentLookup,
260}
261
262#[derive(Debug)]
264enum SegmentLookup {
265 Native,
267 Positions(Vec<(u64, u32)>),
270 BitmapRank(Vec<u32>),
274}
275
276impl SegmentEntry {
277 fn position(&self, sequence: &RowIdSequence, row_id: u64) -> Option<usize> {
279 let segment = &sequence.0[self.seq_idx];
280 match &self.lookup {
281 SegmentLookup::Native => segment.position(row_id),
282 SegmentLookup::Positions(positions) => positions
283 .binary_search_by_key(&row_id, |(id, _)| *id)
284 .ok()
285 .map(|found| positions[found].1 as usize),
286 SegmentLookup::BitmapRank(rank) => {
287 let U64Segment::RangeWithBitmap { range, bitmap } = segment else {
288 return None;
289 };
290 if !range.contains(&row_id) {
291 return None;
292 }
293 let offset = (row_id - range.start) as usize;
294 if !bitmap.get(offset) {
295 return None;
296 }
297 let block_start = offset / (RANK_BLOCK_BYTES * 8) * (RANK_BLOCK_BYTES * 8);
298 let ones_before = rank[offset / (RANK_BLOCK_BYTES * 8)] as usize
299 + bitmap.slice(block_start, offset - block_start).count_ones();
300 Some(ones_before)
301 }
302 }
303 }
304}
305
306fn build_lookup(segment: &U64Segment) -> (SegmentLookup, usize) {
309 match segment {
310 U64Segment::Array(_) => {
311 let mut positions: Vec<(u64, u32)> = segment
312 .iter()
313 .enumerate()
314 .map(|(position, row_id)| (row_id, position as u32))
315 .collect();
316 positions.sort_unstable();
317 positions.dedup_by_key(|(row_id, _)| *row_id);
319 (SegmentLookup::Positions(positions), segment.len())
320 }
321 U64Segment::RangeWithBitmap { bitmap, .. } => {
322 let (rank, ones) = build_bitmap_rank(bitmap);
323 (SegmentLookup::BitmapRank(rank), ones)
324 }
325 _ => (SegmentLookup::Native, segment.len()),
326 }
327}
328
329fn build_bitmap_rank(bitmap: &Bitmap) -> (Vec<u32>, usize) {
331 let mut rank = Vec::with_capacity(bitmap.data.len().div_ceil(RANK_BLOCK_BYTES));
332 let mut ones = 0usize;
333 for block in bitmap.data.chunks(RANK_BLOCK_BYTES) {
334 rank.push(ones as u32);
335 ones += count_ones(block);
336 }
337 (rank, ones)
338}
339
340#[derive(Debug)]
341struct FragmentEntry {
342 fragment_id: u32,
343 sequence: Arc<RowIdSequence>,
344 deletion_vector: Arc<DeletionVector>,
345 segments: Vec<SegmentEntry>,
346 start: u64,
347 end: u64,
348 merge_rows: u64,
350}
351
352impl FragmentEntry {
353 fn new(source: &FragmentRowIdIndex) -> Option<Self> {
354 let mut segments: Vec<SegmentEntry> = Vec::new();
355 let mut start_offset: u32 = 0;
356 let mut merge_rows: u64 = 0;
357 let deleted = !source.deletion_vector.is_empty();
358 for (seq_idx, segment) in source.row_id_sequence.0.iter().enumerate() {
359 let (lookup, len) = build_lookup(segment);
360 if deleted || !matches!(segment, U64Segment::Range(_)) {
362 merge_rows += len as u64;
363 }
364 if len > 0
367 && let Some(range) = segment.range()
368 {
369 segments.push(SegmentEntry {
370 seq_idx,
371 range,
372 start_offset,
373 lookup,
374 });
375 }
376 start_offset += len as u32;
377 }
378 let start = segments.iter().map(|entry| *entry.range.start()).min()?;
379 let end = segments.iter().map(|entry| *entry.range.end()).max()?;
380 Some(Self {
381 fragment_id: source.fragment_id,
382 sequence: source.row_id_sequence.clone(),
383 deletion_vector: source.deletion_vector.clone(),
384 segments,
385 start,
386 end,
387 merge_rows,
388 })
389 }
390
391 fn resolve(&self, row_id: u64) -> Option<RowAddress> {
394 for entry in &self.segments {
395 if !entry.range.contains(&row_id) {
396 continue;
397 }
398 let Some(position) = entry.position(&self.sequence, row_id) else {
399 continue;
400 };
401 let row_offset = entry.start_offset + position as u32;
402 if self.deletion_vector.contains(row_offset) {
403 continue;
404 }
405 return Some(RowAddress::new_from_parts(self.fragment_id, row_offset));
406 }
407 None
408 }
409}
410
411fn probing_beats_merging(fragments: &[FragmentEntry]) -> bool {
418 let merge_rows: u64 = fragments.iter().map(|entry| entry.merge_rows).sum();
419 merge_rows > MERGE_ROWS_BUDGET && max_overlap_depth(fragments) <= MAX_PROBE_DEPTH
420}
421
422fn max_overlap_depth(fragments: &[FragmentEntry]) -> u64 {
424 let mut ends: Vec<u64> = fragments.iter().map(|entry| entry.end).collect();
425 ends.sort_unstable();
426 let mut closed = 0;
427 let mut depth: u64 = 0;
428 for (opened, entry) in fragments.iter().enumerate() {
429 while closed < ends.len() && ends[closed] < entry.start {
430 closed += 1;
431 }
432 depth = depth.max((opened + 1 - closed) as u64);
433 }
434 depth
435}
436
437fn build_end_tree(fragments: &[FragmentEntry]) -> Vec<u64> {
440 if fragments.is_empty() {
441 return Vec::new();
442 }
443 let leaves = fragments.len().next_power_of_two();
444 let mut tree = vec![0_u64; 2 * leaves];
445 for (slot, entry) in fragments.iter().enumerate() {
446 tree[leaves + slot] = entry.end;
447 }
448 for node in (1..leaves).rev() {
449 tree[node] = tree[2 * node].max(tree[2 * node + 1]);
450 }
451 tree
452}
453
454impl DeepSizeOf for RowIdIndex {
455 fn deep_size_of_children(&self, context: &mut lance_core::deepsize::Context) -> usize {
458 let fragment_bytes: usize = self
459 .fragments
460 .iter()
461 .map(|entry| {
462 entry.sequence.deep_size_of_children(context)
463 + entry.deletion_vector.deep_size_of_children(context)
464 + entry.segments.capacity() * std::mem::size_of::<SegmentEntry>()
465 + entry
466 .segments
467 .iter()
468 .map(|segment| match &segment.lookup {
469 SegmentLookup::Native => 0,
470 SegmentLookup::Positions(positions) => {
471 positions.capacity() * std::mem::size_of::<(u64, u32)>()
472 }
473 SegmentLookup::BitmapRank(rank) => {
474 rank.capacity() * std::mem::size_of::<u32>()
475 }
476 })
477 .sum::<usize>()
478 })
479 .sum();
480 let merged_bytes: usize = self
481 .merged
482 .as_ref()
483 .map(|merged| {
484 merged
485 .iter()
486 .map(|(_, (row_id_segment, address_segment))| {
487 (2 * std::mem::size_of::<u64>())
488 + std::mem::size_of::<(U64Segment, U64Segment)>()
489 + row_id_segment.deep_size_of_children(context)
490 + address_segment.deep_size_of_children(context)
491 })
492 .sum()
493 })
494 .unwrap_or(0);
495 fragment_bytes
496 + merged_bytes
497 + self.fragments.capacity() * std::mem::size_of::<FragmentEntry>()
498 + self.end_tree.capacity() * std::mem::size_of::<u64>()
499 }
500}
501
502fn decompose_sequence(
503 frag_index: &FragmentRowIdIndex,
504) -> Vec<(RangeInclusive<u64>, (U64Segment, U64Segment))> {
505 let mut start_address: u64 = RowAddress::first_row(frag_index.fragment_id).into();
506 let mut current_offset = 0u32;
507 let no_deletions = frag_index.deletion_vector.is_empty();
508
509 frag_index
510 .row_id_sequence
511 .0
512 .iter()
513 .filter_map(|segment| {
514 let segment_len = segment.len();
515
516 let result = if no_deletions {
517 decompose_segment_no_deletions(segment, start_address)
518 } else {
519 decompose_segment_with_deletions(
520 segment,
521 start_address,
522 current_offset,
523 &frag_index.deletion_vector,
524 )
525 };
526
527 current_offset += segment_len as u32;
528 start_address += segment_len as u64;
529
530 result
531 })
532 .collect()
533}
534
535fn build_chunk_from_pairs(mut pairs: Vec<(u64, u64)>) -> Option<IndexChunk> {
537 if pairs.is_empty() {
538 return None;
539 }
540 pairs.sort_unstable_by_key(|(row_id, _)| *row_id);
543 let (row_ids, addresses): (Vec<u64>, Vec<u64>) = pairs.into_iter().unzip();
544 let row_id_segment = U64Segment::from_iter(row_ids);
545 let address_segment = U64Segment::from_iter(addresses);
546 let coverage = row_id_segment.range()?;
547 Some((coverage, (row_id_segment, address_segment)))
548}
549
550fn decompose_segment_no_deletions(segment: &U64Segment, start_address: u64) -> Option<IndexChunk> {
552 match segment {
553 U64Segment::Range(range) if !range.is_empty() => {
554 let len = range.end - range.start;
555 let row_id_segment = U64Segment::Range(range.clone());
556 let address_segment = U64Segment::Range(start_address..start_address + len);
557 let coverage = range.start..=range.end - 1;
558 Some((coverage, (row_id_segment, address_segment)))
559 }
560 U64Segment::Ranges { .. } if !segment.is_empty() => {
561 let row_id_segment = segment.clone();
564 let address_segment =
565 U64Segment::Range(start_address..start_address + row_id_segment.len() as u64);
566 let coverage = row_id_segment.range()?;
567 Some((coverage, (row_id_segment, address_segment)))
568 }
569 _ if segment.is_empty() => None,
570 _ => {
571 let pairs: Vec<(u64, u64)> = segment
573 .iter()
574 .enumerate()
575 .map(|(i, row_id)| (row_id, start_address + i as u64))
576 .collect();
577 build_chunk_from_pairs(pairs)
578 }
579 }
580}
581
582fn decompose_segment_with_deletions(
584 segment: &U64Segment,
585 start_address: u64,
586 current_offset: u32,
587 deletion_vector: &DeletionVector,
588) -> Option<IndexChunk> {
589 let pairs: Vec<(u64, u64)> = segment
590 .iter()
591 .enumerate()
592 .filter_map(|(i, row_id)| {
593 let row_offset = current_offset + i as u32;
594 if !deletion_vector.contains(row_offset) {
595 Some((row_id, start_address + i as u64))
596 } else {
597 None
598 }
599 })
600 .collect();
601 build_chunk_from_pairs(pairs)
602}
603
604type IndexChunk = (RangeInclusive<u64>, (U64Segment, U64Segment));
605
606#[derive(Debug)]
607enum RawIndexChunk {
608 NonOverlapping(IndexChunk),
609 Overlapping(RangeInclusive<u64>, Vec<IndexChunk>),
610}
611
612impl RawIndexChunk {
613 fn range_end(&self) -> u64 {
614 match self {
615 Self::NonOverlapping((range, _)) => *range.end(),
616 Self::Overlapping(range, _) => *range.end(),
617 }
618 }
619}
620
621fn prep_index_chunks(mut chunks: Vec<IndexChunk>) -> impl Iterator<Item = RawIndexChunk> {
626 chunks.sort_by_key(|(range, _)| u64::MAX - *range.start());
627
628 let mut output = Vec::new();
629
630 if let Some(first_chunk) = chunks.pop() {
632 output.push(RawIndexChunk::NonOverlapping(first_chunk));
633 } else {
634 return output.into_iter();
636 }
637
638 let mut current_range = 0..=0;
639 let mut current_overlap = Vec::new();
640 while let Some(chunk) = chunks.pop() {
641 debug_assert_eq!(
642 current_overlap
643 .iter()
644 .map(|(range, _): &IndexChunk| *range.start())
645 .min()
646 .unwrap_or_default(),
647 *current_range.start(),
648 );
649 debug_assert_eq!(
650 current_overlap
651 .iter()
652 .map(|(range, _): &IndexChunk| *range.end())
653 .max()
654 .unwrap_or_default(),
655 *current_range.end(),
656 );
657
658 if current_overlap.is_empty() {
659 let last_chunk_end = output.last().unwrap().range_end();
661 if *chunk.0.start() <= last_chunk_end {
662 match output.pop().unwrap() {
664 RawIndexChunk::NonOverlapping(chunk) => {
665 current_overlap.push(chunk);
666 }
667 _ => unreachable!(),
668 }
669 current_overlap.push(chunk);
670
671 let range_start = *current_overlap.first().unwrap().0.start();
672 let range_end = *current_overlap
673 .last()
674 .unwrap()
675 .0
676 .end()
677 .max(current_overlap.first().unwrap().0.end());
678 current_range = range_start..=range_end;
679 } else {
680 output.push(RawIndexChunk::NonOverlapping(chunk));
682 }
683 } else {
684 if chunk.0.start() <= current_range.end() {
686 let range_end = *chunk.0.end().max(current_range.end());
688 current_range = *current_range.start()..=range_end;
689
690 current_overlap.push(chunk);
691 } else {
692 output.push(RawIndexChunk::Overlapping(
694 std::mem::replace(&mut current_range, 0..=0),
695 std::mem::take(&mut current_overlap),
696 ));
697 output.push(RawIndexChunk::NonOverlapping(chunk));
698 }
699 }
700 }
701 debug_assert_eq!(
702 current_overlap
703 .iter()
704 .map(|(range, _): &IndexChunk| *range.start())
705 .min()
706 .unwrap_or_default(),
707 *current_range.start(),
708 );
709 debug_assert_eq!(
710 current_overlap
711 .iter()
712 .map(|(range, _): &IndexChunk| *range.end())
713 .max()
714 .unwrap_or_default(),
715 *current_range.end(),
716 );
717
718 if !current_overlap.is_empty() {
719 output.push(RawIndexChunk::Overlapping(
720 current_range.clone(),
721 current_overlap,
722 ));
723 }
724
725 output.into_iter()
726}
727
728fn merge_overlapping_chunks(overlapping_chunks: Vec<IndexChunk>) -> Result<IndexChunk> {
729 let total_capacity = overlapping_chunks
730 .iter()
731 .map(|(_, (row_ids, _))| row_ids.len())
732 .sum();
733 let mut values = Vec::with_capacity(total_capacity);
734 for (_, (row_ids, row_addrs)) in overlapping_chunks.iter() {
735 values.extend(row_ids.iter().zip(row_addrs.iter()));
736 }
737 values.sort_by_key(|(row_id, _)| *row_id);
738 if let Some(w) = values.windows(2).find(|w| w[0].0 == w[1].0) {
741 return Err(Error::internal(format!(
742 "row id index corrupt: stable row id {} is live in multiple fragments",
743 w[0].0
744 )));
745 }
746 let row_id_segment = U64Segment::from_iter(values.iter().map(|(row_id, _)| *row_id));
747 let address_segment = U64Segment::from_iter(values.iter().map(|(_, row_addr)| *row_addr));
748
749 let range = row_id_segment.range().unwrap();
750
751 Ok((range, (row_id_segment, address_segment)))
752}
753
754#[cfg(test)]
755impl RowIdIndex {
756 fn probing(fragment_indices: &[FragmentRowIdIndex]) -> Result<Self> {
758 let mut fragments: Vec<FragmentEntry> = fragment_indices
759 .iter()
760 .filter_map(FragmentEntry::new)
761 .collect();
762 fragments.sort_unstable_by_key(|entry| entry.start);
763 Ok(Self {
764 end_tree: build_end_tree(&fragments),
765 fragments,
766 merged: None,
767 })
768 }
769}
770
771#[cfg(test)]
772mod tests {
773 use super::*;
774 use proptest::{
775 prelude::{Just, Strategy, any},
776 prop_assert, prop_assert_eq,
777 };
778
779 fn sparse_sequence(len: u64) -> RowIdSequence {
781 RowIdSequence(vec![U64Segment::SortedArray(
782 (0..len).map(|value| value * 2).collect::<Vec<u64>>().into(),
783 )])
784 }
785
786 fn fragment(fragment_id: u32, sequence: RowIdSequence) -> FragmentRowIdIndex {
787 FragmentRowIdIndex {
788 fragment_id,
789 row_id_sequence: Arc::new(sequence),
790 deletion_vector: Arc::new(DeletionVector::default()),
791 }
792 }
793
794 #[test]
795 fn test_new_builds_the_merged_map_unless_probing_wins() {
796 let ranges = fragment(1, RowIdSequence(vec![U64Segment::Range(0..1_000_000)]));
799 assert!(RowIdIndex::new(&[ranges]).unwrap().merged.is_some());
800 let small = fragment(1, sparse_sequence(16));
801 assert!(RowIdIndex::new(&[small]).unwrap().merged.is_some());
802
803 let wide = fragment(1, sparse_sequence(MERGE_ROWS_BUDGET + 1));
805 let index = RowIdIndex::new(&[wide]).unwrap();
806 assert!(index.merged.is_none());
807 assert_eq!(
808 index.get(6).unwrap(),
809 Some(RowAddress::new_from_parts(1, 3))
810 );
811 }
812
813 #[test]
814 fn test_merged_index_keeps_ranges_segments_compact() {
815 let row_id_ranges = [100..103, 108..110, 115..119];
816 let sequence = RowIdSequence(vec![
817 U64Segment::Range(50..53),
818 U64Segment::from_sorted_ranges(&row_id_ranges).unwrap(),
819 ]);
820 let source = fragment(7, sequence);
821 let index = RowIdIndex::new(std::slice::from_ref(&source)).unwrap();
822
823 let merged = index.merged.as_ref().expect("small input should be merged");
824 assert!(matches!(
825 merged.get(&100).unwrap().0,
826 U64Segment::Ranges { .. }
827 ));
828
829 for (row_id, row_offset) in [
830 (50, 0),
831 (51, 1),
832 (52, 2),
833 (100, 3),
834 (101, 4),
835 (102, 5),
836 (108, 6),
837 (109, 7),
838 (115, 8),
839 (116, 9),
840 (117, 10),
841 (118, 11),
842 ] {
843 assert_eq!(
844 index.get(row_id).unwrap(),
845 Some(RowAddress::new_from_parts(7, row_offset)),
846 "row id {row_id}"
847 );
848 }
849 for missing in [49, 53, 99, 103, 107, 110, 114, 119, 120] {
850 assert_eq!(index.get(missing).unwrap(), None, "row id {missing}");
851 }
852 assert_eq!(
853 index.get_many(&[108, 53, 100, 108]).unwrap(),
854 vec![
855 Some(RowAddress::new_from_parts(7, 6)),
856 None,
857 Some(RowAddress::new_from_parts(7, 3)),
858 Some(RowAddress::new_from_parts(7, 6)),
859 ]
860 );
861 }
862
863 #[test]
864 fn test_overlapping_ranges_chunks_keep_row_addresses() {
865 let left_ranges = [100..103, 108..110, 115..119];
866 let right_ranges = [103..108, 110..115];
867 let sources = [
868 fragment(
869 7,
870 RowIdSequence(vec![U64Segment::from_sorted_ranges(&left_ranges).unwrap()]),
871 ),
872 fragment(
873 8,
874 RowIdSequence(vec![U64Segment::from_sorted_ranges(&right_ranges).unwrap()]),
875 ),
876 ];
877 let index = RowIdIndex::new(&sources).unwrap();
878
879 for row_id in 100..119 {
880 let (fragment_id, row_offset) = if (100..103).contains(&row_id)
881 || (108..110).contains(&row_id)
882 || (115..119).contains(&row_id)
883 {
884 let row_offset = match row_id {
885 100..=102 => row_id - 100,
886 108..=109 => row_id - 105,
887 115..=118 => row_id - 110,
888 _ => unreachable!(),
889 };
890 (7, row_offset)
891 } else {
892 let row_offset = match row_id {
893 103..=107 => row_id - 103,
894 110..=114 => row_id - 105,
895 _ => unreachable!(),
896 };
897 (8, row_offset)
898 };
899 assert_eq!(
900 index.get(row_id).unwrap(),
901 Some(RowAddress::new_from_parts(fragment_id, row_offset as u32)),
902 "row id {row_id}"
903 );
904 }
905 }
906
907 #[test]
908 fn test_deep_overlap_merges_however_many_rows_it_reads() {
909 let fragments = MAX_PROBE_DEPTH + 1;
912 let rows_per_fragment = MERGE_ROWS_BUDGET / fragments + 1;
913 let deep: Vec<FragmentRowIdIndex> = (0..fragments as u32)
914 .map(|id| {
915 let ids: Vec<u64> = (0..rows_per_fragment)
916 .map(|value| value * fragments + id as u64)
917 .collect();
918 fragment(id, RowIdSequence(vec![U64Segment::SortedArray(ids.into())]))
919 })
920 .collect();
921
922 assert!(RowIdIndex::new(&deep).unwrap().merged.is_some());
923 }
924
925 #[test]
926 fn test_probe_resolves_a_row_id_the_merged_map_rejects() {
927 let sources = [
928 fragment(1, RowIdSequence::from(&[0, 2][..])),
929 fragment(2, RowIdSequence::from(&[1, 2][..])),
930 ];
931 assert!(RowIdIndex::new(&sources).is_err());
932
933 let index = RowIdIndex::probing(&sources[..1]).unwrap();
934 assert_eq!(
935 index.get(2).unwrap(),
936 Some(RowAddress::new_from_parts(1, 1))
937 );
938 }
939
940 #[test]
941 fn test_probe_errors_when_two_fragments_hold_an_id_live() {
942 let sources = [
943 fragment(1, RowIdSequence::from(&[0, 2][..])),
944 fragment(2, RowIdSequence::from(&[1, 2][..])),
945 ];
946 let index = RowIdIndex::probing(&sources).unwrap();
947 assert_eq!(
948 index.get(0).unwrap(),
949 Some(RowAddress::new_from_parts(1, 0))
950 );
951
952 let error = index.get(2).unwrap_err();
953 assert!(matches!(&error, Error::Internal { .. }));
954 assert!(
955 error
956 .to_string()
957 .contains("stable row id 2 is live in multiple fragments")
958 );
959
960 let error = index.get_many(&[0, 2]).unwrap_err();
961 assert!(matches!(&error, Error::Internal { .. }));
962 }
963
964 #[test]
965 fn test_probe_ranks_a_bitmap_wider_than_one_block() {
966 let span = (RANK_BLOCK_BYTES * 8 * 3 + 17) as u64;
969 let present: Vec<bool> = (0..span).map(|slot| slot % 3 != 1).collect();
970 let segment = U64Segment::RangeWithBitmap {
971 range: 1000..1000 + span,
972 bitmap: present.as_slice().into(),
973 };
974 let index = RowIdIndex::probing(&[fragment(7, RowIdSequence(vec![segment]))]).unwrap();
975 let mut position = 0;
976 for slot in 0..span {
977 let found = index.get(1000 + slot).unwrap();
978 if slot % 3 == 1 {
979 assert_eq!(found, None);
980 } else {
981 assert_eq!(found, Some(RowAddress::new_from_parts(7, position)));
982 position += 1;
983 }
984 }
985 assert_eq!(index.get(999).unwrap(), None);
986 assert_eq!(index.get(1000 + span).unwrap(), None);
987 }
988
989 #[test]
990 fn test_probe_finds_every_position_of_an_unsorted_array() {
991 let row_ids: Vec<u64> = (0..2048).map(|value| (value * 7919) % 2048).collect();
992 let index = RowIdIndex::probing(&[fragment(
993 3,
994 RowIdSequence(vec![U64Segment::Array(row_ids.clone().into())]),
995 )])
996 .unwrap();
997 for (offset, row_id) in row_ids.iter().enumerate() {
998 assert_eq!(
999 index.get(*row_id).unwrap(),
1000 Some(RowAddress::new_from_parts(3, offset as u32))
1001 );
1002 }
1003 assert!(index.merged.is_none());
1004 }
1005
1006 #[test]
1007 fn test_new_index() {
1008 let fragment_indices = vec![
1009 FragmentRowIdIndex {
1010 fragment_id: 10,
1011 row_id_sequence: Arc::new(RowIdSequence(vec![
1012 U64Segment::Range(0..10),
1013 U64Segment::RangeWithHoles {
1014 range: 10..17,
1015 holes: vec![12, 15].into(),
1016 },
1017 U64Segment::SortedArray(vec![20, 25, 30].into()),
1018 ])),
1019 deletion_vector: Arc::new(DeletionVector::default()),
1020 },
1021 FragmentRowIdIndex {
1022 fragment_id: 20,
1023 row_id_sequence: Arc::new(RowIdSequence(vec![
1024 U64Segment::RangeWithBitmap {
1025 range: 17..20,
1026 bitmap: [true, false, true].as_slice().into(),
1027 },
1028 U64Segment::Array(vec![40, 50, 60].into()),
1029 ])),
1030 deletion_vector: Arc::new(DeletionVector::default()),
1031 },
1032 ];
1033
1034 let index = RowIdIndex::new(&fragment_indices).unwrap();
1035
1036 assert_eq!(
1038 index.get(0).unwrap(),
1039 Some(RowAddress::new_from_parts(10, 0))
1040 );
1041 assert_eq!(index.get(15).unwrap(), None);
1042 assert_eq!(
1043 index.get(16).unwrap(),
1044 Some(RowAddress::new_from_parts(10, 14))
1045 );
1046 assert_eq!(
1047 index.get(17).unwrap(),
1048 Some(RowAddress::new_from_parts(20, 0))
1049 );
1050 assert_eq!(
1051 index.get(25).unwrap(),
1052 Some(RowAddress::new_from_parts(10, 16))
1053 );
1054 assert_eq!(
1055 index.get(40).unwrap(),
1056 Some(RowAddress::new_from_parts(20, 2))
1057 );
1058 assert_eq!(
1059 index.get(60).unwrap(),
1060 Some(RowAddress::new_from_parts(20, 4))
1061 );
1062 assert_eq!(index.get(61).unwrap(), None);
1063 }
1064
1065 #[test]
1066 fn test_new_index_overlap() {
1067 let fragment_indices = vec![
1068 FragmentRowIdIndex {
1069 fragment_id: 23,
1070 row_id_sequence: Arc::new(RowIdSequence(vec![U64Segment::SortedArray(
1071 vec![3, 6, 9].into(),
1072 )])),
1073 deletion_vector: Arc::new(DeletionVector::default()),
1074 },
1075 FragmentRowIdIndex {
1076 fragment_id: 42,
1077 row_id_sequence: Arc::new(RowIdSequence(vec![U64Segment::SortedArray(
1078 vec![2, 5, 8].into(),
1079 )])),
1080 deletion_vector: Arc::new(DeletionVector::default()),
1081 },
1082 FragmentRowIdIndex {
1083 fragment_id: 10,
1084 row_id_sequence: Arc::new(RowIdSequence(vec![U64Segment::SortedArray(
1085 vec![1, 4, 7].into(),
1086 )])),
1087 deletion_vector: Arc::new(DeletionVector::default()),
1088 },
1089 ];
1090
1091 let index = RowIdIndex::new(&fragment_indices).unwrap();
1092
1093 assert_eq!(
1095 index.get(1).unwrap(),
1096 Some(RowAddress::new_from_parts(10, 0))
1097 );
1098 assert_eq!(
1099 index.get(2).unwrap(),
1100 Some(RowAddress::new_from_parts(42, 0))
1101 );
1102 assert_eq!(
1103 index.get(3).unwrap(),
1104 Some(RowAddress::new_from_parts(23, 0))
1105 );
1106 assert_eq!(
1107 index.get(4).unwrap(),
1108 Some(RowAddress::new_from_parts(10, 1))
1109 );
1110 assert_eq!(
1111 index.get(5).unwrap(),
1112 Some(RowAddress::new_from_parts(42, 1))
1113 );
1114 assert_eq!(
1115 index.get(6).unwrap(),
1116 Some(RowAddress::new_from_parts(23, 1))
1117 );
1118 assert_eq!(
1119 index.get(7).unwrap(),
1120 Some(RowAddress::new_from_parts(10, 2))
1121 );
1122 assert_eq!(
1123 index.get(8).unwrap(),
1124 Some(RowAddress::new_from_parts(42, 2))
1125 );
1126 assert_eq!(
1127 index.get(9).unwrap(),
1128 Some(RowAddress::new_from_parts(23, 2))
1129 );
1130 }
1131
1132 #[test]
1133 fn test_new_index_unsorted_row_ids() {
1134 let fragment_indices = vec![
1136 FragmentRowIdIndex {
1137 fragment_id: 10,
1138 row_id_sequence: Arc::new(RowIdSequence(vec![U64Segment::Array(
1139 vec![9, 3, 6].into(), )])),
1141 deletion_vector: Arc::new(DeletionVector::default()),
1142 },
1143 FragmentRowIdIndex {
1144 fragment_id: 20,
1145 row_id_sequence: Arc::new(RowIdSequence(vec![U64Segment::Array(
1146 vec![8, 2, 5].into(), )])),
1148 deletion_vector: Arc::new(DeletionVector::default()),
1149 },
1150 FragmentRowIdIndex {
1151 fragment_id: 30,
1152 row_id_sequence: Arc::new(RowIdSequence(vec![U64Segment::Array(
1153 vec![7, 1, 4].into(), )])),
1155 deletion_vector: Arc::new(DeletionVector::default()),
1156 },
1157 ];
1158
1159 let index = RowIdIndex::new(&fragment_indices).unwrap();
1160
1161 assert_eq!(
1163 index.get(1).unwrap(),
1164 Some(RowAddress::new_from_parts(30, 1))
1165 );
1166 assert_eq!(
1167 index.get(2).unwrap(),
1168 Some(RowAddress::new_from_parts(20, 1))
1169 );
1170 assert_eq!(
1171 index.get(3).unwrap(),
1172 Some(RowAddress::new_from_parts(10, 1))
1173 );
1174 assert_eq!(
1175 index.get(4).unwrap(),
1176 Some(RowAddress::new_from_parts(30, 2))
1177 );
1178 assert_eq!(
1179 index.get(5).unwrap(),
1180 Some(RowAddress::new_from_parts(20, 2))
1181 );
1182 assert_eq!(
1183 index.get(6).unwrap(),
1184 Some(RowAddress::new_from_parts(10, 2))
1185 );
1186 assert_eq!(
1187 index.get(7).unwrap(),
1188 Some(RowAddress::new_from_parts(30, 0))
1189 );
1190 assert_eq!(
1191 index.get(8).unwrap(),
1192 Some(RowAddress::new_from_parts(20, 0))
1193 );
1194 assert_eq!(
1195 index.get(9).unwrap(),
1196 Some(RowAddress::new_from_parts(10, 0))
1197 );
1198
1199 assert_eq!(index.get(0).unwrap(), None);
1201 assert_eq!(index.get(10).unwrap(), None);
1202 }
1203
1204 #[test]
1205 fn test_new_index_partial_overlap() {
1206 let fragment_indices = vec![
1207 FragmentRowIdIndex {
1208 fragment_id: 0,
1209 row_id_sequence: Arc::new(RowIdSequence(vec![U64Segment::RangeWithHoles {
1210 range: 0..100,
1211 holes: vec![50].into(),
1212 }])),
1213 deletion_vector: Arc::new(DeletionVector::default()),
1214 },
1215 FragmentRowIdIndex {
1216 fragment_id: 1,
1217 row_id_sequence: Arc::new(RowIdSequence(vec![U64Segment::Range(50..51)])),
1218 deletion_vector: Arc::new(DeletionVector::default()),
1219 },
1220 ];
1221
1222 let index = RowIdIndex::new(&fragment_indices).unwrap();
1223
1224 assert_eq!(
1226 index.get(0).unwrap(),
1227 Some(RowAddress::new_from_parts(0, 0))
1228 );
1229 assert_eq!(
1230 index.get(49).unwrap(),
1231 Some(RowAddress::new_from_parts(0, 49))
1232 );
1233 assert_eq!(
1234 index.get(50).unwrap(),
1235 Some(RowAddress::new_from_parts(1, 0))
1236 );
1237 assert_eq!(
1238 index.get(51).unwrap(),
1239 Some(RowAddress::new_from_parts(0, 50))
1240 );
1241 assert_eq!(
1242 index.get(99).unwrap(),
1243 Some(RowAddress::new_from_parts(0, 98))
1244 );
1245 }
1246
1247 #[test]
1248 fn test_overlapping_chunks_sparse_with_deletions() {
1249 let fragment_indices = vec![
1252 FragmentRowIdIndex {
1253 fragment_id: 10,
1254 row_id_sequence: Arc::new(RowIdSequence(vec![U64Segment::SortedArray(
1255 vec![1, 3, 5, 7, 9].into(),
1256 )])),
1257 deletion_vector: Arc::new(DeletionVector::default()),
1258 },
1259 FragmentRowIdIndex {
1260 fragment_id: 20,
1261 row_id_sequence: Arc::new(RowIdSequence(vec![U64Segment::SortedArray(
1262 vec![0, 2, 4, 6, 8].into(),
1263 )])),
1264 deletion_vector: Arc::new(DeletionVector::from_iter(vec![2])),
1266 },
1267 ];
1268
1269 let index = RowIdIndex::new(&fragment_indices).unwrap();
1270
1271 assert_eq!(
1272 index.get(0).unwrap(),
1273 Some(RowAddress::new_from_parts(20, 0))
1274 );
1275 assert_eq!(
1276 index.get(1).unwrap(),
1277 Some(RowAddress::new_from_parts(10, 0))
1278 );
1279 assert_eq!(
1280 index.get(2).unwrap(),
1281 Some(RowAddress::new_from_parts(20, 1))
1282 );
1283 assert_eq!(
1284 index.get(3).unwrap(),
1285 Some(RowAddress::new_from_parts(10, 1))
1286 );
1287 assert_eq!(index.get(4).unwrap(), None);
1288 assert_eq!(
1290 index.get(6).unwrap(),
1291 Some(RowAddress::new_from_parts(20, 3))
1292 );
1293 assert_eq!(
1294 index.get(8).unwrap(),
1295 Some(RowAddress::new_from_parts(20, 4))
1296 );
1297 assert_eq!(
1298 index.get(9).unwrap(),
1299 Some(RowAddress::new_from_parts(10, 4))
1300 );
1301 }
1302
1303 #[test]
1304 fn test_index_with_deletion_vector() {
1305 let deletion_vector = DeletionVector::from_iter(vec![2, 3]);
1306
1307 let fragment_indices = vec![FragmentRowIdIndex {
1308 fragment_id: 10,
1309 row_id_sequence: Arc::new(RowIdSequence(vec![U64Segment::Range(0..6)])),
1310 deletion_vector: Arc::new(deletion_vector),
1311 }];
1312
1313 let index = RowIdIndex::new(&fragment_indices).unwrap();
1314
1315 assert_eq!(
1316 index.get(0).unwrap(),
1317 Some(RowAddress::new_from_parts(10, 0))
1318 );
1319 assert_eq!(
1320 index.get(1).unwrap(),
1321 Some(RowAddress::new_from_parts(10, 1))
1322 );
1323 assert_eq!(
1324 index.get(4).unwrap(),
1325 Some(RowAddress::new_from_parts(10, 4))
1326 );
1327 assert_eq!(
1328 index.get(5).unwrap(),
1329 Some(RowAddress::new_from_parts(10, 5))
1330 );
1331
1332 assert_eq!(index.get(2).unwrap(), None);
1333 assert_eq!(index.get(3).unwrap(), None);
1334 }
1335
1336 #[test]
1337 fn test_empty_fragment_sequences() {
1338 let fragment_indices = vec![
1339 FragmentRowIdIndex {
1340 fragment_id: 10,
1341 row_id_sequence: Arc::new(RowIdSequence(vec![])),
1342 deletion_vector: Arc::new(DeletionVector::default()),
1343 },
1344 FragmentRowIdIndex {
1345 fragment_id: 20,
1346 row_id_sequence: Arc::new(RowIdSequence(vec![U64Segment::Range(5..8)])),
1347 deletion_vector: Arc::new(DeletionVector::default()),
1348 },
1349 ];
1350
1351 let index = RowIdIndex::new(&fragment_indices).unwrap();
1352
1353 assert_eq!(
1354 index.get(5).unwrap(),
1355 Some(RowAddress::new_from_parts(20, 0))
1356 );
1357 assert_eq!(
1358 index.get(7).unwrap(),
1359 Some(RowAddress::new_from_parts(20, 2))
1360 );
1361 assert_eq!(index.get(4).unwrap(), None);
1362 }
1363
1364 #[test]
1365 fn test_completely_empty_index() {
1366 let fragment_indices = vec![];
1367 let index = RowIdIndex::new(&fragment_indices).unwrap();
1368
1369 assert_eq!(index.get(0).unwrap(), None);
1370 assert_eq!(index.get(100).unwrap(), None);
1371 }
1372
1373 #[test]
1374 fn test_non_overlapping_ranges() {
1375 let fragment_indices = vec![
1376 FragmentRowIdIndex {
1377 fragment_id: 10,
1378 row_id_sequence: Arc::new(RowIdSequence(vec![U64Segment::Range(0..5)])),
1379 deletion_vector: Arc::new(DeletionVector::default()),
1380 },
1381 FragmentRowIdIndex {
1382 fragment_id: 20,
1383 row_id_sequence: Arc::new(RowIdSequence(vec![U64Segment::Range(5..10)])),
1384 deletion_vector: Arc::new(DeletionVector::default()),
1385 },
1386 FragmentRowIdIndex {
1387 fragment_id: 30,
1388 row_id_sequence: Arc::new(RowIdSequence(vec![U64Segment::Range(10..15)])),
1389 deletion_vector: Arc::new(DeletionVector::default()),
1390 },
1391 ];
1392
1393 let index = RowIdIndex::new(&fragment_indices).unwrap();
1394
1395 assert_eq!(
1396 index.get(0).unwrap(),
1397 Some(RowAddress::new_from_parts(10, 0))
1398 );
1399 assert_eq!(
1400 index.get(4).unwrap(),
1401 Some(RowAddress::new_from_parts(10, 4))
1402 );
1403 assert_eq!(
1404 index.get(5).unwrap(),
1405 Some(RowAddress::new_from_parts(20, 0))
1406 );
1407 assert_eq!(
1408 index.get(9).unwrap(),
1409 Some(RowAddress::new_from_parts(20, 4))
1410 );
1411 assert_eq!(
1412 index.get(10).unwrap(),
1413 Some(RowAddress::new_from_parts(30, 0))
1414 );
1415 assert_eq!(
1416 index.get(14).unwrap(),
1417 Some(RowAddress::new_from_parts(30, 4))
1418 );
1419 }
1420
1421 fn arbitrary_row_ids(
1422 num_fragments_range: std::ops::Range<usize>,
1423 frag_size_range: std::ops::Range<usize>,
1424 ) -> impl Strategy<Value = Vec<(u32, Arc<RowIdSequence>)>> {
1425 let fragment_sizes = proptest::collection::vec(frag_size_range, num_fragments_range);
1426 fragment_sizes.prop_flat_map(|fragment_sizes| {
1427 let num_rows = fragment_sizes.iter().sum::<usize>() as u64;
1428 let row_ids = 0..num_rows;
1429 let row_ids = row_ids.collect::<Vec<_>>();
1430 let row_ids_shuffled = proptest::strategy::Just(row_ids).prop_shuffle();
1431 row_ids_shuffled.prop_map(move |row_ids| {
1432 let mut sequences = Vec::with_capacity(fragment_sizes.len());
1433 let mut i = 0;
1434 for size in &fragment_sizes {
1435 let end = i + size;
1436 let sequence =
1437 RowIdSequence(vec![U64Segment::from_slice(row_ids[i..end].into())]);
1438 sequences.push((i as u32, Arc::new(sequence)));
1439 i = end;
1440 }
1441 sequences
1442 })
1443 })
1444 }
1445
1446 fn arbitrary_row_ids_with_deletions(
1447 num_fragments_range: std::ops::Range<usize>,
1448 frag_size_range: std::ops::Range<usize>,
1449 ) -> impl Strategy<Value = Vec<(u32, Arc<RowIdSequence>, Arc<DeletionVector>)>> {
1450 arbitrary_row_ids(num_fragments_range, frag_size_range)
1451 .prop_flat_map(|row_ids| {
1452 let num_rows = row_ids
1453 .iter()
1454 .map(|(_, sequence)| sequence.len() as usize)
1455 .sum::<usize>();
1456 (
1457 Just(row_ids),
1458 proptest::collection::vec(any::<bool>(), num_rows),
1459 )
1460 })
1461 .prop_map(|(row_ids, deleted_rows)| {
1462 let mut deleted_rows = deleted_rows.into_iter();
1463 row_ids
1464 .into_iter()
1465 .map(|(fragment_id, sequence)| {
1466 let mut deletion_bitmap = roaring::RoaringBitmap::new();
1467 for offset in 0..sequence.len() as u32 {
1468 if deleted_rows.next().unwrap() {
1469 deletion_bitmap.insert(offset);
1470 }
1471 }
1472 (
1473 fragment_id,
1474 sequence,
1475 Arc::new(DeletionVector::Bitmap(deletion_bitmap)),
1476 )
1477 })
1478 .collect()
1479 })
1480 }
1481
1482 #[test]
1483 fn test_large_range_segments_no_deletions() {
1484 let rows_per_fragment = 250_000u64;
1488 let num_fragments = 100u32;
1489 let mut offset = 0u64;
1490
1491 let fragment_indices: Vec<FragmentRowIdIndex> = (0..num_fragments)
1492 .map(|frag_id| {
1493 let start = offset;
1494 offset += rows_per_fragment;
1495 FragmentRowIdIndex {
1496 fragment_id: frag_id,
1497 row_id_sequence: Arc::new(RowIdSequence(vec![U64Segment::Range(
1498 start..start + rows_per_fragment,
1499 )])),
1500 deletion_vector: Arc::new(DeletionVector::default()),
1501 }
1502 })
1503 .collect();
1504
1505 let start = std::time::Instant::now();
1506 let index = RowIdIndex::new(&fragment_indices).unwrap();
1507 let elapsed = start.elapsed();
1508
1509 assert_eq!(
1511 index.get(0).unwrap(),
1512 Some(RowAddress::new_from_parts(0, 0))
1513 );
1514 assert_eq!(
1515 index.get(rows_per_fragment - 1).unwrap(),
1516 Some(RowAddress::new_from_parts(0, rows_per_fragment as u32 - 1))
1517 );
1518 assert_eq!(
1519 index.get(rows_per_fragment).unwrap(),
1520 Some(RowAddress::new_from_parts(1, 0))
1521 );
1522 let last_row = num_fragments as u64 * rows_per_fragment - 1;
1523 assert_eq!(
1524 index.get(last_row).unwrap(),
1525 Some(RowAddress::new_from_parts(
1526 num_fragments - 1,
1527 rows_per_fragment as u32 - 1
1528 ))
1529 );
1530 assert_eq!(index.get(last_row + 1).unwrap(), None);
1531
1532 assert!(
1535 elapsed.as_secs() < 1,
1536 "Index build took {:?} for {} fragments x {} rows = {} total rows. \
1537 This suggests the O(rows) -> O(fragments) optimization is not working.",
1538 elapsed,
1539 num_fragments,
1540 rows_per_fragment,
1541 num_fragments as u64 * rows_per_fragment,
1542 );
1543 }
1544
1545 #[test]
1546 fn test_large_range_segments_with_deletions() {
1547 let rows_per_fragment = 1_000u64;
1548 let num_fragments = 10u32;
1549 let mut offset = 0u64;
1550
1551 let fragment_indices: Vec<FragmentRowIdIndex> = (0..num_fragments)
1552 .map(|frag_id| {
1553 let start = offset;
1554 offset += rows_per_fragment;
1555
1556 let mut deleted = roaring::RoaringBitmap::new();
1558 for i in (0..rows_per_fragment as u32).step_by(3) {
1559 deleted.insert(i);
1560 }
1561
1562 FragmentRowIdIndex {
1563 fragment_id: frag_id,
1564 row_id_sequence: Arc::new(RowIdSequence(vec![U64Segment::Range(
1565 start..start + rows_per_fragment,
1566 )])),
1567 deletion_vector: Arc::new(DeletionVector::Bitmap(deleted)),
1568 }
1569 })
1570 .collect();
1571
1572 let index = RowIdIndex::new(&fragment_indices).unwrap();
1573
1574 assert_eq!(index.get(0).unwrap(), None);
1577 assert_eq!(index.get(3).unwrap(), None);
1579
1580 assert_eq!(
1583 index.get(1).unwrap(),
1584 Some(RowAddress::new_from_parts(0, 1))
1585 );
1586 assert_eq!(
1588 index.get(2).unwrap(),
1589 Some(RowAddress::new_from_parts(0, 2))
1590 );
1591 assert_eq!(
1593 index.get(4).unwrap(),
1594 Some(RowAddress::new_from_parts(0, 4))
1595 );
1596
1597 assert_eq!(index.get(rows_per_fragment).unwrap(), None);
1600 assert_eq!(
1602 index.get(rows_per_fragment + 1).unwrap(),
1603 Some(RowAddress::new_from_parts(1, 1))
1604 );
1605
1606 let last_row = num_fragments as u64 * rows_per_fragment - 1;
1609 assert_eq!(index.get(last_row).unwrap(), None);
1610 assert_eq!(
1612 index.get(last_row - 1).unwrap(),
1613 Some(RowAddress::new_from_parts(num_fragments - 1, 998))
1614 );
1615
1616 assert_eq!(index.get(last_row + 1).unwrap(), None);
1618 }
1619
1620 proptest::proptest! {
1621 #[test]
1622 fn test_new_index_robustness(
1623 row_ids in arbitrary_row_ids_with_deletions(0..5, 0..32)
1624 ) {
1625 let fragment_indices: Vec<FragmentRowIdIndex> = row_ids
1626 .iter()
1627 .map(|(frag_id, sequence, deletion_vector)| FragmentRowIdIndex {
1628 fragment_id: *frag_id,
1629 row_id_sequence: sequence.clone(),
1630 deletion_vector: deletion_vector.clone(),
1631 })
1632 .collect();
1633
1634 let merged = RowIdIndex::new(&fragment_indices).unwrap();
1635 let probing = RowIdIndex::probing(&fragment_indices).unwrap();
1636 for index in [&merged, &probing] {
1637 for (frag_id, sequence, deletion_vector) in row_ids.iter() {
1638 for (local_offset, row_id) in sequence.iter().enumerate() {
1639 let expected = if deletion_vector.contains(local_offset as u32) {
1640 None
1641 } else {
1642 Some(RowAddress::new_from_parts(*frag_id, local_offset as u32))
1643 };
1644 prop_assert_eq!(
1645 index.get(row_id).unwrap(),
1646 expected,
1647 "Row id {} in sequence {:?} not found in index {:?}",
1648 row_id,
1649 sequence,
1650 index
1651 );
1652 }
1653 }
1654 }
1655 }
1656
1657 #[test]
1658 fn test_new_index_moved_row_id(
1659 row_id in any::<u64>(),
1660 source_fragment in 0u32..1024,
1661 fragment_delta in 1u32..1024,
1662 ) {
1663 let target_fragment = source_fragment + fragment_delta;
1664 let fragment_indices = [
1665 FragmentRowIdIndex {
1666 fragment_id: source_fragment,
1667 row_id_sequence: Arc::new(RowIdSequence::from(&[row_id][..])),
1668 deletion_vector: Arc::new(DeletionVector::Bitmap(
1669 roaring::RoaringBitmap::from_iter([0]),
1670 )),
1671 },
1672 FragmentRowIdIndex {
1673 fragment_id: target_fragment,
1674 row_id_sequence: Arc::new(RowIdSequence::from(&[row_id][..])),
1675 deletion_vector: Arc::new(DeletionVector::default()),
1676 },
1677 ];
1678
1679 let index = RowIdIndex::new(&fragment_indices).unwrap();
1680 prop_assert_eq!(
1681 index.get(row_id).unwrap(),
1682 Some(RowAddress::new_from_parts(target_fragment, 0))
1683 );
1684 }
1685
1686 #[test]
1687 fn test_new_index_rejects_duplicate_live_row_id(
1688 row_id in any::<u64>(),
1689 first_fragment in 0u32..1024,
1690 fragment_delta in 1u32..1024,
1691 ) {
1692 let second_fragment = first_fragment + fragment_delta;
1693 let fragment_indices = [
1694 FragmentRowIdIndex {
1695 fragment_id: first_fragment,
1696 row_id_sequence: Arc::new(RowIdSequence::from(&[row_id][..])),
1697 deletion_vector: Arc::new(DeletionVector::default()),
1698 },
1699 FragmentRowIdIndex {
1700 fragment_id: second_fragment,
1701 row_id_sequence: Arc::new(RowIdSequence::from(&[row_id][..])),
1702 deletion_vector: Arc::new(DeletionVector::default()),
1703 },
1704 ];
1705
1706 let error = RowIdIndex::new(&fragment_indices).unwrap_err();
1707 let is_internal = matches!(&error, Error::Internal { .. });
1708 let expected_message =
1709 format!("stable row id {row_id} is live in multiple fragments");
1710 let error_message = error.to_string();
1711 prop_assert!(is_internal);
1712 prop_assert!(error_message.contains(&expected_message));
1713 }
1714 }
1715
1716 #[test]
1717 fn test_index_resolves_range_segments() {
1718 let live: Vec<u64> = (1000..3000u64)
1719 .filter(|v| !(1200..1900).contains(v))
1720 .collect();
1721 let mut runs_sequence = RowIdSequence::from(live.as_slice());
1722 assert!(runs_sequence.use_range_segments());
1723 let index = RowIdIndex::new(&[fragment(7, runs_sequence)]).unwrap();
1724 for (position, &row_id) in live.iter().enumerate() {
1725 assert_eq!(
1726 index.get(row_id).unwrap(),
1727 Some(RowAddress::new_from_parts(7, position as u32)),
1728 "row id {row_id}"
1729 );
1730 }
1731 assert_eq!(index.get(1500).unwrap(), None);
1732 let addr = |position: u32| Some(RowAddress::new_from_parts(7, position));
1733 assert_eq!(
1734 index.get_many(&[2999, 1000, 1199, 1900, 1500]).unwrap(),
1735 vec![
1736 addr(live.len() as u32 - 1),
1737 addr(0),
1738 addr(199),
1739 addr(200),
1740 None
1741 ]
1742 );
1743 }
1744}