transfer: send raw payloads, behind feature flags (#10066)

A raw point can carry its payload as the byte blob it is stored as, mirroring
`PointStructRaw.raw_payload` on the internal gRPC API. The blob travels from the
sending node into the receiving node's WAL untouched, so the sender never parses
the payload it read and neither node builds a protobuf value tree for it.

It is parsed exactly once, where the operation is unpacked for apply
(`process_point_operation`), because that is the first place the parsed form is
actually needed: `set_full_payload` goes through the payload index, which cannot
be updated from bytes. The gRPC boundary therefore only checks the encoding tag
and rejects a point that sets both payload fields, the way the enclosing request
already rejects both `points` and `raw_points`.

Moving the parse onto the apply path makes its error classification load-bearing,
so a malformed blob is reported as `OperationError::MalformedPayloadBlob` — the
payload sibling of `MalformedVectorBlob`, mapped to `CollectionError::BadInput`
for the same reason: a bad blob that reached the WAL has to be skipped on replay
instead of crash-looping recovery.

Three consequences of the blob living that long are handled explicitly rather
than by convention:

- `decode_payload_raw` takes the blob only once it has parsed, so a failure
  leaves the point holding it instead of holding neither representation.
- `upsert_points_raw` and `sync_points_raw` refuse a point that still carries a
  blob. They read the parsed payload, so such a point would otherwise be stored
  with no payload at all, and a `debug_assert!` would not catch it in release.
- `is_equal_to` compares blob to stored blob as bytes. A differing encoding costs
  a redundant upsert on sync, never a skipped one.

The `raw_payload_transfer` bench measures the trade, per 100-point batch (one
transfer batch) at payloads of ~200 B / ~700 B / ~7 KB:

- Sender, storage bytes to wire: 16x / 37x / 113x faster. This is where the whole
  win is — no parse of the blob that was read, no value tree built.
- WAL encode: 5x / 11x / 25x faster, writing a byte string instead of a map.
- Receiver, wire to applicable point: 1.09x / 1.10x / 1.06x. Near neutral, as it
  swaps walking a prost value tree for a JSON parse.
- Wire bytes: ~6% smaller. WAL bytes: 10-32% *larger*, because the blob is JSON
  while a parsed payload is written as a compact CBOR map.

The WAL growth is accepted rather than fixed: decoding earlier to win those bytes
back costs a second full deserialization, and would leave the receiving side with
a `payload_raw` that is never populated. Making the blob itself compact belongs in
the payload storage encoding (`RawPayloadEncoding` is the extension point for it),
not here.

Two flags, both off by default and both sender-only (nodes accept raw points and
raw payloads regardless), read where the transfer batch is prepared:

- `transfer_raw_points` transfers every collection as raw points, not only those
  whose vector storage would drift in a decode-encode round-trip.
- `transfer_raw_payloads` ships the blob a raw read hands out; without it the
  prepared batch decodes it back into the parsed payload, and the wire message is
  exactly what it is today.

Neither is enabled by `all`: a node only accepts them once it runs a version that
understands them, so they can only be switched on a release later. Nothing
enforces that yet — the transfer has no peer-version gate.

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>

Cargo.lock

@@ -7481,6 +7481,7 @@ dependencies = [
  "ordered-float 5.3.0",
  "parking_lot",
  "proptest",
+ "prost 0.14.4",
  "rand 0.10.2",
  "rmp-serde",
  "rstest",

docs/redoc/master/openapi.json

@@ -11859,6 +11859,16 @@
             "default": false,
             "type": "boolean"
           },
+          "transfer_raw_points": {
+            "description": "Transfer points as storage-native bytes (raw points), for every collection rather than only those whose vector storage would lose precision in a decode-encode round-trip (TurboQuant).\n\nRead on the sending side only, where the transfer batch is prepared: nodes accept raw points regardless.",
+            "default": false,
+            "type": "boolean"
+          },
+          "transfer_raw_payloads": {
+            "description": "Send the payload of a raw point as the byte blob it is stored as, so the sending node does not parse it and neither node builds a protobuf value tree for it. The receiving node still parses the blob, once, when the operation is applied. Only has an effect on points transferred raw, see [`Self::transfer_raw_points`].\n\nRead on the sending side only: nodes accept raw payloads regardless.",
+            "default": false,
+            "type": "boolean"
+          },
           "serverless_compatible": {
             "description": "Serverless-compatible deployment mode. Automatically enables [`Self::write_segment_manifest`], [`Self::append_only_mutations`], [`Self::compact_bitmask`] and [`Self::append_only_storages`].\n\nNote that this will only be applied when passed into [`init_feature_flags`].",
             "default": false,

lib/api/src/grpc/conversions.rs

@@ -3705,6 +3705,18 @@ impl From<Modifier> for grpc::Modifier {
     }
 }
 
+impl From<RawPayload> for grpc::RawPayload {
+    fn from(value: RawPayload) -> Self {
+        let RawPayload { payload_bytes } = value;
+
+        Self {
+            payload_bytes,
+            // A blob only ever comes from storage, which keeps payloads as serde_json.
+            encoding: grpc::RawPayloadEncoding::JsonBytes as i32,
+        }
+    }
+}
+
 impl TryFrom<grpc::RawPayload> for RawPayload {
     type Error = Status;
 

lib/collection/src/operations/generalizer/update_persisted.rs

@@ -1,5 +1,5 @@
 use itertools::Itertools;
-use segment::types::{Payload, PointIdType};
+use segment::types::{Payload, PointIdType, RawPayload};
 use serde_json::Value;
 use shard::operations::payload_ops::{PayloadOps, SetPayloadOp};
 use shard::operations::point_ops::{
@@ -25,6 +25,18 @@ impl Generalizer for Payload {
     }
 }
 
+impl Generalizer for RawPayload {
+    fn remove_details(&self) -> Self {
+        let Self { payload_bytes } = self;
+
+        // The blob is opaque here, so there are no keys to keep as the parsed-payload
+        // generalizer above does; retain only the byte length, like raw vectors below.
+        Self {
+            payload_bytes: (payload_bytes.len() as u64).to_le_bytes().to_vec(),
+        }
+    }
+}
+
 impl Generalizer for CollectionUpdateOperations {
     fn remove_details(&self) -> Self {
         match self {
@@ -103,6 +115,7 @@ impl Generalizer for PointStructRawPersisted {
             id: _, // ignore actual id for generalization
             vectors,
             payload,
+            payload_raw,
         } = self;
 
         Self {
@@ -113,6 +126,7 @@ impl Generalizer for PointStructRawPersisted {
                 .map(|(name, bytes)| (name.clone(), (bytes.len() as u64).to_le_bytes().to_vec()))
                 .collect(),
             payload: payload.as_ref().map(|p| p.remove_details()),
+            payload_raw: payload_raw.as_ref().map(|p| p.remove_details()),
         }
     }
 }

lib/collection/src/operations/types.rs

@@ -1138,6 +1138,9 @@ impl From<OperationError> for CollectionError {
             OperationError::MalformedVectorBlob { .. } => Self::BadInput {
                 description: err.to_string(),
             },
+            OperationError::MalformedPayloadBlob { .. } => Self::BadInput {
+                description: err.to_string(),
+            },
             OperationError::VectorNameNotExists { .. } => Self::BadInput {
                 description: err.to_string(),
             },

lib/collection/src/shards/forward_proxy_shard.rs

@@ -5,6 +5,7 @@ use std::time::{Duration, Instant};
 use ahash::HashSet;
 use async_trait::async_trait;
 use common::counter::hardware_accumulator::HwMeasurementAcc;
+use common::flags::feature_flags;
 use common::tar_ext;
 use common::types::{DeferredBehavior, TelemetryDetail};
 use parking_lot::Mutex as ParkingMutex;
@@ -211,17 +212,19 @@ impl ForwardProxyShard {
         // When any named vector uses a TurboQuant (`Turbo4`) storage datatype,
         // ship storage-native (raw) vector bytes instead of decoded floats.
         // This avoids a lossy TQ decode -> encode round-trip on the receiving
-        // node, which would drift the encoding and degrade recall.
-        let transfer_raw = self
-            .wrapped_shard
-            .collection_config
-            .read()
-            .await
-            .has_turbo_vector_storage();
+        // node, which would drift the encoding and degrade recall. The feature flag
+        // extends that to every collection, where it is a plain saving.
+        let transfer_raw = feature_flags().transfer_raw_points
+            || self
+                .wrapped_shard
+                .collection_config
+                .read()
+                .await
+                .has_turbo_vector_storage();
 
         let read_start = Instant::now();
         let (point_operation, next_page_offset, count) = if transfer_raw {
-            let (points, next_page_offset) = match hashring_filter {
+            let (mut points, next_page_offset) = match hashring_filter {
                 Some(hashring_filter) => {
                     self.read_batch_with_hashring_raw(
                         offset,
@@ -236,6 +239,16 @@ impl ForwardProxyShard {
                         .await?
                 }
             };
+
+            // A raw read hands out the payload as its stored blob. Shipping it is
+            // feature-flagged for the same reason as raw points: the receiving node has
+            // to understand it first.
+            if !feature_flags().transfer_raw_payloads {
+                for point in &mut points {
+                    point.decode_payload_raw()?;
+                }
+            }
+
             let count = points.len();
             let point_operation = if !merge_points {
                 PointOperations::SyncPointsRaw(PointSyncRawOperation {

lib/collection/src/shards/local_shard/scroll.rs

@@ -324,8 +324,8 @@ impl LocalShard {
             .iter()
             // Use remove to avoid cloning, we take each point ID only once
             .filter_map(|point_id| records_map.remove(point_id))
-            .map(PointStructRawPersisted::try_from)
-            .collect::<Result<_, _>>()?;
+            .map(PointStructRawPersisted::from)
+            .collect();
 
         Ok(ordered_records)
     }

lib/collection/src/shards/local_shard/shard_ops.rs

@@ -652,8 +652,8 @@ impl LocalShard {
             .iter()
             // Use remove to avoid cloning, we take each point ID only once
             .filter_map(|id| records_map.remove(id))
-            .map(PointStructRawPersisted::try_from)
-            .collect::<Result<_, _>>()?;
+            .map(PointStructRawPersisted::from)
+            .collect();
 
         Ok(ordered_records)
     }

lib/collection/src/tests/wal_recovery_test.rs

@@ -7,7 +7,8 @@ use common::save_on_disk::SaveOnDisk;
 use common::types::DeferredBehavior;
 use segment::data_types::vectors::{DEFAULT_VECTOR_NAME, VectorStructInternal};
 use segment::types::{
-    Distance, MultiVectorConfig, PayloadFieldSchema, PayloadSchemaType, WithPayload, WithVector,
+    Distance, MultiVectorConfig, PayloadFieldSchema, PayloadSchemaType, PointIdType, RawPayload,
+    WithPayload, WithVector,
 };
 use shard::operations::CollectionUpdateOperations;
 use shard::operations::point_ops::{
@@ -1590,6 +1591,7 @@ async fn test_malformed_raw_upsert_is_skipped_on_wal_replay() {
                 id: id.into(),
                 vectors: std::iter::once((DEFAULT_VECTOR_NAME.to_owned(), bytes)).collect(),
                 payload: None,
+                payload_raw: None,
             },
         ]))
     };
@@ -1622,6 +1624,40 @@ async fn test_malformed_raw_upsert_is_skipped_on_wal_replay() {
     );
 }
 
+/// Payload counterpart of [`test_malformed_raw_upsert_is_skipped_on_wal_replay`]: a blob
+/// reaches the WAL unparsed, so a bad one is only found on apply and must be skipped.
+#[tokio::test(flavor = "multi_thread")]
+async fn test_malformed_raw_payload_upsert_is_skipped_on_wal_replay() {
+    let vector_bytes: Vec<u8> = [1.0f32, 2.0, 3.0, 4.0]
+        .iter()
+        .flat_map(|f| f.to_ne_bytes())
+        .collect();
+
+    let raw_upsert = |id: u64, payload_bytes: &[u8]| {
+        CollectionUpdateOperations::PointOperation(PointOperations::UpsertPointsRaw(vec![
+            PointStructRawPersisted {
+                id: PointIdType::from(id),
+                vectors: std::iter::once((DEFAULT_VECTOR_NAME.to_owned(), vector_bytes.clone()))
+                    .collect(),
+                payload: None,
+                payload_raw: Some(RawPayload::from_storage_bytes(payload_bytes.to_vec())),
+            },
+        ]))
+    };
+
+    let err = assert_bad_op_skipped_on_wal_replay(
+        create_collection_config(),
+        raw_upsert(1, br#"{"city": "Berlin"}"#),
+        raw_upsert(2, b"not json"),
+    )
+    .await;
+
+    assert!(
+        matches!(err, CollectionError::BadInput { .. }),
+        "malformed payload blob must be a BadInput user error (skipped on replay), got {err:?}",
+    );
+}
+
 /// Sparse counterpart of [`test_malformed_raw_upsert_is_skipped_on_wal_replay`]:
 /// a raw sparse blob that doesn't decode as a stored sparse vector.
 #[tokio::test(flavor = "multi_thread")]
@@ -1662,6 +1698,7 @@ async fn test_malformed_sparse_raw_upsert_is_skipped_on_wal_replay() {
                 id: 2.into(),
                 vectors: std::iter::once((sparse_name.to_owned(), vec![0_u8, 1, 2])).collect(),
                 payload: None,
+                payload_raw: None,
             },
         ]));
 

lib/common/common/src/flags.rs

@@ -54,6 +54,22 @@ pub struct FeatureFlags {
     /// Implies [`Self::append_only_mutations`], enforced by [`init_feature_flags`].
     pub append_only_storages: bool,
 
+    /// Transfer points as storage-native bytes (raw points), for every collection rather than
+    /// only those whose vector storage would lose precision in a decode-encode round-trip
+    /// (TurboQuant).
+    ///
+    /// Read on the sending side only, where the transfer batch is prepared: nodes accept
+    /// raw points regardless.
+    pub transfer_raw_points: bool,
+
+    /// Send the payload of a raw point as the byte blob it is stored as, so the sending
+    /// node does not parse it and neither node builds a protobuf value tree for it. The
+    /// receiving node still parses the blob, once, when the operation is applied. Only has
+    /// an effect on points transferred raw, see [`Self::transfer_raw_points`].
+    ///
+    /// Read on the sending side only: nodes accept raw payloads regardless.
+    pub transfer_raw_payloads: bool,
+
     /// Serverless-compatible deployment mode. Automatically enables [`Self::write_segment_manifest`],
     /// [`Self::append_only_mutations`], [`Self::compact_bitmask`] and
     /// [`Self::append_only_storages`].
@@ -74,6 +90,8 @@ impl Default for FeatureFlags {
             append_only_mutations: false,
             compact_bitmask: false,
             append_only_storages: false,
+            transfer_raw_points: false,
+            transfer_raw_payloads: false,
             serverless_compatible: false,
         }
     }
@@ -104,6 +122,10 @@ impl FeatureFlags {
             append_only_mutations: false,
             append_only_storages: false,
             compact_bitmask: true,
+            // Deliberately not enabled by `all`: a node only accepts these once it runs a
+            // version that understands them, so they can only be switched on a release later.
+            transfer_raw_points: false,
+            transfer_raw_payloads: false,
             serverless_compatible: false,
         }
     }

lib/common/common/src/lib.rs

@@ -39,6 +39,7 @@ pub mod process_counter;
 pub mod process_cpu_usage;
 pub mod progress_tracker;
 pub mod rate_limiting;
+pub mod raw_bytes_serde;
 pub mod save_on_disk;
 pub mod scope_tracker;
 pub mod small_uint;

lib/common/common/src/raw_bytes_serde.rs

@@ -0,0 +1,74 @@
+//! Serde helpers for raw byte blobs.
+//!
+//! By default serde serializes `Vec<u8>` as a sequence of integers, which CBOR (used for
+//! the WAL) writes as one item per byte. The helpers here force blobs through
+//! `serialize_bytes` so they become a single byte string instead.
+
+use std::fmt;
+
+use serde::de::{self, Deserializer, SeqAccess, Visitor};
+use serde::ser::Serializer;
+use serde::{Deserialize, Serialize};
+
+/// Upper bound for the capacity we pre-allocate from an untrusted `size_hint` when deserializing a single raw blob.
+/// This is *not* a hard size limit.
+const MAX_RAW_BLOB_PREALLOC: usize = 128 * 1024 * 1024;
+
+/// Reference wrapper that serializes a byte slice as a byte string.
+pub struct BytesRef<'a>(pub &'a [u8]);
+
+impl serde::Serialize for BytesRef<'_> {
+    fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
+        serializer.serialize_bytes(self.0)
+    }
+}
+
+/// Owned wrapper that deserializes a byte string into a `Vec<u8>`.
+pub struct ByteVec(pub Vec<u8>);
+
+impl<'de> serde::Deserialize<'de> for ByteVec {
+    fn deserialize<D: Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
+        struct ByteVecVisitor;
+
+        impl<'de> Visitor<'de> for ByteVecVisitor {
+            type Value = Vec<u8>;
+
+            fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result {
+                formatter.write_str("a byte string")
+            }
+
+            fn visit_bytes<E: de::Error>(self, value: &[u8]) -> Result<Self::Value, E> {
+                Ok(value.to_vec())
+            }
+
+            fn visit_byte_buf<E: de::Error>(self, value: Vec<u8>) -> Result<Self::Value, E> {
+                Ok(value)
+            }
+
+            /// Formats that lack a native byte-string type (e.g. JSON) fall
+            /// back to a sequence of integers; accept those too.
+            fn visit_seq<A: SeqAccess<'de>>(self, mut seq: A) -> Result<Self::Value, A::Error> {
+                let capacity = seq.size_hint().unwrap_or(0).min(MAX_RAW_BLOB_PREALLOC);
+                let mut bytes = Vec::with_capacity(capacity);
+                while let Some(byte) = seq.next_element()? {
+                    bytes.push(byte);
+                }
+                Ok(bytes)
+            }
+        }
+
+        deserializer
+            .deserialize_byte_buf(ByteVecVisitor)
+            .map(ByteVec)
+    }
+}
+
+/// `#[serde(with = "...")]` entry point for a plain `Vec<u8>` field.
+pub fn serialize<S: Serializer>(bytes: &[u8], serializer: S) -> Result<S::Ok, S::Error> {
+    BytesRef(bytes).serialize(serializer)
+}
+
+/// `#[serde(with = "...")]` entry point for a plain `Vec<u8>` field.
+pub fn deserialize<'de, D: Deserializer<'de>>(deserializer: D) -> Result<Vec<u8>, D::Error> {
+    ByteVec::deserialize(deserializer).map(|bytes| bytes.0)
+}

lib/edge/src/update_only/batch/plan.rs

@@ -104,11 +104,14 @@ impl UpdateBatchPlan {
                 }
             }
             PointOperations::UpsertPointsRaw(points) => {
-                for point in points {
+                for mut point in points {
+                    // Decode, so this path cannot drop the payload of a local operation.
+                    point.decode_payload_raw()?;
                     let PointStructRawPersisted {
                         id,
                         vectors,
                         payload,
+                        payload_raw: _,
                     } = point;
                     self.push(
                         id,

lib/segment/src/common/operation_error.rs

@@ -33,6 +33,11 @@ pub enum OperationError {
     /// crash-looping recovery.
     #[error("{description}")]
     MalformedVectorBlob { description: String },
+    /// A stored-encoding payload blob that does not parse. User error for the same reason
+    /// as [`Self::MalformedVectorBlob`]: it arrives from a peer and is parsed on apply, so
+    /// a bad one has to be skipped on replay instead of crash-looping recovery.
+    #[error("{description}")]
+    MalformedPayloadBlob { description: String },
     #[error("Not existing vector name error: {received_name}")]
     VectorNameNotExists { received_name: VectorNameBuf },
     #[error("No point with id {missed_point_id}")]
@@ -167,6 +172,7 @@ impl IsNotFound for OperationError {
             Self::FileNotFound { .. } => true,
             Self::WrongVectorDimension { .. }
             | Self::MalformedVectorBlob { .. }
+            | Self::MalformedPayloadBlob { .. }
             | Self::VectorNameNotExists { .. }
             | Self::PointIdError { .. }
             | Self::TypeError { .. }

lib/segment/src/types.rs

@@ -16,6 +16,7 @@ use std::sync::Arc;
 
 use ahash::AHashSet;
 use bytemuck::{Pod, Zeroable};
+use common::raw_bytes_serde;
 use common::stable_hash::StableHash;
 use common::types::{PointOffsetType, ScoreType};
 use ecow::EcoString;
@@ -2945,10 +2946,16 @@ impl TryFrom<PayloadIndexInfo> for PayloadFieldSchema {
 /// Byte-blob analogue of [`Payload`]: the whole payload object as a single
 /// encoded blob, tagged with its encoding.
 ///
-/// Read from storage by `retrieve_raw` and shipped to another node as-is, so
-/// neither side has to parse the payload on the way.
-#[derive(Clone, Debug, Eq, PartialEq, Hash)]
+/// Read from storage by `retrieve_raw` and shipped as-is, so it is parsed once: when the
+/// receiving node applies the point.
+#[derive(Clone, Debug, Eq, PartialEq, Deserialize, Serialize, Hash)]
 pub struct RawPayload {
+    /// A compact byte string rather than the serde default of an integer sequence, which
+    /// is what the WAL would otherwise store.
+    ///
+    /// The helper is named through a `use` import rather than a full path: Qdrant Edge's
+    /// amalgamation rewrites paths in code but cannot see into attribute strings.
+    #[serde(with = "raw_bytes_serde")]
     pub payload_bytes: Vec<u8>,
 }
 
@@ -2960,9 +2967,14 @@ impl RawPayload {
     }
 
     /// Parse the blob into a [`Payload`].
+    ///
+    /// Reported as user error, so an operation carrying a bad blob is skipped on WAL
+    /// replay rather than failing recovery.
     pub fn decode(&self) -> OperationResult<Payload> {
         serde_json::from_slice(&self.payload_bytes).map_err(|err| {
-            OperationError::service_error(format!("Malformed raw payload blob: {err}"))
+            OperationError::MalformedPayloadBlob {
+                description: format!("Malformed raw payload blob: {err}"),
+            }
         })
     }
 }

lib/shard/Cargo.toml

@@ -54,6 +54,7 @@ shard = { path = ".", default-features = false, features = ["testing"] }
 
 criterion = { workspace = true }
 fs-err = { workspace = true, features = ["debug"] }
+prost = { workspace = true }
 proptest = { workspace = true }
 rstest = { workspace = true }
 serde_cbor = { workspace = true }
@@ -62,3 +63,7 @@ tar = { workspace = true }
 [[bench]]
 name = "find_duplicated_points"
 harness = false
+
+[[bench]]
+name = "raw_payload_transfer"
+harness = false

lib/shard/benches/raw_payload_transfer.rs

@@ -0,0 +1,197 @@
+//! What `transfer_raw_payloads` actually costs and saves, per point.
+//!
+//! Both arms transfer raw points; they differ only in how the payload travels, which
+//! is exactly what the flag switches:
+//!
+//! - `blob`: the payload stays the stored byte blob from the sending node's storage
+//!   into the receiving node's WAL, and is parsed once when the point is applied.
+//! - `value_tree`: the sender parses the blob and builds a protobuf value tree, the
+//!   receiver turns that tree back into a payload. What happens with the flag off.
+//!
+//! Run with:
+//!     cargo bench -p shard --bench raw_payload_transfer
+//!
+//! Sizes are printed once at startup, since criterion only reports time.
+
+use std::hint::black_box;
+
+use criterion::{BenchmarkId, Criterion, Throughput, criterion_group, criterion_main};
+use prost::Message as _;
+use segment::types::{Payload, PointIdType, RawPayload};
+use shard::operations::point_ops::{PointOperations, PointStructRawPersisted, RawVectorsPersisted};
+
+/// Points per measured batch, in the order of a transfer batch.
+const BATCH: usize = 100;
+
+/// Payload widths to measure, in `field_count`. Payload size drives the whole
+/// comparison, so a narrow and a wide payload are measured separately.
+const WIDTHS: [usize; 3] = [1, 10, 100];
+
+/// A payload shaped like a real one — strings, numbers, a nested object and an array
+/// — rather than a flat map of a single key.
+fn payload(fields: usize) -> Payload {
+    let mut map = serde_json::Map::new();
+    for i in 0..fields {
+        map.insert(
+            format!("field_{i}"),
+            serde_json::json!(format!("value number {i} with some descriptive text")),
+        );
+        map.insert(format!("count_{i}"), serde_json::json!(i));
+    }
+    map.insert(
+        "nested".to_string(),
+        serde_json::json!({"city": "Berlin", "geo": {"lat": 52.52, "lon": 13.405}}),
+    );
+    map.insert("tags".to_string(), serde_json::json!(["a", "b", "c"]));
+    serde_json::from_value(serde_json::Value::Object(map)).unwrap()
+}
+
+/// A batch as a raw read hands it out: vectors and payload both still stored bytes.
+fn batch_as_read(fields: usize) -> Vec<PointStructRawPersisted> {
+    let blob = serde_json::to_vec(&payload(fields)).unwrap();
+    let vector: Vec<u8> = [1.0f32, 2.0, 3.0, 4.0]
+        .iter()
+        .flat_map(|v| v.to_le_bytes())
+        .collect();
+
+    (0..BATCH)
+        .map(|i| PointStructRawPersisted {
+            id: PointIdType::from(i as u64),
+            vectors: RawVectorsPersisted::from(vec![("dense".to_string(), vector.clone())]),
+            payload: None,
+            payload_raw: Some(RawPayload::from_storage_bytes(blob.clone())),
+        })
+        .collect()
+}
+
+/// The same batch with every blob already parsed, i.e. what the sender holds with the
+/// flag off.
+fn batch_decoded(fields: usize) -> Vec<PointStructRawPersisted> {
+    let mut points = batch_as_read(fields);
+    for point in &mut points {
+        point.decode_payload_raw().unwrap();
+    }
+    points
+}
+
+/// Sender: turn a batch into the wire messages.
+fn to_wire(points: &[PointStructRawPersisted]) -> Vec<api::grpc::qdrant::PointStructRaw> {
+    points
+        .iter()
+        .cloned()
+        .map(api::grpc::qdrant::PointStructRaw::from)
+        .collect()
+}
+
+/// Receiver: turn wire messages back into points ready to apply, which for a blob
+/// includes the one parse the whole design leans on.
+fn from_wire(wire: &[api::grpc::qdrant::PointStructRaw]) -> Vec<PointStructRawPersisted> {
+    wire.iter()
+        .cloned()
+        .map(|point| {
+            let mut point = PointStructRawPersisted::try_from(point).unwrap();
+            point.decode_payload_raw().unwrap();
+            point
+        })
+        .collect()
+}
+
+/// What the WAL write serializes.
+fn wal_bytes(points: Vec<PointStructRawPersisted>) -> Vec<u8> {
+    serde_cbor::to_vec(&PointOperations::UpsertPointsRaw(points)).unwrap()
+}
+
+fn report_sizes() {
+    eprintln!(
+        "\n{:>7}  {:>10}  {:>12}  {:>12}",
+        "fields", "arm", "wire bytes", "WAL bytes"
+    );
+    for fields in WIDTHS {
+        for (arm, points) in [
+            ("blob", batch_as_read(fields)),
+            ("value_tree", batch_decoded(fields)),
+        ] {
+            let wire: usize = to_wire(&points)
+                .iter()
+                .map(api::grpc::qdrant::PointStructRaw::encoded_len)
+                .sum();
+            let wal = wal_bytes(points).len();
+            eprintln!("{fields:>7}  {arm:>10}  {wire:>12}  {wal:>12}");
+        }
+    }
+    eprintln!();
+}
+
+fn raw_payload_bench(c: &mut Criterion) {
+    report_sizes();
+
+    let mut sender = c.benchmark_group("raw-payload-sender");
+    for fields in WIDTHS {
+        sender.throughput(Throughput::Elements(BATCH as u64));
+
+        let as_read = batch_as_read(fields);
+        sender.bench_with_input(BenchmarkId::new("blob", fields), &as_read, |b, points| {
+            b.iter(|| black_box(to_wire(points)));
+        });
+
+        // With the flag off the sender also pays the parse, so it starts from the
+        // batch as read too.
+        sender.bench_with_input(
+            BenchmarkId::new("value_tree", fields),
+            &as_read,
+            |b, points| {
+                b.iter(|| {
+                    let mut points = points.clone();
+                    for point in &mut points {
+                        point.decode_payload_raw().unwrap();
+                    }
+                    black_box(to_wire(&points))
+                });
+            },
+        );
+    }
+    sender.finish();
+
+    let mut receiver = c.benchmark_group("raw-payload-receiver");
+    for fields in WIDTHS {
+        receiver.throughput(Throughput::Elements(BATCH as u64));
+
+        let blob_wire = to_wire(&batch_as_read(fields));
+        receiver.bench_with_input(BenchmarkId::new("blob", fields), &blob_wire, |b, wire| {
+            b.iter(|| black_box(from_wire(wire)));
+        });
+
+        let tree_wire = to_wire(&batch_decoded(fields));
+        receiver.bench_with_input(
+            BenchmarkId::new("value_tree", fields),
+            &tree_wire,
+            |b, wire| {
+                b.iter(|| black_box(from_wire(wire)));
+            },
+        );
+    }
+    receiver.finish();
+
+    let mut wal = c.benchmark_group("raw-payload-wal-encode");
+    for fields in WIDTHS {
+        wal.throughput(Throughput::Elements(BATCH as u64));
+
+        let as_read = batch_as_read(fields);
+        wal.bench_with_input(BenchmarkId::new("blob", fields), &as_read, |b, points| {
+            b.iter(|| black_box(wal_bytes(points.clone())));
+        });
+
+        let decoded = batch_decoded(fields);
+        wal.bench_with_input(
+            BenchmarkId::new("value_tree", fields),
+            &decoded,
+            |b, points| {
+                b.iter(|| black_box(wal_bytes(points.clone())));
+            },
+        );
+    }
+    wal.finish();
+}
+
+criterion_group!(benches, raw_payload_bench);
+criterion_main!(benches);

lib/shard/src/operations/mod.rs

@@ -432,11 +432,15 @@ mod tests {
                 points: Vec::new(),
             });
 
-            // Use a non-empty raw point so the byte-blob path is actually exercised
+            // Use a non-empty raw point, with a payload blob, so both byte-blob paths
+            // are actually exercised
             let raw_point = PointStructRawPersisted {
                 id: 1.into(),
                 vectors: vec![("dense".to_string(), vec![0, 1, 2, 3, 255])].into(),
                 payload: None,
+                payload_raw: Some(segment::types::RawPayload::from_storage_bytes(
+                    br#"{"city":"Berlin"}"#.to_vec(),
+                )),
             };
 
             let upsert_raw = Self::UpsertPointsRaw(vec![raw_point.clone()]);

lib/shard/src/operations/point_ops.rs

@@ -7,7 +7,7 @@ use common::validation::validate_multi_vector;
 use itertools::Itertools as _;
 use ordered_float::OrderedFloat;
 use schemars::JsonSchema;
-use segment::common::operation_error::OperationError;
+use segment::common::operation_error::{OperationError, OperationResult};
 use segment::common::utils::unordered_hash_unique;
 use segment::data_types::named_vectors::NamedVectors;
 use segment::data_types::segment_record::{SegmentRecord, SegmentRecordRaw};
@@ -336,41 +336,30 @@ pub struct PointStructRawPersisted {
     pub vectors: RawVectorsPersisted,
     /// Payload values (optional)
     pub payload: Option<Payload>,
+    /// The whole payload as a single encoded blob, as read from storage.
+    ///
+    /// Mutually exclusive with `payload`. Kept from the sender all the way into the WAL
+    /// and parsed once, when the point is applied. Skipped when `None` so WAL entries
+    /// without a blob stay byte-identical to those written before this field existed.
+    ///
+    /// Costs more WAL bytes than `payload` would, since the blob is JSON while a parsed
+    /// payload becomes a compact CBOR map. Decoding earlier to win them back would mean a
+    /// second full deserialization, so the blob is carried as-is.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub payload_raw: Option<RawPayload>,
 }
 
-/// Serde helper for [`PointStructRawPersisted::vectors`].
-///
-/// By default serde serializes `Vec<u8>` as a sequence of integers, which in
-/// CBOR (used for the WAL) costs ~2x for high-entropy data such as raw vector
-/// bytes. This module forces each blob through `serialize_bytes` so it is
-/// encoded as a compact byte string (~1x overhead) instead.
+/// Serde helper for [`PointStructRawPersisted::vectors`]: each vector blob in the
+/// `(name, bytes)` pair list goes through [`common::raw_bytes_serde`], so it is encoded as
+/// a compact byte string instead of the serde default of a sequence of integers.
 mod raw_vectors_serde {
-    use std::fmt;
-
+    use common::raw_bytes_serde::{ByteVec, BytesRef};
     use segment::types::VectorNameBuf;
-    use serde::de::{self, Deserializer, SeqAccess, Visitor};
+    use serde::de::Deserializer;
     use serde::ser::{SerializeSeq, Serializer};
 
     use super::RawVectorsPersisted;
 
-    /// Upper bound for the capacity we pre-allocate from an untrusted `size_hint`
-    /// when deserializing a single vector's raw bytes.
-    ///
-    /// Realistic sizes are far below this: a maximum-size dense vector is
-    /// 65536 dims x 4 bytes (f32) = 256 KiB. The 128 MiB headroom comfortably
-    /// covers large multivectors and sparse vectors while staying orders of
-    /// magnitude away from OOM territory.
-    const MAX_RAW_VECTOR_PREALLOC: usize = 128 * 1024 * 1024;
-
-    /// Reference wrapper that serializes a byte slice as a byte string.
-    struct BytesRef<'a>(&'a [u8]);
-
-    impl serde::Serialize for BytesRef<'_> {
-        fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
-            serializer.serialize_bytes(self.0)
-        }
-    }
-
     pub fn serialize<S: Serializer>(
         vectors: &[(VectorNameBuf, Vec<u8>)],
         serializer: S,
@@ -382,46 +371,6 @@ mod raw_vectors_serde {
         seq.end()
     }
 
-    /// Owned wrapper that deserializes a byte string into a `Vec<u8>`.
-    struct ByteVec(Vec<u8>);
-
-    impl<'de> serde::Deserialize<'de> for ByteVec {
-        fn deserialize<D: Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
-            struct ByteVecVisitor;
-
-            impl<'de> Visitor<'de> for ByteVecVisitor {
-                type Value = Vec<u8>;
-
-                fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result {
-                    formatter.write_str("a byte string")
-                }
-
-                fn visit_bytes<E: de::Error>(self, value: &[u8]) -> Result<Self::Value, E> {
-                    Ok(value.to_vec())
-                }
-
-                fn visit_byte_buf<E: de::Error>(self, value: Vec<u8>) -> Result<Self::Value, E> {
-                    Ok(value)
-                }
-
-                /// Formats that lack a native byte-string type (e.g. JSON) fall
-                /// back to a sequence of integers; accept those too.
-                fn visit_seq<A: SeqAccess<'de>>(self, mut seq: A) -> Result<Self::Value, A::Error> {
-                    let capacity = seq.size_hint().unwrap_or(0).min(MAX_RAW_VECTOR_PREALLOC);
-                    let mut bytes = Vec::with_capacity(capacity);
-                    while let Some(byte) = seq.next_element()? {
-                        bytes.push(byte);
-                    }
-                    Ok(bytes)
-                }
-            }
-
-            deserializer
-                .deserialize_byte_buf(ByteVecVisitor)
-                .map(ByteVec)
-        }
-    }
-
     pub fn deserialize<'de, D: Deserializer<'de>>(
         deserializer: D,
     ) -> Result<RawVectorsPersisted, D::Error> {
@@ -433,33 +382,48 @@ mod raw_vectors_serde {
     }
 }
 
-impl TryFrom<SegmentRecordRaw> for PointStructRawPersisted {
-    type Error = OperationError;
-
-    /// A raw record carries the payload as stored, so it is decoded here — this
-    /// struct goes into the WAL, which holds parsed payloads.
-    fn try_from(record: SegmentRecordRaw) -> Result<Self, Self::Error> {
+impl From<SegmentRecordRaw> for PointStructRawPersisted {
+    /// A raw read hands out the payload as stored, so the blob travels as-is and
+    /// nothing is parsed or encoded here.
+    fn from(record: SegmentRecordRaw) -> Self {
         let SegmentRecordRaw {
             id,
             vectors,
             payload,
         } = record;
 
-        Ok(Self {
+        Self {
             id,
             vectors: vectors.unwrap_or_default(),
-            payload: payload.as_ref().map(RawPayload::decode).transpose()?,
-        })
+            payload: None,
+            payload_raw: payload,
+        }
     }
 }
 
 impl PointStructRawPersisted {
+    /// Move a raw payload blob into the parsed [`Self::payload`], decoding it.
+    ///
+    /// The blob is taken only once it has parsed, so a failure leaves the point holding it
+    /// rather than holding neither representation.
+    pub fn decode_payload_raw(&mut self) -> OperationResult<()> {
+        let Some(payload_raw) = &self.payload_raw else {
+            return Ok(());
+        };
+
+        debug_assert!(self.payload.is_none());
+
+        self.payload = Some(payload_raw.decode()?);
+        self.payload_raw = None;
+
+        Ok(())
+    }
+
     /// Whether this point carries the data stored in `segment_record`.
     ///
-    /// Vectors are compared as bytes, so logically equal vectors in a different
-    /// encoding count as unequal, which only costs a redundant upsert on sync.
-    /// Payloads are compared parsed, decoding the stored blob; a blob that does
-    /// not parse counts as unequal.
+    /// A point reaches here with its payload already parsed — both raw entry points
+    /// require it — so the stored blob is decoded to compare. The blob-to-blob arms are
+    /// an unreachable fallback, kept correct in case a caller ever compares first.
     pub fn is_equal_to(&self, segment_record: &SegmentRecordRaw) -> bool {
         let SegmentRecordRaw {
             id,
@@ -487,13 +451,20 @@ impl PointStructRawPersisted {
         }
 
         // Check if payloads are equal, empty and non-existent payloads are considered equal
-        let self_payload = self.payload.as_ref().filter(|p| !p.is_empty());
-        let Ok(segment_payload) = payload.as_ref().map(RawPayload::decode).transpose() else {
-            // A blob that cannot be parsed is not the data this point carries
-            return false;
-        };
-        let segment_payload = segment_payload.filter(|payload| !payload.is_empty());
-        self_payload == segment_payload.as_ref()
+        match (&self.payload_raw, payload) {
+            (Some(own_blob), Some(segment_blob)) => own_blob == segment_blob,
+            (Some(own_blob), None) => own_blob.decode().is_ok_and(|payload| payload.is_empty()),
+            (None, _) => {
+                let self_payload = self.payload.as_ref().filter(|p| !p.is_empty());
+                let Ok(segment_payload) = payload.as_ref().map(RawPayload::decode).transpose()
+                else {
+                    // A blob that cannot be parsed is not the data this point carries
+                    return false;
+                };
+                let segment_payload = segment_payload.filter(|payload| !payload.is_empty());
+                self_payload == segment_payload.as_ref()
+            }
+        }
     }
 }
 
@@ -504,6 +475,7 @@ impl From<PointStructRawPersisted> for api::grpc::qdrant::PointStructRaw {
             id,
             vectors,
             payload,
+            payload_raw,
         } = value;
 
         Self {
@@ -512,7 +484,7 @@ impl From<PointStructRawPersisted> for api::grpc::qdrant::PointStructRaw {
             payload: payload
                 .map(api::conversions::json::payload_to_proto)
                 .unwrap_or_default(),
-            raw_payload: None,
+            raw_payload: payload_raw.map(api::grpc::qdrant::RawPayload::from),
         }
     }
 }
@@ -533,34 +505,29 @@ impl TryFrom<api::grpc::qdrant::PointStructRaw> for PointStructRawPersisted {
             .ok_or_else(|| tonic::Status::invalid_argument("Empty id is not allowed"))?
             .try_into()?;
 
-        // Prefer the raw payload blob and fall back to the serialized payload otherwise.
+        // Only the encoding tag is checked here; the blob itself is parsed when the point
+        // is applied. Rejecting both fields mirrors how the request rejects both
+        // `points` and `raw_points`.
+        let payload_raw = raw_payload.map(RawPayload::try_from).transpose()?;
+        if payload_raw.is_some() && !payload.is_empty() {
+            return Err(tonic::Status::invalid_argument(
+                "Only one of `payload` and `raw_payload` can be set for a point",
+            ));
+        }
+
         // An empty payload is normalized to `None`.
-        let payload = match raw_payload {
-            Some(raw_payload) => decode_payload(raw_payload)?,
-            None => api::conversions::json::proto_to_payloads(payload)?,
-        };
+        let payload = api::conversions::json::proto_to_payloads(payload)?;
         let payload = (!payload.is_empty()).then_some(payload);
 
         Ok(Self {
             id,
             vectors: vectors.into_iter().collect(),
             payload,
+            payload_raw,
         })
     }
 }
 
-/// Decodes a raw payload received over the wire.
-///
-/// The blob is decoded by [`RawPayload::decode`], the one place that knows how
-/// each encoding is read; this only restates its errors as the bad input they
-/// are on this side.
-#[cfg(feature = "api")]
-fn decode_payload(raw_payload: api::grpc::RawPayload) -> Result<Payload, tonic::Status> {
-    RawPayload::try_from(raw_payload)?
-        .decode()
-        .map_err(|err| tonic::Status::invalid_argument(err.to_string()))
-}
-
 impl Debug for PointStructRawPersisted {
     fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
         let vectors = self
@@ -568,9 +535,13 @@ impl Debug for PointStructRawPersisted {
             .iter()
             .map(|(name, bytes)| format!("{name}: {} bytes", bytes.len()))
             .join(", ");
+        let payload_raw = self
+            .payload_raw
+            .as_ref()
+            .map(|payload| format!("{} bytes", payload.payload_bytes.len()));
         write!(
             f,
-            "PointStructRawPersisted {{ id: {}, vectors: [{vectors}], payload: {:?} }}",
+            "PointStructRawPersisted {{ id: {}, vectors: [{vectors}], payload: {:?}, payload_raw: {payload_raw:?} }}",
             self.id, self.payload,
         )
     }
@@ -1138,17 +1109,15 @@ mod tests {
 
     #[test]
     fn raw_persisted_vectors_use_compact_byte_string() {
-        // High-entropy payload: byte value == (index % 256), most bytes >= 24.
         let blob: Vec<u8> = (0..4096u32).map(|i| i as u8).collect();
         let point = PointStructRawPersisted {
-            id: 1.into(),
+            id: PointIdType::from(1),
             vectors: vec![("dense".to_string(), blob.clone())].into(),
             payload: None,
+            payload_raw: None,
         };
 
         let encoded = serde_cbor::to_vec(&point).unwrap();
-        // A byte string is ~1x; an integer array would be ~1.9x for this data.
-        // Guard well below the naive-array size (>7800 bytes for 4096 bytes).
         assert!(
             encoded.len() < blob.len() + 128,
             "expected compact byte-string encoding, got {} bytes for a {}-byte blob",
@@ -1156,11 +1125,212 @@ mod tests {
             blob.len(),
         );
 
-        // Round-trips losslessly.
         let decoded: PointStructRawPersisted = serde_cbor::from_slice(&encoded).unwrap();
         assert!(decoded == point, "round-trip mismatch");
     }
 
+    #[test]
+    fn raw_persisted_payload_uses_compact_byte_string() {
+        let blob: Vec<u8> = (0..4096u32).map(|i| i as u8).collect();
+        let point = PointStructRawPersisted {
+            id: PointIdType::from(1),
+            vectors: RawVectorsPersisted::default(),
+            payload: None,
+            payload_raw: Some(RawPayload::from_storage_bytes(blob.clone())),
+        };
+
+        let encoded = serde_cbor::to_vec(&point).unwrap();
+        assert!(
+            encoded.len() < blob.len() + 128,
+            "expected compact byte-string encoding, got {} bytes for a {}-byte blob",
+            encoded.len(),
+            blob.len(),
+        );
+
+        let decoded: PointStructRawPersisted = serde_cbor::from_slice(&encoded).unwrap();
+        assert!(decoded == point, "round-trip mismatch");
+    }
+
+    /// A point without a blob must encode exactly as it did before the field existed.
+    #[test]
+    fn raw_persisted_without_payload_blob_keeps_wal_encoding() {
+        /// Mirror of [`PointStructRawPersisted`] without `payload_raw`.
+        #[derive(Serialize)]
+        #[serde(rename_all = "snake_case")]
+        struct LegacyPointStructRawPersisted {
+            id: PointIdType,
+            #[serde(with = "raw_vectors_serde")]
+            vectors: RawVectorsPersisted,
+            payload: Option<Payload>,
+        }
+
+        let vectors = RawVectorsPersisted::from(vec![("dense".to_string(), vec![0_u8, 1, 2, 3])]);
+        let legacy = LegacyPointStructRawPersisted {
+            id: PointIdType::from(1),
+            vectors: vectors.clone(),
+            payload: None,
+        };
+        let point = PointStructRawPersisted {
+            id: PointIdType::from(1),
+            vectors,
+            payload: None,
+            payload_raw: None,
+        };
+
+        assert_eq!(
+            serde_cbor::to_vec(&point).unwrap(),
+            serde_cbor::to_vec(&legacy).unwrap(),
+        );
+    }
+
+    /// The stored blob is decoded for the comparison rather than counting as a difference.
+    #[test]
+    fn raw_point_is_equal_to_decodes_stored_blob() {
+        let record = SegmentRecordRaw {
+            id: PointIdType::from(1),
+            vectors: None,
+            payload: Some(RawPayload::from_storage_bytes(br#"{"a":1}"#.to_vec())),
+        };
+
+        let mut point = PointStructRawPersisted {
+            id: PointIdType::from(1),
+            vectors: RawVectorsPersisted::default(),
+            payload: Some(serde_json::from_str(r#"{"a":1}"#).unwrap()),
+            payload_raw: None,
+        };
+        assert!(point.is_equal_to(&record));
+
+        point.payload = Some(serde_json::from_str(r#"{"a":2}"#).unwrap());
+        assert!(!point.is_equal_to(&record));
+
+        point.payload = None;
+        assert!(!point.is_equal_to(&record));
+    }
+
+    /// A stored blob that does not parse is unequal, so the point is upserted rather than
+    /// silently skipped.
+    #[test]
+    fn raw_point_is_equal_to_rejects_malformed_blob() {
+        let record = SegmentRecordRaw {
+            id: PointIdType::from(1),
+            vectors: None,
+            payload: Some(RawPayload::from_storage_bytes(b"not json".to_vec())),
+        };
+        let point = PointStructRawPersisted {
+            id: PointIdType::from(1),
+            vectors: RawVectorsPersisted::default(),
+            payload: Some(serde_json::from_str(r#"{"a":1}"#).unwrap()),
+            payload_raw: None,
+        };
+
+        assert!(!point.is_equal_to(&record));
+    }
+
+    /// A blob survives the wire types unchanged, with no value tree built on either side.
+    #[cfg(feature = "api")]
+    #[test]
+    fn raw_payload_round_trips_over_the_wire_types() {
+        let payload: Payload =
+            serde_json::from_str(r#"{"city": "Berlin", "count": 3, "nested": {"a": [1, 2]}}"#)
+                .unwrap();
+        let stored_bytes = serde_json::to_vec(&payload).unwrap();
+
+        let point = PointStructRawPersisted {
+            id: PointIdType::from(1),
+            vectors: RawVectorsPersisted::from(vec![("dense".to_string(), vec![0_u8, 1, 2, 3])]),
+            payload: None,
+            payload_raw: Some(RawPayload::from_storage_bytes(stored_bytes.clone())),
+        };
+
+        let sent = api::grpc::qdrant::PointStructRaw::from(point);
+        assert!(
+            sent.payload.is_empty(),
+            "a raw payload must not also be sent as a value tree",
+        );
+        let raw_payload = sent.raw_payload.as_ref().expect("blob must be sent");
+        assert_eq!(raw_payload.payload_bytes, stored_bytes);
+        assert_eq!(
+            raw_payload.encoding(),
+            api::grpc::RawPayloadEncoding::JsonBytes,
+        );
+
+        // The receiving side keeps the blob, which is what puts it in the WAL as it arrived.
+        let received = PointStructRawPersisted::try_from(sent).unwrap();
+        assert!(received.payload.is_none());
+        assert_eq!(
+            received.payload_raw.as_ref().map(|raw| &raw.payload_bytes),
+            Some(&stored_bytes),
+        );
+
+        let mut applied = received;
+        applied.decode_payload_raw().unwrap();
+        assert_eq!(applied.payload, Some(payload));
+    }
+
+    /// Two payloads for one point is malformed, not something to silently pick from.
+    #[cfg(feature = "api")]
+    #[test]
+    fn wire_point_with_both_payload_fields_is_rejected() {
+        let mut sent = api::grpc::qdrant::PointStructRaw::from(PointStructRawPersisted {
+            id: PointIdType::from(1),
+            vectors: RawVectorsPersisted::default(),
+            payload: Some(serde_json::from_str(r#"{"city":"Berlin"}"#).unwrap()),
+            payload_raw: None,
+        });
+        assert!(!sent.payload.is_empty());
+        sent.raw_payload = Some(api::grpc::RawPayload {
+            payload_bytes: br#"{"city":"Berlin"}"#.to_vec(),
+            encoding: api::grpc::RawPayloadEncoding::JsonBytes as i32,
+        });
+
+        let err = PointStructRawPersisted::try_from(sent)
+            .expect_err("two payloads for one point must be rejected");
+
+        assert_eq!(err.code(), tonic::Code::InvalidArgument);
+    }
+
+    #[test]
+    fn decode_payload_raw_moves_blob_into_payload() {
+        let mut point = PointStructRawPersisted {
+            id: PointIdType::from(1),
+            vectors: RawVectorsPersisted::default(),
+            payload: None,
+            payload_raw: Some(RawPayload::from_storage_bytes(br#"{"a":1}"#.to_vec())),
+        };
+
+        point.decode_payload_raw().unwrap();
+        assert_eq!(
+            point.payload,
+            Some(serde_json::from_str(r#"{"a":1}"#).unwrap()),
+        );
+        assert!(point.payload_raw.is_none());
+
+        // Decoding is idempotent.
+        point.decode_payload_raw().unwrap();
+        assert_eq!(
+            point.payload,
+            Some(serde_json::from_str(r#"{"a":1}"#).unwrap()),
+        );
+
+        // A malformed blob fails, and leaves the point holding it rather than nothing.
+        let mut point = PointStructRawPersisted {
+            id: PointIdType::from(1),
+            vectors: RawVectorsPersisted::default(),
+            payload: None,
+            payload_raw: Some(RawPayload::from_storage_bytes(b"not json".to_vec())),
+        };
+        let err = point.decode_payload_raw().expect_err("must not parse");
+        assert!(
+            matches!(err, OperationError::MalformedPayloadBlob { .. }),
+            "{err:?}",
+        );
+        assert!(point.payload.is_none());
+        assert!(
+            point.payload_raw.is_some(),
+            "a failed decode must not consume the blob",
+        );
+    }
+
     fn dense(v: f32) -> VectorPersisted {
         VectorPersisted::Dense(vec![v])
     }
@@ -1173,7 +1343,7 @@ mod tests {
         let mut list =
             PointOperations::UpsertPoints(PointInsertOperationsInternal::PointsList(vec![
                 PointStructPersisted {
-                    id: 1.into(),
+                    id: PointIdType::from(1),
                     vector: VectorStructPersisted::Named(
                         [("a".to_string(), dense(0.1)), ("b".to_string(), dense(0.2))]
                             .into_iter()
@@ -1221,40 +1391,45 @@ mod tests {
         assert_eq!(batch.ids.len(), 2);
     }
 
-    /// A raw payload arriving over the wire is read by the same decoder as one
-    /// read from storage, and a blob that does not parse is reported as the bad
-    /// input it is.
+    /// The boundary does not parse the blob, so a malformed one only surfaces when the
+    /// point is applied.
     #[cfg(feature = "api")]
     #[test]
-    fn wire_raw_payload_is_decoded_by_the_shared_decoder() {
-        let expected: Payload = serde_json::from_str(r#"{"city": "Berlin", "count": 3}"#).unwrap();
+    fn wire_malformed_blob_fails_at_apply_not_at_the_boundary() {
+        let sent = api::grpc::qdrant::PointStructRaw::from(PointStructRawPersisted {
+            id: PointIdType::from(1),
+            vectors: RawVectorsPersisted::default(),
+            payload: None,
+            payload_raw: Some(RawPayload::from_storage_bytes(b"not json".to_vec())),
+        });
 
-        let received = decode_payload(api::grpc::RawPayload {
-            payload_bytes: br#"{"city": "Berlin", "count": 3}"#.to_vec(),
-            encoding: api::grpc::RawPayloadEncoding::JsonBytes as i32,
-        })
-        .expect("a well-formed blob must decode");
-        assert_eq!(received, expected);
+        let mut received =
+            PointStructRawPersisted::try_from(sent).expect("the boundary must not parse the blob");
 
-        let err = decode_payload(api::grpc::RawPayload {
-            payload_bytes: b"not json".to_vec(),
-            encoding: api::grpc::RawPayloadEncoding::JsonBytes as i32,
-        })
-        .expect_err("a malformed blob must not decode");
-        assert_eq!(err.code(), tonic::Code::InvalidArgument);
+        let err = received.decode_payload_raw().expect_err("must not parse");
+        assert!(
+            matches!(err, OperationError::MalformedPayloadBlob { .. }),
+            "{err:?}",
+        );
     }
 
-    /// An encoding this node has no variant for comes from a node that writes
-    /// payloads some other way: it must be rejected rather than read as the
-    /// default encoding.
+    /// Unlike the bytes, the tag is checked at the boundary: a blob this node could never
+    /// read must not reach the WAL.
     #[cfg(feature = "api")]
     #[test]
     fn wire_raw_payload_rejects_an_unknown_encoding() {
-        let err = decode_payload(api::grpc::RawPayload {
-            payload_bytes: br#"{"city": "Berlin"}"#.to_vec(),
-            encoding: 12345,
-        })
-        .expect_err("an unknown encoding must not decode");
+        let mut sent = api::grpc::qdrant::PointStructRaw::from(PointStructRawPersisted {
+            id: PointIdType::from(1),
+            vectors: RawVectorsPersisted::default(),
+            payload: None,
+            payload_raw: Some(RawPayload::from_storage_bytes(
+                br#"{"city":"Berlin"}"#.to_vec(),
+            )),
+        });
+        sent.raw_payload.as_mut().unwrap().encoding = 12345;
+
+        let err = PointStructRawPersisted::try_from(sent)
+            .expect_err("an unknown encoding must not be accepted");
 
         assert_eq!(err.code(), tonic::Code::InvalidArgument);
         assert!(

lib/shard/src/update/mod.rs

@@ -25,7 +25,7 @@ pub use self::points::{
 };
 pub use self::vectors::{delete_vectors, delete_vectors_by_filter, update_vectors_conditional};
 use crate::operations::payload_ops::PayloadOps;
-use crate::operations::point_ops::PointOperations;
+use crate::operations::point_ops::{PointOperations, PointStructRawPersisted};
 use crate::operations::vector_ops::VectorOperations;
 use crate::operations::{
     CreateVectorName, DeleteVectorName, FieldIndexOperations, VectorNameOperations,
@@ -67,15 +67,17 @@ pub fn process_point_operation(
             )?;
             Ok(deleted + new + updated)
         }
-        PointOperations::UpsertPointsRaw(points) => {
+        PointOperations::UpsertPointsRaw(mut points) => {
+            decode_raw_payloads(&mut points)?;
             if points.is_empty() {
                 // An empty upsert touches no segment; bump so WAL can acknowledge it.
                 segments.bump_max_segment_version_overwrite(op_num);
             }
-            let res = upsert_points_raw(segments, op_num, points.iter(), hw_counter)?;
+            let res = upsert_points_raw(segments, op_num, &points, hw_counter)?;
             Ok(res)
         }
-        PointOperations::SyncPointsRaw(operation) => {
+        PointOperations::SyncPointsRaw(mut operation) => {
+            decode_raw_payloads(&mut operation.points)?;
             let (deleted, new, updated) = sync_points_raw(
                 segments,
                 op_num,
@@ -89,6 +91,16 @@ pub fn process_point_operation(
     }
 }
 
+/// Decode the raw payload blobs of a batch into parsed payloads.
+///
+/// The one place a blob is parsed: it reaches the WAL as stored, and applying a point
+/// needs the parsed form because the payload index does.
+fn decode_raw_payloads(points: &mut [PointStructRawPersisted]) -> OperationResult<()> {
+    points
+        .iter_mut()
+        .try_for_each(PointStructRawPersisted::decode_payload_raw)
+}
+
 #[cfg(feature = "staging")]
 pub fn process_staging_operation(
     segments: &SegmentHolder,

lib/shard/src/update/points/sync.rs

@@ -45,6 +45,7 @@ pub fn sync_points_raw(
     points: &[PointStructRawPersisted],
     hw_counter: &HardwareCounterCell,
 ) -> OperationResult<(usize, usize, usize)> {
+    super::upsert::ensure_payloads_decoded(points)?;
     sync_points_impl(segments, op_num, from_id, to_id, points, hw_counter)
 }
 

lib/shard/src/update/points/upsert.rs

@@ -38,18 +38,28 @@ where
 }
 
 /// Same as [`upsert_points`], but for points carrying raw vector bytes verbatim.
-pub fn upsert_points_raw<'a, T>(
+pub fn upsert_points_raw(
     segments: &SegmentHolder,
     op_num: SeqNumberType,
-    points: T,
+    points: &[PointStructRawPersisted],
     hw_counter: &HardwareCounterCell,
-) -> OperationResult<usize>
-where
-    T: IntoIterator<Item = &'a PointStructRawPersisted>,
-{
+) -> OperationResult<usize> {
+    ensure_payloads_decoded(points)?;
     upsert_points_impl(segments, op_num, points, hw_counter)
 }
 
+/// Applying a point writes [`PointStructRawPersisted::payload`], so one still holding a
+/// blob would be stored with no payload at all. Blobs are decoded where the operation is
+/// unpacked (`process_point_operation`); this refuses a point that skipped that.
+pub(super) fn ensure_payloads_decoded(points: &[PointStructRawPersisted]) -> OperationResult<()> {
+    if points.iter().any(|point| point.payload_raw.is_some()) {
+        return Err(OperationError::service_error(
+            "Raw point reached the apply path with an undecoded payload blob",
+        ));
+    }
+    Ok(())
+}
+
 /// Drop from `points_op` every point that the conditional-upsert
 /// `update_mode` excludes, judged against current segment state.
 ///

lib/shard/src/update/tests.rs

@@ -7,14 +7,15 @@ use segment::entry::ReadSegmentEntry as _;
 use segment::entry::entry_point::SegmentEntry as _;
 use segment::payload_json;
 use segment::types::{
-    Condition, FieldCondition, Filter, Match, MatchValue, PayloadKeyType, ValueVariants,
+    Condition, FieldCondition, Filter, Match, MatchValue, PayloadKeyType, PointIdType,
+    ValueVariants,
 };
 use tempfile::Builder;
 
 use crate::fixtures::{
     build_segment_1, build_segment_2, empty_segment, empty_segment_with_deferred,
 };
-use crate::operations::point_ops::PointStructRawPersisted;
+use crate::operations::point_ops::{PointStructRawPersisted, RawVectorsPersisted};
 use crate::segment_holder::{FlushMode, SegmentHolder};
 use crate::update::{
     clear_payload_by_filter, create_field_index, delete_payload_by_filter, delete_points_by_filter,
@@ -94,6 +95,15 @@ fn retrieve_raw_record(
         .remove(&point_id.into())
 }
 
+/// A stored record turned into a point ready to apply, with the blob already decoded.
+fn incoming_raw_point(
+    record: segment::data_types::segment_record::SegmentRecordRaw,
+) -> PointStructRawPersisted {
+    let mut point = PointStructRawPersisted::from(record);
+    point.decode_payload_raw().unwrap();
+    point
+}
+
 /// The payload a raw record carries, parsed; `None` when the point has none
 /// stored or stores an empty one.
 fn stored_payload(
@@ -125,15 +135,17 @@ fn test_upsert_points_raw_moves_point_from_non_appendable() {
             id: 1.into(),
             vectors: vec![(DEFAULT_VECTOR_NAME.to_owned(), new_bytes.clone())].into(),
             payload: Some(payload.clone()),
+            payload_raw: None,
         },
         PointStructRawPersisted {
             id: 100.into(),
             vectors: vec![(DEFAULT_VECTOR_NAME.to_owned(), new_bytes.clone())].into(),
             payload: None,
+            payload_raw: None,
         },
     ];
 
-    let updated = upsert_points_raw(&holder, 100, points.iter(), &hw_counter).unwrap();
+    let updated = upsert_points_raw(&holder, 100, &points, &hw_counter).unwrap();
     assert_eq!(updated, 1);
 
     {
@@ -159,6 +171,40 @@ fn test_upsert_points_raw_moves_point_from_non_appendable() {
     assert_eq!(stored_payload(&record), None);
 }
 
+/// Applying reads the parsed payload, so a point that still holds a blob would be
+/// stored with no payload at all. Both raw entry points refuse it instead.
+#[test]
+fn test_apply_refuses_an_undecoded_payload_blob() {
+    let dir = Builder::new().prefix("segment_dir").tempdir().unwrap();
+    let hw_counter = HardwareCounterCell::new();
+
+    let mut holder = SegmentHolder::default();
+    holder.add_new(empty_segment(dir.path()));
+
+    let points = [PointStructRawPersisted {
+        id: PointIdType::from(1),
+        vectors: RawVectorsPersisted::from(vec![(
+            DEFAULT_VECTOR_NAME.to_owned(),
+            [1.0f32, 2.0, 3.0, 4.0]
+                .iter()
+                .flat_map(|v| v.to_le_bytes())
+                .collect(),
+        )]),
+        payload: None,
+        payload_raw: Some(segment::types::RawPayload::from_storage_bytes(
+            br#"{"city":"Berlin"}"#.to_vec(),
+        )),
+    }];
+
+    assert!(upsert_points_raw(&holder, 100, &points, &hw_counter).is_err());
+    assert!(sync_points_raw(&holder, 101, None, None, &points, &hw_counter).is_err());
+
+    assert!(
+        retrieve_raw_record(&holder, holder.iter().next().unwrap().0, 1).is_none(),
+        "a refused point must not be stored at all",
+    );
+}
+
 #[test]
 fn test_sync_points_raw() {
     let dir = Builder::new().prefix("segment_dir").tempdir().unwrap();
@@ -168,8 +214,7 @@ fn test_sync_points_raw() {
     let mut holder = SegmentHolder::default();
     let sid = holder.add_new(segment);
 
-    let point_2 =
-        PointStructRawPersisted::try_from(retrieve_raw_record(&holder, sid, 2).unwrap()).unwrap();
+    let point_2 = incoming_raw_point(retrieve_raw_record(&holder, sid, 2).unwrap());
     let point_2_version_before = holder
         .get(sid)
         .unwrap()
@@ -177,8 +222,7 @@ fn test_sync_points_raw() {
         .read()
         .point_version(2.into());
 
-    let mut point_3 =
-        PointStructRawPersisted::try_from(retrieve_raw_record(&holder, sid, 3).unwrap()).unwrap();
+    let mut point_3 = incoming_raw_point(retrieve_raw_record(&holder, sid, 3).unwrap());
     let changed_bytes: Vec<u8> = [9.0f32, 8.0, 7.0, 6.0]
         .iter()
         .flat_map(|v| v.to_le_bytes())
@@ -189,6 +233,7 @@ fn test_sync_points_raw() {
         id: 100.into(),
         vectors: vec![(DEFAULT_VECTOR_NAME.to_owned(), changed_bytes.clone())].into(),
         payload: None,
+        payload_raw: None,
     };
 
     let (deleted, new, updated) = sync_points_raw(

lib/storage/src/rbac/ops_checks.rs

@@ -799,6 +799,7 @@ mod tests_ops {
                         id: ExtendedPointId::NumId(12345),
                         vectors: vec![("dense".to_string(), vec![0, 1, 2, 3])].into(),
                         payload: None,
+                        payload_raw: None,
                     }]),
                 );
                 assert_requires_whole_write_access(&op);