diff --git a/src/mito2/AGENTS.md b/src/mito2/AGENTS.md index f70aa622a27..95b9eddae07 100644 --- a/src/mito2/AGENTS.md +++ b/src/mito2/AGENTS.md @@ -26,7 +26,7 @@ snapshot isolation). It implements the `RegionEngine` trait from `store-api`. | `compaction` | `src/mito2/src/compaction/` | Compaction scheduler (`scheduler.rs` + `scheduler/`), TWCS picker, strict-window manual picker, compactor, memory control | | `access_layer` | `src/mito2/src/access_layer.rs` | SST read/write over the object store | | `sst` | `src/mito2/src/sst/` | Parquet format, file metadata, index layout | -| `series_index` | `src/mito2/src/series_index/` | Series index writer/searcher, bucket planning, catalogs, worker-owned reconciliation (`maintenance.rs`), shared approximate usage (`task.rs`), catalogs, and index lifecycle | +| `series_index` | `src/mito2/src/series_index/` | Format interfaces (`format.rs`), index writer/searcher, per-SST samples artifacts (`artifact.rs`), bucket planning, catalogs, reconciliation (`maintenance.rs`), and deferred purge | | `read` | `src/mito2/src/read/` | `ScanRegion`, merge, dedup, projection, streaming | | `manifest` | `src/mito2/src/manifest/` | `RegionManifestManager`, manifest actions/edits | | `cache` | `src/mito2/src/cache.rs` | Write/file/page caches | @@ -66,6 +66,7 @@ filtered `RecordBatch` stream. and defer new builds when the shared estimate is full. Cleanup still prunes obsolete range entries and expires series indexes to reclaim published usage. Concurrent builds may overshoot; closed-region files are not counted. Series handles and the SST purger own file deletion. +- **Series format PoC**: samples are rebuildable local SST sidecars. Optional writer interfaces (`format.rs`) integrate with reconciliation, catalog publication, and deferred purge. Defaults retain the original Parquet writer and disable samples. Prepared sources, read APIs, and concrete new formats are deferred; ordinary scans only use Parquet index artifacts. - **Series-index coverage** (`series_index/catalog.rs`): `SeriesIndexEntry` stores compaction-window width and SST summaries keyed by aligned start in both catalogs and Parquet footers. - **Request types** (`request.rs`): usually tied to proto definitions consumed by `datanode`. diff --git a/src/mito2/src/engine/scan_test.rs b/src/mito2/src/engine/scan_test.rs index c0b155f7311..68469e85324 100644 --- a/src/mito2/src/engine/scan_test.rs +++ b/src/mito2/src/engine/scan_test.rs @@ -1549,6 +1549,7 @@ async fn check_two_phase_series_scan( let store = region.series_index_store.clone().unwrap(); let sequence = region.flushed_sequence(); let entry = SeriesIndexEntry { + artifact: None, file_size: 0, index_uuid: FileId::random(), bucket_start: Timestamp::new_millisecond(0), diff --git a/src/mito2/src/read/series_candidate.rs b/src/mito2/src/read/series_candidate.rs index 3ba5512a7bd..82210e0ffa7 100644 --- a/src/mito2/src/read/series_candidate.rs +++ b/src/mito2/src/read/series_candidate.rs @@ -243,6 +243,10 @@ impl SeriesIndexCoverage { .version .series_indexes .values() + // Unimplemented formats cannot replace authoritative SST candidates. + .filter(|index| { + index.entry().format() == crate::series_index::artifact::ArtifactFormat::Parquet + }) .map(|index| { let files: HashSet<_> = uncovered .iter() @@ -778,6 +782,7 @@ mod tests { )); let store = env.access_layer.object_store().clone(); let entry = SeriesIndexEntry { + artifact: None, file_size: 0, index_uuid: FileId::random(), bucket_start: Timestamp::new_millisecond(0), @@ -964,6 +969,48 @@ mod tests { } } + #[tokio::test] + async fn index_merge_releases_permits_before_yielding() { + let (_env, scanner, _pruner) = indexed_scanner().await; + let semaphore = Arc::new(Semaphore::new(1)); + let context = scanner.stream_ctx.input.series_index.clone().unwrap(); + let streams = (0..3) + .map(|_| { + index_primary_key_stream( + scanner.stream_ctx.clone(), + context.clone(), + scanner.coverage.indexes[0].clone(), + semaphore.clone(), + ) + }) + .collect(); + let merged = merge_primary_key_streams( + streams, + scanner.memory_pool.clone(), + &scanner.metrics_set, + 0, + "permit-regression", + ) + .unwrap(); + let series = + decode_metric_series(merged, scanner.stream_ctx.input.region_metadata().clone()) + .unwrap(); + let batches = tokio::time::timeout( + std::time::Duration::from_secs(10), + series.try_collect::>(), + ) + .await + .expect("index streams must not hold permits across yield") + .unwrap(); + assert_eq!( + (0..1001) + .map(|tsid| MetricSeriesId { table_id: 1, tsid }) + .collect::>(), + batches.into_iter().flatten().collect::>() + ); + assert_eq!(1, semaphore.available_permits()); + } + #[tokio::test] async fn index_read_failure_after_output_is_propagated() { let (_env, scanner, _pruner) = indexed_scanner().await; diff --git a/src/mito2/src/series_index.rs b/src/mito2/src/series_index.rs index 3aa70256709..585120628e8 100644 --- a/src/mito2/src/series_index.rs +++ b/src/mito2/src/series_index.rs @@ -16,9 +16,13 @@ //! //! Under development. Index files are currently stored on the local filesystem. +pub(crate) mod artifact; mod bucket; mod builder; mod catalog; +pub(crate) mod format; +#[cfg(test)] +mod format_tests; mod maintenance; mod purger; mod searcher; diff --git a/src/mito2/src/series_index/artifact.rs b/src/mito2/src/series_index/artifact.rs new file mode 100644 index 00000000000..e1412920f6f --- /dev/null +++ b/src/mito2/src/series_index/artifact.rs @@ -0,0 +1,93 @@ +// Copyright 2023 Greptime Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Immutable, potentially multi-object format outputs and reader-owned lifetimes. + +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; + +use serde::{Deserialize, Serialize}; +use store_api::storage::FileId; + +use crate::series_index::purger::IndexFilePurger; +use crate::sst::file::RegionFileId; + +/// Persisted artifact identifiers. Concrete codecs are added with their implementations. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +pub(crate) enum ArtifactFormat { + #[serde(rename = "parquet")] + Parquet, + #[cfg(test)] + #[serde(rename = "fake-samples")] + FakeSamples, + #[cfg(test)] + #[serde(rename = "fake-index")] + FakeIndex, +} + +/// Identifies the exact bytes produced by one backend build. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub(crate) struct ArtifactMetadata { + pub(crate) format: ArtifactFormat, + pub(crate) generation: FileId, + /// Every object owned by this build, including metadata and auxiliary files. + pub(crate) objects: Vec, + pub(crate) file_size: u64, +} + +/// One samples artifact derived from an immutable Parquet SST. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub(crate) struct SamplesEntry { + pub(crate) file_id: FileId, + pub(crate) artifact: ArtifactMetadata, +} + +#[derive(Debug, Clone)] +pub(crate) struct SamplesFileHandle(Arc); + +#[derive(Debug)] +struct SamplesFileInner { + file_id: RegionFileId, + entry: SamplesEntry, + retired: AtomicBool, + purger: IndexFilePurger, +} + +impl SamplesFileHandle { + pub(crate) fn new(file_id: RegionFileId, entry: SamplesEntry, purger: IndexFilePurger) -> Self { + Self(Arc::new(SamplesFileInner { + file_id, + entry, + retired: AtomicBool::new(false), + purger, + })) + } + + pub(crate) fn entry(&self) -> &SamplesEntry { + &self.0.entry + } + + pub(crate) fn mark_deleted(&self) { + self.0.retired.store(true, Ordering::Release); + } +} + +impl Drop for SamplesFileInner { + fn drop(&mut self) { + if self.retired.load(Ordering::Acquire) { + self.purger + .purge_objects(self.file_id, self.entry.artifact.objects.clone()); + } + } +} diff --git a/src/mito2/src/series_index/bucket.rs b/src/mito2/src/series_index/bucket.rs index 446948838f6..470caeadd3b 100644 --- a/src/mito2/src/series_index/bucket.rs +++ b/src/mito2/src/series_index/bucket.rs @@ -334,6 +334,7 @@ impl SeriesBucket { .filter_map(|file| file.meta_ref().sequence.map(|sequence| sequence.get())) .min()?; Some(SeriesIndexEntry { + artifact: None, file_size: 0, index_uuid: FileId::random(), bucket_start: self.start, diff --git a/src/mito2/src/series_index/builder.rs b/src/mito2/src/series_index/builder.rs index 63a51badaeb..2a9db9503e3 100644 --- a/src/mito2/src/series_index/builder.rs +++ b/src/mito2/src/series_index/builder.rs @@ -35,6 +35,7 @@ use crate::series_index::bucket::SeriesBucket; use crate::series_index::catalog::{ RangeIndexEntry, SeriesIndexEntry, range_index_path, series_index_path, series_metadata, }; +use crate::series_index::format::SeriesIndexFormat; use crate::series_index::purger::{IndexFilePurger, IndexFileType, file_operation}; use crate::series_index::version::SeriesIndexFileHandle; use crate::series_index::{SeriesIndexWriter, SeriesIndexWriterOptions}; @@ -130,6 +131,7 @@ pub(crate) async fn build_series_index( version: &VersionRef, bucket: &SeriesBucket, entry: &SeriesIndexEntry, + backend: Option<&dyn SeriesIndexFormat>, purger: &IndexFilePurger, ) -> Result { let mut sources = Vec::::new(); @@ -180,6 +182,35 @@ pub(crate) async fn build_series_index( ) }; // TODO(yingwen): Deduplicate update-mode rows before series indexes are used by queries. + if let Some(backend) = backend { + let mut writer = backend + .writer(version.metadata.clone(), store.clone(), entry) + .await?; + let result: Result<()> = async { + while let Some(batch) = visible.try_next().await? { + writer.write(&batch).await?; + } + Ok(()) + } + .await; + if let Err(error) = result { + if let Err(cleanup_error) = writer.abort().await { + warn!(cleanup_error; "Failed to abort series-index artifact build"); + } + return Err(error); + } + let artifact = writer.finish().await?; + file_operation(IndexFileType::Series, "build", "success"); + return Ok(SeriesIndexFileHandle::new( + region.region_id, + SeriesIndexEntry { + file_size: artifact.file_size, + artifact: Some(artifact), + ..entry.clone() + }, + purger.clone(), + )); + } let path = series_index_path(region.region_id, entry.index_uuid); let mut writer = SeriesIndexWriter::try_new( version.metadata.clone(), @@ -214,6 +245,79 @@ pub(crate) async fn build_series_index( )) } +/// Converts one immutable SST into an independently rebuildable samples artifact. +pub(crate) async fn build_samples( + store: &ObjectStore, + region: &MitoRegionRef, + version: &VersionRef, + file: FileHandle, + backend: &dyn crate::series_index::format::SamplesFormat, + purger: &IndexFilePurger, +) -> Result> { + use crate::series_index::artifact::{SamplesEntry, SamplesFileHandle}; + let file_id = file.file_id(); + let Some((context, mut selection)) = region + .access_layer + .read_sst(file) + .build_reader_input(&mut ReaderMetrics::default()) + .await? + else { + return Ok(None); + }; + let context = Arc::new(context); + let mapper = FlatProjectionMapper::new( + &version.metadata, + 0..version.metadata.column_metadatas.len(), + )?; + let compat = FlatCompatBatch::try_new(&mapper, context.read_format(), false)?; + let generation = store_api::storage::FileId::random(); + let prefix = format!( + "{}/samples/{}/{generation}/", + region.region_id.as_u64(), + file_id.file_id() + ); + let mut writer = backend + .writer(version.metadata.clone(), store.clone(), &prefix, generation) + .await?; + let result: Result<()> = async { + while let Some((row_group, selection)) = selection.pop_first() { + let reader = context + .reader_builder() + .build(context.build_context(row_group, Some(selection), None)) + .await?; + let mut reader = FlatPruneReader::new_with_row_group_reader( + context.clone(), + FlatRowGroupReader::new(context.clone(), reader), + context.pre_filter_mode().skip_fields(), + ); + while let Some(batch) = reader.next_batch().await? { + let batch = match &compat { + Some(compat) => compat.compat(batch)?, + None => batch, + }; + writer.write(&batch).await?; + } + } + Ok(()) + } + .await; + if let Err(error) = result { + if let Err(cleanup) = writer.abort().await { + warn!(cleanup; "Failed to abort samples build"); + } + return Err(error); + } + let artifact = writer.finish().await?; + Ok(Some(SamplesFileHandle::new( + file_id, + SamplesEntry { + file_id: file_id.file_id(), + artifact, + }, + purger.clone(), + ))) +} + #[cfg(test)] mod tests { use std::collections::{BTreeMap, HashMap}; @@ -228,7 +332,8 @@ mod tests { use super::*; 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, + SeriesIndexCatalog, load_version_control, series_catalog_path, series_index_path, + store_catalog, }; use crate::series_index::purger::series_index_channel; use crate::series_index::tests::prepare_region; @@ -321,6 +426,7 @@ mod tests { &version, &bucket, &entry, + None, &purger, ) .await; @@ -471,7 +577,7 @@ mod tests { .unwrap() .unwrap(); } - let _handle = build_series_index(&store, ®ion, &version, &bucket, &entry, &purger) + let _handle = build_series_index(&store, ®ion, &version, &bucket, &entry, None, &purger) .await .unwrap(); let bytes = store @@ -540,7 +646,7 @@ mod tests { }; completed.push(entry.file_id); } - build_series_index(&store, ®ion, &version, &bucket, &entry, &purger) + build_series_index(&store, ®ion, &version, &bucket, &entry, None, &purger) .await .map(Some) } @@ -746,7 +852,7 @@ mod tests { .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, &bucket, &entry, None, &purger) .await .map(|_| ()) }; diff --git a/src/mito2/src/series_index/catalog.rs b/src/mito2/src/series_index/catalog.rs index 95cf87f83d3..66355ace359 100644 --- a/src/mito2/src/series_index/catalog.rs +++ b/src/mito2/src/series_index/catalog.rs @@ -33,6 +33,7 @@ use crate::series_index::version::{ pub(crate) use crate::sst::range_index::range_index_path; const SERIES_DIR: &str = "series"; +const SAMPLES_CATALOG: &str = "samples.json"; const RANGE_CATALOG: &str = "range-index.json"; const SERIES_CATALOG: &str = "series-index.json"; const SERIES_METADATA_KEY: &str = "greptime.series_index"; @@ -60,6 +61,8 @@ pub(crate) struct WindowSequence { /// relies on this complete-coverage contract, not on `source_file_ids`. #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub(crate) struct SeriesIndexEntry { + #[serde(default, skip_serializing_if = "Option::is_none")] + pub(crate) artifact: Option, /// Completed size in the catalog; zero in the footer written before completion. pub(crate) file_size: u64, pub(crate) index_uuid: FileId, @@ -83,6 +86,14 @@ pub(crate) struct SeriesIndexEntry { } impl SeriesIndexEntry { + /// Entries written before pluggable formats describe the original Parquet layout. + pub(crate) fn format(&self) -> crate::series_index::artifact::ArtifactFormat { + self.artifact.as_ref().map_or( + crate::series_index::artifact::ArtifactFormat::Parquet, + |artifact| artifact.format, + ) + } + /// Whether this index completely covers an SST in the query's sequence domain. pub(crate) fn covers_file( &self, @@ -116,6 +127,18 @@ pub(crate) struct RangeIndexCatalog { pub(crate) indexes: Vec, } +#[derive(Debug, Default, Serialize, Deserialize)] +pub(crate) struct SamplesCatalog { + pub(crate) entries: Vec, + /// Once samples are enabled this single catalog commits all artifact references together. + pub(crate) series: Vec, + pub(crate) ranges: Vec, +} + +pub(crate) fn samples_catalog_path(region_id: RegionId) -> String { + format!("{}/{SAMPLES_CATALOG}", region_id.as_u64()) +} + pub(crate) fn range_catalog_path(region_id: RegionId) -> String { format!("{}/{RANGE_CATALOG}", region_id.as_u64()) } @@ -172,6 +195,7 @@ where pub(crate) async fn delete_catalogs(store: &ObjectStore, region_id: RegionId) { for path in [ series_catalog_path(region_id), + samples_catalog_path(region_id), range_catalog_path(region_id), ] { if let Err(error) = store.delete(&path).await @@ -194,14 +218,17 @@ pub(crate) async fn load_version_control( let series = load_catalog::(store, &series_catalog_path(region_id)) .await .unwrap_or_default(); - let version = SeriesIndexVersion::new( - range - .indexes + let samples = load_catalog::(store, &samples_catalog_path(region_id)).await; + let (ranges, indexes, sample_entries) = match samples { + Some(catalog) => (catalog.ranges, catalog.series, catalog.entries), + None => (range.indexes, series.indexes, Vec::new()), + }; + let mut version = SeriesIndexVersion::new( + ranges .into_iter() .map(|entry| (entry.file_id, entry)) .collect(), - series - .indexes + indexes .into_iter() .map(|entry| { ( @@ -211,6 +238,17 @@ pub(crate) async fn load_version_control( }) .collect(), ); + for entry in sample_entries { + let id = entry.file_id; + version.samples.insert( + id, + crate::series_index::artifact::SamplesFileHandle::new( + crate::sst::file::RegionFileId::new(region_id, id), + entry, + purger.clone(), + ), + ); + } let control = SeriesIndexVersionControl::default(); control.publish(std::sync::Arc::new(version)); control @@ -258,10 +296,37 @@ mod tests { } } + #[test] + fn legacy_series_catalog_fixture() { + use crate::series_index::artifact::ArtifactFormat; + let json = include_str!( + "../../../../tests/compatibility/fixtures/series_index/legacy_catalog.json" + ); + let catalog: SeriesIndexCatalog = serde_json::from_str(json).unwrap(); + assert_eq!(1, catalog.indexes.len()); + let entry = &catalog.indexes[0]; + assert!(entry.artifact.is_none()); + assert_eq!(ArtifactFormat::Parquet, entry.format()); + assert_eq!( + serde_json::from_str::(json).unwrap(), + serde_json::to_value(&catalog).unwrap() + ); + for (format, name) in [ + (ArtifactFormat::Parquet, "parquet"), + (ArtifactFormat::FakeSamples, "fake-samples"), + (ArtifactFormat::FakeIndex, "fake-index"), + ] { + let encoded = serde_json::to_value(format).unwrap(); + assert_eq!(name, encoded.as_str().unwrap()); + assert_eq!(format, serde_json::from_value(encoded).unwrap()); + } + } + #[test] fn coverage_uses_exclusive_time_end_and_inclusive_file_sequences() { let region_id = RegionId::new(1, 1); let entry = SeriesIndexEntry { + artifact: None, file_size: 0, index_uuid: FileId::random(), bucket_start: Timestamp::new_second(1), @@ -359,6 +424,7 @@ mod tests { let store = ObjectStore::new(Memory::default()).unwrap(); let region_id = RegionId::new(1, 1); let entry = SeriesIndexEntry { + artifact: None, file_size: 0, index_uuid: FileId::random(), bucket_start: Timestamp::new_second(0), diff --git a/src/mito2/src/series_index/format.rs b/src/mito2/src/series_index/format.rs new file mode 100644 index 00000000000..b59933f719a --- /dev/null +++ b/src/mito2/src/series_index/format.rs @@ -0,0 +1,72 @@ +// Copyright 2023 Greptime Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Storage-format boundaries. Encodings stay outside reconciliation and source scheduling. + +use std::fmt::Debug; +use std::sync::Arc; + +use async_trait::async_trait; +use datatypes::arrow::record_batch::RecordBatch; +use object_store::ObjectStore; +use store_api::metadata::RegionMetadataRef; +use store_api::storage::FileId; + +use crate::error::Result; +use crate::series_index::artifact::{ArtifactFormat, ArtifactMetadata}; +use crate::series_index::catalog::SeriesIndexEntry; + +/// Writers own partial outputs until finish succeeds; abort removes every partial object. +/// A failed finish must clean up its output before returning; implementations also own +/// cleanup on cancellation while a writer is live. +#[async_trait] +pub(crate) trait FormatWriter: Send { + async fn write(&mut self, batch: &RecordBatch) -> Result<()>; + async fn finish(self: Box) -> Result; + async fn abort(self: Box) -> Result<()>; +} + +/// Builds series-index artifacts; read interfaces are deferred. +#[async_trait] +pub(crate) trait SeriesIndexFormat: Debug + Send + Sync { + fn format(&self) -> ArtifactFormat; + async fn writer( + &self, + metadata: RegionMetadataRef, + store: ObjectStore, + entry: &SeriesIndexEntry, + ) -> Result>; +} + +/// Builds samples artifacts; read interfaces are deferred. +#[async_trait] +pub(crate) trait SamplesFormat: Debug + Send + Sync { + fn format(&self) -> ArtifactFormat; + /// The caller supplies a unique prefix. Finish reports all objects owned below it. + async fn writer( + &self, + metadata: RegionMetadataRef, + store: ObjectStore, + prefix: &str, + generation: FileId, + ) -> Result>; +} + +/// Optional writers, captured once per reconciliation pass. +/// Without an index backend, builds use the existing Parquet writer. +#[derive(Debug, Clone, Default)] +pub(crate) struct SeriesFormats { + pub(crate) index: Option>, + pub(crate) samples: Option>, +} diff --git a/src/mito2/src/series_index/format_tests.rs b/src/mito2/src/series_index/format_tests.rs new file mode 100644 index 00000000000..a7e0c86069d --- /dev/null +++ b/src/mito2/src/series_index/format_tests.rs @@ -0,0 +1,442 @@ +// Copyright 2023 Greptime Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Artifact lifecycle tests; fake formats have no codec. + +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::time::Duration; + +use async_trait::async_trait; +use datatypes::arrow::record_batch::RecordBatch; +use futures::TryStreamExt; +use object_store::ObjectStore; +use object_store::layers::mock::{self, MockLayerBuilder, oio}; +use snafu::ResultExt; +use store_api::metadata::RegionMetadataRef; +use store_api::region_engine::RegionEngine; +use store_api::storage::FileId; + +use crate::config::MitoConfig; +use crate::error::{OpenDalSnafu, Result, UnexpectedSnafu}; +use crate::series_index::artifact::{ArtifactFormat, ArtifactMetadata}; +use crate::series_index::catalog::{SeriesIndexEntry, load_version_control}; +use crate::series_index::format::{FormatWriter, SamplesFormat, SeriesFormats, SeriesIndexFormat}; +use crate::series_index::maintenance::reconcile_series_indexes; +use crate::series_index::purger::series_index_channel; +use crate::series_index::tests::prepare_region_with_config; +use crate::test_util::TestEnv; + +#[derive(Debug, Default)] +struct FakeSamples { + fail: Arc, +} + +struct FakeWriter { + store: ObjectStore, + artifact: ArtifactMetadata, + fail: Arc, +} + +#[async_trait] +impl FormatWriter for FakeWriter { + async fn write(&mut self, _batch: &RecordBatch) -> Result<()> { + if self.fail.load(Ordering::Relaxed) { + return UnexpectedSnafu { + reason: "injected samples write failure", + } + .fail(); + } + Ok(()) + } + async fn finish(self: Box) -> Result { + Ok(self.artifact) + } + async fn abort(self: Box) -> Result<()> { + for path in &self.artifact.objects { + self.store.delete(path).await.context(OpenDalSnafu)?; + } + Ok(()) + } +} + +#[async_trait] +impl SamplesFormat for FakeSamples { + fn format(&self) -> ArtifactFormat { + ArtifactFormat::FakeSamples + } + async fn writer( + &self, + _metadata: RegionMetadataRef, + store: ObjectStore, + prefix: &str, + generation: FileId, + ) -> Result> { + let objects = vec![format!("{prefix}data"), format!("{prefix}directory")]; + for path in &objects { + store.write(path, "fake").await.context(OpenDalSnafu)?; + } + Ok(Box::new(FakeWriter { + store, + artifact: ArtifactMetadata { + format: self.format(), + generation, + objects, + file_size: 8, + }, + fail: self.fail.clone(), + })) + } +} + +#[tokio::test] +async fn format_skeleton_build_reopen_and_retire() { + let mut env = TestEnv::with_prefix("format-skeleton").await; + let (engine, region) = prepare_region_with_config( + &mut env, + &[1000, 2000], + MitoConfig { + experimental_enable_series_index: true, + experimental_series_index_maintenance_interval: Duration::from_secs(3600), + min_compaction_interval: Duration::from_secs(3600), + ..Default::default() + }, + ) + .await; + let fake = Arc::new(FakeSamples::default()); + *region.series_index_version_control.formats.write().unwrap() = Arc::new(SeriesFormats { + samples: Some(fake.clone()), + ..Default::default() + }); + let store = region.series_index_store.clone().unwrap(); + let (purger, mut purges) = series_index_channel(store.clone()); + let stats = reconcile_series_indexes( + 0, + store.clone(), + region.clone(), + Duration::from_secs(100), + 0, + purger.clone(), + false, + true, + ) + .await + .unwrap(); + assert_eq!(2, stats.built_samples); + let snapshot = region.series_index_version(); + assert_eq!(2, snapshot.samples.len()); + let reopened = load_version_control(&store, region.region_id, &purger).await; + assert_eq!(2, reopened.current().samples.len()); + assert!(snapshot.disk_usage() >= 16); + let stats = reconcile_series_indexes( + 0, + store.clone(), + region.clone(), + Duration::from_secs(100), + 0, + purger.clone(), + false, + true, + ) + .await + .unwrap(); + assert_eq!(0, stats.built_samples); + + // Rebuilding a missing sidecar replaces its generation but retains old readers' files. + let old = snapshot.samples.values().next().unwrap().clone(); + store + .delete(&old.entry().artifact.objects[0]) + .await + .unwrap(); + let stats = reconcile_series_indexes( + 0, + store.clone(), + region.clone(), + Duration::from_secs(100), + 0, + purger.clone(), + false, + true, + ) + .await + .unwrap(); + assert_eq!(1, stats.built_samples); + assert!(purges.try_recv().is_err()); + drop(snapshot); + assert!(purges.try_recv().is_err()); + drop(old); + let request = purges.try_recv().unwrap(); + assert_eq!(2, request.objects.unwrap().len()); + engine.stop().await.unwrap(); +} + +#[tokio::test] +async fn format_skeleton_abort_removes_all_objects() { + let mut env = TestEnv::with_prefix("format-abort").await; + let (engine, region) = crate::series_index::tests::prepare_region(&mut env).await; + let store = ObjectStore::new(object_store::services::Memory::default()).unwrap(); + let (purger, _receiver) = series_index_channel(store.clone()); + let version = region.version(); + let file = version + .ssts + .levels() + .iter() + .flat_map(|l| l.files()) + .next() + .unwrap() + .clone(); + let fake = FakeSamples::default(); + fake.fail.store(true, Ordering::Relaxed); + assert!( + crate::series_index::builder::build_samples( + &store, ®ion, &version, file, &fake, &purger + ) + .await + .is_err() + ); + assert!( + store + .list_with(&format!("{}/samples/", region.region_id.as_u64())) + .recursive(true) + .await + .unwrap() + .iter() + .all(|entry| entry.metadata().is_dir()) + ); + assert!(region.series_index_version().samples.is_empty()); + engine.stop().await.unwrap(); +} + +#[derive(Debug, Default)] +struct FakeIndex; + +#[async_trait] +impl SeriesIndexFormat for FakeIndex { + fn format(&self) -> ArtifactFormat { + ArtifactFormat::FakeIndex + } + + async fn writer( + &self, + metadata: RegionMetadataRef, + store: ObjectStore, + entry: &SeriesIndexEntry, + ) -> Result> { + let path = format!( + "{}/series/{}.fake", + metadata.region_id.as_u64(), + entry.index_uuid + ); + store.write(&path, "fake").await.context(OpenDalSnafu)?; + Ok(Box::new(FakeWriter { + store, + artifact: ArtifactMetadata { + format: self.format(), + generation: entry.index_uuid, + objects: vec![path], + file_size: 4, + }, + fail: Arc::new(AtomicBool::new(false)), + })) + } +} + +#[tokio::test] +async fn format_index_rebuild_is_scoped_and_scans_fall_back_to_ssts() { + let mut env = TestEnv::with_prefix("format-index").await; + // Index builds require four SSTs per bucket. + let timestamps = [1000, 2000, 3000, 4000, 101000, 102000, 103000, 104000]; + let (engine, region) = prepare_region_with_config( + &mut env, + ×tamps, + MitoConfig { + experimental_enable_series_index: true, + experimental_series_scan_v2: true, + experimental_series_index_maintenance_interval: Duration::from_secs(3600), + min_compaction_interval: Duration::from_secs(3600), + ..Default::default() + }, + ) + .await; + *region.series_index_version_control.formats.write().unwrap() = Arc::new(SeriesFormats { + index: Some(Arc::new(FakeIndex)), + ..Default::default() + }); + let store = region.series_index_store.clone().unwrap(); + let (purger, mut purges) = series_index_channel(store.clone()); + let reconcile = || { + reconcile_series_indexes( + 0, + store.clone(), + region.clone(), + Duration::from_secs(100), + 0, + purger.clone(), + false, + true, + ) + }; + assert_eq!(2, reconcile().await.unwrap().built_series); + let before = region.series_index_version(); + assert!( + before + .series_indexes + .values() + .all(|h| h.entry().format() == ArtifactFormat::FakeIndex) + ); + assert_eq!(0, reconcile().await.unwrap().built_series); + let scanner = engine + .scanner( + region.region_id, + store_api::storage::ScanRequest { + distribution: Some(store_api::storage::TimeSeriesDistribution::PerSeries), + ..Default::default() + }, + ) + .await + .unwrap(); + let batches = scanner + .scan() + .await + .unwrap() + .try_collect::>() + .await + .unwrap(); + assert_eq!( + timestamps.len(), + batches.iter().map(|batch| batch.num_rows()).sum::() + ); + drop(scanner); + + let missing = before.series_indexes.values().next().unwrap(); + store + .delete(&missing.entry().artifact.as_ref().unwrap().objects[0]) + .await + .unwrap(); + assert_eq!(1, reconcile().await.unwrap().built_series); + let after = region.series_index_version(); + assert_eq!( + 1, + after + .series_indexes + .keys() + .filter(|id| before.series_indexes.contains_key(id)) + .count() + ); + assert!(purges.try_recv().is_err()); + drop(before); + assert!(purges.try_recv().is_ok()); + + // Disabling the injected writer restores ordinary Parquet artifacts without a read adapter. + *region.series_index_version_control.formats.write().unwrap() = + Arc::new(SeriesFormats::default()); + assert_eq!(2, reconcile().await.unwrap().built_series); + assert!( + region + .series_index_version() + .series_indexes + .values() + .all(|h| h.entry().artifact.is_none()) + ); + assert_eq!(0, reconcile().await.unwrap().built_series); + engine.stop().await.unwrap(); +} + +struct FailedCatalogWriter(oio::Writer); + +impl mock::Write for FailedCatalogWriter { + async fn write(&mut self, buffer: mock::Buffer) -> mock::Result<()> { + self.0.write(buffer).await + } + async fn close(&mut self) -> mock::Result { + Err(mock::Error::new( + mock::ErrorKind::Unexpected, + "injected artifact catalog failure", + )) + } + async fn abort(&mut self) -> mock::Result<()> { + self.0.abort().await + } +} + +#[tokio::test] +async fn format_catalog_failure_preserves_published_artifacts() { + let mut env = TestEnv::with_prefix("format-catalog-failure").await; + let (engine, region) = crate::series_index::tests::prepare_region(&mut env).await; + *region.series_index_version_control.formats.write().unwrap() = Arc::new(SeriesFormats { + index: Some(Arc::new(FakeIndex)), + samples: Some(Arc::new(FakeSamples::default())), + }); + let store = ObjectStore::new(object_store::services::Memory::default()).unwrap(); + let (purger, mut purges) = series_index_channel(store.clone()); + reconcile_series_indexes( + 0, + store.clone(), + region.clone(), + Duration::from_secs(100), + 0, + purger.clone(), + false, + true, + ) + .await + .unwrap(); + let before = region.series_index_version(); + let catalog_path = crate::series_index::catalog::samples_catalog_path(region.region_id); + let catalog = store.read(&catalog_path).await.unwrap().to_bytes(); + let old = before.samples.values().next().unwrap(); + store + .delete(&old.entry().artifact.objects[0]) + .await + .unwrap(); + let layer = MockLayerBuilder::default() + .writer_factory(Arc::new(|path, _, inner| { + if path.ends_with("samples.json") { + Box::new(FailedCatalogWriter(inner)) + } else { + inner + } + })) + .build() + .unwrap(); + let error = reconcile_series_indexes( + 0, + store.clone().layer(layer), + region.clone(), + Duration::from_secs(100), + 0, + purger.clone(), + false, + true, + ) + .await + .unwrap_err(); + assert!(format!("{error:?}").contains("injected artifact catalog failure")); + assert!(Arc::ptr_eq(&before, ®ion.series_index_version())); + assert_eq!(catalog, store.read(&catalog_path).await.unwrap().to_bytes()); + let cleanup = purges.try_recv().unwrap().objects.unwrap(); + assert_eq!(2, cleanup.len()); + assert!( + cleanup + .iter() + .all(|path| !old.entry().artifact.objects.contains(path)) + ); + assert!(purges.try_recv().is_err()); + let reopened = load_version_control(&store, region.region_id, &purger).await; + assert_eq!( + old.entry(), + reopened.current().samples[&old.entry().file_id].entry() + ); + engine.stop().await.unwrap(); +} diff --git a/src/mito2/src/series_index/maintenance.rs b/src/mito2/src/series_index/maintenance.rs index f5221bb1ebe..232e17ac266 100644 --- a/src/mito2/src/series_index/maintenance.rs +++ b/src/mito2/src/series_index/maintenance.rs @@ -20,6 +20,7 @@ use std::time::{Duration, Instant}; use common_telemetry::{debug, info}; use object_store::ObjectStore; +use snafu::ResultExt; use store_api::storage::{FileId, RegionId}; use crate::error::Result; @@ -40,16 +41,23 @@ use crate::series_index::version::{SeriesIndexFileHandle, SeriesIndexVersion}; /// Retires newly completed series files unless their snapshot is published. #[derive(Default)] -struct UnpublishedSeriesFiles(Vec); +struct UnpublishedSeriesFiles( + Vec, + Vec, +); impl UnpublishedSeriesFiles { fn disarm(&mut self) { self.0.clear(); + self.1.clear(); } } impl Drop for UnpublishedSeriesFiles { fn drop(&mut self) { + for handle in &self.1 { + handle.mark_deleted(); + } for handle in &self.0 { handle.mark_deleted(); } @@ -58,6 +66,8 @@ impl Drop for UnpublishedSeriesFiles { #[derive(Debug, Default)] pub(crate) struct ReconcileStats { + pub(crate) built_samples: usize, + pub(crate) removed_samples: usize, pub(crate) source_files: usize, pub(crate) built_range: usize, pub(crate) built_series: usize, @@ -69,7 +79,13 @@ pub(crate) struct ReconcileStats { impl ReconcileStats { fn changed(&self) -> bool { - self.built_range + self.built_series + self.removed_range + self.removed_series > 0 + self.built_samples + + self.removed_samples + + self.built_range + + self.built_series + + self.removed_range + + self.removed_series + > 0 } } @@ -201,12 +217,43 @@ async fn build_index_version( } (false, _) => Vec::new(), }; - let plan = plan_series_indexes( - buckets, - current.index_buckets.clone(), - version.options.ttl, - now_ms, + let formats = region + .series_index_version_control + .formats + .read() + .unwrap() + .clone(); + let selected_format = formats.index.as_ref().map_or( + crate::series_index::artifact::ArtifactFormat::Parquet, + |backend| backend.format(), ); + let mut unusable = HashSet::new(); + for handle in current.series_indexes.values() { + let entry = handle.entry(); + let mut reusable = entry.format() == selected_format; + // Legacy Parquet indexes retain their existing maintenance behavior. + if reusable && let Some(artifact) = &entry.artifact { + for object in &artifact.objects { + reusable &= store + .exists(object) + .await + .context(crate::error::OpenDalSnafu)?; + } + } + if !reusable { + unusable.insert(entry.index_uuid); + } + } + let mut index_buckets = current.index_buckets.clone(); + for bucket in index_buckets.values_mut() { + if bucket.index_ids.iter().any(|id| unusable.contains(id)) { + // Force replacement only for the affected bucket. Retain its geometry and + // handles until a replacement is built and its catalog is published. + bucket.compaction_window_secs = 0; + bucket.window_sequences.clear(); + } + } + let plan = plan_series_indexes(buckets, index_buckets, version.options.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); @@ -218,6 +265,7 @@ async fn build_index_version( } if plan.builds.is_empty() && next.is_none() + && formats.samples.is_none() && (!enable_range_index || current.range_indexes.len() == visible.len()) { return Ok((None, stats)); @@ -225,13 +273,50 @@ async fn build_index_version( let SeriesIndexVersion { mut range_indexes, mut series_indexes, + mut samples, .. } = next.unwrap_or_else(|| SeriesIndexVersion { + samples: current.samples.clone(), range_indexes: current.range_indexes.clone(), series_indexes: current.series_indexes.clone(), index_buckets: Default::default(), }); let index_buckets = plan.index_buckets; + if let Some(backend) = &formats.samples { + for file in &files { + let id = file.file_id().file_id(); + let reusable = if let Some(handle) = samples.get(&id) { + let artifact = &handle.entry().artifact; + let mut exists = artifact.format == backend.format(); + for path in &artifact.objects { + exists &= store + .exists(path) + .await + .context(crate::error::OpenDalSnafu)?; + } + exists + } else { + false + }; + if reusable { + continue; + } + if let Some(handle) = crate::series_index::builder::build_samples( + store, + region, + version, + file.clone(), + backend.as_ref(), + purger, + ) + .await? + { + unpublished.1.push(handle.clone()); + samples.insert(id, handle); + stats.built_samples += 1; + } + } + } for id in &plan.superseded_index_ids { series_indexes.remove(id); } @@ -246,8 +331,16 @@ async fn build_index_version( range_indexes.insert(entry.file_id, entry); } } - let series_handle = - build_series_index(store, region, version, &bucket, &expected, purger).await?; + let series_handle = build_series_index( + store, + region, + version, + &bucket, + &expected, + formats.index.as_deref(), + purger, + ) + .await?; unpublished.0.push(series_handle.clone()); stats.built_series += 1; series_indexes.insert(expected.index_uuid, series_handle); @@ -276,6 +369,7 @@ async fn build_index_version( return Ok((None, stats)); } let next = SeriesIndexVersion { + samples, range_indexes, series_indexes, index_buckets, @@ -293,7 +387,8 @@ fn prune_index_version( expired_index_ids: &[FileId], stats: &mut ReconcileStats, ) -> Option { - if current.range_indexes.keys().all(|id| visible.contains(id)) + if current.samples.keys().all(|id| visible.contains(id)) + && current.range_indexes.keys().all(|id| visible.contains(id)) && expired_index_ids .iter() .all(|id| !current.series_indexes.contains_key(id)) @@ -301,10 +396,13 @@ fn prune_index_version( return None; } let mut next = SeriesIndexVersion { + samples: current.samples.clone(), range_indexes: current.range_indexes.clone(), series_indexes: current.series_indexes.clone(), index_buckets: current.index_buckets.clone(), }; + next.samples.retain(|id, _| visible.contains(id)); + stats.removed_samples = current.samples.len() - next.samples.len(); next.range_indexes .retain(|file_id, _| visible.contains(file_id)); for id in expired_index_ids { @@ -322,6 +420,31 @@ async fn persist_index_catalogs( next: &SeriesIndexVersion, stats: &ReconcileStats, ) -> Result<()> { + let samples_path = crate::series_index::catalog::samples_catalog_path(region_id); + if !next.samples.is_empty() + || stats.removed_samples > 0 + || store + .exists(&samples_path) + .await + .context(crate::error::OpenDalSnafu)? + { + // A single object write commits all references. A failed write leaves the prior + // catalog usable while the unpublished-output guard retires the new artifacts. + return store_catalog( + store, + &samples_path, + &crate::series_index::catalog::SamplesCatalog { + entries: next.samples.values().map(|h| h.entry().clone()).collect(), + series: next + .series_indexes + .values() + .map(|h| h.entry().clone()) + .collect(), + ranges: next.range_indexes.values().copied().collect(), + }, + ) + .await; + } if stats.built_range + stats.removed_range > 0 { let mut range_entries = next.range_indexes.values().copied().collect::>(); range_entries @@ -364,6 +487,13 @@ async fn persist_index_catalogs( /// Publishes the snapshot and retires series files absent from the new version. fn publish_index_version(region: &MitoRegionRef, next: Arc) { let previous = region.series_index_version_control.publish(next.clone()); + for (id, handle) in &previous.samples { + if next.samples.get(id).is_none_or(|next| { + next.entry().artifact.generation != handle.entry().artifact.generation + }) { + handle.mark_deleted(); + } + } for (id, handle) in &previous.series_indexes { if !next.series_indexes.contains_key(id) { // Purge only after readers release their retained handles. @@ -405,6 +535,7 @@ mod tests { SeriesIndexFileHandle::new( RegionId::new(1, 1), SeriesIndexEntry { + artifact: None, index_uuid: series_id, file_size: 20, bucket_start: Timestamp::new_second(0), diff --git a/src/mito2/src/series_index/purger.rs b/src/mito2/src/series_index/purger.rs index a3eb6d89e38..66eda84c327 100644 --- a/src/mito2/src/series_index/purger.rs +++ b/src/mito2/src/series_index/purger.rs @@ -39,9 +39,10 @@ impl IndexFileType { } } -#[derive(Debug, Clone, Copy)] +#[derive(Debug, Clone)] pub(crate) struct PurgeRequest { pub(crate) file_id: RegionFileId, + pub(crate) objects: Option>, } #[derive(Clone)] @@ -57,6 +58,13 @@ impl Debug for IndexFilePurger { } impl IndexFilePurger { + pub(crate) fn purge_objects(&self, file_id: RegionFileId, objects: Vec) { + self.purge(PurgeRequest { + file_id, + objects: Some(objects), + }); + } + pub(crate) fn purge(&self, request: PurgeRequest) { if let Err(error) = self.sender.send(request) { let store = self.store.clone(); @@ -87,17 +95,24 @@ pub(crate) async fn run_index_purge_task( } async fn purge_file(store: &ObjectStore, request: PurgeRequest) { - let path = series_index_path(request.file_id.region_id(), request.file_id.file_id()); - match store.delete(&path).await { - Ok(()) => { - file_operation(IndexFileType::Series, "delete", "success"); - } - Err(error) if error.kind() == ErrorKind::NotFound => { - file_operation(IndexFileType::Series, "delete", "success"); - } - Err(error) => { - file_operation(IndexFileType::Series, "delete", "failure"); - warn!(error; "Failed to delete series index, index_type: {}, path: {}, phase: deletion", IndexFileType::Series.as_str(), path); + let objects = request.objects.unwrap_or_else(|| { + vec![series_index_path( + request.file_id.region_id(), + request.file_id.file_id(), + )] + }); + for path in objects { + match store.delete(&path).await { + Ok(()) => { + file_operation(IndexFileType::Series, "delete", "success"); + } + Err(error) if error.kind() == ErrorKind::NotFound => { + file_operation(IndexFileType::Series, "delete", "success"); + } + Err(error) => { + file_operation(IndexFileType::Series, "delete", "failure"); + warn!(error; "Failed to delete series index, index_type: {}, path: {}, phase: deletion", IndexFileType::Series.as_str(), path); + } } } } @@ -129,7 +144,10 @@ mod tests { let path = series_index_path(file_id.region_id(), file_id.file_id()); store.write(&path, "index").await.unwrap(); paths.push(path); - purger.purge(PurgeRequest { file_id }); + purger.purge(PurgeRequest { + file_id, + objects: None, + }); } drop(purger); tokio::time::timeout( @@ -151,7 +169,10 @@ mod tests { let file_id = RegionFileId::new(RegionId::new(1, 1), FileId::random()); let path = series_index_path(file_id.region_id(), file_id.file_id()); store.write(&path, "index").await.unwrap(); - purger.purge(PurgeRequest { file_id }); + purger.purge(PurgeRequest { + file_id, + objects: None, + }); tokio::time::timeout(Duration::from_secs(10), async { while store.exists(&path).await.unwrap() { tokio::task::yield_now().await; diff --git a/src/mito2/src/series_index/searcher.rs b/src/mito2/src/series_index/searcher.rs index 0b9befe4b8a..a2589b3bda9 100644 --- a/src/mito2/src/series_index/searcher.rs +++ b/src/mito2/src/series_index/searcher.rs @@ -312,6 +312,7 @@ mod tests { ) -> (SeriesIndexFileHandle, UnboundedReceiver) { let (purger, receiver) = series_index_channel(store.clone()); let entry = SeriesIndexEntry { + artifact: None, file_size: 0, index_uuid: FileId::random(), bucket_start: common_time::Timestamp::new_second(0), diff --git a/src/mito2/src/series_index/tests.rs b/src/mito2/src/series_index/tests.rs index 631bdc75b87..145c2088cce 100644 --- a/src/mito2/src/series_index/tests.rs +++ b/src/mito2/src/series_index/tests.rs @@ -63,7 +63,7 @@ pub(crate) async fn prepare_region_with_timestamps( .await } -async fn prepare_region_with_config( +pub(crate) async fn prepare_region_with_config( env: &mut TestEnv, timestamps: &[i64], config: MitoConfig, @@ -726,7 +726,7 @@ async fn test_failed_reconcile_retires_only_unpublished_series() { .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, bucket, entry, None, purger) .await .unwrap(); let reused_path = series_index_path(region.region_id, entry.index_uuid); @@ -891,6 +891,7 @@ async fn test_drop_catalogs_and_retained_snapshot_after_task_stop() { let store = ObjectStore::new(Memory::default()).unwrap(); let region_id = RegionId::new(1, 1); let entry = SeriesIndexEntry { + artifact: None, file_size: 0, index_uuid: FileId::random(), bucket_start: Timestamp::new_second(0), diff --git a/src/mito2/src/series_index/version.rs b/src/mito2/src/series_index/version.rs index fa2115fc1f2..7d819a0df55 100644 --- a/src/mito2/src/series_index/version.rs +++ b/src/mito2/src/series_index/version.rs @@ -82,9 +82,15 @@ struct SeriesIndexFileHandleInner { impl Drop for SeriesIndexFileHandleInner { fn drop(&mut self) { if self.deleted.load(Ordering::Acquire) { - self.purger.purge(PurgeRequest { - file_id: self.file_id, - }); + if let Some(artifact) = &self.entry.artifact { + self.purger + .purge_objects(self.file_id, artifact.objects.clone()); + } else { + self.purger.purge(PurgeRequest { + file_id: self.file_id, + objects: None, + }); + } } } } @@ -94,6 +100,7 @@ impl Drop for SeriesIndexFileHandleInner { pub(crate) struct SeriesIndexVersion { /// Range indexes for visible SSTs; reconciliation removes IDs absent from its SST snapshot. /// Physical deletion is independently handled by the SST file purger. + pub(crate) samples: HashMap, pub(crate) range_indexes: HashMap, pub(crate) series_indexes: HashMap, pub(crate) index_buckets: BTreeMap, @@ -110,6 +117,7 @@ impl SeriesIndexVersion { IndexBucket::from_entry(handle.entry()).insert_into(&mut index_buckets); } Self { + samples: Default::default(), range_indexes, series_indexes, index_buckets, @@ -118,10 +126,15 @@ impl SeriesIndexVersion { /// Approximate installed usage; old snapshots and unpublished outputs are excluded. pub(crate) fn disk_usage(&self) -> u64 { - self.range_indexes + self.samples .values() - .map(|entry| entry.file_size) + .map(|handle| handle.entry().artifact.file_size) .sum::() + + self + .range_indexes + .values() + .map(|entry| entry.file_size) + .sum::() + self .series_indexes .values() @@ -130,6 +143,9 @@ impl SeriesIndexVersion { } fn mark_all_deleted(&self) { + self.samples + .values() + .for_each(|handle| handle.mark_deleted()); self.series_indexes .values() .for_each(SeriesIndexFileHandle::mark_deleted); @@ -140,6 +156,7 @@ impl SeriesIndexVersion { #[derive(Debug, Default)] pub(crate) struct SeriesIndexVersionControl { current: RwLock>, + pub(crate) formats: RwLock>, } impl SeriesIndexVersionControl { diff --git a/tests/compatibility/fixtures/series_index/README.md b/tests/compatibility/fixtures/series_index/README.md new file mode 100644 index 00000000000..29c3000f838 --- /dev/null +++ b/tests/compatibility/fixtures/series_index/README.md @@ -0,0 +1,10 @@ +# Series-index catalog compatibility + +`legacy_catalog.json` is a synthetic fixture matching `SeriesIndexEntry` at +`3ad5f963285e51166fd8789920e4b3bf2cf60423`, before optional artifact metadata. +It is not captured from a released binary. Its entry must remain a reusable +Parquet index and serialize without adding an artifact field. + +Run `cargo nextest run -p mito2 -E 'test(legacy_series_catalog_fixture)'`. +The owning test also checks the persisted artifact format names. Ordinary +index-reader tests exercise Parquet files without a full-primary-key column. diff --git a/tests/compatibility/fixtures/series_index/legacy_catalog.json b/tests/compatibility/fixtures/series_index/legacy_catalog.json new file mode 100644 index 00000000000..e119e602a8d --- /dev/null +++ b/tests/compatibility/fixtures/series_index/legacy_catalog.json @@ -0,0 +1,27 @@ +{ + "indexes": [ + { + "file_size": 123, + "index_uuid": "00000000-0000-0000-0000-000000000001", + "bucket_start": { + "value": 0, + "unit": "Second" + }, + "bucket_end": { + "value": 100, + "unit": "Second" + }, + "source_file_ids": [], + "min_file_sequence": 1, + "max_file_sequence": 2, + "compaction_window_secs": 100, + "window_sequences": { + "0": { + "start": 0, + "end": 100, + "max_sequence": 2 + } + } + } + ] +}