Fix nullability mismatch in CometArrowStreamSuite.

Author: mbutrovich

spark/src/main/scala/org/apache/spark/sql/comet/execution/arrow/CometNativeArrowSource.scala

@@ -24,7 +24,7 @@ import scala.jdk.CollectionConverters._
 import org.apache.arrow.c.{ArrowArrayStream, Data}
 import org.apache.arrow.memory.BufferAllocator
 import org.apache.arrow.vector.ipc.ArrowReader
-import org.apache.arrow.vector.types.pojo.{Field, Schema}
+import org.apache.arrow.vector.types.pojo.{Field, FieldType, Schema}
 import org.apache.spark.TaskContext
 import org.apache.spark.internal.Logging
 import org.apache.spark.rdd.RDD
@@ -132,8 +132,14 @@ object CometArrowStream extends Logging {
    * `CometDictionaryVector`, [[ColumnarBatchArrowReader]] decodes it via
    * `DictionaryEncoder.decode` before unloading, so the wire-level field is the dictionary's
    * *value* type, not `Dictionary<index, value>`. For everything else, use the underlying value
-   * vector's field. Field name / nullability / metadata come from `expected` so that consumers
-   * indexing by name keep working.
+   * vector's field.
+   *
+   * Field name and metadata come from `expected` so that consumers indexing by name keep working.
+   * Nullability is the union of the two — a CometVector that happens to hold no nulls in this
+   * batch can still be nullable per Spark's contract (the next batch may have one), and a column
+   * whose actual buffer carries validity bits must stay nullable even if Spark thought otherwise.
+   * Taking only `raw.isNullable` here would advertise non-nullable when the next batch does carry
+   * a null and crash native validation.
    */
   private def actualFieldOf(col: CometVector, expected: Field): Field = {
     val raw = col match {
@@ -143,7 +149,10 @@ object CometArrowStream extends Logging {
         dict.getVector.getField
       case _ => col.getValueVector.getField
     }
-    new Field(expected.getName, raw.getFieldType, raw.getChildren)
+    val nullable = expected.isNullable || raw.isNullable
+    val fieldType =
+      new FieldType(nullable, raw.getType, raw.getDictionary, expected.getMetadata)
+    new Field(expected.getName, fieldType, raw.getChildren)
   }
 
   /**

spark/src/test/scala/org/apache/spark/sql/comet/execution/arrow/CometArrowStreamSuite.scala

@@ -112,4 +112,36 @@ class CometArrowStreamSuite extends AnyFunSuite with Matchers {
       allocator.close()
     }
   }
+
+  test("reconcileStreamSchema preserves nullability when expected is nullable but actual is not") {
+    val allocator = new RootAllocator(Integer.MAX_VALUE)
+    try {
+      // Spark catalyst declares the column nullable; the first batch happens to come from a
+      // vector whose Field reports non-nullable. Subsequent batches may carry nulls, so the
+      // wire schema must stay nullable or native validation rejects the next null with
+      // "declared as non-nullable but contains null values".
+      val v = new BigIntVector(
+        new Field(
+          "col_0",
+          new FieldType(false, new ArrowType.Int(64, true), null),
+          java.util.Collections.emptyList[Field]()),
+        allocator)
+      v.allocateNew()
+      v.setSafe(0, 1L)
+      v.setValueCount(1)
+      val cv = new CometPlainVector(v, false)
+      val batch = batchOf(cv)
+      val expected = expectedSchema("c0" -> new ArrowType.Int(64, true)) // nullable=true
+
+      val (returned, _) = CometArrowStream
+        .reconcileStreamSchema("test", expected, Iterator.single(batch))
+
+      val returnedField = returned.getFields.get(0)
+      returnedField.isNullable shouldBe true
+
+      cv.close()
+    } finally {
+      allocator.close()
+    }
+  }
 }