rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-10-17 09:22:29 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: avoid pinning memtables during series-index reconciliation (#9588)

Browse Source 
Signed-off-by: evenyag <realevenyag@gmail.com>
This commit is contained in:
Yingwen
2026-10-15 03:51:05 +00:00
committed by GitHub
parent 5cfdc30a69
commit ce66d8d4ac
3 changed files with 70 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/mito2/src/series_index/builder.rs
+19 -17
View File
@@ -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<Option<RangeIndexEntry>> {
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<SeriesIndexFileHandle> {
let mut sources = Vec::<BoxedRecordBatchStream>::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, &region, &version, file.clone())
let completed = build_range_index(&store, &region, &version.metadata, file.clone())
.await
.unwrap()
.unwrap();
@@ -318,7 +319,7 @@ mod tests {
let result = build_series_index(
&store.clone().layer(layer),
&region,
&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, &region, &version, file.clone())
build_range_index(&store, &region, &version.metadata, file.clone())
.await
.unwrap()
.unwrap();
}
let _handle = build_series_index(&store, &region, &version, &bucket, &entry, &purger)
.await
.unwrap();
let _handle =
build_series_index(&store, &region, &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<Option<SeriesIndexFileHandle>> = async {
for file in files {
let Some(entry) =
build_range_index(&store, &region, &version, file.clone()).await?
build_range_index(&store, &region, &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, &region, &version, &bucket, &entry, &purger)
build_series_index(&store, &region, &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, &region, &version, bucket.files[0].clone())
build_range_index(&store, &region, &version.metadata, bucket.files[0].clone())
.await
.map(|_| ())
} else {
let (purger, _receiver) = series_index_channel(store.clone());
build_series_index(&store, &region, &version, &bucket, &entry, &purger)
build_series_index(&store, &region, &version.metadata, &bucket, &entry, &purger)
.await
.map(|_| ())
};
src/mito2/src/series_index/maintenance.rs
+49 -29
View File
@@ -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<FileHandle>,
compaction_time_window: Option<Duration>,
ttl: Option<TimeToLive>,
}
/// 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<ReconcileStats> {
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,
&region,
&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<SeriesIndexVersion>, ReconcileStats)> {
let mut stats = ReconcileStats::default();
let files = version
.ssts
.levels()
.iter()
.flat_map(|level| level.files())
.cloned()
.collect::<Vec<_>>();
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::<HashSet<_>>();
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(&current, &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);
}
src/mito2/src/series_index/tests.rs
+2 -2
View File
@@ -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, &region, &version, file.clone())
let entry = build_range_index(store, &region, &version.metadata, file.clone())
.await
.unwrap()
.unwrap();
ranges.insert(entry.file_id, entry);
}
let handle = build_series_index(store, &region, &version, bucket, entry, purger)
let handle = build_series_index(store, &region, &version.metadata, bucket, entry, purger)
.await
.unwrap();
let reused_path = series_index_path(region.region_id, entry.index_uuid);