Merge branch 'main' into lukim/rust-1.93

Author: lukekim

.github/workflows/docs.yml

@@ -41,6 +41,6 @@ jobs:
       name: github-pages
       url: ${{ steps.deployment.outputs.page_url }}
     steps:
-      - uses: actions/configure-pages@v5
-      - uses: actions/deploy-pages@v4
+      - uses: actions/configure-pages@v6
+      - uses: actions/deploy-pages@v5
         id: deployment

Cargo.toml

@@ -4,17 +4,17 @@ resolver = "2"
 members = ["datafusion-federation"]
 
 [workspace.package]
-version = "0.5.1"
+version = "0.5.3"
 edition = "2021"
 license = "Apache-2.0"
 readme = "README.md"
 repository = "https://github.com/datafusion-contrib/datafusion-federation"
 
 [workspace.dependencies]
-arrow-json = "57.2"
+arrow-json = "58.0"
 async-stream = "0.3"
 async-trait = "0.1"
-datafusion = "52.0"
-datafusion-federation = { path = "./datafusion-federation", version = "0.5.1" }
+datafusion = "53"
+datafusion-federation = { path = "./datafusion-federation", version = "0.5.3" }
 futures = "0.3"
 tokio = { version = "1.41", features = ["full"] }

datafusion-federation/CHANGELOG.md

@@ -7,6 +7,18 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [0.5.3](https://github.com/datafusion-contrib/datafusion-federation/compare/v0.5.2...v0.5.3) - 2026-03-30
+
+### Other
+
+- Fix `Explain Analyze` ([#168](https://github.com/datafusion-contrib/datafusion-federation/pull/168))
+
+## [0.5.2](https://github.com/datafusion-contrib/datafusion-federation/compare/v0.5.1...v0.5.2) - 2026-03-24
+
+### Other
+
+- Upgrade to DataFusion 53 ([#164](https://github.com/datafusion-contrib/datafusion-federation/pull/164))
+
 ## [0.5.1](https://github.com/datafusion-contrib/datafusion-federation/compare/v0.5.0...v0.5.1) - 2026-03-04
 
 ### Added

datafusion-federation/src/optimizer/mod.rs

@@ -191,14 +191,22 @@ impl FederationOptimizerRule {
                 return Ok((None, ScanResult::Distinct(provider)));
             }
 
-            let Some(optimizer) = provider.optimizer() else {
-                // No optimizer provided
-                return Ok((None, ScanResult::None));
-            };
+            // Analyze plans (EXPLAIN ANALYZE) cannot be converted to SQL by
+            // the Unparser, so they must not be federated as a whole. Only the
+            // inner query should be federated; DataFusion's AnalyzeExec will
+            // handle executing it and collecting metrics.
+            if matches!(plan, LogicalPlan::Analyze(_)) {
+                // Fall through to federate children instead.
+            } else {
+                let Some(optimizer) = provider.optimizer() else {
+                    // No optimizer provided
+                    return Ok((None, ScanResult::None));
+                };
 
-            // If this is the root plan node; federate the entire plan
-            let optimized = optimizer.optimize(plan.clone(), _config, |_, _| {})?;
-            return Ok((Some(optimized), ScanResult::None));
+                // If this is the root plan node; federate the entire plan
+                let optimized = optimizer.optimize(plan.clone(), _config, |_, _| {})?;
+                return Ok((Some(optimized), ScanResult::None));
+            }
         }
 
         // The plan is ambiguous; any input that is not yet optimized and has a

datafusion-federation/src/schema_cast/mod.rs

@@ -27,7 +27,7 @@ mod struct_cast;
 pub struct SchemaCastScanExec {
     input: Arc<dyn ExecutionPlan>,
     schema: SchemaRef,
-    properties: PlanProperties,
+    properties: Arc<PlanProperties>,
     metrics_set: ExecutionPlanMetricsSet,
 }
 
@@ -36,12 +36,12 @@ impl SchemaCastScanExec {
         let eq_properties = input.equivalence_properties().clone();
         let emission_type = input.pipeline_behavior();
         let boundedness = input.boundedness();
-        let properties = PlanProperties::new(
+        let properties = Arc::new(PlanProperties::new(
             eq_properties,
             input.output_partitioning().clone(),
             emission_type,
             boundedness,
-        );
+        ));
         Self {
             input,
             schema,
@@ -66,7 +66,7 @@ impl ExecutionPlan for SchemaCastScanExec {
         self
     }
 
-    fn properties(&self) -> &PlanProperties {
+    fn properties(&self) -> &Arc<PlanProperties> {
         &self.properties
     }
 

datafusion-federation/src/sql/analyzer.rs

@@ -5,7 +5,7 @@ use datafusion::{
     logical_expr::{
         expr::{
             AggregateFunction, AggregateFunctionParams, Alias, Exists, InList, InSubquery,
-            PlannedReplaceSelectItem, ScalarFunction, Sort, Unnest, WildcardOptions,
+            PlannedReplaceSelectItem, ScalarFunction, SetComparison, Sort, Unnest, WildcardOptions,
             WindowFunction, WindowFunctionParams,
         },
         Between, BinaryExpr, Case, Cast, Expr, GroupingSet, Like, Limit, LogicalPlan, Subquery,
@@ -563,6 +563,27 @@ fn rewrite_table_scans_in_expr(
             Ok(Expr::Unnest(Unnest::new(expr)))
         }
         Expr::ScalarVariable(_, _) | Expr::Literal(_, _) | Expr::Placeholder(_) => Ok(expr),
+        Expr::SetComparison(sc) => {
+            let expr = rewrite_table_scans_in_expr(*sc.expr, known_rewrites)?;
+            let subquery_plan = rewrite_table_scans(&sc.subquery.subquery, known_rewrites)?;
+            let outer_ref_columns = sc
+                .subquery
+                .outer_ref_columns
+                .into_iter()
+                .map(|e| rewrite_table_scans_in_expr(e, known_rewrites))
+                .collect::<Result<Vec<Expr>>>()?;
+            let subquery = Subquery {
+                subquery: Arc::new(subquery_plan),
+                outer_ref_columns,
+                spans: Spans::new(),
+            };
+            Ok(Expr::SetComparison(SetComparison::new(
+                Box::new(expr),
+                subquery,
+                sc.op,
+                sc.quantifier,
+            )))
+        }
     }
 }
 

datafusion-federation/src/sql/ast_analyzer.rs

@@ -97,6 +97,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![],
                     })

datafusion-federation/src/sql/mod.rs

@@ -164,20 +164,20 @@ impl FederationPlanner for SQLFederationPlanner {
 pub struct VirtualExecutionPlan {
     plan: LogicalPlan,
     executor: Arc<dyn SQLExecutor>,
-    props: PlanProperties,
+    props: Arc<PlanProperties>,
     statistics: Statistics,
     filters: Vec<Arc<dyn PhysicalExpr>>,
 }
 
 impl VirtualExecutionPlan {
     pub fn new(plan: LogicalPlan, executor: Arc<dyn SQLExecutor>, statistics: Statistics) -> Self {
         let schema: Schema = plan.schema().as_arrow().clone();
-        let props = PlanProperties::new(
+        let props = Arc::new(PlanProperties::new(
             EquivalenceProperties::new(Arc::new(schema)),
             Partitioning::UnknownPartitioning(1),
             EmissionType::Incremental,
             Boundedness::Bounded,
-        );
+        ));
         Self {
             plan,
             executor,
@@ -401,7 +401,7 @@ impl ExecutionPlan for VirtualExecutionPlan {
             .execute(&self.final_sql()?, self.schema(), &self.filters)
     }
 
-    fn properties(&self) -> &PlanProperties {
+    fn properties(&self) -> &Arc<PlanProperties> {
         &self.props
     }
 
@@ -766,6 +766,57 @@ mod tests {
         Ok(())
     }
 
+    /// EXPLAIN ANALYZE must not federate the Analyze wrapper — only the inner
+    /// query should be federated. Otherwise the SQL Unparser fails because it
+    /// cannot convert Analyze to SQL.
+    #[tokio::test]
+    async fn explain_analyze_not_federated() -> Result<(), DataFusionError> {
+        let executor = TestExecutor {
+            compute_context: "a".into(),
+        };
+
+        let table_ref = "test_table".to_string();
+        let table = get_test_table_provider(table_ref.clone(), executor);
+
+        let state = crate::default_session_state();
+        let ctx = SessionContext::new_with_state(state);
+        ctx.register_table(table_ref, table).unwrap();
+
+        let plan = ctx
+            .sql("EXPLAIN ANALYZE SELECT * FROM test_table")
+            .await?
+            .into_optimized_plan()?;
+
+        // The top-level node must be Analyze, not Federated.
+        assert!(
+            matches!(plan, LogicalPlan::Analyze(_)),
+            "Expected Analyze at root, got: {}",
+            plan.display_indent()
+        );
+
+        // The inner plan should contain a Federated extension node.
+        let mut found_federated = false;
+        plan.apply(|node| {
+            if let LogicalPlan::Extension(ext) = node {
+                if ext.node.name() == "Federated" {
+                    found_federated = true;
+                    return Ok(TreeNodeRecursion::Stop);
+                }
+            }
+            Ok(TreeNodeRecursion::Continue)
+        })?;
+        assert!(
+            found_federated,
+            "Expected a Federated node inside the Analyze plan"
+        );
+
+        // Physical planning should succeed (this is where it used to fail).
+        let physical_plan = ctx.state().create_physical_plan(&plan).await?;
+        assert_eq!(physical_plan.name(), "AnalyzeExec");
+
+        Ok(())
+    }
+
     #[tokio::test]
     async fn sql_query_rewriter_hook_invoked_and_rewrites_sql() -> Result<(), DataFusionError> {
         let executor = TestExecutor {