pure storage, fixed compile issues

Author: CTTY

crates/iceberg/src/arrow/reader.rs

@@ -327,7 +327,7 @@ impl ArrowReader {
         data_file_path: &str,
         file_io: FileIO,
         should_load_page_index: bool,
-    ) -> Result<ParquetRecordBatchStreamBuilder<ArrowFileReader<impl FileRead + Sized>>> {
+    ) -> Result<ParquetRecordBatchStreamBuilder<ArrowFileReader>> {
         // Get the metadata for the Parquet file we need to read and build
         // a reader for the data within
         let parquet_file = file_io.new_input(data_file_path)?;
@@ -1312,18 +1312,18 @@ impl BoundPredicateVisitor for PredicateConverter<'_> {
 }
 
 /// ArrowFileReader is a wrapper around a FileRead that impls parquets AsyncFileReader.
-pub struct ArrowFileReader<R: FileRead> {
+pub struct ArrowFileReader {
     meta: FileMetadata,
     preload_column_index: bool,
     preload_offset_index: bool,
     preload_page_index: bool,
     metadata_size_hint: Option<usize>,
-    r: R,
+    r: Box<dyn FileRead>,
 }
 
-impl<R: FileRead> ArrowFileReader<R> {
+impl ArrowFileReader {
     /// Create a new ArrowFileReader
-    pub fn new(meta: FileMetadata, r: R) -> Self {
+    pub fn new(meta: FileMetadata, r: Box<dyn FileRead>) -> Self {
         Self {
             meta,
             preload_column_index: false,
@@ -1362,7 +1362,7 @@ impl<R: FileRead> ArrowFileReader<R> {
     }
 }
 
-impl<R: FileRead> AsyncFileReader for ArrowFileReader<R> {
+impl AsyncFileReader for ArrowFileReader {
     fn get_bytes(&mut self, range: Range<u64>) -> BoxFuture<'_, parquet::errors::Result<Bytes>> {
         Box::pin(
             self.r

crates/iceberg/src/io/file_io.rs

@@ -27,6 +27,7 @@ use url::Url;
 use super::storage::OpenDALStorage;
 use crate::{Error, ErrorKind, Result};
 
+/// todo doc
 #[async_trait]
 pub trait Storage: Debug + Send + Sync {
     /// Check if a file exists at the given path
@@ -326,8 +327,10 @@ impl FileRead for opendal::Reader {
 /// Input file is used for reading from files.
 #[derive(Debug)]
 pub struct InputFile {
+    /// todo doc
     pub storage: Arc<dyn Storage>,
     // Absolution path of file.
+    /// todo doc
     pub path: String,
     // todo should remove this? Should always pass down a full path
     // // Relative path of file to uri, starts at [`relative_path_pos`]
@@ -353,7 +356,7 @@ impl InputFile {
     /// Read and returns whole content of file.
     ///
     /// For continuous reading, use [`Self::reader`] instead.
-    pub async fn read(&self) -> crate::Result<Bytes> {
+    pub async fn read(&self) -> Result<Bytes> {
         self
             .storage
             .read(&self.path)
@@ -363,7 +366,7 @@ impl InputFile {
     /// Creates [`FileRead`] for continuous reading.
     ///
     /// For one-time reading, use [`Self::read`] instead.
-    pub async fn reader(&self) -> crate::Result<Box<dyn FileRead>> {
+    pub async fn reader(&self) -> Result<Box<dyn FileRead>> {
         self.storage.reader(&self.path).await
     }
 }
@@ -374,17 +377,17 @@ impl InputFile {
 ///
 /// It's possible for us to remove the async_trait, but we need to figure
 /// out how to handle the object safety.
-#[async_trait::async_trait]
-pub trait FileWrite: Send + Unpin + 'static {
+#[async_trait]
+pub trait FileWrite: Send + Sync + Unpin + 'static {
     /// Write bytes to file.
     ///
     /// TODO: we can support writing non-contiguous bytes in the future.
-    async fn write(&mut self, bs: Bytes) -> crate::Result<()>;
+    async fn write(&mut self, bs: Bytes) -> Result<()>;
 
     /// Close file.
     ///
     /// Calling close on closed file will generate an error.
-    async fn close(&mut self) -> crate::Result<()>;
+    async fn close(&mut self) -> Result<()>;
 }
 
 #[async_trait::async_trait]
@@ -413,8 +416,10 @@ impl FileWrite for Box<dyn FileWrite> {
 /// Output file is used for writing to files..
 #[derive(Debug)]
 pub struct OutputFile {
+    /// todo fix pub qualifier
     pub storage: Arc<dyn Storage>,
     // Absolution path of file.
+    /// todo fix pub qualifier
     pub path: String,
     // todo should always pass down a full path
     // // Relative path of file to uri, starts at [`relative_path_pos`]

crates/iceberg/src/io/storage.rs

@@ -15,7 +15,9 @@
 // specific language governing permissions and limitations
 // under the License.
 
+use crate::Result;
 use std::sync::Arc;
+use async_trait::async_trait;
 use bytes::Bytes;
 use opendal::layers::RetryLayer;
 #[cfg(feature = "storage-azdls")]
@@ -67,14 +69,14 @@ pub(crate) enum OpenDALStorage {
     },
 }
 
-#[async_trait::async_trait]
+#[async_trait]
 impl Storage for OpenDALStorage {
-    async fn exists(&self, path: &str) -> crate::Result<bool> {
+    async fn exists(&self, path: &str) -> Result<bool> {
         let (op, relative_path) = self.create_operator(&path)?;
         Ok(op.exists(relative_path).await?)
     }
 
-    async fn metadata(&self, path: &str) -> crate::Result<FileMetadata> {
+    async fn metadata(&self, path: &str) -> Result<FileMetadata> {
         let (op, relative_path) = self.create_operator(&path)?;
         let meta = op.stat(relative_path).await?;
 
@@ -83,35 +85,35 @@ impl Storage for OpenDALStorage {
         })
     }
 
-    async fn read(&self, path: &str) -> crate::Result<Bytes> {
+    async fn read(&self, path: &str) -> Result<Bytes> {
         let (op, relative_path) = self.create_operator(&path)?;
         Ok(op.read(relative_path).await?.to_bytes())
     }
 
-    async fn reader(&self, path: &str) -> crate::Result<Box<dyn FileRead>> {
+    async fn reader(&self, path: &str) -> Result<Box<dyn FileRead>> {
         let (op, relative_path) = self.create_operator(&path)?;
         Ok(Box::new(op.reader(relative_path).await?))
     }
 
-    async fn write(&self, path: &str, bs: Bytes) -> crate::Result<()> {
+    async fn write(&self, path: &str, bs: Bytes) -> Result<()> {
         let mut writer = self.writer(path).await?;
         writer.write(bs).await?;
         writer.close().await
     }
 
-    async fn writer(&self, path: &str) -> crate::Result<Box<dyn FileWrite>> {
+    async fn writer(&self, path: &str) -> Result<Box<dyn FileWrite>> {
         let (op, relative_path) = self.create_operator(&path)?;
         Ok(Box::new(
             op.writer(relative_path).await?,
         ))
     }
 
-    async fn delete(&self, path: &str) -> crate::Result<()> {
+    async fn delete(&self, path: &str) -> Result<()> {
         let (op, relative_path) = self.create_operator(&path)?;
         Ok(op.delete(relative_path).await?)
     }
 
-    async fn remove_dir_all(&self, path: &str) -> crate::Result<()> {
+    async fn remove_dir_all(&self, path: &str) -> Result<()> {
         let (op, relative_path) = self.create_operator(&path)?;
         let path = if relative_path.ends_with('/') {
             relative_path.to_string()
@@ -121,7 +123,7 @@ impl Storage for OpenDALStorage {
         Ok(op.remove_all(&path).await?)
     }
 
-    fn new_input(&self, path: &str) -> crate::Result<InputFile> {
+    fn new_input(&self, path: &str) -> Result<InputFile> {
         let storage = Arc::new(self.clone());
         let path = path.to_string();
         Ok(InputFile {
@@ -130,7 +132,7 @@ impl Storage for OpenDALStorage {
         })
     }
 
-    fn new_output(&self, path: &str) -> crate::Result<OutputFile> {
+    fn new_output(&self, path: &str) -> Result<OutputFile> {
         let storage = Arc::new(self.clone());
         let path = path.to_string();
         Ok(OutputFile {
@@ -142,7 +144,7 @@ impl Storage for OpenDALStorage {
 
 impl OpenDALStorage {
     /// Convert iceberg config to opendal config.
-    pub(crate) fn build(file_io_builder: FileIOBuilder) -> crate::Result<Self> {
+    pub(crate) fn build(file_io_builder: FileIOBuilder) -> Result<Self> {
         let (scheme_str, props, extensions) = file_io_builder.into_parts();
         let scheme = Self::parse_scheme(&scheme_str)?;
 

crates/iceberg/src/puffin/metadata.rs

@@ -16,7 +16,6 @@
 // under the License.
 
 use std::collections::{HashMap, HashSet};
-
 use bytes::Bytes;
 use serde::{Deserialize, Serialize};
 
@@ -288,9 +287,9 @@ impl FileMetadata {
 
         let input_file_length = input_file.metadata().await?.size;
         let footer_payload_length =
-            FileMetadata::read_footer_payload_length(&file_read, input_file_length).await?;
+            FileMetadata::read_footer_payload_length(file_read.as_ref(), input_file_length).await?;
         let footer_bytes =
-            FileMetadata::read_footer_bytes(&file_read, input_file_length, footer_payload_length)
+            FileMetadata::read_footer_bytes(file_read.as_ref(), input_file_length, footer_payload_length)
                 .await?;
 
         let magic_length = FileMetadata::MAGIC_LENGTH as usize;

crates/iceberg/src/puffin/reader.rs

@@ -18,7 +18,7 @@
 use tokio::sync::OnceCell;
 
 use crate::Result;
-use crate::io::{FileRead, InputFile};
+use crate::io::InputFile;
 use crate::puffin::blob::Blob;
 use crate::puffin::metadata::{BlobMetadata, FileMetadata};