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..3979c14553c 100644 --- a/src/mito2/src/engine/file_ref_test.rs +++ b/src/mito2/src/engine/file_ref_test.rs @@ -53,3 +53,204 @@ 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_related_region_old_index_readers() { + 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::{FileHandle, FileMeta, IndexType}; + 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 source_region = RegionId::new(1, 1); + let target_region = RegionId::new(1, 2); + engine + .handle_request( + target_region, + RegionRequest::Create(CreateRequestBuilder::new().build()), + ) + .await + .unwrap(); + engine + .set_region_role(target_region, RegionRole::Leader) + .unwrap(); + let meta = FileMeta { + region_id: source_region, + file_id: FileId::random(), + index_version: 2, + available_indexes: smallvec::smallvec![IndexType::InvertedIndex], + ..Default::default() + }; + let target = engine.get_region(target_region).unwrap(); + let old_reader = FileHandle::new( + FileMeta { + index_version: 0, + ..meta.clone() + }, + target.file_purger.clone(), + ); + target + .manifest_ctx + .update_manifest( + RegionLeaderState::Writable, + RegionMetaActionList::with_action(RegionMetaAction::Edit(RegionEdit { + files_to_add: vec![meta.clone()], + files_to_remove: vec![], + timestamp_ms: None, + compaction_time_window: None, + flushed_entry_id: None, + flushed_sequence: None, + committed_sequence: None, + })), + false, + ) + .await + .unwrap(); + + // The source is not open here, but a query on the destination still uses its v0 index. + let related_regions = HashMap::from([(source_region, HashSet::from([target_region]))]); + let snapshot = engine + .get_snapshot_of_file_refs([], related_regions.clone()) + .await + .unwrap(); + assert_eq!( + snapshot.file_refs[&source_region], + [0, 2] + .into_iter() + .map(|version| FileRef::new(source_region, meta.file_id, Some(version))) + .collect() + ); + assert_eq!( + snapshot.cross_region_refs[&source_region], + HashSet::from([target_region]) + ); + drop(old_reader); + let snapshot = engine + .get_snapshot_of_file_refs([], related_regions) + .await + .unwrap(); + assert_eq!( + snapshot.file_refs[&source_region], + HashSet::from([FileRef::new(source_region, meta.file_id, Some(2))]) + ); +} + +#[rstest::rstest] +#[tokio::test] +async fn get_snapshot_preserves_old_index_readers_until_release( + #[values(false, true)] old_has_index: bool, +) { + 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: if old_has_index { + smallvec::smallvec![IndexType::InvertedIndex] + } else { + Default::default() + }, + ..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, + available_indexes: smallvec::smallvec![IndexType::InvertedIndex], + ..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(); + // Only v1 is covered by the manifest; keep the older view's exact reference. + let expected = HashSet::from([FileRef::new( + region_id, + meta.file_id, + old_has_index.then_some(0), + )]); + assert_eq!(snapshot.file_refs[®ion_id], expected); + // Closing the current view without retirement must keep the old reader protected. + drop(version); + let snapshot = engine + .get_snapshot_of_file_refs([region_id], HashMap::new()) + .await + .unwrap(); + assert_eq!(snapshot.file_refs[®ion_id], expected); + drop(old_reader); + 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 541bb4872b4..8d917fbcffb 100644 --- a/src/mito2/src/gc.rs +++ b/src/mito2/src/gc.rs @@ -929,17 +929,8 @@ impl LocalGcWorker { unknown_file_may_linger_until: chrono::DateTime, ) -> Vec { let mut ready_for_delete = vec![]; - // all group by file id for easier checking - let in_tmp_ref: HashMap> = - in_tmp_ref - .iter() - .fold(HashMap::new(), |mut acc, (file, version)| { - let indices = acc.entry(*file).or_default(); - if let Some(version) = version { - indices.insert(*version); - } - acc - }); + // Every index-version reader needs the same Parquet file. + let referenced_ssts: HashSet<_> = in_tmp_ref.iter().map(|(file_id, _)| *file_id).collect(); let may_linger_files: HashMap> = may_linger_files .iter() @@ -976,7 +967,7 @@ impl LocalGcWorker { let should_delete = match file_type { FileType::Parquet => { let is_in_manifest = in_manifest.contains_key(&file_id); - let is_in_tmp_ref = in_tmp_ref.contains_key(&file_id); + let is_in_tmp_ref = referenced_ssts.contains(&file_id); let is_linger = may_linger_files.contains_key(&file_id); let is_eligible_for_delete = eligible_for_delete.contains_key(&file_id); @@ -991,15 +982,8 @@ impl LocalGcWorker { ) } FileType::Puffin(version) => { - // notice need to check both file id and version - let is_in_manifest = in_manifest - .get(&file_id) - .map(|opt_ver| *opt_ver == Some(version)) - .unwrap_or(false); - let is_in_tmp_ref = in_tmp_ref - .get(&file_id) - .map(|versions| versions.contains(&version)) - .unwrap_or(false); + let is_in_manifest = in_manifest.get(&file_id) == Some(&Some(version)); + let is_in_tmp_ref = in_tmp_ref.contains(&(file_id, Some(version))); let is_linger = may_linger_files .get(&file_id) .map(|files| files.contains(&&RemovedFile::Index(file_id, version))) @@ -1100,21 +1084,22 @@ 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) // 3. Have passed the lingering time let files_to_delete: Vec = eligible_for_removal .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)) + .filter(|file| { + let in_use = match file { + RemovedFile::File(file_id, _) => { + 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)) - || in_tmp_ref.contains(&(*file_id, Some(*index_version))) + RemovedFile::Index(file_id, version) => { + in_manifest.get(file_id) == Some(&Some(*version)) + || in_tmp_ref.contains(&(*file_id, Some(*version))) } }; !in_use diff --git a/src/mito2/src/gc/worker_test.rs b/src/mito2/src/gc/worker_test.rs index 3d389e7e1d7..f439e01a951 100644 --- a/src/mito2/src/gc/worker_test.rs +++ b/src/mito2/src/gc/worker_test.rs @@ -21,12 +21,13 @@ use async_trait::async_trait; use bytes::Bytes; use common_base::Plugins; use common_telemetry::init_default_ut_logging; +use common_time::Timestamp; use futures::TryStreamExt; use object_store::{Entry, ObjectStore, services}; use store_api::metadata::RegionMetadataRef; -use store_api::region_engine::RegionEngine as _; +use store_api::region_engine::{RegionEngine as _, RegionRole}; use store_api::region_request::{RegionCompactRequest, RegionRequest}; -use store_api::storage::{FileRef, FileRefsManifest, RegionId}; +use store_api::storage::{FileId, FileRef, FileRefsManifest, IndexVersion, RegionId}; use crate::access_layer::AccessLayerRef; use crate::config::MitoConfig; @@ -35,9 +36,11 @@ use crate::engine::compaction_test::{delete_and_flush, put_and_flush}; use crate::engine::region_hook::{RegionGcInfo, RegionHook, RegionHookRef}; use crate::error::UnexpectedSnafu; use crate::gc::{GcConfig, LocalGcWorker, should_delete_file}; -use crate::manifest::action::RemovedFile; +use crate::manifest::action::{RegionEdit, RegionMetaAction, RegionMetaActionList, RemovedFile}; use crate::metrics::GC_SKIPPED_UNPARSABLE_FILES; -use crate::region::MitoRegionRef; +use crate::region::{MitoRegionRef, RegionLeaderState}; +use crate::sst::file::{FileHandle, FileMeta, IndexType, RegionIndexId}; +use crate::sst::location; use crate::test_util::{ CreateRequestBuilder, TestEnv, build_rows, flush_region, put_rows, rows_schema, }; @@ -74,6 +77,296 @@ async fn create_gc_worker( .unwrap() } +struct IndexGcTestEnv { + engine: MitoEngine, + region: MitoRegionRef, + worker: LocalGcWorker, + meta: FileMeta, + entries: Vec, + removed_files: BTreeMap>, + _env: TestEnv, +} + +impl IndexGcTestEnv { + async fn new(full_file_listing: bool) -> Self { + let mut env = TestEnv::new().await; + let engine = env + .create_engine(MitoConfig { + gc: GcConfig { + enable: true, + lingering_time: Some(std::time::Duration::from_secs(60 * 60)), + ..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: FileId::random(), + available_indexes: smallvec::smallvec![IndexType::InvertedIndex], + ..Default::default() + }; + let layer = ®ion.access_layer; + let paths = [ + location::sst_file_path(layer.table_dir(), meta.file_id(), layer.path_type()), + location::index_file_path(layer.table_dir(), meta.index_id(), layer.path_type()), + location::index_file_path( + layer.table_dir(), + RegionIndexId::new(meta.file_id(), 1), + layer.path_type(), + ), + ]; + let mut entries = Vec::new(); + for path in paths { + entries.push(write_and_list_entry(layer.object_store(), &path).await); + } + let removed_files = BTreeMap::from([( + Timestamp::new_millisecond(0), + HashSet::from([ + RemovedFile::File(meta.file_id, Some(1)), + RemovedFile::Index(meta.file_id, 0), + RemovedFile::Index(meta.file_id, 1), + ]), + )]); + let worker = create_gc_worker( + &engine, + BTreeMap::from([(region_id, Some(region.clone()))]), + &FileRefsManifest::default(), + full_file_listing, + ) + .await; + Self { + engine, + region, + worker, + meta, + entries, + removed_files, + _env: env, + } + } +} + +#[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::version::SstVersion; + + let env = IndexGcTestEnv::new(full_file_listing).await; + let meta = &env.meta; + let region_id = meta.region_id; + let mut version = SstVersion::new(crate::test_util::memtable_util::metadata_for_test()); + version.add_files(env.region.file_purger.clone(), [meta.clone()].into_iter()); + let old = version.levels()[0].files[&meta.file_id].clone(); + let updated = FileMeta { + index_version: 1, + ..meta.clone() + }; + version.add_files(env.region.file_purger.clone(), [updated].into_iter()); + let new = version.levels()[0].files[&meta.file_id].clone(); + version.mark_all_deleted(); + drop(version); + + 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 = env + .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, 1], + 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<_> = env + .worker + .list_to_be_deleted_files( + region_id, + false, + &HashMap::new(), + &tmp_refs, + env.removed_files.clone(), + env.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(1) }), + RemovedFile::Index(meta.file_id, 0), + RemovedFile::Index(meta.file_id, 1), + ]), + }; + assert_eq!(deleted, expected); + } +} + +#[rstest::rstest] +#[case::manifest(true, Some(3))] +#[case::no_index_reader(false, None)] +#[case::old_index_reader(false, Some(0))] +#[case::other_index_reader(false, Some(3))] +#[tokio::test] +async fn test_gc_reference_protects_parquet_and_only_its_index( + #[values(false, true)] full_file_listing: bool, + #[case] in_manifest: bool, + #[case] index_version: Option, +) { + let env = IndexGcTestEnv::new(full_file_listing).await; + let file_id = env.meta.file_id; + let (manifest_refs, tmp_refs) = if in_manifest { + (HashMap::from([(file_id, index_version)]), HashSet::new()) + } else { + (HashMap::new(), HashSet::from([(file_id, index_version)])) + }; + let deleted = env + .worker + .list_to_be_deleted_files( + env.meta.region_id, + false, + &manifest_refs, + &tmp_refs, + env.removed_files.clone(), + env.entries.clone(), + ) + .await + .unwrap(); + let expected: HashSet<_> = [0, 1] + .into_iter() + .filter(|version| Some(*version) != index_version) + .map(|version| RemovedFile::Index(file_id, version)) + .collect(); + assert_eq!(deleted.into_iter().collect::>(), expected); +} + +#[rstest::rstest] +#[tokio::test] +async fn test_gc_reclaims_obsolete_index_after_readers_and_linger_release( + #[values(false, true)] full_file_listing: bool, + #[values(false, true)] linger_expired: bool, +) { + let env = IndexGcTestEnv::new(full_file_listing).await; + let region_id = env.meta.region_id; + env.engine + .set_region_role(region_id, RegionRole::Leader) + .unwrap(); + let old_reader = FileHandle::new(env.meta.clone(), env.region.file_purger.clone()); + let current_meta = FileMeta { + index_version: 1, + ..env.meta.clone() + }; + let current_reader = old_reader.with_updated_index(current_meta.clone()); + let removed_at = if linger_expired { + 0 + } else { + chrono::Utc::now().timestamp_millis() + }; + for meta in [env.meta.clone(), current_meta] { + env.region + .manifest_ctx + .update_manifest( + RegionLeaderState::Writable, + RegionMetaActionList::with_action(RegionMetaAction::Edit(RegionEdit { + files_to_add: vec![meta], + files_to_remove: vec![], + timestamp_ms: Some(removed_at), + compaction_time_window: None, + flushed_entry_id: None, + flushed_sequence: None, + committed_sequence: None, + })), + false, + ) + .await + .unwrap(); + } + + let layer = &env.region.access_layer; + let parquet = location::sst_file_path(layer.table_dir(), env.meta.file_id(), layer.path_type()); + let old_index = + location::index_file_path(layer.table_dir(), env.meta.index_id(), layer.path_type()); + let current_index = location::index_file_path( + layer.table_dir(), + current_reader.index_id(), + layer.path_type(), + ); + let mut old_reader = Some(old_reader); + for released in [false, true] { + if released { + drop(old_reader.take()); + } + let refs = env + .engine + .get_snapshot_of_file_refs([region_id], HashMap::new()) + .await + .unwrap(); + let worker = create_gc_worker( + &env.engine, + BTreeMap::from([(region_id, Some(env.region.clone()))]), + &refs, + full_file_listing, + ) + .await; + let report = worker.run().await.unwrap(); + let reclaimed = released && linger_expired; + assert!(report.deleted_files[®ion_id].is_empty()); + assert_eq!( + report.deleted_indexes[®ion_id], + if reclaimed { + vec![(env.meta.file_id, 0)] + } else { + vec![] + } + ); + assert!(layer.object_store().exists(&parquet).await.unwrap()); + assert!(layer.object_store().exists(¤t_index).await.unwrap()); + assert_eq!( + layer.object_store().exists(&old_index).await.unwrap(), + !reclaimed + ); + } + drop(current_reader); +} + #[tokio::test] async fn test_gc_worker_ignores_dropping_marker_file() { init_default_ut_logging(); diff --git a/src/mito2/src/region.rs b/src/mito2/src/region.rs index e6e3de2df1e..ef5b24f9de5 100644 --- a/src/mito2/src/region.rs +++ b/src/mito2/src/region.rs @@ -2363,10 +2363,8 @@ mod tests { struct CountingPurger(AtomicUsize); impl FilePurger for CountingPurger { - fn remove_file(&self, _: FileMeta, is_delete: bool, _: bool) { - if is_delete { - self.0.fetch_add(1, Ordering::SeqCst); - } + fn purge_file(&self, _: FileMeta) { + self.0.fetch_add(1, Ordering::SeqCst); } } diff --git a/src/mito2/src/sst/file.rs b/src/mito2/src/sst/file.rs index 28b6218cd7b..1d4caaf9127 100644 --- a/src/mito2/src/sst/file.rs +++ b/src/mito2/src/sst/file.rs @@ -39,6 +39,7 @@ use crate::cache::CacheManagerRef; use crate::cache::file_cache::{FileType, IndexKey}; use crate::region::options::IndexOptions; use crate::sst::file_purger::FilePurgerRef; +use crate::sst::file_ref::FileReferenceManagerRef; use crate::sst::index::fulltext_index::FULLTEXT_BLOOM_VERSION; use crate::sst::location; use crate::sst::parquet::SstInfo; @@ -551,13 +552,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 +577,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 +633,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 { @@ -634,23 +652,13 @@ impl FileHandle { .is_ok() } - pub fn index_outdated(&self) -> bool { - self.inner.index_outdated.load(Ordering::Relaxed) - } - - pub fn set_index_outdated(&self, index_outdated: bool) { - self.inner - .index_outdated - .store(index_outdated, Ordering::Relaxed); - } - /// Returns a reference to the [FileMeta]. pub fn meta_ref(&self) -> &FileMeta { &self.inner.meta } pub fn file_purger(&self) -> FilePurgerRef { - self.inner.file_purger.clone() + self.inner.lifetime.file_purger.clone() } pub fn size(&self) -> u64 { @@ -670,7 +678,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 +784,69 @@ 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.meta.clone(), - self.deleted.load(Ordering::Acquire), - self.index_outdated.load(Ordering::Acquire), - ); + if let Some(manager) = &self.lifetime.file_ref_manager { + manager.remove_file(&self.meta); + } } } 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 file_ref_manager = file_purger.file_ref_manager(); + let lifetime = Arc::new(SstLifetime { + meta: Mutex::new(meta.clone()), + deleted: AtomicBool::new(false), + file_purger, + file_ref_manager, + }); + Self::with_lifetime(meta, lifetime, primary_key_range) + } + + fn with_lifetime( + meta: FileMeta, + lifetime: Arc, + primary_key_range: Option<(Bytes, Bytes)>, + ) -> Self { + if let Some(manager) = &lifetime.file_ref_manager { + manager.register_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, + /// Tracks each metadata view's readers independently for object-store index GC. + file_ref_manager: Option, +} + +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..727c6c1cc9e 100644 --- a/src/mito2/src/sst/file_purger.rs +++ b/src/mito2/src/sst/file_purger.rs @@ -23,23 +23,20 @@ use crate::access_layer::AccessLayerRef; use crate::cache::CacheManagerRef; use crate::error::Result; use crate::schedule::scheduler::SchedulerRef; -use crate::sst::file::{FileMeta, delete_files, delete_index}; +use crate::sst::file::{FileMeta, delete_files}; use crate::sst::file_ref::FileReferenceManagerRef; 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); + /// Purges a retired SST and its indexes after all metadata views release their readers. + /// Each metadata view has already released its reader reference. + 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 - /// The default implementation is a no-op. - fn new_file(&self, _: &FileMeta) { - // noop + /// Returns the object-store reference manager used by SST metadata views. + /// Local and no-op purgers do not track GC references. + fn file_ref_manager(&self) -> Option { + None } } @@ -50,9 +47,7 @@ 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 purge_file(&self, _file_meta: FileMeta) {} } /// Purger that purges file for current region. @@ -182,33 +177,16 @@ impl LocalFilePurger { error!(e; "Failed to schedule the file purge request"); } } - - fn delete_index(&self, file_meta: FileMeta) { - let sst_layer = self.sst_layer.clone(); - let cache_manager = self.cache_manager.clone(); - if let Err(e) = self.scheduler.schedule(Box::pin(async move { - let index_id = file_meta.index_id(); - if let Err(e) = delete_index(index_id, &sst_layer, &cache_manager).await { - error!(e; "Failed to delete index for file {:?} from storage", file_meta); - } - })) { - error!(e; "Failed to schedule the index purge request"); - } - } } 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 { - 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); } } @@ -246,31 +224,26 @@ 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. - 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) { - self.file_ref_manager.add_file(file_meta); + fn file_ref_manager(&self) -> Option { + Some(self.file_ref_manager.clone()) } } #[cfg(test)] mod tests { use std::num::NonZeroU64; + use std::sync::Barrier; + use std::thread; - use common_test_util::temp_dir::create_temp_dir; + use common_test_util::temp_dir::{TempDir, create_temp_dir}; use object_store::ObjectStore; use object_store::services::Fs; use smallvec::SmallVec; @@ -288,6 +261,133 @@ mod tests { use crate::sst::index::puffin_manager::PuffinManagerFactory; use crate::sst::location; + struct PurgeTestEnv { + layer: AccessLayerRef, + scheduler: Arc, + file_id: RegionFileId, + _dir: TempDir, + } + + impl PurgeTestEnv { + async fn new() -> Self { + common_telemetry::init_default_ut_logging(); + let dir = create_temp_dir("file-purge"); + let index_aux_path = dir.path().join("index_aux"); + let puffin_mgr = PuffinManagerFactory::new(&index_aux_path, 4096, None, None) + .await + .unwrap(); + let intm_mgr = IntermediateManager::init_fs(index_aux_path.to_str().unwrap()) + .await + .unwrap(); + let store = + ObjectStore::new(Fs::default().root(&dir.path().display().to_string())).unwrap(); + let env = Self { + layer: Arc::new(AccessLayer::new( + "table1", + PathType::Bare, + store, + puffin_mgr, + intm_mgr, + )), + scheduler: Arc::new(LocalScheduler::new(1)), + file_id: RegionFileId::new(RegionId::new(0, 0), FileId::random()), + _dir: dir, + }; + env.layer + .object_store() + .write(&env.sst_path(), vec![0; 4096]) + .await + .unwrap(); + env + } + + fn sst_path(&self) -> String { + location::sst_file_path(self.layer.table_dir(), self.file_id, self.layer.path_type()) + } + + fn index_path(&self, version: u64) -> String { + location::index_file_path( + self.layer.table_dir(), + RegionIndexId::new(self.file_id, version), + self.layer.path_type(), + ) + } + + async fn wait_for_purges(&self) { + // The single worker finishes all earlier purge jobs before this barrier. + let (tx, rx) = tokio::sync::oneshot::channel(); + self.scheduler + .schedule(Box::pin(async move { + tx.send(()).unwrap(); + })) + .unwrap(); + rx.await.unwrap(); + } + } + + #[rstest::rstest] + #[tokio::test] + async fn test_concurrent_index_view_release_keeps_live_files_and_purges_retired_files( + #[values(false, true)] retired: bool, + ) { + let env = PurgeTestEnv::new().await; + let store = env.layer.object_store(); + let purger = Arc::new(LocalFilePurger::new( + env.scheduler.clone(), + env.layer.clone(), + None, + )); + let meta = FileMeta { + region_id: env.file_id.region_id(), + file_id: env.file_id.file_id(), + ..Default::default() + }; + let old_reader = FileHandle::new(meta.clone(), purger); + let mut readers = vec![old_reader]; + for index_version in 1..=2 { + store + .write(&env.index_path(index_version), "index") + .await + .unwrap(); + readers.push(readers.last().unwrap().with_updated_index(FileMeta { + index_version, + available_indexes: smallvec::smallvec![IndexType::InvertedIndex], + ..meta.clone() + })); + } + if retired { + readers.last().unwrap().mark_deleted(); + } + // Dropping one view must not remove files still needed by the other views. + drop(readers.remove(1)); + env.wait_for_purges().await; + assert!(store.exists(&env.sst_path()).await.unwrap()); + for version in 1..=2 { + assert!(store.exists(&env.index_path(version)).await.unwrap()); + } + + // Regression coverage for final releases on different threads. The start barrier + // encourages overlap; it does not exhaustively control destructor interleavings. + let barrier = Barrier::new(readers.len()); + thread::scope(|scope| { + for reader in readers { + let barrier = &barrier; + scope.spawn(move || { + barrier.wait(); + drop(reader); + }); + } + }); + env.scheduler.stop(true).await.unwrap(); + assert_eq!(store.exists(&env.sst_path()).await.unwrap(), !retired); + for version in 1..=2 { + assert_eq!( + store.exists(&env.index_path(version)).await.unwrap(), + !retired + ); + } + } + #[rstest::rstest] #[case(false)] #[case(true)] @@ -295,36 +395,13 @@ 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(); - - let dir = create_temp_dir("file-purge"); - let dir_path = dir.path().display().to_string(); - let builder = Fs::default().root(&dir_path); - let sst_file_id = RegionFileId::new(RegionId::new(0, 0), FileId::random()); - let sst_dir = "table1"; - - let index_aux_path = dir.path().join("index_aux"); - let puffin_mgr = PuffinManagerFactory::new(&index_aux_path, 4096, None, None) - .await - .unwrap(); - let intm_mgr = IntermediateManager::init_fs(index_aux_path.to_str().unwrap()) - .await - .unwrap(); - - let object_store = ObjectStore::new(builder).unwrap(); - - let layer = Arc::new(AccessLayer::new( - sst_dir, - PathType::Bare, - object_store.clone(), - puffin_mgr, - intm_mgr, - )); - 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 env = PurgeTestEnv::new().await; + let object_store = env.layer.object_store(); + let path = env.sst_path(); + let sst_file_id = env.file_id; + let scheduler = &env.scheduler; let index_store = ObjectStore::new(object_store::services::Memory::default()).unwrap(); let owner = RegionId::new(9, 1); @@ -334,40 +411,48 @@ mod tests { gc_enabled, PathType::Bare, scheduler.clone(), - layer, + env.layer.clone(), None, Arc::new(crate::sst::file_ref::FileReferenceManager::new(None)), Some(RangeIndexDeleter::new(index_store.clone(), owner)), ); { - 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 = env.index_path(index_version); + 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); + env.wait_for_purges().await; + 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 = env.index_path(index_version); + assert!(object_store.exists(&path).await.unwrap()); + } + } drop(reader); } @@ -378,48 +463,35 @@ 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 = env.index_path(index_version); + assert_eq!( + object_store.exists(&path).await.unwrap(), + gc_enabled || !is_delete + ); + } + } } #[tokio::test] async fn test_file_purge_with_index() { - common_telemetry::init_default_ut_logging(); - - let dir = create_temp_dir("file-purge"); - let dir_path = dir.path().display().to_string(); - let builder = Fs::default().root(&dir_path); - let sst_file_id = RegionFileId::new(RegionId::new(0, 0), FileId::random()); - let index_file_id = RegionIndexId::new(sst_file_id, 0); - let sst_dir = "table1"; - - let index_aux_path = dir.path().join("index_aux"); - let puffin_mgr = PuffinManagerFactory::new(&index_aux_path, 4096, None, None) - .await - .unwrap(); - let intm_mgr = IntermediateManager::init_fs(index_aux_path.to_str().unwrap()) - .await - .unwrap(); - - let object_store = ObjectStore::new(builder).unwrap(); - - let layer = Arc::new(AccessLayer::new( - sst_dir, - PathType::Bare, - object_store.clone(), - puffin_mgr, - intm_mgr, - )); - let path = location::sst_file_path(sst_dir, sst_file_id, layer.path_type()); - object_store.write(&path, vec![0; 4096]).await.unwrap(); - - let index_path = location::index_file_path(sst_dir, index_file_id, layer.path_type()); + let env = PurgeTestEnv::new().await; + let object_store = env.layer.object_store(); + let sst_file_id = env.file_id; + let path = env.sst_path(); + let index_path = env.index_path(0); object_store .write(&index_path, vec![0; 4096]) .await .unwrap(); - let scheduler = Arc::new(LocalScheduler::new(3)); - - let file_purger = Arc::new(LocalFilePurger::new(scheduler.clone(), layer, None)); + let scheduler = &env.scheduler; + let file_purger = Arc::new(LocalFilePurger::new( + scheduler.clone(), + env.layer.clone(), + None, + )); { let handle = FileHandle::new( diff --git a/src/mito2/src/sst/file_ref.rs b/src/mito2/src/sst/file_ref.rs index 4dee84ca1fa..68b0facaab4 100644 --- a/src/mito2/src/sst/file_ref.rs +++ b/src/mito2/src/sst/file_ref.rs @@ -24,23 +24,18 @@ use crate::metrics::GC_REF_FILE_CNT; use crate::region::MitoRegionRef; use crate::sst::file::FileMeta; -/// File references for a region. -/// It contains all files referenced by the region. +/// File references grouped by their physical source region. #[derive(Debug, Clone, Default)] -pub struct RegionFileRefs { - /// (FileRef, Ref Count) meaning how many FileHandleInner is opened for this file. - pub files: HashMap, +struct RegionFileRefs { + /// Counts independent metadata views of each physical file and index version. + files: HashMap, } -/// Manages all file references in one datanode. -/// It keeps track of which files are referenced and group by table ids. -/// This is useful for ensuring that files are not deleted while they are still in use by any -/// query. +/// Tracks reader references to SST metadata views in one datanode for object-store GC. #[derive(Debug)] pub struct FileReferenceManager { /// Datanode id. used to determine tmp ref file name. node_id: Option, - /// TODO(discord9): use no hash hasher since table id is sequential. files_per_region: DashMap, /// Whether global GC is enabled. /// This is only meaningful when using object store (not local filesystem). @@ -68,58 +63,49 @@ impl FileReferenceManager { } /// Returns whether global GC is enabled. - /// - /// This is useful for determining the file deletion strategy: - /// - If GC is enabled (and using object store), files are tracked but not immediately deleted - /// - If GC is disabled (or using local filesystem), files are deleted immediately pub fn is_gc_enabled(&self) -> bool { self.gc_enabled } fn ref_file_set(&self, region_id: RegionId) -> Option> { - // region id not found. - let file_refs = self.files_per_region.get(®ion_id)?.clone(); - - if file_refs.files.is_empty() { - // still return an empty manifest to indicate no files are referenced. - // and differentiate from error case where table_id not found. - return Some(HashSet::new()); - } - - let ref_file_set: HashSet = file_refs.files.keys().cloned().collect(); - + let region_refs = self.files_per_region.get(®ion_id)?.clone(); + let refs: HashSet<_> = region_refs.files.keys().cloned().collect(); debug!( "Get file refs for region {}, node {:?}, {} files", region_id, self.node_id, - ref_file_set.len(), + region_refs.files.len(), ); - - Some(ref_file_set) + Some(refs) } - /// Gets all ref files for the given regions, meaning all open FileHandles for those regions - /// and from related regions' manifests. - /// `query_regions_for_mem` queries for in memory file handles. - /// `related_regions_in_manifest` queries for related regions' manifests to get more file refs of given region ids. + /// Gets references from live SST metadata views and related regions' manifests. pub(crate) async fn get_snapshot_of_file_refs( &self, query_regions_for_mem: Vec, dst_region_to_src_regions: Vec<(MitoRegionRef, HashSet)>, ) -> Result { + // Imported SSTs are tracked under their source region even if that region is + // not open on this node. Capture their old readers before reading manifests. + let region_ids: HashSet<_> = query_regions_for_mem + .iter() + .map(|r| r.region_id()) + .chain( + dst_region_to_src_regions + .iter() + .flat_map(|(_, sources)| sources.iter().copied()), + ) + .collect(); let mut ref_files = HashMap::new(); - // get from in memory file handles - for region_id in query_regions_for_mem.iter().map(|r| r.region_id()) { + for region_id in region_ids { if let Some(files) = self.ref_file_set(region_id) { ref_files.insert(region_id, files); } } let mut manifest_version = HashMap::new(); - let mut cross_region_refs = HashMap::new(); - // get file refs from related regions' manifests for (dst_region, src_regions) in &dst_region_to_src_regions { let manifest = dst_region.manifest_ctx.manifest().await; for meta in manifest.files.values() { @@ -128,9 +114,8 @@ impl FileReferenceManager { .entry(meta.region_id) .or_insert_with(HashSet::new) .insert(dst_region.region_id()); - // since gc couldn't happen together with repartition - // (both the queries and related_region acquire region read lock), no need to worry about - // staging manifest in repartition here. + // GC and repartition hold region read locks, so staging manifests + // cannot change ownership while these references are collected. ref_files .entry(meta.region_id) .or_insert_with(HashSet::new) @@ -141,23 +126,24 @@ impl FileReferenceManager { )); } } - // not sure if related region's manifest version is needed, but record it for now. manifest_version.insert(dst_region.region_id(), manifest.manifest_version); } 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. + // The manifest only protects its current index generation; older readers + // still need their exact-version 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)) - .collect(); + refs.retain(|f| { + !manifest + .files + .get(&f.file_id) + .is_some_and(|meta| meta.index_version() == f.index_version) + }); }); manifest_version.insert(r.region_id(), manifest.manifest_version); } - // simply return all ref files, no manifest version filtering for now. Ok(FileRefsManifest { file_refs: ref_files, manifest_version, @@ -165,77 +151,44 @@ impl FileReferenceManager { }) } - /// Adds a new file reference. - /// Also records the access layer for the table if not exists. - /// The access layer will be used to upload ref file to object storage. - pub fn add_file(&self, file_meta: &FileMeta) { - let region_id = file_meta.region_id; - let mut is_new = false; - { - let file_ref = FileRef::new( - file_meta.region_id, - file_meta.file_id, - file_meta.index_version(), - ); - self.files_per_region - .entry(region_id) - .and_modify(|refs| { - refs.files - .entry(file_ref.clone()) - .and_modify(|count| *count += 1) - .or_insert_with(|| { - is_new = true; - 1 - }); - }) - .or_insert_with(|| RegionFileRefs { - files: HashMap::from_iter([(file_ref, 1)]), - }); - } - if is_new { + /// Registers a metadata view, paired with `remove_file` when its last reader leaves. + pub(crate) fn register_file(&self, file_meta: &FileMeta) { + let mut region = self + .files_per_region + .entry(file_meta.region_id) + .or_default(); + let file_ref = FileRef::new( + file_meta.region_id, + file_meta.file_id, + file_meta.index_version(), + ); + let count = region.files.entry(file_ref).or_insert_with(|| { GC_REF_FILE_CNT.inc(); - } + 0 + }); + *count += 1; } - /// Removes a file reference. - /// If the reference count reaches zero, the file reference will be removed from the manager. - pub fn remove_file(&self, file_meta: &FileMeta) { - let region_id = file_meta.region_id; - let file_ref = FileRef::new(region_id, file_meta.file_id, file_meta.index_version()); - - let mut remove_table_entry = false; - let mut remove_file_ref = false; - let mut file_cnt = 0; - - let region_ref = self.files_per_region.entry(region_id).and_modify(|refs| { - let entry = refs.files.entry(file_ref.clone()).and_modify(|count| { - if *count > 0 { - *count -= 1; - } - if *count == 0 { - remove_file_ref = true; - } - }); - if let std::collections::hash_map::Entry::Occupied(o) = entry - && remove_file_ref + /// Releases a metadata view's reference, regardless of whether the SST was retired. + pub(crate) fn remove_file(&self, file_meta: &FileMeta) { + let file_ref = FileRef::new( + file_meta.region_id, + file_meta.file_id, + file_meta.index_version(), + ); + if let Entry::Occupied(mut region) = self.files_per_region.entry(file_meta.region_id) { + if let std::collections::hash_map::Entry::Occupied(mut file) = + region.get_mut().files.entry(file_ref) { - o.remove_entry(); + *file.get_mut() -= 1; + if *file.get() == 0 { + file.remove(); + GC_REF_FILE_CNT.dec(); + } } - - file_cnt = refs.files.len(); - - if refs.files.is_empty() { - remove_table_entry = true; + if region.get().files.is_empty() { + region.remove(); } - }); - - if let Entry::Occupied(o) = region_ref - && remove_table_entry - { - o.remove_entry(); - } - if remove_file_ref { - GC_REF_FILE_CNT.dec(); } } @@ -246,135 +199,42 @@ impl FileReferenceManager { #[cfg(test)] mod tests { - use std::num::NonZeroU64; - - use smallvec::SmallVec; - use store_api::storage::{FileId, RegionId}; + use store_api::storage::FileId; use super::*; - use crate::sst::file::{ColumnIndexMetadata, FileMeta, FileTimeRange, IndexType, RegionFileId}; + use crate::sst::file::IndexType; - #[tokio::test] - async fn test_file_ref_mgr() { - common_telemetry::init_default_ut_logging(); - - let sst_file_id = RegionFileId::new(RegionId::new(0, 0), FileId::random()); - - let file_ref_mgr = FileReferenceManager::new(None); - - let file_meta = 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: SmallVec::from_iter([IndexType::InvertedIndex]), - indexes: vec![ColumnIndexMetadata { - column_id: 0, - created_indexes: SmallVec::from_iter([IndexType::InvertedIndex]), - ..Default::default() - }], - index_file_size: 4096, - index_version: 0, - num_rows: 1024, - num_row_groups: 1, - sequence: NonZeroU64::new(4096), - partition_expr: None, - num_series: 0, + #[test] + fn test_index_view_references_release_generations_independently() { + let manager = FileReferenceManager::new(None); + let meta = FileMeta { + region_id: RegionId::new(1, 1), + file_id: FileId::random(), + available_indexes: smallvec::smallvec![IndexType::InvertedIndex], ..Default::default() }; + let updated = FileMeta { + index_version: 2, + ..meta.clone() + }; + manager.register_file(&meta); + manager.register_file(&meta); + manager.register_file(&updated); + let expected: HashSet<_> = [0, 2] + .into_iter() + .map(|version| FileRef::new(meta.region_id, meta.file_id, Some(version))) + .collect(); + assert_eq!(manager.ref_file_set(meta.region_id).unwrap(), expected); - file_ref_mgr.add_file(&file_meta); - + // Independent readers of the same index version must both release it. + manager.remove_file(&meta); + assert_eq!(manager.ref_file_set(meta.region_id).unwrap(), expected); + manager.remove_file(&meta); assert_eq!( - file_ref_mgr - .files_per_region - .get(&file_meta.region_id) - .unwrap() - .files, - HashMap::from_iter([( - FileRef::new( - file_meta.region_id, - file_meta.file_id, - file_meta.index_version() - ), - 1 - )]) - ); - - file_ref_mgr.add_file(&file_meta); - - let expected_region_ref_manifest = HashSet::from_iter([FileRef::new( - file_meta.region_id, - file_meta.file_id, - file_meta.index_version(), - )]); - - assert_eq!( - file_ref_mgr.ref_file_set(file_meta.region_id).unwrap(), - expected_region_ref_manifest - ); - - assert_eq!( - file_ref_mgr - .files_per_region - .get(&file_meta.region_id) - .unwrap() - .files, - HashMap::from_iter([( - FileRef::new( - file_meta.region_id, - file_meta.file_id, - file_meta.index_version() - ), - 2 - )]) - ); - - assert_eq!( - file_ref_mgr.ref_file_set(file_meta.region_id).unwrap(), - expected_region_ref_manifest - ); - - file_ref_mgr.remove_file(&file_meta); - - assert_eq!( - file_ref_mgr - .files_per_region - .get(&file_meta.region_id) - .unwrap() - .files, - HashMap::from_iter([( - FileRef::new( - file_meta.region_id, - file_meta.file_id, - file_meta.index_version() - ), - 1 - )]) - ); - - assert_eq!( - file_ref_mgr.ref_file_set(file_meta.region_id).unwrap(), - expected_region_ref_manifest - ); - - file_ref_mgr.remove_file(&file_meta); - - assert!( - file_ref_mgr - .files_per_region - .get(&file_meta.region_id) - .is_none(), - "{:?}", - file_ref_mgr.files_per_region - ); - - assert!( - file_ref_mgr.ref_file_set(file_meta.region_id).is_none(), - "{:?}", - file_ref_mgr.files_per_region + manager.ref_file_set(meta.region_id).unwrap(), + HashSet::from([FileRef::new(meta.region_id, meta.file_id, Some(2))]) ); + manager.remove_file(&updated); + assert!(manager.ref_file_set(meta.region_id).is_none()); } } 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();