diff --git a/src/catalog/src/system_schema/information_schema/region_statistics.rs b/src/catalog/src/system_schema/information_schema/region_statistics.rs index 8a7c19adc00..ab2a30580b0 100644 --- a/src/catalog/src/system_schema/information_schema/region_statistics.rs +++ b/src/catalog/src/system_schema/information_schema/region_statistics.rs @@ -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. diff --git a/src/mito2/src/manifest/checkpointer.rs b/src/mito2/src/manifest/checkpointer.rs index 232421a2621..3d33942a2f9 100644 --- a/src/mito2/src/manifest/checkpointer.rs +++ b/src/mito2/src/manifest/checkpointer.rs @@ -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); } diff --git a/src/mito2/src/manifest/manager.rs b/src/mito2/src/manifest/manager.rs index 5471b1883a9..1bd69475111 100644 --- a/src/mito2/src/manifest/manager.rs +++ b/src/mito2/src/manifest/manager.rs @@ -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 { + 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 { 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> { + ) -> Result> { 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 + ); } } diff --git a/src/mito2/src/manifest/storage.rs b/src/mito2/src/manifest/storage.rs index 9ebeef2e9ab..f060e55b54c 100644 --- a/src/mito2/src/manifest/storage.rs +++ b/src/mito2/src/manifest/storage.rs @@ -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)>> { - 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] diff --git a/src/mito2/src/manifest/storage/checkpoint.rs b/src/mito2/src/manifest/storage/checkpoint.rs index 77b4ff9ac68..3f40b41c771 100644 --- a/src/mito2/src/manifest/storage/checkpoint.rs +++ b/src/mito2/src/manifest/storage/checkpoint.rs @@ -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 CheckpointStorage { // 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 CheckpointStorage { 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 CheckpointStorage { /// 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)>> { + 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> { let last_checkpoint_path = self.last_checkpoint_path(); // Fetch from remote object store without cache @@ -206,7 +217,37 @@ impl CheckpointStorage { 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 CheckpointStorage { 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 CheckpointStorage { .await .context(OpenDalSnafu)?; + self.size_tracker.record(version, checkpoint_size as u64); Ok(()) } } diff --git a/src/mito2/src/manifest/storage/delta.rs b/src/mito2/src/manifest/storage/delta.rs index 8f9962a3c69..e9ef192c6d4 100644 --- a/src/mito2/src/manifest/storage/delta.rs +++ b/src/mito2/src/manifest/storage/delta.rs @@ -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 DeltaStorage { 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 DeltaStorage { } Err(error) => return Err(error).context(OpenDalSnafu), }; + self.delta_tracker.record(*v, bytes.len() as u64); let data = compress_type .decode(bytes) .await diff --git a/src/mito2/src/manifest/storage/size_tracker.rs b/src/mito2/src/manifest/storage/size_tracker.rs index 184839e74b0..60344d47cc9 100644 --- a/src/mito2/src/manifest/storage/size_tracker.rs +++ b/src/mito2/src/manifest/storage/size_tracker.rs @@ -65,15 +65,21 @@ impl Tracker for NoopTracker { #[derive(Debug, Clone, Default)] pub(crate) struct SizeTracker { - file_sizes: Arc>>, + inventory: Arc>, total_size: Arc, } +#[derive(Debug, Default)] +struct Inventory { + file_sizes: HashMap, + checkpoint_version: Option, +} + impl SizeTracker { /// Returns a new [SizeTracker]. pub fn new(total_size: Arc) -> 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); + } +} diff --git a/src/mito2/src/manifest/tests/checkpoint.rs b/src/mito2/src/manifest/tests/checkpoint.rs index 0fe77502e16..4a4c57db7f5 100644 --- a/src/mito2/src/manifest/tests/checkpoint.rs +++ b/src/mito2/src/manifest/tests/checkpoint.rs @@ -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, Vec) { @@ -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, + failure_calls: Arc, + 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 = 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> { + 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()))) + } +} diff --git a/src/mito2/src/manifest/tests/utils.rs b/src/mito2/src/manifest/tests/utils.rs index 8f41f245812..743c088e0d4 100644 --- a/src/mito2/src/manifest/tests/utils.rs +++ b/src/mito2/src/manifest/tests/utils.rs @@ -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) -> 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) -> u64 { + let path = path.as_ref(); + let checkpoint = match tokio::fs::read(path.join("_last_checkpoint")).await { + Ok(bytes) => Some( + serde_json::from_slice::(&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::() 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 +} diff --git a/src/store-api/src/region_engine.rs b/src/store-api/src/region_engine.rs index 5d7af812685..dc9124d36f8 100644 --- a/src/store-api/src/region_engine.rs +++ b/src/store-api/src/region_engine.rs @@ -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. ///