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    /// Each attempt holds the maintenance, state, writer and WAL locks, so
79    /// no other write lands between resolving the matches and deleting them.
80    /// Reclaim debt discards the attempt and restarts resolution after recovery.
81    /// The order is: resolve the ids; durably record a purge intent for them;
82    /// tombstone them all in one WAL record at one generation (readers see all
83    /// of them or none of them); then run the physical purge, which rewrites
84    /// every affected segment and the WAL. When this returns, no deleted byte
85    /// remains in any store file. A crash before the intent is durable deletes
86    /// nothing; after it, the next writable open completes the purge.
87    ///
88    /// A cascade (ZE-225) extends the id set between resolution and
89    /// `delete_and_purge_locked`, inside the same held locks.
90    pub fn delete_matching(
91        &self,
92        predicate: &Predicate,
93    ) -> Result<DeleteMatchingReport, DeleteMatchingError> {
94        #[cfg(feature = "graph-cypher")]
95        {
96            self.with_native_document_recovery(
97                || self.delete_matching_attempt(predicate),
98                |error| matches!(error, DeleteMatchingError::Delete(error) if error.is_native_reclaim_debt()),
99                DeleteMatchingError::Delete,
100            )
101        }
102        #[cfg(not(feature = "graph-cypher"))]
103        self.delete_matching_attempt(predicate)
104    }
105
106    fn delete_matching_attempt(
107        &self,
108        predicate: &Predicate,
109    ) -> Result<DeleteMatchingReport, DeleteMatchingError> {
110        crate::planner::validate_predicate(predicate, &self.schema)
111            .map_err(DeleteMatchingError::Predicate)?;
112        let _maintenance = self
113            .maintenance
114            .lock()
115            .map_err(|_| StoreError::Synchronization {
116                component: "delete matching",
117            })?;
118        #[cfg(feature = "graph-cypher")]
119        let mut native_writer = self.native_document_writer()?;
120        #[cfg(feature = "graph-cypher")]
121        let native_deleted = if let Some(writer) =
122            native_writer.as_mut().and_then(|slot| slot.as_mut())
123        {
124            let ids = {
125                let active = self
126                    .active
127                    .lock()
128                    .map_err(|_| StoreError::Synchronization {
129                        component: "active segment",
130                    })?;
131                let current = active.as_ref().ok_or(StoreError::Closed)?;
132                let snapshot = self
133                    .snapshot
134                    .read()
135                    .map_err(|_| StoreError::Synchronization {
136                        component: "published snapshot",
137                    })?;
138                let snapshot = snapshot.as_ref().ok_or(StoreError::Closed)?;
139                self.live_document_ids_matching(&current.segment, snapshot.segments(), predicate)
140                    .map_err(DeleteMatchingError::Query)?
141            };
142            // Release the snapshot read guard before the mixed writer publishes.
143            if !ids.is_empty()
144                && self
145                    .documents_have_native_nodes(&ids)
146                    .map_err(DeleteMatchingError::Delete)?
147            {
148                // Validate only; the durable intent follows the mixed commit.
149                self.prepare_purge_intent(
150                    &ids,
151                    self.available_purge_bytes()
152                        .map_err(DeleteMatchingError::Purge)?,
153                    self.vfs.as_ref(),
154                )
155                .map_err(DeleteMatchingError::Purge)?;
156                self.delete_native_documents_locked(&super::DeleteBatch::new(ids.clone()), writer)
157                    .map_err(DeleteMatchingError::Delete)?;
158                Some(ids)
159            } else {
160                None
161            }
162        } else {
163            None
164        };
165        let state = self
166            .state
167            .lock()
168            .map_err(|_| StoreError::Synchronization { component: "state" })?;
169        match *state {
170            StoreState::Open => {}
171            StoreState::Closing => return Err(StoreError::Closing.into()),
172            StoreState::Closed => return Err(StoreError::Closed.into()),
173        }
174        let writer_lock = self
175            .writer_lock
176            .lock()
177            .map_err(|_| StoreError::Synchronization {
178                component: "writer lock",
179            })?;
180        if writer_lock.is_none() {
181            return Err(StoreError::ReadOnly.into());
182        }
183        self.require_no_snapshot_views()?;
184        let mut wal = self
185            .wal_writer
186            .lock()
187            .map_err(|_| StoreError::Synchronization {
188                component: "WAL writer",
189            })?;
190        let writer = wal.as_mut().ok_or(StoreError::ReadOnly)?;
191        let (generation, active) = {
192            let active = self
193                .active
194                .lock()
195                .map_err(|_| StoreError::Synchronization {
196                    component: "active segment",
197                })?;
198            let current = active.as_ref().ok_or(StoreError::Closed)?;
199            (current.generation, std::sync::Arc::clone(&current.segment))
200        };
201        let snapshot = self
202            .snapshot
203            .read()
204            .map_err(|_| StoreError::Synchronization {
205                component: "published snapshot",
206            })?
207            .as_ref()
208            .cloned()
209            .ok_or(StoreError::Closed)?;
210        let ids = self
211            .live_document_ids_matching(&active, snapshot.segments(), predicate)
212            .map_err(DeleteMatchingError::Query)?;
213        #[cfg(feature = "graph-cypher")]
214        let (ids, already_deleted) = match native_deleted {
215            Some(ids) => (ids, true),
216            None => (ids, false),
217        };
218        #[cfg(not(feature = "graph-cypher"))]
219        let already_deleted = false;
220        drop(snapshot);
221        drop(active);
222        let report = if ids.is_empty() {
223            DeleteMatchingReport {
224                deleted_ids: ids,
225                generation,
226            }
227        } else {
228            self.delete_and_purge_locked(writer, ids, already_deleted)?
229        };
230        drop(wal);
231        drop(writer_lock);
232        drop(state);
233        Ok(report)
234    }
235
236    /// Records the purge intent for `ids`, tombstones them in one mutation
237    /// and completes the physical purge. The caller holds the maintenance,
238    /// state, writer and WAL locks; `ids` are ascending, unique and non-empty.
239    fn delete_and_purge_locked(
240        &self,
241        writer: &mut super::StoreWal,
242        ids: Vec<DocId>,
243        already_deleted: bool,
244    ) -> Result<DeleteMatchingReport, DeleteMatchingError> {
245        let vfs = self.vfs.as_ref();
246        let available = self
247            .available_purge_bytes()
248            .map_err(DeleteMatchingError::Purge)?;
249        let token = self
250            .schedule_purge_locked(&ids, available, vfs, writer)
251            .map_err(DeleteMatchingError::Purge)?;
252        if !already_deleted && let Err(error) = self.delete_with_writer(writer, &ids, &[]) {
253            // The tombstones did not commit, so the intent must not purge
254            // these documents at the next open.
255            self.abandon_scheduled_purge_locked(vfs, writer)
256                .map_err(DeleteMatchingError::Purge)?;
257            return Err(DeleteMatchingError::Delete(error));
258        }
259        let purged = self
260            .complete_physical_purge_locked(vfs, writer, token)
261            .map_err(DeleteMatchingError::Purge)?;
262        Ok(DeleteMatchingReport {
263            deleted_ids: ids,
264            generation: purged.generation(),
265        })
266    }
267}
zeppelin_embed/ingest/delete_matching.rs

zeppelin_embed/ingest/
delete_matching.rs
