physical-plan: coerce UNION/INTERLEAVE schema mismatches at plan time

[dariocurr]

Which issue does this PR close?

Follow-up to #23861 (issue #15394). Not closing a new issue.

Rationale for this change

#23861 fixed UNION ALL batches carrying the wrong nullability when one leg
is NOT NULL and another isn't, by re-stamping each batch's schema inside
UnionExec/InterleaveExec's own execute(). In review, @alamb noted:

ideally we could coerce the schema at plan time but I don't know how to
coerce nullability

This PR does that: the coercion becomes an explicit node in the plan tree,
inserted when the plan is built, instead of invisible logic inside
execute().

What changes are included in this PR?

• Adds CoerceSchemaExec, a single-child passthrough ExecutionPlan node.
  UnionExec::try_new/InterleaveExec::try_new insert it above any child
  whose own output schema disagrees with the computed union schema (in
  practice, only nullability differs -- UnionExec::try_new already rejects
  real data-type mismatches via calculate_union).
• The actual batch re-stamping logic (SchemaConformingStream) is unchanged;
  it just lives under CoerceSchemaExec::execute() now instead of being
  called directly from UnionExec/InterleaveExec::execute().
• Because it's a real plan node, CoerceSchemaExec implements the full
  ExecutionPlan surface a pure 1:1 passthrough needs to stay transparent
  to the optimizer: statistics passthrough, filter/limit pushdown,
  benefits_from_input_partitioning() -> false (so it doesn't trigger a
  spurious repartition), and proto (de)serialization -- the node erases
  itself on encode and is reconstructed by try_new on decode, so no
  protobuf schema change was needed.
• A genuine data-type mismatch (as opposed to nullability-only) is now
  rejected eagerly at plan-build time (via EquivalenceProperties:: with_new_schema) rather than lazily at execute().

Are these changes tested?

• New unit test test_union_partition_statistics_with_mismatched_nullability
  in union.rs, proving statistics aren't poisoned to Absent through the
  new node.
• Existing union_nullable/union_nullable_spill regression tests from
  fix: UnionExec now conforms each batch to the union's declared schema #23861 continue to pass unchanged.
• Updated the sqllogictest golden file (union.slt) where EXPLAIN
  output now shows the new node for pre-existing nullability-mismatched
  UNION ALL cases.
• Benchmarked against the previous (inline) approach: no measurable
  performance difference in either the coerced or matched-schema case
  (differences were within run-to-run noise).

Are there any user-facing changes?

EXPLAIN output for a UNION ALL/interleaved plan with a nullability
mismatch across legs will now show a CoerceSchemaExec node that wasn't
there before. No behavioral or correctness change.

[codecov-commenter]

Codecov Report

❌ Patch coverage is 88.32117% with 16 lines in your changes missing coverage. Please review.
✅ Project coverage is 81.02%. Comparing base (c4a539b) to head (bc2f35f).
⚠️ Report is 4 commits behind head on main.

Files with missing lines	Patch %	Lines
datafusion/physical-plan/src/union.rs	87.38%	2 Missing and 12 partials ⚠️
datafusion/physical-expr/src/projection.rs	92.30%	1 Missing and 1 partial ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main   #24094      +/-   ##
==========================================
+ Coverage   80.91%   81.02%   +0.11%     
==========================================
  Files        1103     1105       +2     
  Lines      377134   379804    +2670     
  Branches   377134   379804    +2670     
==========================================
+ Hits       305155   307749    +2594     
- Misses      53787    53821      +34     
- Partials    18192    18234      +42     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[kosiew]

@dariocurr,

Thanks for working on this. The plan-visible CoerceSchemaExec approach looks good, and I like that it removes the schema restamping from UnionExec::execute() and InterleaveExec::execute() while keeping the coercion explicit in the plan.

I left one non-blocking suggestion for additional protobuf round-trip coverage. Otherwise, this looks good to me.

[kosiew]

Could we add a protobuf round-trip regression test for nullable and non-nullable UNION and INTERLEAVE inputs?

CoerceSchemaExec::try_to_proto intentionally leaves the wrapper out of the serialized plan, while UnionExec::try_from_proto and InterleaveExec::try_from_proto rebuild it through try_new. It would be helpful to verify that the decoded plans contain the expected coercion and that their emitted batches expose the final nullable schema.

[dariocurr]

Added roundtrip_union_with_mismatched_nullability_executes and roundtrip_interleave_with_mismatched_nullability_executes — one non-nullable + one nullable literal leg, serialize/deserialize, confirm CoerceSchemaExec reappears in the decoded plan, then execute it and check every emitted batch carries the nullable schema.

[alamb]

Thanks for working on this. The plan-visible CoerceSchemaExec approach looks good, and I like that it removes the schema restamping from UnionExec::execute() and InterleaveExec::execute() while keeping the coercion explicit in the plan.

I wonder if instead of using an entirely new exec, we could use the existing ProjectionExec operator with Cast exprs? In theory it should already be capable of doing the schema tranformation (assuming the physical CastExpr can have a field -- not just a DataType)

Thank you for follow up on this @dariocurr and @kosiew