diff --git a/crates/iceberg/public-api.txt b/crates/iceberg/public-api.txt index 41d4d7691b..6da21aed05 100644 --- a/crates/iceberg/public-api.txt +++ b/crates/iceberg/public-api.txt @@ -688,6 +688,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 @@ -770,6 +771,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 @@ -784,6 +797,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 @@ -822,6 +836,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 @@ -1010,6 +1025,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 @@ -1022,6 +1038,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 @@ -1034,6 +1051,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 diff --git a/crates/iceberg/src/io/file_io.rs b/crates/iceberg/src/io/file_io.rs index 227d8f4d5b..194b1abb26 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/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 52f4e68ed3..ebcd8d26a0 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(); @@ -654,6 +685,36 @@ mod tests { ); } + #[cfg(feature = "opendal-memory")] + #[tokio::test] + async fn test_list_prefix_is_directory_style_for_opendal_memory() { + let storage = OpenDalStorage::Memory(default_memory_operator()); + let file = "memory:/warehouse/t/data/a.parquet"; + storage + .write(file, Bytes::from_static(b"aa")) + .await + .unwrap(); + + let (op, relative_path) = storage.create_operator(&file).unwrap(); + let raw_entries = op.list_with(relative_path).recursive(true).await.unwrap(); + assert!(raw_entries.iter().any( + |entry| entry.path() == "warehouse/t/data/a.parquet" && !entry.metadata().is_dir() + )); + + 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(), 1); + assert_eq!(entries[0].path, file); + assert_eq!(entries[0].size, 2); + + let slash_entries = storage.list_prefix("memory:/warehouse/t/").await.unwrap(); + assert_eq!(slash_entries.len(), 1); + assert_eq!(slash_entries[0].path, file); + + assert!(storage.list_prefix(file).await.unwrap().is_empty()); + assert!(storage.exists(file).await.unwrap()); + } + #[cfg(feature = "opendal-fs")] #[test] fn test_relativize_path_fs() { 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.