feat(maintenance): RemoveOrphanFilesAction with dry-run-first safety

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<str>) -> Result<Vec<ListEntry>> {
+        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<i64>,
+}
+
 /// Trait for reading file.
 ///
 /// # TODO

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<Vec<ListEntry>> {
+        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::<String>::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());
+    }
 }