Repository: tikv/tikv (language: rust)

Issue:
compact_log_backup: add minimal compaction size
## Development Task
Some test shows that compacting small logs to small SSTs will make performance even worse. We may need skip small compactions.


Discussion / hints:

Fix patch:
diff --git a/cmd/tikv-ctl/src/cmd.rs b/cmd/tikv-ctl/src/cmd.rs
index ae562292b2f..a42e1781383 100644
--- a/cmd/tikv-ctl/src/cmd.rs
+++ b/cmd/tikv-ctl/src/cmd.rs
@@ -4,6 +4,7 @@ use std::{borrow::ToOwned, str, string::ToString, sync::LazyLock, u64};
 
 use clap::{crate_authors, AppSettings};
 use engine_traits::{SstCompressionType, CF_DEFAULT};
+use raft_engine::ReadableSize;
 use structopt::StructOpt;
 
 const RAW_KEY_HINT: &str = "Raw key (generally starts with \"z\") in escaped form";
@@ -688,8 +689,22 @@ pub enum Cmd {
         )]
         compression_level: Option<i32>,
 
-        #[structopt(long, help("Don't try to skip already finished compactions."))]
+        #[structopt(
+            long,
+            help(
+                "if set, all checkpoints will be ignored. i.e. all finished compaction will be regenerated."
+            )
+        )]
         force_regenerate: bool,
+
+        #[structopt(
+            long,
+            default_value = "16M",
+            help(
+                "specify the minimal compaction size in bytes, if backup data of a region doesn't reach this threshold, it won't be compacted"
+            )
+        )]
+        minimal_compaction_size: ReadableSize,
     },
     /// Get the state of a region's RegionReadProgress.
     GetRegionReadProgress {
diff --git a/cmd/tikv-ctl/src/main.rs b/cmd/tikv-ctl/src/main.rs
index 9b8852c62b1..7e3832a6137 100644
--- a/cmd/tikv-ctl/src/main.rs
+++ b/cmd/tikv-ctl/src/main.rs
@@ -20,7 +20,10 @@ use std::{
 };
 
 use collections::HashMap;
-use compact_log_backup::{exec_hooks as compact_log_hooks, execute as compact_log, TraceResultExt};
+use compact_log_backup::{
+    exec_hooks::{self as compact_log_hooks, skip_small_compaction::SkipSmallCompaction},
+    execute as compact_log, TraceResultExt,
+};
 use crypto::fips;
 use encryption_export::{
     create_backend, data_key_manager_from_config, DataKeyManager, DecrypterReader, Iv,
@@ -397,6 +400,7 @@ fn main() {
             compression_level,
             name,
             force_regenerate,
+            minimal_compaction_size,
         } => {
             let tmp_engine =
                 TemporaryRocks::new(&cfg).expect("failed to create temp engine for writing SSTs.");
@@ -467,8 +471,12 @@ fn main() {
             } else {
                 Some(compact_log_hooks::checkpoint::Checkpoint::default())
             };
+            let skip_small_compaction = SkipSmallCompaction::new(minimal_compaction_size.0);
             let hooks = (
-                ((log_to_term, checkpoint), with_status_server),
+                (
+                    (log_to_term, checkpoint),
+                    (with_status_server, skip_small_compaction),
+                ),
                 (save_meta, with_lock),
             );
             match exec.run(hooks) {
diff --git a/components/compact-log-backup/src/exec_hooks/mod.rs b/components/compact-log-backup/src/exec_hooks/mod.rs
index 134887051a6..872114a87b4 100644
--- a/components/compact-log-backup/src/exec_hooks/mod.rs
+++ b/components/compact-log-backup/src/exec_hooks/mod.rs
@@ -13,6 +13,7 @@ pub mod checkpoint;
 pub mod consistency;
 pub mod observability;
 pub mod save_meta;
+pub mod skip_small_compaction;
 
 #[derive(Default)]
 pub struct CollectStatistic {
diff --git a/components/compact-log-backup/src/exec_hooks/skip_small_compaction.rs b/components/compact-log-backup/src/exec_hooks/skip_small_compaction.rs
new file mode 100644
index 00000000000..c16bc3cfd38
--- /dev/null
+++ b/components/compact-log-backup/src/exec_hooks/skip_small_compaction.rs
@@ -0,0 +1,30 @@
+// Copyright 2024 TiKV Project Authors. Licensed under Apache-2.0.
+
+use tikv_util::info;
+
+use crate::execute::hooking::{CId, ExecHooks, SubcompactionStartCtx};
+
+/// A hook that skips small compaction.
+///
+/// The size threshold is compared with "original KV size":
+/// the length of each key value pairs in its original form.
+pub struct SkipSmallCompaction {
+    size_threshold: u64,
+}
+
+impl SkipSmallCompaction {
+    /// Creates a new `SkipSmallCompaction` hook.
+    pub fn new(size_threshold: u64) -> Self {
+        Self { size_threshold }
+    }
+}
+
+impl ExecHooks for SkipSmallCompaction {
+    fn before_a_subcompaction_start(&mut self, _cid: CId, cx: SubcompactionStartCtx<'_>) {
+        if cx.subc.size < self.size_threshold {
+            info!("Skipped a small compaction."; 
+                "size" => cx.subc.size, "threshold" => self.size_threshold);
+            cx.skip();
+        }
+    }
+}
diff --git a/components/compact-log-backup/src/execute/mod.rs b/components/compact-log-backup/src/execute/mod.rs
index d96c16abcb3..b9f2cbc71c5 100644
--- a/components/compact-log-backup/src/execute/mod.rs
+++ b/components/compact-log-backup/src/execute/mod.rs
@@ -41,8 +41,8 @@ const COMPACTION_V1_PREFIX: &str = "v1/compactions";
 /// The config for an execution of a compaction.
 ///
 /// This structure itself fully defines what work the compaction need to do.
-/// That is, keeping this structure unchanged, the compaction should always
-/// generate the same artifices.
+/// That is, keeping this structure unchanged, the compaction task generated
+/// should be the same. (But some of them may be filtered out later.)
 #[derive(Debug)]
 pub struct ExecutionConfig {
     /// Filter out files doesn't contain any record with TS great or equal than


Test patch:
diff --git a/components/compact-log-backup/src/execute/test.rs b/components/compact-log-backup/src/execute/test.rs
index 68e21bbd477..743316cdf69 100644
--- a/components/compact-log-backup/src/execute/test.rs
+++ b/components/compact-log-backup/src/execute/test.rs
@@ -21,6 +21,7 @@ use crate::{
     errors::OtherErrExt,
     exec_hooks::{
         checkpoint::Checkpoint, consistency::StorageConsistencyGuard, save_meta::SaveMeta,
+        skip_small_compaction::SkipSmallCompaction,
     },
     execute::hooking::{CId, ExecHooks, SubcompactionFinishCtx},
     storage::LOCK_PREFIX,
@@ -309,3 +310,28 @@ async fn test_abort_unlocking() {
     let l = load_locks(st.storage().as_ref()).await;
     assert_eq!(l.len(), 0, "it is {:?}", l);
 }
+
+#[tokio::test]
+async fn test_filter_out_small_compactions() {
+    let st = TmpStorage::create();
+    let mut cm = HashMap::new();
+    st.build_flush("0.log", "v1/backupmeta/0.meta", gen_builder(&mut cm, 0, 15))
+        .await;
+
+    let exec = create_compaction(st.backend());
+
+    tokio::task::spawn_blocking(move || {
+        exec.run((SkipSmallCompaction::new(27800), SaveMeta::default()))
+    })
+    .await
+    .unwrap()
+    .unwrap();
+
+    let migs = st.load_migrations().await.unwrap();
+    let c_path = &migs[0].1.compactions[0];
+    let cs = st.load_subcompactions(&c_path.artifacts).await.unwrap();
+    assert_eq!(cs.len(), 8, "{:?}", cs);
+    for c in &cs {
+        assert!(c.get_meta().get_size() >= 27800, "{:?}", c.get_meta());
+    }
+}