fix: many bugs

- when `Project` is pushed down, `InputRefs` loses its related `ColumnRef` information
- when the `Project` is pushed down to Scan, the `Alias` content may not be `ColumnRef`, causing problems.
- `Binary` expression implements `output_columns`
- `max_logic_type` returns the `DateTime` type for a combination of `DateTime` and `VarChar`
- `raw_len` of `DateTime` returns incorrectly

Author: KKould

src/db.rs

@@ -263,6 +263,10 @@ mod test {
         let tuples_time_max = kipsql.run("select max(e) as max_time from t2").await?;
         println!("{}", create_table(&tuples_time_max));
 
+        println!("time where:");
+        let tuples_time_where_t2 = kipsql.run("select (c + 1) from t2 where e > '2021-05-20 21:00:00'").await?;
+        println!("{}", create_table(&tuples_time_where_t2));
+
         assert!(kipsql.run("select max(d) from t2 group by c").await.is_err());
 
         println!("distinct t1:");

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

@@ -48,7 +48,7 @@ impl HashAggExecutor {
                 self.agg_calls
                     .iter()
                     .chain(self.groupby_exprs.iter())
-                    .map(|expr| expr.output_column(&tuple))
+                    .map(|expr| expr.output_columns(&tuple))
                     .collect_vec()
             });
 

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

@@ -42,7 +42,7 @@ impl SimpleAggExecutor {
             columns_option.get_or_insert_with(|| {
                 self.agg_calls
                     .iter()
-                    .map(|expr| expr.output_column(&tuple))
+                    .map(|expr| expr.output_columns(&tuple))
                     .collect_vec()
             });
 

src/execution/executor/dql/projection.rs

@@ -40,7 +40,7 @@ impl Projection {
 
             for expr in exprs.iter() {
                 values.push(expr.eval_column(&tuple));
-                columns.push(expr.output_column(&tuple));
+                columns.push(expr.output_columns(&tuple));
             }
 
             yield Tuple { id: None, columns, values, };

src/expression/mod.rs

@@ -1,3 +1,5 @@
+use std::fmt;
+use std::fmt::Formatter;
 use std::sync::Arc;
 use itertools::Itertools;
 
@@ -59,6 +61,14 @@ pub enum ScalarExpression {
 }
 
 impl ScalarExpression {
+    pub fn unpack_alias(&self) -> &ScalarExpression {
+        if let ScalarExpression::Alias { expr, .. } = self {
+            expr.unpack_alias()
+        } else {
+            self
+        }
+    }
+
     pub fn nullable(&self) -> bool {
         match self {
             ScalarExpression::Constant(_) => false,
@@ -154,14 +164,14 @@ impl ScalarExpression {
         }
     }
 
-    pub fn output_column(&self, tuple: &Tuple) -> ColumnRef {
+    pub fn output_columns(&self, tuple: &Tuple) -> ColumnRef {
         match self {
             ScalarExpression::ColumnRef(col) => {
                 col.clone()
             }
             ScalarExpression::Constant(value) => {
                 Arc::new(ColumnCatalog::new(
-                    String::new(),
+                    format!("{}", value),
                     true,
                     ColumnDesc::new(value.logical_type(), false)
                 ))
@@ -175,7 +185,7 @@ impl ScalarExpression {
             }
             ScalarExpression::AggCall { kind, args, ty, distinct } => {
                 let args_str = args.iter()
-                    .map(|expr| expr.output_column(tuple).name.clone())
+                    .map(|expr| expr.output_columns(tuple).name.clone())
                     .join(", ");
                 let op = |allow_distinct, distinct| {
                     if allow_distinct && distinct {
@@ -200,6 +210,25 @@ impl ScalarExpression {
             ScalarExpression::InputRef { index, .. } => {
                 tuple.columns[*index].clone()
             }
+            ScalarExpression::Binary {
+                left_expr,
+                right_expr,
+                op,
+                ty
+            } => {
+                let column_name = format!(
+                    "({} {} {})",
+                    left_expr.output_columns(tuple).name,
+                    op,
+                    right_expr.output_columns(tuple).name,
+                );
+
+                Arc::new(ColumnCatalog::new(
+                    column_name,
+                    true,
+                    ColumnDesc::new(ty.clone(), false)
+                ))
+            }
             _ => unreachable!()
         }
     }
@@ -238,9 +267,29 @@ pub enum BinaryOperator {
     And,
     Or,
     Xor,
-    BitwiseOr,
-    BitwiseAnd,
-    BitwiseXor,
+}
+
+impl fmt::Display for BinaryOperator {
+    fn fmt(&self, f: &mut Formatter) -> fmt::Result {
+        match self {
+            BinaryOperator::Plus => write!(f, "+"),
+            BinaryOperator::Minus => write!(f, "-"),
+            BinaryOperator::Multiply => write!(f, "*"),
+            BinaryOperator::Divide => write!(f, "/"),
+            BinaryOperator::Modulo => write!(f, "mod"),
+            BinaryOperator::StringConcat => write!(f, "&"),
+            BinaryOperator::Gt => write!(f, ">"),
+            BinaryOperator::Lt => write!(f, "<"),
+            BinaryOperator::GtEq => write!(f, ">="),
+            BinaryOperator::LtEq => write!(f, "<="),
+            BinaryOperator::Spaceship => write!(f, "<=>"),
+            BinaryOperator::Eq => write!(f, "="),
+            BinaryOperator::NotEq => write!(f, "!="),
+            BinaryOperator::And => write!(f, "&&"),
+            BinaryOperator::Or => write!(f, "||"),
+            BinaryOperator::Xor => write!(f, "^"),
+        }
+    }
 }
 
 impl From<SqlBinaryOperator> for BinaryOperator {
@@ -262,9 +311,6 @@ impl From<SqlBinaryOperator> for BinaryOperator {
             SqlBinaryOperator::And => BinaryOperator::And,
             SqlBinaryOperator::Or => BinaryOperator::Or,
             SqlBinaryOperator::Xor => BinaryOperator::Xor,
-            SqlBinaryOperator::BitwiseOr => BinaryOperator::BitwiseOr,
-            SqlBinaryOperator::BitwiseAnd => BinaryOperator::BitwiseAnd,
-            SqlBinaryOperator::BitwiseXor => BinaryOperator::BitwiseXor,
             _ => todo!()
         }
     }

src/optimizer/rule/column_pruning.rs

@@ -51,10 +51,7 @@ impl Rule for PushProjectIntoScan {
 
                 new_scan_op.columns = project_op.columns
                     .iter()
-                    .filter(|expr| matches!(
-                        expr,
-                        ScalarExpression::ColumnRef(_) | ScalarExpression::Alias { .. }
-                    ))
+                    .filter(|expr| matches!(expr.unpack_alias(),ScalarExpression::ColumnRef(_)))
                     .cloned()
                     .collect_vec();
 
@@ -82,16 +79,27 @@ impl Rule for PushProjectThroughChild {
 
         if let Operator::Project(_) = node_operator {
             let child_index = graph.children_at(node_id)[0];
-            let node_referenced_columns = node_operator.referenced_columns();
+            let mut node_referenced_columns = node_operator.referenced_columns();
+            let child_operator = graph.operator(child_index);
+            let child_referenced_columns = child_operator.referenced_columns();
+            let is_child_agg = matches!(child_operator, Operator::Aggregate(_));
+
+            // When the aggregate function is a child node,
+            // the pushdown will lose the corresponding ColumnRef due to `InputRef`.
+            // Therefore, it is necessary to map the InputRef to the corresponding ColumnRef
+            // and push it down.
+            if is_child_agg && !input_refs.is_empty() {
+                node_referenced_columns.append(
+                    &mut child_operator.agg_mapping_col_refs(&input_refs)
+                )
+            }
 
-            let intersection_columns_ids = graph
-                .operator(child_index)
-                .referenced_columns()
-                .into_iter()
+            let intersection_columns_ids = child_referenced_columns
+                .iter()
                 .map(|col| col.id)
                 .chain(
                     node_referenced_columns
-                        .into_iter()
+                        .iter()
                         .map(|col| col.id)
                 )
                 .collect::<HashSet<ColumnId>>();
@@ -101,25 +109,28 @@ impl Rule for PushProjectThroughChild {
             }
 
             for grandson_id in graph.children_at(child_index) {
-                let columns = graph.operator(grandson_id)
+                let mut columns = graph.operator(grandson_id)
                     .referenced_columns()
                     .into_iter()
                     .unique_by(|col| col.id)
                     .filter(|u| intersection_columns_ids.contains(&u.id))
                     .map(|col| ScalarExpression::ColumnRef(col))
                     .collect_vec();
-                // Tips: Aggregation fields take precedence
-                let full_columns = input_refs.iter()
-                    .cloned()
-                    .chain(columns)
-                    .collect_vec();
 
-                if !full_columns.is_empty() {
+                if !is_child_agg && !input_refs.is_empty() {
+                    // Tips: Aggregation InputRefs fields take precedence
+                    columns = input_refs.iter()
+                        .cloned()
+                        .chain(columns)
+                        .collect_vec();
+                }
+
+                if !columns.is_empty() {
                     graph.add_node(
                         child_index,
                         Some(grandson_id),
                         OptExprNode::OperatorRef(
-                            Operator::Project(ProjectOperator { columns: full_columns })
+                            Operator::Project(ProjectOperator { columns })
                         )
                     );
                 }

src/planner/operator/mod.rs

@@ -58,6 +58,7 @@ impl Operator {
             Operator::Project(op) => {
                 op.columns
                     .iter()
+                    .map(ScalarExpression::unpack_alias)
                     .filter(|expr| matches!(expr, ScalarExpression::InputRef { .. }))
                     .cloned()
                     .collect_vec()
@@ -66,6 +67,25 @@ impl Operator {
         }
     }
 
+    pub fn agg_mapping_col_refs(&self, input_refs: &[ScalarExpression]) -> Vec<ColumnRef> {
+        match self {
+            Operator::Aggregate(AggregateOperator { agg_calls, .. }) => {
+                input_refs.iter()
+                    .filter_map(|expr| {
+                        if let ScalarExpression::InputRef { index, .. } = expr {
+                            Some(agg_calls[*index].clone())
+                        } else {
+                            None
+                        }
+                    })
+                    .map(|expr| expr.referenced_columns())
+                    .flatten()
+                    .collect_vec()
+            }
+            _ => vec![],
+        }
+    }
+
     pub fn referenced_columns(&self) -> Vec<ColumnRef> {
         match self {
             Operator::Aggregate(op) => {

src/storage/kip.rs

@@ -218,7 +218,7 @@ impl Transaction for KipTraction<'_> {
 
                             for expr in self.projections.iter() {
                                 values.push(expr.eval_column(&tuple));
-                                columns.push(expr.output_column(&tuple));
+                                columns.push(expr.output_columns(&tuple));
                             }
 
                             self.limit = self.limit.map(|num| num - 1);

src/storage/memory.rs

@@ -220,7 +220,7 @@ impl Transaction for MemTraction<'_> {
 
                 for expr in self.projections.iter() {
                     values.push(expr.eval_column(&tuple));
-                    columns.push(expr.output_column(&tuple));
+                    columns.push(expr.output_columns(&tuple));
                 }
 
                 self.limit = self.limit.map(|num| num - 1);

src/types/mod.rs

@@ -75,7 +75,7 @@ impl LogicalType {
             LogicalType::Float => Some(4),
             LogicalType::Double => Some(8),
             LogicalType::Varchar => None,
-            LogicalType::Date => Some(4),
+            LogicalType::Date => Some(8),
         }
     }
 
@@ -146,6 +146,9 @@ impl LogicalType {
         if left.is_numeric() && right.is_numeric() {
             return LogicalType::combine_numeric_types(left, right);
         }
+        if matches!((left, right), (LogicalType::Date, LogicalType::Varchar) | (LogicalType::Varchar, LogicalType::Date)) {
+            return Ok(LogicalType::Date);
+        }
         Err(TypeError::InternalError(format!(
             "can not compare two types: {:?} and {:?}",
             left, right