Repository: gluesql/gluesql
Pull request #74 (2020-09-18 18:17:29)

Implement GROUP BY (#12)
#12 
It was... far beyond my expectation, I totally underestimate this issue.
Already too many changes, I'll implement `HAVING` by another PR.

```sql
SELECT SUM(quantity), COUNT(*), city FROM Item GROUP BY city
```
Now this kind of `GROUP BY` queries are supported!


Patch:
diff --git a/Cargo.toml b/Cargo.toml
index 037ca36b..e69044f2 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -18,8 +18,10 @@ sled-storage = ["sled", "bincode"]
 
 [dependencies]
 boolinator = "2.4.0"
+indexmap = "1.6.0"
 im-rc = "15.0.0"
 iter-enum = "0.2"
+itertools = "0.9.0"
 or-iterator = "0.1.1"
 serde = { version = "1.0.104", features = ["derive"] }
 sqlparser = { version = "0.6.1", features = ["serde"] }
diff --git a/src/data/value.rs b/src/data/value.rs
index 4e61358d..6e65bdfb 100644
--- a/src/data/value.rs
+++ b/src/data/value.rs
@@ -1,12 +1,13 @@
 use boolinator::Boolinator;
 use serde::{Deserialize, Serialize};
 use std::cmp::Ordering;
-use std::convert::TryFrom;
+use std::convert::{TryFrom, TryInto};
 use std::fmt::Debug;
 use thiserror::Error as ThisError;
 
 use sqlparser::ast::{DataType, Value as AstValue};
 
+use crate::executor::GroupKey;
 use crate::result::{Error, Result};
 
 #[derive(ThisError, Serialize, Debug, PartialEq)]
@@ -34,6 +35,9 @@ pub enum ValueError {
 
     #[error("null value on not null field")]
     NullValueOnNotNullField,
+
+    #[error("floating numbers cannot be grouped by")]
+    FloatCannotBeGroupedBy,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
@@ -159,6 +163,38 @@ impl TryFrom<&AstValue> for Value {
     }
 }
 
+impl TryInto<GroupKey> for &Value {
+    type Error = Error;
+
+    fn try_into(self) -> Result<GroupKey> {
+        use Value::*;
+
+        match self {
+            Bool(v) | OptBool(Some(v)) => Ok(GroupKey::Bool(*v)),
+            I64(v) | OptI64(Some(v)) => Ok(GroupKey::I64(*v)),
+            Str(v) | OptStr(Some(v)) => Ok(GroupKey::Str(v.clone())),
+            Empty | OptBool(None) | OptI64(None) | OptStr(None) => Ok(GroupKey::Null),
+            F64(_) | OptF64(_) => Err(ValueError::FloatCannotBeGroupedBy.into()),
+        }
+    }
+}
+
+impl TryInto<GroupKey> for Value {
+    type Error = Error;
+
+    fn try_into(self) -> Result<GroupKey> {
+        use Value::*;
+
+        match self {
+            Bool(v) | OptBool(Some(v)) => Ok(GroupKey::Bool(v)),
+            I64(v) | OptI64(Some(v)) => Ok(GroupKey::I64(v)),
+            Str(v) | OptStr(Some(v)) => Ok(GroupKey::Str(v)),
+            Empty | OptBool(None) | OptI64(None) | OptStr(None) => Ok(GroupKey::Null),
+            F64(_) | OptF64(_) => Err(ValueError::FloatCannotBeGroupedBy.into()),
+        }
+    }
+}
+
 trait BoolToValue: Sized {
     fn as_value(self, v1: Value, v2: Value) -> Value;
 }
diff --git a/src/executor/aggregate.rs b/src/executor/aggregate.rs
deleted file mode 100644
index ec6a42d0..00000000
--- a/src/executor/aggregate.rs
+++ /dev/null
@@ -1,250 +0,0 @@
-use im_rc::HashMap;
-use iter_enum::Iterator;
-use serde::Serialize;
-use std::cmp::Ordering;
-use std::fmt::Debug;
-use std::iter::{empty, once};
-use std::rc::Rc;
-use thiserror::Error;
-
-use sqlparser::ast::{Expr, Function, SelectItem};
-
-use super::context::{AggregateContext, BlendContext};
-use crate::data::{get_name, Value};
-use crate::result::Result;
-
-#[derive(Error, Serialize, Debug, PartialEq)]
-pub enum AggregateError {
-    #[error("unsupported compound identifier: {0}")]
-    UnsupportedCompoundIdentifier(String),
-
-    #[error("unsupported aggregation: {0}")]
-    UnsupportedAggregation(String),
-
-    #[error("only identifier is allowed in aggregation")]
-    OnlyIdentifierAllowed,
-
-    #[error("unreachable")]
-    Unreachable,
-}
-
-#[derive(Iterator)]
-enum Aggregated<I1, I2, I3> {
-    Applied(I1),
-    Skipped(I2),
-    Empty(I3),
-}
-
-pub struct Aggregate<'a> {
-    fields: &'a [SelectItem],
-}
-
-impl<'a> Aggregate<'a> {
-    pub fn new(fields: &'a [SelectItem]) -> Self {
-        Self { fields }
-    }
-
-    pub fn apply(
-        &self,
-        rows: impl Iterator<Item = Result<Rc<BlendContext<'a>>>>,
-    ) -> Result<impl Iterator<Item = Result<AggregateContext<'a>>>> {
-        if !self.check_aggregate() {
-            let rows = rows.map(|row| {
-                row.map(|blend_context| AggregateContext {
-                    aggregated: None,
-                    next: blend_context,
-                })
-            });
-
-            return Ok(Aggregated::Skipped(rows));
-        }
-
-        let (aggregated, next) = rows.enumerate().try_fold::<_, _, Result<_>>(
-            (HashMap::<&Function, (usize, Value)>::new(), None),
-            |(aggregated, _), (index, row)| {
-                let context = row?;
-
-                let aggregated = self
-                    .fields
-                    .iter()
-                    .try_fold(aggregated, |aggregated, field| match field {
-                        SelectItem::UnnamedExpr(expr) | SelectItem::ExprWithAlias { expr, .. } => {
-                            aggregate(index, aggregated, &context, &expr)
-                        }
-                        _ => Ok(aggregated),
-                    })?;
-
-                Ok((aggregated, Some(context)))
-            },
-        )?;
-
-        let next = match next {
-            Some(next) => next,
-            None => {
-                return Ok(Aggregated::Empty(empty()));
-            }
-        };
-
-        let aggregated: HashMap<&Function, Value> = aggregated
-            .iter()
-            .map(|(key, (_, value))| (*key, value.clone()))
-            .collect();
-        let aggregated = Some(aggregated);
-        let rows = once(Ok(AggregateContext { aggregated, next }));
-
-        Ok(Aggregated::Applied(rows))
-    }
-
-    fn check_aggregate(&self) -> bool {
-        self.fields.iter().any(|field| match field {
-            SelectItem::UnnamedExpr(expr) => check(expr),
-            SelectItem::ExprWithAlias { expr, .. } => check(expr),
-            _ => false,
-        })
-    }
-}
-
-fn aggregate<'a>(
-    index: usize,
-    aggregated: HashMap<&'a Function, (usize, Value)>,
-    context: &BlendContext<'_>,
-    expr: &'a Expr,
-) -> Result<HashMap<&'a Function, (usize, Value)>> {
-    let aggr = |aggregated, expr| aggregate(index, aggregated, context, expr);
-    let get_value = |expr: &Expr| match expr {
-        Expr::Identifier(ident) => context.get_value(&ident.value),
-        Expr::CompoundIdentifier(idents) => {
-            if idents.len() != 2 {
-                return Err(AggregateError::UnsupportedCompoundIdentifier(expr.to_string()).into());
-            }
-
-            let table_alias = &idents[0].value;
-            let column = &idents[1].value;
-
-            context.get_alias_value(table_alias, column)
-        }
-        _ => Err(AggregateError::OnlyIdentifierAllowed.into()),
-    };
-    let get_first_value = |args: &[Expr]| {
-        let expr = args.get(0).ok_or(AggregateError::Unreachable)?;
-
-        get_value(expr)
-    };
-
-    match expr {
-        Expr::Between {
-            expr, low, high, ..
-        } => [expr, low, high]
-            .iter()
-            .try_fold(aggregated, |aggregated, expr| aggr(aggregated, expr)),
-        Expr::BinaryOp { left, right, .. } => [left, right]
-            .iter()
-            .try_fold(aggregated, |aggregated, expr| aggr(aggregated, expr)),
-        Expr::UnaryOp { expr, .. } => aggr(aggregated, expr),
-        Expr::Nested(expr) => aggr(aggregated, expr),
-        Expr::Function(func) => {
-            let Function { name, args, .. } = func;
-
-            match get_name(name)?.to_uppercase().as_str() {
-                "COUNT" => {
-                    let expr = args.get(0).ok_or(AggregateError::Unreachable)?;
-
-                    let value_to_incr = Value::I64(match expr {
-                        Expr::Wildcard => 1,
-                        _ => {
-                            if get_value(expr)?.is_some() {
-                                1
-                            } else {
-                                0
-                            }
-                        }
-                    });
-
-                    match aggregated.get(func) {
-                        Some((current, value)) => {
-                            if &index <= current {
-                                return Ok(aggregated);
-                            }
-
-                            Ok(aggregated.update(func, (index, value_to_incr.add(value)?)))
-                        }
-                        None => Ok(aggregated.update(func, (index, value_to_incr))),
-                    }
-                }
-                "SUM" => {
-                    let value_to_sum = get_first_value(args)?;
-
-                    match aggregated.get(func) {
-                        Some((current, value)) => {
-                            if &index <= current {
-                                return Ok(aggregated);
-                            }
-
-                            Ok(aggregated.update(func, (index, value.add(value_to_sum)?)))
-                        }
-                        None => Ok(aggregated.update(func, (index, value_to_sum.clone()))),
-                    }
-                }
-                "MAX" => {
-                    let value_to_cmp = get_first_value(args)?;
-
-                    match aggregated.get(func) {
-                        Some((current, value)) => {
-                            if &index <= current {
-                                return Ok(aggregated);
-                            }
-
-                            match value.partial_cmp(value_to_cmp) {
-                                Some(ordering) => match ordering {
-                                    Ordering::Greater | Ordering::Equal => Ok(aggregated),
-                                    Ordering::Less => {
-                                        Ok(aggregated.update(func, (index, value_to_cmp.clone())))
-                                    }
-                                },
-                                None => Ok(aggregated),
-                            }
-                        }
-                        None => Ok(aggregated.update(func, (index, value_to_cmp.clone()))),
-                    }
-                }
-                "MIN" => {
-                    let value_to_cmp = get_first_value(args)?;
-
-                    match aggregated.get(func) {
-                        Some((current, value)) => {
-                            if &index <= current {
-                                return Ok(aggregated);
-                            }
-
-                            match value.partial_cmp(value_to_cmp) {
-                                Some(ordering) => match ordering {
-                                    Ordering::Less => Ok(aggregated),
-                                    Ordering::Equal | Ordering::Greater => {
-                                        Ok(aggregated.update(func, (index, value_to_cmp.clone())))
-                                    }
-                                },
-                                None => Ok(aggregated),
-                            }
-                        }
-                        None => Ok(aggregated.update(func, (index, value_to_cmp.clone()))),
-                    }
-                }
-                name => Err(AggregateError::UnsupportedAggregation(name.to_string()).into()),
-            }
-        }
-        _ => Ok(aggregated),
-    }
-}
-
-fn check(expr: &Expr) -> bool {
-    match expr {
-        Expr::Between {
-            expr, low, high, ..
-        } => check(expr) || check(low) || check(high),
-        Expr::BinaryOp { left, right, .. } => check(left) || check(right),
-        Expr::UnaryOp { expr, .. } => check(expr),
-        Expr::Nested(expr) => check(expr),
-        Expr::Function(_) => true,
-        _ => false,
-    }
-}
diff --git a/src/executor/aggregate/error.rs b/src/executor/aggregate/error.rs
new file mode 100644
index 00000000..77f8c500
--- /dev/null
+++ b/src/executor/aggregate/error.rs
@@ -0,0 +1,18 @@
+use serde::Serialize;
+use std::fmt::Debug;
+use thiserror::Error;
+
+#[derive(Error, Serialize, Debug, PartialEq)]
+pub enum AggregateError {
+    #[error("unsupported compound identifier: {0}")]
+    UnsupportedCompoundIdentifier(String),
+
+    #[error("unsupported aggregation: {0}")]
+    UnsupportedAggregation(String),
+
+    #[error("only identifier is allowed in aggregation")]
+    OnlyIdentifierAllowed,
+
+    #[error("unreachable")]
+    Unreachable,
+}
diff --git a/src/executor/aggregate/hash.rs b/src/executor/aggregate/hash.rs
new file mode 100644
index 00000000..7c210c72
--- /dev/null
+++ b/src/executor/aggregate/hash.rs
@@ -0,0 +1,27 @@
+use std::convert::{TryFrom, TryInto};
+
+use crate::data::Value;
+use crate::executor::evaluate::Evaluated;
+use crate::result::{Error, Result};
+
+#[derive(PartialEq, Eq, Hash, Clone, std::fmt::Debug)]
+pub enum GroupKey {
+    I64(i64),
+    Bool(bool),
+    Str(String),
+    Null,
+}
+
+impl TryFrom<&Evaluated<'_>> for GroupKey {
+    type Error = Error;
+
+    fn try_from(evaluated: &Evaluated<'_>) -> Result<Self> {
+        match evaluated {
+            Evaluated::LiteralRef(l) => Value::try_from(*l)?.try_into(),
+            Evaluated::Literal(l) => Value::try_from(l)?.try_into(),
+            Evaluated::ValueRef(v) => (*v).try_into(),
+            Evaluated::Value(v) => v.try_into(),
+            Evaluated::StringRef(s) => Ok(GroupKey::Str(s.to_string())),
+        }
+    }
+}
diff --git a/src/executor/aggregate/mod.rs b/src/executor/aggregate/mod.rs
new file mode 100644
index 00000000..d8fb3732
--- /dev/null
+++ b/src/executor/aggregate/mod.rs
@@ -0,0 +1,198 @@
+mod error;
+mod hash;
+mod state;
+
+use iter_enum::Iterator;
+use std::convert::TryFrom;
+use std::fmt::Debug;
+use std::rc::Rc;
+
+use sqlparser::ast::{Expr, Function, SelectItem};
+
+use super::context::{AggregateContext, BlendContext, FilterContext, UnionContext};
+use super::evaluate::{evaluate_union, Evaluated};
+use crate::data::{get_name, Value};
+use crate::result::Result;
+use crate::store::Store;
+
+pub use error::AggregateError;
+pub use hash::GroupKey;
+use state::State;
+
+#[derive(Iterator)]
+enum Aggregated<I1, I2> {
+    Applied(I1),
+    Skipped(I2),
+}
+
+pub struct Aggregate<'a, T: 'static + Debug> {
+    storage: &'a dyn Store<T>,
+    fields: &'a [SelectItem],
+    group_by: &'a [Expr],
+    filter_context: Option<&'a FilterContext<'a>>,
+}
+
+impl<'a, T: 'static + Debug> Aggregate<'a, T> {
+    pub fn new(
+        storage: &'a dyn Store<T>,
+        fields: &'a [SelectItem],
+        group_by: &'a [Expr],
+        filter_context: Option<&'a FilterContext<'a>>,
+    ) -> Self {
+        Self {
+            storage,
+            fields,
+            group_by,
+            filter_context,
+        }
+    }
+
+    pub fn apply(
+        &self,
+        rows: impl Iterator<Item = Result<Rc<BlendContext<'a>>>>,
+    ) -> Result<impl Iterator<Item = Result<AggregateContext<'a>>>> {
+        if !self.check_aggregate() {
+            let rows = rows.map(|row| {
+                row.map(|blend_context| AggregateContext {
+                    aggregated: None,
+                    next: blend_context,
+                })
+            });
+
+            return Ok(Aggregated::Skipped(rows));
+        }
+
+        let state =
+            rows.enumerate()
+                .try_fold::<_, _, Result<_>>(State::new(), |state, (index, row)| {
+                    let context = row?;
+                    let evaluated: Vec<Evaluated<'_>> = self
+                        .group_by
+                        .iter()
+                        .map(|expr| {
+                            let union_context =
+                                UnionContext::new(self.filter_context, Some(&context));
+
+                            evaluate_union(self.storage, union_context, None, expr)
+                        })
+                        .collect::<Result<_>>()?;
+                    let group = evaluated
+                        .iter()
+                        .map(GroupKey::try_from)
+                        .collect::<Result<Vec<GroupKey>>>()?;
+
+                    let state = state.apply(index, group, Rc::clone(&context));
+                    let state = self
+                        .fields
+                        .iter()
+                        .try_fold(state, |state, field| match field {
+                            SelectItem::UnnamedExpr(expr)
+                            | SelectItem::ExprWithAlias { expr, .. } => {
+                                aggregate(state, &context, &expr)
+                            }
+                            _ => Ok(state),
+                        })?;
+
+                    Ok(state)
+                })?;
+
+        let rows = state
+            .export()
+            .into_iter()
+            .filter_map(|(aggregated, next)| next.map(|next| (aggregated, next)))
+            .map(|(aggregated, next)| Ok(AggregateContext { aggregated, next }));
+
+        Ok(Aggregated::Applied(rows))
+    }
+
+    fn check_aggregate(&self) -> bool {
+        if !self.group_by.is_empty() {
+            return true;
+        }
+
+        self.fields.iter().any(|field| match field {
+            SelectItem::UnnamedExpr(expr) => check(expr),
+            SelectItem::ExprWithAlias { expr, .. } => check(expr),
+            _ => false,
+        })
+    }
+}
+
+fn aggregate<'a>(
+    state: State<'a>,
+    context: &BlendContext<'_>,
+    expr: &'a Expr,
+) -> Result<State<'a>> {
+    let aggr = |state, expr| aggregate(state, context, expr);
+    let get_value = |expr: &Expr| match expr {
+        Expr::Identifier(ident) => context.get_value(&ident.value),
+        Expr::CompoundIdentifier(idents) => {
+            if idents.len() != 2 {
+                return Err(AggregateError::UnsupportedCompoundIdentifier(expr.to_string()).into());
+            }
+
+            let table_alias = &idents[0].value;
+            let column = &idents[1].value;
+
+            context.get_alias_value(table_alias, column)
+        }
+        _ => Err(AggregateError::OnlyIdentifierAllowed.into()),
+    };
+    let get_first_value = |args: &[Expr]| {
+        let expr = args.get(0).ok_or(AggregateError::Unreachable)?;
+
+        get_value(expr)
+    };
+
+    match expr {
+        Expr::Between {
+            expr, low, high, ..
+        } => [expr, low, high]
+            .iter()
+            .try_fold(state, |state, expr| aggr(state, expr)),
+        Expr::BinaryOp { left, right, .. } => [left, right]
+            .iter()
+            .try_fold(state, |state, expr| aggr(state, expr)),
+        Expr::UnaryOp { expr, .. } => aggr(state, expr),
+        Expr::Nested(expr) => aggr(state, expr),
+        Expr::Function(func) => {
+            let Function { name, args, .. } = func;
+
+            match get_name(name)?.to_uppercase().as_str() {
+                "COUNT" => {
+                    let expr = args.get(0).ok_or(AggregateError::Unreachable)?;
+                    let value = Value::I64(match expr {
+                        Expr::Wildcard => 1,
+                        _ => {
+                            if get_value(expr)?.is_some() {
+                                1
+                            } else {
+                                0
+                            }
+                        }
+                    });
+
+                    state.add(func, &value)
+                }
+                "SUM" => state.add(func, get_first_value(args)?),
+                "MAX" => Ok(state.set_max(func, get_first_value(args)?)),
+                "MIN" => Ok(state.set_min(func, get_first_value(args)?)),
+                name => Err(AggregateError::UnsupportedAggregation(name.to_string()).into()),
+            }
+        }
+        _ => Ok(state),
+    }
+}
+
+fn check(expr: &Expr) -> bool {
+    match expr {
+        Expr::Between {
+            expr, low, high, ..
+        } => check(expr) || check(low) || check(high),
+        Expr::BinaryOp { left, right, .. } => check(left) || check(right),
+        Expr::UnaryOp { expr, .. } => check(expr),
+        Expr::Nested(expr) => check(expr),
+        Expr::Function(_) => true,
+        _ => false,
+    }
+}
diff --git a/src/executor/aggregate/state.rs b/src/executor/aggregate/state.rs
new file mode 100644
index 00000000..805377b1
--- /dev/null
+++ b/src/executor/aggregate/state.rs
@@ -0,0 +1,156 @@
+use im_rc::{HashMap, HashSet};
+use itertools::Itertools;
+use std::cmp::Ordering;
+use std::rc::Rc;
+
+use sqlparser::ast::Function;
+
+use crate::utils::{IndexMap, Vector};
+
+use super::hash::GroupKey;
+use crate::data::Value;
+use crate::executor::context::BlendContext;
+use crate::result::Result;
+
+type Group = Rc<Vec<GroupKey>>;
+type ValuesMap<'a> = HashMap<&'a Function, Value>;
+
+pub struct State<'a> {
+    index: usize,
+    group: Group,
+    values: IndexMap<(Group, &'a Function), (usize, Value)>,
+    groups: HashSet<Group>,
+    contexts: Vector<Rc<BlendContext<'a>>>,
+}
+
+impl<'a> State<'a> {
+    pub fn new() -> Self {
+        State {
+            index: 0,
+            group: Rc::new(vec![GroupKey::Null]),
+            values: IndexMap::new(),
+            groups: HashSet::new(),
+            contexts: Vector::new(),
+        }
+    }
+
+    pub fn apply(self, index: usize, group: Vec<GroupKey>, context: Rc<BlendContext<'a>>) -> Self {
+        let group = Rc::new(group);
+        let (groups, contexts) = if self.groups.contains(&group) {
+            (self.groups, self.contexts)
+        } else {
+            (
+                self.groups.update(Rc::clone(&group)),
+                self.contexts.push(context),
+            )
+        };
+
+        Self {
+            index,
+            group,
+            values: self.values,
+            groups,
+            contexts,
+        }
+    }
+
+    fn update(self, func: &'a Function, value: Value) -> Self {
+        let key = (Rc::clone(&self.group), func);
+        let (values, _) = self.values.insert(key, (self.index, value));
+
+        Self {
+            index: self.index,
+            group: self.group,
+            values,
+            groups: self.groups,
+            contexts: self.contexts,
+        }
+    }
+
+    fn get(&self, func: &'a Function) -> Option<&(usize, Value)> {
+        let group = Rc::clone(&self.group);
+
+        self.values.get(&(group, func))
+    }
+
+    pub fn export(self) -> Vec<(Option<ValuesMap<'a>>, Option<Rc<BlendContext<'a>>>)> {
+        let size = match self.values.keys().next() {
+            Some((target, _)) => match self.values.keys().position(|(group, _)| group != target) {
+                Some(size) => size,
+                None => self.values.len(),
+            },
+            None => {
+                return self.contexts.into_iter().map(|c| (None, Some(c))).collect();
+            }
+        };
+
+        let Self {
+            values, contexts, ..
+        } = self;
+
+        values
+            .into_iter()
+            .map(|(k, (_, v))| (k, v))
+            .chunks(size)
+            .into_iter()
+            .enumerate()
+            .map(|(i, entries)| {
+                let aggregated = entries
+                    .map(|((_, func), value)| (func, value))
+                    .collect::<HashMap<&'a Function, Value>>();
+                let next = contexts.get(i).map(Rc::clone);
+
+                (Some(aggregated), next)
+            })
+            .collect::<Vec<(Option<ValuesMap<'a>>, Option<Rc<BlendContext<'a>>>)>>()
+    }
+
+    pub fn add(self, func: &'a Function, target: &Value) -> Result<Self> {
+        let value = match self.get(func) {
+            Some((index, value)) => {
+                if &self.index <= index {
+                    return Ok(self);
+                }
+
+                target.add(value)?
+            }
+            None => target.clone(),
+        };
+
+        Ok(self.update(func, value))
+    }
+
+    pub fn set_max(self, func: &'a Function, target: &Value) -> Self {
+        if let Some((index, value)) = self.get(func) {
+            if &self.index <= index {
+                return self;
+            }
+
+            match value.partial_cmp(target) {
+                None | Some(Ordering::Greater) | Some(Ordering::Equal) => {
+                    return self;
+                }
+                Some(Ordering::Less) => (),
+            }
+        };
+
+        self.update(func, target.clone())
+    }
+
+    pub fn set_min(self, func: &'a Function, target: &Value) -> Self {
+        if let Some((index, value)) = self.get(func) {
+            if &self.index <= index {
+                return self;
+            }
+
+            match value.partial_cmp(target) {
+                None | Some(Ordering::Less) => {
+                    return self;
+                }
+                Some(Ordering::Equal) | Some(Ordering::Greater) => (),
+            }
+        }
+
+        self.update(func, target.clone())
+    }
+}
diff --git a/src/executor/context/mod.rs b/src/executor/context/mod.rs
index 329f96e3..bd2da1d1 100644
--- a/src/executor/context/mod.rs
+++ b/src/executor/context/mod.rs
@@ -1,7 +1,9 @@
 mod aggregate_context;
 mod blend_context;
 mod filter_context;
+mod union_context;
 
 pub use aggregate_context::AggregateContext;
 pub use blend_context::{BlendContext, BlendContextError};
 pub use filter_context::{FilterContext, FilterContextError};
+pub use union_context::{UnionContext, UnionContextError};
diff --git a/src/executor/context/union_context.rs b/src/executor/context/union_context.rs
new file mode 100644
index 00000000..911ddf84
--- /dev/null
+++ b/src/executor/context/union_context.rs
@@ -0,0 +1,52 @@
+use serde::Serialize;
+use std::fmt::Debug;
+use thiserror::Error;
+
+use super::{BlendContext, FilterContext};
+use crate::data::Value;
+use crate::result::Result;
+
+// TODO: add error test case
+#[derive(Error, Serialize, Debug, PartialEq)]
+pub enum UnionContextError {
+    #[error("value not found")]
+    ValueNotFound,
+}
+
+#[derive(Clone)]
+pub struct UnionContext<'a> {
+    pub filter_context: Option<&'a FilterContext<'a>>,
+    blend_context: Option<&'a BlendContext<'a>>,
+}
+
+impl<'a> UnionContext<'a> {
+    pub fn new(
+        filter_context: Option<&'a FilterContext<'a>>,
+        blend_context: Option<&'a BlendContext<'a>>,
+    ) -> Self {
+        Self {
+            filter_context,
+            blend_context,
+        }
+    }
+
+    pub fn get_value(&self, target: &str) -> Result<&'a Value> {
+        match (self.filter_context, self.blend_context) {
+            (Some(fc), Some(bc)) => fc.get_value(target).or_else(|_| bc.get_value(target)),
+            (Some(fc), None) => fc.get_value(target),
+            (None, Some(bc)) => bc.get_value(target),
+            (None, None) => Err(UnionContextError::ValueNotFound.into()),
+        }
+    }
+
+    pub fn get_alias_value(&self, table_alias: &str, target: &str) -> Result<&'a Value> {
+        match (self.filter_context, self.blend_context) {
+            (Some(fc), Some(bc)) => fc
+                .get_alias_value(table_alias, target)
+                .or_else(|_| bc.get_alias_value(table_alias, target)),
+            (Some(fc), None) => fc.get_alias_value(table_alias, target),
+            (None, Some(bc)) => bc.get_alias_value(table_alias, target),
+            (None, None) => Err(UnionContextError::ValueNotFound.into()),
+        }
+    }
+}
diff --git a/src/executor/evaluate/error.rs b/src/executor/evaluate/error.rs
index bb645df8..9f6c9f64 100644
--- a/src/executor/evaluate/error.rs
+++ b/src/executor/evaluate/error.rs
@@ -22,9 +22,6 @@ pub enum EvaluateError {
     #[error("unreachable literal arithmetic")]
     UnreachableLiteralArithmetic,
 
-    #[error("unreachable, filter context is empty: {0}")]
-    UnreachableEmptyFilterContext(String),
-
     #[error("unreachable, aggregated field not found {0}")]
     UnreachableAggregatedField(String),
 
diff --git a/src/executor/evaluate/mod.rs b/src/executor/evaluate/mod.rs
index 14d0a22f..31777770 100644
--- a/src/executor/evaluate/mod.rs
+++ b/src/executor/evaluate/mod.rs
@@ -6,7 +6,7 @@ use std::fmt::Debug;
 
 use sqlparser::ast::{BinaryOperator, Expr, Function, Value as AstValue};
 
-use super::context::FilterContext;
+use super::context::{FilterContext, UnionContext};
 use super::select::select;
 use crate::data::Value;
 use crate::result::Result;
@@ -21,7 +21,18 @@ pub fn evaluate<'a, T: 'static + Debug>(
     aggregated: Option<&HashMap<&Function, Value>>,
     expr: &'a Expr,
 ) -> Result<Evaluated<'a>> {
-    let eval = |expr| evaluate(storage, filter_context, aggregated, expr);
+    let context = UnionContext::new(filter_context, None);
+
+    evaluate_union(storage, context, aggregated, expr)
+}
+
+pub fn evaluate_union<'a, T: 'static + Debug>(
+    storage: &'a dyn Store<T>,
+    context: UnionContext<'a>,
+    aggregated: Option<&HashMap<&Function, Value>>,
+    expr: &'a Expr,
+) -> Result<Evaluated<'a>> {
+    let eval = |expr| evaluate_union(storage, context.clone(), aggregated, expr);
 
     match expr {
         Expr::Value(value) => match value {
@@ -33,14 +44,7 @@ pub fn evaluate<'a, T: 'static + Debug>(
         },
         Expr::Identifier(ident) => match ident.quote_style {
             Some(_) => Ok(Evaluated::StringRef(&ident.value)),
-            None => filter_context
-                .ok_or_else(|| {
-                    let name = ident.value.to_string();
-
-                    EvaluateError::UnreachableEmptyFilterContext(name)
-                })?
-                .get_value(&ident.value)
-                .map(Evaluated::ValueRef),
+            None => context.get_value(&ident.value).map(Evaluated::ValueRef),
         },
         Expr::Nested(expr) => eval(&expr),
         Expr::CompoundIdentifier(idents) => {
@@ -51,16 +55,11 @@ pub fn evaluate<'a, T: 'static + Debug>(
             let table_alias = &idents[0].value;
             let column = &idents[1].value;
 
-            filter_context
-                .ok_or_else(|| {
-                    let name = format!("{}.{}", table_alias, column);
-
-                    EvaluateError::UnreachableEmptyFilterContext(name)
-                })?
+            context
                 .get_alias_value(table_alias, column)
                 .map(Evaluated::ValueRef)
         }
-        Expr::Subquery(query) => select(storage, &query, filter_context)?
+        Expr::Subquery(query) => select(storage, &query, context.filter_context)?
             .map(|row| row?.take_first_value())
             .map(|value| value.map(Evaluated::Value))
             .next()
diff --git a/src/executor/mod.rs b/src/executor/mod.rs
index 99758100..c706a972 100644
--- a/src/executor/mod.rs
+++ b/src/executor/mod.rs
@@ -10,9 +10,9 @@ mod limit;
 mod select;
 mod update;
 
-pub use aggregate::AggregateError;
+pub use aggregate::{AggregateError, GroupKey};
 pub use blend::BlendError;
-pub use context::{BlendContextError, FilterContextError};
+pub use context::{BlendContextError, FilterContextError, UnionContextError};
 pub use evaluate::EvaluateError;
 pub use execute::{execute, ExecuteError, Payload};
 pub use filter::FilterError;
diff --git a/src/executor/select.rs b/src/executor/select.rs
index b2b2b312..2fd322f7 100644
--- a/src/executor/select.rs
+++ b/src/executor/select.rs
@@ -58,7 +58,7 @@ pub fn select<'a, T: 'static + Debug>(
     query: &'a Query,
     filter_context: Option<&'a FilterContext<'a>>,
 ) -> Result<impl Iterator<Item = Result<Row>> + 'a> {
-    let (table_with_joins, where_clause, projection) = match &query.body {
+    let (table_with_joins, where_clause, projection, group_by) = match &query.body {
         SetExpr::Select(statement) => {
             let tables = &statement.from;
             let table_with_joins = match tables.len() {
@@ -71,6 +71,7 @@ pub fn select<'a, T: 'static + Debug>(
                 table_with_joins,
                 statement.selection.as_ref(),
                 statement.projection.as_ref(),
+                &statement.group_by,
             )
         }
         _ => err!(SelectError::Unreachable),
@@ -93,7 +94,7 @@ pub fn select<'a, T: 'static + Debug>(
     let join_columns = Rc::new(join_columns);
 
     let join = Join::new(storage, joins, filter_context);
-    let aggregate = Aggregate::new(projection);
+    let aggregate = Aggregate::new(storage, projection, group_by, filter_context);
     let blend = Blend::new(storage, projection);
     let filter = Filter::new(storage, where_clause, filter_context);
     let limit = Limit::new(query.limit.as_ref(), query.offset.as_ref())?;
diff --git a/src/lib.rs b/src/lib.rs
index e2ed6fd6..c6cfe2be 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -62,6 +62,7 @@ mod executor;
 mod glue;
 mod parse;
 mod storages;
+mod utils;
 
 pub mod data;
 pub mod result;
diff --git a/src/result.rs b/src/result.rs
index ba290b9e..c3c77473 100644
--- a/src/result.rs
+++ b/src/result.rs
@@ -4,7 +4,7 @@ use thiserror::Error as ThisError;
 use crate::data::{RowError, TableError, ValueError};
 use crate::executor::{
     AggregateError, BlendContextError, BlendError, EvaluateError, ExecuteError, FilterContextError,
-    FilterError, JoinError, LimitError, SelectError, UpdateError,
+    FilterError, JoinError, LimitError, SelectError, UnionContextError, UpdateError,
 };
 use crate::store::StoreError;
 
@@ -37,6 +37,8 @@ pub enum Error {
     #[error(transparent)]
     FilterContext(#[from] FilterContextError),
     #[error(transparent)]
+    UnionContext(#[from] UnionContextError),
+    #[error(transparent)]
     Limit(#[from] LimitError),
     #[error(transparent)]
     Row(#[from] RowError),
@@ -65,6 +67,7 @@ impl PartialEq for Error {
             (Update(e), Update(e2)) => e == e2,
             (Filter(e), Filter(e2)) => e == e2,
             (FilterContext(e), FilterContext(e2)) => e == e2,
+            (UnionContext(e), UnionContext(e2)) => e == e2,
             (Limit(e), Limit(e2)) => e == e2,
             (Row(e), Row(e2)) => e == e2,
             (Table(e), Table(e2)) => e == e2,
diff --git a/src/utils/indexmap.rs b/src/utils/indexmap.rs
new file mode 100644
index 00000000..18b8d813
--- /dev/null
+++ b/src/utils/indexmap.rs
@@ -0,0 +1,36 @@
+use indexmap::map::{IntoIter, Keys};
+use std::cmp::Eq;
+use std::hash::Hash;
+
+/// HashMap which provides
+/// 1. Immutable APIs
+/// 2. Preserving insertion order
+pub struct IndexMap<K, V>(indexmap::IndexMap<K, V>);
+
+impl<K: Hash + Eq, V> IndexMap<K, V> {
+    pub fn new() -> Self {
+        Self(indexmap::IndexMap::new())
+    }
+
+    pub fn insert(mut self, key: K, value: V) -> (Self, Option<V>) {
+        let existing = self.0.insert(key, value);
+
+        (self, existing)
+    }
+
+    pub fn get(&self, key: &K) -> Option<&V> {
+        self.0.get(key)
+    }
+
+    pub fn keys(&self) -> Keys<K, V> {
+        self.0.keys()
+    }
+
+    pub fn len(&self) -> usize {
+        self.0.len()
+    }
+
+    pub fn into_iter(self) -> IntoIter<K, V> {
+        self.0.into_iter()
+    }
+}
diff --git a/src/utils/mod.rs b/src/utils/mod.rs
new file mode 100644
index 00000000..5a95ad8a
--- /dev/null
+++ b/src/utils/mod.rs
@@ -0,0 +1,5 @@
+mod indexmap;
+mod vector;
+
+pub use self::indexmap::IndexMap;
+pub use vector::Vector;
diff --git a/src/utils/vector.rs b/src/utils/vector.rs
new file mode 100644
index 00000000..de812561
--- /dev/null
+++ b/src/utils/vector.rs
@@ -0,0 +1,23 @@
+use std::vec::IntoIter;
+
+pub struct Vector<T>(Vec<T>);
+
+impl<T> Vector<T> {
+    pub fn new() -> Self {
+        Self(vec![])
+    }
+
+    pub fn push(mut self, value: T) -> Self {
+        self.0.push(value);
+
+        self
+    }
+
+    pub fn get(&self, i: usize) -> Option<&T> {
+        self.0.get(i)
+    }
+
+    pub fn into_iter(self) -> IntoIter<T> {
+        self.0.into_iter()
+    }
+}


Test patch:
diff --git a/src/tests/aggregate.rs b/src/tests/aggregate.rs
index 2816584f..b93d7fc2 100644
--- a/src/tests/aggregate.rs
+++ b/src/tests/aggregate.rs
@@ -83,3 +83,79 @@ pub fn aggregate(mut tester: impl tests::Tester) {
         .into_iter()
         .for_each(|(error, sql)| tester.test_error(sql, error));
 }
+
+pub fn group_by(mut tester: impl tests::Tester) {
+    let create_sql = "
+        CREATE TABLE Item (
+            id INTEGER,
+            quantity INTEGER NULL,
+            city TEXT,
+            ratio FLOAT,
+        );
+    ";
+
+    tester.run_and_print(create_sql);
+
+    let insert_sqls = [
+        "INSERT INTO Item (id, quantity, city, ratio) VALUES (1, 10, \"Seoul\", 0.2);",
+        "INSERT INTO Item (id, quantity, city, ratio) VALUES (2, 0, \"Dhaka\", 0.9);",
+        "INSERT INTO Item (id, quantity, city, ratio) VALUES (3, NULL, \"Beijing\", 1.1);",
+        "INSERT INTO Item (id, quantity, city, ratio) VALUES (3, 30, \"Daejeon\", 3.2);",
+        "INSERT INTO Item (id, quantity, city, ratio) VALUES (4, 11, \"Seoul\", 11.1);",
+        "INSERT INTO Item (id, quantity, city, ratio) VALUES (5, 24, \"Seattle\", 6.11);",
+    ];
+
+    for insert_sql in insert_sqls.iter() {
+        tester.run(insert_sql).unwrap();
+    }
+
+    let mut run = |sql| tester.run(sql).expect("select");
+
+    use Value::*;
+
+    let test_cases = vec![
+        (
+            "SELECT id, COUNT(*) FROM Item GROUP BY id",
+            select!(I64 I64; 1 1; 2 1; 3 2; 4 1; 5 1),
+        ),
+        (
+            "SELECT id FROM Item GROUP BY id",
+            select!(I64; 1; 2; 3; 4; 5),
+        ),
+        (
+            "SELECT SUM(quantity), COUNT(*), city FROM Item GROUP BY city",
+            select!(
+                 OptI64   I64 Str;
+                 Some(21) 2 "Seoul".to_owned();
+                 Some(0)  1 "Dhaka".to_owned();
+                 None     1 "Beijing".to_owned();
+                 Some(30) 1 "Daejeon".to_owned();
+                 Some(24) 1 "Seattle".to_owned()
+            ),
+        ),
+        (
+            "SELECT id, city FROM Item GROUP BY city",
+            select!(
+                 I64 Str;
+                 1 "Seoul".to_owned();
+                 2 "Dhaka".to_owned();
+                 3 "Beijing".to_owned();
+                 3 "Daejeon".to_owned();
+                 5 "Seattle".to_owned()
+            ),
+        ),
+    ];
+
+    test_cases
+        .into_iter()
+        .for_each(|(sql, expected)| assert_eq!(expected, run(sql)));
+
+    let error_cases = vec![(
+        ValueError::FloatCannotBeGroupedBy.into(),
+        "SELECT * FROM Item GROUP BY ratio;",
+    )];
+
+    error_cases
+        .into_iter()
+        .for_each(|(error, sql)| tester.test_error(sql, error));
+}
diff --git a/src/tests/mod.rs b/src/tests/mod.rs
index 511ffeaa..f7ace23e 100644
--- a/src/tests/mod.rs
+++ b/src/tests/mod.rs
@@ -35,6 +35,7 @@ macro_rules! generate_tests {
 
         glue!(basic, basic::basic);
         glue!(aggregate, aggregate::aggregate);
+        glue!(aggregate_group_by, aggregate::group_by);
         glue!(arithmetic, arithmetic::arithmetic);
         glue!(arithmetic_blend, arithmetic::blend);
         glue!(blend, blend::blend);