[TQDT] Roundtrip fix for clone_and_mutate_point (#9761)

* Fix TQDT roundtrip issue for `clone_and_mutate_point`

* Use TinyMap

* Fix doc string

* Rebase docstring fixes

* [TQDT] Reduce allocations (#9833)

* Reduce allocations

* Fix edge

lib/segment/src/data_types/segment_record.rs

@@ -1,7 +1,11 @@
+use smallvec::SmallVec;
+
 use crate::data_types::vectors::VectorInternal;
 use crate::types::{Payload, PointIdType, VectorNameBuf};
 
-pub type NamedVectorsOwned = Vec<(VectorNameBuf, VectorInternal)>;
+/// A point almost always has a single (default) named vector, so keep it inline
+/// to avoid a heap allocation on the common retrieve path.
+pub type NamedVectorsOwned = SmallVec<[(VectorNameBuf, VectorInternal); 1]>;
 
 /// A retrieved point: id, optional vectors, optional payload.
 ///
@@ -9,7 +13,7 @@ pub type NamedVectorsOwned = Vec<(VectorNameBuf, VectorInternal)>;
 /// storage-native bytes for [`SegmentRecordRaw`].
 pub struct SegmentRecordGeneric<V> {
     pub id: PointIdType,
-    pub vectors: Option<Vec<(VectorNameBuf, V)>>,
+    pub vectors: Option<SmallVec<[(VectorNameBuf, V); 1]>>,
     pub payload: Option<Payload>,
 }
 

lib/segment/src/segment/entry.rs

@@ -766,8 +766,9 @@ impl SegmentEntry for Segment {
                     segment.replace_all_vectors(internal_id, op_num, &vectors, hw_counter)?;
                     Ok(true)
                 },
-                |snapshot_vectors, _payload| {
-                    *snapshot_vectors = vectors.clone().into_owned();
+                |raw_vectors, updated_vectors, _payload| {
+                    raw_vectors.clear();
+                    *updated_vectors = vectors.clone().into_owned();
                     Ok(true)
                 },
             ),
@@ -797,11 +798,9 @@ impl SegmentEntry for Segment {
             .borrow()
             .internal_id_with_behavior(point_id, DeferredBehavior::WithDeferred);
         match stored_internal_point {
-            // Not `handle_point_mutate`: its append-only arm snapshots the old
-            // vectors into decoded `NamedVectors`, which cannot carry
-            // storage-native bytes (and would be a lossy read for TurboQuant).
-            // The raw clone path skips the vector snapshot entirely — upsert
-            // discards all old vectors anyway.
+            // Not `handle_point_mutate`: upsert replaces the whole point, so
+            // the raw clone path skips the vector snapshot entirely instead
+            // of reading raw bytes it would immediately discard.
             Some(existing_internal_id) => {
                 let append_only = self.is_append_only();
                 self.handle_point_version_and_failure(
@@ -936,8 +935,8 @@ impl SegmentEntry for Segment {
                 segment.update_vectors(internal_id, op_num, vectors.clone(), hw_counter)?;
                 Ok(true)
             },
-            |snapshot_vectors, _payload| {
-                snapshot_vectors.merge(vectors.clone().into_owned());
+            |_raw_vectors, updated_vectors, _payload| {
+                *updated_vectors = vectors.clone().into_owned();
                 Ok(true)
             },
         )
@@ -982,8 +981,8 @@ impl SegmentEntry for Segment {
                 let mut vector_storage = vector_data.vector_storage.borrow_mut();
                 vector_storage.delete_vector(internal_id)
             },
-            |snapshot_vectors, _payload| {
-                snapshot_vectors.remove_ref(vector_name);
+            |raw_vectors, _updated_vectors, _payload| {
+                raw_vectors.remove(vector_name);
                 Ok(was_present)
             },
         )?;
@@ -1028,7 +1027,7 @@ impl SegmentEntry for Segment {
                 segment.version_tracker.set_payload(Some(op_num));
                 Ok(true)
             },
-            |_vectors, snapshot_payload| {
+            |_raw_vectors, _updated_vectors, snapshot_payload| {
                 *snapshot_payload = full_payload.clone();
                 Ok(true)
             },
@@ -1067,7 +1066,7 @@ impl SegmentEntry for Segment {
                 segment.version_tracker.set_payload(Some(op_num));
                 Ok(true)
             },
-            |_vectors, snapshot_payload| {
+            |_raw_vectors, _updated_vectors, snapshot_payload| {
                 match key {
                     Some(k) => snapshot_payload.merge_by_key(payload, k),
                     None => snapshot_payload.merge(payload),
@@ -1106,7 +1105,7 @@ impl SegmentEntry for Segment {
                 segment.version_tracker.set_payload(Some(op_num));
                 Ok(true)
             },
-            |_vectors, snapshot_payload| {
+            |_raw_vectors, _updated_vectors, snapshot_payload| {
                 snapshot_payload.remove(key);
                 Ok(true)
             },
@@ -1141,7 +1140,7 @@ impl SegmentEntry for Segment {
                 segment.version_tracker.set_payload(Some(op_num));
                 Ok(true)
             },
-            |_vectors, snapshot_payload| {
+            |_raw_vectors, _updated_vectors, snapshot_payload| {
                 *snapshot_payload = Payload::default();
                 Ok(true)
             },

lib/segment/src/segment/read_view/search.rs

@@ -5,6 +5,7 @@ use common::counter::hardware_counter::HardwareCounterCell;
 use common::generic_consts::Random;
 use common::iterator_ext::IteratorExt;
 use common::types::{DeferredBehavior, PointOffsetType, ScoredPointOffset};
+use smallvec::SmallVec;
 
 use crate::common::operation_error::{OperationError, OperationResult};
 use crate::common::{check_query_vectors, check_stopped};
@@ -119,7 +120,7 @@ where
             .map(|&id| {
                 let record = SegmentRecordGeneric {
                     id,
-                    vectors: needs_vectors.then(Vec::new),
+                    vectors: needs_vectors.then(SmallVec::new),
                     payload: None,
                 };
                 (id, record)

lib/segment/src/segment/segment_ops.rs

@@ -19,6 +19,7 @@ use crate::common::operation_error::{
 };
 use crate::common::{check_named_vectors, check_vector_name};
 use crate::data_types::named_vectors::NamedVectors;
+use crate::data_types::tiny_map::TinyMap;
 use crate::entry::entry_point::StorageSegmentEntry as _;
 use crate::entry::{NonAppendableSegmentEntry as _, ReadSegmentEntry};
 use crate::id_tracker::{IdTracker, IdTrackerRead};
@@ -271,23 +272,30 @@ impl Segment {
         Ok(())
     }
 
-    /// Append-only update: snapshot the point at `old_id` into owned vectors
-    /// and payload, hand them to `mutate` for in-memory modification, then
-    /// write the result at a fresh internal id and repoint the id tracker.
+    /// Append-only update: snapshot the point at `old_id` into owned raw
+    /// vectors and payload, hand them to `mutate` for in-memory modification,
+    /// then write the result at a fresh internal id and repoint the id
+    /// tracker.
     ///
     /// Step order:
     ///
-    /// 1. Read all named vectors at `old_id` into an owned `NamedVectors`,
-    ///    and the full payload at `old_id` into an owned `Payload`.
-    /// 2. Call `mutate(&mut NamedVectors, &mut Payload)` to apply the
-    ///    op-specific change in memory; its return value is propagated to
-    ///    the caller alongside the new internal id.
+    /// 1. Read all named vectors at `old_id` as storage-native bytes into an
+    ///    owned name → bytes map, and the full payload at `old_id` into an
+    ///    owned `Payload`.
+    /// 2. Call `mutate(&mut raw_vectors, &mut updated_vectors, &mut payload)`
+    ///    to apply the op-specific change in memory: drop raw entries to
+    ///    delete names, insert decoded vectors into the (initially empty)
+    ///    `updated_vectors` overlay to overwrite names. The return value is
+    ///    propagated to the caller alongside the new internal id.
     /// 3. Allocate `new_id = total_point_count()`.
-    /// 4. Write the mutated vectors at `new_id`. Every configured named
-    ///    vector storage gets touched: present entries are written, absent
-    ///    ones are inserted as `None` (so the slot is grown and marked
-    ///    deleted, matching `insert_new_vectors`'s behavior).
-    /// 5. Write the mutated payload at `new_id` (skipped if empty).
+    /// 4. Write every configured named vector at `new_id`, overlay first:
+    ///    overlaid names are written decoded, remaining raw entries are
+    ///    ingested verbatim — lossless for requantizing storages
+    ///    (TurboQuant-as-datatype) — and names in neither are inserted as
+    ///    `None` (so the slot is grown and marked deleted, matching
+    ///    `insert_new_vectors`'s behavior).
+    /// 5. Write the mutated payload at `new_id` — always, even when empty,
+    ///    so every field index covers `new_id` (see the body comment).
     /// 6. `set_link(point_id, new_id)` — auto-tombstones `old_id` in the id
     ///    tracker so it becomes invisible to queries.
     ///
@@ -306,7 +314,9 @@ impl Segment {
     ///
     /// Available for appendable segments only. Callers route into this
     /// helper from the `SegmentEntry` mutation paths when
-    /// [`Segment::is_append_only`] is true.
+    /// [`Segment::is_append_only`] is true — except for slots written by the
+    /// current operation, which are mutated in place instead (see
+    /// [`Segment::handle_point_mutate`]).
     pub(super) fn clone_and_mutate_point<F, R>(
         &mut self,
         op_num: SeqNumberType,
@@ -316,45 +326,63 @@ impl Segment {
         mutate: F,
     ) -> OperationResult<(R, PointOffsetType)>
     where
-        F: FnOnce(&mut NamedVectors<'static>, &mut Payload) -> OperationResult<R>,
+        F: FnOnce(
+            &mut TinyMap<VectorNameBuf, Vec<u8>>,
+            &mut NamedVectors<'static>,
+            &mut Payload,
+        ) -> OperationResult<R>,
     {
         debug_assert!(self.is_appendable());
 
         // 1. Snapshot vectors and payload at old_id into owned containers,
-        //    dropping all storage borrows before we start writing. Slots
-        //    that are marked deleted in the per-vector bitslice carry
-        //    leftover default-vector bytes from the original
+        //    dropping all storage borrows before we start writing. Vectors
+        //    are read as storage-native bytes: names the operation does not
+        //    overwrite travel to new_id verbatim, avoiding the lossy
+        //    dequantize→requantize round-trip of TurboQuant-as-datatype
+        //    storages. Slots that are marked deleted in the per-vector
+        //    bitslice carry leftover default-vector bytes from the original
         //    `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).
-        let mut vectors: NamedVectors<'static> = NamedVectors::default();
+        //    and write `None` at new_id (re-tombstoning the slot). `TinyMap`
+        //    keeps the container inline (no allocation), like the decoded
+        //    `NamedVectors` snapshot this replaced.
+        let mut raw_vectors: TinyMap<VectorNameBuf, Vec<u8>> = TinyMap::new();
         for (vector_name, vector_data) in self.vector_data.iter() {
             let storage = vector_data.vector_storage.borrow();
             if storage.is_deleted_vector(old_id) {
                 continue;
             }
-            if let Some(existing) = storage.get_vector_opt::<Random>(old_id) {
-                vectors.insert(vector_name.clone(), existing.to_owned());
+            if let Some(bytes) = storage.vector_bytes_opt::<Random>(old_id)? {
+                raw_vectors.insert(vector_name.clone(), bytes);
             }
         }
+        let mut updated_vectors: NamedVectors<'static> = NamedVectors::default();
         let mut payload = self
             .payload_index
             .borrow()
             .with_view(|view| view.get_payload(old_id, hw_counter))?;
 
         // 2. Let the caller apply the op-specific change in memory.
-        let mutate_result = mutate(&mut vectors, &mut payload)?;
+        let mutate_result = mutate(&mut raw_vectors, &mut updated_vectors, &mut payload)?;
 
         // 3. Allocate the fresh internal id.
         let new_id = self.id_tracker.borrow().total_point_count() as PointOffsetType;
 
-        // 4. Write every configured named vector at new_id. Absent entries
+        // 4. Write every configured named vector at new_id, overlay first:
+        //    names the closure overwrote are written decoded, the rest are
+        //    ingested as their snapshotted bytes. Names in neither container
         //    are written as None so the storage slot grows in lockstep and
         //    is marked deleted, matching insert_new_vectors's contract.
         for (vector_name, vector_data) in self.vector_data.iter_mut() {
-            let vector_opt = vectors.get(vector_name);
             let mut vector_index = vector_data.vector_index.borrow_mut();
-            vector_index.update_vector(new_id, vector_opt, hw_counter)?;
+            match updated_vectors.get(vector_name) {
+                Some(vector) => vector_index.update_vector(new_id, Some(vector), hw_counter)?,
+                None => vector_index.update_vector_raw(
+                    new_id,
+                    raw_vectors.get(vector_name).map(Vec::as_slice),
+                    hw_counter,
+                )?,
+            }
             self.version_tracker.set_vector(vector_name, Some(op_num));
         }
 
@@ -392,11 +420,12 @@ impl Segment {
     ///   benefit from being moved), runs `in_place(&mut Segment, old_id)`.
     /// - On an [`Segment::is_append_only`] segment, routes through
     ///   [`Segment::clone_and_mutate_point`] with `snapshot_mutate`, which
-    ///   sees an owned snapshot of the point's vectors and payload and
-    ///   modifies it in memory; the helper writes the result at a fresh
-    ///   internal id and tombstones the old one. Exception: a slot written
-    ///   by the current operation (its version equals `op_num`) is mutated
-    ///   in place — it is not durable yet, so cloning it would only chain
+    ///   sees the point's vectors as a raw-bytes snapshot plus an empty
+    ///   decoded overlay, and the payload as an owned snapshot, and modifies
+    ///   them in memory; the helper writes the result at a fresh internal id
+    ///   and tombstones the old one. Exception: a slot written by the
+    ///   current operation (its version equals `op_num`) is mutated in
+    ///   place — it is not durable yet, so cloning it would only chain
     ///   dead slots for multi-step point writes.
     ///
     /// Both closures return the op-specific result bool (e.g. "was anything
@@ -417,7 +446,11 @@ impl Segment {
     ) -> OperationResult<bool>
     where
         InPlace: FnOnce(&mut Segment, PointOffsetType) -> OperationResult<bool>,
-        SnapshotMutate: FnOnce(&mut NamedVectors<'static>, &mut Payload) -> OperationResult<bool>,
+        SnapshotMutate: FnOnce(
+            &mut TinyMap<VectorNameBuf, Vec<u8>>,
+            &mut NamedVectors<'static>,
+            &mut Payload,
+        ) -> OperationResult<bool>,
     {
         // A slot whose version already equals `op_num` was written by the
         // current operation (an earlier step of a multi-step point write,

lib/segment/src/segment/tests/mod.rs

@@ -1513,6 +1513,115 @@ fn test_upsert_raw_multivec_turbo_bytes() {
     );
 }
 
+/// TurboQuant-as-datatype vectors must survive append-only mutations that
+/// don't touch them — e.g. payload-only ops — without degrading.
+///
+/// On an append-only segment every mutating op (except same-operation
+/// follow-up steps, which mutate the just-written slot in place) routes
+/// through `Segment::clone_and_mutate_point`, which rewrites the point at a
+/// fresh internal id. It used to snapshot the point's vectors decoded to `f32`,
+/// dequantizing and requantizing a TQ-datatype vector on every mutation,
+/// even one that only touched the payload. 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 op degraded the vector a
+/// little further beyond the initial (expected, one-off) quantization loss.
+/// Vectors the op doesn't overwrite now travel to the fresh id as
+/// storage-native bytes, verbatim; this test guards that.
+///
+/// This is the segment-level analogue of the segment-holder test
+/// `test_cow_move_does_not_degrade_turbo_vectors` (CoW moves between
+/// segments); here the clone-and-tombstone cycle happens within one segment.
+///
+/// The test applies repeated payload-only ops to one point on an append-only
+/// TQ segment and asserts the decoded vector never drifts from its
+/// first-generation value (the read-back right after initial ingestion).
+#[test]
+fn test_append_only_mutate_does_not_degrade_turbo_vectors() {
+    init_logger();
+    const DIM: usize = 128;
+    const ROUNDS: 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, _) = build_segment(dir.path(), &config, None, true).unwrap();
+    segment.append_only_mutations = true;
+
+    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
+        .upsert_point(100, point_id, only_default_vector(&original), &hw_counter)
+        .unwrap();
+
+    let read_dense = |segment: &Segment| -> 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. Payload ops must preserve it exactly.
+    let first_generation = read_dense(&segment);
+
+    let payload: Payload = serde_json::from_str(r#"{"color": "red"}"#).unwrap();
+
+    // L2 distance of each round's read-back from the first generation.
+    let mut drift = Vec::new();
+    for round in 0..ROUNDS {
+        let old_internal_id = segment
+            .with_view(|v| v.lookup_internal_id(point_id, DeferredBehavior::VisibleOnly))
+            .unwrap();
+        segment
+            .set_payload(101 + round, point_id, &payload, &None, &hw_counter)
+            .unwrap();
+
+        // The op must have taken the append-only clone-and-tombstone path
+        // (fresh internal id), not the in-place path.
+        let new_internal_id = segment
+            .with_view(|v| v.lookup_internal_id(point_id, DeferredBehavior::VisibleOnly))
+            .unwrap();
+        assert_ne!(old_internal_id, new_internal_id);
+
+        let read_back = read_dense(&segment);
+        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 append-only payload ops; L2 \
+         distance from the first-generation read-back after each op: {drift:?}",
+    );
+}
+
 /// Tests segment functions to ensure invalid requests do error
 #[test]
 fn test_vector_compatibility_checks() {

lib/shard/src/operations/point_ops.rs

@@ -13,7 +13,7 @@ use segment::data_types::named_vectors::NamedVectors;
 use segment::data_types::segment_record::SegmentRecord;
 use segment::data_types::vectors::{
     BatchVectorStructInternal, DEFAULT_VECTOR_NAME, DenseVector, MultiDenseVector,
-    MultiDenseVectorInternal, VectorInternal, VectorStructInternal,
+    MultiDenseVectorInternal, VectorInternal, VectorRef, VectorStructInternal,
 };
 use segment::types::{Filter, Payload, PointIdType, VectorNameBuf};
 use serde::{Deserialize, Serialize};
@@ -447,14 +447,14 @@ impl PointStructPersisted {
             return false;
         }
 
-        let self_vectors = self.get_vectors().into_owned_map();
+        let self_vectors = self.get_vectors();
 
         if let Some(segment_vectors) = vectors {
             if self_vectors.len() != segment_vectors.len() {
                 return false;
             }
             for (name, vec) in segment_vectors {
-                if self_vectors.get(name) != Some(vec) {
+                if self_vectors.get(name) != Some(VectorRef::from(vec)) {
                     return false;
                 }
             }

lib/shard/src/segment_holder/mod.rs

@@ -996,7 +996,7 @@ impl SegmentHolder {
         F: FnMut(PointIdType, &mut RwLockWriteGuard<dyn SegmentEntry>) -> OperationResult<bool>,
         G: FnMut(
             PointIdType,
-            &mut Vec<(VectorNameBuf, Vec<u8>)>,
+            &mut SmallVec<[(VectorNameBuf, Vec<u8>); 1]>,
             &mut NamedVectors<'op>,
             &mut Payload,
         ),