mbutrovich commented on code in PR #26415:
URL: https://github.com/apache/datafusion/pull/26415#discussion_r4296649074
##########
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:
This PR changes what `probe_hit_rate` and `avg_fanout` report for these
joins, so what do you think about checking both here? Nothing covers them on
the new path today. When I replaced `probe_hit_rate.add_part(matched.len())`
([stream.rs#L1030](https://github.com/apache/datafusion/blob/73ecef5e658fb9466650077be1cdc79b2f9c19ef/datafusion/physical-plan/src/joins/hash_join/stream.rs#L1030))
with a no-op, neither the `joins::` unit tests nor the sqllogictests caught
it. `join_collect` returns the metrics, and
`join_probe_metrics_count_each_probe_row_once` already checks the same metrics
with `assert_ratio_metric`
([exec.rs#L5064-L5065](https://github.com/apache/datafusion/blob/73ecef5e658fb9466650077be1cdc79b2f9c19ef/datafusion/physical-plan/src/joins/hash_join/exec.rs#L5064-L5065)).
```suggestion
let (_, batches, metrics) =
join_collect(left, right, on, &join_type, null_equality,
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:?}");
// Probe rows 1, 3 and 4 have a build-side match.
assert_ratio_metric(&metrics, "probe_hit_rate", 3, 6);
// The probe stops at the first match, so the fanout is unknown.
assert_ratio_metric(&metrics, "avg_fanout", 0, 0);
```
##########
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:
Should this take the buffer instead of cloning it? `clone` allocates and
copies the matched indices on every probe batch. The `RightAnti` branch already
hands `matched` back to `self.probe_indices_buffer`
([stream.rs#L1036-L1039](https://github.com/apache/datafusion/blob/73ecef5e658fb9466650077be1cdc79b2f9c19ef/datafusion/physical-plan/src/joins/hash_join/stream.rs#L1036-L1039)).
With `take`, the `RightSemi` branch would also need to reclaim the buffer from
`right_indices` once the output batch is built, the way the general path does
with `array_map_buffers`
([stream.rs#L1221-L1227](https://github.com/apache/datafusion/blob/73ecef5e658fb9466650077be1cdc79b2f9c19ef/datafusion/physical-plan/src/joins/hash_join/stream.rs#L1221-L1227)).
Otherwise the next batch regrows the buffer from empty.
```suggestion
Ok(Some(std::mem::take(probe_indices_buffer)))
```
--
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]
