viirya commented on code in PR #26214:
URL: https://github.com/apache/datafusion/pull/26214#discussion_r4447256342
##########
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:
Right, a full buffer only says there may be more. Fixed in 05b4c9fb3 as you
suggested:
- The first walk now takes `batch_size + 1` probe rows. If the extra one
exists, it is popped off `kept` and its row index is where the rest starts; if
not, there is no rest and nothing is replayed.
- `probe_rows` now takes a start row. The range closure starts its range
there, and the NULL-mask closure starts a `BitIndexIterator` at that bit offset
of the same buffer, so a replay no longer scans and skips the kept prefix.
- The replay also runs only when the chunk still has unmarked rows after the
kept ones. The refresh in `pair_unmarked` is factored out so `cross_product`
can check that before it creates the iterator.
`cross_product_chunks_the_build_rows` now counts both the calls of
`probe_rows` and the probe rows they read. It covers exactly `batch_size` probe
rows with an unmarked chunk (one walk; dropping the extra read gives 158), and
`batch_size + 1` probe rows (`1 + 157` walks reading `65 + 157` rows; scanning
from the start of the batch reads 10,270). The slt sources slice probe batches
to the batch size, so they never reach the replay;
`test_null_aware_left_anti_multi_column_null_probe_rows_past_batch` in
`exec.rs` uses a 6-row probe batch with four NULL-valued rows at batch sizes 1
and 2, where the rows past the first batch decide two build rows. It fails if
the NULL-mask closure drops the start offset.
--
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]
