viirya commented on code in PR #26214:
URL: https://github.com/apache/datafusion/pull/26214#discussion_r4444316071
##########
datafusion/physical-plan/src/joins/hash_join/not_in.rs:
##########
@@ -1136,25 +1233,71 @@ where
build_lookups: 0,
#[cfg(test)]
mark_checks: 0,
+ #[cfg(test)]
+ max_decided_rows: 0,
f,
}
}
- /// Pairs each of `probe_rows` with each of `build_rows`. The pairing stops
- /// when every row that the pairs decide is marked.
- fn cross_product(
+ /// Pairs each of the rows that `probe_rows` yields with each of
+ /// `build_rows`. The pairing stops when every row that the pairs decide
+ /// is marked.
+ ///
+ /// When the pairs decide the build rows, the unmarked build rows are
+ /// paired in chunks of `batch_size` rows. Neither side is then held
+ /// whole: a large build side, as when this pairing stands in for a scope
+ /// map that the memory pool declined, or a probe batch larger than the
+ /// batch size. Up to `batch_size` probe rows are kept for every chunk.
+ /// More are produced again for each chunk by calling `probe_rows`, which
+ /// costs less than pairing them with the chunk. A chunk is filled with
+ /// unmarked rows only, so a build side with few unmarked rows left pairs
+ /// them all at once, as a single chunk would.
+ fn cross_product<P: Iterator<Item = u32>>(
&mut self,
build_rows: impl Iterator<Item = u64>,
- probe_rows: impl Iterator<Item = u32>,
+ probe_rows: impl Fn() -> P,
) -> Result<()> {
- // Drop the rows that are already marked before the first pair.
- self.needs_refresh = true;
match self.marks {
- UnknownMarks::Build(_) => {
- self.pair_unmarked(&mut build_rows.collect(), probe_rows, |b,
p| (b, p))
+ UnknownMarks::Build(bitmap) => {
+ // A few probe rows can be spread over a large probe batch, so
+ // keep them rather than walk the batch for every chunk.
+ let mut kept: Vec<u32> = probe_rows().take(self.batch_size +
1).collect();
+ if kept.is_empty() {
+ return Ok(());
+ }
+ let keep_probe_rows = kept.len() <= self.batch_size;
+ if !keep_probe_rows {
+ kept = vec![];
Review Comment:
Thanks, that is right: discarding `kept` above the bound brought back the
long walk to the first probe row once per chunk. Fixed in 853f41972 along the
lines you suggested:
- `cross_product` now always keeps the first `batch_size` probe rows. Each
build chunk is paired with them first, and the rest of the probe rows are
produced again (`probe_rows().skip(kept.len())`) only when the chunk still has
unmarked rows. Memory stays bounded by `batch_size`.
- `pair_unmarked` now refreshes and checks for an empty chunk before it
advances the other rows. Without that, the replayed suffix would still walk up
to its first row before noticing the chunk was already marked, which would
again be the long leading stretch on your input.
`cross_product_chunks_the_build_rows` gains a case with `batch_size + 1`
probe rows where every pair marks its build row, and asserts a single walk of
the probe rows. It now counts walks that produce a row rather than calls of
`probe_rows`. Reverting either of the two changes gives 158 walks there instead
of 1.
--
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]
