@@ -396,6 +396,13 @@ object Utils extends CometTypeShim with Logging {
             }
             while (reader.loadNextBatch()) {
               val sourceRoot = reader.getVectorSchemaRoot
+              if (targetRoot != null && targetRoot.getSchema != sourceRoot.getSchema) {
+                logWarning(
+                  "Arrow schemas differ during BroadcastExchange coalescing; skipping coalesce")
+                targetRoot.close()
+                targetRoot = null
+                return (buffers, 0L, 0L)
+              }
               if (targetRoot == null) {
                 targetRoot = VectorSchemaRoot.create(sourceRoot.getSchema, allocator)
                 targetRoot.allocateNew()
 
@@ -25,7 +25,7 @@ import org.scalatest.Tag
 import org.apache.spark.sql.CometTestBase
 import org.apache.spark.sql.catalyst.TableIdentifier
 import org.apache.spark.sql.catalyst.analysis.UnresolvedRelation
-import org.apache.spark.sql.comet.{CometBroadcastExchangeExec, CometBroadcastHashJoinExec, CometBroadcastNestedLoopJoinExec, CometSortMergeJoinExec}
+import org.apache.spark.sql.comet.{CometBroadcastExchangeExec, CometBroadcastHashJoinExec, CometBroadcastNestedLoopJoinExec, CometSortMergeJoinExec, CometUnionExec}
 import org.apache.spark.sql.execution.adaptive.AQEShuffleReadExec
 import org.apache.spark.sql.internal.SQLConf
 
@@ -657,6 +657,30 @@ class CometJoinSuite extends CometTestBase {
     }
   }
 
+  test("Broadcast coalescing falls back when union children have different nullability") {
+    withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+      withParquetTable(Seq((1, 10), (2, 20), (3, 30)), "t") {
+        val (_, cometPlan) = checkSparkAnswerAndOperator(
+          sql("""
+              |SELECT /*+ BROADCAST(b) */ p._1, b.v
+              |FROM t p JOIN (
+              |  SELECT _1 AS k, 99 AS v FROM t
+              |  UNION ALL
+              |  SELECT _1 AS k, _2 + 1 AS v FROM t
+              |) b ON p._1 = b.k
+              |""".stripMargin),
+          Seq(
+            classOf[CometBroadcastExchangeExec],
+            classOf[CometBroadcastHashJoinExec],
+            classOf[CometUnionExec]))
+
+        val broadcast = collect(cometPlan) { case b: CometBroadcastExchangeExec => b }.head
+        assert(broadcast.metrics("numCoalescedBatches").value == 0L)
+        assert(broadcast.metrics("numCoalescedRows").value == 0L)
+      }
+    }
+  }
+
   // Reproducer for SPARK-43113: full outer SMJ with a join filter that references
   // a nullable column should not match when the filter evaluates to NULL.
   test("SPARK-43113: Full outer SMJ with NULL in join filter") {