Upgrade to DataFusion 46 (#346)

* Upgrade to DataFusion 46

* fix

* Use the same DF patch as spice

* fix

* Use 1.85 toolchain

* Fix linking issue

Author: phillipleblanc

Cargo.toml

@@ -20,11 +20,11 @@ byte-unit = { version = "5.1.4", optional = true }
 bigdecimal = "0.4.5"
 byteorder = "1.5.0"
 chrono = "0.4"
-datafusion = "45"
-datafusion-expr = { version = "45", optional = true }
-datafusion-physical-expr = { version = "45", optional = true }
-datafusion-physical-plan = { version = "45", optional = true }
-datafusion-proto = { version = "45", optional = true }
+datafusion = "46"
+datafusion-expr = { version = "46", optional = true }
+datafusion-physical-expr = { version = "46", optional = true }
+datafusion-physical-plan = { version = "46", optional = true }
+datafusion-proto = { version = "46", optional = true }
 duckdb = { version = "1", features = [
   "bundled",
   "r2d2",
@@ -59,7 +59,7 @@ pem = { version = "3.0.4", optional = true }
 tokio-rusqlite = { version = "0.5.1", optional = true }
 tonic = { version = "0.12.2", optional = true }
 datafusion-federation = "0.1"
-datafusion-federation-sql = { git = "https://github.com/spiceai/datafusion-federation.git", rev = "1c814422cfbdeb7f7bd2399e3369656710413aa8" }
+datafusion-federation-sql = { git = "https://github.com/spiceai/datafusion-federation.git", rev = "a4c752a172904a0c214fc37ab2d62e907eb16c8f" }
 itertools = "0.13.0"
 dyn-clone = { version = "1.0.17", optional = true }
 geo-types = "0.7.13"
@@ -77,7 +77,7 @@ rstest = "0.22.0"
 geozero = { version = "0.13.0", features = ["with-wkb"] }
 tokio-stream = { version = "0.1.15", features = ["net"] }
 insta = { version = "1.40.0", features = ["filters"] }
-datafusion-physical-plan = { version = "45" }
+datafusion-physical-plan = { version = "46" }
 tempfile = "3.8.1"
 
 [features]
@@ -103,11 +103,11 @@ sqlite-federation = ["sqlite"]
 postgres-federation = ["postgres"]
 
 [patch.crates-io]
-datafusion-federation = { git = "https://github.com/spiceai/datafusion-federation.git", rev = "1c814422cfbdeb7f7bd2399e3369656710413aa8" } # spiceai-45
+datafusion-federation = { git = "https://github.com/spiceai/datafusion-federation.git", rev = "a4c752a172904a0c214fc37ab2d62e907eb16c8f" } # spiceai-46
 duckdb = { git = "https://github.com/spiceai/duckdb-rs.git", rev = "2e24b958e44ec7419290249e27a15f1a19703fff" } # spiceai-1.1.3-backported-arrow-54
 
-datafusion = { git = "https://github.com/spiceai/datafusion.git", rev = "e0e423e9e29a711d076892865d51747b06429d1b"} # spiceai-45
-datafusion-expr =  { git = "https://github.com/spiceai/datafusion.git", rev = "e0e423e9e29a711d076892865d51747b06429d1b"} # spiceai-45
-datafusion-physical-expr =  { git = "https://github.com/spiceai/datafusion.git", rev = "e0e423e9e29a711d076892865d51747b06429d1b"} # spiceai-45
-datafusion-physical-plan =  { git = "https://github.com/spiceai/datafusion.git", rev = "e0e423e9e29a711d076892865d51747b06429d1b"} # spiceai-45
-datafusion-proto =  { git = "https://github.com/spiceai/datafusion.git", rev = "e0e423e9e29a711d076892865d51747b06429d1b"} # spiceai-45
+datafusion = { git = "https://github.com/spiceai/datafusion.git", rev = "987d143985657065c8fc4d3d347a2010fbf36fe4" }               # spiceai-46
+datafusion-expr = { git = "https://github.com/spiceai/datafusion.git", rev = "987d143985657065c8fc4d3d347a2010fbf36fe4" }          # spiceai-46
+datafusion-physical-expr = { git = "https://github.com/spiceai/datafusion.git", rev = "987d143985657065c8fc4d3d347a2010fbf36fe4" } # spiceai-46
+datafusion-physical-plan = { git = "https://github.com/spiceai/datafusion.git", rev = "987d143985657065c8fc4d3d347a2010fbf36fe4" } # spiceai-46
+datafusion-proto = { git = "https://github.com/spiceai/datafusion.git", rev = "987d143985657065c8fc4d3d347a2010fbf36fe4" }         # spiceai-46

rust-toolchain.toml

@@ -0,0 +1,3 @@
+[toolchain]
+channel = "1.85.0"
+components = ["rustc", "cargo", "rustfmt", "clippy"] 

tests/duckdb/mod.rs

@@ -3,11 +3,12 @@ use arrow::array::RecordBatch;
 use arrow::datatypes::SchemaRef;
 use datafusion::catalog::TableProviderFactory;
 use datafusion::common::{Constraints, ToDFSchema};
+use datafusion::datasource::memory::MemorySourceConfig;
+use datafusion::datasource::source::DataSourceExec;
 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,8 +44,10 @@ 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 = DataSourceExec::new(Arc::new(
+        MemorySourceConfig::try_new(&[vec![arrow_record.clone()]], arrow_record.schema(), None)
+            .expect("memory source config created"),
+    ));
     let insert_plan = table_provider
         .insert_into(&ctx.state(), Arc::new(mem_exec), InsertOp::Append)
         .await
@@ -89,7 +92,8 @@ async fn arrow_duckdb_round_trip(
 #[case::struct_type(get_arrow_struct_record_batch(), "struct")]
 #[ignore] // DuckDB does not support Decimal256 / duckdb_arrow_scan failed to register view
 #[case::decimal(get_arrow_decimal_record_batch(), "decimal")]
-#[ignore] // Interval(DayTime) is not supported: / "Conversion Error: Could not convert Interval to Microsecond"
+#[ignore]
+// Interval(DayTime) is not supported: / "Conversion Error: Could not convert Interval to Microsecond"
 #[case::interval(get_arrow_interval_record_batch(), "interval")]
 #[ignore] // TimeUnit::Nanosecond is not correctly supported; written values are zeros
 #[case::duration(get_arrow_duration_record_batch(), "duration")]

tests/postgres/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::source::DataSourceExec,
+};
+use datafusion::{datasource::memory::MemorySourceConfig, execution::context::SessionContext};
 use datafusion_federation::schema_cast::record_convert::try_cast_to;
 use datafusion_table_providers::{
     postgres::{DynPostgresConnectionPool, PostgresTableProviderFactory},
@@ -53,8 +55,10 @@ 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 = DataSourceExec::new(Arc::new(
+        MemorySourceConfig::try_new(&[vec![arrow_record.clone()]], arrow_record.schema(), None)
+            .expect("memory source config created"),
+    ));
     let insert_plan = table_provider
         .insert_into(&ctx.state(), Arc::new(mem_exec), InsertOp::Append)
         .await
@@ -279,11 +283,7 @@ async fn test_postgres_jsonb_type(port: usize) {
     ('{"nested": {"key": "value"}}');
     "#;
 
-    let schema = Arc::new(Schema::new(vec![Field::new(
-        "data",
-        DataType::Utf8,
-        true,
-    )]));
+    let schema = Arc::new(Schema::new(vec![Field::new("data", DataType::Utf8, true)]));
 
     // Parse and re-serialize the JSON to ensure consistent ordering
     let expected_values = vec![
@@ -304,9 +304,7 @@ async fn test_postgres_jsonb_type(port: usize) {
 
     let expected_record = RecordBatch::try_new(
         Arc::clone(&schema),
-        vec![Arc::new(arrow::array::StringArray::from(
-            expected_values,
-        ))],
+        vec![Arc::new(arrow::array::StringArray::from(expected_values))],
     )
     .expect("Failed to create arrow record batch");
 

tests/sqlite/mod.rs

@@ -40,9 +40,12 @@ async fn arrow_sqlite_round_trip(
     // Create sqlite table from arrow records and insert arrow records
     let schema = Arc::clone(&arrow_record.schema());
     let create_table_stmts = CreateTableBuilder::new(schema, table_name).build_sqlite();
-    let insert_table_stmt = InsertBuilder::new(&TableReference::from(table_name), vec![arrow_record.clone()])
-        .build_sqlite(None)
-        .expect("SQLite insert statement should be constructed");
+    let insert_table_stmt = InsertBuilder::new(
+        &TableReference::from(table_name),
+        vec![arrow_record.clone()],
+    )
+    .build_sqlite(None)
+    .expect("SQLite insert statement should be constructed");
 
     // Test arrow -> Sqlite row coverage
     let _ = conn
@@ -59,8 +62,11 @@ async fn arrow_sqlite_round_trip(
 
     // Perform the test twice: first, simulate a request without a known schema;
     // then, test result conversion with a known projected schema (matching the test RecordBatch).
-    for projected_schema in vec![None, Some(arrow_record.schema())] {
-        if ctx.table_exist(table_name).expect("should be able to check if table exists") {
+    for projected_schema in [None, Some(arrow_record.schema())] {
+        if ctx
+            .table_exist(table_name)
+            .expect("should be able to check if table exists")
+        {
             ctx.deregister_table(table_name)
                 .expect("Table should be deregistered");
         }