qdrant_edge/edge/update_only/
preview.rs

1//! Dry-run of a batch: the apply pipeline run up to — but not including — the
2//! writes.
3//!
4//! [`preview_batch`] and [`apply_batch`] share one resolution stage
5//! ([`resolve_batch`]), so a preview reports exactly what an apply would do.
6//!
7//! [`preview_batch`]: UpdateOnlyEdgeShard::preview_batch
8//! [`apply_batch`]: UpdateOnlyEdgeShard::apply_batch
9
10use crate::common::types::PointOffsetType;
11use crate::common::universal_io::UniversalRead;
12use rayon::ThreadPool;
13use crate::segment::common::operation_error::OperationResult;
14use crate::segment::data_types::fully_qualified_point::FullyQualifiedPoint;
15use crate::segment::types::{PointIdType, SeqNumberType};
16use crate::shard::operations::CollectionUpdateOperations;
17use uuid::Uuid;
18
19use crate::edge::update_only::UpdateOnlyEdgeShard;
20use crate::edge::update_only::apply::{locate_points, read_stored_points};
21use crate::edge::update_only::batch::UpdateBatchPlan;
22use crate::edge::update_only::holder::UpdateOnlySegmentHolder;
23
24/// A resolved batch: one entry per touched point, in first-touched order.
25pub struct UpdateBatchPreview {
26    pub points: Vec<PointPreview>,
27}
28
29/// What the batch does to one point.
30pub struct PointPreview {
31    pub id: PointIdType,
32    /// The newest stored copy of the point; `None` when no segment holds it.
33    pub current: Option<PointCopy>,
34    /// Every slot the point occupies across segments, the newest's included —
35    /// all of them are tombstoned when the action stores or deletes the point.
36    pub slots: Vec<(Uuid, PointOffsetType)>,
37    pub action: PointAction,
38}
39
40/// One stored copy of a point: which segment holds it, in which slot, at what
41/// version.
42pub struct PointCopy {
43    pub segment: Uuid,
44    pub internal_id: PointOffsetType,
45    pub version: SeqNumberType,
46}
47
48/// The write one point's folded mutations resolved to.
49pub enum PointAction {
50    /// The point is appended to the write target in this fully qualified
51    /// form, and every slot in [`PointPreview::slots`] is tombstoned.
52    /// Boxed: a resolved point is hundreds of bytes, the other variants none.
53    Store(Box<FullyQualifiedPoint>),
54    /// The point is removed: every slot is tombstoned, nothing is stored.
55    Delete,
56    /// Left untouched: the stored copy is already at or beyond the batch's
57    /// version, so re-applying would move the point backwards.
58    Skip,
59    /// An operation that can only modify an existing point named one that no
60    /// segment holds; there is nothing to write.
61    Missing,
62}
63
64/// Resolve a folded batch against the segments: locate every touched point,
65/// read the ones whose mutations need the stored form, and materialize each
66/// into its [`PointAction`]. Reads only — the single decision stage behind
67/// both [`UpdateOnlyEdgeShard::preview_batch`] and
68/// [`UpdateOnlyEdgeShard::apply_batch`].
69pub(super) fn resolve_batch<S: UniversalRead + 'static>(
70    segments: &UpdateOnlySegmentHolder<S>,
71    plan: UpdateBatchPlan,
72    pool: &ThreadPool,
73) -> OperationResult<Vec<PointPreview>> {
74    let locations = locate_points(segments, &plan, pool)?;
75    let mut stored = read_stored_points(segments, &plan, &locations, pool)?;
76
77    let mut points = Vec::with_capacity(plan.len());
78    for (id, updates) in plan.into_point_updates() {
79        let location = locations.get(&id);
80        let current = location.map(|location| PointCopy {
81            segment: location.newest.segment,
82            internal_id: location.newest.internal_id,
83            version: location.newest.version,
84        });
85        let slots = location
86            .map(|location| location.slots.clone())
87            .unwrap_or_default();
88
89        // Already applied: the stored point is at or beyond this batch's
90        // version, so re-applying would move it backwards.
91        let already_applied = current
92            .as_ref()
93            .is_some_and(|current| current.version >= updates.version());
94
95        let action = if already_applied {
96            PointAction::Skip
97        } else {
98            match updates.materialize(id, stored.remove(&id))? {
99                Some(point) => PointAction::Store(Box::new(point)),
100                None if current.is_some() => PointAction::Delete,
101                None => PointAction::Missing,
102            }
103        };
104
105        points.push(PointPreview {
106            id,
107            current,
108            slots,
109            action,
110        });
111    }
112
113    Ok(points)
114}
115
116impl<S: UniversalRead + 'static> UpdateOnlyEdgeShard<S> {
117    /// Resolve a batch without writing anything: what
118    /// [`apply_batch`](Self::apply_batch) would do, reported per point.
119    ///
120    /// Runs the same resolution code path as the real apply — the same
121    /// operations are rejected, the same points read — so the report cannot
122    /// drift from the apply's behavior.
123    pub fn preview_batch(
124        &self,
125        operations: impl IntoIterator<Item = (SeqNumberType, CollectionUpdateOperations)>,
126    ) -> OperationResult<UpdateBatchPreview> {
127        let plan = UpdateBatchPlan::build(operations)?;
128        if plan.is_empty() {
129            return Ok(UpdateBatchPreview { points: Vec::new() });
130        }
131
132        let segments = self.segments.read();
133        let points = resolve_batch(&segments, plan, &self.pool)?;
134        Ok(UpdateBatchPreview { points })
135    }
136}
qdrant_edge/edge/update_only/preview.rs

qdrant_edge/edge/update_only/
preview.rs
