Upgrade to Datafusion 54 (#675)

* Upgrade to Datafusion 54

* cargo fmt

* Switch to datafusion-python 54

Hope fully fixes segfault

---------

Co-authored-by: Phillip LeBlanc <879445+phillipleblanc@users.noreply.github.com>

Author: simonvandel

Cargo.lock

@@ -38,14 +38,14 @@ arrow-ipc = { version = "58.0" }
 arrow-schema = { version = "58.0", features = ["serde"] }
 arrow-json = "58.0"
 arrow-odbc = { version = "25.2" }
-datafusion = { version = "53.0", default-features = false }
-datafusion-expr = { version = "53.0" }
-datafusion-federation = { version = "0.5.3" }
-datafusion-ffi = { version = "53.0" }
-datafusion-proto = { version = "53.0" }
-datafusion-physical-expr = { version = "53.0" }
-datafusion-physical-plan = { version = "53.0" }
-datafusion-python = { version = "53.0" }
+datafusion = { version = "54.0", default-features = false }
+datafusion-expr = { version = "54.0" }
+datafusion-federation = { version = "0.5.5" }
+datafusion-ffi = { version = "54.0" }
+datafusion-proto = { version = "54.0" }
+datafusion-physical-expr = { version = "54.0" }
+datafusion-physical-plan = { version = "54.0" }
+datafusion-python = { version = "54.0" }
 datafusion-table-providers = { path = "core" }
 duckdb = { version = "=1.4.4", 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.23" }

Cargo.toml

@@ -18,7 +18,6 @@ use futures::TryStreamExt;
 use crate::sql::sql_provider_datafusion::{
     get_stream, to_execution_error, Result as SqlResult, SqlExec, SqlTable,
 };
-use std::any::Any;
 use std::sync::Arc;
 
 use datafusion::catalog::Session;
@@ -84,10 +83,6 @@ impl<T, P> AdbcDBTable<T, P> {
 
 #[async_trait]
 impl<T, P> TableProvider for AdbcDBTable<T, P> {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn schema(&self) -> SchemaRef {
         self.base_table.schema()
     }
@@ -155,10 +150,6 @@ impl<T: 'static, P: 'static> ExecutionPlan for AdbcSqlExec<T, P> {
         "AdbcSqlExec"
     }
 
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn schema(&self) -> SchemaRef {
         self.base_exec.schema()
     }
@@ -185,7 +176,6 @@ impl<T: 'static, P: 'static> ExecutionPlan for AdbcSqlExec<T, P> {
         match self.base_exec.try_pushdown_sort(order)? {
             SortOrderPushdownResult::Exact { inner } => {
                 let base_exec = inner
-                    .as_any()
                     .downcast_ref::<SqlExec<T, P>>()
                     .ok_or_else(|| {
                         DataFusionError::Internal(
@@ -199,7 +189,6 @@ impl<T: 'static, P: 'static> ExecutionPlan for AdbcSqlExec<T, P> {
             }
             SortOrderPushdownResult::Inexact { inner } => {
                 let base_exec = inner
-                    .as_any()
                     .downcast_ref::<SqlExec<T, P>>()
                     .ok_or_else(|| {
                         DataFusionError::Internal(
@@ -227,7 +216,6 @@ impl<T: 'static, P: 'static> ExecutionPlan for AdbcSqlExec<T, P> {
         let base_exec = self
             .base_exec
             .with_fetch(limit)?
-            .as_any()
             .downcast_ref::<SqlExec<T, P>>()?
             .clone();
         Some(Arc::new(AdbcSqlExec { base_exec }))
@@ -246,7 +234,6 @@ impl<T: 'static, P: 'static> ExecutionPlan for AdbcSqlExec<T, P> {
             .updated_node
             .map(|node| {
                 let base_exec = node
-                    .as_any()
                     .downcast_ref::<SqlExec<T, P>>()
                     .ok_or_else(|| {
                         DataFusionError::Internal(

core/src/adbc/sql_table.rs

@@ -35,7 +35,7 @@ use datafusion::{
 };
 use futures::StreamExt;
 use snafu::ResultExt;
-use std::{any::Any, fmt, sync::Arc};
+use std::{fmt, sync::Arc};
 use tokio::sync::mpsc::{self, Receiver, Sender};
 use tokio::task::JoinHandle;
 
@@ -132,10 +132,6 @@ where
     D: Database + Send + 'static,
     D::ConnectionType: Connection + Send + Sync,
 {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn schema(&self) -> SchemaRef {
         self.read_provider.schema()
     }
@@ -221,10 +217,6 @@ where
     D: Database + Send + 'static,
     D::ConnectionType: Connection + Send + Sync,
 {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn metrics(&self) -> Option<MetricsSet> {
         None
     }

core/src/adbc/write.rs

@@ -4,9 +4,9 @@ use datafusion::logical_expr::{LogicalPlan, LogicalPlanBuilder, LogicalTableSour
 use datafusion::physical_plan::ExecutionPlan;
 use datafusion::sql::sqlparser::ast::VisitMut;
 use datafusion::sql::unparser::Unparser;
+use std::fmt;
 use std::fmt::Display;
 use std::sync::Arc;
-use std::{any::Any, fmt};
 
 use datafusion::{
     arrow::datatypes::SchemaRef,
@@ -55,10 +55,6 @@ impl ClickHouseTable {
 
 #[async_trait]
 impl TableProvider for ClickHouseTable {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn schema(&self) -> SchemaRef {
         self.schema.clone()
     }

core/src/clickhouse/sql_table.rs

@@ -1,4 +1,4 @@
-use std::{any::Any, sync::Arc};
+use std::sync::Arc;
 
 use crate::sql::db_connection_pool::dbconnection::{get_schemas, get_tables};
 use crate::sql::db_connection_pool::DbConnectionPool;
@@ -38,10 +38,6 @@ impl DatabaseCatalogProvider {
 }
 
 impl CatalogProvider for DatabaseCatalogProvider {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn schema_names(&self) -> Vec<String> {
         self.schemas.iter().map(|s| s.key().clone()).collect()
     }
@@ -74,10 +70,6 @@ impl<T, P: 'static> DatabaseSchemaProvider<T, P> {
 
 #[async_trait]
 impl<T: 'static, P: 'static> SchemaProvider for DatabaseSchemaProvider<T, P> {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn table_names(&self) -> Vec<String> {
         self.tables.clone()
     }

core/src/common.rs

@@ -731,7 +731,6 @@ pub(crate) mod tests {
             .expect("table provider created");
 
         let writer = table_provider
-            .as_any()
             .downcast_ref::<DuckDBTableWriter>()
             .expect("cast to DuckDBTableWriter");
 
@@ -793,7 +792,6 @@ pub(crate) mod tests {
             .expect("table provider created");
 
         let writer = table_provider
-            .as_any()
             .downcast_ref::<DuckDBTableWriter>()
             .expect("cast to DuckDBTableWriter");
 
@@ -851,7 +849,6 @@ pub(crate) mod tests {
             .expect("table provider created");
 
         let writer = table_provider
-            .as_any()
             .downcast_ref::<DuckDBTableWriter>()
             .expect("cast to DuckDBTableWriter");
 
@@ -907,7 +904,6 @@ pub(crate) mod tests {
             .expect("table provider created");
 
         let writer = table_provider
-            .as_any()
             .downcast_ref::<DuckDBTableWriter>()
             .expect("cast to DuckDBTableWriter");
 

core/src/duckdb.rs

@@ -5,7 +5,7 @@ use datafusion::sql::unparser::dialect::Dialect;
 use futures::TryStreamExt;
 use std::collections::HashMap;
 use std::fmt::Display;
-use std::{any::Any, fmt, sync::Arc};
+use std::{fmt, sync::Arc};
 
 use crate::sql::sql_provider_datafusion::{
     get_stream, to_execution_error, Result as SqlResult, SqlExec, SqlTable,
@@ -78,10 +78,6 @@ impl<T, P> DuckDBTable<T, P> {
 
 #[async_trait]
 impl<T, P> TableProvider for DuckDBTable<T, P> {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn schema(&self) -> SchemaRef {
         self.base_table.schema()
     }
@@ -167,10 +163,6 @@ impl<T: 'static, P: 'static> ExecutionPlan for DuckSqlExec<T, P> {
         "DuckSqlExec"
     }
 
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn schema(&self) -> SchemaRef {
         self.base_exec.schema()
     }
@@ -197,7 +189,6 @@ impl<T: 'static, P: 'static> ExecutionPlan for DuckSqlExec<T, P> {
         match self.base_exec.try_pushdown_sort(order)? {
             SortOrderPushdownResult::Exact { inner } => {
                 let base_exec = inner
-                    .as_any()
                     .downcast_ref::<SqlExec<T, P>>()
                     .ok_or_else(|| {
                         DataFusionError::Internal(
@@ -214,7 +205,6 @@ impl<T: 'static, P: 'static> ExecutionPlan for DuckSqlExec<T, P> {
             }
             SortOrderPushdownResult::Inexact { inner } => {
                 let base_exec = inner
-                    .as_any()
                     .downcast_ref::<SqlExec<T, P>>()
                     .ok_or_else(|| {
                         DataFusionError::Internal(
@@ -245,7 +235,6 @@ impl<T: 'static, P: 'static> ExecutionPlan for DuckSqlExec<T, P> {
         let base_exec = self
             .base_exec
             .with_fetch(limit)?
-            .as_any()
             .downcast_ref::<SqlExec<T, P>>()?
             .clone();
         Some(Arc::new(DuckSqlExec {
@@ -267,7 +256,6 @@ impl<T: 'static, P: 'static> ExecutionPlan for DuckSqlExec<T, P> {
             .updated_node
             .map(|node| {
                 let base_exec = node
-                    .as_any()
                     .downcast_ref::<SqlExec<T, P>>()
                     .ok_or_else(|| {
                         DataFusionError::Internal(

core/src/duckdb/sql_table.rs

@@ -1,5 +1,5 @@
 use std::time::{SystemTime, UNIX_EPOCH};
-use std::{any::Any, fmt, sync::Arc};
+use std::{fmt, sync::Arc};
 
 use crate::duckdb::DuckDB;
 use crate::sql::db_connection_pool::duckdbpool::DuckDbConnectionPool;
@@ -135,10 +135,6 @@ impl DuckDBTableWriter {
 
 #[async_trait]
 impl TableProvider for DuckDBTableWriter {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn schema(&self) -> SchemaRef {
         self.read_provider.schema()
     }
@@ -194,10 +190,6 @@ pub(crate) struct DuckDBDataSink {
 
 #[async_trait]
 impl DataSink for DuckDBDataSink {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn metrics(&self) -> Option<MetricsSet> {
         None
     }

core/src/duckdb/write.rs

@@ -18,7 +18,6 @@
 //! Generic [FlightTableFactory] that can connect to Arrow Flight services,
 //! with a [sql::FlightSqlDriver] provided out-of-the-box.
 
-use std::any::Any;
 use std::collections::HashMap;
 use std::error::Error;
 use std::fmt::Debug;
@@ -318,10 +317,6 @@ impl std::fmt::Debug for FlightTable {
 
 #[async_trait]
 impl TableProvider for FlightTable {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn schema(&self) -> SchemaRef {
         self.logical_schema.clone()
     }