Error with union and optimize_projections: Physical input schema should be the same as the one converted from logical input schema

Describe the bug

DataFusion fails with schema mismatch error when processing UNION ALL query on parquet files with field metadata.

Error while planning query: Internal error: Physical input schema should be the same as the one converted from logical input schema. Differences: .

To Reproduce

Parquet: union_all_repo.zip

Query:

SET datafusion.execution.parquet.skip_metadata = false;

SELECT AVG(usage_idle), AVG(usage_system)
FROM (
    SELECT time, usage_idle, NULL::DOUBLE as usage_system FROM 'union_all_repro.parquet'
    UNION ALL
    SELECT time, NULL::DOUBLE as usage_idle, usage_system FROM 'union_all_repro.parquet'
);

Expected behavior

Query should succeed with matching logical and physical schema

Additional context

Sequence of Events

1. Projection Optimization

optimize_projections sees that time column is unused in the projection and removes it from the UnionExec

+------------------------------------------------------------+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| plan_type                                                  | plan                                                                                                                                                                                |
+------------------------------------------------------------+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| initial_logical_plan                                       | Projection: avg(usage_idle), avg(usage_system)                                                                                                                                      |
|                                                            |   Aggregate: groupBy=[[]], aggr=[[avg(usage_idle), avg(usage_system)]]                                                                                                              |
|                                                            |     Union                                                                                                                                                                           |
|                                                            |       Projection: ./union_all_repro.parquet.time, ./union_all_repro.parquet.usage_idle, CAST(NULL AS Float64) AS usage_system                                                       |
|                                                            |         TableScan: ./union_all_repro.parquet                                                                                                                                        |
|                                                            |       Projection: ./union_all_repro.parquet.time, CAST(NULL AS Float64) AS usage_idle, ./union_all_repro.parquet.usage_system                                                       |
|                                                            |         TableScan: ./union_all_repro.parquet                                                                                                                                        |
| logical_plan after resolve_grouping_function               | SAME TEXT AS ABOVE                                                                                                                                                                  |
| logical_plan after type_coercion                           | SAME TEXT AS ABOVE                                                                                                                                                                  |
| analyzed_logical_plan                                      | SAME TEXT AS ABOVE                                                                                                                                                                  |
| logical_plan after eliminate_nested_union                  | SAME TEXT AS ABOVE                                                                                                                                                                  |
| logical_plan after simplify_expressions                    | Projection: avg(usage_idle), avg(usage_system)                                                                                                                                      |
|                                                            |   Aggregate: groupBy=[[]], aggr=[[avg(usage_idle), avg(usage_system)]]                                                                                                              |
|                                                            |     Union                                                                                                                                                                           |
|                                                            |       Projection: ./union_all_repro.parquet.time, ./union_all_repro.parquet.usage_idle, Float64(NULL) AS usage_system                                                               |
|                                                            |         TableScan: ./union_all_repro.parquet                                                                                                                                        |
|                                                            |       Projection: ./union_all_repro.parquet.time, Float64(NULL) AS usage_idle, ./union_all_repro.parquet.usage_system                                                               |
|                                                            |         TableScan: ./union_all_repro.parquet                                                                                                                                        |
| logical_plan after replace_distinct_aggregate              | SAME TEXT AS ABOVE                                                                                                                                                                  |
| logical_plan after eliminate_join                          | SAME TEXT AS ABOVE                                                                                                                                                                  |
| logical_plan after decorrelate_predicate_subquery          | SAME TEXT AS ABOVE                                                                                                                                                                  |
| logical_plan after scalar_subquery_to_join                 | SAME TEXT AS ABOVE                                                                                                                                                                  |
| logical_plan after decorrelate_lateral_join                | SAME TEXT AS ABOVE                                                                                                                                                                  |
| logical_plan after extract_equijoin_predicate              | SAME TEXT AS ABOVE                                                                                                                                                                  |
| logical_plan after eliminate_duplicated_expr               | SAME TEXT AS ABOVE                                                                                                                                                                  |
| logical_plan after eliminate_filter                        | SAME TEXT AS ABOVE                                                                                                                                                                  |
| logical_plan after eliminate_cross_join                    | SAME TEXT AS ABOVE                                                                                                                                                                  |
| logical_plan after eliminate_limit                         | SAME TEXT AS ABOVE                                                                                                                                                                  |
| logical_plan after propagate_empty_relation                | SAME TEXT AS ABOVE                                                                                                                                                                  |
| logical_plan after eliminate_one_union                     | SAME TEXT AS ABOVE                                                                                                                                                                  |
| logical_plan after filter_null_join_keys                   | SAME TEXT AS ABOVE                                                                                                                                                                  |
| logical_plan after eliminate_outer_join                    | SAME TEXT AS ABOVE                                                                                                                                                                  |
| logical_plan after push_down_limit                         | SAME TEXT AS ABOVE                                                                                                                                                                  |
| logical_plan after push_down_filter                        | SAME TEXT AS ABOVE                                                                                                                                                                  |
| logical_plan after single_distinct_aggregation_to_group_by | SAME TEXT AS ABOVE                                                                                                                                                                  |
| logical_plan after eliminate_group_by_constant             | SAME TEXT AS ABOVE                                                                                                                                                                  |
| logical_plan after common_sub_expression_eliminate         | SAME TEXT AS ABOVE                                                                                                                                                                  |
| logical_plan after optimize_projections                    | Aggregate: groupBy=[[]], aggr=[[avg(usage_idle), avg(usage_system)]]                                                                                                                |
|                                                            |   Union                                                                                                                                                                             |
|                                                            |     Projection: ./union_all_repro.parquet.usage_idle, Float64(NULL) AS usage_system                                                                                                 |
|                                                            |       TableScan: ./union_all_repro.parquet projection=[usage_idle]                                                                                                                  |
|                                                            |     Projection: Float64(NULL) AS usage_idle, ./union_all_repro.parquet.usage_system                                                                                                 |
|                                                            |       TableScan: ./union_all_repro.parquet projection=[usage_system]

2. Schema Recomputation

optimize_projections calls recompute_schema since the plan has changed. recompute_schema sees that the number of fields has changed and creates a new Union node with Union::try_new

Union::try_new calls Union::derive_schema_from_inputs to recreate the Union schema from the child inputs. For each field in the schema, it calls intersect_metadata_for_union which only keeps metadata when the field has the same metadata across all child inputs. Since the NULL literal doesn't have the same metadata as our column from parquet, the metadata gets removed from the logical schema.

3. Physical/Logical Schema Mismatch

During physical planning, we call DefaultPhysicalPlanner::map_logical_node_to_physical which checks the input of the Aggregate node (in this case Union) to see if its logical schema matches its physical schema. However, because we previously removed the metadata from Union, the metadata no longer matches when compared.

nice find

With the error message changes in #19070 applied, the reproducer described in this issue now gives a more helpful error message:

Internal error: Physical input schema should be the same as the one converted from logical input schema. Differences:
        - field metadata at index 0 [usage_idle]: (physical) {"iox::column::type": "iox::column_type::field::float"} vs (logical) {}
        - field metadata at index 1 [usage_system]: (physical) {"iox::column::type": "iox::column_type::field::float"} vs (logical) {}.
This was likely caused by a bug in DataFusion's code and we would welcome that you file an bug report in our issue tracker

@erratic-pattern

@erratic-pattern I can't assign this to you can you do so

@alamb

@erratic-pattern I can't assign this to you can you do so

I also cannot. Perhaps @alamb can?

@erratic-pattern I can't assign this to you can you do so

I also cannot. Perhaps @alamb can?

Done!

(you can also assign yourself tickets by issuing a comment take: https://datafusion.apache.org/contributor-guide/index.html#open-contribution-and-assigning-tickets

It's not clear to me whether or not this is an actual bug. It seems reasonable to expect metadata to be consistent for field names across union branches. However it could also be problematic for queries that either:

Shadow existing column names with new, unrelated columns. Variable shadowing in Rust is very normal, but I'm not sure if this is considered reasonable in SQL queries.
Queries that introduce constant literals for fields on one side of a union, such as this reproducer.

Perhaps we need to adjust intersect_metadata_for_union to either:

Avoid intersecting a branch that contains empty metadata, and instead preserve/intersect only the branches that contain non-empty metadata. This avoids destructive loss of metadata when one union branch is empty.
Avoid intersecting a branch that contains empty metadata on a field that is a constant literal. This is a more restrictive version of option 1 that might result in fewer unintended consequences.
union the metadata instead of intersecting the metadata. This ensures there is no metadata lost, but I am not sure what consequences this might have since it could populate metadata in the output schema when it was not intended to be there.

I am curious if anyone has opinions about any of these approaches, or if there is another way to look at this.

@adriangb

It's not clear to me whether or not this is an actual bug. It seems reasonable to expect metadata to be consistent for field names across union branches. However it could also be problematic for queries that either:

In my mind it is a bug because the query is very reasonable -- I think there are other types of plans in queries where the metadata from different inputs needs to be combined (and thus might not be the same at the input and output -- for example joins and aggregates)

It seems like there is an implicit assumption that "the schema (including metadata) of a plan should remain the same after an optimizer pass". If this is indeed correct, then by your analysis above

optimize_projections calls recompute_schema since the plan has changed. recompute_schema sees that the number of fields has changed and creates a new Union node with Union::try_new

It seems like we should fix optimize_projections so it maintains the schema (either by attaching metadata to the NULL literal, or perhaps by simply reusing the previous schema).

For example, @adriangb just aded code that does something similar (though during execution) in this PR #19111 :

datafusion/datafusion/datasource-parquet/src/opener.rs

Lines 529 to 545 in bde1608

    
           // Ensure the output batch has the expected schema. 
        
           // This handles things like schema level and field level metadata, which may not be present 
        
           // in the physical file schema. 
        
           // It is also possible for nullability to differ; some writers create files with 
        
           // OPTIONAL fields even when there are no nulls in the data. 
        
           // In these cases it may make sense for the logical schema to be `NOT NULL`. 
        
           // RecordBatch::try_new_with_options checks that if the schema is NOT NULL 
        
           // the array cannot contain nulls, amongst other checks. 
        
           let (_stream_schema, arrays, num_rows) = b.into_parts(); 
        
           let options = 
        
               RecordBatchOptions::new().with_row_count(Some(num_rows)); 
        
           RecordBatch::try_new_with_options( 
        
               Arc::clone(&output_schema), 
        
               arrays, 
        
               &options, 
        
           ) 
        
           .map_err(Into::into)

Current error is

> SET datafusion.execution.parquet.skip_metadata = false;

SELECT AVG(usage_idle), AVG(usage_system)
FROM (
    SELECT time, usage_idle, NULL::DOUBLE as usage_system FROM 'union_all_repro.parquet'
    UNION ALL
    SELECT time, NULL::DOUBLE as usage_idle, usage_system FROM 'union_all_repro.parquet'
);
0 row(s) fetched.
Elapsed 0.013 seconds.

Internal error: Physical input schema should be the same as the one converted from logical input schema. Differences:
	- field metadata at index 0 [usage_idle]: (physical) {"iox::column::type": "iox::column_type::field::float"} vs (logical) {}
	- field metadata at index 1 [usage_system]: (physical) {"iox::column::type": "iox::column_type::field::float"} vs (logical) {}.
This issue was likely caused by a bug in DataFusion's code. Please help us to resolve this by filing a bug report in our issue tracker: https://github.com/apache/datafusion/issues

fix: skip empty metadata in intersect_metadata_for_union to prevent s…

We hit this while running benchmarks against Parquet files with field metadata on a SQL engine we're building on DataFusion. Opened a PR: #21127

fix: add SLT reproducer for apache#19049 and fix cargo fmt

fix: skip empty metadata in intersect_metadata_for_union to prevent s… (

backport fix for issue apache#19049

fix: skip empty metadata in intersect_metadata_for_union to prevent s… (

Navigation Menu

Repository navigation

Error with union and optimize_projections: Physical input schema should be the same as the one converted from logical input schema #19049

Describe the bug

To Reproduce

Expected behavior

Additional context

Sequence of Events

1. Projection Optimization

2. Schema Recomputation

3. Physical/Logical Schema Mismatch

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

	// Ensure the output batch has the expected schema.
	// This handles things like schema level and field level metadata, which may not be present
	// in the physical file schema.
	// It is also possible for nullability to differ; some writers create files with
	// OPTIONAL fields even when there are no nulls in the data.
	// In these cases it may make sense for the logical schema to be `NOT NULL`.
	// RecordBatch::try_new_with_options checks that if the schema is NOT NULL
	// the array cannot contain nulls, amongst other checks.
	let (_stream_schema, arrays, num_rows) = b.into_parts();
	let options =
	RecordBatchOptions::new().with_row_count(Some(num_rows));
	RecordBatch::try_new_with_options(
	Arc::clone(&output_schema),
	arrays,
	&options,
	)
	.map_err(Into::into)

Error with union and optimize_projections: Physical input schema should be the same as the one converted from logical input schema #19049

Description

Describe the bug

To Reproduce

Expected behavior

Additional context

Sequence of Events

1. Projection Optimization

2. Schema Recomputation

3. Physical/Logical Schema Mismatch

Activity

alamb commented on Dec 2, 2025

kallisti-dev commented on Dec 3, 2025

helenosheaa commented on Dec 8, 2025

kallisti-dev commented on Dec 8, 2025

alamb commented on Dec 9, 2025

kallisti-dev commented on Dec 11, 2025

alamb commented on Dec 11, 2025

alamb commented on Jan 13, 2026

RafaelHerrero commented on Mar 23, 2026

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions