turbo_persistence/
meta_file.rs

1use std::{
2    cmp::Ordering,
3    fmt::Display,
4    mem::take,
5    ops::Deref,
6    path::{Path, PathBuf},
7    sync::{Arc, OnceLock},
8};
9
10use anyhow::{Context, Result, bail, ensure};
11use bitfield::bitfield;
12use byteorder::{BE, ReadBytesExt};
13#[cfg(feature = "mmap")]
14use fs_err::File;
15#[cfg(feature = "mmap")]
16use memmap2::{Mmap, MmapOptions};
17use smallvec::SmallVec;
18use zerocopy::{FromBytes, Immutable, IntoBytes, KnownLayout, Ref, big_endian as be};
19
20#[cfg(feature = "mmap")]
21use crate::mmap_helper::advise_mmap_for_persistence;
22use crate::{
23    AccessMode, Compression, FamilyConfig, QueryKey,
24    lookup_entry::LookupValue,
25    static_sorted_file::{BlockCache, SstLookupResult, StaticSortedFile, StaticSortedFileMetaData},
26};
27
28bitfield! {
29    #[derive(Clone, Copy, Default)]
30    pub struct MetaEntryFlags(u32);
31    impl Debug;
32    impl From<u32>;
33    /// The SST file was compacted and none of the entries have been accessed recently.
34    pub cold, set_cold: 0;
35    /// The SST file was freshly written and has not been compacted yet.
36    pub fresh, set_fresh: 1;
37}
38
39impl MetaEntryFlags {
40    pub const FRESH: MetaEntryFlags = MetaEntryFlags(0b10);
41    pub const COLD: MetaEntryFlags = MetaEntryFlags(0b01);
42    pub const WARM: MetaEntryFlags = MetaEntryFlags(0b00);
43}
44
45impl Display for MetaEntryFlags {
46    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
47        if self.fresh() {
48            f.pad_integral(true, "", "fresh")
49        } else if self.cold() {
50            f.pad_integral(true, "", "cold")
51        } else {
52            f.pad_integral(true, "", "warm")
53        }
54    }
55}
56
57/// Magic number identifying a `.meta` file.
58pub(crate) const META_FILE_MAGIC: u32 = 0xFE4ADA4A;
59
60/// On-disk layout of a single entry header in the `.meta` file.
61///
62/// Fields are big-endian to match the existing wire format written by [`MetaFileBuilder`].
63#[repr(C, packed)]
64#[derive(FromBytes, IntoBytes, Immutable, KnownLayout, Clone, Copy)]
65pub(crate) struct EntryHeader {
66    sequence_number: be::U32,
67    block_count: be::U16,
68    min_hash: be::U64,
69    max_hash: be::U64,
70    size: be::U64,
71    flags: be::U32,
72    amqf_end_offset: be::U32,
73}
74
75impl EntryHeader {
76    pub(crate) fn new(
77        sequence_number: u32,
78        block_count: u16,
79        min_hash: u64,
80        max_hash: u64,
81        size: u64,
82        flags: MetaEntryFlags,
83        amqf_end_offset: u32,
84    ) -> Self {
85        Self {
86            sequence_number: be::U32::new(sequence_number),
87            block_count: be::U16::new(block_count),
88            min_hash: be::U64::new(min_hash),
89            max_hash: be::U64::new(max_hash),
90            size: be::U64::new(size),
91            flags: be::U32::new(flags.0),
92            amqf_end_offset: be::U32::new(amqf_end_offset),
93        }
94    }
95}
96
97/// # Safety
98///
99/// `MetaEntry` stores a `FilterRef<'static>` with a transmuted lifetime that actually borrows
100/// from the parent [`MetaFile`]'s stable backing bytes. This is safe as long as an entry never
101/// outlives that backing: entries are only handed out by reference, and the one place that moves
102/// them ([`MetaFile::retain_entries`]) keeps them inside the same `MetaFile`.
103///
104/// For this reason this type should not implement Clone or Copy — a copy could outlive the
105/// `MetaFile` that owns the backing it points into.
106pub struct MetaEntry {
107    /// The metadata for the static sorted file.
108    sst_data: StaticSortedFileMetaData,
109    /// The size of the SST file in bytes.
110    size: u64,
111    /// The status flags for this entry.
112    flags: MetaEntryFlags,
113    /// Byte offset range of the raw AMQF data within the backing, used for carrying forward
114    /// serialized bytes during compaction without re-serializing.
115    amqf_data_offset: std::ops::Range<u32>,
116    /// The AMQF filter for this file, eagerly deserialized as a zero-copy [`qfilter::FilterRef`]
117    /// that borrows directly from the parent [`MetaFile`]'s memory-mapped file.
118    ///
119    /// The `'static` lifetime is transmuted — the actual borrow is from `MetaFile::backing`.
120    amqf: qfilter::FilterRef<'static>,
121    /// Compression recorded in this entry's meta file.
122    compression: Compression,
123    /// The static sorted file that is lazily loaded
124    sst: OnceLock<StaticSortedFile>,
125}
126
127// Safety: FilterRef is a read-only view into stable backing bytes which are Send+Sync.
128unsafe impl Send for MetaEntry {}
129unsafe impl Sync for MetaEntry {}
130
131impl MetaEntry {
132    pub fn sequence_number(&self) -> u32 {
133        self.sst_data.sequence_number
134    }
135
136    pub fn size(&self) -> u64 {
137        self.size
138    }
139
140    pub fn flags(&self) -> MetaEntryFlags {
141        self.flags
142    }
143
144    pub fn amqf_size(&self) -> u32 {
145        self.amqf_data_offset.end - self.amqf_data_offset.start
146    }
147
148    pub fn amqf(&self) -> &qfilter::FilterRef<'static> {
149        &self.amqf
150    }
151
152    /// Returns the raw serialized AMQF bytes from the stable backing.
153    pub fn raw_amqf<'l>(&self, amqf_data: &'l [u8]) -> &'l [u8] {
154        &amqf_data[self.amqf_data_offset.start as usize..self.amqf_data_offset.end as usize]
155    }
156
157    fn sst(&self, meta: &MetaFile) -> Result<&StaticSortedFile> {
158        self.sst.get_or_try_init(|| {
159            StaticSortedFile::open(
160                &meta.db_path,
161                self.sst_data,
162                self.compression,
163                meta.access_mode,
164            )
165            .with_context(|| {
166                format!(
167                    "Unable to open static sorted file referenced from {:08}.meta",
168                    meta.sequence_number()
169                )
170            })
171        })
172    }
173
174    pub fn block_count(&self) -> u16 {
175        self.sst_data.block_count
176    }
177
178    /// Returns the SST metadata needed to open the file independently.
179    /// Used during compaction to avoid caching mmaps on the MetaEntry.
180    pub fn sst_metadata(&self) -> StaticSortedFileMetaData {
181        self.sst_data
182    }
183}
184
185/// The result of a lookup operation.
186pub enum MetaLookupResult {
187    /// The key was not found because it is from a different key family.
188    FamilyMiss,
189    /// The key was not found because it is out of the range of this SST file. But it was the
190    /// correct key family.
191    RangeMiss,
192    /// The key was not found because it was not in the AMQF filter. But it was in the range.
193    QuickFilterMiss,
194    /// The key was looked up in the SST file. It was in the AMQF filter.
195    SstLookup(SstLookupResult),
196}
197
198/// The result of a batch lookup operation.
199#[derive(Default)]
200pub struct MetaBatchLookupResult {
201    /// The key was not found because it is from a different key family.
202    #[cfg(feature = "stats")]
203    pub family_miss: bool,
204    /// The key was not found because it is out of the range of this SST file. But it was the
205    /// correct key family.
206    #[cfg(feature = "stats")]
207    pub range_misses: usize,
208    /// The key was not found because it was not in the AMQF filter. But it was in the range.
209    #[cfg(feature = "stats")]
210    pub quick_filter_misses: usize,
211    /// The key was unsuccessfully looked up in the SST file. It was in the AMQF filter.
212    #[cfg(feature = "stats")]
213    pub sst_misses: usize,
214    /// The key was found in the SST file.
215    #[cfg(feature = "stats")]
216    pub hits: usize,
217}
218
219/// The key family and hash range of an SST file.
220#[derive(Clone, Copy)]
221pub struct StaticSortedFileRange {
222    pub min_hash: u64,
223    pub max_hash: u64,
224}
225
226impl StaticSortedFileRange {
227    /// Whether `hash` falls within this file's span. A lookup can skip the file entirely if not.
228    #[inline(always)]
229    pub fn contains(&self, hash: u64) -> bool {
230        hash >= self.min_hash && hash <= self.max_hash
231    }
232}
233
234enum MetaFileBacking {
235    #[cfg(feature = "mmap")]
236    Mmap(Mmap),
237    /// Heap bytes for [`AccessMode::File`].
238    ///
239    /// This is an `Arc<[u8]>` rather than a `Box<[u8]>` so that moving the backing into
240    /// [`MetaFile`] does not reborrow the bytes: a `Box` is a unique pointer, so the move
241    /// invalidates the `FilterRef`s that already borrow from it, which Miri reports as undefined
242    /// behavior under Stacked Borrows. An `Arc` moves its handle without retagging the allocation.
243    Bytes(Arc<[u8]>),
244}
245
246impl Deref for MetaFileBacking {
247    type Target = [u8];
248
249    fn deref(&self) -> &Self::Target {
250        match self {
251            #[cfg(feature = "mmap")]
252            MetaFileBacking::Mmap(mmap) => mmap,
253            MetaFileBacking::Bytes(bytes) => bytes,
254        }
255    }
256}
257
258/// # Safety
259///
260/// `entries` must be declared before `backing` so every borrowed `FilterRef` is dropped before
261/// its stable mmap or heap storage.
262pub struct MetaFile {
263    /// The database path
264    db_path: PathBuf,
265    /// The sequence number of this file.
266    sequence_number: u32,
267    /// The key family of the SST files in this meta file.
268    family: u32,
269    /// Compression recorded for this family.
270    compression: Compression,
271    /// Stored separately from [`MetaEntry`] so that lookups can operate over a denser data
272    /// structure that's hotter in cache.
273    hash_ranges: Box<[StaticSortedFileRange]>,
274    /// The entries of the file. Dropped before `backing` (field declaration order).
275    entries: Box<[MetaEntry]>,
276    /// The entries that have been marked as obsolete.
277    obsolete_entries: Vec<u32>,
278    /// The obsolete SST files.
279    obsolete_sst_files: Vec<u32>,
280    /// Byte offset within the backing where the AMQF data region starts.
281    /// Entry AMQF offsets and used-keys offsets are relative to this position.
282    amqf_data_start: u32,
283    /// The offset of the start of the "used keys" AMQF data relative to the AMQF data region.
284    start_of_used_keys_amqf_data_offset: u32,
285    /// The offset of the end of the "used keys" AMQF data relative to the AMQF data region.
286    end_of_used_keys_amqf_data_offset: u32,
287    /// The access mode inherited by referenced SST files.
288    access_mode: AccessMode,
289    /// Stable bytes backing the parsed filters. Must be declared after `entries`.
290    backing: MetaFileBacking,
291}
292
293impl MetaFile {
294    /// Opens a meta file using mmap or stable heap bytes according to `access_mode`.
295    pub fn open(
296        db_path: &Path,
297        sequence_number: u32,
298        family_configs: Option<&[FamilyConfig]>,
299        access_mode: AccessMode,
300    ) -> Result<Self> {
301        let filename = format!("{sequence_number:08}.meta");
302        let path = db_path.join(&filename);
303        Self::open_internal(
304            db_path.to_path_buf(),
305            sequence_number,
306            &path,
307            family_configs,
308            access_mode,
309        )
310        .with_context(|| format!("Unable to open meta file {filename}"))
311    }
312
313    fn open_internal(
314        db_path: PathBuf,
315        sequence_number: u32,
316        path: &Path,
317        family_configs: Option<&[FamilyConfig]>,
318        access_mode: AccessMode,
319    ) -> Result<Self> {
320        let backing = match access_mode {
321            #[cfg(feature = "mmap")]
322            AccessMode::Mmap => {
323                let file = File::open(path)?;
324                let mmap = unsafe { MmapOptions::new().map(file.file()) }
325                    .context("Failed to mmap meta file")?;
326                #[cfg(unix)]
327                mmap.advise(memmap2::Advice::Random)
328                    .context("Failed to advise mmap")?;
329                advise_mmap_for_persistence(&mmap)?;
330                MetaFileBacking::Mmap(mmap)
331            }
332            AccessMode::File => MetaFileBacking::Bytes(fs_err::read(path)?.into()),
333        };
334        // Parse the header from stable backing bytes via ReadBytesExt on &[u8].
335        let mut reader: &[u8] = &backing;
336        let magic = reader.read_u32::<BE>()?;
337        if magic != META_FILE_MAGIC {
338            bail!("Invalid magic number");
339        }
340        let family = reader.read_u32::<BE>()?;
341        let compression = match reader.read_u8()? {
342            value if value == Compression::Lz4 as u8 => Compression::Lz4,
343            value if value == Compression::Zstd3 as u8 => Compression::Zstd3,
344            value => bail!("Invalid compression algorithm {value}"),
345        };
346        if let Some(configs) = family_configs {
347            let configured = configs
348                .get(family as usize)
349                .with_context(|| format!("No configuration for family {family}"))?
350                .compression;
351            ensure!(
352                compression == configured,
353                "Compression configuration mismatch for family {family}: meta file uses \
354                 {compression:?}, runtime config uses {configured:?}"
355            );
356        }
357        let obsolete_count = reader.read_u32::<BE>()?;
358        let mut obsolete_sst_files = Vec::with_capacity(obsolete_count as usize);
359        for _ in 0..obsolete_count {
360            obsolete_sst_files.push(reader.read_u32::<BE>()?);
361        }
362
363        let count = reader.read_u32::<BE>()?;
364
365        // Compute where the AMQF data region starts so we can deserialize filters inline.
366        // Remaining header: count * ENTRY_HEADER_SIZE + used_keys_end_offset.
367        let header_so_far = (backing.len() - reader.len()) as u32;
368        let amqf_data_start =
369            header_so_far + count * (size_of::<EntryHeader>() as u32) + size_of::<u32>() as u32;
370        let amqf_data = &backing[amqf_data_start as usize..];
371
372        // Parse entries and eagerly deserialize AMQF filters as zero-copy FilterRefs.
373        let mut entries = Vec::with_capacity(count as usize);
374        let mut hash_ranges = Vec::with_capacity(count as usize);
375        let mut start_of_amqf_data_offset: u32 = 0;
376        for _ in 0..count {
377            let (header, rest): (Ref<&[u8], EntryHeader>, _) = Ref::from_prefix(reader)
378                .ok()
379                .context("Entry header out of bounds")?;
380            reader = rest;
381            let sst_data = StaticSortedFileMetaData {
382                sequence_number: header.sequence_number.get(),
383                block_count: header.block_count.get(),
384            };
385            let min_hash = header.min_hash.get();
386            let max_hash = header.max_hash.get();
387            let size = header.size.get();
388            let flags = MetaEntryFlags(header.flags.get());
389            let end_of_amqf_data_offset = header.amqf_end_offset.get();
390
391            let amqf_bytes = amqf_data
392                .get(start_of_amqf_data_offset as usize..end_of_amqf_data_offset as usize)
393                .expect("AMQF data out of bounds");
394            // Deserialize the filter borrowing from the stable backing, then erase the lifetime.
395            let amqf: qfilter::FilterRef<'_> =
396                postcard::from_bytes(amqf_bytes).with_context(|| {
397                    format!(
398                        "Failed to deserialize AMQF from {:08}.meta for {:08}.sst",
399                        sequence_number, sst_data.sequence_number
400                    )
401                })?;
402            // Safety: the backing is kept alive by MetaFile and is dropped after entries (field
403            // declaration order), so the borrow remains valid for the lifetime of the MetaEntry.
404            let amqf: qfilter::FilterRef<'static> = unsafe { std::mem::transmute(amqf) };
405
406            hash_ranges.push(StaticSortedFileRange { min_hash, max_hash });
407            entries.push(MetaEntry {
408                sst_data,
409                size,
410                flags,
411                amqf_data_offset: start_of_amqf_data_offset..end_of_amqf_data_offset,
412                amqf,
413                compression,
414                sst: OnceLock::new(),
415            });
416            start_of_amqf_data_offset = end_of_amqf_data_offset;
417        }
418
419        let start_of_used_keys_amqf_data_offset = start_of_amqf_data_offset;
420        let end_of_used_keys_amqf_data_offset = reader.read_u32::<BE>()?;
421
422        Ok(Self {
423            db_path,
424            sequence_number,
425            family,
426            compression,
427            hash_ranges: hash_ranges.into_boxed_slice(),
428            entries: entries.into_boxed_slice(),
429            obsolete_entries: Vec::new(),
430            obsolete_sst_files,
431            amqf_data_start,
432            start_of_used_keys_amqf_data_offset,
433            end_of_used_keys_amqf_data_offset,
434            access_mode,
435            backing,
436        })
437    }
438
439    pub fn clear_cache(&mut self) {
440        for entry in self.entries.iter_mut() {
441            entry.sst.take();
442        }
443    }
444
445    pub fn prepare_sst_cache(&self) {
446        for entry in self.entries.iter() {
447            let _ = entry.sst(self);
448        }
449    }
450
451    pub fn sequence_number(&self) -> u32 {
452        self.sequence_number
453    }
454
455    pub fn family(&self) -> u32 {
456        self.family
457    }
458
459    pub fn compression(&self) -> Compression {
460        self.compression
461    }
462
463    /// The on-disk size of this meta file in bytes (the length of its memory map).
464    pub fn byte_size(&self) -> u64 {
465        self.backing.len() as u64
466    }
467
468    pub fn entries(&self) -> &[MetaEntry] {
469        &self.entries
470    }
471
472    /// The hash ranges of this file's entries, in the same order as [`Self::entries`].
473    pub fn hash_ranges(&self) -> &[StaticSortedFileRange] {
474        &self.hash_ranges
475    }
476
477    /// The hash range of the entry at `index`.
478    pub fn hash_range(&self, index: u32) -> StaticSortedFileRange {
479        self.hash_ranges[index as usize]
480    }
481
482    /// The key family and hash range of the entry at `index`.
483    pub fn range(&self, index: u32) -> StaticSortedFileRange {
484        self.hash_range(index)
485    }
486
487    pub fn entry(&self, index: u32) -> &MetaEntry {
488        let index = index as usize;
489        &self.entries[index]
490    }
491
492    pub fn amqf_data(&self) -> &[u8] {
493        &self.backing[self.amqf_data_start as usize..]
494    }
495
496    pub fn deserialize_used_key_hashes_amqf(&self) -> Result<Option<qfilter::FilterRef<'_>>> {
497        if self.start_of_used_keys_amqf_data_offset == self.end_of_used_keys_amqf_data_offset {
498            return Ok(None);
499        }
500        let amqf = &self.amqf_data()[self.start_of_used_keys_amqf_data_offset as usize
501            ..self.end_of_used_keys_amqf_data_offset as usize];
502        Ok(Some(postcard::from_bytes(amqf).with_context(|| {
503            format!(
504                "Failed to deserialize used key hashes AMQF from {:08}.meta",
505                self.sequence_number
506            )
507        })?))
508    }
509
510    pub fn retain_entries(&mut self, mut predicate: impl FnMut(u32) -> bool) -> bool {
511        debug_assert_eq!(
512            self.entries.len(),
513            self.hash_ranges.len(),
514            "hash_ranges must stay parallel to entries"
515        );
516        let old_len = self.entries.len();
517        // Filter the two vectors as pairs so they cannot drift apart. Retaining them separately
518        // would leave a lookup indexing one by a position that means something else in the other.
519        //
520        // This rebuilds both vectors rather than compacting in place, which is the more expensive
521        // shape but a fine trade here: the callers are commit and compaction, never a lookup.
522        //
523        // Entries move between slots but never leave this `MetaFile`, so the `FilterRef`s they
524        // hold keep borrowing a mmap that is neither touched nor dropped.
525        let obsolete = &mut self.obsolete_entries;
526        let (entries, hash_ranges): (Vec<_>, Vec<_>) = take(&mut self.entries)
527            .into_iter()
528            .zip(take(&mut self.hash_ranges))
529            .filter(|(entry, _)| {
530                let retain = predicate(entry.sst_data.sequence_number);
531                if !retain {
532                    obsolete.push(entry.sst_data.sequence_number);
533                }
534                retain
535            })
536            .unzip();
537        self.entries = entries.into_boxed_slice();
538        self.hash_ranges = hash_ranges.into_boxed_slice();
539        old_len != self.entries.len()
540    }
541
542    pub fn obsolete_entries(&self) -> &[u32] {
543        &self.obsolete_entries
544    }
545
546    pub fn obsolete_sst_files(&self) -> &[u32] {
547        &self.obsolete_sst_files
548    }
549
550    /// Looks up a key in this meta file.
551    ///
552    /// If `FIND_ALL` is false, returns after finding the first match.
553    /// If `FIND_ALL` is true, returns all entries with the same key from all SST files
554    /// (useful for keyspaces where keys are hashes and collisions are possible).
555    pub fn lookup<K: QueryKey, const FIND_ALL: bool>(
556        &self,
557        key_family: u32,
558        key_hash: u64,
559        key: &K,
560        key_block_cache: &BlockCache,
561        value_block_cache: &BlockCache,
562    ) -> Result<MetaLookupResult> {
563        if key_family != self.family {
564            return Ok(MetaLookupResult::FamilyMiss);
565        }
566        let mut miss_result = MetaLookupResult::RangeMiss;
567        let mut all_results: SmallVec<[LookupValue; 1]> = SmallVec::new();
568
569        for (index, range) in self.hash_ranges.iter().enumerate().rev() {
570            if !range.contains(key_hash) {
571                continue;
572            }
573            let entry = &self.entries[index];
574            if !entry.amqf.contains_fingerprint(key_hash) {
575                miss_result = MetaLookupResult::QuickFilterMiss;
576                continue;
577            }
578
579            let result = entry.sst(self)?.lookup::<K, FIND_ALL>(
580                key_hash,
581                key,
582                key_block_cache,
583                value_block_cache,
584            )?;
585
586            match result {
587                SstLookupResult::NotFound => {
588                    // continue searching other sst files
589                }
590                SstLookupResult::Found(values) => {
591                    if !FIND_ALL {
592                        // Return immediately with the first result
593                        return Ok(MetaLookupResult::SstLookup(SstLookupResult::Found(values)));
594                    }
595                    // A key tombstone stops the search across older SSTs within this meta file.
596                    // It sorts last within a key group, so it is the last value if present.
597                    // Key-value tombstones do not stop the search: they delete a single value,
598                    // and older SSTs may hold others for this key.
599                    let has_tombstone =
600                        values.last().is_some_and(|v| *v == LookupValue::KeyDeleted);
601                    all_results.extend(values);
602                    if has_tombstone {
603                        return Ok(MetaLookupResult::SstLookup(SstLookupResult::Found(
604                            all_results,
605                        )));
606                    }
607                }
608            }
609        }
610
611        if FIND_ALL && !all_results.is_empty() {
612            return Ok(MetaLookupResult::SstLookup(SstLookupResult::Found(
613                all_results,
614            )));
615        }
616
617        Ok(miss_result)
618    }
619
620    pub fn batch_lookup<K: QueryKey>(
621        &self,
622        key_family: u32,
623        keys: &[K],
624        cells: &mut [(u64, usize, Option<LookupValue>)],
625        empty_cells: &mut usize,
626        key_block_cache: &BlockCache,
627        value_block_cache: &BlockCache,
628    ) -> Result<MetaBatchLookupResult> {
629        if key_family != self.family {
630            #[cfg(feature = "stats")]
631            return Ok(MetaBatchLookupResult {
632                family_miss: true,
633                ..Default::default()
634            });
635            #[cfg(not(feature = "stats"))]
636            return Ok(MetaBatchLookupResult {});
637        }
638        debug_assert!(
639            cells.is_sorted_by_key(|(hash, _, _)| *hash),
640            "Cells must be sorted by key hash"
641        );
642        #[allow(unused_mut, reason = "It's used when stats are enabled")]
643        let mut lookup_result = MetaBatchLookupResult::default();
644        for (entry_index, range) in self.hash_ranges.iter().enumerate().rev() {
645            let start_index = cells
646                .binary_search_by(|(hash, _, _)| hash.cmp(&range.min_hash).then(Ordering::Greater))
647                .err()
648                .unwrap();
649            if start_index >= cells.len() {
650                #[cfg(feature = "stats")]
651                {
652                    lookup_result.range_misses += 1;
653                }
654                continue;
655            }
656            let end_index = cells
657                .binary_search_by(|(hash, _, _)| hash.cmp(&range.max_hash).then(Ordering::Less))
658                .err()
659                .unwrap()
660                .checked_sub(1);
661            let Some(end_index) = end_index else {
662                #[cfg(feature = "stats")]
663                {
664                    lookup_result.range_misses += 1;
665                }
666                continue;
667            };
668            if start_index > end_index {
669                #[cfg(feature = "stats")]
670                {
671                    lookup_result.range_misses += 1;
672                }
673                continue;
674            }
675            let entry = &self.entries[entry_index];
676            for (hash, index, result) in &mut cells[start_index..=end_index] {
677                debug_assert!(range.contains(*hash), "Key hash out of range");
678                if result.is_some() {
679                    continue;
680                }
681                if !entry.amqf.contains_fingerprint(*hash) {
682                    #[cfg(feature = "stats")]
683                    {
684                        lookup_result.quick_filter_misses += 1;
685                    }
686                    continue;
687                }
688                let sst_result = entry.sst(self)?.lookup::<_, false>(
689                    *hash,
690                    &keys[*index],
691                    key_block_cache,
692                    value_block_cache,
693                )?;
694                if let SstLookupResult::Found(mut values) = sst_result {
695                    // find_all=false guarantees exactly one result
696                    debug_assert!(values.len() == 1);
697                    let Some(value) = values.pop() else {
698                        unreachable!()
699                    };
700                    *result = Some(value);
701                    *empty_cells -= 1;
702                    #[cfg(feature = "stats")]
703                    {
704                        lookup_result.hits += 1;
705                    }
706                    if *empty_cells == 0 {
707                        return Ok(lookup_result);
708                    }
709                } else {
710                    #[cfg(feature = "stats")]
711                    {
712                        lookup_result.sst_misses += 1;
713                    }
714                }
715            }
716        }
717        Ok(lookup_result)
718    }
719}
turbo_persistence/meta_file.rs

turbo_persistence/
meta_file.rs
