feat: add IVF-PQ L2 distance range search

Author: JunRuiLee

core/README.md

@@ -22,16 +22,28 @@
 `paimon-vindex-core` contains the Rust implementations and seek-based readers
 for IVF-FLAT, IVF-SQ, IVF-PQ, IVF-RQ, and DiskANN.
 
-The Rust reader supports distance range search for IVF-FLAT, IVF-RQ, and IVF-SQ with
+The Rust reader supports distance range search for IVF-FLAT, IVF-RQ, IVF-SQ, and IVF-PQ with
 squared L2, using `DistanceBand`, `VectorRangeSearchParams`, and CSR
-`RangeSearchResult` buffers. All three families support single and batch queries,
+`RangeSearchResult` buffers. All four families support single and batch queries,
 with or without a serialized Roaring allow-list, and a fixed positive `nprobe`.
 IVF-FLAT tests exact distances; IVF-RQ tests its one-bit or full multi-bit
-estimated distances; IVF-SQ tests scalar-quantized estimates. Results are uncapped
-and unordered. Probing every list removes the IVF coverage gap, but not the
-quantization error of IVF-RQ or IVF-SQ. The range path does not change top-K
+estimated distances; IVF-SQ tests scalar-quantized estimates; IVF-PQ tests
+floating-point ADC estimates for 4-bit and
+8-bit codes, with optional residual encoding and OPQ. Results are uncapped and
+unordered. Probing every list removes the IVF coverage gap, but not the
+compressed families' quantization error. The range path does not change top-K
 search or the v1 storage format.
 
+PQ range uses direct squared-L2 subvector lookup tables and sums their selected
+entries in subquantizer order. It does not use top-K's u8 FastScan tables or
+precomputed norm identities; membership is independent of list size, batch
+size, and `optimize_for_search`. Finite estimates can therefore differ from
+top-K's distances. Every filter-eligible row is fully evaluated, with no early
+abandonment. Non-finite consumed estimates, rotated queries, or coarse distances
+return `InvalidData`, including overflow and distances to unselected centroids.
+Unique non-empty lists are read once per call, with oversized lists streamed
+through the existing bounded reader. DiskANN range remains unsupported.
+
 See the [range search guide](../docs/range-search.html) for membership,
 validation, filtering, and statistics. C/JNI range bindings are not included.
 
@@ -61,7 +73,7 @@ Large lists stream in bounded chunks; scan scratch is reused, and a finite upper
 allows entire SQ blocks to stop after their partial distances reach that cut.
 Result memory still grows with the number of hits. Cache hits are excluded from
 `call_stats().list_reads()`. Range support does not extend to other metrics,
-IVF-PQ, DiskANN, or language bindings in this change.
+DiskANN, or language bindings.
 
 The crate ships its [normative v1 storage-format specification](STORAGE_FORMAT.md)
 and byte-exact fixtures. Project documentation, language bindings, and

core/src/collect.rs

@@ -64,7 +64,7 @@ pub(crate) trait Collector {
     fn cutoff(&self) -> f32;
 
     /// Delivers one row, with the value the family's scan computed for it. For
-    /// IVF-Flat that value is an exact distance; for IVF-RQ and IVF-SQ it is an
+    /// IVF-Flat that value is an exact distance; for IVF-RQ, IVF-SQ and IVF-PQ it is an
     /// estimate.
     ///
     /// Fallible because a collector may own a resource the scan cannot see: the

core/src/index.rs

@@ -1867,8 +1867,9 @@ impl<R: SeekRead> VectorIndexReader<R> {
 
     /// Distance range search. For the contract see
     /// [`IVFFlatIndexReader::range_search`] (exact distances),
-    /// [`IVFRQIndexReader::range_search`] (RQ estimates), and
-    /// [`IVFSQIndexReader::range_search`] (SQ estimates). Only L2 is supported.
+    /// [`IVFRQIndexReader::range_search`] (RQ estimates),
+    /// [`IVFSQIndexReader::range_search`] (SQ estimates), and
+    /// [`IVFPQIndexReader::range_search`] (PQ estimates). Only L2 is supported.
     ///
     /// The empty-band short-circuit lives **inside each family's reader**, so a
     /// family that cannot do range search at all still fails loud for every
@@ -1884,15 +1885,16 @@ impl<R: SeekRead> VectorIndexReader<R> {
             Self::IvfFlat(reader) => reader.range_search(query, params),
             Self::IvfRq(reader) => reader.range_search(query, params),
             Self::IvfSq(reader) => reader.range_search(query, params),
-            Self::IvfPq(_) => Err(range_unsupported("ivf_pq")),
+            Self::IvfPq(reader) => reader.range_search(query, params),
             Self::DiskAnn(_) => Err(range_unsupported("diskann")),
         }
     }
 
     /// Range search restricted to a serialized Roaring allow-list. For the
     /// contract see [`IVFFlatIndexReader::range_search_with_roaring_filter`],
-    /// [`IVFRQIndexReader::range_search`], and
-    /// [`IVFSQIndexReader::range_search_with_roaring_filter`].
+    /// [`IVFRQIndexReader::range_search`],
+    /// [`IVFSQIndexReader::range_search_with_roaring_filter`], and
+    /// [`IVFPQIndexReader::range_search_with_roaring_filter`].
     pub fn range_search_with_roaring_filter(
         &mut self,
         query: &[f32],
@@ -1909,7 +1911,7 @@ impl<R: SeekRead> VectorIndexReader<R> {
             Self::IvfFlat(reader) => reader.range_search_with_filter(query, params, Some(&filter)),
             Self::IvfRq(reader) => reader.range_search_with_filter(query, params, Some(&filter)),
             Self::IvfSq(reader) => reader.range_search_with_filter(query, params, Some(&filter)),
-            Self::IvfPq(_) => Err(range_unsupported("ivf_pq")),
+            Self::IvfPq(reader) => reader.range_search_with_filter(query, params, Some(&filter)),
             Self::DiskAnn(_) => Err(range_unsupported("diskann")),
         }
     }
@@ -1928,7 +1930,7 @@ impl<R: SeekRead> VectorIndexReader<R> {
             Self::IvfFlat(reader) => reader.range_search_batch(queries, query_count, params),
             Self::IvfRq(reader) => reader.range_search_batch(queries, query_count, params),
             Self::IvfSq(reader) => reader.range_search_batch(queries, query_count, params),
-            Self::IvfPq(_) => Err(range_unsupported("ivf_pq")),
+            Self::IvfPq(reader) => reader.range_search_batch(queries, query_count, params),
             Self::DiskAnn(_) => Err(range_unsupported("diskann")),
         }
     }
@@ -1954,7 +1956,9 @@ impl<R: SeekRead> VectorIndexReader<R> {
             Self::IvfSq(reader) => {
                 reader.range_search_batch_with_filter(queries, query_count, params, Some(&filter))
             }
-            Self::IvfPq(_) => Err(range_unsupported("ivf_pq")),
+            Self::IvfPq(reader) => {
+                reader.range_search_batch_with_filter(queries, query_count, params, Some(&filter))
+            }
             Self::DiskAnn(_) => Err(range_unsupported("diskann")),
         }
     }

core/src/io.rs

@@ -934,6 +934,17 @@ impl<R: SeekRead> IVFPQIndexReader<R> {
         &mut self,
         list_id: usize,
         mut consume: impl FnMut(&ProductQuantizer, &[i64], &[u8]),
+    ) -> io::Result<()> {
+        self.try_for_each_streamed_list_chunk(list_id, |pq, ids, codes| {
+            consume(pq, ids, codes);
+            Ok(())
+        })
+    }
+
+    pub(crate) fn try_for_each_streamed_list_chunk(
+        &mut self,
+        list_id: usize,
+        mut consume: impl FnMut(&ProductQuantizer, &[i64], &[u8]) -> io::Result<()>,
     ) -> io::Result<()> {
         self.ensure_loaded()?;
         let count = self.list_counts[list_id] as usize;
@@ -1004,7 +1015,7 @@ impl<R: SeekRead> IVFPQIndexReader<R> {
                     .pread(&mut [ReadRequest::new(chunk_offset, payload.codes_mut())])?;
             }
             let row_end = row_start + chunk_rows;
-            consume(&self.pq, &ids[row_start..row_end], payload.codes());
+            consume(&self.pq, &ids[row_start..row_end], payload.codes())?;
             row_start = row_end;
         }
         Ok(())