Address PR review comments and fix clippy/test issues

- Fix redundant clone in handle_child_pushdown_result
- Fix missing cannot_federate field in TestExecutor instantiation
- Fix SqlRewriteTable::table_reference return type to MultiPartTableReference
- Prefix unused sql_query_rewriters with _ in DisplayAs
- Propagate errors from physical_plan.apply in rewriter test
- Update rewrite_table_scan.rs execute signature to include filters param
- Remove unused std::clone::Clone import in schema_cast/mod.rs
- Remove unused sqlparser::ast import in executor.rs
- Replace deprecated EliminateNestedUnion with OptimizeUnions
- Update test expectations for DataFusion v52.3.0 SQL unparser behavior

Author: lukekim

datafusion-federation/src/analyzer/mod.rs

@@ -3,7 +3,7 @@ mod scan_result;
 use crate::{FederatedTableProviderAdaptor, FederatedTableSource, FederationProviderRef};
 use crate::{FederationAnalyzerForLogicalPlan, FederationProvider};
 use datafusion::logical_expr::{col, expr::InSubquery, LogicalPlanBuilder};
-use datafusion::optimizer::eliminate_nested_union::EliminateNestedUnion;
+use datafusion::optimizer::optimize_unions::OptimizeUnions;
 use datafusion::optimizer::push_down_filter::PushDownFilter;
 use datafusion::optimizer::{Optimizer, OptimizerContext, OptimizerRule};
 use datafusion::{
@@ -74,7 +74,7 @@ impl FederationAnalyzerRule {
 
     pub fn default_optimizer_rules() -> Vec<Arc<dyn OptimizerRule + Send + Sync>> {
         vec![
-            Arc::new(EliminateNestedUnion::new()),
+            Arc::new(OptimizeUnions::new()),
             Arc::new(PushDownFilter::new()),
         ]
     }

datafusion-federation/src/schema_cast/mod.rs

@@ -13,7 +13,6 @@ use datafusion::physical_plan::{
 };
 use futures::StreamExt;
 use std::any::Any;
-use std::clone::Clone;
 use std::fmt;
 use std::sync::Arc;
 

datafusion-federation/src/sql/analyzer/rewrite_table_scan.rs

@@ -490,6 +490,7 @@ mod tests {
     use async_trait::async_trait;
     use datafusion::arrow::datatypes::{Schema, SchemaRef};
     use datafusion::execution::SendableRecordBatchStream;
+    use datafusion::physical_plan::PhysicalExpr;
     use datafusion::sql::unparser::dialect::Dialect;
     use datafusion::sql::unparser::plan_to_sql;
     use datafusion::{
@@ -520,7 +521,12 @@ mod tests {
             unimplemented!()
         }
 
-        fn execute(&self, _query: &str, _schema: SchemaRef) -> Result<SendableRecordBatchStream> {
+        fn execute(
+            &self,
+            _query: &str,
+            _schema: SchemaRef,
+            _filters: &[Arc<dyn PhysicalExpr>],
+        ) -> Result<SendableRecordBatchStream> {
             unimplemented!()
         }
 
@@ -876,7 +882,7 @@ mod tests {
             ),
             (
                 "SELECT foo.df_table.a FROM foo.df_table",
-                r#"SELECT a FROM "default".remote_table"#,
+                r#"SELECT remote_table.a FROM "default".remote_table"#,
             ),
             (
                 "SELECT MIN(a) FROM foo.df_table",

datafusion-federation/src/sql/executor.rs

@@ -6,7 +6,7 @@ use datafusion::{
     error::Result,
     logical_expr::LogicalPlan,
     physical_plan::{metrics::MetricsSet, PhysicalExpr, SendableRecordBatchStream},
-    sql::{sqlparser::ast, unparser::dialect::Dialect},
+    sql::unparser::dialect::Dialect,
 };
 use std::sync::Arc;
 

datafusion-federation/src/sql/mod.rs

@@ -22,7 +22,7 @@ use datafusion::{
     error::{DataFusionError, Result},
     execution::{context::SessionState, TaskContext},
     logical_expr::{Extension, LogicalPlan},
-    optimizer::{eliminate_nested_union::EliminateNestedUnion, Analyzer, AnalyzerRule, Optimizer},
+    optimizer::{optimize_unions::OptimizeUnions, Analyzer, AnalyzerRule, Optimizer},
     physical_expr::EquivalenceProperties,
     physical_plan::{
         execution_plan::{Boundedness, EmissionType},
@@ -51,7 +51,7 @@ use crate::{
 /// Returns a federation analyzer rule that is optimized for SQL federation.
 pub fn federation_analyzer_rule() -> FederationAnalyzerRule {
     FederationAnalyzerRule::new().with_optimizer(Optimizer::with_rules(vec![
-        Arc::new(EliminateNestedUnion::new()),
+        Arc::new(OptimizeUnions::new()),
         Arc::new(PushDownFilterFederation::new()),
         Arc::new(OptimizeProjectionsFederation::new()),
     ]))
@@ -323,7 +323,7 @@ impl DisplayAs for VirtualExecutionPlan {
             write!(f, " base_sql={statement}")?;
         }
 
-        let (logical_optimizers, ast_analyzers, sql_query_rewriters) = match gather_analyzers(&plan)
+        let (logical_optimizers, ast_analyzers, _sql_query_rewriters) = match gather_analyzers(&plan)
         {
             Ok(analyzers) => analyzers,
             Err(_) => return Ok(()),
@@ -427,7 +427,6 @@ impl ExecutionPlan for VirtualExecutionPlan {
         _config: &ConfigOptions,
     ) -> Result<FilterPushdownPropagation<Arc<dyn ExecutionPlan>>> {
         let parent_filters: Vec<_> = child_pushdown_result
-            .clone()
             .parent_filters
             .into_iter()
             .map(|f| f.filter)
@@ -588,7 +587,7 @@ mod tests {
             self
         }
 
-        fn table_reference(&self) -> TableReference {
+        fn table_reference(&self) -> MultiPartTableReference {
             self.table.table_reference().clone()
         }
 
@@ -914,7 +913,7 @@ mod tests {
         });
 
         let expected = vec![
-            r#"SELECT a, b, c FROM "default"."table" UNION ALL SELECT a, b, c FROM "default"."Table"(1) AS Table"#,
+            r#"SELECT "table".a, "table".b, "table".c FROM "default"."table" UNION ALL SELECT "Table".a, "Table".b, "Table".c FROM "default"."Table"(1) AS Table"#,
         ];
 
         assert_eq!(
@@ -929,6 +928,7 @@ mod tests {
     async fn sql_query_rewriter_hook_invoked_and_rewrites_sql() -> Result<(), DataFusionError> {
         let executor = TestExecutor {
             compute_context: "rewrite".into(),
+            cannot_federate: None,
         };
         let rewrite_calls = Arc::new(AtomicUsize::new(0));
         let table_ref = "table_with_rewriter".to_string();
@@ -955,7 +955,7 @@ mod tests {
         let physical_plan = ctx.state().create_physical_plan(&logical_plan).await?;
 
         let mut final_queries = vec![];
-        let _ = physical_plan.apply(|node| {
+        physical_plan.apply(|node| {
             if node.name() == "sql_federation_exec" {
                 let node = node
                     .as_any()
@@ -964,7 +964,7 @@ mod tests {
                 final_queries.push(node.final_sql()?);
             }
             Ok(TreeNodeRecursion::Continue)
-        });
+        })?;
 
         let [final_query] = final_queries.as_slice() else {
             panic!("expected a single federated SQL query");