From 42c4ff3088af41cf801ae040d6ae13ad52504cb1 Mon Sep 17 00:00:00 2001
From: Jiawei Zhao <Phoenix500526@163.com>
Date: Sat, 31 Oct 2026 21:45:01 +0800
Subject: [PATCH 1/2] fix: preserve union output schema metadata

Output batches must retain the merged schema metadata.

Refs #24098
---
 datafusion/physical-plan/src/union.rs | 10 ++++++----
 1 file changed, 6 insertions(+), 4 deletions(-)

diff --git a/datafusion/physical-plan/src/union.rs b/datafusion/physical-plan/src/union.rs
index 042f3217e22c5..7ce54abd1a58d 100644
--- a/datafusion/physical-plan/src/union.rs
+++ b/datafusion/physical-plan/src/union.rs
@@ -116,7 +116,9 @@ fn coerce_schema(
         })
         .collect::<Result<Vec<_>>>()?;
 
-    Ok(Arc::new(ProjectionExec::try_new(exprs, input)?))
+    Ok(Arc::new(ProjectionExec::try_new_with_schema_metadata(
+        exprs, input, schema,
+    )?))
 }
 
 /// `UnionExec`: `UNION ALL` execution plan.
@@ -1851,9 +1853,9 @@ mod tests {
             );
             for batch in batches {
                 assert_eq!(
-                    batch.schema().fields(),
-                    expected_schema.fields(),
-                    "{nullability:?}"
+                    batch.schema(),
+                    expected_schema,
+                    "output batch schema for {nullability:?}"
                 );
             }
         }

From 615925dca3a1a51403037aa1ed5ad938ad78d15b Mon Sep 17 00:00:00 2001
From: Jiawei Zhao <Phoenix500526@163.com>
Date: Tue, 3 Nov 2026 15:04:50 +0800
Subject: [PATCH 2/2] test: cover union schema-only metadata changes

Exercise schema coercion when all input fields are identical.

Refs #24098
---
 datafusion/physical-plan/src/union.rs | 35 ++++++++++++++++++---------
 1 file changed, 24 insertions(+), 11 deletions(-)

diff --git a/datafusion/physical-plan/src/union.rs b/datafusion/physical-plan/src/union.rs
index 7ce54abd1a58d..af1a77b77b7ea 100644
--- a/datafusion/physical-plan/src/union.rs
+++ b/datafusion/physical-plan/src/union.rs
@@ -1792,12 +1792,14 @@ mod tests {
 
         // Field and schema metadata are merged in input order, so the last
         // input wins regardless of which field supplies the nullability.
-        for nullability in [
-            [false, false, false],
-            [true, false, false],
-            [false, true, false],
-            [false, false, true],
-            [true, true, true],
+        for (nullability, vary_fields) in [
+            ([false, false, false], true),
+            ([true, false, false], true),
+            ([false, true, false], true),
+            ([false, false, true], true),
+            ([true, true, true], true),
+            // Identical fields use plain Column expressions during schema coercion.
+            ([false, false, false], false),
         ] {
             let inputs = nullability
                 .iter()
@@ -1807,9 +1809,12 @@ mod tests {
                         ("shared".to_owned(), format!("input{i}")),
                         (format!("only{i}"), "kept".to_owned()),
                     ]);
-                    let field =
+                    let field = if vary_fields {
                         Field::new(format!("value{i}"), DataType::Int32, nullable)
-                            .with_metadata(metadata.clone());
+                            .with_metadata(metadata.clone())
+                    } else {
+                        Field::new("value0", DataType::Int32, nullable)
+                    };
                     let schema =
                         Arc::new(Schema::new_with_metadata(vec![field], metadata));
                     let batch = RecordBatch::try_new(
@@ -1831,13 +1836,21 @@ mod tests {
             let expected_schema = Arc::new(Schema::new_with_metadata(
                 vec![
                     Field::new("value0", DataType::Int32, nullability.contains(&true))
-                        .with_metadata(expected_metadata.clone()),
+                        .with_metadata(if vary_fields {
+                            expected_metadata.clone()
+                        } else {
+                            Metadata::default()
+                        }),
                 ],
                 expected_metadata,
             ));
 
             let union = UnionExec::try_new(inputs)?;
-            assert_eq!(union.schema(), expected_schema, "{nullability:?}");
+            assert_eq!(
+                union.schema(),
+                expected_schema,
+                "{nullability:?}, vary_fields={vary_fields}"
+            );
             let batches = collect(union, Arc::new(TaskContext::default())).await?;
             datafusion_common::assert_batches_sorted_eq!(
                 [
@@ -1855,7 +1868,7 @@ mod tests {
                 assert_eq!(
                     batch.schema(),
                     expected_schema,
-                    "output batch schema for {nullability:?}"
+                    "output batch schema for {nullability:?}, vary_fields={vary_fields}"
                 );
             }
         }