qdrant-edge 0.8.0

//! # Search on level functions
//!
//! This module contains multiple variations of the SEARCH-LAYER function.
//! All of them implement a beam (greedy) search for closest points within a
//! single graph layer.
//!
//! - [`GraphLayersBase::search_on_level`]
//!   Regular search, as described in the original HNSW paper.
//!   Usually used on layer 0.
//!
//! - [`GraphLayersBase::search_on_level_acorn`]
//!   Variation of `search_on_level` that implements the ACORN-1 algorithm.
//!   Usually used on layer 0.
//!
//! - [`GraphLayersBase::search_entry_on_level`]
//!   Simplified version of `search_on_level` that uses beam size of 1.
//!   Usually used on all levels above level 0.
//!
//! - [`GraphLayersWithVectors::search_on_level_with_vectors`]
//!   Like `search_on_level`, but for graphs with [inline storage].
//!
//! - [`GraphLayersWithVectors::search_entry_on_level_with_vectors`]
//!   Like `search_entry_on_level`, but for graphs with [inline storage].
//!
//! [inline storage]: crate::types::HnswConfig::inline_storage

use std::borrow::Cow;
use std::cmp::max;
use std::ops::ControlFlow;
use std::path::{Path, PathBuf};
use std::sync::atomic::AtomicBool;

use crate::common::fixed_length_priority_queue::FixedLengthPriorityQueue;
use crate::common::fs::atomic_save;
use crate::common::types::{PointOffsetType, ScoredPointOffset};
use crate::common::universal_io::{
    CachedReadFs, MmapFs, OpenOptions, Populate, UniversalReadFs, read_bin_via,
};
use fs_err as fs;
use itertools::Itertools;
use serde::{Deserialize, Serialize};

use super::HnswM;
use super::entry_points::{EntryPoint, EntryPoints};
use super::graph_links::{GraphLinks, GraphLinksFormat, GraphLinksResidency};
use crate::segment::common::operation_error::{
    CancellableResult, OperationError, OperationResult, check_process_stopped,
};
use crate::segment::common::utils::rev_range;
use crate::segment::index::hnsw_index::graph_links::{GraphLinksFormatParam, serialize_graph_links};
use crate::segment::index::hnsw_index::point_scorer::{FilteredBytesScorer, FilteredScorer, ScorerFilters};
use crate::segment::index::hnsw_index::search_context::SearchContext;
use crate::segment::index::visited_pool::{VisitedListHandle, VisitedPool};
use crate::segment::vector_storage::RawScorer;
use crate::segment::vector_storage::query_scorer::QueryScorerBytes;

pub const HNSW_GRAPH_FILE: &str = "graph.bin";

pub const HNSW_LINKS_FILE: &str = "links.bin";
pub const COMPRESSED_HNSW_LINKS_FILE: &str = "links_compressed.bin";
pub const COMPRESSED_WITH_VECTORS_HNSW_LINKS_FILE: &str = "links_comp_vec.bin";

/// Contents of the `graph.bin` file.
#[derive(Deserialize, Serialize, Debug)]
pub(super) struct GraphLayerData<'a> {
    pub(super) m: usize,
    pub(super) m0: usize,
    pub(super) ef_construct: usize,
    pub(super) entry_points: Cow<'a, EntryPoints>,
}

#[derive(Debug)]
pub struct GraphLayers {
    pub(super) hnsw_m: HnswM,
    pub(super) links: GraphLinks,
    pub(super) entry_points: EntryPoints,
    pub(super) visited_pool: VisitedPool,
}

#[derive(Debug, Clone, Copy, Eq, PartialEq)]
pub enum SearchAlgorithm {
    Hnsw,
    Acorn,
}

pub trait GraphLayersBase {
    fn get_visited_list_from_pool(&self) -> VisitedListHandle<'_>;

    fn for_each_link<F>(&self, point_id: PointOffsetType, level: usize, f: F)
    where
        F: FnMut(PointOffsetType);

    fn try_for_each_link<F>(
        &self,
        point_id: PointOffsetType,
        level: usize,
        f: F,
    ) -> ControlFlow<(), ()>
    where
        F: FnMut(PointOffsetType) -> ControlFlow<(), ()>;

    /// Get M based on current level
    fn get_m(&self, level: usize) -> usize;

    /// Beam search for closest points within a single graph layer.
    ///
    /// See [module docs](self) for comparison with other search functions.
    fn search_on_level(
        &self,
        level_entry: ScoredPointOffset,
        level: usize,
        ef: usize,
        points_scorer: &mut FilteredScorer,
        is_stopped: &AtomicBool,
    ) -> CancellableResult<FixedLengthPriorityQueue<ScoredPointOffset>> {
        let mut visited_list = self.get_visited_list_from_pool();
        visited_list.check_and_update_visited(level_entry.idx);

        let mut search_context = SearchContext::new(ef);
        search_context.process_candidate(level_entry);

        let limit = self.get_m(level);
        let mut points_ids: Vec<PointOffsetType> = Vec::with_capacity(2 * limit);

        while let Some(candidate) = search_context.candidates.pop() {
            check_process_stopped(is_stopped)?;

            if candidate.score < search_context.lower_bound() {
                break;
            }

            points_ids.clear();
            self.for_each_link(candidate.idx, level, |link| {
                if !visited_list.check(link) {
                    points_ids.push(link);
                }
            });

            points_scorer
                .score_points(&mut points_ids, limit)
                .for_each(|score_point| {
                    search_context.process_candidate(score_point);
                    visited_list.check_and_update_visited(score_point.idx);
                });
        }

        Ok(search_context.nearest)
    }

    /// Variation of [`GraphLayersBase::search_on_level`] that implements the
    /// ACORN-1 algorithm.
    ///
    /// See [module docs](self) for comparison with other search functions.
    fn search_on_level_acorn(
        &self,
        level_entry: ScoredPointOffset,
        level: usize,
        ef: usize,
        points_scorer: &mut FilteredScorer,
        is_stopped: &AtomicBool,
    ) -> CancellableResult<FixedLengthPriorityQueue<ScoredPointOffset>> {
        // Each node in `hop1_visited_list` either:
        // a) Non-deleted node that going to be scored and added to
        //    `search_context` for further expansion. (or already added)
        // b) Deleted node that scheduled for exploration for 2-hop neighbors.
        let mut hop1_visited_list = self.get_visited_list_from_pool();
        hop1_visited_list.check_and_update_visited(level_entry.idx);

        // Nodes in `hop2_visited_list` are already explored as 2-hop neighbors.
        // Being in this list doesn't prevent the node to be handled again as
        // 1-hop neighbor.
        let mut hop2_visited_list = self.get_visited_list_from_pool();

        let mut search_context = SearchContext::new(ef);
        search_context.process_candidate(level_entry);

        // Limits are per every explored 1-hop or 2-hop neighbors, not total.
        // This is necessary to avoid over-scoring when there are many
        // additional graph links.
        let hop1_limit = self.get_m(level);
        let hop2_limit = self.get_m(level);
        debug_assert_ne!(self.get_m(level), 0); // See `FilteredBytesScorer::score_points`

        let mut to_score = Vec::with_capacity(hop1_limit * hop2_limit.min(16));
        let mut to_explore = Vec::with_capacity(hop1_limit * hop2_limit.min(16));

        while let Some(candidate) = search_context.candidates.pop() {
            check_process_stopped(is_stopped)?;

            if candidate.score < search_context.lower_bound() {
                break;
            }

            to_explore.clear();
            to_score.clear();

            // Collect 1-hop neighbors (direct neighbors)
            _ = self.try_for_each_link(candidate.idx, level, |hop1| {
                if hop1_visited_list.check_and_update_visited(hop1) {
                    return ControlFlow::Continue(());
                }

                if points_scorer.filters().check_vector(hop1) {
                    to_score.push(hop1);
                    if to_score.len() >= hop1_limit {
                        return ControlFlow::Break(());
                    }
                } else {
                    to_explore.push(hop1);
                }
                ControlFlow::Continue(())
            });

            // Collect 2-hop neighbors (neighbors of neighbors)
            for &hop1 in &to_explore {
                check_process_stopped(is_stopped)?;

                let total_limit = to_score.len() + hop2_limit;
                _ = self.try_for_each_link(hop1, level, |hop2| {
                    if hop1_visited_list.check(hop2)
                        || hop2_visited_list.check_and_update_visited(hop2)
                    {
                        return ControlFlow::Continue(());
                    }

                    if points_scorer.filters().check_vector(hop2) {
                        hop1_visited_list.check_and_update_visited(hop2);
                        to_score.push(hop2);
                        if to_score.len() >= total_limit {
                            return ControlFlow::Break(());
                        }
                    }
                    ControlFlow::Continue(())
                });
            }

            points_scorer
                .score_points_unfiltered(&to_score)
                .for_each(|score_point| search_context.process_candidate(score_point));
        }

        Ok(search_context.nearest)
    }

    /// Greedy searches for entry point of level `target_level`.
    /// Beam size is 1.
    fn search_entry(
        &self,
        entry_point: PointOffsetType,
        top_level: usize,
        target_level: usize,
        points_scorer: &mut FilteredScorer,
        is_stopped: &AtomicBool,
    ) -> CancellableResult<ScoredPointOffset> {
        let mut links_buffer = Vec::new();
        let mut result = None;
        let mut level_entry = entry_point;
        for level in rev_range(top_level, target_level) {
            check_process_stopped(is_stopped)?;
            let search_result =
                self.search_entry_on_level(level_entry, level, points_scorer, &mut links_buffer);
            level_entry = search_result.idx;
            result = Some(search_result);
        }
        if let Some(result) = result {
            Ok(result)
        } else {
            // If no levels, return the entry point with it's score
            Ok(ScoredPointOffset {
                idx: entry_point,
                score: points_scorer.score_point(entry_point),
            })
        }
    }

    /// Simplified version of `search_on_level` that uses beam size of 1.
    ///
    /// See [module docs](self) for comparison with other search functions.
    fn search_entry_on_level(
        &self,
        entry_point: PointOffsetType,
        level: usize,
        points_scorer: &mut FilteredScorer,
        // Temporary buffer for links to avoid unnecessary allocations.
        // 'links' is reused if `search_entry_on_level` is called multiple times.
        links: &mut Vec<PointOffsetType>,
    ) -> ScoredPointOffset {
        let limit = self.get_m(level);

        links.clear();
        links.reserve(2 * self.get_m(0));

        let mut changed = true;
        let mut current_point = ScoredPointOffset {
            idx: entry_point,
            score: points_scorer.score_point(entry_point),
        };
        while changed {
            changed = false;

            links.clear();
            self.for_each_link(current_point.idx, level, |link| {
                links.push(link);
            });

            points_scorer
                .score_points(links, limit)
                .for_each(|score_point| {
                    if score_point.score > current_point.score {
                        changed = true;
                        current_point = score_point;
                    }
                });
        }
        current_point
    }
}

pub trait GraphLayersWithVectors: GraphLayersBase {
    /// Returns `true` if the current graph format contains vectors.
    fn has_inline_vectors(&self) -> bool;

    /// # Panics
    ///
    /// Panics when using a format that does not support vectors.
    /// Check with [`Self::has_inline_vectors()`] before calling this method.
    fn links_with_vectors(
        &self,
        point_id: PointOffsetType,
        level: usize,
    ) -> (&[u8], impl Iterator<Item = (PointOffsetType, &[u8])> + '_);

    /// Similar to [`GraphLayersBase::search_on_level`].
    ///
    /// See [module docs](self) for comparison with other search functions.
    fn search_on_level_with_vectors(
        &self,
        level_entry: ScoredPointOffset,
        level: usize,
        ef: usize,
        links_scorer: &FilteredBytesScorer,
        base_scorer: &dyn QueryScorerBytes,
        is_stopped: &AtomicBool,
    ) -> CancellableResult<FixedLengthPriorityQueue<ScoredPointOffset>> {
        let mut visited_list = self.get_visited_list_from_pool();
        visited_list.check_and_update_visited(level_entry.idx);

        let mut links_search_context = SearchContext::new(ef);
        let mut base_search_context = SearchContext::new(ef);
        links_search_context.process_candidate(level_entry);

        let limit = self.get_m(level);
        let mut points: Vec<(PointOffsetType, &[u8])> = Vec::with_capacity(2 * limit);

        while let Some(candidate) = links_search_context.candidates.pop() {
            check_process_stopped(is_stopped)?;

            if candidate.score < links_search_context.lower_bound() {
                let (base_vector, _) = self.links_with_vectors(candidate.idx, level);
                base_search_context.process_candidate(ScoredPointOffset {
                    idx: candidate.idx,
                    score: base_scorer.score_bytes(base_vector),
                });
                break;
            }

            points.clear();
            let (base_vector, links_iter) = self.links_with_vectors(candidate.idx, level);
            links_iter.for_each(|(link, link_vector)| {
                if !visited_list.check(link) {
                    points.push((link, link_vector));
                }
            });
            base_search_context.process_candidate(ScoredPointOffset {
                idx: candidate.idx,
                score: base_scorer.score_bytes(base_vector),
            });

            links_scorer
                .score_points(&mut points, limit)
                .for_each(|score_point| {
                    links_search_context.process_candidate(score_point);
                    visited_list.check_and_update_visited(score_point.idx);
                });
        }

        Ok(base_search_context.nearest)
    }

    /// Similar to [`GraphLayersBase::search_entry`].
    fn search_entry_with_vectors(
        &self,
        entry_point: PointOffsetType,
        top_level: usize,
        target_level: usize,
        links_scorer_raw: &dyn RawScorer,
        links_scorer: &FilteredBytesScorer,
        is_stopped: &AtomicBool,
    ) -> CancellableResult<ScoredPointOffset> {
        let mut links_buffer = Vec::new();
        let mut current_point = ScoredPointOffset {
            idx: entry_point,
            score: links_scorer_raw.score_point(entry_point),
        };
        for level in rev_range(top_level, target_level) {
            check_process_stopped(is_stopped)?;
            current_point = self.search_entry_on_level_with_vectors(
                current_point,
                level,
                links_scorer,
                &mut links_buffer,
            );
        }
        Ok(current_point)
    }

    /// Similar to [`GraphLayersBase::search_entry_on_level`].
    ///
    /// See [module docs](self) for comparison with other search functions.
    fn search_entry_on_level_with_vectors<'a>(
        &'a self,
        entry_point: ScoredPointOffset,
        level: usize,
        links_scorer: &FilteredBytesScorer,
        links: &mut Vec<(PointOffsetType, &'a [u8])>,
    ) -> ScoredPointOffset {
        let limit = self.get_m(level);

        links.clear();
        links.reserve(limit);

        let mut changed = true;
        let mut current_point = entry_point;
        while changed {
            changed = false;

            links.clear();
            let (_, links_iter) = self.links_with_vectors(current_point.idx, level);
            links_iter.for_each(|(link, vector)| links.push((link, vector)));

            links_scorer
                .score_points(links, limit)
                .for_each(|score_point| {
                    if score_point.score > current_point.score {
                        changed = true;
                        current_point = score_point;
                    }
                });
        }
        current_point
    }
}

impl GraphLayersBase for GraphLayers {
    fn get_visited_list_from_pool(&self) -> VisitedListHandle<'_> {
        self.visited_pool.get(self.links.num_points())
    }

    fn for_each_link<F>(&self, point_id: PointOffsetType, level: usize, f: F)
    where
        F: FnMut(PointOffsetType),
    {
        self.links.links(point_id, level).for_each(f);
    }

    fn try_for_each_link<F>(
        &self,
        point_id: PointOffsetType,
        level: usize,
        f: F,
    ) -> ControlFlow<(), ()>
    where
        F: FnMut(PointOffsetType) -> ControlFlow<(), ()>,
    {
        self.links.links(point_id, level).try_for_each(f)
    }

    fn get_m(&self, level: usize) -> usize {
        self.hnsw_m.level_m(level)
    }
}

impl GraphLayersWithVectors for GraphLayers {
    fn has_inline_vectors(&self) -> bool {
        self.links.format().is_with_vectors()
    }

    fn links_with_vectors(
        &self,
        point_id: PointOffsetType,
        level: usize,
    ) -> (&[u8], impl Iterator<Item = (PointOffsetType, &[u8])> + '_) {
        self.links.links_with_vectors(point_id, level)
    }
}

/// Object contains links between nodes for HNSW search
///
/// Assume all scores are similarities. Larger score = closer points
impl GraphLayers {
    /// Returns the highest level this point is included in
    pub fn point_level(&self, point_id: PointOffsetType) -> usize {
        self.links.point_level(point_id)
    }

    fn get_entry_point(
        &self,
        filters: &ScorerFilters,
        custom_entry_points: Option<&[PointOffsetType]>,
    ) -> Option<EntryPoint> {
        // Try to get it from custom entry points
        custom_entry_points
            .and_then(|custom_entry_points| {
                custom_entry_points
                    .iter()
                    .filter(|&&point_id| filters.check_vector(point_id))
                    .map(|&point_id| {
                        let level = self.point_level(point_id);
                        EntryPoint { point_id, level }
                    })
                    .max_by_key(|ep| ep.level)
            })
            .or_else(|| {
                // Otherwise use normal entry points
                self.entry_points
                    .get_entry_point(|point_id| filters.check_vector(point_id))
            })
    }

    pub fn search(
        &self,
        top: usize,
        ef: usize,
        algorithm: SearchAlgorithm,
        mut points_scorer: FilteredScorer,
        custom_entry_points: Option<&[PointOffsetType]>,
        is_stopped: &AtomicBool,
    ) -> CancellableResult<Vec<ScoredPointOffset>> {
        let Some(entry_point) = self.get_entry_point(points_scorer.filters(), custom_entry_points)
        else {
            return Ok(Vec::default());
        };

        let zero_level_entry = self.search_entry(
            entry_point.point_id,
            entry_point.level,
            0,
            &mut points_scorer,
            is_stopped,
        )?;
        let ef = max(ef, top);
        let nearest = match algorithm {
            SearchAlgorithm::Hnsw => {
                self.search_on_level(zero_level_entry, 0, ef, &mut points_scorer, is_stopped)
            }
            SearchAlgorithm::Acorn => {
                self.search_on_level_acorn(zero_level_entry, 0, ef, &mut points_scorer, is_stopped)
            }
        }?;
        Ok(nearest.into_iter_sorted().take(top).collect_vec())
    }

    #[allow(clippy::too_many_arguments)]
    pub fn search_with_vectors(
        &self,
        top: usize,
        ef: usize,
        links_scorer: &FilteredScorer,
        links_scorer_bytes: &FilteredBytesScorer,
        base_scorer: &dyn QueryScorerBytes,
        custom_entry_points: Option<&[PointOffsetType]>,
        is_stopped: &AtomicBool,
    ) -> CancellableResult<Vec<ScoredPointOffset>> {
        let Some(entry_point) = self.get_entry_point(links_scorer.filters(), custom_entry_points)
        else {
            return Ok(Vec::default());
        };

        let zero_level_entry = self.search_entry_with_vectors(
            entry_point.point_id,
            entry_point.level,
            0,
            links_scorer.raw_scorer(),
            links_scorer_bytes,
            is_stopped,
        )?;
        let nearest = self.search_on_level_with_vectors(
            zero_level_entry,
            0,
            max(top, ef),
            links_scorer_bytes,
            base_scorer,
            is_stopped,
        )?;
        Ok(nearest.into_iter_sorted().take(top).collect_vec())
    }

    pub fn get_path(path: &Path) -> PathBuf {
        path.join(HNSW_GRAPH_FILE)
    }

    pub fn get_links_path(path: &Path, format: GraphLinksFormat) -> PathBuf {
        match format {
            GraphLinksFormat::Plain => path.join(HNSW_LINKS_FILE),
            GraphLinksFormat::Compressed => path.join(COMPRESSED_HNSW_LINKS_FILE),
            GraphLinksFormat::CompressedWithVectors => {
                path.join(COMPRESSED_WITH_VECTORS_HNSW_LINKS_FILE)
            }
        }
    }

    pub fn files(&self, path: &Path) -> Vec<PathBuf> {
        vec![
            GraphLayers::get_path(path),
            GraphLayers::get_links_path(path, self.links.format()),
        ]
    }

    pub fn num_points(&self) -> usize {
        self.links.num_points()
    }

    /// Heap RAM held by the graph links, in bytes.
    /// Zero when the links are backed by a live (mmap-backed) file handle;
    /// see [`GraphLinks::heap_size_bytes`].
    pub fn links_heap_size_bytes(&self) -> usize {
        self.links.heap_size_bytes()
    }
}

impl GraphLayers {
    /// Load via local mmap, optionally converting the links to the compressed
    /// format first. Used by the (mutable) on-disk HNSW index.
    ///
    /// `residency` controls how the links reside in memory after loading;
    /// see [`GraphLinksResidency`].
    pub fn load(
        dir: &Path,
        residency: GraphLinksResidency,
        compress: bool,
    ) -> OperationResult<Self> {
        if compress {
            // `convert_to_compressed` writes data, and we don't have a
            // `UniversalWriteFs` yet, so it stays on local mmap IO. It is not
            // enabled as per `super::hnsw::LINK_COMPRESSION_CONVERT_EXISTING`.
            let graph_data: GraphLayerData = read_bin_via(&MmapFs, GraphLayers::get_path(dir))?;
            Self::convert_to_compressed(dir, HnswM::new(graph_data.m, graph_data.m0))?;
        }

        Self::load_universal(&MmapFs, dir, residency)
    }

    /// Schedule background prefetch of the files [`Self::load_universal`] will
    /// read: the graph data plus whichever links format is present, probed in
    /// the same order as the load.
    pub fn preopen_universal(
        fs: &impl CachedReadFs,
        dir: &Path,
        residency: GraphLinksResidency,
    ) -> OperationResult<()> {
        // Graph data
        fs.schedule_prefetch(&Self::get_path(dir), None, None)?;

        // The load reads `Cached` links with a *blocking* populate and
        // materializes `Pinned` links into heap; at prefetch time both become
        // a background populate so the fetch overlaps the rest of the open.
        // `Cold` links stay unpopulated, matching the load.
        let populate = match residency {
            GraphLinksResidency::Cold => Populate::No,
            GraphLinksResidency::Cached | GraphLinksResidency::Pinned => Populate::PreferBackground,
        };
        let options = OpenOptions {
            populate,
            ..GraphLinks::open_options(residency)
        };

        // Links
        for format in [
            GraphLinksFormat::CompressedWithVectors,
            GraphLinksFormat::Compressed,
            GraphLinksFormat::Plain,
        ] {
            let path = Self::get_links_path(dir, format);
            if fs.exists(&path)? {
                fs.schedule_prefetch(&path, Some(options), None)?;
                break;
            }
        }
        Ok(())
    }

    /// Load purely through universal IO, without the format conversion path of
    /// [`Self::load`]. Used by the read-only index.
    ///
    /// `residency` controls how the links reside in memory after loading;
    /// see [`GraphLinksResidency`].
    pub fn load_universal<Fs>(
        fs: &Fs,
        dir: &Path,
        residency: GraphLinksResidency,
    ) -> OperationResult<Self>
    where
        Fs: UniversalReadFs,
        Fs::File: 'static,
    {
        let graph_data: GraphLayerData = read_bin_via(fs, GraphLayers::get_path(dir))?;

        Ok(Self {
            hnsw_m: HnswM::new(graph_data.m, graph_data.m0),
            links: Self::load_links_universal(fs, dir, residency)?,
            entry_points: graph_data.entry_points.into_owned(),
            visited_pool: VisitedPool::new(),
        })
    }

    fn load_links_universal<Fs>(
        fs: &Fs,
        dir: &Path,
        residency: GraphLinksResidency,
    ) -> OperationResult<GraphLinks>
    where
        Fs: UniversalReadFs,
        Fs::File: 'static,
    {
        for format in [
            GraphLinksFormat::CompressedWithVectors,
            GraphLinksFormat::Compressed,
            GraphLinksFormat::Plain,
        ] {
            let path = GraphLayers::get_links_path(dir, format);
            if fs.exists(&path)? {
                return GraphLinks::load_universal(fs, &path, format, residency);
            }
        }
        Err(OperationError::service_error("No links file found"))
    }

    /// Convert the "plain" format into the "compressed" format.
    /// Note: conversion into the "compressed with vectors" format is not
    /// supported at the moment, though it is possible to implement.
    /// As far as [`super::hnsw::LINK_COMPRESSION_CONVERT_EXISTING`] is false,
    /// this code is not used in production.
    fn convert_to_compressed(dir: &Path, hnsw_m: HnswM) -> OperationResult<()> {
        let plain_path = Self::get_links_path(dir, GraphLinksFormat::Plain);
        let compressed_path = Self::get_links_path(dir, GraphLinksFormat::Compressed);
        let compressed_with_vectors_path =
            Self::get_links_path(dir, GraphLinksFormat::CompressedWithVectors);

        if compressed_path.exists() || compressed_with_vectors_path.exists() {
            return Ok(());
        }

        let start = std::time::Instant::now();

        let links = GraphLinks::load_universal(
            &MmapFs,
            &plain_path,
            GraphLinksFormat::Plain,
            GraphLinksResidency::Cold,
        )?;
        let original_size = fs::metadata(&plain_path)?.len();
        atomic_save(&compressed_path, |writer| {
            let edges = links.to_edges();
            serialize_graph_links(edges, GraphLinksFormatParam::Compressed, hnsw_m, writer)
        })?;
        let new_size = fs::metadata(&compressed_path)?.len();

        // Remove the original file
        fs::remove_file(&plain_path)?;

        log::debug!(
            "Compressed HNSW graph links in {:.1?}: {:.1}MB -> {:.1}MB ({:.1}%)",
            start.elapsed(),
            original_size as f64 / 1024.0 / 1024.0,
            new_size as f64 / 1024.0 / 1024.0,
            new_size as f64 / original_size as f64 * 100.0,
        );

        Ok(())
    }

    #[cfg(feature = "testing")]
    pub fn compress_ram(&mut self) {
        assert_eq!(self.links.format(), GraphLinksFormat::Plain);
        let dummy =
            GraphLinks::new_from_edges(Vec::new(), GraphLinksFormatParam::Plain, HnswM::new2(0))
                .unwrap();
        let links = std::mem::replace(&mut self.links, dummy);
        self.links = GraphLinks::new_from_edges(
            links.to_edges(),
            GraphLinksFormatParam::Compressed,
            self.hnsw_m,
        )
        .unwrap();
    }

    pub fn populate(&self) -> OperationResult<()> {
        self.links.populate()?;
        Ok(())
    }

    pub fn clear_cache(&self) -> OperationResult<()> {
        let Self {
            hnsw_m: _,
            links,
            entry_points: _,
            visited_pool: _,
        } = self;
        links.clear_cache()?;
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use rand::SeedableRng;
    use rand::rngs::StdRng;
    use rstest::rstest;
    use tempfile::Builder;

    use super::*;
    use crate::segment::data_types::vectors::VectorElementType;
    use crate::segment::fixtures::index_fixtures::{TestRawScorerProducer, random_vector};
    use crate::segment::index::hnsw_index::tests::{
        create_graph_layer_builder_fixture, create_graph_layer_fixture,
    };
    use crate::segment::spaces::metric::Metric;
    use crate::segment::spaces::simple::CosineMetric;
    use crate::segment::types::Distance;
    use crate::segment::vector_storage::{DEFAULT_STOPPED, VectorStorageRead};

    /// `preopen_universal` must schedule exactly the files `load_universal`
    /// goes on to consume.
    ///
    /// Merely loading after a `preopen_universal` proves nothing: `CachedFs`
    /// falls back to a plain inner open for any path that was never scheduled.
    /// To make the prefetch pool the *only* possible source, the graph
    /// directory is emptied between the two calls: the already-open handles
    /// parked in the pool stay readable, while any fallback open hits
    /// `NotFound`.
    #[rstest]
    #[case::uncompressed(GraphLinksFormat::Plain, GraphLinksResidency::Cold)]
    #[case::compressed(GraphLinksFormat::Compressed, GraphLinksResidency::Cold)]
    #[case::compressed_cached(GraphLinksFormat::Compressed, GraphLinksResidency::Cached)]
    fn preopen_then_load_through_cached_fs(
        #[case] format: GraphLinksFormat,
        #[case] residency: GraphLinksResidency,
    ) {
        use crate::common::universal_io::{CachedFs, CachedReadFs};

        let num_vectors = 100;
        let dim = 8;
        let top = 5;

        let mut rng = StdRng::seed_from_u64(42);
        let dir = Builder::new().prefix("graph_preopen").tempdir().unwrap();

        let (vector_holder, graph_layers_builder) = create_graph_layer_builder_fixture(
            num_vectors,
            M,
            dim,
            false,
            false,
            Distance::Cosine,
            &mut rng,
        );
        let graph_links_vectors = vector_holder.graph_links_vectors();
        let graph = graph_layers_builder
            .into_graph_layers(
                dir.path(),
                format.with_param_for_tests(graph_links_vectors.as_ref()),
                false,
            )
            .unwrap();

        let query = random_vector(&mut rng, dim);
        let expected = search_in_graph(&query, top, &vector_holder, &graph);
        drop(graph);

        // Same order as the segment open path: snapshot, then preopen, then load.
        let mut cached_fs = CachedFs::new(MmapFs, dir.path()).unwrap();
        cached_fs.cache_file_info().unwrap();
        GraphLayers::preopen_universal(&cached_fs, dir.path(), residency).unwrap();

        // Everything `load_universal` reads must now come from the prefetch pool.
        for entry in fs_err::read_dir(dir.path()).unwrap() {
            fs_err::remove_file(entry.unwrap().path()).unwrap();
        }

        let graph = GraphLayers::load_universal(&cached_fs, dir.path(), residency).unwrap();
        assert_eq!(
            search_in_graph(&query, top, &vector_holder, &graph),
            expected
        );
    }

    fn search_in_graph(
        query: &[VectorElementType],
        top: usize,
        vector_storage: &TestRawScorerProducer,
        graph: &GraphLayers,
    ) -> Vec<ScoredPointOffset> {
        let scorer = vector_storage.scorer(query.to_owned());

        let ef = 16;
        graph
            .search(
                top,
                ef,
                SearchAlgorithm::Hnsw,
                scorer,
                None,
                &DEFAULT_STOPPED,
            )
            .unwrap()
    }

    const M: usize = 8;

    #[rstest]
    #[case::uncompressed(GraphLinksFormat::Plain)]
    #[case::compressed(GraphLinksFormat::Compressed)]
    #[case::compressed_with_vectors(GraphLinksFormat::CompressedWithVectors)]
    fn test_search_on_level(#[case] format: GraphLinksFormat) {
        let dim = 8;
        let hnsw_m = HnswM::new2(8);
        let entry_points_num = 10;
        let num_vectors = 10;

        let mut rng = StdRng::seed_from_u64(42);

        let vector_holder = TestRawScorerProducer::new(
            dim,
            Distance::Dot,
            num_vectors,
            format.is_with_vectors(),
            &mut rng,
        );

        let mut graph_links = vec![vec![Vec::new()]; num_vectors];
        graph_links[0][0] = vec![1, 2, 3, 4, 5, 6];

        let graph_links_vectors = vector_holder.graph_links_vectors();
        let format_param = format.with_param_for_tests(graph_links_vectors.as_ref());
        let graph_layers = GraphLayers {
            hnsw_m,
            links: GraphLinks::new_from_edges(graph_links.clone(), format_param, hnsw_m).unwrap(),
            entry_points: EntryPoints::new(entry_points_num),
            visited_pool: VisitedPool::new(),
        };

        let linking_idx: PointOffsetType = 7;

        let mut scorer = vector_holder.internal_scorer(linking_idx);

        let nearest_on_level = graph_layers
            .search_on_level(
                ScoredPointOffset {
                    idx: 0,
                    score: scorer.score_point(0),
                },
                0,
                32,
                &mut scorer,
                &DEFAULT_STOPPED,
            )
            .unwrap();

        assert_eq!(nearest_on_level.len(), graph_links[0][0].len() + 1);

        for nearest in nearest_on_level.iter_unsorted() {
            // eprintln!("nearest = {:#?}", nearest);
            assert_eq!(
                nearest.score,
                scorer.score_internal(linking_idx, nearest.idx)
            )
        }
    }

    #[rstest]
    #[case::uncompressed((GraphLinksFormat::Plain, false))]
    #[case::converted((GraphLinksFormat::Plain, true))]
    #[case::compressed((GraphLinksFormat::Compressed, false))]
    #[case::recompressed((GraphLinksFormat::Compressed, true))]
    #[case::compressed_with_vectors((GraphLinksFormat::CompressedWithVectors, false))]
    fn test_save_and_load(#[case] (initial_format, compress): (GraphLinksFormat, bool)) {
        let distance = Distance::Cosine;
        let num_vectors = 100;
        let dim = 8;
        let top = 5;

        let mut rng = StdRng::seed_from_u64(42);

        let dir = Builder::new().prefix("graph_dir").tempdir().unwrap();

        let query = random_vector(&mut rng, dim);

        let (vector_holder, graph_layers_builder) = create_graph_layer_builder_fixture(
            num_vectors,
            M,
            dim,
            false,
            true,
            distance,
            &mut rng,
        );
        let graph_links_vectors = vector_holder.graph_links_vectors();
        let graph1 = graph_layers_builder
            .into_graph_layers(
                dir.path(),
                initial_format.with_param_for_tests(graph_links_vectors.as_ref()),
                true,
            )
            .unwrap();
        assert_eq!(graph1.links.format(), initial_format);
        // The built graph must be mmap-backed, not pinned in heap.
        assert_eq!(graph1.links.heap_size_bytes(), 0);
        let res1 = search_in_graph(&query, top, &vector_holder, &graph1);
        drop(graph1);

        let graph2 = GraphLayers::load(dir.path(), GraphLinksResidency::Cached, compress).unwrap();
        if compress {
            assert_eq!(graph2.links.format(), GraphLinksFormat::Compressed);
        } else {
            assert_eq!(graph2.links.format(), initial_format);
        }
        // `Cached` residency keeps the links in the page cache, not in heap.
        assert_eq!(graph2.links.heap_size_bytes(), 0);
        let res2 = search_in_graph(&query, top, &vector_holder, &graph2);

        assert_eq!(res1, res2)
    }

    /// A freshly built graph must have the same, single-copy residency as one
    /// loaded from disk: mmap-backed for both `on_disk` values, never pinned
    /// in heap.
    #[rstest]
    fn test_built_graph_is_not_pinned(#[values(false, true)] on_disk: bool) {
        let mut rng = StdRng::seed_from_u64(42);
        let dir = Builder::new().prefix("graph_dir").tempdir().unwrap();

        let (_vector_holder, graph_layers_builder) =
            create_graph_layer_builder_fixture(100, M, 8, false, true, Distance::Cosine, &mut rng);
        let graph = graph_layers_builder
            .into_graph_layers(dir.path(), GraphLinksFormatParam::Compressed, on_disk)
            .unwrap();
        assert_eq!(graph.links.heap_size_bytes(), 0);
    }

    #[rstest]
    #[case::uncompressed(GraphLinksFormat::Plain)]
    #[case::compressed(GraphLinksFormat::Compressed)]
    #[case::compressed_with_vectors(GraphLinksFormat::CompressedWithVectors)]
    fn test_add_points(#[case] format: GraphLinksFormat) {
        type M = CosineMetric;
        let distance = <M as Metric<VectorElementType>>::distance();
        let num_vectors = 1000;
        let dim = 8;

        let mut rng = StdRng::seed_from_u64(42);

        let (vector_holder, graph_layers) = create_graph_layer_fixture(
            num_vectors,
            M,
            dim,
            format,
            false,
            format.is_with_vectors(),
            distance,
            &mut rng,
        );

        let main_entry = graph_layers
            .entry_points
            .get_entry_point(|_x| true)
            .expect("Expect entry point to exists");

        assert!(main_entry.level > 0);

        let num_levels = (0..num_vectors)
            .map(|i| graph_layers.links.point_level(i as PointOffsetType))
            .max()
            .unwrap();
        assert_eq!(main_entry.level, num_levels);

        let total_links_0 = (0..num_vectors)
            .map(|i| graph_layers.links.links(i as PointOffsetType, 0).len())
            .sum::<usize>();

        eprintln!("total_links_0 = {total_links_0:#?}");
        eprintln!("num_vectors = {num_vectors:#?}");
        assert!(total_links_0 > 0);
        assert!(total_links_0 as f64 / num_vectors as f64 > M as f64);

        let top = 5;
        let query = random_vector(&mut rng, dim);
        let processed_query = distance.preprocess_vector::<VectorElementType>(query.clone());
        let scorer = vector_holder.scorer(processed_query);
        let mut reference_top = FixedLengthPriorityQueue::new(top);
        for idx in 0..vector_holder.storage().total_vector_count() as PointOffsetType {
            let score = scorer.score_point(idx);
            reference_top.push(ScoredPointOffset { idx, score });
        }

        let graph_search = search_in_graph(&query, top, &vector_holder, &graph_layers);

        assert_eq!(reference_top.into_sorted_vec(), graph_search);
    }
}