fix: Derive Substrait intersection nullability from every input

[namanjain24-sudo]

Which issue does this PR close?

• Closes Substrait intersection schemas retain primary-input nullability #25042.

Rationale for this change

The Substrait consumer derived all three intersection schemas from the primary
input alone, so a field the intersection makes required stayed nullable in the
logical output schema.

The Set Operation rules give intersections a different rule: for the multiset
intersections a field is required as soon as any input requires it, and for
INTERSECTION_PRIMARY it is nullable only when it is nullable in the primary
input and in at least one secondary input.

from_set_rel builds intersections with LogicalPlanBuilder::intersect, which
compiles to a left semi join and therefore keeps the left input's nullability.

What changes are included in this PR?

Intersections now go through a small helper that gives the result
left AND right nullability per field.

The join matches nulls with nulls (NullEquality::NullEqualsNull) on every
field, so a left row holding a null in some field only survives when the right
input holds a null there too — a field is nullable in the result only when both
inputs make it nullable. Applied to each step, that single rule reproduces both
spec rules:

• the multiset intersections chain pairwise, so the result is required when any
  input requires it;
• for INTERSECTION_PRIMARY the right side is the union of the secondary
  inputs, whose field is nullable exactly when some secondary makes it nullable,
  which yields "nullable in the primary and in at least one secondary".

How the narrowing is built matters, because it has to hold in the physical plan
too, not only in the logical schema. When the right input requires a field that
the left input leaves nullable, the intersection is built as an inner join
(nulls equal nulls) against the distinct right rows, and that field is read from
the right side:

• matched rows hold equal values, so the result rows are unchanged;
• the field is non-nullable because its source column is, and the logical and
  the physical planner both derive that from the input schema, so the logical
  schema, the physical plan and the batches agree without any schema override;
• joining against distinct right rows keeps each left row at most once, as the
  semi join does (and the left side is still made distinct unless is_all).

A field is only read from the right side when its type matches the left's; its
metadata does not matter, since metadata is no part of nullability. The result
should still describe the left input, so where the inputs' metadata disagree the
left's wins: a column read from the right is aliased with the left field's
metadata, and an inner join's schema lets the left input's schema metadata win
(in the logical and in the physical plan alike). Keys that only the right input
carries, on a field or on the schema, are merged into the result.

When nothing needs narrowing, the plan is exactly the one
LogicalPlanBuilder::intersect builds today, so the common all-nullable case is
untouched. Unions and the MINUS operations are not affected.

What is the testing strategy for this PR?

New test intersect_nullability in datafusion/substrait/tests/cases/logical_plans.rs,
with three plans added under tests/testdata/test_plans/. They intersect three
tables carrying the same six columns (? marks nullable, ~ marks
NULLABILITY_UNSPECIFIED, which the consumer reads as nullable):

primary     a? b? c? d? e? f?
secondary   a  b  c? d? e~ f~
secondary   a  b? c  d? e? f

Operation	Result
INTERSECTION_PRIMARY	a, b?, c?, d?, e?, f?
INTERSECTION_MULTISET	a, b, c, d?, e?, f
INTERSECTION_MULTISET_ALL	a, b, c, d?, e?, f

Columns e and f pin the unspecified-nullability boundary: if it were read as
required, e would come out required for the multiset intersections and f
for INTERSECTION_PRIMARY.

The tables are given rows, including nulls, and for each plan the test checks
that the logical schema, the physical plan's schema and the schema of every
collected batch are equal, and checks the result rows. It fails on main
(a? where a is expected), and the physical-schema check also fails against
a narrowing done with a logical-only Projection.

Each plan also runs a second time over tables with differing schema and field
metadata: the secondary tables share their metadata, which differs from the
primary's and adds a key the primary lacks. That run checks that the primary's
metadata wins on the keys the tables disagree on, and that the logical schema,
the physical plan's schema and the collected batches' schemas agree. It fails
without the metadata handling (a? where a is expected).

The secondary tables share their metadata on purpose: a Union of inputs with
differing field metadata can report different metadata in its logical and in its
physical schema. That is independent of this change (plain SQL,
SELECT a FROM t1 UNION ALL SELECT a FROM t2, shows it on a checkout without
it), and I have not root-caused it.

With the spec's full input pattern from the issue (R = required,
N = nullable), both the logical and the physical schema now match the
expected column:

primary:     R R R R N N N N
secondary 1: R R N N R R N N
secondary 2: R N R N R N R N

INTERSECTION_PRIMARY       R R R R R N N N
INTERSECTION_MULTISET      R R R R R R R N
INTERSECTION_MULTISET_ALL  R R R R R R R N

The existing datafusion-substrait suite passes unchanged, including the
intersection roundtrip tests, and a Substrait round trip of the sqllogictest
files that use INTERSECT gives the same output as main.

Are there any user-facing changes?

Intersections consumed from Substrait now report a narrower, spec-conforming
nullability. No public API changes.

One note for reviewers: the same narrowing would apply to SQL INTERSECT, since
LogicalPlanBuilder::intersect keeps the left nullability for every caller. I
kept this change inside the Substrait consumer to match the scope of the issue
and to avoid changing SQL plans in the same PR. If you would rather see the rule
live in LogicalPlanBuilder, I am happy to move it.

[codecov-commenter]

Codecov Report

❌ Patch coverage is 89.83051% with 6 lines in your changes missing coverage. Please review.
✅ Project coverage is 82.42%. Comparing base (edc936f) to head (2a8d8f5).
⚠️ Report is 57 commits behind head on main.

Files with missing lines	Patch %	Lines
...substrait/src/logical_plan/consumer/rel/set_rel.rs	89.83%	2 Missing and 4 partials ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main   #25091      +/-   ##
==========================================
+ Coverage   82.33%   82.42%   +0.08%     
==========================================
  Files        1137     1140       +3     
  Lines      432498   435639    +3141     
  Branches   432498   435639    +3141     
==========================================
+ Hits       356116   359080    +2964     
+ Misses      54843    54835       -8     
- Partials    21539    21724     +185     

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

@namanjain24-sudo,

Thanks for working on this. The logical nullability handling looks like it is heading in the right direction, but I think there is still a schema consistency issue between the logical and physical plans that needs to be addressed before this is ready.

I left one blocking comment on the physical projection behavior, plus one small test suggestion.

[kosiew]

I don't think this Projection can carry the nullability narrowing through to execution. DefaultPhysicalPlanner creates it with ProjectionExec::try_new_with_schema_metadata, and that API preserves nullability from the physical column expressions and input. The supplied schema is only used for metadata.

That means the left semi join can still expose nullable fields in the physical plan and collected RecordBatch schema, while plan.schema() says those fields are required. This leaves the logical and physical schemas inconsistent.

Could we use an execution-plan or schema adapter that deliberately rebinds the batches to the proven non-null schema while preserving the exact field attributes, or extend the physical projection path so this validated narrowing is supported there as well?

It would also be good to add a regression assertion that checks both the physical-plan schema and the collected batch schema. The current show() call exercises execution, but it does not verify that those schemas actually match the narrowed logical schema.

[kosiew]

Could we also add a small NULLABILITY_UNSPECIFIED case, ideally on one of the secondary inputs? Since the consumer treats unspecified nullability as nullable, this would lock down that Substrait boundary in addition to the explicit REQUIRED and NULLABLE cases.

[namanjain24-sudo]

Thanks @kosiew, you were right. I added the assertion you suggested and it failed on the previous commit: the logical schema had a as required, but create_physical_plan().schema() still had it nullable, because ProjectionExec takes only metadata from the supplied schema.

Instead of a rebinding exec or a change to ProjectionExec, the latest commit removes the need to override the schema at all:

• A new execution plan or schema adapter would need its own logical node, and DefaultPhysicalPlanner can't plan that unless an ExtensionPlanner is registered. A consumed Substrait plan has to run in any SessionContext.
• Letting ProjectionExec narrow nullability would change every projection in core. The narrowing would also have to survive the physical optimizer's rebuild and removal paths, like the metadata override from fix: preserve projection field metadata during physical planning #23981 does. That felt like too much for a Substrait consumer fix.

What it does now: when the right input requires a field that the left input leaves nullable, the intersection is built as an inner join (nulls equal nulls) against the distinct right rows, and that field is read from the right side. Matched rows hold equal values, so the result rows don't change. The field is non-nullable because its source column is, and both the logical and the physical planner derive that from the input schema, so the two stay consistent without any override. Joining against distinct right rows keeps each left row at most once, the same as the semi join. A field is only taken from the right when its type and metadata match the left's (and the schema metadata matches too), so the result's attributes stay the left input's. When nothing needs narrowing, it still uses LogicalPlanBuilder::intersect as before.

Tests:

• The tables now have rows (including nulls). For each of the three plans, the test checks that the logical schema, create_physical_plan().schema() and the schema of every collected batch are equal, and checks the result rows.
• Added NULLABILITY_UNSPECIFIED on a secondary input (columns e and f). e would come out required for the multiset intersections, and f required for INTERSECTION_PRIMARY, if unspecified were read as required.

[kosiew]

@namanjain24-sudo,

Thanks for working on this. The inner-join approach looks good for the cases covered by the new tests, and the added physical-plan and batch-schema assertions are helpful. The NULLABILITY_UNSPECIFIED secondary case is also covered now.

There is still one path where the original nullability issue can come back, though. When schema or field metadata differs between the two inputs, the implementation falls back to LogicalPlanBuilder::intersect, which uses the left-semi plan and keeps the nullable left-side schema. Since ensure_schema_compatibility does not require metadata equality, these inputs can still be valid.

I think the narrowing path should continue to be used even when metadata differs, while explicitly preserving the left-side metadata on projected right-side columns. It would also be good to add a regression test with differing schema and field metadata that checks the logical schema, physical schema, and collected batch schema all match.

Once that case is handled, this should address the reported issue more completely.

[kosiew]

I think this fallback still leaves the original bug reachable. If the schema metadata differs, we fall back to LogicalPlanBuilder::intersect, which uses the left-semi plan and keeps the nullable left-side field. ensure_schema_compatibility only checks type and nullability compatibility, so differing metadata can still be valid here.

Could we keep using the narrowing path and explicitly preserve the left-side metadata when projecting a required column from the right? The alias API supports attaching metadata, so that should let us retain the left schema metadata without giving up the nullability fix.

The same issue can happen with field metadata: if a nullable left field and required right field differ only in metadata, that field will not be selected from the right unless some other field happens to trigger this path.

It would be useful to add a regression test with differing schema and field metadata and assert the exact logical, physical, and collected-batch schemas.

[namanjain24-sudo]

Thanks @kosiew, good catch. Pushed 650a5de:

• Dropped the metadata-equality conditions: a field is read from the right whenever the left leaves it nullable and the right requires it, and the schema-metadata fallback to LogicalPlanBuilder::intersect is gone, so the narrowing path is always taken.
• A column read from the right is aliased with the left field's metadata. An inner join's schema (logical and physical) already lets the left side's schema metadata win, so on conflicting keys the left's metadata wins.
• intersect_nullability now runs each plan a second time over tables with differing schema and field metadata. It checks that the logical, physical and collected-batch schemas agree, and that the primary's metadata wins where the tables disagree. It fails without the change (a? stays nullable).

Two things to flag:

1. Alias metadata is merged over the right field's metadata, not replacing it, so keys present only on the right (field or schema level) still appear in the result. I first tried stating the exact left schema with an explicit-schema Projection, but the optimizer re-derives it and the physical plan then disagreed with the logical one. Happy to go another way if you want exact-left semantics.
2. While writing the test I saw that a Union of inputs with differing field metadata can report different metadata in its logical and in its physical schema, and in my test that tripped the aggregate schema check. It is independent of this change: plain SQL reproduces the mismatch on a checkout without it (SELECT a FROM t1 UNION ALL SELECT a FROM t2, where t1.a and t2.a carry different field metadata: the logical field has none, the physical one has t2's). So the test gives the secondary tables the same metadata. I haven't root-caused it; I can file a separate issue.

Edited: an earlier version of this comment said which input's metadata each schema reports, which I had only seen in one case and which does not hold in general.

Locally I ran cargo test -p datafusion-substrait, cargo fmt, and clippy with --all-features -D warnings on that crate (with -A clippy::unused_async_trait_impl, which datafusion-datasource-parquet trips on my toolchain; not touched here). I didn't run the full workspace suite and am relying on CI for it.

I also updated the PR description, which still said a field is only read from the right when its metadata matches.

[kosiew]

@namanjain24-sudo,

Thanks for working through the earlier feedback. The physical and collected schema nullability checks look good now, and the added coverage for NULLABILITY_UNSPECIFIED is helpful. The metadata fallback issue also looks addressed.

There is still one metadata preservation issue in the new right-column projection path that I think needs to be fixed before merging. I left the details inline. In particular, the narrowed intersection result should preserve the left input's schema and field metadata exactly, while only changing the nullability required by the intersection rules.

Could you also extend the regression test to compare the complete metadata maps on the logical, physical, and collected schemas? That should catch right-only metadata leaking into the result.

[kosiew]

I think there is still a metadata preservation issue here. alias_qualified_with_metadata does not replace the right column's metadata. Expr::Alias::to_field starts with the right column metadata and then extends it with the supplied left metadata, so keys that exist only on the right can remain in the projected field.

The inner join has a similar effect on schema metadata: right-only keys can be merged into the join schema, and the projection preserves that merged metadata. This differs from LogicalPlanBuilder::intersect, where the left-semi join exposes only the left input's metadata.

Could we preserve the left schema and field metadata maps exactly in this narrowed result, while retaining the intentionally narrowed nullability? It would also be good to extend the regression test to compare the complete metadata maps, including checking that right-only keys are absent, for the logical plan, physical plan, and collected batches.

[namanjain24-sudo]

Good catch — you're right that alias_qualified_with_metadata merges rather than replaces, since Expr::Alias's field derivation extends the aliased expression's own metadata with the alias's supplied metadata instead of overriding it.

Fixed in 2a8d8f5: instead of aliasing the right column directly, I route it through a Cast to an explicit target field (Cast::new_from_field) carrying the left field's exact type and metadata. Both cast_output_field (logical) and cast_with_target_field/CastExpr::new_with_target_field (physical) use an explicit target field's metadata exactly, with nothing merged in from the source - that's the documented, intended way to get "target metadata exactly" rather than Alias's merge semantics. The qualifier/name still need alias_qualified on top since a bare Cast's own field isn't renamed to the target's name.

Extended the intersect_nullability regression test to compare the complete metadata map (not just the "column" key) for the tagged case, so a leaked only_in_secondary key now fails it. Since the test already asserts physical_plan.schema() == logical_schema and batch.schema() == logical_schema, that same check covers the physical plan and collected batches too.