lance 14.0.1

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

//! A task to clean up a lance dataset, removing files that are no longer
//! needed.
//!
//! Currently we try and be rather conservative about what we delete.
//!
//! The following types of files may be deleted by the cleanup function:
//!
//! * Old manifest files - If a manifest file is older than the threshold
//!   and is not the latest manifest then it will be deleted.
//! * Unreferenced data files - If a data file is not referenced by any
//!   fragment in a valid manifest file then it will be deleted.
//! * Unreferenced delete files - If a delete file is not referenced by
//!   any fragment in a valid manifest file then it will be deleted.
//! * Unreferenced index files - If an index file is not referenced by
//!   any valid manifest file then it will be deleted.
//!
//! It is also difficult to distinguish between a data/tx/idx file which was
//! leftover from an abandoned transaction and a data file which is part
//! of an ongoing operation (both will look like unreferenced data files).
//!
//! If the file is referenced by at least one manifest (even if that manifest
//! is old and being deleted) then we assume it is not part of an ongoing
//! operation and can be safely deleted.
//!
//! If the data is not referenced by any manifest then we look at the age of
//! the file.  If the file is at least 7 days old then we assume it is probably
//! not part of any ongoing operation and we will delete it.
//!
//! Otherwise we will leave the file unless delete_unverified is set to true.
//! (which should only be done if the caller can guarantee there are no updates
//! happening at the same time)

use super::refs::TagContents;
use crate::dataset::TRANSACTIONS_DIR;
use crate::{Dataset, utils::temporal::utc_now};
use chrono::{DateTime, TimeDelta, Utc};
use dashmap::DashSet;
use futures::future::try_join_all;
use futures::{FutureExt, StreamExt, TryStreamExt, stream};
use humantime::parse_duration;
use lance_core::{
    Error, Result,
    utils::tracing::{
        AUDIT_MODE_DELETE, AUDIT_MODE_DELETE_UNVERIFIED, AUDIT_TYPE_DATA, AUDIT_TYPE_DELETION,
        AUDIT_TYPE_INDEX, AUDIT_TYPE_MANIFEST, DATASET_CLEANING_EVENT, TRACE_DATASET_EVENTS,
        TRACE_FILE_AUDIT,
    },
};
use lance_io::object_store::ObjectStore;
use lance_table::{
    feature_flags::{ensure_can_read_manifest, ensure_can_write_manifest},
    format::{IndexMetadata, Manifest},
    io::{
        commit::ManifestLocation,
        deletion::deletion_file_path,
        manifest::{read_manifest, read_manifest_indexes},
    },
};
use object_store::ObjectMeta;
use object_store::path::Path;
use prost::Message;
use std::fmt::Debug;
use std::{
    collections::{HashMap, HashSet},
    future,
    sync::{Arc, Mutex, MutexGuard},
    time::Duration,
};
use tokio::sync::OnceCell;
use tokio::time::{MissedTickBehavior, interval};
use tracing::{Span, debug, info, instrument, warn};

#[derive(Clone, Debug, Default)]
struct ReferencedFiles {
    data_paths: HashSet<Path>,
    managed_blob_paths: HashSet<Path>,
    delete_paths: HashSet<Path>,
    tx_paths: HashSet<Path>,
    index_uuids: HashSet<String>,
    /// Stable-partition row-map ids (`_fri/<map_id>/`) referenced by tagged
    /// fragment-reuse-index entries. Resolved from
    /// [`CleanupInspection::frag_reuse_entries`] before deletion decisions.
    frag_reuse_map_ids: HashSet<String>,
}

#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct RemovalStats {
    pub bytes_removed: u64,
    pub old_versions: u64,
    pub data_files_removed: u64,
    pub transaction_files_removed: u64,
    pub index_files_removed: u64,
    pub deletion_files_removed: u64,
    /// Files that could not be deleted. They are not counted as removed; the next
    /// cleanup finds them again by listing.
    pub failed_deletes: u64,
}

/// A read-only explanation of what a cleanup operation would remove.
///
/// This is an explanation, not a deletion plan.  Calling
/// [`CleanupOperation::execute`] re-evaluates the current dataset and reference
/// state before deleting files.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct CleanupExplanation {
    /// Dataset version observed when the explanation was produced.
    pub read_version: u64,
    /// Aggregate statistics for files that would be removed.
    pub stats: RemovalStats,
    /// Candidate files that would be removed, capped by `candidate_file_limit`.
    pub candidate_files: Vec<CleanupCandidateFile>,
    /// True if more candidate files were found than are included.
    pub candidate_files_truncated: bool,
    /// Maximum number of candidate files included in this explanation.
    pub candidate_file_limit: usize,
    /// Referenced child branches and whether cleanup would cascade into them.
    pub referenced_branches: Vec<CleanupReferencedBranch>,
    /// Non-fatal warnings about the explanation.
    pub warnings: Vec<String>,
}

/// A file that cleanup identified as removable.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct CleanupCandidateFile {
    /// Dataset-relative or storage path for the candidate file.
    pub path: String,
    /// Kind of file identified by cleanup.
    pub kind: CleanupFileKind,
    /// True if the file is removable only because it aged past the unverified
    /// retention threshold or `delete_unverified` is enabled.
    pub unverified: bool,
    /// Candidate file size in bytes.
    pub size_bytes: u64,
}

/// A branch that references the current branch lineage.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct CleanupReferencedBranch {
    /// Branch name.
    pub name: String,
    /// Version of the current lineage referenced by this branch.
    pub referenced_version: u64,
    /// True if this branch would be cleaned when cascading cleanup is enabled.
    pub cleanup_candidate: bool,
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum CleanupFileKind {
    Manifest,
    Data,
    Transaction,
    Index,
    Deletion,
    /// A leftover `_versions/.tmp` manifest from a failed transaction.  These
    /// are deleted but excluded from per-kind `RemovalStats` counts and audit
    /// logs to match the long-standing cleanup behavior.  Their bytes
    /// are still included in `bytes_removed`.
    TemporaryManifest,
}

impl CleanupCandidateFile {
    fn from_cleanup_file(file: &CleanupFile) -> Self {
        Self {
            path: file.path.to_string(),
            kind: file.kind,
            unverified: file.unverified,
            size_bytes: file.size_bytes,
        }
    }
}

fn cleanup_file(
    path: Path,
    kind: CleanupFileKind,
    unverified: bool,
    size_bytes: u64,
) -> Option<CleanupFile> {
    Some(CleanupFile {
        path,
        kind,
        unverified,
        size_bytes,
    })
}

#[derive(Clone, Debug)]
struct CleanupFile {
    path: Path,
    kind: CleanupFileKind,
    /// True when the file was kept on disk past its referenced lifetime
    /// because we could not verify it was safe to remove (e.g. produced by an
    /// unfinished commit) and is being deleted only because it has aged past
    /// the unverified-retention threshold or `delete_unverified` is set.
    unverified: bool,
    size_bytes: u64,
}

impl RemovalStats {
    fn record_file(&mut self, file: &CleanupFile) {
        self.bytes_removed += file.size_bytes;
        match file.kind {
            CleanupFileKind::Manifest => self.old_versions += 1,
            CleanupFileKind::Data => self.data_files_removed += 1,
            CleanupFileKind::Transaction => self.transaction_files_removed += 1,
            CleanupFileKind::Index => self.index_files_removed += 1,
            CleanupFileKind::Deletion => self.deletion_files_removed += 1,
            CleanupFileKind::TemporaryManifest => {}
        }
    }

    fn merge(&mut self, other: &Self) {
        self.bytes_removed += other.bytes_removed;
        self.old_versions += other.old_versions;
        self.data_files_removed += other.data_files_removed;
        self.transaction_files_removed += other.transaction_files_removed;
        self.index_files_removed += other.index_files_removed;
        self.deletion_files_removed += other.deletion_files_removed;
        self.failed_deletes += other.failed_deletes;
    }
}

#[derive(Debug, Default)]
struct CleanupRunResult {
    stats: RemovalStats,
    removed_manifests: HashSet<Path>,
    candidate_files: Vec<CleanupCandidateFile>,
    candidate_files_truncated: bool,
    referenced_branches: Vec<CleanupReferencedBranch>,
}

impl CleanupRunResult {
    fn record_file(
        &mut self,
        file: &CleanupFile,
        candidate_file_limit: Option<usize>,
        track_removed_manifests: bool,
    ) {
        self.stats.record_file(file);
        if track_removed_manifests && matches!(file.kind, CleanupFileKind::Manifest) {
            self.removed_manifests.insert(file.path.clone());
        }
        if let Some(limit) = candidate_file_limit {
            if self.candidate_files.len() < limit {
                self.candidate_files
                    .push(CleanupCandidateFile::from_cleanup_file(file));
            } else {
                self.candidate_files_truncated = true;
            }
        }
    }

    fn merge(&mut self, other: Self, candidate_file_limit: Option<usize>) {
        self.stats.merge(&other.stats);
        self.removed_manifests.extend(other.removed_manifests);
        self.referenced_branches.extend(other.referenced_branches);
        if let Some(limit) = candidate_file_limit {
            for file in other.candidate_files {
                if self.candidate_files.len() < limit {
                    self.candidate_files.push(file);
                } else {
                    self.candidate_files_truncated = true;
                }
            }
            self.candidate_files_truncated |= other.candidate_files_truncated;
        }
    }
}

#[derive(Clone, Copy, Debug)]
enum CleanupAction {
    Execute,
    Explain { max_candidate_files: usize },
}

impl CleanupAction {
    fn deletes_files(self) -> bool {
        matches!(self, Self::Execute)
    }

    fn candidate_file_limit(self) -> Option<usize> {
        match self {
            Self::Execute => None,
            Self::Explain {
                max_candidate_files,
            } => Some(max_candidate_files),
        }
    }
}

fn remove_prefix(path: &Path, prefix: &Path) -> Path {
    let relative_parts = path.prefix_match(prefix);
    if relative_parts.is_none() {
        return path.clone();
    }
    Path::from_iter(relative_parts.unwrap())
}

#[derive(Debug)]
struct DeleteExecution {
    concurrency: usize,
    permits: tokio::sync::Semaphore,
    limiter: Option<tokio::sync::Mutex<tokio::time::Interval>>,
}

#[derive(Clone, Debug)]
struct CleanupTask<'a> {
    delete_execution: Option<Arc<DeleteExecution>>,
    dataset: &'a Dataset,
    policy: CleanupPolicy,
    action: CleanupAction,
    read_version: u64,
    ignored_manifests: HashSet<Path>,
    track_removed_manifests: bool,
    include_referenced_branches: bool,
}

/// The visible descriptor rows and base namespace used by one fragment scan.
#[derive(Clone, Debug, PartialEq, Eq, Hash)]
struct ManagedScanKey {
    store_prefix: String,
    data_dir: Path,
    bases: Vec<(u32, String, bool)>,
    fragment: Vec<u8>,
    fields: Vec<arrow_schema::Field>,
    field_ids: Vec<i32>,
    retained: bool,
}

/// A manifest that has aged out and is queued for deletion.
#[derive(Clone, Debug)]
struct ExpiredManifest {
    version: u64,
    /// Size from the manifest listing. `None` means it must be fetched before it can
    /// be counted in `RemovalStats::bytes_removed`.
    size_bytes: Option<u64>,
}

/// Information about the dataset that we learn by inspecting all of the manifests
#[derive(Clone, Debug, Default)]
struct CleanupInspection {
    // Completed scans have already added their paths to the sets below. Keep
    // only whether they found owner paths, not another copy of all references.
    managed_scans: HashMap<ManagedScanKey, Arc<OnceCell<bool>>>,
    old_manifests: HashMap<Path, ExpiredManifest>,
    /// Store records to retire once their manifests are gone, by version;
    /// see `CommitHandler::forget_version`.
    retired_records: HashMap<u64, String>,
    /// Referenced files are part of our working set
    referenced_files: ReferencedFiles,
    /// Verified files may or may not be part of the working set but they are
    /// referenced by at least one manifest file (potentially an old one) and
    /// so we know that they are not part of an ongoing operation.
    verified_files: ReferencedFiles,
    /// Track tagged old versions in case we want to raise a `CleanupError`.
    tagged_old_versions: HashSet<u64>,
    /// Tagged fragment-reuse-index entries seen across manifests, deduped by
    /// entry uuid so each is decoded at most once; the flag records whether
    /// some retained (working-set) manifest carries the entry.
    frag_reuse_entries: HashMap<uuid::Uuid, (IndexMetadata, bool)>,
    /// Tagged FRI entry uuids per manifest version, so a branch rescue that
    /// retains a parent manifest can flip its entries into the working set
    /// (their row maps become referenced instead of merely verified).
    frag_reuse_entry_versions: HashMap<u64, Vec<uuid::Uuid>>,
    /// Set when a branch manifest's FRI history could not be interpreted:
    /// `_fri/` garbage collection is skipped for the run rather than risking
    /// row maps the branch references invisibly.
    skip_frag_reuse_gc: bool,
    /// The earliest timestamp of all retained manifests.
    earliest_retained_manifest_time: Option<DateTime<Utc>>,
    /// The latest timestamp of all manifests that will be removed.
    latest_deleted_manifest_time: Option<DateTime<Utc>>,
}

impl CleanupInspection {
    /// Cutoff for `read_dir_all(..., unmodified_since)`.
    ///
    /// Listing only files with `last_modified <= earliest_retained` is valid
    /// when the working set is a time suffix: every retained version is newer
    /// than every deleted one. A tagged old version (or any other sparse
    /// retain) pulls that cutoff backwards, so files from newer deleted
    /// versions are never listed. Their manifests are still removed, which
    /// permanently orphans the data files ([#8705](https://github.com/lance-format/lance/issues/8705)).
    ///
    /// When a deleted manifest is newer than the earliest retained one, drop
    /// the cutoff and scan the whole subtree — the same approach already used
    /// for `_indices/`.
    fn listing_unmodified_since(&self) -> Option<DateTime<Utc>> {
        match (
            self.earliest_retained_manifest_time,
            self.latest_deleted_manifest_time,
        ) {
            (Some(retained), Some(deleted)) if deleted > retained => None,
            (retained, _) => retained,
        }
    }
}

/// If a file cannot be verified then it will only be deleted if it is at least
/// this many days old.
const UNVERIFIED_THRESHOLD_DAYS: i64 = 7;
const DEFAULT_EXPLANATION_MAX_CANDIDATE_FILES: usize = 1_000;

/// Builder-style cleanup operation.
///
/// Call [`Self::explain`] for a read-only explanation of what cleanup would
/// remove, or [`Self::execute`] to re-evaluate the current dataset state and
/// delete files.
pub struct CleanupOperation<'a> {
    dataset: &'a Dataset,
    policy: CleanupPolicy,
    max_candidate_files: usize,
    delete_concurrency: Option<usize>,
    delete_rate_limit: Option<u64>,
}

impl<'a> CleanupOperation<'a> {
    pub(crate) fn new(dataset: &'a Dataset, policy: CleanupPolicy) -> Self {
        Self {
            dataset,
            policy,
            max_candidate_files: DEFAULT_EXPLANATION_MAX_CANDIDATE_FILES,
            delete_concurrency: None,
            delete_rate_limit: None,
        }
    }

    /// Set the maximum in-flight file deletes shared with all cascaded branches.
    ///
    /// If unset, the initiating dataset's object store I/O parallelism is used.
    /// Returns an invalid-input error outside `1..=tokio::sync::Semaphore::MAX_PERMITS`.
    ///
    /// ```
    /// # use lance::{Dataset, Result};
    /// # use lance::dataset::cleanup::CleanupPolicy;
    /// # async fn example(dataset: &Dataset) -> Result<()> {
    /// dataset.cleanup(CleanupPolicy::default())
    ///     .with_delete_concurrency(32)?
    ///     .with_delete_rate_limit(100)?
    ///     .execute().await?;
    /// # Ok(())
    /// # }
    /// ```
    pub fn with_delete_concurrency(mut self, concurrency: usize) -> Result<Self> {
        validate_delete_concurrency(concurrency)?;
        self.delete_concurrency = Some(concurrency);
        Ok(self)
    }

    /// Set the maximum delete requests per second shared with all cascaded branches.
    ///
    /// Overrides [`CleanupPolicy::delete_rate_limit`]; without an override, the
    /// policy's rate applies. If neither is set, deletes are not rate limited.
    /// Returns a cleanup error if `rate` is zero. See [`Self::with_delete_concurrency`]
    /// for an example configuring both limits.
    pub fn with_delete_rate_limit(mut self, rate: u64) -> Result<Self> {
        validate_delete_rate_limit(rate)?;
        self.delete_rate_limit = Some(rate);
        Ok(self)
    }

    /// Set the maximum number of candidate files included in explanations.
    ///
    /// The aggregate [`RemovalStats`] in [`CleanupExplanation`] still include
    /// all files that would be removed.
    pub fn with_max_candidate_files(mut self, max_candidate_files: usize) -> Self {
        self.max_candidate_files = max_candidate_files;
        self
    }

    /// Explain what cleanup would remove without deleting files.
    /// Deletion concurrency and rate limits do not affect this preview.
    pub async fn explain(&self) -> Result<CleanupExplanation> {
        let cleanup = CleanupTask::new(
            self,
            CleanupAction::Explain {
                max_candidate_files: self.max_candidate_files,
            },
        )?;
        let read_version = cleanup.read_version;
        let result = cleanup.run().await?;
        let warnings = if result.candidate_files_truncated {
            vec![format!(
                "candidate_files truncated to {} entries",
                self.max_candidate_files
            )]
        } else {
            Vec::new()
        };
        Ok(CleanupExplanation {
            read_version,
            stats: result.stats,
            candidate_files: result.candidate_files,
            candidate_files_truncated: result.candidate_files_truncated,
            candidate_file_limit: self.max_candidate_files,
            referenced_branches: result.referenced_branches,
            warnings,
        })
    }

    /// Execute cleanup by re-evaluating the current dataset state.
    pub async fn execute(&self) -> Result<RemovalStats> {
        info!(target: TRACE_DATASET_EVENTS, event=DATASET_CLEANING_EVENT, uri=&self.dataset.uri);
        let cleanup = CleanupTask::new(self, CleanupAction::Execute)?;
        Ok(cleanup.run().await?.stats)
    }
}

impl<'a> CleanupTask<'a> {
    fn new(operation: &CleanupOperation<'a>, action: CleanupAction) -> Result<Self> {
        let track_removed_manifests = operation.policy.clean_referenced_branches;
        let include_referenced_branches = action.candidate_file_limit().is_some();
        Self::new_with_ignored_manifests(
            operation,
            action,
            HashSet::new(),
            track_removed_manifests,
            include_referenced_branches,
            None,
        )
    }

    fn new_with_ignored_manifests(
        operation: &CleanupOperation<'a>,
        action: CleanupAction,
        ignored_manifests: HashSet<Path>,
        track_removed_manifests: bool,
        include_referenced_branches: bool,
        mut delete_execution: Option<Arc<DeleteExecution>>,
    ) -> Result<Self> {
        if action.deletes_files() && delete_execution.is_none() {
            let concurrency = operation
                .delete_concurrency
                .unwrap_or_else(|| operation.dataset.object_store.io_parallelism());
            validate_delete_concurrency(concurrency)?;
            delete_execution = Some(Arc::new(DeleteExecution {
                concurrency,
                permits: tokio::sync::Semaphore::new(concurrency),
                limiter: create_delete_limiter(
                    operation
                        .delete_rate_limit
                        .or(operation.policy.delete_rate_limit),
                ),
            }));
        }
        Ok(Self {
            delete_execution,
            dataset: operation.dataset,
            policy: operation.policy.clone(),
            action,
            read_version: operation.dataset.version().version,
            ignored_manifests,
            track_removed_manifests,
            include_referenced_branches,
        })
    }

    async fn run(self) -> Result<CleanupRunResult> {
        let mut final_result = CleanupRunResult::default();
        let candidate_file_limit = self.action.candidate_file_limit();
        // First check if we need to clean referenced branches
        // For cases that referenced branches never clean and the current cleanup cannot clean anything
        // This must happen before cleaning the current branch if the setting is enabled.

        let referenced_branches: Vec<(String, u64)> = self.find_referenced_branches().await?;
        if self.include_referenced_branches {
            final_result.referenced_branches = referenced_branches
                .iter()
                .map(|(name, referenced_version)| CleanupReferencedBranch {
                    name: name.clone(),
                    referenced_version: *referenced_version,
                    cleanup_candidate: self.policy.clean_referenced_branches,
                })
                .collect();
        }
        if self.policy.clean_referenced_branches {
            final_result.merge(
                self.clean_referenced_branches(&referenced_branches).await?,
                candidate_file_limit,
            );
        }

        // we process all manifest files in parallel to figure
        // out which files are referenced by valid manifests

        // get protected manifests first, and include those in process_manifests
        // pass on option to process manifests around whether to return error
        // or clean around the manifest
        let tags = self.dataset.tags().list().await?;
        let current_branch = &self.dataset.manifest.branch;

        // Only retain tags on the current branch.
        // Tags on other branches would take effect in retain_branch_lineage_files
        let tagged_versions: HashSet<u64> = tags
            .values()
            .filter(|tag| match (tag.branch.as_ref(), current_branch.as_ref()) {
                (Some(branch_of_tag), Some(current_branch)) => branch_of_tag == current_branch,
                (None, None) => true,
                _ => false,
            })
            .map(|tag_content| tag_content.version)
            .collect();

        let mut inspection = self.process_manifests(&tagged_versions).await?;

        if self.policy.error_if_tagged_old_versions && !inspection.tagged_old_versions.is_empty() {
            return Err(tagged_old_versions_cleanup_error(
                &tags,
                &inspection.tagged_old_versions,
            ));
        }

        if !referenced_branches.is_empty() {
            let ignored_manifests: HashSet<_> = final_result
                .removed_manifests
                .union(&self.ignored_manifests)
                .cloned()
                .collect();
            inspection = self
                .retain_branch_lineage_files(inspection, &referenced_branches, &ignored_manifests)
                .await?
        };

        final_result.merge(
            self.delete_unreferenced_files(inspection).await?,
            candidate_file_limit,
        );
        Ok(final_result)
    }

    #[instrument(level = "debug", skip_all)]
    async fn process_manifests(
        &'a self,
        tagged_versions: &HashSet<u64>,
    ) -> Result<CleanupInspection> {
        let inspection = Mutex::new(CleanupInspection::default());
        self.dataset
            .commit_handler
            .list_manifest_locations(&self.dataset.base, &self.dataset.object_store, false)
            .try_filter(|location| future::ready(!self.ignored_manifests.contains(&location.path)))
            .try_for_each_concurrent(self.dataset.object_store.io_parallelism(), |location| {
                self.process_manifest_file(location, &inspection, tagged_versions)
            })
            .await?;
        Ok(inspection.into_inner().unwrap())
    }

    async fn inspect_managed_blobs(
        &self,
        dataset: &Dataset,
        retained: bool,
        inspection: &Mutex<CleanupInspection>,
    ) -> Result<bool> {
        let names = dataset
            .schema()
            .fields
            .iter()
            .filter(|field| super::blob::field_contains_blob(field))
            .map(|field| field.name.as_str())
            .collect::<Vec<_>>();
        if names.is_empty() {
            return Ok(false);
        }
        let projection = dataset.schema().project(&names)?;
        let fields = projection
            .fields
            .iter()
            .map(arrow_schema::Field::from)
            .collect::<Vec<_>>();
        let field_ids = projection.field_ids();
        let mut bases = dataset
            .manifest
            .base_paths
            .iter()
            .map(|(id, base)| (*id, base.path.clone(), base.is_dataset_root))
            .collect::<Vec<_>>();
        bases.sort_unstable();
        let mut found = false;
        // Manifests are already read concurrently. Scan their fragments in order
        // to avoid multiplying concurrent descriptor buffers and path sets.
        for fragment in dataset.get_fragments() {
            let key = ManagedScanKey {
                store_prefix: dataset.object_store.store_prefix.clone(),
                data_dir: dataset.data_dir(),
                bases: bases.clone(),
                fragment: lance_table::format::pb::DataFragment::from(&fragment.metadata)
                    .encode_to_vec(),
                fields: fields.clone(),
                field_ids: field_ids.clone(),
                retained,
            };
            let cell = inspection
                .lock()
                .unwrap()
                .managed_scans
                .entry(key)
                .or_default()
                .clone();
            let scanned = cell
                .get_or_try_init(|| async {
                    let paths =
                        super::blob::managed_paths(dataset, self.dataset, fragment.scan()).await?;
                    let found = !paths.is_empty();
                    let mut inspection = inspection.lock().unwrap();
                    for path in paths {
                        let relative = remove_prefix(&path, &self.dataset.base);
                        if retained {
                            inspection
                                .verified_files
                                .managed_blob_paths
                                .remove(&relative);
                            inspection
                                .referenced_files
                                .managed_blob_paths
                                .insert(relative);
                        } else {
                            inspection
                                .verified_files
                                .managed_blob_paths
                                .insert(relative);
                        }
                    }
                    Ok::<_, Error>(found)
                })
                .await;
            match scanned {
                Ok(has_paths) => found |= *has_paths,
                // An expired descriptor can disappear during concurrent cleanup.
                // Do not mark failed scans complete: retained reads must still fail.
                Err(error) if !retained && error.is_not_found() => {}
                Err(error) => return Err(error),
            }
        }
        Ok(found)
    }

    async fn process_manifest_file(
        &self,
        location: ManifestLocation,
        inspection: &Mutex<CleanupInspection>,
        tagged_versions: &HashSet<u64>,
    ) -> Result<()> {
        // TODO: We can't cleanup invalid manifests.  There is no way to distinguish
        // between an invalid manifest and a temporary I/O error.  It's also not safe
        // to ignore a manifest error because if it is a temporary I/O error and we
        // ignore it then we might delete valid data files thinking they are not
        // referenced.

        let manifest_and_indexes = async {
            let manifest =
                read_manifest(&self.dataset.object_store, &location.path, location.size).await?;
            ensure_can_read_manifest(&manifest)?;
            ensure_can_write_manifest(&manifest)?;
            let indexes =
                read_manifest_indexes(&self.dataset.object_store, &location, &manifest).await?;
            Ok::<_, Error>((manifest, indexes))
        }
        .await;
        let (manifest, indexes) = match manifest_and_indexes {
            Ok(manifest_and_indexes) => manifest_and_indexes,
            Err(error) if location.version < self.read_version && error.is_not_found() => {
                // Another cleanup may remove an old manifest after this cleanup lists it.
                // The current manifest is never safe to skip because it anchors our snapshot.
                debug!(
                    manifest_version = location.version,
                    read_version = self.read_version,
                    manifest_path = %location.path,
                    "Skipping old manifest removed by concurrent cleanup"
                );
                // Its record may still be there if that cleanup stopped early.
                if let Some(identity) = location.identity {
                    inspection
                        .lock()
                        .unwrap()
                        .retired_records
                        .insert(location.version, identity);
                }
                return Ok(());
            }
            Err(error) => return Err(error),
        };
        // Don't delete the latest version, even if it is old. Don't delete tagged versions,
        // regardless of age. Don't delete manifests if their version is newer than the dataset
        // version.  These are either in-progress or newly added since we started.
        let is_latest = self.read_version <= manifest.version;
        let is_tagged = tagged_versions.contains(&manifest.version);
        let in_working_set = is_latest || !self.policy.should_clean(&manifest) || is_tagged;
        if manifest.has_managed_blobs() {
            let snapshot = self
                .dataset
                .checkout_version((manifest.branch.as_deref(), Some(manifest.version)))
                .await?;
            self.inspect_managed_blobs(&snapshot, in_working_set, inspection)
                .await?;
        }
        let mut inspection = inspection.lock().unwrap();

        // Track tagged old versions in case we want to return a `CleanupError` later.
        // Only track tagged when it is old.
        if is_tagged && !is_latest && self.policy.should_clean(&manifest) {
            inspection.tagged_old_versions.insert(manifest.version);
        }

        self.process_manifest(&manifest, &indexes, in_working_set, &mut inspection)?;
        let commit_ts = manifest.timestamp();
        if !in_working_set {
            inspection.old_manifests.insert(
                location.path.clone(),
                ExpiredManifest {
                    version: manifest.version,
                    // Carried from the listing so the delete phase need not HEAD
                    // every manifest just to report bytes removed.
                    size_bytes: location.size,
                },
            );
            if let Some(identity) = location.identity.clone() {
                inspection
                    .retired_records
                    .insert(manifest.version, identity);
            }
            match inspection.latest_deleted_manifest_time {
                Some(ts) if commit_ts <= ts => {}
                _ => inspection.latest_deleted_manifest_time = Some(commit_ts),
            }
        } else {
            match inspection.earliest_retained_manifest_time {
                Some(ts) if commit_ts >= ts => {}
                _ => inspection.earliest_retained_manifest_time = Some(commit_ts),
            }
        }
        Ok(())
    }

    fn process_manifest(
        &self,
        manifest: &Manifest,
        indexes: &Vec<IndexMetadata>,
        in_working_set: bool,
        inspection: &mut MutexGuard<CleanupInspection>,
    ) -> Result<()> {
        // If this part of our working set then update referenced_files.  Otherwise, just mark the
        // file as verified.
        let referenced_files = if in_working_set {
            &mut inspection.referenced_files
        } else {
            &mut inspection.verified_files
        };

        for fragment in manifest.fragments.iter() {
            for file in fragment.referenced_lance_files() {
                let full_data_path = self.dataset.data_dir().clone().join(file.path.as_str());
                let relative_data_path = remove_prefix(&full_data_path, &self.dataset.base);
                referenced_files.data_paths.insert(relative_data_path);
            }
            let delpath = fragment
                .deletion_file
                .as_ref()
                .map(|delfile| deletion_file_path(&self.dataset.base, fragment.id, delfile));
            if let Some(delpath) = delpath {
                let relative_path = remove_prefix(&delpath, &self.dataset.base);
                referenced_files.delete_paths.insert(relative_path);
            }
        }
        if let Some(relative_tx_path) = &manifest.transaction_file {
            referenced_files
                .tx_paths
                .insert(Path::parse(TRANSACTIONS_DIR)?.join(relative_tx_path.as_str()));
        }

        for index in indexes {
            let uuid_str = index.uuid.to_string();
            referenced_files.index_uuids.insert(uuid_str);
        }

        // Tagged FRI entries are recorded for row-map reference resolution
        // (their content is decoded later, once per unique entry). A v0 entry
        // cannot reference row maps, so v0-only datasets record nothing.
        for index in indexes {
            if index.name == lance_index::frag_reuse::FRAG_REUSE_INDEX_NAME
                && index.index_version != 0
            {
                let entry = inspection
                    .frag_reuse_entries
                    .entry(index.uuid)
                    .or_insert_with(|| (index.clone(), false));
                entry.1 |= in_working_set;
                inspection
                    .frag_reuse_entry_versions
                    .entry(manifest.version)
                    .or_default()
                    .push(index.uuid);
            }
        }
        Ok(())
    }

    #[instrument(
        level = "debug",
        skip_all,
        fields(
            old_versions = inspection.old_manifests.len(),
            bytes_removed = tracing::field::Empty,
            data_files_removed = tracing::field::Empty,
            transaction_files_removed = tracing::field::Empty,
            index_files_removed = tracing::field::Empty,
            deletion_files_removed = tracing::field::Empty
        )
    )]
    /// Resolve the row-map ids referenced by the recorded tagged FRI entries
    /// into the inspection's referenced (retained manifests) and verified
    /// (expiring manifests) sets. Returns false when some entry cannot be
    /// interpreted, in which case `_fri/` garbage collection is skipped for
    /// this run: an uninterpretable entry may reference maps invisibly.
    async fn resolve_frag_reuse_map_ids(&self, inspection: &mut CleanupInspection) -> bool {
        let entries = std::mem::take(&mut inspection.frag_reuse_entries);
        for (uuid, (entry, in_working_set)) in entries.iter() {
            let map_ids = async {
                let content =
                    crate::index::frag_reuse::load_raw_frag_reuse_content(self.dataset, entry)
                        .await?;
                let ledger = crate::index::frag_reuse::decode_frag_reuse_ledger_from_content(
                    entry.index_version,
                    &content,
                )
                .await?;
                if ledger.has_unsupported_transitions() {
                    return Err(Error::not_supported(
                        "the tagged FRI history carries transitions this client cannot interpret",
                    ));
                }
                Ok::<Vec<String>, Error>(
                    ledger
                        .transitions()
                        .iter()
                        .filter_map(|transition| {
                            match transition.mapping() {
                            lance_table::system_index::frag_reuse::ledger::Mapping::StablePartition(
                                partition,
                            ) => Some(partition.map_id.clone()),
                            _ => None,
                        }
                        })
                        .collect(),
                )
            }
            .await;
            match map_ids {
                Ok(map_ids) => {
                    let target = if *in_working_set {
                        &mut inspection.referenced_files
                    } else {
                        &mut inspection.verified_files
                    };
                    target.frag_reuse_map_ids.extend(map_ids);
                }
                Err(error) => {
                    warn!(
                        entry_uuid = %uuid,
                        error = %error,
                        "Cannot resolve the row-map references of a fragment reuse index \
                         entry; skipping _fri garbage collection for this run"
                    );
                    return false;
                }
            }
        }
        true
    }

    async fn delete_unreferenced_files(
        &self,
        mut inspection: CleanupInspection,
    ) -> Result<CleanupRunResult> {
        let collect_frag_reuse_maps = !inspection.skip_frag_reuse_gc
            && self.resolve_frag_reuse_map_ids(&mut inspection).await;
        let inspection = inspection;
        let cleanup_result = Mutex::new(CleanupRunResult::default());
        let deletes_files = self.action.deletes_files();
        let removes_empty_dirs = matches!(
            self.dataset.object_store.scheme(),
            "file" | "file+uring" | "file-object-store"
        );
        let indices_dir = self.dataset.indices_dir();
        let retained_index_dirs = inspection
            .referenced_files
            .index_uuids
            .iter()
            .map(|uuid| indices_dir.clone().join(uuid.as_str()))
            .collect::<HashSet<_>>();
        let index_dirs_to_remove = Mutex::new(HashSet::new());
        // Versions whose manifest could not be deleted. Their store records must
        // survive with them: a record outliving its manifest is retired by the next
        // cleanup, but a manifest outliving its record is a lost version.
        let undeleted_manifest_versions = Mutex::new(HashSet::new());
        let candidate_file_limit = self.action.candidate_file_limit();
        let verification_threshold = utc_now()
            - TimeDelta::try_days(UNVERIFIED_THRESHOLD_DAYS).expect("TimeDelta::try_days");

        // Build stream for a managed subtree
        let build_listing_stream = |dir: Path, unmodified_since| {
            let inspection_ref = &inspection;
            self.dataset
                .object_store
                .read_dir_all(&dir, unmodified_since)
                .map_ok(|obj| stream::once(future::ready(Ok(obj))).boxed())
                .or_else(|e| {
                    // If the directory doesn't exist then we can just return an empty stream.
                    if e.is_not_found() {
                        future::ready(Ok(stream::empty::<Result<ObjectMeta>>().boxed()))
                    } else {
                        future::ready(Err(e))
                    }
                })
                .try_flatten()
                .try_filter_map(move |obj_meta| {
                    // If a file is new-ish then it might be part of an ongoing operation and so we only
                    // delete it if we can verify it is part of an old version.
                    let maybe_in_progress = !self.policy.delete_unverified
                        && obj_meta.last_modified >= verification_threshold;
                    let file_to_remove = self.cleanup_file_if_not_referenced(
                        obj_meta,
                        maybe_in_progress,
                        inspection_ref,
                    );
                    future::ready(file_to_remove)
                })
                .boxed()
        };

        // Restrict scanning to Lance-managed subtrees for safety and performance.
        // Drop the retained-manifest cutoff when a sparse retain (e.g. a tag)
        // would hide files that belong to newer deleted versions. See
        // [`CleanupInspection::listing_unmodified_since`].
        let unmodified_since = inspection.listing_unmodified_since();
        // Data files owned by no manifest can be newer than the retained
        // manifest floor. List through the moving verification threshold so
        // they become candidates once old enough, or list all data files when
        // the caller explicitly permits deleting unverified files.
        let data_unmodified_since = if self.policy.delete_unverified {
            None
        } else {
            unmodified_since.map(|cutoff| cutoff.max(verification_threshold))
        };
        let mut streams = vec![
            build_listing_stream(self.dataset.versions_dir(), unmodified_since),
            build_listing_stream(self.dataset.transactions_dir(), unmodified_since),
            build_listing_stream(self.dataset.data_dir(), data_unmodified_since),
            build_listing_stream(
                self.dataset.base.clone().join("_blobs"),
                data_unmodified_since,
            ),
            // Index UUIDs from manifests being removed are proof that their files are
            // safe to delete. Scan every index artifact while that proof is available;
            // a retained-manifest cutoff can otherwise skip newer artifacts and lose
            // the proof when the old manifests are removed by this cleanup pass.
            build_listing_stream(self.dataset.indices_dir(), None),
            build_listing_stream(self.dataset.deletions_dir(), unmodified_since),
        ];
        if collect_frag_reuse_maps {
            // Stable-partition row maps, referenced by map id from tagged FRI
            // entries. Like `_indices/`, scanned without a cutoff: references
            // from manifests removed by this pass are proof of deletability.
            // A dataset without a `_fri/` directory yields an empty stream.
            streams.push(build_listing_stream(
                self.dataset.base.clone().join("_fri"),
                None,
            ));
        }
        let unreferenced_files = stream::iter(streams).flatten().boxed();

        let old_manifests = inspection.old_manifests.clone();
        let manifest_files = stream::iter(old_manifests)
            .map(|(path, expired)| async move {
                let size_bytes =
                    expired_manifest_size(&self.dataset.object_store, &path, expired.size_bytes)
                        .await?;
                Ok::<CleanupFile, Error>(CleanupFile {
                    path,
                    kind: CleanupFileKind::Manifest,
                    unverified: false,
                    size_bytes,
                })
            })
            .buffer_unordered(self.dataset.object_store.io_parallelism())
            .boxed();

        let all_files = stream::iter(vec![unreferenced_files, manifest_files]).flatten();
        let all_files_to_remove = all_files.map(|file| -> Result<CleanupFile> {
            let file = file?;
            if deletes_files {
                let mode = if file.unverified {
                    AUDIT_MODE_DELETE_UNVERIFIED
                } else {
                    AUDIT_MODE_DELETE
                };
                let path_str = file.path.as_ref();
                match file.kind {
                    CleanupFileKind::Manifest => {
                        info!(target: TRACE_FILE_AUDIT, mode=AUDIT_MODE_DELETE, r#type=AUDIT_TYPE_MANIFEST, path = path_str);
                    }
                    CleanupFileKind::Data => {
                        info!(target: TRACE_FILE_AUDIT, mode=mode, r#type=AUDIT_TYPE_DATA, path = path_str);
                    }
                    CleanupFileKind::Deletion => {
                        info!(target: TRACE_FILE_AUDIT, mode=mode, r#type=AUDIT_TYPE_DELETION, path = path_str);
                    }
                    CleanupFileKind::Index => {
                        info!(target: TRACE_FILE_AUDIT, mode=mode, r#type=AUDIT_TYPE_INDEX, path = path_str);
                    }
                    CleanupFileKind::Transaction | CleanupFileKind::TemporaryManifest => {}
                }
            }
            if deletes_files && removes_empty_dirs && matches!(file.kind, CleanupFileKind::Index) {
                let mut parent = file.path.parent();
                let mut index_dirs = index_dirs_to_remove.lock().unwrap();
                while let Some(dir_path) = parent {
                    if dir_path == indices_dir || !dir_path.prefix_matches(&indices_dir) {
                        break;
                    }
                    index_dirs.insert(dir_path.clone());
                    parent = dir_path.parent();
                }
            }
            Ok(file)
        });

        if deletes_files {
            let execution = self
                .delete_execution
                .as_ref()
                .expect("execute requires a deletion context");
            // Deleting here rather than through `remove_stream` keeps each outcome
            // attached to its file, which is what lets the stats below count what was
            // actually removed instead of what was merely attempted.
            let store = &self.dataset.object_store;
            all_files_to_remove
                .map(|file| async move {
                    let file = file?;
                    // Acquire concurrency before QPS so queued deletes cannot spend
                    // rate permits early and then start together when a slot opens.
                    let _permit = execution.permits.acquire().await.map_err(|error| {
                        Error::invalid_input(format!("cleanup delete semaphore closed: {error}"))
                    })?;
                    if let Some(limiter) = &execution.limiter {
                        let mut ticker = limiter.lock().await;
                        ticker.tick().await;
                    }
                    let outcome = match store.delete(&file.path).await {
                        Ok(()) => Ok(()),
                        // Cleanup lists first and deletes after, so a concurrent
                        // writer or a second cleanup can remove a path in between.
                        // Already gone is the outcome we wanted.
                        Err(error) if error.is_not_found() => Ok(()),
                        Err(error) => Err(error),
                    };
                    Ok::<_, Error>((file, outcome))
                })
                .buffer_unordered(execution.concurrency)
                .try_for_each(|(file, outcome)| {
                    let mut result = cleanup_result.lock().unwrap();
                    match outcome {
                        Ok(()) => result.record_file(
                            &file,
                            candidate_file_limit,
                            self.track_removed_manifests,
                        ),
                        // One transient failure must not discard a sweep over
                        // millions of objects. The file stays, is not counted as
                        // removed, and the next run finds it again by listing.
                        Err(error) => {
                            // Only the first is warned; a failing backend would
                            // otherwise emit a line per object.
                            if result.stats.failed_deletes == 0 {
                                warn!(path = %file.path, error = %error,
                                      "failed to delete file; continuing and counting it");
                            } else {
                                debug!(path = %file.path, error = %error, "failed to delete file");
                            }
                            result.stats.failed_deletes += 1;
                            if matches!(file.kind, CleanupFileKind::Manifest)
                                && let Some(expired) = inspection.old_manifests.get(&file.path)
                            {
                                undeleted_manifest_versions
                                    .lock()
                                    .unwrap()
                                    .insert(expired.version);
                            }
                        }
                    }
                    future::ready(Ok(()))
                })
                .await?;

            // Only after the objects are gone: a record that outlives its
            // manifest is retired by the next cleanup, the reverse is a lost
            // version.
            let undeleted = undeleted_manifest_versions.into_inner().unwrap();
            for (version, identity) in &inspection.retired_records {
                if undeleted.contains(version) {
                    continue;
                }
                self.dataset
                    .commit_handler
                    .forget_version(&self.dataset.base, *version, identity)
                    .await?;
            }

            if removes_empty_dirs
                && let Err(error) = self
                    .dataset
                    .object_store
                    .remove_empty_dirs(
                        indices_dir.clone(),
                        retained_index_dirs,
                        index_dirs_to_remove.into_inner().unwrap(),
                        (!self.policy.delete_unverified).then_some(verification_threshold),
                    )
                    .await
            {
                warn!(
                    path = indices_dir.as_ref(),
                    error = %error,
                    "Failed to remove empty index directories"
                );
            }
        } else {
            // Nothing is deleted, so the stats describe what would be removed.
            all_files_to_remove
                .try_for_each(|file| {
                    cleanup_result.lock().unwrap().record_file(
                        &file,
                        candidate_file_limit,
                        self.track_removed_manifests,
                    );
                    future::ready(Ok(()))
                })
                .await?;
        }

        let cleanup_result = cleanup_result.into_inner().unwrap();

        let span = Span::current();
        span.record("bytes_removed", cleanup_result.stats.bytes_removed);
        span.record(
            "data_files_removed",
            cleanup_result.stats.data_files_removed,
        );
        span.record(
            "transaction_files_removed",
            cleanup_result.stats.transaction_files_removed,
        );
        span.record(
            "index_files_removed",
            cleanup_result.stats.index_files_removed,
        );
        span.record(
            "deletion_files_removed",
            cleanup_result.stats.deletion_files_removed,
        );

        Ok(cleanup_result)
    }

    fn cleanup_file_if_not_referenced(
        &self,
        obj_meta: ObjectMeta,
        maybe_in_progress: bool,
        inspection: &CleanupInspection,
    ) -> Result<Option<CleanupFile>> {
        let path = obj_meta.location;
        let relative_path = remove_prefix(&path, &self.dataset.base);
        let size_bytes = obj_meta.size;
        if relative_path.as_ref().starts_with("_versions/.tmp") {
            // This is a temporary manifest file.
            //
            // If the file is old (or the user has verified there are no writes in progress) then
            // it must be leftover from a failed tx.
            if maybe_in_progress {
                return Ok(None);
            } else {
                return Ok(cleanup_file(
                    path,
                    CleanupFileKind::TemporaryManifest,
                    true,
                    size_bytes,
                ));
            }
        }
        if relative_path.as_ref().starts_with("_fri") {
            // Row maps are referenced by map id (`_fri/<map_id>/...`) from
            // tagged FRI entries, resolved in `resolve_frag_reuse_map_ids`.
            // The same lifecycle as `_indices/`: kept while some retained
            // manifest references the map; deletable immediately when only
            // expiring manifests reference it; an unreferenced map (a failed
            // commit's orphan, or one released by an already-expired history)
            // ages past the unverified-retention threshold first, which also
            // protects a just-written map whose commit has not landed yet.
            if let Some(map_id) = relative_path.parts().nth(1) {
                if inspection
                    .referenced_files
                    .frag_reuse_map_ids
                    .contains(map_id.as_ref())
                {
                    return Ok(None);
                } else if !maybe_in_progress {
                    return Ok(cleanup_file(path, CleanupFileKind::Index, true, size_bytes));
                } else if inspection
                    .verified_files
                    .frag_reuse_map_ids
                    .contains(map_id.as_ref())
                {
                    return Ok(cleanup_file(
                        path,
                        CleanupFileKind::Index,
                        false,
                        size_bytes,
                    ));
                }
            }
            return Ok(None);
        }
        if relative_path.as_ref().starts_with("_indices") {
            // Indices are referenced by UUID so we need to examine the UUID
            // portion of the path.
            if let Some(uuid) = relative_path.parts().nth(1) {
                if inspection
                    .referenced_files
                    .index_uuids
                    .contains(uuid.as_ref())
                {
                    return Ok(None);
                } else if !maybe_in_progress {
                    return Ok(cleanup_file(path, CleanupFileKind::Index, true, size_bytes));
                } else if inspection
                    .verified_files
                    .index_uuids
                    .contains(uuid.as_ref())
                {
                    return Ok(cleanup_file(
                        path,
                        CleanupFileKind::Index,
                        false,
                        size_bytes,
                    ));
                }
            } else {
                return Ok(None);
            }
        }
        match path.extension() {
            Some("lance") => {
                if relative_path.as_ref().starts_with("data") {
                    if inspection
                        .referenced_files
                        .data_paths
                        .contains(&relative_path)
                    {
                        Ok(None)
                    } else if !maybe_in_progress {
                        Ok(cleanup_file(path, CleanupFileKind::Data, true, size_bytes))
                    } else if inspection
                        .verified_files
                        .data_paths
                        .contains(&relative_path)
                    {
                        Ok(cleanup_file(path, CleanupFileKind::Data, false, size_bytes))
                    } else {
                        Ok(None)
                    }
                } else {
                    // If a .lance file isn't in the data directory we err on the side of leaving it alone
                    Ok(None)
                }
            }
            Some("blob") => {
                if inspection
                    .referenced_files
                    .managed_blob_paths
                    .contains(&relative_path)
                {
                    return Ok(None);
                }
                if relative_path
                    .parts()
                    .next()
                    .is_some_and(|part| part.as_ref() == "_blobs")
                {
                    let verified = inspection
                        .verified_files
                        .managed_blob_paths
                        .contains(&relative_path);
                    return if verified || !maybe_in_progress {
                        Ok(cleanup_file(
                            path,
                            CleanupFileKind::Data,
                            !verified,
                            size_bytes,
                        ))
                    } else {
                        Ok(None)
                    };
                }
                // Blob v2 sidecar files are keyed by the data file stem:
                //   data/{data_file_key}/{obfuscated_blob_id:032b}.blob
                //
                // These files are not referenced directly by the manifest.  Instead, treat them
                // as referenced if their parent data file is referenced.
                if !relative_path.as_ref().starts_with("data") {
                    debug!(
                        path = relative_path.as_ref(),
                        "Will not garbage collect blob file because it does not follow convention"
                    );
                    return Ok(None);
                }

                let mut parts = relative_path.parts();
                let data_dir = parts.next();
                let data_file_key = parts.next();
                let blob_file = parts.next();
                // Be conservative: only handle the expected 3-part layout.
                if !matches!(data_dir, Some(dir) if dir.as_ref() == "data")
                    || data_file_key.is_none()
                    || blob_file.is_none()
                {
                    debug!(
                        path = relative_path.as_ref(),
                        "Will not garbage collect blob file because it does not follow convention"
                    );
                    return Ok(None);
                }
                if parts.next().is_some() {
                    debug!(
                        path = relative_path.as_ref(),
                        "Will not garbage collect blob file because it does not follow convention"
                    );
                    return Ok(None);
                }

                let data_file_key = data_file_key.expect("checked is_some");
                let Ok(parent_data_path) =
                    Path::parse(format!("data/{}.lance", data_file_key.as_ref()))
                else {
                    debug!(
                        path = relative_path.as_ref(),
                        derived_parent = format!("data/{}.lance", data_file_key.as_ref()),
                        "Will not garbage collect blob file because derived parent data file path is invalid"
                    );
                    return Ok(None);
                };

                if inspection
                    .referenced_files
                    .data_paths
                    .contains(&parent_data_path)
                {
                    Ok(None)
                } else if !maybe_in_progress {
                    Ok(cleanup_file(path, CleanupFileKind::Data, true, size_bytes))
                } else if inspection
                    .verified_files
                    .data_paths
                    .contains(&parent_data_path)
                    || inspection
                        .verified_files
                        .managed_blob_paths
                        .contains(&relative_path)
                {
                    Ok(cleanup_file(path, CleanupFileKind::Data, false, size_bytes))
                } else {
                    Ok(None)
                }
            }
            Some("manifest") => {
                // We already scanned the manifest files
                Ok(None)
            }
            Some("arrow") | Some("bin") => {
                if relative_path.as_ref().starts_with("_deletions") {
                    if inspection
                        .referenced_files
                        .delete_paths
                        .contains(&relative_path)
                    {
                        Ok(None)
                    } else if !maybe_in_progress {
                        Ok(cleanup_file(
                            path,
                            CleanupFileKind::Deletion,
                            true,
                            size_bytes,
                        ))
                    } else if inspection
                        .verified_files
                        .delete_paths
                        .contains(&relative_path)
                    {
                        Ok(cleanup_file(
                            path,
                            CleanupFileKind::Deletion,
                            false,
                            size_bytes,
                        ))
                    } else {
                        Ok(None)
                    }
                } else {
                    Ok(None)
                }
            }
            Some("txn") => {
                if relative_path.as_ref().starts_with(TRANSACTIONS_DIR) {
                    if inspection
                        .referenced_files
                        .tx_paths
                        .contains(&relative_path)
                    {
                        Ok(None)
                    } else if !maybe_in_progress
                        || inspection.verified_files.tx_paths.contains(&relative_path)
                    {
                        let unverified =
                            !inspection.verified_files.tx_paths.contains(&relative_path);
                        Ok(cleanup_file(
                            path,
                            CleanupFileKind::Transaction,
                            unverified,
                            size_bytes,
                        ))
                    } else {
                        Ok(None)
                    }
                } else {
                    Ok(None)
                }
            }
            // Local writers stage files in their destination directory before renaming
            // them. These temporary files are never referenced by a manifest, even
            // when they are inside a referenced blob sidecar directory.
            _ if relative_path.as_ref().starts_with("data/")
                && path
                    .filename()
                    .is_some_and(|filename| filename.starts_with(".tmp"))
                && !maybe_in_progress =>
            {
                Ok(cleanup_file(path, CleanupFileKind::Data, true, size_bytes))
            }
            _ => Ok(None),
        }
    }

    async fn find_referenced_branches(&self) -> Result<Vec<(String, u64)>> {
        let current_branch_id = self.dataset.branch_identifier().await?;
        let all_branches = self.dataset.branches().list().await?;
        let children = current_branch_id.collect_referenced_versions(&all_branches);

        // Use a concurrent set to identify branches eligible for cleanup.
        // The filter below preserves the original (branch_name, version) tuples.
        let referenced_branches: DashSet<String> = DashSet::new();
        let tasks: Vec<_> = children
            .iter()
            .map(|(branch_name, referenced_version)| {
                let dataset = &self.dataset;
                let policy = &self.policy;
                let referenced_branches = &referenced_branches;

                async move {
                    let manifest = async {
                        let manifest_location = dataset
                            .commit_handler
                            .resolve_version_location(
                                &dataset.base,
                                *referenced_version,
                                &dataset.object_store.inner,
                            )
                            .await?;

                        read_manifest(
                            &dataset.object_store,
                            &manifest_location.path,
                            manifest_location.size,
                        )
                        .await
                    }
                    .await;
                    match manifest {
                        Ok(manifest) => {
                            ensure_can_read_manifest(&manifest)?;
                            ensure_can_write_manifest(&manifest)?;
                            if policy.should_clean(&manifest) {
                                referenced_branches.insert(branch_name.clone());
                            }
                        }
                        Err(error) if error.is_not_found() => {
                            // The source may be gone while descendants still use its files.
                            // Scan their manifests before deleting any parent data.
                            referenced_branches.insert(branch_name.clone());
                        }
                        Err(error) => return Err(error),
                    }
                    Ok::<(), Error>(())
                }
            })
            .collect();

        try_join_all(tasks).await?;

        // Filter children to only include branches that should be cleaned.
        // The DashSet contains branch names found eligible during concurrent scan.
        let referenced_branches = children
            .iter()
            .filter(|(branch_name, _)| referenced_branches.contains(branch_name))
            .cloned()
            .collect();
        Ok(referenced_branches)
    }

    async fn clean_referenced_branches(
        &self,
        referenced_branches: &[(String, u64)],
    ) -> Result<CleanupRunResult> {
        let final_result = Mutex::new(CleanupRunResult::default());

        // Group branches by their lineage identifier (BranchIdentifier).
        // Branches with the same identifier share a lineage and must be cleaned sequentially
        // to preserve cleanup order. Different lineages can be cleaned concurrently.
        let mut branches_chains = HashMap::new();
        for (branch, id) in referenced_branches {
            branches_chains
                .entry(*id)
                .or_insert_with(Vec::new)
                .push(branch.clone());
        }
        let action = self.action;
        let candidate_file_limit = self.action.candidate_file_limit();
        let tasks: Vec<_> = branches_chains
            .values()
            .map(|branch_chain| {
                let final_result = &final_result;
                async move {
                    for branch in branch_chain {
                        let branch_dataset = self
                            .dataset
                            .checkout_version((branch.as_str(), None))
                            .await?;
                        let ignored_manifests =
                            final_result.lock().unwrap().removed_manifests.clone();
                        if let Some(operation) = build_cleanup_operation(
                            &branch_dataset,
                            branch_dataset.manifest.as_ref(),
                            true,
                        )
                        .await?
                        {
                            let result = cleanup_cascade_branch_run(
                                operation,
                                action,
                                ignored_manifests,
                                self.delete_execution.clone(),
                            )
                            .await?;
                            final_result
                                .lock()
                                .unwrap()
                                .merge(result, candidate_file_limit);
                        }
                    }
                    Ok::<(), Error>(())
                }
            })
            .collect();
        try_join_all(tasks).await?;
        Ok(final_result.into_inner().unwrap())
    }

    // Retain manifests containing files referenced by descendant branches.
    // This protects parent branch files that are still needed by child branches.
    async fn retain_branch_lineage_files(
        &self,
        inspection: CleanupInspection,
        referenced_branches: &[(String, u64)],
        removed_branch_manifests: &HashSet<Path>,
    ) -> Result<CleanupInspection> {
        let inspection = Mutex::new(inspection);
        for (branch, root_version_number) in referenced_branches {
            // Use find_branch to get the branch path directly without checkout.
            // This avoids creating a dataset instance and prevents manifest deletion
            // during the retain operation.
            let branch_location = self.dataset.branch_location().find_branch(Some(branch))?;
            self.dataset
                .commit_handler
                .list_manifest_locations(&branch_location.path, &self.dataset.object_store, false)
                .try_filter(|location| {
                    future::ready(!removed_branch_manifests.contains(&location.path))
                })
                .try_for_each_concurrent(self.dataset.object_store.io_parallelism(), |location| {
                    self.process_branch_referenced_manifests(
                        &branch_location.path,
                        location,
                        *root_version_number,
                        &inspection,
                    )
                })
                .await?;
        }
        Ok(inspection.into_inner().unwrap())
    }

    /// The `_fri/<map_id>/` row maps a branch manifest's tagged FRI entry
    /// resolves into THIS dataset's base: stable-partition mappings whose
    /// `base_id` names the parent (a branch clone's relocated entry keeps
    /// reading the parent's row maps in place). The entry's own details are
    /// read base-aware, exactly as a reader on the branch would.
    async fn branch_frag_reuse_parent_map_ids(
        &self,
        branch_root: &Path,
        manifest: &Manifest,
        entry: &IndexMetadata,
    ) -> Result<Vec<String>> {
        use lance_table::system_index::frag_reuse::ledger::Mapping;

        let details = entry
            .index_details
            .as_ref()
            .filter(|details| details.type_url.ends_with("FragmentReuseIndexDetails"))
            .ok_or_else(|| Error::index("Index details is not for the fragment reuse index"))?;
        let content =
            crate::index::frag_reuse::extract_raw_frag_reuse_content(details, |file| async move {
                let end = file
                    .offset
                    .checked_add(file.size)
                    .and_then(|n| usize::try_from(n).ok())
                    .ok_or_else(|| {
                        Error::corrupt_file_named("FRI details", "external FRI range overflow")
                    })?;
                let (store, indices_dir) = match entry.base_id {
                    None => (None, branch_root.clone().join(crate::dataset::INDICES_DIR)),
                    Some(id) => {
                        let base_path = manifest.base_paths.get(&id).ok_or_else(|| {
                            Error::invalid_input(format!(
                                "base_path id {} not found for index {}",
                                id, entry.uuid
                            ))
                        })?;
                        let path = base_path.extract_path(self.dataset.session.store_registry())?;
                        let dir = if base_path.is_dataset_root {
                            path.join(crate::dataset::INDICES_DIR)
                        } else {
                            path
                        };
                        let store = if base_path.path == self.dataset.uri {
                            None
                        } else {
                            // Foreign bases are opened with default store
                            // params; per-base credentials are not plumbed
                            // through cleanup (see
                            // <https://github.com/lance-format/lance/issues/6093>).
                            Some(
                                lance_io::object_store::ObjectStore::from_uri_and_params(
                                    self.dataset.session.store_registry(),
                                    &base_path.path,
                                    &Default::default(),
                                )
                                .await?
                                .0,
                            )
                        };
                        (store, dir)
                    }
                };
                let path = indices_dir
                    .join(entry.uuid.to_string())
                    .join(file.path.as_str());
                let store = store.as_deref().unwrap_or(&self.dataset.object_store);
                store
                    .open(&path)
                    .await?
                    .get_range(file.offset as usize..end)
                    .await
                    .map_err(Error::from)
            })
            .await?;
        let ledger = crate::index::frag_reuse::decode_frag_reuse_ledger_from_content(
            entry.index_version,
            &content,
        )
        .await?;
        if ledger.has_unsupported_transitions() {
            return Err(Error::not_supported(
                "the branch's tagged FRI history carries transitions this client cannot interpret",
            ));
        }
        Ok(ledger
            .transitions()
            .iter()
            .filter_map(|transition| match transition.mapping() {
                Mapping::StablePartition(partition) => partition
                    .base_id
                    .and_then(|id| manifest.base_paths.get(&id))
                    .filter(|base_path| base_path.path == self.dataset.uri)
                    .map(|_| partition.map_id.clone()),
                _ => None,
            })
            .collect())
    }

    async fn process_branch_referenced_manifests(
        &self,
        branch_root: &Path,
        location: ManifestLocation,
        referenced_version: u64,
        inspection: &Mutex<CleanupInspection>,
    ) -> Result<()> {
        let manifest =
            read_manifest(&self.dataset.object_store, &location.path, location.size).await?;
        ensure_can_read_manifest(&manifest)?;
        ensure_can_write_manifest(&manifest)?;
        let managed_referenced = if manifest.has_managed_blobs() {
            let snapshot = self
                .dataset
                .checkout_version((manifest.branch.as_deref(), Some(manifest.version)))
                .await?;
            self.inspect_managed_blobs(&snapshot, true, inspection)
                .await?
        } else {
            false
        };
        let indexes =
            read_manifest_indexes(&self.dataset.object_store, &location, &manifest).await?;

        // Resolve tagged FRI references into the parent before taking the
        // lock (content reads are async). A history this client cannot
        // interpret disables `_fri/` GC for the whole run: it may reference
        // parent row maps invisibly.
        let mut branch_parent_map_ids: Vec<String> = Vec::new();
        let mut fri_unresolvable = false;
        for index in indexes.iter() {
            if index.name == lance_index::frag_reuse::FRAG_REUSE_INDEX_NAME
                && index.index_version != 0
            {
                match self
                    .branch_frag_reuse_parent_map_ids(branch_root, &manifest, index)
                    .await
                {
                    Ok(map_ids) => branch_parent_map_ids.extend(map_ids),
                    Err(error) => {
                        warn!(
                            entry_uuid = %index.uuid,
                            error = %error,
                            "Cannot resolve the row-map references of a branch's fragment \
                             reuse index entry; skipping _fri garbage collection for this run"
                        );
                        fri_unresolvable = true;
                    }
                }
            }
        }

        let mut inspection = inspection.lock().unwrap();
        let mut is_referenced = managed_referenced;

        for fragment in manifest.fragments.iter() {
            for file in fragment.referenced_lance_files() {
                if let Some(base_id) = file.base_id {
                    let base_path = manifest.base_paths.get(&base_id);
                    if let Some(base_path) = base_path
                        && base_path.path == self.dataset.uri
                    {
                        let full_data_path =
                            self.dataset.data_dir().clone().join(file.path.as_str());
                        let relative_data_path = remove_prefix(&full_data_path, &self.dataset.base);
                        inspection
                            .verified_files
                            .data_paths
                            .remove(&relative_data_path);
                        inspection
                            .referenced_files
                            .data_paths
                            .insert(relative_data_path);
                        is_referenced = true;
                    }
                }
            }
            if let Some(del_file) = fragment.deletion_file.as_ref()
                && let Some(base_id) = del_file.base_id
            {
                let base_path = manifest.base_paths.get(&base_id);
                if let Some(base_path) = base_path {
                    let deletion_path = fragment.deletion_file.as_ref().map(|deletion_file| {
                        deletion_file_path(&self.dataset.base, fragment.id, deletion_file)
                    });
                    if base_path.path == self.dataset.uri {
                        if let Some(deletion_path) = deletion_path {
                            let relative_del_path =
                                remove_prefix(&deletion_path, &self.dataset.base);
                            inspection
                                .verified_files
                                .delete_paths
                                .remove(&relative_del_path);
                            inspection
                                .referenced_files
                                .delete_paths
                                .insert(relative_del_path);
                        }
                        is_referenced = true;
                    }
                }
            }
        }
        for index in indexes {
            if let Some(base_id) = index.base_id {
                let base_path = manifest.base_paths.get(&base_id);
                if let Some(base_path) = base_path
                    && base_path.path == self.dataset.uri
                {
                    let uuid_str = index.uuid.to_string();
                    inspection.verified_files.index_uuids.remove(&uuid_str);
                    inspection.referenced_files.index_uuids.insert(uuid_str);
                    is_referenced = true;
                }
            }
        }
        if fri_unresolvable {
            inspection.skip_frag_reuse_gc = true;
        }
        // The branch's relocated FRI entry keeps translating through row
        // maps that live in the parent's `_fri/`; those maps are part of the
        // working set for as long as the branch's history references them.
        for map_id in branch_parent_map_ids {
            inspection.verified_files.frag_reuse_map_ids.remove(&map_id);
            inspection
                .referenced_files
                .frag_reuse_map_ids
                .insert(map_id);
            is_referenced = true;
        }
        if is_referenced {
            inspection
                .old_manifests
                .retain(|_path, expired| expired.version != referenced_version);
            // Kept on disk, so its record stays too.
            inspection.retired_records.remove(&referenced_version);
            // The rescued parent manifest's own FRI entries join the working
            // set with it: their row maps become referenced, not merely
            // verified, when `resolve_frag_reuse_map_ids` decodes them.
            if let Some(uuids) = inspection
                .frag_reuse_entry_versions
                .get(&referenced_version)
            {
                for uuid in uuids.clone() {
                    if let Some(entry) = inspection.frag_reuse_entries.get_mut(&uuid) {
                        entry.1 = true;
                    }
                }
            }
        }

        Ok(())
    }
}

/// Size of an expired manifest, for `bytes_removed`.
///
/// `known` is what the listing reported. The HEAD fallback only runs when the listing did
/// not say, because issuing it unconditionally costs a request per expired manifest.
///
/// A manifest that has already vanished counts as zero rather than failing. Cleanup lists
/// first and acts after, so a concurrent cleanup can remove one in between — and that is
/// the outcome this sweep wanted. Propagating `NotFound` here would abandon an entire
/// sweep over a single file that is already in the desired state.
async fn expired_manifest_size(
    object_store: &ObjectStore,
    path: &Path,
    known: Option<u64>,
) -> Result<u64> {
    match known {
        Some(size_bytes) => Ok(size_bytes),
        None => match object_store.size(path).await {
            Ok(size_bytes) => Ok(size_bytes),
            Err(error) if error.is_not_found() => Ok(0),
            Err(error) => Err(error),
        },
    }
}

fn create_delete_limiter(rate: Option<u64>) -> Option<tokio::sync::Mutex<tokio::time::Interval>> {
    rate.map(|rate| {
        let mut ticker = interval(calculate_duration(rate));
        ticker.set_missed_tick_behavior(MissedTickBehavior::Delay);
        tokio::sync::Mutex::new(ticker)
    })
}

/// The interval between delete permits for a `delete_rate_limit` of `rate` requests/s.
///
/// One permit issues exactly one `delete`, so the interval is the reciprocal of the
/// rate. Scaling it by a bulk-delete batch size would only be correct if a permit
/// covered a whole batch; nothing coalesces paths into a single request, so doing so
/// lets the limiter issue `batch_size` times the configured rate.
fn calculate_duration(rate: u64) -> Duration {
    let effective_rate = rate.max(1);
    info!(
        "delete_rate_limit enabled: limit {} delete requests/sec",
        effective_rate
    );
    Duration::from_nanos(1_000_000_000u64.div_ceil(effective_rate).max(1))
}

fn validate_delete_concurrency(value: usize) -> Result<()> {
    if !(1..=tokio::sync::Semaphore::MAX_PERMITS).contains(&value) {
        return Err(Error::invalid_input(format!(
            "delete_concurrency must be between 1 and {}, got {value}",
            tokio::sync::Semaphore::MAX_PERMITS,
        )));
    }
    Ok(())
}

fn validate_delete_rate_limit(rate: u64) -> Result<()> {
    if rate == 0 {
        return Err(Error::Cleanup {
            message: format!("delete_rate_limit must be greater than 0, got {rate}"),
        });
    }
    Ok(())
}

#[derive(Clone, Debug)]
pub struct CleanupPolicy {
    /// If not none, cleanup all versions before the specified timestamp.
    pub before_timestamp: Option<DateTime<Utc>>,
    /// If not none, cleanup all versions before the specified version.
    pub before_version: Option<u64>,
    /// If not none, cleanup only the specified versions.
    pub versions: Option<HashSet<u64>>,
    /// If true, delete unverified data files even if they are recent
    pub delete_unverified: bool,
    /// If true, return an Error if a tagged version is old
    pub error_if_tagged_old_versions: bool,
    /// If clean the referenced branches
    pub clean_referenced_branches: bool,
    /// Maximum number of delete requests per second. If None, no rate limiting is applied.
    ///
    /// Use this to avoid hitting S3 (or other object store) request rate limits during
    /// cleanup. Cleanup deletes one path per request, so this is also the number of
    /// paths removed per second. For example, `Some(100)` limits deletions to 100
    /// delete requests per second.
    /// The budget is shared by this cleanup and all cascaded branches, ignoring
    /// their own rate settings. `None` leaves deletes unthrottled unless overridden.
    /// An explicit [`CleanupOperation::with_delete_rate_limit`] overrides this value.
    pub delete_rate_limit: Option<u64>,
}

impl CleanupPolicy {
    pub fn should_clean(&self, manifest: &Manifest) -> bool {
        let mut should_clean = true;
        if let Some(before_timestamp) = self.before_timestamp {
            should_clean &= manifest.timestamp() < before_timestamp;
        }
        if let Some(before_version) = self.before_version {
            should_clean &= manifest.version < before_version;
        }
        if let Some(versions) = self.versions.as_ref() {
            should_clean &= versions.contains(&manifest.version);
        }
        should_clean
    }
}

impl Default for CleanupPolicy {
    fn default() -> Self {
        Self {
            before_timestamp: None,
            before_version: None,
            versions: None,
            delete_unverified: false,
            error_if_tagged_old_versions: true,
            clean_referenced_branches: false,
            delete_rate_limit: None,
        }
    }
}

#[derive(Default)]
pub struct CleanupPolicyBuilder {
    policy: CleanupPolicy,
}

impl CleanupPolicyBuilder {
    /// If auto clean referenced branches.
    pub fn clean_referenced_branches(mut self, clean_referenced_branches: bool) -> Self {
        self.policy.clean_referenced_branches = clean_referenced_branches;
        self
    }

    /// Cleanup all versions before the specified timestamp.
    pub fn before_timestamp(mut self, timestamp: DateTime<Utc>) -> Self {
        self.policy.before_timestamp = Some(timestamp);
        self
    }

    /// Cleanup only the specified dataset versions.
    ///
    /// This is an exact-version filter. If other policy filters are also
    /// configured, a manifest is removed only when it satisfies all of them.
    ///
    /// # Errors
    ///
    /// Returns an error if `versions` is empty.
    pub fn versions(mut self, versions: Vec<u64>) -> Result<Self> {
        if versions.is_empty() {
            return Err(Error::invalid_input(
                "versions must not be empty when specified",
            ));
        }
        self.policy.versions = Some(versions.into_iter().collect());
        Ok(self)
    }

    /// Cleanup all versions except the last `n` versions of the dataset.
    ///
    /// # Errors
    ///
    /// Returns an error if `n` is zero.
    pub async fn retain_n_versions(mut self, dataset: &Dataset, n: usize) -> Result<Self> {
        if n == 0 {
            return Err(Error::invalid_input(format!(
                "retain_versions must be greater than 0, got {n}"
            )));
        }
        let versions = dataset.version_refs().await?;
        self.policy.before_version = if versions.len() <= n {
            Some(versions[0].version)
        } else {
            Some(versions[versions.len() - n].version)
        };

        Ok(self)
    }

    /// Delete without verification.
    ///
    /// By default, files will only be deleted if they are not referenced and are not in
    /// progress(at least 7 days old). Setting delete_unverified to true will not verify whether the
    /// file is in progress.
    /// This config is dangerous, only set to true when you are sure there are no other in-progress
    /// dataset operations.
    pub fn delete_unverified(mut self, delete: bool) -> Self {
        self.policy.delete_unverified = delete;
        self
    }

    /// If this argument True, an exception will be raised if any tagged versions match the
    /// parameters.
    pub fn error_if_tagged_old_versions(mut self, error: bool) -> Self {
        self.policy.error_if_tagged_old_versions = error;
        self
    }

    /// Limit the number of delete requests per second during cleanup.
    ///
    /// By default (None), deletions run at full speed. Set this to a positive value to
    /// throttle deletions and avoid hitting object store request rate limits (e.g. S3 HTTP 503).
    /// The request budget is shared with all cascaded branches, overriding their rate settings.
    ///
    /// # Errors
    ///
    /// Returns an error if `rate` is zero.
    pub fn delete_rate_limit(mut self, rate: u64) -> Result<Self> {
        validate_delete_rate_limit(rate)?;
        self.policy.delete_rate_limit = Some(rate);
        Ok(self)
    }

    pub fn build(self) -> CleanupPolicy {
        self.policy
    }
}

/// Deletes old versions of a dataset, removing files that are no longer
/// needed.
///
/// This function will remove old manifest files, data files, indexes,
/// delete files, and transaction files.
///
/// It will only remove files that are not referenced by any valid manifest.
///
/// The latest manifest is always considered valid and will not be removed
/// even if it satisfied the cleanup policy.
pub async fn cleanup_old_versions(
    dataset: &Dataset,
    policy: CleanupPolicy,
) -> Result<RemovalStats> {
    CleanupOperation::new(dataset, policy).execute().await
}

/// If the dataset config has `lance.auto_cleanup` parameters set,
/// this function automatically calls `dataset.cleanup_old_versions`
/// every `lance.auto_cleanup.interval` versions. This function calls
/// `dataset.cleanup_old_versions` with `lance.auto_cleanup.older_than`
/// for `older_than` and `Some(false)` for both `delete_unverified` and
/// `error_if_tagged_old_versions`.
pub async fn auto_cleanup_hook(
    dataset: &Dataset,
    manifest: &Manifest,
) -> Result<Option<RemovalStats>> {
    let Some(operation) = build_cleanup_operation(dataset, manifest, false).await? else {
        return Ok(None);
    };
    // Bound the cleanup future's type depth in commit callers.
    Ok(Some(operation.execute().boxed().await?))
}

/// This is trigger when a parent branch is cleaning and `clean_referenced_branches` is set as true
/// For cascade branches, some cleanup parameters need be overridden.
pub async fn cleanup_cascade_branch(
    dataset: &Dataset,
    manifest: &Manifest,
) -> Result<Option<RemovalStats>> {
    let Some(operation) = build_cleanup_operation(dataset, manifest, false).await? else {
        return Ok(None);
    };
    let result =
        cleanup_cascade_branch_run(operation, CleanupAction::Execute, HashSet::new(), None).await?;
    Ok(Some(result.stats))
}

async fn cleanup_cascade_branch_run(
    mut operation: CleanupOperation<'_>,
    action: CleanupAction,
    ignored_manifests: HashSet<Path>,
    delete_execution: Option<Arc<DeleteExecution>>,
) -> Result<CleanupRunResult> {
    operation.policy.clean_referenced_branches = false;
    operation.policy.error_if_tagged_old_versions = false;
    if action.deletes_files() {
        info!(target: TRACE_DATASET_EVENTS, event=DATASET_CLEANING_EVENT, uri=&operation.dataset.uri);
    }
    CleanupTask::new_with_ignored_manifests(
        &operation,
        action,
        ignored_manifests,
        true,
        false,
        delete_execution,
    )?
    .run()
    .await
}

/// Build the eligible automatic cleanup policy from the manifest configuration.
///
/// The returned policy includes the legacy QPS field, but not operation-level
/// concurrency. Use [`auto_cleanup_hook`] to execute with all configured limits.
///
/// ```
/// # use lance::{Dataset, Result};
/// # use lance::dataset::cleanup::build_cleanup_policy;
/// # async fn example(dataset: &Dataset, manifest: &lance_table::format::Manifest) -> Result<()> {
/// let policy = build_cleanup_policy(dataset, manifest).await?;
/// # Ok(())
/// # }
/// ```
pub async fn build_cleanup_policy(
    dataset: &Dataset,
    manifest: &Manifest,
) -> Result<Option<CleanupPolicy>> {
    Ok(build_cleanup_operation(dataset, manifest, false)
        .await?
        .map(|operation| operation.policy))
}

async fn build_cleanup_operation<'a>(
    dataset: &'a Dataset,
    manifest: &Manifest,
    cascade: bool,
) -> Result<Option<CleanupOperation<'a>>> {
    if let Some(interval) = manifest.config.get("lance.auto_cleanup.interval") {
        let interval: u64 = match interval.parse() {
            Ok(i) => i,
            Err(e) => {
                return Err(Error::Cleanup {
                    message: format!(
                        "Error encountered while parsing lance.auto_cleanup.interval as u64: {}",
                        e
                    ),
                });
            }
        };

        if interval != 0 && !manifest.version.is_multiple_of(interval) {
            return Ok(None);
        }
    } else {
        return Ok(None);
    }

    let mut builder = CleanupPolicyBuilder::default();
    if let Some(older_than) = manifest.config.get("lance.auto_cleanup.older_than") {
        let std_older_than = match parse_duration(older_than) {
            Ok(t) => t,
            Err(e) => {
                return Err(Error::Cleanup {
                    message: format!(
                        "Error encountered while parsing lance.auto_cleanup.older_than as std::time::Duration: {}",
                        e
                    ),
                });
            }
        };
        let timestamp = utc_now() - TimeDelta::from_std(std_older_than).unwrap_or(TimeDelta::MAX);
        builder = builder.before_timestamp(timestamp);
    }
    if let Some(retain_versions) = manifest.config.get("lance.auto_cleanup.retain_versions") {
        let retain_versions: usize = match retain_versions.parse() {
            Ok(n) => n,
            Err(e) => {
                return Err(Error::Cleanup {
                    message: format!(
                        "Error encountered while parsing lance.auto_cleanup.retain_versions as u64: {}",
                        e
                    ),
                });
            }
        };
        builder = builder.retain_n_versions(dataset, retain_versions).await?;
    }
    if let Some(referenced_branch) = manifest.config.get("lance.auto_cleanup.referenced_branch") {
        let clean_referenced: bool = match referenced_branch.parse() {
            Ok(b) => b,
            Err(e) => {
                return Err(Error::Cleanup {
                    message: format!(
                        "Error encountered while parsing lance.auto_cleanup.referenced_branch as bool: {}",
                        e
                    ),
                });
            }
        };
        // Map config to policy flag controlling whether referenced branches are cleaned
        builder = builder.clean_referenced_branches(clean_referenced);
    }

    if !cascade
        && let Some(delete_rate_limit) = manifest.config.get("lance.auto_cleanup.delete_rate_limit")
    {
        let rate: u64 = delete_rate_limit.parse().map_err(|error| Error::Cleanup {
            message: format!(
                "Error encountered while parsing lance.auto_cleanup.delete_rate_limit as u64: {error}"
            ),
        })?;
        builder = builder.delete_rate_limit(rate)?;
    }

    let mut operation = dataset.cleanup(builder.build());
    if !cascade && let Some(value) = manifest.config.get("lance.auto_cleanup.delete_concurrency") {
        let concurrency = value.parse::<usize>().map_err(|error| Error::Cleanup {
            message: format!(
                "Error encountered while parsing lance.auto_cleanup.delete_concurrency={value:?} as usize: {error}"
            ),
        })?;
        operation = operation.with_delete_concurrency(concurrency)?;
    }

    Ok(Some(operation))
}

fn tagged_old_versions_cleanup_error(
    tags: &HashMap<String, TagContents>,
    tagged_old_versions: &HashSet<u64>,
) -> Error {
    let unreferenced_tags: HashMap<String, u64> = tags
        .iter()
        .filter_map(|(k, v)| {
            if tagged_old_versions.contains(&v.version) {
                Some((k.clone(), v.version))
            } else {
                None
            }
        })
        .collect();

    Error::Cleanup {
        message: format!(
            "{} tagged version(s) have been marked for cleanup. Either set `error_if_tagged_old_versions=false` or delete the following tag(s) to enable cleanup: {:?}",
            unreferenced_tags.len(),
            unreferenced_tags
        ),
    }
}

#[cfg(test)]
mod tests {
    use std::{
        collections::HashMap,
        sync::{
            Arc, Mutex,
            atomic::{AtomicUsize, Ordering},
        },
    };

    use super::*;
    use crate::blob::{BlobArrayBuilder, blob_field};
    use crate::index::DatasetIndexExt;
    use crate::{
        dataset::transaction::{Operation, Transaction},
        dataset::{AutoCleanupParams, ReadParams, WriteMode, WriteParams, builder::DatasetBuilder},
        index::vector::VectorIndexParams,
    };
    use all_asserts::{assert_gt, assert_lt};
    use arrow::compute;
    use arrow_array::{
        Int32Array, RecordBatch, RecordBatchIterator, RecordBatchReader, UInt64Array,
    };
    use arrow_schema::{DataType, Field, Schema as ArrowSchema};
    use datafusion::common::assert_contains;
    use lance_core::utils::tempfile::TempStrDir;
    use lance_core::utils::testing::{ProxyObjectStore, ProxyObjectStorePolicy};
    use lance_index::IndexType;
    use lance_io::object_store::{
        ObjectStore, ObjectStoreParams, ObjectStoreRegistry, WrappingObjectStore,
    };
    use lance_linalg::distance::MetricType;
    use lance_table::io::commit::RenameCommitHandler;
    use lance_testing::datagen::{BatchGenerator, IncrementingInt32, RandomVector, some_batch};
    use mock_instant::thread_local::MockClock;
    use object_store::ObjectStoreExt;
    use object_store::throttle::{ThrottleConfig, ThrottledStore};
    use rstest::rstest;
    use uuid::Uuid;

    #[rstest]
    #[case::default(None, false)]
    #[case::serial(Some(1), false)]
    #[case::override_default(Some(4), false)]
    #[case::auto_config(Some(4), true)]
    #[tokio::test]
    async fn test_delete_concurrency(
        #[case] concurrency: Option<usize>,
        #[case] auto_config: bool,
    ) {
        let data = || {
            lance_datagen::gen_batch()
                .col(
                    "id",
                    lance_datagen::array::step::<arrow_array::types::Int32Type>(),
                )
                .into_reader_rows(2.into(), 2.into())
        };
        let delete_latency = Duration::from_secs(1);
        let mut dataset = Dataset::write(
            data(),
            "memory://",
            Some(WriteParams {
                max_rows_per_file: 2,
                store_params: Some(ObjectStoreParams {
                    object_store_wrapper: Some(Arc::new(MockObjectStore {
                        delete_latency,
                        ..MockObjectStore::new()
                    })),
                    ..Default::default()
                }),
                ..Default::default()
            }),
        )
        .await
        .unwrap();
        dataset.append(data(), None).await.unwrap();
        dataset = Dataset::write(
            data(),
            Arc::new(dataset),
            Some(WriteParams {
                mode: WriteMode::Overwrite,
                max_rows_per_file: 2,
                ..Default::default()
            }),
        )
        .await
        .unwrap();
        let expected = dataset.scan().try_into_batch().await.unwrap();
        // A small store default exercises explicit windows both below and above it.
        dataset.object_store = Arc::new(ObjectStore::new(
            dataset.object_store.inner.clone(),
            url::Url::parse("memory://").unwrap(),
            None,
            None,
            false,
            true,
            2,
            0,
            None,
        ));
        tokio::time::pause();
        let policy = CleanupPolicy {
            // Keep the pre-existing exhaustive literal source-compatible.
            before_timestamp: None,
            before_version: Some(dataset.version().version),
            versions: None,
            delete_unverified: true,
            error_if_tagged_old_versions: true,
            clean_referenced_branches: false,
            delete_rate_limit: None,
        };
        let operation = if auto_config {
            let mut manifest = dataset.manifest.as_ref().clone();
            manifest.config.extend([
                ("lance.auto_cleanup.interval".into(), "0".into()),
                ("lance.auto_cleanup.retain_versions".into(), "1".into()),
                ("lance.auto_cleanup.delete_concurrency".into(), "4".into()),
            ]);
            let operation = build_cleanup_operation(&dataset, &manifest, false)
                .await
                .unwrap()
                .unwrap();
            for value in ["0", "abc"] {
                manifest
                    .config
                    .insert("lance.auto_cleanup.delete_concurrency".into(), value.into());
                let error = build_cleanup_operation(&dataset, &manifest, false)
                    .await
                    .err()
                    .unwrap();
                if value == "0" {
                    assert!(matches!(error, Error::InvalidInput { .. }));
                } else {
                    assert!(matches!(error, Error::Cleanup { .. }));
                }
                assert_contains!(error.to_string(), "delete_concurrency");
                assert_contains!(error.to_string(), value);
            }
            operation
        } else {
            let operation = dataset.cleanup(policy);
            match concurrency {
                Some(value) => operation.with_delete_concurrency(value).unwrap(),
                None => operation,
            }
        };
        let explanation = operation.explain().await.unwrap();
        let window = concurrency.unwrap_or_else(|| dataset.object_store.io_parallelism());
        let candidate_count = explanation.candidate_files.len();
        if concurrency.is_some() {
            assert!(candidate_count >= window);
        }
        let expected_duration = delete_latency * candidate_count.div_ceil(window) as u32;
        let start = tokio::time::Instant::now();
        let stats = operation.execute().await.unwrap();
        let elapsed = start.elapsed();
        assert_eq!(stats.old_versions, 2);
        assert_eq!(stats, explanation.stats);
        assert!(stats.data_files_removed > 0);
        // Bound both sides so serial execution and an ignored override cannot pass.
        assert!(elapsed >= expected_duration);
        assert!(elapsed < expected_duration + Duration::from_millis(50));
        assert_eq!(dataset.scan().try_into_batch().await.unwrap(), expected);
    }

    #[rstest]
    #[case::zero(0)]
    #[case::above_limit(tokio::sync::Semaphore::MAX_PERMITS + 1)]
    fn test_invalid_delete_concurrency(#[case] value: usize) {
        let error = validate_delete_concurrency(value).unwrap_err();
        assert!(matches!(error, Error::InvalidInput { .. }));
        assert_contains!(error.to_string(), "delete_concurrency");
        assert_contains!(error.to_string(), value.to_string());
    }

    #[derive(Debug)]
    struct MockObjectStore {
        delete_latency: Duration,
        policy: Arc<Mutex<ProxyObjectStorePolicy>>,
        last_modified_times: Arc<Mutex<HashMap<Path, DateTime<Utc>>>>,
    }

    impl WrappingObjectStore for MockObjectStore {
        fn wrap(
            &self,
            _storage_prefix: &str,
            original: Arc<dyn object_store::ObjectStore>,
        ) -> Arc<dyn object_store::ObjectStore> {
            let original: Arc<dyn object_store::ObjectStore> = if self.delete_latency.is_zero() {
                original
            } else {
                Arc::new(ThrottledStore::new(
                    original,
                    ThrottleConfig {
                        wait_delete_per_call: self.delete_latency,
                        ..Default::default()
                    },
                ))
            };
            Arc::new(ProxyObjectStore::new(original, self.policy.clone()))
        }

        // Injects behaviour into every request, so a listing must not go around it.
        fn wrap_paginated(
            &self,
            _store_prefix: &str,
            _original: Arc<dyn object_store::list::PaginatedListStore>,
        ) -> Option<Arc<dyn object_store::list::PaginatedListStore>> {
            None
        }
    }

    impl MockObjectStore {
        pub(crate) fn new() -> Self {
            let instance = Self {
                delete_latency: Duration::ZERO,
                policy: Arc::new(Mutex::new(ProxyObjectStorePolicy::new())),
                last_modified_times: Arc::new(Mutex::new(HashMap::new())),
            };
            instance.add_timestamp_policy();
            instance
        }

        fn add_timestamp_policy(&self) {
            let mut policy = self.policy.lock().unwrap();
            let times_map = self.last_modified_times.clone();
            policy.set_before_policy(
                "record_file_time",
                Arc::new(move |_, path| {
                    let mut times_map = times_map.lock().unwrap();
                    times_map.insert(path.clone(), utc_now());
                    Ok(())
                }),
            );
            let times_map = self.last_modified_times.clone();
            policy.set_obj_meta_policy(
                "add_recorded_file_time",
                Arc::new(move |_, meta| {
                    let mut meta = meta;
                    if let Some(recorded) = times_map.lock().unwrap().get(&meta.location) {
                        meta.last_modified = *recorded;
                    }
                    Ok(meta)
                }),
            );
        }
    }

    #[derive(Debug, PartialEq, Clone, Copy)]
    struct FileCounts {
        num_data_files: usize,
        num_manifest_files: usize,
        num_index_files: usize,
        num_delete_files: usize,
        num_tx_files: usize,
        num_bytes: u64,
    }

    struct MockDatasetFixture {
        // This is a temporary directory that will be deleted when the fixture
        // is dropped
        tmpdir: TempStrDir,
        dataset_path: String,
        mock_store: Arc<MockObjectStore>,
    }

    impl MockDatasetFixture {
        fn try_new() -> Result<Self> {
            let tmpdir = TempStrDir::default();
            let tmpdir_path = tmpdir.as_str();
            // Use file-object-store:// scheme so that writes go through the ObjectStore
            // wrapper chain (MockObjectStore) instead of the optimized local writer path.
            // The path must always start with "/" (three slashes after the scheme) so that
            // on Windows, a drive letter like "C:" isn't parsed as the URL authority.
            let path_prefix = if tmpdir_path.starts_with('/') {
                ""
            } else {
                "/"
            };
            let dataset_path = format!("file-object-store://{path_prefix}{tmpdir_path}/my_db");
            Ok(Self {
                tmpdir,
                dataset_path,
                mock_store: Arc::new(MockObjectStore::new()),
            })
        }

        fn local_index_dir(&self, uuid: Uuid) -> std::path::PathBuf {
            std::path::Path::new(self.tmpdir.as_str())
                .join("my_db")
                .join(crate::dataset::INDICES_DIR)
                .join(uuid.to_string())
        }

        fn os_params(&self) -> ObjectStoreParams {
            ObjectStoreParams {
                object_store_wrapper: Some(self.mock_store.clone()),
                ..Default::default()
            }
        }

        async fn write_data_impl(
            &self,
            data: impl RecordBatchReader + Send + 'static,
            mode: WriteMode,
        ) -> Result<()> {
            Dataset::write(
                data,
                &self.dataset_path,
                Some(WriteParams {
                    store_params: Some(self.os_params()),
                    commit_handler: Some(Arc::new(RenameCommitHandler)),
                    mode,
                    ..Default::default()
                }),
            )
            .await?;
            Ok(())
        }

        async fn write_some_data_impl(&self, mode: WriteMode) -> Result<()> {
            self.write_data_impl(some_batch(), mode).await?;
            Ok(())
        }

        async fn create_some_data(&self) -> Result<()> {
            self.write_some_data_impl(WriteMode::Create).await
        }

        // Auto-cleanup is disabled by default; this helper creates a dataset
        // with auto-cleanup enabled using the default interval/older_than.
        async fn create_some_data_with_auto_cleanup(&self) -> Result<()> {
            Dataset::write(
                some_batch(),
                &self.dataset_path,
                Some(WriteParams {
                    store_params: Some(self.os_params()),
                    commit_handler: Some(Arc::new(RenameCommitHandler)),
                    mode: WriteMode::Create,
                    auto_cleanup: Some(AutoCleanupParams::default()),
                    ..Default::default()
                }),
            )
            .await?;
            Ok(())
        }

        async fn overwrite_some_data(&self) -> Result<()> {
            self.write_some_data_impl(WriteMode::Overwrite).await
        }

        async fn append_some_data(&self) -> Result<()> {
            self.write_some_data_impl(WriteMode::Append).await
        }

        async fn create_with_data(
            &self,
            data: impl RecordBatchReader + Send + 'static,
        ) -> Result<()> {
            self.write_data_impl(data, WriteMode::Create).await
        }

        async fn append_data(&self, data: impl RecordBatchReader + Send + 'static) -> Result<()> {
            self.write_data_impl(data, WriteMode::Append).await
        }

        async fn overwrite_data(
            &self,
            data: impl RecordBatchReader + Send + 'static,
        ) -> Result<()> {
            self.write_data_impl(data, WriteMode::Overwrite).await
        }

        async fn delete_data(&self, predicate: &str) -> Result<()> {
            let mut db = self.open().await?;
            db.delete(predicate).await?;
            Ok(())
        }

        async fn create_some_index(&self) -> Result<()> {
            let mut db = self.open().await?;
            let index_params = Box::new(VectorIndexParams::ivf_pq(2, 8, 2, MetricType::L2, 5));
            db.create_index(
                &["indexable"],
                IndexType::Vector,
                Some("some_index".to_owned()),
                &*index_params,
                false,
            )
            .await?;
            Ok(())
        }

        fn block_commits(&mut self) {
            let mut policy = self.mock_store.policy.lock().unwrap();
            policy.set_before_policy(
                "block_commit",
                Arc::new(|op, _| -> Result<()> {
                    if op.contains("copy") || op.contains("rename") {
                        return Err(Error::internal("Commit blocked".to_string()));
                    }
                    Ok(())
                }),
            );
        }

        fn block_delete_manifest(&mut self) {
            let mut policy = self.mock_store.policy.lock().unwrap();
            policy.set_before_policy(
                "block_delete_manifest",
                Arc::new(|op, path| -> Result<()> {
                    if op.contains("delete") && path.extension() == Some("manifest") {
                        Err(Error::internal("Delete manifest blocked".to_string()))
                    } else {
                        Ok(())
                    }
                }),
            );
        }

        fn unblock_delete_manifest(&mut self) {
            let mut policy = self.mock_store.policy.lock().unwrap();
            policy.clear_before_policy("block_delete_manifest");
        }

        async fn run_cleanup(&self, before: DateTime<Utc>) -> Result<RemovalStats> {
            let db = self.open().await?;
            cleanup_old_versions(
                &db,
                CleanupPolicyBuilder::default()
                    .before_timestamp(before)
                    .build(),
            )
            .await
        }

        async fn run_cleanup_with_policy(&self, policy: CleanupPolicy) -> Result<RemovalStats> {
            let db = self.open().await?;
            cleanup_old_versions(&db, policy).await
        }

        async fn run_cleanup_with_override(
            &self,
            before: DateTime<Utc>,
            delete_unverified: Option<bool>,
            error_if_tagged_old_versions: Option<bool>,
        ) -> Result<RemovalStats> {
            let db = self.open().await?;
            cleanup_old_versions(
                &db,
                CleanupPolicyBuilder::default()
                    .before_timestamp(before)
                    .delete_unverified(delete_unverified.unwrap_or(false))
                    .error_if_tagged_old_versions(error_if_tagged_old_versions.unwrap_or(true))
                    .build(),
            )
            .await
        }

        async fn open(&self) -> Result<Box<Dataset>> {
            let ds = DatasetBuilder::from_uri(&self.dataset_path)
                .with_read_params(ReadParams {
                    store_options: Some(self.os_params()),
                    ..Default::default()
                })
                .load()
                .await?;
            Ok(Box::new(ds))
        }

        // Load the fixture's dataset.
        async fn load(&self) -> Result<Dataset> {
            self.load_dataset(&self.dataset_path).await
        }

        // Helper to load a dataset with the mock store configured.
        async fn load_dataset(&self, uri: &str) -> Result<Dataset> {
            DatasetBuilder::from_uri(uri)
                .with_read_params(ReadParams {
                    store_options: Some(self.os_params()),
                    ..Default::default()
                })
                .load()
                .await
        }

        // Helper to create a branch and load it as a Dataset.
        async fn create_branch_and_load<V: Into<crate::dataset::refs::Ref>>(
            &self,
            from_dataset: &mut Dataset,
            branch_name: &str,
            source_ref: V,
        ) -> Result<Dataset> {
            let branch_ds = from_dataset
                .create_branch(branch_name, source_ref, Some(self.os_params()))
                .await?;
            self.load_dataset(&branch_ds.uri).await
        }

        async fn count_files(&self) -> Result<FileCounts> {
            let registry = Arc::new(ObjectStoreRegistry::default());
            let (os, path) =
                ObjectStore::from_uri_and_params(registry, &self.dataset_path, &self.os_params())
                    .await?;
            let mut file_stream = os.read_dir_all(&path, None);
            let mut file_count = FileCounts {
                num_data_files: 0,
                num_delete_files: 0,
                num_index_files: 0,
                num_manifest_files: 0,
                num_tx_files: 0,
                num_bytes: 0,
            };
            while let Some(path) = file_stream.try_next().await? {
                file_count.num_bytes += path.size;
                match path.location.extension() {
                    Some("lance") => file_count.num_data_files += 1,
                    Some("manifest") => file_count.num_manifest_files += 1,
                    Some("arrow") | Some("bin") => file_count.num_delete_files += 1,
                    Some("idx") => file_count.num_index_files += 1,
                    Some("txn") => file_count.num_tx_files += 1,
                    _ => (),
                }
            }
            Ok(file_count)
        }

        async fn count_blob_files(&self) -> Result<usize> {
            let registry = Arc::new(ObjectStoreRegistry::default());
            let (os, path) =
                ObjectStore::from_uri_and_params(registry, &self.dataset_path, &self.os_params())
                    .await?;
            let mut file_stream = os.read_dir_all(&path, None);
            let mut blob_count = 0usize;
            while let Some(path) = file_stream.try_next().await? {
                if path.location.extension() == Some("blob") {
                    blob_count += 1;
                }
            }
            Ok(blob_count)
        }

        async fn count_rows(&self) -> Result<usize> {
            let db = self.open().await?;
            let count = db.count_rows(None).await?;
            Ok(count)
        }
    }

    async fn write_dummy_index_artifact(dataset: &Dataset, uuid: Uuid) -> Result<()> {
        let index_dir = dataset.indices_dir().join(uuid.to_string());
        dataset
            .object_store
            .as_ref()
            .put(&index_dir.clone().join("index.idx"), b"idx")
            .await?;
        dataset
            .object_store
            .as_ref()
            .put(&index_dir.clone().join("auxiliary.idx"), b"aux")
            .await?;
        Ok(())
    }

    async fn write_dummy_staging_partial(
        dataset: &Dataset,
        staging_uuid: Uuid,
        shard_uuid: Uuid,
    ) -> Result<()> {
        let shard_dir = dataset
            .indices_dir()
            .join(staging_uuid.to_string())
            .join(format!("partial_{}", shard_uuid));
        dataset
            .object_store
            .as_ref()
            .put(&shard_dir.clone().join("index.idx"), b"idx")
            .await?;
        dataset
            .object_store
            .as_ref()
            .put(&shard_dir.clone().join("auxiliary.idx"), b"aux")
            .await?;
        Ok(())
    }

    fn dummy_index_metadata(
        dataset: &Dataset,
        field_id: i32,
        uuid: Uuid,
        fragment_bitmap: impl IntoIterator<Item = u32>,
    ) -> IndexMetadata {
        IndexMetadata {
            uuid,
            name: "some_index".to_string(),
            fields: vec![field_id],
            covering_fields: vec![],
            dataset_version: dataset.version().version,
            fragment_bitmap: Some(fragment_bitmap.into_iter().collect()),
            index_details: None,
            index_version: IndexType::Vector.version(),
            created_at: None,
            base_id: None,
            files: None,
        }
    }

    fn blob_v2_batch(blob_len: usize) -> Box<dyn RecordBatchReader + Send> {
        let mut blobs = BlobArrayBuilder::new(1);
        blobs.push_bytes(vec![0u8; blob_len]).unwrap();

        let schema = Arc::new(ArrowSchema::new(vec![
            Field::new("id", DataType::Int32, false),
            blob_field("blob", true),
        ]));

        let batch = RecordBatch::try_new(
            schema.clone(),
            vec![Arc::new(Int32Array::from(vec![1])), blobs.finish().unwrap()],
        )
        .unwrap();

        Box::new(RecordBatchIterator::new(
            vec![Ok(batch)].into_iter(),
            schema,
        ))
    }

    // Tiny dedicated objects make individual row retention observable without
    // large payloads or partially live packed objects masking a missing reference.
    fn managed_blob_batch() -> RecordBatch {
        let mut blobs = BlobArrayBuilder::new(2);
        blobs.push_bytes(b"first").unwrap();
        blobs.push_bytes(b"second").unwrap();
        let field = blob_field("blob", false).with_metadata(HashMap::from([
            ("ARROW:extension:name".into(), "lance.blob.v2".into()),
            (
                "lance-encoding:blob-inline-size-threshold".into(),
                "0".into(),
            ),
            (
                "lance-encoding:blob-dedicated-size-threshold".into(),
                "1".into(),
            ),
        ]));
        let schema = Arc::new(ArrowSchema::new_with_metadata(
            vec![Field::new("id", DataType::Int32, false), field],
            HashMap::from([(
                lance_table::feature_flags::MANAGED_BLOBS_META_KEY.to_string(),
                "true".to_string(),
            )]),
        ));
        RecordBatch::try_new(
            schema,
            vec![
                Arc::new(Int32Array::from(vec![0, 1])),
                blobs.finish().unwrap(),
            ],
        )
        .unwrap()
    }

    #[rstest]
    #[case::unprotected(None)]
    #[case::tag(Some("tag"))]
    #[case::branch(Some("branch"))]
    #[tokio::test]
    async fn managed_cleanup_preserves_protected_rows(#[case] protection: Option<&str>) {
        let fixture = MockDatasetFixture::try_new().unwrap();
        let batch = managed_blob_batch();
        let mut dataset = Dataset::write(
            RecordBatchIterator::new([Ok(batch.clone())], batch.schema()),
            &fixture.dataset_path,
            Some(WriteParams {
                store_params: Some(fixture.os_params()),
                data_storage_version: Some(lance_file::version::LanceFileVersion::V2_2),
                ..Default::default()
            }),
        )
        .await
        .unwrap();
        dataset.update_config([("before", "delete")]).await.unwrap();
        let protected_version = dataset.version_id();
        let protected = match protection {
            Some("tag") => {
                dataset
                    .tags()
                    .create("keep", protected_version)
                    .await
                    .unwrap();
                Some(dataset.clone())
            }
            Some("branch") => Some(
                fixture
                    .create_branch_and_load(&mut dataset, "keep", protected_version)
                    .await
                    .unwrap(),
            ),
            _ => None,
        };
        dataset.delete("id = 0").await.unwrap();
        dataset.update_config([("after", "delete")]).await.unwrap();
        dataset
            .cleanup(CleanupPolicy {
                before_version: Some(dataset.version_id()),
                error_if_tagged_old_versions: false,
                ..Default::default()
            })
            .execute()
            .await
            .unwrap();
        assert_eq!(
            fixture.count_blob_files().await.unwrap(),
            if protected.is_some() { 2 } else { 1 }
        );
        let values = Arc::new(dataset)
            .take_blobs_by_indices(&[0], "blob")
            .await
            .unwrap();
        assert_eq!(
            values[0].as_ref().unwrap().read().await.unwrap().as_ref(),
            b"second"
        );
        if let Some(protected) = protected {
            let values = Arc::new(protected)
                .take_blobs_by_indices(&[0], "blob")
                .await
                .unwrap();
            assert_eq!(
                values[0].as_ref().unwrap().read().await.unwrap().as_ref(),
                b"first"
            );
        }
    }

    #[rstest]
    #[tokio::test]
    async fn managed_cleanup_distinguishes_column_projections(
        #[values(false, true)] keep_old: bool,
    ) {
        let fixture = MockDatasetFixture::try_new().unwrap();
        let batch = managed_blob_batch();
        let schema = Arc::new(ArrowSchema::new_with_metadata(
            vec![
                batch.schema().field(0).clone(),
                batch.schema().field(1).clone(),
                batch.schema().field(1).clone().with_name("other"),
            ],
            batch.schema().metadata().clone(),
        ));
        let batch = RecordBatch::try_new(
            schema.clone(),
            vec![
                batch.column(0).clone(),
                batch.column(1).clone(),
                batch.column(1).clone(),
            ],
        )
        .unwrap();
        let mut dataset = Dataset::write(
            RecordBatchIterator::new([Ok(batch)], schema),
            &fixture.dataset_path,
            Some(WriteParams {
                store_params: Some(fixture.os_params()),
                data_storage_version: Some(lance_file::version::LanceFileVersion::V2_2),
                ..Default::default()
            }),
        )
        .await
        .unwrap();
        dataset.update_config([("before", "drop")]).await.unwrap();
        let old = dataset.clone();
        if keep_old {
            dataset
                .tags()
                .create("keep", dataset.version_id())
                .await
                .unwrap();
        }
        dataset.drop_columns(&["other"]).await.unwrap();
        dataset.update_config([("after", "drop")]).await.unwrap();
        assert_eq!(fixture.count_blob_files().await.unwrap(), 4);
        if keep_old {
            // Both views are retained and share the physical file. Scanning the
            // narrower view first must not hide the old view's second column.
            let task = CleanupTask::new(
                &dataset.cleanup(CleanupPolicy::default()),
                CleanupAction::Execute,
            )
            .unwrap();
            let inspection = Mutex::new(CleanupInspection::default());
            task.inspect_managed_blobs(&dataset, true, &inspection)
                .await
                .unwrap();
            task.inspect_managed_blobs(&old, true, &inspection)
                .await
                .unwrap();
            assert_eq!(
                inspection
                    .lock()
                    .unwrap()
                    .referenced_files
                    .managed_blob_paths
                    .len(),
                4
            );
        }
        dataset
            .cleanup(CleanupPolicy {
                before_version: Some(dataset.version_id()),
                error_if_tagged_old_versions: false,
                ..Default::default()
            })
            .execute()
            .await
            .unwrap();
        assert_eq!(
            fixture.count_blob_files().await.unwrap(),
            if keep_old { 4 } else { 2 }
        );
        let values = Arc::new(dataset)
            .take_blobs_by_indices(&[0, 1], "blob")
            .await
            .unwrap();
        for (value, expected) in values
            .iter()
            .zip([b"first".as_slice(), b"second".as_slice()])
        {
            assert_eq!(
                value.as_ref().unwrap().read().await.unwrap().as_ref(),
                expected
            );
        }
    }

    #[tokio::test]
    async fn managed_cleanup_deduplicates_reads_and_retries_failures() {
        let fixture = MockDatasetFixture::try_new().unwrap();
        let batch = managed_blob_batch();
        let mut dataset = Dataset::write(
            RecordBatchIterator::new([Ok(batch.clone())], batch.schema()),
            &fixture.dataset_path,
            Some(WriteParams {
                store_params: Some(fixture.os_params()),
                data_storage_version: Some(lance_file::version::LanceFileVersion::V2_2),
                ..Default::default()
            }),
        )
        .await
        .unwrap();
        dataset.update_config([("version", "2")]).await.unwrap();
        let reads = Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let read_count = reads.clone();
        fixture.mock_store.policy.lock().unwrap().set_before_policy(
            "count_descriptors",
            Arc::new(move |op, path| {
                if op.starts_with("get") && path.extension() == Some("lance") {
                    read_count.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
                }
                Ok(())
            }),
        );
        let task = CleanupTask::new(
            &dataset.cleanup(CleanupPolicy::default()),
            CleanupAction::Execute,
        )
        .unwrap();
        // Warm the session's file metadata cache, then measure one descriptor scan.
        for _ in 0..2 {
            reads.store(0, std::sync::atomic::Ordering::SeqCst);
            task.inspect_managed_blobs(&dataset, false, &Mutex::new(CleanupInspection::default()))
                .await
                .unwrap();
        }
        let single_scan_reads = reads.load(std::sync::atomic::Ordering::SeqCst);
        assert!(single_scan_reads > 0);
        let inspection = Mutex::new(CleanupInspection::default());
        for retained in [false, true] {
            let before = reads.load(std::sync::atomic::Ordering::SeqCst);
            let scans = (0..8).map(|_| task.inspect_managed_blobs(&dataset, retained, &inspection));
            for result in futures::future::join_all(scans).await {
                assert!(result.unwrap());
            }
            let after = reads.load(std::sync::atomic::Ordering::SeqCst);
            assert_eq!(after - before, single_scan_reads);
            task.inspect_managed_blobs(&dataset, retained, &inspection)
                .await
                .unwrap();
            assert_eq!(reads.load(std::sync::atomic::Ordering::SeqCst), after);
        }
        assert_eq!(inspection.lock().unwrap().managed_scans.len(), 2);
        assert_eq!(
            inspection
                .lock()
                .unwrap()
                .referenced_files
                .managed_blob_paths
                .len(),
            2
        );

        let file = dataset
            .data_dir()
            .join(dataset.manifest.fragments[0].files[0].path.as_str());
        let saved = dataset
            .object_store
            .inner
            .get(&file)
            .await
            .unwrap()
            .bytes()
            .await
            .unwrap();
        dataset.object_store.inner.delete(&file).await.unwrap();
        let before = fixture.count_files().await.unwrap();
        let result = dataset
            .cleanup(CleanupPolicy {
                before_version: Some(dataset.version_id()),
                ..Default::default()
            })
            .execute()
            .await;
        assert!(result.unwrap_err().is_not_found());
        assert_eq!(fixture.count_files().await.unwrap(), before);
        assert_eq!(fixture.count_blob_files().await.unwrap(), 2);

        // A missing expired descriptor must not publish successful completion.
        let inspection = Mutex::new(CleanupInspection::default());
        assert!(
            !task
                .inspect_managed_blobs(&dataset, false, &inspection)
                .await
                .unwrap()
        );
        assert!(
            inspection
                .lock()
                .unwrap()
                .managed_scans
                .values()
                .all(|cell| cell.get().is_none())
        );
        dataset
            .object_store
            .inner
            .put(&file, saved.into())
            .await
            .unwrap();
        assert!(
            task.inspect_managed_blobs(&dataset, false, &inspection)
                .await
                .unwrap()
        );
        dataset
            .cleanup(CleanupPolicy {
                before_version: Some(dataset.version_id()),
                ..Default::default()
            })
            .execute()
            .await
            .unwrap();
        assert_eq!(fixture.count_blob_files().await.unwrap(), 2);
    }

    #[tokio::test]
    async fn cleanup_unreferenced_data_files() {
        // We should clean up data files that are only referenced
        // by old versions.  This can happen, for example, due to
        // an overwrite
        let fixture = MockDatasetFixture::try_new().unwrap();
        fixture.create_some_data().await.unwrap();
        fixture.overwrite_some_data().await.unwrap();

        MockClock::set_system_time(TimeDelta::try_days(10).unwrap().to_std().unwrap());

        let before_count = fixture.count_files().await.unwrap();

        let removed = fixture
            .run_cleanup(utc_now() - TimeDelta::try_days(8).unwrap())
            .await
            .unwrap();

        let after_count = fixture.count_files().await.unwrap();
        assert_eq!(removed.old_versions, 1);
        assert_eq!(removed.data_files_removed, 1);
        assert_eq!(
            removed.bytes_removed,
            before_count.num_bytes - after_count.num_bytes
        );

        // There should be one less data file
        assert_lt!(after_count.num_data_files, before_count.num_data_files);
        // And one less manifest file
        assert_lt!(
            after_count.num_manifest_files,
            before_count.num_manifest_files
        );
        assert_lt!(after_count.num_tx_files, before_count.num_tx_files);

        assert_gt!(after_count.num_manifest_files, 0);
        assert_gt!(after_count.num_data_files, 0);
        // We should keep referenced tx files
        assert_gt!(after_count.num_tx_files, 0);
    }

    #[tokio::test]
    async fn cleanup_ignores_old_manifest_removed_after_listing() {
        let fixture = MockDatasetFixture::try_new().unwrap();
        fixture.create_some_data().await.unwrap();
        fixture.overwrite_some_data().await.unwrap();
        let dataset = fixture.open().await.unwrap();

        let old_manifest = dataset
            .commit_handler
            .list_manifest_locations(&dataset.base, &dataset.object_store, false)
            .try_filter(|location| future::ready(location.version == 1))
            .try_next()
            .await
            .unwrap()
            .unwrap();
        dataset
            .object_store
            .delete(&old_manifest.path)
            .await
            .unwrap();

        let cleanup = CleanupTask::new(
            &dataset.cleanup(CleanupPolicyBuilder::default().build()),
            CleanupAction::Execute,
        )
        .unwrap();
        cleanup
            .process_manifest_file(
                old_manifest,
                &Mutex::new(CleanupInspection::default()),
                &HashSet::new(),
            )
            .await
            .unwrap();
    }

    #[tokio::test]
    async fn explain_cleanup_does_not_delete_files() {
        let fixture = MockDatasetFixture::try_new().unwrap();
        fixture.create_some_data().await.unwrap();
        MockClock::set_system_time(TimeDelta::try_seconds(1).unwrap().to_std().unwrap());
        fixture.overwrite_some_data().await.unwrap();

        let before_count = fixture.count_files().await.unwrap();
        let policy = CleanupPolicyBuilder::default()
            .before_timestamp(utc_now())
            .build();

        let dataset = fixture.open().await.unwrap();
        let operation = dataset.cleanup(policy);
        let preview = CleanupTask::new(
            &operation,
            CleanupAction::Explain {
                max_candidate_files: 1000,
            },
        )
        .unwrap();
        assert!(preview.delete_execution.is_none());

        // Explain must remain read-only even if a caller forwards an execution context.
        let execution = CleanupTask::new(&operation, CleanupAction::Execute).unwrap();
        let preview = CleanupTask::new_with_ignored_manifests(
            &operation,
            preview.action,
            HashSet::new(),
            false,
            true,
            execution.delete_execution,
        )
        .unwrap()
        .run()
        .await
        .unwrap();
        let explanation = operation.explain().await.unwrap();
        assert_eq!(preview.stats, explanation.stats);
        let after_preview_count = fixture.count_files().await.unwrap();

        // Files are not actually removed when explaining cleanup.
        assert_eq!(before_count, after_preview_count);
        assert_eq!(explanation.read_version, 2);
        assert_eq!(explanation.stats.old_versions, 1);
        assert_eq!(explanation.stats.data_files_removed, 1);
        assert_eq!(explanation.stats.transaction_files_removed, 1);
        assert_gt!(explanation.stats.bytes_removed, 0);
        assert!(!explanation.candidate_files.is_empty());
        assert!(!explanation.candidate_files_truncated);

        // Running cleanup with the same policy should remove the same files the
        // explanation reported for this unchanged dataset.
        let removed = operation.execute().await.unwrap();
        let after_cleanup_count = fixture.count_files().await.unwrap();

        assert_eq!(
            removed.bytes_removed,
            before_count.num_bytes - after_cleanup_count.num_bytes
        );
        assert_eq!(removed.old_versions, explanation.stats.old_versions);
        assert_eq!(
            removed.data_files_removed,
            explanation.stats.data_files_removed
        );
        assert_eq!(removed.bytes_removed, explanation.stats.bytes_removed);
    }

    #[tokio::test]
    async fn every_removed_file_kind_contributes_its_bytes() {
        // Each kind must contribute its size to `bytes_removed`, not merely increment its
        // per-kind counter. Index files dominate some tables, so an undercount there makes
        // cleanup look far less effective than it is; deletion and transaction files are
        // individually small but numerous.
        let fixture = MockDatasetFixture::try_new().unwrap();
        fixture.create_some_data().await.unwrap();
        fixture.create_some_index().await.unwrap();
        // Produces a deletion file that the overwrite below then orphans. The only data
        // column here is a vector, so delete on the row id instead.
        fixture.delete_data("_rowid < 20").await.unwrap();
        MockClock::set_system_time(TimeDelta::try_days(10).unwrap().to_std().unwrap());
        // Overwrite drops the index and the old fragments, leaving all of it unreferenced.
        fixture.overwrite_some_data().await.unwrap();

        let before = fixture.count_files().await.unwrap();
        let removed = fixture.run_cleanup(utc_now()).await.unwrap();
        let after = fixture.count_files().await.unwrap();

        assert_gt!(removed.index_files_removed, 0);
        assert_gt!(removed.deletion_files_removed, 0);
        assert_gt!(removed.transaction_files_removed, 0);
        assert_gt!(removed.data_files_removed, 0);
        assert_gt!(removed.old_versions, 0);
        // The whole point: reported bytes match the storage delta exactly with every kind
        // in the mix, so no kind is silently contributing zero.
        assert_eq!(removed.bytes_removed, before.num_bytes - after.num_bytes);
    }

    #[tokio::test]
    async fn cleanup_blob_v2_sidecar_files() {
        let fixture = MockDatasetFixture::try_new().unwrap();

        // First version: write a packed blob (sidecar .blob file).
        Dataset::write(
            blob_v2_batch(100 * 1024),
            &fixture.dataset_path,
            Some(WriteParams {
                store_params: Some(fixture.os_params()),
                commit_handler: Some(Arc::new(RenameCommitHandler)),
                mode: WriteMode::Create,
                data_storage_version: Some(lance_file::version::LanceFileVersion::V2_2),
                ..Default::default()
            }),
        )
        .await
        .unwrap();
        assert_gt!(fixture.count_blob_files().await.unwrap(), 0);

        // Second version: overwrite with an inline blob (no sidecar).
        Dataset::write(
            blob_v2_batch(1024),
            &fixture.dataset_path,
            Some(WriteParams {
                store_params: Some(fixture.os_params()),
                commit_handler: Some(Arc::new(RenameCommitHandler)),
                mode: WriteMode::Overwrite,
                data_storage_version: Some(lance_file::version::LanceFileVersion::V2_2),
                ..Default::default()
            }),
        )
        .await
        .unwrap();

        // Advance time so the unverified threshold doesn't interfere.
        MockClock::set_system_time(TimeDelta::try_days(10).unwrap().to_std().unwrap());

        fixture
            .run_cleanup(utc_now() - TimeDelta::try_days(8).unwrap())
            .await
            .unwrap();

        assert_eq!(fixture.count_blob_files().await.unwrap(), 0);
    }

    #[rstest]
    #[case::recent(false, 1, false)]
    #[case::at_threshold(false, 8, false)]
    #[case::aged(false, 9, true)]
    #[case::delete_unverified(true, 1, true)]
    #[tokio::test]
    async fn cleanup_temporary_data_files(
        #[case] delete_unverified: bool,
        #[case] cleanup_day: i64,
        #[case] should_remove: bool,
    ) {
        MockClock::set_system_time(std::time::Duration::ZERO);
        let fixture = MockDatasetFixture::try_new().unwrap();
        fixture.create_some_data().await.unwrap();
        fixture.append_some_data().await.unwrap();
        let dataset = fixture.open().await.unwrap();
        assert_eq!(dataset.manifest.fragments.len(), 2);
        let expected_rows = dataset.count_rows(None).await.unwrap();

        let data_file = &dataset.manifest.fragments[0].files[0].path;
        let data_file_key = data_file.strip_suffix(".lance").unwrap();
        let temporary_paths = [
            dataset.data_dir().join(".tmpAb3Xz9"),
            dataset.data_dir().join("uncommitted").join(".tmpAb3Xz9"),
            dataset.data_dir().join(data_file_key).join(".tmpAb3Xz9"),
        ];
        let preserved_paths = [
            dataset.data_dir().join("unrelated"),
            dataset.data_dir().join(".tmpDirectory").join("unrelated"),
            dataset
                .data_dir()
                .join(data_file_key)
                .join("10000000000000000000000000000000.blob"),
            dataset.base.clone().join("database").join(".tmpAb3Xz9"),
        ];

        // The debris is newer than every retained manifest, as with an interrupted
        // append. Its eligibility must depend on its own age, not the manifest age.
        MockClock::set_system_time(TimeDelta::days(1).to_std().unwrap());
        for path in temporary_paths.iter().chain(&preserved_paths) {
            dataset.object_store.put(path, b"debris").await.unwrap();
        }
        MockClock::set_system_time(TimeDelta::days(cleanup_day).to_std().unwrap());

        let policy = CleanupPolicyBuilder::default()
            .before_timestamp(DateTime::<Utc>::UNIX_EPOCH)
            .delete_unverified(delete_unverified)
            .build();
        let before_count = fixture.count_files().await.unwrap();
        let explanation = dataset.cleanup(policy.clone()).explain().await.unwrap();
        let expected_files = if should_remove {
            temporary_paths.len() as u64
        } else {
            0
        };
        assert_eq!(explanation.stats.data_files_removed, expected_files);
        assert_eq!(explanation.stats.bytes_removed, expected_files * 6);
        assert_eq!(explanation.candidate_files.len() as u64, expected_files);
        for file in &explanation.candidate_files {
            assert_eq!(file.kind, CleanupFileKind::Data);
            assert!(file.unverified);
            assert!(
                temporary_paths
                    .iter()
                    .any(|path| path.as_ref() == file.path)
            );
        }
        assert_eq!(fixture.count_files().await.unwrap(), before_count);

        let removed = dataset.cleanup(policy).execute().await.unwrap();
        assert_eq!(removed.data_files_removed, expected_files);
        assert_eq!(removed.bytes_removed, expected_files * 6);
        assert_eq!(removed.old_versions, 0);
        for path in &temporary_paths {
            assert_eq!(
                dataset.object_store.exists(path).await.unwrap(),
                !should_remove,
                "{path}"
            );
        }
        for path in &preserved_paths {
            assert!(dataset.object_store.exists(path).await.unwrap(), "{path}");
        }
        assert_eq!(fixture.count_rows().await.unwrap(), expected_rows);
    }

    #[tokio::test]
    async fn cleanup_recent_blob_v2_sidecar_files_when_verified() {
        let fixture = MockDatasetFixture::try_new().unwrap();

        Dataset::write(
            blob_v2_batch(100 * 1024),
            &fixture.dataset_path,
            Some(WriteParams {
                store_params: Some(fixture.os_params()),
                commit_handler: Some(Arc::new(RenameCommitHandler)),
                mode: WriteMode::Create,
                data_storage_version: Some(lance_file::version::LanceFileVersion::V2_2),
                ..Default::default()
            }),
        )
        .await
        .unwrap();

        Dataset::write(
            blob_v2_batch(1024),
            &fixture.dataset_path,
            Some(WriteParams {
                store_params: Some(fixture.os_params()),
                commit_handler: Some(Arc::new(RenameCommitHandler)),
                mode: WriteMode::Overwrite,
                data_storage_version: Some(lance_file::version::LanceFileVersion::V2_2),
                ..Default::default()
            }),
        )
        .await
        .unwrap();

        // Old version is verified (referenced by an old manifest) even though the files are
        // recent; cleanup should remove them without waiting 7 days.
        fixture
            .run_cleanup(utc_now() + TimeDelta::seconds(1))
            .await
            .unwrap();

        assert_eq!(fixture.count_blob_files().await.unwrap(), 0);
    }

    #[tokio::test]
    async fn do_not_cleanup_newer_data() {
        // Even though an old manifest is removed the data files should
        // remain if they are still referenced by newer manifests
        let fixture = MockDatasetFixture::try_new().unwrap();
        fixture.create_some_data().await.unwrap();
        MockClock::set_system_time(TimeDelta::try_days(10).unwrap().to_std().unwrap());
        fixture.append_some_data().await.unwrap();
        fixture.append_some_data().await.unwrap();

        let before_count = fixture.count_files().await.unwrap();

        // 3 versions
        assert_eq!(before_count.num_data_files, 3);
        assert_eq!(before_count.num_manifest_files, 3);

        let before = utc_now() - TimeDelta::try_days(7).unwrap();
        let removed = fixture.run_cleanup(before).await.unwrap();

        let after_count = fixture.count_files().await.unwrap();

        assert_eq!(removed.old_versions, 1);
        assert_eq!(
            removed.bytes_removed,
            before_count.num_bytes - after_count.num_bytes
        );

        // The data files should all remain since they are referenced by
        // the latest version
        assert_eq!(after_count.num_data_files, 3);
        // Only the oldest manifest file should be removed
        assert_eq!(after_count.num_manifest_files, 2);
        assert_eq!(after_count.num_tx_files, 2);
    }

    #[rstest]
    #[case::version_number(false)]
    #[case::raw_main_alias(true)]
    #[tokio::test]
    async fn cleanup_error_when_tagged_old_versions(#[case] use_main_alias: bool) {
        // We should not clean up old versions that are tagged.
        // This tests when `error_if_tagged_old_version=true`.
        // When `true`, no files should be cleaned and a `Error::CleanupError`
        // should be returned.
        let fixture = MockDatasetFixture::try_new().unwrap();
        fixture.create_some_data().await.unwrap();
        fixture.overwrite_some_data().await.unwrap();
        fixture.overwrite_some_data().await.unwrap();

        let dataset = *(fixture.open().await.unwrap());

        let reference = |version| {
            if use_main_alias {
                crate::dataset::refs::Ref::Version(Some("main".to_string()), Some(version))
            } else {
                crate::dataset::refs::Ref::VersionNumber(version)
            }
        };
        dataset
            .tags()
            .create("old-tag", reference(1))
            .await
            .unwrap();
        dataset
            .tags()
            .create("another-old-tag", reference(3))
            .await
            .unwrap();
        dataset
            .tags()
            .update("another-old-tag", reference(2))
            .await
            .unwrap();
        for tag in dataset.tags().list().await.unwrap().values() {
            assert_eq!(tag.branch, None);
        }

        MockClock::set_system_time(TimeDelta::try_days(10).unwrap().to_std().unwrap());

        let removed = fixture
            .run_cleanup(utc_now() - TimeDelta::try_days(20).unwrap())
            .await
            .unwrap();
        assert_eq!(removed.old_versions, 0);

        let mut cleanup_error = fixture
            .run_cleanup(utc_now() - TimeDelta::try_days(8).unwrap())
            .await
            .err()
            .unwrap();
        assert_contains!(
            cleanup_error.to_string(),
            "Cleanup error: 2 tagged version(s) have been marked for cleanup. Either set `error_if_tagged_old_versions=false` or delete the following tag(s) to enable cleanup:"
        );

        dataset.tags().delete("old-tag").await.unwrap();

        cleanup_error = fixture
            .run_cleanup(utc_now() - TimeDelta::try_days(8).unwrap())
            .await
            .err()
            .unwrap();
        assert_contains!(
            cleanup_error.to_string(),
            "Cleanup error: 1 tagged version(s) have been marked for cleanup. Either set `error_if_tagged_old_versions=false` or delete the following tag(s) to enable cleanup:"
        );

        dataset.tags().delete("another-old-tag").await.unwrap();

        let removed = fixture
            .run_cleanup(utc_now() - TimeDelta::try_days(8).unwrap())
            .await
            .unwrap();
        assert_eq!(removed.old_versions, 2);
    }

    #[tokio::test]
    async fn cleanup_around_tagged_old_versions() {
        // We should not clean up old versions that are tagged.
        // This tests when `error_if_tagged_old_version=false`.
        // When `false`, old versions should be cleaned up except
        // latest and those that are tagged.
        let fixture = MockDatasetFixture::try_new().unwrap();
        fixture.create_some_data().await.unwrap();
        fixture.overwrite_some_data().await.unwrap();
        fixture.overwrite_some_data().await.unwrap();

        let dataset = *(fixture.open().await.unwrap());

        dataset.tags().create("old-tag", 1).await.unwrap();
        dataset.tags().create("another-old-tag", 2).await.unwrap();
        dataset.tags().create("tag-latest", 3).await.unwrap();

        MockClock::set_system_time(TimeDelta::try_days(10).unwrap().to_std().unwrap());

        let mut removed = fixture
            .run_cleanup_with_override(
                utc_now() - TimeDelta::try_days(8).unwrap(),
                None,
                Some(false),
            )
            .await
            .unwrap();

        assert_eq!(removed.old_versions, 0);

        dataset.tags().delete("old-tag").await.unwrap();

        removed = fixture
            .run_cleanup_with_override(
                utc_now() - TimeDelta::try_days(8).unwrap(),
                None,
                Some(false),
            )
            .await
            .unwrap();
        assert_eq!(removed.old_versions, 1);

        dataset.tags().delete("another-old-tag").await.unwrap();

        removed = fixture
            .run_cleanup_with_override(
                utc_now() - TimeDelta::try_days(8).unwrap(),
                None,
                Some(false),
            )
            .await
            .unwrap();

        assert_eq!(removed.old_versions, 1);
    }

    #[tokio::test]
    async fn cleanup_deletes_data_files_newer_than_tagged_version() {
        // A tag on an old version must not prevent cleanup from deleting data
        // files that belong only to newer, untagged versions. The listing
        // cutoff used to be the earliest retained manifest time; with a tag
        // that pulled the cutoff backwards and skipped those newer files.
        // After their manifests were removed they became permanent orphans
        // (https://github.com/lance-format/lance/issues/8705).
        MockClock::set_system_time(std::time::Duration::from_secs(0));
        let fixture = MockDatasetFixture::try_new().unwrap();
        fixture.create_some_data().await.unwrap();
        MockClock::set_system_time(TimeDelta::try_days(1).unwrap().to_std().unwrap());
        fixture.overwrite_some_data().await.unwrap();
        MockClock::set_system_time(TimeDelta::try_days(2).unwrap().to_std().unwrap());
        fixture.overwrite_some_data().await.unwrap();

        let dataset = *(fixture.open().await.unwrap());
        dataset.tags().create("keep-v1", 1).await.unwrap();

        MockClock::set_system_time(TimeDelta::try_days(10).unwrap().to_std().unwrap());

        let before_count = fixture.count_files().await.unwrap();
        assert_eq!(before_count.num_data_files, 3);
        assert_eq!(before_count.num_manifest_files, 3);

        let removed = fixture
            .run_cleanup_with_override(
                utc_now() - TimeDelta::try_days(8).unwrap(),
                None,
                Some(false),
            )
            .await
            .unwrap();

        assert_eq!(removed.old_versions, 1);
        assert_eq!(removed.data_files_removed, 1);

        let after_count = fixture.count_files().await.unwrap();
        assert_eq!(after_count.num_manifest_files, 2);
        assert_eq!(after_count.num_data_files, 2);
        assert_eq!(after_count.num_tx_files, 2);
    }

    // Helper function to check that the number of files is correct.
    async fn check_num_files(fixture: &MockDatasetFixture, num_expected_files: usize) {
        let file_count = fixture.count_files().await.unwrap();

        assert_eq!(file_count.num_data_files, num_expected_files);
        assert_eq!(file_count.num_manifest_files, num_expected_files);
        assert_eq!(file_count.num_tx_files, num_expected_files);
    }

    #[tokio::test]
    async fn auto_cleanup_old_versions() {
        // Every n commits, all versions older than T should be deleted.
        //
        // We first make many commits and check that all of the versions are
        // present. We then wait until the "older_than" period has elapsed and
        // make many more commits. We check that, without explicitly calling
        // `fixture.run_cleanup`, the old versions are automatically cleaned
        // up and only the new ones remain. File counts are made after every
        // commit.
        let fixture = MockDatasetFixture::try_new().unwrap();

        fixture.create_some_data_with_auto_cleanup().await.unwrap();

        let dataset_config = &fixture.open().await.unwrap().manifest.config;
        let cleanup_interval: usize = dataset_config
            .get("lance.auto_cleanup.interval")
            .unwrap()
            .parse()
            .unwrap();

        let cleanup_older_than = TimeDelta::from_std(
            parse_duration(dataset_config.get("lance.auto_cleanup.older_than").unwrap()).unwrap(),
        )
        .unwrap();

        // First, write many files within the "older_than" window. Check that
        // no files are automatically cleaned up.
        for num_expected_files in 2..2 * cleanup_interval {
            fixture.overwrite_some_data().await.unwrap();
            check_num_files(&fixture, num_expected_files).await;
        }

        // Fast forward so we are outside of the "older_than" window.
        MockClock::set_system_time(
            (cleanup_older_than + TimeDelta::minutes(1))
                .to_std()
                .unwrap(),
        );

        // Write more files and check that those outside of the "older_than" window
        // are cleaned up.
        for num_expected_files in 2..cleanup_interval {
            fixture.overwrite_some_data().await.unwrap();
            check_num_files(&fixture, num_expected_files).await;
        }

        // Overwrite auto cleanup params with custom values
        let mut dataset = *(fixture.open().await.unwrap());
        let mut new_autoclean_params = HashMap::new();

        let new_cleanup_older_than_str = "1month 2days 2h 42min 6sec";
        let new_cleanup_older_than =
            TimeDelta::from_std(parse_duration(new_cleanup_older_than_str).unwrap()).unwrap();
        new_autoclean_params.insert(
            "lance.auto_cleanup.older_than".to_string(),
            new_cleanup_older_than_str.to_string(),
        );

        let new_cleanup_interval = 5;
        new_autoclean_params.insert(
            "lance.auto_cleanup.interval".to_string(),
            new_cleanup_interval.to_string(),
        );

        // Convert to new API format
        let config_updates = new_autoclean_params
            .into_iter()
            .map(|(k, v)| (k, Some(v)))
            .collect::<HashMap<String, Option<String>>>();
        dataset.update_config(config_updates).await.unwrap();

        // Fast forward so we are outside of the new "older_than" window.
        MockClock::set_system_time(
            (cleanup_older_than + new_cleanup_older_than + TimeDelta::minutes(2))
                .to_std()
                .unwrap(),
        );

        fixture.overwrite_some_data().await.unwrap();

        for num_expected_files in 2..new_cleanup_interval {
            fixture.overwrite_some_data().await.unwrap();
            check_num_files(&fixture, num_expected_files).await;
        }
    }

    #[tokio::test]
    async fn test_auto_cleanup_interval_zero() {
        let fixture = MockDatasetFixture::try_new().unwrap();

        fixture.create_some_data().await.unwrap();
        fixture.overwrite_some_data().await.unwrap();
        fixture.overwrite_some_data().await.unwrap();
        check_num_files(&fixture, 3).await;

        let mut dataset = fixture.open().await.unwrap();
        let mut config_updates = HashMap::new();
        config_updates.insert(
            "lance.auto_cleanup.interval".to_string(),
            Some("0".to_string()),
        );
        config_updates.insert(
            "lance.auto_cleanup.retain_versions".to_string(),
            Some("1".to_string()),
        );
        dataset
            .update_config(config_updates)
            .replace()
            .await
            .unwrap();

        fixture.overwrite_some_data().await.unwrap();
        fixture.overwrite_some_data().await.unwrap();
        // The last version before the new commit is retained, means we have 2 versions to assert
        check_num_files(&fixture, 2).await;

        fixture.overwrite_some_data().await.unwrap();
        check_num_files(&fixture, 2).await;
    }

    #[tokio::test]
    async fn cleanup_recent_verified_files() {
        let fixture = MockDatasetFixture::try_new().unwrap();
        fixture.create_some_data().await.unwrap();
        MockClock::set_system_time(TimeDelta::try_seconds(1).unwrap().to_std().unwrap());
        fixture.overwrite_some_data().await.unwrap();

        let before_count = fixture.count_files().await.unwrap();
        assert_eq!(before_count.num_data_files, 2);
        assert_eq!(before_count.num_manifest_files, 2);

        // Not much time has passed but we can still delete the old manifest
        // and the related data files
        let before = utc_now();
        let removed = fixture.run_cleanup(before).await.unwrap();

        let after_count = fixture.count_files().await.unwrap();
        assert_eq!(removed.old_versions, 1);
        assert_eq!(
            removed.bytes_removed,
            before_count.num_bytes - after_count.num_bytes
        );

        assert_eq!(after_count.num_data_files, 1);
        assert_eq!(after_count.num_manifest_files, 1);
    }

    #[tokio::test]
    async fn dont_cleanup_recent_unverified_files() {
        for (override_opt, old_files) in [
            (Some(false), false), // User provides false, files are new - do not delete
            (Some(true), false),  // User provides true, files are new - delete
            (None, true),         // Default, files are old - delete
            (None, false),        // Default, files are new - do not delete
        ] {
            MockClock::set_system_time(std::time::Duration::from_secs(0));
            let mut fixture = MockDatasetFixture::try_new().unwrap();
            fixture.create_some_data().await.unwrap();
            fixture.block_commits();
            assert!(fixture.append_some_data().await.is_err());
            let dataset = fixture.open().await.unwrap();
            let blob_path = dataset.base.clone().join("_blobs").join("orphan.blob");
            dataset
                .object_store
                .put(&blob_path, b"orphan")
                .await
                .unwrap();

            let age = if old_files {
                TimeDelta::try_days(UNVERIFIED_THRESHOLD_DAYS + 1).unwrap()
            } else {
                TimeDelta::try_days(UNVERIFIED_THRESHOLD_DAYS - 1).unwrap()
            };
            MockClock::set_system_time(age.to_std().unwrap());

            // The above created some unreferenced data files but, since they
            // are not referenced in any manifest, and 7 days has not passed, we
            // cannot safely delete them unless the user overrides the safety check

            let before_count = fixture.count_files().await.unwrap();
            assert_eq!(before_count.num_data_files, 2);
            assert_eq!(before_count.num_manifest_files, 1);

            let before = utc_now();
            let removed = fixture
                .run_cleanup_with_override(before, override_opt, None)
                .await
                .unwrap();

            let should_delete = override_opt.unwrap_or(false) || old_files;

            let after_count = fixture.count_files().await.unwrap();
            assert_eq!(
                dataset.object_store.exists(&blob_path).await.unwrap(),
                !should_delete,
                "override={override_opt:?}, old_files={old_files}"
            );
            assert_eq!(removed.old_versions, 0);
            assert_eq!(
                removed.bytes_removed,
                before_count.num_bytes - after_count.num_bytes
            );

            if should_delete {
                assert_gt!(removed.bytes_removed, 0);
            } else {
                assert_eq!(removed.bytes_removed, 0);
            }
        }
    }

    #[rstest]
    #[case::head_error(Some(0))]
    #[case::read_error(Some(1))]
    #[case::missing_source(None)]
    #[tokio::test]
    async fn cleanup_preserves_child_after_source_manifest_failure(
        #[case] fail_request: Option<usize>,
    ) {
        MockClock::set_system_time(std::time::Duration::ZERO);
        let fixture = MockDatasetFixture::try_new().unwrap();
        fixture.create_some_data().await.unwrap();
        let mut parent = fixture.open().await.unwrap();
        let source_manifest = parent.manifest_location.path.clone();
        let child = fixture
            .create_branch_and_load(&mut parent, "child", (None, None))
            .await
            .unwrap();
        let expected = child.scan().try_into_batch().await.unwrap();
        MockClock::set_system_time(TimeDelta::try_days(10).unwrap().to_std().unwrap());
        fixture.overwrite_some_data().await.unwrap();

        let source_reads = Arc::new(AtomicUsize::new(0));
        if let Some(fail_request) = fail_request {
            let source_reads = source_reads.clone();
            fixture.mock_store.policy.lock().unwrap().set_before_policy(
                "fail_branch_source",
                Arc::new(move |operation, path| {
                    // get_opts serves both HEAD (first) and the subsequent range read.
                    if operation == "get_opts"
                        && path == &source_manifest
                        && source_reads.fetch_add(1, Ordering::SeqCst) == fail_request
                    {
                        return Err(Error::internal("transient branch source read"));
                    }
                    Ok(())
                }),
            );
        } else {
            parent
                .object_store
                .inner
                .delete(&source_manifest)
                .await
                .unwrap();
        }
        let before = fixture.count_files().await.unwrap();
        let policy = CleanupPolicy {
            before_version: Some(2),
            error_if_tagged_old_versions: false,
            ..Default::default()
        };
        let result = fixture.run_cleanup_with_policy(policy.clone()).await;
        if let Some(fail_request) = fail_request {
            let error = result.unwrap_err();
            assert!(matches!(error, Error::IO { .. }), "{error}");
            assert_contains!(error.to_string(), "transient branch source read");
            assert_eq!(source_reads.load(Ordering::SeqCst), fail_request + 1);
            let after = fixture.count_files().await.unwrap();
            assert_eq!(after.num_data_files, before.num_data_files);
            assert_eq!(after.num_manifest_files, before.num_manifest_files);
            fixture
                .mock_store
                .policy
                .lock()
                .unwrap()
                .clear_before_policy("fail_branch_source");
            let retried = fixture.run_cleanup_with_policy(policy).await.unwrap();
            assert_eq!(retried.data_files_removed, 0);
        } else {
            assert_eq!(result.unwrap().data_files_removed, 0);
        }
        assert_eq!(child.scan().try_into_batch().await.unwrap(), expected);
    }

    #[tokio::test]
    async fn cleanup_old_index() {
        let fixture = MockDatasetFixture::try_new().unwrap();
        fixture.create_some_data().await.unwrap();
        fixture.create_some_index().await.unwrap();
        MockClock::set_system_time(TimeDelta::try_days(10).unwrap().to_std().unwrap());
        fixture.overwrite_some_data().await.unwrap();

        let before_count = fixture.count_files().await.unwrap();
        // we store 2 files (index and quantized storage) for each index
        assert_eq!(before_count.num_index_files, 2);
        // Two user data files
        assert_eq!(before_count.num_data_files, 2);
        // Creating an index creates a new manifest so there are 3 total
        assert_eq!(before_count.num_manifest_files, 3);

        let before = utc_now() - TimeDelta::try_days(8).unwrap();
        let removed = fixture.run_cleanup(before).await.unwrap();

        let after_count = fixture.count_files().await.unwrap();
        assert_eq!(removed.old_versions, 2);
        assert_eq!(
            removed.bytes_removed,
            before_count.num_bytes - after_count.num_bytes
        );

        assert_eq!(after_count.num_index_files, 0);
        assert_eq!(after_count.num_data_files, 1);
        assert_eq!(after_count.num_manifest_files, 1);
        assert_eq!(after_count.num_tx_files, 1);
    }

    #[tokio::test]
    async fn clean_old_delete_files() {
        let fixture = MockDatasetFixture::try_new().unwrap();
        let mut data_gen = BatchGenerator::new().col(Box::new(
            IncrementingInt32::new().named("filter_me".to_owned()),
        ));

        fixture.create_with_data(data_gen.batch(16)).await.unwrap();
        fixture.append_data(data_gen.batch(16)).await.unwrap();
        // This will keep some data from the appended file and should
        // completely remove the first file
        fixture.delete_data("filter_me < 20").await.unwrap();
        MockClock::set_system_time(TimeDelta::try_days(10).unwrap().to_std().unwrap());
        fixture.overwrite_data(data_gen.batch(16)).await.unwrap();
        // This will delete half of the last fragment
        fixture.delete_data("filter_me >= 40").await.unwrap();

        let before_count = fixture.count_files().await.unwrap();
        assert_eq!(before_count.num_data_files, 3);
        assert_eq!(before_count.num_delete_files, 2);
        assert_eq!(before_count.num_manifest_files, 5);

        let before = utc_now() - TimeDelta::try_days(8).unwrap();
        let removed = fixture.run_cleanup(before).await.unwrap();

        let after_count = fixture.count_files().await.unwrap();
        assert_eq!(removed.old_versions, 3);
        assert_eq!(
            removed.bytes_removed,
            before_count.num_bytes - after_count.num_bytes
        );

        assert_eq!(after_count.num_data_files, 1);
        assert_eq!(after_count.num_delete_files, 1);
        assert_eq!(after_count.num_manifest_files, 2);
        assert_eq!(after_count.num_tx_files, 2);

        // Ensure we can still read the dataset
        let row_count_after = fixture.count_rows().await.unwrap();
        assert_eq!(row_count_after, 8);
    }

    #[tokio::test]
    async fn cleanup_collects_removed_file_metrics() {
        let fixture = MockDatasetFixture::try_new().unwrap();
        let row_count = 512;
        let mut data_gen = BatchGenerator::new()
            .col(Box::new(
                IncrementingInt32::new().named("filter_me".to_owned()),
            ))
            .col(Box::new(RandomVector::new().named("indexable".to_owned())));

        fixture
            .create_with_data(data_gen.batch(row_count))
            .await
            .unwrap();
        fixture
            .append_data(data_gen.batch(row_count))
            .await
            .unwrap();
        fixture.create_some_index().await.unwrap();
        fixture.delete_data("filter_me < 20").await.unwrap();
        MockClock::set_system_time(TimeDelta::try_days(10).unwrap().to_std().unwrap());
        fixture
            .overwrite_data(data_gen.batch(row_count))
            .await
            .unwrap();
        fixture.delete_data("filter_me >= 40").await.unwrap();

        let before_count = fixture.count_files().await.unwrap();
        let removed = fixture
            .run_cleanup(utc_now() - TimeDelta::try_days(8).unwrap())
            .await
            .unwrap();
        let after_count = fixture.count_files().await.unwrap();

        let data_files_removed = (before_count.num_data_files - after_count.num_data_files) as u64;
        let transaction_files_removed =
            (before_count.num_tx_files - after_count.num_tx_files) as u64;
        let index_files_removed =
            (before_count.num_index_files - after_count.num_index_files) as u64;
        let deletion_files_removed =
            (before_count.num_delete_files - after_count.num_delete_files) as u64;

        assert_eq!(removed.data_files_removed, data_files_removed);
        assert_eq!(removed.transaction_files_removed, transaction_files_removed);
        assert_eq!(removed.index_files_removed, index_files_removed);
        assert_eq!(removed.deletion_files_removed, deletion_files_removed);
        assert_gt!(removed.data_files_removed, 0);
        assert_gt!(removed.transaction_files_removed, 0);
        assert_gt!(removed.index_files_removed, 0);
        assert_gt!(removed.deletion_files_removed, 0);
    }

    /// A branch reaches its parent's files through `base_id`, and
    /// `retain_branch_lineage_files` promotes those into the parent's keep set. An
    /// overlay inherited that way must be promoted too, or the parent deletes a
    /// file the branch still reads.
    #[tokio::test]
    async fn lineage_retention_covers_inherited_overlay_files() {
        use lance_table::format::overlay::{DataOverlayFile, OverlayCoverage};
        use roaring::RoaringBitmap;

        let fixture = MockDatasetFixture::try_new().unwrap();
        fixture.create_some_data().await.unwrap();

        // Give the parent an overlay, then branch from it: `shallow_clone` stamps
        // the overlay's `base_id` so the branch resolves it against the parent.
        let mut dataset = fixture.open().await.unwrap();
        let mut fragments: Vec<_> = dataset
            .get_fragments()
            .iter()
            .map(|f| f.metadata().clone())
            .collect();
        let mut overlay_file = fragments[0].files[0].clone();
        overlay_file.path = "overlay.lance".to_string();
        fragments[0].overlays = vec![DataOverlayFile {
            data_file: overlay_file,
            coverage: OverlayCoverage::Shared(Arc::new(RoaringBitmap::from_iter([0_u32]))),
            committed_version: dataset.manifest.version,
        }];
        let transaction = Transaction::new(
            dataset.manifest.version,
            Operation::Overwrite {
                fragments,
                schema: dataset.schema().clone(),
                config_upsert_values: None,
                initial_bases: None,
            },
            None,
        );
        dataset
            .apply_commit(transaction, &Default::default(), &Default::default())
            .await
            .unwrap();

        let root_version = dataset.manifest.version;
        let branch = fixture
            .create_branch_and_load(&mut dataset, "child", (None, None))
            .await
            .unwrap();
        let branch_fragments = branch.get_fragments();
        let inherited = &branch_fragments[0].metadata().overlays[0].data_file;
        assert!(
            inherited.base_id.is_some(),
            "the branch must reach the parent's overlay through base_id"
        );

        // The parent's cleanup walks the branch's manifest and must promote that
        // overlay out of `verified_files`.
        let task = CleanupTask::new(
            &dataset.cleanup(
                CleanupPolicyBuilder::default()
                    .before_timestamp(utc_now())
                    .build(),
            ),
            CleanupAction::Execute,
        )
        .unwrap();
        let inspection = task.process_manifests(&HashSet::new()).await.unwrap();
        // Queue the branch root for removal on both sides; rescuing the
        // manifest must rescue its store record with it.
        let inspection = Mutex::new(inspection);
        {
            let mut queued = inspection.lock().unwrap();
            queued.old_manifests.insert(
                Path::from("_versions/root.manifest"),
                ExpiredManifest {
                    version: root_version,
                    size_bytes: None,
                },
            );
            queued
                .retired_records
                .insert(root_version, "root-identity".to_string());
        }
        task.process_branch_referenced_manifests(
            &branch.base,
            branch.manifest_location.clone(),
            root_version,
            &inspection,
        )
        .await
        .unwrap();
        let inspection = inspection.into_inner().unwrap();
        assert!(
            !inspection
                .old_manifests
                .values()
                .any(|expired| expired.version == root_version)
        );
        assert!(
            !inspection.retired_records.contains_key(&root_version),
            "a retained branch root must not be retired from authoritative history"
        );
        let referenced_branches = task.find_referenced_branches().await.unwrap();
        let inspection = task
            .retain_branch_lineage_files(inspection, &referenced_branches, &HashSet::new())
            .await
            .unwrap();

        let overlay_path = Path::from("data/overlay.lance");
        assert!(
            inspection
                .referenced_files
                .data_paths
                .contains(&overlay_path),
            "the inherited overlay must be promoted into the parent's keep set"
        );
    }

    /// A keep set built from `fragment.files` alone omits overlay data files, so
    /// cleanup would delete live data.
    #[tokio::test]
    async fn keep_set_covers_referenced_overlay_files() {
        use lance_table::format::overlay::{DataOverlayFile, OverlayCoverage};
        use roaring::RoaringBitmap;

        let fixture = MockDatasetFixture::try_new().unwrap();
        fixture.create_some_data().await.unwrap();

        // The overlay file need not exist: the keep set comes from manifest
        // metadata alone.
        let mut dataset = fixture.open().await.unwrap();
        let mut fragments: Vec<_> = dataset
            .get_fragments()
            .iter()
            .map(|f| f.metadata().clone())
            .collect();
        let mut overlay_file = fragments[0].files[0].clone();
        overlay_file.path = "overlay.lance".to_string();
        fragments[0].overlays = vec![DataOverlayFile {
            data_file: overlay_file,
            coverage: OverlayCoverage::Shared(Arc::new(RoaringBitmap::from_iter([0_u32]))),
            committed_version: dataset.manifest.version,
        }];
        let transaction = Transaction::new(
            dataset.manifest.version,
            Operation::Overwrite {
                fragments,
                schema: dataset.schema().clone(),
                config_upsert_values: None,
                initial_bases: None,
            },
            None,
        );
        dataset
            .apply_commit(transaction, &Default::default(), &Default::default())
            .await
            .unwrap();

        let task = CleanupTask::new(
            &dataset.cleanup(
                CleanupPolicyBuilder::default()
                    .before_timestamp(utc_now())
                    .build(),
            ),
            CleanupAction::Execute,
        )
        .unwrap();
        let inspection = task.process_manifests(&HashSet::new()).await.unwrap();
        let kept: HashSet<&Path> = inspection
            .referenced_files
            .data_paths
            .iter()
            .chain(inspection.verified_files.data_paths.iter())
            .collect();

        let overlay_path = Path::from("data/overlay.lance");
        assert!(
            kept.contains(&overlay_path),
            "the overlay's data file must be in the keep set, got {kept:?}"
        );
    }

    #[tokio::test]
    async fn dont_clean_index_data_files() {
        // Indexes have .lance files in them that are not referenced
        // by any fragment.  We need to make sure the cleanup routine
        // doesn't over-zealously delete these
        let fixture = MockDatasetFixture::try_new().unwrap();
        MockClock::set_system_time(TimeDelta::try_days(10).unwrap().to_std().unwrap());
        fixture.create_some_data().await.unwrap();
        fixture.create_some_index().await.unwrap();

        let before_count = fixture.count_files().await.unwrap();
        let before = utc_now() - TimeDelta::try_days(8).unwrap();
        let removed = fixture.run_cleanup(before).await.unwrap();
        assert_eq!(removed.old_versions, 0);
        assert_eq!(removed.bytes_removed, 0);

        let after_count = fixture.count_files().await.unwrap();

        assert_eq!(before_count, after_count);
    }

    #[tokio::test]
    async fn cleanup_removes_preexisting_empty_index_directories() {
        let fixture = MockDatasetFixture::try_new().unwrap();
        fixture.create_some_data().await.unwrap();

        let mut dataset = fixture.open().await.unwrap();
        let field_id = dataset.schema().field("indexable").unwrap().id;
        let stale_uuid = Uuid::new_v4();
        let nested_stale_uuid = Uuid::new_v4();
        let referenced_uuid = Uuid::new_v4();

        std::fs::create_dir_all(fixture.local_index_dir(stale_uuid)).unwrap();
        std::fs::create_dir_all(
            fixture
                .local_index_dir(nested_stale_uuid)
                .join("empty_nested_dir"),
        )
        .unwrap();
        std::fs::create_dir_all(fixture.local_index_dir(referenced_uuid)).unwrap();

        let referenced_index = dummy_index_metadata(&dataset, field_id, referenced_uuid, [0_u32]);
        let create_index_tx = Transaction::new(
            dataset.manifest.version,
            Operation::CreateIndex {
                new_indices: vec![referenced_index],
                removed_indices: vec![],
            },
            None,
        );
        dataset
            .apply_commit(create_index_tx, &Default::default(), &Default::default())
            .await
            .unwrap();

        let real_now = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap();
        MockClock::set_system_time(real_now + TimeDelta::try_days(10).unwrap().to_std().unwrap());
        let in_progress_uuid = Uuid::new_v4();
        write_dummy_index_artifact(&dataset, in_progress_uuid)
            .await
            .unwrap();
        let in_progress_empty_dir = fixture
            .local_index_dir(in_progress_uuid)
            .join("empty_in_progress_dir");
        std::fs::create_dir_all(&in_progress_empty_dir).unwrap();

        let removed = fixture
            .run_cleanup(utc_now() - TimeDelta::try_days(7).unwrap())
            .await
            .unwrap();

        assert_eq!(removed.index_files_removed, 0);
        assert!(!fixture.local_index_dir(stale_uuid).exists());
        assert!(!fixture.local_index_dir(nested_stale_uuid).exists());
        assert!(fixture.local_index_dir(referenced_uuid).exists());
        assert!(fixture.local_index_dir(in_progress_uuid).exists());
        assert!(in_progress_empty_dir.exists());
    }

    #[rstest]
    #[case::default_policy(false, true)]
    #[case::delete_unverified(true, false)]
    #[tokio::test]
    async fn cleanup_applies_unverified_policy_to_fresh_empty_index_directory(
        #[case] delete_unverified: bool,
        #[case] should_preserve: bool,
    ) {
        let real_now = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap();
        MockClock::set_system_time(real_now);

        let fixture = MockDatasetFixture::try_new().unwrap();
        fixture.create_some_data().await.unwrap();
        let in_progress_dir = fixture.local_index_dir(Uuid::new_v4());
        std::fs::create_dir_all(&in_progress_dir).unwrap();

        fixture
            .run_cleanup_with_override(
                utc_now() - TimeDelta::try_days(7).unwrap(),
                Some(delete_unverified),
                None,
            )
            .await
            .unwrap();

        assert_eq!(in_progress_dir.exists(), should_preserve);
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn cleanup_does_not_remove_empty_directory_through_index_symlink() {
        let fixture = MockDatasetFixture::try_new().unwrap();
        fixture.create_some_data().await.unwrap();

        let outside = tempfile::tempdir().unwrap();
        let outside_empty_dir = outside.path().join("must_remain");
        std::fs::create_dir_all(&outside_empty_dir).unwrap();

        let link_path = fixture.local_index_dir(Uuid::new_v4());
        std::fs::create_dir_all(link_path.parent().unwrap()).unwrap();
        std::os::unix::fs::symlink(outside.path(), &link_path).unwrap();

        fixture
            .run_cleanup_with_override(utc_now(), Some(true), None)
            .await
            .unwrap();

        assert!(outside_empty_dir.exists());
        assert!(link_path.is_symlink());
    }

    #[tokio::test]
    async fn cleanup_old_replaced_segment_keeps_still_referenced_segments() {
        let fixture = MockDatasetFixture::try_new().unwrap();
        fixture.create_some_data().await.unwrap();

        let mut dataset = fixture.open().await.unwrap();
        let field_id = dataset.schema().field("indexable").unwrap().id;

        let seg_a = Uuid::new_v4();
        let seg_b = Uuid::new_v4();
        write_dummy_index_artifact(&dataset, seg_a).await.unwrap();
        write_dummy_index_artifact(&dataset, seg_b).await.unwrap();

        let index_a = dummy_index_metadata(&dataset, field_id, seg_a, [0_u32]);
        let index_b = dummy_index_metadata(&dataset, field_id, seg_b, [1_u32]);
        let initial_tx = Transaction::new(
            dataset.manifest.version,
            Operation::CreateIndex {
                new_indices: vec![index_a.clone(), index_b.clone()],
                removed_indices: vec![],
            },
            None,
        );
        dataset
            .apply_commit(initial_tx, &Default::default(), &Default::default())
            .await
            .unwrap();

        MockClock::set_system_time(TimeDelta::try_days(10).unwrap().to_std().unwrap());

        let seg_c = Uuid::new_v4();
        write_dummy_index_artifact(&dataset, seg_c).await.unwrap();
        let index_c = dummy_index_metadata(&dataset, field_id, seg_c, [2_u32]);
        let replace_tx = Transaction::new(
            dataset.manifest.version,
            Operation::CreateIndex {
                new_indices: vec![index_c.clone()],
                removed_indices: vec![index_a.clone()],
            },
            None,
        );
        dataset
            .apply_commit(replace_tx, &Default::default(), &Default::default())
            .await
            .unwrap();

        let removed = fixture
            .run_cleanup(utc_now() - TimeDelta::try_days(7).unwrap())
            .await
            .unwrap();

        assert_eq!(removed.index_files_removed, 2);
        assert!(!fixture.local_index_dir(seg_a).exists());
        assert!(
            !dataset
                .object_store
                .as_ref()
                .exists(
                    &dataset
                        .indices_dir()
                        .clone()
                        .join(seg_a.to_string())
                        .join("index.idx")
                )
                .await
                .unwrap()
        );
        assert!(
            dataset
                .object_store
                .as_ref()
                .exists(
                    &dataset
                        .indices_dir()
                        .clone()
                        .join(seg_b.to_string())
                        .join("index.idx")
                )
                .await
                .unwrap()
        );
        assert!(
            dataset
                .object_store
                .as_ref()
                .exists(
                    &dataset
                        .indices_dir()
                        .clone()
                        .join(seg_c.to_string())
                        .join("index.idx")
                )
                .await
                .unwrap()
        );
    }

    #[tokio::test]
    async fn cleanup_recent_replaced_index_with_short_retention() {
        let fixture = MockDatasetFixture::try_new().unwrap();
        fixture.create_some_data().await.unwrap();
        fixture.append_some_data().await.unwrap();

        let mut dataset = fixture.open().await.unwrap();
        let field_id = dataset.schema().field("indexable").unwrap().id;
        let old_uuid = Uuid::new_v4();
        let current_uuid = Uuid::new_v4();

        let old_index = dummy_index_metadata(&dataset, field_id, old_uuid, [0_u32, 1]);
        dataset
            .apply_commit(
                Transaction::new(
                    dataset.manifest.version,
                    Operation::CreateIndex {
                        new_indices: vec![old_index.clone()],
                        removed_indices: vec![],
                    },
                    None,
                ),
                &Default::default(),
                &Default::default(),
            )
            .await
            .unwrap();

        MockClock::set_system_time(TimeDelta::try_minutes(1).unwrap().to_std().unwrap());
        let current_index = dummy_index_metadata(&dataset, field_id, current_uuid, [0_u32, 1]);
        dataset
            .apply_commit(
                Transaction::new(
                    dataset.manifest.version,
                    Operation::CreateIndex {
                        new_indices: vec![current_index],
                        removed_indices: vec![old_index],
                    },
                    None,
                ),
                &Default::default(),
                &Default::default(),
            )
            .await
            .unwrap();

        // Model index artifacts whose storage timestamp is newer than the retained
        // manifest. UUID verification must not be hidden by the manifest cutoff.
        MockClock::set_system_time(TimeDelta::try_minutes(2).unwrap().to_std().unwrap());
        write_dummy_index_artifact(&dataset, old_uuid)
            .await
            .unwrap();
        write_dummy_index_artifact(&dataset, current_uuid)
            .await
            .unwrap();

        let short_retention = TimeDelta::try_seconds(30).unwrap();
        let removed = fixture
            .run_cleanup(utc_now() - short_retention)
            .await
            .unwrap();
        assert_eq!(removed.old_versions, 3);
        assert_eq!(removed.index_files_removed, 2);

        let old_index_file = dataset
            .indices_dir()
            .join(old_uuid.to_string())
            .join("index.idx");
        let current_index_file = dataset
            .indices_dir()
            .join(current_uuid.to_string())
            .join("index.idx");
        assert!(
            !dataset
                .object_store
                .as_ref()
                .exists(&old_index_file)
                .await
                .unwrap()
        );
        assert!(
            dataset
                .object_store
                .as_ref()
                .exists(&current_index_file)
                .await
                .unwrap()
        );
    }

    #[tokio::test]
    async fn cleanup_old_uncommitted_index_artifacts() {
        let fixture = MockDatasetFixture::try_new().unwrap();
        fixture.create_some_data().await.unwrap();

        let dataset = fixture.open().await.unwrap();
        let staging_uuid = Uuid::new_v4();
        let shard_uuid = Uuid::new_v4();
        let built_segment_uuid = Uuid::new_v4();

        write_dummy_staging_partial(&dataset, staging_uuid, shard_uuid)
            .await
            .unwrap();
        write_dummy_index_artifact(&dataset, built_segment_uuid)
            .await
            .unwrap();

        MockClock::set_system_time(TimeDelta::try_days(10).unwrap().to_std().unwrap());

        let removed = fixture
            .run_cleanup(utc_now() - TimeDelta::try_days(7).unwrap())
            .await
            .unwrap();

        assert_eq!(removed.old_versions, 0);
        assert_eq!(removed.index_files_removed, 4);
        assert!(!fixture.local_index_dir(staging_uuid).exists());
        assert!(!fixture.local_index_dir(built_segment_uuid).exists());
        assert!(
            !dataset
                .object_store
                .as_ref()
                .exists(
                    &dataset
                        .indices_dir()
                        .clone()
                        .join(staging_uuid.to_string())
                        .join(format!("partial_{}", shard_uuid))
                        .join("index.idx"),
                )
                .await
                .unwrap()
        );
        assert!(
            !dataset
                .object_store
                .as_ref()
                .exists(
                    &dataset
                        .indices_dir()
                        .clone()
                        .join(built_segment_uuid.to_string())
                        .join("index.idx"),
                )
                .await
                .unwrap()
        );
    }

    #[rstest]
    #[case::aged(false, 10)]
    #[case::delete_unverified(true, 1)]
    #[tokio::test]
    async fn cleanup_failed_commit_data_file(
        #[case] delete_unverified: bool,
        #[case] cleanup_day: i64,
    ) {
        // We should clean up data files that are written but the commit failed
        // for whatever reason

        MockClock::set_system_time(std::time::Duration::from_secs(0));
        let mut fixture = MockDatasetFixture::try_new().unwrap();
        fixture.create_some_data().await.unwrap();

        // The failed write is newer than every manifest, so a retained-manifest
        // listing cutoff must not hide it from the age and policy checks.
        MockClock::set_system_time(TimeDelta::try_days(1).unwrap().to_std().unwrap());
        fixture.block_commits();
        assert!(fixture.append_some_data().await.is_err());
        MockClock::set_system_time(TimeDelta::try_days(cleanup_day).unwrap().to_std().unwrap());

        let before_count = fixture.count_files().await.unwrap();
        assert_eq!(before_count.num_data_files, 2);
        assert_eq!(before_count.num_manifest_files, 1);
        // Only 1 txn file: the failed commit's txn file was already cleaned up.
        assert_eq!(before_count.num_tx_files, 1);

        let removed = fixture
            .run_cleanup_with_override(
                utc_now() - TimeDelta::try_days(7).unwrap(),
                Some(delete_unverified),
                None,
            )
            .await
            .unwrap();

        let after_count = fixture.count_files().await.unwrap();
        assert_eq!(removed.old_versions, 0);
        assert_eq!(removed.data_files_removed, 1);
        assert_eq!(
            removed.bytes_removed,
            before_count.num_bytes - after_count.num_bytes
        );

        assert_eq!(after_count.num_data_files, 1);
        assert_eq!(after_count.num_manifest_files, 1);
        assert_eq!(after_count.num_tx_files, 1);
    }

    #[tokio::test]
    async fn dont_cleanup_in_progress_write() {
        // We should not cleanup data files newer than our threshold as they might
        // belong to in-progress writes

        // For testing purposes we actually create these files with a failed write
        // but the cleanup routine has no way of detecting this.  They should look
        // just like an in-progress write.
        let mut fixture = MockDatasetFixture::try_new().unwrap();
        MockClock::set_system_time(TimeDelta::try_days(10).unwrap().to_std().unwrap());
        fixture.create_some_data().await.unwrap();
        fixture.block_commits();
        assert!(fixture.append_some_data().await.is_err());

        let before_count = fixture.count_files().await.unwrap();

        let removed = fixture
            .run_cleanup(utc_now() - TimeDelta::try_days(7).unwrap())
            .await
            .unwrap();

        assert_eq!(removed.old_versions, 0);
        assert_eq!(removed.bytes_removed, 0);
        assert_eq!(removed.data_files_removed, 0);

        let after_count = fixture.count_files().await.unwrap();
        assert_eq!(before_count, after_count);
    }

    #[tokio::test]
    async fn can_recover_delete_failure() {
        // We want to make sure that an I/O error during the cleanup process doesn't
        // prevent us from running cleanup again later.
        let mut fixture = MockDatasetFixture::try_new().unwrap();
        fixture.create_some_data().await.unwrap();
        MockClock::set_system_time(TimeDelta::try_days(10).unwrap().to_std().unwrap());
        fixture.overwrite_some_data().await.unwrap();

        // The delete operation should delete the first version and its
        // data file.  However, we will block the manifest file from getting
        // cleaned up by simulating an I/O error.
        fixture.block_delete_manifest();

        let before_count = fixture.count_files().await.unwrap();
        assert_eq!(before_count.num_data_files, 2);
        assert_eq!(before_count.num_manifest_files, 2);

        // A file that cannot be deleted is counted and skipped, not fatal: one
        // transient failure must not discard a sweep over millions of objects.
        let interrupted = fixture
            .run_cleanup(utc_now() - TimeDelta::try_days(7).unwrap())
            .await
            .unwrap();
        assert_eq!(interrupted.failed_deletes, 1);
        assert_eq!(
            interrupted.old_versions, 0,
            "a manifest that failed to delete is not a removed version"
        );

        // Every candidate is attempted, so the data file goes even though the
        // manifest delete fails. This no longer depends on buffering order.
        let mid_count = fixture.count_files().await.unwrap();
        assert_eq!(mid_count.num_data_files, 1);
        assert_eq!(mid_count.num_manifest_files, 2);

        fixture.unblock_delete_manifest();

        let removed = fixture
            .run_cleanup(utc_now() - TimeDelta::try_days(7).unwrap())
            .await
            .unwrap();

        let after_count = fixture.count_files().await.unwrap();
        assert_eq!(removed.old_versions, 1);
        assert_eq!(
            removed.bytes_removed,
            mid_count.num_bytes - after_count.num_bytes
        );

        assert_eq!(after_count.num_data_files, 1);
        assert_eq!(after_count.num_manifest_files, 1);
    }

    #[rstest]
    #[case::missing(true, 0)]
    #[case::permission_denied(false, 2)]
    #[tokio::test]
    async fn cleanup_handles_wrapped_delete_errors(
        #[case] is_missing: bool,
        #[case] expected_failed_deletes: u64,
    ) {
        let fixture = MockDatasetFixture::try_new().unwrap();
        let data = || {
            lance_datagen::gen_batch()
                .col(
                    "id",
                    lance_datagen::array::step::<arrow_array::types::Int32Type>(),
                )
                .into_reader_rows(2.into(), 1.into())
        };
        fixture.create_with_data(data()).await.unwrap();
        fixture.append_data(data()).await.unwrap();
        fixture.overwrite_data(data()).await.unwrap();

        fixture.mock_store.policy.lock().unwrap().set_before_policy(
            "concurrent_data_delete",
            Arc::new(move |method, path| {
                if method == "delete" && path.extension() == Some("lance") {
                    if is_missing {
                        // Simulate another cleanup removing the file after listing.
                        std::fs::remove_file(lance_io::local::to_local_path(path))?;
                        return Err(Error::not_found(path.to_string()));
                    }
                    return Err(Error::from(std::io::Error::new(
                        std::io::ErrorKind::PermissionDenied,
                        "cleanup delete denied",
                    )));
                }
                Ok(())
            }),
        );

        let dataset = fixture.load().await.unwrap();
        let logs = tempfile::NamedTempFile::new().unwrap();
        let subscriber = tracing_subscriber::fmt()
            .with_ansi(false)
            .without_time()
            .with_max_level(tracing::Level::WARN)
            .with_writer(logs.reopen().unwrap())
            .finish();
        let _guard = tracing::subscriber::set_default(subscriber);
        let stats = dataset
            .cleanup(
                CleanupPolicyBuilder::default()
                    .versions(vec![1, 2])
                    .unwrap()
                    .build(),
            )
            .with_delete_concurrency(2)
            .unwrap()
            .execute()
            .await
            .unwrap();
        assert_eq!(stats.old_versions, 2);
        assert_eq!(stats.failed_deletes, expected_failed_deletes);
        assert_eq!(stats.data_files_removed, if is_missing { 2 } else { 0 });
        assert_eq!(
            fixture.count_files().await.unwrap().num_data_files,
            if is_missing { 1 } else { 3 }
        );
        assert_eq!(fixture.count_rows().await.unwrap(), 2);
        let output = std::fs::read_to_string(logs.path()).unwrap();
        assert_eq!(
            output
                .matches("failed to delete file; continuing and counting it")
                .count(),
            usize::from(!is_missing),
            "missing files must be silent; other failures warn once per sweep: {output}"
        );
    }

    #[tokio::test]
    async fn cleanup_rejects_retain_zero_versions() {
        let fixture = MockDatasetFixture::try_new().unwrap();
        fixture.create_some_data().await.unwrap();

        let error = CleanupPolicyBuilder::default()
            .retain_n_versions(&fixture.open().await.unwrap(), 0)
            .await
            .err()
            .expect("retaining zero versions should return an error");

        assert!(matches!(&error, Error::InvalidInput { .. }));
        assert!(
            error
                .to_string()
                .contains("retain_versions must be greater than 0, got 0"),
            "unexpected error: {error}"
        );
    }

    #[tokio::test]
    async fn retain_n_versions_does_not_read_manifests() {
        let fixture = MockDatasetFixture::try_new().unwrap();
        fixture.create_some_data().await.unwrap();
        fixture.overwrite_some_data().await.unwrap();
        fixture.overwrite_some_data().await.unwrap();
        let dataset = fixture.open().await.unwrap();

        let _ = dataset.object_store.as_ref().io_stats_incremental();
        let policy = CleanupPolicyBuilder::default()
            .retain_n_versions(&dataset, 2)
            .await
            .unwrap()
            .build();
        let io_stats = dataset.object_store.as_ref().io_stats_incremental();

        assert_eq!(policy.before_version, Some(2));
        assert_eq!(io_stats.read_bytes, 0);
    }

    #[tokio::test]
    async fn cleanup_and_retain_3_recent_versions() {
        let fixture = MockDatasetFixture::try_new().unwrap();
        fixture.create_some_data().await.unwrap();
        let mut time = 10i64;
        for _ in 0..4 {
            MockClock::set_system_time(TimeDelta::try_seconds(time).unwrap().to_std().unwrap());
            time += 10i64;
            fixture.overwrite_some_data().await.unwrap();
        }

        let before_count = fixture.count_files().await.unwrap();
        assert_eq!(before_count.num_data_files, 5);
        assert_eq!(before_count.num_manifest_files, 5);

        // Retain 3 recent versions
        let policy = CleanupPolicyBuilder::default()
            .retain_n_versions(&fixture.open().await.unwrap(), 3)
            .await
            .unwrap()
            .build();
        let removed = fixture.run_cleanup_with_policy(policy).await.unwrap();

        let after_count = fixture.count_files().await.unwrap();
        assert_eq!(removed.old_versions, 2);
        assert_eq!(
            removed.bytes_removed,
            before_count.num_bytes - after_count.num_bytes
        );

        assert_eq!(after_count.num_data_files, 3);
        assert_eq!(after_count.num_manifest_files, 3);
        assert_eq!(
            fixture
                .open()
                .await
                .unwrap()
                .version_refs()
                .await
                .unwrap()
                .iter()
                .map(|version| version.version)
                .collect::<Vec<_>>(),
            vec![3, 4, 5]
        );
    }

    #[tokio::test]
    async fn cleanup_specific_versions_only() {
        let fixture = MockDatasetFixture::try_new().unwrap();
        fixture.create_some_data().await.unwrap();
        fixture.overwrite_some_data().await.unwrap();
        fixture.overwrite_some_data().await.unwrap();

        let before_count = fixture.count_files().await.unwrap();
        assert_eq!(before_count.num_manifest_files, 3);

        let policy = CleanupPolicyBuilder::default()
            .versions(vec![2])
            .unwrap()
            .build();
        let removed = fixture.run_cleanup_with_policy(policy).await.unwrap();

        assert_eq!(removed.old_versions, 1);

        let versions = fixture
            .open()
            .await
            .unwrap()
            .version_refs()
            .await
            .unwrap()
            .iter()
            .map(|version| version.version)
            .collect::<Vec<_>>();
        assert_eq!(versions, vec![1, 3]);
    }

    #[tokio::test]
    async fn cleanup_before_ts_and_retain_n_recent_versions() {
        let fixture = MockDatasetFixture::try_new().unwrap();
        fixture.create_some_data().await.unwrap();
        for time in (1i64..).take(4) {
            MockClock::set_system_time(TimeDelta::try_days(time).unwrap().to_std().unwrap());
            fixture.overwrite_some_data().await.unwrap();
        }

        let before_count = fixture.count_files().await.unwrap();
        assert_eq!(before_count.num_data_files, 5);
        assert_eq!(before_count.num_manifest_files, 5);

        // Retain 3 recent versions before timestamp now - 6days
        let policy = CleanupPolicyBuilder::default()
            .before_timestamp(utc_now() - TimeDelta::try_days(6).unwrap())
            .retain_n_versions(&fixture.open().await.unwrap(), 3)
            .await
            .unwrap()
            .build();
        let removed = fixture.run_cleanup_with_policy(policy).await.unwrap();
        assert_eq!(removed.old_versions, 0);

        // Retain 10 recent versions before timestamp now
        let policy = CleanupPolicyBuilder::default()
            .before_timestamp(utc_now())
            .retain_n_versions(&fixture.open().await.unwrap(), 10)
            .await
            .unwrap()
            .build();
        let removed = fixture.run_cleanup_with_policy(policy).await.unwrap();
        assert_eq!(removed.old_versions, 0);

        // Retain 3 recent versions before timestamp now - 1days
        let policy = CleanupPolicyBuilder::default()
            .before_timestamp(utc_now() - TimeDelta::try_days(2).unwrap())
            .retain_n_versions(&fixture.open().await.unwrap(), 3)
            .await
            .unwrap()
            .build();
        let removed = fixture.run_cleanup_with_policy(policy).await.unwrap();

        let after_count = fixture.count_files().await.unwrap();
        assert_eq!(removed.old_versions, 2);
        assert_eq!(
            removed.bytes_removed,
            before_count.num_bytes - after_count.num_bytes
        );
        assert_eq!(after_count.num_data_files, 3);
        assert_eq!(after_count.num_manifest_files, 3);
    }

    #[tokio::test]
    async fn cleanup_preserves_unmanaged_dirs_and_files() {
        // Ensure cleanup does not delete unmanaged directories/files under the dataset root
        // Uses MockDatasetFixture and run_cleanup_with_override to match other tests' style
        let fixture = MockDatasetFixture::try_new().unwrap();
        fixture.create_some_data().await.unwrap();

        let registry = Arc::new(ObjectStoreRegistry::default());
        let (os, base) =
            ObjectStore::from_uri_and_params(registry, &fixture.dataset_path, &fixture.os_params())
                .await
                .unwrap();

        // Create unmanaged directories/files under dataset root
        let img = base.clone().join("images").join("clip.mp4");
        let misc = base.clone().join("misc").join("notes.txt");
        let branch_file = base.clone().join("tree").join("branchA").join("data.bin");
        os.put(&img, b"video").await.unwrap();
        os.put(&misc, b"notes").await.unwrap();
        os.put(&branch_file, b"branch").await.unwrap();

        // Create a temporary manifest file that should be cleaned
        let tmp_manifest = base.clone().join("_versions").join(".tmp").join("orphan");
        os.put(&tmp_manifest, b"tmp").await.unwrap();
        // Delete the _transactions directory so that we can test that if not_found err will be swallowed
        os.remove_dir_all(base.clone().join(TRANSACTIONS_DIR))
            .await
            .unwrap();

        fixture
            .run_cleanup_with_override(utc_now(), Some(true), Some(false))
            .await
            .unwrap();

        // Temp manifest file is managed by Lance and should be removed
        assert!(!os.exists(&tmp_manifest).await.unwrap());
        // Unrelated files must remain
        assert!(os.exists(&img).await.unwrap());
        assert!(os.exists(&misc).await.unwrap());
        assert!(os.exists(&branch_file).await.unwrap());
    }

    // Lineage overview with annotated base versions:
    // - branch1 is created from main@v1
    // - branch4 is created from main@v2 (after main receives a second write)
    // - dev/branch2 is created from branch1@latest
    // - feature/nathan/branch3 is created from dev/branch2@latest
    //
    // ASCII lineage with versions:
    //    main:v1 ──▶ branch1:v1 ──▶ dev/branch2:v2 ──▶ feature/nathan/branch3:v3
    //        │
    //    (main:v2) ──▶ branch4:v2
    //
    // Cleanup policy focus (unless explicitly overridden in a test):
    // - retain_n_versions = 1: keep the latest manifest per branch
    // - referenced branches: when enabled, protect parent files referenced by descendants
    // - file counts reported per branch:
    //   manifest: number of manifest files under _versions
    //   data: .lance files under data directory
    //   tx: .txn files count under _transactions
    //   delete: deletion files count under _deletions
    //   index: index files count under _indices
    //
    // Note: branch2 is stored as "dev/branch2"; comments may refer to it as branch2 for brevity.
    // Important: auto_cleanup_hook uses policy derived from manifest config; it does not flip
    // clean_referenced_branches unless tests call cleanup_old_versions with a custom policy.
    struct LineageSetup {
        main: BranchDatasetFixture,
        branch1: BranchDatasetFixture,
        branch2: BranchDatasetFixture,
        branch3: BranchDatasetFixture,
        branch4: BranchDatasetFixture,
    }

    impl LineageSetup {
        /// Assert all branches and main are unchanged since last refresh.
        pub async fn assert_all_unchanged(&mut self) {
            self.main.assert_not_changed().await.unwrap();
            self.branch1.assert_not_changed().await.unwrap();
            self.branch2.assert_not_changed().await.unwrap();
            self.branch3.assert_not_changed().await.unwrap();
            self.branch4.assert_not_changed().await.unwrap();
        }

        /// Assert specified branches are unchanged.
        pub async fn assert_unchanged(&mut self, branches: &[&str]) {
            for &b in branches {
                match b {
                    "main" => self.main.assert_not_changed().await.unwrap(),
                    "branch1" => self.branch1.assert_not_changed().await.unwrap(),
                    "branch2" => self.branch2.assert_not_changed().await.unwrap(),
                    "branch3" => self.branch3.assert_not_changed().await.unwrap(),
                    "branch4" => self.branch4.assert_not_changed().await.unwrap(),
                    _ => panic!("unknown branch: {}", b),
                }
            }
        }

        pub async fn enable_auto_cleanup(&mut self) -> Result<()> {
            let updates = [
                ("lance.auto_cleanup.interval", "1"),
                ("lance.auto_cleanup.retain_versions", "1"),
                ("lance.auto_cleanup.referenced_branch", "true"),
            ];
            self.main.dataset.update_config(updates).await?;
            self.branch1.dataset.update_config(updates).await?;
            self.branch2.dataset.update_config(updates).await?;
            self.branch3.dataset.update_config(updates).await?;
            self.branch4.dataset.update_config(updates).await?;
            self.main.refresh().await?;
            self.branch1.refresh().await?;
            self.branch2.refresh().await?;
            self.branch3.refresh().await?;
            self.branch4.refresh().await?;
            Ok(())
        }

        pub async fn disable_auto_cleanup(&mut self) -> Result<()> {
            let updates = [
                ("lance.auto_cleanup.interval", None),
                ("lance.auto_cleanup.retain_versions", None),
                ("lance.auto_cleanup.older_than", None),
            ];
            self.main.dataset.update_config(updates).await?;
            self.branch1.dataset.update_config(updates).await?;
            self.branch2.dataset.update_config(updates).await?;
            self.branch3.dataset.update_config(updates).await?;
            self.branch4.dataset.update_config(updates).await?;
            self.main.refresh().await?;
            self.branch1.refresh().await?;
            self.branch2.refresh().await?;
            self.branch3.refresh().await?;
            self.branch4.refresh().await?;
            Ok(())
        }
    }

    // Build the lineage and configure per-branch auto-cleanup to retain latest version.
    async fn build_lineage_datasets() -> Result<LineageSetup> {
        let fixture = Arc::new(MockDatasetFixture::try_new()?);

        MockClock::set_system_time(TimeDelta::try_seconds(1).unwrap().to_std().unwrap());

        // Create main (initial write) with id and text columns for inverted index
        use arrow_array::{Int32Array, RecordBatch, RecordBatchIterator, StringArray};
        use arrow_schema::{DataType, Field};
        let ids = Int32Array::from_iter_values(0..50i32);
        let texts = StringArray::from_iter_values((0..50i32).map(|i| format!("text_{}", i)));
        let schema = Arc::new(arrow_schema::Schema::new(vec![
            Field::new("id", DataType::Int32, false),
            Field::new("text", DataType::Utf8, false),
        ]));
        let batch =
            RecordBatch::try_new(schema.clone(), vec![Arc::new(ids), Arc::new(texts)]).unwrap();
        let reader = RecordBatchIterator::new(vec![batch].into_iter().map(Ok), schema);

        Dataset::write(
            reader,
            &fixture.dataset_path,
            Some(WriteParams {
                mode: WriteMode::Create,
                store_params: Some(fixture.os_params()),
                ..Default::default()
            }),
        )
        .await?;
        let mut main = BranchDatasetFixture::new(fixture.clone(), fixture.load().await?);
        // Initial index creation and refresh counts
        main.create_text_index().await?;
        main.write_data().await?;

        // Create branch1 from main@v1, then do an initial append + deterministic delete
        let mut branch1 = BranchDatasetFixture::new(
            fixture.clone(),
            fixture
                .create_branch_and_load(&mut main.dataset, "branch1", (None, None))
                .await?,
        );
        branch1.write_data().await?;

        // Create branch2 from branch1@latest
        let mut branch2 = BranchDatasetFixture::new(
            fixture.clone(),
            fixture
                .create_branch_and_load(&mut branch1.dataset, "dev/branch2", ("branch1", None))
                .await?,
        );
        branch2.write_data().await?;

        // Create branch3 from branch2@latest, initial append + delete
        let mut branch3 = BranchDatasetFixture::new(
            fixture.clone(),
            fixture
                .create_branch_and_load(
                    &mut branch2.dataset,
                    "feature/nathan/branch3",
                    ("dev/branch2", None),
                )
                .await?,
        );
        branch3.write_data().await?;

        // Create branch4 from a new version in main
        main.write_data().await?;
        let mut branch4 = BranchDatasetFixture::new(
            fixture.clone(),
            fixture
                .create_branch_and_load(&mut main.dataset, "branch4", (None, None))
                .await?,
        );
        branch4.write_data().await?;

        let mut lineage = LineageSetup {
            main,
            branch1,
            branch2,
            branch3,
            branch4,
        };

        lineage.disable_auto_cleanup().await?;
        Ok(lineage)
    }

    // BranchDatasetFixture combines dataset with branch-specific state and file counting.
    // It provides:
    // - Shared fixture for temporary directory and mock store
    // - Dataset holding for stateful operations (checkout, write, etc.)
    // - File counting for cleanup verification
    struct BranchDatasetFixture {
        fixture: Arc<MockDatasetFixture>,
        dataset: Dataset,
        counts: FileCounts,
    }

    impl BranchDatasetFixture {
        fn new(fixture: Arc<MockDatasetFixture>, dataset: Dataset) -> Self {
            Self {
                fixture,
                dataset,
                counts: FileCounts {
                    num_manifest_files: 0,
                    num_data_files: 0,
                    num_tx_files: 0,
                    num_delete_files: 0,
                    num_index_files: 0,
                    num_bytes: 0,
                },
            }
        }

        // Create a full-text index (Inverted) on the "text" column once.
        // We only create this on main during dataset creation. Branches inherit the index configuration.
        async fn create_text_index(&mut self) -> Result<()> {
            use crate::index::DatasetIndexExt;
            use lance_index::IndexType;
            use lance_index::scalar::InvertedIndexParams;
            let params = InvertedIndexParams::default();
            self.dataset
                .create_index(&["text"], IndexType::Inverted, None, &params, true)
                .await?;
            Ok(())
        }

        // Append a batch, then read exactly one row and delete that row; finally optimize indices.
        async fn append_delete_and_optimize_index(&mut self) -> Result<()> {
            // Append a small batch with id and text columns
            self.write_batch(5).await?;
            // Delete the last row to create a deletion file
            self.delete_last_row().await?;
            // Optimize indices after write and delete
            use lance_index::optimize::OptimizeOptions;
            self.dataset
                .optimize_indices(&OptimizeOptions::merge(1))
                .await?;
            Ok(())
        }

        // Append a batch with id and text columns.
        async fn write_batch(&mut self, rows: i32) -> Result<()> {
            use crate::dataset::WriteParams;
            use arrow_array::{Int32Array, RecordBatch, RecordBatchIterator, StringArray};
            use arrow_schema::{DataType, Field};

            let ids = Int32Array::from_iter_values(0..rows);
            let texts = StringArray::from_iter_values((0..rows).map(|i| format!("text_{}", i)));
            let schema = Arc::new(arrow_schema::Schema::new(vec![
                Field::new("id", DataType::Int32, false),
                Field::new("text", DataType::Utf8, false),
            ]));
            let batch =
                RecordBatch::try_new(schema.clone(), vec![Arc::new(ids), Arc::new(texts)]).unwrap();
            let reader = RecordBatchIterator::new(vec![batch].into_iter().map(Ok), schema);

            self.dataset
                .append(
                    reader,
                    Some(WriteParams {
                        mode: WriteMode::Append,
                        store_params: Some(self.fixture.os_params()),
                        ..Default::default()
                    }),
                )
                .await?;
            self.dataset.checkout_latest().await?;
            Ok(())
        }

        // Delete the last row to generate a deletion file.
        async fn delete_last_row(&mut self) -> Result<()> {
            let batch = self.dataset.scan().with_row_id().try_into_batch().await?;
            if batch.num_rows() > 0 {
                let row_id_col = batch.column_by_name(lance_core::ROW_ID).unwrap();
                let uint64_array = row_id_col.as_any().downcast_ref::<UInt64Array>().unwrap();
                let max_row_id = compute::max(uint64_array).unwrap_or(0);
                self.dataset
                    .delete(&format!("_rowid = {}", max_row_id))
                    .await?;
            }
            Ok(())
        }

        // Update counters by listing authoritative branch directories instead of reading the latest manifest.
        async fn refresh(&mut self) -> Result<()> {
            use futures::TryStreamExt;
            let branch_path = self.dataset.base.clone();

            // Count files in a directory, filtering by optional extension(s).
            async fn count_dir(
                os: &ObjectStore,
                dir: &Path,
                exts: Option<&[&str]>,
            ) -> Result<usize> {
                let mut count = 0usize;
                let mut s = os.read_dir_all(dir, None);
                while let Some(meta) = s.try_next().await? {
                    match exts {
                        Some(exts) => {
                            if let Some(e) = meta.location.extension()
                                && exts.contains(&e)
                            {
                                count += 1;
                            }
                        }
                        None => count += 1,
                    }
                }
                Ok(count)
            }

            let manifest_dir = branch_path.clone().join("_versions");
            self.counts.num_manifest_files = count_dir(
                &self.dataset.object_store,
                &manifest_dir,
                Some(&["manifest"]),
            )
            .await
            .unwrap_or(0);

            // Transactions: count files under _transactions (extension .txn)
            let txn_dir = branch_path.clone().join("_transactions");
            self.counts.num_tx_files =
                count_dir(&self.dataset.object_store, &txn_dir, Some(&["txn"]))
                    .await
                    .unwrap_or(0);

            // Indices: count files under _indices
            let idx_dir = branch_path.clone().join(crate::dataset::INDICES_DIR);
            self.counts.num_index_files = count_dir(&self.dataset.object_store, &idx_dir, None)
                .await
                .unwrap_or(0);

            // Deletions: count files under _deletions (extensions .arrow / .bin)
            let del_dir = branch_path.clone().join("_deletions");
            self.counts.num_delete_files = count_dir(
                &self.dataset.object_store,
                &del_dir,
                Some(&["arrow", "bin"]),
            )
            .await
            .unwrap_or(0);

            // Data files: count .lance files under data/
            let data_dir = branch_path.clone().join(crate::dataset::DATA_DIR);
            self.counts.num_data_files =
                count_dir(&self.dataset.object_store, &data_dir, Some(&["lance"]))
                    .await
                    .unwrap_or(0);

            Ok(())
        }

        async fn count_data(&self) -> Result<usize> {
            use futures::TryStreamExt;
            let mut count = 0usize;
            let mut s = self.dataset.scan().try_into_stream().await?;
            while let Some(_batch) = s.try_next().await? {
                count += 1;
            }
            Ok(count)
        }

        // Strict equality assertion for all counters.
        async fn assert_not_changed(&mut self) -> Result<()> {
            let pre_counts = self.counts;
            let pre_data_count = self.count_data().await?;

            self.refresh().await?;
            assert_eq!(
                self.counts.num_manifest_files,
                pre_counts.num_manifest_files
            );
            assert_eq!(self.counts.num_data_files, pre_counts.num_data_files);
            assert_eq!(self.counts.num_tx_files, pre_counts.num_tx_files);
            assert_eq!(self.counts.num_delete_files, pre_counts.num_delete_files);
            assert_eq!(self.counts.num_index_files, pre_counts.num_index_files);
            assert_eq!(self.count_data().await?, pre_data_count);
            Ok(())
        }

        // Append, delete top row, and optimize indices.
        async fn write_data(&mut self) -> Result<()> {
            self.append_delete_and_optimize_index().await?;
            self.refresh().await
        }

        // Compact files for a given branch and optimize indices to stabilize index files.
        async fn compact(&mut self) -> Result<()> {
            use crate::dataset::optimize::{CompactionOptions, compact_files};
            compact_files(&mut self.dataset, CompactionOptions::default(), None).await?;
            self.refresh().await
        }

        async fn run_cleanup(&mut self) -> Result<RemovalStats> {
            let policy = CleanupPolicyBuilder::default()
                .error_if_tagged_old_versions(false)
                .retain_n_versions(&self.dataset, 1)
                .await?
                .build();
            self.run_cleanup_inner(policy).await
        }

        async fn run_cleanup_with_referenced_branches(&mut self) -> Result<RemovalStats> {
            let policy = CleanupPolicyBuilder::default()
                .error_if_tagged_old_versions(false)
                .clean_referenced_branches(true)
                .retain_n_versions(&self.dataset, 1)
                .await?
                .build();
            self.run_cleanup_inner(policy).await
        }

        async fn explain_cleanup_with_referenced_branches(&mut self) -> Result<CleanupExplanation> {
            let policy = CleanupPolicyBuilder::default()
                .error_if_tagged_old_versions(false)
                .clean_referenced_branches(true)
                .retain_n_versions(&self.dataset, 1)
                .await?
                .build();
            self.dataset.checkout_latest().await?;
            self.dataset.cleanup(policy).explain().await
        }

        async fn run_cleanup_inner(&mut self, policy: CleanupPolicy) -> Result<RemovalStats> {
            let pre_count = self.count_data().await?;
            self.dataset.checkout_latest().await?;
            let stats = cleanup_old_versions(&self.dataset, policy).await;
            self.refresh().await?;
            // Assert data could be read again and did't change
            assert_eq!(self.count_data().await?, pre_count);
            stats
        }
    }

    // ===================== Tests =====================
    #[tokio::test]
    async fn cleanup_lineage_branch1() {
        let mut setup = build_lineage_datasets().await.unwrap();

        setup.branch1.write_data().await.unwrap();
        setup.branch1.run_cleanup().await.unwrap();
        // Branch2 and branch3 hold references from branch1:
        // - 1 manifest file
        // - 1 data file
        // - 1 deletion file
        // - 4 index files
        // The left is the counts for the latest version of appending
        assert_eq!(setup.branch1.counts.num_manifest_files, 2);
        assert_eq!(setup.branch1.counts.num_data_files, 2);
        assert_eq!(setup.branch1.counts.num_tx_files, 1);
        assert_eq!(setup.branch1.counts.num_delete_files, 2);
        assert_eq!(setup.branch1.counts.num_index_files, 14);
        setup.assert_all_unchanged().await;

        setup.branch1.compact().await.unwrap();
        setup.branch1.run_cleanup().await.unwrap();
        // Branch2 and branch3 hold references from branch1:
        // - 1 manifest file
        // - 1 data file
        // - 1 deletion file
        // - 4 index files
        // The left (1, 1, 1, 0, 4) is the counts for the latest version of compaction
        assert_eq!(setup.branch1.counts.num_manifest_files, 2);
        assert_eq!(setup.branch1.counts.num_data_files, 2);
        assert_eq!(setup.branch1.counts.num_tx_files, 1);
        assert_eq!(setup.branch1.counts.num_delete_files, 1);
        assert_eq!(setup.branch1.counts.num_index_files, 14);
        setup.assert_all_unchanged().await;

        // Now we clean the referenced files of branch1 by branch2 and branch3
        setup.branch2.compact().await.unwrap();
        setup.branch3.compact().await.unwrap();
        setup.branch3.run_cleanup().await.unwrap();
        setup.branch2.run_cleanup().await.unwrap();
        // Only the latest manifest is retained.
        // (1, 1, 1, 0, 4) is the counts for the latest version of compaction
        assert_eq!(setup.branch2.counts.num_manifest_files, 1);
        assert_eq!(setup.branch2.counts.num_data_files, 1);
        assert_eq!(setup.branch2.counts.num_tx_files, 1);
        assert_eq!(setup.branch2.counts.num_delete_files, 0);
        assert_eq!(setup.branch2.counts.num_index_files, 7);
        // Only the latest manifest is retained.
        // (1, 1, 1, 0, 4) is the counts for the latest version of compaction
        assert_eq!(setup.branch3.counts.num_manifest_files, 1);
        assert_eq!(setup.branch3.counts.num_data_files, 1);
        assert_eq!(setup.branch3.counts.num_tx_files, 1);
        assert_eq!(setup.branch3.counts.num_delete_files, 0);
        assert_eq!(setup.branch3.counts.num_index_files, 7);
        setup.branch1.run_cleanup().await.unwrap();

        // Only the latest manifest is retained.
        // (1, 1, 1, 0, 4) is the counts for the latest version of compaction
        assert_eq!(setup.branch1.counts.num_manifest_files, 1);
        assert_eq!(setup.branch1.counts.num_data_files, 1);
        assert_eq!(setup.branch1.counts.num_tx_files, 1);
        assert_eq!(setup.branch1.counts.num_delete_files, 0);
        assert_eq!(setup.branch1.counts.num_index_files, 7);
        setup.assert_all_unchanged().await;
    }

    #[tokio::test]
    async fn cleanup_lineage_branch3() {
        let mut setup = build_lineage_datasets().await.unwrap();

        setup.branch3.write_data().await.unwrap();
        setup.branch3.run_cleanup().await.unwrap();
        // Two writes produced:
        // - 2 data files
        // - 2 deletion files
        assert_eq!(setup.branch3.counts.num_manifest_files, 1);
        assert_eq!(setup.branch3.counts.num_data_files, 2);
        assert_eq!(setup.branch3.counts.num_tx_files, 1);
        assert_eq!(setup.branch3.counts.num_delete_files, 2);
        assert_eq!(setup.branch3.counts.num_index_files, 7);
        setup
            .assert_unchanged(&["branch1", "branch2", "branch4", "main"])
            .await;

        setup.branch2.compact().await.unwrap();
        setup.branch2.run_cleanup().await.unwrap();
        // Branch3 hold references from branch2:
        // - 1 manifest file
        // - 1 data file
        // - 1 deletion file
        // The left is the counts for the latest version of compaction
        assert_eq!(setup.branch2.counts.num_manifest_files, 2);
        assert_eq!(setup.branch2.counts.num_data_files, 2);
        assert_eq!(setup.branch2.counts.num_tx_files, 1);
        assert_eq!(setup.branch2.counts.num_delete_files, 1);
        assert_eq!(setup.branch2.counts.num_index_files, 7);

        setup.branch3.compact().await.unwrap();
        setup.branch3.run_cleanup().await.unwrap();
        // Only the latest manifest is retained.
        // (1, 1, 1, 0, 4) is the counts for the latest version
        assert_eq!(setup.branch3.counts.num_manifest_files, 1);
        assert_eq!(setup.branch3.counts.num_data_files, 1);
        assert_eq!(setup.branch3.counts.num_tx_files, 1);
        assert_eq!(setup.branch3.counts.num_delete_files, 0);
        assert_eq!(setup.branch3.counts.num_index_files, 7);
        setup
            .assert_unchanged(&["branch1", "branch2", "branch4", "main"])
            .await;

        setup.branch2.compact().await.unwrap();
        setup.branch2.run_cleanup().await.unwrap();
        // Only the latest manifest is retained.
        // (1, 1, 1, 0, 4) is the counts for the latest version
        assert_eq!(setup.branch2.counts.num_manifest_files, 1);
        assert_eq!(setup.branch2.counts.num_data_files, 1);
        assert_eq!(setup.branch2.counts.num_tx_files, 1);
        assert_eq!(setup.branch2.counts.num_delete_files, 0);
        assert_eq!(setup.branch2.counts.num_index_files, 7);
    }

    #[tokio::test]
    async fn cleanup_lineage_branch4() {
        // Setup shared lineage and per-branch auto-clean config
        let mut setup = build_lineage_datasets().await.unwrap();

        setup.branch4.write_data().await.unwrap();
        setup.branch4.run_cleanup().await.unwrap();
        // Two writes produced:
        // - 2 data files
        // - 2 deletion files
        assert_eq!(setup.branch4.counts.num_manifest_files, 1);
        assert_eq!(setup.branch4.counts.num_data_files, 2);
        assert_eq!(setup.branch4.counts.num_tx_files, 1);
        assert_eq!(setup.branch4.counts.num_delete_files, 2);
        assert_eq!(setup.branch4.counts.num_index_files, 7);
        setup.assert_all_unchanged().await;

        setup.main.compact().await.unwrap();
        setup.main.run_cleanup().await.unwrap();
        // Branch1-branch2 hold references from main:
        // - 1 manifest file
        // - 2 data files
        // - 1 deletion file
        // - 4 index files
        // Branch4 holds references from main:
        // - 1 manifest file
        // - 3 data files
        // - 1 deletion file
        // - 4 index files
        // The left(1, 1, 1, 0, 0) is the counts for the latest version of compaction
        assert_eq!(setup.main.counts.num_manifest_files, 3);
        assert_eq!(setup.main.counts.num_data_files, 4);
        assert_eq!(setup.main.counts.num_tx_files, 1);
        assert_eq!(setup.main.counts.num_delete_files, 2);
        assert_eq!(setup.main.counts.num_index_files, 14);

        setup.branch4.compact().await.unwrap();
        setup.branch4.run_cleanup().await.unwrap();
        // Only the latest manifest is retained.
        // (1, 1, 1, 0, 4) is the counts of one version
        assert_eq!(setup.branch4.counts.num_manifest_files, 1);
        assert_eq!(setup.branch4.counts.num_data_files, 1);
        assert_eq!(setup.branch4.counts.num_tx_files, 1);
        assert_eq!(setup.branch4.counts.num_delete_files, 0);
        assert_eq!(setup.branch4.counts.num_index_files, 7);
        setup.assert_all_unchanged().await;

        setup.main.run_cleanup().await.unwrap();
        // Branch1-branch2 hold references from main:
        // - 1 manifest file
        // - 2 data files
        // - 1 deletion file
        // - 4 index files
        // The left(1, 1, 1, 0, 4) is the counts for the latest version of compaction
        assert_eq!(setup.main.counts.num_manifest_files, 2);
        assert_eq!(setup.main.counts.num_data_files, 3);
        assert_eq!(setup.main.counts.num_tx_files, 1);
        assert_eq!(setup.main.counts.num_delete_files, 1);
        assert_eq!(setup.main.counts.num_index_files, 14);
    }

    #[tokio::test]
    async fn cleanup_lineage_main() {
        // Setup shared lineage and per-branch auto-clean config
        let mut setup = build_lineage_datasets().await.unwrap();

        setup.main.write_data().await.unwrap();
        setup.main.run_cleanup().await.unwrap();
        // Branch1-branch2 hold references from main:
        // - 1 manifest file
        // - 2 data files
        // - 1 deletion file
        // - 4 index files(only for branch1)
        // Branch4 holds references from main:
        // - 1 manifest file
        // - 3 data files
        // - 1 deletion file
        // - 4 index files
        // The left(1, 1, 1, 1, 4) is the counts for the latest version of compaction
        assert_eq!(setup.main.counts.num_manifest_files, 3);
        assert_eq!(setup.main.counts.num_data_files, 4);
        assert_eq!(setup.main.counts.num_tx_files, 1);
        assert_eq!(setup.main.counts.num_delete_files, 3);
        assert_eq!(setup.main.counts.num_index_files, 21);
        setup.assert_all_unchanged().await;

        setup.main.compact().await.unwrap();
        setup.main.run_cleanup().await.unwrap();
        // Cleanup the deletion file
        // Produce 1 datafile and cleanup 1
        assert_eq!(setup.main.counts.num_manifest_files, 3);
        assert_eq!(setup.main.counts.num_data_files, 4);
        assert_eq!(setup.main.counts.num_tx_files, 1);
        assert_eq!(setup.main.counts.num_delete_files, 2);
        assert_eq!(setup.main.counts.num_index_files, 21);
        setup.assert_all_unchanged().await;

        setup.branch1.write_data().await.unwrap();
        setup.branch1.compact().await.unwrap();
        setup.branch2.write_data().await.unwrap();
        setup.branch2.compact().await.unwrap();
        setup.branch2.run_cleanup().await.unwrap();
        // Branch3 holds references from branch2:
        // - 1 manifest file
        // - 1 data files
        // - 1 deletion file
        // Branch3 holds reference from branch1:
        // - 1 manifest file
        // - 1 data files
        // - 2 deletion files
        // - 4 index files
        assert_eq!(setup.branch2.counts.num_manifest_files, 2);
        assert_eq!(setup.branch2.counts.num_data_files, 2);
        assert_eq!(setup.branch2.counts.num_tx_files, 1);
        assert_eq!(setup.branch2.counts.num_delete_files, 1);
        assert_eq!(setup.branch2.counts.num_index_files, 14);
        setup.branch1.run_cleanup().await.unwrap();
        // Cleanup 4 index files referenced from branch2
        assert_eq!(setup.branch1.counts.num_manifest_files, 2);
        assert_eq!(setup.branch1.counts.num_data_files, 2);
        assert_eq!(setup.branch1.counts.num_tx_files, 1);
        assert_eq!(setup.branch1.counts.num_delete_files, 1);
        assert_eq!(setup.branch1.counts.num_index_files, 7);

        setup.main.run_cleanup().await.unwrap();
        // Branch3 holds references from main:
        // - 1 manifest file
        // - 1 data files
        // - 1 deletion file
        // Branch4 holds references from main:
        // - 1 manifest file
        // - 3 data files
        // - 2 deletion files
        // - 4 index files
        assert_eq!(setup.main.counts.num_manifest_files, 3);
        assert_eq!(setup.main.counts.num_data_files, 4);
        assert_eq!(setup.main.counts.num_tx_files, 1);
        assert_eq!(setup.main.counts.num_delete_files, 2);
        assert_eq!(setup.main.counts.num_index_files, 14);

        setup.branch3.write_data().await.unwrap();
        setup.branch3.compact().await.unwrap();
        setup.branch3.run_cleanup().await.unwrap();
        // Only the counts for the latest version
        assert_eq!(setup.branch3.counts.num_manifest_files, 1);
        assert_eq!(setup.branch3.counts.num_data_files, 1);
        assert_eq!(setup.branch3.counts.num_tx_files, 1);
        assert_eq!(setup.branch3.counts.num_delete_files, 0);
        assert_eq!(setup.branch3.counts.num_index_files, 7);

        setup.main.run_cleanup().await.unwrap();
        // Cleanup doesn't take effects if we don't clean branch2 and branch1 first
        assert_eq!(setup.main.counts.num_manifest_files, 3);
        assert_eq!(setup.main.counts.num_data_files, 4);
        assert_eq!(setup.main.counts.num_tx_files, 1);
        assert_eq!(setup.main.counts.num_delete_files, 2);
        assert_eq!(setup.main.counts.num_index_files, 14);

        // Cleanup doesn't take effect if we don't clean branch2 first
        setup.branch1.run_cleanup().await.unwrap();
        assert_eq!(setup.branch1.counts.num_manifest_files, 2);
        assert_eq!(setup.branch1.counts.num_data_files, 2);
        assert_eq!(setup.branch1.counts.num_tx_files, 1);
        assert_eq!(setup.branch1.counts.num_delete_files, 1);
        assert_eq!(setup.branch1.counts.num_index_files, 7);

        setup.branch2.run_cleanup().await.unwrap();
        // Only the latest manifest is retained.
        // (1, 1, 1, 0, 4) is the counts for the latest version
        assert_eq!(setup.branch2.counts.num_manifest_files, 1);
        assert_eq!(setup.branch2.counts.num_data_files, 1);
        assert_eq!(setup.branch2.counts.num_tx_files, 1);
        assert_eq!(setup.branch2.counts.num_delete_files, 0);
        assert_eq!(setup.branch2.counts.num_index_files, 7);

        setup.branch1.run_cleanup().await.unwrap();
        // Only the latest manifest is retained.
        // (1, 1, 1, 0, 4) is the counts for the latest version
        assert_eq!(setup.branch1.counts.num_manifest_files, 1);
        assert_eq!(setup.branch1.counts.num_data_files, 1);
        assert_eq!(setup.branch1.counts.num_tx_files, 1);
        assert_eq!(setup.branch1.counts.num_delete_files, 0);
        assert_eq!(setup.branch1.counts.num_index_files, 7);

        setup.main.run_cleanup().await.unwrap();
        // Branch4 holds references from main:
        // - 1 manifest file
        // - 3 data files
        // - 2 deletion files
        // - 4 index files
        assert_eq!(setup.main.counts.num_manifest_files, 2);
        assert_eq!(setup.main.counts.num_data_files, 4);
        assert_eq!(setup.main.counts.num_tx_files, 1);
        assert_eq!(setup.main.counts.num_delete_files, 2);
        assert_eq!(setup.main.counts.num_index_files, 14);

        setup.branch4.write_data().await.unwrap();
        setup.branch4.compact().await.unwrap();
        setup.branch4.run_cleanup().await.unwrap();
        // Only the latest manifest is retained.
        // (1, 1, 1, 0, 4) is the counts for the latest version
        assert_eq!(setup.branch4.counts.num_manifest_files, 1);
        assert_eq!(setup.branch4.counts.num_data_files, 1);
        assert_eq!(setup.branch4.counts.num_tx_files, 1);
        assert_eq!(setup.branch4.counts.num_delete_files, 0);
        assert_eq!(setup.branch4.counts.num_index_files, 7);

        setup.main.run_cleanup().await.unwrap();
        // Only the latest manifest is retained.
        // (1, 1, 1, 0, 4) is the counts for the latest version
        assert_eq!(setup.main.counts.num_manifest_files, 1);
        assert_eq!(setup.main.counts.num_data_files, 1);
        assert_eq!(setup.main.counts.num_tx_files, 1);
        assert_eq!(setup.main.counts.num_delete_files, 0);
        assert_eq!(setup.main.counts.num_index_files, 7);
    }

    #[tokio::test]
    async fn auto_clean_referenced_branches_from_branch2() {
        // Setup shared lineage and per-branch auto-clean config
        let mut setup = build_lineage_datasets().await.unwrap();

        setup.branch3.write_data().await.unwrap();
        setup.enable_auto_cleanup().await.unwrap();
        setup
            .branch2
            .run_cleanup_with_referenced_branches()
            .await
            .unwrap();
        setup.branch3.refresh().await.unwrap();
        // Branch3 holds references from branch2:
        // - 1 manifest file
        // - 1 data file
        // - 1 deletion file
        assert_eq!(setup.branch2.counts.num_manifest_files, 2);
        assert_eq!(setup.branch2.counts.num_data_files, 1);
        assert_eq!(setup.branch2.counts.num_tx_files, 1);
        assert_eq!(setup.branch2.counts.num_delete_files, 1);
        assert_eq!(setup.branch2.counts.num_index_files, 7);
        // After auto-clean: branch3
        // 2 appends produced 2 data files
        // 2 deletes produced 2 deletion files
        assert_eq!(setup.branch3.counts.num_manifest_files, 1);
        assert_eq!(setup.branch3.counts.num_data_files, 2);
        assert_eq!(setup.branch3.counts.num_tx_files, 1);
        assert_eq!(setup.branch3.counts.num_delete_files, 2);
        assert_eq!(setup.branch3.counts.num_index_files, 7);
        setup
            .assert_unchanged(&["branch1", "branch4", "main"])
            .await;

        setup.disable_auto_cleanup().await.unwrap();
        setup.branch2.write_data().await.unwrap();
        setup.branch2.compact().await.unwrap();
        setup.branch3.compact().await.unwrap();
        setup.enable_auto_cleanup().await.unwrap();
        setup
            .branch2
            .run_cleanup_with_referenced_branches()
            .await
            .unwrap();
        setup.branch3.refresh().await.unwrap();
        // Only the latest manifest is retained.
        // (1, 1, 1, 0, 4) is the counts of one version
        assert_eq!(setup.branch2.counts.num_manifest_files, 1);
        assert_eq!(setup.branch2.counts.num_data_files, 1);
        assert_eq!(setup.branch2.counts.num_tx_files, 1);
        assert_eq!(setup.branch2.counts.num_delete_files, 0);
        assert_eq!(setup.branch2.counts.num_index_files, 7);
        // Only the latest manifest is retained.
        // (1, 1, 1, 0, 4) is the counts of one version
        assert_eq!(setup.branch3.counts.num_manifest_files, 1);
        assert_eq!(setup.branch3.counts.num_data_files, 1);
        assert_eq!(setup.branch3.counts.num_tx_files, 1);
        assert_eq!(setup.branch3.counts.num_delete_files, 0);
        assert_eq!(setup.branch3.counts.num_index_files, 7);
        setup
            .assert_unchanged(&["branch1", "branch4", "main"])
            .await;
    }

    #[tokio::test]
    async fn auto_clean_referenced_branches_from_main() {
        let mut setup = build_lineage_datasets().await.unwrap();

        setup.enable_auto_cleanup().await.unwrap();
        setup.main.write_data().await.unwrap();
        setup
            .main
            .run_cleanup_with_referenced_branches()
            .await
            .unwrap();
        // Branch3, branch2 and branch1 hold references from main:
        // - 1 manifest file
        // - 2 data files
        // - 1 deletion file
        // Branch4 holds references from main:
        // - 1 manifest file
        // - 3 data files
        // - 1 deletion file
        // - 4 index files
        assert_eq!(setup.main.counts.num_manifest_files, 3);
        assert_eq!(setup.main.counts.num_data_files, 4);
        assert_eq!(setup.main.counts.num_tx_files, 1);
        assert_eq!(setup.main.counts.num_delete_files, 3);
        assert_eq!(setup.main.counts.num_index_files, 7);

        setup.main.compact().await.unwrap();
        setup
            .main
            .run_cleanup_with_referenced_branches()
            .await
            .unwrap();
        // Branch3, branch2 and branch1 hold references from main:
        // - 1 manifest file
        // - 2 data files
        // - 1 deletion file
        // Branch4 holds references from main:
        // - 1 manifest file
        // - 3 data files
        // - 1 deletion file
        assert_eq!(setup.main.counts.num_manifest_files, 3);
        assert_eq!(setup.main.counts.num_data_files, 4);
        assert_eq!(setup.main.counts.num_tx_files, 1);
        assert_eq!(setup.main.counts.num_delete_files, 2);
        assert_eq!(setup.main.counts.num_index_files, 7);

        setup.branch4.compact().await.unwrap();
        setup
            .main
            .run_cleanup_with_referenced_branches()
            .await
            .unwrap();
        setup.branch4.refresh().await.unwrap();
        // Branch3, branch2 and branch1 hold references from main:
        // - 1 manifest file
        // - 2 data files
        // - 1 deletion file
        assert_eq!(setup.main.counts.num_manifest_files, 2);
        assert_eq!(setup.main.counts.num_data_files, 3);
        assert_eq!(setup.main.counts.num_tx_files, 1);
        assert_eq!(setup.main.counts.num_delete_files, 1);
        assert_eq!(setup.main.counts.num_index_files, 7);
        // (1, 1, 1, 0, 4) is the counts of one version
        assert_eq!(setup.branch4.counts.num_manifest_files, 1);
        assert_eq!(setup.branch4.counts.num_data_files, 1);
        assert_eq!(setup.branch4.counts.num_tx_files, 1);
        assert_eq!(setup.branch4.counts.num_delete_files, 0);
        assert_eq!(setup.branch4.counts.num_index_files, 7);

        setup.branch1.write_data().await.unwrap();
        setup.branch1.compact().await.unwrap();
        setup
            .main
            .run_cleanup_with_referenced_branches()
            .await
            .unwrap();
        setup.branch1.refresh().await.unwrap();
        // Branch3 and branch2 still hold references from main:
        // - 1 manifest file
        // - 2 data files
        // - 1 deletion file
        assert_eq!(setup.main.counts.num_manifest_files, 2);
        assert_eq!(setup.main.counts.num_data_files, 3);
        assert_eq!(setup.main.counts.num_tx_files, 1);
        assert_eq!(setup.main.counts.num_delete_files, 1);
        assert_eq!(setup.main.counts.num_index_files, 7);
        // Branch3 and branch2 still hold references from branch1:
        // - 1 manifest file
        // - 1 data files
        // - 1 deletion file
        assert_eq!(setup.branch1.counts.num_manifest_files, 2);
        assert_eq!(setup.branch1.counts.num_data_files, 2);
        assert_eq!(setup.branch1.counts.num_tx_files, 1);
        assert_eq!(setup.branch1.counts.num_delete_files, 1);
        assert_eq!(setup.branch1.counts.num_index_files, 7);

        setup.branch2.write_data().await.unwrap();
        setup.branch2.compact().await.unwrap();
        setup
            .main
            .run_cleanup_with_referenced_branches()
            .await
            .unwrap();
        setup.branch2.refresh().await.unwrap();
        // Branch3 still holds references from main:
        // - 1 manifest file
        // - 2 data files
        // - 1 deletion file
        assert_eq!(setup.main.counts.num_manifest_files, 2);
        assert_eq!(setup.main.counts.num_data_files, 3);
        assert_eq!(setup.main.counts.num_tx_files, 1);
        assert_eq!(setup.main.counts.num_delete_files, 1);
        assert_eq!(setup.main.counts.num_index_files, 7);
        // Branch3 still holds references from branch1:
        // - 1 manifest file
        // - 1 data files
        // - 1 deletion file
        assert_eq!(setup.branch1.counts.num_manifest_files, 2);
        assert_eq!(setup.branch1.counts.num_data_files, 2);
        assert_eq!(setup.branch1.counts.num_tx_files, 1);
        assert_eq!(setup.branch1.counts.num_delete_files, 1);
        assert_eq!(setup.branch1.counts.num_index_files, 7);
        // Branch3 still holds references from branch2:
        // - 1 manifest file
        // - 1 data files
        // - 1 deletion file
        assert_eq!(setup.branch2.counts.num_manifest_files, 2);
        assert_eq!(setup.branch2.counts.num_data_files, 2);
        assert_eq!(setup.branch2.counts.num_tx_files, 1);
        assert_eq!(setup.branch2.counts.num_delete_files, 1);
        assert_eq!(setup.branch2.counts.num_index_files, 7);

        setup.branch3.write_data().await.unwrap();
        setup.branch3.compact().await.unwrap();
        setup
            .main
            .run_cleanup_with_referenced_branches()
            .await
            .unwrap();
        setup.branch1.refresh().await.unwrap();
        setup.branch2.refresh().await.unwrap();
        setup.branch3.refresh().await.unwrap();
        // For all branches, only the latest manifest is retained.
        // (1, 1, 1, 0, 4) is the counts of one version
        assert_eq!(setup.main.counts.num_manifest_files, 1);
        assert_eq!(setup.main.counts.num_data_files, 1);
        assert_eq!(setup.main.counts.num_tx_files, 1);
        assert_eq!(setup.main.counts.num_delete_files, 0);
        assert_eq!(setup.main.counts.num_index_files, 7);
        assert_eq!(setup.branch1.counts.num_manifest_files, 1);
        assert_eq!(setup.branch1.counts.num_data_files, 1);
        assert_eq!(setup.branch1.counts.num_tx_files, 1);
        assert_eq!(setup.branch1.counts.num_delete_files, 0);
        assert_eq!(setup.branch1.counts.num_index_files, 7);
        assert_eq!(setup.branch2.counts.num_manifest_files, 1);
        assert_eq!(setup.branch2.counts.num_data_files, 1);
        assert_eq!(setup.branch2.counts.num_tx_files, 1);
        assert_eq!(setup.branch2.counts.num_delete_files, 0);
        assert_eq!(setup.branch2.counts.num_index_files, 7);
        assert_eq!(setup.branch3.counts.num_manifest_files, 1);
        assert_eq!(setup.branch3.counts.num_data_files, 1);
        assert_eq!(setup.branch3.counts.num_tx_files, 1);
        assert_eq!(setup.branch3.counts.num_delete_files, 0);
        assert_eq!(setup.branch3.counts.num_index_files, 7);
        setup.assert_unchanged(&["branch4"]).await;
    }

    #[tokio::test]
    async fn explain_cleanup_with_referenced_branches_matches_cleanup() {
        let mut setup = build_lineage_datasets().await.unwrap();

        setup.enable_auto_cleanup().await.unwrap();
        setup.main.write_data().await.unwrap();
        setup.main.compact().await.unwrap();
        setup.branch4.compact().await.unwrap();
        setup.branch1.write_data().await.unwrap();
        setup.branch1.compact().await.unwrap();
        setup.branch2.write_data().await.unwrap();
        setup.branch2.compact().await.unwrap();
        setup.branch3.write_data().await.unwrap();
        setup.branch3.compact().await.unwrap();

        setup.main.refresh().await.unwrap();
        setup.branch1.refresh().await.unwrap();
        setup.branch2.refresh().await.unwrap();
        setup.branch3.refresh().await.unwrap();
        setup.branch4.refresh().await.unwrap();
        let main_counts_before = setup.main.counts;
        let branch1_counts_before = setup.branch1.counts;
        let branch2_counts_before = setup.branch2.counts;
        let branch3_counts_before = setup.branch3.counts;
        let branch4_counts_before = setup.branch4.counts;

        let explanation = setup
            .main
            .explain_cleanup_with_referenced_branches()
            .await
            .unwrap();

        setup.main.refresh().await.unwrap();
        setup.branch1.refresh().await.unwrap();
        setup.branch2.refresh().await.unwrap();
        setup.branch3.refresh().await.unwrap();
        setup.branch4.refresh().await.unwrap();
        assert_eq!(setup.main.counts, main_counts_before);
        assert_eq!(setup.branch1.counts, branch1_counts_before);
        assert_eq!(setup.branch2.counts, branch2_counts_before);
        assert_eq!(setup.branch3.counts, branch3_counts_before);
        assert_eq!(setup.branch4.counts, branch4_counts_before);

        let removed = setup
            .main
            .run_cleanup_with_referenced_branches()
            .await
            .unwrap();

        assert!(!explanation.referenced_branches.is_empty());
        assert!(
            explanation
                .referenced_branches
                .iter()
                .any(|branch| branch.cleanup_candidate)
        );
        assert_eq!(explanation.stats, removed);
        setup.branch1.refresh().await.unwrap();
        setup.branch2.refresh().await.unwrap();
        setup.branch3.refresh().await.unwrap();
        setup.branch4.refresh().await.unwrap();
        assert_eq!(setup.main.counts.num_manifest_files, 1);
        assert_eq!(setup.branch1.counts.num_manifest_files, 1);
        assert_eq!(setup.branch2.counts.num_manifest_files, 1);
        assert_eq!(setup.branch3.counts.num_manifest_files, 1);
        assert_eq!(setup.branch4.counts.num_manifest_files, 1);
    }

    #[tokio::test]
    async fn auto_clean_referenced_branches_with_tags() {
        let mut setup = build_lineage_datasets().await.unwrap();

        setup
            .branch3
            .dataset
            .tags()
            .create("branch3-tag", setup.branch3.dataset.version().version)
            .await
            .unwrap();
        setup
            .main
            .dataset
            .tags()
            .create("main-tag", setup.main.dataset.version().version)
            .await
            .unwrap();

        setup.branch1.compact().await.unwrap();
        setup.branch2.compact().await.unwrap();
        setup.branch3.compact().await.unwrap();
        setup.branch4.compact().await.unwrap();
        setup.main.compact().await.unwrap();
        setup.enable_auto_cleanup().await.unwrap();
        setup
            .main
            .run_cleanup_with_referenced_branches()
            .await
            .unwrap();
        setup.branch1.refresh().await.unwrap();
        setup.branch2.refresh().await.unwrap();
        setup.branch3.refresh().await.unwrap();
        setup.branch4.refresh().await.unwrap();
        // Two tags hold two manifest references
        // Main tag holds 1 tx file, 3 data files, 2 deletion files and 4 index files
        assert_eq!(setup.main.counts.num_manifest_files, 3);
        assert_eq!(setup.main.counts.num_data_files, 4);
        assert_eq!(setup.main.counts.num_tx_files, 2);
        assert_eq!(setup.main.counts.num_delete_files, 2);
        assert_eq!(setup.main.counts.num_index_files, 14);
        // Branch3 tag holds branch1 with 1 tx file, 1 data files, 1 deletion files and 4 index files
        assert_eq!(setup.branch2.counts.num_manifest_files, 2);
        assert_eq!(setup.branch2.counts.num_data_files, 2);
        assert_eq!(setup.branch2.counts.num_tx_files, 1);
        assert_eq!(setup.branch2.counts.num_delete_files, 1);
        assert_eq!(setup.branch2.counts.num_index_files, 7);
        // Branch3 tag holds branch2 with 1 tx file, 1 data files, 1 deletion files and 4 index files
        assert_eq!(setup.branch2.counts.num_manifest_files, 2);
        assert_eq!(setup.branch2.counts.num_data_files, 2);
        assert_eq!(setup.branch2.counts.num_tx_files, 1);
        assert_eq!(setup.branch2.counts.num_delete_files, 1);
        assert_eq!(setup.branch2.counts.num_index_files, 7);
        assert_eq!(setup.branch4.counts.num_manifest_files, 1);
        assert_eq!(setup.branch4.counts.num_data_files, 1);
        assert_eq!(setup.branch4.counts.num_tx_files, 1);
        assert_eq!(setup.branch4.counts.num_delete_files, 0);
        assert_eq!(setup.branch4.counts.num_index_files, 7);

        setup
            .branch3
            .dataset
            .tags()
            .delete("branch3-tag")
            .await
            .unwrap();
        setup
            .main
            .run_cleanup_with_referenced_branches()
            .await
            .unwrap();
        setup.branch1.refresh().await.unwrap();
        setup.branch2.refresh().await.unwrap();
        setup.branch3.refresh().await.unwrap();
        setup.branch4.refresh().await.unwrap();
        // 1 manifest file referenced by branch3-tag is cleaned
        assert_eq!(setup.main.counts.num_manifest_files, 2);
        assert_eq!(setup.main.counts.num_data_files, 4);
        assert_eq!(setup.main.counts.num_tx_files, 2);
        assert_eq!(setup.main.counts.num_delete_files, 2);
        assert_eq!(setup.main.counts.num_index_files, 14);
        assert_eq!(setup.branch1.counts.num_manifest_files, 1);
        assert_eq!(setup.branch1.counts.num_data_files, 1);
        assert_eq!(setup.branch1.counts.num_tx_files, 1);
        assert_eq!(setup.branch1.counts.num_delete_files, 0);
        assert_eq!(setup.branch1.counts.num_index_files, 7);
        assert_eq!(setup.branch2.counts.num_manifest_files, 1);
        assert_eq!(setup.branch2.counts.num_data_files, 1);
        assert_eq!(setup.branch2.counts.num_tx_files, 1);
        assert_eq!(setup.branch2.counts.num_delete_files, 0);
        assert_eq!(setup.branch2.counts.num_index_files, 7);
        assert_eq!(setup.branch3.counts.num_manifest_files, 1);
        assert_eq!(setup.branch3.counts.num_data_files, 1);
        assert_eq!(setup.branch3.counts.num_tx_files, 1);
        assert_eq!(setup.branch3.counts.num_delete_files, 0);
        assert_eq!(setup.branch3.counts.num_index_files, 7);
        assert_eq!(setup.branch4.counts.num_manifest_files, 1);
        assert_eq!(setup.branch4.counts.num_data_files, 1);
        assert_eq!(setup.branch4.counts.num_tx_files, 1);
        assert_eq!(setup.branch4.counts.num_delete_files, 0);
        assert_eq!(setup.branch4.counts.num_index_files, 7);

        setup.main.dataset.tags().delete("main-tag").await.unwrap();
        setup
            .main
            .run_cleanup_with_referenced_branches()
            .await
            .unwrap();
        setup.branch2.refresh().await.unwrap();
        setup.branch3.refresh().await.unwrap();
        setup.branch4.refresh().await.unwrap();
        // All cleaned up
        assert_eq!(setup.main.counts.num_manifest_files, 1);
        assert_eq!(setup.main.counts.num_data_files, 1);
        assert_eq!(setup.main.counts.num_tx_files, 1);
        assert_eq!(setup.main.counts.num_delete_files, 0);
        assert_eq!(setup.main.counts.num_index_files, 7);
        assert_eq!(setup.branch2.counts.num_manifest_files, 1);
        assert_eq!(setup.branch2.counts.num_data_files, 1);
        assert_eq!(setup.branch2.counts.num_tx_files, 1);
        assert_eq!(setup.branch2.counts.num_delete_files, 0);
        assert_eq!(setup.branch2.counts.num_index_files, 7);
        assert_eq!(setup.branch3.counts.num_manifest_files, 1);
        assert_eq!(setup.branch3.counts.num_data_files, 1);
        assert_eq!(setup.branch3.counts.num_tx_files, 1);
        assert_eq!(setup.branch3.counts.num_delete_files, 0);
        assert_eq!(setup.branch3.counts.num_index_files, 7);
        assert_eq!(setup.branch4.counts.num_manifest_files, 1);
        assert_eq!(setup.branch4.counts.num_data_files, 1);
        assert_eq!(setup.branch4.counts.num_tx_files, 1);
        assert_eq!(setup.branch4.counts.num_delete_files, 0);
        assert_eq!(setup.branch4.counts.num_index_files, 7);
    }

    #[rstest]
    #[case::plain(false)]
    #[case::wrapped(true)]
    #[tokio::test]
    async fn expired_manifest_size_tolerates_a_vanished_manifest(#[case] is_wrapped: bool) {
        let wrapper = Arc::new(MockObjectStore::new());
        let (store, _) = ObjectStore::from_uri_and_params(
            Arc::new(ObjectStoreRegistry::default()),
            "memory://",
            &ObjectStoreParams {
                object_store_wrapper: Some(wrapper.clone()),
                ..Default::default()
            },
        )
        .await
        .unwrap();
        let path = Path::from("_versions/1.manifest");
        if is_wrapped {
            wrapper.policy.lock().unwrap().set_before_policy(
                "missing_manifest",
                Arc::new(|_, path| Err(Error::not_found(path.to_string()))),
            );
        }

        // A size the listing already reported costs no request at all.
        assert_eq!(
            expired_manifest_size(&store, &path, Some(42))
                .await
                .unwrap(),
            42
        );

        // Unknown size and the object is gone: zero, not an error. A concurrent cleanup
        // can remove a manifest between listing and here, and propagating NotFound would
        // abandon the whole sweep over one file already in the state we wanted.
        assert_eq!(expired_manifest_size(&store, &path, None).await.unwrap(), 0);

        // A present object still reports its real size through the fallback.
        wrapper
            .policy
            .lock()
            .unwrap()
            .clear_before_policy("missing_manifest");
        store.put(&path, b"1234".as_slice()).await.unwrap();
        assert_eq!(expired_manifest_size(&store, &path, None).await.unwrap(), 4);
    }

    #[test]
    fn test_calculate_duration() {
        // One permit is one delete request, so the interval is the reciprocal of the
        // configured rate. Scaling by a bulk-delete batch size here would let the
        // limiter issue batch_size times the rate the caller asked for: at 100
        // requests/s an S3 multiplier of 1,000 would give 10us instead of 10ms.
        assert_eq!(calculate_duration(100), Duration::from_millis(10));
        assert_eq!(calculate_duration(1_000), Duration::from_millis(1));
        assert_eq!(calculate_duration(1), Duration::from_secs(1));

        // Edge case: rate too small is clamped to 1.
        assert_eq!(calculate_duration(0), calculate_duration(1));

        // Edge case: a rate finer than 1ns is clamped to 1ns.
        assert_eq!(calculate_duration(2_000_000_000), Duration::from_nanos(1));
    }

    #[rstest]
    #[case::single(None, Some(1), false, None, false)]
    #[case::single_unlimited(None, None, false, None, false)]
    #[case::cascade_missing(Some(1), None, true, None, false)]
    #[case::cascade_different(Some(100), Some(1), true, Some("100"), false)]
    #[case::cascade_unlimited(None, None, true, Some("1"), false)]
    #[case::cascade_invalid(Some(1), None, true, Some("invalid"), false)]
    #[case::cascade_zero(Some(1), None, true, Some("0"), false)]
    #[case::cascade_unlimited_invalid(None, None, true, Some("invalid"), false)]
    #[case::standalone(Some(1), None, false, None, true)]
    #[case::standalone_unlimited(None, None, false, None, true)]
    #[tokio::test(start_paused = true)]
    async fn test_cleanup_with_rate_limit(
        #[case] rate: Option<u64>,
        #[case] operation_rate: Option<u64>,
        #[case] clean_referenced_branches: bool,
        #[case] child_rate: Option<&str>,
        #[case] standalone: bool,
    ) {
        let data = || {
            lance_datagen::gen_batch()
                .col(
                    "id",
                    lance_datagen::array::step::<arrow_array::types::Int32Type>(),
                )
                .into_reader_rows(2.into(), 1.into())
        };
        let concurrency = 2;
        let delete_latency = if clean_referenced_branches && rate.is_none() {
            Duration::from_secs(1)
        } else {
            Duration::ZERO
        };
        let mock_store = Arc::new(MockObjectStore {
            delete_latency,
            ..MockObjectStore::new()
        });
        let mut parent = Dataset::write(
            data(),
            "memory://cleanup",
            Some(WriteParams {
                store_params: Some(ObjectStoreParams {
                    object_store_wrapper: Some(mock_store.clone()),
                    ..Default::default()
                }),
                ..Default::default()
            }),
        )
        .await
        .unwrap();
        let mut children = Vec::new();
        // Cover sequential siblings and independently scheduled branch chains.
        let branches: &[&str] = if clean_referenced_branches {
            &["first", "sibling", "second", "disabled"]
        } else {
            &[]
        };
        for &name in branches {
            let mut child = parent
                .create_branch(name, (None, None), None)
                .await
                .unwrap();
            let mut updates = vec![("lance.auto_cleanup.retain_versions", "1")];
            if name != "disabled" {
                updates.push(("lance.auto_cleanup.interval", "1"));
            }
            updates.push(("lance.auto_cleanup.delete_concurrency", "invalid"));
            if let Some(value) = child_rate {
                updates.push(("lance.auto_cleanup.delete_rate_limit", value));
            }
            child.update_config(updates).await.unwrap();
            for _ in 0..2 {
                child = Dataset::write(
                    data(),
                    Arc::new(child),
                    Some(WriteParams {
                        mode: WriteMode::Overwrite,
                        max_rows_per_file: 1,
                        skip_auto_cleanup: true,
                        ..Default::default()
                    }),
                )
                .await
                .unwrap();
            }
            assert!(child.get_fragments().len() > 1);
            let versions = child.versions().await.unwrap().len();
            let expected = child.scan().try_into_batch().await.unwrap();
            children.push((child, versions, expected));
            if name != "first" {
                parent.append(data(), None).await.unwrap();
            }
        }
        // These versions are not branch sources, so the parent also has deletions.
        for _ in 0..2 {
            parent.append(data(), None).await.unwrap();
        }
        let mut policy = CleanupPolicyBuilder::default()
            .clean_referenced_branches(clean_referenced_branches)
            .retain_n_versions(&parent, 1)
            .await
            .unwrap()
            .build();
        policy.delete_rate_limit = rate;
        let requests = Arc::new(Mutex::new(Vec::new()));
        let recorded = requests.clone();
        mock_store.policy.lock().unwrap().set_before_policy(
            "record_deletes",
            Arc::new(move |operation, path| {
                if operation == "delete" {
                    recorded
                        .lock()
                        .unwrap()
                        .push((path.clone(), tokio::time::Instant::now()));
                }
                Ok(())
            }),
        );
        let mut operation = parent
            .cleanup(policy)
            .with_delete_concurrency(concurrency)
            .unwrap();
        if let Some(rate) = operation_rate {
            operation = operation.with_delete_rate_limit(rate).unwrap();
        }
        let start = tokio::time::Instant::now();
        if matches!(child_rate, Some("invalid" | "0")) {
            let child = parent.checkout_version(("first", None)).await.unwrap();
            let error = build_cleanup_policy(&child, child.manifest.as_ref())
                .await
                .unwrap_err();
            assert!(matches!(error, Error::Cleanup { .. }));
            assert_contains!(error.to_string(), "delete_rate_limit");
        }
        let explanation = operation.explain().await.unwrap();
        assert_eq!(start.elapsed(), Duration::ZERO);
        assert!(requests.lock().unwrap().is_empty());
        let stats = if standalone {
            let mut manifest = parent.manifest.as_ref().clone();
            manifest
                .config
                .insert("lance.auto_cleanup.interval".into(), "1".into());
            manifest
                .config
                .insert("lance.auto_cleanup.retain_versions".into(), "1".into());
            if let Some(rate) = rate {
                manifest.config.insert(
                    "lance.auto_cleanup.delete_rate_limit".into(),
                    rate.to_string(),
                );
            }
            cleanup_cascade_branch(&parent, &manifest)
                .await
                .unwrap()
                .unwrap()
        } else {
            operation.execute().await.unwrap()
        };
        assert_eq!(stats, explanation.stats);
        for (child, previous_versions, expected) in children {
            let deleted = requests
                .lock()
                .unwrap()
                .iter()
                .any(|(path, _)| path.prefix_matches(&child.base));
            let enabled = child
                .manifest
                .config
                .contains_key("lance.auto_cleanup.interval");
            assert_eq!(deleted, enabled);
            assert_eq!(
                child.versions().await.unwrap().len(),
                if enabled { 1 } else { previous_versions }
            );
            assert_eq!(child.scan().try_into_batch().await.unwrap(), expected);
        }
        let requests = requests.lock().unwrap();
        assert!(requests.len() > 2);
        assert!(
            requests
                .iter()
                .any(|(path, _)| path.prefix_matches(&parent.versions_dir()))
        );
        if let Some(rate) = operation_rate.or(rate) {
            let interval = calculate_duration(rate);
            for pair in requests.windows(2) {
                let elapsed = pair[1].1.duration_since(pair[0].1);
                assert!(elapsed >= interval);
                assert!(elapsed < interval + Duration::from_millis(50));
            }
        } else if !delete_latency.is_zero() {
            // The two active lineages must share the window, but still run in parallel.
            assert!(
                start.elapsed() >= delete_latency * requests.len().div_ceil(concurrency) as u32
            );
            assert!(start.elapsed() < delete_latency * requests.len() as u32);
        } else {
            assert_eq!(start.elapsed(), Duration::ZERO);
        }
    }

    use lance_table::io::commit::external_manifest::ExternalManifestStore;
    use lance_table::io::commit::{ManifestLocation, ManifestNamingScheme};

    /// `(path, size, identity)` per version.
    #[derive(Debug, Default)]
    struct IdentifiedStore {
        rows: Mutex<HashMap<u64, (String, u64, String)>>,
        next_identity: std::sync::atomic::AtomicU64,
    }

    #[async_trait::async_trait]
    impl ExternalManifestStore for IdentifiedStore {
        async fn get(&self, _base_uri: &str, version: u64) -> Result<String> {
            self.rows
                .lock()
                .unwrap()
                .get(&version)
                .map(|row| row.0.clone())
                .ok_or_else(|| Error::not_found(format!("@{version}")))
        }

        async fn get_manifest_location(
            &self,
            _base_uri: &str,
            version: u64,
        ) -> Result<ManifestLocation> {
            let row = self
                .rows
                .lock()
                .unwrap()
                .get(&version)
                .cloned()
                .ok_or_else(|| Error::not_found(format!("@{version}")))?;
            Ok(ManifestLocation {
                version,
                path: Path::parse(&row.0).unwrap(),
                size: Some(row.1),
                naming_scheme: ManifestNamingScheme::V2,
                e_tag: None,
                identity: Some(row.2),
            })
        }

        async fn get_latest_version(&self, _base_uri: &str) -> Result<Option<(u64, String)>> {
            Ok(self
                .rows
                .lock()
                .unwrap()
                .iter()
                .max_by_key(|(version, _)| **version)
                .map(|(version, row)| (*version, row.0.clone())))
        }

        async fn get_latest_manifest_location(
            &self,
            base_uri: &str,
        ) -> Result<Option<ManifestLocation>> {
            match self.get_latest_version(base_uri).await? {
                Some((version, _)) => self
                    .get_manifest_location(base_uri, version)
                    .await
                    .map(Some),
                None => Ok(None),
            }
        }

        async fn put_if_not_exists(
            &self,
            _base_uri: &str,
            version: u64,
            path: &str,
            size: u64,
            _e_tag: Option<String>,
        ) -> Result<()> {
            let identity = format!(
                "identity-{}",
                self.next_identity
                    .fetch_add(1, std::sync::atomic::Ordering::SeqCst)
            );
            let mut rows = self.rows.lock().unwrap();
            if rows.contains_key(&version) {
                return Err(Error::commit_conflict_source(version, "exists".into()));
            }
            rows.insert(version, (path.to_string(), size, identity));
            Ok(())
        }

        async fn put_if_exists(
            &self,
            _base_uri: &str,
            version: u64,
            path: &str,
            size: u64,
            _e_tag: Option<String>,
        ) -> Result<()> {
            let mut rows = self.rows.lock().unwrap();
            let row = rows
                .get_mut(&version)
                .ok_or_else(|| Error::not_found(format!("@{version}")))?;
            row.0 = path.to_string();
            row.1 = size;
            Ok(())
        }

        fn supports_predecessor_condition(&self) -> bool {
            true
        }

        async fn get_identity(&self, _base_uri: &str, version: u64) -> Result<Option<String>> {
            Ok(self
                .rows
                .lock()
                .unwrap()
                .get(&version)
                .map(|row| row.2.clone()))
        }

        async fn list_versions(
            &self,
            base_uri: &str,
            since: Option<u64>,
        ) -> Result<Option<Vec<ManifestLocation>>> {
            let versions: Vec<u64> = self.rows.lock().unwrap().keys().copied().collect();
            let mut locations = Vec::new();
            for version in versions {
                if since.is_none_or(|since| version > since) {
                    locations.push(self.get_manifest_location(base_uri, version).await?);
                }
            }
            Ok(Some(locations))
        }

        async fn forget_version(
            &self,
            _base_uri: &str,
            version: u64,
            identity: &str,
        ) -> Result<()> {
            let mut rows = self.rows.lock().unwrap();
            if rows.get(&version).is_some_and(|row| row.2 == identity) {
                rows.remove(&version);
            }
            Ok(())
        }
    }

    /// Cleanup retires the store record of every manifest it removes, one
    /// whose object was already gone included, so store-backed history
    /// matches what is on disk.
    #[tokio::test]
    async fn test_cleanup_forgets_removed_versions_in_the_external_store() {
        use crate::dataset::{InsertBuilder, WriteDestination};
        use lance_table::io::commit::CommitHandler;
        use lance_table::io::commit::external_manifest::{
            ExternalManifestCommitHandler, ExternalManifestStore,
        };

        let store = Arc::new(IdentifiedStore::default());
        let handler: Arc<dyn CommitHandler> = Arc::new(ExternalManifestCommitHandler {
            external_manifest_store: store.clone(),
        });
        let uri = TempStrDir::default();
        let batch = || arrow_array::record_batch!(("i", Int32, [1, 2, 3])).unwrap();
        let mut dataset = InsertBuilder::new(uri.as_str())
            .with_params(&WriteParams {
                commit_handler: Some(handler.clone()),
                ..Default::default()
            })
            .execute(vec![batch()])
            .await
            .unwrap();
        for _ in 0..2 {
            dataset = InsertBuilder::new(WriteDestination::Dataset(Arc::new(dataset)))
                .with_params(&WriteParams {
                    mode: WriteMode::Append,
                    commit_handler: Some(handler.clone()),
                    ..Default::default()
                })
                .execute(vec![batch()])
                .await
                .unwrap();
        }
        assert_eq!(dataset.count_versions().await.unwrap(), 3);

        // Version 1's object is already gone, as after a cleanup that stopped
        // before retiring records.
        let v1 = Path::parse(store.get("", 1).await.unwrap()).unwrap();
        dataset.object_store.delete(&v1).await.unwrap();

        cleanup_old_versions(
            &dataset,
            CleanupPolicyBuilder::default()
                .before_timestamp(chrono::Utc::now())
                .build(),
        )
        .await
        .unwrap();

        let mut remaining: Vec<u64> = store.rows.lock().unwrap().keys().copied().collect();
        remaining.sort();
        assert_eq!(remaining, vec![3]);
        assert_eq!(dataset.count_versions().await.unwrap(), 1);
        assert_eq!(dataset.versions().await.unwrap().len(), 1);
    }

    /// A manifest whose delete fails keeps its store record. A record outliving its
    /// manifest is retired by the next cleanup; a manifest outliving its record is a
    /// version nothing can find again.
    #[tokio::test]
    async fn test_a_manifest_that_fails_to_delete_keeps_its_record() {
        use crate::dataset::{InsertBuilder, WriteDestination};
        use lance_table::io::commit::CommitHandler;
        use lance_table::io::commit::external_manifest::ExternalManifestCommitHandler;

        let store = Arc::new(IdentifiedStore::default());
        let handler: Arc<dyn CommitHandler> = Arc::new(ExternalManifestCommitHandler {
            external_manifest_store: store.clone(),
        });
        let mock_store = Arc::new(MockObjectStore::new());
        let os_params = ObjectStoreParams {
            object_store_wrapper: Some(mock_store.clone()),
            ..Default::default()
        };
        let uri = TempStrDir::default();
        let batch = || arrow_array::record_batch!(("i", Int32, [1, 2, 3])).unwrap();
        let params = |mode| WriteParams {
            mode,
            commit_handler: Some(handler.clone()),
            store_params: Some(os_params.clone()),
            ..Default::default()
        };
        let mut dataset = InsertBuilder::new(uri.as_str())
            .with_params(&params(WriteMode::Create))
            .execute(vec![batch()])
            .await
            .unwrap();
        for _ in 0..2 {
            dataset = InsertBuilder::new(WriteDestination::Dataset(Arc::new(dataset)))
                .with_params(&params(WriteMode::Append))
                .execute(vec![batch()])
                .await
                .unwrap();
        }
        assert_eq!(dataset.count_versions().await.unwrap(), 3);

        mock_store.policy.lock().unwrap().set_before_policy(
            "block_delete_manifest",
            Arc::new(|op: &str, path: &Path| -> Result<()> {
                if op.contains("delete") && path.extension() == Some("manifest") {
                    Err(Error::internal("Delete manifest blocked".to_string()))
                } else {
                    Ok(())
                }
            }),
        );

        let stats = cleanup_old_versions(
            &dataset,
            CleanupPolicyBuilder::default()
                .before_timestamp(chrono::Utc::now())
                .build(),
        )
        .await
        .expect("a blocked manifest delete must not fail the sweep");

        assert!(
            stats.failed_deletes > 0,
            "the blocked manifests are counted"
        );
        assert_eq!(
            stats.old_versions, 0,
            "a manifest still on disk is not a removed version"
        );
        // Every record survives, because every manifest survives.
        let mut remaining: Vec<u64> = store.rows.lock().unwrap().keys().copied().collect();
        remaining.sort();
        assert_eq!(remaining, vec![1, 2, 3]);
    }

    /// D-rows: `_fri/` row-map garbage collection.
    mod frag_reuse_map_gc {
        use super::*;
        use crate::dataset::index::frag_reuse::cleanup_frag_reuse_index;
        use crate::dataset::write::CommitBuilder;
        use crate::index::frag_reuse_reader::tests as reader_tests;
        use arrow_array::types::Int32Type;
        use chrono::TimeDelta;
        use lance_index::frag_reuse::FRAG_REUSE_INDEX_NAME;
        use lance_index::scalar::ScalarIndexParams;
        use lance_table::format::Fragment;
        use lance_table::format::IndexMetadata;
        use lance_table::format::pb::fragment_reuse_index_details as pb_fri;
        use lance_table::transaction::RewriteGroup;
        use prost::Message;

        impl MockDatasetFixture {
            /// The `_fri/<map_id>/` directories currently on disk.
            async fn list_fri_map_dirs(&self) -> Vec<String> {
                let registry = Arc::new(ObjectStoreRegistry::default());
                let (os, path) = ObjectStore::from_uri_and_params(
                    registry,
                    &self.dataset_path,
                    &self.os_params(),
                )
                .await
                .unwrap();
                let mut ids = std::collections::HashSet::new();
                let mut stream = os.read_dir_all(&path.clone().join("_fri"), None);
                loop {
                    match stream.try_next().await {
                        Ok(Some(meta)) => {
                            let relative = remove_prefix(&meta.location, &path);
                            if let Some(map_id) = relative.parts().nth(1) {
                                ids.insert(map_id.as_ref().to_string());
                            }
                        }
                        Ok(None) => break,
                        Err(Error::NotFound { .. }) => break,
                        Err(e) => panic!("{e}"),
                    }
                }
                let mut ids: Vec<String> = ids.into_iter().collect();
                ids.sort();
                ids
            }
        }

        async fn reserve_fragments(dataset: &mut Dataset, num_fragments: u32) {
            dataset
                .apply_commit(
                    Transaction::new(
                        dataset.manifest.version,
                        Operation::ReserveFragments { num_fragments },
                        None,
                    ),
                    &Default::default(),
                    &Default::default(),
                )
                .await
                .unwrap();
        }

        /// A persisted dataset (mock store, recorded mtimes) with one
        /// committed stable-partition rewrite; returns the map id of its
        /// row map.
        async fn make_tagged(fixture: &MockDatasetFixture) -> (Dataset, String) {
            let data = lance_datagen::gen_batch()
                .col("i", lance_datagen::array::step::<Int32Type>())
                .into_reader_rows(
                    lance_datagen::RowCount::from(4),
                    lance_datagen::BatchCount::from(2),
                );
            Dataset::write(
                data,
                &fixture.dataset_path,
                Some(WriteParams {
                    store_params: Some(fixture.os_params()),
                    commit_handler: Some(Arc::new(RenameCommitHandler)),
                    mode: WriteMode::Create,
                    max_rows_per_file: 4,
                    ..Default::default()
                }),
            )
            .await
            .unwrap();
            let mut dataset = *fixture.open().await.unwrap();
            dataset
                .create_index(
                    &["i"],
                    IndexType::Scalar,
                    Some("i_idx".into()),
                    &ScalarIndexParams::default(),
                    false,
                )
                .await
                .unwrap();
            reserve_fragments(&mut dataset, 20).await;
            let old_fragments: Vec<Fragment> = dataset.fragments().iter().cloned().collect();
            let (transition, destinations) = reader_tests::prepare(&dataset).await;
            let Some(pb_fri::transition::Mapping::StablePartition(mapping)) = &transition.mapping
            else {
                unreachable!()
            };
            let map_id = mapping.map_id.clone();
            let read_version = dataset.manifest.version;
            let frag_reuse_index = Some(
                crate::index::frag_reuse::frag_reuse_entry_appending(&dataset, vec![transition])
                    .await
                    .unwrap(),
            );
            let dataset = CommitBuilder::new(Arc::new(dataset))
                .execute(Transaction::new(
                    read_version,
                    Operation::Rewrite {
                        groups: vec![RewriteGroup {
                            old_fragments,
                            new_fragments: destinations,
                        }],
                        rewritten_indices: vec![],
                        frag_reuse_index,
                    },
                    None,
                ))
                .await
                .unwrap();
            (dataset, map_id)
        }

        /// D1: a map referenced by a retained manifest survives cleanup, no
        /// matter how old it is.
        #[tokio::test]
        #[serial_test::serial(frag_reuse_maintenance)]
        async fn referenced_map_is_kept() {
            let fixture = MockDatasetFixture::try_new().unwrap();
            let (_, map_id) = make_tagged(&fixture).await;
            MockClock::set_system_time(TimeDelta::try_days(10).unwrap().to_std().unwrap());
            fixture
                .run_cleanup(utc_now() - TimeDelta::try_seconds(1).unwrap())
                .await
                .unwrap();
            assert_eq!(fixture.list_fri_map_dirs().await, vec![map_id]);
        }

        /// D2: a map referenced only by expiring manifests goes away in the
        /// same pass that removes them (the trim released it from the entry
        /// first).
        #[tokio::test]
        #[serial_test::serial(frag_reuse_maintenance)]
        async fn map_released_by_trim_is_collected_with_its_manifests() {
            let fixture = MockDatasetFixture::try_new().unwrap();
            let (mut dataset, map_id) = make_tagged(&fixture).await;
            assert_eq!(fixture.list_fri_map_dirs().await, vec![map_id]);

            // Drain the index onto the destinations, then trim: the entry is
            // deleted, so only old manifests reference the map.
            dataset
                .create_index(
                    &["i"],
                    IndexType::Scalar,
                    Some("i_idx".into()),
                    &ScalarIndexParams::default(),
                    true,
                )
                .await
                .unwrap();
            cleanup_frag_reuse_index(&mut dataset).await.unwrap();
            assert!(
                dataset
                    .load_index_by_name(FRAG_REUSE_INDEX_NAME)
                    .await
                    .unwrap()
                    .is_none()
            );

            // Young files referenced by expiring manifests still wait for the
            // unverified retention window unless verified; the reference from
            // the removed manifest is that verification.
            MockClock::set_system_time(TimeDelta::try_days(10).unwrap().to_std().unwrap());
            fixture
                .run_cleanup(utc_now() - TimeDelta::try_seconds(1).unwrap())
                .await
                .unwrap();
            assert!(fixture.list_fri_map_dirs().await.is_empty());
        }

        /// D3 + D4: an unreferenced map (a failed commit's orphan) is kept
        /// while young and collected once it ages past the unverified
        /// retention threshold.
        #[tokio::test]
        #[serial_test::serial(frag_reuse_maintenance)]
        async fn orphan_map_ages_out() {
            let fixture = MockDatasetFixture::try_new().unwrap();
            let (dataset, committed_map_id) = make_tagged(&fixture).await;

            // A row map written at day 10 whose commit never lands.
            MockClock::set_system_time(TimeDelta::try_days(10).unwrap().to_std().unwrap());
            let (orphan_transition, _) = reader_tests::prepare(&dataset).await;
            let Some(pb_fri::transition::Mapping::StablePartition(mapping)) =
                &orphan_transition.mapping
            else {
                unreachable!()
            };
            let orphan_map_id = mapping.map_id.clone();
            let mut expected = vec![committed_map_id.clone(), orphan_map_id.clone()];
            expected.sort();
            assert_eq!(fixture.list_fri_map_dirs().await, expected);

            // D3: unreferenced but young, kept.
            fixture
                .run_cleanup(utc_now() + TimeDelta::try_seconds(1).unwrap())
                .await
                .unwrap();
            assert_eq!(fixture.list_fri_map_dirs().await, expected);

            // D4: unreferenced and old, collected; the referenced map stays.
            MockClock::set_system_time(TimeDelta::try_days(20).unwrap().to_std().unwrap());
            fixture
                .run_cleanup(utc_now() + TimeDelta::try_seconds(1).unwrap())
                .await
                .unwrap();
            assert_eq!(fixture.list_fri_map_dirs().await, vec![committed_map_id]);
        }

        /// D5: reference resolution decodes external entry content too. The
        /// integration tests above cover inline entries; this drives the
        /// resolver directly over an entry whose details live in an external
        /// file.
        #[tokio::test]
        #[serial_test::serial(frag_reuse_maintenance)]
        async fn reference_resolution_reads_external_entries() {
            let fixture = MockDatasetFixture::try_new().unwrap();
            let (dataset, _) = make_tagged(&fixture).await;

            let map_id = Uuid::new_v4().to_string();
            let digest = |id: u64| pb_fri::FragmentDigest {
                id,
                physical_rows: 4,
                num_deleted_rows: 0,
            };
            let content = {
                let mut content = Vec::new();
                let transition = pb_fri::Transition {
                    sources: vec![digest(100)],
                    destinations: vec![digest(101)],
                    mapping: Some(pb_fri::transition::Mapping::StablePartition(
                        pb_fri::StablePartition {
                            map_id: map_id.clone(),
                            map_size_bytes: 1,
                            base_id: None,
                        },
                    )),
                };
                content.extend(reader_tests::field(2, &transition.encode_to_vec()));
                content
            };
            let uuid = Uuid::new_v4();
            let details_path = dataset
                .indices_dir()
                .join(uuid.to_string())
                .join("frag_reuse_details.binpb");
            let mut writer = dataset.object_store.create(&details_path).await.unwrap();
            use tokio::io::AsyncWriteExt;
            writer.write_all(&content).await.unwrap();
            writer.shutdown().await.unwrap();
            let entry = IndexMetadata {
                uuid,
                fields: vec![],
                covering_fields: vec![],
                name: FRAG_REUSE_INDEX_NAME.into(),
                dataset_version: dataset.manifest.version,
                fragment_bitmap: None,
                index_details: Some(Arc::new(prost_types::Any {
                    type_url: "/lance.table.FragmentReuseIndexDetails".into(),
                    value: reader_tests::field(
                        2,
                        &lance_table::format::pb::ExternalFile {
                            path: "frag_reuse_details.binpb".into(),
                            offset: 0,
                            size: content.len() as u64,
                        }
                        .encode_to_vec(),
                    ),
                })),
                index_version: 1,
                created_at: None,
                base_id: None,
                files: None,
            };

            let task = CleanupTask::new(
                &dataset.cleanup(CleanupPolicy::default()),
                CleanupAction::Execute,
            )
            .unwrap();
            let mut inspection = CleanupInspection::default();
            inspection
                .frag_reuse_entries
                .insert(uuid, (entry.clone(), true));
            assert!(task.resolve_frag_reuse_map_ids(&mut inspection).await);
            assert!(
                inspection
                    .referenced_files
                    .frag_reuse_map_ids
                    .contains(&map_id)
            );

            // An uninterpretable entry (a future index_version) disables
            // `_fri` collection for the whole run instead of guessing.
            let future_entry = IndexMetadata {
                index_version: 2,
                ..entry.clone()
            };
            let mut inspection = CleanupInspection::default();
            inspection
                .frag_reuse_entries
                .insert(uuid, (future_entry, true));
            assert!(!task.resolve_frag_reuse_map_ids(&mut inspection).await);

            // So does an unknown envelope-level record inside a supported
            // index_version: it may reference row maps this build cannot
            // see, so no map may be treated as unreferenced this run.
            let mut unknown_content = content.clone();
            unknown_content.extend(reader_tests::field(9, b"future envelope record"));
            let unknown_entry = IndexMetadata {
                index_details: Some(Arc::new(prost_types::Any {
                    type_url: "/lance.table.FragmentReuseIndexDetails".into(),
                    value: reader_tests::field(1, &unknown_content),
                })),
                ..entry
            };
            let mut inspection = CleanupInspection::default();
            inspection
                .frag_reuse_entries
                .insert(uuid, (unknown_entry, true));
            assert!(
                !task.resolve_frag_reuse_map_ids(&mut inspection).await,
                "an unknown envelope record must disable _fri collection"
            );
        }

        /// E: the full lifecycle, asserting every stage. A stable-partition
        /// commit installs the tagged entry and its row map; an index rebuild
        /// over the destinations simulates the drain and drops the old
        /// segment; the trim then deletes the fully drained entry; a young
        /// cleanup keeps the map (still referenced by retained manifests);
        /// an aged cleanup removes those manifests and the map with them,
        /// leaving a working table.
        #[tokio::test]
        #[serial_test::serial(frag_reuse_maintenance)]
        async fn end_to_end_lifecycle() {
            use crate::index::frag_reuse::decode_frag_reuse_ledger;
            use lance_table::system_index::frag_reuse::ledger::Mapping;

            let fixture = MockDatasetFixture::try_new().unwrap();
            let (mut dataset, map_id) = make_tagged(&fixture).await;

            // Stage 1: the tagged entry holds one stable-partition
            // transition referencing the on-disk row map, and the old index
            // segment still hangs off the sources.
            let entry = dataset
                .load_index_by_name(FRAG_REUSE_INDEX_NAME)
                .await
                .unwrap()
                .expect("the stable-partition commit installs the entry");
            assert_eq!(entry.index_version, 1);
            let ledger = decode_frag_reuse_ledger(&dataset, &entry).await.unwrap();
            assert_eq!(ledger.transitions().len(), 1);
            let Mapping::StablePartition(partition) = ledger.transitions()[0].mapping() else {
                panic!("expected a stable-partition transition");
            };
            assert_eq!(partition.map_id, map_id);
            assert_eq!(fixture.list_fri_map_dirs().await, vec![map_id.clone()]);
            let old_segment = dataset
                .load_index_by_name("i_idx")
                .await
                .unwrap()
                .expect("fixture index");

            // Stage 2: the trim retains the entry while the segment still
            // derives from the sources.
            cleanup_frag_reuse_index(&mut dataset).await.unwrap();
            assert!(
                dataset
                    .load_index_by_name(FRAG_REUSE_INDEX_NAME)
                    .await
                    .unwrap()
                    .is_some()
            );

            // Stage 3: drain -- rebuild the index over the destinations,
            // dropping the old segment.
            dataset
                .create_index(
                    &["i"],
                    IndexType::Scalar,
                    Some("i_idx".into()),
                    &ScalarIndexParams::default(),
                    true,
                )
                .await
                .unwrap();
            let rebuilt = dataset
                .load_index_by_name("i_idx")
                .await
                .unwrap()
                .expect("rebuilt index");
            assert_ne!(rebuilt.uuid, old_segment.uuid);

            // Stage 4: the trim now deletes the fully drained entry.
            cleanup_frag_reuse_index(&mut dataset).await.unwrap();
            assert!(
                dataset
                    .load_index_by_name(FRAG_REUSE_INDEX_NAME)
                    .await
                    .unwrap()
                    .is_none()
            );

            // Stage 5: a young cleanup keeps the map -- the manifests that
            // reference it are still retained.
            fixture
                .run_cleanup(utc_now() - TimeDelta::try_days(7).unwrap())
                .await
                .unwrap();
            assert_eq!(fixture.list_fri_map_dirs().await, vec![map_id]);

            // Stage 6: once aged, the referencing manifests expire and the
            // map goes with them; the table keeps working.
            MockClock::set_system_time(TimeDelta::try_days(10).unwrap().to_std().unwrap());
            fixture
                .run_cleanup(utc_now() - TimeDelta::try_seconds(1).unwrap())
                .await
                .unwrap();
            assert!(fixture.list_fri_map_dirs().await.is_empty());
            let reopened = fixture.open().await.unwrap();
            assert_eq!(reopened.count_rows(None).await.unwrap(), 8);
            assert_eq!(
                reopened
                    .count_rows(Some("i >= 4".to_string()))
                    .await
                    .unwrap(),
                4
            );
        }

        /// Commit a fresh `i_idx` delta segment built over the current
        /// (translated) table state; `covered` optionally narrows the
        /// committed bitmap to an under-claim.
        async fn commit_delta_segment(dataset: &mut Dataset, covered: Option<&[u32]>) -> Uuid {
            let params = ScalarIndexParams::default();
            let mut delta =
                crate::index::CreateIndexBuilder::new(dataset, &["i"], IndexType::BTree, &params)
                    .name("i_idx_delta".into())
                    .execute_uncommitted()
                    .await
                    .unwrap();
            delta.name = "i_idx".into();
            if let Some(covered) = covered {
                delta.fragment_bitmap = Some(covered.iter().copied().collect());
            }
            let uuid = delta.uuid;
            dataset
                .apply_commit(
                    Transaction::new(
                        dataset.manifest.version,
                        Operation::CreateIndex {
                            new_indices: vec![delta],
                            removed_indices: vec![],
                        },
                        None,
                    ),
                    &Default::default(),
                    &Default::default(),
                )
                .await
                .unwrap();
            uuid
        }

        async fn stored_segments(dataset: &Dataset, name: &str) -> Vec<IndexMetadata> {
            lance_table::io::manifest::read_manifest_indexes(
                &dataset.object_store,
                &dataset.manifest_location,
                &dataset.manifest,
            )
            .await
            .unwrap()
            .into_iter()
            .filter(|idx| idx.name == name)
            .collect()
        }

        /// Draining through `optimize_indices` merge on a tagged table. A
        /// provenance-only segment still owning translated coverage is opened
        /// through the translating loader, so the merged file holds live
        /// addresses. The merged segment commits the UNION of the selected
        /// segments' stored bitmaps (provenance, retired sources included):
        /// the tagged reader treats a segment's bitmap as provenance, stops
        /// translating at a live fragment, and derives coverage of a
        /// destination whenever every contributing source is present, so the
        /// merged segment keeps the translated destination coverage instead
        /// of degrading to scan fallback, and the transition it still names
        /// stays retained.
        ///
        /// A provenance-only segment whose translated coverage is fully taken
        /// over by siblings is excluded from the query listing by
        /// direct-coverage-wins and cannot be selected for a merge; it is
        /// dead weight that the prune step removes
        /// (`delta_drain_stages_partial_then_full`).
        #[tokio::test]
        #[serial_test::serial(frag_reuse_maintenance)]
        async fn merge_keeps_translated_coverage_on_tagged_tables() {
            use lance_index::optimize::OptimizeOptions;

            let fixture = MockDatasetFixture::try_new().unwrap();
            let (mut dataset, _map_id) = make_tagged(&fixture).await;
            // A partial delta: the old segment keeps exclusive translated
            // coverage of destination 11, so it stays openable.
            commit_delta_segment(&mut dataset, Some(&[10])).await;
            assert_eq!(stored_segments(&dataset, "i_idx").await.len(), 2);
            let all_rows = dataset.count_rows(None).await.unwrap();
            let scan_counts = counts_without_index(&dataset).await;

            dataset
                .optimize_indices(&OptimizeOptions::merge(2))
                .await
                .unwrap();

            // One merged segment; its stored bitmap is provenance: the retired
            // sources of the old segment plus the delta's direct coverage.
            let merged = stored_segments(&dataset, "i_idx").await;
            assert_eq!(merged.len(), 1, "{merged:?}");
            let stored = merged[0].fragment_bitmap.clone().unwrap();
            assert!(
                stored.contains(0) && stored.contains(1) && stored.contains(10),
                "{stored:?}"
            );

            // The reader derives coverage of BOTH destinations from that
            // provenance, so nothing falls back to scan.
            let derived = derived_coverage(&dataset, "i_idx").await;
            assert_eq!(
                derived,
                dataset.fragment_bitmap.as_ref().clone(),
                "the merged segment must answer for every live fragment"
            );
            assert_index_used(&dataset).await;
            assert_eq!(dataset.count_rows(None).await.unwrap(), all_rows);
            assert_eq!(counts_with_index(&dataset).await, scan_counts);

            // Trim keeps the transition: the merged segment's provenance still
            // names its sources and destination 11 has no direct coverage.
            cleanup_frag_reuse_index(&mut dataset).await.unwrap();
            assert!(
                dataset
                    .load_index_by_name(FRAG_REUSE_INDEX_NAME)
                    .await
                    .unwrap()
                    .is_some(),
                "the merged segment still depends on the transition"
            );
            assert_eq!(stored_segments(&dataset, "i_idx").await.len(), 1);
            assert_eq!(counts_with_index(&dataset).await, scan_counts);
        }

        /// A merge that selects only SOME of the segments contributing to a
        /// destination must not strip the unselected sibling of its coverage.
        /// Two per-fragment segments S1 (fragment 0) and S2 (fragment 1) both
        /// contribute to destinations 10 and 11 after the recluster. Merging
        /// S2 with newly appended data produces M whose provenance still names
        /// fragment 1, so the reader keeps resolving 10 and 11 from {S1, M}.
        /// With a `stored ∩ live` bitmap M would drop fragment 1 and S1 would
        /// lose both destinations.
        #[tokio::test]
        #[serial_test::serial(frag_reuse_maintenance)]
        async fn merge_keeps_unselected_sibling_coverage_on_tagged_tables() {
            use lance_index::optimize::OptimizeOptions;

            let fixture = MockDatasetFixture::try_new().unwrap();
            let (mut dataset, _map_id) = make_tagged_per_fragment(&fixture).await;
            let segments = stored_segments(&dataset, "i_idx").await;
            assert_eq!(segments.len(), 2);
            let s1 = segments
                .iter()
                .find(|s| s.fragment_bitmap.as_ref().unwrap().contains(0))
                .unwrap()
                .uuid;

            // New data so a one-segment merge has something to fold in.
            let appended = arrow_array::record_batch!(("i", Int32, [8, 9, 10, 11])).unwrap();
            dataset
                .append(
                    RecordBatchIterator::new(vec![Ok(appended.clone())], appended.schema()),
                    Some(WriteParams {
                        store_params: Some(fixture.os_params()),
                        commit_handler: Some(Arc::new(RenameCommitHandler)),
                        ..Default::default()
                    }),
                )
                .await
                .unwrap();
            let all_rows = dataset.count_rows(None).await.unwrap();
            let scan_counts = counts_without_index(&dataset).await;

            dataset
                .optimize_indices(&OptimizeOptions::merge(1))
                .await
                .unwrap();

            let after = stored_segments(&dataset, "i_idx").await;
            assert_eq!(after.len(), 2, "{after:?}");
            assert!(after.iter().any(|s| s.uuid == s1), "S1 was not selected");
            // Both destinations are still derived for S1 and for the merged
            // segment, and every live fragment is covered.
            let derived = dataset.load_indices().await.unwrap();
            for segment in derived.iter().filter(|s| s.name == "i_idx") {
                let coverage = segment.fragment_bitmap.as_ref().unwrap();
                assert!(
                    coverage.contains(10) && coverage.contains(11),
                    "segment {} lost translated coverage: {coverage:?}",
                    segment.uuid
                );
            }
            assert_eq!(
                derived_coverage(&dataset, "i_idx").await,
                dataset.fragment_bitmap.as_ref().clone()
            );
            assert_index_used(&dataset).await;
            assert_eq!(dataset.count_rows(None).await.unwrap(), all_rows);
            assert_eq!(counts_with_index(&dataset).await, scan_counts);
        }

        /// Sorted per-value row counts through the scalar index.
        async fn counts_with_index(dataset: &Dataset) -> Vec<usize> {
            let mut counts = Vec::new();
            for value in 0..12 {
                let mut scan = dataset.scan();
                scan.filter(&format!("i = {value}")).unwrap();
                counts.push(scan.try_into_batch().await.unwrap().num_rows());
            }
            counts
        }

        /// The same counts with the scalar index disabled: the ground truth.
        async fn counts_without_index(dataset: &Dataset) -> Vec<usize> {
            let mut counts = Vec::new();
            for value in 0..12 {
                let mut scan = dataset.scan();
                scan.filter(&format!("i = {value}")).unwrap();
                scan.use_scalar_index(false);
                counts.push(scan.try_into_batch().await.unwrap().num_rows());
            }
            counts
        }

        /// Union of the coverage the tagged reader derives for `name`.
        async fn derived_coverage(dataset: &Dataset, name: &str) -> roaring::RoaringBitmap {
            dataset
                .load_indices()
                .await
                .unwrap()
                .iter()
                .filter(|s| s.name == name)
                .filter_map(|s| s.fragment_bitmap.clone())
                .fold(roaring::RoaringBitmap::new(), |acc, b| acc | b)
        }

        /// A filtered scan plans through the scalar index (no scan fallback).
        async fn assert_index_used(dataset: &Dataset) {
            let mut scan = dataset.scan();
            scan.filter("i >= 4").unwrap();
            let plan = scan.explain_plan(false).await.unwrap();
            assert!(plan.contains("ScalarIndexQuery"), "{plan}");
        }

        /// `make_tagged` with `i_idx` built as one segment per source fragment.
        async fn make_tagged_per_fragment(fixture: &MockDatasetFixture) -> (Dataset, String) {
            let data = lance_datagen::gen_batch()
                .col("i", lance_datagen::array::step::<Int32Type>())
                .into_reader_rows(
                    lance_datagen::RowCount::from(4),
                    lance_datagen::BatchCount::from(2),
                );
            Dataset::write(
                data,
                &fixture.dataset_path,
                Some(WriteParams {
                    store_params: Some(fixture.os_params()),
                    commit_handler: Some(Arc::new(RenameCommitHandler)),
                    mode: WriteMode::Create,
                    max_rows_per_file: 4,
                    ..Default::default()
                }),
            )
            .await
            .unwrap();
            let mut dataset = *fixture.open().await.unwrap();
            let params = ScalarIndexParams::default();
            let fragment_ids: Vec<u32> = dataset.fragments().iter().map(|f| f.id as u32).collect();
            let mut segments = Vec::new();
            for fragment in fragment_ids {
                segments.push(
                    crate::index::CreateIndexBuilder::new(
                        &mut dataset,
                        &["i"],
                        IndexType::BTree,
                        &params,
                    )
                    .name("i_idx".into())
                    .fragments(vec![fragment])
                    .execute_uncommitted()
                    .await
                    .unwrap(),
                );
            }
            dataset
                .commit_existing_index_segments("i_idx", "i", segments)
                .await
                .unwrap();
            reserve_fragments(&mut dataset, 20).await;
            let old_fragments: Vec<Fragment> = dataset.fragments().iter().cloned().collect();
            let (transition, destinations) = reader_tests::prepare(&dataset).await;
            let Some(pb_fri::transition::Mapping::StablePartition(mapping)) = &transition.mapping
            else {
                unreachable!()
            };
            let map_id = mapping.map_id.clone();
            let read_version = dataset.manifest.version;
            let frag_reuse_index = Some(
                crate::index::frag_reuse::frag_reuse_entry_appending(&dataset, vec![transition])
                    .await
                    .unwrap(),
            );
            let dataset = CommitBuilder::new(Arc::new(dataset))
                .execute(Transaction::new(
                    read_version,
                    Operation::Rewrite {
                        groups: vec![RewriteGroup {
                            old_fragments,
                            new_fragments: destinations,
                        }],
                        rewritten_indices: vec![],
                        frag_reuse_index,
                    },
                    None,
                ))
                .await
                .unwrap();
            (dataset, map_id)
        }

        /// The staged drain the merge tests were after, via delta segments:
        /// partial direct coverage retains the transition; completing the
        /// coverage lets one maintenance run prune the superseded segment,
        /// release the transition, and (aged) collect the row map.
        #[tokio::test]
        #[serial_test::serial(frag_reuse_maintenance)]
        async fn delta_drain_stages_partial_then_full() {
            let fixture = MockDatasetFixture::try_new().unwrap();
            let (mut dataset, map_id) = make_tagged(&fixture).await;
            let all_rows = dataset.count_rows(None).await.unwrap();

            // Stage 1: only destination 10 gains direct coverage; the
            // transition and the old segment must both stay.
            commit_delta_segment(&mut dataset, Some(&[10])).await;
            cleanup_frag_reuse_index(&mut dataset).await.unwrap();
            assert_eq!(stored_segments(&dataset, "i_idx").await.len(), 2);
            assert!(
                dataset
                    .load_index_by_name(FRAG_REUSE_INDEX_NAME)
                    .await
                    .unwrap()
                    .is_some(),
                "destination 11 lacks direct coverage, so the transition stays"
            );
            assert_eq!(fixture.list_fri_map_dirs().await, vec![map_id.clone()]);

            // Stage 2: destination 11 gains direct coverage too. One
            // maintenance run prunes the superseded old segment and releases
            // the transition.
            let second = commit_delta_segment(&mut dataset, Some(&[11])).await;
            cleanup_frag_reuse_index(&mut dataset).await.unwrap();
            let segments = stored_segments(&dataset, "i_idx").await;
            assert_eq!(segments.len(), 2, "the two deltas remain");
            assert!(segments.iter().any(|s| s.uuid == second));
            assert!(
                dataset
                    .load_index_by_name(FRAG_REUSE_INDEX_NAME)
                    .await
                    .unwrap()
                    .is_none(),
                "with the old segment pruned, nothing needs the transition"
            );
            assert_eq!(dataset.count_rows(None).await.unwrap(), all_rows);

            // Stage 3: the released row map ages out with its manifests.
            MockClock::set_system_time(TimeDelta::try_days(10).unwrap().to_std().unwrap());
            fixture
                .run_cleanup(utc_now() - TimeDelta::try_seconds(1).unwrap())
                .await
                .unwrap();
            assert!(fixture.list_fri_map_dirs().await.is_empty());
            let reopened = fixture.open().await.unwrap();
            assert_eq!(reopened.count_rows(None).await.unwrap(), all_rows);
        }

        /// D-guard: a v0 dataset records no tagged entries and has no `_fri`
        /// directory; cleanup behaves exactly as before.
        #[tokio::test]
        async fn v0_dataset_is_untouched() {
            let fixture = MockDatasetFixture::try_new().unwrap();
            fixture.create_some_data().await.unwrap();
            fixture.overwrite_some_data().await.unwrap();
            MockClock::set_system_time(TimeDelta::try_days(10).unwrap().to_std().unwrap());
            let removed = fixture
                .run_cleanup(utc_now() - TimeDelta::try_days(8).unwrap())
                .await
                .unwrap();
            assert_eq!(removed.old_versions, 1);
            assert!(fixture.list_fri_map_dirs().await.is_empty());
        }
    }
}