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>(&self, query: &[f32], mut partials: rowmajor::Mut<'_, f32>)
289    where
290        T: pivots::ProcessInto,
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::{PureDistanceFunction, distance};
510    use rand::{
511        Rng, SeedableRng,
512        distr::{Distribution, StandardUniform, Uniform},
513        rngs::StdRng,
514    };
515
516    use super::*;
517    #[cfg(not(miri))]
518    use crate::product::tables::test::{
519        check_pqtable_batch_compression_errors, check_pqtable_single_compression_errors,
520    };
521    use crate::{
522        distances::{InnerProduct, SquaredL2},
523        error::format,
524        product::tables::test::{create_dataset, create_pivot_tables},
525    };
526    use diskann_utils::lazy_format;
527
528    //////////////////////////////
529    // Transposed Table Methods //
530    //////////////////////////////
531
532    // Test that an error is returned when the dimension between the pivots and offsets
533    // disagree.
534    #[test]
535    fn error_on_mismatch_dim() {
536        let pivots = views::rowmajor::Owned::from_element(3, 5, 0.0);
537        let offsets = ChunkOffsets::new(Box::new([0, 1, 6])).unwrap();
538        let result = TransposedTable::from_parts(pivots.as_view(), offsets);
539        assert!(result.is_err(), "dimensions are not equal");
540        assert_eq!(
541            result.unwrap_err().to_string(),
542            "pivots have 5 dimensions while the offsets expect 6"
543        );
544    }
545
546    // Test that an error is returned when the dimension between the pivots and offsets
547    // disagree.
548    #[test]
549    fn error_on_empty() {
550        let pivots = views::rowmajor::Owned::from_element(0, 5, 0.0);
551        let offsets = ChunkOffsets::new(Box::new([0, 1, 5])).unwrap();
552        let result = TransposedTable::from_parts(pivots.as_view(), offsets);
553        assert!(result.is_err(), "dimensions are not equal");
554
555        let expected = [
556            "error constructing pivot 0 of 2",
557            "    caused by: cannot construct a Chunk from a source with zero length",
558        ]
559        .join("\n");
560
561        assert_eq!(format(&result.unwrap_err()), expected,);
562    }
563
564    #[test]
565    fn basic_table() {
566        let mut rng = StdRng::seed_from_u64(0xd96bac968083ec29);
567        for dim in [5, 10, 12] {
568            // Sweep over enough totals to ensure the inner chunks have a non-trivial layout.
569            for total in [1, 2, 3, 7, 8, 9, 10] {
570                let pivots = views::rowmajor::Owned::from_fn(total, dim, |_| -> f32 {
571                    StandardUniform {}.sample(&mut rng)
572                });
573                let offsets = ChunkOffsets::new(Box::new([0, 1, 3, dim])).unwrap();
574                let table = TransposedTable::from_parts(pivots.as_view(), offsets.clone()).unwrap();
575
576                assert_eq!(table.ncenters(), total);
577                assert_eq!(table.nchunks(), offsets.len());
578                assert_eq!(table.dim(), offsets.dim());
579
580                // This kind of looks into the guts of this data structure, but is an extra
581                // check that the plumbing was performed properly.
582                for chunk in 0..offsets.len() {
583                    let range = offsets.at(chunk);
584                    let pivot = &table.pivots[chunk];
585                    for row in 0..total {
586                        let r = &pivots.row(row)[range.clone()];
587                        for (col, expected) in r.iter().enumerate() {
588                            assert_eq!(pivot.get(row, col), *expected);
589                        }
590                    }
591                }
592
593                assert_eq!(table.view_offsets(), offsets.as_view());
594            }
595        }
596    }
597
598    /////////////////
599    // Compression //
600    /////////////////
601
602    #[derive(Error, Debug)]
603    #[error("unreachable reached")]
604    struct Infallible;
605
606    #[test]
607    fn test_happy_path() {
608        // Feed in chunks of dimension 1, 2, 3, ... 16.
609        //
610        // If we're using MIRI, max out at 7 dimensions.
611        let offsets: Vec<usize> = if cfg!(miri) {
612            vec![0, 1, 3, 6, 10, 15, 21, 28, 36]
613        } else {
614            vec![
615                0, 1, 3, 6, 10, 15, 21, 28, 36, 45, 55, 66, 78, 91, 105, 120, 136,
616            ]
617        };
618
619        let schema = ChunkOffsetsView::new(&offsets).unwrap();
620        let mut rng = StdRng::seed_from_u64(0x88e3d3366501ad6c);
621
622        let num_data = if cfg!(miri) {
623            vec![0, 7, 8, 10]
624        } else {
625            vec![0, 1, 2, 3, 4, 16, 17, 18, 19]
626        };
627
628        let num_trials = if cfg!(miri) { 1 } else { 10 };
629
630        // Strategically pick `num_centers` so we cover the corner cases in lower-level
631        // handling.
632        //
633        // This includes:
634        // * Full blocks with an even number of total blocks (num centers = 16);
635        // * Full blocks with an odd number of total blocks (num centers = 24);
636        // * Partially full blocks with the last block even (num centers = 13);
637        // * Partially full blocks with the last block odd (num centers = 17);
638        for &num_centers in [16, 24, 13, 17].iter() {
639            for &num_data in num_data.iter() {
640                for trial in 0..num_trials {
641                    let context = lazy_format!(
642                        "happy path, num centers = {}, num data = {}, trial = {}",
643                        num_centers,
644                        num_data,
645                        trial,
646                    );
647
648                    println!("Currently = {}", context);
649
650                    let (pivots, offsets) = create_pivot_tables(schema.to_owned(), num_centers);
651                    let table = TransposedTable::from_parts(pivots.as_view(), offsets).unwrap();
652                    let (data, expected) = create_dataset(schema, num_centers, num_data, &mut rng);
653
654                    let mut called = HashSet::<(usize, usize)>::new();
655
656                    // Direct method call.
657                    table
658                        .compress_batch(
659                            data.as_view(),
660                            |RowChunk { row, chunk }, value| -> Result<(), Infallible> {
661                                assert!(value.is_okay());
662                                // Ensure that this is the expected value.
663                                assert_eq!(
664                                value.unwrap() as usize,
665                                *expected.element(row, chunk),
666                                "failed at (row = {row}, chunk = {chunk}). data = {:?}, context: {}",
667                                &(data.row(row)[schema.at(chunk)]),
668                                context,
669                            );
670
671                                // (A) Ensure that this combination of row and chunk hasn't
672                                //     been called before.
673                                // (B) Record that this combination has been called.
674                                assert!(
675                                    called.insert((row, chunk)),
676                                    "row {row} and chunk {chunk}, called multiple times. Context = {}",
677                                    context,
678                                );
679
680                                Ok(())
681                            },
682                        )
683                        .unwrap();
684
685                    assert_eq!(called.len(), num_data * schema.len());
686
687                    // Trait Interface.
688                    let mut output =
689                        views::rowmajor::Owned::from_element(num_data, schema.len(), 0);
690                    table
691                        .compress_into(data.as_view(), output.as_view_mut())
692                        .unwrap();
693
694                    assert_eq!(output.nrows(), expected.nrows());
695                    assert_eq!(output.ncols(), expected.ncols());
696                    for row in 0..output.nrows() {
697                        for col in 0..output.ncols() {
698                            assert_eq!(
699                                *output.element(row, col) as usize,
700                                *expected.element(row, col),
701                                "failed on row {}, col {}. Context = {}",
702                                row,
703                                col,
704                                context,
705                            );
706                        }
707                    }
708
709                    // Trait inteface - single step.
710                    let mut output = vec![0; schema.len()];
711                    for (i, (row, expected)) in
712                        std::iter::zip(data.rows(), expected.rows()).enumerate()
713                    {
714                        table.compress_into(row, output.as_mut_slice()).unwrap();
715                        for (d, (o, e)) in
716                            std::iter::zip(output.iter(), expected.iter()).enumerate()
717                        {
718                            assert_eq!(
719                                *o as usize, *e,
720                                "failed on row {}, col {}. Context = {}",
721                                i, d, context
722                            );
723                        }
724                    }
725                }
726            }
727        }
728    }
729
730    ////////////////////
731    // Error Handling //
732    ////////////////////
733
734    #[test]
735    #[should_panic(expected = "schema expects 4 dimensions but data has 5")]
736    fn panic_on_dim_mismatch() {
737        let offsets = [0, 4];
738        let data: Vec<f32> = vec![0.0; 5];
739
740        let schema = ChunkOffsetsView::new(&offsets).unwrap();
741        let (pivots, offsets) = create_pivot_tables(schema.to_owned(), 3);
742        let table = TransposedTable::from_parts(pivots.as_view(), offsets).unwrap();
743
744        // should panic
745        let _ = table.compress_batch(
746            views::rowmajor::Ref::try_from_data(data.as_slice(), 1, 5).unwrap(),
747            |_, _| -> Result<(), Infallible> { panic!("this shouldn't be called") },
748        );
749    }
750
751    #[derive(Error, Debug)]
752    #[error("compression delegate error with {0}")]
753    struct DelegateError(u64);
754
755    // The strategy here is to construct a delegate that returns an error on a specific
756    // row and chunk. We then make sure that the error is propagated successfully along
757    // with the recorded row and chunk.
758    #[test]
759    fn test_delegate_error_propagation() {
760        let offsets: Vec<usize> = vec![0, 1, 7];
761        let schema = ChunkOffsetsView::new(&offsets).unwrap();
762        let mut rng = StdRng::seed_from_u64(0xc35a90da17fafa2a);
763
764        let num_centers = 3;
765        let num_data = 7;
766
767        let (pivots, offsets) = create_pivot_tables(schema.to_owned(), num_centers);
768        let table = TransposedTable::from_parts(pivots.as_view(), offsets).unwrap();
769        let (data, _) = create_dataset(schema, num_centers, num_data, &mut rng);
770
771        let data_view =
772            views::rowmajor::Ref::try_from_data(data.as_slice(), num_data, schema.dim()).unwrap();
773        let distribution = rand_distr::StandardUniform {};
774
775        for row in 0..data_view.nrows() {
776            for chunk in 0..schema.len() {
777                let context = lazy_format!("row = {row}, chunk = {chunk}");
778
779                // Generate a random number for the delegate error.
780                let value: u64 = rng.sample(distribution);
781
782                let result = table.compress_batch(
783                    data_view,
784                    |RowChunk {
785                         row: this_row,
786                         chunk: this_chunk,
787                     },
788                     _| {
789                        if this_row == row && this_chunk == chunk {
790                            Err(DelegateError(value))
791                        } else {
792                            Ok(())
793                        }
794                    },
795                );
796                assert!(result.is_err(), "{}", context);
797
798                let message = result.unwrap_err().to_string();
799                assert!(
800                    message.contains(&format!("{}", DelegateError(value))),
801                    "{}",
802                    context
803                );
804                assert!(message.contains("delegate returned"));
805                assert!(message.contains(&format!("when processing row {row} and chunk {chunk}")));
806            }
807        }
808    }
809
810    #[test]
811    #[cfg(not(miri))]
812    fn test_table_single_compression_errors() {
813        check_pqtable_single_compression_errors(
814            &|pivots: views::rowmajor::Owned<f32>, offsets| {
815                TransposedTable::from_parts(pivots.as_view(), offsets).unwrap()
816            },
817            &"TranposedTable",
818        )
819    }
820
821    #[test]
822    #[cfg(not(miri))]
823    fn test_table_batch_compression_errors() {
824        check_pqtable_batch_compression_errors(
825            &|pivots: views::rowmajor::Owned<f32>, offsets| {
826                TransposedTable::from_parts(pivots.as_view(), offsets).unwrap()
827            },
828            &"TranposedTable",
829        )
830    }
831
832    /////////////////////////
833    // Test `process_into` //
834    /////////////////////////
835
836    fn test_process_into_impl(
837        num_chunks: usize,
838        num_centers: usize,
839        num_trials: usize,
840        rng: &mut StdRng,
841    ) {
842        // Choose the chunk size randomly from this distribution. Keep choosing chunks
843        // sizes until the desired `num_chunks` is reached.
844        //
845        // The sum of chunk sizes gives the dimensionality.
846        let chunk_size_distribution = Uniform::<usize>::new(1, 6).unwrap();
847
848        // Use integer values for the value distribution to avoid dealing with floating
849        // point rounding.
850        let value_distribution = Uniform::<i32>::new(-10, 10).unwrap();
851
852        for trial in 0..num_trials {
853            let mut offsets: Vec<usize> = vec![0];
854            for _ in 0..num_chunks {
855                let chunk_size = chunk_size_distribution.sample(rng);
856                offsets.push(offsets.last().unwrap() + chunk_size);
857            }
858
859            let offsets = ChunkOffsets::new(offsets.into()).unwrap();
860            let dim = offsets.dim();
861            let pivots = views::rowmajor::Owned::<f32>::from_fn(num_centers, dim, |_| {
862                value_distribution.sample(rng) as f32
863            });
864
865            let table = TransposedTable::from_parts(pivots.as_view(), offsets.clone()).unwrap();
866
867            let mut output =
868                views::rowmajor::Owned::<f32>::from_element(num_chunks, num_centers, 0.0);
869            let query: Vec<_> = (0..dim)
870                .map(|_| value_distribution.sample(rng) as f32)
871                .collect();
872
873            // Inner Product
874            table.process_into::<InnerProduct>(&query, output.as_view_mut());
875
876            for chunk in 0..num_chunks {
877                let range = offsets.at(chunk);
878                let query_chunk = &query[range.clone()];
879                for center in 0..num_centers {
880                    let data_chunk = &pivots.row(center)[range.clone()];
881                    let expected: f32 = distance::InnerProduct::evaluate(query_chunk, data_chunk);
882                    assert_eq!(
883                        *output.element(chunk, center),
884                        expected,
885                        "failed on (chunk, center) = ({}, {}) - offsets = {:?} - trial = {}",
886                        chunk,
887                        center,
888                        offsets,
889                        trial,
890                    );
891                }
892            }
893
894            // Squared L2
895            table.process_into::<SquaredL2>(&query, output.as_view_mut());
896
897            for chunk in 0..num_chunks {
898                let range = offsets.at(chunk);
899                let query_chunk = &query[range.clone()];
900                for center in 0..num_centers {
901                    let data_chunk = &pivots.row(center)[range.clone()];
902                    let expected: f32 = distance::SquaredL2::evaluate(query_chunk, data_chunk);
903                    assert_eq!(
904                        *output.element(chunk, center),
905                        expected,
906                        "failed on (chunk, center) = ({}, {}) - offsets = {:?} - trial = {}",
907                        chunk,
908                        center,
909                        offsets,
910                        trial,
911                    );
912                }
913            }
914        }
915    }
916
917    #[test]
918    fn test_process_into() {
919        let mut rng = StdRng::seed_from_u64(0x0e3cf3ba4b27e7f8);
920
921        let num_chunks_range = if cfg!(miri) { 4..5 } else { 1..5 };
922
923        let num_centers: Vec<usize> = if cfg!(miri) {
924            vec![1, 7, 16, 33, 47]
925        } else {
926            (1..48).collect()
927        };
928
929        for num_chunks in num_chunks_range {
930            for num_centers in num_centers.clone() {
931                test_process_into_impl(num_chunks, num_centers, 2, &mut rng);
932            }
933        }
934    }
935
936    #[test]
937    #[should_panic(expected = "query has the wrong number of dimensions")]
938    fn test_process_into_panics_query() {
939        let offsets = ChunkOffsets::new(Box::new([0, 1, 5])).unwrap();
940        let data = views::rowmajor::Owned::<f32>::from_element(3, 5, 0.0);
941        let table = TransposedTable::from_parts(data.as_view(), offsets).unwrap();
942        assert_eq!(table.dim(), 5);
943
944        // query has the wrong length.
945        let query = vec![0.0; table.dim() - 1];
946        let mut partials =
947            views::rowmajor::Owned::from_element(table.nchunks(), table.ncenters(), 0.0);
948        table.process_into::<InnerProduct>(&query, partials.as_view_mut());
949    }
950
951    #[test]
952    #[should_panic(expected = "output has the wrong number of rows")]
953    fn test_process_into_panics_partials_rows() {
954        let offsets = ChunkOffsets::new(Box::new([0, 1, 5])).unwrap();
955        let data = views::rowmajor::Owned::<f32>::from_element(3, 5, 0.0);
956        let table = TransposedTable::from_parts(data.as_view(), offsets).unwrap();
957        assert_eq!(table.dim(), 5);
958
959        let query = vec![0.0; table.dim()];
960        // partials has the wrong numbers of rows.
961        let mut partials =
962            views::rowmajor::Owned::from_element(table.nchunks() - 1, table.ncenters(), 0.0);
963        table.process_into::<InnerProduct>(&query, partials.as_view_mut());
964    }
965
966    #[test]
967    #[should_panic(expected = "output has the wrong number of columns")]
968    fn test_process_into_panics_partials_cols() {
969        let offsets = ChunkOffsets::new(Box::new([0, 1, 5])).unwrap();
970        let data = views::rowmajor::Owned::<f32>::from_element(3, 5, 0.0);
971        let table = TransposedTable::from_parts(data.as_view(), offsets).unwrap();
972        assert_eq!(table.dim(), 5);
973
974        let query = vec![0.0; table.dim()];
975        // partials has the wrong numbers of rows.
976        let mut partials =
977            views::rowmajor::Owned::from_element(table.nchunks(), table.ncenters() - 1, 0.0);
978        table.process_into::<InnerProduct>(&query, partials.as_view_mut());
979    }
980}