Fix error formatting

Author: alamb

datafusion/datasource/Cargo.toml

@@ -31,6 +31,7 @@ version.workspace = true
 all-features = true
 
 [features]
+backtrace = ["datafusion-common/backtrace"]
 compression = ["async-compression", "liblzma", "bzip2", "flate2", "zstd", "tokio-util"]
 default = ["compression"]
 

datafusion/datasource/src/file_stream/mod.rs

@@ -175,6 +175,7 @@ mod tests {
     use crate::{PartitionedFile, TableSchema};
     use arrow::array::{AsArray, RecordBatch};
     use arrow::datatypes::{DataType, Field, Int32Type, Schema};
+    use datafusion_common::DataFusionError;
     use datafusion_common::error::Result;
     use datafusion_execution::object_store::ObjectStoreUrl;
     use datafusion_physical_plan::metrics::ExecutionPlanMetricsSet;
@@ -796,7 +797,7 @@ mod tests {
 
         insta::assert_snapshot!(test.run().await.unwrap(), @r"
         ----- Output Stream -----
-        DataFusionError::Internal(io failed while opening file)
+        Error: io failed while opening file
         Done
         ----- File Stream Events -----
         morselize_file: file1.parquet
@@ -822,7 +823,7 @@ mod tests {
 
         insta::assert_snapshot!(test.run().await.unwrap(), @r"
         ----- Output Stream -----
-        DataFusionError::Internal(planner failed after io)
+        Error: planner failed after io
         Done
         ----- File Stream Events -----
         morselize_file: file1.parquet
@@ -967,22 +968,15 @@ mod tests {
                         stream_contents.push(format!("Batch: {batch_id}"));
                     }
                     Err(e) => {
-                        // Pull the actual message for common types rather than
+                        // Pull the actual message for exteral errors rather than
                         // relying on DataFusionError formatting, which changes
                         // if backtraces are enabled, etc
-                        let message = match e {
-                            datafusion_common::DataFusionError::Execution(msg) => {
-                                format!("DataFusionError::Execution({msg})")
-                            }
-                            datafusion_common::DataFusionError::Internal(msg) => {
-                                format!("DataFusionError::Internal({msg})")
-                            }
-                            datafusion_common::DataFusionError::Plan(msg) => {
-                                format!("DataFusionError::Plan({msg})")
-                            }
-                            other => format!("{other}"),
+                        let message = if let DataFusionError::External(generic) = e {
+                            generic.to_string()
+                        } else {
+                            e.to_string()
                         };
-                        stream_contents.push(message)
+                        stream_contents.push(format!("Error: {message}"));
                     }
                 }
             }

datafusion/datasource/src/morsel/mocks.rs

@@ -27,7 +27,7 @@ use crate::PartitionedFile;
 use crate::morsel::{Morsel, MorselPlan, MorselPlanner, Morselizer};
 use arrow::array::{Int32Array, RecordBatch};
 use arrow::datatypes::{DataType, Field, Schema};
-use datafusion_common::{Result, internal_datafusion_err};
+use datafusion_common::{DataFusionError, Result, internal_datafusion_err};
 use futures::Future;
 use futures::future::BoxFuture;
 use futures::stream::BoxStream;
@@ -54,6 +54,18 @@ pub(crate) struct PollsToResolve(pub usize);
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub(crate) struct MockError(pub String);
 
+impl Display for MockError {
+    fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
+        write!(f, "{}", self.0)
+    }
+}
+
+impl std::error::Error for MockError {
+    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
+        None
+    }
+}
+
 /// Scheduler-visible event captured by the mock morsel test harness.
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub(crate) enum MorselEvent {
@@ -216,7 +228,7 @@ enum PlannerStep {
         result: std::result::Result<(), MockError>,
     },
     Error {
-        message: String,
+        error: MockError,
     },
     None,
 }
@@ -289,7 +301,7 @@ impl MockPlannerBuilder {
     /// Adds one planning step that fails during CPU planning.
     pub(crate) fn return_error(mut self, message: impl Into<String>) -> Self {
         self.steps.push(PlannerStep::Error {
-            message: message.into(),
+            error: MockError(message.into()),
         });
         self
     }
@@ -413,7 +425,9 @@ impl MorselPlanner for MockMorselPlanner {
                     )) as BoxFuture<'static, Result<()>>,
                 )))
             }
-            PlannerStep::Error { message } => Err(internal_datafusion_err!("{message}")),
+            PlannerStep::Error { error } => {
+                Err(DataFusionError::External(Box::new(error)))
+            }
             PlannerStep::None => Ok(None),
         }
     }
@@ -537,13 +551,13 @@ impl Future for MockIoFuture {
                 });
                 Poll::Ready(Ok(()))
             }
-            Err(MockError(message)) => {
+            Err(e) => {
                 self.observer.push(MorselEvent::IoFutureErrored {
                     planner_name: self.planner_name.clone(),
                     io_future_id: self.io_future_id,
-                    message: message.clone(),
+                    message: e.0.clone(),
                 });
-                Poll::Ready(Err(internal_datafusion_err!("{message}")))
+                Poll::Ready(Err(DataFusionError::External(Box::new(e.clone()))))
             }
         }
     }