fix(storage): skip caching splits larger than the split cache budget

quickwit/quickwit-indexing/src/actors/uploader.rs

@@ -338,6 +338,10 @@ impl Handler<PackagedSplitBatch> for Uploader {
                     report_splits.push(ReportSplit {
                         storage_uri: split_store.remote_uri().to_string(),
                         split_id: packaged_split.split_id_str().to_string(),
+                        // The footer sits at the end of the split file, so its end
+                        // offset equals the split file size. The searcher split
+                        // cache uses this to skip caching splits larger than its budget.
+                        num_bytes: split_streamer.footer_range.end,
                     });
 
                     split_metadata_list.push(split_metadata);
@@ -1058,6 +1062,9 @@ mod tests {
         let split = &report_splits.report_splits[0];
         assert_eq!(split.storage_uri, "ram:///");
         assert_eq!(split.split_id, SPLIT_ULID_STR);
+        // The reported size is the split file size (footer end offset), which the
+        // searcher split cache uses to skip caching oversized splits.
+        assert!(split.num_bytes > 0);
         universe.assert_quit().await;
         Ok(())
     }

quickwit/quickwit-proto/protos/quickwit/search.proto

@@ -105,6 +105,10 @@ message ReportSplit {
   string split_id = 2;
   // The storage uri. This URI does NOT include the split id.
   string storage_uri = 1;
+  // Size of the split file in bytes (i.e. `footer_offsets.end`).
+  // A value of 0 means the size is unknown to the reporter; the split
+  // cache treats an unknown size as "fits" so behavior is unchanged.
+  uint64 num_bytes = 3;
 }
 
 message ReportSplitsRequest {

quickwit/quickwit-search/src/leaf.rs

@@ -179,6 +179,15 @@ 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() {
+            // Register the split together with its file size (the footer end
+            // offset) so the download guard can skip it when it is larger than
+            // the cache budget. This is the only size signal for pre-existing
+            // splits that no indexer ever reported a size for.
+            split_cache.report_split_size(
+                split_and_footer_offsets.split_id.clone().into(),
+                index_storage.uri(),
+                split_and_footer_offsets.split_footer_end,
+            );
             SearchSplitCache::wrap_storage(split_cache.clone(), index_storage.clone())
         } else {
             index_storage.clone()

quickwit/quickwit-storage/src/metrics.rs

@@ -335,6 +335,19 @@ pub static PREDICATE_CACHE: LazyLock<CacheMetrics> =
 pub(crate) static SEARCHER_SPLIT_CACHE: LazyLock<CacheMetrics> =
     LazyLock::new(|| CacheMetrics::for_component("searcher_split"));
 
+/// Number of times a split download opportunity was skipped because the split
+/// is larger than the entire split-cache byte budget (`split_cache.max_num_bytes`).
+///
+/// Such a split can never coexist with any other split and would evict the whole
+/// cache without ever fitting, so it is left to be served by the cold-storage
+/// warmup path. A steadily rising counter means a searcher keeps re-evaluating a
+/// hot oversized split; it is a "skip events" rate, not a distinct-split count.
+pub(crate) static SEARCHER_SPLIT_CACHE_DOWNLOADS_SKIPPED_TOO_LARGE: LazyCounter = lazy_counter!(
+    name: "searcher_split_cache_downloads_skipped_too_large_total",
+    description: "Number of split downloads skipped because the split is larger than the split cache byte budget.",
+    subsystem: "storage",
+);
+
 /// Cache metrics for short-lived byte range caches (used during leaf search
 /// and caching directory warmup).
 pub static SHORTLIVED_CACHE: LazyLock<CacheMetrics> =

quickwit/quickwit-storage/src/split_cache/download_task.rs

@@ -32,6 +32,7 @@ async fn download_split(
         split_id,
         storage_uri,
         living_token: _,
+        num_bytes: _,
     } = candidate_split;
     let split_filename = split_file(split_id);
     let target_filepath = root_path.join(&split_filename);

quickwit/quickwit-storage/src/split_cache/mod.rs

@@ -94,7 +94,8 @@ impl SearchSplitCache {
         let mut split_table = SplitTable::with_limits_and_existing_splits(limits, existing_splits);
 
         // In case of a setting change, it could be useful to evict some splits on startup.
-        let splits_to_remove_res = split_table.make_room_for_split_if_necessary(u64::MAX);
+        // No specific split is incoming here; `0` just trims the table to the current limits.
+        let splits_to_remove_res = split_table.make_room_for_split_if_necessary(u64::MAX, 0);
         if let Ok(splits_to_remove) = splits_to_remove_res {
             info!(
                 num_splits = splits_to_remove.len(),
@@ -144,10 +145,27 @@ impl SearchSplitCache {
                 error!(storage_uri=%report_split.storage_uri, "received invalid storage uri: ignoring");
                 continue;
             };
-            split_table.report(split_id, storage_uri);
+            split_table.report(split_id, storage_uri, report_split.num_bytes);
         }
     }
 
+    /// Informs the cache that a split is about to be searched, together with its
+    /// file size in bytes (the split footer end offset).
+    ///
+    /// This is the only size signal for splits that no indexer reported a size
+    /// for — most importantly pre-existing oversized splits. Those are otherwise
+    /// discovered via `get_split_file`/`touch` without a size and would bypass
+    /// the download guard entirely. Attaching the size here lets the guard skip
+    /// them just like freshly-reported oversized splits. A zero size (unknown)
+    /// is ignored, so behavior is unchanged when the size is genuinely unknown.
+    pub fn report_split_size(&self, split_id: SplitId, storage_uri: &Uri, num_bytes: u64) {
+        if num_bytes == 0 {
+            return;
+        }
+        let mut split_table = self.split_table.lock().unwrap();
+        split_table.report(split_id, storage_uri.clone(), num_bytes);
+    }
+
     // Returns a split guard object. As long as it is not dropped, the
     // split won't be evinced from the cache.
     async fn get_split_file(&self, split_id: SplitId, storage_uri: &Uri) -> Option<SplitFile> {