sunchao commented on code in PR #26214:
URL: https://github.com/apache/datafusion/pull/26214#discussion_r4443932027
##########
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:
The one-NULL case from my [earlier
comment](https://github.com/apache/datafusion/pull/26214#discussion_r4436475953)
is fixed. I missed an early-exit case when I suggested discarding the cache
above `batch_size`. `pair_unmarked` can mark a whole build chunk using its
**first** eligible probe row. Discarding `kept` makes the next chunk walk the
long leading stretch of the bitmap to reach that same row again.
I reran `(a,b) NOT IN (SELECT x,y FROM i)` with 4,194,304 build rows `(a,7)`
(`a >= 10`) and one 8,388,608-row streaming probe batch. The probe starts with
`(3,7)` rows and ends with `(NULL,7)` rows. At the default batch size of 8,192,
changing the number of trailing NULLs from 8,192 to **8,193** takes the current
code from one bitmap walk to 513. The merge base walks it once. Both evaluate
4,194,304 candidate pairs and return the same result. With 8,193 NULLs the
median query CPU time over three local debug runs was 4.31 s on the merge base
and 5.92 s here (+37%).
Could we retain the first `batch_size` indices even on overflow, iterate
that prefix first, and lazily replay the suffix only if the chunk still has
unmarked rows? That keeps memory bounded. In the pairing reproducer it also
avoids the repeated walks on this input.
--
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]
