feat(storage): support w-tinylfu as block cache eviction algorithm

Cargo.toml

@@ -73,6 +73,7 @@ license = "Apache-2.0"
 repository = "https://github.com/risingwavelabs/risingwave"
 
 [workspace.dependencies]
+foyer = "0.6"
 await-tree = "0.1.1"
 aws-config = { version = "1", default-features = false, features = [
   "behavior-version-latest",
@@ -113,11 +114,7 @@ hytra = "0.1"
 rdkafka = { package = "madsim-rdkafka", version = "0.3.1", features = [
   "cmake-build",
 ] }
-hashbrown = { version = "0.14.0", features = [
-  "ahash",
-  "inline-more",
-  "nightly",
-] }
+hashbrown = { version = "0.14", features = ["ahash", "inline-more", "nightly"] }
 criterion = { version = "0.5", features = ["async_futures"] }
 tonic = { package = "madsim-tonic", version = "0.4.1" }
 tonic-build = { package = "madsim-tonic-build", version = "0.4.2" }

src/batch/Cargo.toml

@@ -21,6 +21,7 @@ assert_matches = "1"
 async-recursion = "1"
 async-trait = "0.1"
 either = "1"
+foyer = { workspace = true }
 futures = { version = "0.3", default-features = false, features = ["alloc"] }
 futures-async-stream = { workspace = true }
 futures-util = "0.3"

src/batch/src/executor/hash_agg.rs

@@ -466,6 +466,7 @@ mod tests {
 
     /// A test to verify that `HashMap` may leak memory counter when using `into_iter`.
     #[test]
+    #[should_panic] // TODO(MrCroxx): This bug is fixed and the test should panic. Remove the test and fix the related code later.
     fn test_hashmap_into_iter_bug() {
         let dropped: Arc<AtomicBool> = Arc::new(AtomicBool::new(false));
 

src/batch/src/executor/insert.rs

@@ -267,10 +267,10 @@ mod tests {
     use std::sync::Arc;
 
     use assert_matches::assert_matches;
+    use foyer::memory::CacheContext;
     use futures::StreamExt;
     use itertools::Itertools;
     use risingwave_common::array::{Array, ArrayImpl, I32Array, StructArray};
-    use risingwave_common::cache::CachePriority;
     use risingwave_common::catalog::{
         schema_test_utils, ColumnDesc, ColumnId, INITIAL_TABLE_VERSION_ID,
     };
@@ -400,7 +400,7 @@ mod tests {
                 epoch,
                 None,
                 ReadOptions {
-                    cache_policy: CachePolicy::Fill(CachePriority::High),
+                    cache_policy: CachePolicy::Fill(CacheContext::Default),
                     ..Default::default()
                 },
             )

src/common/Cargo.toml

@@ -45,6 +45,7 @@ enum-as-inner = "0.6"
 enumflags2 = { version = "0.7.8" }
 ethnum = { version = "1", features = ["serde"] }
 fixedbitset = { version = "0.5", features = ["std"] }
+foyer = { workspace = true }
 fs-err = "2"
 futures = { version = "0.3", default-features = false, features = ["alloc"] }
 governor = { version = "0.6", default-features = false, features = ["std"] }

src/common/src/config.rs

@@ -25,6 +25,7 @@ use std::num::NonZeroUsize;
 use anyhow::Context;
 use clap::ValueEnum;
 use educe::Educe;
+use foyer::memory::{LfuConfig, LruConfig};
 use risingwave_common_proc_macro::ConfigDoc;
 pub use risingwave_common_proc_macro::OverrideConfig;
 use risingwave_pb::meta::SystemParams;
@@ -567,6 +568,44 @@ impl PartialOrd for MetricLevel {
     }
 }
 
+/// the section `[storage.cache]` in `risingwave.toml`.
+#[derive(Clone, Debug, Serialize, Deserialize, DefaultFromSerde, ConfigDoc)]
+pub struct CacheConfig {
+    /// Capacity of sstable block cache.
+    #[serde(default)]
+    pub block_cache_capacity_mb: Option<usize>,
+
+    /// Capacity of sstable meta cache.
+    #[serde(default)]
+    pub meta_cache_capacity_mb: Option<usize>,
+
+    #[serde(default)]
+    pub eviction: CacheEvictionConfig,
+}
+
+/// the section `[storage.cache.eviction]` in `risingwave.toml`.
+#[derive(Clone, Debug, Serialize, Deserialize)]
+#[serde(tag = "algorithm")]
+pub enum CacheEvictionConfig {
+    Lru {
+        high_priority_ratio_in_percent: Option<usize>,
+    },
+    Lfu {
+        window_capacity_ratio_in_percent: Option<usize>,
+        protected_capacity_ratio_in_percent: Option<usize>,
+        cmsketch_eps: Option<f64>,
+        cmsketch_confidence: Option<f64>,
+    },
+}
+
+impl Default for CacheEvictionConfig {
+    fn default() -> Self {
+        Self::Lru {
+            high_priority_ratio_in_percent: None,
+        }
+    }
+}
+
 /// The section `[storage]` in `risingwave.toml`.
 #[derive(Clone, Debug, Serialize, Deserialize, DefaultFromSerde, ConfigDoc)]
 pub struct StorageConfig {
@@ -597,16 +636,8 @@ pub struct StorageConfig {
     #[serde(default = "default::storage::write_conflict_detection_enabled")]
     pub write_conflict_detection_enabled: bool,
 
-    /// Capacity of sstable block cache.
-    #[serde(default)]
-    pub block_cache_capacity_mb: Option<usize>,
-
-    #[serde(default)]
-    pub high_priority_ratio_in_percent: Option<usize>,
-
-    /// Capacity of sstable meta cache.
     #[serde(default)]
-    pub meta_cache_capacity_mb: Option<usize>,
+    pub cache: CacheConfig,
 
     /// max memory usage for large query
     #[serde(default)]
@@ -1215,6 +1246,19 @@ pub mod default {
             70
         }
 
+        pub fn window_capacity_ratio_in_percent() -> usize {
+            10
+        }
+        pub fn protected_capacity_ratio_in_percent() -> usize {
+            80
+        }
+        pub fn cmsketch_eps() -> f64 {
+            0.002
+        }
+        pub fn cmsketch_confidence() -> f64 {
+            0.95
+        }
+
         pub fn meta_cache_capacity_mb() -> usize {
             128
         }
@@ -1642,6 +1686,20 @@ pub mod default {
     }
 }
 
+#[derive(Debug, Clone)]
+pub enum EvictionConfig {
+    Lru(LruConfig),
+    Lfu(LfuConfig),
+}
+
+impl EvictionConfig {
+    pub fn for_test() -> Self {
+        Self::Lru(LruConfig {
+            high_priority_pool_ratio: 0.0,
+        })
+    }
+}
+
 pub struct StorageMemoryConfig {
     pub block_cache_capacity_mb: usize,
     pub meta_cache_capacity_mb: usize,
@@ -1650,16 +1708,18 @@ pub struct StorageMemoryConfig {
     pub meta_file_cache_ring_buffer_capacity_mb: usize,
     pub compactor_memory_limit_mb: usize,
     pub prefetch_buffer_capacity_mb: usize,
-    pub high_priority_ratio_in_percent: usize,
+    pub cache_eviction_config: EvictionConfig,
 }
 
 pub fn extract_storage_memory_config(s: &RwConfig) -> StorageMemoryConfig {
     let block_cache_capacity_mb = s
         .storage
+        .cache
         .block_cache_capacity_mb
         .unwrap_or(default::storage::block_cache_capacity_mb());
     let meta_cache_capacity_mb = s
         .storage
+        .cache
         .meta_cache_capacity_mb
         .unwrap_or(default::storage::meta_cache_capacity_mb());
     let shared_buffer_capacity_mb = s
@@ -1672,14 +1732,47 @@ pub fn extract_storage_memory_config(s: &RwConfig) -> StorageMemoryConfig {
         .storage
         .compactor_memory_limit_mb
         .unwrap_or(default::storage::compactor_memory_limit_mb());
-    let high_priority_ratio_in_percent = s
-        .storage
-        .high_priority_ratio_in_percent
-        .unwrap_or(default::storage::high_priority_ratio_in_percent());
-    let prefetch_buffer_capacity_mb = s
-        .storage
-        .shared_buffer_capacity_mb
-        .unwrap_or((100 - high_priority_ratio_in_percent) * block_cache_capacity_mb / 100);
+
+    let cache_eviction_config = match s.storage.cache.eviction {
+        CacheEvictionConfig::Lru {
+            high_priority_ratio_in_percent,
+        } => EvictionConfig::Lru(LruConfig {
+            high_priority_pool_ratio: high_priority_ratio_in_percent
+                .unwrap_or(default::storage::high_priority_ratio_in_percent())
+                as f64
+                / 100.0,
+        }),
+        CacheEvictionConfig::Lfu {
+            window_capacity_ratio_in_percent,
+            protected_capacity_ratio_in_percent,
+            cmsketch_eps,
+            cmsketch_confidence,
+        } => EvictionConfig::Lfu(LfuConfig {
+            window_capacity_ratio: window_capacity_ratio_in_percent
+                .unwrap_or(default::storage::window_capacity_ratio_in_percent())
+                as f64
+                / 100.0,
+            protected_capacity_ratio: protected_capacity_ratio_in_percent
+                .unwrap_or(default::storage::protected_capacity_ratio_in_percent())
+                as f64
+                / 100.0,
+            cmsketch_eps: cmsketch_eps.unwrap_or(default::storage::cmsketch_eps()),
+            cmsketch_confidence: cmsketch_confidence
+                .unwrap_or(default::storage::cmsketch_confidence()),
+        }),
+    };
+
+    let prefetch_buffer_capacity_mb =
+        s.storage
+            .shared_buffer_capacity_mb
+            .unwrap_or(match &cache_eviction_config {
+                EvictionConfig::Lru(lru) => {
+                    ((1.0 - lru.high_priority_pool_ratio) * block_cache_capacity_mb as f64) as usize
+                }
+                EvictionConfig::Lfu(lfu) => {
+                    ((1.0 - lfu.protected_capacity_ratio) * block_cache_capacity_mb as f64) as usize
+                }
+            });
 
     StorageMemoryConfig {
         block_cache_capacity_mb,
@@ -1689,7 +1782,7 @@ pub fn extract_storage_memory_config(s: &RwConfig) -> StorageMemoryConfig {
         meta_file_cache_ring_buffer_capacity_mb,
         compactor_memory_limit_mb,
         prefetch_buffer_capacity_mb,
-        high_priority_ratio_in_percent,
+        cache_eviction_config,
     }
 }
 

src/compute/Cargo.toml

@@ -20,6 +20,7 @@ await-tree = { workspace = true }
 chrono = { version = "0.4" }
 clap = { version = "4", features = ["derive"] }
 either = "1"
+foyer = { workspace = true }
 futures = { version = "0.3", default-features = false, features = ["alloc"] }
 futures-async-stream = { workspace = true }
 hyper = "0.14"

src/compute/src/memory/config.rs

@@ -12,7 +12,10 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use risingwave_common::config::{StorageConfig, StorageMemoryConfig};
+use foyer::memory::{LfuConfig, LruConfig};
+use risingwave_common::config::{
+    CacheEvictionConfig, EvictionConfig, StorageConfig, StorageMemoryConfig,
+};
 use risingwave_common::util::pretty_bytes::convert;
 
 /// The minimal memory requirement of computing tasks in megabytes.
@@ -32,7 +35,6 @@ const STORAGE_BLOCK_CACHE_MEMORY_PROPORTION: f64 = 0.3;
 const STORAGE_META_CACHE_MAX_MEMORY_MB: usize = 4096;
 const STORAGE_META_CACHE_MEMORY_PROPORTION: f64 = 0.35;
 const STORAGE_SHARED_BUFFER_MEMORY_PROPORTION: f64 = 0.3;
-const STORAGE_DEFAULT_HIGH_PRIORITY_BLOCK_CACHE_RATIO: usize = 50;
 
 /// Each compute node reserves some memory for stack and code segment of processes, allocation
 /// overhead, network buffer, etc. based on `SYSTEM_RESERVED_MEMORY_PROPORTION`. The reserve memory
@@ -65,30 +67,58 @@ pub fn storage_memory_config(
     } else {
         (STORAGE_MEMORY_PROPORTION + COMPACTOR_MEMORY_PROPORTION, 0.0)
     };
-    let mut block_cache_capacity_mb = storage_config.block_cache_capacity_mb.unwrap_or(
+    let mut block_cache_capacity_mb = storage_config.cache.block_cache_capacity_mb.unwrap_or(
         ((non_reserved_memory_bytes as f64
             * storage_memory_proportion
             * STORAGE_BLOCK_CACHE_MEMORY_PROPORTION)
             .ceil() as usize)
             >> 20,
     );
-    let high_priority_ratio_in_percent = storage_config
-        .high_priority_ratio_in_percent
-        .unwrap_or(STORAGE_DEFAULT_HIGH_PRIORITY_BLOCK_CACHE_RATIO);
     let default_meta_cache_capacity = (non_reserved_memory_bytes as f64
         * storage_memory_proportion
         * STORAGE_META_CACHE_MEMORY_PROPORTION)
         .ceil() as usize;
-    let meta_cache_capacity_mb = storage_config
-        .meta_cache_capacity_mb
-        .unwrap_or(std::cmp::min(
-            default_meta_cache_capacity >> 20,
-            STORAGE_META_CACHE_MAX_MEMORY_MB,
-        ));
+    let meta_cache_capacity_mb =
+        storage_config
+            .cache
+            .meta_cache_capacity_mb
+            .unwrap_or(std::cmp::min(
+                default_meta_cache_capacity >> 20,
+                STORAGE_META_CACHE_MAX_MEMORY_MB,
+            ));
 
     let prefetch_buffer_capacity_mb = storage_config
         .prefetch_buffer_capacity_mb
         .unwrap_or(block_cache_capacity_mb);
+    let cache_eviction_config = match &storage_config.cache.eviction {
+        CacheEvictionConfig::Lru {
+            high_priority_ratio_in_percent,
+        } => EvictionConfig::Lru(LruConfig {
+            high_priority_pool_ratio: high_priority_ratio_in_percent.unwrap_or(
+                risingwave_common::config::default::storage::high_priority_ratio_in_percent(),
+            ) as f64
+                / 100.0,
+        }),
+        CacheEvictionConfig::Lfu {
+            window_capacity_ratio_in_percent,
+            protected_capacity_ratio_in_percent,
+            cmsketch_eps,
+            cmsketch_confidence,
+        } => EvictionConfig::Lfu(LfuConfig {
+            window_capacity_ratio: window_capacity_ratio_in_percent.unwrap_or(
+                risingwave_common::config::default::storage::window_capacity_ratio_in_percent(),
+            ) as f64
+                / 100.0,
+            protected_capacity_ratio: protected_capacity_ratio_in_percent.unwrap_or(
+                risingwave_common::config::default::storage::protected_capacity_ratio_in_percent(),
+            ) as f64
+                / 100.0,
+            cmsketch_eps: cmsketch_eps
+                .unwrap_or(risingwave_common::config::default::storage::cmsketch_eps()),
+            cmsketch_confidence: cmsketch_confidence
+                .unwrap_or(risingwave_common::config::default::storage::cmsketch_confidence()),
+        }),
+    };
 
     if meta_cache_capacity_mb == STORAGE_META_CACHE_MAX_MEMORY_MB {
         block_cache_capacity_mb += (default_meta_cache_capacity >> 20) - meta_cache_capacity_mb;
@@ -143,7 +173,7 @@ pub fn storage_memory_config(
         meta_file_cache_ring_buffer_capacity_mb,
         compactor_memory_limit_mb,
         prefetch_buffer_capacity_mb,
-        high_priority_ratio_in_percent,
+        cache_eviction_config,
     }
 }
 
@@ -192,8 +222,8 @@ mod tests {
         assert_eq!(memory_config.meta_file_cache_ring_buffer_capacity_mb, 256);
         assert_eq!(memory_config.compactor_memory_limit_mb, 819);
 
-        storage_config.block_cache_capacity_mb = Some(512);
-        storage_config.meta_cache_capacity_mb = Some(128);
+        storage_config.cache.block_cache_capacity_mb = Some(512);
+        storage_config.cache.meta_cache_capacity_mb = Some(128);
         storage_config.shared_buffer_capacity_mb = Some(1024);
         storage_config.compactor_memory_limit_mb = Some(512);
         let memory_config = storage_memory_config(0, true, &storage_config);