lance_table/system_index/
frag_reuse.rs

1// SPDX-License-Identifier: Apache-2.0
2// SPDX-FileCopyrightText: Copyright The Lance Authors
3
4pub mod gate;
5pub mod ledger;
6pub mod lineage;
7pub mod metadata;
8
9use std::{collections::HashMap, io::Cursor, sync::Arc};
10
11use arrow_array::cast::AsArray;
12use arrow_array::types::UInt64Type;
13use arrow_array::{Array, ArrayRef, PrimitiveArray, RecordBatch, UInt64Array};
14use lance_core::deepsize::{Context, DeepSizeOf};
15use lance_core::utils::row_addr_remap::{GroupInputWithLayout, RowAddrRemap};
16use lance_core::{Error, Result};
17use lance_select::RowAddrTreeMap;
18use prost::Name;
19use roaring::{RoaringBitmap, RoaringTreemap};
20use serde::{Deserialize, Serialize};
21use uuid::Uuid;
22
23use crate::format::pb::fragment_reuse_index_details::InlineContent;
24use crate::format::{ExternalFile, Fragment, IndexMetadata, pb};
25
26pub const FRAG_REUSE_INDEX_NAME: &str = "__lance_frag_reuse";
27pub const FRAG_REUSE_DETAILS_FILE_NAME: &str = "details.binpb";
28
29/// Whether `index` is the fragment reuse index itself, not a user index that
30/// borrows its name.
31pub fn is_frag_reuse_index_entry(index: &IndexMetadata) -> bool {
32    index.name == FRAG_REUSE_INDEX_NAME
33        && index.index_details.as_ref().is_some_and(|details| {
34            details
35                .type_url
36                .rsplit_once('/')
37                .is_some_and(|(_, type_name)| {
38                    type_name.eq_ignore_ascii_case(&pb::FragmentReuseIndexDetails::full_name())
39                })
40        })
41}
42
43#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, DeepSizeOf)]
44pub struct FragDigest {
45    pub id: u64,
46    pub physical_rows: usize,
47    pub num_deleted_rows: usize,
48}
49
50impl From<&FragDigest> for pb::fragment_reuse_index_details::FragmentDigest {
51    fn from(digest: &FragDigest) -> Self {
52        Self {
53            id: digest.id,
54            physical_rows: digest.physical_rows as u64,
55            num_deleted_rows: digest.num_deleted_rows as u64,
56        }
57    }
58}
59
60impl From<&Fragment> for FragDigest {
61    fn from(fragment: &Fragment) -> Self {
62        Self {
63            id: fragment.id,
64            physical_rows: fragment
65                .physical_rows
66                .expect("Fragment doesn't have physical rows recorded"),
67            num_deleted_rows: fragment
68                .deletion_file
69                .as_ref()
70                .and_then(|d| d.num_deleted_rows)
71                .unwrap_or(0),
72        }
73    }
74}
75
76impl TryFrom<pb::fragment_reuse_index_details::FragmentDigest> for FragDigest {
77    type Error = Error;
78
79    fn try_from(digest: pb::fragment_reuse_index_details::FragmentDigest) -> Result<Self> {
80        Ok(Self {
81            id: digest.id,
82            physical_rows: digest.physical_rows as usize,
83            num_deleted_rows: digest.num_deleted_rows as usize,
84        })
85    }
86}
87
88#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, DeepSizeOf)]
89pub struct FragReuseGroup {
90    pub changed_row_addrs: Vec<u8>,
91    pub old_frags: Vec<FragDigest>,
92    pub new_frags: Vec<FragDigest>,
93}
94
95impl From<&FragReuseGroup> for pb::fragment_reuse_index_details::Group {
96    fn from(group: &FragReuseGroup) -> Self {
97        Self {
98            changed_row_addrs: group.changed_row_addrs.clone(),
99            old_fragments: group.old_frags.iter().map(|f| f.into()).collect(),
100            new_fragments: group.new_frags.iter().map(|f| f.into()).collect(),
101        }
102    }
103}
104
105impl TryFrom<pb::fragment_reuse_index_details::Group> for FragReuseGroup {
106    type Error = Error;
107
108    fn try_from(group: pb::fragment_reuse_index_details::Group) -> Result<Self> {
109        Ok(Self {
110            changed_row_addrs: group.changed_row_addrs,
111            old_frags: group
112                .old_fragments
113                .into_iter()
114                .map(FragDigest::try_from)
115                .collect::<Result<_>>()?,
116            new_frags: group
117                .new_fragments
118                .into_iter()
119                .map(FragDigest::try_from)
120                .collect::<Result<_>>()?,
121        })
122    }
123}
124
125#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, DeepSizeOf)]
126pub struct FragReuseVersion {
127    pub dataset_version: u64,
128    pub groups: Vec<FragReuseGroup>,
129}
130
131impl From<&FragReuseVersion> for pb::fragment_reuse_index_details::Version {
132    fn from(version: &FragReuseVersion) -> Self {
133        Self {
134            dataset_version: version.dataset_version,
135            groups: version.groups.iter().map(|g| g.into()).collect(),
136        }
137    }
138}
139
140impl TryFrom<pb::fragment_reuse_index_details::Version> for FragReuseVersion {
141    type Error = Error;
142
143    fn try_from(version: pb::fragment_reuse_index_details::Version) -> Result<Self> {
144        Ok(Self {
145            dataset_version: version.dataset_version,
146            groups: version
147                .groups
148                .into_iter()
149                .map(FragReuseGroup::try_from)
150                .collect::<Result<_>>()?,
151        })
152    }
153}
154
155impl FragReuseVersion {
156    pub fn old_frag_ids(&self) -> Vec<u64> {
157        self.groups
158            .iter()
159            .flat_map(|g| g.old_frags.iter().map(|f| f.id))
160            .collect::<Vec<_>>()
161    }
162
163    pub fn new_frag_ids(&self) -> Vec<u64> {
164        self.groups
165            .iter()
166            .flat_map(|g| g.new_frags.iter().map(|f| f.id))
167            .collect::<Vec<_>>()
168    }
169
170    pub fn new_frag_bitmap(&self) -> RoaringBitmap {
171        RoaringBitmap::from_iter(self.new_frag_ids().iter().map(|&id| id as u32))
172    }
173}
174
175#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, DeepSizeOf)]
176pub enum FragReuseIndexDetailsContentType {
177    Inline(FragReuseIndexDetails),
178    External(ExternalFile),
179}
180
181#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, DeepSizeOf)]
182pub struct FragReuseIndexDetails {
183    pub versions: Vec<FragReuseVersion>,
184}
185
186impl From<&FragReuseIndexDetails> for InlineContent {
187    fn from(details: &FragReuseIndexDetails) -> Self {
188        let mut versions: Vec<pb::fragment_reuse_index_details::Version> =
189            details.versions.iter().map(|m| m.into()).collect();
190        // sort from oldest to latest version
191        versions.sort_by_key(|v| v.dataset_version);
192        Self {
193            legacy_versions: versions,
194            transitions: Vec::new(),
195        }
196    }
197}
198
199impl TryFrom<InlineContent> for FragReuseIndexDetails {
200    type Error = Error;
201
202    fn try_from(content: InlineContent) -> Result<Self> {
203        Ok(Self {
204            versions: content
205                .legacy_versions
206                .into_iter()
207                .map(|m| m.try_into())
208                .collect::<Result<Vec<_>>>()?,
209        })
210    }
211}
212
213impl FragReuseIndexDetails {
214    pub fn new_frag_bitmap(&self) -> RoaringBitmap {
215        RoaringBitmap::from_iter(
216            self.versions
217                .iter()
218                .flat_map(|v| v.new_frag_ids().into_iter().map(|id| id as u32)),
219        )
220    }
221}
222
223/// An index that stores materialized row ID maps.
224///
225/// This type is retained for API and serde compatibility. Dataset loading uses
226/// [`CompactFragReuseIndex`] so persisted FRI details are not expanded into a
227/// hash-map entry for every affected row.
228#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
229pub struct FragReuseIndex {
230    pub uuid: Uuid,
231    pub row_id_maps: Vec<HashMap<u64, Option<u64>>>,
232    pub details: FragReuseIndexDetails,
233}
234
235impl DeepSizeOf for FragReuseIndex {
236    fn deep_size_of_children(&self, cx: &mut Context) -> usize {
237        self.row_id_maps.deep_size_of_children(cx) + self.details.deep_size_of_children(cx)
238    }
239}
240
241impl FragReuseIndex {
242    pub fn new(
243        uuid: Uuid,
244        row_id_maps: Vec<HashMap<u64, Option<u64>>>,
245        details: FragReuseIndexDetails,
246    ) -> Self {
247        Self {
248            uuid,
249            row_id_maps,
250            details,
251        }
252    }
253
254    pub fn is_empty(&self) -> bool {
255        self.row_id_maps.iter().all(HashMap::is_empty)
256    }
257
258    pub fn remap_row_id(&self, row_id: u64) -> Option<u64> {
259        let mut mapped = Some(row_id);
260        for row_id_map in &self.row_id_maps {
261            if let Some(current) = mapped {
262                mapped = row_id_map.get(&current).copied().unwrap_or(mapped);
263            }
264        }
265        mapped
266    }
267
268    pub fn remap_row_ids_in_place(&self, row_ids: &mut [Option<u64>]) {
269        for row_id_map in &self.row_id_maps {
270            for row_id in row_ids.iter_mut() {
271                if let Some(current) = *row_id
272                    && let Some(mapped) = row_id_map.get(&current)
273                {
274                    *row_id = *mapped;
275                }
276            }
277        }
278    }
279
280    pub fn remap_row_addrs_tree_map(&self, row_addrs: &RowAddrTreeMap) -> RowAddrTreeMap {
281        RowAddrTreeMap::from_iter(
282            row_addrs
283                .row_addrs()
284                .unwrap()
285                .filter_map(|addr| self.remap_row_id(u64::from(addr))),
286        )
287    }
288
289    pub fn remap_row_ids_roaring_tree_map(&self, row_ids: &RoaringTreemap) -> RoaringTreemap {
290        RoaringTreemap::from_iter(row_ids.iter().filter_map(|addr| self.remap_row_id(addr)))
291    }
292
293    pub fn remap_row_ids_record_batch(
294        &self,
295        batch: RecordBatch,
296        row_id_idx: usize,
297    ) -> Result<RecordBatch> {
298        remap_row_ids_record_batch(batch, row_id_idx, |row_ids| {
299            self.remap_row_ids_in_place(row_ids)
300        })
301    }
302
303    pub fn remap_row_ids_array(&self, array: ArrayRef) -> PrimitiveArray<UInt64Type> {
304        remap_row_ids_array(array, |row_ids| self.remap_row_ids_in_place(row_ids))
305    }
306
307    pub fn remap_fragment_bitmap(&self, fragment_bitmap: &mut RoaringBitmap) -> Result<()> {
308        remap_fragment_bitmap(&self.details, fragment_bitmap)
309    }
310}
311
312/// A compact row-address remap chain for deferred compactions.
313///
314/// Each FRI version retains rewritten-row bitmaps and fragment layouts. Queries
315/// use bitmap rank plus the ordered new-fragment ranges instead of storing one
316/// hash-map entry per affected row.
317#[derive(Debug, Clone, PartialEq, Eq)]
318pub struct CompactFragReuseIndex {
319    pub uuid: Uuid,
320    row_addr_remap: RowAddrRemap,
321    pub details: FragReuseIndexDetails,
322}
323
324impl DeepSizeOf for CompactFragReuseIndex {
325    fn deep_size_of_children(&self, cx: &mut Context) -> usize {
326        self.row_addr_remap.deep_size_of_children(cx) + self.details.deep_size_of_children(cx)
327    }
328}
329
330impl CompactFragReuseIndex {
331    #[doc(hidden)]
332    pub fn from_row_id_maps(
333        uuid: Uuid,
334        row_id_maps: Vec<HashMap<u64, Option<u64>>>,
335        details: FragReuseIndexDetails,
336    ) -> Self {
337        Self {
338            uuid,
339            row_addr_remap: RowAddrRemap::chained(
340                row_id_maps.into_iter().map(RowAddrRemap::direct),
341            ),
342            details,
343        }
344    }
345
346    /// Build a queryable index directly from serialized FRI details without
347    /// expanding each affected row into a hash map.
348    pub fn try_new(uuid: Uuid, details: FragReuseIndexDetails) -> Result<Self> {
349        let mut version_remaps = Vec::with_capacity(details.versions.len());
350        for (version_idx, version) in details.versions.iter().enumerate() {
351            let mut groups = Vec::with_capacity(version.groups.len());
352            for (group_idx, group) in version.groups.iter().enumerate() {
353                let changed_row_addrs = RoaringTreemap::deserialize_from(Cursor::new(
354                    &group.changed_row_addrs,
355                ))
356                .map_err(|error| {
357                    Error::index(format!(
358                        "failed to deserialize changed row addresses for FRI version {version_idx}, group {group_idx}: {error}"
359                    ))
360                })?;
361                let old_frags = group
362                    .old_frags
363                    .iter()
364                    .map(|frag| fragment_layout(frag, "old", version_idx, group_idx))
365                    .collect::<Result<Vec<_>>>()?;
366                let new_frags = group
367                    .new_frags
368                    .iter()
369                    .map(|frag| fragment_layout(frag, "new", version_idx, group_idx))
370                    .collect::<Result<Vec<_>>>()?;
371                groups.push(GroupInputWithLayout {
372                    rewritten_old_row_addrs: changed_row_addrs,
373                    old_frags,
374                    new_frags,
375                });
376            }
377            let remap = RowAddrRemap::compact_with_layout(groups).map_err(|error| {
378                Error::index(format!(
379                    "failed to build compact remap for FRI version {version_idx}: {error}"
380                ))
381            })?;
382            version_remaps.push(remap);
383        }
384
385        Ok(Self {
386            uuid,
387            row_addr_remap: RowAddrRemap::chained(version_remaps),
388            details,
389        })
390    }
391
392    /// The ordered remap chain used by index and transaction remapping paths.
393    pub fn row_addr_remap(&self) -> &RowAddrRemap {
394        &self.row_addr_remap
395    }
396
397    /// Returns whether the index contains no row-address remapping.
398    pub fn is_empty(&self) -> bool {
399        self.row_addr_remap.is_empty()
400    }
401
402    pub fn remap_row_id(&self, row_id: u64) -> Option<u64> {
403        self.row_addr_remap.get(row_id).unwrap_or(Some(row_id))
404    }
405
406    /// Apply all FRI versions to row addresses in place. `None` values remain
407    /// deleted and missing mappings pass through unchanged.
408    pub fn remap_row_ids_in_place(&self, row_ids: &mut [Option<u64>]) {
409        self.row_addr_remap.remap_in_place(row_ids);
410    }
411
412    pub fn remap_row_addrs_tree_map(&self, row_addrs: &RowAddrTreeMap) -> RowAddrTreeMap {
413        RowAddrTreeMap::from_iter(row_addrs.row_addrs().unwrap().filter_map(|addr| {
414            let addr_as_u64 = u64::from(addr);
415            self.remap_row_id(addr_as_u64)
416        }))
417    }
418
419    pub fn remap_row_ids_roaring_tree_map(&self, row_ids: &RoaringTreemap) -> RoaringTreemap {
420        RoaringTreemap::from_iter(row_ids.iter().filter_map(|addr| self.remap_row_id(addr)))
421    }
422
423    /// Remap a record batch that contains a row_id column at index `row_id_idx`
424    /// Currently this assumes there are only 2 columns in the schema,
425    /// which is the case for all indexes.
426    /// For example, for btree, the schema is (value, row_id).
427    /// For vector index storage, the schema is (row_id, vector).
428    pub fn remap_row_ids_record_batch(
429        &self,
430        batch: RecordBatch,
431        row_id_idx: usize,
432    ) -> Result<RecordBatch> {
433        remap_row_ids_record_batch(batch, row_id_idx, |row_ids| {
434            self.remap_row_ids_in_place(row_ids)
435        })
436    }
437
438    pub fn remap_row_ids_array(&self, array: ArrayRef) -> PrimitiveArray<UInt64Type> {
439        remap_row_ids_array(array, |row_ids| self.remap_row_ids_in_place(row_ids))
440    }
441
442    pub fn remap_fragment_bitmap(&self, fragment_bitmap: &mut RoaringBitmap) -> Result<()> {
443        remap_fragment_bitmap(&self.details, fragment_bitmap)
444    }
445}
446
447fn remap_row_ids_record_batch(
448    batch: RecordBatch,
449    row_id_idx: usize,
450    remap: impl FnOnce(&mut [Option<u64>]),
451) -> Result<RecordBatch> {
452    assert_eq!(batch.schema().fields().len(), 2);
453    let other_column_idx = 1 - row_id_idx;
454    let row_ids = batch.column(row_id_idx).as_primitive::<UInt64Type>();
455    let mut remapped_row_ids = row_ids
456        .values()
457        .iter()
458        .copied()
459        .map(Some)
460        .collect::<Vec<_>>();
461    remap(&mut remapped_row_ids);
462    let (val_indices, new_row_ids): (Vec<u64>, Vec<u64>) = remapped_row_ids
463        .iter()
464        .enumerate()
465        .filter_map(|(idx, new_id)| new_id.map(|new_id| (idx as u64, new_id)))
466        .unzip();
467    let new_val_indices = UInt64Array::from_iter_values(val_indices);
468    let new_vals = arrow::compute::take(batch.column(other_column_idx), &new_val_indices, None)?;
469
470    let mut batch_data: Vec<(usize, ArrayRef)> = vec![
471        (
472            row_id_idx,
473            Arc::new(UInt64Array::from_iter_values(new_row_ids)) as ArrayRef,
474        ),
475        (other_column_idx, Arc::new(new_vals)),
476    ];
477    batch_data.sort_by_key(|(i, _)| *i);
478    Ok(RecordBatch::try_new(
479        batch.schema(),
480        batch_data.into_iter().map(|(_, item)| item).collect(),
481    )?)
482}
483
484fn remap_row_ids_array(
485    array: ArrayRef,
486    remap: impl FnOnce(&mut [Option<u64>]),
487) -> PrimitiveArray<UInt64Type> {
488    let primitive_array = array
489        .as_any()
490        .downcast_ref::<PrimitiveArray<UInt64Type>>()
491        .expect("expected row IDs to be uint64 array");
492    let mut remapped = (0..primitive_array.len())
493        .map(|i| {
494            if primitive_array.is_null(i) {
495                None
496            } else {
497                Some(primitive_array.value(i))
498            }
499        })
500        .collect::<Vec<_>>();
501    remap(&mut remapped);
502    PrimitiveArray::from(remapped)
503}
504
505fn remap_fragment_bitmap(
506    details: &FragReuseIndexDetails,
507    fragment_bitmap: &mut RoaringBitmap,
508) -> Result<()> {
509    for version in details.versions.iter() {
510        for group in version.groups.iter() {
511            let mut removed = 0;
512            for old_frag in group.old_frags.iter() {
513                if fragment_bitmap.remove(old_frag.id as u32) {
514                    removed += 1;
515                }
516            }
517
518            if removed > 0 {
519                if removed != group.old_frags.len() {
520                    // Straddle: the index covered only part of this rewrite
521                    // group. Caused by the bug fixed in
522                    // <https://github.com/lance-format/lance/pull/6610>.
523                    // We've already removed the indexed old_frags from the
524                    // bitmap above; deliberately do NOT insert new_frags,
525                    // since the merged fragment also contains rows that
526                    // were never indexed. Affected rows fall through to
527                    // flat scan until the next optimize_indices. The fix
528                    // is persisted on the next write via build_manifest.
529                    tracing::warn!(
530                        "Healing straddling fragment-reuse rewrite group in index bitmap: \
531                             group {:?} was only partially indexed ({} of {} old fragments). \
532                             Affected rows will use flat scan until the next optimize_indices.",
533                        group.old_frags,
534                        removed,
535                        group.old_frags.len(),
536                    );
537                    continue;
538                }
539
540                for new_frag in group.new_frags.iter() {
541                    fragment_bitmap.insert(new_frag.id as u32);
542                }
543            }
544        }
545    }
546    Ok(())
547}
548
549fn fragment_layout(
550    frag: &FragDigest,
551    role: &str,
552    version_idx: usize,
553    group_idx: usize,
554) -> Result<(u32, u32)> {
555    let fragment_id = u32::try_from(frag.id).map_err(|_| {
556        Error::index(format!(
557            "FRI version {version_idx}, group {group_idx} has {role} fragment id {} outside the row-address range",
558            frag.id
559        ))
560    })?;
561    let physical_rows = u32::try_from(frag.physical_rows).map_err(|_| {
562        Error::index(format!(
563            "FRI version {version_idx}, group {group_idx} has {role} fragment {fragment_id} with physical_rows={} outside the row-address range",
564            frag.physical_rows
565        ))
566    })?;
567    Ok((fragment_id, physical_rows))
568}
569
570#[cfg(test)]
571mod tests {
572
573    use super::*;
574    use rstest::rstest;
575
576    fn addr(fragment_id: u32, offset: u32) -> u64 {
577        u64::from(lance_core::utils::address::RowAddress::new_from_parts(
578            fragment_id,
579            offset,
580        ))
581    }
582
583    fn serialize_changed(addrs: impl IntoIterator<Item = u64>) -> Vec<u8> {
584        let changed = RoaringTreemap::from_iter(addrs);
585        let mut bytes = Vec::with_capacity(changed.serialized_size());
586        changed.serialize_into(&mut bytes).unwrap();
587        bytes
588    }
589
590    fn digest(id: u64, physical_rows: usize) -> FragDigest {
591        FragDigest {
592            id,
593            physical_rows,
594            num_deleted_rows: 0,
595        }
596    }
597
598    #[test]
599    fn test_compact_fri_tristate_one_to_many_and_chain() {
600        let details = FragReuseIndexDetails {
601            versions: vec![
602                FragReuseVersion {
603                    dataset_version: 1,
604                    groups: vec![
605                        // One old fragment is split into two new fragments.
606                        FragReuseGroup {
607                            changed_row_addrs: serialize_changed([
608                                addr(1, 0),
609                                addr(1, 2),
610                                addr(1, 3),
611                            ]),
612                            old_frags: vec![digest(1, 4)],
613                            new_frags: vec![digest(10, 1), digest(11, 2)],
614                        },
615                        // A separate rewrite group deletes an entire fragment.
616                        FragReuseGroup {
617                            changed_row_addrs: serialize_changed([]),
618                            old_frags: vec![digest(3, 2)],
619                            new_frags: vec![],
620                        },
621                    ],
622                },
623                FragReuseVersion {
624                    dataset_version: 2,
625                    groups: vec![FragReuseGroup {
626                        changed_row_addrs: serialize_changed([addr(10, 0), addr(11, 1)]),
627                        old_frags: vec![digest(10, 1), digest(11, 2)],
628                        new_frags: vec![digest(20, 2)],
629                    }],
630                },
631            ],
632        };
633        let details = FragReuseIndexDetails::try_from(InlineContent::from(&details)).unwrap();
634        let fri = CompactFragReuseIndex::try_new(Uuid::new_v4(), details).unwrap();
635
636        // Surviving rows follow both versions in oldest-to-newest order.
637        assert_eq!(fri.remap_row_id(addr(1, 0)), Some(addr(20, 0)));
638        assert_eq!(fri.remap_row_id(addr(1, 3)), Some(addr(20, 1)));
639        // Deletes can happen in either the first or a later version.
640        assert_eq!(fri.remap_row_id(addr(1, 1)), None);
641        assert_eq!(fri.remap_row_id(addr(1, 2)), None);
642        assert_eq!(fri.remap_row_id(addr(3, 0)), None);
643        // Uncovered fragments and out-of-range offsets retain the existing
644        // missing-map pass-through semantics.
645        assert_eq!(fri.remap_row_id(addr(2, 0)), Some(addr(2, 0)));
646        assert_eq!(fri.remap_row_id(addr(1, 4)), Some(addr(1, 4)));
647
648        let mut batch = vec![
649            Some(addr(1, 0)),
650            Some(addr(1, 1)),
651            Some(addr(1, 2)),
652            Some(addr(1, 3)),
653            Some(addr(2, 0)),
654            None,
655        ];
656        fri.remap_row_ids_in_place(&mut batch);
657        assert_eq!(
658            batch,
659            vec![
660                Some(addr(20, 0)),
661                None,
662                None,
663                Some(addr(20, 1)),
664                Some(addr(2, 0)),
665                None,
666            ]
667        );
668    }
669
670    #[test]
671    fn test_compact_fri_rejects_invalid_changed_row_bitmap() {
672        let details = FragReuseIndexDetails {
673            versions: vec![FragReuseVersion {
674                dataset_version: 1,
675                groups: vec![FragReuseGroup {
676                    changed_row_addrs: vec![1, 2, 3],
677                    old_frags: vec![digest(1, 1)],
678                    new_frags: vec![digest(2, 1)],
679                }],
680            }],
681        };
682        let error = CompactFragReuseIndex::try_new(Uuid::new_v4(), details).unwrap_err();
683        assert!(matches!(error, Error::Index { .. }));
684        assert!(
685            error
686                .to_string()
687                .contains("failed to deserialize changed row addresses for FRI version 0, group 0"),
688            "{error}"
689        );
690    }
691
692    #[rstest]
693    #[case::unknown_fragment(
694        vec![addr(2, 0)],
695        vec![digest(1, 1)],
696        "from fragments [2] not in its old fragments"
697    )]
698    #[case::offset_out_of_range(
699        vec![addr(1, 1)],
700        vec![digest(1, 1)],
701        "row offset outside old fragment 1 with physical_rows=1"
702    )]
703    #[case::duplicate_old_fragment(
704        vec![addr(1, 0)],
705        vec![digest(1, 1), digest(1, 1)],
706        "old fragment 1 more than once"
707    )]
708    fn test_compact_fri_preserves_layout_validation(
709        #[case] changed_addrs: Vec<u64>,
710        #[case] old_frags: Vec<FragDigest>,
711        #[case] expected_message: &str,
712    ) {
713        let details = FragReuseIndexDetails {
714            versions: vec![FragReuseVersion {
715                dataset_version: 1,
716                groups: vec![FragReuseGroup {
717                    changed_row_addrs: serialize_changed(changed_addrs),
718                    old_frags,
719                    new_frags: vec![digest(10, 1)],
720                }],
721            }],
722        };
723
724        let error = CompactFragReuseIndex::try_new(Uuid::new_v4(), details).unwrap_err();
725        assert!(matches!(error, Error::Index { .. }));
726        let message = error.to_string();
727        assert!(message.contains("FRI version 0"), "{message}");
728        assert!(message.contains("rewrite group 0"), "{message}");
729        assert!(message.contains(expected_message), "{message}");
730    }
731
732    #[tokio::test]
733    async fn test_serialize_deserialize_index_details() {
734        // Create sample FragReuseVersions with different dataset versions
735        let version1 = FragReuseVersion {
736            dataset_version: 2,
737            groups: vec![FragReuseGroup {
738                changed_row_addrs: vec![1, 2, 3],
739                old_frags: vec![FragDigest {
740                    id: 1,
741                    physical_rows: 1,
742                    num_deleted_rows: 0,
743                }],
744                new_frags: vec![
745                    FragDigest {
746                        id: 2,
747                        physical_rows: 1,
748                        num_deleted_rows: 0,
749                    },
750                    FragDigest {
751                        id: 3,
752                        physical_rows: 1,
753                        num_deleted_rows: 0,
754                    },
755                ],
756            }],
757        };
758
759        let version2 = FragReuseVersion {
760            dataset_version: 1,
761            groups: vec![FragReuseGroup {
762                changed_row_addrs: vec![4, 5, 6],
763                old_frags: vec![FragDigest {
764                    id: 2,
765                    physical_rows: 1,
766                    num_deleted_rows: 0,
767                }],
768                new_frags: vec![
769                    FragDigest {
770                        id: 4,
771                        physical_rows: 1,
772                        num_deleted_rows: 0,
773                    },
774                    FragDigest {
775                        id: 5,
776                        physical_rows: 1,
777                        num_deleted_rows: 0,
778                    },
779                ],
780            }],
781        };
782
783        // Create FragReuseIndexDetails with versions in reverse order
784        let details = FragReuseIndexDetails {
785            versions: vec![version1, version2],
786        };
787
788        // Convert to protobuf format
789        let inline_content: InlineContent = (&details).into();
790
791        // Convert back to FragReuseIndexDetails
792        let roundtrip_details = FragReuseIndexDetails::try_from(inline_content).unwrap();
793
794        // Verify the roundtrip
795        assert_eq!(roundtrip_details.versions.len(), 2);
796
797        // Verify versions are sorted by dataset_version (oldest to latest)
798        assert_eq!(roundtrip_details.versions[0].dataset_version, 1);
799        assert_eq!(
800            roundtrip_details.versions[0].groups[0].changed_row_addrs,
801            vec![4, 5, 6]
802        );
803        assert_eq!(
804            roundtrip_details.versions[0].groups[0].new_frags,
805            vec![
806                FragDigest {
807                    id: 4,
808                    physical_rows: 1,
809                    num_deleted_rows: 0,
810                },
811                FragDigest {
812                    id: 5,
813                    physical_rows: 1,
814                    num_deleted_rows: 0,
815                }
816            ]
817        );
818        assert_eq!(
819            roundtrip_details.versions[0].groups[0].old_frags,
820            vec![FragDigest {
821                id: 2,
822                physical_rows: 1,
823                num_deleted_rows: 0,
824            }]
825        );
826
827        assert_eq!(roundtrip_details.versions[1].dataset_version, 2);
828        assert_eq!(
829            roundtrip_details.versions[1].groups[0].changed_row_addrs,
830            vec![1, 2, 3]
831        );
832        assert_eq!(
833            roundtrip_details.versions[1].groups[0].new_frags,
834            vec![
835                FragDigest {
836                    id: 2,
837                    physical_rows: 1,
838                    num_deleted_rows: 0,
839                },
840                FragDigest {
841                    id: 3,
842                    physical_rows: 1,
843                    num_deleted_rows: 0,
844                }
845            ]
846        );
847        assert_eq!(
848            roundtrip_details.versions[1].groups[0].old_frags,
849            vec![FragDigest {
850                id: 1,
851                physical_rows: 1,
852                num_deleted_rows: 0,
853            }]
854        );
855    }
856}
lance_table/system_index/frag_reuse.rs

lance_table/system_index/
frag_reuse.rs
