fix equal_to in PrimitiveGroupValueBuilder

[Rachelint]

Which issue does this PR close?

Closes #.

Part of #12680

Rationale for this change

I found New added fuzz tests in #12667 failed after #12703

And after debugging, I found it is due to the change of equal_to function in PrimitiveGroupValueBuilder.

I think we should only check row's value when this row is not null in both exist array and input array.Because the actual value of row may be meaningless when it is null.

What changes are included in this PR?

Don't check the equality of actual values in exist row and input row, when both them are nulls.

Are these changes tested?

Test by new uts.

Are there any user-facing changes?

No

[alamb]

Thank you @Rachelint -- this is a really great find. Thank you 🙏

I left some suggestions

cc @Dandandan @jayzhan211

[alamb]

It might be faster to avoid calling is_null twice:

Suggested change

	let null_match = self.nulls.is_null(lhs_row) == array.is_null(rhs_row);
	let null_match = is_exist_null == array.is_null(rhs_row);

[Rachelint]

Thanks! Fixed.

[alamb]

I wonder if there is some way to write a reproducer in an end to end test (as in .slt as well) 🤔

[Rachelint]

🤔 I am trying it, but I still have no idea.
I found it through the fuzz tests in #12667 . And to be honest, I am still confused about why the null row will have the non-default value...

[Rachelint]

I got it !

It may be due to the compute::take function using to generate the random dataset.
https://github.com/Rachelint/arrow-datafusion/blob/9ad971be0c6c808e77a74cdfc571a33732a0838a/test-utils/src/array_gen/primitive.rs#L48-L62

Let's see take's implementation:

// In `take_primitive`:
let values_buf = take_native(values.values(), indices);
let nulls = take_nulls(values.nulls(), indices);

// In `take_native`:
    match indices.nulls().filter(|n| n.null_count() > 0) {
        Some(n) => indices
            .values()
            .iter()
            .enumerate()
            .map(|(idx, index)| match values.get(index.as_usize()) {
                Some(v) => *v,
                None => match n.is_null(idx) {
                    true => T::default(),
                    false => panic!("Out-of-bounds index {index:?}"),
                },
            })
            .collect(),

It will still try to take value from values ranther than using default value, even the row in indicies is null...

This logic seems unreasonable actually?

[Rachelint]

It may be a bit hard to produce it in end to end test.

The null row with a non-default value is only possible to exist in some special cases,
like generating through take as mentioned above, or as I remember we use it to improve filter performance in avg accumulator.

[jayzhan211]

👍

[alamb]

I was thinking about this last night and I think the byte buffer below has the same problem. I will make a follow on PR

[alamb]

Indeed it does -- #12770 to fix

[alamb]

THanks again @Rachelint -- very nice