Pwmj metrics accounting

[SubhamSinghal]

Which issue does this PR close?

Part of #17427.

Rationale for this change

PiecewiseMergeJoinExec's metrics accounting had several gaps left over from earlier PRs in the epic:

• The classic-join stream (Left/Right/Full/Inner) never routed poll_next through BaselineMetrics::record_poll, so output_rows, output_bytes, and output_batches stayed at 0 in EXPLAIN ANALYZE regardless of how many rows the join actually produced.
• join_time was never measured on either the classic or existence-join stream — only build_time was timed, so elapsed_compute understated total operator cost for large probe sides.
• probe_hit_rate/avg_fanout exist on the shared BuildProbeJoinMetrics struct (populated by HashJoinExec) but PWMJ never populated them, always showing N/A (0/0).

What changes are included in this PR?

• ClassicPWMJStream::poll_next now calls self.join_metrics.baseline.record_poll(poll), matching the existence-join stream.
• join_time is timed around the actual comparison work: resolve_classic_join and the ProcessUnmatched bitmap/take pass on the classic path; extreme_key + mark_matched_buffered_rows on the existence path.
• probe_hit_rate/avg_fanout are populated for classic join, using the range size (buffered_len - buffer_idx) already computed at each match as the fanout.
• probe_hit_rate is populated for existence join: a streamed batch counts as a hit if its extreme key lowers the shared watermark, a miss otherwise. avg_fanout is intentionally left unset for existence join — its watermark-based semantics don't have a natural per-row fanout equivalent.
• Added a metrics-focused test to each stream module: classic_join::tests::inner_join_records_output_and_probe_metrics and existence_join::tests::probe_hit_rate_counts_batches_that_advance_the_watermark, both hand-derived and verified against actual runs.

Are these changes tested?

Yes — two new unit tests

Are there any user-facing changes?

EXPLAIN ANALYZE on a PiecewiseMergeJoinExec plan now reports accurate output_rows/output_bytes/output_batches/join_time/probe_hit_rate/avg_fanout instead of zeros/N/A. No API or behavior change to query results.

[codecov-commenter]

Codecov Report

❌ Patch coverage is 87.92271% with 25 lines in your changes missing coverage. Please review.
✅ Project coverage is 81.63%. Comparing base (35f58f5) to head (a0ba9a7).
⚠️ Report is 11 commits behind head on main.

Files with missing lines	Patch %	Lines
...n/src/joins/piecewise_merge_join/existence_join.rs	87.71%	4 Missing and 10 partials ⚠️
...lan/src/joins/piecewise_merge_join/classic_join.rs	88.17%	0 Missing and 11 partials ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main   #24956      +/-   ##
==========================================
+ Coverage   81.61%   81.63%   +0.01%     
==========================================
  Files        1124     1124              
  Lines      411978   412700     +722     
  Branches   411978   412700     +722     
==========================================
+ Hits       336236   336887     +651     
- Misses      55936    55976      +40     
- Partials    19806    19837      +31     

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

Thanks @SubhamSinghal , I found what look like duplicated timers in a few places. If there's a reason to keep them, a short comment explaining why would help.

[jayzhan211]

Do we need another timer here?

[SubhamSinghal]

Addressed in a0ba9a7

[jayzhan211]

Do we need either join_time or join_timer here?

[SubhamSinghal]

Addressed in a0ba9a7

[xudong963]

it starts timers inside existing timers

[SubhamSinghal]

Addressed in a0ba9a7

[xudong963]

ditto

[SubhamSinghal]

Addressed in a0ba9a7

[xudong963]

ditto

[SubhamSinghal]

Addressed in a0ba9a7

[jayzhan211]

I found a possible issue

probe_hit_rate for existence join is defined as "lowered the watermark", which isn't a hit rate. The new test shows it: buffered 1..5, Gt, streamed key 4 matches buffered 5, but it's asserted as a miss because key 3 arrived first. I ran the same test with the two batches swapped and got 2/2 instead of 1/2. With multiple streamed partitions, whichever partition lowers min_marked first makes every later partition's matches look like misses, and once the watermark reaches first_non_null_buffered the whole block is skipped so every remaining batch is a miss. docs/source/user-guide/metrics.md defines this as "fraction of probe-side rows with a build-side join-key match", so an EXPLAIN ANALYZE reader will take 50% to mean half the probe side didn't match.

Since is_match is monotone over the sorted buffered side, "does this batch match anything" is one comparison against the last buffered row, independent of the watermark. Decide the hit there, then keep the bounded binary search only for the watermark update:

if row_idx < stream_values.len() && first_non_null_buffered < buffered_len {
    let cmp = JoinKeyComparator::new(/* unchanged */)?;
    let is_match = |buffer_idx: usize| { /* unchanged */ };

    // `is_match` is monotone over the sorted buffered side, so the extreme key
    // matches *something* iff it matches the last buffered key. That decides
    // `probe_hit_rate` independently of the watermark, which other batches or
    // partitions may already have lowered past this batch's match range. A
    // batch that matches nothing can't lower the watermark either.
    if !is_match(buffered_len - 1) {
        return Ok(());
    }
    self.join_metrics.probe_hit_rate.add_part(1);

    if first_non_null_buffered >= scan_limit {
        // Everything this batch could mark is already marked.
        return Ok(());
    }

    // existing binary search over [first_non_null_buffered, scan_limit) ...
    let buffer_idx = lo;
    if buffer_idx < scan_limit {
        buffered_data.min_marked.fetch_min(buffer_idx, AtomicOrdering::SeqCst);
    }
}

The existing test then expects (2, 2) since key 4 really does match buffered 5. To guard the order dependence, a test that runs the same batches in both orders, with one batch that genuinely matches nothing (key 5 under >) and one pair of nested hits:

/// Runs a `LeftSemi` `buffered.b1 > streamed.b1` join over buffered keys 1..=5 with
/// the given streamed batches in one partition and returns `probe_hit_rate` as
/// `(part, total)`.
async fn existence_probe_hit_rate(
    streamed_batches: Vec<RecordBatch>,
) -> Result<(usize, usize)> {
    let left = build_table(
        ("a1", &vec![1, 2, 3, 4, 5]),
        ("b1", &vec![1, 2, 3, 4, 5]),
        ("c1", &vec![10, 20, 30, 40, 50]),
    );
    let streamed_schema = Schema::new(vec![
        Field::new("a2", DataType::Int32, false),
        Field::new("b1", DataType::Int32, false),
        Field::new("c2", DataType::Int32, false),
    ]);
    let right = TestMemoryExec::try_new_exec(
        &[streamed_batches],
        Arc::new(streamed_schema),
        None,
    )?;
    let on = (
        Arc::new(Column::new_with_schema("b1", &left.schema())?) as _,
        Arc::new(Column::new_with_schema("b1", &right.schema())?) as _,
    );
    let join = PiecewiseMergeJoinExec::try_new(
        left, right, on, Operator::Gt, JoinType::LeftSemi, 1,
    )?;
    let stream = join.execute(0, Arc::new(TaskContext::default()))?;
    common::collect(stream).await?;

    let metrics = join.metrics().unwrap();
    Ok(metrics
        .iter()
        .find_map(|m| match m.value() {
            crate::metrics::MetricValue::Ratio { name, ratio_metrics }
                if name == "probe_hit_rate" =>
            {
                Some((ratio_metrics.part(), ratio_metrics.total()))
            }
            _ => None,
        })
        .expect("probe_hit_rate metric"))
}

/// `probe_hit_rate` must describe the data, not the order it arrived in. Key 3 matches
/// buffered 4 and 5; key 5 matches nothing under `>`. Whichever batch lowers the
/// watermark first, the answer is one hit out of two.
#[tokio::test]
async fn existence_probe_hit_rate_is_independent_of_batch_order() -> Result<()> {
    let hit = || build_table_i32(("a2", &vec![10]), ("b1", &vec![3]), ("c2", &vec![70]));
    let miss = || build_table_i32(("a2", &vec![20]), ("b1", &vec![5]), ("c2", &vec![80]));

    assert_eq!(existence_probe_hit_rate(vec![hit(), miss()]).await?, (1, 2));
    assert_eq!(existence_probe_hit_rate(vec![miss(), hit()]).await?, (1, 2));

    // Two hits whose match ranges nest: the second cannot lower the watermark but
    // still matched, so it must not be reported as a miss in either order.
    let inner = || build_table_i32(("a2", &vec![30]), ("b1", &vec![4]), ("c2", &vec![90]));
    assert_eq!(existence_probe_hit_rate(vec![hit(), inner()]).await?, (2, 2));
    assert_eq!(existence_probe_hit_rate(vec![inner(), hit()]).await?, (2, 2));

    Ok(())
}

On the current branch this test fails at the nested-hits assertion with (1, 2); with the change above it passes along with the rest of the module and clippy. If you'd rather not add the compare, leaving the metric unset (as you did for avg_fanout) is better than a number that changes with batch order. Either way the metrics doc table should say PWMJ existence joins count streamed batches rather than rows.