fix(mito): preserve SST lifetime across index replacement

Share retirement and final purge across immutable index metadata views. Keep version-specific GC references until readers release them and match Parquet liveness by file identity.

Fixes #9944

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

src/cmd/src/datanode/objbench.rs

@@ -27,7 +27,7 @@ use mito2::config::{FulltextIndexConfig, MitoConfig, Mode};
 use mito2::read::FlatSource;
 use mito2::sst::FormatType;
 use mito2::sst::file::{FileHandle, FileMeta};
-use mito2::sst::file_purger::{FilePurger, FilePurgerRef};
+use mito2::sst::file_purger::{FilePurgerRef, NoopFilePurger};
 use mito2::sst::index::intermediate::IntermediateManager;
 use mito2::sst::index::puffin_manager::PuffinManagerFactory;
 use mito2::sst::parquet::WriteOptions;
@@ -508,12 +508,7 @@ async fn build_cache_manager(
 }
 
 fn new_noop_file_purger() -> FilePurgerRef {
-    #[derive(Debug)]
-    struct Noop;
-    impl FilePurger for Noop {
-        fn remove_file(&self, _file_meta: FileMeta, _is_delete: bool, _index_outdated: bool) {}
-    }
-    Arc::new(Noop)
+    Arc::new(NoopFilePurger)
 }
 
 async fn load_parquet_metadata(

src/mito2/src/engine/file_ref_test.rs

@@ -53,3 +53,98 @@ async fn get_snapshot_of_file_refs_skips_missing_file_handle_region() {
     assert!(manifest.manifest_version.is_empty());
     assert!(manifest.cross_region_refs.is_empty());
 }
+
+#[tokio::test]
+async fn get_snapshot_preserves_old_index_readers_until_release() {
+    use store_api::region_engine::{RegionEngine, RegionRole};
+    use store_api::region_request::RegionRequest;
+    use store_api::storage::{FileId, FileRef};
+
+    use crate::manifest::action::{RegionEdit, RegionMetaAction, RegionMetaActionList};
+    use crate::region::RegionLeaderState;
+    use crate::sst::file::{FileMeta, IndexType};
+    use crate::sst::version::SstVersion;
+    use crate::test_util::CreateRequestBuilder;
+
+    let mut env = TestEnv::new().await;
+    let engine = env
+        .create_engine(MitoConfig {
+            gc: crate::gc::GcConfig {
+                enable: true,
+                ..Default::default()
+            },
+            ..Default::default()
+        })
+        .await;
+    let region_id = RegionId::new(1, 1);
+    engine
+        .handle_request(
+            region_id,
+            RegionRequest::Create(CreateRequestBuilder::new().build()),
+        )
+        .await
+        .unwrap();
+    engine
+        .set_region_role(region_id, RegionRole::Leader)
+        .unwrap();
+    let region = engine.get_region(region_id).unwrap();
+    let meta = FileMeta {
+        region_id,
+        file_id: FileId::random(),
+        available_indexes: smallvec::smallvec![IndexType::InvertedIndex],
+        ..Default::default()
+    };
+    let mut version = SstVersion::new(crate::test_util::memtable_util::metadata_for_test());
+    version.add_files(region.file_purger.clone(), [meta.clone()].into_iter());
+    let old_reader = version.levels()[0].files[&meta.file_id].clone();
+    let current_meta = FileMeta {
+        index_version: 1,
+        ..meta.clone()
+    };
+    version.add_files(
+        region.file_purger.clone(),
+        [current_meta.clone()].into_iter(),
+    );
+    region
+        .manifest_ctx
+        .update_manifest(
+            RegionLeaderState::Writable,
+            RegionMetaActionList::with_action(RegionMetaAction::Edit(RegionEdit {
+                files_to_add: vec![current_meta],
+                files_to_remove: vec![],
+                timestamp_ms: None,
+                compaction_time_window: None,
+                flushed_entry_id: None,
+                flushed_sequence: None,
+                committed_sequence: None,
+            })),
+            false,
+        )
+        .await
+        .unwrap();
+    let snapshot = engine
+        .get_snapshot_of_file_refs([region_id], HashMap::new())
+        .await
+        .unwrap();
+    assert_eq!(
+        snapshot.file_refs[&region_id],
+        HashSet::from([FileRef::new(region_id, meta.file_id, Some(0))])
+    );
+    drop(old_reader);
+    let snapshot = engine
+        .get_snapshot_of_file_refs([region_id], HashMap::new())
+        .await
+        .unwrap();
+    assert!(snapshot.file_refs[&region_id].is_empty());
+    drop(version);
+    let snapshot = engine
+        .get_snapshot_of_file_refs([region_id], HashMap::new())
+        .await
+        .unwrap();
+    assert!(
+        snapshot
+            .file_refs
+            .get(&region_id)
+            .is_none_or(HashSet::is_empty)
+    );
+}

src/mito2/src/gc.rs

@@ -1091,6 +1091,8 @@ impl LocalGcWorker {
         // When full_file_listing is false, skip expensive list operations and only delete
         // files that are tracked in recently_removed_files
         if !self.full_file_listing {
+            let referenced_ssts: HashSet<_> =
+                in_tmp_ref.iter().map(|(file_id, _)| *file_id).collect();
             // Only delete files that:
             // 1. Are in recently_removed_files (tracked in manifest)
             // 2. Are not in use(in manifest or tmp ref)
@@ -1099,9 +1101,9 @@ impl LocalGcWorker {
                 .iter()
                 .filter(|file_id| {
                     let in_use = match file_id {
-                        RemovedFile::File(file_id, index_version) => {
-                            in_manifest.get(file_id) == Some(index_version)
-                                || in_tmp_ref.contains(&(*file_id, *index_version))
+                        RemovedFile::File(file_id, _) => {
+                            // Any index-version reader still needs the same Parquet file.
+                            in_manifest.contains_key(file_id) || referenced_ssts.contains(file_id)
                         }
                         RemovedFile::Index(file_id, index_version) => {
                             in_manifest.get(file_id) == Some(&Some(*index_version))

src/mito2/src/gc/worker_test.rs

@@ -74,6 +74,146 @@ async fn create_gc_worker(
     .unwrap()
 }
 
+#[rstest::rstest]
+#[tokio::test]
+async fn test_gc_keeps_sst_across_index_versions(
+    #[values(false, true)] full_file_listing: bool,
+    #[values(false, true)] old_reader_last: bool,
+) {
+    use crate::sst::file::{FileMeta, IndexType};
+    use crate::sst::version::SstVersion;
+    use common_time::Timestamp;
+
+    let mut env = TestEnv::new().await;
+    let engine = env
+        .create_engine(MitoConfig {
+            gc: GcConfig {
+                enable: true,
+                lingering_time: None,
+                ..Default::default()
+            },
+            ..Default::default()
+        })
+        .await;
+    let region_id = RegionId::new(1, 1);
+    engine
+        .handle_request(
+            region_id,
+            RegionRequest::Create(CreateRequestBuilder::new().build()),
+        )
+        .await
+        .unwrap();
+    let region = engine.get_region(region_id).unwrap();
+    let meta = FileMeta {
+        region_id,
+        file_id: store_api::storage::FileId::random(),
+        available_indexes: smallvec::smallvec![IndexType::InvertedIndex],
+        ..Default::default()
+    };
+    let mut version = SstVersion::new(crate::test_util::memtable_util::metadata_for_test());
+    version.add_files(region.file_purger.clone(), [meta.clone()].into_iter());
+    let old = version.levels()[0].files[&meta.file_id].clone();
+    let updated = FileMeta {
+        index_version: 2,
+        ..meta.clone()
+    };
+    version.add_files(region.file_purger.clone(), [updated].into_iter());
+    let new = version.levels()[0].files[&meta.file_id].clone();
+    version.mark_all_deleted();
+    drop(version);
+
+    let store = region.access_layer.object_store();
+    let mut entries = Vec::new();
+    for suffix in ["parquet", "puffin", "2.puffin"] {
+        entries.push(write_and_list_entry(store, &format!("{}.{}", meta.file_id, suffix)).await);
+    }
+    let removed = BTreeMap::from([(
+        Timestamp::new_millisecond(0),
+        HashSet::from([
+            RemovedFile::File(meta.file_id, Some(2)),
+            RemovedFile::Index(meta.file_id, 0),
+            RemovedFile::Index(meta.file_id, 2),
+        ]),
+    )]);
+    let worker = create_gc_worker(
+        &engine,
+        BTreeMap::from([(region_id, Some(region.clone()))]),
+        &FileRefsManifest::default(),
+        full_file_listing,
+    )
+    .await;
+    let (first, last) = if old_reader_last {
+        (new, old)
+    } else {
+        (old, new)
+    };
+    let released_version = first.index_id().version;
+    let live_version = last.index_id().version;
+    let mut handles = [Some(first), Some(last)];
+    for released in 0..=2 {
+        if released > 0 {
+            drop(handles[released - 1].take());
+        }
+        let refs = engine
+            .get_snapshot_of_file_refs([region_id], HashMap::new())
+            .await
+            .unwrap();
+        let refs = refs.file_refs.get(&region_id).cloned().unwrap_or_default();
+        let expected_versions = match released {
+            0 => vec![0, 2],
+            1 => vec![live_version],
+            _ => vec![],
+        };
+        assert_eq!(
+            refs,
+            expected_versions
+                .into_iter()
+                .map(|v| FileRef::new(region_id, meta.file_id, Some(v)))
+                .collect()
+        );
+        let tmp_refs = refs
+            .into_iter()
+            .map(|f| (f.file_id, f.index_version))
+            .collect();
+        let deleted: HashSet<_> = worker
+            .list_to_be_deleted_files(
+                region_id,
+                false,
+                &HashMap::new(),
+                &tmp_refs,
+                removed.clone(),
+                entries.clone(),
+            )
+            .await
+            .unwrap()
+            .into_iter()
+            .collect();
+        let expected = match released {
+            0 => HashSet::new(),
+            1 => HashSet::from([RemovedFile::Index(meta.file_id, released_version)]),
+            _ => HashSet::from([
+                RemovedFile::File(meta.file_id, if full_file_listing { None } else { Some(2) }),
+                RemovedFile::Index(meta.file_id, 0),
+                RemovedFile::Index(meta.file_id, 2),
+            ]),
+        };
+        assert_eq!(deleted, expected);
+    }
+    // A manifest reference also protects Parquet regardless of its index version.
+    let deleted = worker
+        .list_to_be_deleted_files(
+            region_id,
+            false,
+            &HashMap::from([(meta.file_id, Some(3))]),
+            &HashSet::new(),
+            removed,
+            entries,
+        )
+        .await
+        .unwrap();
+    assert!(deleted.iter().all(|f| matches!(f, RemovedFile::Index(..))));
+}
+
 #[tokio::test]
 async fn test_gc_worker_ignores_dropping_marker_file() {
     init_default_ut_logging();

src/mito2/src/sst/file.rs

@@ -551,13 +551,13 @@ impl fmt::Debug for FileHandle {
         f.debug_struct("FileHandle")
             .field("meta", self.meta_ref())
             .field("compacting", &self.compacting())
-            .field("deleted", &self.inner.deleted.load(Ordering::Relaxed))
+            .field("deleted", &self.is_deleted())
             .finish()
     }
 }
 
 impl FileHandle {
-    /// Creates a handle sharing the file's physical state and original statistics.
+    /// Creates a new logical SST reference. Index replacements must reuse its lifetime.
     pub fn new(meta: FileMeta, file_purger: FilePurgerRef) -> FileHandle {
         let pk_range = meta.primary_key_range();
         FileHandle {
@@ -576,6 +576,23 @@ impl FileHandle {
         }
     }
 
+    /// Replaces an index metadata view without splitting the underlying SST lifetime.
+    pub(crate) fn with_updated_index(&self, meta: FileMeta) -> Self {
+        debug_assert_eq!(self.file_id(), meta.file_id());
+        let primary_key_range = meta
+            .primary_key_range()
+            .or_else(|| self.raw_primary_key_range());
+        let lifetime = self.inner.lifetime.clone();
+        *lifetime.meta.lock().unwrap() = meta.clone();
+        Self {
+            inner: Arc::new(FileHandleInner::with_lifetime(
+                meta,
+                lifetime,
+                primary_key_range,
+            )),
+        }
+    }
+
     /// Returns the region id of the file.
     pub fn region_id(&self) -> RegionId {
         self.inner.meta.region_id
@@ -615,7 +632,7 @@ impl FileHandle {
 
     /// Mark the file as deleted and will delete it on drop asynchronously
     pub fn mark_deleted(&self) {
-        self.inner.deleted.store(true, Ordering::Relaxed);
+        self.inner.lifetime.deleted.store(true, Ordering::Relaxed);
     }
 
     pub fn compacting(&self) -> bool {
@@ -650,7 +667,7 @@ impl FileHandle {
     }
 
     pub fn file_purger(&self) -> FilePurgerRef {
-        self.inner.file_purger.clone()
+        self.inner.lifetime.file_purger.clone()
     }
 
     pub fn size(&self) -> u64 {
@@ -670,7 +687,7 @@ impl FileHandle {
     }
 
     pub fn is_deleted(&self) -> bool {
-        self.inner.deleted.load(Ordering::Relaxed)
+        self.inner.lifetime.deleted.load(Ordering::Relaxed)
     }
 
     /// Returns bounds aligned to the caller's pinned schema, before any comparison
@@ -776,39 +793,66 @@ impl PrimaryKeyRange {
 struct FileHandleInner {
     meta: FileMeta,
     compacting: AtomicBool,
-    deleted: AtomicBool,
     index_outdated: AtomicBool,
     primary_key_range: RwLock<PrimaryKeyRange>,
-    file_purger: FilePurgerRef,
+    lifetime: Arc<SstLifetime>,
 }
 
 impl Drop for FileHandleInner {
     fn drop(&mut self) {
-        self.file_purger.remove_file(
+        self.lifetime.file_purger.remove_file(
             self.meta.clone(),
-            self.deleted.load(Ordering::Acquire),
             self.index_outdated.load(Ordering::Acquire),
         );
     }
 }
 
 impl FileHandleInner {
-    /// There should only be one `FileHandleInner` for each file on a datanode
     fn new(
         meta: FileMeta,
         file_purger: FilePurgerRef,
         primary_key_range: Option<(Bytes, Bytes)>,
-    ) -> FileHandleInner {
-        file_purger.new_file(&meta);
-        FileHandleInner {
+    ) -> Self {
+        let lifetime = Arc::new(SstLifetime {
+            meta: Mutex::new(meta.clone()),
+            deleted: AtomicBool::new(false),
+            file_purger,
+        });
+        Self::with_lifetime(meta, lifetime, primary_key_range)
+    }
+
+    fn with_lifetime(
+        meta: FileMeta,
+        lifetime: Arc<SstLifetime>,
+        primary_key_range: Option<(Bytes, Bytes)>,
+    ) -> Self {
+        lifetime.file_purger.new_file(&meta);
+        Self {
             meta,
             compacting: AtomicBool::new(false),
-            deleted: AtomicBool::new(false),
             index_outdated: AtomicBool::new(false),
             primary_key_range: RwLock::new(
                 primary_key_range.map_or(PrimaryKeyRange::Missing, PrimaryKeyRange::Raw),
             ),
-            file_purger,
+            lifetime,
+        }
+    }
+}
+
+/// Physical deletion must wait for readers of every index metadata version.
+/// Independent logical references (for example, copied regions) do not share this guard.
+struct SstLifetime {
+    /// Latest metadata ensures final purge includes indexes newer than the last reader's view.
+    meta: Mutex<FileMeta>,
+    deleted: AtomicBool,
+    file_purger: FilePurgerRef,
+}
+
+impl Drop for SstLifetime {
+    fn drop(&mut self) {
+        if self.deleted.load(Ordering::Acquire) {
+            self.file_purger
+                .purge_file(self.meta.get_mut().unwrap().clone());
         }
     }
 }

src/mito2/src/sst/file_purger.rs

@@ -29,11 +29,13 @@ use crate::sst::range_index::RangeIndexDeleter;
 
 /// A worker to delete files in background.
 pub trait FilePurger: Send + Sync + fmt::Debug {
-    /// Send a request to remove the file.
-    /// If `is_delete` is true, the file will be deleted from the storage.
-    /// Otherwise, only the reference will be removed.
-    /// If `index_outdated` is true, the index file will be deleted regardless of `is_delete`.
-    fn remove_file(&self, file_meta: FileMeta, is_delete: bool, index_outdated: bool);
+    /// Releases one metadata version's reference, deleting its index if outdated.
+    /// This never deletes the SST itself.
+    fn remove_file(&self, file_meta: FileMeta, index_outdated: bool);
+
+    /// Purges a retired SST after all metadata versions release their readers.
+    /// This does not release a reference; each version already did so on drop.
+    fn purge_file(&self, file_meta: FileMeta);
 
     /// Notify the purger of a new file created.
     /// This is useful for object store based storage, where we need to track the file references
@@ -50,9 +52,9 @@ pub type FilePurgerRef = Arc<dyn FilePurger>;
 pub struct NoopFilePurger;
 
 impl FilePurger for NoopFilePurger {
-    fn remove_file(&self, _file_meta: FileMeta, _is_delete: bool, _index_outdated: bool) {
-        // noop
-    }
+    fn remove_file(&self, _file_meta: FileMeta, _index_outdated: bool) {}
+
+    fn purge_file(&self, _file_meta: FileMeta) {}
 }
 
 /// Purger that purges file for current region.
@@ -198,18 +200,20 @@ impl LocalFilePurger {
 }
 
 impl FilePurger for LocalFilePurger {
-    fn remove_file(&self, file_meta: FileMeta, is_delete: bool, index_outdated: bool) {
-        if is_delete {
-            schedule_range_index_deletion(
-                &self.scheduler,
-                self.range_index_deleter.as_ref(),
-                file_meta.file_id,
-            );
-            self.delete_file(file_meta);
-        } else if index_outdated {
+    fn remove_file(&self, file_meta: FileMeta, index_outdated: bool) {
+        if index_outdated {
             self.delete_index(file_meta);
         }
     }
+
+    fn purge_file(&self, file_meta: FileMeta) {
+        schedule_range_index_deletion(
+            &self.scheduler,
+            self.range_index_deleter.as_ref(),
+            file_meta.file_id,
+        );
+        self.delete_file(file_meta);
+    }
 }
 
 pub struct ObjectStoreFilePurger {
@@ -246,19 +250,17 @@ fn schedule_range_index_deletion(
 }
 
 impl FilePurger for ObjectStoreFilePurger {
-    fn remove_file(&self, file_meta: FileMeta, is_delete: bool, _index_outdated: bool) {
-        // if not on local file system, instead inform the global file purger to remove the file reference.
-        // notice that no matter whether the file is deleted or not, we need to remove the reference
-        // because the file is no longer in use nonetheless.
-        // for same reason, we don't care about index_outdated here.
+    fn remove_file(&self, file_meta: FileMeta, _index_outdated: bool) {
+        // GC handles physical SST and index deletion after references are released.
         self.file_ref_manager.remove_file(&file_meta);
-        if is_delete {
-            schedule_range_index_deletion(
-                &self.scheduler,
-                self.range_index_deleter.as_ref(),
-                file_meta.file_id,
-            );
-        }
+    }
+
+    fn purge_file(&self, file_meta: FileMeta) {
+        schedule_range_index_deletion(
+            &self.scheduler,
+            self.range_index_deleter.as_ref(),
+            file_meta.file_id,
+        );
     }
 
     fn new_file(&self, file_meta: &FileMeta) {
@@ -295,6 +297,7 @@ mod tests {
     async fn test_range_index_purge_on_handle_release(
         #[case] gc_enabled: bool,
         #[values(false, true)] is_delete: bool,
+        #[values(false, true)] update_index: bool,
     ) {
         common_telemetry::init_default_ut_logging();
 
@@ -324,7 +327,7 @@ mod tests {
         let path = location::sst_file_path(sst_dir, sst_file_id, layer.path_type());
         object_store.write(&path, vec![0; 4096]).await.unwrap();
 
-        let scheduler = Arc::new(LocalScheduler::new(3));
+        let scheduler = Arc::new(LocalScheduler::new(1));
 
         let index_store = ObjectStore::new(object_store::services::Memory::default()).unwrap();
         let owner = RegionId::new(9, 1);
@@ -341,33 +344,52 @@ mod tests {
         );
 
         {
-            let handle = FileHandle::new(
-                FileMeta {
-                    region_id: sst_file_id.region_id(),
-                    file_id: sst_file_id.file_id(),
-                    time_range: FileTimeRange::default(),
-                    level: 0,
-                    file_size: 4096,
-                    max_row_group_uncompressed_size: 4096,
-                    available_indexes: Default::default(),
-                    indexes: Default::default(),
-                    index_file_size: 0,
-                    index_version: 0,
-                    num_rows: 0,
-                    num_row_groups: 0,
-                    sequence: None,
-                    partition_expr: None,
-                    num_series: 0,
-                    ..Default::default()
-                },
-                file_purger,
+            let meta = FileMeta {
+                region_id: sst_file_id.region_id(),
+                file_id: sst_file_id.file_id(),
+                file_size: 4096,
+                max_row_group_uncompressed_size: 4096,
+                ..Default::default()
+            };
+            let mut version = crate::sst::version::SstVersion::new(
+                crate::test_util::memtable_util::metadata_for_test(),
             );
-            if is_delete {
-                handle.mark_deleted();
+            version.add_files(file_purger, [meta].into_iter());
+            let reader = version.levels()[0].files[&sst_file_id.file_id()].clone();
+            if update_index {
+                for index_version in 1..=2 {
+                    let mut meta = reader.meta_ref().clone();
+                    meta.index_version = index_version;
+                    meta.available_indexes.push(IndexType::InvertedIndex);
+                    let path = location::index_file_path(sst_dir, meta.index_id(), PathType::Bare);
+                    object_store.write(&path, "index").await.unwrap();
+                    version.add_files(reader.file_purger(), [meta].into_iter());
+                }
             }
-            let reader = handle.clone();
-            drop(handle);
+            if is_delete {
+                version.mark_all_deleted();
+            }
+            drop(version);
+            // A single-worker queue barrier makes premature purging deterministic.
+            let (tx, rx) = tokio::sync::oneshot::channel();
+            scheduler
+                .schedule(Box::pin(async move {
+                    tx.send(()).unwrap();
+                }))
+                .unwrap();
+            rx.await.unwrap();
+            assert!(object_store.exists(&path).await.unwrap());
             assert!(index_store.exists(&index_path).await.unwrap());
+            if update_index {
+                for index_version in 1..=2 {
+                    let path = location::index_file_path(
+                        sst_dir,
+                        RegionIndexId::new(sst_file_id, index_version),
+                        PathType::Bare,
+                    );
+                    assert!(object_store.exists(&path).await.unwrap());
+                }
+            }
             drop(reader);
         }
 
@@ -378,6 +400,19 @@ mod tests {
             gc_enabled || !is_delete
         );
         assert_eq!(index_store.exists(&index_path).await.unwrap(), !is_delete);
+        if update_index {
+            for index_version in 1..=2 {
+                let path = location::index_file_path(
+                    sst_dir,
+                    RegionIndexId::new(sst_file_id, index_version),
+                    PathType::Bare,
+                );
+                assert_eq!(
+                    object_store.exists(&path).await.unwrap(),
+                    gc_enabled || !is_delete
+                );
+            }
+        }
     }
 
     #[tokio::test]

src/mito2/src/sst/file_ref.rs

@@ -147,11 +147,17 @@ impl FileReferenceManager {
 
         for r in &query_regions_for_mem {
             let manifest = r.manifest_ctx.manifest().await;
-            // remove in manifest files for smaller size, since gc worker read from manifest later.
+            // Only omit references covered by the exact index version in the manifest.
+            // Older index readers still need their temporary references.
             ref_files.entry(r.region_id()).and_modify(|refs| {
                 *refs = std::mem::take(refs)
                     .into_iter()
-                    .filter(|f| !manifest.files.contains_key(&f.file_id))
+                    .filter(|f| {
+                        !manifest
+                            .files
+                            .get(&f.file_id)
+                            .is_some_and(|meta| meta.index_version() == f.index_version)
+                    })
                     .collect();
             });
             manifest_version.insert(r.region_id(), manifest.manifest_version);

src/mito2/src/sst/version.rs

@@ -100,7 +100,7 @@ impl SstVersion {
                         }
                     } else {
                         // include case like old file have no index or index is outdated
-                        *f = FileHandle::new(file.clone(), file_purger.clone());
+                        *f = f.with_updated_index(file.clone());
                     }
                 })
                 .or_insert_with(|| {
@@ -352,6 +352,42 @@ mod tests {
         assert_eq!(version.time_range(), None);
     }
 
+    #[rstest::rstest]
+    fn test_index_replacement_shares_retirement_with_old_readers(
+        #[values(false, true)] retire_before_update: bool,
+    ) {
+        let purger = new_noop_file_purger();
+        let meta = FileMeta {
+            file_id: FileId::random(),
+            ..Default::default()
+        };
+        let mut version = SstVersion::new(crate::test_util::memtable_util::metadata_for_test());
+        version.add_files(purger.clone(), [meta.clone()].into_iter());
+        let old_snapshot = version.clone();
+        let old_reader = old_snapshot.levels()[0].files[&meta.file_id].clone();
+        let independent = FileHandle::new(meta.clone(), purger.clone());
+        if retire_before_update {
+            old_reader.mark_deleted();
+        }
+        let mut updated = meta.clone();
+        updated.index_version = 1;
+        updated
+            .available_indexes
+            .push(crate::sst::file::IndexType::InvertedIndex);
+        version.add_files(purger, [updated.clone()].into_iter());
+        let new_reader = version.levels()[0].files[&meta.file_id].clone();
+
+        assert_eq!(old_reader.meta_ref(), &meta);
+        assert_eq!(new_reader.meta_ref(), &updated);
+        assert_eq!(new_reader.is_deleted(), retire_before_update);
+        version.remove_files([updated].into_iter());
+        assert!(new_reader.is_deleted());
+        assert!(old_reader.is_deleted());
+        assert!(!independent.is_deleted());
+        drop(version);
+        assert!(old_snapshot.levels()[0].files[&meta.file_id].is_deleted());
+    }
+
     #[test]
     fn test_add_files() {
         let purger = new_noop_file_purger();