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        let state = self
102            .state
103            .lock()
104            .map_err(|_| StoreError::Synchronization { component: "state" })?;
105        match *state {
106            StoreState::Open => {}
107            StoreState::Closing => return Err(StoreError::Closing.into()),
108            StoreState::Closed => return Err(StoreError::Closed.into()),
109        }
110        let writer_lock = self
111            .writer_lock
112            .lock()
113            .map_err(|_| StoreError::Synchronization {
114                component: "writer lock",
115            })?;
116        if writer_lock.is_none() {
117            return Err(StoreError::ReadOnly.into());
118        }
119        self.require_no_snapshot_views()?;
120        let mut wal = self
121            .wal_writer
122            .lock()
123            .map_err(|_| StoreError::Synchronization {
124                component: "WAL writer",
125            })?;
126        let writer = wal.as_mut().ok_or(StoreError::ReadOnly)?;
127        let (generation, active) = {
128            let active = self
129                .active
130                .lock()
131                .map_err(|_| StoreError::Synchronization {
132                    component: "active segment",
133                })?;
134            let current = active.as_ref().ok_or(StoreError::Closed)?;
135            (current.generation, std::sync::Arc::clone(&current.segment))
136        };
137        let snapshot = self
138            .snapshot
139            .read()
140            .map_err(|_| StoreError::Synchronization {
141                component: "published snapshot",
142            })?
143            .as_ref()
144            .cloned()
145            .ok_or(StoreError::Closed)?;
146        let ids = self
147            .live_document_ids_matching(&active, snapshot.segments(), predicate)
148            .map_err(DeleteMatchingError::Query)?;
149        drop(snapshot);
150        drop(active);
151        let report = if ids.is_empty() {
152            DeleteMatchingReport {
153                deleted_ids: ids,
154                generation,
155            }
156        } else {
157            self.delete_and_purge_locked(writer, ids)?
158        };
159        drop(wal);
160        drop(writer_lock);
161        drop(state);
162        Ok(report)
163    }
164
165    /// Records the purge intent for `ids`, tombstones them in one mutation
166    /// and completes the physical purge. The caller holds the maintenance,
167    /// state, writer and WAL locks; `ids` are ascending, unique and non-empty.
168    fn delete_and_purge_locked(
169        &self,
170        writer: &mut super::StoreWal,
171        ids: Vec<DocId>,
172    ) -> Result<DeleteMatchingReport, DeleteMatchingError> {
173        let vfs = self.vfs.as_ref();
174        let available = self
175            .available_purge_bytes()
176            .map_err(DeleteMatchingError::Purge)?;
177        let token = self
178            .schedule_purge_locked(&ids, available, vfs)
179            .map_err(DeleteMatchingError::Purge)?;
180        if let Err(error) = self.delete_with_writer(writer, &ids, &[]) {
181            // The tombstones did not commit, so the intent must not purge
182            // these documents at the next open.
183            self.abandon_scheduled_purge_locked(vfs)
184                .map_err(DeleteMatchingError::Purge)?;
185            return Err(DeleteMatchingError::Delete(error));
186        }
187        let purged = self
188            .complete_physical_purge_locked(vfs, writer, token)
189            .map_err(DeleteMatchingError::Purge)?;
190        Ok(DeleteMatchingReport {
191            deleted_ids: ids,
192            generation: purged.generation(),
193        })
194    }
195}
zeppelin_embed/ingest/delete_matching.rs

zeppelin_embed/ingest/
delete_matching.rs
