From ecd01e9a850326f71eb46e52af627e687d87b2e6 Mon Sep 17 00:00:00 2001 From: Darkheir Date: Wed, 8 Jul 2026 17:26:46 +0200 Subject: [PATCH 1/6] feat: On disk cache for split footers Signed-off-by: Darkheir --- config/quickwit.yaml | 1 + docs/configuration/node-config.md | 1 + docs/overview/concepts/querying.md | 1 + .../quickwit-config/src/node_config/mod.rs | 3 + .../src/node_config/serialize.rs | 1 + quickwit/quickwit-search/src/leaf.rs | 6 +- quickwit/quickwit-search/src/lib.rs | 2 +- quickwit/quickwit-search/src/service.rs | 22 +- quickwit/quickwit-search/src/tests.rs | 4 +- quickwit/quickwit-serve/src/lib.rs | 26 +- .../src/soft_delete_api/handler.rs | 7 +- .../src/cache/disk_sized_cache.rs | 361 ++++++++++++++++++ quickwit/quickwit-storage/src/cache/mod.rs | 4 + .../src/cache/tiered_sized_cache.rs | 121 ++++++ quickwit/quickwit-storage/src/lib.rs | 3 +- quickwit/quickwit-storage/src/metrics.rs | 2 + 16 files changed, 547 insertions(+), 18 deletions(-) create mode 100644 quickwit/quickwit-storage/src/cache/disk_sized_cache.rs create mode 100644 quickwit/quickwit-storage/src/cache/tiered_sized_cache.rs diff --git a/config/quickwit.yaml b/config/quickwit.yaml index 99ebda63717..0b6ba45b236 100644 --- a/config/quickwit.yaml +++ b/config/quickwit.yaml @@ -138,6 +138,7 @@ indexer: # searcher: # fast_field_cache_capacity: 1G # split_footer_cache_capacity: 500M +# split_footer_disk_cache_capacity: 1G # partial_request_cache_capacity: 64M # max_num_concurrent_split_streams: 100 # max_num_concurrent_split_searches: 100 diff --git a/docs/configuration/node-config.md b/docs/configuration/node-config.md index ed4bca51517..1a6811ba76c 100644 --- a/docs/configuration/node-config.md +++ b/docs/configuration/node-config.md @@ -200,6 +200,7 @@ This section contains the configuration options for a Searcher. | `aggregation_bucket_limit` | Determines the maximum number of buckets returned to the client. | `65000` | | `fast_field_cache_capacity` | Fast field in memory cache capacity on a Searcher. If your filter by dates, run aggregations, range queries, or even for tracing, it might worth increasing this parameter. The [metrics](../reference/metrics.md) starting by `quickwit_cache_fastfields_cache` can help you make an informed choice when setting this value. | `1G` | | `split_footer_cache_capacity` | Split footer in memory cache (it is essentially the hotcache) capacity on a Searcher.| `500M` | +| `split_footer_disk_cache_capacity` | Opt-in, persistent on-disk cache for split footers, stored in the data directory under `searcher-split-footer-cache/`. When set, split footers survive a searcher restart, avoiding refetching the hotcache from the object store on startup. Disabled if unspecified. | | | `partial_request_cache_capacity` | Partial request in memory cache capacity on a Searcher. Cache intermediate state for a request, possibly making subsequent requests faster. It can be disabled by setting the size to `0`. | `64M` | | `max_num_concurrent_split_searches` | Maximum number of concurrent split search requests running on a Searcher. | `100` | | `split_cache` | Searcher split cache configuration options defined in the section below. Cache disabled if unspecified. | | diff --git a/docs/overview/concepts/querying.md b/docs/overview/concepts/querying.md index a883ea21dce..79ce0ab90df 100644 --- a/docs/overview/concepts/querying.md +++ b/docs/overview/concepts/querying.md @@ -98,6 +98,7 @@ In memory: On disk: - The split cache stores entire splits on disk. It can be enabled by setting the `split_cache` configuration fields. This cache can help reduce object store costs and load. Searchers populate this cache when splits are created or queried and evict them with a simple LRU strategy. +- The split footer disk cache persists split footers on disk. It can be enabled by setting the `split_footer_disk_cache_capacity` configuration value. It backs the in-memory footer cache with a persistent LRU tier so that, after a searcher restart, footers do not have to be refetched from the object store. Cached footers are promoted back into memory on the first access. Learn more about cache parameters in the [searcher configuration docs](../../configuration/node-config.md#searcher-configuration). diff --git a/quickwit/quickwit-config/src/node_config/mod.rs b/quickwit/quickwit-config/src/node_config/mod.rs index f14752574a8..60423de8200 100644 --- a/quickwit/quickwit-config/src/node_config/mod.rs +++ b/quickwit/quickwit-config/src/node_config/mod.rs @@ -274,6 +274,8 @@ pub struct SearcherConfig { pub aggregation_bucket_limit: u32, pub fast_field_cache_capacity: ByteSize, pub split_footer_cache_capacity: ByteSize, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub split_footer_disk_cache_capacity: Option, pub partial_request_cache_capacity: ByteSize, pub predicate_cache_capacity: ByteSize, pub max_num_concurrent_split_searches: usize, @@ -333,6 +335,7 @@ impl Default for SearcherConfig { SearcherConfig { fast_field_cache_capacity: ByteSize::gb(1), split_footer_cache_capacity: ByteSize::mb(500), + split_footer_disk_cache_capacity: None, partial_request_cache_capacity: ByteSize::mb(64), predicate_cache_capacity: ByteSize::mb(256), max_num_concurrent_split_searches: 100, diff --git a/quickwit/quickwit-config/src/node_config/serialize.rs b/quickwit/quickwit-config/src/node_config/serialize.rs index 1acd648729e..5524a534139 100644 --- a/quickwit/quickwit-config/src/node_config/serialize.rs +++ b/quickwit/quickwit-config/src/node_config/serialize.rs @@ -663,6 +663,7 @@ mod tests { aggregation_bucket_limit: 500_000, fast_field_cache_capacity: ByteSize::gb(10), split_footer_cache_capacity: ByteSize::gb(1), + split_footer_disk_cache_capacity: None, partial_request_cache_capacity: ByteSize::mb(64), predicate_cache_capacity: ByteSize::mb(256), max_num_concurrent_split_searches: 150, diff --git a/quickwit/quickwit-search/src/leaf.rs b/quickwit/quickwit-search/src/leaf.rs index f184375438e..36f328aeae0 100644 --- a/quickwit/quickwit-search/src/leaf.rs +++ b/quickwit/quickwit-search/src/leaf.rs @@ -35,8 +35,8 @@ use quickwit_query::query_ast::{ }; use quickwit_query::tokenizers::TokenizerManager; use quickwit_storage::{ - BundleStorage, ByteRangeCache, MemorySizedCache, OwnedBytes, SplitCache, Storage, - StorageResolver, TimeoutAndRetryStorage, wrap_storage_with_cache, + BundleStorage, ByteRangeCache, OwnedBytes, SplitCache, Storage, StorageResolver, + TieredSizedCache, TimeoutAndRetryStorage, wrap_storage_with_cache, }; use tantivy::aggregation::agg_req::{AggregationVariants, Aggregations}; use tantivy::aggregation::{AggContextParams, AggregationLimitsGuard}; @@ -60,7 +60,7 @@ use crate::{QuickwitAggregations, SearchError}; async fn get_split_footer_from_cache_or_fetch( index_storage: Arc, split_and_footer_offsets: &SplitIdAndFooterOffsets, - footer_cache: &MemorySizedCache, + footer_cache: &TieredSizedCache, ) -> anyhow::Result { { let possible_val = footer_cache.get(&split_and_footer_offsets.split_id); diff --git a/quickwit/quickwit-search/src/lib.rs b/quickwit/quickwit-search/src/lib.rs index 73d9cf05ffe..00f78e8d1c6 100644 --- a/quickwit/quickwit-search/src/lib.rs +++ b/quickwit/quickwit-search/src/lib.rs @@ -345,7 +345,7 @@ pub async fn single_node_search( let search_job_placer = SearchJobPlacer::new(searcher_pool.clone()); let cluster_client = ClusterClient::new(search_job_placer); let searcher_config = SearcherConfig::default(); - let searcher_context = Arc::new(SearcherContext::new(searcher_config, None)); + let searcher_context = Arc::new(SearcherContext::new(searcher_config, None, None)); let search_service = Arc::new(SearchServiceImpl::new( metastore.clone(), storage_resolver, diff --git a/quickwit/quickwit-search/src/service.rs b/quickwit/quickwit-search/src/service.rs index 74648c0e1eb..21ce58f6221 100644 --- a/quickwit/quickwit-search/src/service.rs +++ b/quickwit/quickwit-search/src/service.rs @@ -29,7 +29,8 @@ use quickwit_proto::search::{ SearchResponse, SnippetRequest, }; use quickwit_storage::{ - MemorySizedCache, QuickwitCache, SplitCache, StorageCache, StorageResolver, + DiskSizedCache, MemorySizedCache, QuickwitCache, SplitCache, StorageCache, StorageResolver, + TieredSizedCache, }; use tantivy::aggregation::AggregationLimitsGuard; @@ -405,8 +406,8 @@ pub struct SearcherContext { pub fast_fields_cache: Arc, /// Limit the concurrency for small snappy interactive searches. pub search_permit_provider: SearchPermitProvider, - /// Split footer cache. - pub split_footer_cache: MemorySizedCache, + /// Split footer cache. In-memory, optionally backed by a persistent on-disk tier. + pub split_footer_cache: TieredSizedCache, /// Per-split and per-query cache. pub leaf_search_cache: LeafSearchCache, /// Per-split and per-predicate cache. @@ -432,16 +433,23 @@ impl SearcherContext { #[cfg(test)] pub fn for_test() -> SearcherContext { let searcher_config = SearcherConfig::default(); - SearcherContext::new(searcher_config, None) + SearcherContext::new(searcher_config, None, None) } - /// Creates a new searcher context, given a searcher config, and an optional `SplitCache`. - pub fn new(searcher_config: SearcherConfig, split_cache_opt: Option>) -> Self { + /// Creates a new searcher context, given a searcher config, an optional `SplitCache` and an + /// optional persistent (on-disk) split footer cache. + pub fn new( + searcher_config: SearcherConfig, + split_cache_opt: Option>, + split_footer_disk_cache_opt: Option>, + ) -> Self { let capacity_in_bytes = searcher_config.split_footer_cache_capacity.as_u64() as usize; let global_split_footer_cache = MemorySizedCache::with_capacity_in_bytes( capacity_in_bytes, &quickwit_storage::STORAGE_METRICS.split_footer_cache, ); + let split_footer_cache = + TieredSizedCache::new(global_split_footer_cache, split_footer_disk_cache_opt); let leaf_search_split_semaphore = SearchPermitProvider::new( searcher_config.max_num_concurrent_split_searches, searcher_config.warmup_memory_budget, @@ -465,7 +473,7 @@ impl SearcherContext { fast_fields_cache: storage_long_term_cache, predicate_cache: predicate_cache.into(), search_permit_provider: leaf_search_split_semaphore, - split_footer_cache: global_split_footer_cache, + split_footer_cache, leaf_search_cache, list_fields_cache, split_cache_opt, diff --git a/quickwit/quickwit-search/src/tests.rs b/quickwit/quickwit-search/src/tests.rs index d20fbd881d8..a0b01a3f665 100644 --- a/quickwit/quickwit-search/src/tests.rs +++ b/quickwit/quickwit-search/src/tests.rs @@ -1125,7 +1125,7 @@ async fn test_search_util(test_sandbox: &TestSandbox, query: &str) -> Vec { ..Default::default() }); let searcher_context: Arc = - Arc::new(SearcherContext::new(SearcherConfig::default(), None)); + Arc::new(SearcherContext::new(SearcherConfig::default(), None, None)); let agg_limits = searcher_context.get_aggregation_limits(); @@ -1765,7 +1765,7 @@ async fn test_single_node_list_terms() -> anyhow::Result<()> { .into_iter() .map(|split| extract_split_and_footer_offsets(&split.split_metadata)) .collect(); - let searcher_context = Arc::new(SearcherContext::new(SearcherConfig::default(), None)); + let searcher_context = Arc::new(SearcherContext::new(SearcherConfig::default(), None, None)); { let request = ListTermsRequest { diff --git a/quickwit/quickwit-serve/src/lib.rs b/quickwit/quickwit-serve/src/lib.rs index 390519d33ae..361cb9ecef3 100644 --- a/quickwit/quickwit-serve/src/lib.rs +++ b/quickwit/quickwit-serve/src/lib.rs @@ -113,7 +113,7 @@ use quickwit_search::{ SearchJobPlacer, SearchService, SearchServiceClient, SearcherContext, SearcherPool, create_search_client_from_channel, start_searcher_service, }; -use quickwit_storage::{SplitCache, StorageResolver}; +use quickwit_storage::{DiskSizedCache, SplitCache, StorageResolver}; use tcp_listener::TcpListenerResolver; use tokio::sync::oneshot; use tonic_health::ServingStatus; @@ -658,9 +658,30 @@ pub async fn serve_quickwit( None }; + let split_footer_disk_cache_opt: Option> = if let Some(capacity) = + node_config.searcher_config.split_footer_disk_cache_capacity + { + match DiskSizedCache::open( + node_config + .data_dir_path + .join("searcher-split-footer-cache"), + capacity.as_u64(), + &quickwit_storage::STORAGE_METRICS.split_footer_disk_cache, + ) { + Ok(disk_cache) => Some(disk_cache), + Err(error) => { + error!(%error, "failed to open the persistent split footer cache, disabling it"); + None + } + } + } else { + None + }; + let searcher_context = Arc::new(SearcherContext::new( node_config.searcher_config.clone(), split_cache_opt, + split_footer_disk_cache_opt, )); let (search_job_placer, search_service) = setup_searcher( @@ -1581,7 +1602,8 @@ mod tests { #[tokio::test] async fn test_setup_searcher() { let node_config = NodeConfig::for_test(); - let searcher_context = Arc::new(SearcherContext::new(SearcherConfig::default(), None)); + let searcher_context = + Arc::new(SearcherContext::new(SearcherConfig::default(), None, None)); let metastore = metastore_for_test(); let (change_stream, change_stream_tx) = ClusterChangeStream::new_unbounded(); let storage_resolver = StorageResolver::unconfigured(); diff --git a/quickwit/quickwit-serve/src/soft_delete_api/handler.rs b/quickwit/quickwit-serve/src/soft_delete_api/handler.rs index b7000237573..009b7805cf3 100644 --- a/quickwit/quickwit-serve/src/soft_delete_api/handler.rs +++ b/quickwit/quickwit-serve/src/soft_delete_api/handler.rs @@ -207,8 +207,11 @@ mod tests { let search_job_placer = SearchJobPlacer::new(searcher_pool.clone()); let cluster_client = ClusterClient::new(search_job_placer); let searcher_config = SearcherConfig::default(); - let searcher_context = - Arc::new(quickwit_search::SearcherContext::new(searcher_config, None)); + let searcher_context = Arc::new(quickwit_search::SearcherContext::new( + searcher_config, + None, + None, + )); let search_service: Arc = Arc::new(SearchServiceImpl::new( sandbox.metastore(), sandbox.storage_resolver(), diff --git a/quickwit/quickwit-storage/src/cache/disk_sized_cache.rs b/quickwit/quickwit-storage/src/cache/disk_sized_cache.rs new file mode 100644 index 00000000000..08f179ab21c --- /dev/null +++ b/quickwit/quickwit-storage/src/cache/disk_sized_cache.rs @@ -0,0 +1,361 @@ +// Copyright 2021-Present Datadog, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// 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 std::fmt::Display; +use std::io; +use std::marker::PhantomData; +use std::path::PathBuf; +use std::sync::Mutex; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::SystemTime; + +use lru::LruCache; +use tracing::warn; + +use crate::OwnedBytes; +use crate::metrics::CacheMetrics; + +/// Substring used to mark files that are being written. +const TEMP_MARKER: &str = ".tmp"; + +/// Global counter used to build unique temporary file names. +static TEMP_COUNTER: AtomicU64 = AtomicU64::new(0); + +struct DiskCacheIndex { + /// Maps the on-disk file name to the size of its content. + /// The LRU order tracks the recency of accesses for eviction. + lru_cache: LruCache, + num_bytes: u64, + capacity_in_bytes: u64, + cache_counters: &'static CacheMetrics, +} + +impl DiskCacheIndex { + fn record_item(&mut self, num_bytes: u64) { + self.num_bytes += num_bytes; + self.cache_counters.in_cache_count.inc(); + self.cache_counters.in_cache_num_bytes.add(num_bytes as i64); + } + + fn drop_item(&mut self, num_bytes: u64) { + self.num_bytes -= num_bytes; + self.cache_counters.in_cache_count.dec(); + self.cache_counters.in_cache_num_bytes.sub(num_bytes as i64); + self.cache_counters.evict_num_items.inc(); + self.cache_counters.evict_num_bytes.inc_by(num_bytes); + } + + /// Evicts the least recently used entries until `incoming` extra bytes would fit + /// under the capacity. Returns the file names that must be deleted from disk. + fn evict_to_fit(&mut self, incoming: u64) -> Vec { + let mut victims = Vec::new(); + while self.num_bytes + incoming > self.capacity_in_bytes { + if let Some((file_name, num_bytes)) = self.lru_cache.pop_lru() { + self.drop_item(num_bytes); + victims.push(file_name); + } else { + break; + } + } + victims + } +} + +impl Drop for DiskCacheIndex { + fn drop(&mut self) { + // We don't count this toward evicted entries, as we are clearing the whole index. + self.cache_counters + .in_cache_count + .sub(self.lru_cache.len() as i64); + self.cache_counters + .in_cache_num_bytes + .sub(self.num_bytes as i64); + } +} + +/// A size-bounded cache that persists its entries on disk. +/// +/// This is the on-disk counterpart to [`MemorySizedCache`](super::MemorySizedCache): entries +/// are evicted following an LRU policy once the configured capacity (in bytes) is exceeded. +/// +/// Each entry is stored as a single file named after `key.to_string()`. Keys must therefore +/// produce filesystem-safe, collision-free strings. +pub struct DiskSizedCache { + root_path: PathBuf, + index: Mutex, + _phantom: PhantomData K>, +} + +impl DiskSizedCache { + /// Opens a disk cache rooted at `root_path`. + pub fn open( + root_path: PathBuf, + capacity_in_bytes: u64, + cache_counters: &'static CacheMetrics, + ) -> io::Result { + std::fs::create_dir_all(&root_path)?; + + let mut entries: Vec<(String, u64, SystemTime)> = Vec::new(); + for dir_entry_res in std::fs::read_dir(&root_path)? { + let dir_entry = dir_entry_res?; + let metadata = match dir_entry.metadata() { + Ok(metadata) => metadata, + Err(_) => continue, + }; + if !metadata.is_file() { + continue; + } + let file_name = match dir_entry.file_name().into_string() { + Ok(file_name) => file_name, + Err(_) => continue, + }; + if file_name.contains(TEMP_MARKER) { + // Leftover temporary file from an interrupted write: clean it up. + let _ = std::fs::remove_file(dir_entry.path()); + continue; + } + let modified = metadata.modified().unwrap_or(SystemTime::UNIX_EPOCH); + entries.push((file_name, metadata.len(), modified)); + } + entries.sort_by_key(|(_, _, modified)| *modified); + + let mut index = DiskCacheIndex { + lru_cache: LruCache::unbounded(), + num_bytes: 0, + capacity_in_bytes, + cache_counters, + }; + for (file_name, num_bytes, _) in entries { + index.record_item(num_bytes); + index.lru_cache.put(file_name, num_bytes); + } + let victims = index.evict_to_fit(0); + + let cache = DiskSizedCache { + root_path, + index: Mutex::new(index), + _phantom: PhantomData, + }; + cache.remove_files(&victims); + Ok(cache) + } + + /// Returns the cached payload for the given key, if present on disk. + pub fn get(&self, key: &K) -> Option { + let file_name = key.to_string(); + { + let mut index = self.index.lock().unwrap(); + // `LruCache::get` refreshes the recency of the entry. + if index.lru_cache.get(&file_name).is_none() { + index.cache_counters.misses_num_items.inc(); + return None; + } + } + match std::fs::read(self.root_path.join(&file_name)) { + Ok(buffer) => { + let index = self.index.lock().unwrap(); + index.cache_counters.hits_num_items.inc(); + index + .cache_counters + .hits_num_bytes + .inc_by(buffer.len() as u64); + Some(OwnedBytes::new(buffer)) + } + Err(_) => { + // The file vanished (e.g. concurrent eviction or manual deletion): drop the + // stale index entry and report a miss. + let mut index = self.index.lock().unwrap(); + if let Some(num_bytes) = index.lru_cache.pop(&file_name) { + index.drop_item(num_bytes); + } + index.cache_counters.misses_num_items.inc(); + None + } + } + } + + /// Stores the given payload on disk under the given key. + /// + /// This silently does nothing if the payload is larger than the whole cache capacity. If an + /// entry already exists for the key, it is kept as-is (payloads are assumed immutable) and its + /// recency is refreshed. + pub fn put(&self, key: K, bytes: OwnedBytes) { + let num_bytes = bytes.len() as u64; + let file_name = key.to_string(); + { + let mut index = self.index.lock().unwrap(); + if index.capacity_in_bytes < num_bytes { + if index.capacity_in_bytes != 0 { + warn!( + capacity_in_bytes = index.capacity_in_bytes, + len = num_bytes, + "footer larger than the disk cache capacity, not caching it on disk" + ); + } + return; + } + if index.lru_cache.get(&file_name).is_some() { + // Already cached: payloads are immutable, just keep the refreshed recency. + return; + } + } + + // We write the new file *before* evicting on purpose: if the write fails we return early + // without having deleted any valid entry, and the write stays outside the index lock. The + // tradeoff is that on-disk usage may transiently exceed the capacity by roughly one entry. + if let Err(error) = self.write_file(&file_name, &bytes) { + warn!(%error, file_name, "failed to persist entry to disk cache"); + return; + } + + let victims = { + let mut index = self.index.lock().unwrap(); + if index.lru_cache.get(&file_name).is_some() { + // Lost a race with a concurrent put of the same (immutable) payload. + return; + } + let victims = index.evict_to_fit(num_bytes); + index.record_item(num_bytes); + index.lru_cache.put(file_name, num_bytes); + victims + }; + self.remove_files(&victims); + } + + fn write_file(&self, file_name: &str, bytes: &[u8]) -> io::Result<()> { + let counter = TEMP_COUNTER.fetch_add(1, Ordering::Relaxed); + let tmp_path = self + .root_path + .join(format!("{file_name}{TEMP_MARKER}{counter}")); + std::fs::write(&tmp_path, bytes)?; + std::fs::rename(&tmp_path, self.root_path.join(file_name)) + } + + fn remove_files(&self, file_names: &[String]) { + for file_name in file_names { + let _ = std::fs::remove_file(self.root_path.join(file_name)); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::metrics::CACHE_METRICS_FOR_TESTS; + + fn open_cache(root_path: PathBuf, capacity_in_bytes: u64) -> DiskSizedCache { + DiskSizedCache::open(root_path, capacity_in_bytes, &CACHE_METRICS_FOR_TESTS).unwrap() + } + + #[test] + fn test_put_get() { + let tmp_dir = tempfile::tempdir().unwrap(); + let cache = open_cache(tmp_dir.path().to_path_buf(), 1_000); + assert!(cache.get(&"missing".to_string()).is_none()); + + cache.put("a".to_string(), OwnedBytes::new(&b"hello"[..])); + assert_eq!(cache.get(&"a".to_string()).unwrap(), &b"hello"[..]); + // A file should have been written on disk. + assert!(tmp_dir.path().join("a").exists()); + } + + #[test] + fn test_persists_across_reopen() { + let tmp_dir = tempfile::tempdir().unwrap(); + { + let cache = open_cache(tmp_dir.path().to_path_buf(), 1_000); + cache.put("a".to_string(), OwnedBytes::new(&b"hello"[..])); + cache.put("b".to_string(), OwnedBytes::new(&b"world"[..])); + } + // Re-opening the cache should recover the previously stored entries. + let cache = open_cache(tmp_dir.path().to_path_buf(), 1_000); + assert_eq!(cache.get(&"a".to_string()).unwrap(), &b"hello"[..]); + assert_eq!(cache.get(&"b".to_string()).unwrap(), &b"world"[..]); + } + + #[test] + fn test_lru_eviction() { + let tmp_dir = tempfile::tempdir().unwrap(); + let cache = open_cache(tmp_dir.path().to_path_buf(), 6); + cache.put("a".to_string(), OwnedBytes::new(&b"aaa"[..])); + cache.put("b".to_string(), OwnedBytes::new(&b"bbb"[..])); + // Access "a" so that "b" becomes the least recently used entry. + assert_eq!(cache.get(&"a".to_string()).unwrap(), &b"aaa"[..]); + // Inserting a third entry must evict "b". + cache.put("c".to_string(), OwnedBytes::new(&b"ccc"[..])); + + assert_eq!(cache.get(&"a".to_string()).unwrap(), &b"aaa"[..]); + assert!(cache.get(&"b".to_string()).is_none()); + assert_eq!(cache.get(&"c".to_string()).unwrap(), &b"ccc"[..]); + // The evicted entry's file must be gone. + assert!(!tmp_dir.path().join("b").exists()); + } + + #[test] + fn test_payload_larger_than_capacity_is_ignored() { + let tmp_dir = tempfile::tempdir().unwrap(); + let cache = open_cache(tmp_dir.path().to_path_buf(), 3); + cache.put("big".to_string(), OwnedBytes::new(&b"toolarge"[..])); + assert!(cache.get(&"big".to_string()).is_none()); + assert!(!tmp_dir.path().join("big").exists()); + } + + #[test] + fn test_put_same_key_twice_keeps_single_entry() { + let tmp_dir = tempfile::tempdir().unwrap(); + let cache = open_cache(tmp_dir.path().to_path_buf(), 6); + cache.put("a".to_string(), OwnedBytes::new(&b"aaa"[..])); + // Immutable payload: putting the same key again is a no-op and must not evict others. + cache.put("a".to_string(), OwnedBytes::new(&b"aaa"[..])); + cache.put("b".to_string(), OwnedBytes::new(&b"bbb"[..])); + assert_eq!(cache.get(&"a".to_string()).unwrap(), &b"aaa"[..]); + assert_eq!(cache.get(&"b".to_string()).unwrap(), &b"bbb"[..]); + } + + #[test] + fn test_get_after_manual_file_deletion() { + let tmp_dir = tempfile::tempdir().unwrap(); + let cache = open_cache(tmp_dir.path().to_path_buf(), 1_000); + cache.put("a".to_string(), OwnedBytes::new(&b"hello"[..])); + std::fs::remove_file(tmp_dir.path().join("a")).unwrap(); + // The stale entry should be detected and reported as a miss. + assert!(cache.get(&"a".to_string()).is_none()); + } + + #[test] + fn test_open_evicts_when_over_capacity() { + let tmp_dir = tempfile::tempdir().unwrap(); + { + let cache = open_cache(tmp_dir.path().to_path_buf(), 1_000); + cache.put("a".to_string(), OwnedBytes::new(&b"aaa"[..])); + cache.put("b".to_string(), OwnedBytes::new(&b"bbb"[..])); + } + // Re-open with a capacity that can only hold one of the two entries. + let cache = open_cache(tmp_dir.path().to_path_buf(), 3); + let a = cache.get(&"a".to_string()); + let b = cache.get(&"b".to_string()); + // Exactly one entry should have survived. + assert_ne!(a.is_some(), b.is_some()); + } + + #[test] + fn test_open_cleans_up_temp_files() { + let tmp_dir = tempfile::tempdir().unwrap(); + let leftover = tmp_dir.path().join("a.tmp42"); + std::fs::write(&leftover, b"partial").unwrap(); + let cache = open_cache(tmp_dir.path().to_path_buf(), 1_000); + assert!(!leftover.exists()); + assert!(cache.get(&"a".to_string()).is_none()); + } +} diff --git a/quickwit/quickwit-storage/src/cache/mod.rs b/quickwit/quickwit-storage/src/cache/mod.rs index fe2a45d2f4f..5c6fa20ff61 100644 --- a/quickwit/quickwit-storage/src/cache/mod.rs +++ b/quickwit/quickwit-storage/src/cache/mod.rs @@ -13,11 +13,13 @@ // limitations under the License. mod byte_range_cache; +mod disk_sized_cache; mod memory_sized_cache; mod quickwit_cache; mod slice_address; mod storage_with_cache; mod stored_item; +mod tiered_sized_cache; use std::ops::Range; use std::path::{Path, PathBuf}; @@ -28,7 +30,9 @@ pub use quickwit_cache::QuickwitCache; pub use storage_with_cache::StorageWithCache; pub use self::byte_range_cache::ByteRangeCache; +pub use self::disk_sized_cache::DiskSizedCache; pub use self::memory_sized_cache::MemorySizedCache; +pub use self::tiered_sized_cache::TieredSizedCache; use crate::{OwnedBytes, Storage}; /// Wraps the given directory with a slice cache that is actually global diff --git a/quickwit/quickwit-storage/src/cache/tiered_sized_cache.rs b/quickwit/quickwit-storage/src/cache/tiered_sized_cache.rs new file mode 100644 index 00000000000..dd367dd48db --- /dev/null +++ b/quickwit/quickwit-storage/src/cache/tiered_sized_cache.rs @@ -0,0 +1,121 @@ +// Copyright 2021-Present Datadog, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// 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 std::fmt::Display; +use std::hash::Hash; + +use crate::OwnedBytes; +use crate::cache::disk_sized_cache::DiskSizedCache; +use crate::cache::memory_sized_cache::MemorySizedCache; + +/// A two-tier, size-bounded cache combining a fast in-memory tier (L1) with an optional +/// disk-backed tier (L2). +/// +/// Lookups check memory first, then disk; a disk hit is promoted back into memory. Writes +/// populate both tiers. When no disk tier is configured this behaves exactly like the +/// underlying [`MemorySizedCache`], which makes it a drop-in, opt-in replacement. +pub struct TieredSizedCache { + memory: MemorySizedCache, + disk: Option>, +} + +impl TieredSizedCache { + /// Creates a tiered cache from an in-memory tier and an optional disk tier. + pub fn new(memory: MemorySizedCache, disk: Option>) -> Self { + TieredSizedCache { memory, disk } + } + + /// Returns the cached payload for the given key, checking memory first, then disk. + /// + /// A disk hit is promoted into the memory tier before being returned. + pub fn get(&self, key: &K) -> Option { + if let Some(bytes) = self.memory.get(key) { + return Some(bytes); + } + let bytes = self.disk.as_ref()?.get(key)?; + self.memory.put(key.clone(), bytes.clone()); + Some(bytes) + } + + /// Stores the given payload in both the memory tier and, if configured, the disk tier. + pub fn put(&self, key: K, bytes: OwnedBytes) { + if let Some(disk) = &self.disk { + disk.put(key.clone(), bytes.clone()); + } + self.memory.put(key, bytes); + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::metrics::CACHE_METRICS_FOR_TESTS; + + fn memory_cache() -> MemorySizedCache { + MemorySizedCache::with_capacity_in_bytes(1_000, &CACHE_METRICS_FOR_TESTS) + } + + #[test] + fn test_memory_only() { + let cache = TieredSizedCache::new(memory_cache(), None); + assert!(cache.get(&"a".to_string()).is_none()); + cache.put("a".to_string(), OwnedBytes::new(&b"hello"[..])); + assert_eq!(cache.get(&"a".to_string()).unwrap(), &b"hello"[..]); + } + + #[test] + fn test_put_writes_to_both_tiers() { + let tmp_dir = tempfile::tempdir().unwrap(); + let disk = DiskSizedCache::open( + tmp_dir.path().to_path_buf(), + 1_000, + &CACHE_METRICS_FOR_TESTS, + ) + .unwrap(); + let cache = TieredSizedCache::new(memory_cache(), Some(disk)); + cache.put("a".to_string(), OwnedBytes::new(&b"hello"[..])); + assert_eq!(cache.get(&"a".to_string()).unwrap(), &b"hello"[..]); + // The payload must have been persisted on disk as well. + assert!(tmp_dir.path().join("a").exists()); + } + + #[test] + fn test_disk_hit_is_promoted_to_memory() { + let tmp_dir = tempfile::tempdir().unwrap(); + { + let disk = DiskSizedCache::open( + tmp_dir.path().to_path_buf(), + 1_000, + &CACHE_METRICS_FOR_TESTS, + ) + .unwrap(); + let cache = TieredSizedCache::new(memory_cache(), Some(disk)); + cache.put("a".to_string(), OwnedBytes::new(&b"hello"[..])); + } + // Simulate a fresh process: only the disk tier still holds the data. + let disk = DiskSizedCache::open( + tmp_dir.path().to_path_buf(), + 1_000, + &CACHE_METRICS_FOR_TESTS, + ) + .unwrap(); + let memory = memory_cache(); + let cache = TieredSizedCache::new(memory, Some(disk)); + + assert_eq!(cache.get(&"a".to_string()).unwrap(), &b"hello"[..]); + // After a disk hit, deleting the file must not lose the value: it lives in memory now. + std::fs::remove_file(tmp_dir.path().join("a")).unwrap(); + assert_eq!(cache.get(&"a".to_string()).unwrap(), &b"hello"[..]); + } +} diff --git a/quickwit/quickwit-storage/src/lib.rs b/quickwit/quickwit-storage/src/lib.rs index 31bbddcdd89..18737ec7ad0 100644 --- a/quickwit/quickwit-storage/src/lib.rs +++ b/quickwit/quickwit-storage/src/lib.rs @@ -64,7 +64,8 @@ pub use self::bundle_storage::{BundleStorage, BundleStorageFileOffsets}; #[cfg(any(test, feature = "testsuite"))] pub use self::cache::MockStorageCache; pub use self::cache::{ - ByteRangeCache, MemorySizedCache, QuickwitCache, StorageCache, wrap_storage_with_cache, + ByteRangeCache, DiskSizedCache, MemorySizedCache, QuickwitCache, StorageCache, + TieredSizedCache, wrap_storage_with_cache, }; pub use self::local_file_storage::{LocalFileStorage, LocalFileStorageFactory}; #[cfg(feature = "azure")] diff --git a/quickwit/quickwit-storage/src/metrics.rs b/quickwit/quickwit-storage/src/metrics.rs index a4de7a53db0..4a94ac4e899 100644 --- a/quickwit/quickwit-storage/src/metrics.rs +++ b/quickwit/quickwit-storage/src/metrics.rs @@ -28,6 +28,7 @@ pub struct StorageMetrics { pub fd_cache_metrics: CacheMetrics, pub fast_field_cache: CacheMetrics, pub split_footer_cache: CacheMetrics, + pub split_footer_disk_cache: CacheMetrics, pub searcher_split_cache: CacheMetrics, pub get_slice_timeout_successes: [IntCounter; 3], pub get_slice_timeout_all_timeouts: IntCounter, @@ -66,6 +67,7 @@ impl Default for StorageMetrics { searcher_split_cache: CacheMetrics::for_component("searcher_split"), shortlived_cache: CacheMetrics::for_component("shortlived"), split_footer_cache: CacheMetrics::for_component("splitfooter"), + split_footer_disk_cache: CacheMetrics::for_component("splitfooter_disk"), get_slice_timeout_successes, get_slice_timeout_all_timeouts, object_storage_requests_total: new_counter_vec( From 303dec66f3b9bf3d0380feaf9bed51a40053be2a Mon Sep 17 00:00:00 2001 From: Darkheir Date: Wed, 8 Jul 2026 17:51:06 +0200 Subject: [PATCH 2/6] feat: Async disk cache access Signed-off-by: Darkheir --- quickwit/quickwit-search/src/leaf.rs | 12 +- .../src/cache/disk_sized_cache.rs | 181 +++++++++++------- .../src/cache/tiered_sized_cache.rs | 45 +++-- 3 files changed, 147 insertions(+), 91 deletions(-) diff --git a/quickwit/quickwit-search/src/leaf.rs b/quickwit/quickwit-search/src/leaf.rs index 36f328aeae0..02c86520c8b 100644 --- a/quickwit/quickwit-search/src/leaf.rs +++ b/quickwit/quickwit-search/src/leaf.rs @@ -63,7 +63,7 @@ async fn get_split_footer_from_cache_or_fetch( footer_cache: &TieredSizedCache, ) -> anyhow::Result { { - let possible_val = footer_cache.get(&split_and_footer_offsets.split_id); + let possible_val = footer_cache.get(&split_and_footer_offsets.split_id).await; if let Some(footer_data) = possible_val { return Ok(footer_data); } @@ -84,10 +84,12 @@ async fn get_split_footer_from_cache_or_fetch( ) })?; - footer_cache.put( - split_and_footer_offsets.split_id.to_owned(), - footer_data_opt.clone(), - ); + footer_cache + .put( + split_and_footer_offsets.split_id.to_owned(), + footer_data_opt.clone(), + ) + .await; Ok(footer_data_opt) } diff --git a/quickwit/quickwit-storage/src/cache/disk_sized_cache.rs b/quickwit/quickwit-storage/src/cache/disk_sized_cache.rs index 08f179ab21c..5df3220be3d 100644 --- a/quickwit/quickwit-storage/src/cache/disk_sized_cache.rs +++ b/quickwit/quickwit-storage/src/cache/disk_sized_cache.rs @@ -15,7 +15,7 @@ use std::fmt::Display; use std::io; use std::marker::PhantomData; -use std::path::PathBuf; +use std::path::{Path, PathBuf}; use std::sync::Mutex; use std::sync::atomic::{AtomicU64, Ordering}; use std::time::SystemTime; @@ -147,12 +147,12 @@ impl DiskSizedCache { index: Mutex::new(index), _phantom: PhantomData, }; - cache.remove_files(&victims); + remove_files(&cache.root_path, &victims); Ok(cache) } /// Returns the cached payload for the given key, if present on disk. - pub fn get(&self, key: &K) -> Option { + pub async fn get(&self, key: &K) -> Option { let file_name = key.to_string(); { let mut index = self.index.lock().unwrap(); @@ -162,8 +162,11 @@ impl DiskSizedCache { return None; } } - match std::fs::read(self.root_path.join(&file_name)) { - Ok(buffer) => { + // Offload the blocking read so we don't stall the async runtime worker. + let path = self.root_path.join(&file_name); + let read_res = tokio::task::spawn_blocking(move || std::fs::read(path)).await; + match read_res { + Ok(Ok(buffer)) => { let index = self.index.lock().unwrap(); index.cache_counters.hits_num_items.inc(); index @@ -172,7 +175,7 @@ impl DiskSizedCache { .inc_by(buffer.len() as u64); Some(OwnedBytes::new(buffer)) } - Err(_) => { + Ok(Err(_)) => { // The file vanished (e.g. concurrent eviction or manual deletion): drop the // stale index entry and report a miss. let mut index = self.index.lock().unwrap(); @@ -182,6 +185,13 @@ impl DiskSizedCache { index.cache_counters.misses_num_items.inc(); None } + Err(_join_error) => { + // The blocking read task failed unexpectedly. Keep the index entry (the file is + // likely still valid) and just report a miss. + let index = self.index.lock().unwrap(); + index.cache_counters.misses_num_items.inc(); + None + } } } @@ -190,7 +200,7 @@ impl DiskSizedCache { /// This silently does nothing if the payload is larger than the whole cache capacity. If an /// entry already exists for the key, it is kept as-is (payloads are assumed immutable) and its /// recency is refreshed. - pub fn put(&self, key: K, bytes: OwnedBytes) { + pub async fn put(&self, key: K, bytes: OwnedBytes) { let num_bytes = bytes.len() as u64; let file_name = key.to_string(); { @@ -214,9 +224,19 @@ impl DiskSizedCache { // We write the new file *before* evicting on purpose: if the write fails we return early // without having deleted any valid entry, and the write stays outside the index lock. The // tradeoff is that on-disk usage may transiently exceed the capacity by roughly one entry. - if let Err(error) = self.write_file(&file_name, &bytes) { - warn!(%error, file_name, "failed to persist entry to disk cache"); - return; + // The blocking write is offloaded so we don't stall the async runtime worker. + let write_res = { + let root_path = self.root_path.clone(); + let write_name = file_name.clone(); + tokio::task::spawn_blocking(move || write_file(&root_path, &write_name, &bytes)).await + }; + match write_res { + Ok(Ok(())) => {} + Ok(Err(error)) => { + warn!(%error, file_name, "failed to persist entry to disk cache"); + return; + } + Err(_join_error) => return, } let victims = { @@ -230,22 +250,23 @@ impl DiskSizedCache { index.lru_cache.put(file_name, num_bytes); victims }; - self.remove_files(&victims); + if !victims.is_empty() { + let root_path = self.root_path.clone(); + let _ = tokio::task::spawn_blocking(move || remove_files(&root_path, &victims)).await; + } } +} - fn write_file(&self, file_name: &str, bytes: &[u8]) -> io::Result<()> { - let counter = TEMP_COUNTER.fetch_add(1, Ordering::Relaxed); - let tmp_path = self - .root_path - .join(format!("{file_name}{TEMP_MARKER}{counter}")); - std::fs::write(&tmp_path, bytes)?; - std::fs::rename(&tmp_path, self.root_path.join(file_name)) - } +fn write_file(root_path: &Path, file_name: &str, bytes: &[u8]) -> io::Result<()> { + let counter = TEMP_COUNTER.fetch_add(1, Ordering::Relaxed); + let tmp_path = root_path.join(format!("{file_name}{TEMP_MARKER}{counter}")); + std::fs::write(&tmp_path, bytes)?; + std::fs::rename(&tmp_path, root_path.join(file_name)) +} - fn remove_files(&self, file_names: &[String]) { - for file_name in file_names { - let _ = std::fs::remove_file(self.root_path.join(file_name)); - } +fn remove_files(root_path: &Path, file_names: &[String]) { + for file_name in file_names { + let _ = std::fs::remove_file(root_path.join(file_name)); } } @@ -258,104 +279,130 @@ mod tests { DiskSizedCache::open(root_path, capacity_in_bytes, &CACHE_METRICS_FOR_TESTS).unwrap() } - #[test] - fn test_put_get() { + #[tokio::test] + async fn test_put_get() { let tmp_dir = tempfile::tempdir().unwrap(); let cache = open_cache(tmp_dir.path().to_path_buf(), 1_000); - assert!(cache.get(&"missing".to_string()).is_none()); + assert!(cache.get(&"missing".to_string()).await.is_none()); - cache.put("a".to_string(), OwnedBytes::new(&b"hello"[..])); - assert_eq!(cache.get(&"a".to_string()).unwrap(), &b"hello"[..]); + cache + .put("a".to_string(), OwnedBytes::new(&b"hello"[..])) + .await; + assert_eq!(cache.get(&"a".to_string()).await.unwrap(), &b"hello"[..]); // A file should have been written on disk. assert!(tmp_dir.path().join("a").exists()); } - #[test] - fn test_persists_across_reopen() { + #[tokio::test] + async fn test_persists_across_reopen() { let tmp_dir = tempfile::tempdir().unwrap(); { let cache = open_cache(tmp_dir.path().to_path_buf(), 1_000); - cache.put("a".to_string(), OwnedBytes::new(&b"hello"[..])); - cache.put("b".to_string(), OwnedBytes::new(&b"world"[..])); + cache + .put("a".to_string(), OwnedBytes::new(&b"hello"[..])) + .await; + cache + .put("b".to_string(), OwnedBytes::new(&b"world"[..])) + .await; } // Re-opening the cache should recover the previously stored entries. let cache = open_cache(tmp_dir.path().to_path_buf(), 1_000); - assert_eq!(cache.get(&"a".to_string()).unwrap(), &b"hello"[..]); - assert_eq!(cache.get(&"b".to_string()).unwrap(), &b"world"[..]); + assert_eq!(cache.get(&"a".to_string()).await.unwrap(), &b"hello"[..]); + assert_eq!(cache.get(&"b".to_string()).await.unwrap(), &b"world"[..]); } - #[test] - fn test_lru_eviction() { + #[tokio::test] + async fn test_lru_eviction() { let tmp_dir = tempfile::tempdir().unwrap(); let cache = open_cache(tmp_dir.path().to_path_buf(), 6); - cache.put("a".to_string(), OwnedBytes::new(&b"aaa"[..])); - cache.put("b".to_string(), OwnedBytes::new(&b"bbb"[..])); + cache + .put("a".to_string(), OwnedBytes::new(&b"aaa"[..])) + .await; + cache + .put("b".to_string(), OwnedBytes::new(&b"bbb"[..])) + .await; // Access "a" so that "b" becomes the least recently used entry. - assert_eq!(cache.get(&"a".to_string()).unwrap(), &b"aaa"[..]); + assert_eq!(cache.get(&"a".to_string()).await.unwrap(), &b"aaa"[..]); // Inserting a third entry must evict "b". - cache.put("c".to_string(), OwnedBytes::new(&b"ccc"[..])); + cache + .put("c".to_string(), OwnedBytes::new(&b"ccc"[..])) + .await; - assert_eq!(cache.get(&"a".to_string()).unwrap(), &b"aaa"[..]); - assert!(cache.get(&"b".to_string()).is_none()); - assert_eq!(cache.get(&"c".to_string()).unwrap(), &b"ccc"[..]); + assert_eq!(cache.get(&"a".to_string()).await.unwrap(), &b"aaa"[..]); + assert!(cache.get(&"b".to_string()).await.is_none()); + assert_eq!(cache.get(&"c".to_string()).await.unwrap(), &b"ccc"[..]); // The evicted entry's file must be gone. assert!(!tmp_dir.path().join("b").exists()); } - #[test] - fn test_payload_larger_than_capacity_is_ignored() { + #[tokio::test] + async fn test_payload_larger_than_capacity_is_ignored() { let tmp_dir = tempfile::tempdir().unwrap(); let cache = open_cache(tmp_dir.path().to_path_buf(), 3); - cache.put("big".to_string(), OwnedBytes::new(&b"toolarge"[..])); - assert!(cache.get(&"big".to_string()).is_none()); + cache + .put("big".to_string(), OwnedBytes::new(&b"toolarge"[..])) + .await; + assert!(cache.get(&"big".to_string()).await.is_none()); assert!(!tmp_dir.path().join("big").exists()); } - #[test] - fn test_put_same_key_twice_keeps_single_entry() { + #[tokio::test] + async fn test_put_same_key_twice_keeps_single_entry() { let tmp_dir = tempfile::tempdir().unwrap(); let cache = open_cache(tmp_dir.path().to_path_buf(), 6); - cache.put("a".to_string(), OwnedBytes::new(&b"aaa"[..])); + cache + .put("a".to_string(), OwnedBytes::new(&b"aaa"[..])) + .await; // Immutable payload: putting the same key again is a no-op and must not evict others. - cache.put("a".to_string(), OwnedBytes::new(&b"aaa"[..])); - cache.put("b".to_string(), OwnedBytes::new(&b"bbb"[..])); - assert_eq!(cache.get(&"a".to_string()).unwrap(), &b"aaa"[..]); - assert_eq!(cache.get(&"b".to_string()).unwrap(), &b"bbb"[..]); + cache + .put("a".to_string(), OwnedBytes::new(&b"aaa"[..])) + .await; + cache + .put("b".to_string(), OwnedBytes::new(&b"bbb"[..])) + .await; + assert_eq!(cache.get(&"a".to_string()).await.unwrap(), &b"aaa"[..]); + assert_eq!(cache.get(&"b".to_string()).await.unwrap(), &b"bbb"[..]); } - #[test] - fn test_get_after_manual_file_deletion() { + #[tokio::test] + async fn test_get_after_manual_file_deletion() { let tmp_dir = tempfile::tempdir().unwrap(); let cache = open_cache(tmp_dir.path().to_path_buf(), 1_000); - cache.put("a".to_string(), OwnedBytes::new(&b"hello"[..])); + cache + .put("a".to_string(), OwnedBytes::new(&b"hello"[..])) + .await; std::fs::remove_file(tmp_dir.path().join("a")).unwrap(); // The stale entry should be detected and reported as a miss. - assert!(cache.get(&"a".to_string()).is_none()); + assert!(cache.get(&"a".to_string()).await.is_none()); } - #[test] - fn test_open_evicts_when_over_capacity() { + #[tokio::test] + async fn test_open_evicts_when_over_capacity() { let tmp_dir = tempfile::tempdir().unwrap(); { let cache = open_cache(tmp_dir.path().to_path_buf(), 1_000); - cache.put("a".to_string(), OwnedBytes::new(&b"aaa"[..])); - cache.put("b".to_string(), OwnedBytes::new(&b"bbb"[..])); + cache + .put("a".to_string(), OwnedBytes::new(&b"aaa"[..])) + .await; + cache + .put("b".to_string(), OwnedBytes::new(&b"bbb"[..])) + .await; } // Re-open with a capacity that can only hold one of the two entries. let cache = open_cache(tmp_dir.path().to_path_buf(), 3); - let a = cache.get(&"a".to_string()); - let b = cache.get(&"b".to_string()); + let a = cache.get(&"a".to_string()).await; + let b = cache.get(&"b".to_string()).await; // Exactly one entry should have survived. assert_ne!(a.is_some(), b.is_some()); } - #[test] - fn test_open_cleans_up_temp_files() { + #[tokio::test] + async fn test_open_cleans_up_temp_files() { let tmp_dir = tempfile::tempdir().unwrap(); let leftover = tmp_dir.path().join("a.tmp42"); std::fs::write(&leftover, b"partial").unwrap(); let cache = open_cache(tmp_dir.path().to_path_buf(), 1_000); assert!(!leftover.exists()); - assert!(cache.get(&"a".to_string()).is_none()); + assert!(cache.get(&"a".to_string()).await.is_none()); } } diff --git a/quickwit/quickwit-storage/src/cache/tiered_sized_cache.rs b/quickwit/quickwit-storage/src/cache/tiered_sized_cache.rs index dd367dd48db..20d0fbfbf26 100644 --- a/quickwit/quickwit-storage/src/cache/tiered_sized_cache.rs +++ b/quickwit/quickwit-storage/src/cache/tiered_sized_cache.rs @@ -38,20 +38,21 @@ impl TieredSizedCache { /// Returns the cached payload for the given key, checking memory first, then disk. /// - /// A disk hit is promoted into the memory tier before being returned. - pub fn get(&self, key: &K) -> Option { + /// A disk hit is promoted into the memory tier before being returned. Only the disk tier + /// performs (off-thread) I/O, so an in-memory hit stays fully synchronous and cheap. + pub async fn get(&self, key: &K) -> Option { if let Some(bytes) = self.memory.get(key) { return Some(bytes); } - let bytes = self.disk.as_ref()?.get(key)?; + let bytes = self.disk.as_ref()?.get(key).await?; self.memory.put(key.clone(), bytes.clone()); Some(bytes) } /// Stores the given payload in both the memory tier and, if configured, the disk tier. - pub fn put(&self, key: K, bytes: OwnedBytes) { + pub async fn put(&self, key: K, bytes: OwnedBytes) { if let Some(disk) = &self.disk { - disk.put(key.clone(), bytes.clone()); + disk.put(key.clone(), bytes.clone()).await; } self.memory.put(key, bytes); } @@ -66,16 +67,18 @@ mod tests { MemorySizedCache::with_capacity_in_bytes(1_000, &CACHE_METRICS_FOR_TESTS) } - #[test] - fn test_memory_only() { + #[tokio::test] + async fn test_memory_only() { let cache = TieredSizedCache::new(memory_cache(), None); - assert!(cache.get(&"a".to_string()).is_none()); - cache.put("a".to_string(), OwnedBytes::new(&b"hello"[..])); - assert_eq!(cache.get(&"a".to_string()).unwrap(), &b"hello"[..]); + assert!(cache.get(&"a".to_string()).await.is_none()); + cache + .put("a".to_string(), OwnedBytes::new(&b"hello"[..])) + .await; + assert_eq!(cache.get(&"a".to_string()).await.unwrap(), &b"hello"[..]); } - #[test] - fn test_put_writes_to_both_tiers() { + #[tokio::test] + async fn test_put_writes_to_both_tiers() { let tmp_dir = tempfile::tempdir().unwrap(); let disk = DiskSizedCache::open( tmp_dir.path().to_path_buf(), @@ -84,14 +87,16 @@ mod tests { ) .unwrap(); let cache = TieredSizedCache::new(memory_cache(), Some(disk)); - cache.put("a".to_string(), OwnedBytes::new(&b"hello"[..])); - assert_eq!(cache.get(&"a".to_string()).unwrap(), &b"hello"[..]); + cache + .put("a".to_string(), OwnedBytes::new(&b"hello"[..])) + .await; + assert_eq!(cache.get(&"a".to_string()).await.unwrap(), &b"hello"[..]); // The payload must have been persisted on disk as well. assert!(tmp_dir.path().join("a").exists()); } - #[test] - fn test_disk_hit_is_promoted_to_memory() { + #[tokio::test] + async fn test_disk_hit_is_promoted_to_memory() { let tmp_dir = tempfile::tempdir().unwrap(); { let disk = DiskSizedCache::open( @@ -101,7 +106,9 @@ mod tests { ) .unwrap(); let cache = TieredSizedCache::new(memory_cache(), Some(disk)); - cache.put("a".to_string(), OwnedBytes::new(&b"hello"[..])); + cache + .put("a".to_string(), OwnedBytes::new(&b"hello"[..])) + .await; } // Simulate a fresh process: only the disk tier still holds the data. let disk = DiskSizedCache::open( @@ -113,9 +120,9 @@ mod tests { let memory = memory_cache(); let cache = TieredSizedCache::new(memory, Some(disk)); - assert_eq!(cache.get(&"a".to_string()).unwrap(), &b"hello"[..]); + assert_eq!(cache.get(&"a".to_string()).await.unwrap(), &b"hello"[..]); // After a disk hit, deleting the file must not lose the value: it lives in memory now. std::fs::remove_file(tmp_dir.path().join("a")).unwrap(); - assert_eq!(cache.get(&"a".to_string()).unwrap(), &b"hello"[..]); + assert_eq!(cache.get(&"a".to_string()).await.unwrap(), &b"hello"[..]); } } From aa723c248517b6b5ed9c55cdee511940e67decac Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Rapha=C3=ABl=20Cohen?= Date: Wed, 8 Jul 2026 18:16:09 +0200 Subject: [PATCH 3/6] Apply suggestions from code review Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- .../src/cache/disk_sized_cache.rs | 17 +++++++++++------ .../src/cache/tiered_sized_cache.rs | 7 ++++--- 2 files changed, 15 insertions(+), 9 deletions(-) diff --git a/quickwit/quickwit-storage/src/cache/disk_sized_cache.rs b/quickwit/quickwit-storage/src/cache/disk_sized_cache.rs index 5df3220be3d..6b91da6e981 100644 --- a/quickwit/quickwit-storage/src/cache/disk_sized_cache.rs +++ b/quickwit/quickwit-storage/src/cache/disk_sized_cache.rs @@ -180,7 +180,12 @@ impl DiskSizedCache { // stale index entry and report a miss. let mut index = self.index.lock().unwrap(); if let Some(num_bytes) = index.lru_cache.pop(&file_name) { - index.drop_item(num_bytes); + index.num_bytes -= num_bytes; + index.cache_counters.in_cache_count.dec(); + index + .cache_counters + .in_cache_num_bytes + .sub(num_bytes as i64); } index.cache_counters.misses_num_items.inc(); None @@ -210,7 +215,7 @@ impl DiskSizedCache { warn!( capacity_in_bytes = index.capacity_in_bytes, len = num_bytes, - "footer larger than the disk cache capacity, not caching it on disk" + "payload larger than the disk cache capacity, not caching it on disk" ); } return; @@ -290,7 +295,7 @@ mod tests { .await; assert_eq!(cache.get(&"a".to_string()).await.unwrap(), &b"hello"[..]); // A file should have been written on disk. - assert!(tmp_dir.path().join("a").exists()); + assert!(tmp_dir.path().join("a").try_exists().unwrap()); } #[tokio::test] @@ -332,7 +337,7 @@ mod tests { assert!(cache.get(&"b".to_string()).await.is_none()); assert_eq!(cache.get(&"c".to_string()).await.unwrap(), &b"ccc"[..]); // The evicted entry's file must be gone. - assert!(!tmp_dir.path().join("b").exists()); + assert!(!tmp_dir.path().join("b").try_exists().unwrap()); } #[tokio::test] @@ -343,7 +348,7 @@ mod tests { .put("big".to_string(), OwnedBytes::new(&b"toolarge"[..])) .await; assert!(cache.get(&"big".to_string()).await.is_none()); - assert!(!tmp_dir.path().join("big").exists()); + assert!(!tmp_dir.path().join("big").try_exists().unwrap()); } #[tokio::test] @@ -402,7 +407,7 @@ mod tests { let leftover = tmp_dir.path().join("a.tmp42"); std::fs::write(&leftover, b"partial").unwrap(); let cache = open_cache(tmp_dir.path().to_path_buf(), 1_000); - assert!(!leftover.exists()); + assert!(!leftover.try_exists().unwrap()); assert!(cache.get(&"a".to_string()).await.is_none()); } } diff --git a/quickwit/quickwit-storage/src/cache/tiered_sized_cache.rs b/quickwit/quickwit-storage/src/cache/tiered_sized_cache.rs index 20d0fbfbf26..30f994d6fea 100644 --- a/quickwit/quickwit-storage/src/cache/tiered_sized_cache.rs +++ b/quickwit/quickwit-storage/src/cache/tiered_sized_cache.rs @@ -51,10 +51,11 @@ impl TieredSizedCache { /// Stores the given payload in both the memory tier and, if configured, the disk tier. pub async fn put(&self, key: K, bytes: OwnedBytes) { + // Populate L1 first so callers benefit immediately even if the disk tier is slow. + self.memory.put(key.clone(), bytes.clone()); if let Some(disk) = &self.disk { - disk.put(key.clone(), bytes.clone()).await; + disk.put(key, bytes).await; } - self.memory.put(key, bytes); } } @@ -92,7 +93,7 @@ mod tests { .await; assert_eq!(cache.get(&"a".to_string()).await.unwrap(), &b"hello"[..]); // The payload must have been persisted on disk as well. - assert!(tmp_dir.path().join("a").exists()); + assert!(tmp_dir.path().join("a").try_exists().unwrap()); } #[tokio::test] From 9a1d10f57d543e9ecbdb2c3795ec72ca274ac08b Mon Sep 17 00:00:00 2001 From: Darkheir Date: Thu, 9 Jul 2026 10:08:40 +0200 Subject: [PATCH 4/6] feat: Async disk cache open Signed-off-by: Darkheir --- quickwit/quickwit-serve/src/lib.rs | 4 +- .../src/cache/disk_sized_cache.rs | 58 +++++++++++++------ .../src/cache/tiered_sized_cache.rs | 3 + 3 files changed, 46 insertions(+), 19 deletions(-) diff --git a/quickwit/quickwit-serve/src/lib.rs b/quickwit/quickwit-serve/src/lib.rs index 361cb9ecef3..bb8eefc53ad 100644 --- a/quickwit/quickwit-serve/src/lib.rs +++ b/quickwit/quickwit-serve/src/lib.rs @@ -667,7 +667,9 @@ pub async fn serve_quickwit( .join("searcher-split-footer-cache"), capacity.as_u64(), &quickwit_storage::STORAGE_METRICS.split_footer_disk_cache, - ) { + ) + .await + { Ok(disk_cache) => Some(disk_cache), Err(error) => { error!(%error, "failed to open the persistent split footer cache, disabling it"); diff --git a/quickwit/quickwit-storage/src/cache/disk_sized_cache.rs b/quickwit/quickwit-storage/src/cache/disk_sized_cache.rs index 6b91da6e981..94056d23058 100644 --- a/quickwit/quickwit-storage/src/cache/disk_sized_cache.rs +++ b/quickwit/quickwit-storage/src/cache/disk_sized_cache.rs @@ -99,7 +99,22 @@ pub struct DiskSizedCache { impl DiskSizedCache { /// Opens a disk cache rooted at `root_path`. - pub fn open( + pub async fn open( + root_path: PathBuf, + capacity_in_bytes: u64, + cache_counters: &'static CacheMetrics, + ) -> io::Result + where + K: 'static, + { + tokio::task::spawn_blocking(move || { + Self::open_blocking(root_path, capacity_in_bytes, cache_counters) + }) + .await + .map_err(io::Error::other)? + } + + fn open_blocking( root_path: PathBuf, capacity_in_bytes: u64, cache_counters: &'static CacheMetrics, @@ -109,11 +124,9 @@ impl DiskSizedCache { let mut entries: Vec<(String, u64, SystemTime)> = Vec::new(); for dir_entry_res in std::fs::read_dir(&root_path)? { let dir_entry = dir_entry_res?; - let metadata = match dir_entry.metadata() { - Ok(metadata) => metadata, - Err(_) => continue, - }; - if !metadata.is_file() { + if let Ok(file_type) = dir_entry.file_type() + && !file_type.is_file() + { continue; } let file_name = match dir_entry.file_name().into_string() { @@ -125,6 +138,13 @@ impl DiskSizedCache { let _ = std::fs::remove_file(dir_entry.path()); continue; } + let metadata = match dir_entry.metadata() { + Ok(metadata) => metadata, + Err(_) => continue, + }; + if !metadata.is_file() { + continue; + } let modified = metadata.modified().unwrap_or(SystemTime::UNIX_EPOCH); entries.push((file_name, metadata.len(), modified)); } @@ -280,14 +300,16 @@ mod tests { use super::*; use crate::metrics::CACHE_METRICS_FOR_TESTS; - fn open_cache(root_path: PathBuf, capacity_in_bytes: u64) -> DiskSizedCache { - DiskSizedCache::open(root_path, capacity_in_bytes, &CACHE_METRICS_FOR_TESTS).unwrap() + async fn open_cache(root_path: PathBuf, capacity_in_bytes: u64) -> DiskSizedCache { + DiskSizedCache::open(root_path, capacity_in_bytes, &CACHE_METRICS_FOR_TESTS) + .await + .unwrap() } #[tokio::test] async fn test_put_get() { let tmp_dir = tempfile::tempdir().unwrap(); - let cache = open_cache(tmp_dir.path().to_path_buf(), 1_000); + let cache = open_cache(tmp_dir.path().to_path_buf(), 1_000).await; assert!(cache.get(&"missing".to_string()).await.is_none()); cache @@ -302,7 +324,7 @@ mod tests { async fn test_persists_across_reopen() { let tmp_dir = tempfile::tempdir().unwrap(); { - let cache = open_cache(tmp_dir.path().to_path_buf(), 1_000); + let cache = open_cache(tmp_dir.path().to_path_buf(), 1_000).await; cache .put("a".to_string(), OwnedBytes::new(&b"hello"[..])) .await; @@ -311,7 +333,7 @@ mod tests { .await; } // Re-opening the cache should recover the previously stored entries. - let cache = open_cache(tmp_dir.path().to_path_buf(), 1_000); + let cache = open_cache(tmp_dir.path().to_path_buf(), 1_000).await; assert_eq!(cache.get(&"a".to_string()).await.unwrap(), &b"hello"[..]); assert_eq!(cache.get(&"b".to_string()).await.unwrap(), &b"world"[..]); } @@ -319,7 +341,7 @@ mod tests { #[tokio::test] async fn test_lru_eviction() { let tmp_dir = tempfile::tempdir().unwrap(); - let cache = open_cache(tmp_dir.path().to_path_buf(), 6); + let cache = open_cache(tmp_dir.path().to_path_buf(), 6).await; cache .put("a".to_string(), OwnedBytes::new(&b"aaa"[..])) .await; @@ -343,7 +365,7 @@ mod tests { #[tokio::test] async fn test_payload_larger_than_capacity_is_ignored() { let tmp_dir = tempfile::tempdir().unwrap(); - let cache = open_cache(tmp_dir.path().to_path_buf(), 3); + let cache = open_cache(tmp_dir.path().to_path_buf(), 3).await; cache .put("big".to_string(), OwnedBytes::new(&b"toolarge"[..])) .await; @@ -354,7 +376,7 @@ mod tests { #[tokio::test] async fn test_put_same_key_twice_keeps_single_entry() { let tmp_dir = tempfile::tempdir().unwrap(); - let cache = open_cache(tmp_dir.path().to_path_buf(), 6); + let cache = open_cache(tmp_dir.path().to_path_buf(), 6).await; cache .put("a".to_string(), OwnedBytes::new(&b"aaa"[..])) .await; @@ -372,7 +394,7 @@ mod tests { #[tokio::test] async fn test_get_after_manual_file_deletion() { let tmp_dir = tempfile::tempdir().unwrap(); - let cache = open_cache(tmp_dir.path().to_path_buf(), 1_000); + let cache = open_cache(tmp_dir.path().to_path_buf(), 1_000).await; cache .put("a".to_string(), OwnedBytes::new(&b"hello"[..])) .await; @@ -385,7 +407,7 @@ mod tests { async fn test_open_evicts_when_over_capacity() { let tmp_dir = tempfile::tempdir().unwrap(); { - let cache = open_cache(tmp_dir.path().to_path_buf(), 1_000); + let cache = open_cache(tmp_dir.path().to_path_buf(), 1_000).await; cache .put("a".to_string(), OwnedBytes::new(&b"aaa"[..])) .await; @@ -394,7 +416,7 @@ mod tests { .await; } // Re-open with a capacity that can only hold one of the two entries. - let cache = open_cache(tmp_dir.path().to_path_buf(), 3); + let cache = open_cache(tmp_dir.path().to_path_buf(), 3).await; let a = cache.get(&"a".to_string()).await; let b = cache.get(&"b".to_string()).await; // Exactly one entry should have survived. @@ -406,7 +428,7 @@ mod tests { let tmp_dir = tempfile::tempdir().unwrap(); let leftover = tmp_dir.path().join("a.tmp42"); std::fs::write(&leftover, b"partial").unwrap(); - let cache = open_cache(tmp_dir.path().to_path_buf(), 1_000); + let cache = open_cache(tmp_dir.path().to_path_buf(), 1_000).await; assert!(!leftover.try_exists().unwrap()); assert!(cache.get(&"a".to_string()).await.is_none()); } diff --git a/quickwit/quickwit-storage/src/cache/tiered_sized_cache.rs b/quickwit/quickwit-storage/src/cache/tiered_sized_cache.rs index 30f994d6fea..d2726156488 100644 --- a/quickwit/quickwit-storage/src/cache/tiered_sized_cache.rs +++ b/quickwit/quickwit-storage/src/cache/tiered_sized_cache.rs @@ -86,6 +86,7 @@ mod tests { 1_000, &CACHE_METRICS_FOR_TESTS, ) + .await .unwrap(); let cache = TieredSizedCache::new(memory_cache(), Some(disk)); cache @@ -105,6 +106,7 @@ mod tests { 1_000, &CACHE_METRICS_FOR_TESTS, ) + .await .unwrap(); let cache = TieredSizedCache::new(memory_cache(), Some(disk)); cache @@ -117,6 +119,7 @@ mod tests { 1_000, &CACHE_METRICS_FOR_TESTS, ) + .await .unwrap(); let memory = memory_cache(); let cache = TieredSizedCache::new(memory, Some(disk)); From 44dffd94fadac3e212dae3ed947995b4ce59c402 Mon Sep 17 00:00:00 2001 From: Darkheir Date: Thu, 9 Jul 2026 14:29:21 +0200 Subject: [PATCH 5/6] feat: Sub-folder sharding Signed-off-by: Darkheir --- .../src/cache/disk_sized_cache.rs | 147 ++++++++++++++---- .../src/cache/tiered_sized_cache.rs | 5 +- 2 files changed, 117 insertions(+), 35 deletions(-) diff --git a/quickwit/quickwit-storage/src/cache/disk_sized_cache.rs b/quickwit/quickwit-storage/src/cache/disk_sized_cache.rs index 94056d23058..bd775e30cde 100644 --- a/quickwit/quickwit-storage/src/cache/disk_sized_cache.rs +++ b/quickwit/quickwit-storage/src/cache/disk_sized_cache.rs @@ -13,15 +13,17 @@ // limitations under the License. use std::fmt::Display; +use std::hash::Hasher; use std::io; use std::marker::PhantomData; use std::path::{Path, PathBuf}; use std::sync::Mutex; use std::sync::atomic::{AtomicU64, Ordering}; -use std::time::SystemTime; +use std::time::{Instant, SystemTime}; +use fnv::FnvHasher; use lru::LruCache; -use tracing::warn; +use tracing::{info, warn}; use crate::OwnedBytes; use crate::metrics::CacheMetrics; @@ -29,6 +31,9 @@ use crate::metrics::CacheMetrics; /// Substring used to mark files that are being written. const TEMP_MARKER: &str = ".tmp"; +/// Number of shard sub-directories files are spread across. +const NUM_SHARDS: u64 = 256; + /// Global counter used to build unique temporary file names. static TEMP_COUNTER: AtomicU64 = AtomicU64::new(0); @@ -119,34 +124,50 @@ impl DiskSizedCache { capacity_in_bytes: u64, cache_counters: &'static CacheMetrics, ) -> io::Result { + let start = Instant::now(); std::fs::create_dir_all(&root_path)?; let mut entries: Vec<(String, u64, SystemTime)> = Vec::new(); - for dir_entry_res in std::fs::read_dir(&root_path)? { - let dir_entry = dir_entry_res?; - if let Ok(file_type) = dir_entry.file_type() - && !file_type.is_file() - { - continue; + for shard_entry_res in std::fs::read_dir(&root_path)? { + let shard_entry = shard_entry_res?; + // Entries live inside shard sub-directories; only recurse into those. + match shard_entry.file_type() { + Ok(file_type) if file_type.is_dir() => {} + _ => continue, } - let file_name = match dir_entry.file_name().into_string() { - Ok(file_name) => file_name, + let shard_dir_iter = match std::fs::read_dir(shard_entry.path()) { + Ok(shard_dir_iter) => shard_dir_iter, Err(_) => continue, }; - if file_name.contains(TEMP_MARKER) { - // Leftover temporary file from an interrupted write: clean it up. - let _ = std::fs::remove_file(dir_entry.path()); - continue; - } - let metadata = match dir_entry.metadata() { - Ok(metadata) => metadata, - Err(_) => continue, - }; - if !metadata.is_file() { - continue; + for dir_entry_res in shard_dir_iter { + let dir_entry = match dir_entry_res { + Ok(dir_entry) => dir_entry, + Err(_) => continue, + }; + if let Ok(file_type) = dir_entry.file_type() + && !file_type.is_file() + { + continue; + } + let file_name = match dir_entry.file_name().into_string() { + Ok(file_name) => file_name, + Err(_) => continue, + }; + if file_name.contains(TEMP_MARKER) { + // Leftover temporary file from an interrupted write: clean it up. + let _ = std::fs::remove_file(dir_entry.path()); + continue; + } + let metadata = match dir_entry.metadata() { + Ok(metadata) => metadata, + Err(_) => continue, + }; + if !metadata.is_file() { + continue; + } + let modified = metadata.modified().unwrap_or(SystemTime::UNIX_EPOCH); + entries.push((file_name, metadata.len(), modified)); } - let modified = metadata.modified().unwrap_or(SystemTime::UNIX_EPOCH); - entries.push((file_name, metadata.len(), modified)); } entries.sort_by_key(|(_, _, modified)| *modified); @@ -168,6 +189,17 @@ impl DiskSizedCache { _phantom: PhantomData, }; remove_files(&cache.root_path, &victims); + let num_entries = { + let index = cache.index.lock().unwrap(); + index.lru_cache.len() + }; + info!( + root_path = %cache.root_path.display(), + num_entries, + num_evicted = victims.len(), + elapsed_millis = start.elapsed().as_millis(), + "opened disk cache" + ); Ok(cache) } @@ -183,7 +215,7 @@ impl DiskSizedCache { } } // Offload the blocking read so we don't stall the async runtime worker. - let path = self.root_path.join(&file_name); + let path = path_for(&self.root_path, &file_name); let read_res = tokio::task::spawn_blocking(move || std::fs::read(path)).await; match read_res { Ok(Ok(buffer)) => { @@ -282,16 +314,30 @@ impl DiskSizedCache { } } +/// Returns the shard sub-directory name a file belongs to. +fn shard_dir(file_name: &str) -> String { + let mut hasher = FnvHasher::default(); + hasher.write(file_name.as_bytes()); + format!("{:02x}", hasher.finish() % NUM_SHARDS) +} + +/// Returns the full on-disk path of an entry, including its shard sub-directory. +pub(crate) fn path_for(root_path: &Path, file_name: &str) -> PathBuf { + root_path.join(shard_dir(file_name)).join(file_name) +} + fn write_file(root_path: &Path, file_name: &str, bytes: &[u8]) -> io::Result<()> { + let shard_path = root_path.join(shard_dir(file_name)); + std::fs::create_dir_all(&shard_path)?; let counter = TEMP_COUNTER.fetch_add(1, Ordering::Relaxed); - let tmp_path = root_path.join(format!("{file_name}{TEMP_MARKER}{counter}")); + let tmp_path = shard_path.join(format!("{file_name}{TEMP_MARKER}{counter}")); std::fs::write(&tmp_path, bytes)?; - std::fs::rename(&tmp_path, root_path.join(file_name)) + std::fs::rename(&tmp_path, shard_path.join(file_name)) } fn remove_files(root_path: &Path, file_names: &[String]) { for file_name in file_names { - let _ = std::fs::remove_file(root_path.join(file_name)); + let _ = std::fs::remove_file(path_for(root_path, file_name)); } } @@ -316,8 +362,8 @@ mod tests { .put("a".to_string(), OwnedBytes::new(&b"hello"[..])) .await; assert_eq!(cache.get(&"a".to_string()).await.unwrap(), &b"hello"[..]); - // A file should have been written on disk. - assert!(tmp_dir.path().join("a").try_exists().unwrap()); + // A file should have been written on disk, inside its shard sub-directory. + assert!(path_for(tmp_dir.path(), "a").try_exists().unwrap()); } #[tokio::test] @@ -359,7 +405,7 @@ mod tests { assert!(cache.get(&"b".to_string()).await.is_none()); assert_eq!(cache.get(&"c".to_string()).await.unwrap(), &b"ccc"[..]); // The evicted entry's file must be gone. - assert!(!tmp_dir.path().join("b").try_exists().unwrap()); + assert!(!path_for(tmp_dir.path(), "b").try_exists().unwrap()); } #[tokio::test] @@ -370,7 +416,7 @@ mod tests { .put("big".to_string(), OwnedBytes::new(&b"toolarge"[..])) .await; assert!(cache.get(&"big".to_string()).await.is_none()); - assert!(!tmp_dir.path().join("big").try_exists().unwrap()); + assert!(!path_for(tmp_dir.path(), "big").try_exists().unwrap()); } #[tokio::test] @@ -398,7 +444,7 @@ mod tests { cache .put("a".to_string(), OwnedBytes::new(&b"hello"[..])) .await; - std::fs::remove_file(tmp_dir.path().join("a")).unwrap(); + std::fs::remove_file(path_for(tmp_dir.path(), "a")).unwrap(); // The stale entry should be detected and reported as a miss. assert!(cache.get(&"a".to_string()).await.is_none()); } @@ -426,10 +472,45 @@ mod tests { #[tokio::test] async fn test_open_cleans_up_temp_files() { let tmp_dir = tempfile::tempdir().unwrap(); - let leftover = tmp_dir.path().join("a.tmp42"); + // A leftover temp file lives inside the shard directory of the entry it belongs to. + let shard_path = tmp_dir.path().join(shard_dir("a")); + std::fs::create_dir_all(&shard_path).unwrap(); + let leftover = shard_path.join("a.tmp42"); std::fs::write(&leftover, b"partial").unwrap(); let cache = open_cache(tmp_dir.path().to_path_buf(), 1_000).await; assert!(!leftover.try_exists().unwrap()); assert!(cache.get(&"a".to_string()).await.is_none()); } + + #[tokio::test] + async fn test_entries_are_sharded_and_survive_reopen() { + let tmp_dir = tempfile::tempdir().unwrap(); + let cache = open_cache(tmp_dir.path().to_path_buf(), 100_000).await; + // Insert enough entries that at least two distinct shard directories get used. + for i in 0..64 { + let key = format!("key-{i}"); + cache.put(key, OwnedBytes::new(&b"payload"[..])).await; + } + // Files must be nested under shard sub-directories, not directly in the root. + let mut shard_dirs = 0; + for entry in std::fs::read_dir(tmp_dir.path()).unwrap() { + let entry = entry.unwrap(); + assert!( + entry.file_type().unwrap().is_dir(), + "root should only contain shard directories" + ); + shard_dirs += 1; + } + assert!( + shard_dirs > 1, + "entries should be spread over several shards" + ); + + // Every entry is still retrievable, including after a reopen (which walks the shards). + let cache = open_cache(tmp_dir.path().to_path_buf(), 100_000).await; + for i in 0..64 { + let key = format!("key-{i}"); + assert_eq!(cache.get(&key).await.unwrap(), &b"payload"[..]); + } + } } diff --git a/quickwit/quickwit-storage/src/cache/tiered_sized_cache.rs b/quickwit/quickwit-storage/src/cache/tiered_sized_cache.rs index d2726156488..9546ee8174f 100644 --- a/quickwit/quickwit-storage/src/cache/tiered_sized_cache.rs +++ b/quickwit/quickwit-storage/src/cache/tiered_sized_cache.rs @@ -62,6 +62,7 @@ impl TieredSizedCache { #[cfg(test)] mod tests { use super::*; + use crate::cache::disk_sized_cache::path_for; use crate::metrics::CACHE_METRICS_FOR_TESTS; fn memory_cache() -> MemorySizedCache { @@ -94,7 +95,7 @@ mod tests { .await; assert_eq!(cache.get(&"a".to_string()).await.unwrap(), &b"hello"[..]); // The payload must have been persisted on disk as well. - assert!(tmp_dir.path().join("a").try_exists().unwrap()); + assert!(path_for(tmp_dir.path(), "a").try_exists().unwrap()); } #[tokio::test] @@ -126,7 +127,7 @@ mod tests { assert_eq!(cache.get(&"a".to_string()).await.unwrap(), &b"hello"[..]); // After a disk hit, deleting the file must not lose the value: it lives in memory now. - std::fs::remove_file(tmp_dir.path().join("a")).unwrap(); + std::fs::remove_file(path_for(tmp_dir.path(), "a")).unwrap(); assert_eq!(cache.get(&"a".to_string()).await.unwrap(), &b"hello"[..]); } } From 0b228a702f7f08d4ea679fab91a0fb77f26e6807 Mon Sep 17 00:00:00 2001 From: Darkheir Date: Thu, 9 Jul 2026 17:13:05 +0200 Subject: [PATCH 6/6] feat: Apply PR review suggestion Signed-off-by: Darkheir --- .../src/cache/disk_sized_cache.rs | 30 +++++++++---------- 1 file changed, 15 insertions(+), 15 deletions(-) diff --git a/quickwit/quickwit-storage/src/cache/disk_sized_cache.rs b/quickwit/quickwit-storage/src/cache/disk_sized_cache.rs index bd775e30cde..c1af2186081 100644 --- a/quickwit/quickwit-storage/src/cache/disk_sized_cache.rs +++ b/quickwit/quickwit-storage/src/cache/disk_sized_cache.rs @@ -31,9 +31,6 @@ use crate::metrics::CacheMetrics; /// Substring used to mark files that are being written. const TEMP_MARKER: &str = ".tmp"; -/// Number of shard sub-directories files are spread across. -const NUM_SHARDS: u64 = 256; - /// Global counter used to build unique temporary file names. static TEMP_COUNTER: AtomicU64 = AtomicU64::new(0); @@ -135,32 +132,31 @@ impl DiskSizedCache { Ok(file_type) if file_type.is_dir() => {} _ => continue, } - let shard_dir_iter = match std::fs::read_dir(shard_entry.path()) { - Ok(shard_dir_iter) => shard_dir_iter, - Err(_) => continue, + let Ok(shard_dir_iter) = std::fs::read_dir(shard_entry.path()) else { + continue; }; + for dir_entry_res in shard_dir_iter { - let dir_entry = match dir_entry_res { - Ok(dir_entry) => dir_entry, - Err(_) => continue, + let Ok(dir_entry) = dir_entry_res else { + continue; }; + if let Ok(file_type) = dir_entry.file_type() && !file_type.is_file() { continue; } - let file_name = match dir_entry.file_name().into_string() { - Ok(file_name) => file_name, - Err(_) => continue, + let Ok(file_name) = dir_entry.file_name().into_string() else { + continue; }; + if file_name.contains(TEMP_MARKER) { // Leftover temporary file from an interrupted write: clean it up. let _ = std::fs::remove_file(dir_entry.path()); continue; } - let metadata = match dir_entry.metadata() { - Ok(metadata) => metadata, - Err(_) => continue, + let Ok(metadata) = dir_entry.metadata() else { + continue; }; if !metadata.is_file() { continue; @@ -316,6 +312,9 @@ impl DiskSizedCache { /// Returns the shard sub-directory name a file belongs to. fn shard_dir(file_name: &str) -> String { + // Number of shard sub-directories files are spread across. + const NUM_SHARDS: u64 = 256; + let mut hasher = FnvHasher::default(); hasher.write(file_name.as_bytes()); format!("{:02x}", hasher.finish() % NUM_SHARDS) @@ -329,6 +328,7 @@ pub(crate) fn path_for(root_path: &Path, file_name: &str) -> PathBuf { fn write_file(root_path: &Path, file_name: &str, bytes: &[u8]) -> io::Result<()> { let shard_path = root_path.join(shard_dir(file_name)); std::fs::create_dir_all(&shard_path)?; + // Rely on a counter to guarantee uniqueness of the temporary file name. let counter = TEMP_COUNTER.fetch_add(1, Ordering::Relaxed); let tmp_path = shard_path.join(format!("{file_name}{TEMP_MARKER}{counter}")); std::fs::write(&tmp_path, bytes)?;