feat: add miniz compression

compression is now behind feature flags

Cargo.toml

@@ -19,8 +19,9 @@ name = "value_log"
 path = "src/lib.rs"
 
 [features]
-default = ["lz4"]
+default = []
 lz4 = ["dep:lz4_flex"]
+miniz = ["dep:miniz_oxide"]
 serde = ["dep:serde"]
 
 [dependencies]
@@ -29,6 +30,7 @@ crc32fast = "1.4.2"
 log = "0.4.21"
 lz4_flex = { version = "0.11.3", optional = true }
 min-max-heap = "1.3.0"
+miniz_oxide = { version = "0.7.3", optional = true }
 path-absolutize = "3.1.1"
 quick_cache = "0.5.1"
 serde = { version = "1.0.200", optional = true, features = ["derive", "rc"] }

README.md

@@ -31,7 +31,13 @@ Keys are limited to 65536 bytes, values are limited to 2^32 bytes.
 
 Allows using `LZ4` compression, powered by [`lz4_flex`](https://github.com/PSeitz/lz4_flex).
 
-*Enabled by default.*
+*Disabled by default.*
+
+### miniz
+
+Allows using `DEFLATE/zlib` compression, powered by [`miniz_oxide`](https://github.com/Frommi/miniz_oxide).
+
+*Disabled by default.*
 
 ### serde
 

src/compression.rs

@@ -0,0 +1,67 @@
+/// Compression type
+#[derive(Copy, Clone, Debug, Eq, PartialEq)]
+#[allow(clippy::module_name_repetitions)]
+#[cfg_attr(feature = "serde", derive(serde::Deserialize, serde::Serialize))]
+pub enum CompressionType {
+    /// No compression
+    None,
+
+    /// LZ4 compression (speed-optimized)
+    #[cfg(feature = "lz4")]
+    Lz4,
+
+    // TODO: compression level
+    /// Zlib/DEFLATE compression (space-optimized)
+    #[cfg(feature = "miniz")]
+    Miniz,
+}
+
+impl From<CompressionType> for u8 {
+    fn from(val: CompressionType) -> Self {
+        match val {
+            CompressionType::None => 0,
+
+            #[cfg(feature = "lz4")]
+            CompressionType::Lz4 => 1,
+
+            #[cfg(feature = "miniz")]
+            CompressionType::Miniz => 2,
+        }
+    }
+}
+
+impl TryFrom<u8> for CompressionType {
+    type Error = ();
+
+    fn try_from(value: u8) -> Result<Self, Self::Error> {
+        match value {
+            0 => Ok(Self::None),
+
+            #[cfg(feature = "lz4")]
+            1 => Ok(Self::Lz4),
+
+            #[cfg(feature = "miniz")]
+            2 => Ok(Self::Miniz),
+
+            _ => Err(()),
+        }
+    }
+}
+
+impl std::fmt::Display for CompressionType {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(
+            f,
+            "{}",
+            match self {
+                Self::None => "no compression",
+
+                #[cfg(feature = "lz4")]
+                Self::Lz4 => "lz4",
+
+                #[cfg(feature = "miniz")]
+                Self::Miniz => "miniz",
+            }
+        )
+    }
+}

src/config.rs

@@ -1,18 +1,23 @@
-use crate::blob_cache::BlobCache;
+use crate::{blob_cache::BlobCache, CompressionType};
 use std::sync::Arc;
 
 /// Value log configuration
 #[derive(Debug)]
 pub struct Config {
     pub(crate) segment_size_bytes: u64,
     pub(crate) blob_cache: Arc<BlobCache>,
+
+    pub(crate) compression: CompressionType,
 }
 
 impl Default for Config {
     fn default() -> Self {
         Self {
             segment_size_bytes: 256 * 1_024 * 1_024,
             blob_cache: Arc::new(BlobCache::with_capacity_bytes(16 * 1_024 * 1_024)),
+
+            // TODO: setter method
+            compression: CompressionType::None,
         }
     }
 }

src/error.rs

@@ -1,6 +1,7 @@
 use crate::{
     serde::{DeserializeError, SerializeError},
     version::Version,
+    CompressionType,
 };
 
 /// Represents errors that can occur in the value-log
@@ -17,6 +18,9 @@ pub enum Error {
 
     /// Deserialization failed
     Deserialize(DeserializeError),
+
+    /// Decompression failed
+    Decompress(CompressionType),
     // TODO:
     // /// CRC check failed
     // CrcMismatch,

src/lib.rs

@@ -89,6 +89,7 @@
 #![allow(clippy::missing_const_for_fn)]
 
 mod blob_cache;
+mod compression;
 mod config;
 mod error;
 mod handle;
@@ -106,6 +107,7 @@ mod version;
 
 pub use {
     blob_cache::BlobCache,
+    compression::CompressionType,
     config::Config,
     error::{Error, Result},
     handle::ValueHandle,

src/manifest.rs

@@ -224,6 +224,7 @@ impl SegmentManifest {
                                     .clone()
                                     .expect("should have written at least 1 item"),
                             )),
+                            compression: writer.compression,
                         },
                         gc_stats: GcStats::default(),
                     }),

src/segment/meta.rs

@@ -1,6 +1,7 @@
 use crate::{
     key_range::KeyRange,
     serde::{Deserializable, DeserializeError, Serializable, SerializeError},
+    CompressionType,
 };
 use byteorder::{BigEndian, ReadBytesExt, WriteBytesExt};
 use std::io::{Read, Write};
@@ -19,9 +20,9 @@ pub struct Metadata {
     /// true size in bytes (if no compression were used)
     pub total_uncompressed_bytes: u64,
 
-    // TODO: 1.0.0
-    ///// What type of compression is used
-    // pub compression: CompressionType,
+    /// What type of compression is used
+    pub compression: CompressionType,
+
     /// Key range
     pub key_range: KeyRange,
 }
@@ -35,6 +36,8 @@ impl Serializable for Metadata {
         writer.write_u64::<BigEndian>(self.compressed_bytes)?;
         writer.write_u64::<BigEndian>(self.total_uncompressed_bytes)?;
 
+        writer.write_u8(self.compression.into())?;
+
         self.key_range.serialize(writer)?;
 
         Ok(())
@@ -55,13 +58,17 @@ impl Deserializable for Metadata {
         let compressed_bytes = reader.read_u64::<BigEndian>()?;
         let total_uncompressed_bytes = reader.read_u64::<BigEndian>()?;
 
+        let compression = reader.read_u8()?;
+        let compression = CompressionType::try_from(compression).expect("invalid compression type");
+
         let key_range = KeyRange::deserialize(reader)?;
 
         Ok(Self {
             item_count,
             compressed_bytes,
             total_uncompressed_bytes,
             key_range,
+            compression,
         })
     }
 }

src/segment/multi_writer.rs

@@ -1,5 +1,8 @@
 use super::writer::Writer;
-use crate::id::{IdGenerator, SegmentId};
+use crate::{
+    id::{IdGenerator, SegmentId},
+    CompressionType,
+};
 use std::path::{Path, PathBuf};
 
 /// Segment writer, may write multiple segments
@@ -8,6 +11,8 @@ pub struct MultiWriter {
     target_size: u64,
     writers: Vec<Writer>,
     id_generator: IdGenerator,
+
+    compression: CompressionType,
 }
 
 impl MultiWriter {
@@ -21,6 +26,7 @@ impl MultiWriter {
         id_generator: IdGenerator,
         target_size: u64,
         folder: P,
+        compression: CompressionType,
     ) -> std::io::Result<Self> {
         let folder = folder.as_ref();
 
@@ -31,7 +37,8 @@ impl MultiWriter {
             id_generator,
             folder: folder.into(),
             target_size,
-            writers: vec![Writer::new(segment_path, segment_id)?],
+            writers: vec![Writer::new(segment_path, segment_id, compression)?],
+            compression,
         })
     }
 
@@ -67,7 +74,7 @@ impl MultiWriter {
         let new_segment_id = self.id_generator.next();
 
         self.writers
-            .push(Writer::new(&self.folder, new_segment_id)?);
+            .push(Writer::new(&self.folder, new_segment_id, self.compression)?);
 
         Ok(())
     }

src/segment/writer.rs

@@ -1,5 +1,7 @@
 use super::{meta::Metadata, trailer::SegmentFileTrailer};
-use crate::{id::SegmentId, key_range::KeyRange, serde::Serializable, value::UserKey};
+use crate::{
+    id::SegmentId, key_range::KeyRange, serde::Serializable, value::UserKey, CompressionType,
+};
 use byteorder::{BigEndian, WriteBytesExt};
 use std::{
     fs::File,
@@ -20,8 +22,10 @@ pub struct Writer {
     pub(crate) written_blob_bytes: u64,
     pub(crate) uncompressed_bytes: u64,
 
-    pub first_key: Option<UserKey>,
-    pub last_key: Option<UserKey>,
+    pub(crate) first_key: Option<UserKey>,
+    pub(crate) last_key: Option<UserKey>,
+
+    pub(crate) compression: CompressionType,
 }
 
 impl Writer {
@@ -31,7 +35,11 @@ impl Writer {
     ///
     /// Will return `Err` if an IO error occurs.
     #[doc(hidden)]
-    pub fn new<P: AsRef<Path>>(path: P, segment_id: SegmentId) -> std::io::Result<Self> {
+    pub fn new<P: AsRef<Path>>(
+        path: P,
+        segment_id: SegmentId,
+        compression: CompressionType,
+    ) -> std::io::Result<Self> {
         let path = path.as_ref();
 
         let file = File::create(path)?;
@@ -47,6 +55,8 @@ impl Writer {
 
             first_key: None,
             last_key: None,
+
+            compression,
         })
     }
 
@@ -85,8 +95,15 @@ impl Writer {
 
         self.uncompressed_bytes += value.len() as u64;
 
-        #[cfg(feature = "lz4")]
-        let value = lz4_flex::compress_prepend_size(value);
+        let value = match self.compression {
+            CompressionType::None => value.to_vec(),
+
+            #[cfg(feature = "lz4")]
+            CompressionType::Lz4 => lz4_flex::compress_prepend_size(&value),
+
+            #[cfg(feature = "miniz")]
+            CompressionType::Miniz => miniz_oxide::deflate::compress_to_vec(&value, 10),
+        };
 
         let mut hasher = crc32fast::Hasher::new();
         hasher.update(&value);
@@ -140,6 +157,7 @@ impl Writer {
                     .clone()
                     .expect("should have written at least 1 item"),
             )),
+            compression: self.compression,
         };
         metadata.serialize(&mut self.writer)?;
 

src/value_log.rs

@@ -197,15 +197,24 @@ impl ValueLog {
 
         let val_len = reader.read_u32::<BigEndian>()?;
 
-        let mut val = vec![0; val_len as usize];
-        reader.read_exact(&mut val)?;
+        let mut value = vec![0; val_len as usize];
+        reader.read_exact(&mut value)?;
 
-        #[cfg(feature = "lz4")]
-        let val = lz4_flex::decompress_size_prepended(&val).expect("should decompress");
+        let value = match segment.meta.compression {
+            crate::CompressionType::None => value,
+
+            #[cfg(feature = "lz4")]
+            crate::CompressionType::Lz4 => lz4_flex::decompress_size_prepended(&value)
+                .map_err(|_| crate::Error::Decompress(segment.meta.compression))?,
+
+            #[cfg(feature = "miniz")]
+            crate::CompressionType::Miniz => miniz_oxide::inflate::decompress_to_vec(&value)
+                .map_err(|_| crate::Error::Decompress(segment.meta.compression))?,
+        };
 
         // TODO: handle CRC
 
-        let val: UserValue = val.into();
+        let val: UserValue = value.into();
 
         self.blob_cache
             .insert((self.id, handle.clone()).into(), val.clone());
@@ -223,6 +232,7 @@ impl ValueLog {
             self.id_generator.clone(),
             self.config.segment_size_bytes,
             self.path.join(SEGMENTS_FOLDER),
+            self.config.compression,
         )?)
     }