refactor: use DataFusion errors in integration

crates/datafusion/Cargo.toml

@@ -29,7 +29,6 @@ license = { workspace = true }
 repository = { workspace = true }
 
 [dependencies]
-anyhow = { workspace = true }
 async-trait = { workspace = true }
 dashmap = "6"
 datafusion = { workspace = true }

crates/datafusion/src/catalog.rs

@@ -19,10 +19,12 @@ use std::collections::HashMap;
 use std::sync::Arc;
 
 use datafusion::catalog::{CatalogProvider, SchemaProvider};
+use datafusion::error::Result;
 use futures::future::try_join_all;
-use iceberg::{Catalog, NamespaceIdent, Result};
+use iceberg::{Catalog, NamespaceIdent};
 
 use crate::schema::IcebergSchemaProvider;
+use crate::to_datafusion_error;
 
 /// Provides an interface to manage and access multiple schemas
 /// within an Iceberg [`Catalog`].
@@ -51,7 +53,8 @@ impl IcebergCatalogProvider {
         // As of right now; schemas might become stale.
         let schema_names: Vec<_> = client
             .list_namespaces(None)
-            .await?
+            .await
+            .map_err(to_datafusion_error)?
             .iter()
             .flat_map(|ns| ns.as_ref().clone())
             .collect();

crates/datafusion/src/error.rs

@@ -15,18 +15,109 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use anyhow::anyhow;
+use datafusion::error::DataFusionError;
 use iceberg::{Error, ErrorKind};
 
 /// Converts a datafusion error into an iceberg error.
-pub fn from_datafusion_error(error: datafusion::error::DataFusionError) -> Error {
+pub fn from_datafusion_error(error: DataFusionError) -> Error {
+    let fallback_message = error.to_string();
+    let DataFusionError::Context(context, error) = error else {
+        return unexpected_datafusion_error(fallback_message);
+    };
+
+    let Some(kind) = parse_iceberg_error_kind(&context) else {
+        return unexpected_datafusion_error(fallback_message);
+    };
+    let DataFusionError::Execution(message) = error.as_ref() else {
+        return unexpected_datafusion_error(fallback_message);
+    };
+
+    Error::new(kind, strip_error_kind(kind, message))
+}
+
+/// Converts an iceberg error into a datafusion error.
+pub fn to_datafusion_error(error: Error) -> DataFusionError {
+    DataFusionError::Context(
+        format!("IcebergError({})", error.kind()),
+        Box::new(DataFusionError::Execution(error.to_string())),
+    )
+}
+
+fn unexpected_datafusion_error(message: String) -> Error {
     Error::new(
         ErrorKind::Unexpected,
-        "Operation failed for hitting datafusion error".to_string(),
+        format!("DataFusion execution failed: {message}"),
     )
-    .with_source(anyhow!("datafusion error: {error:?}"))
 }
-/// Converts an iceberg error into a datafusion error.
-pub fn to_datafusion_error(error: Error) -> datafusion::error::DataFusionError {
-    datafusion::error::DataFusionError::External(error.into())
+
+fn parse_iceberg_error_kind(context: &str) -> Option<ErrorKind> {
+    let kind = context.strip_prefix("IcebergError(")?.strip_suffix(')')?;
+
+    match kind {
+        "PreconditionFailed" => Some(ErrorKind::PreconditionFailed),
+        "Unexpected" => Some(ErrorKind::Unexpected),
+        "DataInvalid" => Some(ErrorKind::DataInvalid),
+        "NamespaceAlreadyExists" => Some(ErrorKind::NamespaceAlreadyExists),
+        "TableAlreadyExists" => Some(ErrorKind::TableAlreadyExists),
+        "NamespaceNotFound" => Some(ErrorKind::NamespaceNotFound),
+        "TableNotFound" => Some(ErrorKind::TableNotFound),
+        "FeatureUnsupported" => Some(ErrorKind::FeatureUnsupported),
+        "CatalogCommitConflicts" => Some(ErrorKind::CatalogCommitConflicts),
+        _ => None,
+    }
+}
+
+fn strip_error_kind(kind: ErrorKind, message: &str) -> String {
+    let kind = kind.into_static();
+    if message == kind {
+        String::new()
+    } else {
+        message
+            .strip_prefix(&format!("{kind} => "))
+            .unwrap_or(message)
+            .to_string()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use datafusion::error::DataFusionError;
+    use iceberg::{Error, ErrorKind};
+
+    use super::{from_datafusion_error, to_datafusion_error};
+
+    #[test]
+    fn roundtrips_iceberg_error_kind_and_message() {
+        let error = Error::new(ErrorKind::DataInvalid, "invalid manifest");
+        let roundtripped = from_datafusion_error(to_datafusion_error(error));
+
+        assert_eq!(roundtripped.kind(), ErrorKind::DataInvalid);
+        assert_eq!(roundtripped.to_string(), "DataInvalid => invalid manifest");
+    }
+
+    #[test]
+    fn encodes_iceberg_errors_with_native_datafusion_variants() {
+        let error =
+            to_datafusion_error(Error::new(ErrorKind::DataInvalid, "invalid manifest"));
+
+        assert!(matches!(
+            error,
+            DataFusionError::Context(context, inner)
+                if context == "IcebergError(DataInvalid)"
+                    && matches!(inner.as_ref(), DataFusionError::Execution(message) if message == "DataInvalid => invalid manifest")
+        ));
+    }
+
+    #[test]
+    fn maps_non_iceberg_datafusion_errors_to_unexpected() {
+        let error = from_datafusion_error(DataFusionError::Execution(
+            "worker failed".to_string(),
+        ));
+
+        assert_eq!(error.kind(), ErrorKind::Unexpected);
+        assert_eq!(
+            error.to_string(),
+            "Unexpected => DataFusion execution failed: Execution error: worker failed"
+        );
+    }
 }

crates/datafusion/src/physical_plan/commit.rs

@@ -22,8 +22,10 @@ use datafusion::arrow::array::{Array, ArrayRef, RecordBatch, StringArray, UInt64
 use datafusion::arrow::datatypes::{
     DataType, Field, Schema as ArrowSchema, SchemaRef as ArrowSchemaRef,
 };
-use datafusion::common::tree_node::TreeNodeRecursion;
-use datafusion::common::{DataFusionError, Result as DFResult};
+use datafusion::common::{
+    internal_datafusion_err, internal_err, tree_node::TreeNodeRecursion,
+};
+use datafusion::error::{DataFusionError, Result};
 use datafusion::execution::{SendableRecordBatchStream, TaskContext};
 use datafusion::physical_expr::{EquivalenceProperties, Partitioning, PhysicalExpr};
 use datafusion::physical_plan::execution_plan::{Boundedness, EmissionType};
@@ -84,7 +86,7 @@ impl IcebergCommitExec {
     }
 
     // Create a record batch with just the count of rows written
-    fn make_count_batch(count: u64) -> DFResult<RecordBatch> {
+    fn make_count_batch(count: u64) -> Result<RecordBatch> {
         let count_array = Arc::new(UInt64Array::from(vec![count])) as ArrayRef;
 
         RecordBatch::try_from_iter_with_nullable(vec![("count", count_array, false)])
@@ -142,8 +144,8 @@ impl ExecutionPlan for IcebergCommitExec {
 
     fn apply_expressions(
         &self,
-        _f: &mut dyn FnMut(&Arc<dyn PhysicalExpr>) -> DFResult<TreeNodeRecursion>,
-    ) -> DFResult<TreeNodeRecursion> {
+        _f: &mut dyn FnMut(&Arc<dyn PhysicalExpr>) -> Result<TreeNodeRecursion>,
+    ) -> Result<TreeNodeRecursion> {
         Ok(TreeNodeRecursion::Continue)
     }
 
@@ -163,12 +165,12 @@ impl ExecutionPlan for IcebergCommitExec {
     fn with_new_children(
         self: Arc<Self>,
         children: Vec<Arc<dyn ExecutionPlan>>,
-    ) -> DFResult<Arc<dyn ExecutionPlan>> {
+    ) -> Result<Arc<dyn ExecutionPlan>> {
         if children.len() != 1 {
-            return Err(DataFusionError::Internal(format!(
+            return internal_err!(
                 "IcebergCommitExec expects exactly one child, but provided {}",
                 children.len()
-            )));
+            );
         }
 
         Ok(Arc::new(IcebergCommitExec::new(
@@ -183,12 +185,12 @@ impl ExecutionPlan for IcebergCommitExec {
         &self,
         partition: usize,
         context: Arc<TaskContext>,
-    ) -> DFResult<SendableRecordBatchStream> {
+    ) -> Result<SendableRecordBatchStream> {
         // IcebergCommitExec only has one partition (partition 0)
         if partition != 0 {
-            return Err(DataFusionError::Internal(format!(
+            return internal_err!(
                 "IcebergCommitExec only has one partition, but got partition {partition}"
-            )));
+            );
         }
 
         let table = self.table.clone();
@@ -215,23 +217,23 @@ impl ExecutionPlan for IcebergCommitExec {
                 let files_array = batch
                     .column_by_name(DATA_FILES_COL_NAME)
                     .ok_or_else(|| {
-                        DataFusionError::Internal(
-                            "Expected 'data_files' column in input batch".to_string(),
+                        internal_datafusion_err!(
+                            "Expected 'data_files' column in input batch"
                         )
                     })?
                     .as_any()
                     .downcast_ref::<StringArray>()
                     .ok_or_else(|| {
-                        DataFusionError::Internal(
-                            "Expected 'data_files' column to be StringArray".to_string(),
+                        internal_datafusion_err!(
+                            "Expected 'data_files' column to be StringArray"
                         )
                     })?;
 
                 // Deserialize all data files from the StringArray
                 let batch_files: Vec<DataFile> = files_array
                     .into_iter()
                     .flatten()
-                    .map(|f| -> DFResult<DataFile> {
+                    .map(|f| -> Result<DataFile> {
                         // Parse JSON to DataFileSerde and convert to DataFile
                         deserialize_data_file_from_json(
                             f,
@@ -241,7 +243,7 @@ impl ExecutionPlan for IcebergCommitExec {
                         )
                         .map_err(to_datafusion_error)
                     })
-                    .collect::<datafusion::common::Result<_>>()?;
+                    .collect::<Result<_>>()?;
 
                 // add record_counts from the current batch to total record count
                 total_record_count +=
@@ -361,23 +363,23 @@ mod tests {
 
         fn apply_expressions(
             &self,
-            _f: &mut dyn FnMut(&Arc<dyn PhysicalExpr>) -> DFResult<TreeNodeRecursion>,
-        ) -> DFResult<TreeNodeRecursion> {
+            _f: &mut dyn FnMut(&Arc<dyn PhysicalExpr>) -> Result<TreeNodeRecursion>,
+        ) -> Result<TreeNodeRecursion> {
             Ok(TreeNodeRecursion::Continue)
         }
 
         fn with_new_children(
             self: Arc<Self>,
             _children: Vec<Arc<dyn ExecutionPlan>>,
-        ) -> datafusion::common::Result<Arc<dyn ExecutionPlan>> {
+        ) -> Result<Arc<dyn ExecutionPlan>> {
             Ok(self)
         }
 
         fn execute(
             &self,
             _partition: usize,
             _context: Arc<TaskContext>,
-        ) -> datafusion::common::Result<SendableRecordBatchStream> {
+        ) -> Result<SendableRecordBatchStream> {
             // Create a record batch with the serialized data files
             let array =
                 Arc::new(StringArray::from(self.data_files_json.clone())) as ArrayRef;

crates/datafusion/src/physical_plan/project.rs

@@ -21,7 +21,8 @@ use std::sync::Arc;
 
 use datafusion::arrow::array::RecordBatch;
 use datafusion::arrow::datatypes::{DataType, Schema as ArrowSchema};
-use datafusion::common::{DataFusionError, Result as DFResult};
+use datafusion::common::plan_err;
+use datafusion::error::Result;
 use datafusion::physical_expr::PhysicalExpr;
 use datafusion::physical_expr::expressions::Column;
 use datafusion::physical_plan::projection::ProjectionExec;
@@ -51,7 +52,7 @@ use crate::to_datafusion_error;
 pub fn project_with_partition(
     input: Arc<dyn ExecutionPlan>,
     table: &Table,
-) -> DFResult<Arc<dyn ExecutionPlan>> {
+) -> Result<Arc<dyn ExecutionPlan>> {
     let metadata = table.metadata();
     let partition_spec = metadata.default_partition_spec();
     let table_schema = metadata.current_schema();
@@ -72,11 +73,11 @@ pub fn project_with_partition(
         .map_err(to_datafusion_error)?;
 
     if input_schema_cleaned != expected_schema_cleaned {
-        return Err(DataFusionError::Plan(format!(
+        return plan_err!(
             "Input schema does not match Iceberg table schema.\n\
              Expected schema: {expected_schema_cleaned}\n\
              Input schema: {input_schema_cleaned}"
-        )));
+        );
     }
 
     let calculator =
@@ -129,15 +130,15 @@ impl PartialEq for PartitionExpr {
 impl Eq for PartitionExpr {}
 
 impl PhysicalExpr for PartitionExpr {
-    fn data_type(&self, _input_schema: &ArrowSchema) -> DFResult<DataType> {
+    fn data_type(&self, _input_schema: &ArrowSchema) -> Result<DataType> {
         Ok(self.calculator.partition_arrow_type().clone())
     }
 
-    fn nullable(&self, _input_schema: &ArrowSchema) -> DFResult<bool> {
+    fn nullable(&self, _input_schema: &ArrowSchema) -> Result<bool> {
         Ok(false)
     }
 
-    fn evaluate(&self, batch: &RecordBatch) -> DFResult<ColumnarValue> {
+    fn evaluate(&self, batch: &RecordBatch) -> Result<ColumnarValue> {
         let array = self
             .calculator
             .calculate(batch)
@@ -152,7 +153,7 @@ impl PhysicalExpr for PartitionExpr {
     fn with_new_children(
         self: Arc<Self>,
         _children: Vec<Arc<dyn PhysicalExpr>>,
-    ) -> DFResult<Arc<dyn PhysicalExpr>> {
+    ) -> Result<Arc<dyn PhysicalExpr>> {
         Ok(self)
     }
 

crates/datafusion/src/physical_plan/repartition.rs

@@ -18,7 +18,8 @@
 use std::num::NonZeroUsize;
 use std::sync::Arc;
 
-use datafusion::error::{DataFusionError, Result as DFResult};
+use datafusion::common::plan_err;
+use datafusion::error::Result;
 use datafusion::physical_expr::PhysicalExpr;
 use datafusion::physical_plan::expressions::Column;
 use datafusion::physical_plan::repartition::RepartitionExec;
@@ -90,7 +91,7 @@ pub(crate) fn repartition(
     input: Arc<dyn ExecutionPlan>,
     table_metadata: TableMetadataRef,
     target_partitions: NonZeroUsize,
-) -> DFResult<Arc<dyn ExecutionPlan>> {
+) -> Result<Arc<dyn ExecutionPlan>> {
     let partitioning_strategy =
         determine_partitioning_strategy(&input, &table_metadata, target_partitions)?;
 
@@ -125,7 +126,7 @@ fn determine_partitioning_strategy(
     input: &Arc<dyn ExecutionPlan>,
     table_metadata: &TableMetadata,
     target_partitions: NonZeroUsize,
-) -> DFResult<Partitioning> {
+) -> Result<Partitioning> {
     let partition_spec = table_metadata.default_partition_spec();
     let input_schema = input.schema();
     let target_partition_count = target_partitions.get();
@@ -158,10 +159,10 @@ fn determine_partitioning_strategy(
         }
 
         // Case 2: Partitioned table missing _partition column (normally this should not happen)
-        (true, Err(_)) => Err(DataFusionError::Plan(format!(
+        (true, Err(_)) => plan_err!(
             "Partitioned table input missing {PROJECTED_PARTITION_VALUE_COLUMN} column. \
              Ensure projection happens before repartitioning."
-        ))),
+        ),
 
         // Case 3: Unpartitioned table, always use RoundRobinBatch
         (false, _) => Ok(Partitioning::RoundRobinBatch(target_partition_count)),