From a23fe1c2dc958e49f849e92f1bec82ed218ff764 Mon Sep 17 00:00:00 2001 From: Yingwen Date: Mon, 14 Sep 2026 09:21:31 +0000 Subject: [PATCH] fix: configure series indexes with an enable flag (#9141) * fix: use join_dir for series index config path Signed-off-by: evenyag * fix: configure series indexes with an enable flag Signed-off-by: evenyag * docs: omit experimental series index from example configs Signed-off-by: evenyag * fix: preserve legacy cache cleanup path behavior Signed-off-by: evenyag * test: remove trivial path joining tests Signed-off-by: evenyag * test: isolate worker group WAL directories on Windows Signed-off-by: evenyag --------- Signed-off-by: evenyag --- src/mito2/src/access_layer.rs | 44 ++++++++++++++++++-- src/mito2/src/config.rs | 49 ++-------------------- src/mito2/src/engine.rs | 2 + src/mito2/src/test_util.rs | 2 + src/mito2/src/worker.rs | 74 +++++++++++++++++++++++---------- tests-integration/tests/http.rs | 2 +- 6 files changed, 102 insertions(+), 71 deletions(-) diff --git a/src/mito2/src/access_layer.rs b/src/mito2/src/access_layer.rs index 8580884274..14c629a01b 100644 --- a/src/mito2/src/access_layer.rs +++ b/src/mito2/src/access_layer.rs @@ -12,6 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. +use std::path::Path; use std::sync::Arc; use std::time::{Duration, Instant}; @@ -655,14 +656,17 @@ impl TempFileCleaner { } pub(crate) async fn new_fs_cache_store(root: &str) -> Result { - let atomic_write_dir = join_dir(root, ATOMIC_WRITE_DIR); - clean_dir(&atomic_write_dir).await?; + // Preserve native filesystem prefixes such as Windows UNC shares. + let atomic_write_dir = Path::new(root).join(ATOMIC_WRITE_DIR); + clean_dir(&atomic_write_dir.to_string_lossy()).await?; // Compatible code. Remove this after a major release. let old_atomic_temp_dir = join_dir(root, OLD_ATOMIC_WRITE_DIR); clean_dir(&old_atomic_temp_dir).await?; - let builder = Fs::default().root(root).atomic_write_dir(&atomic_write_dir); + let builder = Fs::default() + .root(root) + .atomic_write_dir(&atomic_write_dir.to_string_lossy()); let store = ObjectStore::new(builder).context(OpenDalSnafu)?; Ok(with_instrument_layers(store, false)) @@ -759,3 +763,37 @@ impl FilePathProvider for RegionFilePathFactory { location::sst_file_path(&self.table_dir, file_id, self.path_type) } } + +#[cfg(test)] +mod tests { + use std::path::PathBuf; + + use super::*; + + #[tokio::test] + async fn test_new_fs_cache_store() { + let root = common_test_util::temp_dir::create_temp_dir("fs-cache-store"); + let dirs = [ + root.path().join(ATOMIC_WRITE_DIR), + PathBuf::from(join_dir( + root.path().to_str().unwrap(), + OLD_ATOMIC_WRITE_DIR, + )), + ]; + for dir in &dirs { + tokio::fs::create_dir_all(dir).await.unwrap(); + tokio::fs::write(dir.join("stale"), b"stale").await.unwrap(); + } + let store = new_fs_cache_store(root.path().to_str().unwrap()) + .await + .unwrap(); + for dir in &dirs { + assert!(!dir.join("stale").exists()); + } + store.write("index", "contents").await.unwrap(); + assert_eq!( + tokio::fs::read(root.path().join("index")).await.unwrap(), + b"contents" + ); + } +} diff --git a/src/mito2/src/config.rs b/src/mito2/src/config.rs index 6ec790c52b..667f156fb2 100644 --- a/src/mito2/src/config.rs +++ b/src/mito2/src/config.rs @@ -89,11 +89,9 @@ pub struct MitoConfig { // Background job configs: /// Max number of running background index build jobs (default: 1/8 of cpu cores). pub max_background_index_builds: usize, - // TODO: Document both series-index settings in the example configs and regenerate configuration - // docs before exposing the feature. - /// Under development; do not enable. Root directory for loading series indexes, currently stored - /// on the local filesystem. Empty disables the feature. Relative paths resolve under `data_home`. - pub experimental_series_index_root: String, + /// Under development; do not enable. Whether to enable series indexes (default false). + /// Indexes are stored on the local filesystem under `{data_home}/series_index`. + pub experimental_enable_series_index: bool, /// Interval between series-index maintenance runs (default 5 min). Zero uses the default. #[serde(with = "humantime_serde")] pub experimental_series_index_maintenance_interval: Duration, @@ -214,7 +212,7 @@ impl Default for MitoConfig { experimental_manifest_keep_removed_file_ttl: Duration::from_secs(60 * 60), compress_manifest: false, max_background_index_builds: divide_num_cpus(8), - experimental_series_index_root: String::new(), + experimental_enable_series_index: false, experimental_series_index_maintenance_interval: DEFAULT_SERIES_INDEX_MAINTENANCE_INTERVAL, max_background_flushes: divide_num_cpus(2), @@ -312,15 +310,6 @@ impl MitoConfig { DEFAULT_SERIES_INDEX_MAINTENANCE_INTERVAL; } - if !self.experimental_series_index_root.trim().is_empty() - && Path::new(&self.experimental_series_index_root).is_relative() - { - self.experimental_series_index_root = Path::new(data_home) - .join(&self.experimental_series_index_root) - .display() - .to_string(); - } - if self.global_write_buffer_reject_size <= self.global_write_buffer_size { self.global_write_buffer_reject_size = self.global_write_buffer_size * 2; warn!( @@ -432,36 +421,6 @@ mod tests { assert_eq!(ReadableSize::mb(128), config.prefilter_result_cache_size); } - #[test] - fn test_series_index_config() { - assert!( - MitoConfig::default() - .experimental_series_index_root - .is_empty() - ); - let mut config: MitoConfig = toml::from_str( - "experimental_series_index_root = 'indexes' - experimental_series_index_maintenance_interval = '30s'", - ) - .unwrap(); - config.sanitize("/data").unwrap(); - assert_eq!(config.experimental_series_index_root, "/data/indexes"); - assert_eq!( - config.experimental_series_index_maintenance_interval, - Duration::from_secs(30) - ); - let restored: MitoConfig = toml::from_str(&toml::to_string(&config).unwrap()).unwrap(); - assert_eq!(config, restored); - - let mut config: MitoConfig = - toml::from_str("experimental_series_index_maintenance_interval = '0s'").unwrap(); - config.sanitize("/data").unwrap(); - assert_eq!( - config.experimental_series_index_maintenance_interval, - MitoConfig::default().experimental_series_index_maintenance_interval - ); - } - #[test] fn test_experimental_series_scan_v2_config() { assert!(MitoConfig::default().experimental_series_scan_v2); diff --git a/src/mito2/src/engine.rs b/src/mito2/src/engine.rs index 1c674abadb..348ec4c122 100644 --- a/src/mito2/src/engine.rs +++ b/src/mito2/src/engine.rs @@ -225,6 +225,7 @@ impl<'a, S: LogStore> MitoEngineBuilder<'a, S> { // so the engine (and thus the GC worker) can fire `on_region_gc`. let region_hook = self.plugins.get::(); let workers = WorkerGroup::start( + self.data_home, config.clone(), self.log_store.clone(), self.object_store_manager, @@ -1620,6 +1621,7 @@ impl MitoEngine { Ok(MitoEngine { inner: Arc::new(EngineInner { workers: WorkerGroup::start_for_test( + data_home, config.clone(), log_store, object_store_manager, diff --git a/src/mito2/src/test_util.rs b/src/mito2/src/test_util.rs index aa03e4bc84..dbfbc53db2 100644 --- a/src/mito2/src/test_util.rs +++ b/src/mito2/src/test_util.rs @@ -682,6 +682,7 @@ impl TestEnv { match log_store { LogStoreImpl::RaftEngine(log_store) => WorkerGroup::start( + &data_home, Arc::new(config), log_store, Arc::new(object_store_manager), @@ -693,6 +694,7 @@ impl TestEnv { .await .unwrap(), LogStoreImpl::Kafka(log_store) => WorkerGroup::start( + &data_home, Arc::new(config), log_store, Arc::new(object_store_manager), diff --git a/src/mito2/src/worker.rs b/src/mito2/src/worker.rs index 8efcf47ab6..0483217f96 100644 --- a/src/mito2/src/worker.rs +++ b/src/mito2/src/worker.rs @@ -169,7 +169,9 @@ impl WorkerGroup { /// Starts a worker group. /// /// The number of workers should be power of two. + #[allow(clippy::too_many_arguments)] pub(crate) async fn start( + data_home: &str, config: Arc, log_store: Arc, object_store_manager: ObjectStoreManagerRef, @@ -195,11 +197,7 @@ impl WorkerGroup { .with_buffer_size(Some(config.index.write_buffer_size.as_bytes() as _)); let index_build_job_pool = Arc::new(LocalScheduler::new(config.max_background_index_builds)); - let series_index_store = if config.experimental_series_index_root.trim().is_empty() { - None - } else { - Some(new_fs_cache_store(&config.experimental_series_index_root).await?) - }; + let series_index_store = series_index_store_from_config(&config, data_home).await?; let flush_job_pool = Arc::new(LocalScheduler::new(config.max_background_flushes)); let compact_job_pool = Arc::new(LocalScheduler::new(config.max_background_compactions)); let flush_semaphore = Arc::new(Semaphore::new(config.max_background_flushes)); @@ -391,6 +389,7 @@ impl WorkerGroup { /// The number of workers should be power of two. #[allow(clippy::too_many_arguments)] pub(crate) async fn start_for_test( + data_home: &str, config: Arc, log_store: Arc, object_store_manager: ObjectStoreManagerRef, @@ -410,11 +409,7 @@ impl WorkerGroup { }); let index_build_job_pool = Arc::new(LocalScheduler::new(config.max_background_index_builds)); - let series_index_store = if config.experimental_series_index_root.trim().is_empty() { - None - } else { - Some(new_fs_cache_store(&config.experimental_series_index_root).await?) - }; + let series_index_store = series_index_store_from_config(&config, data_home).await?; let flush_job_pool = Arc::new(LocalScheduler::new(config.max_background_flushes)); let compact_job_pool = Arc::new(LocalScheduler::new(config.max_background_compactions)); let flush_semaphore = Arc::new(Semaphore::new(config.max_background_flushes)); @@ -517,6 +512,19 @@ fn region_id_to_index(id: RegionId, num_workers: usize) -> usize { % num_workers } +/// Opens the fixed local series-index store only when the feature is enabled. +async fn series_index_store_from_config( + config: &MitoConfig, + data_home: &str, +) -> Result> { + if !config.experimental_enable_series_index { + return Ok(None); + } + + let root = Path::new(data_home).join("series_index"); + new_fs_cache_store(&root.to_string_lossy()).await.map(Some) +} + pub async fn write_cache_from_config( config: &MitoConfig, puffin_manager_factory: PuffinManagerFactory, @@ -1620,18 +1628,40 @@ mod tests { #[tokio::test] async fn test_worker_group_start_stop() { - let env = TestEnv::with_prefix("group-stop").await; - let group = env - .create_worker_group(MitoConfig { - num_workers: 4, - experimental_series_index_root: "series-index".to_string(), - ..Default::default() - }) - .await; + for (enabled, existing_index) in [(false, false), (true, false), (false, true)] { + // Use a fresh WAL directory for each case: stopping workers does not stop the log store. + let env = TestEnv::with_prefix("group-stop").await; + let root = env.data_home().join("series_index"); + if existing_index { + tokio::fs::create_dir_all(&root).await.unwrap(); + tokio::fs::write(root.join("retained"), b"index") + .await + .unwrap(); + } + let group = env + .create_worker_group(MitoConfig { + num_workers: 4, + experimental_enable_series_index: enabled, + ..Default::default() + }) + .await; - tokio::time::timeout(Duration::from_secs(5), group.stop()) - .await - .expect("series-index tasks should stop without waiting for their interval") - .unwrap(); + for worker in &group.workers { + assert_eq!(worker.series_index_task_state.is_some(), enabled); + assert_eq!(worker.series_index_handle.lock().await.is_some(), enabled); + } + assert_eq!(root.is_dir(), enabled || existing_index); + if existing_index { + assert_eq!( + tokio::fs::read(root.join("retained")).await.unwrap(), + b"index" + ); + } + + tokio::time::timeout(Duration::from_secs(5), group.stop()) + .await + .expect("series-index tasks should stop without waiting for their interval") + .unwrap(); + } } } diff --git a/tests-integration/tests/http.rs b/tests-integration/tests/http.rs index 84d1f7d957..da4951a92e 100644 --- a/tests-integration/tests/http.rs +++ b/tests-integration/tests/http.rs @@ -2381,7 +2381,7 @@ manifest_checkpoint_distance = 10 experimental_manifest_keep_removed_file_count = 256 experimental_manifest_keep_removed_file_ttl = "1h" compress_manifest = false -experimental_series_index_root = "" +experimental_enable_series_index = false experimental_series_index_maintenance_interval = "5m" experimental_compaction_memory_limit = "unlimited" experimental_compaction_on_exhausted = "wait"