diff --git a/crates/iceberg/public-api.txt b/crates/iceberg/public-api.txt index 610675c83d..e54e552944 100644 --- a/crates/iceberg/public-api.txt +++ b/crates/iceberg/public-api.txt @@ -726,6 +726,7 @@ pub async fn iceberg::io::FileIO::delete(&self, path: impl core::convert::AsRef< pub async fn iceberg::io::FileIO::delete_prefix(&self, path: impl core::convert::AsRef) -> iceberg::Result<()> pub async fn iceberg::io::FileIO::delete_stream(&self, paths: impl futures_core::stream::Stream + core::marker::Send + 'static) -> iceberg::Result<()> pub async fn iceberg::io::FileIO::exists(&self, path: impl core::convert::AsRef) -> iceberg::Result +pub async fn iceberg::io::FileIO::list_prefix(&self, path: impl core::convert::AsRef) -> iceberg::Result> pub fn iceberg::io::FileIO::new_input(&self, path: impl core::convert::AsRef) -> iceberg::Result pub fn iceberg::io::FileIO::new_output(&self, path: impl core::convert::AsRef) -> iceberg::Result pub fn iceberg::io::FileIO::new_with_fs() -> Self @@ -808,6 +809,18 @@ pub async fn iceberg::io::InputFile::read(&self) -> iceberg::Result iceberg::Result> impl core::fmt::Debug for iceberg::io::InputFile pub fn iceberg::io::InputFile::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result +pub struct iceberg::io::ListEntry +pub iceberg::io::ListEntry::last_modified_ms: core::option::Option +pub iceberg::io::ListEntry::path: alloc::string::String +pub iceberg::io::ListEntry::size: u64 +impl core::clone::Clone for iceberg::io::ListEntry +pub fn iceberg::io::ListEntry::clone(&self) -> iceberg::io::ListEntry +impl core::cmp::Eq for iceberg::io::ListEntry +impl core::cmp::PartialEq for iceberg::io::ListEntry +pub fn iceberg::io::ListEntry::eq(&self, other: &iceberg::io::ListEntry) -> bool +impl core::fmt::Debug for iceberg::io::ListEntry +pub fn iceberg::io::ListEntry::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result +impl core::marker::StructuralPartialEq for iceberg::io::ListEntry pub struct iceberg::io::LocalFsStorage impl iceberg::io::LocalFsStorage pub fn iceberg::io::LocalFsStorage::new() -> Self @@ -822,6 +835,7 @@ pub fn iceberg::io::LocalFsStorage::delete<'life0, 'life1, 'async_trait>(&'life0 pub fn iceberg::io::LocalFsStorage::delete_prefix<'life0, 'life1, 'async_trait>(&'life0 self, path: &'life1 str) -> core::pin::Pin> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait pub fn iceberg::io::LocalFsStorage::delete_stream<'life0, 'async_trait>(&'life0 self, paths: futures_core::stream::BoxStream<'static, alloc::string::String>) -> core::pin::Pin> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait pub fn iceberg::io::LocalFsStorage::exists<'life0, 'life1, 'async_trait>(&'life0 self, path: &'life1 str) -> core::pin::Pin> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait +pub fn iceberg::io::LocalFsStorage::list_prefix<'life0, 'life1, 'async_trait>(&'life0 self, path: &'life1 str) -> core::pin::Pin>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait pub fn iceberg::io::LocalFsStorage::metadata<'life0, 'life1, 'async_trait>(&'life0 self, path: &'life1 str) -> core::pin::Pin> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait pub fn iceberg::io::LocalFsStorage::new_input(&self, path: &str) -> iceberg::Result pub fn iceberg::io::LocalFsStorage::new_output(&self, path: &str) -> iceberg::Result @@ -860,6 +874,7 @@ pub fn iceberg::io::MemoryStorage::delete<'life0, 'life1, 'async_trait>(&'life0 pub fn iceberg::io::MemoryStorage::delete_prefix<'life0, 'life1, 'async_trait>(&'life0 self, path: &'life1 str) -> core::pin::Pin> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait pub fn iceberg::io::MemoryStorage::delete_stream<'life0, 'async_trait>(&'life0 self, paths: futures_core::stream::BoxStream<'static, alloc::string::String>) -> core::pin::Pin> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait pub fn iceberg::io::MemoryStorage::exists<'life0, 'life1, 'async_trait>(&'life0 self, path: &'life1 str) -> core::pin::Pin> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait +pub fn iceberg::io::MemoryStorage::list_prefix<'life0, 'life1, 'async_trait>(&'life0 self, path: &'life1 str) -> core::pin::Pin>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait pub fn iceberg::io::MemoryStorage::metadata<'life0, 'life1, 'async_trait>(&'life0 self, path: &'life1 str) -> core::pin::Pin> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait pub fn iceberg::io::MemoryStorage::new_input(&self, path: &str) -> iceberg::Result pub fn iceberg::io::MemoryStorage::new_output(&self, path: &str) -> iceberg::Result @@ -1048,6 +1063,7 @@ pub fn iceberg::io::Storage::delete<'life0, 'life1, 'async_trait>(&'life0 self, pub fn iceberg::io::Storage::delete_prefix<'life0, 'life1, 'async_trait>(&'life0 self, path: &'life1 str) -> core::pin::Pin> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait pub fn iceberg::io::Storage::delete_stream<'life0, 'async_trait>(&'life0 self, paths: futures_core::stream::BoxStream<'static, alloc::string::String>) -> core::pin::Pin> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait pub fn iceberg::io::Storage::exists<'life0, 'life1, 'async_trait>(&'life0 self, path: &'life1 str) -> core::pin::Pin> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait +pub fn iceberg::io::Storage::list_prefix<'life0, 'life1, 'async_trait>(&'life0 self, path: &'life1 str) -> core::pin::Pin>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait pub fn iceberg::io::Storage::metadata<'life0, 'life1, 'async_trait>(&'life0 self, path: &'life1 str) -> core::pin::Pin> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait pub fn iceberg::io::Storage::new_input(&self, path: &str) -> iceberg::Result pub fn iceberg::io::Storage::new_output(&self, path: &str) -> iceberg::Result @@ -1060,6 +1076,7 @@ pub fn iceberg::io::LocalFsStorage::delete<'life0, 'life1, 'async_trait>(&'life0 pub fn iceberg::io::LocalFsStorage::delete_prefix<'life0, 'life1, 'async_trait>(&'life0 self, path: &'life1 str) -> core::pin::Pin> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait pub fn iceberg::io::LocalFsStorage::delete_stream<'life0, 'async_trait>(&'life0 self, paths: futures_core::stream::BoxStream<'static, alloc::string::String>) -> core::pin::Pin> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait pub fn iceberg::io::LocalFsStorage::exists<'life0, 'life1, 'async_trait>(&'life0 self, path: &'life1 str) -> core::pin::Pin> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait +pub fn iceberg::io::LocalFsStorage::list_prefix<'life0, 'life1, 'async_trait>(&'life0 self, path: &'life1 str) -> core::pin::Pin>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait pub fn iceberg::io::LocalFsStorage::metadata<'life0, 'life1, 'async_trait>(&'life0 self, path: &'life1 str) -> core::pin::Pin> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait pub fn iceberg::io::LocalFsStorage::new_input(&self, path: &str) -> iceberg::Result pub fn iceberg::io::LocalFsStorage::new_output(&self, path: &str) -> iceberg::Result @@ -1072,6 +1089,7 @@ pub fn iceberg::io::MemoryStorage::delete<'life0, 'life1, 'async_trait>(&'life0 pub fn iceberg::io::MemoryStorage::delete_prefix<'life0, 'life1, 'async_trait>(&'life0 self, path: &'life1 str) -> core::pin::Pin> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait pub fn iceberg::io::MemoryStorage::delete_stream<'life0, 'async_trait>(&'life0 self, paths: futures_core::stream::BoxStream<'static, alloc::string::String>) -> core::pin::Pin> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait pub fn iceberg::io::MemoryStorage::exists<'life0, 'life1, 'async_trait>(&'life0 self, path: &'life1 str) -> core::pin::Pin> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait +pub fn iceberg::io::MemoryStorage::list_prefix<'life0, 'life1, 'async_trait>(&'life0 self, path: &'life1 str) -> core::pin::Pin>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait pub fn iceberg::io::MemoryStorage::metadata<'life0, 'life1, 'async_trait>(&'life0 self, path: &'life1 str) -> core::pin::Pin> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait pub fn iceberg::io::MemoryStorage::new_input(&self, path: &str) -> iceberg::Result pub fn iceberg::io::MemoryStorage::new_output(&self, path: &str) -> iceberg::Result @@ -1085,6 +1103,46 @@ impl iceberg::io::StorageFactory for iceberg::io::LocalFsStorageFactory pub fn iceberg::io::LocalFsStorageFactory::build(&self, _config: &iceberg::io::StorageConfig) -> iceberg::Result> impl iceberg::io::StorageFactory for iceberg::io::MemoryStorageFactory pub fn iceberg::io::MemoryStorageFactory::build(&self, _config: &iceberg::io::StorageConfig) -> iceberg::Result> +pub mod iceberg::maintenance +pub enum iceberg::maintenance::PrefixMismatchMode +pub iceberg::maintenance::PrefixMismatchMode::Delete +pub iceberg::maintenance::PrefixMismatchMode::Error +pub iceberg::maintenance::PrefixMismatchMode::Ignore +impl core::clone::Clone for iceberg::maintenance::PrefixMismatchMode +pub fn iceberg::maintenance::PrefixMismatchMode::clone(&self) -> iceberg::maintenance::PrefixMismatchMode +impl core::cmp::Eq for iceberg::maintenance::PrefixMismatchMode +impl core::cmp::PartialEq for iceberg::maintenance::PrefixMismatchMode +pub fn iceberg::maintenance::PrefixMismatchMode::eq(&self, other: &iceberg::maintenance::PrefixMismatchMode) -> bool +impl core::default::Default for iceberg::maintenance::PrefixMismatchMode +pub fn iceberg::maintenance::PrefixMismatchMode::default() -> iceberg::maintenance::PrefixMismatchMode +impl core::fmt::Debug for iceberg::maintenance::PrefixMismatchMode +pub fn iceberg::maintenance::PrefixMismatchMode::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result +impl core::marker::Copy for iceberg::maintenance::PrefixMismatchMode +impl core::marker::StructuralPartialEq for iceberg::maintenance::PrefixMismatchMode +pub struct iceberg::maintenance::RemoveOrphanFilesAction +impl iceberg::maintenance::RemoveOrphanFilesAction +pub fn iceberg::maintenance::RemoveOrphanFilesAction::delete_concurrency(self, concurrency: usize) -> Self +pub fn iceberg::maintenance::RemoveOrphanFilesAction::dry_run(self, dry_run: bool) -> Self +pub fn iceberg::maintenance::RemoveOrphanFilesAction::equal_authorities(self, authorities: std::collections::hash::map::HashMap) -> Self +pub fn iceberg::maintenance::RemoveOrphanFilesAction::equal_schemes(self, schemes: std::collections::hash::map::HashMap) -> Self +pub async fn iceberg::maintenance::RemoveOrphanFilesAction::execute(self) -> iceberg::Result +pub fn iceberg::maintenance::RemoveOrphanFilesAction::location(self, location: impl core::convert::Into) -> Self +pub fn iceberg::maintenance::RemoveOrphanFilesAction::new(table: iceberg::table::Table) -> Self +pub fn iceberg::maintenance::RemoveOrphanFilesAction::older_than_ms(self, timestamp_ms: i64) -> Self +pub fn iceberg::maintenance::RemoveOrphanFilesAction::prefix_mismatch_mode(self, mode: iceberg::maintenance::PrefixMismatchMode) -> Self +pub struct iceberg::maintenance::RemoveOrphanFilesResult +pub iceberg::maintenance::RemoveOrphanFilesResult::deleted_files: alloc::vec::Vec +pub iceberg::maintenance::RemoveOrphanFilesResult::failed_deletes: alloc::vec::Vec<(alloc::string::String, alloc::string::String)> +pub iceberg::maintenance::RemoveOrphanFilesResult::listed_count: usize +pub iceberg::maintenance::RemoveOrphanFilesResult::orphan_files: alloc::vec::Vec +pub iceberg::maintenance::RemoveOrphanFilesResult::referenced_count: usize +pub iceberg::maintenance::RemoveOrphanFilesResult::skipped_missing_mtime: usize +pub iceberg::maintenance::RemoveOrphanFilesResult::skipped_recent: usize +impl core::default::Default for iceberg::maintenance::RemoveOrphanFilesResult +pub fn iceberg::maintenance::RemoveOrphanFilesResult::default() -> iceberg::maintenance::RemoveOrphanFilesResult +impl core::fmt::Debug for iceberg::maintenance::RemoveOrphanFilesResult +pub fn iceberg::maintenance::RemoveOrphanFilesResult::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result +pub const iceberg::maintenance::DEFAULT_OLDER_THAN_MS: i64 pub mod iceberg::memory pub struct iceberg::memory::MemoryCatalog impl core::fmt::Debug for iceberg::memory::MemoryCatalog diff --git a/crates/iceberg/src/io/file_io.rs b/crates/iceberg/src/io/file_io.rs index cd0a4434c4..5a00ac0d0a 100644 --- a/crates/iceberg/src/io/file_io.rs +++ b/crates/iceberg/src/io/file_io.rs @@ -162,6 +162,38 @@ impl FileIO { self.get_storage()?.exists(path.as_ref()).await } + /// Recursively list all FILES under a prefix (directories are not + /// returned). Entry paths are absolute, in the same form as the given + /// prefix, so they can be passed straight back to [`FileIO`] operations + /// and compared against paths recorded in table metadata. + /// + /// # Prefix semantics + /// + /// The prefix is treated as a *directory-style location*, not a raw + /// object-key prefix: implementations normalize a prefix without a + /// trailing `/` by appending one before listing. Consequently: + /// + /// * `list_prefix("s3://bucket/t")` and `list_prefix("s3://bucket/t/")` + /// are equivalent; + /// * a path naming an existing FILE yields an empty list (it is not + /// matched as a raw prefix) — use [`FileIO::exists`] to check for a + /// specific file; + /// * a prefix with nothing under it yields an empty list, not an error. + /// + /// This directory-style contract matches the maintenance use case + /// (listing a table location, or its `data/`/`metadata/` directories, + /// for orphan-file cleanup). Java's `SupportsPrefixOperations.listPrefix` + /// leaves the exact-file/raw-prefix case implementation-defined; here it + /// is fixed to directory-style so callers do not have to defend against + /// both behaviors. + /// + /// # Arguments + /// + /// * path: It should be *absolute* path starting with scheme string used to construct [`FileIO`]. + pub async fn list_prefix(&self, path: impl AsRef) -> Result> { + self.get_storage()?.list_prefix(path.as_ref()).await + } + /// Creates input file. /// /// # Arguments @@ -242,6 +274,21 @@ pub struct FileMetadata { pub size: u64, } +/// A single file returned by [`FileIO::list_prefix`]. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ListEntry { + /// Absolute path, in the same form as the listed prefix — directly usable + /// with [`FileIO`] operations and comparable to paths recorded in table + /// metadata. + pub path: String, + /// The size of the file in bytes. + pub size: u64, + /// Last-modified time in milliseconds since the Unix epoch, when the + /// backing storage tracks it. `None` means unknown — age-based logic + /// (e.g. orphan-file cleanup) must treat such files as NOT eligible. + pub last_modified_ms: Option, +} + /// Trait for reading file. /// /// # TODO diff --git a/crates/iceberg/src/io/storage/local_fs.rs b/crates/iceberg/src/io/storage/local_fs.rs index e96e951baa..c55521e949 100644 --- a/crates/iceberg/src/io/storage/local_fs.rs +++ b/crates/iceberg/src/io/storage/local_fs.rs @@ -34,11 +34,25 @@ use futures::stream::BoxStream; use serde::{Deserialize, Serialize}; use crate::io::{ - FileMetadata, FileRead, FileWrite, InputFile, OutputFile, Storage, StorageConfig, + FileMetadata, FileRead, FileWrite, InputFile, ListEntry, OutputFile, Storage, StorageConfig, StorageFactory, }; use crate::{Error, ErrorKind, Result}; +/// Recursively collect all files under `dir` into `out` as (path, metadata). +fn walk_dir(dir: &std::path::Path, out: &mut Vec<(PathBuf, fs::Metadata)>) -> std::io::Result<()> { + for entry in fs::read_dir(dir)? { + let entry = entry?; + let meta = entry.metadata()?; + if meta.is_dir() { + walk_dir(&entry.path(), out)?; + } else { + out.push((entry.path(), meta)); + } + } + Ok(()) +} + /// Local filesystem storage implementation. /// /// This storage implementation uses standard Rust filesystem operations, @@ -202,6 +216,48 @@ impl Storage for LocalFsStorage { Ok(()) } + async fn list_prefix(&self, path: &str) -> Result> { + let root = Self::normalize_path(path); + if !root.is_dir() { + return Ok(vec![]); + } + let mut files = Vec::new(); + walk_dir(&root, &mut files).map_err(|e| { + Error::new( + ErrorKind::Unexpected, + format!("Failed to list directory {}: {}", root.display(), e), + ) + })?; + let abs_prefix = if path.ends_with('/') { + path.to_string() + } else { + format!("{path}/") + }; + let root_str = root.to_string_lossy().to_string(); + let root_prefix = if root_str.ends_with('/') { + root_str + } else { + format!("{root_str}/") + }; + Ok(files + .into_iter() + .map(|(p, meta)| { + let full = p.to_string_lossy(); + let suffix = full.strip_prefix(&root_prefix).unwrap_or(&full).to_string(); + ListEntry { + // Reconstruct in the caller's prefix form so paths round-trip. + path: format!("{abs_prefix}{suffix}"), + size: meta.len(), + last_modified_ms: meta + .modified() + .ok() + .and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok()) + .map(|d| d.as_millis() as i64), + } + }) + .collect()) + } + async fn delete_stream(&self, mut paths: BoxStream<'static, String>) -> Result<()> { while let Some(path) = paths.next().await { self.delete(&path).await?; @@ -600,4 +656,67 @@ mod tests { let path_stream = stream::iter(Vec::::new()).boxed(); storage.delete_stream(path_stream).await.unwrap(); } + + #[tokio::test] + async fn test_list_prefix() { + let tmp = tempfile::tempdir().unwrap(); + let root = tmp.path().to_str().unwrap().to_string(); + let storage = LocalFsStorage::new(); + storage + .write( + &format!("{root}/t/data/a.parquet"), + Bytes::from_static(b"aa"), + ) + .await + .unwrap(); + storage + .write( + &format!("{root}/t/metadata/v1.json"), + Bytes::from_static(b"m"), + ) + .await + .unwrap(); + storage + .write(&format!("{root}/other.txt"), Bytes::from_static(b"x")) + .await + .unwrap(); + + let mut entries = storage.list_prefix(&format!("{root}/t")).await.unwrap(); + entries.sort_by(|a, b| a.path.cmp(&b.path)); + assert_eq!(entries.len(), 2); + assert_eq!(entries[0].path, format!("{root}/t/data/a.parquet")); + assert_eq!(entries[0].size, 2); + assert!( + entries[0].last_modified_ms.is_some(), + "local fs must report modification times" + ); + // Listing a non-existent prefix is empty, not an error. + assert!( + storage + .list_prefix(&format!("{root}/missing")) + .await + .unwrap() + .is_empty() + ); + } + + #[tokio::test] + async fn test_list_prefix_is_directory_style() { + let tmp = tempfile::tempdir().unwrap(); + let root = tmp.path().to_str().unwrap().to_string(); + let storage = LocalFsStorage::new(); + let file = format!("{root}/t/data/a.parquet"); + storage + .write(&file, Bytes::from_static(b"aa")) + .await + .unwrap(); + + // A trailing-slash prefix is equivalent to the non-slash form. + let entries = storage.list_prefix(&format!("{root}/t/")).await.unwrap(); + assert_eq!(entries.len(), 1); + assert_eq!(entries[0].path, format!("{root}/t/data/a.parquet")); + + // A path naming an existing FILE is not matched as a raw prefix. + assert!(storage.list_prefix(&file).await.unwrap().is_empty()); + } } diff --git a/crates/iceberg/src/io/storage/memory.rs b/crates/iceberg/src/io/storage/memory.rs index f33dbd07b1..9a79467778 100644 --- a/crates/iceberg/src/io/storage/memory.rs +++ b/crates/iceberg/src/io/storage/memory.rs @@ -33,7 +33,7 @@ use futures::stream::BoxStream; use serde::{Deserialize, Serialize}; use crate::io::{ - FileMetadata, FileRead, FileWrite, InputFile, OutputFile, Storage, StorageConfig, + FileMetadata, FileRead, FileWrite, InputFile, ListEntry, OutputFile, Storage, StorageConfig, StorageFactory, }; use crate::{Error, ErrorKind, Result}; @@ -63,17 +63,33 @@ use crate::{Error, ErrorKind, Result}; pub struct MemoryStorage { #[serde(skip, default = "default_memory_data")] data: Arc>>, + /// Last-modified stamps (ms since epoch) per normalized path, so + /// age-based logic (orphan-file cleanup) is testable on memory storage. + #[serde(skip, default = "default_memory_mtimes")] + mtimes: Arc>>, } fn default_memory_data() -> Arc>> { Arc::new(RwLock::new(HashMap::new())) } +fn default_memory_mtimes() -> Arc>> { + Arc::new(RwLock::new(HashMap::new())) +} + +pub(crate) fn now_ms() -> i64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_millis() as i64) + .unwrap_or(0) +} + impl MemoryStorage { /// Create a new empty `MemoryStorage` instance. pub fn new() -> Self { Self { data: Arc::new(RwLock::new(HashMap::new())), + mtimes: Arc::new(RwLock::new(HashMap::new())), } } @@ -169,7 +185,11 @@ impl Storage for MemoryStorage { format!("Failed to acquire write lock: {e}"), ) })?; - data.insert(normalized, bs); + data.insert(normalized.clone(), bs); + drop(data); + if let Ok(mut mtimes) = self.mtimes.write() { + mtimes.insert(normalized, now_ms()); + } Ok(()) } @@ -177,6 +197,7 @@ impl Storage for MemoryStorage { let normalized = Self::normalize_path(path); Ok(Box::new(MemoryFileWrite::new( self.data.clone(), + self.mtimes.clone(), normalized, ))) } @@ -229,6 +250,42 @@ impl Storage for MemoryStorage { Ok(()) } + async fn list_prefix(&self, path: &str) -> Result> { + let normalized = Self::normalize_path(path); + let norm_prefix = if normalized.ends_with('/') || normalized.is_empty() { + normalized + } else { + format!("{normalized}/") + }; + let abs_prefix = if path.ends_with('/') { + path.to_string() + } else { + format!("{path}/") + }; + let data = self.data.read().map_err(|e| { + Error::new( + ErrorKind::Unexpected, + format!("Failed to acquire read lock: {e}"), + ) + })?; + let mtimes = self.mtimes.read().map_err(|e| { + Error::new( + ErrorKind::Unexpected, + format!("Failed to acquire read lock: {e}"), + ) + })?; + Ok(data + .iter() + .filter(|(k, _)| k.starts_with(&norm_prefix)) + .map(|(k, bytes)| ListEntry { + // Reconstruct in the caller's prefix form so paths round-trip. + path: format!("{abs_prefix}{}", &k[norm_prefix.len()..]), + size: bytes.len() as u64, + last_modified_ms: mtimes.get(k).copied(), + }) + .collect()) + } + fn new_input(&self, path: &str) -> Result { Ok(InputFile::new(Arc::new(self.clone()), path.to_string())) } @@ -296,6 +353,7 @@ impl FileRead for MemoryFileRead { #[derive(Debug)] pub struct MemoryFileWrite { data: Arc>>, + mtimes: Arc>>, path: String, buffer: Vec, closed: bool, @@ -303,9 +361,14 @@ pub struct MemoryFileWrite { impl MemoryFileWrite { /// Create a new `MemoryFileWrite` for the given path. - pub fn new(data: Arc>>, path: String) -> Self { + pub fn new( + data: Arc>>, + mtimes: Arc>>, + path: String, + ) -> Self { Self { data, + mtimes, path, buffer: Vec::new(), closed: false, @@ -342,6 +405,10 @@ impl FileWrite for MemoryFileWrite { self.path.clone(), Bytes::from(std::mem::take(&mut self.buffer)), ); + drop(data); + if let Ok(mut mtimes) = self.mtimes.write() { + mtimes.insert(self.path.clone(), now_ms()); + } self.closed = true; Ok(()) } @@ -655,4 +722,67 @@ mod tests { let path_stream = stream::iter(Vec::::new()).boxed(); storage.delete_stream(path_stream).await.unwrap(); } + + #[tokio::test] + async fn test_list_prefix() { + let storage = MemoryStorage::new(); + storage + .write( + "memory://warehouse/t/data/a.parquet", + Bytes::from_static(b"aa"), + ) + .await + .unwrap(); + storage + .write( + "memory://warehouse/t/metadata/v1.json", + Bytes::from_static(b"m"), + ) + .await + .unwrap(); + storage + .write( + "memory://warehouse/other/b.parquet", + Bytes::from_static(b"b"), + ) + .await + .unwrap(); + + let mut entries = storage.list_prefix("memory://warehouse/t").await.unwrap(); + entries.sort_by(|a, b| a.path.cmp(&b.path)); + assert_eq!(entries.len(), 2); + // Paths come back in the caller's prefix form and round-trip to reads. + assert_eq!(entries[0].path, "memory://warehouse/t/data/a.parquet"); + assert_eq!(entries[0].size, 2); + assert!(entries[0].last_modified_ms.is_some()); + assert_eq!(entries[1].path, "memory://warehouse/t/metadata/v1.json"); + assert!(storage.exists(&entries[0].path).await.unwrap()); + } + + #[tokio::test] + async fn test_list_prefix_is_directory_style() { + let storage = MemoryStorage::new(); + storage + .write( + "memory://warehouse/t/data/a.parquet", + Bytes::from_static(b"aa"), + ) + .await + .unwrap(); + + // A trailing-slash prefix is equivalent to the non-slash form. + let entries = storage.list_prefix("memory://warehouse/t/").await.unwrap(); + assert_eq!(entries.len(), 1); + assert_eq!(entries[0].path, "memory://warehouse/t/data/a.parquet"); + + // A path naming an existing FILE is not matched as a raw prefix: + // it is normalized to `.../a.parquet/`, under which nothing exists. + assert!( + storage + .list_prefix("memory://warehouse/t/data/a.parquet") + .await + .unwrap() + .is_empty() + ); + } } diff --git a/crates/iceberg/src/io/storage/mod.rs b/crates/iceberg/src/io/storage/mod.rs index 5276c7771f..ae31c9ca58 100644 --- a/crates/iceberg/src/io/storage/mod.rs +++ b/crates/iceberg/src/io/storage/mod.rs @@ -31,7 +31,7 @@ use futures::stream::BoxStream; pub use local_fs::{LocalFsStorage, LocalFsStorageFactory}; pub use memory::{MemoryStorage, MemoryStorageFactory}; -use super::{FileMetadata, FileRead, FileWrite, InputFile, OutputFile}; +use super::{FileMetadata, FileRead, FileWrite, InputFile, ListEntry, OutputFile}; use crate::Result; /// Trait for storage operations in Iceberg. @@ -94,6 +94,20 @@ pub trait Storage: Debug + Send + Sync { /// Delete all files with the given prefix async fn delete_prefix(&self, path: &str) -> Result<()>; + /// Recursively list all FILES under a prefix (no directories). Entry paths + /// are absolute, in the same form as the given prefix. The prefix is + /// directory-style: implementations MUST normalize a prefix without a + /// trailing `/` by appending one, so a path naming an existing file + /// yields an empty list — see [`crate::io::FileIO::list_prefix`] for the + /// full contract. Default: unsupported. + async fn list_prefix(&self, path: &str) -> Result> { + let _ = path; + Err(crate::Error::new( + crate::ErrorKind::FeatureUnsupported, + "list_prefix is not supported by this storage", + )) + } + /// Delete multiple files from a stream of paths. async fn delete_stream(&self, paths: BoxStream<'static, String>) -> Result<()>; diff --git a/crates/iceberg/src/lib.rs b/crates/iceberg/src/lib.rs index 4e346460f5..a280bba7ed 100644 --- a/crates/iceberg/src/lib.rs +++ b/crates/iceberg/src/lib.rs @@ -86,6 +86,7 @@ pub mod inspect; pub mod scan; pub mod expr; +pub mod maintenance; pub mod transaction; pub mod transform; diff --git a/crates/iceberg/src/maintenance.rs b/crates/iceberg/src/maintenance.rs new file mode 100644 index 0000000000..79e846be2a --- /dev/null +++ b/crates/iceberg/src/maintenance.rs @@ -0,0 +1,443 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you 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. + +//! Table maintenance operations that are NOT snapshot commits. +//! +//! Currently: [`RemoveOrphanFilesAction`] — delete files under the table +//! location that no retained snapshot or table metadata references. Modeled on +//! iceberg-go's `Table.DeleteOrphanFiles` (itself modeled on Java's +//! `DeleteOrphanFiles` action): dry-run support, an `older_than` age guard +//! (protects in-flight commits), and prefix-mismatch handling with optional +//! scheme/authority equivalence. + +use std::collections::{HashMap, HashSet}; + +use futures::{StreamExt, stream}; + +use crate::io::ListEntry; +use crate::table::Table; +use crate::{Error, ErrorKind, Result}; + +/// Default age guard: only files older than 3 days are deletion candidates +/// (iceberg-go/Java default). Protects files written by in-flight commits. +pub const DEFAULT_OLDER_THAN_MS: i64 = 3 * 24 * 60 * 60 * 1000; + +const DEFAULT_DELETE_CONCURRENCY: usize = 8; + +/// How to handle a listed file that matches a referenced file except for its +/// URI scheme/authority (after applying the equivalence maps). Mirrors +/// iceberg-go's `PrefixMismatchMode` / Java's `prefix_mismatch_mode`. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +pub enum PrefixMismatchMode { + /// Fail the action (default — safest; forces the operator to declare + /// scheme/authority equivalence explicitly). + #[default] + Error, + /// Treat mismatched files as referenced (never delete them). + Ignore, + /// Treat mismatched files as orphans (delete them). + Delete, +} + +/// Result of a [`RemoveOrphanFilesAction`] execution. +#[derive(Debug, Default)] +pub struct RemoveOrphanFilesResult { + /// Orphan files identified (deletion candidates). In dry-run mode nothing + /// is deleted — this IS the report. + pub orphan_files: Vec, + /// Files actually deleted (empty in dry-run mode). + pub deleted_files: Vec, + /// Per-file delete failures as `(path, error)` — deletion is best-effort + /// (iceberg-go parity: continue past individual failures and report them). + pub failed_deletes: Vec<(String, String)>, + /// Total files listed under the location. + pub listed_count: usize, + /// Size of the referenced-file set. + pub referenced_count: usize, + /// Candidates skipped because they are newer than `older_than_ms`. + pub skipped_recent: usize, + /// Candidates skipped because the storage reported no modification time + /// (treated as possibly-in-flight; never deleted). + pub skipped_missing_mtime: usize, +} + +/// Deletes files under the table location that are not referenced by any +/// retained snapshot or table metadata. +/// +/// ```ignore +/// let result = RemoveOrphanFilesAction::new(table) +/// .older_than_ms(cutoff_ms) +/// .dry_run(true) +/// .execute() +/// .await?; +/// ``` +/// +/// Safety properties: +/// - **Dry-run first**: `dry_run(true)` produces the full orphan report with +/// zero writes; callers should verify it (e.g. against an independent +/// engine's `all_files` metadata) before ever running the delete mode. +/// - **Age guard**: files newer than `older_than_ms` are never touched — +/// an in-flight commit's freshly-written files are unreferenced until the +/// commit lands. Files without a modification time are never touched. +/// - **Location guard**: only files under the listed location are considered. +/// - **Conservative referenced set**: ALL manifest entries of ALL retained +/// snapshots count as referenced, regardless of entry status. This +/// deliberately diverges from iceberg-go (which drops DELETED-status +/// entries): a superseded file costs storage until `expire_snapshots` +/// drops its last referencing manifest, but can never be deleted while any +/// retained manifest still mentions it — and it keeps the orphan report +/// verifiable against Java's `all_files` metadata table, which also +/// includes logically-deleted files. +pub struct RemoveOrphanFilesAction { + table: Table, + location: Option, + older_than_ms: Option, + dry_run: bool, + delete_concurrency: usize, + prefix_mismatch_mode: PrefixMismatchMode, + equal_schemes: HashMap, + equal_authorities: HashMap, +} + +impl RemoveOrphanFilesAction { + /// Creates the action for `table` with iceberg-go's defaults: table + /// location, 3-day age guard, no dry-run, `PrefixMismatchMode::Error`, + /// no scheme/authority equivalence. + pub fn new(table: Table) -> Self { + Self { + table, + location: None, + older_than_ms: None, + dry_run: false, + delete_concurrency: DEFAULT_DELETE_CONCURRENCY, + prefix_mismatch_mode: PrefixMismatchMode::default(), + equal_schemes: HashMap::new(), + equal_authorities: HashMap::new(), + } + } + + /// Location to scan (defaults to the table location). + pub fn location(mut self, location: impl Into) -> Self { + self.location = Some(location.into()); + self + } + + /// Absolute cutoff: only files last modified BEFORE this timestamp + /// (milliseconds since epoch) are deletion candidates. Defaults to + /// now − 3 days. + pub fn older_than_ms(mut self, timestamp_ms: i64) -> Self { + self.older_than_ms = Some(timestamp_ms); + self + } + + /// Identify orphans without deleting anything. + pub fn dry_run(mut self, dry_run: bool) -> Self { + self.dry_run = dry_run; + self + } + + /// Concurrency for the delete phase. + pub fn delete_concurrency(mut self, concurrency: usize) -> Self { + self.delete_concurrency = concurrency.max(1); + self + } + + /// How to handle candidates that match a referenced file except for + /// scheme/authority. See [`PrefixMismatchMode`]. + pub fn prefix_mismatch_mode(mut self, mode: PrefixMismatchMode) -> Self { + self.prefix_mismatch_mode = mode; + self + } + + /// Declare URI schemes as equivalent, e.g. `{"s3a" => "s3", "s3n" => "s3"}`. + pub fn equal_schemes(mut self, schemes: HashMap) -> Self { + self.equal_schemes = schemes; + self + } + + /// Declare URI authorities as equivalent. + pub fn equal_authorities(mut self, authorities: HashMap) -> Self { + self.equal_authorities = authorities; + self + } + + /// Executes the scan (and, unless dry-run, the delete phase). + pub async fn execute(self) -> Result { + let location = self + .location + .clone() + .unwrap_or_else(|| self.table.metadata().location().to_string()); + let cutoff_ms = self.older_than_ms.unwrap_or_else(|| { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_millis() as i64) + .unwrap_or(0) + - DEFAULT_OLDER_THAN_MS + }); + + let referenced = self.collect_referenced_files().await?; + // normalized form -> one original referenced path (for mismatch reporting). + let mut normalized_referenced: HashMap = HashMap::new(); + for path in &referenced { + normalized_referenced + .entry(self.normalize(path)) + .or_insert_with(|| path.clone()); + } + + let listed = self.table.file_io().list_prefix(&location).await?; + + let mut result = RemoveOrphanFilesResult { + listed_count: listed.len(), + referenced_count: referenced.len(), + ..Default::default() + }; + + for entry in listed { + match self.classify(&entry, &referenced, &normalized_referenced, cutoff_ms)? { + Classification::Referenced => {} + Classification::Recent => result.skipped_recent += 1, + Classification::MissingMtime => result.skipped_missing_mtime += 1, + Classification::Orphan => result.orphan_files.push(entry.path), + } + } + + if self.dry_run || result.orphan_files.is_empty() { + return Ok(result); + } + + // Delete phase: bounded concurrency, best-effort per file. + let file_io = self.table.file_io().clone(); + let outcomes: Vec<(String, std::result::Result<(), String>)> = + stream::iter(result.orphan_files.clone()) + .map(|path| { + let file_io = file_io.clone(); + async move { + let outcome = file_io.delete(&path).await.map_err(|e| e.to_string()); + (path, outcome) + } + }) + .buffer_unordered(self.delete_concurrency) + .collect() + .await; + for (path, outcome) in outcomes { + match outcome { + Ok(()) => result.deleted_files.push(path), + Err(e) => result.failed_deletes.push((path, e)), + } + } + + Ok(result) + } + + fn classify( + &self, + entry: &ListEntry, + referenced: &HashSet, + normalized_referenced: &HashMap, + cutoff_ms: i64, + ) -> Result { + if referenced.contains(&entry.path) { + return Ok(Classification::Referenced); + } + let normalized = self.normalize(&entry.path); + if let Some(original) = normalized_referenced.get(&normalized) { + // Same file modulo scheme/authority — iceberg-go's prefix-mismatch + // handling. (An exact match was already caught above; reaching here + // with `original == entry.path` is impossible.) + return match self.prefix_mismatch_mode { + PrefixMismatchMode::Error => Err(Error::new( + ErrorKind::DataInvalid, + format!( + "Found file with mismatched scheme/authority: listed `{}` vs referenced `{}`. \ + Declare equivalence via equal_schemes/equal_authorities, or set \ + PrefixMismatchMode::Ignore / ::Delete.", + entry.path, original + ), + )), + PrefixMismatchMode::Ignore => Ok(Classification::Referenced), + PrefixMismatchMode::Delete => self.age_gate(entry, cutoff_ms), + }; + } + self.age_gate(entry, cutoff_ms) + } + + fn age_gate(&self, entry: &ListEntry, cutoff_ms: i64) -> Result { + Ok(match entry.last_modified_ms { + None => Classification::MissingMtime, + Some(ts) if ts >= cutoff_ms => Classification::Recent, + Some(_) => Classification::Orphan, + }) + } + + fn normalize(&self, path: &str) -> String { + normalize_uri(path, &self.equal_schemes, &self.equal_authorities) + } + + /// The referenced-file set: everything reachable from table metadata. + async fn collect_referenced_files(&self) -> Result> { + let metadata = self.table.metadata_ref(); + let file_io = self.table.file_io(); + let mut referenced: HashSet = HashSet::new(); + + // Metadata files: current + historical + version hint (harmless if absent). + if let Some(loc) = self.table.metadata_location() { + referenced.insert(loc.to_string()); + } + referenced.extend( + metadata + .metadata_log() + .iter() + .map(|e| e.metadata_file.clone()), + ); + referenced.insert(format!( + "{}/metadata/version-hint.text", + metadata.location().trim_end_matches('/') + )); + + // Statistics + partition statistics (Puffin). + referenced.extend( + metadata + .statistics_iter() + .map(|s| s.statistics_path.clone()), + ); + referenced.extend( + metadata + .partition_statistics_iter() + .map(|s| s.statistics_path.clone()), + ); + + // Every retained snapshot: manifest list, manifests, and ALL entries' + // file paths (any status — see the struct doc for why deleted-status + // entries stay in the referenced set). + let mut manifests_to_load = Vec::new(); + for snapshot in metadata.snapshots() { + if !snapshot.manifest_list().is_empty() { + referenced.insert(snapshot.manifest_list().to_string()); + } + let manifest_list = self.table.manifest_list_reader(snapshot).load().await?; + for manifest_file in manifest_list.entries() { + // insert() returning true = first sighting -> load once. + if referenced.insert(manifest_file.manifest_path.clone()) { + manifests_to_load.push(manifest_file.clone()); + } + } + } + for manifest_file in manifests_to_load { + let manifest = manifest_file.load_manifest(file_io).await?; + for entry in manifest.entries() { + referenced.insert(entry.data_file().file_path().to_string()); + } + } + + Ok(referenced) + } +} + +enum Classification { + Referenced, + Recent, + MissingMtime, + Orphan, +} + +/// Normalize a URI for equality-modulo-scheme/authority comparison: apply the +/// equivalence maps to `scheme` and `authority`, leave the path untouched. +/// Scheme-less paths are returned as-is. +fn normalize_uri( + path: &str, + equal_schemes: &HashMap, + equal_authorities: &HashMap, +) -> String { + let Some((scheme, rest)) = path.split_once("://") else { + return path.to_string(); + }; + let scheme = equal_schemes + .get(scheme) + .map(String::as_str) + .unwrap_or(scheme); + let (authority, tail) = match rest.split_once('/') { + Some((a, t)) => (a, Some(t)), + None => (rest, None), + }; + let authority = equal_authorities + .get(authority) + .map(String::as_str) + .unwrap_or(authority); + match tail { + Some(t) => format!("{scheme}://{authority}/{t}"), + None => format!("{scheme}://{authority}"), + } +} + +#[cfg(test)] +mod tests { + use std::collections::HashMap; + + use super::*; + + fn maps( + schemes: &[(&str, &str)], + authorities: &[(&str, &str)], + ) -> (HashMap, HashMap) { + ( + schemes + .iter() + .map(|(k, v)| (k.to_string(), v.to_string())) + .collect(), + authorities + .iter() + .map(|(k, v)| (k.to_string(), v.to_string())) + .collect(), + ) + } + + #[test] + fn test_normalize_identity_without_maps() { + let (s, a) = maps(&[], &[]); + assert_eq!( + normalize_uri("s3://bucket/a/b.parquet", &s, &a), + "s3://bucket/a/b.parquet" + ); + assert_eq!(normalize_uri("/plain/path", &s, &a), "/plain/path"); + } + + #[test] + fn test_normalize_equal_schemes() { + let (s, a) = maps(&[("s3a", "s3"), ("s3n", "s3")], &[]); + assert_eq!( + normalize_uri("s3a://bucket/a/b.parquet", &s, &a), + "s3://bucket/a/b.parquet" + ); + assert_eq!( + normalize_uri("s3n://bucket/a/b.parquet", &s, &a), + "s3://bucket/a/b.parquet" + ); + assert_eq!( + normalize_uri("s3://bucket/a/b.parquet", &s, &a), + "s3://bucket/a/b.parquet" + ); + } + + #[test] + fn test_normalize_equal_authorities() { + let (s, a) = maps(&[], &[("bucket-alias", "bucket")]); + assert_eq!( + normalize_uri("s3://bucket-alias/a/b.parquet", &s, &a), + "s3://bucket/a/b.parquet" + ); + } +} diff --git a/crates/iceberg/tests/remove_orphan_files.rs b/crates/iceberg/tests/remove_orphan_files.rs new file mode 100644 index 0000000000..98e526c391 --- /dev/null +++ b/crates/iceberg/tests/remove_orphan_files.rs @@ -0,0 +1,250 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you 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. + +//! End-to-end tests for `RemoveOrphanFilesAction`: a real table (memory +//! catalog + parquet writer chain) with extra unreferenced files dropped in. +//! +//! The dry-run report is the safety-critical surface: it must contain EXACTLY +//! the garbage files and never a referenced one (data files, manifests, +//! manifest lists, metadata). Deleting a referenced file is unrecoverable. + +use std::collections::HashMap; +use std::sync::Arc; + +use arrow_array::{Int32Array, RecordBatch}; +use arrow_schema::{DataType, Field, Schema as ArrowSchema}; +use futures::TryStreamExt; +use iceberg::maintenance::RemoveOrphanFilesAction; +use iceberg::memory::{MEMORY_CATALOG_WAREHOUSE, MemoryCatalogBuilder}; +use iceberg::spec::{ + DataFile, DataFileFormat, FormatVersion, NestedField, PrimitiveType, Schema, Type, +}; +use iceberg::table::Table; +use iceberg::transaction::{ApplyTransactionAction, Transaction}; +use iceberg::writer::base_writer::data_file_writer::DataFileWriterBuilder; +use iceberg::writer::file_writer::ParquetWriterBuilder; +use iceberg::writer::file_writer::location_generator::{ + DefaultFileNameGenerator, DefaultLocationGenerator, +}; +use iceberg::writer::file_writer::rolling_writer::RollingFileWriterBuilder; +use iceberg::writer::{IcebergWriter, IcebergWriterBuilder}; +use iceberg::{Catalog, CatalogBuilder, NamespaceIdent, TableCreation}; +use parquet::arrow::PARQUET_FIELD_ID_META_KEY; +use parquet::file::properties::WriterProperties; +use tempfile::TempDir; + +fn now_ms() -> i64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_millis() as i64 +} + +async fn write_one_data_file(table: &Table, prefix: &str, batch: RecordBatch) -> Vec { + let schema = table.metadata().current_schema().clone(); + let rolling = RollingFileWriterBuilder::new_with_default_file_size( + ParquetWriterBuilder::new(WriterProperties::builder().build(), schema), + table.file_io().clone(), + DefaultLocationGenerator::new(table.metadata()).unwrap(), + // Unique per-write prefix: the generator's counter resets per instance, + // so a shared prefix collides consecutive writes onto one path. + DefaultFileNameGenerator::new(prefix.to_string(), None, DataFileFormat::Parquet), + ); + let mut writer = DataFileWriterBuilder::new(rolling) + .build(None) + .await + .unwrap(); + writer.write(batch).await.unwrap(); + writer.close().await.unwrap() +} + +async fn live_row_count(table: &Table) -> usize { + let mut n = 0; + let mut stream = table + .scan() + .select_all() + .build() + .unwrap() + .to_arrow() + .await + .unwrap(); + while let Some(batch) = stream.try_next().await.unwrap() { + n += batch.num_rows(); + } + n +} + +/// Build a table with two appended data files (ids 1-4, 5-8). +async fn seed_table(warehouse: &TempDir) -> (impl Catalog, Table) { + let catalog = MemoryCatalogBuilder::default() + .load( + "memory", + HashMap::from([( + MEMORY_CATALOG_WAREHOUSE.to_string(), + warehouse.path().to_str().unwrap().to_string(), + )]), + ) + .await + .unwrap(); + let ns = NamespaceIdent::new("db".to_string()); + catalog.create_namespace(&ns, HashMap::new()).await.unwrap(); + + let schema = Schema::builder() + .with_schema_id(0) + .with_fields(vec![ + NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int)).into(), + ]) + .build() + .unwrap(); + let mut table = catalog + .create_table( + &ns, + TableCreation::builder() + .name("t".to_string()) + .schema(schema) + .format_version(FormatVersion::V3) + .build(), + ) + .await + .unwrap(); + + let arrow_schema = Arc::new(ArrowSchema::new(vec![ + Field::new("id", DataType::Int32, false).with_metadata(HashMap::from([( + PARQUET_FIELD_ID_META_KEY.to_string(), + "1".to_string(), + )])), + ])); + for (i, ids) in [vec![1, 2, 3, 4], vec![5, 6, 7, 8]].into_iter().enumerate() { + let batch = + RecordBatch::try_new(arrow_schema.clone(), vec![Arc::new(Int32Array::from(ids))]) + .unwrap(); + let files = write_one_data_file(&table, &format!("data-{i}"), batch).await; + let tx = Transaction::new(&table); + table = tx + .fast_append() + .add_data_files(files) + .apply(tx) + .unwrap() + .commit(&catalog) + .await + .unwrap(); + } + assert_eq!(live_row_count(&table).await, 8); + (catalog, table) +} + +/// Write an unreferenced file under the table's data dir; returns its path. +async fn plant_garbage(table: &Table, name: &str) -> String { + let path = format!("{}/data/{name}", table.metadata().location()); + table + .file_io() + .new_output(&path) + .unwrap() + .write(bytes::Bytes::from_static(b"garbage")) + .await + .unwrap(); + path +} + +/// Dry run: the report contains exactly the planted garbage — never a +/// referenced file — and nothing is deleted. +#[tokio::test] +async fn orphan_dry_run_finds_only_garbage() { + let warehouse = TempDir::new().unwrap(); + let (_catalog, table) = seed_table(&warehouse).await; + let garbage = plant_garbage(&table, "zz-garbage.parquet").await; + + let result = RemoveOrphanFilesAction::new(table.clone()) + .older_than_ms(now_ms() + 60_000) // everything qualifies by age + .dry_run(true) + .execute() + .await + .unwrap(); + + assert_eq!(result.orphan_files, vec![garbage.clone()]); + assert!(result.deleted_files.is_empty(), "dry run must not delete"); + assert!(result.failed_deletes.is_empty()); + assert!( + result.listed_count > result.orphan_files.len(), + "listing must cover referenced files too (listed {})", + result.listed_count + ); + // Nothing was deleted — the garbage file still exists, table intact. + assert!(table.file_io().exists(&garbage).await.unwrap()); + assert_eq!(live_row_count(&table).await, 8); +} + +/// The age guard: a freshly-written garbage file is NOT eligible when the +/// cutoff is in the past. +#[tokio::test] +async fn orphan_age_guard_skips_recent_files() { + let warehouse = TempDir::new().unwrap(); + let (_catalog, table) = seed_table(&warehouse).await; + let garbage = plant_garbage(&table, "zz-recent.parquet").await; + + let result = RemoveOrphanFilesAction::new(table.clone()) + .older_than_ms(now_ms() - 3_600_000) // cutoff 1h ago — nothing is that old + .dry_run(true) + .execute() + .await + .unwrap(); + + assert!( + result.orphan_files.is_empty(), + "recent files must be protected: {:?}", + result.orphan_files + ); + assert_eq!(result.skipped_recent, 1, "the garbage file was age-skipped"); + assert!(table.file_io().exists(&garbage).await.unwrap()); +} + +/// Delete mode removes the garbage and ONLY the garbage: the table remains +/// fully readable afterwards, and a second run reports zero orphans. +#[tokio::test] +async fn orphan_delete_removes_garbage_and_preserves_table() { + let warehouse = TempDir::new().unwrap(); + let (_catalog, table) = seed_table(&warehouse).await; + let g1 = plant_garbage(&table, "zz-garbage-1.parquet").await; + let g2 = plant_garbage(&table, "zz-garbage-2.parquet").await; + + let result = RemoveOrphanFilesAction::new(table.clone()) + .older_than_ms(now_ms() + 60_000) + .execute() + .await + .unwrap(); + + let mut deleted = result.deleted_files.clone(); + deleted.sort(); + let mut expected = vec![g1.clone(), g2.clone()]; + expected.sort(); + assert_eq!(deleted, expected); + assert!(result.failed_deletes.is_empty()); + assert!(!table.file_io().exists(&g1).await.unwrap()); + assert!(!table.file_io().exists(&g2).await.unwrap()); + + // The table is untouched: full scan still returns every row. + assert_eq!(live_row_count(&table).await, 8); + + // Convergence: a second pass finds nothing. + let again = RemoveOrphanFilesAction::new(table.clone()) + .older_than_ms(now_ms() + 60_000) + .dry_run(true) + .execute() + .await + .unwrap(); + assert!(again.orphan_files.is_empty()); +} diff --git a/crates/storage/opendal/public-api.txt b/crates/storage/opendal/public-api.txt index fc1ed7cf78..14824ea15f 100644 --- a/crates/storage/opendal/public-api.txt +++ b/crates/storage/opendal/public-api.txt @@ -24,6 +24,7 @@ pub fn iceberg_storage_opendal::OpenDalStorage::delete<'life0, 'life1, 'async_tr pub fn iceberg_storage_opendal::OpenDalStorage::delete_prefix<'life0, 'life1, 'async_trait>(&'life0 self, path: &'life1 str) -> core::pin::Pin> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait pub fn iceberg_storage_opendal::OpenDalStorage::delete_stream<'life0, 'async_trait>(&'life0 self, paths: futures_core::stream::BoxStream<'static, alloc::string::String>) -> core::pin::Pin> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait pub fn iceberg_storage_opendal::OpenDalStorage::exists<'life0, 'life1, 'async_trait>(&'life0 self, path: &'life1 str) -> core::pin::Pin> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait +pub fn iceberg_storage_opendal::OpenDalStorage::list_prefix<'life0, 'life1, 'async_trait>(&'life0 self, path: &'life1 str) -> core::pin::Pin>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait pub fn iceberg_storage_opendal::OpenDalStorage::metadata<'life0, 'life1, 'async_trait>(&'life0 self, path: &'life1 str) -> core::pin::Pin> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait pub fn iceberg_storage_opendal::OpenDalStorage::new_input(&self, path: &str) -> iceberg::error::Result pub fn iceberg_storage_opendal::OpenDalStorage::new_output(&self, path: &str) -> iceberg::error::Result @@ -69,6 +70,7 @@ pub fn iceberg_storage_opendal::OpenDalResolvingStorage::delete<'life0, 'life1, pub fn iceberg_storage_opendal::OpenDalResolvingStorage::delete_prefix<'life0, 'life1, 'async_trait>(&'life0 self, path: &'life1 str) -> core::pin::Pin> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait pub fn iceberg_storage_opendal::OpenDalResolvingStorage::delete_stream<'life0, 'async_trait>(&'life0 self, paths: futures_core::stream::BoxStream<'static, alloc::string::String>) -> core::pin::Pin> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait pub fn iceberg_storage_opendal::OpenDalResolvingStorage::exists<'life0, 'life1, 'async_trait>(&'life0 self, path: &'life1 str) -> core::pin::Pin> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait +pub fn iceberg_storage_opendal::OpenDalResolvingStorage::list_prefix<'life0, 'life1, 'async_trait>(&'life0 self, path: &'life1 str) -> core::pin::Pin>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait pub fn iceberg_storage_opendal::OpenDalResolvingStorage::metadata<'life0, 'life1, 'async_trait>(&'life0 self, path: &'life1 str) -> core::pin::Pin> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait pub fn iceberg_storage_opendal::OpenDalResolvingStorage::new_input(&self, path: &str) -> iceberg::error::Result pub fn iceberg_storage_opendal::OpenDalResolvingStorage::new_output(&self, path: &str) -> iceberg::error::Result diff --git a/crates/storage/opendal/src/lib.rs b/crates/storage/opendal/src/lib.rs index 768d3ff77f..b708eb7282 100644 --- a/crates/storage/opendal/src/lib.rs +++ b/crates/storage/opendal/src/lib.rs @@ -34,7 +34,7 @@ use cfg_if::cfg_if; use futures::StreamExt; use futures::stream::BoxStream; use iceberg::io::{ - FileMetadata, FileRead, FileWrite, InputFile, OutputFile, Storage, StorageConfig, + FileMetadata, FileRead, FileWrite, InputFile, ListEntry, OutputFile, Storage, StorageConfig, StorageFactory, }; use iceberg::{Error, ErrorKind, Result}; @@ -549,6 +549,37 @@ impl Storage for OpenDalStorage { .map_err(from_opendal_error)?) } + async fn list_prefix(&self, path: &str) -> Result> { + let (op, relative_path) = self.create_operator(&path)?; + // The absolute prefix is the input path minus its operator-relative + // suffix; entry paths (operator-root-relative) append onto it. + let absolute_prefix = &path[..path.len() - relative_path.len()]; + let list_path = if relative_path.is_empty() || relative_path.ends_with('/') { + relative_path.to_string() + } else { + format!("{relative_path}/") + }; + let entries = op + .list_with(&list_path) + .recursive(true) + .await + .map_err(from_opendal_error)?; + Ok(entries + .into_iter() + .filter(|e| !e.metadata().is_dir()) + .map(|e| ListEntry { + path: format!("{absolute_prefix}{}", e.path()), + size: e.metadata().content_length(), + last_modified_ms: e.metadata().last_modified().and_then(|dt| { + let t: std::time::SystemTime = dt.into(); + t.duration_since(std::time::UNIX_EPOCH) + .ok() + .map(|d| d.as_millis() as i64) + }), + }) + .collect()) + } + async fn delete_stream(&self, mut paths: BoxStream<'static, String>) -> Result<()> { let mut deleters: HashMap = HashMap::new(); diff --git a/crates/storage/opendal/src/resolving.rs b/crates/storage/opendal/src/resolving.rs index 86993220a8..8faf382355 100644 --- a/crates/storage/opendal/src/resolving.rs +++ b/crates/storage/opendal/src/resolving.rs @@ -289,6 +289,10 @@ impl Storage for OpenDalResolvingStorage { self.resolve(path)?.delete_prefix(path).await } + async fn list_prefix(&self, path: &str) -> Result> { + self.resolve(path)?.list_prefix(path).await + } + async fn delete_stream(&self, mut paths: BoxStream<'static, String>) -> Result<()> { // Group paths by canonical scheme so each resolved storage receives a batch, // avoiding repeated operator creation per path.