Refactor error handling to use boxed errors for DataFusionError variants (apache#16672)

- Refactored the `DataFusionError` enum to use `Box<T>` for:
  - `ArrowError`
  - `ParquetError`
  - `AvroError`
  - `object_store::Error`
  - `ParserError`
  - `SchemaError`
  - `JoinError`
- Updated all relevant match arms and constructors to handle boxed errors.
- Refactored error-related macros (`arrow_datafusion_err!`, `sql_datafusion_err!`, etc.) to use `Box<T>`.
- Adjusted test cases and error assertions for boxed variants.
- Documentation update to the upgrade guide to explain the required changes and rationale.

Author: kosiew

datafusion-cli/src/exec.rs

@@ -330,8 +330,8 @@ impl StatementExecutor {
 
         let df = match ctx.execute_logical_plan(plan).await {
             Ok(df) => Ok(df),
-            Err(DataFusionError::ObjectStore(Generic { store, source: _ }))
-                if "S3".eq_ignore_ascii_case(store)
+            Err(DataFusionError::ObjectStore(err))
+                if matches!(err.as_ref(), Generic { store, source: _ } if "S3".eq_ignore_ascii_case(store))
                     && self.statement_for_retry.is_some() =>
             {
                 warn!("S3 region is incorrect, auto-detecting the correct region (this may be slow). Consider updating your region configuration.");

datafusion-cli/src/object_storage.rs

@@ -15,27 +15,35 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use std::any::Any;
-use std::error::Error;
-use std::fmt::{Debug, Display};
-use std::sync::Arc;
-
-use datafusion::common::config::{
-    ConfigEntry, ConfigExtension, ConfigField, ExtensionOptions, TableOptions, Visit,
-};
-use datafusion::common::{config_err, exec_datafusion_err, exec_err};
-use datafusion::error::{DataFusionError, Result};
-use datafusion::execution::context::SessionState;
-
 use async_trait::async_trait;
 use aws_config::BehaviorVersion;
-use aws_credential_types::provider::error::CredentialsError;
-use aws_credential_types::provider::{ProvideCredentials, SharedCredentialsProvider};
+use aws_credential_types::provider::{
+    error::CredentialsError, ProvideCredentials, SharedCredentialsProvider,
+};
+use datafusion::{
+    common::{
+        config::ConfigEntry, config::ConfigExtension, config::ConfigField,
+        config::ExtensionOptions, config::TableOptions, config::Visit, config_err,
+        exec_datafusion_err, exec_err,
+    },
+    error::{DataFusionError, Result},
+    execution::context::SessionState,
+};
 use log::debug;
-use object_store::aws::{AmazonS3Builder, AmazonS3ConfigKey, AwsCredential};
-use object_store::gcp::GoogleCloudStorageBuilder;
-use object_store::http::HttpBuilder;
-use object_store::{ClientOptions, CredentialProvider, ObjectStore};
+use object_store::{
+    aws::{AmazonS3Builder, AmazonS3ConfigKey, AwsCredential},
+    gcp::GoogleCloudStorageBuilder,
+    http::HttpBuilder,
+    ClientOptions, CredentialProvider,
+    Error::Generic,
+    ObjectStore,
+};
+use std::{
+    any::Any,
+    error::Error,
+    fmt::{Debug, Display},
+    sync::Arc,
+};
 use url::Url;
 
 #[cfg(not(test))]
@@ -153,10 +161,10 @@ impl CredentialsFromConfig {
         let credentials = config
             .credentials_provider()
             .ok_or_else(|| {
-                DataFusionError::ObjectStore(object_store::Error::Generic {
+                DataFusionError::ObjectStore(Box::new(Generic {
                     store: "S3",
                     source: "Failed to get S3 credentials aws_config".into(),
-                })
+                }))
             })?
             .clone();
 
@@ -183,10 +191,10 @@ impl CredentialsFromConfig {
                     "Error getting credentials from provider: {e}{source_message}",
                 );
 
-                return Err(DataFusionError::ObjectStore(object_store::Error::Generic {
+                return Err(DataFusionError::ObjectStore(Box::new(Generic {
                     store: "S3",
                     source: message.into(),
-                }));
+                })));
             }
         };
         Ok(Self {
@@ -206,12 +214,14 @@ impl CredentialProvider for S3CredentialProvider {
     type Credential = AwsCredential;
 
     async fn get_credential(&self) -> object_store::Result<Arc<Self::Credential>> {
-        let creds = self.credentials.provide_credentials().await.map_err(|e| {
-            object_store::Error::Generic {
-                store: "S3",
-                source: Box::new(e),
-            }
-        })?;
+        let creds =
+            self.credentials
+                .provide_credentials()
+                .await
+                .map_err(|e| Generic {
+                    store: "S3",
+                    source: Box::new(e),
+                })?;
         Ok(Arc::new(AwsCredential {
             key_id: creds.access_key_id().to_string(),
             secret_key: creds.secret_access_key().to_string(),

datafusion/catalog/src/stream.rs

@@ -435,6 +435,6 @@ impl DataSink for StreamWrite {
         write_task
             .join_unwind()
             .await
-            .map_err(DataFusionError::ExecutionJoin)?
+            .map_err(|e| DataFusionError::ExecutionJoin(Box::new(e)))?
     }
 }

datafusion/common/src/error.rs

@@ -53,22 +53,22 @@ pub enum DataFusionError {
     /// Error returned by arrow.
     ///
     /// 2nd argument is for optional backtrace
-    ArrowError(ArrowError, Option<String>),
+    ArrowError(Box<ArrowError>, Option<String>),
     /// Error when reading / writing Parquet data.
     #[cfg(feature = "parquet")]
-    ParquetError(ParquetError),
+    ParquetError(Box<ParquetError>),
     /// Error when reading Avro data.
     #[cfg(feature = "avro")]
     AvroError(Box<AvroError>),
     /// Error when reading / writing to / from an object_store (e.g. S3 or LocalFile)
     #[cfg(feature = "object_store")]
-    ObjectStore(object_store::Error),
+    ObjectStore(Box<object_store::Error>),
     /// Error when an I/O operation fails
     IoError(io::Error),
     /// Error when SQL is syntactically incorrect.
     ///
     /// 2nd argument is for optional backtrace
-    SQL(ParserError, Option<String>),
+    SQL(Box<ParserError>, Option<String>),
     /// Error when a feature is not yet implemented.
     ///
     /// These errors are sometimes returned for features that are still in
@@ -107,7 +107,7 @@ pub enum DataFusionError {
     ///
     /// 2nd argument is for optional backtrace
     /// Boxing the optional backtrace to prevent <https://rust-lang.github.io/rust-clippy/master/index.html#/result_large_err>
-    SchemaError(SchemaError, Box<Option<String>>),
+    SchemaError(Box<SchemaError>, Box<Option<String>>),
     /// Error during execution of the query.
     ///
     /// This error is returned when an error happens during execution due to a
@@ -118,7 +118,7 @@ pub enum DataFusionError {
     /// [`JoinError`] during execution of the query.
     ///
     /// This error can't occur for unjoined tasks, such as execution shutdown.
-    ExecutionJoin(JoinError),
+    ExecutionJoin(Box<JoinError>),
     /// Error when resources (such as memory of scratch disk space) are exhausted.
     ///
     /// This error is thrown when a consumer cannot acquire additional memory
@@ -276,14 +276,14 @@ impl From<io::Error> for DataFusionError {
 
 impl From<ArrowError> for DataFusionError {
     fn from(e: ArrowError) -> Self {
-        DataFusionError::ArrowError(e, None)
+        DataFusionError::ArrowError(Box::new(e), None)
     }
 }
 
 impl From<DataFusionError> for ArrowError {
     fn from(e: DataFusionError) -> Self {
         match e {
-            DataFusionError::ArrowError(e, _) => e,
+            DataFusionError::ArrowError(e, _) => *e,
             DataFusionError::External(e) => ArrowError::ExternalError(e),
             other => ArrowError::ExternalError(Box::new(other)),
         }
@@ -304,7 +304,7 @@ impl From<&Arc<DataFusionError>> for DataFusionError {
 #[cfg(feature = "parquet")]
 impl From<ParquetError> for DataFusionError {
     fn from(e: ParquetError) -> Self {
-        DataFusionError::ParquetError(e)
+        DataFusionError::ParquetError(Box::new(e))
     }
 }
 
@@ -318,20 +318,20 @@ impl From<AvroError> for DataFusionError {
 #[cfg(feature = "object_store")]
 impl From<object_store::Error> for DataFusionError {
     fn from(e: object_store::Error) -> Self {
-        DataFusionError::ObjectStore(e)
+        DataFusionError::ObjectStore(Box::new(e))
     }
 }
 
 #[cfg(feature = "object_store")]
 impl From<object_store::path::Error> for DataFusionError {
     fn from(e: object_store::path::Error) -> Self {
-        DataFusionError::ObjectStore(e.into())
+        DataFusionError::ObjectStore(Box::new(e.into()))
     }
 }
 
 impl From<ParserError> for DataFusionError {
     fn from(e: ParserError) -> Self {
-        DataFusionError::SQL(e, None)
+        DataFusionError::SQL(Box::new(e), None)
     }
 }
 
@@ -361,22 +361,22 @@ impl Display for DataFusionError {
 impl Error for DataFusionError {
     fn source(&self) -> Option<&(dyn Error + 'static)> {
         match self {
-            DataFusionError::ArrowError(e, _) => Some(e),
+            DataFusionError::ArrowError(e, _) => Some(e.as_ref()),
             #[cfg(feature = "parquet")]
-            DataFusionError::ParquetError(e) => Some(e),
+            DataFusionError::ParquetError(e) => Some(e.as_ref()),
             #[cfg(feature = "avro")]
-            DataFusionError::AvroError(e) => Some(e),
+            DataFusionError::AvroError(e) => Some(e.as_ref()),
             #[cfg(feature = "object_store")]
-            DataFusionError::ObjectStore(e) => Some(e),
+            DataFusionError::ObjectStore(e) => Some(e.as_ref()),
             DataFusionError::IoError(e) => Some(e),
-            DataFusionError::SQL(e, _) => Some(e),
+            DataFusionError::SQL(e, _) => Some(e.as_ref()),
             DataFusionError::NotImplemented(_) => None,
             DataFusionError::Internal(_) => None,
             DataFusionError::Configuration(_) => None,
             DataFusionError::Plan(_) => None,
-            DataFusionError::SchemaError(e, _) => Some(e),
+            DataFusionError::SchemaError(e, _) => Some(e.as_ref()),
             DataFusionError::Execution(_) => None,
-            DataFusionError::ExecutionJoin(e) => Some(e),
+            DataFusionError::ExecutionJoin(e) => Some(e.as_ref()),
             DataFusionError::ResourcesExhausted(_) => None,
             DataFusionError::External(e) => Some(e.as_ref()),
             DataFusionError::Context(_, e) => Some(e.as_ref()),
@@ -828,7 +828,7 @@ make_error!(resources_err, resources_datafusion_err, ResourcesExhausted);
 #[macro_export]
 macro_rules! sql_datafusion_err {
     ($ERR:expr $(; diagnostic = $DIAG:expr)?) => {{
-        let err = DataFusionError::SQL($ERR, Some(DataFusionError::get_back_trace()));
+        let err = DataFusionError::SQL(Box::new($ERR), Some(DataFusionError::get_back_trace()));
         $(
             let err = err.with_diagnostic($DIAG);
         )?
@@ -852,7 +852,7 @@ macro_rules! sql_err {
 #[macro_export]
 macro_rules! arrow_datafusion_err {
     ($ERR:expr $(; diagnostic = $DIAG:expr)?) => {{
-        let err = DataFusionError::ArrowError($ERR, Some(DataFusionError::get_back_trace()));
+        let err = DataFusionError::ArrowError(Box::new($ERR), Some(DataFusionError::get_back_trace()));
         $(
             let err = err.with_diagnostic($DIAG);
         )?
@@ -878,7 +878,7 @@ macro_rules! arrow_err {
 macro_rules! schema_datafusion_err {
     ($ERR:expr $(; diagnostic = $DIAG:expr)?) => {{
         let err = $crate::error::DataFusionError::SchemaError(
-            $ERR,
+            Box::new($ERR),
             Box::new(Some($crate::error::DataFusionError::get_back_trace())),
         );
         $(
@@ -893,7 +893,7 @@ macro_rules! schema_datafusion_err {
 macro_rules! schema_err {
     ($ERR:expr $(; diagnostic = $DIAG:expr)?) => {{
         let err = $crate::error::DataFusionError::SchemaError(
-            $ERR,
+            Box::new($ERR),
             Box::new(Some($crate::error::DataFusionError::get_back_trace())),
         );
         $(
@@ -951,11 +951,17 @@ pub fn add_possible_columns_to_diag(
 
 #[cfg(test)]
 mod test {
+    use std::mem::size_of;
     use std::sync::Arc;
 
     use crate::error::{DataFusionError, GenericError};
     use arrow::error::ArrowError;
 
+    #[test]
+    fn test_datafusion_error_size() {
+        assert_eq!(size_of::<DataFusionError>(), 40);
+    }
+
     #[test]
     fn datafusion_error_to_arrow() {
         let res = return_arrow_error().unwrap_err();
@@ -1020,8 +1026,8 @@ mod test {
 
         do_root_test(
             DataFusionError::ArrowError(
-                ArrowError::ExternalError(Box::new(DataFusionError::ResourcesExhausted(
-                    "foo".to_string(),
+                Box::new(ArrowError::ExternalError(Box::new(
+                    DataFusionError::ResourcesExhausted("foo".to_string()),
                 ))),
                 None,
             ),
@@ -1044,9 +1050,11 @@ mod test {
 
         do_root_test(
             DataFusionError::ArrowError(
-                ArrowError::ExternalError(Box::new(ArrowError::ExternalError(Box::new(
-                    DataFusionError::ResourcesExhausted("foo".to_string()),
-                )))),
+                Box::new(ArrowError::ExternalError(Box::new(
+                    ArrowError::ExternalError(Box::new(
+                        DataFusionError::ResourcesExhausted("foo".to_string()),
+                    )),
+                ))),
                 None,
             ),
             DataFusionError::ResourcesExhausted("foo".to_string()),

datafusion/common/src/scalar/mod.rs

@@ -2640,7 +2640,7 @@ impl ScalarValue {
                         value_offsets,
                         child_arrays,
                     )
-                    .map_err(|e| DataFusionError::ArrowError(e, None))?;
+                    .map_err(|e| DataFusionError::ArrowError(Box::new(e), None))?;
                     Arc::new(ar)
                 }
                 None => {
@@ -6679,10 +6679,7 @@ mod tests {
                     let err = value.arithmetic_negate().expect_err("Should receive overflow error on negating {value:?}");
                     let root_err = err.find_root();
                     match  root_err{
-                        DataFusionError::ArrowError(
-                            ArrowError::ArithmeticOverflow(_),
-                            _,
-                        ) => {}
+                        DataFusionError::ArrowError(err, _) if matches!(err.as_ref(), ArrowError::ArithmeticOverflow(_)) => {}
                         _ => return Err(err),
                     };
                 }

datafusion/core/src/dataframe/mod.rs

@@ -2036,10 +2036,11 @@ impl DataFrame {
             match self.plan.schema().qualified_field_from_column(&old_column) {
                 Ok(qualifier_and_field) => qualifier_and_field,
                 // no-op if field not found
-                Err(DataFusionError::SchemaError(
-                    SchemaError::FieldNotFound { .. },
-                    _,
-                )) => return Ok(self),
+                Err(DataFusionError::SchemaError(e, _))
+                    if matches!(*e, SchemaError::FieldNotFound { .. }) =>
+                {
+                    return Ok(self);
+                }
                 Err(err) => return Err(err),
             };
         let projection = self

datafusion/core/src/datasource/file_format/arrow.rs

@@ -294,7 +294,7 @@ impl FileSink for ArrowFileSink {
         demux_task
             .join_unwind()
             .await
-            .map_err(DataFusionError::ExecutionJoin)??;
+            .map_err(|e| DataFusionError::ExecutionJoin(Box::new(e)))??;
         Ok(row_count as u64)
     }
 }

datafusion/core/src/datasource/memory_test.rs

@@ -130,12 +130,15 @@ mod tests {
             .scan(&session_ctx.state(), Some(&projection), &[], None)
             .await
         {
-            Err(DataFusionError::ArrowError(ArrowError::SchemaError(e), _)) => {
-                assert_eq!(
-                    "\"project index 4 out of bounds, max field 3\"",
-                    format!("{e:?}")
-                )
-            }
+            Err(DataFusionError::ArrowError(err, _)) => match err.as_ref() {
+                ArrowError::SchemaError(e) => {
+                    assert_eq!(
+                        "\"project index 4 out of bounds, max field 3\"",
+                        format!("{e:?}")
+                    )
+                }
+                _ => panic!("unexpected error"),
+            },
             res => panic!("Scan should failed on invalid projection, got {res:?}"),
         };