rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-10-28 22:56:19 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(mito): preserve SST lifetime across index replacement (#9950)

Browse Source 
* 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>

* refactor(mito): list-based SST purge and single FilePurger::remove_file

The final purge of a retired SST no longer depends on FileMeta: index
files of every version are discovered by listing
({region_dir}/index/{file_id}. prefix), so even when the last live view
is a stale one, the purge still removes all newer index files.

- SstLifetime is slimmed down to { deleted, file_purger }; the purge now
  runs in the last FileHandleInner::drop (Arc::strong_count == 1, which
  is race-free here since only inners of the same file hold the lifetime).
- FilePurger is merged back into a single remove_file(file_meta,
  is_delete) method; is_delete means the file was retired AND this is the
  last live view.
- LocalFilePurger deletes the SST and all its index files via the new
  delete_file_and_all_indexes helper (list before delete, so a listing
  failure leaves the file untouched).
- Remove the dead index_outdated flag and its accessors, which had no
  callers in the entire repo.

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

* fix(mito): elect final SST view release atomically

Track metadata views with an atomic counter instead of inspecting the lifetime Arc strong count. Concurrent destructors now elect exactly one final releaser while preserving the single remove_file interface.

Add a deterministic concurrent-drop regression covering retired and live files.

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

* refactor(mito): restore Arc-owned SST cleanup and clarify reference release

Restore the lifetime design from 835227dcbc, using the final Arc drop and latest metadata instead of manual view counting and per-SST index listing.

Rename FilePurger::remove_file to release_file_ref while retaining purge_file for final cleanup. Preserve and adapt the concurrent view-release regression test.

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

* refactor(mito): track SST references by file lifetime

Register one reference guard per logical SST lifetime and release it on final drop, including normal region close. Remove per-view release_file_ref and index_outdated handling.

Protect all index generations while their SST is referenced. Preserve the existing GC snapshot format by expanding the recorded index generation upper bound for older version-aware workers.

Update coverage for independent references, normal close, cross-region ownership, concurrent final cleanup and both GC listing modes.

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

* refactor(mito): release SST references directly on lifetime drop

Remove the redundant FileReferenceGuard and let SstLifetime register and unregister its file reference directly. Preserve unconditional reference release on normal close and retirement-only purge.

Expose the reference manager through FilePurger and adapt reference manager tests without changing GC or snapshot behavior.

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

* test(mito): verify SST cleanup with real storage and isolate GC policy cases

Replace the recording-only concurrent purge test with filesystem assertions using LocalFilePurger and shared fixtures. Keep intermediate-view retention and retired versus live file cleanup coverage without claiming exhaustive thread interleavings.

Separate GC reference-policy cases from reader-release ordering and reuse their setup. Production code remains unchanged.

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

* style(mito): format SST lifetime changes

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

* fix(mito): reclaim obsolete indexes by reader version

Track exact index-version references per metadata view and restore version-specific Puffin liveness in both GC paths. Keep file-level Parquet protection and shared SST lifetime cleanup, without expanding snapshots into unreferenced index generations.

Include old readers of imported files even when the source region is not open locally. Add real-storage GC coverage for reader release and lingering while retaining Parquet and the current index.

Validated workspace checks with default and all features, all-target all-feature Clippy with warnings denied, and all 1853 mito2 tests.

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

---------

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>
This commit is contained in:
Lei, HUANG
2026-10-27 10:20:45 +00:00
committed by GitHub
parent 2e84833ac8
commit 17ed988f49
9 changed files with 927 additions and 449 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/cmd/src/datanode/objbench.rs
+2 -7
View File
@@ -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
+201
View File
@@ -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[&region_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[&region_id], expected);
drop(old_reader);
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
+14 -29
View File
@@ -929,17 +929,8 @@ impl LocalGcWorker {
unknown_file_may_linger_until: chrono::DateTime<chrono::Utc>,
) -> Vec<RemovedFile> {
let mut ready_for_delete = vec![];
// all group by file id for easier checking
let in_tmp_ref: HashMap<FileId, HashSet<IndexVersion>> =
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<FileId, HashSet<&RemovedFile>> = 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<RemovedFile> = 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
src/mito2/src/gc/worker_test.rs
+297 -4
View File
@@ -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<Entry>,
removed_files: BTreeMap<Timestamp, HashSet<RemovedFile>>,
_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 = &region.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(&region_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<IndexVersion>,
) {
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::<HashSet<_>>(), 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[&region_id].is_empty());
assert_eq!(
report.deleted_indexes[&region_id],
if reclaimed {
vec![(env.meta.file_id, 0)]
} else {
vec![]
}
);
assert!(layer.object_store().exists(&parquet).await.unwrap());
assert!(layer.object_store().exists(&current_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();
src/mito2/src/region.rs
+2 -4
View File
@@ -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);
}
}
src/mito2/src/sst/file.rs
+68 -30
View File
@@ -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<PrimaryKeyRange>,
file_purger: FilePurgerRef,
lifetime: Arc<SstLifetime>,
}
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<SstLifetime>,
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<FileMeta>,
deleted: AtomicBool,
file_purger: FilePurgerRef,
/// Tracks each metadata view's readers independently for object-store index GC.
file_ref_manager: Option<FileReferenceManagerRef>,
}
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
+213 -141
View File
@@ -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<FileReferenceManagerRef> {
None
}
}
@@ -50,9 +47,7 @@ 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 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<FileReferenceManagerRef> {
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<LocalScheduler>,
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(
src/mito2/src/sst/file_ref.rs
+93 -233
View File
@@ -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<FileRef, usize>,
struct RegionFileRefs {
/// Counts independent metadata views of each physical file and index version.
files: HashMap<FileRef, usize>,
}
/// 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<u64>,
/// TODO(discord9): use no hash hasher since table id is sequential.
files_per_region: DashMap<RegionId, RegionFileRefs>,
/// 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<HashSet<FileRef>> {
// region id not found.
let file_refs = self.files_per_region.get(&region_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<FileRef> = file_refs.files.keys().cloned().collect();
let region_refs = self.files_per_region.get(&region_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<MitoRegionRef>,
dst_region_to_src_regions: Vec<(MitoRegionRef, HashSet<RegionId>)>,
) -> Result<FileRefsManifest> {
// 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());
}
}
src/mito2/src/sst/version.rs
+37 -1
View File
@@ -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();