refactor: consolidate RegionManifestOptions creation logic

Signed-off-by: WenyXu <[email protected]>

Author: WenyXu

src/mito2/src/compaction/compactor.rs

@@ -41,11 +41,9 @@ use crate::error::{
     EmptyRegionDirSnafu, InvalidPartitionExprSnafu, JoinSnafu, ObjectStoreNotFoundSnafu, Result,
 };
 use crate::manifest::action::{RegionEdit, RegionMetaAction, RegionMetaActionList};
-use crate::manifest::manager::{RegionManifestManager, RegionManifestOptions, RemoveFileOptions};
-use crate::manifest::storage::manifest_compress_type;
+use crate::manifest::manager::{RegionManifestManager, RegionManifestOptions};
 use crate::metrics;
 use crate::read::{FlatSource, Source};
-use crate::region::opener::new_manifest_dir;
 use crate::region::options::RegionOptions;
 use crate::region::version::VersionRef;
 use crate::region::{ManifestContext, RegionLeaderState, RegionRoleState};
@@ -162,19 +160,9 @@ pub async fn open_compaction_region(
     };
 
     let manifest_manager = {
-        let region_manifest_options = RegionManifestOptions {
-            manifest_dir: new_manifest_dir(&region_dir_from_table_dir(
-                &req.table_dir,
-                req.region_id,
-                req.path_type,
-            )),
-            object_store: object_store.clone(),
-            compress_type: manifest_compress_type(mito_config.compress_manifest),
-            checkpoint_distance: mito_config.manifest_checkpoint_distance,
-            remove_file_options: RemoveFileOptions {
-                enable_gc: mito_config.gc.enable,
-            },
-        };
+        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?

src/mito2/src/manifest/manager.rs

@@ -24,6 +24,7 @@ use store_api::metadata::RegionMetadataRef;
 use store_api::storage::FileId;
 use store_api::{MAX_VERSION, MIN_VERSION, ManifestVersion};
 
+use crate::config::MitoConfig;
 use crate::error::{
     self, InstallManifestToSnafu, NoCheckpointSnafu, NoManifestsSnafu, RegionStoppedSnafu, Result,
 };
@@ -33,7 +34,8 @@ use crate::manifest::action::{
 };
 use crate::manifest::checkpointer::Checkpointer;
 use crate::manifest::storage::{
-    ManifestObjectStore, file_version, is_checkpoint_file, is_delta_file,
+    ManifestObjectStore, file_version, is_checkpoint_file, is_delta_file, manifest_compress_type,
+    manifest_dir,
 };
 use crate::metrics::MANIFEST_OP_ELAPSED;
 use crate::region::{ManifestStats, RegionLeaderState, RegionRoleState};
@@ -52,6 +54,23 @@ pub struct RegionManifestOptions {
     pub remove_file_options: RemoveFileOptions,
 }
 
+impl RegionManifestOptions {
+    /// Creates a new [RegionManifestOptions] with the given region directory, object store, and configuration.
+    pub fn new(config: &MitoConfig, region_dir: &str, object_store: &ObjectStore) -> Self {
+        RegionManifestOptions {
+            manifest_dir: manifest_dir(region_dir),
+            object_store: object_store.clone(),
+            // We don't allow users to set the compression algorithm as we use it as a file suffix.
+            // Currently, the manifest storage doesn't have good support for changing compression algorithms.
+            compress_type: manifest_compress_type(config.compress_manifest),
+            checkpoint_distance: config.manifest_checkpoint_distance,
+            remove_file_options: RemoveFileOptions {
+                enable_gc: config.gc.enable,
+            },
+        }
+    }
+}
+
 /// Options for updating `removed_files` field in [RegionManifest].
 #[derive(Debug, Clone)]
 #[cfg_attr(any(test, feature = "test"), derive(Default))]

src/mito2/src/manifest/storage.rs

@@ -24,6 +24,7 @@ use crc32fast::Hasher;
 use futures::TryStreamExt;
 use futures::future::try_join_all;
 use lazy_static::lazy_static;
+use object_store::util::join_dir;
 use object_store::{Entry, ErrorKind, Lister, ObjectStore, util};
 use regex::Regex;
 use serde::{Deserialize, Serialize};
@@ -49,6 +50,11 @@ const DEFAULT_MANIFEST_COMPRESSION_TYPE: CompressionType = CompressionType::Gzip
 const FALL_BACK_COMPRESS_TYPE: CompressionType = CompressionType::Uncompressed;
 const FETCH_MANIFEST_PARALLELISM: usize = 16;
 
+/// Returns the directory to the manifest files.
+pub fn manifest_dir(region_dir: &str) -> String {
+    join_dir(region_dir, "manifest")
+}
+
 /// Returns the [CompressionType] according to whether to compress manifest files.
 pub const fn manifest_compress_type(compress: bool) -> CompressionType {
     if compress {

src/mito2/src/region/opener.rs

@@ -28,7 +28,7 @@ use log_store::kafka::log_store::KafkaLogStore;
 use log_store::noop::log_store::NoopLogStore;
 use log_store::raft_engine::log_store::RaftEngineLogStore;
 use object_store::manager::ObjectStoreManagerRef;
-use object_store::util::{join_dir, normalize_dir};
+use object_store::util::normalize_dir;
 use snafu::{OptionExt, ResultExt, ensure};
 use store_api::logstore::LogStore;
 use store_api::logstore::provider::Provider;
@@ -49,8 +49,7 @@ use crate::error::{
     Result, StaleLogEntrySnafu,
 };
 use crate::manifest::action::RegionManifest;
-use crate::manifest::manager::{RegionManifestManager, RegionManifestOptions, RemoveFileOptions};
-use crate::manifest::storage::manifest_compress_type;
+use crate::manifest::manager::{RegionManifestManager, RegionManifestOptions};
 use crate::memtable::MemtableBuilderProvider;
 use crate::memtable::bulk::part::BulkPart;
 use crate::memtable::time_partition::{TimePartitions, TimePartitionsRef};
@@ -272,7 +271,7 @@ impl RegionOpener {
         };
         // Create a manifest manager for this region and writes regions to the manifest file.
         let region_manifest_options =
-            Self::manifest_options(config, &options, &region_dir, &self.object_store_manager)?;
+            RegionManifestOptions::new(config, &region_dir, &object_store);
         // For remote WAL, we need to set flushed_entry_id to current topic's latest entry id.
         let flushed_entry_id = provider.initial_flushed_entry_id::<S>(wal.store());
         let manifest_manager = RegionManifestManager::new(
@@ -406,13 +405,9 @@ impl RegionOpener {
     ) -> Result<Option<MitoRegionRef>> {
         let now = Instant::now();
         let mut region_options = self.options.as_ref().unwrap().clone();
-
-        let region_manifest_options = Self::manifest_options(
-            config,
-            &region_options,
-            &self.region_dir(),
-            &self.object_store_manager,
-        )?;
+        let object_storage = get_object_store(&region_options.storage, &self.object_store_manager)?;
+        let region_manifest_options =
+            RegionManifestOptions::new(config, &self.region_dir(), &object_storage);
         let Some(manifest_manager) =
             RegionManifestManager::open(region_manifest_options, &self.stats).await?
         else {
@@ -576,27 +571,6 @@ impl RegionOpener {
 
         Ok(Some(region))
     }
-
-    /// Returns a new manifest options.
-    fn manifest_options(
-        config: &MitoConfig,
-        options: &RegionOptions,
-        region_dir: &str,
-        object_store_manager: &ObjectStoreManagerRef,
-    ) -> Result<RegionManifestOptions> {
-        let object_store = get_object_store(&options.storage, object_store_manager)?;
-        Ok(RegionManifestOptions {
-            manifest_dir: new_manifest_dir(region_dir),
-            object_store,
-            // We don't allow users to set the compression algorithm as we use it as a file suffix.
-            // Currently, the manifest storage doesn't have good support for changing compression algorithms.
-            compress_type: manifest_compress_type(config.compress_manifest),
-            checkpoint_distance: config.manifest_checkpoint_distance,
-            remove_file_options: RemoveFileOptions {
-                enable_gc: config.gc.enable,
-            },
-        })
-    }
 }
 
 /// Creates a version builder from a region manifest.
@@ -678,12 +652,9 @@ impl RegionMetadataLoader {
         region_dir: &str,
         region_options: &RegionOptions,
     ) -> Result<Option<Arc<RegionManifest>>> {
-        let region_manifest_options = RegionOpener::manifest_options(
-            &self.config,
-            region_options,
-            region_dir,
-            &self.object_store_manager,
-        )?;
+        let object_store = get_object_store(&region_options.storage, &self.object_store_manager)?;
+        let region_manifest_options =
+            RegionManifestOptions::new(&self.config, region_dir, &object_store);
         let Some(manifest_manager) =
             RegionManifestManager::open(region_manifest_options, &Default::default()).await?
         else {
@@ -848,11 +819,6 @@ where
     Ok(last_entry_id)
 }
 
-/// Returns the directory to the manifest files.
-pub(crate) fn new_manifest_dir(region_dir: &str) -> String {
-    join_dir(region_dir, "manifest")
-}
-
 /// A task to load and fill the region file cache.
 pub(crate) struct RegionLoadCacheTask {
     region: MitoRegionRef,