rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-11-06 19:06:58 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(mito): use stored recovery-set bytes for manifest size (#10142)

Browse Source 
* fix(mito): track stored bytes for manifest recovery

Signed-off-by: WenyXu <wenymedia@gmail.com>

* fix(mito): retain manifest sizes on checkpoint and sync

Signed-off-by: WenyXu <wenymedia@gmail.com>

* test(mito): cover logical manifest size accounting

Signed-off-by: WenyXu <wenymedia@gmail.com>

* docs(mito): clarify manifest size retention invariants

Signed-off-by: WenyXu <wenymedia@gmail.com>

---------

Signed-off-by: WenyXu <wenymedia@gmail.com>
This commit is contained in:
Weny Xu
2026-11-05 04:55:53 +00:00
committed by GitHub
parent c02e1f58af
commit d7036df73c
10 changed files with 602 additions and 138 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/catalog/src/system_schema/information_schema/region_statistics.rs
+6 -1
View File
@@ -75,7 +75,12 @@ const INIT_CAPACITY: usize = 42;
/// - `query_scanned_bytes`: The total bytes scanned by queries since region opened.
/// - `memtable_size`: The memtable size in bytes.
/// - `disk_size`: The approximate disk size in bytes.
/// - `manifest_size`: The manifest size in bytes.
/// - `manifest_size`: Stored bytes of the latest observed checkpoint and subsequent
/// deltas (compressed when enabled), excluding staging and `_last_checkpoint`.
/// Older files retained on disk, including failed deletions, are not counted.
/// Followers update this estimate on sync; metadata read failures can leave it stale.
/// Promotion does not refresh it, so an incomplete estimate can persist without
/// further syncs or checkpoints.
/// - `sst_size`: The sst data files size in bytes.
/// - `index_size`: The sst index files size in bytes.
/// - `engine`: The engine type.
src/mito2/src/manifest/checkpointer.rs
+3
View File
@@ -78,6 +78,9 @@ impl Inner {
self.last_checkpoint_version
.store(version, Ordering::Relaxed);
// Logical retention changes when the checkpoint is published, even if cleanup fails.
self.manifest_store.retain_manifest_size_after(version);
if let Err(e) = self.manifest_store.delete_until(version, true).await {
warn!(e; "Failed to delete manifest actions until version {} for region {}, leftover files will be ignored on recovery", version, region_id);
}
src/mito2/src/manifest/manager.rs
+37 -49
View File
@@ -17,7 +17,7 @@ use std::sync::atomic::{AtomicU64, Ordering};
use std::time::Instant;
use common_datasource::compression::CompressionType;
use common_telemetry::{debug, info};
use common_telemetry::{debug, info, warn};
use futures::TryStreamExt;
use object_store::ObjectStore;
use snafu::{OptionExt, ResultExt, ensure};
@@ -236,7 +236,7 @@ impl RegionManifestManager {
manifest
.removed_files
.update_file_removed_cnt_to_stats(stats);
Ok(Self {
let manager = Self {
store,
last_version: manifest_version,
checkpointer,
@@ -244,7 +244,8 @@ impl RegionManifestManager {
staging_manifest: None,
stats: stats.clone(),
stopped: false,
})
};
Ok(manager)
}
/// Opens an existing manifest.
@@ -276,9 +277,9 @@ impl RegionManifestManager {
let checkpoint = Self::last_checkpoint(&mut store).await?;
let last_checkpoint_version = checkpoint
.as_ref()
.map(|(checkpoint, _)| checkpoint.last_version)
.map(|checkpoint| checkpoint.last_version)
.unwrap_or(MIN_VERSION);
let mut manifest_builder = if let Some((checkpoint, _)) = checkpoint {
let mut manifest_builder = if let Some(checkpoint) = checkpoint {
info!(
"Recover region manifest {} from checkpoint version {}",
options.manifest_dir, checkpoint.last_version
@@ -305,8 +306,6 @@ impl RegionManifestManager {
for (manifest_version, raw_action_list) in manifests {
let action_list = RegionMetaActionList::decode(&raw_action_list)?;
// set manifest size after last checkpoint
store.set_delta_file_size(manifest_version, raw_action_list.len() as u64);
for action in action_list.actions {
match action {
RegionMetaAction::Change(action) => {
@@ -365,7 +364,7 @@ impl RegionManifestManager {
version,
open_start.elapsed(),
);
Ok(Some(Self {
let manager = Self {
store,
last_version: manifest_version,
checkpointer,
@@ -374,7 +373,8 @@ impl RegionManifestManager {
staging_manifest: None,
stats: stats.clone(),
stopped: false,
}))
};
Ok(Some(manager))
}
/// Stops the manager.
@@ -396,18 +396,30 @@ impl RegionManifestManager {
&mut self,
target_version: ManifestVersion,
) -> Result<ManifestVersion> {
let (version, loaded_checkpoint) = self.install_manifest_to_inner(target_version).await?;
if !loaded_checkpoint && let Err(error) = self.store.update_manifest_size(version).await {
warn!(error; "Failed to update manifest size after sync for region {}", self.manifest.metadata.region_id);
}
Ok(version)
}
async fn install_manifest_to_inner(
&mut self,
target_version: ManifestVersion,
) -> Result<(ManifestVersion, bool)> {
let _t = MANIFEST_OP_ELAPSED
.with_label_values(&["install_manifest_to"])
.start_timer();
let last_version = self.last_version();
let mut loaded_checkpoint = false;
// Case 1: If the target version is less than the current version, return the current version.
if last_version >= target_version {
debug!(
"Target version {} is less than or equal to the current version {}, region: {}, skip install",
target_version, last_version, self.manifest.metadata.region_id
);
return Ok(last_version);
return Ok((last_version, false));
}
ensure!(
@@ -437,9 +449,10 @@ impl RegionManifestManager {
last_version, self.manifest.metadata.region_id
);
let last_version = self.install_last_checkpoint().await?;
loaded_checkpoint = true;
// Case 2.1: If the installed checkpoint version is greater than or equal to the target version, return the last version.
if last_version >= target_version {
return Ok(last_version);
return Ok((last_version, true));
}
// Fetches manifests from the installed version strictly.
@@ -465,8 +478,6 @@ impl RegionManifestManager {
RegionManifestBuilder::with_checkpoint(Some(self.manifest.as_ref().clone()));
for (manifest_version, raw_action_list) in manifests {
self.store
.set_delta_file_size(manifest_version, raw_action_list.len() as u64);
let action_list = RegionMetaActionList::decode(&raw_action_list)?;
for action in action_list.actions {
match action {
@@ -514,23 +525,19 @@ impl RegionManifestManager {
version, last_version, self.manifest.metadata.region_id
);
Ok(last_version)
Ok((last_version, loaded_checkpoint))
}
/// Installs the last checkpoint.
pub(crate) async fn install_last_checkpoint(&mut self) -> Result<ManifestVersion> {
let last_version = self.last_version();
let Some((checkpoint, checkpoint_size)) = Self::last_checkpoint(&mut self.store).await?
else {
let Some(checkpoint) = Self::last_checkpoint(&mut self.store).await? else {
return NoCheckpointSnafu {
region_id: self.manifest.metadata.region_id,
last_version,
}
.fail();
};
self.store.reset_manifest_size();
self.store
.set_checkpoint_file_size(checkpoint.last_version, checkpoint_size);
let builder = RegionManifestBuilder::with_checkpoint(checkpoint.checkpoint);
let manifest = builder.try_build()?;
let last_version = self.set_version(manifest.manifest_version);
@@ -736,15 +743,15 @@ impl RegionManifestManager {
/// Fetches the last [RegionCheckpoint] from storage.
///
/// If the checkpoint is not found, returns `None`.
/// Otherwise, returns the checkpoint and the size of the checkpoint.
/// Otherwise, returns the decoded checkpoint.
pub(crate) async fn last_checkpoint(
store: &mut ManifestObjectStore,
) -> Result<Option<(RegionCheckpoint, u64)>> {
) -> Result<Option<RegionCheckpoint>> {
let last_checkpoint = store.load_last_checkpoint().await?;
if let Some((_, bytes)) = last_checkpoint {
let checkpoint = RegionCheckpoint::decode(&bytes)?;
Ok(Some((checkpoint, bytes.len() as u64)))
Ok(Some(checkpoint))
} else {
Ok(None)
}
@@ -856,7 +863,7 @@ mod test {
use super::*;
use crate::manifest::action::{RegionChange, RegionEdit};
use crate::manifest::tests::utils::basic_region_metadata;
use crate::manifest::tests::utils::{basic_region_metadata, manifest_recovery_usage};
use crate::test_util::TestEnv;
#[tokio::test]
@@ -972,26 +979,6 @@ mod test {
manager.validate_manifest(&new_metadata, 1);
}
/// Just for test, refer to wal_dir_usage in src/store-api/src/logstore.rs.
async fn manifest_dir_usage(path: &str) -> u64 {
let mut size = 0;
let mut read_dir = tokio::fs::read_dir(path).await.unwrap();
while let Ok(dir_entry) = read_dir.next_entry().await {
let Some(entry) = dir_entry else {
break;
};
if entry.file_type().await.unwrap().is_file() {
let file_name = entry.file_name().into_string().unwrap();
if file_name.contains(".checkpoint") || file_name.contains(".json") {
let file_size = entry.metadata().await.unwrap().len() as usize;
debug!("File: {file_name:?}, size: {file_size}");
size += file_size;
}
}
}
size as u64
}
#[tokio::test]
async fn test_manifest_size() {
let metadata = Arc::new(basic_region_metadata());
@@ -1028,7 +1015,7 @@ mod test {
// get manifest size
let manifest_size = manager.manifest_usage();
assert_eq!(manifest_size, manifest_dir_usage(&manifest_dir).await);
assert_eq!(manifest_size, manifest_recovery_usage(&manifest_dir).await);
// update 10 times nop_action to trigger checkpoint
for _ in 0..10 {
@@ -1055,10 +1042,9 @@ mod test {
// check manifest size again
let manifest_size = manager.manifest_usage();
assert_eq!(manifest_size, manifest_dir_usage(&manifest_dir).await);
assert_eq!(manifest_size, manifest_recovery_usage(&manifest_dir).await);
// Reopen the manager,
// we just calculate the size from the latest checkpoint file
// Reopening reconstructs the same logical recovery set from stored bytes.
manager.stop().await;
let manager = env
.create_manifest_manager(CompressionType::Uncompressed, 10, None)
@@ -1068,7 +1054,9 @@ mod test {
manager.validate_manifest(&new_metadata, 11);
// get manifest size again
let manifest_size = manager.manifest_usage();
assert_eq!(manifest_size, 1398);
assert_eq!(
manager.manifest_usage(),
manifest_recovery_usage(&manifest_dir).await
);
}
}
src/mito2/src/manifest/storage.rs
+48 -37
View File
@@ -262,7 +262,7 @@ impl ManifestObjectStore {
/// Delete manifest files that version < end.
/// If keep_last_checkpoint is true, the last checkpoint file will be kept.
/// ### Return
/// The number of deleted files.
/// The number of files whose deletion was acknowledged by the object store.
pub async fn delete_until(
&self,
end: ManifestVersion,
@@ -336,9 +336,9 @@ impl ManifestObjectStore {
// Try batch delete first. On failure, fall back to per-file deletes.
// This is a workaround for S3-compatible object stores that do not support batch delete. See issue #7986.
let mut succeeded = vec![false; del_entries.len()];
let mut deleted = 0;
match self.object_store.delete_iter(paths.clone()).await {
Ok(()) => succeeded.fill(true),
Ok(()) => deleted = total,
Err(batch_err) => {
warn!(
batch_err;
@@ -354,26 +354,11 @@ impl ManifestObjectStore {
);
break;
}
succeeded[i] = true;
deleted += 1;
}
}
}
let mut deleted = 0usize;
for (i, (_, is_checkpoint, version)) in del_entries.iter().enumerate() {
if !succeeded[i] {
continue;
}
deleted += 1;
if *is_checkpoint {
self.size_tracker
.remove(&size_tracker::FileKey::Checkpoint(*version));
} else {
self.size_tracker
.remove(&size_tracker::FileKey::Delta(*version));
}
}
Ok(deleted)
}
@@ -404,8 +389,13 @@ impl ManifestObjectStore {
/// Load the latest checkpoint.
/// Return manifest version and the raw [RegionCheckpoint](crate::manifest::action::RegionCheckpoint) content if any
/// Also applies the checkpoint retention rule to the tracked size.
pub async fn load_last_checkpoint(&mut self) -> Result<Option<(ManifestVersion, Vec<u8>)>> {
self.checkpoint_storage.load_last_checkpoint().await
let checkpoint = self.checkpoint_storage.load_last_checkpoint().await?;
if let Some((version, _)) = &checkpoint {
self.retain_manifest_size_after(*version);
}
Ok(checkpoint)
}
/// Compute the size(Byte) in manifest size map.
@@ -413,19 +403,36 @@ impl ManifestObjectStore {
self.size_tracker.total()
}
/// Resets the size of all files.
pub(crate) fn reset_manifest_size(&mut self) {
self.size_tracker.reset();
/// Accounts for the latest checkpoint already covered by this follower's replay.
/// Only a changed or previously unaccounted checkpoint needs a stat.
pub(crate) async fn update_manifest_size(
&self,
replayed_version: ManifestVersion,
) -> Result<()> {
let Some(metadata) = self
.checkpoint_storage
.read_last_checkpoint_metadata()
.await?
else {
return Ok(());
};
if metadata.version > replayed_version {
// Pruning now would stop counting deltas still in this follower's recovery set.
// Retention applies once replay covers this version or sync installs the checkpoint.
return Ok(());
}
if !self.size_tracker.contains_checkpoint(metadata.version) {
self.checkpoint_storage
.record_checkpoint_size(metadata.version)
.await?;
}
self.retain_manifest_size_after(metadata.version);
Ok(())
}
/// Set the size of the delta file by delta version.
pub(crate) fn set_delta_file_size(&mut self, version: ManifestVersion, size: u64) {
self.size_tracker.record_delta(version, size);
}
/// Set the size of the checkpoint file by checkpoint version.
pub(crate) fn set_checkpoint_file_size(&self, version: ManifestVersion, size: u64) {
self.size_tracker.record_checkpoint(version, size);
/// Prunes logical accounting independently of physical cleanup success.
pub(crate) fn retain_manifest_size_after(&self, version: ManifestVersion) {
self.size_tracker.retain_after(version);
}
/// Fetch all staging manifest files and return them as (version, action_list) pairs.
@@ -701,8 +708,10 @@ mod tests {
// delete 3 manifest files
assert_eq!(log_store.delete_until(3, false).await.unwrap(), 3);
// manifest files size after delete
assert_eq!(log_store.total_manifest_size(), 39);
// Physical deletion does not define the logical recovery set.
assert_eq!(log_store.total_manifest_size(), 63);
log_store.retain_manifest_size_after(5);
assert_eq!(log_store.total_manifest_size(), 23);
// delete all manifest files
assert_eq!(
@@ -713,7 +722,7 @@ mod tests {
3
);
assert_eq!(log_store.total_manifest_size(), 0);
assert_eq!(log_store.total_manifest_size(), 23);
}
#[tokio::test]
@@ -739,8 +748,10 @@ mod tests {
// delete 3 manifest files
assert_eq!(log_store.delete_until(3, false).await.unwrap(), 3);
// manifest files size after delete
assert_eq!(log_store.total_manifest_size(), 97);
// Physical deletion does not define the logical recovery set.
assert_eq!(log_store.total_manifest_size(), 181);
log_store.retain_manifest_size_after(5);
assert_eq!(log_store.total_manifest_size(), 41);
// delete all manifest files
assert_eq!(
@@ -751,7 +762,7 @@ mod tests {
3
);
assert_eq!(log_store.total_manifest_size(), 0);
assert_eq!(log_store.total_manifest_size(), 41);
}
#[tokio::test]
src/mito2/src/manifest/storage/checkpoint.rs
+44 -3
View File
@@ -16,7 +16,7 @@ use std::collections::HashMap;
use std::sync::Arc;
use common_datasource::compression::CompressionType;
use common_telemetry::debug;
use common_telemetry::{debug, warn};
use object_store::{ErrorKind, ObjectStore};
use serde::{Deserialize, Serialize};
use snafu::ResultExt;
@@ -104,6 +104,7 @@ impl<T: Tracker> CheckpointStorage<T> {
// Try to get from cache first
if let Some(data) = get_from_cache(self.manifest_cache.as_ref(), &path).await {
verify_checksum(&data, metadata.checksum)?;
self.record_cached_checkpoint_size(version, &path).await;
return Ok(Some((version, data)));
}
@@ -145,6 +146,8 @@ impl<T: Tracker> CheckpointStorage<T> {
get_from_cache(self.manifest_cache.as_ref(), &fall_back_path).await
{
verify_checksum(&data, metadata.checksum)?;
self.record_cached_checkpoint_size(version, &fall_back_path)
.await;
return Ok(Some((version, data)));
}
@@ -186,6 +189,14 @@ impl<T: Tracker> CheckpointStorage<T> {
/// Load the latest checkpoint.
/// Return manifest version and the raw [RegionCheckpoint](crate::manifest::action::RegionCheckpoint) content if any
pub async fn load_last_checkpoint(&mut self) -> Result<Option<(ManifestVersion, Vec<u8>)>> {
let Some(metadata) = self.read_last_checkpoint_metadata().await? else {
return Ok(None);
};
self.load_checkpoint(metadata).await
}
/// Reads the small checkpoint pointer without loading its payload.
pub(crate) async fn read_last_checkpoint_metadata(&self) -> Result<Option<CheckpointMetadata>> {
let last_checkpoint_path = self.last_checkpoint_path();
// Fetch from remote object store without cache
@@ -206,7 +217,37 @@ impl<T: Tracker> CheckpointStorage<T> {
last_checkpoint_path, checkpoint_metadata
);
self.load_checkpoint(checkpoint_metadata).await
Ok(Some(checkpoint_metadata))
}
/// A cache hit contains decoded bytes; accounting failures must not prevent recovery.
async fn record_cached_checkpoint_size(&self, version: ManifestVersion, path: &str) {
match self.object_store.stat(path).await {
Ok(metadata) => self.size_tracker.record(version, metadata.content_length()),
Err(error) => warn!(error; "Failed to account for cached checkpoint {}", path),
}
}
/// Records stored checkpoint bytes, including the legacy uncompressed path.
pub(crate) async fn record_checkpoint_size(&self, version: ManifestVersion) -> Result<()> {
let path = self.checkpoint_file_path(version);
let metadata = match self.object_store.stat(&path).await {
Ok(metadata) => metadata,
Err(error)
if error.kind() == ErrorKind::NotFound
&& self.compress_type != FALL_BACK_COMPRESS_TYPE =>
{
let path = gen_path(
&self.path,
&checkpoint_file(version),
FALL_BACK_COMPRESS_TYPE,
);
self.object_store.stat(&path).await.context(OpenDalSnafu)?
}
Err(error) => return Err(error).context(OpenDalSnafu),
};
self.size_tracker.record(version, metadata.content_length());
Ok(())
}
/// Save the checkpoint manifest file.
@@ -234,7 +275,6 @@ impl<T: Tracker> CheckpointStorage<T> {
data,
)
.await?;
self.size_tracker.record(version, checkpoint_size as u64);
// Because last checkpoint file only contain size and version, which is tiny, so we don't compress it.
let last_checkpoint_path = self.last_checkpoint_path();
@@ -257,6 +297,7 @@ impl<T: Tracker> CheckpointStorage<T> {
.await
.context(OpenDalSnafu)?;
self.size_tracker.record(version, checkpoint_size as u64);
Ok(())
}
}
src/mito2/src/manifest/storage/delta.rs
+11 -1
View File
@@ -15,7 +15,7 @@
use std::sync::Arc;
use common_datasource::compression::CompressionType;
use common_telemetry::debug;
use common_telemetry::{debug, warn};
use futures::TryStreamExt;
use futures::future::try_join_all;
use object_store::{Entry, ErrorKind, Lister, ObjectStore};
@@ -208,6 +208,15 @@ impl<T: Tracker> DeltaStorage<T> {
let cache_key = entry.path();
// Try to get from cache first
if let Some(data) = get_from_cache(self.manifest_cache.as_ref(), cache_key).await {
let size = entry.metadata().content_length();
if size > 0 {
self.delta_tracker.record(*v, size);
} else {
match self.object_store.stat(entry.path()).await {
Ok(metadata) => self.delta_tracker.record(*v, metadata.content_length()),
Err(error) => warn!(error; "Failed to account for cached manifest delta {}", entry.path()),
}
}
return Ok((*v, data));
}
@@ -223,6 +232,7 @@ impl<T: Tracker> DeltaStorage<T> {
}
Err(error) => return Err(error).context(OpenDalSnafu),
};
self.delta_tracker.record(*v, bytes.len() as u64);
let data = compress_type
.decode(bytes)
.await
src/mito2/src/manifest/storage/size_tracker.rs
+82 -25
View File
@@ -65,15 +65,21 @@ impl Tracker for NoopTracker {
#[derive(Debug, Clone, Default)]
pub(crate) struct SizeTracker {
file_sizes: Arc<RwLock<HashMap<FileKey, u64>>>,
inventory: Arc<RwLock<Inventory>>,
total_size: Arc<AtomicU64>,
}
#[derive(Debug, Default)]
struct Inventory {
file_sizes: HashMap<FileKey, u64>,
checkpoint_version: Option<ManifestVersion>,
}
impl SizeTracker {
/// Returns a new [SizeTracker].
pub fn new(total_size: Arc<AtomicU64>) -> Self {
Self {
file_sizes: Arc::new(RwLock::new(HashMap::new())),
inventory: Arc::default(),
total_size,
}
}
@@ -92,39 +98,90 @@ impl SizeTracker {
}
}
/// Records a delta file size.
pub(crate) fn record_delta(&self, version: ManifestVersion, size: u64) {
self.record(FileKey::Delta(version), size);
}
/// Records a checkpoint file size.
pub(crate) fn record_checkpoint(&self, version: ManifestVersion, size: u64) {
self.record(FileKey::Checkpoint(version), size);
}
/// Removes a file from tracking.
pub(crate) fn remove(&self, key: &FileKey) {
if let Some(size) = self.file_sizes.write().unwrap().remove(key) {
self.total_size.fetch_sub(size, Ordering::Relaxed);
}
}
/// Returns the total tracked size.
pub(crate) fn total(&self) -> u64 {
self.total_size.load(Ordering::Relaxed)
}
/// Resets all tracking.
pub(crate) fn reset(&self) {
self.file_sizes.write().unwrap().clear();
self.total_size.store(0, Ordering::Relaxed);
/// Whether this checkpoint's stored size is already known.
pub(crate) fn contains_checkpoint(&self, version: ManifestVersion) -> bool {
self.inventory
.read()
.unwrap()
.file_sizes
.contains_key(&FileKey::Checkpoint(version))
}
/// Keeps the latest observed checkpoint and deltas after it, regardless of cleanup.
/// Physical storage may retain older checkpoints and deltas for longer.
pub(crate) fn retain_after(&self, version: ManifestVersion) {
let mut inventory = self.inventory.write().unwrap();
if inventory
.checkpoint_version
.is_some_and(|current| current > version)
{
return;
}
inventory.checkpoint_version = Some(version);
inventory.file_sizes.retain(|key, _| match key {
FileKey::Delta(v) => *v > version,
FileKey::Checkpoint(v) => *v >= version,
});
self.total_size
.store(inventory.file_sizes.values().sum(), Ordering::Relaxed);
}
fn record(&self, key: FileKey, size: u64) {
// Remove the old size if present
if let Some(old_size) = self.file_sizes.write().unwrap().insert(key, size) {
let mut inventory = self.inventory.write().unwrap();
// Manifest versions are monotonic: deltas at or below the checkpoint watermark
// are superseded, even if a read started before publication finishes afterward.
// Such a late read must not resurrect old entries.
if inventory
.checkpoint_version
.is_some_and(|version| match key {
FileKey::Delta(v) => v <= version,
FileKey::Checkpoint(v) => v < version,
})
{
return;
}
if let Some(old_size) = inventory.file_sizes.insert(key, size) {
self.total_size.fetch_sub(old_size, Ordering::Relaxed);
}
self.total_size.fetch_add(size, Ordering::Relaxed);
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn retain_checkpoint_and_later_deltas() {
let tracker = SizeTracker::default();
tracker.retain_after(0);
assert_eq!(tracker.total(), 0);
for (key, size) in [
(FileKey::Delta(1), 10),
(FileKey::Delta(2), 20),
(FileKey::Delta(3), 30),
(FileKey::Checkpoint(1), 40),
(FileKey::Checkpoint(2), 50),
] {
tracker.record(key, size);
}
tracker.retain_after(2);
assert_eq!(tracker.total(), 80);
assert_eq!(
tracker.inventory.read().unwrap().file_sizes,
HashMap::from([(FileKey::Delta(3), 30), (FileKey::Checkpoint(2), 50)])
);
// Late reads and an older retention request cannot roll accounting back.
tracker.record(FileKey::Delta(2), 20);
tracker.record(FileKey::Checkpoint(1), 40);
tracker.retain_after(1);
assert_eq!(tracker.total(), 80);
tracker.record(FileKey::Delta(3), 35);
assert_eq!(tracker.total(), 85);
}
}
src/mito2/src/manifest/tests/checkpoint.rs
+312 -21
View File
@@ -33,8 +33,12 @@ use crate::manifest::action::{
};
use crate::manifest::manager::RegionManifestManager;
use crate::manifest::storage::checkpoint::CheckpointMetadata;
use crate::manifest::storage::is_delta_file;
use crate::manifest::tests::utils::basic_region_metadata;
use crate::manifest::storage::{
checkpoint_file, delta_file, file_version, gen_path, is_delta_file,
};
use crate::manifest::tests::utils::{
basic_region_metadata, manifest_dir_usage, manifest_recovery_usage,
};
use crate::sst::file::FileMeta;
use crate::test_util::{CheckpointTaskBlocker, TestEnv};
@@ -332,7 +336,6 @@ async fn generate_checkpoint_with_compression_types(
.await
.unwrap()
.unwrap()
.0
}
fn generate_action_lists(num: usize) -> (Vec<FileId>, Vec<RegionMetaActionList>) {
@@ -480,7 +483,10 @@ async fn manifest_install_manifest_to_with_checkpoint() {
for file_id in files.iter() {
assert!(another_manager.manifest().files.contains_key(file_id));
}
assert!(another_manager.store().total_manifest_size() > 4000);
assert_eq!(
another_manager.manifest_usage(),
manifest_recovery_usage(env.data_home().join("manifest")).await
);
}
#[tokio::test]
@@ -525,20 +531,27 @@ async fn test_checkpoint_bypass_in_staging_mode() {
assert_eq!(last_version, 16);
}
/// A deleter that fails on `close`, simulating the S3 batch-delete failure
/// described in issue #7986.
/// Simulates both a rejected delete and a close failure after eager deletion.
struct FailingDeleter {
inner: oio::Deleter,
close_calls: Arc<AtomicUsize>,
failure_calls: Arc<AtomicUsize>,
fail_delete: bool,
}
impl oio::Delete for FailingDeleter {
async fn delete(&mut self, path: &str, args: OpDelete) -> MockResult<()> {
if self.fail_delete {
self.failure_calls.fetch_add(1, Ordering::Relaxed);
return Err(MockError::new(
ErrorKind::Unexpected,
"mock rejected delete",
));
}
self.inner.delete(path, args).await
}
async fn close(&mut self) -> MockResult<()> {
self.close_calls.fetch_add(1, Ordering::Relaxed);
self.failure_calls.fetch_add(1, Ordering::Relaxed);
Err(MockError::new(
ErrorKind::Unexpected,
"mock manifest delete close failure",
@@ -546,17 +559,21 @@ impl oio::Delete for FailingDeleter {
}
}
#[rstest::rstest]
#[case::close_failure(false)]
#[case::delete_failure(true)]
#[tokio::test]
async fn checkpoint_advances_and_recovery_works_when_delete_fails() {
async fn manifest_size_ignores_leftover_files_after_failed_delete(#[case] fail_delete: bool) {
common_telemetry::init_default_ut_logging();
let close_calls = Arc::new(AtomicUsize::new(0));
let factory_close_calls = close_calls.clone();
let failure_calls = Arc::new(AtomicUsize::new(0));
let factory_failure_calls = failure_calls.clone();
let mock_layer = MockLayerBuilder::default()
.deleter_factory(Arc::new(move |inner| {
Box::new(FailingDeleter {
inner,
close_calls: factory_close_calls.clone(),
failure_calls: factory_failure_calls.clone(),
fail_delete,
})
}))
.build()
@@ -578,7 +595,7 @@ async fn checkpoint_advances_and_recovery_works_when_delete_fails() {
}
// The checkpointer must have attempted to delete stale files at least once.
assert!(close_calls.load(Ordering::Relaxed) > 0);
assert!(failure_calls.load(Ordering::Relaxed) > 0);
// Despite delete failures, the in-memory checkpoint marker advances so
// subsequent `maybe_do_checkpoint` calls compute correct ranges.
@@ -593,22 +610,25 @@ async fn checkpoint_advances_and_recovery_works_when_delete_fails() {
.expect("checkpoint should be durable");
assert_eq!(last_version, 10);
// Stale deltas below the checkpoint version must still be present because
// the mocked deleter refused them.
// A rejected delete preserves stale files; a close failure after eager FS
// deletion does not. Exclude the current delta from the stale-file count.
let file_names: Vec<String> = manager
.store()
.delta_storage()
.get_paths(None, |e| Some(e.name().to_string()))
.await
.unwrap();
let stale_delta_count = file_names.iter().filter(|name| is_delta_file(name)).count();
assert!(
stale_delta_count > 0,
"expected leftover delta files after failed delete, got {:?}",
file_names,
let stale_delta_count = file_names
.iter()
.filter(|name| is_delta_file(name) && file_version(name) < 10)
.count();
assert_eq!(stale_delta_count > 0, fail_delete, "{file_names:?}");
assert_eq!(
manager.manifest_usage(),
manifest_recovery_usage(env.data_home().join("manifest")).await
);
// Recovery must succeed despite the leftover deltas.
// Recovery must succeed in both failure modes.
manager.stop().await;
let reopened = env
.create_manifest_manager(CompressionType::Uncompressed, 1, None)
@@ -616,6 +636,15 @@ async fn checkpoint_advances_and_recovery_works_when_delete_fails() {
.unwrap()
.expect("manifest should be recoverable");
assert_eq!(reopened.manifest().manifest_version, 10);
assert_eq!(
manager.manifest_usage(),
manifest_recovery_usage(env.data_home().join("manifest")).await
);
assert_eq!(
reopened.manifest_usage(),
manifest_recovery_usage(env.data_home().join("manifest")).await
);
assert!(manifest_dir_usage(env.data_home().join("manifest")).await > reopened.manifest_usage());
}
#[tokio::test]
@@ -781,3 +810,265 @@ async fn completed_checkpoint_is_reaped_before_scheduling_the_next_one() {
assert_eq!(2, manager.checkpointer().last_checkpoint_version());
}
/// Retention runs inside the checkpoint task, so waiting for task completion also
/// makes logical size accounting observable to the assertions below.
async fn wait_for_manifest_checkpoint(manager: &RegionManifestManager) {
tokio::time::timeout(Duration::from_secs(10), async {
while manager.checkpointer().is_doing_checkpoint() {
tokio::time::sleep(Duration::from_millis(10)).await;
}
})
.await
.unwrap();
}
#[tokio::test]
async fn manifest_size_matches_recovery_bytes_after_open() {
for compress_type in [
CompressionType::Uncompressed,
CompressionType::Gzip,
CompressionType::Zstd,
] {
for checkpoint_distance in [0, 3] {
let (env, mut manager) = build_manager(checkpoint_distance, compress_type).await;
for _ in 0..7 {
manager.update(nop_action(), false).await.unwrap();
wait_for_manifest_checkpoint(&manager).await;
}
let expected = manifest_recovery_usage(env.data_home().join("manifest")).await;
assert_eq!(manager.manifest_usage(), expected);
manager.stop().await;
let reopened = reopen_manager(&env, checkpoint_distance, compress_type).await;
assert_eq!(reopened.manifest_usage(), expected);
}
}
}
#[rstest::rstest]
#[case::without_checkpoints(0)]
#[case::with_checkpoints(3)]
#[tokio::test]
async fn manifest_size_follower_converges_after_sync(#[case] distance: u64) {
for compression in [
CompressionType::Uncompressed,
CompressionType::Gzip,
CompressionType::Zstd,
] {
let (env, mut leader) = build_manager(distance, compression).await;
let mut follower = reopen_manager(&env, distance, compression).await;
for _ in 0..12 {
leader.update(nop_action(), false).await.unwrap();
leader.wait_for_pending_checkpoint().await;
let expected = manifest_recovery_usage(env.data_home().join("manifest")).await;
assert_eq!(leader.manifest_usage(), expected);
follower
.install_manifest_to(leader.last_version())
.await
.unwrap();
assert_eq!(follower.manifest_usage(), expected);
}
// Promotion requires no accounting hook. New leader writes and checkpoints
// continue from the same logical recovery set.
leader.stop().await;
for _ in 0..4 {
follower.update(nop_action(), false).await.unwrap();
follower.wait_for_pending_checkpoint().await;
assert_eq!(
follower.manifest_usage(),
manifest_recovery_usage(env.data_home().join("manifest")).await
);
}
}
}
#[tokio::test]
async fn noop_sync_accounts_for_new_checkpoint_without_listing() {
let lists = Arc::new(AtomicUsize::new(0));
let counter = lists.clone();
let layer = MockLayerBuilder::default()
.lister_factory(Arc::new(move |_, _, inner| {
counter.fetch_add(1, Ordering::Relaxed);
inner
}))
.build()
.unwrap();
let env = TestEnv::new().await.with_mock_layer(layer);
let mut leader = build_manager_with_initial_metadata(&env, 0, CompressionType::Gzip).await;
for _ in 0..3 {
leader.update(nop_action(), false).await.unwrap();
}
let mut follower = reopen_manager(&env, 0, CompressionType::Gzip).await;
let version = leader.last_version();
let checkpoint = RegionCheckpoint {
last_version: version,
compacted_actions: 4,
checkpoint: Some(leader.manifest().as_ref().clone()),
};
leader
.store()
.save_checkpoint(version, &checkpoint.encode().unwrap())
.await
.unwrap();
let expected = manifest_recovery_usage(env.data_home().join("manifest")).await;
assert_ne!(follower.manifest_usage(), expected);
let before = lists.load(Ordering::Relaxed);
// Pointer decoding and checkpoint stat failures are diagnostic only. The
// next request must retry the missing accounting rather than prune blindly.
let previous = follower.manifest_usage();
let directory = env.data_home().join("manifest");
let pointer = directory.join("_last_checkpoint");
let bytes = tokio::fs::read(&pointer).await.unwrap();
tokio::fs::write(&pointer, b"invalid checkpoint metadata")
.await
.unwrap();
follower.install_manifest_to(version).await.unwrap();
assert_eq!(follower.manifest_usage(), previous);
tokio::fs::write(&pointer, bytes).await.unwrap();
let checkpoint_path = directory.join(gen_path(
"",
&checkpoint_file(version),
CompressionType::Gzip,
));
let hidden = env.data_home().join("temporarily-hidden-checkpoint");
tokio::fs::rename(&checkpoint_path, &hidden).await.unwrap();
follower.install_manifest_to(version).await.unwrap();
assert_eq!(follower.manifest_usage(), previous);
tokio::fs::rename(&hidden, &checkpoint_path).await.unwrap();
for _ in 0..3 {
follower.install_manifest_to(version).await.unwrap();
assert_eq!(follower.manifest_usage(), expected);
}
assert_eq!(lists.load(Ordering::Relaxed), before);
}
#[rstest::rstest]
#[case::listed_sizes(false, false, false)]
#[case::unknown_sizes(true, false, false)]
#[case::fallback_cache(false, true, false)]
#[case::missing_checkpoint(false, false, true)]
#[tokio::test]
async fn manifest_size_warm_cache_matches_recovery_bytes(
#[case] unknown_sizes: bool,
#[case] fallback: bool,
#[case] missing_checkpoint: bool,
) {
use common_base::readable_size::ReadableSize;
use object_store::ObjectStore;
use object_store::services::Fs;
use crate::cache::manifest_cache::ManifestCache;
use crate::manifest::manager::RegionManifestOptions;
for compress_type in [
CompressionType::Uncompressed,
CompressionType::Gzip,
CompressionType::Zstd,
] {
let stored_compression = if fallback {
CompressionType::Uncompressed
} else {
compress_type
};
let (env, mut leader) = build_manager(3, stored_compression).await;
for _ in 0..7 {
leader.update(nop_action(), false).await.unwrap();
wait_for_manifest_checkpoint(&leader).await;
}
let object_store = ObjectStore::new(
Fs::default().root(env.data_home().join("manifest").to_str().unwrap()),
)
.unwrap();
let object_store = if unknown_sizes {
object_store.layer(
MockLayerBuilder::default()
.lister_factory(Arc::new(|_, _, inner| Box::new(ZeroSizeLister(inner))))
.build()
.unwrap(),
)
} else {
object_store
};
let local_store =
ObjectStore::new(Fs::default().root(env.data_home().join("cache").to_str().unwrap()))
.unwrap();
let cache = ManifestCache::new(local_store, ReadableSize::mb(10), None, true).await;
let options = RegionManifestOptions {
manifest_dir: "/".to_string(),
object_store,
compress_type,
checkpoint_distance: 3,
remove_file_options: Default::default(),
manifest_cache: Some(cache.clone()),
};
let expected = manifest_recovery_usage(env.data_home().join("manifest")).await;
// First open fills the cache; the second reuses the cached checkpoint and deltas.
for iteration in 0..2 {
if iteration == 1 {
let checkpoint_version = leader.checkpointer().last_checkpoint_version();
let checkpoint_key = gen_path(
"/",
&checkpoint_file(checkpoint_version),
stored_compression,
);
assert!(
cache.get(&checkpoint_key).await.is_some(),
"{checkpoint_key}"
);
for version in checkpoint_version + 1..=leader.manifest().manifest_version {
let delta_key = gen_path("/", &delta_file(version), stored_compression);
assert!(
cache.get(delta_key.trim_start_matches('/')).await.is_some(),
"{delta_key}"
);
}
}
let checkpoint_path = env.data_home().join("manifest").join(gen_path(
"",
&checkpoint_file(leader.checkpointer().last_checkpoint_version()),
stored_compression,
));
let hidden = env.data_home().join("temporarily-hidden-checkpoint");
let missing_size = if missing_checkpoint && iteration == 1 {
let size = tokio::fs::metadata(&checkpoint_path).await.unwrap().len();
tokio::fs::rename(&checkpoint_path, &hidden).await.unwrap();
size
} else {
0
};
let mut reopened = RegionManifestManager::open(options.clone(), &Default::default())
.await
.unwrap()
.unwrap();
assert_eq!(reopened.manifest_usage(), expected - missing_size);
if missing_size != 0 {
tokio::fs::rename(&hidden, &checkpoint_path).await.unwrap();
reopened
.install_manifest_to(reopened.last_version())
.await
.unwrap();
assert_eq!(reopened.manifest_usage(), expected);
}
reopened.install_last_checkpoint().await.unwrap();
assert_eq!(reopened.manifest_usage(), expected);
}
}
}
/// Exercises cache-hit accounting on backends whose LIST omits object sizes.
struct ZeroSizeLister(oio::Lister);
impl oio::List for ZeroSizeLister {
async fn next(&mut self) -> MockResult<Option<oio::Entry>> {
let Some(entry) = self.0.next().await? else {
return Ok(None);
};
let (path, metadata) = entry.into_parts();
if !metadata.is_file() {
return Ok(Some(oio::Entry::new(&path, metadata)));
}
let mut builder = metadata.into_builder();
builder.set_file(0);
Ok(Some(oio::Entry::new(&path, builder.build())))
}
}
src/mito2/src/manifest/tests/utils.rs
+55
View File
@@ -12,6 +12,8 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use std::path::Path;
use api::v1::SemanticType;
use datatypes::prelude::ConcreteDataType;
use datatypes::schema::ColumnSchema;
@@ -48,3 +50,56 @@ pub fn basic_region_metadata() -> RegionMetadata {
.primary_key(vec![36]);
builder.build().unwrap()
}
/// Measures stored delta and checkpoint bytes independently of the size tracker.
pub async fn manifest_dir_usage(path: impl AsRef<Path>) -> u64 {
let mut entries = tokio::fs::read_dir(path).await.unwrap();
let mut total = 0;
while let Some(entry) = entries.next_entry().await.unwrap() {
let name = entry.file_name();
let name = name.to_str().unwrap();
if crate::manifest::storage::is_delta_file(name)
|| crate::manifest::storage::is_checkpoint_file(name)
{
total += entry.metadata().await.unwrap().len();
}
}
total
}
/// Independently measures the logical recovery set described by the persisted pointer.
/// Older physical files (including the previous checkpoint) are deliberately excluded.
pub async fn manifest_recovery_usage(path: impl AsRef<Path>) -> u64 {
let path = path.as_ref();
let checkpoint = match tokio::fs::read(path.join("_last_checkpoint")).await {
Ok(bytes) => Some(
serde_json::from_slice::<serde_json::Value>(&bytes).unwrap()["version"]
.as_u64()
.unwrap(),
),
Err(error) if error.kind() == std::io::ErrorKind::NotFound => None,
Err(error) => panic!("Cannot read checkpoint pointer: {error}"),
};
let mut entries = tokio::fs::read_dir(path).await.unwrap();
let mut total = 0;
while let Some(entry) = entries.next_entry().await.unwrap() {
let name = entry.file_name();
let Some((version, suffix)) = name.to_str().unwrap().split_once('.') else {
continue;
};
let Ok(version) = version.parse::<u64>() else {
continue;
};
let keep = if suffix.starts_with("json") {
checkpoint.is_none_or(|checkpoint| version > checkpoint)
} else if suffix.starts_with("checkpoint") {
checkpoint == Some(version)
} else {
false
};
if keep {
total += entry.metadata().await.unwrap().len();
}
}
total
}
src/store-api/src/region_engine.rs
+4 -1
View File
@@ -579,7 +579,10 @@ pub struct RegionStatistic {
pub memtable_size: u64,
/// The size of WAL in bytes.
pub wal_size: u64,
/// The size of manifest in bytes.
/// Stored bytes of the latest observed manifest checkpoint and subsequent deltas,
/// compressed when enabled. Excludes staging manifests, `_last_checkpoint`, and
/// superseded files still present on disk. This is a logical recovery-size estimate,
/// not the physical directory usage; observation failures can leave it incomplete.
pub manifest_size: u64,
/// The size of SST data files owned by this region in bytes.
///