mirror of
https://github.com/GreptimeTeam/greptimedb.git
synced 2026-10-29 15:06:59 +00:00
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>
This commit is contained in:
@@ -27,7 +27,7 @@ use mito2::config::{FulltextIndexConfig, MitoConfig, Mode};
|
|||||||
use mito2::read::FlatSource;
|
use mito2::read::FlatSource;
|
||||||
use mito2::sst::FormatType;
|
use mito2::sst::FormatType;
|
||||||
use mito2::sst::file::{FileHandle, FileMeta};
|
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::intermediate::IntermediateManager;
|
||||||
use mito2::sst::index::puffin_manager::PuffinManagerFactory;
|
use mito2::sst::index::puffin_manager::PuffinManagerFactory;
|
||||||
use mito2::sst::parquet::WriteOptions;
|
use mito2::sst::parquet::WriteOptions;
|
||||||
@@ -508,12 +508,7 @@ async fn build_cache_manager(
|
|||||||
}
|
}
|
||||||
|
|
||||||
fn new_noop_file_purger() -> FilePurgerRef {
|
fn new_noop_file_purger() -> FilePurgerRef {
|
||||||
#[derive(Debug)]
|
Arc::new(NoopFilePurger)
|
||||||
struct Noop;
|
|
||||||
impl FilePurger for Noop {
|
|
||||||
fn remove_file(&self, _file_meta: FileMeta, _is_delete: bool, _index_outdated: bool) {}
|
|
||||||
}
|
|
||||||
Arc::new(Noop)
|
|
||||||
}
|
}
|
||||||
|
|
||||||
async fn load_parquet_metadata(
|
async fn load_parquet_metadata(
|
||||||
|
|||||||
@@ -53,3 +53,98 @@ async fn get_snapshot_of_file_refs_skips_missing_file_handle_region() {
|
|||||||
assert!(manifest.manifest_version.is_empty());
|
assert!(manifest.manifest_version.is_empty());
|
||||||
assert!(manifest.cross_region_refs.is_empty());
|
assert!(manifest.cross_region_refs.is_empty());
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn get_snapshot_preserves_old_index_readers_until_release() {
|
||||||
|
use store_api::region_engine::{RegionEngine, RegionRole};
|
||||||
|
use store_api::region_request::RegionRequest;
|
||||||
|
use store_api::storage::{FileId, FileRef};
|
||||||
|
|
||||||
|
use crate::manifest::action::{RegionEdit, RegionMetaAction, RegionMetaActionList};
|
||||||
|
use crate::region::RegionLeaderState;
|
||||||
|
use crate::sst::file::{FileMeta, IndexType};
|
||||||
|
use crate::sst::version::SstVersion;
|
||||||
|
use crate::test_util::CreateRequestBuilder;
|
||||||
|
|
||||||
|
let mut env = TestEnv::new().await;
|
||||||
|
let engine = env
|
||||||
|
.create_engine(MitoConfig {
|
||||||
|
gc: crate::gc::GcConfig {
|
||||||
|
enable: true,
|
||||||
|
..Default::default()
|
||||||
|
},
|
||||||
|
..Default::default()
|
||||||
|
})
|
||||||
|
.await;
|
||||||
|
let region_id = RegionId::new(1, 1);
|
||||||
|
engine
|
||||||
|
.handle_request(
|
||||||
|
region_id,
|
||||||
|
RegionRequest::Create(CreateRequestBuilder::new().build()),
|
||||||
|
)
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
engine
|
||||||
|
.set_region_role(region_id, RegionRole::Leader)
|
||||||
|
.unwrap();
|
||||||
|
let region = engine.get_region(region_id).unwrap();
|
||||||
|
let meta = FileMeta {
|
||||||
|
region_id,
|
||||||
|
file_id: FileId::random(),
|
||||||
|
available_indexes: smallvec::smallvec![IndexType::InvertedIndex],
|
||||||
|
..Default::default()
|
||||||
|
};
|
||||||
|
let mut version = SstVersion::new(crate::test_util::memtable_util::metadata_for_test());
|
||||||
|
version.add_files(region.file_purger.clone(), [meta.clone()].into_iter());
|
||||||
|
let old_reader = version.levels()[0].files[&meta.file_id].clone();
|
||||||
|
let current_meta = FileMeta {
|
||||||
|
index_version: 1,
|
||||||
|
..meta.clone()
|
||||||
|
};
|
||||||
|
version.add_files(
|
||||||
|
region.file_purger.clone(),
|
||||||
|
[current_meta.clone()].into_iter(),
|
||||||
|
);
|
||||||
|
region
|
||||||
|
.manifest_ctx
|
||||||
|
.update_manifest(
|
||||||
|
RegionLeaderState::Writable,
|
||||||
|
RegionMetaActionList::with_action(RegionMetaAction::Edit(RegionEdit {
|
||||||
|
files_to_add: vec![current_meta],
|
||||||
|
files_to_remove: vec![],
|
||||||
|
timestamp_ms: None,
|
||||||
|
compaction_time_window: None,
|
||||||
|
flushed_entry_id: None,
|
||||||
|
flushed_sequence: None,
|
||||||
|
committed_sequence: None,
|
||||||
|
})),
|
||||||
|
false,
|
||||||
|
)
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
let snapshot = engine
|
||||||
|
.get_snapshot_of_file_refs([region_id], HashMap::new())
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
assert_eq!(
|
||||||
|
snapshot.file_refs[®ion_id],
|
||||||
|
HashSet::from([FileRef::new(region_id, meta.file_id, Some(0))])
|
||||||
|
);
|
||||||
|
drop(old_reader);
|
||||||
|
let snapshot = engine
|
||||||
|
.get_snapshot_of_file_refs([region_id], HashMap::new())
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
assert!(snapshot.file_refs[®ion_id].is_empty());
|
||||||
|
drop(version);
|
||||||
|
let snapshot = engine
|
||||||
|
.get_snapshot_of_file_refs([region_id], HashMap::new())
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
assert!(
|
||||||
|
snapshot
|
||||||
|
.file_refs
|
||||||
|
.get(®ion_id)
|
||||||
|
.is_none_or(HashSet::is_empty)
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|||||||
+5
-3
@@ -1091,6 +1091,8 @@ impl LocalGcWorker {
|
|||||||
// When full_file_listing is false, skip expensive list operations and only delete
|
// When full_file_listing is false, skip expensive list operations and only delete
|
||||||
// files that are tracked in recently_removed_files
|
// files that are tracked in recently_removed_files
|
||||||
if !self.full_file_listing {
|
if !self.full_file_listing {
|
||||||
|
let referenced_ssts: HashSet<_> =
|
||||||
|
in_tmp_ref.iter().map(|(file_id, _)| *file_id).collect();
|
||||||
// Only delete files that:
|
// Only delete files that:
|
||||||
// 1. Are in recently_removed_files (tracked in manifest)
|
// 1. Are in recently_removed_files (tracked in manifest)
|
||||||
// 2. Are not in use(in manifest or tmp ref)
|
// 2. Are not in use(in manifest or tmp ref)
|
||||||
@@ -1099,9 +1101,9 @@ impl LocalGcWorker {
|
|||||||
.iter()
|
.iter()
|
||||||
.filter(|file_id| {
|
.filter(|file_id| {
|
||||||
let in_use = match file_id {
|
let in_use = match file_id {
|
||||||
RemovedFile::File(file_id, index_version) => {
|
RemovedFile::File(file_id, _) => {
|
||||||
in_manifest.get(file_id) == Some(index_version)
|
// Any index-version reader still needs the same Parquet file.
|
||||||
|| in_tmp_ref.contains(&(*file_id, *index_version))
|
in_manifest.contains_key(file_id) || referenced_ssts.contains(file_id)
|
||||||
}
|
}
|
||||||
RemovedFile::Index(file_id, index_version) => {
|
RemovedFile::Index(file_id, index_version) => {
|
||||||
in_manifest.get(file_id) == Some(&Some(*index_version))
|
in_manifest.get(file_id) == Some(&Some(*index_version))
|
||||||
|
|||||||
@@ -74,6 +74,146 @@ async fn create_gc_worker(
|
|||||||
.unwrap()
|
.unwrap()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[rstest::rstest]
|
||||||
|
#[tokio::test]
|
||||||
|
async fn test_gc_keeps_sst_across_index_versions(
|
||||||
|
#[values(false, true)] full_file_listing: bool,
|
||||||
|
#[values(false, true)] old_reader_last: bool,
|
||||||
|
) {
|
||||||
|
use crate::sst::file::{FileMeta, IndexType};
|
||||||
|
use crate::sst::version::SstVersion;
|
||||||
|
use common_time::Timestamp;
|
||||||
|
|
||||||
|
let mut env = TestEnv::new().await;
|
||||||
|
let engine = env
|
||||||
|
.create_engine(MitoConfig {
|
||||||
|
gc: GcConfig {
|
||||||
|
enable: true,
|
||||||
|
lingering_time: None,
|
||||||
|
..Default::default()
|
||||||
|
},
|
||||||
|
..Default::default()
|
||||||
|
})
|
||||||
|
.await;
|
||||||
|
let region_id = RegionId::new(1, 1);
|
||||||
|
engine
|
||||||
|
.handle_request(
|
||||||
|
region_id,
|
||||||
|
RegionRequest::Create(CreateRequestBuilder::new().build()),
|
||||||
|
)
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
let region = engine.get_region(region_id).unwrap();
|
||||||
|
let meta = FileMeta {
|
||||||
|
region_id,
|
||||||
|
file_id: store_api::storage::FileId::random(),
|
||||||
|
available_indexes: smallvec::smallvec![IndexType::InvertedIndex],
|
||||||
|
..Default::default()
|
||||||
|
};
|
||||||
|
let mut version = SstVersion::new(crate::test_util::memtable_util::metadata_for_test());
|
||||||
|
version.add_files(region.file_purger.clone(), [meta.clone()].into_iter());
|
||||||
|
let old = version.levels()[0].files[&meta.file_id].clone();
|
||||||
|
let updated = FileMeta {
|
||||||
|
index_version: 2,
|
||||||
|
..meta.clone()
|
||||||
|
};
|
||||||
|
version.add_files(region.file_purger.clone(), [updated].into_iter());
|
||||||
|
let new = version.levels()[0].files[&meta.file_id].clone();
|
||||||
|
version.mark_all_deleted();
|
||||||
|
drop(version);
|
||||||
|
|
||||||
|
let store = region.access_layer.object_store();
|
||||||
|
let mut entries = Vec::new();
|
||||||
|
for suffix in ["parquet", "puffin", "2.puffin"] {
|
||||||
|
entries.push(write_and_list_entry(store, &format!("{}.{}", meta.file_id, suffix)).await);
|
||||||
|
}
|
||||||
|
let removed = BTreeMap::from([(
|
||||||
|
Timestamp::new_millisecond(0),
|
||||||
|
HashSet::from([
|
||||||
|
RemovedFile::File(meta.file_id, Some(2)),
|
||||||
|
RemovedFile::Index(meta.file_id, 0),
|
||||||
|
RemovedFile::Index(meta.file_id, 2),
|
||||||
|
]),
|
||||||
|
)]);
|
||||||
|
let worker = create_gc_worker(
|
||||||
|
&engine,
|
||||||
|
BTreeMap::from([(region_id, Some(region.clone()))]),
|
||||||
|
&FileRefsManifest::default(),
|
||||||
|
full_file_listing,
|
||||||
|
)
|
||||||
|
.await;
|
||||||
|
let (first, last) = if old_reader_last {
|
||||||
|
(new, old)
|
||||||
|
} else {
|
||||||
|
(old, new)
|
||||||
|
};
|
||||||
|
let released_version = first.index_id().version;
|
||||||
|
let live_version = last.index_id().version;
|
||||||
|
let mut handles = [Some(first), Some(last)];
|
||||||
|
for released in 0..=2 {
|
||||||
|
if released > 0 {
|
||||||
|
drop(handles[released - 1].take());
|
||||||
|
}
|
||||||
|
let refs = engine
|
||||||
|
.get_snapshot_of_file_refs([region_id], HashMap::new())
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
let refs = refs.file_refs.get(®ion_id).cloned().unwrap_or_default();
|
||||||
|
let expected_versions = match released {
|
||||||
|
0 => vec![0, 2],
|
||||||
|
1 => vec![live_version],
|
||||||
|
_ => vec![],
|
||||||
|
};
|
||||||
|
assert_eq!(
|
||||||
|
refs,
|
||||||
|
expected_versions
|
||||||
|
.into_iter()
|
||||||
|
.map(|v| FileRef::new(region_id, meta.file_id, Some(v)))
|
||||||
|
.collect()
|
||||||
|
);
|
||||||
|
let tmp_refs = refs
|
||||||
|
.into_iter()
|
||||||
|
.map(|f| (f.file_id, f.index_version))
|
||||||
|
.collect();
|
||||||
|
let deleted: HashSet<_> = worker
|
||||||
|
.list_to_be_deleted_files(
|
||||||
|
region_id,
|
||||||
|
false,
|
||||||
|
&HashMap::new(),
|
||||||
|
&tmp_refs,
|
||||||
|
removed.clone(),
|
||||||
|
entries.clone(),
|
||||||
|
)
|
||||||
|
.await
|
||||||
|
.unwrap()
|
||||||
|
.into_iter()
|
||||||
|
.collect();
|
||||||
|
let expected = match released {
|
||||||
|
0 => HashSet::new(),
|
||||||
|
1 => HashSet::from([RemovedFile::Index(meta.file_id, released_version)]),
|
||||||
|
_ => HashSet::from([
|
||||||
|
RemovedFile::File(meta.file_id, if full_file_listing { None } else { Some(2) }),
|
||||||
|
RemovedFile::Index(meta.file_id, 0),
|
||||||
|
RemovedFile::Index(meta.file_id, 2),
|
||||||
|
]),
|
||||||
|
};
|
||||||
|
assert_eq!(deleted, expected);
|
||||||
|
}
|
||||||
|
// A manifest reference also protects Parquet regardless of its index version.
|
||||||
|
let deleted = worker
|
||||||
|
.list_to_be_deleted_files(
|
||||||
|
region_id,
|
||||||
|
false,
|
||||||
|
&HashMap::from([(meta.file_id, Some(3))]),
|
||||||
|
&HashSet::new(),
|
||||||
|
removed,
|
||||||
|
entries,
|
||||||
|
)
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
assert!(deleted.iter().all(|f| matches!(f, RemovedFile::Index(..))));
|
||||||
|
}
|
||||||
|
|
||||||
#[tokio::test]
|
#[tokio::test]
|
||||||
async fn test_gc_worker_ignores_dropping_marker_file() {
|
async fn test_gc_worker_ignores_dropping_marker_file() {
|
||||||
init_default_ut_logging();
|
init_default_ut_logging();
|
||||||
|
|||||||
+59
-15
@@ -551,13 +551,13 @@ impl fmt::Debug for FileHandle {
|
|||||||
f.debug_struct("FileHandle")
|
f.debug_struct("FileHandle")
|
||||||
.field("meta", self.meta_ref())
|
.field("meta", self.meta_ref())
|
||||||
.field("compacting", &self.compacting())
|
.field("compacting", &self.compacting())
|
||||||
.field("deleted", &self.inner.deleted.load(Ordering::Relaxed))
|
.field("deleted", &self.is_deleted())
|
||||||
.finish()
|
.finish()
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
impl FileHandle {
|
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 {
|
pub fn new(meta: FileMeta, file_purger: FilePurgerRef) -> FileHandle {
|
||||||
let pk_range = meta.primary_key_range();
|
let pk_range = meta.primary_key_range();
|
||||||
FileHandle {
|
FileHandle {
|
||||||
@@ -576,6 +576,23 @@ impl FileHandle {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Replaces an index metadata view without splitting the underlying SST lifetime.
|
||||||
|
pub(crate) fn with_updated_index(&self, meta: FileMeta) -> Self {
|
||||||
|
debug_assert_eq!(self.file_id(), meta.file_id());
|
||||||
|
let primary_key_range = meta
|
||||||
|
.primary_key_range()
|
||||||
|
.or_else(|| self.raw_primary_key_range());
|
||||||
|
let lifetime = self.inner.lifetime.clone();
|
||||||
|
*lifetime.meta.lock().unwrap() = meta.clone();
|
||||||
|
Self {
|
||||||
|
inner: Arc::new(FileHandleInner::with_lifetime(
|
||||||
|
meta,
|
||||||
|
lifetime,
|
||||||
|
primary_key_range,
|
||||||
|
)),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/// Returns the region id of the file.
|
/// Returns the region id of the file.
|
||||||
pub fn region_id(&self) -> RegionId {
|
pub fn region_id(&self) -> RegionId {
|
||||||
self.inner.meta.region_id
|
self.inner.meta.region_id
|
||||||
@@ -615,7 +632,7 @@ impl FileHandle {
|
|||||||
|
|
||||||
/// Mark the file as deleted and will delete it on drop asynchronously
|
/// Mark the file as deleted and will delete it on drop asynchronously
|
||||||
pub fn mark_deleted(&self) {
|
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 {
|
pub fn compacting(&self) -> bool {
|
||||||
@@ -650,7 +667,7 @@ impl FileHandle {
|
|||||||
}
|
}
|
||||||
|
|
||||||
pub fn file_purger(&self) -> FilePurgerRef {
|
pub fn file_purger(&self) -> FilePurgerRef {
|
||||||
self.inner.file_purger.clone()
|
self.inner.lifetime.file_purger.clone()
|
||||||
}
|
}
|
||||||
|
|
||||||
pub fn size(&self) -> u64 {
|
pub fn size(&self) -> u64 {
|
||||||
@@ -670,7 +687,7 @@ impl FileHandle {
|
|||||||
}
|
}
|
||||||
|
|
||||||
pub fn is_deleted(&self) -> bool {
|
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
|
/// Returns bounds aligned to the caller's pinned schema, before any comparison
|
||||||
@@ -776,39 +793,66 @@ impl PrimaryKeyRange {
|
|||||||
struct FileHandleInner {
|
struct FileHandleInner {
|
||||||
meta: FileMeta,
|
meta: FileMeta,
|
||||||
compacting: AtomicBool,
|
compacting: AtomicBool,
|
||||||
deleted: AtomicBool,
|
|
||||||
index_outdated: AtomicBool,
|
index_outdated: AtomicBool,
|
||||||
primary_key_range: RwLock<PrimaryKeyRange>,
|
primary_key_range: RwLock<PrimaryKeyRange>,
|
||||||
file_purger: FilePurgerRef,
|
lifetime: Arc<SstLifetime>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl Drop for FileHandleInner {
|
impl Drop for FileHandleInner {
|
||||||
fn drop(&mut self) {
|
fn drop(&mut self) {
|
||||||
self.file_purger.remove_file(
|
self.lifetime.file_purger.remove_file(
|
||||||
self.meta.clone(),
|
self.meta.clone(),
|
||||||
self.deleted.load(Ordering::Acquire),
|
|
||||||
self.index_outdated.load(Ordering::Acquire),
|
self.index_outdated.load(Ordering::Acquire),
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
impl FileHandleInner {
|
impl FileHandleInner {
|
||||||
/// There should only be one `FileHandleInner` for each file on a datanode
|
|
||||||
fn new(
|
fn new(
|
||||||
meta: FileMeta,
|
meta: FileMeta,
|
||||||
file_purger: FilePurgerRef,
|
file_purger: FilePurgerRef,
|
||||||
primary_key_range: Option<(Bytes, Bytes)>,
|
primary_key_range: Option<(Bytes, Bytes)>,
|
||||||
) -> FileHandleInner {
|
) -> Self {
|
||||||
file_purger.new_file(&meta);
|
let lifetime = Arc::new(SstLifetime {
|
||||||
FileHandleInner {
|
meta: Mutex::new(meta.clone()),
|
||||||
|
deleted: AtomicBool::new(false),
|
||||||
|
file_purger,
|
||||||
|
});
|
||||||
|
Self::with_lifetime(meta, lifetime, primary_key_range)
|
||||||
|
}
|
||||||
|
|
||||||
|
fn with_lifetime(
|
||||||
|
meta: FileMeta,
|
||||||
|
lifetime: Arc<SstLifetime>,
|
||||||
|
primary_key_range: Option<(Bytes, Bytes)>,
|
||||||
|
) -> Self {
|
||||||
|
lifetime.file_purger.new_file(&meta);
|
||||||
|
Self {
|
||||||
meta,
|
meta,
|
||||||
compacting: AtomicBool::new(false),
|
compacting: AtomicBool::new(false),
|
||||||
deleted: AtomicBool::new(false),
|
|
||||||
index_outdated: AtomicBool::new(false),
|
index_outdated: AtomicBool::new(false),
|
||||||
primary_key_range: RwLock::new(
|
primary_key_range: RwLock::new(
|
||||||
primary_key_range.map_or(PrimaryKeyRange::Missing, PrimaryKeyRange::Raw),
|
primary_key_range.map_or(PrimaryKeyRange::Missing, PrimaryKeyRange::Raw),
|
||||||
),
|
),
|
||||||
file_purger,
|
lifetime,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Physical deletion must wait for readers of every index metadata version.
|
||||||
|
/// Independent logical references (for example, copied regions) do not share this guard.
|
||||||
|
struct SstLifetime {
|
||||||
|
/// Latest metadata ensures final purge includes indexes newer than the last reader's view.
|
||||||
|
meta: Mutex<FileMeta>,
|
||||||
|
deleted: AtomicBool,
|
||||||
|
file_purger: FilePurgerRef,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl Drop for SstLifetime {
|
||||||
|
fn drop(&mut self) {
|
||||||
|
if self.deleted.load(Ordering::Acquire) {
|
||||||
|
self.file_purger
|
||||||
|
.purge_file(self.meta.get_mut().unwrap().clone());
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -29,11 +29,13 @@ use crate::sst::range_index::RangeIndexDeleter;
|
|||||||
|
|
||||||
/// A worker to delete files in background.
|
/// A worker to delete files in background.
|
||||||
pub trait FilePurger: Send + Sync + fmt::Debug {
|
pub trait FilePurger: Send + Sync + fmt::Debug {
|
||||||
/// Send a request to remove the file.
|
/// Releases one metadata version's reference, deleting its index if outdated.
|
||||||
/// If `is_delete` is true, the file will be deleted from the storage.
|
/// This never deletes the SST itself.
|
||||||
/// Otherwise, only the reference will be removed.
|
fn remove_file(&self, file_meta: FileMeta, index_outdated: bool);
|
||||||
/// 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 after all metadata versions release their readers.
|
||||||
|
/// This does not release a reference; each version already did so on drop.
|
||||||
|
fn purge_file(&self, file_meta: FileMeta);
|
||||||
|
|
||||||
/// Notify the purger of a new file created.
|
/// Notify the purger of a new file created.
|
||||||
/// This is useful for object store based storage, where we need to track the file references
|
/// This is useful for object store based storage, where we need to track the file references
|
||||||
@@ -50,9 +52,9 @@ pub type FilePurgerRef = Arc<dyn FilePurger>;
|
|||||||
pub struct NoopFilePurger;
|
pub struct NoopFilePurger;
|
||||||
|
|
||||||
impl FilePurger for NoopFilePurger {
|
impl FilePurger for NoopFilePurger {
|
||||||
fn remove_file(&self, _file_meta: FileMeta, _is_delete: bool, _index_outdated: bool) {
|
fn remove_file(&self, _file_meta: FileMeta, _index_outdated: bool) {}
|
||||||
// noop
|
|
||||||
}
|
fn purge_file(&self, _file_meta: FileMeta) {}
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Purger that purges file for current region.
|
/// Purger that purges file for current region.
|
||||||
@@ -198,18 +200,20 @@ impl LocalFilePurger {
|
|||||||
}
|
}
|
||||||
|
|
||||||
impl FilePurger for LocalFilePurger {
|
impl FilePurger for LocalFilePurger {
|
||||||
fn remove_file(&self, file_meta: FileMeta, is_delete: bool, index_outdated: bool) {
|
fn remove_file(&self, file_meta: FileMeta, index_outdated: bool) {
|
||||||
if is_delete {
|
if index_outdated {
|
||||||
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);
|
self.delete_index(file_meta);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
fn purge_file(&self, file_meta: FileMeta) {
|
||||||
|
schedule_range_index_deletion(
|
||||||
|
&self.scheduler,
|
||||||
|
self.range_index_deleter.as_ref(),
|
||||||
|
file_meta.file_id,
|
||||||
|
);
|
||||||
|
self.delete_file(file_meta);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
pub struct ObjectStoreFilePurger {
|
pub struct ObjectStoreFilePurger {
|
||||||
@@ -246,19 +250,17 @@ fn schedule_range_index_deletion(
|
|||||||
}
|
}
|
||||||
|
|
||||||
impl FilePurger for ObjectStoreFilePurger {
|
impl FilePurger for ObjectStoreFilePurger {
|
||||||
fn remove_file(&self, file_meta: FileMeta, is_delete: bool, _index_outdated: bool) {
|
fn remove_file(&self, file_meta: FileMeta, _index_outdated: bool) {
|
||||||
// if not on local file system, instead inform the global file purger to remove the file reference.
|
// GC handles physical SST and index deletion after references are released.
|
||||||
// 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);
|
self.file_ref_manager.remove_file(&file_meta);
|
||||||
if is_delete {
|
}
|
||||||
schedule_range_index_deletion(
|
|
||||||
&self.scheduler,
|
fn purge_file(&self, file_meta: FileMeta) {
|
||||||
self.range_index_deleter.as_ref(),
|
schedule_range_index_deletion(
|
||||||
file_meta.file_id,
|
&self.scheduler,
|
||||||
);
|
self.range_index_deleter.as_ref(),
|
||||||
}
|
file_meta.file_id,
|
||||||
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
fn new_file(&self, file_meta: &FileMeta) {
|
fn new_file(&self, file_meta: &FileMeta) {
|
||||||
@@ -295,6 +297,7 @@ mod tests {
|
|||||||
async fn test_range_index_purge_on_handle_release(
|
async fn test_range_index_purge_on_handle_release(
|
||||||
#[case] gc_enabled: bool,
|
#[case] gc_enabled: bool,
|
||||||
#[values(false, true)] is_delete: bool,
|
#[values(false, true)] is_delete: bool,
|
||||||
|
#[values(false, true)] update_index: bool,
|
||||||
) {
|
) {
|
||||||
common_telemetry::init_default_ut_logging();
|
common_telemetry::init_default_ut_logging();
|
||||||
|
|
||||||
@@ -324,7 +327,7 @@ mod tests {
|
|||||||
let path = location::sst_file_path(sst_dir, sst_file_id, layer.path_type());
|
let path = location::sst_file_path(sst_dir, sst_file_id, layer.path_type());
|
||||||
object_store.write(&path, vec![0; 4096]).await.unwrap();
|
object_store.write(&path, vec![0; 4096]).await.unwrap();
|
||||||
|
|
||||||
let scheduler = Arc::new(LocalScheduler::new(3));
|
let scheduler = Arc::new(LocalScheduler::new(1));
|
||||||
|
|
||||||
let index_store = ObjectStore::new(object_store::services::Memory::default()).unwrap();
|
let index_store = ObjectStore::new(object_store::services::Memory::default()).unwrap();
|
||||||
let owner = RegionId::new(9, 1);
|
let owner = RegionId::new(9, 1);
|
||||||
@@ -341,33 +344,52 @@ mod tests {
|
|||||||
);
|
);
|
||||||
|
|
||||||
{
|
{
|
||||||
let handle = FileHandle::new(
|
let meta = FileMeta {
|
||||||
FileMeta {
|
region_id: sst_file_id.region_id(),
|
||||||
region_id: sst_file_id.region_id(),
|
file_id: sst_file_id.file_id(),
|
||||||
file_id: sst_file_id.file_id(),
|
file_size: 4096,
|
||||||
time_range: FileTimeRange::default(),
|
max_row_group_uncompressed_size: 4096,
|
||||||
level: 0,
|
..Default::default()
|
||||||
file_size: 4096,
|
};
|
||||||
max_row_group_uncompressed_size: 4096,
|
let mut version = crate::sst::version::SstVersion::new(
|
||||||
available_indexes: Default::default(),
|
crate::test_util::memtable_util::metadata_for_test(),
|
||||||
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,
|
|
||||||
);
|
);
|
||||||
if is_delete {
|
version.add_files(file_purger, [meta].into_iter());
|
||||||
handle.mark_deleted();
|
let reader = version.levels()[0].files[&sst_file_id.file_id()].clone();
|
||||||
|
if update_index {
|
||||||
|
for index_version in 1..=2 {
|
||||||
|
let mut meta = reader.meta_ref().clone();
|
||||||
|
meta.index_version = index_version;
|
||||||
|
meta.available_indexes.push(IndexType::InvertedIndex);
|
||||||
|
let path = location::index_file_path(sst_dir, meta.index_id(), PathType::Bare);
|
||||||
|
object_store.write(&path, "index").await.unwrap();
|
||||||
|
version.add_files(reader.file_purger(), [meta].into_iter());
|
||||||
|
}
|
||||||
}
|
}
|
||||||
let reader = handle.clone();
|
if is_delete {
|
||||||
drop(handle);
|
version.mark_all_deleted();
|
||||||
|
}
|
||||||
|
drop(version);
|
||||||
|
// A single-worker queue barrier makes premature purging deterministic.
|
||||||
|
let (tx, rx) = tokio::sync::oneshot::channel();
|
||||||
|
scheduler
|
||||||
|
.schedule(Box::pin(async move {
|
||||||
|
tx.send(()).unwrap();
|
||||||
|
}))
|
||||||
|
.unwrap();
|
||||||
|
rx.await.unwrap();
|
||||||
|
assert!(object_store.exists(&path).await.unwrap());
|
||||||
assert!(index_store.exists(&index_path).await.unwrap());
|
assert!(index_store.exists(&index_path).await.unwrap());
|
||||||
|
if update_index {
|
||||||
|
for index_version in 1..=2 {
|
||||||
|
let path = location::index_file_path(
|
||||||
|
sst_dir,
|
||||||
|
RegionIndexId::new(sst_file_id, index_version),
|
||||||
|
PathType::Bare,
|
||||||
|
);
|
||||||
|
assert!(object_store.exists(&path).await.unwrap());
|
||||||
|
}
|
||||||
|
}
|
||||||
drop(reader);
|
drop(reader);
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -378,6 +400,19 @@ mod tests {
|
|||||||
gc_enabled || !is_delete
|
gc_enabled || !is_delete
|
||||||
);
|
);
|
||||||
assert_eq!(index_store.exists(&index_path).await.unwrap(), !is_delete);
|
assert_eq!(index_store.exists(&index_path).await.unwrap(), !is_delete);
|
||||||
|
if update_index {
|
||||||
|
for index_version in 1..=2 {
|
||||||
|
let path = location::index_file_path(
|
||||||
|
sst_dir,
|
||||||
|
RegionIndexId::new(sst_file_id, index_version),
|
||||||
|
PathType::Bare,
|
||||||
|
);
|
||||||
|
assert_eq!(
|
||||||
|
object_store.exists(&path).await.unwrap(),
|
||||||
|
gc_enabled || !is_delete
|
||||||
|
);
|
||||||
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
#[tokio::test]
|
#[tokio::test]
|
||||||
|
|||||||
@@ -147,11 +147,17 @@ impl FileReferenceManager {
|
|||||||
|
|
||||||
for r in &query_regions_for_mem {
|
for r in &query_regions_for_mem {
|
||||||
let manifest = r.manifest_ctx.manifest().await;
|
let manifest = r.manifest_ctx.manifest().await;
|
||||||
// remove in manifest files for smaller size, since gc worker read from manifest later.
|
// Only omit references covered by the exact index version in the manifest.
|
||||||
|
// Older index readers still need their temporary references.
|
||||||
ref_files.entry(r.region_id()).and_modify(|refs| {
|
ref_files.entry(r.region_id()).and_modify(|refs| {
|
||||||
*refs = std::mem::take(refs)
|
*refs = std::mem::take(refs)
|
||||||
.into_iter()
|
.into_iter()
|
||||||
.filter(|f| !manifest.files.contains_key(&f.file_id))
|
.filter(|f| {
|
||||||
|
!manifest
|
||||||
|
.files
|
||||||
|
.get(&f.file_id)
|
||||||
|
.is_some_and(|meta| meta.index_version() == f.index_version)
|
||||||
|
})
|
||||||
.collect();
|
.collect();
|
||||||
});
|
});
|
||||||
manifest_version.insert(r.region_id(), manifest.manifest_version);
|
manifest_version.insert(r.region_id(), manifest.manifest_version);
|
||||||
|
|||||||
@@ -100,7 +100,7 @@ impl SstVersion {
|
|||||||
}
|
}
|
||||||
} else {
|
} else {
|
||||||
// include case like old file have no index or index is outdated
|
// 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(|| {
|
.or_insert_with(|| {
|
||||||
@@ -352,6 +352,42 @@ mod tests {
|
|||||||
assert_eq!(version.time_range(), None);
|
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]
|
#[test]
|
||||||
fn test_add_files() {
|
fn test_add_files() {
|
||||||
let purger = new_noop_file_purger();
|
let purger = new_noop_file_purger();
|
||||||
|
|||||||
Reference in New Issue
Block a user