feat:progressive compact

Author: brother-jin

include/pika_server.h

@@ -522,6 +522,7 @@ class PikaServer : public pstd::noncopyable {
    */
   void DoTimingTask();
   void AutoCompactRange();
+  pstd::Status AutoCompactOldSST();
   void AutoBinlogPurge();
   void AutoServerlogPurge();
   void AutoDeleteExpiredDump();

src/pika_server.cc

@@ -1126,6 +1126,7 @@ int PikaServer::ClientPubSubChannelPatternSize(const std::shared_ptr<NetConn>& c
 void PikaServer::DoTimingTask() {
   // Maybe schedule compactrange
   AutoCompactRange();
+  AutoCompactOldSST();
   // Purge serverlog
   AutoServerlogPurge();
   // Purge binlog
@@ -1146,6 +1147,21 @@ void PikaServer::DoTimingTask() {
   StatDiskUsage();
 }
 
+Status PikaServer::AutoCompactOldSST() {
+  static uint64_t last_compact_old_time = 0;
+  auto current_time = pstd::NowMicros();
+  if (current_time - last_compact_old_time < 90 * 1000 * 1000) {
+    return Status::OK();
+  }
+
+  last_compact_old_time = current_time;
+
+  for (const auto& db_item : dbs_) {
+    db_item.second->Compact(storage::kSST);
+  }
+  return Status::OK();
+}
+
 void PikaServer::StatDiskUsage() {
   thread_local uint64_t last_update_time = 0;
   auto current_time = pstd::NowMicros();

src/storage/include/storage/storage.h

@@ -126,9 +126,9 @@ struct ScoreMember {
 
 enum BeforeOrAfter { Before, After };
 
-enum DataType { kAll, kStrings, kHashes, kLists, kZSets, kSets, kStreams };
+enum DataType { kAll, kStrings, kHashes, kLists, kZSets, kSets, kStreams, kSST };
 
-const char DataTypeTag[] = {'a', 'k', 'h', 'l', 'z', 's', 'x'};
+const char DataTypeTag[] = {'a', 'k', 'h', 'l', 'z', 's', 'x', 't'};
 
 enum class OptionType {
   kDB,
@@ -150,7 +150,9 @@ enum Operation {
   kCleanSets,
   kCleanLists,
   kCleanStreams,
-  kCompactRange
+  kCompactRange,
+  kCompactSST,
+  kCompactOldSST
 };
 
 struct BGTask {
@@ -1075,6 +1077,8 @@ class Storage {
 
   Status Compact(const DataType& type, bool sync = false);
   Status CompactRange(const DataType& type, const std::string& start, const std::string& end, bool sync = false);
+  Status CompactOldFiles(const DataType& type, bool sync = false);
+  Status DoCompactOldFiles(const DataType& type);
   Status DoCompact(const DataType& type);
   Status DoCompactRange(const DataType& type, const std::string& start, const std::string& end);
 

src/storage/src/redis.cc

@@ -5,6 +5,9 @@
 
 #include "src/redis.h"
 #include <sstream>
+#include "rocksdb/column_family.h"
+#include "rocksdb/compaction_job_stats.h"
+#include "glog/logging.h"  // 确保LOG宏可用
 
 namespace storage {
 
@@ -196,7 +199,67 @@ void Redis::SetCompactRangeOptions(const bool is_canceled) {
     default_compact_range_options_.canceled = new std::atomic<bool>(is_canceled);
   } else {
     default_compact_range_options_.canceled->store(is_canceled);
-  } 
+  }
+}
+
+uint64_t Redis::TableFileNameToNumber(const std::string& name) {
+      uint64_t number = 0;
+      uint64_t base = 1;
+      int pos = static_cast<int>(name.find_last_of('.'));
+      while (--pos >= 0 && name[pos] >= '0' && name[pos] <= '9') {
+        number += (name[pos] - '0') * base;
+        base *= 10;
+      }
+      return number;
+}
+
+Status Redis::CompactOldFiles() {
+    Status s;
+    std::vector<rocksdb::ColumnFamilyHandle*> compact_handles_;
+    for (auto handle : handles_) {
+        compact_handles_.push_back(handle);
+    }
+    if (handles_.empty()) {
+        compact_handles_.push_back(db_->DefaultColumnFamily());
+    }
+    for (auto handle : compact_handles_) {
+        int rate = 0;
+        while (rate < 70) {
+            rate = 100;
+            rocksdb::ColumnFamilyMetaData meta;
+            rocksdb::CompactionOptions compact_options_;
+            std::vector<std::string> input_file_names;
+            db_->GetColumnFamilyMetaData(handle, &meta);
+            uint64_t min_number = -1;
+            int min_level = 0;
+            std::string min_file = "999999.sst";
+            std::string file_path;
+            for (auto level_meta : meta.levels) {
+                for (auto file_meta : level_meta.files) {
+                    file_path = file_meta.db_path;
+                    uint64_t number = 0;
+                    number = TableFileNameToNumber(file_meta.name);
+                    if (number < min_number) {
+                        min_file = file_meta.name;
+                        min_level = level_meta.level;
+                        min_number = number;
+                    }
+                }
+            }
+
+            if (min_level < 6) min_level++;
+
+            if (min_file != "999999.sst") {
+                input_file_names.push_back(min_file);
+                rocksdb::CompactionJobInfo job_info;
+                db_->CompactFiles(compact_options_, handle, input_file_names, min_level, -1, nullptr, &job_info);
+                LOG(WARNING) << "CompactFiles: " << handle->GetName()<< " : " << file_path << " : " << min_file << " : " << min_level << ", job_stats : " << job_info.stats.num_input_records <<  ": "<< job_info.stats.num_output_records;
+                if (job_info.stats.num_input_records == 0) { break; }
+                rate = job_info.stats.num_output_records * 100 / job_info.stats.num_input_records;
+            }
+        }
+    }
+    return s;
 }
 
 }  // namespace storage

src/storage/src/redis.h

@@ -115,6 +115,9 @@ class Redis {
   Status SetSmallCompactionDurationThreshold(uint64_t small_compaction_duration_threshold);
   std::vector<rocksdb::ColumnFamilyHandle*> GetHandles(){ return handles_;};
   void GetRocksDBInfo(std::string &info, const char *prefix);
+  Status CompactOldFiles();
+
+  static uint64_t TableFileNameToNumber(const std::string& name);
 
  protected:
   Storage* const storage_;

src/storage/src/storage.cc

@@ -1628,6 +1628,13 @@ Status Storage::StartBGThread() {
 
 Status Storage::AddBGTask(const BGTask& bg_task) {
   bg_tasks_mutex_.lock();
+  
+  // 如果任务队列不为空且当前任务是 kCompactSST 类型，则直接返回，不添加到队列
+  if (!bg_tasks_queue_.empty() && bg_task.operation == kCompactSST) {
+    bg_tasks_mutex_.unlock();
+    return Status::OK();
+  }
+
   if (bg_task.type == kAll) {
     // if current task it is global compact,
     // clear the bg_tasks_queue_;
@@ -1658,10 +1665,14 @@ Status Storage::RunBGTask() {
 
     if (task.operation == kCleanAll) {
       DoCompact(task.type);
+    } else if (task.operation == kCompactOldSST) {
+      DoCompactRange(task.type, "", "");
     } else if (task.operation == kCompactRange) {
       if (task.argv.size() == 2) {
         DoCompactRange(task.type, task.argv.front(), task.argv.back());
       }
+    } else if (task.operation == kCompactSST) {
+      DoCompactOldFiles(task.type);
     }
   }
   return Status::OK();
@@ -1670,6 +1681,9 @@ Status Storage::RunBGTask() {
 Status Storage::Compact(const DataType& type, bool sync) {
   if (sync) {
     return DoCompact(type);
+  } else if (type == kSST) {
+    LOG(WARNING) << "AddBGTask kCompactOldSST";
+    AddBGTask({type, kCompactOldSST});
   } else {
     AddBGTask({type, kCleanAll});
   }
@@ -1722,6 +1736,48 @@ Status Storage::CompactRange(const DataType& type, const std::string& start, con
   return Status::OK();
 }
 
+Status Storage::CompactOldFiles(const DataType& type, bool sync) {
+  if (sync) {
+    return DoCompactOldFiles(type);
+  } else {
+    AddBGTask({type, kCompactSST});
+  }
+  return Status::OK();
+}
+
+Status Storage::DoCompactOldFiles(const DataType& type) {
+  Status s;
+  if (type == kStrings) {
+    current_task_type_ = Operation::kCompactSST;
+    s = strings_db_->CompactOldFiles();
+  } else if (type == kHashes) {
+    current_task_type_ = Operation::kCompactSST;
+    s = hashes_db_->CompactOldFiles();
+  } else if (type == kSets) {
+    current_task_type_ = Operation::kCompactSST;
+    s = sets_db_->CompactOldFiles();
+  } else if (type == kZSets) {
+    current_task_type_ = Operation::kCompactSST;
+    s = zsets_db_->CompactOldFiles();
+  } else if (type == kLists) {
+    current_task_type_ = Operation::kCompactSST;
+    s = lists_db_->CompactOldFiles();
+  } else if (type == kStreams) {
+    current_task_type_ = Operation::kCompactSST;
+    s = streams_db_->CompactOldFiles();
+  } else {
+    current_task_type_ = Operation::kCompactSST;
+    s = strings_db_->CompactOldFiles();
+    s = hashes_db_->CompactOldFiles();
+    s = sets_db_->CompactOldFiles();
+    s = zsets_db_->CompactOldFiles();
+    s = lists_db_->CompactOldFiles();
+    s = streams_db_->CompactOldFiles();
+  }
+  current_task_type_ = Operation::kNone;
+  return s;
+}
+
 Status Storage::DoCompactRange(const DataType& type, const std::string& start, const std::string& end) {
   Status s;
   if (type == kStrings) {