qdrant_edge/edge/update_only/
apply.rs

1//! Applying a folded batch: locate, resolve, materialize, append — each one
2//! batched pass over the whole point set, so a batch's cost scales with the
3//! points it touches, not the operations in it.
4
5use ahash::AHashMap;
6use crate::common::counter::hardware_counter::HardwareCounterCell;
7use crate::common::types::PointOffsetType;
8use crate::common::universal_io::UniversalRead;
9use rayon::ThreadPool;
10use rayon::prelude::*;
11use crate::segment::common::operation_error::{OperationError, OperationResult};
12use crate::segment::data_types::fully_qualified_point::{FullyQualifiedPoint, StoredPoint};
13use crate::segment::types::{PointIdType, SeqNumberType};
14use crate::shard::operations::CollectionUpdateOperations;
15use uuid::Uuid;
16
17use crate::edge::update_only::UpdateOnlyEdgeShard;
18use crate::edge::update_only::batch::UpdateBatchPlan;
19use crate::edge::update_only::holder::UpdateOnlySegmentHolder;
20use crate::edge::update_only::preview::{PointAction, PointPreview, resolve_batch};
21
22/// What a batch did, counted per point rather than per operation: a point
23/// named by ten operations counts once.
24#[derive(Debug, Default, Clone, PartialEq, Eq)]
25pub struct UpdateBatchOutcome {
26    /// Points written: created, or rewritten into a fresh slot.
27    pub stored: usize,
28    /// Points removed.
29    pub deleted: usize,
30    /// Points already at or beyond the batch's version, left untouched — what
31    /// makes a replayed batch a no-op.
32    pub skipped: usize,
33    /// Points an operation named that no segment holds and the batch did not
34    /// create — a payload update to a point that is not there.
35    pub missing: usize,
36}
37
38/// One copy of a point: where it lives, and at what version.
39#[derive(Debug, Clone, Copy)]
40pub(super) struct PointLocation {
41    pub(super) segment: Uuid,
42    pub(super) internal_id: PointOffsetType,
43    pub(super) version: SeqNumberType,
44    /// Whether the holding segment accepts appends; breaks a version tie.
45    appendable: bool,
46}
47
48impl PointLocation {
49    /// Whether this copy of the point supersedes `other`: the higher version
50    /// wins, and on a tie the appendable copy is the live one (a point being
51    /// moved between segments exists in both at the same version).
52    fn supersedes(&self, other: &Self) -> bool {
53        (self.version, self.appendable) > (other.version, other.appendable)
54    }
55}
56
57/// Every copy of one point across the shard's segments.
58pub(super) struct PointLocations {
59    /// The live copy: its version decides whether the batch is already
60    /// applied, and its slot is the one a resolve reads from.
61    pub(super) newest: PointLocation,
62    /// Every slot the point occupies, `newest`'s included. A rewrite or a
63    /// delete retires them all — tombstoning only the newest slot would let
64    /// an older duplicate (left by an interrupted move) outlive the point
65    /// and, on a delete, resurrect it.
66    pub(super) slots: Vec<(Uuid, PointOffsetType)>,
67}
68
69impl<S: UniversalRead + 'static> UpdateOnlyEdgeShard<S> {
70    /// Apply a batch of update operations, each paired with the operation
71    /// number to record as its version. Operations are expected in ascending
72    /// operation-number order; see [`UpdateBatchPlan::build`] for what is
73    /// rejected.
74    ///
75    /// Atomic in the sense that matters without a WAL: applied in full or the
76    /// error is returned, and re-applying a batch that partially landed skips
77    /// the points that already carry its version.
78    pub fn apply_batch(
79        &self,
80        operations: impl IntoIterator<Item = (SeqNumberType, CollectionUpdateOperations)>,
81    ) -> OperationResult<UpdateBatchOutcome> {
82        let plan = UpdateBatchPlan::build(operations)?;
83        if plan.is_empty() {
84            return Ok(UpdateBatchOutcome::default());
85        }
86
87        let hw_counter = HardwareCounterCell::disposable();
88        let segments = self.segments.read();
89
90        // 1-3. Locate, read, materialize — the decision stage shared with
91        // `preview_batch`, so a preview cannot drift from the real apply.
92        let resolved = resolve_batch(&segments, plan, &self.pool)?;
93
94        let mut outcome = UpdateBatchOutcome::default();
95        let mut to_store: Vec<FullyQualifiedPoint> = Vec::new();
96        let mut to_tombstone: AHashMap<Uuid, Vec<PointOffsetType>> = AHashMap::new();
97
98        for point in resolved {
99            let PointPreview {
100                id: _,
101                current: _,
102                slots,
103                action,
104            } = point;
105
106            match action {
107                PointAction::Skip => {
108                    outcome.skipped += 1;
109                    continue;
110                }
111                PointAction::Missing => {
112                    outcome.missing += 1;
113                    continue;
114                }
115                PointAction::Store(point) => {
116                    to_store.push(*point);
117                    outcome.stored += 1;
118                }
119                PointAction::Delete => outcome.deleted += 1,
120            }
121
122            // Whatever happened to the point, every slot it occupied — in any
123            // segment — is retired: a rewrite left its replacement elsewhere,
124            // a delete left nothing, and an older duplicate must not outlive
125            // either.
126            for (segment, internal_id) in slots {
127                to_tombstone.entry(segment).or_default().push(internal_id);
128            }
129        }
130
131        // 4. Append the resolved points, then retire the slots they replaced.
132        if !to_store.is_empty() {
133            let write_target = segments.write_target()?;
134            write_target.write().store_points(&to_store, &hw_counter)?;
135
136            // The new slots must be durable before the tombstones that retire
137            // the old ones: the reverse order can lose a point outright if the
138            // process dies in between.
139            write_target.read().flush()?;
140        }
141
142        for (uuid, internal_ids) in to_tombstone {
143            let segment = segments.get(&uuid).ok_or_else(|| {
144                OperationError::service_error(format!("Segment {uuid} disappeared mid-batch"))
145            })?;
146            segment.write().tombstone_points(&internal_ids)?;
147            segment.read().flush()?;
148        }
149
150        Ok(outcome)
151    }
152}
153
154/// Locate every point the batch touches: every slot it occupies, with the
155/// newest copy marked, when more than one segment holds the point. Segments
156/// are visited in parallel on `pool`.
157pub(super) fn locate_points<S: UniversalRead + 'static>(
158    segments: &UpdateOnlySegmentHolder<S>,
159    plan: &UpdateBatchPlan,
160    pool: &ThreadPool,
161) -> OperationResult<AHashMap<PointIdType, PointLocations>> {
162    let ids: Vec<PointIdType> = plan.point_ids().collect();
163
164    let per_segment: Vec<Vec<(PointIdType, PointLocation)>> = pool.install(|| {
165        segments
166            .iter()
167            .collect::<Vec<_>>()
168            .into_par_iter()
169            .map(|(uuid, segment)| {
170                let segment = segment.read();
171                let appendable = segment.is_appendable();
172
173                let mut found_ids = Vec::new();
174                let mut internal_ids = Vec::new();
175                segment.locate_points(ids.iter().copied(), |id, internal_id| {
176                    found_ids.push(id);
177                    internal_ids.push(internal_id);
178                })?;
179                let versions = segment.point_versions(&internal_ids)?;
180
181                let located = found_ids
182                    .into_iter()
183                    .zip(internal_ids)
184                    .map(|(id, internal_id)| {
185                        let location = PointLocation {
186                            segment: uuid,
187                            internal_id,
188                            // A slot without a stored version is unwritten,
189                            // which compares as version 0.
190                            version: versions.get(&internal_id).copied().unwrap_or(0),
191                            appendable,
192                        };
193                        (id, location)
194                    })
195                    .collect();
196                Ok(located)
197            })
198            .collect::<OperationResult<Vec<_>>>()
199    })?;
200
201    let mut locations: AHashMap<PointIdType, PointLocations> = AHashMap::new();
202    for (id, location) in per_segment.into_iter().flatten() {
203        let slot = (location.segment, location.internal_id);
204        locations
205            .entry(id)
206            .and_modify(|current| {
207                current.slots.push(slot);
208                if location.supersedes(&current.newest) {
209                    current.newest = location;
210                }
211            })
212            .or_insert_with(|| PointLocations {
213                newest: location,
214                slots: vec![slot],
215            });
216    }
217
218    Ok(locations)
219}
220
221/// Read the stored form of the points whose mutations need it, one batched
222/// pass per segment; segments are read in parallel on `pool`.
223pub(super) fn read_stored_points<S: UniversalRead + 'static>(
224    segments: &UpdateOnlySegmentHolder<S>,
225    plan: &UpdateBatchPlan,
226    locations: &AHashMap<PointIdType, PointLocations>,
227    pool: &ThreadPool,
228) -> OperationResult<AHashMap<PointIdType, StoredPoint>> {
229    let mut by_segment: AHashMap<Uuid, Vec<(PointIdType, PointOffsetType)>> = AHashMap::new();
230    for id in plan.point_ids_needing_stored_point() {
231        // A point no segment holds has nothing to read; its mutations either
232        // create it outright or resolve to nothing. Only the newest copy is
233        // read — older duplicates are stale.
234        if let Some(location) = locations.get(&id) {
235            by_segment
236                .entry(location.newest.segment)
237                .or_default()
238                .push((id, location.newest.internal_id));
239        }
240    }
241
242    let per_segment: Vec<Vec<(PointIdType, StoredPoint)>> = pool.install(|| {
243        by_segment
244            .into_iter()
245            .collect::<Vec<_>>()
246            .into_par_iter()
247            .map(|(uuid, entries)| {
248                let segment = segments.get(&uuid).ok_or_else(|| {
249                    OperationError::service_error(format!("Segment {uuid} disappeared mid-batch"))
250                })?;
251                let segment = segment.read();
252
253                let internal_ids: Vec<PointOffsetType> = entries
254                    .iter()
255                    .map(|(_, internal_id)| *internal_id)
256                    .collect();
257                // Not shared with the caller's counter: `HardwareCounterCell`
258                // is not `Sync`, and the writer's accounting is disposable.
259                let hw_counter = HardwareCounterCell::disposable();
260                let points = segment.read_stored_points(&internal_ids, &hw_counter)?;
261
262                Ok(entries.into_iter().map(|(id, _)| id).zip(points).collect())
263            })
264            .collect::<OperationResult<Vec<_>>>()
265    })?;
266
267    let mut stored = AHashMap::new();
268    for (id, point) in per_segment.into_iter().flatten() {
269        stored.insert(id, point);
270    }
271
272    Ok(stored)
273}
qdrant_edge/edge/update_only/apply.rs

qdrant_edge/edge/update_only/
apply.rs
