viirya commented on code in PR #26214:
URL: https://github.com/apache/datafusion/pull/26214#discussion_r4438724928
##########
datafusion/physical-plan/src/joins/hash_join/not_in.rs:
##########
@@ -1136,25 +1233,56 @@ 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, and `probe_rows` is called
+ /// again for each chunk. 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. 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) => {
+ if probe_rows().next().is_none() {
+ return Ok(());
+ }
+ let mut build_rows = build_rows.peekable();
+ let mut chunk = Vec::with_capacity(self.batch_size);
+ while build_rows.peek().is_some() {
+ chunk.clear();
+ {
+ let bitmap = bitmap.lock();
+ chunk.extend(
+ build_rows
+ .by_ref()
+ .filter(|row| !bitmap.get_bit(*row as usize))
+ .take(self.batch_size),
+ );
+ }
+ // The chunk holds no row marked so far.
+ self.needs_refresh = false;
+ self.pair_unmarked(&mut chunk, probe_rows(), |b, p| (b,
p))?;
Review Comment:
Confirmed, and fixed in a40e07c41 as you suggest. `cross_product` first
takes up to `batch_size + 1` probe rows. If there are at most `batch_size`, it
keeps them and pairs every chunk with them, so your case walks the bitmap once,
as the merge base does. With more, it drops them and produces the probe rows
again for each chunk, which keeps the memory bound. That walk then costs less
than pairing those rows with the chunk, which is at least `batch_size *
batch_size` pairs.
The fix sits in `cross_product`, so it also covers the NULL-valued build
rows paired with the in-scope probe rows, where a few in-scope rows in a large
batch had the same problem. `cross_product_chunks_the_build_rows` now counts
the calls of `probe_rows`: one for three probe rows over 157 chunks (158
without the fix), and one per chunk plus one for `batch_size + 1` probe rows.
##########
datafusion/physical-plan/src/joins/hash_join/not_in.rs:
##########
@@ -1245,33 +1379,46 @@ where
self.batch_size,
)
};
- let rest_hashes = &hashes[group_from..];
// Estimate the groups from a sample first, because a sort of all the
// rest that finds no group worth it costs as much as the wasted
// matches did.
- let worth_sorting = saves_work(rest_hashes.len(),
estimate_groups(rest_hashes));
- let mut rest: Vec<(u64, u32)> = vec![];
- let mut groups = vec![];
- if worth_sorting {
- rest = rest_hashes
- .iter()
- .copied()
- .zip(probe_rows[group_from..].iter().copied())
- .collect();
- rest.sort_unstable_by_key(|(hash, _)| *hash);
- groups = rest.chunk_by(|a, b| a.0 == b.0).collect();
+ let mut rest = vec![];
+ if saves_work(
+ probe_rows.len() - group_from,
+ estimate_groups(&hashes[group_from..]),
+ ) {
+ rest = sort_by_hash(&hashes[group_from..],
&probe_rows[group_from..]);
}
- if !worth_sorting || !saves_work(rest.len(), groups.len()) {
- self.lookup_each_row(
+ let mut groups = hash_groups(&rest);
+ if rest.is_empty() || !saves_work(rest.len(), groups.len()) {
+ self.wasted_matches = 0;
+ let Some((offset, _)) = self.lookup_each_row(
scope_map,
hashes,
(!all_rows).then_some(probe_rows.as_slice()),
offset,
&to_build_row,
- usize::MAX,
+ max_wasted_matches,
lookup,
- )?;
- return Ok(());
+ )?
+ else {
+ return Ok(());
+ };
+ let group_from = match offset {
+ AtRow { row } | MidChain { row, .. } => row,
+ };
+ if rest.is_empty() {
+ rest = sort_by_hash(&hashes[group_from..],
&probe_rows[group_from..]);
+ } else {
+ // `probe_rows` is increasing, so the rows from `group_from` on
+ // are those not below its row, and keeping them keeps the
+ // order by hash.
+ match probe_rows.get(group_from) {
+ Some(&first) => rest.retain(|(_, row)| *row >= first),
+ None => rest.clear(),
+ }
+ }
+ groups = hash_groups(&rest);
Review Comment:
Right, done in a40e07c41: the retry starts with `drop(groups)`, so the
rejected groups are freed before the lookups resume rather than when the new
groups replace them.
##########
docs/source/library-user-guide/upgrading/56.0.0.md:
##########
@@ -934,9 +934,15 @@ Wire compatibility is directional:
### Null-aware joins record their `NOT IN` value key count
Multi-column `(a, b) NOT IN (SELECT x, y ...)` is now planned as a null-aware
-join. Its first equi-join keys are the tuple elements, followed by any
-correlation keys, so a null-aware join now records how many of its leading keys
-are `NOT IN` value keys:
+join. Its first equi-join keys are the tuple elements that can be NULL. A
+nullable element cannot be NULL here when a correlated filter of the subquery
+rejects NULL for it, such as `y > o.c` for `y`. The elements that cannot be
+NULL and any equality correlations follow as further equi-join keys when their
+types can be hashed, and otherwise stay in the join filter. `Null`,
+dictionary, run-end encoded and union elements stay among the first keys,
Review Comment:
Reworded in a40e07c41. The guide now says that these types, and elements
that read a column of them, count as elements that can be NULL. Like any such
element, they must be hashable equi-join keys, so the planner rejects a run-end
encoded one.
--
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]
