rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-09-21 20:55:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: configure series indexes with an enable flag (#9141)

Browse Source 
* fix: use join_dir for series index config path

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

* fix: configure series indexes with an enable flag

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

* docs: omit experimental series index from example configs

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

* fix: preserve legacy cache cleanup path behavior

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

* test: remove trivial path joining tests

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

* test: isolate worker group WAL directories on Windows

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

---------

Signed-off-by: evenyag <realevenyag@gmail.com>
This commit is contained in:
Yingwen
2026-09-14 09:21:31 +00:00
committed by GitHub
parent 577de012b2
commit a23fe1c2dc
6 changed files with 102 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/mito2/src/access_layer.rs
+41 -3
View File
@@ -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<ObjectStore> {
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"
);
}
}
src/mito2/src/config.rs
+4 -45
View File
@@ -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);
src/mito2/src/engine.rs
+2
View File
@@ -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
+2
View File
@@ -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
+52 -22
View File
@@ -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,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<Option<ObjectStore>> {
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();
}
}
}
tests-integration/tests/http.rs
+1 -1
View File
@@ -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"