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