Skip to main content

diskann_quantization/product/tables/transposed/
table.rs

1/*
2 * Copyright (c) Microsoft Corporation.
3 * Licensed under the MIT license.
4 */
5
6use std::num::NonZeroUsize;
7
8use super::pivots;
9use crate::{
10    product::tables::basic::TableCompressionError,
11    traits::CompressInto,
12    views::{ChunkOffsets, ChunkOffsetsView},
13};
14use diskann_utils::{
15    strided::Strided,
16    views::{
17        self,
18        rowmajor::{self, Matrix, MatrixMut},
19    },
20};
21use thiserror::Error;
22
23/// A PQ table that stores the pivots for each chunk in a miniture block-transpose to
24/// facilitate faster compression. The exact layout is not documented (as it is for the
25/// `BasicTable`) because it may be subject to change.
26///
27/// The advantage of this table over the `BasicTable` is that this table uses a more
28/// hardware friendly layout for the pivots, meaning that compression (particularly batch
29/// compression) is much faster than the basic table, at the cost of a slightly higher
30/// memory footprint.
31///
32/// # Invariants (Dev Docs)
33///
34/// * `pivots.len() == schema.len()`: The number of PQ chunks must agree.
35/// * `pivots[i].dimension() == schema.at(i).len()` for all inbounds `i`.
36/// * `pivots[i].total() == ncenters` for all inbounds `i`.
37/// * `largest = max(schema.at(i).len() for i)`.
38#[derive(Debug)]
39pub struct TransposedTable {
40    pivots: Box<[pivots::Chunk]>,
41    offsets: crate::views::ChunkOffsets,
42    /// The largest dimension in offsets.
43    largest: usize,
44    /// The number of centers in each block.
45    ncenters: usize,
46}
47
48#[derive(Debug, Error)]
49#[non_exhaustive]
50pub enum TransposedTableError {
51    #[error("pivots have {pivot_dim} dimensions while the offsets expect {offsets_dim}")]
52    DimMismatch {
53        pivot_dim: usize,
54        offsets_dim: usize,
55    },
56    #[error("error constructing pivot {problem} of {total}")]
57    PivotError {
58        problem: usize,
59        total: usize,
60        source: pivots::ChunkConstructionError,
61    },
62}
63
64impl TransposedTable {
65    /// Construct a new `TransposedTable` from raw parts.
66    ///
67    /// # Error
68    ///
69    /// Returns an error if
70    ///
71    /// * `pivots.ncols() != offsets.dim()`: Pivots must have the dimensionality expected
72    ///   by the offsets.
73    ///
74    /// * `pivots.nrows() == 0`: The pivot table cannot be empty.
75    #[expect(clippy::expect_used)]
76    pub fn from_parts(
77        pivots: views::rowmajor::Ref<f32>,
78        offsets: ChunkOffsets,
79    ) -> Result<Self, TransposedTableError> {
80        let pivot_dim = pivots.ncols();
81        let offsets_dim = offsets.dim();
82        if pivot_dim != offsets_dim {
83            return Err(TransposedTableError::DimMismatch {
84                pivot_dim,
85                offsets_dim,
86            });
87        }
88
89        let ncenters = pivots.nrows();
90        let mut largest = 0;
91        let pivots: Box<[_]> = (0..offsets.len())
92            .map(|i| {
93                let range = offsets.at(i);
94                largest = largest.max(range.len());
95                let view = Strided::try_from_data(
96                    &(pivots.as_slice()[range.start..]),
97                    pivots.nrows(),
98                    range.len(),
99                    offsets.dim(),
100                )
101                .expect(
102                    "the check on `pivot_dim` and `offsets_dim` should cause this to never error",
103                );
104                pivots::Chunk::new(view).map_err(|source| TransposedTableError::PivotError {
105                    problem: i,
106                    total: offsets.len(),
107                    source,
108                })
109            })
110            .collect::<Result<Box<[_]>, TransposedTableError>>()?;
111
112        debug_assert_eq!(pivots.len(), offsets.len());
113        Ok(Self {
114            pivots,
115            offsets,
116            largest,
117            ncenters,
118        })
119    }
120
121    /// Return the number of pivots in each PQ chunk.
122    pub fn ncenters(&self) -> usize {
123        self.ncenters
124    }
125
126    /// Return the number of PQ chunks.
127    pub fn nchunks(&self) -> usize {
128        self.offsets.len()
129    }
130
131    /// Return the dimensionality of the full-precision vectors associated with this table.
132    pub fn dim(&self) -> usize {
133        self.offsets.dim()
134    }
135
136    /// Return a view over the schema offsets.
137    pub fn view_offsets(&self) -> ChunkOffsetsView<'_> {
138        self.offsets.as_view()
139    }
140
141    /// Perform PQ compression on the dataset by mapping each chunk in `data` to its nearest
142    /// neighbor in the corresponding entry in `chunks`.
143    ///
144    /// The index of the nearest neighbor is provided to `compression_delegate` along with its
145    /// corresponding row in data and chunk index.
146    ///
147    /// Calls to `compression_delegate` may occur in any order.
148    ///
149    /// Visitor will be invoked for all rows in `0..data.nrows()` and all chunks in
150    /// `0..schema.len()`.
151    ///
152    /// # Panics
153    ///
154    /// Panics under the following conditions:
155    /// * `data.cols() != self.dim()`: The number of columns in the source dataset must match
156    ///   the number of dimensions expected by the schema.
157    #[expect(clippy::expect_used)]
158    pub fn compress_batch<T, F, DelegateError>(
159        &self,
160        data: views::rowmajor::Ref<'_, T>,
161        mut compression_delegate: F,
162    ) -> Result<(), CompressError<DelegateError>>
163    where
164        T: Copy + Into<f32>,
165        F: FnMut(RowChunk, pivots::CompressionResult) -> Result<(), DelegateError>,
166        DelegateError: std::error::Error,
167    {
168        assert_eq!(
169            data.ncols(),
170            self.dim(),
171            "schema expects {} dimensions but data has {}",
172            self.dim(),
173            data.ncols()
174        );
175
176        // The batch size expected by the compression micro-kernel.
177        const SUB_BATCH_SIZE: usize = pivots::Chunk::batchsize();
178
179        let dim_nonzero = self.offsets.dim_nonzero();
180        let mut packing_buffer: Box<[f32]> =
181            (0..self.largest * SUB_BATCH_SIZE).map(|_| 0.0).collect();
182
183        // Stride along the chunks in the source matrix to keep the associated `Chunk` in the
184        // cache.
185        let nrows = data.nrows();
186        let ncols = data.ncols();
187        let slice = data.as_slice();
188
189        for (i, chunk) in self.pivots.iter().enumerate() {
190            let range = self.offsets.at(i);
191            if let Some(chunk_dim) = NonZeroUsize::new(range.len()) {
192                // Construct a view for the packing buffer for this chunk.
193                let mut packing_view = views::rowmajor::Mut::try_from_data(
194                    &mut packing_buffer[..SUB_BATCH_SIZE * chunk_dim.get()],
195                    SUB_BATCH_SIZE,
196                    chunk_dim.get(),
197                )
198                .expect("the packing buffer should have been sized correctly");
199
200                for row_start in (0..nrows).step_by(SUB_BATCH_SIZE) {
201                    let row_end = nrows.min(row_start + SUB_BATCH_SIZE);
202
203                    // If this is a full batch, the use the batched micro-kernel.
204                    if row_end - row_start == SUB_BATCH_SIZE {
205                        // When computing the stop offset, don't try to adjust for the length
206                        // of the underlying span.
207                        //
208                        // The control on our loop bounds mean we should never be in a situation
209                        // where this would occur, so we'd rather hit the indexing panic early.
210                        let mut linear_start = row_start * ncols + range.start;
211                        packing_view.rows_mut().for_each(|row| {
212                            pack(row, &slice[linear_start..linear_start + chunk_dim.get()]);
213                            linear_start += dim_nonzero.get();
214                        });
215
216                        let result = chunk.find_closest_batch(packing_view.as_view().into());
217
218                        // Invoke the delegate with the results.
219                        // If the delegate returns an error, wrap that error inside a
220                        // `CompressError` to add context and forward the error upward.
221                        for (j, &r) in result.iter().enumerate() {
222                            compression_delegate(
223                                RowChunk {
224                                    row: row_start + j,
225                                    chunk: i,
226                                },
227                                r,
228                            )
229                            .map_err(|inner| CompressError {
230                                inner,
231                                row: row_start + j,
232                                chunk: i,
233                                nearest: r.into_inner(),
234                            })?;
235                        }
236                    } else {
237                        // Handle remainders one at a time.
238                        for row in row_start..row_end {
239                            let linear_start = row * ncols + range.start;
240                            let linear_stop = linear_start + range.len();
241
242                            // Pre-convert to f32.
243                            let packed = &mut packing_view.row_mut(0);
244                            pack(packed, &slice[linear_start..linear_stop]);
245                            let result = chunk.find_closest(packed);
246
247                            compression_delegate(RowChunk { row, chunk: i }, result).map_err(
248                                |inner| CompressError {
249                                    inner,
250                                    row,
251                                    chunk: i,
252                                    nearest: result.into_inner(),
253                                },
254                            )?;
255                        }
256                    }
257                }
258            }
259        }
260        Ok(())
261    }
262
263    /// Compute the operation defined by `T` for each chunk in the query on all corresponding
264    /// pivots for that chunk, storing the result in the output matrix.
265    ///
266    /// For example, this can be used to compute squared L2 distances between chunks of the
267    /// query and the pivot table to create a fast run-time lookup table for these distances.
268    ///
269    /// This is currently implemented for the following operation types `T`:
270    ///
271    /// * `quantization::distances::SquaredL2`
272    /// * `quantization::distances::InnerProduct`
273    ///
274    /// # Arguments
275    ///
276    /// * `query`: The query slice to process. Must have length `self.dim()`.
277    /// * `partials`: Output matrix for the partial results. The result of the computation
278    ///   of chunk `i` against pivot `j` will be stored at `partials.element(i, j)`.
279    ///
280    ///   Must have `nrows = self.nchunks()` and `ncols = self.ncenters()`.
281    ///
282    /// # Panics
283    ///
284    /// Panics if:
285    /// * `query.len() != self.dim()`.
286    /// * `partisl.nrows() != self.nchunks()`.
287    /// * `partisl.ncols() != self.ncenters()`.
288    pub fn process_into<T, U>(&self, query: &[f32], mut partials: rowmajor::Mut<'_, U>)
289    where
290        T: pivots::ProcessInto<U>,
291    {
292        // Check Requirements
293        assert_eq!(
294            query.len(),
295            self.dim(),
296            "query has the wrong number of dimensions"
297        );
298        assert_eq!(
299            partials.ncols(),
300            self.ncenters(),
301            "output has the wrong number of columns"
302        );
303        assert_eq!(
304            partials.nrows(),
305            self.nchunks(),
306            "output has the wrong number of rows"
307        );
308
309        // Loop over each chunk.
310        std::iter::zip(self.pivots.iter(), partials.rows_mut())
311            .enumerate()
312            .for_each(|(i, (pivot, out))| {
313                let range = self.offsets.at(i);
314                T::process_into(pivot, &query[range], out);
315            });
316    }
317}
318
319/// Row and chunk indexes provided to the `compression_delegate` argument of `compress` to
320/// describe the position of the nearest neighbor being provided.
321pub struct RowChunk {
322    row: usize,
323    chunk: usize,
324}
325
326#[derive(Error, Debug)]
327#[error(
328    "compression delegate returned \"{inner}\" when processing row {row} and chunk \
329    {chunk} with nearest center {nearest}"
330)]
331pub struct CompressError<DelegateError: std::error::Error> {
332    inner: DelegateError,
333    row: usize,
334    chunk: usize,
335    nearest: u32,
336}
337
338#[inline(always)]
339fn pack<T>(dst: &mut [f32], src: &[T])
340where
341    T: Copy + Into<f32>,
342{
343    debug_assert_eq!(dst.len(), src.len());
344    std::iter::zip(dst.iter_mut(), src.iter()).for_each(|(d, &s)| *d = s.into());
345}
346
347///////////////
348// Coompress //
349///////////////
350
351impl CompressInto<&[f32], &mut [u8]> for TransposedTable {
352    type Error = TableCompressionError;
353    type Output = ();
354
355    /// Compress the full-precision vector `from` into the PQ byte buffer `to`.
356    ///
357    /// Compression is performed by partitioning `from` into chunks according to the offsets
358    /// schema in the table and then finding the closest pivot according to the L2 distance.
359    ///
360    /// The final compressed value is the index of the closest pivot.
361    ///
362    /// # Errors
363    ///
364    /// Returns errors under the following conditions:
365    ///
366    /// * `self.ncenters() > 256`: If the number of centers exceeds 256, then it cannot be
367    ///   guaranteed that the index of the closest pivot for a chunk will fit losslessly in
368    ///   an 8-bit integer.
369    ///
370    /// * `from.len() != self.dim()`: The full precision vector must have the dimensionality
371    ///   expected by the compression.
372    ///
373    /// * `to.len() != self.nchunks()`: The PQ buffer must be sized appropriately.
374    ///
375    /// * If any chunk is sufficiently far from all centers that its distance becomes
376    ///   infinity to all centers.
377    ///
378    /// # Allocates
379    ///
380    /// This function should not allocate when successful.
381    ///
382    /// # Parallelism
383    ///
384    /// This function is single-threaded.
385    fn compress_into(&self, from: &[f32], to: &mut [u8]) -> Result<(), Self::Error> {
386        if self.ncenters() > 256 {
387            return Err(Self::Error::CannotCompressToByte(self.ncenters()));
388        }
389        if from.len() != self.dim() {
390            return Err(Self::Error::InvalidInputDim(self.dim(), from.len()));
391        }
392        if to.len() != self.nchunks() {
393            return Err(Self::Error::InvalidOutputDim(self.nchunks(), to.len()));
394        }
395
396        std::iter::zip(self.pivots.iter(), to.iter_mut())
397            .enumerate()
398            .try_for_each(|(i, (pivot, to))| {
399                let range = self.offsets.at(i);
400                let result = pivot.find_closest(&from[range]);
401                result.map(
402                    |v| *to = v as u8, // conversion guaranteed to be lossless
403                    || Self::Error::InfinityOrNaN(i),
404                )
405            })
406    }
407}
408
409#[derive(Error, Debug)]
410pub enum TableBatchCompressionError {
411    #[error("num centers ({0}) must be at most 256 to compress into a byte vector")]
412    CannotCompressToByte(usize),
413    #[error("invalid input len - expected {0}, got {1}")]
414    InvalidInputDim(usize, usize),
415    #[error("invalid PQ buffer len - expected {0}, got {1}")]
416    InvalidOutputDim(usize, usize),
417    #[error(
418        "input and output must have the same number of rows - instead, got {0} and {1} \
419         (respectively)"
420    )]
421    UnequalRows(usize, usize),
422    #[error(
423        "a value of infinity or NaN was observed while compressing chunk {0} of batch input {1}"
424    )]
425    InfinityOrNaN(usize, usize),
426}
427
428impl<T> CompressInto<rowmajor::Ref<'_, T>, rowmajor::Mut<'_, u8>> for TransposedTable
429where
430    T: Copy + Into<f32>,
431{
432    type Error = TableBatchCompressionError;
433    type Output = ();
434
435    /// Compress each full-precision row in `from` into the corresponding row in `to`.
436    ///
437    /// Compression is performed by partitioning `from` into chunks according to the offsets
438    /// schema in the table and then finding the closest pivot according to the L2 distance.
439    ///
440    /// The final compressed value is the index of the closest pivot.
441    ///
442    /// # Errors
443    ///
444    /// Returns errors under the following conditions:
445    ///
446    /// * `self.ncenters() > 256`: If the number of centers exceeds 256, then it cannot be
447    ///   guaranteed that the index of the closest pivot for a chunk will fit losslessly in
448    ///   an 8-bit integer.
449    ///
450    /// * `from.ncols() != self.dim()`: The full precision data must have the dimensionality
451    ///   expected by the compression.
452    ///
453    /// * `to.ncols() != self.nchunks()`: The PQ buffer must be sized appropriately.
454    ///
455    /// * `from.nrows() == to.nrows()`: The input and output buffers must have the same
456    ///   number of elements.
457    ///
458    /// * If any chunk is sufficiently far from all centers that its distance becomes
459    ///   infinity to all centers.
460    ///
461    /// # Allocates
462    ///
463    /// Allocates scratch memory proportional to the length of the largest chunk.
464    ///
465    /// # Parallelism
466    ///
467    /// This function is single-threaded.
468    fn compress_into(
469        &self,
470        from: rowmajor::Ref<'_, T>,
471        mut to: rowmajor::Mut<'_, u8>,
472    ) -> Result<(), Self::Error> {
473        if self.ncenters() > 256 {
474            return Err(Self::Error::CannotCompressToByte(self.ncenters()));
475        }
476        if from.ncols() != self.dim() {
477            return Err(Self::Error::InvalidInputDim(self.dim(), from.ncols()));
478        }
479        if to.ncols() != self.nchunks() {
480            return Err(Self::Error::InvalidOutputDim(self.nchunks(), to.ncols()));
481        }
482        if from.nrows() != to.nrows() {
483            return Err(Self::Error::UnequalRows(from.nrows(), to.nrows()));
484        }
485
486        // The `CompressionError` already has all the information we need, so we make the
487        // `Delegate` error light-weight.
488        #[derive(Debug, Error)]
489        #[error("unreachable")]
490        struct PassThrough;
491
492        // Do the compression. We will reformat the NaN warning from `CompressionError`
493        // if needed.
494        let result = self.compress_batch(
495            from,
496            |RowChunk { row, chunk }, result| -> Result<(), PassThrough> {
497                result.map(|v| *to.element_mut(row, chunk) = v as u8, || PassThrough)
498            },
499        );
500
501        result.map_err(|err| Self::Error::InfinityOrNaN(err.chunk, err.row))
502    }
503}
504
505#[cfg(test)]
506mod test_compression {
507    use std::collections::HashSet;
508
509    use diskann_vector::{
510        MathematicalValue, Norm, PureDistanceFunction, distance, norm::FastL2NormSquared,
511    };
512    use rand::{
513        Rng, SeedableRng,
514        distr::{Distribution, StandardUniform, Uniform},
515        rngs::StdRng,
516    };
517
518    use super::*;
519    #[cfg(not(miri))]
520    use crate::product::tables::test::{
521        check_pqtable_batch_compression_errors, check_pqtable_single_compression_errors,
522    };
523    use crate::{
524        distances::{Cosine, InnerProduct, SquaredL2},
525        error::format,
526        product::tables::{
527            lookup::DotAndNorm,
528            test::{create_dataset, create_pivot_tables},
529        },
530    };
531    use diskann_utils::lazy_format;
532
533    //////////////////////////////
534    // Transposed Table Methods //
535    //////////////////////////////
536
537    // Test that an error is returned when the dimension between the pivots and offsets
538    // disagree.
539    #[test]
540    fn error_on_mismatch_dim() {
541        let pivots = views::rowmajor::Owned::from_element(3, 5, 0.0);
542        let offsets = ChunkOffsets::new(Box::new([0, 1, 6])).unwrap();
543        let result = TransposedTable::from_parts(pivots.as_view(), offsets);
544        assert!(result.is_err(), "dimensions are not equal");
545        assert_eq!(
546            result.unwrap_err().to_string(),
547            "pivots have 5 dimensions while the offsets expect 6"
548        );
549    }
550
551    // Test that an error is returned when the dimension between the pivots and offsets
552    // disagree.
553    #[test]
554    fn error_on_empty() {
555        let pivots = views::rowmajor::Owned::from_element(0, 5, 0.0);
556        let offsets = ChunkOffsets::new(Box::new([0, 1, 5])).unwrap();
557        let result = TransposedTable::from_parts(pivots.as_view(), offsets);
558        assert!(result.is_err(), "dimensions are not equal");
559
560        let expected = [
561            "error constructing pivot 0 of 2",
562            "    caused by: cannot construct a Chunk from a source with zero length",
563        ]
564        .join("\n");
565
566        assert_eq!(format(&result.unwrap_err()), expected,);
567    }
568
569    #[test]
570    fn basic_table() {
571        let mut rng = StdRng::seed_from_u64(0xd96bac968083ec29);
572        for dim in [5, 10, 12] {
573            // Sweep over enough totals to ensure the inner chunks have a non-trivial layout.
574            for total in [1, 2, 3, 7, 8, 9, 10] {
575                let pivots = views::rowmajor::Owned::from_fn(total, dim, |_| -> f32 {
576                    StandardUniform {}.sample(&mut rng)
577                });
578                let offsets = ChunkOffsets::new(Box::new([0, 1, 3, dim])).unwrap();
579                let table = TransposedTable::from_parts(pivots.as_view(), offsets.clone()).unwrap();
580
581                assert_eq!(table.ncenters(), total);
582                assert_eq!(table.nchunks(), offsets.len());
583                assert_eq!(table.dim(), offsets.dim());
584
585                // This kind of looks into the guts of this data structure, but is an extra
586                // check that the plumbing was performed properly.
587                for chunk in 0..offsets.len() {
588                    let range = offsets.at(chunk);
589                    let pivot = &table.pivots[chunk];
590                    for row in 0..total {
591                        let r = &pivots.row(row)[range.clone()];
592                        for (col, expected) in r.iter().enumerate() {
593                            assert_eq!(pivot.get(row, col), *expected);
594                        }
595                    }
596                }
597
598                assert_eq!(table.view_offsets(), offsets.as_view());
599            }
600        }
601    }
602
603    /////////////////
604    // Compression //
605    /////////////////
606
607    #[derive(Error, Debug)]
608    #[error("unreachable reached")]
609    struct Infallible;
610
611    #[test]
612    fn test_happy_path() {
613        // Feed in chunks of dimension 1, 2, 3, ... 16.
614        //
615        // If we're using MIRI, max out at 7 dimensions.
616        let offsets: Vec<usize> = if cfg!(miri) {
617            vec![0, 1, 3, 6, 10, 15, 21, 28, 36]
618        } else {
619            vec![
620                0, 1, 3, 6, 10, 15, 21, 28, 36, 45, 55, 66, 78, 91, 105, 120, 136,
621            ]
622        };
623
624        let schema = ChunkOffsetsView::new(&offsets).unwrap();
625        let mut rng = StdRng::seed_from_u64(0x88e3d3366501ad6c);
626
627        let num_data = if cfg!(miri) {
628            vec![0, 7, 8, 10]
629        } else {
630            vec![0, 1, 2, 3, 4, 16, 17, 18, 19]
631        };
632
633        let num_trials = if cfg!(miri) { 1 } else { 10 };
634
635        // Strategically pick `num_centers` so we cover the corner cases in lower-level
636        // handling.
637        //
638        // This includes:
639        // * Full blocks with an even number of total blocks (num centers = 16);
640        // * Full blocks with an odd number of total blocks (num centers = 24);
641        // * Partially full blocks with the last block even (num centers = 13);
642        // * Partially full blocks with the last block odd (num centers = 17);
643        for &num_centers in [16, 24, 13, 17].iter() {
644            for &num_data in num_data.iter() {
645                for trial in 0..num_trials {
646                    let context = lazy_format!(
647                        "happy path, num centers = {}, num data = {}, trial = {}",
648                        num_centers,
649                        num_data,
650                        trial,
651                    );
652
653                    println!("Currently = {}", context);
654
655                    let (pivots, offsets) = create_pivot_tables(schema.to_owned(), num_centers);
656                    let table = TransposedTable::from_parts(pivots.as_view(), offsets).unwrap();
657                    let (data, expected) = create_dataset(schema, num_centers, num_data, &mut rng);
658
659                    let mut called = HashSet::<(usize, usize)>::new();
660
661                    // Direct method call.
662                    table
663                        .compress_batch(
664                            data.as_view(),
665                            |RowChunk { row, chunk }, value| -> Result<(), Infallible> {
666                                assert!(value.is_okay());
667                                // Ensure that this is the expected value.
668                                assert_eq!(
669                                value.unwrap() as usize,
670                                *expected.element(row, chunk),
671                                "failed at (row = {row}, chunk = {chunk}). data = {:?}, context: {}",
672                                &(data.row(row)[schema.at(chunk)]),
673                                context,
674                            );
675
676                                // (A) Ensure that this combination of row and chunk hasn't
677                                //     been called before.
678                                // (B) Record that this combination has been called.
679                                assert!(
680                                    called.insert((row, chunk)),
681                                    "row {row} and chunk {chunk}, called multiple times. Context = {}",
682                                    context,
683                                );
684
685                                Ok(())
686                            },
687                        )
688                        .unwrap();
689
690                    assert_eq!(called.len(), num_data * schema.len());
691
692                    // Trait Interface.
693                    let mut output =
694                        views::rowmajor::Owned::from_element(num_data, schema.len(), 0);
695                    table
696                        .compress_into(data.as_view(), output.as_view_mut())
697                        .unwrap();
698
699                    assert_eq!(output.nrows(), expected.nrows());
700                    assert_eq!(output.ncols(), expected.ncols());
701                    for row in 0..output.nrows() {
702                        for col in 0..output.ncols() {
703                            assert_eq!(
704                                *output.element(row, col) as usize,
705                                *expected.element(row, col),
706                                "failed on row {}, col {}. Context = {}",
707                                row,
708                                col,
709                                context,
710                            );
711                        }
712                    }
713
714                    // Trait inteface - single step.
715                    let mut output = vec![0; schema.len()];
716                    for (i, (row, expected)) in
717                        std::iter::zip(data.rows(), expected.rows()).enumerate()
718                    {
719                        table.compress_into(row, output.as_mut_slice()).unwrap();
720                        for (d, (o, e)) in
721                            std::iter::zip(output.iter(), expected.iter()).enumerate()
722                        {
723                            assert_eq!(
724                                *o as usize, *e,
725                                "failed on row {}, col {}. Context = {}",
726                                i, d, context
727                            );
728                        }
729                    }
730                }
731            }
732        }
733    }
734
735    ////////////////////
736    // Error Handling //
737    ////////////////////
738
739    #[test]
740    #[should_panic(expected = "schema expects 4 dimensions but data has 5")]
741    fn panic_on_dim_mismatch() {
742        let offsets = [0, 4];
743        let data: Vec<f32> = vec![0.0; 5];
744
745        let schema = ChunkOffsetsView::new(&offsets).unwrap();
746        let (pivots, offsets) = create_pivot_tables(schema.to_owned(), 3);
747        let table = TransposedTable::from_parts(pivots.as_view(), offsets).unwrap();
748
749        // should panic
750        let _ = table.compress_batch(
751            views::rowmajor::Ref::try_from_data(data.as_slice(), 1, 5).unwrap(),
752            |_, _| -> Result<(), Infallible> { panic!("this shouldn't be called") },
753        );
754    }
755
756    #[derive(Error, Debug)]
757    #[error("compression delegate error with {0}")]
758    struct DelegateError(u64);
759
760    // The strategy here is to construct a delegate that returns an error on a specific
761    // row and chunk. We then make sure that the error is propagated successfully along
762    // with the recorded row and chunk.
763    #[test]
764    fn test_delegate_error_propagation() {
765        let offsets: Vec<usize> = vec![0, 1, 7];
766        let schema = ChunkOffsetsView::new(&offsets).unwrap();
767        let mut rng = StdRng::seed_from_u64(0xc35a90da17fafa2a);
768
769        let num_centers = 3;
770        let num_data = 7;
771
772        let (pivots, offsets) = create_pivot_tables(schema.to_owned(), num_centers);
773        let table = TransposedTable::from_parts(pivots.as_view(), offsets).unwrap();
774        let (data, _) = create_dataset(schema, num_centers, num_data, &mut rng);
775
776        let data_view =
777            views::rowmajor::Ref::try_from_data(data.as_slice(), num_data, schema.dim()).unwrap();
778        let distribution = rand_distr::StandardUniform {};
779
780        for row in 0..data_view.nrows() {
781            for chunk in 0..schema.len() {
782                let context = lazy_format!("row = {row}, chunk = {chunk}");
783
784                // Generate a random number for the delegate error.
785                let value: u64 = rng.sample(distribution);
786
787                let result = table.compress_batch(
788                    data_view,
789                    |RowChunk {
790                         row: this_row,
791                         chunk: this_chunk,
792                     },
793                     _| {
794                        if this_row == row && this_chunk == chunk {
795                            Err(DelegateError(value))
796                        } else {
797                            Ok(())
798                        }
799                    },
800                );
801                assert!(result.is_err(), "{}", context);
802
803                let message = result.unwrap_err().to_string();
804                assert!(
805                    message.contains(&format!("{}", DelegateError(value))),
806                    "{}",
807                    context
808                );
809                assert!(message.contains("delegate returned"));
810                assert!(message.contains(&format!("when processing row {row} and chunk {chunk}")));
811            }
812        }
813    }
814
815    #[test]
816    #[cfg(not(miri))]
817    fn test_table_single_compression_errors() {
818        check_pqtable_single_compression_errors(
819            &|pivots: views::rowmajor::Owned<f32>, offsets| {
820                TransposedTable::from_parts(pivots.as_view(), offsets).unwrap()
821            },
822            &"TranposedTable",
823        )
824    }
825
826    #[test]
827    #[cfg(not(miri))]
828    fn test_table_batch_compression_errors() {
829        check_pqtable_batch_compression_errors(
830            &|pivots: views::rowmajor::Owned<f32>, offsets| {
831                TransposedTable::from_parts(pivots.as_view(), offsets).unwrap()
832            },
833            &"TranposedTable",
834        )
835    }
836
837    /////////////////////////
838    // Test `process_into` //
839    /////////////////////////
840
841    fn test_process_into_impl(
842        num_chunks: usize,
843        num_centers: usize,
844        num_trials: usize,
845        rng: &mut StdRng,
846    ) {
847        // Choose the chunk size randomly from this distribution. Keep choosing chunks
848        // sizes until the desired `num_chunks` is reached.
849        //
850        // The sum of chunk sizes gives the dimensionality.
851        let chunk_size_distribution = Uniform::<usize>::new(1, 6).unwrap();
852
853        // Use integer values for the value distribution to avoid dealing with floating
854        // point rounding.
855        let value_distribution = Uniform::<i32>::new(-10, 10).unwrap();
856
857        for trial in 0..num_trials {
858            let mut offsets: Vec<usize> = vec![0];
859            for _ in 0..num_chunks {
860                let chunk_size = chunk_size_distribution.sample(rng);
861                offsets.push(offsets.last().unwrap() + chunk_size);
862            }
863
864            let offsets = ChunkOffsets::new(offsets.into()).unwrap();
865            let dim = offsets.dim();
866            let pivots = views::rowmajor::Owned::<f32>::from_fn(num_centers, dim, |_| {
867                value_distribution.sample(rng) as f32
868            });
869
870            let table = TransposedTable::from_parts(pivots.as_view(), offsets.clone()).unwrap();
871
872            let mut output =
873                views::rowmajor::Owned::<f32>::from_element(num_chunks, num_centers, 0.0);
874
875            let query: Vec<_> = (0..dim)
876                .map(|_| value_distribution.sample(rng) as f32)
877                .collect();
878
879            // Inner Product
880            table.process_into::<InnerProduct, f32>(&query, output.as_view_mut());
881
882            for chunk in 0..num_chunks {
883                let range = offsets.at(chunk);
884                let query_chunk = &query[range.clone()];
885                for center in 0..num_centers {
886                    let data_chunk = &pivots.row(center)[range.clone()];
887                    let expected: f32 = distance::InnerProduct::evaluate(query_chunk, data_chunk);
888                    assert_eq!(
889                        *output.element(chunk, center),
890                        expected,
891                        "failed on (chunk, center) = ({}, {}) - offsets = {:?} - trial = {}",
892                        chunk,
893                        center,
894                        offsets,
895                        trial,
896                    );
897                }
898            }
899
900            // Squared L2
901            table.process_into::<SquaredL2, f32>(&query, output.as_view_mut());
902
903            for chunk in 0..num_chunks {
904                let range = offsets.at(chunk);
905                let query_chunk = &query[range.clone()];
906                for center in 0..num_centers {
907                    let data_chunk = &pivots.row(center)[range.clone()];
908                    let expected: f32 = distance::SquaredL2::evaluate(query_chunk, data_chunk);
909                    assert_eq!(
910                        *output.element(chunk, center),
911                        expected,
912                        "failed on (chunk, center) = ({}, {}) - offsets = {:?} - trial = {}",
913                        chunk,
914                        center,
915                        offsets,
916                        trial,
917                    );
918                }
919            }
920
921            // Cosine
922            let mut output = views::rowmajor::Owned::<DotAndNorm>::from_element(
923                num_chunks,
924                num_centers,
925                DotAndNorm::default(),
926            );
927
928            table.process_into::<Cosine, DotAndNorm>(&query, output.as_view_mut());
929            for chunk in 0..num_chunks {
930                let range = offsets.at(chunk);
931                let query_chunk = &query[range.clone()];
932
933                for center in 0..num_centers {
934                    let data_chunk = &pivots.row(center)[range.clone()];
935                    let expected_dot: MathematicalValue<f32> =
936                        distance::InnerProduct::evaluate(query_chunk, data_chunk);
937
938                    assert_eq!(
939                        output.element(chunk, center).dot(),
940                        expected_dot.into_inner(),
941                        "failed on (chunk, center) = ({}, {}) - offsets = {:?} - trial = {}",
942                        chunk,
943                        center,
944                        offsets,
945                        trial,
946                    );
947
948                    let expected_norm: f32 = (FastL2NormSquared).evaluate(data_chunk);
949                    assert_eq!(
950                        output.element(chunk, center).square_norm(),
951                        expected_norm,
952                        "failed on (chunk, center) = ({}, {}) - offsets = {:?} - trial = {}",
953                        chunk,
954                        center,
955                        offsets,
956                        trial,
957                    );
958                }
959            }
960        }
961    }
962
963    #[test]
964    fn test_process_into() {
965        let mut rng = StdRng::seed_from_u64(0x0e3cf3ba4b27e7f8);
966
967        let num_chunks_range = if cfg!(miri) { 4..5 } else { 1..5 };
968
969        let num_centers: Vec<usize> = if cfg!(miri) {
970            vec![1, 7, 16, 33, 47]
971        } else {
972            (1..48).collect()
973        };
974
975        for num_chunks in num_chunks_range {
976            for num_centers in num_centers.clone() {
977                test_process_into_impl(num_chunks, num_centers, 2, &mut rng);
978            }
979        }
980    }
981
982    #[test]
983    #[should_panic(expected = "query has the wrong number of dimensions")]
984    fn test_process_into_panics_query() {
985        let offsets = ChunkOffsets::new(Box::new([0, 1, 5])).unwrap();
986        let data = views::rowmajor::Owned::<f32>::from_element(3, 5, 0.0);
987        let table = TransposedTable::from_parts(data.as_view(), offsets).unwrap();
988        assert_eq!(table.dim(), 5);
989
990        // query has the wrong length.
991        let query = vec![0.0; table.dim() - 1];
992        let mut partials =
993            views::rowmajor::Owned::from_element(table.nchunks(), table.ncenters(), 0.0);
994        table.process_into::<InnerProduct, f32>(&query, partials.as_view_mut());
995    }
996
997    #[test]
998    #[should_panic(expected = "output has the wrong number of rows")]
999    fn test_process_into_panics_partials_rows() {
1000        let offsets = ChunkOffsets::new(Box::new([0, 1, 5])).unwrap();
1001        let data = views::rowmajor::Owned::<f32>::from_element(3, 5, 0.0);
1002        let table = TransposedTable::from_parts(data.as_view(), offsets).unwrap();
1003        assert_eq!(table.dim(), 5);
1004
1005        let query = vec![0.0; table.dim()];
1006        // partials has the wrong numbers of rows.
1007        let mut partials =
1008            views::rowmajor::Owned::from_element(table.nchunks() - 1, table.ncenters(), 0.0);
1009        table.process_into::<InnerProduct, f32>(&query, partials.as_view_mut());
1010    }
1011
1012    #[test]
1013    #[should_panic(expected = "output has the wrong number of columns")]
1014    fn test_process_into_panics_partials_cols() {
1015        let offsets = ChunkOffsets::new(Box::new([0, 1, 5])).unwrap();
1016        let data = views::rowmajor::Owned::<f32>::from_element(3, 5, 0.0);
1017        let table = TransposedTable::from_parts(data.as_view(), offsets).unwrap();
1018        assert_eq!(table.dim(), 5);
1019
1020        let query = vec![0.0; table.dim()];
1021        // partials has the wrong numbers of rows.
1022        let mut partials =
1023            views::rowmajor::Owned::from_element(table.nchunks(), table.ncenters() - 1, 0.0);
1024        table.process_into::<InnerProduct, f32>(&query, partials.as_view_mut());
1025    }
1026}