Fixed and added multiple e2e tests

src/binder/aggregate.rs

@@ -30,7 +30,7 @@ impl<S: Storage> Binder<S> {
         select_items: &mut [ScalarExpression],
     ) -> Result<(), BindError> {
         for column in select_items {
-            self.visit_column_agg_expr(column);
+            self.visit_column_agg_expr(column, true)?;
         }
         Ok(())
     }
@@ -57,7 +57,8 @@ impl<S: Storage> Binder<S> {
         // Extract having expression.
         let return_having = if let Some(having) = having {
             let mut having = self.bind_expr(having).await?;
-            self.visit_column_agg_expr(&mut having);
+            self.visit_column_agg_expr(&mut having, false)?;
+
             Some(having)
         } else {
             None
@@ -73,11 +74,11 @@ impl<S: Storage> Binder<S> {
                     nulls_first,
                 } = orderby;
                 let mut expr = self.bind_expr(expr).await?;
-                self.visit_column_agg_expr(&mut expr);
+                self.visit_column_agg_expr(&mut expr, false)?;
 
                 return_orderby.push(SortField::new(
                     expr,
-                    asc.map_or(true, |asc| !asc),
+                    asc.map_or(true, |asc| asc),
                     nulls_first.map_or(false, |first| first),
                 ));
             }
@@ -88,50 +89,67 @@ impl<S: Storage> Binder<S> {
         Ok((return_having, return_orderby))
     }
 
-    fn visit_column_agg_expr(&mut self, expr: &mut ScalarExpression) {
+    fn visit_column_agg_expr(&mut self, expr: &mut ScalarExpression, is_select: bool) -> Result<(), BindError> {
         match expr {
             ScalarExpression::AggCall {
                 ty: return_type, ..
             } => {
-                let index = self.context.input_ref_index(InputRefType::AggCall);
-                let input_ref = ScalarExpression::InputRef {
-                    index,
-                    ty: return_type.clone(),
-                };
-                match std::mem::replace(expr, input_ref) {
-                    ScalarExpression::AggCall {
-                        kind,
-                        args,
+                let ty = return_type.clone();
+                if is_select {
+                    let index = self.context.input_ref_index(InputRefType::AggCall);
+                    let input_ref = ScalarExpression::InputRef {
+                        index,
                         ty,
-                        distinct
-                    } => {
-                        self.context.agg_calls.push(ScalarExpression::AggCall {
-                            distinct,
+                    };
+                    match std::mem::replace(expr, input_ref) {
+                        ScalarExpression::AggCall {
                             kind,
                             args,
                             ty,
-                        });
+                            distinct
+                        } => {
+                            self.context.agg_calls.push(ScalarExpression::AggCall {
+                                distinct,
+                                kind,
+                                args,
+                                ty,
+                            });
+                        }
+                        _ => unreachable!(),
                     }
-                    _ => unreachable!(),
+                } else {
+                    let (index, _) = self
+                        .context
+                        .agg_calls
+                        .iter()
+                        .find_position(|agg_expr| agg_expr == &expr)
+                        .ok_or_else(|| BindError::AggMiss(format!("{:?}", expr)))?;
+
+                    let _ = std::mem::replace(expr, ScalarExpression::InputRef {
+                        index,
+                        ty,
+                    });
                 }
             }
 
-            ScalarExpression::TypeCast { expr, .. } => self.visit_column_agg_expr(expr),
-            ScalarExpression::IsNull { expr } => self.visit_column_agg_expr(expr),
-            ScalarExpression::Unary { expr, .. } => self.visit_column_agg_expr(expr),
-            ScalarExpression::Alias { expr, .. } => self.visit_column_agg_expr(expr),
+            ScalarExpression::TypeCast { expr, .. } => self.visit_column_agg_expr(expr, is_select)?,
+            ScalarExpression::IsNull { expr } => self.visit_column_agg_expr(expr, is_select)?,
+            ScalarExpression::Unary { expr, .. } => self.visit_column_agg_expr(expr, is_select)?,
+            ScalarExpression::Alias { expr, .. } => self.visit_column_agg_expr(expr, is_select)?,
             ScalarExpression::Binary {
                 left_expr,
                 right_expr,
                 ..
             } => {
-                self.visit_column_agg_expr(left_expr);
-                self.visit_column_agg_expr(right_expr);
+                self.visit_column_agg_expr(left_expr, is_select)?;
+                self.visit_column_agg_expr(right_expr, is_select)?;
             }
             ScalarExpression::Constant(_)
             | ScalarExpression::ColumnRef { .. }
             | ScalarExpression::InputRef { .. } => {}
         }
+
+        Ok(())
     }
 
     /// Validate select exprs must appear in the GROUP BY clause or be used in
@@ -173,6 +191,7 @@ impl<S: Storage> Binder<S> {
             if expr.has_agg_call(&self.context) {
                 continue;
             }
+
             group_raw_set.remove(expr);
 
             if !group_raw_exprs.iter().contains(expr) {
@@ -271,6 +290,9 @@ impl<S: Storage> Binder<S> {
                 if self.context.group_by_exprs.contains(expr) {
                     return Ok(());
                 }
+                if matches!(expr, ScalarExpression::Alias { .. }) {
+                    return self.validate_having_orderby(expr.unpack_alias());
+                }
 
                 Err(BindError::AggMiss(
                     format!(

src/binder/create_table.rs

@@ -1,7 +1,7 @@
 use std::collections::HashSet;
 use std::sync::Arc;
 use itertools::Itertools;
-use sqlparser::ast::{ColumnDef, ObjectName};
+use sqlparser::ast::{ColumnDef, ObjectName, TableConstraint};
 
 use super::Binder;
 use crate::binder::{BindError, lower_case_name, split_name};
@@ -12,10 +12,12 @@ use crate::planner::operator::Operator;
 use crate::storage::Storage;
 
 impl<S: Storage> Binder<S> {
+    // TODO: TableConstraint
     pub(crate) fn bind_create_table(
         &mut self,
         name: &ObjectName,
         columns: &[ColumnDef],
+        constraints: &[TableConstraint]
     ) -> Result<LogicalPlan, BindError> {
         let name = lower_case_name(&name);
         let (_, name) = split_name(&name)?;

src/binder/expr.rs

@@ -85,7 +85,7 @@ impl<S: Storage> Binder<S> {
             }
             if got_column.is_none() {
                 if let Some(expr) = self.context.aliases.get(column_name) {
-                    return Ok(expr.clone());
+                    return Ok(ScalarExpression::Alias { expr: Box::new(expr.clone()), alias: column_name.clone() });
                 }
             }
             let column_catalog =
@@ -167,7 +167,7 @@ impl<S: Storage> Binder<S> {
                 distinct: func.distinct,
                 kind: AggKind::Count,
                 args,
-                ty: LogicalType::UInteger,
+                ty: LogicalType::Integer,
             },
             "sum" => ScalarExpression::AggCall{
                 distinct: func.distinct,

src/binder/mod.rs

@@ -82,7 +82,9 @@ impl<S: Storage> Binder<S> {
     pub async fn bind(mut self, stmt: &Statement) -> Result<LogicalPlan, BindError> {
         let plan = match stmt {
             Statement::Query(query) => self.bind_query(query).await?,
-            Statement::CreateTable { name, columns, .. } => self.bind_create_table(name, &columns)?,
+            Statement::CreateTable { name, columns, constraints, .. } => {
+                self.bind_create_table(name, &columns, &constraints)?
+            },
             Statement::Drop { object_type, names, .. } => {
                 match object_type {
                     ObjectType::Table => {
@@ -168,9 +170,9 @@ pub enum BindError {
     SubqueryMustHaveAlias,
     #[error("agg miss: {0}")]
     AggMiss(String),
-    #[error("catalog error")]
+    #[error("catalog error: {0}")]
     CatalogError(#[from] CatalogError),
-    #[error("type error")]
+    #[error("type error: {0}")]
     TypeError(#[from] TypeError)
 }
 
@@ -193,16 +195,16 @@ pub mod test {
         let _ = storage.create_table(
             Arc::new("t1".to_string()),
             vec![
-                ColumnCatalog::new("c1".to_string(), false, ColumnDesc::new(Integer, true, false)),
-                ColumnCatalog::new("c2".to_string(), false, ColumnDesc::new(Integer, false, true)),
+                ColumnCatalog::new("c1".to_string(), false, ColumnDesc::new(Integer, true, false), None),
+                ColumnCatalog::new("c2".to_string(), false, ColumnDesc::new(Integer, false, true), None),
             ]
         ).await?;
 
         let _ = storage.create_table(
             Arc::new("t2".to_string()),
             vec![
-                ColumnCatalog::new("c3".to_string(), false, ColumnDesc::new(Integer, true, false)),
-                ColumnCatalog::new("c4".to_string(), false, ColumnDesc::new(Integer, false, false)),
+                ColumnCatalog::new("c3".to_string(), false, ColumnDesc::new(Integer, true, false), None),
+                ColumnCatalog::new("c4".to_string(), false, ColumnDesc::new(Integer, false, false), None),
             ]
         ).await?;
 

src/catalog/column.rs

@@ -2,6 +2,7 @@ use std::sync::Arc;
 use serde::{Deserialize, Serialize};
 use sqlparser::ast::{ColumnDef, ColumnOption};
 use crate::catalog::TableName;
+use crate::expression::ScalarExpression;
 
 use crate::types::{ColumnId, LogicalType};
 
@@ -14,16 +15,23 @@ pub struct ColumnCatalog {
     pub table_name: Option<TableName>,
     pub nullable: bool,
     pub desc: ColumnDesc,
+    pub ref_expr: Option<ScalarExpression>,
 }
 
 impl ColumnCatalog {
-    pub(crate) fn new(column_name: String, nullable: bool, column_desc: ColumnDesc) -> ColumnCatalog {
+    pub(crate) fn new(
+        column_name: String,
+        nullable: bool,
+        column_desc: ColumnDesc,
+        ref_expr: Option<ScalarExpression>
+    ) -> ColumnCatalog {
         ColumnCatalog {
             id: None,
             name: column_name,
             table_name: None,
             nullable,
             desc: column_desc,
+            ref_expr,
         }
     }
 
@@ -34,6 +42,7 @@ impl ColumnCatalog {
             table_name: None,
             nullable: false,
             desc: ColumnDesc::new(LogicalType::Varchar(None), false, false),
+            ref_expr: None,
         }
     }
 
@@ -75,7 +84,7 @@ impl From<ColumnDef> for ColumnCatalog {
             }
         }
 
-        ColumnCatalog::new(column_name, nullable, column_desc)
+        ColumnCatalog::new(column_name, nullable, column_desc, None)
     }
 }
 

src/catalog/root.rs

@@ -68,11 +68,13 @@ mod tests {
             "a".to_string(),
             false,
             ColumnDesc::new(LogicalType::Integer, false, false),
+            None
         );
         let col1 = ColumnCatalog::new(
             "b".to_string(),
             false,
             ColumnDesc::new(LogicalType::Boolean, false, false),
+            None
         );
         let col_catalogs = vec![col0, col1];
 

src/catalog/table.rs

@@ -23,10 +23,12 @@ impl TableCatalog {
             .find(|meta| meta.is_unique && &meta.column_ids[0] == col_id)
     }
 
+    #[allow(dead_code)]
     pub(crate) fn get_column_by_id(&self, id: &ColumnId) -> Option<&ColumnRef> {
         self.columns.get(id)
     }
 
+    #[allow(dead_code)]
     pub(crate) fn get_column_id_by_name(&self, name: &String) -> Option<ColumnId> {
         self.column_idxs.get(name).cloned()
     }
@@ -123,8 +125,8 @@ mod tests {
     // | 1         | true     |
     // | 2         | false    |
     fn test_table_catalog() {
-        let col0 = ColumnCatalog::new("a".into(), false, ColumnDesc::new(LogicalType::Integer, false, false));
-        let col1 = ColumnCatalog::new("b".into(), false, ColumnDesc::new(LogicalType::Boolean, false, false));
+        let col0 = ColumnCatalog::new("a".into(), false, ColumnDesc::new(LogicalType::Integer, false, false), None);
+        let col1 = ColumnCatalog::new("b".into(), false, ColumnDesc::new(LogicalType::Boolean, false, false), None);
         let col_catalogs = vec![col0, col1];
         let table_catalog = TableCatalog::new(Arc::new("test".to_string()), col_catalogs).unwrap();
 

src/db.rs

@@ -168,12 +168,14 @@ mod test {
             ColumnCatalog::new(
                 "c1".to_string(),
                 false,
-                ColumnDesc::new(LogicalType::Integer, true, false)
+                ColumnDesc::new(LogicalType::Integer, true, false),
+                None
             ),
             ColumnCatalog::new(
                 "c2".to_string(),
                 false,
-                ColumnDesc::new(LogicalType::Boolean, false, false)
+                ColumnDesc::new(LogicalType::Boolean, false, false),
+                None
             ),
         ];
 

src/execution/executor/dql/aggregate/count.rs

@@ -6,7 +6,7 @@ use crate::execution::ExecutorError;
 use crate::types::value::{DataValue, ValueRef};
 
 pub struct CountAccumulator {
-    result: u32,
+    result: i32,
 }
 
 impl CountAccumulator {
@@ -25,7 +25,7 @@ impl Accumulator for CountAccumulator {
     }
 
     fn evaluate(&self) -> Result<ValueRef, ExecutorError> {
-        Ok(Arc::new(DataValue::UInt32(Some(self.result))))
+        Ok(Arc::new(DataValue::Int32(Some(self.result))))
     }
 }
 
@@ -51,6 +51,6 @@ impl Accumulator for DistinctCountAccumulator {
     }
 
     fn evaluate(&self) -> Result<ValueRef, ExecutorError> {
-        Ok(Arc::new(DataValue::UInt32(Some(self.distinct_values.len() as u32))))
+        Ok(Arc::new(DataValue::Int32(Some(self.distinct_values.len() as i32))))
     }
 }

src/execution/executor/dql/aggregate/hash_agg.rs

@@ -125,9 +125,9 @@ mod test {
         let desc = ColumnDesc::new(LogicalType::Integer, false, false);
 
         let t1_columns = vec![
-            Arc::new(ColumnCatalog::new("c1".to_string(), true, desc.clone())),
-            Arc::new(ColumnCatalog::new("c2".to_string(), true, desc.clone())),
-            Arc::new(ColumnCatalog::new("c3".to_string(), true, desc.clone())),
+            Arc::new(ColumnCatalog::new("c1".to_string(), true, desc.clone(), None)),
+            Arc::new(ColumnCatalog::new("c2".to_string(), true, desc.clone(), None)),
+            Arc::new(ColumnCatalog::new("c3".to_string(), true, desc.clone(), None)),
         ];
 
         let operator = AggregateOperator {

src/execution/executor/dql/aggregate/sum.rs

@@ -23,7 +23,7 @@ impl SumAccumulator {
 impl Accumulator for SumAccumulator {
     fn update_value(&mut self, value: &ValueRef) -> Result<(), ExecutorError> {
         if !value.is_null() {
-            self.result  = binary_op(
+            self.result = binary_op(
                 &self.result,
                 value,
                 &BinaryOperator::Plus

src/execution/executor/dql/filter.rs

@@ -36,7 +36,7 @@ impl Filter {
         for tuple in input {
             let tuple = tuple?;
             if let DataValue::Boolean(option) = predicate.eval_column(&tuple)?.as_ref() {
-                if let Some(true) = option{
+                if let Some(true) = option {
                     yield tuple;
                 } else {
                     continue