List files

crates/iceberg/src/io/file_io.rs

@@ -110,6 +110,16 @@ impl FileIO {
         Ok(op.exists(relative_path).await?)
     }
 
+    /// List 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 list(&self, path: impl AsRef<str>) -> Result<Vec<opendal::Entry>> {
+        let (op, relative_path) = self.inner.create_operator(&path)?;
+        Ok(op.list(relative_path).await?)
+    }
+
     /// Creates input file.
     ///
     /// # Arguments
@@ -355,6 +365,16 @@ impl OutputFile {
         writer.close().await
     }
 
+    /// Create a new output file with given bytes.
+    /// Error out if the file already exists
+    pub async fn write_exclusive(&self, bs: Bytes) -> crate::Result<()> {
+        self.op
+            .write_with(&self.path[self.relative_path_pos..], bs)
+            .if_not_exists(true)
+            .await?;
+        Ok(())
+    }
+
     /// Creates output file for continues writing.
     ///
     /// # Notes

crates/integrations/datafusion/src/table/mod.rs

@@ -107,6 +107,10 @@ impl IcebergTableProvider {
             schema,
         })
     }
+
+    pub fn table(&self) -> Table {
+        self.table.clone()
+    }
 }
 
 #[async_trait]