Use DataFusionError instead of ArrowError in FileOpenFuture (#17397)

adriangb · web-flow · commit 3d5863b05950 · 2025-09-04T13:23:09.000-05:00
diff --git a/datafusion/core/src/datasource/physical_plan/arrow_file.rs b/datafusion/core/src/datasource/physical_plan/arrow_file.rs
@@ -26,7 +26,7 @@ use datafusion_datasource::schema_adapter::SchemaAdapterFactory;
 use arrow::buffer::Buffer;
 use arrow::datatypes::SchemaRef;
 use arrow_ipc::reader::FileDecoder;
-use datafusion_common::Statistics;
+use datafusion_common::{exec_datafusion_err, Statistics};
 use datafusion_datasource::file::FileSource;
 use datafusion_datasource::file_scan_config::FileScanConfig;
 use datafusion_datasource::PartitionedFile;
@@ -140,15 +140,19 @@ impl FileOpener for ArrowOpener {
                             let arrow_reader = arrow::ipc::reader::FileReader::try_new(
                                 file, projection,
                             )?;
-                            Ok(futures::stream::iter(arrow_reader).boxed())
+                            Ok(futures::stream::iter(arrow_reader)
+                                .map(|r| r.map_err(Into::into))
+                                .boxed())
                         }
                         GetResultPayload::Stream(_) => {
                             let bytes = r.bytes().await?;
                             let cursor = std::io::Cursor::new(bytes);
                             let arrow_reader = arrow::ipc::reader::FileReader::try_new(
                                 cursor, projection,
                             )?;
-                            Ok(futures::stream::iter(arrow_reader).boxed())
+                            Ok(futures::stream::iter(arrow_reader)
+                                .map(|r| r.map_err(Into::into))
+                                .boxed())
                         }
                     }
                 }
@@ -179,9 +183,7 @@ impl FileOpener for ArrowOpener {
                         footer_buf[..footer_len].try_into().unwrap(),
                     )
                     .map_err(|err| {
-                        arrow::error::ArrowError::ParseError(format!(
-                            "Unable to get root as footer: {err:?}"
-                        ))
+                        exec_datafusion_err!("Unable to get root as footer: {err:?}")
                     })?;
                     // build decoder according to footer & projection
                     let schema =
@@ -248,6 +250,7 @@ impl FileOpener for ArrowOpener {
                                     .transpose()
                             }),
                     )
+                    .map(|r| r.map_err(Into::into))
                     .boxed())
                 }
             }
diff --git a/datafusion/core/tests/physical_optimizer/filter_pushdown/util.rs b/datafusion/core/tests/physical_optimizer/filter_pushdown/util.rs
@@ -16,7 +16,6 @@
 // under the License.
 
 use arrow::datatypes::SchemaRef;
-use arrow::error::ArrowError;
 use arrow::{array::RecordBatch, compute::concat_batches};
 use datafusion::{datasource::object_store::ObjectStoreUrl, physical_plan::PhysicalExpr};
 use datafusion_common::{config::ConfigOptions, internal_err, Result, Statistics};
@@ -39,7 +38,7 @@ use datafusion_physical_plan::{
     metrics::ExecutionPlanMetricsSet,
     DisplayAs, DisplayFormatType, ExecutionPlan, PlanProperties,
 };
-use futures::stream::BoxStream;
+use futures::StreamExt;
 use futures::{FutureExt, Stream};
 use object_store::ObjectStore;
 use std::{
@@ -93,12 +92,7 @@ impl FileOpener for TestOpener {
 
         let stream = TestStream::new(batches);
 
-        Ok((async {
-            let stream: BoxStream<'static, Result<RecordBatch, ArrowError>> =
-                Box::pin(stream);
-            Ok(stream)
-        })
-        .boxed())
+        Ok((async { Ok(stream.boxed()) }).boxed())
     }
 }
 
@@ -344,7 +338,7 @@ impl TestStream {
 }
 
 impl Stream for TestStream {
-    type Item = Result<RecordBatch, ArrowError>;
+    type Item = Result<RecordBatch>;
 
     fn poll_next(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Option<Self::Item>> {
         let next_batch = self.index.value();
diff --git a/datafusion/datasource-avro/src/source.rs b/datafusion/datasource-avro/src/source.rs
@@ -169,12 +169,16 @@ mod private {
                 match r.payload {
                     GetResultPayload::File(file, _) => {
                         let reader = config.open(file)?;
-                        Ok(futures::stream::iter(reader).boxed())
+                        Ok(futures::stream::iter(reader)
+                            .map(|r| r.map_err(Into::into))
+                            .boxed())
                     }
                     GetResultPayload::Stream(_) => {
                         let bytes = r.bytes().await?;
                         let reader = config.open(bytes.reader())?;
-                        Ok(futures::stream::iter(reader).boxed())
+                        Ok(futures::stream::iter(reader)
+                            .map(|r| r.map_err(Into::into))
+                            .boxed())
                     }
                 }
             }))
diff --git a/datafusion/datasource-csv/src/source.rs b/datafusion/datasource-csv/src/source.rs
@@ -392,17 +392,20 @@ impl FileOpener for CsvOpener {
                         )?
                     };
 
-                    Ok(futures::stream::iter(config.open(decoder)?).boxed())
+                    Ok(futures::stream::iter(config.open(decoder)?)
+                        .map(|r| r.map_err(Into::into))
+                        .boxed())
                 }
                 GetResultPayload::Stream(s) => {
                     let decoder = config.builder().build_decoder();
                     let s = s.map_err(DataFusionError::from);
                     let input = file_compression_type.convert_stream(s.boxed())?.fuse();
 
-                    Ok(deserialize_stream(
+                    let stream = deserialize_stream(
                         input,
                         DecoderDeserializer::new(CsvDecoder::new(decoder)),
-                    ))
+                    );
+                    Ok(stream.map_err(Into::into).boxed())
                 }
             }
         }))
diff --git a/datafusion/datasource-json/src/source.rs b/datafusion/datasource-json/src/source.rs
@@ -222,7 +222,9 @@ impl FileOpener for JsonOpener {
                         .with_batch_size(batch_size)
                         .build(BufReader::new(bytes))?;
 
-                    Ok(futures::stream::iter(reader).boxed())
+                    Ok(futures::stream::iter(reader)
+                        .map(|r| r.map_err(Into::into))
+                        .boxed())
                 }
                 GetResultPayload::Stream(s) => {
                     let s = s.map_err(DataFusionError::from);
@@ -232,10 +234,11 @@ impl FileOpener for JsonOpener {
                         .build_decoder()?;
                     let input = file_compression_type.convert_stream(s.boxed())?.fuse();
 
-                    Ok(deserialize_stream(
+                    let stream = deserialize_stream(
                         input,
                         DecoderDeserializer::new(JsonDecoder::new(decoder)),
-                    ))
+                    );
+                    Ok(stream.map_err(Into::into).boxed())
                 }
             }
         }))
diff --git a/datafusion/datasource-parquet/src/opener.rs b/datafusion/datasource-parquet/src/opener.rs
@@ -32,7 +32,6 @@ use std::sync::Arc;
 use std::task::{Context, Poll};
 
 use arrow::datatypes::{FieldRef, SchemaRef, TimeUnit};
-use arrow::error::ArrowError;
 use datafusion_common::encryption::FileDecryptionProperties;
 
 use datafusion_common::{exec_err, DataFusionError, Result};
@@ -414,8 +413,8 @@ impl FileOpener for ParquetOpener {
                 .build()?;
 
             let stream = stream
-                .map_err(|e| ArrowError::ExternalError(Box::new(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(
@@ -462,12 +461,9 @@ impl<S> EarlyStoppingStream<S> {
 }
 impl<S> EarlyStoppingStream<S>
 where
-    S: Stream<Item = Result<RecordBatch, ArrowError>> + Unpin,
+    S: Stream<Item = Result<RecordBatch>> + Unpin,
 {
-    fn check_prune(
-        &mut self,
-        input: Result<RecordBatch, ArrowError>,
-    ) -> Result<Option<RecordBatch>, ArrowError> {
+    fn check_prune(&mut self, input: Result<RecordBatch>) -> Result<Option<RecordBatch>> {
         let batch = input?;
 
         // Since dynamic filters may have been updated, see if we can stop
@@ -485,9 +481,9 @@ where
 
 impl<S> Stream for EarlyStoppingStream<S>
 where
-    S: Stream<Item = Result<RecordBatch, ArrowError>> + Unpin,
+    S: Stream<Item = Result<RecordBatch>> + Unpin,
 {
-    type Item = Result<RecordBatch, ArrowError>;
+    type Item = Result<RecordBatch>;
 
     fn poll_next(
         mut self: Pin<&mut Self>,
@@ -697,8 +693,8 @@ mod test {
     use bytes::{BufMut, BytesMut};
     use chrono::Utc;
     use datafusion_common::{
-        assert_batches_eq, record_batch, stats::Precision, ColumnStatistics, ScalarValue,
-        Statistics,
+        assert_batches_eq, record_batch, stats::Precision, ColumnStatistics,
+        DataFusionError, ScalarValue, Statistics,
     };
     use datafusion_datasource::{
         file_meta::FileMeta,
@@ -724,12 +720,8 @@ mod test {
     async fn count_batches_and_rows(
         mut stream: std::pin::Pin<
             Box<
-                dyn Stream<
-                        Item = Result<
-                            arrow::array::RecordBatch,
-                            arrow::error::ArrowError,
-                        >,
-                    > + Send,
+                dyn Stream<Item = Result<arrow::array::RecordBatch, DataFusionError>>
+                    + Send,
             >,
         >,
     ) -> (usize, usize) {
@@ -745,12 +737,8 @@ mod test {
     async fn collect_batches(
         mut stream: std::pin::Pin<
             Box<
-                dyn Stream<
-                        Item = Result<
-                            arrow::array::RecordBatch,
-                            arrow::error::ArrowError,
-                        >,
-                    > + Send,
+                dyn Stream<Item = Result<arrow::array::RecordBatch, DataFusionError>>
+                    + Send,
             >,
         >,
     ) -> Vec<arrow::array::RecordBatch> {
diff --git a/datafusion/datasource/src/file_stream.rs b/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::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)));
                                 }
                             }
                         }
@@ -345,7 +343,7 @@ impl RecordBatchStream for FileStream {
 
 /// A fallible future that resolves to a stream of [`RecordBatch`]
 pub type FileOpenFuture =
-    BoxFuture<'static, Result<BoxStream<'static, Result<RecordBatch, ArrowError>>>>;
+    BoxFuture<'static, Result<BoxStream<'static, Result<RecordBatch>>>>;
 
 /// Describes the behavior of the `FileStream` if file opening or scanning fails
 pub enum OnError {
@@ -376,7 +374,7 @@ pub trait FileOpener: Unpin + Send + Sync {
 /// is ready
 pub enum NextOpen {
     Pending(FileOpenFuture),
-    Ready(Result<BoxStream<'static, Result<RecordBatch, ArrowError>>>),
+    Ready(Result<BoxStream<'static, Result<RecordBatch>>>),
 }
 
 pub enum FileStreamState {
@@ -396,7 +394,7 @@ pub enum FileStreamState {
         /// Partitioning column values for the current batch_iter
         partition_values: Vec<ScalarValue>,
         /// The reader instance
-        reader: BoxStream<'static, Result<RecordBatch, ArrowError>>,
+        reader: BoxStream<'static, Result<RecordBatch>>,
         /// A [`FileOpenFuture`] for the next file to be processed,
         /// and its corresponding partition column values, if any.
         /// This allows the next file to be opened in parallel while the
@@ -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, exec_err, internal_err};
 
     /// Test `FileOpener` which will simulate errors during file opening or scanning
     #[derive(Default)]
@@ -566,9 +563,7 @@ 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(
-                    "error scanning".to_owned(),
-                )));
+                let error = futures::future::ready(exec_err!("error scanning"));
                 let stream = futures::stream::once(error).boxed();
                 Ok(futures::future::ready(Ok(stream)).boxed())
             } else {
diff --git a/docs/source/library-user-guide/upgrading.md b/docs/source/library-user-guide/upgrading.md
@@ -265,6 +265,26 @@ Reimplementation for any custom `DataSource` should be relatively straightforwar
 
 [#17395]: https://github.com/apache/datafusion/pull/17395/
 
+### `FileOpenFuture` now uses `DataFusionError` instead of `ArrowError`
+
+The `FileOpenFuture` type alias has been updated to use `DataFusionError` instead of `ArrowError` for its error type. This change affects the `FileOpener` trait and any implementations that work with file streaming operations.
+
+**Before:**
+
+```rust,ignore
+pub type FileOpenFuture = BoxFuture<'static, Result<BoxStream<'static, Result<RecordBatch, ArrowError>>>>;
+```
+
+**After:**
+
+```rust,ignore
+pub type FileOpenFuture = BoxFuture<'static, Result<BoxStream<'static, Result<RecordBatch>>>>;
+```
+
+If you have custom implementations of `FileOpener` or work directly with `FileOpenFuture`, you'll need to update your error handling to use `DataFusionError` instead of `ArrowError`. The `FileStreamState` enum's `Open` variant has also been updated accordingly. See [#17397] for more details.
+
+[#17397]: https://github.com/apache/datafusion/pull/17397
+
 ## DataFusion `49.0.0`
 
 ### `MSRV` updated to 1.85.1

Original file line number	Diff line number	Diff line change
`@@ -169,12 +169,16 @@ mod private {`
`169`	`169`	`match r.payload {`
`170`	`170`	`GetResultPayload::File(file, _) => {`
`171`	`171`	`let reader = config.open(file)?;`
`172`		`- Ok(futures::stream::iter(reader).boxed())`
	`172`	`+ Ok(futures::stream::iter(reader)`
	`173`	`+ .map(\|r\| r.map_err(Into::into))`
	`174`	`+ .boxed())`
`173`	`175`	`}`
`174`	`176`	`GetResultPayload::Stream(_) => {`
`175`	`177`	`let bytes = r.bytes().await?;`
`176`	`178`	`let reader = config.open(bytes.reader())?;`
`177`		`- Ok(futures::stream::iter(reader).boxed())`
	`179`	`+ Ok(futures::stream::iter(reader)`
	`180`	`+ .map(\|r\| r.map_err(Into::into))`
	`181`	`+ .boxed())`
`178`	`182`	`}`
`179`	`183`	`}`
`180`	`184`	`}))`