feat: cf support compress type (#259)

* feat: cf support compress type

* fix

Signed-off-by: happy-v587 <code4happy@gmail.com>

---------

Signed-off-by: happy-v587 <code4happy@gmail.com>

Author: happy-v587

rust-toolchain.toml

@@ -16,5 +16,5 @@
 # limitations under the License.
 
 [toolchain]
-channel = "stable"
+channel = "nightly-2025-08-20"
 components = ["rustfmt", "clippy", "rust-src", "miri", "rust-analyzer"]

src/conf/kiwi.conf

@@ -266,4 +266,14 @@ rocksdb-level-compaction-dynamic-level-bytes yes
 rocksdb-max-open-files 10000
 
 # Target file size for level-1 and above
-rocksdb-target-file-size-base 67108864
+rocksdb-target-file-size-base 67108864
+
+# Compression algorithm for RocksDB column families.
+# Applies to all column families (default, hash_data_cf, set_data_cf, list_data_cf, zset_data_cf, zset_score_cf).
+# Supported values: none, lz4, snappy, zstd, zlib, bz2
+# Default: lz4 (good balance of speed and compression ratio)
+# - lz4:    fast compression/decompression, moderate ratio (recommended for most workloads)
+# - snappy: similar to lz4, slightly lower ratio but very fast
+# - zstd:   best compression ratio, moderate speed (recommended for storage-constrained environments)
+# - none:   no compression (highest performance, largest disk usage)
+rocksdb-compression-type lz4

src/conf/src/config.rs

@@ -14,12 +14,72 @@
 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 // See the License for the specific language governing permissions and
 // limitations under the License.
+use rocksdb::DBCompressionType;
+use serde::{Deserialize, Serialize};
 use snafu::ResultExt;
+use std::fmt::Formatter;
 use validator::Validate;
 
 use crate::de_func::{parse_bool_from_string, parse_memory, parse_redis_config};
 use crate::error::Error;
 
+/// Compression algorithm for RocksDB column families.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub enum CompressionType {
+    None,
+    Snappy,
+    Lz4,
+    Zstd,
+    Zlib,
+    Bz2,
+}
+
+impl CompressionType {
+    pub fn to_rocksdb(&self) -> rocksdb::DBCompressionType {
+        match self {
+            CompressionType::None => rocksdb::DBCompressionType::None,
+            CompressionType::Snappy => rocksdb::DBCompressionType::Snappy,
+            CompressionType::Lz4 => rocksdb::DBCompressionType::Lz4,
+            CompressionType::Zstd => rocksdb::DBCompressionType::Zstd,
+            CompressionType::Zlib => rocksdb::DBCompressionType::Zlib,
+            CompressionType::Bz2 => rocksdb::DBCompressionType::Bz2,
+        }
+    }
+}
+
+impl std::fmt::Display for CompressionType {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        let s = match self {
+            CompressionType::None => "none",
+            CompressionType::Snappy => "snappy",
+            CompressionType::Lz4 => "lz4",
+            CompressionType::Zstd => "zstd",
+            CompressionType::Zlib => "zlib",
+            CompressionType::Bz2 => "bz2",
+        };
+        write!(f, "{}", s)
+    }
+}
+
+impl std::str::FromStr for CompressionType {
+    type Err = String;
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        match s.to_lowercase().as_str() {
+            "none" | "no" => Ok(CompressionType::None),
+            "snappy" => Ok(CompressionType::Snappy),
+            "lz4" => Ok(CompressionType::Lz4),
+            "zstd" => Ok(CompressionType::Zstd),
+            "zlib" => Ok(CompressionType::Zlib),
+            "bz2" => Ok(CompressionType::Bz2),
+            _ => Err(format!(
+                "unknown compression type '{}', valid values: none, snappy, lz4, zstd, zlib, bz2",
+                s
+            )),
+        }
+    }
+}
+
 const DEFAULT_BINDING: &str = "127.0.0.1";
 const DEFAULT_PORT: u16 = 7379; // Redis-compatible port (7xxx variant of 6379)
 // config struct define - keeping original config items but using Redis-style format
@@ -46,6 +106,9 @@ pub struct Config {
     pub rocksdb_level_compaction_dynamic_level_bytes: bool,
     pub rocksdb_max_open_files: i32,
     pub rocksdb_target_file_size_base: u64,
+    /// Compression algorithm applied to all column families.
+    /// Supported values: none, lz4, snappy, zstd, zlib, bz2. Default: lz4.
+    pub rocksdb_compression_type: CompressionType,
 
     // Additional fields from original config
     pub binding: String,
@@ -94,6 +157,7 @@ impl Default for Config {
             rocksdb_level_compaction_dynamic_level_bytes: true,
             rocksdb_max_open_files: 10000,
             rocksdb_target_file_size_base: 64 << 20, // 64MB
+            rocksdb_compression_type: CompressionType::Lz4,
 
             db_instance_num: 3,
             db_path: "./db".to_string(),
@@ -316,6 +380,15 @@ impl Config {
                             )),
                         })?;
                 }
+                "rocksdb-compression-type" => {
+                    config.rocksdb_compression_type =
+                        value.parse().map_err(|e| Error::InvalidConfig {
+                            source: serde_ini::de::Error::Custom(format!(
+                                "Invalid rocksdb-compression-type: {}",
+                                e
+                            )),
+                        })?;
+                }
                 "raft-node-id" => {
                     raft_node_id = Some(value.parse().map_err(|e| Error::InvalidConfig {
                         source: serde_ini::de::Error::Custom(format!(
@@ -395,6 +468,14 @@ impl Config {
         options.set_max_open_files(self.rocksdb_max_open_files);
         options.set_target_file_size_base(self.rocksdb_target_file_size_base);
 
+        // Apply compression to all levels (level 0 and 1 use no compression, levels 2+ use configured type)
+        let compression = self.rocksdb_compression_type.to_rocksdb();
+        let mut compressions = vec![DBCompressionType::None; 3];
+        for _ in 3..self.rocksdb_num_levels {
+            compressions.push(compression);
+        }
+        options.set_compression_per_level(&compressions);
+
         if self.rocksdb_periodic_second > 0 {
             options.set_periodic_compaction_seconds(self.rocksdb_periodic_second);
         }

src/server/src/main.rs

@@ -214,7 +214,7 @@ fn main() -> std::io::Result<()> {
 async fn initialize_storage(config: &Config) -> Result<GlobalStorage, DualRuntimeError> {
     info!("Initializing storage...");
 
-    let storage_options = Arc::new(StorageOptions::default());
+    let storage_options = Arc::new(StorageOptions::from_config(config));
     let db_path = PathBuf::from(&config.db_path);
 
     let mut storage = Storage::new(1, 0);

src/storage/src/options.rs

@@ -110,6 +110,19 @@ impl StorageOptions {
         Self::default()
     }
 
+    /// Build StorageOptions from a loaded [`conf::config::Config`].
+    pub fn from_config(config: &conf::config::Config) -> Self {
+        let rocksdb_opts = config.get_rocksdb_options();
+        Self {
+            options: rocksdb_opts,
+            block_cache_size: config.memory as usize,
+            small_compaction_threshold: config.small_compaction_threshold,
+            small_compaction_duration_threshold: config.small_compaction_duration_threshold,
+            db_instance_num: config.db_instance_num,
+            ..Self::default()
+        }
+    }
+
     /// Set block cache size
     pub fn set_block_cache_size(&mut self, size: usize) -> &mut Self {
         self.block_cache_size = size;