andygrove commented on code in PR #26415:
URL: https://github.com/apache/datafusion/pull/26415#discussion_r4296761379
##########
datafusion/physical-plan/src/joins/hash_join/stream.rs:
##########
@@ -537,6 +518,128 @@ pub(super) fn lookup_join_hashmap(
Ok((build_indices, probe_indices, next_offset))
}
+/// Returns the probe rows of `state`, in ascending order, that have at least
one
+/// build-side row with an equal key. Used by `RightSemi`/`RightAnti` joins
+/// without a join filter.
+///
+/// Returns `None` if the build-side map cannot do first-match probing, in
which
+/// case the caller must use the general path. Unlike [`lookup_join_hashmap`],
+/// this stops at the first key-equal build row of each hash chain, so build
rows
+/// with duplicate keys cost one comparison per probe row instead of one per
+/// duplicate, and no probe row is produced twice.
+fn first_match_probe_rows(
+ left_data: &JoinLeftData,
+ state: &mut ProcessProbeBatchState,
+ hashes: &[u64],
+ null_equality: NullEquality,
+ probe_indices_buffer: &mut Vec<u32>,
+ build_indices_buffer: &mut Vec<u64>,
+) -> Result<Option<Vec<u32>>> {
+ match left_data.map() {
+ Map::ArrayMap(array_map) => {
+ array_map
+ .get_probe_indices_with_any_match(&state.values,
probe_indices_buffer)?;
+ Ok(Some(probe_indices_buffer.clone()))
Review Comment:
Good catch, and you called the regrow cost correctly: with a plain `take`
the dense `d100_h100` and `d50_h100` semi benchmarks were 13–15% slower. This
is addressed in 874c1e50f. The `ArrayMap` arm now takes the buffer, `RightAnti`
hands it back after `get_anti_indices`, and `RightSemi` reclaims it from
`right_indices` once the output batch is built. The benchmarks are back to
within noise.
##########
datafusion/physical-plan/src/joins/hash_join/exec.rs:
##########
@@ -6549,6 +6066,77 @@ mod tests {
Ok(())
}
+ /// `RightSemi`/`RightAnti` emit each probe row at most once, however many
+ /// build rows share its key, and treat NULL keys per the null equality.
+ #[apply(hash_join_exec_configs)]
+ #[tokio::test]
+ async fn join_right_semi_anti_duplicate_build_keys_and_null_probe_keys(
+ batch_size: usize,
+ use_perfect_hash_join_as_possible: bool,
+ ) -> Result<()> {
+ // Key 1 has three build rows, key 2 has one.
+ let build = || {
+ build_table_two_cols(
+ ("a1", &vec![Some(1), Some(1), Some(2), Some(1)]),
+ ("b1", &vec![Some(1), Some(1), Some(2), Some(1)]),
+ )
+ };
+ let probe = || {
+ build_table_two_cols(
+ (
+ "a2",
+ &vec![Some(0), Some(1), Some(2), Some(3), Some(4),
Some(5)],
+ ),
+ (
+ "b2",
+ &vec![None, Some(1), Some(3), Some(2), Some(1), Some(9)],
+ ),
+ )
+ };
+
+ for (join_type, null_equality, expected) in [
+ (
+ JoinType::RightSemi,
+ NullEquality::NullEqualsNothing,
+ vec![1, 3, 4],
+ ),
+ (
+ JoinType::RightAnti,
+ NullEquality::NullEqualsNothing,
+ vec![0, 2, 5],
+ ),
+ (
+ JoinType::RightSemi,
+ NullEquality::NullEqualsNull,
+ vec![1, 3, 4],
+ ),
+ (
+ JoinType::RightAnti,
+ NullEquality::NullEqualsNull,
+ vec![0, 2, 5],
+ ),
+ ] {
+ let task_ctx =
+ prepare_task_ctx(batch_size,
use_perfect_hash_join_as_possible);
+ let left = build();
+ let right = probe();
+ let on = vec![(
+ Arc::new(Column::new_with_schema("b1", &left.schema())?) as _,
+ Arc::new(Column::new_with_schema("b2", &right.schema())?) as _,
+ )];
+ let join = join(left, right, on, &join_type, null_equality)?;
+
+ let batches = common::collect(join.execute(0, task_ctx)?).await?;
+ let mut a2 = vec![];
+ for batch in &batches {
+ let col = batch.column(0).as_primitive::<Int32Type>();
+ a2.extend(col.values().iter().copied());
+ }
+ assert_eq!(a2, expected, "{join_type} with {null_equality:?}");
Review Comment:
Agreed, thanks for checking the no-op mutation. I took your suggestion as
written in d4a59565c: the test now uses `join_collect` and asserts
`probe_hit_rate` is 3/6 and `avg_fanout` is 0/0 for each variant.
--
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]
