upstream-archive/qdrant
1
0
Fork 0
mirror of https://github.com/qdrant/qdrant.git synced 2026-07-23 11:11:00 -05:00
Code Issues Packages Projects Releases Wiki Activity

fix(turbo): make quantized vectors first-class to stop within-node TQ drift

Browse Source 
TurboQuant (TQDT) is a primary vector storage: it keeps only encoded bytes, no
original f32. Reading a vector back as float and re-inserting it into a TQ storage
re-quantizes (TQ -> float -> TQ) and the stored length drifts, degrading recall
over repeated round-trips. Two single-node sites did this: cross-segment
copy-on-write move and within-segment clone-and-mutate.

TQ encoding is fully deterministic (fixed permutation seeds, hardcoded TQDT_*
constants, no per-segment state), so raw encoded bytes are portable and decodable
without the storage. Carry those native bytes wherever a vector is moved/copied
unchanged; decode only where a float is genuinely needed. Dense TQ only.

Make a quantized vector a first-class vector type. A `Quantized` variant joins the
three in-memory vector types — `CowVector` (Cow read form), `VectorRef` (borrowed),
and `VectorInternal` (owned) — so a quantized vector flows through the normal
read/write API:

- The TQ storage returns `CowVector::Quantized` from `get_vector` (borrowed bytes,
  no copy) and has no dequantize method at all — decoding is a deterministic free
  function over the bytes, resolved from a process-wide (dim, distance) quantizer
  cache. The storage holds bytes, not floats.
- `to_owned` preserves the encoding, so `VectorInternal` is "maybe-encoded": every
  consumer that needs floats decodes it, compiler-forced at each match / From /
  TryFrom. Output boundaries (REST, gRPC, edge, pyo3) decode, so raw bytes never
  leak to a user.
- The write path routes a `VectorRef::Quantized` straight into the storage as bytes
  (no re-quantization), so a quantized vector written back unchanged is preserved
  verbatim. Re-quantization drift via read -> write is structurally impossible.

Fix #1 (cross-segment CoW, `SegmentHolder::apply_points_with_conditional_move`):
read the point once via `all_vectors`, run the move closure, `upsert_point`. No
snapshots — the move closures never read vectors as float, so an untouched
quantized vector is written back verbatim and an overwritten one re-quantized.
Fix #2 (within-segment `clone_and_mutate_point`): snapshot native bytes + a float
for the closure, write unchanged vectors verbatim.

`VectorStorageEnum` is untouched. The previously prototyped parallel segment API
(SerializedVector, get/upsert_serialized_vectors, read_native_vectors) and the
speculative VectorEncoding tag are not part of this design.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Ivan Pleshkov
2026-06-26 02:02:42 +02:00
parent dc74d66288
commit 803e0b7a3f
24 changed files with 998 additions and 88 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
lib/api/src/conversions/vectors.rs
View File

@@ -102,6 +102,8 @@ impl TryFrom<rest::VectorStructOutput> for grpc::VectorsOutput {
impl From<VectorInternal> for grpc::VectorOutput {
fn from(vector: VectorInternal) -> Self {
let vector = match vector {
// Output boundary: decode quantized bytes to floats for the user.
VectorInternal::Quantized(q) => return Self::from(q.dequantize()),
VectorInternal::Dense(vector) => {
grpc::vector_output::Vector::Dense(grpc::DenseVector { data: vector })
}
@@ -419,6 +421,8 @@ impl TryFrom<grpc::VectorsOutput> for VectorStructInternal {
impl From<VectorInternal> for grpc::Vector {
fn from(vector: VectorInternal) -> Self {
let vector = match vector {
// Output boundary: decode quantized bytes to floats for the user.
VectorInternal::Quantized(q) => return Self::from(q.dequantize()),
VectorInternal::Dense(vector) => {
grpc::vector::Vector::Dense(grpc::DenseVector { data: vector })
}
@@ -608,6 +612,8 @@ impl From<VectorInternal> for grpc::RawVector {
use crate::grpc::qdrant::raw_vector::Variant;
let variant = match value {
// Output boundary: decode quantized bytes to floats for the user.
VectorInternal::Quantized(q) => return Self::from(q.dequantize()),
VectorInternal::Dense(vector) => Variant::Dense(grpc::DenseVector::from(vector)),
VectorInternal::Sparse(vector) => Variant::Sparse(grpc::SparseVector::from(vector)),
VectorInternal::MultiDense(vector) => {

4
lib/api/src/grpc/conversions.rs
View File

@@ -2711,6 +2711,10 @@ pub fn into_named_vector_struct(
use sparse::common::sparse_vector::SparseVector;
Ok(match vector_internal {
// Decode quantized bytes to floats before building the named struct.
VectorInternal::Quantized(q) => {
return into_named_vector_struct(vector_name, q.dequantize());
}
VectorInternal::Dense(vector) => {
if let Some(name) = vector_name {
NamedVectorStruct::Dense(NamedVector { name, vector })

2
lib/api/src/rest/conversions.rs
View File

@@ -37,6 +37,8 @@ impl From<VectorInternal> for VectorOutput {
VectorInternal::MultiDense(vector) => {
VectorOutput::MultiDense(vector.into_multi_vectors())
}
// Output boundary: a quantized vector is decoded to floats for the user.
VectorInternal::Quantized(q) => VectorOutput::from(q.dequantize()),
}
}
}

9
lib/collection/src/operations/conversions.rs
View File

@@ -1204,6 +1204,15 @@ impl TryFrom<api::grpc::qdrant::Vector> for RecommendExample {
VectorInternal::MultiDense(_vector) => Err(Status::invalid_argument(
"MultiDense vector is not supported in search request",
)),
// User input is never quantized; decode defensively if it ever is.
VectorInternal::Quantized(q) => match q.dequantize() {
VectorInternal::Dense(vector) => Ok(Self::Dense(vector)),
VectorInternal::Sparse(_)
| VectorInternal::MultiDense(_)
| VectorInternal::Quantized(_) => {
Err(Status::invalid_argument("unexpected quantized vector"))
}
},
}
}
}

10
lib/collection/src/operations/generalizer/query.rs
View File

@@ -153,6 +153,16 @@ impl Generalizer for VectorInternal {
2,
))
}
// Strip the bytes; keep just the dimensionality marker, matching the
// shape the dequantized form would generalize to (a multivector keeps
// the 2-element `[num_vectors, dim]` marker, a single stays 1-element).
VectorInternal::Quantized(q) => match q.multivector_count {
None => VectorInternal::Dense(vec![q.dim as f32]),
Some(count) => VectorInternal::MultiDense(MultiDenseVectorInternal::new(
vec![count as f32, q.dim as f32],
2,
)),
},
}
}
}

34
lib/collection/src/recommendations.rs
View File

@@ -68,6 +68,40 @@ fn avg_vectors<'a>(
},
));
}
VectorRef::Quantized(q) => {
// Encoded-primary (TurboQuant) storages hand out native bytes;
// decode and fold into the running average exactly like a plain
// dense or multidense input of the same kind.
match q.dequantize() {
VectorInternal::Dense(vector) => {
dense_count += 1;
for i in 0..vector.len() {
if i >= avg_dense.len() {
avg_dense.push(vector[i])
} else {
avg_dense[i] += vector[i];
}
}
}
VectorInternal::MultiDense(vector) => {
multi_count += 1;
avg_multi = Some(avg_multi.map_or_else(
|| vector.clone(),
|mut avg_multi| {
avg_multi
.flattened_vectors
.extend_from_slice(&vector.flattened_vectors);
avg_multi
},
));
}
VectorInternal::Sparse(_) | VectorInternal::Quantized(_) => {
return Err(CollectionError::service_error(
"TurboQuant vector dequantized to an unexpected form",
));
}
}
}
}
}

5
lib/edge/python/src/types/vector_internal.rs
View File

@@ -150,6 +150,10 @@ impl<'py> IntoPyObject<'py> for &PyNamedVectorInternal {
VectorInternal::Dense(dense) => dense.into_bound_py_any(py),
VectorInternal::Sparse(sparse) => PySparseVector(sparse.clone()).into_bound_py_any(py),
VectorInternal::MultiDense(multi) => multi_dense_into_py(multi, py),
// Output boundary: decode quantized bytes to floats for Python.
VectorInternal::Quantized(q) => {
(&PyNamedVectorInternal(q.dequantize())).into_pyobject(py)
}
}
}
}
@@ -160,6 +164,7 @@ impl Repr for PyNamedVectorInternal {
VectorInternal::Dense(dense) => dense.fmt(f),
VectorInternal::Sparse(sparse) => PySparseVector::wrap_ref(sparse).fmt(f),
VectorInternal::MultiDense(multi) => f.list(multi.multi_vectors()),
VectorInternal::Quantized(q) => Repr::fmt(&PyNamedVectorInternal(q.dequantize()), f),
}
}
}

2
lib/edge/src/types/vector.rs
View File

@@ -173,6 +173,8 @@ impl From<Vector> for Vectors {
VectorInternal::Sparse(v) => Self(VectorStructInternal::Named(
[(String::new(), VectorInternal::Sparse(v))].into(),
)),
// Output boundary: decode quantized bytes to floats for the user.
VectorInternal::Quantized(q) => Self::from(Vector(q.dequantize())),
}
}
}

23
lib/segment/src/common/mod.rs
View File

@@ -18,7 +18,9 @@ use std::sync::atomic::AtomicBool;
use crate::common::operation_error::{OperationError, OperationResult};
use crate::data_types::named_vectors::NamedVectors;
use crate::data_types::vectors::{QueryVector, VectorRef};
use crate::types::{SegmentConfig, SparseVectorDataConfig, VectorDataConfig, VectorName};
use crate::types::{
SegmentConfig, SparseVectorDataConfig, VectorDataConfig, VectorName, VectorStorageDatatype,
};
pub type Flusher = Box<dyn FnOnce() -> OperationResult<()> + Send>;
@@ -216,6 +218,24 @@ fn check_vector_against_config(
}
Ok(())
}
VectorRef::Quantized(quantized) => {
if vector_config.datatype != Some(VectorStorageDatatype::Turbo4) {
return Err(OperationError::service_error(format!(
"Incompatible vector datatype: expected {:?}, received {:?}",
vector_config.datatype,
Some(VectorStorageDatatype::Turbo4)
)));
}
let dim = vector_config.size;
if quantized.dim != dim {
return Err(OperationError::WrongVectorDimension {
expected_dim: dim,
received_dim: quantized.dim,
});
}
Ok(())
}
}
}
@@ -227,6 +247,7 @@ fn check_sparse_vector_against_config(
VectorRef::Dense(_) => Err(OperationError::WrongSparse),
VectorRef::Sparse(_vector) => Ok(()), // TODO(sparse) check vector by config
VectorRef::MultiDense(_) => Err(OperationError::WrongMulti),
VectorRef::Quantized(_) => Err(OperationError::WrongSparse),
}
}

132
lib/segment/src/data_types/named_vectors.rs
View File

@@ -10,7 +10,7 @@ use super::vectors::{
VectorElementType, VectorElementTypeByte, VectorElementTypeHalf, VectorInternal, VectorRef,
};
use crate::common::operation_error::OperationError;
use crate::types::{VectorDataConfig, VectorName, VectorNameBuf, VectorStorageDatatype};
use crate::types::{Distance, VectorDataConfig, VectorName, VectorNameBuf, VectorStorageDatatype};
type CowKey<'a> = Cow<'a, VectorName>;
@@ -46,11 +46,74 @@ where
}
}
/// A storage-native quantized vector carried as opaque encoded bytes plus the
/// local context needed to decode it (TurboQuant is fully deterministic given
/// `dim` + `distance`). This is the in-memory representation that lets a
/// quantized vector travel through `NamedVectors` / CoW / clone-and-mutate
/// without a dequantize -> re-quantize round-trip; only the TQ storage's write
/// path consumes the bytes verbatim, every other consumer dequantizes on demand.
///
/// `dim` / `distance` are local context (from the owning or target storage).
/// Equality is by all fields, so comparing two `Quantized` is a byte compare —
/// exactly the "unchanged" check a copy-on-write needs.
///
/// Handles both dense single vectors and multivectors. `multivector_count`
/// distinguishes them: `None` is a dense single (`dim` floats); `Some(n)` is a
/// multivector of `n` inner vectors, the bytes being `n` concatenated inner
/// encodings of `dim` floats each. `dim` is always the inner-vector dimension.
///
/// There is no encoding-version field: within a single node every blob is
/// produced and consumed by the same process with the same TQDT constants, so a
/// version would always equal itself. Versioning only matters when bytes cross a
/// node/build boundary, where it rides the wire type separately.
#[derive(Clone, PartialEq, Debug, serde::Serialize)]
pub struct CowQuantizedVector<'a> {
pub bytes: Cow<'a, [u8]>,
pub dim: usize,
pub distance: Distance,
/// `None` for a dense single vector; `Some(inner_count)` for a multivector.
pub multivector_count: Option<usize>,
}
impl CowQuantizedVector<'_> {
/// Decode back to a plain float vector: dense for a single, multi for a
/// multivector (TurboQuant).
pub fn dequantize(&self) -> VectorInternal {
match self.multivector_count {
None => VectorInternal::Dense(crate::vector_storage::turbo::dequantize_tqdt_dense(
&self.bytes,
self.dim,
self.distance,
)),
Some(count) => {
VectorInternal::MultiDense(crate::vector_storage::turbo::dequantize_tqdt_multi(
&self.bytes,
self.dim,
self.distance,
count,
))
}
}
}
pub fn into_owned(self) -> CowQuantizedVector<'static> {
CowQuantizedVector {
bytes: Cow::Owned(self.bytes.into_owned()),
dim: self.dim,
distance: self.distance,
multivector_count: self.multivector_count,
}
}
}
#[derive(Clone, PartialEq, Debug)]
pub enum CowVector<'a> {
Dense(Cow<'a, [VectorElementType]>),
Sparse(Cow<'a, SparseVector>),
MultiDense(CowMultiVector<'a, VectorElementType>),
/// Storage-native quantized bytes (e.g. TurboQuant), decoded on demand by
/// any float consumer; preserved verbatim only by a matching TQ storage.
Quantized(CowQuantizedVector<'a>),
}
impl Default for CowVector<'_> {
@@ -67,6 +130,7 @@ impl CowVector<'_> {
CowVector::MultiDense(cow_multi_vector) => {
cow_multi_vector.flattened_len() * size_of::<VectorElementType>()
}
CowVector::Quantized(q) => q.bytes.len(),
}
}
}
@@ -102,11 +166,29 @@ impl CowVector<'_> {
CowVector::Sparse(Cow::Owned(SparseVector::default()))
}
/// Own the vector **preserving its native encoding** (a `Quantized` becomes
/// [`VectorInternal::Quantized`], not a dequantized float). `VectorInternal`
/// is "maybe-encoded": consumers that need floats decode it explicitly (the
/// compiler forces them at each match / `From`/`TryFrom`).
pub fn to_owned(self) -> VectorInternal {
match self {
CowVector::Dense(v) => VectorInternal::Dense(v.into_owned()),
CowVector::Sparse(v) => VectorInternal::Sparse(v.into_owned()),
CowVector::MultiDense(v) => VectorInternal::MultiDense(v.to_owned()),
CowVector::Quantized(q) => VectorInternal::Quantized(q.into_owned()),
}
}
/// Own any borrowed data **without dequantizing**: a `Quantized` stays
/// `Quantized` (with owned bytes), unlike [`Self::to_owned`] which decodes to
/// a float `VectorInternal`. Used to carry a vector across a storage borrow
/// while preserving its native encoding.
pub fn into_owned(self) -> CowVector<'static> {
match self {
CowVector::Dense(v) => CowVector::Dense(Cow::Owned(v.into_owned())),
CowVector::Sparse(v) => CowVector::Sparse(Cow::Owned(v.into_owned())),
CowVector::MultiDense(v) => CowVector::MultiDense(CowMultiVector::Owned(v.to_owned())),
CowVector::Quantized(q) => CowVector::Quantized(q.into_owned()),
}
}
@@ -115,6 +197,7 @@ impl CowVector<'_> {
CowVector::Dense(v) => VectorRef::Dense(v.as_ref()),
CowVector::Sparse(v) => VectorRef::Sparse(v.as_ref()),
CowVector::MultiDense(v) => VectorRef::MultiDense(v.as_vec_ref()),
CowVector::Quantized(q) => VectorRef::Quantized(q),
}
}
}
@@ -134,6 +217,7 @@ impl From<VectorInternal> for CowVector<'_> {
VectorInternal::Dense(v) => CowVector::Dense(Cow::Owned(v)),
VectorInternal::Sparse(v) => CowVector::Sparse(Cow::Owned(v)),
VectorInternal::MultiDense(v) => CowVector::MultiDense(CowMultiVector::Owned(v)),
VectorInternal::Quantized(q) => CowVector::Quantized(q),
}
}
}
@@ -193,6 +277,7 @@ impl<'a> TryFrom<CowVector<'a>> for SparseVector {
CowVector::Dense(_) => Err(OperationError::WrongSparse),
CowVector::Sparse(v) => Ok(v.into_owned()),
CowVector::MultiDense(_) => Err(OperationError::WrongSparse),
CowVector::Quantized(_) => Err(OperationError::WrongSparse),
}
}
}
@@ -205,6 +290,12 @@ impl<'a> TryFrom<CowVector<'a>> for DenseVector {
CowVector::Dense(v) => Ok(v.into_owned()),
CowVector::Sparse(_) => Err(OperationError::WrongSparse),
CowVector::MultiDense(_) => Err(OperationError::WrongMulti),
CowVector::Quantized(q) => match q.dequantize() {
VectorInternal::Dense(v) => Ok(v),
VectorInternal::Sparse(_)
| VectorInternal::MultiDense(_)
| VectorInternal::Quantized(_) => Err(OperationError::WrongMulti),
},
}
}
}
@@ -217,6 +308,12 @@ impl<'a> TryFrom<CowVector<'a>> for Cow<'a, [VectorElementType]> {
CowVector::Dense(v) => Ok(v),
CowVector::Sparse(_) => Err(OperationError::WrongSparse),
CowVector::MultiDense(_) => Err(OperationError::WrongMulti),
CowVector::Quantized(q) => match q.dequantize() {
VectorInternal::Dense(v) => Ok(Cow::Owned(v)),
VectorInternal::Sparse(_)
| VectorInternal::MultiDense(_)
| VectorInternal::Quantized(_) => Err(OperationError::WrongMulti),
},
}
}
}
@@ -227,6 +324,7 @@ impl<'a> From<VectorRef<'a>> for CowVector<'a> {
VectorRef::Dense(v) => CowVector::Dense(Cow::Borrowed(v)),
VectorRef::Sparse(v) => CowVector::Sparse(Cow::Borrowed(v)),
VectorRef::MultiDense(v) => CowVector::MultiDense(CowMultiVector::Borrowed(v)),
VectorRef::Quantized(q) => CowVector::Quantized(q.clone()),
}
}
}
@@ -258,6 +356,13 @@ impl<'a> NamedVectors<'a> {
.insert(CowKey::Owned(name), CowVector::from(vector));
}
/// Insert an already-built [`CowVector`], preserving a
/// [`CowVector::Quantized`] (unlike [`Self::insert`], which takes an owned
/// float `VectorInternal`). Used by the native read path.
pub fn insert_cow(&mut self, name: VectorNameBuf, vector: CowVector<'a>) {
self.map.insert(CowKey::Owned(name), vector);
}
pub fn remove_ref(&mut self, key: &VectorName) {
self.map.remove(key);
}
@@ -278,6 +383,13 @@ impl<'a> NamedVectors<'a> {
self.map.iter().map(|(k, _)| k.as_ref())
}
/// Iterate `(name, &CowVector)` without borrowing a float view, so it is
/// safe in the presence of [`CowVector::Quantized`] (unlike [`Self::iter`],
/// which calls `as_vec_ref`).
pub fn iter_cow(&self) -> impl Iterator<Item = (&VectorName, &CowVector<'a>)> {
self.map.iter().map(|(k, v)| (k.as_ref(), v))
}
pub fn into_owned_map(self) -> HashMap<VectorNameBuf, VectorInternal> {
self.map
.into_iter()
@@ -285,12 +397,13 @@ impl<'a> NamedVectors<'a> {
.collect()
}
/// Materialise into a fully-owned `NamedVectors<'static>` by cloning
/// any borrowed keys/values into owned ones.
/// Materialise into a fully-owned `NamedVectors<'static>` by cloning any
/// borrowed keys/values into owned ones, **preserving native encoding** (a
/// `Quantized` vector is not dequantized — see [`CowVector::into_owned`]).
pub fn into_owned(self) -> NamedVectors<'static> {
let mut out = NamedVectors::default();
for (name, vector) in self {
out.insert(name.into_owned(), vector.to_owned());
out.insert_cow(name.into_owned(), vector.into_owned());
}
out
}
@@ -303,6 +416,13 @@ impl<'a> NamedVectors<'a> {
self.map.get(key).map(|v| v.as_vec_ref())
}
/// Like [`Self::get`] but returns the owning [`CowVector`], which (unlike a
/// borrowed `VectorRef`) can be a [`CowVector::Quantized`]. The write path
/// uses this so it can route quantized vectors without dequantizing.
pub fn get_cow(&self, key: &VectorName) -> Option<&CowVector<'_>> {
self.map.get(key)
}
pub fn preprocess<'b>(
&mut self,
get_vector_data: impl Fn(&VectorName) -> &'b VectorDataConfig,
@@ -338,6 +458,10 @@ impl<'a> NamedVectors<'a> {
}
*multi_vector = CowMultiVector::Owned(owned_multi_vector);
}
// Preprocess chokepoint: a quantized vector is already final encoded
// bytes (the original float was normalized/cast before quantizing),
// so normalization/datatype casting must NOT touch it. Pass through.
CowVector::Quantized(_) => {}
}
}
}

61
lib/segment/src/data_types/vectors.rs
View File

@@ -30,6 +30,12 @@ pub enum VectorInternal {
Dense(DenseVector),
Sparse(SparseVector),
MultiDense(MultiDenseVectorInternal),
/// A storage-native quantized vector (e.g. TurboQuant bytes), carried owned
/// without dequantizing. `VectorInternal` is therefore "maybe-encoded": every
/// consumer that needs floats decodes it (compiler-enforced at each match /
/// `From`/`TryFrom`); only a matching TQ storage write consumes the bytes
/// verbatim, which is what makes a move/transfer lossless.
Quantized(super::named_vectors::CowQuantizedVector<'static>),
}
impl Hash for VectorInternal {
@@ -51,6 +57,9 @@ impl Hash for VectorInternal {
VectorInternal::MultiDense(v) => {
v.hash(state);
}
VectorInternal::Quantized(q) => {
q.bytes.hash(state);
}
}
}
}
@@ -62,6 +71,8 @@ impl VectorInternal {
VectorInternal::Dense(_dense) => 1,
VectorInternal::Sparse(sparse) => sparse.indices.len().div_ceil(SPARSE_DIMS_COST_UNIT),
VectorInternal::MultiDense(multivec) => multivec.vectors_count(),
// A quantized vector is dense — one comparison, like `Dense`.
VectorInternal::Quantized(_) => 1,
}
}
@@ -77,6 +88,8 @@ impl VectorInternal {
}
}
VectorInternal::MultiDense(_) => {}
// Already-final encoded bytes; nothing to preprocess.
VectorInternal::Quantized(_) => {}
}
}
@@ -97,6 +110,10 @@ pub enum VectorRef<'a> {
Dense(&'a [VectorElementType]),
Sparse(&'a SparseVector),
MultiDense(TypedMultiDenseVectorRef<'a, VectorElementType>),
/// A borrowed storage-native quantized vector (e.g. TurboQuant bytes). It has
/// no float view, so float consumers must dequantize (`try_into`); only a
/// matching TQ storage's `insert_vector` consumes the bytes verbatim.
Quantized(&'a super::named_vectors::CowQuantizedVector<'a>),
}
impl<'a> TryFrom<VectorRef<'a>> for &'a [VectorElementType] {
@@ -107,6 +124,11 @@ impl<'a> TryFrom<VectorRef<'a>> for &'a [VectorElementType] {
VectorRef::Dense(v) => Ok(v),
VectorRef::Sparse(_) => Err(OperationError::WrongSparse),
VectorRef::MultiDense(_) => Err(OperationError::WrongMulti),
// A quantized vector has no borrowed float view — it must be owned
// (dequantized) first, or consumed verbatim by a matching TQ storage.
VectorRef::Quantized(_) => Err(OperationError::service_error(
"cannot borrow a quantized vector as a float slice; dequantize it first",
)),
}
}
}
@@ -119,6 +141,7 @@ impl<'a> TryFrom<VectorRef<'a>> for &'a SparseVector {
VectorRef::Dense(_) => Err(OperationError::WrongSparse),
VectorRef::Sparse(v) => Ok(v),
VectorRef::MultiDense(_) => Err(OperationError::WrongMulti),
VectorRef::Quantized(_) => Err(OperationError::WrongSparse),
}
}
}
@@ -134,6 +157,7 @@ impl<'a> TryFrom<VectorRef<'a>> for TypedMultiDenseVectorRef<'a, f32> {
}),
VectorRef::Sparse(_v) => Err(OperationError::WrongSparse),
VectorRef::MultiDense(v) => Ok(v),
VectorRef::Quantized(_) => Err(OperationError::WrongMulti),
}
}
}
@@ -157,6 +181,8 @@ impl TryFrom<VectorInternal> for DenseVector {
VectorInternal::Dense(v) => Ok(v),
VectorInternal::Sparse(_) => Err(OperationError::WrongSparse),
VectorInternal::MultiDense(_) => Err(OperationError::WrongMulti),
// Decode the quantized bytes, then take the dense result.
VectorInternal::Quantized(q) => Self::try_from(q.dequantize()),
}
}
}
@@ -169,6 +195,7 @@ impl TryFrom<VectorInternal> for SparseVector {
VectorInternal::Dense(_) => Err(OperationError::WrongSparse),
VectorInternal::Sparse(v) => Ok(v),
VectorInternal::MultiDense(_) => Err(OperationError::WrongMulti),
VectorInternal::Quantized(_) => Err(OperationError::WrongSparse),
}
}
}
@@ -185,6 +212,7 @@ impl TryFrom<VectorInternal> for MultiDenseVectorInternal {
}
VectorInternal::Sparse(_) => Err(OperationError::WrongSparse),
VectorInternal::MultiDense(v) => Ok(v),
VectorInternal::Quantized(q) => Self::try_from(q.dequantize()),
}
}
}
@@ -245,6 +273,7 @@ impl<'a> From<&'a VectorInternal> for VectorRef<'a> {
VectorInternal::MultiDense(v) => {
VectorRef::MultiDense(TypedMultiDenseVectorRef::from(v))
}
VectorInternal::Quantized(q) => VectorRef::Quantized(q),
}
}
}
@@ -482,6 +511,8 @@ impl VectorRef<'_> {
VectorRef::Dense(v) => VectorInternal::Dense(v.to_vec()),
VectorRef::Sparse(v) => VectorInternal::Sparse(v.clone()),
VectorRef::MultiDense(v) => VectorInternal::MultiDense(v.to_owned()),
// Preserve the native encoding (see `CowVector::to_owned`).
VectorRef::Quantized(q) => VectorInternal::Quantized(q.clone().into_owned()),
}
}
}
@@ -494,6 +525,10 @@ impl<'a> TryInto<&'a [VectorElementType]> for &'a VectorInternal {
VectorInternal::Dense(v) => Ok(v),
VectorInternal::Sparse(_) => Err(OperationError::WrongSparse),
VectorInternal::MultiDense(_) => Err(OperationError::WrongMulti),
// No borrowed float view — it must be owned (dequantized) first.
VectorInternal::Quantized(_) => Err(OperationError::service_error(
"cannot borrow a quantized vector as a float slice; dequantize it first",
)),
}
}
}
@@ -506,6 +541,7 @@ impl<'a> TryInto<&'a SparseVector> for &'a VectorInternal {
VectorInternal::Dense(_) => Err(OperationError::WrongSparse),
VectorInternal::Sparse(v) => Ok(v),
VectorInternal::MultiDense(_) => Err(OperationError::WrongMulti),
VectorInternal::Quantized(_) => Err(OperationError::WrongSparse),
}
}
}
@@ -518,6 +554,7 @@ impl<'a> TryInto<&'a MultiDenseVectorInternal> for &'a VectorInternal {
VectorInternal::Dense(_) => Err(OperationError::WrongMulti), // &Dense vector cannot be converted to &MultiDense
VectorInternal::Sparse(_) => Err(OperationError::WrongSparse),
VectorInternal::MultiDense(v) => Ok(v),
VectorInternal::Quantized(_) => Err(OperationError::WrongMulti),
}
}
}
@@ -572,6 +609,18 @@ impl From<NamedVectors<'_>> for VectorStructInternal {
let vector_ref = v.get(DEFAULT_VECTOR_NAME).unwrap();
match vector_ref {
VectorRef::Quantized(q) => match q.dequantize() {
VectorInternal::Dense(v) => VectorStructInternal::Single(v),
// A quantized multivector must produce the same `MultiDense`
// shape as its plain equivalent (the `VectorRef::MultiDense`
// arm below), not a `Named` wrapper.
VectorInternal::MultiDense(v) => VectorStructInternal::MultiDense(v),
other @ (VectorInternal::Sparse(_) | VectorInternal::Quantized(_)) => {
let mut map = HashMap::new();
map.insert(DEFAULT_VECTOR_NAME.to_owned(), other);
VectorStructInternal::Named(map)
}
},
VectorRef::Dense(v) => VectorStructInternal::Single(v.to_owned()),
VectorRef::Sparse(v) => {
debug_assert!(false, "Sparse vector cannot be default");
@@ -606,6 +655,18 @@ impl From<NamedVectorsOwned> for VectorStructInternal {
VectorStructInternal::Named(map)
}
VectorInternal::MultiDense(v) => VectorStructInternal::MultiDense(v),
// Output boundary: decode the quantized bytes to a float vector.
VectorInternal::Quantized(q) => match q.dequantize() {
VectorInternal::Dense(v) => VectorStructInternal::Single(v),
// Keep a quantized multivector aligned with the plain
// `MultiDense` arm above instead of wrapping it in `Named`.
VectorInternal::MultiDense(v) => VectorStructInternal::MultiDense(v),
other @ (VectorInternal::Sparse(_) | VectorInternal::Quantized(_)) => {
let mut map = HashMap::new();
map.insert(DEFAULT_VECTOR_NAME.to_owned(), other);
VectorStructInternal::Named(map)
}
},
}
} else {
VectorStructInternal::Named(v.into_iter().collect())

4
lib/segment/src/index/sparse_index/sparse_vector_index/read_view/search.rs
View File

@@ -48,7 +48,9 @@ where
let search_results = if query_context.is_require_idf() {
let vector = (*vector).clone().transform(|mut vector| {
match &mut vector {
VectorInternal::Dense(_) | VectorInternal::MultiDense(_) => {
VectorInternal::Dense(_)
| VectorInternal::MultiDense(_)
| VectorInternal::Quantized(_) => {
return Err(OperationError::WrongSparse);
}
VectorInternal::Sparse(sparse) => {

5
lib/segment/src/segment/segment_ops.rs
View File

@@ -174,6 +174,11 @@ impl Segment {
// `update_vector(_, None, _)` insert — including those would
// promote phantom data into the fresh slot, so we skip them
// and write `None` at new_id (re-tombstoning the slot).
// For an encoded-primary storage (e.g. TurboQuant) `get_vector_opt`
// yields a `CowVector::Quantized`; `to_owned` preserves the encoding as
// `VectorInternal::Quantized`, so the native bytes ride through the
// `mutate` closure and back into storage untouched unless the closure
// actually replaces the vector — no lossy TQ -> float -> TQ round-trip.
let mut vectors: NamedVectors<'static> = NamedVectors::default();
for (vector_name, vector_data) in self.vector_data.iter() {
let storage = vector_data.vector_storage.borrow();

222
lib/segment/src/segment/tests/mod.rs
View File

@@ -1712,3 +1712,225 @@ fn test_flush_survives_concurrent_field_index_drop() {
// version at 0.
assert_eq!(segment.persistent_version(), 3);
}
/// Regression test for the TurboQuant round-trip drift bug on within-segment
/// copy-on-write. On an append-only segment a payload mutation clones the point
/// to a fresh slot; the vector must be copied as native encoded bytes, not
/// dequantized and re-quantized, so its encoding never drifts across repeated
/// payload writes.
#[test]
fn test_append_only_cow_preserves_turbo_encoding() {
use common::generic_consts::Random;
use crate::data_types::named_vectors::CowVector;
use crate::types::VectorStorageDatatype;
use crate::vector_storage::VectorStorageRead;
init_logger();
let dim = 128;
let dir = Builder::new()
.prefix("turbo_cow_segment")
.tempdir()
.unwrap();
let (mut segment, _) = build_segment(
dir.path(),
&SegmentConfig {
vector_data: HashMap::from([(
DEFAULT_VECTOR_NAME.to_owned(),
VectorDataConfig {
size: dim,
distance: Distance::Dot,
// Appendable storage so clone-and-mutate can grow it.
storage_type: VectorStorageType::ChunkedMmap,
index: Indexes::Plain {},
quantization_config: None,
multivector_config: None,
datatype: Some(VectorStorageDatatype::Turbo4),
},
)]),
sparse_vector_data: HashMap::new(),
payload_storage_type: Default::default(),
},
None,
true,
)
.unwrap();
// Force the append-only clone-and-tombstone mutation path.
segment.append_only_mutations = true;
assert!(segment.is_append_only());
let hw_counter = HardwareCounterCell::new();
let point = PointIdType::from(0u64);
let mut rng = StdRng::seed_from_u64(0x7142);
let dense: Vec<f32> = {
let v: Vec<f32> = (0..dim).map(|_| rng.random_range(-1.0..1.0)).collect();
let norm = v.iter().map(|&x| x * x).sum::<f32>().sqrt();
v.iter().map(|&x| x / norm).collect()
};
segment
.upsert_point(1, point, only_default_vector(&dense), &hw_counter)
.unwrap();
// Helper: read the encoded bytes of the point's current (deferred) head.
let read_encoded = |segment: &Segment| -> Vec<u8> {
let internal = segment
.id_tracker
.borrow()
.internal_id_with_behavior(point, DeferredBehavior::WithDeferred)
.expect("point must exist");
let vector_data = segment.vector_data.get(DEFAULT_VECTOR_NAME).unwrap();
let storage = vector_data.vector_storage.borrow();
match storage
.get_vector_opt::<Random>(internal)
.expect("turbo storage must yield a vector")
{
CowVector::Quantized(quantized) => quantized.bytes.into_owned(),
other @ (CowVector::Dense(_) | CowVector::Sparse(_) | CowVector::MultiDense(_)) => {
panic!("expected quantized vector, got {other:?}")
}
}
};
let original_encoding = read_encoded(&segment);
// Each payload write clones the point on an append-only segment. The vector
// is untouched, so its encoding must stay byte-identical every time.
for op_num in 2..=20 {
let mut payload = Payload::default();
payload
.0
.insert("iter".to_string(), Value::Number(Number::from(op_num)));
segment
.set_full_payload(op_num, point, &payload, &hw_counter)
.unwrap();
let encoding = read_encoded(&segment);
assert_eq!(
encoding, original_encoding,
"turbo encoding drifted after {op_num} copy-on-write payload writes",
);
}
}
/// Same regression as [`test_append_only_cow_preserves_turbo_encoding`] but for a
/// TurboQuant *multivector* storage: a payload mutation clones the point and must
/// re-store the inner encoded records verbatim, never dequantize-and-re-quantize.
#[test]
fn test_append_only_cow_preserves_turbo_multi_encoding() {
use common::generic_consts::Random;
use crate::data_types::named_vectors::CowVector;
use crate::data_types::vectors::MultiDenseVectorInternal;
use crate::types::{MultiVectorComparator, MultiVectorConfig, VectorStorageDatatype};
use crate::vector_storage::VectorStorageRead;
init_logger();
let dim = 64;
let inner_count = 5;
let dir = Builder::new()
.prefix("turbo_multi_cow_segment")
.tempdir()
.unwrap();
let (mut segment, _) = build_segment(
dir.path(),
&SegmentConfig {
vector_data: HashMap::from([(
DEFAULT_VECTOR_NAME.to_owned(),
VectorDataConfig {
size: dim,
distance: Distance::Dot,
storage_type: VectorStorageType::ChunkedMmap,
index: Indexes::Plain {},
quantization_config: None,
multivector_config: Some(MultiVectorConfig {
comparator: MultiVectorComparator::MaxSim,
}),
datatype: Some(VectorStorageDatatype::Turbo4),
},
)]),
sparse_vector_data: HashMap::new(),
payload_storage_type: Default::default(),
},
None,
true,
)
.unwrap();
segment.append_only_mutations = true;
assert!(segment.is_append_only());
let hw_counter = HardwareCounterCell::new();
let point = PointIdType::from(0u64);
let mut rng = StdRng::seed_from_u64(0x51A6);
let flattened: Vec<f32> = {
let v: Vec<f32> = (0..dim * inner_count)
.map(|_| rng.random_range(-1.0..1.0))
.collect();
// Normalize each inner vector independently.
v.chunks_exact(dim)
.flat_map(|inner| {
let norm = inner.iter().map(|&x| x * x).sum::<f32>().sqrt();
inner.iter().map(move |&x| x / norm)
})
.collect()
};
let multi = MultiDenseVectorInternal::new(flattened, dim);
segment
.upsert_point(
1,
point,
NamedVectors::from_ref(DEFAULT_VECTOR_NAME, VectorRef::from(&multi)),
&hw_counter,
)
.unwrap();
let read_encoded = |segment: &Segment| -> (Vec<u8>, Option<usize>) {
let internal = segment
.id_tracker
.borrow()
.internal_id_with_behavior(point, DeferredBehavior::WithDeferred)
.expect("point must exist");
let vector_data = segment.vector_data.get(DEFAULT_VECTOR_NAME).unwrap();
let storage = vector_data.vector_storage.borrow();
match storage
.get_vector_opt::<Random>(internal)
.expect("turbo storage must yield a vector")
{
CowVector::Quantized(quantized) => {
(quantized.bytes.into_owned(), quantized.multivector_count)
}
other @ (CowVector::Dense(_) | CowVector::Sparse(_) | CowVector::MultiDense(_)) => {
panic!("expected quantized multivector, got {other:?}")
}
}
};
let (original_encoding, original_count) = read_encoded(&segment);
assert_eq!(original_count, Some(inner_count));
for op_num in 2..=20 {
let mut payload = Payload::default();
payload
.0
.insert("iter".to_string(), Value::Number(Number::from(op_num)));
segment
.set_full_payload(op_num, point, &payload, &hw_counter)
.unwrap();
let (encoding, count) = read_encoded(&segment);
assert_eq!(
(encoding, count),
(original_encoding.clone(), original_count),
"turbo multivector encoding drifted after {op_num} copy-on-write payload writes",
);
}
}

6
lib/segment/src/vector_storage/sparse/read_only/mod.rs
View File

@@ -90,7 +90,7 @@ mod tests {
assert_eq!(got.indices, sparse_vectors[id as usize].indices);
assert_eq!(got.values, sparse_vectors[id as usize].values);
}
CowVector::Dense(_) | CowVector::MultiDense(_) => {
CowVector::Dense(_) | CowVector::MultiDense(_) | CowVector::Quantized(_) => {
panic!("expected sparse vector for point {id}")
}
}
@@ -157,7 +157,9 @@ mod tests {
// The appended vector is visible; deleted ones are flagged.
match reader.get_vector::<Random>(first.len() as PointOffsetType) {
CowVector::Sparse(got) => assert_eq!(got.values, second[0].values),
CowVector::Dense(_) | CowVector::MultiDense(_) => panic!("expected sparse vector"),
CowVector::Dense(_) | CowVector::MultiDense(_) | CowVector::Quantized(_) => {
panic!("expected sparse vector")
}
}
for &id in &deleted_ids {
assert!(reader.is_deleted_vector(id));

240
lib/segment/src/vector_storage/turbo/mod.rs
View File

@@ -14,9 +14,11 @@ mod test;
mod turbo_encoded_vectors;
use std::borrow::Cow;
use std::collections::HashMap;
use std::ops::Range;
use std::path::{Path, PathBuf};
use std::sync::atomic::AtomicBool;
use std::sync::{Arc, OnceLock, RwLock};
use common::bitvec::BitSlice;
use common::counter::hardware_counter::HardwareCounterCell;
@@ -30,8 +32,8 @@ use self::turbo_encoded_vectors::TurboEncodedVectorStorage;
use crate::common::Flusher;
use crate::common::flags::bitvec_flags::BitvecFlags;
use crate::common::flags::dynamic_stored_flags::DynamicStoredFlags;
use crate::common::operation_error::OperationResult;
use crate::data_types::named_vectors::CowVector;
use crate::common::operation_error::{OperationError, OperationResult};
use crate::data_types::named_vectors::{CowQuantizedVector, CowVector};
use crate::data_types::vectors::{DenseVector, VectorElementType, VectorRef};
use crate::spaces::metric::Metric;
use crate::spaces::simple::{CosineMetric, DotProductMetric, EuclidMetric, ManhattanMetric};
@@ -43,6 +45,85 @@ const TQDT_BITS: TQBits = TQBits::Bits4;
const TQDT_MODE: TQMode = TQMode::Normal;
const TQDT_ROTATION: TQRotation = TQRotation::Unpadded;
/// Build the canonical TQDT quantizer for a `(dim, distance)`.
///
/// Single source of truth for the `TQDT_*` constants so the owning storage, the
/// process-wide dequant cache, and tests all construct an identical quantizer —
/// which is what makes encoded bytes portable and decodable without a storage.
pub(crate) fn tqdt_quantizer(dim: usize, distance: Distance) -> TurboQuantizer {
TurboQuantizer::new(
dim,
TQDT_BITS,
TQDT_MODE,
distance.into(),
TQDT_ROTATION,
None,
)
}
/// Process-wide cache of TQDT quantizers keyed by `(dim, distance)`.
///
/// The quantizer is fully determined by these two plus the global `TQDT_*`
/// constants, so a cached instance decodes bytes identically to the owning
/// storage. This lets a free-floating `CowVector::Quantized` dequantize without
/// holding a storage reference. Constructing a quantizer rebuilds the Hadamard
/// rotation, so we cache rather than rebuild per decode.
fn tqdt_quantizer_cached(dim: usize, distance: Distance) -> Arc<TurboQuantizer> {
type QuantizerCache = HashMap<(usize, Distance), Arc<TurboQuantizer>>;
static CACHE: OnceLock<RwLock<QuantizerCache>> = OnceLock::new();
let cache = CACHE.get_or_init(|| RwLock::new(HashMap::new()));
if let Some(quantizer) = cache.read().unwrap().get(&(dim, distance)) {
return quantizer.clone();
}
cache
.write()
.unwrap()
.entry((dim, distance))
.or_insert_with(|| Arc::new(tqdt_quantizer(dim, distance)))
.clone()
}
/// Decode one TQDT-encoded dense vector to `f32`, dropping the padding tail so
/// the result has the original dimensionality. Shared by the storage's own
/// `dequantize_vector` and the storage-free [`dequantize_tqdt_dense`].
fn decode_dense(quantizer: &TurboQuantizer, bytes: &[u8], dim: usize) -> DenseVector {
let mut dequantized = quantizer.dequantize::<f64>(bytes);
quantizer.apply_inverse_rotation(&mut dequantized);
dequantized[..dim].iter().map(|i| *i as f32).collect()
}
/// Dequantize TQDT-encoded bytes of a single dense vector, resolving the
/// quantizer from the process-wide cache (see [`tqdt_quantizer_cached`]).
pub fn dequantize_tqdt_dense(bytes: &[u8], dim: usize, distance: Distance) -> DenseVector {
decode_dense(&tqdt_quantizer_cached(dim, distance), bytes, dim)
}
/// Dequantize a TQDT-encoded multivector blob: `count` concatenated inner
/// encodings, each one dense record of `dim` floats. Mirrors the dense decode
/// per inner vector and reassembles the flattened multivector.
pub fn dequantize_tqdt_multi(
bytes: &[u8],
dim: usize,
distance: Distance,
count: usize,
) -> crate::data_types::vectors::MultiDenseVectorInternal {
let quantizer = tqdt_quantizer_cached(dim, distance);
let inner_size = quantizer.quantized_size();
let mut flattened = Vec::with_capacity(count * dim);
for inner in bytes.chunks_exact(inner_size) {
let mut dequantized = quantizer.dequantize::<f64>(inner);
quantizer.apply_inverse_rotation(&mut dequantized);
flattened.extend(dequantized[..dim].iter().map(|i| *i as f32));
}
crate::data_types::vectors::MultiDenseVectorInternal::new(flattened, dim)
}
/// Size in bytes of one TQDT-encoded dense vector for `(dim, distance)`. Used to
/// validate a byte upsert and to split a concatenated multivector blob.
pub fn tqdt_quantized_size(dim: usize, distance: Distance) -> usize {
tqdt_quantizer_cached(dim, distance).quantized_size()
}
const VECTORS_PATH: &str = "tq_vectors.dat";
const DELETED_PATH: &str = "deleted.dat";
@@ -76,6 +157,41 @@ impl TurboVectorStorage {
self.storage.get_quantized_vector(key)
}
/// Wrap raw encoded bytes into a [`CowQuantizedVector`] with this storage's
/// decode context (`dim`, `distance`).
fn quantized_cow<'a>(&self, bytes: Cow<'a, [u8]>) -> CowQuantizedVector<'a> {
CowQuantizedVector {
bytes,
dim: self.dim,
distance: self.distance,
multivector_count: None,
}
}
/// Insert a pre-encoded TQ vector blob at `key`, bypassing quantization.
///
/// The bytes must have been produced by an identical quantizer (same TQDT
/// constants); the caller guarantees that. Used by the native
/// segment-to-segment transfer path to avoid a lossy TQ -> float -> TQ
/// round-trip.
pub fn insert_quantized_vector(
&mut self,
key: PointOffsetType,
bytes: &[u8],
hw_counter: &HardwareCounterCell,
) -> OperationResult<()> {
let expected = self.quantizer.quantized_size();
if bytes.len() != expected {
return Err(OperationError::service_error(format!(
"Cannot insert encoded TQ vector: got {} bytes, expected {expected}",
bytes.len(),
)));
}
self.storage.upsert_vector(key, bytes, hw_counter)?;
self.set_deleted(key, false);
Ok(())
}
/// Populate all pages of the encoded vectors into the page cache.
pub fn populate(&self) -> OperationResult<()> {
// deleted bitvec is already loaded
@@ -151,22 +267,6 @@ impl TurboVectorStorage {
let score = self.quantizer.score_symmetric(&v1, &v2);
if self.invert_score() { -score } else { score }
}
fn dequantize_vector(&self, quantized: Cow<[u8]>) -> CowVector<'_> {
let mut dequantized = self.quantizer.dequantize::<f64>(&quantized);
// Rotate back
self.quantizer.apply_inverse_rotation(&mut dequantized);
// Drop the padding tail: callers expect the original dimensionality.
CowVector::Dense(Cow::Owned(
// Skip the padding
dequantized[..self.dim]
.iter()
.map(|i| *i as f32)
.collect::<Vec<_>>(),
))
}
}
impl std::fmt::Debug for TurboVectorStorage {
@@ -232,14 +332,7 @@ fn open_turbo_vector_storage_impl(
) -> OperationResult<TurboVectorStorage> {
fs_err::create_dir_all(path)?;
let quantizer = TurboQuantizer::new(
dim,
TQDT_BITS,
TQDT_MODE,
distance.into(),
TQDT_ROTATION,
None,
);
let quantizer = tqdt_quantizer(dim, distance);
let storage = open_storage(&path.join(VECTORS_PATH), quantizer.quantized_size())?;
@@ -284,12 +377,19 @@ impl VectorStorageRead for TurboVectorStorage {
self.storage.vectors_count()
}
/// A TurboQuant storage holds only encoded bytes — it has **no float view**.
/// `get_vector` therefore returns the storage-native [`CowVector::Quantized`]
/// (borrowed bytes, no copy); any float consumer decodes on demand via
/// [`CowQuantizedVector::dequantize`], a deterministic free function over the
/// bytes. There is intentionally no dequantize method on the storage.
fn get_vector<P: AccessPattern>(&self, key: PointOffsetType) -> CowVector<'_> {
self.dequantize_vector(self.storage.get_quantized_vector(key))
CowVector::Quantized(self.quantized_cow(self.storage.get_quantized_vector(key)))
}
fn get_vector_opt<P: AccessPattern>(&self, key: PointOffsetType) -> Option<CowVector<'_>> {
Some(self.dequantize_vector(self.storage.get_quantized_vector_opt(key)?))
Some(CowVector::Quantized(
self.quantized_cow(self.storage.get_quantized_vector_opt(key)?),
))
}
fn is_deleted_vector(&self, key: PointOffsetType) -> bool {
@@ -312,6 +412,32 @@ impl VectorStorage for TurboVectorStorage {
vector: VectorRef,
hw_counter: &HardwareCounterCell,
) -> OperationResult<()> {
// Native-encoded input (read back from this or another TQ storage): write
// the bytes verbatim instead of dequantizing and re-quantizing, so the
// encoding never drifts on a copy-on-write or transfer. This is what makes
// `VectorRef::Quantized` a first-class storage input — no separate index
// method or upsert chokepoint is needed.
if let VectorRef::Quantized(quantized) = vector {
// Byte length alone is ambiguous (a count-1 multivector blob has the
// same length as a dense one). Reject a blob from a different TQ
// context so we never copy bytes that would decode/score wrong.
if quantized.dim != self.dim
|| quantized.distance != self.distance
|| quantized.multivector_count.is_some()
{
return Err(OperationError::service_error(format!(
"Cannot insert quantized vector: blob context (dim={}, distance={:?}, \
multivector_count={:?}) does not match this dense storage (dim={}, distance={:?})",
quantized.dim,
quantized.distance,
quantized.multivector_count,
self.dim,
self.distance,
)));
}
return self.insert_quantized_vector(key, &quantized.bytes, hw_counter);
}
let dense: &[VectorElementType] = vector.try_into()?;
let quantized = self
@@ -434,6 +560,64 @@ mod tests {
/// independent random inputs instead of a single fixed one.
const SEEDS: [u64; 6] = [42, 0xC0FFEE, 0x0BAD_C0DE, 0x0DECAF, 0x5128E, 0xD15EA5E];
/// The native byte-copy primitive (`insert_quantized_vector`) must preserve
/// the encoding bit-for-bit, unlike the lossy dequantize -> re-quantize path
/// that drives the TQ round-trip drift bug.
#[test]
fn insert_quantized_vector_preserves_encoding_exactly() {
let dim = 256;
let distance = Distance::Dot;
let dir = Builder::new()
.prefix("turbo_native_copy")
.tempdir()
.unwrap();
let hw_counter = HardwareCounterCell::new();
let mut storage =
open_appendable_turbo_vector_storage(dir.path(), dim, distance, true).unwrap();
let input = &make_vectors(dim, 1, 0xABCD)[0];
// key 0: float -> quantize (the original encoding).
storage
.insert_vector(0, input.as_slice().into(), &hw_counter)
.unwrap();
let b0 = storage.get_quantized_vector(0).into_owned();
// key 1: dequantize(b0) -> re-quantize. This is the lossy round-trip.
// `get_vector` now returns the storage-native `Quantized`; `DenseVector::
// try_from` decodes it to the float the re-quantize path consumes.
let dequantized = DenseVector::try_from(storage.get_vector::<Random>(0).to_owned())
.expect("dense vector");
storage
.insert_vector(1, dequantized.as_slice().into(), &hw_counter)
.unwrap();
let b1 = storage.get_quantized_vector(1).into_owned();
// key 2: native byte copy of b0 -> must be byte-identical to b0.
storage
.insert_quantized_vector(2, &b0, &hw_counter)
.unwrap();
let b2 = storage.get_quantized_vector(2).into_owned();
assert_eq!(
b2, b0,
"native byte copy must preserve the encoding exactly"
);
assert!(!storage.is_deleted_vector(2));
// b1 is the re-quantized round-trip; it is allowed to drift from b0.
// We don't assert it differs (some vectors are fixed points), but the
// native copy (b2) is guaranteed not to.
let _ = b1;
// A wrong-sized blob must be rejected, not silently corrupt the slot.
assert!(
storage
.insert_quantized_vector(3, &b0[..b0.len() - 1], &hw_counter)
.is_err(),
);
}
#[test]
fn upsert_flush_reload_in_ram_matches_independent_oracle() {
const COUNT: usize = 64;

128
lib/segment/src/vector_storage/turbo/multi.rs
View File

@@ -28,10 +28,9 @@ use crate::common::Flusher;
use crate::common::flags::bitvec_flags::BitvecFlags;
use crate::common::flags::dynamic_stored_flags::DynamicStoredFlags;
use crate::common::operation_error::{OperationError, OperationResult, check_process_stopped};
use crate::data_types::named_vectors::{CowMultiVector, CowVector};
use crate::data_types::named_vectors::{CowQuantizedVector, CowVector};
use crate::data_types::vectors::{
MultiDenseVectorInternal, TypedMultiDenseVector, TypedMultiDenseVectorRef, VectorElementType,
VectorRef,
MultiDenseVectorInternal, TypedMultiDenseVectorRef, VectorElementType, VectorRef,
};
use crate::spaces::metric::Metric;
use crate::spaces::simple::{CosineMetric, DotProductMetric, EuclidMetric, ManhattanMetric};
@@ -170,30 +169,6 @@ impl TurboMultiVectorStorage {
previous
}
/// Decode one inner record into `out`: dequantize, rotate back, drop the padding tail.
fn dequantize_inner_into(&self, encoded: &[u8], out: &mut Vec<VectorElementType>) {
let mut dequantized = self.quantizer.dequantize::<f64>(encoded);
self.quantizer.apply_inverse_rotation(&mut dequantized);
out.extend(
dequantized[..self.dim]
.iter()
.map(|&x| x as VectorElementType),
);
}
/// Decode the full multivector behind an offset record.
fn dequantize_multi(&self, offset: MultivectorMmapOffset) -> CowVector<'_> {
let mut flattened = Vec::with_capacity(offset.count as usize * self.dim);
for inner_id in offset.offset..offset.offset + offset.count {
let encoded = self.storage.get_vector_data(inner_id);
self.dequantize_inner_into(&encoded, &mut flattened);
}
CowVector::MultiDense(CowMultiVector::Owned(TypedMultiDenseVector {
flattened_vectors: flattened,
dim: self.dim,
}))
}
/// Start of a fresh range for `count` records, never straddling a chunk
/// boundary: skips the tail when the range wouldn't fit it, errors when
/// even a whole chunk can't hold it.
@@ -260,6 +235,67 @@ impl TurboMultiVectorStorage {
Ok(())
}
/// Upsert a pre-encoded multivector blob at `key`, bypassing quantization:
/// the bytes are `count` concatenated inner encodings produced by an identical
/// quantizer. Mirrors [`Self::insert_multi`] but copies the records verbatim.
fn insert_quantized_multi(
&mut self,
key: PointOffsetType,
quantized: &CowQuantizedVector,
hw_counter: &HardwareCounterCell,
) -> OperationResult<()> {
let Some(inner_count) = quantized.multivector_count else {
return Err(OperationError::service_error(
"Cannot insert a dense-encoded TQ vector into multivector storage",
));
};
// A blob from another TQ storage with the same per-record size but a
// different dim/distance would decode/score wrong; reject it.
if quantized.dim != self.dim || quantized.distance != self.distance {
return Err(OperationError::service_error(format!(
"Cannot insert quantized multivector: blob context (dim={}, distance={:?}) does \
not match this storage (dim={}, distance={:?})",
quantized.dim, quantized.distance, self.dim, self.distance,
)));
}
let inner_size = self.quantizer.quantized_size();
let expected = inner_count * inner_size;
if quantized.bytes.len() != expected {
return Err(OperationError::service_error(format!(
"Cannot insert encoded TQ multivector: got {} bytes, expected {expected} \
({inner_count} x {inner_size})",
quantized.bytes.len(),
)));
}
let count = inner_count as PointOffsetType;
let mut offset = self
.offsets
.get::<Random>(key as VectorOffsetType)
.map(|x| x.first().copied().unwrap_or_default())
.unwrap_or_default();
if count > offset.capacity {
offset = MultivectorMmapOffset {
offset: self.fresh_range_start(count)?,
count,
capacity: count,
};
} else {
offset.count = count;
}
for (i, inner) in quantized.bytes.chunks_exact(inner_size).enumerate() {
self.storage
.upsert_vector(offset.offset + i as PointOffsetType, inner, hw_counter)?;
}
self.offsets
.insert(key as VectorOffsetType, &[offset], hw_counter)?;
self.set_deleted(key, false);
Ok(())
}
/// Populate all pages of the encoded vectors and offsets into the page cache.
pub fn populate(&self) -> OperationResult<()> {
// deleted bitvec is already loaded
@@ -423,7 +459,19 @@ impl VectorStorageRead for TurboMultiVectorStorage {
}
fn get_vector_opt<P: AccessPattern>(&self, key: PointOffsetType) -> Option<CowVector<'_>> {
Some(self.dequantize_multi(self.get_offset::<P>(key)?))
// Hand out the native encoded inner records verbatim (no dequantize), so a
// copy-on-write or transfer re-stores the exact bytes. The blob is the
// point's `count` inner encodings concatenated; `dim` is the inner dim.
let offset = self.get_offset::<P>(key)?;
let bytes = self
.storage
.get_many::<P>(offset.offset, offset.count as usize)?;
Some(CowVector::Quantized(CowQuantizedVector {
bytes,
dim: self.dim,
distance: self.distance,
multivector_count: Some(offset.count as usize),
}))
}
fn is_deleted_vector(&self, key: PointOffsetType) -> bool {
@@ -456,6 +504,13 @@ impl VectorStorage for TurboMultiVectorStorage {
vector: VectorRef,
hw_counter: &HardwareCounterCell,
) -> OperationResult<()> {
// Native-encoded input (read back from this or another TQ multivector
// storage): write the inner byte records verbatim instead of
// dequantizing and re-quantizing, so the encoding never drifts on a
// copy-on-write or transfer.
if let VectorRef::Quantized(quantized) = vector {
return self.insert_quantized_multi(key, quantized, hw_counter);
}
let multi_vector: TypedMultiDenseVectorRef<VectorElementType> = vector.try_into()?;
self.insert_multi(key, multi_vector, hw_counter)
}
@@ -566,7 +621,7 @@ mod tests {
use tempfile::Builder;
use super::*;
use crate::data_types::vectors::{DenseVector, MultiDenseVectorInternal};
use crate::data_types::vectors::{DenseVector, MultiDenseVectorInternal, VectorInternal};
use crate::vector_storage::common::CHUNK_SIZE;
/// Deterministic multivectors of unit inner vectors; point `i` gets `(i % 4) + 1` inner vectors.
@@ -601,12 +656,19 @@ mod tests {
dot / (na * nb)
}
/// Extract the owned multivector; this storage always returns `Owned`.
/// Decode the multivector; this storage always returns native quantized bytes.
fn to_multi(vector: CowVector) -> MultiDenseVectorInternal {
match vector {
CowVector::MultiDense(CowMultiVector::Owned(multi)) => multi,
CowVector::Dense(_) | CowVector::Sparse(_) | CowVector::MultiDense(_) => {
panic!("expected owned multi-dense vector")
CowVector::Quantized(quantized) => match quantized.dequantize() {
VectorInternal::MultiDense(multi) => multi,
other @ (VectorInternal::Dense(_)
| VectorInternal::Sparse(_)
| VectorInternal::Quantized(_)) => {
panic!("expected multi-dense dequantize, got {other:?}")
}
},
other @ (CowVector::Dense(_) | CowVector::Sparse(_) | CowVector::MultiDense(_)) => {
panic!("expected quantized multivector, got {other:?}")
}
}
}

18
lib/segment/src/vector_storage/turbo/test.rs
View File

@@ -12,8 +12,9 @@ use tempfile::Builder;
use super::multi::{TurboMultiVectorStorage, open_appendable_turbo_multi_vector_storage};
use super::*;
use crate::data_types::named_vectors::CowMultiVector;
use crate::data_types::vectors::{MultiDenseVectorInternal, TypedMultiDenseVectorRef};
use crate::data_types::vectors::{
MultiDenseVectorInternal, TypedMultiDenseVectorRef, VectorInternal,
};
use crate::types::MultiVectorConfig;
use crate::vector_storage::MultiTQVectorStorage;
@@ -40,9 +41,16 @@ fn to_dense(vector: CowVector) -> DenseVector {
fn to_multi(vector: CowVector) -> MultiDenseVectorInternal {
match vector {
CowVector::MultiDense(CowMultiVector::Owned(multi)) => multi,
CowVector::Dense(_) | CowVector::Sparse(_) | CowVector::MultiDense(_) => {
panic!("expected owned multi-dense vector")
CowVector::Quantized(quantized) => match quantized.dequantize() {
VectorInternal::MultiDense(multi) => multi,
other @ (VectorInternal::Dense(_)
| VectorInternal::Sparse(_)
| VectorInternal::Quantized(_)) => {
panic!("expected multi-dense dequantize, got {other:?}")
}
},
other @ (CowVector::Dense(_) | CowVector::Sparse(_) | CowVector::MultiDense(_)) => {
panic!("expected quantized multivector, got {other:?}")
}
}
}

2
lib/shard/src/operations/point_ops.rs
View File

@@ -809,6 +809,8 @@ impl From<VectorInternal> for VectorPersisted {
VectorInternal::MultiDense(vector) => {
VectorPersisted::MultiDense(vector.into_multi_vectors())
}
// The float persisted form carries no encoding; decode the bytes.
VectorInternal::Quantized(q) => Self::from(q.dequantize()),
}
}
}

12
lib/shard/src/query/conversions.rs
View File

@@ -495,12 +495,16 @@ impl QueryEnum {
Some(name),
VectorInternal::MultiDense(_)
| VectorInternal::Sparse(_)
| VectorInternal::Dense(_),
| VectorInternal::Dense(_)
| VectorInternal::Quantized(_),
) => name,
(None, VectorInternal::MultiDense(_) | VectorInternal::Dense(_)) => {
DEFAULT_VECTOR_NAME.to_owned()
}
(
None,
VectorInternal::MultiDense(_)
| VectorInternal::Dense(_)
| VectorInternal::Quantized(_),
) => DEFAULT_VECTOR_NAME.to_owned(),
};
let named_vector = NamedQuery::new(vector, name);

22
lib/shard/src/query/mmr/mod.rs
View File

@@ -67,6 +67,19 @@ pub fn mmr_from_points_with_vector(
return Ok(candidates);
}
// Quantized candidates (e.g. TurboQuant) have no borrowed float view; decode
// them once up front so both the volatile storage population *and* the
// pairwise similarity matrix operate on plain floats, not raw bytes.
let vectors: Vec<VectorInternal> = vectors
.into_iter()
.map(|vector| match vector {
VectorInternal::Quantized(quantized) => quantized.dequantize(),
vector @ (VectorInternal::Dense(_)
| VectorInternal::Sparse(_)
| VectorInternal::MultiDense(_)) => vector,
})
.collect();
let volatile_storage = create_volatile_storage(
&vectors,
distance,
@@ -100,6 +113,10 @@ pub fn mmr_from_points_with_vector(
}
/// Creates a volatile (in-memory and not persistent) vector storage and inserts the vectors in the provided order.
///
/// `vectors` must already be decoded (no `VectorInternal::Quantized`); the caller
/// decodes once so the same float vectors feed both this storage and the
/// similarity matrix.
fn create_volatile_storage(
vectors: &[VectorInternal],
distance: Distance,
@@ -128,6 +145,11 @@ fn create_volatile_storage(
}
VectorInternal::Sparse(_) => new_volatile_sparse_vector_storage(),
// The caller decodes quantized candidates before calling, so this is unreachable here.
VectorInternal::Quantized(_) => {
unreachable!("quantized vectors are decoded by the caller")
}
}
};

20
lib/shard/src/query/query_enum.rs
View File

@@ -43,7 +43,9 @@ impl QueryEnum {
match self {
QueryEnum::Nearest(named) => match &named.query {
VectorInternal::Sparse(sparse_vector) => f(named.get_name(), sparse_vector),
VectorInternal::Dense(_) | VectorInternal::MultiDense(_) => {}
VectorInternal::Dense(_)
| VectorInternal::MultiDense(_)
| VectorInternal::Quantized(_) => {}
},
QueryEnum::RecommendBestScore(reco_query)
| QueryEnum::RecommendSumScores(reco_query) => {
@@ -51,7 +53,9 @@ impl QueryEnum {
for vector in reco_query.query.flat_iter() {
match vector {
VectorInternal::Sparse(sparse_vector) => f(name, sparse_vector),
VectorInternal::Dense(_) | VectorInternal::MultiDense(_) => {}
VectorInternal::Dense(_)
| VectorInternal::MultiDense(_)
| VectorInternal::Quantized(_) => {}
}
}
}
@@ -60,7 +64,9 @@ impl QueryEnum {
for vector in discover_query.query.flat_iter() {
match vector {
VectorInternal::Sparse(sparse_vector) => f(name, sparse_vector),
VectorInternal::Dense(_) | VectorInternal::MultiDense(_) => {}
VectorInternal::Dense(_)
| VectorInternal::MultiDense(_)
| VectorInternal::Quantized(_) => {}
}
}
}
@@ -69,7 +75,9 @@ impl QueryEnum {
for vector in context_query.query.flat_iter() {
match vector {
VectorInternal::Sparse(sparse_vector) => f(name, sparse_vector),
VectorInternal::Dense(_) | VectorInternal::MultiDense(_) => {}
VectorInternal::Dense(_)
| VectorInternal::MultiDense(_)
| VectorInternal::Quantized(_) => {}
}
}
}
@@ -78,7 +86,9 @@ impl QueryEnum {
for vector in feedback_query.query.flat_iter() {
match vector {
VectorInternal::Sparse(sparse_vector) => f(name, sparse_vector),
VectorInternal::Dense(_) | VectorInternal::MultiDense(_) => {}
VectorInternal::Dense(_)
| VectorInternal::MultiDense(_)
| VectorInternal::Quantized(_) => {}
}
}
}

20
lib/shard/src/segment_holder/mod.rs
View File

@@ -793,11 +793,21 @@ impl SegmentHolder {
// fail with `PointIdError`, surfacing as a spurious
// "No point with id ... found" on a plain upsert that
// races a `prevent_unoptimized` optimization.
let mut all_vectors = write_segment.all_vectors_with_behavior(
point_id,
DeferredBehavior::WithDeferred,
hw_counter,
)?;
// Read the point's vectors in their storage-native form
// (TurboQuant comes back as `CowVector::Quantized`). A
// vector the cow operation leaves untouched is written back
// verbatim — no lossy dequantize -> re-quantize round-trip;
// a vector it overwrites is re-quantized as usual. The move
// closures only insert/replace vectors, never read them as
// float, so the quantized form passes through safely and
// `upsert_point`'s write path preserves it.
let mut all_vectors = write_segment
.all_vectors_with_behavior(
point_id,
DeferredBehavior::WithDeferred,
hw_counter,
)?
.into_owned();
let mut payload = write_segment.payload_with_behavior(
point_id,
DeferredBehavior::WithDeferred,

99
lib/shard/src/segment_holder/tests.rs
View File

@@ -1576,3 +1576,102 @@ fn test_cow_deletes_source_when_destination_is_not_deferred() {
"Point 100 should be deleted from the source"
);
}
/// Regression test for the TurboQuant round-trip drift bug on cross-segment
/// copy-on-write. A payload write on a point living in a non-appendable segment
/// moves it to an appendable segment; the vector must be relocated as native
/// encoded bytes, not dequantized and re-quantized, so its encoding is identical
/// before and after the move.
#[test]
fn test_cow_preserves_turbo_encoding() {
use common::types::DeferredBehavior;
use segment::data_types::named_vectors::CowVector;
use segment::data_types::vectors::only_default_vector;
use segment::segment_constructor::build_segment;
use segment::types::{
Indexes, SegmentConfig, VectorDataConfig, VectorStorageDatatype, VectorStorageType,
};
let dim = 128;
let dir = Builder::new().prefix("turbo_cow_holder").tempdir().unwrap();
let hw_counter = HardwareCounterCell::new();
let turbo_config = || SegmentConfig {
vector_data: HashMap::from([(
DEFAULT_VECTOR_NAME.to_owned(),
VectorDataConfig {
size: dim,
distance: Distance::Dot,
storage_type: VectorStorageType::ChunkedMmap,
index: Indexes::Plain {},
quantization_config: None,
multivector_config: None,
datatype: Some(VectorStorageDatatype::Turbo4),
},
)]),
sparse_vector_data: HashMap::new(),
payload_storage_type: Default::default(),
};
// Source: a Turbo segment holding the point, then frozen as non-appendable so
// a mutation forces a copy-on-write move into the appendable target.
let (mut source, _) = build_segment(dir.path(), &turbo_config(), None, true).unwrap();
let point = 1.into();
let mut rng = rand::rng();
let dense: Vec<f32> = {
let v: Vec<f32> = (0..dim).map(|_| rng.random_range(-1.0..1.0)).collect();
let norm = v.iter().map(|&x| x * x).sum::<f32>().sqrt();
v.iter().map(|&x| x / norm).collect()
};
source
.upsert_point(1, point, only_default_vector(&dense), &hw_counter)
.unwrap();
let read_encoding = |segment: &dyn SegmentEntry| -> Vec<u8> {
let native = segment
.all_vectors_with_behavior(point, DeferredBehavior::WithDeferred, &hw_counter)
.unwrap();
match native
.get_cow(DEFAULT_VECTOR_NAME)
.expect("default vector must exist")
{
CowVector::Quantized(quantized) => quantized.bytes.to_vec(),
other @ (CowVector::Dense(_) | CowVector::Sparse(_) | CowVector::MultiDense(_)) => {
panic!("expected quantized vector, got {other:?}")
}
}
};
let original_encoding = read_encoding(&source);
source.appendable_flag = false;
// Empty appendable Turbo target to receive the moved point.
let (target, _) = build_segment(dir.path(), &turbo_config(), None, true).unwrap();
let mut holder = SegmentHolder::default();
let target_id = holder.add_new(target);
let _source_id = holder.add_new(source);
holder
.apply_points_with_conditional_move(
100,
&[point],
|_point_id, _segment| Ok(true),
|_point_id, _vectors, payload| {
*payload = payload_json! { "moved": true };
},
&hw_counter,
)
.unwrap();
let target = holder.get(target_id).unwrap().get();
let target = target.read();
let moved_encoding = read_encoding(&*target);
assert_eq!(
moved_encoding, original_encoding,
"turbo encoding drifted across a copy-on-write segment move",
);
}