revert compact and recluster

Author: SkyFan2002

src/query/service/src/pipelines/builders/builder_commit.rs

@@ -20,9 +20,9 @@ use databend_common_sql::executor::physical_plans::CommitSink as PhysicalCommitS
 use databend_common_sql::executor::physical_plans::CommitType;
 use databend_common_sql::executor::physical_plans::MutationKind;
 use databend_common_sql::plans::TruncateMode;
-use databend_common_storages_fuse::operations::add_table_mutation_aggregator;
 use databend_common_storages_fuse::operations::CommitSink;
 use databend_common_storages_fuse::operations::MutationGenerator;
+use databend_common_storages_fuse::operations::TableMutationAggregator;
 use databend_common_storages_fuse::operations::TransformMergeCommitMeta;
 use databend_common_storages_fuse::operations::TruncateGenerator;
 use databend_common_storages_fuse::FuseTable;
@@ -81,28 +81,30 @@ impl PipelineBuilder {
                         TransformMergeCommitMeta::create(cluster_key_id)
                     });
                 } else {
-                    let base_segments = if matches!(
-                        kind,
-                        MutationKind::Compact | MutationKind::Insert | MutationKind::Recluster
-                    ) {
-                        vec![]
-                    } else {
-                        plan.snapshot.segments().to_vec()
-                    };
+                    self.main_pipeline.add_async_accumulating_transformer(|| {
+                        let base_segments = if matches!(
+                            kind,
+                            MutationKind::Compact | MutationKind::Insert | MutationKind::Recluster
+                        ) {
+                            vec![]
+                        } else {
+                            plan.snapshot.segments().to_vec()
+                        };
 
-                    // extract re-cluster related mutations from physical plan
-                    let recluster_info = plan.recluster_info.clone().unwrap_or_default();
-                    add_table_mutation_aggregator(
-                        &mut self.main_pipeline,
-                        table,
-                        self.ctx.clone(),
-                        base_segments,
-                        recluster_info.merged_blocks,
-                        recluster_info.removed_segment_indexes,
-                        recluster_info.removed_statistics,
-                        *kind,
-                        plan.table_meta_timestamps,
-                    );
+                        // extract re-cluster related mutations from physical plan
+                        let recluster_info = plan.recluster_info.clone().unwrap_or_default();
+
+                        TableMutationAggregator::create(
+                            table,
+                            self.ctx.clone(),
+                            base_segments,
+                            recluster_info.merged_blocks,
+                            recluster_info.removed_segment_indexes,
+                            recluster_info.removed_statistics,
+                            *kind,
+                            plan.table_meta_timestamps,
+                        )
+                    });
                 }
 
                 let snapshot_gen = MutationGenerator::new(plan.snapshot.clone(), *kind);

src/query/service/src/pipelines/builders/builder_compact.rs

@@ -26,14 +26,11 @@ use databend_common_pipeline_transforms::processors::TransformPipelineHelper;
 use databend_common_sql::executor::physical_plans::CompactSource as PhysicalCompactSource;
 use databend_common_sql::executor::physical_plans::MutationKind;
 use databend_common_sql::StreamContext;
-use databend_common_storages_fuse::io::read::ColumnOrientedSegmentReader;
-use databend_common_storages_fuse::io::read::RowOrientedSegmentReader;
-use databend_common_storages_fuse::operations::add_table_mutation_aggregator;
 use databend_common_storages_fuse::operations::BlockCompactMutator;
-use databend_common_storages_fuse::operations::ColumnOrientedSegmentsWithIndices;
-use databend_common_storages_fuse::operations::CompactSegmentsWithIndices;
+use databend_common_storages_fuse::operations::CompactLazyPartInfo;
 use databend_common_storages_fuse::operations::CompactSource;
 use databend_common_storages_fuse::operations::CompactTransform;
+use databend_common_storages_fuse::operations::TableMutationAggregator;
 use databend_common_storages_fuse::operations::TransformSerializeBlock;
 use databend_common_storages_fuse::FuseTable;
 
@@ -57,60 +54,39 @@ impl PipelineBuilder {
         let thresholds = table.get_block_thresholds();
         let cluster_key_id = table.cluster_key_id();
         let mut max_threads = self.ctx.get_settings().get_max_threads()? as usize;
-        let partitions = compact_block.parts.partitions.clone();
 
         if is_lazy {
             let query_ctx = self.ctx.clone();
-            let column_ids = compact_block.column_ids.clone();
-            let is_column_oriented = table.is_column_oriented();
 
+            let lazy_parts = compact_block
+                .parts
+                .partitions
+                .iter()
+                .map(|v| {
+                    v.as_any()
+                        .downcast_ref::<CompactLazyPartInfo>()
+                        .unwrap()
+                        .clone()
+                })
+                .collect::<Vec<_>>();
+
+            let column_ids = compact_block.column_ids.clone();
             self.main_pipeline.set_on_init(move || {
                 let ctx = query_ctx.clone();
-                let partitions = Runtime::with_worker_threads(
-                    2,
-                    Some("build_compact_tasks".to_string()),
-                )?
-                .block_on(async move {
-                    let partitions = if is_column_oriented {
-                        let lazy_parts = partitions
-                            .iter()
-                            .map(|v| {
-                                v.as_any()
-                                    .downcast_ref::<ColumnOrientedSegmentsWithIndices>()
-                                    .unwrap()
-                                    .clone()
-                            })
-                            .collect::<Vec<_>>();
-                        BlockCompactMutator::<ColumnOrientedSegmentReader>::build_compact_tasks(
-                            ctx.clone(),
-                            column_ids.clone(),
-                            cluster_key_id,
-                            thresholds,
-                            lazy_parts,
-                        )
-                        .await?
-                    } else {
-                        let lazy_parts = partitions
-                            .iter()
-                            .map(|v| {
-                                v.as_any()
-                                    .downcast_ref::<CompactSegmentsWithIndices>()
-                                    .unwrap()
-                                    .clone()
-                            })
-                            .collect::<Vec<_>>();
-                        BlockCompactMutator::<RowOrientedSegmentReader>::build_compact_tasks(
-                            ctx.clone(),
-                            column_ids.clone(),
-                            cluster_key_id,
-                            thresholds,
-                            lazy_parts,
-                        )
-                        .await?
-                    };
+                let partitions =
+                    Runtime::with_worker_threads(2, Some("build_compact_tasks".to_string()))?
+                        .block_on(async move {
+                            let partitions = BlockCompactMutator::build_compact_tasks(
+                                ctx.clone(),
+                                column_ids.clone(),
+                                cluster_key_id,
+                                thresholds,
+                                lazy_parts,
+                            )
+                            .await?;
 
-                    Result::<_>::Ok(partitions)
-                })?;
+                            Result::<_>::Ok(partitions)
+                        })?;
 
                 let partitions = Partitions::create(PartitionsShuffleKind::Mod, partitions);
                 query_ctx.set_partitions(partitions)?;
@@ -179,17 +155,18 @@ impl PipelineBuilder {
 
         if is_lazy {
             self.main_pipeline.try_resize(1)?;
-            add_table_mutation_aggregator(
-                &mut self.main_pipeline,
-                table,
-                self.ctx.clone(),
-                vec![],
-                vec![],
-                vec![],
-                Default::default(),
-                MutationKind::Compact,
-                compact_block.table_meta_timestamps,
-            );
+            self.main_pipeline.add_async_accumulating_transformer(|| {
+                TableMutationAggregator::create(
+                    table,
+                    self.ctx.clone(),
+                    vec![],
+                    vec![],
+                    vec![],
+                    Default::default(),
+                    MutationKind::Compact,
+                    compact_block.table_meta_timestamps,
+                )
+            });
         }
         Ok(())
     }

src/query/service/src/pipelines/builders/builder_insert_multi_table.rs

@@ -34,8 +34,6 @@ use databend_common_sql::executor::physical_plans::ChunkFilter;
 use databend_common_sql::executor::physical_plans::ChunkMerge;
 use databend_common_sql::executor::physical_plans::Duplicate;
 use databend_common_sql::executor::physical_plans::Shuffle;
-use databend_common_storages_fuse::io::read::ColumnOrientedSegmentReader;
-use databend_common_storages_fuse::io::read::RowOrientedSegmentReader;
 use databend_common_storages_fuse::operations::CommitMultiTableInsert;
 use databend_common_storages_fuse::FuseTable;
 use databend_common_storages_fuse::TableContext;
@@ -291,22 +289,12 @@ impl PipelineBuilder {
                 block_thresholds,
                 target.table_meta_timestamps,
             )?));
-            if table.is_column_oriented() {
-                mutation_aggregator_builders.push(Box::new(
-                    self.mutation_aggregator_transform_builder::<ColumnOrientedSegmentReader>(
-                        table.clone(),
-                        target.table_meta_timestamps,
-                    )?,
-                ));
-            } else {
-                mutation_aggregator_builders.push(Box::new(
-                    self.mutation_aggregator_transform_builder::<RowOrientedSegmentReader>(
-                        table.clone(),
-                        target.table_meta_timestamps,
-                    )?,
-                ));
-            }
-
+            mutation_aggregator_builders.push(Box::new(
+                self.mutation_aggregator_transform_builder(
+                    table.clone(),
+                    target.table_meta_timestamps,
+                )?,
+            ));
             table_meta_timestampss.insert(table.get_id(), target.table_meta_timestamps);
             tables.insert(table.get_id(), table);
         }

src/query/service/src/pipelines/builders/builder_mutation.rs

@@ -34,8 +34,8 @@ use databend_common_pipeline_transforms::processors::TransformPipelineHelper;
 use databend_common_sql::binder::MutationStrategy;
 use databend_common_sql::executor::physical_plans::Mutation;
 use databend_common_sql::executor::physical_plans::MutationKind;
-use databend_common_storages_fuse::operations::new_serialize_segment_pipe_item;
 use databend_common_storages_fuse::operations::TransformSerializeBlock;
+use databend_common_storages_fuse::operations::TransformSerializeSegment;
 use databend_common_storages_fuse::operations::UnMatchedExprs;
 use databend_common_storages_fuse::FuseTable;
 
@@ -61,13 +61,13 @@ impl PipelineBuilder {
         let io_request_semaphore =
             Arc::new(Semaphore::new(self.settings.get_max_threads()? as usize));
 
-        let serialize_segment_transform = new_serialize_segment_pipe_item(
+        let serialize_segment_transform = TransformSerializeSegment::new(
             InputPort::create(),
             OutputPort::create(),
             table,
             block_thresholds,
             merge_into.table_meta_timestamps,
-        )?;
+        );
 
         // For row_id port, create rowid_aggregate_mutator
         // For matched data port and unmatched port, do serialize
@@ -161,7 +161,7 @@ impl PipelineBuilder {
                 vec.push(create_dummy_item());
             }
             // data port
-            vec.push(serialize_segment_transform);
+            vec.push(serialize_segment_transform.into_pipe_item());
             vec
         };
 

src/query/service/src/pipelines/builders/builder_replace_into.rs

@@ -49,10 +49,10 @@ use databend_common_sql::BindContext;
 use databend_common_sql::Metadata;
 use databend_common_sql::MetadataRef;
 use databend_common_sql::NameResolutionContext;
-use databend_common_storages_fuse::operations::new_serialize_segment_pipe_item;
 use databend_common_storages_fuse::operations::BroadcastProcessor;
 use databend_common_storages_fuse::operations::ReplaceIntoProcessor;
 use databend_common_storages_fuse::operations::TransformSerializeBlock;
+use databend_common_storages_fuse::operations::TransformSerializeSegment;
 use databend_common_storages_fuse::operations::UnbranchedReplaceIntoProcessor;
 use databend_common_storages_fuse::FuseTable;
 use parking_lot::RwLock;
@@ -134,13 +134,13 @@ impl PipelineBuilder {
         let mut block_builder = serialize_block_transform.get_block_builder();
         block_builder.source_schema = table.schema_with_stream();
 
-        let serialize_segment_transform = new_serialize_segment_pipe_item(
+        let serialize_segment_transform = TransformSerializeSegment::new(
             InputPort::create(),
             OutputPort::create(),
             table,
             *block_thresholds,
             replace.table_meta_timestamps,
-        )?;
+        );
         if !*need_insert {
             if segment_partition_num == 0 {
                 return Ok(());
@@ -207,7 +207,7 @@ impl PipelineBuilder {
         if segment_partition_num == 0 {
             let dummy_item = create_dummy_item();
             self.main_pipeline.add_pipe(Pipe::create(2, 2, vec![
-                serialize_segment_transform,
+                serialize_segment_transform.into_pipe_item(),
                 dummy_item,
             ]));
         } else {
@@ -230,7 +230,7 @@ impl PipelineBuilder {
             let item_size = segment_partition_num + 1;
             let mut pipe_items = Vec::with_capacity(item_size);
             // setup the dummy transform
-            pipe_items.push(serialize_segment_transform);
+            pipe_items.push(serialize_segment_transform.into_pipe_item());
 
             let max_threads = self.settings.get_max_threads()?;
             let io_request_semaphore = Arc::new(Semaphore::new(max_threads as usize));

src/query/service/src/pipelines/builders/transform_builder.rs

@@ -34,10 +34,9 @@ use databend_common_sql::evaluator::BlockOperator;
 use databend_common_sql::evaluator::CompoundBlockOperator;
 use databend_common_sql::ColumnSet;
 use databend_common_storages_factory::Table;
-use databend_common_storages_fuse::io::read::SegmentReader;
-use databend_common_storages_fuse::operations::new_serialize_segment_processor;
 use databend_common_storages_fuse::operations::TableMutationAggregator;
 use databend_common_storages_fuse::operations::TransformSerializeBlock;
+use databend_common_storages_fuse::operations::TransformSerializeSegment;
 use databend_common_storages_fuse::statistics::ClusterStatsGenerator;
 use databend_common_storages_fuse::FuseTable;
 use databend_storages_common_table_meta::meta;
@@ -146,25 +145,26 @@ impl PipelineBuilder {
     ) -> Result<impl Fn(Arc<InputPort>, Arc<OutputPort>) -> Result<ProcessorPtr>> {
         Ok(move |input, output| {
             let fuse_table = FuseTable::try_from_table(table.as_ref())?;
-            new_serialize_segment_processor(
+            let proc = TransformSerializeSegment::new(
                 input,
                 output,
                 fuse_table,
                 block_thresholds,
                 table_meta_timestamps,
-            )
+            );
+            proc.into_processor()
         })
     }
 
-    pub(crate) fn mutation_aggregator_transform_builder<R: SegmentReader>(
+    pub(crate) fn mutation_aggregator_transform_builder(
         &self,
         table: Arc<dyn Table>,
         table_meta_timestamps: TableMetaTimestamps,
     ) -> Result<impl Fn(Arc<InputPort>, Arc<OutputPort>) -> Result<ProcessorPtr>> {
         let ctx = self.ctx.clone();
         Ok(move |input, output| {
             let fuse_table = FuseTable::try_from_table(table.as_ref())?;
-            let aggregator = TableMutationAggregator::<R>::create(
+            let aggregator = TableMutationAggregator::create(
                 fuse_table,
                 ctx.clone(),
                 vec![],

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

@@ -27,7 +27,6 @@ use databend_common_sql::executor::physical_plans::CommitType;
 use databend_common_sql::executor::physical_plans::CompactSource;
 use databend_common_sql::executor::physical_plans::MutationKind;
 use databend_common_sql::executor::PhysicalPlan;
-use databend_common_storages_fuse::io::read::RowOrientedSegmentReader;
 use databend_common_storages_fuse::io::SegmentsIO;
 use databend_common_storages_fuse::operations::BlockCompactMutator;
 use databend_common_storages_fuse::operations::CompactBlockPartInfo;
@@ -244,7 +243,7 @@ async fn test_safety() -> Result<()> {
         };
 
         eprintln!("running target select");
-        let mut block_compact_mutator = BlockCompactMutator::<RowOrientedSegmentReader>::new(
+        let mut block_compact_mutator = BlockCompactMutator::new(
             ctx.clone(),
             threshold,
             compact_params,
@@ -287,7 +286,7 @@ pub async fn verify_compact_tasks(
                 compact_segment_indices.extend(extra.removed_segment_indexes.iter());
                 actual_blocks_number += extra.unchanged_blocks.len();
                 for b in &extra.unchanged_blocks {
-                    actual_block_ids.insert(b.1.location.0.clone());
+                    actual_block_ids.insert(b.1.location.clone());
                 }
             }
             CompactBlockPartInfo::CompactTaskInfo(task) => {
@@ -318,7 +317,7 @@ pub async fn verify_compact_tasks(
         let segment = SegmentInfo::try_from(compact_segment)?;
         except_blocks_number += segment.blocks.len();
         for b in &segment.blocks {
-            except_block_ids.insert(b.location.0.clone());
+            except_block_ids.insert(b.location.clone());
         }
     }
     assert_eq!(except_blocks_number, actual_blocks_number);