Skip to main content

lance_table/transaction/
manifest_build.rs

1// SPDX-License-Identifier: Apache-2.0
2// SPDX-FileCopyrightText: Copyright The Lance Authors
3
4//! Applying an operation to produce the next manifest.
5//!
6//! [`Transaction::build_manifest`] is the centre of this module and of the
7//! transaction machinery generally: given the current manifest and index list, it
8//! decides the new fragment list, the surviving indices and the next row id, then
9//! assembles the manifest. Everything else in `super` exists to serve it -- the
10//! operation vocabulary it matches on, the index rules it applies, the row version
11//! metadata it stamps, the validation that runs before it.
12
13use 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
60/// Table config key that opts a table into writing clustered deletions as
61/// `Range` segments (`U64Segment::Ranges`). Set it to `true` with
62/// `update_config`.
63///
64/// Manifests written afterwards re-encode bitmap row id segments as runs of
65/// `Range` segments where that is smaller. That is the
66/// existing wire format, so any version reads the result, but a reader without
67/// the compact in-memory form handles thousands of segments per fragment
68/// slowly. That is why it is opt-in rather than the default.
69///
70/// Setting the key back to `false` only stops further conversions. Fragments
71/// already written as `Range` segments keep that encoding until something else
72/// rewrites their row ids (compaction, for example), so coordinate reader
73/// upgrades before enabling it.
74pub 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
83/// Re-encode inline row id sequences as runs of `Range` segments when the
84/// table has opted in.
85///
86/// Fragments that the previous manifest already re-encoded are left alone
87/// (their inline bytes are the very same allocation), so a commit pays only
88/// for the fragments it changed; enabling the key re-encodes every fragment
89/// once.
90fn 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        // Mixed prewritten files cannot imply a default. An explicit default
148        // takes precedence; finalization validates the referenced file versions.
149        let version = match user_requested {
150            Some(ConcreteFileVersion::V1) => {
151                // Preserve legacy creation's homogeneous-file contract.
152                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        // This read bypasses Dataset's feature gates. Refuse unsupported target
182        // manifests before apply_feature_flags can clear their unknown bits and
183        // republish the referenced files as legacy-compatible.
184        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            // Before activation, different fields could share an ID across versions.
195            // Keeping today's high-water mark cannot prevent restoring such a collision.
196            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        // Row ids are a high-water mark like fragment ids: rewinding hands old ids to new rows.
206        manifest.next_row_id = manifest.next_row_id.max(current_manifest.next_row_id);
207        // Turning stable row ids off would revert `_rowid` to row addresses, whose
208        // namespace overlaps the ids this table has already handed out.
209        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    /// Every non-system logical index, mapped to what determines its coverage.
221    ///
222    /// A logical index may be backed by several physical segments, so "did this
223    /// index change" is a question about the whole set. Sorted by UUID so the
224    /// two sides compare positionally.
225    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    /// Apply MemWAL index-coverage rules once the final index list is known.
243    ///
244    /// Coverage records that a base-table index contains the rows a compaction
245    /// copied in, and the WAL pod retires SSTables against it.
246    ///
247    /// It is derived, not reported. An index covering every fragment live at the
248    /// transaction's read version holds every row compaction had copied in by
249    /// then, so it is caught up to that version's `compacted_sstables`. That is
250    /// the only proof available: nothing maps a generation to the fragments its
251    /// rows landed in, so covering the table as the transaction read it is how
252    /// an index shows it covered those rows. Fragments appended since are a
253    /// later gap.
254    ///
255    /// Deriving rather than transmitting means no claim can go stale between
256    /// inspection and commit, the answer survives rebase (`read_version` is
257    /// fixed for a transaction's life), and any operation can earn coverage --
258    /// an ordinary reindex that fully covers no longer has to throw its work
259    /// away and wait for a repair.
260    ///
261    /// Only meaningful once catch-up is required, where a missing entry means
262    /// "not caught up" and the SSTables stay. A legacy table reads a missing
263    /// entry as "fully caught up", so this leaves it untouched rather than
264    /// making the table look more covered than it is.
265    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            // The system index went away with this transaction (MemWAL disable,
276            // or an overwrite). There is no coverage left to maintain.
277            return Ok(());
278        };
279
280        let mut details = load_mem_wal_index_details(final_indices[pos].clone())?;
281
282        // Nothing has ever been compacted, so no index can be behind and there
283        // is no coverage to invalidate.
284        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        // Per shard: what this commit records as compacted, and the most the
292        // read version may credit. Generations compacted after that read landed
293        // in fragments no index under consideration has seen; the committed
294        // value caps it in turn, so a read version since rolled back cannot
295        // retire SSTables no live commit copied in.
296        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        // Every fragment live when the transaction read the table. An index
329        // spanning all of them holds every row compacted by then.
330        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                // Subset-of-empty is trivially true, so this would credit every
345                // index on a table with no fragments. Refused because an empty
346                // fragment list is not only what an emptied table looks like:
347                // it is also what a manifest written before #8438 looks like,
348                // where UpdateMemWalState published no fragments at all. On
349                // such a table the SSTables are the last copy of those rows,
350                // and crediting coverage would retire them. The cost is that a
351                // genuinely emptied table keeps its SSTables.
352                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                    // An unknown bitmap cannot be shown to cover anything.
359                    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            // Compared by [`CoverageIdentity`], not segment UUID: an Update
368            // that touches an indexed field prunes a segment's fragment bitmap
369            // in place while keeping its UUID, so a UUID-only comparison would
370            // carry a position forward that the index no longer earns.
371            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                // Changed, and nothing shows the new index covers the read
379                // version. No entry: a missing one reads as "not caught up".
380                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                    // Takes the better of what this commit proves and what an
391                    // unchanged index already held, so a commit reading an older
392                    // version does not lower a position it cannot re-prove. The
393                    // clamp is the exception: a position above what this commit
394                    // records as compacted describes rows no live commit copied
395                    // in.
396                    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            // The first thing to check when SSTables stop becoming trimmable.
419            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    /// Drop coverage for indices a post-`build_manifest` step narrowed.
431    ///
432    /// The derivation runs while the manifest is being built, but the index list
433    /// is not final there: `migrate_indices` can recalculate a segment's
434    /// fragment bitmap and keep its UUID, so an index can narrow after its
435    /// position was decided. It reports which ones it touched rather than the
436    /// caller re-snapshotting every bitmap to find out. Only ever removes.
437    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    /// Step one of a commit attempt: what the operation settles about the
468    /// index list, computed against the manifest current at the attempt.
469    ///
470    /// Runs the tagged-table write gate (`system_index::frag_reuse::gate`)
471    /// before anything is derived, keeps `current_indices` as the original
472    /// list, and derives the prepared list: on an in-place column rewrite
473    /// (an update with `fields_modified`, a merge swapping a column's data
474    /// file, a data replacement) the coverage it invalidates is withdrawn
475    /// along a tagged fragment reuse history or pruned by fragment id on any
476    /// other table (`withdraw_in_place_rewrites`). Every other operation
477    /// prepares the list unchanged; `Restore` never reaches the builder and
478    /// restores its snapshot's list whole.
479    ///
480    /// `frag_reuse_ledger` is the decoded history of the current entry when
481    /// the caller read it (the commit path decodes it once per attempt for
482    /// the operations above); without it an inline history is decoded from
483    /// the entry and an external one is refused, so a commit never guesses
484    /// at the lineage. A history that does not decode is an error, never
485    /// treated as "no lineage". `frag_reuse` is what the commit path settled
486    /// about the entry itself.
487    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            &current_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                        &current_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    /// Create a new manifest from the current manifest and the transaction.
538    ///
539    /// `current_manifest` should only be None if the dataset does not yet exist.
540    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    /// [`Self::build_manifest`] with the version this transaction read.
557    ///
558    /// Supplied by the commit path, which already materializes that version.
559    /// `None` where there is none to read -- dataset creation and detached
560    /// commits -- in which case no index can be shown to cover it and coverage
561    /// is left as the invalidation rules put it.
562    ///
563    /// Prepares the index list itself with nothing settled about the
564    /// fragment reuse entry and no history read, so it stays safe on its
565    /// own: a tagged entry on a `Rewrite` or a `CreateIndex` touching a
566    /// tagged entry is refused, an inline history is decoded from the entry
567    /// and an external one is refused. The commit path prepares first
568    /// (`prepare_indices`) and builds with `build_manifest_prepared`.
569    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    /// Build the next manifest from a prepared index list
594    /// (`prepare_indices`): the builder applies the operation to the
595    /// prepared list and publishes only the final list; the original list
596    /// is what MemWAL coverage is compared against. The prepared result must
597    /// have been derived against `current_manifest`.
598    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        // The tagged-table write gate, run again at this chokepoint (it is
616        // pure) so a build cannot skip it whatever prepared the list.
617        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        // The fragment-reuse index is internal bookkeeping registered by
637        // compaction with deferred index remap, not something a user can
638        // meaningfully drop and recreate, so it must not gate the migration.
639        // It is dropped from the migrated manifest below, which is what makes
640        // ignoring it here safe. Every other index, including the MemWAL
641        // index, still blocks: only the fragment-reuse index is known to be
642        // discardable.
643        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                // CREATE mode: registering base paths
668                // Base IDs should have been assigned during write operation
669                // Validate uniqueness and insert them into the manifest
670                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                // OVERWRITE mode with initial_bases should have been rejected by validation
681                // This branch should never be reached
682                return Err(Error::invalid_input(
683                    "OVERWRITE mode cannot register new bases. This should have been caught by validation.",
684                ));
685            }
686        }
687
688        // Get the schema and the final fragment list
689        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        // Fragment ids are a high water mark for the whole dataset history: an id
705        // must never name two different sets of rows, or per-fragment state keyed
706        // by id (caches, deletion files, row addresses) can be attributed to the
707        // wrong rows.
708        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        // Release builds refuse to publish a stable-row-id dataset with a fragment
716        // reuse index, and no index needs it.
717        if config.migration_next_row_id.is_some() {
718            final_indices.retain(|idx| idx.name != FRAG_REUSE_INDEX_NAME);
719        }
720
721        // The original list, as read from the current manifest and before
722        // any preparation, so coverage can be compared against what each
723        // logical index looked like going in. Only tables with a MemWAL
724        // index maintain coverage, so every other commit -- and the segment
725        // clones this costs -- pays nothing.
726        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            // Only use row ids if the feature flag is set already, or this is
734            // a migration activation that explicitly provides the next_row_id.
735            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                    // Migration activation: use the provided next_row_id.
743                    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                    // Add version metadata for all new fragments
787                    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                // Remove the deleted fragments
801                // Hash lookups keep this linear on tables with many fragments.
802                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                        // The post-image was built at the transaction's read
810                        // version. Only its deletion file is new; retain all
811                        // other state from the current fragment when rebasing.
812                        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                // Extract existing fragments once for reuse
829                let existing_fragments = maybe_existing_fragments?;
830
831                // Apply updates to existing fragments
832                // Hash lookups keep this linear on tables with many fragments.
833                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                            // Carry forward the fragment's current overlays (which
852                            // may include ones added by a concurrent commit). An
853                            // in-place column rewrite then tombstones the overlaid
854                            // fields it rewrote, since the fresh base values
855                            // supersede them.
856                            updated.overlays = f.overlays.clone();
857                            // A concurrent Project may have pruned files from the
858                            // current fragment after this post-image was staged.
859                            // Spilled lineage fields are absent from the schema,
860                            // but their carriers must survive just as in Project.
861                            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                // Update version metadata for updated fragments if stable row IDs are enabled
881                // Note: We don't update version metadata for fragments with deletion vectors
882                // because the version sequences are indexed by physical row position, not logical position.
883                // Version metadata for deleted rows will be filtered out during scan using the deletion vector.
884                if next_row_id.is_some() {
885                    // Version metadata will be properly set during compaction when deletions are materialized
886                }
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                        // Defense-in-depth: only stamp fragments that were actually
901                        // rewritten. validate_operation enforces this invariant before
902                        // build_manifest is called; this guard catches any path that
903                        // bypasses validation.
904                        if !updated_by_id.contains_key(&fragment.id) {
905                            continue;
906                        }
907                        if bitmap.is_empty() {
908                            continue;
909                        }
910                        // Skip fragments with no existing version metadata: the helper
911                        // would fill unmatched rows with prev_version, fabricating a
912                        // last_updated stamp for rows that never had one.
913                        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                // The fragments whose indexed fields were rewritten were
949                // withdrawn from their indices by `prepare_indices`.
950
951                let mut new_fragments =
952                    Self::fragments_with_ids(new_fragments.clone(), &mut fragment_id)
953                        .collect::<Vec<_>>();
954                // New fragments were staged against the same pre-Project schema.
955                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                // Assign row IDs to any fragments that don't have them yet
965                // (e.g., inserted rows from merge_insert operations)
966                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                    // collect all the original frag ids that contains the updated rows
987                    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                    // The original fragments that carried an overlay: their moved rows may have a
994                    // stale index entry (see `register_pure_rewrite_rows_update_frags_in_indices`).
995                    // Reuse the hash lookups built above instead of scanning
996                    // `original_fragment_ids` per fragment.
997                    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                    // Note: Version metadata is already set above (lines 1627-1755)
1019                    // for Update operations, preserving created_at from original fragments.
1020                    // Don't overwrite it here.
1021                }
1022                // Identify fragments that were updated or newly created in this update
1023                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                // Every fragment in an overwrite is newly written, so all of them
1038                // take fresh ids regardless of the id they arrive with. Fragments
1039                // carried over from the dataset being replaced are rejected by
1040                // `validate_operation`, which is what makes ignoring the incoming
1041                // id safe here.
1042                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                    // Add version metadata for all new fragments
1050                    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                // Groups covered by the stable-partition transitions
1075                // redistribute rows across their destinations, so index
1076                // bitmaps must not follow them: the retired source ids stay
1077                // in the bitmaps as provenance and the tagged fragment reuse
1078                // entry records the row-level translation. Only the
1079                // order-preserving groups take part in bitmap maintenance
1080                // below.
1081                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                    // We can re-use indices, but need to rewrite the fragment bitmaps.
1092                    //
1093                    // An address-domain index the commit path remapped eagerly (no FRI,
1094                    // no deferral) is already fully migrated here: its uuid, files and
1095                    // bitmap were rewritten in place by `handle_rewrite_indices`, below,
1096                    // exactly as a non-stable-row-id rewrite migrates one. The per-index
1097                    // loop that follows must not touch those again, so it skips any index
1098                    // whose uuid is one `handle_rewrite_indices` just produced.
1099                    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                    // If there is an FRI and the index uses addresses then we can
1109                    // migrate fragment support.
1110                    //
1111                    // If there is an FRI and the index uses row ids (this is getting phased
1112                    // out) then we can migrate fragment support without need for remap.
1113                    let deferred_remap = frag_reuse_index.is_some();
1114                    for index in final_indices.iter_mut() {
1115                        // Its bitmap is lineage, not coverage, and a straddling
1116                        // group would fail the recalculation.
1117                        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                                // Stable row ids survive a rewrite, so a row-id-domain index
1126                                // can simply follow its data to the new fragments. An
1127                                // address-domain index this commit did not remap above
1128                                // cannot: its stored addresses point into the fragments the
1129                                // rewrite dropped. Claiming coverage of the new fragments
1130                                // would make it answer queries with addresses that no longer
1131                                // resolve, so drop the rewritten fragments from its coverage
1132                                // instead and let the scanner fall back to a full scan for
1133                                // them.
1134                                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                // A full compaction materializes a fragment's overlays into fresh
1149                // base data. Any index older than one of those overlays was built on
1150                // the pre-overlay values, so drop the rewritten fragment from its
1151                // coverage to keep it from serving stale values.
1152                Self::prune_overlay_stale_fields_from_indices(&mut final_indices, groups);
1153
1154                match (frag_reuse_index, assembly) {
1155                    // Tagged history: splice the entry the commit path
1156                    // assembled against the current manifest, never the
1157                    // operation's own entry (the caller's intent, which the
1158                    // assembly merged and validated).
1159                    (_, Some(assembly)) => {
1160                        // A consistency check on the assembly, not the
1161                        // concurrency protection. Concurrent rewrites are
1162                        // handled by the commit path, which re-assembles the
1163                        // entry against the latest manifest on every attempt
1164                        // and commits with compare-and-swap, so an assembly
1165                        // built on a stale entry never reaches this splice
1166                        // through that path. What this refuses is an
1167                        // assembled entry whose recorded base is not the
1168                        // entry in the manifest being built (an entry
1169                        // assembled by hand, or carried over from another
1170                        // base): splicing it would replace the transitions
1171                        // that entry already holds.
1172                        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                    // v0 snapshot: splice verbatim, byte-identical to the
1191                    // historical behavior. The chokepoint above already
1192                    // rejected tagged snapshots.
1193                    (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                // Check the final set, including retained siblings, on every
1224                // commit attempt. A format migration may change the result
1225                // domain without changing the index type.
1226                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                // For each previous id, the first occurrence in the list is the
1260                // merged existing fragment; everything else is new (staged
1261                // write_fragments output arrives with placeholder id 0, colliding
1262                // with real fragment 0). merge_fragments_valid enforces this rule.
1263                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                // New fragments get Append's treatment: id assignment (0 means
1272                // unassigned, non-zero pre-reserved) and, with stable row ids,
1273                // row ids plus fresh version metadata.
1274                for (fragment, is_new) in merged_fragments.iter_mut().zip(is_new.iter()) {
1275                    if !*is_new {
1276                        continue;
1277                    }
1278                    // Caller-supplied row ids are never checked against live
1279                    // fragments or next_row_id, so keeping them could commit
1280                    // duplicate or unallocated stable row ids.
1281                    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                // A Merge can rewrite a column's data file in place; the field stays
1317                // in the schema, so the index is retained. Its now-stale coverage of
1318                // the rewritten fragments was withdrawn by `prepare_indices`.
1319
1320                // Some fields that have indices may have been removed, so we should
1321                // remove those indices as well.
1322                Self::retain_relevant_indices(&mut final_indices, &schema, &final_fragments)
1323            }
1324            Operation::Project { .. } => {
1325                final_fragments.extend(maybe_existing_fragments?.clone());
1326
1327                // We might have removed all fields for certain data files, so
1328                // we should remove the data files that are no longer relevant.
1329                // A file carrying a spilled row lineage sequence stays: its
1330                // reserved ids are never in the schema, and it is the only copy.
1331                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                // Some fields that have indices may have been removed, so we should
1346                // remove those indices as well.
1347                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                // 1. make sure the new files all have the same fields / or empty
1363                // NOTE: arguably this requirement could be relaxed in the future
1364                // for the sake of simplicity, we require the new files to have the same fields
1365                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                // Collect replaced field IDs before consuming new_datafiles
1388                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                // 2. check that the fragments being modified have isomorphic layouts along the columns being replaced
1402                // 3. add modified fragments to final_fragments
1403                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                    // Physical mappings differ across V2 encodings (a nested
1415                    // parent may have no column of its own). Replacement and
1416                    // tombstoning operate on the logical field coverage.
1417                    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                    // TODO(rmeng): check new file and fragment are the same length
1425
1426                    let mut columns_covered = HashSet::new();
1427                    // Set when an existing file covers exactly the replaced
1428                    // fields, so the whole file swaps rather than part of it.
1429                    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                            // assign the new file path / size / base to the fragment
1436                            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                    // Reject a file whose version does not decode before any
1445                    // arm publishes it.
1446                    new_file.file_version()?;
1447
1448                    // SPECIAL CASE: if the column(s) being replaced are not covered by the fragment
1449                    // Then it means it's a all-NULL column that is being replaced with real data
1450                    // just add it to the final fragments. Push the DataFile as
1451                    // given so every field (including base_id) is preserved.
1452                    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                        // Tombstone the replaced fields where they live and
1466                        // append the new file to answer for them, the idiom
1467                        // `update_columns` uses. Compaction decides that layout,
1468                        // so the fields may sit in one wider file or span
1469                        // several.
1470                        //
1471                        // Legacy V1 is excluded: its reader derives the page table
1472                        // offset from the first field in the metadata, so
1473                        // tombstoning one field leaves its siblings decoding from
1474                        // the wrong pages. A field a V1 file covers keeps
1475                        // exact-match replacement.
1476                        for file in &mut new_frag.files {
1477                            // Same reason as the guard above.
1478                            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                        // Every data file must share at least one field with
1495                        // the dataset schema: a file kept alive only by
1496                        // tombstones or by ids the schema no longer defines is
1497                        // unreachable to readers, uncollectable by cleanup,
1498                        // and reported corrupt by validate(). The exception is
1499                        // a file carrying one of the fragment's spilled row
1500                        // lineage sequences: their reserved ids are never in
1501                        // the schema, the file holds their only copy, and the
1502                        // replaced user columns in it are dead space until
1503                        // compaction.
1504                        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                    // Nothing changed in the current fragment, which is not expected -- error out
1518                    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                    // New base values supersede any overlay still shadowing
1525                    // them, so tombstone the overlaid fields. An overlay
1526                    // committed after this transaction's snapshot is the newer
1527                    // value though -- the conflict resolver rebases these two
1528                    // precisely because the overlay wins -- so it stays, and
1529                    // being newer it stays last, preserving the ordering.
1530                    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                // 4. push fragments that didn't change back to final_fragments
1551                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                // A replacement changes what its rows read as, so stamp them
1560                // updated. Without this, get_updated_rows never reports them and
1561                // an incremental consumer skips them for good.
1562                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                // The replaced fields' coverage of the modified fragments was
1576                // withdrawn by `prepare_indices`.
1577            }
1578            Operation::DataOverlay { groups } => {
1579                // Stamp each overlay with the version this commit is producing.
1580                // build_manifest re-runs on every retry with an updated
1581                // current_manifest, so this is naturally re-stamped on retry.
1582                let new_version = current_manifest.map_or(1, |m| m.version + 1);
1583
1584                let existing_fragments = maybe_existing_fragments?;
1585                // Multiple groups may target the same fragment; merge them in
1586                // order rather than letting a HashMap collapse drop all but the
1587                // last group's overlays.
1588                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                // Every group must target an existing fragment. Build a set of
1597                // existing ids once so this is O(groups + fragments) rather than
1598                // O(groups * fragments).
1599                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                                // Scans interpret missing lineage metadata as version 1,
1632                                // so using 1 as the fallback preserves the observable
1633                                // version of every uncovered row.
1634                                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                        // Appended (not replaced) so concurrently-written overlays
1644                        // survive; later entries are newer.
1645                        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                // Updates the MemWAL index only; the fragments are unchanged.
1660                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                // UpdateBases operation doesn't modify fragments or indices
1669                // Base paths are handled in the manifest creation section below
1670                final_fragments.extend(maybe_existing_fragments?.clone());
1671            }
1672        };
1673
1674        // If a fragment was reserved then it may not belong at the end of the fragments list.
1675        final_fragments.sort_by_key(|frag| frag.id);
1676
1677        // Clean up data files that only contain tombstoned fields
1678        Self::remove_tombstoned_data_files(&mut final_fragments);
1679
1680        // Enforce the newest-last overlay ordering invariant at the write
1681        // boundary. Load normalizes with a sort; this rejects any commit path
1682        // that assembled a fragment's overlays out of order.
1683        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        // A spilled row lineage sequence lives only in its carrier file. An
1690        // operation that dropped or duplicated that file would otherwise go
1691        // unnoticed until the next read, by which time cleanup may have
1692        // deleted the only copy. Every fragment is checked, not only the ones
1693        // this operation touched: the check reads metadata only, skips
1694        // fragments that spill nothing, and also refuses to build on a
1695        // manifest that already lost a carrier.
1696        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        // Applied once the final index list is known, so it sees exactly the
1708        // indices this commit publishes rather than what any one operation arm
1709        // intended.
1710        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        // Blob is one logical column across file versions. Publishing its first
1720        // 2.2+ file also publishes the logical v2 view, without replacing older files.
1721        // Blob children are not independent physical columns; retain the root ID
1722        // and allocate logical child IDs against the current manifest on commit.
1723        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            // OVERWRITE with initial_bases on existing dataset is not allowed (caught by validation)
1761            // So we always use new_from_previous which preserves base_paths
1762            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                // If this is an overwrite operation and the user has requested a specific version
1769                // then overwrite with that version.  Otherwise, if the user didn't request a specific
1770                // version, then overwrite with whatever version we had before.
1771                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            // Internal operations (e.g. CreateIndex) build with the default config,
1800            // which has use_stable_row_ids = false. Without inheriting from the previous
1801            // manifest, apply_feature_flags would clear FLAG_STABLE_ROW_IDS.
1802            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        // Set after apply_feature_flags, which resets both flag words -- and a
1813        // `Manifest` only points at its index section, so the flag cannot be
1814        // derived there.
1815        //
1816        // Derived fresh from `final_indices` on every commit, never inherited.
1817        // Every manifest this reaches starts without the covering bit, so there
1818        // is no stale bit to clear. Dropping the last covering index lifts the
1819        // fence by simply not setting it again. Inheriting it from the previous
1820        // manifest instead would make the fence permanent.
1821        //
1822        // Both words: a reader that selects a vector index by membership of
1823        // `fields` would answer a query on a merely-carried column with an index
1824        // keyed on another one, and a writer that treats every entry of `fields`
1825        // as keyed would mismaintain it.
1826        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                // The unenforced primary and clustering keys are reserved
1872                // schema properties: each is immutable once set, and its
1873                // reserved metadata keys cannot be written with an invalid
1874                // value. Capture the prior keys, and whether this transaction
1875                // writes a reserved key, before applying the updates so
1876                // violations can be rejected below. This runs on every apply,
1877                // including conflict-rebase, so it also rejects the
1878                // concurrent-writer race.
1879                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                        // Also set unenforced primary key based on updated field metadata.
1907                        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                        // Also set unenforced clustering key based on updated
1919                        // field metadata.
1920                        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                    // The primary key is already set: reject any change to it,
1938                    // and any write that touches a reserved primary key.
1939                    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                    // A reserved primary key was written but did not install a
1946                    // valid primary key (e.g. a non-marker flag value or a
1947                    // non-numeric position).
1948                    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                    // Installing by field metadata skips the Arrow-schema
1954                    // conversion that would otherwise validate the key.
1955                    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                    // The clustering key is already set: reject any change to
1965                    // it, and any write that touches the reserved key.
1966                    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                    // The reserved clustering key was written but did not
1973                    // install a valid clustering key (e.g. a non-numeric
1974                    // position value).
1975                    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        // Handle UpdateBases operation to update manifest base_paths
1986        if let Operation::UpdateBases { new_bases } = &self.operation {
1987            // Validate and add new base paths to the manifest
1988            for new_base in new_bases {
1989                // Check for conflicts with existing base paths
1990                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                // Assign a new ID if not already assigned
2002                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    /// Remove data files that only contain tombstoned fields (-2)
2049    /// These files no longer contain any live data and can be safely dropped
2050    fn remove_tombstoned_data_files(fragments: &mut [Fragment]) {
2051        for fragment in fragments {
2052            fragment.files.retain(|file| {
2053                // Keep file if it has at least one non-tombstoned field
2054                file.fields.iter().any(|&field_id| field_id != -2)
2055            });
2056        }
2057    }
2058
2059    /// Remove overlays that no longer supply any field in the schema.
2060    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    /// Coverage of an index that a rewrite invalidates: the rewritten fragments are
2073    /// removed and the fragments they became are *not* added.
2074    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    /// Fragment 0 is the live destination of a rewrite that retired
2112    /// fragment 5; the entry carries that one transition inline.
2113    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    /// Deletes, row-rewriting updates, config updates, projections and merges
2139    /// that rewrite no column touch deletion vectors, new fragments, the
2140    /// schema or the manifest config only: the tagged entry is carried
2141    /// through unchanged and the mutation commits.
2142    #[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    /// An overwrite replaces every fragment and drops every index, the tagged
2180    /// entry included; the sticky flag keeps the tagged record form.
2181    #[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    /// The stable row id migration commits a `Merge` with the activation
2205    /// marker; it is refused while a tagged entry exists.
2206    #[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    /// The manifest's only fragment, 0, carries field 0 in `path`.
2233    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    /// A user segment on field 0 built from fragment 5, which the rewrite
2248    /// that produced fragment 0 retired: the entry's history says so, and
2249    /// the segment covers fragment 0 only through it.
2250    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    /// The three in-place column rewrites of field 0 on fragment 0.
2267    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    /// An in-place rewrite of a column that a translating segment indexes,
2312    /// on a fragment the segment reaches only through the history, is
2313    /// admitted; the segment cannot withdraw that fragment from its bitmap,
2314    /// so the whole transition's sources are withdrawn instead and the
2315    /// segment claims nothing for its destinations (the reader scans them).
2316    #[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    /// Naming the fragment directly is no way out: a merged segment's bitmap
2348    /// is the union of its sources' provenance, and pruning the direct id
2349    /// alone would let the retired sources re-derive the same destination
2350    /// with the pre-rewrite values inside. Both go.
2351    #[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    /// Without a walkable history (a ledger this writer cannot read, or none
2380    /// at hand) the withdrawal is conservative: every retired fragment of a
2381    /// segment indexing the rewritten column goes, on the rewritten lineage
2382    /// or not, and only the live fragments stay. The same rewrite with the
2383    /// history walkable withdraws just the transition reaching the rewritten
2384    /// fragment.
2385    #[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    /// What a withdrawal leaves, per logical index name: a segment emptied
2426    /// while a same-name sibling still has coverage leaves the manifest in
2427    /// the commit; the last segment of a name stays, empty, as the record of
2428    /// what to rebuild; a segment without a bitmap has unknown coverage and
2429    /// is neither withdrawn from nor counted as empty.
2430    #[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        // Beside a serving sibling the emptied segment leaves the manifest.
2463        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        // As the last segment of its name it stays, empty.
2504        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    /// A segment on an unrelated lineage (fragment 6 rewritten into 8) is
2520    /// untouched by a rewrite of fragment 0: the history is walked, not
2521    /// guessed at.
2522    #[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    /// Without a supplied ledger an entry whose history is not inline cannot
2545    /// be walked, and the build refuses rather than guess.
2546    #[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    /// A segment whose bitmap is entirely live has the fragment pruned as on
2569    /// an untagged table; a translating segment that does not index the
2570    /// rewritten field, or whose rows the history moves elsewhere, is
2571    /// untouched. The entry is carried through in every case.
2572    #[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            // Fragment 5 was rewritten into 7, not into the rewritten 0.
2590            "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            // Fragment 0 was rewritten: withdrawn from the segment.
2614            RoaringBitmap::new()
2615        } else {
2616            segment.fragment_bitmap.unwrap()
2617        };
2618        assert_eq!(user.fragment_bitmap.as_ref().unwrap(), &expected, "{shape}");
2619    }
2620
2621    /// A user index is created, replaced or dropped on a tagged table like on
2622    /// any other: the entry is carried through untouched.
2623    #[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    /// Recording MemWAL compaction progress on a tagged table replaces the
2649    /// MemWAL entry and carries the history and the user segment through
2650    /// unchanged.
2651    #[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            // A rewrite carrying no entry, or a v0 entry, would splice away
2701            // the tagged history; only a tagged entry may replace one. The
2702            // bare rewrite touches fragment 0, which the entry's bitmap
2703            // covers (a bare rewrite of only uncovered fragments is allowed,
2704            // see `tagged_history_allows_rewrite_of_uncovered_fragments`).
2705            "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        // A pre-assembled tagged entry that did not come from the commit
2750        // path's assembly (a `ReplaceEntry` snapshot) has bypassed
2751        // validation and may be stale; it must be rejected at the manifest
2752        // chokepoint regardless of the table's current state.
2753        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        // The sticky flag alone forbids a v0 snapshot: a trim may have
2790        // deleted the fully drained tagged entry, but the flag survives and
2791        // remains the final authority on the record form. Publishing the
2792        // snapshot would silently downgrade the table to v0.
2793        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        // Deferred compaction of never-covered fragments commits a plain
2819        // rewrite (no entry, no intent): it cannot invalidate provenance or
2820        // the recorded lineage, so the tagged gate admits it.
2821        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        // The tagged entry rides through untouched.
2851        assert!(final_indices.iter().any(|idx| idx.uuid == fri.uuid));
2852    }
2853
2854    #[test]
2855    fn out_of_range_transition_source_id_rejected() {
2856        // Fragment ids in the reuse domain are bounded by the row-address
2857        // fragment space; an oversized id must error instead of silently
2858        // truncating into an alias of another fragment.
2859        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        // The assembled entry replaced the previous one by name.
2916        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    /// A tagged-table fixture for the CreateIndex gate: the manifest carries
2929    /// the FRI flags and one tagged entry.
2930    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    /// Build with the trim intent the maintenance path settles
2941    /// (`FragReuseUpdate::Trim`), through the prepared entry.
2942    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    /// A CreateIndex replacing a tagged entry without the maintenance path's
2959    /// derivation intent is a hand-built (possibly stale) trim: rejected.
2960    #[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    /// With the intent, exactly two shapes pass: the entry replaced, or the
2990    /// entry removed outright (everything trimmed away; flags stay sticky).
2991    #[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        // Sticky bits survive even full removal.
3021        assert_ne!(
3022            new_manifest.reader_feature_flags & FLAG_FRAGMENT_REUSE_INDEX,
3023            0
3024        );
3025    }
3026
3027    /// Even with the intent, a trim whose removed identity no longer matches
3028    /// the current entry was derived against a stale version: rejected, so a
3029    /// replay cannot splice away a concurrent append.
3030    #[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    /// A trim mixing user indices into the same commit is not a shape the
3052    /// maintenance path produces; rejected even with the intent.
3053    #[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        // Assembled against a different entry (base None while the manifest
3085        // holds version 7): splicing would replace that entry's transitions.
3086        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    /// The preparation keeps the original list as read and withdraws on the
3117    /// prepared list only; the build publishes exactly the prepared list.
3118    #[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        // A segment on the unrelated F6 -> F8 lineage is untouched.
3126        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    /// An ordinary table prepares the same way, pruning by fragment id.
3177    #[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    /// Operations that rewrite no column in place, `Restore` included,
3209    /// prepare the list unchanged.
3210    #[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    /// A history stored externally is walked only when the commit path
3239    /// supplied the decoded ledger; without it the preparation refuses.
3240    #[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    /// Corrupt inline details are an error of their own kind, never
3280    /// "no lineage" and never a successful build.
3281    #[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    /// A result prepared against one manifest cannot build on another:
3318    /// every attempt prepares against the manifest it builds on.
3319    #[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    /// The low-level entry cannot skip what the commit path settles: a
3347    /// tagged rewrite entry is refused without the assembled update, whoever
3348    /// prepared the list.
3349    #[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                // Fragment 99 does not exist in the dataset; it must be ignored,
3579                // not appended.
3580                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        // Model an update staged before a projection removed field 1's file.
3638        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            // Separate carriers ensure each lineage kind is retained even when
3674            // the file has no field from the live schema.
3675            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        // Model a delete staged before a projection removed field 1's file.
3745        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        // Create a fragment with mixed data files: some normal, some fully tombstoned
3793        let mut fragment = Fragment::new(1);
3794
3795        // Add a normal data file with valid field IDs
3796        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        // Add a data file with all fields tombstoned
3807        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        // Add a data file with mixed tombstoned and valid fields
3818        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        // Add another fully tombstoned file
3829        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        // Apply the cleanup
3842        Transaction::remove_tombstoned_data_files(&mut fragments);
3843
3844        // Should have removed the two fully tombstoned files
3845        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    /// When a fragment has no existing last_updated_at_version_meta (None), a
3851    /// partial RewriteColumns refresh must leave it as None rather than fabricating
3852    /// prev_version for unmatched rows.
3853    #[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        // Simulate a RewriteColumns update that matched offsets 1 and 3
3881        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        // Bitmap with 10 offsets but fragment only has 5 physical rows.
3939        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        // Only 2 offsets (within cardinality) but max offset 100 exceeds physical_rows 5.
3996        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        // All 5 offsets on a 5-row fragment — exactly at the boundary, should succeed.
4053        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        // Fragment A is being rewritten; fragment B exists in the manifest but is
4077        // NOT in updated_fragments. Supplying an offset key for B must be rejected
4078        // so that B's version metadata cannot be stamped by an unrelated commit.
4079        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        // updated_fragments contains only A; offsets are keyed to B — must fail.
4106        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        // Field 9 must be empty; field 10 must be populated.
4151        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        // Simulate a manifest written by old Lance: only field 9, no field 10.
4178        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        // When both fields present, field 10 wins.
4222        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                    // Field 9 has different values than field 10.
4243                    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    /// Committed fragment carrying `row_ids`.
4431    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    /// Fragment without row id metadata: staged write_fragments output, or a
4447    /// committed fragment of a non-stable dataset.
4448    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    /// Single-column manifest over `existing`; stable datasets get row id
4462    /// flags and `next_row_id` = 100.
4463    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    /// Validate and build a Merge of `fragments` against `manifest`.
4482    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        // A placeholder id 0 gets a fresh fragment id; a pre-reserved id is
4507        // kept. Either way the staged fragment's row ids come from
4508        // next_row_id at commit time.
4509        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        // Existing fragments keep their row id sequences byte-identical.
4524        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        // The staged fragment was allocated row ids from next_row_id.
4530        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        // Version metadata is stamped like Append.
4539        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        // Regression for the duplicate fragment id corruption: a staged
4546        // fragment with placeholder id 0 must not be committed as-is.
4547        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        // The staged fragment's placeholder id 0 collides with fragment 0;
4565        // it must not withdraw fragment 0 from index coverage.
4566        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        // Supplied row ids duplicate the existing fragment's live ids; keeping
4585        // them (or filling a partial prefix) would corrupt the row id index.
4586        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        // With equal row counts and no row id metadata, a misordered staged
4635        // fragment is indistinguishable by content from the fragment whose id
4636        // it collides with; the order check must reject it on non-stable
4637        // datasets too.
4638        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        // A staged id-0 fragment listed before the real fragment 0 would be
4662        // taken as the existing one; validation must reject the swap.
4663        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        // Existing fragment (id=0) with stable row IDs
4768        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        // New fragment (id=1) not present in prev manifest; row ids are
4791        // allocated at commit time.
4792        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        // last_updated_at_version must be set to the commit version
4828        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        // created_at_version must also be set — must not be None
4838        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        // Overlays targeting two distinct fragments are each applied and stamped.
4851        // A targeted fragment already carrying an overlay (committed at v3) gets
4852        // the new overlay appended and stamped while its existing overlay is
4853        // preserved, and a fragment the operation does not target is passed
4854        // through with its existing overlays untouched.
4855        let mut frag0 = Fragment::new(0);
4856        frag0.overlays = vec![overlay_with_field(5, 3)]; // targeted, pre-existing at v3
4857        let frag1 = Fragment::new(1);
4858        let mut frag2 = Fragment::new(2);
4859        frag2.overlays = vec![overlay_with_field(9, 3)]; // untargeted, committed at v3
4860        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        // The pre-existing overlays were committed at v3, so the current
4868        // manifest must be at least that version; the new commit then stamps
4869        // its overlay at v4, keeping the fragment's overlays newest-last.
4870        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        // The already-overlaid target keeps its v3 overlay and appends the new
4901        // one, stamped to the new version.
4902        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        // The fresh target gets its overlay, stamped to the new version.
4906        assert_eq!(frag(1).overlays.len(), 1);
4907        assert_eq!(frag(1).overlays[0].committed_version, result.version);
4908        // The untargeted fragment is unchanged: same overlay, original version.
4909        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        // Stable-row-id migrations can leave old fragments without explicit
4932        // lineage metadata; scans expose those rows as version 1.
4933        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        // A DataReplacement writing new base values for field 5 must stop any
5009        // overlay already shadowing those cells: field 5 is tombstoned in place
5010        // (preserving the overlay's field 3), and an overlay covering only field
5011        // 5 is dropped entirely. Both overlays predate the transaction's read
5012        // version, which is what makes the replacement the newer value.
5013        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        // The base data file for field 5 was swapped in.
5059        assert!(frag.files.iter().any(|f| f.path == "f5-new.lance"));
5060        // The [3, 5] overlay keeps field 3 and tombstones field 5; the [5]-only
5061        // overlay is dropped.
5062        assert_eq!(frag.overlays.len(), 1);
5063        assert_eq!(frag.overlays[0].data_file.fields.as_ref(), &[3, -2]);
5064    }
5065
5066    /// Replace `fields` in `fragment` at `read_version`, against a manifest
5067    /// at `manifest_version` whose schema declares field ids 3 ("x"), 4 ("a"),
5068    /// 5 ("v") and 6 ("y").
5069    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    /// Replace field 5 in `fragment` at `read_version`, against a manifest at
5117    /// `manifest_version`.
5118    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        // The V1 reader derives its page table offset from the first field in
5129        // the file metadata, so turning `[4, 5]` into `[-2, 5]` would leave
5130        // field 4 decoding from field 5's pages. With no exact match to swap,
5131        // the replacement must be rejected rather than corrupting the sibling.
5132        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        // The replaced fields sit in two different wider files. Each file is
5150        // tombstoned for the field it holds and survives on its remaining
5151        // live one, with the new file answering for both.
5152        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        // Spanning is only resolvable while every covering file can be
5188        // tombstoned. A V1 file holding one of the replaced fields cannot,
5189        // so the replacement must be rejected rather than half applied.
5190        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        // A wider file carrying a tombstone from an earlier round is
5214        // tombstoned again for the newly replaced field and survives on its
5215        // remaining live field.
5216        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        // An overlay committed after this transaction read its snapshot holds
5242        // the newer value; the conflict resolver rebases the two precisely
5243        // because the overlay wins. Tombstoning it would discard a committed
5244        // write, so only overlays the transaction could have seen are superseded.
5245        let mut fragment = Fragment::new(0);
5246        // One wider file, so the replacement takes the tombstone-and-append path.
5247        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        // Staged against version 6, i.e. before the overlay landed.
5269        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    /// A projection keeps only the data files that still share a field with
5280    /// the schema. The reserved ids of spilled row lineage are never in it, so
5281    /// without an exception the file carrying them, their only copy, would go.
5282    #[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        // Drop field 1, as `drop_columns(["b"])` would.
5320        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    /// Replacing part of a wider file tombstones the replaced fields where
5352    /// they live and drops any file left without a schema field. The file
5353    /// carrying the fragment's spilled row ids may have none, yet it is their
5354    /// only copy and must stay next to the new file that answers for the user
5355    /// fields. The carrier is either a file of its own, as an update writes
5356    /// it, or a file that also holds user columns, all of which are replaced.
5357    #[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    /// Whatever hands the commit a fragment whose spilled row ids have no v2
5426    /// carrier -- an operation that dropped the file, or a writer that never
5427    /// attached it -- the commit refuses it rather than publish lineage that
5428    /// no reader can load.
5429    #[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        // Column indices do not matter to the commit, only ids and version.
5441        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        // Two groups targeting the same fragment must both survive (a HashMap
5467        // collapse would have dropped the first).
5468        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        // A writer that predates the field encodes nothing, which decodes as
5518        // false: no assertion, so a legacy tightening or required-field merge
5519        // still conflicts. Only an explicit true skips the barrier.
5520        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        /// A manifest carrying exactly `frags`, standing in for the version a
5601        /// transaction read.
5602        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        /// Drives the production path, so these exercise the real derivation.
5614        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        /// Through the builder, the "before" side of the coverage comparison
5656        /// is the original list as read from the manifest, never the
5657        /// prepared one: an in-place rewrite that narrows an index's
5658        /// coverage in preparation withdraws its catch-up record instead of
5659        /// carrying it over as if nothing changed.
5660        #[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            // A schema with field 0, so the index on it is retained.
5666            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        /// An index spanning every fragment the transaction read is credited
5706        /// with what that version had compacted.
5707        #[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        /// An index short of the read version proves nothing, so it gets no
5717        /// entry -- absence reads as "not caught up".
5718        #[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        /// The hazard that makes the comparison use whole metadata.
5728        ///
5729        /// `Operation::Update` prunes a segment's fragment bitmap in place when
5730        /// it touches an indexed field, keeping the same UUID. A UUID-only
5731        /// "unchanged" test carries the old position forward while the index
5732        /// covers fewer fragments, and the WAL pod then trims on a position the
5733        /// index no longer earns. Reachable from the ordinary SSTable merge.
5734        #[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            // Same UUID, fragment 1 pruned away.
5740            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        /// Carrying a position forward is not the same as extending it. An
5750        /// index that has not moved still only holds the generations it caught
5751        /// up to; the compaction that has landed since is in fragments it does
5752        /// not span.
5753        #[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            // Recorded at generation 2; generation 5 has since been folded in.
5758            let before = table(&[0], uuid, progress_with_catchup(shard, 5, 2));
5759            let mut after = before.clone();
5760            // Fragment 1 arrived with that compaction and this index lacks it.
5761            apply(&mut after, &before, &[0, 1], &before).unwrap();
5762            assert_eq!(coverage_for(&after, "idx"), Some(compacted(shard, 2)));
5763        }
5764
5765        /// A recorded position above what this commit says was compacted is
5766        /// clamped down. Nothing should produce one, but a position the base
5767        /// table cannot back would retire SSTables whose rows are nowhere.
5768        #[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        /// An unchanged index keeps what it recorded even when this commit's
5779        /// own snapshot cannot prove as much.
5780        #[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        /// Credit never exceeds what this commit records as compacted, so a
5791        /// read version since rolled back cannot retire SSTables no live commit
5792        /// copied in.
5793        #[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        /// The cap is the read version's progress, not this commit's. A
5803        /// compaction that landed while the index was being built put its rows
5804        /// in fragments this transaction never inspected, so covering
5805        /// everything it *did* read earns only what had been folded in by then.
5806        #[test]
5807        fn credit_never_reaches_past_the_read_version() {
5808            let shard = Uuid::new_v4();
5809            // Read at generation 2; generation 5 landed while this ran.
5810            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        /// One segment with an unknown bitmap makes the whole index unproven,
5817        /// even when its siblings happen to span everything. Coverage that
5818        /// cannot be read is not coverage that can be relied on.
5819        #[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        /// A dropped index has no coverage left to gate anything.
5835        #[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        /// An index created by this commit is credited if it spans the read
5845        /// version -- it was built over those fragments, so it holds their
5846        /// rows. This is what the advance model could not express: an ordinary
5847        /// build that fully covers had to throw its work away and wait.
5848        #[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            // Covers the read version, but was not there when it was read.
5854            apply(&mut after, &before, &[0], &before).unwrap();
5855            assert_eq!(coverage_for(&after, "idx"), Some(compacted(shard, 5)));
5856        }
5857
5858        /// A table carrying compaction progress but no catch-up entry earns one
5859        /// from an ordinary commit. This is how a table written before catch-up
5860        /// was maintained heals itself: nothing has to be run against it.
5861        #[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        /// Two shards, only one of them compacted.
5871        #[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        /// Two indexes advance independently: one covering, one behind.
5903        #[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        /// An index whose coverage is unknown cannot be shown to cover anything.
5918        #[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        /// Nothing compacted means nothing to be behind on.
5930        #[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        /// No MemWAL system index: nothing to maintain, and no error.
5940        #[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        /// No read version -- dataset creation, detached commits -- credits
5950        /// nothing and lowers nothing.
5951        #[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        /// An untrained index covers nothing that exists, so a sibling's work
5964        /// is no evidence for it.
5965        #[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        /// Shards move independently within one index: one advances on this
5980        /// commit's proof while another keeps the position it already had.
5981        #[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            // The quiet shard was never compacted as of the read, so nothing
5997            // this commit proves reaches it -- it keeps its recorded 7.
5998            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        /// The derivation drops coverage an index no longer earns, but it never
6019        /// rejects the commit -- an ordinary index job must not be blocked by
6020        /// a protocol it knows nothing about.
6021        #[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            // Rebuilt over a subset -- the shape a partial reindex leaves.
6026            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        /// A reader's rule is that a missing entry means "not caught up", so an
6032        /// index caught up to nothing must be absent rather than present at
6033        /// generation zero -- otherwise it reads as known-and-covered.
6034        #[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            // Does not span the read version, so nothing lifts it off zero.
6041            apply(&mut after, &before, &[0, 1], &before).unwrap();
6042            assert_eq!(coverage_for(&after, "idx"), None);
6043        }
6044
6045        /// Each shard carries its own position. Collapsing them to one value
6046        /// would credit a lagging shard with a busier shard's progress.
6047        #[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            // Unchanged and unproven: both shards keep exactly what they had.
6068            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        /// The derivation runs while the manifest is being built, but the
6081        /// index list is not final there: `migrate_indices` recalculates a
6082        /// segment's fragment bitmap and keeps its UUID. A position decided
6083        /// before that must not survive the narrowing, or the WAL pod trims
6084        /// against an index that no longer covers those rows.
6085        #[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            // What migrate_indices leaves behind: same UUID, fewer fragments,
6090            // and it says so.
6091            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        /// Migration routinely fills in file lists and inferred details. Those
6104        /// do not change which rows an index answers for, so withdrawing on
6105        /// them would drop coverage every commit for no reason.
6106        #[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            // Nothing narrowed, so migration reports nothing.
6115            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        /// A commit that changes nothing must not churn the system index: a new
6121        /// UUID on every append would invalidate its cache entry fleet-wide.
6122        #[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        /// A commit with no read version still withdraws. It can prove nothing,
6141        /// so an index it changed keeps no position -- the alternative leaves a
6142        /// position describing an index that no longer exists.
6143        #[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        /// Two attempts against the same read version agree, which is what makes
6155        /// a rebase safe: `read_version` is fixed for a transaction's life.
6156        #[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}