feat: update recursive CTE and schema alignment for nullability

- Reverted oom_recursive_cte query to use SELECT 1 as id.
- Updated schema alignment to permit safe nullability widening (non-null to nullable) through same-type CastExpr.
- Enhanced recursive CTE to reconcile output nullability across static and recursive terms, ensuring alignment of both plans.
- Rejected nullability narrowing.

Author: kosiew

datafusion/core/tests/memory_limit/mod.rs

@@ -343,7 +343,7 @@ async fn oom_recursive_cte() {
     TestCase::new()
         .with_query(
             "WITH RECURSIVE nodes AS (
-            SELECT id FROM (VALUES (1), (NULL)) AS t(id) WHERE id IS NOT NULL
+            SELECT 1 as id
             UNION ALL
             SELECT UNNEST(RANGE(id+1, id+1000)) as id
             FROM nodes

datafusion/physical-plan/src/common.rs

@@ -22,7 +22,7 @@ use std::fs::metadata;
 use std::sync::Arc;
 
 use super::SendableRecordBatchStream;
-use crate::expressions::Column;
+use crate::expressions::{CastExpr, Column};
 use crate::projection::{ProjectionExec, ProjectionExpr};
 use crate::stream::RecordBatchReceiverStream;
 use crate::{ColumnStatistics, ExecutionPlan, Statistics};
@@ -91,18 +91,18 @@ fn build_file_list_recurse(
     Ok(())
 }
 
-/// Align `input`'s physical plan schema with `expected_schema` when only field names differ.
+/// Align `input`'s physical plan schema with `expected_schema`.
 ///
 /// This helper is intended for operators that combine independently planned children but
 /// expose a single declared output schema. It returns `input` unchanged when schemas already
 /// match exactly. Otherwise, it validates that projection can safely produce the expected
 /// schema, then wraps `input` in a [`ProjectionExec`] that keeps columns in their existing
 /// positional order and aliases them to `expected_schema`'s field names.
 ///
-/// [`ProjectionExec`] can rename fields but preserves column data types, nullability, field
-/// metadata, and schema metadata from the input expressions. Therefore, this helper rejects
-/// mismatches in those attributes rather than returning a plan whose schema still differs
-/// from `expected_schema`.
+/// [`ProjectionExec`] can rename fields. When the expected field is nullable and the input
+/// field is not, this helper also widens nullability with a same-type [`CastExpr`]. It rejects
+/// differences that projection cannot safely normalize exactly, such as data type, metadata,
+/// schema metadata, and nullability narrowing.
 pub fn project_plan_to_schema(
     input: Arc<dyn ExecutionPlan>,
     expected_schema: &SchemaRef,
@@ -134,7 +134,7 @@ pub fn project_plan_to_schema(
         .find_map(|(i, (input_field, expected_field))| {
             if input_field.data_type() != expected_field.data_type() {
                 Some((i, input_field, expected_field, "data type"))
-            } else if input_field.is_nullable() != expected_field.is_nullable() {
+            } else if input_field.is_nullable() && !expected_field.is_nullable() {
                 Some((i, input_field, expected_field, "nullability"))
             } else if input_field.metadata() != expected_field.metadata() {
                 Some((i, input_field, expected_field, "metadata"))
@@ -157,9 +157,22 @@ pub fn project_plan_to_schema(
         .fields()
         .iter()
         .enumerate()
-        .map(|(i, expected_field)| ProjectionExpr {
-            expr: Arc::new(Column::new(input_schema.field(i).name(), i)),
-            alias: expected_field.name().clone(),
+        .map(|(i, expected_field)| {
+            let input_field = input_schema.field(i);
+            let column = Arc::new(Column::new(input_field.name(), i));
+            let expr = if !input_field.is_nullable() && expected_field.is_nullable() {
+                Arc::new(CastExpr::new_with_target_field(
+                    column,
+                    Arc::clone(expected_field),
+                    None,
+                )) as _
+            } else {
+                column as _
+            };
+            ProjectionExpr {
+                expr,
+                alias: expected_field.name().clone(),
+            }
         })
         .collect::<Vec<_>>();
 
@@ -484,7 +497,22 @@ mod tests {
     }
 
     #[test]
-    fn project_plan_to_schema_errors_on_nullability_mismatch() {
+    fn project_plan_to_schema_widens_nullability() -> Result<()> {
+        let input = empty_exec(vec![Field::new("a", DataType::Int32, false)]);
+        let expected_schema = Arc::new(Schema::new(vec![Field::new(
+            "renamed",
+            DataType::Int32,
+            true,
+        )]));
+
+        let result = project_plan_to_schema(input, &expected_schema)?;
+
+        assert_eq!(result.schema(), expected_schema);
+        Ok(())
+    }
+
+    #[test]
+    fn project_plan_to_schema_errors_on_nullability_narrowing() {
         let input = empty_exec(vec![Field::new("a", DataType::Int32, true)]);
         let expected_schema = Arc::new(Schema::new(vec![Field::new(
             "renamed",

datafusion/physical-plan/src/recursive_query.rs

@@ -35,7 +35,7 @@ use crate::{
 };
 use arrow::array::{BooleanArray, BooleanBuilder};
 use arrow::compute::filter_record_batch;
-use arrow::datatypes::SchemaRef;
+use arrow::datatypes::{Field, Schema, SchemaRef};
 use arrow::record_batch::RecordBatch;
 use datafusion_common::tree_node::TreeNodeRecursion;
 use datafusion_common::tree_node::{Transformed, TransformedResult, TreeNode};
@@ -91,10 +91,12 @@ impl RecursiveQueryExec {
         // Each recursive query needs its own work table
         let work_table = Arc::new(WorkTable::new(name.clone()));
         // Use the same work table for both the WorkTableExec and the recursive term
+        let output_schema =
+            recursive_output_schema(&static_term.schema(), &recursive_term.schema());
+        let static_term = project_plan_to_schema(static_term, &output_schema)?;
         let recursive_term = assign_work_table(recursive_term, &work_table)?;
-        let recursive_term =
-            project_plan_to_schema(recursive_term, &static_term.schema())?;
-        let cache = Self::compute_properties(static_term.schema());
+        let recursive_term = project_plan_to_schema(recursive_term, &output_schema)?;
+        let cache = Self::compute_properties(output_schema);
         Ok(RecursiveQueryExec {
             name,
             static_term,
@@ -368,6 +370,30 @@ impl RecursiveQueryStream {
     }
 }
 
+fn recursive_output_schema(
+    static_schema: &SchemaRef,
+    recursive_schema: &SchemaRef,
+) -> SchemaRef {
+    let fields = static_schema
+        .fields()
+        .iter()
+        .zip(recursive_schema.fields())
+        .map(|(static_field, recursive_field)| {
+            Field::new(
+                static_field.name(),
+                static_field.data_type().clone(),
+                static_field.is_nullable() || recursive_field.is_nullable(),
+            )
+            .with_metadata(static_field.metadata().clone())
+        })
+        .collect::<Vec<_>>();
+
+    Arc::new(Schema::new_with_metadata(
+        fields,
+        static_schema.metadata().clone(),
+    ))
+}
+
 fn assign_work_table(
     plan: Arc<dyn ExecutionPlan>,
     work_table: &Arc<WorkTable>,
@@ -528,19 +554,21 @@ mod tests {
     }
 
     #[test]
-    fn recursive_query_exec_rejects_nullability_mismatch() {
+    fn recursive_query_exec_reconciles_nullability() -> Result<()> {
         let static_term = empty_exec(vec![Field::new("value", DataType::Int32, false)]);
         let recursive_term =
             empty_exec(vec![Field::new("value + Int32(1)", DataType::Int32, true)]);
 
-        let err = RecursiveQueryExec::try_new(
+        let exec = RecursiveQueryExec::try_new(
             "numbers".to_string(),
             static_term,
             recursive_term,
             false,
-        )
-        .unwrap_err();
+        )?;
 
-        assert!(err.to_string().contains("field nullability differs"));
+        assert!(exec.schema().field(0).is_nullable());
+        assert!(exec.static_term().schema().field(0).is_nullable());
+        assert!(exec.recursive_term().schema().field(0).is_nullable());
+        Ok(())
     }
 }