qdrant_edge/edge/
update.rs

1use std::fmt;
2
3use crate::common::counter::hardware_counter::HardwareCounterCell;
4use crate::segment::common::operation_error::{OperationError, OperationResult};
5use crate::shard::operations::vector_name_ops::VectorNameConfig;
6use crate::shard::operations::{CollectionUpdateOperations, VectorNameOperations};
7use crate::shard::update::*;
8use crate::shard::wal::WalRawRecord;
9
10use crate::edge::EdgeShard;
11use crate::edge::config::vectors::{EdgeSparseVectorParams, EdgeVectorParams};
12
13impl EdgeShard {
14    pub fn update(&self, operation: CollectionUpdateOperations) -> OperationResult<()> {
15        let record = WalRawRecord::new(&operation).map_err(service_error)?;
16
17        let mut wal = self.wal.lock();
18
19        let operation_id = wal.write(&record).map_err(service_error)?;
20        let hw_counter = HardwareCounterCell::disposable();
21        let _update_guard = self.segments.acquire_updates_lock();
22
23        let segments_guard = self.segments.read();
24
25        let result = match operation {
26            CollectionUpdateOperations::PointOperation(point_operation) => {
27                process_point_operation(&segments_guard, operation_id, point_operation, &hw_counter)
28            }
29            CollectionUpdateOperations::VectorOperation(vector_operation) => {
30                process_vector_operation(
31                    &segments_guard,
32                    operation_id,
33                    vector_operation,
34                    &hw_counter,
35                )
36            }
37            CollectionUpdateOperations::PayloadOperation(payload_operation) => {
38                process_payload_operation(
39                    &segments_guard,
40                    operation_id,
41                    payload_operation,
42                    &hw_counter,
43                )
44            }
45            CollectionUpdateOperations::FieldIndexOperation(index_operation) => {
46                process_field_index_operation(
47                    &segments_guard,
48                    operation_id,
49                    &index_operation,
50                    &hw_counter,
51                )
52            }
53            CollectionUpdateOperations::VectorNameOperation(ref vector_name_operation) => {
54                let result = process_vector_name_operation(
55                    &segments_guard,
56                    operation_id,
57                    vector_name_operation,
58                );
59                // Also update the edge shard config so queries can resolve the vector name
60                if result.is_ok() {
61                    self.apply_vector_name_to_config(vector_name_operation)?;
62                }
63                result
64            }
65            #[cfg(feature = "staging")]
66            CollectionUpdateOperations::StagingOperation(staging_operation) => {
67                crate::shard::update::process_staging_operation(
68                    &segments_guard,
69                    operation_id,
70                    staging_operation,
71                )
72            }
73        };
74
75        result.map(|_| ())
76    }
77
78    /// Update the edge shard config to reflect a vector name create/delete operation.
79    fn apply_vector_name_to_config(&self, operation: &VectorNameOperations) -> OperationResult<()> {
80        match operation {
81            VectorNameOperations::CreateVectorName(create) => {
82                self.config
83                    .write(|config| match &create.config {
84                        VectorNameConfig::Dense(wrapper) => {
85                            config.vectors.insert(
86                                create.vector_name.clone(),
87                                EdgeVectorParams {
88                                    size: wrapper.dense.size,
89                                    distance: wrapper.dense.distance,
90                                    on_disk: None,
91                                    multivector_config: wrapper.dense.multivector_config,
92                                    datatype: wrapper.dense.datatype,
93                                    quantization_config: None,
94                                    hnsw_config: None,
95                                },
96                            );
97                        }
98                        VectorNameConfig::Sparse(wrapper) => {
99                            config.sparse_vectors.insert(
100                                create.vector_name.clone(),
101                                EdgeSparseVectorParams {
102                                    full_scan_threshold: None,
103                                    on_disk: None,
104                                    modifier: wrapper.sparse.modifier,
105                                    datatype: wrapper.sparse.datatype,
106                                },
107                            );
108                        }
109                    })
110                    .map_err(service_error)?;
111            }
112            VectorNameOperations::DeleteVectorName(delete) => {
113                self.config
114                    .write(|config| {
115                        config.vectors.remove(&delete.vector_name);
116                        config.sparse_vectors.remove(&delete.vector_name);
117                    })
118                    .map_err(service_error)?;
119            }
120        }
121        Ok(())
122    }
123}
124
125fn service_error(err: impl fmt::Display) -> OperationError {
126    OperationError::service_error(err.to_string())
127}
qdrant_edge/edge/update.rs

qdrant_edge/edge/
update.rs
