Upgrade to datafusion 51, arrow 57 (#497)

* Upgrade to datafusion 51, arrow 57

* Bump duckdb and r2d2_adbc

* Fix clippy

* Bump r2d2_adbc to 0.2.0

Author: nuno-faria

Cargo.lock

@@ -28,24 +28,24 @@ license = "Apache-2.0"
 description = "Extend the capabilities of DataFusion to support additional data sources via implementations of the `TableProvider` trait."
 
 [workspace.dependencies]
-arrow = "56.0.0"
-arrow-array = { version = "56.0.0" }
-arrow-flight = { version = "56.0.0", features = [
+arrow = "57.0.0"
+arrow-array = { version = "57.0.0" }
+arrow-flight = { version = "57.0.0", features = [
   "flight-sql-experimental",
   "tls-ring",
 ] }
-arrow-ipc = { version = "56.0.0" }
-arrow-schema = { version = "56.0.0", features = ["serde"] }
-arrow-json = "56.0.0"
-arrow-odbc = { version = "20.0.0" }
-datafusion = { version = "50", default-features = false }
-datafusion-expr = { version = "50" }
-datafusion-federation = { version = "0.4.10" }
-datafusion-ffi = { version = "50" }
-datafusion-proto = { version = "50" }
-datafusion-physical-expr = { version = "50" }
-datafusion-physical-plan = { version = "50" }
+arrow-ipc = { version = "57.0.0" }
+arrow-schema = { version = "57.0.0", features = ["serde"] }
+arrow-json = "57.0.0"
+arrow-odbc = { version = "21.0.0" }
+datafusion = { version = "51", default-features = false }
+datafusion-expr = { version = "51" }
+datafusion-federation = { version = "0.4.11" }
+datafusion-ffi = { version = "51" }
+datafusion-proto = { version = "51" }
+datafusion-physical-expr = { version = "51" }
+datafusion-physical-plan = { version = "51" }
 datafusion-table-providers = { path = "core" }
-duckdb = { version = "=1.3.1", package = "spiceai_duckdb_fork" } # Forked to add support for duckdb_scan_arrow, pending: https://github.com/duckdb/duckdb-rs/pull/488
-adbc_core = { version = "0.20.0" }
-adbc_driver_manager = { version = "0.20.0" }
+duckdb = { version = "=1.3.2", package = "spiceai_duckdb_fork" } # Forked to add support for duckdb_scan_arrow, pending: https://github.com/duckdb/duckdb-rs/pull/488
+adbc_core = { version = "0.21.0" }
+adbc_driver_manager = { version = "0.21.0" }

Cargo.toml

@@ -179,6 +179,7 @@ async fn run_benchmark(
         constraints: Constraints::new_unverified(vec![]),
         column_defaults: std::collections::HashMap::default(),
         temporary: false,
+        or_replace: false,
     };
 
     let ctx = SessionContext::new();

benches/sqlite_insert_benchmark.rs

@@ -30,7 +30,7 @@ byte-unit = { version = "5.1.4", optional = true }
 chrono = "0.4"
 clickhouse = { version = "0.13.3", optional = true }
 dashmap = "6.1.0"
-datafusion = { workspace = true, default-features = false }
+datafusion = { workspace = true, default-features = false, features = ["sql"]}
 datafusion-expr = { workspace = true, optional = true }
 datafusion-federation = { workspace = true, features = [
   "sql",
@@ -63,7 +63,7 @@ num-bigint = "0.4"
 odbc-api = { version = "19.0", optional = true }
 pem = { version = "3.0.4", optional = true }
 postgres-native-tls = { version = "0.5.0", optional = true }
-prost = { version = "0.14.1", optional = true }
+prost = { version = "=0.14.1", optional = true }
 rand = { version = "0.9" }
 regex = { version = "1" }
 r2d2 = { version = "0.8", optional = true }
@@ -91,7 +91,7 @@ tokio-postgres = { version = "0.7", features = [
   "with-geo-types-0_7",
 ], optional = true }
 tokio-rusqlite = { version = "0.7.0", optional = true }
-tonic = { version = "0.13", optional = true, features = [
+tonic = { version = "0.14", optional = true, features = [
   "tls-native-roots",
   "tls-webpki-roots",
 ] }
@@ -100,16 +100,16 @@ trust-dns-resolver = "0.23.2"
 url = "2.5.4"
 uuid = { version = "1.18", optional = true }
 rust_decimal = { version = "1.38.0", features = ["db-postgres"] }
-adbc_driver_manager = { workspace = true, version = "0.20.0", optional = true }
-adbc_core = { workspace = true, version = "0.20.0", optional = true }
-r2d2_adbc = { version = "0.1.0", optional = true }
+adbc_driver_manager = { workspace = true, optional = true }
+adbc_core = { workspace = true, optional = true }
+r2d2_adbc = { version = "0.2.0", optional = true }
 
 [dev-dependencies]
 anyhow = "1.0"
 bollard = "0.19"
 geozero = { version = "0.14.0", features = ["with-wkb"] }
 insta = { version = "1.43.2", features = ["filters"] }
-prost = { version = "=0.13.5" }
+prost = { version = "=0.14.1" }
 rand = "0.9"
 reqwest = "0.12"
 rstest = "0.26.1"

core/Cargo.toml

@@ -417,7 +417,7 @@ impl TableProviderFactory for DuckDBTableProviderFactory {
 
         let read_pool = read_pool.with_connection_setup_queries(local_settings);
 
-        let schema: SchemaRef = Arc::new(cmd.schema.as_ref().into());
+        let schema: SchemaRef = Arc::new(cmd.schema.as_ref().as_arrow().clone());
 
         let table_definition =
             TableDefinition::new(RelationName::new(name.clone()), Arc::clone(&schema))
@@ -722,6 +722,7 @@ pub(crate) mod tests {
             constraints: Constraints::default(),
             column_defaults: HashMap::new(),
             temporary: false,
+            or_replace: false,
         };
 
         let table_provider = factory
@@ -783,6 +784,7 @@ pub(crate) mod tests {
             constraints: Constraints::default(),
             column_defaults: HashMap::new(),
             temporary: false,
+            or_replace: false,
         };
 
         let table_provider = factory
@@ -840,6 +842,7 @@ pub(crate) mod tests {
             constraints: Constraints::default(),
             column_defaults: HashMap::new(),
             temporary: false,
+            or_replace: false,
         };
 
         let table_provider = factory
@@ -895,6 +898,7 @@ pub(crate) mod tests {
             constraints: Constraints::default(),
             column_defaults: HashMap::new(),
             temporary: false,
+            or_replace: false,
         };
 
         let table_provider = factory
@@ -953,6 +957,7 @@ pub(crate) mod tests {
             constraints: Constraints::default(),
             column_defaults: HashMap::new(),
             temporary: false,
+            or_replace: false,
         };
 
         let result = factory.create(&ctx.state(), &cmd).await;

core/src/duckdb.rs

@@ -22,7 +22,7 @@ use std::sync::Arc;
 use crate::flight::exec::{FlightConfig, FlightExec};
 use crate::flight::to_df_err;
 use datafusion::common::DataFusionError;
-use datafusion::logical_expr::registry::FunctionRegistry;
+use datafusion::execution::TaskContext;
 use datafusion::physical_plan::ExecutionPlan;
 use datafusion_proto::physical_plan::PhysicalExtensionCodec;
 
@@ -35,7 +35,7 @@ impl PhysicalExtensionCodec for FlightPhysicalCodec {
         &self,
         buf: &[u8],
         inputs: &[Arc<dyn ExecutionPlan>],
-        _registry: &dyn FunctionRegistry,
+        _registry: &TaskContext,
     ) -> datafusion::common::Result<Arc<dyn ExecutionPlan>> {
         if inputs.is_empty() {
             let config: FlightConfig = serde_json::from_slice(buf).map_err(to_df_err)?;

core/src/flight/codec.rs

@@ -188,7 +188,7 @@ impl TableProviderFactory for MySQLTableProviderFactory {
     ) -> datafusion::common::Result<Arc<dyn TableProvider>> {
         let name = cmd.name.to_string();
         let mut options = cmd.options.clone();
-        let schema: Schema = cmd.schema.as_ref().into();
+        let schema: Schema = cmd.schema.as_ref().as_arrow().clone();
 
         let indexes_option_str = options.remove("indexes");
         let unparsed_indexes: HashMap<String, IndexType> = match indexes_option_str {

core/src/mysql.rs

@@ -219,7 +219,7 @@ impl TableProviderFactory for PostgresTableProviderFactory {
 
         let name = cmd.name.clone();
         let mut options = cmd.options.clone();
-        let schema: Schema = cmd.schema.as_ref().into();
+        let schema: Schema = cmd.schema.as_ref().as_arrow().clone();
 
         let indexes_option_str = options.remove("indexes");
         let unparsed_indexes: HashMap<String, IndexType> = match indexes_option_str {

core/src/postgres.rs

@@ -311,7 +311,7 @@ impl TableProviderFactory for SqliteTableProviderFactory {
             )
         };
 
-        let schema: SchemaRef = Arc::new(cmd.schema.as_ref().into());
+        let schema: SchemaRef = Arc::new(cmd.schema.as_ref().as_arrow().clone());
         let schema: SchemaRef =
             SqliteConnection::handle_unsupported_schema(&schema, UnsupportedTypeAction::Error)
                 .map_err(|e| DataFusionError::External(e.into()))?;
@@ -1282,6 +1282,7 @@ pub(crate) mod tests {
             constraints: primary_keys_constraints,
             column_defaults: HashMap::default(),
             temporary: false,
+            or_replace: false,
         };
         let ctx = SessionContext::new();
         let table = SqliteTableProviderFactory::default()

core/src/sqlite.rs

@@ -324,6 +324,7 @@ mod tests {
             constraints: Constraints::default(),
             column_defaults: HashMap::default(),
             temporary: false,
+            or_replace: false,
         };
         let ctx = SessionContext::new();
         let table = SqliteTableProviderFactory::default()
@@ -466,6 +467,7 @@ mod tests {
             constraints: Constraints::default(),
             column_defaults: HashMap::default(),
             temporary: false,
+            or_replace: false,
         };
 
         let ctx = SessionContext::new();
@@ -772,6 +774,7 @@ mod tests {
             constraints: Constraints::default(),
             column_defaults: HashMap::default(),
             temporary: false,
+            or_replace: false,
         };
         let ctx = SessionContext::new();
         let table = SqliteTableProviderFactory::default()
@@ -816,6 +819,7 @@ mod tests {
             constraints: Constraints::default(),
             column_defaults: HashMap::default(),
             temporary: false,
+            or_replace: false,
         };
 
         let ctx = SessionContext::new();