fix: amend RecursiveQuery structure and schema handling

- Remove schema field from RecursiveQuery
- Update recompute_schema() to re-widen recursive CTE children as necessary
- Retain widened schema through child projections post recompute drift
- Adjust logical-plan SLT expectation for newly inserted projection

Author: kosiew

datafusion/expr/src/logical_plan/builder.rs

@@ -215,13 +215,11 @@ impl LogicalPlanBuilder {
             Arc::unwrap_or_clone(self.plan),
             Arc::clone(&output_schema),
         )?;
-        let recursive_term =
-            plan_with_schema(coerced_recursive_term, Arc::clone(&output_schema))?;
+        let recursive_term = plan_with_schema(coerced_recursive_term, output_schema)?;
         Ok(Self::from(LogicalPlan::RecursiveQuery(RecursiveQuery {
             name,
             static_term: Arc::new(static_term),
             recursive_term: Arc::new(recursive_term),
-            schema: output_schema,
             is_distinct,
         })))
     }

datafusion/expr/src/logical_plan/plan.rs

@@ -54,6 +54,7 @@ use arrow::datatypes::{DataType, Field, FieldRef, Schema, SchemaRef};
 use datafusion_common::cse::{NormalizeEq, Normalizeable};
 use datafusion_common::format::ExplainFormat;
 use datafusion_common::metadata::check_metadata_with_storage_equal;
+use datafusion_common::recursive_schema::recursive_query_output_schema;
 use datafusion_common::tree_node::{
     Transformed, TreeNode, TreeNodeContainer, TreeNodeRecursion,
 };
@@ -353,7 +354,11 @@ impl LogicalPlan {
             LogicalPlan::Copy(CopyTo { output_schema, .. }) => output_schema,
             LogicalPlan::Ddl(ddl) => ddl.schema(),
             LogicalPlan::Unnest(Unnest { schema, .. }) => schema,
-            LogicalPlan::RecursiveQuery(RecursiveQuery { schema, .. }) => schema,
+            LogicalPlan::RecursiveQuery(RecursiveQuery { static_term, .. }) => {
+                // The static term is coerced to the recursive output schema when
+                // building a RecursiveQuery.
+                static_term.schema()
+            }
         }
     }
 
@@ -737,7 +742,33 @@ impl LogicalPlan {
                 };
                 Ok(LogicalPlan::Distinct(distinct))
             }
-            LogicalPlan::RecursiveQuery(_) => Ok(self),
+            LogicalPlan::RecursiveQuery(RecursiveQuery {
+                name,
+                static_term,
+                recursive_term,
+                is_distinct,
+            }) => {
+                let output_schema = recursive_query_output_schema(
+                    static_term.schema(),
+                    recursive_term.schema(),
+                )?;
+                if output_schema == *static_term.schema() {
+                    Ok(LogicalPlan::RecursiveQuery(RecursiveQuery {
+                        name,
+                        static_term,
+                        recursive_term,
+                        is_distinct,
+                    }))
+                } else {
+                    LogicalPlanBuilder::from(Arc::unwrap_or_clone(static_term))
+                        .to_recursive_query(
+                            name,
+                            Arc::unwrap_or_clone(recursive_term),
+                            is_distinct,
+                        )?
+                        .build()
+                }
+            }
             LogicalPlan::Analyze(_) => Ok(self),
             LogicalPlan::Explain(_) => Ok(self),
             LogicalPlan::TableScan(_) => Ok(self),
@@ -1073,20 +1104,13 @@ impl LogicalPlan {
                 Ok(LogicalPlan::Distinct(distinct))
             }
             LogicalPlan::RecursiveQuery(RecursiveQuery {
-                name,
-                schema,
-                is_distinct,
-                ..
+                name, is_distinct, ..
             }) => {
                 self.assert_no_expressions(expr)?;
                 let (static_term, recursive_term) = self.only_two_inputs(inputs)?;
-                Ok(LogicalPlan::RecursiveQuery(RecursiveQuery {
-                    name: name.clone(),
-                    static_term: Arc::new(static_term),
-                    recursive_term: Arc::new(recursive_term),
-                    schema: Arc::clone(schema),
-                    is_distinct: *is_distinct,
-                }))
+                LogicalPlanBuilder::from(static_term)
+                    .to_recursive_query(name.clone(), recursive_term, *is_distinct)?
+                    .build()
             }
             LogicalPlan::Analyze(a) => {
                 self.assert_no_expressions(expr)?;
@@ -2247,7 +2271,7 @@ impl PartialOrd for EmptyRelation {
 ///   intermediate table, then empty the intermediate table.
 ///
 /// [Postgres Docs]: https://www.postgresql.org/docs/current/queries-with.html#QUERIES-WITH-RECURSIVE
-#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Hash)]
 pub struct RecursiveQuery {
     /// Name of the query
     pub name: String,
@@ -2256,33 +2280,11 @@ pub struct RecursiveQuery {
     /// The recursive term (evaluated on the contents of the working table until
     /// it returns an empty set)
     pub recursive_term: Arc<LogicalPlan>,
-    /// Output schema for the recursive query. This preserves the static term's
-    /// names, types, and metadata, with nullability widened by the recursive term.
-    pub schema: DFSchemaRef,
     /// Should the output of the recursive term be deduplicated (`UNION`) or
     /// not (`UNION ALL`).
     pub is_distinct: bool,
 }
 
-// Manual implementation needed because of `schema` field.
-impl PartialOrd for RecursiveQuery {
-    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
-        (
-            &self.name,
-            &self.static_term,
-            &self.recursive_term,
-            self.is_distinct,
-        )
-            .partial_cmp(&(
-                &other.name,
-                &other.static_term,
-                &other.recursive_term,
-                other.is_distinct,
-            ))
-            .filter(|cmp| *cmp != Ordering::Equal || self == other)
-    }
-}
-
 /// Values expression. See
 /// [Postgres VALUES](https://www.postgresql.org/docs/current/queries-values.html)
 /// documentation for more details.

datafusion/expr/src/logical_plan/tree_node.rs

@@ -328,15 +328,13 @@ impl TreeNode for LogicalPlan {
                 name,
                 static_term,
                 recursive_term,
-                schema,
                 is_distinct,
             }) => (static_term, recursive_term).map_elements(f)?.update_data(
                 |(static_term, recursive_term)| {
                     LogicalPlan::RecursiveQuery(RecursiveQuery {
                         name,
                         static_term,
                         recursive_term,
-                        schema,
                         is_distinct,
                     })
                 },

datafusion/sqllogictest/test_files/cte.slt

@@ -1196,8 +1196,9 @@ logical_plan
 03)----RecursiveQuery: is_distinct=false
 04)------Projection: Int64(0) AS k, Int64(0) AS v
 05)--------EmptyRelation: rows=1
-06)------Sort: r.v ASC NULLS LAST, fetch=1
-07)--------TableScan: r projection=[k, v]
+06)------Projection: r.k, r.v
+07)--------Sort: r.v ASC NULLS LAST, fetch=1
+08)----------TableScan: r projection=[k, v]
 physical_plan
 01)GlobalLimitExec: skip=0, fetch=5
 02)--RecursiveQueryExec: name=r, is_distinct=false