Relax OneOf schema comparison to ignore nullability differences

Different table types (FileScanTable, LiveTable, MV rollups) may expose
different nullability for the same column. For example:
- A partition column in one table is non-nullable
- The same column in a rollup MV is a file column, forced nullable for
  DF 52 RecordBatch validation compatibility

Previously, OneOf rejected candidates with any schema difference
including nullability. Now it only checks field names and data types,
allowing nullable/non-nullable variants of the same column.

Author: zhuqi-lucas

src/rewrite/exploitation.rs

@@ -245,10 +245,21 @@ impl ExtensionPlanner for ViewExploitationPlanner {
             return Ok(None);
         }
 
+        // Compare schemas ignoring nullability differences.
+        // Different table types (FileScanTable, LiveTable, MV) may expose
+        // different nullability for the same column. For example, a partition
+        // column in one table is non-nullable, but the same column is a file
+        // column in a rollup MV (forced nullable for DF 52 RecordBatch
+        // validation compatibility). Field names and data types must match.
         if logical_inputs
             .iter()
             .map(|plan| plan.schema())
-            .any(|schema| schema != logical_inputs[0].schema())
+            .any(|schema| {
+                !schemas_equal_ignoring_nullability(
+                    schema.as_arrow(),
+                    logical_inputs[0].schema().as_arrow(),
+                )
+            })
         {
             return Err(DataFusionError::Plan(
                 "candidate logical plans should have the same schema".to_string(),
@@ -258,7 +269,12 @@ impl ExtensionPlanner for ViewExploitationPlanner {
         if physical_inputs
             .iter()
             .map(|plan| plan.schema())
-            .any(|schema| schema != physical_inputs[0].schema())
+            .any(|schema| {
+                !schemas_equal_ignoring_nullability(
+                    &schema,
+                    &physical_inputs[0].schema(),
+                )
+            })
         {
             return Err(DataFusionError::Plan(
                 "candidate physical plans should have the same schema".to_string(),
@@ -522,3 +538,66 @@ impl PhysicalOptimizerRule for PruneCandidates {
         true
     }
 }
+
+/// Compare two Arrow schemas ignoring field nullability.
+///
+/// Returns true if field count, field names, and data types all match.
+/// Nullability differences are ignored because different table types
+/// (e.g. FileScanTable with forced-nullable file columns vs a table
+/// where the same column is a non-nullable partition column) can
+/// legitimately differ in nullability while being semantically equivalent.
+fn schemas_equal_ignoring_nullability(
+    a: &arrow_schema::Schema,
+    b: &arrow_schema::Schema,
+) -> bool {
+    a.fields().len() == b.fields().len()
+        && a.fields()
+            .iter()
+            .zip(b.fields().iter())
+            .all(|(f1, f2)| f1.name() == f2.name() && f1.data_type() == f2.data_type())
+}
+
+#[cfg(test)]
+mod tests_nullability {
+    use super::*;
+    use arrow_schema::{DataType, Field, Schema};
+
+    #[test]
+    fn schemas_equal_when_only_nullability_differs() {
+        let a = Schema::new(vec![
+            Field::new("ticker", DataType::Utf8, false),
+            Field::new("date", DataType::Utf8, false),
+            Field::new("price", DataType::Float64, false),
+        ]);
+        let b = Schema::new(vec![
+            Field::new("ticker", DataType::Utf8, true),
+            Field::new("date", DataType::Utf8, true),
+            Field::new("price", DataType::Float64, true),
+        ]);
+        assert!(schemas_equal_ignoring_nullability(&a, &b));
+    }
+
+    #[test]
+    fn schemas_not_equal_when_types_differ() {
+        let a = Schema::new(vec![Field::new("x", DataType::Int32, false)]);
+        let b = Schema::new(vec![Field::new("x", DataType::Int64, false)]);
+        assert!(!schemas_equal_ignoring_nullability(&a, &b));
+    }
+
+    #[test]
+    fn schemas_not_equal_when_names_differ() {
+        let a = Schema::new(vec![Field::new("ticker", DataType::Utf8, true)]);
+        let b = Schema::new(vec![Field::new("symbol", DataType::Utf8, true)]);
+        assert!(!schemas_equal_ignoring_nullability(&a, &b));
+    }
+
+    #[test]
+    fn schemas_not_equal_when_field_count_differs() {
+        let a = Schema::new(vec![
+            Field::new("x", DataType::Int32, false),
+            Field::new("y", DataType::Int32, false),
+        ]);
+        let b = Schema::new(vec![Field::new("x", DataType::Int32, false)]);
+        assert!(!schemas_equal_ignoring_nullability(&a, &b));
+    }
+}