fix: evaluate sort-merge join filters that reference no columns

[mrhard9090]

Which issue does this PR close?

• Closes Sort-merge join ignores a join filter that references no columns (wrong FULL JOIN results) #26002.

Rationale for this change

SortMergeJoinExec only evaluated its join filter when the filter read at least one column. A filter that reads none (a bound parameter such as $1 > 5, or a volatile function such as random() > 2) was skipped, so every key match was kept. For FULL JOIN that returns matched pairs the filter rejects, where hash join null-pads both sides.

What changes are included in this PR?

• freeze_streamed_matched (sort_merge_join/materializing_stream.rs) now evaluates the filter whenever the join has one. The filter batch is built with RecordBatchOptions::with_row_count(total_matched_rows), because a batch with no columns cannot infer its row count.
• Two queries in sort_merge_join.slt: FULL JOIN ... AND random() > 2 (every row null-padded) and AND random() < 2 (all key matches kept).

I did not touch bitwise_stream.rs (semi/anti/mark joins), which builds its filter batch in a different place.

What is the testing strategy for this PR?

cargo test -p datafusion-sqllogictest --test sqllogictests -- sort_merge_join: all 4 files pass. With the materializing_stream.rs change reverted, sort_merge_join.slt fails. cargo test -p datafusion-physical-plan --lib sort_merge_join: 247 passed. cargo fmt and cargo clippy -p datafusion-physical-plan --all-targets -- -D warnings pass.

Are there any user-facing changes?

FULL JOIN (and other join types run by sort-merge join) with a column-free ON condition now returns the same rows as hash join. No API change.

AI assistance: written with Claude Code; I reproduced the bug, reviewed the diff and ran the tests above.

[codecov-commenter]

Codecov Report

❌ Patch coverage is 89.36170% with 5 lines in your changes missing coverage. Please review.
✅ Project coverage is 82.64%. Comparing base (95a1961) to head (7c18075).

Files with missing lines	Patch %	Lines
.../src/joins/sort_merge_join/materializing_stream.rs	89.36%	1 Missing and 4 partials ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main   #26010      +/-   ##
==========================================
- Coverage   82.64%   82.64%   -0.01%     
==========================================
  Files        1147     1147              
  Lines      445502   445503       +1     
  Branches   445502   445503       +1     
==========================================
- Hits       368179   368169      -10     
- Misses      54969    54975       +6     
- Partials    22354    22359       +5     

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

[jayzhan211]

@mrhard9090 , overall LGTM, please fix the conflicts

Same bug in the semi/anti/mark stream: with a column-free filter, try_new fails with must either specify a row count or at least one column for all six types. SQL can't reach it because the planner pushes the predicate into an input, but a SortMergeJoinExec built directly can. Fine to handle in a follow-up; it's a one-line fix:

-    let filter_batch = RecordBatch::try_new(Arc::clone(filter.schema()), columns)?;
+    let filter_batch = RecordBatch::try_new_with_options(
+        Arc::clone(filter.schema()),
+        columns,
+        &RecordBatchOptions::new().with_row_count(Some(num_outer_rows)),
+    )?;

Test (fails on this branch, passes with the fix):

#[tokio::test]
async fn join_semi_anti_mark_column_free_filter() -> Result<()> {
    for join_type in [
        LeftSemi, LeftAnti, LeftMark, RightSemi, RightAnti, RightMark,
    ] {
        let left = build_table(
            ("a1", &vec![1, 2, 3]),
            ("b1", &vec![4, 5, 7]),
            ("c1", &vec![7, 8, 9]),
        );
        let right = build_table(
            ("a2", &vec![10, 20, 30]),
            ("b1", &vec![4, 5, 6]),
            ("c2", &vec![70, 80, 90]),
        );
        let on = vec![(
            Arc::new(Column::new_with_schema("b1", &left.schema())?) as _,
            Arc::new(Column::new_with_schema("b1", &right.schema())?) as _,
        )];
        let filter = JoinFilter::new(
            Arc::new(Literal::new(ScalarValue::Boolean(Some(false)))),
            vec![],
            Arc::new(Schema::empty()),
        );
        let (_, batches) =
            join_collect_with_filter(left, right, on, filter, join_type).await?;
        let rows: usize = batches.iter().map(|b| b.num_rows()).sum();
        let expected = match join_type {
            LeftSemi | RightSemi => 0,
            _ => 3,
        };
        assert_eq!(rows, expected, "{join_type:?}");
    }
    Ok(())
}