Upgrade to DataFusion 46 (#309)

* Upgrade to DataFusion 46

* Fix lint

Author: phillipleblanc

Cargo.lock

@@ -40,12 +40,12 @@ arrow-flight = { version = "54.2.1", features = [
 arrow-schema = { version = "54.2.1", features = ["serde"] }
 arrow-json = "54.2.1"
 arrow-odbc = { version = "=15.1.1" }
-datafusion = { version = "45", default-features = false }
-datafusion-expr = { version = "45" }
-datafusion-federation = { version = "=0.3.6" }
-datafusion-ffi = { version = "45" }
-datafusion-proto = { version = "45" }
-datafusion-physical-expr = { version = "45" }
-datafusion-physical-plan = { version = "45" }
+datafusion = { version = "46", default-features = false }
+datafusion-expr = { version = "46" }
+datafusion-federation = { version = "=0.3.7" }
+datafusion-ffi = { version = "46" }
+datafusion-proto = { version = "46" }
+datafusion-physical-expr = { version = "46" }
+datafusion-physical-plan = { version = "46" }
 datafusion-table-providers = { path = "core" }
 duckdb = { version = "=1.2.1" }

Cargo.toml

@@ -616,12 +616,12 @@ impl DuckDB {
             );
         }
         if !extra_in_actual.is_empty() {
-           tracing::warn!(
-    "Unexpected index(es) detected in table '{name}': {}.\n\
+            tracing::warn!(
+                "Unexpected index(es) detected in table '{name}': {}.\n\
      These indexes are not defined in the configuration.",
-    extra_in_actual.iter().join(", "),
-    name = self.table_name
-);
+                extra_in_actual.iter().join(", "),
+                name = self.table_name
+            );
         }
 
         Ok(missing_in_actual.is_empty() && extra_in_actual.is_empty())
@@ -785,7 +785,7 @@ pub(crate) mod tests {
         let ctx = SessionContext::new();
         let cmd = CreateExternalTable {
             schema: Arc::new(schema.to_dfschema().expect("to df schema")),
-            name: table_name.into(),
+            name: table_name,
             location: "".to_string(),
             file_type: "".to_string(),
             table_partition_cols: vec![],

core/src/duckdb.rs

@@ -196,7 +196,7 @@ async fn flight_stream(
             Err(e) => errors.push(Box::new(e)),
         }
     }
-    let err = errors.into_iter().last().unwrap_or_else(|| {
+    let err = errors.into_iter().next_back().unwrap_or_else(|| {
         Box::new(FlightError::ProtocolError(format!(
             "No available location for endpoint {:?}",
             partition.locations

core/src/flight/exec.rs

@@ -3,11 +3,11 @@ use datafusion::arrow::array::RecordBatch;
 use datafusion::arrow::datatypes::SchemaRef;
 use datafusion::catalog::TableProviderFactory;
 use datafusion::common::{Constraints, ToDFSchema};
+use datafusion::datasource::memory::MemorySourceConfig;
 use datafusion::execution::context::SessionContext;
 use datafusion::logical_expr::dml::InsertOp;
 use datafusion::logical_expr::CreateExternalTable;
 use datafusion::physical_plan::collect;
-use datafusion::physical_plan::memory::MemoryExec;
 use datafusion_federation::schema_cast::record_convert::try_cast_to;
 use datafusion_table_providers::duckdb::DuckDBTableProviderFactory;
 use rstest::rstest;
@@ -43,10 +43,14 @@ async fn arrow_duckdb_round_trip(
 
     let ctx = SessionContext::new();
 
-    let mem_exec = MemoryExec::try_new(&[vec![arrow_record.clone()]], arrow_record.schema(), None)
-        .expect("memory exec created");
+    let mem_exec = MemorySourceConfig::try_new_exec(
+        &[vec![arrow_record.clone()]],
+        arrow_record.schema(),
+        None,
+    )
+    .expect("memory exec created");
     let insert_plan = table_provider
-        .insert_into(&ctx.state(), Arc::new(mem_exec), InsertOp::Append)
+        .insert_into(&ctx.state(), mem_exec, InsertOp::Append)
         .await
         .expect("insert plan created");
 

core/tests/duckdb/mod.rs

@@ -2,7 +2,7 @@ use rand::Rng;
 
 mod arrow_record_batch_gen;
 mod docker;
-#[cfg(feature = "duckdb")]
+#[cfg(all(feature = "duckdb", feature = "federation"))]
 mod duckdb;
 #[cfg(feature = "flight")]
 mod flight;

core/tests/integration.rs

@@ -1,5 +1,5 @@
 use crate::arrow_record_batch_gen::*;
-use datafusion::execution::context::SessionContext;
+use datafusion::{datasource::memory::MemorySourceConfig, execution::context::SessionContext};
 use datafusion_table_providers::{
     mysql::DynMySQLConnectionPool, sql::sql_provider_datafusion::SqlTable,
 };
@@ -20,7 +20,6 @@ use datafusion::common::{Constraints, ToDFSchema};
 use datafusion::logical_expr::dml::InsertOp;
 use datafusion::logical_expr::CreateExternalTable;
 use datafusion::physical_plan::collect;
-use datafusion::physical_plan::memory::MemoryExec;
 #[cfg(feature = "mysql-federation")]
 use datafusion_federation::schema_cast::record_convert::try_cast_to;
 use datafusion_table_providers::mysql::MySQLTableProviderFactory;
@@ -708,10 +707,14 @@ async fn arrow_mysql_round_trip(
         .expect("table provider created");
 
     let ctx = SessionContext::new();
-    let mem_exec = MemoryExec::try_new(&[vec![arrow_record.clone()]], arrow_record.schema(), None)
-        .expect("memory exec created");
+    let mem_exec = MemorySourceConfig::try_new_exec(
+        &[vec![arrow_record.clone()]],
+        arrow_record.schema(),
+        None,
+    )
+    .expect("memory exec created");
     let insert_plan = table_provider
-        .insert_into(&ctx.state(), Arc::new(mem_exec), InsertOp::Overwrite)
+        .insert_into(&ctx.state(), mem_exec, InsertOp::Overwrite)
         .await
         .expect("insert plan created");
 

core/tests/mysql/mod.rs

@@ -3,12 +3,14 @@ use arrow::{
     array::{Decimal128Array, RecordBatch},
     datatypes::{DataType, Field, Schema, SchemaRef},
 };
-use datafusion::common::{Constraints, ToDFSchema};
 use datafusion::execution::context::SessionContext;
 use datafusion::logical_expr::CreateExternalTable;
 use datafusion::physical_plan::collect;
-use datafusion::physical_plan::memory::MemoryExec;
 use datafusion::{catalog::TableProviderFactory, logical_expr::dml::InsertOp};
+use datafusion::{
+    common::{Constraints, ToDFSchema},
+    datasource::memory::MemorySourceConfig,
+};
 #[cfg(feature = "postgres-federation")]
 use datafusion_federation::schema_cast::record_convert::try_cast_to;
 
@@ -55,10 +57,14 @@ async fn arrow_postgres_round_trip(
         .expect("table provider created");
 
     let ctx = SessionContext::new();
-    let mem_exec = MemoryExec::try_new(&[vec![arrow_record.clone()]], arrow_record.schema(), None)
-        .expect("memory exec created");
+    let mem_exec = MemorySourceConfig::try_new_exec(
+        &[vec![arrow_record.clone()]],
+        arrow_record.schema(),
+        None,
+    )
+    .expect("memory exec created");
     let insert_plan = table_provider
-        .insert_into(&ctx.state(), Arc::new(mem_exec), InsertOp::Append)
+        .insert_into(&ctx.state(), mem_exec, InsertOp::Append)
         .await
         .expect("insert plan created");