feat: improve query planning and schema handling

- Skip identity projection when schema is already the same in builder.rs.
- Replace split temporary binding with Option<DFSchema> in physical_planner.rs.
- Reuse nullability booleans in recursive_schema.rs for efficiency.
- Simplify formatting in matches! within cte.rs.
- Add repeated schema helper to physical_planner.rs tests.
- Invert branch in plan.rs to reduce clutter in recursive query rebuild path.

Author: kosiew

datafusion/common/src/recursive_schema.rs

@@ -120,13 +120,16 @@ fn widen_dfschema_nullability_with_fields<'a>(
             return Ok(None);
         }
 
-        widened_nullability |= !base_field.is_nullable() && widening_field.is_nullable();
+        let base_nullable = base_field.is_nullable();
+        let widening_nullable = widening_field.is_nullable();
+        let output_nullable = base_nullable || widening_nullable;
+        widened_nullability |= !base_nullable && widening_nullable;
         fields.push((
             qualifier.cloned(),
             base_field
                 .as_ref()
                 .clone()
-                .with_nullable(base_field.is_nullable() || widening_field.is_nullable())
+                .with_nullable(output_nullable)
                 .into(),
         ));
     }

datafusion/core/src/physical_planner.rs

@@ -998,22 +998,22 @@ impl DefaultPhysicalPlanner {
                 let input_exec = children.one()?;
                 let physical_input_schema = input_exec.schema();
                 let logical_input_schema = input.as_ref().schema();
-                let reconciled_logical_schema;
-                let logical_input_schema = if schema_satisfied_by(
-                    logical_input_schema.inner(),
-                    &physical_input_schema,
-                ) || !contains_recursive_query_input(input)
-                {
-                    logical_input_schema
-                } else if let Some(schema) = reconcile_dfschema_with_schema_nullability(
-                    logical_input_schema,
-                    &physical_input_schema,
-                )? {
-                    reconciled_logical_schema = schema;
-                    &reconciled_logical_schema
+                let needs_recursive_reconciliation =
+                    !schema_satisfied_by(
+                        logical_input_schema.inner(),
+                        &physical_input_schema,
+                    ) && contains_recursive_query_input(input);
+                let reconciled_logical_schema = if needs_recursive_reconciliation {
+                    reconcile_dfschema_with_schema_nullability(
+                        logical_input_schema,
+                        &physical_input_schema,
+                    )?
                 } else {
-                    logical_input_schema
+                    None
                 };
+                let logical_input_schema = reconciled_logical_schema
+                    .as_ref()
+                    .unwrap_or(logical_input_schema);
                 let physical_input_schema_from_logical = logical_input_schema.inner();
 
                 if !options.execution.skip_physical_aggregate_schema_check
@@ -4762,6 +4762,10 @@ digraph {
         }
     }
 
+    fn c1_i32_required_schema() -> SchemaRef {
+        Arc::new(Schema::new(vec![Field::new("c1", DataType::Int32, false)]))
+    }
+
     /// Attempts to plan a query with potentially mismatched schemas.
     async fn plan_with_schemas(
         logical_schema: SchemaRef,
@@ -4783,8 +4787,7 @@ digraph {
     // It then panics on unimplemented error in NoOpExecutionPlan.
     #[should_panic(expected = "NoOpExecutionPlan")]
     async fn test_aggregate_schema_check_passes() {
-        let schema =
-            Arc::new(Schema::new(vec![Field::new("c1", DataType::Int32, false)]));
+        let schema = c1_i32_required_schema();
 
         plan_with_schemas(
             Arc::clone(&schema),
@@ -4797,8 +4800,7 @@ digraph {
 
     #[tokio::test]
     async fn test_aggregate_schema_mismatch_metadata() {
-        let logical_schema =
-            Arc::new(Schema::new(vec![Field::new("c1", DataType::Int32, false)]));
+        let logical_schema = c1_i32_required_schema();
         let physical_schema = Arc::new(
             Schema::new(vec![Field::new("c1", DataType::Int32, false)])
                 .with_metadata(HashMap::from([("key".into(), "value".into())])),
@@ -4817,8 +4819,7 @@ digraph {
 
     #[tokio::test]
     async fn test_aggregate_schema_mismatch_field_count() {
-        let logical_schema =
-            Arc::new(Schema::new(vec![Field::new("c1", DataType::Int32, false)]));
+        let logical_schema = c1_i32_required_schema();
         let physical_schema = Arc::new(Schema::new(vec![
             Field::new("c1", DataType::Int32, false),
             Field::new("c2", DataType::Int32, false),
@@ -4837,8 +4838,7 @@ digraph {
 
     #[tokio::test]
     async fn test_aggregate_schema_mismatch_field_name() {
-        let logical_schema =
-            Arc::new(Schema::new(vec![Field::new("c1", DataType::Int32, false)]));
+        let logical_schema = c1_i32_required_schema();
         let physical_schema = Arc::new(Schema::new(vec![Field::new(
             "different_name",
             DataType::Int32,
@@ -4858,8 +4858,7 @@ digraph {
 
     #[tokio::test]
     async fn test_aggregate_schema_mismatch_field_type() {
-        let logical_schema =
-            Arc::new(Schema::new(vec![Field::new("c1", DataType::Int32, false)]));
+        let logical_schema = c1_i32_required_schema();
         let physical_schema =
             Arc::new(Schema::new(vec![Field::new("c1", DataType::Int64, false)]));
 
@@ -4876,8 +4875,7 @@ digraph {
 
     #[tokio::test]
     async fn test_aggregate_schema_mismatch_field_nullability() {
-        let logical_schema =
-            Arc::new(Schema::new(vec![Field::new("c1", DataType::Int32, false)]));
+        let logical_schema = c1_i32_required_schema();
         let physical_schema =
             Arc::new(Schema::new(vec![Field::new("c1", DataType::Int32, true)]));
 
@@ -4894,8 +4892,7 @@ digraph {
 
     #[tokio::test]
     async fn test_aggregate_schema_mismatch_field_metadata() {
-        let logical_schema =
-            Arc::new(Schema::new(vec![Field::new("c1", DataType::Int32, false)]));
+        let logical_schema = c1_i32_required_schema();
         let physical_schema = Arc::new(Schema::new(vec![
             Field::new("c1", DataType::Int32, false)
                 .with_metadata(HashMap::from([("key".into(), "value".into())])),

datafusion/expr/src/logical_plan/builder.rs

@@ -68,6 +68,10 @@ use indexmap::IndexSet;
 pub const UNNAMED_TABLE: &str = "?table?";
 
 fn plan_with_schema(plan: LogicalPlan, schema: DFSchemaRef) -> Result<LogicalPlan> {
+    if schema == *plan.schema() {
+        return Ok(plan);
+    }
+
     match plan {
         LogicalPlan::Projection(Projection { expr, input, .. }) => {
             Projection::try_new_with_schema(expr, input, schema)

datafusion/expr/src/logical_plan/plan.rs

@@ -752,21 +752,21 @@ impl LogicalPlan {
                     static_term.schema(),
                     recursive_term.schema(),
                 )?;
-                if output_schema == *static_term.schema() {
-                    Ok(LogicalPlan::RecursiveQuery(RecursiveQuery {
-                        name,
-                        static_term,
-                        recursive_term,
-                        is_distinct,
-                    }))
-                } else {
+                if output_schema != *static_term.schema() {
                     LogicalPlanBuilder::from(Arc::unwrap_or_clone(static_term))
                         .to_recursive_query(
                             name,
                             Arc::unwrap_or_clone(recursive_term),
                             is_distinct,
                         )?
                         .build()
+                } else {
+                    Ok(LogicalPlan::RecursiveQuery(RecursiveQuery {
+                        name,
+                        static_term,
+                        recursive_term,
+                        is_distinct,
+                    }))
                 }
             }
             LogicalPlan::Analyze(_) => Ok(self),

datafusion/sql/src/cte.rs

@@ -190,7 +190,10 @@ fn has_work_table_reference(
     work_table_source: &Arc<dyn TableSource>,
 ) -> bool {
     plan.exists(|node| {
-        Ok(matches!(node, LogicalPlan::TableScan(scan) if Arc::ptr_eq(&scan.source, work_table_source)))
+        Ok(matches!(
+            node,
+            LogicalPlan::TableScan(scan) if Arc::ptr_eq(&scan.source, work_table_source)
+        ))
     })
     // Closure always returns Ok
     .unwrap()