1use std::borrow::Cow;
5use std::collections::{BinaryHeap, HashMap};
6use std::ops::Sub;
7use std::sync::{
8 Arc, OnceLock,
9 atomic::{AtomicU64, Ordering},
10};
11
12use arrow::array::AsArray;
13use arrow::datatypes::{Float16Type, Float32Type, Float64Type, UInt8Type, UInt64Type};
14use arrow_array::{
15 Array, FixedSizeListArray, Float32Array, RecordBatch, UInt8Array, UInt32Array, UInt64Array,
16};
17use arrow_schema::{DataType, Field, SchemaRef};
18use async_trait::async_trait;
19use bytes::{Bytes, BytesMut};
20use itertools::{Itertools, izip};
21use lance_arrow::{ArrowFloatType, FixedSizeListArrayExt, FloatArray, RecordBatchExt};
22use lance_core::deepsize::DeepSizeOf;
23use lance_core::{Error, ROW_ID, Result};
24use lance_file::previous::reader::FileReader as PreviousFileReader;
25use lance_linalg::distance::{DistanceType, Dot, dot, l2::l2};
26use lance_linalg::simd::{
27 self,
28 dist_table::{BATCH_SIZE, PERM0, PERM0_INVERSE},
29};
30#[cfg(any(
31 target_arch = "x86_64",
32 target_arch = "aarch64",
33 target_arch = "loongarch64"
34))]
35use lance_linalg::simd::{SIMD, f32::f32x16};
36use lance_table::utils::LanceIteratorExtension;
37use num_traits::AsPrimitive;
38use prost::Message;
39use serde::{Deserialize, Serialize};
40
41use crate::frag_reuse::FragReuseIndex;
42use crate::pb;
43use crate::vector::ApproxMode;
44use crate::vector::bq::dist_table_quant::{
45 DistTableDequant, quantize_dist_table_into, quantize_dist_table_u16_into,
46};
47use crate::vector::bq::ex_dot::{
48 EX_DOT_BLOCK_DIMS, ExDotFn, blocked_ex_code_bytes, ex_dot_kernel, pad_query_into,
49 padded_query_len, repack_sequential_row, sequential_matches_blocked,
50};
51use crate::vector::bq::prune::{LowerBoundTerms, PRUNE_LANES, prune_mask_kernel};
52use crate::vector::bq::rotation::{apply_fast_rotation, apply_fast_rotation_in_place};
53use crate::vector::bq::transform::{
54 ADD_FACTORS_COLUMN, ERROR_FACTORS_COLUMN, EX_ADD_FACTORS_COLUMN, EX_SCALE_FACTORS_COLUMN,
55 SCALE_FACTORS_COLUMN,
56};
57use crate::vector::bq::{
58 RQRotationType, rabit_binary_code_bytes, rabit_ex_bits, rabit_ex_code_bytes,
59 validate_rq_num_bits,
60};
61use crate::vector::graph::{OrderedFloat, OrderedNode};
62use crate::vector::pq::storage::transpose;
63use crate::vector::quantizer::{QuantizerMetadata, QuantizerStorage};
64use crate::vector::storage::{
65 DistCalculator, DistanceCalculatorOptions, QueryResidual, RabitRawQueryContext, VectorStore,
66};
67
68pub const RABIT_METADATA_KEY: &str = "lance:rabit";
69pub const RABIT_CODE_COLUMN: &str = "_rabit_codes";
70pub const RABIT_EX_CODE_COLUMN: &str = "__ex_codes";
73pub const RABIT_BLOCKED_EX_CODE_COLUMN: &str = "__blocked_ex_codes";
78pub const SEGMENT_LENGTH: usize = 4;
79pub const SEGMENT_NUM_CODES: usize = 1 << SEGMENT_LENGTH;
80const RABIT_PRUNE_STATS_ENV: &str = "LANCE_RQ_PRUNE_STATS";
81const RABIT_PRUNE_STATS_INTERVAL_ENV: &str = "LANCE_RQ_PRUNE_STATS_INTERVAL";
82const DEFAULT_RABIT_PRUNE_STATS_INTERVAL: u64 = 1024;
83
84#[derive(Default)]
85struct RabitPruneStats {
86 calls: AtomicU64,
87 candidates: AtomicU64,
88 pruned_upper_bound: AtomicU64,
89 pruned_heap: AtomicU64,
90 exact: AtomicU64,
91 exact_rejected: AtomicU64,
92}
93
94#[derive(Default)]
95struct RabitPruneBypassStats {
96 calls: AtomicU64,
97}
98
99static RABIT_PRUNE_STATS: OnceLock<RabitPruneStats> = OnceLock::new();
100static RABIT_PRUNE_BYPASS_STATS: OnceLock<RabitPruneBypassStats> = OnceLock::new();
101static RABIT_PRUNE_STATS_ENABLED: OnceLock<bool> = OnceLock::new();
102static RABIT_PRUNE_STATS_INTERVAL: OnceLock<u64> = OnceLock::new();
103
104fn rabit_prune_stats_enabled() -> bool {
105 *RABIT_PRUNE_STATS_ENABLED.get_or_init(|| match std::env::var(RABIT_PRUNE_STATS_ENV) {
106 Ok(value) => {
107 let value = value.to_ascii_lowercase();
108 !matches!(value.as_str(), "" | "0" | "false" | "off" | "no")
109 }
110 Err(_) => false,
111 })
112}
113
114fn rabit_prune_stats_interval() -> u64 {
115 *RABIT_PRUNE_STATS_INTERVAL.get_or_init(|| {
116 std::env::var(RABIT_PRUNE_STATS_INTERVAL_ENV)
117 .ok()
118 .and_then(|value| value.parse::<u64>().ok())
119 .filter(|interval| *interval > 0)
120 .unwrap_or(DEFAULT_RABIT_PRUNE_STATS_INTERVAL)
121 })
122}
123
124fn ratio(numerator: u64, denominator: u64) -> f64 {
125 if denominator == 0 {
126 0.0
127 } else {
128 numerator as f64 / denominator as f64
129 }
130}
131
132fn emit_rabit_prune_stats(message: &str) {
133 log::warn!(
134 target: "lance_index::vector::bq::prune_stats",
135 "{}",
136 message
137 );
138}
139
140#[derive(Default)]
143struct RabitPruneCounters {
144 candidates: usize,
145 pruned_upper_bound: usize,
146 pruned_heap: usize,
147 exact: usize,
148 exact_rejected: usize,
149}
150
151fn record_rabit_prune_stats(counters: &RabitPruneCounters) {
152 if !rabit_prune_stats_enabled() {
153 return;
154 }
155 let RabitPruneCounters {
156 candidates,
157 pruned_upper_bound,
158 pruned_heap,
159 exact,
160 exact_rejected,
161 } = *counters;
162
163 let stats = RABIT_PRUNE_STATS.get_or_init(RabitPruneStats::default);
164 let calls = stats.calls.fetch_add(1, Ordering::Relaxed) + 1;
165 let candidates = stats
166 .candidates
167 .fetch_add(candidates as u64, Ordering::Relaxed)
168 + candidates as u64;
169 let pruned_upper_bound = stats
170 .pruned_upper_bound
171 .fetch_add(pruned_upper_bound as u64, Ordering::Relaxed)
172 + pruned_upper_bound as u64;
173 let pruned_heap = stats
174 .pruned_heap
175 .fetch_add(pruned_heap as u64, Ordering::Relaxed)
176 + pruned_heap as u64;
177 let exact = stats.exact.fetch_add(exact as u64, Ordering::Relaxed) + exact as u64;
178 let exact_rejected = stats
179 .exact_rejected
180 .fetch_add(exact_rejected as u64, Ordering::Relaxed)
181 + exact_rejected as u64;
182 let interval = rabit_prune_stats_interval();
183 if calls.is_multiple_of(interval) {
184 let pruned = pruned_upper_bound + pruned_heap;
185 emit_rabit_prune_stats(&format!(
186 "ivf_rq_prune_stats calls={} candidates={} pruned={} pruned_upper_bound={} pruned_heap={} prune_ratio={:.6} exact={} exact_ratio={:.6} exact_rejected={} exact_reject_ratio={:.6}",
187 calls,
188 candidates,
189 pruned,
190 pruned_upper_bound,
191 pruned_heap,
192 ratio(pruned, candidates),
193 exact,
194 ratio(exact, candidates),
195 exact_rejected,
196 ratio(exact_rejected, exact),
197 ));
198 }
199}
200
201fn record_rabit_prune_bypass(reason: &'static str) {
202 if !rabit_prune_stats_enabled() {
203 return;
204 }
205
206 let stats = RABIT_PRUNE_BYPASS_STATS.get_or_init(RabitPruneBypassStats::default);
207 let calls = stats.calls.fetch_add(1, Ordering::Relaxed) + 1;
208 if calls.is_multiple_of(rabit_prune_stats_interval()) {
209 emit_rabit_prune_stats(&format!(
210 "ivf_rq_prune_stats_bypass calls={} reason={}",
211 calls, reason
212 ));
213 }
214}
215
216#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
217#[serde(rename_all = "snake_case")]
218pub enum RabitQueryEstimator {
219 ResidualQuery,
220 RawQuery,
221}
222
223pub fn rabit_binary_code_field(rotated_dim: usize) -> Field {
224 Field::new(
225 RABIT_CODE_COLUMN,
226 DataType::FixedSizeList(
227 Arc::new(Field::new("item", DataType::UInt8, true)),
228 rabit_binary_code_bytes(rotated_dim) as i32,
229 ),
230 true,
231 )
232}
233
234pub fn rabit_ex_code_field(rotated_dim: usize, num_bits: u8) -> Result<Option<Field>> {
235 let ex_bits = rabit_ex_bits(num_bits)?;
236 if ex_bits == 0 {
237 return Ok(None);
238 }
239 Ok(Some(Field::new(
240 RABIT_BLOCKED_EX_CODE_COLUMN,
241 DataType::FixedSizeList(
242 Arc::new(Field::new("item", DataType::UInt8, true)),
243 blocked_ex_code_bytes(rotated_dim, ex_bits) as i32,
244 ),
245 true,
246 )))
247}
248
249#[derive(Debug, Clone, Serialize, Deserialize)]
250pub struct RabitQuantizationMetadata {
251 #[serde(skip)]
255 pub rotate_mat: Option<FixedSizeListArray>,
256 #[serde(default)]
257 pub rotate_mat_position: Option<u32>,
258 #[serde(default)]
259 pub fast_rotation_signs: Option<Vec<u8>>,
260 #[serde(default = "default_rotation_type_compat")]
261 pub rotation_type: RQRotationType,
262 #[serde(default)]
263 pub code_dim: u32,
264 pub num_bits: u8,
265 pub packed: bool,
266 #[serde(default = "default_query_estimator_compat")]
267 pub query_estimator: RabitQueryEstimator,
268}
269
270impl RabitQuantizationMetadata {
271 pub fn rotated_dim(&self) -> usize {
272 if self.code_dim > 0 {
273 self.code_dim as usize
274 } else {
275 self.rotate_mat
276 .as_ref()
277 .map(|rotate_mat| rotate_mat.len())
278 .unwrap_or(0)
279 }
280 }
281
282 pub fn binary_code_bytes(&self) -> usize {
283 rabit_binary_code_bytes(self.rotated_dim())
284 }
285}
286
287fn default_rotation_type_compat() -> RQRotationType {
288 RQRotationType::Matrix
290}
291
292fn default_query_estimator_compat() -> RabitQueryEstimator {
293 RabitQueryEstimator::ResidualQuery
295}
296
297impl RabitQuantizationMetadata {
298 fn code_dim(&self) -> usize {
299 self.rotated_dim()
300 }
301
302 fn rotate_vector_with_residual_into(
303 &self,
304 vector: &dyn Array,
305 residual_centroid: Option<&dyn Array>,
306 output: &mut [f32],
307 ) {
308 debug_assert_eq!(output.len(), self.code_dim());
309 match self.rotation_type {
310 RQRotationType::Matrix => {
311 let rotate_mat = self
312 .rotate_mat
313 .as_ref()
314 .expect("RabitQ dense rotation metadata not loaded");
315
316 match rotate_mat.value_type() {
317 DataType::Float16 => {
318 RabitQuantizationStorage::rotate_query_vector_dense_into::<Float16Type>(
319 rotate_mat,
320 vector,
321 residual_centroid,
322 output,
323 )
324 }
325 DataType::Float32 => {
326 RabitQuantizationStorage::rotate_query_vector_dense_into::<Float32Type>(
327 rotate_mat,
328 vector,
329 residual_centroid,
330 output,
331 )
332 }
333 DataType::Float64 => {
334 RabitQuantizationStorage::rotate_query_vector_dense_into::<Float64Type>(
335 rotate_mat,
336 vector,
337 residual_centroid,
338 output,
339 )
340 }
341 dt => unimplemented!("RabitQ does not support data type: {}", dt),
342 }
343 }
344 RQRotationType::Fast => {
345 let signs = self
346 .fast_rotation_signs
347 .as_ref()
348 .expect("RabitQ fast rotation metadata not loaded");
349 match vector.data_type() {
350 DataType::Float16 => RabitQuantizationStorage::rotate_query_vector_fast_into::<
351 Float16Type,
352 >(
353 signs, vector, residual_centroid, output
354 ),
355 DataType::Float32 => {
356 RabitQuantizationStorage::rotate_query_vector_fast_f32_into(
357 signs,
358 vector,
359 residual_centroid,
360 output,
361 )
362 }
363 DataType::Float64 => RabitQuantizationStorage::rotate_query_vector_fast_into::<
364 Float64Type,
365 >(
366 signs, vector, residual_centroid, output
367 ),
368 dt => unimplemented!("RabitQ does not support data type: {}", dt),
369 }
370 }
371 }
372 }
373
374 pub fn prepare_raw_query_context(&self, query: &dyn Array) -> Result<RabitRawQueryContext> {
375 validate_rq_num_bits(self.num_bits)?;
376 let code_dim = self.code_dim();
377 let ex_bits = rabit_ex_bits(self.num_bits)?;
378 let dist_table_len = code_dim * 4;
379
380 let mut rotated_query = vec![0.0; code_dim];
381 self.rotate_vector_with_residual_into(query, None, &mut rotated_query);
382
383 let mut dist_table = vec![0.0; dist_table_len];
384 build_dist_table_direct_into::<Float32Type>(&rotated_query, &mut dist_table);
385
386 let mut ex_query = Vec::new();
389 if ex_bits > 0 && !code_dim.is_multiple_of(EX_DOT_BLOCK_DIMS) {
390 ex_query.resize(padded_query_len(code_dim), 0.0);
391 pad_query_into(&rotated_query, &mut ex_query);
392 }
393
394 let sum_q = rotated_query.iter().copied().sum();
395 Ok(RabitRawQueryContext {
396 code_dim,
397 ex_bits,
398 rotated_query,
399 dist_table,
400 ex_query,
401 sum_q,
402 })
403 }
404}
405
406impl DeepSizeOf for RabitQuantizationMetadata {
407 fn deep_size_of_children(&self, context: &mut lance_core::deepsize::Context) -> usize {
408 self.rotate_mat
409 .as_ref()
410 .map(|inv_p| (inv_p as &dyn arrow_array::Array).deep_size_of_children(context))
411 .unwrap_or(0)
412 + self
413 .fast_rotation_signs
414 .as_ref()
415 .map(|signs| signs.len())
416 .unwrap_or(0)
417 }
418}
419
420#[async_trait]
421impl QuantizerMetadata for RabitQuantizationMetadata {
422 fn buffer_index(&self) -> Option<u32> {
423 match self.rotation_type {
424 RQRotationType::Matrix => self.rotate_mat_position,
425 RQRotationType::Fast => None,
426 }
427 }
428
429 fn set_buffer_index(&mut self, index: u32) {
430 self.rotate_mat_position = Some(index);
431 }
432
433 fn parse_buffer(&mut self, bytes: Bytes) -> Result<()> {
434 if self.rotation_type != RQRotationType::Matrix {
435 return Ok(());
436 }
437 debug_assert!(!bytes.is_empty());
438 let codebook_tensor: pb::Tensor = pb::Tensor::decode(bytes)?;
439 self.rotate_mat = Some(FixedSizeListArray::try_from(&codebook_tensor)?);
440 if self.code_dim == 0 {
441 self.code_dim = self
442 .rotate_mat
443 .as_ref()
444 .map(|rotate_mat| rotate_mat.len() as u32)
445 .unwrap_or(0);
446 }
447 Ok(())
448 }
449
450 fn extra_metadata(&self) -> Result<Option<Bytes>> {
451 match self.rotation_type {
452 RQRotationType::Matrix => {
453 if let Some(inv_p) = &self.rotate_mat {
454 let inv_p_tensor = pb::Tensor::try_from(inv_p)?;
455 let mut bytes = BytesMut::new();
456 inv_p_tensor.encode(&mut bytes)?;
457 Ok(Some(bytes.freeze()))
458 } else {
459 Ok(None)
460 }
461 }
462 RQRotationType::Fast => Ok(None),
463 }
464 }
465
466 async fn load(reader: &PreviousFileReader) -> Result<Self> {
467 let metadata_str = reader
468 .schema()
469 .metadata
470 .get(RABIT_METADATA_KEY)
471 .ok_or(Error::index(format!(
472 "Reading Rabit metadata: metadata key {} not found",
473 RABIT_METADATA_KEY
474 )))?;
475 serde_json::from_str(metadata_str)
476 .map_err(|_| Error::index(format!("Failed to parse index metadata: {}", metadata_str)))
477 }
478}
479
480#[derive(Debug, Clone)]
481pub struct RabitQuantizationStorage {
482 metadata: RabitQuantizationMetadata,
483 batch: RecordBatch,
484 distance_type: DistanceType,
485
486 row_ids: UInt64Array,
488 codes: FixedSizeListArray,
489 add_factors: Float32Array,
490 scale_factors: Float32Array,
491 error_factors: Option<Float32Array>,
492 ex_codes: Option<FixedSizeListArray>,
497 packed_ex_codes: Option<FixedSizeListArray>,
498 ex_add_factors: Option<Float32Array>,
499 ex_scale_factors: Option<Float32Array>,
500}
501
502impl DeepSizeOf for RabitQuantizationStorage {
503 fn deep_size_of_children(&self, context: &mut lance_core::deepsize::Context) -> usize {
504 self.metadata.deep_size_of_children(context)
505 + self.batch.deep_size_of_children(context)
506 + self
507 .packed_ex_codes
508 .as_ref()
509 .map(|codes| (codes as &dyn Array).deep_size_of_children(context))
510 .unwrap_or_default()
511 }
512}
513
514impl RabitQuantizationStorage {
515 fn code_dim(&self) -> usize {
516 self.metadata.code_dim()
517 }
518
519 fn residual_query_factor(&self, dist_q_c: f32) -> f32 {
520 match self.distance_type {
521 DistanceType::L2 => dist_q_c,
522 DistanceType::Cosine | DistanceType::Dot => dist_q_c - 1.0,
523 _ => unimplemented!(
524 "RabitQ does not support distance type: {}",
525 self.distance_type
526 ),
527 }
528 }
529
530 fn raw_query_factor(
531 &self,
532 dist_q_c: f32,
533 rotated_query: &[f32],
534 rotated_centroid: Option<&[f32]>,
535 ) -> f32 {
536 match self.distance_type {
537 DistanceType::L2 => dist_q_c,
538 DistanceType::Dot => rotated_centroid
539 .map(|centroid| -dot(rotated_query, centroid))
540 .unwrap_or(dist_q_c - 1.0),
541 DistanceType::Cosine => dist_q_c - 1.0,
542 _ => unimplemented!(
543 "RabitQ does not support distance type: {}",
544 self.distance_type
545 ),
546 }
547 }
548
549 fn raw_query_error(
550 &self,
551 dist_q_c: f32,
552 rotated_query: &[f32],
553 rotated_centroid: Option<&[f32]>,
554 ) -> f32 {
555 match self.distance_type {
556 DistanceType::L2 => dist_q_c.max(0.0).sqrt(),
557 DistanceType::Dot => rotated_centroid
558 .map(|centroid| l2(rotated_query, centroid).sqrt())
559 .unwrap_or_else(|| dist_q_c.max(0.0).sqrt()),
560 DistanceType::Cosine => dist_q_c.max(0.0).sqrt(),
561 _ => unimplemented!(
562 "RabitQ does not support distance type: {}",
563 self.distance_type
564 ),
565 }
566 }
567
568 fn uses_raw_query_lower_bound_gating(&self) -> bool {
569 self.metadata.query_estimator == RabitQueryEstimator::RawQuery
570 && self.metadata.num_bits > 1
571 && self.error_factors.is_some()
572 }
573
574 fn raw_query_error_for_gating(
575 &self,
576 dist_q_c: f32,
577 rotated_query: &[f32],
578 rotated_centroid: Option<&[f32]>,
579 ) -> f32 {
580 if self.uses_raw_query_lower_bound_gating() {
581 self.raw_query_error(dist_q_c, rotated_query, rotated_centroid)
582 } else {
583 0.0
584 }
585 }
586
587 fn distance_calculator_from_parts<'a>(
588 &'a self,
589 parts: RabitDistCalculatorParts<'a>,
590 ) -> RabitDistCalculator<'a> {
591 let RabitDistCalculatorParts {
592 dim,
593 dist_table,
594 ex_query,
595 sum_q,
596 query_factor,
597 query_error,
598 approx_mode,
599 } = parts;
600 let ex_code_len = self
601 .ex_codes
602 .as_ref()
603 .map(|codes| codes.value_length() as usize)
604 .unwrap_or_default();
605 let ex_codes = self
606 .ex_codes
607 .as_ref()
608 .map(|codes| codes.values().as_primitive::<UInt8Type>().values().as_ref());
609 let packed_ex_codes = self
610 .packed_ex_codes
611 .as_ref()
612 .map(|codes| codes.values().as_primitive::<UInt8Type>().values().as_ref());
613 RabitDistCalculator::new(
614 dim,
615 self.metadata.num_bits,
616 self.metadata.query_estimator,
617 dist_table,
618 ex_query,
619 sum_q,
620 self.codes.values().as_primitive::<UInt8Type>().values(),
621 ex_codes,
622 ex_code_len,
623 self.add_factors.values(),
624 self.scale_factors.values(),
625 self.error_factors
626 .as_ref()
627 .map(|factors| factors.values().as_ref()),
628 self.ex_add_factors
629 .as_ref()
630 .map(|factors| factors.values().as_ref()),
631 self.ex_scale_factors
632 .as_ref()
633 .map(|factors| factors.values().as_ref()),
634 packed_ex_codes,
635 query_factor,
636 query_error,
637 approx_mode,
638 )
639 }
640
641 fn rotate_query_vector(&self, code_dim: usize, qr: &dyn Array) -> Vec<f32> {
642 let mut output = vec![0.0f32; code_dim];
643 self.rotate_query_vector_into(code_dim, qr, None, &mut output);
644 output
645 }
646
647 fn rotate_query_vector_into(
648 &self,
649 code_dim: usize,
650 qr: &dyn Array,
651 residual_centroid: Option<&dyn Array>,
652 output: &mut [f32],
653 ) {
654 debug_assert_eq!(output.len(), code_dim);
655 self.metadata
656 .rotate_vector_with_residual_into(qr, residual_centroid, output);
657 }
658
659 fn rotate_query_vector_dense_into<T: ArrowFloatType>(
660 rotate_mat: &FixedSizeListArray,
661 qr: &dyn Array,
662 residual_centroid: Option<&dyn Array>,
663 output: &mut [f32],
664 ) where
665 T::Native: AsPrimitive<f32> + Dot + Sub<Output = T::Native>,
666 {
667 let d = qr.len();
668 let code_dim = rotate_mat.len();
669 debug_assert_eq!(output.len(), code_dim);
670 let rotate_mat = rotate_mat
671 .values()
672 .as_any()
673 .downcast_ref::<T::ArrayType>()
674 .unwrap()
675 .as_slice();
676
677 let qr = qr
678 .as_any()
679 .downcast_ref::<T::ArrayType>()
680 .unwrap()
681 .as_slice();
682
683 if let Some(residual_centroid) = residual_centroid {
684 let residual_centroid = residual_centroid
685 .as_any()
686 .downcast_ref::<T::ArrayType>()
687 .unwrap()
688 .as_slice();
689 debug_assert_eq!(residual_centroid.len(), d);
690 for (chunk, out) in rotate_mat.chunks_exact(code_dim).zip(output.iter_mut()) {
691 let mut sum = 0.0;
692 for idx in 0..d {
693 let residual = qr[idx] - residual_centroid[idx];
694 sum += chunk[idx].as_() * residual.as_();
695 }
696 *out = sum;
697 }
698 } else {
699 rotate_mat
700 .chunks_exact(code_dim)
701 .zip(output.iter_mut())
702 .for_each(|(chunk, out)| {
703 *out = lance_linalg::distance::dot(&chunk[..d], qr);
704 });
705 }
706 }
707
708 fn rotate_query_vector_fast_into<T: ArrowFloatType>(
709 signs: &[u8],
710 qr: &dyn Array,
711 residual_centroid: Option<&dyn Array>,
712 output: &mut [f32],
713 ) where
714 T::Native: AsPrimitive<f32> + Sub<Output = T::Native>,
715 {
716 let qr = qr
717 .as_any()
718 .downcast_ref::<T::ArrayType>()
719 .unwrap()
720 .as_slice();
721
722 if let Some(residual_centroid) = residual_centroid {
723 let residual_centroid = residual_centroid
724 .as_any()
725 .downcast_ref::<T::ArrayType>()
726 .unwrap()
727 .as_slice();
728 let input_len = qr.len().min(output.len());
729 debug_assert!(residual_centroid.len() >= input_len);
730 for idx in 0..input_len {
731 output[idx] = (qr[idx] - residual_centroid[idx]).as_();
732 }
733 if input_len < output.len() {
734 output[input_len..].fill(0.0);
735 }
736 apply_fast_rotation_in_place(output, signs);
737 } else {
738 apply_fast_rotation(qr, output, signs);
739 }
740 }
741
742 fn rotate_query_vector_fast_f32_into(
743 signs: &[u8],
744 qr: &dyn Array,
745 residual_centroid: Option<&dyn Array>,
746 output: &mut [f32],
747 ) {
748 let qr = qr.as_any().downcast_ref::<Float32Array>().unwrap().values();
749
750 if let Some(residual_centroid) = residual_centroid {
751 let residual_centroid = residual_centroid
752 .as_any()
753 .downcast_ref::<Float32Array>()
754 .unwrap()
755 .values();
756 copy_subtract_f32(qr, residual_centroid, output);
757 apply_fast_rotation_in_place(output, signs);
758 } else {
759 apply_fast_rotation(qr, output, signs);
760 }
761 }
762}
763
764#[inline]
765fn copy_subtract_f32(lhs: &[f32], rhs: &[f32], output: &mut [f32]) {
766 let input_len = lhs.len().min(output.len());
767 debug_assert!(rhs.len() >= input_len);
768
769 #[cfg(any(
770 target_arch = "x86_64",
771 target_arch = "aarch64",
772 target_arch = "loongarch64"
773 ))]
774 let simd_len = input_len / f32x16::LANES * f32x16::LANES;
775 #[cfg(not(any(
776 target_arch = "x86_64",
777 target_arch = "aarch64",
778 target_arch = "loongarch64"
779 )))]
780 let simd_len = 0;
781
782 #[cfg(any(
783 target_arch = "x86_64",
784 target_arch = "aarch64",
785 target_arch = "loongarch64"
786 ))]
787 for idx in (0..simd_len).step_by(f32x16::LANES) {
788 let lhs = f32x16::from(&lhs[idx..]);
789 let rhs = f32x16::from(&rhs[idx..]);
790 let result = lhs - rhs;
791 unsafe {
792 result.store_unaligned(output.as_mut_ptr().add(idx));
793 }
794 }
795
796 for idx in simd_len..input_len {
797 output[idx] = lhs[idx] - rhs[idx];
798 }
799 if input_len < output.len() {
800 output[input_len..].fill(0.0);
801 }
802}
803
804struct RabitDistCalculatorParts<'a> {
805 dim: usize,
806 dist_table: Cow<'a, [f32]>,
807 ex_query: Cow<'a, [f32]>,
808 sum_q: f32,
809 query_factor: f32,
810 query_error: f32,
811 approx_mode: ApproxMode,
812}
813
814struct RawQueryTopkContext<'a> {
818 n: usize,
819 k: usize,
820 ex_bits: u8,
821 ex_codes: &'a [u8],
822 ex_add_factors: &'a [f32],
823 ex_scale_factors: &'a [f32],
824 query_lower_bound: f32,
825 query_upper_bound: f32,
826}
827
828fn kernel_query<'a>(rotated_query: &'a [f32], padded: &'a [f32]) -> &'a [f32] {
831 if rotated_query.len().is_multiple_of(EX_DOT_BLOCK_DIMS) {
832 rotated_query
833 } else {
834 padded
835 }
836}
837
838pub struct RabitDistCalculator<'a> {
839 dim: usize,
840 num_bits: u8,
841 query_estimator: RabitQueryEstimator,
842 codes: &'a [u8],
844 ex_codes: Option<&'a [u8]>,
846 ex_code_len: usize,
849 dist_table: Cow<'a, [f32]>,
853 ex_query: Cow<'a, [f32]>,
856 ex_dot: Option<ExDotFn>,
857 add_factors: &'a [f32],
858 scale_factors: &'a [f32],
859 error_factors: Option<&'a [f32]>,
860 ex_add_factors: Option<&'a [f32]>,
861 ex_scale_factors: Option<&'a [f32]>,
862 packed_ex_codes: Option<&'a [u8]>,
863 query_factor: f32,
864 query_error: f32,
865 approx_mode: ApproxMode,
866
867 sum_q: f32,
868 sqrt_d: f32,
869}
870
871impl<'a> RabitDistCalculator<'a> {
872 #[allow(clippy::too_many_arguments)]
873 pub fn new(
874 dim: usize,
875 num_bits: u8,
876 query_estimator: RabitQueryEstimator,
877 dist_table: Cow<'a, [f32]>,
878 ex_query: Cow<'a, [f32]>,
879 sum_q: f32,
880 codes: &'a [u8],
881 ex_codes: Option<&'a [u8]>,
882 ex_code_len: usize,
883 add_factors: &'a [f32],
884 scale_factors: &'a [f32],
885 error_factors: Option<&'a [f32]>,
886 ex_add_factors: Option<&'a [f32]>,
887 ex_scale_factors: Option<&'a [f32]>,
888 packed_ex_codes: Option<&'a [u8]>,
889 query_factor: f32,
890 query_error: f32,
891 approx_mode: ApproxMode,
892 ) -> Self {
893 let ex_dot = (num_bits > 1).then(|| ex_dot_kernel(num_bits - 1));
894 Self {
895 dim,
896 num_bits,
897 query_estimator,
898 codes,
899 ex_codes,
900 ex_code_len,
901 dist_table,
902 ex_query,
903 ex_dot,
904 add_factors,
905 scale_factors,
906 error_factors,
907 ex_add_factors,
908 ex_scale_factors,
909 packed_ex_codes,
910 query_factor,
911 query_error,
912 approx_mode,
913 sqrt_d: (dim as f32 * num_bits as f32).sqrt(),
914 sum_q,
915 }
916 }
917
918 #[inline]
920 fn ex_code_dot(&self, ex_codes: &[u8], id: usize) -> f32 {
921 let ex_dot = self
922 .ex_dot
923 .expect("raw-query multi-bit RQ requires an ex-dot kernel");
924 ex_dot(
925 self.ex_query.as_ref(),
926 &ex_codes[id * self.ex_code_len..(id + 1) * self.ex_code_len],
927 )
928 }
929
930 #[allow(clippy::uninit_vec)]
934 fn fill_exact_binary_distances(&self, n: usize, code_len: usize, dists: &mut Vec<f32>) {
935 dists.clear();
936 dists.reserve(n);
937 unsafe {
939 dists.set_len(n);
940 }
941 dists.iter_mut().enumerate().for_each(|(id, dist)| {
942 *dist = compute_single_rq_distance(self.codes, id, n, code_len, &self.dist_table);
943 });
944 }
945
946 #[allow(clippy::uninit_vec)]
947 fn binary_distances_with_scratch(
948 &self,
949 n: usize,
950 code_len: usize,
951 dists: &mut Vec<f32>,
952 quantized_dists: &mut Vec<u16>,
953 quantized_dists_table: &mut Vec<u8>,
954 hacc_quantized_dists: &mut Vec<u32>,
955 ) -> usize {
956 if self.approx_mode == ApproxMode::Accurate {
957 return self.binary_distances_hacc_with_scratch(
958 n,
959 code_len,
960 dists,
961 quantized_dists,
962 quantized_dists_table,
963 hacc_quantized_dists,
964 );
965 }
966
967 let (qmin, qmax) = match quantize_dist_table_into(&self.dist_table, quantized_dists_table) {
968 DistTableDequant::Affine { qmin, qmax } => (qmin, qmax),
969 DistTableDequant::Exact => {
970 self.fill_exact_binary_distances(n, code_len, dists);
974 return 0;
975 }
976 };
977 let remainder = n % BATCH_SIZE;
978 let simd_len = n - remainder;
979 quantized_dists.clear();
980 quantized_dists.reserve(simd_len);
981 unsafe {
983 quantized_dists.set_len(simd_len);
984 }
985 simd::dist_table::sum_4bit_dist_table(
986 simd_len,
987 code_len,
988 self.codes,
989 quantized_dists_table,
990 quantized_dists,
991 );
992
993 let range = (qmax - qmin) / 255.0;
994 let num_tables = quantized_dists_table.len() / SEGMENT_NUM_CODES;
995 let sum_min = num_tables as f32 * qmin;
996 dists.clear();
997 dists.reserve(n);
998 unsafe {
1001 dists.set_len(n);
1002 }
1003 let (simd_dists, remainder_dists) = dists.split_at_mut(simd_len);
1004 simd_dists
1005 .iter_mut()
1006 .zip(quantized_dists.iter())
1007 .for_each(|(dist, q_dist)| {
1008 *dist = (*q_dist as f32) * range + sum_min;
1009 });
1010
1011 remainder_dists
1012 .iter_mut()
1013 .enumerate()
1014 .for_each(|(id, dist)| {
1015 *dist = compute_single_rq_distance(
1016 self.codes,
1017 simd_len + id,
1018 n,
1019 code_len,
1020 &self.dist_table,
1021 );
1022 });
1023 simd_len
1024 }
1025
1026 #[allow(clippy::uninit_vec)]
1027 fn binary_distances_hacc_with_scratch(
1028 &self,
1029 n: usize,
1030 code_len: usize,
1031 dists: &mut Vec<f32>,
1032 quantized_dist_table: &mut Vec<u16>,
1033 hacc_dist_table: &mut Vec<u8>,
1034 quantized_dists: &mut Vec<u32>,
1035 ) -> usize {
1036 let (qmin, qmax) =
1037 match quantize_dist_table_u16_into(&self.dist_table, quantized_dist_table) {
1038 DistTableDequant::Affine { qmin, qmax } => (qmin, qmax),
1039 DistTableDequant::Exact => {
1040 self.fill_exact_binary_distances(n, code_len, dists);
1043 return 0;
1044 }
1045 };
1046 simd::dist_table::transfer_4bit_dist_table_u16(quantized_dist_table, hacc_dist_table);
1047 let remainder = n % BATCH_SIZE;
1048 let simd_len = n - remainder;
1049 quantized_dists.clear();
1050 quantized_dists.reserve(simd_len);
1051 unsafe {
1053 quantized_dists.set_len(simd_len);
1054 }
1055 simd::dist_table::sum_4bit_hacc_dist_table(
1056 simd_len,
1057 code_len,
1058 self.codes,
1059 hacc_dist_table,
1060 quantized_dists,
1061 );
1062
1063 let range = (qmax - qmin) / u16::MAX as f32;
1064 let num_tables = quantized_dist_table.len() / SEGMENT_NUM_CODES;
1065 let sum_min = num_tables as f32 * qmin;
1066 dists.clear();
1067 dists.reserve(n);
1068 unsafe {
1071 dists.set_len(n);
1072 }
1073 let (simd_dists, remainder_dists) = dists.split_at_mut(simd_len);
1074 simd_dists
1075 .iter_mut()
1076 .zip(quantized_dists.iter())
1077 .for_each(|(dist, q_dist)| {
1078 *dist = (*q_dist as f32) * range + sum_min;
1079 });
1080
1081 remainder_dists
1082 .iter_mut()
1083 .enumerate()
1084 .for_each(|(id, dist)| {
1085 *dist = compute_single_rq_distance(
1086 self.codes,
1087 simd_len + id,
1088 n,
1089 code_len,
1090 &self.dist_table,
1091 );
1092 });
1093 simd_len
1094 }
1095
1096 #[inline]
1097 fn binary_distance_factor_params(&self) -> (f32, f32) {
1098 match self.query_estimator {
1099 RabitQueryEstimator::ResidualQuery => (2.0 / self.sqrt_d, -self.sum_q / self.sqrt_d),
1100 RabitQueryEstimator::RawQuery => (1.0, -0.5 * self.sum_q),
1101 }
1102 }
1103
1104 #[allow(clippy::uninit_vec)]
1105 fn one_bit_distances_with_scratch(
1106 &self,
1107 n: usize,
1108 code_len: usize,
1109 dists: &mut Vec<f32>,
1110 quantized_dists: &mut Vec<u16>,
1111 quantized_dists_table: &mut Vec<u8>,
1112 hacc_quantized_dists: &mut Vec<u32>,
1113 ) {
1114 self.binary_distances_with_scratch(
1115 n,
1116 code_len,
1117 dists,
1118 quantized_dists,
1119 quantized_dists_table,
1120 hacc_quantized_dists,
1121 );
1122 let (binary_distance_multiplier, binary_distance_offset) =
1123 self.binary_distance_factor_params();
1124 dists.iter_mut().enumerate().for_each(|(id, dist)| {
1125 let binary_dist = *dist;
1126 *dist = (binary_dist * binary_distance_multiplier + binary_distance_offset)
1127 * self.scale_factors[id]
1128 + self.add_factors[id]
1129 + self.query_factor;
1130 });
1131 }
1132
1133 #[allow(clippy::uninit_vec)]
1134 fn apply_raw_query_multi_bit_distances(
1135 &self,
1136 simd_len: usize,
1137 dists: &mut [f32],
1138 quantized_dists: &mut Vec<u16>,
1139 quantized_dists_table: &mut Vec<u8>,
1140 ) {
1141 let ex_bits = self.num_bits - 1;
1142 let ex_codes = self
1143 .ex_codes
1144 .expect("raw-query multi-bit RQ requires ex codes");
1145 let ex_add_factors = self
1146 .ex_add_factors
1147 .expect("raw-query multi-bit RQ requires ex add factors");
1148 let ex_scale_factors = self
1149 .ex_scale_factors
1150 .expect("raw-query multi-bit RQ requires ex scale factors");
1151 let code_scale = (1u32 << ex_bits) as f32;
1152 let code_bias = -(code_scale - 0.5);
1153
1154 let fastscan_len = if simd_len > 0 && supports_ex_fastscan(ex_bits) {
1155 self.packed_ex_codes
1156 .map(|packed_ex_codes| {
1157 let fastscan_len = simd_len;
1158 let fastscan_code_len = self.ex_code_len;
1159 let (qmin, qmax, quantization_max) = quantize_ex_fastscan_dist_table_into(
1160 ex_bits,
1161 self.ex_code_len,
1162 self.ex_query.as_ref(),
1163 quantized_dists_table,
1164 );
1165 quantized_dists.clear();
1166 quantized_dists.reserve(fastscan_len);
1167 unsafe {
1169 quantized_dists.set_len(fastscan_len);
1170 }
1171 simd::dist_table::sum_4bit_dist_table(
1172 fastscan_len,
1173 fastscan_code_len,
1174 packed_ex_codes,
1175 quantized_dists_table,
1176 quantized_dists,
1177 );
1178
1179 let range = (qmax - qmin) / quantization_max;
1180 let num_tables = quantized_dists_table.len() / SEGMENT_NUM_CODES;
1181 let sum_min = num_tables as f32 * qmin;
1182 dists
1183 .iter_mut()
1184 .take(fastscan_len)
1185 .zip(quantized_dists.iter())
1186 .enumerate()
1187 .for_each(|(id, (dist, q_ex_dist))| {
1188 let ex_dist = (*q_ex_dist as f32) * range + sum_min;
1189 let full_dot = code_scale * *dist + ex_dist + code_bias * self.sum_q;
1190 *dist = full_dot * ex_scale_factors[id]
1191 + ex_add_factors[id]
1192 + self.query_factor;
1193 });
1194 fastscan_len
1195 })
1196 .unwrap_or_default()
1197 } else {
1198 0
1199 };
1200
1201 dists
1202 .iter_mut()
1203 .enumerate()
1204 .skip(fastscan_len)
1205 .for_each(|(id, dist)| {
1206 let ex_dist = self.ex_code_dot(ex_codes, id);
1207 let full_dot = code_scale * *dist + ex_dist + code_bias * self.sum_q;
1208 *dist = full_dot * ex_scale_factors[id] + ex_add_factors[id] + self.query_factor;
1209 });
1210 }
1211
1212 #[inline]
1213 fn raw_query_binary_distance(&self, id: usize, binary_ip: f32) -> f32 {
1214 (binary_ip - 0.5 * self.sum_q) * self.scale_factors[id]
1215 + self.add_factors[id]
1216 + self.query_factor
1217 }
1218
1219 #[inline]
1220 fn raw_query_lower_bound(&self, id: usize, binary_ip: f32) -> Option<f32> {
1221 let error_factors = self.error_factors?;
1222 Some(self.raw_query_binary_distance(id, binary_ip) - error_factors[id] * self.query_error)
1223 }
1224
1225 #[inline]
1226 #[allow(clippy::too_many_arguments)]
1227 fn raw_query_multi_bit_exact_distance(
1228 &self,
1229 id: usize,
1230 binary_ip: f32,
1231 ex_bits: u8,
1232 ex_codes: &[u8],
1233 ex_add_factors: &[f32],
1234 ex_scale_factors: &[f32],
1235 ) -> f32 {
1236 let ex_dist = self.ex_code_dot(ex_codes, id);
1237 let code_bias = -((1u32 << ex_bits) as f32 - 0.5);
1238 let full_dot = (1u32 << ex_bits) as f32 * binary_ip + ex_dist + code_bias * self.sum_q;
1239 full_dot * ex_scale_factors[id] + ex_add_factors[id] + self.query_factor
1240 }
1241
1242 #[allow(clippy::too_many_arguments)]
1246 fn raw_query_multi_bit_topk_context(
1247 &self,
1248 k: usize,
1249 lower_bound: Option<f32>,
1250 upper_bound: Option<f32>,
1251 dists: &mut Vec<f32>,
1252 quantized_dists: &mut Vec<u16>,
1253 quantized_dists_table: &mut Vec<u8>,
1254 hacc_quantized_dists: &mut Vec<u32>,
1255 ) -> Option<RawQueryTopkContext<'_>> {
1256 let code_len = rabit_binary_code_bytes(self.dim);
1257 let n = self.codes.len() / code_len;
1258 if n == 0 {
1259 dists.clear();
1260 quantized_dists.clear();
1261 hacc_quantized_dists.clear();
1262 return None;
1263 }
1264
1265 self.binary_distances_with_scratch(
1266 n,
1267 code_len,
1268 dists,
1269 quantized_dists,
1270 quantized_dists_table,
1271 hacc_quantized_dists,
1272 );
1273
1274 Some(RawQueryTopkContext {
1275 n,
1276 k,
1277 ex_bits: self.num_bits - 1,
1278 ex_codes: self
1279 .ex_codes
1280 .expect("raw-query multi-bit RQ requires ex codes"),
1281 ex_add_factors: self
1282 .ex_add_factors
1283 .expect("raw-query multi-bit RQ requires ex add factors"),
1284 ex_scale_factors: self
1285 .ex_scale_factors
1286 .expect("raw-query multi-bit RQ requires ex scale factors"),
1287 query_lower_bound: lower_bound.unwrap_or(f32::MIN),
1288 query_upper_bound: upper_bound.unwrap_or(f32::MAX),
1289 })
1290 }
1291
1292 #[inline]
1296 #[allow(clippy::too_many_arguments)]
1297 fn accumulate_raw_query_multi_bit_row(
1298 &self,
1299 ctx: &RawQueryTopkContext<'_>,
1300 id: usize,
1301 row_id: u64,
1302 binary_ip: f32,
1303 raw_lower_bound: f32,
1304 res: &mut BinaryHeap<OrderedNode<u64>>,
1305 max_dist: &mut Option<OrderedFloat>,
1306 counters: &mut RabitPruneCounters,
1307 ) {
1308 if raw_lower_bound >= ctx.query_upper_bound {
1309 counters.pruned_upper_bound += 1;
1310 return;
1311 }
1312 if res.len() >= ctx.k && max_dist.is_some_and(|max_dist| raw_lower_bound >= max_dist.0) {
1313 counters.pruned_heap += 1;
1314 return;
1315 }
1316
1317 counters.exact += 1;
1318 let dist = self.raw_query_multi_bit_exact_distance(
1319 id,
1320 binary_ip,
1321 ctx.ex_bits,
1322 ctx.ex_codes,
1323 ctx.ex_add_factors,
1324 ctx.ex_scale_factors,
1325 );
1326 if dist < ctx.query_lower_bound || dist >= ctx.query_upper_bound {
1327 counters.exact_rejected += 1;
1328 return;
1329 }
1330 let dist = OrderedFloat(dist);
1331 if res.len() < ctx.k {
1332 res.push(OrderedNode::new(row_id, dist));
1333 if res.len() == ctx.k {
1334 *max_dist = res.peek().map(|node| node.dist);
1335 }
1336 } else if max_dist.is_some_and(|max_dist| max_dist > dist) {
1337 res.pop();
1338 res.push(OrderedNode::new(row_id, dist));
1339 *max_dist = res.peek().map(|node| node.dist);
1340 }
1341 }
1342
1343 #[allow(clippy::too_many_arguments)]
1344 fn accumulate_raw_query_multi_bit_topk_with_scratch(
1345 &self,
1346 k: usize,
1347 lower_bound: Option<f32>,
1348 upper_bound: Option<f32>,
1349 row_ids: impl Iterator<Item = (usize, u64)>,
1350 res: &mut BinaryHeap<OrderedNode<u64>>,
1351 dists: &mut Vec<f32>,
1352 quantized_dists: &mut Vec<u16>,
1353 quantized_dists_table: &mut Vec<u8>,
1354 hacc_quantized_dists: &mut Vec<u32>,
1355 ) {
1356 let Some(ctx) = self.raw_query_multi_bit_topk_context(
1357 k,
1358 lower_bound,
1359 upper_bound,
1360 dists,
1361 quantized_dists,
1362 quantized_dists_table,
1363 hacc_quantized_dists,
1364 ) else {
1365 return;
1366 };
1367 let mut max_dist = res.peek().map(|node| node.dist);
1368 let mut counters = RabitPruneCounters::default();
1369
1370 for (id, row_id) in row_ids {
1371 let Some(binary_ip) = dists.get(id).copied() else {
1372 continue;
1373 };
1374 counters.candidates += 1;
1375 let Some(raw_lower_bound) = self.raw_query_lower_bound(id, binary_ip) else {
1376 continue;
1377 };
1378 self.accumulate_raw_query_multi_bit_row(
1379 &ctx,
1380 id,
1381 row_id,
1382 binary_ip,
1383 raw_lower_bound,
1384 res,
1385 &mut max_dist,
1386 &mut counters,
1387 );
1388 }
1389 record_rabit_prune_stats(&counters);
1390 }
1391
1392 #[allow(clippy::too_many_arguments)]
1396 fn accumulate_raw_query_multi_bit_topk_dense_with_scratch(
1397 &self,
1398 k: usize,
1399 lower_bound: Option<f32>,
1400 upper_bound: Option<f32>,
1401 row_id: impl Fn(u32) -> u64,
1402 res: &mut BinaryHeap<OrderedNode<u64>>,
1403 dists: &mut Vec<f32>,
1404 quantized_dists: &mut Vec<u16>,
1405 quantized_dists_table: &mut Vec<u8>,
1406 hacc_quantized_dists: &mut Vec<u32>,
1407 ) {
1408 let Some(ctx) = self.raw_query_multi_bit_topk_context(
1409 k,
1410 lower_bound,
1411 upper_bound,
1412 dists,
1413 quantized_dists,
1414 quantized_dists_table,
1415 hacc_quantized_dists,
1416 ) else {
1417 return;
1418 };
1419 let dists = dists.as_slice();
1420 debug_assert_eq!(dists.len(), ctx.n);
1421 let scale_factors = &self.scale_factors[..ctx.n];
1422 let add_factors = &self.add_factors[..ctx.n];
1423 let error_factors = &self
1424 .error_factors
1425 .expect("raw-query lower-bound gating requires error factors")[..ctx.n];
1426 let lower_bound_of = |id: usize, binary_ip: f32| {
1429 self.raw_query_binary_distance(id, binary_ip) - error_factors[id] * self.query_error
1430 };
1431 let terms = LowerBoundTerms {
1432 half_sum_q: 0.5 * self.sum_q,
1433 query_factor: self.query_factor,
1434 query_error: self.query_error,
1435 };
1436 let prune_masks = prune_mask_kernel();
1437 let mut max_dist = res.peek().map(|node| node.dist);
1438 let mut counters = RabitPruneCounters::default();
1439
1440 let (dist_groups, dist_tail) = dists.as_chunks::<PRUNE_LANES>();
1441 let (scale_groups, _) = scale_factors.as_chunks::<PRUNE_LANES>();
1442 let (add_groups, _) = add_factors.as_chunks::<PRUNE_LANES>();
1443 let (error_groups, _) = error_factors.as_chunks::<PRUNE_LANES>();
1444 for (group, (dist16, scale16, add16, error16)) in
1445 izip!(dist_groups, scale_groups, add_groups, error_groups).enumerate()
1446 {
1447 counters.candidates += PRUNE_LANES;
1448 let heap_threshold = (res.len() >= ctx.k)
1453 .then(|| max_dist.map(|max_dist| max_dist.0))
1454 .flatten();
1455 let (pruned_upper_bound, pruned_heap) = prune_masks(
1456 dist16,
1457 scale16,
1458 add16,
1459 error16,
1460 terms,
1461 ctx.query_upper_bound,
1462 heap_threshold,
1463 );
1464 counters.pruned_upper_bound += pruned_upper_bound.count_ones() as usize;
1465 counters.pruned_heap += pruned_heap.count_ones() as usize;
1466 let mut survivors = !(pruned_upper_bound | pruned_heap);
1467 while survivors != 0 {
1468 let lane = survivors.trailing_zeros() as usize;
1469 survivors &= survivors - 1;
1470 let id = group * PRUNE_LANES + lane;
1471 let binary_ip = dists[id];
1472 self.accumulate_raw_query_multi_bit_row(
1473 &ctx,
1474 id,
1475 row_id(id as u32),
1476 binary_ip,
1477 lower_bound_of(id, binary_ip),
1478 res,
1479 &mut max_dist,
1480 &mut counters,
1481 );
1482 }
1483 }
1484
1485 let tail_start = ctx.n - dist_tail.len();
1486 for (offset, binary_ip) in dist_tail.iter().copied().enumerate() {
1487 let id = tail_start + offset;
1488 counters.candidates += 1;
1489 self.accumulate_raw_query_multi_bit_row(
1490 &ctx,
1491 id,
1492 row_id(id as u32),
1493 binary_ip,
1494 lower_bound_of(id, binary_ip),
1495 res,
1496 &mut max_dist,
1497 &mut counters,
1498 );
1499 }
1500 record_rabit_prune_stats(&counters);
1501 }
1502
1503 fn raw_query_lower_bound_gating_disabled_reason(&self) -> Option<&'static str> {
1504 if self.approx_mode == ApproxMode::Fast {
1505 Some("approx_mode_fast")
1506 } else if self.query_estimator != RabitQueryEstimator::RawQuery {
1507 Some("residual_query_estimator")
1508 } else if self.num_bits <= 1 {
1509 Some("num_bits_le_one")
1510 } else if self.error_factors.is_none() {
1511 Some("missing_error_factors")
1512 } else {
1513 None
1514 }
1515 }
1516}
1517
1518#[inline]
1519fn lowbit(x: usize) -> usize {
1520 1 << x.trailing_zeros()
1521}
1522
1523#[inline]
1524pub fn build_dist_table_direct<T: ArrowFloatType>(qc: &[T::Native]) -> Vec<f32>
1525where
1526 T::Native: AsPrimitive<f32>,
1527{
1528 let mut dist_table = vec![0.0; qc.len() * 4];
1532 build_dist_table_direct_into::<T>(qc, &mut dist_table);
1533 dist_table
1534}
1535
1536fn build_dist_table_direct_into<T: ArrowFloatType>(qc: &[T::Native], dist_table: &mut [f32])
1537where
1538 T::Native: AsPrimitive<f32>,
1539{
1540 debug_assert_eq!(dist_table.len(), qc.len() * 4);
1541 qc.chunks_exact(SEGMENT_LENGTH)
1542 .zip(dist_table.chunks_exact_mut(SEGMENT_NUM_CODES))
1543 .for_each(|(sub_vec, dist_table)| {
1544 dist_table[0] = 0.0;
1545 build_dist_table_for_subvec::<T>(sub_vec, dist_table);
1546 });
1547}
1548
1549#[inline(always)]
1550fn build_dist_table_for_subvec<T: ArrowFloatType>(sub_vec: &[T::Native], dist_table: &mut [f32])
1551where
1552 T::Native: AsPrimitive<f32>,
1553{
1554 (1..SEGMENT_NUM_CODES).for_each(|j| {
1556 dist_table[j] = dist_table[j - lowbit(j)] + sub_vec[LOWBIT_IDX[j]].as_();
1569 })
1570}
1571
1572fn quantize_ex_fastscan_dist_table_into(
1577 ex_bits: u8,
1578 ex_code_len: usize,
1579 ex_query: &[f32],
1580 quantized_dist_table: &mut Vec<u8>,
1581) -> (f32, f32, f32) {
1582 debug_assert!(supports_ex_fastscan(ex_bits));
1583
1584 let num_split_tables = ex_code_len * 2;
1586 let quantization_max = (u16::MAX as usize / num_split_tables)
1587 .min(u8::MAX as usize)
1588 .max(1) as f32;
1589
1590 let mut qmin = f32::INFINITY;
1591 let mut qmax = f32::NEG_INFINITY;
1592 for table_idx in 0..num_split_tables {
1593 for code in 0..SEGMENT_NUM_CODES {
1594 let value = ex_fastscan_dist_table_value(ex_query, ex_bits, table_idx, code);
1595 qmin = qmin.min(value);
1596 qmax = qmax.max(value);
1597 }
1598 }
1599
1600 quantized_dist_table.clear();
1601 quantized_dist_table.reserve(num_split_tables * SEGMENT_NUM_CODES);
1602 if qmin == qmax {
1603 quantized_dist_table.resize(num_split_tables * SEGMENT_NUM_CODES, 0);
1604 return (qmin, qmax, quantization_max);
1605 }
1606
1607 let factor = quantization_max / (qmax - qmin);
1608 for table_idx in 0..num_split_tables {
1609 for code in 0..SEGMENT_NUM_CODES {
1610 let value = ex_fastscan_dist_table_value(ex_query, ex_bits, table_idx, code);
1611 quantized_dist_table.push(((value - qmin) * factor).round() as u8);
1612 }
1613 }
1614
1615 (qmin, qmax, quantization_max)
1616}
1617
1618#[inline]
1619fn supports_ex_fastscan(ex_bits: u8) -> bool {
1620 matches!(ex_bits, 2 | 4 | 8)
1621}
1622
1623#[inline]
1629fn ex_fastscan_dist_table_value(
1630 ex_query: &[f32],
1631 ex_bits: u8,
1632 table_idx: usize,
1633 code: usize,
1634) -> f32 {
1635 let query = |dim_idx: usize| ex_query.get(dim_idx).copied().unwrap_or(0.0);
1636 let byte_idx = table_idx / 2;
1637 let high_nibble = table_idx % 2 == 1;
1638 match ex_bits {
1639 2 => {
1640 let dim_idx = 64 * (byte_idx / 16) + byte_idx % 16 + 32 * usize::from(high_nibble);
1643 let low = (code & 0b11) as f32;
1644 let high = ((code >> 2) & 0b11) as f32;
1645 query(dim_idx) * low + query(dim_idx + 16) * high
1646 }
1647 4 => {
1648 let in_block = byte_idx % 32;
1650 let dim_idx = 64 * (byte_idx / 32)
1651 + 16 * (in_block / 8)
1652 + in_block % 8
1653 + 8 * usize::from(high_nibble);
1654 query(dim_idx) * code as f32
1655 }
1656 8 => {
1657 let code = if high_nibble {
1659 code << SEGMENT_LENGTH
1660 } else {
1661 code
1662 };
1663 query(byte_idx) * code as f32
1664 }
1665 _ => unreachable!("unsupported RabitQ ex_bits={ex_bits} for FastScan"),
1666 }
1667}
1668
1669fn maybe_pack_ex_codes(
1674 ex_codes: Option<&FixedSizeListArray>,
1675 ex_bits: u8,
1676 error_factors: Option<&Float32Array>,
1677) -> Option<FixedSizeListArray> {
1678 let ex_codes = ex_codes?;
1679 if error_factors.is_some() {
1680 return None;
1681 }
1682 match ex_bits {
1683 2 | 4 | 8 => Some(pack_codes(ex_codes)),
1684 _ => None,
1685 }
1686}
1687
1688fn blocked_ex_codes_from_sequential(
1692 seq_codes: &FixedSizeListArray,
1693 dim: usize,
1694 ex_bits: u8,
1695) -> Result<FixedSizeListArray> {
1696 if sequential_matches_blocked(ex_bits)
1697 && seq_codes.value_length() as usize == blocked_ex_code_bytes(dim, ex_bits)
1698 {
1699 return Ok(seq_codes.clone());
1700 }
1701 let seq_code_len = seq_codes.value_length() as usize;
1702 let seq_values = seq_codes.values().as_primitive::<UInt8Type>().values();
1703 let blocked_code_len = blocked_ex_code_bytes(dim, ex_bits);
1704 let mut blocked_values = vec![0u8; seq_codes.len() * blocked_code_len];
1705 for (seq_row, blocked_row) in seq_values
1706 .chunks_exact(seq_code_len)
1707 .zip(blocked_values.chunks_exact_mut(blocked_code_len))
1708 {
1709 repack_sequential_row(seq_row, dim, ex_bits, blocked_row);
1710 }
1711 Ok(FixedSizeListArray::try_new_from_values(
1712 UInt8Array::from(blocked_values),
1713 blocked_code_len as i32,
1714 )?)
1715}
1716
1717pub(crate) fn load_blocked_ex_codes(
1723 batch: RecordBatch,
1724 rotated_dim: usize,
1725 num_bits: u8,
1726) -> Result<(RecordBatch, FixedSizeListArray)> {
1727 let ex_bits = rabit_ex_bits(num_bits)?;
1728 if let Some(column) = batch.column_by_name(RABIT_BLOCKED_EX_CODE_COLUMN) {
1729 let codes = column.as_fixed_size_list().clone();
1730 let expected_bytes = blocked_ex_code_bytes(rotated_dim, ex_bits);
1731 if codes.value_length() as usize != expected_bytes {
1732 return Err(Error::invalid_input(format!(
1733 "RabitQ ex-code byte width mismatch: column {} has {} bytes, metadata rotated_dim={} ex_bits={} requires {} bytes",
1734 RABIT_BLOCKED_EX_CODE_COLUMN,
1735 codes.value_length(),
1736 rotated_dim,
1737 ex_bits,
1738 expected_bytes
1739 )));
1740 }
1741 return Ok((batch, codes));
1742 }
1743 let column = batch.column_by_name(RABIT_EX_CODE_COLUMN).ok_or_else(|| {
1744 Error::invalid_input(format!(
1745 "RabitQ num_bits={} requires {} column",
1746 num_bits, RABIT_BLOCKED_EX_CODE_COLUMN
1747 ))
1748 })?;
1749 let codes = column.as_fixed_size_list().clone();
1750 let expected_bytes = rabit_ex_code_bytes(rotated_dim, ex_bits)?;
1751 if codes.value_length() as usize != expected_bytes {
1752 return Err(Error::invalid_input(format!(
1753 "RabitQ ex-code byte width mismatch: column {} has {} bytes, metadata rotated_dim={} ex_bits={} requires {} bytes",
1754 RABIT_EX_CODE_COLUMN,
1755 codes.value_length(),
1756 rotated_dim,
1757 ex_bits,
1758 expected_bytes
1759 )));
1760 }
1761 let blocked = blocked_ex_codes_from_sequential(&codes, rotated_dim, ex_bits)?;
1762 let ex_code_field = rabit_ex_code_field(rotated_dim, num_bits)?
1763 .expect("multi-bit RabitQ always has an ex-code field");
1764 let batch = batch
1765 .drop_column(RABIT_EX_CODE_COLUMN)?
1766 .try_with_column(ex_code_field, Arc::new(blocked.clone()))?;
1767 Ok((batch, blocked))
1768}
1769
1770impl DistCalculator for RabitDistCalculator<'_> {
1771 #[inline(always)]
1772 fn distance(&self, id: u32) -> f32 {
1773 let id = id as usize;
1774 let code_len = rabit_binary_code_bytes(self.dim);
1775 let num_vectors = self.codes.len() / code_len;
1776 let dist =
1777 compute_single_rq_distance(self.codes, id, num_vectors, code_len, &self.dist_table);
1778
1779 match self.query_estimator {
1780 RabitQueryEstimator::ResidualQuery => {
1781 let dist_vq_qr = (2.0 * dist - self.sum_q) / self.sqrt_d;
1783 dist_vq_qr * self.scale_factors[id] + self.add_factors[id] + self.query_factor
1784 }
1785 RabitQueryEstimator::RawQuery => {
1786 let ex_bits = self.num_bits - 1;
1787 if ex_bits == 0 || self.approx_mode == ApproxMode::Fast {
1788 return self.raw_query_binary_distance(id, dist);
1789 }
1790
1791 let ex_codes = self
1792 .ex_codes
1793 .expect("raw-query multi-bit RQ requires ex codes");
1794 let ex_add_factors = self
1795 .ex_add_factors
1796 .expect("raw-query multi-bit RQ requires ex add factors");
1797 let ex_scale_factors = self
1798 .ex_scale_factors
1799 .expect("raw-query multi-bit RQ requires ex scale factors");
1800 self.raw_query_multi_bit_exact_distance(
1801 id,
1802 dist,
1803 ex_bits,
1804 ex_codes,
1805 ex_add_factors,
1806 ex_scale_factors,
1807 )
1808 }
1809 }
1810 }
1811
1812 #[inline(always)]
1813 fn distance_all(&self, _: usize) -> Vec<f32> {
1814 let mut dists = Vec::new();
1815 let mut quantized_dists = Vec::new();
1816 let mut quantized_dists_table = Vec::new();
1817 let mut hacc_quantized_dists = Vec::new();
1818 self.distance_all_with_scratch(
1819 0,
1820 &mut dists,
1821 &mut quantized_dists,
1822 &mut quantized_dists_table,
1823 &mut hacc_quantized_dists,
1824 );
1825 dists
1826 }
1827
1828 #[inline(always)]
1829 #[allow(clippy::uninit_vec)]
1830 fn distance_all_with_scratch(
1831 &self,
1832 _: usize,
1833 dists: &mut Vec<f32>,
1834 quantized_dists: &mut Vec<u16>,
1835 quantized_dists_table: &mut Vec<u8>,
1836 hacc_quantized_dists: &mut Vec<u32>,
1837 ) {
1838 let code_len = rabit_binary_code_bytes(self.dim);
1839 let n = self.codes.len() / code_len;
1840 if n == 0 {
1841 dists.clear();
1842 quantized_dists.clear();
1843 return;
1844 }
1845
1846 if self.query_estimator == RabitQueryEstimator::ResidualQuery
1847 || self.num_bits == 1
1848 || self.approx_mode == ApproxMode::Fast
1849 {
1850 self.one_bit_distances_with_scratch(
1851 n,
1852 code_len,
1853 dists,
1854 quantized_dists,
1855 quantized_dists_table,
1856 hacc_quantized_dists,
1857 );
1858 return;
1859 }
1860
1861 let simd_len = self.binary_distances_with_scratch(
1862 n,
1863 code_len,
1864 dists,
1865 quantized_dists,
1866 quantized_dists_table,
1867 hacc_quantized_dists,
1868 );
1869
1870 self.apply_raw_query_multi_bit_distances(
1871 simd_len,
1872 dists,
1873 quantized_dists,
1874 quantized_dists_table,
1875 );
1876 }
1877
1878 #[allow(clippy::too_many_arguments)]
1879 fn accumulate_topk_with_scratch(
1880 &self,
1881 k: usize,
1882 lower_bound: Option<f32>,
1883 upper_bound: Option<f32>,
1884 row_id: impl Fn(u32) -> u64,
1885 res: &mut BinaryHeap<OrderedNode<u64>>,
1886 dists: &mut Vec<f32>,
1887 quantized_dists: &mut Vec<u16>,
1888 quantized_dists_table: &mut Vec<u8>,
1889 hacc_quantized_dists: &mut Vec<u32>,
1890 ) {
1891 if k == 0 {
1892 return;
1893 }
1894 if let Some(reason) = self.raw_query_lower_bound_gating_disabled_reason() {
1895 record_rabit_prune_bypass(reason);
1896 self.distance_all_with_scratch(
1897 k,
1898 dists,
1899 quantized_dists,
1900 quantized_dists_table,
1901 hacc_quantized_dists,
1902 );
1903 accumulate_distances_into_heap(k, lower_bound, upper_bound, row_id, res, dists);
1904 return;
1905 }
1906
1907 self.accumulate_raw_query_multi_bit_topk_dense_with_scratch(
1908 k,
1909 lower_bound,
1910 upper_bound,
1911 row_id,
1912 res,
1913 dists,
1914 quantized_dists,
1915 quantized_dists_table,
1916 hacc_quantized_dists,
1917 );
1918 }
1919
1920 #[allow(clippy::too_many_arguments)]
1921 fn accumulate_filtered_topk_with_scratch(
1922 &self,
1923 k: usize,
1924 lower_bound: Option<f32>,
1925 upper_bound: Option<f32>,
1926 row_ids: impl Iterator<Item = (u32, u64)>,
1927 accept_row: impl Fn(u64) -> bool,
1928 res: &mut BinaryHeap<OrderedNode<u64>>,
1929 dists: &mut Vec<f32>,
1930 quantized_dists: &mut Vec<u16>,
1931 quantized_dists_table: &mut Vec<u8>,
1932 hacc_quantized_dists: &mut Vec<u32>,
1933 ) {
1934 if k == 0 {
1935 return;
1936 }
1937 if let Some(reason) = self.raw_query_lower_bound_gating_disabled_reason() {
1938 record_rabit_prune_bypass(reason);
1939 self.distance_all_with_scratch(
1940 k,
1941 dists,
1942 quantized_dists,
1943 quantized_dists_table,
1944 hacc_quantized_dists,
1945 );
1946 accumulate_filtered_distances_into_heap(
1947 k,
1948 lower_bound,
1949 upper_bound,
1950 row_ids,
1951 accept_row,
1952 res,
1953 dists,
1954 );
1955 return;
1956 }
1957
1958 self.accumulate_raw_query_multi_bit_topk_with_scratch(
1959 k,
1960 lower_bound,
1961 upper_bound,
1962 row_ids
1963 .filter(|(_, row_id)| accept_row(*row_id))
1964 .map(|(id, row_id)| (id as usize, row_id)),
1965 res,
1966 dists,
1967 quantized_dists,
1968 quantized_dists_table,
1969 hacc_quantized_dists,
1970 );
1971 }
1972}
1973
1974fn accumulate_distances_into_heap(
1975 k: usize,
1976 lower_bound: Option<f32>,
1977 upper_bound: Option<f32>,
1978 row_id: impl Fn(u32) -> u64,
1979 res: &mut BinaryHeap<OrderedNode<u64>>,
1980 dists: &[f32],
1981) {
1982 let lower_bound = lower_bound.unwrap_or(f32::MIN).into();
1983 let upper_bound = upper_bound.unwrap_or(f32::MAX).into();
1984 let mut max_dist = res.peek().map(|node| node.dist);
1985 for (id, dist) in dists.iter().copied().enumerate() {
1986 let dist = OrderedFloat(dist);
1987 if dist < lower_bound || dist >= upper_bound {
1988 continue;
1989 }
1990 if res.len() < k {
1991 res.push(OrderedNode::new(row_id(id as u32), dist));
1992 if res.len() == k {
1993 max_dist = res.peek().map(|node| node.dist);
1994 }
1995 } else if max_dist.is_some_and(|max_dist| max_dist > dist) {
1996 res.pop();
1997 res.push(OrderedNode::new(row_id(id as u32), dist));
1998 max_dist = res.peek().map(|node| node.dist);
1999 }
2000 }
2001}
2002
2003fn accumulate_filtered_distances_into_heap(
2004 k: usize,
2005 lower_bound: Option<f32>,
2006 upper_bound: Option<f32>,
2007 row_ids: impl Iterator<Item = (u32, u64)>,
2008 accept_row: impl Fn(u64) -> bool,
2009 res: &mut BinaryHeap<OrderedNode<u64>>,
2010 dists: &[f32],
2011) {
2012 let lower_bound = lower_bound.unwrap_or(f32::MIN).into();
2013 let upper_bound = upper_bound.unwrap_or(f32::MAX).into();
2014 let mut max_dist = res.peek().map(|node| node.dist);
2015 for (id, row_id) in row_ids {
2016 if !accept_row(row_id) {
2017 continue;
2018 }
2019 let Some(dist) = dists.get(id as usize).copied() else {
2020 continue;
2021 };
2022 let dist = OrderedFloat(dist);
2023 if dist < lower_bound || dist >= upper_bound {
2024 continue;
2025 }
2026 if res.len() < k {
2027 res.push(OrderedNode::new(row_id, dist));
2028 if res.len() == k {
2029 max_dist = res.peek().map(|node| node.dist);
2030 }
2031 } else if max_dist.is_some_and(|max_dist| max_dist > dist) {
2032 res.pop();
2033 res.push(OrderedNode::new(row_id, dist));
2034 max_dist = res.peek().map(|node| node.dist);
2035 }
2036 }
2037}
2038
2039impl VectorStore for RabitQuantizationStorage {
2040 type DistanceCalculator<'a> = RabitDistCalculator<'a>;
2041
2042 fn as_any(&self) -> &dyn std::any::Any {
2043 self
2044 }
2045
2046 fn schema(&self) -> &SchemaRef {
2047 self.batch.schema_ref()
2048 }
2049
2050 fn to_batches(&self) -> Result<impl Iterator<Item = RecordBatch> + Send> {
2051 Ok(std::iter::once(self.batch.clone()))
2052 }
2053
2054 fn append_batch(&self, _batch: RecordBatch, _vector_column: &str) -> Result<Self> {
2055 unimplemented!("RabitQ does not support append_batch")
2056 }
2057
2058 fn len(&self) -> usize {
2059 self.batch.num_rows()
2060 }
2061
2062 fn row_id(&self, id: u32) -> u64 {
2063 self.row_ids.value(id as usize)
2064 }
2065
2066 fn row_ids(&self) -> impl Iterator<Item = &u64> {
2067 self.row_ids.values().iter()
2068 }
2069
2070 fn distance_type(&self) -> DistanceType {
2071 self.distance_type
2072 }
2073
2074 #[inline(never)]
2076 fn dist_calculator(&self, qr: Arc<dyn Array>, dist_q_c: f32) -> Self::DistanceCalculator<'_> {
2077 let code_dim = self.code_dim();
2078 let rotated_qr = self.rotate_query_vector(code_dim, &qr);
2079 let dist_table = build_dist_table_direct::<Float32Type>(&rotated_qr);
2080 let query_factor = match self.metadata.query_estimator {
2081 RabitQueryEstimator::ResidualQuery => self.residual_query_factor(dist_q_c),
2082 RabitQueryEstimator::RawQuery => self.raw_query_factor(dist_q_c, &rotated_qr, None),
2083 };
2084 let query_error = match self.metadata.query_estimator {
2085 RabitQueryEstimator::ResidualQuery => 0.0,
2086 RabitQueryEstimator::RawQuery => {
2087 self.raw_query_error_for_gating(dist_q_c, &rotated_qr, None)
2088 }
2089 };
2090 let sum_q = rotated_qr.iter().copied().sum();
2091 let ex_query = if code_dim.is_multiple_of(EX_DOT_BLOCK_DIMS) {
2094 rotated_qr
2095 } else {
2096 let mut padded = vec![0.0; padded_query_len(code_dim)];
2097 pad_query_into(&rotated_qr, &mut padded);
2098 padded
2099 };
2100
2101 self.distance_calculator_from_parts(RabitDistCalculatorParts {
2102 dim: code_dim,
2103 dist_table: Cow::Owned(dist_table),
2104 ex_query: Cow::Owned(ex_query),
2105 sum_q,
2106 query_factor,
2107 query_error,
2108 approx_mode: ApproxMode::Normal,
2109 })
2110 }
2111
2112 #[inline(never)]
2114 fn dist_calculator_with_scratch<'a>(
2115 &'a self,
2116 qr: Arc<dyn Array>,
2117 dist_q_c: f32,
2118 residual: Option<QueryResidual<'a>>,
2119 f32_scratch: &'a mut Vec<f32>,
2120 options: DistanceCalculatorOptions,
2121 ) -> Self::DistanceCalculator<'a> {
2122 let code_dim = self.code_dim();
2123 if let (
2124 RabitQueryEstimator::RawQuery,
2125 Some(QueryResidual::RabitRawQuery {
2126 rotated_centroid,
2127 query: Some(raw_query),
2128 }),
2129 ) = (self.metadata.query_estimator, residual)
2130 {
2131 debug_assert_eq!(raw_query.code_dim, code_dim);
2132 debug_assert_eq!(raw_query.ex_bits, self.metadata.num_bits - 1);
2133 let query_factor =
2134 self.raw_query_factor(dist_q_c, &raw_query.rotated_query, rotated_centroid);
2135 let query_error = self.raw_query_error_for_gating(
2136 dist_q_c,
2137 &raw_query.rotated_query,
2138 rotated_centroid,
2139 );
2140 return self.distance_calculator_from_parts(RabitDistCalculatorParts {
2141 dim: code_dim,
2142 dist_table: Cow::Borrowed(&raw_query.dist_table),
2143 ex_query: Cow::Borrowed(kernel_query(
2144 &raw_query.rotated_query,
2145 &raw_query.ex_query,
2146 )),
2147 sum_q: raw_query.sum_q,
2148 query_factor,
2149 query_error,
2150 approx_mode: options.approx_mode,
2151 });
2152 }
2153
2154 let dist_table_len = code_dim * 4;
2155 let ex_bits = self.metadata.num_bits - 1;
2156 let ex_query_table_len = if ex_bits == 0 || code_dim.is_multiple_of(EX_DOT_BLOCK_DIMS) {
2159 0
2160 } else {
2161 padded_query_len(code_dim)
2162 };
2163 f32_scratch.resize(code_dim + dist_table_len + ex_query_table_len, 0.0);
2164
2165 let query_factor;
2166 let query_error;
2167 let sum_q = {
2168 let (rotated_qr, remaining) = f32_scratch.split_at_mut(code_dim);
2169 let (dist_table, ex_query) = remaining.split_at_mut(dist_table_len);
2170 match residual {
2171 Some(QueryResidual::Centroid(residual_centroid)) => {
2172 self.rotate_query_vector_into(
2173 code_dim,
2174 &qr,
2175 Some(residual_centroid),
2176 rotated_qr,
2177 );
2178 }
2179 Some(QueryResidual::RabitRawQuery { .. }) | None => {
2180 self.rotate_query_vector_into(code_dim, &qr, None, rotated_qr);
2181 }
2182 }
2183 query_factor = match (self.metadata.query_estimator, residual) {
2184 (RabitQueryEstimator::ResidualQuery, _) => self.residual_query_factor(dist_q_c),
2185 (
2186 RabitQueryEstimator::RawQuery,
2187 Some(QueryResidual::RabitRawQuery {
2188 rotated_centroid, ..
2189 }),
2190 ) => self.raw_query_factor(dist_q_c, rotated_qr, rotated_centroid),
2191 (RabitQueryEstimator::RawQuery, _) => {
2192 self.raw_query_factor(dist_q_c, rotated_qr, None)
2193 }
2194 };
2195 query_error = match (self.metadata.query_estimator, residual) {
2196 (RabitQueryEstimator::ResidualQuery, _) => 0.0,
2197 (
2198 RabitQueryEstimator::RawQuery,
2199 Some(QueryResidual::RabitRawQuery {
2200 rotated_centroid, ..
2201 }),
2202 ) => self.raw_query_error_for_gating(dist_q_c, rotated_qr, rotated_centroid),
2203 (RabitQueryEstimator::RawQuery, _) => {
2204 self.raw_query_error_for_gating(dist_q_c, rotated_qr, None)
2205 }
2206 };
2207 build_dist_table_direct_into::<Float32Type>(rotated_qr, dist_table);
2208 if ex_query_table_len > 0 {
2209 pad_query_into(rotated_qr, ex_query);
2210 }
2211 rotated_qr.iter().copied().sum()
2212 };
2213
2214 let ex_query_start = code_dim + dist_table_len;
2215 self.distance_calculator_from_parts(RabitDistCalculatorParts {
2216 dim: code_dim,
2217 dist_table: Cow::Borrowed(&f32_scratch[code_dim..ex_query_start]),
2218 ex_query: Cow::Borrowed(kernel_query(
2219 &f32_scratch[..code_dim],
2220 &f32_scratch[ex_query_start..ex_query_start + ex_query_table_len],
2221 )),
2222 sum_q,
2223 query_factor,
2224 query_error,
2225 approx_mode: options.approx_mode,
2226 })
2227 }
2228
2229 fn dist_calculator_from_id(&self, _: u32) -> Self::DistanceCalculator<'_> {
2232 unimplemented!("RabitQ does not support dist_calculator_from_id")
2233 }
2234}
2235
2236const LOWBIT_IDX: [usize; 16] = {
2237 let mut array = [0; 16];
2238 let mut i = 1;
2239 while i < 16 {
2240 array[i] = i.trailing_zeros() as usize;
2241 i += 1;
2242 }
2243 array
2244};
2245
2246fn get_column(
2247 quantization_code: &[u8],
2248 code_len: usize,
2249 row: usize,
2250 col_idx: usize,
2251 codes: &mut [u8; 32],
2252) {
2253 for (i, code) in codes.iter_mut().enumerate() {
2254 let vec_idx = row + i;
2255 *code = quantization_code[vec_idx * code_len + col_idx];
2256 }
2257}
2258
2259pub fn pack_codes(codes: &FixedSizeListArray) -> FixedSizeListArray {
2260 let code_len = codes.value_length() as usize;
2261
2262 let num_blocks = codes.len() / BATCH_SIZE;
2264 let num_packed_vectors = num_blocks * BATCH_SIZE;
2265
2266 let mut blocks = vec![0u8; codes.values().len()];
2272
2273 let codes_values = codes
2274 .slice(0, num_packed_vectors)
2275 .values()
2276 .as_primitive::<UInt8Type>()
2277 .clone();
2278 let codes_values = codes_values.values();
2279
2280 let mut col = [0u8; 32];
2283 let mut col_0 = [0u8; 32]; let mut col_1 = [0u8; 32]; for row in (0..num_packed_vectors).step_by(BATCH_SIZE) {
2286 for i in 0..code_len {
2290 get_column(codes_values, code_len, row, i, &mut col);
2291
2292 for j in 0..32 {
2293 col_0[j] = col[j] & 0xF;
2294 col_1[j] = col[j] >> 4;
2295 }
2296
2297 let block_offset = (row / BATCH_SIZE) * code_len * BATCH_SIZE + i * BATCH_SIZE;
2298 for j in 0..16 {
2299 let val0 = col_0[PERM0[j]] | (col_0[PERM0[j] + 16] << 4);
2302 let val1 = col_1[PERM0[j]] | (col_1[PERM0[j] + 16] << 4);
2303 blocks[block_offset + j] = val0;
2304 blocks[block_offset + j + 16] = val1;
2305 }
2306 }
2307 }
2308
2309 let transposed_codes = transpose(
2311 &codes.values().as_primitive::<UInt8Type>().slice(
2312 num_packed_vectors * code_len,
2313 (codes.len() - num_packed_vectors) * code_len,
2314 ),
2315 codes.len() - num_packed_vectors,
2316 code_len,
2317 );
2318
2319 let offset = codes.values().len() - transposed_codes.len();
2320 for (i, v) in transposed_codes.values().iter().enumerate() {
2321 blocks[offset + i] = *v;
2322 }
2323
2324 assert_eq!(blocks.len(), codes.values().len());
2325 FixedSizeListArray::try_new_from_values(UInt8Array::from(blocks), code_len as i32).unwrap()
2326}
2327
2328pub fn unpack_codes(codes: &FixedSizeListArray) -> FixedSizeListArray {
2330 let code_len = codes.value_length() as usize;
2331 let num_vectors = codes.len();
2332
2333 let num_blocks = num_vectors / BATCH_SIZE;
2335 let num_packed_vectors = num_blocks * BATCH_SIZE;
2336
2337 let mut unpacked = vec![0u8; codes.values().len()];
2338
2339 let codes_values = codes.values().as_primitive::<UInt8Type>().values();
2340
2341 for batch_idx in 0..num_blocks {
2343 let block_start = batch_idx * code_len * BATCH_SIZE;
2344
2345 for i in 0..code_len {
2346 let block_offset = block_start + i * BATCH_SIZE;
2347 let block = &codes_values[block_offset..block_offset + BATCH_SIZE];
2348
2349 for j in 0..16 {
2351 let val0 = block[j];
2352 let val1 = block[j + 16];
2353
2354 let low_0 = val0 & 0xF;
2355 let high_0 = val0 >> 4;
2356 let low_1 = val1 & 0xF;
2357 let high_1 = val1 >> 4;
2358
2359 let vec_idx_0 = batch_idx * BATCH_SIZE + PERM0[j];
2360 let vec_idx_1 = batch_idx * BATCH_SIZE + PERM0[j] + 16;
2361
2362 unpacked[vec_idx_0 * code_len + i] = low_0 | (low_1 << 4);
2363 unpacked[vec_idx_1 * code_len + i] = high_0 | (high_1 << 4);
2364 }
2365 }
2366 }
2367
2368 if num_packed_vectors < num_vectors {
2370 let remainder = num_vectors - num_packed_vectors;
2371 let offset = num_packed_vectors * code_len;
2372 let transposed_data = &codes_values[offset..];
2373
2374 for row in 0..remainder {
2376 for col in 0..code_len {
2377 unpacked[offset + row * code_len + col] = transposed_data[col * remainder + row];
2378 }
2379 }
2380 }
2381
2382 FixedSizeListArray::try_new_from_values(UInt8Array::from(unpacked), code_len as i32).unwrap()
2383}
2384
2385fn build_frag_reuse_mapping(
2394 fri: Option<&FragReuseIndex>,
2395 row_ids: &UInt64Array,
2396) -> Option<HashMap<u64, Option<u64>>> {
2397 let fri = fri?;
2398 if fri.row_id_maps.is_empty() {
2399 return None;
2400 }
2401 let mut mapping: HashMap<u64, Option<u64>> = HashMap::new();
2402 for row_id in row_ids.values().iter() {
2403 match fri.remap_row_id(*row_id) {
2404 Some(new_id) if new_id == *row_id => {}
2405 mapped => {
2406 mapping.insert(*row_id, mapped);
2407 }
2408 }
2409 }
2410 if mapping.is_empty() {
2411 None
2412 } else {
2413 Some(mapping)
2414 }
2415}
2416
2417#[async_trait]
2418impl QuantizerStorage for RabitQuantizationStorage {
2419 type Metadata = RabitQuantizationMetadata;
2420
2421 fn try_from_batch(
2422 batch: RecordBatch,
2423 metadata: &Self::Metadata,
2424 distance_type: DistanceType,
2425 fri: Option<Arc<FragReuseIndex>>,
2426 ) -> Result<Self> {
2427 let distance_type = match (metadata.query_estimator, distance_type) {
2428 (RabitQueryEstimator::RawQuery, DistanceType::Cosine) => DistanceType::L2,
2429 _ => distance_type,
2430 };
2431 validate_rq_num_bits(metadata.num_bits)?;
2432 let row_ids = batch[ROW_ID].as_primitive::<UInt64Type>().clone();
2433 let codes = batch[RABIT_CODE_COLUMN].as_fixed_size_list().clone();
2434 let expected_code_bytes = metadata.binary_code_bytes();
2435 if expected_code_bytes > 0 && codes.value_length() as usize != expected_code_bytes {
2436 return Err(Error::invalid_input(format!(
2437 "RabitQ code byte width mismatch: column {} has {} bytes, metadata rotated_dim={} requires {} bytes",
2438 RABIT_CODE_COLUMN,
2439 codes.value_length(),
2440 metadata.rotated_dim(),
2441 expected_code_bytes
2442 )));
2443 }
2444 let add_factors = batch[ADD_FACTORS_COLUMN]
2445 .as_primitive::<Float32Type>()
2446 .clone();
2447 let scale_factors = batch[SCALE_FACTORS_COLUMN]
2448 .as_primitive::<Float32Type>()
2449 .clone();
2450 let error_factors = batch
2451 .column_by_name(ERROR_FACTORS_COLUMN)
2452 .map(|factors| factors.as_primitive::<Float32Type>().clone());
2453 let ex_bits = rabit_ex_bits(metadata.num_bits)?;
2454 let mut batch = batch;
2455 let mut ex_codes = None;
2456 let mut ex_add_factors = None;
2457 let mut ex_scale_factors = None;
2458 if ex_bits != 0 {
2459 let (normalized_batch, codes) =
2460 load_blocked_ex_codes(batch, metadata.rotated_dim(), metadata.num_bits)?;
2461 batch = normalized_batch;
2462 ex_codes = Some(codes);
2463 ex_add_factors = Some(
2464 batch
2465 .column_by_name(EX_ADD_FACTORS_COLUMN)
2466 .ok_or_else(|| {
2467 Error::invalid_input(format!(
2468 "RabitQ num_bits={} requires {} column",
2469 metadata.num_bits, EX_ADD_FACTORS_COLUMN
2470 ))
2471 })?
2472 .as_primitive::<Float32Type>()
2473 .clone(),
2474 );
2475 ex_scale_factors = Some(
2476 batch
2477 .column_by_name(EX_SCALE_FACTORS_COLUMN)
2478 .ok_or_else(|| {
2479 Error::invalid_input(format!(
2480 "RabitQ num_bits={} requires {} column",
2481 metadata.num_bits, EX_SCALE_FACTORS_COLUMN
2482 ))
2483 })?
2484 .as_primitive::<Float32Type>()
2485 .clone(),
2486 );
2487 } else if metadata.query_estimator == RabitQueryEstimator::RawQuery {
2488 if batch.column_by_name(EX_ADD_FACTORS_COLUMN).is_some()
2489 || batch.column_by_name(EX_SCALE_FACTORS_COLUMN).is_some()
2490 || batch.column_by_name(RABIT_EX_CODE_COLUMN).is_some()
2491 || batch.column_by_name(RABIT_BLOCKED_EX_CODE_COLUMN).is_some()
2492 {
2493 return Err(Error::invalid_input(
2494 "RabitQ num_bits=1 raw-query indexes must not contain ex-code columns"
2495 .to_string(),
2496 ));
2497 }
2498 } else if batch.column_by_name(RABIT_EX_CODE_COLUMN).is_some()
2499 || batch.column_by_name(RABIT_BLOCKED_EX_CODE_COLUMN).is_some()
2500 {
2501 return Err(Error::invalid_input(format!(
2502 "RabitQ num_bits={} does not support ex-code columns",
2503 metadata.num_bits
2504 )));
2505 }
2506
2507 let (batch, codes) = if !metadata.packed {
2508 let codes = pack_codes(&codes);
2509 let batch = batch.replace_column_by_name(RABIT_CODE_COLUMN, Arc::new(codes))?;
2510 let codes = batch[RABIT_CODE_COLUMN].as_fixed_size_list().clone();
2511 (batch, codes)
2512 } else {
2513 (batch, codes)
2514 };
2515
2516 let mut metadata = metadata.clone();
2517 metadata.packed = true;
2518 let packed_ex_codes =
2519 maybe_pack_ex_codes(ex_codes.as_ref(), ex_bits, error_factors.as_ref());
2520
2521 let storage = Self {
2522 metadata,
2523 batch,
2524 distance_type,
2525 row_ids,
2526 codes,
2527 add_factors,
2528 scale_factors,
2529 error_factors,
2530 ex_codes,
2531 packed_ex_codes,
2532 ex_add_factors,
2533 ex_scale_factors,
2534 };
2535
2536 match build_frag_reuse_mapping(fri.as_deref(), &storage.row_ids) {
2537 Some(mapping) => storage.remap(&mapping),
2538 None => Ok(storage),
2539 }
2540 }
2541
2542 fn metadata(&self) -> &Self::Metadata {
2543 &self.metadata
2544 }
2545
2546 async fn load_partition(
2547 reader: &PreviousFileReader,
2548 range: std::ops::Range<usize>,
2549 distance_type: DistanceType,
2550 metadata: &Self::Metadata,
2551 frag_reuse_index: Option<Arc<FragReuseIndex>>,
2552 ) -> Result<Self> {
2553 let schema = reader.schema();
2554 let batch = reader.read_range(range, schema).await?;
2555 Self::try_from_batch(batch, metadata, distance_type, frag_reuse_index)
2556 }
2557
2558 fn remap(&self, mapping: &HashMap<u64, Option<u64>>) -> Result<Self> {
2559 let num_vectors = self.codes.len();
2560 let num_code_bytes = self.codes.value_length() as usize;
2561 let codes = self.codes.values().as_primitive::<UInt8Type>().values();
2562 let mut indices = Vec::with_capacity(num_vectors);
2563 let mut new_row_ids = Vec::with_capacity(num_vectors);
2564 let mut new_codes = Vec::with_capacity(codes.len());
2565
2566 let row_ids = self.row_ids.values();
2567 for (i, row_id) in row_ids.iter().enumerate() {
2568 match mapping.get(row_id) {
2569 Some(Some(new_id)) => {
2570 indices.push(i as u32);
2571 new_row_ids.push(*new_id);
2572 new_codes.extend(get_rq_code(codes, i, num_vectors, num_code_bytes));
2573 }
2574 Some(None) => {}
2575 None => {
2576 indices.push(i as u32);
2577 new_row_ids.push(*row_id);
2578 new_codes.extend(get_rq_code(codes, i, num_vectors, num_code_bytes));
2579 }
2580 }
2581 }
2582
2583 let new_row_ids = UInt64Array::from(new_row_ids);
2584 let new_codes = FixedSizeListArray::try_new_from_values(
2585 UInt8Array::from(new_codes),
2586 num_code_bytes as i32,
2587 )?;
2588 let batch = if new_row_ids.is_empty() {
2589 RecordBatch::new_empty(self.schema().clone())
2590 } else {
2591 let codes = Arc::new(pack_codes(&new_codes));
2592 self.batch
2593 .take(&UInt32Array::from(indices))?
2594 .replace_column_by_name(ROW_ID, Arc::new(new_row_ids.clone()))?
2595 .replace_column_by_name(RABIT_CODE_COLUMN, codes)?
2596 };
2597 let codes = batch[RABIT_CODE_COLUMN].as_fixed_size_list().clone();
2598 let add_factors = batch[ADD_FACTORS_COLUMN]
2599 .as_primitive::<Float32Type>()
2600 .clone();
2601 let scale_factors = batch[SCALE_FACTORS_COLUMN]
2602 .as_primitive::<Float32Type>()
2603 .clone();
2604 let error_factors = batch
2605 .column_by_name(ERROR_FACTORS_COLUMN)
2606 .map(|factors| factors.as_primitive::<Float32Type>().clone());
2607 let ex_bits = rabit_ex_bits(self.metadata.num_bits)?;
2608 let (batch, ex_codes) = if ex_bits == 0 {
2609 (batch, None)
2610 } else {
2611 let (batch, codes) =
2614 load_blocked_ex_codes(batch, self.metadata.rotated_dim(), self.metadata.num_bits)?;
2615 (batch, Some(codes))
2616 };
2617 let packed_ex_codes =
2618 maybe_pack_ex_codes(ex_codes.as_ref(), ex_bits, error_factors.as_ref());
2619 let ex_add_factors = batch
2620 .column_by_name(EX_ADD_FACTORS_COLUMN)
2621 .map(|factors| factors.as_primitive::<Float32Type>().clone());
2622 let ex_scale_factors = batch
2623 .column_by_name(EX_SCALE_FACTORS_COLUMN)
2624 .map(|factors| factors.as_primitive::<Float32Type>().clone());
2625
2626 Ok(Self {
2627 metadata: self.metadata.clone(),
2628 distance_type: self.distance_type,
2629 batch,
2630 codes,
2631 add_factors,
2632 scale_factors,
2633 error_factors,
2634 ex_codes,
2635 packed_ex_codes,
2636 ex_add_factors,
2637 ex_scale_factors,
2638 row_ids: new_row_ids,
2639 })
2640 }
2641}
2642
2643#[inline]
2649fn compute_single_rq_distance(
2650 codes: &[u8],
2651 id: usize,
2652 num_vectors: usize,
2653 num_code_bytes: usize,
2654 dist_table: &[f32],
2655) -> f32 {
2656 let remainder = num_vectors % BATCH_SIZE;
2657 let mut dist_table_iter = dist_table.chunks_exact(SEGMENT_NUM_CODES).tuples();
2658
2659 if id < num_vectors - remainder {
2660 let batch_codes = &codes[id / BATCH_SIZE * BATCH_SIZE * num_code_bytes
2661 ..(id / BATCH_SIZE + 1) * BATCH_SIZE * num_code_bytes];
2662
2663 let id_in_batch = id % BATCH_SIZE;
2664 let idx = PERM0_INVERSE[id_in_batch % 16];
2665 let is_lower = id_in_batch < 16;
2666
2667 let mut dist = 0.0f32;
2668 for block in batch_codes.chunks_exact(BATCH_SIZE) {
2669 let code_byte = if is_lower {
2670 (block[idx] & 0xF) | (block[idx + 16] << 4)
2671 } else {
2672 (block[idx] >> 4) | (block[idx + 16] & 0xF0)
2673 };
2674 if let Some((current_dt, next_dt)) = dist_table_iter.next() {
2675 let current_code = (code_byte & 0x0F) as usize;
2676 let next_code = (code_byte >> 4) as usize;
2677 dist += current_dt[current_code] + next_dt[next_code];
2678 }
2679 }
2680 dist
2681 } else {
2682 let offset_id = id - (num_vectors - remainder);
2683 let remainder_codes = &codes[(num_vectors - remainder) * num_code_bytes..];
2684
2685 let mut dist = 0.0f32;
2686 for &code_byte in remainder_codes.iter().skip(offset_id).step_by(remainder) {
2687 if let Some((current_dt, next_dt)) = dist_table_iter.next() {
2688 let current_code = (code_byte & 0x0F) as usize;
2689 let next_code = (code_byte >> 4) as usize;
2690 dist += current_dt[current_code] + next_dt[next_code];
2691 }
2692 }
2693 dist
2694 }
2695}
2696
2697#[inline]
2698fn get_rq_code(
2699 codes: &[u8],
2700 id: usize,
2701 num_vectors: usize,
2702 num_code_bytes: usize,
2703) -> impl Iterator<Item = u8> + '_ {
2704 let remainder = num_vectors % BATCH_SIZE;
2705
2706 if id < num_vectors - remainder {
2707 let codes = &codes[id / BATCH_SIZE * BATCH_SIZE * num_code_bytes
2709 ..(id / BATCH_SIZE + 1) * BATCH_SIZE * num_code_bytes];
2710
2711 let id_in_batch = id % BATCH_SIZE;
2712 if id_in_batch < 16 {
2713 let idx = PERM0_INVERSE[id_in_batch];
2714 codes
2715 .chunks_exact(BATCH_SIZE)
2716 .map(|block| (block[idx] & 0xF) | (block[idx + 16] << 4))
2717 .exact_size(num_code_bytes)
2718 .collect_vec()
2719 .into_iter()
2720 } else {
2721 let idx = PERM0_INVERSE[id_in_batch - 16];
2722 codes
2723 .chunks_exact(BATCH_SIZE)
2724 .map(|block| (block[idx] >> 4) | (block[idx + 16] & 0xF0))
2725 .exact_size(num_code_bytes)
2726 .collect_vec()
2727 .into_iter()
2728 }
2729 } else {
2730 let id = id - (num_vectors - remainder);
2731 let codes = &codes[(num_vectors - remainder) * num_code_bytes..];
2732 codes
2733 .iter()
2734 .skip(id)
2735 .step_by(remainder)
2736 .copied()
2737 .exact_size(num_code_bytes)
2738 .collect_vec()
2739 .into_iter()
2740 }
2741}
2742
2743#[cfg(test)]
2744mod tests {
2745 use super::*;
2746 use std::collections::{BinaryHeap, HashMap};
2747
2748 use arrow_array::{ArrayRef, Float32Array, Float64Array, UInt64Array};
2749 use lance_core::ROW_ID;
2750 use lance_linalg::distance::DistanceType;
2751 use rand::rngs::SmallRng;
2752 use rand::{Rng, SeedableRng};
2753 use rstest::rstest;
2754
2755 use crate::vector::bq::{RQRotationType, builder::RabitQuantizer};
2756 use crate::vector::quantizer::{Quantization, QuantizerStorage};
2757
2758 fn build_dist_table_not_optimized<T: ArrowFloatType>(
2759 sub_vec: &[T::Native],
2760 dist_table: &mut [f32],
2761 ) where
2762 T::Native: AsPrimitive<f32>,
2763 {
2764 for (j, dist) in dist_table.iter_mut().enumerate().take(SEGMENT_NUM_CODES) {
2765 for (k, v) in sub_vec.iter().enumerate().take(SEGMENT_LENGTH) {
2766 if j & (1 << k) != 0 {
2767 *dist += v.as_();
2768 }
2769 }
2770 }
2771 }
2772
2773 #[test]
2774 fn test_build_dist_table_not_optimized() {
2775 let sub_vec = vec![1.0, 2.0, 3.0, 4.0];
2776 let mut expected = vec![0.0; SEGMENT_NUM_CODES];
2777 build_dist_table_not_optimized::<Float32Type>(&sub_vec, &mut expected);
2778 let mut dist_table = vec![0.0; SEGMENT_NUM_CODES];
2779 build_dist_table_for_subvec::<Float32Type>(&sub_vec, &mut dist_table);
2780 assert_eq!(dist_table, expected);
2781 }
2782
2783 #[test]
2784 fn test_dist_calculator_with_scratch_matches_owned_and_reuses_buffer() {
2785 let code_dim = 64;
2786 let original_codes = make_test_codes(50, code_dim);
2787 let metadata = make_test_metadata(original_codes.value_length() as usize * 8);
2788 let storage = RabitQuantizationStorage::try_from_batch(
2789 make_test_batch(original_codes),
2790 &metadata,
2791 DistanceType::L2,
2792 None,
2793 )
2794 .unwrap();
2795 let query = Arc::new(Float32Array::from_iter_values(
2796 (0..code_dim).map(|idx| idx as f32 / code_dim as f32),
2797 )) as ArrayRef;
2798
2799 let expected = storage.dist_calculator(query.clone(), 0.25).distance_all(0);
2800 let expected_scratch_len = code_dim as usize + code_dim as usize * 4;
2801 let mut scratch = Vec::with_capacity(expected_scratch_len);
2802 let initial_ptr = scratch.as_ptr();
2803 {
2804 let calc = storage.dist_calculator_with_scratch(
2805 query.clone(),
2806 0.25,
2807 None,
2808 &mut scratch,
2809 DistanceCalculatorOptions::default(),
2810 );
2811 assert_eq!(calc.distance_all(0), expected);
2812 }
2813 assert_eq!(scratch.len(), expected_scratch_len);
2814 assert_eq!(scratch.as_ptr(), initial_ptr);
2815
2816 scratch.fill(f32::NAN);
2817 {
2818 let calc = storage.dist_calculator_with_scratch(
2819 query,
2820 0.25,
2821 None,
2822 &mut scratch,
2823 DistanceCalculatorOptions::default(),
2824 );
2825 assert_eq!(calc.distance_all(0), expected);
2826 }
2827 assert_eq!(scratch.as_ptr(), initial_ptr);
2828 }
2829
2830 #[test]
2831 fn test_dist_calculator_with_scratch_applies_residual_centroid_without_residual_array() {
2832 let code_dim = 64usize;
2833 let original_codes = make_test_codes(50, code_dim as i32);
2834 let mut metadata = make_test_metadata(original_codes.value_length() as usize * 8);
2835 metadata.query_estimator = RabitQueryEstimator::ResidualQuery;
2836 let storage = RabitQuantizationStorage::try_from_batch(
2837 make_test_batch(original_codes),
2838 &metadata,
2839 DistanceType::L2,
2840 None,
2841 )
2842 .unwrap();
2843 let query_values = (0..code_dim)
2844 .map(|idx| idx as f32 / code_dim as f32)
2845 .collect::<Vec<_>>();
2846 let centroid_values = (0..code_dim)
2847 .map(|idx| (idx % 7) as f32 / code_dim as f32)
2848 .collect::<Vec<_>>();
2849 let residual_values = query_values
2850 .iter()
2851 .zip(centroid_values.iter())
2852 .map(|(query, centroid)| query - centroid)
2853 .collect::<Vec<_>>();
2854 let query = Arc::new(Float32Array::from(query_values)) as ArrayRef;
2855 let centroid = Arc::new(Float32Array::from(centroid_values)) as ArrayRef;
2856 let residual = Arc::new(Float32Array::from(residual_values)) as ArrayRef;
2857
2858 let expected = storage.dist_calculator(residual, 0.25).distance_all(0);
2859 let mut scratch = Vec::new();
2860 let calc = storage.dist_calculator_with_scratch(
2861 query.clone(),
2862 0.25,
2863 Some(QueryResidual::Centroid(centroid.as_ref())),
2864 &mut scratch,
2865 DistanceCalculatorOptions::default(),
2866 );
2867
2868 assert_eq!(calc.distance_all(0), expected);
2869 }
2870
2871 #[test]
2872 fn test_dist_calculator_with_scratch_applies_float64_residual_before_f32_cast() {
2873 let code_dim = 64usize;
2874 let original_codes = make_test_codes(50, code_dim as i32);
2875 let mut metadata = make_test_metadata(original_codes.value_length() as usize * 8);
2876 metadata.query_estimator = RabitQueryEstimator::ResidualQuery;
2877 let storage = RabitQuantizationStorage::try_from_batch(
2878 make_test_batch(original_codes),
2879 &metadata,
2880 DistanceType::L2,
2881 None,
2882 )
2883 .unwrap();
2884 let query_values = (0..code_dim)
2885 .map(|idx| 1.0 + idx as f64 * 1.0e-9)
2886 .collect::<Vec<_>>();
2887 let centroid_values = vec![1.0; code_dim];
2888 let residual_values = query_values
2889 .iter()
2890 .zip(centroid_values.iter())
2891 .map(|(query, centroid)| query - centroid)
2892 .collect::<Vec<_>>();
2893 let query = Arc::new(Float64Array::from(query_values)) as ArrayRef;
2894 let centroid = Arc::new(Float64Array::from(centroid_values)) as ArrayRef;
2895 let residual = Arc::new(Float64Array::from(residual_values)) as ArrayRef;
2896
2897 let expected = storage.dist_calculator(residual, 0.25).distance_all(0);
2898 let mut scratch = Vec::new();
2899 let calc = storage.dist_calculator_with_scratch(
2900 query,
2901 0.25,
2902 Some(QueryResidual::Centroid(centroid.as_ref())),
2903 &mut scratch,
2904 DistanceCalculatorOptions::default(),
2905 );
2906
2907 assert_eq!(calc.distance_all(0), expected);
2908 }
2909
2910 #[test]
2911 fn test_pack_unpack_codes() {
2912 for num_vectors in [10, 32, 50, 64, 100] {
2914 let code_len = 8;
2915
2916 let mut codes_data = Vec::new();
2918 for i in 0..num_vectors {
2919 for j in 0..code_len {
2920 codes_data.push((i * code_len + j) as u8);
2921 }
2922 }
2923
2924 let original_codes = FixedSizeListArray::try_new_from_values(
2925 UInt8Array::from(codes_data.clone()),
2926 code_len,
2927 )
2928 .unwrap();
2929
2930 let packed = pack_codes(&original_codes);
2932 let unpacked = unpack_codes(&packed);
2933
2934 assert_eq!(original_codes.len(), unpacked.len());
2936 assert_eq!(original_codes.value_length(), unpacked.value_length());
2937
2938 let original_values = original_codes.values().as_primitive::<UInt8Type>().values();
2939 let unpacked_values = unpacked.values().as_primitive::<UInt8Type>().values();
2940
2941 assert_eq!(
2942 original_values, unpacked_values,
2943 "Mismatch for num_vectors={}",
2944 num_vectors
2945 );
2946 }
2947 }
2948
2949 #[test]
2950 fn test_rabit_split_code_fields() {
2951 let bin_field = rabit_binary_code_field(128);
2952 let DataType::FixedSizeList(_, bin_code_bytes) = bin_field.data_type() else {
2953 panic!("binary code field should be FixedSizeList");
2954 };
2955 assert_eq!(*bin_code_bytes, 16);
2956
2957 assert!(rabit_ex_code_field(128, 1).unwrap().is_none());
2958 let ex_field = rabit_ex_code_field(128, 9).unwrap().unwrap();
2959 assert_eq!(ex_field.name(), RABIT_BLOCKED_EX_CODE_COLUMN);
2960 let DataType::FixedSizeList(_, ex_code_bytes) = ex_field.data_type() else {
2961 panic!("ex-code field should be FixedSizeList");
2962 };
2963 assert_eq!(*ex_code_bytes, 128);
2964 }
2965
2966 fn make_test_codes(num_vectors: usize, code_dim: i32) -> FixedSizeListArray {
2967 let quantizer =
2968 RabitQuantizer::new_with_rotation::<Float32Type>(1, code_dim, RQRotationType::Fast);
2969 let values = Float32Array::from_iter_values(
2970 (0..num_vectors * code_dim as usize).map(|idx| idx as f32 / code_dim as f32),
2971 );
2972 let vectors = FixedSizeListArray::try_new_from_values(values, code_dim).unwrap();
2973 quantizer
2974 .quantize(&vectors)
2975 .unwrap()
2976 .as_fixed_size_list()
2977 .clone()
2978 }
2979
2980 fn make_test_metadata(code_dim: usize) -> RabitQuantizationMetadata {
2981 RabitQuantizer::new_with_rotation::<Float32Type>(1, code_dim as i32, RQRotationType::Fast)
2982 .metadata(None)
2983 }
2984
2985 #[test]
2986 fn test_rabit_metadata_defaults_old_indexes_to_residual_query() {
2987 let metadata: RabitQuantizationMetadata = serde_json::from_str(
2988 r#"{"rotate_mat_position":0,"rotation_type":"matrix","code_dim":64,"num_bits":1,"packed":true}"#,
2989 )
2990 .unwrap();
2991 assert_eq!(metadata.query_estimator, RabitQueryEstimator::ResidualQuery);
2992 }
2993
2994 #[test]
2995 fn test_new_rabit_metadata_uses_raw_query_estimator() {
2996 let metadata = make_test_metadata(64);
2997 assert_eq!(metadata.query_estimator, RabitQueryEstimator::RawQuery);
2998 }
2999
3000 fn make_test_batch(codes: FixedSizeListArray) -> RecordBatch {
3001 let num_rows = codes.len();
3002 RecordBatch::try_from_iter(vec![
3003 (
3004 ROW_ID,
3005 Arc::new(UInt64Array::from_iter_values(0..num_rows as u64)) as ArrayRef,
3006 ),
3007 (RABIT_CODE_COLUMN, Arc::new(codes) as ArrayRef),
3008 (
3009 ADD_FACTORS_COLUMN,
3010 Arc::new(Float32Array::from_iter_values(
3011 (0..num_rows).map(|v| v as f32),
3012 )) as ArrayRef,
3013 ),
3014 (
3015 SCALE_FACTORS_COLUMN,
3016 Arc::new(Float32Array::from_iter_values(
3017 (0..num_rows).map(|v| v as f32 + 0.5),
3018 )) as ArrayRef,
3019 ),
3020 (
3021 ERROR_FACTORS_COLUMN,
3022 Arc::new(Float32Array::from_iter_values(
3023 (0..num_rows).map(|v| v as f32 + 0.25),
3024 )) as ArrayRef,
3025 ),
3026 ])
3027 .unwrap()
3028 }
3029
3030 fn make_test_ex_codes(num_vectors: usize, code_dim: usize, num_bits: u8) -> FixedSizeListArray {
3031 let ex_bits = rabit_ex_bits(num_bits).unwrap();
3032 let ex_code_bytes = rabit_ex_code_bytes(code_dim, ex_bits).unwrap();
3033 let values = (0..num_vectors * ex_code_bytes)
3034 .map(|idx| (idx % 251) as u8)
3035 .collect::<Vec<_>>();
3036 FixedSizeListArray::try_new_from_values(UInt8Array::from(values), ex_code_bytes as i32)
3037 .unwrap()
3038 }
3039
3040 fn make_test_batch_with_ex(
3041 codes: FixedSizeListArray,
3042 ex_codes: FixedSizeListArray,
3043 ) -> RecordBatch {
3044 let num_rows = codes.len();
3045 RecordBatch::try_from_iter(vec![
3046 (
3047 ROW_ID,
3048 Arc::new(UInt64Array::from_iter_values(0..num_rows as u64)) as ArrayRef,
3049 ),
3050 (RABIT_CODE_COLUMN, Arc::new(codes) as ArrayRef),
3051 (
3052 ADD_FACTORS_COLUMN,
3053 Arc::new(Float32Array::from_iter_values(
3054 (0..num_rows).map(|v| v as f32),
3055 )) as ArrayRef,
3056 ),
3057 (
3058 SCALE_FACTORS_COLUMN,
3059 Arc::new(Float32Array::from_iter_values(
3060 (0..num_rows).map(|v| v as f32 + 0.5),
3061 )) as ArrayRef,
3062 ),
3063 (
3064 ERROR_FACTORS_COLUMN,
3065 Arc::new(Float32Array::from_iter_values(
3066 (0..num_rows).map(|v| v as f32 + 0.25),
3067 )) as ArrayRef,
3068 ),
3069 (RABIT_EX_CODE_COLUMN, Arc::new(ex_codes) as ArrayRef),
3070 (
3071 EX_ADD_FACTORS_COLUMN,
3072 Arc::new(Float32Array::from_iter_values(
3073 (0..num_rows).map(|v| v as f32 + 10.5),
3074 )) as ArrayRef,
3075 ),
3076 (
3077 EX_SCALE_FACTORS_COLUMN,
3078 Arc::new(Float32Array::from_iter_values(
3079 (0..num_rows).map(|v| v as f32 + 1.5),
3080 )) as ArrayRef,
3081 ),
3082 ])
3083 .unwrap()
3084 }
3085
3086 fn assert_codes_eq(actual: &FixedSizeListArray, expected: &FixedSizeListArray) {
3087 assert_eq!(actual.len(), expected.len());
3088 assert_eq!(actual.value_length(), expected.value_length());
3089 assert_eq!(
3090 actual.values().as_primitive::<UInt8Type>().values(),
3091 expected.values().as_primitive::<UInt8Type>().values()
3092 );
3093 }
3094
3095 #[test]
3096 fn test_raw_query_multi_bit_distance_uses_ex_factors() {
3097 let code_dim = 8usize;
3098 let identity = Float32Array::from_iter_values(
3099 (0..code_dim)
3100 .flat_map(|row| (0..code_dim).map(move |col| if row == col { 1.0 } else { 0.0 })),
3101 );
3102 let rotate_mat =
3103 FixedSizeListArray::try_new_from_values(identity, code_dim as i32).unwrap();
3104 let metadata = RabitQuantizationMetadata {
3105 rotate_mat: Some(rotate_mat),
3106 rotate_mat_position: None,
3107 fast_rotation_signs: None,
3108 rotation_type: RQRotationType::Matrix,
3109 code_dim: code_dim as u32,
3110 num_bits: 2,
3111 packed: false,
3112 query_estimator: RabitQueryEstimator::RawQuery,
3113 };
3114 let codes =
3115 FixedSizeListArray::try_new_from_values(UInt8Array::from(vec![0xff, 0xff]), 1).unwrap();
3116 let ex_codes =
3117 FixedSizeListArray::try_new_from_values(UInt8Array::from(vec![0x00, 0xff]), 1).unwrap();
3118 let batch = RecordBatch::try_from_iter(vec![
3119 (ROW_ID, Arc::new(UInt64Array::from(vec![0, 1])) as ArrayRef),
3120 (RABIT_CODE_COLUMN, Arc::new(codes) as ArrayRef),
3121 (
3122 ADD_FACTORS_COLUMN,
3123 Arc::new(Float32Array::from(vec![0.0, 0.0])) as ArrayRef,
3124 ),
3125 (
3126 SCALE_FACTORS_COLUMN,
3127 Arc::new(Float32Array::from(vec![0.0, 0.0])) as ArrayRef,
3128 ),
3129 (RABIT_EX_CODE_COLUMN, Arc::new(ex_codes) as ArrayRef),
3130 (
3131 EX_ADD_FACTORS_COLUMN,
3132 Arc::new(Float32Array::from(vec![100.0, 10.0])) as ArrayRef,
3133 ),
3134 (
3135 EX_SCALE_FACTORS_COLUMN,
3136 Arc::new(Float32Array::from(vec![1.0, 1.0])) as ArrayRef,
3137 ),
3138 ])
3139 .unwrap();
3140 let storage =
3141 RabitQuantizationStorage::try_from_batch(batch, &metadata, DistanceType::L2, None)
3142 .unwrap();
3143 let query = Arc::new(Float32Array::from(vec![1.0; code_dim])) as ArrayRef;
3144 let calc = storage.dist_calculator(query, 0.0);
3145
3146 assert_eq!(calc.distance(0), 104.0);
3147 assert_eq!(calc.distance(1), 22.0);
3148 let mut distances = Vec::new();
3149 let mut u16_scratch = Vec::new();
3150 let mut u8_scratch = Vec::new();
3151 let mut u32_scratch = Vec::new();
3152 calc.distance_all_with_scratch(
3153 0,
3154 &mut distances,
3155 &mut u16_scratch,
3156 &mut u8_scratch,
3157 &mut u32_scratch,
3158 );
3159 assert_eq!(distances, vec![104.0, 22.0]);
3160 }
3161
3162 #[test]
3169 fn test_raw_query_multi_bit_distance_matches_reference_for_all_ex_widths() {
3170 use rand::rngs::SmallRng;
3171 use rand::{Rng, SeedableRng};
3172
3173 for (code_dim, num_rows) in [(72usize, 33usize), (1536, 33)] {
3178 for num_bits in 2..=9u8 {
3179 for legacy_format in [false, true] {
3180 let ex_bits = num_bits - 1;
3181 let mut rng = SmallRng::seed_from_u64(num_bits as u64);
3182
3183 let sign_bits = (0..num_rows * code_dim)
3184 .map(|_| rng.random_bool(0.5))
3185 .collect::<Vec<_>>();
3186 let max_code = ((1u16 << ex_bits) - 1) as u8;
3187 let ex_values = (0..num_rows * code_dim)
3188 .map(|_| rng.random_range(0..=max_code))
3189 .collect::<Vec<_>>();
3190
3191 let code_len = rabit_binary_code_bytes(code_dim);
3192 let mut code_bytes = vec![0u8; num_rows * code_len];
3193 for (row, bits) in sign_bits.chunks_exact(code_dim).enumerate() {
3194 for (dim, &bit) in bits.iter().enumerate() {
3195 code_bytes[row * code_len + dim / 8] |= (bit as u8) << (dim % 8);
3196 }
3197 }
3198 let (ex_code_column, ex_code_len, ex_code_bytes) = if legacy_format {
3199 let ex_code_len = rabit_ex_code_bytes(code_dim, ex_bits).unwrap();
3200 let mut ex_code_bytes = vec![0u8; num_rows * ex_code_len];
3201 for (row, values) in ex_values.chunks_exact(code_dim).enumerate() {
3202 for (dim, &value) in values.iter().enumerate() {
3203 let bit_offset = dim * ex_bits as usize;
3204 let bits = (value as u16) << (bit_offset % 8);
3205 ex_code_bytes[row * ex_code_len + bit_offset / 8] |= bits as u8;
3206 if bits >> 8 != 0 {
3207 ex_code_bytes[row * ex_code_len + bit_offset / 8 + 1] |=
3208 (bits >> 8) as u8;
3209 }
3210 }
3211 }
3212 (RABIT_EX_CODE_COLUMN, ex_code_len, ex_code_bytes)
3213 } else {
3214 let ex_code_len = blocked_ex_code_bytes(code_dim, ex_bits);
3215 let mut ex_code_bytes = vec![0u8; num_rows * ex_code_len];
3216 for (row, values) in ex_code_bytes
3217 .chunks_exact_mut(ex_code_len)
3218 .zip(ex_values.chunks_exact(code_dim))
3219 {
3220 crate::vector::bq::ex_dot::pack_blocked_row(values, ex_bits, row);
3221 }
3222 (RABIT_BLOCKED_EX_CODE_COLUMN, ex_code_len, ex_code_bytes)
3223 };
3224
3225 let identity = Float32Array::from_iter_values((0..code_dim).flat_map(|row| {
3226 (0..code_dim).map(move |col| if row == col { 1.0 } else { 0.0 })
3227 }));
3228 let rotate_mat =
3229 FixedSizeListArray::try_new_from_values(identity, code_dim as i32).unwrap();
3230 let metadata = RabitQuantizationMetadata {
3231 rotate_mat: Some(rotate_mat),
3232 rotate_mat_position: None,
3233 fast_rotation_signs: None,
3234 rotation_type: RQRotationType::Matrix,
3235 code_dim: code_dim as u32,
3236 num_bits,
3237 packed: false,
3238 query_estimator: RabitQueryEstimator::RawQuery,
3239 };
3240 let codes = FixedSizeListArray::try_new_from_values(
3241 UInt8Array::from(code_bytes),
3242 code_len as i32,
3243 )
3244 .unwrap();
3245 let ex_codes = FixedSizeListArray::try_new_from_values(
3246 UInt8Array::from(ex_code_bytes),
3247 ex_code_len as i32,
3248 )
3249 .unwrap();
3250 let ex_add_factors = (0..num_rows)
3251 .map(|_| rng.random_range(-1.0f32..1.0))
3252 .collect::<Vec<_>>();
3253 let ex_scale_factors = (0..num_rows)
3254 .map(|_| rng.random_range(0.1f32..1.0))
3255 .collect::<Vec<_>>();
3256 let batch = RecordBatch::try_from_iter(vec![
3257 (
3258 ROW_ID,
3259 Arc::new(UInt64Array::from_iter_values(0..num_rows as u64)) as ArrayRef,
3260 ),
3261 (RABIT_CODE_COLUMN, Arc::new(codes) as ArrayRef),
3262 (
3263 ADD_FACTORS_COLUMN,
3264 Arc::new(Float32Array::from(vec![0.0; num_rows])) as ArrayRef,
3265 ),
3266 (
3267 SCALE_FACTORS_COLUMN,
3268 Arc::new(Float32Array::from(vec![0.0; num_rows])) as ArrayRef,
3269 ),
3270 (ex_code_column, Arc::new(ex_codes) as ArrayRef),
3271 (
3272 EX_ADD_FACTORS_COLUMN,
3273 Arc::new(Float32Array::from(ex_add_factors.clone())) as ArrayRef,
3274 ),
3275 (
3276 EX_SCALE_FACTORS_COLUMN,
3277 Arc::new(Float32Array::from(ex_scale_factors.clone())) as ArrayRef,
3278 ),
3279 ])
3280 .unwrap();
3281 let storage = RabitQuantizationStorage::try_from_batch(
3282 batch,
3283 &metadata,
3284 DistanceType::L2,
3285 None,
3286 )
3287 .unwrap();
3288
3289 let query = (0..code_dim)
3290 .map(|_| rng.random_range(-1.0f32..1.0))
3291 .collect::<Vec<_>>();
3292 let sum_q = query.iter().sum::<f32>();
3293 let calc = storage.dist_calculator(
3294 Arc::new(Float32Array::from(query.clone())) as ArrayRef,
3295 0.0,
3296 );
3297
3298 let code_scale = (1u32 << ex_bits) as f32;
3299 let code_bias = -(code_scale - 0.5);
3300 let expected = (0..num_rows)
3301 .map(|row| {
3302 let binary_ip = (0..code_dim)
3303 .map(|dim| {
3304 query[dim] * sign_bits[row * code_dim + dim] as u8 as f32
3305 })
3306 .sum::<f32>();
3307 let ex_dist = (0..code_dim)
3308 .map(|dim| query[dim] * ex_values[row * code_dim + dim] as f32)
3309 .sum::<f32>();
3310 let full_dot = code_scale * binary_ip + ex_dist + code_bias * sum_q;
3311 full_dot * ex_scale_factors[row] + ex_add_factors[row]
3312 })
3313 .collect::<Vec<_>>();
3314
3315 for (row, &want) in expected.iter().enumerate() {
3316 let got = calc.distance(row as u32);
3317 assert!(
3318 (got - want).abs() <= 1e-3 * want.abs().max(1.0),
3319 "num_bits={num_bits} row={row}: {got} != {want}"
3320 );
3321 }
3322
3323 let mut distances = Vec::new();
3324 let mut u16_scratch = Vec::new();
3325 let mut u8_scratch = Vec::new();
3326 let mut u32_scratch = Vec::new();
3327 calc.distance_all_with_scratch(
3328 0,
3329 &mut distances,
3330 &mut u16_scratch,
3331 &mut u8_scratch,
3332 &mut u32_scratch,
3333 );
3334 assert_eq!(distances.len(), num_rows);
3335 if !matches!(ex_bits, 2 | 4 | 8) {
3341 let num_tables = code_dim.div_ceil(4);
3345 let mut table_min = f32::INFINITY;
3346 let mut table_max = f32::NEG_INFINITY;
3347 for segment in query.chunks(4) {
3348 for subset in 0..16usize {
3349 let value = segment
3350 .iter()
3351 .enumerate()
3352 .filter(|(idx, _)| subset & (1 << idx) != 0)
3353 .map(|(_, q)| *q)
3354 .sum::<f32>();
3355 table_min = table_min.min(value);
3356 table_max = table_max.max(value);
3357 }
3358 }
3359 let binary_bound =
3360 code_scale * num_tables as f32 * (table_max - table_min) / 255.0 / 2.0
3361 * ex_scale_factors.iter().fold(0.0f32, |max, &s| max.max(s));
3362 for (row, (&got, &want)) in
3363 distances.iter().zip(expected.iter()).enumerate()
3364 {
3365 assert!(
3366 (got - want).abs() <= binary_bound + 1e-3,
3367 "num_bits={num_bits} row={row} (distance_all): {got} != {want} (bound {binary_bound})"
3368 );
3369 }
3370 let remainder_row = num_rows - 1;
3373 let got = distances[remainder_row];
3374 let want = calc.distance(remainder_row as u32);
3375 assert!(
3376 (got - want).abs() <= 1e-3 * want.abs().max(1.0),
3377 "num_bits={num_bits} remainder row (distance_all): {got} != {want}"
3378 );
3379 }
3380 }
3381 }
3382 }
3383 }
3384
3385 #[test]
3386 fn test_fast_approx_mode_uses_one_bit_scores_for_multi_bit_raw_query() {
3387 let code_dim = 8usize;
3388 let identity = Float32Array::from_iter_values(
3389 (0..code_dim)
3390 .flat_map(|row| (0..code_dim).map(move |col| if row == col { 1.0 } else { 0.0 })),
3391 );
3392 let rotate_mat =
3393 FixedSizeListArray::try_new_from_values(identity, code_dim as i32).unwrap();
3394 let metadata = RabitQuantizationMetadata {
3395 rotate_mat: Some(rotate_mat),
3396 rotate_mat_position: None,
3397 fast_rotation_signs: None,
3398 rotation_type: RQRotationType::Matrix,
3399 code_dim: code_dim as u32,
3400 num_bits: 2,
3401 packed: false,
3402 query_estimator: RabitQueryEstimator::RawQuery,
3403 };
3404 let codes =
3405 FixedSizeListArray::try_new_from_values(UInt8Array::from(vec![0xff, 0xff]), 1).unwrap();
3406 let ex_codes =
3407 FixedSizeListArray::try_new_from_values(UInt8Array::from(vec![0x00, 0xff]), 1).unwrap();
3408 let batch = make_test_batch_with_ex(codes, ex_codes)
3409 .replace_column_by_name(
3410 SCALE_FACTORS_COLUMN,
3411 Arc::new(Float32Array::from(vec![0.0, 0.0])),
3412 )
3413 .unwrap();
3414 let storage =
3415 RabitQuantizationStorage::try_from_batch(batch, &metadata, DistanceType::L2, None)
3416 .unwrap();
3417 let query = Arc::new(Float32Array::from(vec![1.0; code_dim])) as ArrayRef;
3418 let normal = storage.dist_calculator(query.clone(), 0.0).distance_all(0);
3419
3420 let mut f32_scratch = Vec::new();
3421 let calc = storage.dist_calculator_with_scratch(
3422 query,
3423 0.0,
3424 None,
3425 &mut f32_scratch,
3426 DistanceCalculatorOptions {
3427 approx_mode: ApproxMode::Fast,
3428 },
3429 );
3430 let mut distances = Vec::new();
3431 let mut u16_scratch = Vec::new();
3432 let mut u8_scratch = Vec::new();
3433 let mut u32_scratch = Vec::new();
3434 calc.distance_all_with_scratch(
3435 0,
3436 &mut distances,
3437 &mut u16_scratch,
3438 &mut u8_scratch,
3439 &mut u32_scratch,
3440 );
3441
3442 let expected_fast = (0..2)
3443 .map(|id| calc.distance(id as u32))
3444 .collect::<Vec<_>>();
3445 assert_ne!(normal, distances);
3446 assert_eq!(distances, expected_fast);
3447 assert_eq!(
3448 calc.raw_query_lower_bound_gating_disabled_reason(),
3449 Some("approx_mode_fast")
3450 );
3451 }
3452
3453 #[test]
3454 fn test_accurate_approx_mode_reduces_binary_lut_quantization_error() {
3455 let code_dim = 64usize;
3456 let num_rows = BATCH_SIZE;
3457 let original_codes = make_test_codes(num_rows, code_dim as i32);
3458 let metadata = make_test_metadata(code_dim);
3459 let storage = RabitQuantizationStorage::try_from_batch(
3460 make_test_batch(original_codes),
3461 &metadata,
3462 DistanceType::L2,
3463 None,
3464 )
3465 .unwrap();
3466 let query = Arc::new(Float32Array::from_iter_values(
3467 (0..code_dim).map(|idx| (idx as f32 * 0.137).sin() + idx as f32 * 0.003),
3468 )) as ArrayRef;
3469 let exact_calc = storage.dist_calculator(query.clone(), 0.0);
3470 let exact = (0..num_rows)
3471 .map(|id| exact_calc.distance(id as u32))
3472 .collect::<Vec<_>>();
3473
3474 let normal = {
3475 let mut f32_scratch = Vec::new();
3476 let calc = storage.dist_calculator_with_scratch(
3477 query.clone(),
3478 0.0,
3479 None,
3480 &mut f32_scratch,
3481 DistanceCalculatorOptions::default(),
3482 );
3483 let mut distances = Vec::new();
3484 let mut u16_scratch = Vec::new();
3485 let mut u8_scratch = Vec::new();
3486 let mut u32_scratch = Vec::new();
3487 calc.distance_all_with_scratch(
3488 0,
3489 &mut distances,
3490 &mut u16_scratch,
3491 &mut u8_scratch,
3492 &mut u32_scratch,
3493 );
3494 distances
3495 };
3496
3497 let (accurate, hacc_table_len, hacc_packed_table_len, hacc_accum_len) = {
3498 let mut f32_scratch = Vec::new();
3499 let calc = storage.dist_calculator_with_scratch(
3500 query,
3501 0.0,
3502 None,
3503 &mut f32_scratch,
3504 DistanceCalculatorOptions {
3505 approx_mode: ApproxMode::Accurate,
3506 },
3507 );
3508 let mut distances = Vec::new();
3509 let mut u16_scratch = Vec::new();
3510 let mut u8_scratch = Vec::new();
3511 let mut u32_scratch = Vec::new();
3512 calc.distance_all_with_scratch(
3513 0,
3514 &mut distances,
3515 &mut u16_scratch,
3516 &mut u8_scratch,
3517 &mut u32_scratch,
3518 );
3519 (
3520 distances,
3521 u16_scratch.len(),
3522 u8_scratch.len(),
3523 u32_scratch.len(),
3524 )
3525 };
3526
3527 let normal_error = normal
3528 .iter()
3529 .zip(exact.iter())
3530 .map(|(actual, expected)| (actual - expected).abs())
3531 .sum::<f32>();
3532 let accurate_error = accurate
3533 .iter()
3534 .zip(exact.iter())
3535 .map(|(actual, expected)| (actual - expected).abs())
3536 .sum::<f32>();
3537
3538 assert!(normal_error > 0.0);
3539 assert!(
3540 accurate_error < normal_error,
3541 "accurate_error={accurate_error}, normal_error={normal_error}"
3542 );
3543 assert_eq!(hacc_table_len, code_dim * 4);
3544 assert_eq!(hacc_packed_table_len, code_dim * 8);
3545 assert_eq!(hacc_accum_len, num_rows);
3546 }
3547
3548 fn assert_raw_query_multi_bit_distance_all_uses_fastscan(
3549 num_bits: u8,
3550 legacy_format: bool,
3551 with_error_factors: bool,
3552 ) {
3553 let code_dim = 72usize;
3556 let num_rows = BATCH_SIZE + 1;
3557 let ex_bits = rabit_ex_bits(num_bits).unwrap();
3558 let max_code = ((1u16 << ex_bits) - 1) as u8;
3559 let identity = Float32Array::from_iter_values(
3560 (0..code_dim)
3561 .flat_map(|row| (0..code_dim).map(move |col| if row == col { 1.0 } else { 0.0 })),
3562 );
3563 let rotate_mat =
3564 FixedSizeListArray::try_new_from_values(identity, code_dim as i32).unwrap();
3565 let metadata = RabitQuantizationMetadata {
3566 rotate_mat: Some(rotate_mat),
3567 rotate_mat_position: None,
3568 fast_rotation_signs: None,
3569 rotation_type: RQRotationType::Matrix,
3570 code_dim: code_dim as u32,
3571 num_bits,
3572 packed: false,
3573 query_estimator: RabitQueryEstimator::RawQuery,
3574 };
3575 let code_len = rabit_binary_code_bytes(code_dim);
3576 let codes = FixedSizeListArray::try_new_from_values(
3577 UInt8Array::from_iter_values((0..num_rows * code_len).map(|idx| (idx * 13) as u8)),
3578 code_len as i32,
3579 )
3580 .unwrap();
3581 let ex_values = (0..num_rows * code_dim)
3582 .map(|idx| ((idx * 37) % (max_code as usize + 1)) as u8)
3583 .collect::<Vec<_>>();
3584 let (ex_code_column, ex_code_len, ex_code_bytes) = if legacy_format {
3585 let ex_code_len = rabit_ex_code_bytes(code_dim, ex_bits).unwrap();
3586 let mut ex_code_bytes = vec![0u8; num_rows * ex_code_len];
3587 for (row, values) in ex_values.chunks_exact(code_dim).enumerate() {
3588 for (dim, &value) in values.iter().enumerate() {
3589 let bit_offset = dim * ex_bits as usize;
3590 let bits = (value as u16) << (bit_offset % 8);
3591 ex_code_bytes[row * ex_code_len + bit_offset / 8] |= bits as u8;
3592 if bits >> 8 != 0 {
3593 ex_code_bytes[row * ex_code_len + bit_offset / 8 + 1] |= (bits >> 8) as u8;
3594 }
3595 }
3596 }
3597 (RABIT_EX_CODE_COLUMN, ex_code_len, ex_code_bytes)
3598 } else {
3599 let ex_code_len = blocked_ex_code_bytes(code_dim, ex_bits);
3600 let mut ex_code_bytes = vec![0u8; num_rows * ex_code_len];
3601 for (row, values) in ex_code_bytes
3602 .chunks_exact_mut(ex_code_len)
3603 .zip(ex_values.chunks_exact(code_dim))
3604 {
3605 crate::vector::bq::ex_dot::pack_blocked_row(values, ex_bits, row);
3606 }
3607 (RABIT_BLOCKED_EX_CODE_COLUMN, ex_code_len, ex_code_bytes)
3608 };
3609 let ex_codes = FixedSizeListArray::try_new_from_values(
3610 UInt8Array::from(ex_code_bytes),
3611 ex_code_len as i32,
3612 )
3613 .unwrap();
3614 let batch = RecordBatch::try_from_iter(vec![
3615 (
3616 ROW_ID,
3617 Arc::new(UInt64Array::from_iter_values(0..num_rows as u64)) as ArrayRef,
3618 ),
3619 (RABIT_CODE_COLUMN, Arc::new(codes) as ArrayRef),
3620 (
3621 ADD_FACTORS_COLUMN,
3622 Arc::new(Float32Array::from(vec![0.0; num_rows])) as ArrayRef,
3623 ),
3624 (
3625 SCALE_FACTORS_COLUMN,
3626 Arc::new(Float32Array::from(vec![1.0; num_rows])) as ArrayRef,
3627 ),
3628 (ex_code_column, Arc::new(ex_codes) as ArrayRef),
3629 (
3630 EX_ADD_FACTORS_COLUMN,
3631 Arc::new(Float32Array::from(vec![0.0; num_rows])) as ArrayRef,
3632 ),
3633 (
3634 EX_SCALE_FACTORS_COLUMN,
3635 Arc::new(Float32Array::from(vec![1.0; num_rows])) as ArrayRef,
3636 ),
3637 ])
3638 .unwrap();
3639 let batch = if with_error_factors {
3640 batch
3641 .try_with_column(
3642 crate::vector::bq::transform::ERROR_FACTORS_FIELD.clone(),
3643 Arc::new(Float32Array::from(vec![1000.0; num_rows])) as ArrayRef,
3644 )
3645 .unwrap()
3646 } else {
3647 batch
3648 };
3649 let storage =
3650 RabitQuantizationStorage::try_from_batch(batch, &metadata, DistanceType::L2, None)
3651 .unwrap();
3652 assert_eq!(storage.packed_ex_codes.is_some(), !with_error_factors);
3656
3657 let query_values = (0..code_dim)
3660 .map(|dim| (dim % 11) as f32 * 0.3 - 1.5)
3661 .collect::<Vec<_>>();
3662 let query = Arc::new(Float32Array::from(query_values.clone())) as ArrayRef;
3663 let calc = storage.dist_calculator(query, 0.0);
3664 let mut distances = Vec::new();
3665 let mut u16_scratch = Vec::new();
3666 let mut u8_scratch = Vec::new();
3667 let mut u32_scratch = Vec::new();
3668 calc.distance_all_with_scratch(
3669 0,
3670 &mut distances,
3671 &mut u16_scratch,
3672 &mut u8_scratch,
3673 &mut u32_scratch,
3674 );
3675
3676 assert_eq!(distances.len(), num_rows);
3677 assert_eq!(u16_scratch.len(), BATCH_SIZE);
3678 let loaded_ex_code_len = storage.ex_codes.as_ref().unwrap().value_length() as usize;
3679 if with_error_factors {
3680 assert_eq!(u8_scratch.len(), code_dim * 4);
3683 } else {
3684 assert_eq!(u8_scratch.len(), loaded_ex_code_len * 2 * SEGMENT_NUM_CODES);
3685 }
3686
3687 let mut table_min = f32::INFINITY;
3691 let mut table_max = f32::NEG_INFINITY;
3692 for segment in query_values.chunks(4) {
3693 for subset in 0..SEGMENT_NUM_CODES {
3694 let value = segment
3695 .iter()
3696 .enumerate()
3697 .filter(|(idx, _)| subset & (1 << idx) != 0)
3698 .map(|(_, q)| *q)
3699 .sum::<f32>();
3700 table_min = table_min.min(value);
3701 table_max = table_max.max(value);
3702 }
3703 }
3704 let code_scale = (1u32 << ex_bits) as f32;
3705 let binary_bound =
3706 code_scale * code_dim.div_ceil(4) as f32 * (table_max - table_min) / 510.0;
3707 let mut padded_query = vec![0.0f32; crate::vector::bq::ex_dot::padded_query_len(code_dim)];
3708 crate::vector::bq::ex_dot::pad_query_into(&query_values, &mut padded_query);
3709 let mut quantized_table = Vec::new();
3710 let (ex_qmin, ex_qmax, ex_qcap) = quantize_ex_fastscan_dist_table_into(
3711 ex_bits,
3712 loaded_ex_code_len,
3713 &padded_query,
3714 &mut quantized_table,
3715 );
3716 let ex_bound = if with_error_factors {
3719 0.0
3720 } else {
3721 (loaded_ex_code_len * 2) as f32 * (ex_qmax - ex_qmin) / ex_qcap / 2.0
3722 };
3723 let bound = (binary_bound + ex_bound) * 1.5 + 1e-3;
3724 for (id, distance) in distances.iter().take(BATCH_SIZE).enumerate() {
3725 let exact = calc.distance(id as u32);
3726 assert!(
3727 (*distance - exact).abs() <= bound,
3728 "distance_all fastscan mismatch for id {id} (num_bits={num_bits} legacy={legacy_format}): actual={distance}, exact={exact}, bound={bound}"
3729 );
3730 }
3731 assert_eq!(distances[BATCH_SIZE], calc.distance(BATCH_SIZE as u32));
3732 }
3733
3734 #[test]
3735 fn test_raw_query_multi_bit_distance_all_uses_fastscan_for_split_ex_codes() {
3736 for num_bits in [3, 5, 9] {
3737 for legacy_format in [false, true] {
3738 assert_raw_query_multi_bit_distance_all_uses_fastscan(
3739 num_bits,
3740 legacy_format,
3741 false,
3742 );
3743 }
3744 assert_raw_query_multi_bit_distance_all_uses_fastscan(num_bits, false, true);
3747 }
3748 }
3749
3750 #[rstest]
3756 fn test_degenerate_dist_table_falls_back_to_exact_distances(
3757 #[values(ApproxMode::Normal, ApproxMode::Accurate)] approx_mode: ApproxMode,
3758 ) {
3759 let code_dim = 8usize;
3760 let num_rows = BATCH_SIZE + 5;
3761 let num_bits = 3;
3762 let ex_bits = rabit_ex_bits(num_bits).unwrap();
3763 let identity = Float32Array::from_iter_values(
3764 (0..code_dim)
3765 .flat_map(|row| (0..code_dim).map(move |col| if row == col { 1.0 } else { 0.0 })),
3766 );
3767 let rotate_mat =
3768 FixedSizeListArray::try_new_from_values(identity, code_dim as i32).unwrap();
3769 let metadata = RabitQuantizationMetadata {
3770 rotate_mat: Some(rotate_mat),
3771 rotate_mat_position: None,
3772 fast_rotation_signs: None,
3773 rotation_type: RQRotationType::Matrix,
3774 code_dim: code_dim as u32,
3775 num_bits,
3776 packed: false,
3777 query_estimator: RabitQueryEstimator::RawQuery,
3778 };
3779 let codes = FixedSizeListArray::try_new_from_values(
3780 UInt8Array::from_iter_values((0..num_rows).map(|idx| (idx * 19) as u8)),
3781 rabit_binary_code_bytes(code_dim) as i32,
3782 )
3783 .unwrap();
3784 let ex_codes = make_test_ex_codes(num_rows, code_dim, num_bits);
3785 let batch = make_test_batch_with_ex(codes, ex_codes);
3786 let storage =
3787 RabitQuantizationStorage::try_from_batch(batch, &metadata, DistanceType::L2, None)
3788 .unwrap();
3789 let query = Arc::new(Float32Array::from(vec![1.0; code_dim])) as ArrayRef;
3790
3791 let mut calc = storage.dist_calculator(query, 4.0);
3792 calc.approx_mode = approx_mode;
3793 let mut degenerate = vec![0.0f32; code_dim * 4];
3798 degenerate[0] = -2e38;
3799 degenerate[1] = 2e38;
3800 calc.dist_table = Cow::Owned(degenerate);
3801
3802 let code_len = rabit_binary_code_bytes(code_dim);
3803 let ex_codes = calc.ex_codes.unwrap();
3804 let ex_add_factors = calc.ex_add_factors.unwrap();
3805 let ex_scale_factors = calc.ex_scale_factors.unwrap();
3806 let expected = (0..num_rows)
3807 .map(|id| {
3808 let binary_ip = compute_single_rq_distance(
3809 calc.codes,
3810 id,
3811 num_rows,
3812 code_len,
3813 &calc.dist_table,
3814 );
3815 calc.raw_query_multi_bit_exact_distance(
3816 id,
3817 binary_ip,
3818 ex_bits,
3819 ex_codes,
3820 ex_add_factors,
3821 ex_scale_factors,
3822 )
3823 })
3824 .collect::<Vec<_>>();
3825
3826 let actual = calc.distance_all(0);
3827 assert_eq!(actual.len(), num_rows);
3828 for id in 0..num_rows {
3829 assert!(
3830 !actual[id].is_nan(),
3831 "approx_mode={approx_mode:?} id={id}: degenerate table produced NaN"
3832 );
3833 assert_eq!(
3834 actual[id].to_bits(),
3835 expected[id].to_bits(),
3836 "approx_mode={approx_mode:?} id={id}: distance_all must match the exact path"
3837 );
3838 }
3839 }
3840
3841 #[test]
3842 fn test_raw_query_multi_bit_accumulate_topk_uses_lower_bound_gating() {
3843 let code_dim = 8usize;
3844 let num_rows = BATCH_SIZE + 9;
3845 let num_bits = 3;
3846 let ex_bits = rabit_ex_bits(num_bits).unwrap();
3847 let identity = Float32Array::from_iter_values(
3848 (0..code_dim)
3849 .flat_map(|row| (0..code_dim).map(move |col| if row == col { 1.0 } else { 0.0 })),
3850 );
3851 let rotate_mat =
3852 FixedSizeListArray::try_new_from_values(identity, code_dim as i32).unwrap();
3853 let metadata = RabitQuantizationMetadata {
3854 rotate_mat: Some(rotate_mat),
3855 rotate_mat_position: None,
3856 fast_rotation_signs: None,
3857 rotation_type: RQRotationType::Matrix,
3858 code_dim: code_dim as u32,
3859 num_bits,
3860 packed: false,
3861 query_estimator: RabitQueryEstimator::RawQuery,
3862 };
3863 let codes = FixedSizeListArray::try_new_from_values(
3864 UInt8Array::from_iter_values((0..num_rows).map(|idx| (idx * 19) as u8)),
3865 1,
3866 )
3867 .unwrap();
3868 let ex_code_len = rabit_ex_code_bytes(code_dim, ex_bits).unwrap();
3869 let ex_codes = FixedSizeListArray::try_new_from_values(
3870 UInt8Array::from_iter_values(
3871 (0..num_rows * ex_code_len).map(|idx| (idx * 29 % 251) as u8),
3872 ),
3873 ex_code_len as i32,
3874 )
3875 .unwrap();
3876 let batch = make_test_batch_with_ex(codes, ex_codes)
3877 .replace_column_by_name(
3878 ERROR_FACTORS_COLUMN,
3879 Arc::new(Float32Array::from(vec![1000.0; num_rows])),
3880 )
3881 .unwrap();
3882 let storage =
3883 RabitQuantizationStorage::try_from_batch(batch, &metadata, DistanceType::L2, None)
3884 .unwrap();
3885 let query = Arc::new(Float32Array::from(vec![1.0; code_dim])) as ArrayRef;
3886 let calc = storage.dist_calculator(query, 4.0);
3887 assert!(
3888 calc.raw_query_lower_bound_gating_disabled_reason()
3889 .is_none()
3890 );
3891
3892 let k = 5;
3893 let mut binary_ips = Vec::new();
3894 let mut binary_u16_scratch = Vec::new();
3895 let mut binary_u8_scratch = Vec::new();
3896 let mut binary_u32_scratch = Vec::new();
3897 calc.binary_distances_with_scratch(
3898 num_rows,
3899 rabit_binary_code_bytes(code_dim),
3900 &mut binary_ips,
3901 &mut binary_u16_scratch,
3902 &mut binary_u8_scratch,
3903 &mut binary_u32_scratch,
3904 );
3905 let ex_codes = calc.ex_codes.unwrap();
3906 let ex_add_factors = calc.ex_add_factors.unwrap();
3907 let ex_scale_factors = calc.ex_scale_factors.unwrap();
3908 let mut expected = binary_ips
3909 .iter()
3910 .copied()
3911 .enumerate()
3912 .map(|(id, binary_ip)| {
3913 (
3914 id,
3915 calc.raw_query_multi_bit_exact_distance(
3916 id,
3917 binary_ip,
3918 ex_bits,
3919 ex_codes,
3920 ex_add_factors,
3921 ex_scale_factors,
3922 ),
3923 )
3924 })
3925 .collect::<Vec<_>>();
3926 expected.sort_by(|left, right| left.1.total_cmp(&right.1));
3927 expected.truncate(k);
3928 let mut expected = expected
3929 .into_iter()
3930 .map(|(id, dist)| (id as u64, dist))
3931 .collect::<Vec<_>>();
3932 expected.sort_by(|left, right| left.0.cmp(&right.0));
3933
3934 let mut heap = BinaryHeap::with_capacity(k);
3935 let mut distances = Vec::new();
3936 let mut u16_scratch = Vec::new();
3937 let mut u8_scratch = Vec::new();
3938 let mut u32_scratch = Vec::new();
3939 calc.accumulate_topk_with_scratch(
3940 k,
3941 None,
3942 None,
3943 |id| id as u64,
3944 &mut heap,
3945 &mut distances,
3946 &mut u16_scratch,
3947 &mut u8_scratch,
3948 &mut u32_scratch,
3949 );
3950 let mut actual = heap
3951 .into_iter()
3952 .map(|node| (node.id, node.dist.0))
3953 .collect::<Vec<_>>();
3954 actual.sort_by(|left, right| left.0.cmp(&right.0));
3955
3956 assert_eq!(actual.len(), expected.len());
3957 for ((actual_id, actual_dist), (expected_id, expected_dist)) in
3958 actual.into_iter().zip(expected)
3959 {
3960 assert_eq!(actual_id, expected_id);
3961 assert!(
3962 (actual_dist - expected_dist).abs() < 1e-5,
3963 "actual={actual_dist}, expected={expected_dist}"
3964 );
3965 }
3966 }
3967
3968 struct CraftedTopkData {
3975 codes: Vec<u8>,
3976 ex_codes: Vec<u8>,
3977 dist_table: Vec<f32>,
3978 ex_query: Vec<f32>,
3979 scale_factors: Vec<f32>,
3980 add_factors: Vec<f32>,
3981 error_factors: Vec<f32>,
3982 ex_scale_factors: Vec<f32>,
3983 ex_add_factors: Vec<f32>,
3984 }
3985
3986 const CRAFTED_TOPK_DIM: usize = 64;
3987 const CRAFTED_TOPK_NUM_BITS: u8 = 5;
3988
3989 impl CraftedTopkData {
3990 fn new(
3991 exact_dists: &[f32],
3992 lower_bound_margins: &[f32],
3993 error_factors: Vec<f32>,
3994 rng: &mut SmallRng,
3995 ) -> Self {
3996 let n = exact_dists.len();
3997 let code_len = rabit_binary_code_bytes(CRAFTED_TOPK_DIM);
3998 let ex_code_len = blocked_ex_code_bytes(CRAFTED_TOPK_DIM, CRAFTED_TOPK_NUM_BITS - 1);
3999 let add_factors = izip!(exact_dists, lower_bound_margins, &error_factors)
4000 .map(|(dist, margin, error)| dist - margin + error)
4001 .collect();
4002 Self {
4003 codes: (0..n * code_len).map(|_| rng.random()).collect(),
4004 ex_codes: (0..n * ex_code_len).map(|_| rng.random()).collect(),
4005 dist_table: (0..CRAFTED_TOPK_DIM * 4)
4006 .map(|_| rng.random_range(-1.0f32..1.0))
4007 .collect(),
4008 ex_query: (0..CRAFTED_TOPK_DIM)
4009 .map(|_| rng.random_range(-1.0f32..1.0))
4010 .collect(),
4011 scale_factors: vec![0.0; n],
4012 add_factors,
4013 error_factors,
4014 ex_scale_factors: vec![0.0; n],
4015 ex_add_factors: exact_dists.to_vec(),
4016 }
4017 }
4018
4019 fn calculator(&self, approx_mode: ApproxMode) -> RabitDistCalculator<'_> {
4020 RabitDistCalculator::new(
4021 CRAFTED_TOPK_DIM,
4022 CRAFTED_TOPK_NUM_BITS,
4023 RabitQueryEstimator::RawQuery,
4024 Cow::Borrowed(self.dist_table.as_slice()),
4025 Cow::Borrowed(self.ex_query.as_slice()),
4026 0.7,
4027 &self.codes,
4028 Some(&self.ex_codes),
4029 blocked_ex_code_bytes(CRAFTED_TOPK_DIM, CRAFTED_TOPK_NUM_BITS - 1),
4030 &self.add_factors,
4031 &self.scale_factors,
4032 Some(&self.error_factors),
4033 Some(&self.ex_add_factors),
4034 Some(&self.ex_scale_factors),
4035 None,
4036 0.0,
4037 1.0,
4038 approx_mode,
4039 )
4040 }
4041 }
4042
4043 fn canonical_heap_rows(heap: BinaryHeap<OrderedNode<u64>>) -> Vec<(u32, u64)> {
4044 let mut rows = heap
4045 .into_iter()
4046 .map(|node| (node.dist.0.to_bits(), node.id))
4047 .collect::<Vec<_>>();
4048 rows.sort_unstable();
4049 rows
4050 }
4051
4052 #[rstest]
4056 fn test_raw_query_multi_bit_topk_dense_matches_sparse(
4057 #[values(ApproxMode::Normal, ApproxMode::Accurate)] approx_mode: ApproxMode,
4058 #[values("descending", "ascending", "random", "duplicates", "duplicate_ties")]
4059 ordering: &str,
4060 ) {
4061 for n in [1usize, 15, 16, 17, 100, 4109] {
4062 let mut rng = SmallRng::seed_from_u64(n as u64 * 31 + ordering.len() as u64);
4063 let exact_dists: Vec<f32> = match ordering {
4064 "descending" => (0..n).map(|id| (n - id) as f32).collect(),
4066 "ascending" => (0..n).map(|id| id as f32).collect(),
4068 "random" => (0..n).map(|_| rng.random_range(0.0..n as f32)).collect(),
4069 "duplicates" => (0..n).map(|id| (id % 7) as f32).collect(),
4070 "duplicate_ties" => (0..n).map(|id| (id % 5) as f32).collect(),
4073 _ => unreachable!(),
4074 };
4075 let (margins, error_factors) = if ordering == "duplicate_ties" {
4076 (vec![0.0; n], vec![0.0; n])
4077 } else if ordering == "random" {
4078 (
4079 (0..n).map(|_| rng.random_range(0.0f32..2.0)).collect(),
4080 (0..n).map(|_| rng.random_range(0.0f32..1.0)).collect(),
4081 )
4082 } else {
4083 (
4084 vec![1.0; n],
4085 (0..n).map(|_| rng.random_range(0.0f32..1.0)).collect(),
4086 )
4087 };
4088 let data = CraftedTopkData::new(&exact_dists, &margins, error_factors, &mut rng);
4089 let calc = data.calculator(approx_mode);
4090 assert!(
4091 calc.raw_query_lower_bound_gating_disabled_reason()
4092 .is_none()
4093 );
4094
4095 let max_dist = exact_dists.iter().fold(0.0f32, |acc, dist| acc.max(*dist));
4096 for k in [1usize, 10, n + 7] {
4097 for bounds in [(None, None), (Some(max_dist * 0.25), Some(max_dist * 0.7))] {
4098 let (lower_bound, upper_bound) = bounds;
4099 let mut dense_heap = BinaryHeap::new();
4100 let mut sparse_heap = BinaryHeap::new();
4101 let mut dists = Vec::new();
4102 let mut u16_scratch = Vec::new();
4103 let mut u8_scratch = Vec::new();
4104 let mut u32_scratch = Vec::new();
4105 for pass in 0..2u64 {
4108 let offset = pass * n as u64;
4109 calc.accumulate_topk_with_scratch(
4110 k,
4111 lower_bound,
4112 upper_bound,
4113 |id| id as u64 + offset,
4114 &mut dense_heap,
4115 &mut dists,
4116 &mut u16_scratch,
4117 &mut u8_scratch,
4118 &mut u32_scratch,
4119 );
4120 calc.accumulate_filtered_topk_with_scratch(
4121 k,
4122 lower_bound,
4123 upper_bound,
4124 (0..n as u32).map(|id| (id, id as u64 + offset)),
4125 |_| true,
4126 &mut sparse_heap,
4127 &mut dists,
4128 &mut u16_scratch,
4129 &mut u8_scratch,
4130 &mut u32_scratch,
4131 );
4132 }
4133 let dense = canonical_heap_rows(dense_heap);
4134 let sparse = canonical_heap_rows(sparse_heap);
4135 assert_eq!(
4136 dense, sparse,
4137 "ordering={ordering} n={n} k={k} bounds={bounds:?} mode={approx_mode:?}"
4138 );
4139
4140 let query_lower_bound = lower_bound.unwrap_or(f32::MIN);
4144 let query_upper_bound = upper_bound.unwrap_or(f32::MAX);
4145 let mut expected = (0..2 * n)
4146 .map(|row| exact_dists[row % n])
4147 .filter(|dist| *dist >= query_lower_bound && *dist < query_upper_bound)
4148 .map(|dist| dist.to_bits())
4149 .collect::<Vec<_>>();
4150 expected.sort_unstable();
4151 expected.truncate(k);
4152 let actual = dense.iter().map(|(dist, _)| *dist).collect::<Vec<_>>();
4153 assert_eq!(
4154 actual, expected,
4155 "ordering={ordering} n={n} k={k} bounds={bounds:?} mode={approx_mode:?}"
4156 );
4157 }
4158 }
4159 }
4160 }
4161
4162 #[test]
4163 fn test_raw_query_one_bit_distance_uses_binary_factors_without_ex_columns() {
4164 let code_dim = 8usize;
4165 let identity = Float32Array::from_iter_values(
4166 (0..code_dim)
4167 .flat_map(|row| (0..code_dim).map(move |col| if row == col { 1.0 } else { 0.0 })),
4168 );
4169 let rotate_mat =
4170 FixedSizeListArray::try_new_from_values(identity, code_dim as i32).unwrap();
4171 let metadata = RabitQuantizationMetadata {
4172 rotate_mat: Some(rotate_mat),
4173 rotate_mat_position: None,
4174 fast_rotation_signs: None,
4175 rotation_type: RQRotationType::Matrix,
4176 code_dim: code_dim as u32,
4177 num_bits: 1,
4178 packed: false,
4179 query_estimator: RabitQueryEstimator::RawQuery,
4180 };
4181 let codes =
4182 FixedSizeListArray::try_new_from_values(UInt8Array::from(vec![0xff, 0x00]), 1).unwrap();
4183 let storage = RabitQuantizationStorage::try_from_batch(
4184 make_test_batch(codes),
4185 &metadata,
4186 DistanceType::L2,
4187 None,
4188 )
4189 .unwrap();
4190 let query = Arc::new(Float32Array::from(vec![1.0; code_dim])) as ArrayRef;
4191 let calc = storage.dist_calculator(query, 3.0);
4192
4193 assert_eq!(calc.distance_all(0), vec![5.0, -2.0]);
4194 }
4195
4196 #[test]
4197 fn test_raw_query_context_matches_fallback_and_only_updates_partition_factor() {
4198 let code_dim = 8usize;
4199 let identity = Float32Array::from_iter_values(
4200 (0..code_dim)
4201 .flat_map(|row| (0..code_dim).map(move |col| if row == col { 1.0 } else { 0.0 })),
4202 );
4203 let rotate_mat =
4204 FixedSizeListArray::try_new_from_values(identity, code_dim as i32).unwrap();
4205 let metadata = RabitQuantizationMetadata {
4206 rotate_mat: Some(rotate_mat),
4207 rotate_mat_position: None,
4208 fast_rotation_signs: None,
4209 rotation_type: RQRotationType::Matrix,
4210 code_dim: code_dim as u32,
4211 num_bits: 2,
4212 packed: false,
4213 query_estimator: RabitQueryEstimator::RawQuery,
4214 };
4215 let codes =
4216 FixedSizeListArray::try_new_from_values(UInt8Array::from(vec![0xff, 0xff]), 1).unwrap();
4217 let ex_codes =
4218 FixedSizeListArray::try_new_from_values(UInt8Array::from(vec![0x00, 0xff]), 1).unwrap();
4219 let storage = RabitQuantizationStorage::try_from_batch(
4220 make_test_batch_with_ex(codes, ex_codes),
4221 &metadata,
4222 DistanceType::Dot,
4223 None,
4224 )
4225 .unwrap();
4226 let query = Arc::new(Float32Array::from(vec![1.0; code_dim])) as ArrayRef;
4227 let rotated_centroid = vec![0.25; code_dim];
4228 let raw_query = metadata.prepare_raw_query_context(query.as_ref()).unwrap();
4229
4230 let mut fallback_scratch = Vec::new();
4231 let expected = storage
4232 .dist_calculator_with_scratch(
4233 query.clone(),
4234 123.0,
4235 Some(QueryResidual::RabitRawQuery {
4236 rotated_centroid: Some(&rotated_centroid),
4237 query: None,
4238 }),
4239 &mut fallback_scratch,
4240 DistanceCalculatorOptions::default(),
4241 )
4242 .distance_all(0);
4243
4244 let mut prepared_scratch = Vec::new();
4245 let actual = storage
4246 .dist_calculator_with_scratch(
4247 query,
4248 456.0,
4249 Some(QueryResidual::RabitRawQuery {
4250 rotated_centroid: Some(&rotated_centroid),
4251 query: Some(&raw_query),
4252 }),
4253 &mut prepared_scratch,
4254 DistanceCalculatorOptions::default(),
4255 )
4256 .distance_all(0);
4257
4258 assert_eq!(actual, expected);
4259 assert!(prepared_scratch.is_empty());
4260 }
4261
4262 #[test]
4263 fn test_try_from_batch_canonicalizes_rq_codes_to_packed_layout() {
4264 let original_codes = make_test_codes(50, 64);
4265 let metadata = make_test_metadata(original_codes.value_length() as usize * 8);
4266 assert!(!metadata.packed);
4267
4268 let storage = RabitQuantizationStorage::try_from_batch(
4269 make_test_batch(original_codes.clone()),
4270 &metadata,
4271 DistanceType::L2,
4272 None,
4273 )
4274 .unwrap();
4275
4276 assert!(storage.metadata().packed);
4277 let stored_batch = storage.to_batches().unwrap().next().unwrap();
4278 let stored_codes = stored_batch[RABIT_CODE_COLUMN].as_fixed_size_list();
4279 let expected_codes = pack_codes(&original_codes);
4280 assert_codes_eq(stored_codes, &expected_codes);
4281 }
4282
4283 #[test]
4284 fn test_try_from_batch_uses_l2_for_cosine() {
4285 let original_codes = make_test_codes(50, 64);
4286 let metadata = make_test_metadata(original_codes.value_length() as usize * 8);
4287
4288 let storage = RabitQuantizationStorage::try_from_batch(
4289 make_test_batch(original_codes),
4290 &metadata,
4291 DistanceType::Cosine,
4292 None,
4293 )
4294 .unwrap();
4295
4296 assert_eq!(storage.distance_type(), DistanceType::L2);
4297 }
4298
4299 #[test]
4300 fn test_try_from_batch_keeps_cosine_for_legacy_residual_query() {
4301 let original_codes = make_test_codes(50, 64);
4302 let mut metadata = make_test_metadata(original_codes.value_length() as usize * 8);
4303 metadata.query_estimator = RabitQueryEstimator::ResidualQuery;
4304
4305 let storage = RabitQuantizationStorage::try_from_batch(
4306 make_test_batch(original_codes),
4307 &metadata,
4308 DistanceType::Cosine,
4309 None,
4310 )
4311 .unwrap();
4312
4313 assert_eq!(storage.distance_type(), DistanceType::Cosine);
4314 }
4315
4316 #[test]
4317 fn test_try_from_batch_requires_ex_columns_for_multi_bit_rq() {
4318 let original_codes = make_test_codes(50, 64);
4319 let mut metadata = make_test_metadata(original_codes.value_length() as usize * 8);
4320 metadata.num_bits = 2;
4321
4322 let err = RabitQuantizationStorage::try_from_batch(
4323 make_test_batch(original_codes),
4324 &metadata,
4325 DistanceType::L2,
4326 None,
4327 )
4328 .unwrap_err();
4329 assert!(
4330 err.to_string()
4331 .contains("requires __blocked_ex_codes column"),
4332 "{}",
4333 err
4334 );
4335 }
4336
4337 #[test]
4338 fn test_try_from_batch_requires_ex_add_factors_for_multi_bit_rq() {
4339 let original_codes = make_test_codes(50, 64);
4340 let code_dim = original_codes.value_length() as usize * 8;
4341 let ex_codes = make_test_ex_codes(original_codes.len(), code_dim, 9);
4342 let mut metadata = make_test_metadata(code_dim);
4343 metadata.num_bits = 9;
4344 let batch = make_test_batch_with_ex(original_codes, ex_codes)
4345 .drop_column(EX_ADD_FACTORS_COLUMN)
4346 .unwrap();
4347
4348 let err =
4349 RabitQuantizationStorage::try_from_batch(batch, &metadata, DistanceType::L2, None)
4350 .unwrap_err();
4351 assert!(
4352 err.to_string().contains("requires __add_factors_ex column"),
4353 "{}",
4354 err
4355 );
4356 }
4357
4358 #[test]
4359 fn test_try_from_batch_accepts_multi_bit_rq_split_codes() {
4360 let original_codes = make_test_codes(50, 64);
4361 let code_dim = original_codes.value_length() as usize * 8;
4362 let ex_codes = make_test_ex_codes(original_codes.len(), code_dim, 9);
4363 let mut metadata = make_test_metadata(code_dim);
4364 metadata.num_bits = 9;
4365
4366 let storage = RabitQuantizationStorage::try_from_batch(
4367 make_test_batch_with_ex(original_codes, ex_codes),
4368 &metadata,
4369 DistanceType::L2,
4370 None,
4371 )
4372 .unwrap();
4373
4374 assert!(storage.metadata().packed);
4375 let stored_batch = storage.to_batches().unwrap().next().unwrap();
4377 assert!(stored_batch.column_by_name(RABIT_EX_CODE_COLUMN).is_none());
4378 assert_eq!(
4379 stored_batch[RABIT_BLOCKED_EX_CODE_COLUMN]
4380 .as_fixed_size_list()
4381 .value_length(),
4382 64
4383 );
4384 assert!(stored_batch.column_by_name(ERROR_FACTORS_COLUMN).is_some());
4385 }
4386
4387 #[test]
4388 fn test_try_from_batch_accepts_missing_error_factors_for_compatibility() {
4389 let original_codes = make_test_codes(50, 64);
4390 let code_dim = original_codes.value_length() as usize * 8;
4391 let ex_codes = make_test_ex_codes(original_codes.len(), code_dim, 9);
4392 let mut metadata = make_test_metadata(code_dim);
4393 metadata.num_bits = 9;
4394 let batch = make_test_batch_with_ex(original_codes, ex_codes)
4395 .drop_column(ERROR_FACTORS_COLUMN)
4396 .unwrap();
4397
4398 let storage =
4399 RabitQuantizationStorage::try_from_batch(batch, &metadata, DistanceType::L2, None)
4400 .unwrap();
4401 let query = Arc::new(Float32Array::from(vec![1.0; code_dim])) as ArrayRef;
4402 let calc = storage.dist_calculator(query, 4.0);
4403
4404 assert!(storage.error_factors.is_none());
4405 assert_eq!(
4406 calc.raw_query_lower_bound_gating_disabled_reason(),
4407 Some("missing_error_factors")
4408 );
4409 }
4410
4411 #[test]
4412 fn test_remap_preserves_packed_rq_storage_layout() {
4413 let original_codes = make_test_codes(50, 64);
4414 let metadata = make_test_metadata(original_codes.value_length() as usize * 8);
4415 let storage = RabitQuantizationStorage::try_from_batch(
4416 make_test_batch(original_codes.clone()),
4417 &metadata,
4418 DistanceType::L2,
4419 None,
4420 )
4421 .unwrap();
4422
4423 let mut mapping = HashMap::new();
4424 mapping.insert(1, Some(101));
4425 mapping.insert(3, None);
4426 mapping.insert(4, Some(104));
4427
4428 let remapped = storage.remap(&mapping).unwrap();
4429 assert!(remapped.metadata().packed);
4430
4431 let remapped_batch = remapped.to_batches().unwrap().next().unwrap();
4432 let remapped_row_ids = remapped_batch[ROW_ID].as_primitive::<UInt64Type>().values();
4433 let expected_row_ids = UInt64Array::from_iter_values(
4434 [0, 101, 2, 104]
4435 .into_iter()
4436 .chain(5..original_codes.len() as u64),
4437 );
4438 assert_eq!(remapped_row_ids, expected_row_ids.values());
4439
4440 let remapped_codes = remapped_batch[RABIT_CODE_COLUMN].as_fixed_size_list();
4441 let repacked = pack_codes(&unpack_codes(remapped_codes));
4442 assert_codes_eq(remapped_codes, &repacked);
4443 }
4444
4445 #[test]
4446 fn test_remap_preserves_multi_bit_rq_split_columns() {
4447 for num_bits in [4, 6, 8, 9u8] {
4450 test_remap_preserves_multi_bit_rq_split_columns_impl(num_bits);
4451 }
4452 }
4453
4454 fn test_remap_preserves_multi_bit_rq_split_columns_impl(num_bits: u8) {
4455 let original_codes = make_test_codes(50, 64);
4456 let code_dim = original_codes.value_length() as usize * 8;
4457 let ex_codes = make_test_ex_codes(original_codes.len(), code_dim, num_bits);
4458 let mut metadata = make_test_metadata(code_dim);
4459 metadata.num_bits = num_bits;
4460 let storage = RabitQuantizationStorage::try_from_batch(
4461 make_test_batch_with_ex(original_codes.clone(), ex_codes),
4462 &metadata,
4463 DistanceType::L2,
4464 None,
4465 )
4466 .unwrap();
4467
4468 let mut mapping = HashMap::new();
4469 mapping.insert(1, Some(101));
4470 mapping.insert(3, None);
4471 mapping.insert(4, Some(104));
4472
4473 let remapped = storage.remap(&mapping).unwrap();
4474 let remapped_batch = remapped.to_batches().unwrap().next().unwrap();
4475 let remapped_row_ids = remapped_batch[ROW_ID].as_primitive::<UInt64Type>().values();
4476 let expected_row_ids = UInt64Array::from_iter_values(
4477 [0, 101, 2, 104]
4478 .into_iter()
4479 .chain(5..original_codes.len() as u64),
4480 );
4481 assert_eq!(remapped_row_ids, expected_row_ids.values());
4482
4483 let ex_code_len = blocked_ex_code_bytes(code_dim, rabit_ex_bits(num_bits).unwrap());
4486 assert_eq!(
4487 remapped_batch[RABIT_BLOCKED_EX_CODE_COLUMN]
4488 .as_fixed_size_list()
4489 .value_length(),
4490 ex_code_len as i32
4491 );
4492 assert_eq!(
4493 &remapped_batch[EX_ADD_FACTORS_COLUMN]
4494 .as_primitive::<Float32Type>()
4495 .values()[..5],
4496 &[10.5, 11.5, 12.5, 14.5, 15.5]
4497 );
4498 assert_eq!(
4499 &remapped_batch[EX_SCALE_FACTORS_COLUMN]
4500 .as_primitive::<Float32Type>()
4501 .values()[..5],
4502 &[1.5, 2.5, 3.5, 5.5, 6.5]
4503 );
4504 assert_eq!(
4505 &remapped_batch[ERROR_FACTORS_COLUMN]
4506 .as_primitive::<Float32Type>()
4507 .values()[..5],
4508 &[0.25, 1.25, 2.25, 4.25, 5.25]
4509 );
4510
4511 let reloaded = RabitQuantizationStorage::try_from_batch(
4514 remapped_batch,
4515 &remapped.metadata,
4516 DistanceType::L2,
4517 None,
4518 )
4519 .unwrap();
4520 assert_eq!(remapped.ex_codes, reloaded.ex_codes);
4521 assert_eq!(
4522 remapped.ex_codes.as_ref().unwrap().value_length() as usize,
4523 blocked_ex_code_bytes(code_dim, rabit_ex_bits(num_bits).unwrap())
4524 );
4525 }
4526}