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

fix(mito2): preserve live index files across remote compactions (#10046)

Browse Source 
fix(mito2): reuse executor-owned compaction index resources

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>
This commit is contained in:
Lei, HUANG
2026-10-29 17:58:37 +08:00
committed by GitHub
parent 694ca48517
commit c7ea80558f
3 changed files with 266 additions and 109 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/mito2/src/compaction/compactor.rs
+132 -98
View File
@@ -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<MitoConfig>,
object_store_manager: ObjectStoreManagerRef,
ttl_provider: Either<TimeToLive, SchemaMetadataManagerRef>,
) -> Result<CompactionRegion> {
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<Self> {
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<TimeToLive, SchemaMetadataManagerRef>,
) -> Result<CompactionRegion> {
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, &region_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, &region_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<RegionHookRef> = 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<RegionHookRef> = 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 {
src/mito2/src/engine/open_test.rs
+132 -9
View File
@@ -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.
src/mito2/src/engine/region_hook.rs
+2 -2
View File
@@ -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