diff --git a/crates/iceberg/src/io/file_io.rs b/crates/iceberg/src/io/file_io.rs
index 3a54b1b0b..5e30968f5 100644
--- a/crates/iceberg/src/io/file_io.rs
+++ b/crates/iceberg/src/io/file_io.rs
@@ -71,18 +71,40 @@ impl FileIO {
     }
 
     /// Deletes file.
+    ///
+    /// # Arguments
+    ///
+    /// * path: It should be *absolute* path starting with scheme string used to construct [`FileIO`].
     pub async fn delete(&self, path: impl AsRef<str>) -> Result<()> {
         let (op, relative_path) = self.inner.create_operator(&path)?;
         Ok(op.delete(relative_path).await?)
     }
 
+    /// Remove the path and all nested dirs and files recursively.
+    ///
+    /// # Arguments
+    ///
+    /// * path: It should be *absolute* path starting with scheme string used to construct [`FileIO`].
+    pub async fn remove_all(&self, path: impl AsRef<str>) -> Result<()> {
+        let (op, relative_path) = self.inner.create_operator(&path)?;
+        Ok(op.remove_all(relative_path).await?)
+    }
+
     /// Check file exists.
+    ///
+    /// # Arguments
+    ///
+    /// * path: It should be *absolute* path starting with scheme string used to construct [`FileIO`].
     pub async fn is_exist(&self, path: impl AsRef<str>) -> Result<bool> {
         let (op, relative_path) = self.inner.create_operator(&path)?;
         Ok(op.is_exist(relative_path).await?)
     }
 
     /// Creates input file.
+    ///
+    /// # Arguments
+    ///
+    /// * path: It should be *absolute* path starting with scheme string used to construct [`FileIO`].
     pub fn new_input(&self, path: impl AsRef<str>) -> Result<InputFile> {
         let (op, relative_path) = self.inner.create_operator(&path)?;
         let path = path.as_ref().to_string();
@@ -95,6 +117,10 @@ impl FileIO {
     }
 
     /// Creates output file.
+    ///
+    /// # Arguments
+    ///
+    /// * path: It should be *absolute* path starting with scheme string used to construct [`FileIO`].
     pub fn new_output(&self, path: impl AsRef<str>) -> Result<OutputFile> {
         let (op, relative_path) = self.inner.create_operator(&path)?;
         let path = path.as_ref().to_string();
@@ -338,7 +364,7 @@ impl OutputFile {
 
 #[cfg(test)]
 mod tests {
-    use std::fs::File;
+    use std::fs::{create_dir_all, File};
     use std::io::Write;
     use std::path::Path;
 
@@ -353,6 +379,7 @@ mod tests {
     }
 
     fn write_to_file<P: AsRef<Path>>(s: &str, path: P) {
+        create_dir_all(path.as_ref().parent().unwrap()).unwrap();
         let mut f = File::create(path).unwrap();
         write!(f, "{s}").unwrap();
     }
@@ -389,16 +416,24 @@ mod tests {
     async fn test_delete_local_file() {
         let tmp_dir = TempDir::new().unwrap();
 
-        let file_name = "a.txt";
-        let content = "Iceberg loves rust.";
-
-        let full_path = format!("{}/{}", tmp_dir.path().to_str().unwrap(), file_name);
-        write_to_file(content, &full_path);
+        let a_path = format!("{}/{}", tmp_dir.path().to_str().unwrap(), "a.txt");
+        let sub_dir_path = format!("{}/sub", tmp_dir.path().to_str().unwrap());
+        let b_path = format!("{}/{}", sub_dir_path, "b.txt");
+        let c_path = format!("{}/{}", sub_dir_path, "c.txt");
+        write_to_file("Iceberg loves rust.", &a_path);
+        write_to_file("Iceberg loves rust.", &b_path);
+        write_to_file("Iceberg loves rust.", &c_path);
 
         let file_io = create_local_file_io();
-        assert!(file_io.is_exist(&full_path).await.unwrap());
-        file_io.delete(&full_path).await.unwrap();
-        assert!(!file_io.is_exist(&full_path).await.unwrap());
+        assert!(file_io.is_exist(&a_path).await.unwrap());
+
+        file_io.remove_all(&sub_dir_path).await.unwrap();
+        assert!(!file_io.is_exist(&b_path).await.unwrap());
+        assert!(!file_io.is_exist(&c_path).await.unwrap());
+        assert!(file_io.is_exist(&a_path).await.unwrap());
+
+        file_io.delete(&a_path).await.unwrap();
+        assert!(!file_io.is_exist(&a_path).await.unwrap());
     }
 
     #[tokio::test]
@@ -411,6 +446,7 @@ mod tests {
         let file_io = create_local_file_io();
         assert!(!file_io.is_exist(&full_path).await.unwrap());
         assert!(file_io.delete(&full_path).await.is_ok());
+        assert!(file_io.remove_all(&full_path).await.is_ok());
     }
 
     #[tokio::test]