chore: Update to DataFusion 49.0.0

Signed-off-by: Andrew Lamb <andrew@nerdnetworks.org>

Author: alamb

Cargo.toml

@@ -47,9 +47,9 @@ object_store = { version = "0.12.1" }
 parquet = { version = "55.2.0" }
 
 # datafusion
-datafusion = "48.0.0"
-datafusion-ffi = "48.0.0"
-datafusion-proto = "48.0.0"
+datafusion = "49.0.0"
+datafusion-ffi = "49.0.0"
+datafusion-proto = "49.0.0"
 
 # serde
 serde = { version = "1.0.194", features = ["derive"] }
@@ -94,3 +94,10 @@ Arro3 = "Arro3"
 AKS = "AKS"
 # to avoid using 'type' as a field name.
 tpe = "tpe"
+
+# temp patch to https://github.com/apache/datafusion/pull/16822
+# before merge to branch-49
+[patch.crates-io]
+datafusion = { git = "https://github.com/apache/datafusion.git", rev = "fb02f4ec1d17f268ca7be9757fac121213c9bdc9" }
+datafusion-ffi = { git = "https://github.com/apache/datafusion.git", rev = "fb02f4ec1d17f268ca7be9757fac121213c9bdc9" }
+datafusion-proto = { git = "https://github.com/apache/datafusion.git", rev = "fb02f4ec1d17f268ca7be9757fac121213c9bdc9" }

crates/core/src/delta_datafusion/mod.rs

@@ -114,10 +114,10 @@ mod schema_adapter;
 impl From<DeltaTableError> for DataFusionError {
     fn from(err: DeltaTableError) -> Self {
         match err {
-            DeltaTableError::Arrow { source } => DataFusionError::ArrowError(source, None),
+            DeltaTableError::Arrow { source } => DataFusionError::from(source),
             DeltaTableError::Io { source } => DataFusionError::IoError(source),
-            DeltaTableError::ObjectStore { source } => DataFusionError::ObjectStore(source),
-            DeltaTableError::Parquet { source } => DataFusionError::ParquetError(source),
+            DeltaTableError::ObjectStore { source } => DataFusionError::from(source),
+            DeltaTableError::Parquet { source } => DataFusionError::from(source),
             _ => DataFusionError::External(Box::new(err)),
         }
     }
@@ -126,10 +126,10 @@ impl From<DeltaTableError> for DataFusionError {
 impl From<DataFusionError> for DeltaTableError {
     fn from(err: DataFusionError) -> Self {
         match err {
-            DataFusionError::ArrowError(source, _) => DeltaTableError::Arrow { source },
+            DataFusionError::ArrowError(source, _) => DeltaTableError::from(*source),
             DataFusionError::IoError(source) => DeltaTableError::Io { source },
-            DataFusionError::ObjectStore(source) => DeltaTableError::ObjectStore { source },
-            DataFusionError::ParquetError(source) => DeltaTableError::Parquet { source },
+            DataFusionError::ObjectStore(source) => DeltaTableError::from(*source),
+            DataFusionError::ParquetError(source) => DeltaTableError::from(*source),
             _ => DeltaTableError::Generic(err.to_string()),
         }
     }

crates/core/src/operations/merge/barrier.rs

@@ -299,10 +299,11 @@ impl Stream for MergeBarrierStream {
                                             .columns()
                                             .iter()
                                             .map(|c| {
-                                                arrow::compute::take(c.as_ref(), &indices, None)
-                                                    .map_err(|err| {
-                                                        DataFusionError::ArrowError(err, None)
-                                                    })
+                                                Ok(arrow::compute::take(
+                                                    c.as_ref(),
+                                                    &indices,
+                                                    None,
+                                                )?)
                                             })
                                             .collect::<DataFusionResult<Vec<ArrayRef>>>()?;