Option to disable `Physical input schema should be the same as the one converted from logical schema` error

[alamb]

Is your feature request related to a problem or challenge?

This bug, released in DataFusion 42.0.0 ,

• Fix the schema mismatch between logical and physical for aggregate function, add AggregateUDFImpl::is_null #11989

Added a new check in the DefaultPhysicalPlanner that the schema of the output plan is the same as the input plan

datafusion/datafusion/core/src/physical_planner.rs

Lines 660 to 662 in 818ce3f

	if physical_input_schema != physical_input_schema_from_logical {
	return internal_err!("Physical input schema should be the same as the one converted from logical input schema.");
	}

While @jayzhan211 's heroic efforts has this passing in all the DataFusion tests, it turned out this check failed on many downstream implementations:

• [EPIC] Schema metadata handling / bugs #12733 lists issues we hit during our upgrade in InfluxDB 3.0
• chore: upgrade to datafusion 43 delta-io/delta-rs#2886 @ion-elgreco and @rtyler and @Xuanwo while updating delta.rs to DataFusion 42.0.0

Downstream in InfluxDB 3.0 we turned the check into a warning in our fork to unblock our upgrade

We even made a patch release to try and get the delta-rs upgrade working:

• Release DataFusion 42.1.0 #12813

But it is still failing when I write this (see delta-io/delta-rs#2886 (comment))

Internal error: Failed due to a difference in schemas, original schema: DFSchema { inner: Schema { fields: [Field { name: "id", data_type: Utf8, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }, Field { name: "price", data_type: Int64, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }, Field { name: "sold", data_type: Int64, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }, Field { name: "price_float", data_type: Float64, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }, Field { name: "items_in_bucket", data_type: List(Field { name: "element", data_type: Utf8, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }), nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }, Field { name: "deleted", data_type: Boolean, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }, Field { name: "__delta_rs_update_predicate", data_type: Boolean, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }], metadata: {} }, field_qualifiers: [None, Some(Bare { table: "target" }), None, None, None, None, None], functional_dependencies: FunctionalDependencies { deps: [] } }, new schema: DFSchema { inner: Schema { fields: [Field { name: "id", data_type: Utf8, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }, Field { name: "price", data_type: Int64, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }, Field { name: "sold", data_type: Int64, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }, Field { name: "price_float", data_type: Float64, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }, Field { name: "items_in_bucket", data_type: List(Field { name: "item", data_type: Utf8, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }), nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }, Field { name: "deleted", data_type: Boolean, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }, Field { name: "__delta_rs_update_predicate", data_type: Boolean, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }], metadata: {} }, field_qualifiers: [None, Some(Bare { table: "target" }), None, None, None, None, None], functional_dependencies: FunctionalDependencies { deps: [] } }.

Describe the solution you'd like

Note there is at least one open outstanding bug: #13010

I would like some way to disable this check to unblock upgrades in downstream crates.

Describe alternatives you've considered

I propose we add a new config value that lets downstream crates opt in / out of this check, similarly to datafusion.optimizer.skip_failed_rules (see Config Docs)

Something like:

• datafusion.execution.validate_schema: If true, the DefaultPhysicalPlanner will error if the input plan's schema does not exactly match the output plan.

Additional context

No response

[alamb]

I should clarify -- ideally this check should be enabled by default and it is a goal we should shoot for.

However, as there are clearly bugs in the code that currently prevent it from passing cleanly in all cases (which were previously in the code), I think it is better to relax the check and sort out the errors rather than hard failing plans.

[wiedld]

However, as there are clearly bugs in the code that currently prevent it from passing cleanly in all cases (which were previously in the code), I think it is better to relax the check and sort out the errors rather than hard failing plans.

Some additional context -- we have definitely not isolated all the bugs this check uncovered. Even with this know bug (#13010) patched for us, we are still encountering this failed check every few minutes. Changing this check to a warning, not an error, has been necessary for us.

Assuming that we are not the only ones, having the feature @alamb proposed here (to also convert to a warning based on configuration) would help unblock others from ungrading datafusion IMO.

[alamb]

We (InfluxData) will likely contribute fixes back upstream into DataFusion as we find issues as well

[comphead]

It is probably happens when users utilize only physical planner, although there are schema alignments that happening on logical planner?

[comphead]

I remember bunch of issues on schema comparison when different null flag for the column caused a problem.Since DF reworked the Schema methods this issue might come up again

[findepi]

@wiedld are you able to capture failing queries (or rather: queries giving a warning) and maybe turn some into actionable issues for the community?

[timsaucer]

The question I'd want answered is: does disabling this check hide an actual problem or are we confident the problem exists within the check itself? That is, do we know for certain that disabling the check isn't allowing faulty output? In general I recommend against features to disable checks but I also recognize that unblocking downstream users may be more important.

[alamb]

The question I'd want answered is: does disabling this check hide an actual problem or are we confident the problem exists within the check itself?

I believe disabling the check hides actual problems -- a list of such problems is here #12733

However, those plans ran in DataFusion 41 (and sometimes metadata was dropped from the output schema) and now the plans refuse to run.

So depending on your perspective this is either an improvement or regression:

• It is a regression for a user running SQL queries perspective because a query that used to run great no longer does
• It is an improvement for a developer as it now makes it easy to find queries that violate invariants

So having a flag to change behaviors based on your point of view I think makes sense

That is, do we know for certain that disabling the check isn't allowing faulty output? In general I recommend against features to disable checks but I also recognize that unblocking downstream users may be more important.

No, we do not know this for certain. I think my point of view is that running with the check disabled is no worse than DataFusion 41 so while not ideal, it isn't worse

[alamb]

It is probably happens when users utilize only physical planner, although there are schema alignments that happening on logical planner?

There are two classes of errors we have seen so far in InfluxDB:

1. Null declarations as predicated by @comphead
2. Metadata mismatches (metadata is dropped by some physical optimizer)

I believe delta-rs is also seeing field name mismatches (e.g. the embedded fieldname on ListArray is "item" vs "element" or something)

[comphead]

Before DF 41 when doing a schema comparison we relied on helper methods which excludes nullability and metadata comparison. I suppose this method is gone after the refactoring and we need to revive it and use for schema comparison

[wiedld]

@wiedld are you able to capture failing queries (or rather: queries giving a warning) and maybe turn some into actionable issues for the community?

@findepi I'll try to get us actionable issues this week. Specifically, I need to isolate reproducers in either our code base (e.g. our physical optimizers) versus datafusion.