jayzhan211 commented on PR #24956:
URL: https://github.com/apache/datafusion/pull/24956#issuecomment-5603695263
@SubhamSinghal I still have a suggestion for the latest commit,
**`probe_hit_rate` is per-row everywhere else — the existence path publishes
batches under that name.**
`existence_join.rs:301` does `add_total(1)` per streamed batch and `:360`
does `add_part(1)` per matching batch. But the metric is defined as *rows*:
- `joins/utils.rs:1843` — "Fraction of probe rows with at least one
build-side join-key match", under a `// Keep these metric descriptions in sync
with docs/source/user-guide/metrics.md` note
- `docs/source/user-guide/metrics.md:60` — same wording
- `classic_join.rs:245` (this PR) —
`probe_hit_rate.add_total(batch.num_rows())`
- `hash_join/stream.rs:765` —
`probe_hit_rate.add_total(state.batch.num_rows())`
So a `LeftSemi` PWMJ over two 8192-row batches in which 3 rows actually
match renders `probe_hit_rate: 100% (2/2)`. That doesn't just lose resolution,
it reports the opposite of the truth, and `probe_hit_rate` now means two
different things in two streams of the same operator. It also breaks the
invariant the other two paths hold: `probe_hit_rate.total == input_rows`.
The good news is the fix is cheap. The predicate you already evaluate for
the hit decision — `is_match(buffered_len - 1)`, i.e. "does this key beat the
most-matching buffered key" — is exactly the per-row predicate, and it
vectorizes, so the O(log n) watermark search is untouched:
```rust
// scan_stream_batch, before reducing to the extreme key
if batch.num_rows() > 0 {
self.record_probe_hits(&stream_values)?;
let extreme = extreme_key(&stream_values, self.sort_option.descending)?;
self.mark_matched_buffered_rows(&extreme)?;
}
```
```rust
/// `probe_hit_rate` is defined per probe *row*. A streamed row matches iff
it satisfies
/// the predicate against the most-matching buffered key, which is the last
non-null
/// buffered row (`is_match` is monotone over the sorted buffered side) --
so one
/// vectorized comparison per batch is enough.
fn record_probe_hits(&self, stream_values: &ArrayRef) -> Result<()> {
self.join_metrics.probe_hit_rate.add_total(stream_values.len());
let buffered_values =
self.buffered_side.try_as_ready()?.buffered_data.values();
let len = buffered_values.len();
if len == buffered_values.null_count() {
return Ok(()); // no non-null buffered key: nothing can match
}
let last = Scalar::new(buffered_values.slice(len - 1, 1));
// The join predicate is `buffered OP streamed`, so flip it to compare
// streamed against the key. Null streamed keys compare to null and are
// excluded by `true_count()`, which is what we want.
let hits = match self.operator {
Operator::Gt => lt(stream_values, &last)?,
Operator::GtEq => lt_eq(stream_values, &last)?,
Operator::Lt => gt(stream_values, &last)?,
Operator::LtEq => gt_eq(stream_values, &last)?,
op => return internal_err!("PiecewiseMergeJoin should not contain
operator, {op}"),
};
self.join_metrics.probe_hit_rate.add_part(hits.true_count());
Ok(())
}
```
Then `mark_matched_buffered_rows` drops both metric lines and goes back to
being purely about the watermark, and the existing order-independence tests
still hold (they'd assert `(2, 2)` on row counts rather than batch counts).
If you'd rather not add the comparison, the alternative I'd accept is to
leave `probe_hit_rate` unset for existence joins exactly as you've done for
`avg_fanout`, and say why in the same comment. `N/A` is honest; `100% (2/2)`
for a 3-row-in-16384 hit rate is not.
One related note for whichever way you go: `nothing_left_to_mark()` stops
polling the streamed side once the watermark saturates, so the denominator
covers only the batches actually read. That's fine and matches `input_rows`,
but it's worth a line in the comment so nobody later reads the ratio as
covering the whole probe side.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
