feat(udf): POC faster min max accumulator

[devanbenz]

Which issue does this PR close?

Closes #6906

Rationale for this change

What changes are included in this PR?

Are these changes tested?

Are there any user-facing changes?

[alamb]

I think much of the filter logic is handled by accumulate_indices

datafusion/datafusion/functions-aggregate/src/count.rs

Lines 417 to 424 in f54712d

	accumulate_indices(
	group_indices,
	values.logical_nulls().as_ref(),
	opt_filter,
	|group_index| {
	self.counts[group_index] += 1;
	},
	);

You could likely avoid much of this repetition (and likely it would be faster)

It woudl also be nice to avoid the duplication between min /max by using generics. Here is how the primitive one does it (passes in a comparison function)

https://github.com/apache/datafusion/blob/main/datafusion/functions-aggregate/src/min_max.rs#L119

[devanbenz]

Okay so I would probably want to do something like this?

accumulate_indices(group_indices, input_array.logical_nulls().as_ref(), opt_filter, |group_index| {
        let value = input_array.as_binary_view().value(i);
    
        let value_str = std::str::from_utf8(value).map_err(|e| {
            DataFusionError::Execution(format!(
                "could not build utf8 from binary view {}",
                e
            ))
        }).unwrap();
    
        if self.states[group_index].is_empty() {
            self.states[group_index] = value_str.to_string();
        } else {
            let curr_value_bytes = self.states[group_index].as_bytes();
            if value < curr_value_bytes {
                self.states[group_index] = value_str.parse().unwrap();
            }
        } 
});

And then make this generic. I.E. I can pass a generic function in instead of:

        if self.states[group_index].is_empty() {
            self.states[group_index] = value_str.to_string();
        } else {
            let curr_value_bytes = self.states[group_index].as_bytes();
            if value < curr_value_bytes {
                self.states[group_index] = value_str.parse().unwrap();
            }
        } 

Afterwards I can likely use a const generic for deciding how to down-cast here with string array or string view?

let value = input_array.as_binary_view().value(i);

[alamb]

Thanks @devanbenz -- I think this is on the right track

What I suggest is:

1. We start with StringArray (rather than StringViewArray)
2. Run some benchmarks (clickbench) -- I can do this if it would help

All in all, thanks again

[devanbenz]

@alamb thanks for taking a peek. Will go ahead and implement this for StringArray as well and modify the core functionality to lean on already existing methods + make it more generic so that MAX and MIN can share code. 🫡

[alamb]

BTW here is an example of using a const generic: #12703

[jayzhan211]

could replace with macro exec_err

[devanbenz]

TODO: Optimize downcasts (remove them)

[devanbenz]

TODO: Optimize from_utf8 and remove this validation

[devanbenz]

TODO: Remove downcasts

[alamb]

Some other notes I had from our talkL

1. Recommend you start with just support for StringArray (not StringView, LargeString, etc)
2. We will then successively build up to the fully general solution
3. The structure should likely be implemented in terms of StringArray (not ArrayRef)
4. When creating the final output we'll need to do some unsafe to avoid re-checking all UTF8 values

Should be a hoot

Thanks again

[alamb]

@devanbenz -- I am interested in pushing this one along. If I found some time, would you mind if I spent some time tweaking it?

[devanbenz]

@devanbenz -- I am interested in pushing this one along. If I found some time, would you mind if I spent some time tweaking it?

Please feel free to, I've been spending some time this morning trying to get it working with just StringArray but am having issues. Probably just a fundamental misunderstanding of how this should be achieved (skill issue :P). I'm attempting to set the inner state to Vec<StringArray> instead of Vec<String> but I often find that I need to do some hacky things to convert to Binary and then back to StringArray especially in the evaluate method. I am using the following signature and when I do get it working its actually less performant then my original methodology :(

pub struct StringGroupsAccumulator<F> {
    states: Vec<StringArray>,
    fun: F,
}

impl<F> StringGroupsAccumulator<F>
where
    F: Fn(&[u8], &[u8]) -> bool + Send + Sync,
{
    pub fn new(s_fn: F) -> Self {
        Self {
            states: Vec::new(),
            fun: s_fn,
        }
    }
}

This is certainly a task that has made me realize how much more I need to learn about Arrow semantics 🫡 😆 and just memory operations in general. I definitely take on more of a 'hacker' way of coding like just throwing things at the editor until they work in instances like this where I don't fully understand the underlying mechanisms. I have so much more to learn 😮

[jayzhan211]

I'm attempting to set the inner state to Vec instead of Vec

Why replace Vec with Vec? 🤔 Doesn't Vec<String> represent the best current value for each group? Since we only need to store one value per group, why make the change?

[alamb]

I started hacking on this in #12792 -- not quite done yet but we are getting close

[alamb]

BTW I think #12792 has all the right code -- I need to write a few more tests, but it has the basic idea completed

[alamb]

Given #12792 has been merged, let's close this one for now. Thank you @devanbenz and everyone else for the help.