harmos_storage/
read_batch.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10/* === Modules === */
11
12/* === Re-Exports === */
13
14/* === Imports === */
15
16// Standard Library Imports
17use std::borrow::Borrow;
18use std::ops::Deref;
19use std::sync::Arc;
20// Third Party Imports
21// Local Imports
22use crate::MemorySize;
23
24/* === Constants === */
25
26/* === Macros === */
27
28/* === Type Aliases === */
29
30/// A retained collection of original typed row pages.
31pub type ReadPages<T> = ReadBatch<Vec<ReadBatch<Vec<T>>>>;
32
33/* === Structs === */
34
35pub struct ReadBatch<T> {
36    inner: Arc<ReadOwner<T>>,
37}
38
39struct ReadOwner<T> {
40    value: T,
41}
42
43/* === Enums === */
44
45/* === Traits === */
46
47/* === Implementations === */
48
49impl<T> ReadBatch<T> {
50    pub fn new(value: T) -> Self {
51        Self { inner: Arc::new(ReadOwner { value }) }
52    }
53
54    /// Moves A Unique Batch Through A Fallible Projection.
55    #[tracing::instrument(name = "storage.read_batch.try_map", level = "trace", skip_all, fields(outcome = "cancelled"))]
56    pub fn try_map<U, E>(self, map: impl FnOnce(T) -> std::result::Result<U, E>) -> std::result::Result<ReadBatch<U>, E>
57    where
58        E: From<crate::Error>,
59    {
60        let owner = Arc::try_unwrap(self.inner).map_err(|_| {
61            E::from(crate::Error::InvalidInput {
62                field: "read projection",
63                reason: "requires a unique owner",
64            })
65        })?;
66        let value = map(owner.value)?;
67        tracing::Span::current().record("outcome", "completed");
68        Ok(ReadBatch::new(value))
69    }
70
71    pub fn try_map_with<U, V, E>(
72        self, other: ReadBatch<U>, map: impl FnOnce(T, U) -> std::result::Result<V, E>,
73    ) -> std::result::Result<ReadBatch<V>, E>
74    where
75        E: From<crate::Error>,
76    {
77        let shared = || {
78            E::from(crate::Error::InvalidInput {
79                field: "read projection",
80                reason: "requires unique owners",
81            })
82        };
83        let left = Arc::try_unwrap(self.inner).map_err(|_| shared())?;
84        let right = Arc::try_unwrap(other.inner).map_err(|_| shared())?;
85        Ok(ReadBatch::new(map(left.value, right.value)?))
86    }
87
88    pub fn owner(&self) -> Arc<impl Borrow<T> + Send + Sync + 'static>
89    where
90        T: Send + Sync + 'static,
91    {
92        Arc::clone(&self.inner)
93    }
94}
95
96impl<T: std::fmt::Debug> std::fmt::Debug for ReadBatch<T> {
97    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
98        formatter.debug_tuple("ReadBatch").field(&self.inner.value).finish()
99    }
100}
101
102impl<T> Clone for ReadBatch<T> {
103    fn clone(&self) -> Self {
104        Self { inner: Arc::clone(&self.inner) }
105    }
106}
107
108impl<T> Deref for ReadBatch<T> {
109    type Target = T;
110
111    fn deref(&self) -> &Self::Target {
112        &self.inner.value
113    }
114}
115
116impl<T> AsRef<T> for ReadBatch<T> {
117    fn as_ref(&self) -> &T {
118        &self.inner.value
119    }
120}
121
122impl<T> Borrow<T> for ReadOwner<T> {
123    fn borrow(&self) -> &T {
124        &self.value
125    }
126}
127
128impl<T: MemorySize> MemorySize for ReadBatch<T> {
129    fn memory_size(&self) -> usize {
130        std::mem::size_of::<Self>()
131            .saturating_add(std::mem::size_of::<ReadOwner<T>>())
132            .saturating_add(2 * std::mem::size_of::<usize>())
133            .saturating_add(self.inner.value.memory_size().saturating_sub(std::mem::size_of::<T>()))
134    }
135}
136
137/* === Public Functions === */
138
139/* === Private Functions === */
140
141/* === Tests === */
142
143#[cfg(test)]
144mod tests {
145    use super::*;
146
147    #[test]
148    fn shared_projection_preserves_payload() {
149        let bytes = vec![1u8, 2, 3];
150        let pointer = bytes.as_ptr();
151        let batch = ReadBatch::new(bytes);
152        let alias = batch.clone();
153        assert!(batch.try_map(Ok::<_, crate::Error>).is_err());
154        assert_eq!(alias.as_ptr(), pointer);
155        let projected = alias.try_map(|bytes| Ok::<_, crate::Error>(vec![bytes])).unwrap();
156        assert_eq!(projected[0].as_ptr(), pointer);
157        assert_eq!(projected[0], [1, 2, 3]);
158    }
159}
harmos_storage/read_batch.rs

harmos_storage/
read_batch.rs
