mito2/sst/
file_purger.rs

1// Copyright 2023 Greptime Team
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use std::fmt;
16use std::sync::Arc;
17
18use common_telemetry::error;
19use store_api::region_request::PathType;
20use store_api::storage::FileId;
21
22use crate::access_layer::AccessLayerRef;
23use crate::cache::CacheManagerRef;
24use crate::error::Result;
25use crate::schedule::scheduler::SchedulerRef;
26use crate::sst::file::{FileMeta, delete_files, delete_index};
27use crate::sst::file_ref::FileReferenceManagerRef;
28use crate::sst::range_index::RangeIndexDeleter;
29
30/// A worker to delete files in background.
31pub trait FilePurger: Send + Sync + fmt::Debug {
32    /// Send a request to remove the file.
33    /// If `is_delete` is true, the file will be deleted from the storage.
34    /// Otherwise, only the reference will be removed.
35    /// If `index_outdated` is true, the index file will be deleted regardless of `is_delete`.
36    fn remove_file(&self, file_meta: FileMeta, is_delete: bool, index_outdated: bool);
37
38    /// Notify the purger of a new file created.
39    /// This is useful for object store based storage, where we need to track the file references
40    /// The default implementation is a no-op.
41    fn new_file(&self, _: &FileMeta) {
42        // noop
43    }
44}
45
46pub type FilePurgerRef = Arc<dyn FilePurger>;
47
48/// A no-op file purger can be used in combination with reading SST files outside of this region.
49#[derive(Debug)]
50pub struct NoopFilePurger;
51
52impl FilePurger for NoopFilePurger {
53    fn remove_file(&self, _file_meta: FileMeta, _is_delete: bool, _index_outdated: bool) {
54        // noop
55    }
56}
57
58/// Purger that purges file for current region.
59pub struct LocalFilePurger {
60    scheduler: SchedulerRef,
61    sst_layer: AccessLayerRef,
62    cache_manager: Option<CacheManagerRef>,
63    range_index_deleter: Option<RangeIndexDeleter>,
64}
65
66impl fmt::Debug for LocalFilePurger {
67    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
68        f.debug_struct("LocalFilePurger")
69            .field("sst_layer", &self.sst_layer)
70            .finish()
71    }
72}
73
74#[cfg(not(any(debug_assertions, test)))]
75/// Whether to enable GC for the file purger.
76pub fn should_enable_gc(global_gc_enabled: bool, object_store_scheme: &'static str) -> bool {
77    global_gc_enabled && object_store_scheme != object_store::services::FS_SCHEME
78}
79
80#[cfg(any(debug_assertions, test))]
81/// Debug builds and unit tests use Fs to exercise object-store GC, including
82/// unit tests compiled with the release profile.
83pub fn should_enable_gc(global_gc_enabled: bool, _object_store_scheme: &'static str) -> bool {
84    global_gc_enabled
85}
86
87/// Creates a file purger based on the storage type of the access layer.
88/// Should be use in combination with Gc Worker.
89///
90/// If the storage is local file system, a `LocalFilePurger` is created, which deletes
91/// the files from both the storage and the cache.
92///
93/// If the storage is an object store, an `ObjectStoreFilePurger` is created, which
94/// only manages SST file references. Companion range indexes are deleted directly in either
95/// mode on final handle release when the SST is marked deleted and a deleter is provided.
96///
97pub fn create_file_purger(
98    gc_enabled: bool,
99    path_type: PathType,
100    scheduler: SchedulerRef,
101    sst_layer: AccessLayerRef,
102    cache_manager: Option<CacheManagerRef>,
103    file_ref_manager: FileReferenceManagerRef,
104    range_index_deleter: Option<RangeIndexDeleter>,
105) -> FilePurgerRef {
106    // Only enable GC for:
107    // - object store based storage
108    // - data or bare path type (metadata region doesn't need to be GCed)
109    if should_enable_gc(gc_enabled, sst_layer.object_store().info().scheme())
110        && matches!(path_type, PathType::Data | PathType::Bare)
111    {
112        Arc::new(ObjectStoreFilePurger {
113            file_ref_manager,
114            scheduler,
115            range_index_deleter,
116        })
117    } else {
118        Arc::new(
119            LocalFilePurger::new(scheduler, sst_layer, cache_manager)
120                .with_range_index_deleter(range_index_deleter),
121        )
122    }
123}
124
125/// Creates a local file purger that deletes files from both the storage and the cache.
126pub fn create_local_file_purger(
127    scheduler: SchedulerRef,
128    sst_layer: AccessLayerRef,
129    cache_manager: Option<CacheManagerRef>,
130    _file_ref_manager: FileReferenceManagerRef,
131    range_index_deleter: Option<RangeIndexDeleter>,
132) -> FilePurgerRef {
133    Arc::new(
134        LocalFilePurger::new(scheduler, sst_layer, cache_manager)
135            .with_range_index_deleter(range_index_deleter),
136    )
137}
138
139impl LocalFilePurger {
140    /// Creates a new purger.
141    pub fn new(
142        scheduler: SchedulerRef,
143        sst_layer: AccessLayerRef,
144        cache_manager: Option<CacheManagerRef>,
145    ) -> Self {
146        Self {
147            scheduler,
148            sst_layer,
149            cache_manager,
150            range_index_deleter: None,
151        }
152    }
153
154    /// Attaches deletion of companion range indexes.
155    pub fn with_range_index_deleter(mut self, deleter: Option<RangeIndexDeleter>) -> Self {
156        self.range_index_deleter = deleter;
157        self
158    }
159
160    /// Stop the scheduler of the file purger.
161    pub async fn stop_scheduler(&self) -> Result<()> {
162        self.scheduler.stop(true).await
163    }
164
165    /// Deletes the file(and it's index, if any) from cache and storage.
166    fn delete_file(&self, file_meta: FileMeta) {
167        let sst_layer = self.sst_layer.clone();
168        let cache_manager = self.cache_manager.clone();
169        if let Err(e) = self.scheduler.schedule(Box::pin(async move {
170            if let Err(e) = delete_files(
171                file_meta.region_id,
172                &[(file_meta.file_id, file_meta.index_id().version)],
173                file_meta.exists_index(),
174                &sst_layer,
175                &cache_manager,
176            )
177            .await
178            {
179                error!(e; "Failed to delete file {:?} from storage", file_meta);
180            }
181        })) {
182            error!(e; "Failed to schedule the file purge request");
183        }
184    }
185
186    fn delete_index(&self, file_meta: FileMeta) {
187        let sst_layer = self.sst_layer.clone();
188        let cache_manager = self.cache_manager.clone();
189        if let Err(e) = self.scheduler.schedule(Box::pin(async move {
190            let index_id = file_meta.index_id();
191            if let Err(e) = delete_index(index_id, &sst_layer, &cache_manager).await {
192                error!(e; "Failed to delete index for file {:?} from storage", file_meta);
193            }
194        })) {
195            error!(e; "Failed to schedule the index purge request");
196        }
197    }
198}
199
200impl FilePurger for LocalFilePurger {
201    fn remove_file(&self, file_meta: FileMeta, is_delete: bool, index_outdated: bool) {
202        if is_delete {
203            schedule_range_index_deletion(
204                &self.scheduler,
205                self.range_index_deleter.as_ref(),
206                file_meta.file_id,
207            );
208            self.delete_file(file_meta);
209        } else if index_outdated {
210            self.delete_index(file_meta);
211        }
212    }
213}
214
215pub struct ObjectStoreFilePurger {
216    file_ref_manager: FileReferenceManagerRef,
217    scheduler: SchedulerRef,
218    range_index_deleter: Option<RangeIndexDeleter>,
219}
220
221impl fmt::Debug for ObjectStoreFilePurger {
222    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
223        f.debug_struct("ObjectStoreFilePurger")
224            .field("file_ref_manager", &self.file_ref_manager)
225            .field("range_index_deleter", &self.range_index_deleter)
226            .finish_non_exhaustive()
227    }
228}
229
230/// Range indexes are deleted directly even when SST deletion is delegated to GC.
231fn schedule_range_index_deletion(
232    scheduler: &SchedulerRef,
233    deleter: Option<&RangeIndexDeleter>,
234    file_id: FileId,
235) {
236    let Some(deleter) = deleter.cloned() else {
237        return;
238    };
239    if let Err(error) = scheduler.schedule(Box::pin(async move {
240        if let Err(error) = deleter.delete(file_id).await {
241            error!(error; "Failed to delete range index, file_id: {file_id}");
242        }
243    })) {
244        error!(error; "Failed to schedule range-index deletion, file_id: {file_id}");
245    }
246}
247
248impl FilePurger for ObjectStoreFilePurger {
249    fn remove_file(&self, file_meta: FileMeta, is_delete: bool, _index_outdated: bool) {
250        // if not on local file system, instead inform the global file purger to remove the file reference.
251        // notice that no matter whether the file is deleted or not, we need to remove the reference
252        // because the file is no longer in use nonetheless.
253        // for same reason, we don't care about index_outdated here.
254        self.file_ref_manager.remove_file(&file_meta);
255        if is_delete {
256            schedule_range_index_deletion(
257                &self.scheduler,
258                self.range_index_deleter.as_ref(),
259                file_meta.file_id,
260            );
261        }
262    }
263
264    fn new_file(&self, file_meta: &FileMeta) {
265        self.file_ref_manager.add_file(file_meta);
266    }
267}
268
269#[cfg(test)]
270mod tests {
271    use std::num::NonZeroU64;
272
273    use common_test_util::temp_dir::create_temp_dir;
274    use object_store::ObjectStore;
275    use object_store::services::Fs;
276    use smallvec::SmallVec;
277    use store_api::region_request::PathType;
278    use store_api::storage::{FileId, RegionId};
279
280    use super::*;
281    use crate::access_layer::AccessLayer;
282    use crate::schedule::scheduler::{LocalScheduler, Scheduler};
283    use crate::sst::file::{
284        ColumnIndexMetadata, FileHandle, FileMeta, FileTimeRange, IndexType, RegionFileId,
285        RegionIndexId,
286    };
287    use crate::sst::index::intermediate::IntermediateManager;
288    use crate::sst::index::puffin_manager::PuffinManagerFactory;
289    use crate::sst::location;
290
291    #[rstest::rstest]
292    #[case(false)]
293    #[case(true)]
294    #[tokio::test]
295    async fn test_range_index_purge_on_handle_release(
296        #[case] gc_enabled: bool,
297        #[values(false, true)] is_delete: bool,
298    ) {
299        common_telemetry::init_default_ut_logging();
300
301        let dir = create_temp_dir("file-purge");
302        let dir_path = dir.path().display().to_string();
303        let builder = Fs::default().root(&dir_path);
304        let sst_file_id = RegionFileId::new(RegionId::new(0, 0), FileId::random());
305        let sst_dir = "table1";
306
307        let index_aux_path = dir.path().join("index_aux");
308        let puffin_mgr = PuffinManagerFactory::new(&index_aux_path, 4096, None, None)
309            .await
310            .unwrap();
311        let intm_mgr = IntermediateManager::init_fs(index_aux_path.to_str().unwrap())
312            .await
313            .unwrap();
314
315        let object_store = ObjectStore::new(builder).unwrap();
316
317        let layer = Arc::new(AccessLayer::new(
318            sst_dir,
319            PathType::Bare,
320            object_store.clone(),
321            puffin_mgr,
322            intm_mgr,
323        ));
324        let path = location::sst_file_path(sst_dir, sst_file_id, layer.path_type());
325        object_store.write(&path, vec![0; 4096]).await.unwrap();
326
327        let scheduler = Arc::new(LocalScheduler::new(3));
328
329        let index_store = ObjectStore::new(object_store::services::Memory::default()).unwrap();
330        let owner = RegionId::new(9, 1);
331        let index_path = crate::sst::range_index::range_index_path(owner, sst_file_id.file_id());
332        index_store.write(&index_path, "range index").await.unwrap();
333        let file_purger = create_file_purger(
334            gc_enabled,
335            PathType::Bare,
336            scheduler.clone(),
337            layer,
338            None,
339            Arc::new(crate::sst::file_ref::FileReferenceManager::new(None)),
340            Some(RangeIndexDeleter::new(index_store.clone(), owner)),
341        );
342
343        {
344            let handle = FileHandle::new(
345                FileMeta {
346                    region_id: sst_file_id.region_id(),
347                    file_id: sst_file_id.file_id(),
348                    time_range: FileTimeRange::default(),
349                    level: 0,
350                    file_size: 4096,
351                    max_row_group_uncompressed_size: 4096,
352                    available_indexes: Default::default(),
353                    indexes: Default::default(),
354                    index_file_size: 0,
355                    index_version: 0,
356                    num_rows: 0,
357                    num_row_groups: 0,
358                    sequence: None,
359                    partition_expr: None,
360                    num_series: 0,
361                    ..Default::default()
362                },
363                file_purger,
364            );
365            if is_delete {
366                handle.mark_deleted();
367            }
368            let reader = handle.clone();
369            drop(handle);
370            assert!(index_store.exists(&index_path).await.unwrap());
371            drop(reader);
372        }
373
374        scheduler.stop(true).await.unwrap();
375
376        assert_eq!(
377            object_store.exists(&path).await.unwrap(),
378            gc_enabled || !is_delete
379        );
380        assert_eq!(index_store.exists(&index_path).await.unwrap(), !is_delete);
381    }
382
383    #[tokio::test]
384    async fn test_file_purge_with_index() {
385        common_telemetry::init_default_ut_logging();
386
387        let dir = create_temp_dir("file-purge");
388        let dir_path = dir.path().display().to_string();
389        let builder = Fs::default().root(&dir_path);
390        let sst_file_id = RegionFileId::new(RegionId::new(0, 0), FileId::random());
391        let index_file_id = RegionIndexId::new(sst_file_id, 0);
392        let sst_dir = "table1";
393
394        let index_aux_path = dir.path().join("index_aux");
395        let puffin_mgr = PuffinManagerFactory::new(&index_aux_path, 4096, None, None)
396            .await
397            .unwrap();
398        let intm_mgr = IntermediateManager::init_fs(index_aux_path.to_str().unwrap())
399            .await
400            .unwrap();
401
402        let object_store = ObjectStore::new(builder).unwrap();
403
404        let layer = Arc::new(AccessLayer::new(
405            sst_dir,
406            PathType::Bare,
407            object_store.clone(),
408            puffin_mgr,
409            intm_mgr,
410        ));
411        let path = location::sst_file_path(sst_dir, sst_file_id, layer.path_type());
412        object_store.write(&path, vec![0; 4096]).await.unwrap();
413
414        let index_path = location::index_file_path(sst_dir, index_file_id, layer.path_type());
415        object_store
416            .write(&index_path, vec![0; 4096])
417            .await
418            .unwrap();
419
420        let scheduler = Arc::new(LocalScheduler::new(3));
421
422        let file_purger = Arc::new(LocalFilePurger::new(scheduler.clone(), layer, None));
423
424        {
425            let handle = FileHandle::new(
426                FileMeta {
427                    region_id: sst_file_id.region_id(),
428                    file_id: sst_file_id.file_id(),
429                    time_range: FileTimeRange::default(),
430                    level: 0,
431                    file_size: 4096,
432                    max_row_group_uncompressed_size: 4096,
433                    available_indexes: SmallVec::from_iter([IndexType::InvertedIndex]),
434                    indexes: vec![ColumnIndexMetadata {
435                        column_id: 0,
436                        created_indexes: SmallVec::from_iter([IndexType::InvertedIndex]),
437                        ..Default::default()
438                    }],
439                    index_file_size: 4096,
440                    index_version: 0,
441                    num_rows: 1024,
442                    num_row_groups: 1,
443                    sequence: NonZeroU64::new(4096),
444                    partition_expr: None,
445                    num_series: 0,
446                    ..Default::default()
447                },
448                file_purger,
449            );
450            // mark file as deleted and drop the handle, we expect the sst file and the index file are deleted.
451            handle.mark_deleted();
452        }
453
454        scheduler.stop(true).await.unwrap();
455
456        assert!(!object_store.exists(&path).await.unwrap());
457        assert!(!object_store.exists(&index_path).await.unwrap());
458    }
459}
mito2/sst/file_purger.rs

mito2/sst/
file_purger.rs
