mirror of
https://github.com/GreptimeTeam/greptimedb.git
synced 2026-11-06 19:06:58 +00:00
feat(mito): scaffold series format writers and artifact lifecycle (#9788)
* feat(mito): add pluggable series artifact reconciliation Signed-off-by: evenyag <realevenyag@gmail.com> * feat(mito): prepare independent series reads from SST artifacts Signed-off-by: evenyag <realevenyag@gmail.com> * feat(mito): prepare mixed sidecar SST and memtable series reads Signed-off-by: evenyag <realevenyag@gmail.com> * refactor(mito): make prepared series sources replaceable Signed-off-by: evenyag <realevenyag@gmail.com> * fix(mito): narrow series formats to artifact lifecycle integration Signed-off-by: evenyag <realevenyag@gmail.com> * test(mito): populate artifact regression buckets above build threshold Signed-off-by: evenyag <realevenyag@gmail.com> --------- Signed-off-by: evenyag <realevenyag@gmail.com>
This commit is contained in:
Yingwen
+2
-1
@@ -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`.
|
||||
|
||||
@@ -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),
|
||||
|
||||
@@ -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::<Vec<_>>(),
|
||||
)
|
||||
.await
|
||||
.expect("index streams must not hold permits across yield")
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
(0..1001)
|
||||
.map(|tsid| MetricSeriesId { table_id: 1, tsid })
|
||||
.collect::<Vec<_>>(),
|
||||
batches.into_iter().flatten().collect::<Vec<_>>()
|
||||
);
|
||||
assert_eq!(1, semaphore.available_permits());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn index_read_failure_after_output_is_propagated() {
|
||||
let (_env, scanner, _pruner) = indexed_scanner().await;
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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<String>,
|
||||
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<SamplesFileInner>);
|
||||
|
||||
#[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());
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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,
|
||||
|
||||
@@ -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<SeriesIndexFileHandle> {
|
||||
let mut sources = Vec::<BoxedRecordBatchStream>::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<Option<crate::series_index::artifact::SamplesFileHandle>> {
|
||||
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(|_| ())
|
||||
};
|
||||
|
||||
@@ -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<crate::series_index::artifact::ArtifactMetadata>,
|
||||
/// 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<RangeIndexEntry>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Default, Serialize, Deserialize)]
|
||||
pub(crate) struct SamplesCatalog {
|
||||
pub(crate) entries: Vec<crate::series_index::artifact::SamplesEntry>,
|
||||
/// Once samples are enabled this single catalog commits all artifact references together.
|
||||
pub(crate) series: Vec<SeriesIndexEntry>,
|
||||
pub(crate) ranges: Vec<RangeIndexEntry>,
|
||||
}
|
||||
|
||||
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::<SeriesIndexCatalog>(store, &series_catalog_path(region_id))
|
||||
.await
|
||||
.unwrap_or_default();
|
||||
let version = SeriesIndexVersion::new(
|
||||
range
|
||||
.indexes
|
||||
let samples = load_catalog::<SamplesCatalog>(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::<serde_json::Value>(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),
|
||||
|
||||
@@ -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<Self>) -> Result<ArtifactMetadata>;
|
||||
async fn abort(self: Box<Self>) -> 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<Box<dyn FormatWriter>>;
|
||||
}
|
||||
|
||||
/// 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<Box<dyn FormatWriter>>;
|
||||
}
|
||||
|
||||
/// 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<Arc<dyn SeriesIndexFormat>>,
|
||||
pub(crate) samples: Option<Arc<dyn SamplesFormat>>,
|
||||
}
|
||||
@@ -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<AtomicBool>,
|
||||
}
|
||||
|
||||
struct FakeWriter {
|
||||
store: ObjectStore,
|
||||
artifact: ArtifactMetadata,
|
||||
fail: Arc<AtomicBool>,
|
||||
}
|
||||
|
||||
#[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<Self>) -> Result<ArtifactMetadata> {
|
||||
Ok(self.artifact)
|
||||
}
|
||||
async fn abort(self: Box<Self>) -> 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<Box<dyn FormatWriter>> {
|
||||
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<Box<dyn FormatWriter>> {
|
||||
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::<Vec<_>>()
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
timestamps.len(),
|
||||
batches.iter().map(|batch| batch.num_rows()).sum::<usize>()
|
||||
);
|
||||
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<mock::Metadata> {
|
||||
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();
|
||||
}
|
||||
@@ -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<SeriesIndexFileHandle>);
|
||||
struct UnpublishedSeriesFiles(
|
||||
Vec<SeriesIndexFileHandle>,
|
||||
Vec<crate::series_index::artifact::SamplesFileHandle>,
|
||||
);
|
||||
|
||||
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<SeriesIndexVersion> {
|
||||
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::<Vec<_>>();
|
||||
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<SeriesIndexVersion>) {
|
||||
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),
|
||||
|
||||
@@ -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<Vec<String>>,
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
@@ -57,6 +58,13 @@ impl Debug for IndexFilePurger {
|
||||
}
|
||||
|
||||
impl IndexFilePurger {
|
||||
pub(crate) fn purge_objects(&self, file_id: RegionFileId, objects: Vec<String>) {
|
||||
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;
|
||||
|
||||
@@ -312,6 +312,7 @@ mod tests {
|
||||
) -> (SeriesIndexFileHandle, UnboundedReceiver<PurgeRequest>) {
|
||||
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),
|
||||
|
||||
@@ -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),
|
||||
|
||||
@@ -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<FileId, crate::series_index::artifact::SamplesFileHandle>,
|
||||
pub(crate) range_indexes: HashMap<FileId, RangeIndexEntry>,
|
||||
pub(crate) series_indexes: HashMap<FileId, SeriesIndexFileHandle>,
|
||||
pub(crate) index_buckets: BTreeMap<Timestamp, IndexBucket>,
|
||||
@@ -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::<u64>()
|
||||
+ self
|
||||
.range_indexes
|
||||
.values()
|
||||
.map(|entry| entry.file_size)
|
||||
.sum::<u64>()
|
||||
+ 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<Arc<SeriesIndexVersion>>,
|
||||
pub(crate) formats: RwLock<Arc<crate::series_index::format::SeriesFormats>>,
|
||||
}
|
||||
|
||||
impl SeriesIndexVersionControl {
|
||||
|
||||
@@ -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.
|
||||
@@ -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
|
||||
}
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
Reference in New Issue
Block a user