diff --git a/lib/segment/src/data_types/named_vectors.rs b/lib/segment/src/data_types/named_vectors.rs index 1c28b8e9d9..308010ae20 100644 --- a/lib/segment/src/data_types/named_vectors.rs +++ b/lib/segment/src/data_types/named_vectors.rs @@ -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); } diff --git a/lib/segment/src/entry/entry_point.rs b/lib/segment/src/entry/entry_point.rs index 030fa025d5..b86c554584 100644 --- a/lib/segment/src/entry/entry_point.rs +++ b/lib/segment/src/entry/entry_point.rs @@ -87,19 +87,6 @@ pub trait ReadSegmentEntry { hw_counter: &HardwareCounterCell, ) -> OperationResult>; - /// 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>; - /// 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; - /// 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; - /// Paginate over points which satisfies filtering condition starting with `offset` id including. /// /// Cancelled by `is_stopped` flag. diff --git a/lib/segment/src/segment/entry.rs b/lib/segment/src/segment/entry.rs index 1b03003060..a3f4188ada 100644 --- a/lib/segment/src/segment/entry.rs +++ b/lib/segment/src/segment/entry.rs @@ -104,22 +104,16 @@ impl ReadSegmentEntry for Segment { &self, point_id: PointIdType, hw_counter: &HardwareCounterCell, - ) -> OperationResult> { - 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> { 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 { - self.with_view(|view| view.payload_with_behavior(point_id, deferred_behavior, hw_counter)) - } - fn retrieve( &self, point_ids: &[PointIdType], diff --git a/lib/segment/src/segment/read_only/read_entry.rs b/lib/segment/src/segment/read_only/read_entry.rs index 9852bc2408..dbed7461d9 100644 --- a/lib/segment/src/segment/read_only/read_entry.rs +++ b/lib/segment/src/segment/read_only/read_entry.rs @@ -100,22 +100,16 @@ impl ReadSegmentEntry for ReadOnlySegment { &self, point_id: PointIdType, hw_counter: &HardwareCounterCell, - ) -> OperationResult> { - 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> { 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 ReadSegmentEntry for ReadOnlySegment { 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 { - self.with_view(|view| view.payload_with_behavior(point_id, deferred_behavior, hw_counter)) - } - fn retrieve( &self, point_ids: &[PointIdType], diff --git a/lib/segment/src/segment/tests/mod.rs b/lib/segment/src/segment/tests/mod.rs index e444718cff..a27eb58d57 100644 --- a/lib/segment/src/segment/tests/mod.rs +++ b/lib/segment/src/segment/tests/mod.rs @@ -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", ); } diff --git a/lib/segment/src/vector_storage/vector_storage_base.rs b/lib/segment/src/vector_storage/vector_storage_base.rs index bb69619012..6812546b04 100644 --- a/lib/segment/src/vector_storage/vector_storage_base.rs +++ b/lib/segment/src/vector_storage/vector_storage_base.rs @@ -876,10 +876,16 @@ fn insert_dense_bytes + bytes.len(), ))); } - // `pod_collect_to_vec` instead of `cast_slice`: incoming bytes may not be - // aligned for `T`. - let elements: Vec = 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 + bytes.len(), ))); } - let elements: Vec = 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) } diff --git a/lib/shard/src/proxy_segment/segment_entry.rs b/lib/shard/src/proxy_segment/segment_entry.rs index 97437c1abf..b4609b4e4e 100644 --- a/lib/shard/src/proxy_segment/segment_entry.rs +++ b/lib/shard/src/proxy_segment/segment_entry.rs @@ -305,15 +305,6 @@ impl ReadSegmentEntry for ProxySegment { &self, point_id: PointIdType, hw_counter: &HardwareCounterCell, - ) -> OperationResult> { - 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> { 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 { - 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 { 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) } } diff --git a/lib/shard/src/segment_holder/mod.rs b/lib/shard/src/segment_holder/mod.rs index 202efa7425..b677e01e26 100644 --- a/lib/shard/src/segment_holder/mod.rs +++ b/lib/shard/src/segment_holder/mod.rs @@ -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( + pub fn apply_points_with_conditional_move<'op, F, G>( &self, op_num: SeqNumberType, ids: &[PointIdType], @@ -983,12 +994,18 @@ impl SegmentHolder { ) -> OperationResult> where F: FnMut(PointIdType, &mut RwLockWriteGuard) -> OperationResult, - for<'n, 'o, 'p> G: FnMut(PointIdType, &'n mut NamedVectors<'o>, &'p mut Payload), + G: FnMut( + PointIdType, + &mut Vec<(VectorNameBuf, Vec)>, + &mut NamedVectors<'op>, + &mut Payload, + ), { // Choose random appendable segment from this let appendable_segments = self.appendable_segments_ids(); let mut applied_points: AHashSet = 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)?; diff --git a/lib/shard/src/segment_holder/tests.rs b/lib/shard/src/segment_holder/tests.rs index 0686a648c5..3787fb20c5 100644 --- a/lib/shard/src/segment_holder/tests.rs +++ b/lib/shard/src/segment_holder/tests.rs @@ -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 = (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 { + 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::() + .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 = (0..DIM).map(|i| (i as f32 * 0.37).sin()).collect(); + let old_replace: Vec = (0..DIM).map(|i| (i as f32 * 0.11).cos()).collect(); + let fresh_replace: Vec = (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 { + 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 = (0..DIM).map(|i| (i as f32 * 0.37).sin()).collect(); + let drop_vec: Vec = (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 { + 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 = (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(); diff --git a/lib/shard/src/update.rs b/lib/shard/src/update.rs index 061c9dbee6..fca97cbee8 100644 --- a/lib/shard/src/update.rs +++ b/lib/shard/src/update.rs @@ -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"); + } }