Upgrade to DataFusion 53 (#559)

nuno-faria · web-flow · commit 00d1b280ef0f · 2026-04-19T18:44:49.000+09:00
* Upgrade to DataFusion 53

* Update dependencies

* Fix merge

* Update dependencies

* Update dependencies

* Fix clippy

* Update Rust toolchain

* Fix MongoDBExec

* Fix merge

* Update pyproject.toml

---------

Signed-off-by: Nuno Faria &lt;nunofpfaria@gmail.com&gt;
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/Cargo.toml b/Cargo.toml
@@ -28,25 +28,25 @@ license = "Apache-2.0"
 description = "Extend the capabilities of DataFusion to support additional data sources via implementations of the `TableProvider` trait."
 
 [workspace.dependencies]
-arrow = "57.2"
-arrow-array = { version = "57.2" }
-arrow-flight = { version = "57.2", features = [
+arrow = "58.0"
+arrow-array = { version = "58.0" }
+arrow-flight = { version = "58.0", features = [
     "flight-sql-experimental",
     "tls-ring",
 ] }
-arrow-ipc = { version = "57.2" }
-arrow-schema = { version = "57.2", features = ["serde"] }
-arrow-json = "57.2"
+arrow-ipc = { version = "58.0" }
+arrow-schema = { version = "58.0", features = ["serde"] }
+arrow-json = "58.0"
 arrow-odbc = { version = "23.1" }
-datafusion = { version = "52.0", default-features = false }
-datafusion-expr = { version = "52.0" }
-datafusion-federation = { version = "0.4" }
-datafusion-ffi = { version = "52.0" }
-datafusion-proto = { version = "52.0" }
-datafusion-physical-expr = { version = "52.0" }
-datafusion-physical-plan = { version = "52.0" }
-datafusion-python = { version = "52.0" }
+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-table-providers = { path = "core" }
-duckdb = { version = "=1.3", package = "spiceai_duckdb_fork" } # Forked to add support for duckdb_scan_arrow, pending: https://github.com/duckdb/duckdb-rs/pull/488
+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" }
 adbc_driver_manager = { version = "0.23" }
diff --git a/core/Cargo.toml b/core/Cargo.toml
@@ -46,10 +46,10 @@ bigdecimal = "0.4"
 byteorder = "1.5"
 bytes = { version = "1.11", optional = true }
 byte-unit = { version = "5.2", optional = true }
-chrono = "0.4"
+chrono = "0.4.44"
 clickhouse = { version = "0.14", optional = true }
 dashmap = "6.1"
-libduckdb-sys = { version = "=1.3", optional = true }
+libduckdb-sys = { version = "=1.4.4", optional = true }
 dyn-clone = { version = "1.0", optional = true }
 fallible-iterator = "0.3"
 fundu = "2.0"
@@ -69,7 +69,7 @@ num-traits = { version = "0.2", optional = true }
 odbc-api = { version = "24.1", optional = true }
 pem = { version = "3.0", optional = true }
 postgres-native-tls = { version = "0.5.0", optional = true }
-prost = { version = "=0.14", optional = true }
+prost = { version = "=0.14.3", optional = true }
 rand = { version = "0.9" }
 regex = { version = "1" }
 r2d2 = { version = "0.8", optional = true }
diff --git a/core/src/adbc/sql_table.rs b/core/src/adbc/sql_table.rs
@@ -163,7 +163,7 @@ impl<T: 'static, P: 'static> ExecutionPlan for AdbcSqlExec<T, P> {
         self.base_exec.schema()
     }
 
-    fn properties(&self) -> &PlanProperties {
+    fn properties(&self) -> &Arc<PlanProperties> {
         self.base_exec.properties()
     }
 
diff --git a/core/src/clickhouse/federation.rs b/core/src/clickhouse/federation.rs
@@ -3,6 +3,7 @@ use crate::sql::db_connection_pool::{DbConnectionPool, JoinPushDown};
 use crate::sql::sql_provider_datafusion::{get_stream, to_execution_error};
 use arrow::datatypes::SchemaRef;
 use async_trait::async_trait;
+use datafusion::physical_expr::PhysicalExpr;
 use datafusion::sql::unparser::dialect::Dialect;
 use datafusion_federation::sql::{
     AstAnalyzer, RemoteTableRef, SQLExecutor, SQLFederationProvider, SQLTableSource,
@@ -71,6 +72,7 @@ impl SQLExecutor for ClickHouseTable {
         &self,
         query: &str,
         schema: SchemaRef,
+        _filters: &[Arc<dyn PhysicalExpr>],
     ) -> DataFusionResult<SendableRecordBatchStream> {
         let fut = get_stream(self.pool.clone(), query.to_string(), Arc::clone(&schema));
         let stream = futures::stream::once(fut).try_flatten();
diff --git a/core/src/duckdb/federation.rs b/core/src/duckdb/federation.rs
@@ -1,6 +1,7 @@
 use crate::sql::db_connection_pool::dbconnection::{get_schema, Error as DbError};
 use crate::sql::sql_provider_datafusion::{get_stream, to_execution_error};
 use datafusion::arrow::datatypes::SchemaRef;
+use datafusion::physical_expr::PhysicalExpr;
 use datafusion::sql::unparser::dialect::Dialect;
 use datafusion_federation::sql::{
     RemoteTableRef, SQLExecutor, SQLFederationProvider, SQLTableSource,
@@ -64,6 +65,7 @@ impl<T, P> SQLExecutor for DuckDBTable<T, P> {
         &self,
         query: &str,
         schema: SchemaRef,
+        _filters: &[Arc<dyn PhysicalExpr>],
     ) -> DataFusionResult<SendableRecordBatchStream> {
         let fut = get_stream(
             self.base_table.clone_pool(),
diff --git a/core/src/duckdb/sql_table.rs b/core/src/duckdb/sql_table.rs
@@ -175,7 +175,7 @@ impl<T: 'static, P: 'static> ExecutionPlan for DuckSqlExec<T, P> {
         self.base_exec.schema()
     }
 
-    fn properties(&self) -> &PlanProperties {
+    fn properties(&self) -> &Arc<PlanProperties> {
         self.base_exec.properties()
     }
 
diff --git a/core/src/flight/exec.rs b/core/src/flight/exec.rs
@@ -48,7 +48,7 @@ use tonic::metadata::{AsciiMetadataKey, MetadataMap};
 #[derive(Clone, Debug)]
 pub(crate) struct FlightExec {
     config: FlightConfig,
-    plan_properties: PlanProperties,
+    plan_properties: Arc<PlanProperties>,
     metadata_map: Arc<MetadataMap>,
 }
 
@@ -90,12 +90,12 @@ impl From<FlightConfig> for FlightExec {
         } else {
             Boundedness::Bounded
         };
-        let plan_properties = PlanProperties::new(
+        let plan_properties = Arc::new(PlanProperties::new(
             EquivalenceProperties::new(config.schema.clone()),
             Partitioning::UnknownPartitioning(config.partitions.len()),
             EmissionType::Incremental,
             exec_mode,
-        );
+        ));
         let mut mm = MetadataMap::new();
         for (k, v) in config.properties.grpc_headers.iter() {
             let key = AsciiMetadataKey::from_str(k.as_str()).expect("invalid header name");
@@ -292,7 +292,7 @@ impl ExecutionPlan for FlightExec {
         self
     }
 
-    fn properties(&self) -> &PlanProperties {
+    fn properties(&self) -> &Arc<PlanProperties> {
         &self.plan_properties
     }
 
diff --git a/core/src/mongodb/table.rs b/core/src/mongodb/table.rs
@@ -106,7 +106,7 @@ struct MongoDBExec {
     filters_doc: Document,
     sort_doc: Document,
     limit: Option<i32>,
-    properties: PlanProperties,
+    properties: Arc<PlanProperties>,
 }
 
 impl MongoDBExec {
@@ -159,12 +159,12 @@ impl MongoDBExec {
             filters_doc: mongo_filters_doc,
             sort_doc: Document::new(),
             limit,
-            properties: PlanProperties::new(
+            properties: Arc::new(PlanProperties::new(
                 EquivalenceProperties::new(projected_schema),
                 Partitioning::UnknownPartitioning(1),
                 EmissionType::Final,
                 Boundedness::Bounded,
-            ),
+            )),
         })
     }
 }
@@ -213,7 +213,7 @@ impl ExecutionPlan for MongoDBExec {
         Arc::clone(&self.projected_schema)
     }
 
-    fn properties(&self) -> &PlanProperties {
+    fn properties(&self) -> &Arc<PlanProperties> {
         &self.properties
     }
 
@@ -259,7 +259,8 @@ impl ExecutionPlan for MongoDBExec {
             Arc::clone(&self.projected_schema),
             vec![order.to_vec()],
         );
-        new_exec.properties = new_exec.properties.with_eq_properties(eq_properties);
+        new_exec.properties =
+            Arc::new(PlanProperties::clone(&new_exec.properties).with_eq_properties(eq_properties));
 
         // Return Inexact rather than Exact so DataFusion keeps the SortExec wrapper
         // above us. Exact would replace the SortExec with `inner`, which loses the
diff --git a/core/src/mysql/federation.rs b/core/src/mysql/federation.rs
@@ -2,6 +2,7 @@ use crate::sql::db_connection_pool::dbconnection::{get_schema, Error as DbError}
 use crate::sql::sql_provider_datafusion::{get_stream, to_execution_error};
 use arrow::datatypes::SchemaRef;
 use async_trait::async_trait;
+use datafusion::physical_expr::PhysicalExpr;
 use datafusion::sql::sqlparser::ast::{self, VisitMut};
 use datafusion::sql::unparser::dialect::Dialect;
 use datafusion_federation::sql::{
@@ -84,6 +85,7 @@ impl SQLExecutor for MySQLTable {
         &self,
         query: &str,
         schema: SchemaRef,
+        _filters: &[Arc<dyn PhysicalExpr>],
     ) -> DataFusionResult<SendableRecordBatchStream> {
         let fut = get_stream(
             self.base_table.clone_pool(),
diff --git a/core/src/mysql/sql_table.rs b/core/src/mysql/sql_table.rs
@@ -167,7 +167,7 @@ impl ExecutionPlan for MySQLSQLExec {
         self.base_exec.schema()
     }
 
-    fn properties(&self) -> &PlanProperties {
+    fn properties(&self) -> &Arc<PlanProperties> {
         self.base_exec.properties()
     }
 
diff --git a/core/src/sql/sql_provider_datafusion/federation.rs b/core/src/sql/sql_provider_datafusion/federation.rs
@@ -1,5 +1,6 @@
 use crate::sql::db_connection_pool::{dbconnection::get_schema, JoinPushDown};
 use async_trait::async_trait;
+use datafusion::physical_expr::PhysicalExpr;
 use datafusion_federation::sql::{
     RemoteTableRef, SQLExecutor, SQLFederationProvider, SQLTableSource,
 };
@@ -81,6 +82,7 @@ impl<T, P> SQLExecutor for SqlTable<T, P> {
         &self,
         query: &str,
         schema: SchemaRef,
+        _filters: &[Arc<dyn PhysicalExpr>],
     ) -> DataFusionResult<SendableRecordBatchStream> {
         let fut = get_stream(
             Arc::clone(&self.pool),
diff --git a/core/src/sql/sql_provider_datafusion/mod.rs b/core/src/sql/sql_provider_datafusion/mod.rs
@@ -283,7 +283,7 @@ pub struct SqlExec<T, P> {
     projected_schema: SchemaRef,
     pool: Arc<dyn DbConnectionPool<T, P> + Send + Sync>,
     sql: String,
-    properties: PlanProperties,
+    properties: Arc<PlanProperties>,
     dialect: Arc<dyn Dialect + Send + Sync>,
 }
 
@@ -313,12 +313,12 @@ impl<T, P> SqlExec<T, P> {
             projected_schema: Arc::clone(&projected_schema),
             pool,
             sql,
-            properties: PlanProperties::new(
+            properties: Arc::new(PlanProperties::new(
                 EquivalenceProperties::new(projected_schema),
                 Partitioning::UnknownPartitioning(1),
                 EmissionType::Incremental,
                 Boundedness::Bounded,
-            ),
+            )),
             dialect,
         })
     }
@@ -495,7 +495,7 @@ impl<T: 'static, P: 'static> ExecutionPlan for SqlExec<T, P> {
         Arc::clone(&self.projected_schema)
     }
 
-    fn properties(&self) -> &PlanProperties {
+    fn properties(&self) -> &Arc<PlanProperties> {
         &self.properties
     }
 
@@ -578,7 +578,8 @@ impl<T: 'static, P: 'static> ExecutionPlan for SqlExec<T, P> {
             Arc::clone(&self.projected_schema),
             vec![order.to_vec()],
         );
-        new_exec.properties = new_exec.properties.with_eq_properties(eq_properties);
+        new_exec.properties =
+            Arc::new(PlanProperties::clone(&new_exec.properties).with_eq_properties(eq_properties));
 
         // Return Inexact rather than Exact so DataFusion keeps the SortExec wrapper
         // above us. Exact would replace the SortExec with `inner`, which loses the
diff --git a/core/src/sqlite/federation.rs b/core/src/sqlite/federation.rs
@@ -2,6 +2,7 @@ use crate::sql::db_connection_pool::dbconnection::{get_schema, Error as DbError}
 use crate::sql::sql_provider_datafusion::{get_stream, to_execution_error};
 use async_trait::async_trait;
 use datafusion::arrow::datatypes::SchemaRef;
+use datafusion::physical_expr::PhysicalExpr;
 use datafusion::sql::sqlparser::ast::{self, VisitMut};
 use datafusion::sql::unparser::dialect::Dialect;
 use datafusion_federation::sql::{
@@ -86,6 +87,7 @@ impl<T, P> SQLExecutor for SQLiteTable<T, P> {
         &self,
         query: &str,
         schema: SchemaRef,
+        _filters: &[Arc<dyn PhysicalExpr>],
     ) -> DataFusionResult<SendableRecordBatchStream> {
         let fut = get_stream(
             self.base_table.clone_pool(),
diff --git a/core/src/sqlite/sql_table.rs b/core/src/sqlite/sql_table.rs
@@ -155,7 +155,7 @@ impl<T: 'static, P: 'static> ExecutionPlan for SQLiteSqlExec<T, P> {
         self.base_exec.schema()
     }
 
-    fn properties(&self) -> &PlanProperties {
+    fn properties(&self) -> &Arc<PlanProperties> {
         self.base_exec.properties()
     }
 
diff --git a/core/src/sqlite/sqlite_interval.rs b/core/src/sqlite/sqlite_interval.rs
@@ -253,6 +253,7 @@ impl SQLiteIntervalVisitor {
             data_type: ast::DataType::Text,
             format: None,
             kind: ast::CastKind::Cast,
+            array: false,
         }
     }
 
@@ -420,6 +421,7 @@ mod test {
             data_type: ast::DataType::Text,
             format: None,
             kind: ast::CastKind::Cast,
+            array: false,
         };
 
         assert_eq!(datetime_function, expected);
@@ -470,6 +472,7 @@ mod test {
             data_type: ast::DataType::Text,
             format: None,
             kind: ast::CastKind::Cast,
+            array: false,
         };
 
         assert_eq!(datetime_function, expected);
diff --git a/core/src/util/table_arg_replace.rs b/core/src/util/table_arg_replace.rs
@@ -74,6 +74,7 @@ impl VisitorMut for TableArgReplace {
                 *args = Some(arg.clone());
                 if alias.is_none() {
                     *alias = Some(TableAlias {
+                        explicit: true,
                         name: Ident::new(table.table()),
                         columns: vec![],
                     })
diff --git a/core/src/util/test.rs b/core/src/util/test.rs
@@ -24,7 +24,7 @@ pub struct MockExec {
     /// if true (the default), sends data using a separate task to ensure the
     /// batches are not available without this stream yielding first
     use_task: bool,
-    cache: PlanProperties,
+    cache: Arc<PlanProperties>,
 }
 
 impl MockExec {
@@ -36,7 +36,7 @@ impl MockExec {
     /// ensure any poll loops are correct. This behavior can be
     /// changed with `with_use_task`
     pub fn new(data: Vec<Result<RecordBatch>>, schema: SchemaRef) -> Self {
-        let cache = Self::compute_properties(Arc::clone(&schema));
+        let cache = Arc::new(Self::compute_properties(Arc::clone(&schema)));
         Self {
             data,
             schema,
@@ -87,7 +87,7 @@ impl ExecutionPlan for MockExec {
         self
     }
 
-    fn properties(&self) -> &PlanProperties {
+    fn properties(&self) -> &Arc<PlanProperties> {
         &self.cache
     }
 
@@ -149,7 +149,7 @@ impl ExecutionPlan for MockExec {
     }
 
     // Panics if one of the batches is an error
-    fn statistics(&self) -> Result<Statistics> {
+    fn partition_statistics(&self, _partition: Option<usize>) -> Result<Statistics> {
         let data: Result<Vec<_>> = self
             .data
             .iter()
diff --git a/python/Cargo.toml b/python/Cargo.toml
@@ -22,7 +22,7 @@ datafusion-python = { workspace = true }
 datafusion-table-providers = { workspace = true }
 duckdb = { workspace = true, optional = true }
 
-pyo3 = { version = "0.26" }
+pyo3 = { version = "0.28" }
 tokio = { version = "1.51", features = [
     "macros",
     "rt",
diff --git a/python/pyproject.toml b/python/pyproject.toml
@@ -24,7 +24,7 @@ classifier = [
     "Programming Language :: Python",
     "Programming Language :: Rust",
 ]
-dependencies = ["datafusion>=52.0.0,<53.0.0"]
+dependencies = ["datafusion>=53.0.0,<54.0.0"]
 
 [project.urls]
 repository = "https://github.com/datafusion-contrib/datafusion-table-providers"
diff --git a/python/src/clickhouse.rs b/python/src/clickhouse.rs
@@ -63,7 +63,7 @@ impl RawClickHouseTableFactory {
         args: Option<Py<PyAny>>,
     ) -> PyResult<RawTableProvider> {
         let args_vec = if let Some(args) = args {
-            let seq = args.downcast_bound::<PyList>(py).map_err(|_| {
+            let seq = args.cast_bound::<PyList>(py).map_err(|_| {
                 PyTypeError::new_err("Argument must be list of int (signed/unsigned) or string")
             })?;
 
diff --git a/rust-toolchain.toml b/rust-toolchain.toml
@@ -1,3 +1,3 @@
 [toolchain]
-channel = "1.92.0"
+channel = "1.93.0"
 components = ["rustc", "cargo", "rustfmt", "clippy"] 

Original file line number	Diff line number	Diff line change
`@@ -163,7 +163,7 @@ impl<T: 'static, P: 'static> ExecutionPlan for AdbcSqlExec<T, P> {`
`163`	`163`	`self.base_exec.schema()`
`164`	`164`	`}`
`165`	`165`
`166`		`- fn properties(&self) -> &PlanProperties {`
	`166`	`+ fn properties(&self) -> &Arc<PlanProperties> {`
`167`	`167`	`self.base_exec.properties()`
`168`	`168`	`}`
`169`	`169`
Original file line number	Diff line number	Diff line change
`@@ -175,7 +175,7 @@ impl<T: 'static, P: 'static> ExecutionPlan for DuckSqlExec<T, P> {`
`175`	`175`	`self.base_exec.schema()`
`176`	`176`	`}`
`177`	`177`
`178`		`- fn properties(&self) -> &PlanProperties {`
	`178`	`+ fn properties(&self) -> &Arc<PlanProperties> {`
`179`	`179`	`self.base_exec.properties()`
`180`	`180`	`}`
`181`	`181`