chore: Merge the logic that triggers automatic compression upon write

Author: zhyass

src/query/service/src/interpreters/hook/compact_hook.rs

@@ -26,7 +26,6 @@ use databend_common_exception::Result;
 use databend_common_pipeline::core::always_callback;
 use databend_common_pipeline::core::ExecutionInfo;
 use databend_common_pipeline::core::Pipeline;
-use databend_common_sql::executor::physical_plans::MutationKind;
 use databend_common_sql::optimizer::ir::SExpr;
 use databend_common_sql::plans::OptimizeCompactBlock;
 use databend_common_sql::plans::ReclusterPlan;
@@ -50,7 +49,6 @@ pub struct CompactTargetTableDescription {
     pub catalog: String,
     pub database: String,
     pub table: String,
-    pub mutation_kind: MutationKind,
 }
 
 pub struct CompactHookTraceCtx {
@@ -95,30 +93,18 @@ async fn do_hook_compact(
             info!("Operation {op_name} completed successfully, starting table optimization job.");
 
             let compact_start_at = Instant::now();
-            let compaction_limits = match compact_target.mutation_kind {
-                MutationKind::Insert => {
-                    let compaction_num_block_hint =
-                        ctx.get_compaction_num_block_hint(&compact_target.table);
-                    info!(
-                        "Table {} requires compaction of {} blocks",
-                        compact_target.table, compaction_num_block_hint
-                    );
-                    if compaction_num_block_hint == 0 {
-                        return Ok(());
-                    }
-                    CompactionLimits {
-                        segment_limit: None,
-                        block_limit: Some(compaction_num_block_hint as usize),
-                    }
-                }
-                _ => {
-                    let auto_compaction_segments_limit =
-                        ctx.get_settings().get_auto_compaction_segments_limit()?;
-                    CompactionLimits {
-                        segment_limit: Some(auto_compaction_segments_limit as usize),
-                        block_limit: None,
-                    }
-                }
+            let compaction_num_block_hint =
+                ctx.get_compaction_num_block_hint(&compact_target.table);
+            info!(
+                "Table {} requires compaction of {} blocks",
+                compact_target.table, compaction_num_block_hint
+            );
+            if compaction_num_block_hint == 0 {
+                return Ok(());
+            }
+            let compaction_limits = CompactionLimits {
+                segment_limit: None,
+                block_limit: Some(compaction_num_block_hint as usize),
             };
 
             // keep the original progress value

src/query/service/src/interpreters/hook/hook.rs

@@ -102,7 +102,6 @@ impl HookOperator {
             catalog: self.catalog.to_owned(),
             database: self.database.to_owned(),
             table: self.table.to_owned(),
-            mutation_kind: self.mutation_kind,
         };
 
         let trace_ctx = CompactHookTraceCtx {

src/query/settings/src/settings_default.rs

@@ -892,7 +892,7 @@ impl DefaultSettings {
                 }),
                 ("auto_compaction_segments_limit", DefaultSettingValue {
                     value: UserSettingValue::UInt64(3),
-                    desc: "The maximum number of segments that can be compacted automatically triggered after write(replace-into/merge-into).",
+                    desc: "The maximum number of segments that can be reclustered automatically triggered after write.",
                     mode: SettingMode::Both,
                     scope: SettingScope::Both,
                     range: Some(SettingRange::Numeric(2..=u64::MAX)),

src/query/storages/fuse/src/operations/common/generators/append_generator.rs

@@ -30,7 +30,6 @@ use databend_storages_common_table_meta::meta::ColumnStatistics;
 use databend_storages_common_table_meta::meta::Statistics;
 use databend_storages_common_table_meta::meta::TableMetaTimestamps;
 use databend_storages_common_table_meta::meta::TableSnapshot;
-use log::info;
 use log::warn;
 
 use crate::operations::common::ConflictResolveContext;
@@ -41,7 +40,8 @@ use crate::statistics::TableStatsGenerator;
 
 #[derive(Clone)]
 pub struct AppendGenerator {
-    ctx: Arc<dyn TableContext>,
+    pub(crate) ctx: Arc<dyn TableContext>,
+
     leaf_default_values: HashMap<ColumnId, Scalar>,
     overwrite: bool,
     conflict_resolve_ctx: ConflictResolveContext,
@@ -194,28 +194,6 @@ impl SnapshotGenerator for AppendGenerator {
             }
         }
 
-        // check if need to auto compact
-        // the algorithm is: if the number of imperfect blocks is greater than the threshold, then auto compact.
-        // the threshold is set by the setting `auto_compaction_imperfect_blocks_threshold`, default is 25.
-        let imperfect_count = new_summary.block_count - new_summary.perfect_block_count;
-        let auto_compaction_imperfect_blocks_threshold = self
-            .ctx
-            .get_settings()
-            .get_auto_compaction_imperfect_blocks_threshold()?;
-
-        if imperfect_count >= auto_compaction_imperfect_blocks_threshold {
-            // If imperfect_count is larger, SLIGHTLY increase the number of blocks
-            // eligible for auto-compaction, this adjustment is intended to help reduce
-            // fragmentation over time.
-            let compact_num_block_hint = std::cmp::min(
-                imperfect_count,
-                (auto_compaction_imperfect_blocks_threshold as f64 * 1.5).ceil() as u64,
-            );
-            info!("set compact_num_block_hint to {compact_num_block_hint }");
-            self.ctx
-                .set_compaction_num_block_hint(table_info.name.as_str(), compact_num_block_hint);
-        }
-
         // merge statistics will set the additional_stats_meta to none,
         // so reset additional_stats_meta here.
         let table_statistics_location = table_stats_gen.table_statistics_location();

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

@@ -24,5 +24,6 @@ pub use conflict_resolve_context::SnapshotChanges;
 pub use conflict_resolve_context::SnapshotMerged;
 pub use mutation_generator::MutationGenerator;
 pub use snapshot_generator::decorate_snapshot;
+pub(crate) use snapshot_generator::set_compaction_num_block_hint;
 pub use snapshot_generator::SnapshotGenerator;
 pub use truncate_generator::TruncateGenerator;

src/query/storages/fuse/src/operations/common/generators/snapshot_generator.rs

@@ -15,12 +15,15 @@
 use std::any::Any;
 use std::sync::Arc;
 
+use databend_common_catalog::table_context::TableContext;
 use databend_common_exception::Result;
 use databend_common_expression::TableSchema;
 use databend_common_meta_app::schema::TableInfo;
 use databend_storages_common_session::TxnManagerRef;
+use databend_storages_common_table_meta::meta::Statistics;
 use databend_storages_common_table_meta::meta::TableMetaTimestamps;
 use databend_storages_common_table_meta::meta::TableSnapshot;
+use log::info;
 
 use crate::operations::common::ConflictResolveContext;
 use crate::statistics::TableStatsGenerator;
@@ -92,3 +95,29 @@ pub fn decorate_snapshot(
     }
     Ok(())
 }
+
+pub(crate) fn set_compaction_num_block_hint(
+    ctx: &dyn TableContext,
+    table_name: &str,
+    summary: &Statistics,
+) -> Result<()> {
+    // check if need to auto compact
+    // the algorithm is: if the number of imperfect blocks is greater than the threshold, then auto compact.
+    // the threshold is set by the setting `auto_compaction_imperfect_blocks_threshold`, default is 25.
+    let imperfect_count = summary.block_count - summary.perfect_block_count;
+    let auto_compaction_imperfect_blocks_threshold = ctx
+        .get_settings()
+        .get_auto_compaction_imperfect_blocks_threshold()?;
+    if imperfect_count >= auto_compaction_imperfect_blocks_threshold {
+        // If imperfect_count is larger, SLIGHTLY increase the number of blocks
+        // eligible for auto-compaction, this adjustment is intended to help reduce
+        // fragmentation over time.
+        let compact_num_block_hint = std::cmp::min(
+            imperfect_count,
+            (auto_compaction_imperfect_blocks_threshold as f64 * 1.5).ceil() as u64,
+        );
+        info!("set compact_num_block_hint to {compact_num_block_hint }");
+        ctx.set_compaction_num_block_hint(table_name, compact_num_block_hint);
+    }
+    Ok(())
+}

src/query/storages/fuse/src/operations/common/processors/multi_table_insert_commit.rs

@@ -42,6 +42,7 @@ use log::error;
 use log::info;
 
 use crate::operations::set_backoff;
+use crate::operations::set_compaction_num_block_hint;
 use crate::operations::AppendGenerator;
 use crate::operations::CommitMeta;
 use crate::operations::SnapshotGenerator;
@@ -272,15 +273,21 @@ async fn build_update_table_meta_req(
     let table_stats_gen = fuse_table
         .generate_table_stats(&previous, insert_hll, insert_rows)
         .await?;
+    let table_info = table.get_table_info();
     let snapshot = snapshot_generator.generate_new_snapshot(
-        table.get_table_info(),
+        table_info,
         fuse_table.cluster_key_id(),
         previous,
         txn_mgr,
         table_meta_timestamps,
         table_stats_gen,
     )?;
     snapshot.ensure_segments_unique()?;
+    let _ = set_compaction_num_block_hint(
+        snapshot_generator.ctx.as_ref(),
+        table_info.name.as_str(),
+        &snapshot.summary,
+    );
 
     // write snapshot
     let dal = fuse_table.get_operator();

src/query/storages/fuse/src/operations/common/processors/sink_commit.rs

@@ -58,6 +58,7 @@ use opendal::Operator;
 
 use crate::io::TableMetaLocationGenerator;
 use crate::operations::set_backoff;
+use crate::operations::set_compaction_num_block_hint;
 use crate::operations::vacuum::vacuum_table;
 use crate::operations::AppendGenerator;
 use crate::operations::CommitMeta;
@@ -442,6 +443,11 @@ where F: SnapshotGenerator + Send + Sync + 'static
                     table_stats_gen,
                 ) {
                     Ok(snapshot) => {
+                        let _ = set_compaction_num_block_hint(
+                            self.ctx.as_ref(),
+                            table_info.name.as_str(),
+                            &snapshot.summary,
+                        );
                         self.state = State::TryCommit {
                             data: snapshot.to_bytes()?,
                             snapshot,