lance-table 13.0.0

// SPDX-License-Identifier: Apache-2.0
// SPDX-FileCopyrightText: Copyright The Lance Authors

//! Applying an operation to produce the next manifest.
//!
//! [`Transaction::build_manifest`] is the centre of this module and of the
//! transaction machinery generally: given the current manifest and index list, it
//! decides the new fragment list, the surviving indices and the next row id, then
//! assembles the manifest. Everything else in `super` exists to serve it -- the
//! operation vocabulary it matches on, the index rules it applies, the row version
//! metadata it stamps, the validation that runs before it.

use crate::feature_flags::{
    FLAG_COVERED_INDEX_METADATA, FLAG_FRAGMENT_REUSE_INDEX, FLAG_STABLE_ROW_IDS,
    apply_feature_flags, ensure_can_read_manifest, ensure_can_write_manifest,
    inherit_sticky_feature_flags,
};
use crate::format::overlay::{OverlayCoverage, TOMBSTONE_FIELD_ID};
use crate::format::{
    DataFile, DataStorageFormat, Fragment, IndexMetadata, Manifest, ManifestBuildConfig, RowIdMeta,
    overlay::DataOverlayFile,
};
use crate::io::{
    commit::CommitHandler,
    manifest::{read_manifest, read_manifest_indexes},
};
use crate::rowids::version::build_version_meta;
use crate::rowids::{read_row_ids, write_row_ids};
use crate::system_index::frag_reuse::metadata::{is_tagged, validate_flags};
use crate::system_index::frag_reuse::{FRAG_REUSE_INDEX_NAME, is_frag_reuse_index_entry};
use crate::system_index::is_system_index;
use crate::system_index::mem_wal::{
    CompactedSsTable, IndexCatchupProgress, MEM_WAL_INDEX_NAME, load_mem_wal_index_details,
    new_mem_wal_index_meta, update_mem_wal_index_compacted_sstables,
};
use crate::transaction::UpdateMode::{RewriteColumns, RewriteRows};
use crate::transaction::prepare::{FragReuseUpdate, PreparedIndices};
use crate::transaction::row_version::resolve_update_version_metadata;
use crate::transaction::update_map::apply_update_map;
use crate::transaction::validate::merge_fragment_physically_rewritten;
use crate::transaction::{
    CoverageIdentity, DataReplacementGroup, LogicalIndexSegments, Operation, ReadVersionState,
    RewriteGroup, Transaction, UpdatedFragmentOffsets,
};
use lance_core::datatypes::{
    LANCE_UNENFORCED_CLUSTERING_KEY_POSITION, LANCE_UNENFORCED_PRIMARY_KEY,
    LANCE_UNENFORCED_PRIMARY_KEY_POSITION,
};
use lance_core::utils::parse::str_is_truthy;
use lance_core::{Error, Result};
use lance_file::version::ConcreteFileVersion;
use lance_io::object_store::ObjectStore;
use object_store::path::Path;
use roaring::RoaringBitmap;
use std::collections::{BTreeMap, HashMap, HashSet};
use std::sync::Arc;
use uuid::Uuid;

/// Table config key that opts a table into writing clustered deletions as
/// `Range` segments (`U64Segment::Ranges`). Set it to `true` with
/// `update_config`.
///
/// Manifests written afterwards re-encode bitmap row id segments as runs of
/// `Range` segments where that is smaller. That is the
/// existing wire format, so any version reads the result, but a reader without
/// the compact in-memory form handles thousands of segments per fragment
/// slowly. That is why it is opt-in rather than the default.
///
/// Setting the key back to `false` only stops further conversions. Fragments
/// already written as `Range` segments keep that encoding until something else
/// rewrites their row ids (compaction, for example), so coordinate reader
/// upgrades before enabling it.
pub const RANGE_SEGMENTS_CONFIG_KEY: &str = "lance.row_ids.range_segments";

fn range_segments_enabled(manifest: &Manifest) -> bool {
    manifest
        .config
        .get(RANGE_SEGMENTS_CONFIG_KEY)
        .is_some_and(|value| str_is_truthy(value))
}

/// Re-encode inline row id sequences as runs of `Range` segments when the
/// table has opted in.
///
/// Fragments that the previous manifest already re-encoded are left alone
/// (their inline bytes are the very same allocation), so a commit pays only
/// for the fragments it changed; enabling the key re-encodes every fragment
/// once.
fn apply_range_segments(
    manifest: &mut Manifest,
    current_manifest: Option<&Manifest>,
) -> Result<()> {
    if !range_segments_enabled(manifest) {
        return Ok(());
    }
    let unchanged: HashMap<u64, *const u8> = current_manifest
        .filter(|current| range_segments_enabled(current))
        .map(|current| {
            current
                .fragments
                .iter()
                .filter_map(|fragment| match &fragment.row_id_meta {
                    Some(RowIdMeta::Inline(data)) => Some((fragment.id, data.as_ptr())),
                    _ => None,
                })
                .collect()
        })
        .unwrap_or_default();
    let fragments = Arc::make_mut(&mut manifest.fragments);
    for fragment in fragments.iter_mut() {
        let Some(RowIdMeta::Inline(data)) = &fragment.row_id_meta else {
            continue;
        };
        if unchanged.get(&fragment.id) == Some(&data.as_ptr()) {
            continue;
        }
        let mut sequence = read_row_ids(&data[..])?;
        if sequence.use_range_segments() {
            fragment.row_id_meta = Some(RowIdMeta::Inline(write_row_ids(&sequence).into()));
        }
    }
    Ok(())
}

impl Transaction {
    pub(super) fn fragments_with_ids<'a, T>(
        new_fragments: T,
        fragment_id: &'a mut u64,
    ) -> impl Iterator<Item = Fragment> + 'a
    where
        T: IntoIterator<Item = Fragment> + 'a,
    {
        new_fragments.into_iter().map(move |mut f| {
            if f.id == 0 {
                f.id = *fragment_id;
                *fragment_id += 1;
            }
            f
        })
    }

    fn data_storage_format_from_files(
        fragments: &[Fragment],
        user_requested: Option<ConcreteFileVersion>,
    ) -> Result<DataStorageFormat> {
        // Mixed prewritten files cannot imply a default. An explicit default
        // takes precedence; finalization validates the referenced file versions.
        let version = match user_requested {
            Some(ConcreteFileVersion::V1) => {
                // Preserve legacy creation's homogeneous-file contract.
                if let Some(actual) = Fragment::try_infer_version(fragments)?
                    && actual != ConcreteFileVersion::V1
                {
                    return Err(Error::invalid_input(format!(
                        "User requested data storage version ({}) does not match version in data files ({actual})",
                        ConcreteFileVersion::V1
                    )));
                }
                Some(ConcreteFileVersion::V1)
            }
            Some(version) => Some(version),
            None => Fragment::try_infer_version(fragments)?,
        };
        Ok(version.map(DataStorageFormat::new).unwrap_or_default())
    }

    pub async fn restore_old_manifest(
        object_store: &ObjectStore,
        commit_handler: &dyn CommitHandler,
        base_path: &Path,
        version: u64,
        config: &ManifestBuildConfig,
        tx_path: &str,
        current_manifest: &Manifest,
    ) -> Result<(Manifest, Vec<IndexMetadata>)> {
        let location = commit_handler
            .resolve_version_location(base_path, version, &object_store.inner)
            .await?;
        let mut manifest = read_manifest(object_store, &location.path, location.size).await?;
        // This read bypasses Dataset's feature gates. Refuse unsupported target
        // manifests before apply_feature_flags can clear their unknown bits and
        // republish the referenced files as legacy-compatible.
        ensure_can_read_manifest(&manifest)?;
        ensure_can_write_manifest(&manifest)?;
        manifest.set_timestamp(config.timestamp_nanos);
        manifest.transaction_file = Some(tx_path.to_string());
        let indices = read_manifest_indexes(object_store, &location, &manifest).await?;
        validate_flags(&manifest, &indices)?;
        manifest.max_fragment_id = manifest
            .max_fragment_id
            .max(current_manifest.max_fragment_id);
        // Row ids are a high-water mark like fragment ids: rewinding hands old ids to new rows.
        manifest.next_row_id = manifest.next_row_id.max(current_manifest.next_row_id);
        // Turning stable row ids off would revert `_rowid` to row addresses, whose
        // namespace overlaps the ids this table has already handed out.
        if current_manifest.uses_stable_row_ids() && !manifest.uses_stable_row_ids() {
            return Err(Error::invalid_input(format!(
                "Cannot restore version {version}: stable row ids were enabled \
                 after it, and turning them back off would let row addresses \
                 collide with ids this table has already used"
            )));
        }
        inherit_sticky_feature_flags(&mut manifest, current_manifest)?;
        Ok((manifest, indices))
    }

    /// Every non-system logical index, mapped to what determines its coverage.
    ///
    /// A logical index may be backed by several physical segments, so "did this
    /// index change" is a question about the whole set. Sorted by UUID so the
    /// two sides compare positionally.
    pub fn logical_index_segments(indices: &[IndexMetadata]) -> LogicalIndexSegments {
        let mut by_name: LogicalIndexSegments = BTreeMap::new();
        for idx in indices.iter().filter(|idx| !is_system_index(idx)) {
            by_name
                .entry(idx.name.clone())
                .or_default()
                .push(CoverageIdentity {
                    uuid: idx.uuid,
                    fragment_bitmap: idx.fragment_bitmap.clone(),
                });
        }
        for segments in by_name.values_mut() {
            segments.sort_unstable_by_key(|segment| segment.uuid);
        }
        by_name
    }

    /// Apply MemWAL index-coverage rules once the final index list is known.
    ///
    /// Coverage records that a base-table index contains the rows a compaction
    /// copied in, and the WAL pod retires SSTables against it.
    ///
    /// It is derived, not reported. An index covering every fragment live at the
    /// transaction's read version holds every row compaction had copied in by
    /// then, so it is caught up to that version's `compacted_sstables`. That is
    /// the only proof available: nothing maps a generation to the fragments its
    /// rows landed in, so covering the table as the transaction read it is how
    /// an index shows it covered those rows. Fragments appended since are a
    /// later gap.
    ///
    /// Deriving rather than transmitting means no claim can go stale between
    /// inspection and commit, the answer survives rebase (`read_version` is
    /// fixed for a transaction's life), and any operation can earn coverage --
    /// an ordinary reindex that fully covers no longer has to throw its work
    /// away and wait for a repair.
    ///
    /// Only meaningful once catch-up is required, where a missing entry means
    /// "not caught up" and the SSTables stay. A legacy table reads a missing
    /// entry as "fully caught up", so this leaves it untouched rather than
    /// making the table look more covered than it is.
    pub fn apply_mem_wal_index_coverage(
        final_indices: &mut [IndexMetadata],
        segments_before: &LogicalIndexSegments,
        read_version_state: Option<ReadVersionState<'_>>,
        new_version: u64,
    ) -> Result<()> {
        let Some(pos) = final_indices
            .iter()
            .position(|idx| idx.name == MEM_WAL_INDEX_NAME)
        else {
            // The system index went away with this transaction (MemWAL disable,
            // or an overwrite). There is no coverage left to maintain.
            return Ok(());
        };

        let mut details = load_mem_wal_index_details(final_indices[pos].clone())?;

        // Nothing has ever been compacted, so no index can be behind and there
        // is no coverage to invalidate.
        if details.compacted_sstables.is_empty() && details.index_catchup.is_empty() {
            return Ok(());
        }

        let segments_after = Self::logical_index_segments(final_indices);
        let catchup_before = std::mem::take(&mut details.index_catchup);

        // Per shard: what this commit records as compacted, and the most the
        // read version may credit. Generations compacted after that read landed
        // in fragments no index under consideration has seen; the committed
        // value caps it in turn, so a read version since rolled back cannot
        // retire SSTables no live commit copied in.
        let read_details = read_version_state
            .map(|state| {
                state
                    .indices
                    .iter()
                    .find(|idx| idx.name == MEM_WAL_INDEX_NAME)
                    .cloned()
                    .map(load_mem_wal_index_details)
                    .transpose()
            })
            .transpose()?
            .flatten();
        let shards: Vec<(Uuid, u64, u64)> = details
            .compacted_sstables
            .iter()
            .map(|committed| {
                let at_read = read_details
                    .as_ref()
                    .and_then(|read| {
                        read.compacted_sstables
                            .iter()
                            .find(|s| s.shard_id == committed.shard_id)
                    })
                    .map_or(0, |s| s.generation);
                (
                    committed.shard_id,
                    committed.generation,
                    at_read.min(committed.generation),
                )
            })
            .collect();

        // Every fragment live when the transaction read the table. An index
        // spanning all of them holds every row compacted by then.
        let read_fragments: Option<RoaringBitmap> = read_version_state.map(|state| {
            state
                .manifest
                .fragments
                .iter()
                .map(|fragment| fragment.id as u32)
                .collect()
        });

        let covers_read_version = |segments: &[CoverageIdentity]| -> bool {
            let Some(required) = read_fragments.as_ref() else {
                return false;
            };
            if required.is_empty() {
                // Subset-of-empty is trivially true, so this would credit every
                // index on a table with no fragments. Refused because an empty
                // fragment list is not only what an emptied table looks like:
                // it is also what a manifest written before #8438 looks like,
                // where UpdateMemWalState published no fragments at all. On
                // such a table the SSTables are the last copy of those rows,
                // and crediting coverage would retire them. The cost is that a
                // genuinely emptied table keeps its SSTables.
                return false;
            }
            let mut covered = RoaringBitmap::new();
            for segment in segments {
                match segment.fragment_bitmap.as_ref() {
                    Some(bitmap) => covered |= bitmap,
                    // An unknown bitmap cannot be shown to cover anything.
                    None => return false,
                }
            }
            required.is_subset(&covered)
        };

        let mut rebuilt: Vec<IndexCatchupProgress> = Vec::new();
        for (name, after) in segments_after.iter() {
            // Compared by [`CoverageIdentity`], not segment UUID: an Update
            // that touches an indexed field prunes a segment's fragment bitmap
            // in place while keeping its UUID, so a UUID-only comparison would
            // carry a position forward that the index no longer earns.
            let unchanged = segments_before.get(name) == Some(after);
            let carried = unchanged
                .then(|| catchup_before.iter().find(|e| e.index_name == *name))
                .flatten();
            let proven = covers_read_version(after);

            if carried.is_none() && !proven {
                // Changed, and nothing shows the new index covers the read
                // version. No entry: a missing one reads as "not caught up".
                continue;
            }

            let generations = shards
                .iter()
                .map(|&(shard_id, committed, creditable)| {
                    let prior = carried
                        .and_then(|entry| entry.caught_up_generation_for_shard(&shard_id))
                        .unwrap_or(0);
                    let credited = if proven { creditable } else { 0 };
                    // Takes the better of what this commit proves and what an
                    // unchanged index already held, so a commit reading an older
                    // version does not lower a position it cannot re-prove. The
                    // clamp is the exception: a position above what this commit
                    // records as compacted describes rows no live commit copied
                    // in.
                    CompactedSsTable::new(shard_id, prior.max(credited).min(committed))
                })
                .collect::<Vec<_>>();
            if generations.iter().all(|g| g.generation == 0) {
                continue;
            }
            rebuilt.push(IndexCatchupProgress::new(name.clone(), generations));
        }
        rebuilt.sort_by(|a, b| a.index_name.cmp(&b.index_name));

        let mut before_sorted = catchup_before;
        before_sorted.sort_by(|a, b| a.index_name.cmp(&b.index_name));
        if rebuilt == before_sorted {
            return Ok(());
        }

        let dropped: Vec<&str> = before_sorted
            .iter()
            .map(|e| e.index_name.as_str())
            .filter(|name| !rebuilt.iter().any(|kept| kept.index_name == *name))
            .collect();
        if !dropped.is_empty() {
            // The first thing to check when SSTables stop becoming trimmable.
            log::info!(
                "MemWAL index catch-up invalidated at version {new_version} for {dropped:?}: \
                 these indices changed and no longer cover the version this commit read"
            );
        }

        details.index_catchup = rebuilt;
        final_indices[pos] = new_mem_wal_index_meta(new_version, details)?;
        Ok(())
    }

    /// Drop coverage for indices a post-`build_manifest` step narrowed.
    ///
    /// The derivation runs while the manifest is being built, but the index list
    /// is not final there: `migrate_indices` can recalculate a segment's
    /// fragment bitmap and keep its UUID, so an index can narrow after its
    /// position was decided. It reports which ones it touched rather than the
    /// caller re-snapshotting every bitmap to find out. Only ever removes.
    pub fn withdraw_coverage_invalidated_after_build(
        indices: &mut [IndexMetadata],
        changed: &[String],
        new_version: u64,
    ) -> Result<()> {
        if changed.is_empty() {
            return Ok(());
        }
        let Some(pos) = indices
            .iter()
            .position(|idx| idx.name == MEM_WAL_INDEX_NAME)
        else {
            return Ok(());
        };
        let mut details = load_mem_wal_index_details(indices[pos].clone())?;
        let before = details.index_catchup.len();
        details
            .index_catchup
            .retain(|entry| !changed.contains(&entry.index_name));
        if details.index_catchup.len() == before {
            return Ok(());
        }
        log::info!(
            "MemWAL index catch-up withdrawn at version {new_version} for {changed:?}: \
             these indices were recalculated after their coverage was derived"
        );
        indices[pos] = new_mem_wal_index_meta(new_version, details)?;
        Ok(())
    }

    /// Step one of a commit attempt: what the operation settles about the
    /// index list, computed against the manifest current at the attempt.
    ///
    /// Runs the tagged-table write gate (`system_index::frag_reuse::gate`)
    /// before anything is derived, keeps `current_indices` as the original
    /// list, and derives the prepared list: on an in-place column rewrite
    /// (an update with `fields_modified`, a merge swapping a column's data
    /// file, a data replacement) the coverage it invalidates is withdrawn
    /// along a tagged fragment reuse history or pruned by fragment id on any
    /// other table (`withdraw_in_place_rewrites`). Every other operation
    /// prepares the list unchanged; `Restore` never reaches the builder and
    /// restores its snapshot's list whole.
    ///
    /// `frag_reuse_ledger` is the decoded history of the current entry when
    /// the caller read it (the commit path decodes it once per attempt for
    /// the operations above); without it an inline history is decoded from
    /// the entry and an external one is refused, so a commit never guesses
    /// at the lineage. A history that does not decode is an error, never
    /// treated as "no lineage". `frag_reuse` is what the commit path settled
    /// about the entry itself.
    pub fn prepare_indices(
        &self,
        current_manifest: Option<&Manifest>,
        current_indices: Vec<IndexMetadata>,
        config: &ManifestBuildConfig,
        frag_reuse_ledger: Option<Arc<crate::system_index::frag_reuse::ledger::FragReuseLedger>>,
        frag_reuse: FragReuseUpdate,
    ) -> Result<PreparedIndices> {
        crate::system_index::frag_reuse::gate::classify(
            &self.operation,
            current_manifest,
            &current_indices,
            &frag_reuse,
            config.migration_next_row_id,
        )?;
        let mut prepared = current_indices.clone();
        if let Some(manifest) = current_manifest {
            let rewrites = Self::rewritten_physical_columns(
                &self.operation,
                &manifest.schema,
                &manifest.fragments,
            );
            if !rewrites.is_empty() {
                let lineage =
                    crate::system_index::frag_reuse::lineage::TaggedLineage::from_indices_with_ledger(
                        &current_indices,
                        frag_reuse_ledger,
                    )?;
                let live: RoaringBitmap = manifest
                    .fragments
                    .iter()
                    .filter_map(|fragment| u32::try_from(fragment.id).ok())
                    .collect();
                Self::withdraw_in_place_rewrites(
                    &mut prepared,
                    rewrites,
                    &manifest.schema,
                    &live,
                    lineage.as_ref(),
                );
            }
        }
        Ok(PreparedIndices::new(
            current_manifest.map(|manifest| manifest.version),
            current_indices,
            prepared,
            frag_reuse,
        ))
    }

    /// Create a new manifest from the current manifest and the transaction.
    ///
    /// `current_manifest` should only be None if the dataset does not yet exist.
    pub fn build_manifest(
        &self,
        current_manifest: Option<&Manifest>,
        current_indices: Vec<IndexMetadata>,
        transaction_file_path: &str,
        config: &ManifestBuildConfig,
    ) -> Result<(Manifest, Vec<IndexMetadata>)> {
        self.build_manifest_with_read_version(
            current_manifest,
            current_indices,
            transaction_file_path,
            config,
            None,
        )
    }

    /// [`Self::build_manifest`] with the version this transaction read.
    ///
    /// Supplied by the commit path, which already materializes that version.
    /// `None` where there is none to read -- dataset creation and detached
    /// commits -- in which case no index can be shown to cover it and coverage
    /// is left as the invalidation rules put it.
    ///
    /// Prepares the index list itself with nothing settled about the
    /// fragment reuse entry and no history read, so it stays safe on its
    /// own: a tagged entry on a `Rewrite` or a `CreateIndex` touching a
    /// tagged entry is refused, an inline history is decoded from the entry
    /// and an external one is refused. The commit path prepares first
    /// (`prepare_indices`) and builds with `build_manifest_prepared`.
    pub fn build_manifest_with_read_version(
        &self,
        current_manifest: Option<&Manifest>,
        current_indices: Vec<IndexMetadata>,
        transaction_file_path: &str,
        config: &ManifestBuildConfig,
        read_version_state: Option<ReadVersionState<'_>>,
    ) -> Result<(Manifest, Vec<IndexMetadata>)> {
        let prepared = self.prepare_indices(
            current_manifest,
            current_indices,
            config,
            None,
            FragReuseUpdate::None,
        )?;
        self.build_manifest_prepared(
            current_manifest,
            prepared,
            transaction_file_path,
            config,
            read_version_state,
        )
    }

    /// Build the next manifest from a prepared index list
    /// (`prepare_indices`): the builder applies the operation to the
    /// prepared list and publishes only the final list; the original list
    /// is what MemWAL coverage is compared against. The prepared result must
    /// have been derived against `current_manifest`.
    pub fn build_manifest_prepared(
        &self,
        current_manifest: Option<&Manifest>,
        prepared: PreparedIndices,
        transaction_file_path: &str,
        config: &ManifestBuildConfig,
        read_version_state: Option<ReadVersionState<'_>>,
    ) -> Result<(Manifest, Vec<IndexMetadata>)> {
        if prepared.manifest_version() != current_manifest.map(|manifest| manifest.version) {
            return Err(Error::invalid_input(format!(
                "the index list was prepared against manifest version {:?} but the manifest \
                 being built on is at version {:?}; every commit attempt prepares against the \
                 manifest it builds on",
                prepared.manifest_version(),
                current_manifest.map(|manifest| manifest.version)
            )));
        }
        // The tagged-table write gate, run again at this chokepoint (it is
        // pure) so a build cannot skip it whatever prepared the list.
        crate::system_index::frag_reuse::gate::classify(
            &self.operation,
            current_manifest,
            prepared.original(),
            prepared.frag_reuse_update(),
            config.migration_next_row_id,
        )?;
        let (original_indices, prepared_indices, frag_reuse) = prepared.into_parts();
        if config.use_stable_row_ids
            && config.migration_next_row_id.is_none()
            && current_manifest
                .map(|m| !m.uses_stable_row_ids())
                .unwrap_or_default()
        {
            return Err(Error::not_supported_source(
                "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(),
            ));
        }

        // The fragment-reuse index is internal bookkeeping registered by
        // compaction with deferred index remap, not something a user can
        // meaningfully drop and recreate, so it must not gate the migration.
        // It is dropped from the migrated manifest below, which is what makes
        // ignoring it here safe. Every other index, including the MemWAL
        // index, still blocks: only the fragment-reuse index is known to be
        // discardable.
        let blocking_indices: Vec<&str> = original_indices
            .iter()
            .filter(|idx| idx.name != FRAG_REUSE_INDEX_NAME)
            .map(|idx| idx.name.as_str())
            .collect();
        if config.migration_next_row_id.is_some() && !blocking_indices.is_empty() {
            return Err(Error::invalid_input(format!(
                "Cannot migrate to stable row IDs while indexes exist on the dataset. \
                 Drop the following indexes first, then re-run the migration, and \
                 recreate them afterwards: {}",
                blocking_indices.join(", ")
            )));
        }
        let mut reference_paths = match current_manifest {
            Some(m) => m.base_paths.clone(),
            None => HashMap::new(),
        };

        if let Operation::Overwrite {
            initial_bases: Some(initial_bases),
            ..
        } = &self.operation
        {
            if current_manifest.is_none() {
                // CREATE mode: registering base paths
                // Base IDs should have been assigned during write operation
                // Validate uniqueness and insert them into the manifest
                for base_path in initial_bases.iter() {
                    if reference_paths.contains_key(&base_path.id) {
                        return Err(Error::invalid_input(format!(
                            "Duplicate base path ID {} detected. Base path IDs must be unique.",
                            base_path.id
                        )));
                    }
                    reference_paths.insert(base_path.id, base_path.clone());
                }
            } else {
                // OVERWRITE mode with initial_bases should have been rejected by validation
                // This branch should never be reached
                return Err(Error::invalid_input(
                    "OVERWRITE mode cannot register new bases. This should have been caught by validation.",
                ));
            }
        }

        // Get the schema and the final fragment list
        let mut schema = match self.operation {
            Operation::Overwrite { ref schema, .. } => schema.clone(),
            Operation::Merge { ref schema, .. } => schema.clone(),
            Operation::Project { ref schema, .. } => schema.clone(),
            _ => {
                if let Some(current_manifest) = current_manifest {
                    current_manifest.schema.clone()
                } else {
                    return Err(Error::internal(
                        "Cannot create a new dataset without a schema".to_string(),
                    ));
                }
            }
        };

        // Fragment ids are a high water mark for the whole dataset history: an id
        // must never name two different sets of rows, or per-fragment state keyed
        // by id (caches, deletion files, row addresses) can be attributed to the
        // wrong rows.
        let mut fragment_id = current_manifest
            .and_then(|m| m.max_fragment_id())
            .map(|id| id + 1)
            .unwrap_or(0);
        let mut final_fragments = Vec::new();
        let mut final_indices = prepared_indices;

        // Release builds refuse to publish a stable-row-id dataset with a fragment
        // reuse index, and nothing needs it: compaction rejects deferred index
        // remap there.
        if config.migration_next_row_id.is_some() {
            final_indices.retain(|idx| idx.name != FRAG_REUSE_INDEX_NAME);
        }

        // The original list, as read from the current manifest and before
        // any preparation, so coverage can be compared against what each
        // logical index looked like going in. Only tables with a MemWAL
        // index maintain coverage, so every other commit -- and the segment
        // clones this costs -- pays nothing.
        let mem_wal_segments_before = original_indices
            .iter()
            .any(|idx| idx.name == MEM_WAL_INDEX_NAME)
            .then(|| Self::logical_index_segments(&original_indices));
        drop(original_indices);

        let mut next_row_id = {
            // Only use row ids if the feature flag is set already, or this is
            // a migration activation that explicitly provides the next_row_id.
            match (current_manifest, config.use_stable_row_ids) {
                (Some(manifest), _) if manifest.reader_feature_flags & FLAG_STABLE_ROW_IDS != 0 => {
                    Some(manifest.next_row_id)
                }
                (None, true) => Some(0),
                (_, false) => None,
                (Some(_), true) => {
                    // Migration activation: use the provided next_row_id.
                    if let Some(migration_nri) = config.migration_next_row_id {
                        Some(migration_nri)
                    } else {
                        return Err(Error::not_supported_source(
                            "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(),
                        ));
                    }
                }
            }
        };

        let maybe_existing_fragments =
            current_manifest
                .map(|m| m.fragments.as_ref())
                .ok_or_else(|| {
                    Error::internal(format!(
                        "No current manifest was provided while building manifest for operation {}",
                        self.operation.name()
                    ))
                });

        let new_version = current_manifest.map_or(1, |m| m.version + 1);

        match &self.operation {
            Operation::Clone { .. } => {
                return Err(Error::internal(
                    "Clone operation should not enter build_manifest.".to_string(),
                ));
            }
            Operation::Append { fragments } => {
                final_fragments.extend(maybe_existing_fragments?.clone());
                let mut new_fragments =
                    Self::fragments_with_ids(fragments.clone(), &mut fragment_id)
                        .collect::<Vec<_>>();
                if let Some(next_row_id) = &mut next_row_id {
                    Self::assign_row_ids(next_row_id, new_fragments.as_mut_slice())?;
                    // Add version metadata for all new fragments
                    for fragment in new_fragments.iter_mut() {
                        let version_meta = build_version_meta(fragment, new_version);
                        fragment.last_updated_at_version_meta = version_meta.clone();
                        fragment.created_at_version_meta = version_meta;
                    }
                }
                final_fragments.extend(new_fragments);
            }
            Operation::Delete {
                updated_fragments,
                deleted_fragment_ids,
                ..
            } => {
                // Remove the deleted fragments
                // Hash lookups keep this linear on tables with many fragments.
                let deleted_ids: HashSet<u64> = deleted_fragment_ids.iter().copied().collect();
                let updated_by_id: HashMap<u64, &Fragment> =
                    updated_fragments.iter().map(|f| (f.id, f)).collect();
                final_fragments.extend(maybe_existing_fragments?.clone());
                final_fragments.retain(|f| !deleted_ids.contains(&f.id));
                final_fragments.iter_mut().for_each(|f| {
                    if let Some(updated) = updated_by_id.get(&f.id) {
                        // The post-image was built at the transaction's read
                        // version. Only its deletion file is new; retain all
                        // other state from the current fragment when rebasing.
                        f.deletion_file = updated.deletion_file.clone();
                    }
                });
                Self::retain_relevant_indices(&mut final_indices, &schema, &final_fragments)
            }
            Operation::Update {
                removed_fragment_ids,
                updated_fragments,
                new_fragments,
                fields_modified,
                compacted_sstables,
                fields_for_preserving_frag_bitmap,
                update_mode,
                updated_fragment_offsets,
                ..
            } => {
                // Extract existing fragments once for reuse
                let existing_fragments = maybe_existing_fragments?;

                // Apply updates to existing fragments
                // Hash lookups keep this linear on tables with many fragments.
                let removed_ids: HashSet<u64> = removed_fragment_ids.iter().copied().collect();
                let mut updated_by_id: HashMap<u64, &Fragment> =
                    HashMap::with_capacity(updated_fragments.len());
                for fragment in updated_fragments {
                    updated_by_id.entry(fragment.id).or_insert(fragment);
                }
                let live_field_ids = schema
                    .fields_pre_order()
                    .map(|field| field.id)
                    .collect::<HashSet<_>>();
                let updated_frags: Vec<Fragment> = existing_fragments
                    .iter()
                    .filter_map(|f| {
                        if removed_ids.contains(&f.id) {
                            return None;
                        }
                        if let Some(&updated) = updated_by_id.get(&f.id) {
                            let mut updated = updated.clone();
                            // Carry forward the fragment's current overlays (which
                            // may include ones added by a concurrent commit). An
                            // in-place column rewrite then tombstones the overlaid
                            // fields it rewrote, since the fresh base values
                            // supersede them.
                            updated.overlays = f.overlays.clone();
                            // A concurrent Project may have pruned files from the
                            // current fragment after this post-image was staged.
                            // Spilled lineage fields are absent from the schema,
                            // but their carriers must survive just as in Project.
                            let spilled = updated.spilled_row_lineage_field_ids();
                            updated.files.retain(|file| {
                                file.fields.iter().any(|field_id| {
                                    live_field_ids.contains(field_id) || spilled.contains(field_id)
                                })
                            });
                            if matches!(update_mode, Some(RewriteColumns)) {
                                crate::format::overlay::tombstone_overlay_fields(
                                    &mut updated.overlays,
                                    fields_modified,
                                );
                            }
                            Some(updated)
                        } else {
                            Some(f.clone())
                        }
                    })
                    .collect();

                // Update version metadata for updated fragments if stable row IDs are enabled
                // Note: We don't update version metadata for fragments with deletion vectors
                // because the version sequences are indexed by physical row position, not logical position.
                // Version metadata for deleted rows will be filtered out during scan using the deletion vector.
                if next_row_id.is_some() {
                    // Version metadata will be properly set during compaction when deletions are materialized
                }

                final_fragments.extend(updated_frags);

                if next_row_id.is_some()
                    && matches!(update_mode, Some(RewriteColumns))
                    && let Some(UpdatedFragmentOffsets(off_map)) = updated_fragment_offsets
                    && !off_map.is_empty()
                {
                    let prev_version = current_manifest.map(|m| m.version).unwrap_or(0);
                    for fragment in final_fragments.iter_mut() {
                        let Some(bitmap) = off_map.get(&fragment.id) else {
                            continue;
                        };
                        // Defense-in-depth: only stamp fragments that were actually
                        // rewritten. validate_operation enforces this invariant before
                        // build_manifest is called; this guard catches any path that
                        // bypasses validation.
                        if !updated_by_id.contains_key(&fragment.id) {
                            continue;
                        }
                        if bitmap.is_empty() {
                            continue;
                        }
                        // Skip fragments with no existing version metadata: the helper
                        // would fill unmatched rows with prev_version, fabricating a
                        // last_updated stamp for rows that never had one.
                        if fragment.last_updated_at_version_meta.is_none() {
                            continue;
                        }
                        let max_allowed = existing_fragments
                            .iter()
                            .find(|f| f.id == fragment.id)
                            .and_then(|f| f.physical_rows)
                            .unwrap_or(1 << 24);
                        if bitmap.len() as usize > max_allowed {
                            return Err(Error::invalid_input(format!(
                                "updatedFragmentOffsets cardinality {} exceeds fragment {} limit {}",
                                bitmap.len(),
                                fragment.id,
                                max_allowed
                            )));
                        }
                        if let Some(max_off) = bitmap.max()
                            && max_off as usize >= max_allowed
                        {
                            return Err(Error::invalid_input(format!(
                                "updatedFragmentOffsets max offset {} exceeds fragment {} limit {}",
                                max_off, fragment.id, max_allowed
                            )));
                        }
                        let offsets: Vec<usize> = bitmap.iter().map(|o| o as usize).collect();
                        crate::rowids::version::refresh_row_latest_update_meta_for_partial_frag_rewrite_cols(
                            fragment,
                            &offsets,
                            new_version,
                            prev_version,
                            &config.spilled_row_lineage,
                        )?;
                    }
                }

                // The fragments whose indexed fields were rewritten were
                // withdrawn from their indices by `prepare_indices`.

                let mut new_fragments =
                    Self::fragments_with_ids(new_fragments.clone(), &mut fragment_id)
                        .collect::<Vec<_>>();
                // New fragments were staged against the same pre-Project schema.
                for fragment in &mut new_fragments {
                    let spilled = fragment.spilled_row_lineage_field_ids();
                    fragment.files.retain(|file| {
                        file.fields.iter().any(|field_id| {
                            live_field_ids.contains(field_id) || spilled.contains(field_id)
                        })
                    });
                }

                // Assign row IDs to any fragments that don't have them yet
                // (e.g., inserted rows from merge_insert operations)
                if let Some(next_row_id) = &mut next_row_id {
                    Self::assign_row_ids(next_row_id, new_fragments.as_mut_slice())?;
                }

                if next_row_id.is_some() {
                    resolve_update_version_metadata(
                        existing_fragments,
                        new_fragments.as_mut_slice(),
                        new_version,
                        &config.spilled_row_lineage,
                    )?;
                }

                if config.use_stable_row_ids
                    && update_mode.is_some()
                    && *update_mode == Some(RewriteRows)
                {
                    let pure_updated_frag_ids =
                        Self::collect_pure_rewrite_row_update_frags_ids(&new_fragments)?;

                    // collect all the original frag ids that contains the updated rows
                    let original_fragment_ids: Vec<u64> = removed_fragment_ids
                        .iter()
                        .chain(updated_fragments.iter().map(|f| &f.id))
                        .copied()
                        .collect();

                    // The original fragments that carried an overlay: their moved rows may have a
                    // stale index entry (see `register_pure_rewrite_rows_update_frags_in_indices`).
                    // Reuse the hash lookups built above instead of scanning
                    // `original_fragment_ids` per fragment.
                    let original_overlaid_frags: HashMap<u32, &Fragment> = existing_fragments
                        .iter()
                        .filter(|f| {
                            (removed_ids.contains(&f.id) || updated_by_id.contains_key(&f.id))
                                && !f.overlays.is_empty()
                        })
                        .map(|f| (f.id as u32, f))
                        .collect();

                    Self::register_pure_rewrite_rows_update_frags_in_indices(
                        &mut final_indices,
                        &pure_updated_frag_ids,
                        &original_fragment_ids,
                        fields_for_preserving_frag_bitmap,
                        &original_overlaid_frags,
                        &schema,
                    )?;
                }

                if let Some(next_row_id) = &mut next_row_id {
                    Self::assign_row_ids(next_row_id, new_fragments.as_mut_slice())?;
                    // Note: Version metadata is already set above (lines 1627-1755)
                    // for Update operations, preserving created_at from original fragments.
                    // Don't overwrite it here.
                }
                // Identify fragments that were updated or newly created in this update
                let mut target_ids: HashSet<u64> = HashSet::new();
                target_ids.extend(new_fragments.iter().map(|f| f.id));
                final_fragments.extend(new_fragments);
                Self::retain_relevant_indices(&mut final_indices, &schema, &final_fragments);

                if !compacted_sstables.is_empty() {
                    update_mem_wal_index_compacted_sstables(
                        &mut final_indices,
                        new_version,
                        compacted_sstables.clone(),
                    )?;
                }
            }
            Operation::Overwrite { fragments, .. } => {
                // Every fragment in an overwrite is newly written, so all of them
                // take fresh ids regardless of the id they arrive with. Fragments
                // carried over from the dataset being replaced are rejected by
                // `validate_operation`, which is what makes ignoring the incoming
                // id safe here.
                let mut new_fragments = fragments.clone();
                for fragment in new_fragments.iter_mut() {
                    fragment.id = fragment_id;
                    fragment_id += 1;
                }
                if let Some(next_row_id) = &mut next_row_id {
                    Self::assign_row_ids(next_row_id, new_fragments.as_mut_slice())?;
                    // Add version metadata for all new fragments
                    for fragment in new_fragments.iter_mut() {
                        let version_meta = build_version_meta(fragment, new_version);
                        fragment.last_updated_at_version_meta = version_meta.clone();
                        fragment.created_at_version_meta = version_meta;
                    }
                }
                final_fragments.extend(new_fragments);
                final_indices = Vec::new();
            }
            Operation::Rewrite {
                groups,
                rewritten_indices,
                frag_reuse_index,
            } => {
                final_fragments.extend(maybe_existing_fragments?.clone());
                let current_version = current_manifest.map(|m| m.version).unwrap_or_default();
                Self::handle_rewrite_fragments(
                    &mut final_fragments,
                    groups,
                    &mut fragment_id,
                    current_version,
                    next_row_id.as_ref(),
                )?;

                // Groups covered by the stable-partition transitions
                // redistribute rows across their destinations, so index
                // bitmaps must not follow them: the retired source ids stay
                // in the bitmaps as provenance and the tagged fragment reuse
                // entry records the row-level translation. Only the
                // order-preserving groups take part in bitmap maintenance
                // below.
                let assembly = match &frag_reuse {
                    FragReuseUpdate::Rewrite(assembly) => Some(assembly.as_ref()),
                    _ => None,
                };
                let ordered_groups = Self::ordered_rewrite_groups(
                    groups,
                    assembly.map(|assembly| &assembly.reordered_sources),
                )?;

                if next_row_id.is_some() {
                    // We can re-use indices, but need to rewrite the fragment bitmaps
                    debug_assert!(rewritten_indices.is_empty());
                    for index in final_indices.iter_mut() {
                        // Its bitmap is lineage, not coverage, and a straddling
                        // group would fail the recalculation.
                        if is_frag_reuse_index_entry(index) {
                            continue;
                        }
                        let results_are_row_addrs = index.results_are_row_addrs();
                        if let Some(fragment_bitmap) = &mut index.fragment_bitmap {
                            *fragment_bitmap = if results_are_row_addrs {
                                // Stable row ids survive a rewrite, so a row-id-domain index
                                // can simply follow its data to the new fragments. An
                                // address-domain index cannot: its stored addresses point into
                                // the fragments the rewrite dropped. Claiming coverage of the
                                // new fragments would make it answer queries with addresses
                                // that no longer resolve, so drop the rewritten fragments from
                                // its coverage instead and let the scanner fall back to a full
                                // scan for them.
                                Self::drop_rewritten_fragments(fragment_bitmap, &ordered_groups)
                            } else {
                                Self::recalculate_fragment_bitmap(fragment_bitmap, &ordered_groups)?
                            };
                        }
                    }
                } else {
                    Self::handle_rewrite_indices(
                        &mut final_indices,
                        rewritten_indices,
                        &ordered_groups,
                    )?;
                }

                // A full compaction materializes a fragment's overlays into fresh
                // base data. Any index older than one of those overlays was built on
                // the pre-overlay values, so drop the rewritten fragment from its
                // coverage to keep it from serving stale values.
                Self::prune_overlay_stale_fields_from_indices(&mut final_indices, groups);

                match (frag_reuse_index, assembly) {
                    // Tagged history: splice the entry the commit path
                    // assembled against the current manifest, never the
                    // operation's own entry (the caller's intent, which the
                    // assembly merged and validated).
                    (_, Some(assembly)) => {
                        // A consistency check on the assembly, not the
                        // concurrency protection. Concurrent rewrites are
                        // handled by the commit path, which re-assembles the
                        // entry against the latest manifest on every attempt
                        // and commits with compare-and-swap, so an assembly
                        // built on a stale entry never reaches this splice
                        // through that path. What this refuses is an
                        // assembled entry whose recorded base is not the
                        // entry in the manifest being built (an entry
                        // assembled by hand, or carried over from another
                        // base): splicing it would replace the transitions
                        // that entry already holds.
                        let existing_version = final_indices
                            .iter()
                            .find(|idx| idx.name == FRAG_REUSE_INDEX_NAME)
                            .map(|idx| idx.dataset_version);
                        if existing_version != assembly.base_entry_version {
                            return Err(Error::invalid_input(format!(
                                "the {} index entry in this manifest is at version {:?} but the \
                                 rewrite's entry was assembled against version {:?}; rebuild it \
                                 through the lance commit path, which assembles the entry against \
                                 the manifest it commits to",
                                FRAG_REUSE_INDEX_NAME,
                                existing_version,
                                assembly.base_entry_version,
                            )));
                        }
                        final_indices.retain(|idx| idx.name != assembly.entry.name);
                        final_indices.push(assembly.entry.clone());
                    }
                    // v0 snapshot: splice verbatim, byte-identical to the
                    // historical behavior. The chokepoint above already
                    // rejected tagged snapshots.
                    (Some(entry), None) => {
                        final_indices.retain(|idx| idx.name != entry.name);
                        final_indices.push(entry.clone());
                    }
                    (None, None) => {}
                }
            }
            Operation::CreateIndex {
                new_indices,
                removed_indices,
                ..
            } => {
                final_fragments.extend(maybe_existing_fragments?.clone());
                let removed_uuids = removed_indices
                    .iter()
                    .map(|old_index| old_index.uuid)
                    .collect::<HashSet<_>>();
                let new_uuids = new_indices
                    .iter()
                    .map(|new_index| new_index.uuid)
                    .collect::<HashSet<_>>();
                final_indices.retain(|existing_index| {
                    !removed_uuids.contains(&existing_index.uuid)
                        && !new_uuids.contains(&existing_index.uuid)
                });
                for new_index in new_indices {
                    new_index.validate_covering_fields()?;
                }
                final_indices.extend(new_indices.clone());
            }
            Operation::ReserveFragments { .. } | Operation::UpdateConfig { .. } => {
                final_fragments.extend(maybe_existing_fragments?.clone());
            }
            Operation::Merge { fragments, .. } => {
                let existing_fragments = maybe_existing_fragments?;
                let mut merged_fragments = fragments.clone();
                if next_row_id.is_some() {
                    let prev_by_id: HashMap<u64, &Fragment> =
                        existing_fragments.iter().map(|f| (f.id, f)).collect();
                    for fragment in merged_fragments.iter_mut() {
                        match prev_by_id.get(&fragment.id) {
                            Some(prev) => {
                                if merge_fragment_physically_rewritten(prev, fragment) {
                                    crate::rowids::version::refresh_row_latest_update_meta_for_full_frag_rewrite_cols(
                                        fragment,
                                        new_version,
                                    )?;
                                }
                            }
                            None => {
                                // Brand-new fragment ID not present in the previous manifest.
                                // Set both last_updated and created version meta, consistent
                                // with Append/Overwrite for genuinely new fragments.
                                crate::rowids::version::refresh_row_latest_update_meta_for_full_frag_rewrite_cols(
                                    fragment,
                                    new_version,
                                )?;
                                fragment.created_at_version_meta =
                                    fragment.last_updated_at_version_meta.clone();
                            }
                        }
                    }
                }
                final_fragments.extend(merged_fragments);

                // A Merge can rewrite a column's data file in place; the field stays
                // in the schema, so the index is retained. Its now-stale coverage of
                // the rewritten fragments was withdrawn by `prepare_indices`.

                // Some fields that have indices may have been removed, so we should
                // remove those indices as well.
                Self::retain_relevant_indices(&mut final_indices, &schema, &final_fragments)
            }
            Operation::Project { .. } => {
                final_fragments.extend(maybe_existing_fragments?.clone());

                // We might have removed all fields for certain data files, so
                // we should remove the data files that are no longer relevant.
                // A file carrying a spilled row lineage sequence stays: its
                // reserved ids are never in the schema, and it is the only copy.
                let remaining_field_ids = schema
                    .fields_pre_order()
                    .map(|f| f.id)
                    .collect::<HashSet<_>>();
                for fragment in final_fragments.iter_mut() {
                    let spilled = fragment.spilled_row_lineage_field_ids();
                    fragment.files.retain(|file| {
                        file.fields.iter().any(|field_id| {
                            remaining_field_ids.contains(field_id) || spilled.contains(field_id)
                        })
                    });
                }

                // Some fields that have indices may have been removed, so we should
                // remove those indices as well.
                Self::retain_relevant_indices(&mut final_indices, &schema, &final_fragments)
            }
            Operation::Restore { .. } => {
                unreachable!()
            }
            Operation::DataReplacement { replacements } => {
                log::warn!(
                    "Building manifest with DataReplacement operation. This operation is not stable yet, please use with caution."
                );

                let (old_fragment_ids, new_datafiles): (Vec<&u64>, Vec<&DataFile>) = replacements
                    .iter()
                    .map(|DataReplacementGroup(fragment_id, new_file)| (fragment_id, new_file))
                    .unzip();

                // 1. make sure the new files all have the same fields / or empty
                // NOTE: arguably this requirement could be relaxed in the future
                // for the sake of simplicity, we require the new files to have the same fields
                if new_datafiles
                    .iter()
                    .map(|f| f.fields.clone())
                    .collect::<HashSet<_>>()
                    .len()
                    > 1
                {
                    let field_info = new_datafiles
                        .iter()
                        .enumerate()
                        .map(|(id, f)| (id, f.fields.clone()))
                        .fold("".to_string(), |acc, (id, fields)| {
                            format!("{}File {}: {:?}\n", acc, id, fields)
                        });

                    return Err(Error::invalid_input(format!(
                        "All new data files must have the same fields, but found different fields:\n{field_info}"
                    )));
                }

                let existing_fragments = maybe_existing_fragments?;

                // Collect replaced field IDs before consuming new_datafiles
                let replaced_fields: Vec<u32> = new_datafiles
                    .first()
                    .map(|f| {
                        f.schema(&schema)
                            .field_ids()
                            .iter()
                            .chain(f.fields.iter())
                            .filter(|&&id| id >= 0)
                            .map(|&id| id as u32)
                            .collect()
                    })
                    .unwrap_or_default();

                // 2. check that the fragments being modified have isomorphic layouts along the columns being replaced
                // 3. add modified fragments to final_fragments
                for (frag_id, new_file) in old_fragment_ids.iter().zip(new_datafiles) {
                    let frag = existing_fragments
                        .iter()
                        .find(|f| f.id == **frag_id)
                        .ok_or_else(|| {
                            Error::invalid_input(
                                "Fragment being replaced not found in existing fragments",
                            )
                        })?;
                    let mut new_frag = frag.clone();

                    // Physical mappings differ across V2 encodings (a nested
                    // parent may have no column of its own). Replacement and
                    // tombstoning operate on the logical field coverage.
                    let replacement_ids = new_file
                        .schema(&schema)
                        .field_ids()
                        .into_iter()
                        .chain(new_file.fields.iter().copied())
                        .collect::<HashSet<_>>();

                    // TODO(rmeng): check new file and fragment are the same length

                    let mut columns_covered = HashSet::new();
                    // Set when an existing file covers exactly the replaced
                    // fields, so the whole file swaps rather than part of it.
                    let mut replaced_in_place = false;
                    for file in &mut new_frag.files {
                        if file.fields == new_file.fields
                            && file.file_major_version == new_file.file_major_version
                            && file.file_minor_version == new_file.file_minor_version
                        {
                            // assign the new file path / size / base to the fragment
                            file.path = new_file.path.clone();
                            file.file_size_bytes = new_file.file_size_bytes.clone();
                            file.base_id = new_file.base_id;
                            replaced_in_place = true;
                        }
                        columns_covered.extend(file.schema(&schema).field_ids());
                        columns_covered.extend(file.fields.iter().copied());
                    }
                    // Reject a file whose version does not decode before any
                    // arm publishes it.
                    new_file.file_version()?;

                    // SPECIAL CASE: if the column(s) being replaced are not covered by the fragment
                    // Then it means it's a all-NULL column that is being replaced with real data
                    // just add it to the final fragments. Push the DataFile as
                    // given so every field (including base_id) is preserved.
                    if columns_covered.is_disjoint(&replacement_ids) {
                        new_frag.files.push(new_file.clone());
                    } else if !replaced_in_place
                        && replacement_ids.is_subset(&columns_covered)
                        && new_frag.files.iter().all(|file| {
                            file.file_version()
                                .is_ok_and(|version| version != ConcreteFileVersion::V1)
                                || file
                                    .fields
                                    .iter()
                                    .all(|field| !replacement_ids.contains(field))
                        })
                    {
                        // Tombstone the replaced fields where they live and
                        // append the new file to answer for them, the idiom
                        // `update_columns` uses. Compaction decides that layout,
                        // so the fields may sit in one wider file or span
                        // several.
                        //
                        // Legacy V1 is excluded: its reader derives the page table
                        // offset from the first field in the metadata, so
                        // tombstoning one field leaves its siblings decoding from
                        // the wrong pages. A field a V1 file covers keeps
                        // exact-match replacement.
                        for file in &mut new_frag.files {
                            // Same reason as the guard above.
                            if file.file_version()? == ConcreteFileVersion::V1 {
                                continue;
                            }
                            file.fields = file
                                .fields
                                .iter()
                                .map(|field| {
                                    if replacement_ids.contains(field) {
                                        TOMBSTONE_FIELD_ID
                                    } else {
                                        *field
                                    }
                                })
                                .collect::<Vec<_>>()
                                .into();
                        }
                        // Every data file must share at least one field with
                        // the dataset schema: a file kept alive only by
                        // tombstones or by ids the schema no longer defines is
                        // unreachable to readers, uncollectable by cleanup,
                        // and reported corrupt by validate(). The exception is
                        // a file carrying one of the fragment's spilled row
                        // lineage sequences: their reserved ids are never in
                        // the schema, the file holds their only copy, and the
                        // replaced user columns in it are dead space until
                        // compaction.
                        let live_ids = schema
                            .fields_pre_order()
                            .map(|field| field.id)
                            .collect::<HashSet<i32>>();
                        let spilled = new_frag.spilled_row_lineage_field_ids();
                        new_frag.files.retain(|file| {
                            file.fields
                                .iter()
                                .any(|f| live_ids.contains(f) || spilled.contains(f))
                        });
                        new_frag.files.push(new_file.clone());
                    }

                    // Nothing changed in the current fragment, which is not expected -- error out
                    if &new_frag == frag {
                        return Err(Error::invalid_input(
                            "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",
                        ));
                    }

                    // New base values supersede any overlay still shadowing
                    // them, so tombstone the overlaid fields. An overlay
                    // committed after this transaction's snapshot is the newer
                    // value though -- the conflict resolver rebases these two
                    // precisely because the overlay wins -- so it stays, and
                    // being newer it stays last, preserving the ordering.
                    let (mut superseded, newer): (Vec<_>, Vec<_>) = new_frag
                        .overlays
                        .drain(..)
                        .partition(|overlay| overlay.committed_version <= self.read_version);
                    crate::format::overlay::tombstone_overlay_fields(
                        &mut superseded,
                        &replaced_fields,
                    );
                    superseded.extend(newer);
                    new_frag.overlays = superseded;

                    final_fragments.push(new_frag);
                }

                let fragments_changed = old_fragment_ids
                    .iter()
                    .cloned()
                    .cloned()
                    .collect::<HashSet<_>>();

                // 4. push fragments that didn't change back to final_fragments
                let unmodified_fragments = existing_fragments
                    .iter()
                    .filter(|f| !fragments_changed.contains(&f.id))
                    .cloned()
                    .collect::<Vec<_>>();

                final_fragments.extend(unmodified_fragments);

                // A replacement changes what its rows read as, so stamp them
                // updated. Without this, get_updated_rows never reports them and
                // an incremental consumer skips them for good.
                if next_row_id.is_some() {
                    let new_version = current_manifest.map_or(1, |m| m.version + 1);
                    for fragment in final_fragments
                        .iter_mut()
                        .filter(|f| fragments_changed.contains(&f.id))
                    {
                        crate::rowids::version::refresh_row_latest_update_meta_for_full_frag_rewrite_cols(
                            fragment,
                            new_version,
                        )?;
                    }
                }

                // The replaced fields' coverage of the modified fragments was
                // withdrawn by `prepare_indices`.
            }
            Operation::DataOverlay { groups } => {
                // Stamp each overlay with the version this commit is producing.
                // build_manifest re-runs on every retry with an updated
                // current_manifest, so this is naturally re-stamped on retry.
                let new_version = current_manifest.map_or(1, |m| m.version + 1);

                let existing_fragments = maybe_existing_fragments?;
                // Multiple groups may target the same fragment; merge them in
                // order rather than letting a HashMap collapse drop all but the
                // last group's overlays.
                let mut overlays_by_fragment: HashMap<u64, Vec<&DataOverlayFile>> = HashMap::new();
                for group in groups {
                    overlays_by_fragment
                        .entry(group.fragment_id)
                        .or_default()
                        .extend(group.overlays.iter());
                }

                // Every group must target an existing fragment. Build a set of
                // existing ids once so this is O(groups + fragments) rather than
                // O(groups * fragments).
                let existing_fragment_ids: HashSet<u64> =
                    existing_fragments.iter().map(|f| f.id).collect();
                for fragment_id in overlays_by_fragment.keys() {
                    if !existing_fragment_ids.contains(fragment_id) {
                        return Err(Error::invalid_input(format!(
                            "DataOverlay targets fragment {fragment_id}, which does not exist"
                        )));
                    }
                }

                for fragment in existing_fragments {
                    let mut fragment = fragment.clone();
                    if let Some(new_overlays) = overlays_by_fragment.get(&fragment.id) {
                        if next_row_id.is_some() {
                            let mut covered_offsets = RoaringBitmap::new();
                            for overlay in new_overlays {
                                match &overlay.coverage {
                                    OverlayCoverage::Shared(bitmap) => {
                                        covered_offsets |= bitmap.as_ref();
                                    }
                                    OverlayCoverage::PerField(bitmaps) => {
                                        for bitmap in bitmaps {
                                            covered_offsets |= bitmap.as_ref();
                                        }
                                    }
                                }
                            }
                            if !covered_offsets.is_empty() {
                                let covered_offsets: Vec<usize> = covered_offsets
                                    .iter()
                                    .map(|offset| offset as usize)
                                    .collect();
                                // Scans interpret missing lineage metadata as version 1,
                                // so using 1 as the fallback preserves the observable
                                // version of every uncovered row.
                                crate::rowids::version::refresh_row_latest_update_meta_for_partial_frag_rewrite_cols(
                                    &mut fragment,
                                    &covered_offsets,
                                    new_version,
                                    1,
                                    &config.spilled_row_lineage,
                                )?;
                            }
                        }
                        // Appended (not replaced) so concurrently-written overlays
                        // survive; later entries are newer.
                        fragment
                            .overlays
                            .extend(new_overlays.iter().map(|&overlay| {
                                let mut overlay = overlay.clone();
                                overlay.committed_version = new_version;
                                overlay
                            }));
                    }
                    final_fragments.push(fragment);
                }
            }
            Operation::UpdateMemWalState {
                compacted_sstables, ..
            } => {
                // Updates the MemWAL index only; the fragments are unchanged.
                final_fragments.extend(maybe_existing_fragments?.clone());
                update_mem_wal_index_compacted_sstables(
                    &mut final_indices,
                    new_version,
                    compacted_sstables.clone(),
                )?;
            }
            Operation::UpdateBases { .. } => {
                // UpdateBases operation doesn't modify fragments or indices
                // Base paths are handled in the manifest creation section below
                final_fragments.extend(maybe_existing_fragments?.clone());
            }
        };

        // If a fragment was reserved then it may not belong at the end of the fragments list.
        final_fragments.sort_by_key(|frag| frag.id);

        // Clean up data files that only contain tombstoned fields
        Self::remove_tombstoned_data_files(&mut final_fragments);

        // Enforce the newest-last overlay ordering invariant at the write
        // boundary. Load normalizes with a sort; this rejects any commit path
        // that assembled a fragment's overlays out of order.
        for fragment in &final_fragments {
            if !fragment.overlays.is_empty() {
                crate::format::overlay::verify_overlays_newest_last(&fragment.overlays)?;
            }
        }

        // A spilled row lineage sequence lives only in its carrier file. An
        // operation that dropped or duplicated that file would otherwise go
        // unnoticed until the next read, by which time cleanup may have
        // deleted the only copy. Every fragment is checked, not only the ones
        // this operation touched: the check reads metadata only, skips
        // fragments that spill nothing, and also refuses to build on a
        // manifest that already lost a carrier.
        for fragment in &final_fragments {
            fragment.validate_row_lineage_carriers()?;
        }

        let user_requested_version = match (&config.storage_format, config.use_legacy_format) {
            (Some(storage_format), _) => Some(storage_format.lance_file_format()),
            (None, Some(true)) => Some(ConcreteFileVersion::V1),
            (None, Some(false)) => Some(ConcreteFileVersion::V2_0),
            (None, None) => None,
        };

        // Applied once the final index list is known, so it sees exactly the
        // indices this commit publishes rather than what any one operation arm
        // intended.
        if let Some(segments_before) = mem_wal_segments_before.as_ref() {
            Self::apply_mem_wal_index_coverage(
                &mut final_indices,
                segments_before,
                read_version_state,
                new_version,
            )?;
        }

        // Blob is one logical column across file versions. Publishing its first
        // 2.2+ file also publishes the logical v2 view, without replacing older files.
        // Blob children are not independent physical columns; retain the root ID
        // and allocate logical child IDs against the current manifest on commit.
        let legacy_blob_ids = schema
            .fields_pre_order()
            .filter(|field| field.is_blob() && !field.is_blob_v2())
            .map(|field| field.id)
            .collect::<HashSet<_>>();
        if !legacy_blob_ids.is_empty() {
            let mut blob_v2_fields = std::collections::BTreeSet::new();
            for file in final_fragments
                .iter()
                .flat_map(|fragment| fragment.referenced_lance_files())
            {
                if matches!(
                    file.file_version()?,
                    ConcreteFileVersion::V2_2 | ConcreteFileVersion::V2_3
                ) {
                    blob_v2_fields.extend(
                        file.fields
                            .iter()
                            .copied()
                            .filter(|id| legacy_blob_ids.contains(id)),
                    );
                }
            }
            let mut next_field_id = current_manifest
                .map(|manifest| manifest.max_field_id())
                .unwrap_or(-1)
                .max(schema.max_field_id().unwrap_or(-1))
                + 1;
            for id in blob_v2_fields {
                if let Some(field) = schema.mut_field_by_id(id) {
                    field.promote_blob_v2()?;
                    field.set_id(field.parent_id, &mut next_field_id);
                }
            }
        }

        let mut manifest = if let Some(current_manifest) = current_manifest {
            // OVERWRITE with initial_bases on existing dataset is not allowed (caught by validation)
            // So we always use new_from_previous which preserves base_paths
            let mut prev_manifest =
                Manifest::new_from_previous(current_manifest, schema, Arc::new(final_fragments));

            if let (Some(user_requested_version), Operation::Overwrite { .. }) =
                (user_requested_version, &self.operation)
            {
                // If this is an overwrite operation and the user has requested a specific version
                // then overwrite with that version.  Otherwise, if the user didn't request a specific
                // version, then overwrite with whatever version we had before.
                prev_manifest.data_storage_format = DataStorageFormat::new(user_requested_version);
            }

            prev_manifest
        } else {
            let data_storage_format =
                Self::data_storage_format_from_files(&final_fragments, user_requested_version)?;
            Manifest::new(
                schema,
                Arc::new(final_fragments),
                data_storage_format,
                reference_paths,
            )
        };

        manifest.tag.clone_from(&self.tag);

        if config.auto_set_feature_flags {
            // Internal operations (e.g. CreateIndex) build with the default config,
            // which has use_stable_row_ids = false. Without inheriting from the previous
            // manifest, apply_feature_flags would clear FLAG_STABLE_ROW_IDS.
            let inherited = current_manifest
                .map(|m| m.uses_stable_row_ids())
                .unwrap_or(false);
            let use_stable_row_ids = config.use_stable_row_ids || inherited;
            apply_feature_flags(
                &mut manifest,
                use_stable_row_ids,
                config.disable_transaction_file,
            )?;
        }
        // Set after apply_feature_flags, which resets both flag words -- and a
        // `Manifest` only points at its index section, so the flag cannot be
        // derived there.
        //
        // Derived fresh from `final_indices` on every commit, never inherited.
        // Every manifest this reaches starts without the covering bit, so there
        // is no stale bit to clear. Dropping the last covering index lifts the
        // fence by simply not setting it again. Inheriting it from the previous
        // manifest instead would make the fence permanent.
        //
        // Both words: a reader that selects a vector index by membership of
        // `fields` would answer a query on a merely-carried column with an index
        // keyed on another one, and a writer that treats every entry of `fields`
        // as keyed would mismaintain it.
        if final_indices
            .iter()
            .any(|index| !index.covering_fields.is_empty())
        {
            manifest.reader_feature_flags |= FLAG_COVERED_INDEX_METADATA;
            manifest.writer_feature_flags |= FLAG_COVERED_INDEX_METADATA;
        }

        if let Some(current_manifest) = current_manifest {
            inherit_sticky_feature_flags(&mut manifest, current_manifest)?;
        }

        manifest.set_timestamp(config.timestamp_nanos);

        manifest.update_max_fragment_id();

        match &self.operation {
            Operation::Overwrite {
                config_upsert_values: Some(tm),
                ..
            } => {
                manifest.config_mut().extend(tm.clone());
            }
            Operation::UpdateConfig {
                config_updates,
                table_metadata_updates,
                schema_metadata_updates,
                field_metadata_updates,
            } => {
                if let Some(config_updates) = config_updates {
                    let mut config = manifest.config.clone();
                    apply_update_map(&mut config, config_updates);
                    manifest.config = config;
                }
                if let Some(table_metadata_updates) = table_metadata_updates {
                    let mut table_metadata = manifest.table_metadata.clone();
                    apply_update_map(&mut table_metadata, table_metadata_updates);
                    manifest.table_metadata = table_metadata;
                }
                if let Some(schema_metadata_updates) = schema_metadata_updates {
                    let mut schema_metadata = manifest.schema.metadata.clone();
                    apply_update_map(&mut schema_metadata, schema_metadata_updates);
                    manifest.schema.metadata = schema_metadata;
                }
                // The unenforced primary and clustering keys are reserved
                // schema properties: each is immutable once set, and its
                // reserved metadata keys cannot be written with an invalid
                // value. Capture the prior keys, and whether this transaction
                // writes a reserved key, before applying the updates so
                // violations can be rejected below. This runs on every apply,
                // including conflict-rebase, so it also rejects the
                // concurrent-writer race.
                let primary_key_before: Vec<i32> = manifest
                    .schema
                    .unenforced_primary_key()
                    .iter()
                    .map(|field| field.id)
                    .collect();
                let writes_primary_key = field_metadata_updates.values().any(|update| {
                    update.update_entries.iter().any(|entry| {
                        entry.key == LANCE_UNENFORCED_PRIMARY_KEY
                            || entry.key == LANCE_UNENFORCED_PRIMARY_KEY_POSITION
                    })
                });
                let clustering_key_before: Vec<i32> = manifest
                    .schema
                    .unenforced_clustering_key()
                    .iter()
                    .map(|field| field.id)
                    .collect();
                let writes_clustering_key = field_metadata_updates.values().any(|update| {
                    update
                        .update_entries
                        .iter()
                        .any(|entry| entry.key == LANCE_UNENFORCED_CLUSTERING_KEY_POSITION)
                });
                for (field_id, field_metadata_update) in field_metadata_updates {
                    if let Some(field) = manifest.schema.field_by_id_mut(*field_id) {
                        apply_update_map(&mut field.metadata, field_metadata_update);
                        // Also set unenforced primary key based on updated field metadata.
                        field.unenforced_primary_key_position = field
                            .metadata
                            .get(LANCE_UNENFORCED_PRIMARY_KEY_POSITION)
                            .and_then(|s| s.parse::<u32>().ok())
                            .or_else(|| {
                                field
                                    .metadata
                                    .get(LANCE_UNENFORCED_PRIMARY_KEY)
                                    .filter(|s| str_is_truthy(s))
                                    .map(|_| 0)
                            });
                        // Also set unenforced clustering key based on updated
                        // field metadata.
                        field.unenforced_clustering_key_position = field
                            .metadata
                            .get(LANCE_UNENFORCED_CLUSTERING_KEY_POSITION)
                            .and_then(|s| s.parse::<u32>().ok());
                    } else {
                        return Err(Error::invalid_input_source(
                            format!("Field with id {} does not exist", field_id).into(),
                        ));
                    }
                }
                let primary_key_after: Vec<i32> = manifest
                    .schema
                    .unenforced_primary_key()
                    .iter()
                    .map(|field| field.id)
                    .collect();
                if !primary_key_before.is_empty() {
                    // The primary key is already set: reject any change to it,
                    // and any write that touches a reserved primary key.
                    if writes_primary_key || primary_key_after != primary_key_before {
                        return Err(Error::invalid_input(
                            "the unenforced primary key is a reserved key and cannot be changed once set",
                        ));
                    }
                } else if writes_primary_key && primary_key_after.is_empty() {
                    // A reserved primary key was written but did not install a
                    // valid primary key (e.g. a non-marker flag value or a
                    // non-numeric position).
                    return Err(Error::invalid_input(
                        "the unenforced primary key is a reserved key and cannot be set to an invalid value",
                    ));
                }
                if writes_primary_key {
                    // Installing by field metadata skips the Arrow-schema
                    // conversion that would otherwise validate the key.
                    manifest.schema.verify_primary_key()?;
                }
                let clustering_key_after: Vec<i32> = manifest
                    .schema
                    .unenforced_clustering_key()
                    .iter()
                    .map(|field| field.id)
                    .collect();
                if !clustering_key_before.is_empty() {
                    // The clustering key is already set: reject any change to
                    // it, and any write that touches the reserved key.
                    if writes_clustering_key || clustering_key_after != clustering_key_before {
                        return Err(Error::invalid_input(
                            "the unenforced clustering key is a reserved key and cannot be changed once set",
                        ));
                    }
                } else if writes_clustering_key && clustering_key_after.is_empty() {
                    // The reserved clustering key was written but did not
                    // install a valid clustering key (e.g. a non-numeric
                    // position value).
                    return Err(Error::invalid_input(
                        "the unenforced clustering key is a reserved key and cannot be set to an invalid value",
                    ));
                }
            }
            _ => {}
        }

        apply_range_segments(&mut manifest, current_manifest)?;

        // Handle UpdateBases operation to update manifest base_paths
        if let Operation::UpdateBases { new_bases } = &self.operation {
            // Validate and add new base paths to the manifest
            for new_base in new_bases {
                // Check for conflicts with existing base paths
                if let Some(existing_base) = manifest
                    .base_paths
                    .values()
                    .find(|bp| bp.name == new_base.name || bp.path == new_base.path)
                {
                    return Err(Error::invalid_input(format!(
                        "Conflict detected: Base path with name '{:?}' or path '{}' already exists. Existing: name='{:?}', path='{}'",
                        new_base.name, new_base.path, existing_base.name, existing_base.path
                    )));
                }

                // Assign a new ID if not already assigned
                let mut base_to_add = new_base.clone();
                if base_to_add.id == 0 {
                    let next_id = manifest
                        .base_paths
                        .keys()
                        .max()
                        .map(|&id| id + 1)
                        .unwrap_or(1);
                    base_to_add.id = next_id;
                }

                manifest.base_paths.insert(base_to_add.id, base_to_add);
            }
        }

        if let Operation::ReserveFragments { num_fragments } = self.operation {
            manifest.max_fragment_id = Some(manifest.max_fragment_id.unwrap_or(0) + num_fragments);
        }

        manifest.transaction_file = Some(transaction_file_path.to_string());

        if let Some(next_row_id) = next_row_id {
            manifest.next_row_id = next_row_id;
        }

        if final_indices.iter().any(is_tagged) {
            manifest.reader_feature_flags |= FLAG_FRAGMENT_REUSE_INDEX;
            manifest.writer_feature_flags |= FLAG_FRAGMENT_REUSE_INDEX;
        }
        Ok((manifest, final_indices))
    }

    /// Remove data files that only contain tombstoned fields (-2)
    /// These files no longer contain any live data and can be safely dropped
    fn remove_tombstoned_data_files(fragments: &mut [Fragment]) {
        for fragment in fragments {
            fragment.files.retain(|file| {
                // Keep file if it has at least one non-tombstoned field
                file.fields.iter().any(|&field_id| field_id != -2)
            });
        }
    }
    /// Coverage of an index that a rewrite invalidates: the rewritten fragments are
    /// removed and the fragments they became are *not* added.
    fn drop_rewritten_fragments(old: &RoaringBitmap, groups: &[RewriteGroup]) -> RoaringBitmap {
        let mut new_bitmap = old.clone();
        for group in groups {
            for old_fragment in &group.old_fragments {
                new_bitmap.remove(old_fragment.id as u32);
            }
        }
        new_bitmap
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::format::overlay::OverlayCoverage;
    use crate::format::pb;
    use crate::format::{
        DeletionFile, DeletionFileType, ROW_CREATED_AT_VERSION_FIELD_ID, ROW_ID_FIELD_ID,
        ROW_LAST_UPDATED_AT_VERSION_FIELD_ID, RowDatasetVersionMeta, RowDatasetVersionSequence,
        RowIdMeta,
    };
    use crate::rowids::{RowIdSequence, write_row_ids};
    use crate::transaction::test_support::{
        default_build_config, last_updated_at_versions, make_stable_row_id_manifest,
        overlay_with_field, sample_index_metadata, sample_manifest, tagged_entry,
    };
    use crate::transaction::{DataOverlayGroup, UpdateMode, validate_operation};
    use arrow_schema::{DataType, Field as ArrowField, Schema as ArrowSchema};
    use lance_core::datatypes::Schema as LanceSchema;
    use lance_file::version::{ConcreteFileVersion, LanceFileVersion};
    use lance_io::utils::CachedFileSize;
    use std::collections::HashMap;
    use std::sync::Arc;
    use uuid::Uuid;

    /// Fragment 0 is the live destination of a rewrite that retired
    /// fragment 5; the entry carries that one transition inline.
    fn tagged_sample_manifest() -> (Manifest, IndexMetadata) {
        let mut manifest = sample_manifest();
        manifest.reader_feature_flags |= FLAG_FRAGMENT_REUSE_INDEX;
        manifest.writer_feature_flags |= FLAG_FRAGMENT_REUSE_INDEX;
        (manifest, tagged_entry(&[(&[5], &[0])]))
    }

    fn row_mutation(kind: &str) -> Operation {
        match kind {
            "delete" => Operation::Delete {
                updated_fragments: vec![],
                deleted_fragment_ids: vec![0],
                predicate: "true".into(),
            },
            "update" => crate::transaction::test_support::update_txn(vec![]).operation,
            "config" => Operation::UpdateConfig {
                config_updates: None,
                table_metadata_updates: None,
                schema_metadata_updates: None,
                field_metadata_updates: HashMap::new(),
            },
            _ => unreachable!(),
        }
    }

    /// Deletes, row-rewriting updates, config updates, projections and merges
    /// that rewrite no column touch deletion vectors, new fragments, the
    /// schema or the manifest config only: the tagged entry is carried
    /// through unchanged and the mutation commits.
    #[rstest::rstest]
    #[case::delete("delete")]
    #[case::update("update")]
    #[case::config("config")]
    #[case::project("project")]
    #[case::merge("merge")]
    fn tagged_history_allows_row_mutations(#[case] kind: &str) {
        let (manifest, fri) = tagged_sample_manifest();
        let operation = match kind {
            "project" => Operation::Project {
                schema: manifest.schema.clone(),
                preserves_nullability: true,
            },
            "merge" => Operation::Merge {
                fragments: manifest.fragments.as_ref().clone(),
                schema: manifest.schema.clone(),
                preserves_nullability: true,
            },
            _ => row_mutation(kind),
        };
        let transaction = Transaction::new(manifest.version, operation, None);
        let (_, indices) = transaction
            .build_manifest(
                Some(&manifest),
                vec![fri.clone()],
                "txn",
                &default_build_config(),
            )
            .unwrap();
        let carried = indices
            .iter()
            .find(|idx| idx.name == crate::system_index::frag_reuse::FRAG_REUSE_INDEX_NAME)
            .expect("the tagged entry is carried through");
        assert_eq!(carried.uuid, fri.uuid);
        assert_eq!(carried.index_version, fri.index_version);
    }

    /// An overwrite replaces every fragment and drops every index, the tagged
    /// entry included; the sticky flag keeps the tagged record form.
    #[test]
    fn tagged_history_overwrite_discards_the_entry() {
        let (manifest, fri) = tagged_sample_manifest();
        let transaction = Transaction::new(
            manifest.version,
            Operation::Overwrite {
                fragments: vec![Fragment::new(0)],
                schema: manifest.schema.clone(),
                config_upsert_values: None,
                initial_bases: None,
            },
            None,
        );
        let (new_manifest, indices) = transaction
            .build_manifest(Some(&manifest), vec![fri], "txn", &default_build_config())
            .unwrap();
        assert!(indices.is_empty(), "{indices:?}");
        assert_ne!(
            new_manifest.reader_feature_flags & FLAG_FRAGMENT_REUSE_INDEX,
            0
        );
    }

    /// The stable row id migration commits a `Merge` with the activation
    /// marker; it is refused while a tagged entry exists.
    #[test]
    fn tagged_history_refuses_stable_row_id_migration() {
        let (manifest, fri) = tagged_sample_manifest();
        let transaction = Transaction::new(
            manifest.version,
            Operation::Merge {
                fragments: manifest.fragments.as_ref().clone(),
                schema: manifest.schema.clone(),
                preserves_nullability: true,
            },
            None,
        );
        let config = ManifestBuildConfig {
            migration_next_row_id: Some(100),
            ..default_build_config()
        };
        let error = transaction
            .build_manifest(Some(&manifest), vec![fri], "txn", &config)
            .unwrap_err();
        assert!(matches!(error, Error::NotSupported { .. }), "{error}");
        assert!(
            error.to_string().contains("stable row id migration"),
            "{error}"
        );
    }

    /// The manifest's only fragment, 0, carries field 0 in `path`.
    fn manifest_with_file(path: &str) -> Manifest {
        let mut manifest = sample_manifest();
        manifest.fragments = Arc::new(vec![Fragment::new(0).with_file(
            path,
            vec![0],
            vec![0],
            ConcreteFileVersion::V2_0,
            None,
        )]);
        manifest.reader_feature_flags |= FLAG_FRAGMENT_REUSE_INDEX;
        manifest.writer_feature_flags |= FLAG_FRAGMENT_REUSE_INDEX;
        manifest
    }

    /// A user segment on field 0 built from fragment 5, which the rewrite
    /// that produced fragment 0 retired: the entry's history says so, and
    /// the segment covers fragment 0 only through it.
    fn translating_segment_and_entry() -> (IndexMetadata, IndexMetadata) {
        let mut segment = sample_index_metadata("id_idx");
        segment.fragment_bitmap = Some([5u32].into_iter().collect());
        (segment, tagged_entry(&[(&[5], &[0])]))
    }

    fn user_bitmap(indices: &[IndexMetadata]) -> RoaringBitmap {
        indices
            .iter()
            .find(|idx| idx.name == "id_idx")
            .unwrap()
            .fragment_bitmap
            .clone()
            .unwrap()
    }

    /// The three in-place column rewrites of field 0 on fragment 0.
    fn in_place_rewrite(kind: &str, manifest: &Manifest, fields: Vec<u32>) -> Operation {
        match kind {
            "update" => {
                let mut update = crate::transaction::test_support::update_txn(vec![]);
                let Operation::Update {
                    updated_fragments,
                    fields_modified,
                    ..
                } = &mut update.operation
                else {
                    unreachable!()
                };
                *updated_fragments = vec![Fragment::new(0)];
                *fields_modified = fields;
                update.operation
            }
            "merge" => Operation::Merge {
                fragments: vec![Fragment::new(0).with_file(
                    "b.lance",
                    fields.iter().map(|f| *f as i32).collect(),
                    vec![0],
                    ConcreteFileVersion::V2_0,
                    None,
                )],
                schema: manifest.schema.clone(),
                preserves_nullability: true,
            },
            "data_replacement" => Operation::DataReplacement {
                replacements: vec![DataReplacementGroup(
                    0,
                    DataFile::new(
                        "b.lance",
                        fields.iter().map(|f| *f as i32).collect(),
                        vec![0],
                        ConcreteFileVersion::V2_0,
                        None,
                        None,
                    ),
                )],
            },
            _ => unreachable!(),
        }
    }

    /// An in-place rewrite of a column that a translating segment indexes,
    /// on a fragment the segment reaches only through the history, is
    /// admitted; the segment cannot withdraw that fragment from its bitmap,
    /// so the whole transition's sources are withdrawn instead and the
    /// segment claims nothing for its destinations (the reader scans them).
    #[rstest::rstest]
    #[case::update("update")]
    #[case::merge("merge")]
    #[case::data_replacement("data_replacement")]
    fn tagged_history_withdraws_in_place_rewrite_covered_only_by_translation(#[case] kind: &str) {
        let manifest = manifest_with_file("a.lance");
        let (segment, fri) = translating_segment_and_entry();
        let transaction = Transaction::new(
            manifest.version,
            in_place_rewrite(kind, &manifest, vec![0]),
            None,
        );
        let (_, indices) = transaction
            .build_manifest(
                Some(&manifest),
                vec![segment, fri.clone()],
                "txn",
                &default_build_config(),
            )
            .unwrap();
        assert!(
            indices.iter().any(|idx| idx.uuid == fri.uuid),
            "entry carried"
        );
        assert!(
            user_bitmap(&indices).is_empty(),
            "{:?}",
            user_bitmap(&indices)
        );
    }

    /// Naming the fragment directly is no way out: a merged segment's bitmap
    /// is the union of its sources' provenance, and pruning the direct id
    /// alone would let the retired sources re-derive the same destination
    /// with the pre-rewrite values inside. Both go.
    #[rstest::rstest]
    #[case::update("update")]
    #[case::merge("merge")]
    #[case::data_replacement("data_replacement")]
    fn tagged_history_withdraws_mixed_provenance_whole(#[case] kind: &str) {
        let manifest = manifest_with_file("a.lance");
        let (mut segment, fri) = translating_segment_and_entry();
        segment.fragment_bitmap = Some([0u32, 5].into_iter().collect());
        let transaction = Transaction::new(
            manifest.version,
            in_place_rewrite(kind, &manifest, vec![0]),
            None,
        );
        let (_, indices) = transaction
            .build_manifest(
                Some(&manifest),
                vec![segment, fri],
                "txn",
                &default_build_config(),
            )
            .unwrap();
        assert!(
            user_bitmap(&indices).is_empty(),
            "{:?}",
            user_bitmap(&indices)
        );
    }

    /// Without a walkable history (a ledger this writer cannot read, or none
    /// at hand) the withdrawal is conservative: every retired fragment of a
    /// segment indexing the rewritten column goes, on the rewritten lineage
    /// or not, and only the live fragments stay. The same rewrite with the
    /// history walkable withdraws just the transition reaching the rewritten
    /// fragment.
    #[test]
    fn tagged_history_without_a_walkable_lineage_withdraws_every_retired_source() {
        use crate::system_index::frag_reuse::lineage::TaggedLineage;
        let entry = tagged_entry(&[(&[5], &[0]), (&[6], &[8])]);
        let mut segment = sample_index_metadata("id_idx");
        segment.fragment_bitmap = Some([0u32, 5, 6, 7].into_iter().collect());
        let live: RoaringBitmap = [0u32, 7, 8].into_iter().collect();

        let unwalkable = TaggedLineage::new(&entry, None);
        assert!(!unwalkable.has_ledger());
        let mut indices = vec![segment.clone()];
        Transaction::withdraw_rewritten_coverage(
            &mut indices,
            Some(&unwalkable),
            &live,
            &[(0, vec![0])],
        );
        assert_eq!(
            user_bitmap(&indices),
            [7u32].into_iter().collect::<RoaringBitmap>(),
            "the rewritten fragment and every retired fragment go"
        );

        let walkable = TaggedLineage::from_indices_with_ledger(&[entry], None)
            .unwrap()
            .unwrap();
        let mut indices = vec![segment];
        Transaction::withdraw_rewritten_coverage(
            &mut indices,
            Some(&walkable),
            &live,
            &[(0, vec![0])],
        );
        assert_eq!(
            user_bitmap(&indices),
            [6u32, 7].into_iter().collect::<RoaringBitmap>(),
            "only the transition reaching fragment 0 is withdrawn"
        );
    }

    /// What a withdrawal leaves, per logical index name: a segment emptied
    /// while a same-name sibling still has coverage leaves the manifest in
    /// the commit; the last segment of a name stays, empty, as the record of
    /// what to rebuild; a segment without a bitmap has unknown coverage and
    /// is neither withdrawn from nor counted as empty.
    #[test]
    fn tagged_history_retires_an_emptied_segment_only_beside_a_serving_sibling() {
        let mut manifest = manifest_with_file("a.lance");
        manifest.fragments = Arc::new(vec![
            Fragment::new(0).with_file(
                "a.lance",
                vec![0],
                vec![0],
                ConcreteFileVersion::V2_0,
                None,
            ),
            Fragment::new(1).with_file(
                "c.lance",
                vec![0],
                vec![0],
                ConcreteFileVersion::V2_0,
                None,
            ),
        ]);
        let (translating, fri) = translating_segment_and_entry();
        let mut sibling = sample_index_metadata("id_idx");
        sibling.uuid = Uuid::new_v4();
        sibling.fragment_bitmap = Some([1u32].into_iter().collect());
        let mut unknown = sample_index_metadata("other_idx");
        unknown.uuid = Uuid::new_v4();
        unknown.fragment_bitmap = None;
        let transaction = Transaction::new(
            manifest.version,
            in_place_rewrite("update", &manifest, vec![0]),
            None,
        );

        // Beside a serving sibling the emptied segment leaves the manifest.
        let (_, indices) = transaction
            .build_manifest(
                Some(&manifest),
                vec![
                    translating.clone(),
                    sibling.clone(),
                    unknown.clone(),
                    fri.clone(),
                ],
                "txn",
                &default_build_config(),
            )
            .unwrap();
        let names: Vec<(String, Uuid)> = indices
            .iter()
            .map(|idx| (idx.name.clone(), idx.uuid))
            .collect();
        assert!(
            !names.contains(&("id_idx".to_string(), translating.uuid)),
            "{names:?}"
        );
        assert_eq!(
            indices
                .iter()
                .find(|idx| idx.uuid == sibling.uuid)
                .unwrap()
                .fragment_bitmap,
            sibling.fragment_bitmap,
            "the sibling is untouched"
        );
        assert_eq!(
            indices
                .iter()
                .find(|idx| idx.uuid == unknown.uuid)
                .unwrap()
                .fragment_bitmap,
            None,
            "unknown coverage is neither withdrawn nor emptied"
        );

        // As the last segment of its name it stays, empty.
        let (_, indices) = transaction
            .build_manifest(
                Some(&manifest),
                vec![translating.clone(), fri],
                "txn",
                &default_build_config(),
            )
            .unwrap();
        let kept = indices
            .iter()
            .find(|idx| idx.uuid == translating.uuid)
            .expect("the last segment stays");
        assert!(kept.fragment_bitmap.as_ref().unwrap().is_empty());
    }

    /// A segment on an unrelated lineage (fragment 6 rewritten into 8) is
    /// untouched by a rewrite of fragment 0: the history is walked, not
    /// guessed at.
    #[test]
    fn tagged_history_leaves_unrelated_lineage_alone() {
        let manifest = manifest_with_file("a.lance");
        let fri = tagged_entry(&[(&[5], &[0]), (&[6], &[8])]);
        let mut segment = sample_index_metadata("id_idx");
        segment.fragment_bitmap = Some([6u32].into_iter().collect());
        let transaction = Transaction::new(
            manifest.version,
            in_place_rewrite("update", &manifest, vec![0]),
            None,
        );
        let (_, indices) = transaction
            .build_manifest(
                Some(&manifest),
                vec![segment, fri],
                "txn",
                &default_build_config(),
            )
            .unwrap();
        assert_eq!(user_bitmap(&indices), RoaringBitmap::from_iter([6u32]));
    }

    /// Without a supplied ledger an entry whose history is not inline cannot
    /// be walked, and the build refuses rather than guess.
    #[test]
    fn tagged_history_without_readable_history_is_an_error() {
        let manifest = manifest_with_file("a.lance");
        let (segment, mut fri) = translating_segment_and_entry();
        fri.index_details = None;
        let transaction = Transaction::new(
            manifest.version,
            in_place_rewrite("update", &manifest, vec![0]),
            None,
        );
        let error = transaction
            .build_manifest(
                Some(&manifest),
                vec![segment, fri],
                "txn",
                &default_build_config(),
            )
            .unwrap_err();
        assert!(matches!(error, Error::InvalidInput { .. }), "{error}");
        assert!(error.to_string().contains("history"), "{error}");
    }

    /// A segment whose bitmap is entirely live has the fragment pruned as on
    /// an untagged table; a translating segment that does not index the
    /// rewritten field, or whose rows the history moves elsewhere, is
    /// untouched. The entry is carried through in every case.
    #[rstest::rstest]
    #[case::update_direct("update", "direct")]
    #[case::merge_direct("merge", "direct")]
    #[case::data_replacement_direct("data_replacement", "direct")]
    #[case::update_unindexed_field("update", "unindexed_field")]
    #[case::merge_unindexed_field("merge", "unindexed_field")]
    #[case::update_off_lineage("update", "off_lineage")]
    fn tagged_history_prunes_or_keeps_what_the_history_says(
        #[case] kind: &str,
        #[case] shape: &str,
    ) {
        let manifest = manifest_with_file("a.lance");
        let (mut segment, mut fri) = translating_segment_and_entry();
        let mut fields = vec![0];
        match shape {
            "direct" => segment.fragment_bitmap = Some([0u32].into_iter().collect()),
            "unindexed_field" => fields = vec![1],
            // Fragment 5 was rewritten into 7, not into the rewritten 0.
            "off_lineage" => fri = tagged_entry(&[(&[5], &[7])]),
            _ => unreachable!(),
        }
        let transaction = Transaction::new(
            manifest.version,
            in_place_rewrite(kind, &manifest, fields),
            None,
        );
        let (_, indices) = transaction
            .build_manifest(
                Some(&manifest),
                vec![segment.clone(), fri.clone()],
                "txn",
                &default_build_config(),
            )
            .unwrap();
        let carried = indices
            .iter()
            .find(|idx| idx.name == crate::system_index::frag_reuse::FRAG_REUSE_INDEX_NAME)
            .expect("the tagged entry is carried through");
        assert_eq!(carried.uuid, fri.uuid);
        let user = indices.iter().find(|idx| idx.name == "id_idx").unwrap();
        let expected: RoaringBitmap = if shape == "direct" {
            // Fragment 0 was rewritten: withdrawn from the segment.
            RoaringBitmap::new()
        } else {
            segment.fragment_bitmap.unwrap()
        };
        assert_eq!(user.fragment_bitmap.as_ref().unwrap(), &expected, "{shape}");
    }

    /// A user index is created, replaced or dropped on a tagged table like on
    /// any other: the entry is carried through untouched.
    #[test]
    fn tagged_history_allows_user_index_create_index() {
        let (manifest, fri) = tagged_sample_manifest();
        let transaction = Transaction::new(
            manifest.version,
            Operation::CreateIndex {
                new_indices: vec![sample_index_metadata("id_idx")],
                removed_indices: vec![],
            },
            None,
        );
        let (_, indices) = transaction
            .build_manifest(
                Some(&manifest),
                vec![fri.clone()],
                "txn",
                &default_build_config(),
            )
            .unwrap();
        assert!(indices.iter().any(|idx| idx.name == "id_idx"));
        let carried = indices.iter().find(|idx| idx.uuid == fri.uuid).unwrap();
        assert_eq!(carried.index_details, fri.index_details);
        assert_eq!(carried.fragment_bitmap, fri.fragment_bitmap);
    }

    #[rstest::rstest]
    #[case::memwal("memwal")]
    #[case::bare_rewrite("bare_rewrite")]
    #[case::rewrite_with_v0_entry("rewrite_with_v0_entry")]
    fn tagged_history_rejects_unsupported_transactions(#[case] kind: &str) {
        let (manifest, fri) = tagged_sample_manifest();
        let operation = match kind {
            "memwal" => Operation::UpdateMemWalState {
                compacted_sstables: vec![],
            },
            // A rewrite carrying no entry, or a v0 entry, would splice away
            // the tagged history; only a tagged entry may replace one. The
            // bare rewrite touches fragment 0, which the entry's bitmap
            // covers (a bare rewrite of only uncovered fragments is allowed,
            // see `tagged_history_allows_rewrite_of_uncovered_fragments`).
            "bare_rewrite" => Operation::Rewrite {
                groups: vec![RewriteGroup {
                    old_fragments: vec![Fragment::new(0)],
                    new_fragments: vec![Fragment::new(10)],
                }],
                rewritten_indices: vec![],
                frag_reuse_index: None,
            },
            "rewrite_with_v0_entry" => {
                let mut v0_entry = fri.clone();
                v0_entry.index_version = 0;
                Operation::Rewrite {
                    groups: vec![],
                    rewritten_indices: vec![],
                    frag_reuse_index: Some(v0_entry),
                }
            }
            _ => unreachable!(),
        };
        let transaction = Transaction::new(manifest.version, operation, None);
        let error = transaction
            .build_manifest(Some(&manifest), vec![fri], "txn", &default_build_config())
            .unwrap_err();
        assert!(matches!(error, Error::NotSupported { .. }), "{error}");
        assert!(
            error.to_string().contains("Tagged FRI history maintenance"),
            "{error}"
        );
    }

    #[rstest::rstest]
    #[case::untagged_table(false)]
    #[case::tagged_table(true)]
    fn tagged_snapshot_replace_rejected(#[case] table_is_tagged: bool) {
        // A pre-assembled tagged entry that did not come from the commit
        // path's assembly (a `ReplaceEntry` snapshot) has bypassed
        // validation and may be stale; it must be rejected at the manifest
        // chokepoint regardless of the table's current state.
        let mut manifest = sample_manifest();
        let mut current_indices = vec![];
        if table_is_tagged {
            manifest.reader_feature_flags |= FLAG_FRAGMENT_REUSE_INDEX;
            manifest.writer_feature_flags |= FLAG_FRAGMENT_REUSE_INDEX;
            let mut current =
                sample_index_metadata(crate::system_index::frag_reuse::FRAG_REUSE_INDEX_NAME);
            current.fields.clear();
            current_indices.push(current);
        }
        let mut entry =
            sample_index_metadata(crate::system_index::frag_reuse::FRAG_REUSE_INDEX_NAME);
        entry.fields.clear();
        let transaction = Transaction::new(
            manifest.version,
            Operation::Rewrite {
                groups: vec![],
                rewritten_indices: vec![],
                frag_reuse_index: Some(entry),
            },
            None,
        );
        let error = transaction
            .build_manifest(
                Some(&manifest),
                current_indices,
                "txn",
                &default_build_config(),
            )
            .unwrap_err();
        assert!(matches!(error, Error::InvalidInput { .. }), "{error}");
        assert!(error.to_string().contains("must be assembled"), "{error}");
    }

    #[test]
    fn v0_snapshot_replace_rejected_on_flag_carrying_manifest_without_entry() {
        // The sticky flag alone forbids a v0 snapshot: a trim may have
        // deleted the fully drained tagged entry, but the flag survives and
        // remains the final authority on the record form. Publishing the
        // snapshot would silently downgrade the table to v0.
        let mut manifest = sample_manifest();
        manifest.reader_feature_flags |= FLAG_FRAGMENT_REUSE_INDEX;
        manifest.writer_feature_flags |= FLAG_FRAGMENT_REUSE_INDEX;
        let mut entry =
            sample_index_metadata(crate::system_index::frag_reuse::FRAG_REUSE_INDEX_NAME);
        entry.fields.clear();
        entry.index_version = 0;
        let transaction = Transaction::new(
            manifest.version,
            Operation::Rewrite {
                groups: vec![],
                rewritten_indices: vec![],
                frag_reuse_index: Some(entry),
            },
            None,
        );
        let error = transaction
            .build_manifest(Some(&manifest), vec![], "txn", &default_build_config())
            .unwrap_err();
        assert!(matches!(error, Error::InvalidInput { .. }), "{error}");
        assert!(error.to_string().contains("sticky feature flag"), "{error}");
    }

    #[test]
    fn tagged_history_allows_rewrite_of_uncovered_fragments() {
        // Deferred compaction of never-covered fragments commits a plain
        // rewrite (no entry, no intent): it cannot invalidate provenance or
        // the recorded lineage, so the tagged gate admits it.
        let mut manifest = sample_manifest_with_fragments(0..6);
        manifest.reader_feature_flags |= FLAG_FRAGMENT_REUSE_INDEX;
        manifest.writer_feature_flags |= FLAG_FRAGMENT_REUSE_INDEX;
        let mut fri = sample_index_metadata(crate::system_index::frag_reuse::FRAG_REUSE_INDEX_NAME);
        fri.fields.clear();
        assert_eq!(
            fri.fragment_bitmap.as_ref().unwrap(),
            &roaring::RoaringBitmap::from_iter([0u32])
        );
        let transaction = Transaction::new(
            manifest.version,
            Operation::Rewrite {
                groups: vec![RewriteGroup {
                    old_fragments: vec![Fragment::new(5)],
                    new_fragments: vec![Fragment::new(10)],
                }],
                rewritten_indices: vec![],
                frag_reuse_index: None,
            },
            None,
        );
        let (_, final_indices) = transaction
            .build_manifest(
                Some(&manifest),
                vec![fri.clone()],
                "txn",
                &default_build_config(),
            )
            .unwrap();
        // The tagged entry rides through untouched.
        assert!(final_indices.iter().any(|idx| idx.uuid == fri.uuid));
    }

    #[test]
    fn out_of_range_transition_source_id_rejected() {
        // Fragment ids in the reuse domain are bounded by the row-address
        // fragment space; an oversized id must error instead of silently
        // truncating into an alias of another fragment.
        let error = crate::transaction::reordered_sources(&[
            crate::format::pb::fragment_reuse_index_details::Transition {
                sources: vec![
                    crate::format::pb::fragment_reuse_index_details::FragmentDigest {
                        id: u64::from(u32::MAX) + 1,
                        physical_rows: 4,
                        num_deleted_rows: 0,
                    },
                ],
                destinations: vec![],
                mapping: None,
            },
        ])
        .unwrap_err();
        assert!(error.to_string().contains("row-address range"), "{error}");
    }

    #[test]
    fn tagged_history_allows_rewrite_appending_tagged_entry() {
        let mut manifest = sample_manifest();
        manifest.reader_feature_flags |= FLAG_FRAGMENT_REUSE_INDEX;
        manifest.writer_feature_flags |= FLAG_FRAGMENT_REUSE_INDEX;
        let mut current =
            sample_index_metadata(crate::system_index::frag_reuse::FRAG_REUSE_INDEX_NAME);
        current.fields.clear();
        current.dataset_version = 7;
        let mut appended =
            sample_index_metadata(crate::system_index::frag_reuse::FRAG_REUSE_INDEX_NAME);
        appended.fields.clear();

        let transaction = Transaction::new(
            manifest.version,
            Operation::Rewrite {
                groups: vec![],
                rewritten_indices: vec![],
                frag_reuse_index: Some(appended.clone()),
            },
            None,
        );
        let config = default_build_config();
        let prepared = transaction
            .prepare_indices(
                Some(&manifest),
                vec![current.clone()],
                &config,
                None,
                FragReuseUpdate::Rewrite(Arc::new(crate::transaction::TaggedRewriteAssembly {
                    entry: appended.clone(),
                    base_entry_version: Some(7),
                    reordered_sources: RoaringBitmap::new(),
                })),
            )
            .unwrap();
        let (new_manifest, final_indices) = transaction
            .build_manifest_prepared(Some(&manifest), prepared, "txn", &config, None)
            .unwrap();
        // The assembled entry replaced the previous one by name.
        let entries: Vec<_> = final_indices
            .iter()
            .filter(|idx| idx.name == crate::system_index::frag_reuse::FRAG_REUSE_INDEX_NAME)
            .collect();
        assert_eq!(entries.len(), 1);
        assert_eq!(entries[0].uuid, appended.uuid);
        assert_ne!(
            new_manifest.reader_feature_flags & FLAG_FRAGMENT_REUSE_INDEX,
            0
        );
    }

    #[test]
    fn frag_reuse_rewrite_guards_base_entry_version() {
        let mut manifest = sample_manifest();
        manifest.reader_feature_flags |= FLAG_FRAGMENT_REUSE_INDEX;
        manifest.writer_feature_flags |= FLAG_FRAGMENT_REUSE_INDEX;
        let mut current =
            sample_index_metadata(crate::system_index::frag_reuse::FRAG_REUSE_INDEX_NAME);
        current.fields.clear();
        current.dataset_version = 7;
        let mut appended =
            sample_index_metadata(crate::system_index::frag_reuse::FRAG_REUSE_INDEX_NAME);
        appended.fields.clear();

        // Assembled against a different entry (base None while the manifest
        // holds version 7): splicing would replace that entry's transitions.
        let transaction = Transaction::new(
            manifest.version,
            Operation::Rewrite {
                groups: vec![],
                rewritten_indices: vec![],
                frag_reuse_index: Some(appended.clone()),
            },
            None,
        );
        let config = default_build_config();
        let prepared = transaction
            .prepare_indices(
                Some(&manifest),
                vec![current],
                &config,
                None,
                FragReuseUpdate::Rewrite(Arc::new(crate::transaction::TaggedRewriteAssembly {
                    entry: appended.clone(),
                    base_entry_version: None,
                    reordered_sources: RoaringBitmap::new(),
                })),
            )
            .unwrap();
        let error = transaction
            .build_manifest_prepared(Some(&manifest), prepared, "txn", &config, None)
            .unwrap_err();
        assert!(matches!(error, Error::InvalidInput { .. }), "{error}");
        assert!(error.to_string().contains("rebuild"), "{error}");
    }

    /// The preparation keeps the original list as read and withdraws on the
    /// prepared list only; the build publishes exactly the prepared list.
    #[rstest::rstest]
    #[case::update("update")]
    #[case::merge("merge")]
    #[case::data_replacement("data_replacement")]
    fn prepare_indices_keeps_original_and_withdraws_prepared(#[case] kind: &str) {
        let manifest = manifest_with_file("a.lance");
        let (segment, fri) = translating_segment_and_entry();
        // A segment on the unrelated F6 -> F8 lineage is untouched.
        let fri = {
            let mut merged = fri;
            merged.index_details = tagged_entry(&[(&[5], &[0]), (&[6], &[8])]).index_details;
            merged.fragment_bitmap = Some([0u32, 5, 6, 8].into_iter().collect());
            merged
        };
        let mut unrelated = sample_index_metadata("other_idx");
        unrelated.fragment_bitmap = Some([6u32].into_iter().collect());
        let transaction = Transaction::new(
            manifest.version,
            in_place_rewrite(kind, &manifest, vec![0]),
            None,
        );
        let config = default_build_config();
        let prepared = transaction
            .prepare_indices(
                Some(&manifest),
                vec![segment.clone(), unrelated.clone(), fri.clone()],
                &config,
                None,
                FragReuseUpdate::None,
            )
            .unwrap();
        assert_eq!(prepared.manifest_version(), Some(manifest.version));
        assert_eq!(
            prepared.original(),
            &[segment.clone(), unrelated.clone(), fri],
            "the original list is what was read"
        );
        let prepared_user = prepared
            .prepared()
            .iter()
            .find(|idx| idx.uuid == segment.uuid)
            .unwrap();
        assert!(prepared_user.fragment_bitmap.as_ref().unwrap().is_empty());
        let prepared_unrelated = prepared
            .prepared()
            .iter()
            .find(|idx| idx.uuid == unrelated.uuid)
            .unwrap();
        assert_eq!(
            prepared_unrelated.fragment_bitmap,
            unrelated.fragment_bitmap
        );
        let (_, indices) = transaction
            .build_manifest_prepared(Some(&manifest), prepared.clone(), "txn", &config, None)
            .unwrap();
        assert_eq!(indices, prepared.prepared());
    }

    /// An ordinary table prepares the same way, pruning by fragment id.
    #[rstest::rstest]
    #[case::update("update")]
    #[case::merge("merge")]
    #[case::data_replacement("data_replacement")]
    fn prepare_indices_prunes_by_id_on_an_untagged_table(#[case] kind: &str) {
        let mut manifest = manifest_with_file("a.lance");
        manifest.reader_feature_flags &= !FLAG_FRAGMENT_REUSE_INDEX;
        manifest.writer_feature_flags &= !FLAG_FRAGMENT_REUSE_INDEX;
        let mut segment = sample_index_metadata("id_idx");
        segment.fragment_bitmap = Some([0u32, 3].into_iter().collect());
        let transaction = Transaction::new(
            manifest.version,
            in_place_rewrite(kind, &manifest, vec![0]),
            None,
        );
        let prepared = transaction
            .prepare_indices(
                Some(&manifest),
                vec![segment.clone()],
                &default_build_config(),
                None,
                FragReuseUpdate::None,
            )
            .unwrap();
        assert_eq!(prepared.original(), &[segment]);
        assert_eq!(
            prepared.prepared()[0].fragment_bitmap,
            Some(RoaringBitmap::from_iter([3u32]))
        );
    }

    /// Operations that rewrite no column in place, `Restore` included,
    /// prepare the list unchanged.
    #[test]
    fn prepare_indices_restore_and_non_rewrites_are_identity() {
        let manifest = manifest_with_file("a.lance");
        let (segment, fri) = translating_segment_and_entry();
        for operation in [
            Operation::Restore { version: 1 },
            Operation::Append { fragments: vec![] },
            Operation::Delete {
                updated_fragments: vec![],
                deleted_fragment_ids: vec![0],
                predicate: "true".into(),
            },
            in_place_rewrite("update", &manifest, vec![]),
        ] {
            let transaction = Transaction::new(manifest.version, operation, None);
            let prepared = transaction
                .prepare_indices(
                    Some(&manifest),
                    vec![segment.clone(), fri.clone()],
                    &default_build_config(),
                    None,
                    FragReuseUpdate::None,
                )
                .unwrap();
            assert_eq!(prepared.prepared(), prepared.original());
        }
    }

    /// A history stored externally is walked only when the commit path
    /// supplied the decoded ledger; without it the preparation refuses.
    #[test]
    fn prepare_indices_external_history_needs_the_supplied_ledger() {
        let manifest = manifest_with_file("a.lance");
        let (segment, fri) = translating_segment_and_entry();
        let ledger = crate::system_index::frag_reuse::ledger::FragReuseLedger::decode_inline(
            fri.index_version,
            fri.index_details.as_ref().unwrap(),
        )
        .unwrap()
        .unwrap();
        let mut external = fri;
        external.index_details = None;
        let transaction = Transaction::new(
            manifest.version,
            in_place_rewrite("update", &manifest, vec![0]),
            None,
        );
        let error = transaction
            .prepare_indices(
                Some(&manifest),
                vec![segment.clone(), external.clone()],
                &default_build_config(),
                None,
                FragReuseUpdate::None,
            )
            .unwrap_err();
        assert!(matches!(error, Error::InvalidInput { .. }), "{error}");
        let prepared = transaction
            .prepare_indices(
                Some(&manifest),
                vec![segment, external],
                &default_build_config(),
                Some(Arc::new(ledger)),
                FragReuseUpdate::None,
            )
            .unwrap();
        assert!(user_bitmap(prepared.prepared()).is_empty());
    }

    /// Corrupt inline details are an error of their own kind, never
    /// "no lineage" and never a successful build.
    #[test]
    fn prepare_indices_corrupt_inline_details_is_an_error() {
        let manifest = manifest_with_file("a.lance");
        let (segment, mut fri) = translating_segment_and_entry();
        let mut details = fri.index_details.as_ref().unwrap().as_ref().clone();
        details.value = vec![0xff, 0xff, 0xff, 0xff, 0xff];
        fri.index_details = Some(Arc::new(details));
        let transaction = Transaction::new(
            manifest.version,
            in_place_rewrite("update", &manifest, vec![0]),
            None,
        );
        let error = transaction
            .prepare_indices(
                Some(&manifest),
                vec![segment.clone(), fri.clone()],
                &default_build_config(),
                None,
                FragReuseUpdate::None,
            )
            .unwrap_err();
        assert!(
            matches!(error, Error::CorruptFile { .. }),
            "corrupt history must surface as corruption: {error:?}"
        );
        let error = transaction
            .build_manifest(
                Some(&manifest),
                vec![segment, fri],
                "txn",
                &default_build_config(),
            )
            .unwrap_err();
        assert!(matches!(error, Error::CorruptFile { .. }), "{error:?}");
    }

    /// A result prepared against one manifest cannot build on another:
    /// every attempt prepares against the manifest it builds on.
    #[test]
    fn build_manifest_prepared_refuses_a_result_prepared_against_another_manifest() {
        let manifest = manifest_with_file("a.lance");
        let (segment, fri) = translating_segment_and_entry();
        let transaction = Transaction::new(
            manifest.version,
            in_place_rewrite("update", &manifest, vec![0]),
            None,
        );
        let prepared = transaction
            .prepare_indices(
                Some(&manifest),
                vec![segment, fri],
                &default_build_config(),
                None,
                FragReuseUpdate::None,
            )
            .unwrap();
        let mut later = manifest;
        later.version += 1;
        let error = transaction
            .build_manifest_prepared(Some(&later), prepared, "txn", &default_build_config(), None)
            .unwrap_err();
        assert!(matches!(error, Error::InvalidInput { .. }), "{error}");
        assert!(error.to_string().contains("prepared against"), "{error}");
    }

    /// The low-level entry cannot skip what the commit path settles: a
    /// tagged rewrite entry is refused without the assembled update, whoever
    /// prepared the list.
    #[test]
    fn build_manifest_prepared_still_runs_the_gate() {
        let manifest = manifest_with_file("a.lance");
        let (_, fri) = translating_segment_and_entry();
        let mut appended = fri.clone();
        appended.uuid = uuid::Uuid::new_v4();
        let transaction = Transaction::new(
            manifest.version,
            Operation::Rewrite {
                groups: vec![],
                rewritten_indices: vec![],
                frag_reuse_index: Some(appended),
            },
            None,
        );
        let error = transaction
            .prepare_indices(
                Some(&manifest),
                vec![fri.clone()],
                &default_build_config(),
                None,
                FragReuseUpdate::None,
            )
            .unwrap_err();
        assert!(
            error.to_string().contains("assembled by the commit path"),
            "{error}"
        );
        let error = transaction
            .build_manifest(Some(&manifest), vec![fri], "txn", &default_build_config())
            .unwrap_err();
        assert!(
            error.to_string().contains("assembled by the commit path"),
            "{error}"
        );
    }

    fn sample_manifest_with_fragments(ids: std::ops::Range<u64>) -> Manifest {
        let schema = ArrowSchema::new(vec![ArrowField::new("id", DataType::Int32, false)]);
        Manifest::new(
            LanceSchema::try_from(&schema).unwrap(),
            Arc::new(ids.map(Fragment::new).collect()),
            DataStorageFormat::new(ConcreteFileVersion::V2_0),
            HashMap::new(),
        )
    }

    #[test]
    fn test_create_index_build_manifest_keeps_unremoved_same_name_indices() {
        let manifest = sample_manifest();
        let first_index = sample_index_metadata("vector_idx");
        let second_index = sample_index_metadata("vector_idx");
        let third_index = sample_index_metadata("vector_idx");

        let transaction = Transaction::new(
            manifest.version,
            Operation::CreateIndex {
                new_indices: vec![third_index.clone()],
                removed_indices: vec![second_index.clone()],
            },
            None,
        );

        let (_, final_indices) = transaction
            .build_manifest(
                Some(&manifest),
                vec![first_index.clone(), second_index.clone()],
                "txn",
                &default_build_config(),
            )
            .unwrap();

        assert_eq!(final_indices.len(), 2);
        assert!(final_indices.iter().any(|idx| idx.uuid == first_index.uuid));
        assert!(final_indices.iter().any(|idx| idx.uuid == third_index.uuid));
        assert!(
            !final_indices
                .iter()
                .any(|idx| idx.uuid == second_index.uuid)
        );
    }

    #[test]
    fn test_create_index_build_manifest_deduplicates_relisted_indices_by_uuid() {
        let manifest = sample_manifest();
        let first_index = sample_index_metadata("vector_idx");
        let second_index = sample_index_metadata("vector_idx");
        let third_index = sample_index_metadata("vector_idx");

        let transaction = Transaction::new(
            manifest.version,
            Operation::CreateIndex {
                new_indices: vec![first_index.clone(), third_index.clone()],
                removed_indices: vec![second_index.clone()],
            },
            None,
        );

        let (_, final_indices) = transaction
            .build_manifest(
                Some(&manifest),
                vec![first_index.clone(), second_index.clone()],
                "txn",
                &default_build_config(),
            )
            .unwrap();

        assert_eq!(final_indices.len(), 2);
        assert_eq!(
            final_indices
                .iter()
                .filter(|idx| idx.uuid == first_index.uuid)
                .count(),
            1
        );
        assert!(final_indices.iter().any(|idx| idx.uuid == third_index.uuid));
        assert!(
            !final_indices
                .iter()
                .any(|idx| idx.uuid == second_index.uuid)
        );
    }

    #[test]
    fn test_update_build_manifest_replaces_and_removes_fragments() {
        let manifest = sample_manifest_with_fragments(0..5);

        let mut updated2 = Fragment::new(2);
        updated2.physical_rows = Some(42);
        let mut updated4 = Fragment::new(4);
        updated4.physical_rows = Some(43);

        let transaction = Transaction::new(
            manifest.version,
            Operation::Update {
                removed_fragment_ids: vec![1],
                // Fragment 99 does not exist in the dataset; it must be ignored,
                // not appended.
                updated_fragments: vec![updated2, updated4, Fragment::new(99)],
                new_fragments: vec![],
                fields_modified: vec![],
                compacted_sstables: vec![],
                fields_for_preserving_frag_bitmap: vec![],
                update_mode: None,
                inserted_rows_filter: None,
                updated_fragment_offsets: None,
            },
            None,
        );

        let (new_manifest, _) = transaction
            .build_manifest(Some(&manifest), vec![], "txn", &default_build_config())
            .unwrap();

        let ids: Vec<u64> = new_manifest.fragments.iter().map(|f| f.id).collect();
        assert_eq!(ids, vec![0, 2, 3, 4]);
        let rows: Vec<Option<usize>> = new_manifest
            .fragments
            .iter()
            .map(|f| f.physical_rows)
            .collect();
        assert_eq!(rows, vec![None, Some(42), None, Some(43)]);
    }

    #[rstest::rstest]
    #[case::unspecified(None)]
    #[case::rewrite_rows(Some(UpdateMode::RewriteRows))]
    #[case::rewrite_columns(Some(UpdateMode::RewriteColumns))]
    fn test_update_build_manifest_does_not_restore_projected_files(
        #[case] update_mode: Option<UpdateMode>,
        #[values(false, true)] with_spilled_lineage: bool,
    ) {
        let mut manifest = sample_manifest_with_fragments(0..3);
        if with_spilled_lineage {
            manifest.reader_feature_flags = FLAG_STABLE_ROW_IDS;
            manifest.writer_feature_flags = FLAG_STABLE_ROW_IDS;
            manifest.next_row_id = 126;
            for fragment in Arc::make_mut(&mut manifest.fragments) {
                let start = fragment.id * 42;
                fragment.physical_rows = Some(42);
                fragment.row_id_meta = Some(RowIdMeta::Inline(
                    write_row_ids(&RowIdSequence::from(start..start + 42)).into(),
                ));
            }
        }
        let projected_file = DataFile::new(
            "projected.lance",
            vec![0],
            vec![0],
            ConcreteFileVersion::V2_0,
            None,
            None,
        );
        Arc::make_mut(&mut manifest.fragments)[1].files = vec![projected_file.clone()];

        // Model an update staged before a projection removed field 1's file.
        let mut updated = manifest.fragments[1].clone();
        updated.files.push(DataFile::new(
            "dropped.lance",
            vec![1],
            vec![0],
            ConcreteFileVersion::V2_0,
            None,
            None,
        ));
        updated.physical_rows = Some(42);
        let inserted_projected_file = DataFile::new(
            "inserted-projected.lance",
            vec![0],
            vec![0],
            ConcreteFileVersion::V2_0,
            None,
            None,
        );
        let mut inserted = Fragment::new(0);
        inserted.physical_rows = Some(42);
        inserted.files = vec![
            inserted_projected_file.clone(),
            DataFile::new(
                "inserted-dropped.lance",
                vec![1],
                vec![0],
                ConcreteFileVersion::V2_0,
                None,
                None,
            ),
        ];

        let mut expected_updated_files = vec![projected_file];
        let mut expected_inserted_files = vec![inserted_projected_file];
        if with_spilled_lineage {
            // Separate carriers ensure each lineage kind is retained even when
            // the file has no field from the live schema.
            for (fragment, expected_files) in [
                (&mut updated, &mut expected_updated_files),
                (&mut inserted, &mut expected_inserted_files),
            ] {
                fragment.row_id_meta = Some(RowIdMeta::Column);
                fragment.created_at_version_meta = Some(RowDatasetVersionMeta::Column);
                fragment.last_updated_at_version_meta = Some(RowDatasetVersionMeta::Column);
                for field_id in [
                    ROW_ID_FIELD_ID,
                    ROW_CREATED_AT_VERSION_FIELD_ID,
                    ROW_LAST_UPDATED_AT_VERSION_FIELD_ID,
                ] {
                    let file = DataFile::new(
                        format!("lineage-{field_id}.lance"),
                        vec![field_id],
                        vec![0],
                        ConcreteFileVersion::V2_0,
                        None,
                        None,
                    );
                    fragment.files.push(file.clone());
                    expected_files.push(file);
                }
            }
        }

        let transaction = Transaction::new(
            manifest.version,
            Operation::Update {
                removed_fragment_ids: vec![],
                updated_fragments: vec![updated],
                new_fragments: vec![inserted],
                fields_modified: vec![1],
                compacted_sstables: vec![],
                fields_for_preserving_frag_bitmap: vec![],
                update_mode,
                inserted_rows_filter: None,
                updated_fragment_offsets: None,
            },
            None,
        );

        let (new_manifest, _) = transaction
            .build_manifest(Some(&manifest), vec![], "txn", &default_build_config())
            .unwrap();

        let fragment = &new_manifest.fragments[1];
        assert_eq!(fragment.files, expected_updated_files);
        assert_eq!(fragment.physical_rows, Some(42));
        assert_eq!(new_manifest.fragments[3].files, expected_inserted_files);
        assert_eq!(new_manifest.max_field_id(), 0);
    }

    #[test]
    fn test_delete_build_manifest_applies_deletion_to_current_fragment() {
        let mut manifest = sample_manifest_with_fragments(0..5);
        manifest.version = 2;
        let projected_file = DataFile::new_legacy_from_fields("projected.lance", vec![0], None);
        let current_fragment = &mut Arc::make_mut(&mut manifest.fragments)[2];
        current_fragment.files = vec![projected_file.clone()];
        current_fragment.physical_rows = Some(42);
        current_fragment.overlays = vec![overlay_with_field(0, 2)];
        current_fragment.last_updated_at_version_meta = Some(
            RowDatasetVersionMeta::from_sequence(
                &RowDatasetVersionSequence::from_uniform_row_count(42, 2),
            )
            .unwrap(),
        );

        // Model a delete staged before a projection removed field 1's file.
        let mut updated2 = Fragment::new(2);
        updated2.physical_rows = Some(42);
        updated2.files = vec![
            projected_file.clone(),
            DataFile::new_legacy_from_fields("dropped.lance", vec![1], None),
        ];
        let deletion_file = DeletionFile {
            read_version: 1,
            id: 10,
            file_type: DeletionFileType::Array,
            num_deleted_rows: Some(1),
            base_id: None,
        };
        updated2.deletion_file = Some(deletion_file.clone());

        let transaction = Transaction::new(
            1,
            Operation::Delete {
                updated_fragments: vec![updated2],
                deleted_fragment_ids: vec![1, 3],
                predicate: "id > 0".to_string(),
            },
            None,
        );

        let (new_manifest, _) = transaction
            .build_manifest(Some(&manifest), vec![], "txn", &default_build_config())
            .unwrap();

        let ids: Vec<u64> = new_manifest.fragments.iter().map(|f| f.id).collect();
        assert_eq!(ids, vec![0, 2, 4]);
        let fragment2 = new_manifest
            .fragments
            .iter()
            .find(|fragment| fragment.id == 2)
            .unwrap();
        assert_eq!(fragment2.files, vec![projected_file]);
        assert_eq!(fragment2.deletion_file, Some(deletion_file));
        assert_eq!(fragment2.overlays.len(), 1);
        assert_eq!(fragment2.overlays[0].committed_version, 2);
        assert_eq!(fragment2.physical_rows, Some(42));
        assert_eq!(new_manifest.max_field_id(), 0);
        assert_eq!(last_updated_at_versions(&new_manifest, 2), vec![2; 42]);
    }

    #[test]
    fn test_remove_tombstoned_data_files() {
        // Create a fragment with mixed data files: some normal, some fully tombstoned
        let mut fragment = Fragment::new(1);

        // Add a normal data file with valid field IDs
        fragment.files.push(DataFile {
            path: "normal.lance".to_string(),
            fields: Arc::from([1, 2, 3]),
            column_indices: Arc::from([]),
            file_major_version: 2,
            file_minor_version: 0,
            file_size_bytes: CachedFileSize::new(1000),
            base_id: None,
        });

        // Add a data file with all fields tombstoned
        fragment.files.push(DataFile {
            path: "all_tombstoned.lance".to_string(),
            fields: Arc::from([-2, -2, -2]),
            column_indices: Arc::from([]),
            file_major_version: 2,
            file_minor_version: 0,
            file_size_bytes: CachedFileSize::new(500),
            base_id: None,
        });

        // Add a data file with mixed tombstoned and valid fields
        fragment.files.push(DataFile {
            path: "mixed.lance".to_string(),
            fields: Arc::from([4, -2, 5]),
            column_indices: Arc::from([]),
            file_major_version: 2,
            file_minor_version: 0,
            file_size_bytes: CachedFileSize::new(750),
            base_id: None,
        });

        // Add another fully tombstoned file
        fragment.files.push(DataFile {
            path: "another_tombstoned.lance".to_string(),
            fields: Arc::from([-2_i32]),
            column_indices: Arc::from([]),
            file_major_version: 2,
            file_minor_version: 0,
            file_size_bytes: CachedFileSize::new(250),
            base_id: None,
        });

        let mut fragments = vec![fragment];

        // Apply the cleanup
        Transaction::remove_tombstoned_data_files(&mut fragments);

        // Should have removed the two fully tombstoned files
        assert_eq!(fragments[0].files.len(), 2);
        assert_eq!(fragments[0].files[0].path, "normal.lance");
        assert_eq!(fragments[0].files[1].path, "mixed.lance");
    }

    /// When a fragment has no existing last_updated_at_version_meta (None), a
    /// partial RewriteColumns refresh must leave it as None rather than fabricating
    /// prev_version for unmatched rows.
    #[test]
    fn test_partial_rewrite_skips_fragment_with_no_version_meta() {
        let row_ids = RowIdSequence::from([10u64, 11, 12, 13, 14].as_slice());
        let row_id_meta = Some(RowIdMeta::Inline(write_row_ids(&row_ids).into()));

        let data_file = DataFile::new(
            "data.lance",
            vec![0],
            vec![0],
            LanceFileVersion::Stable.resolve(),
            None,
            None,
        );

        let fragment = Fragment {
            id: 1,
            files: vec![data_file],
            overlays: vec![],
            deletion_file: None,
            row_id_meta,
            physical_rows: Some(5),
            last_updated_at_version_meta: None,
            created_at_version_meta: None,
        };

        let manifest = make_stable_row_id_manifest(vec![fragment.clone()]);

        // Simulate a RewriteColumns update that matched offsets 1 and 3
        let off_map = HashMap::from([(1u64, RoaringBitmap::from_iter([1u32, 3]))]);
        let tx = Transaction::new(
            manifest.version,
            Operation::Update {
                removed_fragment_ids: vec![],
                updated_fragments: vec![fragment],
                new_fragments: vec![],
                fields_modified: vec![],
                compacted_sstables: vec![],
                fields_for_preserving_frag_bitmap: vec![],
                update_mode: Some(UpdateMode::RewriteColumns),
                inserted_rows_filter: None,
                updated_fragment_offsets: Some(UpdatedFragmentOffsets(off_map)),
            },
            None,
        );

        let (out, _) = tx
            .build_manifest(Some(&manifest), vec![], "txn", &default_build_config())
            .unwrap();

        assert!(
            out.fragments[0].last_updated_at_version_meta.is_none(),
            "fragment with no prior version metadata must not have fabricated prev_version stamped on unmatched rows"
        );
    }

    #[test]
    fn test_bitmap_cardinality_exceeds_physical_rows() {
        let row_ids = RowIdSequence::from([10u64, 11, 12, 13, 14].as_slice());
        let row_id_meta = Some(RowIdMeta::Inline(write_row_ids(&row_ids).into()));

        let data_file = DataFile::new(
            "data.lance",
            vec![0],
            vec![0],
            LanceFileVersion::Stable.resolve(),
            None,
            None,
        );

        let version_seq = RowDatasetVersionSequence::from_uniform_row_count(5, 1);
        let version_meta = RowDatasetVersionMeta::from_sequence(&version_seq).unwrap();

        let fragment = Fragment {
            id: 1,
            files: vec![data_file],
            overlays: vec![],
            deletion_file: None,
            row_id_meta,
            physical_rows: Some(5),
            last_updated_at_version_meta: Some(version_meta.clone()),
            created_at_version_meta: Some(version_meta),
        };

        let manifest = make_stable_row_id_manifest(vec![fragment.clone()]);

        // Bitmap with 10 offsets but fragment only has 5 physical rows.
        let off_map = HashMap::from([(1u64, RoaringBitmap::from_iter(0u32..10))]);
        let tx = Transaction::new(
            manifest.version,
            Operation::Update {
                removed_fragment_ids: vec![],
                updated_fragments: vec![fragment],
                new_fragments: vec![],
                fields_modified: vec![],
                compacted_sstables: vec![],
                fields_for_preserving_frag_bitmap: vec![],
                update_mode: Some(UpdateMode::RewriteColumns),
                inserted_rows_filter: None,
                updated_fragment_offsets: Some(UpdatedFragmentOffsets(off_map)),
            },
            None,
        );

        let result = tx.build_manifest(Some(&manifest), vec![], "txn", &default_build_config());
        assert!(result.is_err());
        let msg = result.unwrap_err().to_string();
        assert!(
            msg.contains("cardinality"),
            "expected cardinality error, got: {msg}"
        );
    }

    #[test]
    fn test_bitmap_max_offset_exceeds_physical_rows() {
        let row_ids = RowIdSequence::from([10u64, 11, 12, 13, 14].as_slice());
        let row_id_meta = Some(RowIdMeta::Inline(write_row_ids(&row_ids).into()));

        let data_file = DataFile::new(
            "data.lance",
            vec![0],
            vec![0],
            LanceFileVersion::Stable.resolve(),
            None,
            None,
        );

        let version_seq = RowDatasetVersionSequence::from_uniform_row_count(5, 1);
        let version_meta = RowDatasetVersionMeta::from_sequence(&version_seq).unwrap();

        let fragment = Fragment {
            id: 1,
            files: vec![data_file],
            overlays: vec![],
            deletion_file: None,
            row_id_meta,
            physical_rows: Some(5),
            last_updated_at_version_meta: Some(version_meta.clone()),
            created_at_version_meta: Some(version_meta),
        };

        let manifest = make_stable_row_id_manifest(vec![fragment.clone()]);

        // Only 2 offsets (within cardinality) but max offset 100 exceeds physical_rows 5.
        let off_map = HashMap::from([(1u64, RoaringBitmap::from_iter([0u32, 100]))]);
        let tx = Transaction::new(
            manifest.version,
            Operation::Update {
                removed_fragment_ids: vec![],
                updated_fragments: vec![fragment],
                new_fragments: vec![],
                fields_modified: vec![],
                compacted_sstables: vec![],
                fields_for_preserving_frag_bitmap: vec![],
                update_mode: Some(UpdateMode::RewriteColumns),
                inserted_rows_filter: None,
                updated_fragment_offsets: Some(UpdatedFragmentOffsets(off_map)),
            },
            None,
        );

        let result = tx.build_manifest(Some(&manifest), vec![], "txn", &default_build_config());
        assert!(result.is_err());
        let msg = result.unwrap_err().to_string();
        assert!(
            msg.contains("max offset"),
            "expected max offset error, got: {msg}"
        );
    }

    #[test]
    fn test_bitmap_at_exact_physical_rows_boundary_succeeds() {
        let row_ids = RowIdSequence::from([10u64, 11, 12, 13, 14].as_slice());
        let row_id_meta = Some(RowIdMeta::Inline(write_row_ids(&row_ids).into()));

        let data_file = DataFile::new(
            "data.lance",
            vec![0],
            vec![0],
            LanceFileVersion::Stable.resolve(),
            None,
            None,
        );

        let version_seq = RowDatasetVersionSequence::from_uniform_row_count(5, 1);
        let version_meta = RowDatasetVersionMeta::from_sequence(&version_seq).unwrap();

        let fragment = Fragment {
            id: 1,
            files: vec![data_file],
            overlays: vec![],
            deletion_file: None,
            row_id_meta,
            physical_rows: Some(5),
            last_updated_at_version_meta: Some(version_meta.clone()),
            created_at_version_meta: Some(version_meta),
        };

        let manifest = make_stable_row_id_manifest(vec![fragment.clone()]);

        // All 5 offsets on a 5-row fragment — exactly at the boundary, should succeed.
        let off_map = HashMap::from([(1u64, RoaringBitmap::from_iter(0u32..5))]);
        let tx = Transaction::new(
            manifest.version,
            Operation::Update {
                removed_fragment_ids: vec![],
                updated_fragments: vec![fragment],
                new_fragments: vec![],
                fields_modified: vec![],
                compacted_sstables: vec![],
                fields_for_preserving_frag_bitmap: vec![],
                update_mode: Some(UpdateMode::RewriteColumns),
                inserted_rows_filter: None,
                updated_fragment_offsets: Some(UpdatedFragmentOffsets(off_map)),
            },
            None,
        );

        tx.build_manifest(Some(&manifest), vec![], "txn", &default_build_config())
            .expect("bitmap at exact physical_rows boundary should succeed");
    }

    #[test]
    fn test_updated_fragment_offsets_key_not_in_updated_fragments_is_rejected() {
        // Fragment A is being rewritten; fragment B exists in the manifest but is
        // NOT in updated_fragments. Supplying an offset key for B must be rejected
        // so that B's version metadata cannot be stamped by an unrelated commit.
        let make_fragment = |id: u64| {
            let row_ids = RowIdSequence::from([id * 10].as_slice());
            let row_id_meta = Some(RowIdMeta::Inline(write_row_ids(&row_ids).into()));
            Fragment {
                id,
                files: vec![DataFile::new(
                    format!("{id}.lance"),
                    vec![0],
                    vec![0],
                    LanceFileVersion::Stable.resolve(),
                    None,
                    None,
                )],
                overlays: vec![],
                deletion_file: None,
                row_id_meta,
                physical_rows: Some(5),
                last_updated_at_version_meta: None,
                created_at_version_meta: None,
            }
        };

        let frag_a = make_fragment(1);
        let frag_b = make_fragment(2);
        let manifest = make_stable_row_id_manifest(vec![frag_a.clone(), frag_b.clone()]);

        // updated_fragments contains only A; offsets are keyed to B — must fail.
        let off_map = HashMap::from([(frag_b.id, RoaringBitmap::from_iter([0u32, 1, 2]))]);
        let operation = Operation::Update {
            removed_fragment_ids: vec![],
            updated_fragments: vec![frag_a],
            new_fragments: vec![],
            fields_modified: vec![],
            compacted_sstables: vec![],
            fields_for_preserving_frag_bitmap: vec![],
            update_mode: Some(UpdateMode::RewriteColumns),
            inserted_rows_filter: None,
            updated_fragment_offsets: Some(UpdatedFragmentOffsets(off_map)),
        };

        let err = validate_operation(Some(&manifest), &operation).unwrap_err();
        assert!(
            err.to_string().contains("not in updated_fragments"),
            "expected key-presence error, got: {err}"
        );
    }

    #[test]
    fn test_proto_round_trip_field_10() {
        let off_map = HashMap::from([
            (1u64, RoaringBitmap::from_iter([1u32, 3, 5])),
            (2u64, RoaringBitmap::from_iter([0u32, 2, 4, 6])),
        ]);
        let tx = Transaction::new(
            1,
            Operation::Update {
                removed_fragment_ids: vec![],
                updated_fragments: vec![],
                new_fragments: vec![],
                fields_modified: vec![],
                compacted_sstables: vec![],
                fields_for_preserving_frag_bitmap: vec![],
                update_mode: Some(UpdateMode::RewriteColumns),
                inserted_rows_filter: None,
                updated_fragment_offsets: Some(UpdatedFragmentOffsets(off_map.clone())),
            },
            None,
        );

        let pb_tx: pb::Transaction = pb::Transaction::from(&tx);

        // Field 9 must be empty; field 10 must be populated.
        if let Some(pb::transaction::Operation::Update(ref update)) = pb_tx.operation {
            assert!(
                update.updated_fragment_offsets.is_empty(),
                "field 9 should be empty"
            );
            assert_eq!(update.updated_fragment_offset_bitmaps.len(), 2);
        } else {
            panic!("expected Update operation");
        }

        let tx2 = Transaction::try_from(pb_tx).unwrap();
        if let Operation::Update {
            updated_fragment_offsets: Some(UpdatedFragmentOffsets(m)),
            ..
        } = &tx2.operation
        {
            assert_eq!(m.len(), 2);
            assert_eq!(*m.get(&1).unwrap(), off_map[&1]);
            assert_eq!(*m.get(&2).unwrap(), off_map[&2]);
        } else {
            panic!("expected Update with offsets");
        }
    }

    #[test]
    fn test_proto_legacy_field_9_read() {
        // Simulate a manifest written by old Lance: only field 9, no field 10.
        let pb_tx = pb::Transaction {
            read_version: 1,
            uuid: "test".to_string(),
            tag: String::new(),
            transaction_properties: HashMap::new(),
            operation: Some(pb::transaction::Operation::Update(
                pb::transaction::Update {
                    removed_fragment_ids: vec![],
                    updated_fragments: vec![],
                    new_fragments: vec![],
                    fields_modified: vec![],
                    compacted_sstables: vec![],
                    fields_for_preserving_frag_bitmap: vec![],
                    update_mode: 1,
                    inserted_rows: None,
                    updated_fragment_offsets: HashMap::from([(
                        1u64,
                        pb::transaction::UInt32List {
                            values: vec![1, 3, 5],
                        },
                    )]),
                    updated_fragment_offset_bitmaps: HashMap::new(),
                },
            )),
        };

        let tx = Transaction::try_from(pb_tx).unwrap();
        if let Operation::Update {
            updated_fragment_offsets: Some(UpdatedFragmentOffsets(m)),
            ..
        } = &tx.operation
        {
            assert_eq!(m.len(), 1);
            let bitmap = m.get(&1).unwrap();
            let offsets: Vec<u32> = bitmap.iter().collect();
            assert_eq!(offsets, vec![1, 3, 5]);
        } else {
            panic!("expected Update with offsets from legacy field 9");
        }
    }

    #[test]
    fn test_proto_field_10_takes_precedence_over_field_9() {
        // When both fields present, field 10 wins.
        let mut bitmap_bytes = Vec::new();
        RoaringBitmap::from_iter([10u32, 20, 30])
            .serialize_into(&mut bitmap_bytes)
            .unwrap();

        let pb_tx = pb::Transaction {
            read_version: 1,
            uuid: "test".to_string(),
            tag: String::new(),
            transaction_properties: HashMap::new(),
            operation: Some(pb::transaction::Operation::Update(
                pb::transaction::Update {
                    removed_fragment_ids: vec![],
                    updated_fragments: vec![],
                    new_fragments: vec![],
                    fields_modified: vec![],
                    compacted_sstables: vec![],
                    fields_for_preserving_frag_bitmap: vec![],
                    update_mode: 1,
                    inserted_rows: None,
                    // Field 9 has different values than field 10.
                    updated_fragment_offsets: HashMap::from([(
                        1u64,
                        pb::transaction::UInt32List {
                            values: vec![99, 100],
                        },
                    )]),
                    updated_fragment_offset_bitmaps: HashMap::from([(1u64, bitmap_bytes)]),
                },
            )),
        };

        let tx = Transaction::try_from(pb_tx).unwrap();
        if let Operation::Update {
            updated_fragment_offsets: Some(UpdatedFragmentOffsets(m)),
            ..
        } = &tx.operation
        {
            let offsets: Vec<u32> = m.get(&1).unwrap().iter().collect();
            assert_eq!(offsets, vec![10, 20, 30], "field 10 should take precedence");
        } else {
            panic!("expected Update with offsets from field 10");
        }
    }

    #[test]
    fn merge_build_manifest_refreshes_last_updated_when_data_files_change_stable_row_ids() {
        use crate::feature_flags::FLAG_STABLE_ROW_IDS;
        use lance_file::version::LanceFileVersion;

        let mk_file = |path: &str| {
            DataFile::new(
                path,
                vec![0],
                vec![0],
                LanceFileVersion::Stable.resolve(),
                None,
                None,
            )
        };

        let arrow_schema = ArrowSchema::new(vec![ArrowField::new("id", DataType::Int32, false)]);
        let lance_schema = LanceSchema::try_from(&arrow_schema).unwrap();

        let row_ids = RowIdSequence::from([100u64, 101, 102, 103, 104].as_slice());
        let row_id_meta = Some(RowIdMeta::Inline(write_row_ids(&row_ids).into()));

        let prev_fragment = Fragment {
            id: 0,
            files: vec![mk_file("before.lance")],
            overlays: vec![],
            deletion_file: None,
            row_id_meta,
            physical_rows: Some(5),
            last_updated_at_version_meta: None,
            created_at_version_meta: None,
        };

        let mut manifest = Manifest::new(
            lance_schema.clone(),
            Arc::new(vec![prev_fragment.clone()]),
            DataStorageFormat::new(ConcreteFileVersion::V2_0),
            HashMap::new(),
        );
        manifest.reader_feature_flags |= FLAG_STABLE_ROW_IDS;
        manifest.next_row_id = 100;

        let merged_fragment = Fragment {
            files: vec![mk_file("after.lance")],
            ..prev_fragment
        };

        let tx = Transaction::new(
            manifest.version,
            Operation::Merge {
                fragments: vec![merged_fragment],
                schema: lance_schema,
                preserves_nullability: true,
            },
            None,
        );

        let (out, _) = tx
            .build_manifest(Some(&manifest), vec![], "txn", &default_build_config())
            .unwrap();

        assert_eq!(out.version, 2);
        let frag = &out.fragments[0];
        let seq = frag
            .last_updated_at_version_meta
            .as_ref()
            .unwrap()
            .load_sequence()
            .unwrap();
        assert_eq!(seq.version_at(0).unwrap(), 2);
        assert_eq!(seq.version_at(4).unwrap(), 2);
    }

    #[test]
    fn merge_build_manifest_skips_refresh_when_carry_forward_stable_row_ids() {
        use crate::feature_flags::FLAG_STABLE_ROW_IDS;
        use crate::rowids::version::{RowDatasetVersionMeta, RowDatasetVersionSequence};
        use lance_file::version::LanceFileVersion;

        let data_file = DataFile::new(
            "same.lance",
            vec![0],
            vec![0],
            LanceFileVersion::Stable.resolve(),
            None,
            None,
        );

        let arrow_schema = ArrowSchema::new(vec![ArrowField::new("id", DataType::Int32, false)]);
        let lance_schema = LanceSchema::try_from(&arrow_schema).unwrap();

        let row_ids = RowIdSequence::from([200u64, 201, 202, 203, 204].as_slice());
        let row_id_meta = Some(RowIdMeta::Inline(write_row_ids(&row_ids).into()));

        let uniform_v1 = RowDatasetVersionSequence::from_uniform_row_count(5, 1);
        let meta_v1 = RowDatasetVersionMeta::from_sequence(&uniform_v1).unwrap();

        let prev_fragment = Fragment {
            id: 0,
            files: vec![data_file.clone()],
            overlays: vec![],
            deletion_file: None,
            row_id_meta: row_id_meta.clone(),
            physical_rows: Some(5),
            last_updated_at_version_meta: Some(meta_v1.clone()),
            created_at_version_meta: None,
        };

        let mut manifest = Manifest::new(
            lance_schema.clone(),
            Arc::new(vec![prev_fragment]),
            DataStorageFormat::new(ConcreteFileVersion::V2_0),
            HashMap::new(),
        );
        manifest.reader_feature_flags |= FLAG_STABLE_ROW_IDS;
        manifest.next_row_id = 100;

        let merged_fragment = Fragment {
            id: 0,
            files: vec![data_file],
            overlays: vec![],
            deletion_file: None,
            row_id_meta,
            physical_rows: Some(5),
            last_updated_at_version_meta: Some(meta_v1),
            created_at_version_meta: None,
        };

        let tx = Transaction::new(
            manifest.version,
            Operation::Merge {
                fragments: vec![merged_fragment],
                schema: lance_schema,
                preserves_nullability: true,
            },
            None,
        );

        let (out, _) = tx
            .build_manifest(Some(&manifest), vec![], "txn", &default_build_config())
            .unwrap();

        let seq = out.fragments[0]
            .last_updated_at_version_meta
            .as_ref()
            .unwrap()
            .load_sequence()
            .unwrap();
        assert_eq!(seq.version_at(0).unwrap(), 1);
        assert_eq!(seq.version_at(4).unwrap(), 1);
    }

    #[test]
    fn merge_build_manifest_no_last_updated_refresh_without_stable_row_ids() {
        use crate::feature_flags::FLAG_STABLE_ROW_IDS;
        use lance_file::version::LanceFileVersion;

        let mk_file = |path: &str| {
            DataFile::new(
                path,
                vec![0],
                vec![0],
                LanceFileVersion::Stable.resolve(),
                None,
                None,
            )
        };

        let arrow_schema = ArrowSchema::new(vec![ArrowField::new("id", DataType::Int32, false)]);
        let lance_schema = LanceSchema::try_from(&arrow_schema).unwrap();

        let prev_fragment = Fragment {
            id: 0,
            files: vec![mk_file("before.lance")],
            overlays: vec![],
            deletion_file: None,
            row_id_meta: None,
            physical_rows: Some(5),
            last_updated_at_version_meta: None,
            created_at_version_meta: None,
        };

        let manifest = Manifest::new(
            lance_schema.clone(),
            Arc::new(vec![prev_fragment.clone()]),
            DataStorageFormat::new(ConcreteFileVersion::V2_0),
            HashMap::new(),
        );
        assert_eq!(
            manifest.reader_feature_flags & FLAG_STABLE_ROW_IDS,
            0,
            "manifest must not use stable row IDs for this guard test"
        );

        let merged_fragment = Fragment {
            files: vec![mk_file("after.lance")],
            ..prev_fragment
        };

        let tx = Transaction::new(
            manifest.version,
            Operation::Merge {
                fragments: vec![merged_fragment],
                schema: lance_schema,
                preserves_nullability: true,
            },
            None,
        );

        let (out, _) = tx
            .build_manifest(Some(&manifest), vec![], "txn", &default_build_config())
            .unwrap();

        assert!(
            out.fragments[0].last_updated_at_version_meta.is_none(),
            "without stable row IDs, Merge must not populate per-row last_updated metadata"
        );
    }

    #[test]
    fn merge_build_manifest_sets_both_version_meta_for_new_fragment_id_stable_row_ids() {
        use crate::feature_flags::FLAG_STABLE_ROW_IDS;
        use lance_file::version::LanceFileVersion;

        let mk_file = |path: &str| {
            DataFile::new(
                path,
                vec![0],
                vec![0],
                LanceFileVersion::Stable.resolve(),
                None,
                None,
            )
        };

        let arrow_schema = ArrowSchema::new(vec![ArrowField::new("id", DataType::Int32, false)]);
        let lance_schema = LanceSchema::try_from(&arrow_schema).unwrap();

        // Existing fragment (id=0) with stable row IDs
        let row_ids_0 = RowIdSequence::from([10u64, 11, 12].as_slice());
        let existing_fragment = Fragment {
            id: 0,
            files: vec![mk_file("existing.lance")],
            overlays: vec![],
            deletion_file: None,
            row_id_meta: Some(RowIdMeta::Inline(write_row_ids(&row_ids_0).into())),
            physical_rows: Some(3),
            last_updated_at_version_meta: None,
            created_at_version_meta: None,
        };

        let mut manifest = Manifest::new(
            lance_schema.clone(),
            Arc::new(vec![existing_fragment.clone()]),
            DataStorageFormat::new(ConcreteFileVersion::V2_0),
            HashMap::new(),
        );
        manifest.reader_feature_flags |= FLAG_STABLE_ROW_IDS;
        manifest.next_row_id = 100;
        manifest.version = 1;

        // New fragment (id=1) not present in prev manifest — exercises the None branch
        let row_ids_1 = RowIdSequence::from([20u64, 21, 22, 23].as_slice());
        let new_fragment = Fragment {
            id: 1,
            files: vec![mk_file("new.lance")],
            overlays: vec![],
            deletion_file: None,
            row_id_meta: Some(RowIdMeta::Inline(write_row_ids(&row_ids_1).into())),
            physical_rows: Some(4),
            last_updated_at_version_meta: None,
            created_at_version_meta: None,
        };

        let tx = Transaction::new(
            manifest.version,
            Operation::Merge {
                fragments: vec![existing_fragment, new_fragment],
                schema: lance_schema,
                preserves_nullability: true,
            },
            None,
        );

        let (out, _) = tx
            .build_manifest(Some(&manifest), vec![], "txn", &default_build_config())
            .unwrap();

        assert_eq!(out.version, 2);

        let new_frag = out.fragments.iter().find(|f| f.id == 1).unwrap();

        // last_updated_at_version must be set to the commit version
        let last_updated_seq = new_frag
            .last_updated_at_version_meta
            .as_ref()
            .expect("new fragment must have last_updated_at_version_meta")
            .load_sequence()
            .unwrap();
        assert_eq!(last_updated_seq.version_at(0).unwrap(), 2);
        assert_eq!(last_updated_seq.version_at(3).unwrap(), 2);

        // created_at_version must also be set — must not be None
        let created_seq = new_frag
            .created_at_version_meta
            .as_ref()
            .expect("new fragment must have created_at_version_meta")
            .load_sequence()
            .unwrap();
        assert_eq!(created_seq.version_at(0).unwrap(), 2);
        assert_eq!(created_seq.version_at(3).unwrap(), 2);
    }

    #[test]
    fn test_data_overlay_build_manifest_multi_fragment() {
        // Overlays targeting two distinct fragments are each applied and stamped.
        // A targeted fragment already carrying an overlay (committed at v3) gets
        // the new overlay appended and stamped while its existing overlay is
        // preserved, and a fragment the operation does not target is passed
        // through with its existing overlays untouched.
        let mut frag0 = Fragment::new(0);
        frag0.overlays = vec![overlay_with_field(5, 3)]; // targeted, pre-existing at v3
        let frag1 = Fragment::new(1);
        let mut frag2 = Fragment::new(2);
        frag2.overlays = vec![overlay_with_field(9, 3)]; // untargeted, committed at v3
        let schema = ArrowSchema::new(vec![ArrowField::new("id", DataType::Int32, false)]);
        let mut manifest = Manifest::new(
            LanceSchema::try_from(&schema).unwrap(),
            Arc::new(vec![frag0, frag1, frag2]),
            crate::format::DataStorageFormat::new(ConcreteFileVersion::V2_0),
            HashMap::new(),
        );
        // The pre-existing overlays were committed at v3, so the current
        // manifest must be at least that version; the new commit then stamps
        // its overlay at v4, keeping the fragment's overlays newest-last.
        manifest.version = 3;

        let txn = Transaction::new(
            manifest.version,
            Operation::DataOverlay {
                groups: vec![
                    DataOverlayGroup {
                        fragment_id: 0,
                        overlays: vec![overlay_with_field(1, 0)],
                    },
                    DataOverlayGroup {
                        fragment_id: 1,
                        overlays: vec![overlay_with_field(2, 0)],
                    },
                ],
            },
            None,
        );

        let (result, _) = txn
            .build_manifest(Some(&manifest), vec![], "txn", &default_build_config())
            .unwrap();

        let frag = |id: u64| {
            result
                .fragments
                .iter()
                .find(|f| f.id == id)
                .unwrap_or_else(|| panic!("fragment {id} missing from result"))
        };
        // The already-overlaid target keeps its v3 overlay and appends the new
        // one, stamped to the new version.
        assert_eq!(frag(0).overlays.len(), 2);
        assert_eq!(frag(0).overlays[0].committed_version, 3);
        assert_eq!(frag(0).overlays[1].committed_version, result.version);
        // The fresh target gets its overlay, stamped to the new version.
        assert_eq!(frag(1).overlays.len(), 1);
        assert_eq!(frag(1).overlays[0].committed_version, result.version);
        // The untargeted fragment is unchanged: same overlay, original version.
        assert_eq!(frag(2).overlays.len(), 1);
        assert_eq!(frag(2).overlays[0].committed_version, 3);
        assert!(result.version > manifest.version);
    }

    #[test]
    fn test_data_overlay_build_manifest_updates_row_lineage() {
        let row_ids = RowIdSequence::from([10u64, 11, 12, 13].as_slice());
        let version_meta = RowDatasetVersionMeta::from_sequence(
            &RowDatasetVersionSequence::from_uniform_row_count(4, 1),
        )
        .unwrap();
        let fragment = Fragment {
            id: 1,
            files: vec![],
            overlays: vec![],
            deletion_file: None,
            row_id_meta: Some(RowIdMeta::Inline(write_row_ids(&row_ids).into())),
            physical_rows: Some(4),
            last_updated_at_version_meta: Some(version_meta.clone()),
            created_at_version_meta: Some(version_meta),
        };
        // Stable-row-id migrations can leave old fragments without explicit
        // lineage metadata; scans expose those rows as version 1.
        let untracked_row_ids = RowIdSequence::from([20u64, 21].as_slice());
        let untracked_fragment = Fragment {
            id: 2,
            files: vec![],
            overlays: vec![],
            deletion_file: None,
            row_id_meta: Some(RowIdMeta::Inline(write_row_ids(&untracked_row_ids).into())),
            physical_rows: Some(2),
            last_updated_at_version_meta: None,
            created_at_version_meta: None,
        };
        let manifest = make_stable_row_id_manifest(vec![fragment, untracked_fragment]);
        let txn = Transaction::new(
            manifest.version,
            Operation::DataOverlay {
                groups: vec![
                    DataOverlayGroup {
                        fragment_id: 1,
                        overlays: vec![
                            DataOverlayFile {
                                data_file: DataFile::new_legacy_from_fields(
                                    "dense.lance",
                                    vec![1],
                                    None,
                                ),
                                coverage: OverlayCoverage::dense(RoaringBitmap::from_iter([0u32])),
                                committed_version: 0,
                            },
                            DataOverlayFile {
                                data_file: DataFile::new_legacy_from_fields(
                                    "sparse.lance",
                                    vec![1, 2],
                                    None,
                                ),
                                coverage: OverlayCoverage::sparse(vec![
                                    RoaringBitmap::from_iter([2u32]),
                                    RoaringBitmap::from_iter([3u32]),
                                ]),
                                committed_version: 0,
                            },
                        ],
                    },
                    DataOverlayGroup {
                        fragment_id: 2,
                        overlays: vec![DataOverlayFile {
                            data_file: DataFile::new_legacy_from_fields(
                                "migrated.lance",
                                vec![1],
                                None,
                            ),
                            coverage: OverlayCoverage::dense(RoaringBitmap::from_iter([1u32])),
                            committed_version: 0,
                        }],
                    },
                ],
            },
            None,
        );

        let (result, _) = txn
            .build_manifest(Some(&manifest), vec![], "txn", &default_build_config())
            .unwrap();

        assert_eq!(last_updated_at_versions(&result, 1), vec![5, 1, 5, 5]);
        assert_eq!(last_updated_at_versions(&result, 2), vec![1, 5]);
        assert!(
            result.fragments[0]
                .overlays
                .iter()
                .all(|overlay| overlay.committed_version == result.version)
        );
    }

    #[test]
    fn test_data_replacement_tombstones_overlaid_fields() {
        // A DataReplacement writing new base values for field 5 must stop any
        // overlay already shadowing those cells: field 5 is tombstoned in place
        // (preserving the overlay's field 3), and an overlay covering only field
        // 5 is dropped entirely. Both overlays predate the transaction's read
        // version, which is what makes the replacement the newer value.
        let mut fragment = Fragment::new(0);
        fragment.files = vec![
            DataFile::new_legacy_from_fields("f3.lance", vec![3], None),
            DataFile::new_legacy_from_fields("f5.lance", vec![5], None),
        ];
        fragment.overlays = vec![
            DataOverlayFile {
                data_file: DataFile::new_legacy_from_fields("o35.lance", vec![3, 5], None),
                coverage: OverlayCoverage::sparse(vec![
                    roaring::RoaringBitmap::from_iter([0u32]),
                    roaring::RoaringBitmap::from_iter([0u32]),
                ]),
                committed_version: 1,
            },
            DataOverlayFile {
                data_file: DataFile::new_legacy_from_fields("o5.lance", vec![5], None),
                coverage: OverlayCoverage::dense(roaring::RoaringBitmap::from_iter([0u32])),
                committed_version: 1,
            },
        ];

        let schema = ArrowSchema::new(vec![ArrowField::new("id", DataType::Int32, false)]);
        let manifest = Manifest::new(
            LanceSchema::try_from(&schema).unwrap(),
            Arc::new(vec![fragment]),
            crate::format::DataStorageFormat::new(ConcreteFileVersion::V2_0),
            HashMap::new(),
        );

        let txn = Transaction::new(
            manifest.version,
            Operation::DataReplacement {
                replacements: vec![DataReplacementGroup(
                    0,
                    DataFile::new_legacy_from_fields("f5-new.lance", vec![5], None),
                )],
            },
            None,
        );

        let (result, _) = txn
            .build_manifest(Some(&manifest), vec![], "txn", &default_build_config())
            .unwrap();

        let frag = &result.fragments[0];
        // The base data file for field 5 was swapped in.
        assert!(frag.files.iter().any(|f| f.path == "f5-new.lance"));
        // The [3, 5] overlay keeps field 3 and tombstones field 5; the [5]-only
        // overlay is dropped.
        assert_eq!(frag.overlays.len(), 1);
        assert_eq!(frag.overlays[0].data_file.fields.as_ref(), &[3, -2]);
    }

    /// Replace `fields` in `fragment` at `read_version`, against a manifest
    /// at `manifest_version` whose schema declares field ids 3 ("x"), 4 ("a"),
    /// 5 ("v") and 6 ("y").
    fn replace_fields(
        fragment: Fragment,
        fields: Vec<i32>,
        manifest_version: u64,
        read_version: u64,
    ) -> Result<Fragment> {
        let schema = ArrowSchema::new(vec![
            ArrowField::new("x", DataType::Int32, true),
            ArrowField::new("a", DataType::Int32, true),
            ArrowField::new("v", DataType::Int32, true),
            ArrowField::new("y", DataType::Int32, true),
        ]);
        let mut lance_schema = LanceSchema::try_from(&schema).unwrap();
        lance_schema.fields[0].id = 3;
        lance_schema.fields[1].id = 4;
        lance_schema.fields[2].id = 5;
        lance_schema.fields[3].id = 6;
        let mut manifest = Manifest::new(
            lance_schema,
            Arc::new(vec![fragment]),
            crate::format::DataStorageFormat::new(ConcreteFileVersion::V2_0),
            HashMap::new(),
        );
        manifest.version = manifest_version;

        let column_indices = (0..fields.len() as i32).collect();
        let txn = Transaction::new(
            read_version,
            Operation::DataReplacement {
                replacements: vec![DataReplacementGroup(
                    0,
                    DataFile::new(
                        "v-new.lance",
                        fields,
                        column_indices,
                        ConcreteFileVersion::V2_0,
                        None,
                        None,
                    ),
                )],
            },
            None,
        );
        txn.build_manifest(Some(&manifest), vec![], "txn", &default_build_config())
            .map(|(manifest, _)| manifest.fragments[0].clone())
    }

    /// Replace field 5 in `fragment` at `read_version`, against a manifest at
    /// `manifest_version`.
    fn replace_field_5(
        fragment: Fragment,
        manifest_version: u64,
        read_version: u64,
    ) -> Result<Fragment> {
        replace_fields(fragment, vec![5], manifest_version, read_version)
    }

    #[test]
    fn test_data_replacement_rejects_subset_of_legacy_file() {
        // The V1 reader derives its page table offset from the first field in
        // the file metadata, so turning `[4, 5]` into `[-2, 5]` would leave
        // field 4 decoding from field 5's pages. With no exact match to swap,
        // the replacement must be rejected rather than corrupting the sibling.
        let mut fragment = Fragment::new(0);
        fragment.files = vec![DataFile::new_legacy_from_fields(
            "wide.lance",
            vec![4, 5],
            None,
        )];

        let result = replace_field_5(fragment, 1, 1);
        assert!(
            result.is_err(),
            "legacy subset replacement must be rejected, got: {:?}",
            result.map(|fragment| fragment.files)
        );
    }

    #[test]
    fn test_data_replacement_tombstones_fields_spanning_files() {
        // The replaced fields sit in two different wider files. Each file is
        // tombstoned for the field it holds and survives on its remaining
        // live one, with the new file answering for both.
        let mut fragment = Fragment::new(0);
        fragment.files = vec![
            DataFile::new(
                "ab.lance",
                vec![3, 4],
                vec![0, 1],
                ConcreteFileVersion::V2_0,
                None,
                None,
            ),
            DataFile::new(
                "cd.lance",
                vec![5, 6],
                vec![0, 1],
                ConcreteFileVersion::V2_0,
                None,
                None,
            ),
        ];

        let fragment = replace_fields(fragment, vec![4, 5], 1, 1).unwrap();
        let file = |path| {
            fragment
                .files
                .iter()
                .find(|file| file.path == path)
                .unwrap_or_else(|| panic!("{path} survives on its live field"))
        };
        assert_eq!(file("ab.lance").fields.as_ref(), &[3, TOMBSTONE_FIELD_ID]);
        assert_eq!(file("cd.lance").fields.as_ref(), &[TOMBSTONE_FIELD_ID, 6]);
        assert!(fragment.files.iter().any(|file| file.path == "v-new.lance"));
    }

    #[test]
    fn test_data_replacement_rejects_fields_spanning_a_legacy_file() {
        // Spanning is only resolvable while every covering file can be
        // tombstoned. A V1 file holding one of the replaced fields cannot,
        // so the replacement must be rejected rather than half applied.
        let mut fragment = Fragment::new(0);
        fragment.files = vec![
            DataFile::new(
                "ab.lance",
                vec![3, 4],
                vec![0, 1],
                ConcreteFileVersion::V2_0,
                None,
                None,
            ),
            DataFile::new_legacy_from_fields("cd.lance", vec![5, 6], None),
        ];

        let result = replace_fields(fragment, vec![4, 5], 1, 1);
        assert!(
            result.is_err(),
            "spanning a legacy file must be rejected, got: {:?}",
            result.map(|fragment| fragment.files)
        );
    }

    #[test]
    fn test_data_replacement_retombstones_wider_file() {
        // A wider file carrying a tombstone from an earlier round is
        // tombstoned again for the newly replaced field and survives on its
        // remaining live field.
        let mut fragment = Fragment::new(0);
        fragment.files = vec![DataFile::new(
            "wide.lance",
            vec![4, TOMBSTONE_FIELD_ID, 5],
            vec![0, 1, 2],
            ConcreteFileVersion::V2_0,
            None,
            None,
        )];

        let fragment = replace_fields(fragment, vec![5], 1, 1).unwrap();
        let wide = fragment
            .files
            .iter()
            .find(|file| file.path == "wide.lance")
            .expect("wider file survives on its live field");
        assert_eq!(
            wide.fields.as_ref(),
            &[4, TOMBSTONE_FIELD_ID, TOMBSTONE_FIELD_ID]
        );
        assert!(fragment.files.iter().any(|file| file.path == "v-new.lance"));
    }

    #[test]
    fn test_data_replacement_preserves_overlay_newer_than_snapshot() {
        // An overlay committed after this transaction read its snapshot holds
        // the newer value; the conflict resolver rebases the two precisely
        // because the overlay wins. Tombstoning it would discard a committed
        // write, so only overlays the transaction could have seen are superseded.
        let mut fragment = Fragment::new(0);
        // One wider file, so the replacement takes the tombstone-and-append path.
        fragment.files = vec![DataFile::new(
            "wide.lance",
            vec![4, 5],
            vec![0, 1],
            ConcreteFileVersion::V2_0,
            None,
            None,
        )];
        fragment.overlays = vec![DataOverlayFile {
            data_file: DataFile::new(
                "newer.lance",
                vec![5],
                vec![0],
                ConcreteFileVersion::V2_0,
                None,
                None,
            ),
            coverage: OverlayCoverage::dense(roaring::RoaringBitmap::from_iter([0u32])),
            committed_version: 7,
        }];

        // Staged against version 6, i.e. before the overlay landed.
        let fragment = replace_field_5(fragment, 7, 6).unwrap();
        assert!(fragment.files.iter().any(|f| f.path == "v-new.lance"));
        assert_eq!(
            fragment.overlays.len(),
            1,
            "overlay committed after the snapshot must survive"
        );
        assert_eq!(fragment.overlays[0].data_file.fields.as_ref(), &[5]);
    }

    /// A projection keeps only the data files that still share a field with
    /// the schema. The reserved ids of spilled row lineage are never in it, so
    /// without an exception the file carrying them, their only copy, would go.
    #[test]
    fn project_keeps_file_carrying_spilled_row_lineage() {
        let mut fragment = Fragment::new(0);
        fragment.physical_rows = Some(4);
        fragment.row_id_meta = Some(RowIdMeta::Column);
        fragment.created_at_version_meta = Some(RowDatasetVersionMeta::Column);
        fragment.files = vec![
            DataFile::new(
                "d.lance",
                vec![0, 1],
                vec![0, 1],
                ConcreteFileVersion::V2_0,
                None,
                None,
            ),
            DataFile::new(
                "l.lance",
                vec![ROW_ID_FIELD_ID, ROW_CREATED_AT_VERSION_FIELD_ID],
                vec![0, 1],
                ConcreteFileVersion::V2_0,
                None,
                None,
            ),
        ];
        let schema = ArrowSchema::new(vec![
            ArrowField::new("a", DataType::Int32, false),
            ArrowField::new("b", DataType::Int32, false),
        ]);
        let mut manifest = Manifest::new(
            LanceSchema::try_from(&schema).unwrap(),
            Arc::new(vec![fragment]),
            DataStorageFormat::new(ConcreteFileVersion::V2_0),
            HashMap::new(),
        );
        manifest.reader_feature_flags = FLAG_STABLE_ROW_IDS;
        manifest.writer_feature_flags = FLAG_STABLE_ROW_IDS;

        // Drop field 1, as `drop_columns(["b"])` would.
        let transaction = Transaction::new(
            manifest.version,
            Operation::Project {
                schema: manifest.schema.project_by_ids(&[0], true),
                preserves_nullability: true,
            },
            None,
        );
        let (result, _) = transaction
            .build_manifest(Some(&manifest), vec![], "txn", &default_build_config())
            .unwrap();

        let fragment = &result.fragments[0];
        let paths = fragment
            .files
            .iter()
            .map(|file| file.path.as_str())
            .collect::<Vec<_>>();
        assert_eq!(paths, ["d.lance", "l.lance"]);
        for field_id in [ROW_ID_FIELD_ID, ROW_CREATED_AT_VERSION_FIELD_ID] {
            assert_eq!(
                fragment
                    .row_lineage_file(field_id)
                    .unwrap()
                    .map(|file| file.path.as_str()),
                Some("l.lance"),
                "field {field_id}"
            );
        }
    }

    /// Replacing part of a wider file tombstones the replaced fields where
    /// they live and drops any file left without a schema field. The file
    /// carrying the fragment's spilled row ids may have none, yet it is their
    /// only copy and must stay next to the new file that answers for the user
    /// fields. The carrier is either a file of its own, as an update writes
    /// it, or a file that also holds user columns, all of which are replaced.
    #[rstest::rstest]
    #[case::lineage_only_file(
        vec![("v.lance", vec![3, 4, 5, 6]), ("l.lance", vec![ROW_ID_FIELD_ID])],
        vec![3, 4],
        vec![
            ("v.lance", vec![TOMBSTONE_FIELD_ID, TOMBSTONE_FIELD_ID, 5, 6]),
            ("l.lance", vec![ROW_ID_FIELD_ID]),
            ("v-new.lance", vec![3, 4]),
        ],
        "l.lance"
    )]
    #[case::in_file_carrier(
        vec![("v.lance", vec![3, 4, 5, 6, ROW_ID_FIELD_ID])],
        vec![3, 4, 5, 6],
        vec![
            (
                "v.lance",
                vec![
                    TOMBSTONE_FIELD_ID,
                    TOMBSTONE_FIELD_ID,
                    TOMBSTONE_FIELD_ID,
                    TOMBSTONE_FIELD_ID,
                    ROW_ID_FIELD_ID,
                ],
            ),
            ("v-new.lance", vec![3, 4, 5, 6]),
        ],
        "v.lance"
    )]
    fn data_replacement_keeps_lineage_carrier(
        #[case] files: Vec<(&str, Vec<i32>)>,
        #[case] replaced_fields: Vec<i32>,
        #[case] expected_files: Vec<(&str, Vec<i32>)>,
        #[case] carrier: &str,
    ) {
        let mut fragment = Fragment::new(0);
        fragment.row_id_meta = Some(RowIdMeta::Column);
        fragment.files = files
            .into_iter()
            .map(|(path, fields)| {
                let column_indices = (0..fields.len() as i32).collect();
                DataFile::new(
                    path,
                    fields,
                    column_indices,
                    ConcreteFileVersion::V2_0,
                    None,
                    None,
                )
            })
            .collect();

        let fragment = replace_fields(fragment, replaced_fields, 1, 1).unwrap();
        let files = fragment
            .files
            .iter()
            .map(|file| (file.path.as_str(), file.fields.to_vec()))
            .collect::<Vec<_>>();
        assert_eq!(files, expected_files);
        assert_eq!(
            fragment
                .row_lineage_file(ROW_ID_FIELD_ID)
                .unwrap()
                .map(|file| file.path.as_str()),
            Some(carrier)
        );
    }

    /// Whatever hands the commit a fragment whose spilled row ids have no v2
    /// carrier -- an operation that dropped the file, or a writer that never
    /// attached it -- the commit refuses it rather than publish lineage that
    /// no reader can load.
    #[rstest::rstest]
    #[case::no_carrier(vec![0], ConcreteFileVersion::V2_0)]
    #[case::v1_carrier(vec![0, ROW_ID_FIELD_ID], ConcreteFileVersion::V1)]
    fn build_manifest_rejects_spilled_arm_without_carrier(
        #[case] fields: Vec<i32>,
        #[case] version: ConcreteFileVersion,
    ) {
        let manifest = sample_manifest();
        let mut fragment = Fragment::new(0);
        fragment.physical_rows = Some(4);
        fragment.row_id_meta = Some(RowIdMeta::Column);
        // Column indices do not matter to the commit, only ids and version.
        let data_file = DataFile::new("d.lance", fields, vec![], version, None, None);
        fragment.files = vec![data_file];
        let transaction = Transaction::new(
            manifest.version,
            Operation::Merge {
                fragments: vec![fragment],
                schema: manifest.schema.clone(),
                preserves_nullability: true,
            },
            None,
        );

        let err = transaction
            .build_manifest(Some(&manifest), vec![], "txn", &default_build_config())
            .unwrap_err();
        assert!(matches!(err, Error::Internal { .. }), "{err:?}");
        let message = err.to_string();
        assert!(
            message.contains("fragment 0") && message.contains(&ROW_ID_FIELD_ID.to_string()),
            "{message}"
        );
    }

    #[test]
    fn test_data_overlay_build_manifest_merges_duplicate_groups() {
        // Two groups targeting the same fragment must both survive (a HashMap
        // collapse would have dropped the first).
        let manifest = sample_manifest();
        let txn = Transaction::new(
            manifest.version,
            Operation::DataOverlay {
                groups: vec![
                    DataOverlayGroup {
                        fragment_id: 0,
                        overlays: vec![overlay_with_field(1, 0)],
                    },
                    DataOverlayGroup {
                        fragment_id: 0,
                        overlays: vec![overlay_with_field(2, 0)],
                    },
                ],
            },
            None,
        );

        let (result, _) = txn
            .build_manifest(Some(&manifest), vec![], "txn", &default_build_config())
            .unwrap();

        let overlays = &result.fragments[0].overlays;
        assert_eq!(overlays.len(), 2);
        assert_eq!(overlays[0].data_file.fields.as_ref(), [1i32].as_slice());
        assert_eq!(overlays[1].data_file.fields.as_ref(), [2i32].as_slice());
    }

    #[test]
    fn test_data_overlay_build_manifest_rejects_unknown_fragment() {
        let manifest = sample_manifest();
        let txn = Transaction::new(
            manifest.version,
            Operation::DataOverlay {
                groups: vec![DataOverlayGroup {
                    fragment_id: 99,
                    overlays: vec![overlay_with_field(1, 0)],
                }],
            },
            None,
        );
        let err = txn
            .build_manifest(Some(&manifest), vec![], "txn", &default_build_config())
            .unwrap_err();
        assert!(err.to_string().contains("does not exist"), "{err}");
    }

    #[test]
    fn test_nullability_assertion_defaults_conservative() {
        // A writer that predates the field encodes nothing, which decodes as
        // false: no assertion, so a legacy tightening or required-field merge
        // still conflicts. Only an explicit true skips the barrier.
        for encoded in [false, true] {
            let txn = Transaction::try_from(pb::Transaction {
                read_version: 1,
                uuid: "test".to_string(),
                operation: Some(pb::transaction::Operation::Project(
                    pb::transaction::Project {
                        schema: vec![],
                        preserves_nullability: encoded,
                    },
                )),
                ..Default::default()
            })
            .unwrap();
            assert!(
                matches!(txn.operation, Operation::Project { preserves_nullability, .. } if preserves_nullability == encoded),
                "encoded={encoded:?}"
            );

            let txn = Transaction::try_from(pb::Transaction {
                read_version: 1,
                uuid: "test".to_string(),
                operation: Some(pb::transaction::Operation::Merge(pb::transaction::Merge {
                    fragments: vec![],
                    schema: vec![],
                    schema_metadata: Default::default(),
                    preserves_nullability: encoded,
                })),
                ..Default::default()
            })
            .unwrap();
            assert!(
                matches!(txn.operation, Operation::Merge { preserves_nullability, .. } if preserves_nullability == encoded),
                "encoded={encoded:?}"
            );
        }
    }

    mod mem_wal_index_coverage {
        use super::*;
        use crate::system_index::mem_wal::{
            CompactedSsTable, IndexCatchupProgress, MEM_WAL_INDEX_NAME, MemWalIndexDetails,
        };

        fn user_index(name: &str, uuid: Uuid, frags: &[u32]) -> IndexMetadata {
            IndexMetadata {
                uuid,
                name: name.to_string(),
                fields: vec![0],
                covering_fields: vec![],
                dataset_version: 1,
                fragment_bitmap: Some(RoaringBitmap::from_iter(frags.iter().copied())),
                index_details: None,
                index_version: 0,
                created_at: None,
                base_id: None,
                files: None,
            }
        }

        fn mem_wal_index(details: MemWalIndexDetails) -> IndexMetadata {
            crate::system_index::mem_wal::new_mem_wal_index_meta(1, details).unwrap()
        }

        fn coverage_for(indices: &[IndexMetadata], name: &str) -> Option<Vec<CompactedSsTable>> {
            let meta = indices
                .iter()
                .find(|idx| idx.name == MEM_WAL_INDEX_NAME)
                .expect("mem wal index present");
            load_mem_wal_index_details(meta.clone())
                .unwrap()
                .index_catchup
                .into_iter()
                .find(|entry| entry.index_name == name)
                .map(|entry| entry.caught_up_generations)
        }

        fn compacted(shard: Uuid, generation: u64) -> Vec<CompactedSsTable> {
            vec![CompactedSsTable::new(shard, generation)]
        }

        /// A manifest carrying exactly `frags`, standing in for the version a
        /// transaction read.
        fn manifest_with(frags: &[u32]) -> Manifest {
            let fragments: Vec<Fragment> =
                frags.iter().map(|id| Fragment::new(*id as u64)).collect();
            Manifest::new(
                LanceSchema::default(),
                Arc::new(fragments),
                DataStorageFormat::default(),
                Default::default(),
            )
        }

        /// Drives the production path, so these exercise the real derivation.
        fn apply(
            after: &mut [IndexMetadata],
            before: &[IndexMetadata],
            read_frags: &[u32],
            read_indices: &[IndexMetadata],
        ) -> Result<()> {
            let manifest = manifest_with(read_frags);
            let segments_before = Transaction::logical_index_segments(before);
            Transaction::apply_mem_wal_index_coverage(
                after,
                &segments_before,
                Some(ReadVersionState {
                    manifest: &manifest,
                    indices: read_indices,
                }),
                2,
            )
        }

        fn table(idx_frags: &[u32], uuid: Uuid, details: MemWalIndexDetails) -> Vec<IndexMetadata> {
            vec![user_index("idx", uuid, idx_frags), mem_wal_index(details)]
        }

        fn progress(shard: Uuid, generation: u64) -> MemWalIndexDetails {
            MemWalIndexDetails {
                compacted_sstables: compacted(shard, generation),
                ..Default::default()
            }
        }

        fn progress_with_catchup(shard: Uuid, generation: u64, caught: u64) -> MemWalIndexDetails {
            MemWalIndexDetails {
                compacted_sstables: compacted(shard, generation),
                index_catchup: vec![IndexCatchupProgress::new(
                    "idx".to_string(),
                    compacted(shard, caught),
                )],
                ..Default::default()
            }
        }

        /// Through the builder, the "before" side of the coverage comparison
        /// is the original list as read from the manifest, never the
        /// prepared one: an in-place rewrite that narrows an index's
        /// coverage in preparation withdraws its catch-up record instead of
        /// carrying it over as if nothing changed.
        #[test]
        fn a_withdrawal_in_preparation_withdraws_the_catch_up_record() {
            let shard = Uuid::new_v4();
            let uuid = Uuid::new_v4();
            let read = table(&[0, 1], uuid, progress_with_catchup(shard, 5, 5));
            // A schema with field 0, so the index on it is retained.
            let mut manifest = crate::transaction::test_support::sample_manifest();
            manifest.fragments = Arc::new(vec![Fragment::new(0), Fragment::new(1)]);
            let mut update = crate::transaction::test_support::update_txn(vec![]);
            let Operation::Update {
                updated_fragments,
                fields_modified,
                ..
            } = &mut update.operation
            else {
                unreachable!()
            };
            *updated_fragments = vec![Fragment::new(0)];
            *fields_modified = vec![0];
            let transaction = Transaction::new(manifest.version, update.operation, None);
            let (_, after) = transaction
                .build_manifest_with_read_version(
                    Some(&manifest),
                    read.clone(),
                    "txn",
                    &default_build_config(),
                    Some(ReadVersionState {
                        manifest: &manifest,
                        indices: &read,
                    }),
                )
                .unwrap();
            let idx = after.iter().find(|idx| idx.name == "idx").unwrap();
            assert_eq!(
                idx.fragment_bitmap,
                Some(RoaringBitmap::from_iter([1u32])),
                "the rewritten fragment left the coverage in preparation"
            );
            assert_eq!(
                coverage_for(&after, "idx"),
                None,
                "a record derived over the original coverage is not carried"
            );
        }

        /// An index spanning every fragment the transaction read is credited
        /// with what that version had compacted.
        #[test]
        fn an_index_covering_the_read_version_is_credited() {
            let shard = Uuid::new_v4();
            let read = table(&[0, 1], Uuid::new_v4(), progress(shard, 5));
            let mut after = table(&[0, 1], Uuid::new_v4(), progress(shard, 5));
            apply(&mut after, &read, &[0, 1], &read).unwrap();
            assert_eq!(coverage_for(&after, "idx"), Some(compacted(shard, 5)));
        }

        /// An index short of the read version proves nothing, so it gets no
        /// entry -- absence reads as "not caught up".
        #[test]
        fn an_index_short_of_the_read_version_is_not_credited() {
            let shard = Uuid::new_v4();
            let read = table(&[0], Uuid::new_v4(), progress(shard, 5));
            let mut after = table(&[0], Uuid::new_v4(), progress(shard, 5));
            apply(&mut after, &read, &[0, 1], &read).unwrap();
            assert_eq!(coverage_for(&after, "idx"), None);
        }

        /// The hazard that makes the comparison use whole metadata.
        ///
        /// `Operation::Update` prunes a segment's fragment bitmap in place when
        /// it touches an indexed field, keeping the same UUID. A UUID-only
        /// "unchanged" test carries the old position forward while the index
        /// covers fewer fragments, and the WAL pod then trims on a position the
        /// index no longer earns. Reachable from the ordinary SSTable merge.
        #[test]
        fn a_bitmap_pruned_in_place_does_not_keep_its_position() {
            let shard = Uuid::new_v4();
            let uuid = Uuid::new_v4();
            let before = table(&[0, 1], uuid, progress_with_catchup(shard, 5, 5));
            // Same UUID, fragment 1 pruned away.
            let mut after = table(&[0], uuid, progress_with_catchup(shard, 5, 5));
            apply(&mut after, &before, &[0, 1], &before).unwrap();
            assert_eq!(
                coverage_for(&after, "idx"),
                None,
                "a shrunken index kept a position it no longer earns"
            );
        }

        /// Carrying a position forward is not the same as extending it. An
        /// index that has not moved still only holds the generations it caught
        /// up to; the compaction that has landed since is in fragments it does
        /// not span.
        #[test]
        fn an_unchanged_index_is_not_raised_beyond_what_it_proves() {
            let shard = Uuid::new_v4();
            let uuid = Uuid::new_v4();
            // Recorded at generation 2; generation 5 has since been folded in.
            let before = table(&[0], uuid, progress_with_catchup(shard, 5, 2));
            let mut after = before.clone();
            // Fragment 1 arrived with that compaction and this index lacks it.
            apply(&mut after, &before, &[0, 1], &before).unwrap();
            assert_eq!(coverage_for(&after, "idx"), Some(compacted(shard, 2)));
        }

        /// A recorded position above what this commit says was compacted is
        /// clamped down. Nothing should produce one, but a position the base
        /// table cannot back would retire SSTables whose rows are nowhere.
        #[test]
        fn a_carried_position_cannot_exceed_the_committed_progress() {
            let shard = Uuid::new_v4();
            let uuid = Uuid::new_v4();
            let before = table(&[0], uuid, progress_with_catchup(shard, 3, 9));
            let mut after = before.clone();
            apply(&mut after, &before, &[0], &before).unwrap();
            assert_eq!(coverage_for(&after, "idx"), Some(compacted(shard, 3)));
        }

        /// An unchanged index keeps what it recorded even when this commit's
        /// own snapshot cannot prove as much.
        #[test]
        fn an_unchanged_index_is_never_lowered() {
            let shard = Uuid::new_v4();
            let uuid = Uuid::new_v4();
            let before = table(&[0], uuid, progress_with_catchup(shard, 9, 9));
            let mut after = before.clone();
            apply(&mut after, &before, &[0, 1], &before).unwrap();
            assert_eq!(coverage_for(&after, "idx"), Some(compacted(shard, 9)));
        }

        /// Credit never exceeds what this commit records as compacted, so a
        /// read version since rolled back cannot retire SSTables no live commit
        /// copied in.
        #[test]
        fn credit_is_capped_by_the_committed_progress() {
            let shard = Uuid::new_v4();
            let read = table(&[0], Uuid::new_v4(), progress(shard, 9));
            let mut after = table(&[0], Uuid::new_v4(), progress(shard, 3));
            apply(&mut after, &read, &[0], &read).unwrap();
            assert_eq!(coverage_for(&after, "idx"), Some(compacted(shard, 3)));
        }

        /// The cap is the read version's progress, not this commit's. A
        /// compaction that landed while the index was being built put its rows
        /// in fragments this transaction never inspected, so covering
        /// everything it *did* read earns only what had been folded in by then.
        #[test]
        fn credit_never_reaches_past_the_read_version() {
            let shard = Uuid::new_v4();
            // Read at generation 2; generation 5 landed while this ran.
            let read = table(&[0], Uuid::new_v4(), progress(shard, 2));
            let mut after = table(&[0], Uuid::new_v4(), progress(shard, 5));
            apply(&mut after, &read, &[0], &read).unwrap();
            assert_eq!(coverage_for(&after, "idx"), Some(compacted(shard, 2)));
        }

        /// One segment with an unknown bitmap makes the whole index unproven,
        /// even when its siblings happen to span everything. Coverage that
        /// cannot be read is not coverage that can be relied on.
        #[test]
        fn an_index_with_an_unknown_segment_is_not_credited() {
            let shard = Uuid::new_v4();
            let mut unknown = user_index("idx", Uuid::new_v4(), &[]);
            unknown.fragment_bitmap = None;
            let read = vec![
                user_index("idx", Uuid::new_v4(), &[0, 1]),
                unknown,
                mem_wal_index(progress(shard, 5)),
            ];
            let mut after = read.clone();
            apply(&mut after, &read, &[0, 1], &read).unwrap();
            assert_eq!(coverage_for(&after, "idx"), None);
        }

        /// A dropped index has no coverage left to gate anything.
        #[test]
        fn a_dropped_index_loses_its_entry() {
            let shard = Uuid::new_v4();
            let before = table(&[0], Uuid::new_v4(), progress_with_catchup(shard, 5, 5));
            let mut after = vec![mem_wal_index(progress_with_catchup(shard, 5, 5))];
            apply(&mut after, &before, &[0], &before).unwrap();
            assert_eq!(coverage_for(&after, "idx"), None);
        }

        /// An index created by this commit is credited if it spans the read
        /// version -- it was built over those fragments, so it holds their
        /// rows. This is what the advance model could not express: an ordinary
        /// build that fully covers had to throw its work away and wait.
        #[test]
        fn a_new_index_covering_the_read_version_is_credited() {
            let shard = Uuid::new_v4();
            let before = vec![mem_wal_index(progress(shard, 5))];
            let mut after = table(&[0], Uuid::new_v4(), progress(shard, 5));
            // Covers the read version, but was not there when it was read.
            apply(&mut after, &before, &[0], &before).unwrap();
            assert_eq!(coverage_for(&after, "idx"), Some(compacted(shard, 5)));
        }

        /// A table carrying compaction progress but no catch-up entry earns one
        /// from an ordinary commit. This is how a table written before catch-up
        /// was maintained heals itself: nothing has to be run against it.
        #[test]
        fn a_table_with_no_catchup_entry_earns_one() {
            let shard = Uuid::new_v4();
            let before = table(&[0], Uuid::new_v4(), progress(shard, 5));
            let mut after = before.clone();
            apply(&mut after, &before, &[0], &before).unwrap();
            assert_eq!(coverage_for(&after, "idx"), Some(compacted(shard, 5)));
        }

        /// Two shards, only one of them compacted.
        #[test]
        fn each_shard_is_credited_independently() {
            let merged = Uuid::new_v4();
            let idle = Uuid::new_v4();
            let details = MemWalIndexDetails {
                compacted_sstables: vec![
                    CompactedSsTable::new(merged, 4),
                    CompactedSsTable::new(idle, 0),
                ],
                ..Default::default()
            };
            let read = table(&[0], Uuid::new_v4(), details.clone());
            let mut after = table(&[0], Uuid::new_v4(), details);
            apply(&mut after, &read, &[0], &read).unwrap();
            let coverage = coverage_for(&after, "idx").expect("credited");
            assert_eq!(
                coverage
                    .iter()
                    .find(|g| g.shard_id == merged)
                    .map(|g| g.generation),
                Some(4)
            );
            assert_eq!(
                coverage
                    .iter()
                    .find(|g| g.shard_id == idle)
                    .map(|g| g.generation),
                Some(0)
            );
        }

        /// Two indexes advance independently: one covering, one behind.
        #[test]
        fn indexes_are_credited_independently() {
            let shard = Uuid::new_v4();
            let read = vec![
                user_index("fast", Uuid::new_v4(), &[0, 1]),
                user_index("slow", Uuid::new_v4(), &[0]),
                mem_wal_index(progress(shard, 6)),
            ];
            let mut after = read.clone();
            apply(&mut after, &read, &[0, 1], &read).unwrap();
            assert_eq!(coverage_for(&after, "fast"), Some(compacted(shard, 6)));
            assert_eq!(coverage_for(&after, "slow"), None);
        }

        /// An index whose coverage is unknown cannot be shown to cover anything.
        #[test]
        fn an_index_without_a_bitmap_is_not_credited() {
            let shard = Uuid::new_v4();
            let mut idx = user_index("idx", Uuid::new_v4(), &[0]);
            idx.fragment_bitmap = None;
            let read = vec![idx, mem_wal_index(progress(shard, 5))];
            let mut after = read.clone();
            apply(&mut after, &read, &[0], &read).unwrap();
            assert_eq!(coverage_for(&after, "idx"), None);
        }

        /// Nothing compacted means nothing to be behind on.
        #[test]
        fn no_compaction_progress_writes_no_entries() {
            let before = table(&[0], Uuid::new_v4(), MemWalIndexDetails::default());
            let mut after = before.clone();
            let untouched = after.clone();
            apply(&mut after, &before, &[0], &before).unwrap();
            assert_eq!(after, untouched);
        }

        /// No MemWAL system index: nothing to maintain, and no error.
        #[test]
        fn a_table_without_mem_wal_is_a_no_op() {
            let before = vec![user_index("idx", Uuid::new_v4(), &[0])];
            let mut after = before.clone();
            let untouched = after.clone();
            apply(&mut after, &before, &[0], &before).unwrap();
            assert_eq!(after, untouched);
        }

        /// No read version -- dataset creation, detached commits -- credits
        /// nothing and lowers nothing.
        #[test]
        fn without_a_read_version_nothing_changes() {
            let shard = Uuid::new_v4();
            let uuid = Uuid::new_v4();
            let before = table(&[0], uuid, progress_with_catchup(shard, 5, 5));
            let mut after = before.clone();
            let segments_before = Transaction::logical_index_segments(&before);
            Transaction::apply_mem_wal_index_coverage(&mut after, &segments_before, None, 2)
                .unwrap();
            assert_eq!(coverage_for(&after, "idx"), Some(compacted(shard, 5)));
        }

        /// An untrained index covers nothing that exists, so a sibling's work
        /// is no evidence for it.
        #[test]
        fn an_untrained_index_earns_nothing() {
            let shard = Uuid::new_v4();
            let read = vec![
                user_index("untrained", Uuid::new_v4(), &[]),
                user_index("trained", Uuid::new_v4(), &[0]),
                mem_wal_index(progress(shard, 10)),
            ];
            let mut after = read.clone();
            apply(&mut after, &read, &[0], &read).unwrap();
            assert_eq!(coverage_for(&after, "untrained"), None);
            assert_eq!(coverage_for(&after, "trained"), Some(compacted(shard, 10)));
        }

        /// Shards move independently within one index: one advances on this
        /// commit's proof while another keeps the position it already had.
        #[test]
        fn a_shard_keeps_its_position_while_another_advances() {
            let (advancing, quiet) = (Uuid::new_v4(), Uuid::new_v4());
            let uuid = Uuid::new_v4();
            let details = |advancing_gen: u64| MemWalIndexDetails {
                compacted_sstables: vec![
                    CompactedSsTable::new(advancing, advancing_gen),
                    CompactedSsTable::new(quiet, 10),
                ],
                index_catchup: vec![IndexCatchupProgress::new(
                    "idx".to_string(),
                    vec![CompactedSsTable::new(quiet, 7)],
                )],
                ..Default::default()
            };
            // The quiet shard was never compacted as of the read, so nothing
            // this commit proves reaches it -- it keeps its recorded 7.
            let read = vec![
                user_index("idx", uuid, &[0]),
                mem_wal_index(MemWalIndexDetails {
                    compacted_sstables: vec![CompactedSsTable::new(advancing, 9)],
                    ..details(9)
                }),
            ];
            let mut after = vec![user_index("idx", uuid, &[0]), mem_wal_index(details(10))];
            apply(&mut after, &read, &[0], &read).unwrap();

            let mut coverage = coverage_for(&after, "idx").expect("credited");
            coverage.sort_unstable_by_key(|sstable| sstable.shard_id);
            let mut expected = vec![
                CompactedSsTable::new(advancing, 9),
                CompactedSsTable::new(quiet, 7),
            ];
            expected.sort_unstable_by_key(|sstable| sstable.shard_id);
            assert_eq!(coverage, expected);
        }

        /// The derivation drops coverage an index no longer earns, but it never
        /// rejects the commit -- an ordinary index job must not be blocked by
        /// a protocol it knows nothing about.
        #[test]
        fn an_ordinary_index_job_is_never_blocked() {
            let shard = Uuid::new_v4();
            let before = table(&[0, 1], Uuid::new_v4(), progress_with_catchup(shard, 5, 5));
            // Rebuilt over a subset -- the shape a partial reindex leaves.
            let mut after = table(&[0], Uuid::new_v4(), progress_with_catchup(shard, 5, 5));
            apply(&mut after, &before, &[0, 1], &before).unwrap();
            assert_eq!(coverage_for(&after, "idx"), None);
        }

        /// A reader's rule is that a missing entry means "not caught up", so an
        /// index caught up to nothing must be absent rather than present at
        /// generation zero -- otherwise it reads as known-and-covered.
        #[test]
        fn an_index_caught_up_to_nothing_gets_no_entry() {
            let shard = Uuid::new_v4();
            let uuid = Uuid::new_v4();
            let before = table(&[0], uuid, progress_with_catchup(shard, 5, 0));
            let mut after = before.clone();
            // Does not span the read version, so nothing lifts it off zero.
            apply(&mut after, &before, &[0, 1], &before).unwrap();
            assert_eq!(coverage_for(&after, "idx"), None);
        }

        /// Each shard carries its own position. Collapsing them to one value
        /// would credit a lagging shard with a busier shard's progress.
        #[test]
        fn carried_positions_do_not_leak_between_shards() {
            let (ahead, behind) = (Uuid::new_v4(), Uuid::new_v4());
            let uuid = Uuid::new_v4();
            let details = MemWalIndexDetails {
                compacted_sstables: vec![
                    CompactedSsTable::new(ahead, 10),
                    CompactedSsTable::new(behind, 10),
                ],
                index_catchup: vec![IndexCatchupProgress::new(
                    "idx".to_string(),
                    vec![
                        CompactedSsTable::new(ahead, 8),
                        CompactedSsTable::new(behind, 2),
                    ],
                )],
                ..Default::default()
            };
            let before = vec![user_index("idx", uuid, &[0]), mem_wal_index(details)];
            let mut after = before.clone();
            // Unchanged and unproven: both shards keep exactly what they had.
            apply(&mut after, &before, &[0, 1], &before).unwrap();

            let mut coverage = coverage_for(&after, "idx").expect("carried");
            coverage.sort_unstable_by_key(|sstable| sstable.shard_id);
            let mut expected = vec![
                CompactedSsTable::new(ahead, 8),
                CompactedSsTable::new(behind, 2),
            ];
            expected.sort_unstable_by_key(|sstable| sstable.shard_id);
            assert_eq!(coverage, expected);
        }

        /// The derivation runs while the manifest is being built, but the
        /// index list is not final there: `migrate_indices` recalculates a
        /// segment's fragment bitmap and keeps its UUID. A position decided
        /// before that must not survive the narrowing, or the WAL pod trims
        /// against an index that no longer covers those rows.
        #[test]
        fn a_bitmap_narrowed_after_the_build_loses_its_position() {
            let shard = Uuid::new_v4();
            let uuid = Uuid::new_v4();
            // What migrate_indices leaves behind: same UUID, fewer fragments,
            // and it says so.
            let mut migrated = table(&[0], uuid, progress_with_catchup(shard, 5, 5));

            Transaction::withdraw_coverage_invalidated_after_build(
                &mut migrated,
                &["idx".to_string()],
                3,
            )
            .unwrap();

            assert_eq!(coverage_for(&migrated, "idx"), None);
        }

        /// Migration routinely fills in file lists and inferred details. Those
        /// do not change which rows an index answers for, so withdrawing on
        /// them would drop coverage every commit for no reason.
        #[test]
        fn metadata_migration_that_does_not_narrow_keeps_its_position() {
            let shard = Uuid::new_v4();
            let uuid = Uuid::new_v4();
            let mut migrated = table(&[0, 1], uuid, progress_with_catchup(shard, 5, 5));
            migrated[0].files = Some(Vec::new());
            migrated[0].created_at = Some(chrono::Utc::now());

            // Nothing narrowed, so migration reports nothing.
            Transaction::withdraw_coverage_invalidated_after_build(&mut migrated, &[], 3).unwrap();

            assert_eq!(coverage_for(&migrated, "idx"), Some(compacted(shard, 5)));
        }

        /// A commit that changes nothing must not churn the system index: a new
        /// UUID on every append would invalidate its cache entry fleet-wide.
        #[test]
        fn an_unchanged_commit_does_not_rewrite_the_system_index() {
            let shard = Uuid::new_v4();
            let uuid = Uuid::new_v4();
            let before = table(&[0], uuid, progress_with_catchup(shard, 5, 5));
            let mut after = before.clone();
            apply(&mut after, &before, &[0], &before).unwrap();

            let system_uuid = |indices: &[IndexMetadata]| {
                indices
                    .iter()
                    .find(|idx| idx.name == MEM_WAL_INDEX_NAME)
                    .unwrap()
                    .uuid
            };
            assert_eq!(system_uuid(&after), system_uuid(&before));
        }

        /// A commit with no read version still withdraws. It can prove nothing,
        /// so an index it changed keeps no position -- the alternative leaves a
        /// position describing an index that no longer exists.
        #[test]
        fn without_a_read_version_a_changed_index_still_loses_its_position() {
            let shard = Uuid::new_v4();
            let before = table(&[0, 1], Uuid::new_v4(), progress_with_catchup(shard, 5, 5));
            let mut after = table(&[0], Uuid::new_v4(), progress_with_catchup(shard, 5, 5));
            let segments_before = Transaction::logical_index_segments(&before);
            Transaction::apply_mem_wal_index_coverage(&mut after, &segments_before, None, 2)
                .unwrap();
            assert_eq!(coverage_for(&after, "idx"), None);
        }

        /// Two attempts against the same read version agree, which is what makes
        /// a rebase safe: `read_version` is fixed for a transaction's life.
        #[test]
        fn the_derivation_is_stable_across_attempts() {
            let shard = Uuid::new_v4();
            let read = table(&[0, 1], Uuid::new_v4(), progress(shard, 5));
            let mut first = table(&[0, 1], Uuid::new_v4(), progress(shard, 5));
            let mut second = first.clone();
            apply(&mut first, &read, &[0, 1], &read).unwrap();
            apply(&mut second, &read, &[0, 1], &read).unwrap();
            assert_eq!(coverage_for(&first, "idx"), coverage_for(&second, "idx"));
        }
    }
}