diskann_benchmark_core/search/graph/
multihop.rs1use std::sync::Arc;
7
8use diskann::{
9 ANNResult,
10 graph::{self, ext::labeled, glue},
11 provider,
12};
13use diskann_utils::{future::AsyncFriendly, views::Matrix};
14
15use crate::search::{self, Search, graph::Strategy};
16
17#[derive(Debug)]
28pub struct MultiHop<DP, T, S>
29where
30 DP: provider::DataProvider,
31{
32 index: Arc<graph::DiskANNIndex<DP>>,
33 queries: Arc<Matrix<T>>,
34 strategy: Strategy<S>,
35 labels: Arc<[Arc<dyn labeled::QueryLabelProvider<DP::InternalId>>]>,
36}
37
38impl<DP, T, S> MultiHop<DP, T, S>
39where
40 DP: provider::DataProvider,
41{
42 pub fn new(
62 index: Arc<graph::DiskANNIndex<DP>>,
63 queries: Arc<Matrix<T>>,
64 strategy: Strategy<S>,
65 labels: Arc<[Arc<dyn labeled::QueryLabelProvider<DP::InternalId>>]>,
66 ) -> anyhow::Result<Arc<Self>> {
67 strategy.length_compatible(queries.nrows())?;
68
69 if labels.len() != queries.nrows() {
70 Err(anyhow::anyhow!(
71 "Number of label providers ({}) must be equal to the number of queries ({})",
72 labels.len(),
73 queries.nrows()
74 ))
75 } else {
76 Ok(Arc::new(Self {
77 index,
78 queries,
79 strategy,
80 labels,
81 }))
82 }
83 }
84}
85
86impl<DP, T, S> Search for MultiHop<DP, T, S>
87where
88 DP: provider::DataProvider<Context: Default, ExternalId: search::Id>,
89 S: for<'a> glue::DefaultSearchStrategy<
90 'a,
91 DP,
92 &'a [T],
93 DP::ExternalId,
94 SearchAccessor: glue::SearchAccessor,
95 > + Clone
96 + AsyncFriendly,
97 T: AsyncFriendly + Clone,
98{
99 type Id = DP::ExternalId;
100 type Parameters = graph::search::Knn;
101 type Output = super::knn::Metrics;
102
103 fn num_queries(&self) -> usize {
104 self.queries.nrows()
105 }
106
107 fn id_count(&self, parameters: &Self::Parameters) -> search::IdCount {
108 search::IdCount::Fixed(parameters.k_value())
109 }
110
111 async fn search<O>(
112 &self,
113 parameters: &Self::Parameters,
114 buffer: &mut O,
115 index: usize,
116 ) -> ANNResult<Self::Output>
117 where
118 O: graph::SearchOutputBuffer<DP::ExternalId> + Send,
119 {
120 let context = DP::Context::default();
121 let multihop_search = graph::search::MultihopFilterSearch::new(*parameters);
122 let strategy =
123 labeled::Filtered::new(self.strategy.get(index)?.clone(), &*self.labels[index]);
124 let stats = self
125 .index
126 .search(
127 multihop_search,
128 &strategy,
129 &context,
130 self.queries.row(index),
131 buffer,
132 )
133 .await?;
134
135 Ok(super::knn::Metrics {
136 comparisons: stats.cmps,
137 hops: stats.hops,
138 })
139 }
140}
141
142#[cfg(test)]
147mod tests {
148 use std::num::NonZeroUsize;
149
150 use super::*;
151
152 use crate::recall::GroundTruthMode;
153 use diskann::graph::{ext::labeled::QueryLabelProvider, test::provider};
154
155 #[derive(Debug)]
157 struct NoOdds;
158
159 impl labeled::QueryLabelProvider<u32> for NoOdds {
160 fn is_match(&self, id: u32) -> bool {
161 id.is_multiple_of(2)
162 }
163 }
164
165 #[test]
166 fn test_multihop() {
167 let nearest_neighbors = 5;
168
169 let index = search::graph::test_grid_provider();
170
171 let mut queries = Matrix::new(0.0f32, 5, index.provider().dim());
172 queries.row_mut(0).copy_from_slice(&[0.0, 0.0, 0.0, 0.0]);
173 queries.row_mut(1).copy_from_slice(&[4.0, 0.0, 0.0, 0.0]);
174 queries.row_mut(2).copy_from_slice(&[0.0, 4.0, 0.0, 0.0]);
175 queries.row_mut(3).copy_from_slice(&[0.0, 0.0, 4.0, 0.0]);
176 queries.row_mut(4).copy_from_slice(&[0.0, 0.0, 0.0, 4.0]);
177
178 let queries = Arc::new(queries);
179
180 let multihop = MultiHop::new(
181 index,
182 queries.clone(),
183 Strategy::broadcast(provider::Strategy::new()),
184 (0..queries.nrows())
185 .map(|_| -> Arc<dyn QueryLabelProvider<_>> { Arc::new(NoOdds {}) })
186 .collect(),
187 )
188 .unwrap();
189
190 let rt = crate::tokio::runtime(2).unwrap();
192 let results = search::search(
193 multihop.clone(),
194 graph::search::Knn::new(nearest_neighbors, 10, None).unwrap(),
195 NonZeroUsize::new(2).unwrap(),
196 &rt,
197 )
198 .unwrap();
199
200 assert_eq!(results.len(), queries.nrows());
201 let rows = results.ids().as_rows();
202 assert_eq!(*rows.row(0).first().unwrap(), 0);
203
204 for r in 0..rows.nrows() {
206 assert_eq!(rows.row(r).len(), nearest_neighbors);
207 for &id in rows.row(r) {
208 assert_eq!(id % 2, 0, "Found odd ID {} in row {}", id, r);
209 }
210 }
211
212 const TWO: NonZeroUsize = NonZeroUsize::new(2).unwrap();
213 let setup = search::Setup {
214 threads: TWO,
215 tasks: TWO,
216 reps: TWO,
217 };
218
219 let parameters = [
221 search::Run::new(
222 graph::search::Knn::new(nearest_neighbors, 10, None).unwrap(),
223 setup.clone(),
224 ),
225 search::Run::new(
226 graph::search::Knn::new(nearest_neighbors, 15, None).unwrap(),
227 setup.clone(),
228 ),
229 ];
230
231 let recall_k = nearest_neighbors;
232 let recall_n = nearest_neighbors;
233
234 let all = search::search_all(
235 multihop,
236 parameters,
237 search::graph::knn::Aggregator::new(
238 rows,
239 recall_k,
240 recall_n,
241 GroundTruthMode::Flexible,
242 ),
243 )
244 .unwrap();
245
246 assert_eq!(all.len(), 2);
247 for summary in all {
248 assert_eq!(summary.setup, setup);
249 assert_eq!(summary.end_to_end_latencies.len(), TWO.get());
250 assert_eq!(summary.mean_latencies.len(), TWO.get());
251 assert_eq!(summary.p90_latencies.len(), TWO.get());
252 assert_eq!(summary.p99_latencies.len(), TWO.get());
253
254 assert_ne!(summary.mean_cmps, 0.0);
255 assert_ne!(summary.mean_hops, 0.0);
256
257 let recall = summary.recall;
258 assert_eq!(recall.recall_k, recall_k);
259 assert_eq!(recall.recall_n, recall_n);
260 assert_eq!(recall.num_queries, queries.nrows());
261 assert_eq!(recall.average, 1.0, "we used a search as the groundtruth");
262 }
263 }
264
265 #[test]
266 fn test_multihop_error() {
267 let index = search::graph::test_grid_provider();
268 let queries = Arc::new(Matrix::new(0.0f32, 2, index.provider().dim()));
269
270 let labels: Arc<[_]> = (0..queries.nrows() + 1)
271 .map(|_| -> Arc<dyn QueryLabelProvider<_>> { Arc::new(NoOdds {}) })
272 .collect();
273
274 let strategy = provider::Strategy::new();
275
276 let err = MultiHop::new(
278 index.clone(),
279 queries.clone(),
280 Strategy::collection([strategy.clone()]),
281 labels.clone(),
282 )
283 .unwrap_err();
284 let msg = err.to_string();
285 assert!(
286 msg.contains("1 strategy was provided when 2 were expected"),
287 "failed with {msg}"
288 );
289
290 let err = MultiHop::new(
292 index,
293 queries.clone(),
294 Strategy::broadcast(strategy.clone()),
295 labels.clone(),
296 )
297 .unwrap_err();
298 let msg = err.to_string();
299 assert!(
300 msg.contains(
301 "Number of label providers (3) must be equal to the number of queries (2)"
302 ),
303 "failed with {msg}"
304 );
305 }
306}