fix: use consistent intersect semantics for UNION field metadata

[milesrichardson-edb]

Intra-fork draft PR to validate CI before proposing this upstream. Do not merge.

Which issue does this PR close?

• Closes #TBD (upstream issue will be filed together with the upstream PR).

Rationale for this change

DataFusion derives a union's schema in three places: Union::try_new in datafusion-expr, coerce_union_schema in the type coercion analyzer, and union_schema in datafusion-physical-plan (shared by UnionExec and InterleaveExec). Since apache#17248, Union::try_new intersects field metadata across inputs with intersect_metadata_for_union: a key survives only if every input with non-empty metadata agrees on its value. The other two sites still merge with HashMap::extend, and they pick the surviving value differently: union_schema keeps the metadata of the field selected by find_or_first(Field::is_nullable), while coerce_union_schema folds inputs in order, so the last input wins.

The analyzer rewrites every LogicalPlan::Union with the schema from coerce_union_schema, so for a SQL query the logical schema comes from one merge site and the physical schema from the other. When the union branches carry conflicting field metadata, for example different PARQUET:field_id values from two Parquet files, and the branches differ in nullability, the two sites keep different values and the physical planner's aggregate schema check fails:

Internal error: Physical input schema should be the same as the one converted from logical input schema.
Differences:
  - field metadata at index 0 [v]: (physical) {"PARQUET:field_id": "1"} vs (logical) {"PARQUET:field_id": "2"}

This is the same kind of two-path divergence as apache#14356 (nullability) and apache#19049 (empty metadata); fixing only one of the two remaining sites would leave it disagreeing with the other.

What changes are included in this PR?

• coerce_union_schema_with_schema (datafusion-optimizer) collects each field's metadata from every input and intersects it once with intersect_metadata_for_union, instead of merging in place.
• union_schema (datafusion-physical-plan, which already depends on datafusion-expr) intersects field metadata with the same helper. Field name, type and nullability selection are unchanged.
• The coerce_union_schema doc comment from Add documentation for UNION schema handling. apache/datafusion#17248 is updated to describe the intersect semantics.

Schema-level metadata is still merged with last-writer-wins on both sites; only field-level metadata changes. This is a known inconsistency with Union::try_new, which already intersects schema-level metadata, and is left as a follow-up rather than folded into this fix.

Are these changes tested?

Yes.

• Unit tests for both union_schema and coerce_union_schema_with_schema covering the same-value, conflicting, one-input-empty and all-empty cases. The remaining intersect_metadata_for_union cases from fix: skip empty metadata in intersect_metadata_for_union to prevent s… apache/datafusion#21127 are covered by its own tests in datafusion-expr, since both sites now delegate to it.
• A sqllogictest in metadata.slt reproducing the failure above: an aggregate over a UNION ALL of two tables with conflicting PARQUET:field_id metadata and differing nullability (registered in the test harness, since field metadata cannot be declared in SQL). It fails with the internal error on main and passes with this change.
• A second sqllogictest in metadata.slt unions two tables that agree on field metadata but differ in type (Int32/Int64), so type coercion inserts a CAST above one branch. It pins the assumption that a coercion cast preserves field metadata on both the logical and physical side, which is what makes the two intersects in this PR agree when a cast is present.

We audited existing coverage for a test that pinned the prior last-writer-wins behavior and found none: datafusion/sqllogictest/test_files/union.slt, union_by_name.slt, and arrow_field.slt carry no union field-metadata assertions, and the existing metadata.slt union coverage (the apache#19049 carve-out) only asserts data values through NULL branches, not metadata survival. No existing test was modified or deleted; every change in this PR is an addition. These are the first tests to pin union field-metadata semantics in either direction.

Are there any user-facing changes?

Conflicting field-level metadata no longer survives a UNION: a key is kept only if every branch that has non-empty metadata agrees on its value, matching what Union::try_new already did; branches with empty metadata (e.g. NULL literals) are skipped rather than erasing others. Previously one branch's value survived, chosen by input order in the logical schema and by nullability in the physical schema. Because the union result schema is what COPY ... TO (parquet/arrow) writes and what CREATE VIEW, DataFrame::schema(), and the FFI/C-Data exports expose, a conflicting PARQUET:field_id or ARROW:extension:name that used to be written or reported is now dropped from those outputs. Schema-level (table) metadata is unchanged — still merged last-writer-wins.

[milesrichardson-edb]

@gruuya — requesting your sign-off on this before I file anything upstream. This is the fix for the TPC-DS Q71 DirectScan fallback we discussed in Slack (BEAC-411): union field metadata is computed at three sites in DataFusion with two different semantics, and the physical planner's aggregate schema check trips on the divergence. This PR makes coerce_union_schema and union_schema use the same intersect_metadata_for_union helper that Union::try_new already uses.

The diff and PR body above are in final proposed form (CI green on this fork, including a test pinning cast-metadata-preservation, which is what keeps the logical and physical intersections in agreement when coercion casts fire). Below is the rest of the package: the proposed upstream issue text and the submission workflow. Please review all of it — code, issue text, and choreography — and flag anything you'd change.

---

Proposed upstream issue

Title: Union field metadata is computed differently in type coercion and physical planning, failing the aggregate schema check

Describe the bug

DataFusion derives a union's schema in three places: Union::try_new in
datafusion-expr, coerce_union_schema in the type coercion analyzer, and
union_schema in datafusion-physical-plan (shared by UnionExec and
InterleaveExec). Since apache#17248, Union::try_new intersects field metadata
across inputs with intersect_metadata_for_union. A key survives only if every
input with non-empty metadata agrees on its value. The other two sites still
merge with HashMap::extend, and they pick the surviving value differently.
union_schema keeps the metadata of the field selected by
find_or_first(Field::is_nullable). coerce_union_schema folds inputs in
order, so the last input wins.

The analyzer rewrites every LogicalPlan::Union with the schema from
coerce_union_schema, so for a SQL query the logical schema comes from one
merge site and the physical schema from the other. When the union branches
carry conflicting field metadata, for example different PARQUET:field_id
values from two Parquet files, and the branches also differ in nullability,
the two sites keep different values. The physical planner's aggregate schema
check then fails:

Internal error: Physical input schema should be the same as the one converted from logical input schema.
Differences:
  - field metadata at index 0 [v]: (physical) {"PARQUET:field_id": "1"} vs (logical) {"PARQUET:field_id": "2"}

To Reproduce

Reproducing this needs branches with conflicting non-empty field metadata that
also differ in nullability. If nullability is equal on both sides,
find_or_first picks the same field the last-writer-wins fold would have
picked anyway, and the two sites agree by accident.

Fails on current main (ad7d6ea) and on 53/54:

use std::collections::HashMap;
use std::sync::Arc;
use arrow::array::Int64Array;
use arrow::datatypes::{DataType, Field, Schema};
use arrow::record_batch::RecordBatch;
use datafusion::catalog::MemTable;
use datafusion::prelude::*;

#[tokio::main]
async fn main() -> datafusion::error::Result<()> {
    let ctx = SessionContext::new();

    let field_a = Field::new("a", DataType::Int64, false)
        .with_metadata(HashMap::from([("PARQUET:field_id".into(), "1".into())]));
    let schema_a = Arc::new(Schema::new(vec![field_a]));
    let batch_a = RecordBatch::try_new(
        Arc::clone(&schema_a),
        vec![Arc::new(Int64Array::from(vec![1, 2, 3]))],
    )?;
    ctx.register_table("t1", Arc::new(MemTable::try_new(schema_a, vec![vec![batch_a]])?))?;

    let field_b = Field::new("b", DataType::Int64, true)
        .with_metadata(HashMap::from([("PARQUET:field_id".into(), "2".into())]));
    let schema_b = Arc::new(Schema::new(vec![field_b]));
    let batch_b = RecordBatch::try_new(
        Arc::clone(&schema_b),
        vec![Arc::new(Int64Array::from(vec![Some(10), None, Some(20)]))],
    )?;
    ctx.register_table("t2", Arc::new(MemTable::try_new(schema_b, vec![vec![batch_b]])?))?;

    ctx.sql("SELECT sum(v) FROM (SELECT a AS v FROM t1 UNION ALL SELECT b AS v FROM t2) u")
        .await?
        .collect()
        .await?;
    Ok(())
}

Expected behavior

The query succeeds, with the logical and physical union schemas agreeing on
field metadata. coerce_union_schema and union_schema should use the same
intersect semantics as Union::try_new.

Additional context

This is the same kind of two-path divergence as apache#14356 (nullability) and
apache#19049 (empty metadata). apache#21127's intersect_metadata_for_union handles
conflicting and empty metadata correctly; the problem here is that
coerce_union_schema and union_schema never call it.

Fixing this by reusing intersect_metadata_for_union at both sites is a
behavior change worth flagging: a conflicting field-metadata key (for example
PARQUET:field_id) will no longer survive a union, which changes what
COPY ... TO writes to a parquet/arrow file for a UNION over independently
schematized inputs. Separately, this fix only unifies field-level metadata;
schema-level (table) metadata is left as last-writer-wins in both
coerce_union_schema and union_schema, and remains inconsistent with
Union::try_new, which already intersects it. That's a scoped follow-up, not
part of this fix.

I have a fix ready and will open a PR shortly.

---

Submission workflow (once you sign off)

1. File the issue above via the "Bug report" form on apache/datafusion (form sets Bug type/label).
2. Comment take on the issue to self-assign (the repo's automated self-assign mechanism).
3. Open the upstream PR from fix/intersect-union-field-metadata with this PR's body, replacing #TBD with the issue number and dropping the intra-fork banner line. No separate backlink comment — Closes #N does the linking.
4. Afterwards: file a scoped follow-up issue for the schema-level (table) metadata inconsistency (Union::try_new intersects it; coerce_union_schema/union_schema still merge it) and link it from the main issue.

Notes for your review: the choreography (same-person issue+PR minutes apart, take, no backlink) matches current repo practice (apache#23242→apache#23243, apache#23077→apache#23078, apache#23160→apache#23161). The fix deliberately does NOT touch schema-level metadata or gate behind datafusion.execution.skip_physical_aggregate_schema_check — that flag would disable the safety check globally instead of fixing the divergence.

[gruuya]

Union::try_new intersects field metadata across inputs with intersect_metadata_for_union

I don't think this is true; intersect_metadata_for_union has been applied only at the logical planning level.

In fact, this is what I'd push back against: can we fix the original problem without shoehorning intersect_metadata_for_union at the physical (analyzer) level?

Maybe we can just extend the logic in intersect_metadata_for_union?

[milesrichardson-edb]

@copilot Give a review in context of the feedback from @gruuya 's comment.

[Copilot]

@copilot Give a review in context of the feedback from @gruuya 's comment.

I reviewed it, and I don’t think @gruuya’s feedback requires a code change here.

Union::try_new already uses intersect_metadata_for_union today in datafusion/expr/src/logical_plan/plan.rs, so that part of the PR text is accurate.

On the underlying design point, extending intersect_metadata_for_union alone would not fix this failure. The internal error comes from Aggregate comparing the logical input schema against the schema recomputed by physical union_schema(). Before this patch, coerce_union_schema_with_schema and union_schema() merged field metadata differently, so they could still diverge even if Union::try_new stayed as-is. To eliminate the mismatch, the physical path either has to adopt the same field-metadata semantics as the logical path (what this patch does) or stop recomputing union field metadata independently. That makes this feedback something to clarify in the discussion rather than a reason to rework the patch.