feat(storage): cache immutable split byte ranges

quickwit/quickwit-lambda-server/src/context.rs

@@ -33,8 +33,11 @@ impl LambdaSearcherContext {
         info!("initializing lambda searcher context");
 
         let searcher_config = try_searcher_config_from_env()?;
-        let searcher_context =
-            Arc::new(SearcherContext::new_without_invoker(searcher_config, None));
+        let searcher_context = Arc::new(SearcherContext::new_without_invoker(
+            searcher_config,
+            None,
+            None,
+        ));
         let storage_resolver = StorageResolver::configured(&Default::default());
 
         Ok(Self {

quickwit/quickwit-search/Cargo.toml

@@ -53,6 +53,7 @@ assert-json-diff = { workspace = true }
 proptest = { workspace = true }
 rand = { workspace = true }
 serde_json = { workspace = true }
+tempfile = { workspace = true }
 
 quickwit-indexing = { workspace = true, features = ["testsuite"] }
 quickwit-metastore = { workspace = true, features = ["testsuite"] }

quickwit/quickwit-search/src/leaf.rs

@@ -46,6 +46,7 @@ use quickwit_query::tokenizers::TokenizerManager;
 use quickwit_storage::{
     BundleStorage, ByteRangeCache, CountingStorage, MemorySizedCache, OwnedBytes, SearchSplitCache,
     Storage, StorageResolver, TimeoutAndRetryStorage, wrap_storage_with_cache,
+    wrap_storage_with_split_range_cache,
 };
 use tantivy::aggregation::AggContextParams;
 use tantivy::aggregation::agg_req::{AggregationVariants, Aggregations};
@@ -160,16 +161,19 @@ async fn get_split_footer_from_cache_or_fetch(
     Ok(footer_data_opt)
 }
 
-/// Returns hotcache_bytes and the split directory (`BundleStorage`) with cache layer:
-/// - A split footer cache given by `SearcherContext.split_footer_cache`.
+/// Returns hotcache_bytes and the split directory (`BundleStorage`).
 pub(crate) async fn open_split_bundle(
     searcher_context: &SearcherContext,
     index_storage: Arc<dyn Storage>,
     split_and_footer_offsets: &SplitIdAndFooterOffsets,
 ) -> anyhow::Result<(FileSlice, BundleStorage)> {
     let split_file = PathBuf::from(format!("{}.split", split_and_footer_offsets.split_id));
+    let foyer_storage: Arc<dyn Storage> = match &searcher_context.split_range_disk_cache_opt {
+        Some(cache) => wrap_storage_with_split_range_cache(cache.clone(), index_storage.clone()),
+        None => index_storage.clone(),
+    };
     let footer_data = get_split_footer_from_cache_or_fetch(
-        index_storage.clone(),
+        foyer_storage.clone(),
         split_and_footer_offsets,
         &searcher_context.split_footer_cache,
     )
@@ -179,9 +183,9 @@ pub(crate) async fn open_split_bundle(
     // This is before the bundle storage: at this point, this storage is reading `.split` files.
     let index_storage_with_split_cache =
         if let Some(split_cache) = searcher_context.split_cache_opt.as_ref() {
-            SearchSplitCache::wrap_storage(split_cache.clone(), index_storage.clone())
+            SearchSplitCache::wrap_storage(split_cache.clone(), foyer_storage)
         } else {
-            index_storage.clone()
+            foyer_storage
         };
 
     let (hotcache_bytes, bundle_storage) = BundleStorage::open_from_split_data(
@@ -3113,7 +3117,8 @@ mod tests {
             offload_threshold: 3,
             ..LambdaConfig::for_test()
         });
-        let searcher_context = SearcherContext::new(config, None, Some(Arc::new(DummyInvoker)));
+        let searcher_context =
+            SearcherContext::new(config, None, None, Some(Arc::new(DummyInvoker)));
         let splits = make_splits_with_requests(7);
         let result = super::schedule_search_tasks(splits, &searcher_context).await;
         assert_eq!(result.local_search_tasks.len(), 3);
@@ -3133,7 +3138,8 @@ mod tests {
             offload_threshold: 0,
             ..LambdaConfig::for_test()
         });
-        let searcher_context = SearcherContext::new(config, None, Some(Arc::new(DummyInvoker)));
+        let searcher_context =
+            SearcherContext::new(config, None, None, Some(Arc::new(DummyInvoker)));
         let splits = make_splits_with_requests(5);
         let result = super::schedule_search_tasks(splits, &searcher_context).await;
         assert!(result.local_search_tasks.is_empty());
@@ -3147,7 +3153,8 @@ mod tests {
             offload_threshold: 100,
             ..LambdaConfig::for_test()
         });
-        let searcher_context = SearcherContext::new(config, None, Some(Arc::new(DummyInvoker)));
+        let searcher_context =
+            SearcherContext::new(config, None, None, Some(Arc::new(DummyInvoker)));
         let splits = make_splits_with_requests(5);
         let result = super::schedule_search_tasks(splits, &searcher_context).await;
         assert_eq!(result.local_search_tasks.len(), 5);

quickwit/quickwit-search/src/lib.rs

@@ -42,6 +42,8 @@ pub(crate) mod top_k_collector;
 mod metrics;
 mod search_permit_provider;
 
+#[cfg(test)]
+mod split_range_cache_layer_tests;
 #[cfg(test)]
 mod tests;
 
@@ -289,7 +291,11 @@ 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_without_invoker(searcher_config, None));
+    let searcher_context = Arc::new(SearcherContext::new_without_invoker(
+        searcher_config,
+        None,
+        None,
+    ));
     let search_service = Arc::new(SearchServiceImpl::new(
         metastore.clone(),
         storage_resolver,

quickwit/quickwit-search/src/service.rs

@@ -29,7 +29,8 @@ use quickwit_proto::search::{
     SearchPlanResponse, SearchRequest, SearchResponse, SnippetRequest,
 };
 use quickwit_storage::{
-    MemorySizedCache, QuickwitCache, SearchSplitCache, StorageCache, StorageResolver,
+    FoyerSplitRangeCache, MemorySizedCache, QuickwitCache, SearchSplitCache, StorageCache,
+    StorageResolver,
 };
 use tantivy::aggregation::AggregationLimitsGuard;
 
@@ -417,6 +418,8 @@ pub struct SearcherContext {
     pub predicate_cache: Arc<PredicateCacheImpl>,
     /// Search split cache. `None` if no split cache is configured.
     pub split_cache_opt: Option<Arc<SearchSplitCache>>,
+    /// Process-wide split range disk cache. `None` if not configured.
+    pub split_range_disk_cache_opt: Option<Arc<FoyerSplitRangeCache>>,
     /// List fields cache. Caches the raw fields-metadata blob for a given split.
     pub list_fields_cache: ListFieldsCache,
     /// The aggregation limits are passed to limit the memory usage.
@@ -439,17 +442,19 @@ impl SearcherContext {
     #[cfg(test)]
     pub fn for_test() -> SearcherContext {
         let searcher_config = SearcherConfig::default();
-        SearcherContext::new_without_invoker(searcher_config, None)
+        SearcherContext::new_without_invoker(searcher_config, None, None)
     }
 
     /// Creates a new searcher context without a lambda invoker.
     pub fn new_without_invoker(
         searcher_config: SearcherConfig,
         split_cache_opt: Option<Arc<SearchSplitCache>>,
+        split_range_disk_cache_opt: Option<Arc<FoyerSplitRangeCache>>,
     ) -> Self {
         Self::new(
             searcher_config,
             split_cache_opt,
+            split_range_disk_cache_opt,
             None::<Box<dyn LambdaLeafSearchInvoker>>,
         )
     }
@@ -458,6 +463,7 @@ impl SearcherContext {
     pub fn new(
         searcher_config: SearcherConfig,
         split_cache_opt: Option<Arc<SearchSplitCache>>,
+        split_range_disk_cache_opt: Option<Arc<FoyerSplitRangeCache>>,
         lambda_invoker: Option<impl LambdaLeafSearchInvoker + 'static>,
     ) -> Self {
         let global_split_footer_cache = MemorySizedCache::from_config(
@@ -490,6 +496,7 @@ impl SearcherContext {
             leaf_search_cache,
             list_fields_cache,
             split_cache_opt,
+            split_range_disk_cache_opt,
             aggregation_limit,
             lambda_invoker,
         }