Update errors

Author: matthewmturner

datafusion/src/datasource/listing/helpers.rs

@@ -35,7 +35,7 @@ use futures::{
 use log::debug;
 
 use crate::{
-    error::Result,
+    error::{DataFusionError, Result},
     execution::context::ExecutionContext,
     logical_plan::{self, Expr, ExpressionVisitor, Recursion},
     physical_plan::functions::Volatility,
@@ -191,12 +191,13 @@ pub async fn pruned_partition_list(
         Ok(Box::pin(
             store
                 .list_file_with_suffix(table_path, file_extension)
+                // .map_err(DataFusionError::from)
                 .await?
                 .filter_map(move |f| {
                     let stream_path = stream_path.clone();
                     let table_partition_cols_stream = table_partition_cols_stream.clone();
                     async move {
-                        let file_meta = match f {
+                        let file_meta = match f.map_err(DataFusionError::from) {
                             Ok(fm) => fm,
                             Err(err) => return Some(Err(err)),
                         };
@@ -230,6 +231,7 @@ pub async fn pruned_partition_list(
             // all the files anyway. This number will need to be adjusted according to the object
             // store if we switch to a streaming-stlye pruning of the files. For instance S3 lists
             // 1000 items at a time so batches of 1000 would be ideal with S3 as store.
+            .map_err(DataFusionError::from)
             .chunks(1024)
             .map(|v| v.into_iter().collect::<Result<Vec<_>>>())
             .map(move |metas| paths_to_batch(table_partition_cols, &stream_path, &metas?))

datafusion/src/datasource/object_store/local.rs

@@ -17,6 +17,7 @@
 
 //! Object store that represents the Local File System.
 
+use std::error::Error;
 use std::fs::{self, File, Metadata};
 use std::io::{BufReader, Read, Seek, SeekFrom};
 use std::sync::Arc;
@@ -28,8 +29,7 @@ use crate::datasource::object_store::{
     FileMeta, FileMetaStream, ListEntryStream, ObjectReader, ObjectStore,
 };
 use crate::datasource::PartitionedFile;
-use crate::error::DataFusionError;
-use crate::error::Result;
+use crate::error::{DataFusionError, Result as DataFusionResult};
 
 use super::{ObjectReaderStream, SizedFile};
 
@@ -39,19 +39,25 @@ pub struct LocalFileSystem;
 
 #[async_trait]
 impl ObjectStore for LocalFileSystem {
-    async fn list_file(&self, prefix: &str) -> Result<FileMetaStream> {
+    async fn list_file(
+        &self,
+        prefix: &str,
+    ) -> Result<FileMetaStream, Box<dyn Error + Send + Sync>> {
         list_all(prefix.to_owned()).await
     }
 
     async fn list_dir(
         &self,
         _prefix: &str,
         _delimiter: Option<String>,
-    ) -> Result<ListEntryStream> {
+    ) -> Result<ListEntryStream, Box<dyn Error + Send + Sync>> {
         todo!()
     }
 
-    fn file_reader(&self, file: SizedFile) -> Result<Arc<dyn ObjectReader>> {
+    fn file_reader(
+        &self,
+        file: SizedFile,
+    ) -> Result<Arc<dyn ObjectReader>, Box<dyn Error + Send + Sync>> {
         Ok(Arc::new(LocalFileReader::new(file)?))
     }
 }
@@ -61,7 +67,7 @@ struct LocalFileReader {
 }
 
 impl LocalFileReader {
-    fn new(file: SizedFile) -> Result<Self> {
+    fn new(file: SizedFile) -> DataFusionResult<Self> {
         Ok(Self { file })
     }
 }
@@ -72,7 +78,7 @@ impl ObjectReader for LocalFileReader {
         &self,
         _start: u64,
         _length: usize,
-    ) -> Result<Box<dyn AsyncRead>> {
+    ) -> Result<Box<dyn AsyncRead>, Box<dyn Error + Send + Sync>> {
         todo!(
             "implement once async file readers are available (arrow-rs#78, arrow-rs#111)"
         )
@@ -82,7 +88,7 @@ impl ObjectReader for LocalFileReader {
         &self,
         start: u64,
         length: usize,
-    ) -> Result<Box<dyn Read + Send + Sync>> {
+    ) -> Result<Box<dyn Read + Send + Sync>, Box<dyn Error + Send + Sync>> {
         // A new file descriptor is opened for each chunk reader.
         // This okay because chunks are usually fairly large.
         let mut file = File::open(&self.file.path)?;
@@ -98,7 +104,9 @@ impl ObjectReader for LocalFileReader {
     }
 }
 
-async fn list_all(prefix: String) -> Result<FileMetaStream> {
+async fn list_all(
+    prefix: String,
+) -> Result<FileMetaStream, Box<dyn Error + Send + Sync>> {
     fn get_meta(path: String, metadata: Metadata) -> FileMeta {
         FileMeta {
             sized_file: SizedFile {
@@ -112,7 +120,7 @@ async fn list_all(prefix: String) -> Result<FileMetaStream> {
     async fn find_files_in_dir(
         path: String,
         to_visit: &mut Vec<String>,
-    ) -> Result<Vec<FileMeta>> {
+    ) -> DataFusionResult<Vec<FileMeta>> {
         let mut dir = tokio::fs::read_dir(path).await?;
         let mut files = Vec::new();
 
@@ -194,7 +202,7 @@ mod tests {
     use tempfile::tempdir;
 
     #[tokio::test]
-    async fn test_recursive_listing() -> Result<()> {
+    async fn test_recursive_listing() -> DataFusionResult<()> {
         // tmp/a.txt
         // tmp/x/b.txt
         // tmp/y/c.txt

datafusion/src/datasource/object_store/mod.rs

@@ -20,6 +20,7 @@
 pub mod local;
 
 use std::collections::HashMap;
+use std::error::Error;
 use std::fmt::{self, Debug};
 use std::io::Read;
 use std::pin::Pin;
@@ -31,7 +32,7 @@ use futures::{AsyncRead, Stream, StreamExt};
 
 use local::LocalFileSystem;
 
-use crate::error::{DataFusionError, Result};
+use crate::error::{DataFusionError, Result as DataFusionResult};
 
 /// Object Reader for one file in an object store.
 ///
@@ -40,18 +41,23 @@ use crate::error::{DataFusionError, Result};
 #[async_trait]
 pub trait ObjectReader: Send + Sync {
     /// Get reader for a part [start, start + length] in the file asynchronously
-    async fn chunk_reader(&self, start: u64, length: usize)
-        -> Result<Box<dyn AsyncRead>>;
+    async fn chunk_reader(
+        &self,
+        start: u64,
+        length: usize,
+    ) -> Result<Box<dyn AsyncRead>, Box<dyn Error + Send + Sync>>;
 
     /// Get reader for a part [start, start + length] in the file
     fn sync_chunk_reader(
         &self,
         start: u64,
         length: usize,
-    ) -> Result<Box<dyn Read + Send + Sync>>;
+    ) -> Result<Box<dyn Read + Send + Sync>, Box<dyn Error + Send + Sync>>;
 
     /// Get reader for the entire file
-    fn sync_reader(&self) -> Result<Box<dyn Read + Send + Sync>> {
+    fn sync_reader(
+        &self,
+    ) -> Result<Box<dyn Read + Send + Sync>, Box<dyn Error + Send + Sync>> {
         self.sync_chunk_reader(0, self.length() as usize)
     }
 
@@ -114,29 +120,38 @@ impl std::fmt::Display for FileMeta {
 
 /// Stream of files listed from object store
 pub type FileMetaStream =
-    Pin<Box<dyn Stream<Item = Result<FileMeta>> + Send + Sync + 'static>>;
+    Pin<Box<dyn Stream<Item = DataFusionResult<FileMeta>> + Send + Sync + 'static>>;
 
 /// Stream of list entries obtained from object store
 pub type ListEntryStream =
-    Pin<Box<dyn Stream<Item = Result<ListEntry>> + Send + Sync + 'static>>;
+    Pin<Box<dyn Stream<Item = DataFusionResult<ListEntry>> + Send + Sync + 'static>>;
 
 /// Stream readers opened on a given object store
-pub type ObjectReaderStream =
-    Pin<Box<dyn Stream<Item = Result<Arc<dyn ObjectReader>>> + Send + Sync + 'static>>;
+pub type ObjectReaderStream = Pin<
+    Box<
+        dyn Stream<Item = Result<Arc<dyn ObjectReader>, Box<dyn Error + Send + Sync>>>
+            + Send
+            + Sync
+            + 'static,
+    >,
+>;
 
 /// A ObjectStore abstracts access to an underlying file/object storage.
 /// It maps strings (e.g. URLs, filesystem paths, etc) to sources of bytes
 #[async_trait]
 pub trait ObjectStore: Sync + Send + Debug {
     /// Returns all the files in path `prefix`
-    async fn list_file(&self, prefix: &str) -> Result<FileMetaStream>;
+    async fn list_file(
+        &self,
+        prefix: &str,
+    ) -> Result<FileMetaStream, Box<dyn Error + Send + Sync>>;
 
     /// Calls `list_file` with a suffix filter
     async fn list_file_with_suffix(
         &self,
         prefix: &str,
         suffix: &str,
-    ) -> Result<FileMetaStream> {
+    ) -> Result<FileMetaStream, Box<dyn Error + Send + Sync>> {
         let file_stream = self.list_file(prefix).await?;
         let suffix = suffix.to_owned();
         Ok(Box::pin(file_stream.filter(move |fr| {
@@ -154,10 +169,13 @@ pub trait ObjectStore: Sync + Send + Debug {
         &self,
         prefix: &str,
         delimiter: Option<String>,
-    ) -> Result<ListEntryStream>;
+    ) -> Result<ListEntryStream, Box<dyn Error + Send + Sync>>;
 
     /// Get object reader for one file
-    fn file_reader(&self, file: SizedFile) -> Result<Arc<dyn ObjectReader>>;
+    fn file_reader(
+        &self,
+        file: SizedFile,
+    ) -> Result<Arc<dyn ObjectReader>, Box<dyn Error + Send + Sync>>;
 }
 
 static LOCAL_SCHEME: &str = "file";
@@ -223,7 +241,7 @@ impl ObjectStoreRegistry {
     pub fn get_by_uri<'a>(
         &self,
         uri: &'a str,
-    ) -> Result<(Arc<dyn ObjectStore>, &'a str)> {
+    ) -> Result<(Arc<dyn ObjectStore>, &'a str), Box<dyn Error + Send + Sync>> {
         if let Some((scheme, path)) = uri.split_once("://") {
             let stores = self.object_stores.read().unwrap();
             let store = stores

datafusion/src/error.rs

@@ -61,6 +61,9 @@ pub enum DataFusionError {
     /// Error returned during execution of the query.
     /// Examples include files not found, errors in parsing certain types.
     Execution(String),
+    /// Errors originating from outside DataFusion's core codebase.
+    /// For example, a custom S3Error from the crate datafusion-objectstore-s3
+    External(Box<dyn error::Error + Send + Sync>),
 }
 
 impl DataFusionError {
@@ -101,6 +104,12 @@ impl From<ParserError> for DataFusionError {
     }
 }
 
+impl From<Box<dyn error::Error + Send + Sync>> for DataFusionError {
+    fn from(err: Box<dyn error::Error + Send + Sync>) -> Self {
+        DataFusionError::External(err)
+    }
+}
+
 impl Display for DataFusionError {
     fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
         match *self {
@@ -129,6 +138,9 @@ impl Display for DataFusionError {
             DataFusionError::Execution(ref desc) => {
                 write!(f, "Execution error: {}", desc)
             }
+            DataFusionError::External(ref desc) => {
+                write!(f, "External error: {}", desc)
+            }
         }
     }
 }

datafusion/src/execution/context.rs

@@ -566,6 +566,7 @@ impl ExecutionContext {
             .unwrap()
             .object_store_registry
             .get_by_uri(uri)
+            .map_err(DataFusionError::from)
     }
 
     /// Registers a table using a custom `TableProvider` so that

datafusion/src/physical_plan/file_format/file_stream.rs

@@ -41,6 +41,7 @@ use std::{
 };
 
 use super::PartitionColumnProjector;
+use crate::error::DataFusionError;
 
 pub type FileIter = Box<dyn Iterator<Item = PartitionedFile> + Send + Sync>;
 pub type BatchIter = Box<dyn Iterator<Item = ArrowResult<RecordBatch>> + Send + Sync>;
@@ -124,7 +125,9 @@ impl<F: FormatReaderOpener> FileStream<F> {
                     self.object_store
                         .file_reader(f.file_meta.sized_file)
                         .and_then(|r| r.sync_reader())
-                        .map_err(|e| ArrowError::ExternalError(Box::new(e)))
+                        .map_err(|e| {
+                            ArrowError::ExternalError(Box::new(DataFusionError::from(e)))
+                        })
                         .and_then(|f| {
                             self.batch_iter = (self.file_reader)(f, &self.remain);
                             self.next_batch().transpose()