Skip to main content

lance_table/rowids/
index.rs

1// SPDX-License-Identifier: Apache-2.0
2// SPDX-FileCopyrightText: Copyright The Lance Authors
3
4use 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
15/// Fragments one lookup may have to probe before the merged map is worth its
16/// build, whatever that build costs. A compacted table interleaves its
17/// fragments, and one measured at 46.
18const MAX_PROBE_DEPTH: u64 = 64;
19
20/// Row ids the merged build may read before probing is worth its per-lookup
21/// cost instead.
22const MERGE_ROWS_BUDGET: u64 = 1 << 20;
23
24/// Bitmap bytes summarized by one entry of a bitmap rank directory. A lookup
25/// counts at most this many bytes; the directory costs one `u32` per block,
26/// 1/64 of the bitmap.
27const RANK_BLOCK_BYTES: usize = 256;
28
29/// An index of row ids
30///
31/// This index is used to map row ids to their corresponding addresses. These
32/// addresses correspond to physical positions in the dataset. See [RowAddress].
33///
34/// This structure only contains rows that physically exist. However, it may
35/// map to addresses that have been tombstoned. A separate tombstone index is
36/// used to track tombstoned rows.
37// (Implementation)
38// Two representations answer the same lookups, chosen once by `new`. The merged
39// map keys disjoint ranges of row ids to a pair of segments, the row ids and
40// the addresses, and reads every row id to build. A probe instead reads each
41// segment's bounds and asks the covering segment for the position of the id;
42// `new` takes that when few fragments cover one id and the merged build would
43// read a lot of them.
44#[derive(Debug)]
45pub struct RowIdIndex {
46    /// Fragments that hold at least one row id, sorted by their lowest row id.
47    fragments: Vec<FragmentEntry>,
48    /// Max-`end` heap over `fragments`: `end_tree[1]` is the root and leaf `i`
49    /// sits at `end_tree[len() / 2 + i]`.
50    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    /// Create a new index from a list of fragment ids and their corresponding row id sequences.
64    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                    // Intersecting row-id ranges don't imply intersecting id sets;
105                    // sparse ids and deletion holes leave the union short of the span.
106                    // The real invariant (no id in two fragments) is checked in the merge.
107                    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    /// Get the address for a given row id.
117    ///
118    /// Will return None if the row id does not exist in the index.
119    ///
120    /// # Errors
121    ///
122    /// Returns an error if the row id is live in more than one fragment,
123    /// which means the stable row ids are corrupt.
124    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    /// Get addresses for many row ids in one pass over the index.
132    ///
133    /// Returns one entry per input id, in input order (`None` for missing).
134    /// Sorts a working copy of the input internally so the chunk iterator
135    /// is advanced at most once per chunk, amortizing the per-id tree walk
136    /// from O(N ยท log F) to O(F + N).
137    ///
138    /// # Errors
139    ///
140    /// Returns an error if any requested row id is live in more than one
141    /// fragment, which means the stable row ids are corrupt.
142    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            // Sorted ids keep one fragment and its segments warm across the run.
154            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            // Advance past chunks that end before this id.
163            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; // falls in a gap between chunks
175            }
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    /// Address of `row_id`, from the fragment that holds it live. Descends the
186    /// max-`end` tree, so a fragment out of reach of the id costs nothing.
187    ///
188    /// Visits every candidate rather than stopping at the first hit, and
189    /// errors when a second fragment holds the id live.
190    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        // Only a fragment that starts at or below the id can hold it.
196        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        // Depth-first, left to right, without a stack: `node` covers `width`
204        // slots from `lo`. A pruned subtree or a visited leaf moves on to the
205        // next subtree: up while `node` is a right child, then over to the
206        // sibling.
207        let (mut node, mut lo, mut width) = (1usize, 0usize, self.end_tree.len() / 2);
208        loop {
209            if lo >= upper {
210                // Every slot from here on starts above the id.
211                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/// One segment of a sequence, and the offset its first row sits at.
254#[derive(Debug)]
255struct SegmentEntry {
256    seq_idx: usize,
257    range: RangeInclusive<u64>,
258    start_offset: u32,
259    lookup: SegmentLookup,
260}
261
262/// How a [`SegmentEntry`] finds the position of a row id.
263#[derive(Debug)]
264enum SegmentLookup {
265    /// The encoding searches itself in logarithmic or constant time.
266    Native,
267    /// Row id to position for an unsorted [`U64Segment::Array`], whose own
268    /// `position` scans.
269    Positions(Vec<(u64, u32)>),
270    /// Set bits before each [`RANK_BLOCK_BYTES`] block of a
271    /// [`U64Segment::RangeWithBitmap`] bitmap, whose own `position` counts
272    /// the whole prefix.
273    BitmapRank(Vec<u32>),
274}
275
276impl SegmentEntry {
277    /// Position of `row_id` in this segment, or `None` if it holds no such id.
278    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
306/// The lookup structure for `segment`, and the ids it holds. A bitmap's rank
307/// directory counts its ones on the way, so the segment is read once.
308fn 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            // The first position of a repeated id wins, as `position` does.
318            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
329/// Set bits before each [`RANK_BLOCK_BYTES`] block of `bitmap`, and in total.
330fn 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    /// Row ids the merged build reads one by one.
349    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            // A `Range` without deletions decomposes in constant time.
361            if deleted || !matches!(segment, U64Segment::Range(_)) {
362                merge_rows += len as u64;
363            }
364            // `range()` reports the span of a holed encoding, so ask `len` which
365            // ids the segment actually holds before trusting those bounds.
366            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    /// Address of `row_id` here, or `None` when the fragment lacks it or holds
392    /// it deleted.
393    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
411/// Whether to answer lookups by probing the fragments rather than by merging
412/// every row id.
413///
414/// Probing costs the fragments that cover one id, per lookup; merging costs the
415/// row ids it reads, once. So probe only when both stay on the right side of
416/// [`MAX_PROBE_DEPTH`] and [`MERGE_ROWS_BUDGET`].
417fn 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
422/// Most fragments that cover any one row id.
423fn 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
437/// Implicit max-`end` heap over `fragments`, padded to a power of two. Padding
438/// leaves hold 0, which prunes for every id above 0 and is filtered by slot.
439fn 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    /// Charges the sequences and deletion vectors the `Arc`s keep alive, which
456    /// a sequence cached under its own key is charged for as well.
457    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
535/// Build an IndexChunk from a list of (row_id, address) pairs.
536fn build_chunk_from_pairs(mut pairs: Vec<(u64, u64)>) -> Option<IndexChunk> {
537    if pairs.is_empty() {
538        return None;
539    }
540    // Sorted, so the row id segment encodes as one a lookup can search rather
541    // than an `Array` it has to scan. The address segment follows the pairing.
542    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
550/// Fast path: no deletions. O(1) for Range segments.
551fn 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            // Without deletions, a live ID's position is also its physical
562            // offset within this segment.
563            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            // Non-Range segments: must iterate to build address mapping.
572            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
582/// Slow path: has deletions, must check each row.
583fn 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
621/// Given a vector of index chunks, sort them and return an iterator of index chunks.
622///
623/// The iterator will yield chunks that are non-overlapping or a set of chunks
624/// that are overlapping.
625fn 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    // Start assuming non-overlapping in first chunk.
631    if let Some(first_chunk) = chunks.pop() {
632        output.push(RawIndexChunk::NonOverlapping(first_chunk));
633    } else {
634        // Early return for empty.
635        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            // We haven't found overlap yet.
660            let last_chunk_end = output.last().unwrap().range_end();
661            if *chunk.0.start() <= last_chunk_end {
662                // We have found overlap.
663                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                // We are still in non-overlapping space.
681                output.push(RawIndexChunk::NonOverlapping(chunk));
682            }
683        } else {
684            // We are making an overlap chunk
685            if chunk.0.start() <= current_range.end() {
686                // We are still in overlap.
687                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                // We have exited overlap.
693                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    // A duplicate row id here means two fragments claim the same live id: a
739    // corrupt index, not a resolvable sparse-coverage case.
740    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    /// Index that answers from the probe path, whatever the gate decided.
757    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    /// Sequence of `len` even row ids, held as a sorted array.
780    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        // Ranges decompose in constant time, and a small sequence is cheap to
797        // read whatever its encoding.
798        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        // Past the row budget, with one fragment covering any id.
804        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        // Just past the row budget in total, interleaved so every fragment
910        // covers every id: the depth alone forces the merged build.
911        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        // Three rank blocks and a partial fourth, with every third slot a hole,
967        // so each lookup combines a directory entry with a partial popcount.
968        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        // Check various queries.
1037        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        // Check various queries.
1094        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        // Test case with unsorted row ids within fragments
1135        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(), // Unsorted array
1140                )])),
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(), // Unsorted array
1147                )])),
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(), // Unsorted array
1154                )])),
1155                deletion_vector: Arc::new(DeletionVector::default()),
1156            },
1157        ];
1158
1159        let index = RowIdIndex::new(&fragment_indices).unwrap();
1160
1161        // Check that all row ids can be found regardless of their order in the segments
1162        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        // Check that non-existent row ids return None
1200        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        // Check various queries.
1225        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        // Interleaved (overlapping) id ranges plus a deletion that leaves a hole,
1250        // so the union doesn't tile the span. Every live id must still resolve.
1251        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                // Delete offset 2 (id 4) -> a hole in the span.
1265                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        // Surviving ids keep their original offsets (the hole is not compacted).
1289        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        // Simulates a real-world scenario: many fragments with large Range segments
1485        // and no deletions. Before optimization, this would iterate over all rows
1486        // (O(total_rows)). After optimization, it's O(num_fragments).
1487        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        // Verify correctness at boundaries
1510        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        // With the optimization, building an index for 25M rows across 100 fragments
1533        // should complete in well under 1 second (typically < 1ms).
1534        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                // Delete every 3rd row (offsets 0, 3, 6, ...) within each fragment.
1557                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        // Deleted rows (offset 0, 3, 6, ...) should not be found.
1575        // Row ID 0 has offset 0 in fragment 0 -> deleted.
1576        assert_eq!(index.get(0).unwrap(), None);
1577        // Row ID 3 has offset 3 in fragment 0 -> deleted.
1578        assert_eq!(index.get(3).unwrap(), None);
1579
1580        // Non-deleted rows should resolve correctly.
1581        // Row ID 1 has offset 1 in fragment 0 -> address (frag=0, row=1).
1582        assert_eq!(
1583            index.get(1).unwrap(),
1584            Some(RowAddress::new_from_parts(0, 1))
1585        );
1586        // Row ID 2 has offset 2 in fragment 0 -> address (frag=0, row=2).
1587        assert_eq!(
1588            index.get(2).unwrap(),
1589            Some(RowAddress::new_from_parts(0, 2))
1590        );
1591        // Row ID 4 has offset 4 in fragment 0 -> address (frag=0, row=4).
1592        assert_eq!(
1593            index.get(4).unwrap(),
1594            Some(RowAddress::new_from_parts(0, 4))
1595        );
1596
1597        // Check second fragment: row IDs start at 1000.
1598        // Row ID 1000 has offset 0 in fragment 1 -> deleted.
1599        assert_eq!(index.get(rows_per_fragment).unwrap(), None);
1600        // Row ID 1001 has offset 1 in fragment 1 -> address (frag=1, row=1).
1601        assert_eq!(
1602            index.get(rows_per_fragment + 1).unwrap(),
1603            Some(RowAddress::new_from_parts(1, 1))
1604        );
1605
1606        // Last fragment, last non-deleted row.
1607        // Row ID 9999 has offset 999 in fragment 9 -> 999 % 3 == 0 -> deleted.
1608        let last_row = num_fragments as u64 * rows_per_fragment - 1;
1609        assert_eq!(index.get(last_row).unwrap(), None);
1610        // Row ID 9998 has offset 998 -> 998 % 3 == 2 -> not deleted.
1611        assert_eq!(
1612            index.get(last_row - 1).unwrap(),
1613            Some(RowAddress::new_from_parts(num_fragments - 1, 998))
1614        );
1615
1616        // Out of range.
1617        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}