Substrait join consumer should not merge nullability of join keys

[hareshkh]

Which issue does this PR close?

• Closes Substrait join consumer produces incorrect logical plan #21124

Rationale for this change

When a Substrait join expression contains both equal and is_not_distinct_from predicates (e.g. Spark pushes a null-safe filter into a join that already has a regular equality key), the split_eq_and_noneq_join_predicate_with_nulls_equality function uses a single nulls_equal_nulls boolean that gets overwritten per-predicate. Whichever operator is processed last determines the NullEquality for all keys, silently dropping NULL-matching rows.

Since NullEquality is a join-level setting (not per-key) across all physical join implementations (HashJoinExec, SortMergeJoinExec, SymmetricHashJoinExec), the correct fix is to match DataFusion's own SQL planner behavior: demote IS NOT DISTINCT FROM keys to the join filter when mixed with Eq keys. This is already correctly handled for SQL as shown in join_is_not_distinct_from.slt:L188

# Test mixed equal and IS NOT DISTINCT FROM conditions
# The `IS NOT DISTINCT FROM` expr should NOT in HashJoin's `on` predicate
query TT
EXPLAIN SELECT t1.id AS t1_id, t2.id AS t2_id, t1.val, t2.val
FROM t1
JOIN t2 ON t1.id = t2.id AND t1.val IS NOT DISTINCT FROM t2.val
----
logical_plan
01)Projection: t1.id AS t1_id, t2.id AS t2_id, t1.val, t2.val
02)--Inner Join: t1.id = t2.id Filter: t1.val IS NOT DISTINCT FROM t2.val
03)----TableScan: t1 projection=[id, val]
04)----TableScan: t2 projection=[id, val]

What changes are included in this PR?

datafusion/substrait/src/logical_plan/consumer/rel/join_rel.rs:

• Collect eq_keys and indistinct_keys separately instead of using a single vec with an overwritable boolean
• When both are present (mixed case), use eq_keys as equijoin keys with NullEqualsNothing and reconstruct the IsNotDistinctFrom expressions into the join filter
• Return NullEquality directly instead of converting from bool

Are these changes tested?

Yes, three levels of coverage:

1. Unit tests (join_rel.rs) — directly assert the output of split_eq_and_noneq_join_predicate_with_nulls_equality for eq-only, indistinct-only, mixed, and non-column-operand cases
2. Integration test (consumer_integration.rs) — loads a JSON-encoded Substrait plan with a JoinRel containing both operators through from_substrait_plan, executes it, and asserts 6 rows (including NULL=NULL matches)
3. Existing SLT (join_is_not_distinct_from.slt:179-205) — confirms the SQL planner already exhibits the same demotion behavior that this PR adds to the Substrait consumer

Are there any user-facing changes?

No API changes. Substrait plans with mixed equal/is_not_distinct_from join predicates now correctly preserve null-safe semantics instead of silently dropping NULL-matching rows.

[hareshkh]

Hey @gabotechs, @alamb, @LiaCastaneda! would you like to take a look at this one please?

[Satyr09]

We unconditionally favour Eq keys here, but if we have a case where there exists multiple (say 4) IS NOT DISTINCT FROM column pairs and 1 Eq column pair, this demotes all 4 to filter and keeps just the 1 eq key, right?

But, in this case, would the inverse (demote the single eq to filter) not allow more columns to participate in the hash partitioning/pruning and therefore be a bit more performant?
More selective hash key = frwer candidate pairs survive and need fewer row-by-row filter evaluation, if I understand correctly?

[hareshkh]

The optimiser currently already has this behaviour of favouring Eq predicates over Indistinct predicates. Added a SLT to confirm that behaviour - https://github.com/apache/datafusion/pull/21121/changes#diff-63fc43cf735eb03abd4d114cfbbf24982939425938a74b354fb7db6da7d499d7R305, and replicating that behaviour in this change.

I also think that selectivity is a function of data i.e. having a hash join on 3 indistinct keys could produce more data than 1 eq key.

[hareshkh]

Thanks for the review @Satyr09 - I have addressed all your comments.

[gabotechs]

Looks good, just have a small suggestion, thanks @hareshkh for the PR, and @Satyr09 for the review!

[hareshkh]

Thanks for the review @gabotechs - I have addressed the comment as well.

[gabotechs]

Nice, I'll leave this open until tomorrow in case someone wants to chime in, otherwise good work!

[alamb]

Thank you @hareshkh and @gabotechs