From 02ba2a611b04f5e9b24c26abe01288ee200b9422 Mon Sep 17 00:00:00 2001 From: Pascal Seitz Date: Tue, 30 Jun 2026 16:51:51 +0200 Subject: [PATCH 1/2] add analysis.md --- analysis.md | 343 ++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 343 insertions(+) create mode 100644 analysis.md diff --git a/analysis.md b/analysis.md new file mode 100644 index 00000000000..64f2591c987 --- /dev/null +++ b/analysis.md @@ -0,0 +1,343 @@ +# Analysis: A Download Manager in `quickwit-storage` + +## Goal + +Today, fetching split data from object storage is spread across two crates and a +deep stack of `async`-wrapping decorators ("async onions"). Caching is bolted on +by **wrapping the tantivy `Directory`**, and every other concern (debouncing, +retry/timeout, bundle-offset translation, the on-disk split cache, the +fast-field cache) is a separate `#[async_trait] impl Storage` decorator. A single +`get_slice` therefore traverses 5–9 boxed futures even when it is a pure cache +hit that performs no I/O. + +This document analyzes replacing that with a **`DownloadManager` in +`quickwit-storage`**: a *synchronous core* (cache lookups, bundle-offset math, +range merging, request dedup keying, bookkeeping) plus **one thin, genuinely +async layer** that performs the network/disk download. It is the result of three +agreed design decisions: + +| Decision | Choice | +|----------|--------| +| **Scope** | Search read path **+** the on-disk split-cache downloads (`get_slice`/`get_all`/`copy_to_file`). Indexing/upload path is out of scope for now. | +| **Approach** | **Collapse the onion** into a single manager: replace the decorator chain with a sync core + one async download fn. | +| **Caching** | **Unify** all read caching in the storage download manager, keyed by physical split path + absolute range. The directory wrappers become thin views. | + +--- + +## 1. Current architecture + +### 1.1 The decorator stack (storage crate) + +`quickwit-storage` composes behavior with the `Storage` trait +(`quickwit-storage/src/storage.rs`), every method `async`. Each decorator wraps +`Arc` and is itself `#[async_trait]`, so each call site allocates a +`Pin>`: + +| Layer | File | What it adds | Sync or async work? | +|-------|------|--------------|---------------------| +| `TimeoutAndRetryStorage` | `src/timeout_and_retry_storage.rs` | aggressive per-attempt timeout + retry on `get_slice` | sync (loop/policy) + async (the awaited attempt) | +| `StorageWithCache` (split cache) | `src/cache/storage_with_cache.rs` + `src/split_cache/mod.rs` | on-disk full-split LRU cache (`SplitCacheBackingStorage`) | sync lookup wrapped in `async fn get`; async only on disk read | +| `StorageWithCache` (fast fields) | `src/cache/storage_with_cache.rs` | in-memory `MemorySizedCache` for `.fast` data via `QuickwitCache` | **fully sync lookup**, needlessly `.await`ed | +| `DebouncedStorage` | `src/debouncer.rs` | coalesce concurrent identical `(path, range)` requests | sync hashing/lookup; async only awaiting the shared future | +| `S3CompatibleObjectStorage` | `src/object_storage/s3_compatible_storage.rs` | the real HTTP GET (`REQUEST_SEMAPHORE`, range header, retry, `download_all`) | **genuinely async** | + +Assembly order (search node), bottom → top: +``` +S3CompatibleObjectStorage (s3_compatible_storage_resolver.rs:64) + └─ DebouncedStorage + └─ SplitCache.wrap_storage → StorageWithCache(SplitCacheBackingStorage) (split_cache/mod.rs:129) + └─ (per split) StorageWithCache(fast_fields QuickwitCache) (leaf.rs:233) + └─ BundleStorage (leaf.rs:226) + └─ TimeoutAndRetryStorage (optional) (leaf.rs:200) +``` + +### 1.2 The directory stack (directories crate) + +On top of `Storage`, search builds a tantivy `Directory` tower in +`open_index_with_caches` (`quickwit-search/src/leaf.rs:216`): +``` +S3 …storage… → BundleStorage + └─ StorageDirectory (storage_directory.rs — sync read_bytes() PANICS, async only) + └─ CachingDirectory (caching_directory.rs — ByteRangeCache, "ephemeral unbounded") + └─ HotDirectory (hot_directory.rs — static slice cache from the split footer) +``` +- `BundleStorage` (`src/bundle_storage.rs`) translates a logical file path inside + the `.split` to an absolute byte range in the bundle — **pure arithmetic + + metadata-map lookup**, but still an `async fn get_slice`. +- `CachingDirectory` holds the per-split `ByteRangeCache` + (`caching_directory.rs:31`). Its `get_slice` lookup is **synchronous** — + `ByteRangeCache::get_slice` returns `Option` with no `.await`. +- `HotDirectory` serves a static, pre-computed slice cache (the hotcache built at + index time). Lookups are **synchronous**. + +### 1.3 The core problem: the async download path is an N-layer onion + +The waste we are removing is on the **download path** (`get_slice`/`get_all`), +which `warmup` drives at high volume. A single fetch threads through every +decorator in §1.1/§1.2, and **each layer is `#[async_trait]`**, so each one +allocates a `Box::pin`'d future and adds an `.await` — *regardless of whether that +layer does any I/O*: + +``` +TimeoutAndRetry → Bundle → StorageWithCache(fast) → StorageWithCache(splitcache) → Debounced → S3 +``` + +Of those six layers, exactly one (S3, plus the split-cache's disk read) actually +performs I/O. The other five do **synchronous work** — cache lookups, bundle +offset math, dedup keying — yet each still pays the full async cost, because an +`async fn` whose body is synchronous is **not** free: + +1. **A future (state machine) is constructed** for the call. +2. With `#[async_trait]`, that future is **`Box::pin`'d → a heap allocation per + call** (`async-trait` desugars to `Pin>`). +3. The caller **`.await`s it → a poll**, and the `.await` is an optimization + barrier (no inlining across it). + +A pure-sync body only avoids the task suspend/wake (it returns `Poll::Ready` on +the first poll, no scheduler round-trip). The **boxed allocation + poll + await +barrier remain — per layer, per call**. Multiplied by layer count × warmup call +volume, that is the cost that "adds up." We are paying async machinery for +**synchronous decisions**; the network wait itself is unavoidable and not the +target. + +On top of that, caching is split across **five** places (split-footer cache, +fast-field `QuickwitCache`, on-disk `SplitCache`, ephemeral `ByteRangeCache`, +static `HotDirectory` cache), each keyed differently (`split_id` string, logical +path, physical path, range), at two different abstraction levels — so there is no +single place that owns "fetch a split byte range." + +### 1.4 Aside: the sync read path falls out for free + +A secondary (not load-bearing) observation: tantivy's `FileHandle` already exposes +both `read_bytes` (sync) and `read_bytes_async`. Warmup populates the caches via +the async path; the subsequent collection/scoring phase reads via **sync** +`read_bytes`, which today must hit `CachingDirectory`'s `ByteRangeCache` because +`StorageDirectory::read_bytes` panics by design. Since the proposed `resolve()` is +already a plain synchronous `fn`, that sync handle can call it directly — a nice +side benefit, but the rationale for this refactor stands entirely on §1.3. + +--- + +## 2. Proposed design + +### 2.1 Principle + +> **Decide synchronously, download asynchronously.** + +Split the work into: + +- **Sync core** — everything that is CPU/memory-only: cache lookups across all + tiers, bundle logical→physical offset translation, range coalescing, dedup + keying, metrics, eviction bookkeeping. No futures, no allocation of boxed + async state. Returns a *resolution*. +- **Thin async layer** — exactly one future type that performs the actual + download (object-store HTTP GET, or on-disk split-cache file read), bounded by + the existing semaphore, with retry/timeout, and that writes results back into + the caches synchronously on completion. + +### 2.2 Shape + +```rust +// quickwit-storage/src/download_manager/mod.rs +pub struct DownloadManager { + caches: CacheStack, // footer + fast-field + byte-range, unified, sync API + split_cache: Option>, // on-disk full-split cache (sync lookup, async fill) + inflight: AsyncDebouncer>, // dedup + retry: RetryParams, + timeout_policy: Option, + backend: Arc, // the RAW object-store backend, no decorators (see 2.4) +} + +/// Outcome of the synchronous resolution step. No I/O has happened. +pub enum Resolution { + /// Served entirely from a cache tier — zero futures, return immediately. + Hit(OwnedBytes), + /// Needs a download; carries the physical key + range to fetch. + Miss(DownloadPlan), +} + +impl DownloadManager { + /// 100% synchronous. Safe to call from tantivy's sync `read_bytes`. + pub fn resolve(&self, split: &SplitId, range: LogicalRange) -> Resolution { ... } + + /// The thin async layer. Only entered on a Miss. Debounced + retried + + /// semaphore-bounded; writes the result into the caches before returning. + pub async fn download(&self, plan: DownloadPlan) -> StorageResult { ... } + + /// Convenience used by warmup: resolve, and await only if it's a miss. + pub async fn get(&self, split: &SplitId, range: LogicalRange) -> StorageResult { + match self.resolve(split, range) { + Resolution::Hit(bytes) => Ok(bytes), + Resolution::Miss(plan) => self.download(plan).await, + } + } +} +``` + +`resolve()` walks the cache tiers in order (byte-range / fast-field → on-disk +split cache index → footer), applies the bundle-offset translation, and merges +the request against already-cached touching ranges — all synchronously. Only a +genuine miss produces a `DownloadPlan`, and only then is a single future created. + +### 2.3 Unified cache stack + +Replace the five scattered caches and the two caching directories with one +`CacheStack` owned by the manager, keyed uniformly by **physical split path + +absolute byte range**: + +- byte-range cache (replaces `CachingDirectory`'s `ByteRangeCache` + the + fast-field `MemorySizedCache`/`QuickwitCache`), +- static hotcache slices (replaces `HotDirectory`'s static cache — loaded once + per split, still keyed by physical range), +- split-footer cache, +- on-disk `SplitCache` (kept; its lookup becomes a sync index check, its fill the + async path). + +The tantivy directories (`StorageDirectory`/`CachingDirectory`/`HotDirectory`) +collapse into **one thin `DownloadManagerDirectory`** whose `FileHandle`: +- `read_bytes` (sync) → `manager.resolve(...)`, returning `Hit` bytes or erroring + on a miss (preserving today's "everything must be warmed" invariant); +- `read_bytes_async` → `manager.get(...)` (resolve + thin download). + +### 2.4 We keep the `Storage` trait + +The `Storage` trait stays as-is. It is the backend abstraction used across the +whole codebase (indexing, upload, metastore, CLI, and every backend: S3, Azure, +GCS, local, RAM), and it already exposes exactly the genuinely-async I/O the +manager needs: `get_slice`, `get_all`, `copy_to_file`. There is **no new download +trait** — that would be a redundant abstraction over what `Storage` already does. + +What the manager holds is the **raw backend** `Arc` — i.e. +`S3CompatibleObjectStorage` (or the local/Azure/GCS impl) with **none of the +read-path decorators** wrapped around it. On a `Miss`, `download()` calls the +backend's existing `storage.get_slice(...)` / `get_all(...)` / `copy_to_file(...)`. + +So the refactor removes the *decorator stack on top of `Storage`* +(`DebouncedStorage`, `StorageWithCache`, `TimeoutAndRetryStorage`) and folds their +behavior into the manager's sync core + thin `download()` — it does **not** touch +the `Storage` trait itself, nor the write path (`put`/`delete`/indexing), which is +unaffected by this read-only change. + +### 2.5 Where debounce/retry/timeout go + +These collapse from standalone decorators into the manager's `download()`: +- **debounce**: keyed in `download()` on the physical `DownloadKey`, so dedup + happens once, at the only place a download is issued (today's `DebouncedStorage` + logic in `debouncer.rs` is reused as-is — it is already a sync map + shared + future). +- **retry + timeout**: a loop inside `download()` around the single + `Storage::get_slice` await on the raw backend (reuses `TimeoutAndRetryStorage`'s + policy iteration and `aws_retry`). + +--- + +## 3. Before / after + +The headline is the **download path** (`warmup`), where the onion multiplies +futures across layers that mostly do synchronous work. + +**Before — one `get_slice` on a cache miss, a boxed future per layer:** +``` +TimeoutAndRetry.get_slice().await ← Box::pin + Bundle.get_slice().await ← Box::pin (offset math — sync work) + StorageWithCache(fast).get_slice() ← Box::pin (sync lookup, awaited) + StorageWithCache(splitcache) ← Box::pin (sync keying, awaited) + Debounced.get_slice().await ← Box::pin (sync keying, awaited) + S3.get_slice().await ← Box::pin (real HTTP) ← only this does I/O +``` +6 boxed futures, 5 of them for synchronous work — *even though the call hits the +network exactly once*. + +**After — `resolve()` does all the sync work future-free; one future does the I/O:** +``` +manager.get(split, range) + → resolve(...) ← sync: caches + offset math + dedup keying, 0 futures + → download(plan).await ← 1 future: debounce + retry + raw-backend get_slice +``` +**Miss: 6 boxed futures → 1.** And a fetch whose range is already cached (in the +long-term fast-field cache, footer cache, or on-disk split cache) resolves as a +`Hit` with **0 futures — even though warmup calls it from an async context**, +because `resolve()` is synchronous and `get()` returns `Ready` without building the +download future. + +*(Aside, per §1.4: the post-warmup sync collection path calls `resolve()` directly +from `read_bytes`, so it too is future-free — but the win above stands without +relying on that.)* + +--- + +## 4. Migration plan (incremental, each step compiles & ships) + +1. **Introduce `ObjectDownload`** and implement it for the existing backends as a + thin pass-through next to today's `Storage` impl. No behavior change. +2. **Build `CacheStack`** — move `ByteRangeCache`, fast-field cache, footer cache + behind one sync-lookup API keyed by physical path+range. Port the static + hotcache slices into it. Cover with the existing cache unit tests. +3. **Build `DownloadManager`** with `resolve()` + `download()`, folding in + `DebouncedStorage` and `TimeoutAndRetryStorage` logic and the on-disk + `SplitCache` (sync index lookup, async fill). +4. **Add `DownloadManagerDirectory`** (sync `read_bytes` via `resolve`, async via + `get`); switch `open_index_with_caches` to build it instead of the + `Storage→StorageDirectory→Caching→Hot` tower. +5. **Repoint `warmup`** to call `manager.get(...)`; verify the + warmup-then-sync-read invariant still holds (search reads are all hits). +6. **Delete** `StorageWithCache`, `CachingDirectory`, `HotDirectory`, + `StorageDirectory`, the per-layer `Storage` decorators for reads, and the + resolver wiring that stacked them. +7. **Validate** through the production path: ingest via OTLP, query via REST, + confirm hit-rate metrics and end-to-end latency. Run the existing + `quickwit-search` + `quickwit-storage` test suites and the integration tests. + +Steps 1–4 are additive and can land behind the old path; step 6 is the deletion +once parity is confirmed. + +--- + +## 5. Risks & things to watch + +- **Tantivy fork coupling.** The sync-read-after-warmup contract depends on the + custom tantivy fork's `FileHandle`. `resolve()` returning a miss inside sync + `read_bytes` must error exactly as `StorageDirectory::read_bytes` panics today, + so an un-warmed read is still a loud bug, not a silent sync stall. +- **Debounce + delete race.** The existing race noted in `debouncer.rs` (get/ + delete/get returning stale cached bytes) must be preserved-or-fixed + deliberately, not lost in the move. +- **Cache key unification.** Today keys differ (footer by `split_id` string, + fast-field by logical `.fast` suffix routing, byte-range by physical path). + Collapsing to physical-path+range must keep the `.fast` long-term vs. ephemeral + short-lived distinction (different sizes/lifetimes/metrics) — the `CacheStack` + needs per-tier capacity & metrics, not one flat map. +- **`get_slice_stream`** currently bypasses the debouncer and cache. Decide + whether the manager exposes a streaming download or it stays on the raw + backend. +- **Metrics continuity.** Preserve the existing per-cache `CacheMetrics` + (`fast_field_cache`, `split_footer_cache`, `searcher_split_cache`, + `shortlived_cache`, …) so dashboards keep working. + +--- + +## 6. Expected payoff + +- Collapse the download onion: **6 boxed futures per `get_slice` → 1 on a miss**, + and **0 on an already-cached range** — on the async warmup path that drives the + bulk of the calls. +- One place that owns "fetch a split byte range," replacing two crates' worth of + layering — easier to reason about, profile, and tune. +- Caching keyed consistently on physical bytes, no longer entangled with the + tantivy `Directory` abstraction. + +--- + +## 7. Open questions (non-blocking, for review) + +1. **Streaming reads** — should `DownloadManager` own a streaming download + (`get_slice_stream` equivalent), or is range+full-file enough for search + + split-cache fill? (Today streaming bypasses cache/debounce.) +2. **Crate boundary** — the unified directory (`DownloadManagerDirectory`) needs + tantivy types, so it likely lives in `quickwit-directories` while the manager + lives in `quickwit-storage`. Confirm that split, or whether the thin directory + should move into `quickwit-storage` to keep the manager self-contained. +3. **`copy_to_file` for split-cache** — the background full-split download task + should route through `DownloadManager::download()` (so dedup/retry/semaphore are + shared), calling the backend's existing `Storage::copy_to_file`. Confirm we want + that rather than letting the task call the raw backend directly. From a43a3788d4dd1757f3e7df9ef0a5bb0b2e9b43f2 Mon Sep 17 00:00:00 2001 From: Pascal Seitz Date: Wed, 1 Jul 2026 22:24:49 +0200 Subject: [PATCH 2/2] add DownloadManager --- .../src/download_manager_directory.rs | 174 +++++++++ .../quickwit-directories/src/hot_directory.rs | 2 +- quickwit/quickwit-directories/src/lib.rs | 8 +- .../src/storage_directory.rs | 142 -------- quickwit/quickwit-search/src/leaf.rs | 140 ++++---- quickwit/quickwit-search/src/service.rs | 15 +- .../quickwit-storage/src/bundle_storage.rs | 2 +- quickwit/quickwit-storage/src/cache/mod.rs | 2 - .../src/cache/quickwit_cache.rs | 207 ----------- quickwit/quickwit-storage/src/debouncer.rs | 118 +----- .../src/download_manager/download.rs | 325 +++++++++++++++++ .../src/download_manager/mod.rs | 335 ++++++++++++++++++ quickwit/quickwit-storage/src/lib.rs | 11 +- .../src/local_file_storage.rs | 6 +- .../src/object_storage/azure_blob_storage.rs | 3 +- .../s3_compatible_storage_resolver.rs | 8 +- .../opendal_storage/google_cloud_storage.rs | 3 +- .../quickwit-storage/src/split_cache/mod.rs | 8 +- .../src/timeout_and_retry_storage.rs | 279 --------------- 19 files changed, 953 insertions(+), 835 deletions(-) create mode 100644 quickwit/quickwit-directories/src/download_manager_directory.rs delete mode 100644 quickwit/quickwit-directories/src/storage_directory.rs delete mode 100644 quickwit/quickwit-storage/src/cache/quickwit_cache.rs create mode 100644 quickwit/quickwit-storage/src/download_manager/download.rs create mode 100644 quickwit/quickwit-storage/src/download_manager/mod.rs delete mode 100644 quickwit/quickwit-storage/src/timeout_and_retry_storage.rs diff --git a/quickwit/quickwit-directories/src/download_manager_directory.rs b/quickwit/quickwit-directories/src/download_manager_directory.rs new file mode 100644 index 00000000000..9a39ad7c254 --- /dev/null +++ b/quickwit/quickwit-directories/src/download_manager_directory.rs @@ -0,0 +1,174 @@ +// 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::ops::Range; +use std::path::{Path, PathBuf}; +use std::sync::Arc; +use std::{fmt, io}; + +use async_trait::async_trait; +use quickwit_storage::{Resolution, SplitDownloadView}; +use tantivy::directory::error::OpenReadError; +use tantivy::directory::{FileHandle, OwnedBytes}; +use tantivy::{Directory, HasLen}; +use tracing::{error, instrument}; + +use crate::hot_directory::{StaticDirectoryCache, StaticSliceCache}; + +/// A tantivy [`Directory`] backed by the storage download manager. +/// +/// It replaces the historical `StorageDirectory → CachingDirectory → HotDirectory` +/// tower with a single thin adapter: +/// - the static hotcache (built at index time, keyed by logical file) is consulted first, +/// synchronously; +/// - otherwise reads are delegated to the synchronous [`SplitDownloadView::resolve`] (for tantivy's +/// sync `read_bytes`) or the thin async [`SplitDownloadView::get`] (for warmup). +/// +/// As with the former `StorageDirectory`, a synchronous read of a range that was +/// not warmed up (and is not in the hotcache) is a loud error rather than a silent +/// blocking download — search must warm up everything it reads. +#[derive(Clone)] +pub struct DownloadManagerDirectory { + inner: Arc, +} + +struct InnerDownloadManagerDirectory { + view: Arc, + hotcache: Arc, +} + +impl DownloadManagerDirectory { + /// Opens a directory over the given per-split download view, using the + /// split's static hotcache bytes (the same bytes the former `HotDirectory` + /// consumed). + pub fn open( + view: Arc, + hotcache_bytes: OwnedBytes, + ) -> anyhow::Result { + let hotcache = Arc::new(StaticDirectoryCache::open(hotcache_bytes)?); + Ok(DownloadManagerDirectory { + inner: Arc::new(InnerDownloadManagerDirectory { view, hotcache }), + }) + } + + /// Returns all the files in the directory and their sizes. + /// + /// The actual cached data is a very small fraction of this length. This is + /// the replacement for `HotDirectory::get_file_lengths`, used to estimate the + /// in-memory index size. + pub fn get_file_lengths(&self) -> Vec<(PathBuf, u64)> { + self.inner.hotcache.get_file_lengths() + } +} + +impl fmt::Debug for DownloadManagerDirectory { + fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { + write!(f, "DownloadManagerDirectory") + } +} + +struct DownloadManagerFileHandle { + view: Arc, + static_cache: Arc, + path: PathBuf, + file_length: u64, +} + +impl fmt::Debug for DownloadManagerFileHandle { + fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { + write!(f, "DownloadManagerFileHandle({:?})", &self.path) + } +} + +impl HasLen for DownloadManagerFileHandle { + fn len(&self) -> usize { + self.file_length as usize + } +} + +#[async_trait] +impl FileHandle for DownloadManagerFileHandle { + fn read_bytes(&self, byte_range: Range) -> io::Result { + if let Some(found_bytes) = self.static_cache.try_read_bytes(byte_range.clone()) { + return Ok(found_bytes); + } + match self.view.resolve(&self.path, byte_range) { + Resolution::Hit(bytes) => Ok(bytes), + Resolution::Miss(_) => Err(unsupported_sync_read(&self.path)), + } + } + + #[instrument(level = "debug", fields(path = %self.path.to_string_lossy(), byte_range_size = byte_range.end - byte_range.start), skip(self))] + async fn read_bytes_async(&self, byte_range: Range) -> io::Result { + if let Some(found_bytes) = self.static_cache.try_read_bytes(byte_range.clone()) { + return Ok(found_bytes); + } + let bytes = self.view.get(&self.path, byte_range).await?; + Ok(bytes) + } +} + +/// Error returned when a synchronous read hits a range that was neither in the +/// static hotcache nor warmed up. Mirrors the former `StorageDirectory` +/// behavior: such a read is a bug, not a silent blocking download. +fn unsupported_sync_read(path: &Path) -> io::Error { + let error = "unsupported operation: synchronous read of an un-warmed byte range"; + error!(error, ?path); + io::Error::other(format!("{error}: {}", path.display())) +} + +impl Directory for DownloadManagerDirectory { + fn get_file_handle(&self, path: &Path) -> Result, OpenReadError> { + let file_length = self + .inner + .hotcache + .get_file_length(path) + .ok_or_else(|| OpenReadError::FileDoesNotExist(path.to_owned()))?; + Ok(Arc::new(DownloadManagerFileHandle { + view: self.inner.view.clone(), + static_cache: self.inner.hotcache.get_slice(path), + path: path.to_owned(), + file_length, + })) + } + + fn exists(&self, path: &Path) -> Result { + Ok(self.inner.hotcache.get_file_length(path).is_some()) + } + + fn atomic_read(&self, path: &Path) -> Result, OpenReadError> { + let slice_cache = self.inner.hotcache.get_slice(path); + if let Some(all_bytes) = slice_cache.try_read_all() { + return Ok(all_bytes.as_slice().to_owned()); + } + // Fallback for files not fully present in the hotcache: serve from the + // synchronous cache tiers (must be warmed). This matches the former + // tower, where a non-hotcached atomic_read fell through to a sync read + // that errored if the data was not already cached. + let file_length = self + .inner + .hotcache + .get_file_length(path) + .ok_or_else(|| OpenReadError::FileDoesNotExist(path.to_owned()))?; + match self.inner.view.resolve(path, 0..file_length as usize) { + Resolution::Hit(bytes) => Ok(bytes.as_slice().to_owned()), + Resolution::Miss(_) => Err(OpenReadError::wrap_io_error( + unsupported_sync_read(path), + path.to_owned(), + )), + } + } + + crate::read_only_directory!(); +} diff --git a/quickwit/quickwit-directories/src/hot_directory.rs b/quickwit/quickwit-directories/src/hot_directory.rs index 2019505b542..8f3bb49e8f3 100644 --- a/quickwit/quickwit-directories/src/hot_directory.rs +++ b/quickwit/quickwit-directories/src/hot_directory.rs @@ -165,7 +165,7 @@ impl StaticDirectoryCacheBuilder { } #[derive(Debug)] -struct StaticDirectoryCache { +pub(crate) struct StaticDirectoryCache { file_lengths: HashMap, slices: HashMap>, } diff --git a/quickwit/quickwit-directories/src/lib.rs b/quickwit/quickwit-directories/src/lib.rs index e9b1add1692..ff6dafa7168 100644 --- a/quickwit/quickwit-directories/src/lib.rs +++ b/quickwit/quickwit-directories/src/lib.rs @@ -14,8 +14,8 @@ //! This crate contains all of the building pieces that make quickwit's IO possible. //! -//! - The `StorageDirectory` just wraps a `Storage` trait to make it compatible with tantivy's -//! Directory API. +//! - The `DownloadManagerDirectory` is the search read-path directory: it serves reads through the +//! storage download manager (static hotcache + unified caches + a single thin download). //! - The `BundleDirectory` bundles multiple files into a single file. //! - The `HotDirectory` wraps another directory with a static cache. //! - The `CachingDirectory` wraps a Directory with a dynamic cache. @@ -27,15 +27,15 @@ mod bundle_directory; mod caching_directory; mod debug_proxy_directory; +mod download_manager_directory; mod hot_directory; -mod storage_directory; mod union_directory; pub use self::bundle_directory::{BundleDirectory, get_hotcache_from_split, read_split_footer}; pub use self::caching_directory::CachingDirectory; pub use self::debug_proxy_directory::{DebugProxyDirectory, ReadOperation}; +pub use self::download_manager_directory::DownloadManagerDirectory; pub use self::hot_directory::{HotDirectory, write_hotcache}; -pub use self::storage_directory::StorageDirectory; pub use self::union_directory::UnionDirectory; macro_rules! read_only_directory { diff --git a/quickwit/quickwit-directories/src/storage_directory.rs b/quickwit/quickwit-directories/src/storage_directory.rs deleted file mode 100644 index 31ee6e3f736..00000000000 --- a/quickwit/quickwit-directories/src/storage_directory.rs +++ /dev/null @@ -1,142 +0,0 @@ -// 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::Debug; -use std::ops::Range; -use std::path::{Path, PathBuf}; -use std::sync::Arc; -use std::{fmt, io}; - -use async_trait::async_trait; -use quickwit_common::uri::Uri; -use quickwit_storage::{OwnedBytes, Storage}; -use tantivy::directory::FileHandle; -use tantivy::directory::error::OpenReadError; -use tantivy::{Directory, HasLen}; -use tracing::{error, instrument}; - -struct StorageDirectoryFileHandle { - storage_directory: StorageDirectory, - path: PathBuf, -} - -impl HasLen for StorageDirectoryFileHandle { - fn len(&self) -> usize { - unimplemented!() - } -} - -impl fmt::Debug for StorageDirectoryFileHandle { - fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { - write!( - f, - "StorageDirectoryFileHandle({:?}, dir={:?})", - &self.path, self.storage_directory - ) - } -} - -#[async_trait] -impl FileHandle for StorageDirectoryFileHandle { - fn read_bytes(&self, _byte_range: Range) -> io::Result { - Err(unsupported_operation(&self.path)) - } - - #[instrument(level = "debug", fields(path = %self.path.to_string_lossy(), byte_range_size = byte_range.end - byte_range.start), skip(self))] - async fn read_bytes_async(&self, byte_range: Range) -> io::Result { - if byte_range.is_empty() { - return Ok(OwnedBytes::empty()); - } - let object_bytes = self - .storage_directory - .get_slice(&self.path, byte_range) - .await?; - Ok(object_bytes) - } -} - -/// Directory backed a quickwit `Storage` abstraction. -/// -/// It should not be used in a context outside quickwit, as it contains -/// several pitfalls: -/// Fetching data synchronously panics. -/// Writing data panics. -/// -/// This directory is fetch slices of data to a possibly distant storage -/// everytime `read_bytes` is called. -#[derive(Clone)] -pub struct StorageDirectory { - storage: Arc, -} - -impl Debug for StorageDirectory { - fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { - write!(f, "StorageDirectory({:?})", self.uri()) - } -} - -impl StorageDirectory { - /// Creates a new StorageDirectory, backed by the given `storage`. - pub fn new(storage: Arc) -> StorageDirectory { - StorageDirectory { storage } - } - - /// Fetches a slice of byte from a file asynchronously. - pub async fn get_slice(&self, path: &Path, range: Range) -> io::Result { - let payload: OwnedBytes = self.storage.get_slice(path, range).await?; - Ok(payload) - } - - /// Fetches an entire file asynchronously. - pub async fn get_all(&self, path: &Path) -> io::Result { - let payload: OwnedBytes = self.storage.get_all(path).await?; - Ok(payload) - } - - /// Returns the uri associated to the underlying storage. - pub fn uri(&self) -> &Uri { - self.storage.uri() - } -} - -fn unsupported_operation(path: &Path) -> io::Error { - let error = "unsupported operation: `StorageDirectory` only supports async reads"; - error!(error, ?path); - io::Error::other(format!("{error}: {}", path.display())) -} - -impl Directory for StorageDirectory { - fn get_file_handle(&self, path: &Path) -> Result, OpenReadError> { - Ok(Arc::new(StorageDirectoryFileHandle { - storage_directory: self.clone(), - path: path.to_path_buf(), - })) - } - - fn atomic_read(&self, path: &Path) -> Result, OpenReadError> { - Err(OpenReadError::wrap_io_error( - unsupported_operation(path), - path.to_path_buf(), - )) - } - - fn exists(&self, path: &std::path::Path) -> Result { - Err(OpenReadError::wrap_io_error( - unsupported_operation(path), - path.to_path_buf(), - )) - } - - crate::read_only_directory!(); -} diff --git a/quickwit/quickwit-search/src/leaf.rs b/quickwit/quickwit-search/src/leaf.rs index c752afc8999..f8963f06347 100644 --- a/quickwit/quickwit-search/src/leaf.rs +++ b/quickwit/quickwit-search/src/leaf.rs @@ -29,7 +29,7 @@ use futures::future::try_join_all; use quickwit_common::pretty::PrettySample; use quickwit_common::thread_pool::with_priority::Priority; use quickwit_common::uri::Uri; -use quickwit_directories::{CachingDirectory, HotDirectory, StorageDirectory}; +use quickwit_directories::DownloadManagerDirectory; use quickwit_doc_mapper::{Automaton, DocMapper, FastFieldWarmupInfo, TermRange, WarmupInfo}; use quickwit_metrics::{GaugeGuard, HistogramTimer}; use quickwit_proto::search::lambda_single_split_result::Outcome; @@ -42,8 +42,8 @@ use quickwit_query::query_ast::{ }; use quickwit_query::tokenizers::TokenizerManager; use quickwit_storage::{ - BundleStorage, ByteRangeCache, CountingStorage, MemorySizedCache, OwnedBytes, SplitCache, - Storage, StorageResolver, TimeoutAndRetryStorage, wrap_storage_with_cache, + BundleStorage, BundleStorageFileOffsets, ByteRangeCache, CountingStorage, MemorySizedCache, + OwnedBytes, SplitCache, SplitDownloadView, Storage, StorageResolver, fetch_split_footer, }; use tantivy::aggregation::AggContextParams; use tantivy::aggregation::agg_req::{AggregationVariants, Aggregations}; @@ -190,61 +190,79 @@ pub(crate) async fn open_split_bundle( Ok((hotcache_bytes, bundle_storage)) } -/// Add a storage proxy to retry `get_slice` requests if they are taking too long, -/// if configured in the searcher config. +/// Opens a `tantivy::Index` for the given split, backed by the download manager. /// -/// The goal here is too ensure a low latency. -fn configure_storage_retries( - searcher_context: &SearcherContext, - index_storage: Arc, -) -> Arc { - if let Some(storage_timeout_policy) = &searcher_context.searcher_config.storage_timeout_policy { - Arc::new(TimeoutAndRetryStorage::new( - index_storage, - storage_timeout_policy.clone(), - )) - } else { - index_storage - } -} - -/// Opens a `tantivy::Index` for the given split with several cache layers: -/// - A split footer cache given by `SearcherContext.split_footer_cache`. -/// - A fast fields cache given by `SearcherContext.storage_long_term_cache`. -/// - An ephemeral unbounded cache directory (whose lifetime is tied to the returned `Index` if no -/// `ByteRangeCache` is provided). +/// The returned [`DownloadManagerDirectory`] unifies the former +/// `StorageDirectory → CachingDirectory → HotDirectory` tower: the static +/// hotcache, the long-term fast-field cache, the on-disk split cache and the +/// ephemeral byte-range cache are all consulted through a single synchronous +/// `resolve`, with a single thin async download on a miss. +/// +/// - The split footer is cached by `SearcherContext.split_footer_cache`. +/// - `ephemeral_unbounded_cache`, when provided, is the per-request short-lived byte-range cache +/// populated during warmup. pub(crate) async fn open_index_with_caches( searcher_context: &SearcherContext, index_storage: Arc, split_and_footer_offsets: &SplitIdAndFooterOffsets, tokenizer_manager: Option<&TokenizerManager>, ephemeral_unbounded_cache: Option, -) -> anyhow::Result<(Index, HotDirectory)> { - let index_storage_with_retry_on_timeout = - configure_storage_retries(searcher_context, index_storage); - - let (hotcache_bytes, bundle_storage) = open_split_bundle( - searcher_context, - index_storage_with_retry_on_timeout, - split_and_footer_offsets, - ) - .await?; - - let bundle_storage_with_cache = wrap_storage_with_cache( - searcher_context.fast_fields_cache.clone(), - Arc::new(bundle_storage), - ); - - let directory = StorageDirectory::new(bundle_storage_with_cache); +) -> anyhow::Result<(Index, DownloadManagerDirectory)> { + let split_path = PathBuf::from(format!("{}.split", split_and_footer_offsets.split_id)); + let timeout_policy = searcher_context + .searcher_config + .storage_timeout_policy + .clone(); + + // Fetch (or read from cache) the split footer, which holds both the hotcache + // and the bundle file offsets. On a cache miss the footer is fetched through + // the download manager (request dedup + the aggressive timeout policy). + let footer_cache = &searcher_context.split_footer_cache; + let footer_data = + if let Some(footer_data) = footer_cache.get(&split_and_footer_offsets.split_id) { + footer_data + } else { + let footer_range = split_and_footer_offsets.split_footer_start as usize + ..split_and_footer_offsets.split_footer_end as usize; + let footer_data = fetch_split_footer( + &searcher_context.download_caches, + index_storage.clone(), + timeout_policy.clone(), + split_path.clone(), + footer_range, + ) + .await + .with_context(|| { + format!( + "failed to fetch hotcache and footer from {} for split `{}`", + index_storage.uri(), + split_and_footer_offsets.split_id + ) + })?; + footer_cache.put( + split_and_footer_offsets.split_id.to_owned(), + footer_data.clone(), + ); + footer_data + }; - let hot_directory = if let Some(cache) = ephemeral_unbounded_cache { - let caching_directory = CachingDirectory::new(Arc::new(directory), cache); - HotDirectory::open(caching_directory, hotcache_bytes.read_bytes()?)? - } else { - HotDirectory::open(directory, hotcache_bytes.read_bytes()?)? - }; + // Parse the hotcache and the bundle logical→physical offsets out of the + // footer. This is pure arithmetic + metadata lookup, no I/O. + let (hotcache_slice, bundle_offsets) = + BundleStorageFileOffsets::open_from_split_data(FileSlice::new(Arc::new(footer_data)))?; + let hotcache_bytes = hotcache_slice.read_bytes()?; - let mut index = Index::open(hot_directory.clone())?; + let view = Arc::new(SplitDownloadView::new( + &searcher_context.download_caches, + index_storage, + timeout_policy, + Arc::new(bundle_offsets), + split_path, + ephemeral_unbounded_cache, + )); + let directory = DownloadManagerDirectory::open(view, hotcache_bytes)?; + + let mut index = Index::open(directory.clone())?; if let Some(tokenizer_manager) = tokenizer_manager { index.set_tokenizers(tokenizer_manager.tantivy_manager().clone()); } @@ -253,7 +271,7 @@ pub(crate) async fn open_index_with_caches( .tantivy_manager() .clone(), ); - Ok((index, hot_directory)) + Ok((index, directory)) } /// Outcome of [`warmup`]. @@ -611,9 +629,8 @@ fn leaf_resource_stats_for_split(split_stats: SplitResourceStats) -> LeafResourc } /// Compute the size of the index, store excluded. -fn compute_index_size(hot_directory: &HotDirectory) -> ByteSize { - let size_bytes = hot_directory - .get_file_lengths() +fn compute_index_size(file_lengths: &[(PathBuf, u64)]) -> ByteSize { + let size_bytes = file_lengths .iter() .filter(|(path, _)| !path.to_string_lossy().ends_with("store")) .map(|(_, size)| *size) @@ -662,7 +679,7 @@ async fn leaf_search_single_split( // ABOVE this wrapper, so reads served from cache do not contribute to the // counters — that is the desired "downloaded from object storage" semantics. let (storage, download_counters) = CountingStorage::instrument_storage(storage); - let (index, hot_directory) = open_index_with_caches( + let (index, directory) = open_index_with_caches( &ctx.searcher_context, storage, &split, @@ -671,7 +688,7 @@ async fn leaf_search_single_split( ) .await?; - let index_size = compute_index_size(&hot_directory); + let index_size = compute_index_size(&directory.get_file_lengths()); if index_size < search_permit.memory_allocation() { search_permit.update_memory_usage(index_size); } @@ -2164,7 +2181,7 @@ mod tests { use async_trait::async_trait; use bytes::BufMut; use quickwit_config::{LambdaConfig, SearcherConfig}; - use quickwit_directories::write_hotcache; + use quickwit_directories::{HotDirectory, write_hotcache}; use quickwit_proto::search::LambdaSingleSplitResult; use rand::RngExt; use tantivy::TantivyDocument; @@ -2678,14 +2695,15 @@ mod tests { &payload, ); let size_with_stored_payload = - compute_index_size(&hotcache_directory_stored_payload).as_u64(); + compute_index_size(&hotcache_directory_stored_payload.get_file_lengths()).as_u64(); let (hotcache_directory_index_only, directory_size_index_only) = create_tantivy_dir_with_hotcache( FieldEntry::new_bytes("payload".to_string(), BytesOptions::default()), &payload, ); - let size_index_only = compute_index_size(&hotcache_directory_index_only).as_u64(); + let size_index_only = + compute_index_size(&hotcache_directory_index_only.get_file_lengths()).as_u64(); assert!(directory_size_stored_payload > directory_size_index_only + 1000); assert!(size_with_stored_payload.abs_diff(size_index_only) < 10); @@ -2714,13 +2732,15 @@ mod tests { iure reprehenderit qui in ea voluptate velit esse quam nihil molestiae consequatur, \ vel illum qui dolorem eum fugiat quo voluptas nulla pariatur?", ); - let larger_size = compute_index_size(&hotcache_directory_larger).as_u64(); + let larger_size = + compute_index_size(&hotcache_directory_larger.get_file_lengths()).as_u64(); let (hotcache_directory_smaller, directory_size_smaller) = create_tantivy_dir_with_hotcache( FieldEntry::new_text("text".to_string(), indexing_options), "hi", ); - let smaller_size = compute_index_size(&hotcache_directory_smaller).as_u64(); + let smaller_size = + compute_index_size(&hotcache_directory_smaller.get_file_lengths()).as_u64(); assert!(directory_size_larger > directory_size_smaller + 100); assert!(larger_size > smaller_size + 100); diff --git a/quickwit/quickwit-search/src/service.rs b/quickwit/quickwit-search/src/service.rs index db2b066ae83..3c26e44cc54 100644 --- a/quickwit/quickwit-search/src/service.rs +++ b/quickwit/quickwit-search/src/service.rs @@ -28,9 +28,7 @@ use quickwit_proto::search::{ ReportSplitsRequest, ReportSplitsResponse, RootResourceStats, ScrollRequest, SearchPlanResponse, SearchRequest, SearchResponse, SnippetRequest, }; -use quickwit_storage::{ - MemorySizedCache, QuickwitCache, SplitCache, StorageCache, StorageResolver, -}; +use quickwit_storage::{MemorySizedCache, SearchDownloadCaches, SplitCache, StorageResolver}; use tantivy::aggregation::AggregationLimitsGuard; use crate::invoker::LambdaLeafSearchInvoker; @@ -414,8 +412,9 @@ pub(crate) async fn scroll( pub struct SearcherContext { /// Searcher config. pub searcher_config: SearcherConfig, - /// Fast fields cache. - pub fast_fields_cache: Arc, + /// Shared, cross-request download-path caches (long-term fast-field cache, + /// on-disk split cache, and request debouncer) used by the search read path. + pub download_caches: SearchDownloadCaches, /// Counting semaphore to limit concurrent leaf search split requests. pub search_permit_provider: SearchPermitProvider, /// Split footer cache. @@ -477,8 +476,8 @@ impl SearcherContext { searcher_config.max_num_concurrent_split_searches, searcher_config.warmup_memory_budget, ); - let storage_long_term_cache = - Arc::new(QuickwitCache::new(&searcher_config.fast_field_cache)); + let download_caches = + SearchDownloadCaches::new(&searcher_config.fast_field_cache, split_cache_opt.clone()); let leaf_search_cache = LeafSearchCache::new(&searcher_config.partial_request_cache); let predicate_cache = PredicateCacheImpl::new(&searcher_config.predicate_cache); let list_fields_cache = ListFieldsCache::new(&searcher_config.partial_request_cache); @@ -492,7 +491,7 @@ impl SearcherContext { Self { searcher_config, - fast_fields_cache: storage_long_term_cache, + download_caches, predicate_cache: predicate_cache.into(), search_permit_provider: leaf_search_split_semaphore, split_footer_cache: global_split_footer_cache, diff --git a/quickwit/quickwit-storage/src/bundle_storage.rs b/quickwit/quickwit-storage/src/bundle_storage.rs index ea073c830ba..785446dc40c 100644 --- a/quickwit/quickwit-storage/src/bundle_storage.rs +++ b/quickwit/quickwit-storage/src/bundle_storage.rs @@ -138,7 +138,7 @@ impl BundleStorageFileOffsets { /// See docs/internals/split-format.md /// [Files, FileMetadata, FileMetadata Len, HotCache, HotCache Len] /// Returns (Hotcache, Self) - fn open_from_split_data(file: FileSlice) -> anyhow::Result<(FileSlice, Self)> { + pub fn open_from_split_data(file: FileSlice) -> anyhow::Result<(FileSlice, Self)> { let (bundle_and_hotcache_bytes, hotcache_num_bytes_data) = file.split_from_end(SPLIT_HOTBYTES_FOOTER_LENGTH_NUM_BYTES); let hotcache_num_bytes: u32 = u32::from_le_bytes( diff --git a/quickwit/quickwit-storage/src/cache/mod.rs b/quickwit/quickwit-storage/src/cache/mod.rs index f73a96b90f4..b97d88b510f 100644 --- a/quickwit/quickwit-storage/src/cache/mod.rs +++ b/quickwit/quickwit-storage/src/cache/mod.rs @@ -15,7 +15,6 @@ mod base_cache; mod byte_range_cache; mod memory_sized_cache; -mod quickwit_cache; mod slice_address; mod storage_with_cache; mod stored_item; @@ -25,7 +24,6 @@ use std::path::{Path, PathBuf}; use std::sync::Arc; use async_trait::async_trait; -pub use quickwit_cache::QuickwitCache; pub use storage_with_cache::StorageWithCache; pub use self::byte_range_cache::ByteRangeCache; diff --git a/quickwit/quickwit-storage/src/cache/quickwit_cache.rs b/quickwit/quickwit-storage/src/cache/quickwit_cache.rs deleted file mode 100644 index f0da1266164..00000000000 --- a/quickwit/quickwit-storage/src/cache/quickwit_cache.rs +++ /dev/null @@ -1,207 +0,0 @@ -// 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::ops::Range; -use std::path::{Path, PathBuf}; -use std::sync::Arc; - -use async_trait::async_trait; -use quickwit_config::CacheConfig; - -use crate::OwnedBytes; -use crate::cache::{MemorySizedCache, StorageCache}; -use crate::metrics::{CacheMetrics, FAST_FIELD_CACHE}; - -const FULL_SLICE: Range = 0..usize::MAX; - -/// Quickwit storage cache with a size limit. -/// It is used currently by to cache only fast fields data. -pub struct QuickwitCache { - router: Vec<(&'static str, Arc)>, -} - -impl From)>> for QuickwitCache { - fn from(router: Vec<(&'static str, Arc)>) -> Self { - QuickwitCache { router } - } -} - -impl QuickwitCache { - /// Creates a [`QuickwitCache`] with a cache on fast fields. - pub fn new(cache_config: &CacheConfig) -> Self { - let mut quickwit_cache = QuickwitCache::empty(); - let fast_field_cache_counters: &'static CacheMetrics = &FAST_FIELD_CACHE; - quickwit_cache.add_route( - ".fast", - Arc::new(SimpleCache::from_config( - cache_config, - fast_field_cache_counters, - )), - ); - quickwit_cache - } - - /// Empties cache. - pub fn empty() -> QuickwitCache { - QuickwitCache::from(Vec::new()) - } - - /// Adds a caching route defined by a path suffix. All elements with a path matching - /// this suffix will be cached. - pub fn add_route(&mut self, path_suffix: &'static str, route_cache: Arc) { - self.router.push((path_suffix, route_cache)); - } - - fn get_relevant_cache(&self, path: &Path) -> Option<&dyn StorageCache> { - for (suffix, cache) in &self.router { - if path.to_string_lossy().ends_with(suffix) { - return Some(cache.as_ref()); - } - } - None - } -} - -#[async_trait] -impl StorageCache for QuickwitCache { - async fn get(&self, path: &Path, byte_range: Range) -> Option { - // We don't check for the presence of the entire file in the - // cache. - // That's voluntary to avoid messing with the cache miss counts. - if let Some(cache) = self.get_relevant_cache(path) { - return cache.get(path, byte_range).await; - } - None - } - - async fn get_all(&self, path: &Path) -> Option { - if let Some(cache) = self.get_relevant_cache(path) { - return cache.get_all(path).await; - } - None - } - - async fn put(&self, path: PathBuf, byte_range: Range, bytes: OwnedBytes) { - if let Some(cache) = self.get_relevant_cache(&path) { - cache.put(path, byte_range, bytes).await; - } - } - - async fn put_all(&self, path: PathBuf, bytes: OwnedBytes) { - if let Some(cache) = self.get_relevant_cache(&path) { - cache.put(path, FULL_SLICE, bytes).await; - } - } -} - -/// The Quickwit cache logic is very simple for the moment. -/// -/// It stores hotcache files using an LRU cache. -/// -/// HACK! We use `0..usize::MAX` to signify the "entire file". -/// TODO fixme -struct SimpleCache { - slice_cache: MemorySizedCache, -} - -impl SimpleCache { - fn from_config(cache_config: &CacheConfig, cache_counters: &'static CacheMetrics) -> Self { - SimpleCache { - slice_cache: MemorySizedCache::from_config(cache_config, cache_counters), - } - } -} - -#[async_trait] -impl StorageCache for SimpleCache { - async fn get(&self, path: &Path, byte_range: Range) -> Option { - if let Some(bytes) = self.slice_cache.get_slice(path, byte_range) { - return Some(bytes); - } - None - } - - async fn put(&self, path: PathBuf, byte_range: Range, bytes: OwnedBytes) { - self.slice_cache.put_slice(path, byte_range, bytes); - } - - async fn get_all(&self, path: &Path) -> Option { - self.slice_cache.get_slice(path, FULL_SLICE) - } - - async fn put_all(&self, path: PathBuf, bytes: OwnedBytes) { - self.slice_cache.put_slice(path, FULL_SLICE.clone(), bytes); - } -} - -#[cfg(test)] -mod tests { - use std::path::Path; - use std::sync::Arc; - - use super::QuickwitCache; - use crate::cache::StorageCache; - use crate::{MockStorageCache, OwnedBytes}; - - #[tokio::test] - async fn test_quickwit_cache_get_all() { - let mock_cache_hotcache = MockStorageCache::default(); - let mut mock_cache_fast = MockStorageCache::default(); - mock_cache_fast - .expect_get_all() - .times(1) - .withf(|path| path == Path::new("bubu/toto.fast")) - .returning(|_| Some(OwnedBytes::new(&b"aaaa"[..]))); - let mut quickwit_cache = QuickwitCache::empty(); - quickwit_cache.add_route("hotcache", Arc::new(mock_cache_hotcache)); - quickwit_cache.add_route("fast", Arc::new(mock_cache_fast)); - quickwit_cache.get_all(Path::new("bubu/toto.fast")).await; - } - - #[tokio::test] - async fn test_quickwit_cache_get() { - let mock_cache_hotcache = MockStorageCache::default(); - let mut mock_cache = MockStorageCache::default(); - mock_cache - .expect_get() - .times(1) - .withf(|path, _| path == Path::new("bubu/toto.fast")) - .returning(|_, _| Some(OwnedBytes::new(&b"aaaaa"[..]))); - let mut quickwit_cache = QuickwitCache::empty(); - quickwit_cache.add_route("hotcache", Arc::new(mock_cache_hotcache)); - quickwit_cache.add_route("fast", Arc::new(mock_cache)); - quickwit_cache.get(Path::new("bubu/toto.fast"), 5..10).await; - } - - #[tokio::test] - async fn test_quickwit_cache_priority() { - let mut mock_cache_ast = MockStorageCache::default(); - mock_cache_ast - .expect_get() - .times(1) - .withf(|path, _| path == Path::new("bubu/toto.fast")) - .returning(|_, _| Some(OwnedBytes::new(&b"aaaaa"[..]))); - let mock_cache_fast = MockStorageCache::default(); - let mut quickwit_cache = QuickwitCache::empty(); - quickwit_cache.add_route("ast", Arc::new(mock_cache_ast)); - quickwit_cache.add_route("fast", Arc::new(mock_cache_fast)); - assert_eq!( - quickwit_cache - .get(Path::new("bubu/toto.fast"), 5..10) - .await - .unwrap(), - &b"aaaaa"[..] - ); - } -} diff --git a/quickwit/quickwit-storage/src/debouncer.rs b/quickwit/quickwit-storage/src/debouncer.rs index 4841d979ee4..5746c0e9ee9 100644 --- a/quickwit/quickwit-storage/src/debouncer.rs +++ b/quickwit/quickwit-storage/src/debouncer.rs @@ -12,23 +12,13 @@ // See the License for the specific language governing permissions and // limitations under the License. -use std::fmt; use std::hash::Hash; -use std::ops::Range; -use std::path::{Path, PathBuf}; +use std::sync::Mutex; use std::sync::atomic::{AtomicUsize, Ordering}; -use std::sync::{Arc, Mutex}; use ahash::HashMap; -use async_trait::async_trait; use futures::future::{BoxFuture, Shared, WeakShared}; use futures::{Future, FutureExt}; -use quickwit_common::uri::Uri; -use tantivy::directory::OwnedBytes; -use tokio::io::AsyncRead; - -use crate::storage::SendableAsync; -use crate::{BulkDeleteError, Storage, StorageResult}; /// The AsyncDebouncer debounces inflight Futures, so that concurrent async request to the same data /// source can be deduplicated. @@ -87,7 +77,11 @@ impl AsyncDebouncer { /// Holding the lock across both the lookup and the insert is deliberate: it makes the /// lookup-then-insert atomic, so two concurrent callers for the same key cannot both build /// and race to insert. The lock is always released before the future is awaited. - fn get_or_create(&self, key: K, build_a_future_fast: T) -> Shared> + pub(crate) fn get_or_create( + &self, + key: K, + build_a_future_fast: T, + ) -> Shared> where T: FnOnce() -> F, F: Future + Send + 'static, @@ -126,106 +120,6 @@ impl AsyncDebouncer { } } -type DebouncerKey = (PathBuf, Range); - -/// Just to keep in mind there is a race condition on debouncing, when combined with delete -/// -/// All on the same key -/// start get R1 -/// start delete R2 -/// end delete R2 -/// start get R3 -/// end get R1 -/// end get R3 -/// -/// ==> R3 would return the cached result, although the resource has been deleted. -pub(crate) struct DebouncedStorage { - // wrap both in Arc, because the Future is stored in the cache, which has 'static lifetime - // associated - underlying: Arc, - slice_debouncer: Arc>>, -} - -impl fmt::Debug for DebouncedStorage { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - f.debug_struct("DebouncedStorage").finish() - } -} - -impl DebouncedStorage { - pub(crate) fn new(underlying: T) -> Self { - Self { - underlying: Arc::new(underlying), - slice_debouncer: Arc::new(AsyncDebouncer::default()), - } - } -} - -#[async_trait] -impl Storage for DebouncedStorage { - async fn check_connectivity(&self) -> anyhow::Result<()> { - self.underlying.check_connectivity().await - } - - async fn put( - &self, - path: &Path, - payload: Box, - ) -> crate::StorageResult<()> { - self.underlying.put(path, payload).await - } - - async fn copy_to(&self, path: &Path, output: &mut dyn SendableAsync) -> StorageResult<()> { - self.underlying.copy_to(path, output).await - } - - async fn get_slice(&self, path: &Path, range: Range) -> StorageResult { - let (debouncer, underlying) = (self.slice_debouncer.clone(), self.underlying.clone()); - let key = (path.to_owned(), range); - debouncer - .get_or_create(key.clone(), || async move { - underlying.get_slice(&key.0, key.1).await - }) - .await - } - - async fn get_slice_stream( - &self, - path: &Path, - range: Range, - ) -> StorageResult> { - // Getting a stream bypasses the debouncer - self.underlying.get_slice_stream(path, range).await - } - - async fn delete(&self, path: &Path) -> StorageResult<()> { - self.underlying.delete(path).await - } - - async fn bulk_delete<'a>(&self, paths: &[&'a Path]) -> Result<(), BulkDeleteError> { - self.underlying.bulk_delete(paths).await - } - - async fn get_all(&self, path: &Path) -> StorageResult { - let (debouncer, underlying) = (self.slice_debouncer.clone(), self.underlying.clone()); - let key = (path.to_owned(), 0..usize::MAX); - debouncer - .get_or_create( - key.clone(), - || async move { underlying.get_all(&key.0).await }, - ) - .await - } - - fn uri(&self) -> &Uri { - self.underlying.uri() - } - - async fn file_num_bytes(&self, path: &Path) -> StorageResult { - self.underlying.file_num_bytes(path).await - } -} - #[cfg(test)] mod tests { diff --git a/quickwit/quickwit-storage/src/download_manager/download.rs b/quickwit/quickwit-storage/src/download_manager/download.rs new file mode 100644 index 00000000000..55dd9b3d217 --- /dev/null +++ b/quickwit/quickwit-storage/src/download_manager/download.rs @@ -0,0 +1,325 @@ +// 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. + +//! The single thin async layer of the download manager. +//! +//! Everything here is entered only on a cache miss. It folds together what used +//! to be three separate `Storage` decorators: request debouncing +//! ([`crate::AsyncDebouncer`]), the on-disk split cache fill, and the aggressive +//! per-attempt timeout + retry. The actual object-store concurrency limiting and +//! AWS retry still live in the backend, so they are not re-implemented here. + +use std::ops::Range; +use std::path::{Path, PathBuf}; +use std::sync::Arc; + +use quickwit_common::uri::Uri; +use quickwit_common::{rate_limited_info, rate_limited_warn}; +use quickwit_config::StorageTimeoutPolicy; + +use super::{DownloadKey, DownloadPlan, SearchDownloadCaches, SplitDownloadView}; +use crate::split_cache::split_id_from_path; +use crate::{OwnedBytes, SplitCache, Storage, StorageErrorKind, StorageResult}; + +impl SplitDownloadView { + /// The thin async layer. Entered only on a [`super::Resolution::Miss`]. + /// + /// Translates the logical (file-relative) range to an absolute range in the + /// physical split, fetches it (debounced, via the on-disk split cache then the + /// backend with the timeout policy), and writes the result back into the + /// ephemeral and — for `.fast` files — long-term caches so that subsequent + /// resolves are synchronous hits. + pub(crate) async fn download(&self, plan: DownloadPlan) -> StorageResult { + let file_offsets = self + .bundle_offsets + .get(&plan.file_in_split) + .ok_or_else(|| { + StorageErrorKind::NotFound.with_error(anyhow::anyhow!( + "missing file `{}` in split `{}`", + plan.file_in_split.display(), + self.split_path.display(), + )) + })?; + let abs_range = file_offsets.start as usize + plan.byte_range.start + ..file_offsets.start as usize + plan.byte_range.end; + + let key = DownloadKey { + split_path: self.split_path.clone(), + file_in_split: plan.file_in_split.clone(), + byte_range: plan.byte_range.clone(), + }; + let backend = self.backend.clone(); + let split_cache_opt = self.split_cache_opt.clone(); + let timeout_policy = self.timeout_policy.clone(); + let split_path = self.split_path.clone(); + // The builder runs while the debouncer lock is held, so it must be cheap: + // it only constructs the future (an `async fn` call), it does not poll it. + let download_fut = self.inflight.get_or_create(key, move || { + fetch_slice( + backend, + split_cache_opt, + split_path, + abs_range, + timeout_policy, + ) + }); + let bytes = download_fut.await?; + + let cache_tag = self.cache_tag(&plan.file_in_split); + if let Some(byte_range_cache) = &self.byte_range_cache { + byte_range_cache.put_slice(cache_tag.clone(), plan.byte_range.clone(), bytes.clone()); + } + if plan.is_fast { + self.fast_field_cache + .put_slice(cache_tag, plan.byte_range, bytes.clone()); + } + Ok(bytes) + } +} + +/// Fetches the footer byte range of a split, with request dedup and the optional +/// aggressive-timeout policy (mirroring the historical timeout+retry on the footer +/// fetch). The footer is addressed by its absolute range in the physical `.split` +/// file and does not go through the on-disk split cache. +pub async fn fetch_split_footer( + caches: &SearchDownloadCaches, + backend: Arc, + timeout_policy: Option, + split_path: PathBuf, + footer_range: Range, +) -> StorageResult { + let key = DownloadKey { + split_path: split_path.clone(), + file_in_split: PathBuf::new(), + byte_range: footer_range.clone(), + }; + let download_fut = caches.inflight.get_or_create(key, move || async move { + download_slice_with_timeout( + backend.as_ref(), + &split_path, + footer_range, + timeout_policy.as_ref(), + ) + .await + }); + download_fut.await +} + +/// Fetches an absolute byte range of a physical `.split` file: first the on-disk +/// split cache (if configured), then the backend with the timeout policy. +async fn fetch_slice( + backend: Arc, + split_cache_opt: Option>, + split_path: PathBuf, + abs_range: Range, + timeout_policy: Option, +) -> StorageResult { + if let Some(split_cache) = &split_cache_opt { + let from_split_cache = + read_from_split_cache(split_cache, backend.uri(), &split_path, abs_range.clone()).await; + record_split_cache_outcome(from_split_cache.as_ref()); + if let Some(bytes) = from_split_cache { + return Ok(bytes); + } + } + download_slice_with_timeout( + backend.as_ref(), + &split_path, + abs_range, + timeout_policy.as_ref(), + ) + .await +} + +/// Reads an absolute range from the on-disk split cache, if the split is present. +async fn read_from_split_cache( + split_cache: &SplitCache, + storage_uri: &Uri, + split_path: &Path, + abs_range: Range, +) -> Option { + let split_id = split_id_from_path(split_path)?; + let split_file = split_cache.get_split_file(split_id, storage_uri).await?; + split_file.get_range(abs_range).await.ok() +} + +/// Records a hit/miss against the `searcher_split` cache metrics, preserving the +/// behavior of the former `SplitCacheBackingStorage::record_hit_metrics`. +fn record_split_cache_outcome(result_opt: Option<&OwnedBytes>) { + let split_metrics = &crate::metrics::SEARCHER_SPLIT_CACHE.cache_metrics; + if let Some(result) = result_opt { + split_metrics.hits_num_items.inc(); + split_metrics.hits_num_bytes.inc_by(result.len() as u64); + } else { + split_metrics.misses_num_items.inc(); + } +} + +/// Downloads a slice from the backend, applying the aggressive per-attempt +/// timeout + retry policy if one is configured. Ported from the former +/// `TimeoutAndRetryStorage::get_slice` (the backend still owns AWS retry + the +/// concurrency semaphore, so neither is re-implemented here). +async fn download_slice_with_timeout( + backend: &dyn Storage, + split_path: &Path, + range: Range, + timeout_policy: Option<&StorageTimeoutPolicy>, +) -> StorageResult { + let Some(timeout_policy) = timeout_policy else { + return backend.get_slice(split_path, range).await; + }; + let num_bytes = range.len(); + for (attempt_id, timeout_duration) in timeout_policy.compute_timeout(num_bytes).enumerate() { + let get_slice_fut = backend.get_slice(split_path, range.clone()); + // TODO test avoid aborting timed out requests. #5468 + match tokio::time::timeout(timeout_duration, get_slice_fut).await { + Ok(result) => { + match attempt_id { + 0 => crate::metrics::GET_SLICE_TIMEOUT_SUCCESS_AFTER_0_TIMEOUT.inc(), + 1 => crate::metrics::GET_SLICE_TIMEOUT_SUCCESS_AFTER_1_TIMEOUT.inc(), + _ => crate::metrics::GET_SLICE_TIMEOUT_SUCCESS_AFTER_2_PLUS_TIMEOUT.inc(), + } + return result; + } + Err(_elapsed) => { + rate_limited_info!(limit_per_min = 60, num_bytes = num_bytes, path = %split_path.display(), timeout_secs = timeout_duration.as_secs_f32(), "get timeout elapsed"); + continue; + } + } + } + rate_limited_warn!(limit_per_min = 60, num_bytes = num_bytes, path = %split_path.display(), "all get_slice attempts timeouted"); + crate::metrics::GET_SLICE_TIMEOUT_ALL_TIMEOUTS.inc(); + Err(StorageErrorKind::Timeout.with_error(anyhow::anyhow!("internal timeout on get_slice"))) +} + +#[cfg(test)] +mod tests { + use std::sync::Mutex; + use std::time::Duration; + + use quickwit_common::uri::Uri; + use quickwit_config::StorageTimeoutPolicy; + use tokio::io::AsyncRead; + use tokio::time::Instant; + + use super::*; + use crate::storage::SendableAsync; + use crate::{BulkDeleteError, PutPayload, StorageErrorKind}; + + /// Test backend that returns after a configurable per-call delay (consumed in + /// order). Used to exercise the timeout + retry loop in + /// [`download_slice_with_timeout`]. + #[derive(Debug)] + struct StorageWithDelay { + delays: Mutex>, + } + + impl StorageWithDelay { + fn new(mut delays: Vec) -> StorageWithDelay { + delays.reverse(); + StorageWithDelay { + delays: Mutex::new(delays), + } + } + } + + #[async_trait::async_trait] + impl Storage for StorageWithDelay { + fn uri(&self) -> &Uri { + todo!(); + } + async fn check_connectivity(&self) -> anyhow::Result<()> { + todo!() + } + async fn put(&self, _path: &Path, _payload: Box) -> StorageResult<()> { + todo!(); + } + async fn copy_to( + &self, + _path: &Path, + _output: &mut dyn SendableAsync, + ) -> StorageResult<()> { + todo!(); + } + async fn get_slice(&self, _path: &Path, range: Range) -> StorageResult { + let duration_opt = self.delays.lock().unwrap().pop(); + let Some(delay) = duration_opt else { + return Err( + StorageErrorKind::Internal.with_error(anyhow::anyhow!("internal error")) + ); + }; + tokio::time::sleep(delay).await; + let buf = vec![0u8; range.len()]; + Ok(OwnedBytes::new(buf)) + } + async fn get_slice_stream( + &self, + _path: &Path, + _range: Range, + ) -> StorageResult> { + todo!() + } + async fn get_all(&self, _path: &Path) -> StorageResult { + todo!(); + } + async fn delete(&self, _path: &Path) -> StorageResult<()> { + todo!(); + } + async fn bulk_delete<'a>(&self, _paths: &[&'a Path]) -> Result<(), BulkDeleteError> { + todo!(); + } + async fn exists(&self, _path: &Path) -> StorageResult { + todo!() + } + async fn file_num_bytes(&self, _path: &Path) -> StorageResult { + todo!(); + } + } + + #[tokio::test] + async fn test_download_slice_with_timeout() { + tokio::time::pause(); + let timeout_policy = StorageTimeoutPolicy { + min_throughtput_bytes_per_secs: 100_000, + timeout_millis: 2_000, + max_num_retries: 1, + }; + let path = Path::new("foo/bar"); + { + // Both attempts exceed the 2s timeout -> overall timeout after 2 * 2s. + let now = Instant::now(); + let backend = + StorageWithDelay::new(vec![Duration::from_secs(5), Duration::from_secs(3)]); + let err = download_slice_with_timeout(&backend, path, 10..100, Some(&timeout_policy)) + .await + .unwrap_err(); + assert_eq!(err.kind(), StorageErrorKind::Timeout); + let elapsed = now.elapsed().as_millis(); + assert!(elapsed.abs_diff(2 * 2_000) < 100); + } + { + // First attempt times out, retry succeeds after 1s. + let now = Instant::now(); + let backend = + StorageWithDelay::new(vec![Duration::from_secs(5), Duration::from_secs(1)]); + assert!( + download_slice_with_timeout(&backend, path, 10..100, Some(&timeout_policy)) + .await + .is_ok() + ); + let elapsed = now.elapsed().as_millis(); + assert!(elapsed.abs_diff(2_000 + 1_000) < 100); + } + } +} diff --git a/quickwit/quickwit-storage/src/download_manager/mod.rs b/quickwit/quickwit-storage/src/download_manager/mod.rs new file mode 100644 index 00000000000..2aa12df4feb --- /dev/null +++ b/quickwit/quickwit-storage/src/download_manager/mod.rs @@ -0,0 +1,335 @@ +// 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. + +//! Download manager for the search read path. +//! +//! Historically, fetching a byte range of a split for search traversed a deep +//! stack of `#[async_trait] impl Storage` decorators and tantivy `Directory` +//! wrappers (timeout/retry, fast-field cache, on-disk split cache, request +//! debouncing, bundle-offset translation, the ephemeral byte-range cache and the +//! static hotcache). Every layer allocated a boxed future and added an `.await`, +//! even though only the actual network/disk read performs I/O. +//! +//! The [`SplitDownloadView`] collapses that into: +//! - a **synchronous core** ([`SplitDownloadView::resolve`]) that performs every cache lookup, the +//! bundle logical→physical offset translation and the dedup keying without allocating a single +//! future, and +//! - **one thin async layer** ([`SplitDownloadView::download`]) that performs the actual download +//! (on-disk split cache fill or object-store GET), bounded by the backend's existing semaphore, +//! debounced and retried, and that writes the result back into the caches synchronously on +//! completion. +//! +//! Caching is unified and keyed by the **physical split path joined with the +//! logical file name inside the split**, plus the file-relative byte range. This +//! keeps the long-term `.fast` field cache and the ephemeral byte-range cache +//! cleanly differentiated per file while remaining globally unique (split files +//! are ULID-named). + +mod download; + +use std::ops::Range; +use std::path::{Path, PathBuf}; +use std::sync::Arc; + +use quickwit_config::{CacheConfig, StorageTimeoutPolicy}; + +pub use self::download::fetch_split_footer; +use crate::{ + AsyncDebouncer, BundleStorageFileOffsets, ByteRangeCache, MemorySizedCache, OwnedBytes, + SplitCache, Storage, StorageResult, +}; + +/// Identifies a single download. Used both as the request-deduplication key and +/// as the logical identity of a cached byte range. +/// +/// The key carries the logical file name inside the split (not just the physical +/// split path + absolute range) so that caching is differentiated per file. +#[derive(Clone, Hash, Eq, PartialEq, Debug)] +pub(crate) struct DownloadKey { + /// Physical `.split` file path, e.g. `01HX….split`. + split_path: PathBuf, + /// Logical file inside the split, e.g. a tantivy segment file. Empty for + /// reads that address the physical split directly (e.g. the footer). + file_in_split: PathBuf, + /// Byte range. File-relative for logical reads, absolute for physical reads. + byte_range: Range, +} + +/// Outcome of the synchronous [`SplitDownloadView::resolve`] step. No I/O has +/// happened yet. +pub enum Resolution { + /// Served entirely from a cache tier. Zero futures, return immediately. + Hit(OwnedBytes), + /// Not cached. Carries the plan for the (single) async download. + Miss(DownloadPlan), +} + +/// A pending download produced by a cache miss. Opaque to callers: it is fed +/// straight back into [`SplitDownloadView::download`]. +pub struct DownloadPlan { + file_in_split: PathBuf, + /// File-relative byte range. + byte_range: Range, + is_fast: bool, +} + +/// Process-wide, cross-request pieces of the download path shared by every split +/// search: the long-term fast-field cache, the optional on-disk split cache, and +/// the request debouncer. Held by the searcher context and cloned into a +/// per-split [`SplitDownloadView`]. +#[derive(Clone)] +pub struct SearchDownloadCaches { + /// Long-term `.fast` field cache (cross request). + fast_field_cache: Arc, + /// On-disk full-split cache. `None` if not configured. + split_cache_opt: Option>, + /// Coalesces concurrent identical downloads. Global so concurrent searches + /// over the same split share a single inflight download. + inflight: Arc>>, +} + +impl SearchDownloadCaches { + /// Builds the shared download caches from the fast-field cache config and the + /// optional on-disk split cache. + pub fn new( + fast_field_cache_config: &CacheConfig, + split_cache_opt: Option>, + ) -> SearchDownloadCaches { + let fast_field_cache = Arc::new(MemorySizedCache::from_config( + fast_field_cache_config, + &crate::metrics::FAST_FIELD_CACHE, + )); + SearchDownloadCaches { + fast_field_cache, + split_cache_opt, + inflight: Arc::new(AsyncDebouncer::default()), + } + } +} + +/// Per-split handle over the download path. Built once per split search; the +/// ephemeral byte-range cache and bundle offsets are specific to that split, +/// while the fast-field cache, split cache and debouncer are shared. +pub struct SplitDownloadView { + backend: Arc, + fast_field_cache: Arc, + split_cache_opt: Option>, + inflight: Arc>>, + timeout_policy: Option, + bundle_offsets: Arc, + /// Physical `.split` path, e.g. `01HX….split`. + split_path: PathBuf, + /// Ephemeral, per-request byte-range cache. `None` when the caller does not + /// provide one (e.g. doc fetching), in which case only the long-term and + /// static tiers serve reads. + byte_range_cache: Option, +} + +impl SplitDownloadView { + /// Builds a per-split view from the shared caches and the per-split pieces: + /// the raw backend storage, the optional aggressive-timeout policy, the + /// bundle logical→physical offset map, the physical split path, and an + /// optional ephemeral byte-range cache. + pub fn new( + caches: &SearchDownloadCaches, + backend: Arc, + timeout_policy: Option, + bundle_offsets: Arc, + split_path: PathBuf, + byte_range_cache: Option, + ) -> SplitDownloadView { + SplitDownloadView { + backend, + fast_field_cache: caches.fast_field_cache.clone(), + split_cache_opt: caches.split_cache_opt.clone(), + inflight: caches.inflight.clone(), + timeout_policy, + bundle_offsets, + split_path, + byte_range_cache, + } + } + + /// Resolves a read against the cache tiers, synchronously. Safe to call from + /// tantivy's synchronous `read_bytes`. + /// + /// Lookup order: ephemeral byte-range tier (per request) → long-term + /// fast-field tier (only for `.fast` files). A miss returns a [`DownloadPlan`] + /// without performing any network I/O. + pub fn resolve(&self, file_in_split: &Path, byte_range: Range) -> Resolution { + if byte_range.is_empty() { + return Resolution::Hit(OwnedBytes::empty()); + } + let cache_tag = self.cache_tag(file_in_split); + if let Some(byte_range_cache) = &self.byte_range_cache + && let Some(bytes) = byte_range_cache.get_slice(&cache_tag, byte_range.clone()) + { + return Resolution::Hit(bytes); + } + let is_fast = is_fast_field_file(file_in_split); + if is_fast + && let Some(bytes) = self + .fast_field_cache + .get_slice(&cache_tag, byte_range.clone()) + { + return Resolution::Hit(bytes); + } + Resolution::Miss(DownloadPlan { + file_in_split: file_in_split.to_owned(), + byte_range, + is_fast, + }) + } + + /// Resolves the read, and only awaits if it is a miss. This is the entry + /// point used by warmup (async context). + pub async fn get( + &self, + file_in_split: &Path, + byte_range: Range, + ) -> StorageResult { + match self.resolve(file_in_split, byte_range) { + Resolution::Hit(bytes) => Ok(bytes), + Resolution::Miss(plan) => self.download(plan).await, + } + } + + /// The cache identity of a logical file inside this split: the physical split + /// path joined with the logical file name. Globally unique (split files are + /// ULID-named) while keeping per-file differentiation. + fn cache_tag(&self, file_in_split: &Path) -> PathBuf { + self.split_path.join(file_in_split) + } +} + +/// Returns whether the logical file is a fast-field file, i.e. eligible for the +/// long-term fast-field cache. Mirrors the historical `.fast` suffix routing. +fn is_fast_field_file(file_in_split: &Path) -> bool { + file_in_split.to_string_lossy().ends_with(".fast") +} + +#[cfg(test)] +mod tests { + use std::path::Path; + use std::sync::Arc; + + use bytesize::ByteSize; + + use super::*; + use crate::{RamStorage, RamStorageBuilder}; + + fn bundle_offsets(files: &[(&str, Range)]) -> Arc { + let files = files + .iter() + .map(|(name, range)| (PathBuf::from(name), range.clone())) + .collect(); + Arc::new(BundleStorageFileOffsets { files }) + } + + /// A valid ULID, so the physical split path is recognized by the (unused + /// here) on-disk split cache. + const SPLIT_PATH: &str = "01ARZ3NDEKTSV4RRFFQ69G5FAV.split"; + + fn build_view( + backend: Arc, + byte_range_cache: Option, + offsets: Arc, + ) -> SplitDownloadView { + let caches = SearchDownloadCaches::new(&ByteSize::mb(10).into(), None); + SplitDownloadView::new( + &caches, + backend, + None, + offsets, + PathBuf::from(SPLIT_PATH), + byte_range_cache, + ) + } + + #[tokio::test] + async fn test_resolve_miss_then_download_then_hit() { + // The split physical file contains "0123456789", file "seg.fast" maps to + // bytes [2, 8) i.e. "234567". + let ram_storage = RamStorageBuilder::default() + .put(SPLIT_PATH, b"0123456789") + .build(); + let offsets = bundle_offsets(&[("seg.fast", 2..8)]); + let byte_range_cache = + ByteRangeCache::with_infinite_capacity(&crate::metrics::SHORTLIVED_CACHE); + let view = build_view( + Arc::new(ram_storage), + Some(byte_range_cache.clone()), + offsets, + ); + + // First read is a miss (no cache populated yet). + assert!(matches!( + view.resolve(Path::new("seg.fast"), 0..4), + Resolution::Miss(_) + )); + + // `get` downloads (logical [0,4) -> physical [2,6) -> "2345"). + let bytes = view.get(Path::new("seg.fast"), 0..4).await.unwrap(); + assert_eq!(bytes.as_slice(), b"2345"); + + // The download wrote back into both the ephemeral and the fast-field + // tiers, so a subsequent resolve is a synchronous hit. + match view.resolve(Path::new("seg.fast"), 0..4) { + Resolution::Hit(bytes) => assert_eq!(bytes.as_slice(), b"2345"), + Resolution::Miss(_) => panic!("expected a cache hit after download"), + } + } + + #[tokio::test] + async fn test_non_fast_field_not_in_fast_cache() { + let ram_storage = RamStorageBuilder::default() + .put(SPLIT_PATH, b"0123456789") + .build(); + let offsets = bundle_offsets(&[("seg.idx", 0..10)]); + // No ephemeral cache: only the fast-field (long-term) tier could serve a + // synchronous hit, and a non-`.fast` file must never land there. + let view = build_view(Arc::new(ram_storage), None, offsets); + + let bytes = view.get(Path::new("seg.idx"), 0..3).await.unwrap(); + assert_eq!(bytes.as_slice(), b"012"); + // Still a miss: the read was not cached anywhere we consult synchronously. + assert!(matches!( + view.resolve(Path::new("seg.idx"), 0..3), + Resolution::Miss(_) + )); + } + + #[tokio::test] + async fn test_empty_range_is_hit() { + let ram_storage = RamStorage::default(); + let offsets = bundle_offsets(&[("seg.fast", 0..0)]); + let view = build_view(Arc::new(ram_storage), None, offsets); + match view.resolve(Path::new("seg.fast"), 0..0) { + Resolution::Hit(bytes) => assert!(bytes.is_empty()), + Resolution::Miss(_) => panic!("empty range must resolve to an empty hit"), + } + } + + #[tokio::test] + async fn test_unknown_file_is_not_found() { + let ram_storage = RamStorage::default(); + let offsets = bundle_offsets(&[("seg.fast", 0..0)]); + let view = build_view(Arc::new(ram_storage), None, offsets); + let err = view + .get(Path::new("does_not_exist"), 0..4) + .await + .unwrap_err(); + assert_eq!(err.kind(), crate::StorageErrorKind::NotFound); + } +} diff --git a/quickwit/quickwit-storage/src/lib.rs b/quickwit/quickwit-storage/src/lib.rs index 5f2c6057b5c..a5a7b70a31a 100644 --- a/quickwit/quickwit-storage/src/lib.rs +++ b/quickwit/quickwit-storage/src/lib.rs @@ -28,12 +28,14 @@ mod cache; mod counting_storage; mod debouncer; +mod download_manager; mod file_descriptor_cache; pub mod metrics; mod storage; -mod timeout_and_retry_storage; pub use debouncer::AsyncDebouncer; -pub(crate) use debouncer::DebouncedStorage; +pub use download_manager::{ + DownloadPlan, Resolution, SearchDownloadCaches, SplitDownloadView, fetch_split_footer, +}; pub use self::payload::PutPayload; pub use self::storage::Storage; @@ -63,9 +65,7 @@ pub use versioned_component::VersionedComponent; 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, -}; +pub use self::cache::{ByteRangeCache, MemorySizedCache, StorageCache, wrap_storage_with_cache}; pub use self::counting_storage::{CountingStorage, DownloadCounters}; pub use self::local_file_storage::{LocalFileStorage, LocalFileStorageFactory}; #[cfg(feature = "azure")] @@ -90,7 +90,6 @@ pub use self::test_suite::{ storage_test_multi_part_upload, storage_test_single_part_upload, storage_test_suite, test_write_and_bulk_delete, }; -pub use self::timeout_and_retry_storage::TimeoutAndRetryStorage; pub use crate::error::{ BulkDeleteError, DeleteFailure, StorageError, StorageErrorKind, StorageResolverError, StorageResult, diff --git a/quickwit/quickwit-storage/src/local_file_storage.rs b/quickwit/quickwit-storage/src/local_file_storage.rs index c0e36b5ba4f..25b263f1fa3 100644 --- a/quickwit/quickwit-storage/src/local_file_storage.rs +++ b/quickwit/quickwit-storage/src/local_file_storage.rs @@ -31,8 +31,8 @@ use tracing::warn; use crate::metrics::object_storage_get_slice_in_flight_guards; use crate::storage::SendableAsync; use crate::{ - BulkDeleteError, DebouncedStorage, DeleteFailure, OwnedBytes, Storage, StorageError, - StorageErrorKind, StorageFactory, StorageResolverError, StorageResult, + BulkDeleteError, DeleteFailure, OwnedBytes, Storage, StorageError, StorageErrorKind, + StorageFactory, StorageResolverError, StorageResult, }; /// File system compatible storage implementation. @@ -376,7 +376,7 @@ impl StorageFactory for LocalFileStorageFactory { async fn resolve(&self, uri: &Uri) -> Result, StorageResolverError> { let storage = LocalFileStorage::from_uri(uri)?; - Ok(Arc::new(DebouncedStorage::new(storage))) + Ok(Arc::new(storage)) } } diff --git a/quickwit/quickwit-storage/src/object_storage/azure_blob_storage.rs b/quickwit/quickwit-storage/src/object_storage/azure_blob_storage.rs index 42e08591699..c302ce1af47 100644 --- a/quickwit/quickwit-storage/src/object_storage/azure_blob_storage.rs +++ b/quickwit/quickwit-storage/src/object_storage/azure_blob_storage.rs @@ -43,7 +43,6 @@ use tokio_util::compat::FuturesAsyncReadCompatExt; use tokio_util::io::StreamReader; use tracing::{instrument, warn}; -use crate::debouncer::DebouncedStorage; use crate::metrics::object_storage_get_slice_in_flight_guards; use crate::stable_deref_bytes::into_owned_bytes; use crate::storage::SendableAsync; @@ -72,7 +71,7 @@ impl StorageFactory for AzureBlobStorageFactory { async fn resolve(&self, uri: &Uri) -> Result, StorageResolverError> { let storage = AzureBlobStorage::from_uri(&self.storage_config, uri)?; - Ok(Arc::new(DebouncedStorage::new(storage))) + Ok(Arc::new(storage)) } } diff --git a/quickwit/quickwit-storage/src/object_storage/s3_compatible_storage_resolver.rs b/quickwit/quickwit-storage/src/object_storage/s3_compatible_storage_resolver.rs index 92b4406486b..09b5077564c 100644 --- a/quickwit/quickwit-storage/src/object_storage/s3_compatible_storage_resolver.rs +++ b/quickwit/quickwit-storage/src/object_storage/s3_compatible_storage_resolver.rs @@ -21,9 +21,7 @@ use quickwit_config::{S3StorageConfig, StorageBackend}; use tokio::sync::OnceCell; use super::s3_compatible_storage::create_s3_client; -use crate::{ - DebouncedStorage, S3CompatibleObjectStorage, Storage, StorageFactory, StorageResolverError, -}; +use crate::{S3CompatibleObjectStorage, Storage, StorageFactory, StorageResolverError}; /// S3 compatible object storage resolver. pub struct S3CompatibleObjectStorageFactory { @@ -61,6 +59,8 @@ impl StorageFactory for S3CompatibleObjectStorageFactory { let storage = S3CompatibleObjectStorage::from_uri_and_client(&self.storage_config, uri, s3_client) .await?; - Ok(Arc::new(DebouncedStorage::new(storage))) + // Request debouncing now lives in the search download manager, so the + // resolver hands out the raw backend (see analysis.md §2.5). + Ok(Arc::new(storage)) } } diff --git a/quickwit/quickwit-storage/src/opendal_storage/google_cloud_storage.rs b/quickwit/quickwit-storage/src/opendal_storage/google_cloud_storage.rs index 23397e90a7c..6f961481c5b 100644 --- a/quickwit/quickwit-storage/src/opendal_storage/google_cloud_storage.rs +++ b/quickwit/quickwit-storage/src/opendal_storage/google_cloud_storage.rs @@ -22,7 +22,6 @@ use regex::Regex; use tracing::info; use super::OpendalStorage; -use crate::debouncer::DebouncedStorage; use crate::{Storage, StorageFactory, StorageResolverError}; /// Google cloud storage resolver. @@ -45,7 +44,7 @@ impl StorageFactory for GoogleCloudStorageFactory { async fn resolve(&self, uri: &Uri) -> Result, StorageResolverError> { let storage = from_uri(&self.storage_config, uri)?; - Ok(Arc::new(DebouncedStorage::new(storage))) + Ok(Arc::new(storage)) } } diff --git a/quickwit/quickwit-storage/src/split_cache/mod.rs b/quickwit/quickwit-storage/src/split_cache/mod.rs index 1c0de124d73..d7711250140 100644 --- a/quickwit/quickwit-storage/src/split_cache/mod.rs +++ b/quickwit/quickwit-storage/src/split_cache/mod.rs @@ -152,7 +152,11 @@ impl SplitCache { // 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: Ulid, storage_uri: &Uri) -> Option { + pub(crate) async fn get_split_file( + &self, + split_id: Ulid, + storage_uri: &Uri, + ) -> Option { // We touch before even checking the fd cache in order to update the file's last access time // for the file cache. let num_bytes_opt: Option = self @@ -186,7 +190,7 @@ fn delete_evicted_splits(root_path: &Path, splits_to_delete: &[Ulid]) { } } -fn split_id_from_path(split_path: &Path) -> Option { +pub(crate) fn split_id_from_path(split_path: &Path) -> Option { let split_filename = split_path.file_name()?.to_str()?; let split_id_str = split_filename.strip_suffix(".split")?; Ulid::from_str(split_id_str).ok() diff --git a/quickwit/quickwit-storage/src/timeout_and_retry_storage.rs b/quickwit/quickwit-storage/src/timeout_and_retry_storage.rs deleted file mode 100644 index d42aca9d55b..00000000000 --- a/quickwit/quickwit-storage/src/timeout_and_retry_storage.rs +++ /dev/null @@ -1,279 +0,0 @@ -// 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::ops::Range; -use std::path::Path; -use std::sync::Arc; - -use async_trait::async_trait; -use quickwit_common::uri::Uri; -use quickwit_common::{rate_limited_info, rate_limited_warn}; -use quickwit_config::StorageTimeoutPolicy; -use tantivy::directory::OwnedBytes; -use tokio::io::AsyncRead; - -use crate::storage::SendableAsync; -use crate::{BulkDeleteError, PutPayload, Storage, StorageErrorKind, StorageResult}; - -/// Storage proxy that implements a retry operation if the underlying storage -/// takes too long. -/// -/// This is useful in order to ensure a low latency on S3. -/// Retrying agressively is recommended for S3. -/// -/// -#[derive(Clone, Debug)] -pub struct TimeoutAndRetryStorage { - underlying: Arc, - storage_timeout_policy: StorageTimeoutPolicy, -} - -impl TimeoutAndRetryStorage { - /// Creates a new `TimeoutAndRetryStorage`. - /// - /// See [StorageTimeoutPolicy] for more information. - pub fn new(storage: Arc, storage_timeout_policy: StorageTimeoutPolicy) -> Self { - TimeoutAndRetryStorage { - underlying: storage, - storage_timeout_policy, - } - } -} - -#[async_trait] -impl Storage for TimeoutAndRetryStorage { - async fn check_connectivity(&self) -> anyhow::Result<()> { - self.underlying.check_connectivity().await - } - - async fn put(&self, path: &Path, payload: Box) -> StorageResult<()> { - self.underlying.put(path, payload).await - } - - fn copy_to<'life0, 'life1, 'life2, 'async_trait>( - &'life0 self, - path: &'life1 Path, - output: &'life2 mut dyn SendableAsync, - ) -> ::core::pin::Pin< - Box< - dyn ::core::future::Future> - + ::core::marker::Send - + 'async_trait, - >, - > - where - 'life0: 'async_trait, - 'life1: 'async_trait, - 'life2: 'async_trait, - Self: 'async_trait, - { - self.underlying.copy_to(path, output) - } - - async fn copy_to_file(&self, path: &Path, output_path: &Path) -> StorageResult { - self.underlying.copy_to_file(path, output_path).await - } - - /// Downloads a slice of a file from the storage, and returns an in memory buffer - async fn get_slice(&self, path: &Path, range: Range) -> StorageResult { - let num_bytes = range.len(); - for (attempt_id, timeout_duration) in self - .storage_timeout_policy - .compute_timeout(num_bytes) - .enumerate() - { - let get_slice_fut = self.underlying.get_slice(path, range.clone()); - // TODO test avoid aborting timed out requests. #5468 - match tokio::time::timeout(timeout_duration, get_slice_fut).await { - Ok(result) => { - match attempt_id { - 0 => crate::metrics::GET_SLICE_TIMEOUT_SUCCESS_AFTER_0_TIMEOUT.inc(), - 1 => crate::metrics::GET_SLICE_TIMEOUT_SUCCESS_AFTER_1_TIMEOUT.inc(), - _ => crate::metrics::GET_SLICE_TIMEOUT_SUCCESS_AFTER_2_PLUS_TIMEOUT.inc(), - } - return result; - } - Err(_elapsed) => { - rate_limited_info!(limit_per_min=60, num_bytes=num_bytes, path=%path.display(), timeout_secs=timeout_duration.as_secs_f32(), "get timeout elapsed"); - continue; - } - } - } - rate_limited_warn!(limit_per_min=60, num_bytes=num_bytes, path=%path.display(), "all get_slice attempts timeouted"); - crate::metrics::GET_SLICE_TIMEOUT_ALL_TIMEOUTS.inc(); - return Err( - StorageErrorKind::Timeout.with_error(anyhow::anyhow!("internal timeout on get_slice")) - ); - } - - async fn get_slice_stream( - &self, - path: &Path, - range: Range, - ) -> StorageResult> { - self.underlying.get_slice_stream(path, range).await - } - - async fn get_all(&self, path: &Path) -> StorageResult { - self.underlying.get_all(path).await - } - - async fn delete(&self, path: &Path) -> StorageResult<()> { - self.underlying.delete(path).await - } - - async fn bulk_delete<'a>(&self, paths: &[&'a Path]) -> Result<(), BulkDeleteError> { - self.underlying.bulk_delete(paths).await - } - - async fn exists(&self, path: &Path) -> StorageResult { - self.underlying.exists(path).await - } - - async fn file_num_bytes(&self, path: &Path) -> StorageResult { - self.underlying.file_num_bytes(path).await - } - - fn uri(&self) -> &Uri { - self.underlying.uri() - } -} - -#[cfg(test)] -mod tests { - - use std::sync::Mutex; - use std::time::Duration; - - use tokio::time::Instant; - - use super::*; - - #[derive(Debug)] - struct StorageWithDelay { - delays: Mutex>, - } - - impl StorageWithDelay { - pub fn new(mut delays: Vec) -> StorageWithDelay { - delays.reverse(); - StorageWithDelay { - delays: Mutex::new(delays), - } - } - } - - #[async_trait] - impl Storage for StorageWithDelay { - fn uri(&self) -> &Uri { - todo!(); - } - - async fn check_connectivity(&self) -> anyhow::Result<()> { - todo!() - } - async fn put(&self, _path: &Path, _payload: Box) -> StorageResult<()> { - todo!(); - } - fn copy_to<'life0, 'life1, 'life2, 'async_trait>( - &'life0 self, - _path: &'life1 Path, - _output: &'life2 mut dyn SendableAsync, - ) -> ::core::pin::Pin< - Box< - dyn ::core::future::Future> - + ::core::marker::Send - + 'async_trait, - >, - > - where - 'life0: 'async_trait, - 'life1: 'async_trait, - 'life2: 'async_trait, - Self: 'async_trait, - { - todo!(); - } - - async fn get_slice(&self, _path: &Path, range: Range) -> StorageResult { - let duration_opt = self.delays.lock().unwrap().pop(); - let Some(delay) = duration_opt else { - return Err( - StorageErrorKind::Internal.with_error(anyhow::anyhow!("internal error")) - ); - }; - tokio::time::sleep(delay).await; - let buf = vec![0u8; range.len()]; - Ok(OwnedBytes::new(buf)) - } - async fn get_slice_stream( - &self, - _path: &Path, - _range: Range, - ) -> StorageResult> { - todo!() - } - async fn get_all(&self, _path: &Path) -> StorageResult { - todo!(); - } - async fn delete(&self, _path: &Path) -> StorageResult<()> { - todo!(); - } - async fn bulk_delete<'a>(&self, _paths: &[&'a Path]) -> Result<(), BulkDeleteError> { - todo!(); - } - async fn exists(&self, _path: &Path) -> StorageResult { - todo!() - } - async fn file_num_bytes(&self, _path: &Path) -> StorageResult { - todo!(); - } - } - - #[tokio::test] - async fn test_timeout_and_retry_storage() { - tokio::time::pause(); - - let timeout_policy = StorageTimeoutPolicy { - min_throughtput_bytes_per_secs: 100_000, - timeout_millis: 2_000, - max_num_retries: 1, - }; - - let path = Path::new("foo/bar"); - - { - let now = Instant::now(); - let storage_with_delay = - StorageWithDelay::new(vec![Duration::from_secs(5), Duration::from_secs(3)]); - let storage = - TimeoutAndRetryStorage::new(Arc::new(storage_with_delay), timeout_policy.clone()); - assert_eq!( - storage.get_slice(path, 10..100).await.unwrap_err().kind, - StorageErrorKind::Timeout - ); - let elapsed = now.elapsed().as_millis(); - assert!(elapsed.abs_diff(2 * 2_000) < 100); - } - { - let now = Instant::now(); - let storage_with_delay = - StorageWithDelay::new(vec![Duration::from_secs(5), Duration::from_secs(1)]); - let storage = TimeoutAndRetryStorage::new(Arc::new(storage_with_delay), timeout_policy); - assert!(storage.get_slice(path, 10..100).await.is_ok(),); - let elapsed = now.elapsed().as_millis(); - assert!(elapsed.abs_diff(2_000 + 1_000) < 100); - } - } -}