1use crate::feature_flags::{
14 FLAG_COVERED_INDEX_METADATA, FLAG_FRAGMENT_REUSE_INDEX, FLAG_MANAGED_BLOBS,
15 FLAG_NON_REUSABLE_FIELD_IDS, FLAG_STABLE_ROW_IDS, apply_feature_flags,
16 ensure_can_read_manifest, ensure_can_write_manifest, inherit_sticky_feature_flags,
17 managed_blobs_enabled, validate_non_reusable_field_id_flags,
18};
19use crate::format::overlay::{OverlayCoverage, TOMBSTONE_FIELD_ID};
20use crate::format::{
21 DataFile, DataStorageFormat, Fragment, IndexMetadata, Manifest, ManifestBuildConfig, RowIdMeta,
22 overlay::DataOverlayFile,
23};
24use crate::io::{
25 commit::CommitHandler,
26 manifest::{read_manifest, read_manifest_indexes},
27};
28use crate::rowids::version::build_version_meta;
29use crate::rowids::{read_row_ids, write_row_ids};
30use crate::system_index::frag_reuse::metadata::{is_tagged, validate_flags};
31use crate::system_index::frag_reuse::{FRAG_REUSE_INDEX_NAME, is_frag_reuse_index_entry};
32use crate::system_index::is_system_index;
33use crate::system_index::mem_wal::{
34 CompactedSsTable, IndexCatchupProgress, MEM_WAL_INDEX_NAME, load_mem_wal_index_details,
35 new_mem_wal_index_meta, update_mem_wal_index_compacted_sstables,
36};
37use crate::transaction::UpdateMode::{RewriteColumns, RewriteRows};
38use crate::transaction::prepare::{FragReuseUpdate, PreparedIndices};
39use crate::transaction::row_version::resolve_update_version_metadata;
40use crate::transaction::update_map::apply_update_map;
41use crate::transaction::validate::merge_fragment_physically_rewritten;
42use crate::transaction::{
43 CoverageIdentity, DataReplacementGroup, LogicalIndexSegments, Operation, ReadVersionState,
44 RewriteGroup, Transaction, UpdatedFragmentOffsets,
45};
46use lance_core::datatypes::{
47 LANCE_UNENFORCED_CLUSTERING_KEY_POSITION, LANCE_UNENFORCED_PRIMARY_KEY,
48 LANCE_UNENFORCED_PRIMARY_KEY_POSITION,
49};
50use lance_core::utils::parse::str_is_truthy;
51use lance_core::{Error, Result};
52use lance_file::version::ConcreteFileVersion;
53use lance_io::object_store::ObjectStore;
54use object_store::path::Path;
55use roaring::RoaringBitmap;
56use std::collections::{BTreeMap, HashMap, HashSet};
57use std::sync::Arc;
58use uuid::Uuid;
59
60pub const RANGE_SEGMENTS_CONFIG_KEY: &str = "lance.row_ids.range_segments";
75
76fn range_segments_enabled(manifest: &Manifest) -> bool {
77 manifest
78 .config
79 .get(RANGE_SEGMENTS_CONFIG_KEY)
80 .is_some_and(|value| str_is_truthy(value))
81}
82
83fn apply_range_segments(
91 manifest: &mut Manifest,
92 current_manifest: Option<&Manifest>,
93) -> Result<()> {
94 if !range_segments_enabled(manifest) {
95 return Ok(());
96 }
97 let unchanged: HashMap<u64, *const u8> = current_manifest
98 .filter(|current| range_segments_enabled(current))
99 .map(|current| {
100 current
101 .fragments
102 .iter()
103 .filter_map(|fragment| match &fragment.row_id_meta {
104 Some(RowIdMeta::Inline(data)) => Some((fragment.id, data.as_ptr())),
105 _ => None,
106 })
107 .collect()
108 })
109 .unwrap_or_default();
110 let fragments = Arc::make_mut(&mut manifest.fragments);
111 for fragment in fragments.iter_mut() {
112 let Some(RowIdMeta::Inline(data)) = &fragment.row_id_meta else {
113 continue;
114 };
115 if unchanged.get(&fragment.id) == Some(&data.as_ptr()) {
116 continue;
117 }
118 let mut sequence = read_row_ids(&data[..])?;
119 if sequence.use_range_segments() {
120 fragment.row_id_meta = Some(RowIdMeta::Inline(write_row_ids(&sequence).into()));
121 }
122 }
123 Ok(())
124}
125
126impl Transaction {
127 pub(super) fn fragments_with_ids<'a, T>(
128 new_fragments: T,
129 fragment_id: &'a mut u64,
130 ) -> impl Iterator<Item = Fragment> + 'a
131 where
132 T: IntoIterator<Item = Fragment> + 'a,
133 {
134 new_fragments.into_iter().map(move |mut f| {
135 if f.id == 0 {
136 f.id = *fragment_id;
137 *fragment_id += 1;
138 }
139 f
140 })
141 }
142
143 fn data_storage_format_from_files(
144 fragments: &[Fragment],
145 user_requested: Option<ConcreteFileVersion>,
146 ) -> Result<DataStorageFormat> {
147 let version = match user_requested {
150 Some(ConcreteFileVersion::V1) => {
151 if let Some(actual) = Fragment::try_infer_version(fragments)?
153 && actual != ConcreteFileVersion::V1
154 {
155 return Err(Error::invalid_input(format!(
156 "User requested data storage version ({}) does not match version in data files ({actual})",
157 ConcreteFileVersion::V1
158 )));
159 }
160 Some(ConcreteFileVersion::V1)
161 }
162 Some(version) => Some(version),
163 None => Fragment::try_infer_version(fragments)?,
164 };
165 Ok(version.map(DataStorageFormat::new).unwrap_or_default())
166 }
167
168 pub async fn restore_old_manifest(
169 object_store: &ObjectStore,
170 commit_handler: &dyn CommitHandler,
171 base_path: &Path,
172 version: u64,
173 config: &ManifestBuildConfig,
174 tx_path: &str,
175 current_manifest: &Manifest,
176 ) -> Result<(Manifest, Vec<IndexMetadata>)> {
177 let location = commit_handler
178 .resolve_version_location(base_path, version, &object_store.inner)
179 .await?;
180 let mut manifest = read_manifest(object_store, &location.path, location.size).await?;
181 ensure_can_read_manifest(&manifest)?;
185 ensure_can_write_manifest(&manifest)?;
186 manifest.set_timestamp(config.timestamp_nanos);
187 manifest.transaction_file = Some(tx_path.to_string());
188 let indices = read_manifest_indexes(object_store, &location, &manifest).await?;
189 validate_flags(&manifest, &indices)?;
190 manifest.max_fragment_id = manifest
191 .max_fragment_id
192 .max(current_manifest.max_fragment_id);
193 if current_manifest.uses_non_reusable_field_ids() {
194 let Some(restored_max_field_id) = manifest.max_allocated_field_id else {
197 return Err(Error::invalid_input(format!(
198 "Cannot restore version {version}: non-reusable field IDs were activated after that version"
199 )));
200 };
201 manifest.max_allocated_field_id =
202 Some(restored_max_field_id.max(current_manifest.max_field_id()));
203 manifest.writer_feature_flags |= FLAG_NON_REUSABLE_FIELD_IDS;
204 }
205 manifest.next_row_id = manifest.next_row_id.max(current_manifest.next_row_id);
207 if current_manifest.uses_stable_row_ids() && !manifest.uses_stable_row_ids() {
210 return Err(Error::invalid_input(format!(
211 "Cannot restore version {version}: stable row ids were enabled \
212 after it, and turning them back off would let row addresses \
213 collide with ids this table has already used"
214 )));
215 }
216 inherit_sticky_feature_flags(&mut manifest, current_manifest)?;
217 Ok((manifest, indices))
218 }
219
220 pub fn logical_index_segments(indices: &[IndexMetadata]) -> LogicalIndexSegments {
226 let mut by_name: LogicalIndexSegments = BTreeMap::new();
227 for idx in indices.iter().filter(|idx| !is_system_index(idx)) {
228 by_name
229 .entry(idx.name.clone())
230 .or_default()
231 .push(CoverageIdentity {
232 uuid: idx.uuid,
233 fragment_bitmap: idx.fragment_bitmap.clone(),
234 });
235 }
236 for segments in by_name.values_mut() {
237 segments.sort_unstable_by_key(|segment| segment.uuid);
238 }
239 by_name
240 }
241
242 pub fn apply_mem_wal_index_coverage(
266 final_indices: &mut [IndexMetadata],
267 segments_before: &LogicalIndexSegments,
268 read_version_state: Option<ReadVersionState<'_>>,
269 new_version: u64,
270 ) -> Result<()> {
271 let Some(pos) = final_indices
272 .iter()
273 .position(|idx| idx.name == MEM_WAL_INDEX_NAME)
274 else {
275 return Ok(());
278 };
279
280 let mut details = load_mem_wal_index_details(final_indices[pos].clone())?;
281
282 if details.compacted_sstables.is_empty() && details.index_catchup.is_empty() {
285 return Ok(());
286 }
287
288 let segments_after = Self::logical_index_segments(final_indices);
289 let catchup_before = std::mem::take(&mut details.index_catchup);
290
291 let read_details = read_version_state
297 .map(|state| {
298 state
299 .indices
300 .iter()
301 .find(|idx| idx.name == MEM_WAL_INDEX_NAME)
302 .cloned()
303 .map(load_mem_wal_index_details)
304 .transpose()
305 })
306 .transpose()?
307 .flatten();
308 let shards: Vec<(Uuid, u64, u64)> = details
309 .compacted_sstables
310 .iter()
311 .map(|committed| {
312 let at_read = read_details
313 .as_ref()
314 .and_then(|read| {
315 read.compacted_sstables
316 .iter()
317 .find(|s| s.shard_id == committed.shard_id)
318 })
319 .map_or(0, |s| s.generation);
320 (
321 committed.shard_id,
322 committed.generation,
323 at_read.min(committed.generation),
324 )
325 })
326 .collect();
327
328 let read_fragments: Option<RoaringBitmap> = read_version_state.map(|state| {
331 state
332 .manifest
333 .fragments
334 .iter()
335 .map(|fragment| fragment.id as u32)
336 .collect()
337 });
338
339 let covers_read_version = |segments: &[CoverageIdentity]| -> bool {
340 let Some(required) = read_fragments.as_ref() else {
341 return false;
342 };
343 if required.is_empty() {
344 return false;
353 }
354 let mut covered = RoaringBitmap::new();
355 for segment in segments {
356 match segment.fragment_bitmap.as_ref() {
357 Some(bitmap) => covered |= bitmap,
358 None => return false,
360 }
361 }
362 required.is_subset(&covered)
363 };
364
365 let mut rebuilt: Vec<IndexCatchupProgress> = Vec::new();
366 for (name, after) in segments_after.iter() {
367 let unchanged = segments_before.get(name) == Some(after);
372 let carried = unchanged
373 .then(|| catchup_before.iter().find(|e| e.index_name == *name))
374 .flatten();
375 let proven = covers_read_version(after);
376
377 if carried.is_none() && !proven {
378 continue;
381 }
382
383 let generations = shards
384 .iter()
385 .map(|&(shard_id, committed, creditable)| {
386 let prior = carried
387 .and_then(|entry| entry.caught_up_generation_for_shard(&shard_id))
388 .unwrap_or(0);
389 let credited = if proven { creditable } else { 0 };
390 CompactedSsTable::new(shard_id, prior.max(credited).min(committed))
397 })
398 .collect::<Vec<_>>();
399 if generations.iter().all(|g| g.generation == 0) {
400 continue;
401 }
402 rebuilt.push(IndexCatchupProgress::new(name.clone(), generations));
403 }
404 rebuilt.sort_by(|a, b| a.index_name.cmp(&b.index_name));
405
406 let mut before_sorted = catchup_before;
407 before_sorted.sort_by(|a, b| a.index_name.cmp(&b.index_name));
408 if rebuilt == before_sorted {
409 return Ok(());
410 }
411
412 let dropped: Vec<&str> = before_sorted
413 .iter()
414 .map(|e| e.index_name.as_str())
415 .filter(|name| !rebuilt.iter().any(|kept| kept.index_name == *name))
416 .collect();
417 if !dropped.is_empty() {
418 log::info!(
420 "MemWAL index catch-up invalidated at version {new_version} for {dropped:?}: \
421 these indices changed and no longer cover the version this commit read"
422 );
423 }
424
425 details.index_catchup = rebuilt;
426 final_indices[pos] = new_mem_wal_index_meta(new_version, details)?;
427 Ok(())
428 }
429
430 pub fn withdraw_coverage_invalidated_after_build(
438 indices: &mut [IndexMetadata],
439 changed: &[String],
440 new_version: u64,
441 ) -> Result<()> {
442 if changed.is_empty() {
443 return Ok(());
444 }
445 let Some(pos) = indices
446 .iter()
447 .position(|idx| idx.name == MEM_WAL_INDEX_NAME)
448 else {
449 return Ok(());
450 };
451 let mut details = load_mem_wal_index_details(indices[pos].clone())?;
452 let before = details.index_catchup.len();
453 details
454 .index_catchup
455 .retain(|entry| !changed.contains(&entry.index_name));
456 if details.index_catchup.len() == before {
457 return Ok(());
458 }
459 log::info!(
460 "MemWAL index catch-up withdrawn at version {new_version} for {changed:?}: \
461 these indices were recalculated after their coverage was derived"
462 );
463 indices[pos] = new_mem_wal_index_meta(new_version, details)?;
464 Ok(())
465 }
466
467 pub fn prepare_indices(
488 &self,
489 current_manifest: Option<&Manifest>,
490 current_indices: Vec<IndexMetadata>,
491 config: &ManifestBuildConfig,
492 frag_reuse_ledger: Option<Arc<crate::system_index::frag_reuse::ledger::FragReuseLedger>>,
493 frag_reuse: FragReuseUpdate,
494 ) -> Result<PreparedIndices> {
495 crate::system_index::frag_reuse::gate::classify(
496 &self.operation,
497 current_manifest,
498 ¤t_indices,
499 &frag_reuse,
500 config.migration_next_row_id,
501 )?;
502 let mut prepared = current_indices.clone();
503 if let Some(manifest) = current_manifest {
504 let rewrites = Self::rewritten_physical_columns(
505 &self.operation,
506 &manifest.schema,
507 &manifest.fragments,
508 );
509 if !rewrites.is_empty() {
510 let lineage =
511 crate::system_index::frag_reuse::lineage::TaggedLineage::from_indices_with_ledger(
512 ¤t_indices,
513 frag_reuse_ledger,
514 )?;
515 let live: RoaringBitmap = manifest
516 .fragments
517 .iter()
518 .filter_map(|fragment| u32::try_from(fragment.id).ok())
519 .collect();
520 Self::withdraw_in_place_rewrites(
521 &mut prepared,
522 rewrites,
523 &manifest.schema,
524 &live,
525 lineage.as_ref(),
526 );
527 }
528 }
529 Ok(PreparedIndices::new(
530 current_manifest.map(|manifest| manifest.version),
531 current_indices,
532 prepared,
533 frag_reuse,
534 ))
535 }
536
537 pub fn build_manifest(
541 &self,
542 current_manifest: Option<&Manifest>,
543 current_indices: Vec<IndexMetadata>,
544 transaction_file_path: &str,
545 config: &ManifestBuildConfig,
546 ) -> Result<(Manifest, Vec<IndexMetadata>)> {
547 self.build_manifest_with_read_version(
548 current_manifest,
549 current_indices,
550 transaction_file_path,
551 config,
552 None,
553 )
554 }
555
556 pub fn build_manifest_with_read_version(
570 &self,
571 current_manifest: Option<&Manifest>,
572 current_indices: Vec<IndexMetadata>,
573 transaction_file_path: &str,
574 config: &ManifestBuildConfig,
575 read_version_state: Option<ReadVersionState<'_>>,
576 ) -> Result<(Manifest, Vec<IndexMetadata>)> {
577 let prepared = self.prepare_indices(
578 current_manifest,
579 current_indices,
580 config,
581 None,
582 FragReuseUpdate::None,
583 )?;
584 self.build_manifest_prepared(
585 current_manifest,
586 prepared,
587 transaction_file_path,
588 config,
589 read_version_state,
590 )
591 }
592
593 pub fn build_manifest_prepared(
599 &self,
600 current_manifest: Option<&Manifest>,
601 prepared: PreparedIndices,
602 transaction_file_path: &str,
603 config: &ManifestBuildConfig,
604 read_version_state: Option<ReadVersionState<'_>>,
605 ) -> Result<(Manifest, Vec<IndexMetadata>)> {
606 if prepared.manifest_version() != current_manifest.map(|manifest| manifest.version) {
607 return Err(Error::invalid_input(format!(
608 "the index list was prepared against manifest version {:?} but the manifest \
609 being built on is at version {:?}; every commit attempt prepares against the \
610 manifest it builds on",
611 prepared.manifest_version(),
612 current_manifest.map(|manifest| manifest.version)
613 )));
614 }
615 crate::system_index::frag_reuse::gate::classify(
618 &self.operation,
619 current_manifest,
620 prepared.original(),
621 prepared.frag_reuse_update(),
622 config.migration_next_row_id,
623 )?;
624 let (original_indices, prepared_indices, frag_reuse) = prepared.into_parts();
625 if config.use_stable_row_ids
626 && config.migration_next_row_id.is_none()
627 && current_manifest
628 .map(|m| !m.uses_stable_row_ids())
629 .unwrap_or_default()
630 {
631 return Err(Error::not_supported_source(
632 "This dataset was not created with the stable row ids feature. Please run `migrate_to_stable_row_ids` before attempting to use stable row ids".into(),
633 ));
634 }
635
636 let blocking_indices: Vec<&str> = original_indices
644 .iter()
645 .filter(|idx| idx.name != FRAG_REUSE_INDEX_NAME)
646 .map(|idx| idx.name.as_str())
647 .collect();
648 if config.migration_next_row_id.is_some() && !blocking_indices.is_empty() {
649 return Err(Error::invalid_input(format!(
650 "Cannot migrate to stable row IDs while indexes exist on the dataset. \
651 Drop the following indexes first, then re-run the migration, and \
652 recreate them afterwards: {}",
653 blocking_indices.join(", ")
654 )));
655 }
656 let mut reference_paths = match current_manifest {
657 Some(m) => m.base_paths.clone(),
658 None => HashMap::new(),
659 };
660
661 if let Operation::Overwrite {
662 initial_bases: Some(initial_bases),
663 ..
664 } = &self.operation
665 {
666 if current_manifest.is_none() {
667 for base_path in initial_bases.iter() {
671 if reference_paths.contains_key(&base_path.id) {
672 return Err(Error::invalid_input(format!(
673 "Duplicate base path ID {} detected. Base path IDs must be unique.",
674 base_path.id
675 )));
676 }
677 reference_paths.insert(base_path.id, base_path.clone());
678 }
679 } else {
680 return Err(Error::invalid_input(
683 "OVERWRITE mode cannot register new bases. This should have been caught by validation.",
684 ));
685 }
686 }
687
688 let mut schema = match self.operation {
690 Operation::Overwrite { ref schema, .. } => schema.clone(),
691 Operation::Merge { ref schema, .. } => schema.clone(),
692 Operation::Project { ref schema, .. } => schema.clone(),
693 _ => {
694 if let Some(current_manifest) = current_manifest {
695 current_manifest.schema.clone()
696 } else {
697 return Err(Error::internal(
698 "Cannot create a new dataset without a schema".to_string(),
699 ));
700 }
701 }
702 };
703
704 let mut fragment_id = current_manifest
709 .and_then(|m| m.max_fragment_id())
710 .map(|id| id + 1)
711 .unwrap_or(0);
712 let mut final_fragments = Vec::new();
713 let mut final_indices = prepared_indices;
714
715 if config.migration_next_row_id.is_some() {
718 final_indices.retain(|idx| idx.name != FRAG_REUSE_INDEX_NAME);
719 }
720
721 let mem_wal_segments_before = original_indices
727 .iter()
728 .any(|idx| idx.name == MEM_WAL_INDEX_NAME)
729 .then(|| Self::logical_index_segments(&original_indices));
730 drop(original_indices);
731
732 let mut next_row_id = {
733 match (current_manifest, config.use_stable_row_ids) {
736 (Some(manifest), _) if manifest.reader_feature_flags & FLAG_STABLE_ROW_IDS != 0 => {
737 Some(manifest.next_row_id)
738 }
739 (None, true) => Some(0),
740 (_, false) => None,
741 (Some(_), true) => {
742 if let Some(migration_nri) = config.migration_next_row_id {
744 Some(migration_nri)
745 } else {
746 return Err(Error::not_supported_source(
747 "This dataset was not created with the stable row ids feature. Please run `migrate_to_stable_row_ids` before attempting to use stable row ids".into(),
748 ));
749 }
750 }
751 }
752 };
753
754 let maybe_existing_fragments =
755 current_manifest
756 .map(|m| m.fragments.as_ref())
757 .ok_or_else(|| {
758 Error::internal(format!(
759 "No current manifest was provided while building manifest for operation {}",
760 self.operation.name()
761 ))
762 });
763
764 let new_version = current_manifest.map_or(1, |m| m.version + 1);
765
766 match &self.operation {
767 Operation::Clone { .. } => {
768 return Err(Error::internal(
769 "Clone operation should not enter build_manifest.".to_string(),
770 ));
771 }
772 Operation::Unknown { .. } => {
773 return Err(Error::not_supported(format!(
774 "Transaction {} has an operation written by a newer version of Lance \
775 and cannot be committed by this version",
776 self.uuid
777 )));
778 }
779 Operation::Append { fragments } => {
780 final_fragments.extend(maybe_existing_fragments?.clone());
781 let mut new_fragments =
782 Self::fragments_with_ids(fragments.clone(), &mut fragment_id)
783 .collect::<Vec<_>>();
784 if let Some(next_row_id) = &mut next_row_id {
785 Self::assign_row_ids(next_row_id, new_fragments.as_mut_slice())?;
786 for fragment in new_fragments.iter_mut() {
788 let version_meta = build_version_meta(fragment, new_version);
789 fragment.last_updated_at_version_meta = version_meta.clone();
790 fragment.created_at_version_meta = version_meta;
791 }
792 }
793 final_fragments.extend(new_fragments);
794 }
795 Operation::Delete {
796 updated_fragments,
797 deleted_fragment_ids,
798 ..
799 } => {
800 let deleted_ids: HashSet<u64> = deleted_fragment_ids.iter().copied().collect();
803 let updated_by_id: HashMap<u64, &Fragment> =
804 updated_fragments.iter().map(|f| (f.id, f)).collect();
805 final_fragments.extend(maybe_existing_fragments?.clone());
806 final_fragments.retain(|f| !deleted_ids.contains(&f.id));
807 final_fragments.iter_mut().for_each(|f| {
808 if let Some(updated) = updated_by_id.get(&f.id) {
809 f.deletion_file = updated.deletion_file.clone();
813 }
814 });
815 Self::retain_relevant_indices(&mut final_indices, &schema, &final_fragments)
816 }
817 Operation::Update {
818 removed_fragment_ids,
819 updated_fragments,
820 new_fragments,
821 fields_modified,
822 compacted_sstables,
823 fields_for_preserving_frag_bitmap,
824 update_mode,
825 updated_fragment_offsets,
826 ..
827 } => {
828 let existing_fragments = maybe_existing_fragments?;
830
831 let removed_ids: HashSet<u64> = removed_fragment_ids.iter().copied().collect();
834 let mut updated_by_id: HashMap<u64, &Fragment> =
835 HashMap::with_capacity(updated_fragments.len());
836 for fragment in updated_fragments {
837 updated_by_id.entry(fragment.id).or_insert(fragment);
838 }
839 let live_field_ids = schema
840 .fields_pre_order()
841 .map(|field| field.id)
842 .collect::<HashSet<_>>();
843 let updated_frags: Vec<Fragment> = existing_fragments
844 .iter()
845 .filter_map(|f| {
846 if removed_ids.contains(&f.id) {
847 return None;
848 }
849 if let Some(&updated) = updated_by_id.get(&f.id) {
850 let mut updated = updated.clone();
851 updated.overlays = f.overlays.clone();
857 let spilled = updated.spilled_row_lineage_field_ids();
862 updated.files.retain(|file| {
863 file.fields.iter().any(|field_id| {
864 live_field_ids.contains(field_id) || spilled.contains(field_id)
865 })
866 });
867 if matches!(update_mode, Some(RewriteColumns)) {
868 crate::format::overlay::tombstone_overlay_fields(
869 &mut updated.overlays,
870 fields_modified,
871 );
872 }
873 Some(updated)
874 } else {
875 Some(f.clone())
876 }
877 })
878 .collect();
879
880 if next_row_id.is_some() {
885 }
887
888 final_fragments.extend(updated_frags);
889
890 if next_row_id.is_some()
891 && matches!(update_mode, Some(RewriteColumns))
892 && let Some(UpdatedFragmentOffsets(off_map)) = updated_fragment_offsets
893 && !off_map.is_empty()
894 {
895 let prev_version = current_manifest.map(|m| m.version).unwrap_or(0);
896 for fragment in final_fragments.iter_mut() {
897 let Some(bitmap) = off_map.get(&fragment.id) else {
898 continue;
899 };
900 if !updated_by_id.contains_key(&fragment.id) {
905 continue;
906 }
907 if bitmap.is_empty() {
908 continue;
909 }
910 if fragment.last_updated_at_version_meta.is_none() {
914 continue;
915 }
916 let max_allowed = existing_fragments
917 .iter()
918 .find(|f| f.id == fragment.id)
919 .and_then(|f| f.physical_rows)
920 .unwrap_or(1 << 24);
921 if bitmap.len() as usize > max_allowed {
922 return Err(Error::invalid_input(format!(
923 "updatedFragmentOffsets cardinality {} exceeds fragment {} limit {}",
924 bitmap.len(),
925 fragment.id,
926 max_allowed
927 )));
928 }
929 if let Some(max_off) = bitmap.max()
930 && max_off as usize >= max_allowed
931 {
932 return Err(Error::invalid_input(format!(
933 "updatedFragmentOffsets max offset {} exceeds fragment {} limit {}",
934 max_off, fragment.id, max_allowed
935 )));
936 }
937 let offsets: Vec<usize> = bitmap.iter().map(|o| o as usize).collect();
938 crate::rowids::version::refresh_row_latest_update_meta_for_partial_frag_rewrite_cols(
939 fragment,
940 &offsets,
941 new_version,
942 prev_version,
943 &config.spilled_row_lineage,
944 )?;
945 }
946 }
947
948 let mut new_fragments =
952 Self::fragments_with_ids(new_fragments.clone(), &mut fragment_id)
953 .collect::<Vec<_>>();
954 for fragment in &mut new_fragments {
956 let spilled = fragment.spilled_row_lineage_field_ids();
957 fragment.files.retain(|file| {
958 file.fields.iter().any(|field_id| {
959 live_field_ids.contains(field_id) || spilled.contains(field_id)
960 })
961 });
962 }
963
964 if let Some(next_row_id) = &mut next_row_id {
967 Self::assign_row_ids(next_row_id, new_fragments.as_mut_slice())?;
968 }
969
970 if next_row_id.is_some() {
971 resolve_update_version_metadata(
972 existing_fragments,
973 new_fragments.as_mut_slice(),
974 new_version,
975 &config.spilled_row_lineage,
976 )?;
977 }
978
979 if config.use_stable_row_ids
980 && update_mode.is_some()
981 && *update_mode == Some(RewriteRows)
982 {
983 let pure_updated_frag_ids =
984 Self::collect_pure_rewrite_row_update_frags_ids(&new_fragments)?;
985
986 let original_fragment_ids: Vec<u64> = removed_fragment_ids
988 .iter()
989 .chain(updated_fragments.iter().map(|f| &f.id))
990 .copied()
991 .collect();
992
993 let original_overlaid_frags: HashMap<u32, &Fragment> = existing_fragments
998 .iter()
999 .filter(|f| {
1000 (removed_ids.contains(&f.id) || updated_by_id.contains_key(&f.id))
1001 && !f.overlays.is_empty()
1002 })
1003 .map(|f| (f.id as u32, f))
1004 .collect();
1005
1006 Self::register_pure_rewrite_rows_update_frags_in_indices(
1007 &mut final_indices,
1008 &pure_updated_frag_ids,
1009 &original_fragment_ids,
1010 fields_for_preserving_frag_bitmap,
1011 &original_overlaid_frags,
1012 &schema,
1013 )?;
1014 }
1015
1016 if let Some(next_row_id) = &mut next_row_id {
1017 Self::assign_row_ids(next_row_id, new_fragments.as_mut_slice())?;
1018 }
1022 let mut target_ids: HashSet<u64> = HashSet::new();
1024 target_ids.extend(new_fragments.iter().map(|f| f.id));
1025 final_fragments.extend(new_fragments);
1026 Self::retain_relevant_indices(&mut final_indices, &schema, &final_fragments);
1027
1028 if !compacted_sstables.is_empty() {
1029 update_mem_wal_index_compacted_sstables(
1030 &mut final_indices,
1031 new_version,
1032 compacted_sstables.clone(),
1033 )?;
1034 }
1035 }
1036 Operation::Overwrite { fragments, .. } => {
1037 let mut new_fragments = fragments.clone();
1043 for fragment in new_fragments.iter_mut() {
1044 fragment.id = fragment_id;
1045 fragment_id += 1;
1046 }
1047 if let Some(next_row_id) = &mut next_row_id {
1048 Self::assign_row_ids(next_row_id, new_fragments.as_mut_slice())?;
1049 for fragment in new_fragments.iter_mut() {
1051 let version_meta = build_version_meta(fragment, new_version);
1052 fragment.last_updated_at_version_meta = version_meta.clone();
1053 fragment.created_at_version_meta = version_meta;
1054 }
1055 }
1056 final_fragments.extend(new_fragments);
1057 final_indices = Vec::new();
1058 }
1059 Operation::Rewrite {
1060 groups,
1061 rewritten_indices,
1062 frag_reuse_index,
1063 } => {
1064 final_fragments.extend(maybe_existing_fragments?.clone());
1065 let current_version = current_manifest.map(|m| m.version).unwrap_or_default();
1066 Self::handle_rewrite_fragments(
1067 &mut final_fragments,
1068 groups,
1069 &mut fragment_id,
1070 current_version,
1071 next_row_id.as_ref(),
1072 )?;
1073
1074 let assembly = match &frag_reuse {
1082 FragReuseUpdate::Rewrite(assembly) => Some(assembly.as_ref()),
1083 _ => None,
1084 };
1085 let ordered_groups = Self::ordered_rewrite_groups(
1086 groups,
1087 assembly.map(|assembly| &assembly.reordered_sources),
1088 )?;
1089
1090 if next_row_id.is_some() {
1091 let remapped_new_ids: HashSet<Uuid> =
1100 rewritten_indices.iter().map(|idx| idx.new_id).collect();
1101 if !rewritten_indices.is_empty() {
1102 Self::handle_rewrite_indices(
1103 &mut final_indices,
1104 rewritten_indices,
1105 &ordered_groups,
1106 )?;
1107 }
1108 let deferred_remap = frag_reuse_index.is_some();
1114 for index in final_indices.iter_mut() {
1115 if is_frag_reuse_index_entry(index)
1118 || remapped_new_ids.contains(&index.uuid)
1119 {
1120 continue;
1121 }
1122 let results_are_row_addrs = index.results_are_row_addrs();
1123 if let Some(fragment_bitmap) = &mut index.fragment_bitmap {
1124 *fragment_bitmap = if results_are_row_addrs && !deferred_remap {
1125 Self::drop_rewritten_fragments(fragment_bitmap, &ordered_groups)
1135 } else {
1136 Self::recalculate_fragment_bitmap(fragment_bitmap, &ordered_groups)?
1137 };
1138 }
1139 }
1140 } else {
1141 Self::handle_rewrite_indices(
1142 &mut final_indices,
1143 rewritten_indices,
1144 &ordered_groups,
1145 )?;
1146 }
1147
1148 Self::prune_overlay_stale_fields_from_indices(&mut final_indices, groups);
1153
1154 match (frag_reuse_index, assembly) {
1155 (_, Some(assembly)) => {
1160 let existing_version = final_indices
1173 .iter()
1174 .find(|idx| idx.name == FRAG_REUSE_INDEX_NAME)
1175 .map(|idx| idx.dataset_version);
1176 if existing_version != assembly.base_entry_version {
1177 return Err(Error::invalid_input(format!(
1178 "the {} index entry in this manifest is at version {:?} but the \
1179 rewrite's entry was assembled against version {:?}; rebuild it \
1180 through the lance commit path, which assembles the entry against \
1181 the manifest it commits to",
1182 FRAG_REUSE_INDEX_NAME,
1183 existing_version,
1184 assembly.base_entry_version,
1185 )));
1186 }
1187 final_indices.retain(|idx| idx.name != assembly.entry.name);
1188 final_indices.push(assembly.entry.clone());
1189 }
1190 (Some(entry), None) => {
1194 final_indices.retain(|idx| idx.name != entry.name);
1195 final_indices.push(entry.clone());
1196 }
1197 (None, None) => {}
1198 }
1199 }
1200 Operation::CreateIndex {
1201 new_indices,
1202 removed_indices,
1203 ..
1204 } => {
1205 final_fragments.extend(maybe_existing_fragments?.clone());
1206 let removed_uuids = removed_indices
1207 .iter()
1208 .map(|old_index| old_index.uuid)
1209 .collect::<HashSet<_>>();
1210 let new_uuids = new_indices
1211 .iter()
1212 .map(|new_index| new_index.uuid)
1213 .collect::<HashSet<_>>();
1214 final_indices.retain(|existing_index| {
1215 !removed_uuids.contains(&existing_index.uuid)
1216 && !new_uuids.contains(&existing_index.uuid)
1217 });
1218 for new_index in new_indices {
1219 new_index.validate_covering_fields()?;
1220 }
1221 final_indices.extend(new_indices.clone());
1222
1223 let changed_names = new_indices
1227 .iter()
1228 .map(|index| index.name.as_str())
1229 .collect::<HashSet<_>>();
1230 let mut first_segments: HashMap<&str, &IndexMetadata> = HashMap::new();
1231 for index in &final_indices {
1232 if !changed_names.contains(index.name.as_str()) {
1233 continue;
1234 }
1235 if let Some(first) = first_segments.get(index.name.as_str()) {
1236 if first.results_are_row_addrs() != index.results_are_row_addrs() {
1237 return Err(Error::invalid_input(format!(
1238 "Index '{}' mixes row-ID and row-address segments {} (version {}) \
1239 and {} (version {}); rebuild every segment in one commit",
1240 index.name,
1241 first.uuid,
1242 first.index_version,
1243 index.uuid,
1244 index.index_version,
1245 )));
1246 }
1247 } else {
1248 first_segments.insert(index.name.as_str(), index);
1249 }
1250 }
1251 }
1252 Operation::ReserveFragments { .. } | Operation::UpdateConfig { .. } => {
1253 final_fragments.extend(maybe_existing_fragments?.clone());
1254 }
1255 Operation::Merge { fragments, .. } => {
1256 let existing_fragments = maybe_existing_fragments?;
1257 let mut merged_fragments = fragments.clone();
1258
1259 let prev_by_id: HashMap<u64, &Fragment> =
1264 existing_fragments.iter().map(|f| (f.id, f)).collect();
1265 let mut seen_prev_ids = HashSet::new();
1266 let is_new = merged_fragments
1267 .iter()
1268 .map(|f| !prev_by_id.contains_key(&f.id) || !seen_prev_ids.insert(f.id))
1269 .collect::<Vec<_>>();
1270
1271 for (fragment, is_new) in merged_fragments.iter_mut().zip(is_new.iter()) {
1275 if !*is_new {
1276 continue;
1277 }
1278 if fragment.row_id_meta.is_some() {
1282 return Err(Error::invalid_input(format!(
1283 "Merge operation includes new fragment (id {}) with row id \
1284 metadata. New fragments must not carry row ids; they are \
1285 assigned at commit time.",
1286 fragment.id
1287 )));
1288 }
1289 if fragment.id == 0 {
1290 fragment.id = fragment_id;
1291 fragment_id += 1;
1292 }
1293 }
1294
1295 if let Some(next_row_id) = &mut next_row_id {
1296 for (fragment, is_new) in merged_fragments.iter_mut().zip(is_new.iter()) {
1297 if *is_new {
1298 Self::assign_row_ids(next_row_id, std::slice::from_mut(fragment))?;
1299 let version_meta = build_version_meta(fragment, new_version);
1300 fragment.last_updated_at_version_meta = version_meta.clone();
1301 fragment.created_at_version_meta = version_meta;
1302 } else if let Some(prev) = prev_by_id.get(&fragment.id)
1303 && merge_fragment_physically_rewritten(prev, fragment)
1304 {
1305 crate::rowids::version::refresh_row_latest_update_meta_for_full_frag_rewrite_cols(
1306 fragment,
1307 new_version,
1308 )?;
1309 }
1310 }
1311 }
1312 let remaining_field_ids = schema.field_ids().into_iter().collect::<HashSet<_>>();
1313 Self::retain_relevant_overlays(&mut merged_fragments, &remaining_field_ids);
1314 final_fragments.extend(merged_fragments);
1315
1316 Self::retain_relevant_indices(&mut final_indices, &schema, &final_fragments)
1323 }
1324 Operation::Project { .. } => {
1325 final_fragments.extend(maybe_existing_fragments?.clone());
1326
1327 let remaining_field_ids = schema
1332 .fields_pre_order()
1333 .map(|f| f.id)
1334 .collect::<HashSet<_>>();
1335 for fragment in final_fragments.iter_mut() {
1336 let spilled = fragment.spilled_row_lineage_field_ids();
1337 fragment.files.retain(|file| {
1338 file.fields.iter().any(|field_id| {
1339 remaining_field_ids.contains(field_id) || spilled.contains(field_id)
1340 })
1341 });
1342 }
1343 Self::retain_relevant_overlays(&mut final_fragments, &remaining_field_ids);
1344
1345 Self::retain_relevant_indices(&mut final_indices, &schema, &final_fragments)
1348 }
1349 Operation::Restore { .. } => {
1350 unreachable!()
1351 }
1352 Operation::DataReplacement { replacements } => {
1353 log::warn!(
1354 "Building manifest with DataReplacement operation. This operation is not stable yet, please use with caution."
1355 );
1356
1357 let (old_fragment_ids, new_datafiles): (Vec<&u64>, Vec<&DataFile>) = replacements
1358 .iter()
1359 .map(|DataReplacementGroup(fragment_id, new_file)| (fragment_id, new_file))
1360 .unzip();
1361
1362 if new_datafiles
1366 .iter()
1367 .map(|f| f.fields.clone())
1368 .collect::<HashSet<_>>()
1369 .len()
1370 > 1
1371 {
1372 let field_info = new_datafiles
1373 .iter()
1374 .enumerate()
1375 .map(|(id, f)| (id, f.fields.clone()))
1376 .fold("".to_string(), |acc, (id, fields)| {
1377 format!("{}File {}: {:?}\n", acc, id, fields)
1378 });
1379
1380 return Err(Error::invalid_input(format!(
1381 "All new data files must have the same fields, but found different fields:\n{field_info}"
1382 )));
1383 }
1384
1385 let existing_fragments = maybe_existing_fragments?;
1386
1387 let replaced_fields: Vec<u32> = new_datafiles
1389 .first()
1390 .map(|f| {
1391 f.schema(&schema)
1392 .field_ids()
1393 .iter()
1394 .chain(f.fields.iter())
1395 .filter(|&&id| id >= 0)
1396 .map(|&id| id as u32)
1397 .collect()
1398 })
1399 .unwrap_or_default();
1400
1401 for (frag_id, new_file) in old_fragment_ids.iter().zip(new_datafiles) {
1404 let frag = existing_fragments
1405 .iter()
1406 .find(|f| f.id == **frag_id)
1407 .ok_or_else(|| {
1408 Error::invalid_input(
1409 "Fragment being replaced not found in existing fragments",
1410 )
1411 })?;
1412 let mut new_frag = frag.clone();
1413
1414 let replacement_ids = new_file
1418 .schema(&schema)
1419 .field_ids()
1420 .into_iter()
1421 .chain(new_file.fields.iter().copied())
1422 .collect::<HashSet<_>>();
1423
1424 let mut columns_covered = HashSet::new();
1427 let mut replaced_in_place = false;
1430 for file in &mut new_frag.files {
1431 if file.fields == new_file.fields
1432 && file.file_major_version == new_file.file_major_version
1433 && file.file_minor_version == new_file.file_minor_version
1434 {
1435 file.path = new_file.path.clone();
1437 file.file_size_bytes = new_file.file_size_bytes.clone();
1438 file.base_id = new_file.base_id;
1439 replaced_in_place = true;
1440 }
1441 columns_covered.extend(file.schema(&schema).field_ids());
1442 columns_covered.extend(file.fields.iter().copied());
1443 }
1444 new_file.file_version()?;
1447
1448 if columns_covered.is_disjoint(&replacement_ids) {
1453 new_frag.files.push(new_file.clone());
1454 } else if !replaced_in_place
1455 && replacement_ids.is_subset(&columns_covered)
1456 && new_frag.files.iter().all(|file| {
1457 file.file_version()
1458 .is_ok_and(|version| version != ConcreteFileVersion::V1)
1459 || file
1460 .fields
1461 .iter()
1462 .all(|field| !replacement_ids.contains(field))
1463 })
1464 {
1465 for file in &mut new_frag.files {
1477 if file.file_version()? == ConcreteFileVersion::V1 {
1479 continue;
1480 }
1481 file.fields = file
1482 .fields
1483 .iter()
1484 .map(|field| {
1485 if replacement_ids.contains(field) {
1486 TOMBSTONE_FIELD_ID
1487 } else {
1488 *field
1489 }
1490 })
1491 .collect::<Vec<_>>()
1492 .into();
1493 }
1494 let live_ids = schema
1505 .fields_pre_order()
1506 .map(|field| field.id)
1507 .collect::<HashSet<i32>>();
1508 let spilled = new_frag.spilled_row_lineage_field_ids();
1509 new_frag.files.retain(|file| {
1510 file.fields
1511 .iter()
1512 .any(|f| live_ids.contains(f) || spilled.contains(f))
1513 });
1514 new_frag.files.push(new_file.clone());
1515 }
1516
1517 if &new_frag == frag {
1519 return Err(Error::invalid_input(
1520 "Expected to modify the fragment but no changes were made. This means the new data files does not align with any exiting datafiles. Please check if the schema of the new data files matches the schema of the old data files including the file major and minor versions",
1521 ));
1522 }
1523
1524 let (mut superseded, newer): (Vec<_>, Vec<_>) = new_frag
1531 .overlays
1532 .drain(..)
1533 .partition(|overlay| overlay.committed_version <= self.read_version);
1534 crate::format::overlay::tombstone_overlay_fields(
1535 &mut superseded,
1536 &replaced_fields,
1537 );
1538 superseded.extend(newer);
1539 new_frag.overlays = superseded;
1540
1541 final_fragments.push(new_frag);
1542 }
1543
1544 let fragments_changed = old_fragment_ids
1545 .iter()
1546 .cloned()
1547 .cloned()
1548 .collect::<HashSet<_>>();
1549
1550 let unmodified_fragments = existing_fragments
1552 .iter()
1553 .filter(|f| !fragments_changed.contains(&f.id))
1554 .cloned()
1555 .collect::<Vec<_>>();
1556
1557 final_fragments.extend(unmodified_fragments);
1558
1559 if next_row_id.is_some() {
1563 let new_version = current_manifest.map_or(1, |m| m.version + 1);
1564 for fragment in final_fragments
1565 .iter_mut()
1566 .filter(|f| fragments_changed.contains(&f.id))
1567 {
1568 crate::rowids::version::refresh_row_latest_update_meta_for_full_frag_rewrite_cols(
1569 fragment,
1570 new_version,
1571 )?;
1572 }
1573 }
1574
1575 }
1578 Operation::DataOverlay { groups } => {
1579 let new_version = current_manifest.map_or(1, |m| m.version + 1);
1583
1584 let existing_fragments = maybe_existing_fragments?;
1585 let mut overlays_by_fragment: HashMap<u64, Vec<&DataOverlayFile>> = HashMap::new();
1589 for group in groups {
1590 overlays_by_fragment
1591 .entry(group.fragment_id)
1592 .or_default()
1593 .extend(group.overlays.iter());
1594 }
1595
1596 let existing_fragment_ids: HashSet<u64> =
1600 existing_fragments.iter().map(|f| f.id).collect();
1601 for fragment_id in overlays_by_fragment.keys() {
1602 if !existing_fragment_ids.contains(fragment_id) {
1603 return Err(Error::invalid_input(format!(
1604 "DataOverlay targets fragment {fragment_id}, which does not exist"
1605 )));
1606 }
1607 }
1608
1609 for fragment in existing_fragments {
1610 let mut fragment = fragment.clone();
1611 if let Some(new_overlays) = overlays_by_fragment.get(&fragment.id) {
1612 if next_row_id.is_some() {
1613 let mut covered_offsets = RoaringBitmap::new();
1614 for overlay in new_overlays {
1615 match &overlay.coverage {
1616 OverlayCoverage::Shared(bitmap) => {
1617 covered_offsets |= bitmap.as_ref();
1618 }
1619 OverlayCoverage::PerField(bitmaps) => {
1620 for bitmap in bitmaps {
1621 covered_offsets |= bitmap.as_ref();
1622 }
1623 }
1624 }
1625 }
1626 if !covered_offsets.is_empty() {
1627 let covered_offsets: Vec<usize> = covered_offsets
1628 .iter()
1629 .map(|offset| offset as usize)
1630 .collect();
1631 crate::rowids::version::refresh_row_latest_update_meta_for_partial_frag_rewrite_cols(
1635 &mut fragment,
1636 &covered_offsets,
1637 new_version,
1638 1,
1639 &config.spilled_row_lineage,
1640 )?;
1641 }
1642 }
1643 fragment
1646 .overlays
1647 .extend(new_overlays.iter().map(|&overlay| {
1648 let mut overlay = overlay.clone();
1649 overlay.committed_version = new_version;
1650 overlay
1651 }));
1652 }
1653 final_fragments.push(fragment);
1654 }
1655 }
1656 Operation::UpdateMemWalState {
1657 compacted_sstables, ..
1658 } => {
1659 final_fragments.extend(maybe_existing_fragments?.clone());
1661 update_mem_wal_index_compacted_sstables(
1662 &mut final_indices,
1663 new_version,
1664 compacted_sstables.clone(),
1665 )?;
1666 }
1667 Operation::UpdateBases { .. } => {
1668 final_fragments.extend(maybe_existing_fragments?.clone());
1671 }
1672 };
1673
1674 final_fragments.sort_by_key(|frag| frag.id);
1676
1677 Self::remove_tombstoned_data_files(&mut final_fragments);
1679
1680 for fragment in &final_fragments {
1684 if !fragment.overlays.is_empty() {
1685 crate::format::overlay::verify_overlays_newest_last(&fragment.overlays)?;
1686 }
1687 }
1688
1689 for fragment in &final_fragments {
1697 fragment.validate_row_lineage_carriers()?;
1698 }
1699
1700 let user_requested_version = match (&config.storage_format, config.use_legacy_format) {
1701 (Some(storage_format), _) => Some(storage_format.lance_file_format()),
1702 (None, Some(true)) => Some(ConcreteFileVersion::V1),
1703 (None, Some(false)) => Some(ConcreteFileVersion::V2_0),
1704 (None, None) => None,
1705 };
1706
1707 if let Some(segments_before) = mem_wal_segments_before.as_ref() {
1711 Self::apply_mem_wal_index_coverage(
1712 &mut final_indices,
1713 segments_before,
1714 read_version_state,
1715 new_version,
1716 )?;
1717 }
1718
1719 let legacy_blob_ids = schema
1724 .fields_pre_order()
1725 .filter(|field| field.is_blob() && !field.is_blob_v2())
1726 .map(|field| field.id)
1727 .collect::<HashSet<_>>();
1728 if !legacy_blob_ids.is_empty() {
1729 let mut blob_v2_fields = std::collections::BTreeSet::new();
1730 for file in final_fragments
1731 .iter()
1732 .flat_map(|fragment| fragment.referenced_lance_files())
1733 {
1734 if matches!(
1735 file.file_version()?,
1736 ConcreteFileVersion::V2_2 | ConcreteFileVersion::V2_3
1737 ) {
1738 blob_v2_fields.extend(
1739 file.fields
1740 .iter()
1741 .copied()
1742 .filter(|id| legacy_blob_ids.contains(id)),
1743 );
1744 }
1745 }
1746 let mut next_field_id = current_manifest
1747 .map(|manifest| manifest.max_field_id())
1748 .unwrap_or(-1)
1749 .max(schema.max_field_id().unwrap_or(-1))
1750 + 1;
1751 for id in blob_v2_fields {
1752 if let Some(field) = schema.mut_field_by_id(id) {
1753 field.promote_blob_v2()?;
1754 field.set_id(field.parent_id, &mut next_field_id);
1755 }
1756 }
1757 }
1758
1759 let mut manifest = if let Some(current_manifest) = current_manifest {
1760 let mut prev_manifest =
1763 Manifest::new_from_previous(current_manifest, schema, Arc::new(final_fragments));
1764
1765 if let (Some(user_requested_version), Operation::Overwrite { .. }) =
1766 (user_requested_version, &self.operation)
1767 {
1768 prev_manifest.data_storage_format = DataStorageFormat::new(user_requested_version);
1772 }
1773
1774 prev_manifest
1775 } else {
1776 let data_storage_format =
1777 Self::data_storage_format_from_files(&final_fragments, user_requested_version)?;
1778 Manifest::new(
1779 schema,
1780 Arc::new(final_fragments),
1781 data_storage_format,
1782 reference_paths,
1783 )
1784 };
1785
1786 if config.activate_non_reusable_field_ids {
1787 let already_active = current_manifest
1788 .map(|manifest| manifest.uses_non_reusable_field_ids())
1789 .unwrap_or(false);
1790 if !already_active {
1791 manifest.activate_non_reusable_field_ids();
1792 manifest.writer_feature_flags |= FLAG_NON_REUSABLE_FIELD_IDS;
1793 }
1794 }
1795
1796 manifest.tag.clone_from(&self.tag);
1797
1798 if config.auto_set_feature_flags {
1799 let inherited = current_manifest
1803 .map(|m| m.uses_stable_row_ids())
1804 .unwrap_or(false);
1805 let use_stable_row_ids = config.use_stable_row_ids || inherited;
1806 apply_feature_flags(
1807 &mut manifest,
1808 use_stable_row_ids,
1809 config.disable_transaction_file,
1810 )?;
1811 }
1812 if final_indices
1827 .iter()
1828 .any(|index| !index.covering_fields.is_empty())
1829 {
1830 manifest.reader_feature_flags |= FLAG_COVERED_INDEX_METADATA;
1831 manifest.writer_feature_flags |= FLAG_COVERED_INDEX_METADATA;
1832 }
1833
1834 if let Some(current_manifest) = current_manifest {
1835 inherit_sticky_feature_flags(&mut manifest, current_manifest)?;
1836 }
1837
1838 manifest.set_timestamp(config.timestamp_nanos);
1839
1840 manifest.update_max_fragment_id();
1841 manifest.update_max_field_id();
1842
1843 match &self.operation {
1844 Operation::Overwrite {
1845 config_upsert_values: Some(tm),
1846 ..
1847 } => {
1848 manifest.config_mut().extend(tm.clone());
1849 }
1850 Operation::UpdateConfig {
1851 config_updates,
1852 table_metadata_updates,
1853 schema_metadata_updates,
1854 field_metadata_updates,
1855 } => {
1856 if let Some(config_updates) = config_updates {
1857 let mut config = manifest.config.clone();
1858 apply_update_map(&mut config, config_updates);
1859 manifest.config = config;
1860 }
1861 if let Some(table_metadata_updates) = table_metadata_updates {
1862 let mut table_metadata = manifest.table_metadata.clone();
1863 apply_update_map(&mut table_metadata, table_metadata_updates);
1864 manifest.table_metadata = table_metadata;
1865 }
1866 if let Some(schema_metadata_updates) = schema_metadata_updates {
1867 let mut schema_metadata = manifest.schema.metadata.clone();
1868 apply_update_map(&mut schema_metadata, schema_metadata_updates);
1869 manifest.schema.metadata = schema_metadata;
1870 }
1871 let primary_key_before: Vec<i32> = manifest
1880 .schema
1881 .unenforced_primary_key()
1882 .iter()
1883 .map(|field| field.id)
1884 .collect();
1885 let writes_primary_key = field_metadata_updates.values().any(|update| {
1886 update.update_entries.iter().any(|entry| {
1887 entry.key == LANCE_UNENFORCED_PRIMARY_KEY
1888 || entry.key == LANCE_UNENFORCED_PRIMARY_KEY_POSITION
1889 })
1890 });
1891 let clustering_key_before: Vec<i32> = manifest
1892 .schema
1893 .unenforced_clustering_key()
1894 .iter()
1895 .map(|field| field.id)
1896 .collect();
1897 let writes_clustering_key = field_metadata_updates.values().any(|update| {
1898 update
1899 .update_entries
1900 .iter()
1901 .any(|entry| entry.key == LANCE_UNENFORCED_CLUSTERING_KEY_POSITION)
1902 });
1903 for (field_id, field_metadata_update) in field_metadata_updates {
1904 if let Some(field) = manifest.schema.field_by_id_mut(*field_id) {
1905 apply_update_map(&mut field.metadata, field_metadata_update);
1906 field.unenforced_primary_key_position = field
1908 .metadata
1909 .get(LANCE_UNENFORCED_PRIMARY_KEY_POSITION)
1910 .and_then(|s| s.parse::<u32>().ok())
1911 .or_else(|| {
1912 field
1913 .metadata
1914 .get(LANCE_UNENFORCED_PRIMARY_KEY)
1915 .filter(|s| str_is_truthy(s))
1916 .map(|_| 0)
1917 });
1918 field.unenforced_clustering_key_position = field
1921 .metadata
1922 .get(LANCE_UNENFORCED_CLUSTERING_KEY_POSITION)
1923 .and_then(|s| s.parse::<u32>().ok());
1924 } else {
1925 return Err(Error::invalid_input_source(
1926 format!("Field with id {} does not exist", field_id).into(),
1927 ));
1928 }
1929 }
1930 let primary_key_after: Vec<i32> = manifest
1931 .schema
1932 .unenforced_primary_key()
1933 .iter()
1934 .map(|field| field.id)
1935 .collect();
1936 if !primary_key_before.is_empty() {
1937 if writes_primary_key || primary_key_after != primary_key_before {
1940 return Err(Error::invalid_input(
1941 "the unenforced primary key is a reserved key and cannot be changed once set",
1942 ));
1943 }
1944 } else if writes_primary_key && primary_key_after.is_empty() {
1945 return Err(Error::invalid_input(
1949 "the unenforced primary key is a reserved key and cannot be set to an invalid value",
1950 ));
1951 }
1952 if writes_primary_key {
1953 manifest.schema.verify_primary_key()?;
1956 }
1957 let clustering_key_after: Vec<i32> = manifest
1958 .schema
1959 .unenforced_clustering_key()
1960 .iter()
1961 .map(|field| field.id)
1962 .collect();
1963 if !clustering_key_before.is_empty() {
1964 if writes_clustering_key || clustering_key_after != clustering_key_before {
1967 return Err(Error::invalid_input(
1968 "the unenforced clustering key is a reserved key and cannot be changed once set",
1969 ));
1970 }
1971 } else if writes_clustering_key && clustering_key_after.is_empty() {
1972 return Err(Error::invalid_input(
1976 "the unenforced clustering key is a reserved key and cannot be set to an invalid value",
1977 ));
1978 }
1979 }
1980 _ => {}
1981 }
1982
1983 apply_range_segments(&mut manifest, current_manifest)?;
1984
1985 if let Operation::UpdateBases { new_bases } = &self.operation {
1987 for new_base in new_bases {
1989 if let Some(existing_base) = manifest
1991 .base_paths
1992 .values()
1993 .find(|bp| bp.name == new_base.name || bp.path == new_base.path)
1994 {
1995 return Err(Error::invalid_input(format!(
1996 "Conflict detected: Base path with name '{:?}' or path '{}' already exists. Existing: name='{:?}', path='{}'",
1997 new_base.name, new_base.path, existing_base.name, existing_base.path
1998 )));
1999 }
2000
2001 let mut base_to_add = new_base.clone();
2003 if base_to_add.id == 0 {
2004 base_to_add.id = crate::format::BasePath::unused_id(
2005 manifest
2006 .base_paths
2007 .keys()
2008 .copied()
2009 .chain(std::iter::once(0)),
2010 )?;
2011 } else if manifest.has_managed_blobs()
2012 && let Some(existing) = manifest.base_paths.get(&base_to_add.id)
2013 {
2014 return Err(Error::invalid_input(format!(
2015 "Cannot replace base ID {} bound to {:?} with {:?}",
2016 base_to_add.id, existing.path, base_to_add.path
2017 )));
2018 }
2019
2020 manifest.base_paths.insert(base_to_add.id, base_to_add);
2021 }
2022 }
2023
2024 if let Operation::ReserveFragments { num_fragments } = self.operation {
2025 manifest.max_fragment_id = Some(manifest.max_fragment_id.unwrap_or(0) + num_fragments);
2026 }
2027
2028 manifest.transaction_file = Some(transaction_file_path.to_string());
2029
2030 if let Some(next_row_id) = next_row_id {
2031 manifest.next_row_id = next_row_id;
2032 }
2033
2034 if final_indices.iter().any(is_tagged) {
2035 manifest.reader_feature_flags |= FLAG_FRAGMENT_REUSE_INDEX;
2036 manifest.writer_feature_flags |= FLAG_FRAGMENT_REUSE_INDEX;
2037 }
2038
2039 if managed_blobs_enabled(&manifest.schema)? {
2040 manifest.reader_feature_flags |= FLAG_MANAGED_BLOBS;
2041 manifest.writer_feature_flags |= FLAG_MANAGED_BLOBS;
2042 }
2043
2044 validate_non_reusable_field_id_flags(&manifest)?;
2045 Ok((manifest, final_indices))
2046 }
2047
2048 fn remove_tombstoned_data_files(fragments: &mut [Fragment]) {
2051 for fragment in fragments {
2052 fragment.files.retain(|file| {
2053 file.fields.iter().any(|&field_id| field_id != -2)
2055 });
2056 }
2057 }
2058
2059 fn retain_relevant_overlays(fragments: &mut [Fragment], remaining_field_ids: &HashSet<i32>) {
2061 for fragment in fragments {
2062 fragment.overlays.retain(|overlay| {
2063 overlay
2064 .data_file
2065 .fields
2066 .iter()
2067 .any(|field_id| remaining_field_ids.contains(field_id))
2068 });
2069 }
2070 }
2071
2072 fn drop_rewritten_fragments(old: &RoaringBitmap, groups: &[RewriteGroup]) -> RoaringBitmap {
2075 let mut new_bitmap = old.clone();
2076 for group in groups {
2077 for old_fragment in &group.old_fragments {
2078 new_bitmap.remove(old_fragment.id as u32);
2079 }
2080 }
2081 new_bitmap
2082 }
2083}
2084
2085#[cfg(test)]
2086mod tests {
2087 use super::*;
2088 use crate::format::overlay::OverlayCoverage;
2089 use crate::format::pb;
2090 use crate::format::{
2091 DeletionFile, DeletionFileType, ROW_CREATED_AT_VERSION_FIELD_ID, ROW_ID_FIELD_ID,
2092 ROW_LAST_UPDATED_AT_VERSION_FIELD_ID, RowDatasetVersionMeta, RowDatasetVersionSequence,
2093 RowIdMeta,
2094 };
2095 use crate::rowids::{RowIdSequence, write_row_ids};
2096 use crate::transaction::test_support::{
2097 created_at_versions, default_build_config, last_updated_at_versions,
2098 make_stable_row_id_manifest, overlay_with_field, sample_index_metadata, sample_manifest,
2099 tagged_entry,
2100 };
2101 use crate::transaction::{DataOverlayGroup, UpdateMode, validate_operation};
2102 use arrow_schema::{DataType, Field as ArrowField, Schema as ArrowSchema};
2103 use lance_core::datatypes::Schema as LanceSchema;
2104 use lance_file::version::{ConcreteFileVersion, LanceFileVersion};
2105 use lance_io::utils::CachedFileSize;
2106 use rstest::rstest;
2107 use std::collections::HashMap;
2108 use std::sync::Arc;
2109 use uuid::Uuid;
2110
2111 fn tagged_sample_manifest() -> (Manifest, IndexMetadata) {
2114 let mut manifest = sample_manifest();
2115 manifest.reader_feature_flags |= FLAG_FRAGMENT_REUSE_INDEX;
2116 manifest.writer_feature_flags |= FLAG_FRAGMENT_REUSE_INDEX;
2117 (manifest, tagged_entry(&[(&[5], &[0])]))
2118 }
2119
2120 fn row_mutation(kind: &str) -> Operation {
2121 match kind {
2122 "delete" => Operation::Delete {
2123 updated_fragments: vec![],
2124 deleted_fragment_ids: vec![0],
2125 predicate: "true".into(),
2126 },
2127 "update" => crate::transaction::test_support::update_txn(vec![]).operation,
2128 "config" => Operation::UpdateConfig {
2129 config_updates: None,
2130 table_metadata_updates: None,
2131 schema_metadata_updates: None,
2132 field_metadata_updates: HashMap::new(),
2133 },
2134 _ => unreachable!(),
2135 }
2136 }
2137
2138 #[rstest::rstest]
2143 #[case::delete("delete")]
2144 #[case::update("update")]
2145 #[case::config("config")]
2146 #[case::project("project")]
2147 #[case::merge("merge")]
2148 fn tagged_history_allows_row_mutations(#[case] kind: &str) {
2149 let (manifest, fri) = tagged_sample_manifest();
2150 let operation = match kind {
2151 "project" => Operation::Project {
2152 schema: manifest.schema.clone(),
2153 preserves_nullability: true,
2154 },
2155 "merge" => Operation::Merge {
2156 fragments: manifest.fragments.as_ref().clone(),
2157 schema: manifest.schema.clone(),
2158 preserves_nullability: true,
2159 },
2160 _ => row_mutation(kind),
2161 };
2162 let transaction = Transaction::new(manifest.version, operation, None);
2163 let (_, indices) = transaction
2164 .build_manifest(
2165 Some(&manifest),
2166 vec![fri.clone()],
2167 "txn",
2168 &default_build_config(),
2169 )
2170 .unwrap();
2171 let carried = indices
2172 .iter()
2173 .find(|idx| idx.name == crate::system_index::frag_reuse::FRAG_REUSE_INDEX_NAME)
2174 .expect("the tagged entry is carried through");
2175 assert_eq!(carried.uuid, fri.uuid);
2176 assert_eq!(carried.index_version, fri.index_version);
2177 }
2178
2179 #[test]
2182 fn tagged_history_overwrite_discards_the_entry() {
2183 let (manifest, fri) = tagged_sample_manifest();
2184 let transaction = Transaction::new(
2185 manifest.version,
2186 Operation::Overwrite {
2187 fragments: vec![Fragment::new(0)],
2188 schema: manifest.schema.clone(),
2189 config_upsert_values: None,
2190 initial_bases: None,
2191 },
2192 None,
2193 );
2194 let (new_manifest, indices) = transaction
2195 .build_manifest(Some(&manifest), vec![fri], "txn", &default_build_config())
2196 .unwrap();
2197 assert!(indices.is_empty(), "{indices:?}");
2198 assert_ne!(
2199 new_manifest.reader_feature_flags & FLAG_FRAGMENT_REUSE_INDEX,
2200 0
2201 );
2202 }
2203
2204 #[test]
2207 fn tagged_history_refuses_stable_row_id_migration() {
2208 let (manifest, fri) = tagged_sample_manifest();
2209 let transaction = Transaction::new(
2210 manifest.version,
2211 Operation::Merge {
2212 fragments: manifest.fragments.as_ref().clone(),
2213 schema: manifest.schema.clone(),
2214 preserves_nullability: true,
2215 },
2216 None,
2217 );
2218 let config = ManifestBuildConfig {
2219 migration_next_row_id: Some(100),
2220 ..default_build_config()
2221 };
2222 let error = transaction
2223 .build_manifest(Some(&manifest), vec![fri], "txn", &config)
2224 .unwrap_err();
2225 assert!(matches!(error, Error::NotSupported { .. }), "{error}");
2226 assert!(
2227 error.to_string().contains("stable row id migration"),
2228 "{error}"
2229 );
2230 }
2231
2232 fn manifest_with_file(path: &str) -> Manifest {
2234 let mut manifest = sample_manifest();
2235 manifest.fragments = Arc::new(vec![Fragment::new(0).with_file(
2236 path,
2237 vec![0],
2238 vec![0],
2239 ConcreteFileVersion::V2_0,
2240 None,
2241 )]);
2242 manifest.reader_feature_flags |= FLAG_FRAGMENT_REUSE_INDEX;
2243 manifest.writer_feature_flags |= FLAG_FRAGMENT_REUSE_INDEX;
2244 manifest
2245 }
2246
2247 fn translating_segment_and_entry() -> (IndexMetadata, IndexMetadata) {
2251 let mut segment = sample_index_metadata("id_idx");
2252 segment.fragment_bitmap = Some([5u32].into_iter().collect());
2253 (segment, tagged_entry(&[(&[5], &[0])]))
2254 }
2255
2256 fn user_bitmap(indices: &[IndexMetadata]) -> RoaringBitmap {
2257 indices
2258 .iter()
2259 .find(|idx| idx.name == "id_idx")
2260 .unwrap()
2261 .fragment_bitmap
2262 .clone()
2263 .unwrap()
2264 }
2265
2266 fn in_place_rewrite(kind: &str, manifest: &Manifest, fields: Vec<u32>) -> Operation {
2268 match kind {
2269 "update" => {
2270 let mut update = crate::transaction::test_support::update_txn(vec![]);
2271 let Operation::Update {
2272 updated_fragments,
2273 fields_modified,
2274 ..
2275 } = &mut update.operation
2276 else {
2277 unreachable!()
2278 };
2279 *updated_fragments = vec![Fragment::new(0)];
2280 *fields_modified = fields;
2281 update.operation
2282 }
2283 "merge" => Operation::Merge {
2284 fragments: vec![Fragment::new(0).with_file(
2285 "b.lance",
2286 fields.iter().map(|f| *f as i32).collect(),
2287 vec![0],
2288 ConcreteFileVersion::V2_0,
2289 None,
2290 )],
2291 schema: manifest.schema.clone(),
2292 preserves_nullability: true,
2293 },
2294 "data_replacement" => Operation::DataReplacement {
2295 replacements: vec![DataReplacementGroup(
2296 0,
2297 DataFile::new(
2298 "b.lance",
2299 fields.iter().map(|f| *f as i32).collect(),
2300 vec![0],
2301 ConcreteFileVersion::V2_0,
2302 None,
2303 None,
2304 ),
2305 )],
2306 },
2307 _ => unreachable!(),
2308 }
2309 }
2310
2311 #[rstest::rstest]
2317 #[case::update("update")]
2318 #[case::merge("merge")]
2319 #[case::data_replacement("data_replacement")]
2320 fn tagged_history_withdraws_in_place_rewrite_covered_only_by_translation(#[case] kind: &str) {
2321 let manifest = manifest_with_file("a.lance");
2322 let (segment, fri) = translating_segment_and_entry();
2323 let transaction = Transaction::new(
2324 manifest.version,
2325 in_place_rewrite(kind, &manifest, vec![0]),
2326 None,
2327 );
2328 let (_, indices) = transaction
2329 .build_manifest(
2330 Some(&manifest),
2331 vec![segment, fri.clone()],
2332 "txn",
2333 &default_build_config(),
2334 )
2335 .unwrap();
2336 assert!(
2337 indices.iter().any(|idx| idx.uuid == fri.uuid),
2338 "entry carried"
2339 );
2340 assert!(
2341 user_bitmap(&indices).is_empty(),
2342 "{:?}",
2343 user_bitmap(&indices)
2344 );
2345 }
2346
2347 #[rstest::rstest]
2352 #[case::update("update")]
2353 #[case::merge("merge")]
2354 #[case::data_replacement("data_replacement")]
2355 fn tagged_history_withdraws_mixed_provenance_whole(#[case] kind: &str) {
2356 let manifest = manifest_with_file("a.lance");
2357 let (mut segment, fri) = translating_segment_and_entry();
2358 segment.fragment_bitmap = Some([0u32, 5].into_iter().collect());
2359 let transaction = Transaction::new(
2360 manifest.version,
2361 in_place_rewrite(kind, &manifest, vec![0]),
2362 None,
2363 );
2364 let (_, indices) = transaction
2365 .build_manifest(
2366 Some(&manifest),
2367 vec![segment, fri],
2368 "txn",
2369 &default_build_config(),
2370 )
2371 .unwrap();
2372 assert!(
2373 user_bitmap(&indices).is_empty(),
2374 "{:?}",
2375 user_bitmap(&indices)
2376 );
2377 }
2378
2379 #[test]
2386 fn tagged_history_without_a_walkable_lineage_withdraws_every_retired_source() {
2387 use crate::system_index::frag_reuse::lineage::TaggedLineage;
2388 let entry = tagged_entry(&[(&[5], &[0]), (&[6], &[8])]);
2389 let mut segment = sample_index_metadata("id_idx");
2390 segment.fragment_bitmap = Some([0u32, 5, 6, 7].into_iter().collect());
2391 let live: RoaringBitmap = [0u32, 7, 8].into_iter().collect();
2392
2393 let unwalkable = TaggedLineage::new(&entry, None);
2394 assert!(!unwalkable.has_ledger());
2395 let mut indices = vec![segment.clone()];
2396 Transaction::withdraw_rewritten_coverage(
2397 &mut indices,
2398 Some(&unwalkable),
2399 &live,
2400 &[(0, vec![0])],
2401 );
2402 assert_eq!(
2403 user_bitmap(&indices),
2404 [7u32].into_iter().collect::<RoaringBitmap>(),
2405 "the rewritten fragment and every retired fragment go"
2406 );
2407
2408 let walkable = TaggedLineage::from_indices_with_ledger(&[entry], None)
2409 .unwrap()
2410 .unwrap();
2411 let mut indices = vec![segment];
2412 Transaction::withdraw_rewritten_coverage(
2413 &mut indices,
2414 Some(&walkable),
2415 &live,
2416 &[(0, vec![0])],
2417 );
2418 assert_eq!(
2419 user_bitmap(&indices),
2420 [6u32, 7].into_iter().collect::<RoaringBitmap>(),
2421 "only the transition reaching fragment 0 is withdrawn"
2422 );
2423 }
2424
2425 #[test]
2431 fn tagged_history_retires_an_emptied_segment_only_beside_a_serving_sibling() {
2432 let mut manifest = manifest_with_file("a.lance");
2433 manifest.fragments = Arc::new(vec![
2434 Fragment::new(0).with_file(
2435 "a.lance",
2436 vec![0],
2437 vec![0],
2438 ConcreteFileVersion::V2_0,
2439 None,
2440 ),
2441 Fragment::new(1).with_file(
2442 "c.lance",
2443 vec![0],
2444 vec![0],
2445 ConcreteFileVersion::V2_0,
2446 None,
2447 ),
2448 ]);
2449 let (translating, fri) = translating_segment_and_entry();
2450 let mut sibling = sample_index_metadata("id_idx");
2451 sibling.uuid = Uuid::new_v4();
2452 sibling.fragment_bitmap = Some([1u32].into_iter().collect());
2453 let mut unknown = sample_index_metadata("other_idx");
2454 unknown.uuid = Uuid::new_v4();
2455 unknown.fragment_bitmap = None;
2456 let transaction = Transaction::new(
2457 manifest.version,
2458 in_place_rewrite("update", &manifest, vec![0]),
2459 None,
2460 );
2461
2462 let (_, indices) = transaction
2464 .build_manifest(
2465 Some(&manifest),
2466 vec![
2467 translating.clone(),
2468 sibling.clone(),
2469 unknown.clone(),
2470 fri.clone(),
2471 ],
2472 "txn",
2473 &default_build_config(),
2474 )
2475 .unwrap();
2476 let names: Vec<(String, Uuid)> = indices
2477 .iter()
2478 .map(|idx| (idx.name.clone(), idx.uuid))
2479 .collect();
2480 assert!(
2481 !names.contains(&("id_idx".to_string(), translating.uuid)),
2482 "{names:?}"
2483 );
2484 assert_eq!(
2485 indices
2486 .iter()
2487 .find(|idx| idx.uuid == sibling.uuid)
2488 .unwrap()
2489 .fragment_bitmap,
2490 sibling.fragment_bitmap,
2491 "the sibling is untouched"
2492 );
2493 assert_eq!(
2494 indices
2495 .iter()
2496 .find(|idx| idx.uuid == unknown.uuid)
2497 .unwrap()
2498 .fragment_bitmap,
2499 None,
2500 "unknown coverage is neither withdrawn nor emptied"
2501 );
2502
2503 let (_, indices) = transaction
2505 .build_manifest(
2506 Some(&manifest),
2507 vec![translating.clone(), fri],
2508 "txn",
2509 &default_build_config(),
2510 )
2511 .unwrap();
2512 let kept = indices
2513 .iter()
2514 .find(|idx| idx.uuid == translating.uuid)
2515 .expect("the last segment stays");
2516 assert!(kept.fragment_bitmap.as_ref().unwrap().is_empty());
2517 }
2518
2519 #[test]
2523 fn tagged_history_leaves_unrelated_lineage_alone() {
2524 let manifest = manifest_with_file("a.lance");
2525 let fri = tagged_entry(&[(&[5], &[0]), (&[6], &[8])]);
2526 let mut segment = sample_index_metadata("id_idx");
2527 segment.fragment_bitmap = Some([6u32].into_iter().collect());
2528 let transaction = Transaction::new(
2529 manifest.version,
2530 in_place_rewrite("update", &manifest, vec![0]),
2531 None,
2532 );
2533 let (_, indices) = transaction
2534 .build_manifest(
2535 Some(&manifest),
2536 vec![segment, fri],
2537 "txn",
2538 &default_build_config(),
2539 )
2540 .unwrap();
2541 assert_eq!(user_bitmap(&indices), RoaringBitmap::from_iter([6u32]));
2542 }
2543
2544 #[test]
2547 fn tagged_history_without_readable_history_is_an_error() {
2548 let manifest = manifest_with_file("a.lance");
2549 let (segment, mut fri) = translating_segment_and_entry();
2550 fri.index_details = None;
2551 let transaction = Transaction::new(
2552 manifest.version,
2553 in_place_rewrite("update", &manifest, vec![0]),
2554 None,
2555 );
2556 let error = transaction
2557 .build_manifest(
2558 Some(&manifest),
2559 vec![segment, fri],
2560 "txn",
2561 &default_build_config(),
2562 )
2563 .unwrap_err();
2564 assert!(matches!(error, Error::InvalidInput { .. }), "{error}");
2565 assert!(error.to_string().contains("history"), "{error}");
2566 }
2567
2568 #[rstest::rstest]
2573 #[case::update_direct("update", "direct")]
2574 #[case::merge_direct("merge", "direct")]
2575 #[case::data_replacement_direct("data_replacement", "direct")]
2576 #[case::update_unindexed_field("update", "unindexed_field")]
2577 #[case::merge_unindexed_field("merge", "unindexed_field")]
2578 #[case::update_off_lineage("update", "off_lineage")]
2579 fn tagged_history_prunes_or_keeps_what_the_history_says(
2580 #[case] kind: &str,
2581 #[case] shape: &str,
2582 ) {
2583 let manifest = manifest_with_file("a.lance");
2584 let (mut segment, mut fri) = translating_segment_and_entry();
2585 let mut fields = vec![0];
2586 match shape {
2587 "direct" => segment.fragment_bitmap = Some([0u32].into_iter().collect()),
2588 "unindexed_field" => fields = vec![1],
2589 "off_lineage" => fri = tagged_entry(&[(&[5], &[7])]),
2591 _ => unreachable!(),
2592 }
2593 let transaction = Transaction::new(
2594 manifest.version,
2595 in_place_rewrite(kind, &manifest, fields),
2596 None,
2597 );
2598 let (_, indices) = transaction
2599 .build_manifest(
2600 Some(&manifest),
2601 vec![segment.clone(), fri.clone()],
2602 "txn",
2603 &default_build_config(),
2604 )
2605 .unwrap();
2606 let carried = indices
2607 .iter()
2608 .find(|idx| idx.name == crate::system_index::frag_reuse::FRAG_REUSE_INDEX_NAME)
2609 .expect("the tagged entry is carried through");
2610 assert_eq!(carried.uuid, fri.uuid);
2611 let user = indices.iter().find(|idx| idx.name == "id_idx").unwrap();
2612 let expected: RoaringBitmap = if shape == "direct" {
2613 RoaringBitmap::new()
2615 } else {
2616 segment.fragment_bitmap.unwrap()
2617 };
2618 assert_eq!(user.fragment_bitmap.as_ref().unwrap(), &expected, "{shape}");
2619 }
2620
2621 #[test]
2624 fn tagged_history_allows_user_index_create_index() {
2625 let (manifest, fri) = tagged_sample_manifest();
2626 let transaction = Transaction::new(
2627 manifest.version,
2628 Operation::CreateIndex {
2629 new_indices: vec![sample_index_metadata("id_idx")],
2630 removed_indices: vec![],
2631 },
2632 None,
2633 );
2634 let (_, indices) = transaction
2635 .build_manifest(
2636 Some(&manifest),
2637 vec![fri.clone()],
2638 "txn",
2639 &default_build_config(),
2640 )
2641 .unwrap();
2642 assert!(indices.iter().any(|idx| idx.name == "id_idx"));
2643 let carried = indices.iter().find(|idx| idx.uuid == fri.uuid).unwrap();
2644 assert_eq!(carried.index_details, fri.index_details);
2645 assert_eq!(carried.fragment_bitmap, fri.fragment_bitmap);
2646 }
2647
2648 #[test]
2652 fn tagged_history_allows_mem_wal_state_update() {
2653 let (manifest, fri) = tagged_sample_manifest();
2654 let user = sample_index_metadata("id_idx");
2655 let mem_wal = new_mem_wal_index_meta(manifest.version, Default::default()).unwrap();
2656 let shard = Uuid::new_v4();
2657 let transaction = Transaction::new(
2658 manifest.version,
2659 Operation::UpdateMemWalState {
2660 compacted_sstables: vec![CompactedSsTable::new(shard, 3)],
2661 },
2662 None,
2663 );
2664 let (built, indices) = transaction
2665 .build_manifest(
2666 Some(&manifest),
2667 vec![fri.clone(), user.clone(), mem_wal.clone()],
2668 "txn",
2669 &default_build_config(),
2670 )
2671 .unwrap();
2672 assert_eq!(built.fragments.len(), manifest.fragments.len());
2673 assert_eq!(indices.len(), 3);
2674 let carried_fri = indices.iter().find(|idx| idx.uuid == fri.uuid).unwrap();
2675 assert_eq!(carried_fri.index_details, fri.index_details);
2676 assert_eq!(carried_fri.fragment_bitmap, fri.fragment_bitmap);
2677 assert_eq!(carried_fri.index_version, fri.index_version);
2678 let carried_user = indices.iter().find(|idx| idx.uuid == user.uuid).unwrap();
2679 assert_eq!(carried_user.fragment_bitmap, user.fragment_bitmap);
2680 assert_eq!(carried_user.dataset_version, user.dataset_version);
2681 let updated = indices
2682 .iter()
2683 .find(|idx| idx.name == MEM_WAL_INDEX_NAME)
2684 .unwrap();
2685 assert_ne!(updated.uuid, mem_wal.uuid);
2686 assert_eq!(updated.dataset_version, built.version);
2687 let details = load_mem_wal_index_details(updated.clone()).unwrap();
2688 assert_eq!(
2689 details.compacted_sstables,
2690 vec![CompactedSsTable::new(shard, 3)]
2691 );
2692 }
2693
2694 #[rstest::rstest]
2695 #[case::bare_rewrite("bare_rewrite")]
2696 #[case::rewrite_with_v0_entry("rewrite_with_v0_entry")]
2697 fn tagged_history_rejects_unsupported_transactions(#[case] kind: &str) {
2698 let (manifest, fri) = tagged_sample_manifest();
2699 let operation = match kind {
2700 "bare_rewrite" => Operation::Rewrite {
2706 groups: vec![RewriteGroup {
2707 old_fragments: vec![Fragment::new(0)],
2708 new_fragments: vec![Fragment::new(10)],
2709 }],
2710 rewritten_indices: vec![],
2711 frag_reuse_index: None,
2712 },
2713 "rewrite_with_v0_entry" => {
2714 let mut v0_entry = fri.clone();
2715 v0_entry.index_version = 0;
2716 Operation::Rewrite {
2717 groups: vec![],
2718 rewritten_indices: vec![],
2719 frag_reuse_index: Some(v0_entry),
2720 }
2721 }
2722 _ => unreachable!(),
2723 };
2724 let transaction = Transaction::new(manifest.version, operation, None);
2725 let error = transaction
2726 .build_manifest(Some(&manifest), vec![fri], "txn", &default_build_config())
2727 .unwrap_err();
2728 if kind == "bare_rewrite" {
2729 assert!(matches!(error, Error::InvalidInput { .. }), "{error}");
2730 assert!(
2731 error
2732 .to_string()
2733 .contains("carries no fragment reuse transition"),
2734 "{error}"
2735 );
2736 } else {
2737 assert!(matches!(error, Error::NotSupported { .. }), "{error}");
2738 assert!(
2739 error.to_string().contains("Tagged FRI history maintenance"),
2740 "{error}"
2741 );
2742 }
2743 }
2744
2745 #[rstest::rstest]
2746 #[case::untagged_table(false)]
2747 #[case::tagged_table(true)]
2748 fn tagged_snapshot_replace_rejected(#[case] table_is_tagged: bool) {
2749 let mut manifest = sample_manifest();
2754 let mut current_indices = vec![];
2755 if table_is_tagged {
2756 manifest.reader_feature_flags |= FLAG_FRAGMENT_REUSE_INDEX;
2757 manifest.writer_feature_flags |= FLAG_FRAGMENT_REUSE_INDEX;
2758 let mut current =
2759 sample_index_metadata(crate::system_index::frag_reuse::FRAG_REUSE_INDEX_NAME);
2760 current.fields.clear();
2761 current_indices.push(current);
2762 }
2763 let mut entry =
2764 sample_index_metadata(crate::system_index::frag_reuse::FRAG_REUSE_INDEX_NAME);
2765 entry.fields.clear();
2766 let transaction = Transaction::new(
2767 manifest.version,
2768 Operation::Rewrite {
2769 groups: vec![],
2770 rewritten_indices: vec![],
2771 frag_reuse_index: Some(entry),
2772 },
2773 None,
2774 );
2775 let error = transaction
2776 .build_manifest(
2777 Some(&manifest),
2778 current_indices,
2779 "txn",
2780 &default_build_config(),
2781 )
2782 .unwrap_err();
2783 assert!(matches!(error, Error::InvalidInput { .. }), "{error}");
2784 assert!(error.to_string().contains("must be assembled"), "{error}");
2785 }
2786
2787 #[test]
2788 fn v0_snapshot_replace_rejected_on_flag_carrying_manifest_without_entry() {
2789 let mut manifest = sample_manifest();
2794 manifest.reader_feature_flags |= FLAG_FRAGMENT_REUSE_INDEX;
2795 manifest.writer_feature_flags |= FLAG_FRAGMENT_REUSE_INDEX;
2796 let mut entry =
2797 sample_index_metadata(crate::system_index::frag_reuse::FRAG_REUSE_INDEX_NAME);
2798 entry.fields.clear();
2799 entry.index_version = 0;
2800 let transaction = Transaction::new(
2801 manifest.version,
2802 Operation::Rewrite {
2803 groups: vec![],
2804 rewritten_indices: vec![],
2805 frag_reuse_index: Some(entry),
2806 },
2807 None,
2808 );
2809 let error = transaction
2810 .build_manifest(Some(&manifest), vec![], "txn", &default_build_config())
2811 .unwrap_err();
2812 assert!(matches!(error, Error::InvalidInput { .. }), "{error}");
2813 assert!(error.to_string().contains("sticky feature flag"), "{error}");
2814 }
2815
2816 #[test]
2817 fn tagged_history_allows_rewrite_of_uncovered_fragments() {
2818 let mut manifest = sample_manifest_with_fragments(0..6);
2822 manifest.reader_feature_flags |= FLAG_FRAGMENT_REUSE_INDEX;
2823 manifest.writer_feature_flags |= FLAG_FRAGMENT_REUSE_INDEX;
2824 let mut fri = sample_index_metadata(crate::system_index::frag_reuse::FRAG_REUSE_INDEX_NAME);
2825 fri.fields.clear();
2826 assert_eq!(
2827 fri.fragment_bitmap.as_ref().unwrap(),
2828 &roaring::RoaringBitmap::from_iter([0u32])
2829 );
2830 let transaction = Transaction::new(
2831 manifest.version,
2832 Operation::Rewrite {
2833 groups: vec![RewriteGroup {
2834 old_fragments: vec![Fragment::new(5)],
2835 new_fragments: vec![Fragment::new(10)],
2836 }],
2837 rewritten_indices: vec![],
2838 frag_reuse_index: None,
2839 },
2840 None,
2841 );
2842 let (_, final_indices) = transaction
2843 .build_manifest(
2844 Some(&manifest),
2845 vec![fri.clone()],
2846 "txn",
2847 &default_build_config(),
2848 )
2849 .unwrap();
2850 assert!(final_indices.iter().any(|idx| idx.uuid == fri.uuid));
2852 }
2853
2854 #[test]
2855 fn out_of_range_transition_source_id_rejected() {
2856 let error = crate::transaction::reordered_sources(&[
2860 crate::format::pb::fragment_reuse_index_details::Transition {
2861 sources: vec![
2862 crate::format::pb::fragment_reuse_index_details::FragmentDigest {
2863 id: u64::from(u32::MAX) + 1,
2864 physical_rows: 4,
2865 num_deleted_rows: 0,
2866 },
2867 ],
2868 destinations: vec![],
2869 mapping: None,
2870 },
2871 ])
2872 .unwrap_err();
2873 assert!(error.to_string().contains("row-address range"), "{error}");
2874 }
2875
2876 #[test]
2877 fn tagged_history_allows_rewrite_appending_tagged_entry() {
2878 let mut manifest = sample_manifest();
2879 manifest.reader_feature_flags |= FLAG_FRAGMENT_REUSE_INDEX;
2880 manifest.writer_feature_flags |= FLAG_FRAGMENT_REUSE_INDEX;
2881 let mut current =
2882 sample_index_metadata(crate::system_index::frag_reuse::FRAG_REUSE_INDEX_NAME);
2883 current.fields.clear();
2884 current.dataset_version = 7;
2885 let mut appended =
2886 sample_index_metadata(crate::system_index::frag_reuse::FRAG_REUSE_INDEX_NAME);
2887 appended.fields.clear();
2888
2889 let transaction = Transaction::new(
2890 manifest.version,
2891 Operation::Rewrite {
2892 groups: vec![],
2893 rewritten_indices: vec![],
2894 frag_reuse_index: Some(appended.clone()),
2895 },
2896 None,
2897 );
2898 let config = default_build_config();
2899 let prepared = transaction
2900 .prepare_indices(
2901 Some(&manifest),
2902 vec![current.clone()],
2903 &config,
2904 None,
2905 FragReuseUpdate::Rewrite(Arc::new(crate::transaction::TaggedRewriteAssembly {
2906 entry: appended.clone(),
2907 base_entry_version: Some(7),
2908 reordered_sources: RoaringBitmap::new(),
2909 })),
2910 )
2911 .unwrap();
2912 let (new_manifest, final_indices) = transaction
2913 .build_manifest_prepared(Some(&manifest), prepared, "txn", &config, None)
2914 .unwrap();
2915 let entries: Vec<_> = final_indices
2917 .iter()
2918 .filter(|idx| idx.name == crate::system_index::frag_reuse::FRAG_REUSE_INDEX_NAME)
2919 .collect();
2920 assert_eq!(entries.len(), 1);
2921 assert_eq!(entries[0].uuid, appended.uuid);
2922 assert_ne!(
2923 new_manifest.reader_feature_flags & FLAG_FRAGMENT_REUSE_INDEX,
2924 0
2925 );
2926 }
2927
2928 fn tagged_table_fixture() -> (Manifest, IndexMetadata) {
2931 let mut manifest = sample_manifest();
2932 manifest.reader_feature_flags |= FLAG_FRAGMENT_REUSE_INDEX;
2933 manifest.writer_feature_flags |= FLAG_FRAGMENT_REUSE_INDEX;
2934 let mut current =
2935 sample_index_metadata(crate::system_index::frag_reuse::FRAG_REUSE_INDEX_NAME);
2936 current.fields.clear();
2937 (manifest, current)
2938 }
2939
2940 fn build_trimmed(
2943 transaction: &Transaction,
2944 manifest: &Manifest,
2945 indices: Vec<IndexMetadata>,
2946 ) -> Result<(Manifest, Vec<IndexMetadata>)> {
2947 let config = default_build_config();
2948 let prepared = transaction.prepare_indices(
2949 Some(manifest),
2950 indices,
2951 &config,
2952 None,
2953 FragReuseUpdate::Trim,
2954 )?;
2955 transaction.build_manifest_prepared(Some(manifest), prepared, "txn", &config, None)
2956 }
2957
2958 #[test]
2961 fn tagged_trim_without_intent_rejected() {
2962 let (manifest, current) = tagged_table_fixture();
2963 let mut replacement =
2964 sample_index_metadata(crate::system_index::frag_reuse::FRAG_REUSE_INDEX_NAME);
2965 replacement.fields.clear();
2966 let transaction = Transaction::new(
2967 manifest.version,
2968 Operation::CreateIndex {
2969 new_indices: vec![replacement],
2970 removed_indices: vec![current.clone()],
2971 },
2972 None,
2973 );
2974 let error = transaction
2975 .build_manifest(
2976 Some(&manifest),
2977 vec![current],
2978 "txn",
2979 &default_build_config(),
2980 )
2981 .unwrap_err();
2982 assert!(matches!(error, Error::InvalidInput { .. }), "{error}");
2983 assert!(
2984 error.to_string().contains("cleanup_frag_reuse_index"),
2985 "{error}"
2986 );
2987 }
2988
2989 #[rstest::rstest]
2992 #[case::replace(true)]
2993 #[case::remove(false)]
2994 fn tagged_trim_with_intent_accepts_the_safe_shapes(#[case] replace: bool) {
2995 let (manifest, current) = tagged_table_fixture();
2996 let new_indices = if replace {
2997 let mut replacement =
2998 sample_index_metadata(crate::system_index::frag_reuse::FRAG_REUSE_INDEX_NAME);
2999 replacement.fields.clear();
3000 vec![replacement]
3001 } else {
3002 vec![]
3003 };
3004 let expected_uuid = new_indices.first().map(|idx| idx.uuid);
3005 let transaction = Transaction::new(
3006 manifest.version,
3007 Operation::CreateIndex {
3008 new_indices,
3009 removed_indices: vec![current.clone()],
3010 },
3011 None,
3012 );
3013 let (new_manifest, final_indices) =
3014 build_trimmed(&transaction, &manifest, vec![current]).unwrap();
3015 let entries: Vec<_> = final_indices
3016 .iter()
3017 .filter(|idx| idx.name == crate::system_index::frag_reuse::FRAG_REUSE_INDEX_NAME)
3018 .collect();
3019 assert_eq!(entries.first().map(|idx| idx.uuid), expected_uuid);
3020 assert_ne!(
3022 new_manifest.reader_feature_flags & FLAG_FRAGMENT_REUSE_INDEX,
3023 0
3024 );
3025 }
3026
3027 #[test]
3031 fn tagged_trim_with_stale_identity_rejected() {
3032 let (manifest, current) = tagged_table_fixture();
3033 let mut stale = current.clone();
3034 stale.uuid = Uuid::new_v4();
3035 let mut replacement =
3036 sample_index_metadata(crate::system_index::frag_reuse::FRAG_REUSE_INDEX_NAME);
3037 replacement.fields.clear();
3038 let transaction = Transaction::new(
3039 manifest.version,
3040 Operation::CreateIndex {
3041 new_indices: vec![replacement],
3042 removed_indices: vec![stale],
3043 },
3044 None,
3045 );
3046 let error = build_trimmed(&transaction, &manifest, vec![current]).unwrap_err();
3047 assert!(matches!(error, Error::InvalidInput { .. }), "{error}");
3048 assert!(error.to_string().contains("stale"), "{error}");
3049 }
3050
3051 #[test]
3054 fn tagged_trim_mixed_with_user_indices_rejected() {
3055 let (manifest, current) = tagged_table_fixture();
3056 let mut replacement =
3057 sample_index_metadata(crate::system_index::frag_reuse::FRAG_REUSE_INDEX_NAME);
3058 replacement.fields.clear();
3059 let transaction = Transaction::new(
3060 manifest.version,
3061 Operation::CreateIndex {
3062 new_indices: vec![replacement, sample_index_metadata("id_idx")],
3063 removed_indices: vec![current.clone()],
3064 },
3065 None,
3066 );
3067 let error = build_trimmed(&transaction, &manifest, vec![current]).unwrap_err();
3068 assert!(matches!(error, Error::InvalidInput { .. }), "{error}");
3069 }
3070
3071 #[test]
3072 fn frag_reuse_rewrite_guards_base_entry_version() {
3073 let mut manifest = sample_manifest();
3074 manifest.reader_feature_flags |= FLAG_FRAGMENT_REUSE_INDEX;
3075 manifest.writer_feature_flags |= FLAG_FRAGMENT_REUSE_INDEX;
3076 let mut current =
3077 sample_index_metadata(crate::system_index::frag_reuse::FRAG_REUSE_INDEX_NAME);
3078 current.fields.clear();
3079 current.dataset_version = 7;
3080 let mut appended =
3081 sample_index_metadata(crate::system_index::frag_reuse::FRAG_REUSE_INDEX_NAME);
3082 appended.fields.clear();
3083
3084 let transaction = Transaction::new(
3087 manifest.version,
3088 Operation::Rewrite {
3089 groups: vec![],
3090 rewritten_indices: vec![],
3091 frag_reuse_index: Some(appended.clone()),
3092 },
3093 None,
3094 );
3095 let config = default_build_config();
3096 let prepared = transaction
3097 .prepare_indices(
3098 Some(&manifest),
3099 vec![current],
3100 &config,
3101 None,
3102 FragReuseUpdate::Rewrite(Arc::new(crate::transaction::TaggedRewriteAssembly {
3103 entry: appended.clone(),
3104 base_entry_version: None,
3105 reordered_sources: RoaringBitmap::new(),
3106 })),
3107 )
3108 .unwrap();
3109 let error = transaction
3110 .build_manifest_prepared(Some(&manifest), prepared, "txn", &config, None)
3111 .unwrap_err();
3112 assert!(matches!(error, Error::InvalidInput { .. }), "{error}");
3113 assert!(error.to_string().contains("rebuild"), "{error}");
3114 }
3115
3116 #[rstest::rstest]
3119 #[case::update("update")]
3120 #[case::merge("merge")]
3121 #[case::data_replacement("data_replacement")]
3122 fn prepare_indices_keeps_original_and_withdraws_prepared(#[case] kind: &str) {
3123 let manifest = manifest_with_file("a.lance");
3124 let (segment, fri) = translating_segment_and_entry();
3125 let fri = {
3127 let mut merged = fri;
3128 merged.index_details = tagged_entry(&[(&[5], &[0]), (&[6], &[8])]).index_details;
3129 merged.fragment_bitmap = Some([0u32, 5, 6, 8].into_iter().collect());
3130 merged
3131 };
3132 let mut unrelated = sample_index_metadata("other_idx");
3133 unrelated.fragment_bitmap = Some([6u32].into_iter().collect());
3134 let transaction = Transaction::new(
3135 manifest.version,
3136 in_place_rewrite(kind, &manifest, vec![0]),
3137 None,
3138 );
3139 let config = default_build_config();
3140 let prepared = transaction
3141 .prepare_indices(
3142 Some(&manifest),
3143 vec![segment.clone(), unrelated.clone(), fri.clone()],
3144 &config,
3145 None,
3146 FragReuseUpdate::None,
3147 )
3148 .unwrap();
3149 assert_eq!(prepared.manifest_version(), Some(manifest.version));
3150 assert_eq!(
3151 prepared.original(),
3152 &[segment.clone(), unrelated.clone(), fri],
3153 "the original list is what was read"
3154 );
3155 let prepared_user = prepared
3156 .prepared()
3157 .iter()
3158 .find(|idx| idx.uuid == segment.uuid)
3159 .unwrap();
3160 assert!(prepared_user.fragment_bitmap.as_ref().unwrap().is_empty());
3161 let prepared_unrelated = prepared
3162 .prepared()
3163 .iter()
3164 .find(|idx| idx.uuid == unrelated.uuid)
3165 .unwrap();
3166 assert_eq!(
3167 prepared_unrelated.fragment_bitmap,
3168 unrelated.fragment_bitmap
3169 );
3170 let (_, indices) = transaction
3171 .build_manifest_prepared(Some(&manifest), prepared.clone(), "txn", &config, None)
3172 .unwrap();
3173 assert_eq!(indices, prepared.prepared());
3174 }
3175
3176 #[rstest::rstest]
3178 #[case::update("update")]
3179 #[case::merge("merge")]
3180 #[case::data_replacement("data_replacement")]
3181 fn prepare_indices_prunes_by_id_on_an_untagged_table(#[case] kind: &str) {
3182 let mut manifest = manifest_with_file("a.lance");
3183 manifest.reader_feature_flags &= !FLAG_FRAGMENT_REUSE_INDEX;
3184 manifest.writer_feature_flags &= !FLAG_FRAGMENT_REUSE_INDEX;
3185 let mut segment = sample_index_metadata("id_idx");
3186 segment.fragment_bitmap = Some([0u32, 3].into_iter().collect());
3187 let transaction = Transaction::new(
3188 manifest.version,
3189 in_place_rewrite(kind, &manifest, vec![0]),
3190 None,
3191 );
3192 let prepared = transaction
3193 .prepare_indices(
3194 Some(&manifest),
3195 vec![segment.clone()],
3196 &default_build_config(),
3197 None,
3198 FragReuseUpdate::None,
3199 )
3200 .unwrap();
3201 assert_eq!(prepared.original(), &[segment]);
3202 assert_eq!(
3203 prepared.prepared()[0].fragment_bitmap,
3204 Some(RoaringBitmap::from_iter([3u32]))
3205 );
3206 }
3207
3208 #[test]
3211 fn prepare_indices_restore_and_non_rewrites_are_identity() {
3212 let manifest = manifest_with_file("a.lance");
3213 let (segment, fri) = translating_segment_and_entry();
3214 for operation in [
3215 Operation::Restore { version: 1 },
3216 Operation::Append { fragments: vec![] },
3217 Operation::Delete {
3218 updated_fragments: vec![],
3219 deleted_fragment_ids: vec![0],
3220 predicate: "true".into(),
3221 },
3222 in_place_rewrite("update", &manifest, vec![]),
3223 ] {
3224 let transaction = Transaction::new(manifest.version, operation, None);
3225 let prepared = transaction
3226 .prepare_indices(
3227 Some(&manifest),
3228 vec![segment.clone(), fri.clone()],
3229 &default_build_config(),
3230 None,
3231 FragReuseUpdate::None,
3232 )
3233 .unwrap();
3234 assert_eq!(prepared.prepared(), prepared.original());
3235 }
3236 }
3237
3238 #[test]
3241 fn prepare_indices_external_history_needs_the_supplied_ledger() {
3242 let manifest = manifest_with_file("a.lance");
3243 let (segment, fri) = translating_segment_and_entry();
3244 let ledger = crate::system_index::frag_reuse::ledger::FragReuseLedger::decode_inline(
3245 fri.index_version,
3246 fri.index_details.as_ref().unwrap(),
3247 )
3248 .unwrap()
3249 .unwrap();
3250 let mut external = fri;
3251 external.index_details = None;
3252 let transaction = Transaction::new(
3253 manifest.version,
3254 in_place_rewrite("update", &manifest, vec![0]),
3255 None,
3256 );
3257 let error = transaction
3258 .prepare_indices(
3259 Some(&manifest),
3260 vec![segment.clone(), external.clone()],
3261 &default_build_config(),
3262 None,
3263 FragReuseUpdate::None,
3264 )
3265 .unwrap_err();
3266 assert!(matches!(error, Error::InvalidInput { .. }), "{error}");
3267 let prepared = transaction
3268 .prepare_indices(
3269 Some(&manifest),
3270 vec![segment, external],
3271 &default_build_config(),
3272 Some(Arc::new(ledger)),
3273 FragReuseUpdate::None,
3274 )
3275 .unwrap();
3276 assert!(user_bitmap(prepared.prepared()).is_empty());
3277 }
3278
3279 #[test]
3282 fn prepare_indices_corrupt_inline_details_is_an_error() {
3283 let manifest = manifest_with_file("a.lance");
3284 let (segment, mut fri) = translating_segment_and_entry();
3285 let mut details = fri.index_details.as_ref().unwrap().as_ref().clone();
3286 details.value = vec![0xff, 0xff, 0xff, 0xff, 0xff];
3287 fri.index_details = Some(Arc::new(details));
3288 let transaction = Transaction::new(
3289 manifest.version,
3290 in_place_rewrite("update", &manifest, vec![0]),
3291 None,
3292 );
3293 let error = transaction
3294 .prepare_indices(
3295 Some(&manifest),
3296 vec![segment.clone(), fri.clone()],
3297 &default_build_config(),
3298 None,
3299 FragReuseUpdate::None,
3300 )
3301 .unwrap_err();
3302 assert!(
3303 matches!(error, Error::CorruptFile { .. }),
3304 "corrupt history must surface as corruption: {error:?}"
3305 );
3306 let error = transaction
3307 .build_manifest(
3308 Some(&manifest),
3309 vec![segment, fri],
3310 "txn",
3311 &default_build_config(),
3312 )
3313 .unwrap_err();
3314 assert!(matches!(error, Error::CorruptFile { .. }), "{error:?}");
3315 }
3316
3317 #[test]
3320 fn build_manifest_prepared_refuses_a_result_prepared_against_another_manifest() {
3321 let manifest = manifest_with_file("a.lance");
3322 let (segment, fri) = translating_segment_and_entry();
3323 let transaction = Transaction::new(
3324 manifest.version,
3325 in_place_rewrite("update", &manifest, vec![0]),
3326 None,
3327 );
3328 let prepared = transaction
3329 .prepare_indices(
3330 Some(&manifest),
3331 vec![segment, fri],
3332 &default_build_config(),
3333 None,
3334 FragReuseUpdate::None,
3335 )
3336 .unwrap();
3337 let mut later = manifest;
3338 later.version += 1;
3339 let error = transaction
3340 .build_manifest_prepared(Some(&later), prepared, "txn", &default_build_config(), None)
3341 .unwrap_err();
3342 assert!(matches!(error, Error::InvalidInput { .. }), "{error}");
3343 assert!(error.to_string().contains("prepared against"), "{error}");
3344 }
3345
3346 #[test]
3350 fn build_manifest_prepared_still_runs_the_gate() {
3351 let manifest = manifest_with_file("a.lance");
3352 let (_, fri) = translating_segment_and_entry();
3353 let mut appended = fri.clone();
3354 appended.uuid = uuid::Uuid::new_v4();
3355 let transaction = Transaction::new(
3356 manifest.version,
3357 Operation::Rewrite {
3358 groups: vec![],
3359 rewritten_indices: vec![],
3360 frag_reuse_index: Some(appended),
3361 },
3362 None,
3363 );
3364 let error = transaction
3365 .prepare_indices(
3366 Some(&manifest),
3367 vec![fri.clone()],
3368 &default_build_config(),
3369 None,
3370 FragReuseUpdate::None,
3371 )
3372 .unwrap_err();
3373 assert!(
3374 error.to_string().contains("assembled by the commit path"),
3375 "{error}"
3376 );
3377 let error = transaction
3378 .build_manifest(Some(&manifest), vec![fri], "txn", &default_build_config())
3379 .unwrap_err();
3380 assert!(
3381 error.to_string().contains("assembled by the commit path"),
3382 "{error}"
3383 );
3384 }
3385
3386 fn sample_manifest_with_fragments(ids: std::ops::Range<u64>) -> Manifest {
3387 let schema = ArrowSchema::new(vec![ArrowField::new("id", DataType::Int32, false)]);
3388 Manifest::new(
3389 LanceSchema::try_from(&schema).unwrap(),
3390 Arc::new(ids.map(Fragment::new).collect()),
3391 DataStorageFormat::new(ConcreteFileVersion::V2_0),
3392 HashMap::new(),
3393 )
3394 }
3395
3396 #[rstest]
3397 #[case::project(false)]
3398 #[case::merge(true)]
3399 fn test_schema_change_prunes_irrelevant_overlays(#[case] is_merge: bool) {
3400 let arrow_schema = ArrowSchema::new(vec![
3401 ArrowField::new("a", DataType::Int32, false),
3402 ArrowField::new("b", DataType::Int32, false),
3403 ArrowField::new("c", DataType::Int32, false),
3404 ]);
3405 let schema = LanceSchema::try_from(&arrow_schema).unwrap();
3406 let mut projected_schema = schema.clone();
3407 projected_schema.fields.retain(|field| field.name != "c");
3408
3409 let overlay = |path: &str, fields: Vec<i32>| DataOverlayFile {
3410 data_file: DataFile::new_legacy_from_fields(path, fields, None),
3411 coverage: OverlayCoverage::dense(RoaringBitmap::from_iter([0u32])),
3412 committed_version: 1,
3413 };
3414 let mut fragment0 = Fragment::new(0);
3415 fragment0.overlays = vec![
3416 overlay("dropped-0.lance", vec![2]),
3417 overlay("kept-mixed.lance", vec![0, 2]),
3418 ];
3419 let mut fragment1 = Fragment::new(1);
3420 fragment1.overlays = vec![
3421 overlay("dropped-1.lance", vec![2]),
3422 overlay("kept-live.lance", vec![1]),
3423 ];
3424 let manifest = Manifest::new(
3425 schema,
3426 Arc::new(vec![fragment0, fragment1]),
3427 DataStorageFormat::new(ConcreteFileVersion::V2_0),
3428 HashMap::new(),
3429 );
3430
3431 let operation = if is_merge {
3432 Operation::Merge {
3433 fragments: manifest.fragments.as_ref().clone(),
3434 schema: projected_schema,
3435 preserves_nullability: true,
3436 }
3437 } else {
3438 Operation::Project {
3439 schema: projected_schema,
3440 preserves_nullability: true,
3441 }
3442 };
3443 let transaction = Transaction::new(manifest.version, operation, None);
3444 let (result, _) = transaction
3445 .build_manifest(Some(&manifest), vec![], "txn", &default_build_config())
3446 .unwrap();
3447
3448 let overlay_paths = result
3449 .fragments
3450 .iter()
3451 .flat_map(|fragment| &fragment.overlays)
3452 .map(|overlay| overlay.data_file.path.as_str())
3453 .collect::<Vec<_>>();
3454 assert_eq!(overlay_paths, ["kept-mixed.lance", "kept-live.lance"]);
3455 }
3456
3457 #[test]
3458 fn activation_sets_writer_gate_when_auto_flags_are_disabled() {
3459 let manifest = sample_manifest();
3460 let transaction = Transaction::new(
3461 manifest.version,
3462 Operation::UpdateConfig {
3463 config_updates: None,
3464 table_metadata_updates: None,
3465 schema_metadata_updates: None,
3466 field_metadata_updates: HashMap::new(),
3467 },
3468 None,
3469 );
3470 let mut config = default_build_config();
3471 config.auto_set_feature_flags = false;
3472 config.activate_non_reusable_field_ids = true;
3473
3474 let (activated, _) = transaction
3475 .build_manifest(Some(&manifest), vec![], "txn", &config)
3476 .unwrap();
3477
3478 assert!(activated.uses_non_reusable_field_ids());
3479 assert_eq!(
3480 activated.reader_feature_flags & FLAG_NON_REUSABLE_FIELD_IDS,
3481 0
3482 );
3483 assert_ne!(
3484 activated.writer_feature_flags & FLAG_NON_REUSABLE_FIELD_IDS,
3485 0
3486 );
3487 }
3488
3489 #[test]
3490 fn test_create_index_build_manifest_keeps_unremoved_same_name_indices() {
3491 let manifest = sample_manifest();
3492 let first_index = sample_index_metadata("vector_idx");
3493 let second_index = sample_index_metadata("vector_idx");
3494 let third_index = sample_index_metadata("vector_idx");
3495
3496 let transaction = Transaction::new(
3497 manifest.version,
3498 Operation::CreateIndex {
3499 new_indices: vec![third_index.clone()],
3500 removed_indices: vec![second_index.clone()],
3501 },
3502 None,
3503 );
3504
3505 let (_, final_indices) = transaction
3506 .build_manifest(
3507 Some(&manifest),
3508 vec![first_index.clone(), second_index.clone()],
3509 "txn",
3510 &default_build_config(),
3511 )
3512 .unwrap();
3513
3514 assert_eq!(final_indices.len(), 2);
3515 assert!(final_indices.iter().any(|idx| idx.uuid == first_index.uuid));
3516 assert!(final_indices.iter().any(|idx| idx.uuid == third_index.uuid));
3517 assert!(
3518 !final_indices
3519 .iter()
3520 .any(|idx| idx.uuid == second_index.uuid)
3521 );
3522 }
3523
3524 #[test]
3525 fn test_create_index_build_manifest_deduplicates_relisted_indices_by_uuid() {
3526 let manifest = sample_manifest();
3527 let first_index = sample_index_metadata("vector_idx");
3528 let second_index = sample_index_metadata("vector_idx");
3529 let third_index = sample_index_metadata("vector_idx");
3530
3531 let transaction = Transaction::new(
3532 manifest.version,
3533 Operation::CreateIndex {
3534 new_indices: vec![first_index.clone(), third_index.clone()],
3535 removed_indices: vec![second_index.clone()],
3536 },
3537 None,
3538 );
3539
3540 let (_, final_indices) = transaction
3541 .build_manifest(
3542 Some(&manifest),
3543 vec![first_index.clone(), second_index.clone()],
3544 "txn",
3545 &default_build_config(),
3546 )
3547 .unwrap();
3548
3549 assert_eq!(final_indices.len(), 2);
3550 assert_eq!(
3551 final_indices
3552 .iter()
3553 .filter(|idx| idx.uuid == first_index.uuid)
3554 .count(),
3555 1
3556 );
3557 assert!(final_indices.iter().any(|idx| idx.uuid == third_index.uuid));
3558 assert!(
3559 !final_indices
3560 .iter()
3561 .any(|idx| idx.uuid == second_index.uuid)
3562 );
3563 }
3564
3565 #[test]
3566 fn test_update_build_manifest_replaces_and_removes_fragments() {
3567 let manifest = sample_manifest_with_fragments(0..5);
3568
3569 let mut updated2 = Fragment::new(2);
3570 updated2.physical_rows = Some(42);
3571 let mut updated4 = Fragment::new(4);
3572 updated4.physical_rows = Some(43);
3573
3574 let transaction = Transaction::new(
3575 manifest.version,
3576 Operation::Update {
3577 removed_fragment_ids: vec![1],
3578 updated_fragments: vec![updated2, updated4, Fragment::new(99)],
3581 new_fragments: vec![],
3582 fields_modified: vec![],
3583 compacted_sstables: vec![],
3584 fields_for_preserving_frag_bitmap: vec![],
3585 update_mode: None,
3586 inserted_rows_filter: None,
3587 updated_fragment_offsets: None,
3588 },
3589 None,
3590 );
3591
3592 let (new_manifest, _) = transaction
3593 .build_manifest(Some(&manifest), vec![], "txn", &default_build_config())
3594 .unwrap();
3595
3596 let ids: Vec<u64> = new_manifest.fragments.iter().map(|f| f.id).collect();
3597 assert_eq!(ids, vec![0, 2, 3, 4]);
3598 let rows: Vec<Option<usize>> = new_manifest
3599 .fragments
3600 .iter()
3601 .map(|f| f.physical_rows)
3602 .collect();
3603 assert_eq!(rows, vec![None, Some(42), None, Some(43)]);
3604 }
3605
3606 #[rstest::rstest]
3607 #[case::unspecified(None)]
3608 #[case::rewrite_rows(Some(UpdateMode::RewriteRows))]
3609 #[case::rewrite_columns(Some(UpdateMode::RewriteColumns))]
3610 fn test_update_build_manifest_does_not_restore_projected_files(
3611 #[case] update_mode: Option<UpdateMode>,
3612 #[values(false, true)] with_spilled_lineage: bool,
3613 ) {
3614 let mut manifest = sample_manifest_with_fragments(0..3);
3615 if with_spilled_lineage {
3616 manifest.reader_feature_flags = FLAG_STABLE_ROW_IDS;
3617 manifest.writer_feature_flags = FLAG_STABLE_ROW_IDS;
3618 manifest.next_row_id = 126;
3619 for fragment in Arc::make_mut(&mut manifest.fragments) {
3620 let start = fragment.id * 42;
3621 fragment.physical_rows = Some(42);
3622 fragment.row_id_meta = Some(RowIdMeta::Inline(
3623 write_row_ids(&RowIdSequence::from(start..start + 42)).into(),
3624 ));
3625 }
3626 }
3627 let projected_file = DataFile::new(
3628 "projected.lance",
3629 vec![0],
3630 vec![0],
3631 ConcreteFileVersion::V2_0,
3632 None,
3633 None,
3634 );
3635 Arc::make_mut(&mut manifest.fragments)[1].files = vec![projected_file.clone()];
3636
3637 let mut updated = manifest.fragments[1].clone();
3639 updated.files.push(DataFile::new(
3640 "dropped.lance",
3641 vec![1],
3642 vec![0],
3643 ConcreteFileVersion::V2_0,
3644 None,
3645 None,
3646 ));
3647 updated.physical_rows = Some(42);
3648 let inserted_projected_file = DataFile::new(
3649 "inserted-projected.lance",
3650 vec![0],
3651 vec![0],
3652 ConcreteFileVersion::V2_0,
3653 None,
3654 None,
3655 );
3656 let mut inserted = Fragment::new(0);
3657 inserted.physical_rows = Some(42);
3658 inserted.files = vec![
3659 inserted_projected_file.clone(),
3660 DataFile::new(
3661 "inserted-dropped.lance",
3662 vec![1],
3663 vec![0],
3664 ConcreteFileVersion::V2_0,
3665 None,
3666 None,
3667 ),
3668 ];
3669
3670 let mut expected_updated_files = vec![projected_file];
3671 let mut expected_inserted_files = vec![inserted_projected_file];
3672 if with_spilled_lineage {
3673 for (fragment, expected_files) in [
3676 (&mut updated, &mut expected_updated_files),
3677 (&mut inserted, &mut expected_inserted_files),
3678 ] {
3679 fragment.row_id_meta = Some(RowIdMeta::Column);
3680 fragment.created_at_version_meta = Some(RowDatasetVersionMeta::Column);
3681 fragment.last_updated_at_version_meta = Some(RowDatasetVersionMeta::Column);
3682 for field_id in [
3683 ROW_ID_FIELD_ID,
3684 ROW_CREATED_AT_VERSION_FIELD_ID,
3685 ROW_LAST_UPDATED_AT_VERSION_FIELD_ID,
3686 ] {
3687 let file = DataFile::new(
3688 format!("lineage-{field_id}.lance"),
3689 vec![field_id],
3690 vec![0],
3691 ConcreteFileVersion::V2_0,
3692 None,
3693 None,
3694 );
3695 fragment.files.push(file.clone());
3696 expected_files.push(file);
3697 }
3698 }
3699 }
3700
3701 let transaction = Transaction::new(
3702 manifest.version,
3703 Operation::Update {
3704 removed_fragment_ids: vec![],
3705 updated_fragments: vec![updated],
3706 new_fragments: vec![inserted],
3707 fields_modified: vec![1],
3708 compacted_sstables: vec![],
3709 fields_for_preserving_frag_bitmap: vec![],
3710 update_mode,
3711 inserted_rows_filter: None,
3712 updated_fragment_offsets: None,
3713 },
3714 None,
3715 );
3716
3717 let (new_manifest, _) = transaction
3718 .build_manifest(Some(&manifest), vec![], "txn", &default_build_config())
3719 .unwrap();
3720
3721 let fragment = &new_manifest.fragments[1];
3722 assert_eq!(fragment.files, expected_updated_files);
3723 assert_eq!(fragment.physical_rows, Some(42));
3724 assert_eq!(new_manifest.fragments[3].files, expected_inserted_files);
3725 assert_eq!(new_manifest.max_field_id(), 0);
3726 }
3727
3728 #[test]
3729 fn test_delete_build_manifest_applies_deletion_to_current_fragment() {
3730 let mut manifest = sample_manifest_with_fragments(0..5);
3731 manifest.version = 2;
3732 let projected_file = DataFile::new_legacy_from_fields("projected.lance", vec![0], None);
3733 let current_fragment = &mut Arc::make_mut(&mut manifest.fragments)[2];
3734 current_fragment.files = vec![projected_file.clone()];
3735 current_fragment.physical_rows = Some(42);
3736 current_fragment.overlays = vec![overlay_with_field(0, 2)];
3737 current_fragment.last_updated_at_version_meta = Some(
3738 RowDatasetVersionMeta::from_sequence(
3739 &RowDatasetVersionSequence::from_uniform_row_count(42, 2),
3740 )
3741 .unwrap(),
3742 );
3743
3744 let mut updated2 = Fragment::new(2);
3746 updated2.physical_rows = Some(42);
3747 updated2.files = vec![
3748 projected_file.clone(),
3749 DataFile::new_legacy_from_fields("dropped.lance", vec![1], None),
3750 ];
3751 let deletion_file = DeletionFile {
3752 read_version: 1,
3753 id: 10,
3754 file_type: DeletionFileType::Array,
3755 num_deleted_rows: Some(1),
3756 base_id: None,
3757 };
3758 updated2.deletion_file = Some(deletion_file.clone());
3759
3760 let transaction = Transaction::new(
3761 1,
3762 Operation::Delete {
3763 updated_fragments: vec![updated2],
3764 deleted_fragment_ids: vec![1, 3],
3765 predicate: "id > 0".to_string(),
3766 },
3767 None,
3768 );
3769
3770 let (new_manifest, _) = transaction
3771 .build_manifest(Some(&manifest), vec![], "txn", &default_build_config())
3772 .unwrap();
3773
3774 let ids: Vec<u64> = new_manifest.fragments.iter().map(|f| f.id).collect();
3775 assert_eq!(ids, vec![0, 2, 4]);
3776 let fragment2 = new_manifest
3777 .fragments
3778 .iter()
3779 .find(|fragment| fragment.id == 2)
3780 .unwrap();
3781 assert_eq!(fragment2.files, vec![projected_file]);
3782 assert_eq!(fragment2.deletion_file, Some(deletion_file));
3783 assert_eq!(fragment2.overlays.len(), 1);
3784 assert_eq!(fragment2.overlays[0].committed_version, 2);
3785 assert_eq!(fragment2.physical_rows, Some(42));
3786 assert_eq!(new_manifest.max_field_id(), 0);
3787 assert_eq!(last_updated_at_versions(&new_manifest, 2), vec![2; 42]);
3788 }
3789
3790 #[test]
3791 fn test_remove_tombstoned_data_files() {
3792 let mut fragment = Fragment::new(1);
3794
3795 fragment.files.push(DataFile {
3797 path: "normal.lance".to_string(),
3798 fields: Arc::from([1, 2, 3]),
3799 column_indices: Arc::from([]),
3800 file_major_version: 2,
3801 file_minor_version: 0,
3802 file_size_bytes: CachedFileSize::new(1000),
3803 base_id: None,
3804 });
3805
3806 fragment.files.push(DataFile {
3808 path: "all_tombstoned.lance".to_string(),
3809 fields: Arc::from([-2, -2, -2]),
3810 column_indices: Arc::from([]),
3811 file_major_version: 2,
3812 file_minor_version: 0,
3813 file_size_bytes: CachedFileSize::new(500),
3814 base_id: None,
3815 });
3816
3817 fragment.files.push(DataFile {
3819 path: "mixed.lance".to_string(),
3820 fields: Arc::from([4, -2, 5]),
3821 column_indices: Arc::from([]),
3822 file_major_version: 2,
3823 file_minor_version: 0,
3824 file_size_bytes: CachedFileSize::new(750),
3825 base_id: None,
3826 });
3827
3828 fragment.files.push(DataFile {
3830 path: "another_tombstoned.lance".to_string(),
3831 fields: Arc::from([-2_i32]),
3832 column_indices: Arc::from([]),
3833 file_major_version: 2,
3834 file_minor_version: 0,
3835 file_size_bytes: CachedFileSize::new(250),
3836 base_id: None,
3837 });
3838
3839 let mut fragments = vec![fragment];
3840
3841 Transaction::remove_tombstoned_data_files(&mut fragments);
3843
3844 assert_eq!(fragments[0].files.len(), 2);
3846 assert_eq!(fragments[0].files[0].path, "normal.lance");
3847 assert_eq!(fragments[0].files[1].path, "mixed.lance");
3848 }
3849
3850 #[test]
3854 fn test_partial_rewrite_skips_fragment_with_no_version_meta() {
3855 let row_ids = RowIdSequence::from([10u64, 11, 12, 13, 14].as_slice());
3856 let row_id_meta = Some(RowIdMeta::Inline(write_row_ids(&row_ids).into()));
3857
3858 let data_file = DataFile::new(
3859 "data.lance",
3860 vec![0],
3861 vec![0],
3862 LanceFileVersion::Stable.resolve(),
3863 None,
3864 None,
3865 );
3866
3867 let fragment = Fragment {
3868 id: 1,
3869 files: vec![data_file],
3870 overlays: vec![],
3871 deletion_file: None,
3872 row_id_meta,
3873 physical_rows: Some(5),
3874 last_updated_at_version_meta: None,
3875 created_at_version_meta: None,
3876 };
3877
3878 let manifest = make_stable_row_id_manifest(vec![fragment.clone()]);
3879
3880 let off_map = HashMap::from([(1u64, RoaringBitmap::from_iter([1u32, 3]))]);
3882 let tx = Transaction::new(
3883 manifest.version,
3884 Operation::Update {
3885 removed_fragment_ids: vec![],
3886 updated_fragments: vec![fragment],
3887 new_fragments: vec![],
3888 fields_modified: vec![],
3889 compacted_sstables: vec![],
3890 fields_for_preserving_frag_bitmap: vec![],
3891 update_mode: Some(UpdateMode::RewriteColumns),
3892 inserted_rows_filter: None,
3893 updated_fragment_offsets: Some(UpdatedFragmentOffsets(off_map)),
3894 },
3895 None,
3896 );
3897
3898 let (out, _) = tx
3899 .build_manifest(Some(&manifest), vec![], "txn", &default_build_config())
3900 .unwrap();
3901
3902 assert!(
3903 out.fragments[0].last_updated_at_version_meta.is_none(),
3904 "fragment with no prior version metadata must not have fabricated prev_version stamped on unmatched rows"
3905 );
3906 }
3907
3908 #[test]
3909 fn test_bitmap_cardinality_exceeds_physical_rows() {
3910 let row_ids = RowIdSequence::from([10u64, 11, 12, 13, 14].as_slice());
3911 let row_id_meta = Some(RowIdMeta::Inline(write_row_ids(&row_ids).into()));
3912
3913 let data_file = DataFile::new(
3914 "data.lance",
3915 vec![0],
3916 vec![0],
3917 LanceFileVersion::Stable.resolve(),
3918 None,
3919 None,
3920 );
3921
3922 let version_seq = RowDatasetVersionSequence::from_uniform_row_count(5, 1);
3923 let version_meta = RowDatasetVersionMeta::from_sequence(&version_seq).unwrap();
3924
3925 let fragment = Fragment {
3926 id: 1,
3927 files: vec![data_file],
3928 overlays: vec![],
3929 deletion_file: None,
3930 row_id_meta,
3931 physical_rows: Some(5),
3932 last_updated_at_version_meta: Some(version_meta.clone()),
3933 created_at_version_meta: Some(version_meta),
3934 };
3935
3936 let manifest = make_stable_row_id_manifest(vec![fragment.clone()]);
3937
3938 let off_map = HashMap::from([(1u64, RoaringBitmap::from_iter(0u32..10))]);
3940 let tx = Transaction::new(
3941 manifest.version,
3942 Operation::Update {
3943 removed_fragment_ids: vec![],
3944 updated_fragments: vec![fragment],
3945 new_fragments: vec![],
3946 fields_modified: vec![],
3947 compacted_sstables: vec![],
3948 fields_for_preserving_frag_bitmap: vec![],
3949 update_mode: Some(UpdateMode::RewriteColumns),
3950 inserted_rows_filter: None,
3951 updated_fragment_offsets: Some(UpdatedFragmentOffsets(off_map)),
3952 },
3953 None,
3954 );
3955
3956 let result = tx.build_manifest(Some(&manifest), vec![], "txn", &default_build_config());
3957 assert!(result.is_err());
3958 let msg = result.unwrap_err().to_string();
3959 assert!(
3960 msg.contains("cardinality"),
3961 "expected cardinality error, got: {msg}"
3962 );
3963 }
3964
3965 #[test]
3966 fn test_bitmap_max_offset_exceeds_physical_rows() {
3967 let row_ids = RowIdSequence::from([10u64, 11, 12, 13, 14].as_slice());
3968 let row_id_meta = Some(RowIdMeta::Inline(write_row_ids(&row_ids).into()));
3969
3970 let data_file = DataFile::new(
3971 "data.lance",
3972 vec![0],
3973 vec![0],
3974 LanceFileVersion::Stable.resolve(),
3975 None,
3976 None,
3977 );
3978
3979 let version_seq = RowDatasetVersionSequence::from_uniform_row_count(5, 1);
3980 let version_meta = RowDatasetVersionMeta::from_sequence(&version_seq).unwrap();
3981
3982 let fragment = Fragment {
3983 id: 1,
3984 files: vec![data_file],
3985 overlays: vec![],
3986 deletion_file: None,
3987 row_id_meta,
3988 physical_rows: Some(5),
3989 last_updated_at_version_meta: Some(version_meta.clone()),
3990 created_at_version_meta: Some(version_meta),
3991 };
3992
3993 let manifest = make_stable_row_id_manifest(vec![fragment.clone()]);
3994
3995 let off_map = HashMap::from([(1u64, RoaringBitmap::from_iter([0u32, 100]))]);
3997 let tx = Transaction::new(
3998 manifest.version,
3999 Operation::Update {
4000 removed_fragment_ids: vec![],
4001 updated_fragments: vec![fragment],
4002 new_fragments: vec![],
4003 fields_modified: vec![],
4004 compacted_sstables: vec![],
4005 fields_for_preserving_frag_bitmap: vec![],
4006 update_mode: Some(UpdateMode::RewriteColumns),
4007 inserted_rows_filter: None,
4008 updated_fragment_offsets: Some(UpdatedFragmentOffsets(off_map)),
4009 },
4010 None,
4011 );
4012
4013 let result = tx.build_manifest(Some(&manifest), vec![], "txn", &default_build_config());
4014 assert!(result.is_err());
4015 let msg = result.unwrap_err().to_string();
4016 assert!(
4017 msg.contains("max offset"),
4018 "expected max offset error, got: {msg}"
4019 );
4020 }
4021
4022 #[test]
4023 fn test_bitmap_at_exact_physical_rows_boundary_succeeds() {
4024 let row_ids = RowIdSequence::from([10u64, 11, 12, 13, 14].as_slice());
4025 let row_id_meta = Some(RowIdMeta::Inline(write_row_ids(&row_ids).into()));
4026
4027 let data_file = DataFile::new(
4028 "data.lance",
4029 vec![0],
4030 vec![0],
4031 LanceFileVersion::Stable.resolve(),
4032 None,
4033 None,
4034 );
4035
4036 let version_seq = RowDatasetVersionSequence::from_uniform_row_count(5, 1);
4037 let version_meta = RowDatasetVersionMeta::from_sequence(&version_seq).unwrap();
4038
4039 let fragment = Fragment {
4040 id: 1,
4041 files: vec![data_file],
4042 overlays: vec![],
4043 deletion_file: None,
4044 row_id_meta,
4045 physical_rows: Some(5),
4046 last_updated_at_version_meta: Some(version_meta.clone()),
4047 created_at_version_meta: Some(version_meta),
4048 };
4049
4050 let manifest = make_stable_row_id_manifest(vec![fragment.clone()]);
4051
4052 let off_map = HashMap::from([(1u64, RoaringBitmap::from_iter(0u32..5))]);
4054 let tx = Transaction::new(
4055 manifest.version,
4056 Operation::Update {
4057 removed_fragment_ids: vec![],
4058 updated_fragments: vec![fragment],
4059 new_fragments: vec![],
4060 fields_modified: vec![],
4061 compacted_sstables: vec![],
4062 fields_for_preserving_frag_bitmap: vec![],
4063 update_mode: Some(UpdateMode::RewriteColumns),
4064 inserted_rows_filter: None,
4065 updated_fragment_offsets: Some(UpdatedFragmentOffsets(off_map)),
4066 },
4067 None,
4068 );
4069
4070 tx.build_manifest(Some(&manifest), vec![], "txn", &default_build_config())
4071 .expect("bitmap at exact physical_rows boundary should succeed");
4072 }
4073
4074 #[test]
4075 fn test_updated_fragment_offsets_key_not_in_updated_fragments_is_rejected() {
4076 let make_fragment = |id: u64| {
4080 let row_ids = RowIdSequence::from([id * 10].as_slice());
4081 let row_id_meta = Some(RowIdMeta::Inline(write_row_ids(&row_ids).into()));
4082 Fragment {
4083 id,
4084 files: vec![DataFile::new(
4085 format!("{id}.lance"),
4086 vec![0],
4087 vec![0],
4088 LanceFileVersion::Stable.resolve(),
4089 None,
4090 None,
4091 )],
4092 overlays: vec![],
4093 deletion_file: None,
4094 row_id_meta,
4095 physical_rows: Some(5),
4096 last_updated_at_version_meta: None,
4097 created_at_version_meta: None,
4098 }
4099 };
4100
4101 let frag_a = make_fragment(1);
4102 let frag_b = make_fragment(2);
4103 let manifest = make_stable_row_id_manifest(vec![frag_a.clone(), frag_b.clone()]);
4104
4105 let off_map = HashMap::from([(frag_b.id, RoaringBitmap::from_iter([0u32, 1, 2]))]);
4107 let operation = Operation::Update {
4108 removed_fragment_ids: vec![],
4109 updated_fragments: vec![frag_a],
4110 new_fragments: vec![],
4111 fields_modified: vec![],
4112 compacted_sstables: vec![],
4113 fields_for_preserving_frag_bitmap: vec![],
4114 update_mode: Some(UpdateMode::RewriteColumns),
4115 inserted_rows_filter: None,
4116 updated_fragment_offsets: Some(UpdatedFragmentOffsets(off_map)),
4117 };
4118
4119 let err = validate_operation(Some(&manifest), &operation).unwrap_err();
4120 assert!(
4121 err.to_string().contains("not in updated_fragments"),
4122 "expected key-presence error, got: {err}"
4123 );
4124 }
4125
4126 #[test]
4127 fn test_proto_round_trip_field_10() {
4128 let off_map = HashMap::from([
4129 (1u64, RoaringBitmap::from_iter([1u32, 3, 5])),
4130 (2u64, RoaringBitmap::from_iter([0u32, 2, 4, 6])),
4131 ]);
4132 let tx = Transaction::new(
4133 1,
4134 Operation::Update {
4135 removed_fragment_ids: vec![],
4136 updated_fragments: vec![],
4137 new_fragments: vec![],
4138 fields_modified: vec![],
4139 compacted_sstables: vec![],
4140 fields_for_preserving_frag_bitmap: vec![],
4141 update_mode: Some(UpdateMode::RewriteColumns),
4142 inserted_rows_filter: None,
4143 updated_fragment_offsets: Some(UpdatedFragmentOffsets(off_map.clone())),
4144 },
4145 None,
4146 );
4147
4148 let pb_tx: pb::Transaction = pb::Transaction::try_from(&tx).unwrap();
4149
4150 if let Some(pb::transaction::Operation::Update(ref update)) = pb_tx.operation {
4152 assert!(
4153 update.updated_fragment_offsets.is_empty(),
4154 "field 9 should be empty"
4155 );
4156 assert_eq!(update.updated_fragment_offset_bitmaps.len(), 2);
4157 } else {
4158 panic!("expected Update operation");
4159 }
4160
4161 let tx2 = Transaction::try_from(pb_tx).unwrap();
4162 if let Operation::Update {
4163 updated_fragment_offsets: Some(UpdatedFragmentOffsets(m)),
4164 ..
4165 } = &tx2.operation
4166 {
4167 assert_eq!(m.len(), 2);
4168 assert_eq!(*m.get(&1).unwrap(), off_map[&1]);
4169 assert_eq!(*m.get(&2).unwrap(), off_map[&2]);
4170 } else {
4171 panic!("expected Update with offsets");
4172 }
4173 }
4174
4175 #[test]
4176 fn test_proto_legacy_field_9_read() {
4177 let pb_tx = pb::Transaction {
4179 read_version: 1,
4180 uuid: "test".to_string(),
4181 tag: String::new(),
4182 transaction_properties: HashMap::new(),
4183 operation: Some(pb::transaction::Operation::Update(
4184 pb::transaction::Update {
4185 removed_fragment_ids: vec![],
4186 updated_fragments: vec![],
4187 new_fragments: vec![],
4188 fields_modified: vec![],
4189 compacted_sstables: vec![],
4190 fields_for_preserving_frag_bitmap: vec![],
4191 update_mode: 1,
4192 inserted_rows: None,
4193 updated_fragment_offsets: HashMap::from([(
4194 1u64,
4195 pb::transaction::UInt32List {
4196 values: vec![1, 3, 5],
4197 },
4198 )]),
4199 updated_fragment_offset_bitmaps: HashMap::new(),
4200 },
4201 )),
4202 };
4203
4204 let tx = Transaction::try_from(pb_tx).unwrap();
4205 if let Operation::Update {
4206 updated_fragment_offsets: Some(UpdatedFragmentOffsets(m)),
4207 ..
4208 } = &tx.operation
4209 {
4210 assert_eq!(m.len(), 1);
4211 let bitmap = m.get(&1).unwrap();
4212 let offsets: Vec<u32> = bitmap.iter().collect();
4213 assert_eq!(offsets, vec![1, 3, 5]);
4214 } else {
4215 panic!("expected Update with offsets from legacy field 9");
4216 }
4217 }
4218
4219 #[test]
4220 fn test_proto_field_10_takes_precedence_over_field_9() {
4221 let mut bitmap_bytes = Vec::new();
4223 RoaringBitmap::from_iter([10u32, 20, 30])
4224 .serialize_into(&mut bitmap_bytes)
4225 .unwrap();
4226
4227 let pb_tx = pb::Transaction {
4228 read_version: 1,
4229 uuid: "test".to_string(),
4230 tag: String::new(),
4231 transaction_properties: HashMap::new(),
4232 operation: Some(pb::transaction::Operation::Update(
4233 pb::transaction::Update {
4234 removed_fragment_ids: vec![],
4235 updated_fragments: vec![],
4236 new_fragments: vec![],
4237 fields_modified: vec![],
4238 compacted_sstables: vec![],
4239 fields_for_preserving_frag_bitmap: vec![],
4240 update_mode: 1,
4241 inserted_rows: None,
4242 updated_fragment_offsets: HashMap::from([(
4244 1u64,
4245 pb::transaction::UInt32List {
4246 values: vec![99, 100],
4247 },
4248 )]),
4249 updated_fragment_offset_bitmaps: HashMap::from([(1u64, bitmap_bytes)]),
4250 },
4251 )),
4252 };
4253
4254 let tx = Transaction::try_from(pb_tx).unwrap();
4255 if let Operation::Update {
4256 updated_fragment_offsets: Some(UpdatedFragmentOffsets(m)),
4257 ..
4258 } = &tx.operation
4259 {
4260 let offsets: Vec<u32> = m.get(&1).unwrap().iter().collect();
4261 assert_eq!(offsets, vec![10, 20, 30], "field 10 should take precedence");
4262 } else {
4263 panic!("expected Update with offsets from field 10");
4264 }
4265 }
4266
4267 #[test]
4268 fn merge_build_manifest_refreshes_last_updated_when_data_files_change_stable_row_ids() {
4269 use crate::feature_flags::FLAG_STABLE_ROW_IDS;
4270 use lance_file::version::LanceFileVersion;
4271
4272 let mk_file = |path: &str| {
4273 DataFile::new(
4274 path,
4275 vec![0],
4276 vec![0],
4277 LanceFileVersion::Stable.resolve(),
4278 None,
4279 None,
4280 )
4281 };
4282
4283 let arrow_schema = ArrowSchema::new(vec![ArrowField::new("id", DataType::Int32, false)]);
4284 let lance_schema = LanceSchema::try_from(&arrow_schema).unwrap();
4285
4286 let row_ids = RowIdSequence::from([100u64, 101, 102, 103, 104].as_slice());
4287 let row_id_meta = Some(RowIdMeta::Inline(write_row_ids(&row_ids).into()));
4288
4289 let prev_fragment = Fragment {
4290 id: 0,
4291 files: vec![mk_file("before.lance")],
4292 overlays: vec![],
4293 deletion_file: None,
4294 row_id_meta,
4295 physical_rows: Some(5),
4296 last_updated_at_version_meta: None,
4297 created_at_version_meta: None,
4298 };
4299
4300 let mut manifest = Manifest::new(
4301 lance_schema.clone(),
4302 Arc::new(vec![prev_fragment.clone()]),
4303 DataStorageFormat::new(ConcreteFileVersion::V2_0),
4304 HashMap::new(),
4305 );
4306 manifest.reader_feature_flags |= FLAG_STABLE_ROW_IDS;
4307 manifest.next_row_id = 100;
4308
4309 let merged_fragment = Fragment {
4310 files: vec![mk_file("after.lance")],
4311 ..prev_fragment
4312 };
4313
4314 let tx = Transaction::new(
4315 manifest.version,
4316 Operation::Merge {
4317 fragments: vec![merged_fragment],
4318 schema: lance_schema,
4319 preserves_nullability: true,
4320 },
4321 None,
4322 );
4323
4324 let (out, _) = tx
4325 .build_manifest(Some(&manifest), vec![], "txn", &default_build_config())
4326 .unwrap();
4327
4328 assert_eq!(out.version, 2);
4329 let frag = &out.fragments[0];
4330 let seq = frag
4331 .last_updated_at_version_meta
4332 .as_ref()
4333 .unwrap()
4334 .load_sequence()
4335 .unwrap();
4336 assert_eq!(seq.version_at(0).unwrap(), 2);
4337 assert_eq!(seq.version_at(4).unwrap(), 2);
4338 }
4339
4340 #[test]
4341 fn merge_build_manifest_skips_refresh_when_carry_forward_stable_row_ids() {
4342 use crate::feature_flags::FLAG_STABLE_ROW_IDS;
4343 use crate::rowids::version::{RowDatasetVersionMeta, RowDatasetVersionSequence};
4344 use lance_file::version::LanceFileVersion;
4345
4346 let data_file = DataFile::new(
4347 "same.lance",
4348 vec![0],
4349 vec![0],
4350 LanceFileVersion::Stable.resolve(),
4351 None,
4352 None,
4353 );
4354
4355 let arrow_schema = ArrowSchema::new(vec![ArrowField::new("id", DataType::Int32, false)]);
4356 let lance_schema = LanceSchema::try_from(&arrow_schema).unwrap();
4357
4358 let row_ids = RowIdSequence::from([200u64, 201, 202, 203, 204].as_slice());
4359 let row_id_meta = Some(RowIdMeta::Inline(write_row_ids(&row_ids).into()));
4360
4361 let uniform_v1 = RowDatasetVersionSequence::from_uniform_row_count(5, 1);
4362 let meta_v1 = RowDatasetVersionMeta::from_sequence(&uniform_v1).unwrap();
4363
4364 let prev_fragment = Fragment {
4365 id: 0,
4366 files: vec![data_file.clone()],
4367 overlays: vec![],
4368 deletion_file: None,
4369 row_id_meta: row_id_meta.clone(),
4370 physical_rows: Some(5),
4371 last_updated_at_version_meta: Some(meta_v1.clone()),
4372 created_at_version_meta: None,
4373 };
4374
4375 let mut manifest = Manifest::new(
4376 lance_schema.clone(),
4377 Arc::new(vec![prev_fragment]),
4378 DataStorageFormat::new(ConcreteFileVersion::V2_0),
4379 HashMap::new(),
4380 );
4381 manifest.reader_feature_flags |= FLAG_STABLE_ROW_IDS;
4382 manifest.next_row_id = 100;
4383
4384 let merged_fragment = Fragment {
4385 id: 0,
4386 files: vec![data_file],
4387 overlays: vec![],
4388 deletion_file: None,
4389 row_id_meta,
4390 physical_rows: Some(5),
4391 last_updated_at_version_meta: Some(meta_v1),
4392 created_at_version_meta: None,
4393 };
4394
4395 let tx = Transaction::new(
4396 manifest.version,
4397 Operation::Merge {
4398 fragments: vec![merged_fragment],
4399 schema: lance_schema,
4400 preserves_nullability: true,
4401 },
4402 None,
4403 );
4404
4405 let (out, _) = tx
4406 .build_manifest(Some(&manifest), vec![], "txn", &default_build_config())
4407 .unwrap();
4408
4409 let seq = out.fragments[0]
4410 .last_updated_at_version_meta
4411 .as_ref()
4412 .unwrap()
4413 .load_sequence()
4414 .unwrap();
4415 assert_eq!(seq.version_at(0).unwrap(), 1);
4416 assert_eq!(seq.version_at(4).unwrap(), 1);
4417 }
4418
4419 fn merge_test_file(path: &str) -> DataFile {
4420 DataFile::new(
4421 path,
4422 vec![0],
4423 vec![0],
4424 LanceFileVersion::Stable.resolve(),
4425 None,
4426 None,
4427 )
4428 }
4429
4430 fn frag_with_row_ids(id: u64, path: &str, row_ids: &[u64]) -> Fragment {
4432 Fragment {
4433 id,
4434 files: vec![merge_test_file(path)],
4435 overlays: vec![],
4436 deletion_file: None,
4437 row_id_meta: Some(RowIdMeta::Inline(
4438 write_row_ids(&RowIdSequence::from(row_ids)).into(),
4439 )),
4440 physical_rows: Some(row_ids.len()),
4441 last_updated_at_version_meta: None,
4442 created_at_version_meta: None,
4443 }
4444 }
4445
4446 fn frag_without_row_ids(id: u64, path: &str, physical_rows: usize) -> Fragment {
4449 Fragment {
4450 id,
4451 files: vec![merge_test_file(path)],
4452 overlays: vec![],
4453 deletion_file: None,
4454 row_id_meta: None,
4455 physical_rows: Some(physical_rows),
4456 last_updated_at_version_meta: None,
4457 created_at_version_meta: None,
4458 }
4459 }
4460
4461 fn merge_test_manifest(existing: Vec<Fragment>, stable: bool) -> (Manifest, LanceSchema) {
4464 use crate::feature_flags::FLAG_STABLE_ROW_IDS;
4465
4466 let arrow_schema = ArrowSchema::new(vec![ArrowField::new("id", DataType::Int32, false)]);
4467 let lance_schema = LanceSchema::try_from(&arrow_schema).unwrap();
4468 let mut manifest = Manifest::new(
4469 lance_schema.clone(),
4470 Arc::new(existing),
4471 DataStorageFormat::new(ConcreteFileVersion::V2_0),
4472 HashMap::new(),
4473 );
4474 if stable {
4475 manifest.reader_feature_flags |= FLAG_STABLE_ROW_IDS;
4476 manifest.next_row_id = 100;
4477 }
4478 (manifest, lance_schema)
4479 }
4480
4481 fn build_merge(
4483 manifest: &Manifest,
4484 schema: LanceSchema,
4485 fragments: Vec<Fragment>,
4486 ) -> Result<Manifest> {
4487 let operation = Operation::Merge {
4488 fragments,
4489 schema,
4490 preserves_nullability: true,
4491 };
4492 validate_operation(Some(manifest), &operation)?;
4493 let tx = Transaction::new(manifest.version, operation, None);
4494 let (out, _) = tx.build_manifest(Some(manifest), vec![], "txn", &default_build_config())?;
4495 Ok(out)
4496 }
4497
4498 #[rstest::rstest]
4499 #[case::placeholder_id_0(0, vec![0, 1, 2], 2)]
4500 #[case::pre_reserved_id_7(7, vec![0, 1, 7], 7)]
4501 fn merge_build_manifest_assigns_ids_and_row_ids_to_staged_fragments(
4502 #[case] staged_id: u64,
4503 #[case] expected_ids: Vec<u64>,
4504 #[case] new_id: u64,
4505 ) {
4506 let existing = vec![
4510 frag_with_row_ids(0, "frag0.lance", &[0, 1, 2]),
4511 frag_with_row_ids(1, "frag1.lance", &[3, 4]),
4512 ];
4513 let (manifest, schema) = merge_test_manifest(existing.clone(), true);
4514
4515 let mut merge_list = existing.clone();
4516 merge_list.push(frag_without_row_ids(staged_id, "staged.lance", 4));
4517 let out = build_merge(&manifest, schema, merge_list).unwrap();
4518
4519 let ids: Vec<u64> = out.fragments.iter().map(|f| f.id).collect();
4520 assert_eq!(ids, expected_ids);
4521 assert_eq!(out.max_fragment_id, Some(new_id.max(1) as u32));
4522
4523 for prev in &existing {
4525 let frag = out.fragments.iter().find(|f| f.id == prev.id).unwrap();
4526 assert_eq!(frag.row_id_meta, prev.row_id_meta);
4527 }
4528
4529 let new_frag = out.fragments.iter().find(|f| f.id == new_id).unwrap();
4531 let Some(RowIdMeta::Inline(data)) = &new_frag.row_id_meta else {
4532 panic!("staged fragment must have inline row id metadata");
4533 };
4534 let row_ids: Vec<u64> = crate::rowids::read_row_ids(data).unwrap().iter().collect();
4535 assert_eq!(row_ids, vec![100, 101, 102, 103]);
4536 assert_eq!(out.next_row_id, 104);
4537
4538 assert_eq!(created_at_versions(&out, new_id), vec![2, 2, 2, 2]);
4540 assert_eq!(last_updated_at_versions(&out, new_id), vec![2, 2, 2, 2]);
4541 }
4542
4543 #[test]
4544 fn merge_build_manifest_assigns_unique_ids_to_staged_fragments_non_stable() {
4545 let existing = vec![
4548 frag_without_row_ids(0, "frag0.lance", 2),
4549 frag_without_row_ids(1, "frag1.lance", 2),
4550 ];
4551 let (manifest, schema) = merge_test_manifest(existing.clone(), false);
4552
4553 let mut merge_list = existing;
4554 merge_list.push(frag_without_row_ids(0, "staged.lance", 2));
4555 let out = build_merge(&manifest, schema, merge_list).unwrap();
4556
4557 let ids: Vec<u64> = out.fragments.iter().map(|f| f.id).collect();
4558 assert_eq!(ids, vec![0, 1, 2]);
4559 assert!(out.fragments.iter().all(|f| f.row_id_meta.is_none()));
4560 }
4561
4562 #[test]
4563 fn merge_staged_fragment_is_not_a_rewrite_of_fragment_0() {
4564 let existing = vec![
4567 frag_without_row_ids(0, "frag0.lance", 2),
4568 frag_without_row_ids(1, "frag1.lance", 2),
4569 ];
4570 let mut merge_list = existing.clone();
4571 merge_list.push(frag_without_row_ids(0, "staged.lance", 2));
4572
4573 assert!(Transaction::merge_rewritten_fields(&existing, &merge_list).is_empty());
4574 }
4575
4576 #[rstest::rstest]
4577 #[case::complete_stable(&[0, 1], true)]
4578 #[case::partial_stable(&[0], true)]
4579 #[case::complete_non_stable(&[0, 1], false)]
4580 fn merge_build_manifest_rejects_row_ids_on_new_fragments(
4581 #[case] supplied_row_ids: &[u64],
4582 #[case] stable: bool,
4583 ) {
4584 let existing = if stable {
4587 frag_with_row_ids(0, "frag0.lance", &[0, 1])
4588 } else {
4589 frag_without_row_ids(0, "frag0.lance", 2)
4590 };
4591 let (mut manifest, schema) = merge_test_manifest(vec![existing.clone()], stable);
4592 if stable {
4593 manifest.next_row_id = 2;
4594 }
4595
4596 let mut staged = frag_with_row_ids(0, "staged.lance", supplied_row_ids);
4597 staged.physical_rows = Some(2);
4598 let err = build_merge(&manifest, schema, vec![existing, staged]).unwrap_err();
4599 assert!(
4600 matches!(err, Error::InvalidInput { .. }),
4601 "unexpected error variant: {}",
4602 err
4603 );
4604 assert!(
4605 err.to_string()
4606 .contains("new fragment (id 0) with row id metadata"),
4607 "unexpected error: {}",
4608 err
4609 );
4610 }
4611
4612 #[test]
4613 fn merge_validate_rejects_duplicate_nonzero_fragment_ids() {
4614 let existing = vec![frag_without_row_ids(1, "frag1.lance", 2)];
4615 let (manifest, schema) = merge_test_manifest(existing.clone(), false);
4616
4617 let mut merge_list = existing;
4618 merge_list.push(frag_without_row_ids(1, "other.lance", 2));
4619 let err = build_merge(&manifest, schema, merge_list).unwrap_err();
4620 assert!(
4621 matches!(err, Error::InvalidInput { .. }),
4622 "unexpected error variant: {}",
4623 err
4624 );
4625 assert!(
4626 err.to_string().contains("duplicate fragment id 1"),
4627 "unexpected error: {}",
4628 err
4629 );
4630 }
4631
4632 #[test]
4633 fn merge_validate_rejects_staged_fragment_listed_before_existing_non_stable() {
4634 let existing = vec![
4639 frag_without_row_ids(0, "frag0.lance", 2),
4640 frag_without_row_ids(1, "frag1.lance", 2),
4641 ];
4642 let (manifest, schema) = merge_test_manifest(existing.clone(), false);
4643
4644 let mut merge_list = vec![frag_without_row_ids(0, "staged.lance", 2)];
4645 merge_list.extend(existing);
4646 let err = build_merge(&manifest, schema, merge_list).unwrap_err();
4647 assert!(
4648 matches!(err, Error::InvalidInput { .. }),
4649 "unexpected error variant: {}",
4650 err
4651 );
4652 assert!(
4653 err.to_string().contains("after a new fragment"),
4654 "unexpected error: {}",
4655 err
4656 );
4657 }
4658
4659 #[test]
4660 fn merge_validate_rejects_staged_fragment_listed_before_existing_stable() {
4661 let existing = frag_with_row_ids(0, "frag0.lance", &[0, 1]);
4664 let (manifest, schema) = merge_test_manifest(vec![existing.clone()], true);
4665
4666 let staged = frag_without_row_ids(0, "staged.lance", 2);
4667 let err = build_merge(&manifest, schema, vec![staged, existing]).unwrap_err();
4668 assert!(
4669 matches!(err, Error::InvalidInput { .. }),
4670 "unexpected error variant: {}",
4671 err
4672 );
4673 assert!(
4674 err.to_string()
4675 .contains("dropped row id metadata for existing fragment 0"),
4676 "unexpected error: {}",
4677 err
4678 );
4679 }
4680
4681 #[test]
4682 fn merge_build_manifest_no_last_updated_refresh_without_stable_row_ids() {
4683 use crate::feature_flags::FLAG_STABLE_ROW_IDS;
4684 use lance_file::version::LanceFileVersion;
4685
4686 let mk_file = |path: &str| {
4687 DataFile::new(
4688 path,
4689 vec![0],
4690 vec![0],
4691 LanceFileVersion::Stable.resolve(),
4692 None,
4693 None,
4694 )
4695 };
4696
4697 let arrow_schema = ArrowSchema::new(vec![ArrowField::new("id", DataType::Int32, false)]);
4698 let lance_schema = LanceSchema::try_from(&arrow_schema).unwrap();
4699
4700 let prev_fragment = Fragment {
4701 id: 0,
4702 files: vec![mk_file("before.lance")],
4703 overlays: vec![],
4704 deletion_file: None,
4705 row_id_meta: None,
4706 physical_rows: Some(5),
4707 last_updated_at_version_meta: None,
4708 created_at_version_meta: None,
4709 };
4710
4711 let manifest = Manifest::new(
4712 lance_schema.clone(),
4713 Arc::new(vec![prev_fragment.clone()]),
4714 DataStorageFormat::new(ConcreteFileVersion::V2_0),
4715 HashMap::new(),
4716 );
4717 assert_eq!(
4718 manifest.reader_feature_flags & FLAG_STABLE_ROW_IDS,
4719 0,
4720 "manifest must not use stable row IDs for this guard test"
4721 );
4722
4723 let merged_fragment = Fragment {
4724 files: vec![mk_file("after.lance")],
4725 ..prev_fragment
4726 };
4727
4728 let tx = Transaction::new(
4729 manifest.version,
4730 Operation::Merge {
4731 fragments: vec![merged_fragment],
4732 schema: lance_schema,
4733 preserves_nullability: true,
4734 },
4735 None,
4736 );
4737
4738 let (out, _) = tx
4739 .build_manifest(Some(&manifest), vec![], "txn", &default_build_config())
4740 .unwrap();
4741
4742 assert!(
4743 out.fragments[0].last_updated_at_version_meta.is_none(),
4744 "without stable row IDs, Merge must not populate per-row last_updated metadata"
4745 );
4746 }
4747
4748 #[test]
4749 fn merge_build_manifest_sets_both_version_meta_for_new_fragment_id_stable_row_ids() {
4750 use crate::feature_flags::FLAG_STABLE_ROW_IDS;
4751 use lance_file::version::LanceFileVersion;
4752
4753 let mk_file = |path: &str| {
4754 DataFile::new(
4755 path,
4756 vec![0],
4757 vec![0],
4758 LanceFileVersion::Stable.resolve(),
4759 None,
4760 None,
4761 )
4762 };
4763
4764 let arrow_schema = ArrowSchema::new(vec![ArrowField::new("id", DataType::Int32, false)]);
4765 let lance_schema = LanceSchema::try_from(&arrow_schema).unwrap();
4766
4767 let row_ids_0 = RowIdSequence::from([10u64, 11, 12].as_slice());
4769 let existing_fragment = Fragment {
4770 id: 0,
4771 files: vec![mk_file("existing.lance")],
4772 overlays: vec![],
4773 deletion_file: None,
4774 row_id_meta: Some(RowIdMeta::Inline(write_row_ids(&row_ids_0).into())),
4775 physical_rows: Some(3),
4776 last_updated_at_version_meta: None,
4777 created_at_version_meta: None,
4778 };
4779
4780 let mut manifest = Manifest::new(
4781 lance_schema.clone(),
4782 Arc::new(vec![existing_fragment.clone()]),
4783 DataStorageFormat::new(ConcreteFileVersion::V2_0),
4784 HashMap::new(),
4785 );
4786 manifest.reader_feature_flags |= FLAG_STABLE_ROW_IDS;
4787 manifest.next_row_id = 100;
4788 manifest.version = 1;
4789
4790 let new_fragment = Fragment {
4793 id: 1,
4794 files: vec![mk_file("new.lance")],
4795 overlays: vec![],
4796 deletion_file: None,
4797 row_id_meta: None,
4798 physical_rows: Some(4),
4799 last_updated_at_version_meta: None,
4800 created_at_version_meta: None,
4801 };
4802
4803 let tx = Transaction::new(
4804 manifest.version,
4805 Operation::Merge {
4806 fragments: vec![existing_fragment, new_fragment],
4807 schema: lance_schema,
4808 preserves_nullability: true,
4809 },
4810 None,
4811 );
4812
4813 let (out, _) = tx
4814 .build_manifest(Some(&manifest), vec![], "txn", &default_build_config())
4815 .unwrap();
4816
4817 assert_eq!(out.version, 2);
4818
4819 let new_frag = out.fragments.iter().find(|f| f.id == 1).unwrap();
4820 let Some(RowIdMeta::Inline(data)) = &new_frag.row_id_meta else {
4821 panic!("new fragment must have inline row id metadata");
4822 };
4823 let row_ids: Vec<u64> = crate::rowids::read_row_ids(data).unwrap().iter().collect();
4824 assert_eq!(row_ids, vec![100, 101, 102, 103]);
4825 assert_eq!(out.next_row_id, 104);
4826
4827 let last_updated_seq = new_frag
4829 .last_updated_at_version_meta
4830 .as_ref()
4831 .expect("new fragment must have last_updated_at_version_meta")
4832 .load_sequence()
4833 .unwrap();
4834 assert_eq!(last_updated_seq.version_at(0).unwrap(), 2);
4835 assert_eq!(last_updated_seq.version_at(3).unwrap(), 2);
4836
4837 let created_seq = new_frag
4839 .created_at_version_meta
4840 .as_ref()
4841 .expect("new fragment must have created_at_version_meta")
4842 .load_sequence()
4843 .unwrap();
4844 assert_eq!(created_seq.version_at(0).unwrap(), 2);
4845 assert_eq!(created_seq.version_at(3).unwrap(), 2);
4846 }
4847
4848 #[test]
4849 fn test_data_overlay_build_manifest_multi_fragment() {
4850 let mut frag0 = Fragment::new(0);
4856 frag0.overlays = vec![overlay_with_field(5, 3)]; let frag1 = Fragment::new(1);
4858 let mut frag2 = Fragment::new(2);
4859 frag2.overlays = vec![overlay_with_field(9, 3)]; let schema = ArrowSchema::new(vec![ArrowField::new("id", DataType::Int32, false)]);
4861 let mut manifest = Manifest::new(
4862 LanceSchema::try_from(&schema).unwrap(),
4863 Arc::new(vec![frag0, frag1, frag2]),
4864 crate::format::DataStorageFormat::new(ConcreteFileVersion::V2_0),
4865 HashMap::new(),
4866 );
4867 manifest.version = 3;
4871
4872 let txn = Transaction::new(
4873 manifest.version,
4874 Operation::DataOverlay {
4875 groups: vec![
4876 DataOverlayGroup {
4877 fragment_id: 0,
4878 overlays: vec![overlay_with_field(1, 0)],
4879 },
4880 DataOverlayGroup {
4881 fragment_id: 1,
4882 overlays: vec![overlay_with_field(2, 0)],
4883 },
4884 ],
4885 },
4886 None,
4887 );
4888
4889 let (result, _) = txn
4890 .build_manifest(Some(&manifest), vec![], "txn", &default_build_config())
4891 .unwrap();
4892
4893 let frag = |id: u64| {
4894 result
4895 .fragments
4896 .iter()
4897 .find(|f| f.id == id)
4898 .unwrap_or_else(|| panic!("fragment {id} missing from result"))
4899 };
4900 assert_eq!(frag(0).overlays.len(), 2);
4903 assert_eq!(frag(0).overlays[0].committed_version, 3);
4904 assert_eq!(frag(0).overlays[1].committed_version, result.version);
4905 assert_eq!(frag(1).overlays.len(), 1);
4907 assert_eq!(frag(1).overlays[0].committed_version, result.version);
4908 assert_eq!(frag(2).overlays.len(), 1);
4910 assert_eq!(frag(2).overlays[0].committed_version, 3);
4911 assert!(result.version > manifest.version);
4912 }
4913
4914 #[test]
4915 fn test_data_overlay_build_manifest_updates_row_lineage() {
4916 let row_ids = RowIdSequence::from([10u64, 11, 12, 13].as_slice());
4917 let version_meta = RowDatasetVersionMeta::from_sequence(
4918 &RowDatasetVersionSequence::from_uniform_row_count(4, 1),
4919 )
4920 .unwrap();
4921 let fragment = Fragment {
4922 id: 1,
4923 files: vec![],
4924 overlays: vec![],
4925 deletion_file: None,
4926 row_id_meta: Some(RowIdMeta::Inline(write_row_ids(&row_ids).into())),
4927 physical_rows: Some(4),
4928 last_updated_at_version_meta: Some(version_meta.clone()),
4929 created_at_version_meta: Some(version_meta),
4930 };
4931 let untracked_row_ids = RowIdSequence::from([20u64, 21].as_slice());
4934 let untracked_fragment = Fragment {
4935 id: 2,
4936 files: vec![],
4937 overlays: vec![],
4938 deletion_file: None,
4939 row_id_meta: Some(RowIdMeta::Inline(write_row_ids(&untracked_row_ids).into())),
4940 physical_rows: Some(2),
4941 last_updated_at_version_meta: None,
4942 created_at_version_meta: None,
4943 };
4944 let manifest = make_stable_row_id_manifest(vec![fragment, untracked_fragment]);
4945 let txn = Transaction::new(
4946 manifest.version,
4947 Operation::DataOverlay {
4948 groups: vec![
4949 DataOverlayGroup {
4950 fragment_id: 1,
4951 overlays: vec![
4952 DataOverlayFile {
4953 data_file: DataFile::new_legacy_from_fields(
4954 "dense.lance",
4955 vec![1],
4956 None,
4957 ),
4958 coverage: OverlayCoverage::dense(RoaringBitmap::from_iter([0u32])),
4959 committed_version: 0,
4960 },
4961 DataOverlayFile {
4962 data_file: DataFile::new_legacy_from_fields(
4963 "sparse.lance",
4964 vec![1, 2],
4965 None,
4966 ),
4967 coverage: OverlayCoverage::sparse(vec![
4968 RoaringBitmap::from_iter([2u32]),
4969 RoaringBitmap::from_iter([3u32]),
4970 ]),
4971 committed_version: 0,
4972 },
4973 ],
4974 },
4975 DataOverlayGroup {
4976 fragment_id: 2,
4977 overlays: vec![DataOverlayFile {
4978 data_file: DataFile::new_legacy_from_fields(
4979 "migrated.lance",
4980 vec![1],
4981 None,
4982 ),
4983 coverage: OverlayCoverage::dense(RoaringBitmap::from_iter([1u32])),
4984 committed_version: 0,
4985 }],
4986 },
4987 ],
4988 },
4989 None,
4990 );
4991
4992 let (result, _) = txn
4993 .build_manifest(Some(&manifest), vec![], "txn", &default_build_config())
4994 .unwrap();
4995
4996 assert_eq!(last_updated_at_versions(&result, 1), vec![5, 1, 5, 5]);
4997 assert_eq!(last_updated_at_versions(&result, 2), vec![1, 5]);
4998 assert!(
4999 result.fragments[0]
5000 .overlays
5001 .iter()
5002 .all(|overlay| overlay.committed_version == result.version)
5003 );
5004 }
5005
5006 #[test]
5007 fn test_data_replacement_tombstones_overlaid_fields() {
5008 let mut fragment = Fragment::new(0);
5014 fragment.files = vec![
5015 DataFile::new_legacy_from_fields("f3.lance", vec![3], None),
5016 DataFile::new_legacy_from_fields("f5.lance", vec![5], None),
5017 ];
5018 fragment.overlays = vec![
5019 DataOverlayFile {
5020 data_file: DataFile::new_legacy_from_fields("o35.lance", vec![3, 5], None),
5021 coverage: OverlayCoverage::sparse(vec![
5022 roaring::RoaringBitmap::from_iter([0u32]),
5023 roaring::RoaringBitmap::from_iter([0u32]),
5024 ]),
5025 committed_version: 1,
5026 },
5027 DataOverlayFile {
5028 data_file: DataFile::new_legacy_from_fields("o5.lance", vec![5], None),
5029 coverage: OverlayCoverage::dense(roaring::RoaringBitmap::from_iter([0u32])),
5030 committed_version: 1,
5031 },
5032 ];
5033
5034 let schema = ArrowSchema::new(vec![ArrowField::new("id", DataType::Int32, false)]);
5035 let manifest = Manifest::new(
5036 LanceSchema::try_from(&schema).unwrap(),
5037 Arc::new(vec![fragment]),
5038 crate::format::DataStorageFormat::new(ConcreteFileVersion::V2_0),
5039 HashMap::new(),
5040 );
5041
5042 let txn = Transaction::new(
5043 manifest.version,
5044 Operation::DataReplacement {
5045 replacements: vec![DataReplacementGroup(
5046 0,
5047 DataFile::new_legacy_from_fields("f5-new.lance", vec![5], None),
5048 )],
5049 },
5050 None,
5051 );
5052
5053 let (result, _) = txn
5054 .build_manifest(Some(&manifest), vec![], "txn", &default_build_config())
5055 .unwrap();
5056
5057 let frag = &result.fragments[0];
5058 assert!(frag.files.iter().any(|f| f.path == "f5-new.lance"));
5060 assert_eq!(frag.overlays.len(), 1);
5063 assert_eq!(frag.overlays[0].data_file.fields.as_ref(), &[3, -2]);
5064 }
5065
5066 fn replace_fields(
5070 fragment: Fragment,
5071 fields: Vec<i32>,
5072 manifest_version: u64,
5073 read_version: u64,
5074 ) -> Result<Fragment> {
5075 let schema = ArrowSchema::new(vec![
5076 ArrowField::new("x", DataType::Int32, true),
5077 ArrowField::new("a", DataType::Int32, true),
5078 ArrowField::new("v", DataType::Int32, true),
5079 ArrowField::new("y", DataType::Int32, true),
5080 ]);
5081 let mut lance_schema = LanceSchema::try_from(&schema).unwrap();
5082 lance_schema.fields[0].id = 3;
5083 lance_schema.fields[1].id = 4;
5084 lance_schema.fields[2].id = 5;
5085 lance_schema.fields[3].id = 6;
5086 let mut manifest = Manifest::new(
5087 lance_schema,
5088 Arc::new(vec![fragment]),
5089 crate::format::DataStorageFormat::new(ConcreteFileVersion::V2_0),
5090 HashMap::new(),
5091 );
5092 manifest.version = manifest_version;
5093
5094 let column_indices = (0..fields.len() as i32).collect();
5095 let txn = Transaction::new(
5096 read_version,
5097 Operation::DataReplacement {
5098 replacements: vec![DataReplacementGroup(
5099 0,
5100 DataFile::new(
5101 "v-new.lance",
5102 fields,
5103 column_indices,
5104 ConcreteFileVersion::V2_0,
5105 None,
5106 None,
5107 ),
5108 )],
5109 },
5110 None,
5111 );
5112 txn.build_manifest(Some(&manifest), vec![], "txn", &default_build_config())
5113 .map(|(manifest, _)| manifest.fragments[0].clone())
5114 }
5115
5116 fn replace_field_5(
5119 fragment: Fragment,
5120 manifest_version: u64,
5121 read_version: u64,
5122 ) -> Result<Fragment> {
5123 replace_fields(fragment, vec![5], manifest_version, read_version)
5124 }
5125
5126 #[test]
5127 fn test_data_replacement_rejects_subset_of_legacy_file() {
5128 let mut fragment = Fragment::new(0);
5133 fragment.files = vec![DataFile::new_legacy_from_fields(
5134 "wide.lance",
5135 vec![4, 5],
5136 None,
5137 )];
5138
5139 let result = replace_field_5(fragment, 1, 1);
5140 assert!(
5141 result.is_err(),
5142 "legacy subset replacement must be rejected, got: {:?}",
5143 result.map(|fragment| fragment.files)
5144 );
5145 }
5146
5147 #[test]
5148 fn test_data_replacement_tombstones_fields_spanning_files() {
5149 let mut fragment = Fragment::new(0);
5153 fragment.files = vec![
5154 DataFile::new(
5155 "ab.lance",
5156 vec![3, 4],
5157 vec![0, 1],
5158 ConcreteFileVersion::V2_0,
5159 None,
5160 None,
5161 ),
5162 DataFile::new(
5163 "cd.lance",
5164 vec![5, 6],
5165 vec![0, 1],
5166 ConcreteFileVersion::V2_0,
5167 None,
5168 None,
5169 ),
5170 ];
5171
5172 let fragment = replace_fields(fragment, vec![4, 5], 1, 1).unwrap();
5173 let file = |path| {
5174 fragment
5175 .files
5176 .iter()
5177 .find(|file| file.path == path)
5178 .unwrap_or_else(|| panic!("{path} survives on its live field"))
5179 };
5180 assert_eq!(file("ab.lance").fields.as_ref(), &[3, TOMBSTONE_FIELD_ID]);
5181 assert_eq!(file("cd.lance").fields.as_ref(), &[TOMBSTONE_FIELD_ID, 6]);
5182 assert!(fragment.files.iter().any(|file| file.path == "v-new.lance"));
5183 }
5184
5185 #[test]
5186 fn test_data_replacement_rejects_fields_spanning_a_legacy_file() {
5187 let mut fragment = Fragment::new(0);
5191 fragment.files = vec![
5192 DataFile::new(
5193 "ab.lance",
5194 vec![3, 4],
5195 vec![0, 1],
5196 ConcreteFileVersion::V2_0,
5197 None,
5198 None,
5199 ),
5200 DataFile::new_legacy_from_fields("cd.lance", vec![5, 6], None),
5201 ];
5202
5203 let result = replace_fields(fragment, vec![4, 5], 1, 1);
5204 assert!(
5205 result.is_err(),
5206 "spanning a legacy file must be rejected, got: {:?}",
5207 result.map(|fragment| fragment.files)
5208 );
5209 }
5210
5211 #[test]
5212 fn test_data_replacement_retombstones_wider_file() {
5213 let mut fragment = Fragment::new(0);
5217 fragment.files = vec![DataFile::new(
5218 "wide.lance",
5219 vec![4, TOMBSTONE_FIELD_ID, 5],
5220 vec![0, 1, 2],
5221 ConcreteFileVersion::V2_0,
5222 None,
5223 None,
5224 )];
5225
5226 let fragment = replace_fields(fragment, vec![5], 1, 1).unwrap();
5227 let wide = fragment
5228 .files
5229 .iter()
5230 .find(|file| file.path == "wide.lance")
5231 .expect("wider file survives on its live field");
5232 assert_eq!(
5233 wide.fields.as_ref(),
5234 &[4, TOMBSTONE_FIELD_ID, TOMBSTONE_FIELD_ID]
5235 );
5236 assert!(fragment.files.iter().any(|file| file.path == "v-new.lance"));
5237 }
5238
5239 #[test]
5240 fn test_data_replacement_preserves_overlay_newer_than_snapshot() {
5241 let mut fragment = Fragment::new(0);
5246 fragment.files = vec![DataFile::new(
5248 "wide.lance",
5249 vec![4, 5],
5250 vec![0, 1],
5251 ConcreteFileVersion::V2_0,
5252 None,
5253 None,
5254 )];
5255 fragment.overlays = vec![DataOverlayFile {
5256 data_file: DataFile::new(
5257 "newer.lance",
5258 vec![5],
5259 vec![0],
5260 ConcreteFileVersion::V2_0,
5261 None,
5262 None,
5263 ),
5264 coverage: OverlayCoverage::dense(roaring::RoaringBitmap::from_iter([0u32])),
5265 committed_version: 7,
5266 }];
5267
5268 let fragment = replace_field_5(fragment, 7, 6).unwrap();
5270 assert!(fragment.files.iter().any(|f| f.path == "v-new.lance"));
5271 assert_eq!(
5272 fragment.overlays.len(),
5273 1,
5274 "overlay committed after the snapshot must survive"
5275 );
5276 assert_eq!(fragment.overlays[0].data_file.fields.as_ref(), &[5]);
5277 }
5278
5279 #[test]
5283 fn project_keeps_file_carrying_spilled_row_lineage() {
5284 let mut fragment = Fragment::new(0);
5285 fragment.physical_rows = Some(4);
5286 fragment.row_id_meta = Some(RowIdMeta::Column);
5287 fragment.created_at_version_meta = Some(RowDatasetVersionMeta::Column);
5288 fragment.files = vec![
5289 DataFile::new(
5290 "d.lance",
5291 vec![0, 1],
5292 vec![0, 1],
5293 ConcreteFileVersion::V2_0,
5294 None,
5295 None,
5296 ),
5297 DataFile::new(
5298 "l.lance",
5299 vec![ROW_ID_FIELD_ID, ROW_CREATED_AT_VERSION_FIELD_ID],
5300 vec![0, 1],
5301 ConcreteFileVersion::V2_0,
5302 None,
5303 None,
5304 ),
5305 ];
5306 let schema = ArrowSchema::new(vec![
5307 ArrowField::new("a", DataType::Int32, false),
5308 ArrowField::new("b", DataType::Int32, false),
5309 ]);
5310 let mut manifest = Manifest::new(
5311 LanceSchema::try_from(&schema).unwrap(),
5312 Arc::new(vec![fragment]),
5313 DataStorageFormat::new(ConcreteFileVersion::V2_0),
5314 HashMap::new(),
5315 );
5316 manifest.reader_feature_flags = FLAG_STABLE_ROW_IDS;
5317 manifest.writer_feature_flags = FLAG_STABLE_ROW_IDS;
5318
5319 let transaction = Transaction::new(
5321 manifest.version,
5322 Operation::Project {
5323 schema: manifest.schema.project_by_ids(&[0], true),
5324 preserves_nullability: true,
5325 },
5326 None,
5327 );
5328 let (result, _) = transaction
5329 .build_manifest(Some(&manifest), vec![], "txn", &default_build_config())
5330 .unwrap();
5331
5332 let fragment = &result.fragments[0];
5333 let paths = fragment
5334 .files
5335 .iter()
5336 .map(|file| file.path.as_str())
5337 .collect::<Vec<_>>();
5338 assert_eq!(paths, ["d.lance", "l.lance"]);
5339 for field_id in [ROW_ID_FIELD_ID, ROW_CREATED_AT_VERSION_FIELD_ID] {
5340 assert_eq!(
5341 fragment
5342 .row_lineage_file(field_id)
5343 .unwrap()
5344 .map(|file| file.path.as_str()),
5345 Some("l.lance"),
5346 "field {field_id}"
5347 );
5348 }
5349 }
5350
5351 #[rstest::rstest]
5358 #[case::lineage_only_file(
5359 vec![("v.lance", vec![3, 4, 5, 6]), ("l.lance", vec![ROW_ID_FIELD_ID])],
5360 vec![3, 4],
5361 vec![
5362 ("v.lance", vec![TOMBSTONE_FIELD_ID, TOMBSTONE_FIELD_ID, 5, 6]),
5363 ("l.lance", vec![ROW_ID_FIELD_ID]),
5364 ("v-new.lance", vec![3, 4]),
5365 ],
5366 "l.lance"
5367 )]
5368 #[case::in_file_carrier(
5369 vec![("v.lance", vec![3, 4, 5, 6, ROW_ID_FIELD_ID])],
5370 vec![3, 4, 5, 6],
5371 vec![
5372 (
5373 "v.lance",
5374 vec![
5375 TOMBSTONE_FIELD_ID,
5376 TOMBSTONE_FIELD_ID,
5377 TOMBSTONE_FIELD_ID,
5378 TOMBSTONE_FIELD_ID,
5379 ROW_ID_FIELD_ID,
5380 ],
5381 ),
5382 ("v-new.lance", vec![3, 4, 5, 6]),
5383 ],
5384 "v.lance"
5385 )]
5386 fn data_replacement_keeps_lineage_carrier(
5387 #[case] files: Vec<(&str, Vec<i32>)>,
5388 #[case] replaced_fields: Vec<i32>,
5389 #[case] expected_files: Vec<(&str, Vec<i32>)>,
5390 #[case] carrier: &str,
5391 ) {
5392 let mut fragment = Fragment::new(0);
5393 fragment.row_id_meta = Some(RowIdMeta::Column);
5394 fragment.files = files
5395 .into_iter()
5396 .map(|(path, fields)| {
5397 let column_indices = (0..fields.len() as i32).collect();
5398 DataFile::new(
5399 path,
5400 fields,
5401 column_indices,
5402 ConcreteFileVersion::V2_0,
5403 None,
5404 None,
5405 )
5406 })
5407 .collect();
5408
5409 let fragment = replace_fields(fragment, replaced_fields, 1, 1).unwrap();
5410 let files = fragment
5411 .files
5412 .iter()
5413 .map(|file| (file.path.as_str(), file.fields.to_vec()))
5414 .collect::<Vec<_>>();
5415 assert_eq!(files, expected_files);
5416 assert_eq!(
5417 fragment
5418 .row_lineage_file(ROW_ID_FIELD_ID)
5419 .unwrap()
5420 .map(|file| file.path.as_str()),
5421 Some(carrier)
5422 );
5423 }
5424
5425 #[rstest::rstest]
5430 #[case::no_carrier(vec![0], ConcreteFileVersion::V2_0)]
5431 #[case::v1_carrier(vec![0, ROW_ID_FIELD_ID], ConcreteFileVersion::V1)]
5432 fn build_manifest_rejects_spilled_arm_without_carrier(
5433 #[case] fields: Vec<i32>,
5434 #[case] version: ConcreteFileVersion,
5435 ) {
5436 let manifest = sample_manifest();
5437 let mut fragment = Fragment::new(0);
5438 fragment.physical_rows = Some(4);
5439 fragment.row_id_meta = Some(RowIdMeta::Column);
5440 let data_file = DataFile::new("d.lance", fields, vec![], version, None, None);
5442 fragment.files = vec![data_file];
5443 let transaction = Transaction::new(
5444 manifest.version,
5445 Operation::Merge {
5446 fragments: vec![fragment],
5447 schema: manifest.schema.clone(),
5448 preserves_nullability: true,
5449 },
5450 None,
5451 );
5452
5453 let err = transaction
5454 .build_manifest(Some(&manifest), vec![], "txn", &default_build_config())
5455 .unwrap_err();
5456 assert!(matches!(err, Error::Internal { .. }), "{err:?}");
5457 let message = err.to_string();
5458 assert!(
5459 message.contains("fragment 0") && message.contains(&ROW_ID_FIELD_ID.to_string()),
5460 "{message}"
5461 );
5462 }
5463
5464 #[test]
5465 fn test_data_overlay_build_manifest_merges_duplicate_groups() {
5466 let manifest = sample_manifest();
5469 let txn = Transaction::new(
5470 manifest.version,
5471 Operation::DataOverlay {
5472 groups: vec![
5473 DataOverlayGroup {
5474 fragment_id: 0,
5475 overlays: vec![overlay_with_field(1, 0)],
5476 },
5477 DataOverlayGroup {
5478 fragment_id: 0,
5479 overlays: vec![overlay_with_field(2, 0)],
5480 },
5481 ],
5482 },
5483 None,
5484 );
5485
5486 let (result, _) = txn
5487 .build_manifest(Some(&manifest), vec![], "txn", &default_build_config())
5488 .unwrap();
5489
5490 let overlays = &result.fragments[0].overlays;
5491 assert_eq!(overlays.len(), 2);
5492 assert_eq!(overlays[0].data_file.fields.as_ref(), [1i32].as_slice());
5493 assert_eq!(overlays[1].data_file.fields.as_ref(), [2i32].as_slice());
5494 }
5495
5496 #[test]
5497 fn test_data_overlay_build_manifest_rejects_unknown_fragment() {
5498 let manifest = sample_manifest();
5499 let txn = Transaction::new(
5500 manifest.version,
5501 Operation::DataOverlay {
5502 groups: vec![DataOverlayGroup {
5503 fragment_id: 99,
5504 overlays: vec![overlay_with_field(1, 0)],
5505 }],
5506 },
5507 None,
5508 );
5509 let err = txn
5510 .build_manifest(Some(&manifest), vec![], "txn", &default_build_config())
5511 .unwrap_err();
5512 assert!(err.to_string().contains("does not exist"), "{err}");
5513 }
5514
5515 #[test]
5516 fn test_nullability_assertion_defaults_conservative() {
5517 for encoded in [false, true] {
5521 let txn = Transaction::try_from(pb::Transaction {
5522 read_version: 1,
5523 uuid: "test".to_string(),
5524 operation: Some(pb::transaction::Operation::Project(
5525 pb::transaction::Project {
5526 schema: vec![],
5527 preserves_nullability: encoded,
5528 },
5529 )),
5530 ..Default::default()
5531 })
5532 .unwrap();
5533 assert!(
5534 matches!(txn.operation, Operation::Project { preserves_nullability, .. } if preserves_nullability == encoded),
5535 "encoded={encoded:?}"
5536 );
5537
5538 let txn = Transaction::try_from(pb::Transaction {
5539 read_version: 1,
5540 uuid: "test".to_string(),
5541 operation: Some(pb::transaction::Operation::Merge(pb::transaction::Merge {
5542 fragments: vec![],
5543 schema: vec![],
5544 schema_metadata: Default::default(),
5545 preserves_nullability: encoded,
5546 })),
5547 ..Default::default()
5548 })
5549 .unwrap();
5550 assert!(
5551 matches!(txn.operation, Operation::Merge { preserves_nullability, .. } if preserves_nullability == encoded),
5552 "encoded={encoded:?}"
5553 );
5554 }
5555 }
5556
5557 mod mem_wal_index_coverage {
5558 use super::*;
5559 use crate::system_index::mem_wal::{
5560 CompactedSsTable, IndexCatchupProgress, MEM_WAL_INDEX_NAME, MemWalIndexDetails,
5561 };
5562
5563 fn user_index(name: &str, uuid: Uuid, frags: &[u32]) -> IndexMetadata {
5564 IndexMetadata {
5565 uuid,
5566 name: name.to_string(),
5567 fields: vec![0],
5568 covering_fields: vec![],
5569 dataset_version: 1,
5570 fragment_bitmap: Some(RoaringBitmap::from_iter(frags.iter().copied())),
5571 index_details: None,
5572 index_version: 0,
5573 created_at: None,
5574 base_id: None,
5575 files: None,
5576 }
5577 }
5578
5579 fn mem_wal_index(details: MemWalIndexDetails) -> IndexMetadata {
5580 crate::system_index::mem_wal::new_mem_wal_index_meta(1, details).unwrap()
5581 }
5582
5583 fn coverage_for(indices: &[IndexMetadata], name: &str) -> Option<Vec<CompactedSsTable>> {
5584 let meta = indices
5585 .iter()
5586 .find(|idx| idx.name == MEM_WAL_INDEX_NAME)
5587 .expect("mem wal index present");
5588 load_mem_wal_index_details(meta.clone())
5589 .unwrap()
5590 .index_catchup
5591 .into_iter()
5592 .find(|entry| entry.index_name == name)
5593 .map(|entry| entry.caught_up_generations)
5594 }
5595
5596 fn compacted(shard: Uuid, generation: u64) -> Vec<CompactedSsTable> {
5597 vec![CompactedSsTable::new(shard, generation)]
5598 }
5599
5600 fn manifest_with(frags: &[u32]) -> Manifest {
5603 let fragments: Vec<Fragment> =
5604 frags.iter().map(|id| Fragment::new(*id as u64)).collect();
5605 Manifest::new(
5606 LanceSchema::default(),
5607 Arc::new(fragments),
5608 DataStorageFormat::default(),
5609 Default::default(),
5610 )
5611 }
5612
5613 fn apply(
5615 after: &mut [IndexMetadata],
5616 before: &[IndexMetadata],
5617 read_frags: &[u32],
5618 read_indices: &[IndexMetadata],
5619 ) -> Result<()> {
5620 let manifest = manifest_with(read_frags);
5621 let segments_before = Transaction::logical_index_segments(before);
5622 Transaction::apply_mem_wal_index_coverage(
5623 after,
5624 &segments_before,
5625 Some(ReadVersionState {
5626 manifest: &manifest,
5627 indices: read_indices,
5628 }),
5629 2,
5630 )
5631 }
5632
5633 fn table(idx_frags: &[u32], uuid: Uuid, details: MemWalIndexDetails) -> Vec<IndexMetadata> {
5634 vec![user_index("idx", uuid, idx_frags), mem_wal_index(details)]
5635 }
5636
5637 fn progress(shard: Uuid, generation: u64) -> MemWalIndexDetails {
5638 MemWalIndexDetails {
5639 compacted_sstables: compacted(shard, generation),
5640 ..Default::default()
5641 }
5642 }
5643
5644 fn progress_with_catchup(shard: Uuid, generation: u64, caught: u64) -> MemWalIndexDetails {
5645 MemWalIndexDetails {
5646 compacted_sstables: compacted(shard, generation),
5647 index_catchup: vec![IndexCatchupProgress::new(
5648 "idx".to_string(),
5649 compacted(shard, caught),
5650 )],
5651 ..Default::default()
5652 }
5653 }
5654
5655 #[test]
5661 fn a_withdrawal_in_preparation_withdraws_the_catch_up_record() {
5662 let shard = Uuid::new_v4();
5663 let uuid = Uuid::new_v4();
5664 let read = table(&[0, 1], uuid, progress_with_catchup(shard, 5, 5));
5665 let mut manifest = crate::transaction::test_support::sample_manifest();
5667 manifest.fragments = Arc::new(vec![Fragment::new(0), Fragment::new(1)]);
5668 let mut update = crate::transaction::test_support::update_txn(vec![]);
5669 let Operation::Update {
5670 updated_fragments,
5671 fields_modified,
5672 ..
5673 } = &mut update.operation
5674 else {
5675 unreachable!()
5676 };
5677 *updated_fragments = vec![Fragment::new(0)];
5678 *fields_modified = vec![0];
5679 let transaction = Transaction::new(manifest.version, update.operation, None);
5680 let (_, after) = transaction
5681 .build_manifest_with_read_version(
5682 Some(&manifest),
5683 read.clone(),
5684 "txn",
5685 &default_build_config(),
5686 Some(ReadVersionState {
5687 manifest: &manifest,
5688 indices: &read,
5689 }),
5690 )
5691 .unwrap();
5692 let idx = after.iter().find(|idx| idx.name == "idx").unwrap();
5693 assert_eq!(
5694 idx.fragment_bitmap,
5695 Some(RoaringBitmap::from_iter([1u32])),
5696 "the rewritten fragment left the coverage in preparation"
5697 );
5698 assert_eq!(
5699 coverage_for(&after, "idx"),
5700 None,
5701 "a record derived over the original coverage is not carried"
5702 );
5703 }
5704
5705 #[test]
5708 fn an_index_covering_the_read_version_is_credited() {
5709 let shard = Uuid::new_v4();
5710 let read = table(&[0, 1], Uuid::new_v4(), progress(shard, 5));
5711 let mut after = table(&[0, 1], Uuid::new_v4(), progress(shard, 5));
5712 apply(&mut after, &read, &[0, 1], &read).unwrap();
5713 assert_eq!(coverage_for(&after, "idx"), Some(compacted(shard, 5)));
5714 }
5715
5716 #[test]
5719 fn an_index_short_of_the_read_version_is_not_credited() {
5720 let shard = Uuid::new_v4();
5721 let read = table(&[0], Uuid::new_v4(), progress(shard, 5));
5722 let mut after = table(&[0], Uuid::new_v4(), progress(shard, 5));
5723 apply(&mut after, &read, &[0, 1], &read).unwrap();
5724 assert_eq!(coverage_for(&after, "idx"), None);
5725 }
5726
5727 #[test]
5735 fn a_bitmap_pruned_in_place_does_not_keep_its_position() {
5736 let shard = Uuid::new_v4();
5737 let uuid = Uuid::new_v4();
5738 let before = table(&[0, 1], uuid, progress_with_catchup(shard, 5, 5));
5739 let mut after = table(&[0], uuid, progress_with_catchup(shard, 5, 5));
5741 apply(&mut after, &before, &[0, 1], &before).unwrap();
5742 assert_eq!(
5743 coverage_for(&after, "idx"),
5744 None,
5745 "a shrunken index kept a position it no longer earns"
5746 );
5747 }
5748
5749 #[test]
5754 fn an_unchanged_index_is_not_raised_beyond_what_it_proves() {
5755 let shard = Uuid::new_v4();
5756 let uuid = Uuid::new_v4();
5757 let before = table(&[0], uuid, progress_with_catchup(shard, 5, 2));
5759 let mut after = before.clone();
5760 apply(&mut after, &before, &[0, 1], &before).unwrap();
5762 assert_eq!(coverage_for(&after, "idx"), Some(compacted(shard, 2)));
5763 }
5764
5765 #[test]
5769 fn a_carried_position_cannot_exceed_the_committed_progress() {
5770 let shard = Uuid::new_v4();
5771 let uuid = Uuid::new_v4();
5772 let before = table(&[0], uuid, progress_with_catchup(shard, 3, 9));
5773 let mut after = before.clone();
5774 apply(&mut after, &before, &[0], &before).unwrap();
5775 assert_eq!(coverage_for(&after, "idx"), Some(compacted(shard, 3)));
5776 }
5777
5778 #[test]
5781 fn an_unchanged_index_is_never_lowered() {
5782 let shard = Uuid::new_v4();
5783 let uuid = Uuid::new_v4();
5784 let before = table(&[0], uuid, progress_with_catchup(shard, 9, 9));
5785 let mut after = before.clone();
5786 apply(&mut after, &before, &[0, 1], &before).unwrap();
5787 assert_eq!(coverage_for(&after, "idx"), Some(compacted(shard, 9)));
5788 }
5789
5790 #[test]
5794 fn credit_is_capped_by_the_committed_progress() {
5795 let shard = Uuid::new_v4();
5796 let read = table(&[0], Uuid::new_v4(), progress(shard, 9));
5797 let mut after = table(&[0], Uuid::new_v4(), progress(shard, 3));
5798 apply(&mut after, &read, &[0], &read).unwrap();
5799 assert_eq!(coverage_for(&after, "idx"), Some(compacted(shard, 3)));
5800 }
5801
5802 #[test]
5807 fn credit_never_reaches_past_the_read_version() {
5808 let shard = Uuid::new_v4();
5809 let read = table(&[0], Uuid::new_v4(), progress(shard, 2));
5811 let mut after = table(&[0], Uuid::new_v4(), progress(shard, 5));
5812 apply(&mut after, &read, &[0], &read).unwrap();
5813 assert_eq!(coverage_for(&after, "idx"), Some(compacted(shard, 2)));
5814 }
5815
5816 #[test]
5820 fn an_index_with_an_unknown_segment_is_not_credited() {
5821 let shard = Uuid::new_v4();
5822 let mut unknown = user_index("idx", Uuid::new_v4(), &[]);
5823 unknown.fragment_bitmap = None;
5824 let read = vec![
5825 user_index("idx", Uuid::new_v4(), &[0, 1]),
5826 unknown,
5827 mem_wal_index(progress(shard, 5)),
5828 ];
5829 let mut after = read.clone();
5830 apply(&mut after, &read, &[0, 1], &read).unwrap();
5831 assert_eq!(coverage_for(&after, "idx"), None);
5832 }
5833
5834 #[test]
5836 fn a_dropped_index_loses_its_entry() {
5837 let shard = Uuid::new_v4();
5838 let before = table(&[0], Uuid::new_v4(), progress_with_catchup(shard, 5, 5));
5839 let mut after = vec![mem_wal_index(progress_with_catchup(shard, 5, 5))];
5840 apply(&mut after, &before, &[0], &before).unwrap();
5841 assert_eq!(coverage_for(&after, "idx"), None);
5842 }
5843
5844 #[test]
5849 fn a_new_index_covering_the_read_version_is_credited() {
5850 let shard = Uuid::new_v4();
5851 let before = vec![mem_wal_index(progress(shard, 5))];
5852 let mut after = table(&[0], Uuid::new_v4(), progress(shard, 5));
5853 apply(&mut after, &before, &[0], &before).unwrap();
5855 assert_eq!(coverage_for(&after, "idx"), Some(compacted(shard, 5)));
5856 }
5857
5858 #[test]
5862 fn a_table_with_no_catchup_entry_earns_one() {
5863 let shard = Uuid::new_v4();
5864 let before = table(&[0], Uuid::new_v4(), progress(shard, 5));
5865 let mut after = before.clone();
5866 apply(&mut after, &before, &[0], &before).unwrap();
5867 assert_eq!(coverage_for(&after, "idx"), Some(compacted(shard, 5)));
5868 }
5869
5870 #[test]
5872 fn each_shard_is_credited_independently() {
5873 let merged = Uuid::new_v4();
5874 let idle = Uuid::new_v4();
5875 let details = MemWalIndexDetails {
5876 compacted_sstables: vec![
5877 CompactedSsTable::new(merged, 4),
5878 CompactedSsTable::new(idle, 0),
5879 ],
5880 ..Default::default()
5881 };
5882 let read = table(&[0], Uuid::new_v4(), details.clone());
5883 let mut after = table(&[0], Uuid::new_v4(), details);
5884 apply(&mut after, &read, &[0], &read).unwrap();
5885 let coverage = coverage_for(&after, "idx").expect("credited");
5886 assert_eq!(
5887 coverage
5888 .iter()
5889 .find(|g| g.shard_id == merged)
5890 .map(|g| g.generation),
5891 Some(4)
5892 );
5893 assert_eq!(
5894 coverage
5895 .iter()
5896 .find(|g| g.shard_id == idle)
5897 .map(|g| g.generation),
5898 Some(0)
5899 );
5900 }
5901
5902 #[test]
5904 fn indexes_are_credited_independently() {
5905 let shard = Uuid::new_v4();
5906 let read = vec![
5907 user_index("fast", Uuid::new_v4(), &[0, 1]),
5908 user_index("slow", Uuid::new_v4(), &[0]),
5909 mem_wal_index(progress(shard, 6)),
5910 ];
5911 let mut after = read.clone();
5912 apply(&mut after, &read, &[0, 1], &read).unwrap();
5913 assert_eq!(coverage_for(&after, "fast"), Some(compacted(shard, 6)));
5914 assert_eq!(coverage_for(&after, "slow"), None);
5915 }
5916
5917 #[test]
5919 fn an_index_without_a_bitmap_is_not_credited() {
5920 let shard = Uuid::new_v4();
5921 let mut idx = user_index("idx", Uuid::new_v4(), &[0]);
5922 idx.fragment_bitmap = None;
5923 let read = vec![idx, mem_wal_index(progress(shard, 5))];
5924 let mut after = read.clone();
5925 apply(&mut after, &read, &[0], &read).unwrap();
5926 assert_eq!(coverage_for(&after, "idx"), None);
5927 }
5928
5929 #[test]
5931 fn no_compaction_progress_writes_no_entries() {
5932 let before = table(&[0], Uuid::new_v4(), MemWalIndexDetails::default());
5933 let mut after = before.clone();
5934 let untouched = after.clone();
5935 apply(&mut after, &before, &[0], &before).unwrap();
5936 assert_eq!(after, untouched);
5937 }
5938
5939 #[test]
5941 fn a_table_without_mem_wal_is_a_no_op() {
5942 let before = vec![user_index("idx", Uuid::new_v4(), &[0])];
5943 let mut after = before.clone();
5944 let untouched = after.clone();
5945 apply(&mut after, &before, &[0], &before).unwrap();
5946 assert_eq!(after, untouched);
5947 }
5948
5949 #[test]
5952 fn without_a_read_version_nothing_changes() {
5953 let shard = Uuid::new_v4();
5954 let uuid = Uuid::new_v4();
5955 let before = table(&[0], uuid, progress_with_catchup(shard, 5, 5));
5956 let mut after = before.clone();
5957 let segments_before = Transaction::logical_index_segments(&before);
5958 Transaction::apply_mem_wal_index_coverage(&mut after, &segments_before, None, 2)
5959 .unwrap();
5960 assert_eq!(coverage_for(&after, "idx"), Some(compacted(shard, 5)));
5961 }
5962
5963 #[test]
5966 fn an_untrained_index_earns_nothing() {
5967 let shard = Uuid::new_v4();
5968 let read = vec![
5969 user_index("untrained", Uuid::new_v4(), &[]),
5970 user_index("trained", Uuid::new_v4(), &[0]),
5971 mem_wal_index(progress(shard, 10)),
5972 ];
5973 let mut after = read.clone();
5974 apply(&mut after, &read, &[0], &read).unwrap();
5975 assert_eq!(coverage_for(&after, "untrained"), None);
5976 assert_eq!(coverage_for(&after, "trained"), Some(compacted(shard, 10)));
5977 }
5978
5979 #[test]
5982 fn a_shard_keeps_its_position_while_another_advances() {
5983 let (advancing, quiet) = (Uuid::new_v4(), Uuid::new_v4());
5984 let uuid = Uuid::new_v4();
5985 let details = |advancing_gen: u64| MemWalIndexDetails {
5986 compacted_sstables: vec![
5987 CompactedSsTable::new(advancing, advancing_gen),
5988 CompactedSsTable::new(quiet, 10),
5989 ],
5990 index_catchup: vec![IndexCatchupProgress::new(
5991 "idx".to_string(),
5992 vec![CompactedSsTable::new(quiet, 7)],
5993 )],
5994 ..Default::default()
5995 };
5996 let read = vec![
5999 user_index("idx", uuid, &[0]),
6000 mem_wal_index(MemWalIndexDetails {
6001 compacted_sstables: vec![CompactedSsTable::new(advancing, 9)],
6002 ..details(9)
6003 }),
6004 ];
6005 let mut after = vec![user_index("idx", uuid, &[0]), mem_wal_index(details(10))];
6006 apply(&mut after, &read, &[0], &read).unwrap();
6007
6008 let mut coverage = coverage_for(&after, "idx").expect("credited");
6009 coverage.sort_unstable_by_key(|sstable| sstable.shard_id);
6010 let mut expected = vec![
6011 CompactedSsTable::new(advancing, 9),
6012 CompactedSsTable::new(quiet, 7),
6013 ];
6014 expected.sort_unstable_by_key(|sstable| sstable.shard_id);
6015 assert_eq!(coverage, expected);
6016 }
6017
6018 #[test]
6022 fn an_ordinary_index_job_is_never_blocked() {
6023 let shard = Uuid::new_v4();
6024 let before = table(&[0, 1], Uuid::new_v4(), progress_with_catchup(shard, 5, 5));
6025 let mut after = table(&[0], Uuid::new_v4(), progress_with_catchup(shard, 5, 5));
6027 apply(&mut after, &before, &[0, 1], &before).unwrap();
6028 assert_eq!(coverage_for(&after, "idx"), None);
6029 }
6030
6031 #[test]
6035 fn an_index_caught_up_to_nothing_gets_no_entry() {
6036 let shard = Uuid::new_v4();
6037 let uuid = Uuid::new_v4();
6038 let before = table(&[0], uuid, progress_with_catchup(shard, 5, 0));
6039 let mut after = before.clone();
6040 apply(&mut after, &before, &[0, 1], &before).unwrap();
6042 assert_eq!(coverage_for(&after, "idx"), None);
6043 }
6044
6045 #[test]
6048 fn carried_positions_do_not_leak_between_shards() {
6049 let (ahead, behind) = (Uuid::new_v4(), Uuid::new_v4());
6050 let uuid = Uuid::new_v4();
6051 let details = MemWalIndexDetails {
6052 compacted_sstables: vec![
6053 CompactedSsTable::new(ahead, 10),
6054 CompactedSsTable::new(behind, 10),
6055 ],
6056 index_catchup: vec![IndexCatchupProgress::new(
6057 "idx".to_string(),
6058 vec![
6059 CompactedSsTable::new(ahead, 8),
6060 CompactedSsTable::new(behind, 2),
6061 ],
6062 )],
6063 ..Default::default()
6064 };
6065 let before = vec![user_index("idx", uuid, &[0]), mem_wal_index(details)];
6066 let mut after = before.clone();
6067 apply(&mut after, &before, &[0, 1], &before).unwrap();
6069
6070 let mut coverage = coverage_for(&after, "idx").expect("carried");
6071 coverage.sort_unstable_by_key(|sstable| sstable.shard_id);
6072 let mut expected = vec![
6073 CompactedSsTable::new(ahead, 8),
6074 CompactedSsTable::new(behind, 2),
6075 ];
6076 expected.sort_unstable_by_key(|sstable| sstable.shard_id);
6077 assert_eq!(coverage, expected);
6078 }
6079
6080 #[test]
6086 fn a_bitmap_narrowed_after_the_build_loses_its_position() {
6087 let shard = Uuid::new_v4();
6088 let uuid = Uuid::new_v4();
6089 let mut migrated = table(&[0], uuid, progress_with_catchup(shard, 5, 5));
6092
6093 Transaction::withdraw_coverage_invalidated_after_build(
6094 &mut migrated,
6095 &["idx".to_string()],
6096 3,
6097 )
6098 .unwrap();
6099
6100 assert_eq!(coverage_for(&migrated, "idx"), None);
6101 }
6102
6103 #[test]
6107 fn metadata_migration_that_does_not_narrow_keeps_its_position() {
6108 let shard = Uuid::new_v4();
6109 let uuid = Uuid::new_v4();
6110 let mut migrated = table(&[0, 1], uuid, progress_with_catchup(shard, 5, 5));
6111 migrated[0].files = Some(Vec::new());
6112 migrated[0].created_at = Some(chrono::Utc::now());
6113
6114 Transaction::withdraw_coverage_invalidated_after_build(&mut migrated, &[], 3).unwrap();
6116
6117 assert_eq!(coverage_for(&migrated, "idx"), Some(compacted(shard, 5)));
6118 }
6119
6120 #[test]
6123 fn an_unchanged_commit_does_not_rewrite_the_system_index() {
6124 let shard = Uuid::new_v4();
6125 let uuid = Uuid::new_v4();
6126 let before = table(&[0], uuid, progress_with_catchup(shard, 5, 5));
6127 let mut after = before.clone();
6128 apply(&mut after, &before, &[0], &before).unwrap();
6129
6130 let system_uuid = |indices: &[IndexMetadata]| {
6131 indices
6132 .iter()
6133 .find(|idx| idx.name == MEM_WAL_INDEX_NAME)
6134 .unwrap()
6135 .uuid
6136 };
6137 assert_eq!(system_uuid(&after), system_uuid(&before));
6138 }
6139
6140 #[test]
6144 fn without_a_read_version_a_changed_index_still_loses_its_position() {
6145 let shard = Uuid::new_v4();
6146 let before = table(&[0, 1], Uuid::new_v4(), progress_with_catchup(shard, 5, 5));
6147 let mut after = table(&[0], Uuid::new_v4(), progress_with_catchup(shard, 5, 5));
6148 let segments_before = Transaction::logical_index_segments(&before);
6149 Transaction::apply_mem_wal_index_coverage(&mut after, &segments_before, None, 2)
6150 .unwrap();
6151 assert_eq!(coverage_for(&after, "idx"), None);
6152 }
6153
6154 #[test]
6157 fn the_derivation_is_stable_across_attempts() {
6158 let shard = Uuid::new_v4();
6159 let read = table(&[0, 1], Uuid::new_v4(), progress(shard, 5));
6160 let mut first = table(&[0, 1], Uuid::new_v4(), progress(shard, 5));
6161 let mut second = first.clone();
6162 apply(&mut first, &read, &[0, 1], &read).unwrap();
6163 apply(&mut second, &read, &[0, 1], &read).unwrap();
6164 assert_eq!(coverage_for(&first, "idx"), coverage_for(&second, "idx"));
6165 }
6166 }
6167}