Fix: Avoid recursive external error wrapping

datafusion/common/src/error.rs

@@ -131,6 +131,10 @@ pub enum DataFusionError {
     /// Errors from either mapping LogicalPlans to/from Substrait plans
     /// or serializing/deserializing protobytes to Substrait plans
     Substrait(String),
+
+    /// Errors for wrapping other errors.
+    /// This is useful when we need to share DatafusionError.
+    WrappedError(Arc<DataFusionError>),
 }
 
 #[macro_export]
@@ -293,7 +297,16 @@ impl From<ParserError> for DataFusionError {
 
 impl From<GenericError> for DataFusionError {
     fn from(err: GenericError) -> Self {
-        DataFusionError::External(err)
+        // If the error is already a DataFusionError, not wrapping it.
+        if err.is::<DataFusionError>() {
+            if let Ok(e) = err.downcast::<DataFusionError>() {
+                *e
+            } else {
+                unreachable!()
+            }
+        } else {
+            DataFusionError::External(err)
+        }
     }
 }
 
@@ -328,6 +341,7 @@ impl Error for DataFusionError {
             DataFusionError::External(e) => Some(e.as_ref()),
             DataFusionError::Context(_, e) => Some(e.as_ref()),
             DataFusionError::Substrait(_) => None,
+            DataFusionError::WrappedError(e) => Some(e.as_ref()),
         }
     }
 }
@@ -441,6 +455,7 @@ impl DataFusionError {
             DataFusionError::External(_) => "External error: ",
             DataFusionError::Context(_, _) => "",
             DataFusionError::Substrait(_) => "Substrait error: ",
+            DataFusionError::WrappedError(_) => "",
         }
     }
 
@@ -481,6 +496,7 @@ impl DataFusionError {
                 Cow::Owned(format!("{desc}\ncaused by\n{}", *err))
             }
             DataFusionError::Substrait(ref desc) => Cow::Owned(desc.to_string()),
+            DataFusionError::WrappedError(ref desc) => Cow::Owned(desc.to_string()),
         }
     }
 }
@@ -656,7 +672,7 @@ pub fn unqualified_field_not_found(name: &str, schema: &DFSchema) -> DataFusionE
 mod test {
     use std::sync::Arc;
 
-    use crate::error::DataFusionError;
+    use crate::error::{DataFusionError, GenericError};
     use arrow::error::ArrowError;
 
     #[test]
@@ -810,6 +826,43 @@ mod test {
         );
     }
 
+    #[test]
+    fn external_error() {
+        // assert not wrapping DataFusionError
+        let generic_error: GenericError =
+            Box::new(DataFusionError::Plan("test".to_string()));
+        let datafusion_error: DataFusionError = generic_error.into();
+        println!("{}", datafusion_error.strip_backtrace());
+        assert_eq!(
+            datafusion_error.strip_backtrace(),
+            "Error during planning: test"
+        );
+
+        // assert wrapping other Error
+        let generic_error: GenericError =
+            Box::new(std::io::Error::new(std::io::ErrorKind::Other, "io error"));
+        let datafusion_error: DataFusionError = generic_error.into();
+        println!("{}", datafusion_error.strip_backtrace());
+        assert_eq!(
+            datafusion_error.strip_backtrace(),
+            "External error: io error"
+        );
+    }
+
+    #[test]
+    fn external_error_no_recursive() {
+        let generic_error_1: GenericError =
+            Box::new(std::io::Error::new(std::io::ErrorKind::Other, "io error"));
+        let external_error_1: DataFusionError = generic_error_1.into();
+        let generic_error_2: GenericError = Box::new(external_error_1);
+        let external_error_2: DataFusionError = generic_error_2.into();
+
+        println!("{}", external_error_2);
+        assert!(external_error_2
+            .to_string()
+            .starts_with("External error: io error"));
+    }
+
     /// Model what happens when implementing SendableRecordBatchStream:
     /// DataFusion code needs to return an ArrowError
     fn return_arrow_error() -> arrow::error::Result<()> {

datafusion/physical-plan/src/joins/cross_join.rs

@@ -867,7 +867,7 @@ mod tests {
 
         assert_contains!(
             err.to_string(),
-            "External error: Resources exhausted: Additional allocation failed with top memory consumers (across reservations) as: CrossJoinExec"
+            "Resources exhausted: Additional allocation failed with top memory consumers (across reservations) as: CrossJoinExec"
         );
 
         Ok(())

datafusion/physical-plan/src/joins/hash_join.rs

@@ -4014,7 +4014,7 @@ mod tests {
             // Asserting that operator-level reservation attempting to overallocate
             assert_contains!(
                 err.to_string(),
-                "External error: Resources exhausted: Additional allocation failed with top memory consumers (across reservations) as: HashJoinInput"
+                "Resources exhausted: Additional allocation failed with top memory consumers (across reservations) as: HashJoinInput"
             );
 
             assert_contains!(
@@ -4099,7 +4099,7 @@ mod tests {
             // Asserting that stream-level reservation attempting to overallocate
             assert_contains!(
                 err.to_string(),
-                "External error: Resources exhausted: Additional allocation failed with top memory consumers (across reservations) as: HashJoinInput[1]"
+                "Resources exhausted: Additional allocation failed with top memory consumers (across reservations) as: HashJoinInput[1]"
 
             );
 

datafusion/physical-plan/src/joins/nested_loop_join.rs

@@ -1513,7 +1513,7 @@ pub(crate) mod tests {
 
             assert_contains!(
                 err.to_string(),
-                "External error: Resources exhausted: Additional allocation failed with top memory consumers (across reservations) as: NestedLoopJoinLoad[0]"
+                "Resources exhausted: Additional allocation failed with top memory consumers (across reservations) as: NestedLoopJoinLoad[0]"
             );
         }
 

datafusion/physical-plan/src/joins/utils.rs

@@ -1076,7 +1076,7 @@ impl<T: 'static> OnceFut<T> {
             OnceFutState::Ready(r) => Poll::Ready(
                 r.as_ref()
                     .map(|r| r.as_ref())
-                    .map_err(|e| DataFusionError::External(Box::new(Arc::clone(e)))),
+                    .map_err(|e| DataFusionError::WrappedError(Arc::clone(e))),
             ),
         }
     }
@@ -1090,10 +1090,9 @@ impl<T: 'static> OnceFut<T> {
 
         match &self.state {
             OnceFutState::Pending(_) => unreachable!(),
-            OnceFutState::Ready(r) => Poll::Ready(
-                r.clone()
-                    .map_err(|e| DataFusionError::External(Box::new(e))),
-            ),
+            OnceFutState::Ready(r) => {
+                Poll::Ready(r.clone().map_err(DataFusionError::WrappedError))
+            }
         }
     }
 }

datafusion/physical-plan/src/repartition/mod.rs

@@ -916,7 +916,7 @@ impl RepartitionExec {
 
                 for (_, tx) in txs {
                     // wrap it because need to send error to all output partitions
-                    let err = Err(DataFusionError::External(Box::new(Arc::clone(&e))));
+                    let err = Err(DataFusionError::WrappedError(Arc::clone(&e)));
                     tx.send(Some(err)).await.ok();
                 }
             }

datafusion/sqllogictest/test_files/aggregate.slt

@@ -86,7 +86,7 @@ statement error DataFusion error: Error during planning: Failed to coerce argume
 SELECT approx_percentile_cont_with_weight(c3, c2, c1) FROM aggregate_test_100
 
 # csv_query_approx_percentile_cont_with_histogram_bins
-statement error DataFusion error: External error: This feature is not implemented: Tdigest max_size value for 'APPROX_PERCENTILE_CONT' must be UInt > 0 literal \(got data type Int64\)\.
+statement error DataFusion error: This feature is not implemented: Tdigest max_size value for 'APPROX_PERCENTILE_CONT' must be UInt > 0 literal \(got data type Int64\)\.
 SELECT c1, approx_percentile_cont(c3, 0.95, -1000) AS c3_p95 FROM aggregate_test_100 GROUP BY 1 ORDER BY 1
 
 statement error DataFusion error: Error during planning: Failed to coerce arguments to satisfy a call to approx_percentile_cont function: coercion from \[Int16, Float64, Utf8\] to the signature OneOf(.*) failed(.|\n)*

datafusion/sqllogictest/test_files/errors.slt

@@ -161,3 +161,12 @@ create table records (timestamp timestamp, value float) as values (
     '2021-01-01 00:00:00', 1.0,
     '2021-01-01 00:00:00', 2.0
 );
+
+statement ok
+CREATE TABLE tab0(col0 INTEGER, col1 INTEGER, col2 INTEGER);
+
+statement ok
+INSERT INTO tab0 VALUES(83,0,38);
+
+query error DataFusion error: Arrow error: Divide by zero error
+SELECT DISTINCT - 84 FROM tab0 AS cor0 WHERE NOT + 96 / + col1 <= NULL GROUP BY col1, col0;