fix: configure series indexes with an enable flag

Signed-off-by: evenyag <realevenyag@gmail.com>

config/datanode.example.toml

@@ -474,6 +474,13 @@ experimental_manifest_keep_removed_file_ttl = "1h"
 ## Whether to compress manifest and checkpoint file by gzip (default false).
 compress_manifest = false
 
+## Under development; do not enable. Series indexes are stored on the local filesystem
+## under `{data_home}/series_index` when enabled.
+experimental_enable_series_index = false
+
+## Interval between series-index maintenance runs. Zero uses the default of 5 minutes.
+experimental_series_index_maintenance_interval = "5m"
+
 ## Max number of running background flush jobs (default: 1/2 of cpu cores).
 ## @toml2docs:none-default="Auto"
 #+ max_background_flushes = 4

config/standalone.example.toml

@@ -635,6 +635,13 @@ manifest_checkpoint_distance = 10
 ## Whether to compress manifest and checkpoint file by gzip (default false).
 compress_manifest = false
 
+## Under development; do not enable. Series indexes are stored on the local filesystem
+## under `{data_home}/series_index` when enabled.
+experimental_enable_series_index = false
+
+## Interval between series-index maintenance runs. Zero uses the default of 5 minutes.
+experimental_series_index_maintenance_interval = "5m"
+
 ## Max number of running background flush jobs (default: 1/2 of cpu cores).
 ## @toml2docs:none-default="Auto"
 #+ max_background_flushes = 4

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, PathBuf};
 use std::sync::Arc;
 use std::time::{Duration, Instant};
 
@@ -20,7 +21,7 @@ use common_telemetry::warn;
 use common_time::Timestamp;
 use futures::{Stream, TryStreamExt};
 use object_store::services::Fs;
-use object_store::util::{join_dir, with_instrument_layers};
+use object_store::util::with_instrument_layers;
 use object_store::{ATOMIC_WRITE_DIR, ErrorKind, OLD_ATOMIC_WRITE_DIR, ObjectStore};
 use parquet::file::metadata::PageIndexPolicy;
 use smallvec::SmallVec;
@@ -655,19 +656,26 @@ impl TempFileCleaner {
 }
 
 pub(crate) async fn new_fs_cache_store(root: &str) -> Result<ObjectStore> {
-    let atomic_write_dir = join_dir(root, ATOMIC_WRITE_DIR);
-    clean_dir(&atomic_write_dir).await?;
+    let [atomic_write_dir, old_atomic_temp_dir] = fs_cache_temp_dirs(root);
+    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?;
+    clean_dir(&old_atomic_temp_dir.to_string_lossy()).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))
 }
 
+// These are filesystem paths, so preserve native prefixes such as Windows UNC shares.
+fn fs_cache_temp_dirs(root: &str) -> [PathBuf; 2] {
+    let root = Path::new(root);
+    [root.join(ATOMIC_WRITE_DIR), root.join(OLD_ATOMIC_WRITE_DIR)]
+}
+
 /// Clean the directory.
 async fn clean_dir(dir: &str) -> Result<()> {
     if tokio::fs::try_exists(dir)
@@ -759,3 +767,51 @@ impl FilePathProvider for RegionFilePathFactory {
         location::sst_file_path(&self.table_dir, file_id, self.path_type)
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[cfg(windows)]
+    #[test]
+    fn test_fs_cache_temp_dirs_windows() {
+        for (root, expected_root) in [
+            (r"C:\data", r"C:\data"),
+            ("C:/data/", r"C:\data"),
+            ("//server/share/data", r"\\server\share\data"),
+            (r"\\server\share\data", r"\\server\share\data"),
+        ] {
+            for (dir, name) in fs_cache_temp_dirs(root)
+                .into_iter()
+                .zip([ATOMIC_WRITE_DIR, OLD_ATOMIC_WRITE_DIR])
+            {
+                assert_eq!(dir.parent().unwrap(), Path::new(expected_root));
+                assert_eq!(
+                    dir.file_name().unwrap(),
+                    Path::new(name).file_name().unwrap()
+                );
+            }
+        }
+    }
+
+    #[tokio::test]
+    async fn test_new_fs_cache_store() {
+        let root = common_test_util::temp_dir::create_temp_dir("fs-cache-store");
+        for name in [ATOMIC_WRITE_DIR, OLD_ATOMIC_WRITE_DIR] {
+            let dir = root.path().join(name);
+            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 name in [ATOMIC_WRITE_DIR, OLD_ATOMIC_WRITE_DIR] {
+            assert!(!root.path().join(name).join("stale").exists());
+        }
+        store.write("index", "contents").await.unwrap();
+        assert_eq!(
+            tokio::fs::read(root.path().join("index")).await.unwrap(),
+            b"contents"
+        );
+    }
+}

src/mito2/src/config.rs

@@ -23,7 +23,6 @@ use common_base::readable_size::ReadableSize;
 use common_memory_manager::OnExhaustedPolicy;
 use common_stat::{get_total_cpu_cores, get_total_memory_readable};
 use common_telemetry::warn;
-use object_store::util::join_dir;
 use serde::{Deserialize, Serialize};
 use serde_with::serde_as;
 
@@ -90,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,
@@ -215,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),
@@ -313,13 +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 =
-                join_dir(data_home, &self.experimental_series_index_root);
-        }
-
         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!(
@@ -433,18 +423,14 @@ mod tests {
 
     #[test]
     fn test_series_index_config() {
-        assert!(
-            MitoConfig::default()
-                .experimental_series_index_root
-                .is_empty()
-        );
+        assert!(!MitoConfig::default().experimental_enable_series_index);
         let mut config: MitoConfig = toml::from_str(
-            "experimental_series_index_root = 'indexes'
+            "experimental_enable_series_index = true
              experimental_series_index_maintenance_interval = '30s'",
         )
         .unwrap();
         config.sanitize("/data").unwrap();
-        assert_eq!(config.experimental_series_index_root, "/data/indexes/");
+        assert!(config.experimental_enable_series_index);
         assert_eq!(
             config.experimental_series_index_maintenance_interval,
             Duration::from_secs(30)
@@ -452,6 +438,10 @@ mod tests {
         let restored: MitoConfig = toml::from_str(&toml::to_string(&config).unwrap()).unwrap();
         assert_eq!(config, restored);
 
+        let disabled: MitoConfig =
+            toml::from_str("experimental_enable_series_index = false").unwrap();
+        assert!(!disabled.experimental_enable_series_index);
+
         let mut config: MitoConfig =
             toml::from_str("experimental_series_index_maintenance_interval = '0s'").unwrap();
         config.sanitize("/data").unwrap();

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::<RegionHookRef>();
         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,

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),

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<S: LogStore>(
+        data_home: &str,
         config: Arc<MitoConfig>,
         log_store: Arc<S>,
         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<S: LogStore>(
+        data_home: &str,
         config: Arc<MitoConfig>,
         log_store: Arc<S>,
         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,23 @@ 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<Option<ObjectStore>> {
+    if !config.experimental_enable_series_index {
+        return Ok(None);
+    }
+
+    let root = series_index_dir(data_home);
+    new_fs_cache_store(&root.to_string_lossy()).await.map(Some)
+}
+
+fn series_index_dir(data_home: &str) -> std::path::PathBuf {
+    Path::new(data_home).join("series_index")
+}
+
 pub async fn write_cache_from_config(
     config: &MitoConfig,
     puffin_manager_factory: PuffinManagerFactory,
@@ -1618,20 +1630,58 @@ mod tests {
         assert_eq!(index, 1);
     }
 
+    #[test]
+    fn test_series_index_dir() {
+        assert_eq!(series_index_dir("/data"), Path::new("/data/series_index"));
+        assert_eq!(series_index_dir("/data/"), Path::new("/data/series_index"));
+    }
+
+    #[cfg(windows)]
+    #[test]
+    fn test_series_index_dir_windows() {
+        for (data_home, expected) in [
+            (r"C:\data", r"C:\data\series_index"),
+            ("C:/data/", r"C:\data\series_index"),
+            ("//server/share/data", r"\\server\share\data\series_index"),
+            (r"\\server\share\data", r"\\server\share\data\series_index"),
+        ] {
+            assert_eq!(series_index_dir(data_home), Path::new(expected));
+        }
+    }
+
     #[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;
+        let root = env.data_home().join("series_index");
+        for (run, enabled) in [false, true, false].into_iter().enumerate() {
+            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(), run > 0);
+            if enabled {
+                tokio::fs::write(root.join("retained"), b"index")
+                    .await
+                    .unwrap();
+            } else if root.exists() {
+                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();
+        }
     }
 }

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"