QuantizedStorage::for_each_run: pipeline run reads on async backends

With `prefers_contiguous_reads()` true for io_uring, every batch goes
through `for_each_run`, which read each run synchronously: a scattered
id list (HNSW neighbours) waited on one disk read per vector, where
`for_each_in_batch` kept the whole batch in flight through `read_batch`.
Submit all runs of a batch together, still one read per run, so
scattered reads stay pipelined while a scan still reads each run in one
request.  Backends without async reads keep the sequential loop.

`turbo_vector_search` (dim 1024, 200k vectors, 4096 shuffled ids per
iteration) against dev: cold scattered io_uring 187 ms -> 26.5 ms
(dev 29.5 ms); the warm scan keeps 198 ms -> 21 ms.  Warm scattered
lands at 5.06 ms (dev 4.67 ms), giving up the 3.68 ms of synchronous
reads, which only holds with the data already in the page cache.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VRiRaiPsm5CEBgpQ7VAQab

lib/segment/src/vector_storage/quantized/quantized_storage.rs

@@ -345,6 +345,30 @@ impl<S: UniversalRead> quantization::EncodedStorage for QuantizedStorage<S> {
         mut callback: impl FnMut(usize, usize, Cow<'_, [u8]>),
     ) {
         let size = self.quantized_vector_size.get() as u64;
+
+        if ReadOnly::<S>::kind().can_be_async() {
+            // Pipelined backends (io_uring): submit every run of the batch
+            // together — still one read per run — so scattered ids keep their
+            // reads in flight concurrently instead of waiting on each run
+            // before submitting the next.
+            let mut runs = Vec::new();
+            quantization::encoded_storage::for_each_consecutive_run(
+                offsets,
+                |first, start, len| {
+                    let range = ReadRange::new(size * u64::from(start), size * len as u64);
+                    runs.push(((first, len), range));
+                },
+            );
+            // Access pattern does not matter for io_uring.
+            self.storage
+                .read_batch(runs, Random, |(first, len), bytes: &[u8]| {
+                    callback(first, len, Cow::Borrowed(bytes));
+                    UioResult::Ok(())
+                })
+                .expect("vectors read from quantized storage failed");
+            return;
+        }
+
         quantization::encoded_storage::for_each_consecutive_run(offsets, |first, start, len| {
             let range = ReadRange::new(size * u64::from(start), size * len as u64);
             let bytes = if len > 1 {