Skip to main content

lance_core/utils/
row_addr_remap.rs

1// SPDX-License-Identifier: Apache-2.0
2// SPDX-FileCopyrightText: Copyright The Lance Authors
3
4//! Compact row-address remapping for compaction.
5//!
6//! Compaction rewrites rows into new fragments, so indices that store physical
7//! row addresses need an old-address to new-address mapping without building an
8//! O(total rows) `HashMap<u64, Option<u64>>`.
9//!
10//! Layout:
11//!
12//! * Old rows: `old_fragment_id -> (old_offsets, old_rows_before)`
13//!     * `old_offsets`: rewritten old row offsets in this old fragment.
14//!     * `old_rows_before`: rewritten row count before this old fragment.
15//! * New rows: ordered new-fragment ranges
16//!   `(fragment_id, new_rows_before, physical_rows)`
17//!     * `new_rows_before`: rewritten row count before this new fragment.
18//!
19//! Lookup:
20//!
21//! * An address whose fragment was not rewritten returns `None`.
22//! * For an address whose fragment was rewritten:
23//!     * Read `(old_offsets, old_rows_before)` from the old-row layout.
24//!     * If `offset` is outside the old fragment's physical row range, return
25//!       `None`; the direct-map representation would not contain that address.
26//!     * If a valid `offset` is not in `old_offsets`, return `Some(None)`
27//!       because the row was deleted.
28//!     * Otherwise, `old_offsets.rank(offset) - 1` is this row's 0-based
29//!       position among rewritten old rows in this old fragment. Add
30//!       `old_rows_before` to get `k`, the row's 0-based position among all
31//!       rewritten old rows.
32//!     * In the new-row layout, find the range
33//!       `(fragment_id, new_rows_before, physical_rows)` where
34//!       `new_rows_before <= k < new_rows_before + physical_rows`.
35//!     * The new address is `(fragment_id, k - new_rows_before)`.
36//!
37//! Ordering:
38//!
39//! Compact remap does not store each old-to-new row mapping. It computes `k`
40//! from the old-row layout, then maps it to the k-th row written to the new
41//! fragments. This requires the reader-to-writer pipeline to preserve row order.
42//!
43//! * `old_frag_ids` must match the order old fragments are read. Within each
44//!   old fragment, rewritten rows are interpreted by ascending old row offset.
45//! * `new_frags` must match the order new rows are written.
46//! * Current compaction satisfies this because it scans selected fragments in
47//!   order and writes the resulting stream without reordering rows.
48
49use crate::deepsize::{Context, DeepSizeOf};
50use crate::utils::address::RowAddress;
51use crate::{Error, Result};
52use roaring::{RoaringBitmap, RoaringTreemap};
53use std::collections::{HashMap, HashSet};
54use std::mem::size_of;
55
56/// A queryable row-address remapping with the exact semantics of
57/// `HashMap<u64, Option<u64>>::get(&addr).copied()`:
58///
59/// * `None` — the address is not affected by this remap (keep it unchanged)
60/// * `Some(None)` — the row was deleted
61/// * `Some(Some(addr))` — the row moved to `addr`
62#[derive(Clone, Debug, PartialEq, Eq)]
63pub enum RowAddrRemap {
64    /// Compact, `O(#fragments)` remap built from per-group rewritten-row
65    /// bitmaps and new-fragment layouts.
66    Compact(CompactRowAddrRemap),
67    /// Full materialized old-to-new address map. Uses `O(#rows)` memory.
68    Direct(HashMap<u64, Option<u64>>),
69}
70
71impl RowAddrRemap {
72    pub fn compact(groups: impl IntoIterator<Item = GroupInput>) -> Result<Self> {
73        Ok(Self::Compact(CompactRowAddrRemap::new(groups)?))
74    }
75
76    /// Build a compact remap with physical row counts for exact validation of
77    /// addresses loaded from persisted fragment layouts.
78    #[doc(hidden)]
79    pub fn compact_with_layout(
80        groups: impl IntoIterator<Item = GroupInputWithLayout>,
81    ) -> Result<Self> {
82        Ok(Self::Compact(CompactRowAddrRemap::new_with_layout(groups)?))
83    }
84
85    /// Build a remap from a fully materialized old-to-new address map.
86    pub fn direct(map: HashMap<u64, Option<u64>>) -> Self {
87        Self::Direct(map)
88    }
89
90    /// Build an ordered remap chain, flattening nested chains and omitting
91    /// empty remaps.
92    pub fn chained(remaps: impl IntoIterator<Item = Self>) -> Self {
93        let mut remaps = remaps
94            .into_iter()
95            .filter(|remap| !remap.is_empty())
96            .collect::<Vec<_>>();
97        match remaps.len() {
98            0 => Self::empty(),
99            1 => remaps.pop().unwrap(),
100            _ => Self::Compact(CompactRowAddrRemap::chained(remaps)),
101        }
102    }
103
104    /// An empty remap that leaves every address unchanged.
105    pub fn empty() -> Self {
106        Self::Direct(HashMap::new())
107    }
108
109    /// Look up `addr`. See [`RowAddrRemap`] for the tri-state return semantics.
110    #[inline]
111    pub fn get(&self, addr: u64) -> Option<Option<u64>> {
112        match self {
113            Self::Compact(c) => c.get(addr),
114            Self::Direct(m) => m.get(&addr).copied(),
115        }
116    }
117
118    /// Apply this remap to a batch in place.
119    ///
120    /// A `None` input remains deleted. An address missing from a remap remains
121    /// unchanged. Chained remaps are applied version-by-version so this path is
122    /// suitable for bulk index and transaction remapping without materializing
123    /// a composed per-row map.
124    pub fn remap_in_place(&self, row_addrs: &mut [Option<u64>]) {
125        match self {
126            Self::Compact(compact) => compact.remap_in_place(row_addrs),
127            Self::Direct(_) => {
128                for row_addr in row_addrs {
129                    if let Some(addr) = *row_addr
130                        && let Some(mapped) = self.get(addr)
131                    {
132                        *row_addr = mapped;
133                    }
134                }
135            }
136        }
137    }
138
139    pub fn is_empty(&self) -> bool {
140        match self {
141            Self::Compact(c) => c.is_empty(),
142            Self::Direct(m) => m.is_empty(),
143        }
144    }
145
146    pub fn affected_fragments(&self) -> RoaringBitmap {
147        match self {
148            Self::Compact(c) => c.affected_fragments(),
149            Self::Direct(m) => RoaringBitmap::from_iter(m.keys().map(|addr| (addr >> 32) as u32)),
150        }
151    }
152
153    pub fn fully_deleted_fragments(&self) -> Option<RoaringBitmap> {
154        match self {
155            Self::Compact(c) => c.fully_deleted_fragments(),
156            Self::Direct(m) => {
157                if m.values().all(|v| v.is_none()) {
158                    Some(RoaringBitmap::from_iter(
159                        m.keys().map(|addr| (addr >> 32) as u32),
160                    ))
161                } else {
162                    None
163                }
164            }
165        }
166    }
167}
168
169impl DeepSizeOf for RowAddrRemap {
170    fn deep_size_of_children(&self, context: &mut Context) -> usize {
171        match self {
172            Self::Compact(compact) => compact.deep_size_of_children(context),
173            Self::Direct(map) => map.deep_size_of_children(context),
174        }
175    }
176}
177
178/// Input describing one rewrite group: the old row addresses that were
179/// rewritten plus the fragment layout before/after the rewrite.
180pub struct GroupInput {
181    /// Old row addresses that were read and re-written into the new fragments.
182    pub rewritten_old_row_addrs: RoaringTreemap,
183    /// Old fragment ids covered by this group.
184    pub old_frag_ids: Vec<u32>,
185    /// New fragments produced by this group, as `(fragment_id, physical_rows)`,
186    pub new_frags: Vec<(u32, u32)>,
187}
188
189/// Internal compact-remap input that includes old-fragment physical row counts.
190#[doc(hidden)]
191pub struct GroupInputWithLayout {
192    pub rewritten_old_row_addrs: RoaringTreemap,
193    pub old_frags: Vec<(u32, u32)>,
194    pub new_frags: Vec<(u32, u32)>,
195}
196
197/// Keep Roaring only when its serialized representation is substantially
198/// smaller than either rank-friendly representation. This preserves compact
199/// run containers while avoiding Roaring's linear word scan for dense rank.
200/// Binary-copy compaction creates these runs with `RoaringTreemap::insert_range`,
201/// and serialization preserves them without an explicit `optimize()` call.
202const ROARING_SIZE_ADVANTAGE_FOR_RANK: usize = 4;
203
204#[derive(Clone, Debug, PartialEq, Eq)]
205enum RankedOffsets {
206    /// Retained for highly compressible run layouts.
207    Roaring(RoaringBitmap),
208    /// Sorted rewritten offsets. Binary search returns membership and rank in
209    /// one operation.
210    Sparse(Vec<u32>),
211    /// Dense bits with the number of rewritten rows before every word.
212    Dense(DenseRankedOffsets),
213}
214
215impl RankedOffsets {
216    fn try_new(offsets: RoaringBitmap, physical_rows: Option<u32>) -> Result<Self> {
217        let universe_rows = physical_rows.map(u64::from).unwrap_or_else(|| {
218            offsets
219                .max()
220                .map(|offset| u64::from(offset) + 1)
221                .unwrap_or(0)
222        });
223        let word_count = usize::try_from(universe_rows.div_ceil(64)).map_err(|_| {
224            Error::invalid_input(format!(
225                "fragment row range {universe_rows} is too large for compact rank lookup"
226            ))
227        })?;
228        let sparse_bytes = usize::try_from(offsets.len())
229            .ok()
230            .and_then(|len| len.checked_mul(size_of::<u32>()))
231            .ok_or_else(|| {
232                Error::invalid_input(format!(
233                    "rewritten row count {} is too large for sparse rank lookup",
234                    offsets.len()
235                ))
236            })?;
237        let dense_bytes = word_count
238            .checked_mul(size_of::<u64>() + size_of::<u32>())
239            .ok_or_else(|| {
240                Error::invalid_input(format!(
241                    "fragment row range {universe_rows} is too large for dense rank lookup"
242                ))
243            })?;
244        let rank_friendly_bytes = sparse_bytes.min(dense_bytes);
245        if offsets
246            .serialized_size()
247            .checked_mul(ROARING_SIZE_ADVANTAGE_FOR_RANK)
248            .is_some_and(|roaring_bytes| roaring_bytes < rank_friendly_bytes)
249        {
250            return Ok(Self::Roaring(offsets));
251        }
252        if sparse_bytes <= dense_bytes {
253            return Ok(Self::Sparse(offsets.into_iter().collect()));
254        }
255        Ok(Self::Dense(DenseRankedOffsets::try_new(
256            offsets, word_count,
257        )?))
258    }
259
260    /// Return the zero-based rank when `offset` was rewritten.
261    #[inline]
262    fn rank_if_present(&self, offset: u32) -> Option<u64> {
263        match self {
264            Self::Roaring(offsets) => offsets.contains(offset).then(|| offsets.rank(offset) - 1),
265            Self::Sparse(offsets) => offsets.binary_search(&offset).ok().map(|rank| rank as u64),
266            Self::Dense(offsets) => offsets.rank_if_present(offset),
267        }
268    }
269
270    fn is_empty(&self) -> bool {
271        match self {
272            Self::Roaring(offsets) => offsets.is_empty(),
273            Self::Sparse(offsets) => offsets.is_empty(),
274            Self::Dense(offsets) => offsets.words.is_empty(),
275        }
276    }
277}
278
279impl DeepSizeOf for RankedOffsets {
280    fn deep_size_of_children(&self, context: &mut Context) -> usize {
281        match self {
282            // Roaring does not expose its allocation capacity. Its serialized
283            // size is a stable proxy for the retained containers.
284            Self::Roaring(offsets) => offsets.serialized_size(),
285            Self::Sparse(offsets) => offsets.deep_size_of_children(context),
286            Self::Dense(offsets) => offsets.deep_size_of_children(context),
287        }
288    }
289}
290
291#[derive(Clone, Debug, PartialEq, Eq)]
292struct DenseRankedOffsets {
293    words: Vec<u64>,
294    rank_before_word: Vec<u32>,
295}
296
297impl DenseRankedOffsets {
298    fn try_new(offsets: RoaringBitmap, word_count: usize) -> Result<Self> {
299        let mut words = vec![0u64; word_count];
300        for offset in offsets {
301            let word_idx = (offset / 64) as usize;
302            let Some(word) = words.get_mut(word_idx) else {
303                return Err(Error::invalid_input(format!(
304                    "rewritten row offset {offset} is outside dense rank word_count={word_count}"
305                )));
306            };
307            *word |= 1u64 << (offset % 64);
308        }
309
310        let mut rank_before_word = Vec::with_capacity(word_count);
311        let mut rewritten_rows_before = 0u64;
312        for word in &words {
313            rank_before_word.push(u32::try_from(rewritten_rows_before).map_err(|_| {
314                Error::invalid_input(format!(
315                    "rewritten row count {rewritten_rows_before} exceeds the row-address offset range"
316                ))
317            })?);
318            rewritten_rows_before += u64::from(word.count_ones());
319        }
320        Ok(Self {
321            words,
322            rank_before_word,
323        })
324    }
325
326    #[inline]
327    fn rank_if_present(&self, offset: u32) -> Option<u64> {
328        let word_idx = (offset / 64) as usize;
329        let word = *self.words.get(word_idx)?;
330        let bit = 1u64 << (offset % 64);
331        if word & bit == 0 {
332            return None;
333        }
334        Some(
335            u64::from(self.rank_before_word[word_idx]) + u64::from((word & (bit - 1)).count_ones()),
336        )
337    }
338}
339
340impl DeepSizeOf for DenseRankedOffsets {
341    fn deep_size_of_children(&self, context: &mut Context) -> usize {
342        self.words.deep_size_of_children(context)
343            + self.rank_before_word.deep_size_of_children(context)
344    }
345}
346
347#[derive(Clone, Debug, PartialEq, Eq)]
348struct OldFragmentRemap {
349    group_idx: usize,
350    rewritten_offsets: RankedOffsets,
351    rewritten_rows_before: u64,
352    physical_rows: Option<u32>,
353}
354
355impl DeepSizeOf for OldFragmentRemap {
356    fn deep_size_of_children(&self, context: &mut Context) -> usize {
357        self.rewritten_offsets.deep_size_of_children(context)
358    }
359}
360
361#[derive(Clone, Debug, PartialEq, Eq)]
362struct GroupRemap {
363    /// New fragment ranges as `(fragment_id, rewritten_rows_before, physical_rows)`,
364    /// used to map a rewritten row's group-local index to its new address via binary search.
365    new_frag_row_ranges: Vec<(u32, u64, u32)>,
366}
367
368impl GroupRemap {
369    fn new(input: GroupInput, group_idx: usize) -> Result<(Self, Vec<(u32, OldFragmentRemap)>)> {
370        Self::new_with_old_frags(
371            input.rewritten_old_row_addrs,
372            input.old_frag_ids.into_iter().map(|id| (id, None)),
373            input.new_frags,
374            group_idx,
375        )
376    }
377
378    fn new_with_layout(
379        input: GroupInputWithLayout,
380        group_idx: usize,
381    ) -> Result<(Self, Vec<(u32, OldFragmentRemap)>)> {
382        Self::new_with_old_frags(
383            input.rewritten_old_row_addrs,
384            input
385                .old_frags
386                .into_iter()
387                .map(|(id, rows)| (id, Some(rows))),
388            input.new_frags,
389            group_idx,
390        )
391    }
392
393    fn new_with_old_frags(
394        rewritten_old_row_addrs: RoaringTreemap,
395        old_frags: impl IntoIterator<Item = (u32, Option<u32>)>,
396        new_frags: Vec<(u32, u32)>,
397        group_idx: usize,
398    ) -> Result<(Self, Vec<(u32, OldFragmentRemap)>)> {
399        // `compute_new_addr` maps a rewritten row's group-local index by
400        // accumulating `physical_rows` in the caller-provided write order.
401        let mut new_frag_row_ranges = Vec::with_capacity(new_frags.len());
402        let mut rewritten_rows_before = 0u64;
403        for (frag_id, physical_rows) in new_frags {
404            if physical_rows == 0 {
405                continue;
406            }
407            new_frag_row_ranges.push((frag_id, rewritten_rows_before, physical_rows));
408            rewritten_rows_before += physical_rows as u64;
409        }
410        let total_new_rows = rewritten_rows_before;
411
412        let mut per_frag: HashMap<u32, RoaringBitmap> = rewritten_old_row_addrs
413            .bitmaps()
414            .map(|(frag_id, bitmap)| (frag_id, bitmap.clone()))
415            .collect();
416        let old_frags = old_frags.into_iter().collect::<Vec<_>>();
417        let mut frags = Vec::with_capacity(old_frags.len());
418        let mut seen_frag_ids = HashSet::with_capacity(old_frags.len());
419        let mut rewritten_rows_before = 0u64;
420        for &(frag_id, physical_rows) in &old_frags {
421            if !seen_frag_ids.insert(frag_id) {
422                return Err(Error::invalid_input(format!(
423                    "rewrite group {group_idx} contains old fragment {frag_id} more than once"
424                )));
425            }
426            let bitmap = per_frag.remove(&frag_id).unwrap_or_default();
427            if let Some(physical_rows) = physical_rows
428                && bitmap.max().is_some_and(|offset| offset >= physical_rows)
429            {
430                return Err(Error::invalid_input(format!(
431                    "rewrite group {group_idx} contains a row offset outside old fragment {frag_id} with physical_rows={physical_rows}"
432                )));
433            }
434            let num_rewritten_rows = bitmap.len();
435            let rewritten_offsets = RankedOffsets::try_new(bitmap, physical_rows)?;
436            frags.push((
437                frag_id,
438                OldFragmentRemap {
439                    group_idx,
440                    rewritten_offsets,
441                    rewritten_rows_before,
442                    physical_rows,
443                },
444            ));
445            rewritten_rows_before += num_rewritten_rows;
446        }
447        // Rewritten old row addresses must reference only listed old fragments.
448        if !per_frag.is_empty() {
449            return Err(Error::invalid_input(format!(
450                "compaction rewrite group {group_idx} references rewritten old row addresses from fragments {:?} not in its old fragments {:?}",
451                per_frag.keys().collect::<Vec<_>>(),
452                old_frags,
453            )));
454        }
455
456        // Rewritten old rows are mapped positionally onto the new rows, so the
457        // two counts must match exactly
458        let total_rewritten_old_rows = rewritten_old_row_addrs.len();
459        if total_new_rows != total_rewritten_old_rows {
460            return Err(Error::invalid_input(format!(
461                "compaction rewrite group {group_idx} rewrote {total_rewritten_old_rows} old rows from fragments {:?} but the new fragments hold {total_new_rows} rows",
462                old_frags,
463            )));
464        }
465
466        Ok((
467            Self {
468                new_frag_row_ranges,
469            },
470            frags,
471        ))
472    }
473
474    fn compute_new_addr(&self, rewritten_row_index: u64) -> u64 {
475        let idx =
476            match self
477                .new_frag_row_ranges
478                .binary_search_by(|(_, rewritten_rows_before, _)| {
479                    rewritten_rows_before.cmp(&rewritten_row_index)
480                }) {
481                Ok(i) => i,
482                Err(i) => i - 1,
483            };
484        let (frag_id, rewritten_rows_before, _rows) = self.new_frag_row_ranges[idx];
485        let offset = (rewritten_row_index - rewritten_rows_before) as u32;
486        u64::from(RowAddress::new_from_parts(frag_id, offset))
487    }
488}
489
490impl DeepSizeOf for GroupRemap {
491    fn deep_size_of_children(&self, context: &mut Context) -> usize {
492        self.new_frag_row_ranges.deep_size_of_children(context)
493    }
494}
495
496#[derive(Clone, Debug, PartialEq, Eq)]
497struct CompactRemapStep {
498    groups: Vec<GroupRemap>,
499    /// Old fragment id -> its bitmap/rank layout and rewrite group. Size is
500    /// O(#fragments), not rows.
501    frags: HashMap<u32, OldFragmentRemap>,
502}
503
504impl CompactRemapStep {
505    fn new(groups: impl IntoIterator<Item = GroupInput>) -> Result<Self> {
506        let mut frags = HashMap::new();
507        let mut group_remaps = Vec::new();
508        for input in groups {
509            let gi = group_remaps.len();
510            let (group_remap, group_frags) = GroupRemap::new(input, gi)?;
511            for (frag_id, frag) in group_frags {
512                if frags.insert(frag_id, frag).is_some() {
513                    return Err(Error::invalid_input(format!(
514                        "old fragment {frag_id} appears in more than one rewrite group, including group {gi}"
515                    )));
516                }
517            }
518            group_remaps.push(group_remap);
519        }
520        Ok(Self {
521            groups: group_remaps,
522            frags,
523        })
524    }
525
526    fn new_with_layout(groups: impl IntoIterator<Item = GroupInputWithLayout>) -> Result<Self> {
527        let mut frags = HashMap::new();
528        let mut group_remaps = Vec::new();
529        for input in groups {
530            let gi = group_remaps.len();
531            let (group_remap, group_frags) = GroupRemap::new_with_layout(input, gi)?;
532            for (frag_id, frag) in group_frags {
533                if frags.insert(frag_id, frag).is_some() {
534                    return Err(Error::invalid_input(format!(
535                        "old fragment {frag_id} appears in more than one rewrite group, including group {gi}"
536                    )));
537                }
538            }
539            group_remaps.push(group_remap);
540        }
541        Ok(Self {
542            groups: group_remaps,
543            frags,
544        })
545    }
546
547    #[inline]
548    pub fn get(&self, addr: u64) -> Option<Option<u64>> {
549        let frag = (addr >> 32) as u32;
550        // Not in any rewrite group -> unaffected by this remap.
551        let old_frag = self.frags.get(&frag)?;
552        let offset = addr as u32;
553        if old_frag
554            .physical_rows
555            .is_some_and(|physical_rows| offset >= physical_rows)
556        {
557            return None;
558        }
559        let Some(rewritten_rank) = old_frag.rewritten_offsets.rank_if_present(offset) else {
560            return Some(None);
561        };
562        let rewritten_row_index = old_frag.rewritten_rows_before + rewritten_rank;
563        Some(Some(
564            self.groups[old_frag.group_idx].compute_new_addr(rewritten_row_index),
565        ))
566    }
567
568    pub fn is_empty(&self) -> bool {
569        self.groups.is_empty()
570    }
571
572    fn fully_deleted_fragments(&self) -> Option<RoaringBitmap> {
573        // A group with any rewritten row moved at least one row.
574        if self
575            .frags
576            .values()
577            .any(|frag| !frag.rewritten_offsets.is_empty())
578        {
579            return None;
580        }
581        Some(RoaringBitmap::from_iter(self.frags.keys().copied()))
582    }
583
584    fn affected_fragments(&self) -> RoaringBitmap {
585        RoaringBitmap::from_iter(self.frags.keys().copied())
586    }
587}
588
589impl DeepSizeOf for CompactRemapStep {
590    fn deep_size_of_children(&self, context: &mut Context) -> usize {
591        self.groups.deep_size_of_children(context) + self.frags.deep_size_of_children(context)
592    }
593}
594
595#[derive(Clone, Debug, PartialEq, Eq)]
596enum RemapStep {
597    Compact(CompactRemapStep),
598    Direct(HashMap<u64, Option<u64>>),
599}
600
601impl RemapStep {
602    fn get(&self, addr: u64) -> Option<Option<u64>> {
603        match self {
604            Self::Compact(compact) => compact.get(addr),
605            Self::Direct(direct) => direct.get(&addr).copied(),
606        }
607    }
608
609    fn is_empty(&self) -> bool {
610        match self {
611            Self::Compact(compact) => compact.is_empty(),
612            Self::Direct(direct) => direct.is_empty(),
613        }
614    }
615
616    fn affected_fragments(&self) -> RoaringBitmap {
617        match self {
618            Self::Compact(compact) => compact.affected_fragments(),
619            Self::Direct(direct) => {
620                RoaringBitmap::from_iter(direct.keys().map(|addr| (addr >> 32) as u32))
621            }
622        }
623    }
624
625    fn fully_deleted_fragments(&self) -> Option<RoaringBitmap> {
626        match self {
627            Self::Compact(compact) => compact.fully_deleted_fragments(),
628            Self::Direct(direct) if direct.values().all(Option::is_none) => Some(
629                RoaringBitmap::from_iter(direct.keys().map(|addr| (addr >> 32) as u32)),
630            ),
631            Self::Direct(_) => None,
632        }
633    }
634}
635
636impl DeepSizeOf for RemapStep {
637    fn deep_size_of_children(&self, context: &mut Context) -> usize {
638        match self {
639            Self::Compact(compact) => compact.deep_size_of_children(context),
640            Self::Direct(direct) => direct.deep_size_of_children(context),
641        }
642    }
643}
644
645/// Compact remap backed by per-group rewritten row bitmaps + new-fragment layouts.
646///
647/// Multiple remaps are retained as ordered private steps so a version chain
648/// does not require another public [`RowAddrRemap`] variant.
649#[derive(Clone, Debug, PartialEq, Eq)]
650pub struct CompactRowAddrRemap {
651    steps: Vec<RemapStep>,
652}
653
654impl CompactRowAddrRemap {
655    fn new(groups: impl IntoIterator<Item = GroupInput>) -> Result<Self> {
656        Ok(Self {
657            steps: vec![RemapStep::Compact(CompactRemapStep::new(groups)?)],
658        })
659    }
660
661    fn new_with_layout(groups: impl IntoIterator<Item = GroupInputWithLayout>) -> Result<Self> {
662        Ok(Self {
663            steps: vec![RemapStep::Compact(CompactRemapStep::new_with_layout(
664                groups,
665            )?)],
666        })
667    }
668
669    fn chained(remaps: Vec<RowAddrRemap>) -> Self {
670        let mut steps = Vec::with_capacity(remaps.len());
671        for remap in remaps {
672            match remap {
673                RowAddrRemap::Compact(compact) => steps.extend(compact.steps),
674                RowAddrRemap::Direct(direct) => steps.push(RemapStep::Direct(direct)),
675            }
676        }
677        Self { steps }
678    }
679
680    #[inline]
681    pub fn get(&self, addr: u64) -> Option<Option<u64>> {
682        let mut current = addr;
683        let mut was_affected = false;
684        for step in &self.steps {
685            match step.get(current) {
686                None => {}
687                Some(None) => return Some(None),
688                Some(Some(mapped)) => {
689                    current = mapped;
690                    was_affected = true;
691                }
692            }
693        }
694        was_affected.then_some(Some(current))
695    }
696
697    fn remap_in_place(&self, row_addrs: &mut [Option<u64>]) {
698        for step in &self.steps {
699            for row_addr in row_addrs.iter_mut() {
700                if let Some(addr) = *row_addr
701                    && let Some(mapped) = step.get(addr)
702                {
703                    *row_addr = mapped;
704                }
705            }
706        }
707    }
708
709    pub fn is_empty(&self) -> bool {
710        self.steps.iter().all(RemapStep::is_empty)
711    }
712
713    fn affected_fragments(&self) -> RoaringBitmap {
714        self.steps
715            .iter()
716            .fold(RoaringBitmap::new(), |mut affected, step| {
717                affected |= step.affected_fragments();
718                affected
719            })
720    }
721
722    fn fully_deleted_fragments(&self) -> Option<RoaringBitmap> {
723        self.steps
724            .iter()
725            .try_fold(RoaringBitmap::new(), |mut deleted, step| {
726                deleted |= step.fully_deleted_fragments()?;
727                Some(deleted)
728            })
729    }
730}
731
732impl DeepSizeOf for CompactRowAddrRemap {
733    fn deep_size_of_children(&self, context: &mut Context) -> usize {
734        self.steps.deep_size_of_children(context)
735    }
736}
737
738#[cfg(test)]
739mod tests {
740    use super::*;
741
742    fn addr(frag: u32, offset: u32) -> u64 {
743        u64::from(RowAddress::new_from_parts(frag, offset))
744    }
745
746    #[derive(Clone, Copy)]
747    enum ExpectedRankedOffsets {
748        Sparse,
749        Dense,
750        Roaring,
751    }
752
753    fn assert_layout_matches_legacy(
754        frag_id: u32,
755        physical_rows: u32,
756        rewritten_old_row_addrs: RoaringTreemap,
757        new_frags: Vec<(u32, u32)>,
758        expected_representation: ExpectedRankedOffsets,
759    ) {
760        let rewritten_addrs = rewritten_old_row_addrs.iter().collect::<Vec<_>>();
761        let new_addrs = new_frags
762            .iter()
763            .flat_map(|(new_frag_id, rows)| (0..*rows).map(|offset| addr(*new_frag_id, offset)))
764            .collect::<Vec<_>>();
765        assert_eq!(rewritten_addrs.len(), new_addrs.len());
766        let expected_moved = rewritten_addrs
767            .iter()
768            .copied()
769            .zip(new_addrs)
770            .collect::<HashMap<_, _>>();
771
772        let remap = RowAddrRemap::compact_with_layout([GroupInputWithLayout {
773            rewritten_old_row_addrs,
774            old_frags: vec![(frag_id, physical_rows)],
775            new_frags,
776        }])
777        .unwrap();
778
779        let RowAddrRemap::Compact(compact) = &remap else {
780            panic!("compact_with_layout must produce a compact remap");
781        };
782        let RemapStep::Compact(step) = &compact.steps[0] else {
783            panic!("compact_with_layout must produce a compact step");
784        };
785        let offsets = &step.frags[&frag_id].rewritten_offsets;
786        assert!(match expected_representation {
787            ExpectedRankedOffsets::Sparse => matches!(offsets, RankedOffsets::Sparse(_)),
788            ExpectedRankedOffsets::Dense => matches!(offsets, RankedOffsets::Dense(_)),
789            ExpectedRankedOffsets::Roaring => matches!(offsets, RankedOffsets::Roaring(_)),
790        });
791
792        for offset in 0..physical_rows {
793            let old_addr = addr(frag_id, offset);
794            assert_eq!(
795                remap.get(old_addr),
796                Some(expected_moved.get(&old_addr).copied()),
797                "mismatch at ({frag_id}, {offset})"
798            );
799        }
800        assert_eq!(remap.get(addr(frag_id, physical_rows)), None);
801        assert_eq!(remap.get(addr(frag_id + 1, 0)), None);
802    }
803
804    #[test]
805    fn test_sparse_ranked_offsets() {
806        let offsets = RankedOffsets::try_new(
807            RoaringBitmap::from_iter([1u32, 63, 511, 9_999]),
808            Some(10_000),
809        )
810        .unwrap();
811        assert!(matches!(offsets, RankedOffsets::Sparse(_)));
812        assert_eq!(offsets.rank_if_present(0), None);
813        assert_eq!(offsets.rank_if_present(1), Some(0));
814        assert_eq!(offsets.rank_if_present(63), Some(1));
815        assert_eq!(offsets.rank_if_present(511), Some(2));
816        assert_eq!(offsets.rank_if_present(9_999), Some(3));
817    }
818
819    #[test]
820    fn test_dense_ranked_offsets_across_words() {
821        let rewritten = (0..1_024u32)
822            .filter(|offset| offset % 10 != 0)
823            .collect::<RoaringBitmap>();
824        let offsets = RankedOffsets::try_new(rewritten.clone(), Some(1_024)).unwrap();
825        assert!(matches!(offsets, RankedOffsets::Dense(_)));
826
827        let mut expected_rank = 0u64;
828        for offset in 0..1_024 {
829            if rewritten.contains(offset) {
830                assert_eq!(offsets.rank_if_present(offset), Some(expected_rank));
831                expected_rank += 1;
832            } else {
833                assert_eq!(offsets.rank_if_present(offset), None);
834            }
835        }
836        assert_eq!(expected_rank, rewritten.len());
837    }
838
839    #[test]
840    fn test_run_compressed_ranked_offsets() {
841        let mut rewritten = RoaringBitmap::new();
842        rewritten.insert_range(100..9_900);
843        let offsets = RankedOffsets::try_new(rewritten, Some(10_000)).unwrap();
844        assert!(matches!(offsets, RankedOffsets::Roaring(_)));
845        assert_eq!(offsets.rank_if_present(99), None);
846        assert_eq!(offsets.rank_if_present(100), Some(0));
847        assert_eq!(offsets.rank_if_present(9_899), Some(9_799));
848        assert_eq!(offsets.rank_if_present(9_900), None);
849    }
850
851    #[test]
852    fn test_compact_with_layout_matches_legacy_across_rank_representations() {
853        assert_layout_matches_legacy(
854            1,
855            10_000,
856            RoaringTreemap::from_iter(
857                [1u32, 63, 511, 9_999]
858                    .into_iter()
859                    .map(|offset| addr(1, offset)),
860            ),
861            vec![(10, 2), (11, 2)],
862            ExpectedRankedOffsets::Sparse,
863        );
864
865        let dense = (0..1_024u32)
866            .filter(|offset| offset % 10 != 0)
867            .map(|offset| addr(2, offset))
868            .collect::<RoaringTreemap>();
869        let dense_rows = u32::try_from(dense.len()).unwrap();
870        assert_layout_matches_legacy(
871            2,
872            1_024,
873            dense,
874            vec![(20, 400), (21, dense_rows - 400)],
875            ExpectedRankedOffsets::Dense,
876        );
877
878        // Binary-copy compaction captures complete fragment ranges with
879        // `RoaringTreemap::insert_range`, then persists that bitmap. The
880        // serialized round trip retains run containers without `optimize()`.
881        let mut captured = RoaringTreemap::new();
882        captured.insert_range(addr(3, 100)..addr(3, 9_900));
883        let mut serialized = Vec::with_capacity(captured.serialized_size());
884        captured.serialize_into(&mut serialized).unwrap();
885        let persisted = RoaringTreemap::deserialize_from(std::io::Cursor::new(serialized)).unwrap();
886        assert_layout_matches_legacy(
887            3,
888            10_000,
889            persisted,
890            vec![(31, 5_000), (30, 4_800)],
891            ExpectedRankedOffsets::Roaring,
892        );
893    }
894
895    #[test]
896    fn test_compact_lookup() {
897        // Group A: out-of-order old frags [4, 3], split new frags (11 empty),
898        // some deletions. frag 4 (5 rows) keeps 0,2,4; frag 3 keeps 0,1, so the
899        // rewritten rows (4,0)(4,2)(4,4)(3,0)(3,1) go to new frags 10(2), 12(3).
900        // Group B is a fully-deleted fragment.
901        let group_a = GroupInput {
902            rewritten_old_row_addrs: RoaringTreemap::from_iter([
903                addr(4, 0),
904                addr(4, 2),
905                addr(4, 4),
906                addr(3, 0),
907                addr(3, 1),
908            ]),
909            old_frag_ids: vec![4, 3],
910            new_frags: vec![(10, 2), (11, 0), (12, 3)],
911        };
912        let group_b = GroupInput {
913            rewritten_old_row_addrs: RoaringTreemap::new(),
914            old_frag_ids: vec![7],
915            new_frags: vec![],
916        };
917        let remap = RowAddrRemap::compact([group_a, group_b]).unwrap();
918
919        // Moves, in rewrite order; frag 4 comes first despite the larger id.
920        assert_eq!(remap.get(addr(4, 0)), Some(Some(addr(10, 0))));
921        assert_eq!(remap.get(addr(4, 2)), Some(Some(addr(10, 1))));
922        // Rank 2 skips the zero-row new fragment 11 and lands in fragment 12.
923        assert_eq!(remap.get(addr(4, 4)), Some(Some(addr(12, 0))));
924        assert_eq!(remap.get(addr(3, 0)), Some(Some(addr(12, 1))));
925        assert_eq!(remap.get(addr(3, 1)), Some(Some(addr(12, 2))));
926        // Deleted offsets inside a rewritten fragment.
927        assert_eq!(remap.get(addr(4, 1)), Some(None));
928        assert_eq!(remap.get(addr(4, 3)), Some(None));
929        // Covered but fully-deleted fragment -> Some(None), not None.
930        assert_eq!(remap.get(addr(7, 0)), Some(None));
931        // Fragment in no group -> unaffected.
932        assert_eq!(remap.get(addr(9, 0)), None);
933        assert_eq!(remap.get(addr(4, 5)), Some(None));
934        assert!(!remap.is_empty());
935    }
936
937    #[test]
938    fn test_fragment_sets() {
939        // Each deferred version deletes a different covered fragment. The
940        // chain must retain the flat direct map's union semantics.
941        let first_dead = RowAddrRemap::compact([GroupInput {
942            rewritten_old_row_addrs: RoaringTreemap::new(),
943            old_frag_ids: vec![3],
944            new_frags: vec![],
945        }])
946        .unwrap();
947        let second_dead = RowAddrRemap::compact([GroupInput {
948            rewritten_old_row_addrs: RoaringTreemap::new(),
949            old_frag_ids: vec![7],
950            new_frags: vec![],
951        }])
952        .unwrap();
953        let dead = RowAddrRemap::chained([first_dead.clone(), second_dead]);
954        assert_eq!(
955            dead.fully_deleted_fragments(),
956            Some(RoaringBitmap::from_iter([3u32, 7u32]))
957        );
958        assert_eq!(
959            dead.affected_fragments(),
960            RoaringBitmap::from_iter([3u32, 7u32])
961        );
962
963        // At least one rewritten row -> not fully deleted, but both covered
964        // fragments (including the fully-deleted frag 1) are still affected.
965        let alive = RowAddrRemap::compact([GroupInput {
966            rewritten_old_row_addrs: RoaringTreemap::from_iter([addr(0, 0)]),
967            old_frag_ids: vec![0, 1],
968            new_frags: vec![(10, 1)],
969        }])
970        .unwrap();
971        assert!(alive.fully_deleted_fragments().is_none());
972        assert_eq!(
973            alive.affected_fragments(),
974            RoaringBitmap::from_iter([0u32, 1u32])
975        );
976        assert!(
977            RowAddrRemap::chained([first_dead, alive])
978                .fully_deleted_fragments()
979                .is_none()
980        );
981    }
982
983    #[test]
984    fn test_compact_rejects_rewritten_addrs_outside_old_frags() {
985        // Rewritten addresses reference frag 5, not in old_frags. The count
986        // still matches (2 == 2), so only the per-fragment split catches it.
987        let input = GroupInput {
988            rewritten_old_row_addrs: RoaringTreemap::from_iter([addr(0, 0), addr(5, 0)]),
989            old_frag_ids: vec![0],
990            new_frags: vec![(10, 2)],
991        };
992        assert!(RowAddrRemap::compact([input]).is_err());
993    }
994
995    #[test]
996    fn test_compact_preserves_explicit_fragment_order() {
997        let remap = RowAddrRemap::compact([GroupInput {
998            rewritten_old_row_addrs: RoaringTreemap::from_iter([addr(0, 0), addr(0, 1)]),
999            old_frag_ids: vec![0],
1000            new_frags: vec![(12, 1), (11, 1)],
1001        }])
1002        .unwrap();
1003        assert_eq!(remap.get(addr(0, 0)), Some(Some(addr(12, 0))));
1004        assert_eq!(remap.get(addr(0, 1)), Some(Some(addr(11, 0))));
1005    }
1006
1007    #[test]
1008    fn test_direct_and_empty() {
1009        // Direct covers arbitrary maps the compact form can't express.
1010        let mut map = HashMap::new();
1011        map.insert(addr(2, 0), Some(addr(9, 9)));
1012        map.insert(addr(5, 1), None);
1013        let remap = RowAddrRemap::direct(map);
1014        assert_eq!(remap.get(addr(2, 0)), Some(Some(addr(9, 9))));
1015        assert_eq!(remap.get(addr(5, 1)), Some(None));
1016        assert_eq!(remap.get(addr(2, 1)), None);
1017        // affected_fragments over an explicit map: the fragment of every key.
1018        assert_eq!(
1019            remap.affected_fragments(),
1020            RoaringBitmap::from_iter([2u32, 5u32])
1021        );
1022
1023        let empty = RowAddrRemap::empty();
1024        assert!(empty.is_empty());
1025        assert_eq!(empty.get(addr(0, 0)), None);
1026    }
1027
1028    #[test]
1029    fn test_chained_lookup_and_batch() {
1030        let first = RowAddrRemap::compact([GroupInput {
1031            rewritten_old_row_addrs: RoaringTreemap::from_iter([addr(0, 0), addr(0, 2)]),
1032            old_frag_ids: vec![0],
1033            new_frags: vec![(10, 2)],
1034        }])
1035        .unwrap();
1036        let second = RowAddrRemap::compact([GroupInput {
1037            rewritten_old_row_addrs: RoaringTreemap::from_iter([addr(10, 1)]),
1038            old_frag_ids: vec![10],
1039            new_frags: vec![(20, 1)],
1040        }])
1041        .unwrap();
1042        let chain = RowAddrRemap::chained([first, second]);
1043
1044        assert_eq!(chain.get(addr(0, 0)), Some(None));
1045        assert_eq!(chain.get(addr(0, 1)), Some(None));
1046        assert_eq!(chain.get(addr(0, 2)), Some(Some(addr(20, 0))));
1047        assert_eq!(chain.get(addr(1, 0)), None);
1048
1049        let mut batch = vec![
1050            Some(addr(0, 0)),
1051            Some(addr(0, 1)),
1052            Some(addr(0, 2)),
1053            Some(addr(1, 0)),
1054            None,
1055        ];
1056        chain.remap_in_place(&mut batch);
1057        assert_eq!(
1058            batch,
1059            vec![None, None, Some(addr(20, 0)), Some(addr(1, 0)), None]
1060        );
1061    }
1062}