Support vectorized append and compare for multi group by #12996
Conversation
| @@ -128,6 +132,15 @@ impl<T: ArrowPrimitiveType, const NULLABLE: bool> GroupColumn | |||
| } | |||
| } | |||
|
|
|||
| fn append_non_nullable_val(&mut self, array: &ArrayRef, row: usize) { | |||
| if NULLABLE { | |||
| self.nulls.append(false); | |||
There was a problem hiding this comment.
This could be optimized to append nulls for entire batch instead of per value
There was a problem hiding this comment.
Yes, I plan to refactor the interface for supporting input a rows: &[usize].
And make all parts' appending vectorized, and see the performance again.
There was a problem hiding this comment.
(i.e. remove it here and call it in such a way we use https://docs.rs/arrow/latest/arrow/array/struct.BooleanBufferBuilder.html#method.append_n
There was a problem hiding this comment.
I add the append_batch function to support vectorized append more better.
But the improvement seems still not obvious. #12996 (comment)
🤔 I guess, it is likely due the new introduced branch of equal_to:
if *group_idx < group_values_len {
for (i, group_val) in self.group_values.iter().enumerate() {
if !check_row_equal(group_val.as_ref(), *group_idx, &cols[i], row)
{
return false;
}
}
} else {
let row_idx_offset = group_idx - group_values_len;
let row_idx = self.append_rows_buffer[row_idx_offset];
return is_rows_eq(cols, row, cols, row_idx).unwrap();
}
There was a problem hiding this comment.
To eliminate this extra branch, I think we need to refactor the intern process metioned in #12821 (comment)
I am trying it.
|
The latest benchmark numbers: |
Rachelint
changed the title
| core(array, row); | ||
| struct AggregationHashTable<T: AggregationHashTableEntry> { | ||
| /// Raw table storing values in a `Vec` | ||
| raw_table: Vec<T>, |
There was a problem hiding this comment.
Based on some experiments in changing hash join algorithm, I think it's likely hashbrown performs much better than implementing a hashtable ourselves although I would like to be surprised 🙂
There was a problem hiding this comment.
Based on some experiments in changing hash join algorithm, I think it's likely
hashbrownperforms much better than implementing a hashtable ourselves although I would like to be surprised 🙂
🤔 Even if we can perform something like vectorized compare or vectorized append in our hashtable?
I found in multi group by case, we will perform the compare for each row leading to the array downcasting again and again... And actually the downcast operation will be compiled to many asm codes....
And I foudn we can't eliminate it and perform the vectorized compare with hashbrown...
fn equal_to_inner(&self, lhs_row: usize, array: &ArrayRef, rhs_row: usize) -> bool {
let array = array.as_byte_view::<B>();
There was a problem hiding this comment.
We can still do "vectorized compare" by doing the lookup in the hashtable (based on hash value only) and the vectorized equality check separately. That way you still can use the fast hashtable, but move the equality check to a separate/vectorized step.
That's at least what is done in the vectorized hash join implementation :). I changed it before to use a Vec-based index like you did here, but that performed significantly worse.
There was a problem hiding this comment.
The reason I think is that the lookup is incredibly well optimized using the swiss table design and you get fewer 'false" candidates to check for, while we can still use the vectorized/type specialized equality check.
There was a problem hiding this comment.
Make sense, thank you!
Rachelint
force-pushed
the
vectorize-append-value
branch
from
601c7b2
to
d4b5820
Compare
Rachelint
force-pushed
the
vectorize-append-value
branch
from
3415659
to
d79b813
Compare
|
The logic is a bit complex, I plan to finish and do benchmark for it today. |
🤔I think It can unify simply, The alternative is that we support a dedicated 🤔 I personally prefer the second one? What do you think about it @alamb ? |
| /// And we use [`GroupIndexView`] to represent such `group indices` in table. | ||
| /// | ||
| /// | ||
| map: RawTable<(u64, GroupIndexView)>, |
There was a problem hiding this comment.
Is it the case
- If group1 and group2 have exactly the same hash value,
GroupIndexViewwill use chaining to resolve the collision - If group1 and group2 have different hash values but map to the same slot in hash table,
hashbrownwill handle the collision for you with probing
…d operations to make code clearer.
I think this makes sense -- thank you |
|
BTW I think this code is fairly well covered by the aggregate fuzz tester (also added by @Rachelint :)) Also, @LeslieKid is adding additional data type coverage which is great: #13226 cargo test --test fuzz -- aggregate |
Rachelint
force-pushed
the
vectorize-append-value
branch
from
695d29c
to
2d982a1
Compare
Have unified the |
Rachelint
force-pushed
the
vectorize-append-value
branch
from
e0635fe
to
c6f8074
Compare
Rachelint
force-pushed
the
vectorize-append-value
branch
from
c6f8074
to
14fffb8
Compare
|
This is top of my list to review tomorrow morning |
|
I am giving this a final review now |
|
Performance results: 🚀 |
There was a problem hiding this comment.
👏 @Rachelint @jayzhan211 @2010YOUY01 and @Dandandan. What great teamwork
This PR is really nice in my opinion. It makes a super tricky and performance sensitive part of the code about as clear as I could imagine it to be.
I also ran some code coverage on this
nice cargo llvm-cov --html test --test fuzz -- aggregate
nice cargo llvm-cov --html test -p datafusion-physical-plan -- group_valuesAnd verified that the new code was well covered
| @@ -75,55 +148,653 @@ pub struct GroupValuesColumn { | |||
| random_state: RandomState, | |||
| } | |||
|
|
|||
| impl GroupValuesColumn { | |||
| /// Buffers to store intermediate results in `vectorized_append` | |||
| // ======================================================================== | ||
| // Initialization functions | ||
| // ======================================================================== | ||
|
|
||
| /// Create a new instance of GroupValuesColumn if supported for the specified schema | ||
| pub fn try_new(schema: SchemaRef) -> Result<Self> { | ||
| let map = RawTable::with_capacity(0); |
There was a problem hiding this comment.
This with_capacity can probably be improved (as a follow on PR) to avoid some smaller allocations
| /// `Group indices` order are against with their input order, and this will lead to error | ||
| /// in `streaming aggregation`. | ||
| /// | ||
| fn scalarized_intern( |
There was a problem hiding this comment.
this is basically the same as GroupValuesColumn::intern was previously, which makes sense to me
| @@ -56,14 +59,40 @@ pub trait GroupColumn: Send + Sync { | |||
| /// | |||
| /// Note that this comparison returns true if both elements are NULL | |||
| fn equal_to(&self, lhs_row: usize, array: &ArrayRef, rhs_row: usize) -> bool; | |||
|
|
|||
| /// Appends the row at `row` in `array` to this builder | |||
| fn append_val(&mut self, array: &ArrayRef, row: usize); | |||
There was a problem hiding this comment.
Maybe as a follow on we can consider removing append_val and equal_to and simpl change all codepaths to use the vectorized version
There was a problem hiding this comment.
I am a bit worried about if we merge them, some extra if else will be introduced.
It hurt much for performance for the row level operation.
There was a problem hiding this comment.
A good thing to benchmark (as a follow on PR) perhaps
| /// it will record the `true` result at the corresponding | ||
| /// position in `equal_to_results`. | ||
| /// | ||
| /// And if found nth result in `equal_to_results` is already |
There was a problem hiding this comment.
this is quite clever to pass in the existing "is equal to results"
|
|
||
| (false, _) => { | ||
| for &row in rows { | ||
| self.group_values.push(arr.value(row)); |
There was a problem hiding this comment.
that uf possible the inner loop just looks like this (memcopy!)
There was a problem hiding this comment.
😆 I think we can even do more, like check if rows.len() == array.len(), if so we just perform extend.
There was a problem hiding this comment.
I think we already could use extend instead of push? extend on Vec is somewhat faster than push as the capacity check / allocation is done once instead of once per value.
There was a problem hiding this comment.
I think there are several things that could be done to make the append even faster:
extend_from_sliceif rows.len() == array.len()- use
extendrather thanpushfor values - Speed up appending nulls (don't append bits one by one)
There was a problem hiding this comment.
I think we already could use extend instead of
push?extendonVecis somewhat faster than push as the capacity check / allocation is done once instead of once per value.
Ok, I got it, I think again and found it indeed simple to do it!
There was a problem hiding this comment.
I think there are several things that could be done to make the append even faster:
1. `extend_from_slice` `if rows.len() == array.len()` 2. use `extend` rather than `push` for values 3. Speed up appending nulls (don't append bits one by one)
I filed an issue to tracking the potential improvements for vecotrized operations.
#13275
| @@ -287,6 +469,63 @@ where | |||
| }; | |||
| } | |||
|
|
|||
| fn vectorized_equal_to( | |||
There was a problem hiding this comment.
What i have been dreaming about with @XiangpengHao is maybe something like adding take / filter to arrow array builders
I took this opportunity to write up the idea (finally) for your amusement:
|
As my admittedly sparse help for this PR I have filed some additional tickets for follow on work after this PR is merged: |
|
I don't think we need to wait on this PR anymore, let's merge it in and keep moving forward. Thank you everyone again! |
Which issue does this PR close?
Closes #.
Related to
Rationale for this change
Although
GroupValuesColumnis stored themulti gourp by valuesincolumn orientedway.However, it still use
row orientedapproach to performappendandequal to.The most obvious overhead is that we need to downcast the
arraywhen processing each row, and instructions for downcast is actually not few, and even worse it will introduce branches.And as I guess, the
row orientedapproach will also increase the random memory accesses but I am not sure.What changes are included in this PR?
This pr introduce the
vectorized appendandvectorized equal toforGroupValuesColumn.But such vectorized appoach is not compatible with
streaming aggregationdepending on the order betweeninput rowsand their correspondinggourp indices.So I define a new
VectorizedGroupValuesColumnfor optimizingnon streaming aggregationcases, and keep the originalGroupValuesColumnfor thestreaming aggregationcases.Are these changes tested?
Yes, I think enough new unit tests are added.
Are there any user-facing changes?
No.