executor: refine HashAgg

[skyzh]

https://github.com/singularity-data/risinglight/pull/69#discussion_r744248427 As mentioned earlier, constructing a visibility bitmap for every unique group key incurs high time and space complexity. Instead, if we use the row-by-row update in the current implementation, we can avoid the cost from bitmap construction.

Originally posted by @pleiadesian in https://github.com/singularity-data/risinglight/pull/69#discussion_r744422237

[skyzh]

From my perspective, it's better to do either of the following:

1. update(&self, array: &ArrayImpl, visibility: impl Iterator<Item = bool>)
2. update_single(&self, array: &ArrayImpl, idx: usize)

@pleiadesian What do you think?

[pleiadesian]

Interface 2 LGTM. But no caller needs to update with an array and visibility in interface 1. So, update(&self, array: &ArrayImpl is enough currently.

[skyzh]

By the way, for every aggregate state, we need to rewrite the same logic twice: once for batch update, and once for single update... Any idea on simplifying this, or do you think this duplicated code is acceptable? In fact, using the first interface could help unify them.

fn update(&self, array: &ArrayImpl, visibility: Option<impl Iterator<Item = bool>>);

[pleiadesian]

I agree. But the cost of the unified interface is that we have to construct a bitmap every time we want to update the aggregate state. And it's O(N^2) time and space overhead when constructing a bitmap for every unique group key. I don't think it's acceptable when the row number is large.

Let's clarify the use case of the two interfaces above:

1. update_single: It's only called in HashAggExecutor, which updates the aggregate states row by row (1. find the corresponding state by the group key on the row; 2. update the state with values on the current row).
2. update: It's only called in SimpleAggExecutor, which updates the whole column to the aggregate state in batch. There is no need to specify a visibility bitmap for simple aggregation, as it can see the whole column instead of the rows corresponding to a group key.

[skyzh]

But the cost of the unified interface is that we have to construct a bitmap every time we want to update the aggregate state.

Actually not. As you can see, we pass an Iterator to the function. You can simply construct the iterator by:

state.update(array, rows.iter(|x| x == current_group_key))

... without constructing a new Vec<bool> or BitVec. This does not incur any additional cost.

[skyzh]

After a second thought, evan a visibility Iterator might require O(N^2) time complexity in the worst case, when all group keys are different. We need to call update N times, and inside each update, we need to iterate N elements in the array.

Then I have no further ideas on this. Would you please investigate a little bit about DuckDB or other systems on how they do it?