Repository: gluesql/gluesql
Pull request #115 (2020-11-15 15:20:44)

Convert store traits to async
Now all existing store traits support `async`; `Store`, `StoreMut` and `AlterTable`.


* Most of internal functions which was returning `Iterator`, now those returns `TryStream` due to `Future` support.

* Change the way of adding integration tests.
Using `test_case!` macro, we can add a new test case and `run!`, `count!` and `test!` macros are now available to use.



Patch:
diff --git a/Cargo.toml b/Cargo.toml
index c794874f..09e557fb 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -25,12 +25,15 @@ alter-table = []
 sled-storage = ["sled", "bincode"]
 
 [dependencies]
+async-trait = "0.1.41"
+async-recursion = "0.3.1"
 boolinator = "2.4.0"
+futures = "0.3"
 indexmap = "1.6.0"
 im-rc = "15.0.0"
 iter-enum = "0.2.4"
 itertools = "0.9.0"
-or-iterator = "0.1.1"
+pin-project = "1.0.1"
 serde = { version = "1.0.117", features = ["derive"] }
 sqlparser = { version = "0.6.1", features = ["serde"] }
 thiserror = "1.0.21"
@@ -38,3 +41,6 @@ thiserror = "1.0.21"
 # dependencies for sled-storage
 bincode = { version = "1.3.1", optional = true }
 sled = { version = "0.34.4", optional = true }
+
+[dev-dependencies]
+tokio = { version = "0.3.3", features = ["macros", "rt"] }
diff --git a/examples/api_usage.rs b/examples/api_usage.rs
index 714f2da4..de48e4e8 100644
--- a/examples/api_usage.rs
+++ b/examples/api_usage.rs
@@ -1,4 +1,6 @@
 #[cfg(feature = "sled-storage")]
+use futures::executor::block_on;
+#[cfg(feature = "sled-storage")]
 use gluesql::{execute, parse, Glue, SledStorage};
 
 #[cfg(feature = "sled-storage")]
@@ -13,7 +15,7 @@ fn immutable_api() {
     ";
 
     parse(sqls).unwrap().iter().fold(storage, |storage, query| {
-        let (storage, _) = execute(storage, query).unwrap();
+        let (storage, _) = block_on(execute(storage, query)).unwrap();
 
         storage
     });
diff --git a/src/executor/aggregate/error.rs b/src/executor/aggregate/error.rs
index 77f8c500..377b596b 100644
--- a/src/executor/aggregate/error.rs
+++ b/src/executor/aggregate/error.rs
@@ -13,6 +13,9 @@ pub enum AggregateError {
     #[error("only identifier is allowed in aggregation")]
     OnlyIdentifierAllowed,
 
+    #[error("value not found: {0}")]
+    ValueNotFound(String),
+
     #[error("unreachable")]
     Unreachable,
 }
diff --git a/src/executor/aggregate/mod.rs b/src/executor/aggregate/mod.rs
index 58b3422f..8ebf5564 100644
--- a/src/executor/aggregate/mod.rs
+++ b/src/executor/aggregate/mod.rs
@@ -3,9 +3,10 @@ mod hash;
 mod state;
 
 use boolinator::Boolinator;
-use iter_enum::Iterator;
+use futures::stream::{self, StreamExt, TryStream, TryStreamExt};
 use std::convert::TryFrom;
 use std::fmt::Debug;
+use std::pin::Pin;
 use std::rc::Rc;
 
 use sqlparser::ast::{Expr, Function, SelectItem};
@@ -14,7 +15,7 @@ use super::context::{AggregateContext, BlendContext, FilterContext};
 use super::evaluate::{evaluate, Evaluated};
 use super::filter::check_expr;
 use crate::data::{get_name, Value};
-use crate::result::Result;
+use crate::result::{Error, Result};
 use crate::store::Store;
 
 pub use error::AggregateError;
@@ -29,6 +30,9 @@ pub struct Aggregate<'a, T: 'static + Debug> {
     filter_context: Option<Rc<FilterContext<'a>>>,
 }
 
+type Applied<'a> = dyn TryStream<Ok = AggregateContext<'a>, Error = Error, Item = Result<AggregateContext<'a>>>
+    + 'a;
+
 impl<'a, T: 'static + Debug> Aggregate<'a, T> {
     pub fn new(
         storage: &'a dyn Store<T>,
@@ -46,81 +50,102 @@ impl<'a, T: 'static + Debug> Aggregate<'a, T> {
         }
     }
 
-    pub fn apply(
+    pub async fn apply(
         &self,
-        rows: impl Iterator<Item = Result<Rc<BlendContext<'a>>>>,
-    ) -> Result<impl Iterator<Item = Result<AggregateContext<'a>>>> {
-        #[derive(Iterator)]
-        enum Aggregated<I1, I2> {
-            Applied(I1),
-            Skipped(I2),
-        }
-
+        rows: impl TryStream<Ok = Rc<BlendContext<'a>>, Error = Error> + 'a,
+    ) -> Result<Pin<Box<Applied<'a>>>> {
         if !self.check_aggregate()? {
-            let rows = rows.map(|row| {
-                row.map(|blend_context| AggregateContext {
-                    aggregated: None,
-                    next: blend_context,
-                })
+            let rows = rows.map_ok(|blend_context| AggregateContext {
+                aggregated: None,
+                next: blend_context,
             });
 
-            return Ok(Aggregated::Skipped(rows));
+            return Ok(Box::pin(rows));
         }
 
-        let state =
-            rows.enumerate()
-                .try_fold::<_, _, Result<_>>(State::new(), |state, (index, row)| {
-                    let blend_context = row?;
-                    let evaluated: Vec<Evaluated<'_>> = self
-                        .group_by
-                        .iter()
-                        .map(|expr| {
-                            let filter_context = self.filter_context.as_ref().map(Rc::clone);
-                            let filter_context = blend_context.concat_into(filter_context);
-
-                            evaluate(self.storage, filter_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(&blend_context));
-                    let state = self
-                        .fields
-                        .iter()
-                        .try_fold(state, |state, field| match field {
-                            SelectItem::UnnamedExpr(expr)
-                            | SelectItem::ExprWithAlias { expr, .. } => {
-                                aggregate(state, &blend_context, &expr)
+        let state = rows
+            .into_stream()
+            .enumerate()
+            .map(|(i, row)| row.map(|row| (i, row)))
+            .try_fold(State::new(), |state, (index, blend_context)| async move {
+                let evaluated: Vec<Evaluated<'_>> =
+                    stream::iter(self.group_by.iter())
+                        .then(|expr| {
+                            let filter_context = FilterContext::concat(
+                                self.filter_context.as_ref().map(Rc::clone),
+                                Some(&blend_context).map(Rc::clone),
+                            );
+                            let filter_context = Some(filter_context).map(Rc::new);
+
+                            async move {
+                                evaluate(self.storage, filter_context, None, expr, false).await
                             }
-                            _ => Ok(state),
-                        })?;
+                        })
+                        .try_collect::<Vec<_>>()
+                        .await?;
+                let group = evaluated
+                    .iter()
+                    .map(GroupKey::try_from)
+                    .collect::<Result<Vec<GroupKey>>>()?;
 
-                    Ok(state)
-                })?;
+                let state = state.apply(index, group, Rc::clone(&blend_context));
+                let state = self
+                    .fields
+                    .iter()
+                    .try_fold(state, |state, field| match field {
+                        SelectItem::UnnamedExpr(expr) | SelectItem::ExprWithAlias { expr, .. } => {
+                            aggregate(state, &blend_context, &expr)
+                        }
+                        _ => Ok(state),
+                    })?;
+
+                Ok(state)
+            })
+            .await?;
 
         let storage = self.storage;
         let filter_context = self.filter_context.as_ref().map(Rc::clone);
         let having = self.having;
+
         let rows = state
             .export()
             .into_iter()
-            .filter_map(|(aggregated, next)| next.map(|next| (aggregated, next)))
-            .filter_map(move |(aggregated, next)| match having {
-                Some(having) => {
-                    let filter_context = filter_context.as_ref().map(Rc::clone);
-                    let filter_context = next.concat_into(filter_context);
-
-                    check_expr(storage, filter_context, aggregated.as_ref(), having)
-                        .map(|pass| pass.as_some(AggregateContext { aggregated, next }))
-                        .transpose()
+            .filter_map(|(aggregated, next)| next.map(|next| (aggregated, next)));
+        let rows = stream::iter(rows)
+            .filter_map(move |(aggregated, next)| {
+                let filter_context = filter_context.as_ref().map(Rc::clone);
+                let aggregated = aggregated.map(Rc::new);
+
+                async move {
+                    match having {
+                        None => Some(Ok((aggregated.as_ref().map(Rc::clone), next))),
+                        Some(having) => {
+                            let filter_context =
+                                FilterContext::concat(filter_context, Some(Rc::clone(&next)));
+                            let filter_context = Some(filter_context).map(Rc::new);
+                            let aggregated = aggregated.as_ref().map(Rc::clone);
+
+                            check_expr(
+                                storage,
+                                filter_context,
+                                aggregated.as_ref().map(Rc::clone),
+                                having,
+                            )
+                            .await
+                            .map(|pass| pass.as_some((aggregated, next)))
+                            .transpose()
+                        }
+                    }
                 }
-                None => Some(Ok(AggregateContext { aggregated, next })),
+            })
+            .map_ok(|(aggregated, next)| {
+                // TODO: Remove unwrap!
+                let aggregated = aggregated.map(|a| Rc::try_unwrap(a).unwrap());
+
+                AggregateContext { aggregated, next }
             });
 
-        Ok(Aggregated::Applied(rows))
+        Ok(Box::pin(rows))
     }
 
     fn check_aggregate(&self) -> Result<bool> {
@@ -147,18 +172,24 @@ fn aggregate<'a>(
 ) -> 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::Identifier(ident) => context
+            .get_value(&ident.value)
+            .ok_or_else(|| AggregateError::ValueNotFound(ident.value.to_string())),
         Expr::CompoundIdentifier(idents) => {
             if idents.len() != 2 {
-                return Err(AggregateError::UnsupportedCompoundIdentifier(expr.to_string()).into());
+                return Err(AggregateError::UnsupportedCompoundIdentifier(
+                    expr.to_string(),
+                ));
             }
 
             let table_alias = &idents[0].value;
             let column = &idents[1].value;
 
-            context.get_alias_value(table_alias, column)
+            context
+                .get_alias_value(table_alias, column)
+                .ok_or_else(|| AggregateError::ValueNotFound(column.to_string()))
         }
-        _ => Err(AggregateError::OnlyIdentifierAllowed.into()),
+        _ => Err(AggregateError::OnlyIdentifierAllowed),
     };
     let get_first_value = |args: &[Expr]| {
         let expr = args.get(0).ok_or(AggregateError::Unreachable)?;
diff --git a/src/executor/blend.rs b/src/executor/blend.rs
index 4191880d..0acf0d90 100644
--- a/src/executor/blend.rs
+++ b/src/executor/blend.rs
@@ -1,21 +1,20 @@
+use futures::stream::{self, StreamExt, TryStreamExt};
 use im_rc::HashMap;
-use iter_enum::Iterator;
 use serde::Serialize;
 use std::convert::TryInto;
 use std::fmt::Debug;
-use std::iter::once;
 use std::rc::Rc;
-use thiserror::Error;
+use thiserror::Error as ThisError;
 
 use sqlparser::ast::{Function, SelectItem};
 
-use super::context::{AggregateContext, BlendContext};
+use super::context::{AggregateContext, BlendContext, FilterContext};
 use super::evaluate::evaluate;
 use crate::data::{get_name, Row, Value};
-use crate::result::Result;
+use crate::result::{Error, Result};
 use crate::store::Store;
 
-#[derive(Error, Serialize, Debug, PartialEq)]
+#[derive(ThisError, Serialize, Debug, PartialEq)]
 pub enum BlendError {
     #[error("table alias not found: {0}")]
     TableAliasNotFound(String),
@@ -26,76 +25,62 @@ pub struct Blend<'a, T: 'static + Debug> {
     fields: &'a [SelectItem],
 }
 
-#[derive(Iterator)]
-enum Blended<I1, I2, I3, I4> {
-    All(I1),
-    AllInTable(I2),
-    Single(I3),
-    Err(I4),
-}
-
 impl<'a, T: 'static + Debug> Blend<'a, T> {
     pub fn new(storage: &'a dyn Store<T>, fields: &'a [SelectItem]) -> Self {
         Self { storage, fields }
     }
 
-    pub fn apply(&self, context: Result<AggregateContext<'a>>) -> Result<Row> {
+    pub async fn apply(&self, context: Result<AggregateContext<'a>>) -> Result<Row> {
         let AggregateContext { aggregated, next } = context?;
-        let values = self.blend(aggregated, next)?;
+        let values = self.blend(aggregated, next).await?;
 
         Ok(Row(values))
     }
 
-    fn blend(
+    async fn blend(
         &self,
         aggregated: Option<HashMap<&'a Function, Value>>,
         context: Rc<BlendContext<'a>>,
     ) -> Result<Vec<Value>> {
-        macro_rules! err {
-            ($err: expr) => {
-                Blended::Err(once(Err($err.into())))
-            };
-        }
-
-        macro_rules! try_into {
-            ($v: expr) => {
-                match $v {
-                    Ok(v) => v,
-                    Err(e) => {
-                        return err!(e);
-                    }
-                }
-            };
-        }
-
-        let filter_context = context.concat_into(None);
-
-        self.fields
-            .iter()
-            .flat_map(|item| match item {
-                SelectItem::Wildcard => {
-                    let values = context.get_all_values().into_iter().map(Ok);
-
-                    Blended::All(values)
-                }
-                SelectItem::QualifiedWildcard(alias) => {
-                    let table_alias = try_into!(get_name(alias));
-
-                    match context.get_alias_values(table_alias) {
-                        Some(values) => Blended::AllInTable(values.into_iter().map(Ok)),
-                        None => err!(BlendError::TableAliasNotFound(table_alias.to_string())),
+        let filter_context = FilterContext::concat(None, Some(Rc::clone(&context)));
+        let filter_context = Some(filter_context).map(Rc::new);
+
+        let aggregated = aggregated.map(Rc::new);
+
+        let values = stream::iter(self.fields.iter())
+            .map(Ok::<&'a SelectItem, Error>)
+            .and_then(|item| {
+                let context = Rc::clone(&context);
+                let filter_context = filter_context.as_ref().map(Rc::clone);
+                let aggregated = aggregated.as_ref().map(Rc::clone);
+
+                async move {
+                    match item {
+                        SelectItem::Wildcard => Ok(context.get_all_values()),
+                        SelectItem::QualifiedWildcard(alias) => {
+                            let table_alias = get_name(alias)?;
+
+                            match context.get_alias_values(table_alias) {
+                                Some(values) => Ok(values),
+                                None => {
+                                    Err(BlendError::TableAliasNotFound(table_alias.to_string())
+                                        .into())
+                                }
+                            }
+                        }
+                        SelectItem::UnnamedExpr(expr) | SelectItem::ExprWithAlias { expr, .. } => {
+                            evaluate(self.storage, filter_context, aggregated, expr, true)
+                                .await
+                                .map(|evaluated| evaluated.try_into())?
+                                .map(|v| vec![v])
+                        }
                     }
                 }
-                SelectItem::UnnamedExpr(expr) | SelectItem::ExprWithAlias { expr, .. } => {
-                    let filter_context = filter_context.as_ref().map(Rc::clone);
-
-                    let value = evaluate(self.storage, filter_context, aggregated.as_ref(), expr)
-                        .map(|evaluated| evaluated.try_into());
-                    let value = try_into!(value);
-
-                    Blended::Single(once(value))
-                }
             })
-            .collect::<Result<_>>()
+            .try_collect::<Vec<Vec<_>>>()
+            .await?
+            .concat();
+
+        Ok(values)
     }
 }
diff --git a/src/executor/context/blend_context.rs b/src/executor/context/blend_context.rs
index 00abd4dc..5bc057dd 100644
--- a/src/executor/context/blend_context.rs
+++ b/src/executor/context/blend_context.rs
@@ -1,25 +1,15 @@
-use serde::Serialize;
 use std::fmt::Debug;
 use std::rc::Rc;
-use thiserror::Error;
 
 use sqlparser::ast::Ident;
 
-use super::FilterContext;
 use crate::data::{Row, Value};
-use crate::result::Result;
-
-#[derive(Error, Serialize, Debug, PartialEq)]
-pub enum BlendContextError {
-    #[error("value not found")]
-    ValueNotFound,
-}
 
 #[derive(Debug)]
 pub struct BlendContext<'a> {
     table_alias: &'a str,
-    columns: Rc<Vec<Ident>>,
-    row: Option<Row>,
+    pub columns: Rc<Vec<Ident>>,
+    pub row: Option<Row>,
     next: Option<Rc<BlendContext<'a>>>,
 }
 
@@ -38,46 +28,24 @@ impl<'a> BlendContext<'a> {
         }
     }
 
-    pub fn concat_into(
-        &'a self,
-        filter_context: Option<Rc<FilterContext<'a>>>,
-    ) -> Option<Rc<FilterContext<'a>>> {
-        let BlendContext {
-            table_alias,
-            columns,
-            row,
-            next,
-            ..
-        } = self;
-
-        let filter_context =
-            FilterContext::new(table_alias, &columns, row.as_ref(), filter_context);
-        let filter_context = Some(Rc::new(filter_context));
-
-        match next {
-            Some(next) => next.concat_into(filter_context),
-            None => filter_context,
-        }
-    }
-
-    pub fn get_value(&'a self, target: &str) -> Result<&'a Value> {
+    pub fn get_value(&'a self, target: &str) -> Option<&'a Value> {
         let get_value = || {
             self.columns
                 .iter()
                 .position(|column| column.value == target)
-                .and_then(|index| self.row.as_ref().and_then(|row| row.get_value(index)))
+                .map(|index| self.row.as_ref().and_then(|row| row.get_value(index)))
         };
 
         match get_value() {
             None => match &self.next {
-                None => Err(BlendContextError::ValueNotFound.into()),
+                None => None,
                 Some(context) => context.get_value(target),
             },
-            Some(value) => Ok(value),
+            Some(value) => value,
         }
     }
 
-    pub fn get_alias_value(&'a self, table_alias: &str, target: &str) -> Result<&'a Value> {
+    pub fn get_alias_value(&'a self, table_alias: &str, target: &str) -> Option<&'a Value> {
         let get_value = || {
             if self.table_alias != table_alias {
                 return None;
@@ -86,15 +54,15 @@ impl<'a> BlendContext<'a> {
             self.columns
                 .iter()
                 .position(|column| column.value == target)
-                .and_then(|index| self.row.as_ref().and_then(|row| row.get_value(index)))
+                .map(|index| self.row.as_ref().and_then(|row| row.get_value(index)))
         };
 
         match get_value() {
             None => match &self.next {
-                None => Err(BlendContextError::ValueNotFound.into()),
+                None => None,
                 Some(context) => context.get_alias_value(table_alias, target),
             },
-            Some(value) => Ok(value),
+            Some(value) => value,
         }
     }
 
diff --git a/src/executor/context/filter_context.rs b/src/executor/context/filter_context.rs
index 71f31c52..4d98c1b8 100644
--- a/src/executor/context/filter_context.rs
+++ b/src/executor/context/filter_context.rs
@@ -1,77 +1,111 @@
-use serde::Serialize;
 use std::fmt::Debug;
 use std::rc::Rc;
-use thiserror::Error;
 
 use sqlparser::ast::Ident;
 
+use super::BlendContext;
 use crate::data::{Row, Value};
-use crate::result::Result;
 
-#[derive(Error, Serialize, Debug, PartialEq)]
-pub enum FilterContextError {
-    #[error("value not found")]
-    ValueNotFound,
+#[derive(Debug)]
+enum Content<'a> {
+    Some {
+        table_alias: &'a str,
+        columns: Rc<Vec<Ident>>,
+        row: Option<&'a Row>,
+    },
+    None,
 }
 
 #[derive(Debug)]
 pub struct FilterContext<'a> {
-    table_alias: &'a str,
-    columns: &'a [Ident],
-    row: Option<&'a Row>,
+    content: Content<'a>,
     next: Option<Rc<FilterContext<'a>>>,
+    next2: Option<Rc<BlendContext<'a>>>,
 }
 
 impl<'a> FilterContext<'a> {
     pub fn new(
         table_alias: &'a str,
-        columns: &'a [Ident],
+        columns: Rc<Vec<Ident>>,
         row: Option<&'a Row>,
         next: Option<Rc<FilterContext<'a>>>,
     ) -> Self {
         Self {
-            table_alias,
-            columns,
-            row,
+            content: Content::Some {
+                table_alias,
+                columns,
+                row,
+            },
             next,
+            next2: None,
+        }
+    }
+
+    pub fn concat(
+        filter_context: Option<Rc<FilterContext<'a>>>,
+        blend_context: Option<Rc<BlendContext<'a>>>,
+    ) -> Self {
+        Self {
+            content: Content::None,
+            next: filter_context,
+            next2: blend_context,
         }
     }
 
-    pub fn get_value(&self, target: &str) -> Result<Option<&'a Value>> {
-        let get_value = || {
-            self.columns
+    pub fn get_value(&'a self, target: &str) -> Option<&'a Value> {
+        if let Content::Some { columns, row, .. } = &self.content {
+            let value = columns
                 .iter()
                 .position(|column| column.value == target)
-                .map(|index| self.row.and_then(|row| row.get_value(index)))
-        };
+                .map(|index| row.and_then(|row| row.get_value(index)));
 
-        match get_value() {
-            None => match &self.next {
-                None => Err(FilterContextError::ValueNotFound.into()),
-                Some(context) => context.get_value(target),
+            if let Some(value) = value {
+                return value;
+            }
+        }
+
+        match (&self.next, &self.next2) {
+            (None, None) => None,
+            (Some(fc), None) => fc.get_value(target),
+            (None, Some(bc)) => bc.get_value(target),
+            (Some(fc), Some(bc)) => match bc.get_value(target) {
+                v @ Some(_) => v,
+                None => fc.get_value(target),
             },
-            Some(value) => Ok(value),
         }
     }
 
-    pub fn get_alias_value(&self, table_alias: &str, target: &str) -> Result<Option<&'a Value>> {
-        let get_value = || {
-            if self.table_alias != table_alias {
-                return None;
-            }
+    pub fn get_alias_value(&'a self, target_alias: &str, target: &str) -> Option<&'a Value> {
+        if let Content::Some {
+            table_alias,
+            columns,
+            row,
+        } = &self.content
+        {
+            let get_value = || {
+                if table_alias != &target_alias {
+                    return None;
+                }
 
-            self.columns
-                .iter()
-                .position(|column| column.value == target)
-                .map(|index| self.row.and_then(|row| row.get_value(index)))
-        };
+                columns
+                    .iter()
+                    .position(|column| column.value == target)
+                    .map(|index| row.and_then(|row| row.get_value(index)))
+            };
+
+            if let Some(value) = get_value() {
+                return value;
+            }
+        }
 
-        match get_value() {
-            None => match &self.next {
-                None => Err(FilterContextError::ValueNotFound.into()),
-                Some(context) => context.get_alias_value(table_alias, target),
+        match (&self.next, &self.next2) {
+            (None, None) => None,
+            (Some(fc), None) => fc.get_alias_value(target_alias, target),
+            (None, Some(bc)) => bc.get_alias_value(target_alias, target),
+            (Some(fc), Some(bc)) => match bc.get_alias_value(target_alias, target) {
+                v @ Some(_) => v,
+                None => fc.get_alias_value(target_alias, target),
             },
-            Some(value) => Ok(value),
         }
     }
 }
diff --git a/src/executor/context/mod.rs b/src/executor/context/mod.rs
index 329f96e3..9c2ca6b1 100644
--- a/src/executor/context/mod.rs
+++ b/src/executor/context/mod.rs
@@ -3,5 +3,5 @@ mod blend_context;
 mod filter_context;
 
 pub use aggregate_context::AggregateContext;
-pub use blend_context::{BlendContext, BlendContextError};
-pub use filter_context::{FilterContext, FilterContextError};
+pub use blend_context::BlendContext;
+pub use filter_context::FilterContext;
diff --git a/src/executor/evaluate/error.rs b/src/executor/evaluate/error.rs
index 3e9714f4..6ca748e1 100644
--- a/src/executor/evaluate/error.rs
+++ b/src/executor/evaluate/error.rs
@@ -27,6 +27,9 @@ pub enum EvaluateError {
     )]
     NumberOfFunctionParamsNotMatching { expected: usize, found: usize },
 
+    #[error("value not found: {0}")]
+    ValueNotFound(String),
+
     #[error("unsupported compound identifier {0}")]
     UnsupportedCompoundIdentifier(String),
 
@@ -39,6 +42,9 @@ pub enum EvaluateError {
     #[error("unreachable literal arithmetic")]
     UnreachableLiteralArithmetic,
 
+    #[error("unreachable empty context")]
+    UnreachableEmptyContext,
+
     #[error("unimplemented")]
     Unimplemented,
 }
diff --git a/src/executor/evaluate/evaluated.rs b/src/executor/evaluate/evaluated.rs
index b9f229aa..a93fa34a 100644
--- a/src/executor/evaluate/evaluated.rs
+++ b/src/executor/evaluate/evaluated.rs
@@ -27,6 +27,7 @@ use super::EvaluateError;
 /// * Similar with `Literal`, `Value` is also generated by any other operation with `ValueRef` or
 /// `Value`.
 /// e.g. `LiteralRef` + `ValueRef`, `LiteralRef` * `Value`, ...
+#[derive(std::fmt::Debug)]
 pub enum Evaluated<'a> {
     LiteralRef(&'a AstValue),
     Literal(AstValue),
@@ -83,8 +84,8 @@ impl<'a> PartialEq for Evaluated<'a> {
                     Literal(r) => l == r,
                     LiteralRef(r) => &l == r,
                     ValueRef(r) => r == &l,
+                    Value(r) => r == l,
                     StringRef(_) => false,
-                    Value(_) => panic!(),
                 },
             }
         }
diff --git a/src/executor/evaluate/mod.rs b/src/executor/evaluate/mod.rs
index 2477c8a2..e4c0d6c7 100644
--- a/src/executor/evaluate/mod.rs
+++ b/src/executor/evaluate/mod.rs
@@ -1,6 +1,8 @@
 mod error;
 mod evaluated;
 
+use async_recursion::async_recursion;
+use futures::stream::{StreamExt, TryStreamExt};
 use im_rc::HashMap;
 use std::convert::TryInto;
 use std::fmt::Debug;
@@ -17,16 +19,19 @@ use crate::store::Store;
 pub use error::EvaluateError;
 pub use evaluated::Evaluated;
 
-pub fn evaluate<'a, T: 'static + Debug>(
+#[async_recursion(?Send)]
+pub async fn evaluate<'a, T: 'static + Debug>(
     storage: &'a dyn Store<T>,
     context: Option<Rc<FilterContext<'a>>>,
-    aggregated: Option<&HashMap<&Function, Value>>,
+    aggregated: Option<Rc<HashMap<&'a Function, Value>>>,
     expr: &'a Expr,
+    use_empty: bool,
 ) -> Result<Evaluated<'a>> {
     let eval = |expr| {
         let context = context.as_ref().map(Rc::clone);
+        let aggregated = aggregated.as_ref().map(Rc::clone);
 
-        evaluate(storage, context, aggregated, expr)
+        evaluate(storage, context, aggregated, expr, use_empty)
     };
 
     match expr {
@@ -39,18 +44,20 @@ pub fn evaluate<'a, T: 'static + Debug>(
         },
         Expr::Identifier(ident) => match ident.quote_style {
             Some(_) => Ok(Evaluated::StringRef(&ident.value)),
-            // TODO: remove panic!
-            None => match context {
-                None => {
-                    panic!();
+            None => {
+                let context = context.ok_or(EvaluateError::UnreachableEmptyContext)?;
+
+                match (context.get_value(&ident.value), use_empty) {
+                    (Some(value), _) => Ok(value.clone()),
+                    (None, true) => Ok(Value::Empty),
+                    (None, false) => {
+                        Err(EvaluateError::ValueNotFound(ident.value.to_string()).into())
+                    }
                 }
-                Some(context) => context.get_value(&ident.value).map(|value| match value {
-                    Some(value) => Evaluated::ValueRef(value),
-                    None => Evaluated::Value(Value::Empty),
-                }),
-            },
+                .map(Evaluated::Value)
+            }
         },
-        Expr::Nested(expr) => eval(&expr),
+        Expr::Nested(expr) => eval(&expr).await,
         Expr::CompoundIdentifier(idents) => {
             if idents.len() != 2 {
                 return Err(EvaluateError::UnsupportedCompoundIdentifier(expr.to_string()).into());
@@ -58,30 +65,27 @@ pub fn evaluate<'a, T: 'static + Debug>(
 
             let table_alias = &idents[0].value;
             let column = &idents[1].value;
+            let context = context.ok_or(EvaluateError::UnreachableEmptyContext)?;
 
-            // TODO: remove panic!
-            match context {
-                None => {
-                    panic!();
-                }
-                Some(context) => {
-                    context
-                        .get_alias_value(table_alias, column)
-                        .map(|value| match value {
-                            Some(value) => Evaluated::ValueRef(value),
-                            None => Evaluated::Value(Value::Empty),
-                        })
-                }
+            match (context.get_alias_value(table_alias, column), use_empty) {
+                (Some(value), _) => Ok(value.clone()),
+                (None, true) => Ok(Value::Empty),
+                (None, false) => Err(EvaluateError::ValueNotFound(column.to_string()).into()),
             }
+            .map(Evaluated::Value)
         }
-        Expr::Subquery(query) => select(storage, &query, context.as_ref().map(Rc::clone))?
-            .map(|row| row?.take_first_value())
-            .map(|value| value.map(Evaluated::Value))
+        Expr::Subquery(query) => select(storage, &query, context.as_ref().map(Rc::clone))
+            .await?
+            .map_ok(|row| row.take_first_value().map(Evaluated::Value))
+            .take(1)
+            .collect::<Vec<_>>()
+            .await
+            .into_iter()
             .next()
-            .ok_or(EvaluateError::NestedSelectRowNotFound)?,
+            .unwrap_or_else(|| Err(EvaluateError::NestedSelectRowNotFound.into()))?,
         Expr::BinaryOp { op, left, right } => {
-            let l = eval(left)?;
-            let r = eval(right)?;
+            let l = eval(left).await?;
+            let r = eval(right).await?;
 
             match op {
                 BinaryOperator::Plus => l.add(&r),
@@ -92,7 +96,7 @@ pub fn evaluate<'a, T: 'static + Debug>(
             }
         }
         Expr::UnaryOp { op, expr } => {
-            let v = eval(expr)?;
+            let v = eval(expr).await?;
 
             match op {
                 UnaryOperator::Plus => v.unary_plus(),
@@ -100,33 +104,31 @@ pub fn evaluate<'a, T: 'static + Debug>(
                 _ => Err(EvaluateError::Unimplemented.into()),
             }
         }
+        Expr::Function(func) => match aggregated.as_ref().map(|aggr| aggr.get(func)).flatten() {
+            Some(value) => Ok(Evaluated::Value(value.clone())),
+            None => {
+                let context = context.as_ref().map(Rc::clone);
+                let aggregated = aggregated.as_ref().map(Rc::clone);
 
-        Expr::Function(func) => aggregated
-            .as_ref()
-            .map(|aggr| aggr.get(func))
-            .flatten()
-            .map_or_else(
-                || {
-                    let context = context.as_ref().map(Rc::clone);
-
-                    evaluate_function(storage, context, aggregated, func)
-                },
-                |value| Ok(Evaluated::Value(value.clone())),
-            ),
+                evaluate_function(storage, context, aggregated, func, use_empty).await
+            }
+        },
         _ => Err(EvaluateError::Unimplemented.into()),
     }
 }
 
-fn evaluate_function<'a, T: 'static + Debug>(
+async fn evaluate_function<'a, T: 'static + Debug>(
     storage: &'a dyn Store<T>,
     context: Option<Rc<FilterContext<'a>>>,
-    aggregated: Option<&HashMap<&Function, Value>>,
+    aggregated: Option<Rc<HashMap<&'a Function, Value>>>,
     func: &'a Function,
+    use_empty: bool,
 ) -> Result<Evaluated<'a>> {
     let eval = |expr| {
         let context = context.as_ref().map(Rc::clone);
+        let aggregated = aggregated.as_ref().map(Rc::clone);
 
-        evaluate(storage, context, aggregated, expr)
+        evaluate(storage, context, aggregated, expr, use_empty)
     };
 
     let Function { name, args, .. } = func;
@@ -150,7 +152,7 @@ fn evaluate_function<'a, T: 'static + Debug>(
             }
 
             let expr = &args[0];
-            let value: Value = match eval(expr)?.try_into()? {
+            let value: Value = match eval(expr).await?.try_into()? {
                 Value::Str(s) => Value::Str(convert(s)),
                 Value::OptStr(s) => Value::OptStr(s.map(convert)),
                 _ => {
diff --git a/src/executor/execute.rs b/src/executor/execute.rs
index 4a35a0ef..35fdae75 100644
--- a/src/executor/execute.rs
+++ b/src/executor/execute.rs
@@ -1,14 +1,16 @@
+use futures::stream::{self, TryStreamExt};
+use serde::Serialize;
 use std::fmt::Debug;
+use std::rc::Rc;
+use thiserror::Error as ThisError;
 
-use serde::Serialize;
 #[cfg(feature = "alter-table")]
 use sqlparser::ast::AlterTableOperation;
 use sqlparser::ast::{ObjectType, SetExpr, Statement, Values};
-use thiserror::Error;
 
 use crate::data::{get_name, Row, Schema};
 use crate::parse::Query;
-use crate::result::{MutResult, Result};
+use crate::result::{Error, MutResult, Result};
 use crate::store::{AlterTable, Store, StoreMut};
 
 use super::fetch::{fetch, fetch_columns};
@@ -16,7 +18,7 @@ use super::filter::Filter;
 use super::select::select_with_labels;
 use super::update::Update;
 
-#[derive(Error, Serialize, Debug, PartialEq)]
+#[derive(ThisError, Serialize, Debug, PartialEq)]
 pub enum ExecuteError {
     #[error("query not supported")]
     QueryNotSupported,
@@ -53,7 +55,7 @@ pub enum Payload {
     AlterTable,
 }
 
-pub fn execute<T: 'static + Debug, U: Store<T> + StoreMut<T> + AlterTable>(
+pub async fn execute<T: 'static + Debug, U: Store<T> + StoreMut<T> + AlterTable>(
     storage: U,
     query: &Query,
 ) -> MutResult<U, Payload> {
@@ -69,14 +71,15 @@ pub fn execute<T: 'static + Debug, U: Store<T> + StoreMut<T> + AlterTable>(
     }
 
     let Query(query) = query;
-    let prepared = try_into!(storage, prepare(&storage, query));
+    let prepared = prepare(&storage, query).await;
+    let prepared = try_into!(storage, prepared);
 
     match prepared {
         Prepared::Create {
             schema,
             if_not_exists,
         } => {
-            if try_into!(storage, storage.fetch_schema(&schema.table_name)).is_some() {
+            if try_into!(storage, storage.fetch_schema(&schema.table_name).await).is_some() {
                 return if if_not_exists {
                     Ok((storage, Payload::Create))
                 } else {
@@ -86,48 +89,42 @@ pub fn execute<T: 'static + Debug, U: Store<T> + StoreMut<T> + AlterTable>(
 
             storage
                 .insert_schema(&schema)
+                .await
                 .map(|(storage, _)| (storage, Payload::Create))
         }
-        Prepared::Insert(table_name, rows) => rows
-            .into_iter()
-            .try_fold((storage, 0), |(storage, num), row| {
-                let (storage, key) = storage.generate_id(&table_name)?;
-                let (storage, _) = storage.insert_data(&key, row)?;
+        Prepared::Insert(table_name, rows) => {
+            stream::iter(rows.into_iter().map(Ok::<Row, (U, Error)>))
+                .try_fold((storage, 0), |(storage, num), row| async move {
+                    let (storage, key) = storage.generate_id(&table_name).await?;
+                    let (storage, _) = storage.insert_data(&key, row).await?;
+
+                    Ok((storage, num + 1))
+                })
+                .await
+                .map(|(storage, num_rows)| (storage, Payload::Insert(num_rows)))
+        }
+        Prepared::Delete(keys) => stream::iter(keys.into_iter().map(Ok::<T, (U, Error)>))
+            .try_fold((storage, 0), |(storage, num), key| async move {
+                let (storage, _) = storage.delete_data(&key).await?;
 
                 Ok((storage, num + 1))
             })
-            .map(|(storage, num_rows)| (storage, Payload::Insert(num_rows))),
-        Prepared::Delete(keys) => {
-            let (storage, num_rows) =
-                keys.into_iter()
-                    .try_fold((storage, 0), |(storage, num), key| {
-                        let (storage, _) = storage.delete_data(&key)?;
+            .await
+            .map(|(storage, num_rows)| (storage, Payload::Delete(num_rows))),
+        Prepared::Update(items) => stream::iter(items.into_iter().map(Ok::<_, (U, Error)>))
+            .try_fold((storage, 0), |(storage, num), (key, row)| async move {
+                let (storage, _) = storage.insert_data(&key, row).await?;
 
-                        Ok((storage, num + 1))
-                    })?;
-
-            Ok((storage, Payload::Delete(num_rows)))
-        }
-        Prepared::Update(items) => {
-            let (storage, num_rows) =
-                items
-                    .into_iter()
-                    .try_fold((storage, 0), |(storage, num), item| {
-                        let (key, row) = item;
-                        let (storage, _) = storage.insert_data(&key, row)?;
-
-                        Ok((storage, num + 1))
-                    })?;
-
-            Ok((storage, Payload::Update(num_rows)))
-        }
+                Ok((storage, num + 1))
+            })
+            .await
+            .map(|(storage, num_rows)| (storage, Payload::Update(num_rows))),
         Prepared::DropTable {
             schema_names,
             if_exists,
-        } => schema_names
-            .iter()
-            .try_fold(storage, |storage, table_name| {
-                let schema = try_into!(storage, storage.fetch_schema(table_name));
+        } => stream::iter(schema_names.into_iter().map(Ok::<&str, (U, Error)>))
+            .try_fold(storage, |storage, table_name| async move {
+                let schema = try_into!(storage, storage.fetch_schema(table_name).await);
 
                 if !if_exists {
                     try_into!(storage, schema.ok_or(ExecuteError::TableNotExists));
@@ -135,8 +132,10 @@ pub fn execute<T: 'static + Debug, U: Store<T> + StoreMut<T> + AlterTable>(
 
                 storage
                     .delete_schema(table_name)
+                    .await
                     .map(|(storage, _)| storage)
             })
+            .await
             .map(|storage| (storage, Payload::DropTable)),
         Prepared::Select { labels, rows } => Ok((storage, Payload::Select { labels, rows })),
 
@@ -145,23 +144,31 @@ pub fn execute<T: 'static + Debug, U: Store<T> + StoreMut<T> + AlterTable>(
             let result = match operation {
                 AlterTableOperation::RenameTable {
                     table_name: new_table_name,
-                } => storage.rename_schema(table_name, &new_table_name.value),
+                } => {
+                    storage
+                        .rename_schema(table_name, &new_table_name.value)
+                        .await
+                }
                 AlterTableOperation::RenameColumn {
                     old_column_name,
                     new_column_name,
-                } => storage.rename_column(
-                    table_name,
-                    &old_column_name.value,
-                    &new_column_name.value,
-                ),
+                } => {
+                    storage
+                        .rename_column(table_name, &old_column_name.value, &new_column_name.value)
+                        .await
+                }
                 AlterTableOperation::AddColumn { column_def } => {
-                    storage.add_column(table_name, column_def)
+                    storage.add_column(table_name, column_def).await
                 }
                 AlterTableOperation::DropColumn {
                     column_name,
                     if_exists,
                     ..
-                } => storage.drop_column(table_name, &column_name.value, *if_exists),
+                } => {
+                    storage
+                        .drop_column(table_name, &column_name.value, *if_exists)
+                        .await
+                }
                 _ => Err((
                     storage,
                     ExecuteError::UnsupportedAlterTableOperation(operation.to_string()).into(),
@@ -194,7 +201,9 @@ enum Prepared<'a, T> {
     AlterTable(&'a str, &'a AlterTableOperation),
 }
 
-fn prepare<'a, T: 'static + Debug>(
+// looks like... false positive
+#[allow(clippy::needless_lifetimes)]
+async fn prepare<'a, T: 'static + Debug>(
     storage: &impl Store<T>,
     sql_query: &'a Statement,
 ) -> Result<Prepared<'a, T>> {
@@ -216,8 +225,8 @@ fn prepare<'a, T: 'static + Debug>(
             })
         }
         Statement::Query(query) => {
-            let (labels, rows) = select_with_labels(storage, &query, None, true)?;
-            let rows = rows.collect::<Result<_>>()?;
+            let (labels, rows) = select_with_labels(storage, &query, None, true).await?;
+            let rows = rows.try_collect::<Vec<_>>().await?;
 
             Ok(Prepared::Select { labels, rows })
         }
@@ -228,7 +237,8 @@ fn prepare<'a, T: 'static + Debug>(
         } => {
             let table_name = get_name(table_name)?;
             let Schema { column_defs, .. } = storage
-                .fetch_schema(table_name)?
+                .fetch_schema(table_name)
+                .await?
                 .ok_or(ExecuteError::TableNotExists)?;
 
             let values_list = match &source.body {
@@ -253,33 +263,40 @@ fn prepare<'a, T: 'static + Debug>(
             assignments,
         } => {
             let table_name = get_name(table_name)?;
-            let columns = fetch_columns(storage, table_name)?;
-            let update = Update::new(storage, table_name, assignments, &columns)?;
+            let columns = Rc::new(fetch_columns(storage, table_name).await?);
+            let update = Update::new(storage, table_name, assignments, Rc::clone(&columns))?;
             let filter = Filter::new(storage, selection.as_ref(), None, None);
 
-            let rows = fetch(storage, table_name, &columns, filter)?
-                .map(|item| {
-                    let (_, key, row) = item?;
+            fetch(storage, table_name, columns, filter)
+                .await?
+                .and_then(|item| {
+                    let update = &update;
+                    let (_, key, row) = item;
 
-                    Ok((key, update.apply(row)?))
-                })
-                .collect::<Result<_>>()?;
+                    async move {
+                        let row = update.apply(row).await?;
 
-            Ok(Prepared::Update(rows))
+                        Ok((key, row))
+                    }
+                })
+                .try_collect::<Vec<_>>()
+                .await
+                .map(Prepared::Update)
         }
         Statement::Delete {
             table_name,
             selection,
         } => {
             let table_name = get_name(table_name)?;
-            let columns = fetch_columns(storage, table_name)?;
+            let columns = Rc::new(fetch_columns(storage, table_name).await?);
             let filter = Filter::new(storage, selection.as_ref(), None, None);
 
-            let rows = fetch(storage, table_name, &columns, filter)?
-                .map(|item| item.map(|(_, key, _)| key))
-                .collect::<Result<_>>()?;
-
-            Ok(Prepared::Delete(rows))
+            fetch(storage, table_name, columns, filter)
+                .await?
+                .map_ok(|(_, key, _)| key)
+                .try_collect::<Vec<_>>()
+                .await
+                .map(Prepared::Delete)
         }
         Statement::Drop {
             object_type,
diff --git a/src/executor/fetch.rs b/src/executor/fetch.rs
index b8c019fb..8d520cb2 100644
--- a/src/executor/fetch.rs
+++ b/src/executor/fetch.rs
@@ -1,27 +1,31 @@
 use boolinator::Boolinator;
+use futures::stream::{self, TryStream, TryStreamExt};
 use serde::Serialize;
 use std::fmt::Debug;
-use thiserror::Error;
+use std::rc::Rc;
+use thiserror::Error as ThisError;
 
 use sqlparser::ast::{ColumnDef, Ident};
 
+use super::context::BlendContext;
 use super::filter::Filter;
 use crate::data::Row;
-use crate::result::Result;
+use crate::result::{Error, Result};
 use crate::store::Store;
 
-#[derive(Error, Serialize, Debug, PartialEq)]
+#[derive(ThisError, Serialize, Debug, PartialEq)]
 pub enum FetchError {
     #[error("table not found: {0}")]
     TableNotFound(String),
 }
 
-pub fn fetch_columns<T: 'static + Debug>(
+pub async fn fetch_columns<T: 'static + Debug>(
     storage: &dyn Store<T>,
     table_name: &str,
 ) -> Result<Vec<Ident>> {
     Ok(storage
-        .fetch_schema(table_name)?
+        .fetch_schema(table_name)
+        .await?
         .ok_or_else(|| FetchError::TableNotFound(table_name.to_string()))?
         .column_defs
         .into_iter()
@@ -29,23 +33,33 @@ pub fn fetch_columns<T: 'static + Debug>(
         .collect::<Vec<Ident>>())
 }
 
-pub fn fetch<'a, T: 'static + Debug>(
+pub async fn fetch<'a, T: 'static + Debug>(
     storage: &dyn Store<T>,
     table_name: &'a str,
-    columns: &'a [Ident],
+    columns: Rc<Vec<Ident>>,
     filter: Filter<'a, T>,
-) -> Result<impl Iterator<Item = Result<(&'a [Ident], T, Row)>> + 'a> {
-    let rows = storage.scan_data(table_name)?.filter_map(move |item| {
-        item.map_or_else(
-            |error| Some(Err(error)),
-            |(key, row)| {
-                filter
-                    .check(&table_name, columns, &row)
-                    .map(|pass| pass.as_some((columns, key, row)))
-                    .transpose()
-            },
-        )
-    });
+) -> Result<impl TryStream<Ok = (Rc<Vec<Ident>>, T, Row), Error = Error> + 'a> {
+    let filter = Rc::new(filter);
+
+    let rows = storage
+        .scan_data(table_name)
+        .await
+        .map(stream::iter)?
+        .try_filter_map(move |(key, row)| {
+            let columns = Rc::clone(&columns);
+            let filter = Rc::clone(&filter);
+
+            let context = Rc::new(BlendContext::new(table_name, columns, Some(row), None));
+
+            // TODO: remove two unwrap() uses.
+            async move {
+                filter.check(Rc::clone(&context)).await.map(|pass| {
+                    let context = Rc::try_unwrap(context).unwrap();
+
+                    pass.as_some((context.columns, key, context.row.unwrap()))
+                })
+            }
+        });
 
     Ok(rows)
 }
diff --git a/src/executor/filter.rs b/src/executor/filter.rs
index c92d0d76..65a92ae1 100644
--- a/src/executor/filter.rs
+++ b/src/executor/filter.rs
@@ -1,16 +1,18 @@
+use async_recursion::async_recursion;
 use boolinator::Boolinator;
+use futures::stream::{self, StreamExt, TryStreamExt};
 use im_rc::HashMap;
 use serde::Serialize;
 use std::fmt::Debug;
 use std::rc::Rc;
 use thiserror::Error;
 
-use sqlparser::ast::{BinaryOperator, Expr, Function, Ident, UnaryOperator};
+use sqlparser::ast::{BinaryOperator, Expr, Function, UnaryOperator};
 
 use super::context::{BlendContext, FilterContext};
 use super::evaluate::{evaluate, Evaluated};
 use super::select::select;
-use crate::data::{Row, Value};
+use crate::data::Value;
 use crate::result::Result;
 use crate::store::Store;
 
@@ -24,7 +26,7 @@ pub struct Filter<'a, T: 'static + Debug> {
     storage: &'a dyn Store<T>,
     where_clause: Option<&'a Expr>,
     context: Option<Rc<FilterContext<'a>>>,
-    aggregated: Option<&'a HashMap<&'a Function, Value>>,
+    aggregated: Option<Rc<HashMap<&'a Function, Value>>>,
 }
 
 impl<'a, T: 'static + Debug> Filter<'a, T> {
@@ -32,7 +34,7 @@ impl<'a, T: 'static + Debug> Filter<'a, T> {
         storage: &'a dyn Store<T>,
         where_clause: Option<&'a Expr>,
         context: Option<Rc<FilterContext<'a>>>,
-        aggregated: Option<&'a HashMap<&'a Function, Value>>,
+        aggregated: Option<Rc<HashMap<&'a Function, Value>>>,
     ) -> Self {
         Self {
             storage,
@@ -42,84 +44,96 @@ impl<'a, T: 'static + Debug> Filter<'a, T> {
         }
     }
 
-    pub fn check(&self, table_alias: &str, columns: &[Ident], row: &Row) -> Result<bool> {
-        let next = self.context.as_ref().map(Rc::clone);
-        let context = FilterContext::new(table_alias, columns, Some(row), next);
-        let context = Some(context).map(Rc::new);
-
-        match self.where_clause {
-            Some(expr) => check_expr(self.storage, context, self.aggregated, expr),
-            None => Ok(true),
-        }
-    }
-
-    pub fn check_blended(&self, blend_context: &BlendContext<'_>) -> Result<bool> {
+    pub async fn check(&self, blend_context: Rc<BlendContext<'a>>) -> Result<bool> {
         match self.where_clause {
             Some(expr) => {
                 let context = self.context.as_ref().map(Rc::clone);
-                let context = blend_context.concat_into(context);
+                let context = FilterContext::concat(context, Some(blend_context));
+                let context = Some(context).map(Rc::new);
+                let aggregated = self.aggregated.as_ref().map(Rc::clone);
 
-                check_expr(self.storage, context, self.aggregated, expr)
+                check_expr(self.storage, context, aggregated, expr).await
             }
             None => Ok(true),
         }
     }
 }
 
-pub fn check_expr<T: 'static + Debug>(
+#[async_recursion(?Send)]
+pub async fn check_expr<T: 'static + Debug>(
     storage: &dyn Store<T>,
-    filter_context: Option<Rc<FilterContext<'_>>>,
-    aggregated: Option<&HashMap<&Function, Value>>,
+    filter_context: Option<Rc<FilterContext<'async_recursion>>>,
+    aggregated: Option<Rc<HashMap<&'async_recursion Function, Value>>>,
     expr: &Expr,
 ) -> Result<bool> {
-    let evaluate = |expr| {
+    let evaluate = |expr: &'async_recursion Expr| {
         let filter_context = filter_context.as_ref().map(Rc::clone);
+        let aggregated = aggregated.as_ref().map(Rc::clone);
 
-        evaluate(storage, filter_context, aggregated, expr)
+        evaluate(storage, filter_context, aggregated, expr, false)
     };
     let check = |expr| {
         let filter_context = filter_context.as_ref().map(Rc::clone);
+        let aggregated = aggregated.as_ref().map(Rc::clone);
 
         check_expr(storage, filter_context, aggregated, expr)
     };
 
     match expr {
         Expr::BinaryOp { op, left, right } => {
-            let zip_evaluate = || Ok((evaluate(left)?, evaluate(right)?));
-            let zip_check = || Ok((check(left)?, check(right)?));
+            let zip_evaluate = || async move {
+                let l = evaluate(left).await?;
+                let r = evaluate(right).await?;
+
+                Ok((l, r))
+            };
+            let zip_check = || async move {
+                let l = check(left).await?;
+                let r = check(right).await?;
+
+                Ok((l, r))
+            };
 
             match op {
-                BinaryOperator::Eq => zip_evaluate().map(|(l, r)| l == r),
-                BinaryOperator::NotEq => zip_evaluate().map(|(l, r)| l != r),
-                BinaryOperator::And => zip_check().map(|(l, r)| l && r),
-                BinaryOperator::Or => zip_check().map(|(l, r)| l || r),
-                BinaryOperator::Lt => zip_evaluate().map(|(l, r)| l < r),
-                BinaryOperator::LtEq => zip_evaluate().map(|(l, r)| l <= r),
-                BinaryOperator::Gt => zip_evaluate().map(|(l, r)| l > r),
-                BinaryOperator::GtEq => zip_evaluate().map(|(l, r)| l >= r),
+                BinaryOperator::Eq => zip_evaluate().await.map(|(l, r)| l == r),
+                BinaryOperator::NotEq => zip_evaluate().await.map(|(l, r)| l != r),
+                BinaryOperator::And => zip_check().await.map(|(l, r)| l && r),
+                BinaryOperator::Or => zip_check().await.map(|(l, r)| l || r),
+                BinaryOperator::Lt => zip_evaluate().await.map(|(l, r)| l < r),
+                BinaryOperator::LtEq => zip_evaluate().await.map(|(l, r)| l <= r),
+                BinaryOperator::Gt => zip_evaluate().await.map(|(l, r)| l > r),
+                BinaryOperator::GtEq => zip_evaluate().await.map(|(l, r)| l >= r),
                 _ => Err(FilterError::Unimplemented.into()),
             }
         }
         Expr::UnaryOp { op, expr } => match op {
-            UnaryOperator::Not => check(&expr).map(|v| !v),
+            UnaryOperator::Not => check(&expr).await.map(|v| !v),
             _ => Err(FilterError::Unimplemented.into()),
         },
-        Expr::Nested(expr) => check(&expr),
+        Expr::Nested(expr) => check(&expr).await,
         Expr::InList {
             expr,
             list,
             negated,
         } => {
             let negated = *negated;
-            let target = evaluate(expr)?;
+            let target = evaluate(expr).await?;
 
-            list.iter()
+            stream::iter(list.iter())
                 .filter_map(|expr| {
-                    evaluate(expr).map_or_else(
-                        |error| Some(Err(error)),
-                        |evaluated| (target == evaluated).as_some(Ok(!negated)),
-                    )
+                    let target = &target;
+
+                    async move {
+                        evaluate(expr).await.map_or_else(
+                            |error| Some(Err(error)),
+                            |evaluated| (target == &evaluated).as_some(Ok(!negated)),
+                        )
+                    }
                 })
+                .take(1)
+                .collect::<Vec<_>>()
+                .await
+                .into_iter()
                 .next()
                 .unwrap_or(Ok(negated))
         }
@@ -128,19 +142,27 @@ pub fn check_expr<T: 'static + Debug>(
             subquery,
             negated,
         } => {
-            let negated = *negated;
-            let target = evaluate(expr)?;
-
-            select(storage, &subquery, filter_context)?
-                .map(|row| row?.take_first_value())
-                .filter_map(|value| {
-                    value.map_or_else(
-                        |error| Some(Err(error)),
-                        |value| (target == Evaluated::ValueRef(&value)).as_some(Ok(!negated)),
-                    )
+            let target = evaluate(expr).await?;
+
+            select(storage, &subquery, filter_context)
+                .await?
+                .try_filter_map(|row| {
+                    let target = &target;
+
+                    async move {
+                        let value = row.take_first_value()?;
+
+                        (target == &Evaluated::ValueRef(&value))
+                            .as_some(Ok(!negated))
+                            .transpose()
+                    }
                 })
+                .take(1)
+                .collect::<Vec<_>>()
+                .await
+                .into_iter()
                 .next()
-                .unwrap_or(Ok(negated))
+                .unwrap_or(Ok(*negated))
         }
         Expr::Between {
             expr,
@@ -149,13 +171,20 @@ pub fn check_expr<T: 'static + Debug>(
             high,
         } => {
             let negated = *negated;
-            let target = evaluate(expr)?;
+            let target = evaluate(expr).await?;
 
-            Ok(negated ^ (evaluate(low)? <= target && target <= evaluate(high)?))
+            Ok(negated ^ (evaluate(low).await? <= target && target <= evaluate(high).await?))
         }
-        Expr::Exists(query) => Ok(select(storage, query, filter_context)?.next().is_some()),
-        Expr::IsNull(expr) => Ok(!evaluate(expr)?.is_some()),
-        Expr::IsNotNull(expr) => Ok(evaluate(expr)?.is_some()),
+        Expr::Exists(query) => Ok(select(storage, query, filter_context)
+            .await?
+            .into_stream()
+            .take(1)
+            .try_collect::<Vec<_>>()
+            .await?
+            .get(0)
+            .is_some()),
+        Expr::IsNull(expr) => Ok(!evaluate(expr).await?.is_some()),
+        Expr::IsNotNull(expr) => Ok(evaluate(expr).await?.is_some()),
         _ => Err(FilterError::Unimplemented.into()),
     }
 }
diff --git a/src/executor/join.rs b/src/executor/join.rs
index 6623fa26..ccd57c8c 100644
--- a/src/executor/join.rs
+++ b/src/executor/join.rs
@@ -1,9 +1,8 @@
 use boolinator::Boolinator;
-use iter_enum::Iterator;
-use or_iterator::OrIterator;
+use futures::stream::{self, once, StreamExt, TryStream, TryStreamExt};
 use serde::Serialize;
 use std::fmt::Debug;
-use std::iter::once;
+use std::pin::Pin;
 use std::rc::Rc;
 use thiserror::Error as ThisError;
 
@@ -12,8 +11,9 @@ use sqlparser::ast::{Ident, Join as AstJoin, JoinConstraint, JoinOperator};
 use super::context::{BlendContext, FilterContext};
 use super::filter::Filter;
 use crate::data::Table;
-use crate::result::Result;
+use crate::result::{Error, Result};
 use crate::store::Store;
+use crate::utils::OrStream;
 
 #[derive(ThisError, Serialize, Debug, PartialEq)]
 pub enum JoinError {
@@ -36,15 +36,9 @@ pub struct Join<'a, T: 'static + Debug> {
     filter_context: Option<Rc<FilterContext<'a>>>,
 }
 
-type JoinItem<'a> = Result<Rc<BlendContext<'a>>>;
-
-#[derive(Iterator)]
-enum Applied<I1, I2, I3, I4> {
-    Init(I1),
-    Once(I2),
-    Boxed(I3),
-    Mapped(I4),
-}
+type JoinItem<'a> = Rc<BlendContext<'a>>;
+type Joined<'a> =
+    Pin<Box<dyn TryStream<Ok = JoinItem<'a>, Error = Error, Item = Result<JoinItem<'a>>> + 'a>>;
 
 impl<'a, T: 'static + Debug> Join<'a, T> {
     pub fn new(
@@ -59,143 +53,148 @@ impl<'a, T: 'static + Debug> Join<'a, T> {
         }
     }
 
-    pub fn apply(
+    pub async fn apply(
         &self,
         init_context: Result<BlendContext<'a>>,
         join_columns: Rc<Vec<Rc<Vec<Ident>>>>,
-    ) -> impl Iterator<Item = JoinItem<'a>> + 'a {
+    ) -> Result<Joined<'a>> {
         let init_context = init_context.map(Rc::new);
-        let init_rows = Applied::Init(once(init_context));
+        let init_rows: Joined<'a> = Box::pin(stream::once(async { init_context }));
         let filter_context = self.filter_context.as_ref().map(Rc::clone);
 
-        self.join_clauses
+        let joins = self
+            .join_clauses
             .iter()
             .enumerate()
             .map(move |(i, join_clause)| {
                 let join_columns = Rc::clone(&join_columns[i]);
 
-                (join_clause, join_columns)
-            })
-            .fold(init_rows, |rows, (join_clause, join_columns)| {
+                Ok::<_, Error>((join_clause, join_columns))
+            });
+
+        let rows = stream::iter(joins)
+            .try_fold(init_rows, |rows, (join_clause, join_columns)| {
                 let storage = self.storage;
                 let filter_context = filter_context.as_ref().map(Rc::clone);
-                let map = move |blend_context| {
-                    let columns = Rc::clone(&join_columns);
-
-                    let filter_context = filter_context.as_ref().map(Rc::clone);
-                    join(storage, filter_context, join_clause, columns, blend_context)
-                };
-
-                match rows {
-                    Applied::Init(rows) => Applied::Once(rows.flat_map(map)),
-                    Applied::Once(rows) => {
-                        let rows: Box<dyn Iterator<Item = _>> = Box::new(rows.flat_map(map));
-
-                        Applied::Boxed(rows)
-                    }
-                    Applied::Boxed(rows) => Applied::Mapped(rows.flat_map(map)),
-                    Applied::Mapped(rows) => Applied::Boxed(Box::new(rows.flat_map(map))),
+
+                async move {
+                    let rows = rows
+                        .map(Ok)
+                        .and_then(move |blend_context| {
+                            let columns = Rc::clone(&join_columns);
+                            let filter_context = filter_context.as_ref().map(Rc::clone);
+
+                            join(storage, filter_context, join_clause, columns, blend_context)
+                        })
+                        .try_flatten();
+
+                    Ok::<Joined<'a>, _>(Box::pin(rows))
                 }
             })
-    }
-}
+            .await?;
 
-#[derive(Iterator)]
-enum Joined<I1, I2, I3> {
-    Err(I1),
-    Inner(I2),
-    LeftOuter(I3),
+        Ok(Box::pin(rows))
+    }
 }
 
-fn join<'a, T: 'static + Debug>(
+async fn join<'a, T: 'static + Debug>(
     storage: &'a dyn Store<T>,
     filter_context: Option<Rc<FilterContext<'a>>>,
     ast_join: &'a AstJoin,
     columns: Rc<Vec<Ident>>,
     blend_context: Result<Rc<BlendContext<'a>>>,
-) -> impl Iterator<Item = JoinItem<'a>> + 'a {
-    let err = |e| Joined::Err(once(Err(e)));
-
-    macro_rules! try_into {
-        ($v: expr) => {
-            match $v {
-                Ok(v) => v,
-                Err(e) => {
-                    return err(e);
-                }
-            }
-        };
-    }
-
+) -> Result<Joined<'a>> {
     let AstJoin {
         relation,
         join_operator,
     } = ast_join;
-    let table = try_into!(Table::new(relation));
+    let table = Table::new(relation)?;
     let table_name = table.get_name();
     let table_alias = table.get_alias();
 
-    let blend_context = try_into!(blend_context);
+    let blend_context = blend_context?;
     let init_context = Rc::new(BlendContext::new(
-        table_alias,
+        table.get_alias(),
         Rc::clone(&columns),
         None,
         Some(Rc::clone(&blend_context)),
     ));
 
-    let fetch_rows = |constraint: &'a JoinConstraint| {
-        let where_clause = match constraint {
-            JoinConstraint::On(where_clause) => Some(where_clause),
-            JoinConstraint::Using(_) => {
-                return Err(JoinError::UsingOnJoinNotSupported.into());
-            }
-            JoinConstraint::Natural => {
-                return Err(JoinError::NaturalOnJoinNotSupported.into());
-            }
-        };
-
-        let rows = storage.scan_data(table_name)?;
-        let rows = rows.filter_map(move |item| {
-            let (_, row) = match item {
-                Ok(v) => v,
-                Err(e) => {
-                    return Some(Err(e));
-                }
-            };
-
-            let filter_context = filter_context.as_ref().map(Rc::clone);
-            let filter_context = blend_context.concat_into(filter_context);
-            let filter = Filter::new(storage, where_clause, filter_context, None);
-
-            filter
-                .check(table_alias, &columns, &row)
-                .map(|pass| {
-                    pass.as_some(Rc::new(BlendContext::new(
-                        table_alias,
-                        Rc::clone(&columns),
-                        Some(row),
-                        Some(Rc::clone(&blend_context)),
-                    )))
-                })
-                .transpose()
-        });
-
-        Ok(rows)
+    let fetch_joined = |constraint: &'a JoinConstraint| {
+        fetch_joined(
+            storage,
+            table_name,
+            table_alias,
+            Rc::clone(&columns),
+            filter_context,
+            blend_context,
+            constraint,
+        )
     };
 
     match join_operator {
-        JoinOperator::Inner(constraint) => match fetch_rows(constraint) {
-            Ok(rows) => Joined::Inner(rows),
-            Err(e) => err(e),
-        },
-        JoinOperator::LeftOuter(constraint) => match fetch_rows(constraint) {
-            Ok(rows) => {
-                let rows = rows.or(once(Ok(init_context)));
-
-                Joined::LeftOuter(rows)
-            }
-            Err(e) => err(e),
-        },
-        _ => err(JoinError::JoinTypeNotSupported.into()),
+        JoinOperator::Inner(constraint) => fetch_joined(constraint).await.map(|rows| {
+            let rows: Joined<'a> = Box::pin(rows);
+
+            rows
+        }),
+        JoinOperator::LeftOuter(constraint) => fetch_joined(constraint).await.map(|rows| {
+            let init_rows = once(async { Ok(init_context) });
+            let rows: Joined<'a> = Box::pin(OrStream::new(rows, init_rows));
+
+            rows
+        }),
+        _ => Err(JoinError::JoinTypeNotSupported.into()),
     }
 }
+
+async fn fetch_joined<'a, T: 'static + Debug>(
+    storage: &'a dyn Store<T>,
+    table_name: &'a str,
+    table_alias: &'a str,
+    columns: Rc<Vec<Ident>>,
+    filter_context: Option<Rc<FilterContext<'a>>>,
+    blend_context: Rc<BlendContext<'a>>,
+    constraint: &'a JoinConstraint,
+) -> Result<impl TryStream<Ok = JoinItem<'a>, Error = Error, Item = Result<JoinItem<'a>>> + 'a> {
+    let where_clause = match constraint {
+        JoinConstraint::On(where_clause) => Some(where_clause),
+        JoinConstraint::Using(_) => {
+            return Err(JoinError::UsingOnJoinNotSupported.into());
+        }
+        JoinConstraint::Natural => {
+            return Err(JoinError::NaturalOnJoinNotSupported.into());
+        }
+    };
+
+    let rows = storage
+        .scan_data(table_name)
+        .await
+        .map(stream::iter)?
+        .try_filter_map(move |(_, row)| {
+            let filter_context = FilterContext::concat(
+                filter_context.as_ref().map(Rc::clone),
+                Some(&blend_context).map(Rc::clone),
+            );
+            let filter_context = Some(filter_context).map(Rc::new);
+            let blend_context = Rc::clone(&blend_context);
+            let columns = Rc::clone(&columns);
+
+            async move {
+                let filter = Filter::new(storage, where_clause, filter_context, None);
+                let context = Rc::new(BlendContext::new(
+                    table_alias,
+                    Rc::clone(&columns),
+                    Some(row),
+                    Some(&blend_context).map(Rc::clone),
+                ));
+
+                filter
+                    .check(Rc::clone(&context))
+                    .await
+                    .map(|pass| pass.as_some(context))
+            }
+        });
+
+    Ok(rows)
+}
diff --git a/src/executor/mod.rs b/src/executor/mod.rs
index 0d9844b3..778a704a 100644
--- a/src/executor/mod.rs
+++ b/src/executor/mod.rs
@@ -12,7 +12,6 @@ mod update;
 
 pub use aggregate::{AggregateError, GroupKey};
 pub use blend::BlendError;
-pub use context::{BlendContextError, FilterContextError};
 pub use evaluate::EvaluateError;
 pub use execute::{execute, ExecuteError, Payload};
 pub use fetch::FetchError;
diff --git a/src/executor/select.rs b/src/executor/select.rs
index 274d673a..db1cc7df 100644
--- a/src/executor/select.rs
+++ b/src/executor/select.rs
@@ -1,10 +1,11 @@
 use boolinator::Boolinator;
+use futures::stream::{self, Stream, StreamExt, TryStream, TryStreamExt};
 use iter_enum::Iterator;
 use serde::Serialize;
 use std::fmt::Debug;
 use std::iter::once;
 use std::rc::Rc;
-use thiserror::Error;
+use thiserror::Error as ThisError;
 
 use sqlparser::ast::{Expr, Ident, Query, SelectItem, SetExpr, TableWithJoins};
 
@@ -16,10 +17,10 @@ use super::filter::Filter;
 use super::join::Join;
 use super::limit::Limit;
 use crate::data::{get_name, Row, Table};
-use crate::result::Result;
+use crate::result::{Error, Result};
 use crate::store::Store;
 
-#[derive(Error, Serialize, Debug, PartialEq)]
+#[derive(ThisError, Serialize, Debug, PartialEq)]
 pub enum SelectError {
     #[error("unimplemented! select on two or more than tables are not supported")]
     TooManyTables,
@@ -31,12 +32,12 @@ pub enum SelectError {
     Unreachable,
 }
 
-fn fetch_blended<'a, T: 'static + Debug>(
+async fn fetch_blended<'a, T: 'static + Debug>(
     storage: &dyn Store<T>,
     table: Table<'a>,
     columns: Rc<Vec<Ident>>,
-) -> Result<impl Iterator<Item = Result<BlendContext<'a>>> + 'a> {
-    let rows = storage.scan_data(table.get_name())?.map(move |data| {
+) -> Result<impl Stream<Item = Result<BlendContext<'a>>> + 'a> {
+    let rows = storage.scan_data(table.get_name()).await?.map(move |data| {
         let (_, row) = data?;
         let row = Some(row);
         let columns = Rc::clone(&columns);
@@ -44,7 +45,7 @@ fn fetch_blended<'a, T: 'static + Debug>(
         Ok(BlendContext::new(table.get_alias(), columns, row, None))
     });
 
-    Ok(rows)
+    Ok(stream::iter(rows))
 }
 
 fn get_labels<'a>(
@@ -125,12 +126,12 @@ fn get_labels<'a>(
         .collect::<Result<_>>()
 }
 
-pub fn select_with_labels<'a, T: 'static + Debug>(
+pub async fn select_with_labels<'a, T: 'static + Debug>(
     storage: &'a dyn Store<T>,
     query: &'a Query,
     filter_context: Option<Rc<FilterContext<'a>>>,
     with_labels: bool,
-) -> Result<(Vec<String>, impl Iterator<Item = Result<Row>> + 'a)> {
+) -> Result<(Vec<String>, impl TryStream<Ok = Row, Error = Error> + 'a)> {
     macro_rules! err {
         ($err: expr) => {{
             return Err($err.into());
@@ -160,18 +161,24 @@ pub fn select_with_labels<'a, T: 'static + Debug>(
     let TableWithJoins { relation, joins } = &table_with_joins;
     let table = Table::new(relation)?;
 
-    let columns = fetch_columns(storage, table.get_name())?;
-    let join_columns = joins
-        .iter()
-        .map(|join| {
-            let table = Table::new(&join.relation)?;
-            let table_alias = table.get_alias();
-            let table_name = table.get_name();
-            let columns = fetch_columns(storage, table_name)?;
+    let columns = fetch_columns(storage, table.get_name()).await?;
+    let join_columns = stream::iter(joins.iter())
+        .map(Ok::<_, Error>)
+        .and_then(|join| {
+            let table = Table::new(&join.relation);
+
+            async move {
+                let table = table?;
+                let table_alias = table.get_alias();
+                let table_name = table.get_name();
+
+                let columns = fetch_columns(storage, table_name).await?;
 
-            Ok((table_alias, columns))
+                Ok((table_alias, columns))
+            }
         })
-        .collect::<Result<Vec<_>>>()?;
+        .try_collect::<Vec<_>>()
+        .await?;
 
     let labels = if with_labels {
         get_labels(&projection, table.get_alias(), &columns, &join_columns)?
@@ -187,7 +194,11 @@ pub fn select_with_labels<'a, T: 'static + Debug>(
         .collect::<Vec<_>>();
     let join_columns = Rc::new(join_columns);
 
-    let join = Join::new(storage, joins, filter_context.as_ref().map(Rc::clone));
+    let join = Rc::new(Join::new(
+        storage,
+        joins,
+        filter_context.as_ref().map(Rc::clone),
+    ));
     let aggregate = Aggregate::new(
         storage,
         projection,
@@ -195,46 +206,55 @@ pub fn select_with_labels<'a, T: 'static + Debug>(
         having,
         filter_context.as_ref().map(Rc::clone),
     );
-    let blend = Blend::new(storage, projection);
-    let filter = Filter::new(storage, where_clause, filter_context, None);
-    let limit = Limit::new(query.limit.as_ref(), query.offset.as_ref())?;
+    let blend = Rc::new(Blend::new(storage, projection));
+    let filter = Rc::new(Filter::new(storage, where_clause, filter_context, None));
+    let limit = Rc::new(Limit::new(query.limit.as_ref(), query.offset.as_ref())?);
 
-    let rows = fetch_blended(storage, table, columns)?
-        .flat_map(move |blend_context| {
+    let rows = fetch_blended(storage, table, columns)
+        .await?
+        .then(move |blend_context| {
             let join_columns = Rc::clone(&join_columns);
+            let join = Rc::clone(&join);
 
-            join.apply(blend_context, join_columns)
+            async move { join.apply(blend_context, join_columns).await }
         })
-        .filter_map(move |blend_context| {
-            blend_context.map_or_else(
-                |error| Some(Err(error)),
-                |blend_context| {
-                    filter
-                        .check_blended(&blend_context)
-                        .map(|pass| pass.as_some(blend_context))
-                        .transpose()
-                },
-            )
+        .try_flatten()
+        .try_filter_map(move |blend_context| {
+            let filter = Rc::clone(&filter);
+
+            async move {
+                filter
+                    .check(Rc::clone(&blend_context))
+                    .await
+                    .map(|pass| pass.as_some(blend_context))
+            }
         })
         .enumerate()
-        .filter_map(move |(i, item)| limit.check(i).as_some(item));
+        .filter_map(move |(i, item)| {
+            let limit = Rc::clone(&limit);
 
-    let rows = {
-        let rows = aggregate.apply(rows)?;
-        let rows = Box::new(rows);
+            async move { limit.check(i).as_some(item) }
+        });
 
-        rows.map(move |blend_context| blend.apply(blend_context))
-    };
+    let rows = aggregate
+        .apply(rows)
+        .await?
+        .into_stream()
+        .and_then(move |aggregate_context| {
+            let blend = Rc::clone(&blend);
+
+            async move { blend.apply(Ok(aggregate_context)).await }
+        });
 
     Ok((labels, rows))
 }
 
-pub fn select<'a, T: 'static + Debug>(
+pub async fn select<'a, T: 'static + Debug>(
     storage: &'a dyn Store<T>,
     query: &'a Query,
     filter_context: Option<Rc<FilterContext<'a>>>,
-) -> Result<impl Iterator<Item = Result<Row>> + 'a> {
-    let (_, rows) = select_with_labels(storage, query, filter_context, false)?;
-
-    Ok(rows)
+) -> Result<impl TryStream<Ok = Row, Error = Error> + 'a> {
+    select_with_labels(storage, query, filter_context, false)
+        .await
+        .map(|(_, rows)| rows)
 }
diff --git a/src/executor/update.rs b/src/executor/update.rs
index ac5023ee..6e21ca5d 100644
--- a/src/executor/update.rs
+++ b/src/executor/update.rs
@@ -1,3 +1,4 @@
+use futures::stream::{self, TryStreamExt};
 use serde::Serialize;
 use std::fmt::Debug;
 use std::rc::Rc;
@@ -27,7 +28,7 @@ pub struct Update<'a, T: 'static + Debug> {
     storage: &'a dyn Store<T>,
     table_name: &'a str,
     fields: &'a [Assignment],
-    columns: &'a [Ident],
+    columns: Rc<Vec<Ident>>,
 }
 
 impl<'a, T: 'static + Debug> Update<'a, T> {
@@ -35,7 +36,7 @@ impl<'a, T: 'static + Debug> Update<'a, T> {
         storage: &'a dyn Store<T>,
         table_name: &'a str,
         fields: &'a [Assignment],
-        columns: &'a [Ident],
+        columns: Rc<Vec<Ident>>,
     ) -> Result<Self> {
         for assignment in fields.iter() {
             let Assignment { id, .. } = assignment;
@@ -53,14 +54,18 @@ impl<'a, T: 'static + Debug> Update<'a, T> {
         })
     }
 
-    fn find(&self, row: &Row, column: &Ident) -> Option<Result<Value>> {
-        let context = FilterContext::new(self.table_name, self.columns, Some(row), None);
+    async fn find(&self, row: &Row, column: &Ident) -> Result<Option<Value>> {
+        let context =
+            FilterContext::new(self.table_name, Rc::clone(&self.columns), Some(row), None);
         let context = Some(Rc::new(context));
 
-        self.fields
+        match self
+            .fields
             .iter()
             .find(|assignment| assignment.id.value == column.value)
-            .map(|assignment| {
+        {
+            None => Ok(None),
+            Some(assignment) => {
                 let Assignment { id, value } = &assignment;
 
                 let index = self
@@ -69,7 +74,8 @@ impl<'a, T: 'static + Debug> Update<'a, T> {
                     .position(|column| column.value == id.value)
                     .ok_or_else(|| UpdateError::Unreachable)?;
 
-                let evaluated = evaluate(self.storage, context, None, value)?;
+                let evaluated = evaluate(self.storage, context, None, value, false).await?;
+
                 let Row(values) = &row;
                 let value = &values[index];
 
@@ -80,25 +86,34 @@ impl<'a, T: 'static + Debug> Update<'a, T> {
                     Evaluated::ValueRef(v) => Ok(v.clone()),
                     Evaluated::Value(v) => Ok(v),
                 }
-            })
+                .map(Some)
+            }
+        }
     }
 
-    pub fn apply(&self, row: Row) -> Result<Row> {
+    pub async fn apply(&self, row: Row) -> Result<Row> {
         let Row(values) = &row;
 
-        values
-            .clone()
-            .into_iter()
-            .enumerate()
-            .map(|(i, value)| {
-                let column = &self
-                    .columns
-                    .get(i)
-                    .ok_or_else(|| UpdateError::ConflictOnSchema)?;
-
-                self.find(&row, column).unwrap_or(Ok(value))
+        let values = values.clone().into_iter().enumerate().map(|(i, value)| {
+            self.columns
+                .get(i)
+                .map(|column| (column, value))
+                .ok_or_else(|| UpdateError::ConflictOnSchema.into())
+        });
+
+        stream::iter(values)
+            .and_then(|(column, value)| {
+                let row = &row;
+
+                async move {
+                    self.find(row, column)
+                        .await
+                        .transpose()
+                        .unwrap_or(Ok(value))
+                }
             })
-            .collect::<Result<_>>()
+            .try_collect::<Vec<_>>()
+            .await
             .map(Row)
     }
 }
diff --git a/src/glue.rs b/src/glue.rs
index 45be18d6..15598421 100644
--- a/src/glue.rs
+++ b/src/glue.rs
@@ -1,5 +1,7 @@
 #[cfg(feature = "sled-storage")]
 use crate::{execute, storages::SledStorage, Payload, Query, Result};
+#[cfg(feature = "sled-storage")]
+use futures::executor::block_on;
 
 #[cfg(feature = "sled-storage")]
 pub struct Glue {
@@ -17,7 +19,7 @@ impl Glue {
     pub fn execute(&mut self, query: &Query) -> Result<Payload> {
         let storage = self.storage.take().unwrap();
 
-        match execute(storage, query) {
+        match block_on(execute(storage, query)) {
             Ok((storage, payload)) => {
                 self.storage = Some(storage);
 
diff --git a/src/result.rs b/src/result.rs
index 8afa13fb..bf4e6d3b 100644
--- a/src/result.rs
+++ b/src/result.rs
@@ -3,8 +3,8 @@ use thiserror::Error as ThisError;
 
 use crate::data::{RowError, TableError, ValueError};
 use crate::executor::{
-    AggregateError, BlendContextError, BlendError, EvaluateError, ExecuteError, FetchError,
-    FilterContextError, FilterError, JoinError, LimitError, SelectError, UpdateError,
+    AggregateError, BlendError, EvaluateError, ExecuteError, FetchError, FilterError, JoinError,
+    LimitError, SelectError, UpdateError,
 };
 
 #[cfg(feature = "alter-table")]
@@ -33,16 +33,12 @@ pub enum Error {
     #[error(transparent)]
     Blend(#[from] BlendError),
     #[error(transparent)]
-    BlendContext(#[from] BlendContextError),
-    #[error(transparent)]
     Aggregate(#[from] AggregateError),
     #[error(transparent)]
     Update(#[from] UpdateError),
     #[error(transparent)]
     Filter(#[from] FilterError),
     #[error(transparent)]
-    FilterContext(#[from] FilterContextError),
-    #[error(transparent)]
     Limit(#[from] LimitError),
     #[error(transparent)]
     Row(#[from] RowError),
@@ -68,11 +64,9 @@ impl PartialEq for Error {
             (Select(e), Select(e2)) => e == e2,
             (Join(e), Join(e2)) => e == e2,
             (Blend(e), Blend(e2)) => e == e2,
-            (BlendContext(e), BlendContext(e2)) => e == e2,
             (Aggregate(e), Aggregate(e2)) => e == e2,
             (Update(e), Update(e2)) => e == e2,
             (Filter(e), Filter(e2)) => e == e2,
-            (FilterContext(e), FilterContext(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/storages/sled_storage/alter_table.rs b/src/storages/sled_storage/alter_table.rs
index 130f16c5..25557d36 100644
--- a/src/storages/sled_storage/alter_table.rs
+++ b/src/storages/sled_storage/alter_table.rs
@@ -1,3 +1,4 @@
+use async_trait::async_trait;
 use boolinator::Boolinator;
 use std::iter::once;
 use std::str;
@@ -31,8 +32,9 @@ macro_rules! fetch_schema {
     }};
 }
 
+#[async_trait]
 impl AlterTable for SledStorage {
-    fn rename_schema(self, table_name: &str, new_table_name: &str) -> MutResult<Self, ()> {
+    async fn rename_schema(self, table_name: &str, new_table_name: &str) -> MutResult<Self, ()> {
         let (_, Schema { column_defs, .. }) = fetch_schema!(self, &self.tree, table_name);
         let schema = Schema {
             table_name: new_table_name.to_string(),
@@ -67,7 +69,7 @@ impl AlterTable for SledStorage {
         Ok((self, ()))
     }
 
-    fn rename_column(
+    async fn rename_column(
         self,
         table_name: &str,
         old_column_name: &str,
@@ -109,7 +111,7 @@ impl AlterTable for SledStorage {
         Ok((self, ()))
     }
 
-    fn add_column(self, table_name: &str, column_def: &ColumnDef) -> MutResult<Self, ()> {
+    async fn add_column(self, table_name: &str, column_def: &ColumnDef) -> MutResult<Self, ()> {
         let (
             key,
             Schema {
@@ -188,7 +190,7 @@ impl AlterTable for SledStorage {
         Ok((self, ()))
     }
 
-    fn drop_column(
+    async fn drop_column(
         self,
         table_name: &str,
         column_name: &str,
diff --git a/src/storages/sled_storage/store.rs b/src/storages/sled_storage/store.rs
index 42e74109..e7b5875d 100644
--- a/src/storages/sled_storage/store.rs
+++ b/src/storages/sled_storage/store.rs
@@ -1,18 +1,27 @@
+use async_trait::async_trait;
 use sled::IVec;
 
 use super::{fetch_schema, SledStorage, StorageError};
 use crate::try_into;
 use crate::{Error, MutResult, Result, Row, RowIter, Schema, Store, StoreMut};
 
+#[async_trait]
 impl StoreMut<IVec> for SledStorage {
-    fn generate_id(self, table_name: &str) -> MutResult<Self, IVec> {
+    async fn generate_id(self, table_name: &str) -> MutResult<Self, IVec> {
         let id = try_into!(self, self.tree.generate_id());
-        let id = format!("data/{}/{}", table_name, id);
+        let id = id.to_be_bytes();
+        let prefix = format!("data/{}/", table_name);
+
+        let bytes = prefix
+            .into_bytes()
+            .into_iter()
+            .chain(id.iter().copied())
+            .collect::<Vec<_>>();
 
-        Ok((self, IVec::from(id.as_bytes())))
+        Ok((self, IVec::from(bytes.as_slice())))
     }
 
-    fn insert_schema(self, schema: &Schema) -> MutResult<Self, ()> {
+    async fn insert_schema(self, schema: &Schema) -> MutResult<Self, ()> {
         let key = format!("schema/{}", schema.table_name);
         let key = key.as_bytes();
         let value = try_into!(self, bincode::serialize(schema));
@@ -22,7 +31,7 @@ impl StoreMut<IVec> for SledStorage {
         Ok((self, ()))
     }
 
-    fn delete_schema(self, table_name: &str) -> MutResult<Self, ()> {
+    async fn delete_schema(self, table_name: &str) -> MutResult<Self, ()> {
         let prefix = format!("data/{}/", table_name);
         let tree = &self.tree;
 
@@ -38,7 +47,7 @@ impl StoreMut<IVec> for SledStorage {
         Ok((self, ()))
     }
 
-    fn insert_data(self, key: &IVec, row: Row) -> MutResult<Self, ()> {
+    async fn insert_data(self, key: &IVec, row: Row) -> MutResult<Self, ()> {
         let value = try_into!(self, bincode::serialize(&row));
 
         try_into!(self, self.tree.insert(key, value));
@@ -46,19 +55,20 @@ impl StoreMut<IVec> for SledStorage {
         Ok((self, ()))
     }
 
-    fn delete_data(self, key: &IVec) -> MutResult<Self, ()> {
+    async fn delete_data(self, key: &IVec) -> MutResult<Self, ()> {
         try_into!(self, self.tree.remove(key));
 
         Ok((self, ()))
     }
 }
 
+#[async_trait]
 impl Store<IVec> for SledStorage {
-    fn fetch_schema(&self, table_name: &str) -> Result<Option<Schema>> {
+    async fn fetch_schema(&self, table_name: &str) -> Result<Option<Schema>> {
         fetch_schema(&self.tree, table_name).map(|(_, schema)| schema)
     }
 
-    fn scan_data(&self, table_name: &str) -> Result<RowIter<IVec>> {
+    async fn scan_data(&self, table_name: &str) -> Result<RowIter<IVec>> {
         let prefix = format!("data/{}/", table_name);
 
         let result_set = self.tree.scan_prefix(prefix.as_bytes()).map(move |item| {
diff --git a/src/store/alter_table.rs b/src/store/alter_table.rs
index 29b735b8..1f037f5b 100644
--- a/src/store/alter_table.rs
+++ b/src/store/alter_table.rs
@@ -1,3 +1,4 @@
+use async_trait::async_trait;
 use serde::Serialize;
 use std::fmt::Debug;
 use thiserror::Error;
@@ -24,22 +25,23 @@ pub enum AlterTableError {
     DroppingColumnNotFound(String),
 }
 
+#[async_trait]
 pub trait AlterTable
 where
     Self: Sized,
 {
-    fn rename_schema(self, table_name: &str, new_table_name: &str) -> MutResult<Self, ()>;
+    async fn rename_schema(self, table_name: &str, new_table_name: &str) -> MutResult<Self, ()>;
 
-    fn rename_column(
+    async fn rename_column(
         self,
         table_name: &str,
         old_column_name: &str,
         new_column_name: &str,
     ) -> MutResult<Self, ()>;
 
-    fn add_column(self, table_name: &str, column_def: &ColumnDef) -> MutResult<Self, ()>;
+    async fn add_column(self, table_name: &str, column_def: &ColumnDef) -> MutResult<Self, ()>;
 
-    fn drop_column(
+    async fn drop_column(
         self,
         table_name: &str,
         column_name: &str,
diff --git a/src/store/mod.rs b/src/store/mod.rs
index 8db9c8d6..b4009333 100644
--- a/src/store/mod.rs
+++ b/src/store/mod.rs
@@ -5,6 +5,7 @@ pub use alter_table::*;
 #[cfg(not(feature = "alter-table"))]
 pub trait AlterTable {}
 
+use async_trait::async_trait;
 use std::fmt::Debug;
 use std::marker::Sized;
 
@@ -14,25 +15,27 @@ use crate::result::{MutResult, Result};
 pub type RowIter<T> = Box<dyn Iterator<Item = Result<(T, Row)>>>;
 
 /// By implementing `Store` trait, you can run `SELECT` queries.
+#[async_trait]
 pub trait Store<T: Debug> {
-    fn fetch_schema(&self, table_name: &str) -> Result<Option<Schema>>;
+    async fn fetch_schema(&self, table_name: &str) -> Result<Option<Schema>>;
 
-    fn scan_data(&self, table_name: &str) -> Result<RowIter<T>>;
+    async fn scan_data(&self, table_name: &str) -> Result<RowIter<T>>;
 }
 
 /// `StoreMut` takes role of mutation, related to `INSERT`, `CREATE`, `DELETE`, `DROP` and
 /// `UPDATE`.
+#[async_trait]
 pub trait StoreMut<T: Debug>
 where
     Self: Sized,
 {
-    fn generate_id(self, table_name: &str) -> MutResult<Self, T>;
+    async fn generate_id(self, table_name: &str) -> MutResult<Self, T>;
 
-    fn insert_schema(self, schema: &Schema) -> MutResult<Self, ()>;
+    async fn insert_schema(self, schema: &Schema) -> MutResult<Self, ()>;
 
-    fn delete_schema(self, table_name: &str) -> MutResult<Self, ()>;
+    async fn delete_schema(self, table_name: &str) -> MutResult<Self, ()>;
 
-    fn insert_data(self, key: &T, row: Row) -> MutResult<Self, ()>;
+    async fn insert_data(self, key: &T, row: Row) -> MutResult<Self, ()>;
 
-    fn delete_data(self, key: &T) -> MutResult<Self, ()>;
+    async fn delete_data(self, key: &T) -> MutResult<Self, ()>;
 }
diff --git a/src/utils/mod.rs b/src/utils/mod.rs
index 5a95ad8a..9a677ec7 100644
--- a/src/utils/mod.rs
+++ b/src/utils/mod.rs
@@ -1,5 +1,7 @@
 mod indexmap;
+mod or_stream;
 mod vector;
 
 pub use self::indexmap::IndexMap;
+pub use or_stream::OrStream;
 pub use vector::Vector;
diff --git a/src/utils/or_stream.rs b/src/utils/or_stream.rs
new file mode 100644
index 00000000..6e54a89f
--- /dev/null
+++ b/src/utils/or_stream.rs
@@ -0,0 +1,114 @@
+use core::cmp::max;
+use core::pin::Pin;
+use futures::ready;
+use futures::stream::Stream;
+use futures::task::{Context, Poll};
+use pin_project::pin_project;
+
+#[derive(Debug)]
+enum State {
+    Initial,
+    InSt1,
+    InSt2,
+}
+
+#[pin_project]
+#[derive(Debug)]
+#[must_use = "streams do nothing unless polled"]
+pub struct OrStream<St1, St2> {
+    #[pin]
+    stream1: St1,
+    #[pin]
+    stream2: St2,
+    state: State,
+}
+
+use State::{InSt1, InSt2, Initial};
+
+impl<St1, St2> OrStream<St1, St2>
+where
+    St1: Stream,
+    St2: Stream<Item = St1::Item>,
+{
+    pub fn new(stream1: St1, stream2: St2) -> Self {
+        Self {
+            stream1,
+            stream2,
+            state: Initial,
+        }
+    }
+}
+
+impl<St1, St2> Stream for OrStream<St1, St2>
+where
+    St1: Stream,
+    St2: Stream<Item = St1::Item>,
+{
+    type Item = St1::Item;
+
+    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        let this = self.project();
+
+        match this.state {
+            Initial => match ready!(this.stream1.poll_next(cx)) {
+                item @ Some(_) => {
+                    *this.state = InSt1;
+
+                    Poll::Ready(item)
+                }
+                None => {
+                    *this.state = InSt2;
+
+                    this.stream2.poll_next(cx)
+                }
+            },
+            InSt1 => this.stream1.poll_next(cx),
+            InSt2 => this.stream2.poll_next(cx),
+        }
+    }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        match self.state {
+            Initial => match self.stream1.size_hint() {
+                (_, Some(0)) => self.stream2.size_hint(),
+                (0, i1_high) => {
+                    let (i2_low, i2_high) = self.stream2.size_hint();
+                    let low = if i2_low > 0 { 1 } else { 0 };
+
+                    let high = i1_high.and_then(|h1| i2_high.map(|h2| max(h1, h2)));
+                    (low, high)
+                }
+                i1_hint => i1_hint,
+            },
+            InSt1 => self.stream1.size_hint(),
+            InSt2 => self.stream2.size_hint(),
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::OrStream;
+    use futures::executor::block_on;
+    use futures::stream::{empty, once, StreamExt};
+
+    #[test]
+    fn basic() {
+        block_on(async move {
+            let s1 = once(async { 1 });
+            let s2 = once(async { 3 });
+            let v = OrStream::new(s1, s2).collect::<Vec<i32>>().await;
+            assert_eq!(vec![1], v);
+
+            let s1 = empty();
+            let s2 = once(async { 3 });
+            let v = OrStream::new(s1, s2).collect::<Vec<i32>>().await;
+            assert_eq!(vec![3], v);
+
+            let s1 = once(async { 3 });
+            let s2 = empty();
+            let v = OrStream::new(s1, s2).collect::<Vec<i32>>().await;
+            assert_eq!(vec![3], v);
+        });
+    }
+}


Test patch:
diff --git a/src/tests/aggregate.rs b/src/tests/aggregate.rs
index 74845818..8338b311 100644
--- a/src/tests/aggregate.rs
+++ b/src/tests/aggregate.rs
@@ -1,6 +1,6 @@
 use crate::*;
 
-pub fn aggregate(mut tester: impl tests::Tester) {
+test_case!(aggregate, async move {
     let create_sql = "
         CREATE TABLE Item (
             id INTEGER,
@@ -9,7 +9,7 @@ pub fn aggregate(mut tester: impl tests::Tester) {
         );
     ";
 
-    tester.run_and_print(create_sql);
+    run!(create_sql);
 
     let insert_sqls = [
         "INSERT INTO Item (id, quantity, age) VALUES (1, 10, 11);",
@@ -20,13 +20,11 @@ pub fn aggregate(mut tester: impl tests::Tester) {
     ];
 
     for insert_sql in insert_sqls.iter() {
-        tester.run(insert_sql).unwrap();
+        run!(insert_sql);
     }
 
     use Value::*;
 
-    let mut run = |sql| tester.run(sql).expect("select");
-
     let test_cases = vec![
         ("SELECT COUNT(*) FROM Item", select!("COUNT(*)"; I64; 5)),
         ("SELECT count(*) FROM Item", select!("count(*)"; I64; 5)),
@@ -64,9 +62,9 @@ pub fn aggregate(mut tester: impl tests::Tester) {
         ),
     ];
 
-    test_cases
-        .into_iter()
-        .for_each(|(sql, expected)| assert_eq!(expected, run(sql)));
+    for (sql, expected) in test_cases.into_iter() {
+        test!(Ok(expected), sql);
+    }
 
     let error_cases = vec![
         (
@@ -82,17 +80,17 @@ pub fn aggregate(mut tester: impl tests::Tester) {
             "SELECT SUM(1 + 2) FROM Item;",
         ),
         (
-            BlendContextError::ValueNotFound.into(),
+            AggregateError::ValueNotFound("num".to_owned()).into(),
             "SELECT SUM(num) FROM Item;",
         ),
     ];
 
-    error_cases
-        .into_iter()
-        .for_each(|(error, sql)| tester.test_error(sql, error));
-}
+    for (error, sql) in error_cases.into_iter() {
+        test!(Err(error), sql);
+    }
+});
 
-pub fn group_by(mut tester: impl tests::Tester) {
+test_case!(group_by, async move {
     let create_sql = "
         CREATE TABLE Item (
             id INTEGER,
@@ -102,7 +100,7 @@ pub fn group_by(mut tester: impl tests::Tester) {
         );
     ";
 
-    tester.run_and_print(create_sql);
+    run!(create_sql);
 
     let insert_sqls = [
         "INSERT INTO Item (id, quantity, city, ratio) VALUES (1, 10, \"Seoul\", 0.2);",
@@ -114,11 +112,9 @@ pub fn group_by(mut tester: impl tests::Tester) {
     ];
 
     for insert_sql in insert_sqls.iter() {
-        tester.run(insert_sql).unwrap();
+        run!(insert_sql);
     }
 
-    let mut run = |sql| tester.run(sql).expect("select");
-
     use Value::*;
 
     let test_cases = vec![
@@ -172,16 +168,16 @@ pub fn group_by(mut tester: impl tests::Tester) {
         ),
     ];
 
-    test_cases
-        .into_iter()
-        .for_each(|(sql, expected)| assert_eq!(expected, run(sql)));
+    for (sql, expected) in test_cases.into_iter() {
+        test!(Ok(expected), 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));
-}
+    for (error, sql) in error_cases.into_iter() {
+        test!(Err(error), sql);
+    }
+});
diff --git a/src/tests/alter_table.rs b/src/tests/alter_table.rs
index 7c245bd8..ed4e6491 100644
--- a/src/tests/alter_table.rs
+++ b/src/tests/alter_table.rs
@@ -3,16 +3,8 @@
 use crate::*;
 use Value::*;
 
-fn run(mut tester: impl tests::Tester, test_cases: &[(&str, Result<Payload>)]) {
-    test_cases.iter().for_each(|(sql, expected)| {
-        let found = tester.run(sql);
-
-        assert_eq!(expected, &found);
-    });
-}
-
-pub fn rename(tester: impl tests::Tester) {
-    let test_cases = [
+test_case!(rename, async move {
+    let test_cases = vec![
         ("CREATE TABLE Foo (id INTEGER);", Ok(Payload::Create)),
         (
             "INSERT INTO Foo VALUES (1), (2), (3);",
@@ -36,11 +28,13 @@ pub fn rename(tester: impl tests::Tester) {
         ),
     ];
 
-    run(tester, &test_cases);
-}
+    for (sql, expected) in test_cases.into_iter() {
+        test!(expected, sql);
+    }
+});
 
-pub fn add_drop(tester: impl tests::Tester) {
-    let test_cases = [
+test_case!(add_drop, async move {
+    let test_cases = vec![
         ("CREATE TABLE Foo (id INTEGER);", Ok(Payload::Create)),
         ("INSERT INTO Foo VALUES (1), (2);", Ok(Payload::Insert(2))),
         ("SELECT * FROM Foo;", Ok(select!(id; I64; 1; 2))),
@@ -126,5 +120,7 @@ pub fn add_drop(tester: impl tests::Tester) {
         ),
     ];
 
-    run(tester, &test_cases);
-}
+    for (sql, expected) in test_cases.into_iter() {
+        test!(expected, sql);
+    }
+});
diff --git a/src/tests/arithmetic.rs b/src/tests/arithmetic.rs
index b68ae231..fca25d82 100644
--- a/src/tests/arithmetic.rs
+++ b/src/tests/arithmetic.rs
@@ -1,6 +1,6 @@
 use crate::*;
 
-pub fn arithmetic(mut tester: impl tests::Tester) {
+test_case!(arithmetic, async move {
     let create_sql = "
         CREATE TABLE Arith (
             id INTEGER,
@@ -8,11 +8,10 @@ pub fn arithmetic(mut tester: impl tests::Tester) {
             name TEXT,
         );
     ";
-
-    tester.run_and_print(create_sql);
+    run!(create_sql);
 
     let delete_sql = "DELETE FROM Arith";
-    tester.run_and_print(delete_sql);
+    run!(delete_sql);
 
     let insert_sqls = [
         "INSERT INTO Arith (id, num, name) VALUES (1, 6, \"A\");",
@@ -23,7 +22,7 @@ pub fn arithmetic(mut tester: impl tests::Tester) {
     ];
 
     for insert_sql in insert_sqls.iter() {
-        tester.run(insert_sql).unwrap();
+        run!(insert_sql);
     }
 
     let test_cases = [
@@ -60,7 +59,7 @@ pub fn arithmetic(mut tester: impl tests::Tester) {
     ];
 
     for (num, sql) in test_cases.iter() {
-        tester.test_rows(sql, *num);
+        count!(*num, sql);
     }
 
     let test_cases = vec![
@@ -86,12 +85,12 @@ pub fn arithmetic(mut tester: impl tests::Tester) {
         ),
     ];
 
-    test_cases
-        .into_iter()
-        .for_each(|(error, sql)| tester.test_error(sql, error));
-}
+    for (error, sql) in test_cases.into_iter() {
+        test!(Err(error), sql);
+    }
+});
 
-pub fn blend(mut tester: impl tests::Tester) {
+test_case!(blend, async move {
     let create_sql = "
         CREATE TABLE Arith (
             id INTEGER,
@@ -99,10 +98,10 @@ pub fn blend(mut tester: impl tests::Tester) {
         );
     ";
 
-    tester.run_and_print(create_sql);
+    run!(create_sql);
 
     let delete_sql = "DELETE FROM Arith";
-    tester.run_and_print(delete_sql);
+    run!(delete_sql);
 
     let insert_sqls = [
         "INSERT INTO Arith (id, num) VALUES (1, 6);",
@@ -113,19 +112,17 @@ pub fn blend(mut tester: impl tests::Tester) {
     ];
 
     for insert_sql in insert_sqls.iter() {
-        tester.run(insert_sql).unwrap();
+        run!(insert_sql);
     }
 
     use Value::I64;
 
     let sql = "SELECT 1 * 2 + 1 - 3 / 1 FROM Arith LIMIT 1;";
-    let found = tester.run(sql).expect("select");
+    let found = run!(sql);
     let expected = select!("1 * 2 + 1 - 3 / 1"; I64; 0);
     assert_eq!(expected, found);
 
-    let found = tester
-        .run("SELECT id, id + 1, id + num, 1 + 1 FROM Arith")
-        .expect("select");
+    let found = run!("SELECT id, id + 1, id + num, 1 + 1 FROM Arith");
     let expected = select!(
         id  | "id + 1" | "id + num" | "1 + 1"
         I64 | I64      | I64        | I64;
@@ -142,7 +139,7 @@ pub fn blend(mut tester: impl tests::Tester) {
       FROM Arith a
       JOIN Arith b ON a.id = b.id + 1
     ";
-    let found = tester.run(sql).expect("select");
+    let found = run!(sql);
     let expected = select!("a.id + b.id"; I64; 3; 5; 7; 9);
     assert_eq!(expected, found);
-}
+});
diff --git a/src/tests/basic.rs b/src/tests/basic.rs
index a8846de0..6409a28c 100644
--- a/src/tests/basic.rs
+++ b/src/tests/basic.rs
@@ -1,42 +1,45 @@
 use crate::*;
 
-pub fn basic(mut tester: impl tests::Tester) {
-    tester.run_and_print(
+test_case!(basic, async move {
+    run!(
         r#"
 CREATE TABLE Test (
     id INTEGER,
     num INTEGER,
     name TEXT
-)"#,
-    );
-    tester.run_and_print("INSERT INTO Test (id, num, name) VALUES (1, 2, \"Hello\")");
-    tester.run_and_print("INSERT INTO Test (id, num, name) VALUES (1, 9, \"World\")");
-    tester.run_and_print(
-        "INSERT INTO Test (id, num, name) VALUES (3, 4, \"Great\"), (4, 7, \"Job\")",
+)"#
     );
+    run!("INSERT INTO Test (id, num, name) VALUES (1, 2, \"Hello\")");
+    run!("INSERT INTO Test (id, num, name) VALUES (1, 9, \"World\")");
+    run!("INSERT INTO Test (id, num, name) VALUES (3, 4, \"Great\"), (4, 7, \"Job\")");
 
     use Value::*;
 
-    let found = tester
-        .run("SELECT id, num, name FROM Test")
-        .expect("select");
-    let expected = select!(
-        id  | num | name
-        I64 | I64 | Str;
-        1     2     "Hello".to_owned();
-        1     9     "World".to_owned();
-        3     4     "Great".to_owned();
-        4     7     "Job".to_owned()
+    test!(
+        Ok(select!(
+            id  | num | name
+            I64 | I64 | Str;
+            1     2     "Hello".to_owned();
+            1     9     "World".to_owned();
+            3     4     "Great".to_owned();
+            4     7     "Job".to_owned()
+        )),
+        "SELECT id, num, name FROM Test"
     );
-    assert_eq!(expected, found);
 
-    tester.run_and_print("UPDATE Test SET id = 2");
+    count!(4, "SELECT * FROM Test");
+
+    run!("UPDATE Test SET id = 2");
 
-    let found = tester.run("SELECT id FROM Test").expect("select");
-    let expected = select!(id; I64; 2; 2; 2; 2);
-    assert_eq!(expected, found);
+    let test_cases = vec![
+        (Ok(select!(id; I64; 2; 2; 2; 2)), "SELECT id FROM Test"),
+        (
+            Ok(select!(id | num; I64 | I64; 2 2; 2 9; 2 4; 2 7)),
+            "SELECT id, num FROM Test",
+        ),
+    ];
 
-    let found = tester.run("SELECT id, num FROM Test").expect("select");
-    let expected = select!(id | num; I64 | I64; 2 2; 2 9; 2 4; 2 7);
-    assert_eq!(expected, found);
-}
+    for (expected, sql) in test_cases.into_iter() {
+        test!(expected, sql);
+    }
+});
diff --git a/src/tests/blend.rs b/src/tests/blend.rs
index be714f33..1b120a3a 100644
--- a/src/tests/blend.rs
+++ b/src/tests/blend.rs
@@ -1,6 +1,6 @@
 use crate::*;
 
-pub fn blend(mut tester: impl tests::Tester) {
+test_case!(blend, async move {
     let create_sqls: [&str; 2] = [
         "
         CREATE TABLE BlendUser (
@@ -17,11 +17,15 @@ pub fn blend(mut tester: impl tests::Tester) {
     ",
     ];
 
-    create_sqls.iter().for_each(|sql| tester.run_and_print(sql));
+    for sql in create_sqls.iter() {
+        run!(sql);
+    }
 
     let delete_sqls = ["DELETE FROM BlendUser", "DELETE FROM BlendItem"];
 
-    delete_sqls.iter().for_each(|sql| tester.run_and_print(sql));
+    for sql in delete_sqls.iter() {
+        run!(sql);
+    }
 
     let insert_sqls = [
         "INSERT INTO BlendUser (id, name) VALUES (1, \"Taehoon\")",
@@ -35,13 +39,11 @@ pub fn blend(mut tester: impl tests::Tester) {
     ];
 
     for insert_sql in insert_sqls.iter() {
-        tester.run(insert_sql).unwrap();
+        run!(insert_sql);
     }
 
     use Value::*;
 
-    let mut run = |sql| tester.run(sql).expect("select");
-
     let test_cases = vec![
         ("SELECT 1 FROM BlendUser", select!(1; I64; 1; 1; 1)),
         (
@@ -121,9 +123,9 @@ pub fn blend(mut tester: impl tests::Tester) {
         ),
     ];
 
-    test_cases
-        .into_iter()
-        .for_each(|(sql, expected)| assert_eq!(expected, run(sql)));
+    for (sql, expected) in test_cases.into_iter() {
+        test!(Ok(expected), sql);
+    }
 
     let error_cases = vec![
         (
@@ -136,7 +138,7 @@ pub fn blend(mut tester: impl tests::Tester) {
         ),
     ];
 
-    error_cases
-        .into_iter()
-        .for_each(|(error, sql)| tester.test_error(sql, error));
-}
+    for (error, sql) in error_cases.into_iter() {
+        test!(Err(error), sql);
+    }
+});
diff --git a/src/tests/create_table.rs b/src/tests/create_table.rs
index a8755a4c..ce0780ea 100644
--- a/src/tests/create_table.rs
+++ b/src/tests/create_table.rs
@@ -1,8 +1,6 @@
 use crate::*;
 
-pub fn create_table(mut tester: impl tests::Tester) {
-    let mut run = |sql| tester.run(sql);
-
+test_case!(create_table, async move {
     let test_cases = vec![
         (
             r#"
@@ -41,7 +39,8 @@ pub fn create_table(mut tester: impl tests::Tester) {
             Ok(Payload::Create),
         ),
     ];
-    test_cases
-        .into_iter()
-        .for_each(|(sql, expected)| assert_eq!(expected, run(sql)));
-}
+
+    for (sql, expected) in test_cases.into_iter() {
+        test!(expected, sql);
+    }
+});
diff --git a/src/tests/default.rs b/src/tests/default.rs
index 1fb14713..adea24ce 100644
--- a/src/tests/default.rs
+++ b/src/tests/default.rs
@@ -1,9 +1,9 @@
 use crate::*;
 
-pub fn default(mut tester: impl tests::Tester) {
+test_case!(default, async move {
     use Value::*;
 
-    let test_cases = [
+    let test_cases = vec![
         (
             "CREATE TABLE Test (
                 id INTEGER DEFAULT 1,
@@ -36,9 +36,7 @@ pub fn default(mut tester: impl tests::Tester) {
         ),
     ];
 
-    test_cases.iter().for_each(|(sql, expected)| {
-        let found = tester.run(sql).unwrap();
-
-        assert_eq!(expected, &found);
-    });
-}
+    for (sql, expected) in test_cases.into_iter() {
+        test!(Ok(expected), sql);
+    }
+});
diff --git a/src/tests/drop_table.rs b/src/tests/drop_table.rs
index 844de747..177d1ef1 100644
--- a/src/tests/drop_table.rs
+++ b/src/tests/drop_table.rs
@@ -1,6 +1,6 @@
 use crate::*;
 
-pub fn drop_table(mut tester: impl tests::Tester) {
+test_case!(drop_table, async move {
     let create_sql = r#"
 CREATE TABLE DropTable (
     id INT,
@@ -8,13 +8,13 @@ CREATE TABLE DropTable (
     name TEXT
 )"#;
 
-    tester.run(create_sql).unwrap();
+    run!(create_sql);
 
     let sqls = ["INSERT INTO DropTable (id, num, name) VALUES (1, 2, \"Hello\")"];
 
-    sqls.iter().for_each(|sql| {
-        tester.run(sql).unwrap();
-    });
+    for sql in sqls.iter() {
+        run!(sql);
+    }
 
     use Value::*;
 
@@ -58,9 +58,7 @@ CREATE TABLE DropTable (
         ),
     ];
 
-    sqls.into_iter().for_each(|(sql, expected)| {
-        let found = tester.run(sql);
-
-        assert_eq!(expected, found);
-    });
-}
+    for (sql, expected) in sqls.into_iter() {
+        test!(expected, sql);
+    }
+});
diff --git a/src/tests/error.rs b/src/tests/error.rs
index b334e156..4362f909 100644
--- a/src/tests/error.rs
+++ b/src/tests/error.rs
@@ -1,8 +1,8 @@
 use crate::*;
 
-pub fn error(mut tester: impl tests::Tester) {
-    tester.run_and_print("CREATE TABLE TableA (id INTEGER);");
-    tester.run_and_print("INSERT INTO TableA (id) VALUES (1);");
+test_case!(error, async move {
+    run!("CREATE TABLE TableA (id INTEGER);");
+    run!("INSERT INTO TableA (id) VALUES (1);");
 
     let test_cases = vec![
         (ExecuteError::QueryNotSupported.into(), "COMMIT;"),
@@ -31,7 +31,7 @@ pub fn error(mut tester: impl tests::Tester) {
             "SELECT * FROM TableA WHERE id = (SELECT id FROM TableA WHERE id = 2);",
         ),
         (
-            FilterContextError::ValueNotFound.into(),
+            EvaluateError::ValueNotFound("noname".to_owned()).into(),
             "SELECT * FROM TableA WHERE noname = 1;",
         ),
         (
@@ -60,13 +60,13 @@ pub fn error(mut tester: impl tests::Tester) {
         ),
     ];
 
-    test_cases
-        .into_iter()
-        .for_each(|(error, sql)| tester.test_error(sql, error));
+    for (error, sql) in test_cases.into_iter() {
+        test!(Err(error), sql);
+    }
 
-    tester.run_and_print("CREATE TABLE TableB (id BOOL);");
-    tester.test_error(
-        "INSERT INTO TableB (id) VALUES (0);",
-        ValueError::SqlTypeNotSupported.into(),
+    run!("CREATE TABLE TableB (id BOOL);");
+    test!(
+        Err(ValueError::SqlTypeNotSupported.into()),
+        "INSERT INTO TableB (id) VALUES (0);"
     );
-}
+});
diff --git a/src/tests/filter.rs b/src/tests/filter.rs
index 34af83c0..40a46cf5 100644
--- a/src/tests/filter.rs
+++ b/src/tests/filter.rs
@@ -1,6 +1,6 @@
 use crate::*;
 
-pub fn filter(mut tester: impl tests::Tester) {
+test_case!(filter, async move {
     let create_sqls = [
         "
         CREATE TABLE Boss (
@@ -15,7 +15,9 @@ pub fn filter(mut tester: impl tests::Tester) {
         );",
     ];
 
-    create_sqls.iter().for_each(|sql| tester.run_and_print(sql));
+    for sql in create_sqls.iter() {
+        run!(sql);
+    }
 
     let insert_sqls = [
         "INSERT INTO Boss (id, name, strength) VALUES (1, \"Amelia\", 10.10);",
@@ -28,7 +30,9 @@ pub fn filter(mut tester: impl tests::Tester) {
         "INSERT INTO Hunter (id, name) VALUES (3, \"Maria\");",
     ];
 
-    insert_sqls.iter().for_each(|sql| tester.run_and_print(sql));
+    for sql in insert_sqls.iter() {
+        run!(sql);
+    }
 
     let select_sqls = [
         (3, "SELECT id, name FROM Boss WHERE id BETWEEN 2 AND 4"),
@@ -68,11 +72,11 @@ pub fn filter(mut tester: impl tests::Tester) {
         (2, "SELECT name FROM Boss WHERE +id <= 2"),
     ];
 
-    select_sqls
-        .iter()
-        .for_each(|(num, sql)| tester.test_rows(sql, *num));
+    for (num, sql) in select_sqls.iter() {
+        count!(*num, sql);
+    }
 
-    let select_sqls_err = vec![
+    let error_sqls = vec![
         (
             EvaluateError::LiteralUnaryPlusOnNonNumeric.into(),
             "SELECT id FROM Hunter WHERE +'abcd' > 1.0",
@@ -91,7 +95,7 @@ pub fn filter(mut tester: impl tests::Tester) {
         ),
     ];
 
-    select_sqls_err
-        .into_iter()
-        .for_each(|(error, sql)| tester.test_error(sql, error));
-}
+    for (error, sql) in error_sqls.into_iter() {
+        test!(Err(error), sql);
+    }
+});
diff --git a/src/tests/function.rs b/src/tests/function.rs
index a4732e8c..a9b9a6bf 100644
--- a/src/tests/function.rs
+++ b/src/tests/function.rs
@@ -1,8 +1,6 @@
 use crate::*;
 
-pub fn function(mut tester: impl tests::Tester) {
-    let mut run = |sql| tester.run(sql);
-
+test_case!(function, async move {
     use Value::Str;
 
     let test_cases = vec![
@@ -61,7 +59,7 @@ pub fn function(mut tester: impl tests::Tester) {
         ),
     ];
 
-    test_cases
-        .into_iter()
-        .for_each(|(sql, expected)| assert_eq!(expected, run(sql)));
-}
+    for (sql, expected) in test_cases.into_iter() {
+        test!(expected, sql);
+    }
+});
diff --git a/src/tests/join.rs b/src/tests/join.rs
index 229c8a78..490dc43e 100644
--- a/src/tests/join.rs
+++ b/src/tests/join.rs
@@ -1,6 +1,6 @@
 use crate::*;
 
-pub fn join(mut tester: impl tests::Tester) {
+test_case!(join, async move {
     let create_sqls: [&str; 2] = [
         "
         CREATE TABLE Player (
@@ -17,11 +17,15 @@ pub fn join(mut tester: impl tests::Tester) {
     ",
     ];
 
-    create_sqls.iter().for_each(|sql| tester.run_and_print(sql));
+    for sql in create_sqls.iter() {
+        run!(sql);
+    }
 
     let delete_sqls = ["DELETE FROM Player", "DELETE FROM Item"];
 
-    delete_sqls.iter().for_each(|sql| tester.run_and_print(sql));
+    for sql in delete_sqls.iter() {
+        run!(sql);
+    }
 
     let insert_sqls = [
         "INSERT INTO Player (id, name) VALUES (1, \"Taehoon\")",
@@ -47,7 +51,7 @@ pub fn join(mut tester: impl tests::Tester) {
     ];
 
     for insert_sql in insert_sqls.iter() {
-        tester.run(insert_sql).unwrap();
+        run!(insert_sql);
     }
 
     let select_sqls = [
@@ -121,14 +125,16 @@ pub fn join(mut tester: impl tests::Tester) {
         (30, "SELECT * FROM Item INNER JOIN Item i2 ON i2.id IN (101, 103);"),
     ];
 
-    select_sqls
-        .iter()
-        .for_each(|(num, sql)| tester.test_rows(sql, *num));
+    for (num, sql) in select_sqls.iter() {
+        count!(*num, sql);
+    }
 
-    delete_sqls.iter().for_each(|sql| tester.run_and_print(sql));
-}
+    for sql in delete_sqls.iter() {
+        run!(sql);
+    }
+});
 
-pub fn blend(mut tester: impl tests::Tester) {
+test_case!(blend, async move {
     let create_sqls: [&str; 2] = [
         "
         CREATE TABLE Player (
@@ -145,7 +151,9 @@ pub fn blend(mut tester: impl tests::Tester) {
     ",
     ];
 
-    create_sqls.iter().for_each(|sql| tester.run_and_print(sql));
+    for sql in create_sqls.iter() {
+        run!(sql);
+    }
 
     let insert_sqls = [
         "INSERT INTO Player (id, name) VALUES (1, \"Taehoon\")",
@@ -159,7 +167,7 @@ pub fn blend(mut tester: impl tests::Tester) {
     ];
 
     for insert_sql in insert_sqls.iter() {
-        tester.run(insert_sql).unwrap();
+        run!(insert_sql);
     }
 
     use Value::{Empty, Str, I64};
@@ -170,7 +178,7 @@ pub fn blend(mut tester: impl tests::Tester) {
         LEFT JOIN Item i
         ON p.id = i.player_id
     ";
-    let found = tester.run(sql).expect("select");
+    let found = run!(sql);
     let expected = select_with_empty!(
         id     | id;
         I64(1)   I64(101);
@@ -187,7 +195,7 @@ pub fn blend(mut tester: impl tests::Tester) {
         LEFT JOIN Item
         ON p.id = player_id
     ";
-    let found = tester.run(sql).expect("select");
+    let found = run!(sql);
     let expected = select_with_empty!(
         id     | player_id;
         I64(1)   I64(1);
@@ -204,7 +212,7 @@ pub fn blend(mut tester: impl tests::Tester) {
         LEFT JOIN Item
         ON p.id = player_id
     ";
-    let found = tester.run(sql).expect("select");
+    let found = run!(sql);
     let expected = select_with_empty!(
         id       | quantity | player_id;
         I64(101)   I64(1)     I64(1);
@@ -221,7 +229,7 @@ pub fn blend(mut tester: impl tests::Tester) {
         LEFT JOIN Item
         ON p.id = player_id
     ";
-    let found = tester.run(sql).expect("select");
+    let found = run!(sql);
     let expected = select_with_empty!(
         id     | name                      | id       | quantity | player_id;
         I64(1)   Str("Taehoon".to_owned())   I64(101)   I64(1)     I64(1);
@@ -231,4 +239,4 @@ pub fn blend(mut tester: impl tests::Tester) {
         I64(5)   Str("Hwan".to_owned())      Empty      Empty      Empty
     );
     assert_eq!(expected, found);
-}
+});
diff --git a/src/tests/migrate.rs b/src/tests/migrate.rs
index 5d998b77..ae756686 100644
--- a/src/tests/migrate.rs
+++ b/src/tests/migrate.rs
@@ -1,6 +1,6 @@
 use crate::*;
 
-pub fn migrate(mut tester: impl tests::Tester) {
+test_case!(migrate, async move {
     let sql = r#"
 CREATE TABLE Test (
     id INT,
@@ -8,7 +8,7 @@ CREATE TABLE Test (
     name TEXT
 )"#;
 
-    tester.run(sql).unwrap();
+    run!(sql);
 
     let sqls = [
         "INSERT INTO Test (id, num, name) VALUES (1, 2, \"Hello\")",
@@ -16,9 +16,9 @@ CREATE TABLE Test (
         "INSERT INTO Test (id, num, name) VALUES (3, 4, \"Great\")",
     ];
 
-    sqls.iter().for_each(|sql| {
-        tester.run(sql).unwrap();
-    });
+    for sql in sqls.iter() {
+        run!(sql);
+    }
 
     let error_cases = vec![
         (
@@ -43,17 +43,13 @@ CREATE TABLE Test (
         ),
     ];
 
-    error_cases.into_iter().for_each(|(error, sql)| {
-        let error = Err(error);
-
-        assert_eq!(error, tester.run(sql));
-    });
+    for (error, sql) in error_cases.into_iter() {
+        test!(Err(error), sql);
+    }
 
     use Value::*;
 
-    let found = tester
-        .run("SELECT id, num, name FROM Test")
-        .expect("select");
+    let found = run!("SELECT id, num, name FROM Test");
     let expected = select!(
         id  | num | name
         I64 | I64 | Str;
@@ -63,9 +59,7 @@ CREATE TABLE Test (
     );
     assert_eq!(expected, found);
 
-    let found = tester
-        .run("SELECT id, num, name FROM Test WHERE id = 1")
-        .expect("select");
+    let found = run!("SELECT id, num, name FROM Test WHERE id = 1");
     let expected = select!(
         id  | num | name
         I64 | I64 | Str;
@@ -74,11 +68,9 @@ CREATE TABLE Test (
     );
     assert_eq!(expected, found);
 
-    tester.run_and_print("UPDATE Test SET id = 2");
+    run!("UPDATE Test SET id = 2");
 
-    let found = tester
-        .run("SELECT id, num, name FROM Test")
-        .expect("select");
+    let found = run!("SELECT id, num, name FROM Test");
     let expected = select!(
         id  | num | name;
         I64 | I64 | Str;
@@ -88,17 +80,15 @@ CREATE TABLE Test (
     );
     assert_eq!(expected, found);
 
-    let found = tester.run("SELECT id FROM Test").expect("select");
+    let found = run!("SELECT id FROM Test");
     let expected = select!(id; I64; 2; 2; 2);
     assert_eq!(expected, found);
 
-    let found = tester.run("SELECT id, num FROM Test").expect("select");
+    let found = run!("SELECT id, num FROM Test");
     let expected = select!(id | num; I64 | I64; 2 2; 2 9; 2 4);
     assert_eq!(expected, found);
 
-    let found = tester
-        .run("SELECT id, num FROM Test LIMIT 1 OFFSET 1")
-        .expect("select");
+    let found = run!("SELECT id, num FROM Test LIMIT 1 OFFSET 1");
     let expected = select!(id | num; I64 | I64; 2 9);
     assert_eq!(expected, found);
-}
+});
diff --git a/src/tests/mod.rs b/src/tests/mod.rs
index 7598a6f9..c3cbfffb 100644
--- a/src/tests/mod.rs
+++ b/src/tests/mod.rs
@@ -45,11 +45,11 @@ macro_rules! generate_tests {
         macro_rules! glue {
             ($title: ident, $func: path) => {
                 #[$test]
-                fn $title() {
+                async fn $title() {
                     let path = stringify!($title);
                     let storage = $storage::new(path);
 
-                    $func(storage);
+                    $func(storage).await;
                 }
             };
         }
@@ -61,20 +61,20 @@ macro_rules! generate_tests {
         glue!(arithmetic_blend, arithmetic::blend);
         glue!(blend, blend::blend);
         glue!(create_table, create_table::create_table);
+        glue!(default, default::default);
         glue!(drop_table, drop_table::drop_table);
         glue!(error, error::error);
+        glue!(filter, filter::filter);
+        glue!(function, function::function);
         glue!(join, join::join);
         glue!(join_blend, join::blend);
         glue!(migrate, migrate::migrate);
         glue!(nested_select, nested_select::nested_select);
-        glue!(default, default::default);
         glue!(nullable, nullable::nullable);
         glue!(nullable_text, nullable::nullable_text);
         glue!(ordering, ordering::ordering);
         glue!(sql_types, sql_types::sql_types);
         glue!(synthesize, synthesize::synthesize);
-        glue!(filter, filter::filter);
-        glue!(function, function::function);
 
         generate_alter_table_tests!();
     };
diff --git a/src/tests/nested_select.rs b/src/tests/nested_select.rs
index 2a0b6c46..a4397f74 100644
--- a/src/tests/nested_select.rs
+++ b/src/tests/nested_select.rs
@@ -1,6 +1,6 @@
 use crate::*;
 
-pub fn nested_select(mut tester: impl tests::Tester) {
+test_case!(nested_select, async move {
     let create_sqls: [&str; 2] = [
         "
         CREATE TABLE User (
@@ -17,7 +17,9 @@ pub fn nested_select(mut tester: impl tests::Tester) {
     ",
     ];
 
-    create_sqls.iter().for_each(|sql| tester.run_and_print(sql));
+    for sql in create_sqls.iter() {
+        run!(sql);
+    }
 
     let insert_sqls = [
         "INSERT INTO User (id, name) VALUES (1, \"Taehoon\")",
@@ -43,7 +45,7 @@ pub fn nested_select(mut tester: impl tests::Tester) {
     ];
 
     for insert_sql in insert_sqls.iter() {
-        tester.run(insert_sql).unwrap();
+        run!(insert_sql);
     }
 
     let select_sqls = [
@@ -66,7 +68,7 @@ pub fn nested_select(mut tester: impl tests::Tester) {
         (9, "SELECT * FROM Request WHERE user_id IN (SELECT id FROM User WHERE name IN (\"Taehoon\", \"Hwan\"));"),
     ];
 
-    select_sqls
-        .iter()
-        .for_each(|(num, sql)| tester.test_rows(sql, *num));
-}
+    for (num, sql) in select_sqls.iter() {
+        count!(*num, sql);
+    }
+});
diff --git a/src/tests/nullable.rs b/src/tests/nullable.rs
index c24c675c..206df356 100644
--- a/src/tests/nullable.rs
+++ b/src/tests/nullable.rs
@@ -1,22 +1,20 @@
 use crate::*;
 
-pub fn nullable(mut tester: impl tests::Tester) {
-    tester.run_and_print(
+test_case!(nullable, async move {
+    run!(
         r#"
 CREATE TABLE Test (
     id INTEGER NULL,
     num INTEGER,
     name TEXT
-)"#,
+)"#
     );
-    tester.run_and_print("INSERT INTO Test (id, num, name) VALUES (NULL, 2, \"Hello\")");
-    tester.run_and_print("INSERT INTO Test (id, num, name) VALUES (1, 9, \"World\")");
-    tester.run_and_print("INSERT INTO Test (id, num, name) VALUES (3, 4, \"Great\")");
+    run!("INSERT INTO Test (id, num, name) VALUES (NULL, 2, \"Hello\")");
+    run!("INSERT INTO Test (id, num, name) VALUES (1, 9, \"World\")");
+    run!("INSERT INTO Test (id, num, name) VALUES (3, 4, \"Great\")");
 
     use Value::*;
 
-    let mut run = |sql| tester.run(sql).expect("select");
-
     let test_cases = vec![
         (
             "SELECT id, num, name FROM Test",
@@ -199,13 +197,12 @@ CREATE TABLE Test (
             ),
         ),
     ];
-    test_cases
-        .into_iter()
-        .for_each(|(sql, expected)| assert_eq!(expected, run(sql)));
 
-    tester.run_and_print("UPDATE Test SET id = 2");
+    for (sql, expected) in test_cases.into_iter() {
+        test!(Ok(expected), sql);
+    }
 
-    let mut run = |sql| tester.run(sql);
+    run!("UPDATE Test SET id = 2");
 
     let test_cases = vec![
         (
@@ -233,19 +230,20 @@ CREATE TABLE Test (
             Err(ValueError::NullValueOnNotNullField.into()),
         ),
     ];
-    test_cases
-        .into_iter()
-        .for_each(|(sql, expected)| assert_eq!(expected, run(sql)));
-}
 
-pub fn nullable_text(mut tester: impl tests::Tester) {
-    tester.run_and_print(
+    for (sql, expected) in test_cases.into_iter() {
+        test!(expected, sql);
+    }
+});
+
+test_case!(nullable_text, async move {
+    run!(
         "
         CREATE TABLE Foo (
             id INTEGER,
             name TEXT NULL
         );
-    ",
+    "
     );
 
     let insert_sqls = [
@@ -254,6 +252,6 @@ pub fn nullable_text(mut tester: impl tests::Tester) {
     ];
 
     for insert_sql in insert_sqls.iter() {
-        tester.run(insert_sql).unwrap();
+        run!(insert_sql);
     }
-}
+});
diff --git a/src/tests/ordering.rs b/src/tests/ordering.rs
index 7cd73b73..ee80ffee 100644
--- a/src/tests/ordering.rs
+++ b/src/tests/ordering.rs
@@ -1,6 +1,6 @@
 use crate::*;
 
-pub fn ordering(mut tester: impl tests::Tester) {
+test_case!(ordering, async move {
     let create_sql = "
         CREATE TABLE Operator (
             id INTEGER,
@@ -8,10 +8,10 @@ pub fn ordering(mut tester: impl tests::Tester) {
         );
     ";
 
-    tester.run_and_print(create_sql);
+    run!(create_sql);
 
     let delete_sql = "DELETE FROM Operator";
-    tester.run_and_print(delete_sql);
+    run!(delete_sql);
 
     let insert_sqls = [
         "INSERT INTO Operator (id, name) VALUES (1, \"Abstract\");",
@@ -22,7 +22,7 @@ pub fn ordering(mut tester: impl tests::Tester) {
     ];
 
     for insert_sql in insert_sqls.iter() {
-        tester.run(insert_sql).unwrap();
+        run!(insert_sql);
     }
 
     let test_cases = [
@@ -66,6 +66,6 @@ pub fn ordering(mut tester: impl tests::Tester) {
     ];
 
     for (num, sql) in test_cases.iter() {
-        tester.test_rows(sql, *num);
+        count!(*num, sql);
     }
-}
+});
diff --git a/src/tests/sql_types.rs b/src/tests/sql_types.rs
index ff120a38..31e260a4 100644
--- a/src/tests/sql_types.rs
+++ b/src/tests/sql_types.rs
@@ -1,6 +1,6 @@
 use crate::*;
 
-pub fn sql_types(mut tester: impl tests::Tester) {
+test_case!(sql_types, async move {
     let create_sql = "
         CREATE TABLE Item (
             id INTEGER,
@@ -10,11 +10,11 @@ pub fn sql_types(mut tester: impl tests::Tester) {
         );
     ";
 
-    tester.run_and_print(create_sql);
+    run!(create_sql);
 
     let delete_sql = "DELETE FROM Item";
 
-    tester.run_and_print(delete_sql);
+    run!(delete_sql);
 
     let insert_sqls = [
         "INSERT INTO Item (id, content, verified, ratio) VALUES (1, \"Hello\", True, 0.1);",
@@ -22,7 +22,7 @@ pub fn sql_types(mut tester: impl tests::Tester) {
     ];
 
     for insert_sql in insert_sqls.iter() {
-        tester.run(insert_sql).unwrap();
+        run!(insert_sql);
     }
 
     let test_sqls = [
@@ -41,8 +41,8 @@ pub fn sql_types(mut tester: impl tests::Tester) {
     ];
 
     for (num, sql) in test_sqls.iter() {
-        tester.test_rows(sql, *num);
+        count!(*num, sql);
     }
 
-    tester.run_and_print(delete_sql);
-}
+    run!(delete_sql);
+});
diff --git a/src/tests/synthesize.rs b/src/tests/synthesize.rs
index 96b44511..fb539e17 100644
--- a/src/tests/synthesize.rs
+++ b/src/tests/synthesize.rs
@@ -1,6 +1,6 @@
 use crate::*;
 
-pub fn synthesize(mut tester: impl tests::Tester) {
+test_case!(synthesize, async move {
     let create_sql = "
         CREATE TABLE TableA (
             id INTEGER,
@@ -9,10 +9,7 @@ pub fn synthesize(mut tester: impl tests::Tester) {
         );
     ";
 
-    tester.run_and_print(create_sql);
-
-    let delete_sql = "DELETE FROM TableA";
-    tester.run_and_print(delete_sql);
+    run!(create_sql);
 
     let insert_sqls = [
         "INSERT INTO TableA (id, test, target_id) VALUES (1, 100, 2);",
@@ -24,7 +21,7 @@ pub fn synthesize(mut tester: impl tests::Tester) {
     ];
 
     for insert_sql in insert_sqls.iter() {
-        tester.run(insert_sql).unwrap();
+        run!(insert_sql);
     }
 
     let test_cases = [
@@ -61,24 +58,28 @@ pub fn synthesize(mut tester: impl tests::Tester) {
     ];
 
     for (num, sql) in test_cases.iter() {
-        tester.test_rows(sql, *num);
+        count!(*num, sql);
     }
 
     for insert_sql in insert_sqls.iter() {
-        tester.run(insert_sql).unwrap();
+        run!(insert_sql);
     }
 
-    let mut test_select = |sql, num| {
-        tester.test_columns(sql, num);
-    };
+    use Value::I64;
 
-    let test_cases = [
-        (2, "SELECT id, test FROM TableA;"),
-        (1, "SELECT id FROM TableA;"),
-        (3, "SELECT * FROM TableA;"),
+    let test_cases = vec![
+        (
+            select!(id | test; I64 | I64; 1 100),
+            "SELECT id, test FROM TableA LIMIT 1;",
+        ),
+        (select!(id; I64; 1), "SELECT id FROM TableA LIMIT 1;"),
+        (
+            select!(id | test | target_id; I64 | I64 | I64; 1 100 2),
+            "SELECT * FROM TableA LIMIT 1;",
+        ),
     ];
 
-    for (num, sql) in test_cases.iter() {
-        test_select(sql, *num);
+    for (expected, sql) in test_cases.into_iter() {
+        test!(Ok(expected), sql);
     }
-}
+});
diff --git a/src/tests/tester.rs b/src/tests/tester.rs
index 6e2622d4..8398073b 100644
--- a/src/tests/tester.rs
+++ b/src/tests/tester.rs
@@ -1,7 +1,40 @@
-use crate::data::Row;
-use crate::executor::Payload;
-use crate::parse::{parse, Query};
-use crate::result::{Error, Result};
+use async_trait::async_trait;
+use std::cell::RefCell;
+use std::fmt::Debug;
+use std::rc::Rc;
+
+use crate::executor::{execute, Payload};
+use crate::parse::parse;
+use crate::result::Result;
+use crate::store::{AlterTable, Store, StoreMut};
+
+pub async fn run<T: 'static + Debug, U: Store<T> + StoreMut<T> + AlterTable>(
+    cell: Rc<RefCell<Option<U>>>,
+    sql: &str,
+) -> Result<Payload> {
+    println!("[Run] {}", sql);
+
+    let mut storage = cell.replace(None).unwrap();
+
+    let query = &parse(sql).unwrap()[0];
+
+    let result = match execute(storage, query).await {
+        Ok((s, payload)) => {
+            storage = s;
+
+            Ok(payload)
+        }
+        Err((s, error)) => {
+            storage = s;
+
+            Err(error)
+        }
+    };
+
+    cell.replace(Some(storage));
+
+    result
+}
 
 /// If you want to make your custom storage and want to run integrate tests,
 /// you should implement this `Tester` trait.
@@ -11,64 +44,54 @@ use crate::result::{Error, Result};
 ///
 /// Actual test cases are in [/src/tests/](https://github.com/gluesql/gluesql/blob/main/src/tests/),
 /// not in `/tests/`.
-pub trait Tester {
+#[async_trait]
+pub trait Tester<T: 'static + Debug, U: Store<T> + StoreMut<T> + AlterTable> {
     fn new(namespace: &str) -> Self;
 
-    fn execute(&mut self, query: &Query) -> Result<Payload>;
-
-    fn run(&mut self, sql: &str) -> Result<Payload> {
-        println!("[Run] {}", sql);
-
-        parse(sql)
-            .unwrap()
-            .iter()
-            .map(|query| self.execute(query))
-            .next()
-            .unwrap()
-    }
-
-    fn run_and_print(&mut self, sql: &str) {
-        let result = self.run(sql);
-
-        match result.unwrap() {
-            Payload::Select { rows, .. } => println!("[Ok ]\n{:#?}\n", rows),
-            Payload::Insert(num) => println!("[Ok ] {} rows inserted.\n", num),
-            Payload::Delete(num) => println!("[Ok ] {} rows deleted.\n", num),
-            Payload::Update(num) => println!("[Ok ] {} rows updated.\n", num),
-            Payload::DropTable | Payload::Create => println!("[Ok ] :)\n"),
-
-            #[cfg(feature = "alter-table")]
-            Payload::AlterTable => println!("[Ok ] :)\n"),
-        };
-    }
-
-    fn test_rows(&mut self, sql: &str, count: usize) {
-        let result = self.run(sql);
-
-        match result.unwrap() {
-            Payload::Select { rows, .. } => assert_eq!(count, rows.len()),
-            Payload::Delete(num) => assert_eq!(count, num),
-            Payload::Update(num) => assert_eq!(count, num),
-            _ => panic!("compare is only for Select, Delete and Update"),
-        };
-    }
-
-    fn test_columns(&mut self, sql: &str, count: usize) {
-        let result = self.run(sql);
-
-        match result.unwrap() {
-            Payload::Select { rows, .. } => {
-                let Row(items) = rows.into_iter().next().unwrap();
-
-                assert_eq!(count, items.len())
+    fn get_cell(&mut self) -> Rc<RefCell<Option<U>>>;
+}
+
+#[macro_export]
+macro_rules! test_case {
+    ($name: ident, $content: expr) => {
+        pub async fn $name<T, U>(mut tester: impl tests::Tester<T, U>)
+        where
+            T: 'static + std::fmt::Debug,
+            U: Store<T> + StoreMut<T> + AlterTable,
+        {
+            use std::rc::Rc;
+
+            let cell = tester.get_cell();
+
+            #[allow(unused_macros)]
+            macro_rules! run {
+                ($sql: expr) => {
+                    tests::run(Rc::clone(&cell), $sql).await.unwrap()
+                };
             }
-            _ => panic!("tests_columns can only handle SELECT"),
-        };
-    }
 
-    fn test_error(&mut self, sql: &str, expected: Error) {
-        let result = self.run(sql);
+            #[allow(unused_macros)]
+            macro_rules! count {
+                ($count: expr, $sql: expr) => {
+                    match tests::run(Rc::clone(&cell), $sql).await.unwrap() {
+                        Payload::Select { rows, .. } => assert_eq!($count, rows.len()),
+                        Payload::Delete(num) => assert_eq!($count, num),
+                        Payload::Update(num) => assert_eq!($count, num),
+                        _ => panic!("compare is only for Select, Delete and Update"),
+                    };
+                };
+            }
+
+            #[allow(unused_macros)]
+            macro_rules! test {
+                ($expected: expr, $sql: expr) => {
+                    let found = tests::run(Rc::clone(&cell), $sql).await;
+
+                    assert_eq!($expected, found);
+                };
+            }
 
-        assert_eq!(result.unwrap_err(), expected);
-    }
+            $content.await
+        }
+    };
 }
diff --git a/tests/sled_storage.rs b/tests/sled_storage.rs
index 3205c585..04369781 100644
--- a/tests/sled_storage.rs
+++ b/tests/sled_storage.rs
@@ -1,19 +1,18 @@
 #[cfg(feature = "sled-storage")]
-use std::convert::TryFrom;
+use sled::IVec;
+#[cfg(feature = "sled-storage")]
+use std::{cell::RefCell, convert::TryFrom, rc::Rc};
 
 #[cfg(feature = "sled-storage")]
-use gluesql::{
-    execute, generate_alter_table_tests, generate_tests, sled, tests::*, Payload, Query, Result,
-    SledStorage,
-};
+use gluesql::{generate_alter_table_tests, generate_tests, sled, tests::*, SledStorage};
 
 #[cfg(feature = "sled-storage")]
 struct SledTester {
-    storage: Option<SledStorage>,
+    storage: Rc<RefCell<Option<SledStorage>>>,
 }
 
 #[cfg(feature = "sled-storage")]
-impl Tester for SledTester {
+impl Tester<IVec, SledStorage> for SledTester {
     fn new(namespace: &str) -> Self {
         let path = format!("data/{}", namespace);
 
@@ -28,29 +27,20 @@ impl Tester for SledTester {
             .path(path)
             .temporary(true)
             .mode(sled::Mode::HighThroughput);
-        let storage = SledStorage::try_from(config).expect("SledStorage::new");
-        let storage = Some(storage);
+
+        let storage = SledStorage::try_from(config)
+            .map(Some)
+            .map(RefCell::new)
+            .map(Rc::new)
+            .expect("SledStorage::new");
 
         SledTester { storage }
     }
 
-    fn execute(&mut self, query: &Query) -> Result<Payload> {
-        let storage = self.storage.take().unwrap();
-
-        match execute(storage, query) {
-            Ok((storage, payload)) => {
-                self.storage = Some(storage);
-
-                Ok(payload)
-            }
-            Err((storage, error)) => {
-                self.storage = Some(storage);
-
-                Err(error)
-            }
-        }
+    fn get_cell(&mut self) -> Rc<RefCell<Option<SledStorage>>> {
+        Rc::clone(&self.storage)
     }
 }
 
 #[cfg(feature = "sled-storage")]
-generate_tests!(test, SledTester);
+generate_tests!(tokio::test, SledTester);