Upgrade to datafusion-federation 0.4.2 (#360)

* Upgrade to latest datafusion-federation

* pin 1.1.3

* Use fixed datafusion-federation

* Fix table into

* Upgrade to DataFusion 47 and Arrow 55 (#355)

* Arrow enable FFI

* upgrade fed

* Update to official commit

---------

Co-authored-by: Sergei Grebnov <sergei.grebnov@gmail.com>

Author: phillipleblanc

Cargo.toml

@@ -6,7 +6,7 @@ edition = "2021"
 repository = "https://github.com/datafusion-contrib/datafusion-table-providers"
 
 [dependencies]
-arrow = "55"
+arrow = { version = "55", features = ["ffi"] }
 arrow-array = { version = "55", optional = true }
 arrow-flight = { version = "55", optional = true, features = ["flight-sql-experimental", "tls"] }
 arrow-schema = { version = "55", optional = true, features = ["serde"] }
@@ -25,7 +25,7 @@ datafusion-expr = { version = "47", optional = true }
 datafusion-physical-expr = { version = "47", optional = true }
 datafusion-physical-plan = { version = "47", optional = true }
 datafusion-proto = { version = "47", optional = true }
-duckdb = { version = "1", features = [
+duckdb = { version = "1.1.3", features = [
   "bundled",
   "r2d2",
   "vtab",
@@ -58,8 +58,7 @@ url = "2.5.1"
 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 = "dc11bd6edec00fb42ea48b8c5256970c8a533aa2" }
+datafusion-federation = { version = "0.4.2", features = ["sql"] }
 itertools = "0.13.0"
 dyn-clone = { version = "1.0.17", optional = true }
 geo-types = "0.7.13"
@@ -103,7 +102,7 @@ sqlite-federation = ["sqlite"]
 postgres-federation = ["postgres"]
 
 [patch.crates-io]
-datafusion-federation = { git = "https://github.com/spiceai/datafusion-federation.git", rev = "dc11bd6edec00fb42ea48b8c5256970c8a533aa2" } # spiceai-47
+datafusion-federation = { git = "https://github.com/spiceai/datafusion-federation.git", rev = "9db74a4b360df6be1bb554c59a474a2fd4bfb7e9" } # spiceai-47
 duckdb = { git = "https://github.com/spiceai/duckdb-rs.git", rev = "69ae7518ee093a1b070e9e4e6f011ef353431086" } # spiceai-1.1.3-backported-arrow-55
 
 datafusion = { git = "https://github.com/spiceai/datafusion.git", rev = "b5c62f29d2c70c5331ff50015b67b5e1cafcd578" }               # spiceai-47

src/duckdb/federation.rs

@@ -2,8 +2,8 @@ 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 datafusion::sql::unparser::dialect::Dialect;
+use datafusion_federation::sql::{SQLExecutor, SQLFederationProvider, SQLTableSource};
 use datafusion_federation::{FederatedTableProviderAdaptor, FederatedTableSource};
-use datafusion_federation_sql::{SQLExecutor, SQLFederationProvider, SQLTableSource};
 use futures::TryStreamExt;
 use snafu::ResultExt;
 
@@ -23,14 +23,14 @@ impl<T, P> DuckDBTable<T, P> {
     fn create_federated_table_source(
         self: Arc<Self>,
     ) -> DataFusionResult<Arc<dyn FederatedTableSource>> {
-        let table_name = self.base_table.table_reference.to_quoted_string();
+        let table_name = self.base_table.table_reference.clone();
         let schema = Arc::clone(&Arc::clone(&self).base_table.schema());
         let fed_provider = Arc::new(SQLFederationProvider::new(self));
         Ok(Arc::new(SQLTableSource::new_with_schema(
             fed_provider,
             table_name,
             schema,
-        )?))
+        )))
     }
 
     pub fn create_federated_table_provider(

src/mysql/federation.rs

@@ -4,8 +4,10 @@ use arrow::datatypes::SchemaRef;
 use async_trait::async_trait;
 use datafusion::sql::sqlparser::ast::{self, VisitMut};
 use datafusion::sql::unparser::dialect::Dialect;
+use datafusion_federation::sql::{
+    ast_analyzer::AstAnalyzer, SQLExecutor, SQLFederationProvider, SQLTableSource,
+};
 use datafusion_federation::{FederatedTableProviderAdaptor, FederatedTableSource};
-use datafusion_federation_sql::{AstAnalyzer, SQLExecutor, SQLFederationProvider, SQLTableSource};
 use futures::TryStreamExt;
 use snafu::ResultExt;
 use std::sync::Arc;
@@ -24,14 +26,14 @@ impl MySQLTable {
     fn create_federated_table_source(
         self: Arc<Self>,
     ) -> DataFusionResult<Arc<dyn FederatedTableSource>> {
-        let table_name = self.base_table.table_reference.to_quoted_string();
+        let table_name = self.base_table.table_reference.clone();
         let schema = Arc::clone(&Arc::clone(&self).base_table.schema());
         let fed_provider = Arc::new(SQLFederationProvider::new(self));
         Ok(Arc::new(SQLTableSource::new_with_schema(
             fed_provider,
             table_name,
             schema,
-        )?))
+        )))
     }
 
     pub fn create_federated_table_provider(
@@ -75,7 +77,7 @@ impl SQLExecutor for MySQLTable {
     }
 
     fn ast_analyzer(&self) -> Option<AstAnalyzer> {
-        Some(Box::new(mysql_ast_analyzer))
+        Some(AstAnalyzer::new(vec![Box::new(mysql_ast_analyzer)]))
     }
 
     fn execute(

src/sql/sql_provider_datafusion/federation.rs

@@ -1,7 +1,7 @@
 use crate::sql::db_connection_pool::{dbconnection::get_schema, JoinPushDown};
 use async_trait::async_trait;
+use datafusion_federation::sql::{SQLExecutor, SQLFederationProvider, SQLTableSource};
 use datafusion_federation::{FederatedTableProviderAdaptor, FederatedTableSource};
-use datafusion_federation_sql::{SQLExecutor, SQLFederationProvider, SQLTableSource};
 use futures::TryStreamExt;
 use snafu::prelude::*;
 use std::sync::Arc;
@@ -23,14 +23,14 @@ impl<T, P> SqlTable<T, P> {
     fn create_federated_table_source(
         self: Arc<Self>,
     ) -> DataFusionResult<Arc<dyn FederatedTableSource>> {
-        let table_name = self.table_reference.to_quoted_string();
+        let table_name = self.table_reference.clone();
         let schema = Arc::clone(&self.schema);
         let fed_provider = Arc::new(SQLFederationProvider::new(self));
         Ok(Arc::new(SQLTableSource::new_with_schema(
             fed_provider,
             table_name,
             schema,
-        )?))
+        )))
     }
 
     pub fn create_federated_table_provider(

src/sqlite/federation.rs

@@ -4,8 +4,11 @@ use arrow::datatypes::SchemaRef;
 use async_trait::async_trait;
 use datafusion::sql::sqlparser::ast::{self, VisitMut};
 use datafusion::sql::unparser::dialect::Dialect;
+use datafusion_federation::sql::ast_analyzer::AstAnalyzerRule;
+use datafusion_federation::sql::{
+    ast_analyzer::AstAnalyzer, SQLExecutor, SQLFederationProvider, SQLTableSource,
+};
 use datafusion_federation::{FederatedTableProviderAdaptor, FederatedTableSource};
-use datafusion_federation_sql::{AstAnalyzer, SQLExecutor, SQLFederationProvider, SQLTableSource};
 use futures::TryStreamExt;
 use snafu::ResultExt;
 use std::sync::Arc;
@@ -25,14 +28,14 @@ impl<T, P> SQLiteTable<T, P> {
     fn create_federated_table_source(
         self: Arc<Self>,
     ) -> DataFusionResult<Arc<dyn FederatedTableSource>> {
-        let table_name = self.base_table.table_reference.to_quoted_string();
+        let table_name = self.base_table.table_reference.clone();
         let schema = Arc::clone(&Arc::clone(&self).base_table.schema());
         let fed_provider = Arc::new(SQLFederationProvider::new(self));
         Ok(Arc::new(SQLTableSource::new_with_schema(
             fed_provider,
             table_name,
             schema,
-        )?))
+        )))
     }
 
     pub fn create_federated_table_provider(
@@ -45,7 +48,7 @@ impl<T, P> SQLiteTable<T, P> {
         ))
     }
 
-    fn sqlite_ast_analyzer(&self) -> AstAnalyzer {
+    fn sqlite_ast_analyzer(&self) -> AstAnalyzerRule {
         let decimal_between = self.decimal_between;
         Box::new(move |ast| {
             match ast {
@@ -85,7 +88,7 @@ impl<T, P> SQLExecutor for SQLiteTable<T, P> {
     }
 
     fn ast_analyzer(&self) -> Option<AstAnalyzer> {
-        Some(self.sqlite_ast_analyzer())
+        Some(AstAnalyzer::new(vec![self.sqlite_ast_analyzer()]))
     }
 
     fn execute(