Add RocksDB L0/L1 compression config

crates/bifrost/src/providers/local_loglet/log_store.rs

@@ -235,14 +235,8 @@ fn cf_data_options(
     opts.set_compaction_style(rocksdb::DBCompactionStyle::Level);
     opts.set_num_levels(7);
 
-    let l0_l1 = if local_loglet_config
-        .rocksdb
-        .rocksdb_disable_l0_l1_compression()
-    {
-        rocksdb::DBCompressionType::None
-    } else {
-        rocksdb::DBCompressionType::Zstd
-    };
+    let l0_l1 =
+        restate_rocksdb::configuration::l0_l1_compression_type(&local_loglet_config.rocksdb);
     let levels = restate_rocksdb::configuration::build_compression_per_level(
         7,
         l0_l1,
@@ -290,14 +284,8 @@ fn cf_metadata_options(
     // Set compactions per level
     //
     opts.set_num_levels(3);
-    let l0_l1 = if local_loglet_config
-        .rocksdb
-        .rocksdb_disable_l0_l1_compression()
-    {
-        rocksdb::DBCompressionType::None
-    } else {
-        rocksdb::DBCompressionType::Zstd
-    };
+    let l0_l1 =
+        restate_rocksdb::configuration::l0_l1_compression_type(&local_loglet_config.rocksdb);
     let levels = restate_rocksdb::configuration::build_compression_per_level(
         3,
         l0_l1,

crates/log-server/src/rocksdb_logstore/builder.rs

@@ -292,14 +292,7 @@ fn cf_data_options(
     opts.set_compaction_style(rocksdb::DBCompactionStyle::Level);
     opts.set_num_levels(7);
 
-    let l0_l1 = if log_server_config
-        .rocksdb
-        .rocksdb_disable_l0_l1_compression()
-    {
-        rocksdb::DBCompressionType::None
-    } else {
-        rocksdb::DBCompressionType::Zstd
-    };
+    let l0_l1 = restate_rocksdb::configuration::l0_l1_compression_type(&log_server_config.rocksdb);
     let levels = restate_rocksdb::configuration::build_compression_per_level(
         7,
         l0_l1,

crates/metadata-server/src/raft/storage/rocksdb_builder.rs

@@ -153,14 +153,8 @@ fn cf_data_options(
     cf_options.set_compaction_style(rocksdb::DBCompactionStyle::Level);
     cf_options.set_num_levels(7);
 
-    let l0_l1 = if metadata_server_config
-        .rocksdb
-        .rocksdb_disable_l0_l1_compression()
-    {
-        rocksdb::DBCompressionType::None
-    } else {
-        rocksdb::DBCompressionType::Zstd
-    };
+    let l0_l1 =
+        restate_rocksdb::configuration::l0_l1_compression_type(&metadata_server_config.rocksdb);
     let levels = restate_rocksdb::configuration::build_compression_per_level(
         7,
         l0_l1,
@@ -205,14 +199,8 @@ fn cf_metadata_options(
     // Set compactions per level
     //
     cf_options.set_num_levels(3);
-    let l0_l1 = if metadata_server_config
-        .rocksdb
-        .rocksdb_disable_l0_l1_compression()
-    {
-        rocksdb::DBCompressionType::None
-    } else {
-        rocksdb::DBCompressionType::Zstd
-    };
+    let l0_l1 =
+        restate_rocksdb::configuration::l0_l1_compression_type(&metadata_server_config.rocksdb);
     let levels = restate_rocksdb::configuration::build_compression_per_level(
         3,
         l0_l1,

crates/partition-store/src/partition_db.rs

@@ -660,11 +660,7 @@ impl CfConfigurator for RocksConfigurator<AllDataCf> {
         // As much as we can to increase the chances to observe a deletion.
         //
         cf_options.set_num_levels(7);
-        let l0_l1 = if config.rocksdb.rocksdb_disable_l0_l1_compression() {
-            rocksdb::DBCompressionType::None
-        } else {
-            rocksdb::DBCompressionType::Zstd
-        };
+        let l0_l1 = restate_rocksdb::configuration::l0_l1_compression_type(&config.rocksdb);
         let levels = restate_rocksdb::configuration::build_compression_per_level(
             7,
             l0_l1,

crates/rocksdb/src/configuration.rs

@@ -12,7 +12,9 @@ use std::num::NonZeroU32;
 
 use rocksdb::{BlockBasedOptions, Cache, WriteBufferManager};
 
-use restate_types::config::{RocksDbLogLevel, RocksDbOptions, StatisticsLevel};
+use restate_types::config::{
+    RocksDbL0L1Compression, RocksDbLogLevel, RocksDbOptions, StatisticsLevel,
+};
 
 use crate::logging::LoggingEventListener;
 use crate::{DbName, OpenMode, RocksAccess};
@@ -198,6 +200,20 @@ pub fn convert_log_level(input: RocksDbLogLevel) -> rocksdb::LogLevel {
     }
 }
 
+pub fn l0_l1_compression_type(opts: &RocksDbOptions) -> rocksdb::DBCompressionType {
+    convert_l0_l1_compression(opts.rocksdb_l0_l1_compression())
+}
+
+fn convert_l0_l1_compression(input: RocksDbL0L1Compression) -> rocksdb::DBCompressionType {
+    use rocksdb::DBCompressionType;
+    match input {
+        RocksDbL0L1Compression::NoCompression => DBCompressionType::None,
+        RocksDbL0L1Compression::Snappy => DBCompressionType::Snappy,
+        RocksDbL0L1Compression::Lz4 => DBCompressionType::Lz4,
+        RocksDbL0L1Compression::Zstd => DBCompressionType::Zstd,
+    }
+}
+
 /// Builds a per-level compression array for `num_levels` levels.
 ///
 /// L0 and L1 use `l0_l1_compression`, remaining levels use `upper_compression`.

crates/types/src/config/rocksdb.rs

@@ -132,6 +132,18 @@ pub struct RocksDbOptions {
     #[serde(skip_serializing_if = "Option::is_none")]
     rocksdb_disable_wal_compression: Option<bool>,
 
+    /// # RocksDB L0/L1 SST compression
+    ///
+    /// Compression algorithm for L0 and L1 SST files. Higher levels (L2+) always use Zstd.
+    /// Set to `none` to disable compression for L0/L1, which can improve write throughput at
+    /// the cost of higher disk usage since these files are short-lived and frequently compacted.
+    ///
+    /// Since v1.7.0
+    ///
+    /// Default: "zstd"
+    #[serde(skip_serializing_if = "Option::is_none")]
+    rocksdb_l0_l1_compression: Option<RocksDbL0L1Compression>,
+
     /// # Disable L0/L1 SST compression
     ///
     /// When false (the default), L0 and L1 SST files are compressed with Zstd.
@@ -140,6 +152,10 @@ pub struct RocksDbOptions {
     /// throughput at the cost of higher disk usage since these files are
     /// short-lived and frequently compacted.
     ///
+    /// Since v1.7.0, `rocksdb-l0-l1-compression` can be used to select a specific
+    /// compression algorithm. This option remains supported for backwards compatibility.
+    /// If both options are set in the same config block, this option takes precedence.
+    ///
     /// Default: false (L0/L1 compression enabled)
     #[serde(skip_serializing_if = "Option::is_none")]
     rocksdb_disable_l0_l1_compression: Option<bool>,
@@ -208,9 +224,11 @@ impl RocksDbOptions {
         if self.rocksdb_disable_wal_compression.is_none() {
             self.rocksdb_disable_wal_compression = Some(common.rocksdb_disable_wal_compression());
         }
-        if self.rocksdb_disable_l0_l1_compression.is_none() {
-            self.rocksdb_disable_l0_l1_compression =
-                Some(common.rocksdb_disable_l0_l1_compression());
+        if self.rocksdb_l0_l1_compression.is_none()
+            && self.rocksdb_disable_l0_l1_compression.is_none()
+        {
+            self.rocksdb_l0_l1_compression = common.rocksdb_l0_l1_compression;
+            self.rocksdb_disable_l0_l1_compression = common.rocksdb_disable_l0_l1_compression;
         }
     }
 
@@ -273,11 +291,35 @@ impl RocksDbOptions {
         self.rocksdb_disable_wal_compression.unwrap_or(false)
     }
 
+    pub fn rocksdb_l0_l1_compression(&self) -> RocksDbL0L1Compression {
+        if let Some(disable_l0_l1_compression) = self.rocksdb_disable_l0_l1_compression {
+            if disable_l0_l1_compression {
+                RocksDbL0L1Compression::NoCompression
+            } else {
+                RocksDbL0L1Compression::Zstd
+            }
+        } else {
+            self.rocksdb_l0_l1_compression.unwrap_or_default()
+        }
+    }
+
     pub fn rocksdb_disable_l0_l1_compression(&self) -> bool {
-        self.rocksdb_disable_l0_l1_compression.unwrap_or(false)
+        self.rocksdb_l0_l1_compression() == RocksDbL0L1Compression::NoCompression
     }
 }
 
+#[derive(Debug, Clone, Copy, Default, Serialize, Deserialize, PartialEq, Eq)]
+#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
+#[serde(rename_all = "kebab-case")]
+pub enum RocksDbL0L1Compression {
+    #[serde(rename = "none")]
+    NoCompression,
+    Snappy,
+    Lz4,
+    #[default]
+    Zstd,
+}
+
 #[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq)]
 #[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
 #[cfg_attr(feature = "schemars", schemars(rename = "RocksbStatistics"))]
@@ -318,3 +360,115 @@ pub enum PerfStatsLevel {
     /// Enables count and time stats
     EnableTime,
 }
+
+#[cfg(test)]
+mod tests {
+    use super::{RocksDbL0L1Compression, RocksDbOptions};
+
+    #[test]
+    fn l0_l1_compression_config_supports_new_and_legacy_options() {
+        let default = RocksDbOptions::default();
+        assert_eq!(
+            default.rocksdb_l0_l1_compression(),
+            RocksDbL0L1Compression::Zstd
+        );
+        assert!(!default.rocksdb_disable_l0_l1_compression());
+
+        let lz4: RocksDbOptions = toml::from_str(r#"rocksdb-l0-l1-compression = "lz4""#)
+            .expect("valid l0/l1 compression config");
+        assert_eq!(lz4.rocksdb_l0_l1_compression(), RocksDbL0L1Compression::Lz4);
+        assert!(!lz4.rocksdb_disable_l0_l1_compression());
+
+        let snappy: RocksDbOptions = toml::from_str(r#"rocksdb-l0-l1-compression = "snappy""#)
+            .expect("valid l0/l1 compression config");
+        assert_eq!(
+            snappy.rocksdb_l0_l1_compression(),
+            RocksDbL0L1Compression::Snappy
+        );
+
+        let legacy_disabled: RocksDbOptions =
+            toml::from_str("rocksdb-disable-l0-l1-compression = true")
+                .expect("valid legacy l0/l1 compression config");
+        assert_eq!(
+            legacy_disabled.rocksdb_l0_l1_compression(),
+            RocksDbL0L1Compression::NoCompression
+        );
+        assert!(legacy_disabled.rocksdb_disable_l0_l1_compression());
+
+        let legacy_enabled: RocksDbOptions =
+            toml::from_str("rocksdb-disable-l0-l1-compression = false")
+                .expect("valid legacy l0/l1 compression config");
+        assert_eq!(
+            legacy_enabled.rocksdb_l0_l1_compression(),
+            RocksDbL0L1Compression::Zstd
+        );
+
+        let legacy_config_wins: RocksDbOptions = toml::from_str(
+            r#"
+            rocksdb-l0-l1-compression = "lz4"
+            rocksdb-disable-l0-l1-compression = true
+            "#,
+        )
+        .expect("valid mixed l0/l1 compression config");
+        assert_eq!(
+            legacy_config_wins.rocksdb_l0_l1_compression(),
+            RocksDbL0L1Compression::NoCompression
+        );
+
+        let legacy_config_wins: RocksDbOptions = toml::from_str(
+            r#"
+            rocksdb-l0-l1-compression = "none"
+            rocksdb-disable-l0-l1-compression = false
+            "#,
+        )
+        .expect("valid mixed l0/l1 compression config");
+        assert_eq!(
+            legacy_config_wins.rocksdb_l0_l1_compression(),
+            RocksDbL0L1Compression::Zstd
+        );
+    }
+
+    #[test]
+    fn l0_l1_compression_common_values_apply_to_unset_local_values() {
+        let common: RocksDbOptions = toml::from_str(r#"rocksdb-l0-l1-compression = "lz4""#)
+            .expect("valid common l0/l1 compression config");
+        let mut local = RocksDbOptions::default();
+        local.apply_common(&common);
+        assert_eq!(
+            local.rocksdb_l0_l1_compression(),
+            RocksDbL0L1Compression::Lz4
+        );
+
+        let common: RocksDbOptions = toml::from_str("rocksdb-disable-l0-l1-compression = true")
+            .expect("valid common legacy l0/l1 compression config");
+        let mut local = RocksDbOptions::default();
+        local.apply_common(&common);
+        assert_eq!(
+            local.rocksdb_l0_l1_compression(),
+            RocksDbL0L1Compression::NoCompression
+        );
+    }
+
+    #[test]
+    fn l0_l1_compression_common_values_do_not_override_local_values() {
+        let common: RocksDbOptions = toml::from_str(r#"rocksdb-l0-l1-compression = "lz4""#)
+            .expect("valid common l0/l1 compression config");
+        let mut local: RocksDbOptions = toml::from_str("rocksdb-disable-l0-l1-compression = true")
+            .expect("valid local legacy l0/l1 compression config");
+        local.apply_common(&common);
+        assert_eq!(
+            local.rocksdb_l0_l1_compression(),
+            RocksDbL0L1Compression::NoCompression
+        );
+
+        let common: RocksDbOptions = toml::from_str("rocksdb-disable-l0-l1-compression = true")
+            .expect("valid common legacy l0/l1 compression config");
+        let mut local: RocksDbOptions = toml::from_str(r#"rocksdb-l0-l1-compression = "lz4""#)
+            .expect("valid local l0/l1 compression config");
+        local.apply_common(&common);
+        assert_eq!(
+            local.rocksdb_l0_l1_compression(),
+            RocksDbL0L1Compression::Lz4
+        );
+    }
+}

release-notes/unreleased/rocksdb-l0-l1-compression.md

@@ -0,0 +1,30 @@
+# Release Notes: Configure RocksDB L0/L1 Compression
+
+## New Feature
+
+### What Changed
+Restate now supports selecting the RocksDB compression algorithm for L0 and L1 SST files with `rocksdb-l0-l1-compression`.
+
+Supported values are `zstd`, `lz4`, `snappy`, and `none`. Higher RocksDB levels continue to use ZSTD.
+
+### Why This Matters
+L0 and L1 files are short-lived and frequently compacted. Operators can now test lighter-weight compression algorithms such as LZ4 or Snappy to reduce CPU overhead while still preserving some I/O reduction.
+
+### Impact on Users
+- Existing deployments keep the current ZSTD default.
+- Existing `rocksdb-disable-l0-l1-compression = true` settings continue to disable L0/L1 compression.
+- Existing `rocksdb-disable-l0-l1-compression` settings take precedence if both L0/L1 compression options are set in the same config block.
+- New deployments can opt into a lighter compression algorithm.
+
+### Migration Guidance
+To test LZ4 for L0/L1 SST files:
+
+```toml
+rocksdb-l0-l1-compression = "lz4"
+```
+
+To keep the previous explicit disable behavior:
+
+```toml
+rocksdb-l0-l1-compression = "none"
+```