fix: Preserve metadata when nl-join is swapped

[mkleen]

Which issue does this PR close?

• Relates to Join swap in join_selection flips schema-level metadata, failing "Schema mismatch" #23461 and resolves it for nl-joins. More joins types have the same issue.
• Same as fix: Preserve metadata when a cross-join is swapped #23605 but for nl-joins.

Rationale for this change

When a nl-join is swapped, the metadata is rebuilt based on the inverted inputs. This can cause a change of metadata when the metadata has conflicting values which results in an error e.g.:

// t1: 1000 rows, schema metadata {"key": "left"}
// t2: 2 rows, schema metadata {"key": "right"}
// Inner-join with non-equijoin predicate resulting in nested-loop join with swapped sides
ctx.sql("select * from t1 inner join t2 on (t1.x + t2.y) > 1").await?.collect().await?;
// -> Internal error: ... 'join_selection' failed. Schema mismatch.
// Metadata should be {"key" : "right" } but changed to {"key" : "left" } 
// because of the swap of sides and the rebuilt of the metadata starting from the other direction.

Therefore this pr avoids rebuilding the metadata based on the swapped inputs for nl-join and uses the original metadata instead

What changes are included in this PR?

• Extends NestedLoopJoinExecBuilder by a metadata field to create a new NestedLoopJoinExec with existing metadata
• Use the NestedLoopJoinExecBuilder in NestedLoopJoinExec::swap_inputs to swap the input and preserve the metadata
• Tests

Are these changes tested?

Yes

Are there any user-facing changes?

No

[mkleen]

This moved into test_swap_joins_on_conflicting_metadata

[codecov-commenter]

Codecov Report

❌ Patch coverage is 97.36842% with 1 line in your changes missing coverage. Please review.
✅ Project coverage is 80.71%. Comparing base (7269d13) to head (d4b7c77).
⚠️ Report is 1 commits behind head on main.

Files with missing lines	Patch %	Lines
...fusion/physical-plan/src/joins/nested_loop_join.rs	96.00%	0 Missing and 1 partial ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main   #23759      +/-   ##
==========================================
- Coverage   80.71%   80.71%   -0.01%     
==========================================
  Files        1090     1090              
  Lines      370370   370380      +10     
  Branches   370370   370380      +10     
==========================================
- Hits       298948   298944       -4     
- Misses      53601    53609       +8     
- Partials    17821    17827       +6     

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

[2010YOUY01]

This is a internal API from optimizer, it requires strict pre-conditions in order to use it correctly, so I think this should not be called directly.

Probably we could either

• mark raw swap_inputs() API as unsafe to use directly
• Add sanity checks inside to ensure it must be used correctly
• Deprecate it from the public API

I'm wondering is it possible to trigger the bug from only SQL script 🤔 ? If yes this seem like a major bug.

[alamb]

If swap_inputs is not safe we should definitely mark it thusly

There are some caveats documented here

• https://docs.rs/datafusion/latest/datafusion/physical_plan/joins/struct.HashJoinExec.html#method.swap_inputs

This function is public so other downstream projects can use it to construct HashJoinExec with right side as the build side.

I do think it is meant to be used publically and the usecase is reasonable in my mind (user control over join order)

If it is hard to use / easy to mess up, perhaps we can find a better API

[2010YOUY01]

I tend to think we should let this query error, but we can provide better error message.

Here are more details in the original issue, let's continue the discussion here: #23461 (comment)