harmos_storage/
selection.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10// Stored Entry Kinds Shared By Scans And Live Subscriptions
11
12/* === Modules === */
13
14mod sealed {
15    pub trait Sealed {}
16}
17
18/* === Re-Exports === */
19
20/* === Imports === */
21
22// Standard Library Imports
23use std::time::SystemTime;
24
25// Third Party Imports
26use serde::Serialize;
27
28// Local Imports
29use super::backend::{BackendSnapshot, ScanPage, StorageBackend};
30use super::manager::Operation;
31use super::query::{FileSelection, ObjectSelection, RecordSelection, SchemaSelection, StreamSelection};
32use super::{FileReference, ObjectEntry, StorageCursor, StorageFuture, StorageSchema, StoredRecord, StoredRow};
33use crate::Identifier;
34use harmos_domain::model::storage::{IdentitySelection, StoredIdentity};
35
36/* === Constants === */
37
38/* === Macros === */
39
40/* === Type Aliases === */
41
42/* === Structs === */
43
44pub struct Queued<'a>(&'a Operation);
45
46/* === Enums === */
47
48/* === Traits === */
49
50pub trait Entry: Clone + Serialize + Send + Sync + 'static + sealed::Sealed {
51    type Selection: Send + Sync;
52
53    fn position(&self) -> &StorageCursor;
54    fn matches(&self, selection: &Self::Selection) -> bool;
55
56    fn start(_selection: &Self::Selection) -> u64 {
57        0
58    }
59
60    fn pending(queued: Queued<'_>) -> &[Self];
61    fn scan<'a>(
62        backend: &'a dyn StorageBackend, snapshot: &'a BackendSnapshot, selection: &'a Self::Selection, page: ScanPage,
63        memory: crate::arrow::RetainedMemory,
64    ) -> StorageFuture<'a, Vec<Self>>;
65}
66
67/* === Implementations === */
68
69impl<'a> Queued<'a> {
70    pub(super) fn new(operation: &'a Operation) -> Self {
71        Self(operation)
72    }
73}
74
75impl sealed::Sealed for ObjectEntry {}
76impl sealed::Sealed for StoredRecord {}
77impl sealed::Sealed for StoredRow {}
78impl sealed::Sealed for StorageSchema {}
79impl sealed::Sealed for FileReference {}
80impl sealed::Sealed for StoredIdentity {}
81
82impl Entry for ObjectEntry {
83    type Selection = ObjectSelection;
84
85    fn position(&self) -> &StorageCursor {
86        &self.position
87    }
88
89    // Latest selection is resolved by the backend; matching checks identity only.
90    fn matches(&self, selection: &ObjectSelection) -> bool {
91        self.version.check_position(&self.position).is_ok()
92            && selection.schema.as_ref().is_none_or(|key| key == &self.value.schema)
93            && selection.key.as_ref().is_none_or(|key| key == &self.value.item)
94    }
95
96    fn pending(queued: Queued<'_>) -> &[Self] {
97        match queued.0 {
98            Operation::ObjectGroup { objects, .. } => objects.as_slice(),
99            _ => &[],
100        }
101    }
102
103    fn scan<'a>(
104        backend: &'a dyn StorageBackend, snapshot: &'a BackendSnapshot, selection: &'a ObjectSelection, page: ScanPage,
105        memory: crate::arrow::RetainedMemory,
106    ) -> StorageFuture<'a, Vec<Self>> {
107        let future = backend.read_objects(snapshot, selection, page, memory);
108        let span = tracing::trace_span!("storage.backend.scan", method = "primary.read_objects");
109        Box::pin(tracing::Instrument::instrument(future, span))
110    }
111}
112
113impl Entry for StoredRecord {
114    type Selection = RecordSelection;
115
116    fn position(&self) -> &StorageCursor {
117        &self.position
118    }
119
120    fn matches(&self, selection: &RecordSelection) -> bool {
121        self.schema == selection.schema
122            && optional(&selection.scope, self.scope.as_ref())
123            && optional(&selection.item, Some(&self.item))
124            && within(selection.source_start, selection.source_end, self.source_time)
125    }
126
127    fn pending(queued: Queued<'_>) -> &[Self] {
128        match queued.0 {
129            Operation::Records(values) => values.as_slice(),
130            _ => &[],
131        }
132    }
133
134    fn scan<'a>(
135        backend: &'a dyn StorageBackend, snapshot: &'a BackendSnapshot, selection: &'a RecordSelection, page: ScanPage,
136        memory: crate::arrow::RetainedMemory,
137    ) -> StorageFuture<'a, Vec<Self>> {
138        let future = backend.read_records(snapshot, selection, page, memory);
139        let span = tracing::trace_span!("storage.backend.scan", method = "primary.read_records");
140        Box::pin(tracing::Instrument::instrument(future, span))
141    }
142}
143
144impl Entry for StoredRow {
145    type Selection = StreamSelection;
146
147    fn position(&self) -> &StorageCursor {
148        &self.record.position
149    }
150
151    fn matches(&self, selection: &StreamSelection) -> bool {
152        let record = &self.record;
153        record.schema == selection.schema
154            && record.position.sequence > selection.after
155            && optional(&selection.scope, record.scope.as_ref())
156            && optional(&selection.item, Some(&record.item))
157            && optional(&selection.stream, Some(&self.stream))
158            && within(selection.source_start, selection.source_end, record.source_time)
159    }
160
161    fn start(selection: &StreamSelection) -> u64 {
162        selection.after
163    }
164
165    fn pending(queued: Queued<'_>) -> &[Self] {
166        match queued.0 {
167            Operation::Stream(values) => values.as_slice(),
168            _ => &[],
169        }
170    }
171
172    fn scan<'a>(
173        backend: &'a dyn StorageBackend, snapshot: &'a BackendSnapshot, selection: &'a StreamSelection, page: ScanPage,
174        memory: crate::arrow::RetainedMemory,
175    ) -> StorageFuture<'a, Vec<Self>> {
176        let future = backend.read_stream(snapshot, selection, page, memory);
177        let span = tracing::trace_span!("storage.backend.scan", method = "primary.read_stream");
178        Box::pin(tracing::Instrument::instrument(future, span))
179    }
180}
181
182impl Entry for StorageSchema {
183    type Selection = SchemaSelection;
184
185    fn position(&self) -> &StorageCursor {
186        &self.position
187    }
188
189    fn matches(&self, selection: &SchemaSelection) -> bool {
190        selection.schema.as_ref().is_none_or(|key| key == &self.key)
191    }
192
193    fn pending(queued: Queued<'_>) -> &[Self] {
194        match queued.0 {
195            Operation::Schema(value) => value.as_slice(),
196            Operation::ObjectGroup { schemas, .. } => schemas.as_slice(),
197            _ => &[],
198        }
199    }
200
201    fn scan<'a>(
202        backend: &'a dyn StorageBackend, snapshot: &'a BackendSnapshot, selection: &'a SchemaSelection, page: ScanPage,
203        memory: crate::arrow::RetainedMemory,
204    ) -> StorageFuture<'a, Vec<Self>> {
205        let future = backend.read_schemas(snapshot, selection, page, memory);
206        let span = tracing::trace_span!("storage.backend.scan", method = "primary.read_schemas");
207        Box::pin(tracing::Instrument::instrument(future, span))
208    }
209}
210
211impl Entry for FileReference {
212    type Selection = FileSelection;
213
214    fn position(&self) -> &StorageCursor {
215        &self.position
216    }
217
218    fn matches(&self, selection: &FileSelection) -> bool {
219        optional(&selection.scope, self.metadata.scope.as_ref()) && optional(&selection.item, Some(&self.metadata.id))
220    }
221
222    fn pending(queued: Queued<'_>) -> &[Self] {
223        match queued.0 {
224            Operation::File(value) => value.as_slice(),
225            _ => &[],
226        }
227    }
228
229    fn scan<'a>(
230        backend: &'a dyn StorageBackend, snapshot: &'a BackendSnapshot, selection: &'a FileSelection, page: ScanPage,
231        memory: crate::arrow::RetainedMemory,
232    ) -> StorageFuture<'a, Vec<Self>> {
233        let future = backend.read_files(snapshot, selection, page, memory);
234        let span = tracing::trace_span!("storage.backend.scan", method = "primary.read_files");
235        Box::pin(tracing::Instrument::instrument(future, span))
236    }
237}
238
239// Identities Are Catalogued From Durable Storage Only; None Wait In The Queue.
240impl Entry for StoredIdentity {
241    type Selection = IdentitySelection;
242
243    fn position(&self) -> &StorageCursor {
244        self.first()
245    }
246
247    fn matches(&self, selection: &IdentitySelection) -> bool {
248        selection.matches(self)
249    }
250
251    fn pending(_: Queued<'_>) -> &[Self] {
252        &[]
253    }
254
255    fn scan<'a>(
256        backend: &'a dyn StorageBackend, snapshot: &'a BackendSnapshot, selection: &'a IdentitySelection, page: ScanPage,
257        memory: crate::arrow::RetainedMemory,
258    ) -> StorageFuture<'a, Vec<Self>> {
259        let future = backend.read_identities(snapshot, selection, page, memory);
260        let span = tracing::trace_span!("storage.backend.scan", method = "primary.read_identities");
261        Box::pin(tracing::Instrument::instrument(future, span))
262    }
263}
264
265/* === Public Functions === */
266
267/* === Private Functions === */
268
269fn optional(expected: &Option<Identifier>, actual: Option<&Identifier>) -> bool {
270    expected.as_ref().is_none_or(|expected| Some(expected) == actual)
271}
272
273// Half-Open Source Windows Select Only Records Carrying A Source Time.
274fn within(start: Option<SystemTime>, end: Option<SystemTime>, time: Option<SystemTime>) -> bool {
275    match (start, end) {
276        (Some(start), Some(end)) => time.is_some_and(|time| time >= start && time < end),
277        _ => true,
278    }
279}
280
281/* === Tests === */
harmos_storage/selection.rs

harmos_storage/
selection.rs
