clean up code paths

rust/lance-io/src/local.rs

@@ -111,13 +111,20 @@ impl Reader for LocalObjectReader {
     }
 
     /// Returns the file size.
-    async fn size(&self) -> Result<usize> {
-        Ok(self.file.metadata()?.len() as usize)
+    async fn size(&self) -> object_store::Result<usize> {
+        let metadata = self
+            .file
+            .metadata()
+            .map_err(|err| object_store::Error::Generic {
+                store: "LocalFileSystem",
+                source: err.into(),
+            })?;
+        Ok(metadata.len() as usize)
     }
 
     /// Reads a range of data.
     #[instrument(level = "debug", skip(self))]
-    async fn get_range(&self, range: Range<usize>) -> Result<Bytes> {
+    async fn get_range(&self, range: Range<usize>) -> object_store::Result<Bytes> {
         let file = self.file.clone();
         tokio::task::spawn_blocking(move || {
             let mut buf = BytesMut::with_capacity(range.len());
@@ -132,6 +139,10 @@ impl Reader for LocalObjectReader {
             Ok(buf.freeze())
         })
         .await?
+        .map_err(|err: std::io::Error| object_store::Error::Generic {
+            store: "LocalFileSystem",
+            source: err.into(),
+        })
     }
 }
 

rust/lance-io/src/object_reader.rs

@@ -42,14 +42,14 @@ impl CloudObjectReader {
     async fn do_with_retry<'a, O>(
         &self,
         f: impl Fn() -> BoxFuture<'a, std::result::Result<O, object_store::Error>>,
-    ) -> Result<O> {
+    ) -> object_store::Result<O> {
         let mut retries = 3;
         loop {
             match f().await {
                 Ok(val) => return Ok(val),
                 Err(err) => {
                     if retries == 0 {
-                        return Err(err.into());
+                        return Err(err);
                     }
                     retries -= 1;
                 }
@@ -69,15 +69,15 @@ impl Reader for CloudObjectReader {
     }
 
     /// Object/File Size.
-    async fn size(&self) -> Result<usize> {
+    async fn size(&self) -> object_store::Result<usize> {
         let meta = self
             .do_with_retry(|| self.object_store.head(&self.path))
             .await?;
         Ok(meta.size)
     }
 
     #[instrument(level = "debug", skip(self))]
-    async fn get_range(&self, range: Range<usize>) -> Result<Bytes> {
+    async fn get_range(&self, range: Range<usize>) -> object_store::Result<Bytes> {
         self.do_with_retry(|| self.object_store.get_range(&self.path, range.clone()))
             .await
     }

rust/lance-io/src/scheduler.rs

@@ -91,7 +91,8 @@ impl IoTask {
         let bytes = self
             .reader
             .get_range(self.to_read.start as usize..self.to_read.end as usize)
-            .await;
+            .await
+            .map_err(Error::from);
         (self.when_done)(bytes);
     }
 }

rust/lance-io/src/traits.rs

@@ -86,10 +86,10 @@ pub trait Reader: std::fmt::Debug + Send + Sync {
     fn block_size(&self) -> usize;
 
     /// Object/File Size.
-    async fn size(&self) -> Result<usize>;
+    async fn size(&self) -> object_store::Result<usize>;
 
     /// Read a range of bytes from the object.
     ///
     /// TODO: change to read_at()?
-    async fn get_range(&self, range: Range<usize>) -> Result<Bytes>;
+    async fn get_range(&self, range: Range<usize>) -> object_store::Result<Bytes>;
 }

rust/lance-io/src/utils.rs

@@ -113,8 +113,15 @@ pub async fn read_struct<
     T::try_from(msg)
 }
 
-pub async fn read_last_block(reader: &dyn Reader) -> Result<Bytes> {
-    let file_size = reader.size().await?;
+pub async fn read_last_block(
+    reader: &dyn Reader,
+    size: Option<u64>,
+) -> object_store::Result<Bytes> {
+    let file_size = if let Some(size) = size {
+        size as usize
+    } else {
+        reader.size().await?
+    };
     let block_size = reader.block_size();
     let begin = if file_size < block_size {
         0