Skip to main content

lance_index/vector/hnsw/
builder.rs

1// SPDX-License-Identifier: Apache-2.0
2// SPDX-FileCopyrightText: Copyright The Lance Authors
3
4//! Builder of Hnsw Graph.
5
6use arrow::array::{AsArray, ListBuilder, UInt32Builder};
7use arrow::compute::concat_batches;
8use arrow::datatypes::{DataType, UInt32Type};
9use arrow_array::{ArrayRef, Float32Array, ListArray, RecordBatch, UInt64Array};
10use crossbeam_queue::ArrayQueue;
11use itertools::Itertools;
12use lance_core::deepsize::DeepSizeOf;
13
14use lance_core::utils::tokio::get_num_compute_intensive_cpus;
15use lance_linalg::distance::DistanceType;
16use rayon::prelude::*;
17use std::cmp::min;
18use std::collections::{BinaryHeap, HashMap, VecDeque};
19use std::fmt::Debug;
20use std::iter;
21use std::sync::Arc;
22use std::sync::RwLock;
23use std::sync::atomic::{AtomicUsize, Ordering};
24use tracing::instrument;
25
26use lance_core::{Error, Result};
27use rand::{Rng, SeedableRng, rngs::SmallRng};
28use serde::{Deserialize, Serialize};
29
30use super::super::graph::beam_search;
31use super::{
32    HNSW_TYPE, HnswMetadata, VECTOR_ID_COL, VECTOR_ID_FIELD, select_neighbors_heuristic_owned,
33};
34use crate::metrics::MetricsCollector;
35use crate::prefilter::PreFilter;
36use crate::vector::flat::storage::{FlatBinStorage, FlatFloatStorage};
37use crate::vector::graph::builder::GraphBuilderNode;
38use crate::vector::graph::{
39    BorrowingGraph, DISTS_FIELD, Graph, NEIGHBORS_COL, NEIGHBORS_FIELD, OrderedFloat, OrderedNode,
40    VisitedGenerator,
41};
42use crate::vector::graph::{Visited, beam_search_borrowed, greedy_search, greedy_search_borrowed};
43use crate::vector::storage::{DistCalculator, VectorStore};
44use crate::vector::v3::subindex::IvfSubIndex;
45use crate::vector::{Query, VECTOR_RESULT_SCHEMA};
46
47pub const HNSW_METADATA_KEY: &str = "lance:hnsw";
48
49/// Fixed seed for HNSW node-level assignment.
50///
51/// A constant seed makes graph construction reproducible (same data + params =>
52/// same graph), which keeps index builds deterministic and tests stable. Recall
53/// is statistically unaffected — the level distribution is identical, only the
54/// random draws become fixed. Shared by the offline ([`HNSWBuilder`]) and online
55/// ([`super::online::OnlineHnswBuilder`]) builders so both produce comparable graphs.
56pub(crate) const HNSW_LEVEL_RNG_SEED: u64 = 42;
57
58/// Parameters of building HNSW index
59#[derive(Debug, Clone, Serialize, Deserialize, DeepSizeOf)]
60pub struct HnswBuildParams {
61    /// max level ofm
62    pub max_level: u16,
63
64    /// number of connections to establish while inserting new element
65    pub m: usize,
66
67    /// size of the dynamic list for the candidates
68    pub ef_construction: usize,
69
70    /// number of vectors ahead to prefetch while building the graph
71    pub prefetch_distance: Option<usize>,
72}
73
74impl From<&HnswBuildParams> for crate::pb::HnswParameters {
75    fn from(params: &HnswBuildParams) -> Self {
76        Self {
77            max_connections: params.m as u32,
78            construction_ef: params.ef_construction as u32,
79            max_level: params.max_level as u32,
80        }
81    }
82}
83
84impl Default for HnswBuildParams {
85    fn default() -> Self {
86        Self {
87            max_level: 7,
88            m: 20,
89            ef_construction: 150,
90            prefetch_distance: Some(2),
91        }
92    }
93}
94
95impl HnswBuildParams {
96    /// The maximum level of the graph.
97    /// The default value is `8`.
98    pub fn max_level(mut self, max_level: u16) -> Self {
99        self.max_level = max_level;
100        self
101    }
102
103    /// The number of connections to establish while inserting new element
104    /// The default value is `30`.
105    pub fn num_edges(mut self, m: usize) -> Self {
106        self.m = m;
107        self
108    }
109
110    /// Number of candidates to be considered when searching for the nearest neighbors
111    /// during the construction of the graph.
112    ///
113    /// The default value is `100`.
114    pub fn ef_construction(mut self, ef_construction: usize) -> Self {
115        self.ef_construction = ef_construction;
116        self
117    }
118
119    /// Build the HNSW index from the given data.
120    ///
121    /// # Parameters
122    /// - `data`: A FixedSizeList to build the HNSW.
123    /// - `distance_type`: The distance type to use.
124    pub async fn build(self, data: ArrayRef, distance_type: DistanceType) -> Result<HNSW> {
125        let vectors = data.as_fixed_size_list().clone();
126        match (vectors.value_type(), distance_type) {
127            (DataType::UInt8, DistanceType::Hamming) => {
128                let vec_store = Arc::new(FlatBinStorage::new(vectors, distance_type));
129                HNSW::index_vectors(vec_store.as_ref(), self)
130            }
131            (DataType::UInt8, _) => Err(Error::invalid_input(format!(
132                "HNSW only supports hamming distance for UInt8 vectors, got {}",
133                distance_type
134            ))),
135            (_, DistanceType::Hamming) => Err(Error::invalid_input(format!(
136                "HNSW hamming distance only supports UInt8 vectors, got {}",
137                vectors.value_type()
138            ))),
139            _ => {
140                let vec_store = Arc::new(FlatFloatStorage::new(vectors, distance_type));
141                HNSW::index_vectors(vec_store.as_ref(), self)
142            }
143        }
144    }
145}
146
147/// Build a HNSW graph.
148///
149/// Currently, the HNSW graph is fully built in memory.
150///
151/// During the build, the graph is built layer by layer.
152///
153/// Each node in the graph has a global ID which is the index on the base layer.
154#[derive(Clone, DeepSizeOf)]
155pub struct HNSW {
156    inner: Arc<HnswCore>,
157}
158
159struct HnswCore {
160    params: HnswBuildParams,
161    graph: HnswGraph,
162    level_count: Vec<usize>,
163    entry_point: u32,
164    visited_generator_queue: Arc<ArrayQueue<VisitedGenerator>>,
165}
166
167impl DeepSizeOf for HnswCore {
168    fn deep_size_of_children(&self, context: &mut lance_core::deepsize::Context) -> usize {
169        self.params.deep_size_of_children(context)
170            + self.graph.deep_size_of_children(context)
171            + self.level_count.deep_size_of_children(context)
172        // Skipping the visited_generator_queue
173    }
174}
175
176impl HnswCore {
177    fn max_level(&self) -> u16 {
178        self.params.max_level
179    }
180
181    fn num_nodes(&self, level: usize) -> usize {
182        self.level_count[level]
183    }
184}
185
186impl Debug for HNSW {
187    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
188        write!(f, "HNSW(max_layers: {})", self.inner.max_level() as usize,)
189    }
190}
191
192impl HNSW {
193    /// Construct an HNSW from its constituent parts. Used by the online
194    /// builder when finalizing.
195    pub(crate) fn from_parts(
196        params: HnswBuildParams,
197        nodes: Vec<GraphBuilderNode>,
198        level_count: Vec<usize>,
199        entry_point: u32,
200    ) -> Self {
201        let queue_size = get_num_compute_intensive_cpus().max(1) * 2;
202        let visited_generator_queue = Arc::new(ArrayQueue::new(queue_size));
203        for _ in 0..queue_size {
204            let _ = visited_generator_queue.push(VisitedGenerator::new(0));
205        }
206        Self {
207            inner: Arc::new(HnswCore {
208                params,
209                graph: HnswGraph::Built(Arc::new(nodes)),
210                level_count,
211                entry_point,
212                visited_generator_queue,
213            }),
214        }
215    }
216
217    pub fn empty() -> Self {
218        Self {
219            inner: Arc::new(HnswCore {
220                params: HnswBuildParams::default(),
221                graph: HnswGraph::Built(Arc::new(Vec::new())),
222                level_count: Vec::new(),
223                entry_point: 0,
224                visited_generator_queue: Arc::new(ArrayQueue::new(1)),
225            }),
226        }
227    }
228
229    pub fn len(&self) -> usize {
230        match &self.inner.graph {
231            HnswGraph::Built(nodes) => nodes.len(),
232            // `level_count[0]` is the bottom-level (== total) node count.
233            HnswGraph::Loaded(graph) => graph.level_count[0],
234        }
235    }
236
237    pub fn is_empty(&self) -> bool {
238        self.len() == 0
239    }
240
241    pub fn max_level(&self) -> u16 {
242        self.inner.max_level()
243    }
244
245    pub fn num_nodes(&self, level: usize) -> usize {
246        self.inner.num_nodes(level)
247    }
248
249    /// Returns the in-memory builder nodes, if this graph was freshly built.
250    ///
251    /// A disk-loaded graph is Arrow-backed and has no `GraphBuilderNode`s,
252    /// so this returns `None` for it.
253    pub fn nodes(&self) -> Option<Arc<Vec<GraphBuilderNode>>> {
254        match &self.inner.graph {
255            HnswGraph::Built(nodes) => Some(nodes.clone()),
256            HnswGraph::Loaded(_) => None,
257        }
258    }
259
260    #[allow(clippy::too_many_arguments)]
261    pub fn search_inner(
262        &self,
263        query: ArrayRef,
264        k: usize,
265        params: &HnswQueryParams,
266        bitset: Option<Visited>,
267        visited_generator: &mut VisitedGenerator,
268        storage: &impl VectorStore,
269        prefetch_distance: Option<usize>,
270    ) -> Result<Vec<OrderedNode>> {
271        let dist_calc = storage.dist_calculator(query, params.dist_q_c);
272        let entry = self.inner.entry_point;
273        let ep = OrderedNode::new(entry, dist_calc.distance(entry).into());
274
275        // The level descent + bottom beam search are identical across
276        // graph backends; only the view types differ. `run_search` is
277        // generic over those view types so the loop is single-sourced:
278        // each backend supplies a per-level view closure and a
279        // bottom-level view.
280        let result = match &self.inner.graph {
281            HnswGraph::Built(nodes) => {
282                let nodes = nodes.as_slice();
283                self.run_search(
284                    ep,
285                    k,
286                    params,
287                    bitset.as_ref(),
288                    visited_generator,
289                    storage.len(),
290                    prefetch_distance,
291                    &dist_calc,
292                    |level| ImmutableHnswLevelView::new(level, nodes),
293                    ImmutableHnswBottomView::new(nodes),
294                )
295            }
296            HnswGraph::Loaded(graph) => {
297                let graph = graph.as_ref();
298                self.run_search(
299                    ep,
300                    k,
301                    params,
302                    bitset.as_ref(),
303                    visited_generator,
304                    storage.len(),
305                    prefetch_distance,
306                    &dist_calc,
307                    |level| LoadedHnswLevelView::new(level, graph),
308                    LoadedHnswBottomView::new(graph),
309                )
310            }
311        };
312        Ok(result)
313    }
314
315    /// Drives the shared HNSW query path over backend-specific graph
316    /// views: a per-level view produced by `make_level` and a
317    /// bottom-level view `bottom`. The views borrow their backing store
318    /// and are created, used, and dropped entirely within this call;
319    /// only the owned result escapes. Monomorphizing over `L`/`B` is the
320    /// single seam that lets the in-memory and disk-loaded backends
321    /// share one search loop.
322    #[allow(clippy::too_many_arguments)]
323    fn run_search<L, B>(
324        &self,
325        ep: OrderedNode,
326        k: usize,
327        params: &HnswQueryParams,
328        bitset: Option<&Visited>,
329        visited_generator: &mut VisitedGenerator,
330        storage_len: usize,
331        prefetch_distance: Option<usize>,
332        dist_calc: &impl DistCalculator,
333        make_level: impl Fn(u16) -> L,
334        bottom: B,
335    ) -> Vec<OrderedNode>
336    where
337        L: BorrowingGraph,
338        B: BorrowingGraph,
339    {
340        let mut ep = ep;
341        for level in (0..self.max_level()).rev() {
342            let cur_level = make_level(level);
343            ep = greedy_search_borrowed(
344                &cur_level,
345                ep,
346                dist_calc,
347                self.inner.params.prefetch_distance,
348            );
349        }
350        let mut visited = visited_generator.generate(storage_len);
351        beam_search_borrowed(
352            &bottom,
353            &ep,
354            params,
355            dist_calc,
356            bitset,
357            prefetch_distance,
358            &mut visited,
359        )
360        .into_iter()
361        .take(k)
362        .collect::<Vec<OrderedNode>>()
363    }
364
365    #[instrument(level = "debug", skip(self, query, bitset, storage))]
366    pub fn search_basic(
367        &self,
368        query: ArrayRef,
369        k: usize,
370        params: &HnswQueryParams,
371        bitset: Option<Visited>,
372        storage: &impl VectorStore,
373    ) -> Result<Vec<OrderedNode>> {
374        let mut visited_generator = self
375            .inner
376            .visited_generator_queue
377            .pop()
378            .unwrap_or_else(|| VisitedGenerator::new(storage.len()));
379        let result = self.search_inner(
380            query,
381            k,
382            params,
383            bitset,
384            &mut visited_generator,
385            storage,
386            Some(2),
387        );
388
389        match self.inner.visited_generator_queue.push(visited_generator) {
390            Ok(_) => {}
391            Err(_) => {
392                log::warn!("visited_generator_queue is full");
393            }
394        }
395
396        result
397    }
398
399    #[instrument(level = "debug", skip(self, storage, query, prefilter_bitset))]
400    fn flat_search(
401        &self,
402        storage: &impl VectorStore,
403        query: ArrayRef,
404        k: usize,
405        prefilter_bitset: Visited,
406        params: &HnswQueryParams,
407    ) -> Vec<OrderedNode> {
408        let lower_bound: OrderedFloat = params.lower_bound.unwrap_or(f32::MIN).into();
409        let upper_bound: OrderedFloat = params.upper_bound.unwrap_or(f32::MAX).into();
410
411        let dist_calc = storage.dist_calculator(query, params.dist_q_c);
412        let mut heap = BinaryHeap::<OrderedNode>::with_capacity(k);
413
414        match self.inner.params.prefetch_distance {
415            Some(ahead) if ahead > 0 => {
416                let mut ids_iter = prefilter_bitset.iter_ones().map(|i| i as u32);
417                let mut buffer = VecDeque::with_capacity(ahead + 1);
418                for _ in 0..=ahead {
419                    if let Some(id) = ids_iter.next() {
420                        buffer.push_back(id);
421                    } else {
422                        break;
423                    }
424                }
425
426                while let Some(node_id) = buffer.pop_front() {
427                    if let Some(&prefetch_id) = buffer.get(ahead - 1) {
428                        dist_calc.prefetch(prefetch_id);
429                    }
430                    if let Some(next) = ids_iter.next() {
431                        buffer.push_back(next);
432                    }
433
434                    let dist: OrderedFloat = dist_calc.distance(node_id).into();
435                    if dist <= lower_bound || dist > upper_bound {
436                        continue;
437                    }
438                    if heap.len() < k {
439                        heap.push((dist, node_id).into());
440                    } else if dist < heap.peek().unwrap().dist {
441                        heap.pop();
442                        heap.push((dist, node_id).into());
443                    }
444                }
445            }
446            _ => {
447                for node_id in prefilter_bitset.iter_ones().map(|i| i as u32) {
448                    let dist: OrderedFloat = dist_calc.distance(node_id).into();
449                    if dist <= lower_bound || dist > upper_bound {
450                        continue;
451                    }
452                    if heap.len() < k {
453                        heap.push((dist, node_id).into());
454                    } else if dist < heap.peek().unwrap().dist {
455                        heap.pop();
456                        heap.push((dist, node_id).into());
457                    }
458                }
459            }
460        };
461        heap.into_sorted_vec()
462    }
463
464    /// Returns the metadata of this [`HNSW`].
465    pub fn metadata(&self) -> HnswMetadata {
466        // calculate the offsets of each level,
467        // start from 0
468        let level_offsets = self
469            .inner
470            .level_count
471            .iter()
472            .chain(iter::once(&0))
473            .scan(0, |state, x| {
474                let start = *state;
475                *state += *x;
476                Some(start)
477            })
478            .collect();
479
480        HnswMetadata {
481            entry_point: self.inner.entry_point,
482            params: self.inner.params.clone(),
483            level_offsets,
484        }
485    }
486}
487
488struct HnswBuilder {
489    params: HnswBuildParams,
490
491    nodes: Arc<Vec<RwLock<GraphBuilderNode>>>,
492    level_count: Vec<AtomicUsize>,
493
494    entry_point: u32,
495
496    visited_generator_queue: Arc<ArrayQueue<VisitedGenerator>>,
497}
498
499impl DeepSizeOf for HnswBuilder {
500    fn deep_size_of_children(&self, context: &mut lance_core::deepsize::Context) -> usize {
501        self.params.deep_size_of_children(context)
502            + self.nodes.deep_size_of_children(context)
503            + self.level_count.deep_size_of_children(context)
504        // Skipping the visited_generator_queue
505    }
506}
507
508impl HnswBuilder {
509    fn finish(self) -> HNSW {
510        let nodes = match Arc::try_unwrap(self.nodes) {
511            Ok(nodes) => nodes
512                .into_iter()
513                .map(|node| node.into_inner().expect("builder lock poisoned"))
514                .collect(),
515            Err(nodes) => nodes
516                .iter()
517                .map(|node| node.read().expect("builder lock poisoned").clone())
518                .collect(),
519        };
520
521        let level_count = self
522            .level_count
523            .into_iter()
524            .map(|count| count.load(Ordering::Relaxed))
525            .collect();
526
527        HNSW {
528            inner: Arc::new(HnswCore {
529                params: self.params,
530                graph: HnswGraph::Built(Arc::new(nodes)),
531                level_count,
532                entry_point: self.entry_point,
533                visited_generator_queue: self.visited_generator_queue,
534            }),
535        }
536    }
537
538    /// Create a new [`HNSWBuilder`] with prepared params and in memory vector storage.
539    pub fn with_params(params: HnswBuildParams, storage: &impl VectorStore) -> Self {
540        let len = storage.len();
541        let max_level = params.max_level;
542
543        let level_count = (0..max_level)
544            .map(|_| AtomicUsize::new(0))
545            .collect::<Vec<_>>();
546
547        let visited_generator_queue = Arc::new(ArrayQueue::new(get_num_compute_intensive_cpus()));
548        for _ in 0..get_num_compute_intensive_cpus() {
549            visited_generator_queue
550                .push(VisitedGenerator::new(0))
551                .unwrap();
552        }
553        let mut builder = Self {
554            params,
555            nodes: Arc::new(Vec::new()),
556            level_count,
557            entry_point: 0,
558            visited_generator_queue,
559        };
560
561        if storage.is_empty() {
562            return builder;
563        }
564
565        let mut nodes = Vec::with_capacity(len);
566        {
567            if len > 0 {
568                nodes.push(RwLock::new(GraphBuilderNode::new(0, max_level as usize)));
569            }
570            let mut level_rng = SmallRng::seed_from_u64(HNSW_LEVEL_RNG_SEED);
571            for i in 1..len {
572                nodes.push(RwLock::new(GraphBuilderNode::new(
573                    i as u32,
574                    builder.random_level(&mut level_rng) as usize + 1,
575                )));
576            }
577        }
578        builder.nodes = Arc::new(nodes);
579
580        builder
581    }
582
583    /// New node's level
584    ///
585    /// See paper `Algorithm 1`
586    fn random_level<R: Rng + ?Sized>(&self, rng: &mut R) -> u16 {
587        let ml = 1.0 / (self.params.m as f32).ln();
588        min(
589            (-rng.random::<f32>().ln() * ml) as u16,
590            self.params.max_level - 1,
591        )
592    }
593
594    /// Insert one node.
595    fn insert(
596        &self,
597        node: u32,
598        visited_generator: &mut VisitedGenerator,
599        storage: &impl VectorStore,
600    ) {
601        let nodes = &self.nodes;
602        let target_level = nodes[node as usize].read().unwrap().level_neighbors.len() as u16 - 1;
603        let dist_calc = storage.dist_calculator_from_id(node);
604        let mut ep = OrderedNode::new(
605            self.entry_point,
606            dist_calc.distance(self.entry_point).into(),
607        );
608
609        //
610        // Search for entry point in paper.
611        // ```
612        //   for l_c in (L..l+1) {
613        //     W = Search-Layer(q, ep, ef=1, l_c)
614        //    ep = Select-Neighbors(W, 1)
615        //  }
616        // ```
617        for level in (target_level + 1..self.params.max_level).rev() {
618            let cur_level = HnswLevelView::new(level, nodes);
619            ep = greedy_search(&cur_level, ep, &dist_calc, self.params.prefetch_distance);
620        }
621
622        let mut pruned_neighbors_per_level: Vec<Vec<_>> =
623            vec![Vec::new(); (target_level + 1) as usize];
624        {
625            let mut current_node = nodes[node as usize].write().unwrap();
626            for level in (0..=target_level).rev() {
627                self.level_count[level as usize].fetch_add(1, Ordering::Relaxed);
628
629                let neighbors = self.search_level(&ep, level, &dist_calc, nodes, visited_generator);
630                for neighbor in &neighbors {
631                    current_node.add_neighbor(neighbor.id, neighbor.dist, level);
632                }
633                self.prune(storage, &mut current_node, level);
634                pruned_neighbors_per_level[level as usize]
635                    .clone_from(&current_node.level_neighbors_ranked[level as usize]);
636
637                ep = neighbors[0].clone();
638            }
639        }
640        for (level, pruned_neighbors) in pruned_neighbors_per_level.iter().enumerate() {
641            for unpruned_edge in pruned_neighbors {
642                let level = level as u16;
643                let m_max = match level {
644                    0 => self.params.m * 2,
645                    _ => self.params.m,
646                };
647                if unpruned_edge.dist
648                    < nodes[unpruned_edge.id as usize]
649                        .read()
650                        .unwrap()
651                        .cutoff(level, m_max)
652                {
653                    let mut chosen_node = nodes[unpruned_edge.id as usize].write().unwrap();
654                    chosen_node.add_neighbor(node, unpruned_edge.dist, level);
655                    self.prune(storage, &mut chosen_node, level);
656                }
657            }
658        }
659    }
660
661    fn search_level(
662        &self,
663        ep: &OrderedNode,
664        level: u16,
665        dist_calc: &impl DistCalculator,
666        nodes: &[RwLock<GraphBuilderNode>],
667        visited_generator: &mut VisitedGenerator,
668    ) -> Vec<OrderedNode> {
669        let cur_level = HnswLevelView::new(level, nodes);
670        let mut visited = visited_generator.generate(nodes.len());
671        beam_search(
672            &cur_level,
673            ep,
674            &HnswQueryParams {
675                ef: self.params.ef_construction,
676                lower_bound: None,
677                upper_bound: None,
678                dist_q_c: 0.0,
679            },
680            dist_calc,
681            None,
682            self.params.prefetch_distance,
683            &mut visited,
684        )
685    }
686
687    fn prune(&self, storage: &impl VectorStore, builder_node: &mut GraphBuilderNode, level: u16) {
688        let m_max = match level {
689            0 => self.params.m * 2,
690            _ => self.params.m,
691        };
692
693        let neighbors_ranked = &mut builder_node.level_neighbors_ranked[level as usize];
694        if neighbors_ranked.len() <= m_max {
695            builder_node.update_from_ranked_neighbors(level);
696            return;
697        }
698
699        let level_neighbors = std::mem::take(neighbors_ranked);
700        *neighbors_ranked = select_neighbors_heuristic_owned(storage, level_neighbors, m_max);
701        builder_node.update_from_ranked_neighbors(level);
702    }
703}
704
705// View of a level in HNSW graph.
706// This is used to iterate over neighbors in a specific level.
707pub(crate) struct HnswLevelView<'a> {
708    level: u16,
709    nodes: &'a [RwLock<GraphBuilderNode>],
710}
711
712impl<'a> HnswLevelView<'a> {
713    pub fn new(level: u16, nodes: &'a [RwLock<GraphBuilderNode>]) -> Self {
714        Self { level, nodes }
715    }
716}
717
718impl Graph for HnswLevelView<'_> {
719    fn len(&self) -> usize {
720        self.nodes.len()
721    }
722
723    fn neighbors(&self, key: u32) -> Arc<Vec<u32>> {
724        let node = &self.nodes[key as usize];
725        node.read().unwrap().level_neighbors[self.level as usize].clone()
726    }
727}
728
729pub(crate) struct ImmutableHnswLevelView<'a> {
730    level: u16,
731    nodes: &'a [GraphBuilderNode],
732}
733
734impl<'a> ImmutableHnswLevelView<'a> {
735    pub fn new(level: u16, nodes: &'a [GraphBuilderNode]) -> Self {
736        Self { level, nodes }
737    }
738}
739
740impl Graph for ImmutableHnswLevelView<'_> {
741    fn len(&self) -> usize {
742        self.nodes.len()
743    }
744
745    fn neighbors(&self, key: u32) -> Arc<Vec<u32>> {
746        self.nodes[key as usize].level_neighbors[self.level as usize].clone()
747    }
748}
749
750impl BorrowingGraph for ImmutableHnswLevelView<'_> {
751    fn len(&self) -> usize {
752        self.nodes.len()
753    }
754
755    fn neighbors(&self, key: u32) -> &[u32] {
756        self.nodes[key as usize].level_neighbors[self.level as usize].as_slice()
757    }
758}
759
760pub(crate) struct ImmutableHnswBottomView<'a> {
761    nodes: &'a [GraphBuilderNode],
762}
763
764impl<'a> ImmutableHnswBottomView<'a> {
765    pub fn new(nodes: &'a [GraphBuilderNode]) -> Self {
766        Self { nodes }
767    }
768}
769
770impl Graph for ImmutableHnswBottomView<'_> {
771    fn len(&self) -> usize {
772        self.nodes.len()
773    }
774
775    fn neighbors(&self, key: u32) -> Arc<Vec<u32>> {
776        self.nodes[key as usize].bottom_neighbors.clone()
777    }
778}
779
780impl BorrowingGraph for ImmutableHnswBottomView<'_> {
781    fn len(&self) -> usize {
782        self.nodes.len()
783    }
784
785    fn neighbors(&self, key: u32) -> &[u32] {
786        self.nodes[key as usize].bottom_neighbors.as_slice()
787    }
788}
789
790/// Per-level node-id -> row-index lookup for a disk-loaded HNSW graph.
791enum LevelLookup {
792    /// `row == node id`. Used only for level 0, where [`HNSW::to_batch`]
793    /// writes every node once in ascending `__vector_id` (== node id) order,
794    /// so the level-0 slice is exactly `[0, N)` with `row == id`.
795    Dense,
796    /// Upper level: an explicit `node_id -> row` map built from the level's
797    /// `__vector_id` column.
798    ///
799    /// We do *not* assume the column is sorted or that the slice is aligned
800    /// to a true level boundary: `level_offsets`/`level_count` omit the
801    /// entry-point node (it is written at every level by `to_batch` but only
802    /// counted at level 0), so upper-level slices can be off-by-one and
803    /// non-monotonic. Keying by the `__vector_id` value -- exactly what the
804    /// old per-node `load` did -- preserves behavior bit-for-bit. Upper
805    /// levels shrink geometrically, so this map stays tiny.
806    Sparse(HashMap<u32, u32>),
807}
808
809/// A search-only HNSW graph backed directly by the Arrow buffers of the
810/// on-disk `RecordBatch`.
811///
812/// Loading performs no per-node reconstruction: neighbor adjacency is served
813/// as `&[u32]` slices straight out of the `__neighbors` `ListArray` value
814/// buffer (zero copy). The full `batch` is retained so [`HNSW::to_batch`] is a
815/// near-free passthrough -- required, because the IVF partition cache
816/// re-serializes loaded indices through `to_batch()`
817/// (`lance/src/index/vector/ivf/partition_serde.rs`) -- and so a future
818/// zero-copy `CacheCodec` (#6745) can write/read it through
819/// `lance_arrow::ipc` without rebuilding the graph.
820struct LoadedHnswGraph {
821    /// The full loaded batch (all levels concatenated, level 0 first),
822    /// retained verbatim for `to_batch()` and #6745.
823    batch: RecordBatch,
824    /// Per-level `__neighbors` `List<UInt32>`, zero-copy slices of `batch`.
825    level_neighbors: Vec<ListArray>,
826    /// Per-level node-id -> row lookup (see [`LevelLookup`]).
827    level_lookup: Vec<LevelLookup>,
828    /// Number of nodes present at each level (`level_count[0]` == total).
829    level_count: Vec<usize>,
830}
831
832impl DeepSizeOf for LoadedHnswGraph {
833    fn deep_size_of_children(&self, _context: &mut lance_core::deepsize::Context) -> usize {
834        // `level_neighbors` are zero-copy views into `batch`, so counting
835        // `batch` alone avoids double counting (mirrors
836        // `vector/flat/storage.rs`). The upper-level `level_lookup` maps are
837        // sized to the geometrically-shrinking node counts above level 0 --
838        // negligible next to the batch and not separately accounted here.
839        self.batch.get_array_memory_size()
840    }
841}
842
843impl LoadedHnswGraph {
844    /// Borrow the neighbor ids of `key` at `level` directly from the Arrow
845    /// `ListArray` value buffer -- no allocation, no copy.
846    #[inline]
847    fn neighbors_at(&self, level: usize, key: u32) -> &[u32] {
848        let row = match &self.level_lookup[level] {
849            LevelLookup::Dense => key as usize,
850            LevelLookup::Sparse(id_to_row) => match id_to_row.get(&key) {
851                Some(&row) => row as usize,
852                // The node is absent at this level -- e.g. an empty upper
853                // level the search descends through, or a node that only
854                // exists at lower levels. Mirror the old representation
855                // (`level_neighbors[level]` defaulted to empty): no
856                // neighbors here, so greedy search stays put and descends.
857                None => return &[],
858            },
859        };
860        let list = &self.level_neighbors[level];
861        let offsets = list.value_offsets();
862        let start = offsets[row] as usize;
863        let end = offsets[row + 1] as usize;
864        // The `__neighbors` list child is `UInt32` per `HNSW::schema()`.
865        // Validity bitmap is ignored on purpose: `to_batch` never writes null
866        // neighbor lists, matching the previous `.unwrap()`-based load.
867        let values = list.values().as_primitive::<UInt32Type>();
868        &values.values()[start..end]
869    }
870}
871
872/// Per-level search view over a disk-loaded [`LoadedHnswGraph`].
873pub(crate) struct LoadedHnswLevelView<'a> {
874    level: usize,
875    graph: &'a LoadedHnswGraph,
876}
877
878impl<'a> LoadedHnswLevelView<'a> {
879    fn new(level: u16, graph: &'a LoadedHnswGraph) -> Self {
880        Self {
881            level: level as usize,
882            graph,
883        }
884    }
885}
886
887impl Graph for LoadedHnswLevelView<'_> {
888    fn len(&self) -> usize {
889        // Mirrors `ImmutableHnswLevelView::len` (total node count).
890        self.graph.level_count[0]
891    }
892
893    fn neighbors(&self, key: u32) -> Arc<Vec<u32>> {
894        // Non-hot fallback: HNSW search goes through `BorrowingGraph`. Kept
895        // only so the `Graph` trait / legacy `greedy_search` need no
896        // special-casing for loaded graphs.
897        Arc::new(self.graph.neighbors_at(self.level, key).to_vec())
898    }
899}
900
901impl BorrowingGraph for LoadedHnswLevelView<'_> {
902    fn len(&self) -> usize {
903        self.graph.level_count[0]
904    }
905
906    fn neighbors(&self, key: u32) -> &[u32] {
907        self.graph.neighbors_at(self.level, key)
908    }
909}
910
911/// Bottom-level (level 0) search view over a disk-loaded [`LoadedHnswGraph`].
912pub(crate) struct LoadedHnswBottomView<'a> {
913    graph: &'a LoadedHnswGraph,
914}
915
916impl<'a> LoadedHnswBottomView<'a> {
917    fn new(graph: &'a LoadedHnswGraph) -> Self {
918        Self { graph }
919    }
920}
921
922impl Graph for LoadedHnswBottomView<'_> {
923    fn len(&self) -> usize {
924        self.graph.level_count[0]
925    }
926
927    fn neighbors(&self, key: u32) -> Arc<Vec<u32>> {
928        Arc::new(self.graph.neighbors_at(0, key).to_vec())
929    }
930}
931
932impl BorrowingGraph for LoadedHnswBottomView<'_> {
933    fn len(&self) -> usize {
934        self.graph.level_count[0]
935    }
936
937    fn neighbors(&self, key: u32) -> &[u32] {
938        self.graph.neighbors_at(0, key)
939    }
940}
941
942/// The graph backing an [`HNSW`]: either built in memory or disk-loaded.
943enum HnswGraph {
944    /// Built in memory by the (online) builder / `index_vectors` /
945    /// `from_parts`. Mutable-shaped `GraphBuilderNode`s; `to_batch()`
946    /// re-encodes from these (it needs the per-node ranked distances).
947    Built(Arc<Vec<GraphBuilderNode>>),
948    /// Loaded from disk, Arrow-backed, search-only.
949    Loaded(Arc<LoadedHnswGraph>),
950}
951
952impl DeepSizeOf for HnswGraph {
953    fn deep_size_of_children(&self, context: &mut lance_core::deepsize::Context) -> usize {
954        match self {
955            Self::Built(nodes) => nodes.deep_size_of_children(context),
956            Self::Loaded(graph) => graph.deep_size_of_children(context),
957        }
958    }
959}
960
961#[derive(Debug, Clone, Copy)]
962pub struct HnswQueryParams {
963    pub ef: usize,
964    pub lower_bound: Option<f32>,
965    pub upper_bound: Option<f32>,
966    pub dist_q_c: f32,
967}
968
969impl From<&Query> for HnswQueryParams {
970    fn from(query: &Query) -> Self {
971        let k = query.k * query.refine_factor.unwrap_or(1) as usize;
972        Self {
973            ef: query.ef.unwrap_or(k + k / 2),
974            lower_bound: query.lower_bound,
975            upper_bound: query.upper_bound,
976            dist_q_c: query.dist_q_c,
977        }
978    }
979}
980
981impl IvfSubIndex for HNSW {
982    type BuildParams = HnswBuildParams;
983    type QueryParams = HnswQueryParams;
984
985    fn load(data: RecordBatch) -> Result<Self>
986    where
987        Self: Sized,
988    {
989        if data.num_rows() == 0 {
990            return Ok(Self::empty());
991        }
992
993        let hnsw_metadata = data
994            .schema_ref()
995            .metadata()
996            .get(HNSW_METADATA_KEY)
997            .ok_or(Error::index(format!("{} not found", HNSW_METADATA_KEY)))?;
998        let hnsw_metadata: HnswMetadata = serde_json::from_str(hnsw_metadata).map_err(|e| {
999            Error::index(format!(
1000                "Failed to decode HNSW metadata: {}, json: {}",
1001                e, hnsw_metadata
1002            ))
1003        })?;
1004
1005        // Slice the concatenated batch into one (zero-copy) view per level.
1006        let level_batches: Vec<RecordBatch> = hnsw_metadata
1007            .level_offsets
1008            .iter()
1009            .tuple_windows()
1010            .map(|(start, end)| data.slice(*start, end - start))
1011            .collect();
1012
1013        let level_count = level_batches
1014            .iter()
1015            .map(|b| b.num_rows())
1016            .collect::<Vec<_>>();
1017
1018        // No per-node reconstruction: keep the Arrow adjacency buffers as-is
1019        // and only build the tiny per-upper-level id->row lookups. The
1020        // `__distance` column is never materialized here -- search doesn't
1021        // need it, and `to_batch()` returns the retained `data` verbatim.
1022        let mut level_neighbors = Vec::with_capacity(level_batches.len());
1023        let mut level_lookup = Vec::with_capacity(level_batches.len());
1024        for (level, batch) in level_batches.iter().enumerate() {
1025            // `.clone()` on an Arrow array bumps a refcount; buffers stay
1026            // shared with `data` (zero copy).
1027            let neighbors = batch[NEIGHBORS_COL].as_list::<i32>().clone();
1028            let ids = batch[VECTOR_ID_COL].as_primitive::<UInt32Type>();
1029            if level == 0 {
1030                // `to_batch` writes every node at level 0 exactly once in
1031                // ascending `__vector_id` (== node id) order, so the level-0
1032                // slice is exactly `[0, N)` and the row index *is* the node
1033                // id. The `Dense` lookup below depends on this: in a release
1034                // build a violated invariant would silently make search read
1035                // the wrong neighbor list, so enforce it at load time (not via
1036                // `debug_assert!`) and reject a malformed or version-
1037                // incompatible batch.
1038                if let Some((row, id)) = ids
1039                    .values()
1040                    .iter()
1041                    .enumerate()
1042                    .find(|&(row, id)| *id != row as u32)
1043                {
1044                    return Err(Error::index(format!(
1045                        "HNSW level-0 __vector_id must equal the row index, but \
1046                         row {row} has __vector_id {id}; the on-disk batch is \
1047                         malformed or was written by an incompatible version"
1048                    )));
1049                }
1050                level_lookup.push(LevelLookup::Dense);
1051            } else {
1052                // Upper levels: explicit id -> row map. No ordering/alignment
1053                // assumption (see `LevelLookup::Sparse`). On the rare
1054                // duplicate id (a misaligned slice can repeat one across a
1055                // level boundary) the last wins, matching the old load's
1056                // `nodes[id].level_neighbors[level] = ...` last-write.
1057                let id_to_row: HashMap<u32, u32> = ids
1058                    .values()
1059                    .iter()
1060                    .enumerate()
1061                    .map(|(row, id)| (*id, row as u32))
1062                    .collect();
1063                level_lookup.push(LevelLookup::Sparse(id_to_row));
1064            }
1065            level_neighbors.push(neighbors);
1066        }
1067
1068        // `entry_point` is read from untrusted metadata and indexes the `Dense`
1069        // level-0 lookup directly; an out-of-range value would read past the
1070        // level-0 neighbor buffer during search. Validate it under the same
1071        // persisted-format invariant as the level-0 ids above.
1072        let num_nodes = level_count[0];
1073        if hnsw_metadata.entry_point as usize >= num_nodes {
1074            return Err(Error::index(format!(
1075                "HNSW entry_point {} is out of range for a graph with {num_nodes} \
1076                 nodes; the on-disk batch is malformed or was written by an \
1077                 incompatible version",
1078                hnsw_metadata.entry_point
1079            )));
1080        }
1081
1082        let visited_generator_queue =
1083            Arc::new(ArrayQueue::new(get_num_compute_intensive_cpus() * 2));
1084        for _ in 0..get_num_compute_intensive_cpus() * 2 {
1085            visited_generator_queue
1086                .push(VisitedGenerator::new(0))
1087                .unwrap();
1088        }
1089
1090        let graph = LoadedHnswGraph {
1091            batch: data,
1092            level_neighbors,
1093            level_lookup,
1094            level_count: level_count.clone(),
1095        };
1096        let inner = HnswCore {
1097            params: hnsw_metadata.params,
1098            graph: HnswGraph::Loaded(Arc::new(graph)),
1099            level_count,
1100            entry_point: hnsw_metadata.entry_point,
1101            visited_generator_queue,
1102        };
1103
1104        Ok(Self {
1105            inner: Arc::new(inner),
1106        })
1107    }
1108
1109    fn name() -> &'static str {
1110        HNSW_TYPE
1111    }
1112
1113    fn metadata_key() -> &'static str {
1114        "lance:hnsw"
1115    }
1116
1117    /// Return the schema of the sub index
1118    fn schema() -> arrow_schema::SchemaRef {
1119        arrow_schema::Schema::new(vec![
1120            VECTOR_ID_FIELD.clone(),
1121            NEIGHBORS_FIELD.clone(),
1122            DISTS_FIELD.clone(),
1123        ])
1124        .into()
1125    }
1126
1127    #[instrument(level = "debug", skip(self, query, storage, prefilter, _metrics))]
1128    fn search(
1129        &self,
1130        query: ArrayRef,
1131        k: usize,
1132        params: Self::QueryParams,
1133        storage: &impl VectorStore,
1134        prefilter: Arc<dyn PreFilter>,
1135        _metrics: &dyn MetricsCollector,
1136    ) -> Result<RecordBatch> {
1137        if params.ef < k {
1138            return Err(Error::index(
1139                "ef must be greater than or equal to k".to_string(),
1140            ));
1141        }
1142
1143        let schema = VECTOR_RESULT_SCHEMA.clone();
1144        if self.is_empty() {
1145            return Ok(RecordBatch::new_empty(schema));
1146        }
1147
1148        let mut prefilter_generator = self
1149            .inner
1150            .visited_generator_queue
1151            .pop()
1152            .unwrap_or_else(|| VisitedGenerator::new(storage.len()));
1153        let prefilter_bitset = if prefilter.is_empty() {
1154            None
1155        } else {
1156            let indices = prefilter.filter_row_ids(Box::new(storage.row_ids()));
1157            let mut bitset = prefilter_generator.generate(storage.len());
1158            for indices in indices {
1159                bitset.insert(indices as u32);
1160            }
1161            Some(bitset)
1162        };
1163
1164        let remained = prefilter_bitset
1165            .as_ref()
1166            .map(|b| b.count_ones())
1167            .unwrap_or(storage.len());
1168        let results = if remained < self.len() * 10 / 100 {
1169            let prefilter_bitset =
1170                prefilter_bitset.expect("the prefilter bitset must be set for flat search");
1171            self.flat_search(storage, query, k, prefilter_bitset, &params)
1172        } else {
1173            self.search_basic(query, k, &params, prefilter_bitset, storage)?
1174        };
1175        // if the queue is full, we just don't push it back, so ignore the error here
1176        let _ = self.inner.visited_generator_queue.push(prefilter_generator);
1177
1178        // need to unique by row ids in case of searching multivector
1179        let (row_ids, dists): (Vec<_>, Vec<_>) = results
1180            .into_iter()
1181            .map(|r| (storage.row_id(r.id), r.dist.0))
1182            .unique_by(|r| r.0)
1183            .unzip();
1184        let row_ids = Arc::new(UInt64Array::from(row_ids));
1185        let distances = Arc::new(Float32Array::from(dists));
1186
1187        Ok(RecordBatch::try_new(schema, vec![distances, row_ids])?)
1188    }
1189
1190    /// Given a vector storage, containing all the data for the IVF partition, build the sub index.
1191    fn index_vectors(storage: &impl VectorStore, params: Self::BuildParams) -> Result<Self>
1192    where
1193        Self: Sized,
1194    {
1195        let builder = HnswBuilder::with_params(params, storage);
1196
1197        log::debug!(
1198            "Building HNSW graph: num={}, max_levels={}, m={}, ef_construction={}, distance_type:{}",
1199            storage.len(),
1200            builder.params.max_level,
1201            builder.params.m,
1202            builder.params.ef_construction,
1203            storage.distance_type(),
1204        );
1205
1206        if storage.is_empty() {
1207            return Ok(builder.finish());
1208        }
1209
1210        let len = storage.len();
1211        builder.level_count[0].fetch_add(1, Ordering::Relaxed);
1212        (1..len).into_par_iter().for_each_init(
1213            || VisitedGenerator::new(len),
1214            |visited_generator, node| {
1215                builder.insert(node as u32, visited_generator, storage);
1216            },
1217        );
1218
1219        assert_eq!(builder.level_count[0].load(Ordering::Relaxed), len);
1220        Ok(builder.finish())
1221    }
1222
1223    fn remap(
1224        &self,
1225        _mapping: &HashMap<u64, Option<u64>>, // we don't need the mapping here because we rebuild the graph from remapped storage
1226        store: &impl VectorStore,
1227    ) -> Result<Self> {
1228        // We can't simply remap the row ids in the graph because the vectors are changed,
1229        // so the graph needs to be rebuilt.
1230        Self::index_vectors(store, self.inner.params.clone())
1231    }
1232
1233    /// Encode the sub index into a record batch
1234    fn to_batch(&self) -> Result<RecordBatch> {
1235        let nodes = match &self.inner.graph {
1236            HnswGraph::Built(nodes) => nodes,
1237            HnswGraph::Loaded(graph) => {
1238                // A loaded graph is already Arrow-backed: return the retained
1239                // batch verbatim, re-stamped with up-to-date HNSW metadata.
1240                // The IVF partition cache re-serializes loaded indices through
1241                // here (`ivf/partition_serde.rs`), so this must round-trip.
1242                let metadata = serde_json::to_string(&self.metadata())?;
1243                let schema =
1244                    graph
1245                        .batch
1246                        .schema()
1247                        .as_ref()
1248                        .clone()
1249                        .with_metadata(HashMap::from_iter(vec![(
1250                            HNSW_METADATA_KEY.to_string(),
1251                            metadata,
1252                        )]));
1253                return Ok(graph.batch.clone().with_schema(Arc::new(schema))?);
1254            }
1255        };
1256
1257        let mut vector_id_builder = UInt32Builder::with_capacity(self.len());
1258        let mut neighbors_builder = ListBuilder::with_capacity(UInt32Builder::new(), self.len());
1259        let mut distances_builder =
1260            ListBuilder::with_capacity(arrow_array::builder::Float32Builder::new(), self.len());
1261        let mut batches = Vec::with_capacity(self.max_level() as usize);
1262        for level in 0..self.max_level() {
1263            let level = level as usize;
1264            for (id, node) in nodes.iter().enumerate() {
1265                if level >= node.level_neighbors.len() {
1266                    continue;
1267                }
1268                let neighbors = node.level_neighbors[level].iter().map(|n| Some(*n));
1269                let distances = node.level_neighbors_ranked[level]
1270                    .iter()
1271                    .map(|n| Some(n.dist.0));
1272                vector_id_builder.append_value(id as u32);
1273                neighbors_builder.append_value(neighbors);
1274                distances_builder.append_value(distances);
1275            }
1276
1277            let batch = RecordBatch::try_new(
1278                Self::schema(),
1279                vec![
1280                    Arc::new(vector_id_builder.finish()),
1281                    Arc::new(neighbors_builder.finish()),
1282                    Arc::new(distances_builder.finish()),
1283                ],
1284            )?;
1285            batches.push(batch);
1286        }
1287
1288        let metadata = self.metadata();
1289        let metadata = serde_json::to_string(&metadata)?;
1290        let schema = Self::schema()
1291            .as_ref()
1292            .clone()
1293            .with_metadata(HashMap::from_iter(vec![(
1294                HNSW_METADATA_KEY.to_string(),
1295                metadata,
1296            )]));
1297        let batch = concat_batches(&Self::schema(), batches.iter())?;
1298        let batch = batch.with_schema(Arc::new(schema))?;
1299        Ok(batch)
1300    }
1301}
1302
1303#[cfg(test)]
1304mod tests {
1305    use std::sync::Arc;
1306
1307    use arrow_array::{ArrayRef, FixedSizeListArray, RecordBatch, UInt8Array, UInt32Array};
1308    use arrow_schema::Schema;
1309    use lance_arrow::FixedSizeListArrayExt;
1310    use lance_core::deepsize::DeepSizeOf;
1311    use lance_file::previous::{
1312        reader::FileReader as PreviousFileReader,
1313        writer::{
1314            FileWriter as PreviousFileWriter, FileWriterOptions as PreviousFileWriterOptions,
1315        },
1316    };
1317    use lance_io::object_store::ObjectStore;
1318    use lance_linalg::distance::DistanceType;
1319    use lance_table::format::SelfDescribingFileReader;
1320    use lance_table::io::manifest::ManifestDescribing;
1321    use lance_testing::datagen::generate_random_array;
1322    use object_store::path::Path;
1323    use rstest::rstest;
1324
1325    use super::HnswGraph;
1326    use crate::scalar::IndexWriter;
1327    use crate::vector::storage::{DistCalculator, VectorStore};
1328    use crate::vector::v3::subindex::IvfSubIndex;
1329    use crate::vector::{
1330        flat::storage::{FlatBinStorage, FlatFloatStorage},
1331        graph::{DISTS_FIELD, NEIGHBORS_FIELD},
1332        hnsw::{
1333            HNSW, VECTOR_ID_FIELD,
1334            builder::{HnswBuildParams, HnswQueryParams},
1335        },
1336    };
1337
1338    #[tokio::test]
1339    async fn test_builder_write_load() {
1340        const DIM: usize = 32;
1341        const TOTAL: usize = 2048;
1342        const NUM_EDGES: usize = 20;
1343        let data = generate_random_array(TOTAL * DIM);
1344        let fsl = FixedSizeListArray::try_new_from_values(data, DIM as i32).unwrap();
1345        let store = Arc::new(FlatFloatStorage::new(fsl.clone(), DistanceType::L2));
1346        let builder = HNSW::index_vectors(
1347            store.as_ref(),
1348            HnswBuildParams::default()
1349                .num_edges(NUM_EDGES)
1350                .ef_construction(50),
1351        )
1352        .unwrap();
1353
1354        let object_store = ObjectStore::memory();
1355        let path = Path::from("test_builder_write_load");
1356        let writer = object_store.create(&path).await.unwrap();
1357        let schema = Schema::new(vec![
1358            VECTOR_ID_FIELD.clone(),
1359            NEIGHBORS_FIELD.clone(),
1360            DISTS_FIELD.clone(),
1361        ]);
1362        let schema = lance_core::datatypes::Schema::try_from(&schema).unwrap();
1363        let mut writer = PreviousFileWriter::<ManifestDescribing>::with_object_writer(
1364            writer,
1365            schema,
1366            &PreviousFileWriterOptions::default(),
1367        )
1368        .unwrap();
1369        let batch = builder.to_batch().unwrap();
1370        let metadata = batch.schema_ref().metadata().clone();
1371        writer.write_record_batch(batch).await.unwrap();
1372        writer.finish_with_metadata(&metadata).await.unwrap();
1373
1374        let reader = PreviousFileReader::try_new_self_described(&object_store, &path, None)
1375            .await
1376            .unwrap();
1377        let batch = reader
1378            .read_range(0..reader.len(), reader.schema())
1379            .await
1380            .unwrap();
1381        let loaded_hnsw = HNSW::load(batch).unwrap();
1382
1383        let query = fsl.value(0);
1384        let k = 10;
1385        let params = HnswQueryParams {
1386            ef: 50,
1387            lower_bound: None,
1388            upper_bound: None,
1389            dist_q_c: 0.0,
1390        };
1391        let builder_results = builder
1392            .search_basic(query.clone(), k, &params, None, store.as_ref())
1393            .unwrap();
1394        let loaded_results = loaded_hnsw
1395            .search_basic(query, k, &params, None, store.as_ref())
1396            .unwrap();
1397        assert_eq!(builder_results, loaded_results);
1398    }
1399
1400    #[tokio::test]
1401    async fn test_builder_write_load_binary_hamming() {
1402        const DIM: usize = 8;
1403        const TOTAL: usize = 256;
1404        const NUM_EDGES: usize = 20;
1405        let data = UInt8Array::from_iter_values((0..TOTAL * DIM).map(|v| (v % 16) as u8));
1406        let fsl = FixedSizeListArray::try_new_from_values(data, DIM as i32).unwrap();
1407        let store = Arc::new(FlatBinStorage::new(fsl.clone(), DistanceType::Hamming));
1408        let builder = HnswBuildParams::default()
1409            .num_edges(NUM_EDGES)
1410            .ef_construction(50)
1411            .build(Arc::new(fsl.clone()), DistanceType::Hamming)
1412            .await
1413            .unwrap();
1414
1415        let object_store = ObjectStore::memory();
1416        let path = Path::from("test_builder_write_load_binary_hamming");
1417        let writer = object_store.create(&path).await.unwrap();
1418        let schema = Schema::new(vec![
1419            VECTOR_ID_FIELD.clone(),
1420            NEIGHBORS_FIELD.clone(),
1421            DISTS_FIELD.clone(),
1422        ]);
1423        let schema = lance_core::datatypes::Schema::try_from(&schema).unwrap();
1424        let mut writer = PreviousFileWriter::<ManifestDescribing>::with_object_writer(
1425            writer,
1426            schema,
1427            &PreviousFileWriterOptions::default(),
1428        )
1429        .unwrap();
1430        let batch = builder.to_batch().unwrap();
1431        let metadata = batch.schema_ref().metadata().clone();
1432        writer.write_record_batch(batch).await.unwrap();
1433        writer.finish_with_metadata(&metadata).await.unwrap();
1434
1435        let reader = PreviousFileReader::try_new_self_described(&object_store, &path, None)
1436            .await
1437            .unwrap();
1438        let batch = reader
1439            .read_range(0..reader.len(), reader.schema())
1440            .await
1441            .unwrap();
1442        let loaded_hnsw = HNSW::load(batch).unwrap();
1443
1444        let query = fsl.value(0);
1445        let k = 10;
1446        let params = HnswQueryParams {
1447            ef: 50,
1448            lower_bound: None,
1449            upper_bound: None,
1450            dist_q_c: 0.0,
1451        };
1452        let builder_results = builder
1453            .search_basic(query.clone(), k, &params, None, store.as_ref())
1454            .unwrap();
1455        let loaded_results = loaded_hnsw
1456            .search_basic(query, k, &params, None, store.as_ref())
1457            .unwrap();
1458        assert_eq!(builder_results, loaded_results);
1459    }
1460
1461    /// Brute-force top-`k` node ids by distance -- recall ground truth.
1462    fn brute_force_topk(store: &FlatFloatStorage, query: ArrayRef, k: usize) -> Vec<u32> {
1463        let dist_calc = store.dist_calculator(query, 0.0);
1464        let mut all: Vec<(f32, u32)> = (0..store.len() as u32)
1465            .map(|id| (dist_calc.distance(id), id))
1466            .collect();
1467        all.sort_by(|a, b| a.0.partial_cmp(&b.0).unwrap());
1468        all.into_iter().take(k).map(|(_, id)| id).collect()
1469    }
1470
1471    /// The Arrow-backed loaded graph must search bit-identically to the
1472    /// in-memory build, across distance types and graph sizes (single node,
1473    /// pair, and a multi-level graph exercising the sparse upper-level
1474    /// id->row lookup).
1475    #[rstest]
1476    #[case::l2_single(DistanceType::L2, 1)]
1477    #[case::l2_pair(DistanceType::L2, 2)]
1478    #[case::l2_multi_level(DistanceType::L2, 2048)]
1479    #[case::dot_multi_level(DistanceType::Dot, 2048)]
1480    #[tokio::test]
1481    async fn test_loaded_search_parity_and_recall(
1482        #[case] distance_type: DistanceType,
1483        #[case] total: usize,
1484    ) {
1485        const DIM: usize = 32;
1486        let fsl =
1487            FixedSizeListArray::try_new_from_values(generate_random_array(total * DIM), DIM as i32)
1488                .unwrap();
1489        let store = Arc::new(FlatFloatStorage::new(fsl.clone(), distance_type));
1490        let builder = HNSW::index_vectors(
1491            store.as_ref(),
1492            HnswBuildParams::default().num_edges(20).ef_construction(50),
1493        )
1494        .unwrap();
1495        assert!(!matches!(builder.inner.graph, HnswGraph::Loaded(_)));
1496
1497        let loaded = HNSW::load(builder.to_batch().unwrap()).unwrap();
1498        assert!(matches!(loaded.inner.graph, HnswGraph::Loaded(_)));
1499        assert_eq!(loaded.len(), total);
1500
1501        let k = total.min(10);
1502        let params = HnswQueryParams {
1503            ef: 50,
1504            lower_bound: None,
1505            upper_bound: None,
1506            dist_q_c: 0.0,
1507        };
1508        let query = fsl.value(0);
1509
1510        let builder_results = builder
1511            .search_basic(query.clone(), k, &params, None, store.as_ref())
1512            .unwrap();
1513        let loaded_results = loaded
1514            .search_basic(query.clone(), k, &params, None, store.as_ref())
1515            .unwrap();
1516        assert_eq!(builder_results, loaded_results);
1517
1518        // Recall vs brute-force ground truth (project rule: >= 0.5).
1519        let truth: std::collections::HashSet<u32> = brute_force_topk(store.as_ref(), query, k)
1520            .into_iter()
1521            .collect();
1522        let hits = loaded_results
1523            .iter()
1524            .filter(|n| truth.contains(&n.id))
1525            .count();
1526        let recall = hits as f32 / k as f32;
1527        assert!(recall >= 0.5, "recall {recall} below 0.5 (k={k})");
1528    }
1529
1530    /// Regression guard for the `level_offsets` misalignment (issue #6746).
1531    /// `to_batch` writes the entry-point node at *every* level, but
1532    /// `level_count` only counts it at level 0, so the serialized batch has
1533    /// strictly more rows than `sum(level_count)` and the upper-level
1534    /// `level_offsets` slices are off-by-one / non-monotonic. The Arrow-backed
1535    /// loaded graph must still search bit-identically to the in-memory build:
1536    /// it keys upper levels by `__vector_id` value via the `Sparse` map
1537    /// (last-write-wins), never `row == id`. A naive `row == id`
1538    /// reimplementation would pass the small cases but break here.
1539    #[tokio::test]
1540    async fn test_loaded_level_offsets_misalignment_invariant() {
1541        use arrow::array::AsArray;
1542        use arrow::datatypes::UInt32Type;
1543
1544        const DIM: usize = 32;
1545        const TOTAL: usize = 2048;
1546        let fsl =
1547            FixedSizeListArray::try_new_from_values(generate_random_array(TOTAL * DIM), DIM as i32)
1548                .unwrap();
1549        let store = Arc::new(FlatFloatStorage::new(fsl.clone(), DistanceType::L2));
1550        let builder = HNSW::index_vectors(
1551            store.as_ref(),
1552            HnswBuildParams::default().num_edges(20).ef_construction(50),
1553        )
1554        .unwrap();
1555
1556        // The scenario only exists on a multi-level graph.
1557        assert!(
1558            builder.max_level() >= 2,
1559            "expected a multi-level graph (got max_level {})",
1560            builder.max_level()
1561        );
1562
1563        let batch = builder.to_batch().unwrap();
1564        let md = builder.metadata();
1565        let total_counted = *md.level_offsets.last().unwrap();
1566
1567        // The exact misalignment: more serialized rows than `level_count` sums
1568        // to, because the entry-point node is written at every level yet
1569        // counted only at level 0.
1570        assert!(
1571            batch.num_rows() > total_counted,
1572            "expected serialized rows ({}) to exceed sum(level_count) ({}) -- \
1573             entry point should be written at every level",
1574            batch.num_rows(),
1575            total_counted,
1576        );
1577
1578        // Level-0 slice must still be exactly `[0, N)` with
1579        // `__vector_id == row` -- the precondition for `LevelLookup::Dense`.
1580        let n = md.level_offsets[1];
1581        assert_eq!(n, TOTAL);
1582        let level0 = batch.slice(0, n);
1583        let ids = level0.column(0).as_primitive::<UInt32Type>();
1584        assert!(
1585            ids.values()
1586                .iter()
1587                .enumerate()
1588                .all(|(row, id)| *id == row as u32),
1589            "level-0 __vector_id must equal the row index",
1590        );
1591
1592        // Despite the surplus rows and off-by-one upper slices, the loaded
1593        // graph searches bit-identically to the in-memory build (old `load`
1594        // semantics preserved via the `Sparse` last-write-wins map).
1595        let loaded = HNSW::load(batch).unwrap();
1596        assert!(matches!(loaded.inner.graph, HnswGraph::Loaded(_)));
1597        let params = HnswQueryParams {
1598            ef: 50,
1599            lower_bound: None,
1600            upper_bound: None,
1601            dist_q_c: 0.0,
1602        };
1603        let query = fsl.value(0);
1604        let builder_results = builder
1605            .search_basic(query.clone(), 10, &params, None, store.as_ref())
1606            .unwrap();
1607        let loaded_results = loaded
1608            .search_basic(query, 10, &params, None, store.as_ref())
1609            .unwrap();
1610        assert_eq!(builder_results, loaded_results);
1611    }
1612
1613    /// `load()` must reject a batch whose level-0 `__vector_id` no longer
1614    /// matches the row index. The `LevelLookup::Dense` fast path relies on
1615    /// `row == id`, and the old `debug_assert!` was compiled out of release
1616    /// builds -- so a corrupt batch must fail at the `load()` boundary instead
1617    /// of silently searching the wrong neighbor lists.
1618    #[tokio::test]
1619    async fn test_load_rejects_misaligned_level0_id() {
1620        use arrow::array::AsArray;
1621        use arrow::datatypes::UInt32Type;
1622
1623        const DIM: usize = 16;
1624        const TOTAL: usize = 256;
1625        let fsl =
1626            FixedSizeListArray::try_new_from_values(generate_random_array(TOTAL * DIM), DIM as i32)
1627                .unwrap();
1628        let store = Arc::new(FlatFloatStorage::new(fsl, DistanceType::L2));
1629        let builder = HNSW::index_vectors(
1630            store.as_ref(),
1631            HnswBuildParams::default().num_edges(20).ef_construction(50),
1632        )
1633        .unwrap();
1634
1635        let batch = builder.to_batch().unwrap();
1636        // Row 0 is always a level-0 node; break its `__vector_id == row`
1637        // invariant while preserving the (metadata-bearing) schema.
1638        let mut ids = batch
1639            .column(0)
1640            .as_primitive::<UInt32Type>()
1641            .values()
1642            .to_vec();
1643        ids[0] = ids.len() as u32;
1644        let mut columns = batch.columns().to_vec();
1645        columns[0] = Arc::new(UInt32Array::from(ids));
1646        let corrupted = RecordBatch::try_new(batch.schema(), columns).unwrap();
1647
1648        assert!(
1649            HNSW::load(corrupted).is_err(),
1650            "load() must reject a misaligned level-0 __vector_id"
1651        );
1652    }
1653
1654    /// `load()` must reject metadata whose `entry_point` is out of range for
1655    /// the node count: it indexes the `Dense` level-0 lookup directly, so an
1656    /// out-of-range value would read past the level-0 neighbor buffer at search
1657    /// time.
1658    #[tokio::test]
1659    async fn test_load_rejects_out_of_range_entry_point() {
1660        use super::{HNSW_METADATA_KEY, HnswMetadata};
1661
1662        const DIM: usize = 16;
1663        const TOTAL: usize = 256;
1664        let fsl =
1665            FixedSizeListArray::try_new_from_values(generate_random_array(TOTAL * DIM), DIM as i32)
1666                .unwrap();
1667        let store = Arc::new(FlatFloatStorage::new(fsl, DistanceType::L2));
1668        let builder = HNSW::index_vectors(
1669            store.as_ref(),
1670            HnswBuildParams::default().num_edges(20).ef_construction(50),
1671        )
1672        .unwrap();
1673
1674        let batch = builder.to_batch().unwrap();
1675        let mut metadata = batch.schema_ref().metadata().clone();
1676        let mut md: HnswMetadata =
1677            serde_json::from_str(metadata.get(HNSW_METADATA_KEY).unwrap()).unwrap();
1678        // Valid entry points are `[0, N)`; `level_offsets[1]` == N is one past.
1679        let n = md.level_offsets[1];
1680        md.entry_point = n as u32;
1681        metadata.insert(
1682            HNSW_METADATA_KEY.to_string(),
1683            serde_json::to_string(&md).unwrap(),
1684        );
1685        // Rebuild the batch under the rewritten metadata. `with_schema` would
1686        // reject this: it requires the new metadata to be a superset, but we
1687        // are changing an existing key's value, not adding one.
1688        let schema = batch.schema().as_ref().clone().with_metadata(metadata);
1689        let corrupted = RecordBatch::try_new(Arc::new(schema), batch.columns().to_vec()).unwrap();
1690
1691        assert!(
1692            HNSW::load(corrupted).is_err(),
1693            "load() must reject an out-of-range entry_point"
1694        );
1695    }
1696
1697    /// An empty index round-trips: 0-row `to_batch` -> `load` -> empty graph.
1698    #[tokio::test]
1699    async fn test_loaded_empty_index() {
1700        const DIM: usize = 16;
1701        let fsl =
1702            FixedSizeListArray::try_new_from_values(generate_random_array(0), DIM as i32).unwrap();
1703        let store = Arc::new(FlatFloatStorage::new(fsl, DistanceType::L2));
1704        let builder = HNSW::index_vectors(store.as_ref(), HnswBuildParams::default()).unwrap();
1705        assert!(builder.is_empty());
1706
1707        let batch = builder.to_batch().unwrap();
1708        assert_eq!(batch.num_rows(), 0);
1709
1710        let loaded = HNSW::load(batch).unwrap();
1711        assert!(loaded.is_empty());
1712        assert_eq!(loaded.len(), 0);
1713        // A 0-row load short-circuits to the empty (Built) graph.
1714        assert!(!matches!(loaded.inner.graph, HnswGraph::Loaded(_)));
1715        assert_eq!(loaded.to_batch().unwrap().num_rows(), 0);
1716    }
1717
1718    /// build -> `to_batch` (b1) -> `load` -> `to_batch` (b2) must satisfy
1719    /// `b1 == b2`, and the round-tripped batch must reload and search
1720    /// identically. This is exactly the IVF partition-cache path:
1721    /// `ivf/partition_serde.rs` calls `to_batch()` on a *loaded* index.
1722    #[tokio::test]
1723    async fn test_to_batch_roundtrip_loaded() {
1724        const DIM: usize = 24;
1725        const TOTAL: usize = 1500;
1726        let fsl =
1727            FixedSizeListArray::try_new_from_values(generate_random_array(TOTAL * DIM), DIM as i32)
1728                .unwrap();
1729        let store = Arc::new(FlatFloatStorage::new(fsl.clone(), DistanceType::L2));
1730        let builder = HNSW::index_vectors(
1731            store.as_ref(),
1732            HnswBuildParams::default().num_edges(16).ef_construction(50),
1733        )
1734        .unwrap();
1735
1736        let b1 = builder.to_batch().unwrap();
1737        let loaded = HNSW::load(b1.clone()).unwrap();
1738        assert!(matches!(loaded.inner.graph, HnswGraph::Loaded(_)));
1739        let b2 = loaded.to_batch().unwrap();
1740        assert_eq!(b1, b2);
1741
1742        let reloaded = HNSW::load(b2).unwrap();
1743        let params = HnswQueryParams {
1744            ef: 50,
1745            lower_bound: None,
1746            upper_bound: None,
1747            dist_q_c: 0.0,
1748        };
1749        let query = fsl.value(7);
1750        let a = builder
1751            .search_basic(query.clone(), 10, &params, None, store.as_ref())
1752            .unwrap();
1753        let b = reloaded
1754            .search_basic(query, 10, &params, None, store.as_ref())
1755            .unwrap();
1756        assert_eq!(a, b);
1757    }
1758
1759    /// The loaded graph shares the Arrow batch and reconstructs no per-node
1760    /// `Vec<GraphBuilderNode>` / `Vec<OrderedNode>`, so it is strictly
1761    /// lighter than the in-memory build representation.
1762    #[tokio::test]
1763    async fn test_loaded_graph_is_arrow_backed() {
1764        const DIM: usize = 32;
1765        const TOTAL: usize = 2048;
1766        let fsl =
1767            FixedSizeListArray::try_new_from_values(generate_random_array(TOTAL * DIM), DIM as i32)
1768                .unwrap();
1769        let store = Arc::new(FlatFloatStorage::new(fsl, DistanceType::L2));
1770        let builder = HNSW::index_vectors(
1771            store.as_ref(),
1772            HnswBuildParams::default().num_edges(20).ef_construction(50),
1773        )
1774        .unwrap();
1775        assert!(!matches!(builder.inner.graph, HnswGraph::Loaded(_)));
1776
1777        let loaded = HNSW::load(builder.to_batch().unwrap()).unwrap();
1778        assert!(matches!(loaded.inner.graph, HnswGraph::Loaded(_)));
1779        assert!(
1780            loaded.deep_size_of() < builder.deep_size_of(),
1781            "loaded graph ({}) should be lighter than built ({})",
1782            loaded.deep_size_of(),
1783            builder.deep_size_of(),
1784        );
1785    }
1786}