From 835227dcbc6dcdc8f5709c57e9409379aec166e6 Mon Sep 17 00:00:00 2001 From: "Lei, HUANG" Date: Sat, 24 Oct 2026 01:07:17 +0800 Subject: [PATCH] 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 --- src/cmd/src/datanode/objbench.rs | 9 +- src/mito2/src/engine/file_ref_test.rs | 95 +++++++++++++++++ src/mito2/src/gc.rs | 8 +- src/mito2/src/gc/worker_test.rs | 140 +++++++++++++++++++++++++ src/mito2/src/sst/file.rs | 74 ++++++++++--- src/mito2/src/sst/file_purger.rs | 143 ++++++++++++++++---------- src/mito2/src/sst/file_ref.rs | 10 +- src/mito2/src/sst/version.rs | 38 ++++++- 8 files changed, 435 insertions(+), 82 deletions(-) diff --git a/src/cmd/src/datanode/objbench.rs b/src/cmd/src/datanode/objbench.rs index 03696c96fc0..248acc09bf6 100644 --- a/src/cmd/src/datanode/objbench.rs +++ b/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( diff --git a/src/mito2/src/engine/file_ref_test.rs b/src/mito2/src/engine/file_ref_test.rs index 96f5f6661f4..c30fb08410f 100644 --- a/src/mito2/src/engine/file_ref_test.rs +++ b/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[®ion_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[®ion_id].is_empty()); + drop(version); + let snapshot = engine + .get_snapshot_of_file_refs([region_id], HashMap::new()) + .await + .unwrap(); + assert!( + snapshot + .file_refs + .get(®ion_id) + .is_none_or(HashSet::is_empty) + ); +} diff --git a/src/mito2/src/gc.rs b/src/mito2/src/gc.rs index a5474099cbf..9535cc96288 100644 --- a/src/mito2/src/gc.rs +++ b/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)) diff --git a/src/mito2/src/gc/worker_test.rs b/src/mito2/src/gc/worker_test.rs index 3d389e7e1d7..22e50fac318 100644 --- a/src/mito2/src/gc/worker_test.rs +++ b/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(®ion_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(); diff --git a/src/mito2/src/sst/file.rs b/src/mito2/src/sst/file.rs index 28b6218cd7b..3ca25d0e453 100644 --- a/src/mito2/src/sst/file.rs +++ b/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, - file_purger: FilePurgerRef, + lifetime: Arc, } 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, + 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, + 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()); } } } diff --git a/src/mito2/src/sst/file_purger.rs b/src/mito2/src/sst/file_purger.rs index 2660a19d4af..2a2516c15b4 100644 --- a/src/mito2/src/sst/file_purger.rs +++ b/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; 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] diff --git a/src/mito2/src/sst/file_ref.rs b/src/mito2/src/sst/file_ref.rs index 4dee84ca1fa..74bd497e7f6 100644 --- a/src/mito2/src/sst/file_ref.rs +++ b/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); diff --git a/src/mito2/src/sst/version.rs b/src/mito2/src/sst/version.rs index 1ca0410db7b..5351195eefb 100644 --- a/src/mito2/src/sst/version.rs +++ b/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();