feat: add binder and executor for aggregation #69

pleiadesian · 2021-10-28T05:25:26Z

TODO in aggregation

Avg
Count (only rowcount is supported yet)
wildcard (e.g., count(*))
distinct keyword
aggregation with multiple arguments (e.g., json_object_agg ( key "any", value "any" ) -> json)

skyzh

Good work!

src/binder/expression/agg.rs

src/executor/aggregation/sum.rs

src/executor/hash_agg.rs

src/types/mod.rs

src/executor/hash_agg.rs

wangrunji0408 · 2021-11-01T15:59:15Z

It seems that aggregation.slt is blocked by float number casting...
You may add simple test cases and leave these big tests for future.

Some tips:

You can add # before subtest command since it's not supported yet.
You can add halt command to stop the test early.

skyzh

Good work! From my perspective, this PR is too large for a single review. I think it can be split at least into two parts:

add binder and executor for simple aggregation, and sum state
add group by support and HashAgg
(if you want to split further, ) add max/min and rowcount state

As this PR generally looks good to me and needs few modification before merging, I think it's okay to submit one big patch this time. For the next time, I think a PR of ~300 LoCs is of a reasonable size. Just add new functionalities little by little.

src/executor/aggregation/min_max.rs

skyzh · 2021-11-07T09:58:31Z

src/executor/aggregation/min_max.rs

+        match (array, &self.input_datatype, self.is_min) {
+            (ArrayImpl::Int32(arr), DataTypeKind::Int, true) => {
+                let mut temp: Option<i32> = None;
+                temp = arr.iter().fold(temp, min_i32);


Why not set the fold initial value to self.result? And we don't need the following match to do extra works.

self.result is DataValue. So, to get the result, we have
self.result = array.iter().fold(self.result.clone(), min_i32);
But we can not call iter() on ArrayImpl. And we have to refactor the min_i32 to calculate on ArrayImpl instead of i32.

src/executor/aggregation/min_max.rs

src/executor/aggregation/rowcount.rs

skyzh · 2021-11-07T10:03:35Z

src/executor/aggregation/min_max.rs

+                array.filter(visibility.iter().copied().collect::<Vec<_>>().into_iter())
+            }
+        };
+        match (array, &self.input_datatype, self.is_min) {


Typically, is_min should be some kind of const generic parameter, so that we can reduce runtime overhead. We may save the optimization later.

src/executor/hash_agg.rs

skyzh · 2021-11-07T10:06:04Z

src/executor/hash_agg.rs

+                for col in group_cols.iter() {
+                    group_key.push(col.get(row_idx));
+                }
+                let vis_map = key_to_vis_maps.entry(group_key.clone()).or_insert_with(|| {


So even if we do not need to insert a new group_key into the key_to_vis_maps, we still need to clone the group_key. I don't think this is necesssary.

Meanwhile, I suggest using Arc<str> instead of String everywhere when we need to handle user inputs, so as to reduce clone overhead. I'll draft a RFC about this later.

I suggest Cow<str>

src/executor/simple_agg.rs

wangrunji0408

LGTM in general. 👍

src/executor/hash_agg.rs

src/array/mod.rs

src/binder/expression/agg_call.rs

wangrunji0408 · 2021-11-07T10:16:34Z

src/types/mod.rs

+impl PartialEq for DataValue {
+    fn eq(&self, other: &Self) -> bool {
+        match (self, other) {
+            (Self::Null, Self::Null) => true,
+            (Self::Bool(left), Self::Bool(right)) => left == right,
+            (Self::Int32(left), Self::Int32(right)) => left == right,
+            (Self::Int64(left), Self::Int64(right)) => left == right,
+            (Self::String(left), Self::String(right)) => left == right,
+            (Self::Float64(left), Self::Float64(right)) => left == right,
+            _ => false,
+        }
+    }
+}


This implementation seems to be equivalent to #[derive(PartialEq)]?

PartialEq is added because of clippy's warning. Firstly, we need a custom Hash function for DataValue, which doesn't take the concrete type into account, and only feed the actual value into the hasher. On that, we need to also implement PartialEq by ourselves, instead of derive(PartialEq). However, I don't think this is a good thing to do in our system. Maybe we can take another look and change the implementation.

src/executor/aggregation/min_max.rs

src/executor/aggregation/rowcount.rs

src/executor/aggregation/min_max.rs

src/executor/simple_agg.rs

wangrunji0408 · 2021-11-07T11:55:51Z

src/executor/hash_agg.rs

+                for col in group_cols.iter() {
+                    group_key.push(col.get(row_idx));
+                }
+                let vis_map = key_to_vis_maps.entry(group_key.clone()).or_insert_with(|| {


I suggest Cow<str>

src/executor/hash_agg.rs

skyzh · 2021-11-07T16:00:29Z

src/executor/aggregation/min_max.rs

+            (ArrayImpl::Int32(arr), DataTypeKind::Int) => {
+                let temp = arr
+                    .iter()
+                    .fold(None, if self.is_min { min_i32 } else { max_i32 });


Moving such branch if self.is_min into a single iteration of loop might be very inefficient. We can make it into a const generics parameter later.

Oops, just find out that the if is used to decide which function to use, instead of inner function. This looks reasonable to me.

src/executor/simple_agg.rs

skyzh

Rest LGTM. I don't want this PR to be held for too many days, so we can get it merged for now, and resolve the minor issues in following PRs.

skyzh · 2021-11-08T05:14:46Z

src/binder/statement/select.rs

                SelectItem::Wildcard => {
+                    // TODO: support wildcard in aggregation


If this is not implemented, just panic.

skyzh · 2021-11-08T05:15:58Z

src/executor/aggregation/min_max.rs

+            (ArrayImpl::Int32(arr), DataTypeKind::Int) => {
+                let temp = arr
+                    .iter()
+                    .fold(None, if self.is_min { min_i32 } else { max_i32 });


Oops, just find out that the if is used to decide which function to use, instead of inner function. This looks reasonable to me.

skyzh · 2021-11-08T05:16:32Z

src/executor/aggregation/mod.rs

+
+/// `AggregationState` records the state of an aggregation
+pub trait AggregationState: 'static + Send + Sync {
+    fn update(&mut self, array: &ArrayImpl) -> Result<(), ExecutorError>;


Please also document the trait functions.

skyzh · 2021-11-08T05:17:25Z

src/executor/aggregation/mod.rs

+pub trait AggregationState: 'static + Send + Sync {
+    fn update(&mut self, array: &ArrayImpl) -> Result<(), ExecutorError>;
+
+    fn update_single(&mut self, value: &DataValue) -> Result<(), ExecutorError>;


Why do we need to implement update_single?

I see, it is used in HashAgg. I still prefer feeding a batch into the aggregator, and use a visibility bitmap or Iterator<Item = bool> to indicate valid entries. Let's do this later.

#69 (comment)
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.

skyzh · 2021-11-08T05:21:49Z

src/executor/hash_agg.rs

+        // Update states
+        let num_rows = chunk.cardinality();
+        for row_idx in 0..num_rows {
+            let mut group_key = HashKey::new();


Group key doesn't need to be constructed before checking its existence in state_entries. We can optimize this later.

skyzh · 2021-11-08T05:24:38Z

src/executor/simple_agg.rs

+                        builder.push(result);
+                        builder.finish()
+                    }
+                    None => ArrayBuilderImpl::new(&DataType::new(DataTypeKind::Int, true)).finish(),


When will the data_type be null in SimpleAgg?

When executing aggregation on an empty table, the result is None. Then the executor should return an empty array in the DataChunk.

But the type of the array should be determined in advance. e.g. SELECT sum(x) + 1.0 FROM table when x is f64. In this case, the SimpleAgg will return a I32Array, and the + 1.0 part might fail with mismatched expression type.

skyzh · 2021-11-08T05:25:04Z

src/executor/simple_agg.rs

+        Ok(())
+    }
+
+    fn finish_agg(states: SmallVec<[Box<dyn AggregationState>; 16]>) -> DataChunk {


Box<dyn AggregationState> can be defined as a separate type. pub type BoxedAggregationState = Box<dyn AggregationState>;

skyzh reviewed Oct 28, 2021

View reviewed changes

src/executor/hash_agg.rs Outdated Show resolved Hide resolved

pleiadesian mentioned this pull request Nov 1, 2021

feat: aggregation design #64

Closed

pleiadesian force-pushed the wzl-hash-aggr branch from b6f2fc1 to 36e4cfc Compare November 1, 2021 15:17

pleiadesian marked this pull request as ready for review November 7, 2021 09:09

pleiadesian force-pushed the wzl-hash-aggr branch from c48001d to 12bb5bc Compare November 7, 2021 09:58

pleiadesian requested review from skyzh, wangrunji0408 and MingjiHan99 November 7, 2021 09:59

skyzh reviewed Nov 7, 2021

View reviewed changes

wangrunji0408 reviewed Nov 7, 2021

View reviewed changes

skyzh reviewed Nov 7, 2021

View reviewed changes

src/executor/simple_agg.rs Outdated Show resolved Hide resolved

pleiadesian requested review from skyzh and wangrunji0408 November 8, 2021 04:44

skyzh approved these changes Nov 8, 2021

View reviewed changes

pleiadesian added 8 commits November 8, 2021 13:41

rebase from main

af22c6d

fix

4be050c

fix

acf8b41

minor fixes

b69bff7

refactor aggr executors

635bad4

fmt

c42788b

use SmallVec in agg

d3843d2

add more SmallVec

ef1dcb6

pleiadesian force-pushed the wzl-hash-aggr branch from 38c212d to ef1dcb6 Compare November 8, 2021 05:42

pleiadesian merged commit f5a6878 into main Nov 8, 2021

pleiadesian deleted the wzl-hash-aggr branch November 8, 2021 05:52

skyzh mentioned this pull request Nov 8, 2021

executor: refine HashAgg #122

Open

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat: add binder and executor for aggregation #69

feat: add binder and executor for aggregation #69

pleiadesian commented Oct 28, 2021 •

edited

skyzh left a comment

wangrunji0408 commented Nov 1, 2021

skyzh left a comment

skyzh Nov 7, 2021

pleiadesian Nov 7, 2021

skyzh Nov 7, 2021

skyzh Nov 7, 2021 •

edited

wangrunji0408 Nov 7, 2021

wangrunji0408 left a comment

wangrunji0408 Nov 7, 2021

skyzh Nov 7, 2021 •

edited

wangrunji0408 Nov 7, 2021

skyzh Nov 7, 2021 •

edited

skyzh Nov 8, 2021

skyzh left a comment •

edited

skyzh Nov 8, 2021

skyzh Nov 8, 2021

skyzh Nov 8, 2021

skyzh Nov 8, 2021

skyzh Nov 8, 2021

pleiadesian Nov 8, 2021

skyzh Nov 8, 2021

skyzh Nov 8, 2021

pleiadesian Nov 8, 2021

skyzh Nov 8, 2021

skyzh Nov 8, 2021

		SelectItem::Wildcard => {
		// TODO: support wildcard in aggregation

feat: add binder and executor for aggregation #69

feat: add binder and executor for aggregation #69

Conversation

pleiadesian commented Oct 28, 2021 • edited

skyzh left a comment

Choose a reason for hiding this comment

wangrunji0408 commented Nov 1, 2021

skyzh left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

skyzh Nov 7, 2021 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

wangrunji0408 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

skyzh Nov 7, 2021 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

skyzh Nov 7, 2021 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

skyzh left a comment • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

pleiadesian commented Oct 28, 2021 •

edited

skyzh Nov 7, 2021 •

edited

skyzh Nov 7, 2021 •

edited

skyzh Nov 7, 2021 •

edited

skyzh left a comment •

edited