aggregate functions

src/algebra/op.rs

@@ -12,6 +12,7 @@ use std::rc::Rc;
 
 mod assoc;
 mod cartesian;
+mod concat;
 mod filter;
 mod from;
 mod group;
@@ -24,7 +25,6 @@ mod select;
 mod sort;
 mod tagged;
 mod values;
-mod concat;
 
 use crate::data::expr::Expr;
 use crate::data::tuple::{DataKind, OwnTuple, Tuple};
@@ -32,6 +32,7 @@ use crate::data::value::StaticValue;
 use crate::runtime::options::default_read_options;
 pub(crate) use assoc::*;
 pub(crate) use cartesian::*;
+pub(crate) use concat::*;
 pub(crate) use filter::*;
 pub(crate) use from::*;
 pub(crate) use group::*;
@@ -44,7 +45,6 @@ pub(crate) use select::*;
 pub(crate) use sort::*;
 pub(crate) use tagged::*;
 pub(crate) use values::*;
-pub(crate) use concat::*;
 
 #[derive(thiserror::Error, Debug)]
 pub(crate) enum QueryError {

src/algebra/op/assoc.rs

@@ -1,7 +1,7 @@
 use crate::algebra::op::{InterpretContext, RelationalAlgebra};
 use crate::algebra::parser::RaBox;
 use crate::context::TempDbContext;
-use crate::data::expr::{Expr};
+use crate::data::expr::Expr;
 use crate::data::tuple::{OwnTuple, Tuple};
 use crate::data::tuple_set::{BindingMap, TupleSet, TupleSetIdx};
 use crate::data::value::Value;

src/algebra/op/concat.rs

@@ -1,7 +1,7 @@
 use crate::algebra::op::RelationalAlgebra;
 use crate::algebra::parser::{build_relational_expr, AlgebraParseError, RaBox};
 use crate::context::TempDbContext;
-use crate::data::expr::{Expr};
+use crate::data::expr::Expr;
 use crate::data::tuple::{DataKind, OwnTuple};
 use crate::data::tuple_set::{BindingMap, TupleSet, TupleSetIdx};
 use crate::data::value::Value;

src/algebra/op/filter.rs

@@ -1,7 +1,7 @@
 use crate::algebra::op::RelationalAlgebra;
 use crate::algebra::parser::{assert_rule, build_relational_expr, AlgebraParseError, RaBox};
 use crate::context::TempDbContext;
-use crate::data::expr::{Expr};
+use crate::data::expr::Expr;
 use crate::data::tuple_set::{BindingMap, BindingMapEvalContext, TupleSet, TupleSetEvalContext};
 use crate::data::value::{StaticValue, Value};
 use crate::ddl::reify::TableInfo;
@@ -70,10 +70,7 @@ impl<'b> RelationalAlgebra for WhereFilter<'b> {
             map: &source_map,
             parent: self.ctx,
         };
-        let condition = self
-            .condition
-            .clone()
-            .partial_eval(&binding_ctx)?;
+        let condition = self.condition.clone().partial_eval(&binding_ctx)?;
         let txn = self.ctx.txn.clone();
         let temp_db = self.ctx.sess.temp.clone();
         let w_opts = default_write_options();

src/algebra/op/from.rs

@@ -3,7 +3,7 @@ use crate::algebra::op::{
 };
 use crate::algebra::parser::{assert_rule, AlgebraParseError, RaBox};
 use crate::context::TempDbContext;
-use crate::data::expr::{Expr};
+use crate::data::expr::Expr;
 use crate::data::uuid::random_uuid_v1;
 use crate::parser::text_identifier::build_name_in_def;
 use crate::parser::{Pair, Pairs, Rule};

src/algebra/op/insert.rs

@@ -3,7 +3,7 @@ use crate::algebra::op::{
 };
 use crate::algebra::parser::{assert_rule, build_relational_expr, AlgebraParseError, RaBox};
 use crate::context::TempDbContext;
-use crate::data::expr::{Expr};
+use crate::data::expr::Expr;
 use crate::data::parser::parse_scoped_dict;
 use crate::data::tuple::{DataKind, OwnTuple};
 use crate::data::tuple_set::{BindingMap, BindingMapEvalContext, TupleSet, TupleSetEvalContext};

src/algebra/op/merge.rs

@@ -1,8 +1,8 @@
 use crate::algebra::op::{drop_temp_table, RelationalAlgebra};
 use crate::algebra::parser::{assert_rule, build_relational_expr, AlgebraParseError, RaBox};
 use crate::context::TempDbContext;
-use crate::data::expr::{Expr};
-use crate::data::op::{OP_EQ};
+use crate::data::expr::Expr;
+use crate::data::op::OP_EQ;
 use crate::data::tuple::{DataKind, OwnTuple, Tuple};
 use crate::data::tuple_set::{merge_binding_maps, BindingMap, BindingMapEvalContext, TupleSet};
 use crate::ddl::reify::{DdlContext, TableInfo};

src/algebra/op/nested_loop.rs

@@ -1,6 +1,7 @@
 use crate::algebra::op::{build_binding_map_from_info, QueryError, RelationalAlgebra};
 use crate::algebra::parser::RaBox;
 use crate::context::TempDbContext;
+use crate::data::expr::Expr;
 use crate::data::tuple::{DataKind, OwnTuple, ReifiedTuple, Tuple};
 use crate::data::tuple_set::{
     shift_merge_binding_map, BindingMap, BindingMapEvalContext, TableId, TupleSet,
@@ -11,7 +12,6 @@ use crate::runtime::options::{default_read_options, default_write_options};
 use anyhow::Result;
 use cozorocks::{DbPtr, PrefixIterator, ReadOptionsPtr, TransactionPtr, WriteOptionsPtr};
 use std::collections::{BTreeMap, BTreeSet};
-use crate::data::expr::Expr;
 
 pub(crate) const NAME_NESTED_LOOP_LEFT: &str = "NestedLoop";
 
@@ -70,11 +70,7 @@ impl<'b> RelationalAlgebra for NestedLoopLeft<'b> {
         let key_extractors = self
             .join_key_extractor
             .iter()
-            .map(|ex| {
-                ex.clone()
-                    .partial_eval(&binding_ctx)
-                    .map(|ex| ex)
-            })
+            .map(|ex| ex.clone().partial_eval(&binding_ctx).map(|ex| ex))
             .collect::<Result<Vec<_>>>()?;
         let table_id = self.right.table_id();
         let mut key_tuple = OwnTuple::with_prefix(table_id.id);

src/algebra/op/select.rs

@@ -1,7 +1,7 @@
 use crate::algebra::op::RelationalAlgebra;
 use crate::algebra::parser::{assert_rule, build_relational_expr, AlgebraParseError, RaBox};
 use crate::context::TempDbContext;
-use crate::data::expr::{Expr};
+use crate::data::expr::Expr;
 use crate::data::parser::{parse_keyed_dict, parse_scoped_dict};
 use crate::data::tuple::{DataKind, OwnTuple};
 use crate::data::tuple_set::{
@@ -131,11 +131,7 @@ impl<'b> RelationalAlgebra for SelectOp<'b> {
             map: &source_map,
             parent: self.ctx,
         };
-        let extraction_vec = match self
-            .extract_map
-            .clone()
-            .partial_eval(&binding_ctx)?
-        {
+        let extraction_vec = match self.extract_map.clone().partial_eval(&binding_ctx)? {
             Expr::Dict(d) => d.values().cloned().collect::<Vec<_>>(),
             Expr::Const(Value::Dict(d)) => d
                 .values()

src/algebra/op/sort.rs

@@ -1,7 +1,7 @@
 use crate::algebra::op::RelationalAlgebra;
 use crate::algebra::parser::{build_relational_expr, AlgebraParseError, RaBox};
 use crate::context::TempDbContext;
-use crate::data::expr::{Expr};
+use crate::data::expr::Expr;
 use crate::data::tuple::{DataKind, OwnTuple, Tuple};
 use crate::data::tuple_set::{BindingMap, BindingMapEvalContext, TupleSet, MIN_TABLE_ID_BOUND};
 use crate::data::value::Value;

src/algebra/parser.rs

@@ -440,6 +440,25 @@ pub(crate) mod tests {
         }
         let duration_concat = start.elapsed();
         let start = Instant::now();
+        {
+            let ctx = sess.temp_ctx(true);
+            let s = r#"
+              From(e:Employee)
+             .Select({id: e.id, sum_id: e.id.sum[], count: count[null]})
+            "#;
+            let ra = build_relational_expr(
+                &ctx,
+                CozoParser::parse(Rule::ra_expr_all, s)
+                    .unwrap()
+                    .into_iter()
+                    .next()
+                    .unwrap(),
+            )?;
+            dbg!(&ra);
+            dbg!(ra.get_values()?);
+        }
+        let duration_aggr = start.elapsed();
+        let start = Instant::now();
         let mut r_opts = default_read_options();
         r_opts.set_total_order_seek(true);
         r_opts.set_prefix_same_as_start(false);
@@ -464,6 +483,7 @@ pub(crate) mod tests {
             duration_concat,
             duration_outer_join,
             duration_list,
+            duration_aggr,
             n
         );
         Ok(())

src/data.rs

@@ -2,6 +2,7 @@ pub(crate) mod eval;
 pub(crate) mod expr;
 pub(crate) mod key_order;
 pub(crate) mod op;
+pub(crate) mod op_agg;
 pub(crate) mod parser;
 pub(crate) mod tuple;
 pub(crate) mod tuple_set;

src/data/eval.rs

@@ -1,5 +1,6 @@
 use crate::data::expr::Expr;
 use crate::data::op::*;
+use crate::data::op_agg::OpAgg;
 use crate::data::tuple_set::TupleSetIdx;
 use crate::data::value::{StaticValue, Value};
 use anyhow::Result;
@@ -21,6 +22,9 @@ pub enum EvalError {
     #[error("Cannot apply `{0}` to `{1:?}`")]
     OpTypeMismatch(String, Vec<StaticValue>),
 
+    #[error("Expect aggregate expression")]
+    NotAggregate,
+
     #[error("Arity mismatch for {0}, {1} arguments given ")]
     ArityMismatch(String, usize),
 
@@ -94,6 +98,101 @@ impl Expr {
         }
     }
 
+    pub(crate) fn extract_agg_heads(&self) -> Result<Vec<(OpAgg, Vec<Expr>)>> {
+        let mut coll = vec![];
+        fn do_extract(ex: Expr, coll: &mut Vec<(OpAgg, Vec<Expr>)>) -> Result<()> {
+            match ex {
+                Expr::Const(_) => {}
+                Expr::List(l) => {
+                    for ex in l {
+                        do_extract(ex, coll)?;
+                    }
+                }
+                Expr::Dict(d) => {
+                    for (_, ex) in d {
+                        do_extract(ex, coll)?;
+                    }
+                }
+                Expr::Variable(_) => return Err(EvalError::NotAggregate.into()),
+                Expr::TupleSetIdx(_) => return Err(EvalError::NotAggregate.into()),
+                Expr::ApplyAgg(op, _, args) => coll.push((op, args)),
+                Expr::FieldAcc(_, arg) => do_extract(*arg, coll)?,
+                Expr::IdxAcc(_, arg) => do_extract(*arg, coll)?,
+                Expr::IfExpr(args) => {
+                    let (a, b, c) = *args;
+                    do_extract(a, coll)?;
+                    do_extract(b, coll)?;
+                    do_extract(c, coll)?;
+                }
+                Expr::SwitchExpr(args) => {
+                    for (cond, expr) in args {
+                        do_extract(cond, coll)?;
+                        do_extract(expr, coll)?;
+                    }
+                }
+                Expr::OpAnd(args) => {
+                    for ex in args {
+                        do_extract(ex, coll)?
+                    }
+                }
+                Expr::OpOr(args) => {
+                    for ex in args {
+                        do_extract(ex, coll)?
+                    }
+                }
+                Expr::OpCoalesce(args) => {
+                    for ex in args {
+                        do_extract(ex, coll)?
+                    }
+                }
+                Expr::OpMerge(args) => {
+                    for ex in args {
+                        do_extract(ex, coll)?
+                    }
+                }
+                Expr::OpConcat(args) => {
+                    for ex in args {
+                        do_extract(ex, coll)?
+                    }
+                }
+                Expr::BuiltinFn(_, args) => {
+                    for ex in args {
+                        do_extract(ex, coll)?
+                    }
+                }
+            }
+            Ok(())
+        }
+        do_extract(self.clone(), &mut coll)?;
+        Ok(coll)
+    }
+
+    pub(crate) fn is_agg_compatible(&self) -> bool {
+        match self {
+            Expr::Const(_) => true,
+            Expr::List(l) => l.iter().all(|el| el.is_agg_compatible()),
+            Expr::Dict(d) => d.values().all(|el| el.is_agg_compatible()),
+            Expr::Variable(_) => false,
+            Expr::TupleSetIdx(_) => false,
+            Expr::ApplyAgg(_, _, _) => true,
+            Expr::FieldAcc(_, arg) => arg.is_agg_compatible(),
+            Expr::IdxAcc(_, arg) => arg.is_agg_compatible(),
+            Expr::IfExpr(args) => {
+                let (a, b, c) = args.as_ref();
+                a.is_agg_compatible() && b.is_agg_compatible() && c.is_agg_compatible()
+            }
+            Expr::SwitchExpr(args) => args
+                .iter()
+                .all(|(cond, expr)| cond.is_agg_compatible() && expr.is_agg_compatible()),
+            Expr::OpAnd(args) => args.iter().all(|el| el.is_agg_compatible()),
+            Expr::OpOr(args) => args.iter().all(|el| el.is_agg_compatible()),
+            Expr::OpCoalesce(args) => args.iter().all(|el| el.is_agg_compatible()),
+            Expr::OpMerge(args) => args.iter().all(|el| el.is_agg_compatible()),
+            Expr::OpConcat(args) => args.iter().all(|el| el.is_agg_compatible()),
+            Expr::BuiltinFn(_, args) => args.iter().all(|el| el.is_agg_compatible()),
+        }
+    }
+
     pub(crate) fn partial_eval<C: PartialEvalContext>(self, ctx: &C) -> Result<Self> {
         let res = match self {
             v @ (Expr::Const(_) | Expr::TupleSetIdx(_)) => v,
@@ -241,13 +340,14 @@ impl Expr {
             Expr::ApplyAgg(op, a_args, args) => {
                 let a_args = a_args
                     .into_iter()
-                    .map(|v| v.partial_eval(ctx))
+                    .map(|v| v.interpret_eval(ctx).map(|v| v.into_static()))
                     .collect::<Result<Vec<_>>>()?;
                 let args = args
                     .into_iter()
                     .map(|v| v.partial_eval(ctx))
                     .collect::<Result<Vec<_>>>()?;
-                Expr::ApplyAgg(op, a_args, args)
+                op.initialize(a_args)?;
+                Expr::ApplyAgg(op, vec![], args)
             }
             Expr::IfExpr(args) => {
                 let (cond, if_part, else_part) = *args;
@@ -275,20 +375,47 @@ impl Expr {
                 .into(),
             Expr::Variable(v) => return Err(EvalError::UnresolvedVariable(v.clone()).into()),
             Expr::TupleSetIdx(idx) => ctx.resolve(idx)?.clone(),
-            Expr::BuiltinFn(op, args) => {
-                let mut eval_args = Vec::with_capacity(args.len());
-                for v in args {
-                    let v = v.row_eval(ctx)?;
+            Expr::BuiltinFn(op, args) => match args.len() {
+                0 => (op.func)(&[])?,
+                1 => {
+                    let v = args.iter().next().unwrap().row_eval(ctx)?;
                     if op.non_null_args && v == Value::Null {
                         return Ok(Value::Null);
                     } else {
-                        eval_args.push(v);
+                        (op.func)(&[v])?
                     }
                 }
-                (op.func)(&eval_args)?
-            }
-            Expr::ApplyAgg(_, _, _) => {
-                todo!()
+                2 => {
+                    let mut args = args.iter();
+                    let v1 = args.next().unwrap().row_eval(ctx)?;
+                    if op.non_null_args && v1 == Value::Null {
+                        return Ok(Value::Null);
+                    }
+                    let v2 = args.next().unwrap().row_eval(ctx)?;
+                    if op.non_null_args && v2 == Value::Null {
+                        return Ok(Value::Null);
+                    }
+                    (op.func)(&[v1, v2])?
+                }
+                _ => {
+                    let mut eval_args = Vec::with_capacity(args.len());
+                    for v in args {
+                        let v = v.row_eval(ctx)?;
+                        if op.non_null_args && v == Value::Null {
+                            return Ok(Value::Null);
+                        } else {
+                            eval_args.push(v);
+                        }
+                    }
+                    (op.func)(&eval_args)?
+                }
+            },
+            Expr::ApplyAgg(op, _, args) => {
+                let mut eval_args = Vec::with_capacity(args.len());
+                for v in args {
+                    eval_args.push(v.row_eval(ctx)?);
+                }
+                op.put_get(&eval_args)?
             }
             Expr::FieldAcc(f, arg) => match arg.row_eval(ctx)? {
                 Value::Null => Value::Null,
@@ -319,7 +446,7 @@ impl Expr {
             Expr::OpOr(args) => row_eval_or(ctx, args)?,
             Expr::OpCoalesce(args) => row_eval_coalesce(ctx, args)?,
             Expr::OpMerge(args) => row_eval_merge(ctx, args)?,
-            Expr::OpConcat(args) => row_eval_concat(ctx, args)?
+            Expr::OpConcat(args) => row_eval_concat(ctx, args)?,
         };
         Ok(res)
     }

src/data/expr.rs

@@ -1,4 +1,5 @@
 use crate::data::op::*;
+use crate::data::op_agg::OpAgg;
 use crate::data::tuple_set::TupleSetIdx;
 use crate::data::value::{StaticValue, Value};
 use crate::parser::{CozoParser, Rule};
@@ -39,15 +40,14 @@ impl PartialEq for BuiltinFn {
 
 impl Eq for BuiltinFn {}
 
-#[derive(Clone, PartialEq, Eq)]
+#[derive(Clone, PartialEq)]
 pub enum Expr {
     Const(StaticValue),
     List(Vec<Expr>),
     Dict(BTreeMap<String, Expr>),
     Variable(String),
     TupleSetIdx(TupleSetIdx),
-    // Apply(Arc<dyn Op + Send + Sync>, Vec<Expr>),
-    ApplyAgg((), Vec<Expr>, Vec<Expr>),
+    ApplyAgg(OpAgg, Vec<Expr>, Vec<Expr>),
     FieldAcc(String, Box<Expr>),
     IdxAcc(usize, Box<Expr>),
     IfExpr(Box<(Expr, Expr, Expr)>),
@@ -163,7 +163,7 @@ impl<'a> Debug for Expr {
             Expr::ApplyAgg(op, a_args, args) => write!(
                 f,
                 "[|{} {} | {}|]",
-                todo!(),
+                op.name(),
                 a_args
                     .iter()
                     .map(|v| format!("{:?}", v))

src/data/op/arithmetic.rs

@@ -1,13 +1,13 @@
 use crate::data::eval::EvalError;
+use crate::data::expr::BuiltinFn;
 use crate::data::value::Value;
 use anyhow::Result;
-use crate::data::expr::BuiltinFn;
 
 pub(crate) const OP_ADD: BuiltinFn = BuiltinFn {
     name: NAME_OP_ADD,
     arity: Some(2),
     non_null_args: true,
-    func: op_add
+    func: op_add,
 };
 
 pub(crate) const NAME_OP_ADD: &str = "+";
@@ -36,7 +36,7 @@ pub(crate) const OP_SUB: BuiltinFn = BuiltinFn {
     name: NAME_OP_SUB,
     arity: Some(2),
     non_null_args: true,
-    func: op_sub
+    func: op_sub,
 };
 
 pub(crate) const NAME_OP_SUB: &str = "-";
@@ -65,7 +65,7 @@ pub(crate) const OP_MUL: BuiltinFn = BuiltinFn {
     name: NAME_OP_MUL,
     arity: Some(2),
     non_null_args: true,
-    func: op_mul
+    func: op_mul,
 };
 
 pub(crate) const NAME_OP_MUL: &str = "*";
@@ -95,7 +95,7 @@ pub(crate) const OP_DIV: BuiltinFn = BuiltinFn {
     name: NAME_OP_DIV,
     arity: Some(2),
     non_null_args: true,
-    func: op_div
+    func: op_div,
 };
 
 pub(crate) const NAME_OP_DIV: &str = "/";
@@ -125,7 +125,7 @@ pub(crate) const OP_MOD: BuiltinFn = BuiltinFn {
     name: NAME_OP_MOD,
     arity: Some(2),
     non_null_args: true,
-    func: op_mod
+    func: op_mod,
 };
 
 pub(crate) const NAME_OP_MOD: &str = "%";
@@ -152,7 +152,7 @@ pub(crate) const OP_POW: BuiltinFn = BuiltinFn {
     name: NAME_OP_POW,
     arity: Some(2),
     non_null_args: true,
-    func: op_pow
+    func: op_pow,
 };
 
 pub(crate) const NAME_OP_POW: &str = "**";
@@ -182,7 +182,7 @@ pub(crate) const OP_MINUS: BuiltinFn = BuiltinFn {
     name: NAME_OP_MINUS,
     arity: Some(1),
     non_null_args: true,
-    func: op_mul
+    func: op_mul,
 };
 
 pub(crate) const NAME_OP_MINUS: &str = "--";
@@ -192,8 +192,10 @@ pub(crate) fn op_minus<'a>(args: &[Value<'a>]) -> Result<Value<'a>> {
     match arg {
         Value::Int(i) => Ok((-i).into()),
         Value::Float(i) => Ok((-i).into()),
-        v => {
-            Err(EvalError::OpTypeMismatch(NAME_OP_MINUS.to_string(), vec![v.clone().into_static()]).into())
-        }
+        v => Err(EvalError::OpTypeMismatch(
+            NAME_OP_MINUS.to_string(),
+            vec![v.clone().into_static()],
+        )
+        .into()),
     }
 }

src/data/op/boolean.rs

@@ -76,7 +76,7 @@ pub(crate) fn partial_eval_or<'a, T: PartialEvalContext>(
 
 pub(crate) fn row_eval_or<'a, T: RowEvalContext + 'a>(
     ctx: &'a T,
-    args: &[Expr]
+    args: &[Expr],
 ) -> Result<Value<'a>> {
     let mut has_null = false;
     for arg in args {
@@ -90,7 +90,7 @@ pub(crate) fn row_eval_or<'a, T: RowEvalContext + 'a>(
                     NAME_OP_OR.to_string(),
                     vec![v.clone().into_static()],
                 )
-                    .into());
+                .into());
             }
         }
     }
@@ -146,7 +146,7 @@ pub(crate) fn partial_eval_and<'a, T: PartialEvalContext>(
 
 pub(crate) fn row_eval_and<'a, T: RowEvalContext + 'a>(
     ctx: &'a T,
-    args: &[Expr]
+    args: &[Expr],
 ) -> Result<Value<'a>> {
     let mut has_null = false;
     for arg in args {
@@ -160,7 +160,7 @@ pub(crate) fn row_eval_and<'a, T: RowEvalContext + 'a>(
                     NAME_OP_AND.to_string(),
                     vec![v.clone().into_static()],
                 )
-                    .into());
+                .into());
             }
         }
     }
@@ -183,6 +183,9 @@ pub(crate) const NAME_OP_NOT: &str = "!";
 pub(crate) fn op_not<'a>(args: &[Value<'a>]) -> Result<Value<'a>> {
     match args.into_iter().next().unwrap() {
         Value::Bool(b) => Ok((!b).into()),
-        v => Err(EvalError::OpTypeMismatch(NAME_OP_NOT.to_string(), vec![v.clone().into_static()]).into()),
+        v => Err(
+            EvalError::OpTypeMismatch(NAME_OP_NOT.to_string(), vec![v.clone().into_static()])
+                .into(),
+        ),
     }
 }

src/data/op/text.rs

@@ -1,13 +1,13 @@
 use crate::data::eval::EvalError;
+use crate::data::expr::BuiltinFn;
 use crate::data::value::Value;
 use anyhow::Result;
-use crate::data::expr::BuiltinFn;
 
 pub(crate) const OP_STR_CAT: BuiltinFn = BuiltinFn {
     name: NAME_OP_STR_CAT,
     arity: None,
     non_null_args: true,
-    func: op_str_cat
+    func: op_str_cat,
 };
 
 pub(crate) const NAME_OP_STR_CAT: &str = "++";

src/data/op_agg.rs

@@ -0,0 +1,81 @@
+use crate::data::eval::EvalError;
+use crate::data::value::{StaticValue, Value};
+use anyhow::Result;
+use std::ops::Deref;
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::sync::Arc;
+
+#[derive(Clone)]
+pub struct OpAgg(pub(crate) Arc<dyn OpAggT + Send + Sync>);
+
+impl Deref for OpAgg {
+    type Target = Arc<dyn OpAggT + Send + Sync>;
+
+    fn deref(&self) -> &Self::Target {
+        &self.0
+    }
+}
+
+impl PartialEq for OpAgg {
+    fn eq(&self, other: &Self) -> bool {
+        self.name() == other.name()
+    }
+}
+
+pub trait OpAggT {
+    fn name(&self) -> &str;
+    fn arity(&self) -> Option<usize>;
+    fn reset(&self);
+    fn initialize(&self, a_args: Vec<StaticValue>) -> Result<()>;
+    fn put(&self, args: &[Value]) -> Result<()>;
+    fn get(&self) -> Result<StaticValue>;
+    fn put_get(&self, args: &[Value]) -> Result<StaticValue> {
+        self.put(args)?;
+        self.get()
+    }
+}
+
+#[derive(Default)]
+pub struct OpSum {
+    total: AtomicUsize,
+}
+
+pub(crate) const NAME_OP_COUNT: &str = "count";
+pub(crate) const NAME_OP_SUM: &str = "sum";
+
+impl OpAggT for OpSum {
+    fn name(&self) -> &str {
+        NAME_OP_COUNT
+    }
+
+    fn arity(&self) -> Option<usize> {
+        Some(1)
+    }
+
+    fn reset(&self) {
+        self.total.swap(0, Ordering::Relaxed);
+    }
+
+    fn initialize(&self, _a_args: Vec<StaticValue>) -> Result<()> {
+        Ok(())
+    }
+
+    fn put(&self, args: &[Value]) -> Result<()> {
+        let arg = args.iter().next().unwrap();
+        match arg {
+            Value::Int(i) => {
+                self.total.fetch_add(*i as usize, Ordering::Relaxed);
+                Ok(())
+            }
+            v => Err(EvalError::OpTypeMismatch(
+                self.name().to_string(),
+                vec![v.clone().into_static()],
+            )
+            .into()),
+        }
+    }
+
+    fn get(&self) -> Result<StaticValue> {
+        Ok((self.total.load(Ordering::Relaxed) as i64).into())
+    }
+}

src/data/parser.rs

@@ -1,5 +1,6 @@
 use crate::data::expr::Expr;
 use crate::data::op::*;
+use crate::data::op_agg::*;
 use crate::data::value::Value;
 use crate::parser::number::parse_int;
 use crate::parser::text_identifier::parse_string;
@@ -9,6 +10,7 @@ use lazy_static::lazy_static;
 use pest::prec_climber::{Assoc, Operator, PrecClimber};
 use std::borrow::Cow;
 use std::collections::BTreeMap;
+use std::sync::Arc;
 
 #[derive(thiserror::Error, Debug)]
 pub enum ExprParseError {
@@ -74,13 +76,6 @@ fn build_cond_expr(pair: Pair) -> Result<Expr> {
     Ok(res)
 }
 
-fn build_call_expr(pair: Pair) -> Result<Expr> {
-    let mut pairs = pair.into_inner();
-    let op_name = pairs.next().unwrap().as_str();
-    let args = pairs.map(Expr::try_from).collect::<Result<Vec<_>>>()?;
-    build_method_call(op_name, args)
-}
-
 fn build_switch_expr(pair: Pair) -> Result<Expr> {
     let mut pairs = pair.into_inner();
     let expr = pairs.next().unwrap();
@@ -142,6 +137,24 @@ fn build_expr_primary(pair: Pair) -> Result<Expr> {
                         args.extend(pairs.map(Expr::try_from).collect::<Result<Vec<_>>>()?);
                         head = build_method_call(op_name, args)?;
                     }
+                    Rule::aggr => {
+                        let mut pairs = p.into_inner();
+                        let op_name = pairs.next().unwrap().as_str();
+                        let mut args = vec![head];
+                        let mut a_args = vec![];
+                        for pair in pairs {
+                            match pair.as_rule() {
+                                Rule::expr => args.push(Expr::try_from(pair)?),
+                                Rule::aggr_params => {
+                                    for pair in pair.into_inner() {
+                                        a_args.push(Expr::try_from(pair)?);
+                                    }
+                                }
+                                _ => unreachable!(),
+                            }
+                        }
+                        head = build_aggr_call(op_name, a_args, args)?;
+                    }
                     _ => todo!(),
                 }
             }
@@ -278,7 +291,30 @@ fn build_expr_primary(pair: Pair) -> Result<Expr> {
         }
         Rule::param => Ok(Expr::Variable(pair.as_str().into())),
         Rule::ident => Ok(Expr::Variable(pair.as_str().into())),
-        Rule::call_expr => build_call_expr(pair),
+        Rule::call_expr => {
+            let mut pairs = pair.into_inner();
+            let op_name = pairs.next().unwrap().as_str();
+            let args = pairs.map(Expr::try_from).collect::<Result<Vec<_>>>()?;
+            build_method_call(op_name, args)
+        }
+        Rule::aggr_expr => {
+            let mut pairs = pair.into_inner();
+            let op_name = pairs.next().unwrap().as_str();
+            let mut args = vec![];
+            let mut a_args = vec![];
+            for pair in pairs {
+                match pair.as_rule() {
+                    Rule::expr => args.push(Expr::try_from(pair)?),
+                    Rule::aggr_params => {
+                        for pair in pair.into_inner() {
+                            a_args.push(Expr::try_from(pair)?);
+                        }
+                    }
+                    _ => unreachable!(),
+                }
+            }
+            build_aggr_call(op_name, a_args, args)
+        }
         _ => {
             println!("Unhandled rule {:?}", pair.as_rule());
             unimplemented!()
@@ -292,7 +328,19 @@ fn build_method_call(name: &str, args: Vec<Expr>) -> Result<Expr> {
         NAME_OP_NOT_NULL => Expr::BuiltinFn(OP_NOT_NULL, args),
         NAME_OP_CONCAT => Expr::OpConcat(args),
         NAME_OP_MERGE => Expr::OpMerge(args),
-        _method_name => unimplemented!(),
+        method_name => unimplemented!("{}", method_name),
+    })
+}
+
+fn build_aggr_call(name: &str, a_args: Vec<Expr>, args: Vec<Expr>) -> Result<Expr> {
+    Ok(match name {
+        NAME_OP_SUM => Expr::ApplyAgg(OpAgg(Arc::new(OpSum::default())), a_args, args),
+        NAME_OP_COUNT => Expr::ApplyAgg(
+            OpAgg(Arc::new(OpSum::default())),
+            a_args,
+            vec![Expr::Const(Value::Int(1))],
+        ),
+        method_name => unimplemented!("{}", method_name),
     })
 }
 

src/ddl/parser.rs

@@ -1,4 +1,4 @@
-use crate::data::expr::{Expr};
+use crate::data::expr::Expr;
 use crate::data::typing::Typing;
 use crate::data::value::{StaticValue, Value};
 use crate::parser::text_identifier::build_name_in_def;

src/grammar.pest

@@ -114,10 +114,11 @@ minus = { "-" }
 negate = { "!" }
 
 call_expr = { fn_ident ~ "(" ~ (expr ~ ",")* ~ expr? ~ ")" }
-aggr_expr = { fn_ident ~ "[" ~ (expr ~ ",")* ~ expr? ~ "]" }
+aggr_expr = { fn_ident ~ "[" ~ (expr ~ ",")* ~ expr? ~ aggr_params? ~ "]" }
+aggr_params = { ";" ~ (expr ~ ",")* ~ expr? }
 term = { (grouping | call_expr | aggr_expr | literal | ident | param | list | dict) ~ (call | aggr | accessor | index_accessor)* }
 call = {"." ~ fn_ident ~ "(" ~ (expr ~ ",")* ~ expr? ~ ")"}
-aggr = {"." ~ fn_ident ~ "[" ~ (expr ~ ",")* ~ expr? ~ "]"}
+aggr = {"." ~ fn_ident ~ "[" ~ (expr ~ ",")* ~ expr? ~ aggr_params? ~ "]"}
 accessor = {"." ~ ident}
 index_accessor = {"[" ~ int ~ "]"}
 grouping = { "(" ~ expr ~ ")" }

src/runtime/session.rs

@@ -1,3 +1,4 @@
+use crate::data::expr::Expr;
 use crate::data::tuple::{DataKind, OwnTuple, Tuple};
 use crate::data::tuple_set::{TableId, MIN_TABLE_ID_BOUND};
 use crate::data::value::{StaticValue, Value};
@@ -14,7 +15,6 @@ use pest::Parser;
 use std::collections::{BTreeMap, BTreeSet};
 use std::sync::atomic::{AtomicU32, Ordering};
 use std::sync::{Arc, Mutex};
-use crate::data::expr::Expr;
 
 pub(crate) enum Definable {
     Value(StaticValue),