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

[TQDT] fix lossy roundtrip in apply_points_with_conditional_move (#9734)

Browse Source 
* TQDT fix lossy roundtrip in apply_points_with_conditional_move

* Reduce allocations + Fix debug_assert

* Refactor _raw methods to reduce allocations

* Rename refactored function

* Drop omitted named vectors in the CoW upsert path.

* Revert refactor of *_raw functions

* Avoid allocations

* Review remarks
This commit is contained in:
Jojii
2026-07-10 16:08:31 +02:00
committed by GitHub
parent a4dd5c6b03
commit e99d433fba
10 changed files with 748 additions and 157 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -258,6 +258,11 @@ impl<'a> NamedVectors<'a> {
.insert(CowKey::Owned(name), CowVector::from(vector));
}
pub fn insert_ref(&mut self, name: &'a VectorName, vector: VectorRef<'a>) {
self.map
.insert(Cow::Borrowed(name), CowVector::from(vector));
}
pub fn remove_ref(&mut self, key: &VectorName) {
self.map.remove(key);
}

26
lib/segment/src/entry/entry_point.rs
View File

@@ -87,19 +87,6 @@ pub trait ReadSegmentEntry {
hw_counter: &HardwareCounterCell,
) -> OperationResult<NamedVectors<'_>>;
/// Like [`SegmentEntry::all_vectors`], but with explicit deferred semantics.
///
/// With [`DeferredBehavior::WithDeferred`] this resolves the latest head of
/// the point, including a deferred head that is invisible to ordinary reads.
/// Used by the copy-on-write move path so deferred points are relocated with
/// their actual data instead of an empty/visible-only snapshot.
fn all_vectors_with_behavior(
&self,
point_id: PointIdType,
deferred_behavior: DeferredBehavior,
hw_counter: &HardwareCounterCell,
) -> OperationResult<NamedVectors<'_>>;
/// Reads Records from the segment, according to specified selectors and a list of point ids.
///
/// WARNING:
@@ -138,19 +125,6 @@ pub trait ReadSegmentEntry {
hw_counter: &HardwareCounterCell,
) -> OperationResult<Payload>;
/// Like [`SegmentEntry::payload`], but with explicit deferred semantics.
///
/// With [`DeferredBehavior::WithDeferred`] this resolves the latest head of
/// the point, including a deferred head that is invisible to ordinary reads.
/// Used by the copy-on-write move path so deferred points are relocated with
/// their actual payload instead of failing with `PointIdError`.
fn payload_with_behavior(
&self,
point_id: PointIdType,
deferred_behavior: DeferredBehavior,
hw_counter: &HardwareCounterCell,
) -> OperationResult<Payload>;
/// Paginate over points which satisfies filtering condition starting with `offset` id including.
///
/// Cancelled by `is_stopped` flag.

27
lib/segment/src/segment/entry.rs
View File

@@ -104,22 +104,16 @@ impl ReadSegmentEntry for Segment {
&self,
point_id: PointIdType,
hw_counter: &HardwareCounterCell,
) -> OperationResult<NamedVectors<'_>> {
self.all_vectors_with_behavior(point_id, DeferredBehavior::VisibleOnly, hw_counter)
}
fn all_vectors_with_behavior(
&self,
point_id: PointIdType,
deferred_behavior: DeferredBehavior,
hw_counter: &HardwareCounterCell,
) -> OperationResult<NamedVectors<'_>> {
self.with_view(|view| {
let mut result = NamedVectors::default();
for vector_name in view.vector_data.keys() {
if let Some(vec) =
view.vector_with_behavior(vector_name, point_id, deferred_behavior, hw_counter)?
{
if let Some(vec) = view.vector_with_behavior(
vector_name,
point_id,
DeferredBehavior::VisibleOnly,
hw_counter,
)? {
result.insert(vector_name.clone(), vec);
}
}
@@ -135,15 +129,6 @@ impl ReadSegmentEntry for Segment {
self.with_view(|view| view.payload(point_id, hw_counter))
}
fn payload_with_behavior(
&self,
point_id: PointIdType,
deferred_behavior: DeferredBehavior,
hw_counter: &HardwareCounterCell,
) -> OperationResult<Payload> {
self.with_view(|view| view.payload_with_behavior(point_id, deferred_behavior, hw_counter))
}
fn retrieve(
&self,
point_ids: &[PointIdType],

27
lib/segment/src/segment/read_only/read_entry.rs
View File

@@ -100,22 +100,16 @@ impl<S: UniversalReadExt + 'static> ReadSegmentEntry for ReadOnlySegment<S> {
&self,
point_id: PointIdType,
hw_counter: &HardwareCounterCell,
) -> OperationResult<NamedVectors<'_>> {
self.all_vectors_with_behavior(point_id, DeferredBehavior::VisibleOnly, hw_counter)
}
fn all_vectors_with_behavior(
&self,
point_id: PointIdType,
deferred_behavior: DeferredBehavior,
hw_counter: &HardwareCounterCell,
) -> OperationResult<NamedVectors<'_>> {
self.with_view(|view| {
let mut result = NamedVectors::default();
for vector_name in view.vector_data.keys() {
if let Some(vec) =
view.vector_with_behavior(vector_name, point_id, deferred_behavior, hw_counter)?
{
if let Some(vec) = view.vector_with_behavior(
vector_name,
point_id,
DeferredBehavior::VisibleOnly,
hw_counter,
)? {
result.insert(vector_name.clone(), vec);
}
}
@@ -131,15 +125,6 @@ impl<S: UniversalReadExt + 'static> ReadSegmentEntry for ReadOnlySegment<S> {
self.with_view(|view| view.payload(point_id, hw_counter))
}
fn payload_with_behavior(
&self,
point_id: PointIdType,
deferred_behavior: DeferredBehavior,
hw_counter: &HardwareCounterCell,
) -> OperationResult<Payload> {
self.with_view(|view| view.payload_with_behavior(point_id, deferred_behavior, hw_counter))
}
fn retrieve(
&self,
point_ids: &[PointIdType],

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

@@ -1929,15 +1929,15 @@ fn test_deferred_point_read_operations() {
}
/// A deferred point is invisible to ordinary (`VisibleOnly`) per-point reads,
/// but the copy-on-write move path must still be able to read its data via the
/// `*_with_behavior(WithDeferred)` accessors.
/// but the copy-on-write move path must still be able to read its data via
/// `retrieve`/`retrieve_raw` with `DeferredBehavior::WithDeferred`.
///
/// Regression test for a spurious "No point with id ... found" error: the CoW
/// move path read the source point's payload with `VisibleOnly`, which raised
/// `PointIdError` for a deferred source point (e.g. a point upserted under
/// `prevent_unoptimized` whose internal id is beyond the deferred threshold).
#[test]
fn test_deferred_point_with_behavior_accessors() {
fn test_deferred_point_with_deferred_reads() {
let hw_counter = HardwareCounterCell::new();
let dir = Builder::new().prefix("segment_dir").tempdir().unwrap();
let dim = 4;
@@ -2002,18 +2002,37 @@ fn test_deferred_point_with_behavior_accessors() {
);
// WithDeferred resolves the deferred point with its real data — this is
// what the CoW move path now uses, fixing the spurious PointNotFound.
let with_deferred_payload = segment
.payload_with_behavior(point_id, DeferredBehavior::WithDeferred, &hw_counter)
.expect("WithDeferred payload read must resolve a deferred point");
assert_eq!(with_deferred_payload, payload);
let with_deferred_vectors = segment
.all_vectors_with_behavior(point_id, DeferredBehavior::WithDeferred, &hw_counter)
.expect("WithDeferred vector read must resolve a deferred point");
// what the CoW move path uses (via `retrieve_raw`), fixing the spurious
// PointNotFound.
let records = segment
.retrieve(
&[point_id],
&WithPayload {
enable: true,
payload_selector: None,
},
&WithVector::Bool(true),
&hw_counter,
&AtomicBool::new(false),
DeferredBehavior::WithDeferred,
)
.unwrap();
let record = records
.get(&point_id)
.expect("WithDeferred retrieve must resolve a deferred point");
assert_eq!(
record.payload.clone().unwrap_or_default(),
payload,
"WithDeferred retrieve must return the deferred point's payload",
);
assert!(
with_deferred_vectors.contains_key(DEFAULT_VECTOR_NAME),
"WithDeferred vector read must return the deferred point's vector",
record
.vectors
.as_ref()
.expect("vectors were requested")
.iter()
.any(|(name, _)| name == DEFAULT_VECTOR_NAME),
"WithDeferred retrieve must return the deferred point's vector",
);
}

29
lib/segment/src/vector_storage/vector_storage_base.rs
View File

@@ -876,10 +876,16 @@ fn insert_dense_bytes<T: PrimitiveVectorElement, S: DenseVectorStorageRead<T> +
bytes.len(),
)));
}
// `pod_collect_to_vec` instead of `cast_slice`: incoming bytes may not be
// aligned for `T`.
let elements: Vec<T> = bytemuck::allocation::pod_collect_to_vec(bytes);
let vector = T::slice_to_float_cow(Cow::Owned(elements));
// Zero-copy cast when the byte buffer is aligned for `T` (heap
// allocations virtually always are); the length already matches, so
// misalignment is the only way the cast can fail, and
// `pod_collect_to_vec` then copies into an aligned buffer.
let elements: Cow<'_, [T]> = match bytemuck::try_cast_slice(bytes) {
Ok(slice) => Cow::Borrowed(slice),
Err(_) => Cow::Owned(bytemuck::allocation::pod_collect_to_vec(bytes)),
};
let vector = T::slice_to_float_cow(elements);
storage.insert_vector(key, VectorRef::from(vector.as_ref()), hw_counter)
}
@@ -898,11 +904,18 @@ fn insert_multi_bytes<T: PrimitiveVectorElement, S: MultiVectorStorageRead<T> +
bytes.len(),
)));
}
let elements: Vec<T> = bytemuck::allocation::pod_collect_to_vec(bytes);
let dim = storage.vector_dim();
let multi = T::into_float_multivector(CowMultiVector::Owned(TypedMultiDenseVector::new(
elements, dim,
)));
// Zero-copy cast when the byte buffer is aligned for `T` (heap
// allocations virtually always are); copy to an aligned buffer otherwise.
let multi = match bytemuck::try_cast_slice(bytes) {
Ok(slice) => T::into_float_multivector(CowMultiVector::Borrowed(
TypedMultiDenseVectorRef::new(slice, dim),
)),
Err(_) => T::into_float_multivector(CowMultiVector::Owned(TypedMultiDenseVector::new(
bytemuck::allocation::pod_collect_to_vec(bytes),
dim,
))),
};
storage.insert_vector(key, VectorRef::MultiDense(multi.as_vec_ref()), hw_counter)
}

36
lib/shard/src/proxy_segment/segment_entry.rs
View File

@@ -305,15 +305,6 @@ impl ReadSegmentEntry for ProxySegment {
&self,
point_id: PointIdType,
hw_counter: &HardwareCounterCell,
) -> OperationResult<NamedVectors<'_>> {
self.all_vectors_with_behavior(point_id, DeferredBehavior::VisibleOnly, hw_counter)
}
fn all_vectors_with_behavior(
&self,
point_id: PointIdType,
deferred_behavior: DeferredBehavior,
hw_counter: &HardwareCounterCell,
) -> OperationResult<NamedVectors<'_>> {
let mut result = NamedVectors::default();
let wrapped = self.wrapped_segment.get();
@@ -332,9 +323,12 @@ impl ReadSegmentEntry for ProxySegment {
drop(wrapped_guard);
for vector_name in vector_names {
if let Some(vector) =
self.vector_with_behavior(&vector_name, point_id, deferred_behavior, hw_counter)?
{
if let Some(vector) = self.vector_with_behavior(
&vector_name,
point_id,
DeferredBehavior::VisibleOnly,
hw_counter,
)? {
result.insert(vector_name, vector);
}
}
@@ -345,24 +339,14 @@ impl ReadSegmentEntry for ProxySegment {
&self,
point_id: PointIdType,
hw_counter: &HardwareCounterCell,
) -> OperationResult<Payload> {
self.payload_with_behavior(point_id, DeferredBehavior::VisibleOnly, hw_counter)
}
fn payload_with_behavior(
&self,
point_id: PointIdType,
deferred_behavior: DeferredBehavior,
hw_counter: &HardwareCounterCell,
) -> OperationResult<Payload> {
if self.deleted_points.contains_key(&point_id) {
Ok(Payload::default())
} else {
self.wrapped_segment.get().read().payload_with_behavior(
point_id,
deferred_behavior,
hw_counter,
)
self.wrapped_segment
.get()
.read()
.payload(point_id, hw_counter)
}
}

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

@@ -11,7 +11,7 @@ use std::collections::{BTreeMap, BTreeSet, HashSet};
use std::ops::{Deref, DerefMut};
use std::path::Path;
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
use std::thread::JoinHandle;
use std::time::Duration;
@@ -31,7 +31,10 @@ use segment::entry::{
};
use segment::segment::Segment;
use segment::segment_constructor::build_segment;
use segment::types::{ExtendedPointId, Payload, PointIdType, SegmentConfig, SeqNumberType};
use segment::types::{
ExtendedPointId, Payload, PointIdType, SegmentConfig, SeqNumberType, VectorNameBuf,
WithPayload, WithVector,
};
use smallvec::SmallVec;
use crate::locked_segment::{DropDataOutcome, LockedSegment};
@@ -965,6 +968,14 @@ impl SegmentHolder {
///
/// Returns set of point ids which were successfully (already) applied to segments.
///
/// `point_cow_operation` receives the moved point's vectors twice: as
/// storage-native bytes (as read from the source; remove a named vector by
/// `retain`ing on the list) and as an initially empty decoded overlay
/// (insert fresh vectors there to overwrite names; entries may borrow from
/// the operation data via `'op`). Untouched names travel to the
/// destination as verbatim bytes, which keeps requantizing storages
/// (TurboQuant-as-datatype) lossless across moves.
///
/// # Warning
///
/// This function must not be used to apply point deletions, and [`apply_points`] must be used
@@ -973,7 +984,7 @@ impl SegmentHolder {
/// 1. moving a point first and deleting it after is unnecessary overhead.
/// 2. this leaves older point versions in place, which may accidentally be revived by some
/// other operation later.
pub fn apply_points_with_conditional_move<F, G>(
pub fn apply_points_with_conditional_move<'op, F, G>(
&self,
op_num: SeqNumberType,
ids: &[PointIdType],
@@ -983,12 +994,18 @@ impl SegmentHolder {
) -> OperationResult<AHashSet<PointIdType>>
where
F: FnMut(PointIdType, &mut RwLockWriteGuard<dyn SegmentEntry>) -> OperationResult<bool>,
for<'n, 'o, 'p> G: FnMut(PointIdType, &'n mut NamedVectors<'o>, &'p mut Payload),
G: FnMut(
PointIdType,
&mut Vec<(VectorNameBuf, Vec<u8>)>,
&mut NamedVectors<'op>,
&mut Payload,
),
{
// Choose random appendable segment from this
let appendable_segments = self.appendable_segments_ids();
let mut applied_points: AHashSet<PointIdType> = Default::default();
let stopped = AtomicBool::new(false);
let _ = self.apply_points(ids, hw_counter, |point_id, idx, write_segment| {
if let Some(point_version) = write_segment.point_version(point_id)
@@ -1017,29 +1034,88 @@ impl SegmentHolder {
// Read the latest head of the point, including a
// deferred head that is invisible to ordinary
// (`VisibleOnly`) reads. A deferred source point would
// otherwise yield empty vectors and make `payload`
// fail with `PointIdError`, surfacing as a spurious
// otherwise yield no record, 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,
)?;
let mut payload = write_segment.payload_with_behavior(
point_id,
DeferredBehavior::WithDeferred,
hw_counter,
)?;
//
// Vectors are read as storage-native bytes: names the
// operation does not touch travel verbatim, avoiding
// the lossy dequantize→requantize round-trip of
// TurboQuant-as-datatype storages.
let mut record = write_segment
.retrieve_raw(
&[point_id],
&WithPayload {
enable: true,
payload_selector: None,
},
&WithVector::Bool(true),
hw_counter,
&stopped,
DeferredBehavior::WithDeferred,
)?
.remove(&point_id)
.ok_or(OperationError::PointIdError {
missed_point_id: point_id,
})?;
point_cow_operation(point_id, &mut all_vectors, &mut payload);
let mut raw_vectors = record.vectors.take().unwrap_or_default();
let mut payload = record.payload.take().unwrap_or_default();
let mut updated_vectors = NamedVectors::default();
appendable_write_segment.upsert_point(
point_cow_operation(
point_id,
&mut raw_vectors,
&mut updated_vectors,
&mut payload,
);
// Names overlaid with fresh data don't travel as bytes.
raw_vectors
.retain(|(name, _)| updated_vectors.get(name).is_none());
// Byte portability requires encoding-compatible vector
// configs on both sides (size, distance, datatype,
// multivector config). Segment-role fields — storage
// type, index, quantization — legitimately differ
// between an optimized source and an appendable
// destination; `check_compatible` ignores them.
debug_assert!(
raw_vectors.iter().all(|(name, _)| {
let src = write_segment.config();
let dst = appendable_write_segment.config();
let dense = (src.vector_data.get(name), dst.vector_data.get(name));
let sparse = (
src.sparse_vector_data.get(name),
dst.sparse_vector_data.get(name),
);
match (dense, sparse) {
((Some(src), Some(dst)), (None, None)) => {
src.check_compatible(dst).is_ok()
}
((None, None), (Some(src), Some(dst))) => {
src.check_compatible(dst).is_ok()
}
_ => false,
}
}),
"CoW raw move requires encoding-compatible vector configs on source and destination",
);
appendable_write_segment.upsert_point_raw(
op_num,
point_id,
all_vectors,
&raw_vectors,
hw_counter,
)?;
if !updated_vectors.is_empty() {
appendable_write_segment.update_vectors(
op_num,
point_id,
updated_vectors,
hw_counter,
)?;
}
appendable_write_segment
.set_full_payload(op_num, point_id, &payload, hw_counter)?;

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

@@ -61,7 +61,7 @@ fn test_apply_to_appendable() {
assert!(segment.has_point(point_id, common::types::DeferredBehavior::WithDeferred));
Ok(true)
},
|point_id, _, _| {
|point_id, _, _, _| {
moved_to_appendable.push(point_id);
},
&HardwareCounterCell::new(),
@@ -187,7 +187,7 @@ fn test_apply_and_move_old_versions(
assert!(segment.has_point(point_id, common::types::DeferredBehavior::WithDeferred));
Ok(true)
},
|point_id, _, _| processed_points2.push(point_id),
|point_id, _, _, _| processed_points2.push(point_id),
&hw_counter,
)
.unwrap();
@@ -272,7 +272,7 @@ fn test_cow_operation() {
1010,
&[123.into()],
|_, _| unreachable!(),
|_point_id, vectors, payload| {
|_point_id, _raw_vectors, vectors, payload| {
vectors.insert(
DEFAULT_VECTOR_NAME.to_owned(),
VectorInternal::Dense(vec![9.0; 4]),
@@ -300,6 +300,448 @@ fn test_cow_operation() {
);
}
/// TurboQuant-as-datatype vectors must survive repeated CoW moves between
/// segments without degrading.
///
/// The CoW arm of [`SegmentHolder::apply_points_with_conditional_move`] reads
/// the point's vectors decoded to `f32` and re-inserts them with
/// `upsert_point`, so a TQ-datatype vector is dequantized and requantized on
/// every move. TurboQuant requantization is not idempotent — for Dot/L2 the
/// stored per-vector scale factor picks up the centroid-norm bias on every
/// cycle — so each move degrades the vector a little further beyond the
/// initial (expected, one-off) quantization loss.
///
/// This test ping-pongs one point between two TQ segments, and asserts the
/// decoded vector never drifts from its first-generation value (the read-back
/// right after initial ingestion). The holder classifies segments as
/// (non-)appendable once, at insertion, so each round uses a fresh holder with
/// the segment roles swapped instead of flipping flags under a live holder.
#[test]
fn test_cow_move_does_not_degrade_turbo_vectors() {
use segment::data_types::vectors::only_default_vector;
use segment::types::{Indexes, VectorDataConfig, VectorStorageDatatype, VectorStorageType};
const DIM: usize = 128;
const ROUNDTRIPS: u64 = 32;
let 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: Default::default(),
payload_storage_type: Default::default(),
};
let dir = Builder::new().prefix("segment_dir").tempdir().unwrap();
let (mut segment_a, _) = build_segment(dir.path(), &config, None, true).unwrap();
let (segment_b, _) = build_segment(dir.path(), &config, None, true).unwrap();
let hw_counter = HardwareCounterCell::new();
let point_id: PointIdType = 7.into();
let original: Vec<f32> = (0..DIM).map(|i| (i as f32 * 0.37).sin()).collect();
segment_a
.upsert_point(100, point_id, only_default_vector(&original), &hw_counter)
.unwrap();
let read_dense = |segment: &dyn SegmentEntry| -> Vec<f32> {
match segment
.vector(DEFAULT_VECTOR_NAME, point_id, &hw_counter)
.unwrap()
.unwrap()
{
VectorInternal::Dense(vector) => vector,
VectorInternal::Sparse(_) | VectorInternal::MultiDense(_) => {
panic!("expected a dense vector")
}
}
};
// First-generation read-back: the original vector after its initial
// (expected, one-off) quantization. Moves must preserve it exactly.
let first_generation = read_dense(&segment_a);
let segment_a = Arc::new(RwLock::new(segment_a));
let segment_b = Arc::new(RwLock::new(segment_b));
// L2 distance of each round's read-back from the first generation.
let mut drift = Vec::new();
for round in 0..ROUNDTRIPS {
let (source, destination) = if round % 2 == 0 {
(&segment_a, &segment_b)
} else {
(&segment_b, &segment_a)
};
// The segment holding the point is non-appendable and the other one is
// the only appendable destination, so the call must take the CoW-move
// arm.
source.write().appendable_flag = false;
destination.write().appendable_flag = true;
let mut holder = SegmentHolder::default();
holder.add_new_locked(LockedSegment::Original(source.clone()));
holder.add_new_locked(LockedSegment::Original(destination.clone()));
holder
.apply_points_with_conditional_move(
101 + round,
&[point_id],
|_, _| unreachable!("the point's segment is non-appendable, it must be moved"),
|_, _, _, _| {}, // no-op: a pure move
&hw_counter,
)
.unwrap();
let source = source.read();
let destination = destination.read();
assert!(!source.has_point(point_id, DeferredBehavior::WithDeferred));
assert!(destination.has_point(point_id, DeferredBehavior::WithDeferred));
let read_back = read_dense(&*destination);
drift.push(
first_generation
.iter()
.zip(&read_back)
.map(|(&a, &b)| (a - b).powi(2))
.sum::<f32>()
.sqrt(),
);
}
assert!(
drift.iter().all(|&distance| distance == 0.0),
"TurboQuant vector degraded across CoW moves; L2 distance from the \
first-generation read-back after each move: {drift:?}",
);
}
/// A CoW move where the operation overlays one named vector must re-encode
/// only that name: the untouched sibling TQ vector travels as verbatim bytes
/// (bit-identical read-back), and the overlaid name reads back exactly like a
/// fresh direct ingest of the same data.
#[test]
fn test_cow_move_overlay_preserves_untouched_turbo_vector() {
use segment::types::{Indexes, VectorDataConfig, VectorStorageDatatype, VectorStorageType};
const DIM: usize = 128;
const KEEP: &str = "keep";
const REPLACE: &str = "replace";
let vector_config = VectorDataConfig {
size: DIM,
distance: Distance::Dot,
storage_type: VectorStorageType::ChunkedMmap,
index: Indexes::Plain {},
quantization_config: None,
multivector_config: None,
datatype: Some(VectorStorageDatatype::Turbo4),
};
let config = SegmentConfig {
vector_data: HashMap::from([
(KEEP.to_owned(), vector_config.clone()),
(REPLACE.to_owned(), vector_config),
]),
sparse_vector_data: Default::default(),
payload_storage_type: Default::default(),
};
let dir = Builder::new().prefix("segment_dir").tempdir().unwrap();
let (mut source, _) = build_segment(dir.path(), &config, None, true).unwrap();
let (destination, _) = build_segment(dir.path(), &config, None, true).unwrap();
let hw_counter = HardwareCounterCell::new();
let point_id: PointIdType = 7.into();
let keep_vec: Vec<f32> = (0..DIM).map(|i| (i as f32 * 0.37).sin()).collect();
let old_replace: Vec<f32> = (0..DIM).map(|i| (i as f32 * 0.11).cos()).collect();
let fresh_replace: Vec<f32> = (0..DIM).map(|i| (i as f32 * 0.73).sin()).collect();
source
.upsert_point(
100,
point_id,
NamedVectors::from_pairs([
(KEEP.to_owned(), keep_vec.clone()),
(REPLACE.to_owned(), old_replace.clone()),
]),
&hw_counter,
)
.unwrap();
let read_dense = |segment: &dyn SegmentEntry, name: &str| -> Vec<f32> {
match segment
.vector(name, point_id, &hw_counter)
.unwrap()
.unwrap()
{
VectorInternal::Dense(vector) => vector,
VectorInternal::Sparse(_) | VectorInternal::MultiDense(_) => {
panic!("expected a dense vector")
}
}
};
// First-generation read-back of the untouched name: must survive the move
// bit-identically.
let first_generation_keep = read_dense(&source, KEEP);
// Oracle for the overlaid name: a fresh direct ingest into a same-config
// segment (TQ encoding is deterministic for equal configs).
let (mut oracle, _) = build_segment(dir.path(), &config, None, true).unwrap();
oracle
.upsert_point(
100,
point_id,
NamedVectors::from_pairs([
(KEEP.to_owned(), keep_vec.clone()),
(REPLACE.to_owned(), fresh_replace.clone()),
]),
&hw_counter,
)
.unwrap();
let expected_replace = read_dense(&oracle, REPLACE);
// Same pattern as test_cow_move_does_not_degrade_turbo_vectors: keep own
// Arc handles so the read-back guard derefs to &Segment (→ &dyn SegmentEntry).
source.appendable_flag = false;
let source = Arc::new(RwLock::new(source));
let destination = Arc::new(RwLock::new(destination));
let mut holder = SegmentHolder::default();
holder.add_new_locked(LockedSegment::Original(source.clone()));
holder.add_new_locked(LockedSegment::Original(destination.clone()));
holder
.apply_points_with_conditional_move(
101,
&[point_id],
|_, _| unreachable!("the point's segment is non-appendable, it must be moved"),
|_, _raw_vectors, updated_vectors, _| {
updated_vectors.insert(
REPLACE.to_owned(),
VectorInternal::Dense(fresh_replace.clone()),
);
},
&hw_counter,
)
.unwrap();
let destination = destination.read();
assert!(destination.has_point(point_id, DeferredBehavior::WithDeferred));
assert_eq!(
read_dense(&*destination, KEEP),
first_generation_keep,
"untouched named vector must travel as verbatim bytes",
);
assert_eq!(
read_dense(&*destination, REPLACE),
expected_replace,
"overlaid named vector must equal a fresh direct ingest",
);
}
/// A CoW move where the operation deletes one named vector must drop exactly
/// that name at the destination while the surviving TQ vector travels as
/// verbatim bytes (bit-identical read-back).
#[test]
fn test_cow_move_delete_name_preserves_survivor() {
use segment::types::{Indexes, VectorDataConfig, VectorStorageDatatype, VectorStorageType};
const DIM: usize = 128;
const KEEP: &str = "keep";
const DROP: &str = "drop";
let vector_config = VectorDataConfig {
size: DIM,
distance: Distance::Dot,
storage_type: VectorStorageType::ChunkedMmap,
index: Indexes::Plain {},
quantization_config: None,
multivector_config: None,
datatype: Some(VectorStorageDatatype::Turbo4),
};
let config = SegmentConfig {
vector_data: HashMap::from([
(KEEP.to_owned(), vector_config.clone()),
(DROP.to_owned(), vector_config),
]),
sparse_vector_data: Default::default(),
payload_storage_type: Default::default(),
};
let dir = Builder::new().prefix("segment_dir").tempdir().unwrap();
let (mut source, _) = build_segment(dir.path(), &config, None, true).unwrap();
let (destination, _) = build_segment(dir.path(), &config, None, true).unwrap();
let hw_counter = HardwareCounterCell::new();
let point_id: PointIdType = 7.into();
let keep_vec: Vec<f32> = (0..DIM).map(|i| (i as f32 * 0.37).sin()).collect();
let drop_vec: Vec<f32> = (0..DIM).map(|i| (i as f32 * 0.11).cos()).collect();
source
.upsert_point(
100,
point_id,
NamedVectors::from_pairs([
(KEEP.to_owned(), keep_vec.clone()),
(DROP.to_owned(), drop_vec.clone()),
]),
&hw_counter,
)
.unwrap();
let read_dense = |segment: &dyn SegmentEntry, name: &str| -> Vec<f32> {
match segment
.vector(name, point_id, &hw_counter)
.unwrap()
.unwrap()
{
VectorInternal::Dense(vector) => vector,
VectorInternal::Sparse(_) | VectorInternal::MultiDense(_) => {
panic!("expected a dense vector")
}
}
};
let first_generation_keep = read_dense(&source, KEEP);
// Own Arc handles, as in test_cow_move_does_not_degrade_turbo_vectors.
source.appendable_flag = false;
let source = Arc::new(RwLock::new(source));
let destination = Arc::new(RwLock::new(destination));
let mut holder = SegmentHolder::default();
holder.add_new_locked(LockedSegment::Original(source.clone()));
holder.add_new_locked(LockedSegment::Original(destination.clone()));
holder
.apply_points_with_conditional_move(
101,
&[point_id],
|_, _| unreachable!("the point's segment is non-appendable, it must be moved"),
|_, raw_vectors, _, _| {
raw_vectors.retain(|(name, _)| name != DROP);
},
&hw_counter,
)
.unwrap();
let destination = destination.read();
assert!(destination.has_point(point_id, DeferredBehavior::WithDeferred));
assert_eq!(
read_dense(&*destination, KEEP),
first_generation_keep,
"surviving named vector must travel as verbatim bytes",
);
assert!(
destination
.vector(DROP, point_id, &hw_counter)
.unwrap()
.is_none(),
"deleted named vector must not exist at the destination",
);
}
/// A CoW move between segments whose vector configs differ only in segment
/// role fields (storage type, index, quantization) must succeed — that is the
/// normal shape of a move out of an optimizer-built segment into an appendable
/// one. Only encoding-relevant fields (size, distance, datatype, multivector
/// config) must match for raw bytes to be portable.
///
/// Regression test: the CoW arm's config guard compared full
/// `VectorDataConfig` structs and panicked (debug builds) on every move out of
/// an optimized segment.
#[test]
fn test_cow_move_allows_role_config_differences() {
use segment::types::{Indexes, VectorDataConfig, VectorStorageType};
const DIM: usize = 128;
let make_config = |storage_type: VectorStorageType| SegmentConfig {
vector_data: HashMap::from([(
DEFAULT_VECTOR_NAME.to_owned(),
VectorDataConfig {
size: DIM,
distance: Distance::Dot,
storage_type,
index: Indexes::Plain {},
quantization_config: None,
multivector_config: None,
datatype: None,
},
)]),
sparse_vector_data: Default::default(),
payload_storage_type: Default::default(),
};
let dir = Builder::new().prefix("segment_dir").tempdir().unwrap();
// Source and destination differ in storage type only; the byte encoding
// (f32 dense) is identical on both sides.
let (mut source, _) = build_segment(
dir.path(),
&make_config(VectorStorageType::InRamChunkedMmap),
None,
true,
)
.unwrap();
let (destination, _) = build_segment(
dir.path(),
&make_config(VectorStorageType::ChunkedMmap),
None,
true,
)
.unwrap();
let hw_counter = HardwareCounterCell::new();
let point_id: PointIdType = 7.into();
let original: Vec<f32> = (0..DIM).map(|i| (i as f32 * 0.37).sin()).collect();
source
.upsert_point(
100,
point_id,
segment::data_types::vectors::only_default_vector(&original),
&hw_counter,
)
.unwrap();
// Own Arc handles, as in test_cow_move_does_not_degrade_turbo_vectors.
source.appendable_flag = false;
let source = Arc::new(RwLock::new(source));
let destination = Arc::new(RwLock::new(destination));
let mut holder = SegmentHolder::default();
holder.add_new_locked(LockedSegment::Original(source.clone()));
holder.add_new_locked(LockedSegment::Original(destination.clone()));
holder
.apply_points_with_conditional_move(
101,
&[point_id],
|_, _| unreachable!("the point's segment is non-appendable, it must be moved"),
|_, _, _, _| {}, // no-op: a pure move
&hw_counter,
)
.unwrap();
let destination = destination.read();
assert!(destination.has_point(point_id, DeferredBehavior::WithDeferred));
match destination
.vector(DEFAULT_VECTOR_NAME, point_id, &hw_counter)
.unwrap()
.unwrap()
{
VectorInternal::Dense(vector) => assert_eq!(
vector, original,
"f32 dense vector must travel losslessly across role-config differences",
),
VectorInternal::Sparse(_) | VectorInternal::MultiDense(_) => {
panic!("expected a dense vector")
}
}
}
#[test]
fn test_points_deduplication() {
let dir = Builder::new().prefix("segment_dir").tempdir().unwrap();
@@ -1498,7 +1940,7 @@ fn test_cow_skips_delete_when_destination_is_deferred() {
20,
&[100.into()],
|_, _| unreachable!("point is in non-appendable, should take CoW path"),
|_, _, _| {},
|_, _, _, _| {},
&hw_counter,
)
.unwrap();
@@ -1665,7 +2107,7 @@ fn test_cow_deletes_source_when_destination_is_not_deferred() {
20,
&[100.into()],
|_, _| unreachable!("point is in non-appendable, should take CoW path"),
|_, _, _| {},
|_, _, _, _| {},
&hw_counter,
)
.unwrap();

142
lib/shard/src/update.rs
View File

@@ -214,14 +214,15 @@ where
hw_counter,
)
},
|id, vectors, old_payload| {
|id, raw_vectors, updated_vectors, old_payload| {
let point = points_map[&id];
for (name, vec) in point.get_vectors() {
vectors.insert(name.into(), vec.to_owned());
}
if let Some(payload) = &point.payload {
*old_payload = payload.clone();
}
// Upsert replaces the whole point: old named vectors and
// payload must not survive the move, matching the in-place
// `upsert_with_payload` path (`replace_all_vectors` +
// clear/set payload).
raw_vectors.clear();
*updated_vectors = point.get_vectors();
*old_payload = point.payload.clone().unwrap_or_default();
},
hw_counter,
)?;
@@ -670,9 +671,9 @@ fn update_vectors(
let vectors = points_map[&id].clone();
write_segment.update_vectors(op_num, id, vectors, hw_counter)
},
|id, owned_vectors, _| {
|id, _raw_vectors, updated_vectors, _| {
for (vector_name, vector_ref) in points_map[&id].iter() {
owned_vectors.insert(vector_name.to_owned(), vector_ref.to_owned());
updated_vectors.insert_ref(vector_name, vector_ref);
}
},
hw_counter,
@@ -711,10 +712,8 @@ pub fn delete_vectors(
}
Ok(res)
},
|_, owned_vectors, _| {
for name in vector_names {
owned_vectors.remove_ref(name);
}
|_, raw_vectors, _, _| {
raw_vectors.retain(|(name, _)| !vector_names.contains(name));
},
hw_counter,
)?;
@@ -770,7 +769,7 @@ pub fn set_payload(
op_num,
chunk,
|id, write_segment| write_segment.set_payload(op_num, id, payload, key, hw_counter),
|_, _, old_payload| match key {
|_, _, _, old_payload| match key {
Some(key) => old_payload.merge_by_key(payload, key),
None => old_payload.merge(payload),
},
@@ -830,7 +829,7 @@ pub fn delete_payload(
}
Ok(res)
},
|_, _, payload| {
|_, _, _, payload| {
for key in keys {
payload.remove(key);
}
@@ -883,7 +882,7 @@ pub fn clear_payload(
op_num,
batch,
|id, write_segment| write_segment.clear_payload(op_num, id, hw_counter),
|_, _, payload| payload.0.clear(),
|_, _, _, payload| payload.0.clear(),
hw_counter,
)?;
check_unprocessed_points(batch, &updated_points)?;
@@ -932,7 +931,7 @@ pub fn overwrite_payload(
op_num,
batch,
|id, write_segment| write_segment.set_full_payload(op_num, id, payload, hw_counter),
|_, _, old_payload| {
|_, _, _, old_payload| {
*old_payload = payload.clone();
},
hw_counter,
@@ -1684,4 +1683,113 @@ mod test {
"Deferred copy must be kept"
);
}
/// Upsert of a point that lives in a non-appendable segment takes the
/// CoW-move arm of `apply_points_with_conditional_move`. The moved record
/// must match in-place upsert semantics (`upsert_with_payload`): named
/// vectors and payload absent from the incoming point are dropped, not
/// carried over from the old record.
#[test]
fn test_upsert_cow_move_replaces_whole_point() {
use std::collections::HashMap;
use common::types::DeferredBehavior;
use segment::data_types::named_vectors::NamedVectors;
use segment::entry::entry_point::SegmentEntry;
use segment::segment_constructor::simple_segment_constructor::{
VECTOR1_NAME, VECTOR2_NAME, build_segment_with_two_named_vecs,
};
use segment::types::{Distance, PointIdType};
use crate::operations::point_ops::{
PointStructPersisted, VectorPersisted, VectorStructPersisted,
};
use crate::update::upsert_points;
const DIM: usize = 4;
let dir = Builder::new().prefix("segment_dir").tempdir().unwrap();
let hw_counter = HardwareCounterCell::new();
let point_id: PointIdType = 7.into();
// Old record: both named vectors plus a payload.
let seed = |segment: &mut segment::segment::Segment| {
segment
.upsert_point(
100,
point_id,
NamedVectors::from_pairs([
(VECTOR1_NAME.to_owned(), vec![0.1, 0.2, 0.3, 0.4]),
(VECTOR2_NAME.to_owned(), vec![0.5, 0.6, 0.7, 0.8]),
]),
&hw_counter,
)
.unwrap();
segment
.set_payload(
100,
point_id,
&payload_json! {"city": "Berlin"},
&None,
&hw_counter,
)
.unwrap();
};
// Incoming upsert: only `vector1`, no payload.
let incoming = PointStructPersisted {
id: point_id,
vector: VectorStructPersisted::Named(HashMap::from([(
VECTOR1_NAME.to_owned(),
VectorPersisted::Dense(vec![1.0, 1.0, 1.0, 1.0]),
)])),
payload: None,
};
let check = |segment: &dyn SegmentEntry, path: &str| {
assert!(segment.has_point(point_id, DeferredBehavior::WithDeferred));
assert!(
segment
.vector(VECTOR1_NAME, point_id, &hw_counter)
.unwrap()
.is_some(),
"{path}: upserted vector must be present",
);
assert!(
segment
.vector(VECTOR2_NAME, point_id, &hw_counter)
.unwrap()
.is_none(),
"{path}: named vector absent from the upsert must be dropped",
);
assert!(
segment.payload(point_id, &hw_counter).unwrap().is_empty(),
"{path}: payload absent from the upsert must be cleared",
);
};
// Oracle: in-place upsert into an appendable segment.
let mut in_place =
build_segment_with_two_named_vecs(dir.path(), DIM, DIM, Distance::Dot).unwrap();
seed(&mut in_place);
let mut holder = SegmentHolder::default();
let sid = holder.add_new(in_place);
upsert_points(&holder, 101, [&incoming], &hw_counter).unwrap();
let segment = holder.get(sid).unwrap().get();
check(&*segment.read(), "in-place");
// CoW move: the point's segment is non-appendable, so the upsert must
// move it into the appendable destination with the same semantics.
let mut source =
build_segment_with_two_named_vecs(dir.path(), DIM, DIM, Distance::Dot).unwrap();
seed(&mut source);
source.appendable_flag = false;
let destination =
build_segment_with_two_named_vecs(dir.path(), DIM, DIM, Distance::Dot).unwrap();
let mut holder = SegmentHolder::default();
holder.add_new(source);
let sid = holder.add_new(destination);
upsert_points(&holder, 101, [&incoming], &hw_counter).unwrap();
let segment = holder.get(sid).unwrap().get();
check(&*segment.read(), "CoW move");
}
}