Use DataFusionError instead of ArrowError in FileOpenFuture

Author: adriangb

datafusion/core/tests/physical_optimizer/filter_pushdown/util.rs

@@ -18,6 +18,7 @@
 use arrow::datatypes::SchemaRef;
 use arrow::{array::RecordBatch, compute::concat_batches};
 use datafusion::{datasource::object_store::ObjectStoreUrl, physical_plan::PhysicalExpr};
+use futures::StreamExt;
 use datafusion_common::{config::ConfigOptions, internal_err, Result, Statistics};
 use datafusion_datasource::{
     file::FileSource, file_meta::FileMeta, file_scan_config::FileScanConfig,
@@ -38,7 +39,6 @@ use datafusion_physical_plan::{
     metrics::ExecutionPlanMetricsSet,
     DisplayAs, DisplayFormatType, ExecutionPlan, PlanProperties,
 };
-use futures::stream::BoxStream;
 use futures::{FutureExt, Stream};
 use object_store::ObjectStore;
 use std::{
@@ -93,11 +93,7 @@ impl FileOpener for TestOpener {
         let stream = TestStream::new(batches);
 
         Ok((async {
-            let stream: BoxStream<
-                'static,
-                Result<RecordBatch, datafusion_common::DataFusionError>,
-            > = Box::pin(stream.map(|r| r.map_err(Into::into)));
-            Ok(stream)
+            Ok(stream.boxed())
         })
         .boxed())
     }

datafusion/datasource-parquet/src/opener.rs

@@ -411,8 +411,8 @@ impl FileOpener for ParquetOpener {
                 .build()?;
 
             let stream = stream
-                .map_err(|e| DataFusionError::from(e))
-                .map(move |b| b.and_then(|b| Ok(schema_mapping.map_batch(b)?)));
+                .map_err(DataFusionError::from)
+                .map(move |b| b.and_then(|b| schema_mapping.map_batch(b)));
 
             if let Some(file_pruner) = file_pruner {
                 Ok(EarlyStoppingStream::new(

datafusion/datasource/src/file_stream.rs

@@ -37,7 +37,6 @@ use datafusion_physical_plan::metrics::{
     BaselineMetrics, Count, ExecutionPlanMetricsSet, MetricBuilder, Time,
 };
 
-use arrow::error::ArrowError;
 use arrow::record_batch::RecordBatch;
 use datafusion_common::instant::Instant;
 use datafusion_common::{DataFusionError, ScalarValue};
@@ -225,7 +224,6 @@ impl FileStream {
                             let result = self
                                 .pc_projector
                                 .project(batch, partition_values)
-                                .map_err(|e| ArrowError::ExternalError(e.into()))
                                 .map(|batch| match &mut self.remain {
                                     Some(remain) => {
                                         if *remain > batch.num_rows() {
@@ -247,7 +245,7 @@ impl FileStream {
                                 self.state = FileStreamState::Error
                             }
                             self.file_stream_metrics.time_scanning_total.start();
-                            return Poll::Ready(Some(result.map_err(Into::into)));
+                            return Poll::Ready(Some(result));
                         }
                         Some(Err(err)) => {
                             self.file_stream_metrics.file_scan_errors.add(1);
@@ -281,7 +279,7 @@ impl FileStream {
                                 },
                                 OnError::Fail => {
                                     self.state = FileStreamState::Error;
-                                    return Poll::Ready(Some(Err(err.into())));
+                                    return Poll::Ready(Some(Err(err)));
                                 }
                             }
                         }
@@ -526,7 +524,6 @@ mod tests {
     use crate::file_scan_config::FileScanConfigBuilder;
     use crate::tests::make_partition;
     use crate::PartitionedFile;
-    use arrow::error::ArrowError;
     use datafusion_common::error::Result;
     use datafusion_execution::object_store::ObjectStoreUrl;
     use datafusion_physical_plan::metrics::ExecutionPlanMetricsSet;
@@ -540,7 +537,7 @@ mod tests {
     use arrow::array::RecordBatch;
     use arrow::datatypes::Schema;
 
-    use datafusion_common::{assert_batches_eq, internal_err};
+    use datafusion_common::{assert_batches_eq, internal_err, DataFusionError};
 
     /// Test `FileOpener` which will simulate errors during file opening or scanning
     #[derive(Default)]
@@ -566,10 +563,9 @@ mod tests {
             if self.error_opening_idx.contains(&idx) {
                 Ok(futures::future::ready(internal_err!("error opening")).boxed())
             } else if self.error_scanning_idx.contains(&idx) {
-                let error = futures::future::ready(Err(ArrowError::IpcError(
+                let error = futures::future::ready(Err(DataFusionError::Execution(
                     "error scanning".to_owned(),
-                )
-                .into()));
+                )));
                 let stream = futures::stream::once(error).boxed();
                 Ok(futures::future::ready(Ok(stream)).boxed())
             } else {