Add update plan

src/query/service/src/interpreters/access/privilege_access.rs

@@ -149,6 +149,18 @@ impl AccessChecker for PrivilegeAccess {
             // Others.
             Plan::Insert(_) => {}
             Plan::Delete(_) => {}
+            Plan::Update(plan) => {
+                session
+                    .validate_privilege(
+                        &GrantObject::Table(
+                            plan.catalog.clone(),
+                            plan.database.clone(),
+                            plan.table.clone(),
+                        ),
+                        UserPrivilegeType::Update,
+                    )
+                    .await?;
+            }
             Plan::CreateView(plan) => {
                 session
                     .validate_privilege(

src/query/service/src/interpreters/interpreter_factory.rs

@@ -183,6 +183,8 @@ impl InterpreterFactory {
                 *delete.clone(),
             )?)),
 
+            Plan::Update(_update) => todo!(),
+
             // Roles
             Plan::CreateRole(create_role) => Ok(Arc::new(CreateRoleInterpreter::try_create(
                 ctx,

src/query/service/src/sql/planner/binder/binder.rs

@@ -212,7 +212,7 @@ impl<'a> Binder {
                 self.bind_delete(bind_context, table_reference, selection)
                     .await?
             }
-            Statement::Update(stmt) => todo!(),
+            Statement::Update(stmt) => self.bind_update(bind_context, stmt).await?,
 
             // Permissions
             Statement::Grant(stmt) => self.bind_grant(stmt).await?,

src/query/service/src/sql/planner/binder/mod.rs

@@ -35,6 +35,7 @@ mod setting;
 mod show;
 mod sort;
 mod table;
+mod update;
 
 pub use aggregate::AggregateInfo;
 pub use bind_context::*;

src/query/service/src/sql/planner/binder/update.rs

@@ -0,0 +1,97 @@
+// Copyright 2022 Datafuse Labs.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::collections::HashMap;
+
+use common_ast::ast::UpdateStmt;
+use common_exception::ErrorCode;
+use common_exception::Result;
+
+use crate::sql::binder::Binder;
+use crate::sql::binder::ScalarBinder;
+use crate::sql::normalize_identifier;
+use crate::sql::plans::Plan;
+use crate::sql::plans::Update;
+use crate::sql::BindContext;
+
+impl<'a> Binder {
+    pub(in crate::sql::planner::binder) async fn bind_update(
+        &mut self,
+        bind_context: &BindContext,
+        stmt: &UpdateStmt<'a>,
+    ) -> Result<Plan> {
+        let UpdateStmt {
+            catalog,
+            database,
+            table,
+            update_list,
+            selection,
+        } = stmt;
+        let catalog_name = catalog.as_ref().map_or_else(
+            || self.ctx.get_current_catalog(),
+            |ident| normalize_identifier(ident, &self.name_resolution_ctx).name,
+        );
+        let database_name = database.as_ref().map_or_else(
+            || self.ctx.get_current_database(),
+            |ident| normalize_identifier(ident, &self.name_resolution_ctx).name,
+        );
+        let table_name = normalize_identifier(table, &self.name_resolution_ctx).name;
+
+        let table = self
+            .ctx
+            .get_table(&catalog_name, &database_name, &table_name)
+            .await?;
+        let table_id = table.get_id();
+
+        let mut scalar_binder = ScalarBinder::new(
+            bind_context,
+            self.ctx.clone(),
+            &self.name_resolution_ctx,
+            self.metadata.clone(),
+            &[],
+        );
+        let schema = table.schema();
+        let mut update_columns = HashMap::with_capacity(update_list.len());
+        for update_expr in update_list {
+            let col_name = normalize_identifier(&update_expr.name, &self.name_resolution_ctx).name;
+            let index = schema.index_of(&col_name)?;
+            if update_columns.contains_key(&index) {
+                return Err(ErrorCode::BadArguments(format!(
+                    "Multiple assignments in the single statement to column `{}`",
+                    col_name
+                )));
+            }
+
+            let (scalar, _) = scalar_binder.bind(&update_expr.expr).await?;
+            update_columns.insert(index, scalar);
+        }
+
+        let push_downs = if let Some(expr) = selection {
+            let (scalar, _) = scalar_binder.bind(expr).await?;
+            Some(scalar)
+        } else {
+            None
+        };
+
+        let plan = Update {
+            catalog: catalog_name,
+            database: database_name,
+            table: table_name,
+            table_id,
+            update_list: update_columns,
+            selection: push_downs,
+        };
+        Ok(Plan::Update(Box::new(plan)))
+    }
+}

src/query/service/src/sql/planner/format/display_plan.rs

@@ -68,6 +68,7 @@ impl Plan {
             // Insert
             Plan::Insert(insert) => Ok(format!("{:?}", insert)),
             Plan::Delete(delete) => Ok(format!("{:?}", delete)),
+            Plan::Update(update) => Ok(format!("{:?}", update)),
 
             // Stages
             Plan::ListStage(s) => Ok(format!("{:?}", s)),

src/query/service/src/sql/planner/plans/mod.rs

@@ -33,6 +33,7 @@ mod scalar;
 pub mod share;
 mod sort;
 mod union_all;
+mod update;
 
 use std::fmt::Display;
 use std::sync::Arc;
@@ -113,6 +114,7 @@ pub use share::*;
 pub use sort::Sort;
 pub use sort::SortItem;
 pub use union_all::UnionAll;
+pub use update::Update;
 
 use super::BindContext;
 use crate::sql::optimizer::SExpr;
@@ -169,6 +171,7 @@ pub enum Plan {
     // Insert
     Insert(Box<Insert>),
     Delete(Box<DeletePlan>),
+    Update(Box<Update>),
 
     // Views
     CreateView(Box<CreateViewPlan>),
@@ -284,6 +287,7 @@ impl Display for Plan {
             Plan::DropUDF(_) => write!(f, "DropUDF"),
             Plan::Insert(_) => write!(f, "Insert"),
             Plan::Delete(_) => write!(f, "Delete"),
+            Plan::Update(_) => write!(f, "Update"),
             Plan::Call(_) => write!(f, "Call"),
             Plan::Presign(_) => write!(f, "Presign"),
             Plan::SetVariable(_) => write!(f, "SetVariable"),
@@ -357,6 +361,7 @@ impl Plan {
             Plan::DropUDF(_) => Arc::new(DataSchema::empty()),
             Plan::Insert(plan) => plan.schema(),
             Plan::Delete(_) => Arc::new(DataSchema::empty()),
+            Plan::Update(_) => Arc::new(DataSchema::empty()),
             Plan::Call(_) => Arc::new(DataSchema::empty()),
             Plan::Presign(plan) => plan.schema(),
             Plan::SetVariable(plan) => plan.schema(),

src/query/service/src/sql/planner/plans/update.rs

@@ -0,0 +1,29 @@
+// Copyright 2022 Datafuse Labs.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::collections::HashMap;
+
+use common_meta_types::MetaId;
+
+use crate::sql::plans::Scalar;
+
+#[derive(Clone, Debug)]
+pub struct Update {
+    pub catalog: String,
+    pub database: String,
+    pub table: String,
+    pub table_id: MetaId,
+    pub update_list: HashMap<usize, Scalar>,
+    pub selection: Option<Scalar>,
+}