Hash Join Vectorized collision checking

[Dandandan]

Which issue does this PR close?

Closes #50

Rationale for this change

Speed up for hash joins:

┏━━━━━━━━━━━━━━┳━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━┓
┃ Query        ┃     main ┃ vectorized_collision ┃        Change ┃
┡━━━━━━━━━━━━━━╇━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━┩
│ QQuery 1     │ 186.15ms │             185.23ms │     no change │
│ QQuery 2     │  68.78ms │              60.46ms │ +1.14x faster │
│ QQuery 3     │  53.50ms │              49.64ms │ +1.08x faster │
│ QQuery 4     │  38.11ms │              38.57ms │     no change │
│ QQuery 5     │ 112.94ms │             101.61ms │ +1.11x faster │
│ QQuery 6     │  11.18ms │              10.97ms │     no change │
│ QQuery 7     │ 228.21ms │             197.74ms │ +1.15x faster │
│ QQuery 8     │  80.80ms │              75.27ms │ +1.07x faster │
│ QQuery 9     │ 166.03ms │             145.26ms │ +1.14x faster │
│ QQuery 10    │ 106.34ms │             100.25ms │ +1.06x faster │
│ QQuery 11    │  55.83ms │              49.45ms │ +1.13x faster │
│ QQuery 12    │  70.98ms │              69.73ms │     no change │
│ QQuery 13    │ 217.69ms │             192.88ms │ +1.13x faster │
│ QQuery 14    │  13.77ms │              12.82ms │ +1.07x faster │
│ QQuery 15    │  23.83ms │              23.92ms │     no change │
│ QQuery 16    │  52.41ms │              53.94ms │     no change │
│ QQuery 17    │ 699.16ms │             699.39ms │     no change │
│ QQuery 18    │ 621.07ms │             547.10ms │ +1.14x faster │
│ QQuery 19    │  61.40ms │              60.98ms │     no change │
│ QQuery 20    │ 218.51ms │             212.73ms │     no change │
│ QQuery 21    │ 341.74ms │             275.30ms │ +1.24x faster │
│ QQuery 22    │  34.32ms │              29.75ms │ +1.15x faster │
└──────────────┴──────────┴──────────────────────┴───────────────┘

What changes are included in this PR?

Changes the collision checking to be vectorized.

Are these changes tested?

Are there any user-facing changes?

[metesynnada]

AFAIK null_equals_null is not supported in the arrow kernel.

[Dandandan]

Indeed - working on implementing this now

[metesynnada]

How about folding?

pub fn equal_rows_arr(
    indices_left: UInt64Array,
    indices_right: UInt32Array,
    left_arrays: &[ArrayRef],
    right_arrays: &[ArrayRef],
    null_equals_null: bool,
) -> Result<(UInt64Array, UInt32Array)> {
    let mut iter = left_arrays.iter().zip(right_arrays.iter());

    let (first_left, first_right) = iter.next().ok_or_else(|| {
        DataFusionError::Internal("At least one array should be provided for both left and right".to_string())
    })?;

    let arr_left = take(first_left.as_ref(), &indices_left, None)?;
    let arr_right = take(first_right.as_ref(), &indices_right, None)?;

    let mut equal = eq_dyn_null(&arr_left, &arr_right, null_equals_null)?;

    // Use map and try_fold to iterate over the remaining pairs of arrays.
    // In each iteration, take is used on the pair of arrays and their equality is determined.
    // The results are then folded (combined) using the and function to get a final equality result.
    equal = iter
        .map(|(left, right)| {
            let arr_left = take(left.as_ref(), &indices_left, None)?;
            let arr_right = take(right.as_ref(), &indices_right, None)?;
            eq_dyn_null(arr_left.as_ref(), arr_right.as_ref(), null_equals_null)
        })
        .try_fold(equal, |acc, res| {
            let equal2 = res?;
            and(&acc, &equal2)
        })?;
    // Continue
}

[metesynnada]

In addition, we do not need to take ownership of indices_left and indices_right.

[Dandandan]

👍

[metesynnada]

A null element can occur in an Int32Array as well. Maybe we should insert an option to arrow kernel, in arrow-rs.

[metesynnada]

I think we can open an issue on arrow-rs for this functionality.

[Dandandan]

apache/arrow-rs#4438

[metesynnada]

There is a significant improvement in the equality check, nice work. Maybe then the apply_join_filter_to_indices can be addressed, since the intermediate batch can be huge in low cardinality cases 😃

[metesynnada]

LGTM, nice work @Dandandan thanks for the effort.

[Dandandan]

LGTM, nice work @Dandandan thanks for the effort.

Thanks for the feedback!

[Dandandan]

Btw I think this can be applied to the symmetric hash join as well without changes.