diff --git a/src/mito2/src/series_index/builder.rs b/src/mito2/src/series_index/builder.rs index 63a51badaeb..447f3d409e0 100644 --- a/src/mito2/src/series_index/builder.rs +++ b/src/mito2/src/series_index/builder.rs @@ -21,6 +21,7 @@ use common_telemetry::warn; use futures::TryStreamExt; use object_store::ObjectStore; use snafu::{OptionExt, ensure}; +use store_api::metadata::RegionMetadataRef; use crate::error::{Result, UnexpectedSnafu}; use crate::read::BoxedRecordBatchStream; @@ -30,7 +31,6 @@ use crate::read::flat_projection::FlatProjectionMapper; use crate::read::prune::FlatPruneReader; use crate::read::read_columns::ReadColumns; use crate::region::MitoRegionRef; -use crate::region::version::VersionRef; use crate::series_index::bucket::SeriesBucket; use crate::series_index::catalog::{ RangeIndexEntry, SeriesIndexEntry, range_index_path, series_index_path, series_metadata, @@ -65,18 +65,18 @@ async fn reader_input( pub(crate) async fn build_range_index( store: &ObjectStore, region: &MitoRegionRef, - version: &VersionRef, + metadata: &RegionMetadataRef, file: FileHandle, ) -> Result> { let file_id = file.file_id().file_id(); let Some((context, mut selection)) = reader_input(region, file).await? else { return Ok(None); }; - let mapper = FlatProjectionMapper::new(&version.metadata, [])?; + let mapper = FlatProjectionMapper::new(metadata, [])?; let compat = FlatCompatBatch::try_new(&mapper, context.read_format(), false)?; let path = range_index_path(region.region_id, file_id); let mut writer = SstRangeIndexWriter::try_new( - version.metadata.clone(), + metadata.clone(), store.clone(), &path, SstRangeIndexWriterOptions::default(), @@ -127,13 +127,13 @@ pub(crate) async fn build_range_index( pub(crate) async fn build_series_index( store: &ObjectStore, region: &MitoRegionRef, - version: &VersionRef, + metadata: &RegionMetadataRef, bucket: &SeriesBucket, entry: &SeriesIndexEntry, purger: &IndexFilePurger, ) -> Result { let mut sources = Vec::::new(); - let mapper = FlatProjectionMapper::new(&version.metadata, [])?; + let mapper = FlatProjectionMapper::new(metadata, [])?; let schema = mapper.input_arrow_schema(false); for file in &bucket.files { let Some((context, mut selection)) = reader_input(region, file.clone()).await? else { @@ -182,7 +182,7 @@ pub(crate) async fn build_series_index( // TODO(yingwen): Deduplicate update-mode rows before series indexes are used by queries. let path = series_index_path(region.region_id, entry.index_uuid); let mut writer = SeriesIndexWriter::try_new( - version.metadata.clone(), + metadata.clone(), store.clone(), &path, SeriesIndexWriterOptions::default(), @@ -226,6 +226,7 @@ mod tests { use store_api::region_engine::RegionEngine; use super::*; + use crate::region::version::VersionRef; use crate::series_index::bucket::{group_files_into_series_buckets, plan_series_indexes}; use crate::series_index::catalog::{ SeriesIndexCatalog, load_version_control, series_catalog_path, store_catalog, @@ -280,7 +281,7 @@ mod tests { let mut range_bytes = HashMap::new(); if build_ranges { for file in files { - let completed = build_range_index(&store, ®ion, &version, file.clone()) + let completed = build_range_index(&store, ®ion, &version.metadata, file.clone()) .await .unwrap() .unwrap(); @@ -318,7 +319,7 @@ mod tests { let result = build_series_index( &store.clone().layer(layer), ®ion, - &version, + &version.metadata, &bucket, &entry, &purger, @@ -466,14 +467,15 @@ mod tests { let store = ObjectStore::new(Memory::default()).unwrap(); let (purger, _receiver) = series_index_channel(store.clone()); for file in &bucket.files { - build_range_index(&store, ®ion, &version, file.clone()) + build_range_index(&store, ®ion, &version.metadata, file.clone()) .await .unwrap() .unwrap(); } - let _handle = build_series_index(&store, ®ion, &version, &bucket, &entry, &purger) - .await - .unwrap(); + let _handle = + build_series_index(&store, ®ion, &version.metadata, &bucket, &entry, &purger) + .await + .unwrap(); let bytes = store .read(&series_index_path(region.region_id, entry.index_uuid)) .await @@ -533,14 +535,14 @@ mod tests { let result: Result> = async { for file in files { let Some(entry) = - build_range_index(&store, ®ion, &version, file.clone()).await? + build_range_index(&store, ®ion, &version.metadata, file.clone()).await? else { // Defer series construction if a needed range is not ready. return Ok(None); }; completed.push(entry.file_id); } - build_series_index(&store, ®ion, &version, &bucket, &entry, &purger) + build_series_index(&store, ®ion, &version.metadata, &bucket, &entry, &purger) .await .map(Some) } @@ -741,12 +743,12 @@ mod tests { .layer(write_layer); fail.store(fail_before_open, Ordering::Relaxed); let result = if !series { - build_range_index(&store, ®ion, &version, bucket.files[0].clone()) + build_range_index(&store, ®ion, &version.metadata, bucket.files[0].clone()) .await .map(|_| ()) } else { let (purger, _receiver) = series_index_channel(store.clone()); - build_series_index(&store, ®ion, &version, &bucket, &entry, &purger) + build_series_index(&store, ®ion, &version.metadata, &bucket, &entry, &purger) .await .map(|_| ()) }; diff --git a/src/mito2/src/series_index/maintenance.rs b/src/mito2/src/series_index/maintenance.rs index f5221bb1ebe..2520e32422e 100644 --- a/src/mito2/src/series_index/maintenance.rs +++ b/src/mito2/src/series_index/maintenance.rs @@ -19,13 +19,14 @@ use std::sync::Arc; use std::time::{Duration, Instant}; use common_telemetry::{debug, info}; +use common_time::TimeToLive; use object_store::ObjectStore; +use store_api::metadata::RegionMetadataRef; use store_api::storage::{FileId, RegionId}; use crate::error::Result; use crate::metrics::{SERIES_INDEX_RECONCILE_ELAPSED, SERIES_INDEX_RECONCILE_TOTAL}; use crate::read::series_candidate::is_sparse_metric_metadata; -use crate::region::version::VersionRef; use crate::region::{MitoRegionRef, RegionLeaderState, RegionRoleState}; use crate::series_index::bucket::{ group_files_into_series_buckets, plan_series_indexes, rounded_bucket_width, @@ -37,6 +38,15 @@ use crate::series_index::catalog::{ }; use crate::series_index::purger::IndexFilePurger; use crate::series_index::version::{SeriesIndexFileHandle, SeriesIndexVersion}; +use crate::sst::file::FileHandle; + +/// Captured index-build inputs that retain SSTs without retaining memtables. +struct ReconcileSnapshot { + metadata: RegionMetadataRef, + files: Vec, + compaction_time_window: Option, + ttl: Option, +} /// Retires newly completed series files unless their snapshot is published. #[derive(Default)] @@ -86,21 +96,36 @@ pub(crate) async fn reconcile_series_indexes( allow_builds: bool, ) -> Result { let total_start = Instant::now(); - // Use this snapshot throughout reconciliation, even if the region version advances. - let version = region.version_control.current().version; - if !is_sparse_metric_metadata(&version.metadata) { - SERIES_INDEX_RECONCILE_TOTAL - .with_label_values(&["noop"]) - .inc(); - return Ok(ReconcileStats::default()); - } + // Retaining the full version during I/O would pin flushed memtables and delay + // write-buffer reclamation. Capture consistent build inputs without the memtables. + let snapshot = { + let version = region.version_control.current().version; + if !is_sparse_metric_metadata(&version.metadata) { + SERIES_INDEX_RECONCILE_TOTAL + .with_label_values(&["noop"]) + .inc(); + return Ok(ReconcileStats::default()); + } + ReconcileSnapshot { + metadata: version.metadata.clone(), + files: version + .ssts + .levels() + .iter() + .flat_map(|level| level.files()) + .cloned() + .collect(), + compaction_time_window: version.compaction_time_window, + ttl: version.options.ttl, + } + }; let build_start = Instant::now(); let mut unpublished = UnpublishedSeriesFiles::default(); let (next, mut stats) = build_index_version( worker_id, &store, ®ion, - &version, + &snapshot, requested_bucket_width, now_ms, &purger, @@ -161,7 +186,7 @@ async fn build_index_version( worker_id: u32, store: &ObjectStore, region: &MitoRegionRef, - version: &VersionRef, + snapshot: &ReconcileSnapshot, requested_bucket_width: Duration, now_ms: i64, purger: &IndexFilePurger, @@ -170,13 +195,12 @@ async fn build_index_version( allow_builds: bool, ) -> Result<(Option, ReconcileStats)> { let mut stats = ReconcileStats::default(); - let files = version - .ssts - .levels() - .iter() - .flat_map(|level| level.files()) - .cloned() - .collect::>(); + let ReconcileSnapshot { + metadata, + files, + compaction_time_window, + ttl, + } = snapshot; stats.source_files = files.len(); let visible = files .iter() @@ -184,12 +208,12 @@ async fn build_index_version( .collect::>(); let current = region.series_index_version(); // Empty build inputs still let the planner expire established bucket coverage. - let buckets = match (allow_builds, version.compaction_time_window) { + let buckets = match (allow_builds, *compaction_time_window) { (true, Some(window)) => rounded_bucket_width(requested_bucket_width, window) // Successful rounding guarantees the window fits in i64; subsecond windows // use the same one-second minimum as rounded_bucket_width. .map(|width| { - group_files_into_series_buckets(&files, width, (window.as_secs() as i64).max(1)) + group_files_into_series_buckets(files, width, (window.as_secs() as i64).max(1)) }) .unwrap_or_default(), (true, None) => { @@ -201,12 +225,7 @@ async fn build_index_version( } (false, _) => Vec::new(), }; - let plan = plan_series_indexes( - buckets, - current.index_buckets.clone(), - version.options.ttl, - now_ms, - ); + let plan = plan_series_indexes(buckets, current.index_buckets.clone(), *ttl, now_ms); stats.computed_buckets = plan.computed_buckets; stats.skipped_buckets = plan.skipped_buckets; let mut next = prune_index_version(¤t, &visible, &plan.expired_index_ids, &mut stats); @@ -240,14 +259,15 @@ async fn build_index_version( for file in &bucket.files { if enable_range_index && !range_indexes.contains_key(&file.file_id().file_id()) - && let Some(entry) = build_range_index(store, region, version, file.clone()).await? + && let Some(entry) = + build_range_index(store, region, metadata, file.clone()).await? { stats.built_range += 1; range_indexes.insert(entry.file_id, entry); } } let series_handle = - build_series_index(store, region, version, &bucket, &expected, purger).await?; + build_series_index(store, region, metadata, &bucket, &expected, purger).await?; unpublished.0.push(series_handle.clone()); stats.built_series += 1; series_indexes.insert(expected.index_uuid, series_handle); @@ -259,7 +279,7 @@ async fn build_index_version( if range_indexes.contains_key(&file_id) { continue; } - if let Some(entry) = build_range_index(store, region, version, file).await? { + if let Some(entry) = build_range_index(store, region, metadata, file.clone()).await? { stats.built_range += 1; range_indexes.insert(entry.file_id, entry); } diff --git a/src/mito2/src/series_index/tests.rs b/src/mito2/src/series_index/tests.rs index 631bdc75b87..5f13d0b4b2d 100644 --- a/src/mito2/src/series_index/tests.rs +++ b/src/mito2/src/series_index/tests.rs @@ -720,13 +720,13 @@ async fn test_failed_reconcile_retires_only_unpublished_series() { let (bucket, entry) = &plan.builds[0]; let mut ranges = HashMap::new(); for file in &bucket.files { - let entry = build_range_index(store, ®ion, &version, file.clone()) + let entry = build_range_index(store, ®ion, &version.metadata, file.clone()) .await .unwrap() .unwrap(); ranges.insert(entry.file_id, entry); } - let handle = build_series_index(store, ®ion, &version, bucket, entry, purger) + let handle = build_series_index(store, ®ion, &version.metadata, bucket, entry, purger) .await .unwrap(); let reused_path = series_index_path(region.region_id, entry.index_uuid);