diff --git a/rust/lance/src/dataset/cleanup.rs b/rust/lance/src/dataset/cleanup.rs index 45f23f877c3..4fcf432ac1a 100644 --- a/rust/lance/src/dataset/cleanup.rs +++ b/rust/lance/src/dataset/cleanup.rs @@ -63,7 +63,10 @@ use std::fmt::Debug; use std::{ collections::{HashMap, HashSet}, future, - sync::{Mutex, MutexGuard}, + sync::{ + Mutex, MutexGuard, + atomic::{AtomicU64, Ordering}, + }, time::Duration, }; use tokio::time::{MissedTickBehavior, interval}; @@ -184,6 +187,16 @@ impl<'a> CleanupTask<'a> { .await? }; + info!( + "Inspection summary: {} old manifests, {} referenced data files, \ + {} referenced deletion files, {} referenced tx files, {} referenced indices", + inspection.old_manifests.len(), + inspection.referenced_files.data_paths.len(), + inspection.referenced_files.delete_paths.len(), + inspection.referenced_files.tx_paths.len(), + inspection.referenced_files.index_uuids.len(), + ); + let stats = self.delete_unreferenced_files(inspection).await?; final_stats.bytes_removed += stats.bytes_removed; final_stats.old_versions += stats.old_versions; @@ -222,12 +235,16 @@ impl<'a> CleanupTask<'a> { ); let failed: Mutex> = Mutex::new(Vec::new()); + let processed_count = AtomicU64::new(0); + let log_interval: u64 = std::cmp::max(total as u64 / 20, 100); // First pass: process all manifests, collecting failures instead of aborting. stream::iter(locations) .for_each_concurrent(io_parallelism, |location| { let failed_ref = &failed; let inspection_ref = &inspection; + let processed_ref = &processed_count; + let total_manifests = total; async move { if let Err(err) = self .process_manifest_file(location.clone(), inspection_ref, tagged_versions) @@ -239,6 +256,15 @@ impl<'a> CleanupTask<'a> { .unwrap() .push((location, format!("{:?}", err))); } + let count = processed_ref.fetch_add(1, Ordering::Relaxed) + 1; + if count % log_interval == 0 || count == total_manifests as u64 { + info!( + "Manifest progress: {}/{} ({:.1}%)", + count, + total_manifests, + count as f64 / total_manifests as f64 * 100.0, + ); + } } }) .await; @@ -530,14 +556,29 @@ impl<'a> CleanupTask<'a> { all_paths_to_remove.boxed() }; + let delete_count = AtomicU64::new(0); + let paths_with_progress = paths_to_delete.map_ok(|path| { + let count = delete_count.fetch_add(1, Ordering::Relaxed) + 1; + if count % 1000 == 0 { + info!("Delete progress: {} files submitted", count); + } + path + }); + let delete_fut = self .dataset .object_store - .remove_stream(paths_to_delete) + .remove_stream(paths_with_progress.boxed()) .try_for_each(|_| future::ready(Ok(()))); delete_fut.await?; + let total_deleted = delete_count.load(Ordering::Relaxed); + info!( + "Delete phase complete: {} files deleted ({} old manifests)", + total_deleted, num_old_manifests, + ); + let mut removal_stats = removal_stats.into_inner().unwrap(); removal_stats.old_versions = num_old_manifests as u64; removal_stats.bytes_removed += manifest_bytes_removed?;