qdrant_edge/shard/operations/
mod.rs

1pub mod operation_name;
2pub mod optimization;
3pub mod payload_ops;
4pub mod point_ops;
5#[cfg(feature = "staging")]
6pub mod staging;
7pub mod vector_name_ops;
8pub mod vector_ops;
9
10use std::collections::HashSet;
11
12use crate::segment::json_path::JsonPath;
13use crate::segment::types::{PayloadFieldSchema, PointIdType, VectorNameBuf};
14use serde::{Deserialize, Serialize};
15use strum::{EnumDiscriminants, EnumIter};
16
17pub use self::vector_name_ops::{
18    CreateVectorName, DeleteVectorName, VectorNameConfig, VectorNameOperations,
19};
20use crate::shard::PeerId;
21use crate::shard::operations::point_ops::PointOperations;
22
23#[derive(Clone, Debug, PartialEq, Deserialize, Serialize, EnumDiscriminants, Hash)]
24#[strum_discriminants(derive(EnumIter))]
25#[serde(untagged, rename_all = "snake_case")]
26pub enum CollectionUpdateOperations {
27    PointOperation(point_ops::PointOperations),
28    VectorOperation(vector_ops::VectorOperations),
29    PayloadOperation(payload_ops::PayloadOps),
30    FieldIndexOperation(FieldIndexOperations),
31    VectorNameOperation(VectorNameOperations),
32    /// Staging-only operations for testing and debugging purposes
33    #[cfg(feature = "staging")]
34    StagingOperation(staging::StagingOperations),
35}
36
37impl CollectionUpdateOperations {
38    pub fn is_upsert_points(&self) -> bool {
39        matches!(
40            self,
41            Self::PointOperation(point_ops::PointOperations::UpsertPoints(_))
42        )
43    }
44
45    pub fn is_delete_points(&self) -> bool {
46        matches!(
47            self,
48            Self::PointOperation(point_ops::PointOperations::DeletePoints { .. })
49        )
50    }
51
52    pub fn point_ids(&self) -> Option<Vec<PointIdType>> {
53        match self {
54            Self::PointOperation(op) => op.point_ids(),
55            Self::VectorOperation(op) => op.point_ids(),
56            Self::PayloadOperation(op) => op.point_ids(),
57            Self::FieldIndexOperation(_) => None,
58            Self::VectorNameOperation(_) => None,
59            #[cfg(feature = "staging")]
60            Self::StagingOperation(_) => None,
61        }
62    }
63
64    /// Whether applying this operation grows what the node holds, which is what
65    /// [`crate::quota`] caps.
66    ///
67    /// Only deleting whole points is excluded, and it has to be: it is the one
68    /// way out of a node that has hit its limit. Deleting a vector or a payload
69    /// key is not — copy-on-write rewrites the point to drop a field, so it
70    /// grows storage before anything is reclaimed.
71    ///
72    /// Shard-transfer syncs are excluded as well: a transfer is sized up once
73    /// before it starts, and refusing its batches partway would abandon work
74    /// that is nearly done, only for the whole transfer to be retried from the
75    /// beginning.
76    pub fn consumes_quota(&self) -> bool {
77        match self {
78            Self::PointOperation(op) => match op {
79                PointOperations::UpsertPoints(_)
80                | PointOperations::UpsertPointsConditional(_)
81                | PointOperations::UpsertPointsRaw(_) => true,
82                PointOperations::DeletePoints { .. } | PointOperations::DeletePointsByFilter(_) => {
83                    false
84                }
85                PointOperations::SyncPoints(_) | PointOperations::SyncPointsRaw(_) => false,
86            },
87            Self::VectorOperation(op) => match op {
88                vector_ops::VectorOperations::UpdateVectors(_) => true,
89                // With CoW all modifications to points create more load.
90                vector_ops::VectorOperations::DeleteVectors(..)
91                | vector_ops::VectorOperations::DeleteVectorsByFilter(..) => true,
92            },
93            Self::PayloadOperation(op) => match op {
94                payload_ops::PayloadOps::SetPayload(_)
95                | payload_ops::PayloadOps::OverwritePayload(_) => true,
96                // With CoW all modifications to points create more load.
97                payload_ops::PayloadOps::DeletePayload(_)
98                | payload_ops::PayloadOps::ClearPayload { .. }
99                | payload_ops::PayloadOps::ClearPayloadByFilter(_) => true,
100            },
101            // Both arrive already committed through consensus, on a path that
102            // does not go past a quota check — they are gated before the
103            // proposal instead, so that a peer cannot refuse what the cluster
104            // has already agreed to.
105            Self::FieldIndexOperation(_) | Self::VectorNameOperation(_) => false,
106            #[cfg(feature = "staging")]
107            Self::StagingOperation(_) => false,
108        }
109    }
110
111    /// List point IDs that can be created during the operation.
112    /// Do not list IDs that are deleted or modified.
113    pub fn upsert_point_ids(&self) -> Option<Vec<PointIdType>> {
114        match self {
115            Self::PointOperation(op) => match op {
116                PointOperations::UpsertPoints(op) => Some(op.point_ids()),
117                PointOperations::UpsertPointsConditional(op) => Some(op.points_op.point_ids()),
118                PointOperations::DeletePoints { .. } => None,
119                PointOperations::DeletePointsByFilter(_) => None,
120                PointOperations::SyncPoints(op) => {
121                    Some(op.points.iter().map(|point| point.id).collect())
122                }
123                PointOperations::UpsertPointsRaw(points) => {
124                    Some(points.iter().map(|point| point.id).collect())
125                }
126                PointOperations::SyncPointsRaw(op) => {
127                    Some(op.points.iter().map(|point| point.id).collect())
128                }
129            },
130            Self::VectorOperation(_) => None,
131            Self::PayloadOperation(_) => None,
132            Self::FieldIndexOperation(_) => None,
133            Self::VectorNameOperation(_) => None,
134            #[cfg(feature = "staging")]
135            Self::StagingOperation(_) => None,
136        }
137    }
138
139    pub fn retain_point_ids<F>(&mut self, filter: F)
140    where
141        F: Fn(&PointIdType) -> bool,
142    {
143        match self {
144            Self::PointOperation(op) => op.retain_point_ids(filter),
145            Self::VectorOperation(op) => op.retain_point_ids(filter),
146            Self::PayloadOperation(op) => op.retain_point_ids(filter),
147            Self::FieldIndexOperation(_) => (),
148            Self::VectorNameOperation(_) => (),
149            #[cfg(feature = "staging")]
150            Self::StagingOperation(_) => (),
151        }
152    }
153
154    /// Drop named-vector references to vector names not in `valid`.
155    ///
156    /// Used during WAL replay: a historical operation may reference a vector name that was
157    /// since removed by `delete_named_vector`. Without this, such an operation fails segment
158    /// validation (`VectorNameNotExists`) and is dropped wholesale on reload, taking its
159    /// points with it. Stripping the dead names lets the rest of the operation apply, matching
160    /// the live outcome (the point survives, just without the deleted vector).
161    ///
162    /// This does not touch `VectorNameOperation` responsible for creating/deleting a named vector.
163    ///
164    /// Only affects the named-vector variants; the default (unnamed) vector is left untouched.
165    ///
166    /// Note: this is best-effort. Stripping a vector from an early operation silently changes the
167    /// behavior of a later operation that depended on it (e.g. a `has_vector` filter or
168    /// `UpdateVectors`), so the replayed timeline can still diverge from the live one. Tracked in
169    /// <https://github.com/qdrant/qdrant/issues/9386>.
170    pub fn retain_vector_names(&mut self, valid: &HashSet<VectorNameBuf>) {
171        match self {
172            Self::PointOperation(op) => op.retain_vector_names(valid),
173            Self::VectorOperation(op) => op.retain_vector_names(valid),
174            Self::PayloadOperation(_) => (),
175            Self::FieldIndexOperation(_) => (),
176            Self::VectorNameOperation(_) => (),
177            #[cfg(feature = "staging")]
178            Self::StagingOperation(_) => (),
179        }
180    }
181
182    /// If this operation creates a named vector, return the name it introduces.
183    ///
184    /// Used during WAL replay to grow the set of valid vector names: a historical
185    /// `CreateVectorName` must make its name valid for the operations that follow it in
186    /// the WAL, otherwise a later upsert referencing it would be wrongly stripped by
187    /// [`Self::retain_vector_names`].
188    pub fn created_vector_name(&self) -> Option<&VectorNameBuf> {
189        match self {
190            Self::VectorNameOperation(VectorNameOperations::CreateVectorName(op)) => {
191                Some(&op.vector_name)
192            }
193            Self::VectorNameOperation(VectorNameOperations::DeleteVectorName(_))
194            | Self::PointOperation(_)
195            | Self::VectorOperation(_)
196            | Self::PayloadOperation(_)
197            | Self::FieldIndexOperation(_) => None,
198            #[cfg(feature = "staging")]
199            Self::StagingOperation(_) => None,
200        }
201    }
202}
203
204#[derive(Clone, Debug, PartialEq, Deserialize, Serialize, EnumDiscriminants, Hash)]
205#[strum_discriminants(derive(EnumIter))]
206#[serde(rename_all = "snake_case")]
207pub enum FieldIndexOperations {
208    /// Create index for payload field
209    CreateIndex(CreateIndex),
210    /// Delete index for the field
211    DeleteIndex(JsonPath),
212}
213
214#[derive(Clone, Debug, PartialEq, Deserialize, Serialize, Hash)]
215#[serde(rename_all = "snake_case")]
216pub struct CreateIndex {
217    pub field_name: JsonPath,
218    pub field_schema: Option<PayloadFieldSchema>,
219}
220
221#[derive(Clone, Debug, PartialEq, Deserialize, Serialize)]
222pub struct OperationWithClockTag {
223    #[serde(flatten)]
224    pub operation: CollectionUpdateOperations,
225
226    #[serde(default, skip_serializing_if = "Option::is_none")]
227    pub clock_tag: Option<ClockTag>,
228}
229
230impl OperationWithClockTag {
231    pub fn new(
232        operation: impl Into<CollectionUpdateOperations>,
233        clock_tag: Option<ClockTag>,
234    ) -> Self {
235        Self {
236            operation: operation.into(),
237            clock_tag,
238        }
239    }
240}
241
242impl From<CollectionUpdateOperations> for OperationWithClockTag {
243    fn from(operation: CollectionUpdateOperations) -> Self {
244        Self::new(operation, None)
245    }
246}
247
248#[derive(Copy, Clone, Debug, Eq, PartialEq, Deserialize, Serialize)]
249pub struct ClockTag {
250    pub peer_id: PeerId,
251    pub clock_id: u32,
252    pub clock_tick: u64,
253    /// A unique token for each clock tag.
254    pub token: ClockToken,
255    pub force: bool,
256}
257
258pub type ClockToken = u64;
259
260impl ClockTag {
261    pub fn new(peer_id: PeerId, clock_id: u32, clock_tick: u64) -> Self {
262        let random_token = rand::random();
263        Self::new_with_token(peer_id, clock_id, clock_tick, random_token)
264    }
265
266    pub fn new_with_token(
267        peer_id: PeerId,
268        clock_id: u32,
269        clock_tick: u64,
270        token: ClockToken,
271    ) -> Self {
272        Self {
273            peer_id,
274            clock_id,
275            clock_tick,
276            token,
277            force: false,
278        }
279    }
280
281    pub fn force(mut self, force: bool) -> Self {
282        self.force = force;
283        self
284    }
285}
286
287#[cfg(feature = "api")]
288impl From<api::grpc::qdrant::ClockTag> for ClockTag {
289    fn from(tag: api::grpc::qdrant::ClockTag) -> Self {
290        let api::grpc::qdrant::ClockTag {
291            peer_id,
292            clock_id,
293            clock_tick,
294            token,
295            force,
296        } = tag;
297        Self {
298            peer_id,
299            clock_id,
300            clock_tick,
301            token,
302            force,
303        }
304    }
305}
306
307#[cfg(feature = "api")]
308impl From<ClockTag> for api::grpc::qdrant::ClockTag {
309    fn from(tag: ClockTag) -> Self {
310        let ClockTag {
311            peer_id,
312            clock_id,
313            clock_tick,
314            token,
315            force,
316        } = tag;
317        Self {
318            peer_id,
319            clock_id,
320            clock_tick,
321            token,
322            force,
323        }
324    }
325}
326
327#[cfg(test)]
328mod tests {
329    #![expect(clippy::wildcard_enum_match_arm, reason = "test code")]
330
331    use proptest::prelude::*;
332    use crate::segment::types::*;
333
334    use super::payload_ops::*;
335    use super::point_ops::*;
336    use super::vector_ops::*;
337    use super::*;
338
339    proptest::proptest! {
340        #[test]
341        fn operation_with_clock_tag_json(operation in any::<OperationWithClockTag>()) {
342            // Assert that `OperationWithClockTag` can be serialized
343            let input = serde_json::to_string(&operation).unwrap();
344            let output: OperationWithClockTag = serde_json::from_str(&input).unwrap();
345            assert_eq!(operation, output);
346
347            // Assert that `OperationWithClockTag` can be deserialized from `CollectionUpdateOperation`
348            let input = serde_json::to_string(&operation.operation).unwrap();
349            let output: OperationWithClockTag = serde_json::from_str(&input).unwrap();
350            assert_eq!(operation.operation, output.operation);
351
352            // Assert that `CollectionUpdateOperation` serializes into JSON object with a single key
353            // (e.g., `{ "upsert_points": <upsert points object> }`)
354            match serde_json::to_value(&operation.operation).unwrap() {
355                serde_json::Value::Object(map) if map.len() == 1 => (),
356                _ => panic!("TODO"),
357            };
358        }
359
360        #[test]
361        fn operation_with_clock_tag_cbor(operation in any::<OperationWithClockTag>()) {
362            // Assert that `OperationWithClockTag` can be serialized
363            let input = serde_cbor::to_vec(&operation).unwrap();
364            let output: OperationWithClockTag = serde_cbor::from_slice(&input).unwrap();
365            assert_eq!(operation, output);
366
367            // Assert that `OperationWithClockTag` can be deserialized from `CollectionUpdateOperation`
368            let input = serde_cbor::to_vec(&operation.operation).unwrap();
369            let output: OperationWithClockTag = serde_cbor::from_slice(&input).unwrap();
370            assert_eq!(operation.operation, output.operation);
371        }
372    }
373
374    impl Arbitrary for OperationWithClockTag {
375        type Parameters = ();
376        type Strategy = BoxedStrategy<Self>;
377
378        fn arbitrary_with(_: Self::Parameters) -> Self::Strategy {
379            any::<(CollectionUpdateOperations, Option<ClockTag>)>()
380                .prop_map(|(operation, clock_tag)| Self::new(operation, clock_tag))
381                .boxed()
382        }
383    }
384
385    impl Arbitrary for ClockTag {
386        type Parameters = ();
387        type Strategy = BoxedStrategy<Self>;
388
389        fn arbitrary_with(_: Self::Parameters) -> Self::Strategy {
390            any::<(PeerId, u32, u64)>()
391                .prop_map(|(peer_id, clock_id, clock_tick)| {
392                    Self::new(peer_id, clock_id, clock_tick)
393                })
394                .boxed()
395        }
396    }
397
398    impl Arbitrary for CollectionUpdateOperations {
399        type Parameters = ();
400        type Strategy = BoxedStrategy<Self>;
401
402        fn arbitrary_with(_: Self::Parameters) -> Self::Strategy {
403            prop_oneof![
404                any::<point_ops::PointOperations>().prop_map(Self::PointOperation),
405                any::<vector_ops::VectorOperations>().prop_map(Self::VectorOperation),
406                any::<payload_ops::PayloadOps>().prop_map(Self::PayloadOperation),
407                any::<FieldIndexOperations>().prop_map(Self::FieldIndexOperation),
408                any::<VectorNameOperations>().prop_map(Self::VectorNameOperation),
409            ]
410            .boxed()
411        }
412    }
413
414    impl Arbitrary for point_ops::PointOperations {
415        type Parameters = ();
416        type Strategy = BoxedStrategy<Self>;
417
418        fn arbitrary_with(_: Self::Parameters) -> Self::Strategy {
419            let upsert = Self::UpsertPoints(PointInsertOperationsInternal::PointsList(Vec::new()));
420            let delete = Self::DeletePoints { ids: Vec::new() };
421
422            let delete_by_filter = Self::DeletePointsByFilter(Filter {
423                should: None,
424                min_should: None,
425                must: None,
426                must_not: None,
427            });
428
429            let sync = Self::SyncPoints(PointSyncOperation {
430                from_id: None,
431                to_id: None,
432                points: Vec::new(),
433            });
434
435            // Use a non-empty raw point so the byte-blob path is actually exercised
436            let raw_point = PointStructRawPersisted {
437                id: 1.into(),
438                vectors: vec![("dense".to_string(), vec![0, 1, 2, 3, 255])].into(),
439                payload: None,
440            };
441
442            let upsert_raw = Self::UpsertPointsRaw(vec![raw_point.clone()]);
443
444            let sync_raw = Self::SyncPointsRaw(PointSyncRawOperation {
445                from_id: Some(1.into()),
446                to_id: None,
447                points: vec![raw_point],
448            });
449
450            prop_oneof![
451                Just(upsert),
452                Just(delete),
453                Just(delete_by_filter),
454                Just(sync),
455                Just(upsert_raw),
456                Just(sync_raw),
457            ]
458            .boxed()
459        }
460    }
461
462    impl Arbitrary for vector_ops::VectorOperations {
463        type Parameters = ();
464        type Strategy = BoxedStrategy<Self>;
465
466        fn arbitrary_with(_: Self::Parameters) -> Self::Strategy {
467            let update = Self::UpdateVectors(UpdateVectorsOp {
468                points: Vec::new(),
469                update_filter: None,
470            });
471
472            let delete = Self::DeleteVectors(
473                PointIdsList {
474                    points: Vec::new(),
475                    #[cfg(feature = "api")]
476                    shard_key: None,
477                },
478                Vec::new(),
479            );
480
481            let delete_by_filter = Self::DeleteVectorsByFilter(
482                Filter {
483                    should: None,
484                    min_should: None,
485                    must: None,
486                    must_not: None,
487                },
488                Vec::new(),
489            );
490
491            prop_oneof![Just(update), Just(delete), Just(delete_by_filter),].boxed()
492        }
493    }
494
495    impl Arbitrary for payload_ops::PayloadOps {
496        type Parameters = ();
497        type Strategy = BoxedStrategy<Self>;
498
499        fn arbitrary_with(_: Self::Parameters) -> Self::Strategy {
500            let set = Self::SetPayload(SetPayloadOp {
501                payload: Payload(Default::default()),
502                points: None,
503                filter: None,
504                key: None,
505            });
506
507            let overwrite = Self::OverwritePayload(SetPayloadOp {
508                payload: Payload(Default::default()),
509                points: None,
510                filter: None,
511                key: None,
512            });
513
514            let delete = Self::DeletePayload(DeletePayloadOp {
515                keys: Vec::new(),
516                points: None,
517                filter: None,
518            });
519
520            let clear = Self::ClearPayload { points: Vec::new() };
521
522            let clear_by_filter = Self::ClearPayloadByFilter(Filter {
523                should: None,
524                min_should: None,
525                must: None,
526                must_not: None,
527            });
528
529            prop_oneof![
530                Just(set),
531                Just(overwrite),
532                Just(delete),
533                Just(clear),
534                Just(clear_by_filter),
535            ]
536            .boxed()
537        }
538    }
539
540    impl Arbitrary for FieldIndexOperations {
541        type Parameters = ();
542        type Strategy = BoxedStrategy<Self>;
543
544        fn arbitrary_with(_: Self::Parameters) -> Self::Strategy {
545            let create = Self::CreateIndex(CreateIndex {
546                field_name: "field_name".parse().unwrap(),
547                field_schema: None,
548            });
549
550            let delete = Self::DeleteIndex("field_name".parse().unwrap());
551
552            prop_oneof![Just(create), Just(delete),].boxed()
553        }
554    }
555
556    impl Arbitrary for VectorNameOperations {
557        type Parameters = ();
558        type Strategy = BoxedStrategy<Self>;
559
560        fn arbitrary_with(_: Self::Parameters) -> Self::Strategy {
561            use crate::shard::operations::vector_name_ops::{
562                self as vnops, DenseVectorConfig, SparseVectorConfig,
563            };
564
565            let create_dense = Self::CreateVectorName(CreateVectorName {
566                vector_name: "test_vector".into(),
567                config: vnops::VectorNameConfig::dense(DenseVectorConfig {
568                    size: 4,
569                    distance: Distance::Cosine,
570                    multivector_config: None,
571                    datatype: None,
572                }),
573            });
574
575            let create_sparse = Self::CreateVectorName(CreateVectorName {
576                vector_name: "sparse_test".into(),
577                config: vnops::VectorNameConfig::sparse(SparseVectorConfig {
578                    modifier: None,
579                    datatype: None,
580                }),
581            });
582
583            let delete = Self::DeleteVectorName(DeleteVectorName {
584                vector_name: "test_vector".into(),
585            });
586
587            prop_oneof![Just(create_dense), Just(create_sparse), Just(delete),].boxed()
588        }
589    }
590
591    #[test]
592    fn test_delete_by_filter_with_has_id_uuids_cbor_roundtrip() {
593        let uuids: Vec<PointIdType> = vec![ExtendedPointId::Uuid(
594            uuid::Uuid::parse_str("6ba7b810-9dad-11d1-80b4-00c04fd430c8").unwrap(),
595        )];
596
597        let filter = Filter {
598            should: None,
599            min_should: None,
600            must: None,
601            must_not: Some(vec![Condition::HasId(HasIdCondition::from(
602                uuids.into_iter().collect::<ahash::AHashSet<_>>(),
603            ))]),
604        };
605
606        let operation = CollectionUpdateOperations::PointOperation(
607            PointOperations::DeletePointsByFilter(filter),
608        );
609
610        let cbor_bytes = serde_cbor::to_vec(&operation).unwrap();
611        let deserialized: CollectionUpdateOperations = serde_cbor::from_slice(&cbor_bytes).unwrap();
612
613        assert_eq!(operation, deserialized);
614    }
615
616    #[test]
617    fn test_wal_roundtrip_delete_by_filter_with_has_id_uuids() {
618        use crate::shard::wal::WalRawRecord;
619
620        let uuids: Vec<PointIdType> = vec![ExtendedPointId::Uuid(
621            uuid::Uuid::parse_str("6ba7b810-9dad-11d1-80b4-00c04fd430c8").unwrap(),
622        )];
623
624        let filter = Filter {
625            should: None,
626            min_should: None,
627            must: None,
628            must_not: Some(vec![Condition::HasId(HasIdCondition::from(
629                uuids.into_iter().collect::<ahash::AHashSet<_>>(),
630            ))]),
631        };
632
633        let operation = CollectionUpdateOperations::PointOperation(
634            PointOperations::DeletePointsByFilter(filter),
635        );
636
637        let raw = WalRawRecord::new(&operation).unwrap();
638        let deserialized: CollectionUpdateOperations = raw.deserialize().unwrap();
639
640        assert_eq!(operation, deserialized);
641    }
642}
qdrant_edge/shard/operations/mod.rs

qdrant_edge/shard/operations/
mod.rs
