feat(frontend): bind project (#615)

* feat(binder): bind project.

* plan projection.

* support multiple tables.

* use reverse_map for BindContext.

* move plan_projection to select.rs.

* add bind_all_columns.

* add test.

* comment one test.

rust/frontend/src/binder/bind_context.rs

@@ -0,0 +1,34 @@
+use std::collections::HashMap;
+
+use risingwave_common::types::DataType;
+
+pub struct ColumnBinding {
+    pub table_name: String,
+    pub index: usize,
+    pub data_type: DataType,
+}
+
+impl ColumnBinding {
+    pub fn new(table_name: String, index: usize, data_type: DataType) -> Self {
+        ColumnBinding {
+            table_name,
+            index,
+            data_type,
+        }
+    }
+}
+
+pub struct BindContext {
+    // Mapping column name to `ColumnBinding`
+    pub columns: HashMap<String, Vec<ColumnBinding>>,
+}
+
+impl BindContext {
+    #[allow(clippy::new_without_default)]
+    pub fn new() -> Self {
+        BindContext {
+            // tables: HashMap::new(),
+            columns: HashMap::new(),
+        }
+    }
+}

rust/frontend/src/binder/expr/column.rs

@@ -0,0 +1,67 @@
+use risingwave_common::array::RwError;
+use risingwave_common::error::{ErrorCode, Result};
+use risingwave_sqlparser::ast::Ident;
+
+use crate::binder::Binder;
+use crate::expr::{ExprImpl, InputRef};
+
+impl Binder {
+    pub fn bind_column(&mut self, idents: &[Ident]) -> Result<ExprImpl> {
+        // TODO: check quote style of `ident`.
+        let (_schema_name, table_name, column_name) = match idents {
+            [column] => (None, None, &column.value),
+            [table, column] => (None, Some(&table.value), &column.value),
+            [schema, table, column] => (Some(&schema.value), Some(&table.value), &column.value),
+            _ => {
+                return Err(
+                    ErrorCode::InternalError(format!("Too many idents: {:?}", idents)).into(),
+                )
+            }
+        };
+        let columns = self.context.columns.get(column_name).ok_or_else(|| {
+            RwError::from(ErrorCode::ItemNotFound(format!(
+                "Invalid column: {}",
+                column_name
+            )))
+        })?;
+        match table_name {
+            Some(table_name) => {
+                match columns
+                    .iter()
+                    .find(|column| column.table_name == *table_name)
+                {
+                    Some(column) => Ok(ExprImpl::InputRef(Box::new(InputRef::new(
+                        column.index,
+                        column.data_type.clone(),
+                    )))),
+                    None => Err(
+                        ErrorCode::ItemNotFound(format!("Invalid table: {}", table_name)).into(),
+                    ),
+                }
+            }
+            None => {
+                if columns.len() > 1 {
+                    Err(ErrorCode::InternalError("Ambiguous column name".into()).into())
+                } else {
+                    Ok(ExprImpl::InputRef(Box::new(InputRef::new(
+                        columns[0].index,
+                        columns[0].data_type.clone(),
+                    ))))
+                }
+            }
+        }
+    }
+
+    pub fn bind_all_columns(&mut self) -> Result<Vec<ExprImpl>> {
+        let mut bound_columns = vec![];
+        self.context.columns.values().for_each(|columns| {
+            columns.iter().for_each(|column| {
+                bound_columns.push(ExprImpl::InputRef(Box::new(InputRef::new(
+                    column.index,
+                    column.data_type.clone(),
+                ))));
+            });
+        });
+        Ok(bound_columns)
+    }
+}

rust/frontend/src/binder/expr/mod.rs

@@ -5,11 +5,14 @@ use crate::binder::Binder;
 use crate::expr::ExprImpl;
 
 mod binary_op;
+mod column;
 mod value;
 
 impl Binder {
     pub(super) fn bind_expr(&mut self, expr: Expr) -> Result<ExprImpl> {
         match expr {
+            Expr::Identifier(ident) => self.bind_column(&[ident]),
+            Expr::CompoundIdentifier(idents) => self.bind_column(&idents),
             Expr::Value(v) => Ok(ExprImpl::Literal(Box::new(self.bind_value(v)?))),
             Expr::BinaryOp { left, op, right } => Ok(ExprImpl::FunctionCall(Box::new(
                 self.bind_binary_op(*left, op, *right)?,

rust/frontend/src/binder/mod.rs

@@ -3,15 +3,18 @@ use std::sync::Arc;
 use risingwave_common::error::Result;
 use risingwave_sqlparser::ast::Statement;
 
+mod bind_context;
 mod expr;
 mod insert;
+mod projection;
 mod query;
 mod select;
 mod set_expr;
 mod statement;
 mod table_ref;
 mod values;
 
+pub use bind_context::BindContext;
 pub use insert::BoundInsert;
 pub use query::BoundQuery;
 pub use select::BoundSelect;
@@ -25,11 +28,17 @@ use crate::catalog::database_catalog::DatabaseCatalog;
 pub struct Binder {
     #[allow(dead_code)]
     catalog: Arc<DatabaseCatalog>,
+
+    // TODO: support subquery.
+    context: BindContext,
 }
 
 impl Binder {
     pub fn new(catalog: Arc<DatabaseCatalog>) -> Binder {
-        Binder { catalog }
+        Binder {
+            catalog,
+            context: BindContext::new(),
+        }
     }
     pub fn bind(&mut self, stmt: Statement) -> Result<BoundStatement> {
         self.bind_statement(stmt)

rust/frontend/src/binder/projection.rs

@@ -0,0 +1,25 @@
+use risingwave_common::error::Result;
+use risingwave_sqlparser::ast::SelectItem;
+
+use crate::binder::Binder;
+use crate::expr::ExprImpl;
+
+impl Binder {
+    pub fn bind_projection(&mut self, projection: Vec<SelectItem>) -> Result<Vec<ExprImpl>> {
+        let mut select_list = vec![];
+        for item in projection {
+            match item {
+                SelectItem::UnnamedExpr(expr) => {
+                    let expr = self.bind_expr(expr)?;
+                    select_list.push(expr);
+                }
+                SelectItem::ExprWithAlias { .. } => todo!(),
+                SelectItem::QualifiedWildcard(_) => todo!(),
+                SelectItem::Wildcard => {
+                    select_list.extend(self.bind_all_columns()?.into_iter());
+                }
+            }
+        }
+        Ok(select_list)
+    }
+}

rust/frontend/src/binder/select.rs

@@ -15,9 +15,10 @@ pub struct BoundSelect {
 impl Binder {
     pub(super) fn bind_select(&mut self, select: Select) -> Result<BoundSelect> {
         let from = self.bind_vec_table_with_joins(select.from)?;
+        let projection = self.bind_projection(select.projection)?;
         Ok(BoundSelect {
             distinct: select.distinct,
-            projection: vec![],
+            projection,
             from,
             selection: None,
         })

rust/frontend/src/binder/table_ref.rs

@@ -1,6 +1,7 @@
 use risingwave_common::error::{ErrorCode, Result};
 use risingwave_sqlparser::ast::{ObjectName, TableFactor, TableWithJoins};
 
+use super::bind_context::ColumnBinding;
 use crate::binder::Binder;
 use crate::catalog::catalog_service::DEFAULT_SCHEMA_NAME;
 use crate::catalog::column_catalog::ColumnCatalog;
@@ -53,10 +54,24 @@ impl Binder {
             .get_schema(&schema_name)
             .and_then(|c| c.get_table(&table_name))
             .ok_or_else(|| ErrorCode::ItemNotFound(format!("relation \"{}\"", table_name)))?;
+        let columns = table_catalog.columns().to_vec();
+
+        columns.iter().enumerate().for_each(|(index, column)| {
+            self.context
+                .columns
+                .entry(column.name().to_string())
+                .or_default()
+                .push(ColumnBinding::new(
+                    table_name.clone(),
+                    index,
+                    column.data_type(),
+                ))
+        });
+
         Ok(BaseTableRef {
             name: table_name,
             table_id: table_catalog.id(),
-            columns: table_catalog.columns().into(),
+            columns,
         })
     }
 }

rust/frontend/src/optimizer/plan_node/logical_project.rs

@@ -95,8 +95,11 @@ impl PlanTreeNodeUnary for LogicalProject {
 impl_plan_tree_node_for_unary! {LogicalProject}
 
 impl fmt::Display for LogicalProject {
-    fn fmt(&self, _f: &mut fmt::Formatter) -> fmt::Result {
-        todo!()
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        f.debug_struct("LogicalProject")
+            .field("exprs", self.exprs())
+            .field("expr_alias", &format_args!("{:?}", self.expr_alias()))
+            .finish()
     }
 }
 

rust/frontend/src/planner/select.rs

@@ -1,15 +1,17 @@
 use risingwave_common::error::Result;
 
 use crate::binder::BoundSelect;
-use crate::optimizer::plan_node::PlanRef;
+use crate::expr::ExprImpl;
+use crate::optimizer::plan_node::{LogicalProject, PlanRef};
 use crate::planner::Planner;
 
 impl Planner {
     pub(super) fn plan_select(&mut self, select: BoundSelect) -> Result<PlanRef> {
-        let root = match select.from {
+        let mut root = match select.from {
             None => self.create_dummy_values()?,
             Some(t) => self.plan_table_ref(t)?,
         };
+        root = self.plan_projection(root, select.projection)?;
         // mut root with LogicalFilter and LogicalProject here
         Ok(root)
     }
@@ -22,4 +24,10 @@ impl Planner {
     fn create_dummy_values(&self) -> Result<PlanRef> {
         todo!()
     }
+
+    fn plan_projection(&mut self, input: PlanRef, projection: Vec<ExprImpl>) -> Result<PlanRef> {
+        // TODO: support alias.
+        let expr_alias = vec![None; projection.len()];
+        Ok(LogicalProject::create(input, projection, expr_alias))
+    }
 }

rust/frontend/tests/testdata/basic_query.yaml

@@ -5,15 +5,24 @@
 - sql: select * from t
   binder_error: "Item not found: relation \"t\""
 
-- sql: |
-    create table t (v1 bigint, v2 double precision);
-    select * from t;
-  plan: |
-    LogicalScan { table: "t", columns: ["_row_id", "v1", "v2"] }
+# Because the order in which these column appear in `exprs` is random, I comment this test. I 
+# - sql: |
+#     create table t (v1 bigint, v2 double precision);
+#     select * from t;
+#   plan: |
+#     LogicalProject { exprs: [InputRef(1), InputRef(0), InputRef(2)], expr_alias: [None, None, None] }
+#       LogicalScan { table: "t", columns: ["_row_id", "v1", "v2"] }
 
 - sql: |
     create table t (v1 int, v2 int);
     insert into t values (22, 33), (44, 55);
   plan: |
     LogicalInsert { table_name: t, columns: [] }
       LogicalValues { rows: [[Literal(Literal { data: Some(Int32(22)), data_type: Int32 }), Literal(Literal { data: Some(Int32(33)), data_type: Int32 })], [Literal(Literal { data: Some(Int32(44)), data_type: Int32 }), Literal(Literal { data: Some(Int32(55)), data_type: Int32 })]], schema: Schema { fields: [Field { name = , data_type = Int32 }, Field { name = , data_type = Int32 }] } }
+
+- sql: |
+    create table t (v1 int, v2 int);
+    select v1 from t;
+  plan: |
+    LogicalProject { exprs: [InputRef(1)], expr_alias: [None] }
+      LogicalScan { table: "t", columns: ["_row_id", "v1", "v2"] }