jayzhan211 commented on PR #24956:
URL: https://github.com/apache/datafusion/pull/24956#issuecomment-5557844772
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:
```rust
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:
```rust
/// 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.
--
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]
