From c7ea80558f9cf53a8feaa4fb6fdcdd07060f4ddf Mon Sep 17 00:00:00 2001 From: "Lei, HUANG" <6406592+v0y4g3r@users.noreply.github.com> Date: Thu, 29 Oct 2026 02:58:37 -0700 Subject: [PATCH] fix(mito2): preserve live index files across remote compactions (#10046) fix(mito2): reuse executor-owned compaction index resources Signed-off-by: Lei, HUANG --- src/mito2/src/compaction/compactor.rs | 230 +++++++++++++++----------- src/mito2/src/engine/open_test.rs | 141 +++++++++++++++- src/mito2/src/engine/region_hook.rs | 4 +- 3 files changed, 266 insertions(+), 109 deletions(-) diff --git a/src/mito2/src/compaction/compactor.rs b/src/mito2/src/compaction/compactor.rs index 1673721ec9b..4c89343b61c 100644 --- a/src/mito2/src/compaction/compactor.rs +++ b/src/mito2/src/compaction/compactor.rs @@ -201,28 +201,26 @@ pub struct OpenCompactionRegionRequest { pub plugins: Plugins, } -/// Open a compaction region from a compaction request. -/// It's simple version of RegionOpener::open(). -pub async fn open_compaction_region( - req: &OpenCompactionRegionRequest, - mito_config: &MitoConfig, +/// Opens manifest-backed compaction regions using executor-owned index resources. +/// +/// Create one opener per auxiliary directory before accepting jobs and reuse it +/// for the executor's lifetime. Initialization recovers staging files and removes +/// stale intermediate files, so repeating it can delete another job's live files. +#[derive(Clone)] +pub struct CompactionRegionOpener { + mito_config: Arc, object_store_manager: ObjectStoreManagerRef, - ttl_provider: Either, -) -> Result { - let object_store = { - let name = &req.region_options.storage; - if let Some(name) = name { - object_store_manager - .find(name) - .with_context(|| ObjectStoreNotFoundSnafu { - object_store: name.clone(), - })? - } else { - object_store_manager.default_object_store() - } - }; + puffin_manager_factory: PuffinManagerFactory, + intermediate_manager: IntermediateManager, +} - let access_layer = { +impl CompactionRegionOpener { + /// Initializes the executor's index resources. The config must be sanitized + /// and its auxiliary directory must not be owned by another engine or opener. + pub async fn new( + mito_config: MitoConfig, + object_store_manager: ObjectStoreManagerRef, + ) -> Result { let puffin_manager_factory = PuffinManagerFactory::new( &mito_config.index.aux_path, mito_config.index.staging_size.as_bytes(), @@ -231,95 +229,131 @@ pub async fn open_compaction_region( ) .await?; let intermediate_manager = - IntermediateManager::init_fs(mito_config.index.aux_path.clone()).await?; + IntermediateManager::init_fs(&mito_config.index.aux_path).await?; - Arc::new(AccessLayer::new( + Ok(Self { + mito_config: Arc::new(mito_config), + object_store_manager, + puffin_manager_factory, + intermediate_manager, + }) + } + + /// Opens a region without reinitializing or cleaning shared index resources. + pub async fn open( + &self, + req: &OpenCompactionRegionRequest, + ttl_provider: Either, + ) -> Result { + let mito_config = &self.mito_config; + let object_store = { + let name = &req.region_options.storage; + if let Some(name) = name { + self.object_store_manager + .find(name) + .with_context(|| ObjectStoreNotFoundSnafu { + object_store: name.clone(), + })? + } else { + self.object_store_manager.default_object_store() + } + }; + + let access_layer = Arc::new(AccessLayer::new( &req.table_dir, req.path_type, object_store.clone(), - puffin_manager_factory, - intermediate_manager, - )) - }; + self.puffin_manager_factory.clone(), + self.intermediate_manager.clone(), + )); - let manifest_manager = { - let region_dir = region_dir_from_table_dir(&req.table_dir, req.region_id, req.path_type); - let region_manifest_options = - RegionManifestOptions::new(mito_config, ®ion_dir, object_store); + let manifest_manager = { + let region_dir = + region_dir_from_table_dir(&req.table_dir, req.region_id, req.path_type); + let region_manifest_options = + RegionManifestOptions::new(mito_config, ®ion_dir, object_store); - RegionManifestManager::open(region_manifest_options, &Default::default()) - .await? - .with_context(|| EmptyRegionDirSnafu { - region_id: req.region_id, - region_dir: region_dir_from_table_dir(&req.table_dir, req.region_id, req.path_type), - })? - }; + RegionManifestManager::open(region_manifest_options, &Default::default()) + .await? + .with_context(|| EmptyRegionDirSnafu { + region_id: req.region_id, + region_dir: region_dir_from_table_dir( + &req.table_dir, + req.region_id, + req.path_type, + ), + })? + }; - let manifest = manifest_manager.manifest(); - let region_metadata = manifest.metadata.clone(); - let hook: Option = req.plugins.get(); - let manifest_ctx = Arc::new(ManifestContext::new( - manifest_manager, - RegionRoleState::Leader(RegionLeaderState::Writable), - hook, - )); + let manifest = manifest_manager.manifest(); + let region_metadata = manifest.metadata.clone(); + let hook: Option = req.plugins.get(); + let manifest_ctx = Arc::new(ManifestContext::new( + manifest_manager, + RegionRoleState::Leader(RegionLeaderState::Writable), + hook, + )); - let file_purger = { - let purge_scheduler = Arc::new(LocalScheduler::new(mito_config.max_background_purges)); - Arc::new(LocalFilePurger::new( - purge_scheduler.clone(), - access_layer.clone(), - None, - )) - }; + let file_purger = { + let purge_scheduler = Arc::new(LocalScheduler::new(mito_config.max_background_purges)); + Arc::new(LocalFilePurger::new( + purge_scheduler.clone(), + access_layer.clone(), + None, + )) + }; - let current_version = { - let mut ssts = SstVersion::new(region_metadata.clone()); - ssts.add_files(file_purger.clone(), manifest.files.values().cloned()); - CompactionVersion { - metadata: region_metadata.clone(), - options: req.region_options.clone(), - ssts: Arc::new(ssts), - // Remote execution uses an already picked output. A manifest-only - // opener cannot prove that a new LastNonNull pick is memtable-safe. - memtable_min_sequence: Some(0), - compaction_time_window: manifest.compaction_time_window, - } - }; + let current_version = { + let mut ssts = SstVersion::new(region_metadata.clone()); + ssts.add_files(file_purger.clone(), manifest.files.values().cloned()); + CompactionVersion { + metadata: region_metadata.clone(), + options: req.region_options.clone(), + ssts: Arc::new(ssts), + // Remote execution uses an already picked output. A manifest-only + // opener cannot prove that a new LastNonNull pick is memtable-safe. + memtable_min_sequence: Some(0), + compaction_time_window: manifest.compaction_time_window, + } + }; - let ttl = match ttl_provider { - // Use the specified ttl. - Either::Left(ttl) => ttl, - // Get the ttl from the schema metadata manager. - Either::Right(schema_metadata_manager) => { - let (_, ttl) = - find_dynamic_options(req.region_id, &req.region_options, &schema_metadata_manager) - .await - .unwrap_or_else(|e| { - warn!(e; "Failed to get ttl for region: {}", region_metadata.region_id); - ( - crate::region::options::CompactionOptions::default(), - TimeToLive::default(), - ) - }); - ttl - } - }; + let ttl = match ttl_provider { + // Use the specified ttl. + Either::Left(ttl) => ttl, + // Get the ttl from the schema metadata manager. + Either::Right(schema_metadata_manager) => { + let (_, ttl) = find_dynamic_options( + req.region_id, + &req.region_options, + &schema_metadata_manager, + ) + .await + .unwrap_or_else(|e| { + warn!(e; "Failed to get ttl for region: {}", region_metadata.region_id); + ( + crate::region::options::CompactionOptions::default(), + TimeToLive::default(), + ) + }); + ttl + } + }; - Ok(CompactionRegion { - region_id: req.region_id, - region_options: req.region_options.clone(), - engine_config: Arc::new(mito_config.clone()), - region_metadata: region_metadata.clone(), - cache_manager: Arc::new(CacheManager::default()), - access_layer, - manifest_ctx, - current_version, - file_purger: Some(file_purger), - ttl: Some(ttl), - max_parallelism: req.max_parallelism, - plugins: req.plugins.clone(), - }) + Ok(CompactionRegion { + region_id: req.region_id, + region_options: req.region_options.clone(), + engine_config: self.mito_config.clone(), + region_metadata: region_metadata.clone(), + cache_manager: Arc::new(CacheManager::default()), + access_layer, + manifest_ctx, + current_version, + file_purger: Some(file_purger), + ttl: Some(ttl), + max_parallelism: req.max_parallelism, + plugins: req.plugins.clone(), + }) + } } impl CompactionRegion { diff --git a/src/mito2/src/engine/open_test.rs b/src/mito2/src/engine/open_test.rs index 126dde56697..13915dc1ef7 100644 --- a/src/mito2/src/engine/open_test.rs +++ b/src/mito2/src/engine/open_test.rs @@ -33,7 +33,7 @@ use store_api::region_request::{ use store_api::storage::{RegionId, ScanRequest}; use tokio::sync::{Notify, oneshot}; -use crate::compaction::compactor::{OpenCompactionRegionRequest, open_compaction_region}; +use crate::compaction::compactor::{CompactionRegionOpener, OpenCompactionRegionRequest}; use crate::config::MitoConfig; use crate::engine::flush_test::MockRegionHook; use crate::engine::listener::EventListener; @@ -1037,16 +1037,139 @@ async fn test_open_compaction_region_with_format(flat_format: bool) { plugins: Plugins::new(), }; - let compaction_region = open_compaction_region( - &req, - &mito_config, - object_store_manager.clone(), - Either::Right(schema_metadata_manager), - ) - .await - .unwrap(); + mito_config.index.aux_path = env + .data_home() + .join("compactor-index") + .display() + .to_string(); + let opener = CompactionRegionOpener::new(mito_config, object_store_manager) + .await + .unwrap(); + let compaction_region = opener + .open(&req, Either::Right(schema_metadata_manager)) + .await + .unwrap(); assert_eq!(region_id, compaction_region.region_id); + compaction_region.stop_purger_scheduler().await.unwrap(); +} + +// Regression: per-job initialization used to delete another compaction's live +// atomic-write, external-sort, and staging files under the shared aux_path. +#[tokio::test] +async fn test_compaction_region_opens_preserve_live_index_files() { + use common_time::TimeToLive; + use futures::{AsyncReadExt, AsyncWriteExt}; + use index::external_provider::ExternalTempFileProvider; + use store_api::storage::FileId; + + use crate::sst::index::intermediate::{IntermediateLocation, TempFileProvider}; + + let mut env = TestEnv::new().await; + let engine = env.create_engine(MitoConfig::default()).await; + let mut requests = Vec::new(); + for (index, path_type) in [PathType::Bare, PathType::Data].into_iter().enumerate() { + let region_id = RegionId::new(100 + index as u32, 0); + let mut create = CreateRequestBuilder::new() + .table_dir(&format!("data/{}/", region_id.table_id())) + .build(); + create.path_type = path_type; + engine + .handle_request(region_id, RegionRequest::Create(create.clone())) + .await + .unwrap(); + put_rows( + &engine, + region_id, + Rows { + schema: rows_schema(&create), + rows: build_rows(0, 3), + }, + ) + .await; + flush_region(&engine, region_id, None).await; + requests.push(OpenCompactionRegionRequest { + region_id, + table_dir: create.table_dir, + path_type, + region_options: RegionOptions::default(), + max_parallelism: 1, + plugins: Plugins::new(), + }); + } + + let mut config = MitoConfig::default(); + config.sanitize(env.data_home().to_str().unwrap()).unwrap(); + let aux_path = env.data_home().join("compactor-index"); + config.index.aux_path = aux_path.display().to_string(); + let opener = CompactionRegionOpener::new(config, env.get_object_store_manager().unwrap()) + .await + .unwrap(); + let first = opener + .open(&requests[0], Either::Left(TimeToLive::default())) + .await + .unwrap(); + let provider = TempFileProvider::new( + IntermediateLocation::new(&first.region_id, &FileId::random()), + first.access_layer.intermediate_manager().clone(), + ); + let mut committed = provider.create("sort", "committed").await.unwrap(); + committed.write_all(b"committed sort input").await.unwrap(); + committed.close().await.unwrap(); + let mut pending = provider.create("sort", "pending").await.unwrap(); + pending.write_all(b"pending sort input").await.unwrap(); + pending.flush().await.unwrap(); + let staging_file = aux_path.join("staging/inflight.tmp"); + tokio::fs::write(&staging_file, b"staging input") + .await + .unwrap(); + + let cloned_opener = opener.clone(); + let (second, reopened) = tokio::try_join!( + opener.open(&requests[1], Either::Left(TimeToLive::default())), + cloned_opener.open(&requests[0], Either::Left(TimeToLive::default())), + ) + .unwrap(); + assert_eq!(second.access_layer.path_type(), PathType::Data); + assert_eq!(reopened.access_layer.path_type(), PathType::Bare); + let missing = OpenCompactionRegionRequest { + region_id: RegionId::new(999, 0), + table_dir: "missing/".into(), + ..requests[0].clone() + }; + assert!(matches!( + opener + .open(&missing, Either::Left(TimeToLive::default())) + .await, + Err(error::Error::EmptyRegionDir { .. }) + )); + drop(cloned_opener); + drop(opener); + + // Open regions retain the resources even after the opener itself is dropped. + pending.close().await.unwrap(); + let mut contents = Vec::new(); + for (_, mut reader) in provider.read_all("sort").await.unwrap() { + let mut bytes = Vec::new(); + reader.read_to_end(&mut bytes).await.unwrap(); + contents.push(bytes); + } + contents.sort(); + assert_eq!( + contents, + vec![ + b"committed sort input".to_vec(), + b"pending sort input".to_vec() + ] + ); + assert_eq!( + tokio::fs::read(&staging_file).await.unwrap(), + b"staging input" + ); + provider.cleanup().await.unwrap(); + for region in [first, second, reopened] { + region.stop_purger_scheduler().await.unwrap(); + } } /// Returns a dummy fetcher that returns the expr_json only once. diff --git a/src/mito2/src/engine/region_hook.rs b/src/mito2/src/engine/region_hook.rs index 90a15c6aa60..c8b736d2894 100644 --- a/src/mito2/src/engine/region_hook.rs +++ b/src/mito2/src/engine/region_hook.rs @@ -62,7 +62,7 @@ //! | Operations during staging | N/A | ❌ (staging dir) | //! | Async index build | ❌ (index files only) | ✅ Yes | //! -//! ¹ Remote compaction runs on a dedicated compactor node via `open_compaction_region()`; +//! ¹ Remote compaction runs on a dedicated compactor node via `CompactionRegionOpener::open()`; //! pass plugins via `OpenCompactionRegionRequest` to enable hooks there. `sst_infos` is //! `#[serde(skip)]` over the wire, so each [`SstInfo`] is rebuilt from [`FileMeta`] with //! empty footer/index — see [`SstFileInfo`] for field-level detail. @@ -328,7 +328,7 @@ pub trait RegionHook: Send + Sync + Debug { /// Fires once when a region becomes active via a create or open request — /// the natural counterpart to [`on_region_closed`] / [`on_region_dropped`]. /// It does **not** fire for the compactor's transient compaction regions - /// (`open_compaction_region`), nor for the internal reopen performed during + /// (`CompactionRegionOpener::open`), nor for the internal reopen performed during /// follower catch-up / leadership promotion. /// /// On the **create** path it runs inline in the region worker loop; on the