feat: Support the DELETE SQL statement

src/catalog/src/system.rs

@@ -211,6 +211,9 @@ pub(crate) fn build_table_deletion_request(
     let table_key = format_table_entry_key(&request.catalog, &request.schema, table_id);
     DeleteRequest {
         key_column_values: build_primary_key_columns(EntryType::Table, table_key.as_bytes()),
+        catalog_name: request.catalog.clone(),
+        schema_name: request.schema.clone(),
+        table_name: request.table_name.clone(),
     }
 }
 

src/datanode/src/error.rs

@@ -114,6 +114,13 @@ pub enum Error {
         source: TableError,
     },
 
+    #[snafu(display("Failed to delete value to table: {}, source: {}", table_name, source))]
+    Delete {
+        table_name: String,
+        #[snafu(backtrace)]
+        source: TableError,
+    },
+
     #[snafu(display("Failed to start server, source: {}", source))]
     StartServer {
         #[snafu(backtrace)]
@@ -161,7 +168,7 @@ pub enum Error {
     },
 
     #[snafu(display("Invalid SQL, error: {}", msg))]
-    InvalidSql { msg: String, backtrace: Backtrace },
+    InvalidSql { msg: String },
 
     #[snafu(display("Failed to create schema when creating table, source: {}", source))]
     CreateSchema {
@@ -343,6 +350,7 @@ impl ErrorExt for Error {
             Error::DropTable { source, .. } => source.status_code(),
 
             Error::Insert { source, .. } => source.status_code(),
+            Error::Delete { source, .. } => source.status_code(),
 
             Error::TableNotFound { .. } => StatusCode::TableNotFound,
             Error::ColumnNotFound { .. } => StatusCode::TableColumnNotFound,

src/datanode/src/instance/sql.rs

@@ -64,6 +64,10 @@ impl Instance {
                 )?;
                 self.sql_handler.execute(request, query_ctx).await
             }
+            QueryStatement::Sql(Statement::Delete(d)) => {
+                let request = self.sql_handler.delete_to_request(*d, query_ctx.clone())?;
+                self.sql_handler.execute(request, query_ctx).await
+            }
 
             QueryStatement::Sql(Statement::CreateDatabase(c)) => {
                 let request = CreateDatabaseRequest {

src/datanode/src/sql.rs

@@ -31,6 +31,7 @@ use crate::instance::sql::table_idents_to_full_name;
 
 mod alter;
 mod create;
+mod delete;
 mod drop_table;
 mod insert;
 
@@ -45,6 +46,7 @@ pub enum SqlRequest {
     ShowTables(ShowTables),
     DescribeTable(DescribeTable),
     Explain(Box<Explain>),
+    Delete(DeleteRequest),
 }
 
 // Handler to execute SQL except query
@@ -78,6 +80,7 @@ impl SqlHandler {
             SqlRequest::CreateDatabase(req) => self.create_database(req).await,
             SqlRequest::Alter(req) => self.alter(req).await,
             SqlRequest::DropTable(req) => self.drop_table(req).await,
+            SqlRequest::Delete(req) => self.delete(req).await,
             SqlRequest::ShowDatabases(stmt) => {
                 show_databases(stmt, self.catalog_manager.clone()).context(ExecuteSqlSnafu)
             }

src/datanode/src/sql/delete.rs

@@ -0,0 +1,139 @@
+// Copyright 2023 Greptime Team
+//
+// 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 std::ops::Deref;
+use std::str::FromStr;
+use std::sync::Arc;
+
+use common_query::Output;
+use common_telemetry::warn;
+use datatypes::prelude::VectorRef;
+use datatypes::vectors::{BooleanVector, Float64Vector, StringVector, TimestampMillisecondVector};
+use session::context::QueryContextRef;
+use snafu::ResultExt;
+use sql::ast::{BinaryOperator, Expr, Value};
+use sql::statements::delete::Delete;
+use table::engine::TableReference;
+use table::requests::DeleteRequest;
+
+use crate::error::{DeleteSnafu, InvalidSqlSnafu, Result};
+use crate::instance::sql::table_idents_to_full_name;
+use crate::sql::{SqlHandler, SqlRequest};
+
+impl SqlHandler {
+    pub(crate) async fn delete(&self, req: DeleteRequest) -> Result<Output> {
+        let table_ref = TableReference {
+            catalog: &req.catalog_name.to_string(),
+            schema: &req.schema_name.to_string(),
+            table: &req.table_name.to_string(),
+        };
+
+        let table = self.get_table(&table_ref)?;
+
+        let affected_rows = table.delete(req).await.with_context(|_| DeleteSnafu {
+            table_name: table_ref.to_string(),
+        })?;
+
+        Ok(Output::AffectedRows(affected_rows))
+    }
+
+    pub(crate) fn delete_to_request(
+        &self,
+        stmt: Delete,
+        query_ctx: QueryContextRef,
+    ) -> Result<SqlRequest> {
+        let (catalog_name, schema_name, table_name) =
+            table_idents_to_full_name(stmt.table_name(), query_ctx)?;
+        let key_column_values = parser_selection(stmt.selection())?;
+        Ok(SqlRequest::Delete(DeleteRequest {
+            key_column_values,
+            catalog_name,
+            schema_name,
+            table_name,
+        }))
+    }
+}
+
+/// parse selection, currently supported format is `tagkey1 = 'tagvalue1' and tagkey2 = 'tagvalue2'`.
+/// (only uses =, and in the where clause and provides all columns needed by the key.)
+fn parser_selection(selection: &Option<Expr>) -> Result<HashMap<String, VectorRef>> {
+    let mut key_column_values = HashMap::new();
+    if let Some(expr) = selection {
+        parser_expr(expr, &mut key_column_values)?;
+    }
+    Ok(key_column_values)
+}
+
+fn parser_expr(expr: &Expr, key_column_values: &mut HashMap<String, VectorRef>) -> Result<()> {
+    // match BinaryOp
+    if let Expr::BinaryOp { left, op, right } = expr {
+        match (left.deref(), op, right.deref()) {
+            // match And operator
+            (Expr::BinaryOp { .. }, BinaryOperator::And, Expr::BinaryOp { .. }) => {
+                parser_expr(left.deref(), key_column_values)?;
+                parser_expr(right.deref(), key_column_values)?;
+                return Ok(());
+            }
+            // match Eq operator
+            (Expr::Identifier(column_name), BinaryOperator::Eq, Expr::Value(value)) => {
+                key_column_values.insert(
+                    column_name.to_string(),
+                    value_to_vector(&column_name.to_string(), value)?,
+                );
+                return Ok(());
+            }
+            (Expr::Identifier(column_name), BinaryOperator::Eq, Expr::Identifier(value)) => {
+                key_column_values.insert(
+                    column_name.to_string(),
+                    Arc::new(StringVector::from(vec![value.to_string()])),
+                );
+                return Ok(());
+            }
+            _ => {}
+        }
+    }
+    return InvalidSqlSnafu {
+        msg: format!("Failed to parse expr:{expr}"),
+    }
+    .fail();
+}
+
+/// parse value to vector
+fn value_to_vector(column_name: &String, value: &Value) -> Result<VectorRef> {
+    match value {
+        Value::Number(n, _) => {
+            if column_name == "ts" {
+                Ok(Arc::new(TimestampMillisecondVector::from_vec(vec![
+                    i64::from_str(n).unwrap(),
+                ])))
+            } else {
+                Ok(Arc::new(Float64Vector::from_vec(vec![
+                    f64::from_str(n).unwrap()
+                ])))
+            }
+        }
+        Value::SingleQuotedString(s) | sql::ast::Value::DoubleQuotedString(s) => {
+            Ok(Arc::new(StringVector::from(vec![s.to_string()])))
+        }
+        Value::Boolean(b) => Ok(Arc::new(BooleanVector::from(vec![*b]))),
+        _ => {
+            warn!("Current value type is not supported, value:{value}");
+            return InvalidSqlSnafu {
+                msg: format!("Failed to parse value:{value}"),
+            }
+            .fail();
+        }
+    }
+}

src/datanode/src/tests/instance_test.rs

@@ -639,6 +639,55 @@ async fn test_use_database() {
     check_output_stream(output, expected).await;
 }
 
+#[tokio::test(flavor = "multi_thread")]
+async fn test_delete() {
+    let instance = MockInstance::new("test_delete").await;
+
+    let output = execute_sql(
+        &instance,
+        r#"create table test_table(
+                            host string,
+                            ts timestamp,
+                            cpu double default 0,
+                            memory double,
+                            TIME INDEX (ts),
+                            PRIMARY KEY(host)
+                        ) engine=mito with(regions=1);"#,
+    )
+    .await;
+    assert!(matches!(output, Output::AffectedRows(0)));
+
+    let output = execute_sql(
+        &instance,
+        r#"insert into test_table(host, cpu, memory, ts) values
+                           ('host1', 66.6, 1024, 1655276557000),
+                           ('host2', 77.7,  2048, 1655276558000),
+                           ('host3', 88.8,  3072, 1655276559000)
+                           "#,
+    )
+    .await;
+    assert!(matches!(output, Output::AffectedRows(3)));
+
+    let output = execute_sql(
+        &instance,
+        "delete from test_table where host = host1 and ts = 1655276557000 ",
+    )
+    .await;
+    assert!(matches!(output, Output::AffectedRows(1)));
+
+    let output = execute_sql(&instance, "select * from test_table").await;
+    let expect = "\
++-------+---------------------+------+--------+
+| host  | ts                  | cpu  | memory |
++-------+---------------------+------+--------+
+| host2 | 2022-06-15T07:02:38 | 77.7 | 2048   |
+| host3 | 2022-06-15T07:02:39 | 88.8 | 3072   |
++-------+---------------------+------+--------+\
+"
+    .to_string();
+    check_output_stream(output, expect).await;
+}
+
 async fn execute_sql(instance: &MockInstance, sql: &str) -> Output {
     execute_sql_in_db(instance, sql, DEFAULT_SCHEMA_NAME).await
 }

src/frontend/src/error.rs

@@ -85,10 +85,7 @@ pub enum Error {
     },
 
     #[snafu(display("Invalid SQL, error: {}", err_msg))]
-    InvalidSql {
-        err_msg: String,
-        backtrace: Backtrace,
-    },
+    InvalidSql { err_msg: String },
 
     #[snafu(display("Illegal Frontend state: {}", err_msg))]
     IllegalFrontendState {

src/frontend/src/instance.rs

@@ -392,6 +392,7 @@ impl Instance {
             | Statement::Explain(_)
             | Statement::Query(_)
             | Statement::Insert(_)
+            | Statement::Delete(_)
             | Statement::Alter(_)
             | Statement::DropTable(_) => self.sql_handler.do_statement_query(stmt, query_ctx).await,
             Statement::Use(db) => self.handle_use(db, query_ctx),

src/mito/src/engine.rs

@@ -1317,7 +1317,12 @@ mod tests {
         let mut key_column_values = HashMap::with_capacity(2);
         key_column_values.insert("host".to_string(), del_hosts);
         key_column_values.insert("ts".to_string(), del_tss);
-        let del_req = DeleteRequest { key_column_values };
+        let del_req = DeleteRequest {
+            key_column_values,
+            catalog_name: "greptime".to_string(),
+            schema_name: "public".to_string(),
+            table_name: TABLE_NAME.to_string(),
+        };
         table.delete(del_req).await.unwrap();
 
         let session_ctx = SessionContext::new();

src/query/src/datafusion/planner.rs

@@ -94,6 +94,7 @@ where
             Statement::Query(qb) => self.query_to_plan(qb),
             Statement::Explain(explain) => self.explain_to_plan(explain),
             Statement::ShowTables(_)
+            | Statement::Delete(_)
             | Statement::ShowDatabases(_)
             | Statement::ShowCreateTable(_)
             | Statement::DescribeTable(_)

src/sql/src/ast.rs

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 pub use sqlparser::ast::{
-    ColumnDef, ColumnOption, ColumnOptionDef, DataType, Expr, Function, FunctionArg,
-    FunctionArgExpr, Ident, ObjectName, SqlOption, TableConstraint, TimezoneInfo, Value,
+    BinaryOperator, ColumnDef, ColumnOption, ColumnOptionDef, DataType, Expr, Function,
+    FunctionArg, FunctionArgExpr, Ident, ObjectName, SqlOption, TableConstraint, TimezoneInfo,
+    Value,
 };

src/sql/src/error.rs

@@ -73,7 +73,7 @@ pub enum Error {
     InvalidTimeIndex { msg: String, backtrace: Backtrace },
 
     #[snafu(display("Invalid SQL, error: {}", msg))]
-    InvalidSql { msg: String, backtrace: Backtrace },
+    InvalidSql { msg: String },
 
     #[snafu(display("Invalid column option, column name: {}, error: {}", name, msg))]
     InvalidColumnOption {

src/sql/src/parser.rs

@@ -88,6 +88,8 @@ impl<'a> ParserContext<'a> {
                         self.parse_show()
                     }
 
+                    Keyword::DELETE => self.parse_delete(),
+
                     Keyword::DESCRIBE | Keyword::DESC => {
                         self.parser.next_token();
                         self.parse_describe()

src/sql/src/parsers.rs

@@ -14,5 +14,6 @@
 
 mod alter_parser;
 pub(crate) mod create_parser;
+pub(crate) mod delete_parser;
 pub(crate) mod insert_parser;
 pub(crate) mod query_parser;