zeppelin_embed/ingest/
delete_matching.rs

1//! Delete every live document matching a predicate in one mutation, then
2//! remove the deleted bytes from every store file (ZE-217).
3
4use crate::lifecycle::{QueryError, Store, StoreError, StoreState};
5use crate::meta::Predicate;
6use crate::planner::PlanError;
7
8use super::{DocId, IngestError, PurgeError};
9
10/// Outcome of [`Store::delete_matching`].
11#[derive(Clone, Debug, Eq, PartialEq)]
12pub struct DeleteMatchingReport {
13    deleted_ids: Vec<DocId>,
14    generation: u64,
15}
16
17impl DeleteMatchingReport {
18    /// Ascending ids of every deleted document; empty when nothing matched.
19    #[must_use]
20    pub fn deleted_ids(&self) -> &[DocId] {
21        &self.deleted_ids
22    }
23
24    /// Store generation when the call returned. Unchanged when nothing
25    /// matched; otherwise the generation after the physical purge.
26    #[must_use]
27    pub const fn generation(&self) -> u64 {
28        self.generation
29    }
30}
31
32/// Why [`Store::delete_matching`] failed.
33#[derive(Debug)]
34pub enum DeleteMatchingError {
35    /// The predicate does not fit the store schema. Nothing changed.
36    Predicate(PlanError),
37    /// Resolving the matching documents failed. Nothing changed.
38    Query(QueryError),
39    /// The tombstone mutation failed. Nothing changed.
40    Delete(IngestError),
41    /// Admission, the purge intent, or the physical purge failed. When the
42    /// documents were already tombstoned, a writable open finishes the purge.
43    Purge(PurgeError),
44}
45
46impl std::fmt::Display for DeleteMatchingError {
47    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
48        match self {
49            Self::Predicate(error) => write!(formatter, "delete filter: {error}"),
50            Self::Query(error) => write!(formatter, "delete match: {error}"),
51            Self::Delete(error) => write!(formatter, "delete: {error}"),
52            Self::Purge(error) => write!(formatter, "delete purge: {error}"),
53        }
54    }
55}
56
57impl std::error::Error for DeleteMatchingError {
58    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
59        match self {
60            Self::Predicate(error) => Some(error),
61            Self::Query(error) => Some(error),
62            Self::Delete(error) => Some(error),
63            Self::Purge(error) => Some(error),
64        }
65    }
66}
67
68impl From<StoreError> for DeleteMatchingError {
69    fn from(error: StoreError) -> Self {
70        Self::Purge(PurgeError::Store(error))
71    }
72}
73
74impl Store {
75    /// Deletes every live document whose current version matches
76    /// `predicate`, atomically, and physically removes their bytes.
77    ///
78    /// The whole call holds the maintenance, state, writer and WAL locks, so
79    /// no other write lands between resolving the matches and deleting them.
80    /// The order is: resolve the ids; durably record a purge intent for them;
81    /// tombstone them all in one WAL record at one generation (readers see all
82    /// of them or none of them); then run the physical purge, which rewrites
83    /// every affected segment and the WAL. When this returns, no deleted byte
84    /// remains in any store file. A crash before the intent is durable deletes
85    /// nothing; after it, the next writable open completes the purge.
86    ///
87    /// A cascade (ZE-225) extends the id set between resolution and
88    /// `delete_and_purge_locked`, inside the same held locks.
89    pub fn delete_matching(
90        &self,
91        predicate: &Predicate,
92    ) -> Result<DeleteMatchingReport, DeleteMatchingError> {
93        crate::planner::validate_predicate(predicate, &self.schema)
94            .map_err(DeleteMatchingError::Predicate)?;
95        let _maintenance = self
96            .maintenance
97            .lock()
98            .map_err(|_| StoreError::Synchronization {
99                component: "delete matching",
100            })?;
101        #[cfg(feature = "graph-cypher")]
102        let mut native_writer = self.native_document_writer()?;
103        #[cfg(feature = "graph-cypher")]
104        let native_deleted = if let Some(writer) =
105            native_writer.as_mut().and_then(|slot| slot.as_mut())
106        {
107            let ids = {
108                let active = self
109                    .active
110                    .lock()
111                    .map_err(|_| StoreError::Synchronization {
112                        component: "active segment",
113                    })?;
114                let current = active.as_ref().ok_or(StoreError::Closed)?;
115                let snapshot = self
116                    .snapshot
117                    .read()
118                    .map_err(|_| StoreError::Synchronization {
119                        component: "published snapshot",
120                    })?;
121                let snapshot = snapshot.as_ref().ok_or(StoreError::Closed)?;
122                self.live_document_ids_matching(&current.segment, snapshot.segments(), predicate)
123                    .map_err(DeleteMatchingError::Query)?
124            };
125            // Release the snapshot read guard before the mixed writer publishes.
126            if !ids.is_empty()
127                && self
128                    .documents_have_native_nodes(&ids)
129                    .map_err(DeleteMatchingError::Delete)?
130            {
131                self.prepare_purge_intent(
132                    &ids,
133                    self.available_purge_bytes()
134                        .map_err(DeleteMatchingError::Purge)?,
135                    self.vfs.as_ref(),
136                )
137                .map_err(DeleteMatchingError::Purge)?;
138                self.delete_native_documents_locked(&super::DeleteBatch::new(ids.clone()), writer)
139                    .map_err(DeleteMatchingError::Delete)?;
140                Some(ids)
141            } else {
142                None
143            }
144        } else {
145            None
146        };
147        let state = self
148            .state
149            .lock()
150            .map_err(|_| StoreError::Synchronization { component: "state" })?;
151        match *state {
152            StoreState::Open => {}
153            StoreState::Closing => return Err(StoreError::Closing.into()),
154            StoreState::Closed => return Err(StoreError::Closed.into()),
155        }
156        let writer_lock = self
157            .writer_lock
158            .lock()
159            .map_err(|_| StoreError::Synchronization {
160                component: "writer lock",
161            })?;
162        if writer_lock.is_none() {
163            return Err(StoreError::ReadOnly.into());
164        }
165        self.require_no_snapshot_views()?;
166        let mut wal = self
167            .wal_writer
168            .lock()
169            .map_err(|_| StoreError::Synchronization {
170                component: "WAL writer",
171            })?;
172        let writer = wal.as_mut().ok_or(StoreError::ReadOnly)?;
173        let (generation, active) = {
174            let active = self
175                .active
176                .lock()
177                .map_err(|_| StoreError::Synchronization {
178                    component: "active segment",
179                })?;
180            let current = active.as_ref().ok_or(StoreError::Closed)?;
181            (current.generation, std::sync::Arc::clone(&current.segment))
182        };
183        let snapshot = self
184            .snapshot
185            .read()
186            .map_err(|_| StoreError::Synchronization {
187                component: "published snapshot",
188            })?
189            .as_ref()
190            .cloned()
191            .ok_or(StoreError::Closed)?;
192        let ids = self
193            .live_document_ids_matching(&active, snapshot.segments(), predicate)
194            .map_err(DeleteMatchingError::Query)?;
195        #[cfg(feature = "graph-cypher")]
196        let (ids, already_deleted) = match native_deleted {
197            Some(ids) => (ids, true),
198            None => (ids, false),
199        };
200        #[cfg(not(feature = "graph-cypher"))]
201        let already_deleted = false;
202        drop(snapshot);
203        drop(active);
204        let report = if ids.is_empty() {
205            DeleteMatchingReport {
206                deleted_ids: ids,
207                generation,
208            }
209        } else {
210            self.delete_and_purge_locked(writer, ids, already_deleted)?
211        };
212        drop(wal);
213        drop(writer_lock);
214        drop(state);
215        Ok(report)
216    }
217
218    /// Records the purge intent for `ids`, tombstones them in one mutation
219    /// and completes the physical purge. The caller holds the maintenance,
220    /// state, writer and WAL locks; `ids` are ascending, unique and non-empty.
221    fn delete_and_purge_locked(
222        &self,
223        writer: &mut super::StoreWal,
224        ids: Vec<DocId>,
225        already_deleted: bool,
226    ) -> Result<DeleteMatchingReport, DeleteMatchingError> {
227        let vfs = self.vfs.as_ref();
228        let available = self
229            .available_purge_bytes()
230            .map_err(DeleteMatchingError::Purge)?;
231        let token = self
232            .schedule_purge_locked(&ids, available, vfs, writer)
233            .map_err(DeleteMatchingError::Purge)?;
234        if !already_deleted && let Err(error) = self.delete_with_writer(writer, &ids, &[]) {
235            // The tombstones did not commit, so the intent must not purge
236            // these documents at the next open.
237            self.abandon_scheduled_purge_locked(vfs, writer)
238                .map_err(DeleteMatchingError::Purge)?;
239            return Err(DeleteMatchingError::Delete(error));
240        }
241        let purged = self
242            .complete_physical_purge_locked(vfs, writer, token)
243            .map_err(DeleteMatchingError::Purge)?;
244        Ok(DeleteMatchingReport {
245            deleted_ids: ids,
246            generation: purged.generation(),
247        })
248    }
249}
zeppelin_embed/ingest/delete_matching.rs

zeppelin_embed/ingest/
delete_matching.rs
