From 24ab8c29e44d05f71aaab13b2cdd29bd844d175c Mon Sep 17 00:00:00 2001 From: Xuanwo Date: Mon, 19 Sep 2022 11:54:20 +0800 Subject: [PATCH 1/3] Add DeletePlanV2 Signed-off-by: Xuanwo --- .../src/sql/planner/plans/delete_v2.rs | 37 +++++++++++++++++++ .../service/src/sql/planner/plans/mod.rs | 1 + 2 files changed, 38 insertions(+) create mode 100644 src/query/service/src/sql/planner/plans/delete_v2.rs diff --git a/src/query/service/src/sql/planner/plans/delete_v2.rs b/src/query/service/src/sql/planner/plans/delete_v2.rs new file mode 100644 index 0000000000000..75b4c58f40e85 --- /dev/null +++ b/src/query/service/src/sql/planner/plans/delete_v2.rs @@ -0,0 +1,37 @@ +// 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::BTreeMap; +use std::sync::Arc; + +use common_datavalues::DataSchema; +use common_datavalues::DataSchemaRef; +use common_legacy_planners::Projection; +use common_meta_types::MetaId; + +#[derive(serde::Serialize, serde::Deserialize, Clone, Debug, PartialEq)] +pub struct DeletePlanV2 { + pub catalog_name: String, + pub database_name: String, + pub table_name: String, + pub table_id: MetaId, + pub selection: Option, + pub projection: Projection, +} + +impl DeletePlanV2 { + pub fn schema(&self) -> DataSchemaRef { + Arc::new(DataSchema::empty()) + } +} diff --git a/src/query/service/src/sql/planner/plans/mod.rs b/src/query/service/src/sql/planner/plans/mod.rs index bcce440896206..c1313de2b48ac 100644 --- a/src/query/service/src/sql/planner/plans/mod.rs +++ b/src/query/service/src/sql/planner/plans/mod.rs @@ -15,6 +15,7 @@ mod aggregate; mod copy_v2; pub mod create_table_v2; +mod delete_v2; mod dummy_table_scan; mod eval_scalar; mod exchange; From 3670612c1e8f7b323802138cb7e3be77cb9b2463 Mon Sep 17 00:00:00 2001 From: Xuanwo Date: Mon, 19 Sep 2022 13:09:05 +0800 Subject: [PATCH 2/3] Use String in DeletePlan Signed-off-by: Xuanwo --- src/meta/api/src/schema_api_test_suite.rs | 18 +++--- src/meta/app/src/schema/table.rs | 2 +- src/query/legacy-planners/src/plan_delete.rs | 20 ++++++- .../src/catalogs/default/immutable_catalog.rs | 2 +- .../service/src/sql/planner/binder/delete.rs | 57 ++++--------------- .../src/sql/planner/plans/delete_v2.rs | 37 ------------ .../service/src/sql/planner/plans/mod.rs | 1 - .../src/sql/statements/statement_delete.rs | 4 +- .../storages/fuse/src/operations/delete.rs | 10 +++- 9 files changed, 50 insertions(+), 101 deletions(-) delete mode 100644 src/query/service/src/sql/planner/plans/delete_v2.rs diff --git a/src/meta/api/src/schema_api_test_suite.rs b/src/meta/api/src/schema_api_test_suite.rs index b2c6f464c13d6..4a63b990eba33 100644 --- a/src/meta/api/src/schema_api_test_suite.rs +++ b/src/meta/api/src/schema_api_test_suite.rs @@ -1522,7 +1522,7 @@ impl SchemaApiTestSuite { let ident = TableIdent::new(tb_id, seq); let want = TableInfo { - ident: ident.clone(), + ident, desc: format!("'{}'.'{}'.'{}'", tenant, db_name, tbl_name), name: tbl_name.into(), meta: table_meta(created_on), @@ -1549,7 +1549,7 @@ impl SchemaApiTestSuite { let got = mt.get_table((tenant, db_name, tbl_name).into()).await?; let want = TableInfo { - ident: tb_ident_2.clone(), + ident: tb_ident_2, desc: format!("'{}'.'{}'.'{}'", tenant, db_name, tbl_name), name: tbl_name.into(), meta: table_meta(created_on), @@ -1584,7 +1584,7 @@ impl SchemaApiTestSuite { let got = mt.get_table((tenant, "db1", "tb2").into()).await.unwrap(); let want = TableInfo { - ident: tb_ident_2.clone(), + ident: tb_ident_2, desc: format!("'{}'.'{}'.'{}'", tenant, db_name, tbl_name), name: tbl_name.into(), meta: table_meta(created_on), @@ -1784,7 +1784,7 @@ impl SchemaApiTestSuite { let cur_db = mt.get_database(Self::req_get_db(tenant, db1_name)).await?; assert!(old_db.ident.seq < cur_db.ident.seq); let got = mt.get_table((tenant, db1_name, tb2_name).into()).await?; - got.ident.clone() + got.ident }; info!("--- rename table, ok"); @@ -1796,7 +1796,7 @@ impl SchemaApiTestSuite { let got = mt.get_table((tenant, db1_name, tb3_name).into()).await?; let want = TableInfo { - ident: tb_ident.clone(), + ident: tb_ident, desc: format!("'{}'.'{}'.'{}'", tenant, db1_name, tb3_name), name: tb3_name.into(), meta: table_meta(created_on), @@ -1842,7 +1842,7 @@ impl SchemaApiTestSuite { let got = mt.get_table((tenant, db1_name, tb2_name).into()).await?; assert_ne!(tb_ident.table_id, got.ident.table_id); assert_ne!(tb_ident.seq, got.ident.seq); - got.ident.clone() + got.ident }; info!("--- db1,tb2(no_nil) -> db1,tb3(no_nil), error"); @@ -2010,7 +2010,7 @@ impl SchemaApiTestSuite { let ident = TableIdent::new(tb_id, seq); let want = TableInfo { - ident: ident.clone(), + ident, desc: format!("'{}'.'{}'.'{}'", tenant, db_name, tbl_name), name: tbl_name.into(), meta: table_meta(created_on), @@ -2140,7 +2140,7 @@ impl SchemaApiTestSuite { let ident = TableIdent::new(tb_id, seq); let want = TableInfo { - ident: ident.clone(), + ident, desc: format!("'{}'.'{}'.'{}'", tenant, db_name, tbl_name), name: tbl_name.into(), meta: table_meta(created_on), @@ -3128,7 +3128,7 @@ impl SchemaApiTestSuite { let ident = TableIdent::new(tb_id, seq); let want = TableInfo { - ident: ident.clone(), + ident, desc: format!("'{}'.'{}'.'{}'", tenant, db_name, tbl_name), name: tbl_name.into(), meta: table_meta(created_on), diff --git a/src/meta/app/src/schema/table.rs b/src/meta/app/src/schema/table.rs index 7357883bf028d..cbc1597d150a7 100644 --- a/src/meta/app/src/schema/table.rs +++ b/src/meta/app/src/schema/table.rs @@ -29,7 +29,7 @@ use maplit::hashmap; use crate::schema::database::DatabaseNameIdent; /// Globally unique identifier of a version of TableMeta. -#[derive(serde::Serialize, serde::Deserialize, Clone, Debug, Eq, PartialEq, Default)] +#[derive(serde::Serialize, serde::Deserialize, Clone, Copy, Debug, Eq, PartialEq, Default)] pub struct TableIdent { /// Globally unique id to identify a table. pub table_id: u64, diff --git a/src/query/legacy-planners/src/plan_delete.rs b/src/query/legacy-planners/src/plan_delete.rs index 1c62d9bd6c537..3f530d104be40 100644 --- a/src/query/legacy-planners/src/plan_delete.rs +++ b/src/query/legacy-planners/src/plan_delete.rs @@ -18,16 +18,30 @@ use common_datavalues::DataSchema; use common_datavalues::DataSchemaRef; use common_meta_app::schema::TableIdent; -use crate::Expression; use crate::Projection; -#[derive(serde::Serialize, serde::Deserialize, Clone, Debug, PartialEq)] +/// # TODO +/// +/// From @xuanwo +/// +/// Ideally, we need to use `Scalar` in DeletePlan.selection. But we met a +/// cycle deps here. So we have to change `selection` in String first, and +/// change into `Scalar` when our `Planner` has been moved out. +/// +/// At this stage, DeletePlan's selection expr will be parsed twice: +/// +/// - Parsed during `bind` to get column index and projection index. +/// - Parsed during `execution` to get the correct columns +/// +/// It's an ugly but necessary price to pay. Without this, we would sink in +/// hell forever. +#[derive(serde::Serialize, serde::Deserialize, Clone, Debug, PartialEq, Eq)] pub struct DeletePlan { pub catalog_name: String, pub database_name: String, pub table_name: String, pub table_id: TableIdent, - pub selection: Option, + pub selection: Option, pub projection: Projection, } diff --git a/src/query/service/src/catalogs/default/immutable_catalog.rs b/src/query/service/src/catalogs/default/immutable_catalog.rs index f7fce84745f30..1f5938bb7350c 100644 --- a/src/query/service/src/catalogs/default/immutable_catalog.rs +++ b/src/query/service/src/catalogs/default/immutable_catalog.rs @@ -130,7 +130,7 @@ impl Catalog for ImmutableCatalog { .get_by_id(&table_id) .ok_or_else(|| ErrorCode::UnknownTable(format!("Unknown table id: '{}'", table_id)))?; let ti = table.get_table_info(); - Ok((ti.ident.clone(), Arc::new(ti.meta.clone()))) + Ok((ti.ident, Arc::new(ti.meta.clone()))) } async fn get_table( diff --git a/src/query/service/src/sql/planner/binder/delete.rs b/src/query/service/src/sql/planner/binder/delete.rs index 53298a7f60f31..4747ff7b3332a 100644 --- a/src/query/service/src/sql/planner/binder/delete.rs +++ b/src/query/service/src/sql/planner/binder/delete.rs @@ -12,39 +12,18 @@ // See the License for the specific language governing permissions and // limitations under the License. -use std::collections::HashSet; - use common_ast::ast::Expr; use common_ast::ast::TableReference; use common_exception::ErrorCode; use common_exception::Result; use common_legacy_planners::DeletePlan; -use common_legacy_planners::Expression; use common_legacy_planners::Projection; use crate::sql::binder::Binder; use crate::sql::binder::ScalarBinder; -use crate::sql::executor::ExpressionBuilderWithoutRenaming; use crate::sql::plans::Plan; -use crate::sql::statements::query::QueryASTIRVisitor; use crate::sql::BindContext; - -pub struct DeleteCollectPushDowns {} -impl QueryASTIRVisitor> for DeleteCollectPushDowns { - fn visit_expr(expr: &mut Expression, require_columns: &mut HashSet) -> Result<()> { - if let Expression::Column(name) = expr { - if !require_columns.contains(name) { - require_columns.insert(name.clone()); - } - } - - Ok(()) - } - - fn visit_filter(predicate: &mut Expression, data: &mut HashSet) -> Result<()> { - Self::visit_recursive_expr(predicate, data) - } -} +use crate::sql::ScalarExpr; impl<'a> Binder { pub(in crate::sql::planner::binder) async fn bind_delete( @@ -88,36 +67,22 @@ impl<'a> Binder { &[], ); - let mut expression = None; - let mut require_columns: HashSet = HashSet::new(); - if let Some(expr) = selection { - let (scalar, _) = scalar_binder.bind(expr).await?; - let eb = ExpressionBuilderWithoutRenaming::create(self.metadata.clone()); - let mut pred_expr = eb.build(&scalar)?; - DeleteCollectPushDowns::visit_filter(&mut pred_expr, &mut require_columns)?; - expression = Some(pred_expr); - } - let table = self .ctx .get_table(&catalog_name, &database_name, &table_name) .await?; let tbl_info = table.get_table_info(); - let table_id = tbl_info.ident.clone(); - let mut col_indices = vec![]; - let schema = tbl_info.meta.schema.as_ref(); - for col_name in require_columns { - if let Some((idx, _)) = schema.column_with_name(col_name.as_str()) { - col_indices.push(idx); - } else { - return Err(ErrorCode::UnknownColumn(format!( - "Column [{}] not found", - col_name - ))); - } - } + let table_id = tbl_info.ident; + // @todo wait delete migrate to new planner + let col_indices: Vec = if let Some(expr) = selection { + let (scalar, _) = scalar_binder.bind(expr).await?; + scalar.used_columns().into_iter().collect() + } else { + vec![] + }; + let selection = selection.as_ref().map(|expr| expr.to_string()); let projection = Projection::Columns(col_indices); let plan = DeletePlan { @@ -125,7 +90,7 @@ impl<'a> Binder { database_name, table_name, table_id, - selection: expression, + selection, projection, }; Ok(Plan::Delete(Box::new(plan))) diff --git a/src/query/service/src/sql/planner/plans/delete_v2.rs b/src/query/service/src/sql/planner/plans/delete_v2.rs deleted file mode 100644 index 75b4c58f40e85..0000000000000 --- a/src/query/service/src/sql/planner/plans/delete_v2.rs +++ /dev/null @@ -1,37 +0,0 @@ -// 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::BTreeMap; -use std::sync::Arc; - -use common_datavalues::DataSchema; -use common_datavalues::DataSchemaRef; -use common_legacy_planners::Projection; -use common_meta_types::MetaId; - -#[derive(serde::Serialize, serde::Deserialize, Clone, Debug, PartialEq)] -pub struct DeletePlanV2 { - pub catalog_name: String, - pub database_name: String, - pub table_name: String, - pub table_id: MetaId, - pub selection: Option, - pub projection: Projection, -} - -impl DeletePlanV2 { - pub fn schema(&self) -> DataSchemaRef { - Arc::new(DataSchema::empty()) - } -} diff --git a/src/query/service/src/sql/planner/plans/mod.rs b/src/query/service/src/sql/planner/plans/mod.rs index c1313de2b48ac..bcce440896206 100644 --- a/src/query/service/src/sql/planner/plans/mod.rs +++ b/src/query/service/src/sql/planner/plans/mod.rs @@ -15,7 +15,6 @@ mod aggregate; mod copy_v2; pub mod create_table_v2; -mod delete_v2; mod dummy_table_scan; mod eval_scalar; mod exchange; diff --git a/src/query/service/src/sql/statements/statement_delete.rs b/src/query/service/src/sql/statements/statement_delete.rs index a188f3178f7dd..bbab4a0af94e7 100644 --- a/src/query/service/src/sql/statements/statement_delete.rs +++ b/src/query/service/src/sql/statements/statement_delete.rs @@ -79,12 +79,12 @@ impl AnalyzableStatement for DfDeleteStatement { let selection = if let Some(predicate) = &self.selection { let mut pred_expr = analyzer.analyze(predicate).await?; DeleteCollectPushDowns::visit_filter(&mut pred_expr, &mut require_columns)?; - Some(pred_expr) + Some(predicate.to_string()) } else { None }; - let table_id = tbl_info.ident.clone(); + let table_id = tbl_info.ident; let mut col_indices = vec![]; let schema = tbl_info.meta.schema.as_ref(); for col_name in require_columns { diff --git a/src/query/storages/fuse/src/operations/delete.rs b/src/query/storages/fuse/src/operations/delete.rs index 040e96beb9a82..0f957a55a0118 100644 --- a/src/query/storages/fuse/src/operations/delete.rs +++ b/src/query/storages/fuse/src/operations/delete.rs @@ -17,8 +17,10 @@ use std::sync::Arc; use common_catalog::table::Table; use common_catalog::table_context::TableContext; use common_datavalues::DataSchemaRefExt; +use common_exception::ErrorCode; use common_exception::Result; use common_fuse_meta::meta::TableSnapshot; +use common_legacy_parser::ExpressionParser; use common_legacy_planners::DeletePlan; use common_legacy_planners::Expression; use common_legacy_planners::Extras; @@ -51,7 +53,13 @@ impl FuseTable { // check if unconditional deletion if let Some(filter) = &plan.selection { - self.delete_rows(ctx, &snapshot, filter, plan).await + let expr = ExpressionParser::parse_exprs(filter)?; + if expr.is_empty() { + return Err(ErrorCode::IndexOutOfBounds( + "expression should be valid, but not", + )); + } + self.delete_rows(ctx, &snapshot, &expr[0], plan).await } else { // deleting the whole table... just a truncate let purge = false; From 8873c5d12bda229bfe21922e7fdfacd9c6314c66 Mon Sep 17 00:00:00 2001 From: Xuanwo Date: Mon, 19 Sep 2022 14:07:39 +0800 Subject: [PATCH 3/3] Fix exprs Signed-off-by: Xuanwo --- src/query/service/src/sql/planner/binder/delete.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/query/service/src/sql/planner/binder/delete.rs b/src/query/service/src/sql/planner/binder/delete.rs index 4747ff7b3332a..da6eb4b230c68 100644 --- a/src/query/service/src/sql/planner/binder/delete.rs +++ b/src/query/service/src/sql/planner/binder/delete.rs @@ -82,7 +82,7 @@ impl<'a> Binder { } else { vec![] }; - let selection = selection.as_ref().map(|expr| expr.to_string()); + let selection = selection.as_ref().map(|expr| format!("({})", expr)); let projection = Projection::Columns(col_indices); let plan = DeletePlan {