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 80.98%. Comparing base (33ad1cc) to head (4667203).
⚠️ Report is 1 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.99%   80.98%   -0.01%     
==========================================
  Files        1106     1106              
  Lines      383352   383495     +143     
  Branches   383352   383495     +143     
==========================================
+ Hits       310488   310581      +93     
- Misses      54544    54578      +34     
- Partials    18320    18336      +16     

☔ 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.

[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

[dariocurr]

Made this change. CoerceSchemaExec is gone. UnionExec/InterleaveExec now insert a ProjectionExec with CastExpr::new_with_target_field (zero-copy for same-type casts), or a plain Column when a leg's field already matches.

A few side effects:

• No more wrapper-erasure trick in proto serialization, ProjectionExec has an ordinary proto message.
• The projection-collapsing optimizer pass fuses the coercion into a leg's existing top-level ProjectionExec for free in several cases.
• Fixed a real gap this surfaced: ProjectionExec statistics were zeroing sum_value/byte_size for same-type casts. Now passed through exactly, with a test.

[alamb]

I merged up to resolve some conflicts

[alamb]

Thank you @dariocurr and @kosiew