feat:support rocksdb compression_per_level (#1365)

* default rocksdb l0 l1 noCompression

conf/pika.conf

@@ -135,8 +135,19 @@ max-write-buffer-num : 2
 # Limit some command response size, like Scan, Keys*
 # Supported Units [K|M|G], default unit bytes
 max-client-response-size : 1073741824
-# Compression type supported [snappy, zlib, lz4, zstd]
+# Compression type supported [snappy, zlib, lz4, zstd], if not compression required, set `none`
 compression : snappy
+
+# if the vector size is smaller than the level number, the undefined lower level uses the
+# last option in the configurable array, for example, for 3 level
+# LSM tree the following settings are the same:
+# configurable array: [none:snappy]
+# LSM settings: [none:snappy:snappy]
+# When this configurable is enabled, compression is ignored,
+# default l0 l1 noCompression, l2 and more use `compression` option
+# https://github.com/facebook/rocksdb/wiki/Compression
+# compression_per_level : [none:none:snappy:lz4:lz4]
+
 # max-background-flushes: default is 1, limited in [1, 4]
 max-background-flushes : 1
 # max-background-compactions: default is 2, limited in [1, 8]

include/pika_conf.h

@@ -17,6 +17,7 @@
 
 #include "include/pika_define.h"
 #include "include/pika_meta.h"
+#include "rocksdb/compression_type.h"
 
 #define kBinlogReadWinDefaultSize 9000
 #define kBinlogReadWinMaxSize 90000
@@ -100,6 +101,8 @@ class PikaConf : public pstd::BaseConf {
   std::string pidfile()                             { return pidfile_; }
   int binlog_file_size()                            { return binlog_file_size_; }
   PikaMeta * local_meta()                           { return local_meta_; }
+  std::vector<rocksdb::CompressionType> compression_per_level();
+  static rocksdb::CompressionType GetCompression(const std::string &value);
 
   // Setter
   void SetPort(const int value) {
@@ -327,6 +330,7 @@ class PikaConf : public pstd::BaseConf {
   std::string pidfile_;
 
   std::string compression_;
+  std::string compression_per_level_;
   int maxclients_;
   int root_connection_num_;
   std::atomic<bool> slowlog_write_errorlog_;

src/pika_conf.cc

@@ -210,6 +210,7 @@ int PikaConf::Load()
       expire_logs_days_ = 1;
   }
   GetConfStr("compression", &compression_);
+  GetConfStr("compression_per_level", &compression_per_level_);
   // set slave read only true as default
   slave_read_only_ = true;
   GetConfInt("slave-priority", &slave_priority_);
@@ -599,3 +600,36 @@ int PikaConf::ConfigRewrite() {
   }
   return WriteBack();
 }
+
+rocksdb::CompressionType PikaConf::GetCompression(const std::string &value) {
+  if (value == "snappy") {
+    return rocksdb::CompressionType::kSnappyCompression;
+  } else if (value == "zlib") {
+    return rocksdb::CompressionType::kZlibCompression;
+  } else if (value == "lz4") {
+    return rocksdb::CompressionType::kLZ4Compression;
+  } else if (value == "zstd") {
+    return rocksdb::CompressionType::kZSTD;
+  }
+  return rocksdb::CompressionType::kNoCompression;
+}
+
+std::vector<rocksdb::CompressionType> PikaConf::compression_per_level() {
+  RWLock l(&rwlock_, false);
+  std::vector<rocksdb::CompressionType> types;
+  if (compression_per_level_.empty()) {
+    return types;
+  }
+  auto left = compression_per_level_.find_first_of('[');
+  auto right = compression_per_level_.find_first_of(']');
+
+  if (left == std::string::npos || right == std::string::npos || right <= left + 1) {
+    return types;
+  }
+  std::vector<std::string> strings;
+  pstd::StringSplit(compression_per_level_.substr(left + 1, right - left - 1), ':', strings);
+  for (const auto &item: strings) {
+    types.push_back(GetCompression(pstd::StringTrim(item)));
+  }
+  return types;
+}

src/pika_server.cc

@@ -1650,22 +1650,17 @@ void PikaServer::InitStorageOptions() {
   storage_options_.options.level_compaction_dynamic_level_bytes =
       g_pika_conf->level_compaction_dynamic_level_bytes();
 
-
-  if (g_pika_conf->compression() == "none") {
-    storage_options_.options.compression =
-        rocksdb::CompressionType::kNoCompression;
-  } else if (g_pika_conf->compression() == "snappy") {
-    storage_options_.options.compression =
-        rocksdb::CompressionType::kSnappyCompression;
-  } else if (g_pika_conf->compression() == "zlib") {
-    storage_options_.options.compression =
-        rocksdb::CompressionType::kZlibCompression;
-  } else if (g_pika_conf->compression() == "lz4") {
-    storage_options_.options.compression =
-        rocksdb::CompressionType::kLZ4Compression;
-  } else if (g_pika_conf->compression() == "zstd") {
-    storage_options_.options.compression =
-        rocksdb::CompressionType::kZSTD;
+  storage_options_.options.compression =
+      PikaConf::GetCompression(g_pika_conf->compression());
+  storage_options_.options.compression_per_level =
+      g_pika_conf->compression_per_level();
+
+  //default l0 l1 noCompression l2 and more use `compression` option
+  if (storage_options_.options.compression_per_level.empty()
+      && storage_options_.options.compression != rocksdb::kNoCompression) {
+    storage_options_.options.compression_per_level.push_back(rocksdb::kNoCompression);
+    storage_options_.options.compression_per_level.push_back(rocksdb::kNoCompression);
+    storage_options_.options.compression_per_level.push_back(storage_options_.options.compression);
   }
 
   // For rocksdb::BlockBasedTableOptions