diff --git a/src/mito2/src/gc.rs b/src/mito2/src/gc.rs index 8c0e72fc08..babebabb74 100644 --- a/src/mito2/src/gc.rs +++ b/src/mito2/src/gc.rs @@ -54,6 +54,7 @@ use crate::metrics::{ use crate::region::{MitoRegionRef, RegionRoleState}; use crate::sst::file::{RegionFileId, RegionIndexId, delete_files, delete_indexes}; use crate::sst::location::{self}; +use crate::worker::DROPPING_MARKER_FILE; #[cfg(test)] mod worker_test; @@ -948,6 +949,10 @@ impl LocalGcWorker { }); for entry in entries { + if entry.name() == DROPPING_MARKER_FILE { + continue; + } + let (file_id, file_type) = match location::parse_file_id_type_from_path(entry.name()) { Ok((file_id, file_type)) => (file_id, file_type), Err(err) => { diff --git a/src/mito2/src/gc/worker_test.rs b/src/mito2/src/gc/worker_test.rs index 1efa70e6e9..cd8b875e9e 100644 --- a/src/mito2/src/gc/worker_test.rs +++ b/src/mito2/src/gc/worker_test.rs @@ -18,6 +18,7 @@ use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use api::v1::Rows; use async_trait::async_trait; +use bytes::Bytes; use common_base::Plugins; use common_telemetry::init_default_ut_logging; use futures::TryStreamExt; @@ -35,10 +36,12 @@ 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::metrics::GC_SKIPPED_UNPARSABLE_FILES; use crate::region::MitoRegionRef; use crate::test_util::{ CreateRequestBuilder, TestEnv, build_rows, flush_region, put_rows, rows_schema, }; +use crate::worker::DROPPING_MARKER_FILE; async fn create_gc_worker( mito_engine: &MitoEngine, @@ -71,6 +74,77 @@ async fn create_gc_worker( .unwrap() } +#[tokio::test] +async fn test_gc_worker_ignores_dropping_marker_file() { + init_default_ut_logging(); + + let mut env = TestEnv::new().await; + env.log_store = Some(env.create_log_store().await); + env.object_store_manager = Some(Arc::new(env.create_in_memory_object_store_manager())); + + let engine = env + .new_mito_engine(MitoConfig { + gc: GcConfig { + enable: true, + lingering_time: None, + ..Default::default() + }, + ..Default::default() + }) + .await; + + let region_id = RegionId::new(1, 1); + env.get_schema_metadata_manager() + .register_region_table_info( + region_id.table_id(), + "test_table", + "test_catalog", + "test_schema", + None, + env.get_kv_backend(), + ) + .await; + + let request = CreateRequestBuilder::new().build(); + engine + .handle_request(region_id, RegionRequest::Create(request)) + .await + .unwrap(); + + let region = engine.get_region(region_id).unwrap(); + let manifest = region.manifest_ctx.manifest().await; + let marker_path = object_store::util::join_path( + ®ion.access_layer.build_region_dir(region_id), + DROPPING_MARKER_FILE, + ); + region + .access_layer + .object_store() + .write(&marker_path, Bytes::new()) + .await + .unwrap(); + + let before = GC_SKIPPED_UNPARSABLE_FILES.get(); + let regions = BTreeMap::from([(region_id, Some(region.clone()))]); + let file_ref_manifest = FileRefsManifest { + file_refs: Default::default(), + manifest_version: [(region_id, manifest.manifest_version)].into(), + cross_region_refs: HashMap::new(), + }; + + let gc_worker = create_gc_worker(&engine, regions, &file_ref_manifest, true).await; + let report = gc_worker.run().await.unwrap(); + + assert!( + report + .deleted_files + .get(®ion_id) + .is_none_or(|files| files.is_empty()) + ); + assert!(report.need_retry_regions.is_empty()); + assert_eq!(GC_SKIPPED_UNPARSABLE_FILES.get(), before); +} + /// Test insert/flush then truncate can allow gc worker to delete files #[tokio::test] async fn test_gc_worker_basic_truncate() {