better cast error

Author: phillipleblanc

datafusion-federation/src/schema_cast/record_convert.rs

@@ -21,6 +21,14 @@ pub enum Error {
         source: datafusion::arrow::error::ArrowError,
     },
 
+    UnableToCastColumn {
+        source: datafusion::arrow::error::ArrowError,
+        column_index: usize,
+        column_name: String,
+        from_type: DataType,
+        to_type: DataType,
+    },
+
     UnexpectedNumberOfColumns {
         expected: usize,
         found: usize,
@@ -35,6 +43,18 @@ impl std::fmt::Display for Error {
             Error::UnableToConvertRecordBatch { source } => {
                 write!(f, "Unable to convert record batch: {source}")
             }
+            Error::UnableToCastColumn {
+                source,
+                column_index,
+                column_name,
+                from_type,
+                to_type,
+            } => {
+                write!(
+                    f,
+                    "Unable to cast column {column_index} '{column_name}' from {from_type} to {to_type}: {source}"
+                )
+            }
             Error::UnexpectedNumberOfColumns { expected, found } => {
                 write!(
                     f,
@@ -70,73 +90,81 @@ pub fn try_cast_to(record_batch: RecordBatch, expected_schema: SchemaRef) -> Res
         .enumerate()
         .map(|(i, expected_field)| {
             let record_batch_col = record_batch.column(i);
+            let from_type = record_batch_col.data_type().clone();
+            let to_type = expected_field.data_type().clone();
+            let make_err = |e| Error::UnableToCastColumn {
+                source: e,
+                column_index: i,
+                column_name: expected_field.name().clone(),
+                from_type: from_type.clone(),
+                to_type: to_type.clone(),
+            };
 
             match (record_batch_col.data_type(), expected_field.data_type()) {
                 (DataType::Utf8, DataType::List(item_type)) => {
                     cast_string_to_list::<i32>(&Arc::clone(record_batch_col), item_type)
-                        .map_err(|e| Error::UnableToConvertRecordBatch { source: e })
+                        .map_err(make_err)
                 }
                 (DataType::Utf8, DataType::LargeList(item_type)) => {
                     cast_string_to_large_list::<i32>(&Arc::clone(record_batch_col), item_type)
-                        .map_err(|e| Error::UnableToConvertRecordBatch { source: e })
+                        .map_err(make_err)
                 }
                 (DataType::Utf8, DataType::FixedSizeList(item_type, value_length)) => {
                     cast_string_to_fixed_size_list::<i32>(
                         &Arc::clone(record_batch_col),
                         item_type,
                         *value_length,
                     )
-                    .map_err(|e| Error::UnableToConvertRecordBatch { source: e })
+                    .map_err(make_err)
                 }
                 (DataType::Utf8, DataType::Struct(_)) => cast_string_to_struct::<i32>(
                     &Arc::clone(record_batch_col),
                     expected_field.clone(),
                 )
-                .map_err(|e| Error::UnableToConvertRecordBatch { source: e }),
+                .map_err(make_err),
                 (DataType::LargeUtf8, DataType::List(item_type)) => {
                     cast_string_to_list::<i64>(&Arc::clone(record_batch_col), item_type)
-                        .map_err(|e| Error::UnableToConvertRecordBatch { source: e })
+                        .map_err(make_err)
                 }
                 (DataType::LargeUtf8, DataType::LargeList(item_type)) => {
                     cast_string_to_large_list::<i64>(&Arc::clone(record_batch_col), item_type)
-                        .map_err(|e| Error::UnableToConvertRecordBatch { source: e })
+                        .map_err(make_err)
                 }
                 (DataType::LargeUtf8, DataType::FixedSizeList(item_type, value_length)) => {
                     cast_string_to_fixed_size_list::<i64>(
                         &Arc::clone(record_batch_col),
                         item_type,
                         *value_length,
                     )
-                    .map_err(|e| Error::UnableToConvertRecordBatch { source: e })
+                    .map_err(make_err)
                 }
                 (DataType::LargeUtf8, DataType::Struct(_)) => cast_string_to_struct::<i64>(
                     &Arc::clone(record_batch_col),
                     expected_field.clone(),
                 )
-                .map_err(|e| Error::UnableToConvertRecordBatch { source: e }),
+                .map_err(make_err),
                 (
                     DataType::Interval(IntervalUnit::MonthDayNano),
                     DataType::Interval(IntervalUnit::YearMonth),
                 ) => cast_interval_monthdaynano_to_yearmonth(&Arc::clone(record_batch_col))
-                    .map_err(|e| Error::UnableToConvertRecordBatch { source: e }),
+                    .map_err(make_err),
                 (
                     DataType::Interval(IntervalUnit::MonthDayNano),
                     DataType::Interval(IntervalUnit::DayTime),
                 ) => cast_interval_monthdaynano_to_daytime(&Arc::clone(record_batch_col))
-                    .map_err(|e| Error::UnableToConvertRecordBatch { source: e }),
+                    .map_err(make_err),
                 _ => cast(&Arc::clone(record_batch_col), expected_field.data_type())
-                    .map_err(|e| Error::UnableToConvertRecordBatch { source: e }),
+                    .map_err(make_err),
             }
         })
         .collect::<Result<Vec<Arc<dyn Array>>>>()
-        .map_err(|e| {
+        .inspect_err(|_| {
             tracing::debug!(
                 actual_schema = ?actual_schema,
                 expected_schema = ?expected_schema,
                 "Cast error in try_cast_to"
             );
             tracing::trace!(record_batch = ?record_batch, "Record batch contents");
-            e
         })?;
 
     RecordBatch::try_new(expected_schema.clone(), cols).map_err(|e| {