feat: refactor error handling to use new external error macros and improve error messages

Author: Cheng-Yuan-Lai

datafusion-cli/src/main.rs

@@ -21,7 +21,7 @@ use std::path::Path;
 use std::process::ExitCode;
 use std::sync::{Arc, LazyLock};
 
-use datafusion::error::{DataFusionError, Result};
+use datafusion::error::Result;
 use datafusion::execution::context::SessionConfig;
 use datafusion::execution::memory_pool::{FairSpillPool, GreedyMemoryPool, MemoryPool};
 use datafusion::execution::runtime_env::RuntimeEnvBuilder;
@@ -381,6 +381,7 @@ pub fn extract_disk_limit(size: &str) -> Result<usize, String> {
 mod tests {
     use super::*;
     use datafusion::common::test_util::batches_to_string;
+    use datafusion::common::DataFusionError;
     use insta::assert_snapshot;
 
     fn assert_conversion(input: &str, expected: Result<usize, String>) {

datafusion-cli/src/print_options.rs

@@ -24,8 +24,8 @@ use crate::print_format::PrintFormat;
 
 use arrow::datatypes::SchemaRef;
 use arrow::record_batch::RecordBatch;
+use datafusion::common::external_err;
 use datafusion::common::instant::Instant;
-use datafusion::common::{external_err, DataFusionError};
 use datafusion::error::Result;
 use datafusion::physical_plan::RecordBatchStream;
 
@@ -143,9 +143,11 @@ impl PrintOptions {
         format_options: &FormatOptions,
     ) -> Result<()> {
         if self.format == PrintFormat::Table {
-            return external_err!("PrintFormat::Table is not implemented"
-                .to_string()
-                .into());
+            let err = std::io::Error::new(
+                std::io::ErrorKind::Other,
+                "PrintFormat::Table is not implemented",
+            );
+            return external_err!(err);
         };
 
         let stdout = std::io::stdout();

datafusion-examples/examples/sql_query.rs

@@ -24,7 +24,7 @@ use datafusion::common::{
 use datafusion::datasource::file_format::parquet::ParquetFormat;
 use datafusion::datasource::listing::ListingOptions;
 use datafusion::datasource::MemTable;
-use datafusion::error::{DataFusionError, Result};
+use datafusion::error::Result;
 use datafusion::prelude::*;
 use object_store::local::LocalFileSystem;
 use std::path::Path;

datafusion/common/src/error.rs

@@ -923,7 +923,10 @@ macro_rules! schema_err {
 #[macro_export]
 macro_rules! external_datafusion_err {
     ($ERR:expr $(; diagnostic = $DIAG:expr)?) => {{
-        let err = DataFusionError::External(Box::new($ERR), Some(DataFusionError::get_back_trace()));
+        let err = $crate::error::DataFusionError::External(
+            Box::new($ERR),
+            Some($crate::error::DataFusionError::get_back_trace())
+        );
         $(
             let err = err.with_diagnostic($DIAG);
         )?
@@ -935,7 +938,7 @@ macro_rules! external_datafusion_err {
 #[macro_export]
 macro_rules! external_err {
     ($ERR:expr $(; diagnostic = $DIAG:expr)?) => {{
-        let err = datafusion_common::external_datafusion_err!($ERR);
+        let err = $crate::external_datafusion_err!($ERR);
         $(
             let err = err.with_diagnostic($DIAG);
         )?
@@ -947,7 +950,10 @@ macro_rules! external_err {
 #[macro_export]
 macro_rules! execution_join_datafusion_err {
     ($ERR:expr $(; diagnostic = $DIAG:expr)?) => {{
-        let err = DataFusionError::ExecutionJoin($ERR, Some(DataFusionError::get_back_trace()));
+        let err = $crate::error::DataFusionError::ExecutionJoin(
+            $ERR,
+            Some($crate::error::DataFusionError::get_back_trace())
+        );
         $(
             let err = err.with_diagnostic($DIAG);
         )?
@@ -959,7 +965,7 @@ macro_rules! execution_join_datafusion_err {
 #[macro_export]
 macro_rules! execution_join_err {
     ($ERR:expr $(; diagnostic = $DIAG:expr)?) => {{
-        let err = datafusion_common::execution_join_datafusion_err!($ERR);
+        let err = $crate::execution_join_datafusion_err!($ERR);
         $(
             let err = err.with_diagnostic($DIAG);
         )?

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

@@ -44,7 +44,8 @@ use arrow::ipc::{root_as_message, CompressionType};
 use datafusion_catalog::Session;
 use datafusion_common::parsers::CompressionTypeVariant;
 use datafusion_common::{
-    not_impl_err, DataFusionError, GetExt, Statistics, DEFAULT_ARROW_EXTENSION,
+    execution_join_datafusion_err, not_impl_err, DataFusionError, GetExt, Statistics,
+    DEFAULT_ARROW_EXTENSION,
 };
 use datafusion_common_runtime::{JoinSet, SpawnedTask};
 use datafusion_datasource::display::FileGroupDisplay;
@@ -294,7 +295,7 @@ impl FileSink for ArrowFileSink {
         demux_task
             .join_unwind()
             .await
-            .map_err(DataFusionError::ExecutionJoin)??;
+            .map_err(|e| execution_join_datafusion_err!(e))??;
         Ok(row_count as u64)
     }
 }

datafusion/core/src/test/mod.rs

@@ -38,7 +38,7 @@ use crate::test_util::{aggr_test_schema, arrow_test_data};
 use arrow::array::{self, Array, ArrayRef, Decimal128Builder, Int32Array};
 use arrow::datatypes::{DataType, Field, Schema};
 use arrow::record_batch::RecordBatch;
-use datafusion_common::{external_datafusion_err, DataFusionError};
+use datafusion_common::external_datafusion_err;
 use datafusion_datasource::source::DataSourceExec;
 
 #[cfg(feature = "compression")]

datafusion/datasource-parquet/src/file_format.rs

@@ -40,8 +40,8 @@ use datafusion_common::config::{ConfigField, ConfigFileType, TableParquetOptions
 use datafusion_common::parsers::CompressionTypeVariant;
 use datafusion_common::stats::Precision;
 use datafusion_common::{
-    internal_datafusion_err, internal_err, not_impl_err, ColumnStatistics,
-    DataFusionError, GetExt, Result, DEFAULT_PARQUET_EXTENSION,
+    execution_join_datafusion_err, internal_datafusion_err, internal_err, not_impl_err,
+    ColumnStatistics, DataFusionError, GetExt, Result, DEFAULT_PARQUET_EXTENSION,
 };
 use datafusion_common::{HashMap, Statistics};
 use datafusion_common_runtime::{JoinSet, SpawnedTask};
@@ -1210,7 +1210,7 @@ impl FileSink for ParquetSink {
         demux_task
             .join_unwind()
             .await
-            .map_err(DataFusionError::ExecutionJoin)??;
+            .map_err(|e| execution_join_datafusion_err!(e))??;
 
         Ok(row_count as u64)
     }
@@ -1338,7 +1338,7 @@ fn spawn_rg_join_and_finalize_task(
             let (writer, _col_reservation) = task
                 .join_unwind()
                 .await
-                .map_err(DataFusionError::ExecutionJoin)??;
+                .map_err(|e| execution_join_datafusion_err!(e))??;
             let encoded_size = writer.get_estimated_total_bytes();
             rg_reservation.grow(encoded_size);
             finalized_rg.push(writer.close()?);
@@ -1475,7 +1475,7 @@ async fn concatenate_parallel_row_groups(
         let result = task.join_unwind().await;
         let mut rg_out = parquet_writer.next_row_group()?;
         let (serialized_columns, mut rg_reservation, _cnt) =
-            result.map_err(DataFusionError::ExecutionJoin)??;
+            result.map_err(|e| execution_join_datafusion_err!(e))??;
         for chunk in serialized_columns {
             chunk.append_to_row_group(&mut rg_out)?;
             rg_reservation.free();
@@ -1542,7 +1542,7 @@ async fn output_single_parquet_file_parallelized(
     launch_serialization_task
         .join_unwind()
         .await
-        .map_err(DataFusionError::ExecutionJoin)??;
+        .map_err(|e| execution_join_datafusion_err!(e))??;
     Ok(file_metadata)
 }
 

datafusion/datasource/src/url.rs

@@ -128,9 +128,11 @@ impl ListingTableUrl {
         };
 
         let url = url_from_filesystem_path(path).ok_or_else(|| {
-            external_datafusion_err!(
-                format!("Failed to convert path to URL: {path}").into()
-            )
+            let err = std::io::Error::new(
+                std::io::ErrorKind::Other,
+                format!("Failed to convert path to URL: {path}"),
+            );
+            external_datafusion_err!(err)
         })?;
 
         Self::try_new(url, glob)

datafusion/datasource/src/write/orchestration.rs

@@ -27,7 +27,9 @@ use crate::file_compression_type::FileCompressionType;
 use datafusion_common::error::Result;
 
 use arrow::array::RecordBatch;
-use datafusion_common::{internal_datafusion_err, internal_err, DataFusionError};
+use datafusion_common::{
+    execution_join_datafusion_err, internal_datafusion_err, internal_err, DataFusionError,
+};
 use datafusion_common_runtime::{JoinSet, SpawnedTask};
 use datafusion_execution::TaskContext;
 
@@ -285,8 +287,8 @@ pub async fn spawn_writer_tasks_and_join(
         write_coordinator_task.join_unwind(),
         demux_task.join_unwind()
     );
-    r1.map_err(DataFusionError::ExecutionJoin)??;
-    r2.map_err(DataFusionError::ExecutionJoin)??;
+    r1.map_err(|e| execution_join_datafusion_err!(e))??;
+    r2.map_err(|e| execution_join_datafusion_err!(e))??;
 
     // Return total row count:
     rx_row_cnt.await.map_err(|_| {

datafusion/functions/src/regex/regexpreplace.rs

@@ -30,9 +30,7 @@ use datafusion_common::external_datafusion_err;
 use datafusion_common::external_err;
 use datafusion_common::plan_err;
 use datafusion_common::ScalarValue;
-use datafusion_common::{
-    cast::as_generic_string_array, internal_err, DataFusionError, Result,
-};
+use datafusion_common::{cast::as_generic_string_array, internal_err, Result};
 use datafusion_expr::function::Hint;
 use datafusion_expr::ColumnarValue;
 use datafusion_expr::TypeSignature;

datafusion/optimizer/src/simplify_expressions/regex.rs

@@ -15,7 +15,7 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use datafusion_common::{external_datafusion_err, DataFusionError, Result, ScalarValue};
+use datafusion_common::{external_datafusion_err, Result, ScalarValue};
 use datafusion_expr::{lit, BinaryExpr, Expr, Like, Operator};
 use regex_syntax::hir::{Capture, Hir, HirKind, Literal, Look};
 

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

@@ -33,9 +33,7 @@ use arrow::datatypes::{Schema, SchemaRef};
 use arrow::ipc::{reader::StreamReader, writer::StreamWriter};
 use arrow::record_batch::RecordBatch;
 
-use datafusion_common::{
-    exec_datafusion_err, external_err, DataFusionError, HashSet, Result,
-};
+use datafusion_common::{exec_datafusion_err, external_err, HashSet, Result};
 use datafusion_common_runtime::SpawnedTask;
 use datafusion_execution::disk_manager::RefCountedTempFile;
 use datafusion_execution::RecordBatchStream;

datafusion/sqllogictest/bin/sqllogictests.rs

@@ -271,7 +271,8 @@ async fn run_file_in_runner<D: AsyncDB, M: MakeConnection<Conn = D>>(
             }
             msg.push_str(&format!("{}. {err}\n\n", i + 1));
         }
-        return external_err!(msg.into());
+        let err = std::io::Error::new(std::io::ErrorKind::Other, msg);
+        return external_err!(err);
     }
 
     Ok(())