feat(storage): Improve optimize table compact

query/src/interpreters/interpreter_table_optimize.rs

@@ -19,13 +19,10 @@ use common_planners::OptimizeTableAction;
 use common_planners::OptimizeTablePlan;
 use common_streams::DataBlockStream;
 use common_streams::SendableDataBlockStream;
-use futures::StreamExt;
 
 use crate::interpreters::Interpreter;
-use crate::interpreters::InterpreterFactory;
 use crate::interpreters::InterpreterPtr;
 use crate::sessions::QueryContext;
-use crate::sql::PlanParser;
 
 pub struct OptimizeTableInterpreter {
     ctx: Arc<QueryContext>,
@@ -65,15 +62,8 @@ impl Interpreter for OptimizeTableInterpreter {
         );
 
         if do_compact {
-            // it is a "simple and violent" strategy, to be optimized later
-            let obj_name = format!("{}.{}", &plan.database, &plan.table);
-            let rewritten_query =
-                format!("INSERT OVERWRITE {} SELECT * FROM {}", obj_name, obj_name);
-            let rewritten_plan =
-                PlanParser::parse(self.ctx.clone(), rewritten_query.as_str()).await?;
-            let interpreter = InterpreterFactory::get(self.ctx.clone(), rewritten_plan)?;
-            let mut stream = interpreter.execute(None).await?;
-            while let Some(Ok(_)) = stream.next().await {}
+            // TODO(zhyass): clustering key.
+            table.compact(self.ctx.clone(), self.plan.clone()).await?;
             if do_purge {
                 // currently, context caches the table, we have to "refresh"
                 // the table by using the catalog API directly

query/src/storages/fuse/fuse_table.rs

@@ -28,6 +28,7 @@ use common_meta_types::MatchSeq;
 use common_planners::DeletePlan;
 use common_planners::Expression;
 use common_planners::Extras;
+use common_planners::OptimizeTablePlan;
 use common_planners::Partitions;
 use common_planners::ReadDataSourcePlan;
 use common_planners::Statistics;
@@ -449,4 +450,9 @@ impl Table for FuseTable {
     async fn delete(&self, ctx: Arc<QueryContext>, delete_plan: DeletePlan) -> Result<()> {
         self.do_delete(ctx, &delete_plan).await
     }
+
+    #[tracing::instrument(level = "debug", name = "fuse_table_compact", skip(self, ctx), fields(ctx.id = ctx.get_id().as_str()))]
+    async fn compact(&self, ctx: Arc<QueryContext>, plan: OptimizeTablePlan) -> Result<()> {
+        self.do_compact(ctx, &plan).await
+    }
 }

query/src/storages/fuse/operations/append.rs

@@ -223,7 +223,7 @@ impl FuseTable {
         Ok(())
     }
 
-    fn get_option<T: FromStr>(&self, opt_key: &str, default: T) -> T {
+    pub fn get_option<T: FromStr>(&self, opt_key: &str, default: T) -> T {
         self.table_info
             .options()
             .get(opt_key)

query/src/storages/fuse/operations/compact.rs

@@ -0,0 +1,88 @@
+//  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::sync::Arc;
+
+use common_cache::Cache;
+use common_exception::Result;
+use common_planners::OptimizeTablePlan;
+
+use super::mutation::CompactMutator;
+use crate::sessions::QueryContext;
+use crate::storages::fuse::FuseTable;
+use crate::storages::fuse::DEFAULT_BLOCK_PER_SEGMENT;
+use crate::storages::fuse::DEFAULT_ROW_PER_BLOCK;
+use crate::storages::fuse::FUSE_OPT_KEY_BLOCK_PER_SEGMENT;
+use crate::storages::fuse::FUSE_OPT_KEY_ROW_PER_BLOCK;
+use crate::storages::storage_table::Table;
+
+impl FuseTable {
+    pub async fn do_compact(&self, ctx: Arc<QueryContext>, plan: &OptimizeTablePlan) -> Result<()> {
+        let snapshot_opt = self.read_table_snapshot(ctx.as_ref()).await?;
+        let snapshot = if let Some(val) = snapshot_opt {
+            val
+        } else {
+            // no snapshot, no compaction.
+            return Ok(());
+        };
+
+        if snapshot.summary.block_count <= 1 {
+            return Ok(());
+        }
+
+        let row_per_block = self.get_option(FUSE_OPT_KEY_ROW_PER_BLOCK, DEFAULT_ROW_PER_BLOCK);
+        let block_per_seg =
+            self.get_option(FUSE_OPT_KEY_BLOCK_PER_SEGMENT, DEFAULT_BLOCK_PER_SEGMENT);
+
+        let mut mutator = CompactMutator::try_create(
+            &ctx,
+            &self.meta_location_generator,
+            &snapshot,
+            row_per_block,
+            block_per_seg,
+        )?;
+
+        mutator.compact(self).await?;
+        let (new_snapshot, loc) = mutator.into_new_snapshot().await?;
+
+        let operator = ctx.get_storage_operator()?;
+        let result = Self::commit_to_meta_server(
+            ctx.as_ref(),
+            &plan.catalog,
+            self.get_table_info(),
+            loc.clone(),
+            &new_snapshot.summary,
+        )
+        .await;
+
+        match result {
+            Ok(_) => {
+                if let Some(snapshot_cache) =
+                    ctx.get_storage_cache_manager().get_table_snapshot_cache()
+                {
+                    let cache = &mut snapshot_cache.write().await;
+                    cache.put(loc, Arc::new(new_snapshot));
+                }
+                Ok(())
+            }
+            Err(e) => {
+                // commit snapshot to meta server failed, try to delete it.
+                // "major GC" will collect this, if deletion failure (even after DAL retried)
+                let _ = operator.object(&loc).delete().await;
+                Err(e)
+            }
+        }
+    }
+}

query/src/storages/fuse/operations/mod.rs

@@ -14,6 +14,7 @@
 
 mod append;
 mod commit;
+mod compact;
 mod delete;
 mod fuse_sink;
 mod gc;

query/src/storages/fuse/operations/mutation/block_filter.rs

@@ -115,7 +115,7 @@ pub async fn delete_from_block(
     Ok(res)
 }
 
-fn all_the_columns_ids(table: &FuseTable) -> Vec<usize> {
+pub fn all_the_columns_ids(table: &FuseTable) -> Vec<usize> {
     (0..table.table_info.schema().fields().len())
         .into_iter()
         .collect::<Vec<usize>>()

query/src/storages/fuse/operations/mutation/compact_mutator.rs

@@ -0,0 +1,164 @@
+//  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::sync::Arc;
+
+use common_exception::Result;
+use opendal::Operator;
+
+use super::block_filter::all_the_columns_ids;
+use crate::sessions::QueryContext;
+use crate::storages::fuse::io::BlockCompactor;
+use crate::storages::fuse::io::BlockWriter;
+use crate::storages::fuse::io::MetaReaders;
+use crate::storages::fuse::io::SegmentWriter;
+use crate::storages::fuse::io::TableMetaLocationGenerator;
+use crate::storages::fuse::meta::Location;
+use crate::storages::fuse::meta::SegmentInfo;
+use crate::storages::fuse::meta::Statistics;
+use crate::storages::fuse::meta::TableSnapshot;
+use crate::storages::fuse::statistics::reducers::reduce_block_metas;
+use crate::storages::fuse::statistics::reducers::reduce_statistics;
+use crate::storages::fuse::FuseTable;
+
+pub struct CompactMutator<'a> {
+    ctx: &'a Arc<QueryContext>,
+    location_generator: &'a TableMetaLocationGenerator,
+    base_snapshot: &'a TableSnapshot,
+    data_accessor: Operator,
+    row_per_block: usize,
+    block_per_seg: usize,
+    segments: Vec<Location>,
+    summarys: Vec<Statistics>,
+}
+
+impl<'a> CompactMutator<'a> {
+    pub fn try_create(
+        ctx: &'a Arc<QueryContext>,
+        location_generator: &'a TableMetaLocationGenerator,
+        base_snapshot: &'a TableSnapshot,
+        row_per_block: usize,
+        block_per_seg: usize,
+    ) -> Result<Self> {
+        let data_accessor = ctx.get_storage_operator()?;
+        Ok(Self {
+            ctx,
+            location_generator,
+            base_snapshot,
+            data_accessor,
+            row_per_block,
+            block_per_seg,
+            segments: Vec::new(),
+            summarys: Vec::new(),
+        })
+    }
+
+    pub async fn into_new_snapshot(self) -> Result<(TableSnapshot, String)> {
+        let snapshot = self.base_snapshot;
+        let mut new_snapshot = TableSnapshot::from_previous(snapshot);
+        new_snapshot.segments = self.segments.clone();
+        // update the summary of new snapshot
+        let new_summary = reduce_statistics(&self.summarys)?;
+        new_snapshot.summary = new_summary;
+
+        // write the new segment out (and keep it in undo log)
+        let snapshot_loc = self.location_generator.snapshot_location_from_uuid(
+            &new_snapshot.snapshot_id,
+            new_snapshot.format_version(),
+        )?;
+        let bytes = serde_json::to_vec(&new_snapshot)?;
+        self.data_accessor
+            .object(&snapshot_loc)
+            .write(bytes)
+            .await?;
+        Ok((new_snapshot, snapshot_loc))
+    }
+
+    pub async fn compact(&mut self, table: &FuseTable) -> Result<()> {
+        // Blocks that need to be reorganized into new segments. 
+        let mut remain_blocks = Vec::new();
+        // Blocks that need to be compacted.
+        let mut merged_blocks = Vec::new();
+        let reader = MetaReaders::segment_info_reader(self.ctx);
+        for segment_location in &self.base_snapshot.segments {
+            let (x, ver) = (segment_location.0.clone(), segment_location.1);
+            let mut need_merge = false;
+            let mut remains = Vec::new();
+            let segment = reader.read(x, None, ver).await?;
+            segment.blocks.iter().for_each(|b| {
+                if b.row_count != self.row_per_block as u64 {
+                    merged_blocks.push(b.clone());
+                    need_merge = true;
+                } else {
+                    remains.push(b.clone());
+                }
+            });
+
+            // If the number of blocks of segment meets block_per_seg, and the blocks in segments donot need to be compacted, 
+            // then record the segment information.
+            if !need_merge && segment.blocks.len() == self.block_per_seg {
+                self.segments.push(segment_location.clone());
+                self.summarys.push(segment.summary.clone());
+                continue;
+            }
+
+            remain_blocks.append(&mut remains);
+        }
+
+        // Compact the blocks.
+        let col_ids = all_the_columns_ids(table);
+        let mut compactor = BlockCompactor::new(self.row_per_block);
+        let block_writer = BlockWriter::new(&self.data_accessor, self.location_generator);
+        for block_meta in &merged_blocks {
+            let block_reader = table.create_block_reader(self.ctx, col_ids.clone())?;
+            let data_block = block_reader.read_with_block_meta(block_meta).await?;
+
+            let res = compactor.compact(data_block)?;
+            if let Some(blocks) = res {
+                for block in blocks {
+                    let new_block_meta = block_writer.write(block).await?;
+                    remain_blocks.push(new_block_meta);
+                }
+            }
+        }
+        let remains = compactor.finish()?;
+        if let Some(blocks) = remains {
+            for block in blocks {
+                let new_block_meta = block_writer.write(block).await?;
+                remain_blocks.push(new_block_meta);
+            }
+        }
+
+        // Create new segments.
+        let segment_info_cache = self
+            .ctx
+            .get_storage_cache_manager()
+            .get_table_segment_cache();
+        let seg_writer = SegmentWriter::new(
+            &self.data_accessor,
+            self.location_generator,
+            &segment_info_cache,
+        );
+        let chunks = remain_blocks.chunks(self.block_per_seg);
+        for chunk in chunks {
+            let new_summary = reduce_block_metas(chunk)?;
+            let new_segment = SegmentInfo::new(chunk.to_vec(), new_summary.clone());
+            let new_segment_location = seg_writer.write_segment(new_segment).await?;
+            self.segments.push(new_segment_location);
+            self.summarys.push(new_summary);
+        }
+
+        Ok(())
+    }
+}

query/src/storages/fuse/operations/mutation/mod.rs

@@ -13,6 +13,8 @@
 //  limitations under the License.
 
 pub mod block_filter;
+pub mod compact_mutator;
 pub mod mutations_collector;
 
 pub use block_filter::delete_from_block;
+pub use compact_mutator::CompactMutator;

query/src/storages/storage_table.rs

@@ -27,6 +27,7 @@ use common_meta_types::MetaId;
 use common_planners::DeletePlan;
 use common_planners::Expression;
 use common_planners::Extras;
+use common_planners::OptimizeTablePlan;
 use common_planners::Partitions;
 use common_planners::ReadDataSourcePlan;
 use common_planners::Statistics;
@@ -202,6 +203,14 @@ pub trait Table: Sync + Send {
             self.get_table_info().engine(),
         )))
     }
+
+    async fn compact(&self, _ctx: Arc<QueryContext>, _plan: OptimizeTablePlan) -> Result<()> {
+        Err(ErrorCode::UnImplement(format!(
+            "table {},  of engine type {}, does not support Optimize",
+            self.name(),
+            self.get_table_info().engine(),
+        )))
+    }
 }
 
 #[derive(Debug)]