HashJoinExec probe_hit_rate metric overcounts probe rows when a probe batch is processed in several chunks #25077

Describe the bug

The probe_hit_rate metric of HashJoinExec is documented as "fraction of probe-side rows with a build-side join-key match before applying any join filter", but its denominator is inflated whenever a probe batch is processed in more than one chunk.

HashJoinStream::process_probe_batch adds state.batch.num_rows() to the metric's total at the top of the function:

https://github.com/apache/datafusion/blob/a5c809f98/datafusion/physical-plan/src/joins/hash_join/stream.rs#L763-L765

When the hash-map lookup hits batch_size output rows before the probe batch is exhausted, process_probe_batch returns with next_offset set and is re-entered for the same probe batch. Each re-entry adds the full batch row count again, so a probe batch that produces k chunks is counted k times in the denominator.

The numerator has a smaller version of the same problem: add_part(distinct_right_indices_count) counts the distinct probe indices of each chunk, so a probe row whose match chain is split across two chunks is counted twice.

To Reproduce

Run any hash join whose fanout makes the output exceed batch_size per probe batch, e.g. a build side with many duplicate keys, and inspect the metric with EXPLAIN ANALYZE. A 4-row probe batch with 8 matches each and batch_size = 8 produces 4 chunks, so the metric reports a total of 16 probe rows instead of 4 and a hit rate of 0.25 instead of 1.0.

Expected behavior

Both counters should reflect each probe row once: the total should be added once per probe batch (for example only when state.offset == (0, None)), and a probe row spanning a chunk boundary should not be counted twice in the part.

Additional context

Found while reviewing process_probe_batch in datafusion/physical-plan/src/joins/hash_join/stream.rs. avg_fanout uses distinct_right_indices_count as its total and has the same chunk-boundary double count, though its part (left_indices.len()) is correct.