Merge pull request #8824 from zhyass/feature_cluster

refactor(storage): deletion

Author: BohuTANG

src/query/catalog/src/table.rs

@@ -34,7 +34,6 @@ use crate::plan::DataSourcePlan;
 use crate::plan::Expression;
 use crate::plan::PartStatistics;
 use crate::plan::Partitions;
-use crate::plan::Projection;
 use crate::plan::PushDownInfo;
 use crate::table::column_stats_provider_impls::DummyColumnStatisticsProvider;
 use crate::table_context::TableContext;
@@ -227,10 +226,11 @@ pub trait Table: Sync + Send {
     async fn delete(
         &self,
         ctx: Arc<dyn TableContext>,
-        projection: &Projection,
-        selection: &Option<String>,
+        filter: Option<Expression>,
+        col_indices: Vec<usize>,
+        pipeline: &mut Pipeline,
     ) -> Result<()> {
-        let (_, _, _) = (ctx, projection, selection);
+        let (_, _, _, _) = (ctx, filter, col_indices, pipeline);
 
         Err(ErrorCode::Unimplemented(format!(
             "table {},  of engine type {}, does not support DELETE FROM",

src/query/service/src/interpreters/interpreter_delete.rs

@@ -16,12 +16,17 @@ use std::sync::Arc;
 
 use common_datavalues::DataSchemaRef;
 use common_exception::Result;
+use common_pipeline_core::Pipeline;
+use common_sql::executor::ExpressionBuilderWithoutRenaming;
 use common_sql::plans::DeletePlan;
 
 use crate::interpreters::Interpreter;
+use crate::pipelines::executor::ExecutorSettings;
+use crate::pipelines::executor::PipelineCompleteExecutor;
 use crate::pipelines::PipelineBuildResult;
 use crate::sessions::QueryContext;
 use crate::sessions::TableContext;
+use crate::sql::plans::ScalarExpr;
 
 /// interprets DeletePlan
 pub struct DeleteInterpreter {
@@ -55,12 +60,29 @@ impl Interpreter for DeleteInterpreter {
         let db_name = self.plan.database_name.as_str();
         let tbl_name = self.plan.table_name.as_str();
         let tbl = self.ctx.get_table(catalog_name, db_name, tbl_name).await?;
-        tbl.delete(
-            self.ctx.clone(),
-            &self.plan.projection,
-            &self.plan.selection,
-        )
-        .await?;
+        let (filter, col_indices) = if let Some(scalar) = &self.plan.selection {
+            let eb = ExpressionBuilderWithoutRenaming::create(self.plan.metadata.clone());
+            (
+                Some(eb.build(scalar)?),
+                scalar.used_columns().into_iter().collect(),
+            )
+        } else {
+            (None, vec![])
+        };
+
+        let mut pipeline = Pipeline::create();
+        tbl.delete(self.ctx.clone(), filter, col_indices, &mut pipeline)
+            .await?;
+        if !pipeline.pipes.is_empty() {
+            let settings = self.ctx.get_settings();
+            pipeline.set_max_threads(settings.get_max_threads()? as usize);
+            let executor_settings = ExecutorSettings::try_create(&settings)?;
+            let executor = PipelineCompleteExecutor::try_create(pipeline, executor_settings)?;
+
+            self.ctx.set_executor(Arc::downgrade(&executor.get_inner()));
+            executor.execute()?;
+            drop(executor);
+        }
 
         Ok(PipelineBuildResult::create())
     }

src/query/service/tests/it/storages/fuse/operations/mutation/deletion_mutator.rs

@@ -12,117 +12,103 @@
 //  See the License for the specific language governing permissions and
 //  limitations under the License.
 
-use std::collections::hash_map::RandomState;
-use std::collections::HashMap;
-use std::collections::HashSet;
 use std::sync::Arc;
 
 use common_base::base::tokio;
-use common_datablocks::BlockCompactThresholds;
-use common_datablocks::DataBlock;
-use common_datavalues::DataSchema;
-use common_exception::ErrorCode;
 use common_exception::Result;
-use common_storages_table_meta::caches::CacheManager;
-use common_storages_table_meta::meta::BlockMeta;
-use common_storages_table_meta::meta::SegmentInfo;
-use common_storages_table_meta::meta::Statistics;
-use common_storages_table_meta::meta::TableSnapshot;
-use common_storages_table_meta::meta::Versioned;
+use common_sql::executor::ExpressionBuilderWithoutRenaming;
+use common_sql::plans::DeletePlan;
+use common_sql::plans::Plan;
+use common_sql::plans::ScalarExpr;
+use common_sql::Planner;
+use common_storages_factory::Table;
+use common_storages_fuse::FuseTable;
+use databend_query::pipelines::executor::ExecutorSettings;
+use databend_query::pipelines::executor::PipelineCompleteExecutor;
+use databend_query::sessions::QueryContext;
 use databend_query::sessions::TableContext;
-use databend_query::storages::fuse::io::SegmentWriter;
-use databend_query::storages::fuse::io::TableMetaLocationGenerator;
-use databend_query::storages::fuse::operations::DeletionMutator;
-use databend_query::storages::fuse::statistics::ClusterStatsGenerator;
-use uuid::Uuid;
 
+use crate::storages::fuse::table_test_fixture::execute_command;
+use crate::storages::fuse::table_test_fixture::execute_query;
+use crate::storages::fuse::table_test_fixture::expects_ok;
 use crate::storages::fuse::table_test_fixture::TestFixture;
 
-/// [issue#6570](https://github.com/datafuselabs/databend/issues/6570)
-/// During deletion, there might be multiple segments become empty
-
 #[tokio::test]
 async fn test_deletion_mutator_multiple_empty_segments() -> Result<()> {
-    // generates a batch of segments, and delete blocks from them
-    // so that half of the segments will be empty
-
     let fixture = TestFixture::new().await;
     let ctx = fixture.ctx();
-    let location_generator = TableMetaLocationGenerator::with_prefix("_prefix".to_owned());
+    let tbl_name = fixture.default_table_name();
+    let db_name = fixture.default_db_name();
 
-    let segment_info_cache = CacheManager::instance().get_table_segment_cache();
-    let data_accessor = ctx.get_data_operator()?.operator();
-    let seg_writer = SegmentWriter::new(&data_accessor, &location_generator, &segment_info_cache);
+    fixture.create_normal_table().await?;
 
-    let gen_test_seg = || async {
-        // generates test segment, each of them contains only one block
-        // structures are filled with arbitrary values, no effects for this test case
-        let block_id = Uuid::new_v4().simple().to_string();
-        let location = (block_id, DataBlock::VERSION);
-        let test_block_meta = Arc::new(BlockMeta::new(
-            1,
-            1,
-            1,
-            HashMap::default(),
-            HashMap::default(),
-            None,
-            location.clone(),
-            None,
-            0,
-        ));
-        let segment = SegmentInfo::new(vec![test_block_meta], Statistics::default());
-        Ok::<_, ErrorCode>((seg_writer.write_segment(segment).await?, location))
-    };
+    // insert
+    for i in 0..10 {
+        let qry = format!("insert into {}.{}(id) values({})", db_name, tbl_name, i);
+        execute_command(ctx.clone(), qry.as_str()).await?;
+    }
 
-    // generates 100 segments, for each segment, contains one block
-    let mut test_segment_locations = vec![];
-    let mut test_block_locations = vec![];
-    for _ in 0..100 {
-        let (segment_location, block_location) = gen_test_seg().await?;
-        test_segment_locations.push(segment_location);
-        test_block_locations.push(block_location);
+    let catalog = ctx.get_catalog(fixture.default_catalog_name().as_str())?;
+    let table = catalog
+        .get_table(ctx.get_tenant().as_str(), &db_name, &tbl_name)
+        .await?;
+    // delete
+    let query = format!("delete from {}.{} where id=1", db_name, tbl_name);
+    let mut planner = Planner::new(ctx.clone());
+    let (plan, _, _) = planner.plan_sql(&query).await?;
+    if let Plan::Delete(delete) = plan {
+        do_deletion(ctx.clone(), table.clone(), *delete).await?;
     }
 
-    let base_snapshot = TableSnapshot::new(
-        Uuid::new_v4(),
-        &None,
-        None,
-        DataSchema::empty(),
-        Statistics::default(),
-        test_segment_locations.clone(),
-        None,
-        None,
+    // check count
+    let expected = vec![
+        "+---------------+-------+",
+        "| segment_count | count |",
+        "+---------------+-------+",
+        "| 9             | 9     |",
+        "+---------------+-------+",
+    ];
+    let qry = format!(
+        "select segment_count, block_count as count from fuse_snapshot('{}', '{}') limit 1",
+        db_name, tbl_name
     );
+    expects_ok(
+        "check segment and block count",
+        execute_query(fixture.ctx(), qry.as_str()).await,
+        expected,
+    )
+    .await?;
+    Ok(())
+}
 
-    let table_ctx: Arc<dyn TableContext> = ctx as Arc<dyn TableContext>;
-    let mut mutator = DeletionMutator::try_create(
-        table_ctx,
-        data_accessor.clone(),
-        location_generator,
-        Arc::new(base_snapshot),
-        ClusterStatsGenerator::default(),
-        BlockCompactThresholds::default(),
-    )?;
+pub async fn do_deletion(
+    ctx: Arc<QueryContext>,
+    table: Arc<dyn Table>,
+    plan: DeletePlan,
+) -> Result<()> {
+    let (filter, col_indices) = if let Some(scalar) = &plan.selection {
+        let eb = ExpressionBuilderWithoutRenaming::create(plan.metadata.clone());
+        (
+            Some(eb.build(scalar)?),
+            scalar.used_columns().into_iter().collect(),
+        )
+    } else {
+        (None, vec![])
+    };
 
-    // clear half of the segments
-    for (i, _) in test_segment_locations.iter().enumerate().take(100) {
-        if i % 2 == 0 {
-            // empty the segment (segment only contains one block)
-            mutator
-                .replace_with(i, test_block_locations[i].clone(), None, DataBlock::empty())
-                .await?;
-        }
+    let fuse_table = FuseTable::try_from_table(table.as_ref())?;
+    let settings = ctx.get_settings();
+    let mut pipeline = common_pipeline_core::Pipeline::create();
+    fuse_table
+        .delete(ctx.clone(), filter, col_indices, &mut pipeline)
+        .await?;
+    if !pipeline.pipes.is_empty() {
+        pipeline.set_max_threads(settings.get_max_threads()? as usize);
+        let executor_settings = ExecutorSettings::try_create(&settings)?;
+        let executor = PipelineCompleteExecutor::try_create(pipeline, executor_settings)?;
+        ctx.set_executor(Arc::downgrade(&executor.get_inner()));
+        executor.execute()?;
+        drop(executor);
     }
-
-    let (segments, _, _) = mutator.generate_segments().await?;
-
-    // half segments left after deletion
-    assert_eq!(segments.len(), 50);
-
-    // new_segments should be a subset of test_segments in our case (no partial deletion of segment)
-    let new_segments = HashSet::<_, RandomState>::from_iter(segments.into_iter());
-    let test_segments = HashSet::from_iter(test_segment_locations.into_iter());
-    assert!(new_segments.is_subset(&test_segments));
-
     Ok(())
 }

src/query/service/tests/it/storages/fuse/operations/mutation/mod.rs

@@ -16,3 +16,5 @@ mod block_compact_mutator;
 mod deletion_mutator;
 mod recluster_mutator;
 mod segments_compact_mutator;
+
+pub use deletion_mutator::do_deletion;

src/query/service/tests/it/storages/fuse/operations/mutation/recluster_mutator.rs

@@ -124,8 +124,15 @@ async fn test_recluster_mutator_block_select() -> Result<()> {
         segments_location,
     )
     .await?;
-    let mut blocks_map = BTreeMap::new();
-    blocks_map.insert(0, block_metas);
+    let mut blocks_map: BTreeMap<i32, Vec<(usize, Arc<BlockMeta>)>> = BTreeMap::new();
+    block_metas.iter().for_each(|(idx, b)| {
+        if let Some(stats) = &b.cluster_stats {
+            blocks_map
+                .entry(stats.level)
+                .or_default()
+                .push((idx.0, b.clone()));
+        }
+    });
 
     let mut mutator = ReclusterMutator::try_create(
         ctx,

src/query/service/tests/it/storages/fuse/operations/table_analyze.rs

@@ -26,6 +26,7 @@ use databend_query::sql::plans::Plan;
 use databend_query::sql::Planner;
 use futures_util::TryStreamExt;
 
+use crate::storages::fuse::operations::mutation::do_deletion;
 use crate::storages::fuse::table_test_fixture::execute_command;
 use crate::storages::fuse::table_test_fixture::execute_query;
 use crate::storages::fuse::table_test_fixture::TestFixture;
@@ -74,9 +75,7 @@ async fn test_table_modify_column_ndv_statistics() -> Result<()> {
     let mut planner = Planner::new(ctx.clone());
     let (plan, _, _) = planner.plan_sql(query).await?;
     if let Plan::Delete(delete) = plan {
-        table
-            .delete(ctx.clone(), &delete.projection, &delete.selection)
-            .await?;
+        do_deletion(ctx.clone(), table.clone(), *delete).await?;
     }
     execute_command(ctx.clone(), statistics_sql).await?;
 

src/query/service/tests/it/storages/fuse/statistics.rs

@@ -19,7 +19,7 @@ use common_datablocks::BlockCompactThresholds;
 use common_datablocks::DataBlock;
 use common_datavalues::prelude::*;
 use common_functions::aggregates::eval_aggr;
-use common_functions::scalars::FunctionContext;
+use common_sql::evaluator::ChunkOperator;
 use common_sql::evaluator::Evaluator;
 use common_sql::executor::add;
 use common_sql::executor::col;
@@ -220,14 +220,14 @@ fn test_ft_stats_cluster_stats() -> common_exception::Result<()> {
     ]);
 
     let block_compactor = BlockCompactThresholds::new(1_000_000, 800_000, 100 * 1024 * 1024);
-    let stats_gen = ClusterStatsGenerator::new(0, vec![0], vec![], 0, block_compactor);
+    let stats_gen = ClusterStatsGenerator::new(0, vec![0], vec![], 0, block_compactor, vec![]);
     let (stats, _) = stats_gen.gen_stats_for_append(&blocks)?;
     assert!(stats.is_some());
     let stats = stats.unwrap();
     assert_eq!(vec![DataValue::Int64(1)], stats.min);
     assert_eq!(vec![DataValue::Int64(3)], stats.max);
 
-    let stats_gen = ClusterStatsGenerator::new(1, vec![1], vec![], 0, block_compactor);
+    let stats_gen = ClusterStatsGenerator::new(1, vec![1], vec![], 0, block_compactor, vec![]);
     let (stats, _) = stats_gen.gen_stats_for_append(&blocks)?;
     assert!(stats.is_some());
     let stats = stats.unwrap();
@@ -249,7 +249,7 @@ async fn test_ft_cluster_stats_with_stats() -> common_exception::Result<()> {
     });
 
     let block_compactor = BlockCompactThresholds::new(1_000_000, 800_000, 100 * 1024 * 1024);
-    let stats_gen = ClusterStatsGenerator::new(0, vec![0], vec![], 0, block_compactor);
+    let stats_gen = ClusterStatsGenerator::new(0, vec![0], vec![], 0, block_compactor, vec![]);
     let stats = stats_gen.gen_with_origin_stats(&blocks, origin.clone())?;
     assert!(stats.is_some());
     let stats = stats.unwrap();
@@ -258,22 +258,21 @@ async fn test_ft_cluster_stats_with_stats() -> common_exception::Result<()> {
 
     // add expression executor.
     let expr = add(col("a", i32::to_data_type()), lit(1));
-    let eval_node = Evaluator::eval_expression(&expr, &schema)?;
-    let func_ctx = FunctionContext::default();
-    let result = eval_node.eval(&func_ctx, &blocks)?;
-    let output_schema =
-        DataSchemaRefExt::create(vec![DataField::new("(a + 1)", i64::to_data_type())]);
-    let blocks = DataBlock::create(output_schema, vec![result.vector]);
-
-    let stats_gen = ClusterStatsGenerator::new(0, vec![0], vec![], 0, block_compactor);
+    let field = DataField::new("(a + 1)", i64::to_data_type());
+    let operators = vec![ChunkOperator::Map {
+        eval: Evaluator::eval_expression(&expr, &schema)?,
+        name: field.name().to_string(),
+    }];
+
+    let stats_gen = ClusterStatsGenerator::new(0, vec![1], vec![], 0, block_compactor, operators);
     let stats = stats_gen.gen_with_origin_stats(&blocks, origin.clone())?;
     assert!(stats.is_some());
     let stats = stats.unwrap();
     assert_eq!(vec![DataValue::Int64(2)], stats.min);
     assert_eq!(vec![DataValue::Int64(4)], stats.max);
 
     // different cluster_key_id.
-    let stats_gen = ClusterStatsGenerator::new(1, vec![0], vec![], 0, block_compactor);
+    let stats_gen = ClusterStatsGenerator::new(1, vec![0], vec![], 0, block_compactor, vec![]);
     let stats = stats_gen.gen_with_origin_stats(&blocks, origin)?;
     assert!(stats.is_none());