RFC: replace DatafusionError::Plam with concise macro

benchmarks/src/tpch/mod.rs

@@ -20,10 +20,10 @@
 use arrow::datatypes::SchemaBuilder;
 use datafusion::{
     arrow::datatypes::{DataType, Field, Schema},
+    common::plan_err,
     error::{DataFusionError, Result},
 };
 use std::fs;
-
 mod run;
 pub use run::RunOpt;
 
@@ -158,13 +158,9 @@ pub fn get_query_sql(query: usize) -> Result<Vec<String>> {
                 Err(e) => errors.push(format!("{filename}: {e}")),
             };
         }
-        Err(DataFusionError::Plan(format!(
-            "invalid query. Could not find query: {errors:?}"
-        )))
+        plan_err!("invalid query. Could not find query: {:?}", errors)
     } else {
-        Err(DataFusionError::Plan(
-            "invalid query. Expected value between 1 and 22".to_owned(),
-        ))
+        plan_err!("invalid query. Expected value between 1 and 22")
     }
 }
 

datafusion/common/src/error.rs

@@ -99,18 +99,6 @@ macro_rules! context {
     };
 }
 
-#[macro_export]
-macro_rules! plan_err {
-    ($desc:expr) => {
-        Err(datafusion_common::DataFusionError::Plan(format!(
-            "{} at {}:{}",
-            $desc,
-            file!(),
-            line!()
-        )))
-    };
-}
-
 /// Schema-related errors
 #[derive(Debug)]
 pub enum SchemaError {
@@ -533,3 +521,28 @@ macro_rules! unwrap_or_internal_err {
         })?
     };
 }
+
+#[macro_export]
+macro_rules! with_dollar_sign {
+    ($($body:tt)*) => {
+        macro_rules! __with_dollar_sign { $($body)* }
+        __with_dollar_sign!($);
+    }
+}
+
+macro_rules! make_error {
+    ($NAME:ident, $ERR:ident) => {
+        with_dollar_sign! {
+            ($d:tt) => {
+                #[macro_export]
+                macro_rules! $NAME {
+                    ($d($d args:expr),*) => {
+                        Err(DataFusionError::$ERR(format!($d($d args),*).into()))
+                    }
+                }
+            }
+        }
+    };
+}
+
+make_error!(plan_err, Plan);

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

@@ -21,7 +21,7 @@ use std::sync::Arc;
 
 use arrow::datatypes::{DataType, Schema, SchemaRef};
 use async_trait::async_trait;
-use datafusion_common::DataFusionError;
+use datafusion_common::{plan_err, DataFusionError};
 
 use crate::datasource::file_format::arrow::{ArrowFormat, DEFAULT_ARROW_EXTENSION};
 use crate::datasource::file_format::avro::DEFAULT_AVRO_EXTENSION;
@@ -439,10 +439,9 @@ pub trait ReadOptions<'a> {
                 .to_listing_options(config)
                 .infer_schema(&state, &table_path)
                 .await?),
-            (None, true) => Err(DataFusionError::Plan(
-                "Schema inference for infinite data sources is not supported."
-                    .to_string(),
-            )),
+            (None, true) => {
+                plan_err!("Schema inference for infinite data sources is not supported.")
+            }
         }
     }
 }

datafusion/core/src/datasource/listing/table.rs

@@ -25,7 +25,7 @@ use arrow::datatypes::{DataType, Field, SchemaBuilder, SchemaRef};
 use arrow_schema::Schema;
 use async_trait::async_trait;
 use dashmap::DashMap;
-use datafusion_common::{SchemaExt, ToDFSchema};
+use datafusion_common::{plan_err, SchemaExt, ToDFSchema};
 use datafusion_expr::expr::Sort;
 use datafusion_optimizer::utils::conjunction;
 use datafusion_physical_expr::{create_physical_expr, LexOrdering, PhysicalSortExpr};
@@ -640,14 +640,10 @@ impl ListingTable {
                         })
                     }
                     else {
-                        Err(DataFusionError::Plan(
-                            format!("Expected single column references in output_ordering, got {expr}")
-                        ))
+                        plan_err!("Expected single column references in output_ordering, got {expr}")
                     }
                 } else {
-                    Err(DataFusionError::Plan(
-                        format!("Expected Expr::Sort in output_ordering, but got {expr}")
-                    ))
+                    plan_err!("Expected Expr::Sort in output_ordering, but got {expr}")
                 }
             })
             .collect::<Result<Vec<_>>>()?;
@@ -777,17 +773,16 @@ impl TableProvider for ListingTable {
     ) -> Result<Arc<dyn ExecutionPlan>> {
         // Check that the schema of the plan matches the schema of this table.
         if !self.schema().equivalent_names_and_types(&input.schema()) {
-            return Err(DataFusionError::Plan(
+            return plan_err!(
                 // Return an error if schema of the input query does not match with the table schema.
-                "Inserting query must have the same schema with the table.".to_string(),
-            ));
+                "Inserting query must have the same schema with the table."
+            );
         }
 
         if self.table_paths().len() > 1 {
-            return Err(DataFusionError::Plan(
+            return plan_err!(
                 "Writing to a table backed by multiple files is not supported yet"
-                    .to_owned(),
-            ));
+            );
         }
 
         let table_path = &self.table_paths()[0];
@@ -806,10 +801,9 @@ impl TableProvider for ListingTable {
         let file_groups = file_list_stream.try_collect::<Vec<_>>().await?;
 
         if file_groups.len() > 1 {
-            return Err(DataFusionError::Plan(
+            return plan_err!(
                 "Datafusion currently supports tables from single partition and/or file."
-                    .to_owned(),
-            ));
+            );
         }
 
         // Sink related option, apart from format

datafusion/core/src/datasource/memory.rs

@@ -26,7 +26,7 @@ use std::sync::Arc;
 use arrow::datatypes::SchemaRef;
 use arrow::record_batch::RecordBatch;
 use async_trait::async_trait;
-use datafusion_common::SchemaExt;
+use datafusion_common::{plan_err, SchemaExt};
 use datafusion_execution::TaskContext;
 use tokio::sync::RwLock;
 use tokio::task::JoinSet;
@@ -64,9 +64,7 @@ impl MemTable {
                     "mem table schema does not contain batches schema. \
                         Target_schema: {schema:?}. Batches Schema: {batches_schema:?}"
                 );
-                return Err(DataFusionError::Plan(
-                    "Mismatch between schema and batches".to_string(),
-                ));
+                return plan_err!("Mismatch between schema and batches");
             }
         }
 
@@ -196,9 +194,9 @@ impl TableProvider for MemTable {
         // Create a physical plan from the logical plan.
         // Check that the schema of the plan matches the schema of this table.
         if !self.schema().equivalent_names_and_types(&input.schema()) {
-            return Err(DataFusionError::Plan(
-                "Inserting query must have the same schema with the table.".to_string(),
-            ));
+            return plan_err!(
+                "Inserting query must have the same schema with the table."
+            );
         }
         let sink = Arc::new(MemSink::new(self.batches.clone()));
         Ok(Arc::new(InsertExec::new(input, sink, self.schema.clone())))

datafusion/core/src/datasource/physical_plan/mod.rs

@@ -61,7 +61,10 @@ use crate::{
     scalar::ScalarValue,
 };
 
-use datafusion_common::tree_node::{TreeNode, VisitRecursion};
+use datafusion_common::{
+    plan_err,
+    tree_node::{TreeNode, VisitRecursion},
+};
 use datafusion_physical_expr::expressions::Column;
 
 use arrow::compute::cast;
@@ -557,12 +560,12 @@ impl SchemaAdapter {
                         projection.push(file_idx);
                     }
                     false => {
-                        return Err(DataFusionError::Plan(format!(
+                        return plan_err!(
                             "Cannot cast file schema field {} of type {:?} to table schema field of type {:?}",
                             file_field.name(),
                             file_field.data_type(),
                             table_field.data_type()
-                        )))
+                        )
                     }
                 }
             }

datafusion/core/src/datasource/streaming.rs

@@ -23,7 +23,7 @@ use std::sync::Arc;
 use arrow::datatypes::SchemaRef;
 use async_trait::async_trait;
 
-use datafusion_common::{DataFusionError, Result};
+use datafusion_common::{plan_err, DataFusionError, Result};
 use datafusion_expr::{Expr, TableType};
 use log::debug;
 
@@ -52,9 +52,7 @@ impl StreamingTable {
                     "target schema does not contain partition schema. \
                         Target_schema: {schema:?}. Partiton Schema: {partition_schema:?}"
                 );
-                return Err(DataFusionError::Plan(
-                    "Mismatch between schema and batches".to_string(),
-                ));
+                return plan_err!("Mismatch between schema and batches");
             }
         }
 

datafusion/core/src/execution/context.rs

@@ -28,7 +28,7 @@ use crate::{
     optimizer::optimizer::Optimizer,
     physical_optimizer::optimizer::{PhysicalOptimizer, PhysicalOptimizerRule},
 };
-use datafusion_common::alias::AliasGenerator;
+use datafusion_common::{alias::AliasGenerator, plan_err};
 use datafusion_execution::registry::SerializerRegistry;
 use datafusion_expr::{
     logical_plan::{DdlStatement, Statement},
@@ -958,10 +958,9 @@ impl SessionContext {
             (Some(s), _) => s,
             (None, false) => options.infer_schema(&self.state(), &table_path).await?,
             (None, true) => {
-                return Err(DataFusionError::Plan(
+                return plan_err!(
                     "Schema inference for infinite data sources is not supported."
-                        .to_string(),
-                ))
+                )
             }
         };
         let config = ListingTableConfig::new(table_path)
@@ -1182,7 +1181,7 @@ impl SessionContext {
         let schema = self.state.read().schema_for_ref(table_ref)?;
         match schema.table(&table).await {
             Some(ref provider) => Ok(Arc::clone(provider)),
-            _ => Err(DataFusionError::Plan(format!("No table named '{table}'"))),
+            _ => plan_err!("No table named '{table}'"),
         }
     }
 

datafusion/core/src/physical_optimizer/pipeline_checker.rs

@@ -26,7 +26,7 @@ use crate::physical_plan::joins::SymmetricHashJoinExec;
 use crate::physical_plan::{with_new_children_if_necessary, ExecutionPlan};
 use datafusion_common::config::OptimizerOptions;
 use datafusion_common::tree_node::{Transformed, TreeNode, VisitRecursion};
-use datafusion_common::DataFusionError;
+use datafusion_common::{plan_err, DataFusionError};
 use datafusion_physical_expr::intervals::{check_support, is_datatype_supported};
 use std::sync::Arc;
 
@@ -142,7 +142,7 @@ pub fn check_finiteness_requirements(
         {
             const MSG: &str = "Join operation cannot operate on a non-prunable stream without enabling \
                                the 'allow_symmetric_joins_without_pruning' configuration flag";
-            return Err(DataFusionError::Plan(MSG.to_owned()));
+            return plan_err!("{}", MSG);
         }
     }
     input