feat: compact segment

src/common/base/src/base/runtime.rs

@@ -205,9 +205,10 @@ impl Runtime {
     {
         let semaphore = Arc::new(semaphore);
         let iter = futures.into_iter().map(|v| {
-            |permit| {
-                let _permit = permit;
-                v
+            |permit| async {
+                let r = v.await;
+                drop(permit);
+                r
             }
         });
         self.try_spawn_batch_with_owned_semaphore(semaphore, iter)

src/query/ast/src/ast/statements/table.rs

@@ -444,19 +444,32 @@ impl Display for Engine {
     }
 }
 
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum CompactTarget {
+    Block,
+    Segment,
+}
+
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum OptimizeTableAction {
     All,
     Purge,
-    Compact,
+    Compact(CompactTarget),
 }
 
 impl Display for OptimizeTableAction {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         match self {
             OptimizeTableAction::All => write!(f, "ALL"),
             OptimizeTableAction::Purge => write!(f, "PURGE"),
-            OptimizeTableAction::Compact => write!(f, "COMPACT"),
+            OptimizeTableAction::Compact(action) => match action {
+                CompactTarget::Block => {
+                    write!(f, "COMPACT BLOCK")
+                }
+                CompactTarget::Segment => {
+                    write!(f, "COMPACT SEGMENT")
+                }
+            },
         }
     }
 }

src/query/ast/src/parser/statement.rs

@@ -941,7 +941,7 @@ pub fn statement(i: Input) -> IResult<StatementMsg> {
             | #alter_table : "`ALTER TABLE [<database>.]<table> <action>`"
             | #rename_table : "`RENAME TABLE [<database>.]<table> TO <new_table>`"
             | #truncate_table : "`TRUNCATE TABLE [<database>.]<table> [PURGE]`"
-            | #optimize_table : "`OPTIMIZE TABLE [<database>.]<table> (ALL | PURGE | COMPACT)`"
+            | #optimize_table : "`OPTIMIZE TABLE [<database>.]<table> (ALL | PURGE | COMPACT [SEGMENT])`"
             | #exists_table : "`EXISTS TABLE [<database>.]<table>`"
         ),
         rule!(
@@ -1364,7 +1364,14 @@ pub fn optimize_table_action(i: Input) -> IResult<OptimizeTableAction> {
     alt((
         value(OptimizeTableAction::All, rule! { ALL }),
         value(OptimizeTableAction::Purge, rule! { PURGE }),
-        value(OptimizeTableAction::Compact, rule! { COMPACT }),
+        value(
+            OptimizeTableAction::Compact(CompactTarget::Segment),
+            rule! { COMPACT ~ SEGMENT},
+        ),
+        value(
+            OptimizeTableAction::Compact(CompactTarget::Block),
+            rule! { COMPACT},
+        ),
     ))(i)
 }
 

src/query/ast/src/parser/token.rs

@@ -601,6 +601,8 @@ pub enum TokenKind {
     SECOND,
     #[token("SELECT", ignore(ascii_case))]
     SELECT,
+    #[token("SEGMENT", ignore(ascii_case))]
+    SEGMENT,
     #[token("SET", ignore(ascii_case))]
     SET,
     #[token("SETTINGS", ignore(ascii_case))]

src/query/catalog/src/table.rs

@@ -232,9 +232,10 @@ pub trait Table: Sync + Send {
     async fn compact(
         &self,
         ctx: Arc<dyn TableContext>,
+        target: CompactTarget,
         pipeline: &mut Pipeline,
     ) -> Result<Option<Arc<dyn TableMutator>>> {
-        let (_, _) = (ctx, pipeline);
+        let (_, _, _) = (ctx, target, pipeline);
 
         Err(ErrorCode::UnImplement(format!(
             "table {},  of engine type {}, does not support compact",
@@ -303,6 +304,11 @@ pub struct ColumnStatistics {
     pub number_of_distinct_values: u64,
 }
 
+pub enum CompactTarget {
+    Blocks,
+    Segments,
+}
+
 pub trait ColumnStatisticsProvider {
     // returns the statistics of the given column, if any.
     // column_id is just the index of the column in table's schema

src/query/catalog/src/table_mutator.rs

@@ -20,6 +20,6 @@ use crate::table::Table;
 
 #[async_trait::async_trait]
 pub trait TableMutator: Send + Sync {
-    async fn blocks_select(&mut self) -> Result<bool>;
+    async fn target_select(&mut self) -> Result<bool>;
     async fn try_commit(&self, table: Arc<dyn Table>) -> Result<()>;
 }

src/query/planner/src/plans/optimize_table.rs

@@ -34,5 +34,6 @@ impl OptimizeTablePlan {
 pub enum OptimizeTableAction {
     All,
     Purge,
-    Compact,
+    CompactBlocks,
+    CompactSegments,
 }

src/query/service/src/interpreters/interpreter_table_optimize.rs

@@ -14,6 +14,7 @@
 
 use std::sync::Arc;
 
+use common_catalog::table::CompactTarget;
 use common_exception::Result;
 use common_planner::plans::OptimizeTableAction;
 use common_planner::plans::OptimizeTablePlan;
@@ -56,14 +57,29 @@ impl Interpreter for OptimizeTableInterpreter {
             action,
             OptimizeTableAction::Purge | OptimizeTableAction::All
         );
-        let do_compact = matches!(
+        let do_compact_blocks = matches!(
             action,
-            OptimizeTableAction::Compact | OptimizeTableAction::All
+            OptimizeTableAction::CompactBlocks | OptimizeTableAction::All
         );
 
-        if do_compact {
+        let do_compact_segments_only = matches!(action, OptimizeTableAction::CompactSegments);
+
+        if do_compact_segments_only {
+            let mut pipeline = Pipeline::create();
+            if let Some(mutator) = table
+                .compact(ctx.clone(), CompactTarget::Segments, &mut pipeline)
+                .await?
+            {
+                mutator.try_commit(table).await?;
+                return Ok(PipelineBuildResult::create());
+            }
+        }
+
+        if do_compact_blocks {
             let mut pipeline = Pipeline::create();
-            let mutator = table.compact(ctx.clone(), &mut pipeline).await?;
+            let mutator = table
+                .compact(ctx.clone(), CompactTarget::Blocks, &mut pipeline)
+                .await?;
 
             if let Some(mutator) = mutator {
                 let settings = ctx.get_settings();

src/query/service/src/servers/mysql/mysql_interactive_worker.rs

@@ -358,7 +358,7 @@ impl<W: AsyncWrite + Send + Unpin> InteractiveWorkerBase<W> {
         }
     }
 
-    //#[tracing::instrument(level = "debug", skip(interpreter, context))]
+    #[tracing::instrument(level = "debug", skip(interpreter, context))]
     async fn exec_query(
         interpreter: Arc<dyn Interpreter>,
         context: &Arc<QueryContext>,

src/query/service/src/sql/planner/binder/ddl/table.rs

@@ -781,7 +781,10 @@ impl<'a> Binder {
         let action = action.map_or(OptimizeTableAction::Purge, |v| match v {
             AstOptimizeTableAction::All => OptimizeTableAction::All,
             AstOptimizeTableAction::Purge => OptimizeTableAction::Purge,
-            AstOptimizeTableAction::Compact => OptimizeTableAction::Compact,
+            AstOptimizeTableAction::Compact(target) => match target {
+                CompactTarget::Block => OptimizeTableAction::CompactBlocks,
+                CompactTarget::Segment => OptimizeTableAction::CompactSegments,
+            },
         });
 
         Ok(Plan::OptimizeTable(Box::new(OptimizeTablePlan {

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

@@ -15,6 +15,7 @@
 use std::sync::Arc;
 
 use common_base::base::tokio;
+use common_catalog::table::CompactTarget;
 use common_catalog::table::Table;
 use common_catalog::table_mutator::TableMutator;
 use common_exception::ErrorCode;
@@ -174,7 +175,9 @@ async fn build_mutator(
     let settings = ctx.get_settings();
     settings.set_max_threads(1)?;
     let mut pipeline = common_pipeline_core::Pipeline::create();
-    let mutator = fuse_table.compact(ctx.clone(), &mut pipeline).await?;
+    let mutator = fuse_table
+        .compact(ctx.clone(), CompactTarget::Blocks, &mut pipeline)
+        .await?;
     assert!(mutator.is_some());
     let mutator = mutator.unwrap();
     pipeline.set_max_threads(1);