1use ndarray::{s, Array1, Array2, ArrayView1, ArrayView2, Axis};
4
5use crate::error::{Error, Result};
6
7const DEFAULT_MAX_NEAREST_CENTROID_MEMORY: usize = 1024 * 1024 * 1024; fn max_nearest_centroid_memory() -> usize {
14 std::env::var("NEXT_PLAID_MAX_NEAREST_CENTROID_MEMORY_MB")
15 .ok()
16 .and_then(|v| v.parse::<usize>().ok())
17 .filter(|&mb| mb > 0)
18 .map(|mb| mb.saturating_mul(1024 * 1024))
19 .unwrap_or(DEFAULT_MAX_NEAREST_CENTROID_MEMORY)
20}
21
22#[inline]
23fn cmp_f32_for_max(a: &f32, b: &f32) -> std::cmp::Ordering {
24 match (a.is_finite(), b.is_finite()) {
25 (true, true) => a.total_cmp(b),
26 (true, false) => std::cmp::Ordering::Greater,
27 (false, true) => std::cmp::Ordering::Less,
28 (false, false) => std::cmp::Ordering::Equal,
29 }
30}
31
32pub enum CentroidStore {
38 Owned(Array2<f32>),
40 Mmap(crate::mmap::MmapNpyArray2F32),
42}
43
44impl CentroidStore {
45 pub fn view(&self) -> ArrayView2<'_, f32> {
49 match self {
50 CentroidStore::Owned(arr) => arr.view(),
51 CentroidStore::Mmap(mmap) => mmap.view(),
52 }
53 }
54
55 pub fn nrows(&self) -> usize {
57 match self {
58 CentroidStore::Owned(arr) => arr.nrows(),
59 CentroidStore::Mmap(mmap) => mmap.nrows(),
60 }
61 }
62
63 pub fn ncols(&self) -> usize {
65 match self {
66 CentroidStore::Owned(arr) => arr.ncols(),
67 CentroidStore::Mmap(mmap) => mmap.ncols(),
68 }
69 }
70
71 pub fn row(&self, idx: usize) -> ArrayView1<'_, f32> {
73 match self {
74 CentroidStore::Owned(arr) => arr.row(idx),
75 CentroidStore::Mmap(mmap) => mmap.row(idx),
76 }
77 }
78
79 pub fn slice_rows(&self, start: usize, end: usize) -> ArrayView2<'_, f32> {
83 match self {
84 CentroidStore::Owned(arr) => arr.slice(s![start..end, ..]),
85 CentroidStore::Mmap(mmap) => mmap.slice_rows(start, end),
86 }
87 }
88}
89
90impl Clone for CentroidStore {
91 fn clone(&self) -> Self {
92 match self {
93 CentroidStore::Owned(arr) => CentroidStore::Owned(arr.clone()),
95 CentroidStore::Mmap(mmap) => CentroidStore::Owned(mmap.to_owned()),
97 }
98 }
99}
100
101#[derive(Clone)]
107pub struct ResidualCodec {
108 pub nbits: usize,
110 pub centroids: CentroidStore,
113 pub avg_residual: Array1<f32>,
115 pub bucket_cutoffs: Option<Array1<f32>>,
117 pub bucket_weights: Option<Array1<f32>>,
119 pub byte_reversed_bits_map: Vec<u8>,
121 pub bucket_weight_indices_lookup: Option<Array2<usize>>,
123}
124
125impl ResidualCodec {
126 pub fn new(
136 nbits: usize,
137 centroids: Array2<f32>,
138 avg_residual: Array1<f32>,
139 bucket_cutoffs: Option<Array1<f32>>,
140 bucket_weights: Option<Array1<f32>>,
141 ) -> Result<Self> {
142 Self::new_with_store(
143 nbits,
144 CentroidStore::Owned(centroids),
145 avg_residual,
146 bucket_cutoffs,
147 bucket_weights,
148 )
149 }
150
151 pub fn new_with_store(
155 nbits: usize,
156 centroids: CentroidStore,
157 avg_residual: Array1<f32>,
158 bucket_cutoffs: Option<Array1<f32>>,
159 bucket_weights: Option<Array1<f32>>,
160 ) -> Result<Self> {
161 if nbits == 0 || 8 % nbits != 0 {
162 return Err(Error::Codec(format!(
163 "nbits must be a divisor of 8, got {}",
164 nbits
165 )));
166 }
167
168 let nbits_mask = (1u32 << nbits) - 1;
170 let mut byte_reversed_bits_map = vec![0u8; 256];
171
172 for (i, byte_slot) in byte_reversed_bits_map.iter_mut().enumerate() {
173 let val = i as u32;
174 let mut out = 0u32;
175 let mut pos = 8i32;
176
177 while pos >= nbits as i32 {
178 let segment = (val >> (pos as u32 - nbits as u32)) & nbits_mask;
179
180 let mut rev_segment = 0u32;
181 for k in 0..nbits {
182 if (segment & (1 << k)) != 0 {
183 rev_segment |= 1 << (nbits - 1 - k);
184 }
185 }
186
187 out |= rev_segment;
188
189 if pos > nbits as i32 {
190 out <<= nbits;
191 }
192
193 pos -= nbits as i32;
194 }
195 *byte_slot = out as u8;
196 }
197
198 let keys_per_byte = 8 / nbits;
200 let bucket_weight_indices_lookup = if bucket_weights.is_some() {
201 let mask = (1usize << nbits) - 1;
202 let mut table = Array2::<usize>::zeros((256, keys_per_byte));
203
204 for byte_val in 0..256usize {
205 for k in (0..keys_per_byte).rev() {
206 let shift = k * nbits;
207 let index = (byte_val >> shift) & mask;
208 table[[byte_val, keys_per_byte - 1 - k]] = index;
209 }
210 }
211 Some(table)
212 } else {
213 None
214 };
215
216 Ok(Self {
217 nbits,
218 centroids,
219 avg_residual,
220 bucket_cutoffs,
221 bucket_weights,
222 byte_reversed_bits_map,
223 bucket_weight_indices_lookup,
224 })
225 }
226
227 pub fn embedding_dim(&self) -> usize {
229 self.centroids.ncols()
230 }
231
232 pub fn num_centroids(&self) -> usize {
234 self.centroids.nrows()
235 }
236
237 pub fn centroids_view(&self) -> ArrayView2<'_, f32> {
241 self.centroids.view()
242 }
243
244 pub fn compress_into_codes(&self, embeddings: &Array2<f32>) -> Array1<usize> {
261 #[cfg(feature = "_cuda")]
263 {
264 let force_gpu = crate::is_force_gpu();
265 if let Some(ctx) = crate::cuda::get_global_context() {
266 let centroids = self.centroids_view();
267 match crate::cuda::compress_into_codes_cuda_batched(
268 &ctx,
269 &embeddings.view(),
270 ¢roids,
271 None,
272 ) {
273 Ok(codes) => return codes,
274 Err(e) => {
275 if force_gpu {
276 panic!(
277 "FORCE_GPU is set but CUDA compress_into_codes failed: {}",
278 e
279 );
280 }
281 eprintln!(
282 "[next-plaid] CUDA compression error: {}. Falling back to CPU.",
283 e
284 );
285 }
286 }
287 } else if force_gpu {
288 panic!("FORCE_GPU is set but CUDA context is unavailable");
289 }
290 }
291
292 self.compress_into_codes_cpu(embeddings)
293 }
294
295 pub fn compress_into_codes_cpu(&self, embeddings: &Array2<f32>) -> Array1<usize> {
298 use rayon::prelude::*;
299
300 let n = embeddings.nrows();
301 if n == 0 {
302 return Array1::zeros(0);
303 }
304
305 let centroids = self.centroids_view();
307 let num_centroids = centroids.nrows();
308
309 let max_batch_by_memory =
313 max_nearest_centroid_memory() / (num_centroids * std::mem::size_of::<f32>());
314 let batch_size = max_batch_by_memory.clamp(1, 1024);
315 let batch_ranges: Vec<(usize, usize)> = (0..n)
316 .step_by(batch_size)
317 .map(|start| (start, (start + batch_size).min(n)))
318 .collect();
319
320 let chunked_codes: Vec<Vec<usize>> = batch_ranges
321 .into_par_iter()
322 .map(|(start, end)| {
323 let batch = embeddings.slice(ndarray::s![start..end, ..]);
324
325 let scores = batch.dot(¢roids.t());
327
328 scores
330 .axis_iter(Axis(0))
331 .map(|row| {
332 row.iter()
333 .enumerate()
334 .max_by(|(_, a), (_, b)| cmp_f32_for_max(a, b))
335 .map(|(idx, _)| idx)
336 .unwrap_or(0)
337 })
338 .collect()
339 })
340 .collect();
341
342 Array1::from_vec(chunked_codes.into_iter().flatten().collect())
343 }
344
345 pub fn quantize_residuals(&self, residuals: &Array2<f32>) -> Result<Array2<u8>> {
357 use rayon::prelude::*;
358
359 let cutoffs = self
360 .bucket_cutoffs
361 .as_ref()
362 .ok_or_else(|| Error::Codec("bucket_cutoffs required for quantization".into()))?;
363
364 let n = residuals.nrows();
365 let dim = residuals.ncols();
366 let packed_dim = dim * self.nbits / 8;
367 let nbits = self.nbits;
368
369 if n == 0 {
370 return Ok(Array2::zeros((0, packed_dim)));
371 }
372
373 let cutoffs_slice = cutoffs.as_slice().unwrap();
375
376 let packed_rows: Vec<Vec<u8>> = residuals
378 .axis_iter(Axis(0))
379 .into_par_iter()
380 .map(|row| {
381 let mut packed_row = vec![0u8; packed_dim];
382 let mut bit_idx = 0;
383
384 for &val in row.iter() {
385 let bucket = cutoffs_slice.iter().filter(|&&c| val > c).count();
387
388 for b in 0..nbits {
390 let bit = ((bucket >> b) & 1) as u8;
391 let byte_idx = bit_idx / 8;
392 let bit_pos = 7 - (bit_idx % 8);
393 packed_row[byte_idx] |= bit << bit_pos;
394 bit_idx += 1;
395 }
396 }
397
398 packed_row
399 })
400 .collect();
401
402 let mut packed = Array2::<u8>::zeros((n, packed_dim));
404 for (i, row) in packed_rows.into_iter().enumerate() {
405 for (j, val) in row.into_iter().enumerate() {
406 packed[[i, j]] = val;
407 }
408 }
409
410 Ok(packed)
411 }
412
413 pub fn decompress(
424 &self,
425 packed_residuals: &Array2<u8>,
426 codes: &ArrayView1<usize>,
427 ) -> Result<Array2<f32>> {
428 let bucket_weights = self
429 .bucket_weights
430 .as_ref()
431 .ok_or_else(|| Error::Codec("bucket_weights required for decompression".into()))?;
432
433 let lookup = self
434 .bucket_weight_indices_lookup
435 .as_ref()
436 .ok_or_else(|| Error::Codec("bucket_weight_indices_lookup required".into()))?;
437
438 let n = packed_residuals.nrows();
439 let dim = self.embedding_dim();
440
441 let mut output = Array2::<f32>::zeros((n, dim));
442
443 for i in 0..n {
444 let centroid = self.centroids.row(codes[i]);
446
447 let mut residual_idx = 0;
449 for &byte_val in packed_residuals.row(i).iter() {
450 let reversed = self.byte_reversed_bits_map[byte_val as usize];
451 let indices = lookup.row(reversed as usize);
452
453 for &bucket_idx in indices.iter() {
454 if residual_idx < dim {
455 output[[i, residual_idx]] =
456 centroid[residual_idx] + bucket_weights[bucket_idx];
457 residual_idx += 1;
458 }
459 }
460 }
461 }
462
463 for mut row in output.axis_iter_mut(Axis(0)) {
465 let norm = row.dot(&row).sqrt().max(1e-12);
466 row /= norm;
467 }
468
469 Ok(output)
470 }
471
472 pub fn load_from_dir(index_path: &std::path::Path) -> Result<Self> {
474 use ndarray_npy::ReadNpyExt;
475 use std::fs::File;
476
477 let centroids_path = index_path.join("centroids.npy");
478 let centroids: Array2<f32> = Array2::read_npy(
479 File::open(¢roids_path)
480 .map_err(|e| Error::IndexLoad(format!("Failed to open centroids.npy: {}", e)))?,
481 )
482 .map_err(|e| Error::IndexLoad(format!("Failed to read centroids.npy: {}", e)))?;
483
484 let avg_residual_path = index_path.join("avg_residual.npy");
485 let avg_residual: Array1<f32> =
486 Array1::read_npy(File::open(&avg_residual_path).map_err(|e| {
487 Error::IndexLoad(format!("Failed to open avg_residual.npy: {}", e))
488 })?)
489 .map_err(|e| Error::IndexLoad(format!("Failed to read avg_residual.npy: {}", e)))?;
490
491 let bucket_cutoffs_path = index_path.join("bucket_cutoffs.npy");
492 let bucket_cutoffs: Option<Array1<f32>> = if bucket_cutoffs_path.exists() {
493 Some(
494 Array1::read_npy(File::open(&bucket_cutoffs_path).map_err(|e| {
495 Error::IndexLoad(format!("Failed to open bucket_cutoffs.npy: {}", e))
496 })?)
497 .map_err(|e| {
498 Error::IndexLoad(format!("Failed to read bucket_cutoffs.npy: {}", e))
499 })?,
500 )
501 } else {
502 None
503 };
504
505 let bucket_weights_path = index_path.join("bucket_weights.npy");
506 let bucket_weights: Option<Array1<f32>> = if bucket_weights_path.exists() {
507 Some(
508 Array1::read_npy(File::open(&bucket_weights_path).map_err(|e| {
509 Error::IndexLoad(format!("Failed to open bucket_weights.npy: {}", e))
510 })?)
511 .map_err(|e| {
512 Error::IndexLoad(format!("Failed to read bucket_weights.npy: {}", e))
513 })?,
514 )
515 } else {
516 None
517 };
518
519 let metadata_path = index_path.join("metadata.json");
521 let metadata: serde_json::Value = serde_json::from_reader(
522 File::open(&metadata_path)
523 .map_err(|e| Error::IndexLoad(format!("Failed to open metadata.json: {}", e)))?,
524 )
525 .map_err(|e| Error::IndexLoad(format!("Failed to parse metadata.json: {}", e)))?;
526
527 let nbits = metadata["nbits"]
528 .as_u64()
529 .ok_or_else(|| Error::IndexLoad("nbits not found in metadata".into()))?
530 as usize;
531
532 Self::new(
533 nbits,
534 centroids,
535 avg_residual,
536 bucket_cutoffs,
537 bucket_weights,
538 )
539 }
540
541 pub fn load_mmap_from_dir(index_path: &std::path::Path) -> Result<Self> {
549 use ndarray_npy::ReadNpyExt;
550 use std::fs::File;
551
552 let centroids_path = index_path.join("centroids.npy");
554 let mmap_centroids = crate::mmap::MmapNpyArray2F32::from_npy_file(¢roids_path)?;
555
556 let avg_residual_path = index_path.join("avg_residual.npy");
558 let avg_residual: Array1<f32> =
559 Array1::read_npy(File::open(&avg_residual_path).map_err(|e| {
560 Error::IndexLoad(format!("Failed to open avg_residual.npy: {}", e))
561 })?)
562 .map_err(|e| Error::IndexLoad(format!("Failed to read avg_residual.npy: {}", e)))?;
563
564 let bucket_cutoffs_path = index_path.join("bucket_cutoffs.npy");
565 let bucket_cutoffs: Option<Array1<f32>> = if bucket_cutoffs_path.exists() {
566 Some(
567 Array1::read_npy(File::open(&bucket_cutoffs_path).map_err(|e| {
568 Error::IndexLoad(format!("Failed to open bucket_cutoffs.npy: {}", e))
569 })?)
570 .map_err(|e| {
571 Error::IndexLoad(format!("Failed to read bucket_cutoffs.npy: {}", e))
572 })?,
573 )
574 } else {
575 None
576 };
577
578 let bucket_weights_path = index_path.join("bucket_weights.npy");
579 let bucket_weights: Option<Array1<f32>> = if bucket_weights_path.exists() {
580 Some(
581 Array1::read_npy(File::open(&bucket_weights_path).map_err(|e| {
582 Error::IndexLoad(format!("Failed to open bucket_weights.npy: {}", e))
583 })?)
584 .map_err(|e| {
585 Error::IndexLoad(format!("Failed to read bucket_weights.npy: {}", e))
586 })?,
587 )
588 } else {
589 None
590 };
591
592 let metadata_path = index_path.join("metadata.json");
594 let metadata: serde_json::Value = serde_json::from_reader(
595 File::open(&metadata_path)
596 .map_err(|e| Error::IndexLoad(format!("Failed to open metadata.json: {}", e)))?,
597 )
598 .map_err(|e| Error::IndexLoad(format!("Failed to parse metadata.json: {}", e)))?;
599
600 let nbits = metadata["nbits"]
601 .as_u64()
602 .ok_or_else(|| Error::IndexLoad("nbits not found in metadata".into()))?
603 as usize;
604
605 Self::new_with_store(
606 nbits,
607 CentroidStore::Mmap(mmap_centroids),
608 avg_residual,
609 bucket_cutoffs,
610 bucket_weights,
611 )
612 }
613}
614
615#[cfg(test)]
616mod tests {
617 use super::*;
618
619 #[test]
620 fn test_codec_creation() {
621 let centroids =
622 Array2::from_shape_vec((4, 8), (0..32).map(|x| x as f32).collect()).unwrap();
623 let avg_residual = Array1::zeros(8);
624 let bucket_cutoffs = Some(Array1::from_vec(vec![-0.5, 0.0, 0.5]));
625 let bucket_weights = Some(Array1::from_vec(vec![-0.75, -0.25, 0.25, 0.75]));
626
627 let codec = ResidualCodec::new(2, centroids, avg_residual, bucket_cutoffs, bucket_weights);
628 assert!(codec.is_ok());
629
630 let codec = codec.unwrap();
631 assert_eq!(codec.nbits, 2);
632 assert_eq!(codec.embedding_dim(), 8);
633 assert_eq!(codec.num_centroids(), 4);
634 }
635
636 #[test]
637 fn test_compress_into_codes() {
638 let centroids = Array2::from_shape_vec(
639 (3, 4),
640 vec![
641 1.0, 0.0, 0.0, 0.0, 0.0, 1.0, 0.0, 0.0, 0.0, 0.0, 1.0, 0.0, ],
645 )
646 .unwrap();
647
648 let avg_residual = Array1::zeros(4);
649 let codec = ResidualCodec::new(2, centroids, avg_residual, None, None).unwrap();
650
651 let embeddings = Array2::from_shape_vec(
652 (2, 4),
653 vec![
654 0.9, 0.1, 0.0, 0.0, 0.0, 0.0, 0.95, 0.05, ],
657 )
658 .unwrap();
659
660 let codes = codec.compress_into_codes(&embeddings);
661 assert_eq!(codes[0], 0);
662 assert_eq!(codes[1], 2);
663 }
664
665 #[test]
666 fn test_quantize_decompress_roundtrip_4bit() {
667 let dim = 8;
669 let centroids = Array2::zeros((4, dim));
670 let avg_residual = Array1::zeros(dim);
671
672 let bucket_cutoffs: Vec<f32> = (1..16).map(|i| (i as f32 / 16.0 - 0.5) * 2.0).collect();
675 let bucket_weights: Vec<f32> = (0..16)
677 .map(|i| ((i as f32 + 0.5) / 16.0 - 0.5) * 2.0)
678 .collect();
679
680 let codec = ResidualCodec::new(
681 4,
682 centroids,
683 avg_residual,
684 Some(Array1::from_vec(bucket_cutoffs)),
685 Some(Array1::from_vec(bucket_weights)),
686 )
687 .unwrap();
688
689 let residuals = Array2::from_shape_vec(
691 (2, dim),
692 vec![
693 -0.9, -0.7, -0.5, -0.3, 0.0, 0.3, 0.5, 0.9, -0.8, -0.4, 0.0, 0.4, 0.8, -0.6, 0.2, 0.6,
695 ],
696 )
697 .unwrap();
698
699 let packed = codec.quantize_residuals(&residuals).unwrap();
701 assert_eq!(packed.ncols(), dim * 4 / 8); let codes = Array1::from_vec(vec![0, 0]);
705
706 let decompressed = codec.decompress(&packed, &codes.view()).unwrap();
708
709 for i in 0..residuals.nrows() {
712 for j in 0..residuals.ncols() {
713 let orig = residuals[[i, j]];
714 let recon = decompressed[[i, j]];
715 if orig.abs() > 0.2 {
719 assert!(
720 (orig > 0.0) == (recon > 0.0) || recon.abs() < 0.1,
721 "Sign mismatch at [{}, {}]: orig={}, recon={}",
722 i,
723 j,
724 orig,
725 recon
726 );
727 }
728 }
729 }
730 }
731
732 #[test]
733 fn test_compress_into_codes_ignores_nan_scores_when_finite_choices_exist() {
734 let centroids = Array2::from_shape_vec(
735 (3, 2),
736 vec![
737 f32::NAN,
738 0.0, 1.0,
740 0.0, 0.0,
742 1.0, ],
744 )
745 .unwrap();
746 let avg_residual = Array1::zeros(2);
747 let codec = ResidualCodec::new(2, centroids, avg_residual, None, None).unwrap();
748 let embeddings = Array2::from_shape_vec((1, 2), vec![1.0, 0.0]).unwrap();
749
750 let codes = codec.compress_into_codes_cpu(&embeddings);
751 assert_eq!(codes[0], 1);
752 }
753}