sunchao commented on code in PR #26214:
URL: https://github.com/apache/datafusion/pull/26214#discussion_r4446958008
##########
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. The first `batch_size` probe rows are kept for every
+ /// chunk. The rest are produced again by calling `probe_rows`, only for
+ /// a chunk that still has unmarked rows after the kept ones, which then
+ /// 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) => {
+ // The probe rows can be spread over a large probe batch, so
+ // keep the first ones rather than walk the batch for every
+ // chunk. The first ones often mark the whole chunk.
+ let kept: Vec<u32> =
probe_rows().take(self.batch_size).collect();
+ if kept.is_empty() {
+ return Ok(());
+ }
+ let more_probe_rows = kept.len() == self.batch_size;
Review Comment:
A full `kept` buffer means there *might* be another eligible row. If there
are exactly `batch_size` and a build row stays unmarked,
`probe_rows().skip(kept.len())` rereads the bitmap for every chunk just to find
the end.
I reused the [previous SQL
fixture](https://github.com/apache/datafusion/pull/26214#discussion_r4443932027)
with 4,194,304 build rows and an 8,388,608-row probe batch. I set `b=8` on one
build row per 8,192-row chunk, kept `b=7` on the rest, and ended the probe with
exactly 8,192 `(NULL,7)` rows. All revisions return `COUNT(r)=512`. The
previous head walks the bitmap once and this head walks it 513 times, with the
same number of candidate pairs. Median query CPU in three local debug runs was
5.10 s on the merge base, 6.08 s on `a40e07c41`, and 7.74 s here.
Could the initial iterator read one more eligible row and save its physical
row index? If it returns `None`, there’s no suffix to replay. If it returns
`Some(start)`, the range or bitmap iterator can restart at `start` instead of
scanning and skipping the prefix. That also handles a real suffix when some
build rows remain unmarked, while keeping the buffer bounded. I tried that in a
local build. With one survivor per chunk, median CPU fell from 7.74 s to 6.01 s
for 8,192 eligible rows and from 6.95 s to 5.41 s for 8,193. Both results
stayed at 512.
--
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]
