1use 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
49pub(crate) const HNSW_LEVEL_RNG_SEED: u64 = 42;
57
58#[derive(Debug, Clone, Serialize, Deserialize, DeepSizeOf)]
60pub struct HnswBuildParams {
61 pub max_level: u16,
63
64 pub m: usize,
66
67 pub ef_construction: usize,
69
70 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 pub fn max_level(mut self, max_level: u16) -> Self {
99 self.max_level = max_level;
100 self
101 }
102
103 pub fn num_edges(mut self, m: usize) -> Self {
106 self.m = m;
107 self
108 }
109
110 pub fn ef_construction(mut self, ef_construction: usize) -> Self {
115 self.ef_construction = ef_construction;
116 self
117 }
118
119 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#[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 }
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 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 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 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 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 #[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 pub fn metadata(&self) -> HnswMetadata {
466 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 }
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 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 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 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 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(¤t_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
705pub(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
790enum LevelLookup {
792 Dense,
796 Sparse(HashMap<u32, u32>),
807}
808
809struct LoadedHnswGraph {
821 batch: RecordBatch,
824 level_neighbors: Vec<ListArray>,
826 level_lookup: Vec<LevelLookup>,
828 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 self.batch.get_array_memory_size()
840 }
841}
842
843impl LoadedHnswGraph {
844 #[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 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 let values = list.values().as_primitive::<UInt32Type>();
868 &values.values()[start..end]
869 }
870}
871
872pub(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 self.graph.level_count[0]
891 }
892
893 fn neighbors(&self, key: u32) -> Arc<Vec<u32>> {
894 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
911pub(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
942enum HnswGraph {
944 Built(Arc<Vec<GraphBuilderNode>>),
948 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 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 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 let neighbors = batch[NEIGHBORS_COL].as_list::<i32>().clone();
1028 let ids = batch[VECTOR_ID_COL].as_primitive::<UInt32Type>();
1029 if level == 0 {
1030 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 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 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 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, ¶ms)
1172 } else {
1173 self.search_basic(query, k, ¶ms, prefilter_bitset, storage)?
1174 };
1175 let _ = self.inner.visited_generator_queue.push(prefilter_generator);
1177
1178 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 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>>, store: &impl VectorStore,
1227 ) -> Result<Self> {
1228 Self::index_vectors(store, self.inner.params.clone())
1231 }
1232
1233 fn to_batch(&self) -> Result<RecordBatch> {
1235 let nodes = match &self.inner.graph {
1236 HnswGraph::Built(nodes) => nodes,
1237 HnswGraph::Loaded(graph) => {
1238 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, ¶ms, None, store.as_ref())
1393 .unwrap();
1394 let loaded_results = loaded_hnsw
1395 .search_basic(query, k, ¶ms, 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, ¶ms, None, store.as_ref())
1454 .unwrap();
1455 let loaded_results = loaded_hnsw
1456 .search_basic(query, k, ¶ms, None, store.as_ref())
1457 .unwrap();
1458 assert_eq!(builder_results, loaded_results);
1459 }
1460
1461 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 #[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, ¶ms, None, store.as_ref())
1512 .unwrap();
1513 let loaded_results = loaded
1514 .search_basic(query.clone(), k, ¶ms, None, store.as_ref())
1515 .unwrap();
1516 assert_eq!(builder_results, loaded_results);
1517
1518 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 #[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 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 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 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 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, ¶ms, None, store.as_ref())
1606 .unwrap();
1607 let loaded_results = loaded
1608 .search_basic(query, 10, ¶ms, None, store.as_ref())
1609 .unwrap();
1610 assert_eq!(builder_results, loaded_results);
1611 }
1612
1613 #[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 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 #[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 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 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 #[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 assert!(!matches!(loaded.inner.graph, HnswGraph::Loaded(_)));
1715 assert_eq!(loaded.to_batch().unwrap().num_rows(), 0);
1716 }
1717
1718 #[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, ¶ms, None, store.as_ref())
1752 .unwrap();
1753 let b = reloaded
1754 .search_basic(query, 10, ¶ms, None, store.as_ref())
1755 .unwrap();
1756 assert_eq!(a, b);
1757 }
1758
1759 #[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}