viirya commented on code in PR #26214:
URL: https://github.com/apache/datafusion/pull/26214#discussion_r4426646751
##########
datafusion/optimizer/src/decorrelate_predicate_subquery.rs:
##########
@@ -1037,6 +1042,38 @@ fn build_join(
}
let left = projected_left.as_ref().unwrap_or(left);
+ // The first `V` keys of a null-aware join are its `NOT IN` value keys.
+ //
+ // A tuple element that cannot be NULL in scope on either side never makes
+ // the tuple comparison UNKNOWN: compared with a definite value it is
either
+ // equal, which leaves the other elements to decide, or a definite
mismatch,
+ // which makes the comparison FALSE whatever the other elements are. It
+ // therefore only selects which subquery rows a NULL in another element is
+ // compared with, which is what a correlation scope key does. Moving such
+ // elements after the ones that can be NULL turns them into scope keys, so
+ // the hash join finds the rows a NULL-valued row is compared with through
a
+ // hash lookup instead of pairing it with every row of the other side
+ // (<https://github.com/apache/datafusion/issues/26088>).
+ //
+ // A null-aware tuple has at least one element that can be NULL, so `V` is
+ // at least one. With exactly one, the join is a scalar `NOT IN` on that
+ // element, correlated by the other elements.
+ let null_aware_value_keys = if null_aware && in_values.len() > 1 {
+ let (nullable, non_nullable): (Vec<_>, Vec<_>) = in_values
+ .into_iter()
+ .zip(elements_may_be_null)
+ .partition(|(_, may_be_null)| *may_be_null);
Review Comment:
Agreed, the tuple query itself regresses here, whatever happens to the
inference. Fixed in 731717da5: an element of a dictionary, run-end encoded or
union type stays a value key, since an array of these types can hold a NULL
without a null bit. Only the key partition changes. Whether the join is
null-aware is decided as before, so the scalar form plans as it did.
`multi_column_not_in_sees_null_dictionary_values` builds your batches with
`RecordBatch::try_from_iter` and checks both the projected NULL for `a = 1` and
the `WHERE` form. It fails without the fix.
A run-end encoded element is not hashable, so it is now rejected again as on
the merge base, instead of being read in the join filter. The slt case for it
expects that error now.
##########
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:
Fixed in f2e861472. With marks on the build rows, `cross_product` now takes
the build rows in chunks of at most `batch_size`. Each chunk drops the rows
already marked and is then paired with the probe rows, which are collected once
and are at most a probe batch. The rows held at once no longer grow with the
build side.
This covers every caller: the fallback for a declined scope map, the
NULL-valued build rows, and the multi-column path without scope keys from
#25679, which allocated the same vector for each probe batch.
`cross_product_chunks_the_build_rows` pairs 10,000 build rows with a batch
size of 64 and checks that no more than 64 are held at once.
##########
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:
Done in f8a35e102, thanks.
--
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]
