ipfrs_semantic/index_merger.rs
1//! # Index Merger
2//!
3//! Merges multiple partial HNSW embedding indexes from distributed nodes into a
4//! unified index, handling deduplication and conflict resolution by cosine distance.
5//!
6//! ## Overview
7//!
8//! - [`MergeConflict`] — describes conflicts detected during merging
9//! - [`ShardEntry`] — a single vector entry within an index shard
10//! - [`IndexShard`] — a partial index shard from a distributed node
11//! - [`MergeStats`] — statistics produced by a merge operation
12//! - [`MergeConfig`] — configuration for the merge process
13//! - [`EmbeddingIndexMerger`] — merges multiple shards into a unified index
14
15use std::collections::HashMap;
16
17// ---------------------------------------------------------------------------
18// MergeConflict
19// ---------------------------------------------------------------------------
20
21/// Describes a conflict detected while merging index shards.
22#[derive(Debug, Clone, PartialEq)]
23pub enum MergeConflict {
24 /// Same embedding ID appears in multiple shards with meaningfully different vectors.
25 ///
26 /// `score_diff` is the cosine distance between the two conflicting vectors.
27 DuplicateId {
28 /// The embedding ID that appears in more than one shard.
29 id: u64,
30 /// Cosine distance between the existing and the incoming vector.
31 score_diff: f32,
32 },
33
34 /// An entry's vector dimension does not match the declared shard dimension.
35 DimensionMismatch {
36 /// The dimension declared by the shard.
37 expected: usize,
38 /// The actual dimension of the offending entry's vector.
39 actual: usize,
40 },
41
42 /// Adding further entries would exceed the configured capacity limit.
43 CapacityExceeded {
44 /// The maximum number of entries allowed.
45 limit: usize,
46 },
47}
48
49impl std::fmt::Display for MergeConflict {
50 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
51 match self {
52 Self::DuplicateId { id, score_diff } => {
53 write!(
54 f,
55 "duplicate embedding id {id} with cosine distance {score_diff:.6}"
56 )
57 }
58 Self::DimensionMismatch { expected, actual } => {
59 write!(f, "dimension mismatch: expected {expected}, got {actual}")
60 }
61 Self::CapacityExceeded { limit } => {
62 write!(f, "merged index would exceed capacity limit of {limit}")
63 }
64 }
65 }
66}
67
68impl std::error::Error for MergeConflict {}
69
70// ---------------------------------------------------------------------------
71// ShardEntry
72// ---------------------------------------------------------------------------
73
74/// A single vector entry stored within an [`IndexShard`].
75#[derive(Debug, Clone, PartialEq)]
76pub struct ShardEntry {
77 /// Globally unique identifier for this embedding.
78 pub id: u64,
79 /// The raw embedding vector.
80 pub vector: Vec<f32>,
81 /// Arbitrary string tag attached to this entry (e.g., document reference).
82 pub metadata: String,
83}
84
85impl ShardEntry {
86 /// Creates a new `ShardEntry`.
87 pub fn new(id: u64, vector: Vec<f32>, metadata: impl Into<String>) -> Self {
88 Self {
89 id,
90 vector,
91 metadata: metadata.into(),
92 }
93 }
94}
95
96// ---------------------------------------------------------------------------
97// IndexShard
98// ---------------------------------------------------------------------------
99
100/// A partial HNSW embedding index from a single distributed node.
101#[derive(Debug, Clone)]
102pub struct IndexShard {
103 /// Human-readable identifier for this shard (e.g., node address or UUID).
104 pub shard_id: String,
105 /// All vector entries stored in this shard.
106 pub entries: Vec<ShardEntry>,
107 /// Expected dimensionality of every vector in `entries`.
108 pub dimension: usize,
109}
110
111impl IndexShard {
112 /// Creates a new `IndexShard`.
113 pub fn new(shard_id: impl Into<String>, entries: Vec<ShardEntry>, dimension: usize) -> Self {
114 Self {
115 shard_id: shard_id.into(),
116 entries,
117 dimension,
118 }
119 }
120
121 /// Validates that every entry's vector has the declared dimension.
122 ///
123 /// Returns `Err(MergeConflict::DimensionMismatch)` on the first offending entry.
124 pub fn validate(&self) -> Result<(), MergeConflict> {
125 for entry in &self.entries {
126 if entry.vector.len() != self.dimension {
127 return Err(MergeConflict::DimensionMismatch {
128 expected: self.dimension,
129 actual: entry.vector.len(),
130 });
131 }
132 }
133 Ok(())
134 }
135}
136
137// ---------------------------------------------------------------------------
138// MergeStats
139// ---------------------------------------------------------------------------
140
141/// Statistics produced by a single [`EmbeddingIndexMerger::merge`] call.
142#[derive(Debug, Clone, Default)]
143pub struct MergeStats {
144 /// Total number of entries across all input shards before deduplication.
145 pub total_input_entries: usize,
146 /// Entries skipped because their ID already existed in the output.
147 pub deduplicated: usize,
148 /// Conflicts recorded during the merge (duplicate IDs beyond the threshold).
149 pub conflicts: Vec<MergeConflict>,
150 /// Number of entries in the resulting merged index.
151 pub output_entries: usize,
152 /// Number of shards that were successfully merged.
153 pub shards_merged: usize,
154}
155
156// ---------------------------------------------------------------------------
157// MergeConfig
158// ---------------------------------------------------------------------------
159
160/// Configuration that governs how [`EmbeddingIndexMerger`] resolves duplicates.
161#[derive(Debug, Clone)]
162pub struct MergeConfig {
163 /// Maximum number of entries in the merged output (default: 100 000).
164 pub max_entries: usize,
165 /// Cosine distance above which a duplicate ID is considered a conflict
166 /// and recorded in [`MergeStats::conflicts`] (default: 0.01).
167 pub conflict_threshold: f32,
168 /// When `true`, the first-seen entry wins on an ID collision.
169 /// When `false`, the later entry replaces the earlier one.
170 pub keep_first: bool,
171}
172
173impl Default for MergeConfig {
174 fn default() -> Self {
175 Self {
176 max_entries: 100_000,
177 conflict_threshold: 0.01,
178 keep_first: true,
179 }
180 }
181}
182
183// ---------------------------------------------------------------------------
184// EmbeddingIndexMerger
185// ---------------------------------------------------------------------------
186
187/// Merges multiple partial HNSW embedding index shards into a unified index.
188///
189/// # Example
190///
191/// ```rust
192/// use ipfrs_semantic::index_merger::{
193/// EmbeddingIndexMerger, IndexShard, MergeConfig, ShardEntry,
194/// };
195///
196/// let config = MergeConfig::default();
197/// let merger = EmbeddingIndexMerger::new(config);
198///
199/// let shard = IndexShard::new(
200/// "node-1",
201/// vec![ShardEntry::new(1, vec![1.0_f32, 0.0], "doc-a")],
202/// 2,
203/// );
204///
205/// let (entries, stats) = merger.merge(&[shard]).expect("merge failed");
206/// assert_eq!(entries.len(), 1);
207/// assert_eq!(stats.output_entries, 1);
208/// ```
209pub struct EmbeddingIndexMerger {
210 /// Configuration for this merger instance.
211 pub config: MergeConfig,
212}
213
214impl EmbeddingIndexMerger {
215 /// Creates a new `EmbeddingIndexMerger` with the supplied configuration.
216 pub fn new(config: MergeConfig) -> Self {
217 Self { config }
218 }
219
220 /// Merges the provided `shards` into a single flat list of [`ShardEntry`] values.
221 ///
222 /// ## Algorithm
223 ///
224 /// 1. Each shard is validated; a dimension mismatch causes an immediate `Err`.
225 /// 2. Entries are walked in shard order. A `HashMap<u64, usize>` maps each seen
226 /// embedding ID to its current position in the output vector.
227 /// 3. On a duplicate ID the cosine distance between the stored vector and the
228 /// incoming vector is computed:
229 /// - Distance > `conflict_threshold` → [`MergeConflict::DuplicateId`] is
230 /// appended to `stats.conflicts`; the new entry is *always* skipped.
231 /// - Distance ≤ `conflict_threshold` → near-identical copy; skipped silently.
232 /// - When `keep_first = false` the existing slot is overwritten regardless of
233 /// whether the conflict was recorded.
234 /// 4. If adding a new (non-duplicate) entry would push the output beyond
235 /// `max_entries`, `Err(CapacityExceeded)` is returned immediately.
236 ///
237 /// Returns `(merged_entries, stats)` on success.
238 pub fn merge(
239 &self,
240 shards: &[IndexShard],
241 ) -> Result<(Vec<ShardEntry>, MergeStats), MergeConflict> {
242 // Validate every shard upfront.
243 for shard in shards {
244 shard.validate()?;
245 }
246
247 let mut output: Vec<ShardEntry> = Vec::new();
248 // Maps embedding id → index into `output`.
249 let mut id_index: HashMap<u64, usize> = HashMap::new();
250
251 let mut stats = MergeStats {
252 total_input_entries: shards.iter().map(|s| s.entries.len()).sum(),
253 shards_merged: shards.len(),
254 ..Default::default()
255 };
256
257 for shard in shards {
258 for entry in &shard.entries {
259 if let Some(&existing_idx) = id_index.get(&entry.id) {
260 // --- duplicate ID ---
261 stats.deduplicated += 1;
262
263 let dist = Self::cosine_distance(&output[existing_idx].vector, &entry.vector);
264
265 if dist > self.config.conflict_threshold {
266 stats.conflicts.push(MergeConflict::DuplicateId {
267 id: entry.id,
268 score_diff: dist,
269 });
270 }
271
272 // When keep_first=false we overwrite the existing slot.
273 if !self.config.keep_first {
274 output[existing_idx] = entry.clone();
275 }
276 } else {
277 // --- new entry ---
278 if output.len() >= self.config.max_entries {
279 return Err(MergeConflict::CapacityExceeded {
280 limit: self.config.max_entries,
281 });
282 }
283 let idx = output.len();
284 id_index.insert(entry.id, idx);
285 output.push(entry.clone());
286 }
287 }
288 }
289
290 stats.output_entries = output.len();
291
292 Ok((output, stats))
293 }
294
295 /// Computes the cosine distance between two vectors.
296 ///
297 /// `cosine_distance = 1.0 - cosine_similarity`
298 ///
299 /// Returns `1.0` when either vector has zero norm (undefined similarity).
300 pub fn cosine_distance(a: &[f32], b: &[f32]) -> f32 {
301 let dot: f32 = a.iter().zip(b.iter()).map(|(x, y)| x * y).sum();
302 let norm_a: f32 = a.iter().map(|x| x * x).sum::<f32>().sqrt();
303 let norm_b: f32 = b.iter().map(|x| x * x).sum::<f32>().sqrt();
304
305 if norm_a == 0.0 || norm_b == 0.0 {
306 return 1.0;
307 }
308
309 let similarity = dot / (norm_a * norm_b);
310 // Clamp to [-1, 1] to guard against floating-point rounding past the boundary.
311 let similarity = similarity.clamp(-1.0_f32, 1.0_f32);
312 1.0 - similarity
313 }
314
315 /// Returns a list of `(shard_id, entry_count)` pairs sorted by entry count
316 /// in **descending** order.
317 pub fn shard_coverage(shards: &[IndexShard]) -> Vec<(String, usize)> {
318 let mut coverage: Vec<(String, usize)> = shards
319 .iter()
320 .map(|s| (s.shard_id.clone(), s.entries.len()))
321 .collect();
322 coverage.sort_by_key(|a| std::cmp::Reverse(a.1));
323 coverage
324 }
325}
326
327// ---------------------------------------------------------------------------
328// Tests
329// ---------------------------------------------------------------------------
330
331#[cfg(test)]
332mod tests {
333 use super::*;
334
335 // -----------------------------------------------------------------------
336 // Helper builders
337 // -----------------------------------------------------------------------
338
339 fn make_entry(id: u64, vector: Vec<f32>) -> ShardEntry {
340 ShardEntry::new(id, vector, format!("meta-{id}"))
341 }
342
343 fn make_shard(shard_id: &str, entries: Vec<ShardEntry>, dimension: usize) -> IndexShard {
344 IndexShard::new(shard_id, entries, dimension)
345 }
346
347 fn default_merger() -> EmbeddingIndexMerger {
348 EmbeddingIndexMerger::new(MergeConfig::default())
349 }
350
351 // -----------------------------------------------------------------------
352 // 1. new() with config
353 // -----------------------------------------------------------------------
354 #[test]
355 fn test_new_with_config() {
356 let config = MergeConfig {
357 max_entries: 500,
358 conflict_threshold: 0.05,
359 keep_first: false,
360 };
361 let merger = EmbeddingIndexMerger::new(config.clone());
362 assert_eq!(merger.config.max_entries, 500);
363 assert!((merger.config.conflict_threshold - 0.05).abs() < 1e-6);
364 assert!(!merger.config.keep_first);
365 }
366
367 // -----------------------------------------------------------------------
368 // 2. merge empty shards returns empty output
369 // -----------------------------------------------------------------------
370 #[test]
371 fn test_merge_empty_shards() {
372 let merger = default_merger();
373 let (entries, stats) = merger.merge(&[]).expect("merge should succeed");
374 assert!(entries.is_empty());
375 assert_eq!(stats.output_entries, 0);
376 assert_eq!(stats.total_input_entries, 0);
377 assert_eq!(stats.shards_merged, 0);
378 }
379
380 // -----------------------------------------------------------------------
381 // 3. merge single shard passes through
382 // -----------------------------------------------------------------------
383 #[test]
384 fn test_merge_single_shard_passthrough() {
385 let merger = default_merger();
386 let shard = make_shard(
387 "s1",
388 vec![make_entry(1, vec![1.0, 0.0]), make_entry(2, vec![0.0, 1.0])],
389 2,
390 );
391 let (entries, stats) = merger.merge(&[shard]).expect("merge should succeed");
392 assert_eq!(entries.len(), 2);
393 assert_eq!(stats.output_entries, 2);
394 assert_eq!(stats.total_input_entries, 2);
395 assert_eq!(stats.shards_merged, 1);
396 assert_eq!(stats.deduplicated, 0);
397 }
398
399 // -----------------------------------------------------------------------
400 // 4. merge two non-overlapping shards combines all
401 // -----------------------------------------------------------------------
402 #[test]
403 fn test_merge_two_non_overlapping_shards() {
404 let merger = default_merger();
405 let s1 = make_shard("s1", vec![make_entry(1, vec![1.0, 0.0])], 2);
406 let s2 = make_shard("s2", vec![make_entry(2, vec![0.0, 1.0])], 2);
407 let (entries, stats) = merger.merge(&[s1, s2]).expect("merge should succeed");
408 assert_eq!(entries.len(), 2);
409 assert_eq!(stats.output_entries, 2);
410 assert_eq!(stats.total_input_entries, 2);
411 assert_eq!(stats.shards_merged, 2);
412 assert_eq!(stats.deduplicated, 0);
413 assert!(stats.conflicts.is_empty());
414 }
415
416 // -----------------------------------------------------------------------
417 // 5. duplicate ID keep_first=true keeps original
418 // -----------------------------------------------------------------------
419 #[test]
420 fn test_duplicate_keep_first_true() {
421 let config = MergeConfig {
422 keep_first: true,
423 conflict_threshold: 0.01,
424 ..Default::default()
425 };
426 let merger = EmbeddingIndexMerger::new(config);
427
428 let first_vec = vec![1.0_f32, 0.0];
429 let second_vec = vec![0.0_f32, 1.0]; // orthogonal — distance = 1.0 > threshold
430
431 let s1 = make_shard(
432 "s1",
433 vec![ShardEntry::new(42, first_vec.clone(), "first")],
434 2,
435 );
436 let s2 = make_shard(
437 "s2",
438 vec![ShardEntry::new(42, second_vec.clone(), "second")],
439 2,
440 );
441
442 let (entries, stats) = merger.merge(&[s1, s2]).expect("merge should succeed");
443 assert_eq!(entries.len(), 1);
444 assert_eq!(entries[0].vector, first_vec, "should keep first entry");
445 assert_eq!(stats.deduplicated, 1);
446 }
447
448 // -----------------------------------------------------------------------
449 // 6. duplicate ID keep_first=false replaces with newer
450 // -----------------------------------------------------------------------
451 #[test]
452 fn test_duplicate_keep_first_false() {
453 let config = MergeConfig {
454 keep_first: false,
455 conflict_threshold: 0.01,
456 ..Default::default()
457 };
458 let merger = EmbeddingIndexMerger::new(config);
459
460 let first_vec = vec![1.0_f32, 0.0];
461 let second_vec = vec![0.0_f32, 1.0]; // orthogonal — high distance
462
463 let s1 = make_shard(
464 "s1",
465 vec![ShardEntry::new(42, first_vec.clone(), "first")],
466 2,
467 );
468 let s2 = make_shard(
469 "s2",
470 vec![ShardEntry::new(42, second_vec.clone(), "second")],
471 2,
472 );
473
474 let (entries, _stats) = merger.merge(&[s1, s2]).expect("merge should succeed");
475 assert_eq!(entries.len(), 1);
476 assert_eq!(entries[0].vector, second_vec, "should keep last entry");
477 }
478
479 // -----------------------------------------------------------------------
480 // 7. near-identical duplicate (distance < threshold) skipped silently
481 // -----------------------------------------------------------------------
482 #[test]
483 fn test_near_identical_duplicate_skipped_silently() {
484 let config = MergeConfig {
485 conflict_threshold: 0.05, // generous threshold
486 keep_first: true,
487 ..Default::default()
488 };
489 let merger = EmbeddingIndexMerger::new(config);
490
491 // Two nearly identical unit vectors (tiny perturbation — well within threshold).
492 let v1 = vec![1.0_f32, 0.0001];
493 let v2 = vec![1.0_f32, 0.0001]; // identical → distance = 0.0
494
495 let s1 = make_shard("s1", vec![ShardEntry::new(7, v1, "a")], 2);
496 let s2 = make_shard("s2", vec![ShardEntry::new(7, v2, "b")], 2);
497
498 let (entries, stats) = merger.merge(&[s1, s2]).expect("merge should succeed");
499 assert_eq!(entries.len(), 1);
500 assert_eq!(stats.deduplicated, 1);
501 assert!(
502 stats.conflicts.is_empty(),
503 "no conflict should be recorded for near-identical vectors"
504 );
505 }
506
507 // -----------------------------------------------------------------------
508 // 8. conflicting duplicate (distance > threshold) recorded in stats
509 // -----------------------------------------------------------------------
510 #[test]
511 fn test_conflicting_duplicate_recorded() {
512 let config = MergeConfig {
513 conflict_threshold: 0.01,
514 keep_first: true,
515 ..Default::default()
516 };
517 let merger = EmbeddingIndexMerger::new(config);
518
519 let v1 = vec![1.0_f32, 0.0];
520 let v2 = vec![0.0_f32, 1.0]; // orthogonal → distance = 1.0
521
522 let s1 = make_shard("s1", vec![ShardEntry::new(99, v1, "a")], 2);
523 let s2 = make_shard("s2", vec![ShardEntry::new(99, v2, "b")], 2);
524
525 let (entries, stats) = merger.merge(&[s1, s2]).expect("merge should succeed");
526 assert_eq!(entries.len(), 1);
527 assert_eq!(stats.deduplicated, 1);
528 assert_eq!(stats.conflicts.len(), 1);
529
530 match &stats.conflicts[0] {
531 MergeConflict::DuplicateId { id, score_diff } => {
532 assert_eq!(*id, 99);
533 assert!(
534 (*score_diff - 1.0).abs() < 1e-5,
535 "expected ~1.0, got {score_diff}"
536 );
537 }
538 other => panic!("unexpected conflict variant: {other:?}"),
539 }
540 }
541
542 // -----------------------------------------------------------------------
543 // 9. dimension mismatch returns Err(DimensionMismatch)
544 // -----------------------------------------------------------------------
545 #[test]
546 fn test_dimension_mismatch_returns_err() {
547 let merger = default_merger();
548 // Shard claims dimension=3 but entry has dimension=2.
549 let bad_shard = make_shard("bad", vec![make_entry(1, vec![1.0, 0.0])], 3);
550 let result = merger.merge(&[bad_shard]);
551 match result {
552 Err(MergeConflict::DimensionMismatch { expected, actual }) => {
553 assert_eq!(expected, 3);
554 assert_eq!(actual, 2);
555 }
556 other => panic!("expected DimensionMismatch, got {other:?}"),
557 }
558 }
559
560 // -----------------------------------------------------------------------
561 // 10. capacity exceeded returns Err(CapacityExceeded)
562 // -----------------------------------------------------------------------
563 #[test]
564 fn test_capacity_exceeded() {
565 let config = MergeConfig {
566 max_entries: 2,
567 ..Default::default()
568 };
569 let merger = EmbeddingIndexMerger::new(config);
570
571 let entries: Vec<ShardEntry> = (1..=3)
572 .map(|i| make_entry(i, vec![i as f32, 0.0]))
573 .collect();
574 let shard = make_shard("big", entries, 2);
575
576 match merger.merge(&[shard]) {
577 Err(MergeConflict::CapacityExceeded { limit }) => assert_eq!(limit, 2),
578 other => panic!("expected CapacityExceeded, got {other:?}"),
579 }
580 }
581
582 // -----------------------------------------------------------------------
583 // 11. cosine_distance identical vectors = 0.0
584 // -----------------------------------------------------------------------
585 #[test]
586 fn test_cosine_distance_identical() {
587 let v = vec![1.0_f32, 2.0, 3.0];
588 let dist = EmbeddingIndexMerger::cosine_distance(&v, &v);
589 assert!(
590 dist.abs() < 1e-6,
591 "identical vectors → distance 0.0, got {dist}"
592 );
593 }
594
595 // -----------------------------------------------------------------------
596 // 12. cosine_distance orthogonal vectors = 1.0
597 // -----------------------------------------------------------------------
598 #[test]
599 fn test_cosine_distance_orthogonal() {
600 let a = vec![1.0_f32, 0.0];
601 let b = vec![0.0_f32, 1.0];
602 let dist = EmbeddingIndexMerger::cosine_distance(&a, &b);
603 assert!(
604 (dist - 1.0).abs() < 1e-6,
605 "orthogonal vectors → distance 1.0, got {dist}"
606 );
607 }
608
609 // -----------------------------------------------------------------------
610 // 13. cosine_distance zero vector = 1.0
611 // -----------------------------------------------------------------------
612 #[test]
613 fn test_cosine_distance_zero_vector() {
614 let a = vec![1.0_f32, 0.0];
615 let zero = vec![0.0_f32, 0.0];
616 let dist = EmbeddingIndexMerger::cosine_distance(&a, &zero);
617 assert!(
618 (dist - 1.0).abs() < 1e-6,
619 "zero vector → distance 1.0, got {dist}"
620 );
621
622 let dist2 = EmbeddingIndexMerger::cosine_distance(&zero, &a);
623 assert!((dist2 - 1.0).abs() < 1e-6);
624 }
625
626 // -----------------------------------------------------------------------
627 // 14. validate() catches wrong-dimension entry
628 // -----------------------------------------------------------------------
629 #[test]
630 fn test_validate_catches_wrong_dimension() {
631 let shard = make_shard(
632 "s",
633 vec![
634 make_entry(1, vec![1.0, 0.0, 0.0]),
635 make_entry(2, vec![0.5, 0.5]), // wrong: dim=2 instead of 3
636 ],
637 3,
638 );
639 match shard.validate() {
640 Err(MergeConflict::DimensionMismatch { expected, actual }) => {
641 assert_eq!(expected, 3);
642 assert_eq!(actual, 2);
643 }
644 other => panic!("expected DimensionMismatch, got {other:?}"),
645 }
646 }
647
648 // -----------------------------------------------------------------------
649 // 15. validate() passes for correct dimensions
650 // -----------------------------------------------------------------------
651 #[test]
652 fn test_validate_passes_for_correct_dimensions() {
653 let shard = make_shard(
654 "s",
655 vec![make_entry(1, vec![1.0, 0.0]), make_entry(2, vec![0.0, 1.0])],
656 2,
657 );
658 assert!(shard.validate().is_ok());
659 }
660
661 // -----------------------------------------------------------------------
662 // 16. shard_coverage sorted descending
663 // -----------------------------------------------------------------------
664 #[test]
665 fn test_shard_coverage_sorted_descending() {
666 let s1 = make_shard(
667 "large",
668 vec![
669 make_entry(1, vec![1.0]),
670 make_entry(2, vec![0.5]),
671 make_entry(3, vec![0.1]),
672 ],
673 1,
674 );
675 let s2 = make_shard("small", vec![make_entry(4, vec![0.9])], 1);
676 let s3 = make_shard(
677 "medium",
678 vec![make_entry(5, vec![0.8]), make_entry(6, vec![0.7])],
679 1,
680 );
681
682 let coverage = EmbeddingIndexMerger::shard_coverage(&[s1, s2, s3]);
683 assert_eq!(coverage.len(), 3);
684 // Should be large(3) > medium(2) > small(1)
685 assert_eq!(coverage[0].0, "large");
686 assert_eq!(coverage[0].1, 3);
687 assert_eq!(coverage[1].0, "medium");
688 assert_eq!(coverage[1].1, 2);
689 assert_eq!(coverage[2].0, "small");
690 assert_eq!(coverage[2].1, 1);
691 }
692
693 // -----------------------------------------------------------------------
694 // 17. stats.deduplicated counted correctly
695 // -----------------------------------------------------------------------
696 #[test]
697 fn test_stats_deduplicated_counted() {
698 let merger = default_merger();
699
700 // 3 shards, each with id=1 and id=2 → 4 duplicates total.
701 let make_pair = |tag: &str| {
702 make_shard(
703 tag,
704 vec![
705 ShardEntry::new(1, vec![1.0, 0.0], "a"),
706 ShardEntry::new(2, vec![0.0, 1.0], "b"),
707 ],
708 2,
709 )
710 };
711
712 let shards = vec![make_pair("s1"), make_pair("s2"), make_pair("s3")];
713 let (entries, stats) = merger.merge(&shards).expect("merge should succeed");
714
715 // Only 2 unique entries.
716 assert_eq!(entries.len(), 2);
717 // Shards s2 and s3 each contribute 2 duplicates → 4 total.
718 assert_eq!(stats.deduplicated, 4);
719 assert_eq!(stats.total_input_entries, 6);
720 assert_eq!(stats.output_entries, 2);
721 }
722
723 // -----------------------------------------------------------------------
724 // 18. stats.shards_merged correct
725 // -----------------------------------------------------------------------
726 #[test]
727 fn test_stats_shards_merged_correct() {
728 let merger = default_merger();
729
730 let shards: Vec<IndexShard> = (0..5)
731 .map(|i| {
732 make_shard(
733 &format!("shard-{i}"),
734 vec![make_entry(i, vec![i as f32, 0.0])],
735 2,
736 )
737 })
738 .collect();
739
740 let (entries, stats) = merger.merge(&shards).expect("merge should succeed");
741 assert_eq!(stats.shards_merged, 5);
742 assert_eq!(entries.len(), 5);
743 assert_eq!(stats.output_entries, 5);
744 }
745}