sunchao commented on code in PR #26214:
URL: https://github.com/apache/datafusion/pull/26214#discussion_r4426153781
##########
datafusion/physical-plan/src/joins/hash_join/not_in.rs:
##########
@@ -1906,6 +2009,79 @@ mod tests {
Ok(())
}
+ /// A hot scope among rare ones: a batch of rare probe rows, then hot
+ /// ones, then rare ones again, as the review of
+ /// <https://github.com/apache/datafusion/pull/26214> found. The rare rows
+ /// keep the estimated groups small, but the hot group alone saves much
+ /// work, so the pairing must still switch to groups instead of walking the
+ /// hot scope once per probe row.
+ #[test]
+ fn hot_scope_among_rare_ones_is_grouped() -> Result<()> {
+ // With marks on the probe rows, the average group fills no chunk. With
+ // marks on the build rows, the groups hold fewer than two rows on
+ // average.
+ for (mark_build_rows, rare, hot) in [(false, 256, 7680), (true, 4096,
2048)] {
+ let stats = hot_scope_among_rare_ones(mark_build_rows, rare, hot,
7679)?;
+ assert!(stats.grouped, "marks on build rows: {mark_build_rows}");
+ // The matches of two rounds of lookups per probe row, each until
+ // it wastes enough matches to pay for a sort, and a few chunks of
+ // the hot scope. The lookups that skip the check of the build row
+ // marks count no wasted match, so a round can last a little
+ // longer. A lookup per probe row walks the hot scope for each of
+ // its probe rows, tens of millions of matches.
+ let bound = 3 * WASTED_MATCHES_PER_SORTED_ROW * (2 * rare + hot) +
4 * 8192;
+ assert!(
+ stats.looked_up_matches <= bound,
+ "marks on build rows: {mark_build_rows}: {} matches walked,
bound {bound}",
+ stats.looked_up_matches
+ );
+ }
+ Ok(())
+ }
+
+ /// `rare` probe rows of their own scope each, `hot` probe rows of the hot
+ /// scope, and `rare` more of their own scope, in that order. Each rare
+ /// scope holds one build row, and the hot scope `hot_build`. Every pair
+ /// passes.
+ fn hot_scope_among_rare_ones(
+ mark_build_rows: bool,
+ rare: usize,
+ hot: usize,
+ hot_build: usize,
+ ) -> Result<Stats> {
+ let rare = rare as u64;
+ // Scope 0 is hot.
+ let build_keys: Vec<u64> = (1..=2 * rare)
+ .chain(std::iter::repeat_n(0, hot_build))
+ .collect();
+ let probe_keys: Vec<u64> = (1..=rare)
+ .chain(std::iter::repeat_n(0, hot))
+ .chain(rare + 1..=2 * rare)
+ .collect();
+ let num_marked = if mark_build_rows {
+ build_keys.len()
+ } else {
+ probe_keys.len()
+ };
+ let case = Case {
+ mark_build_rows,
+ through_scope_map: true,
+ batch_size: 8192,
+ hash_modulus: u64::MAX,
+ probe_rows: (0..probe_keys.len() as u32).collect(),
+ build_rows: (0..build_keys.len() as u64).collect(),
+ build_keys,
+ probe_keys,
+ marked_before: vec![false; num_marked],
+ filter_seed: 0,
+ filter_one_in: 1,
+ foreign_marks: false,
+ };
+ let (marks, stats) = case.run()?;
+ assert_eq!(marks, case.expected_marks());
Review Comment:
**Optional test cleanup.** Every rare scope and the hot scope occurs on both
sides, and `filter_one_in = 1`, so every expected mark is true.
`case.expected_marks()` walks 229,619,712 build/probe pairs across the two
fixtures to derive that bitmap. Could this use
`assert!(marks.iter().all(|marked| *marked))`, as the adjacent one-scope helper
does? The separate bound on `looked_up_matches` would still catch the
performance regression.
##########
datafusion/physical-plan/src/joins/hash_join/not_in.rs:
##########
@@ -879,37 +940,58 @@ fn mark_null_candidates_for_probe_batch(
)?;
pairs.finish()?;
}
- None => {
+ NullValueScope::Unscoped => {
let mut pairs = UnmarkedPairs::new(marks, batch_size, &mark);
pairs.cross_product(
null_rows.build_indices.values().iter().copied(),
probe_rows,
)?;
pairs.finish()?;
}
+ NullValueScope::Unindexed => {
+ let mut pairs =
+ UnmarkedPairs::new(marks, batch_size, &mut
mark_scope_matches);
+ pairs.cross_product(
+ null_rows.build_indices.values().iter().copied(),
+ probe_rows,
+ )?;
+ pairs.finish()?;
+ }
}
}
// Case 2: NULL-valued probe rows paired with the build rows in their
// correlation scope.
if let Some(null_mask) = probe_null_value_mask {
- let null_probe_rows = null_mask
+ let mut null_probe_rows = null_mask
.values()
.set_indices()
.map(|i| i as u32)
- .filter(in_scope);
+ .filter(in_scope)
+ .peekable();
match scope_map {
+ // No NULL-valued probe row is in any scope: nothing to look up,
+ // so the scope map need not be built.
+ Some(_) if null_probe_rows.peek().is_none() => {}
Some(scope_map) => {
let mut pairs =
UnmarkedPairs::new(marks, batch_size, &mut
mark_scope_matches);
- pairs.scope_matches(
- scope_map,
- |build_row| build_row,
- null_probe_rows,
- hashes_buffer,
- lookup,
- )?;
+ match scope_map.get_or_try_init(build_scope_values,
random_state)? {
+ Some(scope_map) => pairs.scope_matches(
+ scope_map,
+ |build_row| build_row,
+ null_probe_rows,
+ hashes_buffer,
+ lookup,
+ )?,
+ // Without the map, every build row is paired, and the
+ // scope keys of each pair are compared.
+ None => pairs.cross_product(
Review Comment:
**The low-memory fallback still allocates an index for every build row.**
For `LeftAnti` and `LeftMark`, `cross_product` collects this range into a
`Vec<u64>` before it checks which build rows are already marked. This
allocation bypasses the reservation that just declined the scope map.
I traced the passing `null_aware_scope_map_falls_back_when_memory_is_short`
test on this head. With 10,000 build rows, this call allocated 80,000 bytes
that were absent from `build_mem_used`. In an isolated pairing run with
8,388,608 already-marked rows and one NULL probe, the vector allocated 64 MiB
and emitted zero candidate pairs. Each concurrent probe partition can allocate
its own vector.
Could the fallback visit bounded chunks of build indices when the marks are
on the build side? That would preserve the scope comparison while bounding the
temporary allocation under pressure.
--
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]
